diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html
index 2841618e7c34..e1bba71c6ed0 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1685,7 +1685,7 @@
target-file-row-num |
9223372036854775807 |
Long |
- 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 support this option and fail fast when it is enabled. Disabled by default. |
+ 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-evolution append tables; its primary-key, blob and vector writers still fail fast when it is enabled. Disabled by default. |
target-file-size |
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 e0c7fd4004c7..96e56d818da3 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -830,8 +830,9 @@ public InlineElement getDescription() {
+ "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 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 20ee9ddb8c25..95084ea7a619 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 000000000000..2dffcf493b41
--- /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/write/file_store_write.py b/paimon-python/pypaimon/write/file_store_write.py
index 92fa5259d7ae..0cdd84664c4e 100644
--- a/paimon-python/pypaimon/write/file_store_write.py
+++ b/paimon-python/pypaimon/write/file_store_write.py
@@ -101,9 +101,20 @@ def _create_data_writer(self, partition: Tuple, bucket: int, options: CoreOption
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/writer/data_writer.py b/paimon-python/pypaimon/write/writer/data_writer.py
index 221f0f049885..3a191bc3f7da 100644
--- a/paimon-python/pypaimon/write/writer/data_writer.py
+++ b/paimon-python/pypaimon/write/writer/data_writer.py
@@ -52,6 +52,10 @@ def __init__(self, table, partition: Tuple, bucket: int, max_seq_number: int, op
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 @@ def _append_file_sequence_range(self, row_count: int) -> Tuple[int, int]:
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: