github-actions[bot] commented on code in PR #67977: URL: https://github.com/apache/doris/pull/67977#discussion_r4011441696
########## fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/IndexDiskUsageScanNode.java: ########## @@ -0,0 +1,211 @@ +// 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 every + * backend gets a single scan range carrying only its own tablets. + */ +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(); + BackendSelector selector = Config.isCloudMode() ? cloudSelector(systemInfo) : localSelector(systemInfo); + Map<Long, List<TabletTarget>> groups = groupByBackend(tvf.getTabletTargets(), selector); + TMetaScanRange template = tvf.getMetaScanRange(Lists.newArrayList()); + scanRanges.clear(); + scanRanges.addAll(buildScanRangeLocations(template, groups, systemInfo::getBackend)); + 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())); + } + + static List<TScanRangeLocations> buildScanRangeLocations(TMetaScanRange template, + Map<Long, List<TabletTarget>> groups, LongFunction<Backend> backendLookup) { + // Drop the table-wide tablet list once, so each backend copy only carries its own tablets. + TMetaScanRange base = template.deepCopy(); + base.getIndexDiskUsageParams().unsetTablets(); + 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()); + TMetaScanRange metaScanRange = base.deepCopy(); + metaScanRange.getIndexDiskUsageParams().setTablets( + group.getValue().stream().map(TabletTarget::toThrift).collect(Collectors.toList())); + + 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); Review Comment: [P2] Preserve eligible replica alternatives instead of publishing only the selected backend. The scheduler's multi-location path skips a backend that became unavailable and chooses another replica, but its one-location path immediately throws `No available workers`. Here `chooseBackend`/grouping discards every alternative before assignment, so a heartbeat change can fail this TVF while the tablet still has a healthy replica. Please use tablet-sized ranges, or chunks whose locations are the common eligible backend intersection for every carried tablet, and cover a backend-state change before assignment. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/IndexDiskUsageScanNode.java: ########## @@ -0,0 +1,211 @@ +// 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 every + * backend gets a single scan range carrying only its own tablets. + */ +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(); + BackendSelector selector = Config.isCloudMode() ? cloudSelector(systemInfo) : localSelector(systemInfo); + Map<Long, List<TabletTarget>> groups = groupByBackend(tvf.getTabletTargets(), selector); + TMetaScanRange template = tvf.getMetaScanRange(Lists.newArrayList()); + scanRanges.clear(); + scanRanges.addAll(buildScanRangeLocations(template, groups, systemInfo::getBackend)); + 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())); + } + + static List<TScanRangeLocations> buildScanRangeLocations(TMetaScanRange template, + Map<Long, List<TabletTarget>> groups, LongFunction<Backend> backendLookup) { + // Drop the table-wide tablet list once, so each backend copy only carries its own tablets. + TMetaScanRange base = template.deepCopy(); + base.getIndexDiskUsageParams().unsetTablets(); + 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()); + TMetaScanRange metaScanRange = base.deepCopy(); + metaScanRange.getIndexDiskUsageParams().setTablets( Review Comment: [P2] Avoid putting the entire backend group into one range. `MetaScanLocalState` creates one `MetaScanner` per range, and this reader processes its tablets sequentially, including synchronous `CloudTablet::sync_rowsets`; a cold many-tablet cloud request therefore pays those metadata operations serially. Ordinary OLAP scans run per-tablet get/sync tasks with bounded `init_scanner_sync_rowsets_parallelism`. Split groups into bounded common-location chunks (also preserving MF-03's valid fallbacks), or add equivalent bounded pre-sync, so configured scanner parallelism can apply. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/IndexDiskUsageScanNode.java: ########## @@ -0,0 +1,211 @@ +// 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 every + * backend gets a single scan range carrying only its own tablets. + */ +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(); + BackendSelector selector = Config.isCloudMode() ? cloudSelector(systemInfo) : localSelector(systemInfo); + Map<Long, List<TabletTarget>> groups = groupByBackend(tvf.getTabletTargets(), selector); + TMetaScanRange template = tvf.getMetaScanRange(Lists.newArrayList()); + scanRanges.clear(); + scanRanges.addAll(buildScanRangeLocations(template, groups, systemInfo::getBackend)); + 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())); + } + + static List<TScanRangeLocations> buildScanRangeLocations(TMetaScanRange template, + Map<Long, List<TabletTarget>> groups, LongFunction<Backend> backendLookup) { + // Drop the table-wide tablet list once, so each backend copy only carries its own tablets. + TMetaScanRange base = template.deepCopy(); + base.getIndexDiskUsageParams().unsetTablets(); + 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()); + TMetaScanRange metaScanRange = base.deepCopy(); + metaScanRange.getIndexDiskUsageParams().setTablets( + group.getValue().stream().map(TabletTarget::toThrift).collect(Collectors.toList())); + + 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); + ranges.add(locations); + } + return ranges; + } + + private static BackendSelector localSelector(SystemInfoService systemInfo) { + ConnectContext context = ConnectContext.get(); + Predicate<Backend> eligible = queryableIn(context == null ? null : context.getComputeGroupSafely()); + return target -> { + Tablet tablet = target.getTablet(); + List<Replica> replicas = tablet.getQueryableReplicas(target.getVersion(), + alivePathHashes(tablet, systemInfo), false); Review Comment: [P2] `alivePathHashes` is rebuilt for every tablet, and each rebuild walks every replica's backend plus every disk on that backend. An unfiltered call plans the whole table, so replicated high-tablet-count tables repeatedly rescan identical disk maps before execution (O(tablets * replicas * disks)). `OlapScanNode.getBackendAlivePathHashes(Map, List)` already caches each backend with `computeIfAbsent`; please take/reuse one backend-path snapshot outside this per-tablet selector as well. ########## fe/fe-core/src/main/java/org/apache/doris/tablefunction/IndexDiskUsageTableValuedFunction.java: ########## @@ -0,0 +1,364 @@ +// 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.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.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"), + 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 base index tablet 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 version; + + public TabletTarget(Tablet tablet, long partitionId, long version) { + this.tablet = tablet; + this.partitionId = partitionId; + this.version = version; + } + + 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.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 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, 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()); + tabletsByPartition.put(partition.getId(), Lists.newArrayList(partition.getBaseIndex().getTablets())); Review Comment: [P1] Include visible rollup tablets here. A rollup is initially created without secondary indexes, but a supported light `ADD INDEX` calls `updateBaseIndexSchema`, which installs the index list on every non-base materialized-index meta; `OlapTableSink` then sends that rollup's `indexesDesc`, so later loads write inverted/ANN files on its tablets when the indexed column is present. This base-only list silently omits those physical bytes from the table-level result. Please enumerate all visible non-row-binlog materialized indexes and cover rollup -> light add index -> load. ########## be/src/storage/index/index_disk_usage.cpp: ########## @@ -0,0 +1,413 @@ +// 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" + +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>(); +} + +// Returns the V1 index file size persisted in the rowset meta, or -1 when it is not recorded. +int64_t persisted_v1_file_size(const InvertedIndexFileInfo& file_info, const TabletIndex& index) { + for (const auto& index_info : file_info.index_info()) { + if (index_info.index_id() == index.index_id() && + index_info.index_suffix() == index.get_index_suffix()) { + return index_info.index_file_size() > 0 ? index_info.index_file_size() : -1; + } + } + return -1; +} + +// 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()); + for (const TabletIndex* index : _schema->inverted_indexes()) { Review Comment: [P1] Enumerate the physical V1 `(index_id, suffix)` entries rather than only formal schema indexes. With local `enable_inverted_index_v1_for_variant`, the writer clones the parent index for every extracted VARIANT path and writes a separate suffixed file, while the rowset schema still exposes only the parent index. This loop probes the empty parent suffix, skips it as absent, and never visits the real files, so the TVF reports none of their usage. The persisted `InvertedIndexFileInfo.index_info` supplies the physical keys for new rowsets; please also provide a segment-derived fallback for legacy metadata and cover multiple extracted paths. ########## be/src/storage/index/index_disk_usage.cpp: ########## @@ -0,0 +1,413 @@ +// 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" + +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>(); +} + +// Returns the V1 index file size persisted in the rowset meta, or -1 when it is not recorded. +int64_t persisted_v1_file_size(const InvertedIndexFileInfo& file_info, const TabletIndex& index) { + for (const auto& index_info : file_info.index_info()) { + if (index_info.index_id() == index.index_id() && + index_info.index_suffix() == index.get_index_suffix()) { + return index_info.index_file_size() > 0 ? index_info.index_file_size() : -1; + } + } + return -1; +} + +// 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()); + for (const TabletIndex* index : _schema->inverted_indexes()) { + if (!is_wanted(options, index->index_id())) { + continue; + } + int64_t file_size = persisted_v1_file_size(_index_file_info, *index); + 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; + } + 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); + } + } + attributed_bytes += record.total_bytes; + 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 collect_rowset_index_disk_usage(const RowsetSharedPtr& rowset, + const IndexDiskUsageOptions& options, int64_t tablet_id, + std::vector<IndexDiskUsageRow>* rows) { + const TabletSchemaSPtr schema = rowset->tablet_schema(); + if (!schema->has_inverted_or_ann_index()) { + return Status::OK(); + } + const InvertedIndexStorageFormatPB format = schema->get_inverted_index_storage_format(); + const std::string rowset_id = rowset->rowset_id().to_string(); + // Rowsets written before per-segment row counts were persisted read them from the footers. + std::vector<uint32_t> footer_rows; + for (auto segment : rowset->segments()) { + RETURN_IF_ERROR(check_cancelled(options)); + const std::string segment_path = DORIS_TRY(segment.path()); + IndexDiskUsageCollector collector( + rowset->rowset_meta()->fs(), + std::string(InvertedIndexDescriptor::get_index_file_path_prefix(segment_path)), + schema, format, tablet_id, segment.inverted_index_file_info()); + std::vector<IndexDiskUsageRecord> records; + RETURN_IF_ERROR(collector.collect(options, &records)); + if (records.empty()) { + continue; + } + int64_t row_count = 0; + if (segment.has_num_rows()) { + row_count = segment.num_rows(); + } else { + if (footer_rows.empty()) { + OlapReaderStatistics stats; + RETURN_IF_ERROR(std::static_pointer_cast<BetaRowset>(rowset)->get_segment_num_rows( Review Comment: [P2] Make this legacy footer fallback cancellation-aware. `get_segment_num_rows` reaches `SegmentLoader::load_segments`, whose loop synchronously loads every segment footer without checking this query; the next `check_cancelled(options)` is reached only after the whole rowset finishes. On an old cloud rowset with many segments, cancellation can therefore leave all remaining remote footer reads running. Please check between footer loads, while keeping cancellation retryable—this helper initializes a shared cache under `DorisCallOnce`, so caching `Cancelled` as its permanent result would poison later queries. ########## be/src/format/table/index_disk_usage_reader.cpp: ########## @@ -0,0 +1,359 @@ +// 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}, {"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"; +} + +const TabletIndex* find_index(const TabletSchema& schema, const IndexDiskUsageRecord& record) { + for (const TabletIndex* index : schema.inverted_and_ann_indexes()) { + if (index->index_id() == record.index_id && + index->get_index_suffix() == record.index_suffix) { Review Comment: [P1] Resolve inherited VARIANT records to their logical parent instead of requiring the physical suffix in the current tablet schema. Direct-load VARIANT writers clone the parent index id and add the extracted path only to a write-local object; V2/V3 directories and SNII metadata preserve that suffix, but the current formal schema used here is not guaranteed to carry every inherited suffix. This exact match therefore makes valid rows report NULL `INDEX_NAME`, `INDEX_TYPE`, and `COLUMN_NAME` (and will do the same for V1 after MF-07). Please map the logical parent by id and derive the extracted column/path from the suffix or segment metadata, retaining NULL only for truly dropped indexes. ########## be/src/storage/index/index_disk_usage.cpp: ########## @@ -0,0 +1,413 @@ +// 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" + +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>(); +} + +// Returns the V1 index file size persisted in the rowset meta, or -1 when it is not recorded. +int64_t persisted_v1_file_size(const InvertedIndexFileInfo& file_info, const TabletIndex& index) { + for (const auto& index_info : file_info.index_info()) { + if (index_info.index_id() == index.index_id() && + index_info.index_suffix() == index.get_index_suffix()) { + return index_info.index_file_size() > 0 ? index_info.index_file_size() : -1; + } + } + return -1; +} + +// 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()); Review Comment: [P2] Validate the compound entry extent before trusting this length. V2/V3 `_init_from` accepts negative/overflowing/out-of-file offsets and lengths, and the V1 reader likewise does not bound positive-offset entries against the physical file (its inline-entry encoding is separate). This accounting path only lists entries, so an invalid extent is never opened and unchecked subtraction can return inflated per-index bytes or a negative container/header total successfully. Reject corrupt V1/V2/V3 headers and add checked accumulation/subtraction tests. -- 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]
