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 0bc71fb402 [python] Use Native scan statistics in explain (#10609)
0bc71fb402 is described below
commit 0bc71fb402eb128e4b73c24ba5809177613859ed
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Oct 17 08:54:49 2026 +0800
[python] Use Native scan statistics in explain (#10609)
---
paimon-python/README.md | 18 +-
paimon-python/conftest.py | 4 +-
paimon-python/pypaimon/catalog/table_query_auth.py | 4 +-
paimon-python/pypaimon/read/explain.py | 21 +-
paimon-python/pypaimon/read/native_plan.py | 5 +-
paimon-python/pypaimon/read/plan.py | 32 +-
paimon-python/pypaimon/read/read_builder.py | 10 +-
paimon-python/pypaimon/read/scan_stats.py | 77 ++++-
.../read/scanner/bucket_select_converter.py | 7 +-
.../pypaimon/read/scanner/file_scanner.py | 1 +
paimon-python/pypaimon/read/table_scan.py | 13 +-
.../pypaimon/tests/native_bucket_selector_test.py | 235 +++++++++++++
.../pypaimon/tests/native_plan_integration_test.py | 11 +-
paimon-python/pypaimon/tests/native_plan_test.py | 39 ++-
.../pypaimon/tests/native_scan_stats_test.py | 375 +++++++++++++++++++++
.../pypaimon/tests/pushdown_bucket_test.py | 19 +-
.../pypaimon/tests/table_query_auth_test.py | 52 +--
17 files changed, 856 insertions(+), 67 deletions(-)
diff --git a/paimon-python/README.md b/paimon-python/README.md
index b0f5a2eb3e..ccb3cb5c18 100644
--- a/paimon-python/README.md
+++ b/paimon-python/README.md
@@ -362,7 +362,23 @@ print(explanation.native_planned)
The adapter checks the installed binding's capabilities and falls back to the
Python planner for unsupported scans. New bindings preserve `plan.snapshot_id`
even when pruning removes every split. Native explain output includes snapshot
-and split metadata; native pruning counters are not exposed.
+and split metadata plus counters for actual Native planning work, available in
+`explanation.planning_counters`. Statistics collection retains early manifest,
+sidecar and LIMIT optimizations. The Python scanner's before/after pruning
funnel
+is unavailable for Native plans because skipped entries are not decoded.
+Final split, file and byte totals describe the returned Plan after Python's
+post-processing and deferred LIMIT; decoding counters retain the Native work.
+Every Native plan includes lightweight counters. Use
+`plan = builder.new_scan().plan()` followed by `plan.stats()`; conversion to
+Python counters is deferred until this getter is called. The existing
+`plan, stats = builder.new_scan().scan_with_stats()` API remains available
+for both planners. Snapshot metadata is available separately as
`plan.snapshot_id`.
+
+Fixed-bucket Native planning also handles ORs of equality/IN predicates,
+intersects repeated key constraints, and specializes predicates per partition.
+It follows Java's 1000-candidate combination limit and keeps scans conservative
+when a predicate cannot determine buckets. Masked fields do not drive raw-value
+bucket pruning.
To run both split planning and data-file reading in Rust, enable the
independent
native-read option:
diff --git a/paimon-python/conftest.py b/paimon-python/conftest.py
index ef969ffb34..ee07dc8313 100644
--- a/paimon-python/conftest.py
+++ b/paimon-python/conftest.py
@@ -104,9 +104,9 @@ def pytest_configure(config):
original_plan = TableScan._try_native_plan
- def tracked_plan(self):
+ def tracked_plan(self, *args, **kwargs):
global _native_plan_count
- plan = original_plan(self)
+ plan = original_plan(self, *args, **kwargs)
if plan is not None and _force_native_for_test:
_native_plan_count += 1
return plan
diff --git a/paimon-python/pypaimon/catalog/table_query_auth.py
b/paimon-python/pypaimon/catalog/table_query_auth.py
index 04983ee97e..b4fc309dd0 100644
--- a/paimon-python/pypaimon/catalog/table_query_auth.py
+++ b/paimon-python/pypaimon/catalog/table_query_auth.py
@@ -17,6 +17,7 @@
################################################################################
import json
+from dataclasses import replace
from typing import Callable, Dict, List, Optional
import pyarrow as pa
@@ -26,7 +27,6 @@ from pypaimon.common.predicate_json_parser import (
extract_referenced_fields,
parse_predicate_to_batch_filter,
)
-from pypaimon.read.plan import Plan
from pypaimon.read.query_auth_split import QueryAuthSplit
from pypaimon.schema.data_types import DataField
from pypaimon.table.special_fields import SpecialFields
@@ -132,7 +132,7 @@ class TableQueryAuthResult:
if not self.has_restrictions:
return plan
auth_splits = [QueryAuthSplit(split, self) for split in plan.splits()]
- return Plan(auth_splits, snapshot_id=plan.snapshot_id)
+ return replace(plan, _splits=auth_splits)
def extract_row_filter(self) -> Optional[Callable[[pa.RecordBatch],
pa.Array]]:
if not self.filter:
diff --git a/paimon-python/pypaimon/read/explain.py
b/paimon-python/pypaimon/read/explain.py
index 5cc1cb1bd8..cc1b0516b2 100644
--- a/paimon-python/pypaimon/read/explain.py
+++ b/paimon-python/pypaimon/read/explain.py
@@ -130,9 +130,10 @@ class ExplainResult:
# Verbose-only
splits: Optional[List[ExplainSplitInfo]] = None
- # Native (pypaimon_rust) plan; pruning funnel not tracked. Kept last to
- # preserve positional-arg compatibility.
+ # Whether the same core Native pass produced the plan and pruning counters.
native_planned: bool = False
+ # Actual work performed by Native planning, retaining early metadata
pruning.
+ planning_counters: Optional[Dict[str, Any]] = None
def __str__(self) -> str:
return render_explain(self)
@@ -168,7 +169,7 @@ def render_explain(result: ExplainResult) -> str:
out.write("\n")
if result.native_planned:
- _line(out, "Planner", "native (pypaimon_rust); pruning not tracked")
+ _line(out, "Planner", "native (pypaimon_rust)")
_line(out, "Partition pruning",
result.partition_pruning.format() if result.partition_pruning else
"n/a")
_line(out, "Bucket pruning",
@@ -176,6 +177,20 @@ def render_explain(result: ExplainResult) -> str:
_line(out, "File skipping",
result.file_skipping.format() if result.file_skipping else "n/a")
+ if result.planning_counters is not None:
+ out.write("\nNative planning counters\n")
+ for label, name in (
+ ('Manifest entries decoded', 'manifest_entries_read'),
+ ('Bucket rejects', 'manifest_entries_pruned_by_bucket'),
+ ('Partition rejects', 'manifest_entries_pruned_by_partition'),
+ ('File statistics rejects',
'manifest_entries_pruned_by_data_stats'),
+ ('Cross-schema rejects',
'manifest_entries_pruned_by_cross_schema_stats'),
+ ('Manifest row-range skips',
'manifest_files_pruned_by_row_ranges'),
+ ('Split candidates built', 'split_candidates_built'),
+ ('LIMIT stopped planning', 'limit_early_stopped'),
+ ):
+ _line(out, label, str(result.planning_counters[name]))
+
out.write("\n")
_line(out, "Splits", str(result.split_count))
if result.split_count > 0:
diff --git a/paimon-python/pypaimon/read/native_plan.py
b/paimon-python/pypaimon/read/native_plan.py
index b3c8267598..f968f85154 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -514,7 +514,8 @@ def native_plan(
scan = scan.with_chunk_shuffle(str(seed), chunk_size)
if shard is not None:
scan = scan.with_shard(*shard)
- return _from_native_plan(table, scan.plan())
+ rust_plan = scan.plan()
+ return _from_native_plan(table, rust_plan)
def _native_stream_auth_supported(table):
@@ -576,4 +577,4 @@ def _from_native_plan(table, rust_plan) -> Plan:
# Older bindings cannot distinguish an empty committed snapshot from a
# table without snapshots. Let the Python scanner recover the metadata.
raise RuntimeError("Native runtime cannot report an empty plan's
snapshot")
- return Plan(splits, snapshot_id=snapshot_id)
+ return Plan(splits, snapshot_id=snapshot_id,
_native_stats=rust_plan.stats())
diff --git a/paimon-python/pypaimon/read/plan.py
b/paimon-python/pypaimon/read/plan.py
index dc7081bea7..88746311dc 100644
--- a/paimon-python/pypaimon/read/plan.py
+++ b/paimon-python/pypaimon/read/plan.py
@@ -16,10 +16,11 @@
# under the License.
from dataclasses import dataclass, field
-from typing import List, Optional
+from typing import Any, List, Optional
from pypaimon.read.split import Split
+from pypaimon.read.scan_stats import ScanStats
@dataclass
@@ -27,6 +28,35 @@ class Plan:
"""Implementation of Plan for native Python reading."""
_splits: List[Split]
snapshot_id: Optional[int] = field(default=None)
+ _scan_stats: Optional[ScanStats] = field(default=None, repr=False,
compare=False)
+ _native_stats: Any = field(default=None, repr=False, compare=False)
+ # A LIMIT hint deferred until after Native row-position selection.
+ _stats_limit: Optional[int] = field(default=None, repr=False,
compare=False)
def splits(self) -> List[Split]:
return self._splits
+
+ def stats(self) -> Optional[ScanStats]:
+ """Planning counters, converted from Native only when requested."""
+ native_stats = self._native_stats
+ if native_stats is not None:
+ self._scan_stats = ScanStats.from_native(native_stats)
+ # Python can filter or limit splits after Native planning. Keep the
+ # actual planning work, but summarize the final returned selection.
+ files, data_bytes = 0, 0
+ for split in self._splits:
+ for file in split.files:
+ files += 1
+ data_bytes += max(0, file.file_size)
+ counters = self._scan_stats.planning_counters
+ counters.update(final_splits=len(self._splits),
splits_after_limit=len(self._splits),
+ final_files=files,
planned_data_file_bytes=min(data_bytes, 2 ** 64 - 1))
+ if self._stats_limit is not None:
+ counters['limit'] = self._stats_limit
+ self._native_stats = None
+ return self._scan_stats
+
+ def __getstate__(self):
+ # Native handles are process-local. Serialize counters as Python
scalars.
+ self.stats()
+ return self.__dict__.copy()
diff --git a/paimon-python/pypaimon/read/read_builder.py
b/paimon-python/pypaimon/read/read_builder.py
index f990d7ded6..cb5373cdc8 100644
--- a/paimon-python/pypaimon/read/read_builder.py
+++ b/paimon-python/pypaimon/read/read_builder.py
@@ -226,8 +226,8 @@ class ReadBuilder:
Cost: ``explain()`` reads manifest list + manifests but never
opens data files. To produce accurate before/after counters it
- suppresses the manifest-reader's early bucket filter and forces
- single-threaded manifest decoding, so it can be measurably
+ suppresses early bucket and sidecar pruning and counts candidates
+ before applying LIMIT, so it can be measurably
heavier than a regular ``new_scan().plan()`` on tables where the
early filter usually prunes aggressively (e.g. very wide
HASH_FIXED tables with a tight predicate).
@@ -482,9 +482,8 @@ def _build_explain_result(table, scan: TableScan, plan,
stats: Optional[ScanStat
table_schema = table.table_schema
bucket_mode_str = _safe_bucket_mode(table)
- # Native plans expose split metadata without Python pruning counters.
- native_planned = stats is None
- if native_planned:
+ native_planned = stats is None or stats.native_planned
+ if stats is None or native_planned:
partition_pruning = bucket_pruning = file_skipping = None
else:
partition_pruning = _partition_pruning(stats, scan)
@@ -586,6 +585,7 @@ def _build_explain_result(table, scan: TableScan, plan,
stats: Optional[ScanStat
split_size_p95=sz_p95,
has_auth=plan_has_auth,
native_planned=native_planned,
+ planning_counters=stats.planning_counters if stats is not None else
None,
splits=split_infos if verbose else None,
)
diff --git a/paimon-python/pypaimon/read/scan_stats.py
b/paimon-python/pypaimon/read/scan_stats.py
index 6626b0c240..2e7de036c3 100644
--- a/paimon-python/pypaimon/read/scan_stats.py
+++ b/paimon-python/pypaimon/read/scan_stats.py
@@ -16,7 +16,38 @@
# under the License.
from dataclasses import dataclass, field
-from typing import Set, Tuple
+from typing import Any, Dict, Optional, Set, Tuple
+
+
+_NATIVE_COUNTER_FIELDS = (
+ 'base_manifest_files',
+ 'delta_manifest_files',
+ 'manifest_files_before_partition_pruning',
+ 'manifest_files_after_partition_pruning',
+ 'manifest_files_pruned_by_row_ranges',
+ 'manifest_entries_read',
+ 'manifest_entries_pruned_by_bucket',
+ 'manifest_entries_pruned_by_partition',
+ 'manifest_entries_after_entry_pruning',
+ 'manifest_entries_pruned_by_level',
+ 'manifest_entries_pruned_by_row_ranges',
+ 'manifest_entries_pruned_by_data_stats',
+ 'manifest_entries_after_manifest_filters',
+ 'manifest_entries_after_merge',
+ 'manifest_entries_pruned_by_cross_schema_stats',
+ 'manifest_entries_after_cross_schema_stats',
+ 'data_evolution_groups_before_stats',
+ 'data_evolution_groups_pruned_by_stats',
+ 'data_evolution_groups_pruned_by_row_ranges',
+ 'split_candidates_built',
+ 'limit_early_stopped',
+ 'splits_before_limit',
+ 'splits_after_limit',
+ 'final_splits',
+ 'final_files',
+ 'planned_data_file_bytes',
+ 'limit',
+)
@dataclass
@@ -38,14 +69,42 @@ class ScanStats:
# manifest-level partition skip. ``entries_total`` is what actually reached
# ``_filter_manifest_entry``; the gap between the two captures
# manifest-level pruning.
- entries_potential_total: int = 0
+ entries_potential_total: Optional[int] = 0
entries_total: int = 0
- entries_after_partition: int = 0
- entries_after_bucket: int = 0
- entries_after_stats: int = 0
+ entries_after_partition: Optional[int] = 0
+ entries_after_bucket: Optional[int] = 0
+ entries_after_stats: Optional[int] = 0
+
+ # Identity sets are Python-scanner bookkeeping. Native returns counters
+ # without transporting partition values back across the language boundary.
+ partition_keys_before: Optional[Set[Tuple]] = field(default_factory=set)
+ partition_keys_after: Optional[Set[Tuple]] = field(default_factory=set)
+
+ buckets_seen: Optional[Set[Tuple[Tuple, int]]] = field(default_factory=set)
+ buckets_after_pruning: Optional[Set[Tuple[Tuple, int]]] =
field(default_factory=set)
+ native_planned: bool = False
+ # Native counters describe actual planning work; skipped metadata is not
decoded
+ # to reconstruct the Python scanner's before/after funnel.
+ planning_counters: Optional[Dict[str, Any]] = None
+
+ @classmethod
+ def from_native(cls, stats):
+ return cls._from_native_counters({name: getattr(stats, name) for name
in _NATIVE_COUNTER_FIELDS})
- partition_keys_before: Set[Tuple] = field(default_factory=set)
- partition_keys_after: Set[Tuple] = field(default_factory=set)
+ @classmethod
+ def empty_native(cls, limit=None):
+ counters = dict.fromkeys(_NATIVE_COUNTER_FIELDS, 0)
+ counters.update(limit=limit, limit_early_stopped=False)
+ return cls._from_native_counters(counters)
- buckets_seen: Set[Tuple[Tuple, int]] = field(default_factory=set)
- buckets_after_pruning: Set[Tuple[Tuple, int]] = field(default_factory=set)
+ @classmethod
+ def _from_native_counters(cls, counters):
+ return cls(
+
manifest_files_total=counters['manifest_files_before_partition_pruning'],
+
manifest_files_after_partition=counters['manifest_files_after_partition_pruning'],
+ entries_total=counters['manifest_entries_read'],
+ entries_potential_total=None, entries_after_partition=None,
+ entries_after_bucket=None, entries_after_stats=None,
+ planning_counters=counters,
+ partition_keys_before=None, partition_keys_after=None,
+ buckets_seen=None, buckets_after_pruning=None, native_planned=True)
diff --git a/paimon-python/pypaimon/read/scanner/bucket_select_converter.py
b/paimon-python/pypaimon/read/scanner/bucket_select_converter.py
index c952ff886a..6a9cd060f5 100644
--- a/paimon-python/pypaimon/read/scanner/bucket_select_converter.py
+++ b/paimon-python/pypaimon/read/scanner/bucket_select_converter.py
@@ -392,7 +392,7 @@ class _Selector:
# ``total_buckets <= 0`` shows up for postpone / legacy / special
# entries and must NOT be pruned: returning False here would drop
# rows the writer hashed under a different convention. Fail open.
- if total_buckets <= 0:
+ if bucket < 0 or total_buckets <= 0:
return True
try:
return bucket in self._compute(partition, total_buckets)
@@ -430,8 +430,9 @@ class _Selector:
self._predicate, partition_field_names, partition_values)
if effective_predicate is _AlwaysFalse:
- # No row in this partition can match — empty bucket set.
- frozen: FrozenSet[int] = frozenset()
+ # Java cannot extract Equal/In values from AlwaysFalse. Leave
+ # rejection to the partition predicate rather than the key hash.
+ frozen: FrozenSet[int] = frozenset(range(total_buckets))
self._cache[cache_key] = frozen
return frozen
diff --git a/paimon-python/pypaimon/read/scanner/file_scanner.py
b/paimon-python/pypaimon/read/scanner/file_scanner.py
index 22779bdfc5..bcbf73e033 100755
--- a/paimon-python/pypaimon/read/scanner/file_scanner.py
+++ b/paimon-python/pypaimon/read/scanner/file_scanner.py
@@ -699,6 +699,7 @@ class FileScanner:
"""
self.scan_stats = ScanStats()
plan = self.scan()
+ plan._scan_stats = self.scan_stats
return plan, self.scan_stats
def with_chunk_shuffle(self, seed: int, chunk_size: int) -> 'FileScanner':
diff --git a/paimon-python/pypaimon/read/table_scan.py
b/paimon-python/pypaimon/read/table_scan.py
index c3d4edc504..ed3bfd9aab 100755
--- a/paimon-python/pypaimon/read/table_scan.py
+++ b/paimon-python/pypaimon/read/table_scan.py
@@ -18,6 +18,7 @@
import json as _json
import logging
import re
+from dataclasses import replace
from typing import List, Optional, Tuple
from pypaimon.catalog.catalog_exception import TableNoPermissionException
@@ -200,7 +201,7 @@ class TableScan:
extra_options = {}
if
self.table.options.options.contains(CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP):
if self._incremental_snapshot_range is None:
- return Plan([])
+ return Plan([],
_scan_stats=ScanStats.empty_native(self.limit))
extra_options['incremental_range'] =
self._incremental_snapshot_range
if self.table.options.changelog_producer() !=
ChangelogProducer.NONE:
extra_options['incremental_mode'] = 'changelog'
@@ -267,7 +268,7 @@ class TableScan:
if not (getattr(fs, 'deletion_vectors_enabled', False)
and not self.table.is_primary_key_table):
splits = fs._apply_push_down_limit(splits)
- return Plan(splits, snapshot_id=plan.snapshot_id)
+ return replace(plan, _splits=splits, _stats_limit=self.limit if
defer_limit else None)
except Exception as e:
_raise_if_native_fork_safety_error(e)
# Any native construction/planning/pruning failure -> fall back.
@@ -296,15 +297,15 @@ class TableScan:
"""Run :meth:`plan` while recording manifest / pruning counters.
Only used by :meth:`ReadBuilder.explain`; the regular read path
- keeps going through :meth:`plan`. Native plans return stats=None;
- explain reports their split metadata without pruning counters.
+ keeps going through :meth:`plan`. Native scans always collect
lightweight
+ counters in the same core planning pass, without opening data files.
"""
native_auth = (self.table.options.query_auth_enabled
and self._native_requested() and
self._native_plan_supported())
if native_auth:
native = self._try_native_plan()
if native is not None:
- return native, None
+ return native, native.stats()
auth_result = self.__auth_query()
if self._query_auth_fn is not None:
prune_scanner_by_auth(self.table, self.file_scanner, auth_result)
@@ -312,7 +313,7 @@ class TableScan:
and self._native_plan_supported()):
native = self._try_native_plan()
if native is not None:
- return native, None
+ return native, native.stats()
plan, stats = self.file_scanner.scan_with_stats()
return wrap_plan_with_auth(auth_result, plan), stats
diff --git a/paimon-python/pypaimon/tests/native_bucket_selector_test.py
b/paimon-python/pypaimon/tests/native_bucket_selector_test.py
new file mode 100644
index 0000000000..eae06eaf8c
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_bucket_selector_test.py
@@ -0,0 +1,235 @@
+# 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.
+
+"""Java BucketSelector semantics through REST, normal and explain planning."""
+
+from contextlib import ExitStack
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.tests.native_scan_stats_test import _append, _create
+from pypaimon.tests import native_plan_rest_test
+
+rest_catalog = native_plan_rest_test.rest_catalog
+
+pytestmark = [pytest.mark.native_plan, pytest.mark.skipif(
+ not native_runtime_available(), reason='Rust planner required')]
+
+
+def _selected(table, predicate, native, stats=True, partition_filter=None,
limit=None):
+ reader = table.copy({'scan.native-plan.enabled': str(native).lower(),
+ 'read.native.enabled':
str(native).lower()}).new_read_builder()
+ reader.with_filter(predicate)
+ if limit is not None:
+ reader.with_limit(limit)
+ if partition_filter is not None:
+ reader.with_partition_filter(partition_filter)
+ scan = reader.new_scan()
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch.object(scan.file_scanner, 'scan',
+
side_effect=AssertionError('Native scan fell back')))
+ stack.enter_context(patch.object(scan.file_scanner,
'scan_with_stats',
+
side_effect=AssertionError('Native statistics fell back')))
+
stack.enter_context(patch('pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError('Native read
fell back')))
+ if stats:
+ plan, counters = scan.scan_with_stats()
+ assert plan.stats() is counters
+ else:
+ plan, counters = scan.plan(), None
+ rows = reader.new_read().to_arrow(plan.splits()).to_pylist()
+ files = sorted((tuple(split.partition.values), split.bucket,
+ tuple(file.file_name for file in split.files)) for split
in plan.splits())
+ return files, sorted(rows, key=lambda row: tuple(row[name] for name in
sorted(row))), counters
+
+
+def _predicate(pb, case):
+ equal = lambda name, value: pb.equal(name, value)
+ and_ = pb.and_predicates
+ or_ = pb.or_predicates
+ if case == 'or':
+ return or_([equal('id', 2), equal('id', 3)])
+ if case == 'intersection':
+ return and_([pb.is_in('id', [1, 2, 3]), pb.is_in('id', [2, 3, 4])])
+ if case == 'partition-or':
+ return or_([and_([equal('pt', 1), equal('id', 3)]),
+ and_([equal('pt', 2), equal('id', 7)])])
+ if case == 'unconstrained-branch':
+ return or_([and_([equal('pt', 1), equal('id', 3)]), equal('pt', 2)])
+ if case == 'unsafe-or':
+ return or_([equal('id', 3), equal('value', 0)])
+ if case == 'safe-and':
+ return and_([equal('id', 3), or_([equal('id', 3), equal('value', 0)])])
+ if case == 'contradiction':
+ return and_([equal('id', 2), equal('id', 3)])
+ raise AssertionError(case)
+
+
[email protected]('primary_key', [False, True], ids=['append', 'pk'])
[email protected]('case', ['or', 'intersection', 'partition-or',
'unconstrained-branch',
+ 'unsafe-or', 'safe-and', 'contradiction'])
+def test_native_bucket_selection_matches_java_compatible_python(rest_catalog,
primary_key, case):
+ catalog, _ = rest_catalog
+ schema = pa.schema([('pt', pa.int32()), ('id', pa.int32()), ('value',
pa.int32())])
+ table, _ = _create(catalog, schema, ['pt'], ['pt', 'id'] if primary_key
else None,
+ {'bucket': '4', 'bucket-key': 'id',
'manifest.sidecar.enabled': 'true'})
+ rows = [{'pt': pt, 'id': index, 'value': index} for pt in [1, 2] for index
in range(64)]
+ _append(table, rows, schema)
+ predicate = _predicate(table.new_read_builder().new_predicate_builder(),
case)
+ files, expected, python = _selected(table, predicate, False)
+ actual_files, actual, native = _selected(table, predicate, True)
+ assert actual == expected
+ assert actual_files == files
+ assert native.entries_potential_total is None
+ assert native.entries_total <= python.entries_total
+ assert native.planning_counters['manifest_entries_after_entry_pruning'] ==
python.entries_after_bucket
+ assert native.planning_counters['manifest_entries_after_manifest_filters']
== python.entries_after_stats
+ assert _selected(table, predicate, True, stats=False)[:2] ==
(actual_files, actual)
+ if case == 'partition-or':
+ assert
native.planning_counters['manifest_entries_after_entry_pruning'] == 2
+ elif case == 'unconstrained-branch':
+ assert
native.planning_counters['manifest_entries_after_entry_pruning'] == 5
+ elif case in ('unsafe-or', 'contradiction'):
+ assert
native.planning_counters['manifest_entries_after_entry_pruning'] == 8
+ else:
+ assert
native.planning_counters['manifest_entries_after_entry_pruning'] < 8
+
+
[email protected]('case', ['one-key-or', 'intersection', 'mixed-or'])
+def
test_native_composite_bucket_keys_preserve_java_clause_boundaries(rest_catalog,
case):
+ catalog, _ = rest_catalog
+ schema = pa.schema([('pt', pa.int32()), ('a', pa.int32()), ('b',
pa.int32()), ('value', pa.int32())])
+ table, _ = _create(catalog, schema, ['pt'], ['pt', 'a', 'b'], {'bucket':
'8', 'bucket-key': 'a,b'})
+ _append(table, [{'pt': pt, 'a': a, 'b': b, 'value': a + b}
+ for pt in [1, 2] for a in range(16) for b in range(16)],
schema)
+ pb = table.new_read_builder().new_predicate_builder()
+ if case == 'one-key-or':
+ predicate = pb.and_predicates([pb.or_predicates([pb.equal('a', 1),
pb.equal('a', 2)]), pb.equal('b', 3)])
+ elif case == 'intersection':
+ predicate = pb.and_predicates([pb.is_in('a', [1, 2, 3]), pb.is_in('a',
[2, 3, 4]), pb.equal('b', 3)])
+ else:
+ predicate = pb.or_predicates([pb.equal('a', 1), pb.equal('b', 3)])
+ files, expected, python = _selected(table, predicate, False)
+ actual_files, actual, native = _selected(table, predicate, True)
+ assert (actual_files, actual) == (files, expected)
+ assert native.planning_counters['manifest_entries_after_entry_pruning'] ==
python.entries_after_bucket
+ assert native.planning_counters['manifest_entries_after_manifest_filters']
== python.entries_after_stats
+ if case == 'mixed-or':
+ assert
native.planning_counters['manifest_entries_after_entry_pruning'] == 16
+ else:
+ assert
native.planning_counters['manifest_entries_after_entry_pruning'] <= 4
+
+
+def
test_explicit_partition_filter_does_not_replace_partition_values_in_bucket_predicate(rest_catalog):
+ catalog, _ = rest_catalog
+ schema = pa.schema([('pt', pa.int32()), ('id', pa.int32()), ('value',
pa.int32())])
+ table, _ = _create(catalog, schema, ['pt'], ['pt', 'id'], {'bucket': '4',
'bucket-key': 'id'})
+ _append(table, [{'pt': pt, 'id': index, 'value': index} for pt in [1, 2]
for index in range(64)], schema)
+ pb = table.new_read_builder().new_predicate_builder()
+ predicate = _predicate(pb, 'partition-or')
+ selector = pb.equal('pt', 2)
+ files, expected, python = _selected(table, predicate, False,
partition_filter=selector)
+ actual_files, actual, native = _selected(table, predicate, True,
partition_filter=selector)
+ assert (actual_files, actual) == (files, expected)
+ assert actual == [{'pt': 2, 'id': 7, 'value': 7}]
+ assert native.planning_counters['manifest_entries_after_entry_pruning'] ==
python.entries_after_bucket == 1
+
+
[email protected]('masked', ['pt', 'value'])
+def
test_native_bucket_specialization_excludes_masked_inputs_and_keeps_safe_conjuncts(rest_catalog,
masked):
+ import json
+ from pypaimon.api.api_response import AuthTableQueryResponse
+ from pypaimon.schema.schema_change import SchemaChange
+ catalog, server = rest_catalog
+ schema = pa.schema([('pt', pa.string()), ('id', pa.int32()), ('value',
pa.string())])
+ table, _ = _create(catalog, schema, ['pt'], ['pt', 'id'], {'bucket': '4',
'bucket-key': 'id'})
+ _append(table, [{'pt': pt, 'id': index, 'value': str(index)}
+ for pt in ['a', 'b'] for index in range(64)], schema)
+ catalog.alter_table(table.identifier,
[SchemaChange.set_option('query-auth.enabled', 'true')])
+ table = catalog.get_table(table.identifier)
+ server.table_query_auth_store[table.identifier.get_full_name()] =
AuthTableQueryResponse(
+ column_masking={masked: json.dumps({'name': 'CONCAT', 'inputs':
['visible']})})
+ pb = table.new_read_builder().new_predicate_builder()
+ if masked == 'pt':
+ # Raw partition 'a' would select id=7, losing the matching id=3 after
masking.
+ predicate = pb.or_predicates([
+ pb.and_predicates([pb.equal('pt', 'a'), pb.equal('id', 7)]),
+ pb.and_predicates([pb.equal('pt', 'visible'), pb.equal('id', 3)]),
+ ])
+ else:
+ predicate = pb.and_predicates([pb.equal('id', 3), pb.equal('value',
'visible')])
+ files, expected, python = _selected(table, predicate, False)
+ actual_files, actual, native = _selected(table, predicate, True)
+ assert (actual_files, actual) == (files, expected)
+ assert [row['id'] for row in actual] == [3, 3]
+ assert all(row[masked] == 'visible' for row in actual)
+ selected_entries =
native.planning_counters['manifest_entries_after_entry_pruning']
+ assert selected_entries == python.entries_after_bucket == (8 if masked ==
'pt' else 2)
+ assert _selected(table, predicate, True, stats=False)[:2] ==
(actual_files, actual)
+
+
[email protected]('limit', [0, 1])
+def test_native_correlated_bucket_statistics_preserve_limit(rest_catalog,
limit):
+ catalog, _ = rest_catalog
+ schema = pa.schema([('pt', pa.int32()), ('id', pa.int32()), ('value',
pa.int32())])
+ table, _ = _create(catalog, schema, ['pt'], ['pt', 'id'], {'bucket': '4',
'bucket-key': 'id'})
+ _append(table, [{'pt': pt, 'id': index, 'value': index} for pt in [1, 2]
for index in range(64)], schema)
+ predicate = _predicate(table.new_read_builder().new_predicate_builder(),
'partition-or')
+ files, actual, stats = _selected(table, predicate, True, limit=limit)
+ assert stats.entries_potential_total is None
+ assert stats.planning_counters['manifest_entries_read'] == (0 if limit ==
0 else 8)
+ assert stats.planning_counters['manifest_entries_after_entry_pruning'] ==
(0 if limit == 0 else 2)
+ assert len(actual) == limit
+ matches = [{'pt': 1, 'id': 3, 'value': 3}, {'pt': 2, 'id': 7, 'value': 7}]
+ assert all(row in matches for row in actual)
+ plain_files, plain_rows, _ = _selected(table, predicate, True,
stats=False, limit=limit)
+ assert plain_files == files
+ assert len(plain_rows) == limit
+ # PK bucket scans have no row order, so LIMIT may select either matching
row.
+ assert all(row in matches for row in plain_rows)
+
+
[email protected]('changelog', [False, True])
+def
test_native_incremental_bucket_specialization_preserves_physical_events(rest_catalog,
changelog):
+ import json
+ catalog, _ = rest_catalog
+ schema = pa.schema([('pt', pa.int32()), ('id', pa.int32()), ('value',
pa.int32())])
+ options = {'bucket': '4', 'bucket-key': 'id'}
+ if changelog:
+ options['changelog-producer'] = 'input'
+ table, _ = _create(catalog, schema, ['pt'], ['pt', 'id'], options)
+ for timestamp in [1000, 2000]:
+ _append(table, [{'pt': pt, 'id': index, 'value': timestamp}
+ for pt in [1, 2] for index in range(64)], schema)
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ path = table.snapshot_manager().get_snapshot_path(snapshot.id)
+ metadata = json.loads(table.file_io.read_file_utf8(path))
+ metadata['timeMillis'] = timestamp
+ table.file_io.overwrite_file_utf8(path, json.dumps(metadata))
+ table = table.copy({'incremental-between-timestamp': '0,2000'})
+ predicate = _predicate(table.new_read_builder().new_predicate_builder(),
'partition-or')
+ files, expected, python = _selected(table, predicate, False)
+ actual_files, actual, native = _selected(table, predicate, True)
+ assert (actual_files, actual) == (files, expected)
+ assert len(actual) == 4
+ assert native.planning_counters['manifest_entries_after_entry_pruning'] ==
python.entries_after_bucket == 4
+ assert native.planning_counters['manifest_entries_after_manifest_filters']
== python.entries_after_stats
+ assert _selected(table, predicate, True, stats=False)[:2] ==
(actual_files, actual)
diff --git a/paimon-python/pypaimon/tests/native_plan_integration_test.py
b/paimon-python/pypaimon/tests/native_plan_integration_test.py
index 7f0647b492..3ea5117708 100644
--- a/paimon-python/pypaimon/tests/native_plan_integration_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_integration_test.py
@@ -1548,7 +1548,7 @@ class NativePlanIntegrationTest(unittest.TestCase):
@unittest.skipUnless(native_method_available('Plan', 'snapshot_id'),
"pypaimon_rust snapshot metadata API not installed")
- def
test_native_explain_reports_split_metadata_without_pruning_counters(self):
+ def test_native_explain_reports_split_metadata_and_pruning_counters(self):
self.cat.create_table(
'default.explain_metadata_t', Schema.from_pyarrow_schema(
self.schema, options={'metadata.stats-mode': 'full'}), False)
@@ -1563,13 +1563,20 @@ class NativePlanIntegrationTest(unittest.TestCase):
self.assertEqual(result.snapshot_id, 2)
self.assertEqual(result.split_count, 1)
self.assertEqual(result.file_count, 1)
- self.assertIn('pruning not tracked', str(result))
+ self.assertNotIn('pruning not tracked', str(result))
self.assertIsNone(result.file_skipping)
+ self.assertEqual(result.planning_counters['manifest_entries_read'], 2)
+
self.assertEqual(result.planning_counters['manifest_entries_pruned_by_data_stats'],
1)
+ self.assertEqual(result.planning_counters['final_files'], 1)
builder.with_filter(builder.new_predicate_builder().equal('k', 99))
empty = builder.explain()
self.assertTrue(empty.native_planned)
self.assertEqual(empty.snapshot_id, 2)
self.assertEqual(empty.split_count, 0)
+ self.assertIsNone(empty.file_skipping)
+ self.assertEqual(empty.planning_counters['manifest_entries_read'], 2)
+
self.assertEqual(empty.planning_counters['manifest_entries_pruned_by_data_stats'],
2)
+ self.assertEqual(empty.planning_counters['final_files'], 0)
def test_empty_table_explain_preserves_native_metadata(self):
self.cat.create_table(
diff --git a/paimon-python/pypaimon/tests/native_plan_test.py
b/paimon-python/pypaimon/tests/native_plan_test.py
index f9b629d4e4..5e1adde4ad 100644
--- a/paimon-python/pypaimon/tests/native_plan_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_test.py
@@ -99,6 +99,32 @@ def _scan(native_enabled, file_scanner):
class NativePlanTest(unittest.TestCase):
+ def
test_native_final_statistics_preserve_work_and_saturate_unknown_sizes(self):
+ from pypaimon.read.scan_stats import ScanStats
+ counters = ScanStats.empty_native()
+ counters.planning_counters.update(
+ manifest_entries_read=10, split_candidates_built=5,
splits_before_limit=5)
+ maximum = 2 ** 63 - 1
+ splits = [SimpleNamespace(files=[SimpleNamespace(file_size=size) for
size in sizes])
+ for sizes in [[maximum, -1], [maximum, 2]]]
+ plan = Plan(splits, 7, _native_stats=object(), _stats_limit=1)
+ with patch.object(ScanStats, 'from_native', return_value=counters) as
convert:
+ stats = plan.stats()
+ self.assertIs(plan.stats(), stats)
+ convert.assert_called_once()
+ actual = stats.planning_counters
+ self.assertEqual(actual['manifest_entries_read'], 10)
+ self.assertEqual(actual['split_candidates_built'], 5)
+ self.assertEqual(actual['splits_before_limit'], 5)
+ self.assertEqual(actual['final_splits'], 2)
+ self.assertEqual(actual['splits_after_limit'], 2)
+ self.assertEqual(actual['final_files'], 4)
+ self.assertEqual(actual['planned_data_file_bytes'], 2 ** 64 - 1)
+ self.assertEqual(actual['limit'], 1)
+ zero = Plan([], 7, _native_stats=object(), _stats_limit=0)
+ with patch.object(ScanStats, 'from_native',
return_value=ScanStats.empty_native()):
+ self.assertEqual(zero.stats().planning_counters['limit'], 0)
+
def test_variant_read_type_matches_java_interop_fixture(self):
actual = _variant_read_type_json(
[DataField(7, 'payload', AtomicType('VARIANT'))],
@@ -643,13 +669,16 @@ class NativePlanTest(unittest.TestCase):
fs.scan.assert_called_once_with()
def test_scan_with_stats_preserves_native_empty_snapshot(self):
+ from pypaimon.read.scan_stats import ScanStats
fs = Mock(partition_key_predicate=None)
scan = _scan(native_enabled=True, file_scanner=fs)
- native = Plan([], 7)
+ counters = ScanStats.empty_native()
+ native = Plan([], 7, _scan_stats=counters)
with patch('pypaimon.read.native_plan.native_plan',
return_value=native) as np:
plan, stats = scan.scan_with_stats()
self.assertEqual(plan.snapshot_id, 7)
- self.assertIsNone(stats)
+ self.assertIs(stats, counters)
+ self.assertIs(plan.stats(), counters)
np.assert_called_once_with(
scan.table, predicate=None, limit=None, projection=None,
row_ranges=None)
@@ -1213,7 +1242,7 @@ class NativePlanTest(unittest.TestCase):
with self.subTest(snapshot_id=snapshot_id, legacy=legacy):
table = _scan(True, Mock()).table
table.table_schema = Mock(fields=[], partition_keys=[])
- rust_plan = SimpleNamespace(splits=lambda: [])
+ rust_plan = SimpleNamespace(splits=lambda: [], stats=lambda:
None)
if not legacy:
rust_plan.snapshot_id = lambda: snapshot_id
scan = SimpleNamespace(plan=lambda: rust_plan)
@@ -1233,7 +1262,7 @@ class NativePlanTest(unittest.TestCase):
table.table_schema = TableSchema(0, [])
table.options = CoreOptions(Options({}))
table.current_branch.return_value = 'b1'
- rust_plan = SimpleNamespace(splits=lambda: [], snapshot_id=lambda: 7)
+ rust_plan = SimpleNamespace(splits=lambda: [], stats=lambda: None,
snapshot_id=lambda: 7)
scan = Mock()
scan.plan.return_value = rust_plan
rt = Mock()
@@ -1357,7 +1386,7 @@ class NativePlanTest(unittest.TestCase):
def test_native_plan_forwards_explicit_incremental_mode(self):
table = _scan(True, Mock()).table
table.table_schema = Mock(fields=[], partition_keys=[])
- rust_plan = SimpleNamespace(splits=lambda: [], snapshot_id=lambda: 4)
+ rust_plan = SimpleNamespace(splits=lambda: [], stats=lambda: None,
snapshot_id=lambda: 4)
rust_scan = Mock()
rust_scan.plan.return_value = rust_plan
builder = Mock()
diff --git a/paimon-python/pypaimon/tests/native_scan_stats_test.py
b/paimon-python/pypaimon/tests/native_scan_stats_test.py
new file mode 100644
index 0000000000..a41280abfb
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_scan_stats_test.py
@@ -0,0 +1,375 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Native explain reports ordinary planning work without disabling early
pruning."""
+
+from contextlib import ExitStack
+import json
+import pickle
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import Schema
+from pypaimon.filesystem.local_file_io import LocalFileIO
+from pypaimon.read.native_plan import _native_table, native_runtime_available
+from pypaimon.tests import native_plan_rest_test, native_query_auth_test
+from pypaimon.utils.file_store_path_factory import canonical_data_file_path
+
+rest_catalog = native_plan_rest_test.rest_catalog
+guarded_table = native_query_auth_test.guarded_table
+pytestmark = [pytest.mark.native_plan, pytest.mark.skipif(
+ not native_runtime_available(), reason='Rust planner required')]
+
+
+def _create(catalog, schema=None, partition_keys=None, primary_keys=None,
options=None):
+ schema = schema or pa.schema([('id', pa.int32()), ('value', pa.int32())])
+ settings = {'bucket': '-1', 'source.split.target-size': '1b',
+ 'source.split.open-file-cost': '1b',
'scan.native-plan.enabled': 'true',
+ 'read.native.enabled': 'false', 'write.native.enabled':
'false',
+ 'commit.native.enabled': 'false', 'metadata.stats-mode':
'full'}
+ settings.update(options or {})
+ catalog.create_table('default.scan_stats', Schema.from_pyarrow_schema(
+ schema, partition_keys=partition_keys or [], primary_keys=primary_keys
or [], options=settings), False)
+ return catalog.get_table('default.scan_stats'), schema
+
+
+def _append(table, rows, schema, overwrite=False):
+ builder = table.new_batch_write_builder()
+ if overwrite:
+ builder.overwrite()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.Table.from_pylist(rows, schema=schema))
+ messages = writer.prepare_commit()
+ commit.commit(messages)
+ return messages
+ finally:
+ writer.close()
+ commit.close()
+
+
+def _native_explain(builder, verbose=False):
+ with
patch('pypaimon.read.scanner.file_scanner.FileScanner.scan_with_stats',
+ side_effect=AssertionError('explain fell back to Python')):
+ result = builder.explain(verbose)
+ assert result.native_planned
+ assert 'pruning not tracked' not in str(result)
+ return result
+
+
[email protected]('limit,expected_files', [(None, 2), (0, 0), (1, 2)])
+def test_native_explain_reports_actual_file_statistics_work(rest_catalog,
limit, expected_files):
+ catalog, _ = rest_catalog
+ table, schema = _create(catalog)
+ for start in [0, 10, 100]:
+ _append(table, [{'id': start + i, 'value': start + i} for i in
range(3)], schema)
+ builder = table.new_read_builder()
+ builder.with_filter(builder.new_predicate_builder().less_or_equal('value',
20))
+ if limit is not None:
+ builder.with_limit(limit)
+ result = _native_explain(builder, True)
+ assert result.snapshot_id == 3
+ assert result.file_count == expected_files
+ assert result.file_skipping is None
+ assert result.planning_counters['manifest_entries_read'] == (0 if limit ==
0 else 3)
+ assert result.planning_counters['manifest_entries_pruned_by_data_stats']
== (0 if limit == 0 else 1)
+ assert 'Manifest entries decoded' in str(result)
+ assert json.loads(json.dumps(result.planning_counters)) ==
result.planning_counters
+ assert result.split_count == len(builder.new_scan().plan().splits())
+ assert result.splits is not None
+
+
+def
test_native_explain_reports_actual_bucket_then_partition_pruning(rest_catalog):
+ catalog, _ = rest_catalog
+ schema = pa.schema([('pt', pa.int32()), ('id', pa.int32()), ('value',
pa.int32())])
+ table, _ = _create(catalog, schema, ['pt'], ['pt', 'id'], {'bucket': '4',
'bucket-key': 'id'})
+ _append(table, [{'pt': pt, 'id': i, 'value': i} for pt in [1, 2] for i in
range(64)], schema)
+
+ def explain(native):
+ selected = table.copy({'scan.native-plan.enabled':
str(native).lower()}).new_read_builder()
+ pb = selected.new_predicate_builder()
+ selected.with_filter(pb.and_predicates([pb.equal('pt', 1),
pb.equal('id', 7)]))
+ return _native_explain(selected) if native else selected.explain()
+
+ expected, actual = explain(False), explain(True)
+ for name in ['partition_pruning', 'bucket_pruning', 'file_skipping']:
+ assert getattr(actual, name) is None
+ assert actual.planning_counters['manifest_entries_read'] == 8
+ # A partition-specialized bucket selector conservatively keeps the other
+ # partition, whose four entries are rejected by the partition filter.
+ assert actual.planning_counters['manifest_entries_pruned_by_bucket'] == 3
+ assert actual.planning_counters['manifest_entries_pruned_by_partition'] ==
4
+ assert actual.file_count == expected.file_count == 1
+
+
[email protected]('limit,expected_files', [(None, 3), (0, 0), (1, 1)])
+def
test_native_explain_preserves_limit_stopping_without_row_filter(rest_catalog,
limit, expected_files):
+ catalog, _ = rest_catalog
+ table, schema = _create(catalog)
+ for start in range(3):
+ _append(table, [{'id': start, 'value': start}], schema)
+ reader = table.new_read_builder()
+ if limit is not None:
+ reader.with_limit(limit)
+ scan = reader.new_scan()
+ with patch.object(scan.file_scanner, 'scan_with_stats',
side_effect=AssertionError('Python stats ran')):
+ plan, stats = scan.scan_with_stats()
+ assert stats.entries_after_stats is None
+ assert stats.planning_counters['final_files'] == expected_files
+ assert pickle.loads(pickle.dumps(stats)) == stats
+ if limit is not None:
+ assert stats.planning_counters['limit_early_stopped']
+ assert stats.planning_counters['split_candidates_built'] == limit
+ assert len(plan.splits()) == expected_files
+ assert [split.file_paths for split in plan.splits()] == [
+ split.file_paths for split in reader.new_scan().plan().splits()]
+
+
[email protected]('selection', ['whole_shard', 'partial_shard',
'whole_slice', 'partial_slice'])
[email protected]('limit', [1, 2])
+def test_native_statistics_follow_distributed_limit(rest_catalog, selection,
limit):
+ catalog, _ = rest_catalog
+ table, schema = _create(catalog)
+ for value in range(3):
+ _append(table, [{'id': value, 'value': value}], schema)
+ reader = table.new_read_builder().with_limit(limit)
+ scan = reader.new_scan()
+ if selection.endswith('shard'):
+ scan.with_shard(0, 1 if selection == 'whole_shard' else 2)
+ else:
+ scan.with_slice(0 if selection == 'whole_slice' else 1, 3)
+ with patch('pypaimon.read.scan_stats.ScanStats.from_native',
+ side_effect=AssertionError('planning eagerly converted Native
statistics')):
+ plan = scan.plan()
+ stats = plan.stats()
+ assert stats.native_planned
+ counters = stats.planning_counters
+ assert counters['manifest_entries_read'] ==
counters['split_candidates_built'] == 3
+ assert counters['splits_before_limit'] == 3
+ assert counters['final_splits'] == counters['splits_after_limit'] ==
len(plan.splits()) == limit
+ assert counters['final_files'] == sum(len(split.files) for split in
plan.splits()) == limit
+ assert counters['planned_data_file_bytes'] == sum(split.file_size for
split in plan.splits())
+ assert counters['limit'] == limit
+ assert not counters['limit_early_stopped'] # Native built all candidates
before Python applied LIMIT.
+ assert plan.stats() is stats
+ assert pickle.loads(pickle.dumps(plan)).stats() == stats
+ assert len(reader.new_read().to_arrow(plan.splits())) == limit
+
+
[email protected]('window', ['future', 'equal'])
[email protected]('limit', [None, 1])
+def test_native_empty_incremental_window_has_zero_statistics(rest_catalog,
window, limit):
+ catalog, _ = rest_catalog
+ table, schema = _create(catalog)
+ _append(table, [{'id': 0, 'value': 0}], schema)
+ timestamp = table.snapshot_manager().get_latest_snapshot().time_millis
+ bounds = (timestamp + 1000, timestamp + 2000) if window == 'future' else
(timestamp, timestamp)
+ selected = table.copy({'incremental-between-timestamp':
'{},{}'.format(*bounds)})
+ reader = selected.new_read_builder()
+ if limit is not None:
+ reader.with_limit(limit)
+ scan = reader.new_scan()
+ with patch.object(scan.file_scanner, 'scan_with_stats',
side_effect=AssertionError('Python stats ran')):
+ plan, stats = scan.scan_with_stats()
+ assert plan.splits() == []
+ assert plan.snapshot_id is None
+ assert stats.native_planned
+ expected = plan.stats().planning_counters
+ reference = table._copy_with_snapshot(None).new_read_builder()
+ if limit is not None:
+ reference.with_limit(limit)
+ assert expected == reference.new_scan().plan().stats().planning_counters
+ assert expected['final_files'] == expected['manifest_entries_read'] == 0
+ assert expected['limit'] == limit
+ assert expected['limit_early_stopped'] is False
+ for name in ['entries_potential_total', 'entries_after_partition',
'entries_after_bucket',
+ 'entries_after_stats', 'partition_keys_before',
'partition_keys_after',
+ 'buckets_seen', 'buckets_after_pruning']:
+ assert getattr(stats, name) is None
+ explanation = _native_explain(reader)
+ assert explanation.planning_counters == expected
+ assert 'Native planning counters' in str(explanation)
+ assert pickle.loads(pickle.dumps(plan)).stats() == stats
+
+
+def test_native_explain_records_cross_schema_overwrite_netting(rest_catalog):
+ from pypaimon.schema.data_types import AtomicType
+ from pypaimon.schema.schema_change import SchemaChange
+ catalog, _ = rest_catalog
+ table, schema = _create(catalog)
+ _append(table, [{'id': 1, 'value': 1}], schema)
+ catalog.alter_table('default.scan_stats',
[SchemaChange.add_column('extra', AtomicType('INT'))], False)
+ table = catalog.get_table('default.scan_stats')
+ schema = schema.append(pa.field('extra', pa.int32()))
+ _append(table, [{'id': 2, 'value': 100, 'extra': 3}], schema,
overwrite=True)
+ outcomes = []
+ for native in [False, True]:
+ reader = table.copy({'scan.native-plan.enabled':
str(native).lower()}).new_read_builder()
+
reader.with_filter(reader.new_predicate_builder().greater_than('value', 50))
+ outcomes.append(_native_explain(reader) if native else
reader.explain())
+ expected, actual = outcomes
+ assert actual.file_count == expected.file_count == 1
+ assert actual.file_skipping is None
+ assert actual.planning_counters['manifest_entries_read'] == 3
+ assert actual.planning_counters['manifest_entries_after_merge'] == 1
+
+
+def
test_native_explain_keeps_empty_and_historical_snapshot_metadata(rest_catalog):
+ catalog, _ = rest_catalog
+ table, schema = _create(catalog)
+ empty = table._copy_with_snapshot(None)
+ result = _native_explain(empty.new_read_builder())
+ assert result.snapshot_id is None
+ assert result.file_count == 0
+ _append(table, [{'id': 1, 'value': 10}], schema)
+ first = table.snapshot_manager().get_latest_snapshot()
+ historical = table._copy_with_snapshot(first)
+ _append(table, [{'id': 2, 'value': 20}], schema)
+ result = _native_explain(historical.new_read_builder())
+ assert result.snapshot_id == 1
+ assert result.file_count == 1
+ assert _native_explain(empty.new_read_builder()).file_count == 0
+ builder = table.new_read_builder()
+ builder.with_filter(builder.new_predicate_builder().greater_than('value',
1000))
+ result = _native_explain(builder)
+ assert result.snapshot_id == 2
+ assert result.file_count == 0
+ assert result.file_skipping is None
+ assert result.planning_counters['manifest_entries_pruned_by_data_stats']
== 2
+
+
+def test_native_explain_does_not_open_data_files(rest_catalog):
+ catalog, _ = rest_catalog
+ table, schema = _create(catalog)
+ messages = _append(table, [{'id': 1, 'value': 10}], schema)
+ with ExitStack() as cleanup:
+ for message in messages:
+ for file in message.new_files:
+ path = LocalFileIO()._to_file(canonical_data_file_path(
+ table, message.partition, message.bucket, file.file_name))
+ hidden = path.with_suffix('.unavailable')
+ path.rename(hidden)
+ cleanup.callback(hidden.rename, path)
+ result = _native_explain(table.new_read_builder())
+ assert result.file_count == 1
+ assert result.estimated_row_count == 1
+
+
+def
test_native_stats_are_readonly_and_normal_planning_stays_lightweight(rest_catalog):
+ catalog, _ = rest_catalog
+ table, schema = _create(catalog)
+ _append(table, [{'id': i, 'value': i} for i in range(3)], schema)
+ scan = _native_table(table).new_read_builder().new_scan()
+ plan = scan.plan()
+ stats = plan.stats()
+ assert stats.manifest_entries_read ==
stats.manifest_entries_after_manifest_filters == 1
+ assert plan.snapshot_id() == 1
+ with pytest.raises(AttributeError):
+ stats.manifest_entries_read = 100
+ with patch('pypaimon.read.scan_stats.ScanStats.from_native',
+ side_effect=AssertionError('ordinary plan requested
statistics')):
+ ordinary = table.new_read_builder().new_scan().plan()
+ assert ordinary.snapshot_id == 1
+ converted = ordinary.stats()
+ assert converted.planning_counters['manifest_entries_read'] == 1
+ assert converted.planning_counters['final_files'] == 1
+ assert ordinary.stats() is converted
+ # An untouched Native handle must be converted during serialization too.
+ restored =
pickle.loads(pickle.dumps(table.new_read_builder().new_scan().plan()))
+ assert restored.stats() == converted
+
+
[email protected]('changelog', [False, True])
+def
test_native_incremental_explain_reports_only_the_selected_window(rest_catalog,
changelog):
+ catalog, _ = rest_catalog
+ table, schema = _create(catalog, primary_keys=['id'] if changelog else
None,
+ options={'bucket': '1', 'changelog-producer':
'input'} if changelog else {})
+ for index, timestamp in enumerate([1000, 2000, 3000]):
+ _append(table, [{'id': index, 'value': timestamp}], schema)
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ path = table.snapshot_manager().snapshot_dir + '/snapshot-' +
str(snapshot.id)
+ with table.file_io.new_input_stream(path) as stream:
+ data = json.loads(stream.read())
+ data['timeMillis'] = timestamp
+ table.file_io.overwrite_file_utf8(path, json.dumps(data))
+ selected = table.copy({'incremental-between-timestamp': '1000,3000'})
+ result = _native_explain(selected.new_read_builder(), True)
+ assert result.snapshot_id == 3
+ assert result.file_count == 2
+ scan = selected.new_read_builder().new_scan()
+ with patch.object(scan.file_scanner, 'scan_with_stats',
side_effect=AssertionError('Python stats ran')):
+ plan, stats = scan.scan_with_stats()
+ assert stats.native_planned
+ assert stats.planning_counters['manifest_entries_read'] ==
stats.planning_counters['final_files'] == 2
+ assert plan.snapshot_id == 3
+ assert all(split.is_streaming for split in plan.splits())
+ assert all(file.file_name.startswith('changelog-') == changelog
+ for split in plan.splits() for file in split.files)
+
+
[email protected]('limit', [None, 0, 1])
+def
test_native_data_evolution_explain_tracks_projection_with_limit(rest_catalog,
limit):
+ catalog, _ = rest_catalog
+ table, schema = _create(catalog, options={
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true'})
+ _append(table, [{'id': i, 'value': i} for i in range(3)], schema)
+ builder = table.new_batch_write_builder()
+ updater = builder.new_update().new_update_by_row_id()
+ messages = updater.update_columns(pa.table({'_ROW_ID': [0, 1, 2], 'value':
pa.array([10, 11, 12], pa.int32())}),
+ ['value'])
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages)
+ finally:
+ commit.close()
+ reader = table.new_read_builder().with_projection(['id'])
+ if limit is not None:
+ reader.with_limit(limit)
+ scan = reader.new_scan()
+ with patch.object(scan.file_scanner, 'scan_with_stats',
side_effect=AssertionError('Python stats ran')):
+ plan, stats = scan.scan_with_stats()
+ assert stats.native_planned
+ assert stats.entries_potential_total is None
+ assert stats.planning_counters['manifest_entries_read'] == (0 if limit ==
0 else 2)
+ assert stats.planning_counters['final_files'] == (0 if limit == 0 else 1)
+ assert len(plan.splits()) == (0 if limit == 0 else 1)
+
+
+def
test_native_explain_preserves_the_authorized_plan_and_captured_policy(guarded_table):
+ import json
+ from pypaimon.api.api_response import AuthTableQueryResponse
+ from pypaimon.common.predicate_builder import PredicateBuilder
+ table, server = guarded_table
+ server.table_query_auth_store[table.identifier.get_full_name()] =
AuthTableQueryResponse(
+ filter=[native_query_auth_test._filter('id', 'GREATER_THAN', [1])],
+ column_masking={'value': json.dumps({
+ 'name': 'SUBSTRING', 'inputs':
[native_query_auth_test._field('source'), 1, 2]})})
+ table = table.copy({'scan.native-plan.enabled': 'true',
'read.native.enabled': 'true'})
+ reader = table.new_read_builder().with_projection(['id', 'value'])
+ reader.with_filter(PredicateBuilder(table.fields).equal('value',
'ab')).with_limit(1)
+ scan = reader.new_scan()
+ with patch.object(scan.file_scanner, 'scan_with_stats',
side_effect=AssertionError('Python stats ran')):
+ plan, stats = scan.scan_with_stats()
+ assert stats.native_planned
+ assert plan.splits()
+ assert all(getattr(split, '_native_split', None) is not None for split in
plan.splits())
+ server.table_query_auth_store[table.identifier.get_full_name()] =
AuthTableQueryResponse()
+ with patch('pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError('Native authorized read fell back')):
+ assert reader.new_read().to_arrow(plan.splits()).to_pylist() ==
[{'id': 2, 'value': 'ab'}]
diff --git a/paimon-python/pypaimon/tests/pushdown_bucket_test.py
b/paimon-python/pypaimon/tests/pushdown_bucket_test.py
index 8f3c0814cc..0e2ddd74c2 100644
--- a/paimon-python/pypaimon/tests/pushdown_bucket_test.py
+++ b/paimon-python/pypaimon/tests/pushdown_bucket_test.py
@@ -524,11 +524,10 @@ class
PartitionAwareBucketSelectorUnitTest(unittest.TestCase):
self.assertTrue(sel(None, b, 8))
self.assertTrue(sel.may_contain(b, b, 8))
- def test_selector_partition_not_matching_returns_empty_bucket_set(self):
+ def test_selector_partition_not_matching_keeps_java_bucket_fallback(self):
# ``part = 'a' AND id = 1`` on partition {part: 'c'} simplifies to
- # AlwaysFalse — the selector returns False for every bucket since
- # no row in this partition can possibly match. Sound: dropping a
- # partition that *can't* contain matches doesn't lose data.
+ # AlwaysFalse. Java cannot extract bucket-key values and falls back
+ # to all buckets; the independent partition filter rejects this row.
pred = PredicateBuilder.and_predicates([
self.pb.equal('part', 'a'),
self.pb.equal('id', 1),
@@ -538,9 +537,15 @@ class
PartitionAwareBucketSelectorUnitTest(unittest.TestCase):
self.assertIsNotNone(sel)
part_c = GenericRow(['c'], [self.part_field], RowKind.INSERT)
for b in range(8):
- self.assertFalse(sel(part_c, b, 8),
- "partition c can't satisfy part='a', "
- "drop every bucket (b=%d)" % b)
+ self.assertTrue(sel(part_c, b, 8))
+
+ def test_pending_buckets_are_never_pruned_by_key_hash(self):
+ pred = self.pb.equal('id', 7)
+ sel = create_bucket_selector(pred, [self.id_field],
partition_fields=[self.part_field])
+ part = GenericRow(['a'], [self.part_field], RowKind.INSERT)
+ for bucket in [-2, -1]:
+ self.assertTrue(sel(part, bucket, 8))
+ self.assertTrue(sel(bucket, 8))
def test_selector_partition_only_constraint_drops_partition(self):
# ``part='a' AND id IN (1,2)`` — same partition value 'a'
diff --git a/paimon-python/pypaimon/tests/table_query_auth_test.py
b/paimon-python/pypaimon/tests/table_query_auth_test.py
index 56ec0ddb94..9a5c304e8b 100644
--- a/paimon-python/pypaimon/tests/table_query_auth_test.py
+++ b/paimon-python/pypaimon/tests/table_query_auth_test.py
@@ -18,6 +18,7 @@
import json
import unittest
+from unittest.mock import patch
import pyarrow as pa
@@ -28,7 +29,9 @@ from pypaimon.catalog.table_query_auth import (
from pypaimon.common.identifier import Identifier
from pypaimon.common.options import Options
from pypaimon.common.options.core_options import CoreOptions
+from pypaimon.read.plan import Plan
from pypaimon.read.query_auth_split import QueryAuthSplit
+from pypaimon.read.scan_stats import ScanStats
class _FakeField:
@@ -60,15 +63,6 @@ class _FakeSplit:
return 100
-class _FakePlan:
- def __init__(self, splits, snapshot_id=None):
- self._splits = splits
- self.snapshot_id = snapshot_id
-
- def splits(self):
- return self._splits
-
-
class _FakeTable:
def __init__(self, identifier=None):
self.identifier = identifier or Identifier("db", "table")
@@ -124,7 +118,7 @@ class
TestTableQueryAuthResultConvertPlan(unittest.TestCase):
def test_no_auth_returns_original_plan(self):
result = TableQueryAuthResult(None, None)
- plan = _FakePlan([_FakeSplit()])
+ plan = Plan([_FakeSplit()])
converted = result.convert_plan(plan)
self.assertIs(converted, plan)
@@ -134,7 +128,7 @@ class
TestTableQueryAuthResultConvertPlan(unittest.TestCase):
def test_empty_filter_and_masking_returns_original(self):
result = TableQueryAuthResult([], {})
- plan = _FakePlan([_FakeSplit()])
+ plan = Plan([_FakeSplit()])
converted = result.convert_plan(plan)
self.assertIs(converted, plan)
@@ -177,22 +171,42 @@ class
TestTableQueryAuthResultConvertPlan(unittest.TestCase):
def test_wraps_splits_with_filter(self):
result = TableQueryAuthResult([_simple_filter_json()], None)
- plan = _FakePlan([_FakeSplit("s1"), _FakeSplit("s2")])
+ stats = ScanStats(entries_after_stats=2)
+ plan = Plan([_FakeSplit("s1"), _FakeSplit("s2")], _scan_stats=stats)
converted = result.convert_plan(plan)
+ self.assertIs(converted.stats(), stats)
self.assertEqual(len(converted.splits()), 2)
for qs in converted.splits():
self.assertIsInstance(qs, QueryAuthSplit)
+ def test_auth_wrapping_preserves_lazy_native_statistics(self):
+ result = TableQueryAuthResult([_simple_filter_json()], None)
+ handle = object()
+ plan = Plan([_FakeSplit("s1")], snapshot_id=7, _native_stats=handle)
+ counters = ScanStats.empty_native()
+ counters.entries_total = 10
+ counters.planning_counters['manifest_entries_read'] = 10
+ with patch.object(ScanStats, 'from_native', return_value=counters) as
convert:
+ wrapped = result.convert_plan(plan)
+ convert.assert_not_called()
+ self.assertEqual(wrapped.snapshot_id, 7)
+ self.assertIsInstance(wrapped.splits()[0], QueryAuthSplit)
+ self.assertIs(wrapped.stats(), counters)
+ self.assertIs(wrapped.stats(), counters)
+
self.assertEqual(counters.planning_counters['manifest_entries_read'], 10)
+ self.assertEqual(counters.planning_counters['final_splits'], 1)
+ convert.assert_called_once_with(handle)
+
def test_wraps_splits_with_masking(self):
result = TableQueryAuthResult(None, {"col": '{"name":"NULL"}'})
- plan = _FakePlan([_FakeSplit()])
+ plan = Plan([_FakeSplit()])
converted = result.convert_plan(plan)
self.assertEqual(len(converted.splits()), 1)
self.assertIsInstance(converted.splits()[0], QueryAuthSplit)
def test_inner_split_preserved(self):
result = TableQueryAuthResult([_simple_filter_json()], None)
- plan = _FakePlan([_FakeSplit("original")])
+ plan = Plan([_FakeSplit("original")])
converted = result.convert_plan(plan)
self.assertEqual(converted.splits()[0].split.name, "original")
@@ -370,7 +384,7 @@ class TestPlanForWrite(unittest.TestCase):
def test_no_restriction_returns_plan_unchanged(self):
from pypaimon.read.table_scan import TableScan
- plain_plan = _FakePlan([_FakeSplit()], snapshot_id=5)
+ plain_plan = Plan([_FakeSplit()], snapshot_id=5)
scan = _FakeScan(plain_plan, auth_result=None)
result = TableScan.plan_for_write(scan)
self.assertIs(result, plain_plan)
@@ -379,7 +393,7 @@ class TestPlanForWrite(unittest.TestCase):
from pypaimon.read.table_scan import TableScan
auth = TableQueryAuthResult([_simple_filter_json()], None)
- plan = _FakePlan([_FakeSplit()], snapshot_id=5)
+ plan = Plan([_FakeSplit()], snapshot_id=5)
scan = _FakeScan(plan, auth_result=auth)
with self.assertRaises(TableNoPermissionException):
TableScan.plan_for_write(scan)
@@ -388,7 +402,7 @@ class TestPlanForWrite(unittest.TestCase):
from pypaimon.read.table_scan import TableScan
auth = TableQueryAuthResult(None, {"col": '{"name":"NULL"}'})
- plan = _FakePlan([_FakeSplit()], snapshot_id=5)
+ plan = Plan([_FakeSplit()], snapshot_id=5)
scan = _FakeScan(plan, auth_result=auth)
with self.assertRaises(TableNoPermissionException):
TableScan.plan_for_write(scan)
@@ -397,7 +411,7 @@ class TestPlanForWrite(unittest.TestCase):
from pypaimon.read.table_scan import TableScan
auth = TableQueryAuthResult([_simple_filter_json()], None)
- empty_plan = _FakePlan([], snapshot_id=5)
+ empty_plan = Plan([], snapshot_id=5)
scan = _FakeScan(empty_plan, auth_result=auth)
with self.assertRaises(TableNoPermissionException):
TableScan.plan_for_write(scan)
@@ -405,7 +419,7 @@ class TestPlanForWrite(unittest.TestCase):
def test_no_restrictions_allows_write(self):
from pypaimon.read.table_scan import TableScan
- plan = _FakePlan([_FakeSplit()], snapshot_id=5)
+ plan = Plan([_FakeSplit()], snapshot_id=5)
scan = _FakeScan(plan, auth_result=None)
result = TableScan.plan_for_write(scan)
self.assertIs(result, plan)
