(paimon) branch master updated: [format] Introduce byte stream split for writing to parquet (#9499)

[lzljs3620320]

This is an automated email from the ASF dual-hosted git repository.

JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new 10fee641c9 [format] Introduce byte stream split for writing to parquet 
(#9499)
10fee641c9 is described below

commit 10fee641c9c20ccd049a3af44fe6f3d05e6e3f54
Author: Arnav Balyan <60175178+arnavbal...@users.noreply.github.com>
AuthorDate: Wed Sep 2 12:13:51 2026 +0530

    [format] Introduce byte stream split for writing to parquet (#9499)
---
 .../parquet/writer/RowDataParquetBuilder.java      |  6 ++++
 .../format/parquet/ParquetFormatReadWriteTest.java | 41 ++++++++++++++++++++++
 2 files changed, 47 insertions(+)

diff --git 
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/RowDataParquetBuilder.java
 
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/RowDataParquetBuilder.java
index 10adbfe35b..08358c4a62 100644
--- 
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/RowDataParquetBuilder.java
+++ 
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/RowDataParquetBuilder.java
@@ -39,6 +39,8 @@ import java.util.function.Supplier;
 /** A {@link ParquetBuilder} for {@link InternalRow}. */
 public class RowDataParquetBuilder implements 
MetadataParquetBuilder<InternalRow> {
 
+    private static final String ENABLE_BYTE_STREAM_SPLIT = 
"parquet.enable.bytestreamsplit";
+
     private final RowType rowType;
     private final Configuration conf;
 
@@ -96,6 +98,10 @@ public class RowDataParquetBuilder implements 
MetadataParquetBuilder<InternalRow
                                 conf.getBoolean(
                                         ParquetOutputFormat.ENABLE_DICTIONARY,
                                         
ParquetProperties.DEFAULT_IS_DICTIONARY_ENABLED))
+                        .withByteStreamSplitEncoding(
+                                conf.getBoolean(
+                                        ENABLE_BYTE_STREAM_SPLIT,
+                                        
ParquetProperties.DEFAULT_IS_BYTE_STREAM_SPLIT_ENABLED))
                         
.withValidation(conf.getBoolean(ParquetOutputFormat.VALIDATION, false))
                         .withWriterVersion(
                                 ParquetProperties.WriterVersion.fromString(
diff --git 
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFormatReadWriteTest.java
 
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFormatReadWriteTest.java
index 2784f157a1..eb393be3c6 100644
--- 
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFormatReadWriteTest.java
+++ 
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFormatReadWriteTest.java
@@ -40,6 +40,7 @@ import org.apache.paimon.reader.RecordReader;
 import org.apache.paimon.types.DataTypes;
 import org.apache.paimon.types.RowType;
 
+import org.apache.parquet.column.Encoding;
 import org.apache.parquet.column.values.bloomfilter.BloomFilter;
 import org.apache.parquet.hadoop.ParquetFileReader;
 import org.apache.parquet.hadoop.metadata.BlockMetaData;
@@ -52,6 +53,7 @@ import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -292,4 +294,43 @@ public class ParquetFormatReadWriteTest extends 
FormatReadWriteTest {
                     .containsEntry("name", CompressionCodecName.UNCOMPRESSED);
         }
     }
+
+    @Test
+    public void testWriteByteStreamSplit() throws Exception {
+        Options options = new Options();
+        options.set("parquet.enable.dictionary", "false");
+        options.set("parquet.enable.bytestreamsplit", "true");
+        ParquetFileFormat format =
+                new ParquetFileFormat(new 
FileFormatFactory.FormatContext(options, 1024, 1024));
+        RowType rowType =
+                DataTypes.ROW(
+                        DataTypes.FIELD(0, "float_value", DataTypes.FLOAT()),
+                        DataTypes.FIELD(1, "double_value", 
DataTypes.DOUBLE()));
+
+        write(
+                format.createWriterFactory(rowType),
+                file,
+                GenericRow.of(1.25f, 2.5d),
+                GenericRow.of(3.75f, 5.0d));
+
+        try (ParquetFileReader reader =
+                ParquetUtil.getParquetReader(
+                        fileIO, file, fileIO.getFileSize(file), new 
Options())) {
+            for (ColumnChunkMetaData column : 
reader.getFooter().getBlocks().get(0).getColumns()) {
+                
Assertions.assertThat(column.getEncodings()).contains(Encoding.BYTE_STREAM_SPLIT);
+            }
+        }
+
+        try (RecordReader<InternalRow> reader =
+                format.createReaderFactory(rowType, rowType, 
java.util.Collections.emptyList())
+                        .createReader(
+                                new FormatReaderContext(
+                                        fileIO, file, 
fileIO.getFileSize(file), null, null))) {
+            InternalRowSerializer serializer = new 
InternalRowSerializer(rowType);
+            List<InternalRow> rows = new ArrayList<>();
+            reader.forEachRemaining(row -> rows.add(serializer.copy(row)));
+            Assertions.assertThat(rows)
+                    .containsExactly(GenericRow.of(1.25f, 2.5d), 
GenericRow.of(3.75f, 5.0d));
+        }
+    }
 }