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 d28225662f [python] Reject unsupported aggregation options before
writing (#10165)
d28225662f is described below
commit d28225662fea4cb546c29ced41e821cd551bbdfc
Author: chaoyang <[email protected]>
AuthorDate: Sat Oct 3 17:13:08 2026 +0800
[python] Reject unsupported aggregation options before writing (#10165)
---
.../pypaimon/tests/test_aggregation_e2e.py | 71 ++++++++++++++++++++--
.../pypaimon/tests/test_partial_update_e2e.py | 22 +++----
paimon-python/pypaimon/write/file_store_write.py | 16 ++---
3 files changed, 80 insertions(+), 29 deletions(-)
diff --git a/paimon-python/pypaimon/tests/test_aggregation_e2e.py
b/paimon-python/pypaimon/tests/test_aggregation_e2e.py
index 2d893e1341..31968d0d78 100644
--- a/paimon-python/pypaimon/tests/test_aggregation_e2e.py
+++ b/paimon-python/pypaimon/tests/test_aggregation_e2e.py
@@ -32,10 +32,12 @@ identifiers) must raise ``NotImplementedError`` at TableRead
construction rather than silently fall back to a wrong answer.
"""
+import glob
import os
import shutil
import tempfile
import unittest
+from unittest.mock import patch
import pyarrow as pa
import pytest
@@ -233,12 +235,15 @@ class AggregationMergeEngineE2ETest(unittest.TestCase):
table = self._create_pk_table(
table_name, extra_options=extra_options
)
- rows = [{'id': 1, 'total': 1, 'max_score': 1, 'label': 'a'}]
- if error_type is ValueError:
- with self.assertRaises(error_type):
- self._write(table, rows)
- else:
- self._write(table, rows)
+ with self.assertRaises(error_type) as write_error:
+ self._write(table, [
+ {'id': 1, 'total': 10, 'max_score': 1, 'label': 'a'},
+ {'id': 1, 'total': 20, 'max_score': 2, 'label': 'b'},
+ ])
+ self.assertIn(expected_substring, str(write_error.exception))
+ self.assertIsNone(table.snapshot_manager().get_latest_snapshot())
+ self.assertEqual(glob.glob(
+ os.path.join(table.table_path, '**', '*.parquet'),
recursive=True), [])
rb = table.new_read_builder()
with self.assertRaises(error_type) as cm:
rb.new_read()
@@ -247,6 +252,7 @@ class AggregationMergeEngineE2ETest(unittest.TestCase):
self.assertIn('aggregation', msg)
self.assertIn(expected_substring, msg)
+ @pytest.mark.python_write
@pytest.mark.python_read
def test_remove_record_on_delete_rejected(self):
self._create_and_expect_unsupported(
@@ -255,6 +261,7 @@ class AggregationMergeEngineE2ETest(unittest.TestCase):
'aggregation.remove-record-on-delete',
)
+ @pytest.mark.python_write
@pytest.mark.python_read
def test_field_ignore_retract_rejected(self):
self._create_and_expect_unsupported(
@@ -263,6 +270,58 @@ class AggregationMergeEngineE2ETest(unittest.TestCase):
'fields.total.ignore-retract',
)
+ @pytest.mark.python_write
+ def test_unsupported_stream_write_rejected_at_construction(self):
+ table = self._create_pk_table(
+ 'agg_stream_reject', field_aggs={'total': 'sum'},
+ extra_options={'aggregation.remove-record-on-delete': 'true'})
+ with self.assertRaisesRegex(
+ NotImplementedError, 'aggregation.remove-record-on-delete'):
+ table.new_stream_write_builder().new_write()
+ self.assertIsNone(table.snapshot_manager().get_latest_snapshot())
+ self.assertEqual(glob.glob(
+ os.path.join(table.table_path, '**', '*.parquet'),
recursive=True), [])
+
+ @pytest.mark.python_write
+ def test_dynamic_bucket_rejected_before_index_creation(self):
+ from pypaimon.write.table_write import BatchTableWrite,
StreamTableWrite
+
+ for streaming in (False, True):
+ table = self._create_pk_table(
+ 'agg_dynamic_reject_{}'.format(streaming), extra_options={
+ 'bucket': '-1',
+ 'aggregation.remove-record-on-delete': 'true',
+ })
+ builder = (table.new_stream_write_builder() if streaming
+ else table.new_batch_write_builder())
+ writer_class = StreamTableWrite if streaming else BatchTableWrite
+ # A rejected writer must never create the bucket index maintainer,
+ # including callers that construct TableWrite directly.
+ with patch.object(table, 'create_row_key_extractor') as extractor:
+ for create in (builder.new_write, lambda: writer_class(table,
'test')):
+ with self.subTest(streaming=streaming, create=create):
+ with self.assertRaisesRegex(
+ NotImplementedError,
'aggregation.remove-record-on-delete'):
+ create()
+ extractor.assert_not_called()
+ self.assertIsNone(table.snapshot_manager().get_latest_snapshot())
+ self.assertEqual(glob.glob(os.path.join(table.table_path, 'index',
'*')), [])
+ self.assertEqual(glob.glob(
+ os.path.join(table.table_path, '**', '*.parquet'),
recursive=True), [])
+
+ @pytest.mark.python_write
+ def test_false_retract_options_remain_writable(self):
+ table = self._create_pk_table(
+ 'agg_false_retract', field_aggs={'total': 'sum'}, extra_options={
+ 'aggregation.remove-record-on-delete': 'false',
+ 'fields.total.ignore-retract': 'false',
+ })
+ self._write(table, [{'id': 1, 'total': 10}])
+ self._write(table, [{'id': 1, 'total': 20}])
+ # This tests write acceptance. Native read support for false-valued
+ # retract options is independent of the write-side guard.
+ self.assertEqual(table.snapshot_manager().get_latest_snapshot().id, 2)
+
def test_sequence_field_supported(self):
# Top-level sequence.field is honored by the aggregation engine:
# aggregators fold in sequence-field order, not file order. Here
diff --git a/paimon-python/pypaimon/tests/test_partial_update_e2e.py
b/paimon-python/pypaimon/tests/test_partial_update_e2e.py
index b94dcc239f..46d225934a 100644
--- a/paimon-python/pypaimon/tests/test_partial_update_e2e.py
+++ b/paimon-python/pypaimon/tests/test_partial_update_e2e.py
@@ -291,22 +291,14 @@ class PartialUpdateMergeEngineE2ETest(unittest.TestCase):
)
@pytest.mark.python_write
- def test_aggregation_engine_write_logs_fallback_warning(self):
- """The write-side fallback to deduplicate for unsupported engines
- is silent in terms of return value -- a ``logging.warning`` is
- the only signal that file contents will not match the table's
- declared semantics. Important when the same table is read back
- by a reader that honours the declared engine; the pypaimon
- read-side raise wouldn't fire there.
- """
- table = self._create_pk_table('agg_warning',
- merge_engine='aggregation')
- with self.assertLogs(
- 'pypaimon.write.file_store_write', level='WARNING') as cm:
+ def test_unsupported_aggregation_write_rejected(self):
+ table = self._create_pk_table(
+ 'agg_unsupported', merge_engine='aggregation',
+ extra_options={'aggregation.remove-record-on-delete': 'true'})
+ with self.assertRaisesRegex(
+ NotImplementedError, 'aggregation.remove-record-on-delete'):
self._write(table, [{'id': 1, 'a': 'x', 'b': None, 'c': None}])
- combined = '\n'.join(cm.output)
- self.assertIn('aggregation', combined)
- self.assertIn('deduplicate', combined)
+ self.assertIsNone(table.snapshot_manager().get_latest_snapshot())
# -- partial-update + out-of-scope option combinations ---------------
#
diff --git a/paimon-python/pypaimon/write/file_store_write.py
b/paimon-python/pypaimon/write/file_store_write.py
index 596cd8594d..28e9f8f67d 100644
--- a/paimon-python/pypaimon/write/file_store_write.py
+++ b/paimon-python/pypaimon/write/file_store_write.py
@@ -41,11 +41,16 @@ class FileStoreWrite:
def __init__(self, table, commit_user):
from pypaimon.table.file_store_table import FileStoreTable
+ from pypaimon.common.options.core_options import MergeEngine
from pypaimon.read.merge_engine_support import
check_sequence_field_supported
# TableWrite constructs this before the row-key extractor, whose
# dynamic bucket index must not retain hashes for rejected writes.
check_sequence_field_supported(table)
+ if table.is_primary_key_table and table.options.merge_engine() ==
MergeEngine.AGGREGATE:
+ from pypaimon.read.merge_engine_support import check_supported
+
+ check_supported(table)
self.table: FileStoreTable = table
self.data_writers: Dict[Tuple, DataWriter] = {}
@@ -219,14 +224,9 @@ class FileStoreWrite:
partial-update with no out-of-scope options) cannot drift
between sides.
- For wholly unsupported engines (``aggregation``) the writer
- falls back to ``DeduplicateMergeFunction`` so the flushed file
- still maintains the LSM "PK unique within a file" invariant.
- The read path's dispatch still raises ``NotImplementedError``,
- so the user gets an explicit error before they observe
- wrong-engine data; the fallback only narrows the damage to
- "file is deduped, not aggregated" rather than the silent
- multi-row-per-PK corruption that existed pre-PR.
+ Aggregation options are validated with the read-side guard at writer
+ construction. Unsupported configurations must not be committed using
+ fallback merge semantics that discard input values.
Partial-update with out-of-scope options (sequence-group,
per-field aggregator, ignore-delete, remove-record-on-*) does
