diff --git a/paimon-python/pypaimon/table/system/branches_table.py b/paimon-python/pypaimon/table/system/branches_table.py index 227e30ba0bf7..78ef2694b654 100644 --- a/paimon-python/pypaimon/table/system/branches_table.py +++ b/paimon-python/pypaimon/table/system/branches_table.py @@ -23,7 +23,7 @@ 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 @@ def _build_arrow_table(self) -> pyarrow.Table: # 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 ddd5e8c6c6f0..f3ae4f434eb1 100644 --- a/paimon-python/pypaimon/table/system/buckets_table.py +++ b/paimon-python/pypaimon/table/system/buckets_table.py @@ -24,7 +24,8 @@ 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 @@ def primary_keys(self) -> List[str]: 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 @@ def _build_arrow_table(self) -> pyarrow.Table: 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 @@ def _build_arrow_table(self) -> pyarrow.Table: 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 @@ def _build_arrow_table(self) -> pyarrow.Table: 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 @@ def _build_arrow_table(self) -> pyarrow.Table: 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 @@ def _empty_table() -> pyarrow.Table: "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 290caae25221..d6df0307c87b 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_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 @@ def primary_keys(self) -> List[str]: 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 @@ def _build_arrow_table(self) -> pyarrow.Table: 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 @@ def _build_arrow_table(self) -> pyarrow.Table: 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 fca9ab6d8481..24e184d241d6 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 @@ _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 @@ def primary_keys(self) -> List[str]: 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 @@ def _build_arrow_table(self) -> pyarrow.Table: 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 @@ def _build_arrow_table(self) -> pyarrow.Table: 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 @@ def _build_arrow_table(self) -> pyarrow.Table: 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 @@ def _build_arrow_table(self) -> pyarrow.Table: 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 a2bcb90877ce..785261c8324b 100644 --- a/paimon-python/pypaimon/table/system/manifests_table.py +++ b/paimon-python/pypaimon/table/system/manifests_table.py @@ -24,7 +24,7 @@ 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 @@ def primary_keys(self) -> List[str]: 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 000000000000..0123576b69ec --- /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 7b917be4e310..77f8fb597e5d 100644 --- a/paimon-python/pypaimon/table/system/schemas_table.py +++ b/paimon-python/pypaimon/table/system/schemas_table.py @@ -23,7 +23,7 @@ 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 @@ def _build_arrow_table(self) -> pyarrow.Table: 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 d32c36163cf8..ee4302ec6a87 100644 --- a/paimon-python/pypaimon/table/system/snapshots_table.py +++ b/paimon-python/pypaimon/table/system/snapshots_table.py @@ -22,7 +22,7 @@ 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 @@ def _build_arrow_table(self) -> pyarrow.Table: 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 7ace7f23f36c..a4f2c11c8216 100644 --- a/paimon-python/pypaimon/table/system/system_table.py +++ b/paimon-python/pypaimon/table/system/system_table.py @@ -41,6 +41,23 @@ 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 e33717227e0f..e7f4c8173636 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 @@ def _materialise(self, splits: List[Split]) -> pyarrow.Table: 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 5bc43302a941..b6d1f6402573 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 @@ def __init__(self, system_table: "SystemTable"): 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 4bf2103f0b1a..f50bf2cf3993 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.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 @@ def primary_keys(self) -> List[str]: 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 @@ def _build_arrow_table(self) -> pyarrow.Table: 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 dc9bc18636a7..cc9bd23381ca 100644 --- a/paimon-python/pypaimon/table/system/tags_table.py +++ b/paimon-python/pypaimon/table/system/tags_table.py @@ -23,7 +23,7 @@ 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 @@ def _build_arrow_table(self) -> pyarrow.Table: 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 738cf43ceadc..ac764e90b731 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 000000000000..7af10c298db9 --- /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())]) + + +@pytest.fixture +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) + + +@pytest.mark.parametrize('name', sorted(_NATIVE_TABLES)) +@pytest.mark.parametrize('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}] + + +@pytest.mark.parametrize('name', ['files', 'file_key_ranges', 'buckets', 'manifests', 'table_indexes']) +@pytest.mark.parametrize('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'}] + + +@pytest.mark.parametrize('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') + + +@pytest.mark.parametrize('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]'] + + +@pytest.mark.parametrize('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 bfd8f9ef9d9b..cdfd1c676ae8 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.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 @@ def test_lists_every_branch_with_create_time(self): 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 57a352392a24..73cfaea0c5ec 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 @@ def test_aggregates_by_partition_and_bucket(self): 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 525ab5224d4d..df15a3a59c16 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 @@ def test_lists_files_with_partition_aggregation(self): 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 @@ def test_lists_files_with_partition_aggregation(self): 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 @@ def test_value_stats_render_actual_values(self): 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 @@ def test_value_stats_follow_field_ids_after_drop_and_re_add(self): # 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 @@ def get_field(self, pos): _row_values(CorruptRow()) +@pytest.mark.parametrize('type_name', ['TIMESTAMP_LTZ(9)', 'TIMESTAMP(9) WITH LOCAL TIME ZONE']) +@pytest.mark.parametrize('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] + '}' + + +@pytest.mark.parametrize('root,expected', [ + ('file:/warehouse/t', 'file:/warehouse/t/'), + ('file:///warehouse/t', 'file:/warehouse/t/'), + ('s3://bucket/warehouse/t', 's3://bucket/warehouse/t/'), +]) +@pytest.mark.parametrize('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 000000000000..4383b9ae761f --- /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))])) + + +@pytest.mark.parametrize('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 aadb69dcaaa8..3cd4e01f03ae 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.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 @@ def test_lists_committed_snapshots_in_id_order(self): 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 e801b656d8d7..290e9974e4f3 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.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 @@ def test_lists_created_tags_with_snapshot_metadata(self): 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():