Skip to content

[format] Support size based ORC memory protection for large rows #9534

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ArnavBalyan wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[format] Support size based ORC memory protection for large rows #9534

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions paimon-format/src/main/java/org/apache/orc/OrcConf.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
10 changes: 8 additions & 2 deletions paimon-format/src/main/java/org/apache/orc/impl/WriterImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);

Expand Down Expand Up @@ -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) {

@JingsongLi JingsongLi Sep 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Honor sub-1 stripe-size ratios

This new gate says a stripe is checked once estimateMemory() exceeds ratio * orc.stripe.size, but the unchanged flush condition below still compares only with memoryLimit (normally the full stripe size). For example, with ratio 0.5 and an unscaled writer, this branch starts firing at half a stripe, resets rowsSinceCheck on every subsequent batch, yet does not flush until the estimate exceeds the full stripe size. That both violates the option description and adds an estimate on every batch in between. Please either flush against the configured threshold (while still respecting a lower MemoryManager limit), or reject/document ratios below 1 if they are not supported.

rowsSinceCheck = 0;
long size = treeWriter.estimateMemory();
if (LOG.isDebugEnabled()) {
LOG.debug(
"ORC writer "
Expand Down
58 changes: 58 additions & 0 deletions paimon-format/src/test/java/org/apache/paimon/format/orc/writer/OrcBulkWriterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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<InternalRow> 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();
Expand Down
Loading