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 3120ac76f0 [python] Support row-count based file rolling for
data-evolution append tables (#8863)
3120ac76f0 is described below
commit 3120ac76f07abc6df71dcca4564de4363fe3e845
Author: XiaoHongbo <[email protected]>
AuthorDate: Tue Jul 28 21:12:03 2026 +0800
[python] Support row-count based file rolling for data-evolution append
tables (#8863)
---
docs/generated/core_configuration.html | 2 +-
.../main/java/org/apache/paimon/CoreOptions.java | 5 +-
.../pypaimon/common/options/core_options.py | 4 +-
.../tests/data_evolution_row_rolling_test.py | 131 +++++++++++++++++++++
.../pypaimon/tests/shard_table_updator_test.py | 76 ++++++++++++
paimon-python/pypaimon/write/file_store_write.py | 17 ++-
paimon-python/pypaimon/write/table_update.py | 4 +
paimon-python/pypaimon/write/writer/data_writer.py | 30 +++--
8 files changed, 252 insertions(+), 17 deletions(-)
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index a5a2412092..a4ea3baecd 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1739,7 +1739,7 @@ If the data size allocated for the sorting task is
uneven,which may lead to perf
<td><h5>target-file-row-num</h5></td>
<td style="word-wrap: break-word;">9223372036854775807</td>
<td>Long</td>
- <td>Target number of rows per newly written data file; a file
rolls when this or target-file-size is reached, whichever comes first. Enforced
at bundle granularity, so a bundled write may exceed it by up to one bundle.
Only constrains files at write time: compaction is size-based and may merge
into larger files, and data-evolution compaction still produces a single file.
Bounds per-file rows for wide columns to avoid data-evolution OOM. PyPaimon
file-store writers do not supp [...]
+ <td>Target number of rows per newly written data file; a file
rolls when this or target-file-size is reached, whichever comes first. Enforced
at bundle granularity, so a bundled write may exceed it by up to one bundle.
Only constrains files at write time: compaction is size-based and may merge
into larger files, and data-evolution compaction still produces a single file.
Bounds per-file rows for wide columns to avoid data-evolution OOM. PyPaimon
supports this for data-evoluti [...]
</tr>
<tr>
<td><h5>target-file-size</h5></td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 1a8bd6755e..92a62c344a 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -870,8 +870,9 @@ public class CoreOptions implements Serializable {
+ "compaction is size-based and may merge
into larger files, and "
+ "data-evolution compaction still
produces a single file. Bounds "
+ "per-file rows for wide columns to avoid
data-evolution OOM. "
- + "PyPaimon file-store writers do not
support this option and "
- + "fail fast when it is enabled. Disabled
by default.");
+ + "PyPaimon supports this for
data-evolution append tables; its "
+ + "primary-key, blob and vector writers
still fail fast when it "
+ + "is enabled. Disabled by default.");
public static final ConfigOption<Double> COMPACTION_SMALL_FILE_RATIO =
key("compaction.small-file-ratio")
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index 392ea653a0..2b168ee09f 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -396,8 +396,8 @@ class CoreOptions:
.default_value((1 << 63) - 1)
.with_description(
"Target number of rows per newly written data file. PyPaimon
format-table "
- "writers split files at this limit; file-store writers fail fast
when "
- "this option is enabled."
+ "and data-evolution append-table writers split files at this
limit; "
+ "primary-key, blob and vector writers fail fast when this option
is enabled."
)
)
diff --git a/paimon-python/pypaimon/tests/data_evolution_row_rolling_test.py
b/paimon-python/pypaimon/tests/data_evolution_row_rolling_test.py
new file mode 100644
index 0000000000..2dffcf493b
--- /dev/null
+++ b/paimon-python/pypaimon/tests/data_evolution_row_rolling_test.py
@@ -0,0 +1,131 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import os
+import shutil
+import tempfile
+import unittest
+import uuid
+
+import pyarrow as pa
+
+from pypaimon import CatalogFactory, Schema
+
+
+class DataEvolutionRowRollingTest(unittest.TestCase):
+ """Row-count based data file rolling (target-file-row-num) for
+ data-evolution append tables."""
+
+ pa_schema = pa.schema([
+ ('id', pa.int32()),
+ ('name', pa.string()),
+ ])
+ de_options = {
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ }
+
+ @classmethod
+ def setUpClass(cls):
+ cls.tempdir = tempfile.mkdtemp()
+ cls.catalog = CatalogFactory.create(
+ {'warehouse': os.path.join(cls.tempdir, 'warehouse')})
+ cls.catalog.create_database('default', True)
+
+ @classmethod
+ def tearDownClass(cls):
+ shutil.rmtree(cls.tempdir, ignore_errors=True)
+
+ def _create(self, options):
+ name = f'default.roll_{uuid.uuid4().hex[:8]}'
+ self.catalog.create_table(
+ name, Schema.from_pyarrow_schema(self.pa_schema, options=options),
+ False)
+ return self.catalog.get_table(name)
+
+ def _rows(self, n):
+ return pa.Table.from_pydict(
+ {'id': list(range(n)), 'name': [f'n{i}' for i in range(n)]},
+ schema=self.pa_schema)
+
+ def _write_files(self, table, data):
+ """Write one Arrow table and return the committed DataFileMeta list."""
+ wb = table.new_batch_write_builder()
+ tw = wb.new_write()
+ tw.write_arrow(data)
+ msgs = tw.prepare_commit()
+ files = [f for m in msgs for f in m.new_files]
+ wb.new_commit().commit(msgs)
+ tw.close()
+ return files
+
+ def _read_ids(self, table):
+ rb = table.new_read_builder()
+ return sorted(
+ rb.new_read().to_arrow(rb.new_scan().plan().splits())
+ ['id'].to_pylist())
+
+ def test_rolls_when_row_count_exceeds_limit(self):
+ table = self._create({**self.de_options, 'target-file-row-num': '3'})
+ files = self._write_files(table, self._rows(10))
+ # 10 rows, limit 3 -> 3 full files + a 1-row remainder.
+ self.assertEqual([1, 3, 3, 3], sorted(f.row_count for f in files))
+ self.assertEqual(list(range(10)), self._read_ids(table))
+
+ def test_exact_multiple_rolls_evenly(self):
+ table = self._create({**self.de_options, 'target-file-row-num': '3'})
+ files = self._write_files(table, self._rows(6))
+ self.assertEqual([3, 3], sorted(f.row_count for f in files))
+ self.assertEqual(list(range(6)), self._read_ids(table))
+
+ def test_below_limit_is_single_file(self):
+ table = self._create({**self.de_options, 'target-file-row-num': '100'})
+ files = self._write_files(table, self._rows(10))
+ self.assertEqual([10], [f.row_count for f in files])
+
+ def test_unset_option_does_not_roll_by_rows(self):
+ table = self._create(self.de_options)
+ files = self._write_files(table, self._rows(50))
+ # No row limit -> a small batch stays one file (size rolling only).
+ self.assertEqual([50], [f.row_count for f in files])
+
+ def test_oversized_row_rolls_by_itself(self):
+ # Each row exceeds target-file-size: the size trigger rolls every row
by
+ # itself even though target-file-row-num is larger.
+ table = self._create({
+ **self.de_options,
+ 'target-file-row-num': '3',
+ 'target-file-size': '100 b',
+ })
+ big = 'x' * 500
+ data = pa.Table.from_pydict(
+ {'id': list(range(4)), 'name': [big] * 4}, schema=self.pa_schema)
+ files = self._write_files(table, data)
+ self.assertEqual([1, 1, 1, 1], [f.row_count for f in files])
+ self.assertEqual(list(range(4)), self._read_ids(table))
+
+ def test_non_de_table_still_fails_fast(self):
+ table = self._create({'target-file-row-num': '3'})
+ wb = table.new_batch_write_builder()
+ tw = wb.new_write()
+ with self.assertRaisesRegex(
+ NotImplementedError, 'row-count based file rolling'):
+ tw.write_arrow(self._rows(4))
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/paimon-python/pypaimon/tests/shard_table_updator_test.py
b/paimon-python/pypaimon/tests/shard_table_updator_test.py
index 00760c033c..7efb58085c 100644
--- a/paimon-python/pypaimon/tests/shard_table_updator_test.py
+++ b/paimon-python/pypaimon/tests/shard_table_updator_test.py
@@ -652,5 +652,81 @@ class ShardTableUpdatorTest(unittest.TestCase):
)
+ def test_shard_update_ignores_target_file_row_num(self):
+ """Regression: a shard maps to exactly one output file, so
+ target-file-row-num must not split it. Before the fix the
+ directly-constructed AppendOnlyDataWriter honoured the option and
+ rolled the 5-row shard into three files, failing SingleWriter.end()
+ with "Should have one file."."""
+ table_schema = pa.schema([
+ ('a', pa.int32()),
+ ('b', pa.int32()),
+ ('c', pa.int32()),
+ ('d', pa.int32()),
+ ])
+ schema = Schema.from_pyarrow_schema(
+ table_schema,
+ options={'row-tracking.enabled': 'true', 'data-evolution.enabled':
'true'},
+ )
+ name = self._create_unique_table_name('row_num_rolling')
+ self.catalog.create_table(name, schema, False)
+ table = self.catalog.get_table(name)
+
+ # One 5-row file for a, b, c.
+ wb = table.new_batch_write_builder()
+ tw = wb.new_write().with_write_type(['a', 'b', 'c'])
+ tc = wb.new_commit()
+ tw.write_arrow(pa.Table.from_pydict({
+ 'a': [1, 2, 3, 4, 5],
+ 'b': [10, 20, 30, 40, 50],
+ 'c': [100, 200, 300, 400, 500],
+ }, schema=pa.schema([
+ ('a', pa.int32()), ('b', pa.int32()), ('c', pa.int32()),
+ ])))
+ tc.commit(tw.prepare_commit())
+ tw.close()
+ tc.close()
+
+ # Enable row-count rolling (limit 2). The shard still covers all 5
+ # rows and must emit a single file.
+ table = table.copy({'target-file-row-num': '2'})
+
+ wb = table.new_batch_write_builder()
+ upd = wb.new_update()
+ upd.with_read_projection(['a', 'b', 'c'])
+ upd.with_update_type(['d'])
+ shard = upd.new_shard_updator(0, 1)
+ reader = shard.arrow_reader()
+ for batch in iter(reader.read_next_batch, None):
+ a_ = batch.column('a').to_pylist()
+ b_ = batch.column('b').to_pylist()
+ c_ = batch.column('c').to_pylist()
+ shard.update_by_arrow_batch(pa.RecordBatch.from_pydict(
+ {'d': [c + b - a for a, b, c in zip(a_, b_, c_)]},
+ schema=pa.schema([('d', pa.int32())]),
+ ))
+
+ # Without the fix this raises "Should have one file."
+ commit_messages = shard.prepare_commit()
+ new_file_count = sum(len(m.new_files) for m in commit_messages)
+ self.assertEqual(
+ 1, new_file_count, "shard update must write exactly one file")
+
+ tc = wb.new_commit()
+ tc.commit(commit_messages)
+ tc.close()
+
+ rb = table.new_read_builder()
+ tr = rb.new_read()
+ actual = tr.to_arrow(rb.new_scan().plan().splits())
+ expected = pa.Table.from_pydict({
+ 'a': [1, 2, 3, 4, 5],
+ 'b': [10, 20, 30, 40, 50],
+ 'c': [100, 200, 300, 400, 500],
+ 'd': [109, 218, 327, 436, 545],
+ }, schema=table_schema)
+ self.assertEqual(actual, expected)
+
+
if __name__ == '__main__':
unittest.main()
diff --git a/paimon-python/pypaimon/write/file_store_write.py
b/paimon-python/pypaimon/write/file_store_write.py
index 92fa5259d7..0cdd84664c 100644
--- a/paimon-python/pypaimon/write/file_store_write.py
+++ b/paimon-python/pypaimon/write/file_store_write.py
@@ -101,9 +101,20 @@ class FileStoreWrite:
raise ValueError(
f"target-file-row-num should be at most {max_value}")
if row_limit != max_value:
- raise NotImplementedError(
- "target-file-row-num is set on this table but pypaimon
file-store writers do not support "
- "row-count based file rolling yet; unset it or write with
Java/Flink/Spark.")
+ # Row-count rolling is implemented in the base append writer only.
+ # DE (data-evolution) append tables are the target; primary-key,
+ # blob and vector writers override rolling and are not supported
yet.
+ row_rolling_supported = (
+ self.table.options.data_evolution_enabled()
+ and not self.table.is_primary_key_table
+ and not self._has_blob_columns()
+ and not (self._has_vector_columns()
+ and options.with_vector_format()))
+ if not row_rolling_supported:
+ raise NotImplementedError(
+ "target-file-row-num is set on this table but pypaimon
supports row-count "
+ "based file rolling only for data-evolution append tables
(no primary key, "
+ "blob or vector columns); unset it or write with
Java/Flink/Spark.")
def max_seq_number():
return self._seq_number_stats(partition).get(bucket, 1)
diff --git a/paimon-python/pypaimon/write/table_update.py
b/paimon-python/pypaimon/write/table_update.py
index 8095c16c2a..9f03f482e3 100644
--- a/paimon-python/pypaimon/write/table_update.py
+++ b/paimon-python/pypaimon/write/table_update.py
@@ -729,7 +729,11 @@ class ShardTableUpdator:
partition = item[0]
row_range = item[1]
writer = AppendOnlyDataWriter(self.table, partition, 0, 0,
self.table.options, self.write_cols)
+ # A shard maps to exactly one output file, so disable both size-
and
+ # row-count based rolling; otherwise target-file-row-num would
split
+ # the shard and SingleWriter.end() fails with "Should have one
file."
writer.target_file_size =
MemorySize.of_mebi_bytes(999999999).get_bytes()
+ writer.target_file_row_num =
CoreOptions.TARGET_FILE_ROW_NUM.default_value()
self.writer = SingleWriter(writer, partition, row_range.from_,
row_range.to - row_range.from_ + 1)
diff --git a/paimon-python/pypaimon/write/writer/data_writer.py
b/paimon-python/pypaimon/write/writer/data_writer.py
index 221f0f0498..3a191bc3f7 100644
--- a/paimon-python/pypaimon/write/writer/data_writer.py
+++ b/paimon-python/pypaimon/write/writer/data_writer.py
@@ -52,6 +52,10 @@ class DataWriter(ABC):
self.options = options
self.target_file_size =
self.options.target_file_size(self.table.is_primary_key_table)
+ # Roll a file when it reaches target_file_row_num rows or
target_file_size,
+ # whichever comes first. Defaults to the max long (disabled), so plain
+ # size-based rolling is unchanged unless the option is set.
+ self.target_file_row_num = self.options.target_file_row_num()
# POSTPONE_BUCKET uses AVRO format, otherwise default to PARQUET
default_format = (
CoreOptions.FILE_FORMAT_AVRO
@@ -188,16 +192,24 @@ class DataWriter(ABC):
def _check_and_roll_if_needed(self):
while self.pending_data is not None:
- current_size = self.pending_data.nbytes
- if current_size <= self.target_file_size:
+ num_rows = self.pending_data.num_rows
+ # Row-count trigger: keep at most target_file_row_num rows per
file.
+ split_row = num_rows
+ if num_rows > self.target_file_row_num:
+ split_row = self.target_file_row_num
+ # Size trigger: roll earlier if the size split point comes first.
+ if self.pending_data.nbytes > self.target_file_size:
+ size_split = self._find_optimal_split_point(
+ self.pending_data, self.target_file_size)
+ # First row alone exceeds target_file_size: roll it by itself.
+ if size_split <= 0:
+ size_split = 1
+ if size_split < split_row:
+ split_row = size_split
+ if split_row <= 0 or split_row >= num_rows:
break
- split_row = self._find_optimal_split_point(self.pending_data,
self.target_file_size)
- if split_row <= 0:
- break
- data_to_write = self.pending_data.slice(0, split_row)
- remaining_data = self.pending_data.slice(split_row)
- self._write_data_to_file(data_to_write)
- self.pending_data = remaining_data
+ self._write_data_to_file(self.pending_data.slice(0, split_row))
+ self.pending_data = self.pending_data.slice(split_row)
def _write_data_to_file(self, data: pa.Table):
if data.num_rows == 0:
