| # 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. | |
| """Tests for ``ReadBuilder.explain``: pruning funnel counters, | |
| split-level execution signals, and pretty-print smoke.""" | |
| import os | |
| import shutil | |
| import tempfile | |
| import json | |
| import unittest | |
| from typing import Any, Dict, List | |
| import pyarrow as pa | |
| import pytest | |
| from pypaimon import CatalogFactory, Schema | |
| from pypaimon.common.predicate import Predicate | |
| from pypaimon.read.explain import ExplainResult | |
| from pypaimon.read.explain_render import render_predicate | |
| def _write(table, rows: List[Dict], pa_schema: pa.Schema) -> None: | |
| wb = table.new_batch_write_builder() | |
| w = wb.new_write() | |
| c = wb.new_commit() | |
| try: | |
| w.write_arrow(pa.Table.from_pylist(rows, schema=pa_schema)) | |
| c.commit(w.prepare_commit()) | |
| finally: | |
| w.close() | |
| c.close() | |
| class ReadBuilderExplainTest(unittest.TestCase): | |
| @classmethod | |
| def setUpClass(cls): | |
| cls.tempdir = tempfile.mkdtemp() | |
| cls.warehouse = os.path.join(cls.tempdir, 'warehouse') | |
| cls.catalog = CatalogFactory.create({'warehouse': cls.warehouse}) | |
| cls.catalog.create_database('default', False) | |
| @classmethod | |
| def tearDownClass(cls): | |
| shutil.rmtree(cls.tempdir, ignore_errors=True) | |
| # ---- helpers -------------------------------------------------------- | |
| def _append_table(self, name: str) -> Any: | |
| pa_schema = pa.schema([ | |
| pa.field('id', pa.int64(), nullable=False), | |
| ('val', pa.int64()), | |
| ]) | |
| schema = Schema.from_pyarrow_schema( | |
| pa_schema, | |
| options={'bucket': '-1', 'file.format': 'parquet'}, | |
| ) | |
| full = 'default.{}'.format(name) | |
| self.catalog.create_table(full, schema, False) | |
| return self.catalog.get_table(full), pa_schema | |
| def _pk_partitioned_bucketed_table(self, name: str, num_buckets: int = 4) -> Any: | |
| pa_schema = pa.schema([ | |
| pa.field('dt', pa.string(), nullable=False), | |
| pa.field('id', pa.int64(), nullable=False), | |
| ('val', pa.int64()), | |
| ]) | |
| schema = Schema.from_pyarrow_schema( | |
| pa_schema, | |
| partition_keys=['dt'], | |
| primary_keys=['dt', 'id'], | |
| options={ | |
| 'bucket': str(num_buckets), | |
| 'bucket-key': 'id', | |
| 'file.format': 'parquet', | |
| }, | |
| ) | |
| full = 'default.{}'.format(name) | |
| self.catalog.create_table(full, schema, False) | |
| return self.catalog.get_table(full), pa_schema | |
| def _pk_dv_table(self, name: str, num_buckets: int = 2) -> Any: | |
| pa_schema = pa.schema([ | |
| pa.field('id', pa.int64(), nullable=False), | |
| ('val', pa.int64()), | |
| ]) | |
| schema = Schema.from_pyarrow_schema( | |
| pa_schema, | |
| primary_keys=['id'], | |
| options={ | |
| 'bucket': str(num_buckets), | |
| 'file.format': 'parquet', | |
| 'deletion-vectors.enabled': 'true', | |
| }, | |
| ) | |
| full = 'default.{}'.format(name) | |
| self.catalog.create_table(full, schema, False) | |
| return self.catalog.get_table(full), pa_schema | |
| # ---- 1. append-only baseline --------------------------------------- | |
| def test_explain_append_only_no_predicate(self): | |
| table, pa_schema = self._append_table('explain_basic') | |
| _write(table, [{'id': i, 'val': i * 2} for i in range(20)], pa_schema) | |
| _write(table, [{'id': i, 'val': i * 3} for i in range(20, 40)], pa_schema) | |
| rb = table.new_read_builder() | |
| result = rb.explain() | |
| plan_splits = rb.new_scan().plan().splits() | |
| self.assertEqual(result.split_count, len(plan_splits)) | |
| self.assertGreater(result.file_count, 0) | |
| self.assertGreater(result.total_file_size, 0) | |
| self.assertIsNone(result.partition_pruning) | |
| self.assertIsNone(result.bucket_pruning) | |
| self.assertIsNone(result.file_skipping) | |
| self.assertIsNone(result.predicate) | |
| # ---- 2. PK + partition + bucket pruning ---------------------------- | |
| @pytest.mark.python_plan | |
| def test_explain_pk_table_with_partition_and_bucket_predicate(self): | |
| table, pa_schema = self._pk_partitioned_bucketed_table( | |
| 'explain_pk_pruning', num_buckets=4) | |
| for day in ['2026-05-01', '2026-05-02', '2026-05-03', '2026-05-04']: | |
| _write( | |
| table, | |
| [{'dt': day, 'id': i, 'val': i + 1} for i in range(20)], | |
| pa_schema, | |
| ) | |
| rb = table.new_read_builder() | |
| pb = rb.new_predicate_builder() | |
| pred = pb.and_predicates([ | |
| pb.equal('dt', '2026-05-01'), | |
| pb.equal('id', 7), | |
| ]) | |
| rb = rb.with_filter(pred) | |
| result = rb.explain() | |
| self.assertIsNotNone(result.partition_pruning) | |
| self.assertIsNotNone(result.bucket_pruning) | |
| self.assertIsNotNone(result.file_skipping) | |
| self.assertGreater( | |
| result.partition_pruning.before, result.partition_pruning.after, | |
| "partition predicate must drop at least one entry") | |
| self.assertGreater( | |
| result.bucket_pruning.before, result.bucket_pruning.after, | |
| "HASH_FIXED bucket pruning must drop at least one entry") | |
| # Cross-check against the actual scan | |
| plan_splits = rb.new_scan().plan().splits() | |
| self.assertEqual(result.split_count, len(plan_splits)) | |
| # ---- 3. predicate rendering ---------------------------------------- | |
| def test_render_predicate_shapes(self): | |
| eq = Predicate(method='equal', index=0, field='dt', literals=['2026-05-01']) | |
| self.assertEqual(render_predicate(eq), "dt = '2026-05-01'") | |
| in_p = Predicate(method='in', index=1, field='id', literals=[1, 2, 3]) | |
| self.assertEqual(render_predicate(in_p), "id IN [1, 2, 3]") | |
| between = Predicate(method='between', index=1, field='id', literals=[5, 10]) | |
| self.assertEqual(render_predicate(between), "id BETWEEN 5 AND 10") | |
| is_null = Predicate(method='isNull', index=0, field='val', literals=None) | |
| self.assertEqual(render_predicate(is_null), "val IS NULL") | |
| and_p = Predicate(method='and', index=None, field=None, literals=[eq, in_p]) | |
| self.assertEqual( | |
| render_predicate(and_p), | |
| "(dt = '2026-05-01') AND (id IN [1, 2, 3])") | |
| or_p = Predicate(method='or', index=None, field=None, literals=[between, is_null]) | |
| self.assertEqual( | |
| render_predicate(or_p), | |
| "(id BETWEEN 5 AND 10) OR (val IS NULL)") | |
| # ---- 4. verbose split detail --------------------------------------- | |
| def test_explain_verbose_lists_per_split_detail(self): | |
| table, pa_schema = self._append_table('explain_verbose') | |
| _write(table, [{'id': i, 'val': i} for i in range(50)], pa_schema) | |
| _write(table, [{'id': i, 'val': i} for i in range(50, 100)], pa_schema) | |
| rb = table.new_read_builder() | |
| result = rb.explain(verbose=True) | |
| self.assertIsNotNone(result.splits) | |
| self.assertEqual(len(result.splits), result.split_count) | |
| plan_splits = rb.new_scan().plan().splits() | |
| for explained, actual in zip(result.splits, plan_splits): | |
| self.assertEqual(explained.bucket, actual.bucket) | |
| self.assertEqual(explained.file_count, len(actual.files)) | |
| self.assertEqual(set(explained.file_paths), set(actual.file_paths)) | |
| # ---- 5. empty snapshot path ---------------------------------------- | |
| def test_explain_empty_snapshot(self): | |
| table, _ = self._append_table('explain_empty') | |
| rb = table.new_read_builder() | |
| result = rb.explain() | |
| self.assertIsNone(result.snapshot_id) | |
| self.assertEqual(result.split_count, 0) | |
| self.assertEqual(result.file_count, 0) | |
| self.assertIn("<none>", str(result)) | |
| # ---- 6. split-level metrics: DV vs append-only --------------------- | |
| def test_explain_split_level_metrics(self): | |
| # Append-only: every split is raw-convertible, no deletion vectors. | |
| table, pa_schema = self._append_table('explain_split_signals_ap') | |
| _write(table, [{'id': i, 'val': i} for i in range(30)], pa_schema) | |
| _write(table, [{'id': i, 'val': i} for i in range(30, 60)], pa_schema) | |
| result = table.new_read_builder().explain() | |
| self.assertGreater(result.split_count, 0) | |
| self.assertEqual(result.splits_with_deletion_vectors, 0) | |
| self.assertEqual(result.splits_raw_convertible, result.split_count) | |
| # All written files are at L0, and append-only doesn't filter them. | |
| self.assertIn(0, result.level_histogram) | |
| self.assertEqual(result.splits_all_above_l0, 0) | |
| # DV-enabled PK table: pypaimon writes alone don't trigger compaction, | |
| # so every file stays at L0. ``_filter_manifest_entry`` then drops | |
| # them all, leaving an empty plan. The vacuous "all above L0" | |
| # invariant (0 / 0) still has to hold. | |
| dv_table, dv_pa = self._pk_dv_table('explain_split_signals_dv', num_buckets=2) | |
| _write(dv_table, [{'id': i, 'val': i} for i in range(20)], dv_pa) | |
| _write(dv_table, [{'id': i, 'val': i * 10} for i in range(10)], dv_pa) | |
| dv_result = dv_table.new_read_builder().explain() | |
| self.assertEqual(dv_result.split_count, 0) | |
| self.assertEqual(dv_result.splits_all_above_l0, 0) | |
| # ---- 8. query auth wraps splits and is surfaced in explain --------- | |
| def test_explain_reports_query_auth_split(self): | |
| from pypaimon.catalog.table_query_auth import TableQueryAuthResult | |
| table, pa_schema = self._append_table('explain_query_auth') | |
| _write(table, [{'id': i, 'val': i} for i in range(20)], pa_schema) | |
| auth_result = TableQueryAuthResult(filter=None, column_masking={'val': json.dumps({'name': 'NULL'})}) | |
| table.catalog_environment.table_query_auth = lambda options, identifier: (lambda select: auth_result) | |
| rb = table.new_read_builder() | |
| result = rb.explain(verbose=True) | |
| self.assertIsNotNone(result.splits) | |
| self.assertGreater(len(result.splits), 0) | |
| self.assertTrue(result.has_auth) | |
| plan_splits = rb.new_scan().plan().splits() | |
| from pypaimon.read.query_auth_split import QueryAuthSplit | |
| self.assertTrue(all(isinstance(s, QueryAuthSplit) for s in plan_splits)) | |
| # ---- 9. pretty-print smoke ----------------------------------------- | |
| def test_pretty_print_smoke(self): | |
| table, pa_schema = self._append_table('explain_print_smoke') | |
| _write(table, [{'id': i, 'val': i} for i in range(40)], pa_schema) | |
| result = table.new_read_builder().explain() | |
| self.assertIsInstance(result, ExplainResult) | |
| printed = str(result) | |
| for anchor in ( | |
| "Snapshot:", | |
| "Splits:", | |
| "raw-convertible:", | |
| "with DV:", | |
| "size/split:", | |
| "Files:", | |
| "Total size:", | |
| ): | |
| self.assertIn(anchor, printed, "missing anchor: " + anchor) | |
| if __name__ == '__main__': | |
| unittest.main() |