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 0bfd0fada1 [python] Use native IO for floating and binary partitions
(#10461)
0bfd0fada1 is described below
commit 0bfd0fada1f61b3f2b637962f5606f6be6ebf708
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Oct 9 16:32:09 2026 +0800
[python] Use native IO for floating and binary partitions (#10461)
---
paimon-python/pypaimon/index/dynamic_bucket.py | 2 +-
.../pypaimon/read/scanner/file_scanner.py | 2 +-
.../pypaimon/tests/deletion_vector_path_test.py | 74 +++---
.../pypaimon/tests/file_store_commit_test.py | 35 +--
.../pypaimon/tests/native_dynamic_write_test.py | 2 +-
.../pypaimon/tests/native_merge_into_test.py | 9 +-
.../pypaimon/tests/native_partition_types_test.py | 269 +++++++++++++++++++++
paimon-python/pypaimon/tests/native_update_test.py | 2 +-
.../pypaimon/tests/py36/rest_ao_read_write_test.py | 4 +-
paimon-python/pypaimon/tests/reader_base_test.py | 4 +-
paimon-python/pypaimon/tests/table_commit_test.py | 2 +-
.../pypaimon/utils/file_store_path_factory.py | 97 +++-----
paimon-python/pypaimon/write/file_store_commit.py | 36 +--
paimon-python/pypaimon/write/native_commit.py | 7 +
paimon-python/pypaimon/write/native_merge_into.py | 6 +-
paimon-python/pypaimon/write/native_update.py | 5 +-
paimon-python/pypaimon/write/native_write.py | 14 +-
paimon-python/pypaimon/write/table_delete.py | 2 +-
18 files changed, 407 insertions(+), 165 deletions(-)
diff --git a/paimon-python/pypaimon/index/dynamic_bucket.py
b/paimon-python/pypaimon/index/dynamic_bucket.py
index 707b03e724..89de907ea0 100644
--- a/paimon-python/pypaimon/index/dynamic_bucket.py
+++ b/paimon-python/pypaimon/index/dynamic_bucket.py
@@ -84,7 +84,7 @@ def validate_bucket_id(bucket: int) -> None:
def _iter_hashes(table, entry: IndexManifestEntry) -> Iterator[int]:
meta = entry.index_file
path = table.path_factory().bucket_index_path(
- tuple(entry.partition.values), entry.bucket, meta, table.file_io
+ tuple(entry.partition.values), entry.bucket, meta
)
with table.file_io.new_input_stream(path) as stream:
remainder = b""
diff --git a/paimon-python/pypaimon/read/scanner/file_scanner.py
b/paimon-python/pypaimon/read/scanner/file_scanner.py
index 503cc78d0a..a7eb538520 100755
--- a/paimon-python/pypaimon/read/scanner/file_scanner.py
+++ b/paimon-python/pypaimon/read/scanner/file_scanner.py
@@ -991,7 +991,7 @@ class FileScanner:
return deletion_files
dv_file_path = self.table.path_factory().bucket_index_path(
- tuple(index_entry.partition.values), index_entry.bucket,
index_file, self.table.file_io)
+ tuple(index_entry.partition.values), index_entry.bucket,
index_file)
# Convert each DeletionVectorMeta to DeletionFile
for data_file_name, dv_meta in index_file.dv_ranges.items():
diff --git a/paimon-python/pypaimon/tests/deletion_vector_path_test.py
b/paimon-python/pypaimon/tests/deletion_vector_path_test.py
index d1f53116e9..6592ddb5fd 100644
--- a/paimon-python/pypaimon/tests/deletion_vector_path_test.py
+++ b/paimon-python/pypaimon/tests/deletion_vector_path_test.py
@@ -157,7 +157,7 @@ def test_bucket_dv_requires_java_directory(tmp_path,
planner, wrong_directory):
file = entry.index_file
assert file.external_path is None
partition = tuple(entry.partition.values)
- path = factory.bucket_index_path(partition, entry.bucket, file,
table.file_io)
+ path = factory.bucket_index_path(partition, entry.bucket, file)
wrong_parent = (factory.index_path() if wrong_directory == 'table-index'
else factory.bucket_path(partition, entry.bucket))
wrong_path = wrong_parent + '/' + file.file_name
@@ -189,14 +189,14 @@ def
test_floating_bucket_dv_does_not_search_python_directory(tmp_path, file_uri,
factory._root = 'file://' + table.table_path
partition = tuple(entry.partition.values)
file = replace(entry.index_file, external_path=None)
- canonical = factory.bucket_index_path(partition, entry.bucket, file,
table.file_io)
+ canonical = factory.bucket_index_path(partition, entry.bucket, file)
python_path = factory.bucket_path(partition, entry.bucket) + '/' +
file.file_name
with table.file_io.new_input_stream(canonical) as stream:
data = stream.read()
table.file_io.delete_quietly(canonical)
with table.file_io.new_output_stream(python_path) as stream:
stream.write(data)
- assert factory.bucket_index_path(partition, entry.bucket, file,
table.file_io) == canonical
+ assert factory.bucket_index_path(partition, entry.bucket, file) ==
canonical
with patch.object(table, 'path_factory', return_value=factory):
with pytest.raises(FileNotFoundError, match=file.file_name):
_read(table, 'python', 2, [1, 2, 3])
@@ -266,33 +266,33 @@ def
test_bucket_dv_writes_java_typed_partition_directory(tmp_path, planner, part
_delete(table, [0])
entry = _entries(table, 2)[0]
file = entry.index_file
- assert (file.external_path is not None) ==
pa.types.is_floating(partition_type)
+ assert file.external_path is None
path = Path(table.table_path) / ('p=' + canonical_name) / ('bucket-' +
str(entry.bucket)) / file.file_name
assert path.is_file()
- if file.external_path is not None:
- # Java IndexInDataFileDirPathFactory uses the explicit path even when
- # this JDK would render the same float with different digits.
- assert Path(file.external_path).samefile(path)
- effective_planner = planner
- if planner == 'native' and pa.types.is_floating(partition_type):
- # Rust intentionally rejects floating partition formatting until it can
- # reproduce Java Float/Double.toString, including boundary values.
- with pytest.raises(NotImplementedError, match='type is not supported
as partition key'):
- native_plan(table)
- native_table = table.copy({'scan.native-plan.enabled': 'true'})
- scan = native_table.new_read_builder().new_scan()
- with patch('pypaimon.read.native_plan.native_plan', wraps=native_plan)
as native_call:
- with patch.object(scan.file_scanner, 'scan',
wraps=scan.file_scanner.scan) as fallback:
- plan = scan.plan()
- assert native_call.call_count == 1
- assert fallback.call_count == 1
- result =
native_table.new_read_builder().new_read().to_arrow(plan.splits())
- assert sorted(result.column('k').to_pylist()) == [1, 2, 3]
- effective_planner = 'python'
- _read(table, effective_planner, 2, [1, 2, 3])
+ _read(table, planner, 2, [1, 2, 3])
_delete(table, [1])
- _read(table, effective_planner, 3, [2, 3])
- _read(table, effective_planner, 2, [1, 2, 3])
+ _read(table, planner, 3, [2, 3])
+ _read(table, planner, 2, [1, 2, 3])
+
+
[email protected]('planner', _PLANNERS)
+def test_floating_bucket_dv_requires_current_java_directory(tmp_path, planner):
+ table = _table(tmp_path, 'bucket', 1e23, pa.float64())
+ _delete(table, [0])
+ entry = _entries(table, 2)[0]
+ canonical = (Path(table.table_path) / 'p=1.0E23' /
+ ('bucket-' + str(entry.bucket)) / entry.index_file.file_name)
+ alternate = (Path(table.table_path) / 'p=9.999999999999999E22' /
+ ('bucket-' + str(entry.bucket)) / entry.index_file.file_name)
+ alternate.parent.mkdir(parents=True, exist_ok=True)
+ canonical.rename(alternate)
+ expected = table.path_factory().bucket_index_path(
+ tuple(entry.partition.values), entry.bucket, entry.index_file)
+ assert expected == str(canonical)
+ with pytest.raises((FileNotFoundError, ValueError)) as error:
+ _read(table, planner, 2, [1, 2, 3])
+ assert entry.index_file.file_name in str(error.value)
+ assert alternate.is_file()
@pytest.mark.parametrize('planner', _PLANNERS)
@@ -304,14 +304,11 @@ def
test_bucket_dv_writes_java_typed_partition_directory(tmp_path, planner, part
(-0.0, pa.float32(), '-0.0'),
(1.4e-45, pa.float32(), '1.4E-45'),
(3.4028234663852886e38, pa.float32(), '3.4028235E38'),
- (1.17549435e-38, pa.float32(), '1.17549435E-38'),
(1.17549435e-38, pa.float32(), '1.1754944E-38'),
- (2.68873286e11, pa.float32(), '2.68873286E11'),
(2.68873286e11, pa.float32(), '2.6887329E11'),
- (9.64991956e24, pa.float32(), '9.6499195E24'),
- (1e23, pa.float64(), '9.999999999999999E22'),
+ (9.64991956e24, pa.float32(), '9.6499196E24'),
(1e23, pa.float64(), '1.0E23'),
- (-2.3345394554987242e17, pa.float64(), '-2.33453945549872416E17'),
+ (-2.3345394554987242e17, pa.float64(), '-2.3345394554987242E17'),
(5e-324, pa.float64(), '4.9E-324'),
(datetime(2026, 9, 15, 12), pa.timestamp('s'), '2026-09-15 12%3A00%3A00'),
(datetime(2026, 9, 15, 12), pa.timestamp('ms'), '2026-09-15
12%3A00%3A00.000'),
@@ -411,7 +408,7 @@ def _check_java_bucket_dv(tmp_path, planner, value,
partition_type, canonical_na
# Model the persisted layout produced by Java: the bucket path is canonical
# and no explicit index location is needed in its manifest metadata.
old_path = table.path_factory().bucket_index_path(
- tuple(old.partition.values), old.bucket, old.index_file, table.file_io)
+ tuple(old.partition.values), old.bucket, old.index_file)
with table.file_io.new_input_stream(old_path) as stream:
data = stream.read()
with table.file_io.new_output_stream(canonical_path) as stream:
@@ -434,16 +431,7 @@ def _check_java_bucket_dv(tmp_path, planner, value,
partition_type, canonical_na
old.index_file.file_name)
with table.file_io.new_output_stream(other_partition) as stream:
stream.write(b'not the requested floating partition')
- if planner == 'native' and partition_type is not None and
pa.types.is_floating(partition_type):
- # Rust does not support floating partitions; exercise the real adapter
- # fallback against the same Java-produced DV layout.
- table = table.copy({'scan.native-plan.enabled': 'true'})
- planner = 'python'
- with patch('pypaimon.read.native_plan.native_plan', wraps=native_plan)
as native_call:
- plan = _read(table, planner, 3, [1, 2, 3])
- assert native_call.call_count == 1
- else:
- plan = _read(table, planner, 3, [1, 2, 3])
+ plan = _read(table, planner, 3, [1, 2, 3])
paths = [dv.dv_index_path for split in plan.splits()
for dv in split.data_deletion_files or [] if dv is not None]
assert paths == [canonical_path]
diff --git a/paimon-python/pypaimon/tests/file_store_commit_test.py
b/paimon-python/pypaimon/tests/file_store_commit_test.py
index 39c0649c8c..e4c3d1d568 100644
--- a/paimon-python/pypaimon/tests/file_store_commit_test.py
+++ b/paimon-python/pypaimon/tests/file_store_commit_test.py
@@ -30,6 +30,7 @@ from pypaimon.manifest.schema.data_file_meta import
DataFileMeta
from pypaimon.manifest.schema.manifest_entry import ManifestEntry
from pypaimon.manifest.schema.manifest_file_meta import ManifestFileMeta
from pypaimon.manifest.schema.simple_stats import SimpleStats
+from pypaimon.schema.data_types import AtomicType, DataField
from pypaimon.snapshot.snapshot_commit import PartitionStatistics
from pypaimon.table.row.binary_row import BinaryRow
from pypaimon.table.row.generic_row import GenericRow, GenericRowSerializer
@@ -43,6 +44,7 @@ from pypaimon.write.file_store_commit import (
_reject_compact_increment,
_row_id_check_from_messages,
)
+from pypaimon.utils.file_store_path_factory import FileStorePathFactory
class TestRowIdCheckFromMessages(unittest.TestCase):
@@ -132,6 +134,9 @@ class TestFileStoreCommitRowTracking(unittest.TestCase):
def setUp(self):
self.mock_table = Mock()
self.mock_table.partition_keys = ['dt', 'region']
+ self.mock_table.path_factory.return_value = FileStorePathFactory(
+ '/test/table/path', ['dt', 'region'], '__DEFAULT_PARTITION__',
'parquet',
+ 'data-', 'changelog-', False, False, 'zstd')
self.mock_table.current_branch.return_value = 'main'
self.mock_table.table_path = '/test/table/path'
self.mock_table.file_io = Mock()
@@ -351,6 +356,13 @@ class TestFileStoreCommit(unittest.TestCase):
# Mock table with required attributes
self.mock_table = Mock()
self.mock_table.partition_keys = ['dt', 'region']
+ self.mock_table.partition_keys_fields = [
+ DataField(0, 'dt', AtomicType('STRING')),
+ DataField(1, 'region', AtomicType('STRING')),
+ ]
+ self.mock_table.path_factory.return_value = FileStorePathFactory(
+ '/test/table/path', ['dt', 'region'], '__DEFAULT_PARTITION__',
'parquet',
+ 'data-', 'changelog-', False, False, 'zstd')
self.mock_table.current_branch.return_value = 'main'
self.mock_table.table_path = '/test/table/path'
self.mock_table.file_io = Mock()
@@ -724,7 +736,7 @@ class TestFileStoreCommit(unittest.TestCase):
for old_buckets, new_buckets in [(-2, 2), (2, 3)]:
with self.subTest(old=old_buckets, new=new_buckets):
- partition = GenericRow(['2024-01-15', 'us-east-1'], None)
+ partition = GenericRow(['2024-01-15', 'us-east-1'],
self.mock_table.partition_keys_fields)
entries = [
ManifestEntry(
kind=1,
@@ -839,6 +851,7 @@ class TestFileStoreCommit(unittest.TestCase):
"""Test partition statistics generation for unpartitioned table."""
# Update mock table to have no partition keys
self.mock_table.partition_keys = []
+ self.mock_table.partition_keys_fields = []
# Create FileStoreCommit instance
file_store_commit = self._create_file_store_commit()
@@ -954,19 +967,8 @@ class TestFileStoreCommit(unittest.TestCase):
new_files=[file_meta]
)
- # Test method
- statistics =
file_store_commit._generate_partition_statistics(self._to_entries([commit_message]))
-
- # Verify results - should fallback to index-based naming
- self.assertEqual(len(statistics), 1)
-
- stat = statistics[0]
- expected_spec = {
- 'partition_0': '2024-01-15',
- 'partition_1': 'us-east-1',
- 'partition_2': 'extra-value'
- }
- self.assertEqual(stat.spec, expected_spec)
+ with self.assertRaisesRegex(ValueError, 'Partition row does not
match'):
+
file_store_commit._generate_partition_statistics(self._to_entries([commit_message]))
def test_generate_partition_statistics_empty_commit_messages(
self, mock_manifest_list_manager, mock_manifest_file_manager):
@@ -1111,11 +1113,10 @@ class TestFileStoreCommit(unittest.TestCase):
self.assertEqual(2, file_store_commit._try_commit_once.call_count)
file_store_commit._commit_retry_wait.assert_called_once_with(0)
- @staticmethod
- def _to_entries(commit_messages):
+ def _to_entries(self, commit_messages):
commit_entries = []
for msg in commit_messages:
- partition = GenericRow(list(msg.partition), None)
+ partition = GenericRow(list(msg.partition),
self.mock_table.partition_keys_fields)
for file in msg.new_files:
commit_entries.append(ManifestEntry(
kind=0,
diff --git a/paimon-python/pypaimon/tests/native_dynamic_write_test.py
b/paimon-python/pypaimon/tests/native_dynamic_write_test.py
index ef84c1b73b..47dd8205a9 100644
--- a/paimon-python/pypaimon/tests/native_dynamic_write_test.py
+++ b/paimon-python/pypaimon/tests/native_dynamic_write_test.py
@@ -118,7 +118,7 @@ def test_dynamic_cross_backend_restart(tmp_path,
bucket_local, first_native):
assert len(indexes) == 1
assert indexes[0].index_file.external_path is None
path = table.path_factory().bucket_index_path(
- tuple(indexes[0].partition.values), 0, indexes[0].index_file,
table.file_io)
+ tuple(indexes[0].partition.values), 0, indexes[0].index_file)
assert table.file_io.exists(path)
assert _indexes(table)[0].index_file.row_count == 5
expected = [{'id': i, 'p': 'a/b', 'v': v} for i, v in
diff --git a/paimon-python/pypaimon/tests/native_merge_into_test.py
b/paimon-python/pypaimon/tests/native_merge_into_test.py
index 1272f129ae..fe15bac7ca 100644
--- a/paimon-python/pypaimon/tests/native_merge_into_test.py
+++ b/paimon-python/pypaimon/tests/native_merge_into_test.py
@@ -508,7 +508,7 @@ def
test_python_written_partition_source_uses_java_paths_for_native_merge(tmp_pa
@pytest.mark.parametrize('native', [False, True])
@pytest.mark.parametrize('stream', [False, True])
@pytest.mark.parametrize('partition_type', [pa.float32(), pa.float64()])
-def test_float_partition_source_selects_python_before_native_merge(tmp_path,
native, stream, partition_type):
+def test_float_partition_source_uses_native_merge(tmp_path, native, stream,
partition_type):
target = _table(tmp_path / 'target', [dict(id=1, value=10,
name='old')]).copy({
'write.native.enabled': str(native).lower()})
schema = pa.schema(list(_SCHEMA) + [pa.field('part', partition_type)])
@@ -522,8 +522,11 @@ def
test_float_partition_source_selects_python_before_native_merge(tmp_path, nat
dict(id=2, value=22, name='insert',
part=2.25)])
builder = target.new_stream_write_builder() if stream else
target.new_batch_write_builder()
kwargs = dict(commit_identifier=57) if stream else {}
- with
patch('pypaimon.write.native_merge_into.NativeTableMergeInto.prepare_commit',
- side_effect=AssertionError('Unsupported source entered native
MERGE')):
+ from contextlib import ExitStack
+ with ExitStack() as stack:
+ if native:
+
stack.enter_context(patch('pypaimon.table.data_evolution_merge_into._build_tables',
+ side_effect=AssertionError('Python
source materialization')))
messages = builder.new_update().merge_into(
source, on=['id'], when_matched=[WhenMatched.update({'value':
source_col('value')})],
when_not_matched=[WhenNotMatched('*')], **kwargs)
diff --git a/paimon-python/pypaimon/tests/native_partition_types_test.py
b/paimon-python/pypaimon/tests/native_partition_types_test.py
new file mode 100644
index 0000000000..783f1540df
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_partition_types_test.py
@@ -0,0 +1,269 @@
+# 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.
+
+"""Java partition values across native append, PK and data-evolution writes."""
+
+from datetime import datetime, timedelta, timezone
+from unittest.mock import Mock, patch
+from urllib.parse import urlparse
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.write.native_write import NativeTableWrite
+
+
+_PARTITIONS = [
+ (pa.float32(), [1.5, 42., 0.0001, None], ['1.5', '42.0', '1.0E-4',
'__DEFAULT_PARTITION__']),
+ (pa.float64(), [2.25, 42., 1e23, None], ['2.25', '42.0', '1.0E23',
'__DEFAULT_PARTITION__']),
+ (pa.binary(), [b'a/b', '\u6c49\u5b57'.encode(), b' \t', None],
+ ['a%2Fb', '\u6c49\u5b57', '__DEFAULT_PARTITION__',
'__DEFAULT_PARTITION__']),
+ (pa.binary(3), [b'a/b', b'x=y', b' \t ', None],
+ ['a%2Fb', 'x%3Dy', '__DEFAULT_PARTITION__', '__DEFAULT_PARTITION__']),
+]
+
+
+def _table(tmp_path, partition_type, mode='append', native=True,
placement='local', extra_options=None):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path / 'warehouse')})
+ catalog.create_database('db', True)
+ options = {'partition.legacy-name': 'false', 'write.native.enabled':
str(native).lower(),
+ 'scan.native-plan.enabled': 'true', 'read.native.enabled':
'true'}
+ if mode == 'pk':
+ options['bucket'] = '1'
+ elif mode == 'evolution':
+ options.update({'data-evolution.enabled': 'true',
'row-tracking.enabled': 'true'})
+ if placement == 'directory':
+ options['data-file.path-directory'] = 'data'
+ elif placement == 'external':
+ options['data-file.external-paths'] = (tmp_path / 'external').as_uri()
+ options['data-file.external-paths.strategy'] = 'round-robin'
+ options.update(extra_options or {})
+ arrow_schema = pa.schema([('id', pa.int32()), ('p', partition_type),
('value', pa.int32())])
+ catalog.create_table('db.t', Schema.from_pyarrow_schema(
+ arrow_schema, partition_keys=['p'], primary_keys=['p', 'id'] if mode
== 'pk' else [],
+ options=options), False)
+ return catalog.get_table('db.t'), arrow_schema
+
+
+def _read(table):
+ builder = table.new_read_builder()
+ return
sorted(builder.new_read().to_arrow(builder.new_scan().plan().splits()).to_pylist(),
+ key=lambda row: row['id'])
+
+
+def _write(table, schema, rows, stream=False):
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ writer = builder.new_write()
+ try:
+ if table.options.native_write_enabled():
+ assert isinstance(writer, NativeTableWrite)
+ writer.write_arrow(pa.Table.from_pylist(rows, schema=schema))
+ messages = writer.prepare_commit(71) if stream else
writer.prepare_commit()
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages, 71) if stream else commit.commit(messages)
+ finally:
+ commit.close()
+ return messages
+ finally:
+ writer.close()
+
+
[email protected]_plan
[email protected]('partition_type,values,names', _PARTITIONS)
[email protected]('mode', ['append', 'pk', 'evolution'])
[email protected]('placement', ['local', 'directory', 'external'])
+def test_native_partition_write_read_and_file_paths(tmp_path, partition_type,
values, names, mode, placement):
+ table, schema = _table(tmp_path, partition_type, mode, placement=placement)
+ rows = [dict(id=i, p=value, value=10 + i) for i, value in
enumerate(values)]
+ if mode == 'pk':
+ rows = [row for row in rows if row['p'] is not None]
+ names = [name for value, name in zip(values, names) if value is not
None]
+ messages = _write(table, schema, rows)
+ assert _read(table) == pa.Table.from_pylist(rows,
schema=schema).to_pylist()
+ paths = []
+ for message in messages:
+ for file in message.new_files:
+ assert table.file_io.exists(file.file_path)
+ if placement == 'external':
+ assert file.external_path is not None
+ assert
urlparse(file.external_path).path.startswith(str(tmp_path / 'external') + '/')
+ paths.append(file.file_path)
+ for name in names:
+ assert any('/p=' + name + '/' in path for path in paths)
+
+
[email protected]_plan
[email protected]('partition_type,values,names', _PARTITIONS)
[email protected]('stream', [False, True])
+def test_native_evolution_updates_use_typed_partition_values(tmp_path,
partition_type, values, names, stream):
+ table, schema = _table(tmp_path, partition_type, 'evolution')
+ rows = [dict(id=i, p=value, value=i) for i, value in enumerate(values)]
+ _write(table, schema, rows, stream)
+ # Fail if the Python update planner or file writer takes over the
operation.
+ with
patch('pypaimon.write.table_update.BatchTableUpdate._build_predicate_update_table',
+ side_effect=AssertionError('Python row-ID update')):
+ builder = table.new_batch_write_builder()
+ messages = builder.new_update().update_by_predicate(None, {'value':
99})
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages)
+ finally:
+ commit.close()
+ expected = pa.Table.from_pylist(rows, schema=schema).to_pylist()
+ assert _read(table) == [dict(row, value=99) for row in expected]
+
+
[email protected]_plan
[email protected]_plan
[email protected]_read
[email protected]('partition_type', [pa.float32(), pa.float64()])
[email protected]('legacy', [False, True])
+def
test_native_non_finite_partitions_use_java_names_for_python_reads(tmp_path,
partition_type, legacy):
+ table, schema = _table(tmp_path, partition_type, extra_options={
+ 'partition.legacy-name': str(legacy).lower(),
+ 'scan.native-plan.enabled': 'false', 'read.native.enabled': 'false'})
+ rows = [dict(id=i, p=value, value=i) for i, value in
enumerate([float('inf'), float('-inf'), 1.5])]
+ messages = _write(table, schema, rows)
+ paths = [file.file_path for message in messages for file in
message.new_files]
+ assert paths and all(table.file_io.exists(path) for path in paths)
+ for name in ['Infinity', '-Infinity', '1.5']:
+ assert any('/p=' + name + '/' in path for path in paths)
+ assert _read(table) == rows
+
+
[email protected]_write
[email protected]_plan
[email protected]('partition_type,values,names', _PARTITIONS[2:])
+def test_python_binary_partitions_use_java_paths_and_native_reads(tmp_path,
partition_type, values, names):
+ table, schema = _table(tmp_path, partition_type, native=False)
+ rows = [dict(id=i, p=value, value=i) for i, value in enumerate(values)]
+ messages = _write(table, schema, rows)
+ for message in messages:
+ for file in message.new_files:
+ assert table.file_io.exists(file.file_path)
+ assert _read(table) == rows
+
+
[email protected]_plan
+def test_native_deletion_vector_restore_uses_java_partition_path(tmp_path):
+ table, schema = _table(tmp_path, pa.float64(), 'evolution', extra_options={
+ 'deletion-vectors.enabled': 'true', 'index-file-in-data-file-dir':
'true'})
+ _write(table, schema, [dict(id=i, p=1e23, value=i) for i in range(3)])
+
+ def delete(row_id):
+ builder = table.new_batch_write_builder()
+ messages = builder.new_update().delete_by_row_id([row_id])
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages)
+ finally:
+ commit.close()
+
+ delete(0)
+ assert [row['id'] for row in _read(table)] == [1, 2]
+ delete(1)
+ assert [row['id'] for row in _read(table)] == [2]
+
+
[email protected]_plan
[email protected]('action', ['DELETE', 'UPDATE SET value = 99'])
+def test_native_cow_merge_uses_java_partition_path(tmp_path, action):
+ from pypaimon_rust.datafusion import SQLContext
+
+ table, schema = _table(tmp_path, pa.float64())
+ _write(table, schema, [dict(id=i, p=1e23, value=i) for i in range(3)])
+ context = SQLContext()
+ context.register_catalog('paimon', {'warehouse': str(tmp_path /
'warehouse')})
+ context.sql('MERGE INTO paimon.db.t t USING (SELECT CAST(1 AS INT) AS id)
s '
+ 'ON t.id = s.id WHEN MATCHED THEN ' + action)
+ expected = [dict(id=i, p=1e23, value=99 if action.startswith('UPDATE') and
i == 1 else i)
+ for i in range(3) if action != 'DELETE' or i != 1]
+ assert _read(table) == expected
+
+
[email protected]('value,expected', [
+ (b'a/b=c', 'a%2Fb%3Dc'), (b'\xed\xa0\x80', '\ufffd'), (b'\xed\xa0A',
'\ufffdA'),
+ (b'\xe2\x82', '\ufffd'), (b'\xe0\x80\x80', '\ufffd\ufffd\ufffd'),
+ (b'', '__DEFAULT_PARTITION__'), (b' \t\x1c', '__DEFAULT_PARTITION__'),
+ ('\u00a0'.encode(), '\u00a0'), ('\u0085'.encode(), '\u0085'),
+])
+def test_binary_partition_string_conversion_matches_java(tmp_path, value,
expected):
+ table, _ = _table(tmp_path, pa.binary(), native=False)
+ assert table.path_factory().relative_bucket_path((value,), 0,
canonical_partition=True) == (
+ 'p=' + expected + '/bucket-0')
+
+
+def
test_python_partition_statistics_keep_java_values_and_typed_identity(tmp_path):
+ from pypaimon.manifest.schema.manifest_entry import ManifestEntry
+ from pypaimon.table.row.generic_row import GenericRow
+ from pypaimon.write.file_store_commit import FileStoreCommit
+
+ table, _ = _table(tmp_path, pa.binary(), native=False)
+ commit = FileStoreCommit.__new__(FileStoreCommit)
+ commit.table = table
+ entries = [ManifestEntry(0, GenericRow([value],
table.partition_keys_fields), 0, 1,
+ Mock(row_count=2, file_size=10,
creation_time=None))
+ for value in [None, b' \t', b'a/b', b'\xed\xa0\x80',
bytearray(b'a/b'), memoryview(b'a/b')]]
+ statistics = commit._generate_partition_statistics(entries)
+ # NULL and whitespace name the same directory, but remain different typed
+ # partitions, as in Java PartitionEntry.merge and Rust BinaryRow grouping.
+ assert [stat.spec for stat in statistics] == [
+ {'p': '__DEFAULT_PARTITION__'}, {'p': '__DEFAULT_PARTITION__'}, {'p':
'a/b'}, {'p': '\ufffd'}]
+ assert [stat.record_count for stat in statistics] == [2, 2, 6, 2]
+
+
[email protected]('partition_type,values,counts', [
+ (pa.float32(), [1.00000001, 1.0, 2.0], [4, 2]),
+ (pa.float64(), [1.00000001, 1.0, 2.0], [2, 2, 2]),
+ (pa.timestamp('ms', tz='UTC'), [
+ datetime(2024, 1, 1),
+ datetime(2024, 1, 1, 8, tzinfo=timezone(timedelta(hours=8))),
+ datetime(2024, 1, 2, tzinfo=timezone.utc)], [4, 2]),
+])
+def test_python_partition_statistics_group_by_schema_encoded_values(tmp_path,
partition_type, values, counts):
+ from pypaimon.manifest.schema.manifest_entry import ManifestEntry
+ from pypaimon.table.row.generic_row import GenericRow
+ from pypaimon.write.file_store_commit import FileStoreCommit
+
+ table, _ = _table(tmp_path, partition_type, native=False)
+ commit = FileStoreCommit.__new__(FileStoreCommit)
+ commit.table = table
+ entries = [ManifestEntry(0, GenericRow([value],
table.partition_keys_fields), 0, 1,
+ Mock(row_count=2, file_size=10,
creation_time=None)) for value in values]
+ statistics = commit._generate_partition_statistics(entries)
+ assert [stat.record_count for stat in statistics] == counts
+ assert [stat.file_count for stat in statistics] == [count // 2 for count
in counts]
+ assert [stat.file_size_in_bytes for stat in statistics] == [count * 5 for
count in counts]
+
+
[email protected]('legacy', [False, True])
+def
test_native_commit_binary_partition_capability_precedes_execution(tmp_path,
legacy):
+ from pypaimon.write.native_commit import create_native_commit
+
+ table, _ = _table(tmp_path, pa.binary(), native=False,
+ extra_options={'partition.legacy-name':
str(legacy).lower()})
+ with patch('pypaimon.write.native_commit._rest_catalog_supported',
return_value=True), \
+ patch('pypaimon.write.native_commit.native_commit_available',
return_value=True), \
+ patch('pypaimon.write.native_commit.create_native_write_table') as
native:
+ committer = create_native_commit(table, 'user')
+ if legacy:
+ assert committer is None
+ native.assert_not_called()
+ else:
+ assert committer is not None
+ native.assert_called_once_with(table)
diff --git a/paimon-python/pypaimon/tests/native_update_test.py
b/paimon-python/pypaimon/tests/native_update_test.py
index e59e81813a..8ed01dbf62 100644
--- a/paimon-python/pypaimon/tests/native_update_test.py
+++ b/paimon-python/pypaimon/tests/native_update_test.py
@@ -1027,5 +1027,5 @@ 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 = 2 if layout in ('python-escaped', 'canonical',
'ordinary', 'directory', 'external') else 0
+ expected_calls = 0 if layout == 'external-binary' else 2
assert len(native_calls) == expected_calls
diff --git a/paimon-python/pypaimon/tests/py36/rest_ao_read_write_test.py
b/paimon-python/pypaimon/tests/py36/rest_ao_read_write_test.py
index 03fd763a9e..889c2e125b 100644
--- a/paimon-python/pypaimon/tests/py36/rest_ao_read_write_test.py
+++ b/paimon-python/pypaimon/tests/py36/rest_ao_read_write_test.py
@@ -283,7 +283,7 @@ class RESTAOReadWritePy36Test(RESTBaseTest):
('region', pa.string()),
('city', pa.string())
])
- schema = Schema.from_pyarrow_schema(pa_schema)
+ schema = Schema.from_pyarrow_schema(pa_schema,
partition_keys=['region', 'city'])
self.rest_catalog.create_table('default.tb', schema, False)
table = self.rest_catalog.get_table('default.tb')
partition_fields = [
@@ -341,7 +341,7 @@ class RESTAOReadWritePy36Test(RESTBaseTest):
('region', pa.string()),
('city', pa.string())
])
- schema = Schema.from_pyarrow_schema(pa_schema)
+ schema = Schema.from_pyarrow_schema(pa_schema,
partition_keys=['region', 'city'])
self.rest_catalog.create_table('default.tb1', schema, False)
table = self.rest_catalog.get_table('default.tb1')
partition_fields = [
diff --git a/paimon-python/pypaimon/tests/reader_base_test.py
b/paimon-python/pypaimon/tests/reader_base_test.py
index a54540a27e..b6a0ce47a8 100644
--- a/paimon-python/pypaimon/tests/reader_base_test.py
+++ b/paimon-python/pypaimon/tests/reader_base_test.py
@@ -413,7 +413,7 @@ class ReaderBasicTest(unittest.TestCase):
('region', pa.string()),
('city', pa.string())
])
- schema = Schema.from_pyarrow_schema(pa_schema)
+ schema = Schema.from_pyarrow_schema(pa_schema,
partition_keys=['region', 'city'])
self.catalog.create_table('default.tb', schema, False)
table = self.catalog.get_table('default.tb')
partition_fields = [
@@ -472,7 +472,7 @@ class ReaderBasicTest(unittest.TestCase):
('region', pa.string()),
('city', pa.string())
])
- schema = Schema.from_pyarrow_schema(pa_schema)
+ schema = Schema.from_pyarrow_schema(pa_schema,
partition_keys=['region', 'city'])
self.catalog.create_table('default.tb1', schema, False)
table = self.catalog.get_table('default.tb1')
partition_fields = [
diff --git a/paimon-python/pypaimon/tests/table_commit_test.py
b/paimon-python/pypaimon/tests/table_commit_test.py
index d606a94331..8a213fd10d 100644
--- a/paimon-python/pypaimon/tests/table_commit_test.py
+++ b/paimon-python/pypaimon/tests/table_commit_test.py
@@ -65,7 +65,7 @@ class TestTableCommit(unittest.TestCase):
if mode == 'dynamic':
self.assertTrue(any(message.index_adds for message in
messages))
paths.extend(table.path_factory().bucket_index_path(
- tuple(entry.partition.values), entry.bucket,
entry.index_file, table.file_io)
+ tuple(entry.partition.values), entry.bucket,
entry.index_file)
for message in messages for entry in
message.index_adds)
self.assertTrue(paths)
publish = commit.file_store_commit.commit
diff --git a/paimon-python/pypaimon/utils/file_store_path_factory.py
b/paimon-python/pypaimon/utils/file_store_path_factory.py
index 1536ffd93c..8da0b11a72 100644
--- a/paimon-python/pypaimon/utils/file_store_path_factory.py
+++ b/paimon-python/pypaimon/utils/file_store_path_factory.py
@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
+import math
import struct
from datetime import date, timezone
from decimal import Decimal
@@ -32,7 +33,29 @@ def _is_null_or_whitespace_only(value) -> bool:
if value is None:
return True
s = str(value)
- return len(s) == 0 or s.isspace()
+ # Character.isWhitespace, as used by Java StringUtils. Python isspace also
+ # folds NBSP, NEL and other characters Java keeps in a partition name.
+ return all(char in '\t\n\v\f\r\x1c\x1d\x1e\x1f
\u1680\u2000\u2001\u2002\u2003\u2004\u2005\u2006'
+ '\u2008\u2009\u200a\u2028\u2029\u205f\u3000' for char in s)
+
+
+def _decode_java_utf8(value: bytes) -> str:
+ """BinaryString.toString replaces a malformed surrogate as one sequence."""
+ value = bytes(value)
+ parts = []
+ while value:
+ try:
+ parts.append(value.decode('utf-8'))
+ break
+ except UnicodeDecodeError as error:
+ parts.append(value[:error.start].decode('utf-8'))
+ value = value[error.start:]
+ length = error.end - error.start
+ if len(value) >= 2 and value[0] == 0xed and 0xa0 <= value[1] <=
0xbf:
+ length = 3 if len(value) >= 3 and value[2] & 0xc0 == 0x80 else
2
+ parts.append('\ufffd')
+ value = value[length:]
+ return ''.join(parts)
def _escape_partition_component(value: str) -> str:
@@ -60,11 +83,15 @@ def canonical_data_file_path(table, partition, bucket,
file_name):
def _floating_partition_string(value, single_precision: bool) -> str:
- # Use a shortest round-tripping form for the initial lookup. Older JVMs
- # can use different digits; the read fallback matches their stored values.
+ # Java's JDK 19+ contract selects a shortest round-tripping form with at
+ # least two significant digits before applying its notation thresholds.
encoding = '>f' if single_precision else '>d'
bits = struct.pack(encoding, value)
value = struct.unpack(encoding, bits)[0]
+ if math.isnan(value):
+ return 'NaN'
+ if math.isinf(value):
+ return '-Infinity' if value < 0 else 'Infinity'
for precision in range(2, 10 if single_precision else 18):
text = format(value, '.{}g'.format(precision))
try:
@@ -74,8 +101,6 @@ def _floating_partition_string(value, single_precision:
bool) -> str:
if rounded != bits:
continue
decimal = Decimal(text)
- if not decimal.is_finite():
- return str(value)
if decimal.is_zero() or Decimal('0.001') <= abs(decimal) <
Decimal('1e7'):
text = format(decimal, 'f')
if '.' in text:
@@ -175,6 +200,8 @@ class FileStorePathFactory:
for i, value in enumerate(partition):
data_type = self.partition_types[i] if self.partition_types is not
None else None
type_name = str(data_type).split('(', 1)[0].split()[0]
+ if value is not None and type_name in ('BINARY', 'VARBINARY',
'BYTES') and not self.legacy_partition_name:
+ value = _decode_java_utf8(value)
if value is not None and _is_ltz_type(str(data_type).upper()):
# Legacy Timestamp.toString() uses UTC fields. Java's
non-legacy
# cast uses TimeZone.getDefault(); neither includes an offset.
@@ -199,6 +226,8 @@ class FileStorePathFactory:
elif self.legacy_partition_name and type_name.startswith('TIME'):
text = str(((value.hour * 60 + value.minute) * 60 +
value.second) * 1000
+ value.microsecond // 1000)
+ elif type_name in ('BINARY', 'VARBINARY', 'BYTES') and not
self.legacy_partition_name:
+ text = value
elif data_type is not None and not _is_unsupported(data_type):
text = cast_value_to_string(value, data_type)
else:
@@ -272,67 +301,19 @@ class FileStorePathFactory:
if external is not None:
return external.get_next_external_data_path(file_name), True
path = f"{self.bucket_path(partition, bucket,
canonical_partition=True)}/{file_name}"
- # Java Float/Double.toString differs across JDK versions. Record
- # the actual Java-style directory so every Java reader resolves
- # the same file through IndexFileMeta.externalPath.
- floating = any(value is not None and str(data_type).split()[0] in
('FLOAT', 'REAL', 'DOUBLE')
- for value, data_type in zip(partition,
self.partition_types or []))
- return to_file_io_path(path), floating
+ # Java IndexInDataFileDirPathFactory marks a path external only
+ # when a configured ExternalPathProvider selected it.
+ return to_file_io_path(path), False
factory = self.global_index_path_factory()
return factory.to_path(file_name), factory.is_external_path()
- def bucket_index_path(self, partition: Tuple, bucket: int, index_file,
file_io=None) -> str:
+ def bucket_index_path(self, partition: Tuple, bucket: int, index_file) ->
str:
"""Resolve a bucket index using Java partition paths and explicit
locations."""
if index_file.external_path:
return to_file_io_path(index_file.external_path)
if not self.index_file_in_data_file_dir:
return f"{self.index_path()}/{index_file.file_name}"
- path = to_file_io_path(f"{self.bucket_path(partition, bucket,
True)}/{index_file.file_name}")
- # Java Float/Double.toString can produce different digits across JDK
- # versions. Only these floating partition spellings need a lookup.
- if file_io is not None and index_file.index_type in
('DELETION_VECTORS', 'HASH') and not file_io.exists(path):
- python_path = to_file_io_path(f"{self.bucket_path(partition,
bucket)}/{index_file.file_name}")
- alternate = self._find_floating_bucket_index(partition, bucket,
index_file.file_name, file_io, python_path)
- if alternate is not None:
- return alternate
- return path
-
- def _find_floating_bucket_index(self, partition, bucket, file_name,
file_io, python_path):
- floating = [str(data_type).split()[0] in ('FLOAT', 'REAL', 'DOUBLE')
- for data_type in self.partition_types or []]
- if not any(is_float and value is not None for is_float, value in
zip(floating, partition)):
- return None
- # Float/Double.toString changed across JDK releases. Only if the usual
- # path is missing, inspect floating partition components and compare
- # their exact encoded values (including the sign of zero).
- paths = [to_file_io_path(resolve_path(self.data_file_path(), '.'))]
- for i, text in enumerate(self._canonical_partition(partition)):
- prefix = _escape_partition_component(self.partition_keys[i]) + '='
- if not floating[i] or partition[i] is None:
- paths = [path + '/' + prefix +
_escape_partition_component(text) for path in paths]
- continue
- encoding = '>d' if str(self.partition_types[i]).split()[0] ==
'DOUBLE' else '>f'
- expected = struct.pack(encoding, partition[i])
- matched = []
- for path in paths:
- if not file_io.exists(path):
- continue
- for status in file_io.list_status(path):
- name = status.base_name
- if not name.startswith(prefix):
- continue
- try:
- if struct.pack(encoding, float(name[len(prefix):])) ==
expected:
- matched.append(path + '/' + name)
- except (ValueError, OverflowError):
- continue
- paths = sorted(matched)
- bucket_name = 'postpone' if bucket == BucketMode.POSTPONE_BUCKET.value
else str(bucket)
- for path in paths:
- candidate = '{}/{}{}/{}'.format(path, self.BUCKET_PATH_PREFIX,
bucket_name, file_name)
- if candidate != python_path and file_io.exists(candidate):
- return candidate
- return None
+ return to_file_io_path(f"{self.bucket_path(partition, bucket,
True)}/{index_file.file_name}")
class IndexPathFactory:
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index 10d8ce23c6..c00f138921 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -33,7 +33,7 @@ from pypaimon.manifest.schema.manifest_entry import
ManifestEntry
from pypaimon.snapshot.snapshot import Snapshot
from pypaimon.snapshot.snapshot_commit import (PartitionStatistics,
SnapshotCommit)
-from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.table.row.generic_row import GenericRow, GenericRowSerializer
from pypaimon.table.row.offset_row import OffsetRow
from pypaimon.utils.file_store_path_factory import canonical_data_file_path
from pypaimon.write.commit.commit_rollback import CommitRollback
@@ -1046,7 +1046,7 @@ class FileStoreCommit:
file_name = index_file.file_name
if index_file.index_type in ('DELETION_VECTORS', 'HASH'):
path = self.table.path_factory().bucket_index_path(
- tuple(entry.partition.values), entry.bucket,
index_file, self.table.file_io)
+ tuple(entry.partition.values), entry.bucket,
index_file)
else:
path = (
index_file.external_path
@@ -1087,31 +1087,17 @@ class FileStoreCommit:
List of PartitionStatistics for each unique partition
"""
partition_stats = {}
+ path_factory = self.table.path_factory()
for entry in commit_entries:
- # Convert partition tuple to dictionary for PartitionStatistics
- partition_value = tuple(entry.partition.values) # Call the method
to get partition value
- if partition_value:
- # Assuming partition is a tuple and we need to convert it to a
dict
- # This may need adjustment based on actual partition format
- if isinstance(partition_value, tuple):
- # Create partition spec from partition tuple and table
partition keys
- partition_spec = {}
- if len(partition_value) == len(self.table.partition_keys):
- for i, key in enumerate(self.table.partition_keys):
- partition_spec[key] = str(partition_value[i])
- else:
- # Fallback: use indices as keys
- for i, value in enumerate(partition_value):
- partition_spec[f"partition_{i}"] = str(value)
- else:
- # If partition is already a dict or other format
- partition_spec = dict(partition_value) if partition_value
else {}
- else:
- # Default partition for unpartitioned tables
- partition_spec = {}
-
- partition_key = tuple(sorted(partition_spec.items()))
+ partition_value = tuple(entry.partition.values)
+ if len(partition_value) != len(self.table.partition_keys):
+ raise ValueError('Partition row does not match the table
partition keys')
+ # Java PartitionEntry.toPartitionStatistics reuses the path value
+ # computer, without path escaping. Group by serialized partition
rows:
+ # NULL and whitespace remain distinct even if their names coincide.
+ partition_spec = dict(zip(self.table.partition_keys,
path_factory._canonical_partition(partition_value)))
+ partition_key = GenericRowSerializer.to_bytes(entry.partition)
if partition_key not in partition_stats:
partition_stats[partition_key] = {
diff --git a/paimon-python/pypaimon/write/native_commit.py
b/paimon-python/pypaimon/write/native_commit.py
index 91716cc357..2abcdd7f4e 100644
--- a/paimon-python/pypaimon/write/native_commit.py
+++ b/paimon-python/pypaimon/write/native_commit.py
@@ -60,6 +60,13 @@ def create_native_commit(table, commit_user,
overwrite_partition=None):
"""Return a native committer only when its publication protocol matches
Python."""
if not _rest_catalog_supported(table) or not native_commit_available():
return None
+ from pypaimon.common.options.core_options import CoreOptions
+ from pypaimon.schema.data_types import PyarrowFieldParser
+ from pypaimon.write.native_write import _native_partition_types_supported
+ if not _native_partition_types_supported(
+ PyarrowFieldParser.from_paimon_schema(table.table_schema.fields),
table.partition_keys,
+
table.options.options.get(CoreOptions.PARTITION_GENERATE_LEGACY_NAME)):
+ return None
native_table = create_native_write_table(table)
if native_table is None:
return None
diff --git a/paimon-python/pypaimon/write/native_merge_into.py
b/paimon-python/pypaimon/write/native_merge_into.py
index 8652079f15..7999c00d72 100644
--- a/paimon-python/pypaimon/write/native_merge_into.py
+++ b/paimon-python/pypaimon/write/native_merge_into.py
@@ -19,6 +19,8 @@
import pyarrow as pa
+from pypaimon.common.options.core_options import CoreOptions
+
from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
from pypaimon.write.native_commit import create_native_write_table,
from_native_commit_messages
from pypaimon.write.native_update import (
@@ -62,7 +64,9 @@ def create_native_merge_into(table, source, on, matched,
not_matched, commit_use
return None
from pypaimon.schema.data_types import PyarrowFieldParser
source_schema =
PyarrowFieldParser.from_paimon_schema(source.table_schema.fields)
- if not _native_partition_types_supported(source_schema,
source.partition_keys):
+ if not _native_partition_types_supported(
+ source_schema, source.partition_keys,
+
source.options.options.get(CoreOptions.PARTITION_GENERATE_LEGACY_NAME)):
return None
native_source = create_native_write_table(source)
if native_source is None:
diff --git a/paimon-python/pypaimon/write/native_update.py
b/paimon-python/pypaimon/write/native_update.py
index 1abf1ba4f9..f421a42e7d 100644
--- a/paimon-python/pypaimon/write/native_update.py
+++ b/paimon-python/pypaimon/write/native_update.py
@@ -20,7 +20,7 @@
import pyarrow as pa
from pypaimon.schema.data_types import MapType, PyarrowFieldParser,
is_blob_file_field
-from pypaimon.common.options.core_options import ChangelogProducer
+from pypaimon.common.options.core_options import ChangelogProducer, CoreOptions
from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
from pypaimon.snapshot.time_travel_util import SCAN_KEYS
from pypaimon.table.file_store_table import FileStoreTable
@@ -46,7 +46,8 @@ def _native_row_id_table(table):
or not native_write_available()):
return None
schema = PyarrowFieldParser.from_paimon_schema(table.table_schema.fields)
- if not _native_partition_types_supported(schema, table.partition_keys):
+ if not _native_partition_types_supported(
+ schema, table.partition_keys,
table.options.options.get(CoreOptions.PARTITION_GENERATE_LEGACY_NAME)):
return None
return create_native_write_table(table)
diff --git a/paimon-python/pypaimon/write/native_write.py
b/paimon-python/pypaimon/write/native_write.py
index 9a353f60a9..7bbccd3ebf 100644
--- a/paimon-python/pypaimon/write/native_write.py
+++ b/paimon-python/pypaimon/write/native_write.py
@@ -20,7 +20,7 @@ from importlib import import_module
import pyarrow as pa
-from pypaimon.common.options.core_options import MergeEngine
+from pypaimon.common.options.core_options import CoreOptions, MergeEngine
from pypaimon.schema.arrow_schema import arrow_schemas_compatible,
normalize_arrow_strings
from pypaimon.schema.data_types import PyarrowFieldParser, is_blob_file_field
from pypaimon.table.bucket_mode import BucketMode
@@ -39,11 +39,13 @@ def native_write_available() -> bool:
return True
-def _native_partition_types_supported(schema, partition_keys):
+def _native_partition_types_supported(schema, partition_keys,
legacy_partition_name=True):
"""Partition keys which Rust can encode and use to locate existing
files."""
- return not any(
+ # Java's legacy byte[].toString() contains an allocation identity, so
binary
+ # partitions only have portable names with the cast-based option.
+ return not legacy_partition_name or not any(
pa.types.is_binary(data_type) or pa.types.is_large_binary(data_type)
- or pa.types.is_fixed_size_binary(data_type) or
pa.types.is_floating(data_type)
+ or pa.types.is_fixed_size_binary(data_type)
for data_type in (schema.field(name).type for name in partition_keys))
@@ -62,8 +64,8 @@ def create_native_write(table, commit_user,
static_partition=None, stream=False,
MergeEngine.AGGREGATE))
or table.options.changelog_file_format() not in (None, 'parquet')
or table.options.file_format() != 'parquet'
- # Rust cannot encode these partition keys yet.
- or not _native_partition_types_supported(schema,
table.partition_keys)
+ or not _native_partition_types_supported(
+ schema, table.partition_keys,
table.options.options.get(CoreOptions.PARTITION_GENERATE_LEGACY_NAME))
or table.options.video_frame_fields()):
return None
native_table = create_native_write_table(table)
diff --git a/paimon-python/pypaimon/write/table_delete.py
b/paimon-python/pypaimon/write/table_delete.py
index 18dff8b910..00ea6b1cec 100644
--- a/paimon-python/pypaimon/write/table_delete.py
+++ b/paimon-python/pypaimon/write/table_delete.py
@@ -327,4 +327,4 @@ class TableDeleteByRowId:
def _index_file_path(self, index_file: IndexFileMeta, partition:
GenericRow, bucket: int) -> str:
return self.table.path_factory().bucket_index_path(
- tuple(partition.values), bucket, index_file, self.file_io)
+ tuple(partition.values), bucket, index_file)
