github-actions[bot] commented on code in PR #67977: URL: https://github.com/apache/doris/pull/67977#discussion_r4013505923
########## be/src/format/table/index_disk_usage_reader.cpp: ########## @@ -0,0 +1,373 @@ +// 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 "format/table/index_disk_usage_reader.h" + +#include <boost/algorithm/string/case_conv.hpp> +#include <shared_mutex> +#include <string> +#include <string_view> +#include <utility> +#include <variant> + +#include "cloud/cloud_tablet.h" +#include "common/cast_set.h" +#include "core/block/block.h" +#include "core/column/column_nullable.h" +#include "core/column/column_string.h" +#include "core/column/column_vector.h" +#include "runtime/exec_env.h" +#include "runtime/runtime_state.h" +#include "storage/rowset/rowset.h" +#include "storage/tablet/base_tablet.h" +#include "storage/tablet/tablet_schema.h" + +namespace doris { + +using segment_v2::IndexDiskUsageLevel; +using segment_v2::IndexDiskUsageRecord; +using segment_v2::IndexDiskUsageRow; +using segment_v2::IndexDiskUsageStructure; + +namespace { + +const std::vector<std::pair<std::string_view, IndexDiskUsageReader::Column>>& column_names() { + using C = IndexDiskUsageReader::Column; + static const std::vector<std::pair<std::string_view, C>> names = { + {"PARTITION_NAME", C::kPartitionName}, + {"MATERIALIZED_INDEX_NAME", C::kMaterializedIndexName}, + {"TABLET_ID", C::kTabletId}, + {"BACKEND_ID", C::kBackendId}, + {"ROWSET_ID", C::kRowsetId}, + {"SEGMENT_ID", C::kSegmentId}, + {"INDEX_ID", C::kIndexId}, + {"INDEX_NAME", C::kIndexName}, + {"INDEX_TYPE", C::kIndexType}, + {"COLUMN_NAME", C::kColumnName}, + {"INDEX_SUFFIX", C::kIndexSuffix}, + {"STRUCTURE", C::kStructure}, + {"STORAGE_FORMAT", C::kStorageFormat}, + {"SEGMENT_COUNT", C::kSegmentCount}, + {"ROW_COUNT", C::kRowCount}, + {"TOTAL_BYTES", C::kTotalBytes}, + {"DICT_BYTES", C::kDictBytes}, + {"POSTING_BYTES", C::kPostingBytes}, + {"POSITION_BYTES", C::kPositionBytes}, + {"STATS_BYTES", C::kStatsBytes}, + {"OTHER_BYTES", C::kOtherBytes}, + {"STATS_SOURCE", C::kStatsSource}, + }; + return names; +} + +Result<IndexDiskUsageReader::Column> column_of(const std::string& slot_name) { + const std::string upper = boost::to_upper_copy(slot_name); + for (const auto& [name, column] : column_names()) { + if (upper == name) { + return column; + } + } + return ResultError(Status::InternalError("unknown index_disk_usage column {}", slot_name)); +} + +Result<IndexDiskUsageLevel> parse_level(const std::string& level) { + if (level == "tablet") { + return IndexDiskUsageLevel::kTablet; + } + if (level == "rowset") { + return IndexDiskUsageLevel::kRowset; + } + if (level == "segment") { + return IndexDiskUsageLevel::kSegment; + } + return ResultError(Status::InvalidArgument("unsupported index_disk_usage level {}", level)); +} + +std::string_view structure_name(IndexDiskUsageStructure structure) { + switch (structure) { + case IndexDiskUsageStructure::kTerm: + return "TERM"; + case IndexDiskUsageStructure::kBkd: + return "BKD"; + case IndexDiskUsageStructure::kAnn: + return "ANN"; + case IndexDiskUsageStructure::kContainer: + return "CONTAINER"; + } + return "UNKNOWN"; +} + +void insert_null(IColumn* column) { + auto& nullable = reinterpret_cast<ColumnNullable&>(*column); + nullable.get_nested_column().insert_default(); + nullable.get_null_map_data().push_back(1); +} + +IColumn* non_null_nested(IColumn* column) { + auto& nullable = reinterpret_cast<ColumnNullable&>(*column); + nullable.get_null_map_data().push_back(0); + return nullable.get_nested_column_ptr().get(); +} + +void insert_int64(IColumn* column, int64_t value) { + assert_cast<ColumnInt64*>(non_null_nested(column))->insert_value(value); +} + +void insert_int32(IColumn* column, int32_t value) { + assert_cast<ColumnInt32*>(non_null_nested(column))->insert_value(value); +} + +void insert_string(IColumn* column, std::string_view value) { + assert_cast<ColumnString*>(non_null_nested(column))->insert_data(value.data(), value.size()); +} + +} // namespace + +IndexDiskUsageReader::IndexDiskUsageReader(std::vector<SlotDescriptor*> slots, RuntimeState* state, + RuntimeProfile* /*profile*/, TMetaScanRange scan_range) + : _state(state), _slots(std::move(slots)), _scan_range(std::move(scan_range)) {} + +Status IndexDiskUsageReader::init_reader() { + if (!_scan_range.__isset.index_disk_usage_params) { + return Status::InvalidArgument("index_disk_usage scan range has no parameters"); + } + const TIndexDiskUsageMetadataParams& params = _scan_range.index_disk_usage_params; + _level = DORIS_TRY(parse_level(params.level)); + _options.position_detail = params.position_detail; + _options.index_ids.insert(params.index_ids.begin(), params.index_ids.end()); + _options.check_cancelled = [state = _state]() { + RETURN_IF_CANCELLED(state); + return Status::OK(); + }; + _slot_columns.clear(); + for (const SlotDescriptor* slot : _slots) { + const Column column = DORIS_TRY(column_of(slot->col_name())); + _slot_columns.push_back(column); + } + return Status::OK(); +} + +Status IndexDiskUsageReader::_do_get_next_block(Block* block, size_t* read_rows, bool* eof) { + const auto& tablets = _scan_range.index_disk_usage_params.tablets; + *read_rows = 0; + while (_next_tablet < tablets.size()) { + RETURN_IF_CANCELLED(_state); + const TIndexDiskUsageTablet& target = tablets[_next_tablet++]; + std::vector<IndexDiskUsageRow> rows; + TabletSchemaSPtr current_schema; + RETURN_IF_ERROR(_collect_tablet(target, &rows, ¤t_schema)); + rows = segment_v2::aggregate_index_disk_usage(std::move(rows), _level); + if (rows.empty()) { + continue; + } + RETURN_IF_ERROR(_fill_block(block, target, *current_schema, rows)); + *read_rows = rows.size(); + *eof = false; + return Status::OK(); + } + *eof = true; + return Status::OK(); +} + +Status IndexDiskUsageReader::_collect_tablet(const TIndexDiskUsageTablet& target, + std::vector<IndexDiskUsageRow>* rows, + TabletSchemaSPtr* current_schema) const { + BaseTabletSPtr tablet = DORIS_TRY(ExecEnv::get_tablet(target.tablet_id)); + if (auto cloud_tablet = std::dynamic_pointer_cast<CloudTablet>(tablet)) { + SyncOptions options; + options.query_version = target.version; + RETURN_IF_ERROR(cloud_tablet->sync_rowsets(options)); Review Comment: [P1] Refresh same-version compaction replacements before collecting physical usage. On a cloud cache hit, `_max_version` can already equal `target.version`, so this query-version sync returns before sending the cached compaction counters to meta-service. If another BE/compute group compacts the tablet, the visible version stays V while its authoritative rowsets and index files change; this TVF then reports stale pre-compaction rowset IDs and byte totals until periodic tablet refresh (1800-second default), or can fail after those files are recycled. Use a freshness path that reaches compaction-counter reconciliation even when V is cached (without the unrelated delete-bitmap work), and add a cache-hit test that replaces rowsets at the same visible version. ########## fe/fe-core/src/main/java/org/apache/doris/tablefunction/IndexDiskUsageTableValuedFunction.java: ########## @@ -0,0 +1,391 @@ +// 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(); + List<Long> versions = null; + 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); + } + // Local replica choice filters replicas by version, so read it with the tablets it applies to. + if (!Config.isCloudMode()) { + versions = partitions.stream().map(Partition::getVisibleVersion).collect(Collectors.toList()); + } + } finally { + table.readUnlock(); + } + // Cloud versions come from meta-service, so they are fetched without holding the table lock. + if (versions == null) { + versions = cloudVisibleVersions(partitions); + } + List<TabletTarget> targets = Lists.newArrayList(); + for (int i = 0; i < partitions.size(); ++i) { + long partitionId = partitions.get(i).getId(); + for (Pair<Long, List<Tablet>> indexTablets : tabletsByPartition.get(partitionId)) { + for (Tablet tablet : indexTablets.second) { + targets.add(new TabletTarget(tablet, partitionId, indexTablets.first, versions.get(i))); + } + } + } + this.indexIds = ImmutableList.copyOf(resolvedIndexIds); + this.partitionNames = resolvedPartitionNames; + this.materializedIndexNames = resolvedMaterializedIndexNames; + this.tabletTargets = ImmutableList.copyOf(targets); + checkPositionDetailLimit(); + } + + public List<TabletTarget> getTabletTargets() { + return tabletTargets; + } + + @Override + public TMetadataType getMetadataType() { + return TMetadataType.INDEX_DISK_USAGE; + } + + @Override + public TMetaScanRange getMetaScanRange(List<String> requiredFields) { + TIndexDiskUsageMetadataParams params = new TIndexDiskUsageMetadataParams(); + params.setLevel(level); + params.setPositionDetail(positionDetail); + params.setIndexIds(Lists.newArrayList(indexIds)); + params.setPartitionNames(Maps.newHashMap(partitionNames)); + params.setMaterializedIndexNames(Maps.newHashMap(materializedIndexNames)); + params.setTablets(tabletTargets.stream().map(TabletTarget::toThrift).collect(Collectors.toList())); + TMetaScanRange metaScanRange = new TMetaScanRange(); + metaScanRange.setMetadataType(TMetadataType.INDEX_DISK_USAGE); + metaScanRange.setIndexDiskUsageParams(params); + return metaScanRange; + } + + @Override + public ScanNode getScanNode(PlanNodeId id, TupleDescriptor desc, SessionVariable sv) { + return new IndexDiskUsageScanNode(id, desc, this, + ScanContext.builder().clusterName(sv.resolveCloudClusterName()).build()); + } + + @Override + public String getTableName() { + return "IndexDiskUsageTableValuedFunction"; + } + + @Override + public List<Column> getTableColumns() { + return SCHEMA; + } + + private static Column varcharColumn(String name) { + return new Column(name, ScalarType.createVarcharType(ScalarType.MAX_VARCHAR_LENGTH), true); + } + + private static Column bigintColumn(String name) { + return new Column(name, ScalarType.createType(PrimitiveType.BIGINT), true); + } + + private static String parseLevel(String raw) { + String normalized = raw.toLowerCase(); + if (!LEVELS.contains(normalized)) { + throw new AnalysisException("Unsupported level '" + raw + + "' for index_disk_usage, expected tablet, rowset or segment"); + } + return normalized; + } + + private static boolean parsePositionDetail(String raw) { + if ("true".equalsIgnoreCase(raw)) { + return true; + } + if ("false".equalsIgnoreCase(raw)) { + return false; + } + throw new AnalysisException("Invalid position_detail '" + raw + "', expected true or false"); + } + + private static void checkShowPrivilege(String dbName, String tableName) { + ConnectContext ctx = ConnectContext.get(); + if (!Env.getCurrentEnv().getAccessManager().checkTblPriv(ctx, InternalCatalog.INTERNAL_CATALOG_NAME, + dbName, tableName, PrivPredicate.SHOW)) { + throw new AnalysisException(ErrorCode.ERR_TABLEACCESS_DENIED_ERROR.formatErrorMsg("SHOW", + ctx.getQualifiedUser(), ctx.getRemoteIP(), dbName + ": " + tableName)); + } + } + + private static OlapTable getOlapTable(String dbName, String tableName) { + TableIf table; + try { + Database db = Env.getCurrentInternalCatalog().getDbOrAnalysisException(dbName); + table = db.getTableOrAnalysisException(tableName); + } catch (org.apache.doris.common.AnalysisException e) { + throw new AnalysisException(e.getMessage(), e); + } + if (!(table instanceof OlapTable)) { + throw new AnalysisException("index_disk_usage only supports OLAP table"); + } + return (OlapTable) table; + } + + private static List<Partition> resolvePartitions(OlapTable table, String raw, String qualifiedName) { + if (StringUtils.isEmpty(raw)) { + return Lists.newArrayList(table.getPartitions()); + } + List<Partition> partitions = Lists.newArrayList(); + for (String name : splitNames(raw)) { + // Temporary partitions are excluded, so look the name up among formal partitions only. + Partition partition = table.getPartition(name, false); + if (partition == null) { + throw new AnalysisException("Unknown partition '" + name + "' in table " + qualifiedName); + } + partitions.add(partition); + } + return partitions; + } + + private static List<Long> resolveIndexIds(OlapTable table, String raw, String qualifiedName) { + if (StringUtils.isEmpty(raw)) { + return Lists.newArrayList(); + } + List<Long> ids = Lists.newArrayList(); + for (String name : splitNames(raw)) { Review Comment: [P2] Reject explicit filters that contain no names. Values such as `"indexes" = " "` or `"indexes" = ","` pass `StringUtils.isEmpty`, but `splitNames` removes every token and this method serializes an empty `index_ids` list, which BE deliberately interprets as all indexes. The analogous `partitions` value silently creates no scan targets. A malformed narrowing request can therefore become an expensive full-index scan or an unexplained empty result; validate the tokenized collection before resolving it and cover whitespace/delimiter-only values for both properties. ########## fe/fe-core/src/main/java/org/apache/doris/tablefunction/IndexDiskUsageTableValuedFunction.java: ########## @@ -0,0 +1,391 @@ +// 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(); + List<Long> versions = null; + 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)) { Review Comment: [P1] Include populated shadow indexes in this physical-usage inventory. `IndexExtState.VISIBLE` excludes `idToShadowIndex`, but schema change registers those tablets and index metadata, loads into them, and waits for their replicas to catch up to the partition visible version before promotion. During a long schema change, another table-sized set of inverted-index files can therefore exist while this TVF—documented to report every physical index—silently omits it. Enumerate non-row-binlog shadow indexes and route their caught-up `ALTER` replicas through an inventory-specific eligibility path (or explicitly narrow the contract to query-visible generations); add an in-progress schema-change case with populated shadow tablets. -- 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]
