| # 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. | |
| """The ``$table_indexes`` system table.""" | |
| from typing import Dict, List, Optional | |
| import pyarrow | |
| from pypaimon.index.index_file_handler import IndexFileHandler | |
| from pypaimon.schema.data_types import (ArrayType, AtomicType, DataField, | |
| RowType) | |
| from pypaimon.table.system.files_table import _render_partition | |
| from pypaimon.table.system.system_table import SystemTable, metadata_snapshot | |
| _DV_RANGE_ROW_TYPE = RowType(False, [ | |
| DataField(0, "f0", AtomicType("STRING", nullable=False)), | |
| DataField(1, "f1", AtomicType("INT", nullable=False)), | |
| DataField(2, "f2", AtomicType("INT", nullable=False)), | |
| DataField(3, "_CARDINALITY", AtomicType("BIGINT", nullable=True)), | |
| ]) | |
| _DV_RANGES_TYPE = pyarrow.list_(pyarrow.struct([ | |
| pyarrow.field("f0", pyarrow.string(), nullable=False), | |
| pyarrow.field("f1", pyarrow.int32(), nullable=False), | |
| pyarrow.field("f2", pyarrow.int32(), nullable=False), | |
| pyarrow.field("_CARDINALITY", pyarrow.int64(), nullable=True), | |
| ])) | |
| TABLE_TYPE = RowType(False, [ | |
| DataField(0, "partition", AtomicType("STRING", nullable=True)), | |
| DataField(1, "bucket", AtomicType("INT", nullable=False)), | |
| DataField(2, "index_type", AtomicType("STRING", nullable=False)), | |
| DataField(3, "file_name", AtomicType("STRING", nullable=False)), | |
| DataField(4, "file_size", AtomicType("BIGINT", nullable=False)), | |
| DataField(5, "row_count", AtomicType("BIGINT", nullable=False)), | |
| DataField( | |
| 6, | |
| "dv_ranges", | |
| ArrayType(nullable=True, | |
| element_type=_DV_RANGE_ROW_TYPE)), | |
| DataField(7, "row_range_start", AtomicType("BIGINT", nullable=True)), | |
| DataField(8, "row_range_end", AtomicType("BIGINT", nullable=True)), | |
| DataField(9, "index_field_id", AtomicType("INT", nullable=True)), | |
| DataField(10, "index_field_name", AtomicType("STRING", nullable=True)), | |
| ]) | |
| class TableIndexesTable(SystemTable): | |
| """The ``$table_indexes`` system table.""" | |
| def system_table_name(self) -> str: | |
| return "table_indexes" | |
| def row_type(self) -> RowType: | |
| return TABLE_TYPE | |
| def primary_keys(self) -> List[str]: | |
| return ["file_name"] | |
| def _build_arrow_table(self) -> pyarrow.Table: | |
| snapshot = metadata_snapshot(self.base_table) | |
| # scan(None) means latest; a resolved empty view must stay empty. | |
| entries = [] if snapshot is None else IndexFileHandler(self.base_table).scan(snapshot) | |
| rows = { | |
| "partition": [], | |
| "bucket": [], | |
| "index_type": [], | |
| "file_name": [], | |
| "file_size": [], | |
| "row_count": [], | |
| "dv_ranges": [], | |
| "row_range_start": [], | |
| "row_range_end": [], | |
| "index_field_id": [], | |
| "index_field_name": [], | |
| } | |
| for entry in entries: | |
| index_file = entry.index_file | |
| global_meta = index_file.global_index_meta | |
| rows["partition"].append(_render_partition(entry.partition)) | |
| rows["bucket"].append(int(entry.bucket)) | |
| rows["index_type"].append(index_file.index_type) | |
| rows["file_name"].append(index_file.file_name) | |
| rows["file_size"].append(int(index_file.file_size)) | |
| rows["row_count"].append(int(index_file.row_count)) | |
| rows["dv_ranges"].append(_render_dv_ranges(index_file.dv_ranges)) | |
| rows["row_range_start"].append( | |
| None if global_meta is None | |
| else int(global_meta.row_range_start)) | |
| rows["row_range_end"].append( | |
| None if global_meta is None | |
| else int(global_meta.row_range_end)) | |
| rows["index_field_id"].append( | |
| None if global_meta is None | |
| else int(global_meta.index_field_id)) | |
| rows["index_field_name"].append( | |
| None if global_meta is None | |
| else _index_field_names(self.base_table, global_meta)) | |
| return pyarrow.table({ | |
| "partition": pyarrow.array( | |
| rows["partition"], type=pyarrow.string()), | |
| "bucket": pyarrow.array(rows["bucket"], type=pyarrow.int32()), | |
| "index_type": pyarrow.array( | |
| rows["index_type"], type=pyarrow.string()), | |
| "file_name": pyarrow.array( | |
| rows["file_name"], type=pyarrow.string()), | |
| "file_size": pyarrow.array( | |
| rows["file_size"], type=pyarrow.int64()), | |
| "row_count": pyarrow.array( | |
| rows["row_count"], type=pyarrow.int64()), | |
| "dv_ranges": pyarrow.array( | |
| rows["dv_ranges"], type=_DV_RANGES_TYPE), | |
| "row_range_start": pyarrow.array( | |
| rows["row_range_start"], type=pyarrow.int64()), | |
| "row_range_end": pyarrow.array( | |
| rows["row_range_end"], type=pyarrow.int64()), | |
| "index_field_id": pyarrow.array( | |
| rows["index_field_id"], type=pyarrow.int32()), | |
| "index_field_name": pyarrow.array( | |
| rows["index_field_name"], type=pyarrow.string()), | |
| }) | |
| def _render_dv_ranges(dv_ranges) -> Optional[List[Dict[str, object]]]: | |
| if dv_ranges is None: | |
| return None | |
| rendered = [] | |
| for meta in dv_ranges.values(): | |
| rendered.append({ | |
| "f0": meta.data_file_name, | |
| "f1": int(meta.offset), | |
| "f2": int(meta.length), | |
| "_CARDINALITY": ( | |
| None if meta.cardinality is None else int(meta.cardinality) | |
| ), | |
| }) | |
| return rendered | |
| def _index_field_names(table, global_meta) -> Optional[str]: | |
| field_by_id = {field.id: field.name for field in table.fields} | |
| field_ids = [global_meta.index_field_id] | |
| if global_meta.extra_field_ids: | |
| field_ids.extend(global_meta.extra_field_ids) | |
| names = [] | |
| for field_id in field_ids: | |
| name = field_by_id.get(field_id) | |
| if name is None: | |
| return None | |
| names.append(name) | |
| return ",".join(names) |