Handle compressed empty DataPage v2 (#7389)

* Work with empty datapage v2 (only null values)

* Remove obsolete TODO

* Check datapage header values for plausibility

* Replace `as usize` with `usize::try_from(…)?` where useful

* Assert reading page_v2_empty_compressed.parquet

* Make get_next_page in test fail on errors

* Split long error message

* Move parquet-testing to latest master

Author: EnricoMi

parquet-testing

@@ -322,7 +322,7 @@ impl<R: 'static + ChunkReader> RowGroupReader for SerializedRowGroupReader<'_, R
         Ok(Box::new(SerializedPageReader::new_with_properties(
             Arc::clone(&self.chunk_reader),
             col,
-            self.metadata.num_rows() as usize,
+            usize::try_from(self.metadata.num_rows())?,
             page_locations,
             props,
         )?))
@@ -442,8 +442,24 @@ pub(crate) fn decode_page(
     let mut can_decompress = true;
 
     if let Some(ref header_v2) = page_header.data_page_header_v2 {
-        offset = (header_v2.definition_levels_byte_length + header_v2.repetition_levels_byte_length)
-            as usize;
+        if header_v2.definition_levels_byte_length < 0
+            || header_v2.repetition_levels_byte_length < 0
+            || header_v2.definition_levels_byte_length + header_v2.repetition_levels_byte_length
+                > page_header.uncompressed_page_size
+        {
+            return Err(general_err!(
+                "DataPage v2 header contains implausible values \
+                    for definition_levels_byte_length ({}) \
+                    and repetition_levels_byte_length ({}) \
+                    given DataPage header provides uncompressed_page_size ({})",
+                header_v2.definition_levels_byte_length,
+                header_v2.repetition_levels_byte_length,
+                page_header.uncompressed_page_size
+            ));
+        }
+        offset = usize::try_from(
+            header_v2.definition_levels_byte_length + header_v2.repetition_levels_byte_length,
+        )?;
         // When is_compressed flag is missing the page is considered compressed
         can_decompress = header_v2.is_compressed.unwrap_or(true);
     }
@@ -452,21 +468,20 @@ pub(crate) fn decode_page(
     // maximum page header size and abort if that is exceeded.
     let buffer = match decompressor {
         Some(decompressor) if can_decompress => {
-            let uncompressed_size = page_header.uncompressed_page_size as usize;
-            let mut decompressed = Vec::with_capacity(uncompressed_size);
-            let compressed = &buffer.as_ref()[offset..];
+            let uncompressed_page_size = usize::try_from(page_header.uncompressed_page_size)?;
+            let decompressed_size = uncompressed_page_size - offset;
+            let mut decompressed = Vec::with_capacity(uncompressed_page_size);
             decompressed.extend_from_slice(&buffer.as_ref()[..offset]);
-            decompressor.decompress(
-                compressed,
-                &mut decompressed,
-                Some(uncompressed_size - offset),
-            )?;
+            if decompressed_size > 0 {
+                let compressed = &buffer.as_ref()[offset..];
+                decompressor.decompress(compressed, &mut decompressed, Some(decompressed_size))?;
+            }
 
-            if decompressed.len() != uncompressed_size {
+            if decompressed.len() != uncompressed_page_size {
                 return Err(general_err!(
                     "Actual decompressed size doesn't match the expected one ({} vs {})",
                     decompressed.len(),
-                    uncompressed_size
+                    uncompressed_page_size
                 ));
             }
 
@@ -652,8 +667,8 @@ impl<R: ChunkReader> SerializedPageReader<R> {
                 }
             }
             None => SerializedPageReaderState::Values {
-                offset: start as usize,
-                remaining_bytes: len as usize,
+                offset: usize::try_from(start)?,
+                remaining_bytes: usize::try_from(len)?,
                 next_page_header: None,
                 page_ordinal: 0,
                 require_dictionary: meta.dictionary_page_offset().is_some(),
@@ -717,9 +732,9 @@ impl<R: ChunkReader> SerializedPageReader<R> {
                 ..
             } => {
                 if let Some(page) = dictionary_page {
-                    Ok(Some(page.offset as usize))
+                    Ok(Some(usize::try_from(page.offset)?))
                 } else if let Some(page) = page_locations.front() {
-                    Ok(Some(page.offset as usize))
+                    Ok(Some(usize::try_from(page.offset)?))
                 } else {
                     Ok(None)
                 }
@@ -861,7 +876,7 @@ impl<R: ChunkReader> PageReader for SerializedPageReader<R> {
                         None => return Ok(None),
                     };
 
-                    let page_len = front.compressed_page_size as usize;
+                    let page_len = usize::try_from(front.compressed_page_size)?;
 
                     let buffer = self.reader.get_bytes(front.offset as u64, page_len)?;
 
@@ -1031,7 +1046,7 @@ mod tests {
     use crate::file::properties::{EnabledStatistics, WriterProperties};
     use crate::format::BoundaryOrder;
 
-    use crate::basic::{self, ColumnOrder};
+    use crate::basic::{self, ColumnOrder, SortOrder};
     use crate::column::reader::ColumnReader;
     use crate::data_type::private::ParquetValueType;
     use crate::data_type::{AsBytes, FixedLenByteArrayType, Int32Type};
@@ -1193,7 +1208,7 @@ mod tests {
         assert!(page_reader_0_result.is_ok());
         let mut page_reader_0: Box<dyn PageReader> = page_reader_0_result.unwrap();
         let mut page_count = 0;
-        while let Ok(Some(page)) = page_reader_0.get_next_page() {
+        while let Some(page) = page_reader_0.get_next_page().unwrap() {
             let is_expected_page = match page {
                 Page::DictionaryPage {
                     buf,
@@ -1287,7 +1302,7 @@ mod tests {
         assert!(page_reader_0_result.is_ok());
         let mut page_reader_0: Box<dyn PageReader> = page_reader_0_result.unwrap();
         let mut page_count = 0;
-        while let Ok(Some(page)) = page_reader_0.get_next_page() {
+        while let Some(page) = page_reader_0.get_next_page().unwrap() {
             let is_expected_page = match page {
                 Page::DictionaryPage {
                     buf,
@@ -1331,6 +1346,198 @@ mod tests {
         assert_eq!(page_count, 2);
     }
 
+    #[test]
+    fn test_file_reader_empty_compressed_datapage_v2() {
+        // this file has a compressed datapage that un-compresses to 0 bytes
+        let test_file = get_test_file("page_v2_empty_compressed.parquet");
+        let reader_result = SerializedFileReader::new(test_file);
+        assert!(reader_result.is_ok());
+        let reader = reader_result.unwrap();
+
+        // Test contents in Parquet metadata
+        let metadata = reader.metadata();
+        assert_eq!(metadata.num_row_groups(), 1);
+
+        // Test contents in file metadata
+        let file_metadata = metadata.file_metadata();
+        assert!(file_metadata.created_by().is_some());
+        assert_eq!(
+            file_metadata.created_by().unwrap(),
+            "parquet-cpp-arrow version 14.0.2"
+        );
+        assert!(file_metadata.key_value_metadata().is_some());
+        assert_eq!(
+            file_metadata.key_value_metadata().to_owned().unwrap().len(),
+            1
+        );
+
+        assert_eq!(file_metadata.num_rows(), 10);
+        assert_eq!(file_metadata.version(), 2);
+        let expected_order = ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::SIGNED);
+        assert_eq!(
+            file_metadata.column_orders(),
+            Some(vec![expected_order].as_ref())
+        );
+
+        let row_group_metadata = metadata.row_group(0);
+
+        // Check each column order
+        for i in 0..row_group_metadata.num_columns() {
+            assert_eq!(file_metadata.column_order(i), expected_order);
+        }
+
+        // Test row group reader
+        let row_group_reader_result = reader.get_row_group(0);
+        assert!(row_group_reader_result.is_ok());
+        let row_group_reader: Box<dyn RowGroupReader> = row_group_reader_result.unwrap();
+        assert_eq!(
+            row_group_reader.num_columns(),
+            row_group_metadata.num_columns()
+        );
+        assert_eq!(
+            row_group_reader.metadata().total_byte_size(),
+            row_group_metadata.total_byte_size()
+        );
+
+        // Test page readers
+        let page_reader_0_result = row_group_reader.get_column_page_reader(0);
+        assert!(page_reader_0_result.is_ok());
+        let mut page_reader_0: Box<dyn PageReader> = page_reader_0_result.unwrap();
+        let mut page_count = 0;
+        while let Some(page) = page_reader_0.get_next_page().unwrap() {
+            let is_expected_page = match page {
+                Page::DictionaryPage {
+                    buf,
+                    num_values,
+                    encoding,
+                    is_sorted,
+                } => {
+                    assert_eq!(buf.len(), 0);
+                    assert_eq!(num_values, 0);
+                    assert_eq!(encoding, Encoding::PLAIN);
+                    assert!(!is_sorted);
+                    true
+                }
+                Page::DataPageV2 {
+                    buf,
+                    num_values,
+                    encoding,
+                    num_nulls,
+                    num_rows,
+                    def_levels_byte_len,
+                    rep_levels_byte_len,
+                    is_compressed,
+                    statistics,
+                } => {
+                    assert_eq!(buf.len(), 3);
+                    assert_eq!(num_values, 10);
+                    assert_eq!(encoding, Encoding::RLE_DICTIONARY);
+                    assert_eq!(num_nulls, 10);
+                    assert_eq!(num_rows, 10);
+                    assert_eq!(def_levels_byte_len, 2);
+                    assert_eq!(rep_levels_byte_len, 0);
+                    assert!(is_compressed);
+                    assert!(statistics.is_some());
+                    true
+                }
+                _ => false,
+            };
+            assert!(is_expected_page);
+            page_count += 1;
+        }
+        assert_eq!(page_count, 2);
+    }
+
+    #[test]
+    fn test_file_reader_empty_datapage_v2() {
+        // this file has 0 bytes compressed datapage that un-compresses to 0 bytes
+        let test_file = get_test_file("datapage_v2_empty_datapage.snappy.parquet");
+        let reader_result = SerializedFileReader::new(test_file);
+        assert!(reader_result.is_ok());
+        let reader = reader_result.unwrap();
+
+        // Test contents in Parquet metadata
+        let metadata = reader.metadata();
+        assert_eq!(metadata.num_row_groups(), 1);
+
+        // Test contents in file metadata
+        let file_metadata = metadata.file_metadata();
+        assert!(file_metadata.created_by().is_some());
+        assert_eq!(
+            file_metadata.created_by().unwrap(),
+            "parquet-mr version 1.13.1 (build db4183109d5b734ec5930d870cdae161e408ddba)"
+        );
+        assert!(file_metadata.key_value_metadata().is_some());
+        assert_eq!(
+            file_metadata.key_value_metadata().to_owned().unwrap().len(),
+            2
+        );
+
+        assert_eq!(file_metadata.num_rows(), 1);
+        assert_eq!(file_metadata.version(), 1);
+        let expected_order = ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::SIGNED);
+        assert_eq!(
+            file_metadata.column_orders(),
+            Some(vec![expected_order].as_ref())
+        );
+
+        let row_group_metadata = metadata.row_group(0);
+
+        // Check each column order
+        for i in 0..row_group_metadata.num_columns() {
+            assert_eq!(file_metadata.column_order(i), expected_order);
+        }
+
+        // Test row group reader
+        let row_group_reader_result = reader.get_row_group(0);
+        assert!(row_group_reader_result.is_ok());
+        let row_group_reader: Box<dyn RowGroupReader> = row_group_reader_result.unwrap();
+        assert_eq!(
+            row_group_reader.num_columns(),
+            row_group_metadata.num_columns()
+        );
+        assert_eq!(
+            row_group_reader.metadata().total_byte_size(),
+            row_group_metadata.total_byte_size()
+        );
+
+        // Test page readers
+        let page_reader_0_result = row_group_reader.get_column_page_reader(0);
+        assert!(page_reader_0_result.is_ok());
+        let mut page_reader_0: Box<dyn PageReader> = page_reader_0_result.unwrap();
+        let mut page_count = 0;
+        while let Some(page) = page_reader_0.get_next_page().unwrap() {
+            let is_expected_page = match page {
+                Page::DataPageV2 {
+                    buf,
+                    num_values,
+                    encoding,
+                    num_nulls,
+                    num_rows,
+                    def_levels_byte_len,
+                    rep_levels_byte_len,
+                    is_compressed,
+                    statistics,
+                } => {
+                    assert_eq!(buf.len(), 2);
+                    assert_eq!(num_values, 1);
+                    assert_eq!(encoding, Encoding::PLAIN);
+                    assert_eq!(num_nulls, 1);
+                    assert_eq!(num_rows, 1);
+                    assert_eq!(def_levels_byte_len, 2);
+                    assert_eq!(rep_levels_byte_len, 0);
+                    assert!(is_compressed);
+                    assert!(statistics.is_none());
+                    true
+                }
+                _ => false,
+            };
+            assert!(is_expected_page);
+            page_count += 1;
+        }
+        assert_eq!(page_count, 1);
+    }
+
     fn get_serialized_page_reader<R: ChunkReader>(
         file_reader: &SerializedFileReader<R>,
         row_group: usize,
@@ -1361,7 +1568,7 @@ mod tests {
         SerializedPageReader::new_with_properties(
             Arc::clone(&row_group.chunk_reader),
             col,
-            row_group.metadata.num_rows() as usize,
+            usize::try_from(row_group.metadata.num_rows())?,
             page_locations,
             props,
         )