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 8e5b6c21d4 [Python] Align Native write validation and file metadata
with Java (#10660)
8e5b6c21d4 is described below
commit 8e5b6c21d4d4e13a41c069ee15c513006718d8e7
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Oct 19 22:09:21 2026 +0800
[Python] Align Native write validation and file metadata with Java (#10660)
---
.../paimon/utils/InternalRowPartitionComputer.java | 17 +-
.../utils/InternalRowPartitionComputerTest.java | 73 +++++
paimon-python/README.md | 4 +
.../pypaimon/manifest/manifest_file_manager.py | 9 +-
.../pypaimon/manifest/schema/data_file_meta.py | 17 +-
paimon-python/pypaimon/snapshot/snapshot_commit.py | 3 +-
.../pypaimon/table/source/native_vector_search.py | 7 -
.../pypaimon/tests/data_file_creation_time_test.py | 175 ++++++++++++
.../tests/manifest/creation_time_precision_test.py | 74 ++++++
paimon-python/pypaimon/tests/native_commit_test.py | 10 +
.../tests/native_filtered_write_validation_test.py | 94 +++++++
.../pypaimon/tests/native_partition_types_test.py | 52 +++-
.../pypaimon/tests/native_vector_context_test.py | 78 ++++++
.../pypaimon/tests/native_vector_search_test.py | 35 ++-
.../tests/pure_writer_creation_time_test.py | 149 +++++++++++
.../pypaimon/tests/reader_primary_key_test.py | 15 +-
.../tests/write/row_kind_nullability_test.py | 295 +++++++++++++++++++++
.../pypaimon/utils/file_store_path_factory.py | 6 +-
paimon-python/pypaimon/write/file_store_commit.py | 15 +-
paimon-python/pypaimon/write/native_commit.py | 11 +-
paimon-python/pypaimon/write/native_merge_into.py | 7 -
paimon-python/pypaimon/write/native_update.py | 7 +-
paimon-python/pypaimon/write/native_write.py | 30 +--
paimon-python/pypaimon/write/row_kind.py | 27 +-
paimon-python/pypaimon/write/table_write.py | 52 +++-
paimon-python/pypaimon/write/writer/blob_writer.py | 2 +-
.../pypaimon/write/writer/composite_data_writer.py | 2 +-
paimon-python/pypaimon/write/writer/data_writer.py | 10 +-
.../pypaimon/write/writer/vector_writer.py | 2 +-
29 files changed, 1144 insertions(+), 134 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/utils/InternalRowPartitionComputer.java
b/paimon-common/src/main/java/org/apache/paimon/utils/InternalRowPartitionComputer.java
index 0145ee3f78..d82ef6b532 100644
---
a/paimon-common/src/main/java/org/apache/paimon/utils/InternalRowPartitionComputer.java
+++
b/paimon-common/src/main/java/org/apache/paimon/utils/InternalRowPartitionComputer.java
@@ -65,7 +65,12 @@ public class InternalRowPartitionComputer {
int i = columnList.indexOf(partitionColumns[j]);
DataType type = rowType.getTypeAt(i);
partitionFieldGetters[j] = createNullCheckingFieldGetter(type, i);
- partitionCastExecutors[j] = CastExecutors.resolve(type,
VarCharType.STRING_TYPE);
+ // byte[].toString contains an allocation identity, which changes
every time a
+ // BinaryRow field is read. Always use the UTF-8 cast for binary
partition names.
+ partitionCastExecutors[j] =
+ legacyPartitionName && !TypeCheckUtils.isBinaryString(type)
+ ? value -> value
+ : CastExecutors.resolve(type,
VarCharType.STRING_TYPE);
}
}
@@ -76,13 +81,9 @@ public class InternalRowPartitionComputer {
Object field = partitionFieldGetters[i].getFieldOrNull(in);
String partitionValue = null;
if (field != null) {
- if (legacyPartitionName) {
- partitionValue = field.toString();
- } else {
- Object casted = partitionCastExecutors[i].cast(field);
- if (casted != null) {
- partitionValue = casted.toString();
- }
+ Object casted = partitionCastExecutors[i].cast(field);
+ if (casted != null) {
+ partitionValue = casted.toString();
}
}
if (StringUtils.isNullOrWhitespaceOnly(partitionValue)) {
diff --git
a/paimon-common/src/test/java/org/apache/paimon/utils/InternalRowPartitionComputerTest.java
b/paimon-common/src/test/java/org/apache/paimon/utils/InternalRowPartitionComputerTest.java
index 6f92d72ed9..42019a4ff8 100644
---
a/paimon-common/src/test/java/org/apache/paimon/utils/InternalRowPartitionComputerTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/utils/InternalRowPartitionComputerTest.java
@@ -27,6 +27,8 @@ import org.apache.paimon.types.RowType;
import org.junit.jupiter.api.Test;
+import java.nio.charset.StandardCharsets;
+
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.entry;
@@ -108,4 +110,75 @@ public class InternalRowPartitionComputerTest {
assertThat(computer.generatePartValues(binaryRow))
.containsExactly(entry("dt", "20240731"), entry("id", "1"));
}
+
+ @Test
+ public void testBinaryPartitionNamesUseStableUtf8Values() {
+ DataType[] binaryTypes = {DataTypes.BINARY(16),
DataTypes.VARBINARY(16), DataTypes.BYTES()};
+ byte[][] values = {
+ null,
+ new byte[0],
+ " \t\u001c".getBytes(StandardCharsets.UTF_8),
+ "a/b=c".getBytes(StandardCharsets.UTF_8),
+ "汉字".getBytes(StandardCharsets.UTF_8),
+ "\u00a0".getBytes(StandardCharsets.UTF_8),
+ new byte[] {(byte) 0xed, (byte) 0xa0, (byte) 0x80}
+ };
+ String[] expected = {
+ "__DEFAULT_PARTITION__",
+ "__DEFAULT_PARTITION__",
+ "__DEFAULT_PARTITION__",
+ "a/b=c",
+ "汉字",
+ "\u00a0",
+ "\ufffd"
+ };
+ for (boolean legacy : new boolean[] {true, false}) {
+ for (DataType type : binaryTypes) {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), type}, new
String[] {"id", "p"});
+ InternalRowPartitionComputer computer =
+ new InternalRowPartitionComputer(
+ "__DEFAULT_PARTITION__", rowType, new String[]
{"p"}, legacy);
+ for (int valueIndex = 0; valueIndex < values.length;
valueIndex++) {
+ BinaryRow row = new BinaryRow(2);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.writeInt(0, 123);
+ byte[] value = values[valueIndex];
+ if (value == null) {
+ writer.setNullAt(1);
+ } else {
+ writer.writeBinary(1, value, 0, value.length);
+ }
+ writer.complete();
+ // BinaryRow.getBinary creates a fresh byte[] on every
call. Partition
+ // names must depend on its contents, including after row
serialization.
+ assertThat(computer.generatePartValues(row))
+ .containsExactly(entry("p", expected[valueIndex]));
+ assertThat(computer.generatePartValues(row))
+ .containsExactly(entry("p", expected[valueIndex]));
+ assertThat(computer.generatePartValues(row.copy()))
+ .containsExactly(entry("p", expected[valueIndex]));
+ }
+ }
+ }
+ }
+
+ @Test
+ public void testBinaryPartitionKeepsOtherLegacyValues() {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.DATE(), DataTypes.BYTES()},
+ new String[] {"dt", "p"});
+ InternalRowPartitionComputer computer =
+ new InternalRowPartitionComputer(
+ "__DEFAULT_PARTITION__", rowType, new String[] {"p",
"dt"}, true);
+ BinaryRow row = new BinaryRow(2);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.writeInt(0, 1);
+ writer.writeBinary(1, new byte[] {65}, 0, 1);
+ writer.complete();
+ assertThat(computer.generatePartValues(row))
+ .containsExactly(entry("p", "A"), entry("dt", "1"));
+ }
}
diff --git a/paimon-python/README.md b/paimon-python/README.md
index 55f53762d3..5f0e4c81d5 100644
--- a/paimon-python/README.md
+++ b/paimon-python/README.md
@@ -551,6 +551,10 @@ Writer restore, overwrite and index construction use this
entrypoint and delegat
planning to Rust core when Native planning is enabled.
Native planning remains optional and is disabled by default.
+Binary partition keys use UTF-8 text and normal partition escaping in both
+`partition.legacy-name` modes, matching Java. This makes repeated writes to the
+same binary value use a stable directory.
+
# Coalesced BLOB reads
FileIO merges nearby BLOB ranges before reading. Set
diff --git a/paimon-python/pypaimon/manifest/manifest_file_manager.py
b/paimon-python/pypaimon/manifest/manifest_file_manager.py
index 27ae7b32d8..d52b9f50d1 100644
--- a/paimon-python/pypaimon/manifest/manifest_file_manager.py
+++ b/paimon-python/pypaimon/manifest/manifest_file_manager.py
@@ -29,7 +29,7 @@ try:
except ImportError: # pragma: no cover - supported fastavro versions provide
these
_read_bytes = _read_long = _read_record = _skip_record = None
-from datetime import datetime
+from datetime import datetime, timezone
from pypaimon.filesystem.caching_file_io import (
CachingInputStream, _open_input_stream_with_known_size)
@@ -265,15 +265,16 @@ class ManifestFileManager:
max_values=BinaryRow(value_dict['_MAX_VALUES'], fields),
null_counts=value_dict['_NULL_COUNTS'],
)
- # fastavro returns UTC-aware datetime for timestamp-millis, we
need to convert properly
+ # Avro exposes the stored local TIMESTAMP fields as a UTC datetime.
+ # Keep integer millisecond precision, without applying the system
zone.
from pypaimon.data.timestamp import Timestamp
creation_time_value = file_dict['_CREATION_TIME']
creation_time_ts = None
if creation_time_value is not None:
if isinstance(creation_time_value, datetime):
if creation_time_value.tzinfo:
- epoch_millis = int(creation_time_value.timestamp() *
1000)
- creation_time_ts =
Timestamp.from_epoch_millis(epoch_millis)
+ creation_time_ts = Timestamp.from_local_date_time(
+
creation_time_value.astimezone(timezone.utc).replace(tzinfo=None))
else:
creation_time_ts =
Timestamp.from_local_date_time(creation_time_value)
elif isinstance(creation_time_value, (int, float)):
diff --git a/paimon-python/pypaimon/manifest/schema/data_file_meta.py
b/paimon-python/pypaimon/manifest/schema/data_file_meta.py
index 8e454bc84a..3a16086668 100644
--- a/paimon-python/pypaimon/manifest/schema/data_file_meta.py
+++ b/paimon-python/pypaimon/manifest/schema/data_file_meta.py
@@ -18,7 +18,6 @@
from dataclasses import dataclass
from datetime import datetime
from typing import List, Optional
-import time
from pypaimon.utils.range import Range
from pypaimon.data.timestamp import Timestamp
@@ -75,12 +74,13 @@ class DataFileMeta:
def creation_time_epoch_millis(self) -> Optional[int]:
if self.creation_time is None:
return None
- local_dt = self.creation_time.to_local_date_time()
- local_time_struct = local_dt.timetuple()
- local_timestamp = time.mktime(local_time_struct)
- utc_timestamp = time.mktime(time.gmtime(local_timestamp))
- tz_offset_seconds = int(local_timestamp - utc_timestamp)
- return int((local_timestamp - tz_offset_seconds) * 1000)
+ from pypaimon.casting.temporal_utils import _local_to_utc_seconds
+
+ # Java stores Timestamp.now() as local wall-clock fields, then resolves
+ # LocalDateTime.atZone(systemDefault()) when an epoch instant is
needed.
+ # Preserve the millisecond remainder, including for pre-epoch values.
+ seconds, millis = divmod(self.creation_time.get_millisecond(), 1000)
+ return _local_to_utc_seconds(seconds) * 1000 + millis
def creation_time_as_datetime(self) -> Optional[datetime]:
if self.creation_time is None:
@@ -114,7 +114,8 @@ class DataFileMeta:
write_cols_sequences: Optional[List[int]] = None,
) -> 'DataFileMeta':
if creation_time is None:
- creation_time = Timestamp.now()
+ # Java forAppend/forLevel0 factories persist TIMESTAMP(3).
+ creation_time = Timestamp.now().to_millis_timestamp()
return cls(
file_name=file_name,
diff --git a/paimon-python/pypaimon/snapshot/snapshot_commit.py
b/paimon-python/pypaimon/snapshot/snapshot_commit.py
index dd9c90aead..7c11b2608c 100644
--- a/paimon-python/pypaimon/snapshot/snapshot_commit.py
+++ b/paimon-python/pypaimon/snapshot/snapshot_commit.py
@@ -63,7 +63,8 @@ class PartitionStatistics:
record_count=record_count,
file_count=file_count,
file_size_in_bytes=file_size_in_bytes,
- last_file_creation_time=last_file_creation_time or int(time.time()
* 1000),
+ last_file_creation_time=(int(time.time() * 1000)
+ if last_file_creation_time is None else
last_file_creation_time),
total_buckets=total_buckets
)
diff --git a/paimon-python/pypaimon/table/source/native_vector_search.py
b/paimon-python/pypaimon/table/source/native_vector_search.py
index f700d4a9e5..01db9cbd2a 100644
--- a/paimon-python/pypaimon/table/source/native_vector_search.py
+++ b/paimon-python/pypaimon/table/source/native_vector_search.py
@@ -37,13 +37,6 @@ def _native_vector_builder(builder, batch=False):
loader = getattr(getattr(table, 'catalog_environment', None),
'catalog_loader', None)
if loader is None or _catalog_metastore(loader) != 'rest':
return None
- context_fn = getattr(loader, 'context', None)
- if not callable(context_fn):
- return None
- context = context_fn()
- if any(getattr(context, attr, None) is not None for attr in (
- 'hadoop_conf', 'prefer_io_loader', 'fallback_io_loader')):
- return None
# Python's batch reader currently supports only global row-ID results.
if batch and builder._is_primary_key_vector_search():
return None
diff --git a/paimon-python/pypaimon/tests/data_file_creation_time_test.py
b/paimon-python/pypaimon/tests/data_file_creation_time_test.py
new file mode 100644
index 0000000000..3c921a13bc
--- /dev/null
+++ b/paimon-python/pypaimon/tests/data_file_creation_time_test.py
@@ -0,0 +1,175 @@
+# 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.
+
+"""Data-file creation times follow Java local Timestamp/epoch semantics."""
+
+import json
+import os
+import subprocess
+import sys
+import time
+from datetime import datetime
+
+import pytest
+
+
+pytestmark = [pytest.mark.python_write, pytest.mark.python_commit]
+_CASES = {
+ 'UTC': [('2024-01-02T12:34:56.789', '2024-01-02T12:34:56.789'),
+ ('1970-01-01T00:00:00.000', '1970-01-01T00:00:00.000'),
+ ('1969-12-31T23:59:59.999', '1969-12-31T23:59:59.999')],
+ 'Asia/Shanghai': [('2024-01-02T12:34:56.789', '2024-01-02T04:34:56.789'),
+ ('1970-01-01T07:59:59.999', '1969-12-31T23:59:59.999')],
+ 'America/New_York': [('2024-01-02T12:34:56.789',
'2024-01-02T17:34:56.789'),
+ ('2024-03-10T02:30:00.123',
'2024-03-10T07:30:00.123'),
+ ('2024-11-03T01:30:00.456',
'2024-11-03T05:30:00.456')],
+}
+_SCRIPT = r'''import json, sys, time
+from datetime import datetime
+from pypaimon.data.timestamp import Timestamp
+from pypaimon.manifest.schema.data_file_meta import DataFileMeta
+if hasattr(time, 'tzset'):
+ time.tzset()
+result = []
+for wall in json.loads(sys.argv[1]):
+ metadata = DataFileMeta('data.parquet', 1, 1, None, None, None, None, 0,
0, 0, 0, [])
+ metadata.creation_time =
Timestamp.from_local_date_time(datetime.strptime(wall, '%Y-%m-%dT%H:%M:%S.%f'))
+ result.append(metadata.creation_time_epoch_millis())
+metadata.creation_time = None
+assert metadata.creation_time_epoch_millis() is None
+print(json.dumps(result))
+'''
+
+
[email protected](not hasattr(time, 'tzset'), reason='process TZ cannot be
changed on this platform')
[email protected]('zone', sorted(_CASES))
+def
test_creation_time_epoch_millis_matches_java_zone_rules_and_precision(zone):
+ # Run separate processes so TZ changes cannot race other tests or cached
+ # native timezone detection. Goldens use LocalDateTime.atZone semantics:
+ # earlier instant in an overlap, forward through a spring gap.
+ environment = dict(os.environ, TZ=zone)
+ output = subprocess.check_output([sys.executable, '-c', _SCRIPT,
+ json.dumps([wall for wall, _ in
_CASES[zone]])], env=environment)
+ assert json.loads(output) == _expected_epoch_millis(zone)
+
+
+def test_create_preserves_an_explicit_creation_timestamp():
+ from pypaimon.data.timestamp import Timestamp
+ from pypaimon.manifest.schema.data_file_meta import DataFileMeta
+
+ # Java's generic factory accepts supplied timestamps unchanged, while its
+ # normal writer factories create millisecond timestamps.
+ explicit = Timestamp(1001, 987654)
+ metadata = DataFileMeta.create('data.parquet', 1, 1, None, None, None,
None, 0, 0, 0, 0, [],
+ creation_time=explicit)
+ assert metadata.creation_time == explicit
+
+
+_REST_SCRIPT = r'''import json, sys, time
+from contextlib import ExitStack
+from datetime import datetime
+from unittest.mock import Mock, patch
+import pyarrow as pa
+from pypaimon import CatalogFactory, Schema
+from pypaimon.api.api_response import ConfigResponse
+from pypaimon.api.auth import BearTokenAuthProvider
+from pypaimon.data.timestamp import Timestamp
+from pypaimon.tests.rest.rest_server import RESTCatalogServer
+if hasattr(time, 'tzset'):
+ time.tzset()
+server = RESTCatalogServer(sys.argv[1], BearTokenAuthProvider('test-token'),
+ ConfigResponse(defaults={'prefix': 'native-test'}),
'warehouse')
+server.start()
+try:
+ catalog = CatalogFactory.create({'metastore': 'rest', 'uri':
server.get_url(), 'warehouse': 'warehouse',
+ 'token.provider': 'bear', 'token':
'test-token', 'data-token.enabled': 'false'})
+ catalog.create_database('default', True)
+ schema = pa.schema([('id', pa.int32()), ('pt', pa.string())])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
partition_keys=['pt'], options={
+ 'write.native.enabled': 'false', 'read.native.enabled': 'false',
'scan.native-plan.enabled': 'false',
+ 'commit.native.enabled': sys.argv[2]}), False)
+ table = catalog.get_table('default.t')
+ captured = []
+ original = server._table_commit_handle
+ def publish(method, data, *args, **kwargs):
+ captured.append(json.loads(data)['statistics'])
+ return original(method, data, *args, **kwargs)
+ for wall in json.loads(sys.argv[3]):
+ builder = table.new_batch_write_builder()
+ writer, earlier_writer, commit = builder.new_write(),
builder.new_write(), builder.new_commit()
+ callback = Mock()
+ commit.add_commit_callback(callback)
+ timestamp = Timestamp.from_local_date_time(datetime.strptime(wall,
'%Y-%m-%dT%H:%M:%S.%f'))
+ try:
+ writer.write_arrow(pa.Table.from_pylist([{'id': 1, 'pt': 'p'}],
schema=schema))
+ messages = writer.prepare_commit()
+ for message in messages:
+ for file in message.new_files:
+ file.creation_time = timestamp
+ earlier_writer.write_arrow(pa.Table.from_pylist([{'id': 2, 'pt':
'p'}], schema=schema))
+ earlier_messages = earlier_writer.prepare_commit()
+ earlier = Timestamp(timestamp.get_millisecond() - 1)
+ for message in earlier_messages:
+ for file in message.new_files:
+ file.creation_time = earlier
+ messages.extend(earlier_messages)
+ if sys.argv[4] == 'true':
+ messages.reverse()
+ with ExitStack() as stack:
+ stack.enter_context(patch.object(server,
'_table_commit_handle', side_effect=publish))
+ if sys.argv[2] == 'true':
+ stack.enter_context(patch.object(commit.file_store_commit,
'_try_commit',
+
side_effect=AssertionError('native commit fell back')))
+ commit.commit(messages)
+ callback.call.assert_called_once()
+ entries = callback.call.call_args[0][0].commit_entries
+ assert sorted(entry.file.creation_time.get_millisecond() for entry
in entries) == [
+ earlier.get_millisecond(), timestamp.get_millisecond()]
+ assert captured[-1][0]['recordCount'] == 2
+ assert captured[-1][0]['fileCount'] == 2
+ finally:
+ writer.close()
+ earlier_writer.close()
+ commit.close()
+ print('CREATION_TIME_RESULT=' +
json.dumps([stats[0]['lastFileCreationTime'] for stats in captured]))
+finally:
+ server.shutdown()
+'''
+
+
+def _expected_epoch_millis(zone):
+ values = []
+ for _, instant in _CASES[zone]:
+ delta = datetime.strptime(instant, '%Y-%m-%dT%H:%M:%S.%f') -
datetime(1970, 1, 1)
+ values.append((delta.days * 86400 + delta.seconds) * 1000 +
delta.microseconds // 1000)
+ return values
+
+
[email protected](not hasattr(time, 'tzset'), reason='process TZ cannot be
changed on this platform')
[email protected]('zone', sorted(_CASES))
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('reverse', [False, True], ids=['newer-first',
'older-first'])
+def test_rest_commit_publishes_creation_time_instants(tmp_path, zone, native,
reverse):
+ from pypaimon.read.native_plan import native_runtime_available
+
+ if native and not native_runtime_available():
+ pytest.skip('pypaimon-rust not installed')
+ output = subprocess.check_output(
+ [sys.executable, '-c', _REST_SCRIPT, str(tmp_path),
str(native).lower(),
+ json.dumps([wall for wall, _ in _CASES[zone]]),
str(reverse).lower()], env=dict(os.environ, TZ=zone))
+ result = next(line for line in output.decode().splitlines() if
line.startswith('CREATION_TIME_RESULT='))
+ assert json.loads(result.split('=', 1)[1]) == _expected_epoch_millis(zone)
diff --git
a/paimon-python/pypaimon/tests/manifest/creation_time_precision_test.py
b/paimon-python/pypaimon/tests/manifest/creation_time_precision_test.py
new file mode 100644
index 0000000000..1d0fa35583
--- /dev/null
+++ b/paimon-python/pypaimon/tests/manifest/creation_time_precision_test.py
@@ -0,0 +1,74 @@
+# 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.
+
+"""Avro creation timestamps retain Java's exact local millisecond fields."""
+
+from datetime import datetime, timedelta, timezone
+
+import fastavro
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.data.timestamp import Timestamp
+from pypaimon.manifest.manifest_file_manager import ManifestFileManager
+from pypaimon.manifest.schema.data_file_meta import DataFileMeta
+from pypaimon.manifest.schema.manifest_entry import ManifestEntry
+from pypaimon.manifest.schema.simple_stats import SimpleStats
+from pypaimon.table.row.generic_row import GenericRow
+
+
+pytestmark = [pytest.mark.python_write, pytest.mark.python_commit]
+
+
[email protected]('prefix_filter', [False, True], ids=['generic',
'filtered-block'])
+def test_creation_time_milliseconds_survive_real_avro_decoding(tmp_path,
prefix_filter):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ catalog.create_table('default.t',
Schema.from_pyarrow_schema(pa.schema([('id', pa.int32())])), False)
+ table = catalog.get_table('default.t')
+ manager = ManifestFileManager(table)
+ empty = GenericRow([], [])
+ stats = SimpleStats(empty, empty, [])
+ millis = [None, -1001, -1, 0, 1, 1001, 1005, 2017, 1_700_000_000_123]
+ entries = [ManifestEntry(
+ kind=0, partition=empty, bucket=0, total_buckets=1,
+ file=DataFileMeta(
+ file_name='data-%d.parquet' % index, file_size=1, row_count=1,
+ min_key=empty, max_key=empty, key_stats=stats, value_stats=stats,
+ min_sequence_number=0, max_sequence_number=0,
schema_id=table.table_schema.id,
+ level=0, extra_files=[], value_stats_cols=[],
+ creation_time=None if value is None else
Timestamp.from_epoch_millis(value),
+ )) for index, value in enumerate(millis)]
+ manager.write('manifest-creation-time', entries)
+
+ # Fastavro returns aware UTC datetimes even though Paimon's TIMESTAMP
+ # contains local wall fields. Verify the actual wire values independently.
+ with table.file_io.new_input_stream(manager.manifest_path +
'/manifest-creation-time') as stream:
+ wire_times = [record['_FILE']['_CREATION_TIME'] for record in
fastavro.reader(stream)]
+ epoch = datetime(1970, 1, 1, tzinfo=timezone.utc)
+ assert wire_times == [None if value is None else epoch +
timedelta(milliseconds=value) for value in millis]
+
+ options = {'early_entry_filter': lambda bucket, total: bucket == 0 and
total == 1} if prefix_filter else {}
+ decoded = manager.read('manifest-creation-time', drop_stats=False,
**options)
+ assert [None if entry.file.creation_time is None else
entry.file.creation_time.get_millisecond()
+ for entry in decoded] == millis
+
+ # Rewriting a manifest must not alter metadata handed over by a writer.
+ manager.write('manifest-creation-time-rewritten', decoded)
+ rewritten = manager.read('manifest-creation-time-rewritten', **options)
+ assert [entry.file.creation_time for entry in rewritten] ==
[entry.file.creation_time for entry in entries]
diff --git a/paimon-python/pypaimon/tests/native_commit_test.py
b/paimon-python/pypaimon/tests/native_commit_test.py
index ff369e8d65..d677751a2e 100644
--- a/paimon-python/pypaimon/tests/native_commit_test.py
+++ b/paimon-python/pypaimon/tests/native_commit_test.py
@@ -555,6 +555,16 @@ def
test_incompatible_publication_environment_is_not_reconstructed(
resolve.assert_not_called()
[email protected]('option', ['hadoop_conf', 'prefer_io_loader',
'fallback_io_loader'])
+def test_inactive_context_loader_fields_do_not_disable_native_commit(tmp_path,
native_rest_catalog, option):
+ from pypaimon.write.native_commit import _rest_catalog_supported
+ table = _table(tmp_path, catalog=native_rest_catalog)
+ unused = Mock(side_effect=AssertionError('Inactive IO loader was invoked'))
+ setattr(table.catalog_environment.catalog_loader.context(), option, unused)
+ assert _rest_catalog_supported(table)
+ unused.assert_not_called()
+
+
def test_missing_runtime_falls_back_without_reconstructing_table(
tmp_path, native_rest_catalog):
table = _table(tmp_path, catalog=native_rest_catalog)
diff --git
a/paimon-python/pypaimon/tests/native_filtered_write_validation_test.py
b/paimon-python/pypaimon/tests/native_filtered_write_validation_test.py
new file mode 100644
index 0000000000..ef4cee28a5
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_filtered_write_validation_test.py
@@ -0,0 +1,94 @@
+# 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.
+
+"""Ignored events do not reach nested value validation or physical writers."""
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.write.native_write import NativeTableWrite
+
+
+pytestmark = [pytest.mark.python_write, pytest.mark.python_commit]
+_NATIVE = [False, pytest.param(True, marks=pytest.mark.skipif(
+ not native_runtime_available(), reason='pypaimon-rust not installed'))]
+
+
+def _payload(mode, value):
+ if mode == 'array':
+ return pa.list_(pa.field('item', pa.int32(), nullable=False)), [value]
+ if mode == 'row':
+ return pa.struct([pa.field('value', pa.int32(), nullable=False)]),
{'value': value}
+ return pa.map_(pa.string(), pa.field('value', pa.int32(),
nullable=False)), [('key', value)]
+
+
+def _table(tmp_path, native, mode):
+ payload_type, _ = _payload(mode, None)
+ schema = pa.schema([('id', pa.int32()), ('op', pa.string()), ('payload',
payload_type)])
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
primary_keys=['id'], options={
+ 'bucket': '1', 'rowkind.field': 'op', 'ignore-delete': 'true',
+ 'write.native.enabled': str(native).lower(), 'commit.native.enabled':
'false',
+ 'read.native.enabled': 'false', 'scan.native-plan.enabled': 'false'}),
False)
+ return catalog.get_table('default.t'), schema
+
+
[email protected]('native', _NATIVE)
[email protected]('mode', ['array', 'row', 'map'])
[email protected]('ignored_kind', ['-D', '-U'])
[email protected]('bucket', [None, 0])
+def test_ignored_nested_nulls_leave_only_valid_rows(tmp_path, native, mode,
ignored_kind, bucket):
+ table, schema = _table(tmp_path, native, mode)
+ _, invalid = _payload(mode, None)
+ _, valid = _payload(mode, 42)
+ data = pa.Table.from_pylist([
+ {'id': 1, 'op': ignored_kind, 'payload': invalid},
+ {'id': 2, 'op': '+I', 'payload': valid},
+ ], schema=schema)
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ assert isinstance(writer, NativeTableWrite) == native
+ try:
+ writer.write_arrow(data, bucket)
+ messages = writer.prepare_commit()
+ assert sum(file.row_count for message in messages for file in
message.new_files) == 1
+ commit.commit(messages)
+ finally:
+ writer.close()
+ commit.close()
+ reader = table.new_read_builder()
+ assert
reader.new_read().to_arrow(reader.new_scan().plan().splits()).to_pylist() == [
+ {'id': 2, 'op': '+I', 'payload': valid}]
+
+
[email protected]('native', _NATIVE)
[email protected]('mode', ['array', 'row', 'map'])
+def test_surviving_nested_nulls_are_rejected(tmp_path, native, mode):
+ table, schema = _table(tmp_path, native, mode)
+ _, invalid = _payload(mode, None)
+ data = pa.Table.from_pylist([{'id': 1, 'op': '+I', 'payload': invalid}],
schema=schema)
+ writer = table.new_batch_write_builder().new_write()
+ assert isinstance(writer, NativeTableWrite) == native
+ try:
+ with pytest.raises((ValueError, RuntimeError, pa.ArrowInvalid)):
+ writer.write_arrow(data)
+ writer.prepare_commit()
+ finally:
+ writer.abort()
diff --git a/paimon-python/pypaimon/tests/native_partition_types_test.py
b/paimon-python/pypaimon/tests/native_partition_types_test.py
index a60295907a..cfe49286c8 100644
--- a/paimon-python/pypaimon/tests/native_partition_types_test.py
+++ b/paimon-python/pypaimon/tests/native_partition_types_test.py
@@ -85,10 +85,12 @@ def _write(table, schema, rows, stream=False):
@pytest.mark.native_plan
@pytest.mark.parametrize('partition_type,values,names', _PARTITIONS)
[email protected]('legacy', [False, True])
@pytest.mark.parametrize('mode', ['append', 'pk', 'evolution'])
@pytest.mark.parametrize('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)
+def test_native_partition_write_read_and_file_paths(tmp_path, partition_type,
values, names, legacy, mode, placement):
+ table, schema = _table(tmp_path, partition_type, mode, placement=placement,
+ extra_options={'partition.legacy-name':
str(legacy).lower()})
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]
@@ -149,8 +151,10 @@ def
test_native_non_finite_partitions_use_java_names_for_python_reads(tmp_path,
@pytest.mark.python_write
@pytest.mark.native_plan
@pytest.mark.parametrize('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)
[email protected]('legacy', [False, True])
+def test_python_binary_partitions_use_java_paths_and_native_reads(tmp_path,
partition_type, values, names, legacy):
+ table, schema = _table(tmp_path, partition_type, native=False,
+ extra_options={'partition.legacy-name':
str(legacy).lower()})
rows = [dict(id=i, p=value, value=i) for i, value in enumerate(values)]
messages = _write(table, schema, rows)
for message in messages:
@@ -202,8 +206,10 @@ def
test_native_cow_merge_uses_java_partition_path(tmp_path, action):
(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)
[email protected]('legacy', [False, True])
+def test_binary_partition_string_conversion_matches_java(tmp_path, value,
expected, legacy):
+ table, _ = _table(tmp_path, pa.binary(), native=False,
+ extra_options={'partition.legacy-name':
str(legacy).lower()})
assert table.path_factory().relative_bucket_path((value,), 0,
canonical_partition=True) == (
'p=' + expected + '/bucket-0')
@@ -261,9 +267,31 @@ def
test_native_commit_binary_partition_capability_precedes_execution(tmp_path,
patch('pypaimon.write.native_commit.native_runtime_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)
+ assert committer is not None
+ native.assert_called_once_with(table)
+
+
[email protected]_plan
[email protected]_read
[email protected]_write
[email protected]('legacy', [False, True])
[email protected]('mode', ['append', 'pk', 'evolution'])
+def test_python_binary_partition_write_commit_and_reopen(tmp_path, legacy,
mode):
+ table, schema = _table(tmp_path, pa.binary(), mode, native=False,
extra_options={
+ 'partition.legacy-name': str(legacy).lower(),
+ 'scan.native-plan.enabled': 'false', 'read.native.enabled': 'false',
+ 'commit.native.enabled': 'false'})
+ values = [b'a/b=c', '汉字'.encode('utf-8'), b'\xed\xa0\x80', b' \t\x1c', b'']
+ 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)
+ # Resolve independently from the typed partition bytes, as a
+ # reopened reader does; no Python bytes repr or object identity.
+ from pypaimon.utils.file_store_path_factory import
canonical_data_file_path
+ assert file.file_path == canonical_data_file_path(
+ table, message.partition, message.bucket, file.file_name)
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path / 'warehouse')})
+ reopened = catalog.get_table('db.t')
+ assert _read(reopened) == rows
diff --git a/paimon-python/pypaimon/tests/native_vector_context_test.py
b/paimon-python/pypaimon/tests/native_vector_context_test.py
new file mode 100644
index 0000000000..d42dc8a92b
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_vector_context_test.py
@@ -0,0 +1,78 @@
+# 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.
+
+"""Unused CatalogContext attributes must not select Python vector search."""
+
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
+
+import pytest
+
+from pypaimon.catalog.catalog_context import CatalogContext
+from pypaimon.catalog.rest.rest_catalog_loader import RESTCatalogLoader
+from pypaimon.common.predicate import Predicate
+from pypaimon.table.source import native_vector_search
+
+
[email protected]('option', ['hadoop_conf', 'prefer_io_loader',
'fallback_io_loader'])
[email protected]('batch', [False, True], ids=['single', 'batch'])
+def test_unused_context_attributes_preserve_native_vector_execution(option,
batch):
+ context = CatalogContext.create_from_options({'metastore': 'rest', 'uri':
'http://unused'})
+ unused = Mock(side_effect=AssertionError('An unused IO loader was
invoked'))
+ setattr(context, option, {'unused': 'value'} if option == 'hadoop_conf'
else unused)
+ loader = RESTCatalogLoader(context)
+ table = SimpleNamespace(
+ options=SimpleNamespace(native_read_enabled=lambda: True,
query_auth_enabled=False),
+ catalog_environment=SimpleNamespace(catalog_loader=loader))
+ builder = SimpleNamespace(
+ _table=table, _vector_column=SimpleNamespace(name='embedding'),
_limit=2,
+ _options={'ivf-flat.metric': 'l2'}, _filter=Predicate('equal', 0,
'id', [7]),
+ _partition_filter=Predicate('equal', 1, 'pt', [1]),
_is_primary_key_vector_search=lambda: False)
+ core = Mock()
+ core.with_limit.return_value = core
+ core.with_options.return_value = core
+ native_table = Mock()
+ native_table.new_vector_search_builder.return_value = core
+ native_table.new_batch_vector_search_builder.return_value = core
+ expected = [{7: -0.25, 9: -1.5}, {11: -2.0}]
+ core.execute_local.return_value.row_ids.return_value = expected[0]
+ core.execute_batch_local.return_value =
[Mock(row_ids=Mock(return_value=scores)) for scores in expected]
+ queries = [[1.0, 0.0], [0.0, 1.0]] if batch else [1.0, 0.0]
+ with patch.object(loader, 'load', side_effect=AssertionError('Python
catalog was reconstructed')), \
+ patch.object(native_vector_search, 'native_runtime_available',
return_value=True), \
+ patch.object(native_vector_search, '_native_table',
return_value=native_table) as resolve:
+ result = native_vector_search.try_native_vector_search(builder,
queries, batch=batch)
+ assert result is not None
+ resolve.assert_called_once_with(table)
+ core.with_vector_column.assert_called_once_with('embedding')
+ core.with_limit.assert_called_once_with(2)
+ core.with_options.assert_called_once_with({'ivf-flat.metric': 'l2'})
+ core.with_filter.assert_called_once_with({'method': 'equal', 'field':
'id', 'literals': [7]})
+ core.with_partition_filter.assert_called_once_with({'method': 'equal',
'field': 'pt', 'literals': [1]})
+ unused.assert_not_called()
+ if batch:
+ core.with_query_vectors.assert_called_once_with(queries)
+ core.execute_batch_local.assert_called_once_with()
+ core.execute_local.assert_not_called()
+ results = result
+ else:
+ core.with_query_vector.assert_called_once_with(queries)
+ core.execute_local.assert_called_once_with()
+ core.execute_batch_local.assert_not_called()
+ results = [result]
+ expected = expected[:1]
+ assert [{row_id: item.score_getter()(row_id) for row_id in item.results()}
for item in results] == expected
diff --git a/paimon-python/pypaimon/tests/native_vector_search_test.py
b/paimon-python/pypaimon/tests/native_vector_search_test.py
index d493962e8b..99cf6639f4 100644
--- a/paimon-python/pypaimon/tests/native_vector_search_test.py
+++ b/paimon-python/pypaimon/tests/native_vector_search_test.py
@@ -19,7 +19,7 @@
import shutil
from pathlib import Path
-from unittest.mock import patch
+from unittest.mock import Mock, patch
import pyarrow as pa
import pytest
@@ -308,18 +308,37 @@ def test_mixed_index_metrics_are_rejected(vector_table,
filtered, mode):
@pytest.mark.parametrize('option', ['hadoop_conf', 'prefer_io_loader',
'fallback_io_loader'])
-def test_custom_io_context_keeps_python_search(vector_table, option):
[email protected]('batch', [False, True], ids=['single', 'batch'])
+def test_unused_context_attributes_preserve_native_search(vector_table,
option, batch):
from copy import copy
- from pypaimon.table.source import native_vector_search
+ from pypaimon.table.source.vector_search_read import
DataEvolutionVectorRead, BatchVectorSearchReadImpl
table = vector_table.copy({'read.native.enabled': 'true'})
loader = table.catalog_environment.catalog_loader
context = copy(loader.context())
- setattr(context, option, object())
- expected = _scores(_single(table.copy({'read.native.enabled':
'false'})).execute_local())
+ unused = Mock(side_effect=AssertionError('An unused IO loader was
invoked'))
+ setattr(context, option, {'unused': 'value'} if option == 'hadoop_conf'
else unused)
+ classic = table.copy({'read.native.enabled': 'false',
'scan.native-plan.enabled': 'false'})
+ predicate = classic.new_read_builder().new_predicate_builder()
+ queries = [[1, 0], [0, 1]] if batch else [[1, 0]]
+
+ def configure(builder):
+ return (builder.with_options({'ivf-flat.metric': 'l2'})
+ .with_filter(predicate.greater_than('id',
1)).with_partition_filter(predicate.equal('pt', 1)))
+
+ expected = [_scores(configure(_single(classic, query, 2)).execute_local())
for query in queries]
with patch.object(loader, 'context', return_value=context), \
- patch.object(native_vector_search, '_native_table',
- side_effect=AssertionError('Unsupported Native
context')):
- _assert_scores(_scores(_single(table).execute_local()), expected)
+ patch.object(DataEvolutionVectorRead, 'read_plan',
side_effect=AssertionError('Python search ran')), \
+ patch.object(BatchVectorSearchReadImpl, 'read_batch_plan',
+ side_effect=AssertionError('Python batch search
ran')):
+ if batch:
+ actual =
configure(table.new_batch_vector_search_builder().with_vector_column('embedding')
+
.with_query_vectors(queries).with_limit(2)).execute_batch_local()
+ else:
+ actual = [configure(_single(table, queries[0], 2)).execute_local()]
+ assert len(actual) == len(expected)
+ for result, scores in zip(actual, expected):
+ _assert_scores(_scores(result), scores)
+ unused.assert_not_called()
@pytest.mark.parametrize('mode', ['fast', 'full', 'detail'])
diff --git a/paimon-python/pypaimon/tests/pure_writer_creation_time_test.py
b/paimon-python/pypaimon/tests/pure_writer_creation_time_test.py
new file mode 100644
index 0000000000..d4f85ac0ac
--- /dev/null
+++ b/paimon-python/pypaimon/tests/pure_writer_creation_time_test.py
@@ -0,0 +1,149 @@
+# 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.
+
+"""Prepared data and dedicated files use Java DataFileMeta's TIMESTAMP(3)."""
+
+from datetime import datetime, timedelta
+from unittest.mock import patch
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.data.timestamp import Timestamp
+from pypaimon.manifest.manifest_file_manager import ManifestFileManager
+from pypaimon.manifest.manifest_list_manager import ManifestListManager
+from pypaimon.write.native_write import NativeTableWrite
+
+
+pytestmark = [pytest.mark.python_write, pytest.mark.python_commit]
+
+
[email protected]('stream', [False, True], ids=['batch', 'stream'])
[email protected]('mode', ['append', 'pk-input', 'postpone', 'blob',
'vector'])
+def test_prepared_file_creation_times_have_millisecond_precision(tmp_path,
stream, mode):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('id', pa.int32()), ('value', pa.int32())])
+ rows = [dict(id=1, value=10), dict(id=2, value=20)]
+ options = {'file.format': 'parquet', 'write.native.enabled': 'false',
+ 'commit.native.enabled': 'false', 'read.native.enabled':
'false',
+ 'scan.native-plan.enabled': 'false'}
+ if mode in ('pk-input', 'postpone'):
+ options.update({'bucket': '-2' if mode == 'postpone' else '1',
+ 'changelog-producer': 'input'})
+ if mode in ('blob', 'vector'):
+ options.update({'data-evolution.enabled': 'true',
'row-tracking.enabled': 'true'})
+ if mode == 'blob':
+ schema = schema.append(pa.field('payload', pa.large_binary()))
+ for row in rows:
+ row['payload'] = ('blob-%d' % row['id']).encode()
+ else:
+ options['vector.file.format'] = 'parquet'
+ schema = schema.append(pa.field('embedding',
pa.list_(pa.float32(), 2)))
+ for row in rows:
+ row['embedding'] = [float(row['id']), 2.0]
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ schema, primary_keys=['id'] if mode in ('pk-input', 'postpone') else
[], options=options), False)
+ table = catalog.get_table('default.t')
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ assert not isinstance(writer, NativeTableWrite)
+ now = Timestamp.from_local_date_time(datetime(2024, 6, 1, 12, 30, 40,
123456))
+ expected = now.to_millis_timestamp()
+ try:
+ with patch.object(Timestamp, 'now', return_value=now):
+ writer.write_arrow(pa.Table.from_pylist(rows, schema=schema))
+ messages = writer.prepare_commit(1) if stream else
writer.prepare_commit()
+ files = [file for message in messages for file in message.new_files +
message.changelog_files]
+ assert files
+ assert all(file.creation_time.get_nano_of_millisecond() == 0 for file
in files)
+ assert all(file.creation_time == expected for file in files)
+ assert now.get_nano_of_millisecond() == 456000
+ if mode == 'blob':
+ assert any(file.file_name.endswith('.blob') for file in files)
+ assert any(file.file_name.endswith('.parquet') for file in files)
+ if mode == 'vector':
+ assert any('.vector.parquet' in file.file_name for file in files)
+ if mode == 'pk-input':
+ assert any(message.changelog_files for message in messages)
+ if mode == 'postpone':
+ assert all(message.bucket == -2 and not message.changelog_files
for message in messages)
+ # Pending events are invisible to normal scans until assigned real
+ # buckets; verify their physical Parquet rows before publication.
+ with table.file_io.new_input_stream(files[0].file_path) as stream_:
+ physical =
pq.ParquetFile(stream_).read().select(schema.names).to_pylist()
+ assert physical == rows
+ commit.commit(messages, 1) if stream else commit.commit(messages)
+ finally:
+ writer.close()
+ commit.close()
+
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ entries =
ManifestFileManager(table).read_entries_parallel(ManifestListManager(table).read_all(snapshot))
+ assert entries and all(entry.file.creation_time == expected for entry in
entries)
+ read = table.new_read_builder()
+ actual =
read.new_read().to_arrow(read.new_scan().plan().splits()).sort_by('id').to_pylist()
+ assert actual == ([] if mode == 'postpone' else rows)
+
+
[email protected]('stream', [False, True], ids=['batch', 'stream'])
+def test_input_changelog_creation_time_is_taken_after_its_write(tmp_path,
stream):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('id', pa.int32()), ('value', pa.int32())])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ schema, primary_keys=['id'], options={
+ 'bucket': '1', 'changelog-producer': 'input',
+ 'write.native.enabled': 'false', 'commit.native.enabled': 'false',
+ 'read.native.enabled': 'false', 'scan.native-plan.enabled':
'false'}), False)
+ table = catalog.get_table('default.t')
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ clock = [datetime(2024, 6, 1, 12, 30, 40, 123456)]
+ completed = {}
+ write_parquet = table.file_io.write_parquet
+
+ def write_and_advance(path, *args, **kwargs):
+ result = write_parquet(path, *args, **kwargs)
+ # Advance a deterministic clock only after a real file was written.
+ # Each file's result must use its own completion time, including the
+ # changelog whose statistics were collected before the I/O.
+ clock[0] += timedelta(seconds=1)
+ completed[path] =
Timestamp.from_local_date_time(clock[0]).to_millis_timestamp()
+ return result
+
+ try:
+ with patch.object(table.file_io, 'write_parquet',
side_effect=write_and_advance), \
+ patch.object(Timestamp, 'now', side_effect=lambda:
Timestamp.from_local_date_time(clock[0])):
+ writer.write_arrow(pa.Table.from_pylist([
+ {'id': 1, 'value': 10}, {'id': 1, 'value': 20}],
schema=schema))
+ messages = writer.prepare_commit(1) if stream else
writer.prepare_commit()
+ data_files = [file for message in messages for file in
message.new_files]
+ changelogs = [file for message in messages for file in
message.changelog_files]
+ assert data_files and changelogs
+ assert sum(file.row_count for file in data_files) == 1
+ assert sum(file.row_count for file in changelogs) == 2
+ for file in data_files + changelogs:
+ assert file.creation_time == completed[file.file_path]
+ commit.commit(messages, 1) if stream else commit.commit(messages)
+ finally:
+ writer.close()
+ commit.close()
+ read = table.new_read_builder()
+ assert
read.new_read().to_arrow(read.new_scan().plan().splits()).to_pylist() ==
[{'id': 1, 'value': 20}]
diff --git a/paimon-python/pypaimon/tests/reader_primary_key_test.py
b/paimon-python/pypaimon/tests/reader_primary_key_test.py
index 438e57ac64..c8927429cb 100644
--- a/paimon-python/pypaimon/tests/reader_primary_key_test.py
+++ b/paimon-python/pypaimon/tests/reader_primary_key_test.py
@@ -20,6 +20,7 @@ import shutil
import tempfile
import time
import unittest
+from datetime import datetime, timezone
import pyarrow as pa
@@ -562,14 +563,12 @@ class PkReaderTest(unittest.TestCase):
epoch_millis = entry.file.creation_time_epoch_millis()
self.assertIsNotNone(epoch_millis)
self.assertGreater(epoch_millis, 0)
- import time
- expected_epoch_millis = creation_time.get_millisecond()
- local_dt = creation_time.to_local_date_time()
- local_time_struct = local_dt.timetuple()
- local_timestamp = time.mktime(local_time_struct)
- local_time_struct_utc = time.gmtime(local_timestamp)
- utc_timestamp = time.mktime(local_time_struct_utc)
- expected_epoch_millis = int(utc_timestamp * 1000)
+ # Java resolves the stored local TIMESTAMP in the system
+ # zone once and retains all millisecond digits.
+ instant =
creation_time.to_local_date_time().astimezone(timezone.utc)
+ delta = instant - datetime(1970, 1, 1, tzinfo=timezone.utc)
+ expected_epoch_millis = ((delta.days * 86400 +
delta.seconds) * 1000
+ + delta.microseconds // 1000)
self.assertEqual(epoch_millis, expected_epoch_millis)
creation_times_found.append(epoch_millis)
diff --git a/paimon-python/pypaimon/tests/write/row_kind_nullability_test.py
b/paimon-python/pypaimon/tests/write/row_kind_nullability_test.py
new file mode 100644
index 0000000000..62652600f1
--- /dev/null
+++ b/paimon-python/pypaimon/tests/write/row_kind_nullability_test.py
@@ -0,0 +1,295 @@
+# 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.
+
+"""Filtered events retain Java's row-kind-specific top-level NULL checks."""
+
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.table.row.row_kind import RowKind
+from pypaimon.write.native_write import NativeTableWrite
+from pypaimon.write.row_kind import skip_write_row
+
+
+pytestmark = [pytest.mark.python_write, pytest.mark.python_commit]
+_NATIVE = pytest.mark.skipif(not native_runtime_available(),
reason='pypaimon-rust not installed')
+_ENGINES = [False, pytest.param(True, marks=[pytest.mark.native_plan,
_NATIVE])]
+
+
+def _table(tmp_path, **options):
+ schema = pa.schema([pa.field('id', pa.int32(), nullable=False),
+ pa.field('op', pa.string()),
+ pa.field('seq', pa.int64(), nullable=False),
+ pa.field('value', pa.int32(), nullable=False)])
+ config = {'bucket': '1', 'rowkind.field': 'op', 'ignore-delete': 'true',
+ 'sequence.field': 'seq', 'write.native.enabled': 'false',
+ 'commit.native.enabled': 'false', 'read.native.enabled': 'false',
+ 'scan.native-plan.enabled': 'false'}
+ config.update(options)
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ schema, primary_keys=['id'], options=config), False)
+ return catalog.get_table('default.t'), schema
+
+
+def _write(writer, table, schema, rows, input_):
+ if input_ == 'row':
+ for row in rows:
+ writer.write_row(GenericRow([row[field.name] for field in
table.fields], table.fields))
+ elif input_ == 'batch':
+ writer.write_arrow_batch(pa.RecordBatch.from_pylist(rows,
schema=schema))
+ elif input_ == 'pandas':
+ import pandas as pd
+ writer.write_pandas(pd.DataFrame.from_records(rows,
columns=schema.names))
+ else:
+ writer.write_arrow(pa.Table.from_pylist(rows, schema=schema))
+
+
[email protected]('input_', ['row', 'batch', 'table', 'pandas'])
[email protected]('stream', [False, True], ids=['batch', 'stream'])
[email protected]('kind,field', [('-D', 'id'), ('-D', 'seq'), ('-U',
'value'), ('+I', 'value')])
+def test_filtered_events_check_required_nulls_before_routing(tmp_path, input_,
stream, kind, field):
+ table, schema = _table(tmp_path)
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ writer = builder.new_write()
+ invalid = dict(id=1, op=kind, seq=1, value=10)
+ invalid[field] = None
+ try:
+ with pytest.raises(ValueError, match=r'non-null column\(%s\)' %
field), \
+ patch.object(writer.row_key_extractor,
'extract_partition_bucket_groups',
+ side_effect=AssertionError('invalid event reached
bucket routing')):
+ _write(writer, table, schema, [invalid], input_)
+ assert not writer.file_store_write.data_writers
+ finally:
+ writer.close()
+
+
[email protected]('input_', ['row', 'batch', 'table', 'pandas'])
[email protected]('stream', [False, True], ids=['batch', 'stream'])
[email protected]('native', _ENGINES, ids=['python', 'native'])
+def test_ignored_delete_may_omit_non_key_not_null_values(tmp_path, input_,
stream, native):
+ table, schema = _table(tmp_path, **{'write.native.enabled':
str(native).lower()})
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ expected = dict(id=2, op='+I', seq=2, value=42)
+ try:
+ assert isinstance(writer, NativeTableWrite) == native
+ _write(writer, table, schema, [dict(id=1, op='-D', seq=1, value=None),
expected], input_)
+ if native:
+ assert writer._python_writer is None
+ messages = writer.prepare_commit(1) if stream else
writer.prepare_commit()
+ commit.commit(messages, 1) if stream else commit.commit(messages)
+ finally:
+ writer.close()
+ commit.close()
+ builder = table.new_read_builder()
+ assert
builder.new_read().to_arrow(builder.new_scan().plan().splits()).to_pylist() ==
[expected]
+
+
[email protected]('input_', ['batch', 'table'])
[email protected]('stream', [False, True], ids=['batch', 'stream'])
[email protected]('native', _ENGINES, ids=['python', 'native'])
+def test_large_string_row_kinds_filter_before_casting_payloads(tmp_path,
input_, stream, native):
+ table, schema = _table(tmp_path, **{'write.native.enabled':
str(native).lower()})
+ transport = pa.schema([field.with_type(pa.large_string()) if field.name ==
'op' else field
+ for field in schema])
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ expected = dict(id=2, op='+I', seq=2, value=42)
+ try:
+ assert isinstance(writer, NativeTableWrite) == native
+ # A whole-table STRING cast checks unrelated top-level NOT NULL
+ # fields, and would reject this DELETE before its row kind is applied.
+ _write(writer, table, transport, [dict(id=1, op='-D', seq=1,
value=None), expected], input_)
+ if native:
+ assert writer._python_writer is None
+ messages = writer.prepare_commit(1) if stream else
writer.prepare_commit()
+ commit.commit(messages, 1) if stream else commit.commit(messages)
+ finally:
+ writer.close()
+ commit.close()
+ builder = table.new_read_builder()
+ assert
builder.new_read().to_arrow(builder.new_scan().plan().splits()).to_pylist() ==
[expected]
+
+
[email protected]_plan
+@_NATIVE
[email protected]('input_', ['row', 'batch', 'table', 'pandas'])
[email protected]('stream', [False, True], ids=['batch', 'stream'])
[email protected]('kind,field', [('-D', 'id'), ('-D', 'seq'), ('-U',
'value'), ('+I', 'value')])
+def test_native_filtered_events_keep_required_null_checks(tmp_path, input_,
stream, kind, field):
+ table, schema = _table(tmp_path, **{'write.native.enabled': 'true'})
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ writer = builder.new_write()
+ invalid = dict(id=1, op=kind, seq=1, value=10)
+ invalid[field] = None
+ try:
+ assert isinstance(writer, NativeTableWrite)
+ with pytest.raises(ValueError, match=r'non-null column\(%s\)' % field):
+ _write(writer, table, schema, [invalid], input_)
+ assert writer._python_writer is None
+ assert not list(tmp_path.rglob('*.parquet'))
+ assert table.snapshot_manager().get_latest_snapshot() is None
+ finally:
+ writer.close()
+
+
+def
test_native_pandas_facade_preserves_null_payload_for_core_filtering(tmp_path):
+ import pandas as pd
+
+ table, schema = _table(tmp_path)
+ native = Mock()
+ writer = NativeTableWrite(table, 'user', None, False, native)
+ rows = [dict(id=1, op='-D', seq=1, value=None), dict(id=2, op='+I', seq=2,
value=42)]
+ writer.write_pandas(pd.DataFrame.from_records(rows, columns=schema.names))
+ native.write_arrow.assert_called_once()
+ transported = native.write_arrow.call_args[0][0]
+ assert transported.to_pylist() == rows
+ assert transported.schema.field('value').nullable
+ assert not writer._schema.field('value').nullable
+ assert writer._python_writer is None
+
+
[email protected]('primary_key', [False, True], ids=['append',
'pk-without-filter'])
+def test_pandas_non_filtering_writers_keep_not_null_validation(tmp_path,
primary_key):
+ import pandas as pd
+
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([pa.field('id', pa.int32(), nullable=False),
+ pa.field('value', pa.int32(), nullable=False)])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ schema, primary_keys=['id'] if primary_key else [], options={
+ 'bucket': '1', 'write.native.enabled': 'false'}), False)
+ table = catalog.get_table('default.t')
+ data = pd.DataFrame([dict(id=1, value=None)])
+ writer = table.new_batch_write_builder().new_write()
+ facade = NativeTableWrite(table, 'native-user', None, False, Mock())
+ try:
+ for candidate in [writer, facade]:
+ with pytest.raises(ValueError, match='non-nullable'):
+ candidate.write_pandas(data)
+ assert not writer.file_store_write.data_writers
+ facade._native_writer.write_arrow.assert_not_called()
+ finally:
+ writer.close()
+
+
+def
test_delete_null_exception_is_limited_to_non_cross_partition_deduplicate(tmp_path):
+ table, _ = _table(tmp_path)
+ values = dict(id=1, op='-D', seq=1, value=None)
+ assert skip_write_row(table, values)
+ for variant, cross_partition in [(table.copy({'merge-engine':
'first-row'}), False), (table, True)]:
+ with patch.object(variant, 'cross_partition_update', cross_partition),
\
+ pytest.raises(ValueError, match=r'non-null column\(value\)'):
+ skip_write_row(variant, values)
+
+
[email protected]('kind,field', [
+ (RowKind.DELETE, 'id'), (RowKind.DELETE, 'seq'), (RowKind.UPDATE_BEFORE,
'value')])
+def test_internal_row_kinds_check_required_nulls_before_filtering(tmp_path,
kind, field):
+ table, _ = _table(tmp_path, **{'rowkind.field': None})
+ writer = table.new_batch_write_builder().new_write()
+ values = dict(id=1, op=None, seq=1, value=10)
+ values[field] = None
+ try:
+ with pytest.raises(ValueError, match=r'non-null column\(%s\)' % field):
+ writer.write_row(GenericRow([values[field.name] for field in
table.fields], table.fields, kind))
+ assert not writer.file_store_write.data_writers
+ finally:
+ writer.close()
+
+
+def test_native_bridge_leaves_nested_value_validation_to_core(tmp_path):
+ transport = pa.schema([('items', pa.list_(pa.int32()))])
+ declared = pa.schema([('items', pa.list_(pa.field('element', pa.int32(),
nullable=False)))])
+ data = pa.RecordBatch.from_arrays([pa.array([[None]],
type=transport.field(0).type)], schema=transport)
+ writer = NativeTableWrite.__new__(NativeTableWrite)
+ writer._table_schema = writer._schema = declared
+ writer._write_cols = None
+ with patch('pypaimon.write.native_write.normalize_arrow_write_nullability',
+ side_effect=AssertionError('bridge inspected values before core
filtering'), create=True), \
+ patch('pypaimon.write.native_write.normalize_arrow_strings',
+ side_effect=AssertionError('bridge cast values before core
filtering'), create=True):
+ assert writer._prepare_native_arrow_data(data) is data
+
+
+def test_schema_only_writer_keeps_strict_nested_validation():
+ from pypaimon.write.table_write import TableWrite
+
+ writer = TableWrite.__new__(TableWrite)
+ writer.file_store_write = SimpleNamespace(write_cols=None)
+ writer.table_pyarrow_schema = pa.schema([
+ ('items', pa.list_(pa.field('element', pa.int32(), nullable=False)))])
+ data = pa.table({'items': [[None]]}, schema=pa.schema([('items',
pa.list_(pa.int32()))]))
+ with pytest.raises(ValueError, match='consistent'):
+ writer._prepare_arrow_data(data)
+
+
+def test_python_filters_before_nested_normalization(tmp_path):
+ from pypaimon.write.table_write import TableWrite
+
+ table, _ = _table(tmp_path)
+ writer = table.new_batch_write_builder().new_write()
+ declared = pa.schema([field if field.name != 'value' else field.with_type(
+ pa.list_(pa.field('element', pa.int32(), nullable=False))) for field
in writer.table_pyarrow_schema])
+ # Use a real table for options and routing, but a schema-only writer facade
+ # to observe normalization before any actual Parquet output is attempted.
+ writer.table_pyarrow_schema = declared
+ transport = pa.schema([field if field.name != 'value' else
field.with_type(pa.list_(pa.int32()))
+ for field in declared])
+ data = pa.table({'id': [1, 2], 'op': ['-D', '+I'], 'seq': [1, 2], 'value':
[[None], [42]]}, schema=transport)
+
+ def normalize(selected, expected):
+ assert selected['id'].to_pylist() == [2]
+ assert expected == declared
+ return selected.cast(expected)
+
+ try:
+ with
patch('pypaimon.write.table_write.normalize_arrow_write_nullability',
side_effect=normalize):
+ selected = TableWrite._prepare_arrow_data(writer, data)
+ assert selected.to_pydict() == {'id': [2], 'op': ['+I'], 'seq': [2],
'value': [[42]]}
+ finally:
+ writer.close()
+
+
[email protected]('ignored', [False, True])
+def
test_python_retained_nested_schema_stays_strict_without_native_extension(tmp_path,
ignored):
+ table, _ = _table(tmp_path)
+ writer = table.new_batch_write_builder().new_write()
+ writer.table_pyarrow_schema = pa.schema([field if field.name != 'value'
else field.with_type(
+ pa.list_(pa.field('element', pa.int32(), nullable=False))) for field
in writer.table_pyarrow_schema])
+ transport = pa.schema([field if field.name != 'value' else
field.with_type(pa.list_(pa.int32()))
+ for field in writer.table_pyarrow_schema])
+ data = pa.table({'id': [1], 'op': ['-D' if ignored else '+I'], 'seq': [1],
'value': [[None]]}, schema=transport)
+ try:
+ with patch('pypaimon.read.native_plan.native_runtime_available',
return_value=False):
+ if ignored:
+ assert writer._prepare_arrow_data(data).num_rows == 0
+ else:
+ with pytest.raises(ValueError, match='consistent'):
+ writer._prepare_arrow_data(data)
+ assert not writer.file_store_write.data_writers
+ finally:
+ writer.close()
diff --git a/paimon-python/pypaimon/utils/file_store_path_factory.py
b/paimon-python/pypaimon/utils/file_store_path_factory.py
index c1c956e560..d5920e1cdb 100644
--- a/paimon-python/pypaimon/utils/file_store_path_factory.py
+++ b/paimon-python/pypaimon/utils/file_store_path_factory.py
@@ -163,7 +163,7 @@ 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:
+ if value is not None and type_name in ('BINARY', 'VARBINARY',
'BYTES'):
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
@@ -189,7 +189,9 @@ 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:
+ elif type_name in ('BINARY', 'VARBINARY', 'BYTES'):
+ # Java uses the binary-to-string cast even in legacy mode:
+ # byte[].toString contains a fresh allocation identity.
text = value
elif data_type is not None and not _is_unsupported(data_type):
text = cast_value_to_string(value, data_type)
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index adc230e51a..ac07c73312 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -1079,31 +1079,30 @@ class FileStoreCommit:
partition_spec = dict(zip(self.table.partition_keys,
path_factory._canonical_partition(partition_value)))
partition_key = GenericRowSerializer.to_bytes(entry.partition)
+ file_meta = entry.file
+ file_creation_time = (file_meta.creation_time_epoch_millis()
+ if file_meta.creation_time is not None else
int(time.time() * 1000))
+
if partition_key not in partition_stats:
partition_stats[partition_key] = {
'partition_spec': partition_spec,
'record_count': 0,
'file_count': 0,
'file_size_in_bytes': 0,
- 'last_file_creation_time': 0,
+ # Java starts with the first entry, including times before
+ # the epoch, rather than clamping negative times to zero.
+ 'last_file_creation_time': file_creation_time,
'total_buckets': entry.total_buckets
}
partition_stats[partition_key]['total_buckets'] = (
entry.total_buckets)
# Following Java implementation: PartitionEntry.fromDataFile()
- file_meta = entry.file
# Extract actual file metadata (following Java DataFileMeta
pattern)
record_count = file_meta.row_count if entry.kind == 0 else
file_meta.row_count * -1
file_size_in_bytes = file_meta.file_size if entry.kind == 0 else
file_meta.file_size * -1
file_count = 1 if entry.kind == 0 else -1
- # Use epoch millis
- if file_meta.creation_time:
- file_creation_time = file_meta.creation_time_epoch_millis()
- else:
- file_creation_time = int(time.time() * 1000)
-
# Accumulate statistics (following Java PartitionEntry.merge()
logic)
partition_stats[partition_key]['record_count'] += record_count
partition_stats[partition_key]['file_size_in_bytes'] +=
file_size_in_bytes
diff --git a/paimon-python/pypaimon/write/native_commit.py
b/paimon-python/pypaimon/write/native_commit.py
index d3cb5235dd..a1731a528a 100644
--- a/paimon-python/pypaimon/write/native_commit.py
+++ b/paimon-python/pypaimon/write/native_commit.py
@@ -51,13 +51,6 @@ def create_native_commit(table, commit_user,
overwrite_partition=None, *, batch=
"""Return a native committer only when its publication protocol matches
Python."""
if not _rest_catalog_supported(table) or not native_runtime_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
@@ -94,9 +87,7 @@ def _rest_catalog_supported(table):
and environment.supports_version_management
and _rest_source_uuid(table) is not None
and context is not None
- and context.options is not None
- and all(getattr(context, attr, None) is None for attr in (
- 'hadoop_conf', 'prefer_io_loader', 'fallback_io_loader')))
+ and context.options is not None)
def _rest_source_uuid(table):
diff --git a/paimon-python/pypaimon/write/native_merge_into.py
b/paimon-python/pypaimon/write/native_merge_into.py
index 717950763f..c9999217ca 100644
--- a/paimon-python/pypaimon/write/native_merge_into.py
+++ b/paimon-python/pypaimon/write/native_merge_into.py
@@ -19,15 +19,12 @@
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 (
_dictionary_value_type, _native_row_id_table,
_native_update_columns_supported,
_native_update_formats_supported, _supported_upsert_key_type,
)
-from pypaimon.write.native_write import _native_partition_types_supported
def _callable_assignment(function, on_pairs):
@@ -76,10 +73,6 @@ 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,
-
source.options.options.get(CoreOptions.PARTITION_GENERATE_LEGACY_NAME)):
- return None
if source.options.query_auth_enabled:
from pypaimon.read.native_plan import
_native_query_auth_supported, _native_table
from pypaimon.write.native_commit import _rest_catalog_supported
diff --git a/paimon-python/pypaimon/write/native_update.py
b/paimon-python/pypaimon/write/native_update.py
index a7825c13dd..dd1bb33b83 100644
--- a/paimon-python/pypaimon/write/native_update.py
+++ b/paimon-python/pypaimon/write/native_update.py
@@ -25,14 +25,13 @@ import pyarrow.compute as pc
from pypaimon.read.native_plan import _native_query_auth_supported,
native_runtime_available
from pypaimon.schema.arrow_schema import arrow_schemas_compatible
from pypaimon.schema.data_types import MapType, PyarrowFieldParser, RowType,
VectorType, is_blob_file_field
-from pypaimon.common.options.core_options import ChangelogProducer, CoreOptions
+from pypaimon.common.options.core_options import ChangelogProducer
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
from pypaimon.write.native_commit import (
create_native_write_table, from_native_commit_messages,
)
-from pypaimon.write.native_write import _native_partition_types_supported
from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
from pypaimon.write.row_utils import _contains_blob_value, value_for_arrow
@@ -53,10 +52,6 @@ def _native_update_table(table, *, allow_changelog=False):
or not native_runtime_available()
or not _native_query_auth_supported(table)):
return None
- schema = PyarrowFieldParser.from_paimon_schema(table.table_schema.fields)
- 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 ce74056e24..ef2efe179e 100644
--- a/paimon-python/pypaimon/write/native_write.py
+++ b/paimon-python/pypaimon/write/native_write.py
@@ -18,10 +18,9 @@
import pyarrow as pa
-from pypaimon.common.options.core_options import ChangelogProducer, CoreOptions
+from pypaimon.common.options.core_options import ChangelogProducer
from pypaimon.read.native_plan import native_runtime_available
-from pypaimon.schema.arrow_schema import (
- arrow_schemas_compatible, normalize_arrow_strings,
normalize_arrow_write_nullability)
+from pypaimon.schema.arrow_schema import arrow_schemas_compatible
from pypaimon.schema.data_types import PyarrowFieldParser, is_blob_file_field
from pypaimon.table.bucket_mode import BucketMode
from pypaimon.write.native_commit import (
@@ -30,20 +29,9 @@ from pypaimon.write.native_commit import (
from pypaimon.write.row_utils import require_columns, row_to_named_values,
row_values_to_arrow_table
-def _native_partition_types_supported(schema, partition_keys,
legacy_partition_name=True):
- """Partition keys which Rust can encode and use to locate existing
files."""
- # 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)
- for data_type in (schema.field(name).type for name in partition_keys))
-
-
def create_native_write(table, commit_user, static_partition=None,
stream=False,
*, fixed_bucket=False, bucket_plan=None,
restore_snapshot_id=None):
"""Return a native writer if the table can use the filesystem write
path."""
- schema = PyarrowFieldParser.from_paimon_schema(table.table_schema.fields)
# DV tables flush normal level-0 files using the configured merge engine.
# Batch visibility is controlled by deletion-vectors.merge-on-read.
if (not native_runtime_available()
@@ -57,8 +45,6 @@ def create_native_write(table, commit_user,
static_partition=None, stream=False,
and table.options.changelog_producer() ==
ChangelogProducer.INPUT
and (stream or static_partition is None))
or table.options.file_format() != 'parquet'
- 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)
@@ -226,10 +212,10 @@ class NativeTableWrite:
self._native_writer.write_arrow(data, bucket)
def _prepare_native_arrow_data(self, data):
- data = normalize_arrow_strings(data)
- target = self._table_schema if data.schema.names ==
self._table_schema.names else self._schema
- data = normalize_arrow_write_nullability(data, target)
- compatible = dict(check_top_level_nullability=False,
allow_binary_compatibility=True)
+ # Core resolves row kinds and filters ignored events before inspecting
+ # payload values. This facade checks only the logical Arrow schema.
+ compatible = dict(check_top_level_nullability=False,
allow_binary_compatibility=True,
+ check_nested_nullability=False)
if not (arrow_schemas_compatible(data.schema, self._schema,
**compatible)
or (self._write_cols is not None
and arrow_schemas_compatible(data.schema,
self._table_schema, **compatible))):
@@ -244,7 +230,9 @@ class NativeTableWrite:
def write_pandas(self, dataframe):
if self._python_writer is not None:
return self._python_writer.write_pandas(dataframe)
- self.write_arrow_batch(pa.RecordBatch.from_pandas(dataframe,
schema=self._schema))
+ from pypaimon.write.table_write import _pandas_write_schema
+ transport = _pandas_write_schema(self.table, self._schema)
+ self.write_arrow_batch(pa.RecordBatch.from_pandas(dataframe,
schema=transport))
def write_ray(self, dataset, overwrite=False, concurrency=None,
ray_remote_args=None,
hash_fixed_precluster='auto', static_partition=None):
diff --git a/paimon-python/pypaimon/write/row_kind.py
b/paimon-python/pypaimon/write/row_kind.py
index df279aadf6..ad66bf0cbc 100644
--- a/paimon-python/pypaimon/write/row_kind.py
+++ b/paimon-python/pypaimon/write/row_kind.py
@@ -21,7 +21,7 @@ from typing import List, Union
import pyarrow as pa
import pyarrow.compute as pc
-from pypaimon.common.options.core_options import CoreOptions
+from pypaimon.common.options.core_options import CoreOptions, MergeEngine
from pypaimon.table.row.row_kind import RowKind
from pypaimon.table.special_fields import SpecialFields
@@ -41,7 +41,7 @@ def row_kinds(
'Cannot find rowkind field %s in table schema' %
row_kind_field_name
)
column = data.column(row_kind_field_name)
- if not pa.types.is_string(column.type):
+ if not (pa.types.is_string(column.type) or
pa.types.is_large_string(column.type)):
raise ValueError('rowkind.field column must be a string')
kinds: List[int] = []
for value in column.to_pylist():
@@ -63,6 +63,15 @@ def filter_write_batch(table, data):
ignore_before = str(raw.get('ignore-update-before', 'false')).lower() ==
'true'
if not ignore_delete and not ignore_before:
return data
+ # Java checks top-level NULLs before its RowKindFilter. A DELETE in a
+ # non-cross-partition deduplicate table only requires non-NULL keys and
+ # sequence fields; UPDATE_BEFORE retains all declared NOT NULL fields.
+ for name, check_delete in _not_null_fields(table):
+ invalid = data.column(name).is_null()
+ if not check_delete:
+ invalid = pc.and_(invalid, pc.not_equal(kinds,
RowKind.DELETE.value))
+ if pc.any(invalid).as_py():
+ raise ValueError('Cannot write null to non-null column(%s)' % name)
ignored = [RowKind.UPDATE_BEFORE.value]
if ignore_delete:
ignored.append(RowKind.DELETE.value)
@@ -100,9 +109,23 @@ def skip_write_row(table, values, row_kind=RowKind.INSERT):
if row_kind_field_name is None
else RowKind.from_string(values[row_kind_field_name]).value
)
+ if options.ignore_delete() or str(raw.get('ignore-update-before',
'false')).lower() == 'true':
+ for name, check_delete in _not_null_fields(table):
+ if (check_delete or kind != RowKind.DELETE.value) and values[name]
is None:
+ raise ValueError('Cannot write null to non-null column(%s)' %
name)
return _is_filtered(kind, options.ignore_delete(),
str(raw.get('ignore-update-before', 'false')).lower()
== 'true')
+def _not_null_fields(table):
+ """Fields Java checks before filtering, with its DELETE-only exception."""
+ delete_fields = None
+ if (table.options.merge_engine() == MergeEngine.DEDUPLICATE
+ and not table.cross_partition_update):
+ delete_fields = set(table.table_schema.primary_keys +
table.options.sequence_field())
+ return [(field.name, delete_fields is None or field.name in delete_fields)
+ for field in table.fields if not field.type.nullable]
+
+
def _is_filtered(kind, ignore_delete, ignore_before):
return (ignore_delete and kind in (1, 3)) or (ignore_before and kind == 1)
diff --git a/paimon-python/pypaimon/write/table_write.py
b/paimon-python/pypaimon/write/table_write.py
index 65c9996326..6c02c33a0b 100644
--- a/paimon-python/pypaimon/write/table_write.py
+++ b/paimon-python/pypaimon/write/table_write.py
@@ -36,6 +36,20 @@ if TYPE_CHECKING:
from ray.data import Dataset
+def _pandas_write_schema(table, schema):
+ """Keep ignored events available for the normal row-kind NULL checks."""
+ if table is None or not table.is_primary_key_table:
+ return schema
+ options = table.options
+ raw = options.options.to_map()
+ if (raw.get('rowkind.field') is None
+ or not (options.ignore_delete() or
str(raw.get('ignore-update-before', 'false')).lower() == 'true')):
+ return schema
+ # Pandas checks top-level NOT NULL before Arrow reaches the writer. Only
+ # relax the transport; the declared schema and nested fields stay intact.
+ return pa.schema([field.with_nullable(True) for field in schema],
metadata=schema.metadata)
+
+
def _write_ray(writer, dataset, overwrite=False, concurrency=None,
ray_remote_args=None,
hash_fixed_precluster='auto', static_partition=None):
"""Share Ray coordination without constructing another local data
writer."""
@@ -205,7 +219,8 @@ class TableWrite:
pa_schema = self._write_cols_pyarrow_schema(write_cols)
else:
pa_schema = self.table_pyarrow_schema
- record_batch = pa.RecordBatch.from_pandas(dataframe, schema=pa_schema)
+ transport = _pandas_write_schema(getattr(self, 'table', None),
pa_schema)
+ record_batch = pa.RecordBatch.from_pandas(dataframe, schema=transport)
return self.write_arrow_batch(record_batch)
def with_write_type(self, write_cols: List[str]):
@@ -251,7 +266,7 @@ class TableWrite:
) -> None:
"""
Write a Ray Dataset to Paimon table.
-
+
Args:
dataset: Ray Dataset to write. This is a distributed data
collection
from Ray Data (ray.data.Dataset).
@@ -328,18 +343,26 @@ class TableWrite:
return data
data = self._normalize_arrow_write_schema(data)
from pypaimon.write.row_kind import filter_write_batch
- return filter_write_batch(table, data)
-
- def _prepare_arrow_schema(self, data):
+ data = filter_write_batch(table, data)
+ if not data.num_rows:
+ return data
data = normalize_arrow_strings(data)
target = self.table_pyarrow_schema
if data.schema.names != target.names and
self.file_store_write.write_cols is not None:
target =
self._write_cols_pyarrow_schema(self.file_store_write.write_cols)
+ # Validate only surviving values. Keep the input transport metadata:
+ # PyArrow otherwise rejects hidden NULL children of NULL ROW parents.
validated = normalize_arrow_write_nullability(data, target)
self._validate_pyarrow_schema(validated.schema)
- # Core checks only visible NOT NULL children. PyArrow's Parquet writer
- # also rejects hidden NULL children under NULL ROW parents when their
- # Arrow field is nonnullable, so retain the input's transport metadata.
+ return data
+
+ def _prepare_arrow_schema(self, data):
+ # Legacy schema-only callers have no row-kind options to evaluate.
+ # A real writer must defer casts until ignored events have been
removed.
+ if getattr(self, 'table', None) is None:
+ data = normalize_arrow_strings(data)
+ self._validate_pyarrow_schema(
+ data.schema, check_nested_nullability=getattr(self, 'table', None)
is None)
return data
def _normalize_arrow_write_schema(self, data):
@@ -358,22 +381,25 @@ class TableWrite:
return pa.RecordBatch.from_arrays(data.columns, schema=schema)
return pa.Table.from_arrays(data.columns, schema=schema)
- def _validate_pyarrow_schema(self, data_schema: pa.Schema):
- if self._is_compatible_pyarrow_schema(data_schema,
self.table_pyarrow_schema):
+ def _validate_pyarrow_schema(self, data_schema: pa.Schema, *,
check_nested_nullability=True):
+ if self._is_compatible_pyarrow_schema(
+ data_schema, self.table_pyarrow_schema,
check_nested_nullability=check_nested_nullability):
return
write_cols = self.file_store_write.write_cols
if write_cols is not None:
write_cols_schema = self._write_cols_pyarrow_schema(write_cols)
- if self._is_compatible_pyarrow_schema(data_schema,
write_cols_schema):
+ if self._is_compatible_pyarrow_schema(
+ data_schema, write_cols_schema,
check_nested_nullability=check_nested_nullability):
return
self._raise_inconsistent_schema(data_schema)
def _is_compatible_pyarrow_schema(
- self, data_schema: pa.Schema, expected_schema: pa.Schema) -> bool:
+ self, data_schema: pa.Schema, expected_schema: pa.Schema, *,
check_nested_nullability=True) -> bool:
return arrow_schemas_compatible(
- data_schema, expected_schema, check_top_level_nullability=False,
allow_binary_compatibility=True)
+ data_schema, expected_schema, check_top_level_nullability=False,
+ allow_binary_compatibility=True,
check_nested_nullability=check_nested_nullability)
def _write_cols_pyarrow_schema(self, write_cols: List[str]) -> pa.Schema:
table_fields = {
diff --git a/paimon-python/pypaimon/write/writer/blob_writer.py
b/paimon-python/pypaimon/write/writer/blob_writer.py
index 166144e5c6..47ba95778c 100644
--- a/paimon-python/pypaimon/write/writer/blob_writer.py
+++ b/paimon-python/pypaimon/write/writer/blob_writer.py
@@ -268,7 +268,7 @@ class BlobWriter(AppendOnlyDataWriter):
schema_id=self.table.table_schema.id,
level=0,
extra_files=[],
- creation_time=Timestamp.now(),
+ creation_time=Timestamp.now().to_millis_timestamp(),
delete_row_count=0,
file_source=0, # FileSource.APPEND = 0
value_stats_cols=None,
diff --git a/paimon-python/pypaimon/write/writer/composite_data_writer.py
b/paimon-python/pypaimon/write/writer/composite_data_writer.py
index a9d5c66e6f..7887582cd9 100644
--- a/paimon-python/pypaimon/write/writer/composite_data_writer.py
+++ b/paimon-python/pypaimon/write/writer/composite_data_writer.py
@@ -169,7 +169,7 @@ class CompositeDataWriter(DataWriter):
schema_id=self.table.table_schema.id,
level=0,
extra_files=[],
- creation_time=Timestamp.now(),
+ creation_time=Timestamp.now().to_millis_timestamp(),
delete_row_count=0,
file_source=0,
value_stats_cols=[column.name for column in stats_columns],
diff --git a/paimon-python/pypaimon/write/writer/data_writer.py
b/paimon-python/pypaimon/write/writer/data_writer.py
index 98cdd66d2b..f2a68b1218 100644
--- a/paimon-python/pypaimon/write/writer/data_writer.py
+++ b/paimon-python/pypaimon/write/writer/data_writer.py
@@ -323,7 +323,6 @@ class DataWriter(ABC):
min_sequence_number=statistics['min_seq'],
max_sequence_number=statistics['max_seq'],
extra_files=extra_files,
- creation_time=statistics['creation_time'],
delete_row_count=self._count_delete_rows(data),
)
@@ -382,10 +381,9 @@ class DataWriter(ABC):
else:
min_seq = self.sequence_generator.start
max_seq = self.sequence_generator.current
- creation_time = Timestamp.now()
return dict(min_key=min_key, max_key=max_key, key_stats=key_stats,
value_stats=value_stats, min_seq=min_seq, max_seq=max_seq,
- creation_time=creation_time,
value_stats_enabled=value_stats_enabled)
+ value_stats_enabled=value_stats_enabled)
def _writes_changelog_before_merge(self):
return False
@@ -424,7 +422,7 @@ class DataWriter(ABC):
max_sequence_number=max_sequence_number,
schema_id=self.table.table_schema.id, level=0,
extra_files=extra_files if extra_files is not None else [],
- creation_time=creation_time if creation_time is not None else
Timestamp.now(),
+ creation_time=creation_time if creation_time is not None else
Timestamp.now().to_millis_timestamp(),
delete_row_count=delete_row_count, file_source=0,
value_stats_cols=None if self.options.metadata_stats_enabled()
else [],
external_path=file_path if self.external_path_provider is not None
else None,
@@ -468,7 +466,7 @@ class DataWriter(ABC):
return pa.Table.from_arrays(columns, schema=pa.schema(fields))
def _write_changelog_file(self, data, min_key, max_key, key_stats,
value_stats,
- min_seq, max_seq, creation_time,
+ min_seq, max_seq,
value_stats_enabled, is_external) ->
DataFileMeta:
"""Write a complete changelog file and return its metadata.
@@ -502,7 +500,7 @@ class DataWriter(ABC):
schema_id=self.table.table_schema.id,
level=0,
extra_files=[],
- creation_time=creation_time,
+ creation_time=Timestamp.now().to_millis_timestamp(),
delete_row_count=self._count_delete_rows(data),
file_source=0,
value_stats_cols=None if value_stats_enabled else [],
diff --git a/paimon-python/pypaimon/write/writer/vector_writer.py
b/paimon-python/pypaimon/write/writer/vector_writer.py
index 50663af7b0..68010ce4a4 100644
--- a/paimon-python/pypaimon/write/writer/vector_writer.py
+++ b/paimon-python/pypaimon/write/writer/vector_writer.py
@@ -117,7 +117,7 @@ class VectorWriter(AppendOnlyDataWriter):
schema_id=self.table.table_schema.id,
level=0,
extra_files=[],
- creation_time=Timestamp.now(),
+ creation_time=Timestamp.now().to_millis_timestamp(),
delete_row_count=0,
file_source=0,
value_stats_cols=None,
