From 8725301466f27984de0a2b07f3f10ad9625b4d88 Mon Sep 17 00:00:00 2001 From: Arnav Balyan Date: Tue, 1 Sep 2026 19:36:54 +0530 Subject: [PATCH] update --- .../src/main/java/org/apache/orc/OrcConf.java | 6 ++ .../java/org/apache/orc/impl/WriterImpl.java | 10 +++- .../format/orc/writer/OrcBulkWriterTest.java | 58 +++++++++++++++++++ 3 files changed, 72 insertions(+), 2 deletions(-) diff --git a/paimon-format/src/main/java/org/apache/orc/OrcConf.java b/paimon-format/src/main/java/org/apache/orc/OrcConf.java index 22eb39c5c384..0410f0a09090 100644 --- a/paimon-format/src/main/java/org/apache/orc/OrcConf.java +++ b/paimon-format/src/main/java/org/apache/orc/OrcConf.java @@ -281,6 +281,12 @@ public enum OrcConf { + "added to all of the writers. Valid range is [1,10000] and is primarily meant for" + "testing. Setting this too low may negatively affect performance." + " Use orc.stripe.row.count instead if the value larger than orc.stripe.row.count."), + STRIPE_SIZE_CHECKRATIO( + "orc.stripe.size.check.ratio", + "orc.stripe.size.check.ratio", + 0.0, + "Flush stripe if the tree writer size in bytes is larger than " + + "(this * orc.stripe.size). Use 0 to disable this check."), OVERWRITE_OUTPUT_FILE( "orc.overwrite.output.file", "orc.overwrite.output.file", diff --git a/paimon-format/src/main/java/org/apache/orc/impl/WriterImpl.java b/paimon-format/src/main/java/org/apache/orc/impl/WriterImpl.java index c6b027bfa504..fec89035c6c7 100644 --- a/paimon-format/src/main/java/org/apache/orc/impl/WriterImpl.java +++ b/paimon-format/src/main/java/org/apache/orc/impl/WriterImpl.java @@ -113,6 +113,7 @@ public class WriterImpl implements WriterInternal, MemoryManager.Callback { private long previousAllocation = -1; private long memoryLimit; private final long rowsPerCheck; + private final double stripeSizePerCheck; private long rowsSinceCheck = 0; private final OrcFile.Version version; private final Configuration conf; @@ -221,6 +222,8 @@ public WriterImpl(FileSystem fs, Path path, OrcFile.WriterOptions opts) throws I this.stripeRowCount = opts.getStripeRowCountValue(); this.stripeSize = opts.getStripeSize(); memoryLimit = stripeSize; + double stripeSizeCheckRatio = OrcConf.STRIPE_SIZE_CHECKRATIO.getDouble(conf); + stripeSizePerCheck = stripeSizeCheckRatio <= 0 ? 0 : stripeSizeCheckRatio * stripeSize; memoryManager = opts.getMemoryManager(); memoryManager.addWriter(path, stripeSize, this); @@ -321,9 +324,12 @@ public boolean checkMemory(double newScale) throws IOException { } private boolean checkMemory() throws IOException { - if (rowsSinceCheck >= rowsPerCheck) { + long size = + rowsSinceCheck < rowsPerCheck && stripeSizePerCheck == 0 + ? 0 + : treeWriter.estimateMemory(); + if (rowsSinceCheck >= rowsPerCheck || size > stripeSizePerCheck) { rowsSinceCheck = 0; - long size = treeWriter.estimateMemory(); if (LOG.isDebugEnabled()) { LOG.debug( "ORC writer " diff --git a/paimon-format/src/test/java/org/apache/paimon/format/orc/writer/OrcBulkWriterTest.java b/paimon-format/src/test/java/org/apache/paimon/format/orc/writer/OrcBulkWriterTest.java index 4690437c6a3b..4b7bc14d67a4 100644 --- a/paimon-format/src/test/java/org/apache/paimon/format/orc/writer/OrcBulkWriterTest.java +++ b/paimon-format/src/test/java/org/apache/paimon/format/orc/writer/OrcBulkWriterTest.java @@ -19,19 +19,26 @@ package org.apache.paimon.format.orc.writer; import org.apache.paimon.CoreOptions; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; import org.apache.paimon.format.FileFormat; +import org.apache.paimon.format.FormatReaderContext; import org.apache.paimon.format.FormatWriter; import org.apache.paimon.format.FormatWriterFactory; import org.apache.paimon.format.orc.OrcFileFormat; +import org.apache.paimon.format.orc.OrcReaderFactory; import org.apache.paimon.format.orc.OrcWriterFactory; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.PositionOutputStream; import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.Options; +import org.apache.paimon.reader.RecordReader; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowType; +import org.apache.orc.Reader; import org.assertj.core.api.Assertions; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -40,6 +47,57 @@ class OrcBulkWriterTest { + @Test + void testStripeSizeCheckRatio(@TempDir java.nio.file.Path tempDir) throws IOException { + Options options = new Options(); + options.set(CoreOptions.WRITE_BATCH_SIZE, 128); + options.set("orc.stripe.size", "65536"); + options.set("orc.rows.between.memory.checks", "5000"); + options.set("orc.stripe.size.check.ratio", "1"); + options.set("orc.column.encoding.direct", "payload"); + FileFormat orc = FileFormat.fromIdentifier("orc", options); + + RowType rowType = + RowType.builder() + .field("id", DataTypes.INT()) + .field("payload", DataTypes.STRING()) + .build(); + Path path = new Path(tempDir.toUri().toString(), "large-rows.orc"); + LocalFileIO fileIO = LocalFileIO.create(); + String payload = new String(new char[2048]).replace('\0', 'x'); + int rowCount = 512; + + try (PositionOutputStream out = fileIO.newOutputStream(path, false); + FormatWriter writer = orc.createWriterFactory(rowType).create(out, "none")) { + for (int i = 0; i < rowCount; i++) { + writer.addElement(GenericRow.of(i, BinaryString.fromString(i + "-" + payload))); + } + } + + try (Reader reader = + OrcReaderFactory.createReader( + new org.apache.hadoop.conf.Configuration(false), fileIO, path, null)) { + Assertions.assertThat(reader.getStripes()).hasSizeGreaterThan(1); + Assertions.assertThat(reader.getNumberOfRows()).isEqualTo(rowCount); + } + + int[] actualRowCount = {0}; + try (RecordReader reader = + orc.createReaderFactory(rowType, rowType, null) + .createReader( + new FormatReaderContext( + fileIO, path, fileIO.getFileSize(path), null, null))) { + reader.forEachRemaining( + row -> { + int id = actualRowCount[0]++; + Assertions.assertThat(row.getInt(0)).isEqualTo(id); + Assertions.assertThat(row.getString(1).toString()) + .isEqualTo(id + "-" + payload); + }); + } + Assertions.assertThat(actualRowCount[0]).isEqualTo(rowCount); + } + @Test void testRowBatch(@TempDir java.nio.file.Path tempDir) throws IOException { Options options = new Options();