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 a6ae9d8cf8 [python] Read partition metadata with Native SystemRead
(#10605)
a6ae9d8cf8 is described below
commit a6ae9d8cf858137bdfce28e70ac4f66162858b40
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Oct 16 20:28:18 2026 +0800
[python] Read partition metadata with Native SystemRead (#10605)
---
.github/workflows/ci-python.yml | 22 +-
.../pypaimon/table/system/native_system_table.py | 4 +-
.../pypaimon/table/system/partitions_table.py | 96 +++++---
.../pypaimon/table/system/system_table.py | 7 +
.../tests/native_partition_metadata_test.py | 248 +++++++++++++++++++++
.../pypaimon/tests/native_system_table_test.py | 5 +-
.../tests/system/native_system_table_test.py | 4 +-
.../pypaimon/tests/system/partitions_table_test.py | 25 ++-
8 files changed, 371 insertions(+), 40 deletions(-)
diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml
index c34c60d745..b6a0fb9f81 100644
--- a/.github/workflows/ci-python.yml
+++ b/.github/workflows/ci-python.yml
@@ -62,11 +62,6 @@ jobs:
restore-keys: |
ci-maven-python-${{ runner.os }}-${{ runner.arch }}-
- - name: Set up Maven
- uses: stCarolas/setup-maven@v5
- with:
- maven-version: 3.8.8
-
- name: Install system dependencies
shell: bash
run: |
@@ -86,6 +81,22 @@ jobs:
&& apt-get clean \
&& rm -rf /var/lib/apt/lists/*
+ - name: Set up Maven
+ shell: bash
+ env:
+ MAVEN_VERSION: '3.8.8'
+ run: |
+
maven_archive="${RUNNER_TEMP}/apache-maven-${MAVEN_VERSION}-bin.tar.gz"
+ # Central can return transient 404s even for an existing release.
+ curl --fail --location --retry 3 --retry-all-errors --retry-delay 2 \
+ --output "$maven_archive" \
+
"https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/${MAVEN_VERSION}/apache-maven-${MAVEN_VERSION}-bin.tar.gz";
\
+ || curl --fail --location --retry 3 --retry-all-errors
--retry-delay 2 \
+ --output "$maven_archive" \
+
"https://archive.apache.org/dist/maven/maven-3/${MAVEN_VERSION}/binaries/apache-maven-${MAVEN_VERSION}-bin.tar.gz";
+ tar -xzf "$maven_archive" -C "$RUNNER_TEMP"
+ echo "${RUNNER_TEMP}/apache-maven-${MAVEN_VERSION}/bin" >>
"$GITHUB_PATH"
+
- name: Verify Java and Maven installation
run: |
java -version
@@ -248,6 +259,7 @@ jobs:
paimon-python/pypaimon/tests/native_cast_completion_test.py \
paimon-python/pypaimon/tests/native_temporal_zone_cast_test.py \
paimon-python/pypaimon/tests/native_schema_cast_test.py \
+ paimon-python/pypaimon/tests/native_partition_metadata_test.py \
-k 'TIMESTAMP or temporal'
done
diff --git a/paimon-python/pypaimon/table/system/native_system_table.py
b/paimon-python/pypaimon/table/system/native_system_table.py
index 0123576b69..834dff23fe 100644
--- a/paimon-python/pypaimon/table/system/native_system_table.py
+++ b/paimon-python/pypaimon/table/system/native_system_table.py
@@ -23,11 +23,9 @@ from pypaimon.schema.data_types import PyarrowFieldParser
from pypaimon.read.native_plan import _native_table, native_method_available
from pypaimon.write.native_commit import _rest_catalog_supported
-# Partitions need the catalog's list-partitions operation, which is not carried
-# by a resolved Native Table. Keep that table on the existing catalog path.
_NATIVE_TABLES = frozenset((
'snapshots', 'schemas', 'options', 'manifests', 'files', 'buckets',
- 'tags', 'branches', 'file_key_ranges', 'table_indexes', 'consumers',
+ 'tags', 'branches', 'file_key_ranges', 'table_indexes', 'consumers',
'partitions',
))
diff --git a/paimon-python/pypaimon/table/system/partitions_table.py
b/paimon-python/pypaimon/table/system/partitions_table.py
index 70685fee0b..af0b805b07 100644
--- a/paimon-python/pypaimon/table/system/partitions_table.py
+++ b/paimon-python/pypaimon/table/system/partitions_table.py
@@ -21,12 +21,17 @@ from typing import List, Optional
import pyarrow
+from pypaimon.common.options.core_options import CoreOptions
from pypaimon.index.index_file_handler import IndexFileHandler
from pypaimon.manifest.index_manifest_file import IndexManifestFile
from pypaimon.manifest.manifest_file_manager import ManifestFileManager
from pypaimon.manifest.manifest_list_manager import ManifestListManager
from pypaimon.schema.data_types import AtomicType, DataField, RowType
-from pypaimon.table.system.system_table import SystemTable
+from pypaimon.table.system.files_table import _render_object, _render_value
+from pypaimon.table.system.system_table import (
+ SystemTable, metadata_snapshot, metadata_local_millis,
metadata_epoch_millis)
+from pypaimon.table.row.generic_row import GenericRowSerializer
+from pypaimon.utils.file_store_path_factory import _is_null_or_whitespace_only
TABLE_TYPE = RowType(False, [
@@ -67,7 +72,7 @@ class PartitionsTable(SystemTable):
return ["partition"]
def _build_arrow_table(self) -> pyarrow.Table:
- snapshot = self.base_table.snapshot_manager().get_latest_snapshot()
+ snapshot = metadata_snapshot(self.base_table)
if snapshot is None:
return self._empty_table()
@@ -79,24 +84,22 @@ class PartitionsTable(SystemTable):
deleted_record_counts = self._deleted_record_counts(snapshot)
partition_map: dict = {}
+ default_name =
self.base_table.options.options.get(CoreOptions.PARTITION_DEFAULT_NAME)
+ legacy =
self.base_table.options.options.get(CoreOptions.PARTITION_GENERATE_LEGACY_NAME)
for entry in entries:
- partition_key = tuple(entry.partition.values)
- spec_items = tuple(
- (field.name, str(value))
- for field, value in zip(
- entry.partition.fields, entry.partition.values))
- spec_key = spec_items
-
+ partition_key =
bytes(GenericRowSerializer.to_bytes(entry.partition))
+ spec_key = partition_key
stats = partition_map.get(spec_key)
if stats is None:
+ partition = _normalize_partition(entry.partition,
default_name, legacy)
stats = {
- "spec_items": spec_items,
- "partition_key": partition_key,
+ "partition_string": _render_partition(partition,
default_name),
+ "deletion_key":
bytes(GenericRowSerializer.to_bytes(partition)),
"record_count": 0,
"file_size_in_bytes": 0,
"file_count": 0,
"last_update_time": None,
- "buckets": set(),
+ "total_buckets": entry.total_buckets,
}
partition_map[spec_key] = stats
@@ -104,11 +107,11 @@ class PartitionsTable(SystemTable):
stats["file_size_in_bytes"] += int(entry.file.file_size)
stats["file_count"] += 1
if entry.file.creation_time is not None:
- ct_ms = entry.file.creation_time.get_millisecond()
+ ct_ms =
metadata_epoch_millis(entry.file.creation_time.get_millisecond())
if (stats["last_update_time"] is None
or ct_ms > stats["last_update_time"]):
stats["last_update_time"] = ct_ms
- stats["buckets"].add(int(entry.bucket))
+ stats["total_buckets"] = entry.total_buckets
partition_strings: List[Optional[str]] = []
record_counts: List[int] = []
@@ -118,15 +121,16 @@ class PartitionsTable(SystemTable):
total_buckets: List[int] = []
deleted_counts: List[Optional[int]] = []
- for stats in partition_map.values():
- partition_strings.append(_render_partition(stats["spec_items"]))
+ for stats in sorted(partition_map.values(), key=lambda stats:
stats["partition_string"]):
+ partition_strings.append(stats["partition_string"])
record_counts.append(stats["record_count"])
file_sizes.append(stats["file_size_in_bytes"])
file_counts.append(stats["file_count"])
- last_update_times.append(stats["last_update_time"])
- total_buckets.append(len(stats["buckets"]))
+ last_update_times.append(None if stats["last_update_time"] is None
+ else
metadata_local_millis(stats["last_update_time"]))
+ total_buckets.append(stats["total_buckets"])
deleted_counts.append(
- deleted_record_counts.get(stats["partition_key"], 0))
+ deleted_record_counts.get(stats["deletion_key"], 0))
n = len(partition_map)
return pyarrow.table({
@@ -155,12 +159,12 @@ class PartitionsTable(SystemTable):
entries = IndexFileHandler(self.base_table).scan(
snapshot,
- lambda entry: entry.index_file.index_type
+ lambda entry: entry.kind == 0 and entry.index_file.index_type
== IndexManifestFile.DELETION_VECTORS_INDEX,
)
result = {}
for entry in entries:
- partition_key = tuple(entry.partition.values)
+ partition_key =
bytes(GenericRowSerializer.to_bytes(entry.partition))
if partition_key in result and result[partition_key] is None:
continue
@@ -197,8 +201,48 @@ class PartitionsTable(SystemTable):
})
-def _render_partition(spec_items) -> Optional[str]:
- """Render a partition spec as ``pt=v/pt2=v2`` or None when empty."""
- if not spec_items:
- return None
- return "/".join("{}={}".format(name, value) for name, value in spec_items)
+def _render_partition(partition, default_name) -> str:
+ """Java PartitionsTable uses typed SQL casts and an unescaped key=value
spec."""
+ parts = []
+ for index, field in enumerate(partition.fields):
+ value = partition.get_field(index)
+ text = default_name if value is None else _render_value(value,
field.type)
+ if _is_null_or_whitespace_only(text):
+ text = default_name
+ parts.append("{}={}".format(field.name, text))
+ return "/".join(parts)
+
+
+def _normalize_partition(partition, default_name, legacy):
+ """Java reconstructs one typed spec row for both display and deletion
lookup."""
+ from pypaimon.table.row.generic_row import (
+ GenericRow, _parse_type_precision_scale, _is_ltz_type, _normalize_ltz,
+ _datetime_to_millis_and_nanos, _millis_nanos_to_datetime)
+ values = []
+ for index, field in enumerate(partition.fields):
+ value = partition.get_field(index)
+ if value is not None:
+ text = _render_object(value, field.type) if legacy else
_render_value(value, field.type)
+ if text == default_name or _is_null_or_whitespace_only(text):
+ value = None
+ elif isinstance(value, (bytes, bytearray, memoryview)):
+ value = text.encode('utf-8')
+ elif str(field.type).upper().startswith('TIMESTAMP'):
+ precision, _ = _parse_type_precision_scale(field.type)
+ millis, nanos =
_datetime_to_millis_and_nanos(_normalize_ltz(value))
+ if _is_ltz_type(str(field.type).upper()):
+ wall_millis = millis if legacy else
metadata_local_millis(millis)
+ millis = metadata_epoch_millis(wall_millis)
+ if precision <= 3:
+ quantum = 10 ** (3 - precision)
+ millis, nanos = millis // quantum * quantum, 0
+ else:
+ quantum = 10 ** (9 - precision)
+ nanos = nanos // quantum * quantum
+ value = _millis_nanos_to_datetime(millis, nanos)
+ elif not legacy and str(field.type).upper().startswith('TIME'):
+ precision, _ = _parse_type_precision_scale(field.type)
+ quantum = 10 ** (6 - min(precision, 3))
+ value = value.replace(microsecond=value.microsecond // quantum
* quantum)
+ values.append(value)
+ return GenericRow(values, partition.fields)
diff --git a/paimon-python/pypaimon/table/system/system_table.py
b/paimon-python/pypaimon/table/system/system_table.py
index a4f2c11c82..989435c04a 100644
--- a/paimon-python/pypaimon/table/system/system_table.py
+++ b/paimon-python/pypaimon/table/system/system_table.py
@@ -47,6 +47,13 @@ def metadata_local_millis(millis):
return _signed(int(millis) + _local_offset(int(millis) // 1000) * 1000, 64)
+def metadata_epoch_millis(millis):
+ """Java DataFileMeta.creationTimeEpochMillis for a stored local
TIMESTAMP."""
+ from pypaimon.casting.temporal_utils import _local_to_utc_seconds, _signed
+ seconds, fraction = divmod(int(millis), 1000)
+ return _signed(_local_to_utc_seconds(seconds) * 1000 + fraction, 64)
+
+
def metadata_snapshot(table):
"""Use the table's captured view, including an explicitly empty
snapshot."""
if '_read_snapshot' in vars(table):
diff --git a/paimon-python/pypaimon/tests/native_partition_metadata_test.py
b/paimon-python/pypaimon/tests/native_partition_metadata_test.py
new file mode 100644
index 0000000000..2ca0a8d1bd
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_partition_metadata_test.py
@@ -0,0 +1,248 @@
+# 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.
+
+"""Native partition metadata uses the same manifest view and Java casts as
Python."""
+
+from datetime import date, datetime, time
+from decimal import Decimal
+import json
+import os
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import Schema
+from pypaimon.read.native_plan import native_method_available
+from pypaimon.index.deletion_vector_meta import DeletionVectorMeta
+from pypaimon.index.index_file_meta import IndexFileMeta
+from pypaimon.manifest.index_manifest_entry import IndexManifestEntry
+from pypaimon.manifest.index_manifest_file import IndexManifestFile
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.tests import native_plan_rest_test
+from pypaimon.tests.native_system_table_test import _canonical, _read, _system
+
+
+rest_catalog = native_plan_rest_test.rest_catalog
+
+pytestmark = [pytest.mark.native_plan, pytest.mark.skipif(
+ not native_method_available('Table', 'new_system_read'), reason='Native
system-table reader required')]
+
+
+def _create(catalog, schema, partition_keys, options=None):
+ settings = {'write.native.enabled': 'false', 'commit.native.enabled':
'false'}
+ settings.update(options or {})
+ catalog.create_table('default.part_metadata', Schema.from_pyarrow_schema(
+ schema, partition_keys=partition_keys, options=settings), False)
+ return catalog.get_table('default.part_metadata')
+
+
+def _append(table, rows, schema):
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.Table.from_pylist(rows, schema=schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+
+
+def _assert_parity(table):
+ expected = _read(_system(table, 'partitions', False))
+ native = _system(table, 'partitions', True)
+ with patch.object(type(native), '_build_arrow_table',
+ side_effect=AssertionError('Python partition aggregation
ran')):
+ actual = _read(native)
+ assert actual.schema == expected.schema
+ assert _canonical(actual, 'partitions') == _canonical(expected,
'partitions')
+ return actual.to_pylist()
+
+
[email protected]('typ,value,expected', [
+ (pa.bool_(), True, 'true'),
+ (pa.decimal128(10, 2), Decimal('1.20'), '1.20'),
+ (pa.date32(), date(2024, 2, 29), '2024-02-29'),
+ (pa.time32('ms'), time(12, 34, 56, 123000), '12:34:56'),
+ (pa.timestamp('us'), datetime(2024, 2, 29, 12, 34, 56, 123456),
'2024-02-29 12:34:56.123456'),
+ (pa.string(), 'a/b=c', 'a/b=c'),
+ (pa.float32(), 1.401298464324817e-45, '1.4E-45'),
+ (pa.float64(), 5e-324, '4.9E-324'),
+ (pa.binary(), b'hello', 'hello'),
+ (pa.binary(), b'\xed\xa0\x80', '\ufffd'),
+ (pa.string(), '\u00a0', '\u00a0'),
+ (pa.string(), '\u0085', '\u0085'),
+ (pa.string(), '\u001c', 'MISSING'),
+])
+def test_partition_values_are_java_casts(rest_catalog, typ, value, expected):
+ catalog, _ = rest_catalog
+ schema = pa.schema([('id', pa.int32()), ('p', typ)])
+ table = _create(catalog, schema, ['p'], {
+ 'partition.default-name': 'MISSING', 'partition.legacy-name': 'false'})
+ _append(table, [{'id': 1, 'p': value}, {'id': 2, 'p': None}], schema)
+ rows = _assert_parity(table)
+ assert sorted(row['partition'] for row in rows) == sorted(['p=' +
expected, 'p=MISSING'])
+ assert [row['total_buckets'] for row in rows] == [-1, -1]
+ assert [row['deleted_record_count'] for row in rows] == [0, 0]
+
+
+def test_unpartitioned_metadata_uses_empty_string(rest_catalog):
+ catalog, _ = rest_catalog
+ schema = pa.schema([('id', pa.int32())])
+ table = _create(catalog, schema, [])
+ _append(table, [{'id': 1}, {'id': 2}], schema)
+ rows = _assert_parity(table)
+ assert len(rows) == 1
+ assert rows[0]['partition'] == ''
+ assert rows[0]['record_count'] == 2
+
+
+def
test_declared_bucket_count_does_not_count_only_occupied_buckets(rest_catalog):
+ catalog, _ = rest_catalog
+ schema = pa.schema([('id', pa.int32()), ('p', pa.string())])
+ table = _create(catalog, schema, ['p'], {'bucket': '8', 'bucket-key':
'id'})
+ _append(table, [{'id': 1, 'p': 'one'}], schema)
+ rows = _assert_parity(table)
+ assert rows[0]['file_count'] == 1
+ assert rows[0]['total_buckets'] == 8
+
+
+def test_rendered_partition_collisions_are_not_aggregated(rest_catalog):
+ catalog, _ = rest_catalog
+ schema = pa.schema([('id', pa.int32()), ('p', pa.string()), ('p2',
pa.string())])
+ table = _create(catalog, schema, ['p', 'p2'])
+ _append(table, [{'id': 1, 'p': 'a/p2=b', 'p2': 'c'},
+ {'id': 2, 'p': 'a', 'p2': 'b/p2=c'}], schema)
+ rows = _assert_parity(table)
+ assert [row['partition'] for row in rows] == ['p=a/p2=b/p2=c'] * 2
+ assert [row['record_count'] for row in rows] == [1, 1]
+
+
+def
test_deletion_vector_counts_read_avro_metadata_and_keep_unknown(rest_catalog):
+ catalog, _ = rest_catalog
+ schema = pa.schema([('id', pa.int32()), ('p', pa.string())])
+ table = _create(catalog, schema, ['p'], {'deletion-vectors.enabled':
'true'})
+ _append(table, [{'id': 1, 'p': 'known'}, {'id': 2, 'p': 'unknown'},
+ {'id': 3, 'p': 'zero'}], schema)
+ entries = []
+ for partition, bucket, counts in [
+ ('known', 0, [1, 2]), ('known', 1, [3]),
+ ('unknown', 0, [None]), ('unknown', 1, [5])]:
+ ranges = {str(i): DeletionVectorMeta(str(i), 0, 1, count) for i, count
in enumerate(counts)}
+ entries.append(IndexManifestEntry(
+ 0, GenericRow([partition], table.partition_keys_fields), bucket,
+ IndexFileMeta('DELETION_VECTORS', partition + str(bucket), 1, 1,
dv_ranges=ranges)))
+ # Only manifest metadata is needed; no deletion-vector payload is opened.
+ table = _with_index_entries(table, entries)
+ rows = _assert_parity(table)
+ assert {row['partition']: row['deleted_record_count'] for row in rows} == {
+ 'p=known': 6, 'p=unknown': None, 'p=zero': 0}
+
+
+def _with_index_entries(table, entries):
+ name = IndexManifestFile(table).write(entries)
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ path = table.snapshot_manager().snapshot_dir + '/snapshot-' +
str(snapshot.id)
+ with table.file_io.new_input_stream(path) as stream:
+ data = json.loads(stream.read())
+ data['indexManifest'] = name
+ table.file_io.overwrite_file_utf8(path, json.dumps(data))
+ return
table._copy_with_snapshot(table.snapshot_manager().get_latest_snapshot())
+
+
[email protected]('typ,value,canonical,legacy', [
+ (pa.string(), 'MISSING', None, True),
+ (pa.string(), 'MISSING', None, False),
+ (pa.string(), '\x1c', None, True),
+ (pa.string(), '\x1c', None, False),
+ # Java byte[].toString has no stable legacy spec; exercise its SQL cast.
+ (pa.binary(), b'\xed\xa0\x80', '\ufffd'.encode('utf-8'), False),
+ (pa.time32('ms'), time(12, 34, 56, 123000), time(12, 34, 56), True),
+ (pa.time32('ms'), time(12, 34, 56, 123000), time(12, 34, 56), False),
+])
+def test_deletion_lookup_uses_java_partition_spec_round_trip(rest_catalog,
typ, value, canonical, legacy):
+ catalog, _ = rest_catalog
+ schema = pa.schema([('id', pa.int32()), ('p', typ)])
+ table = _create(catalog, schema, ['p'], {
+ 'deletion-vectors.enabled': 'true', 'partition.default-name':
'MISSING',
+ 'partition.legacy-name': str(legacy).lower()})
+ _append(table, [{'id': i, 'p': value} for i in range(3)]
+ + [{'id': i + 3, 'p': canonical} for i in range(3)], schema)
+ entries = []
+ for partition, count in [(value, 1), (canonical, 3)]:
+ ranges = {'file': DeletionVectorMeta('file', 0, 1, count)}
+ entries.append(IndexManifestEntry(
+ 0, GenericRow([partition], table.partition_keys_fields), 0,
+ IndexFileMeta('DELETION_VECTORS', str(count), 1, 1,
dv_ranges=ranges)))
+ table = _with_index_entries(table, entries)
+ rows = _assert_parity(table)
+ assert len(rows) == 2
+ # A legacy TIME spec is an integer millisecond count, so Java keeps its
+ # sub-second digits even when the declared TIME precision is zero.
+ expected = [1, 3] if legacy and pa.types.is_time(typ) else [3, 3]
+ assert sorted(row['deleted_record_count'] for row in rows) == expected
+
+
+def test_default_legacy_time_partition_keeps_deletion_counts(rest_catalog):
+ from pypaimon.tests.deletion_vector_path_test import _delete
+ catalog, _ = rest_catalog
+ schema = pa.schema([('id', pa.int32()), ('p', pa.time32('ms'))])
+ table = _create(catalog, schema, ['p'], {
+ 'deletion-vectors.enabled': 'true', 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true'})
+ _append(table, [{'id': i, 'p': time(12, 34, 56, 120000)} for i in
range(2)], schema)
+ _delete(table, [0])
+ rows = _assert_parity(table)
+ assert [(row['partition'], row['deleted_record_count']) for row in rows]
== [('p=12:34:56', 1)]
+
+
[email protected]('legacy,instant,expected,deleted', [
+ (True, '2024-01-01T00:00:00+00:00', '2024-01-01 00:00:00.000', 0),
+ (False, '2024-01-01T00:00:00+00:00', '2023-12-31 19:00:00.000', 1),
+ (True, '2024-11-03T06:30:00+00:00', '2024-11-03 06:30:00.000', 0),
+ (False, '2024-11-03T06:30:00+00:00', '2024-11-03 01:30:00.000', 0),
+])
+def test_temporal_ltz_partition_specs_match_java_in_new_york(rest_catalog,
legacy, instant, expected, deleted):
+ from pypaimon.tests.deletion_vector_path_test import _delete
+ # Native caches the process system zone. CI runs this golden case in a
+ # separate process initialized with New York, as for temporal cast tests.
+ if os.environ.get('TZ') != 'America/New_York':
+ pytest.skip('Java golden values require TZ=America/New_York')
+ catalog, _ = rest_catalog
+ schema = pa.schema([('id', pa.int32()), ('p', pa.timestamp('ms',
tz='UTC'))])
+ table = _create(catalog, schema, ['p'], {
+ 'deletion-vectors.enabled': 'true', 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true', 'partition.legacy-name':
str(legacy).lower()})
+ _append(table, [{'id': i, 'p': datetime.fromisoformat(instant)} for i in
range(2)], schema)
+ _delete(table, [0])
+ rows = _assert_parity(table)
+ assert [(row['partition'], row['deleted_record_count']) for row in rows]
== [('p=' + expected, deleted)]
+
+
[email protected]('native', [False, True])
+def test_partition_split_is_captured_before_later_commits(rest_catalog,
native):
+ catalog, _ = rest_catalog
+ schema = pa.schema([('id', pa.int32()), ('p', pa.string())])
+ table = _create(catalog, schema, ['p'])
+ _append(table, [{'id': 1, 'p': 'old'}], schema)
+ system = _system(table, 'partitions', native)
+ builder = system.new_read_builder()
+ splits = builder.new_scan().plan().splits()
+ _append(table, [{'id': 2, 'p': 'new'}], schema)
+ rows = builder.new_read().to_arrow(splits).to_pylist()
+ assert [row['partition'] for row in rows] == ['p=old']
+ assert len(_assert_parity(table)) == 2
diff --git a/paimon-python/pypaimon/tests/native_system_table_test.py
b/paimon-python/pypaimon/tests/native_system_table_test.py
index 7af10c298d..7d501ceae9 100644
--- a/paimon-python/pypaimon/tests/native_system_table_test.py
+++ b/paimon-python/pypaimon/tests/native_system_table_test.py
@@ -125,7 +125,7 @@ def test_projection_filter_limit_and_pickled_split(source):
assert builder.new_read().to_arrow(splits).to_pylist() == [{'snapshot_id':
1}, {'snapshot_id': 2}]
[email protected]('name', ['files', 'file_key_ranges', 'buckets',
'manifests', 'table_indexes'])
[email protected]('name', ['files', 'file_key_ranges', 'buckets',
'manifests', 'table_indexes', 'partitions'])
@pytest.mark.parametrize('view', ['first', 'empty', 'tag'])
def test_captured_metadata_view(source, name, view):
catalog, table, _ = source
@@ -191,8 +191,7 @@ def
test_native_reader_preserves_schema_on_empty_and_rejects_catalog_tables(sour
assert result.schema.names[16] == 'deleteRowCount'
assert result.num_rows == 0
assert [field.name() for field in read.read_type()] == result.schema.names
- with pytest.raises(Exception, match='requires a catalog|unknown'):
- native.new_system_read('partitions')
+ assert native.new_system_read('partitions').read().num_rows == 0
with pytest.raises(Exception, match='requires a catalog|unknown'):
native.new_system_read('unknown')
diff --git a/paimon-python/pypaimon/tests/system/native_system_table_test.py
b/paimon-python/pypaimon/tests/system/native_system_table_test.py
index 4383b9ae76..60dc35eedf 100644
--- a/paimon-python/pypaimon/tests/system/native_system_table_test.py
+++ b/paimon-python/pypaimon/tests/system/native_system_table_test.py
@@ -34,9 +34,9 @@ def _table(name='files', enabled=True):
row_type=lambda: RowType(False, [DataField(0, 'id', AtomicType('INT',
nullable=False))]))
[email protected]('reason', ['disabled', 'old-binding', 'custom-io',
'catalog-required'])
[email protected]('reason', ['disabled', 'old-binding', 'custom-io',
'unknown-table'])
def test_unsupported_metadata_reads_do_not_create_native_tables(reason):
- table = _table('partitions' if reason == 'catalog-required' else 'files',
enabled=reason != 'disabled')
+ table = _table('unknown' if reason == 'unknown-table' else 'files',
enabled=reason != 'disabled')
with
patch('pypaimon.table.system.native_system_table.native_method_available',
return_value=reason != 'old-binding'):
with
patch('pypaimon.table.system.native_system_table._rest_catalog_supported',
diff --git a/paimon-python/pypaimon/tests/system/partitions_table_test.py
b/paimon-python/pypaimon/tests/system/partitions_table_test.py
index 108cc6d358..ac3177fe22 100644
--- a/paimon-python/pypaimon/tests/system/partitions_table_test.py
+++ b/paimon-python/pypaimon/tests/system/partitions_table_test.py
@@ -124,7 +124,7 @@ class PartitionsTableTest(unittest.TestCase):
for count in arrow_table.column("file_count").to_pylist():
self.assertGreaterEqual(count, 1)
for total_buckets in arrow_table.column("total_buckets").to_pylist():
- self.assertGreaterEqual(total_buckets, 1)
+ self.assertEqual(total_buckets, -1)
# FileSystem catalog does not maintain a "done" flag.
for done in arrow_table.column("done").to_pylist():
self.assertFalse(done)
@@ -192,6 +192,29 @@ class PartitionsTableTest(unittest.TestCase):
self.assertEqual(3, deleted_counts["dt=2024-01-01"])
self.assertIsNone(deleted_counts["dt=2024-01-02"])
+ def test_creation_times_use_java_gap_and_overlap_resolution(self):
+ import os
+ import time
+ from pypaimon.table.system.system_table import metadata_epoch_millis,
metadata_local_millis
+ original = os.environ.get('TZ')
+ try:
+ os.environ['TZ'] = 'America/New_York'
+ time.tzset()
+ # 02:55 lies in the spring gap: Java shifts to 03:55 EDT, 07:55
UTC.
+ gap_wall = 1710039300123
+ self.assertEqual(1710057300123, metadata_epoch_millis(gap_wall))
+ self.assertEqual(1710042900123,
metadata_local_millis(metadata_epoch_millis(gap_wall)))
+ # 01:30 occurs twice in the fall overlap. Java chooses 05:30 UTC
(EDT).
+ overlap_wall = 1730597400456
+ self.assertEqual(1730611800456,
metadata_epoch_millis(overlap_wall))
+ self.assertEqual(overlap_wall,
metadata_local_millis(metadata_epoch_millis(overlap_wall)))
+ finally:
+ if original is None:
+ os.environ.pop('TZ', None)
+ else:
+ os.environ['TZ'] = original
+ time.tzset()
+
if __name__ == "__main__":
unittest.main()
