[python] Support row-count based file rolling for data-evolution append tables

docs/generated/core_configuration.html

@@ -1685,7 +1685,7 @@
             <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 support this option and fail fast when it is enabled. Disabled by default.</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 supports this for data-evolution append tables; its primary-key, blob and vector writers still fail fast when it is enabled. Disabled by default.</td>
         </tr>
         <tr>
             <td><h5>target-file-size</h5></td>

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<Double> COMPACTION_SMALL_FILE_RATIO =
             key("compaction.small-file-ratio")

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."
         )
     )
 

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()

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)

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: