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
commit 9a0dfdde78f1a34f0167e37239b488d1c9f621d9 Author: JingsongLi <[email protected]> AuthorDate: Tue Oct 20 08:41:29 2026 +0800 [hotfix] Fix Python CI regressions on master --- .../pypaimon/tests/native_row_sidecar_test.py | 35 ++++++++++++---------- paimon-python/pypaimon/tests/native_update_test.py | 7 ++--- .../tests/native_upsert_key_normalization_test.py | 10 +++++-- .../tests/native_write_path_completion_test.py | 4 +-- paimon-python/pypaimon/tests/table_update_test.py | 1 + .../pypaimon/tests/test_write_merge_buffer.py | 4 +++ .../pypaimon/tests/write/table_write_test.py | 2 +- paimon-python/pypaimon/write/native_update.py | 8 +++++ paimon-python/pypaimon/write/table_update.py | 3 ++ paimon-python/pypaimon/write/table_write.py | 6 +++- 10 files changed, 54 insertions(+), 26 deletions(-) diff --git a/paimon-python/pypaimon/tests/native_row_sidecar_test.py b/paimon-python/pypaimon/tests/native_row_sidecar_test.py index eb62f5cb97..3b4b83c499 100644 --- a/paimon-python/pypaimon/tests/native_row_sidecar_test.py +++ b/paimon-python/pypaimon/tests/native_row_sidecar_test.py @@ -360,23 +360,28 @@ def test_old_sidecar_schema_evolves_by_field_id(tmp_path, external): assert actual.schema.field('id').type == pa.int64() -def test_bitmap_and_limit_keep_primary_without_requested_row_ranges(tmp_path): [email protected]('mode', ['bitmap', 'limit']) +def test_bitmap_and_limit_select_expected_source_without_requested_row_ranges(tmp_path, mode): table = _table(tmp_path, options={'file-index.bitmap.columns': 'id'}) messages = _append(table) - for message in messages: - for file in message.new_files: - os.remove(next(path for path in file.collect_files() if path.endswith('.row'))) - for mode in ['bitmap', 'limit']: - builder = table.new_read_builder() - if mode == 'bitmap': - builder.with_filter(builder.new_predicate_builder().equal('id', 5)) - else: - builder.with_limit(1) - read = builder.new_read() - with patch.object(read, '_create_split_read', side_effect=AssertionError('Python fallback')): - actual = read.to_arrow(builder.new_scan().plan().splits()) - expected_id = 5 if mode == 'bitmap' else 0 - assert actual.to_pydict() == {'id': [expected_id], 'value': ['v%d' % expected_id]} + if mode == 'bitmap': + # A selective bitmap now chooses ROW even without caller row ranges. + _remove_primaries(messages) + else: + # LIMIT alone still reads the primary Parquet file. + for message in messages: + for file in message.new_files: + os.remove(next(path for path in file.collect_files() if path.endswith('.row'))) + builder = table.new_read_builder() + if mode == 'bitmap': + builder.with_filter(builder.new_predicate_builder().equal('id', 5)) + else: + builder.with_limit(1) + read = builder.new_read() + with patch.object(read, '_create_split_read', side_effect=AssertionError('Python fallback')): + actual = read.to_arrow(builder.new_scan().plan().splits()) + expected_id = 5 if mode == 'bitmap' else 0 + assert actual.to_pydict() == {'id': [expected_id], 'value': ['v%d' % expected_id]} @pytest.mark.parametrize('option', ['blob-descriptor-field', 'blob.stored-descriptor-fields']) diff --git a/paimon-python/pypaimon/tests/native_update_test.py b/paimon-python/pypaimon/tests/native_update_test.py index 709c79d540..d7e4ec1bac 100644 --- a/paimon-python/pypaimon/tests/native_update_test.py +++ b/paimon-python/pypaimon/tests/native_update_test.py @@ -1015,8 +1015,8 @@ def test_update_dispatch_preserves_existing_partition_paths(tmp_path, layout, op native_calls.append(operation) return original(*args, **kwargs) - # Check a second update too: fallback outputs must remain readable on the - # next operation, and canonical files must continue using native updates. + # Check a second update too: outputs must remain readable and all supported + # partition layouts must continue using native updates. with patch.object(wrapper, method, track): for value in [11, 12]: builder = table.new_batch_write_builder() @@ -1029,5 +1029,4 @@ def test_update_dispatch_preserves_existing_partition_paths(tmp_path, layout, op read = table.new_read_builder() actual = read.new_read().to_arrow(read.new_scan().plan().splits()).sort_by('id') assert actual.to_pydict() == {'id': [1, 2], 'part': parts, 'value': [value, value]} - expected_calls = 0 if layout == 'external-binary' else 2 - assert len(native_calls) == expected_calls + assert len(native_calls) == 2 diff --git a/paimon-python/pypaimon/tests/native_upsert_key_normalization_test.py b/paimon-python/pypaimon/tests/native_upsert_key_normalization_test.py index aa9c186e85..d1ad0a63d0 100644 --- a/paimon-python/pypaimon/tests/native_upsert_key_normalization_test.py +++ b/paimon-python/pypaimon/tests/native_upsert_key_normalization_test.py @@ -332,10 +332,14 @@ def _dictionary_array(values, value_type): @pytest.mark.parametrize('native', [False, True], ids=['python', 'native']) -def test_untyped_null_keys_match_only_existing_null_key_rows(tmp_path, native): - table, _ = _table(tmp_path, native, pa.int32(), 1, 2) [email protected]('stream', [False, True], ids=['batch', 'stream']) [email protected]('partitioned', [False, True]) +def test_untyped_null_keys_match_only_existing_null_key_rows(tmp_path, native, stream, partitioned): + table, _ = _table(tmp_path, native, pa.int32(), 1, 2, partitioned=partitioned) source = pa.table({'key': pa.nulls(1), 'value': pa.array([9], type=pa.int32())}) - _upsert(table, source, native, False) + if partitioned: + source = source.append_column('pt', pa.array(['a'])) + _upsert(table, source, native, stream) for rows in _rows(table): assert [row['value'] for row in rows] == [0, 0, 0, 9] diff --git a/paimon-python/pypaimon/tests/native_write_path_completion_test.py b/paimon-python/pypaimon/tests/native_write_path_completion_test.py index 760d735978..90b7324db2 100644 --- a/paimon-python/pypaimon/tests/native_write_path_completion_test.py +++ b/paimon-python/pypaimon/tests/native_write_path_completion_test.py @@ -470,9 +470,9 @@ def test_postpone_blob_row_bridge_keeps_sources_until_prepare(fail): fields = [DataField(0, 'id', AtomicType('INT')), DataField(1, 'pt', AtomicType('STRING')), DataField(2, 'payload', AtomicType('BLOB'))] table = SimpleNamespace( - table_schema=SimpleNamespace(fields=fields), fields=fields, + table_schema=SimpleNamespace(fields=fields, primary_keys=['id', 'pt']), fields=fields, field_dict={field.name: field for field in fields}, partition_keys=['pt'], - is_primary_key_table=True, + is_primary_key_table=True, cross_partition_update=False, options=CoreOptions(Options({'bucket': '-2', 'postpone.target-row-num-per-bucket': '1'}))) backend = Mock() staged, payloads = [], [] diff --git a/paimon-python/pypaimon/tests/table_update_test.py b/paimon-python/pypaimon/tests/table_update_test.py index 552d3d0d29..ddbc4d9a82 100644 --- a/paimon-python/pypaimon/tests/table_update_test.py +++ b/paimon-python/pypaimon/tests/table_update_test.py @@ -50,6 +50,7 @@ def test_write_builder_creates_row_id_updater_with_commit_identity( stream, commit_identifier): table = mock.Mock() table.options.commit_user_prefix.return_value = None + table.catalog_environment.table_query_auth.return_value = None files_info = mock.Mock( snapshot_id=42, first_row_ids=[0], diff --git a/paimon-python/pypaimon/tests/test_write_merge_buffer.py b/paimon-python/pypaimon/tests/test_write_merge_buffer.py index d24c65cb73..8be5850641 100644 --- a/paimon-python/pypaimon/tests/test_write_merge_buffer.py +++ b/paimon-python/pypaimon/tests/test_write_merge_buffer.py @@ -27,6 +27,7 @@ spinning up the real catalog/write stack. """ import unittest +from types import SimpleNamespace from unittest.mock import Mock import pyarrow as pa @@ -86,6 +87,9 @@ class _Harness(KeyValueDataWriter): def __init__(self, merge_function, target_file_size: int = 10 ** 12): self.options = CoreOptions(Options({})) + self.table = SimpleNamespace( + is_primary_key_table=True, cross_partition_update=False, + options=self.options, table_schema=SimpleNamespace(primary_keys=['id'])) self.trimmed_primary_keys = ['id'] self._merge_function = merge_function # Large enough that ``_check_and_roll_if_needed`` does not diff --git a/paimon-python/pypaimon/tests/write/table_write_test.py b/paimon-python/pypaimon/tests/write/table_write_test.py index 47ddaef1f5..ef5bc57350 100644 --- a/paimon-python/pypaimon/tests/write/table_write_test.py +++ b/paimon-python/pypaimon/tests/write/table_write_test.py @@ -830,7 +830,7 @@ class TableWriteTest(unittest.TestCase): read_builder = table.new_read_builder() table_read = read_builder.new_read() splits = read_builder.new_scan().plan().splits() - actual = table_read.to_arrow(splits) + actual = table_read.to_arrow(splits).sort_by('user_id') self.assertEqual(expect.cast(read_arrow_schema(table.fields)), actual) def test_postpone_file_store_write_validates_runtime_bucket_count(self): diff --git a/paimon-python/pypaimon/write/native_update.py b/paimon-python/pypaimon/write/native_update.py index dd1bb33b83..61bc0d6ac6 100644 --- a/paimon-python/pypaimon/write/native_update.py +++ b/paimon-python/pypaimon/write/native_update.py @@ -310,6 +310,14 @@ def create_native_upsert(table, commit_user, data, keys, columns): # surviving payloads before applying row-ID update or append rules; # validating all payload types here would reject valid matched updates. key_names = list(dict.fromkeys(keys + table.partition_keys)) + for name in key_names: + field = data.schema.field(name) + if pa.types.is_null(field.type): + # An untyped NULL has the same match identity as a typed NULL. + # Give core the declared key layout without casting payloads. + data = data.set_column( + data.schema.get_field_index(name), field.with_type(schema.field(name).type), + pa.nulls(data.num_rows, type=schema.field(name).type)) if not all(_native_upsert_key_compatible(data.column(name), schema.field(name).type) for name in key_names): return None diff --git a/paimon-python/pypaimon/write/table_update.py b/paimon-python/pypaimon/write/table_update.py index ec881bf65e..c2fcf180c9 100644 --- a/paimon-python/pypaimon/write/table_update.py +++ b/paimon-python/pypaimon/write/table_update.py @@ -142,6 +142,9 @@ class TableUpdate: def _new_row_id_updater( self, commit_identifier: int, _precomputed_files_info=None ) -> TableUpdateByRowId: + # Authorize outside the native preparation fallback so a denied or + # restricted grant cannot be swallowed and retried by another reader. + _authorize_update(self.table) if _precomputed_files_info is None: from pypaimon.write.native_update import create_native_update_by_row_id try: diff --git a/paimon-python/pypaimon/write/table_write.py b/paimon-python/pypaimon/write/table_write.py index 5f6f35e2b4..dfefe3adcc 100644 --- a/paimon-python/pypaimon/write/table_write.py +++ b/paimon-python/pypaimon/write/table_write.py @@ -151,7 +151,11 @@ class TableWrite: if write_cols is not None: # Keep routing columns available for full-schema inputs, then # write only the columns and order recorded in the file metadata. - data = data.select(write_cols) + # RecordBatch.select is unavailable in Arrow 6 (Python 3.6). + data = pa.RecordBatch.from_arrays( + [data.column(name) for name in write_cols], + schema=pa.schema([data.schema.field(name) for name in write_cols], + metadata=data.schema.metadata)) self.file_store_write.write(partition, bucket, data) def _write_precomputed_bucket(self, data, bucket):
