| # 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. | |
| import logging | |
| import os | |
| import time | |
| from typing import Callable, Dict, List, NamedTuple, Optional, Set, Tuple | |
| logger = logging.getLogger(__name__) | |
| from pypaimon.common.predicate import Predicate | |
| from pypaimon.globalindex import ScoredGlobalIndexResult | |
| from pypaimon.globalindex.global_index_result import GlobalIndexResult | |
| from pypaimon.manifest.index_manifest_file import IndexManifestFile | |
| from pypaimon.manifest.manifest_file_manager import ManifestFileManager | |
| from pypaimon.manifest.manifest_list_manager import ManifestListManager | |
| from pypaimon.manifest.schema.manifest_entry import ManifestEntry | |
| from pypaimon.manifest.schema.manifest_file_meta import ManifestFileMeta | |
| from pypaimon.manifest.simple_stats_evolutions import SimpleStatsEvolutions | |
| from pypaimon.schema.data_types import DataField | |
| from pypaimon.read.plan import Plan | |
| from pypaimon.read.push_down_utils import ( | |
| exclude_predicate_with_fields, | |
| predicate_for_physical_fields, | |
| remove_row_id_filter, | |
| trim_and_transform_predicate) | |
| from pypaimon.read.scan_stats import ScanStats | |
| from pypaimon.read.scan_request import ( | |
| apply_push_down_limit, has_non_partition_filter, validate_chunk_shuffle_compat, | |
| normalize_row_ranges, scan_partition_predicate, validate_chunk_shuffle, | |
| validate_global_index_selection, validate_shard_selection, validate_slice_selection) | |
| from pypaimon.execution.python.read.scanner.append_table_split_generator import \ | |
| AppendTableSplitGenerator | |
| from pypaimon.execution.python.read.scanner.bucket_select_converter import \ | |
| create_bucket_selector | |
| from pypaimon.execution.python.read.scanner.chunk_shuffle_split_generator import ( | |
| AppendChunkShuffleSplitGenerator, | |
| DataEvolutionChunkShuffleSplitGenerator, | |
| ) | |
| from pypaimon.execution.python.read.scanner.data_evolution_split_generator import \ | |
| DataEvolutionSplitGenerator | |
| from pypaimon.execution.python.read.scanner.data_evolution_stats import \ | |
| DataEvolutionGroupStatsFilter | |
| from pypaimon.execution.python.read.scanner.primary_key_table_split_generator import \ | |
| PrimaryKeyTableSplitGenerator | |
| from pypaimon.read.split import DataSplit | |
| from pypaimon.snapshot.snapshot import Snapshot | |
| from pypaimon.table.bucket_mode import BucketMode | |
| from pypaimon.table.special_fields import SpecialFields | |
| from pypaimon.table.source.deletion_file import DeletionFile | |
| from pypaimon.utils.range import Range | |
| class _GlobalIndexPlanningResult(NamedTuple): | |
| indexed_result: GlobalIndexResult | |
| unindexed_ranges: List[Range] | |
| def _row_ranges_from_predicate(predicate: Optional[Predicate]) -> Optional[List]: | |
| from pypaimon.table.special_fields import SpecialFields | |
| if predicate is None: | |
| return None | |
| def visit(p: Predicate): | |
| if p.method == 'and': | |
| result = None | |
| for child in p.literals: | |
| sub = visit(child) | |
| if sub is None: | |
| continue | |
| result = Range.and_(result, sub) if result is not None else sub | |
| if not result: | |
| return result | |
| return result | |
| if p.method == 'or': | |
| parts = [] | |
| for child in p.literals: | |
| sub = visit(child) | |
| if sub is None: | |
| return None | |
| parts.extend(sub) | |
| if not parts: | |
| return [] | |
| return Range.sort_and_merge_overlap(parts, merge=True, adjacent=True) | |
| if p.field != SpecialFields.ROW_ID.name: | |
| return None | |
| if p.method == 'equal': | |
| if not p.literals: | |
| return [] | |
| return Range.to_ranges([int(p.literals[0])]) | |
| if p.method == 'in': | |
| if not p.literals: | |
| return [] | |
| return Range.to_ranges([int(x) for x in p.literals]) | |
| if p.method == 'between': | |
| if not p.literals or len(p.literals) < 2: | |
| return [] | |
| return [Range(int(p.literals[0]), int(p.literals[1]))] | |
| return None | |
| return visit(predicate) | |
| def _build_early_row_range_filter(row_ranges): | |
| """Skip entries whose row-id range doesn't intersect ``row_ranges``. | |
| Runs on the raw fastavro record (OrderedDict) before the expensive | |
| Python object construction (BinaryRow, GenericRow, SimpleStats, | |
| DataFileMeta). fastavro has already parsed the Avro bytes; this | |
| filter avoids the construction cost, not I/O or parsing. | |
| Safe for DELETE entries because ADD and DELETE for the same file | |
| share the same ``_FIRST_ROW_ID``. | |
| """ | |
| if row_ranges is None or not row_ranges: | |
| return None | |
| def _filter(record): | |
| file_dict = record.get('_FILE') | |
| if file_dict is None: | |
| return True | |
| first_row_id = file_dict.get('_FIRST_ROW_ID') | |
| if first_row_id is None: | |
| return True | |
| row_count = file_dict.get('_ROW_COUNT') | |
| if row_count is None: | |
| return True | |
| file_start = int(first_row_id) | |
| file_end = file_start + int(row_count) - 1 | |
| for r in row_ranges: | |
| if Range.intersect(file_start, file_end, r.from_, r.to): | |
| return True | |
| return False | |
| return _filter | |
| def _filter_manifest_files_by_row_ranges( | |
| manifest_files: List[ManifestFileMeta], | |
| row_ranges: List) -> List[ManifestFileMeta]: | |
| """ | |
| Filter manifest files by row ranges. | |
| Only keep manifest files that have min_row_id and max_row_id and overlap with the given row ranges. | |
| Args: | |
| manifest_files: List of manifest file metadata | |
| row_ranges: List of row ranges to filter by | |
| Returns: | |
| Filtered list of manifest files | |
| """ | |
| filtered_files = [] | |
| for manifest in manifest_files: | |
| min_row_id = manifest.min_row_id | |
| max_row_id = manifest.max_row_id | |
| # If min_row_id or max_row_id is None, we cannot filter, keep the file | |
| if min_row_id is None or max_row_id is None: | |
| filtered_files.append(manifest) | |
| continue | |
| # Check if manifest row range overlaps with any of the expected row ranges | |
| manifest_row_range = Range(min_row_id, max_row_id) | |
| should_keep = False | |
| for expected_range in row_ranges: | |
| # Check if ranges intersect | |
| intersect = Range.intersect( | |
| manifest_row_range.from_, | |
| manifest_row_range.to, | |
| expected_range.from_, | |
| expected_range.to) | |
| if intersect: | |
| should_keep = True | |
| break | |
| if should_keep: | |
| filtered_files.append(manifest) | |
| return filtered_files | |
| def _filter_manifest_entries_by_row_ranges( | |
| entries: List[ManifestEntry], | |
| row_ranges: List) -> List[ManifestEntry]: | |
| if not row_ranges: | |
| return [] | |
| filtered = [] | |
| for entry in entries: | |
| first_row_id = entry.file.first_row_id | |
| if first_row_id is None: | |
| filtered.append(entry) | |
| continue | |
| file_range = entry.file.row_id_range() | |
| for r in row_ranges: | |
| if file_range.overlaps(r): | |
| filtered.append(entry) | |
| break | |
| return filtered | |
| class FileScanner: | |
| def __init__( | |
| self, | |
| table, | |
| manifest_scanner: Callable[[], Tuple[List[ManifestFileMeta], Optional[Snapshot]]], | |
| predicate: Optional[Predicate] = None, | |
| limit: Optional[int] = None, | |
| partition_predicate: Optional[Predicate] = None, | |
| skip_level0: bool = False, | |
| is_streaming: bool = False, | |
| ): | |
| from pypaimon.table.file_store_table import FileStoreTable | |
| self.table: FileStoreTable = table | |
| self.manifest_scanner = manifest_scanner | |
| self._user_predicate = predicate | |
| self._partition_predicate = partition_predicate | |
| self._explicit_partition_filter = partition_predicate is not None | |
| self.limit = limit | |
| self.snapshot_manager = table.snapshot_manager() | |
| self.manifest_list_manager = ManifestListManager(table) | |
| self.manifest_file_manager = ManifestFileManager(table) | |
| options = self.table.options | |
| # Get split target size and open file cost from table options | |
| self.target_split_size = options.source_split_target_size() | |
| self.open_file_cost = options.source_split_open_file_cost() | |
| self.idx_of_this_subtask = None | |
| self.number_of_para_subtasks = None | |
| self.start_pos_of_this_subtask = None | |
| self.end_pos_of_this_subtask = None | |
| self.chunk_shuffle: Optional[Tuple[int, int]] = None | |
| self.only_read_real_buckets = options.bucket() == BucketMode.POSTPONE_BUCKET.value | |
| self.data_evolution = options.data_evolution_enabled() | |
| self.deletion_vectors_enabled = options.deletion_vectors_enabled() | |
| self.skip_level0 = skip_level0 | |
| self.level_filter: Optional[Callable[[int], bool]] = None | |
| self.is_streaming = is_streaming | |
| self._global_index_result = None | |
| self._row_ranges = None | |
| self._scanned_snapshot = None | |
| self._scanned_snapshot_id = None | |
| # Opt-in scan-plan tracking. Stays ``None`` for the read hot path; | |
| # ``scan_with_stats()`` flips it on for a single explain pass and | |
| # the filter callbacks below increment counters when present. | |
| self.scan_stats: Optional[ScanStats] = None | |
| self.auth_partition_predicate = None | |
| self.auth_has_non_partition_filter = False | |
| self._set_predicate(predicate) | |
| self.simple_stats_evolutions = SimpleStatsEvolutions( | |
| self._schema_fields, | |
| self.table.table_schema.id | |
| ) | |
| def _set_predicate(self, predicate): | |
| self.predicate = predicate | |
| physical = predicate_for_physical_fields(predicate, self.table.fields) | |
| row_ranges = _row_ranges_from_predicate(physical) if physical else None | |
| self.predicate_for_stats = ( | |
| remove_row_id_filter(physical) if row_ranges is not None else physical) | |
| self.predicate_for_stats = exclude_predicate_with_fields( | |
| self.predicate_for_stats, SpecialFields.SYSTEM_FIELD_NAMES | set(self.table.partition_keys)) | |
| self.primary_key_predicate = trim_and_transform_predicate( | |
| physical, self.table.field_names, self.table.trimmed_primary_keys) | |
| self.partition_key_predicate = scan_partition_predicate(self.table, predicate, self._partition_predicate) | |
| self._bucket_selector = self._init_bucket_selector() | |
| def _schema_fields(self, schema_id: int): | |
| """Resolve schema fields, short-circuiting current table schema id to avoid | |
| filesystem access (REST catalog would get 403). | |
| """ | |
| if schema_id == self.table.table_schema.id: | |
| return self.table.table_schema.fields | |
| return self.table.schema_manager.get_schema(schema_id).fields | |
| def _schema(self, schema_id: int): | |
| """Resolve the complete schema including its dedicated-file options.""" | |
| if schema_id == self.table.table_schema.id: | |
| return self.table.table_schema | |
| return self.table.schema_manager.get_schema(schema_id) | |
| def _deletion_files_map(self, entries: List[ManifestEntry]) -> Dict[tuple, Dict[str, DeletionFile]]: | |
| if self.is_streaming or not self.deletion_vectors_enabled: | |
| return {} | |
| # Extract unique partition-bucket pairs from file entries | |
| bucket_files = set() | |
| for e in entries: | |
| bucket_files.add((tuple(e.partition.values), e.bucket)) | |
| snapshot = self._scanned_snapshot if self._scanned_snapshot else self.snapshot_manager.get_latest_snapshot() | |
| return self._scan_dv_index(snapshot, bucket_files) | |
| def scan(self) -> Plan: | |
| start_ms = time.time() * 1000 | |
| if self._global_index_result is not None: | |
| from pypaimon.table.source.global_index_split_result import GlobalIndexSplitResult | |
| if self.table.is_primary_key_table: | |
| if not isinstance(self._global_index_result, GlobalIndexSplitResult): | |
| raise ValueError( | |
| "Primary-key scan requires a GlobalIndexSplitResult, but found %s." | |
| % type(self._global_index_result).__name__) | |
| # Java ignores external raw-index results under query auth: | |
| # masking can invalidate both their positions and their scores. | |
| if not self.table.options.query_auth_enabled: | |
| return Plan( | |
| list(self._global_index_result.splits), | |
| snapshot_id=self._global_index_result.snapshot_id, | |
| ) | |
| # Create appropriate split generator based on table type | |
| if self.chunk_shuffle is not None: | |
| self._validate_chunk_shuffle_compat() | |
| seed, chunk_size = self.chunk_shuffle | |
| # Both append and DE paths use plan_files() directly: the | |
| # predicate is partition-only (enforced by | |
| # _validate_chunk_shuffle_compat), so manifest_entry-level | |
| # partition pruning in plan_files() is the only filter we | |
| # want — no row_id range pushdown, no global index lookup. | |
| entries = self.plan_files() | |
| if self.data_evolution: | |
| split_generator = DataEvolutionChunkShuffleSplitGenerator( | |
| self.table, | |
| self.target_split_size, | |
| self.open_file_cost, | |
| self._deletion_files_map(entries), | |
| seed=seed, | |
| chunk_size=chunk_size, | |
| snapshot_id=self._scanned_snapshot_id, | |
| ) | |
| else: | |
| split_generator = AppendChunkShuffleSplitGenerator( | |
| self.table, | |
| self.target_split_size, | |
| self.open_file_cost, | |
| self._deletion_files_map(entries), | |
| seed=seed, | |
| chunk_size=chunk_size, | |
| snapshot_id=self._scanned_snapshot_id, | |
| ) | |
| elif self.table.is_primary_key_table: | |
| entries = self.plan_files() | |
| split_generator = PrimaryKeyTableSplitGenerator( | |
| self.table, | |
| self.target_split_size, | |
| self.open_file_cost, | |
| self._deletion_files_map(entries), | |
| snapshot_id=self._scanned_snapshot_id, | |
| ) | |
| elif self.data_evolution: | |
| entries, split_generator = self._create_data_evolution_split_generator() | |
| else: | |
| entries = self.plan_files() | |
| split_generator = AppendTableSplitGenerator( | |
| self.table, | |
| self.target_split_size, | |
| self.open_file_cost, | |
| self._deletion_files_map(entries), | |
| snapshot_id=self._scanned_snapshot_id, | |
| ) | |
| if not entries: | |
| return Plan([], snapshot_id=self._scanned_snapshot_id) | |
| # Configure sharding if needed | |
| if self.idx_of_this_subtask is not None: | |
| split_generator.with_shard(self.idx_of_this_subtask, self.number_of_para_subtasks) | |
| elif self.start_pos_of_this_subtask is not None: | |
| split_generator.with_slice(self.start_pos_of_this_subtask, self.end_pos_of_this_subtask) | |
| # Generate splits | |
| splits = split_generator.create_splits(entries) | |
| if self.is_streaming: | |
| for split in splits: | |
| while callable(getattr(split, 'data_split', None)): | |
| split = split.data_split() | |
| split.is_streaming = True | |
| if self.data_evolution and self.scan_stats is not None: | |
| # Data-evolution stats pruning happens on complete row-id groups | |
| # inside the split generator, not in _filter_manifest_entry. | |
| self.scan_stats.entries_after_stats = sum( | |
| len(split.files) for split in splits) | |
| if self.table.is_primary_key_table and not self.is_streaming: | |
| splits = self._apply_primary_key_sorted_indexes(splits) | |
| splits = self._apply_push_down_limit(splits) | |
| duration_ms = int(time.time() * 1000 - start_ms) | |
| logger.info( | |
| "File store scan plan completed in %d ms. Files size: %d", | |
| duration_ms, len(entries) | |
| ) | |
| return Plan(splits, snapshot_id=self._scanned_snapshot_id) | |
| def _apply_primary_key_sorted_indexes(self, splits, snapshot=None): | |
| snapshot = snapshot if snapshot is not None else self._scanned_snapshot | |
| predicate = predicate_for_physical_fields(self.predicate, self.table.fields) | |
| if (not self.table.options.global_index_enabled() | |
| or predicate is None | |
| or snapshot is None | |
| or not splits): | |
| return splits | |
| from pypaimon.index.index_file_handler import IndexFileHandler | |
| from pypaimon.index.pk.primary_key_index_definitions import PrimaryKeyIndexDefinitions | |
| from pypaimon.table.source.primary_key_sorted_index_result import ( | |
| PrimaryKeySortedIndexResult, | |
| ) | |
| from pypaimon.execution.python.read.search import primary_key_sorted_index_scan | |
| definitions = PrimaryKeyIndexDefinitions.create(self.table.table_schema).definitions | |
| if not definitions: | |
| return splits | |
| field_ids = {definition.field_id for definition in definitions} | |
| entries = IndexFileHandler(self.table).scan( | |
| snapshot, | |
| lambda entry: ( | |
| entry.kind == 0 | |
| and entry.index_file.global_index_meta is not None | |
| and entry.index_file.global_index_meta.source_meta is not None | |
| and entry.index_file.global_index_meta.index_field_id in field_ids | |
| ), | |
| ) | |
| index_plan = primary_key_sorted_index_scan.plan( | |
| snapshot.id, splits, definitions, entries) | |
| evaluated = primary_key_sorted_index_scan.evaluate( | |
| index_plan, | |
| self.table.fields, | |
| predicate, | |
| definitions, | |
| primary_key_sorted_index_scan.reader_factory(self.table), | |
| ) | |
| return list(PrimaryKeySortedIndexResult(evaluated).splits) | |
| def _create_data_evolution_split_generator(self): | |
| row_ranges = getattr(self, '_row_ranges', None) | |
| score_getter = None | |
| # Fetch snapshot once and share with global index evaluation to avoid | |
| # a duplicate /snapshot REST round-trip (#7513). | |
| manifest_files, snapshot = self.manifest_scanner() | |
| self._scanned_snapshot = snapshot | |
| self._scanned_snapshot_id = snapshot.id if snapshot else None | |
| if row_ranges is None and not self.table.options.query_auth_enabled: | |
| global_index_plan = self._global_index_result \ | |
| if self._global_index_result is not None \ | |
| else self._eval_global_index(snapshot) | |
| if global_index_plan is not None: | |
| if isinstance(global_index_plan, _GlobalIndexPlanningResult): | |
| global_index_result = global_index_plan.indexed_result | |
| row_ranges = Range.sort_and_merge_overlap( | |
| global_index_result.results().to_range_list() | |
| + global_index_plan.unindexed_ranges, | |
| True, | |
| ) | |
| else: | |
| global_index_result = global_index_plan | |
| row_ranges = global_index_result.results().to_range_list() | |
| if isinstance(global_index_result, ScoredGlobalIndexResult): | |
| score_getter = global_index_result.score_getter() | |
| if row_ranges is None and self.predicate is not None: | |
| row_ranges = _row_ranges_from_predicate( | |
| predicate_for_physical_fields(self.predicate, self.table.fields)) | |
| if row_ranges is not None and not row_ranges: | |
| return [], DataEvolutionSplitGenerator( | |
| self.table, | |
| self.target_split_size, | |
| self.open_file_cost, | |
| {}, | |
| row_ranges, | |
| score_getter, | |
| None, | |
| snapshot_id=self._scanned_snapshot_id, | |
| ) | |
| # Position selection counts the complete candidate row-id space. Early | |
| # range pruning would renumber later files; the split generator applies | |
| # the range intersection after assigning slice/shard positions. | |
| positional = self.idx_of_this_subtask is not None or self.start_pos_of_this_subtask is not None | |
| early_row_ranges = None if positional else row_ranges | |
| if early_row_ranges is not None: | |
| manifest_files = _filter_manifest_files_by_row_ranges(manifest_files, early_row_ranges) | |
| stats_predicate = getattr(self, 'predicate_for_stats', None) | |
| group_stats_enabled = stats_predicate is not None and score_getter is None | |
| entries = self.read_manifest_entries( | |
| manifest_files, | |
| row_ranges=early_row_ranges, | |
| keep_stats=group_stats_enabled, | |
| ) | |
| # Redundant when early_record_filter ran; kept for explain mode and as safety net. | |
| if early_row_ranges is not None: | |
| entries = _filter_manifest_entries_by_row_ranges(entries, early_row_ranges) | |
| group_stats_filter = None | |
| if group_stats_enabled: | |
| group_stats_filter = DataEvolutionGroupStatsFilter( | |
| stats_predicate, | |
| self.table.fields, | |
| self._schema, | |
| ) | |
| return entries, DataEvolutionSplitGenerator( | |
| self.table, | |
| self.target_split_size, | |
| self.open_file_cost, | |
| self._deletion_files_map(entries), | |
| row_ranges, | |
| score_getter, | |
| group_stats_filter, | |
| snapshot_id=self._scanned_snapshot_id, | |
| ) | |
| def plan_files(self) -> List[ManifestEntry]: | |
| manifest_files, snapshot = self.manifest_scanner() | |
| self._scanned_snapshot = snapshot | |
| self._scanned_snapshot_id = snapshot.id if snapshot else None | |
| if len(manifest_files) == 0: | |
| return [] | |
| return self.read_manifest_entries(manifest_files) | |
| def _eval_global_index(self, snapshot=None): | |
| # Snapshot indexes describe current state, not historical change events. | |
| predicate = predicate_for_physical_fields(self.predicate, self.table.fields) | |
| if self.is_streaming or predicate is None: | |
| return None | |
| # Check if global index is enabled | |
| if not self.table.options.global_index_enabled(): | |
| return None | |
| from pypaimon.execution.python.globalindex.data_evolution_global_index_scanner import ( | |
| DataEvolutionGlobalIndexScanner) | |
| try: | |
| scanner = DataEvolutionGlobalIndexScanner.create( | |
| self.table, | |
| partition_filter=self.partition_key_predicate, | |
| predicate=predicate, | |
| snapshot=snapshot, | |
| ) | |
| if scanner is None: | |
| return None | |
| with scanner: | |
| evaluation = scanner.scan_with_coverage(predicate) | |
| if evaluation is None: | |
| return None | |
| scalar_mode = self.table.options.scalar_index_search_mode() | |
| return _GlobalIndexPlanningResult( | |
| evaluation.result, | |
| scanner.unindexed_ranges( | |
| predicate, | |
| search_mode=scalar_mode, | |
| contributing_field_ids=( | |
| evaluation.contributing_field_ids), | |
| ), | |
| ) | |
| except Exception: | |
| return None | |
| def read_manifest_entries(self, manifest_files: List[ManifestFileMeta], | |
| row_ranges=None, | |
| keep_stats=False) -> List[ManifestEntry]: | |
| max_workers = self.table.options.scan_manifest_parallelism(os.cpu_count() or 8) | |
| if self.scan_stats is not None: | |
| self.scan_stats.manifest_files_total += len(manifest_files) | |
| # ``num_added_files + num_deleted_files`` is the entry count | |
| # recorded in the manifest-file metadata; combined across | |
| # all input manifest files this is the partition-prune-free | |
| # baseline. The difference against ``entries_total`` reveals | |
| # how much manifest-level pruning saved. | |
| self.scan_stats.entries_potential_total += sum( | |
| f.num_added_files + f.num_deleted_files for f in manifest_files) | |
| manifest_files = [entry for entry in manifest_files if self._filter_manifest_file(entry)] | |
| if self.scan_stats is not None: | |
| self.scan_stats.manifest_files_after_partition += len(manifest_files) | |
| # Force single-threaded so we can mutate stats without locking. | |
| max_workers = 1 | |
| # Disable early entry filters and sidecar pruning in explain mode so all entries | |
| # flow through _filter_manifest_entry for accurate funnel counting. | |
| early_row_filter = None if self.scan_stats is not None \ | |
| else _build_early_row_range_filter(row_ranges) | |
| partition_filter = None if self.scan_stats is not None \ | |
| else self.partition_key_predicate | |
| return self.manifest_file_manager.read_entries_parallel( | |
| manifest_files, | |
| self._filter_manifest_entry, | |
| # Transported guarded splits must match the snapshot metadata when | |
| # Rust restores their catalog grant, even if Python planned them. | |
| drop_stats=not (keep_stats or self.table.options.query_auth_enabled), | |
| max_workers=max_workers, | |
| early_entry_filter=self._build_early_bucket_filter(), | |
| early_record_filter=early_row_filter, | |
| partition_filter=partition_filter, | |
| row_ranges=row_ranges if self.scan_stats is None else None, | |
| ) | |
| def _build_early_bucket_filter(self): | |
| """Compose the (bucket, total_buckets) -> bool used by the manifest | |
| reader to drop entries before deserialising ``_FILE`` / partition. | |
| The selector is partition-aware now, but at this early stage the | |
| partition field has not been deserialised yet, so callers stick | |
| with the two-arg form. The selector internally falls back to a | |
| partition-agnostic over-approximation; per-partition tightening | |
| still happens later in ``_filter_manifest_entry`` once the entry | |
| is fully decoded. | |
| """ | |
| # explain() needs accurate before/after pruning counters; suppress | |
| # the early bucket filter so every entry reaches | |
| # ``_filter_manifest_entry`` where each rejection stage is counted. | |
| if self.scan_stats is not None: | |
| return None | |
| only_real = self.only_read_real_buckets | |
| selector = self._bucket_selector | |
| if not only_real and selector is None: | |
| return None | |
| def _filter(bucket: int, total_buckets: int) -> bool: | |
| if only_real and bucket < 0: | |
| return False | |
| if (selector is not None | |
| and bucket >= 0 | |
| and not selector(bucket, total_buckets)): | |
| return False | |
| return True | |
| return _filter | |
| def with_all_buckets(self) -> 'FileScanner': | |
| """Include pending buckets for commit metadata scans, as in Java.""" | |
| self.only_read_real_buckets = False | |
| return self | |
| def with_shard(self, idx_of_this_subtask: int, number_of_para_subtasks: int) -> 'FileScanner': | |
| validate_shard_selection(idx_of_this_subtask, number_of_para_subtasks, self.start_pos_of_this_subtask) | |
| self.idx_of_this_subtask = idx_of_this_subtask | |
| self.number_of_para_subtasks = number_of_para_subtasks | |
| return self | |
| def with_slice(self, start_pos: int, end_pos: int) -> 'FileScanner': | |
| validate_slice_selection(start_pos, end_pos, self.table.is_primary_key_table, self.idx_of_this_subtask) | |
| self.start_pos_of_this_subtask = start_pos | |
| self.end_pos_of_this_subtask = end_pos | |
| return self | |
| def with_global_index_result(self, result) -> 'FileScanner': | |
| validate_global_index_selection(self._row_ranges) | |
| self._global_index_result = result | |
| return self | |
| def with_row_ranges(self, row_ranges) -> 'FileScanner': | |
| self._row_ranges = normalize_row_ranges(row_ranges, self.data_evolution, self._global_index_result) | |
| return self | |
| def scan_with_stats(self) -> Tuple[Plan, ScanStats]: | |
| """Run one scan pass while recording :class:`ScanStats` counters. | |
| Side-effects: forces single-thread manifest reads and disables the | |
| early bucket filter so every entry reaches | |
| ``_filter_manifest_entry`` exactly once. The scanner is one-shot | |
| in this mode — call ``scan()`` on a fresh instance afterwards if | |
| you need the regular hot path. | |
| """ | |
| 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': | |
| validate_chunk_shuffle(seed, chunk_size) | |
| self.chunk_shuffle = (seed, chunk_size) | |
| return self | |
| def _validate_chunk_shuffle_compat(self) -> None: | |
| validate_chunk_shuffle_compat( | |
| is_primary_key_table=self.table.is_primary_key_table, | |
| start_pos_of_this_subtask=self.start_pos_of_this_subtask, | |
| limit=self.limit, | |
| global_index_result=self._global_index_result, | |
| row_ranges=self._row_ranges, | |
| auth_has_non_partition_filter=self.auth_has_non_partition_filter, | |
| user_predicate=self._user_predicate, | |
| partition_keys=self.table.partition_keys, | |
| ) | |
| def _apply_push_down_limit(self, splits: List[DataSplit]) -> List[DataSplit]: | |
| # Preserve short-circuit validation and the existing private method hook. | |
| non_partition_filter = ( | |
| self.limit is not None | |
| and not (self.data_evolution and self.deletion_vectors_enabled) | |
| and self._has_non_partition_filter()) | |
| return apply_push_down_limit( | |
| splits, | |
| limit=self.limit, | |
| data_evolution=self.data_evolution, | |
| deletion_vectors_enabled=self.deletion_vectors_enabled, | |
| non_partition_filter=non_partition_filter, | |
| auth_has_non_partition_filter=self.auth_has_non_partition_filter, | |
| explicit_partition_filter=self._explicit_partition_filter, | |
| predicate=self.predicate, | |
| ) | |
| def _has_non_partition_filter(self) -> bool: | |
| return has_non_partition_filter(self.predicate, self.table.partition_keys) | |
| def _filter_manifest_file(self, file: ManifestFileMeta) -> bool: | |
| # explain() counts bucket rejections at entry level, as with the early | |
| # bucket filter. Keep reading those entries when collecting scan stats. | |
| if self.scan_stats is None and file.min_bucket is not None and file.max_bucket is not None: | |
| if self.only_read_real_buckets and file.max_bucket < 0: | |
| return False | |
| if (self._bucket_selector is not None | |
| and file.min_bucket >= 0 | |
| and file.total_buckets is not None | |
| and file.total_buckets > 0 | |
| and not self._bucket_selector.may_contain( | |
| file.min_bucket, file.max_bucket, file.total_buckets)): | |
| return False | |
| if not self.partition_key_predicate: | |
| return True | |
| return self.partition_key_predicate.test_by_simple_stats( | |
| file.partition_stats, | |
| file.num_added_files + file.num_deleted_files) | |
| def _init_bucket_selector(self): | |
| """Build the predicate-driven bucket selector if (and only if) the | |
| table is in HASH_FIXED mode and the predicate pins all bucket-key | |
| fields to Equal/In literals. Anything else returns None — the | |
| caller treats None as "no bucket-level pruning". | |
| Bucket-key fields come from ``TableSchema.logical_bucket_key_fields`` | |
| — the same source the writer's ``FixedBucketRowKeyExtractor`` reads | |
| from, which is what makes the read/write hash agreement a property | |
| of the schema rather than of any particular extractor instance. | |
| Sound across rescale: ``_Selector`` caches per ``total_buckets``, | |
| which can vary between manifest entries after a bucket rescale. | |
| """ | |
| if self.predicate is None: | |
| return None | |
| predicate = predicate_for_physical_fields(self.predicate, self.table.fields) | |
| if predicate is None: | |
| return None | |
| # ``bucket_mode()`` returns HASH_FIXED only when ``options.bucket() | |
| # > 0``; other modes (DYNAMIC / POSTPONE / UNAWARE / CROSS_PARTITION) | |
| # have no fixed hash → bucket mapping at write time and must NOT | |
| # be pruned here. | |
| try: | |
| if self.table.bucket_mode() != BucketMode.HASH_FIXED: | |
| return None | |
| except Exception: | |
| # Defensive: any catalog/proxy table that fails the mode check | |
| # falls back to no pruning rather than crashing the scan. | |
| return None | |
| # Only the default hash function (Math.abs(hash % numBuckets)) is | |
| # supported for bucket pruning. Non-default functions (mod, hive) | |
| # use different algorithms and would produce wrong bucket sets. | |
| bucket_func = self.table.table_schema.options.get('bucket-function.type') | |
| if bucket_func is not None and bucket_func.lower() != 'default': | |
| return None | |
| try: | |
| bucket_key_fields = self.table.table_schema.logical_bucket_key_fields | |
| except Exception: | |
| # ``bucket_keys`` raises on misconfigured ``bucket-key`` (e.g. | |
| # references an unknown column). The previous extractor-based | |
| # path failed open here; preserve that — pruning is an | |
| # optimisation, never a correctness requirement. | |
| return None | |
| if not bucket_key_fields: | |
| return None | |
| # Partition fields are passed so the selector can specialise | |
| # the predicate per partition value at the late filter stage, | |
| # turning ``(part='a' AND bk=1) OR (part='b' AND bk=2)`` into a | |
| # precise bucket pick per partition instead of an over-scan. | |
| partition_fields: Optional[List[DataField]] = None | |
| if self.table.partition_keys: | |
| partition_fields = [ | |
| self.table.field_dict[name] | |
| for name in self.table.partition_keys | |
| if name in self.table.field_dict | |
| ] | |
| return create_bucket_selector( | |
| predicate, bucket_key_fields, | |
| partition_fields=partition_fields, | |
| ) | |
| def _filter_manifest_entry(self, entry: ManifestEntry) -> bool: | |
| if self.is_streaming and entry.kind != 0: | |
| raise ValueError("Incremental delta manifests must contain only ADD entries") | |
| stats = self.scan_stats | |
| if stats is not None: | |
| stats.entries_total += 1 | |
| partition_key = tuple(entry.partition.values) | |
| stats.partition_keys_before.add(partition_key) | |
| stats.buckets_seen.add((partition_key, entry.bucket)) | |
| # Stage 1: partition predicate. The early manifest-reader filter | |
| # only sees ``(bucket, total_buckets)`` and never enforces | |
| # partition predicates, so this check is the sole partition gate | |
| # at the entry level — not a "redundant safety net". | |
| if self.partition_key_predicate and not self.partition_key_predicate.test(entry.partition): | |
| return False | |
| if self.auth_partition_predicate and not self.auth_partition_predicate.test(entry.partition): | |
| return False | |
| if stats is not None: | |
| stats.entries_after_partition += 1 | |
| stats.partition_keys_after.add(partition_key) | |
| # Stage 2: bucket rejection. Two reasons land here: | |
| # * ``only_read_real_buckets`` drops the synthetic | |
| # POSTPONE_BUCKET bucket id (also enforced by the early | |
| # filter when present; kept here so the method is correct | |
| # standalone). | |
| # * ``_bucket_selector`` is the HASH_FIXED predicate-driven | |
| # selector built by ``_init_bucket_selector``. | |
| # Both are accounted for under ``entries_after_bucket`` so the | |
| # explain funnel reports bucket-level pruning end-to-end. | |
| if self.only_read_real_buckets and entry.bucket < 0: | |
| return False | |
| if (self._bucket_selector is not None | |
| and entry.bucket >= 0 | |
| and not self._bucket_selector( | |
| entry.partition, entry.bucket, entry.total_buckets)): | |
| return False | |
| if stats is not None: | |
| stats.entries_after_bucket += 1 | |
| stats.buckets_after_pruning.add((partition_key, entry.bucket)) | |
| if self.level_filter is not None and not self.level_filter(entry.file.level): | |
| return False | |
| # Get SimpleStatsEvolution for this schema | |
| evolution = self.simple_stats_evolutions.get_or_create(entry.file.schema_id) | |
| stats_predicate = self.simple_stats_evolutions.filter_unsafe_filter( | |
| entry.file.schema_id, self.predicate_for_stats) | |
| # Apply evolution to stats | |
| if self.table.is_primary_key_table: | |
| if self.skip_level0 and entry.file.level == 0: | |
| return False | |
| if self.primary_key_predicate: | |
| if not self.primary_key_predicate.test_by_simple_stats( | |
| entry.file.key_stats, | |
| entry.file.row_count | |
| ): | |
| return False | |
| # Java enables value filtering only when batch scans exclude L0. | |
| # With L0 present, pruning an update can expose an older value. | |
| if self.skip_level0 and stats_predicate: | |
| if entry.file.value_stats_cols is None and entry.file.write_cols is not None: | |
| stats_fields = entry.file.write_cols | |
| else: | |
| stats_fields = entry.file.value_stats_cols | |
| evolved_stats = evolution.evolution( | |
| entry.file.value_stats, | |
| entry.file.row_count, | |
| stats_fields | |
| ) | |
| if not stats_predicate.test_by_simple_stats( | |
| evolved_stats, | |
| entry.file.row_count | |
| ): | |
| return False | |
| if stats is not None: | |
| stats.entries_after_stats += 1 | |
| return True | |
| else: | |
| if not self.predicate or stats_predicate is None: | |
| if stats is not None: | |
| stats.entries_after_stats += 1 | |
| return True | |
| # Data evolution: file stats may be from another schema, skip stats filter and filter in reader. | |
| if self.data_evolution: | |
| return True | |
| if entry.file.value_stats_cols is None and entry.file.write_cols is not None: | |
| stats_fields = entry.file.write_cols | |
| else: | |
| stats_fields = entry.file.value_stats_cols | |
| evolved_stats = evolution.evolution( | |
| entry.file.value_stats, | |
| entry.file.row_count, | |
| stats_fields | |
| ) | |
| kept = stats_predicate.test_by_simple_stats( | |
| evolved_stats, | |
| entry.file.row_count | |
| ) | |
| if kept and stats is not None: | |
| stats.entries_after_stats += 1 | |
| return kept | |
| def _scan_dv_index(self, snapshot, buckets: Set[tuple]) -> Dict[tuple, Dict[str, DeletionFile]]: | |
| """ | |
| Scan deletion vector index from snapshot. | |
| Returns a map of (partition, bucket) -> {filename -> DeletionFile} | |
| Reference: SnapshotReaderImpl.scanDvIndex() in Java | |
| """ | |
| if not snapshot or not snapshot.index_manifest: | |
| return {} | |
| result = {} | |
| # Read index manifest file | |
| index_manifest_file = IndexManifestFile(self.table) | |
| index_entries = index_manifest_file.read(snapshot.index_manifest) | |
| # Filter by DELETION_VECTORS_INDEX type and requested buckets | |
| for entry in index_entries: | |
| if entry.index_file.index_type != IndexManifestFile.DELETION_VECTORS_INDEX: | |
| continue | |
| partition_bucket = (tuple(entry.partition.values), entry.bucket) | |
| if partition_bucket not in buckets: | |
| continue | |
| # Convert to deletion files | |
| deletion_files = self._to_deletion_files(entry) | |
| if deletion_files: | |
| result.setdefault(partition_bucket, {}).update(deletion_files) | |
| return result | |
| def _to_deletion_files(self, index_entry) -> Dict[str, DeletionFile]: | |
| """ | |
| Convert index manifest entry to deletion files map. | |
| Returns {filename -> DeletionFile} | |
| """ | |
| deletion_files = {} | |
| index_file = index_entry.index_file | |
| # Check if dv_ranges exists | |
| if not index_file.dv_ranges: | |
| return deletion_files | |
| dv_file_path = self.table.path_factory().bucket_index_path( | |
| tuple(index_entry.partition.values), index_entry.bucket, index_file) | |
| # Convert each DeletionVectorMeta to DeletionFile | |
| for data_file_name, dv_meta in index_file.dv_ranges.items(): | |
| deletion_file = DeletionFile( | |
| dv_index_path=dv_file_path, | |
| offset=dv_meta.offset, | |
| length=dv_meta.length, | |
| cardinality=dv_meta.cardinality | |
| ) | |
| deletion_files[data_file_name] = deletion_file | |
| return deletion_files |