github-actions[bot] commented on code in PR #67977: URL: https://github.com/apache/doris/pull/67977#discussion_r4012971709
########## fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/IndexDiskUsageScanNode.java: ########## @@ -0,0 +1,244 @@ +// 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.datasource.tvf.source; + +import org.apache.doris.analysis.TupleDescriptor; +import org.apache.doris.catalog.DiskInfo; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.Replica; +import org.apache.doris.catalog.Tablet; +import org.apache.doris.cloud.catalog.CloudReplica; +import org.apache.doris.cloud.system.CloudSystemInfoService; +import org.apache.doris.common.Config; +import org.apache.doris.common.UserException; +import org.apache.doris.planner.OlapScanNode; +import org.apache.doris.planner.PlanNodeId; +import org.apache.doris.planner.ScanContext; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.resource.computegroup.ComputeGroup; +import org.apache.doris.system.Backend; +import org.apache.doris.system.SystemInfoService; +import org.apache.doris.tablefunction.IndexDiskUsageTableValuedFunction; +import org.apache.doris.tablefunction.IndexDiskUsageTableValuedFunction.TabletTarget; +import org.apache.doris.thrift.TMetaScanRange; +import org.apache.doris.thrift.TNetworkAddress; +import org.apache.doris.thrift.TScanRange; +import org.apache.doris.thrift.TScanRangeLocation; +import org.apache.doris.thrift.TScanRangeLocations; + +import com.google.common.base.Preconditions; +import com.google.common.collect.Lists; +import com.google.common.collect.Maps; + +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.LongFunction; +import java.util.function.Predicate; +import java.util.stream.Collectors; + +/** + * Scan node of index_disk_usage. Each tablet is read by one backend that holds it, and the tablets + * of a backend are split into a few scan ranges so that its scanners can read them in parallel. + */ +public class IndexDiskUsageScanNode extends MetadataScanNode { + + /** + * Picks the backend that reads one tablet. + */ + @FunctionalInterface + interface BackendSelector { + long select(TabletTarget target) throws UserException; + } + + private final IndexDiskUsageTableValuedFunction tvf; + private final List<TScanRangeLocations> scanRanges = Lists.newArrayList(); + + public IndexDiskUsageScanNode(PlanNodeId id, TupleDescriptor desc, IndexDiskUsageTableValuedFunction tvf, + ScanContext scanContext) { + super(id, desc, tvf, scanContext); + this.tvf = tvf; + } + + @Override + public void init() throws UserException { + super.init(); + SystemInfoService systemInfo = Env.getCurrentSystemInfo(); + ConnectContext context = ConnectContext.get(); + BackendSelector selector; + if (Config.isCloudMode()) { + selector = cloudSelector(systemInfo); + } else { + selector = localSelector(systemInfo::getBackend, + queryableIn(context == null ? null : context.getComputeGroupSafely())); + } + Map<Long, List<TabletTarget>> groups = groupByBackend(tvf.getTabletTargets(), selector); + TMetaScanRange template = tvf.getMetaScanRange(Lists.newArrayList()); + int rangesPerBackend = context == null ? 1 : context.getSessionVariable().getMaxScannersConcurrency(); + scanRanges.clear(); + scanRanges.addAll(buildScanRangeLocations(template, groups, systemInfo::getBackend, rangesPerBackend)); + numNodes = scanRanges.size(); + } + + @Override + protected void initBackendPolicy() { + // Tablet replicas decide where this scan runs, so the external file backend policy, which + // also requires load-available backends, does not apply. + } + + @Override + protected void createScanRangeLocations() { + // Scan ranges are built in init(), where replica selection can report a user error. + } + + @Override + public List<TScanRangeLocations> getScanRangeLocations(long maxScanRangeLength) { + return scanRanges; + } + + @Override + public int getNumInstances() { + return scanRanges.size(); + } + + static Map<Long, List<TabletTarget>> groupByBackend(List<TabletTarget> targets, BackendSelector selector) + throws UserException { + Map<Long, List<TabletTarget>> groups = Maps.newLinkedHashMap(); + for (TabletTarget target : targets) { + groups.computeIfAbsent(selector.select(target), backendId -> Lists.newArrayList()).add(target); + } + return groups; + } + + // Applies the replica rule of OLAP scans: a queryable mix node inside the caller's compute group. + static Predicate<Backend> queryableIn(ComputeGroup computeGroup) { + boolean invalidComputeGroup = ComputeGroup.INVALID_COMPUTE_GROUP.equals(computeGroup); + boolean notCloudComputeGroup = computeGroup != null && !Config.isCloudMode(); + return backend -> backend.isQueryAvailable() && backend.isMixNode() + && !OlapScanNode.shouldFilterReplicaByResourceTag(invalidComputeGroup, notCloudComputeGroup, + computeGroup, backend.getLocationTag().value); + } + + // Spreads tablets over their eligible backends by tablet id, so one backend does not read a + // whole table while the choice stays deterministic. + static long chooseBackend(long tabletId, List<Replica> replicas, LongFunction<Backend> backendLookup, + Predicate<Backend> eligible) throws UserException { + List<Long> candidates = replicas.stream() + .map(Replica::getBackendIdWithoutException) + .filter(backendId -> { + Backend backend = backendLookup.apply(backendId); + return backend != null && eligible.test(backend); + }) + .sorted() + .collect(Collectors.toList()); + if (candidates.isEmpty()) { + throw new UserException("No queryable replica for tablet " + tabletId); + } + return candidates.get((int) Math.floorMod(tabletId, (long) candidates.size())); + } + + // Splits the tablets of each backend into at most `rangesPerBackend` consecutive ranges, the + // way MetadataScanNode splits serialized splits by scanner concurrency. + static List<TScanRangeLocations> buildScanRangeLocations(TMetaScanRange template, + Map<Long, List<TabletTarget>> groups, LongFunction<Backend> backendLookup, int rangesPerBackend) { + Map<Long, String> partitionNames = template.getIndexDiskUsageParams().getPartitionNames(); + // Drop the table-wide lists once, so each range copy only carries its own tablets and partitions. + TMetaScanRange base = template.deepCopy(); + base.getIndexDiskUsageParams().unsetTablets(); + base.getIndexDiskUsageParams().unsetPartitionNames(); + List<TScanRangeLocations> ranges = Lists.newArrayList(); + for (Map.Entry<Long, List<TabletTarget>> group : groups.entrySet()) { + Backend backend = backendLookup.apply(group.getKey()); + Preconditions.checkState(backend != null, "backend %s is not found", group.getKey()); + List<TabletTarget> tablets = group.getValue(); + int chunkSize = (int) Math.ceil((double) tablets.size() / Math.max(1, rangesPerBackend)); + for (int from = 0; from < tablets.size(); from += chunkSize) { + List<TabletTarget> chunk = tablets.subList(from, Math.min(from + chunkSize, tablets.size())); + ranges.add(buildScanRange(base, chunk, partitionNames, backend)); + } + } + return ranges; + } + + private static TScanRangeLocations buildScanRange(TMetaScanRange base, List<TabletTarget> tablets, + Map<Long, String> partitionNames, Backend backend) { + TMetaScanRange metaScanRange = base.deepCopy(); + metaScanRange.getIndexDiskUsageParams().setTablets( + tablets.stream().map(TabletTarget::toThrift).collect(Collectors.toList())); + if (partitionNames != null) { + Map<Long, String> names = Maps.newHashMap(); + for (TabletTarget tablet : tablets) { + String name = partitionNames.get(tablet.getPartitionId()); + if (name != null) { + names.put(tablet.getPartitionId(), name); + } + } + metaScanRange.getIndexDiskUsageParams().setPartitionNames(names); + } + + TScanRange scanRange = new TScanRange(); + scanRange.setMetaScanRange(metaScanRange); + TScanRangeLocation location = new TScanRangeLocation(); + location.setBackendId(backend.getId()); + location.setServer(new TNetworkAddress(backend.getHost(), backend.getBePort())); + TScanRangeLocations locations = new TScanRangeLocations(); + locations.addToLocations(location); + locations.setScanRange(scanRange); + return locations; + } + + static BackendSelector localSelector(LongFunction<Backend> backendLookup, Predicate<Backend> eligible) { + // Tablets share backends, so the alive disks of each backend are collected once per scan. + Map<Long, Set<Long>> alivePathHashes = Maps.newHashMap(); + return target -> { + Tablet tablet = target.getTablet(); + for (Replica replica : tablet.getReplicas()) { + long backendId = replica.getBackendIdWithoutException(); + if (!alivePathHashes.containsKey(backendId)) { + Backend backend = backendLookup.apply(backendId); + if (backend != null) { + alivePathHashes.put(backendId, alivePathHashes(backend)); + } + } + } + List<Replica> replicas = tablet.getQueryableReplicas(target.getVersion(), alivePathHashes, false); + return chooseBackend(target.getTabletId(), replicas, backendLookup, eligible); + }; + } + + private static BackendSelector cloudSelector(SystemInfoService systemInfo) throws UserException { + String clusterId = ((CloudSystemInfoService) systemInfo).getCurrentClusterId(); + return target -> { + for (Replica replica : target.getTablet().getReplicas()) { + long backendId = ((CloudReplica) replica).getBackendIdWithClusterId(clusterId); + Backend backend = systemInfo.getBackend(backendId); + if (backend != null && backend.isQueryAvailable()) { Review Comment: [P1] Exclude an old smooth-upgrade source from this new metadata scan. `CloudReplica` deliberately retains the source BE as a secondary query fallback until the new destination is alive, and `getBackendIdWithClusterId` can return it because that cached-primary/secondary availability check does not reject `isSmoothUpgradeSrc`; this selector also accepts it solely on `isQueryAvailable()`. A pre-feature source BE deserializes the new `TMetadataType` value, misses the `INDEX_DISK_USAGE` `_open_impl` branch, and reaches the existing default path that sets EOF and returns OK, so this supported cloud upgrade state silently drops the affected tablets from the result. Reject/reroute smooth-upgrade sources, and gate any other supported mixed-version path, until the backend supports this reader; test an unavailable destination with the old source retained as secondary. ########## 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; Review Comment: [P2] Do not synchronize delete bitmaps for this metadata-only scan. `SyncOptions` defaults `sync_delete_bitmap` to true, so every running MoW tablet can download, merge, persist/log, and validate its delete bitmap here even though the reader only inspects rowset, segment, and index metadata. The preceding `ExecEnv::get_tablet` call also hard-codes delete-bitmap sync on a cloud cache miss. Large MoW tablets therefore add substantial unrelated work and can make this TVF fail on a delete-bitmap error that cannot affect its result. Use a no-delete-bitmap tablet lookup and set this versioned sync option to false; add a cloud MoW test proving the scan does not invoke delete-bitmap synchronization. ########## 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: [P2] Make this cloud synchronization cancellation-aware. The outer scanner checks `RuntimeState` only before entering `_collect_tablet`; both a cache-miss `ExecEnv::get_tablet` and this stale-version sync can then block in `CloudMetaMgr::sync_tablet_rowsets_unlocked`, whose synchronous `get_rowset` call uses a 10-second timeout and up to 20 retries plus backoff without receiving the query cancellation signal. Cancelling this TVF after the call starts can therefore retain the scanner for roughly 210 seconds or more. This remains even if ranges are parallelized, because one in-flight sync is still uninterruptible. Thread cancellation through the cold lookup and versioned sync, check it between retries, and arrange prompt cancellation or bounding of the active RPC; add a blocked-`get_rowset` cancellation test. ########## be/src/storage/index/index_disk_usage.cpp: ########## @@ -0,0 +1,460 @@ +// 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); + // A rowset returns no file system when its tablet or storage resource cannot be resolved. + if (_fs == nullptr) { + return Status::Error<ErrorCode::INIT_FAILED>("no file system for inverted index files {}", + _index_path_prefix); + } + 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)) { + const TabletIndex& index = *file.index; + if (!is_wanted(options, index.index_id())) { + continue; + } + RETURN_IF_ERROR(check_cancelled(options)); + int64_t file_size = file.persisted_size; + if (file_size < 0) { + const std::string path = InvertedIndexDescriptor::get_index_file_path_v1( + _index_path_prefix, index.index_id(), index.get_index_suffix()); + const Status size_status = _fs->file_size(path, &file_size); + if (is_absent_index_file(size_status)) { + continue; + } + RETURN_IF_ERROR(size_status); + } + auto directory = reader.open(&index); + if (!directory.has_value()) { + if (is_absent_index_file(directory.error())) { + continue; + } + return directory.error(); + } + + IndexDiskUsageRecord record; + record.index_id = index.index_id(); + record.index_suffix = index.get_index_suffix(); + int64_t files_bytes = 0; + RETURN_IF_ERROR(add_directory_files(*directory.value(), &record, &files_bytes)); + // Each V1 index owns its file, so its compound header is part of the index. + record.other_bytes += file_size - files_bytes; + record.total_bytes += file_size - files_bytes; + out->push_back(std::move(record)); + } + return Status::OK(); +} + +Status IndexDiskUsageCollector::_collect_compound(const IndexDiskUsageOptions& options, + std::vector<IndexDiskUsageRecord>* out) { + IndexFileReader reader(_fs, _index_path_prefix, _format, _index_file_info, _tablet_id); + if (const Status st = reader.init(); !st.ok()) { + return is_absent_index_file(st) ? Status::OK() : st; + } + auto directories = DORIS_TRY(reader.get_all_directories()); + + // Every index counts toward the attributed bytes, even the filtered ones, so the container + // record only holds the shared header. + int64_t attributed_bytes = 0; + for (const auto& [key, directory] : directories) { + IndexDiskUsageRecord record; + record.index_id = key.first; + record.index_suffix = key.second; + RETURN_IF_ERROR(add_directory_files(*directory, &record, &attributed_bytes)); + if (is_wanted(options, record.index_id)) { + out->push_back(std::move(record)); + } + } + if (options.index_ids.empty()) { + IndexDiskUsageRecord container; + container.structure = IndexDiskUsageStructure::kContainer; + container.total_bytes = reader.get_inverted_file_size() - attributed_bytes; + container.other_bytes = container.total_bytes; + out->push_back(std::move(container)); + } + return Status::OK(); +} + +Status IndexDiskUsageCollector::_collect_snii(const IndexDiskUsageOptions& options, + std::vector<IndexDiskUsageRecord>* out) { + IndexFileReader reader(_fs, _index_path_prefix, _format, _index_file_info, _tablet_id); + if (const Status st = reader.init(); !st.ok()) { + return is_absent_index_file(st) ? Status::OK() : st; + } + const auto entries = DORIS_TRY(reader.snii_logical_indexes()); + + int64_t attributed_bytes = 0; + for (const auto& entry : entries) { + const auto index_id = cast_set<int64_t>(entry.index_id); + // The container record is only reported without a filter, so filtered-out indexes + // need no metadata read. + if (!is_wanted(options, index_id)) { + continue; + } + RETURN_IF_ERROR(check_cancelled(options)); + IndexDiskUsageRecord record; + record.index_id = index_id; + record.index_suffix = entry.index_suffix; + if (entry.kind == snii::format::LogicalIndexKind::kInverted) { + snii::format::CoreMetadata core; + RETURN_IF_ERROR(reader.snii_core_metadata(entry.index_id, entry.index_suffix, &core)); + const auto& refs = core.section_refs; + record.structure = IndexDiskUsageStructure::kTerm; + record.dict_bytes = + cast_set<int64_t>(refs.dict_region.length + entry.sampled_term_index.length + + entry.dict_block_directory.length + refs.bsbf.length); + record.posting_bytes = cast_set<int64_t>(refs.posting_region.length); + record.stats_bytes = cast_set<int64_t>(entry.core_metadata.length + refs.norms.length); + record.other_bytes = cast_set<int64_t>(refs.null_bitmap.length); + if (!snii::format::has_positions(core.index_config)) { + record.position_bytes = 0; + } else if (options.position_detail) { + int64_t position_bytes = 0; + RETURN_IF_ERROR(sum_snii_position_bytes(reader, entry.index_id, entry.index_suffix, + options, &position_bytes)); + record.position_bytes = position_bytes; + record.posting_bytes -= position_bytes; + } else { + record.position_bytes = -1; + } + record.total_bytes = record.dict_bytes + record.posting_bytes + + std::max<int64_t>(record.position_bytes, 0) + record.stats_bytes + + record.other_bytes; + } else { + record.structure = entry.kind == snii::format::LogicalIndexKind::kBkd + ? IndexDiskUsageStructure::kBkd + : IndexDiskUsageStructure::kAnn; + for (const auto& file : entry.files) { + record.total_bytes += cast_set<int64_t>(file.length); Review Comment: [P2] Reject overlapping SNII blob extents before accounting them. The directory decoder verifies fields/overflow and duplicate names, while `validate_blob_files` only checks that each named ANN/BKD range ends before the metadata directory; two differently named, positive-length files may still reference the same in-bounds bytes. This loop then counts that range twice, and the unfiltered path can emit an inflated index total plus a negative container total instead of reporting corruption. Validate the physical extents globally as disjoint while preserving legal zero-length BKD files, use checked accumulation/subtraction, and add a CRC-valid directory case with overlapping named blobs. ########## 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()); Review Comment: [P1] Include temporary partitions in this physical-usage inventory. `OlapTable.getPartitions()` explicitly excludes them, and the filtered branch also calls `getPartition(name, false)`, so neither an unfiltered query nor an explicit partition filter can see their tablets. Temporary partitions hold real segment/index files while `INSERT OVERWRITE` or `REPLACE PARTITION` is staged, and detailed `SHOW DATA` correspondingly accounts through `getAllPartitions()`; a table-sized staged overwrite can therefore consume disk while this TVF reports none of it despite promising every physical inverted index of the table. Include temporary partitions by default and provide an unambiguous way to select them, or narrow the documented contract if only formal query-visible data is intended; add a loaded temporary-partition case. -- 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]
