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 750aaea265 [python] Enable native dynamic bucket writes and align HASH
commits (#10353)
750aaea265 is described below
commit 750aaea26544a77e670ca8d2d2c2ae303ce3e93a
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Oct 3 21:52:36 2026 +0800
[python] Enable native dynamic bucket writes and align HASH commits (#10353)
---
paimon-python/README.md | 22 +-
paimon-python/pypaimon/index/dynamic_bucket.py | 20 +-
.../pypaimon/manifest/index_manifest_file.py | 50 ++--
.../pypaimon/tests/deletion_vector_path_test.py | 123 ++++++---
.../pypaimon/tests/native_dynamic_write_test.py | 280 +++++++++++++++++++++
.../tests/native_plan_dynamic_bucket_test.py | 4 +-
paimon-python/pypaimon/tests/native_write_test.py | 11 +-
.../pypaimon/tests/write/dynamic_bucket_test.py | 54 ++--
.../pypaimon/utils/file_store_path_factory.py | 47 ++--
.../pypaimon/write/commit/conflict_detection.py | 27 +-
paimon-python/pypaimon/write/file_store_commit.py | 2 +-
paimon-python/pypaimon/write/native_write.py | 2 +-
paimon-python/pypaimon/write/row_key_extractor.py | 3 +
13 files changed, 461 insertions(+), 184 deletions(-)
diff --git a/paimon-python/README.md b/paimon-python/README.md
index 1b7c200b89..b1b904d6c1 100644
--- a/paimon-python/README.md
+++ b/paimon-python/README.md
@@ -168,6 +168,15 @@ path select the Python writer before native data is
written. If the runtime or t
unavailable, write uses Python. Once Rust starts writing a batch, errors
propagate without retrying that batch through Python.
+Primary-key dynamic buckets (`bucket=-1`, with partition fields included in the
+primary key) also support native writes. HASH indexes are restored across
writer
+restarts; `dynamic-bucket.max-buckets` bounds bucket growth and reuses existing
+buckets after that limit. Dynamic buckets use the trimmed primary key, so
+`bucket-key` must not be configured. As in Java, each bucket must have one
writer
+owner: a HASH ADD replaces the complete previous index and does not carry a
+concurrent-writer baseline. Partitions with existing data but missing HASH
indexes
+must be rewritten before incremental writes.
+
Native writes honor `data-file.path-directory` and the configured
`data-file.external-paths` strategy. Existing files keep their recorded
locations
when the write destinations change. Python and native readers and committers
@@ -317,14 +326,15 @@ Explicit row ranges on data-evolution tables require
`ReadBuilder.with_row_range
Watermark time travel requires Rust 0.4 or newer. Branch reads require the
branch-aware binding exposing `Table.branch()`, and the resolved branch is
checked before planning. Deletion-vector scans require `pypaimon-rust>=0.4.0`,
-which includes schema-aware decoding of Python-written index manifests and
-legacy bucket-index path compatibility. The reader honors explicit paths, then
-bucket paths, and can read older Python files placed in `table/index`.
+which includes schema-aware decoding of index manifests. Index paths follow
+Java: an explicit external path takes precedence; otherwise
+`index-file-in-data-file-dir` selects the bucket or table index directory.
Bucket paths use the partition field types and `partition.legacy-name` to match
Java formatting, including timestamp precision and different JVM float
spellings.
-New Python writes honor `index-file-in-data-file-dir` and retain explicit paths
-when Python and Java partition-directory formatting differs. Older releases
-and prereleases before 0.4.0 use the Python planner for deletion vectors.
+Python bucket-index writes use Java partition paths. Explicit paths identify
+external files and floating partition directories whose spelling depends on the
+JDK version. Missing bucket indexes fail instead of searching Python layouts.
Older releases and prereleases
+before 0.4.0 use the Python planner for deletion vectors.
When using an unreleased 0.4.0 development wheel, rebuild it with these fixes;
package version checks cannot distinguish local builds with identical versions.
diff --git a/paimon-python/pypaimon/index/dynamic_bucket.py
b/paimon-python/pypaimon/index/dynamic_bucket.py
index 90a1c49230..f89862875c 100644
--- a/paimon-python/pypaimon/index/dynamic_bucket.py
+++ b/paimon-python/pypaimon/index/dynamic_bucket.py
@@ -20,6 +20,7 @@
import random
import struct
+import uuid
from dataclasses import dataclass
from typing import Dict, Iterator, List, Optional, Set, Tuple
@@ -83,12 +84,9 @@ def validate_bucket_id(bucket: int) -> None:
def _iter_hashes(table, entry: IndexManifestEntry) -> Iterator[int]:
meta = entry.index_file
- path = meta.external_path
- if path is None:
- # Files without an external path always live in the table's index
- # directory, even if global-index.external-path was configured later.
- index_path =
table.path_factory().global_index_path_factory().index_path()
- path = f"{index_path}/{meta.file_name}"
+ path = table.path_factory().bucket_index_path(
+ tuple(entry.partition.values), entry.bucket, meta, table.file_io
+ )
with table.file_io.new_input_stream(path) as stream:
remainder = b""
while True:
@@ -507,9 +505,11 @@ class DynamicBucketIndexMaintainer:
def _write_index(
self, partition: Tuple, bucket: int, hashes: Set[int]
) -> IndexManifestEntry:
- path_factory = self.table.path_factory().global_index_path_factory()
-
self.table.file_io.check_or_mkdirs(path_factory.global_index_root_path())
- path = path_factory.new_path()
+ file_name = f"index-{uuid.uuid4()}-0"
+ path, external = self.table.path_factory().new_bucket_index_path(
+ partition, bucket, file_name
+ )
+ self.table.file_io.check_or_mkdirs(path.rsplit("/", 1)[0])
payload = b"".join(struct.pack(">i", value) for value in
sorted(hashes))
try:
with self.table.file_io.new_output_stream(path) as stream:
@@ -524,7 +524,7 @@ class DynamicBucketIndexMaintainer:
file_name=path.rsplit("/", 1)[-1],
file_size=self.table.file_io.get_file_size(path),
row_count=len(hashes),
- external_path=path if path_factory.is_external_path() else None,
+ external_path=path if external else None,
)
return IndexManifestEntry(
kind=_ADD,
diff --git a/paimon-python/pypaimon/manifest/index_manifest_file.py
b/paimon-python/pypaimon/manifest/index_manifest_file.py
index 3c1c9448ba..ee78f08d4a 100644
--- a/paimon-python/pypaimon/manifest/index_manifest_file.py
+++ b/paimon-python/pypaimon/manifest/index_manifest_file.py
@@ -255,8 +255,7 @@ class IndexManifestFile:
delete_names = {e.index_file.file_name for e in deletes}
survivors = [e for e in previous if e.index_file.file_name not in
delete_names]
_validate_retained_global_index_files(survivors, adds)
- _validate_hash_index_changes(survivors, adds)
- combined = survivors + adds
+ combined = _combine_hash_indexes(survivors, adds, deletes)
if not combined:
return None
return self.write(combined)
@@ -353,33 +352,26 @@ def _validate_retained_global_index_files(
)
-def _validate_hash_index_changes(
- retained_entries: List[IndexManifestEntry],
- added_entries: List[IndexManifestEntry],
-) -> None:
- """Keep the one-HASH-file-per-bucket manifest invariant."""
- by_bucket = {
- (tuple(entry.partition.values), entry.bucket): entry
- for entry in retained_entries
- if entry.index_file.index_type == _HASH_INDEX
- }
- for added in added_entries:
- if added.kind != _ADD or added.index_file.index_type != _HASH_INDEX:
- continue
- key = (tuple(added.partition.values), added.bucket)
- previous = by_bucket.get(key)
- if previous is not None:
- raise RuntimeError(
- "Trying to add HASH index file {} for partition {}, bucket {},
"
- "but HASH index file {} still exists. Remove the previous file
"
- "first.".format(
- added.index_file.file_name,
- key[0],
- key[1],
- previous.index_file.file_name,
- )
- )
- by_bucket[key] = added
+def _combine_hash_indexes(retained_entries, added_entries, deleted_entries):
+ """Java BucketedCombiner replaces HASH by (partition, bucket, index type).
+
+ Writers submit a complete replacement file, so an ADD need not carry the
+ previous file as a DELETE. Each bucket must have a single writer owner.
+ """
+ def key(entry):
+ return tuple(entry.partition.values), entry.bucket
+
+ hash_indexes = {key(entry): entry for entry in retained_entries
+ if entry.index_file.index_type == _HASH_INDEX}
+ for entry in deleted_entries:
+ if entry.index_file.index_type == _HASH_INDEX:
+ hash_indexes.pop(key(entry), None)
+ for entry in added_entries:
+ if entry.index_file.index_type == _HASH_INDEX:
+ hash_indexes[key(entry)] = entry
+ return ([entry for entry in retained_entries + added_entries
+ if entry.index_file.index_type != _HASH_INDEX]
+ + list(hash_indexes.values()))
def _is_global_index(index_type: str) -> bool:
diff --git a/paimon-python/pypaimon/tests/deletion_vector_path_test.py
b/paimon-python/pypaimon/tests/deletion_vector_path_test.py
index bfa2100448..9ed9b6179d 100644
--- a/paimon-python/pypaimon/tests/deletion_vector_path_test.py
+++ b/paimon-python/pypaimon/tests/deletion_vector_path_test.py
@@ -68,9 +68,8 @@ def _table(tmp_path, layout, first_partition='a',
partition_type=None, legacy_pa
catalog.create_table('db.t', Schema.from_pyarrow_schema(
schema, partition_keys=['p'], options=options), False)
table = catalog.get_table('db.t')
- # Seed the legacy Python partition paths and deterministic row-id ordering
- # which these compatibility/decoy tests deliberately reference. Updates and
- # reads below still exercise the configured native backend.
+ # Seed through Python with deterministic row-id ordering. Updates and
+ # reads below exercise the configured backend against Java index paths.
builder = table.copy({'write.native.enabled':
'false'}).new_batch_write_builder()
writer, commit = builder.new_write(), builder.new_commit()
try:
@@ -150,30 +149,64 @@ def
test_delete_paths_preserve_repeated_deletes_and_historical_reads(tmp_path, p
@pytest.mark.parametrize('planner', _PLANNERS)
-def
test_legacy_python_index_directory_remains_readable_and_new_deletes_use_bucket(tmp_path,
planner):
- table = _table(tmp_path, 'bucket')
- # Old Python writers ignored the option and placed the index under
table/index.
- legacy_factory = table.path_factory()
- legacy_factory.index_file_in_data_file_dir = False
- legacy_table = table.copy({'write.native.enabled': 'false'})
- with patch.object(legacy_table, 'path_factory',
return_value=legacy_factory):
- _delete(legacy_table, [0, 2])
- old_paths = [Path(table.table_path) / 'index' / entry.index_file.file_name
for entry in _entries(table, 2)]
- assert all(path.is_file() for path in old_paths)
- _read(table, planner, 2, [1, 3])
- _delete(table, [1])
- _read(table, planner, 3, [3])
- _read(table, planner, 2, [1, 3])
- for entry in _entries(table, 3):
- if entry.partition.values == ['a']:
- assert (Path(table.path_factory().bucket_path(('a',),
entry.bucket)) /
- entry.index_file.file_name).is_file()
- assert all(path.is_file() for path in old_paths)
[email protected]('wrong_directory', ['table-index',
'unescaped-partition'])
+def test_bucket_dv_requires_java_directory(tmp_path, planner, wrong_directory):
+ table = _table(tmp_path, 'bucket', 'a/b')
+ _delete(table, [0])
+ entry = _entries(table, 2)[0]
+ factory = table.path_factory()
+ 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)
+ wrong_parent = (factory.index_path() if wrong_directory == 'table-index'
+ else factory.bucket_path(partition, entry.bucket))
+ wrong_path = wrong_parent + '/' + file.file_name
+ with table.file_io.new_input_stream(path) as stream:
+ data = stream.read()
+ table.file_io.delete_quietly(path)
+ with table.file_io.new_output_stream(wrong_path) as stream:
+ stream.write(data)
+ with pytest.raises((FileNotFoundError, ValueError)) as error:
+ _read(table, planner, 2, [1, 2, 3])
+ assert file.file_name in str(error.value)
+ with pytest.raises((FileNotFoundError, ValueError)) as error:
+ _delete(table, [1])
+ assert file.file_name in str(error.value)
+ assert table.snapshot_manager().get_latest_snapshot().id == 2
+ assert table.file_io.exists(wrong_path)
+
+
[email protected]('file_uri', [False, True])
[email protected]('escaped_root', [False, True])
+def test_floating_bucket_dv_does_not_search_python_directory(tmp_path,
file_uri, escaped_root):
+ if escaped_root:
+ tmp_path = tmp_path / 'root%2Fdir'
+ table = _table(tmp_path, 'bucket', 0.1,
pa.float32()).copy({'read.native.enabled': 'false'})
+ _delete(table, [0])
+ entry = _entries(table, 2)[0]
+ factory = table.path_factory()
+ if 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)
+ 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
+ with patch.object(table, 'path_factory', return_value=factory):
+ with pytest.raises(FileNotFoundError, match=file.file_name):
+ _read(table, 'python', 2, [1, 2, 3])
+ assert table.file_io.exists(python_path)
@pytest.mark.parametrize('layout', ['bucket', 'bucket-external',
'global-external'])
def test_abort_removes_uncommitted_dv_from_its_actual_directory(tmp_path,
layout):
- table = _table(tmp_path, layout)
+ table = _table(tmp_path, layout, 'a/b')
_delete(table, [0])
builder = table.new_batch_write_builder()
messages = builder.new_update().delete_by_row_id([1])
@@ -208,28 +241,39 @@ def
test_missing_explicit_dv_is_not_replaced_by_a_local_copy(tmp_path, planner):
@pytest.mark.parametrize('planner', _PLANNERS)
[email protected]('partition', ['a/b', 'a%2Fb', 'a=b', 'a#b', 'a b',
'中文'])
-def test_bucket_dv_in_partition_requiring_path_escaping(tmp_path, planner,
partition):
- table = _table(tmp_path, 'bucket', partition)
[email protected]('layout', ['bucket', 'bucket-external'])
[email protected]('partition,escaped', [
+ ('a/b', 'a%2Fb'), ('a%2Fb', 'a%252Fb'), ('a=b', 'a%3Db'),
+ ('a#b', 'a%23b'), ('a b', 'a b'), ('中文', '中文')])
+def test_bucket_dv_in_partition_requiring_path_escaping(tmp_path, planner,
layout, partition, escaped):
+ table = _table(tmp_path, layout, partition)
_delete(table, [0])
- file = _entries(table, 2)[0].index_file
- if partition in ['a b', '中文']:
- assert file.external_path is None
- else:
- assert file.external_path is not None
- assert table.file_io.exists(file.external_path)
+ entry = _entries(table, 2)[0]
+ file = entry.index_file
+ assert (file.external_path is not None) == (layout == 'bucket-external')
+ root = tmp_path / 'external-data' if layout == 'bucket-external' else
Path(table.table_path)
+ path = root / ('p=' + escaped) / ('bucket-' + str(entry.bucket)) /
file.file_name
+ assert path.is_file()
_read(table, planner, 2, [1, 2, 3])
@pytest.mark.parametrize('planner', _PLANNERS)
[email protected]('partition_type,value', [(pa.bool_(), True),
(pa.float32(), 0.1), (pa.float64(), 0.1)],
- ids=['BOOLEAN', 'FLOAT', 'DOUBLE'])
-def test_bucket_dv_preserves_python_typed_partition_directory(tmp_path,
planner, partition_type, value):
[email protected]('partition_type,value,canonical_name', [
+ (pa.bool_(), True, 'true'), (pa.float32(), 0.1, '0.1'), (pa.float64(),
0.1, '0.1'),
+ (pa.float32(), 1.1754943508222875e-38, '1.1754944E-38'),
+ (pa.float64(), 1e23, '1.0E23')])
+def test_bucket_dv_writes_java_typed_partition_directory(tmp_path, planner,
partition_type, value, canonical_name):
table = _table(tmp_path, 'bucket', value, partition_type)
_delete(table, [0])
- file = _entries(table, 2)[0].index_file
- assert file.external_path is not None
- assert table.file_io.exists(file.external_path)
+ entry = _entries(table, 2)[0]
+ file = entry.index_file
+ assert (file.external_path is not None) ==
pa.types.is_floating(partition_type)
+ 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
@@ -384,9 +428,6 @@ def _check_java_bucket_dv(tmp_path, planner, value,
partition_type, canonical_na
# Case-insensitive local filesystems consider p=True and p=true identical.
if not Path(old_path).samefile(canonical_path):
table.file_io.delete_quietly(old_path)
- # Both previous Python locations must lose to every Java spelling.
- with table.file_io.new_output_stream(old_path) as stream:
- stream.write(b'not the canonical deletion vector')
with table.file_io.new_output_stream(table.path_factory().index_path() +
'/' + old.index_file.file_name) as stream:
stream.write(b'not a bucket deletion vector')
if partition_type is not None and pa.types.is_floating(partition_type):
diff --git a/paimon-python/pypaimon/tests/native_dynamic_write_test.py
b/paimon-python/pypaimon/tests/native_dynamic_write_test.py
new file mode 100644
index 0000000000..11ec98ddbf
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_dynamic_write_test.py
@@ -0,0 +1,280 @@
+# 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.
+
+"""Dynamic bucket writer and HASH index interoperability with Rust core."""
+
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.index.index_file_handler import IndexFileHandler
+from pypaimon.write.native_commit import from_native_commit_messages
+from pypaimon.write.native_write import NativeTableWrite
+
+pytestmark = pytest.mark.native_plan
+SCHEMA = pa.schema([('id', pa.int64()), ('p', pa.string()), ('v',
pa.string())])
+
+
+def _table(tmp_path, options=None, catalog=None):
+ catalog = catalog or CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ opts = {'bucket': '-1', 'file.format': 'parquet',
+ 'dynamic-bucket.target-row-num': '2', 'write.native.enabled':
'true'}
+ opts.update(options or {})
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ SCHEMA, primary_keys=['id', 'p'], partition_keys=['p'], options=opts),
False)
+ return catalog.get_table('default.t')
+
+
+def _batch(ids, p='a', value='v'):
+ return pa.RecordBatch.from_pylist(
+ [{'id': i, 'p': p, 'v': value} for i in ids], schema=SCHEMA)
+
+
+def _indexes(table):
+ return
IndexFileHandler(table).scan(table.snapshot_manager().get_latest_snapshot())
+
+
+def _write(table, batch, native=True):
+ table = table.copy({'write.native.enabled': str(native).lower()})
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ assert isinstance(writer, NativeTableWrite) == native
+ try:
+ writer.write_arrow_batch(batch)
+ if native:
+ assert writer._python_writer is None
+ messages = writer.prepare_commit()
+ commit.commit(messages)
+ return messages
+ finally:
+ writer.close()
+ commit.close()
+
+
+def _rows(table, native):
+ builder = table.copy({'scan.native-plan.enabled': str(native).lower(),
+ 'read.native.enabled':
str(native).lower()}).new_read_builder()
+ splits = builder.new_scan().plan().splits()
+ reader = builder.new_read()
+ if native:
+ with patch.object(reader, '_create_split_read',
side_effect=AssertionError('fallback')):
+ rows = reader.to_arrow(splits).to_pylist()
+ else:
+ rows = reader.to_arrow(splits).to_pylist()
+ return sorted(rows, key=lambda row: (row['p'], row['id']))
+
+
+def test_python_commit_accepts_java_hash_replacement(tmp_path):
+ from pypaimon_rust.datafusion import PaimonCatalog
+
+ table = _table(tmp_path, {'dynamic-bucket.target-row-num': '100'})
+ for ids in ([1], [2]):
+ native = PaimonCatalog({'warehouse':
str(tmp_path)}).get_table('default.t')
+ builder = native.new_batch_write_builder()
+ writer = builder.new_write()
+ writer.write_arrow(_batch(ids))
+ messages = from_native_commit_messages(table, writer.prepare_commit())
+ assert len(messages) == 1
+ assert len(messages[0].index_adds) == 1
+ assert messages[0].index_deletes == []
+ table.new_batch_write_builder().new_commit().commit(messages)
+ writer.close()
+ assert len(_indexes(table)) == 1
+ assert _indexes(table)[0].index_file.row_count == 2
+ for native in (False, True):
+ assert _rows(table, native) == _batch([1, 2]).to_pylist()
+
+
[email protected]('bucket_local', [False, True])
[email protected]('first_native', [False, True])
+def test_dynamic_cross_backend_restart(tmp_path, bucket_local, first_native):
+ table = _table(tmp_path, {
+ 'index-file-in-data-file-dir': str(bucket_local).lower(),
+ 'data-file.path-directory': 'data',
+ 'dynamic-bucket.max-buckets': '1',
+ })
+ for native, ids, value in [(first_native, [1, 2, 3], 'old'),
+ (not first_native, [2, 4], 'new'),
+ (first_native, [1, 5], 'last')]:
+ messages = _write(table, _batch(ids, 'a/b', value), native)
+ assert {m.bucket for m in messages} == {0}
+ indexes = _indexes(table)
+ 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)
+ 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
+ [(1, 'last'), (2, 'new'), (3, 'old'), (4, 'new'), (5, 'last')]]
+ for native in (False, True):
+ assert _rows(table, native) == expected
+
+
[email protected]('native_commit', [False, True])
+def test_stream_dynamic_index_checkpoints(tmp_path, native_rest_catalog,
native_commit):
+ table = _table(tmp_path, {'commit.native.enabled':
str(native_commit).lower(),
+ 'dynamic-bucket.max-buckets': '1'},
native_rest_catalog)
+ builder = table.new_stream_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ assert isinstance(writer, NativeTableWrite)
+ try:
+ for checkpoint, ids in enumerate(([1, 2], [2], [3]), 1):
+ writer.write_arrow_batch(_batch(ids, value=str(checkpoint)))
+ messages = writer.prepare_commit(checkpoint)
+ adds = [entry for m in messages for entry in m.index_adds]
+ assert len(adds) == (0 if checkpoint == 2 else 1)
+ if native_commit:
+ with patch.object(commit.file_store_commit, 'commit',
+ side_effect=AssertionError('Python commit
fallback')):
+ commit.commit(messages, checkpoint)
+ else:
+ commit.commit(messages, checkpoint)
+ assert writer.prepare_commit(4) == []
+ assert _indexes(table)[0].index_file.row_count == 3
+ finally:
+ writer.close()
+ commit.close()
+ for native in (False, True):
+ assert _rows(table, native) == [
+ {'id': 1, 'p': 'a', 'v': '1'}, {'id': 2, 'p': 'a', 'v': '2'},
+ {'id': 3, 'p': 'a', 'v': '3'}]
+
+
[email protected]('bucket_local', [False, True])
[email protected]('commit_first', [False, True])
+def test_dynamic_abort_index_ownership(tmp_path, bucket_local, commit_first):
+ table = _table(tmp_path, {'index-file-in-data-file-dir':
str(bucket_local).lower()})
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ assert isinstance(writer, NativeTableWrite)
+ writer.write_arrow_batch(_batch([1]))
+ messages = writer.prepare_commit()
+ entry = messages[0].index_adds[0]
+ path = table.path_factory().bucket_index_path(('a',), entry.bucket,
entry.index_file)
+ assert table.file_io.exists(path)
+ if commit_first:
+ builder.new_commit().commit(messages)
+ writer.abort()
+ assert table.file_io.exists(path) == commit_first
+ assert all(table.file_io.exists(f.file_path) == commit_first
+ for m in messages for f in m.new_files)
+
+
[email protected]('native', [False, True])
+def test_hash_index_requires_configured_directory(tmp_path, native):
+ # A file in table/index cannot replace a missing bucket-local index.
+ table = _table(tmp_path, {'index-file-in-data-file-dir': 'false'})
+ _write(table, _batch([1]), False)
+ old = _indexes(table)[0]
+ assert old.index_file.external_path is None
+ table = table.copy({'index-file-in-data-file-dir': 'true'})
+ with pytest.raises((FileNotFoundError, ValueError)) as error:
+ _write(table, _batch([2]), native)
+ assert old.index_file.file_name in str(error.value)
+ assert table.snapshot_manager().get_latest_snapshot().id == 1
+ assert _indexes(table) == [old]
+
+
[email protected]('missing_all', [False, True])
+def test_native_rejects_incomplete_hash_index(tmp_path, missing_all):
+ table = _table(tmp_path, {'dynamic-bucket.target-row-num': '1'})
+ builder = table.copy({'write.native.enabled':
'false'}).new_batch_write_builder()
+ writer = builder.new_write()
+ writer.write_arrow_batch(_batch([1, 2]))
+ messages = writer.prepare_commit()
+ for message in messages:
+ if missing_all or message.bucket == 1:
+ message.index_adds.clear()
+ builder.new_commit().commit(messages)
+ writer.close()
+ writer = table.new_batch_write_builder().new_write()
+ assert isinstance(writer, NativeTableWrite)
+ try:
+ with pytest.raises(Exception, match='complete HASH index'):
+ writer.write_arrow_batch(_batch([2], value='update'))
+ with pytest.raises(Exception, match='cannot be reused'):
+ writer.prepare_commit()
+ finally:
+ writer.close()
+ assert table.snapshot_manager().get_latest_snapshot().id == 1
+
+
[email protected]('native_commit', [False, True])
[email protected]('dynamic', [False, True])
+def test_dynamic_overwrite_resets_only_replaced_indexes(
+ tmp_path, native_rest_catalog, native_commit, dynamic):
+ table = _table(
+ tmp_path, {'commit.native.enabled': str(native_commit).lower(),
+ 'dynamic-partition-overwrite': str(dynamic).lower()},
native_rest_catalog)
+ _write(table, _batch([1, 2], 'a'))
+ _write(table, _batch([3], 'b'))
+ previous_b = next(e.index_file.file_name for e in _indexes(table) if
e.partition.values == ['b'])
+ builder = table.new_batch_write_builder().overwrite({'p': 'a'})
+ writer = builder.new_write()
+ assert isinstance(writer, NativeTableWrite)
+ writer.write_arrow_batch(_batch([4], 'a'))
+ builder.new_commit().commit(writer.prepare_commit())
+ writer.close()
+ counts = sorted((tuple(e.partition.values), e.index_file.row_count) for e
in _indexes(table))
+ assert counts == [(('a',), 1), (('b',), 1)]
+ assert next(e.index_file.file_name for e in _indexes(table) if
e.partition.values == ['b']) == previous_b
+ _write(table, _batch([4, 5], 'a', 'next'))
+ for native in (False, True):
+ assert _rows(table, native) == _batch([4, 5], 'a', 'next').to_pylist()
+ _batch([3], 'b').to_pylist()
+ # Static empty overwrite removes the old partition's HASH as well as data.
+ builder = table.copy({'dynamic-partition-overwrite':
'false'}).new_batch_write_builder().overwrite({'p': 'a'})
+ writer = builder.new_write()
+ builder.new_commit().commit(writer.prepare_commit())
+ writer.close()
+ assert [e.partition.values for e in _indexes(table)] == [['b']]
+
+
[email protected]('native', [False, True])
[email protected]('bucket_key', ['id', ''])
+def test_dynamic_writer_rejects_custom_bucket_key(tmp_path, native,
bucket_key):
+ table = _table(tmp_path).copy({'write.native.enabled': str(native).lower(),
+ 'bucket-key': bucket_key})
+ with pytest.raises(Exception, match="Cannot define 'bucket-key'"):
+ table.new_batch_write_builder().new_write()
+
+
[email protected]('bucket_local', [False, True])
+def test_dynamic_external_indexes_cross_backends(tmp_path, bucket_local):
+ external = tmp_path / 'external-index'
+ options = {'index-file-in-data-file-dir': str(bucket_local).lower(),
+ 'data-file.path-directory': 'data',
+ 'dynamic-bucket.max-buckets': '1'}
+ if bucket_local:
+ options.update({'data-file.external-paths': external.as_uri(),
+ 'data-file.external-paths.strategy': 'round-robin'})
+ else:
+ options['global-index.external-path'] = external.as_uri()
+ table = _table(tmp_path, options)
+ for native, ids in [(True, [1]), (False, [2]), (True, [3])]:
+ _write(table, _batch(ids), native)
+ indexes = _indexes(table)
+ assert len(indexes) == 1
+ path = indexes[0].index_file.external_path
+ assert path is not None
+ assert '/external-index/' in path
+ assert ('/data/p=a/bucket-0/' in path) == bucket_local
+ assert table.file_io.exists(path)
+ for native in (False, True):
+ assert _rows(table, native) == _batch([1, 2, 3]).to_pylist()
diff --git a/paimon-python/pypaimon/tests/native_plan_dynamic_bucket_test.py
b/paimon-python/pypaimon/tests/native_plan_dynamic_bucket_test.py
index f4d58b6422..beb241f6df 100644
--- a/paimon-python/pypaimon/tests/native_plan_dynamic_bucket_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_dynamic_bucket_test.py
@@ -29,7 +29,7 @@ from pypaimon.read.native_plan import native_method_available
pytestmark = [pytest.mark.native_plan, pytest.mark.skipif(
- not native_method_available('Table', 'new_write_builder'),
+ not native_method_available('Table', 'new_batch_write_builder'),
reason='Rust main writer and planner required')]
@@ -84,7 +84,7 @@ def
test_native_pk_bucket_growth_and_partition_migration(tmp_path, cross_partiti
# The Rust writer maintains the dynamic/cross-partition hash index and
# emits the old-partition DELETE when a key moves.
table = PaimonCatalog({'warehouse':
str(tmp_path)}).get_table('default.t')
- builder = table.new_write_builder()
+ builder = table.new_batch_write_builder()
writer = builder.new_write()
writer.write_arrow(pa.RecordBatch.from_pylist(rows, schema=schema))
builder.new_commit().commit(writer.prepare_commit())
diff --git a/paimon-python/pypaimon/tests/native_write_test.py
b/paimon-python/pypaimon/tests/native_write_test.py
index ea6646e457..8403d06c4d 100644
--- a/paimon-python/pypaimon/tests/native_write_test.py
+++ b/paimon-python/pypaimon/tests/native_write_test.py
@@ -366,15 +366,8 @@ def
test_external_data_paths_fall_back_before_native_write(tmp_path):
writer.close()
[email protected]('options', [
- {'bucket': '-1'},
- {'changelog-file.format': 'orc'},
-])
-def test_unsupported_primary_key_write_falls_back_before_native_reconstruction(
- tmp_path, options):
- table = (_table(tmp_path, primary_key=True, table_options=options)
- if 'bucket' in options else
- _table(tmp_path, primary_key=True).copy(options))
+def
test_unsupported_primary_key_write_falls_back_before_native_reconstruction(tmp_path):
+ table = _table(tmp_path, primary_key=True).copy({'changelog-file.format':
'orc'})
with patch('pypaimon.write.native_write.native_write_available',
return_value=True), \
patch('pypaimon.write.native_write.create_native_write_table',
side_effect=AssertionError('must not reconstruct')):
diff --git a/paimon-python/pypaimon/tests/write/dynamic_bucket_test.py
b/paimon-python/pypaimon/tests/write/dynamic_bucket_test.py
index 7eb7f2049e..e32d5f5a72 100644
--- a/paimon-python/pypaimon/tests/write/dynamic_bucket_test.py
+++ b/paimon-python/pypaimon/tests/write/dynamic_bucket_test.py
@@ -23,6 +23,7 @@ from decimal import Decimal
from unittest.mock import Mock, patch
import pyarrow as pa
+import pytest
from pypaimon import CatalogFactory, Schema
from pypaimon.index.dynamic_bucket import (
@@ -343,6 +344,7 @@ class DynamicBucketTest(unittest.TestCase):
result = self._read_arrow(table).sort_by('id').to_pydict()
self.assertEqual({'id': [1, 2], 'value': ['new', 'other']}, result)
+ @pytest.mark.python_write
def test_regular_dynamic_writer_retains_only_requested_index_hashes(self):
with tempfile.TemporaryDirectory() as root:
table = self._create_table(root, 'bounded_restore')
@@ -357,6 +359,7 @@ class DynamicBucketTest(unittest.TestCase):
self.assertEqual(1, len(partition_index.hash_to_bucket))
self.assertEqual({},
writer.row_key_extractor._index_maintainer._states)
+ @pytest.mark.python_write
def test_legacy_dynamic_data_without_hash_index_fails_fast(self):
with tempfile.TemporaryDirectory() as root:
table = self._create_table(root, 'legacy_no_index')
@@ -416,6 +419,7 @@ class DynamicBucketTest(unittest.TestCase):
self.assertFalse(table.file_io.exists(index_path))
+ @pytest.mark.python_write
def test_stream_writer_releases_prepared_hash_index_ownership(self):
with tempfile.TemporaryDirectory() as root:
table = self._create_table(root, 'stream_prepared')
@@ -481,44 +485,21 @@ class DynamicBucketTest(unittest.TestCase):
[call.args[0] for call in hash_code.call_args_list],
)
- def test_concurrent_initial_hash_index_add_conflicts(self):
+ def test_hash_add_replaces_previous_index_without_explicit_delete(self):
+ # Java DynamicBucketIndexMaintainer sends only the complete new HASH
+ # file. Its bucket-owner protocol does not require an old-file DELETE.
with tempfile.TemporaryDirectory() as root:
- table = self._create_table(root, 'concurrent_initial')
- writer1, commit1, messages1 = self._prepare_indexed_write(
- table, [1]
- )
- writer2, commit2, messages2 = self._prepare_indexed_write(
- table, [2]
- )
-
- commit1.commit(messages1)
- stale_data_paths = [
- file.file_path for message in messages2 for file in
message.new_files
- ]
- stale_index_paths = [
- entry.index_file.external_path
- or table.path_factory().global_index_path_factory().to_path(
- entry.index_file.file_name
- )
- for message in messages2 for entry in message.index_adds
- ]
- with self.assertRaisesRegex(
- RuntimeError, 'HASH index conflict detected'
- ):
- commit2.commit(messages2)
-
- self.assertTrue(all(
- table.file_io.exists(path)
- for path in stale_data_paths + stale_index_paths
- ))
-
+ table = self._create_table(root, 'implicit_hash_replace')
+ self._commit_arrow(table, [1], ['one'])
+ writer, commit, messages = self._prepare_indexed_write(table, [2])
+ self.assertEqual(1, len(messages[0].index_deletes))
+ messages[0].index_deletes.clear()
+ commit.commit(messages)
indexes = self._hash_indexes(table)
self.assertEqual(1, len(indexes))
- self.assertEqual(1, indexes[0].index_file.row_count)
- writer1.close()
- writer2.close()
- commit1.close()
- commit2.close()
+ self.assertEqual(2, indexes[0].index_file.row_count)
+ writer.close()
+ commit.close()
def test_concurrent_hash_index_replacement_conflicts(self):
with tempfile.TemporaryDirectory() as root:
@@ -638,6 +619,7 @@ class DynamicBucketTest(unittest.TestCase):
def test_retry_then_hash_index_conflict_preserves_prepared_files(self):
with tempfile.TemporaryDirectory() as root:
table = self._create_table(root, 'retry_hash_conflict')
+ self._commit_arrow(table, [0], ['seed'])
writer, commit, messages = self._prepare_indexed_write(table, [1])
prepared_paths = [
file.file_path
@@ -694,7 +676,7 @@ class DynamicBucketTest(unittest.TestCase):
invalid_schema = table.table_schema.copy(options)
with self.assertRaisesRegex(
- ValueError, 'bucket-key references unknown columns'
+ ValueError, "Cannot define 'bucket-key' in dynamic bucket mode"
):
DynamicBucketRowKeyExtractor(invalid_schema)
diff --git a/paimon-python/pypaimon/utils/file_store_path_factory.py
b/paimon-python/pypaimon/utils/file_store_path_factory.py
index 15cfcceb22..1536ffd93c 100644
--- a/paimon-python/pypaimon/utils/file_store_path_factory.py
+++ b/paimon-python/pypaimon/utils/file_store_path_factory.py
@@ -245,12 +245,12 @@ class FileStorePathFactory:
return to_file_io_path(path) if self.data_file_path_directory is not
None else path
def create_external_path_provider(
- self, partition: Tuple, bucket: int
+ self, partition: Tuple, bucket: int, canonical_partition: bool = False
) -> Optional[ExternalPathProvider]:
if not self.external_paths:
return None
- relative_bucket_path = self.relative_bucket_path(partition, bucket)
+ relative_bucket_path = self.relative_bucket_path(partition, bucket,
canonical_partition)
return ExternalPathProvider.create(
self.external_path_strategy,
self.external_paths,
@@ -268,42 +268,33 @@ class FileStorePathFactory:
def new_bucket_index_path(self, partition: Tuple, bucket: int, file_name:
str) -> Tuple[str, bool]:
"""Return a new bucket index's path and whether to persist its
external location."""
if self.index_file_in_data_file_dir:
- external = self.create_external_path_provider(partition, bucket)
+ external = self.create_external_path_provider(partition, bucket,
canonical_partition=True)
if external is not None:
return external.get_next_external_data_path(file_name), True
- # Python data directories historically use str(value) without
- # escaping. Record the actual location when Java renders it
- # differently, so its readers can find the DV beside those files.
- return (f"{self.bucket_path(partition, bucket)}/{file_name}",
- self._partition_path_requires_explicit_location(partition))
+ 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
factory = self.global_index_path_factory()
return factory.to_path(file_name), factory.is_external_path()
- def _partition_path_requires_explicit_location(self, partition: Tuple) ->
bool:
- # FLOAT/DOUBLE spellings also vary between JVM versions, so persist
- # their actual Python location even if one Java spelling matches it.
- return (any(isinstance(value, float) for value in partition)
- or self.relative_bucket_path(partition, 0) !=
self.relative_bucket_path(partition, 0, True))
-
def bucket_index_path(self, partition: Tuple, bucket: int, index_file,
file_io=None) -> str:
- """Resolve an existing bucket index, including the legacy Python DV
layout."""
+ """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)
- legacy_path = f"{self.index_path()}/{index_file.file_name}"
if not self.index_file_in_data_file_dir:
- return legacy_path
- path = f"{self.bucket_path(partition, bucket,
True)}/{index_file.file_name}"
- # Older Python DV writers ignored the option. Prefer the Java location
- # when present, and use the old directory only for an existing DV file.
- if file_io is not None and index_file.index_type == 'DELETION_VECTORS'
and not file_io.exists(path):
- python_path = f"{self.bucket_path(partition,
bucket)}/{index_file.file_name}"
+ 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
- if python_path != path and file_io.exists(python_path):
- return python_path
- if file_io.exists(legacy_path):
- return legacy_path
return path
def _find_floating_bucket_index(self, partition, bucket, file_name,
file_io, python_path):
@@ -314,7 +305,7 @@ class FileStorePathFactory:
# 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 = [self.data_file_path()]
+ 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:
@@ -358,7 +349,7 @@ class IndexPathFactory:
self._file_count = 0
def index_path(self) -> str:
- """Return the table index path used as a read fallback."""
+ """Return the default table index directory."""
return self._index_path
def global_index_root_path(self) -> str:
diff --git a/paimon-python/pypaimon/write/commit/conflict_detection.py
b/paimon-python/pypaimon/write/commit/conflict_detection.py
index 2792c84fca..f15917c3c2 100644
--- a/paimon-python/pypaimon/write/commit/conflict_detection.py
+++ b/paimon-python/pypaimon/write/commit/conflict_detection.py
@@ -301,7 +301,12 @@ class ConflictDetection:
def check_hash_index_conflicts(
self, latest_snapshot, delta_index_entries=None):
- """Detect stale full-file replacements of dynamic-bucket HASH
indexes."""
+ """Validate explicit HASH deletes and reject competing additions.
+
+ Java/Rust ADDs replace the complete bucket index without a DELETE.
+ A retained file alone does not prove a conflict; that protocol requires
+ one writer owner per bucket. Explicit DELETEs still detect stale
writers.
+ """
hash_entries = [
entry for entry in (delta_index_entries or [])
if entry.index_file.index_type == IndexManifestFile.HASH_INDEX
@@ -311,10 +316,6 @@ class ConflictDetection:
delete_entries = [entry for entry in hash_entries if entry.kind == 1]
add_entries = [entry for entry in hash_entries if entry.kind == 0]
- delete_names = {
- entry.index_file.file_name for entry in delete_entries
- }
-
current_entries = []
if latest_snapshot is not None and latest_snapshot.index_manifest is
not None:
current_entries = [
@@ -353,22 +354,6 @@ class ConflictDetection:
)
additions_by_bucket[key] = add
- retained = [
- entry for entry in current_entries
- if entry.index_file.file_name not in delete_names
- and tuple(entry.partition.values) == key[0]
- and entry.bucket == key[1]
- ]
- if retained:
- return RuntimeError(
- "HASH index conflict detected: partition {}, bucket {} "
- "already has newer index file {}.".format(
- key[0],
- key[1],
- retained[0].index_file.file_name,
- )
- )
-
return None
def check_deletion_vector_index_conflicts(self,
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index 1b70cf8a53..d7396d367a 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -108,7 +108,7 @@ def _abort_commit_messages(table, commit_messages:
List[CommitMessage]):
try:
index_file = entry.index_file
file_name = index_file.file_name
- if index_file.index_type == 'DELETION_VECTORS':
+ if index_file.index_type in ('DELETION_VECTORS', 'HASH'):
path = table.path_factory().bucket_index_path(
tuple(entry.partition.values), entry.bucket,
index_file, table.file_io)
else:
diff --git a/paimon-python/pypaimon/write/native_write.py
b/paimon-python/pypaimon/write/native_write.py
index 8e00e13171..ca56bd6f7e 100644
--- a/paimon-python/pypaimon/write/native_write.py
+++ b/paimon-python/pypaimon/write/native_write.py
@@ -71,7 +71,7 @@ def create_native_write(table, commit_user,
static_partition=None, stream=False)
# Rust does not produce the optional random-access .row sidecars.
or (table.options.data_evolution_enabled()
and table.options.data_evolution_row_sidecar_enabled())
- or table.bucket_mode() not in (BucketMode.HASH_FIXED,
+ or table.bucket_mode() not in (BucketMode.HASH_FIXED,
BucketMode.HASH_DYNAMIC,
BucketMode.BUCKET_UNAWARE)
or (table.options.deletion_vectors_enabled()
and table.options.merge_engine() in
(MergeEngine.PARTIAL_UPDATE,
diff --git a/paimon-python/pypaimon/write/row_key_extractor.py
b/paimon-python/pypaimon/write/row_key_extractor.py
index d5053e983f..16c46fa6f2 100644
--- a/paimon-python/pypaimon/write/row_key_extractor.py
+++ b/paimon-python/pypaimon/write/row_key_extractor.py
@@ -443,6 +443,9 @@ class DynamicBucketRowKeyExtractor(RowKeyExtractor):
f"'DynamicBucketRowKeyExtractor', but found: {num_buckets}"
)
+ if table_schema.options.get(CoreOptions.BUCKET_KEY.key()) is not None:
+ raise ValueError("Cannot define 'bucket-key' in dynamic bucket
mode")
+
opts = CoreOptions.from_dict(table_schema.options)
self._table = table
self.base_snapshot_id = 0
