From 7557c7e1448a17a5bbfe92993cb03e783d4e1705 Mon Sep 17 00:00:00 2001 From: zhengguangzhuo <100548634+zhengguangzhuo@users.noreply.github.com> Date: Tue, 29 Sep 2026 18:01:54 +0800 Subject: [PATCH 1/5] [python] Align DATE partition paths with Java --- .../tests/date_partition_path_test.py | 157 ++++++++++++++++++ .../pypaimon/tests/file_store_commit_test.py | 4 +- .../tests/mosaic_writer_options_test.py | 3 + .../pypaimon/utils/file_store_path_factory.py | 32 +++- .../pypaimon/write/file_store_commit.py | 4 +- .../pypaimon/write/writer/data_writer.py | 3 +- 6 files changed, 196 insertions(+), 7 deletions(-) create mode 100644 paimon-python/pypaimon/tests/date_partition_path_test.py diff --git a/paimon-python/pypaimon/tests/date_partition_path_test.py b/paimon-python/pypaimon/tests/date_partition_path_test.py new file mode 100644 index 000000000000..44f2648e6cbc --- /dev/null +++ b/paimon-python/pypaimon/tests/date_partition_path_test.py @@ -0,0 +1,157 @@ +# 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. + +"""验证 DATE 分区路径与 Java 命名规则的一致性及历史目录兼容性。""" + +from datetime import date +from pathlib import Path + +import pyarrow as pa + +from pypaimon.common.identifier import Identifier +from pypaimon.filesystem.local_file_io import LocalFileIO +from pypaimon.schema.data_types import AtomicType +from pypaimon.schema.schema import Schema +from pypaimon.schema.schema_manager import SchemaManager +from pypaimon.table.file_store_table import FileStoreTable +from pypaimon.utils.file_store_path_factory import FileStorePathFactory + + +def _create_table(tmp_path, legacy_partition_name=True): + table_path = str(tmp_path / "table") + file_io = LocalFileIO() + arrow_schema = pa.schema([("id", pa.int64()), ("day", pa.date32())]) + schema = Schema.from_pyarrow_schema( + arrow_schema, + partition_keys=["day"], + options={ + "partition.legacy-name": str(legacy_partition_name).lower(), + "scan.native-plan.enabled": "false", + "write.native.enabled": "false", + }, + ) + table_schema = SchemaManager(file_io, table_path).create_table(schema) + table = FileStoreTable( + file_io, + Identifier.create("default", "t"), + table_path, + table_schema, + ) + return table, arrow_schema + + +def _write_one_row(table, arrow_schema): + builder = table.new_batch_write_builder() + writer, commit = builder.new_write(), builder.new_commit() + try: + writer.write_arrow( + pa.table( + {"id": [1], "day": [date(1970, 1, 2)]}, + schema=arrow_schema, + ) + ) + commit.commit(writer.prepare_commit()) + finally: + writer.close() + commit.close() + + +def _read_rows(table): + builder = table.new_read_builder() + splits = builder.new_scan().plan().splits() + return builder.new_read().to_arrow(splits).to_pylist() + + +def test_legacy_date_partition_write_uses_java_epoch_day(tmp_path): + table, arrow_schema = _create_table(tmp_path) + + _write_one_row(table, arrow_schema) + + partition_directories = sorted( + path.name for path in Path(table.table_path).glob("day=*") + ) + assert partition_directories == ["day=1"] + assert _read_rows(table) == [{"id": 1, "day": date(1970, 1, 2)}] + + +def test_legacy_python_date_partition_directory_remains_readable(tmp_path): + table, arrow_schema = _create_table(tmp_path) + + _write_one_row(table, arrow_schema) + canonical_directory = Path(table.table_path) / "day=1" + historical_directory = Path(table.table_path) / "day=1970-01-02" + canonical_directory.rename(historical_directory) + + assert _read_rows(table) == [{"id": 1, "day": date(1970, 1, 2)}] + + +def test_non_legacy_date_partition_write_keeps_iso_name(tmp_path): + table, arrow_schema = _create_table(tmp_path, legacy_partition_name=False) + + _write_one_row(table, arrow_schema) + + partition_directories = sorted( + path.name for path in Path(table.table_path).glob("day=*") + ) + assert partition_directories == ["day=1970-01-02"] + + +def test_date_compatibility_does_not_reformat_other_partition_types(): + factory = FileStorePathFactory( + "/table", + ["day", "region"], + "__DEFAULT_PARTITION__", + "parquet", + "data-", + "changelog-", + True, + False, + None, + partition_types=[AtomicType("DATE"), AtomicType("STRING")], + ) + + assert factory.data_file_relative_bucket_path( + (date(1970, 1, 2), "a/b"), 0 + ) == "day=1/region=a/b/bucket-0" + + +def test_date_compatibility_applies_to_external_data_and_bucket_index_paths(): + partition = (date(1970, 1, 2),) + factory = FileStorePathFactory( + "/table", + ["day"], + "__DEFAULT_PARTITION__", + "parquet", + "data-", + "changelog-", + True, + False, + None, + external_paths=["/external"], + external_path_strategy="round-robin", + index_file_in_data_file_dir=True, + partition_types=[AtomicType("DATE")], + ) + + external_provider = factory.create_external_path_provider(partition, 0) + assert external_provider.get_next_external_data_path("data.parquet") == ( + "/external/day=1/bucket-0/data.parquet" + ) + assert factory.new_bucket_index_path(partition, 0, "index-1") == ( + "/external/day=1/bucket-0/index-1", + True, + ) diff --git a/paimon-python/pypaimon/tests/file_store_commit_test.py b/paimon-python/pypaimon/tests/file_store_commit_test.py index 559b76e56785..b294effe7c06 100644 --- a/paimon-python/pypaimon/tests/file_store_commit_test.py +++ b/paimon-python/pypaimon/tests/file_store_commit_test.py @@ -91,7 +91,7 @@ def _file_meta(**kwargs): def test_reconstructs_local_path_after_wire_decode(self): table = Mock() - table.path_factory.return_value.bucket_path.return_value = '/table/p=1/bucket-0' + table.path_factory.return_value.data_file_bucket_path.return_value = '/table/p=1/bucket-0' file = self._file_meta() message = CommitMessage((1,), 0, [file]) _abort_commit_messages(table, [message]) @@ -100,7 +100,7 @@ def test_reconstructs_local_path_after_wire_decode(self): def test_reconstructs_aligned_sidecars_after_wire_decode(self): table = Mock() - table.path_factory.return_value.bucket_path.return_value = '/table/p=1/bucket-0' + table.path_factory.return_value.data_file_bucket_path.return_value = '/table/p=1/bucket-0' file = self._file_meta(extra_files=['data.parquet.index']) _abort_commit_messages(table, [CommitMessage((1,), 0, [file])]) self.assertEqual(table.file_io.delete_quietly.call_args_list, [ diff --git a/paimon-python/pypaimon/tests/mosaic_writer_options_test.py b/paimon-python/pypaimon/tests/mosaic_writer_options_test.py index 0eac7c38491d..14a00da27121 100644 --- a/paimon-python/pypaimon/tests/mosaic_writer_options_test.py +++ b/paimon-python/pypaimon/tests/mosaic_writer_options_test.py @@ -43,6 +43,9 @@ def create_external_path_provider(self, partition, bucket): def bucket_path(self, partition, bucket): return self._bucket_path + def data_file_bucket_path(self, partition, bucket): + return self._bucket_path + class _FileIO: def __init__(self): diff --git a/paimon-python/pypaimon/utils/file_store_path_factory.py b/paimon-python/pypaimon/utils/file_store_path_factory.py index 15cfcceb221b..e3050432dfde 100644 --- a/paimon-python/pypaimon/utils/file_store_path_factory.py +++ b/paimon-python/pypaimon/utils/file_store_path_factory.py @@ -244,13 +244,41 @@ def bucket_path(self, partition: Tuple, bucket: int, canonical_partition: bool = path = resolve_path(self._root, relative_path) return to_file_io_path(path) if self.data_file_path_directory is not None else path + def _data_file_partition(self, partition: Tuple) -> Tuple: + if not self.legacy_partition_name or not self.partition_types or not partition: + return partition + + date_indexes = [ + index + for index, data_type in enumerate(self.partition_types) + if str(data_type).split('(', 1)[0].split()[0] == 'DATE' + ] + if not date_indexes: + return partition + + data_partition = list(partition) + for index in date_indexes: + value = partition[index] + if not _is_null_or_whitespace_only(value): + data_partition[index] = str((value - date(1970, 1, 1)).days) + return tuple(data_partition) + + def data_file_relative_bucket_path(self, partition: Tuple, bucket: int) -> str: + """生成新数据文件使用的分区路径，并兼容历史非 DATE 格式。""" + return self.relative_bucket_path( + self._data_file_partition(partition), bucket) + + def data_file_bucket_path(self, partition: Tuple, bucket: int) -> str: + """生成新数据文件使用的完整 bucket 路径。""" + return self.bucket_path(self._data_file_partition(partition), bucket) + def create_external_path_provider( self, partition: Tuple, bucket: int ) -> Optional[ExternalPathProvider]: if not self.external_paths: return None - relative_bucket_path = self.relative_bucket_path(partition, bucket) + relative_bucket_path = self.data_file_relative_bucket_path(partition, bucket) return ExternalPathProvider.create( self.external_path_strategy, self.external_paths, @@ -274,7 +302,7 @@ def new_bucket_index_path(self, partition: Tuple, bucket: int, file_name: str) - # 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}", + return (f"{self.data_file_bucket_path(partition, bucket)}/{file_name}", self._partition_path_requires_explicit_location(partition)) factory = self.global_index_path_factory() return factory.to_path(file_name), factory.is_external_path() diff --git a/paimon-python/pypaimon/write/file_store_commit.py b/paimon-python/pypaimon/write/file_store_commit.py index 1b70cf8a5318..12cc1b499a67 100644 --- a/paimon-python/pypaimon/write/file_store_commit.py +++ b/paimon-python/pypaimon/write/file_store_commit.py @@ -93,7 +93,7 @@ def _abort_commit_messages(table, commit_messages: List[CommitMessage]): + list(message.compact_changelog_files)): path = None try: - bucket_path = None if file.physical_path() else table.path_factory().bucket_path( + bucket_path = None if file.physical_path() else table.path_factory().data_file_bucket_path( tuple(message.partition), message.bucket) for path in file.collect_files(bucket_path): table.file_io.delete_quietly(path) @@ -895,7 +895,7 @@ def _is_duplicate_commit( for entry in entries: file = entry.file file.file_path = file.physical_path() if file.external_path else "%s/%s" % ( - path_factory.bucket_path( + path_factory.data_file_bucket_path( tuple(entry.partition.values), entry.bucket, ).rstrip("/"), diff --git a/paimon-python/pypaimon/write/writer/data_writer.py b/paimon-python/pypaimon/write/writer/data_writer.py index 731727fbc25c..bbe852132136 100644 --- a/paimon-python/pypaimon/write/writer/data_writer.py +++ b/paimon-python/pypaimon/write/writer/data_writer.py @@ -490,7 +490,8 @@ def _generate_file_path(self, file_name: str) -> str: if self.external_path_provider: return self.external_path_provider.get_next_external_data_path(file_name) - bucket_path = self.path_factory.bucket_path(self.partition, self.bucket) + bucket_path = self.path_factory.data_file_bucket_path( + self.partition, self.bucket) return f"{bucket_path.rstrip('/')}/{file_name}" def _should_write_row_sidecar(self) -> bool: From 4181aa262f50e3a81ab4d09637d78d2ccd64afce Mon Sep 17 00:00:00 2001 From: zhengguangzhuo <100548634+zhengguangzhuo@users.noreply.github.com> Date: Thu, 8 Oct 2026 10:25:48 +0800 Subject: [PATCH 2/5] [python] Fix composite DATE partition paths --- .../java/org/apache/paimon/JavaPyE2ETest.java | 74 +++++++++++- paimon-python/dev/run_mixed_tests.sh | 9 +- .../tests/date_partition_path_test.py | 107 +++++++++++++++--- .../tests/e2e/java_py_read_write_test.py | 43 +++++++ .../tests/mosaic_writer_options_test.py | 7 +- .../pypaimon/write/writer/data_writer.py | 6 +- 6 files changed, 218 insertions(+), 28 deletions(-) diff --git a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java index 0cdcb9879106..c76fc9eee4ff 100644 --- a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java +++ b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java @@ -159,7 +159,11 @@ public void before() throws Exception { Files.createDirectories(tempDir.resolve("warehouse")); } - warehouse = new Path(TraceableFileIO.SCHEME + "://" + tempDir.resolve("warehouse")); + warehouse = + new Path( + TraceableFileIO.SCHEME + + "://" + + tempDir.resolve("warehouse").toUri().getPath()); catalog = CatalogFactory.createCatalog(CatalogContext.create(warehouse)); // Create database if it doesn't exist @@ -384,6 +388,37 @@ public void testJavaWriteDynamicBucketHashIndex() throws Exception { } } + @Test + @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true") + public void testJavaWriteCompositeDatePartition() throws Exception { + for (boolean legacyName : Arrays.asList(true, false)) { + String suffix = legacyName ? "legacy" : "canonical"; + Identifier identifier = identifier("composite_date_java_to_python_" + suffix); + catalog.dropTable(identifier, true); + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("day", DataTypes.DATE()) + .column("region", DataTypes.STRING()) + .partitionKeys("day", "region") + .option("partition.legacy-name", Boolean.toString(legacyName)) + .option("bucket", "1") + .option("bucket-key", "id") + .option("write.native.enabled", "false") + .option("commit.native.enabled", "false") + .option("scan.native-plan.enabled", "false") + .build(); + catalog.createTable(identifier, schema, false); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = writeBuilder.newWrite(); + BatchTableCommit commit = writeBuilder.newCommit()) { + write.write(GenericRow.of(1, 1, BinaryString.fromString("a/b"))); + commit.commit(write.prepareCommit()); + } + } + } + @Test @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true") public void testPKDeletionVectorWrite() throws Exception { @@ -672,6 +707,43 @@ public void testReadPythonDynamicBucketHashIndex() throws Exception { "hello-java, 42, java-new", "python-only, 7, python-only"); } + @Test + @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true") + public void testReadPythonCompositeDatePartition() throws Exception { + for (boolean legacyName : Arrays.asList(true, false)) { + String suffix = legacyName ? "legacy" : "canonical"; + FileStoreTable table = + (FileStoreTable) + catalog.getTable( + identifier("composite_date_java_to_python_" + suffix)); + List rows = + getResult( + table.newRead(), + table.newScan().plan().splits(), + row -> + row.getInt(0) + + ":" + + row.getInt(1) + + ":" + + row.getString(2).toString()); + assertThat(rows) + .containsExactlyInAnyOrder("1:1:a/b", "2:1:a/b"); + + table.rollbackTo(1L); + List rowsAfterRollback = + getResult( + table.newRead(), + table.newScan().plan().splits(), + row -> + row.getInt(0) + + ":" + + row.getInt(1) + + ":" + + row.getString(2).toString()); + assertThat(rowsAfterRollback).containsExactly("1:1:a/b"); + } + } + private int assignDynamicBucket(FileStoreTable table, InternalRow row) { InternalRowSerializer serializer = new InternalRowSerializer(DataTypes.STRING(), DataTypes.BIGINT()); diff --git a/paimon-python/dev/run_mixed_tests.sh b/paimon-python/dev/run_mixed_tests.sh index 84a54e21dbae..9b8784b973c1 100755 --- a/paimon-python/dev/run_mixed_tests.sh +++ b/paimon-python/dev/run_mixed_tests.sh @@ -99,6 +99,7 @@ run_batched_java_write_tests() { local core_tests="org.apache.paimon.JavaPyE2ETest#testJavaWriteReadPkTable" core_tests="${core_tests}+testJavaWriteDynamicBucketHashIndex" + core_tests="${core_tests}+testJavaWriteCompositeDatePartition" core_tests="${core_tests}+testPKDeletionVectorWrite" core_tests="${core_tests}+testBtreeIndexWrite" core_tests="${core_tests}+testBtreeRawFallbackWrite" @@ -187,7 +188,7 @@ run_java_write_test() { echo "Running Maven test for JavaPyE2ETest.testJavaWriteReadPkTable (Parquet/Orc/Avro)..." echo "Note: Maven may download dependencies on first run, this may take a while..." local parquet_result=0 - if mvn test -Dtest=org.apache.paimon.JavaPyE2ETest#testJavaWriteReadPkTable+testJavaWriteDynamicBucketHashIndex -pl paimon-core -Drun.e2e.tests=true; then + if mvn test -Dtest=org.apache.paimon.JavaPyE2ETest#testJavaWriteReadPkTable+testJavaWriteDynamicBucketHashIndex+testJavaWriteCompositeDatePartition -pl paimon-core -Drun.e2e.tests=true; then echo -e "${GREEN}✓ Java write Parquet/Orc/Avro test completed successfully${NC}" else echo -e "${RED}✗ Java write Parquet/Orc/Avro test failed${NC}" @@ -225,7 +226,7 @@ run_python_read_test() { # Run the parameterized Python test method (runs for both Parquet/Orc/Avro and Lance) echo "Running Python test for JavaPyReadWriteTest.test_read_pk_table..." - if python -m pytest java_py_read_write_test.py::JavaPyReadWriteTest -k "test_read_pk_table or test_read_java_dynamic_bucket_hash_index" -v; then + if python -m pytest java_py_read_write_test.py::JavaPyReadWriteTest -k "test_read_pk_table or test_read_java_dynamic_bucket_hash_index or test_read_java_composite_date_partition" -v; then echo -e "${GREEN}✓ Python test completed successfully${NC}" # source deactivate return 0 @@ -244,7 +245,7 @@ run_python_write_test() { # Run the parameterized Python test method for writing data (pk table, includes bucket num assertion) echo "Running Python test for JavaPyReadWriteTest (test_py_write_read_pk_table)..." - if python -m pytest java_py_read_write_test.py::JavaPyReadWriteTest -k "test_py_write_read_pk_table or test_py_write_dynamic_bucket_hash_index or test_py_write_floating_sequence" -v; then + if python -m pytest java_py_read_write_test.py::JavaPyReadWriteTest -k "test_py_write_read_pk_table or test_py_write_dynamic_bucket_hash_index or test_py_write_floating_sequence or test_py_write_composite_date_partition" -v; then echo -e "${GREEN}✓ Python write test completed successfully${NC}" return 0 else @@ -263,7 +264,7 @@ run_java_read_test() { echo "Running Maven test for JavaPyE2ETest.testReadPkTable (Java Read Parquet/Orc/Avro)..." echo "Note: Maven may download dependencies on first run, this may take a while..." local parquet_result=0 - if mvn test -Dtest=org.apache.paimon.JavaPyE2ETest#testReadPkTable+testReadPythonDynamicBucketHashIndex+testReadPythonFloatingSequence -pl paimon-core -Drun.e2e.tests=true -Dpython.version="$PYTHON_VERSION"; then + if mvn test -Dtest=org.apache.paimon.JavaPyE2ETest#testReadPkTable+testReadPythonDynamicBucketHashIndex+testReadPythonFloatingSequence+testReadPythonCompositeDatePartition -pl paimon-core -Drun.e2e.tests=true -Dpython.version="$PYTHON_VERSION"; then echo -e "${GREEN}✓ Java read Parquet/Orc/Avro test completed successfully${NC}" else echo -e "${RED}✗ Java read Parquet/Orc/Avro test failed${NC}" diff --git a/paimon-python/pypaimon/tests/date_partition_path_test.py b/paimon-python/pypaimon/tests/date_partition_path_test.py index 44f2648e6cbc..dc4c05e9eba7 100644 --- a/paimon-python/pypaimon/tests/date_partition_path_test.py +++ b/paimon-python/pypaimon/tests/date_partition_path_test.py @@ -15,10 +15,11 @@ # specific language governing permissions and limitations # under the License. -"""验证 DATE 分区路径与 Java 命名规则的一致性及历史目录兼容性。""" +"""Verify DATE partition paths match Java and preserve historical reads.""" from datetime import date from pathlib import Path +from unittest.mock import patch import pyarrow as pa @@ -29,19 +30,26 @@ from pypaimon.schema.schema_manager import SchemaManager from pypaimon.table.file_store_table import FileStoreTable from pypaimon.utils.file_store_path_factory import FileStorePathFactory +from pypaimon.write.writer.data_writer import DataWriter -def _create_table(tmp_path, legacy_partition_name=True): +def _create_table(tmp_path, legacy_partition_name=True, composite_partition=False): table_path = str(tmp_path / "table") file_io = LocalFileIO() - arrow_schema = pa.schema([("id", pa.int64()), ("day", pa.date32())]) + fields = [("id", pa.int64()), ("day", pa.date32())] + partition_keys = ["day"] + if composite_partition: + fields.append(("region", pa.string())) + partition_keys.append("region") + arrow_schema = pa.schema(fields) schema = Schema.from_pyarrow_schema( arrow_schema, - partition_keys=["day"], + partition_keys=partition_keys, options={ "partition.legacy-name": str(legacy_partition_name).lower(), "scan.native-plan.enabled": "false", "write.native.enabled": "false", + "commit.native.enabled": "false", }, ) table_schema = SchemaManager(file_io, table_path).create_table(schema) @@ -54,22 +62,37 @@ def _create_table(tmp_path, legacy_partition_name=True): return table, arrow_schema -def _write_one_row(table, arrow_schema): +def _write_rows(table, arrow_schema, rows): builder = table.new_batch_write_builder() writer, commit = builder.new_write(), builder.new_commit() try: - writer.write_arrow( - pa.table( - {"id": [1], "day": [date(1970, 1, 2)]}, - schema=arrow_schema, - ) - ) + writer.write_arrow(pa.Table.from_pylist(rows, schema=arrow_schema)) commit.commit(writer.prepare_commit()) finally: writer.close() commit.close() +def _write_one_row(table, arrow_schema): + _write_rows(table, arrow_schema, [{"id": 1, "day": date(1970, 1, 2)}]) + + +def _write_legacy_composite_row(table, arrow_schema, row): + historical_bucket = ( + Path(table.table_path) + / "day=1970-01-02" + / "region=a" + / "b" + / "bucket-0" + ) + with patch.object( + DataWriter, + "_generate_file_path", + lambda writer, file_name: str(historical_bucket / file_name), + ): + _write_rows(table, arrow_schema, [row]) + + def _read_rows(table): builder = table.new_read_builder() splits = builder.new_scan().plan().splits() @@ -110,6 +133,59 @@ def test_non_legacy_date_partition_write_keeps_iso_name(tmp_path): assert partition_directories == ["day=1970-01-02"] +def test_legacy_composite_date_write_reads_back_from_canonical_path(tmp_path): + table, arrow_schema = _create_table(tmp_path, composite_partition=True) + + _write_rows(table, arrow_schema, [{ + "id": 1, "day": date(1970, 1, 2), "region": "a/b", + }]) + + assert _read_rows(table) == [{ + "id": 1, "day": date(1970, 1, 2), "region": "a/b", + }] + assert (Path(table.table_path) / "day=1" / "region=a%2Fb" / "bucket-0").is_dir() + + +def test_old_and_new_composite_partition_files_coexist_after_append(tmp_path): + table, arrow_schema = _create_table(tmp_path, composite_partition=True) + row = {"day": date(1970, 1, 2), "region": "a/b"} + + _write_legacy_composite_row(table, arrow_schema, {"id": 1, **row}) + first_snapshot_id = table.snapshot_manager().get_latest_snapshot().id + historical_bucket = ( + Path(table.table_path) + / "day=1970-01-02" + / "region=a" + / "b" + / "bucket-0" + ) + assert historical_bucket.is_dir() + + _write_rows(table, arrow_schema, [{"id": 2, **row}]) + canonical_directory = Path(table.table_path) / "day=1" / "region=a%2Fb" + + assert {item["id"] for item in _read_rows(table)} == {1, 2} + assert canonical_directory.joinpath("bucket-0").is_dir() + + table.rollback_to(first_snapshot_id) + + assert [item["id"] for item in _read_rows(table)] == [1] + + +def test_non_legacy_composite_date_write_uses_canonical_escaping(tmp_path): + table, arrow_schema = _create_table( + tmp_path, legacy_partition_name=False, composite_partition=True) + + _write_rows(table, arrow_schema, [{ + "id": 1, "day": date(1970, 1, 2), "region": "a/b", + }]) + + assert _read_rows(table) == [{ + "id": 1, "day": date(1970, 1, 2), "region": "a/b", + }] + assert (Path(table.table_path) / "day=1970-01-02" / "region=a%2Fb" / "bucket-0").is_dir() + + def test_date_compatibility_does_not_reformat_other_partition_types(): factory = FileStorePathFactory( "/table", @@ -124,9 +200,9 @@ def test_date_compatibility_does_not_reformat_other_partition_types(): partition_types=[AtomicType("DATE"), AtomicType("STRING")], ) - assert factory.data_file_relative_bucket_path( - (date(1970, 1, 2), "a/b"), 0 - ) == "day=1/region=a/b/bucket-0" + assert factory.relative_bucket_path( + (date(1970, 1, 2), "a/b"), 0, canonical_partition=True + ) == "day=1/region=a%2Fb/bucket-0" def test_date_compatibility_applies_to_external_data_and_bucket_index_paths(): @@ -147,7 +223,8 @@ def test_date_compatibility_applies_to_external_data_and_bucket_index_paths(): partition_types=[AtomicType("DATE")], ) - external_provider = factory.create_external_path_provider(partition, 0) + external_provider = factory.create_external_path_provider( + partition, 0, canonical_partition=True) assert external_provider.get_next_external_data_path("data.parquet") == ( "/external/day=1/bucket-0/data.parquet" ) diff --git a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py index fa6ff03103bc..82b2e180369a 100644 --- a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py +++ b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py @@ -142,6 +142,49 @@ def test_py_write_dynamic_bucket_hash_index(self): writer.close() commit.close() + def test_read_java_composite_date_partition(self): + for legacy_name in (True, False): + suffix = 'legacy' if legacy_name else 'canonical' + table = self.catalog.get_table( + 'default.composite_date_java_to_python_' + suffix) + read_builder = table.new_read_builder() + result = read_builder.new_read().to_arrow( + read_builder.new_scan().plan().splits()) + self.assertEqual(result.to_pylist(), [{ + 'id': 1, + 'day': datetime.date(1970, 1, 2), + 'region': 'a/b', + }]) + + def test_py_write_composite_date_partition(self): + arrow_schema = pa.schema([ + ('id', pa.int32()), + ('day', pa.date32()), + ('region', pa.string()), + ]) + for legacy_name in (True, False): + suffix = 'legacy' if legacy_name else 'canonical' + table = self.catalog.get_table( + 'default.composite_date_java_to_python_' + suffix) + builder = table.new_batch_write_builder() + writer, commit = builder.new_write(), builder.new_commit() + try: + writer.write_arrow(pa.Table.from_pylist([{ + 'id': 2, + 'day': datetime.date(1970, 1, 2), + 'region': 'a/b', + }], schema=arrow_schema)) + commit.commit(writer.prepare_commit()) + finally: + writer.close() + commit.close() + + read_builder = table.new_read_builder() + result = read_builder.new_read().to_arrow( + read_builder.new_scan().plan().splits()) + self.assertEqual( + {row['id'] for row in result.to_pylist()}, {1, 2}) + @parameterized.expand([ (type_name, order, grouping) for type_name in ('float', 'double') diff --git a/paimon-python/pypaimon/tests/mosaic_writer_options_test.py b/paimon-python/pypaimon/tests/mosaic_writer_options_test.py index 14a00da27121..770f8c8bd300 100644 --- a/paimon-python/pypaimon/tests/mosaic_writer_options_test.py +++ b/paimon-python/pypaimon/tests/mosaic_writer_options_test.py @@ -37,13 +37,10 @@ class _PathFactory: def __init__(self, bucket_path): self._bucket_path = bucket_path - def create_external_path_provider(self, partition, bucket): + def create_external_path_provider(self, partition, bucket, canonical_partition=False): return None - def bucket_path(self, partition, bucket): - return self._bucket_path - - def data_file_bucket_path(self, partition, bucket): + def bucket_path(self, partition, bucket, canonical_partition=False): return self._bucket_path diff --git a/paimon-python/pypaimon/write/writer/data_writer.py b/paimon-python/pypaimon/write/writer/data_writer.py index 612c4784379b..ad7e3b897e71 100644 --- a/paimon-python/pypaimon/write/writer/data_writer.py +++ b/paimon-python/pypaimon/write/writer/data_writer.py @@ -95,7 +95,7 @@ def __init__(self, table, partition: Tuple, bucket: int, max_seq_number: int, op self.path_factory = self.table.path_factory() self.external_path_provider: Optional[ExternalPathProvider] = self.path_factory.create_external_path_provider( - self.partition, self.bucket + self.partition, self.bucket, canonical_partition=True ) # Variant shredding (static mode) — col_name → (obj_fields, target_arrow_type) self._variant_shredding: Dict[str, Tuple] = {} @@ -498,8 +498,8 @@ def _generate_file_path(self, file_name: str) -> str: if self.external_path_provider: return self.external_path_provider.get_next_external_data_path(file_name) - bucket_path = self.path_factory.data_file_bucket_path( - self.partition, self.bucket) + bucket_path = self.path_factory.bucket_path( + self.partition, self.bucket, canonical_partition=True) return f"{bucket_path.rstrip('/')}/{file_name}" def _should_write_row_sidecar(self) -> bool: From 1156834035e3d4d570381db871f284f6bfb85aef Mon Sep 17 00:00:00 2001 From: zhengguangzhuo <100548634+zhengguangzhuo@users.noreply.github.com> Date: Fri, 9 Oct 2026 09:58:57 +0800 Subject: [PATCH 3/5] [python] Fix DATE partition CI compatibility --- .../tests/date_partition_path_test.py | 24 +++++++++++++++++-- .../tests/e2e/java_py_read_write_test.py | 22 ++++++++--------- 2 files changed, 33 insertions(+), 13 deletions(-) diff --git a/paimon-python/pypaimon/tests/date_partition_path_test.py b/paimon-python/pypaimon/tests/date_partition_path_test.py index cf5df5fc8815..d09cec49d54c 100644 --- a/paimon-python/pypaimon/tests/date_partition_path_test.py +++ b/paimon-python/pypaimon/tests/date_partition_path_test.py @@ -22,6 +22,7 @@ from unittest.mock import patch import pyarrow as pa +import pytest from pypaimon.common.identifier import Identifier from pypaimon.filesystem.local_file_io import LocalFileIO @@ -66,7 +67,13 @@ def _write_rows(table, arrow_schema, rows): builder = table.new_batch_write_builder() writer, commit = builder.new_write(), builder.new_commit() try: - writer.write_arrow(pa.Table.from_pylist(rows, schema=arrow_schema)) + writer.write_arrow(pa.table( + { + name: [row[name] for row in rows] + for name in arrow_schema.names + }, + schema=arrow_schema, + )) commit.commit(writer.prepare_commit()) finally: writer.close() @@ -106,7 +113,12 @@ def _write_legacy_date_row(table, arrow_schema, row): def _read_rows(table): builder = table.new_read_builder() splits = builder.new_scan().plan().splits() - return builder.new_read().to_arrow(splits).to_pylist() + result = builder.new_read().to_arrow(splits) + columns = result.to_pydict() + return [ + {name: columns[name][index] for name in result.column_names} + for index in range(result.num_rows) + ] def test_legacy_date_partition_write_uses_java_epoch_day(tmp_path): @@ -121,6 +133,10 @@ def test_legacy_date_partition_write_uses_java_epoch_day(tmp_path): assert _read_rows(table) == [{"id": 1, "day": date(1970, 1, 2)}] +@pytest.mark.python_plan +@pytest.mark.python_read +@pytest.mark.python_write +@pytest.mark.python_commit def test_legacy_python_date_partition_directory_remains_readable(tmp_path): table, arrow_schema = _create_table(tmp_path) @@ -204,6 +220,10 @@ def test_legacy_composite_date_write_reads_back_from_canonical_path(tmp_path): assert (Path(table.table_path) / "day=1" / "region=a%2Fb" / "bucket-0").is_dir() +@pytest.mark.python_plan +@pytest.mark.python_read +@pytest.mark.python_write +@pytest.mark.python_commit def test_old_and_new_composite_partition_files_coexist_after_append(tmp_path): table, arrow_schema = _create_table(tmp_path, composite_partition=True) row = {"day": date(1970, 1, 2), "region": "a/b"} diff --git a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py index 82b2e180369a..afa6e26ab7db 100644 --- a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py +++ b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py @@ -150,11 +150,11 @@ def test_read_java_composite_date_partition(self): read_builder = table.new_read_builder() result = read_builder.new_read().to_arrow( read_builder.new_scan().plan().splits()) - self.assertEqual(result.to_pylist(), [{ - 'id': 1, - 'day': datetime.date(1970, 1, 2), - 'region': 'a/b', - }]) + self.assertEqual(result.to_pydict(), { + 'id': [1], + 'day': [datetime.date(1970, 1, 2)], + 'region': ['a/b'], + }) def test_py_write_composite_date_partition(self): arrow_schema = pa.schema([ @@ -169,11 +169,11 @@ def test_py_write_composite_date_partition(self): builder = table.new_batch_write_builder() writer, commit = builder.new_write(), builder.new_commit() try: - writer.write_arrow(pa.Table.from_pylist([{ - 'id': 2, - 'day': datetime.date(1970, 1, 2), - 'region': 'a/b', - }], schema=arrow_schema)) + writer.write_arrow(pa.table({ + 'id': [2], + 'day': [datetime.date(1970, 1, 2)], + 'region': ['a/b'], + }, schema=arrow_schema)) commit.commit(writer.prepare_commit()) finally: writer.close() @@ -183,7 +183,7 @@ def test_py_write_composite_date_partition(self): result = read_builder.new_read().to_arrow( read_builder.new_scan().plan().splits()) self.assertEqual( - {row['id'] for row in result.to_pylist()}, {1, 2}) + set(result.to_pydict()['id']), {1, 2}) @parameterized.expand([ (type_name, order, grouping) From 6c5cd582b83a5d8cbb1d8fc7213a7a6ed32e5a9d Mon Sep 17 00:00:00 2001 From: zhengguangzhuo <100548634+zhengguangzhuo@users.noreply.github.com> Date: Fri, 9 Oct 2026 11:19:55 +0800 Subject: [PATCH 4/5] [python] Batch partition compatibility lookups --- .../scanner/data_evolution_split_generator.py | 14 ++++- .../pypaimon/read/scanner/split_generator.py | 18 +++++- .../tests/date_partition_path_test.py | 58 ++++++++++++++++--- 3 files changed, 77 insertions(+), 13 deletions(-) diff --git a/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py b/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py index be995c5acdcd..1efa3fc93105 100644 --- a/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py +++ b/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py @@ -128,15 +128,23 @@ def _build_split_from_pack_for_data_evolution( raw_convertible is True only when each range (pack) contains exactly one file. """ splits = [] + data_files = [ + data_file + for file_group in flatten_packed_files + for data_file in file_group + ] + self._set_data_file_paths( + data_files, + file_entries[0].partition, + file_entries[0].bucket, + ) + for i, file_group in enumerate(flatten_packed_files): # In Java: rawConvertible = f.stream().allMatch(file -> file.size() == 1) # This means raw_convertible is True only when each range contains exactly one file pack = packed_files[i] if i < len(packed_files) else [] raw_convertible = all(len(sub_pack) == 1 for sub_pack in pack) - self._set_data_file_paths( - file_group, file_entries[0].partition, file_entries[0].bucket) - if file_group: # Get deletion files for this split data_deletion_files = None diff --git a/paimon-python/pypaimon/read/scanner/split_generator.py b/paimon-python/pypaimon/read/scanner/split_generator.py index 83f8f8fdf783..d0c4ce14421d 100644 --- a/paimon-python/pypaimon/read/scanner/split_generator.py +++ b/paimon-python/pypaimon/read/scanner/split_generator.py @@ -97,6 +97,21 @@ def _build_split_from_pack( splits = [] if not packed_files or not file_entries: return splits + + # All packed groups are derived from the same partition and bucket. + # Resolve every file together so path compatibility needs at most one + # canonical lookup and one historical fallback per partition/bucket. + data_files = [ + data_file + for file_group in packed_files + for data_file in file_group + ] + self._set_data_file_paths( + data_files, + file_entries[0].partition, + file_entries[0].bucket, + ) + for file_group in packed_files: if use_optimized_path: raw_convertible = True @@ -105,9 +120,6 @@ def _build_split_from_pack( else: raw_convertible = True - self._set_data_file_paths( - file_group, file_entries[0].partition, file_entries[0].bucket) - if file_group: # Get deletion files for this split data_deletion_files = None diff --git a/paimon-python/pypaimon/tests/date_partition_path_test.py b/paimon-python/pypaimon/tests/date_partition_path_test.py index d09cec49d54c..7175640ea18a 100644 --- a/paimon-python/pypaimon/tests/date_partition_path_test.py +++ b/paimon-python/pypaimon/tests/date_partition_path_test.py @@ -25,6 +25,7 @@ import pytest from pypaimon.common.identifier import Identifier +from pypaimon.common.options.core_options import CoreOptions from pypaimon.filesystem.local_file_io import LocalFileIO from pypaimon.schema.data_types import AtomicType from pypaimon.schema.schema import Schema @@ -34,7 +35,9 @@ from pypaimon.write.writer.data_writer import DataWriter -def _create_table(tmp_path, legacy_partition_name=True, composite_partition=False): +def _create_table( + tmp_path, legacy_partition_name=True, composite_partition=False, table_options=None +): table_path = str(tmp_path / "table") file_io = LocalFileIO() fields = [("id", pa.int64()), ("day", pa.date32())] @@ -43,15 +46,17 @@ def _create_table(tmp_path, legacy_partition_name=True, composite_partition=Fals fields.append(("region", pa.string())) partition_keys.append("region") arrow_schema = pa.schema(fields) + options = { + "partition.legacy-name": str(legacy_partition_name).lower(), + "scan.native-plan.enabled": "false", + "write.native.enabled": "false", + "commit.native.enabled": "false", + } + options.update(table_options or {}) schema = Schema.from_pyarrow_schema( arrow_schema, partition_keys=partition_keys, - options={ - "partition.legacy-name": str(legacy_partition_name).lower(), - "scan.native-plan.enabled": "false", - "write.native.enabled": "false", - "commit.native.enabled": "false", - }, + options=options, ) table_schema = SchemaManager(file_io, table_path).create_table(schema) table = FileStoreTable( @@ -250,6 +255,45 @@ def test_old_and_new_composite_partition_files_coexist_after_append(tmp_path): assert [item["id"] for item in _read_rows(table)] == [1] +@pytest.mark.parametrize("historical_layout", [False, True]) +def test_composite_date_path_lookups_are_batched_across_split_packs( + tmp_path, monkeypatch, historical_layout +): + table, arrow_schema = _create_table( + tmp_path, + composite_partition=True, + table_options={ + CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(): "1b", + CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST.key(): "1b", + }, + ) + row = {"day": date(1970, 1, 2), "region": "a/b"} + + for row_id in (1, 2): + data = {"id": row_id, **row} + if historical_layout: + _write_legacy_composite_row(table, arrow_schema, data) + else: + _write_rows(table, arrow_schema, [data]) + + lookup_batches = [] + file_io = table.file_io + exists_batch = file_io.exists_batch + + def track_exists_batch(paths): + lookup_batches.append(list(paths)) + return exists_batch(paths) + + monkeypatch.setattr(file_io, "exists_batch", track_exists_batch) + splits = table.new_read_builder().new_scan().plan().splits() + + assert len(splits) == 2 + assert sum(len(split.files) for split in splits) == 2 + expected_batch_sizes = [2, 2] if historical_layout else [2] + assert [len(paths) for paths in lookup_batches] == expected_batch_sizes + assert {item["id"] for item in _read_rows(table)} == {1, 2} + + def test_non_legacy_composite_date_write_uses_canonical_escaping(tmp_path): table, arrow_schema = _create_table( tmp_path, legacy_partition_name=False, composite_partition=True) From 0e1ac11a2ce1517af2ed8d85a2e029dd2f5cb414 Mon Sep 17 00:00:00 2001 From: zhengguangzhuo <100548634+zhengguangzhuo@users.noreply.github.com> Date: Fri, 9 Oct 2026 12:08:13 +0800 Subject: [PATCH 5/5] [python] Keep path batching test on Python backends --- paimon-python/pypaimon/tests/date_partition_path_test.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/paimon-python/pypaimon/tests/date_partition_path_test.py b/paimon-python/pypaimon/tests/date_partition_path_test.py index 7175640ea18a..0bf91ff7e5ac 100644 --- a/paimon-python/pypaimon/tests/date_partition_path_test.py +++ b/paimon-python/pypaimon/tests/date_partition_path_test.py @@ -255,6 +255,10 @@ def test_old_and_new_composite_partition_files_coexist_after_append(tmp_path): assert [item["id"] for item in _read_rows(table)] == [1] +@pytest.mark.python_plan +@pytest.mark.python_read +@pytest.mark.python_write +@pytest.mark.python_commit @pytest.mark.parametrize("historical_layout", [False, True]) def test_composite_date_path_lookups_are_batched_across_split_packs( tmp_path, monkeypatch, historical_layout