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 5d019f5a6d [python] Read REST system tables with Native SystemRead
(#10604)
5d019f5a6d is described below
commit 5d019f5a6d94bff0187a96c94c9952b8e824bbde
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Oct 16 15:19:49 2026 +0800
[python] Read REST system tables with Native SystemRead (#10604)
---
.../pypaimon/table/system/branches_table.py | 4 +-
.../pypaimon/table/system/buckets_table.py | 32 +-
.../pypaimon/table/system/file_key_ranges_table.py | 18 +-
paimon-python/pypaimon/table/system/files_table.py | 223 ++++++-----
.../pypaimon/table/system/manifests_table.py | 4 +-
.../pypaimon/table/system/native_system_table.py | 46 +++
.../pypaimon/table/system/schemas_table.py | 4 +-
.../pypaimon/table/system/snapshots_table.py | 4 +-
.../pypaimon/table/system/system_table.py | 17 +
.../pypaimon/table/system/system_table_read.py | 2 +-
.../pypaimon/table/system/system_table_scan.py | 5 +-
.../pypaimon/table/system/table_indexes_table.py | 9 +-
paimon-python/pypaimon/table/system/tags_table.py | 4 +-
.../pypaimon/tests/native_merge_into_test.py | 5 +-
.../pypaimon/tests/native_system_table_test.py | 406 +++++++++++++++++++++
.../pypaimon/tests/system/branches_table_test.py | 5 +-
.../pypaimon/tests/system/buckets_table_test.py | 2 +-
.../pypaimon/tests/system/files_table_test.py | 114 +++++-
.../tests/system/native_system_table_test.py | 68 ++++
.../pypaimon/tests/system/snapshots_table_test.py | 3 +-
.../pypaimon/tests/system/tags_table_test.py | 3 +-
21 files changed, 813 insertions(+), 165 deletions(-)
diff --git a/paimon-python/pypaimon/table/system/branches_table.py
b/paimon-python/pypaimon/table/system/branches_table.py
index 227e30ba0b..78ef2694b6 100644
--- a/paimon-python/pypaimon/table/system/branches_table.py
+++ b/paimon-python/pypaimon/table/system/branches_table.py
@@ -23,7 +23,7 @@ import pyarrow
from pypaimon.branch.branch_manager import BranchManager
from pypaimon.schema.data_types import AtomicType, DataField, RowType
-from pypaimon.table.system.system_table import SystemTable
+from pypaimon.table.system.system_table import SystemTable,
metadata_local_millis
TABLE_TYPE = RowType(False, [
@@ -59,7 +59,7 @@ class BranchesTable(SystemTable):
# store cannot return an mtime (some remote object stores
# via PyArrowFileIO) fall back to epoch 0 so the schema
# contract holds.
- create_times.append(0 if ms is None else int(ms))
+ create_times.append(metadata_local_millis(0 if ms is None else ms))
return pyarrow.table({
"branch_name": pyarrow.array(names, type=pyarrow.string()),
"create_time": pyarrow.array(create_times, type=_TIMESTAMP_TYPE),
diff --git a/paimon-python/pypaimon/table/system/buckets_table.py
b/paimon-python/pypaimon/table/system/buckets_table.py
index ddd5e8c6c6..f3ae4f434e 100644
--- a/paimon-python/pypaimon/table/system/buckets_table.py
+++ b/paimon-python/pypaimon/table/system/buckets_table.py
@@ -24,7 +24,8 @@ import pyarrow
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_partition
+from pypaimon.table.system.system_table import SystemTable, metadata_snapshot
TABLE_TYPE = RowType(False, [
@@ -57,7 +58,7 @@ class BucketsTable(SystemTable):
return ["partition", "bucket"]
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()
@@ -80,11 +81,8 @@ class BucketsTable(SystemTable):
stats = bucket_map.get(key)
if stats is None:
- render_items = tuple(
- (name, str(val) if val is not _NULL else None)
- for name, val in raw_key)
stats = {
- "render_items": render_items,
+ "partition": _render_partition(entry.partition),
"bucket": bucket_id,
"record_count": 0,
"file_size_in_bytes": 0,
@@ -96,7 +94,7 @@ class BucketsTable(SystemTable):
stats["record_count"] += int(entry.file.row_count)
stats["file_size_in_bytes"] += int(entry.file.file_size)
stats["file_count"] += 1
- ct_ms = entry.file.creation_time_epoch_millis()
+ ct_ms = None if entry.file.creation_time is None else
entry.file.creation_time.get_millisecond()
if ct_ms is not None:
if (stats["last_update_time"] is None
or ct_ms > stats["last_update_time"]):
@@ -105,7 +103,7 @@ class BucketsTable(SystemTable):
sorted_keys = sorted(
bucket_map.keys(),
key=lambda k: (
- _render_partition(bucket_map[k]["render_items"]) or "",
+ bucket_map[k]["partition"] or "",
k[1]))
partition_strings: List[Optional[str]] = []
@@ -117,7 +115,7 @@ class BucketsTable(SystemTable):
for key in sorted_keys:
stats = bucket_map[key]
- partition_strings.append(_render_partition(stats["render_items"]))
+ partition_strings.append(stats["partition"])
buckets.append(stats["bucket"])
record_counts.append(stats["record_count"])
file_sizes.append(stats["file_size_in_bytes"])
@@ -147,19 +145,3 @@ class BucketsTable(SystemTable):
"file_count": pyarrow.array([], type=pyarrow.int64()),
"last_update_time": pyarrow.array([], type=_TIMESTAMP_TYPE),
})
-
-
-def _render_partition(spec_items) -> Optional[str]:
- """Render a partition spec as ``pt=v/pt2=v2`` or None when empty.
-
- Null partition values are rendered as ``__NULL__`` to distinguish them
- from the literal string ``"None"``. A partition whose value is
- literally ``"__NULL__"`` will produce the same rendered string —
- aggregation keys are still distinct, but the displayed partition
- column will collide. This is a display-only limitation.
- """
- if not spec_items:
- return None
- return "/".join(
- "{}={}".format(name, "__NULL__" if value is None else value)
- for name, value in spec_items)
diff --git a/paimon-python/pypaimon/table/system/file_key_ranges_table.py
b/paimon-python/pypaimon/table/system/file_key_ranges_table.py
index 290caae252..d6df0307c8 100644
--- a/paimon-python/pypaimon/table/system/file_key_ranges_table.py
+++ b/paimon-python/pypaimon/table/system/file_key_ranges_table.py
@@ -25,12 +25,11 @@ 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.files_table import (
- _render_key,
+ _data_file_format, _render_key,
_render_partition,
- _stringify_path,
+ _metadata_file_path,
)
-from pypaimon.table.system.system_table import SystemTable
-from pypaimon.utils.file_store_path_factory import canonical_data_file_path
+from pypaimon.table.system.system_table import SystemTable, metadata_snapshot
TABLE_TYPE = RowType(False, [
@@ -61,7 +60,7 @@ class FileKeyRangesTable(SystemTable):
return ["file_path"]
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 _empty_table()
@@ -71,7 +70,6 @@ class FileKeyRangesTable(SystemTable):
entries = manifest_file_manager.read_entries_parallel(
manifest_files, drop_stats=False)
- file_format = self.base_table.options.file_format()
rows = {
"partition": [],
"bucket": [],
@@ -90,12 +88,8 @@ class FileKeyRangesTable(SystemTable):
meta = entry.file
rows["partition"].append(_render_partition(entry.partition))
rows["bucket"].append(int(entry.bucket))
- rows["file_path"].append(_stringify_path(
- meta.external_path
- or meta.file_path
- or canonical_data_file_path(
- self.base_table, tuple(entry.partition.values),
int(entry.bucket), meta.file_name)))
- rows["file_format"].append(file_format)
+ rows["file_path"].append(_metadata_file_path(self.base_table,
entry))
+ rows["file_format"].append(_data_file_format(meta.file_name))
rows["schema_id"].append(int(meta.schema_id))
rows["level"].append(int(meta.level))
rows["record_count"].append(int(meta.row_count))
diff --git a/paimon-python/pypaimon/table/system/files_table.py
b/paimon-python/pypaimon/table/system/files_table.py
index fca9ab6d84..24e184d241 100644
--- a/paimon-python/pypaimon/table/system/files_table.py
+++ b/paimon-python/pypaimon/table/system/files_table.py
@@ -17,17 +17,15 @@
"""The ``$files`` system table — per-data-file detail of the latest
snapshot."""
-import json
from typing import Any, List, Optional
import pyarrow
from pypaimon.manifest.manifest_file_manager import ManifestFileManager
from pypaimon.manifest.manifest_list_manager import ManifestListManager
-from pypaimon.manifest.simple_stats_evolutions import SimpleStatsEvolutions
from pypaimon.schema.data_types import (ArrayType, AtomicType, DataField,
RowType)
-from pypaimon.table.system.system_table import SystemTable
+from pypaimon.table.system.system_table import SystemTable, metadata_snapshot
TABLE_TYPE = RowType(False, [
@@ -64,59 +62,103 @@ _TIMESTAMP_TYPE = pyarrow.timestamp("ms")
_WRITE_COLS_TYPE = pyarrow.list_(pyarrow.string())
-def _to_json(obj: Any) -> str:
- return json.dumps(obj, separators=(',', ':'), ensure_ascii=False,
- default=str)
-
-
-def _stringify_path(value: Any) -> str:
+def _render_value(value, data_type=None) -> str:
if value is None:
- return ""
- if isinstance(value, bytes):
- try:
- return value.decode("utf-8")
- except UnicodeDecodeError:
- return value.hex()
+ return "null"
+ if data_type is not None:
+ from pypaimon.casting.row_to_string import cast_value_to_string
+ from pypaimon.casting.numeric_utils import java_floating_string
+ type_name = str(getattr(data_type, 'type', data_type)).upper()
+ if type_name in ('FLOAT', 'REAL', 'DOUBLE'):
+ return java_floating_string(value, type_name != 'DOUBLE')
+ if type_name in ('BYTES', 'BINARY', 'VARBINARY') or
type_name.startswith(('BINARY(', 'VARBINARY(')):
+ from pypaimon.utils.file_store_path_factory import
_decode_java_utf8
+ return _decode_java_utf8(value)
+ if 'TIMESTAMP_LTZ' in type_name or 'WITH LOCAL TIME ZONE' in type_name:
+ from pypaimon.casting.temporal_string import format_timestamp_parts
+ from pypaimon.casting.temporal_utils import _epoch_day_to_date,
_local_offset
+ from pypaimon.table.row.generic_row import (
+ _datetime_to_millis_and_nanos, _normalize_ltz,
_parse_type_precision_scale)
+ # BinaryRow timestamps need not fit Arrow's i64 nanosecond range.
+ value = _normalize_ltz(value)
+ millis, nanos = _datetime_to_millis_and_nanos(value)
+ seconds = millis // 1000
+ day, second_of_day = divmod(seconds + _local_offset(seconds),
86400)
+ year, month, day = _epoch_day_to_date(day)
+ hour, second_of_day = divmod(second_of_day, 3600)
+ minute, second = divmod(second_of_day, 60)
+ nanos += millis % 1000 * 1000000 + getattr(value, 'nanosecond', 0)
+ precision, _ = _parse_type_precision_scale(data_type)
+ return format_timestamp_parts(year, month, day, hour, minute,
second, nanos, precision)
+ return cast_value_to_string(value, data_type)
+ if isinstance(value, bool):
+ return 'true' if value else 'false'
return str(value)
-def _to_python(value: Any) -> Any:
- """Render an internal-row cell value into a JSON-safe primitive."""
+def _render_object(value, data_type=None) -> str:
+ # Java InternalRowUtils.get returns integer DATE/TIME and a Timestamp
+ # whose toString is ISO LocalDateTime, including for LTZ key/stats cells.
if value is None:
- return None
- if isinstance(value, bytes):
- try:
- return value.decode("utf-8")
- except UnicodeDecodeError:
- return value.hex()
- if hasattr(value, "isoformat"):
- try:
- return value.isoformat()
- except Exception:
- return str(value)
- if isinstance(value, (int, float, bool, str, list, dict)):
- return value
- return str(value)
+ return 'null'
+ type_name = str(getattr(data_type, 'type', data_type)).upper()
+ if type_name == 'DATE':
+ from datetime import date
+ return str((value - date(1970, 1, 1)).days)
+ if type_name.startswith('TIME') and not type_name.startswith('TIMESTAMP'):
+ return str(((value.hour * 60 + value.minute) * 60 + value.second) *
1000 + value.microsecond // 1000)
+ if type_name.startswith('TIMESTAMP'):
+ from datetime import timezone
+ if value.tzinfo is not None:
+ value = value.astimezone(timezone.utc).replace(tzinfo=None)
+ nanos = value.microsecond * 1000 + getattr(value, 'nanosecond', 0)
+ text = '{:04d}-{:02d}-{:02d}T{:02d}:{:02d}'.format(
+ value.year, value.month, value.day, value.hour, value.minute)
+ if value.second or nanos:
+ text += ':{:02d}'.format(value.second)
+ if nanos:
+ precision = 3 if nanos % 1000000 == 0 else 6 if nanos % 1000 == 0
else 9
+ text += '.' + '{:09d}'.format(nanos)[:precision]
+ return text
+ return _render_value(value, data_type)
def _render_key(row) -> Optional[str]:
- if row is None:
- return None
- values = getattr(row, "values", None)
- if not values:
+ if row is None or not len(row):
return None
- return _to_json([_to_python(v) for v in values])
+ fields = getattr(row, 'fields', [])
+ return '[' + ', '.join(_render_object(row.get_field(i), fields[i].type)
for i in range(len(row))) + ']'
def _render_partition(partition_row) -> Optional[str]:
if partition_row is None:
return None
- fields = getattr(partition_row, "fields", None) or []
- values = getattr(partition_row, "values", None) or []
- if not fields:
- return None
- return "/".join("{}={}".format(field.name, value)
- for field, value in zip(fields, values))
+ fields = getattr(partition_row, 'fields', [])
+ return '{' + ', '.join(_render_value(partition_row.get_field(i),
field.type)
+ for i, field in enumerate(fields)) + '}'
+
+
+def _metadata_file_path(table, entry):
+ if entry.file.external_path is not None:
+ return entry.file.external_path
+ # Display Java's logical bucket path; the IO path may have decoded URI
+ # escapes or dropped file: to reach a literal escaped local directory.
+ from pypaimon.utils.path import resolve_path
+ relative_bucket = table.path_factory().relative_bucket_path(
+ tuple(entry.partition.values), int(entry.bucket),
canonical_partition=True)
+ bucket = resolve_path(str(table.table_path), relative_bucket)
+ return bucket.rstrip('/') + '/' + entry.file.file_name
+
+
+def _data_file_format(file_name):
+ parts = file_name.rsplit('.', 2)
+ if len(parts) < 2:
+ raise ValueError('{} is not a legal file name'.format(file_name))
+ if parts[-1].lower() in ('gz', 'bz2', 'deflate', 'snappy', 'lz4', 'zst'):
+ if len(parts) < 3:
+ raise ValueError('{} is not a legal file name'.format(file_name))
+ return parts[-2]
+ return parts[-1]
def _row_values(row) -> List[Any]:
@@ -129,23 +171,44 @@ def _stats_fields(stats_row) -> List[str]:
return [field.name for field in stats_row.fields]
-def _render_stats_map(values: List[Any], columns: List[str]) -> str:
- pairs = {}
- n = min(len(columns), len(values) if values is not None else 0)
- for i in range(n):
- pairs[columns[i]] = _to_python(values[i])
- return _to_json(pairs)
-
-
-def _render_null_counts(null_counts: Optional[List[int]],
- columns: List[str]) -> str:
- pairs = {}
- if null_counts:
- n = min(len(columns), len(null_counts))
- for i in range(n):
- pairs[columns[i]] = (None if null_counts[i] is None
- else int(null_counts[i]))
- return _to_json(pairs)
+def _render_stats_map(values: List[Any], columns: List[str], fields=None) ->
str:
+ values = values or []
+ pairs = {name: _render_object(values[i] if i < len(values) else None,
+ fields[i].type if fields else None)
+ for i, name in enumerate(columns)}
+ return '{' + ', '.join(name + '=' + pairs[name] for name in sorted(pairs))
+ '}'
+
+
+def _render_null_counts(null_counts: Optional[List[int]], columns: List[str])
-> str:
+ return _render_stats_map(null_counts, columns)
+
+
+def _file_stats(table, meta):
+ from pypaimon.casting.schema_evolution import
cast_array_for_schema_evolution
+ from pypaimon.schema.data_types import PyarrowFieldParser
+ file_schema = table.schema_manager.get_schema(meta.schema_id)
+ physical = {field.id: field for field in
file_schema.data_file_fields(meta.write_cols)}
+ stats = meta.value_stats
+ names = _stats_fields(stats.min_values)
+ nulls, lower, upper = [], [], []
+ for field in table.fields:
+ source = physical.get(field.id)
+ if source is None:
+ nulls.append(meta.row_count)
+ lower.append(None)
+ upper.append(None)
+ continue
+ index = names.index(source.name) if source.name in names else None
+ nulls.append(stats.null_counts[index] if index is not None and
stats.null_counts is not None
+ and index < len(stats.null_counts) else None)
+ for row, output in ((stats.min_values, lower), (stats.max_values,
upper)):
+ value = row.get_field(index) if index is not None and index <
len(row) else None
+ if value is not None and source.type != field.type:
+ value = cast_array_for_schema_evolution(
+ pyarrow.array([value],
type=PyarrowFieldParser.from_paimon_type(source.type)),
+ PyarrowFieldParser.from_paimon_type(field.type),
source.type, field.type)[0].as_py()
+ output.append(value)
+ return nulls, lower, upper
class FilesTable(SystemTable):
@@ -161,7 +224,7 @@ class FilesTable(SystemTable):
return ["file_path"]
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()
@@ -171,11 +234,6 @@ class FilesTable(SystemTable):
entries = manifest_file_manager.read_entries_parallel(
manifest_files, drop_stats=False)
- file_format = self.base_table.options.file_format()
- stats_evolutions = SimpleStatsEvolutions(
- self._schema_fields, self.base_table.table_schema.id)
- stats_columns = [field.name for field in stats_evolutions.table_fields]
-
rows = {
"partition": [],
"bucket": [],
@@ -203,9 +261,8 @@ class FilesTable(SystemTable):
meta = entry.file
rows["partition"].append(_render_partition(entry.partition))
rows["bucket"].append(int(entry.bucket))
- rows["file_path"].append(_stringify_path(
- meta.file_path or meta.file_name))
- rows["file_format"].append(file_format)
+ rows["file_path"].append(_metadata_file_path(self.base_table,
entry))
+ rows["file_format"].append(_data_file_format(meta.file_name))
rows["schema_id"].append(int(meta.schema_id))
rows["level"].append(int(meta.level))
rows["record_count"].append(int(meta.row_count))
@@ -213,28 +270,22 @@ class FilesTable(SystemTable):
rows["min_key"].append(_render_key(meta.min_key))
rows["max_key"].append(_render_key(meta.max_key))
- # Evolve stats to the current schema by field id, so a dropped or
- # re-added column never inherits another column's stats.
- value_stats =
stats_evolutions.get_or_create(meta.schema_id).evolution(
- meta.value_stats, meta.row_count,
- _stats_fields(meta.value_stats.min_values))
- rows["null_value_counts"].append(
- _render_null_counts(value_stats.null_counts, stats_columns))
- rows["min_value_stats"].append(_render_stats_map(
- _row_values(value_stats.min_values),
- stats_columns))
- rows["max_value_stats"].append(_render_stats_map(
- _row_values(value_stats.max_values),
- stats_columns))
+ nulls, lower, upper = _file_stats(self.base_table, meta)
+ columns = [field.name for field in self.base_table.fields]
+ rows["null_value_counts"].append(_render_null_counts(nulls,
columns))
+ rows["min_value_stats"].append(_render_stats_map(lower, columns,
self.base_table.fields))
+ rows["max_value_stats"].append(_render_stats_map(upper, columns,
self.base_table.fields))
rows["min_sequence_number"].append(int(meta.min_sequence_number))
rows["max_sequence_number"].append(int(meta.max_sequence_number))
- rows["creation_time"].append(meta.creation_time_epoch_millis())
+ rows["creation_time"].append(
+ None if meta.creation_time is None else
meta.creation_time.get_millisecond())
rows["deleteRowCount"].append(
None if meta.delete_row_count is None
else int(meta.delete_row_count))
rows["file_source"].append(
- None if meta.file_source is None else str(meta.file_source))
+ None if meta.file_source is None else
+ {0: "APPEND", 1: "COMPACT"}.get(meta.file_source,
str(meta.file_source)))
rows["first_row_id"].append(
None if meta.first_row_id is None
else int(meta.first_row_id))
@@ -280,12 +331,6 @@ class FilesTable(SystemTable):
rows["write_cols"], type=_WRITE_COLS_TYPE),
})
- def _schema_fields(self, schema_id: int) -> List[DataField]:
- table_schema = self.base_table.table_schema
- if schema_id == table_schema.id:
- return table_schema.fields
- return self.base_table.schema_manager.get_schema(schema_id).fields
-
@staticmethod
def _empty_table() -> pyarrow.Table:
return pyarrow.table({
diff --git a/paimon-python/pypaimon/table/system/manifests_table.py
b/paimon-python/pypaimon/table/system/manifests_table.py
index a2bcb90877..785261c832 100644
--- a/paimon-python/pypaimon/table/system/manifests_table.py
+++ b/paimon-python/pypaimon/table/system/manifests_table.py
@@ -24,7 +24,7 @@ import pyarrow
from pypaimon.casting.row_to_string import cast_row_to_string
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.system_table import SystemTable, metadata_snapshot
TABLE_TYPE = RowType(False, [
@@ -53,7 +53,7 @@ class ManifestsTable(SystemTable):
return ["file_name"]
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()
diff --git a/paimon-python/pypaimon/table/system/native_system_table.py
b/paimon-python/pypaimon/table/system/native_system_table.py
new file mode 100644
index 0000000000..0123576b69
--- /dev/null
+++ b/paimon-python/pypaimon/table/system/native_system_table.py
@@ -0,0 +1,46 @@
+# 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.
+
+"""Materialize REST metadata through the Rust core system reader."""
+
+import pyarrow
+
+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',
+))
+
+
+def try_read_native_system_table(system_table):
+ base = system_table.base_table
+ name = system_table.system_table_name()
+ if (name not in _NATIVE_TABLES or not base.options.native_read_enabled()
+ or not native_method_available('Table', 'new_system_read')
+ or not _rest_catalog_supported(base)):
+ return None
+ # Capture the full result for the existing in-memory SystemSplit. Reading
+ # or pickling that split never recreates REST credentials or file IO.
+ batch = _native_table(base).new_system_read(name).read()
+ table = pyarrow.Table.from_batches([batch])
+ schema =
PyarrowFieldParser.from_paimon_schema(system_table.row_type().fields)
+ return table.cast(schema)
diff --git a/paimon-python/pypaimon/table/system/schemas_table.py
b/paimon-python/pypaimon/table/system/schemas_table.py
index 7b917be4e3..77f8fb597e 100644
--- a/paimon-python/pypaimon/table/system/schemas_table.py
+++ b/paimon-python/pypaimon/table/system/schemas_table.py
@@ -23,7 +23,7 @@ from typing import Any, List, Optional
import pyarrow
from pypaimon.schema.data_types import AtomicType, DataField, RowType
-from pypaimon.table.system.system_table import SystemTable
+from pypaimon.table.system.system_table import SystemTable,
metadata_local_millis
TABLE_TYPE = RowType(False, [
@@ -82,7 +82,7 @@ class SchemasTable(SystemTable):
primary_keys_jsons.append(_to_json(list(schema.primary_keys)))
options_jsons.append(_to_json(dict(schema.options)))
comments.append(schema.comment)
- update_times.append(int(schema.time_millis))
+ update_times.append(metadata_local_millis(schema.time_millis))
return pyarrow.table({
"schema_id": pyarrow.array(schema_ids, type=pyarrow.int64()),
diff --git a/paimon-python/pypaimon/table/system/snapshots_table.py
b/paimon-python/pypaimon/table/system/snapshots_table.py
index d32c36163c..ee4302ec6a 100644
--- a/paimon-python/pypaimon/table/system/snapshots_table.py
+++ b/paimon-python/pypaimon/table/system/snapshots_table.py
@@ -22,7 +22,7 @@ from typing import List, Optional
import pyarrow
from pypaimon.schema.data_types import AtomicType, DataField, RowType
-from pypaimon.table.system.system_table import SystemTable
+from pypaimon.table.system.system_table import SystemTable,
metadata_local_millis
TABLE_TYPE = RowType(False, [
@@ -85,7 +85,7 @@ class SnapshotsTable(SystemTable):
commit_users.append(snap.commit_user)
commit_identifiers.append(int(snap.commit_identifier))
commit_kinds.append(snap.commit_kind)
- commit_times.append(int(snap.time_millis))
+ commit_times.append(metadata_local_millis(snap.time_millis))
base_manifest_lists.append(snap.base_manifest_list)
delta_manifest_lists.append(snap.delta_manifest_list)
changelog_manifest_lists.append(snap.changelog_manifest_list)
diff --git a/paimon-python/pypaimon/table/system/system_table.py
b/paimon-python/pypaimon/table/system/system_table.py
index 7ace7f23f3..a4f2c11c82 100644
--- a/paimon-python/pypaimon/table/system/system_table.py
+++ b/paimon-python/pypaimon/table/system/system_table.py
@@ -41,6 +41,23 @@ if TYPE_CHECKING: # pragma: no cover - import for type
hints only
from pypaimon.table.system.system_table_scan import SystemTableScan
+def metadata_local_millis(millis):
+ """Java DateTimeUtils.toLocalDateTime for an epoch-millisecond instant."""
+ from pypaimon.casting.temporal_utils import _local_offset, _signed
+ return _signed(int(millis) + _local_offset(int(millis) // 1000) * 1000, 64)
+
+
+def metadata_snapshot(table):
+ """Use the table's captured view, including an explicitly empty
snapshot."""
+ if '_read_snapshot' in vars(table):
+ return table._read_snapshot
+ from pypaimon.common.options.options import Options
+ from pypaimon.snapshot.time_travel_util import TimeTravelUtil
+ snapshot = TimeTravelUtil.try_travel_to_snapshot(
+ Options(table.table_schema.options), table.tag_manager(),
table.snapshot_manager())
+ return snapshot if snapshot is not None else
table.snapshot_manager().get_latest_snapshot()
+
+
_READ_ONLY_MESSAGE = "System table is read-only"
diff --git a/paimon-python/pypaimon/table/system/system_table_read.py
b/paimon-python/pypaimon/table/system/system_table_read.py
index e33717227e..e7f4c81736 100644
--- a/paimon-python/pypaimon/table/system/system_table_read.py
+++ b/paimon-python/pypaimon/table/system/system_table_read.py
@@ -145,7 +145,7 @@ class SystemTableRead:
combined = pyarrow.concat_tables(slices)
if self.limit is not None and combined.num_rows > self.limit:
combined = combined.slice(0, self.limit)
- return combined
+ return
combined.cast(PyarrowFieldParser.from_paimon_schema(self.read_type))
def _reject_non_scalar_filter_fields(self, predicate: Predicate) -> None:
"""Reject value comparisons on nested system-table columns.
diff --git a/paimon-python/pypaimon/table/system/system_table_scan.py
b/paimon-python/pypaimon/table/system/system_table_scan.py
index 5bc43302a9..b6d1f64025 100644
--- a/paimon-python/pypaimon/table/system/system_table_scan.py
+++ b/paimon-python/pypaimon/table/system/system_table_scan.py
@@ -70,5 +70,8 @@ class SystemTableScan:
self.system_table = system_table
def plan(self) -> Plan:
- arrow_table = self.system_table._build_arrow_table()
+ from pypaimon.table.system.native_system_table import
try_read_native_system_table
+ arrow_table = try_read_native_system_table(self.system_table)
+ if arrow_table is None:
+ arrow_table = self.system_table._build_arrow_table()
return Plan(_splits=[SystemSplit(arrow_table)])
diff --git a/paimon-python/pypaimon/table/system/table_indexes_table.py
b/paimon-python/pypaimon/table/system/table_indexes_table.py
index 4bf2103f0b..f50bf2cf39 100644
--- a/paimon-python/pypaimon/table/system/table_indexes_table.py
+++ b/paimon-python/pypaimon/table/system/table_indexes_table.py
@@ -25,7 +25,7 @@ from pypaimon.index.index_file_handler import IndexFileHandler
from pypaimon.schema.data_types import (ArrayType, AtomicType, DataField,
RowType)
from pypaimon.table.system.files_table import _render_partition
-from pypaimon.table.system.system_table import SystemTable
+from pypaimon.table.system.system_table import SystemTable, metadata_snapshot
_DV_RANGE_ROW_TYPE = RowType(False, [
@@ -73,8 +73,9 @@ class TableIndexesTable(SystemTable):
return ["file_name"]
def _build_arrow_table(self) -> pyarrow.Table:
- snapshot = self.base_table.snapshot_manager().get_latest_snapshot()
- entries = IndexFileHandler(self.base_table).scan(snapshot)
+ snapshot = metadata_snapshot(self.base_table)
+ # scan(None) means latest; a resolved empty view must stay empty.
+ entries = [] if snapshot is None else
IndexFileHandler(self.base_table).scan(snapshot)
rows = {
"partition": [],
@@ -139,7 +140,7 @@ class TableIndexesTable(SystemTable):
def _render_dv_ranges(dv_ranges) -> Optional[List[Dict[str, object]]]:
- if not dv_ranges:
+ if dv_ranges is None:
return None
rendered = []
for meta in dv_ranges.values():
diff --git a/paimon-python/pypaimon/table/system/tags_table.py
b/paimon-python/pypaimon/table/system/tags_table.py
index dc9bc18636..cc9bd23381 100644
--- a/paimon-python/pypaimon/table/system/tags_table.py
+++ b/paimon-python/pypaimon/table/system/tags_table.py
@@ -23,7 +23,7 @@ import pyarrow
from pypaimon.common.time_utils import duration_to_iso8601,
local_datetime_to_millis
from pypaimon.schema.data_types import AtomicType, DataField, RowType
-from pypaimon.table.system.system_table import SystemTable
+from pypaimon.table.system.system_table import SystemTable,
metadata_local_millis
TABLE_TYPE = RowType(False, [
@@ -70,7 +70,7 @@ class TagsTable(SystemTable):
names.append(name)
snapshot_ids.append(int(tag.id))
schema_ids.append(int(tag.schema_id))
- commit_times.append(int(tag.time_millis))
+ commit_times.append(metadata_local_millis(tag.time_millis))
record_counts.append(
None if tag.total_record_count is None
else int(tag.total_record_count))
diff --git a/paimon-python/pypaimon/tests/native_merge_into_test.py
b/paimon-python/pypaimon/tests/native_merge_into_test.py
index 738cf43cea..ac764e90b7 100644
--- a/paimon-python/pypaimon/tests/native_merge_into_test.py
+++ b/paimon-python/pypaimon/tests/native_merge_into_test.py
@@ -496,7 +496,10 @@ def
test_python_written_partition_source_uses_java_paths_for_native_merge(tmp_pa
assert _rows(source) == values
ranges = catalog.get_table('db.source$file_key_ranges').new_read_builder()
reported =
ranges.new_read().to_arrow(ranges.new_scan().plan().splits())['file_path'].to_pylist()
- assert reported and all(source.file_io.exists(path) for path in reported)
+ from pypaimon.utils.path import to_file_io_path
+ # System tables display Java logical paths; preserve literal escaped
+ # components when passing these strings into Python's URI-based FileIO.
+ assert reported and all(source.file_io.exists(to_file_io_path(path)) for
path in reported)
assert all('pa%23rt=' in path for path in reported)
builder = target.new_stream_write_builder() if stream else
target.new_batch_write_builder()
kwargs = dict(commit_identifier=57) if stream else {}
diff --git a/paimon-python/pypaimon/tests/native_system_table_test.py
b/paimon-python/pypaimon/tests/native_system_table_test.py
new file mode 100644
index 0000000000..7af10c298d
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_system_table_test.py
@@ -0,0 +1,406 @@
+# 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.
+
+"""REST system-table reads use Rust metadata providers with Java semantics."""
+
+import json
+import pickle
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import Schema
+from pypaimon.common.predicate_builder import PredicateBuilder
+from pypaimon.read.native_plan import _native_table, native_method_available
+from pypaimon.schema.data_types import AtomicType
+from pypaimon.schema.schema_change import SchemaChange
+from pypaimon.table.system.native_system_table import _NATIVE_TABLES
+from pypaimon.table.system.system_table_loader import load
+from pypaimon.tests import native_plan_rest_test
+
+rest_catalog = native_plan_rest_test.rest_catalog
+pytestmark = pytest.mark.skipif(
+ not native_method_available('Table', 'new_system_read'),
+ reason='Native system-table reader required')
+
+SCHEMA = pa.schema([('id', pa.int32()), ('value', pa.string()), ('pt',
pa.string())])
+
+
[email protected]
+def source(rest_catalog):
+ catalog, server = rest_catalog
+ catalog.create_table('default.metadata', Schema.from_pyarrow_schema(
+ SCHEMA, partition_keys=['pt'], options={
+ 'file.format': 'parquet', 'metadata.stats-mode': 'full',
+ 'write.native.enabled': 'false', 'commit.native.enabled': 'false',
+ }), False)
+ table = catalog.get_table('default.metadata')
+ return catalog, table, server
+
+
+def _append(table, rows):
+ builder = table.copy({'write.native.enabled':
'false'}).new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.Table.from_pylist(rows,
schema=SCHEMA).select(table.field_names))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+
+
+def _system(table, name, native):
+ return load(name, table.copy({'read.native.enabled': str(native).lower()}))
+
+
+def _read(table, projection=None, predicate=None, limit=None):
+ builder = table.new_read_builder()
+ if projection is not None:
+ builder.with_projection(projection)
+ if predicate is not None:
+ builder.with_filter(predicate)
+ if limit is not None:
+ builder.with_limit(limit)
+ return builder.new_read().to_arrow(builder.new_scan().plan().splits())
+
+
+def _canonical(table, name):
+ rows = table.to_pylist()
+ if name == 'schemas':
+ for row in rows:
+ for field in ('fields', 'partition_keys', 'primary_keys',
'options'):
+ row[field] = json.loads(row[field])
+ return sorted(json.dumps(row, sort_keys=True, default=str) for row in rows)
+
+
[email protected]('name', sorted(_NATIVE_TABLES))
[email protected]('empty', [True, False])
+def test_rest_system_table_parity_and_no_python_materialization(source, name,
empty):
+ catalog, table, _ = source
+ if not empty:
+ _append(table, [{'id': 1, 'value': 'old', 'pt': 'a'},
+ {'id': 2, 'value': None, 'pt': None}])
+ table.create_tag('first', snapshot_id=1, time_retained='1d')
+ catalog.create_tag(table.identifier, 'first', 1)
+ catalog.create_branch(table.identifier, 'dev', tag_name='first')
+ from pypaimon.consumer.consumer import Consumer
+ table.consumer_manager().reset_consumer('reader', Consumer(2))
+ expected = _read(_system(table, name, False))
+ native = _system(table, name, True)
+ with patch.object(type(native), '_build_arrow_table',
+ side_effect=AssertionError('Python metadata
materialized')):
+ actual = _read(native)
+ if not empty and name in ('tags', 'branches', 'consumers'):
+ assert actual.num_rows > 0
+ assert actual.schema == expected.schema
+ assert _canonical(actual, name) == _canonical(expected, name)
+
+
+def test_projection_filter_limit_and_pickled_split(source):
+ _, table, _ = source
+ _append(table, [{'id': 1, 'value': 'a', 'pt': 'old'}])
+ _append(table, [{'id': 2, 'value': 'b', 'pt': 'new'}])
+ system = _system(table, 'snapshots', True)
+ predicate =
PredicateBuilder(system.row_type().fields).greater_than('snapshot_id', 1)
+ assert _read(system, ['snapshot_id'], predicate, 1).to_pylist() ==
[{'snapshot_id': 2}]
+ assert _read(system, ['snapshot_id'], limit=0).num_rows == 0
+ builder = system.new_read_builder().with_projection(['snapshot_id'])
+ splits = pickle.loads(pickle.dumps(builder.new_scan().plan().splits()))
+ _append(table, [{'id': 3, 'value': 'c', 'pt': 'later'}])
+ 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]('view', ['first', 'empty', 'tag'])
+def test_captured_metadata_view(source, name, view):
+ catalog, table, _ = source
+ _append(table, [{'id': 1, 'value': 'a', 'pt': 'old'}])
+ first = table.snapshot_manager().get_latest_snapshot()
+ table.create_tag('first', snapshot_id=1)
+ catalog.create_tag(table.identifier, 'first', 1)
+ _append(table, [{'id': 2, 'value': 'b', 'pt': 'new'}])
+ if view == 'tag':
+ table = table.copy({'scan.tag-name': 'first'})
+ else:
+ table = table._copy_with_snapshot(first if view == 'first' else None)
+ expected = _read(_system(table, name, False))
+ actual = _read(_system(table, name, True))
+ assert _canonical(actual, name) == _canonical(expected, name)
+ if view == 'empty':
+ assert actual.num_rows == 0
+ elif name in ('files', 'file_key_ranges'):
+ assert actual.column('partition').to_pylist() == ['{old}']
+
+
+def test_options_use_latest_persisted_schema(source):
+ catalog, table, _ = source
+ _append(table, [{'id': 1, 'value': 'a', 'pt': 'a'}])
+ catalog.alter_table(table.identifier,
[SchemaChange.set_option('metadata.marker', 'persisted')], False)
+ table = table.copy({'scan.snapshot-id': '1', 'metadata.marker': 'dynamic'})
+ for native in (False, True):
+ result = _read(_system(table, 'options', native)).to_pylist()
+ options = {row['key']: row['value'] for row in result}
+ assert options['metadata.marker'] == 'persisted'
+ assert 'scan.snapshot-id' not in options
+
+
+def test_stats_follow_readded_column_ids(source):
+ catalog, table, _ = source
+ _append(table, [{'id': 1, 'value': 'old', 'pt': 'a'}])
+ catalog.alter_table(table.identifier, [SchemaChange.drop_column('value')],
False)
+ catalog.alter_table(table.identifier, [SchemaChange.add_column('value',
AtomicType('STRING'))], False)
+ table = catalog.get_table(table.identifier)
+ _append(table, [{'id': 2, 'value': 'new', 'pt': 'a'}])
+ for native in (False, True):
+ rows = sorted(_read(_system(table, 'files', native)).to_pylist(),
key=lambda row: row['schema_id'])
+ assert rows[0]['min_value_stats'] == '{id=1, pt=a, value=null}'
+ assert rows[0]['null_value_counts'] == '{id=0, pt=0, value=1}'
+ assert rows[1]['min_value_stats'] == '{id=2, pt=a, value=new}'
+
+
+def test_native_error_is_not_retried_in_python(source):
+ _, table, _ = source
+ system = _system(table, 'files', True)
+ with patch('pypaimon.table.system.native_system_table._native_table',
side_effect=RuntimeError('denied')):
+ with patch.object(type(system), '_build_arrow_table') as fallback:
+ with pytest.raises(RuntimeError, match='denied'):
+ _read(system)
+ fallback.assert_not_called()
+
+
+def
test_native_reader_preserves_schema_on_empty_and_rejects_catalog_tables(source):
+ _, table, _ = source
+ native = _native_table(table)
+ read = native.new_system_read('files')
+ result = read.read()
+ 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')
+ with pytest.raises(Exception, match='requires a catalog|unknown'):
+ native.new_system_read('unknown')
+
+
+def test_writer_version_is_loaded_from_snapshot_metadata(source):
+ _, table, _ = source
+ _append(table, [{'id': 1, 'value': 'a', 'pt': 'a'}])
+ path = table.snapshot_manager().snapshot_dir + '/snapshot-1'
+ with table.file_io.new_input_stream(path) as stream:
+ snapshot = json.loads(stream.read())
+ snapshot['writerVersion'] = 'Java test writer'
+ table.file_io.overwrite_file_utf8(path, json.dumps(snapshot))
+ assert _read(_system(table, 'snapshots', True),
['writer_version']).to_pylist() == [
+ {'writer_version': 'Java test writer'}]
+
+
[email protected]('name', ['files', 'file_key_ranges', 'buckets',
'snapshots', 'schemas', 'consumers'])
+def test_branch_namespace(source, name):
+ from pypaimon.common.identifier import Identifier
+ from pypaimon.consumer.consumer import Consumer
+ catalog, table, _ = source
+ _append(table, [{'id': 1, 'value': 'main', 'pt': 'a'}])
+ table.create_tag('first', snapshot_id=1)
+ catalog.create_tag(table.identifier, 'first', 1)
+ catalog.create_branch(table.identifier, 'dev', tag_name='first')
+ branch = catalog.get_table(Identifier('default', 'metadata', branch='dev'))
+ branch.consumer_manager().reset_consumer('branch-reader', Consumer(17))
+ table.consumer_manager().reset_consumer('main-reader', Consumer(91))
+ actual = _read(_system(branch, name, True))
+ expected = _read(_system(branch, name, False))
+ assert _canonical(actual, name) == _canonical(expected, name)
+ if name == 'consumers':
+ assert actual.to_pylist() == [{'consumer_id': 'branch-reader',
'next_snapshot_id': 17}]
+
+
+def test_vertical_file_statistics_use_physical_columns(rest_catalog):
+ catalog, _ = rest_catalog
+ schema = pa.schema([('id', pa.int32()), ('value', pa.int32())])
+ catalog.create_table('default.vertical',
Schema.from_pyarrow_schema(schema, options={
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'metadata.stats-mode': 'full', 'write.native.enabled': 'false',
+ 'commit.native.enabled': 'false', 'file.format': 'parquet',
+ }), False)
+ table = catalog.get_table('default.vertical')
+ builder = table.new_batch_write_builder()
+ left = builder.new_write().with_write_type(['id'])
+ right = builder.new_write().with_write_type(['value'])
+ commit = builder.new_commit()
+ try:
+ left.write_arrow(pa.table({'id': pa.array([1, 2], type=pa.int32())}))
+ right.write_arrow(pa.table({'value': pa.array([41, 42],
type=pa.int32())}))
+ commit.commit(left.prepare_commit() + right.prepare_commit())
+ finally:
+ left.close()
+ right.close()
+ commit.close()
+ actual = _read(_system(table, 'files', True))
+ expected = _read(_system(table, 'files', False))
+ assert _canonical(actual, 'files') == _canonical(expected, 'files')
+ stats = {tuple(row['write_cols']): row for row in actual.to_pylist()}
+ assert stats[('id',)]['min_value_stats'] == '{id=1, value=null}'
+ assert stats[('value',)]['min_value_stats'] == '{id=null, value=41}'
+ assert stats[('value',)]['null_value_counts'] == '{id=2, value=0}'
+
+
+def test_primary_key_ranges_are_java_arrays(rest_catalog):
+ catalog, _ = rest_catalog
+ catalog.create_table('default.keyed', Schema.from_pyarrow_schema(
+ SCHEMA, partition_keys=['pt'], primary_keys=['pt', 'id'], options={
+ 'bucket': '1', 'file.format': 'parquet', 'write.native.enabled':
'false',
+ 'commit.native.enabled': 'false'}), False)
+ table = catalog.get_table('default.keyed')
+ _append(table, [{'id': 2, 'value': 'b', 'pt': 'a'}, {'id': 1, 'value':
'a', 'pt': 'a'}])
+ for name in ('files', 'file_key_ranges'):
+ actual, expected = _read(_system(table, name, True)),
_read(_system(table, name, False))
+ assert _canonical(actual, name) == _canonical(expected, name)
+ assert actual.column('min_key').to_pylist() == ['[1]']
+ assert actual.column('max_key').to_pylist() == ['[2]']
+
+
+def test_nonempty_table_indexes(rest_catalog):
+ from pypaimon.tests.native_sorted_index_build_test import _create, _append
as append, _build, _commit
+ catalog, _ = rest_catalog
+ table = _create(catalog)
+ append(table, [{'id': 1, 'name': 'a', 'pt': 0}, {'id': 2, 'name': 'b',
'pt': 0}])
+ _commit(table, _build(table, 'btree'))
+ actual = _read(_system(table, 'table_indexes', True))
+ expected = _read(_system(table, 'table_indexes', False))
+ assert _canonical(actual, 'table_indexes') == _canonical(expected,
'table_indexes')
+ rows = actual.to_pylist()
+ assert rows and rows[0]['index_field_name'] == 'name'
+ # Populate an index before fixing the view to empty: otherwise both paths
+ # return no indexes even if the Python reader incorrectly reloads latest.
+ empty = table._copy_with_snapshot(None)
+ for native in (False, True):
+ result = _read(_system(empty, 'table_indexes', native))
+ assert result.num_rows == 0
+ assert result.schema == actual.schema
+
+
+def test_native_metadata_auth_guard(source):
+ _, table, _ = source
+ denied = _native_table(table.copy({'query-auth.enabled': 'true'}))
+ with pytest.raises(Exception, match='query-auth.enabled'):
+ denied.new_system_read('options')
+
+
[email protected]('keyed', [False, True])
+def test_metadata_temporal_cells_use_java_object_strings(rest_catalog, keyed):
+ from datetime import date, datetime, time, timezone
+ catalog, _ = rest_catalog
+ schema = pa.schema([('day', pa.date32()), ('clock', pa.time32('ms')),
+ ('ts', pa.timestamp('ms')), ('ltz', pa.timestamp('ms',
tz='UTC'))])
+ catalog.create_table('default.temporal', Schema.from_pyarrow_schema(
+ schema, primary_keys=['day', 'clock', 'ts'] if keyed else [], options={
+ 'bucket': '1' if keyed else '-1', 'file.format': 'parquet',
+ 'metadata.stats-mode': 'full', 'write.native.enabled': 'false',
+ 'commit.native.enabled': 'false'}), False)
+ table = catalog.get_table('default.temporal')
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.Table.from_pylist([{
+ 'day': date(2024, 1, 2), 'clock': time(1, 2, 3, 4000),
+ 'ts': datetime(2024, 1, 2, 0, 0, 0, 123000),
+ 'ltz': datetime(2024, 1, 2, tzinfo=timezone.utc),
+ }], schema=schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ actual = _read(_system(table, 'files', True))
+ expected = _read(_system(table, 'files', False))
+ assert _canonical(actual, 'files') == _canonical(expected, 'files')
+ row = actual.to_pylist()[0]
+ assert row['min_value_stats'] == (
+ '{clock=3723004, day=19724, ltz=2024-01-02T00:00,
ts=2024-01-02T00:00:00.123}')
+ if keyed:
+ ranges = _read(_system(table, 'file_key_ranges', True))
+ assert ranges.column('min_key').to_pylist() == ['[19724, 3723004,
2024-01-02T00:00:00.123]']
+
+
[email protected]('name', ['files', 'file_key_ranges'])
+def test_metadata_paths_keep_escaped_partition_components(source, name):
+ _, table, _ = source
+ _append(table, [{'id': 1, 'value': 'a', 'pt': 'a/b%=λ'}])
+ actual = _read(_system(table, name, True))
+ expected = _read(_system(table, name, False))
+ assert _canonical(actual, name) == _canonical(expected, name)
+ path = actual.column('file_path')[0].as_py()
+ assert '/pt=a%2Fb%25%3Dλ/bucket-0/' in path
+ from pypaimon.utils.path import to_file_io_path
+ assert table.file_io.exists(to_file_io_path(path))
+ assert actual.column('partition').to_pylist() == ['{a/b%=λ}']
+
+
+def
test_system_read_filters_before_projection_without_python_arrow_filter(source):
+ from pypaimon.read.native_plan import _predicate_to_native
+ _, table, _ = source
+ _append(table, [{'id': 1, 'value': 'a', 'pt': 'a'}])
+ _append(table, [{'id': 2, 'value': 'b', 'pt': 'b'}])
+ system = _system(table, 'snapshots', True)
+ predicate =
PredicateBuilder(system.row_type().fields).greater_than('snapshot_id', 1)
+ read =
_native_table(table).new_system_read('snapshots').with_projection(['commit_user'])
+ read.with_filter(_predicate_to_native(predicate)).with_limit(1)
+ with patch.object(type(predicate), 'to_arrow',
side_effect=AssertionError('Python Arrow filter ran')):
+ result = read.read()
+ assert result.num_rows == 1
+ assert result.schema.names == ['commit_user']
+ assert [field.name() for field in read.read_type()] == ['commit_user']
+ read.with_limit(0)
+ empty = read.read()
+ assert empty.num_rows == 0
+ assert empty.schema == result.schema
+
+
+def
test_captured_native_split_can_be_reused_with_another_read_configuration(source):
+ _, table, _ = source
+ _append(table, [{'id': 1, 'value': 'a', 'pt': 'a'}])
+ system = _system(table, 'snapshots', True)
+ builder = system.new_read_builder().with_projection(['snapshot_id'])
+ splits = pickle.loads(pickle.dumps(builder.new_scan().plan().splits()))
+ assert splits[0].row_count == 1
+ other =
system.new_read_builder().with_projection(['commit_user']).new_read()
+ assert other.to_arrow(splits).num_rows == 1
+
+
+def
test_captured_native_split_reads_without_recreating_rest_environment(source):
+ _, table, _ = source
+ _append(table, [{'id': 1, 'value': 'a', 'pt': 'a'}])
+ system = _system(table, 'snapshots', True)
+ splits =
pickle.loads(pickle.dumps(system.new_read_builder().new_scan().plan().splits()))
+ reader =
system.new_read_builder().with_projection(['snapshot_id']).new_read()
+ with patch('pypaimon.table.system.native_system_table._native_table',
+ side_effect=AssertionError('REST environment recreated')):
+ assert reader.to_arrow(splits).to_pylist() == [{'snapshot_id': 1}]
+
+
+def test_system_read_repeated_and_empty_projections(source):
+ _, table, _ = source
+ read = _native_table(table).new_system_read('options')
+ read.with_projection(['value', 'key', 'value'])
+ result = read.read()
+ assert result.num_rows > 0
+ assert result.schema.names == ['value', 'key', 'value']
+ assert result.column(0).to_pylist() == result.column(2).to_pylist()
+ read.with_projection([])
+ empty = read.read()
+ assert empty.num_columns == 0
+ assert empty.num_rows == result.num_rows
+ assert read.read_type() == []
diff --git a/paimon-python/pypaimon/tests/system/branches_table_test.py
b/paimon-python/pypaimon/tests/system/branches_table_test.py
index bfd8f9ef9d..cdfd1c676a 100644
--- a/paimon-python/pypaimon/tests/system/branches_table_test.py
+++ b/paimon-python/pypaimon/tests/system/branches_table_test.py
@@ -28,6 +28,7 @@ from pypaimon import CatalogFactory, Schema
from pypaimon.common.json_util import JSON
from pypaimon.schema.data_types import DataField
from pypaimon.snapshot.snapshot import Snapshot
+from pypaimon.table.system.system_table import metadata_local_millis
from pypaimon.table.system.branches_table import BranchesTable
@@ -113,8 +114,8 @@ class BranchesTableTest(unittest.TestCase):
self.assertIsInstance(ts, datetime.datetime)
ms = int(ts.replace(
tzinfo=datetime.timezone.utc).timestamp() * 1000)
- self.assertGreaterEqual(ms, before - 5000)
- self.assertLessEqual(ms, after + 5000)
+ self.assertGreaterEqual(ms, metadata_local_millis(before) - 5000)
+ self.assertLessEqual(ms, metadata_local_millis(after) + 5000)
if __name__ == "__main__":
diff --git a/paimon-python/pypaimon/tests/system/buckets_table_test.py
b/paimon-python/pypaimon/tests/system/buckets_table_test.py
index 57a352392a..73cfaea0c5 100644
--- a/paimon-python/pypaimon/tests/system/buckets_table_test.py
+++ b/paimon-python/pypaimon/tests/system/buckets_table_test.py
@@ -109,7 +109,7 @@ class BucketsTableTest(unittest.TestCase):
self.assertGreater(arrow_table.num_rows, 0)
partitions = arrow_table.column("partition").to_pylist()
- self.assertTrue(all(p in ("dt=2024-01-01", "dt=2024-01-02")
+ self.assertTrue(all(p in ("{2024-01-01}", "{2024-01-02}")
for p in partitions))
for size in arrow_table.column("file_size_in_bytes").to_pylist():
diff --git a/paimon-python/pypaimon/tests/system/files_table_test.py
b/paimon-python/pypaimon/tests/system/files_table_test.py
index 525ab5224d..df15a3a59c 100644
--- a/paimon-python/pypaimon/tests/system/files_table_test.py
+++ b/paimon-python/pypaimon/tests/system/files_table_test.py
@@ -17,18 +17,23 @@
"""End-to-end tests for the ``$files`` system table."""
-import json
import os
import shutil
import tempfile
+import time
import unittest
+from datetime import datetime, timezone
+from types import SimpleNamespace
import pyarrow as pa
+import pytest
from pypaimon import CatalogFactory, Schema
from pypaimon.schema.data_types import AtomicType, DataField
from pypaimon.schema.schema_change import SchemaChange
-from pypaimon.table.system.files_table import FilesTable, _row_values
+from pypaimon.table.row.generic_row import GenericRow, GenericRowDeserializer,
GenericRowSerializer
+from pypaimon.table.system.files_table import FilesTable, _metadata_file_path,
_render_partition, _row_values
+from pypaimon.utils.file_store_path_factory import FileStorePathFactory
def _read(table):
@@ -36,6 +41,20 @@ def _read(table):
return rb.new_read().to_arrow(rb.new_scan().plan().splits())
+def _stats_map(text):
+ # FilesTable exposes Java Map.toString, with sorted names and unquoted
cells.
+ assert text.startswith('{') and text.endswith('}')
+ pairs = {}
+ for item in text[1:-1].split(', ') if text != '{}' else []:
+ name, value = item.split('=', 1)
+ try:
+ value = int(value)
+ except ValueError:
+ value = None if value == 'null' else value
+ pairs[name] = value
+ return pairs
+
+
class FilesTableTest(unittest.TestCase):
def setUp(self):
@@ -114,10 +133,10 @@ class FilesTableTest(unittest.TestCase):
arrow_table = _read(self.catalog.get_table("db.t$files"))
self.assertGreater(arrow_table.num_rows, 0)
- # One row per file; partitions captured as `pt=value` strings.
+ # One row per file; partitions rendered with Java Row to STRING.
partitions = set(arrow_table.column("partition").to_pylist())
- self.assertIn("dt=2024-01-01", partitions)
- self.assertIn("dt=2024-01-02", partitions)
+ self.assertIn("{2024-01-01}", partitions)
+ self.assertIn("{2024-01-02}", partitions)
for path in arrow_table.column("file_path").to_pylist():
self.assertTrue(path)
@@ -126,17 +145,17 @@ class FilesTableTest(unittest.TestCase):
for size in arrow_table.column("file_size_in_bytes").to_pylist():
self.assertGreater(size, 0)
- # Stats columns are JSON dicts keyed by column name.
+ # Stats columns use Java Map.toString keyed by column name.
for value in arrow_table.column("null_value_counts").to_pylist():
self.assertIsInstance(value, str)
- decoded = json.loads(value)
+ decoded = _stats_map(value)
self.assertIsInstance(decoded, dict)
for value in arrow_table.column("min_value_stats").to_pylist():
self.assertIsInstance(value, str)
- self.assertIsInstance(json.loads(value), dict)
+ self.assertIsInstance(_stats_map(value), dict)
for value in arrow_table.column("max_value_stats").to_pylist():
self.assertIsInstance(value, str)
- self.assertIsInstance(json.loads(value), dict)
+ self.assertIsInstance(_stats_map(value), dict)
# Append-only tables expose empty min_key / max_key.
for value in arrow_table.column("min_key").to_pylist():
@@ -178,9 +197,9 @@ class FilesTableTest(unittest.TestCase):
arrow_table = _read(self.catalog.get_table("db.stats_t$files"))
self.assertGreater(arrow_table.num_rows, 0)
- mins = [json.loads(s) for s in
+ mins = [_stats_map(s) for s in
arrow_table.column("min_value_stats").to_pylist()]
- maxs = [json.loads(s) for s in
+ maxs = [_stats_map(s) for s in
arrow_table.column("max_value_stats").to_pylist()]
# Real per-column values, not an empty {} (the BinaryRow-lacks-values
bug).
self.assertTrue(any(m for m in mins),
@@ -227,17 +246,17 @@ class FilesTableTest(unittest.TestCase):
# The re-added ``v`` is a different column: the old file has no stats
# for it and all of its rows are null there.
self.assertEqual({"id": 1, "v": None},
- json.loads(old_file["min_value_stats"]))
+ _stats_map(old_file["min_value_stats"]))
self.assertEqual({"id": 1, "v": None},
- json.loads(old_file["max_value_stats"]))
+ _stats_map(old_file["max_value_stats"]))
self.assertEqual({"id": 0, "v": 1},
- json.loads(old_file["null_value_counts"]))
+ _stats_map(old_file["null_value_counts"]))
self.assertEqual({"id": 2, "v": 20},
- json.loads(new_file["min_value_stats"]))
+ _stats_map(new_file["min_value_stats"]))
self.assertEqual({"id": 2, "v": 20},
- json.loads(new_file["max_value_stats"]))
+ _stats_map(new_file["max_value_stats"]))
self.assertEqual({"id": 0, "v": 0},
- json.loads(new_file["null_value_counts"]))
+ _stats_map(new_file["null_value_counts"]))
def test_stats_decode_errors_are_not_swallowed(self):
class CorruptRow:
@@ -251,5 +270,66 @@ class FilesTableTest(unittest.TestCase):
_row_values(CorruptRow())
[email protected]('type_name', ['TIMESTAMP_LTZ(9)', 'TIMESTAMP(9) WITH
LOCAL TIME ZONE'])
[email protected]('value,expected', [
+ (datetime(2500, 1, 1, tzinfo=timezone.utc), {
+ 'UTC': '2500-01-01 00:00:00.000000000',
+ 'America/New_York': '2499-12-31 19:00:00.000000000',
+ 'Asia/Shanghai': '2500-01-01 08:00:00.000000000',
+ }),
+ (datetime(2500, 7, 1, microsecond=123456, tzinfo=timezone.utc), {
+ 'UTC': '2500-07-01 00:00:00.123456000',
+ 'America/New_York': '2500-06-30 20:00:00.123456000',
+ 'Asia/Shanghai': '2500-07-01 08:00:00.123456000',
+ }),
+ (datetime(1600, 1, 1, microsecond=123456, tzinfo=timezone.utc), {
+ 'UTC': '1600-01-01 00:00:00.123456000',
+ 'America/New_York': '1599-12-31 19:03:58.123456000',
+ 'Asia/Shanghai': '1600-01-01 08:05:43.123456000',
+ }),
+])
+def test_ltz_partition_strings_outside_arrow_nanosecond_range(type_name,
value, expected):
+ zone = os.environ.get('TZ', time.tzname[0])
+ if zone not in expected:
+ pytest.skip('Run Java timezone golden cases with TZ set to UTC, New
York or Shanghai')
+ fields = [DataField(0, 'pt', AtomicType(type_name))]
+ row = GenericRow([value], fields)
+ decoded =
GenericRowDeserializer.from_bytes(GenericRowSerializer.to_bytes(row), fields)
+ assert _render_partition(decoded) == '{' + expected[zone] + '}'
+
+
+def test_ltz_partition_strings_preserve_submicrosecond_fraction():
+ import pandas as pd
+ expected = {
+ 'UTC': '2024-01-01 00:00:00.123456789',
+ 'America/New_York': '2023-12-31 19:00:00.123456789',
+ 'Asia/Shanghai': '2024-01-01 08:00:00.123456789',
+ }
+ zone = os.environ.get('TZ', time.tzname[0])
+ if zone not in expected:
+ pytest.skip('Run Java timezone golden cases with TZ set to UTC, New
York or Shanghai')
+ row = GenericRow([pd.Timestamp('2024-01-01 00:00:00.123456789', tz='UTC')],
+ [DataField(0, 'pt', AtomicType('TIMESTAMP_LTZ(9)'))])
+ assert _render_partition(row) == '{' + expected[zone] + '}'
+
+
[email protected]('root,expected', [
+ ('file:/warehouse/t', 'file:/warehouse/t/'),
+ ('file:///warehouse/t', 'file:/warehouse/t/'),
+ ('s3://bucket/warehouse/t', 's3://bucket/warehouse/t/'),
+])
[email protected]('data_directory', [None, 'data'])
+def test_metadata_file_paths_keep_logical_scheme_and_escaped_partition(root,
expected, data_directory):
+ factory = FileStorePathFactory(
+ root=root, partition_keys=['pt'],
default_part_value='__DEFAULT_PARTITION__',
+ format_identifier='parquet', data_file_prefix='data-',
changelog_file_prefix='changelog-',
+ legacy_partition_name=False, file_suffix_include_compression=False,
file_compression='zstd',
+ data_file_path_directory=data_directory,
partition_types=[AtomicType('STRING')])
+ table = SimpleNamespace(table_path=root, path_factory=lambda: factory)
+ entry = SimpleNamespace(file=SimpleNamespace(external_path=None,
file_name='data.parquet'),
+ partition=GenericRow(['a/b'], [DataField(0, 'pt',
AtomicType('STRING'))]), bucket=0)
+ assert _metadata_file_path(table, entry) == (
+ expected + ('data/' if data_directory else '') +
'pt=a%2Fb/bucket-0/data.parquet')
+
if __name__ == "__main__":
unittest.main()
diff --git a/paimon-python/pypaimon/tests/system/native_system_table_test.py
b/paimon-python/pypaimon/tests/system/native_system_table_test.py
new file mode 100644
index 0000000000..4383b9ae76
--- /dev/null
+++ b/paimon-python/pypaimon/tests/system/native_system_table_test.py
@@ -0,0 +1,68 @@
+# 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.
+
+"""End-to-end tests for the ``$files`` system table."""
+
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon.schema.data_types import AtomicType, DataField, RowType
+from pypaimon.table.system.native_system_table import
try_read_native_system_table
+
+
+def _table(name='files', enabled=True):
+ return SimpleNamespace(
+
base_table=SimpleNamespace(options=SimpleNamespace(native_read_enabled=lambda:
enabled)),
+ system_table_name=lambda: name,
+ row_type=lambda: RowType(False, [DataField(0, 'id', AtomicType('INT',
nullable=False))]))
+
+
[email protected]('reason', ['disabled', 'old-binding', 'custom-io',
'catalog-required'])
+def test_unsupported_metadata_reads_do_not_create_native_tables(reason):
+ table = _table('partitions' if reason == 'catalog-required' 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',
+ return_value=reason != 'custom-io'):
+ with
patch('pypaimon.table.system.native_system_table._native_table') as create:
+ assert try_read_native_system_table(table) is None
+ create.assert_not_called()
+
+
+def _read_with_native_result(result=None, error=None):
+ with
patch('pypaimon.table.system.native_system_table.native_method_available',
return_value=True):
+ with
patch('pypaimon.table.system.native_system_table._rest_catalog_supported',
return_value=True):
+ with
patch('pypaimon.table.system.native_system_table._native_table') as create:
+ read = create.return_value.new_system_read.return_value.read
+ read.return_value = result
+ read.side_effect = error
+ return try_read_native_system_table(_table())
+
+
+def test_native_read_failure_is_propagated():
+ with pytest.raises(RuntimeError, match='metadata read failed'):
+ _read_with_native_result(error=RuntimeError('metadata read failed'))
+
+
+def test_native_empty_result_uses_declared_schema():
+ batch = pa.record_batch([pa.array([], type=pa.int32())], names=['id'])
+ result = _read_with_native_result(batch)
+ assert result.num_rows == 0
+ assert result.schema == pa.schema([pa.field('id', pa.int32(),
nullable=False)])
diff --git a/paimon-python/pypaimon/tests/system/snapshots_table_test.py
b/paimon-python/pypaimon/tests/system/snapshots_table_test.py
index aadb69dcaa..3cd4e01f03 100644
--- a/paimon-python/pypaimon/tests/system/snapshots_table_test.py
+++ b/paimon-python/pypaimon/tests/system/snapshots_table_test.py
@@ -27,6 +27,7 @@ from pypaimon import CatalogFactory, Schema
from pypaimon.common.json_util import JSON
from pypaimon.schema.data_types import DataField
from pypaimon.snapshot.snapshot import Snapshot
+from pypaimon.table.system.system_table import metadata_local_millis
from pypaimon.table.system.snapshots_table import SnapshotsTable
@@ -161,7 +162,7 @@ class SnapshotsTableTest(unittest.TestCase):
self.assertIsInstance(times[0], datetime.datetime)
as_ms = [int(t.replace(tzinfo=datetime.timezone.utc).timestamp() *
1000)
for t in times]
- self.assertEqual([_T0, _T1], as_ms)
+ self.assertEqual([metadata_local_millis(_T0),
metadata_local_millis(_T1)], as_ms)
if __name__ == "__main__":
diff --git a/paimon-python/pypaimon/tests/system/tags_table_test.py
b/paimon-python/pypaimon/tests/system/tags_table_test.py
index e801b656d8..290e9974e4 100644
--- a/paimon-python/pypaimon/tests/system/tags_table_test.py
+++ b/paimon-python/pypaimon/tests/system/tags_table_test.py
@@ -27,6 +27,7 @@ from pypaimon import CatalogFactory, Schema
from pypaimon.common.json_util import JSON
from pypaimon.schema.data_types import DataField
from pypaimon.snapshot.snapshot import Snapshot
+from pypaimon.table.system.system_table import metadata_local_millis
from pypaimon.table.system.tags_table import TagsTable
@@ -118,7 +119,7 @@ class TagsTableTest(unittest.TestCase):
self.assertIsInstance(ts, datetime.datetime)
ms = int(ts.replace(
tzinfo=datetime.timezone.utc).timestamp() * 1000)
- self.assertEqual(_TAG_COMMIT_MS, ms)
+ self.assertEqual(metadata_local_millis(_TAG_COMMIT_MS), ms)
# Tags created without a retention carry no create-time / TTL.
for value in arrow_table.column("create_time").to_pylist():
