|
| 1 | +# Licensed to the Apache Software Foundation (ASF) under one |
| 2 | +# or more contributor license agreements. See the NOTICE file |
| 3 | +# distributed with this work for additional information |
| 4 | +# regarding copyright ownership. The ASF licenses this file |
| 5 | +# to you under the Apache License, Version 2.0 (the |
| 6 | +# "License"); you may not use this file except in compliance |
| 7 | +# with the License. You may obtain a copy of the License at |
| 8 | +# |
| 9 | +# http://www.apache.org/licenses/LICENSE-2.0 |
| 10 | +# |
| 11 | +# Unless required by applicable law or agreed to in writing, |
| 12 | +# software distributed under the License is distributed on an |
| 13 | +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| 14 | +# KIND, either express or implied. See the License for the |
| 15 | +# specific language governing permissions and limitations |
| 16 | +# under the License. |
| 17 | + |
| 18 | +"""Freshness checks for global indexes on data-evolution tables.""" |
| 19 | + |
| 20 | +import logging |
| 21 | + |
| 22 | +from pypaimon.common.options.core_options import CoreOptions, GlobalIndexColumnUpdateAction |
| 23 | +from pypaimon.table.special_fields import SpecialFields |
| 24 | + |
| 25 | +logger = logging.getLogger(__name__) |
| 26 | + |
| 27 | + |
| 28 | +def key_indexes_are_fresh(table, snapshot, match_keys, partition_predicate): |
| 29 | + """Check whether IGNORE may have left indexed match keys stale.""" |
| 30 | + if table.options.global_index_column_update_action() != GlobalIndexColumnUpdateAction.IGNORE: |
| 31 | + return True |
| 32 | + if snapshot is None or not match_keys: |
| 33 | + return True |
| 34 | + from pypaimon.execution.python.globalindex.data_evolution_global_index_scanner import ( |
| 35 | + is_supported_scalar_index, |
| 36 | + ) |
| 37 | + from pypaimon.index.data_evolution_index_source_meta import DataEvolutionIndexSourceMeta |
| 38 | + from pypaimon.index.index_file_handler import IndexFileHandler |
| 39 | + from pypaimon.read.push_down_utils import trim_and_transform_predicate |
| 40 | + |
| 41 | + try: |
| 42 | + key_ids = {table.field_dict[key].id for key in match_keys} |
| 43 | + partition_filter = trim_and_transform_predicate( |
| 44 | + partition_predicate, table.field_names, table.partition_keys) |
| 45 | + native_planning = table.options.native_read_enabled() or table.options.native_plan_enabled() |
| 46 | + |
| 47 | + def relevant_index(entry): |
| 48 | + meta = entry.index_file.global_index_meta |
| 49 | + # Native planning can use more index types, including composite |
| 50 | + # BTree, even when Python performs the final key matching. |
| 51 | + supported = (is_supported_scalar_index(entry.index_file) |
| 52 | + or (native_planning and meta is not None |
| 53 | + and entry.index_file.index_type.lower() in ('btree', 'bitmap', 'fm', 'multivalue'))) |
| 54 | + return (supported |
| 55 | + and bool(key_ids.intersection( |
| 56 | + [meta.index_field_id] + (meta.extra_field_ids or []))) |
| 57 | + and (partition_filter is None or partition_filter.test(entry.partition))) |
| 58 | + |
| 59 | + indexes = IndexFileHandler(table).scan(snapshot, relevant_index) |
| 60 | + if not indexes: |
| 61 | + return True |
| 62 | + baselines = [] |
| 63 | + for index in indexes: |
| 64 | + meta = index.index_file.global_index_meta |
| 65 | + source = DataEvolutionIndexSourceMeta.deserialize(meta.source_meta) |
| 66 | + if source.scan_snapshot_id > snapshot.id or meta.row_range_start > meta.row_range_end: |
| 67 | + return False |
| 68 | + baselines.append((index, source.scan_snapshot_id)) |
| 69 | + |
| 70 | + # Inspect active file metadata before either key stats or an index can |
| 71 | + # hide the files containing changed keys. Both plans use this snapshot. |
| 72 | + builder = table.copy_without_time_travel({ |
| 73 | + CoreOptions.GLOBAL_INDEX_ENABLED.key(): 'false', |
| 74 | + }).new_read_builder().with_filter(partition_predicate) |
| 75 | + indexed_groups = {(tuple(index.partition.values), index.bucket) for index, _ in baselines} |
| 76 | + files_by_group = {} |
| 77 | + for split in builder.new_scan().plan_all().splits(): |
| 78 | + group = (tuple(split.partition.values), split.bucket) |
| 79 | + if group in indexed_groups: |
| 80 | + files = files_by_group.setdefault(group, {}) |
| 81 | + files.update((file.file_name, file) for file in split.files) |
| 82 | + ranges_by_group = {} |
| 83 | + for group, files in files_by_group.items(): |
| 84 | + ranges = [] |
| 85 | + for file in files.values(): |
| 86 | + row_range = file.row_id_range() |
| 87 | + if row_range is None: |
| 88 | + return False |
| 89 | + ranges.append((row_range.from_, row_range.to, file)) |
| 90 | + ranges_by_group[group] = _FileRangeIndex(ranges) |
| 91 | + |
| 92 | + schema = table.table_schema |
| 93 | + known_names = set(table.field_names) | SpecialFields.SYSTEM_FIELD_NAMES |
| 94 | + for index, baseline in baselines: |
| 95 | + meta = index.index_file.global_index_meta |
| 96 | + indexed_keys = key_ids.intersection([meta.index_field_id] + (meta.extra_field_ids or [])) |
| 97 | + ranges = ranges_by_group.get((tuple(index.partition.values), index.bucket)) |
| 98 | + if ranges is None: |
| 99 | + continue |
| 100 | + for file in ranges.overlapping(meta.row_range_start, meta.row_range_end): |
| 101 | + # Historical schemas are outside this optimization's proof. |
| 102 | + if file.schema_id != schema.id: |
| 103 | + return False |
| 104 | + if file.write_cols is not None and ( |
| 105 | + any(name not in known_names for name in file.write_cols) |
| 106 | + or len(file.write_cols) != len(set(file.write_cols))): |
| 107 | + return False |
| 108 | + fields = schema.data_file_fields(file.write_cols) |
| 109 | + sequences = file.write_cols_sequences |
| 110 | + for position, field in enumerate(fields): |
| 111 | + if field.id not in indexed_keys: |
| 112 | + continue |
| 113 | + sequence = (sequences[position] |
| 114 | + if sequences is not None and len(sequences) == len(fields) |
| 115 | + else file.max_sequence_number) |
| 116 | + if sequence <= 0 or sequence > file.max_sequence_number or sequence > baseline: |
| 117 | + return False |
| 118 | + return True |
| 119 | + except Exception: |
| 120 | + logger.debug("Cannot verify upsert key index freshness; scanning without indexes.", exc_info=True) |
| 121 | + return False |
| 122 | + |
| 123 | + |
| 124 | +class _FileRangeIndex: |
| 125 | + """Centered interval tree for inclusive physical-file row ranges. |
| 126 | +
|
| 127 | + Build once per partition/bucket in O(F log F); each shard visits only |
| 128 | + O(log F + overlapping files), including nested and long column files. |
| 129 | + """ |
| 130 | + |
| 131 | + def __init__(self, ranges): |
| 132 | + self._root = self._build(sorted(ranges, key=lambda item: item[0])) |
| 133 | + |
| 134 | + @classmethod |
| 135 | + def _build(cls, ranges): |
| 136 | + if not ranges: |
| 137 | + return None |
| 138 | + center = ranges[len(ranges) // 2][0] |
| 139 | + left, crossing, right = [], [], [] |
| 140 | + for item in ranges: |
| 141 | + start, end, _ = item |
| 142 | + if end < center: |
| 143 | + left.append(item) |
| 144 | + elif start > center: |
| 145 | + right.append(item) |
| 146 | + else: |
| 147 | + crossing.append(item) |
| 148 | + # Filtering preserves start order; each file belongs to one node only. |
| 149 | + by_end = sorted(crossing, key=lambda item: item[1], reverse=True) |
| 150 | + return center, crossing, by_end, cls._build(left), cls._build(right) |
| 151 | + |
| 152 | + def overlapping(self, start, end): |
| 153 | + pending = [self._root] |
| 154 | + while pending: |
| 155 | + node = pending.pop() |
| 156 | + if node is None: |
| 157 | + continue |
| 158 | + center, by_start, by_end, left, right = node |
| 159 | + if end < center: |
| 160 | + for file_start, _, file in by_start: |
| 161 | + if file_start > end: |
| 162 | + break |
| 163 | + yield file |
| 164 | + pending.append(left) |
| 165 | + elif start > center: |
| 166 | + for _, file_end, file in by_end: |
| 167 | + if file_end < start: |
| 168 | + break |
| 169 | + yield file |
| 170 | + pending.append(right) |
| 171 | + else: |
| 172 | + for _, _, file in by_start: |
| 173 | + yield file |
| 174 | + pending.extend((left, right)) |
0 commit comments