github-actions[bot] commented on code in PR #67977: URL: https://github.com/apache/doris/pull/67977#discussion_r4012041021
########## fe/fe-core/src/main/java/org/apache/doris/tablefunction/IndexDiskUsageTableValuedFunction.java: ########## @@ -0,0 +1,387 @@ +// 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. + +package org.apache.doris.tablefunction; + +import org.apache.doris.analysis.TupleDescriptor; +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.Database; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.Index; +import org.apache.doris.catalog.MaterializedIndex; +import org.apache.doris.catalog.MaterializedIndex.IndexExtState; +import org.apache.doris.catalog.OlapTable; +import org.apache.doris.catalog.Partition; +import org.apache.doris.catalog.PrimitiveType; +import org.apache.doris.catalog.ScalarType; +import org.apache.doris.catalog.TableIf; +import org.apache.doris.catalog.Tablet; +import org.apache.doris.catalog.info.IndexType; +import org.apache.doris.cloud.catalog.CloudPartition; +import org.apache.doris.common.Config; +import org.apache.doris.common.ErrorCode; +import org.apache.doris.common.Pair; +import org.apache.doris.datasource.InternalCatalog; +import org.apache.doris.datasource.tvf.source.IndexDiskUsageScanNode; +import org.apache.doris.mysql.privilege.PrivPredicate; +import org.apache.doris.nereids.exceptions.AnalysisException; +import org.apache.doris.planner.PlanNodeId; +import org.apache.doris.planner.ScanContext; +import org.apache.doris.planner.ScanNode; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.SessionVariable; +import org.apache.doris.rpc.RpcException; +import org.apache.doris.thrift.TIndexDiskUsageMetadataParams; +import org.apache.doris.thrift.TIndexDiskUsageTablet; +import org.apache.doris.thrift.TMetaScanRange; +import org.apache.doris.thrift.TMetadataType; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSet; +import com.google.common.collect.Lists; +import com.google.common.collect.Maps; +import org.apache.commons.lang3.StringUtils; + +import java.util.Arrays; +import java.util.Collection; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +/** + * The implement of table valued function + * index_disk_usage("database" = "db1", "table" = "table1"). + * It reports the physical bytes of every inverted index of the table, split by component. + */ +public class IndexDiskUsageTableValuedFunction extends MetadataTableValuedFunction { + public static final String NAME = "index_disk_usage"; + + private static final String DATABASE = "database"; + private static final String TABLE = "table"; + private static final String PARTITIONS = "partitions"; + private static final String INDEXES = "indexes"; + private static final String LEVEL = "level"; + private static final String POSITION_DETAIL = "position_detail"; + + private static final ImmutableSet<String> PROPERTIES_SET = + ImmutableSet.of(DATABASE, TABLE, PARTITIONS, INDEXES, LEVEL, POSITION_DETAIL); + private static final ImmutableSet<String> LEVELS = ImmutableSet.of("tablet", "rowset", "segment"); + + private static final ImmutableList<Column> SCHEMA = ImmutableList.of( + varcharColumn("PARTITION_NAME"), + varcharColumn("MATERIALIZED_INDEX_NAME"), + bigintColumn("TABLET_ID"), + bigintColumn("BACKEND_ID"), + varcharColumn("ROWSET_ID"), + new Column("SEGMENT_ID", ScalarType.createType(PrimitiveType.INT), true), + bigintColumn("INDEX_ID"), + varcharColumn("INDEX_NAME"), + varcharColumn("INDEX_TYPE"), + varcharColumn("COLUMN_NAME"), + varcharColumn("INDEX_SUFFIX"), + varcharColumn("STRUCTURE"), + varcharColumn("STORAGE_FORMAT"), + bigintColumn("SEGMENT_COUNT"), + bigintColumn("ROW_COUNT"), + bigintColumn("TOTAL_BYTES"), + bigintColumn("DICT_BYTES"), + bigintColumn("POSTING_BYTES"), + bigintColumn("POSITION_BYTES"), + bigintColumn("STATS_BYTES"), + bigintColumn("OTHER_BYTES"), + varcharColumn("STATS_SOURCE")); + + /** + * A tablet of a base or rollup index to inspect, pinned to the visible version of its partition. + */ + public static class TabletTarget { + private final Tablet tablet; + private final long partitionId; + private final long materializedIndexId; + private final long version; + + public TabletTarget(Tablet tablet, long partitionId, long materializedIndexId, long version) { + this.tablet = tablet; + this.partitionId = partitionId; + this.materializedIndexId = materializedIndexId; + this.version = version; + } + + public long getMaterializedIndexId() { + return materializedIndexId; + } + + public Tablet getTablet() { + return tablet; + } + + public long getTabletId() { + return tablet.getId(); + } + + public long getPartitionId() { + return partitionId; + } + + public long getVersion() { + return version; + } + + public TIndexDiskUsageTablet toThrift() { + TIndexDiskUsageTablet target = new TIndexDiskUsageTablet(); + target.setTabletId(getTabletId()); + target.setPartitionId(partitionId); + target.setMaterializedIndexId(materializedIndexId); + target.setVersion(version); + return target; + } + } + + private final String level; + private final boolean positionDetail; + private final List<Long> indexIds; + private final Map<Long, String> partitionNames; + private final Map<Long, String> materializedIndexNames; + private final List<TabletTarget> tabletTargets; + + public IndexDiskUsageTableValuedFunction(Map<String, String> params) throws AnalysisException { + Map<String, String> validParams = Maps.newHashMap(); + for (Map.Entry<String, String> entry : params.entrySet()) { + String key = entry.getKey().toLowerCase(); + if (!PROPERTIES_SET.contains(key)) { + throw new AnalysisException("'" + entry.getKey() + "' is invalid property"); + } + validParams.put(key, entry.getValue()); + } + String dbName = validParams.get(DATABASE); + String tableName = validParams.get(TABLE); + if (StringUtils.isEmpty(dbName) || StringUtils.isEmpty(tableName)) { + throw new AnalysisException("'database' and 'table' are required for index_disk_usage"); + } + this.level = parseLevel(validParams.getOrDefault(LEVEL, "tablet")); + this.positionDetail = parsePositionDetail(validParams.getOrDefault(POSITION_DETAIL, "false")); + checkShowPrivilege(dbName, tableName); + + OlapTable table = getOlapTable(dbName, tableName); + String qualifiedName = dbName + "." + tableName; + List<Long> resolvedIndexIds; + List<Partition> partitions; + Map<Long, String> resolvedPartitionNames = Maps.newLinkedHashMap(); + Map<Long, String> resolvedMaterializedIndexNames = Maps.newLinkedHashMap(); + Map<Long, List<Pair<Long, List<Tablet>>>> tabletsByPartition = Maps.newHashMap(); + table.readLock(); + try { + resolvedIndexIds = resolveIndexIds(table, validParams.get(INDEXES), qualifiedName); + partitions = resolvePartitions(table, validParams.get(PARTITIONS), qualifiedName); + for (Partition partition : partitions) { + resolvedPartitionNames.put(partition.getId(), partition.getName()); + // A light ADD INDEX also installs indexes on rollups, so their tablets can hold index files. + List<Pair<Long, List<Tablet>>> indexTablets = Lists.newArrayList(); + for (MaterializedIndex index : partition.getMaterializedIndices(IndexExtState.VISIBLE)) { + resolvedMaterializedIndexNames.putIfAbsent(index.getId(), table.getIndexNameById(index.getId())); + indexTablets.add(Pair.of(index.getId(), Lists.newArrayList(index.getTablets()))); + } + tabletsByPartition.put(partition.getId(), indexTablets); + } + } finally { + table.readUnlock(); + } + // Cloud partitions may fetch visible versions from meta-service, so read them without the table lock. + List<Long> versions = visibleVersions(partitions); Review Comment: [P1] Keep the tablet set and version in one coherent snapshot The visible rollups/tablets are copied under the table lock, but their partition version is fetched only after the lock is released. A reachable local-mode interleaving is: copy rollup R at version V; DROP ROLLUP removes R; publish V+1 updates only the remaining indexes; then this call returns V+1. The retained replicas of R never receive V+1, so `getQueryableReplicas(V+1, ..., false)` rejects them all and this TVF fails with `No queryable replica` (the inverse rollup-visibility transition can also omit a newly visible rollup). This remains after the earlier base-only enumeration issue was fixed. Please establish the opposite ordering—obtain the versions first, then reacquire the read lock and resolve/copy the matching partition/index/tablet snapshot, validating identities or retrying if necessary—and add latch-controlled rollup/drop-plus-publish coverage. ########## be/src/storage/index/index_disk_usage.cpp: ########## @@ -0,0 +1,453 @@ +// 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. + +#include "storage/index/index_disk_usage.h" + +#include <algorithm> +#include <map> +#include <memory> +#include <tuple> +#include <utility> + +#include "common/cast_set.h" +#include "common/check.h" +#include "io/io_common.h" +#include "storage/index/ann/ann_index_files.h" +#include "storage/index/index_file_reader.h" +#include "storage/index/inverted/inverted_index_desc.h" +#include "storage/index/snii/format/dict_entry.h" +#include "storage/index/snii/format/format_constants.h" +#include "storage/index/snii/reader/logical_index_reader.h" +#include "storage/olap_common.h" +#include "storage/rowset/beta_rowset.h" +#include "storage/rowset/rowset.h" +#include "storage/rowset/rowset_meta.h" +#include "storage/segment/segment.h" + +namespace doris::segment_v2 { + +namespace { + +bool is_bkd_file(std::string_view name) { + return name == InvertedIndexDescriptor::get_temporary_bkd_index_data_file_name() || + name == InvertedIndexDescriptor::get_temporary_bkd_index_meta_file_name() || + name == InvertedIndexDescriptor::get_temporary_bkd_index_file_name(); +} + +bool is_ann_file(std::string_view name) { + return name == faiss_index_fila_name || name == faiss_ivfdata_file_name; +} + +bool is_wanted(const IndexDiskUsageOptions& options, int64_t index_id) { + return options.index_ids.empty() || options.index_ids.contains(index_id); +} + +Status check_cancelled(const IndexDiskUsageOptions& options) { + return options.check_cancelled ? options.check_cancelled() : Status::OK(); +} + +// A segment may have no index file on purpose, for example when every ANN index skipped a segment +// too small to train, or when a legacy table skipped writing indexes on load. Such a file holds +// no index bytes, while any other error is still reported. +bool is_absent_index_file(const Status& status) { + return status.is<ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND>() || + status.is<ErrorCode::INVERTED_INDEX_BYPASS>() || status.is<ErrorCode::NOT_FOUND>(); +} + +// A V1 index file and its size persisted in the rowset meta, or -1 when the size is not recorded. +struct V1IndexFile { + const TabletIndex* index; + int64_t persisted_size; +}; + +// Lists the V1 index files of a segment. The rowset meta records every file, including the +// extracted VARIANT paths that the schema does not list; rowsets written without that record fall +// back to the schema indexes. `owned` keeps the indexes built from the rowset meta. +std::vector<V1IndexFile> list_v1_index_files(const TabletSchema& schema, + const InvertedIndexFileInfo& file_info, + std::vector<TabletIndex>* owned) { + std::vector<V1IndexFile> files; + if (file_info.index_info_size() == 0) { + for (const TabletIndex* index : schema.inverted_indexes()) { + files.push_back({.index = index, .persisted_size = -1}); + } + return files; + } + owned->reserve(file_info.index_info_size()); + for (const auto& index_info : file_info.index_info()) { + TabletIndexPB index_pb; + index_pb.set_index_type(IndexType::INVERTED); + index_pb.set_index_id(index_info.index_id()); + index_pb.set_index_suffix_name(index_info.index_suffix()); + owned->emplace_back().init_from_pb(index_pb); + files.push_back({.index = &owned->back(), + .persisted_size = index_info.index_file_size() > 0 + ? index_info.index_file_size() + : -1}); + } + return files; +} + +// Classifies every sub-file of a CLucene directory into `record` and adds their total length to +// `files_bytes`. +Status add_directory_files(const lucene::store::Directory& dir, IndexDiskUsageRecord* record, + int64_t* files_bytes) { + try { + std::vector<std::string> names; + if (!dir.list(&names)) { + return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>( + "failed to list inverted index sub-files"); + } + for (const auto& name : names) { + const int64_t length = dir.fileLength(name.c_str()); + classify_clucene_file(name, length, record); + *files_bytes += length; + } + } catch (CLuceneError& e) { + return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>( + "failed to read inverted index sub-files: {}", e.what()); + } + return Status::OK(); +} + +// Sums the position bytes of the dictionary entries whose postings live in the posting region. +// Inline postings stay in the dictionary region and are not counted. +Status sum_snii_position_bytes(const IndexFileReader& reader, uint64_t index_id, + std::string_view suffix, const IndexDiskUsageOptions& options, + int64_t* position_bytes) { + // A full dictionary scan should not evict blocks that queries keep in the file cache. + io::IOContext io_ctx; + io_ctx.is_disposable = true; + io_ctx.is_inverted_index = true; + auto logical = DORIS_TRY(reader.open_snii_logical_index( + index_id, suffix, &io_ctx, snii::reader::LogicalIndexOpenMode::kCompaction)); + snii_doris::DorisSniiFileReader::ScopedIOContext io_context_scope(&io_ctx); + std::vector<snii::format::DictEntry> entries; + for (uint32_t block = 0; block < logical->n_dict_blocks(); ++block) { + RETURN_IF_ERROR(check_cancelled(options)); + uint64_t frq_base = 0; + uint64_t prx_base = 0; + RETURN_IF_ERROR(logical->decode_dict_block(block, &entries, &frq_base, &prx_base)); + for (const auto& entry : entries) { + if (entry.kind == snii::format::DictEntryKind::kPodRef) { + *position_bytes += cast_set<int64_t>(entry.prx_len); + } + } + } + return Status::OK(); +} + +int64_t merge_component(int64_t lhs, int64_t rhs) { + return lhs < 0 || rhs < 0 ? -1 : lhs + rhs; +} + +} // namespace + +std::vector<IndexDiskUsageRow> aggregate_index_disk_usage(std::vector<IndexDiskUsageRow> rows, + IndexDiskUsageLevel level) { + if (level == IndexDiskUsageLevel::kSegment) { + return rows; + } + using Key = std::tuple<std::string, int64_t, std::string, int, int>; + std::map<Key, size_t> positions; + std::vector<IndexDiskUsageRow> merged; + for (auto& row : rows) { + row.segment_id = -1; + if (level == IndexDiskUsageLevel::kTablet) { + row.rowset_id.clear(); + } + Key key {row.rowset_id, row.record.index_id, row.record.index_suffix, + static_cast<int>(row.format), static_cast<int>(row.record.structure)}; + auto [it, inserted] = positions.emplace(std::move(key), merged.size()); + if (inserted) { + merged.push_back(std::move(row)); + continue; + } + IndexDiskUsageRow& target = merged[it->second]; + target.segment_count += row.segment_count; + target.row_count += row.row_count; + IndexDiskUsageRecord& dst = target.record; + const IndexDiskUsageRecord& src = row.record; + dst.total_bytes += src.total_bytes; + dst.dict_bytes = merge_component(dst.dict_bytes, src.dict_bytes); + dst.posting_bytes = merge_component(dst.posting_bytes, src.posting_bytes); + dst.position_bytes = merge_component(dst.position_bytes, src.position_bytes); + dst.stats_bytes = merge_component(dst.stats_bytes, src.stats_bytes); + dst.other_bytes = merge_component(dst.other_bytes, src.other_bytes); + } + return merged; +} + +void classify_clucene_file(std::string_view name, int64_t length, IndexDiskUsageRecord* record) { + record->total_bytes += length; + if (is_bkd_file(name)) { + record->structure = IndexDiskUsageStructure::kBkd; + return; + } + if (is_ann_file(name)) { + record->structure = IndexDiskUsageStructure::kAnn; + return; + } + const size_t dot = name.rfind('.'); + const std::string_view extension = + dot == std::string_view::npos ? std::string_view() : name.substr(dot + 1); + if (extension == "tis" || extension == "tii") { + record->dict_bytes += length; + } else if (extension == "frq") { + record->posting_bytes += length; + } else if (extension == "prx") { + record->position_bytes += length; + } else if (extension == "nrm") { + record->stats_bytes += length; + } else { + record->other_bytes += length; + } +} + +IndexDiskUsageCollector::IndexDiskUsageCollector(io::FileSystemSPtr fs, + std::string index_path_prefix, + TabletSchemaSPtr schema, + InvertedIndexStorageFormatPB format, + int64_t tablet_id, + InvertedIndexFileInfo index_file_info) + : _fs(std::move(fs)), + _index_path_prefix(std::move(index_path_prefix)), + _schema(std::move(schema)), + _format(format), + _tablet_id(tablet_id), + _index_file_info(std::move(index_file_info)) {} + +Status IndexDiskUsageCollector::collect(const IndexDiskUsageOptions& options, + std::vector<IndexDiskUsageRecord>* out) { + DORIS_CHECK(out != nullptr); + switch (_format) { + case InvertedIndexStorageFormatPB::V1: + return _collect_v1(options, out); + case InvertedIndexStorageFormatPB::V2: + case InvertedIndexStorageFormatPB::V3: + return _collect_compound(options, out); + case InvertedIndexStorageFormatPB::SNII: + return _collect_snii(options, out); + default: + return Status::NotSupported("index disk usage does not support inverted index format {}", + InvertedIndexStorageFormatPB_Name(_format)); + } +} + +Status IndexDiskUsageCollector::_collect_v1(const IndexDiskUsageOptions& options, + std::vector<IndexDiskUsageRecord>* out) { + IndexFileReader reader(_fs, _index_path_prefix, _format, _index_file_info, _tablet_id); + RETURN_IF_ERROR(reader.init()); + std::vector<TabletIndex> file_indexes; + for (const V1IndexFile& file : list_v1_index_files(*_schema, _index_file_info, &file_indexes)) { Review Comment: [P2] Observe cancellation while enumerating physical indexes The cancellation check before each segment does not cover these per-index loops. A valid VARIANT segment can inherit the parent index for every extracted path (configured counts can reach 100000, and zero means unbounded), yet V1 walks every file, V2/V3 first materialize every compound directory, and SNII performs a core-metadata read per selected text entry without checking cancellation. With the default `position_detail=false`, the dictionary-block check added for the earlier thread is never reached, so a cancelled remote scan can continue many reads before returning. Please check at least once per physical entry and make the V2/V3 bulk-directory enumeration cancellable/filtered, with a many-entry cancellation test. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
