diff --git a/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedDeltaEncodedValuesReader.java b/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedDeltaEncodedValuesReader.java index 446b83c7f10d..a0f4b113a85e 100644 --- a/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedDeltaEncodedValuesReader.java +++ b/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedDeltaEncodedValuesReader.java @@ -149,7 +149,7 @@ private void readValues( int remaining = total; int currentRowId = rowId; // First value - if (valuesRead == 0) { + if (valuesRead == 0 && total > 0) { outputWriter.write(vec, ((long) (currentRowId + valuesRead) * typeWidth), firstValue); lastValueRead = firstValue; currentRowId++; diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetVectorizedReads.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetVectorizedReads.java index 24a3603dc7f2..1a5e9cd93f17 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetVectorizedReads.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetVectorizedReads.java @@ -65,6 +65,7 @@ import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.types.Types; import org.apache.parquet.column.ParquetProperties; +import org.apache.parquet.hadoop.ParquetOutputFormat; import org.apache.parquet.schema.GroupType; import org.apache.parquet.schema.MessageType; import org.apache.parquet.schema.Type; @@ -254,6 +255,17 @@ FileAppender getParquetV2Writer(Schema schema, File testFile) throws IOE .build(); } + FileAppender parquetV2WriterWithoutDictionary(Schema schema, File testFile) + throws IOException { + return Parquet.write(Files.localOutput(testFile)) + .schema(schema) + .createWriterFunc(GenericParquetWriter::create) + .named("test") + .writerVersion(ParquetProperties.WriterVersion.PARQUET_2_0) + .set(ParquetOutputFormat.ENABLE_DICTIONARY, "false") + .build(); + } + void assertRecordsMatch( Schema schema, int expectedSize, @@ -430,6 +442,37 @@ public void testSupportedReadsForParquetV2() throws Exception { assertRecordsMatch(schema, 30000, data, dataFile, false, BATCH_SIZE); } + @Test + public void testAllNullStringAndBinaryColumnsParquetV2() throws Exception { + // V2 string/binary uses DELTA_BYTE_ARRAY; an all-null page decodes to a delta length stream + // with totalValueCount == 0. + Schema schema = + new Schema( + required(100, "id", Types.LongType.get()), + optional(101, "string_data", Types.StringType.get()), + optional(102, "binary_data", Types.BinaryType.get())); + + int stringOrdinal = 1; + int binaryOrdinal = 2; + File dataFile = temp.resolve("all-null-v2.parquet").toFile(); + Iterable data = + generateData( + schema, + 100, + 0L, + RandomData.DEFAULT_NULL_PERCENTAGE, + record -> { + record.set(stringOrdinal, null); + record.set(binaryOrdinal, null); + return record; + }); + try (FileAppender writer = parquetV2WriterWithoutDictionary(schema, dataFile)) { + writer.addAll(data); + } + + assertRecordsMatch(schema, 100, data, dataFile, false, BATCH_SIZE); + } + @Test public void testUnsupportedReadsForParquetV2() throws Exception { // Some types use delta encoding and which are not supported for vectorized reads diff --git a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetVectorizedReads.java b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetVectorizedReads.java index 6011c6dad7d2..bf6ff7907f75 100644 --- a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetVectorizedReads.java +++ b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetVectorizedReads.java @@ -65,6 +65,7 @@ import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.types.Types; import org.apache.parquet.column.ParquetProperties; +import org.apache.parquet.hadoop.ParquetOutputFormat; import org.apache.parquet.schema.GroupType; import org.apache.parquet.schema.MessageType; import org.apache.parquet.schema.Type; @@ -254,6 +255,17 @@ FileAppender getParquetV2Writer(Schema schema, File testFile) throws IOE .build(); } + FileAppender parquetV2WriterWithoutDictionary(Schema schema, File testFile) + throws IOException { + return Parquet.write(Files.localOutput(testFile)) + .schema(schema) + .createWriterFunc(GenericParquetWriter::create) + .named("test") + .writerVersion(ParquetProperties.WriterVersion.PARQUET_2_0) + .set(ParquetOutputFormat.ENABLE_DICTIONARY, "false") + .build(); + } + void assertRecordsMatch( Schema schema, int expectedSize, @@ -430,6 +442,37 @@ public void testSupportedReadsForParquetV2() throws Exception { assertRecordsMatch(schema, 30000, data, dataFile, false, BATCH_SIZE); } + @Test + public void testAllNullStringAndBinaryColumnsParquetV2() throws Exception { + // V2 string/binary uses DELTA_BYTE_ARRAY; an all-null page decodes to a delta length stream + // with totalValueCount == 0. + Schema schema = + new Schema( + required(100, "id", Types.LongType.get()), + optional(101, "string_data", Types.StringType.get()), + optional(102, "binary_data", Types.BinaryType.get())); + + int stringOrdinal = 1; + int binaryOrdinal = 2; + File dataFile = temp.resolve("all-null-v2.parquet").toFile(); + Iterable data = + generateData( + schema, + 100, + 0L, + RandomData.DEFAULT_NULL_PERCENTAGE, + record -> { + record.set(stringOrdinal, null); + record.set(binaryOrdinal, null); + return record; + }); + try (FileAppender writer = parquetV2WriterWithoutDictionary(schema, dataFile)) { + writer.addAll(data); + } + + assertRecordsMatch(schema, 100, data, dataFile, false, BATCH_SIZE); + } + @Test public void testUnsupportedReadsForParquetV2() throws Exception { // Some types use delta encoding and which are not supported for vectorized reads diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetVectorizedReads.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetVectorizedReads.java index 4a24a725e0f5..12047bd12847 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetVectorizedReads.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetVectorizedReads.java @@ -267,6 +267,17 @@ FileAppender parquetWriterWithoutDictionary(Schema schema, File testFile .build(); } + FileAppender parquetV2WriterWithoutDictionary(Schema schema, File testFile) + throws IOException { + return Parquet.write(Files.localOutput(testFile)) + .schema(schema) + .createWriterFunc(GenericParquetWriter::create) + .named("test") + .writerVersion(ParquetProperties.WriterVersion.PARQUET_2_0) + .set(ParquetOutputFormat.ENABLE_DICTIONARY, "false") + .build(); + } + void assertRecordsMatch( Schema schema, int expectedSize, @@ -443,6 +454,37 @@ public void testSupportedReadsForParquetV2() throws Exception { assertRecordsMatch(schema, 30000, data, dataFile, false, BATCH_SIZE); } + @Test + public void testAllNullStringAndBinaryColumnsParquetV2() throws Exception { + // V2 string/binary uses DELTA_BYTE_ARRAY; an all-null page decodes to a delta length stream + // with totalValueCount == 0. + Schema schema = + new Schema( + required(100, "id", Types.LongType.get()), + optional(101, "string_data", Types.StringType.get()), + optional(102, "binary_data", Types.BinaryType.get())); + + int stringOrdinal = 1; + int binaryOrdinal = 2; + File dataFile = temp.resolve("all-null-v2.parquet").toFile(); + Iterable data = + generateData( + schema, + 100, + 0L, + RandomData.DEFAULT_NULL_PERCENTAGE, + record -> { + record.set(stringOrdinal, null); + record.set(binaryOrdinal, null); + return record; + }); + try (FileAppender writer = parquetV2WriterWithoutDictionary(schema, dataFile)) { + writer.addAll(data); + } + + assertRecordsMatch(schema, 100, data, dataFile, false, BATCH_SIZE); + } + @Test public void testUnsupportedReadsForParquetV2() throws Exception { // Some types use delta encoding and which are not supported for vectorized reads