You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2026/10/21 01:53:36 UTC
(doris) branch master updated: [feature](inverted index) Add index_disk_usage table function (#67977)
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 3c6af92291b [feature](inverted index) Add index_disk_usage table function (#67977)
3c6af92291b is described below
commit 3c6af92291b14979298a25b02b1abc264287f6f4
Author: Jack <ji...@selectdb.com>
AuthorDate: Wed Oct 21 09:53:27 2026 +0800
[feature](inverted index) Add index_disk_usage table function (#67977)
### What problem does this PR solve?
Issue Number: N/A
Related PR: N/A
Problem Summary:
Doris reports the total inverted index size of a rowset or table (for
example `information_schema.rowsets.INDEX_DISK_SIZE` and `SHOW DATA`),
but SQL users cannot see how much space each index takes, or how that
space splits between the term dictionary, postings, positions and
statistics. Deciding whether to drop `support_phrase`, change a parser
or switch the index storage format therefore requires inspecting index
files by hand on each backend.
This PR adds the table function `index_disk_usage`:
```sql
SELECT index_name, column_name, structure, storage_format,
SUM(total_bytes), SUM(dict_bytes), SUM(posting_bytes), SUM(position_bytes), SUM(stats_bytes)
FROM index_disk_usage("database" = "db1", "table" = "t1")
GROUP BY 1, 2, 3, 4;
```
- Parameters: `database` and `table` (required), `partitions` and
`indexes` (comma-separated names), `level` (`tablet` by default,
`rowset` or `segment`), and `position_detail` (default `false`).
- Columns: `PARTITION_NAME`, `MATERIALIZED_INDEX_NAME`, `TABLET_ID`,
`BACKEND_ID`, `ROWSET_ID`, `SEGMENT_ID`, `INDEX_ID`, `INDEX_NAME`,
`INDEX_TYPE`, `COLUMN_NAME`, `INDEX_SUFFIX`, `STRUCTURE`
(`TERM`/`BKD`/`ANN`/`CONTAINER`), `STORAGE_FORMAT`
(`V1`/`V2`/`V3`/`SNII`), `SEGMENT_COUNT`, `ROW_COUNT`, `TOTAL_BYTES`,
`DICT_BYTES`, `POSTING_BYTES`, `POSITION_BYTES`, `STATS_BYTES`,
`OTHER_BYTES`, `STATS_SOURCE`.
How it works:
- FE resolves the table, its formal partitions and the tablets of the
base and rollup indexes, and checks the `SHOW` privilege. In local mode
the partition visible versions are read together with the tablets under
the table lock; in cloud mode they are fetched from meta-service after
the lock is released, with the batched snapshot version API. Each tablet
is sent to one queryable replica through a new `INDEX_DISK_USAGE`
metadata scan range; a cloud replica routed to a smooth upgrade source
backend fails the query, because that backend cannot run this scan. A
`partitions` or `indexes` filter that names nothing is rejected.
- BE captures the rowsets visible at that version (syncing rowsets first
in cloud mode) and reads only index file metadata of each segment:
- V1/V2/V3: CLucene compound directory entries are attributed by
extension (`tis`/`tii` to dictionary, `frq` to postings, `prx` to
positions, `nrm` to statistics, everything else to other). BKD files
form a `BKD` row with only `TOTAL_BYTES`. The shared V2/V3 compound
header is reported as a `CONTAINER` row.
- SNII: the logical index directory and core metadata are read. The
dictionary region, sampled term index, dictionary block directory and
BSBF count as dictionary; the posting region as postings; core metadata
and norms as statistics; the null bitmap as other. SNII interleaves
positions with postings, so `POSITION_BYTES` is NULL unless
`position_detail=true` decodes the posting region to split them out.
Docs-only indexes report 0 positions.
- Rows are aggregated to the requested level. Within a segment,
`TOTAL_BYTES` of all rows adds up to the index file sizes; unknown
components are NULL.
- Every index file read carries the IO context of the query (query id,
reader type and the remote scan cache write limiter), so file cache
accounting and `file_cache_query_limit_bytes` apply to this scan like to
any other scan of the query.
- A request covering more tablets than the session variable
`index_disk_usage_max_tablets` (default 20000) is rejected, and
`position_detail=true` is additionally limited by
`index_disk_usage_position_detail_max_tablets` (default 10000).
Collecting these statistics at write time and an FE-aggregated
`information_schema` table are left to follow-up PRs.
### Release note
Add the table function `index_disk_usage` to show the on-disk size of
each inverted or ANN index, broken down into dictionary, posting,
position and statistics bytes, at tablet, rowset or segment level.
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [x] Regression test
- `external_table_p0/tvf/test_index_disk_usage_tvf`: V2, V3 and SNII
tables with two rowsets each; component presence per structure,
per-rowset totals reconciled with
`information_schema.rowsets.INDEX_DISK_SIZE`, all three levels, SNII
`position_detail`, index filtering and parameter errors. Passed on a
local single FE/BE cluster.
- [x] Unit Test
- BE: `IndexDiskUsage*` (V1/V2/V3/SNII collectors, docs-only and
positional indexes, `position_detail`, BKD, index filtering, missing
files, level aggregation) and `SniiSegmentReaderLogicalIndexesTest`: 22
passed.
- FE: `IndexDiskUsageTableValuedFunctionTest` (parameters, privileges,
partition/index resolution, visible version read outside the table lock,
`position_detail` tablet limit) and `IndexDiskUsageScanNodeTest`
(replica selection and grouping by backend): 20 passed.
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [x] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [x] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
- The new table function needs a page in doris-website; not submitted
yet.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
be/src/exec/scan/meta_scanner.cpp | 6 +
be/src/format/table/index_disk_usage_reader.cpp | 436 +++++++
be/src/format/table/index_disk_usage_reader.h | 121 ++
be/src/storage/CMakeLists.txt | 3 +
be/src/storage/index/index_disk_usage.cpp | 581 +++++++++
be/src/storage/index/index_disk_usage.h | 147 +++
be/src/storage/index/index_file_reader.cpp | 52 +-
be/src/storage/index/index_file_reader.h | 10 +
.../inverted/snii/reader/snii_segment_reader.cpp | 13 +-
.../inverted/snii/reader/snii_segment_reader.h | 9 +
be/src/storage/segment/page_decoder.h | 1 +
.../format/table/index_disk_usage_reader_test.cpp | 235 ++++
.../index/index_disk_usage_aggregate_test.cpp | 135 ++
.../storage/index/index_disk_usage_rowset_test.cpp | 249 ++++
be/test/storage/index/index_disk_usage_test.cpp | 1301 ++++++++++++++++++++
.../snii_segment_reader_logical_indexes_test.cpp | 140 +++
.../doris/catalog/BuiltinTableValuedFunctions.java | 2 +
.../tvf/source/IndexDiskUsageScanNode.java | 257 ++++
.../functions/table/IndexDiskUsage.java | 58 +
.../visitor/TableValuedFunctionVisitor.java | 5 +
.../org/apache/doris/planner/OlapScanNode.java | 5 +-
.../java/org/apache/doris/qe/SessionVariable.java | 12 +
.../IndexDiskUsageTableValuedFunction.java | 478 +++++++
.../doris/tablefunction/TableValuedFunctionIf.java | 2 +
.../tvf/source/IndexDiskUsageScanNodeTest.java | 341 +++++
.../org/apache/doris/qe/SessionVariablesTest.java | 17 +
.../IndexDiskUsageTableValuedFunctionTest.java | 489 ++++++++
gensrc/thrift/PlanNodes.thrift | 24 +
gensrc/thrift/Types.thrift | 1 +
.../tvf/test_index_disk_usage_tvf.out | 71 ++
.../tvf/test_index_disk_usage_tvf.groovy | 187 +++
31 files changed, 5374 insertions(+), 14 deletions(-)
diff --git a/be/src/exec/scan/meta_scanner.cpp b/be/src/exec/scan/meta_scanner.cpp
index 5bb3d0c6672..282713257d7 100644
--- a/be/src/exec/scan/meta_scanner.cpp
+++ b/be/src/exec/scan/meta_scanner.cpp
@@ -37,6 +37,7 @@
#include "core/column/column_vector.h"
#include "core/data_type/define_primitive_type.h"
#include "core/types.h"
+#include "format/table/index_disk_usage_reader.h"
#include "format/table/parquet_metadata_reader.h"
#include "runtime/cluster_info.h"
#include "runtime/descriptors.h"
@@ -69,6 +70,11 @@ Status MetaScanner::_open_impl(RuntimeState* state) {
_scan_range.meta_scan_range);
RETURN_IF_ERROR(reader->init_reader());
_reader = std::move(reader);
+ } else if (_scan_range.meta_scan_range.metadata_type == TMetadataType::INDEX_DISK_USAGE) {
+ auto reader = IndexDiskUsageReader::create_unique(_tuple_desc->slots(), state, _profile,
+ _scan_range.meta_scan_range);
+ RETURN_IF_ERROR(reader->init_reader());
+ _reader = std::move(reader);
} else {
RETURN_IF_ERROR(_fetch_metadata(_scan_range.meta_scan_range));
}
diff --git a/be/src/format/table/index_disk_usage_reader.cpp b/be/src/format/table/index_disk_usage_reader.cpp
new file mode 100644
index 00000000000..6febd0e9ad2
--- /dev/null
+++ b/be/src/format/table/index_disk_usage_reader.cpp
@@ -0,0 +1,436 @@
+// 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/query_context.h"
+#include "runtime/runtime_state.h"
+#include "storage/rowset/rowset.h"
+#include "storage/rowset/rowset_meta.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();
+ };
+ // Index file reads are accounted to the query like the reads of any other scan.
+ _io_ctx = io::IOContext {
+ .reader_type = ReaderType::READER_QUERY,
+ .query_id = &_state->query_id(),
+ .is_inverted_index = true,
+ };
+ _io_ctx.inverted_index_snii_read_no_write_file_cache =
+ _state->query_options().inverted_index_snii_read_no_write_file_cache;
+ if (auto* query_ctx = _state->get_query_ctx(); query_ctx != nullptr) {
+ _io_ctx.remote_scan_cache_write_limiter = query_ctx->remote_scan_cache_write_limiter();
+ }
+ _options.io_ctx = &_io_ctx;
+ _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 {
+ const TabletRowsets captured =
+ DORIS_TRY(capture_tablet_rowsets(target.tablet_id, target.version));
+ *current_schema = label_schema(captured.tablet->tablet_schema(), captured.rowsets);
+
+ const io::IOContext io_ctx = tablet_io_context(_io_ctx, captured.tablet->ttl_seconds());
+ segment_v2::DirectoryFileNames directory_files;
+ segment_v2::IndexDiskUsageOptions options = _options;
+ options.io_ctx = &io_ctx;
+ options.directory_files = &directory_files;
+ for (const RowsetSharedPtr& rowset : captured.rowsets) {
+ RETURN_IF_ERROR(segment_v2::collect_rowset_index_disk_usage(rowset, options,
+ target.tablet_id, rows));
+ }
+ return Status::OK();
+}
+
+Result<IndexDiskUsageReader::TabletRowsets> IndexDiskUsageReader::capture_tablet_rowsets(
+ int64_t tablet_id, int64_t version) {
+ SyncRowsetStats sync_stats;
+ auto looked_up = ExecEnv::get_tablet(tablet_id, &sync_stats);
+ if (!looked_up.has_value()) {
+ return ResultError(std::move(looked_up.error()));
+ }
+ TabletRowsets result;
+ result.tablet = std::move(looked_up.value());
+ if (auto cloud_tablet = std::dynamic_pointer_cast<CloudTablet>(result.tablet);
+ cloud_tablet != nullptr) {
+ SyncOptions options;
+ // A tablet that the lookup loaded is current, except that the load may have begun before
+ // the scan version was committed. Syncing to that version sends no RPC when the load
+ // already saw it. A cached tablet may have missed a compaction done elsewhere at its
+ // cached version, which only a sync without a query version picks up.
+ if (sync_stats.tablet_meta_cache_miss != 0) {
+ options.query_version = version;
+ }
+ RETURN_IF_ERROR_RESULT(cloud_tablet->sync_rowsets(options));
+ }
+ {
+ std::shared_lock rdlock(result.tablet->get_header_lock());
+ auto captured = result.tablet->capture_consistent_rowsets_unlocked(Version(0, version),
+ CaptureRowsetOps {});
+ if (!captured.has_value()) {
+ return ResultError(std::move(captured.error()));
+ }
+ result.rowsets = std::move(captured.value().rowsets);
+ }
+ return result;
+}
+
+TabletSchemaSPtr IndexDiskUsageReader::label_schema(const TabletSchemaSPtr& tablet_schema,
+ const std::vector<RowsetSharedPtr>& rowsets) {
+ std::vector<RowsetMetaSharedPtr> rowset_metas;
+ rowset_metas.reserve(rowsets.size());
+ for (const RowsetSharedPtr& rowset : rowsets) {
+ if (rowset->rowset_meta()->tablet_schema() != nullptr) {
+ rowset_metas.push_back(rowset->rowset_meta());
+ }
+ }
+ if (rowset_metas.empty()) {
+ return tablet_schema;
+ }
+ return BaseTablet::tablet_schema_with_merged_max_schema_version(rowset_metas);
+}
+
+io::IOContext IndexDiskUsageReader::tablet_io_context(const io::IOContext& query_io_ctx,
+ int64_t ttl_seconds) {
+ io::IOContext io_ctx = query_io_ctx;
+ io_ctx.expiration_time = ttl_seconds;
+ return io_ctx;
+}
+
+Status IndexDiskUsageReader::_fill_block(Block* block, const TIndexDiskUsageTablet& target,
+ const TabletSchema& schema,
+ const std::vector<IndexDiskUsageRow>& rows) const {
+ if (!block->mem_reuse()) {
+ std::vector<MutableColumnPtr> columns(_slots.size());
+ for (size_t i = 0; i < _slots.size(); ++i) {
+ columns[i] = _slots[i]->get_empty_mutable_column();
+ }
+ _append_rows(columns, target, schema, rows);
+ for (size_t i = 0; i < _slots.size(); ++i) {
+ block->insert(ColumnWithTypeAndName(std::move(columns[i]), _slots[i]->type(),
+ _slots[i]->col_name()));
+ }
+ } else {
+ auto columns_guard = block->mutate_columns_scoped();
+ auto& columns = columns_guard.mutable_columns();
+ for (size_t i = 0; i < _slots.size(); ++i) {
+ columns[i]->clear();
+ }
+ _append_rows(columns, target, schema, rows);
+ }
+ return Status::OK();
+}
+
+namespace {
+
+// One output row with the names resolved from the tablet schema.
+struct RowView {
+ const TIndexDiskUsageTablet* target = nullptr;
+ // nullptr when FE sent no name for the partition.
+ const std::string* partition_name = nullptr;
+ // nullptr when FE sent no name for the base or rollup index of the tablet.
+ const std::string* materialized_index_name = nullptr;
+ const IndexDiskUsageRow* row = nullptr;
+ // nullptr for container rows and for indexes no longer in the schema.
+ const TabletIndex* index = nullptr;
+ std::string column_name;
+};
+
+// A cell value: NULL, an integer or a string that outlives the insertion.
+using Cell = std::variant<std::monostate, int64_t, std::string_view>;
+
+// Components are reported only for term indexes, except that a container row carries its
+// overhead as other bytes. Unknown components (-1) are NULL.
+Cell component_cell(const IndexDiskUsageRecord& record, int64_t value, bool is_other) {
+ const bool reported = record.structure == IndexDiskUsageStructure::kTerm ||
+ (is_other && record.structure == IndexDiskUsageStructure::kContainer);
+ return reported && value >= 0 ? Cell {value} : Cell {};
+}
+
+Cell cell_of(IndexDiskUsageReader::Column column, const RowView& view, IndexDiskUsageLevel level,
+ int64_t backend_id) {
+ using C = IndexDiskUsageReader::Column;
+ const IndexDiskUsageRow& row = *view.row;
+ const IndexDiskUsageRecord& record = row.record;
+ const bool is_container = record.structure == IndexDiskUsageStructure::kContainer;
+ switch (column) {
+ case C::kPartitionName:
+ return view.partition_name == nullptr ? Cell {}
+ : Cell {std::string_view(*view.partition_name)};
+ case C::kMaterializedIndexName:
+ return view.materialized_index_name == nullptr
+ ? Cell {}
+ : Cell {std::string_view(*view.materialized_index_name)};
+ case C::kTabletId:
+ return Cell {view.target->tablet_id};
+ case C::kBackendId:
+ return Cell {backend_id};
+ case C::kRowsetId:
+ return level == IndexDiskUsageLevel::kTablet ? Cell {}
+ : Cell {std::string_view(row.rowset_id)};
+ case C::kSegmentId:
+ return level == IndexDiskUsageLevel::kSegment ? Cell {int64_t {row.segment_id}} : Cell {};
+ case C::kIndexId:
+ return is_container ? Cell {} : Cell {record.index_id};
+ case C::kIndexName:
+ return view.index == nullptr ? Cell {} : Cell {std::string_view(view.index->index_name())};
+ case C::kIndexType:
+ if (view.index == nullptr) {
+ return Cell {};
+ }
+ return Cell {std::string_view(view.index->is_ann_index() ? "ANN" : "INVERTED")};
+ case C::kColumnName:
+ return view.column_name.empty() ? Cell {} : Cell {std::string_view(view.column_name)};
+ case C::kIndexSuffix:
+ return is_container ? Cell {} : Cell {std::string_view(record.index_suffix)};
+ case C::kStructure:
+ return Cell {structure_name(record.structure)};
+ case C::kStorageFormat:
+ return Cell {std::string_view(InvertedIndexStorageFormatPB_Name(row.format))};
+ case C::kSegmentCount:
+ return Cell {row.segment_count};
+ case C::kRowCount:
+ return Cell {row.row_count};
+ case C::kTotalBytes:
+ return Cell {record.total_bytes};
+ case C::kDictBytes:
+ return component_cell(record, record.dict_bytes, false);
+ case C::kPostingBytes:
+ return component_cell(record, record.posting_bytes, false);
+ case C::kPositionBytes:
+ return component_cell(record, record.position_bytes, false);
+ case C::kStatsBytes:
+ return component_cell(record, record.stats_bytes, false);
+ case C::kOtherBytes:
+ return component_cell(record, record.other_bytes, true);
+ case C::kStatsSource:
+ return Cell {std::string_view("FILE")};
+ }
+ return Cell {};
+}
+
+void insert_cell(IColumn* column, IndexDiskUsageReader::Column kind, const Cell& cell) {
+ if (std::holds_alternative<std::monostate>(cell)) {
+ insert_null(column);
+ } else if (const auto* value = std::get_if<int64_t>(&cell)) {
+ if (kind == IndexDiskUsageReader::Column::kSegmentId) {
+ insert_int32(column, cast_set<int32_t>(*value));
+ } else {
+ insert_int64(column, *value);
+ }
+ } else {
+ insert_string(column, std::get<std::string_view>(cell));
+ }
+}
+
+} // namespace
+
+void IndexDiskUsageReader::_append_rows(std::vector<MutableColumnPtr>& columns,
+ const TIndexDiskUsageTablet& target,
+ const TabletSchema& schema,
+ const std::vector<IndexDiskUsageRow>& rows) const {
+ const auto& params = _scan_range.index_disk_usage_params;
+ const auto partition = params.partition_names.find(target.partition_id);
+ const auto materialized_index =
+ params.materialized_index_names.find(target.materialized_index_id);
+ for (const IndexDiskUsageRow& row : rows) {
+ RowView view;
+ view.target = ⌖
+ view.partition_name =
+ partition == params.partition_names.end() ? nullptr : &partition->second;
+ view.materialized_index_name = materialized_index == params.materialized_index_names.end()
+ ? nullptr
+ : &materialized_index->second;
+ view.row = &row;
+ if (row.record.structure != IndexDiskUsageStructure::kContainer) {
+ view.index = segment_v2::resolve_disk_usage_index(schema, row.record.index_id,
+ row.record.index_suffix);
+ }
+ if (view.index != nullptr && !view.index->col_unique_ids().empty()) {
+ const int32_t ordinal = schema.field_index(view.index->col_unique_ids()[0]);
+ if (ordinal >= 0) {
+ view.column_name = schema.column(ordinal).name();
+ }
+ }
+ for (size_t i = 0; i < _slot_columns.size(); ++i) {
+ insert_cell(columns[i].get(), _slot_columns[i],
+ cell_of(_slot_columns[i], view, _level, _state->backend_id()));
+ }
+ }
+}
+
+} // namespace doris
diff --git a/be/src/format/table/index_disk_usage_reader.h b/be/src/format/table/index_disk_usage_reader.h
new file mode 100644
index 00000000000..e0be374e923
--- /dev/null
+++ b/be/src/format/table/index_disk_usage_reader.h
@@ -0,0 +1,121 @@
+// 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.
+
+#pragma once
+
+#include <gen_cpp/PlanNodes_types.h>
+
+#include <cstdint>
+#include <vector>
+
+#include "common/factory_creator.h"
+#include "common/status.h"
+#include "format/generic_reader.h"
+#include "io/io_common.h"
+#include "runtime/descriptors.h"
+#include "storage/index/index_disk_usage.h"
+#include "storage/tablet/tablet_fwd.h"
+
+namespace doris {
+class Block;
+class RuntimeProfile;
+class RuntimeState;
+
+// Produces the rows of the index_disk_usage table function for the tablets of one scan range.
+// Each block holds the rows of one tablet, so memory stays bounded by the largest tablet.
+class IndexDiskUsageReader : public GenericReader {
+ ENABLE_FACTORY_CREATOR(IndexDiskUsageReader);
+
+public:
+ // Output columns, in the order the FE schema declares them.
+ enum class Column : uint8_t {
+ kPartitionName,
+ kMaterializedIndexName,
+ kTabletId,
+ kBackendId,
+ kRowsetId,
+ kSegmentId,
+ kIndexId,
+ kIndexName,
+ kIndexType,
+ kColumnName,
+ kIndexSuffix,
+ kStructure,
+ kStorageFormat,
+ kSegmentCount,
+ kRowCount,
+ kTotalBytes,
+ kDictBytes,
+ kPostingBytes,
+ kPositionBytes,
+ kStatsBytes,
+ kOtherBytes,
+ kStatsSource,
+ };
+
+ IndexDiskUsageReader(std::vector<SlotDescriptor*> slots, RuntimeState* state,
+ RuntimeProfile* profile, TMetaScanRange scan_range);
+ ~IndexDiskUsageReader() override = default;
+
+ Status init_reader();
+ Status _do_get_next_block(Block* block, size_t* read_rows, bool* eof) override;
+ Status close() override { return Status::OK(); }
+ const segment_v2::IndexDiskUsageOptions& options() const { return _options; }
+
+ // The schema that labels the rows of a tablet: the newest schema among its rowsets, because a
+ // cached cloud tablet may not know an index that later rowsets already carry.
+ static TabletSchemaSPtr label_schema(const TabletSchemaSPtr& tablet_schema,
+ const std::vector<RowsetSharedPtr>& rowsets);
+ // The IO context of one tablet: the query context plus the tablet TTL that classifies its
+ // reads in the file cache.
+ static io::IOContext tablet_io_context(const io::IOContext& query_io_ctx, int64_t ttl_seconds);
+ // A tablet and its rowsets through the scan version.
+ struct TabletRowsets {
+ BaseTabletSPtr tablet;
+ std::vector<RowsetSharedPtr> rowsets;
+ };
+ // Looks up a tablet and captures its rowsets through `version`. A cloud tablet is brought up
+ // to date with meta-service, so rowsets that a compaction on another backend replaced are not
+ // reported, and a tablet that was just loaded is only synchronized if it lacks `version`.
+ static Result<TabletRowsets> capture_tablet_rowsets(int64_t tablet_id, int64_t version);
+
+protected:
+ Status _do_init_reader(ReaderInitContext* /*ctx*/) override { return init_reader(); }
+
+private:
+ Status _collect_tablet(const TIndexDiskUsageTablet& target,
+ std::vector<segment_v2::IndexDiskUsageRow>* rows,
+ TabletSchemaSPtr* current_schema) const;
+ Status _fill_block(Block* block, const TIndexDiskUsageTablet& target,
+ const TabletSchema& schema,
+ const std::vector<segment_v2::IndexDiskUsageRow>& rows) const;
+ void _append_rows(std::vector<MutableColumnPtr>& columns, const TIndexDiskUsageTablet& target,
+ const TabletSchema& schema,
+ const std::vector<segment_v2::IndexDiskUsageRow>& rows) const;
+
+ RuntimeState* _state = nullptr;
+ std::vector<SlotDescriptor*> _slots;
+ TMetaScanRange _scan_range;
+ std::vector<Column> _slot_columns;
+ segment_v2::IndexDiskUsageLevel _level = segment_v2::IndexDiskUsageLevel::kTablet;
+ // Referenced by _options, so the reader must not move after init_reader().
+ io::IOContext _io_ctx;
+ segment_v2::IndexDiskUsageOptions _options;
+ size_t _next_tablet = 0;
+};
+
+} // namespace doris
diff --git a/be/src/storage/CMakeLists.txt b/be/src/storage/CMakeLists.txt
index a2a0ccf0f9b..54fd6bd2c97 100644
--- a/be/src/storage/CMakeLists.txt
+++ b/be/src/storage/CMakeLists.txt
@@ -56,6 +56,9 @@ pch_reuse(Storage)
# - the three heaviest template-instantiation TUs (predicate creators) which
# would dominate any batch they join
set(STORAGE_UNITY_SKIP
+ # Keep storage batches stable: null_predicate's using directive makes
+ # later CLucene sse_util references ambiguous.
+ ${CMAKE_CURRENT_SOURCE_DIR}/index/index_disk_usage.cpp
# Keep the spill helpers independent: the PRX encoder requests ZSTD's static
# workspace API, and inserting these files into the existing unity batches
# exposes an unrelated LIMIT_REACH enum collision in olap_scan_common.cpp.
diff --git a/be/src/storage/index/index_disk_usage.cpp b/be/src/storage/index/index_disk_usage.cpp
new file mode 100644
index 00000000000..2927410e84e
--- /dev/null
+++ b/be/src/storage/index/index_disk_usage.cpp
@@ -0,0 +1,581 @@
+// 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 <charconv>
+#include <deque>
+#include <filesystem>
+#include <limits>
+#include <map>
+#include <memory>
+#include <optional>
+#include <set>
+#include <system_error>
+#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/common/inverted_index_desc.h"
+#include "storage/index/inverted/snii/format/dict_entry.h"
+#include "storage/index/inverted/snii/format/format_constants.h"
+#include "storage/index/inverted/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>();
+}
+
+// Whether `status` says that the segment has no index container, which is only expected when the
+// rowset meta records no container size: the size is recorded after the container is written.
+bool is_skipped_container(const Status& status, const InvertedIndexFileInfo& file_info) {
+ const bool recorded = file_info.has_index_size() && file_info.index_size() > 0;
+ return !recorded && is_absent_index_file(status);
+}
+
+// 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 from the rowset meta, which records every written file,
+// including the extracted VARIANT paths that the schema does not list. Rowsets without that record
+// fall back to the inverted and ANN indexes of their own schema.
+// `owned` keeps the indexes built from the rowset meta.
+std::vector<V1IndexFile> list_v1_index_files(const TabletSchema& schema,
+ const InvertedIndexFileInfo& file_info,
+ std::deque<TabletIndex>* owned) {
+ std::vector<V1IndexFile> files;
+ if (file_info.index_info_size() == 0) {
+ for (const TabletIndex* index : schema.inverted_and_ann_indexes()) {
+ files.push_back({.index = index, .persisted_size = -1});
+ }
+ return files;
+ }
+ 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;
+}
+
+// Parses the index id and suffix out of `{segment}_{index_id}[@{suffix}].idx`, the name of a V1
+// index file of the segment named `segment`.
+std::optional<std::pair<int64_t, std::string>> parse_v1_index_file_name(std::string_view name,
+ std::string_view segment) {
+ constexpr std::string_view extension = InvertedIndexDescriptor::index_suffix;
+ if (name.size() < segment.size() + extension.size() + 2 || !name.starts_with(segment) ||
+ !name.ends_with(extension) || name[segment.size()] != '_') {
+ return std::nullopt;
+ }
+ std::string_view rest = name.substr(segment.size() + 1);
+ rest.remove_suffix(extension.size());
+ uint64_t index_id = 0;
+ const auto [digits_end, error] =
+ std::from_chars(rest.data(), rest.data() + rest.size(), index_id);
+ if (error != std::errc() || index_id > std::numeric_limits<int64_t>::max()) {
+ return std::nullopt;
+ }
+ std::string_view tail = rest.substr(digits_end - rest.data());
+ if (tail.empty()) {
+ return std::make_pair(static_cast<int64_t>(index_id), std::string());
+ }
+ if (tail.front() != '@') {
+ return std::nullopt;
+ }
+ return std::make_pair(static_cast<int64_t>(index_id), std::string(tail.substr(1)));
+}
+
+// A legacy local VARIANT rowset has no index file info, and its schema may not name every file
+// that was written for an extracted path. Adds the V1 index files of the segment that exist in its
+// directory and are not in `files` yet, in index id and suffix order.
+Status add_v1_index_files_on_disk(const std::string& index_path_prefix,
+ DirectoryFileNames* directory_files,
+ std::deque<TabletIndex>* owned, std::vector<V1IndexFile>* files) {
+ const std::filesystem::path prefix(index_path_prefix);
+ const std::vector<std::string>* names = nullptr;
+ RETURN_IF_ERROR(directory_files->list(prefix.parent_path().string(), &names));
+ std::set<std::pair<int64_t, std::string>> listed;
+ for (const V1IndexFile& file : *files) {
+ listed.emplace(file.index->index_id(), file.index->get_index_suffix());
+ }
+ // The names are sorted, so the files of this segment are adjacent.
+ const std::string segment_name = prefix.filename().string();
+ const std::string name_prefix = segment_name + "_";
+ std::set<std::pair<int64_t, std::string>> on_disk;
+ for (auto it = std::lower_bound(names->begin(), names->end(), name_prefix);
+ it != names->end() && it->starts_with(name_prefix); ++it) {
+ if (auto parsed = parse_v1_index_file_name(*it, segment_name);
+ parsed.has_value() && !listed.contains(*parsed)) {
+ on_disk.insert(std::move(*parsed));
+ }
+ }
+ for (const auto& [index_id, suffix] : on_disk) {
+ TabletIndexPB index_pb;
+ index_pb.set_index_type(IndexType::INVERTED);
+ index_pb.set_index_id(index_id);
+ index_pb.set_index_suffix_name(suffix);
+ owned->emplace_back().init_from_pb(index_pb);
+ files->push_back({.index = &owned->back(), .persisted_size = -1});
+ }
+ return Status::OK();
+}
+
+// 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 = options.io_ctx != nullptr ? *options.io_ctx : io::IOContext {};
+ 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
+
+Status DirectoryFileNames::list(const std::string& dir, const std::vector<std::string>** names) {
+ if (!_listed || _dir != dir) {
+ // The file sizes are not read: unused rowsets delete their files from the tablet directory
+ // while it is listed, and the stat of a vanished file would fail the listing. An entry whose
+ // type cannot be read is gone and is skipped.
+ std::error_code ec;
+ std::filesystem::directory_iterator it(dir, ec);
+ if (ec && ec != std::errc::no_such_file_or_directory) {
+ return Status::IOError("failed to list {}: {}", dir, ec.message());
+ }
+ _names.clear();
+ for (const std::filesystem::directory_iterator end; !ec && it != end; it.increment(ec)) {
+ std::error_code type_ec;
+ if (it->is_regular_file(type_ec)) {
+ _names.push_back(it->path().filename().string());
+ }
+ }
+ if (ec && ec != std::errc::no_such_file_or_directory) {
+ return Status::IOError("failed to list {}: {}", dir, ec.message());
+ }
+ std::sort(_names.begin(), _names.end());
+ _dir = dir;
+ _listed = true;
+ ++_listings;
+ }
+ *names = &_names;
+ return Status::OK();
+}
+
+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(config::inverted_index_read_buffer_size, options.io_ctx));
+ std::deque<TabletIndex> file_indexes;
+ std::vector<V1IndexFile> v1_files =
+ list_v1_index_files(*_schema, _index_file_info, &file_indexes);
+ if (_index_file_info.index_info_size() == 0 && _schema->num_variant_columns() > 0 &&
+ _fs->type() == io::FileSystemType::LOCAL) {
+ DirectoryFileNames own_listing;
+ RETURN_IF_ERROR(add_v1_index_files_on_disk(
+ _index_path_prefix,
+ options.directory_files != nullptr ? options.directory_files : &own_listing,
+ &file_indexes, &v1_files));
+ }
+ for (const V1IndexFile& file : v1_files) {
+ const TabletIndex& index = *file.index;
+ if (!is_wanted(options, index.index_id())) {
+ continue;
+ }
+ RETURN_IF_ERROR(check_cancelled(options));
+ // The rowset meta records only files that were written, so only a file listed from the
+ // schema may be absent.
+ const bool recorded = file.persisted_size > 0;
+ int64_t file_size = file.persisted_size;
+ if (!recorded) {
+ 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, options.io_ctx);
+ if (!directory.has_value()) {
+ if (!recorded && 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(config::inverted_index_read_buffer_size, options.io_ctx);
+ !st.ok()) {
+ return is_skipped_container(st, _index_file_info) ? 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(config::inverted_index_read_buffer_size, options.io_ctx);
+ !st.ok()) {
+ return is_skipped_container(st, _index_file_info) ? 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,
+ options.io_ctx));
+ 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();
+}
+
+const TabletIndex* resolve_disk_usage_index(const TabletSchema& schema, int64_t index_id,
+ std::string_view suffix) {
+ // An extracted VARIANT path keeps its parent index id under a path suffix that the current
+ // schema may not list, so an unmatched suffix resolves to the index of the same id.
+ const TabletIndex* same_id = nullptr;
+ for (const TabletIndex* index : schema.inverted_and_ann_indexes()) {
+ if (index->index_id() != index_id) {
+ continue;
+ }
+ if (index->get_index_suffix() == suffix) {
+ return index;
+ }
+ if (same_id == nullptr || index->get_index_suffix().empty()) {
+ same_id = index;
+ }
+ }
+ return same_id;
+}
+
+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();
+ 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 {
+ // Rowsets without persisted row counts read them from the segment footer. Loading the
+ // segment directly keeps a transient failure out of the rowset's run-once row cache.
+ SegmentSharedPtr loaded;
+ OlapReaderStatistics stats;
+ RETURN_IF_ERROR(std::static_pointer_cast<BetaRowset>(rowset)->load_segment(
+ segment.ref(), &stats, &loaded, options.io_ctx));
+ row_count = loaded->num_rows();
+ }
+ for (auto& record : records) {
+ IndexDiskUsageRow row;
+ row.rowset_id = rowset_id;
+ row.segment_id = cast_set<int32_t>(segment.id());
+ row.segment_count = 1;
+ row.row_count = row_count;
+ row.format = format;
+ row.record = std::move(record);
+ rows->push_back(std::move(row));
+ }
+ }
+ return Status::OK();
+}
+
+} // namespace doris::segment_v2
diff --git a/be/src/storage/index/index_disk_usage.h b/be/src/storage/index/index_disk_usage.h
new file mode 100644
index 00000000000..fa1a43323f8
--- /dev/null
+++ b/be/src/storage/index/index_disk_usage.h
@@ -0,0 +1,147 @@
+// 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.
+
+#pragma once
+
+#include <gen_cpp/olap_common.pb.h>
+#include <gen_cpp/olap_file.pb.h>
+
+#include <cstdint>
+#include <functional>
+#include <set>
+#include <string>
+#include <string_view>
+#include <vector>
+
+#include "common/status.h"
+#include "io/fs/file_system.h"
+#include "io/io_common.h"
+#include "storage/rowset/rowset_fwd.h"
+#include "storage/tablet/tablet_schema.h"
+
+namespace doris::segment_v2 {
+
+enum class IndexDiskUsageStructure : uint8_t { kTerm, kBkd, kAnn, kContainer };
+
+// Physical bytes of one index, or of a container's shared overhead, in one segment.
+// A component of -1 means it is unknown.
+struct IndexDiskUsageRecord {
+ int64_t index_id = -1;
+ std::string index_suffix;
+ IndexDiskUsageStructure structure = IndexDiskUsageStructure::kTerm;
+ int64_t total_bytes = 0;
+ int64_t dict_bytes = 0;
+ int64_t posting_bytes = 0;
+ int64_t position_bytes = 0;
+ int64_t stats_bytes = 0;
+ int64_t other_bytes = 0;
+};
+
+// The sorted names of the files in one local directory, listed on first use and shared by the
+// segments that live there.
+class DirectoryFileNames {
+public:
+ // Sets `names` to the sorted file names in `dir`, which are none when `dir` does not exist.
+ Status list(const std::string& dir, const std::vector<std::string>** names);
+ // How many times a directory was listed.
+ int listings() const { return _listings; }
+
+private:
+ std::string _dir;
+ std::vector<std::string> _names;
+ bool _listed = false;
+ int _listings = 0;
+};
+
+struct IndexDiskUsageOptions {
+ // Scan SNII dictionary blocks to split positions out of postings.
+ bool position_detail = false;
+ // Empty means all indexes.
+ std::set<int64_t> index_ids;
+ // Returns an error once the query is cancelled. Called before each segment and each SNII
+ // dictionary block; unset means the work cannot be cancelled.
+ std::function<Status()> check_cancelled;
+ // The query context passed to every index file read, so the file cache accounts these reads
+ // to the query. Unset means the default read behavior.
+ const io::IOContext* io_ctx = nullptr;
+ // The listing of the directory that holds the segments of one tablet. Unset means each
+ // segment that needs a listing lists the directory itself.
+ DirectoryFileNames* directory_files = nullptr;
+};
+
+enum class IndexDiskUsageLevel : uint8_t { kTablet, kRowset, kSegment };
+
+// A record tagged with the rowset and segment it was collected from.
+struct IndexDiskUsageRow {
+ std::string rowset_id;
+ int32_t segment_id = -1;
+ int64_t segment_count = 0;
+ int64_t row_count = 0;
+ InvertedIndexStorageFormatPB format = InvertedIndexStorageFormatPB::V2;
+ IndexDiskUsageRecord record;
+};
+
+// Merges rows of the same index, format and structure within the granularity of `level`.
+// Merged rows keep first-appearance order; a component unknown in any input stays unknown.
+std::vector<IndexDiskUsageRow> aggregate_index_disk_usage(std::vector<IndexDiskUsageRow> rows,
+ IndexDiskUsageLevel level);
+
+// Adds a CLucene sub-file to the component it belongs to. BKD and ANN sub-files only count toward
+// the total and mark the record with their structure.
+void classify_clucene_file(std::string_view name, int64_t length, IndexDiskUsageRecord* record);
+
+// Reads the index file metadata of one segment and reports the bytes of each index.
+class IndexDiskUsageCollector {
+public:
+ // `index_file_info` holds the index file sizes persisted in the rowset meta. Without them the
+ // file sizes are requested from the filesystem.
+ IndexDiskUsageCollector(io::FileSystemSPtr fs, std::string index_path_prefix,
+ TabletSchemaSPtr schema, InvertedIndexStorageFormatPB format,
+ int64_t tablet_id,
+ InvertedIndexFileInfo index_file_info = InvertedIndexFileInfo());
+
+ // Appends one record per index, plus a container record for a shared file when no index
+ // filter is given.
+ Status collect(const IndexDiskUsageOptions& options, std::vector<IndexDiskUsageRecord>* out);
+
+private:
+ Status _collect_v1(const IndexDiskUsageOptions& options,
+ std::vector<IndexDiskUsageRecord>* out);
+ Status _collect_compound(const IndexDiskUsageOptions& options,
+ std::vector<IndexDiskUsageRecord>* out);
+ Status _collect_snii(const IndexDiskUsageOptions& options,
+ std::vector<IndexDiskUsageRecord>* out);
+
+ io::FileSystemSPtr _fs;
+ std::string _index_path_prefix;
+ TabletSchemaSPtr _schema;
+ InvertedIndexStorageFormatPB _format;
+ int64_t _tablet_id;
+ InvertedIndexFileInfo _index_file_info;
+};
+
+// Returns the schema index that describes the physical index `(index_id, suffix)`, or nullptr
+// when the index no longer exists in `schema`.
+const TabletIndex* resolve_disk_usage_index(const TabletSchema& schema, int64_t index_id,
+ std::string_view suffix);
+
+// Appends one row per index record of every segment in `rowset`.
+Status collect_rowset_index_disk_usage(const RowsetSharedPtr& rowset,
+ const IndexDiskUsageOptions& options, int64_t tablet_id,
+ std::vector<IndexDiskUsageRow>* rows);
+
+} // namespace doris::segment_v2
diff --git a/be/src/storage/index/index_file_reader.cpp b/be/src/storage/index/index_file_reader.cpp
index 7612567f9d2..ad3f19d3d97 100644
--- a/be/src/storage/index/index_file_reader.cpp
+++ b/be/src/storage/index/index_file_reader.cpp
@@ -201,7 +201,7 @@ Result<InvertedIndexDirectoryMap> IndexFileReader::get_all_directories() {
std::shared_lock<std::shared_mutex> lock(_mutex); // Lock for reading
for (auto& [index, _] : _indices_entries) {
auto&& [index_id, index_suffix] = index;
- LOG(INFO) << "index_id:" << index_id << " index_suffix:" << index_suffix;
+ VLOG_DEBUG << "index_id:" << index_id << " index_suffix:" << index_suffix;
auto ret = _open(index_id, index_suffix);
if (!ret.has_value()) {
return ResultError(ret.error());
@@ -320,8 +320,8 @@ Result<std::unique_ptr<DorisCompoundReader, DirectoryDeleter>> IndexFileReader::
err.what()));
}
- // 3. read file in DorisCompoundReader
- compound_reader.reset(new DorisCompoundReader(index_input, _read_buffer_size));
+ // 3. read file in DorisCompoundReader with the caller's IO context
+ compound_reader.reset(new DorisCompoundReader(index_input, _read_buffer_size, io_ctx));
} catch (CLuceneError& err) {
// Lazy open can surface a missing file as a read error; keep NotFound distinguishable
if (err.number() == CL_ERR_FileNotFound) {
@@ -360,6 +360,14 @@ Result<std::unique_ptr<DorisCompoundReader, DirectoryDeleter>> IndexFileReader::
Result<std::unique_ptr<doris::snii::reader::LogicalIndexReader>> IndexFileReader::open_snii_index(
const TabletIndex* index_meta, const io::IOContext* io_ctx,
doris::snii::reader::LogicalIndexOpenMode open_mode) const {
+ return open_snii_logical_index(cast_set<uint64_t>(index_meta->index_id()),
+ index_meta->get_index_suffix(), io_ctx, open_mode);
+}
+
+Result<std::unique_ptr<doris::snii::reader::LogicalIndexReader>>
+IndexFileReader::open_snii_logical_index(
+ uint64_t index_id, std::string_view suffix, const io::IOContext* io_ctx,
+ doris::snii::reader::LogicalIndexOpenMode open_mode) const {
DCHECK(_storage_format == InvertedIndexStorageFormatPB::SNII);
std::shared_lock<std::shared_mutex> lock(_mutex);
if (_snii_segment_reader == nullptr) {
@@ -376,16 +384,40 @@ Result<std::unique_ptr<doris::snii::reader::LogicalIndexReader>> IndexFileReader
snii_doris::DorisSniiFileReader::ScopedIOContext io_context_scope(&meta_io_ctx);
auto logical_reader = std::make_unique<doris::snii::reader::LogicalIndexReader>();
- auto status = _snii_segment_reader->open_index(cast_set<uint64_t>(index_meta->index_id()),
- index_meta->get_index_suffix(),
- logical_reader.get(), open_mode);
- auto doris_status = status;
- if (!doris_status.ok()) {
- return ResultError(doris_status);
- }
+ RETURN_IF_ERROR_RESULT(
+ _snii_segment_reader->open_index(index_id, suffix, logical_reader.get(), open_mode));
return logical_reader;
}
+Result<std::vector<doris::snii::format::LogicalIndexMetadataRef>>
+IndexFileReader::snii_logical_indexes() const {
+ DCHECK(_storage_format == InvertedIndexStorageFormatPB::SNII);
+ std::shared_lock<std::shared_mutex> lock(_mutex);
+ if (_snii_segment_reader == nullptr) {
+ return ResultError(Status::Error<ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND>(
+ "SNII index file {} is not opened",
+ InvertedIndexDescriptor::get_index_file_path_v2(_index_path_prefix)));
+ }
+ return _snii_segment_reader->logical_indexes();
+}
+
+Status IndexFileReader::snii_core_metadata(uint64_t index_id, std::string_view suffix,
+ doris::snii::format::CoreMetadata* out,
+ const io::IOContext* io_ctx) const {
+ DCHECK(_storage_format == InvertedIndexStorageFormatPB::SNII);
+ std::shared_lock<std::shared_mutex> lock(_mutex);
+ if (_snii_segment_reader == nullptr) {
+ return Status::Error<ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND>(
+ "SNII index file {} is not opened",
+ InvertedIndexDescriptor::get_index_file_path_v2(_index_path_prefix));
+ }
+ io::IOContext meta_io_ctx = io_ctx != nullptr ? *io_ctx : io::IOContext {};
+ meta_io_ctx.is_inverted_index = true;
+ meta_io_ctx.is_index_data = true;
+ snii_doris::DorisSniiFileReader::ScopedIOContext io_context_scope(&meta_io_ctx);
+ return _snii_segment_reader->core_metadata_for_index(index_id, suffix, out);
+}
+
Result<std::unique_ptr<doris::snii::bkd::BkdSearcher>> IndexFileReader::open_snii_bkd_index(
const TabletIndex* index_meta, const io::IOContext* io_ctx) const {
DCHECK(_storage_format == InvertedIndexStorageFormatPB::SNII);
diff --git a/be/src/storage/index/index_file_reader.h b/be/src/storage/index/index_file_reader.h
index 29bcfe012d9..d0c3628631c 100644
--- a/be/src/storage/index/index_file_reader.h
+++ b/be/src/storage/index/index_file_reader.h
@@ -95,6 +95,16 @@ public:
const TabletIndex* index_meta, const io::IOContext* io_ctx = nullptr,
doris::snii::reader::LogicalIndexOpenMode open_mode =
doris::snii::reader::LogicalIndexOpenMode::kQuery) const;
+ // SNII only: opens one text logical index by its directory key.
+ Result<std::unique_ptr<doris::snii::reader::LogicalIndexReader>> open_snii_logical_index(
+ uint64_t index_id, std::string_view suffix, const io::IOContext* io_ctx,
+ doris::snii::reader::LogicalIndexOpenMode open_mode) const;
+ // SNII only: every entry of the container directory, text and blob kinds alike.
+ Result<std::vector<doris::snii::format::LogicalIndexMetadataRef>> snii_logical_indexes() const;
+ // SNII only: reads the CoreMetadata of one text logical index with the caller's IO context.
+ Status snii_core_metadata(uint64_t index_id, std::string_view suffix,
+ doris::snii::format::CoreMetadata* out,
+ const io::IOContext* io_ctx = nullptr) const;
// SNII only: builds the fully validated inheritance view of this container
// for a BUILD INDEX rewrite. `segment_doc_count` is the segment's row count;
// every kept logical index must agree with it.
diff --git a/be/src/storage/index/inverted/snii/reader/snii_segment_reader.cpp b/be/src/storage/index/inverted/snii/reader/snii_segment_reader.cpp
index 43553a18867..816a2f4e4d4 100644
--- a/be/src/storage/index/inverted/snii/reader/snii_segment_reader.cpp
+++ b/be/src/storage/index/inverted/snii/reader/snii_segment_reader.cpp
@@ -354,6 +354,17 @@ Status SniiSegmentReader::section_refs_for_index(uint64_t index_id, std::string_
if (out == nullptr) {
return Status::Error<ErrorCode::INVALID_ARGUMENT, false>("segment: null section refs out");
}
+ format::CoreMetadata core;
+ RETURN_IF_ERROR(core_metadata_for_index(index_id, suffix, &core));
+ *out = core.section_refs;
+ return Status::OK();
+}
+
+Status SniiSegmentReader::core_metadata_for_index(uint64_t index_id, std::string_view suffix,
+ format::CoreMetadata* const out) const {
+ if (out == nullptr) {
+ return Status::Error<ErrorCode::INVALID_ARGUMENT, false>("segment: null core metadata out");
+ }
if (reader_ == nullptr) {
return Status::Error<ErrorCode::INVALID_ARGUMENT, false>("segment: not opened");
}
@@ -365,7 +376,7 @@ Status SniiSegmentReader::section_refs_for_index(uint64_t index_id, std::string_
&core_bytes));
format::CoreMetadata core;
RETURN_IF_ERROR(format::decode_core_metadata(Slice(core_bytes), &core));
- *out = core.section_refs;
+ *out = core;
return Status::OK();
}
diff --git a/be/src/storage/index/inverted/snii/reader/snii_segment_reader.h b/be/src/storage/index/inverted/snii/reader/snii_segment_reader.h
index 55d1242ae09..c28329d66e3 100644
--- a/be/src/storage/index/inverted/snii/reader/snii_segment_reader.h
+++ b/be/src/storage/index/inverted/snii/reader/snii_segment_reader.h
@@ -110,6 +110,11 @@ public:
uint32_t n_logical_indexes() const { return static_cast<uint32_t>(directory_.size()); }
+ // Every directory entry, text inverted and blob kinds alike, in directory order.
+ const std::vector<format::LogicalIndexMetadataRef>& logical_indexes() const {
+ return directory_.entries();
+ }
+
Status index_exists(uint64_t index_id, std::string_view suffix, bool* const exists) const;
// Loads the adjacent Core/STI/DBD group for (index_id, suffix) and builds a
@@ -118,6 +123,10 @@ public:
LogicalIndexOpenMode open_mode = LogicalIndexOpenMode::kQuery) const;
Status section_refs_for_index(uint64_t index_id, std::string_view suffix,
format::SectionRefs* const out) const;
+ // Reads only the CoreMetadata blob of a text inverted index. Absent index -> NotFound;
+ // a blob entry -> Unsupported.
+ Status core_metadata_for_index(uint64_t index_id, std::string_view suffix,
+ format::CoreMetadata* const out) const;
// Looks up a BLOB logical index entry (kind != kInverted) and exposes its
// validated directory entry (kind + named-file table). Absent -> NotFound;
diff --git a/be/src/storage/segment/page_decoder.h b/be/src/storage/segment/page_decoder.h
index fc48ae3037a..39a1c62b1a1 100644
--- a/be/src/storage/segment/page_decoder.h
+++ b/be/src/storage/segment/page_decoder.h
@@ -20,6 +20,7 @@
#include "common/status.h" // for Status
#include "core/column/column.h"
#include "storage/olap_define.h" // for DISALLOW_COPY_AND_ASSIGN
+#include "storage/segment/common.h"
namespace doris {
namespace segment_v2 {
diff --git a/be/test/format/table/index_disk_usage_reader_test.cpp b/be/test/format/table/index_disk_usage_reader_test.cpp
new file mode 100644
index 00000000000..7fc19c24deb
--- /dev/null
+++ b/be/test/format/table/index_disk_usage_reader_test.cpp
@@ -0,0 +1,235 @@
+// 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 <gen_cpp/PaloInternalService_types.h>
+#include <gen_cpp/PlanNodes_types.h>
+#include <gen_cpp/Types_types.h>
+#include <gtest/gtest.h>
+
+#include <functional>
+#include <memory>
+#include <unordered_map>
+#include <vector>
+
+#include "cloud/cloud_storage_engine.h"
+#include "cloud/cloud_tablet.h"
+#include "cpp/sync_point.h"
+#include "io/io_common.h"
+#include "runtime/exec_env.h"
+#include "runtime/runtime_state.h"
+#include "storage/rowset/rowset_factory.h"
+#include "storage/rowset/rowset_meta.h"
+#include "storage/tablet/tablet_meta.h"
+
+namespace doris {
+
+namespace {
+
+std::unique_ptr<IndexDiskUsageReader> make_reader(RuntimeState* state) {
+ TIndexDiskUsageMetadataParams params;
+ params.__set_level("tablet");
+ TMetaScanRange scan_range;
+ scan_range.__set_index_disk_usage_params(params);
+ return std::make_unique<IndexDiskUsageReader>(std::vector<SlotDescriptor*> {}, state, nullptr,
+ scan_range);
+}
+
+} // namespace
+
+// The reader tags every index file read with its query, so the file cache and the remote scan
+// cache write limiter treat index_disk_usage like any other scan of that query.
+TEST(IndexDiskUsageReaderTest, InitReaderBuildsQueryIoContext) {
+ RuntimeState state;
+ auto reader = make_reader(&state);
+ const Status st = reader->init_reader();
+ ASSERT_TRUE(st.ok()) << st;
+ const io::IOContext* io_ctx = reader->options().io_ctx;
+ ASSERT_NE(io_ctx, nullptr);
+ EXPECT_EQ(ReaderType::READER_QUERY, io_ctx->reader_type);
+ EXPECT_EQ(&state.query_id(), io_ctx->query_id);
+ EXPECT_TRUE(io_ctx->is_inverted_index);
+ EXPECT_FALSE(io_ctx->inverted_index_snii_read_no_write_file_cache);
+}
+
+// The SNII cache write policy of the session applies to this scan as it does to a tablet read.
+TEST(IndexDiskUsageReaderTest, InitReaderCopiesSniiCachePolicy) {
+ RuntimeState state;
+ TQueryOptions query_options;
+ query_options.__set_inverted_index_snii_read_no_write_file_cache(true);
+ state.set_query_options(query_options);
+ auto reader = make_reader(&state);
+ const Status st = reader->init_reader();
+ ASSERT_TRUE(st.ok()) << st;
+ EXPECT_TRUE(reader->options().io_ctx->inverted_index_snii_read_no_write_file_cache);
+}
+
+// Reads of a TTL table are classified by the tablet TTL, so each tablet gets its own context.
+TEST(IndexDiskUsageReaderTest, TabletIoContextCarriesTtl) {
+ TUniqueId query_id;
+ io::IOContext root;
+ root.reader_type = ReaderType::READER_QUERY;
+ root.query_id = &query_id;
+ const io::IOContext tablet_ctx = IndexDiskUsageReader::tablet_io_context(root, 3600);
+ EXPECT_EQ(3600, tablet_ctx.expiration_time);
+ EXPECT_EQ(ReaderType::READER_QUERY, tablet_ctx.reader_type);
+ EXPECT_EQ(&query_id, tablet_ctx.query_id);
+ EXPECT_EQ(0, root.expiration_time);
+}
+
+// Drives IndexDiskUsageReader::capture_tablet_rowsets through a cloud tablet manager whose
+// meta-service calls are replaced, so that the calls can be counted.
+class IndexDiskUsageCaptureTest : public testing::Test {
+protected:
+ static constexpr int64_t kTabletId = 99001;
+
+ void SetUp() override {
+ ExecEnv::GetInstance()->set_storage_engine(
+ std::make_unique<CloudStorageEngine>(EngineOptions {}));
+ auto* sp = SyncPoint::get_instance();
+ sp->clear_all_call_backs();
+ sp->enable_processing();
+ sp->set_call_back("CloudMetaMgr::get_tablet_meta", [](auto&& args) {
+ *try_any_cast<TabletMetaSharedPtr*>(args[1]) = std::make_shared<TabletMeta>(
+ 1, 2, kTabletId, 15674, 4, 5, TTabletSchema(), 6,
+ std::unordered_map<uint32_t, uint32_t> {{7, 8}}, UniqueId(9, 10),
+ TTabletType::TABLET_TYPE_DISK, TCompressionType::LZ4F);
+ try_any_cast_ret<Status>(args)->second = true;
+ });
+ sp->set_call_back("CloudMetaMgr::sync_tablet_rowsets", [this](auto&& args) {
+ ++_syncs;
+ _on_sync(try_any_cast<CloudTablet*>(args[0]));
+ try_any_cast_ret<Status>(args)->second = true;
+ });
+ }
+
+ void TearDown() override {
+ auto* sp = SyncPoint::get_instance();
+ sp->disable_processing();
+ sp->clear_all_call_backs();
+ ExecEnv::GetInstance()->set_storage_engine(nullptr);
+ }
+
+ static RowsetSharedPtr create_rowset(Version version) {
+ auto rs_meta = std::make_shared<RowsetMeta>();
+ rs_meta->set_rowset_type(BETA_ROWSET);
+ rs_meta->set_version(version);
+ rs_meta->set_rowset_id(ExecEnv::GetInstance()->storage_engine().next_rowset_id());
+ RowsetSharedPtr rowset;
+ const Status st = RowsetFactory::create_rowset(nullptr, "", rs_meta, &rowset);
+ EXPECT_TRUE(st.ok()) << st;
+ return rowset;
+ }
+
+ static void add_rowsets(CloudTablet* tablet, std::vector<RowsetSharedPtr> rowsets,
+ bool version_overlap) {
+ std::unique_lock wlock(tablet->get_header_lock());
+ tablet->add_rowsets(std::move(rowsets), version_overlap, wlock, false);
+ }
+
+ // Makes the next meta-service sync deliver one rowset per version through `max_version`.
+ void sync_delivers_versions(int64_t max_version) {
+ _on_sync = [max_version](CloudTablet* tablet) {
+ std::vector<RowsetSharedPtr> rowsets {create_rowset({0, 1})};
+ for (int64_t version = 2; version <= max_version; ++version) {
+ rowsets.push_back(create_rowset({version, version}));
+ }
+ add_rowsets(tablet, std::move(rowsets), false);
+ };
+ }
+
+ // Makes the next meta-service sync deliver a compaction output that replaces cached rowsets.
+ void sync_delivers_compaction(Version output) {
+ _on_sync = [output](CloudTablet* tablet) {
+ add_rowsets(tablet, {create_rowset(output)}, true);
+ };
+ }
+
+ static std::vector<Version> versions_of(const std::vector<RowsetSharedPtr>& rowsets) {
+ std::vector<Version> versions;
+ versions.reserve(rowsets.size());
+ for (const auto& rowset : rowsets) {
+ versions.push_back(rowset->version());
+ }
+ return versions;
+ }
+
+ int _syncs = 0;
+ std::function<void(CloudTablet*)> _on_sync = [](CloudTablet*) {};
+};
+
+// The lookup of a tablet that is not cached loads and synchronizes it, so the scan adds no second
+// meta-service sync.
+TEST_F(IndexDiskUsageCaptureTest, ColdLookupSyncsOnce) {
+ sync_delivers_versions(3);
+
+ auto captured = IndexDiskUsageReader::capture_tablet_rowsets(kTabletId, 3);
+ ASSERT_TRUE(captured.has_value()) << captured.error();
+ EXPECT_EQ(1, _syncs);
+ EXPECT_EQ((std::vector<Version> {{0, 1}, {2, 2}, {3, 3}}), versions_of(captured->rowsets));
+}
+
+// A lookup can join a load that began before the scan version was committed, so a loaded tablet
+// that lacks the scan version synchronizes up to it.
+TEST_F(IndexDiskUsageCaptureTest, ColdLookupSyncsToTheScanVersionWhenTheLoadIsOlder) {
+ _on_sync = [this](CloudTablet* tablet) {
+ if (_syncs == 1) {
+ add_rowsets(tablet,
+ {create_rowset({0, 1}), create_rowset({2, 2}), create_rowset({3, 3})},
+ false);
+ } else {
+ add_rowsets(tablet, {create_rowset({4, 4})}, false);
+ }
+ };
+
+ auto captured = IndexDiskUsageReader::capture_tablet_rowsets(kTabletId, 4);
+ ASSERT_TRUE(captured.has_value()) << captured.error();
+ EXPECT_EQ(2, _syncs);
+ EXPECT_EQ((std::vector<Version> {{0, 1}, {2, 2}, {3, 3}, {4, 4}}),
+ versions_of(captured->rowsets));
+}
+
+// A compaction on another backend can replace the rowsets of a cached tablet without changing the
+// visible version, so a cached tablet synchronizes again even when its cache is current.
+TEST_F(IndexDiskUsageCaptureTest, CachedLookupSyncsCompactionAtCachedVersion) {
+ sync_delivers_versions(3);
+ ASSERT_TRUE(IndexDiskUsageReader::capture_tablet_rowsets(kTabletId, 3).has_value());
+ ASSERT_EQ(1, _syncs);
+
+ sync_delivers_compaction({2, 3});
+ auto captured = IndexDiskUsageReader::capture_tablet_rowsets(kTabletId, 3);
+ ASSERT_TRUE(captured.has_value()) << captured.error();
+ EXPECT_EQ(2, _syncs);
+ EXPECT_EQ((std::vector<Version> {{0, 1}, {2, 3}}), versions_of(captured->rowsets));
+}
+
+// A compaction that also covers a version newer than the scan version keeps the rowsets it
+// replaced as stale rowsets, so the scan version can still be captured.
+TEST_F(IndexDiskUsageCaptureTest, CompactionPastScanVersionKeepsScanVersionReadable) {
+ sync_delivers_versions(3);
+ ASSERT_TRUE(IndexDiskUsageReader::capture_tablet_rowsets(kTabletId, 3).has_value());
+
+ sync_delivers_compaction({2, 4});
+ auto captured = IndexDiskUsageReader::capture_tablet_rowsets(kTabletId, 3);
+ ASSERT_TRUE(captured.has_value()) << captured.error();
+ EXPECT_EQ(2, _syncs);
+ EXPECT_NE(nullptr, captured->tablet->get_rowset_by_version(Version(2, 4)));
+ EXPECT_EQ((std::vector<Version> {{0, 1}, {2, 2}, {3, 3}}), versions_of(captured->rowsets));
+}
+
+} // namespace doris
diff --git a/be/test/storage/index/index_disk_usage_aggregate_test.cpp b/be/test/storage/index/index_disk_usage_aggregate_test.cpp
new file mode 100644
index 00000000000..039a2e2e417
--- /dev/null
+++ b/be/test/storage/index/index_disk_usage_aggregate_test.cpp
@@ -0,0 +1,135 @@
+// 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 <gtest/gtest.h>
+
+#include <string>
+#include <vector>
+
+#include "storage/index/index_disk_usage.h"
+
+namespace doris::segment_v2 {
+
+namespace {
+
+IndexDiskUsageRow make_row(const std::string& rowset_id, int32_t segment_id, int64_t index_id,
+ IndexDiskUsageStructure structure, InvertedIndexStorageFormatPB format,
+ int64_t total_bytes, int64_t position_bytes) {
+ IndexDiskUsageRow row;
+ row.rowset_id = rowset_id;
+ row.segment_id = segment_id;
+ row.segment_count = 1;
+ row.row_count = 100;
+ row.format = format;
+ row.record.index_id = index_id;
+ row.record.structure = structure;
+ row.record.total_bytes = total_bytes;
+ row.record.dict_bytes = total_bytes / 2;
+ row.record.posting_bytes = total_bytes / 2;
+ row.record.position_bytes = position_bytes;
+ return row;
+}
+
+// Two rowsets with two segments each, all holding index 7 in V2.
+std::vector<IndexDiskUsageRow> four_segments() {
+ return {make_row("r1", 0, 7, IndexDiskUsageStructure::kTerm, InvertedIndexStorageFormatPB::V2,
+ 10, 1),
+ make_row("r1", 1, 7, IndexDiskUsageStructure::kTerm, InvertedIndexStorageFormatPB::V2,
+ 20, 2),
+ make_row("r2", 0, 7, IndexDiskUsageStructure::kTerm, InvertedIndexStorageFormatPB::V2,
+ 30, 3),
+ make_row("r2", 1, 7, IndexDiskUsageStructure::kTerm, InvertedIndexStorageFormatPB::V2,
+ 40, 4)};
+}
+
+const IndexDiskUsageRow* find_rowset(const std::vector<IndexDiskUsageRow>& rows,
+ const std::string& rowset_id) {
+ for (const auto& row : rows) {
+ if (row.rowset_id == rowset_id) {
+ return &row;
+ }
+ }
+ return nullptr;
+}
+
+} // namespace
+
+TEST(IndexDiskUsageAggregateTest, SegmentLevelKeepsEveryRow) {
+ auto rows = aggregate_index_disk_usage(four_segments(), IndexDiskUsageLevel::kSegment);
+ ASSERT_EQ(4U, rows.size());
+ for (const auto& row : rows) {
+ EXPECT_EQ(1, row.segment_count);
+ EXPECT_GE(row.segment_id, 0);
+ }
+}
+
+TEST(IndexDiskUsageAggregateTest, RowsetLevelMergesSegments) {
+ auto rows = aggregate_index_disk_usage(four_segments(), IndexDiskUsageLevel::kRowset);
+ ASSERT_EQ(2U, rows.size());
+ const IndexDiskUsageRow* r1 = find_rowset(rows, "r1");
+ ASSERT_NE(r1, nullptr);
+ EXPECT_EQ(-1, r1->segment_id);
+ EXPECT_EQ(2, r1->segment_count);
+ EXPECT_EQ(200, r1->row_count);
+ EXPECT_EQ(30, r1->record.total_bytes);
+ EXPECT_EQ(3, r1->record.position_bytes);
+}
+
+TEST(IndexDiskUsageAggregateTest, TabletLevelMergesRowsets) {
+ auto rows = aggregate_index_disk_usage(four_segments(), IndexDiskUsageLevel::kTablet);
+ ASSERT_EQ(1U, rows.size());
+ EXPECT_EQ("", rows[0].rowset_id);
+ EXPECT_EQ(-1, rows[0].segment_id);
+ EXPECT_EQ(4, rows[0].segment_count);
+ EXPECT_EQ(400, rows[0].row_count);
+ EXPECT_EQ(100, rows[0].record.total_bytes);
+ EXPECT_EQ(50, rows[0].record.dict_bytes);
+ EXPECT_EQ(10, rows[0].record.position_bytes);
+}
+
+TEST(IndexDiskUsageAggregateTest, UnknownComponentStaysUnknown) {
+ auto input = four_segments();
+ input[2].record.position_bytes = -1;
+ auto rows = aggregate_index_disk_usage(std::move(input), IndexDiskUsageLevel::kTablet);
+ ASSERT_EQ(1U, rows.size());
+ EXPECT_EQ(-1, rows[0].record.position_bytes);
+ EXPECT_EQ(100, rows[0].record.total_bytes);
+}
+
+TEST(IndexDiskUsageAggregateTest, ContainerRowsStaySeparate) {
+ auto input = four_segments();
+ input.push_back(make_row("r1", 0, -1, IndexDiskUsageStructure::kContainer,
+ InvertedIndexStorageFormatPB::V2, 5, 0));
+ auto rows = aggregate_index_disk_usage(std::move(input), IndexDiskUsageLevel::kTablet);
+ ASSERT_EQ(2U, rows.size());
+ int64_t container_bytes = -1;
+ for (const auto& row : rows) {
+ if (row.record.structure == IndexDiskUsageStructure::kContainer) {
+ container_bytes = row.record.total_bytes;
+ }
+ }
+ EXPECT_EQ(5, container_bytes);
+}
+
+TEST(IndexDiskUsageAggregateTest, StorageFormatsStaySeparate) {
+ auto input = four_segments();
+ input[3].format = InvertedIndexStorageFormatPB::SNII;
+ auto rows = aggregate_index_disk_usage(std::move(input), IndexDiskUsageLevel::kTablet);
+ ASSERT_EQ(2U, rows.size());
+}
+
+} // namespace doris::segment_v2
diff --git a/be/test/storage/index/index_disk_usage_rowset_test.cpp b/be/test/storage/index/index_disk_usage_rowset_test.cpp
new file mode 100644
index 00000000000..30d9970fce2
--- /dev/null
+++ b/be/test/storage/index/index_disk_usage_rowset_test.cpp
@@ -0,0 +1,249 @@
+// 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 <gen_cpp/Types_types.h>
+#include <gtest/gtest.h>
+
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "common/config.h"
+#include "common/status.h"
+#include "cpp/sync_point.h"
+#include "format/table/index_disk_usage_reader.h"
+#include "io/fs/local_file_system.h"
+#include "io/io_common.h"
+#include "runtime/exec_env.h"
+#include "storage/index/index_disk_usage.h"
+#include "storage/index/index_writer.h"
+#include "storage/olap_common.h"
+#include "storage/options.h"
+#include "storage/rowset/beta_rowset.h"
+#include "storage/rowset/rowset.h"
+#include "storage/rowset/rowset_factory.h"
+#include "storage/rowset/rowset_meta.h"
+#include "storage/rowset/rowset_writer.h"
+#include "storage/rowset/rowset_writer_context.h"
+#include "storage/storage_engine.h"
+#include "storage/tablet/tablet_schema.h"
+#include "util/defer_op.h"
+
+namespace doris::segment_v2 {
+
+class IndexDiskUsageRowsetTest : public ::testing::Test {
+protected:
+ const std::string kTestDir = "./ut_dir/index_disk_usage_rowset_test";
+
+ void SetUp() override {
+ ASSERT_TRUE(io::global_local_filesystem()->delete_directory(kTestDir).ok());
+ ASSERT_TRUE(io::global_local_filesystem()->create_directory(kTestDir).ok());
+ std::vector<StorePath> paths;
+ paths.emplace_back(kTestDir, 1024000000);
+ auto tmp_file_dirs = std::make_unique<TmpFileDirs>(paths);
+ ASSERT_TRUE(tmp_file_dirs->init().ok());
+ ExecEnv::GetInstance()->set_tmp_file_dir(std::move(tmp_file_dirs));
+ EngineOptions options;
+ options.store_paths = paths;
+ auto engine = std::make_unique<StorageEngine>(options);
+ _engine = engine.get();
+ ExecEnv::GetInstance()->set_storage_engine(std::move(engine));
+ }
+
+ void TearDown() override {
+ ExecEnv::GetInstance()->set_storage_engine(nullptr);
+ EXPECT_TRUE(io::global_local_filesystem()->delete_directory(kTestDir).ok());
+ }
+
+ // A duplicate-key table whose value column has an inverted index in the V2 format.
+ static TabletSchemaSPtr create_schema() {
+ auto schema = std::make_shared<TabletSchema>();
+ schema->_keys_type = KeysType::DUP_KEYS;
+ schema->_inverted_index_storage_format = InvertedIndexStorageFormatPB::V2;
+
+ TabletColumn key;
+ key.set_type(FieldType::OLAP_FIELD_TYPE_INT);
+ key.set_unique_id(1);
+ key.set_name("k1");
+ key.set_is_key(true);
+ key.set_index_length(4);
+ schema->append_column(key);
+
+ TabletColumn value;
+ value.set_type(FieldType::OLAP_FIELD_TYPE_INT);
+ value.set_unique_id(2);
+ value.set_name("v1");
+ value.set_is_key(false);
+ schema->append_column(value);
+
+ TabletIndexPB index_pb;
+ index_pb.set_index_type(IndexType::INVERTED);
+ index_pb.set_index_id(10001);
+ index_pb.set_index_name("idx_v1");
+ index_pb.add_col_unique_id(2);
+ TabletIndex index;
+ index.init_from_pb(index_pb);
+ schema->append_index(std::move(index));
+ return schema;
+ }
+
+ // Each test uses its own rowset id, so segment footers cached by another test are not reused.
+ Status write_rowset(const TabletSchemaSPtr& schema, int64_t rowset_id, int32_t num_rows,
+ RowsetSharedPtr* rowset) {
+ RowsetWriterContext context;
+ context.rowset_id.init(rowset_id);
+ context.tablet_id = 1001;
+ context.tablet_schema_hash = 1111;
+ context.partition_id = 10;
+ context.rowset_type = BETA_ROWSET;
+ context.tablet_path = kTestDir;
+ context.rowset_state = VISIBLE;
+ context.tablet_schema = schema;
+ context.version = Version(2, 2);
+ auto writer = DORIS_TRY(RowsetFactory::create_rowset_writer(*_engine, context, false));
+
+ Block block = schema->create_storage_block();
+ auto columns = std::move(block).mutate_columns();
+ for (int32_t i = 0; i < num_rows; ++i) {
+ columns[0]->insert_data(reinterpret_cast<const char*>(&i), sizeof(i));
+ columns[1]->insert_data(reinterpret_cast<const char*>(&i), sizeof(i));
+ }
+ block = schema->create_storage_block();
+ block.set_columns(std::move(columns));
+ RETURN_IF_ERROR(writer->add_block(&block));
+ RETURN_IF_ERROR(writer->flush());
+ return writer->build(*rowset);
+ }
+
+ StorageEngine* _engine = nullptr;
+};
+
+TEST_F(IndexDiskUsageRowsetTest, RowCountFallsBackToSegmentFooters) {
+ RowsetSharedPtr rowset;
+ const Status write_status = write_rowset(create_schema(), 20001, 8, &rowset);
+ ASSERT_TRUE(write_status.ok()) << write_status;
+ // Rowsets written before per-segment row counts were persisted have none in their meta.
+ rowset->rowset_meta()->set_num_segment_rows({});
+
+ TUniqueId query_id;
+ query_id.hi = 7;
+ query_id.lo = 9;
+ io::IOContext io_ctx;
+ io_ctx.reader_type = ReaderType::READER_QUERY;
+ io_ctx.query_id = &query_id;
+ IndexDiskUsageOptions options;
+ options.io_ctx = &io_ctx;
+
+ // The footer reads carry the query context like the index file reads do.
+ int footer_reads = 0;
+ int footer_reads_with_context = 0;
+ auto* sync_point = SyncPoint::get_instance();
+ sync_point->enable_processing();
+ std::vector<IndexDiskUsageRow> rows;
+ Status st;
+ {
+ SyncPoint::CallbackGuard guard;
+ sync_point->set_call_back(
+ "Segment::_parse_footer::io_ctx",
+ [&](auto&& args) {
+ const auto* ctx = try_any_cast<io::IOContext*>(args[0]);
+ ++footer_reads;
+ if (ctx->reader_type == ReaderType::READER_QUERY &&
+ ctx->query_id == &query_id) {
+ ++footer_reads_with_context;
+ }
+ },
+ &guard);
+ st = collect_rowset_index_disk_usage(rowset, options, /*tablet_id=*/1001, &rows);
+ }
+ sync_point->disable_processing();
+ ASSERT_TRUE(st.ok()) << st;
+ ASSERT_FALSE(rows.empty());
+ for (const auto& row : rows) {
+ EXPECT_EQ(8, row.row_count) << "index " << row.record.index_id;
+ }
+ EXPECT_GT(footer_reads, 0);
+ EXPECT_EQ(footer_reads, footer_reads_with_context);
+}
+
+// A failed footer read must not stay cached in the rowset, where later queries and compactions
+// would keep getting it.
+TEST_F(IndexDiskUsageRowsetTest, RowCountFailureIsNotCachedInRowset) {
+ // The corruption test uses local files without a file cache.
+ const bool saved_enable_file_cache = config::enable_file_cache;
+ config::enable_file_cache = false;
+ Defer restore_file_cache {[&] { config::enable_file_cache = saved_enable_file_cache; }};
+
+ RowsetSharedPtr rowset;
+ const Status write_status = write_rowset(create_schema(), 20002, 8, &rowset);
+ ASSERT_TRUE(write_status.ok()) << write_status;
+ rowset->rowset_meta()->set_num_segment_rows({});
+
+ auto* sync_point = SyncPoint::get_instance();
+ sync_point->enable_processing();
+ {
+ SyncPoint::CallbackGuard guard;
+ sync_point->set_call_back(
+ "Segment::parse_footer:magic_number_corruption",
+ [](auto&& args) {
+ auto* buf = try_any_cast<uint8_t*>(args[0]);
+ buf[8] = 0xFF;
+ },
+ &guard);
+ std::vector<IndexDiskUsageRow> rows;
+ EXPECT_FALSE(collect_rowset_index_disk_usage(rowset, IndexDiskUsageOptions {},
+ /*tablet_id=*/1001, &rows)
+ .ok());
+ }
+ sync_point->disable_processing();
+
+ std::vector<uint32_t> segment_rows;
+ OlapReaderStatistics stats;
+ const Status st = std::static_pointer_cast<BetaRowset>(rowset)->get_segment_num_rows(
+ &segment_rows, /*enable_segment_cache=*/false, &stats);
+ ASSERT_TRUE(st.ok()) << st;
+ EXPECT_EQ(std::vector<uint32_t>({8}), segment_rows);
+}
+
+// Rows are labeled with the newest schema among the rowsets, so an index added after the tablet
+// was cached still resolves to its name.
+TEST_F(IndexDiskUsageRowsetTest, LabelSchemaFollowsNewestRowsetSchema) {
+ auto old_schema = create_schema();
+ auto new_schema = std::make_shared<TabletSchema>();
+ new_schema->copy_from(*old_schema);
+ TabletIndexPB index_pb;
+ index_pb.set_index_type(IndexType::INVERTED);
+ index_pb.set_index_id(10002);
+ index_pb.set_index_name("idx_k1");
+ index_pb.add_col_unique_id(1);
+ TabletIndex index;
+ index.init_from_pb(index_pb);
+ new_schema->append_index(std::move(index));
+ new_schema->set_schema_version(old_schema->schema_version() + 1);
+
+ RowsetSharedPtr old_rowset;
+ RowsetSharedPtr new_rowset;
+ ASSERT_TRUE(write_rowset(old_schema, 20003, 4, &old_rowset).ok());
+ ASSERT_TRUE(write_rowset(new_schema, 20004, 4, &new_rowset).ok());
+
+ TabletSchemaSPtr label =
+ IndexDiskUsageReader::label_schema(old_schema, {new_rowset, old_rowset});
+ EXPECT_NE(nullptr, resolve_disk_usage_index(*label, 10002, ""));
+ EXPECT_EQ(old_schema.get(), IndexDiskUsageReader::label_schema(old_schema, {}).get());
+}
+
+} // namespace doris::segment_v2
diff --git a/be/test/storage/index/index_disk_usage_test.cpp b/be/test/storage/index/index_disk_usage_test.cpp
new file mode 100644
index 00000000000..ab1d51d5d78
--- /dev/null
+++ b/be/test/storage/index/index_disk_usage_test.cpp
@@ -0,0 +1,1301 @@
+// 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 <fmt/format.h>
+#include <gen_cpp/Types_types.h>
+#include <gen_cpp/olap_file.pb.h>
+#include <gtest/gtest.h>
+
+#include <functional>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "common/config.h"
+#include "io/fs/file_reader.h"
+#include "io/fs/local_file_system.h"
+#include "io/io_common.h"
+#include "runtime/exec_env.h"
+#include "storage/index/index_file_reader.h"
+#include "storage/index/index_file_writer.h"
+#include "storage/index/index_fs_directory.h"
+#include "storage/index/index_writer.h"
+#include "storage/index/inverted/common/inverted_index_desc.h"
+#include "storage/index/inverted/snii/format/dict_entry.h"
+#include "storage/index/inverted/snii/reader/logical_index_reader.h"
+#include "storage/options.h"
+#include "storage/tablet/tablet_schema.h"
+#include "util/defer_op.h"
+#include "util/slice.h"
+
+namespace doris::segment_v2 {
+
+namespace {
+
+const IndexDiskUsageRecord* find_record(const std::vector<IndexDiskUsageRecord>& records,
+ int64_t index_id, IndexDiskUsageStructure structure) {
+ for (const auto& record : records) {
+ if (record.index_id == index_id && record.structure == structure) {
+ return &record;
+ }
+ }
+ return nullptr;
+}
+
+int64_t sum_total(const std::vector<IndexDiskUsageRecord>& records) {
+ int64_t total = 0;
+ for (const auto& record : records) {
+ total += record.total_bytes;
+ }
+ return total;
+}
+
+} // namespace
+
+TEST(IndexDiskUsageClassifyTest, ClassifiesCluceneFiles) {
+ IndexDiskUsageRecord record;
+ classify_clucene_file("_0.tis", 10, &record);
+ classify_clucene_file("_0.tii", 1, &record);
+ classify_clucene_file("_0.frq", 20, &record);
+ classify_clucene_file("_0.prx", 30, &record);
+ classify_clucene_file("_0.nrm", 4, &record);
+ classify_clucene_file("_0.fnm", 5, &record);
+ classify_clucene_file("segments_3", 6, &record);
+ classify_clucene_file("segments.gen", 7, &record);
+ classify_clucene_file("null_bitmap", 8, &record);
+ EXPECT_EQ(IndexDiskUsageStructure::kTerm, record.structure);
+ EXPECT_EQ(11, record.dict_bytes);
+ EXPECT_EQ(20, record.posting_bytes);
+ EXPECT_EQ(30, record.position_bytes);
+ EXPECT_EQ(4, record.stats_bytes);
+ EXPECT_EQ(26, record.other_bytes);
+ EXPECT_EQ(91, record.total_bytes);
+}
+
+TEST(IndexDiskUsageClassifyTest, BkdFilesMarkBkdStructure) {
+ IndexDiskUsageRecord record;
+ classify_clucene_file("bkd", 100, &record);
+ classify_clucene_file("bkd_meta", 10, &record);
+ classify_clucene_file("bkd_index", 20, &record);
+ classify_clucene_file("null_bitmap", 8, &record);
+ EXPECT_EQ(IndexDiskUsageStructure::kBkd, record.structure);
+ EXPECT_EQ(138, record.total_bytes);
+}
+
+class IndexDiskUsageCollectorTest : public ::testing::Test {
+protected:
+ struct IndexSpec {
+ TabletIndex index;
+ int column_index;
+ std::function<void(IndexColumnWriter*)> feed;
+ };
+
+ const std::string kTestDir = "./ut_dir/index_disk_usage_collector_test";
+
+ void SetUp() override {
+ auto st = io::global_local_filesystem()->delete_directory(kTestDir);
+ ASSERT_TRUE(st.ok()) << st;
+ st = io::global_local_filesystem()->create_directory(kTestDir);
+ ASSERT_TRUE(st.ok()) << st;
+ std::vector<StorePath> paths;
+ paths.emplace_back(kTestDir, 1024);
+ auto tmp_file_dirs = std::make_unique<TmpFileDirs>(paths);
+ st = tmp_file_dirs->init();
+ ASSERT_TRUE(st.ok()) << st;
+ ExecEnv::GetInstance()->set_tmp_file_dir(std::move(tmp_file_dirs));
+ }
+
+ void TearDown() override {
+ EXPECT_TRUE(io::global_local_filesystem()->delete_directory(kTestDir).ok());
+ }
+
+ static TabletSchemaSPtr create_schema() {
+ auto schema = std::make_shared<TabletSchema>();
+ TabletSchemaPB schema_pb;
+ schema_pb.set_keys_type(DUP_KEYS);
+ schema_pb.set_num_short_key_columns(1);
+ schema_pb.set_num_rows_per_row_block(1024);
+ schema_pb.set_compress_kind(COMPRESS_NONE);
+ schema_pb.set_next_column_unique_id(2);
+ schema->init_from_pb(schema_pb);
+
+ TabletColumn key;
+ key.set_name("c1");
+ key.set_type(FieldType::OLAP_FIELD_TYPE_INT);
+ key.set_length(4);
+ key.set_index_length(4);
+ key.set_is_key(true);
+ key.set_is_nullable(true);
+ schema->append_column(key);
+
+ TabletColumn text;
+ text.set_name("c2");
+ text.set_type(FieldType::OLAP_FIELD_TYPE_VARCHAR);
+ text.set_length(65535);
+ text.set_is_key(false);
+ text.set_is_nullable(false);
+ schema->append_column(text);
+ return schema;
+ }
+
+ // The test schema plus a VARIANT column, whose extracted paths can have their own index files.
+ static TabletSchemaSPtr create_variant_schema() {
+ auto schema = create_schema();
+ TabletColumn variant;
+ variant.set_name("v");
+ variant.set_type(FieldType::OLAP_FIELD_TYPE_VARIANT);
+ variant.set_is_key(false);
+ variant.set_is_nullable(true);
+ schema->append_column(variant);
+ return schema;
+ }
+
+ static TabletIndex text_index(int64_t index_id, bool support_phrase) {
+ TabletIndexPB index_pb;
+ index_pb.set_index_type(IndexType::INVERTED);
+ index_pb.set_index_id(index_id);
+ index_pb.set_index_name("idx_text");
+ index_pb.add_col_unique_id(1);
+ (*index_pb.mutable_properties())["parser"] = "english";
+ if (support_phrase) {
+ (*index_pb.mutable_properties())["support_phrase"] = "true";
+ }
+ TabletIndex index;
+ index.init_from_pb(index_pb);
+ return index;
+ }
+
+ static TabletIndex numeric_index(int64_t index_id) {
+ TabletIndexPB index_pb;
+ index_pb.set_index_type(IndexType::INVERTED);
+ index_pb.set_index_id(index_id);
+ index_pb.set_index_name("idx_num");
+ index_pb.add_col_unique_id(0);
+ (*index_pb.mutable_properties())["type"] = "bkd";
+ TabletIndex index;
+ index.init_from_pb(index_pb);
+ return index;
+ }
+
+ static TabletIndex ann_index(int64_t index_id) {
+ TabletIndexPB index_pb;
+ index_pb.set_index_type(IndexType::ANN);
+ index_pb.set_index_id(index_id);
+ index_pb.set_index_name("idx_ann");
+ index_pb.add_col_unique_id(1);
+ TabletIndex index;
+ index.init_from_pb(index_pb);
+ return index;
+ }
+
+ static TabletIndexPB path_index_pb(const std::string& suffix, int64_t index_id = 1) {
+ TabletIndexPB index_pb;
+ index_pb.set_index_type(IndexType::INVERTED);
+ index_pb.set_index_id(index_id);
+ index_pb.set_index_name("idx_text");
+ index_pb.add_col_unique_id(1);
+ (*index_pb.mutable_properties())["parser"] = "english";
+ (*index_pb.mutable_properties())["support_phrase"] = "true";
+ index_pb.set_index_suffix_name(suffix);
+ return index_pb;
+ }
+
+ static TabletIndex path_index(const std::string& suffix, int64_t index_id = 1) {
+ TabletIndex index;
+ index.init_from_pb(path_index_pb(suffix, index_id));
+ return index;
+ }
+
+ // Writes raw sub-files as the V1 index file of `index`, standing in for an index writer such
+ // as the ANN one.
+ static void write_v1_raw_index(const std::string& prefix, const std::string& rowset_id,
+ const TabletIndex& index,
+ const std::vector<std::pair<std::string, std::string>>& files) {
+ IndexFileWriter writer(io::global_local_filesystem(), prefix, rowset_id, 0,
+ InvertedIndexStorageFormatPB::V1);
+ auto directory = writer.open(&index);
+ ASSERT_TRUE(directory.has_value()) << directory.error();
+ for (const auto& [name, bytes] : files) {
+ std::unique_ptr<lucene::store::IndexOutput> output(
+ directory.value()->createOutput(name.c_str()));
+ output->writeBytes(reinterpret_cast<const uint8_t*>(bytes.data()),
+ static_cast<int32_t>(bytes.size()));
+ output->close();
+ }
+ ASSERT_TRUE(writer.begin_close().ok());
+ ASSERT_TRUE(writer.finish_close().ok());
+ }
+
+ static int64_t v1_file_size(const std::string& prefix, int64_t index_id,
+ const std::string& suffix) {
+ int64_t size = 0;
+ EXPECT_TRUE(io::global_local_filesystem()
+ ->file_size(InvertedIndexDescriptor::get_index_file_path_v1(
+ prefix, index_id, suffix),
+ &size)
+ .ok());
+ return size;
+ }
+
+ static bool is_missing_file(const Status& status) {
+ return status.is<ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND>();
+ }
+
+ static void feed_text(IndexColumnWriter* writer) {
+ std::vector<Slice> values = {Slice("hello world"), Slice("quick brown fox"),
+ Slice("hello quick fox")};
+ ASSERT_TRUE(writer->add_values("c2", values.data(), values.size()).ok());
+ }
+
+ static void feed_numbers(IndexColumnWriter* writer) {
+ std::vector<int32_t> values = {42, 100, 42, 200};
+ ASSERT_TRUE(writer->add_values("c1", values.data(), values.size()).ok());
+ }
+
+ // Writes the given indexes of one segment and returns the index path prefix.
+ std::string write_segment(InvertedIndexStorageFormatPB format, const std::string& rowset_id,
+ const TabletSchemaSPtr& schema, std::vector<IndexSpec>* specs) {
+ std::string prefix {InvertedIndexDescriptor::get_index_file_path_prefix(
+ fmt::format("{}/{}_0.dat", kTestDir, rowset_id))};
+ auto fs = io::global_local_filesystem();
+ io::FileWriterPtr file_writer;
+ if (format != InvertedIndexStorageFormatPB::V1) {
+ io::FileWriterOptions opts;
+ EXPECT_TRUE(fs->create_file(InvertedIndexDescriptor::get_index_file_path_v2(prefix),
+ &file_writer, &opts)
+ .ok());
+ }
+ IndexFileWriter index_file_writer(fs, prefix, rowset_id, 0, format, std::move(file_writer));
+ std::vector<std::unique_ptr<IndexColumnWriter>> writers;
+ for (auto& spec : *specs) {
+ std::unique_ptr<IndexColumnWriter> writer;
+ const TabletColumn& column = schema->column(spec.column_index);
+ EXPECT_TRUE(IndexColumnWriter::create(&column, &writer, &index_file_writer, &spec.index)
+ .ok());
+ spec.feed(writer.get());
+ EXPECT_TRUE(writer->finish().ok());
+ writers.push_back(std::move(writer));
+ }
+ EXPECT_TRUE(index_file_writer.begin_close().ok());
+ EXPECT_TRUE(index_file_writer.finish_close().ok());
+ return prefix;
+ }
+
+ static int64_t container_file_size(const std::string& prefix) {
+ int64_t size = 0;
+ EXPECT_TRUE(
+ io::global_local_filesystem()
+ ->file_size(InvertedIndexDescriptor::get_index_file_path_v2(prefix), &size)
+ .ok());
+ return size;
+ }
+
+ void check_text_index_with_positions(InvertedIndexStorageFormatPB format) {
+ auto schema = create_schema();
+ std::vector<IndexSpec> specs {
+ {.index = text_index(1, true), .column_index = 1, .feed = feed_text}};
+ const std::string prefix = write_segment(format, "rs_text", schema, &specs);
+
+ IndexDiskUsageCollector collector(io::global_local_filesystem(), prefix, schema, format,
+ /*tablet_id=*/1001);
+ std::vector<IndexDiskUsageRecord> records;
+ const Status st = collector.collect(IndexDiskUsageOptions {}, &records);
+ ASSERT_TRUE(st.ok()) << st;
+
+ const IndexDiskUsageRecord* text = find_record(records, 1, IndexDiskUsageStructure::kTerm);
+ ASSERT_NE(text, nullptr);
+ EXPECT_GT(text->dict_bytes, 0);
+ EXPECT_GT(text->posting_bytes, 0);
+ EXPECT_GT(text->position_bytes, 0);
+ EXPECT_EQ(text->total_bytes, text->dict_bytes + text->posting_bytes + text->position_bytes +
+ text->stats_bytes + text->other_bytes);
+ const IndexDiskUsageRecord* container =
+ find_record(records, -1, IndexDiskUsageStructure::kContainer);
+ ASSERT_NE(container, nullptr);
+ EXPECT_GE(container->total_bytes, 0);
+ EXPECT_EQ(container_file_size(prefix), sum_total(records));
+ }
+
+ // Enough rows that common terms get non-inline SNII postings with positions.
+ static void feed_many_text(IndexColumnWriter* writer) {
+ const std::vector<std::string> sentences = {"hello world", "quick brown fox",
+ "hello quick fox"};
+ std::vector<std::string> storage;
+ storage.reserve(1200);
+ for (int i = 0; i < 1200; ++i) {
+ storage.push_back(sentences[i % sentences.size()] + " doc" + std::to_string(i));
+ }
+ std::vector<Slice> values(storage.begin(), storage.end());
+ ASSERT_TRUE(writer->add_values("c2", values.data(), values.size()).ok());
+ }
+
+ static std::vector<IndexDiskUsageRecord> collect_snii(const std::string& prefix,
+ const TabletSchemaSPtr& schema,
+ bool position_detail) {
+ IndexDiskUsageCollector collector(io::global_local_filesystem(), prefix, schema,
+ InvertedIndexStorageFormatPB::SNII, 1001);
+ IndexDiskUsageOptions options;
+ options.position_detail = position_detail;
+ std::vector<IndexDiskUsageRecord> records;
+ const Status st = collector.collect(options, &records);
+ EXPECT_TRUE(st.ok()) << st;
+ return records;
+ }
+
+ // Independently sums the position bytes of every pod_ref dictionary entry.
+ static int64_t sum_pod_ref_prx_bytes(const std::string& prefix, const TabletIndex& index) {
+ IndexFileReader reader(io::global_local_filesystem(), prefix,
+ InvertedIndexStorageFormatPB::SNII);
+ const Status init_status = reader.init();
+ if (!init_status.ok()) {
+ ADD_FAILURE() << init_status;
+ return -1;
+ }
+ auto logical = reader.open_snii_index(&index, nullptr,
+ snii::reader::LogicalIndexOpenMode::kCompaction);
+ if (!logical.has_value()) {
+ ADD_FAILURE() << logical.error();
+ return -1;
+ }
+ int64_t bytes = 0;
+ std::vector<snii::format::DictEntry> entries;
+ for (uint32_t block = 0; block < logical.value()->n_dict_blocks(); ++block) {
+ uint64_t frq_base = 0;
+ uint64_t prx_base = 0;
+ const Status st =
+ logical.value()->decode_dict_block(block, &entries, &frq_base, &prx_base);
+ if (!st.ok()) {
+ ADD_FAILURE() << st;
+ return -1;
+ }
+ for (const auto& entry : entries) {
+ if (entry.kind == snii::format::DictEntryKind::kPodRef) {
+ bytes += static_cast<int64_t>(entry.prx_len);
+ }
+ }
+ }
+ return bytes;
+ }
+};
+
+TEST_F(IndexDiskUsageCollectorTest, CollectV2TextIndexWithPositions) {
+ check_text_index_with_positions(InvertedIndexStorageFormatPB::V2);
+}
+
+TEST_F(IndexDiskUsageCollectorTest, CollectV3TextIndexWithPositions) {
+ check_text_index_with_positions(InvertedIndexStorageFormatPB::V3);
+}
+
+TEST_F(IndexDiskUsageCollectorTest, CollectV2DocsOnlyHasNoPositions) {
+ auto schema = create_schema();
+ std::vector<IndexSpec> specs {
+ {.index = text_index(1, false), .column_index = 1, .feed = feed_text}};
+ const std::string prefix =
+ write_segment(InvertedIndexStorageFormatPB::V2, "rs_docs", schema, &specs);
+
+ IndexDiskUsageCollector collector(io::global_local_filesystem(), prefix, schema,
+ InvertedIndexStorageFormatPB::V2, 1001);
+ std::vector<IndexDiskUsageRecord> records;
+ const Status st = collector.collect(IndexDiskUsageOptions {}, &records);
+ ASSERT_TRUE(st.ok()) << st;
+ const IndexDiskUsageRecord* text = find_record(records, 1, IndexDiskUsageStructure::kTerm);
+ ASSERT_NE(text, nullptr);
+ EXPECT_EQ(0, text->position_bytes);
+ EXPECT_GT(text->posting_bytes, 0);
+}
+
+TEST_F(IndexDiskUsageCollectorTest, CollectV2NumericIndexIsBkd) {
+ auto schema = create_schema();
+ std::vector<IndexSpec> specs {
+ {.index = numeric_index(2), .column_index = 0, .feed = feed_numbers}};
+ const std::string prefix =
+ write_segment(InvertedIndexStorageFormatPB::V2, "rs_num", schema, &specs);
+
+ IndexDiskUsageCollector collector(io::global_local_filesystem(), prefix, schema,
+ InvertedIndexStorageFormatPB::V2, 1001);
+ std::vector<IndexDiskUsageRecord> records;
+ const Status st = collector.collect(IndexDiskUsageOptions {}, &records);
+ ASSERT_TRUE(st.ok()) << st;
+ const IndexDiskUsageRecord* bkd = find_record(records, 2, IndexDiskUsageStructure::kBkd);
+ ASSERT_NE(bkd, nullptr);
+ EXPECT_GT(bkd->total_bytes, 0);
+ EXPECT_EQ(container_file_size(prefix), sum_total(records));
+}
+
+TEST_F(IndexDiskUsageCollectorTest, CollectFiltersIndexIds) {
+ auto schema = create_schema();
+ std::vector<IndexSpec> specs {
+ {.index = text_index(1, true), .column_index = 1, .feed = feed_text},
+ {.index = numeric_index(2), .column_index = 0, .feed = feed_numbers}};
+ const std::string prefix =
+ write_segment(InvertedIndexStorageFormatPB::V2, "rs_filter", schema, &specs);
+
+ IndexDiskUsageCollector collector(io::global_local_filesystem(), prefix, schema,
+ InvertedIndexStorageFormatPB::V2, 1001);
+ IndexDiskUsageOptions options;
+ options.index_ids = {2};
+ std::vector<IndexDiskUsageRecord> records;
+ const Status st = collector.collect(options, &records);
+ ASSERT_TRUE(st.ok()) << st;
+ ASSERT_EQ(1U, records.size());
+ EXPECT_EQ(2, records[0].index_id);
+ EXPECT_EQ(IndexDiskUsageStructure::kBkd, records[0].structure);
+}
+
+// A segment may have no index file, for example when every ANN index skipped a segment too small
+// to train, so a missing or empty container holds no index bytes.
+TEST_F(IndexDiskUsageCollectorTest, CollectMissingContainerReportsNothing) {
+ auto schema = create_schema();
+ for (auto format : {InvertedIndexStorageFormatPB::V2, InvertedIndexStorageFormatPB::V3,
+ InvertedIndexStorageFormatPB::SNII}) {
+ const std::string name = InvertedIndexStorageFormatPB_Name(format);
+ const std::string prefix = kTestDir + "/missing_" + name + "_0";
+ IndexDiskUsageCollector collector(io::global_local_filesystem(), prefix, schema, format,
+ 1001);
+ std::vector<IndexDiskUsageRecord> records;
+ const Status st = collector.collect(IndexDiskUsageOptions {}, &records);
+ ASSERT_TRUE(st.ok()) << name << ": " << st;
+ EXPECT_TRUE(records.empty()) << name;
+ }
+}
+
+TEST_F(IndexDiskUsageCollectorTest, CollectEmptyContainerReportsNothing) {
+ auto schema = create_schema();
+ const std::string prefix = kTestDir + "/empty_0";
+ io::FileWriterPtr writer;
+ ASSERT_TRUE(
+ io::global_local_filesystem()
+ ->create_file(InvertedIndexDescriptor::get_index_file_path_v2(prefix), &writer)
+ .ok());
+ ASSERT_TRUE(writer->close().ok());
+
+ IndexDiskUsageCollector collector(io::global_local_filesystem(), prefix, schema,
+ InvertedIndexStorageFormatPB::V2, 1001);
+ std::vector<IndexDiskUsageRecord> records;
+ const Status st = collector.collect(IndexDiskUsageOptions {}, &records);
+ ASSERT_TRUE(st.ok()) << st;
+ EXPECT_TRUE(records.empty());
+}
+
+TEST_F(IndexDiskUsageCollectorTest, CollectMissingV1IndexFileSkipsIndex) {
+ auto schema = create_schema();
+ schema->append_index(text_index(1, true));
+ const std::string prefix = kTestDir + "/missing_v1_0";
+ IndexDiskUsageCollector collector(io::global_local_filesystem(), prefix, schema,
+ InvertedIndexStorageFormatPB::V1, 1001);
+ std::vector<IndexDiskUsageRecord> records;
+ const Status st = collector.collect(IndexDiskUsageOptions {}, &records);
+ ASSERT_TRUE(st.ok()) << st;
+ EXPECT_TRUE(records.empty());
+}
+
+// A file that the rowset meta records with its size was written, so losing it is reported instead
+// of being taken for an index the segment skipped.
+TEST_F(IndexDiskUsageCollectorTest, CollectMissingRecordedContainerFails) {
+ auto schema = create_schema();
+ InvertedIndexFileInfo file_info;
+ file_info.set_index_size(4096);
+ for (auto format : {InvertedIndexStorageFormatPB::V2, InvertedIndexStorageFormatPB::V3,
+ InvertedIndexStorageFormatPB::SNII}) {
+ const std::string name = InvertedIndexStorageFormatPB_Name(format);
+ IndexDiskUsageCollector collector(io::global_local_filesystem(),
+ kTestDir + "/lost_" + name + "_0", schema, format, 1001,
+ file_info);
+ std::vector<IndexDiskUsageRecord> records;
+ const Status st = collector.collect(IndexDiskUsageOptions {}, &records);
+ EXPECT_TRUE(is_missing_file(st)) << name << ": " << st;
+ EXPECT_TRUE(records.empty()) << name;
+ }
+}
+
+TEST_F(IndexDiskUsageCollectorTest, CollectMissingRecordedV1IndexFileFails) {
+ auto schema = create_schema();
+ schema->append_index(text_index(1, true));
+ std::vector<IndexSpec> specs {
+ {.index = text_index(1, true), .column_index = 1, .feed = feed_text}};
+ const std::string prefix =
+ write_segment(InvertedIndexStorageFormatPB::V1, "rs_v1_lost", schema, &specs);
+ InvertedIndexFileInfo file_info;
+ for (const char* suffix : {"", "v%2Ea"}) {
+ auto* index_info = file_info.add_index_info();
+ index_info->set_index_id(1);
+ index_info->set_index_suffix(suffix);
+ index_info->set_index_file_size(v1_file_size(prefix, 1, ""));
+ }
+
+ IndexDiskUsageCollector collector(io::global_local_filesystem(), prefix, schema,
+ InvertedIndexStorageFormatPB::V1, 1001, file_info);
+ std::vector<IndexDiskUsageRecord> records;
+ const Status st = collector.collect(IndexDiskUsageOptions {}, &records);
+ EXPECT_TRUE(is_missing_file(st)) << st;
+}
+
+// A V1 rowset without index file info lists its ANN index files only in the schema.
+TEST_F(IndexDiskUsageCollectorTest, CollectV1AnnIndexWithoutFileInfo) {
+ for (bool with_text_index : {false, true}) {
+ SCOPED_TRACE(with_text_index ? "mixed" : "ann only");
+ const std::string rowset_id = with_text_index ? "rs_v1_ann_mixed" : "rs_v1_ann_only";
+ auto schema = create_schema();
+ std::string prefix {InvertedIndexDescriptor::get_index_file_path_prefix(
+ fmt::format("{}/{}_0.dat", kTestDir, rowset_id))};
+ if (with_text_index) {
+ schema->append_index(text_index(1, true));
+ std::vector<IndexSpec> specs {
+ {.index = text_index(1, true), .column_index = 1, .feed = feed_text}};
+ prefix = write_segment(InvertedIndexStorageFormatPB::V1, rowset_id, schema, &specs);
+ }
+ schema->append_index(ann_index(2));
+ write_v1_raw_index(prefix, rowset_id, ann_index(2),
+ {{"ann.faiss", std::string(200, 'f')}, {"ann.ivfdata", "ivf"}});
+ ASSERT_FALSE(testing::Test::HasFatalFailure());
+
+ IndexDiskUsageCollector collector(io::global_local_filesystem(), prefix, schema,
+ InvertedIndexStorageFormatPB::V1, 1001);
+ std::vector<IndexDiskUsageRecord> records;
+ const Status st = collector.collect(IndexDiskUsageOptions {}, &records);
+ ASSERT_TRUE(st.ok()) << st;
+ const IndexDiskUsageRecord* ann = find_record(records, 2, IndexDiskUsageStructure::kAnn);
+ ASSERT_NE(ann, nullptr);
+ EXPECT_EQ(v1_file_size(prefix, 2, ""), ann->total_bytes);
+ EXPECT_EQ(with_text_index ? 2U : 1U, records.size());
+ if (with_text_index) {
+ EXPECT_NE(find_record(records, 1, IndexDiskUsageStructure::kTerm), nullptr);
+ }
+ }
+}
+
+// A V1 rowset without index file info lists one schema index per extracted VARIANT path, so the
+// schema fallback finds the path files.
+TEST_F(IndexDiskUsageCollectorTest, CollectV1VariantPathFilesFromLegacyRowsetSchema) {
+ TabletSchemaPB schema_pb;
+ create_schema()->to_schema_pb(&schema_pb);
+ const std::vector<std::string> suffixes {"", "v%2Ea", "v%2Eb"};
+ for (const auto& suffix : suffixes) {
+ *schema_pb.add_index() = path_index_pb(suffix);
+ }
+ auto schema = std::make_shared<TabletSchema>();
+ schema->init_from_pb(schema_pb);
+ std::vector<IndexSpec> specs;
+ for (const auto& suffix : suffixes) {
+ specs.push_back({.index = path_index(suffix), .column_index = 1, .feed = feed_text});
+ }
+ const std::string prefix =
+ write_segment(InvertedIndexStorageFormatPB::V1, "rs_v1_legacy_variant", schema, &specs);
+
+ IndexDiskUsageCollector collector(io::global_local_filesystem(), prefix, schema,
+ InvertedIndexStorageFormatPB::V1, 1001);
+ std::vector<IndexDiskUsageRecord> records;
+ const Status st = collector.collect(IndexDiskUsageOptions {}, &records);
+ ASSERT_TRUE(st.ok()) << st;
+ ASSERT_EQ(suffixes.size(), records.size());
+ int64_t files_size = 0;
+ for (size_t i = 0; i < suffixes.size(); ++i) {
+ EXPECT_EQ(suffixes[i], records[i].index_suffix);
+ EXPECT_EQ(IndexDiskUsageStructure::kTerm, records[i].structure);
+ files_size += v1_file_size(prefix, 1, suffixes[i]);
+ }
+ EXPECT_EQ(files_size, sum_total(records));
+}
+
+TEST_F(IndexDiskUsageCollectorTest, CollectCorruptContainerFails) {
+ auto schema = create_schema();
+ const std::string prefix = kTestDir + "/corrupt_0";
+ io::FileWriterPtr writer;
+ ASSERT_TRUE(
+ io::global_local_filesystem()
+ ->create_file(InvertedIndexDescriptor::get_index_file_path_v2(prefix), &writer)
+ .ok());
+ ASSERT_TRUE(writer->append(Slice("not an inverted index container")).ok());
+ ASSERT_TRUE(writer->close().ok());
+
+ // SNII validates its tail, so a damaged file is reported as an error instead of a crash.
+ IndexDiskUsageCollector collector(io::global_local_filesystem(), prefix, schema,
+ InvertedIndexStorageFormatPB::SNII, 1001);
+ std::vector<IndexDiskUsageRecord> records;
+ EXPECT_FALSE(collector.collect(IndexDiskUsageOptions {}, &records).ok());
+}
+
+TEST_F(IndexDiskUsageCollectorTest, CollectV1TextIndex) {
+ auto schema = create_schema();
+ schema->append_index(text_index(1, true));
+ std::vector<IndexSpec> specs {
+ {.index = text_index(1, true), .column_index = 1, .feed = feed_text}};
+ const std::string prefix =
+ write_segment(InvertedIndexStorageFormatPB::V1, "rs_v1", schema, &specs);
+
+ IndexDiskUsageCollector collector(io::global_local_filesystem(), prefix, schema,
+ InvertedIndexStorageFormatPB::V1, 1001);
+ std::vector<IndexDiskUsageRecord> records;
+ const Status st = collector.collect(IndexDiskUsageOptions {}, &records);
+ ASSERT_TRUE(st.ok()) << st;
+ ASSERT_EQ(1U, records.size());
+ EXPECT_EQ(1, records[0].index_id);
+ EXPECT_EQ(IndexDiskUsageStructure::kTerm, records[0].structure);
+ EXPECT_GT(records[0].dict_bytes, 0);
+ EXPECT_GT(records[0].position_bytes, 0);
+ int64_t file_size = 0;
+ ASSERT_TRUE(io::global_local_filesystem()
+ ->file_size(InvertedIndexDescriptor::get_index_file_path_v1(prefix, 1, ""),
+ &file_size)
+ .ok());
+ EXPECT_EQ(file_size, records[0].total_bytes);
+}
+
+TEST_F(IndexDiskUsageCollectorTest, CollectSniiTextIndex) {
+ auto schema = create_schema();
+ std::vector<IndexSpec> specs {
+ {.index = text_index(1, true), .column_index = 1, .feed = feed_many_text}};
+ const std::string prefix =
+ write_segment(InvertedIndexStorageFormatPB::SNII, "rs_snii", schema, &specs);
+
+ auto records = collect_snii(prefix, schema, /*position_detail=*/false);
+ const IndexDiskUsageRecord* text = find_record(records, 1, IndexDiskUsageStructure::kTerm);
+ ASSERT_NE(text, nullptr);
+ EXPECT_GT(text->dict_bytes, 0);
+ EXPECT_GT(text->posting_bytes, 0);
+ EXPECT_GT(text->stats_bytes, 0);
+ EXPECT_EQ(-1, text->position_bytes);
+ const IndexDiskUsageRecord* container =
+ find_record(records, -1, IndexDiskUsageStructure::kContainer);
+ ASSERT_NE(container, nullptr);
+ EXPECT_GE(container->total_bytes, 0);
+ EXPECT_EQ(container_file_size(prefix), sum_total(records));
+}
+
+TEST_F(IndexDiskUsageCollectorTest, CollectSniiDocsOnlyHasNoPositions) {
+ auto schema = create_schema();
+ std::vector<IndexSpec> specs {
+ {.index = text_index(1, false), .column_index = 1, .feed = feed_many_text}};
+ const std::string prefix =
+ write_segment(InvertedIndexStorageFormatPB::SNII, "rs_snii_docs", schema, &specs);
+
+ auto records = collect_snii(prefix, schema, /*position_detail=*/false);
+ const IndexDiskUsageRecord* text = find_record(records, 1, IndexDiskUsageStructure::kTerm);
+ ASSERT_NE(text, nullptr);
+ EXPECT_EQ(0, text->position_bytes);
+ EXPECT_GT(text->posting_bytes, 0);
+}
+
+TEST_F(IndexDiskUsageCollectorTest, CollectSniiPositionDetail) {
+ auto schema = create_schema();
+ TabletIndex index = text_index(1, true);
+ std::vector<IndexSpec> specs {
+ {.index = text_index(1, true), .column_index = 1, .feed = feed_many_text}};
+ const std::string prefix =
+ write_segment(InvertedIndexStorageFormatPB::SNII, "rs_snii_detail", schema, &specs);
+
+ auto coarse = collect_snii(prefix, schema, /*position_detail=*/false);
+ auto detailed = collect_snii(prefix, schema, /*position_detail=*/true);
+ const IndexDiskUsageRecord* c = find_record(coarse, 1, IndexDiskUsageStructure::kTerm);
+ const IndexDiskUsageRecord* d = find_record(detailed, 1, IndexDiskUsageStructure::kTerm);
+ ASSERT_NE(c, nullptr);
+ ASSERT_NE(d, nullptr);
+ const int64_t expected_positions = sum_pod_ref_prx_bytes(prefix, index);
+ EXPECT_GT(expected_positions, 0);
+ EXPECT_EQ(expected_positions, d->position_bytes);
+ EXPECT_EQ(c->posting_bytes, d->posting_bytes + d->position_bytes);
+ EXPECT_EQ(c->total_bytes, d->total_bytes);
+}
+
+TEST_F(IndexDiskUsageCollectorTest, CollectSniiBkdIndex) {
+ auto schema = create_schema();
+ std::vector<IndexSpec> specs {
+ {.index = numeric_index(2), .column_index = 0, .feed = feed_numbers}};
+ const std::string prefix =
+ write_segment(InvertedIndexStorageFormatPB::SNII, "rs_snii_bkd", schema, &specs);
+
+ auto records = collect_snii(prefix, schema, /*position_detail=*/false);
+ const IndexDiskUsageRecord* bkd = find_record(records, 2, IndexDiskUsageStructure::kBkd);
+ ASSERT_NE(bkd, nullptr);
+ EXPECT_GT(bkd->total_bytes, 0);
+ EXPECT_EQ(container_file_size(prefix), sum_total(records));
+}
+
+TEST(IndexDiskUsageClassifyTest, AnnFilesMarkAnnStructure) {
+ IndexDiskUsageRecord record;
+ classify_clucene_file("ann.faiss", 100, &record);
+ classify_clucene_file("ann.ivfdata", 50, &record);
+ EXPECT_EQ(IndexDiskUsageStructure::kAnn, record.structure);
+ EXPECT_EQ(150, record.total_bytes);
+ EXPECT_EQ(0, record.other_bytes);
+}
+
+namespace {
+
+// Forwards to the local filesystem and counts file size lookups, each of which is a remote request
+// on object storage.
+class SizeCountingFileSystem final : public io::FileSystem {
+public:
+ SizeCountingFileSystem() : io::FileSystem("size_counting_fs", io::FileSystemType::LOCAL) {}
+
+ int size_requests() const { return _size_requests; }
+ void reset() { _size_requests = 0; }
+
+protected:
+ Status create_file_impl(const io::Path& file, io::FileWriterPtr* writer,
+ const io::FileWriterOptions* opts) override {
+ return _local->create_file_impl(file, writer, opts);
+ }
+ Status open_file_impl(const io::Path& file, io::FileReaderSPtr* reader,
+ const io::FileReaderOptions* opts) override {
+ // An open without a known size makes the filesystem look the size up.
+ if (opts == nullptr || opts->file_size < 0) {
+ ++_size_requests;
+ }
+ return _local->open_file_impl(file, reader, opts);
+ }
+ Status create_directory_impl(const io::Path& dir, bool failed_if_exists) override {
+ return _local->create_directory_impl(dir, failed_if_exists);
+ }
+ Status delete_file_impl(const io::Path& file) override {
+ return _local->delete_file_impl(file);
+ }
+ Status batch_delete_impl(const std::vector<io::Path>& files) override {
+ return _local->batch_delete_impl(files);
+ }
+ Status delete_directory_impl(const io::Path& dir) override {
+ return _local->delete_directory_impl(dir);
+ }
+ Status exists_impl(const io::Path& path, bool* res) const override {
+ return _local->exists_impl(path, res);
+ }
+ Status file_size_impl(const io::Path& file, int64_t* file_size) const override {
+ ++_size_requests;
+ return _local->file_size_impl(file, file_size);
+ }
+ Status list_impl(const io::Path& dir, bool only_file, std::vector<io::FileInfo>* files,
+ bool* exists) override {
+ return _local->list_impl(dir, only_file, files, exists);
+ }
+ Status rename_impl(const io::Path& orig_name, const io::Path& new_name) override {
+ return _local->rename_impl(orig_name, new_name);
+ }
+ Status absolute_path(const io::Path& path, io::Path& abs_path) const override {
+ return _local->absolute_path(path, abs_path);
+ }
+
+private:
+ std::shared_ptr<io::LocalFileSystem> _local = io::global_local_filesystem();
+ mutable int _size_requests = 0;
+};
+
+} // namespace
+
+// A persisted size is trusted as on the query path, which saves a size request per file.
+TEST_F(IndexDiskUsageCollectorTest, CollectUsesPersistedContainerSize) {
+ auto schema = create_schema();
+ std::vector<IndexSpec> specs {
+ {.index = text_index(1, true), .column_index = 1, .feed = feed_text}};
+ const std::string prefix =
+ write_segment(InvertedIndexStorageFormatPB::V2, "rs_persisted_v2", schema, &specs);
+ const int64_t file_size = container_file_size(prefix);
+ auto fs = std::make_shared<SizeCountingFileSystem>();
+ std::vector<IndexDiskUsageRecord> records;
+
+ IndexDiskUsageCollector unknown_size(fs, prefix, schema, InvertedIndexStorageFormatPB::V2,
+ 1001);
+ ASSERT_TRUE(unknown_size.collect(IndexDiskUsageOptions {}, &records).ok());
+ EXPECT_GT(fs->size_requests(), 0);
+
+ fs->reset();
+ records.clear();
+ InvertedIndexFileInfo file_info;
+ file_info.set_index_size(file_size);
+ IndexDiskUsageCollector persisted_size(fs, prefix, schema, InvertedIndexStorageFormatPB::V2,
+ 1001, file_info);
+ const Status st = persisted_size.collect(IndexDiskUsageOptions {}, &records);
+ ASSERT_TRUE(st.ok()) << st;
+ EXPECT_EQ(0, fs->size_requests());
+ EXPECT_EQ(file_size, sum_total(records));
+}
+
+TEST_F(IndexDiskUsageCollectorTest, CollectUsesPersistedV1IndexFileSize) {
+ auto schema = create_schema();
+ schema->append_index(text_index(1, true));
+ std::vector<IndexSpec> specs {
+ {.index = text_index(1, true), .column_index = 1, .feed = feed_text}};
+ const std::string prefix =
+ write_segment(InvertedIndexStorageFormatPB::V1, "rs_persisted_v1", schema, &specs);
+ int64_t file_size = 0;
+ ASSERT_TRUE(io::global_local_filesystem()
+ ->file_size(InvertedIndexDescriptor::get_index_file_path_v1(prefix, 1, ""),
+ &file_size)
+ .ok());
+ auto fs = std::make_shared<SizeCountingFileSystem>();
+ std::vector<IndexDiskUsageRecord> records;
+
+ IndexDiskUsageCollector unknown_size(fs, prefix, schema, InvertedIndexStorageFormatPB::V1,
+ 1001);
+ ASSERT_TRUE(unknown_size.collect(IndexDiskUsageOptions {}, &records).ok());
+ EXPECT_GT(fs->size_requests(), 0);
+
+ fs->reset();
+ records.clear();
+ InvertedIndexFileInfo file_info;
+ auto* index_info = file_info.add_index_info();
+ index_info->set_index_id(1);
+ index_info->set_index_suffix("");
+ index_info->set_index_file_size(file_size);
+ IndexDiskUsageCollector persisted_size(fs, prefix, schema, InvertedIndexStorageFormatPB::V1,
+ 1001, file_info);
+ const Status st = persisted_size.collect(IndexDiskUsageOptions {}, &records);
+ ASSERT_TRUE(st.ok()) << st;
+ EXPECT_EQ(0, fs->size_requests());
+ EXPECT_EQ(file_size, sum_total(records));
+}
+
+TEST_F(IndexDiskUsageCollectorTest, CollectSniiPositionDetailChecksCancellationPerBlock) {
+ // Small dictionary blocks make the position scan cross many blocks.
+ const int32_t saved_block_bytes = config::snii_target_dict_block_bytes;
+ config::snii_target_dict_block_bytes = 256;
+ Defer restore_block_bytes {[&] { config::snii_target_dict_block_bytes = saved_block_bytes; }};
+ auto schema = create_schema();
+ std::vector<IndexSpec> specs {
+ {.index = text_index(1, true), .column_index = 1, .feed = feed_many_text}};
+ const std::string prefix =
+ write_segment(InvertedIndexStorageFormatPB::SNII, "rs_snii_cancel", schema, &specs);
+
+ int checks = 0;
+ IndexDiskUsageOptions options;
+ options.position_detail = true;
+ options.check_cancelled = [&checks]() {
+ return ++checks < 2 ? Status::OK() : Status::Cancelled("query is cancelled");
+ };
+ IndexDiskUsageCollector collector(io::global_local_filesystem(), prefix, schema,
+ InvertedIndexStorageFormatPB::SNII, 1001);
+ std::vector<IndexDiskUsageRecord> records;
+ const Status st = collector.collect(options, &records);
+ EXPECT_TRUE(st.is<ErrorCode::CANCELLED>()) << st;
+ EXPECT_EQ(2, checks);
+}
+
+// Each SNII logical index costs a core metadata read, so cancellation is checked before it even
+// without a dictionary scan.
+TEST_F(IndexDiskUsageCollectorTest, CollectSniiChecksCancellationPerIndex) {
+ auto schema = create_schema();
+ std::vector<IndexSpec> specs {
+ {.index = text_index(1, true), .column_index = 1, .feed = feed_many_text}};
+ const std::string prefix =
+ write_segment(InvertedIndexStorageFormatPB::SNII, "rs_snii_cancel_idx", schema, &specs);
+
+ int checks = 0;
+ IndexDiskUsageOptions options;
+ options.check_cancelled = [&checks]() {
+ ++checks;
+ return Status::Cancelled("query is cancelled");
+ };
+ IndexDiskUsageCollector collector(io::global_local_filesystem(), prefix, schema,
+ InvertedIndexStorageFormatPB::SNII, 1001);
+ std::vector<IndexDiskUsageRecord> records;
+ const Status st = collector.collect(options, &records);
+ EXPECT_TRUE(st.is<ErrorCode::CANCELLED>()) << st;
+ EXPECT_EQ(1, checks);
+ EXPECT_TRUE(records.empty());
+}
+
+// Each V1 index file is opened separately, so cancellation is checked before every file.
+TEST_F(IndexDiskUsageCollectorTest, CollectV1ChecksCancellationPerIndexFile) {
+ auto schema = create_schema();
+ schema->append_index(text_index(1, true));
+ std::vector<IndexSpec> specs {
+ {.index = text_index(1, true), .column_index = 1, .feed = feed_text}};
+ const std::string prefix =
+ write_segment(InvertedIndexStorageFormatPB::V1, "rs_v1_cancel", schema, &specs);
+
+ int checks = 0;
+ IndexDiskUsageOptions options;
+ options.check_cancelled = [&checks]() {
+ ++checks;
+ return Status::Cancelled("query is cancelled");
+ };
+ IndexDiskUsageCollector collector(io::global_local_filesystem(), prefix, schema,
+ InvertedIndexStorageFormatPB::V1, 1001);
+ std::vector<IndexDiskUsageRecord> records;
+ const Status st = collector.collect(options, &records);
+ EXPECT_TRUE(st.is<ErrorCode::CANCELLED>()) << st;
+ EXPECT_EQ(1, checks);
+ EXPECT_TRUE(records.empty());
+}
+
+// A rowset file system that cannot be resolved, for example after its tablet is dropped, fails
+// the query instead of being dereferenced.
+TEST_F(IndexDiskUsageCollectorTest, CollectWithoutFileSystemFails) {
+ auto schema = create_schema();
+ schema->append_index(text_index(1, true));
+ for (auto format : {InvertedIndexStorageFormatPB::V1, InvertedIndexStorageFormatPB::V2,
+ InvertedIndexStorageFormatPB::SNII}) {
+ IndexDiskUsageCollector collector(nullptr, kTestDir + "/rs_no_fs", schema, format, 1001);
+ std::vector<IndexDiskUsageRecord> records;
+ const Status st = collector.collect(IndexDiskUsageOptions {}, &records);
+ EXPECT_TRUE(st.is<ErrorCode::INIT_FAILED>()) << st;
+ EXPECT_TRUE(records.empty());
+ }
+}
+
+namespace {
+
+// Counts file reads and how many of them carry the query context the collector was given.
+struct ReadContextCounter {
+ const TUniqueId* expected_query_id = nullptr;
+ int reads = 0;
+ int reads_with_context = 0;
+
+ void record(const io::IOContext* io_ctx) {
+ ++reads;
+ if (io_ctx != nullptr && io_ctx->reader_type == ReaderType::READER_QUERY &&
+ io_ctx->query_id == expected_query_id) {
+ ++reads_with_context;
+ }
+ }
+};
+
+class ReadContextRecordingReader final : public io::FileReader {
+public:
+ ReadContextRecordingReader(io::FileReaderSPtr inner, ReadContextCounter* counter)
+ : _inner(std::move(inner)), _counter(counter) {}
+
+ Status close() override { return _inner->close(); }
+ const io::Path& path() const override { return _inner->path(); }
+ size_t size() const override { return _inner->size(); }
+ bool closed() const override { return _inner->closed(); }
+ int64_t mtime() const override { return _inner->mtime(); }
+
+protected:
+ Status read_at_impl(size_t offset, Slice result, size_t* bytes_read,
+ const io::IOContext* io_ctx) override {
+ _counter->record(io_ctx);
+ return _inner->read_at(offset, result, bytes_read, io_ctx);
+ }
+
+private:
+ io::FileReaderSPtr _inner;
+ ReadContextCounter* _counter;
+};
+
+// Forwards to the local filesystem and records the IOContext of every read.
+class ReadContextRecordingFileSystem final : public io::FileSystem {
+public:
+ explicit ReadContextRecordingFileSystem(const TUniqueId* expected_query_id)
+ : io::FileSystem("read_context_recording_fs", io::FileSystemType::LOCAL) {
+ _counter.expected_query_id = expected_query_id;
+ }
+
+ const ReadContextCounter& counter() const { return _counter; }
+
+protected:
+ Status create_file_impl(const io::Path& file, io::FileWriterPtr* writer,
+ const io::FileWriterOptions* opts) override {
+ return _local->create_file_impl(file, writer, opts);
+ }
+ Status open_file_impl(const io::Path& file, io::FileReaderSPtr* reader,
+ const io::FileReaderOptions* opts) override {
+ io::FileReaderSPtr inner;
+ RETURN_IF_ERROR(_local->open_file_impl(file, &inner, opts));
+ *reader = std::make_shared<ReadContextRecordingReader>(std::move(inner), &_counter);
+ return Status::OK();
+ }
+ Status create_directory_impl(const io::Path& dir, bool failed_if_exists) override {
+ return _local->create_directory_impl(dir, failed_if_exists);
+ }
+ Status delete_file_impl(const io::Path& file) override {
+ return _local->delete_file_impl(file);
+ }
+ Status batch_delete_impl(const std::vector<io::Path>& files) override {
+ return _local->batch_delete_impl(files);
+ }
+ Status delete_directory_impl(const io::Path& dir) override {
+ return _local->delete_directory_impl(dir);
+ }
+ Status exists_impl(const io::Path& path, bool* res) const override {
+ return _local->exists_impl(path, res);
+ }
+ Status file_size_impl(const io::Path& file, int64_t* file_size) const override {
+ return _local->file_size_impl(file, file_size);
+ }
+ Status list_impl(const io::Path& dir, bool only_file, std::vector<io::FileInfo>* files,
+ bool* exists) override {
+ return _local->list_impl(dir, only_file, files, exists);
+ }
+ Status rename_impl(const io::Path& orig_name, const io::Path& new_name) override {
+ return _local->rename_impl(orig_name, new_name);
+ }
+ Status absolute_path(const io::Path& path, io::Path& abs_path) const override {
+ return _local->absolute_path(path, abs_path);
+ }
+
+private:
+ std::shared_ptr<io::LocalFileSystem> _local = io::global_local_filesystem();
+ ReadContextCounter _counter;
+};
+
+} // namespace
+
+// Every index file read carries the query context, so the file cache and the remote scan cache
+// write limiter treat index_disk_usage like any other scan of the query.
+TEST_F(IndexDiskUsageCollectorTest, CollectPassesQueryContextToFileReads) {
+ TUniqueId query_id;
+ query_id.hi = 7;
+ query_id.lo = 9;
+ io::IOContext io_ctx;
+ io_ctx.reader_type = ReaderType::READER_QUERY;
+ io_ctx.query_id = &query_id;
+
+ struct Case {
+ InvertedIndexStorageFormatPB format;
+ std::string rowset;
+ bool position_detail = false;
+ };
+ const std::vector<Case> cases {
+ {.format = InvertedIndexStorageFormatPB::V1, .rowset = "rs_ctx_v1"},
+ {.format = InvertedIndexStorageFormatPB::V2, .rowset = "rs_ctx_v2"},
+ {.format = InvertedIndexStorageFormatPB::SNII, .rowset = "rs_ctx_snii"},
+ {.format = InvertedIndexStorageFormatPB::SNII,
+ .rowset = "rs_ctx_snii_pos",
+ .position_detail = true}};
+ for (const Case& c : cases) {
+ SCOPED_TRACE(c.rowset);
+ auto schema = create_schema();
+ if (c.format == InvertedIndexStorageFormatPB::V1) {
+ schema->append_index(text_index(1, true));
+ }
+ std::vector<IndexSpec> specs {
+ {.index = text_index(1, true), .column_index = 1, .feed = feed_many_text}};
+ const std::string prefix = write_segment(c.format, c.rowset, schema, &specs);
+
+ auto fs = std::make_shared<ReadContextRecordingFileSystem>(&query_id);
+ IndexDiskUsageOptions options;
+ options.position_detail = c.position_detail;
+ options.io_ctx = &io_ctx;
+ IndexDiskUsageCollector collector(fs, prefix, schema, c.format, 1001);
+ std::vector<IndexDiskUsageRecord> records;
+ const Status st = collector.collect(options, &records);
+ ASSERT_TRUE(st.ok()) << st;
+ EXPECT_GT(fs->counter().reads, 0);
+ EXPECT_EQ(fs->counter().reads, fs->counter().reads_with_context);
+ }
+}
+
+// A V1 VARIANT index writes one file per extracted path under the parent index id, while the
+// rowset schema only lists the parent index.
+TEST_F(IndexDiskUsageCollectorTest, CollectV1VariantPathFilesFromFileInfo) {
+ auto schema = create_variant_schema();
+ schema->append_index(text_index(1, true));
+ auto path_index = [](const std::string& suffix) {
+ TabletIndexPB index_pb;
+ index_pb.set_index_type(IndexType::INVERTED);
+ index_pb.set_index_id(1);
+ index_pb.set_index_name("idx_text");
+ index_pb.add_col_unique_id(1);
+ (*index_pb.mutable_properties())["parser"] = "english";
+ (*index_pb.mutable_properties())["support_phrase"] = "true";
+ index_pb.set_index_suffix_name(suffix);
+ TabletIndex index;
+ index.init_from_pb(index_pb);
+ return index;
+ };
+ std::vector<IndexSpec> specs {
+ {.index = path_index("v%2Ea"), .column_index = 1, .feed = feed_text},
+ {.index = path_index("v%2Eb"), .column_index = 1, .feed = feed_text}};
+ const std::string prefix =
+ write_segment(InvertedIndexStorageFormatPB::V1, "rs_v1_variant", schema, &specs);
+
+ InvertedIndexFileInfo file_info;
+ int64_t files_size = 0;
+ for (const char* suffix : {"v%2Ea", "v%2Eb"}) {
+ int64_t size = 0;
+ ASSERT_TRUE(io::global_local_filesystem()
+ ->file_size(InvertedIndexDescriptor::get_index_file_path_v1(prefix, 1,
+ suffix),
+ &size)
+ .ok());
+ auto* index_info = file_info.add_index_info();
+ index_info->set_index_id(1);
+ index_info->set_index_suffix(suffix);
+ index_info->set_index_file_size(size);
+ files_size += size;
+ }
+ DirectoryFileNames listing;
+ IndexDiskUsageOptions options;
+ options.directory_files = &listing;
+ IndexDiskUsageCollector collector(io::global_local_filesystem(), prefix, schema,
+ InvertedIndexStorageFormatPB::V1, 1001, file_info);
+ std::vector<IndexDiskUsageRecord> records;
+ const Status st = collector.collect(options, &records);
+ ASSERT_TRUE(st.ok()) << st;
+ ASSERT_EQ(2U, records.size());
+ EXPECT_EQ("v%2Ea", records[0].index_suffix);
+ EXPECT_EQ("v%2Eb", records[1].index_suffix);
+ EXPECT_EQ(files_size, sum_total(records));
+ // The rowset meta names every file, so the directory is not listed.
+ EXPECT_EQ(0, listing.listings());
+}
+
+// A VARIANT rowset without index file info may hold files of extracted paths that its schema does
+// not list, so the segment directory is searched for them.
+TEST_F(IndexDiskUsageCollectorTest, CollectV1VariantPathFilesFromDirectory) {
+ auto schema = create_variant_schema();
+ schema->append_index(text_index(1, true));
+ std::vector<IndexSpec> specs {
+ {.index = text_index(1, true), .column_index = 1, .feed = feed_text},
+ {.index = path_index("v%2Ea"), .column_index = 1, .feed = feed_text},
+ {.index = path_index("v%2Eb"), .column_index = 1, .feed = feed_text},
+ {.index = path_index("v%2Ec", 12), .column_index = 1, .feed = feed_text}};
+ const std::string prefix =
+ write_segment(InvertedIndexStorageFormatPB::V1, "rs_v1_dir", schema, &specs);
+ // Files of other segments, a V2 container, a non-numeric index id and another extension must
+ // not be taken for index files of this segment. Segment 12 shares the prefix of segment 1.
+ for (const char* other : {"rs_v1_dir_1_1@v%2Ec.idx", "rs_v1_dir_012_1.idx", "rs_v1_dir_0.idx",
+ "rs_v1_dir_0_x.idx", "rs_v1_dir_0_1.tmp", "rs_v1_dir_0_-1.idx"}) {
+ io::FileWriterPtr writer;
+ ASSERT_TRUE(
+ io::global_local_filesystem()->create_file(kTestDir + "/" + other, &writer).ok());
+ ASSERT_TRUE(writer->close().ok());
+ }
+
+ IndexDiskUsageCollector collector(io::global_local_filesystem(), prefix, schema,
+ InvertedIndexStorageFormatPB::V1, 1001);
+ std::vector<IndexDiskUsageRecord> records;
+ const Status st = collector.collect(IndexDiskUsageOptions {}, &records);
+ ASSERT_TRUE(st.ok()) << st;
+ ASSERT_EQ(4U, records.size());
+ EXPECT_EQ(std::make_pair(int64_t {1}, std::string()),
+ std::make_pair(records[0].index_id, records[0].index_suffix));
+ EXPECT_EQ("v%2Ea", records[1].index_suffix);
+ EXPECT_EQ("v%2Eb", records[2].index_suffix);
+ EXPECT_EQ(std::make_pair(int64_t {12}, std::string("v%2Ec")),
+ std::make_pair(records[3].index_id, records[3].index_suffix));
+ EXPECT_EQ(v1_file_size(prefix, 1, "") + v1_file_size(prefix, 1, "v%2Ea") +
+ v1_file_size(prefix, 1, "v%2Eb") + v1_file_size(prefix, 12, "v%2Ec"),
+ sum_total(records));
+}
+
+// Only a VARIANT schema can have extracted paths, so other V1 rowsets never list the directory.
+TEST_F(IndexDiskUsageCollectorTest, CollectV1WithoutVariantColumnDoesNotListDirectory) {
+ auto schema = create_schema();
+ schema->append_index(text_index(1, true));
+ std::vector<IndexSpec> specs {
+ {.index = text_index(1, true), .column_index = 1, .feed = feed_text},
+ {.index = path_index("v%2Ea"), .column_index = 1, .feed = feed_text}};
+ const std::string prefix =
+ write_segment(InvertedIndexStorageFormatPB::V1, "rs_v1_nodir", schema, &specs);
+
+ DirectoryFileNames listing;
+ IndexDiskUsageOptions options;
+ options.directory_files = &listing;
+ IndexDiskUsageCollector collector(io::global_local_filesystem(), prefix, schema,
+ InvertedIndexStorageFormatPB::V1, 1001);
+ std::vector<IndexDiskUsageRecord> records;
+ const Status st = collector.collect(options, &records);
+ ASSERT_TRUE(st.ok()) << st;
+ ASSERT_EQ(1U, records.size());
+ EXPECT_EQ(0, listing.listings());
+}
+
+// The segments of a tablet share its directory, so one listing serves all of them.
+TEST_F(IndexDiskUsageCollectorTest, CollectV1SegmentsShareOneDirectoryListing) {
+ auto schema = create_variant_schema();
+ schema->append_index(text_index(1, true));
+ std::vector<std::string> prefixes;
+ for (const char* rowset_id : {"rs_v1_share_a", "rs_v1_share_b"}) {
+ std::vector<IndexSpec> specs {
+ {.index = text_index(1, true), .column_index = 1, .feed = feed_text},
+ {.index = path_index("v%2Ea"), .column_index = 1, .feed = feed_text}};
+ prefixes.push_back(
+ write_segment(InvertedIndexStorageFormatPB::V1, rowset_id, schema, &specs));
+ }
+
+ DirectoryFileNames listing;
+ IndexDiskUsageOptions options;
+ options.directory_files = &listing;
+ for (const std::string& prefix : prefixes) {
+ IndexDiskUsageCollector collector(io::global_local_filesystem(), prefix, schema,
+ InvertedIndexStorageFormatPB::V1, 1001);
+ std::vector<IndexDiskUsageRecord> records;
+ const Status st = collector.collect(options, &records);
+ ASSERT_TRUE(st.ok()) << st;
+ ASSERT_EQ(2U, records.size()) << prefix;
+ EXPECT_EQ("v%2Ea", records[1].index_suffix) << prefix;
+ }
+ EXPECT_EQ(1, listing.listings());
+}
+
+TEST_F(IndexDiskUsageCollectorTest, DirectoryFileNamesListsSortedNamesOnce) {
+ for (const char* name : {"b.idx", "a.idx", "c_1.idx"}) {
+ io::FileWriterPtr writer;
+ ASSERT_TRUE(
+ io::global_local_filesystem()->create_file(kTestDir + "/" + name, &writer).ok());
+ ASSERT_TRUE(writer->close().ok());
+ }
+ DirectoryFileNames listing;
+ const std::vector<std::string>* names = nullptr;
+ ASSERT_TRUE(listing.list(kTestDir, &names).ok());
+ EXPECT_EQ((std::vector<std::string> {"a.idx", "b.idx", "c_1.idx"}), *names);
+ ASSERT_TRUE(listing.list(kTestDir, &names).ok());
+ EXPECT_EQ(1, listing.listings());
+}
+
+TEST_F(IndexDiskUsageCollectorTest, DirectoryFileNamesOfMissingDirectoryIsEmpty) {
+ DirectoryFileNames listing;
+ const std::vector<std::string>* names = nullptr;
+ const Status st = listing.list(kTestDir + "/no_such_dir", &names);
+ ASSERT_TRUE(st.ok()) << st;
+ EXPECT_TRUE(names->empty());
+}
+
+static TabletIndex usage_test_index(int64_t index_id, const std::string& name,
+ const std::string& suffix) {
+ TabletIndexPB index_pb;
+ index_pb.set_index_type(IndexType::INVERTED);
+ index_pb.set_index_id(index_id);
+ index_pb.set_index_name(name);
+ index_pb.add_col_unique_id(1);
+ index_pb.set_index_suffix_name(suffix);
+ TabletIndex index;
+ index.init_from_pb(index_pb);
+ return index;
+}
+
+// An extracted VARIANT path keeps the parent index id with its own suffix, which the current
+// schema does not list.
+TEST(IndexDiskUsageResolveIndexTest, ResolvesVariantPathToParentIndex) {
+ TabletSchema schema;
+ schema.append_index(usage_test_index(1, "idx_v", ""));
+ schema.append_index(usage_test_index(2, "idx_other", ""));
+
+ const TabletIndex* path_index = resolve_disk_usage_index(schema, 1, "v%2Ea");
+ ASSERT_NE(nullptr, path_index);
+ EXPECT_EQ("idx_v", path_index->index_name());
+ const TabletIndex* other = resolve_disk_usage_index(schema, 2, "");
+ ASSERT_NE(nullptr, other);
+ EXPECT_EQ("idx_other", other->index_name());
+ // A dropped index is not resolved.
+ EXPECT_EQ(nullptr, resolve_disk_usage_index(schema, 3, ""));
+}
+
+TEST(IndexDiskUsageResolveIndexTest, PrefersExactSuffixMatch) {
+ TabletSchema schema;
+ schema.append_index(usage_test_index(1, "idx_v", ""));
+ schema.append_index(usage_test_index(1, "idx_v", "v%2Ea"));
+
+ const TabletIndex* index = resolve_disk_usage_index(schema, 1, "v%2Ea");
+ ASSERT_NE(nullptr, index);
+ EXPECT_EQ("v%2Ea", index->get_index_suffix());
+}
+
+} // namespace doris::segment_v2
diff --git a/be/test/storage/index/inverted/snii/reader/snii_segment_reader_logical_indexes_test.cpp b/be/test/storage/index/inverted/snii/reader/snii_segment_reader_logical_indexes_test.cpp
new file mode 100644
index 00000000000..c97e13c5478
--- /dev/null
+++ b/be/test/storage/index/inverted/snii/reader/snii_segment_reader_logical_indexes_test.cpp
@@ -0,0 +1,140 @@
+// 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 <gtest/gtest.h>
+#include <unistd.h>
+
+#include <cstdint>
+#include <cstdio>
+#include <string>
+#include <vector>
+
+#include "common/status.h"
+#include "storage/index/inverted/snii/format/core_metadata.h"
+#include "storage/index/inverted/snii/format/format_constants.h"
+#include "storage/index/inverted/snii/format/metadata_directory.h"
+#include "storage/index/inverted/snii/io/local_file.h"
+#include "storage/index/inverted/snii/reader/snii_segment_reader.h"
+#include "storage/index/inverted/snii/writer/snii_compound_writer.h"
+#include "storage/index/inverted/snii/writer/spimi_term_buffer.h"
+
+using namespace doris::snii;
+using namespace doris::snii::format;
+using namespace doris::snii::reader;
+using namespace doris::snii::writer;
+namespace ErrorCode = doris::ErrorCode;
+
+namespace {
+
+std::string TempPath() {
+ static int counter = 0;
+ return "/tmp/snii_logical_indexes_test_" + std::to_string(getpid()) + "_" +
+ std::to_string(counter++) + ".idx";
+}
+
+SniiIndexInput MakeInput(uint64_t index_id, const std::string& suffix, IndexConfig config) {
+ SpimiTermBuffer buf(/*has_positions=*/has_positions(config));
+ const std::vector<std::vector<std::string>> docs = {
+ {"alpha", "bravo"}, {"bravo", "charlie"}, {"alpha", "charlie", "delta"}};
+ for (uint32_t doc = 0; doc < docs.size(); ++doc) {
+ for (uint32_t pos = 0; pos < docs[doc].size(); ++pos) {
+ buf.add_token(docs[doc][pos], doc, pos);
+ }
+ }
+ SniiIndexInput input;
+ input.index_id = index_id;
+ input.index_suffix = suffix;
+ input.config = config;
+ input.doc_count = static_cast<uint32_t>(docs.size());
+ input.terms = buf.finalize_sorted();
+ input.target_dict_block_bytes = 256;
+ return input;
+}
+
+class SniiSegmentReaderLogicalIndexesTest : public ::testing::Test {
+protected:
+ void SetUp() override {
+ path_ = TempPath();
+ {
+ io::LocalFileWriter writer;
+ ASSERT_TRUE(writer.open(path_).ok());
+ SniiCompoundWriter compound(&writer);
+ ASSERT_TRUE(compound.add_logical_index(MakeInput(7, "", IndexConfig::kDocsOnly)).ok());
+ ASSERT_TRUE(
+ compound.add_logical_index(MakeInput(9, "body", IndexConfig::kDocsPositions))
+ .ok());
+ ASSERT_TRUE(compound.finish().ok());
+ }
+ ASSERT_TRUE(file_.open(path_).ok());
+ ASSERT_TRUE(SniiSegmentReader::open(&file_, &segment_).ok());
+ }
+
+ void TearDown() override { std::remove(path_.c_str()); }
+
+ std::string path_;
+ io::LocalFileReader file_;
+ SniiSegmentReader segment_;
+};
+
+} // namespace
+
+TEST_F(SniiSegmentReaderLogicalIndexesTest, ListsEveryLogicalIndex) {
+ const std::vector<LogicalIndexMetadataRef>& entries = segment_.logical_indexes();
+ ASSERT_EQ(2U, entries.size());
+ bool saw_docs_only = false;
+ bool saw_positions = false;
+ for (const LogicalIndexMetadataRef& entry : entries) {
+ EXPECT_EQ(LogicalIndexKind::kInverted, entry.kind);
+ EXPECT_GT(entry.core_metadata.length, 0U);
+ if (entry.index_id == 7) {
+ EXPECT_EQ("", entry.index_suffix);
+ saw_docs_only = true;
+ } else if (entry.index_id == 9) {
+ EXPECT_EQ("body", entry.index_suffix);
+ saw_positions = true;
+ }
+ }
+ EXPECT_TRUE(saw_docs_only);
+ EXPECT_TRUE(saw_positions);
+}
+
+TEST_F(SniiSegmentReaderLogicalIndexesTest, ReadsCoreMetadataOfEachIndex) {
+ CoreMetadata docs_only;
+ ASSERT_TRUE(segment_.core_metadata_for_index(7, "", &docs_only).ok());
+ EXPECT_EQ(IndexConfig::kDocsOnly, docs_only.index_config);
+ EXPECT_GT(docs_only.section_refs.dict_region.length, 0U);
+ EXPECT_EQ(3U, docs_only.stats.doc_count);
+
+ // This tiny corpus inlines every posting into dictionary blocks, so only the dictionary
+ // region is guaranteed to be non-empty.
+ CoreMetadata positions;
+ ASSERT_TRUE(segment_.core_metadata_for_index(9, "body", &positions).ok());
+ EXPECT_EQ(IndexConfig::kDocsPositions, positions.index_config);
+ EXPECT_GT(positions.section_refs.dict_region.length, 0U);
+ EXPECT_EQ(3U, positions.stats.doc_count);
+
+ SectionRefs refs;
+ ASSERT_TRUE(segment_.section_refs_for_index(9, "body", &refs).ok());
+ EXPECT_EQ(positions.section_refs.dict_region.offset, refs.dict_region.offset);
+ EXPECT_EQ(positions.section_refs.dict_region.length, refs.dict_region.length);
+}
+
+TEST_F(SniiSegmentReaderLogicalIndexesTest, MissingIndexIsNotFound) {
+ CoreMetadata core;
+ const doris::Status status = segment_.core_metadata_for_index(8, "", &core);
+ EXPECT_TRUE(status.is<ErrorCode::INVERTED_INDEX_SNII_NOT_FOUND>()) << status;
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinTableValuedFunctions.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinTableValuedFunctions.java
index 508eaa0967e..029dddb9eb2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinTableValuedFunctions.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinTableValuedFunctions.java
@@ -28,6 +28,7 @@ import org.apache.doris.nereids.trees.expressions.functions.table.GroupCommit;
import org.apache.doris.nereids.trees.expressions.functions.table.Hdfs;
import org.apache.doris.nereids.trees.expressions.functions.table.Http;
import org.apache.doris.nereids.trees.expressions.functions.table.HttpStream;
+import org.apache.doris.nereids.trees.expressions.functions.table.IndexDiskUsage;
import org.apache.doris.nereids.trees.expressions.functions.table.Jobs;
import org.apache.doris.nereids.trees.expressions.functions.table.Local;
import org.apache.doris.nereids.trees.expressions.functions.table.MvInfos;
@@ -65,6 +66,7 @@ public class BuiltinTableValuedFunctions implements FunctionHelper {
tableValued(S3.class, "s3"),
tableValued(MvInfos.class, "mv_infos"),
tableValued(Partitions.class, "partitions"),
+ tableValued(IndexDiskUsage.class, "index_disk_usage"),
tableValued(Jobs.class, "jobs"),
tableValued(Tasks.class, "tasks"),
tableValued(Query.class, "query"),
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/IndexDiskUsageScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/IndexDiskUsageScanNode.java
new file mode 100644
index 00000000000..2bfbc8c3697
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/IndexDiskUsageScanNode.java
@@ -0,0 +1,257 @@
+// 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()) {
+ String clusterId = ((CloudSystemInfoService) systemInfo).getCurrentClusterId();
+ selector = cloudSelector(replica -> ((CloudReplica) replica).getBackendIdWithClusterId(clusterId),
+ systemInfo::getBackend);
+ } 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);
+ };
+ }
+
+ // Resolves the backend that a cloud replica routes queries to in the current compute group.
+ interface ReplicaBackendResolver {
+ long backendId(Replica replica) throws UserException;
+ }
+
+ static BackendSelector cloudSelector(ReplicaBackendResolver resolver, LongFunction<Backend> backendLookup) {
+ return target -> {
+ for (Replica replica : target.getTablet().getReplicas()) {
+ long backendId = resolver.backendId(replica);
+ Backend backend = backendLookup.apply(backendId);
+ if (backend == null || !backend.isQueryAvailable()) {
+ continue;
+ }
+ // A smooth upgrade keeps the old backend as a query fallback, and that version
+ // returns no rows for this metadata scan.
+ if (backend.isSmoothUpgradeSrc()) {
+ throw new UserException("index_disk_usage is unavailable while backend " + backendId
+ + " is a smooth upgrade source");
+ }
+ return backendId;
+ }
+ throw new UserException("No queryable replica for tablet " + target.getTabletId());
+ };
+ }
+
+ private static Set<Long> alivePathHashes(Backend backend) {
+ return backend.getDisks().values().stream()
+ .filter(DiskInfo::isAlive)
+ .map(DiskInfo::getPathHash)
+ .collect(Collectors.toSet());
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/table/IndexDiskUsage.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/table/IndexDiskUsage.java
new file mode 100644
index 00000000000..4557489f6ca
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/table/IndexDiskUsage.java
@@ -0,0 +1,58 @@
+// 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.nereids.trees.expressions.functions.table;
+
+import org.apache.doris.catalog.FunctionSignature;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.Properties;
+import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor;
+import org.apache.doris.nereids.types.coercion.AnyDataType;
+import org.apache.doris.tablefunction.IndexDiskUsageTableValuedFunction;
+import org.apache.doris.tablefunction.TableValuedFunctionIf;
+
+import java.util.Map;
+
+/**
+ * index_disk_usage
+ */
+public class IndexDiskUsage extends TableValuedFunction {
+ public IndexDiskUsage(Properties properties) {
+ super(IndexDiskUsageTableValuedFunction.NAME, properties);
+ }
+
+ @Override
+ public FunctionSignature customSignature() {
+ return FunctionSignature.of(AnyDataType.INSTANCE_WITHOUT_INDEX, getArgumentsTypes());
+ }
+
+ @Override
+ protected TableValuedFunctionIf toCatalogFunction() {
+ try {
+ Map<String, String> arguments = getTVFProperties().getMap();
+ return new IndexDiskUsageTableValuedFunction(arguments);
+ } catch (Throwable t) {
+ throw new AnalysisException("Can not build IndexDiskUsageTableValuedFunction by "
+ + this + ": " + t.getMessage(), t);
+ }
+ }
+
+ @Override
+ public <R, C> R accept(ExpressionVisitor<R, C> visitor, C context) {
+ return visitor.visitIndexDiskUsage(this, context);
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/TableValuedFunctionVisitor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/TableValuedFunctionVisitor.java
index d91bb2610c0..fd5154a0018 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/TableValuedFunctionVisitor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/TableValuedFunctionVisitor.java
@@ -28,6 +28,7 @@ import org.apache.doris.nereids.trees.expressions.functions.table.GroupCommit;
import org.apache.doris.nereids.trees.expressions.functions.table.Hdfs;
import org.apache.doris.nereids.trees.expressions.functions.table.Http;
import org.apache.doris.nereids.trees.expressions.functions.table.HttpStream;
+import org.apache.doris.nereids.trees.expressions.functions.table.IndexDiskUsage;
import org.apache.doris.nereids.trees.expressions.functions.table.Jobs;
import org.apache.doris.nereids.trees.expressions.functions.table.Local;
import org.apache.doris.nereids.trees.expressions.functions.table.MvInfos;
@@ -68,6 +69,10 @@ public interface TableValuedFunctionVisitor<R, C> {
return visitTableValuedFunction(partitions, context);
}
+ default R visitIndexDiskUsage(IndexDiskUsage indexDiskUsage, C context) {
+ return visitTableValuedFunction(indexDiskUsage, context);
+ }
+
default R visitPartitionValues(PartitionValues partitionValues, C context) {
return visitTableValuedFunction(partitionValues, context);
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
index fce942267a0..26a504ba9b0 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
@@ -1154,9 +1154,8 @@ public class OlapScanNode extends ScanNode {
return scanRangeLocations;
}
- @VisibleForTesting
- static boolean shouldFilterReplicaByResourceTag(boolean isInvalidComputeGroup, boolean isNotCloudComputeGroup,
- ComputeGroup computeGroup, String beTagName) {
+ public static boolean shouldFilterReplicaByResourceTag(boolean isInvalidComputeGroup,
+ boolean isNotCloudComputeGroup, ComputeGroup computeGroup, String beTagName) {
return isInvalidComputeGroup
|| (Config.enable_resource_tag_location_check
&& isNotCloudComputeGroup && !computeGroup.containsBackend(beTagName));
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index 81cb2a8cf3a..542ddbc5174 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -95,6 +95,9 @@ public class SessionVariable implements Serializable, Writable {
public static final String LOCAL_EXCHANGE_FREE_BLOCKS_LIMIT = "local_exchange_free_blocks_limit";
public static final String SCAN_QUEUE_MEM_LIMIT = "scan_queue_mem_limit";
public static final String MAX_SCANNERS_CONCURRENCY = "max_scanners_concurrency";
+ public static final String INDEX_DISK_USAGE_MAX_TABLETS = "index_disk_usage_max_tablets";
+ public static final String INDEX_DISK_USAGE_POSITION_DETAIL_MAX_TABLETS =
+ "index_disk_usage_position_detail_max_tablets";
public static final String MAX_FILE_SCANNERS_CONCURRENCY = "max_file_scanners_concurrency";
public static final String ENABLE_FILE_SCANNER_V2 = "enable_file_scanner_v2";
public static final String MIN_SCANNERS_CONCURRENCY = "min_scanners_concurrency";
@@ -1172,6 +1175,15 @@ public class SessionVariable implements Serializable, Writable {
+ "data of ScanNode, default 4")
public int maxScannersConcurrency = 4;
+ @VarAttrDef.VarAttr(name = INDEX_DISK_USAGE_MAX_TABLETS, needForward = true, description = "The max number "
+ + "of tablets index_disk_usage may scan in one query, default 20000")
+ public int indexDiskUsageMaxTablets = 20000;
+
+ @VarAttrDef.VarAttr(name = INDEX_DISK_USAGE_POSITION_DETAIL_MAX_TABLETS, needForward = true,
+ description = "The max number of tablets index_disk_usage may scan when position_detail is enabled, "
+ + "default 10000")
+ public int indexDiskUsagePositionDetailMaxTablets = 10000;
+
@VarAttrDef.VarAttr(name = MAX_FILE_SCANNERS_CONCURRENCY, needForward = true,
checker = "checkMaxFileScannersConcurrency", description = "The max threads to "
+ "read data of FileScanNode, default 16")
diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/IndexDiskUsageTableValuedFunction.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/IndexDiskUsageTableValuedFunction.java
new file mode 100644
index 00000000000..f06f99cb5ba
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/IndexDiskUsageTableValuedFunction.java
@@ -0,0 +1,478 @@
+// 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.annotations.VisibleForTesting;
+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.Set;
+import java.util.function.Function;
+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");
+ // How many times the cloud snapshot is attempted when a rollup or partition changes during the fetch.
+ private static final int MAX_SNAPSHOT_ATTEMPTS = 3;
+
+ 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 {
+ this(params, IndexDiskUsageTableValuedFunction::cloudVisibleVersions);
+ }
+
+ // `cloudVersions` reads the visible versions of partitions from meta-service in cloud mode.
+ @VisibleForTesting
+ IndexDiskUsageTableValuedFunction(Map<String, String> params,
+ Function<List<Partition>, List<Long>> cloudVersions) 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);
+ Snapshot snapshot = takeConsistentSnapshot(table, validParams, dbName + "." + tableName, cloudVersions);
+ List<TabletTarget> targets = Lists.newArrayList();
+ for (int i = 0; i < snapshot.partitions.size(); ++i) {
+ long partitionId = snapshot.partitions.get(i).getId();
+ for (Pair<Long, List<Tablet>> indexTablets : snapshot.tabletsByPartition.get(partitionId)) {
+ for (Tablet tablet : indexTablets.second) {
+ targets.add(new TabletTarget(tablet, partitionId, indexTablets.first, snapshot.versions.get(i)));
+ }
+ }
+ }
+ this.indexIds = ImmutableList.copyOf(snapshot.indexIds);
+ this.partitionNames = snapshot.partitionNames;
+ this.materializedIndexNames = snapshot.materializedIndexNames;
+ this.tabletTargets = ImmutableList.copyOf(targets);
+ }
+
+ // The tablets of the selected partitions, copied under the table lock, with the visible version of
+ // each partition.
+ private static final class Snapshot {
+ private List<Long> indexIds;
+ private List<Partition> partitions;
+ private final Map<Long, String> partitionNames = Maps.newLinkedHashMap();
+ private final Map<Long, String> materializedIndexNames = Maps.newLinkedHashMap();
+ private final Map<Long, List<Pair<Long, List<Tablet>>>> tabletsByPartition = Maps.newHashMap();
+ private List<Long> versions;
+ }
+
+ private Snapshot takeSnapshot(OlapTable table, Map<String, String> params, String qualifiedName) {
+ Snapshot snapshot = new Snapshot();
+ table.readLock();
+ try {
+ snapshot.indexIds = resolveIndexIds(table, params.get(INDEXES), qualifiedName);
+ snapshot.partitions = resolvePartitions(table, params.get(PARTITIONS), qualifiedName);
+ // Counting is cheap, so an over-limit request fails before anything is copied or fetched.
+ checkTabletLimits(countTablets(snapshot.partitions));
+ for (Partition partition : snapshot.partitions) {
+ snapshot.partitionNames.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)) {
+ snapshot.materializedIndexNames.putIfAbsent(index.getId(),
+ table.getIndexNameById(index.getId()));
+ indexTablets.add(Pair.of(index.getId(), Lists.newArrayList(index.getTablets())));
+ }
+ snapshot.tabletsByPartition.put(partition.getId(), indexTablets);
+ }
+ // Local replica choice filters replicas by version, so read it with the tablets it applies to.
+ if (!Config.isCloudMode()) {
+ snapshot.versions = snapshot.partitions.stream().map(Partition::getVisibleVersion)
+ .collect(Collectors.toList());
+ }
+ } finally {
+ table.readUnlock();
+ }
+ return snapshot;
+ }
+
+ // Cloud versions come from meta-service, so they are fetched without holding the table lock. A
+ // rollup that is dropped meanwhile would pair its tablets with a version they never received, and
+ // one that becomes visible would be missing from the copy, so the partitions and their visible
+ // indexes are checked again after the fetch.
+ private Snapshot takeConsistentSnapshot(OlapTable table, Map<String, String> params,
+ String qualifiedName, Function<List<Partition>, List<Long>> cloudVersions) {
+ for (int attempt = 1;; ++attempt) {
+ Snapshot snapshot = takeSnapshot(table, params, qualifiedName);
+ if (!Config.isCloudMode()) {
+ return snapshot;
+ }
+ snapshot.versions = cloudVersions.apply(snapshot.partitions);
+ if (isUnchanged(table, snapshot)) {
+ return snapshot;
+ }
+ if (attempt == MAX_SNAPSHOT_ATTEMPTS) {
+ throw new AnalysisException("The partitions or rollups of table " + qualifiedName
+ + " kept changing while index_disk_usage collected their tablets, please retry");
+ }
+ }
+ }
+
+ private static boolean isUnchanged(OlapTable table, Snapshot snapshot) {
+ table.readLock();
+ try {
+ for (Partition partition : snapshot.partitions) {
+ Partition current = table.getPartition(partition.getId());
+ if (current == null) {
+ return false;
+ }
+ Set<Long> currentIndexIds = current.getMaterializedIndices(IndexExtState.VISIBLE).stream()
+ .map(MaterializedIndex::getId).collect(Collectors.toSet());
+ Set<Long> copiedIndexIds = snapshot.tabletsByPartition.get(partition.getId()).stream()
+ .map(indexTablets -> indexTablets.first).collect(Collectors.toSet());
+ if (!currentIndexIds.equals(copiedIndexIds)) {
+ return false;
+ }
+ }
+ return true;
+ } finally {
+ table.readUnlock();
+ }
+ }
+
+ public List<TabletTarget> getTabletTargets() {
+ return tabletTargets;
+ }
+
+ @Override
+ public TMetadataType getMetadataType() {
+ return TMetadataType.INDEX_DISK_USAGE;
+ }
+
+ @Override
+ public TMetaScanRange getMetaScanRange(List<String> requiredFields) {
+ TIndexDiskUsageMetadataParams params = new TIndexDiskUsageMetadataParams();
+ params.setLevel(level);
+ params.setPositionDetail(positionDetail);
+ params.setIndexIds(Lists.newArrayList(indexIds));
+ params.setPartitionNames(Maps.newHashMap(partitionNames));
+ params.setMaterializedIndexNames(Maps.newHashMap(materializedIndexNames));
+ params.setTablets(tabletTargets.stream().map(TabletTarget::toThrift).collect(Collectors.toList()));
+ TMetaScanRange metaScanRange = new TMetaScanRange();
+ metaScanRange.setMetadataType(TMetadataType.INDEX_DISK_USAGE);
+ metaScanRange.setIndexDiskUsageParams(params);
+ return metaScanRange;
+ }
+
+ @Override
+ public ScanNode getScanNode(PlanNodeId id, TupleDescriptor desc, SessionVariable sv) {
+ return new IndexDiskUsageScanNode(id, desc, this,
+ ScanContext.builder().clusterName(sv.resolveCloudClusterName()).build());
+ }
+
+ @Override
+ public String getTableName() {
+ return "IndexDiskUsageTableValuedFunction";
+ }
+
+ @Override
+ public List<Column> getTableColumns() {
+ return SCHEMA;
+ }
+
+ private static Column varcharColumn(String name) {
+ return new Column(name, ScalarType.createVarcharType(ScalarType.MAX_VARCHAR_LENGTH), true);
+ }
+
+ private static Column bigintColumn(String name) {
+ return new Column(name, ScalarType.createType(PrimitiveType.BIGINT), true);
+ }
+
+ private static String parseLevel(String raw) {
+ String normalized = raw.toLowerCase();
+ if (!LEVELS.contains(normalized)) {
+ throw new AnalysisException("Unsupported level '" + raw
+ + "' for index_disk_usage, expected tablet, rowset or segment");
+ }
+ return normalized;
+ }
+
+ private static boolean parsePositionDetail(String raw) {
+ if ("true".equalsIgnoreCase(raw)) {
+ return true;
+ }
+ if ("false".equalsIgnoreCase(raw)) {
+ return false;
+ }
+ throw new AnalysisException("Invalid position_detail '" + raw + "', expected true or false");
+ }
+
+ private static void checkShowPrivilege(String dbName, String tableName) {
+ ConnectContext ctx = ConnectContext.get();
+ if (!Env.getCurrentEnv().getAccessManager().checkTblPriv(ctx, InternalCatalog.INTERNAL_CATALOG_NAME,
+ dbName, tableName, PrivPredicate.SHOW)) {
+ throw new AnalysisException(ErrorCode.ERR_TABLEACCESS_DENIED_ERROR.formatErrorMsg("SHOW",
+ ctx.getQualifiedUser(), ctx.getRemoteIP(), dbName + ": " + tableName));
+ }
+ }
+
+ private static OlapTable getOlapTable(String dbName, String tableName) {
+ TableIf table;
+ try {
+ Database db = Env.getCurrentInternalCatalog().getDbOrAnalysisException(dbName);
+ table = db.getTableOrAnalysisException(tableName);
+ } catch (org.apache.doris.common.AnalysisException e) {
+ throw new AnalysisException(e.getMessage(), e);
+ }
+ if (!(table instanceof OlapTable)) {
+ throw new AnalysisException("index_disk_usage only supports OLAP table");
+ }
+ return (OlapTable) table;
+ }
+
+ private static List<Partition> resolvePartitions(OlapTable table, String raw, String qualifiedName) {
+ if (StringUtils.isEmpty(raw)) {
+ return Lists.newArrayList(table.getPartitions());
+ }
+ List<Partition> partitions = Lists.newArrayList();
+ for (String name : splitNames(PARTITIONS, raw)) {
+ // Temporary partitions are excluded, so look the name up among formal partitions only.
+ Partition partition = table.getPartition(name, false);
+ if (partition == null) {
+ throw new AnalysisException("Unknown partition '" + name + "' in table " + qualifiedName);
+ }
+ partitions.add(partition);
+ }
+ return partitions;
+ }
+
+ private static List<Long> resolveIndexIds(OlapTable table, String raw, String qualifiedName) {
+ if (StringUtils.isEmpty(raw)) {
+ return Lists.newArrayList();
+ }
+ List<Long> ids = Lists.newArrayList();
+ for (String name : splitNames(INDEXES, raw)) {
+ Index index = table.getIndexes().stream()
+ .filter(candidate -> candidate.getIndexName().equalsIgnoreCase(name))
+ .findFirst()
+ .orElseThrow(() -> new AnalysisException(
+ "Unknown index '" + name + "' in table " + qualifiedName));
+ if (index.getIndexType() != IndexType.INVERTED && index.getIndexType() != IndexType.ANN) {
+ throw new AnalysisException("Index '" + name + "' in table " + qualifiedName
+ + " is not an inverted or ANN index");
+ }
+ ids.add(index.getIndexId());
+ }
+ return ids;
+ }
+
+ private static List<Long> cloudVisibleVersions(List<Partition> partitions) {
+ List<CloudPartition> cloudPartitions =
+ partitions.stream().map(CloudPartition.class::cast).collect(Collectors.toList());
+ List<Long> versions;
+ try {
+ versions = CloudPartition.getSnapshotVisibleVersion(cloudPartitions);
+ } catch (RpcException e) {
+ throw new AnalysisException("Failed to get partition visible versions: " + e.getMessage(), e);
+ }
+ for (int i = 0; i < versions.size(); ++i) {
+ if (versions.get(i) < 0) {
+ throw new AnalysisException(
+ "Visible version of partition '" + partitions.get(i).getName() + "' is not found");
+ }
+ }
+ return versions;
+ }
+
+ // A filter that names nothing would silently widen or empty the scan, so it is rejected.
+ private static Collection<String> splitNames(String property, String raw) {
+ Collection<String> names = Arrays.stream(raw.split(","))
+ .map(String::trim)
+ .filter(name -> !name.isEmpty())
+ .collect(Collectors.toCollection(LinkedHashSet::new));
+ if (names.isEmpty()) {
+ throw new AnalysisException("'" + property + "' must list at least one name");
+ }
+ return names;
+ }
+
+ private static int countTablets(List<Partition> partitions) {
+ int count = 0;
+ for (Partition partition : partitions) {
+ for (MaterializedIndex index : partition.getMaterializedIndices(IndexExtState.VISIBLE)) {
+ count += index.getTablets().size();
+ }
+ }
+ return count;
+ }
+
+ // Every tablet costs index file metadata reads, so a wide scan must be narrowed explicitly.
+ private void checkTabletLimits(int tabletCount) {
+ SessionVariable sessionVariable = ConnectContext.get().getSessionVariable();
+ int limit = sessionVariable.indexDiskUsageMaxTablets;
+ if (tabletCount > limit) {
+ throw new AnalysisException(String.format("index_disk_usage covers %d tablets, exceeding "
+ + "index_disk_usage_max_tablets=%d; narrow partitions or indexes", tabletCount, limit));
+ }
+ if (!positionDetail) {
+ return;
+ }
+ int positionDetailLimit = sessionVariable.indexDiskUsagePositionDetailMaxTablets;
+ if (tabletCount > positionDetailLimit) {
+ throw new AnalysisException(String.format("position_detail covers %d tablets, exceeding "
+ + "index_disk_usage_position_detail_max_tablets=%d; narrow partitions or indexes",
+ tabletCount, positionDetailLimit));
+ }
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/TableValuedFunctionIf.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/TableValuedFunctionIf.java
index 67f32ba0651..69bc5c0aca9 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/TableValuedFunctionIf.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/TableValuedFunctionIf.java
@@ -69,6 +69,8 @@ public abstract class TableValuedFunctionIf {
return new MvInfosTableValuedFunction(params);
case PartitionsTableValuedFunction.NAME:
return new PartitionsTableValuedFunction(params);
+ case IndexDiskUsageTableValuedFunction.NAME:
+ return new IndexDiskUsageTableValuedFunction(params);
case JobsTableValuedFunction.NAME:
return new JobsTableValuedFunction(params);
case TasksTableValuedFunction.NAME:
diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/tvf/source/IndexDiskUsageScanNodeTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/tvf/source/IndexDiskUsageScanNodeTest.java
new file mode 100644
index 00000000000..a986005bcc1
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/tvf/source/IndexDiskUsageScanNodeTest.java
@@ -0,0 +1,341 @@
+// 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.analysis.TupleId;
+import org.apache.doris.catalog.Replica;
+import org.apache.doris.catalog.Tablet;
+import org.apache.doris.common.UserException;
+import org.apache.doris.planner.PlanNodeId;
+import org.apache.doris.planner.ScanContext;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.resource.Tag;
+import org.apache.doris.resource.computegroup.ComputeGroup;
+import org.apache.doris.system.Backend;
+import org.apache.doris.tablefunction.IndexDiskUsageTableValuedFunction;
+import org.apache.doris.tablefunction.IndexDiskUsageTableValuedFunction.TabletTarget;
+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 org.apache.doris.thrift.TScanRangeLocation;
+import org.apache.doris.thrift.TScanRangeLocations;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class IndexDiskUsageScanNodeTest {
+
+ @Test
+ public void testGroupsTabletsByBackend() throws Exception {
+ List<TabletTarget> targets = Arrays.asList(target(101, 10, 5), target(102, 10, 5), target(103, 11, 7));
+ Map<Long, List<TabletTarget>> groups = IndexDiskUsageScanNode.groupByBackend(targets,
+ t -> t.getTabletId() == 102 ? 2L : 1L);
+ Assertions.assertEquals(2, groups.size());
+ Assertions.assertEquals(Arrays.asList(101L, 103L), tabletIds(groups.get(1L)));
+ Assertions.assertEquals(Arrays.asList(102L), tabletIds(groups.get(2L)));
+ }
+
+ @Test
+ public void testSelectorFailureIsPropagated() {
+ List<TabletTarget> targets = Arrays.asList(target(101, 10, 5));
+ UserException e = Assertions.assertThrows(UserException.class,
+ () -> IndexDiskUsageScanNode.groupByBackend(targets, t -> {
+ throw new UserException("No queryable replica for tablet 101");
+ }));
+ Assertions.assertTrue(e.getMessage().contains("No queryable replica for tablet 101"), e.getMessage());
+ }
+
+ @Test
+ public void testSpreadsTabletsOverQueryableBackends() throws Exception {
+ Map<Long, Backend> backends = ImmutableMap.of(
+ 1L, backend(1L, false), 2L, backend(2L, true), 3L, backend(3L, true));
+ List<Replica> replicas = Arrays.asList(replica(3L), replica(1L), replica(2L));
+ // Backend 1 is not queryable, so tablets pick from [2, 3] by tablet id.
+ Assertions.assertEquals(2L, IndexDiskUsageScanNode.chooseBackend(100L, replicas, backends::get,
+ IndexDiskUsageScanNode.queryableIn(null)));
+ Assertions.assertEquals(3L, IndexDiskUsageScanNode.chooseBackend(101L, replicas, backends::get,
+ IndexDiskUsageScanNode.queryableIn(null)));
+ }
+
+ @Test
+ public void testNoQueryableReplicaFails() {
+ Map<Long, Backend> backends = ImmutableMap.of(1L, backend(1L, false));
+ UserException e = Assertions.assertThrows(UserException.class,
+ () -> IndexDiskUsageScanNode.chooseBackend(101L, Arrays.asList(replica(1L)), backends::get,
+ IndexDiskUsageScanNode.queryableIn(null)));
+ Assertions.assertTrue(e.getMessage().contains("No queryable replica for tablet 101"), e.getMessage());
+ }
+
+ @Test
+ public void testSkipsReplicasOutsideComputeGroup() throws Exception {
+ Map<Long, Backend> backends = ImmutableMap.of(
+ 1L, backend(1L, true, "zone_a", true), 2L, backend(2L, true, "zone_b", true));
+ List<Replica> replicas = Arrays.asList(replica(1L), replica(2L));
+ ComputeGroup zoneA = new ComputeGroup("zone_a", "zone_a", null);
+ // Tablet 101 would hash to backend 2 if the zone_b replica were a candidate.
+ Assertions.assertEquals(1L, IndexDiskUsageScanNode.chooseBackend(101L, replicas, backends::get,
+ IndexDiskUsageScanNode.queryableIn(zoneA)));
+ }
+
+ @Test
+ public void testSkipsComputeOnlyBackends() throws Exception {
+ Map<Long, Backend> backends = ImmutableMap.of(
+ 1L, backend(1L, true, Tag.VALUE_DEFAULT_TAG, true),
+ 2L, backend(2L, true, Tag.VALUE_DEFAULT_TAG, false));
+ List<Replica> replicas = Arrays.asList(replica(1L), replica(2L));
+ Assertions.assertEquals(1L, IndexDiskUsageScanNode.chooseBackend(101L, replicas, backends::get,
+ IndexDiskUsageScanNode.queryableIn(null)));
+ }
+
+ @Test
+ public void testInvalidComputeGroupHasNoCandidate() {
+ Map<Long, Backend> backends = ImmutableMap.of(1L, backend(1L, true));
+ UserException e = Assertions.assertThrows(UserException.class,
+ () -> IndexDiskUsageScanNode.chooseBackend(101L, Arrays.asList(replica(1L)), backends::get,
+ IndexDiskUsageScanNode.queryableIn(ComputeGroup.INVALID_COMPUTE_GROUP)));
+ Assertions.assertTrue(e.getMessage().contains("No queryable replica for tablet 101"), e.getMessage());
+ }
+
+ @Test
+ public void testTemplateTabletsAreNotCopiedPerBackend() {
+ CountingList<TIndexDiskUsageTablet> allTablets = new CountingList<>();
+ for (long tabletId = 101; tabletId <= 103; ++tabletId) {
+ allTablets.add(new TIndexDiskUsageTablet().setTabletId(tabletId).setPartitionId(10).setVersion(5));
+ }
+ TIndexDiskUsageMetadataParams params = new TIndexDiskUsageMetadataParams();
+ params.setLevel("tablet");
+ params.setTablets(allTablets);
+ TMetaScanRange template = new TMetaScanRange();
+ template.setMetadataType(TMetadataType.INDEX_DISK_USAGE);
+ template.setIndexDiskUsageParams(params);
+
+ Map<Long, List<TabletTarget>> groups = ImmutableMap.of(
+ 1L, Arrays.asList(target(101, 10, 5)),
+ 2L, Arrays.asList(target(102, 10, 5)),
+ 3L, Arrays.asList(target(103, 10, 5)));
+ Map<Long, Backend> backends = ImmutableMap.of(
+ 1L, backend(1L, true), 2L, backend(2L, true), 3L, backend(3L, true));
+
+ List<TScanRangeLocations> ranges =
+ IndexDiskUsageScanNode.buildScanRangeLocations(template, groups, backends::get, 1);
+ Assertions.assertEquals(3, ranges.size());
+ for (TScanRangeLocations range : ranges) {
+ long backendId = range.getLocations().get(0).getBackendId();
+ Assertions.assertEquals(tabletIds(groups.get(backendId)),
+ range.getScanRange().getMetaScanRange().getIndexDiskUsageParams().getTablets().stream()
+ .map(TIndexDiskUsageTablet::getTabletId).collect(Collectors.toList()));
+ }
+ // Planning must not copy every tablet once per backend.
+ Assertions.assertTrue(allTablets.iterations <= 1, "template tablets iterated " + allTablets.iterations);
+ Assertions.assertSame(allTablets, params.getTablets(), "the template must not be mutated");
+ }
+
+ @Test
+ public void testSplitsBackendGroupIntoBoundedRanges() {
+ TIndexDiskUsageMetadataParams params = new TIndexDiskUsageMetadataParams();
+ params.setLevel("tablet");
+ params.setPartitionNames(ImmutableMap.of(10L, "p1", 11L, "p2"));
+ TMetaScanRange template = new TMetaScanRange();
+ template.setMetadataType(TMetadataType.INDEX_DISK_USAGE);
+ template.setIndexDiskUsageParams(params);
+
+ Map<Long, List<TabletTarget>> groups = ImmutableMap.of(
+ 1L, Arrays.asList(target(101, 10, 5), target(102, 10, 5), target(103, 11, 7),
+ target(104, 11, 7), target(105, 11, 7)),
+ 2L, Arrays.asList(target(106, 10, 5)));
+ Map<Long, Backend> backends = ImmutableMap.of(1L, backend(1L, true), 2L, backend(2L, true));
+
+ // Two ranges per backend: backend 1 splits its five tablets, backend 2 keeps its single one.
+ List<TScanRangeLocations> ranges =
+ IndexDiskUsageScanNode.buildScanRangeLocations(template, groups, backends::get, 2);
+ Assertions.assertEquals(3, ranges.size());
+ Assertions.assertEquals(Arrays.asList(1L, 1L, 2L), ranges.stream()
+ .map(range -> range.getLocations().get(0).getBackendId()).collect(Collectors.toList()));
+ Assertions.assertEquals(Arrays.asList(101L, 102L, 103L), rangeTabletIds(ranges.get(0)));
+ Assertions.assertEquals(Arrays.asList(104L, 105L), rangeTabletIds(ranges.get(1)));
+ Assertions.assertEquals(Arrays.asList(106L), rangeTabletIds(ranges.get(2)));
+ // Each range only names the partitions of its own tablets.
+ Assertions.assertEquals(ImmutableMap.of(10L, "p1", 11L, "p2"), rangePartitionNames(ranges.get(0)));
+ Assertions.assertEquals(ImmutableMap.of(11L, "p2"), rangePartitionNames(ranges.get(1)));
+ Assertions.assertEquals(ImmutableMap.of(10L, "p1"), rangePartitionNames(ranges.get(2)));
+ }
+
+ @Test
+ public void testAlivePathsAreBuiltOncePerBackend() throws Exception {
+ Backend be1 = backend(1L, true);
+ Backend be2 = backend(2L, true);
+ Map<Long, Backend> backends = ImmutableMap.of(1L, be1, 2L, be2);
+ List<Replica> replicas = Arrays.asList(replica(1L), replica(2L));
+ IndexDiskUsageScanNode.BackendSelector selector =
+ IndexDiskUsageScanNode.localSelector(backends::get, IndexDiskUsageScanNode.queryableIn(null));
+ for (long tabletId = 101; tabletId <= 103; ++tabletId) {
+ Tablet tablet = Mockito.mock(Tablet.class);
+ Mockito.when(tablet.getId()).thenReturn(tabletId);
+ Mockito.when(tablet.getReplicas()).thenReturn(replicas);
+ Mockito.when(tablet.getQueryableReplicas(Mockito.anyLong(), Mockito.anyMap(), Mockito.eq(false)))
+ .thenReturn(replicas);
+ selector.select(new TabletTarget(tablet, 10, 1000L, 5));
+ }
+ // Every tablet shares the backends' disk snapshot instead of rebuilding it.
+ Mockito.verify(be1, Mockito.times(1)).getDisks();
+ Mockito.verify(be2, Mockito.times(1)).getDisks();
+ }
+
+ private static List<Long> rangeTabletIds(TScanRangeLocations range) {
+ return range.getScanRange().getMetaScanRange().getIndexDiskUsageParams().getTablets().stream()
+ .map(TIndexDiskUsageTablet::getTabletId).collect(Collectors.toList());
+ }
+
+ private static Map<Long, String> rangePartitionNames(TScanRangeLocations range) {
+ return range.getScanRange().getMetaScanRange().getIndexDiskUsageParams().getPartitionNames();
+ }
+
+ // Counts full traversals, which thrift deep copies perform once per copy.
+ private static class CountingList<T> extends ArrayList<T> {
+ private int iterations;
+
+ @Override
+ public Iterator<T> iterator() {
+ ++iterations;
+ return super.iterator();
+ }
+ }
+
+ @Test
+ public void testBuildsOneScanRangePerBackend() {
+ TIndexDiskUsageMetadataParams params = new TIndexDiskUsageMetadataParams();
+ params.setLevel("rowset");
+ params.setPositionDetail(true);
+ params.setIndexIds(Arrays.asList(1002L));
+ params.setPartitionNames(ImmutableMap.of(10L, "p1", 11L, "p2"));
+ TMetaScanRange template = new TMetaScanRange();
+ template.setMetadataType(TMetadataType.INDEX_DISK_USAGE);
+ template.setIndexDiskUsageParams(params);
+
+ Map<Long, List<TabletTarget>> groups = ImmutableMap.of(
+ 1L, Arrays.asList(target(101, 10, 5), target(103, 11, 7)),
+ 2L, Arrays.asList(target(102, 10, 5)));
+ Map<Long, Backend> backends = ImmutableMap.of(1L, backend(1L, true), 2L, backend(2L, true));
+
+ List<TScanRangeLocations> ranges =
+ IndexDiskUsageScanNode.buildScanRangeLocations(template, groups, backends::get, 1);
+ Assertions.assertEquals(2, ranges.size());
+ for (TScanRangeLocations range : ranges) {
+ Assertions.assertEquals(1, range.getLocationsSize());
+ TScanRangeLocation location = range.getLocations().get(0);
+ TIndexDiskUsageMetadataParams sub = range.getScanRange().getMetaScanRange().getIndexDiskUsageParams();
+ Assertions.assertEquals("rowset", sub.getLevel());
+ Assertions.assertTrue(sub.isPositionDetail());
+ Assertions.assertEquals(Arrays.asList(1002L), sub.getIndexIds());
+ Assertions.assertEquals("host" + location.getBackendId(), location.getServer().getHostname());
+ Assertions.assertEquals(9060, location.getServer().getPort());
+ List<Long> expected = tabletIds(groups.get(location.getBackendId()));
+ Assertions.assertEquals(expected,
+ sub.getTablets().stream().map(TIndexDiskUsageTablet::getTabletId).collect(Collectors.toList()));
+ }
+ Assertions.assertFalse(params.isSetTablets(), "the template must not be mutated");
+ }
+
+ private static TabletTarget target(long tabletId, long partitionId, long version) {
+ Tablet tablet = Mockito.mock(Tablet.class);
+ Mockito.when(tablet.getId()).thenReturn(tabletId);
+ return new TabletTarget(tablet, partitionId, 1000L, version);
+ }
+
+ private static List<Long> tabletIds(List<TabletTarget> targets) {
+ return targets.stream().map(TabletTarget::getTabletId).collect(Collectors.toList());
+ }
+
+ private static Replica replica(long backendId) {
+ Replica replica = Mockito.mock(Replica.class);
+ Mockito.when(replica.getBackendIdWithoutException()).thenReturn(backendId);
+ return replica;
+ }
+
+ @Test
+ public void testDoesNotRequireLoadAvailableBackends() throws Exception {
+ // OLAP reads may use a queryable backend whose load is disabled, so this scan must too.
+ Backend loadDisabled = backend(1L, true);
+ ComputeGroup group = Mockito.mock(ComputeGroup.class);
+ Mockito.when(group.getBackendList()).thenReturn(Arrays.asList(loadDisabled));
+ ConnectContext ctx = Mockito.mock(ConnectContext.class);
+ Mockito.when(ctx.getSessionVariable()).thenReturn(new SessionVariable());
+ Mockito.when(ctx.getComputeGroup()).thenReturn(group);
+ try (MockedStatic<ConnectContext> mockedContext = Mockito.mockStatic(ConnectContext.class)) {
+ mockedContext.when(ConnectContext::get).thenReturn(ctx);
+ IndexDiskUsageScanNode node = new IndexDiskUsageScanNode(new PlanNodeId(0),
+ new TupleDescriptor(new TupleId(0)), Mockito.mock(IndexDiskUsageTableValuedFunction.class),
+ ScanContext.EMPTY);
+ Assertions.assertDoesNotThrow(node::initBackendPolicy);
+ }
+ }
+
+ @Test
+ public void testCloudSelectorUsesQueryableBackend() throws Exception {
+ Backend queryable = backend(2L, true);
+ TabletTarget target = target(102L, 10L, 5L);
+ Replica replica = replica(2L);
+ Mockito.when(target.getTablet().getReplicas()).thenReturn(Arrays.asList(replica));
+ IndexDiskUsageScanNode.BackendSelector selector = IndexDiskUsageScanNode.cloudSelector(
+ Replica::getBackendIdWithoutException, id -> id == 2L ? queryable : null);
+ Assertions.assertEquals(2L, selector.select(target));
+ }
+
+ @Test
+ public void testCloudSelectorRejectsSmoothUpgradeSource() {
+ // The old backend kept as a smooth upgrade fallback cannot run this scan and would return no rows.
+ Backend source = backend(1L, true);
+ Mockito.when(source.isSmoothUpgradeSrc()).thenReturn(true);
+ TabletTarget target = target(101L, 10L, 5L);
+ Replica replica = replica(1L);
+ Mockito.when(target.getTablet().getReplicas()).thenReturn(Arrays.asList(replica));
+ IndexDiskUsageScanNode.BackendSelector selector = IndexDiskUsageScanNode.cloudSelector(
+ Replica::getBackendIdWithoutException, id -> id == 1L ? source : null);
+ UserException e = Assertions.assertThrows(UserException.class, () -> selector.select(target));
+ Assertions.assertTrue(e.getMessage().contains("smooth upgrade source"), e.getMessage());
+ }
+
+ private static Backend backend(long id, boolean queryAvailable) {
+ return backend(id, queryAvailable, Tag.VALUE_DEFAULT_TAG, true);
+ }
+
+ private static Backend backend(long id, boolean queryAvailable, String locationTag, boolean mixNode) {
+ Backend backend = Mockito.mock(Backend.class);
+ Mockito.when(backend.getId()).thenReturn(id);
+ Mockito.when(backend.getHost()).thenReturn("host" + id);
+ Mockito.when(backend.getBePort()).thenReturn(9060);
+ Mockito.when(backend.isQueryAvailable()).thenReturn(queryAvailable);
+ Mockito.when(backend.isMixNode()).thenReturn(mixNode);
+ Mockito.when(backend.getDisks()).thenReturn(ImmutableMap.of());
+ Mockito.when(backend.getLocationTag()).thenReturn(Tag.createNotCheck(Tag.TYPE_LOCATION, locationTag));
+ return backend;
+ }
+}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
index 9c4f041db66..5760ea5fb10 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
@@ -79,6 +79,23 @@ public class SessionVariablesTest extends TestWithFeService {
sessionVariable.getInsertVisibleTimeoutReturnModeEnum());
}
+ @Test
+ public void testForwardIndexDiskUsageLimits() {
+ // A forwarded index_disk_usage query is planned by the master, so the scan limits the
+ // client set must travel with the statement.
+ SessionVariable follower = new SessionVariable();
+ follower.indexDiskUsageMaxTablets = 7;
+ follower.indexDiskUsagePositionDetailMaxTablets = 3;
+ Map<String, String> vars = follower.getForwardVariables();
+ Assertions.assertEquals("7", vars.get(SessionVariable.INDEX_DISK_USAGE_MAX_TABLETS));
+ Assertions.assertEquals("3", vars.get(SessionVariable.INDEX_DISK_USAGE_POSITION_DETAIL_MAX_TABLETS));
+
+ SessionVariable master = new SessionVariable();
+ master.setForwardedSessionVariables(vars);
+ Assertions.assertEquals(7, master.indexDiskUsageMaxTablets);
+ Assertions.assertEquals(3, master.indexDiskUsagePositionDetailMaxTablets);
+ }
+
@Test
public void testForwardQueryCacheVariables() {
// A forwarded statement is planned by the master in a fresh
diff --git a/fe/fe-core/src/test/java/org/apache/doris/tablefunction/IndexDiskUsageTableValuedFunctionTest.java b/fe/fe-core/src/test/java/org/apache/doris/tablefunction/IndexDiskUsageTableValuedFunctionTest.java
new file mode 100644
index 00000000000..729c971a8bd
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/tablefunction/IndexDiskUsageTableValuedFunctionTest.java
@@ -0,0 +1,489 @@
+// 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.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.OlapTable;
+import org.apache.doris.catalog.Partition;
+import org.apache.doris.catalog.PrimitiveType;
+import org.apache.doris.catalog.Tablet;
+import org.apache.doris.catalog.View;
+import org.apache.doris.catalog.info.IndexType;
+import org.apache.doris.common.Config;
+import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.mysql.privilege.AccessControllerManager;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.SessionVariable;
+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.ImmutableMap;
+import com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.InOrder;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
+public class IndexDiskUsageTableValuedFunctionTest {
+ private static final long BASE_INDEX_ID = 1000L;
+
+ private MockedStatic<Env> mockedEnv;
+ private MockedStatic<ConnectContext> mockedContext;
+ private AccessControllerManager accessManager;
+ private Database db;
+ private SessionVariable sessionVariable;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ sessionVariable = new SessionVariable();
+ ConnectContext ctx = Mockito.mock(ConnectContext.class);
+ Mockito.when(ctx.getSessionVariable()).thenReturn(sessionVariable);
+ Mockito.when(ctx.getQualifiedUser()).thenReturn("u1");
+ Mockito.when(ctx.getRemoteIP()).thenReturn("127.0.0.1");
+
+ Env env = Mockito.mock(Env.class);
+ accessManager = Mockito.mock(AccessControllerManager.class);
+ Mockito.when(env.getAccessManager()).thenReturn(accessManager);
+ allowShow(true);
+
+ InternalCatalog catalog = Mockito.mock(InternalCatalog.class);
+ db = Mockito.mock(Database.class);
+ Mockito.when(catalog.getDbOrAnalysisException("db")).thenReturn(db);
+ OlapTable table = mockOlapTable();
+ Mockito.when(db.getTableOrAnalysisException("logs")).thenReturn(table);
+
+ mockedContext = Mockito.mockStatic(ConnectContext.class);
+ mockedContext.when(ConnectContext::get).thenReturn(ctx);
+ mockedEnv = Mockito.mockStatic(Env.class);
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+ mockedEnv.when(Env::getCurrentInternalCatalog).thenReturn(catalog);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ mockedEnv.close();
+ mockedContext.close();
+ }
+
+ @Test
+ public void testRejectsUnknownProperty() {
+ assertAnalysisError("'foo' is invalid property", params("foo", "bar"));
+ }
+
+ @Test
+ public void testRequiresDatabaseAndTable() {
+ assertAnalysisError("'database' and 'table' are required for index_disk_usage",
+ ImmutableMap.of("table", "logs"));
+ }
+
+ @Test
+ public void testRejectsUnsupportedLevel() {
+ assertAnalysisError("Unsupported level 'partition' for index_disk_usage, expected tablet, rowset or segment",
+ params("level", "partition"));
+ }
+
+ @Test
+ public void testRejectsInvalidPositionDetail() {
+ assertAnalysisError("Invalid position_detail 'abc', expected true or false",
+ params("position_detail", "abc"));
+ }
+
+ @Test
+ public void testRejectsNonOlapTable() throws Exception {
+ View view = Mockito.mock(View.class);
+ Mockito.when(db.getTableOrAnalysisException("v1")).thenReturn(view);
+ assertAnalysisError("index_disk_usage only supports OLAP table", params("table", "v1"));
+ }
+
+ @Test
+ public void testRejectsUnknownPartition() {
+ assertAnalysisError("Unknown partition 'p9' in table db.logs", params("partitions", "p1,p9"));
+ }
+
+ @Test
+ public void testRejectsIndexFilterWithoutNames() {
+ // An empty index list would reach BE as "all indexes" and turn a narrowed query into a full scan.
+ assertAnalysisError("'indexes' must list at least one name", params("indexes", " , "));
+ }
+
+ @Test
+ public void testRejectsPartitionFilterWithoutNames() {
+ assertAnalysisError("'partitions' must list at least one name", params("partitions", ","));
+ }
+
+ @Test
+ public void testRejectsUnknownIndex() {
+ assertAnalysisError("Unknown index 'idx9' in table db.logs", params("indexes", "idx9"));
+ }
+
+ @Test
+ public void testRejectsNonInvertedIndex() {
+ assertAnalysisError("Index 'idx_bf' in table db.logs is not an inverted or ANN index",
+ params("indexes", "idx_bf"));
+ }
+
+ @Test
+ public void testDeniesWithoutShowPrivilege() {
+ allowShow(false);
+ AnalysisException e = Assertions.assertThrows(AnalysisException.class,
+ () -> new IndexDiskUsageTableValuedFunction(params()));
+ Assertions.assertTrue(e.getMessage().contains("denied"), e.getMessage());
+ }
+
+ @Test
+ public void testSchema() {
+ List<Column> columns = new IndexDiskUsageTableValuedFunction(params()).getTableColumns();
+ Assertions.assertEquals(Arrays.asList("PARTITION_NAME", "MATERIALIZED_INDEX_NAME", "TABLET_ID",
+ "BACKEND_ID", "ROWSET_ID", "SEGMENT_ID", "INDEX_ID", "INDEX_NAME", "INDEX_TYPE", "COLUMN_NAME",
+ "INDEX_SUFFIX", "STRUCTURE", "STORAGE_FORMAT", "SEGMENT_COUNT", "ROW_COUNT", "TOTAL_BYTES",
+ "DICT_BYTES", "POSTING_BYTES", "POSITION_BYTES", "STATS_BYTES", "OTHER_BYTES", "STATS_SOURCE"),
+ columns.stream().map(Column::getName).collect(Collectors.toList()));
+ Assertions.assertEquals(Arrays.asList(PrimitiveType.VARCHAR, PrimitiveType.VARCHAR,
+ PrimitiveType.BIGINT, PrimitiveType.BIGINT, PrimitiveType.VARCHAR, PrimitiveType.INT,
+ PrimitiveType.BIGINT, PrimitiveType.VARCHAR, PrimitiveType.VARCHAR, PrimitiveType.VARCHAR,
+ PrimitiveType.VARCHAR, PrimitiveType.VARCHAR, PrimitiveType.VARCHAR, PrimitiveType.BIGINT,
+ PrimitiveType.BIGINT, PrimitiveType.BIGINT, PrimitiveType.BIGINT, PrimitiveType.BIGINT,
+ PrimitiveType.BIGINT, PrimitiveType.BIGINT, PrimitiveType.BIGINT, PrimitiveType.VARCHAR),
+ columns.stream().map(c -> c.getType().getPrimitiveType()).collect(Collectors.toList()));
+ Assertions.assertTrue(columns.stream().allMatch(Column::isAllowNull));
+ }
+
+ @Test
+ public void testMetaScanRangeDefaults() {
+ IndexDiskUsageTableValuedFunction tvf = new IndexDiskUsageTableValuedFunction(params());
+ Assertions.assertEquals(TMetadataType.INDEX_DISK_USAGE, tvf.getMetadataType());
+ TMetaScanRange range = tvf.getMetaScanRange(Lists.newArrayList());
+ Assertions.assertEquals(TMetadataType.INDEX_DISK_USAGE, range.getMetadataType());
+ TIndexDiskUsageMetadataParams p = range.getIndexDiskUsageParams();
+ Assertions.assertEquals("tablet", p.getLevel());
+ Assertions.assertFalse(p.isPositionDetail());
+ Assertions.assertTrue(p.getIndexIds().isEmpty());
+ Assertions.assertEquals(ImmutableMap.of(10L, "p1", 11L, "p2"), p.getPartitionNames());
+ Assertions.assertEquals(ImmutableMap.of(BASE_INDEX_ID, "logs"), p.getMaterializedIndexNames());
+ Assertions.assertEquals(Arrays.asList(tablet(101, 10, 5), tablet(102, 10, 5), tablet(103, 11, 7)),
+ p.getTablets());
+ }
+
+ @Test
+ public void testMetaScanRangeWithFilters() {
+ IndexDiskUsageTableValuedFunction tvf = new IndexDiskUsageTableValuedFunction(
+ params("partitions", "p2", "indexes", "idx_ts", "level", "SEGMENT", "position_detail", "true"));
+ TIndexDiskUsageMetadataParams p = tvf.getMetaScanRange(Lists.newArrayList()).getIndexDiskUsageParams();
+ Assertions.assertEquals("segment", p.getLevel());
+ Assertions.assertTrue(p.isPositionDetail());
+ Assertions.assertEquals(Arrays.asList(1002L), p.getIndexIds());
+ Assertions.assertEquals(ImmutableMap.of(11L, "p2"), p.getPartitionNames());
+ Assertions.assertEquals(Arrays.asList(tablet(103, 11, 7)), p.getTablets());
+ }
+
+ @Test
+ public void testTabletLimit() throws Exception {
+ // The default path reads index file metadata of every segment, so a wide scan must be narrowed.
+ sessionVariable.indexDiskUsageMaxTablets = 2;
+ assertAnalysisError("index_disk_usage covers 3 tablets, exceeding "
+ + "index_disk_usage_max_tablets=2; narrow partitions or indexes", params());
+ // The limit is checked before the tablets are copied and their versions are read.
+ OlapTable table = (OlapTable) db.getTableOrAnalysisException("logs");
+ Mockito.verify(table.getPartition("p1", false), Mockito.never()).getVisibleVersion();
+ }
+
+ @Test
+ public void testPositionDetailTabletLimit() {
+ sessionVariable.indexDiskUsagePositionDetailMaxTablets = 2;
+ assertAnalysisError("position_detail covers 3 tablets, exceeding "
+ + "index_disk_usage_position_detail_max_tablets=2; narrow partitions or indexes",
+ params("position_detail", "true"));
+ }
+
+ @Test
+ public void testTabletTargets() {
+ List<IndexDiskUsageTableValuedFunction.TabletTarget> targets =
+ new IndexDiskUsageTableValuedFunction(params("partitions", "p1")).getTabletTargets();
+ Assertions.assertEquals(2, targets.size());
+ Assertions.assertEquals(101L, targets.get(0).getTabletId());
+ Assertions.assertEquals(10L, targets.get(0).getPartitionId());
+ Assertions.assertEquals(5L, targets.get(0).getVersion());
+ Assertions.assertEquals(101L, targets.get(0).getTablet().getId());
+ Assertions.assertEquals(BASE_INDEX_ID, targets.get(0).getMaterializedIndexId());
+ Assertions.assertEquals(102L, targets.get(1).getTabletId());
+ }
+
+ @Test
+ public void testLocalVisibleVersionIsReadUnderTableLock() throws Exception {
+ OlapTable table = mockOlapTable();
+ Mockito.when(db.getTableOrAnalysisException("locked")).thenReturn(table);
+ new IndexDiskUsageTableValuedFunction(params("table", "locked"));
+ // Local replica choice filters replicas by this version, so it is read together with the
+ // tablets under the table lock.
+ Partition p1 = table.getPartition("p1", false);
+ InOrder inOrder = Mockito.inOrder(table, p1);
+ inOrder.verify(table).readLock();
+ inOrder.verify(p1).getVisibleVersion();
+ inOrder.verify(table).readUnlock();
+ }
+
+ @Test
+ public void testCollectsVisibleRollupTablets() throws Exception {
+ OlapTable table = mockOlapTable();
+ Partition p1 = table.getPartition("p1", false);
+ MaterializedIndex baseIndex = p1.getBaseIndex();
+ Tablet rollupTablet = Mockito.mock(Tablet.class);
+ Mockito.when(rollupTablet.getId()).thenReturn(201L);
+ MaterializedIndex rollup = Mockito.mock(MaterializedIndex.class);
+ Mockito.when(rollup.getId()).thenReturn(2000L);
+ Mockito.when(rollup.getTablets()).thenReturn(Arrays.asList(rollupTablet));
+ // A light ADD INDEX installs the index on rollups too, so their tablets hold index files.
+ Mockito.when(p1.getMaterializedIndices(MaterializedIndex.IndexExtState.VISIBLE))
+ .thenReturn(Arrays.asList(baseIndex, rollup));
+ Mockito.when(table.getIndexNameById(2000L)).thenReturn("r_msg");
+ Mockito.when(db.getTableOrAnalysisException("with_rollup")).thenReturn(table);
+
+ IndexDiskUsageTableValuedFunction tvf = new IndexDiskUsageTableValuedFunction(
+ params("table", "with_rollup", "partitions", "p1"));
+ List<IndexDiskUsageTableValuedFunction.TabletTarget> targets = tvf.getTabletTargets();
+ Assertions.assertEquals(Arrays.asList(101L, 102L, 201L), targets.stream()
+ .map(IndexDiskUsageTableValuedFunction.TabletTarget::getTabletId).collect(Collectors.toList()));
+ Assertions.assertEquals(2000L, targets.get(2).getMaterializedIndexId());
+ Assertions.assertEquals(5L, targets.get(2).getVersion());
+ Assertions.assertEquals(ImmutableMap.of(BASE_INDEX_ID, "logs", 2000L, "r_msg"),
+ tvf.getMetaScanRange(Lists.newArrayList()).getIndexDiskUsageParams().getMaterializedIndexNames());
+ }
+
+ @Test
+ public void testCloudSnapshotFetchesVersionsOnceWhenLayoutIsStable() throws Exception {
+ withCloudMode(() -> {
+ mockCloudTable();
+ AtomicInteger fetches = new AtomicInteger();
+ IndexDiskUsageTableValuedFunction tvf = new IndexDiskUsageTableValuedFunction(
+ params("table", "cloud_logs", "partitions", "p1"), partitions -> {
+ fetches.incrementAndGet();
+ return partitions.stream().map(partition -> 9L).collect(Collectors.toList());
+ });
+ Assertions.assertEquals(1, fetches.get());
+ Assertions.assertEquals(Arrays.asList(101L, 102L), tabletIds(tvf));
+ Assertions.assertEquals(9L, tvf.getTabletTargets().get(0).getVersion());
+ });
+ }
+
+ // A rollup dropped while the versions are read would be paired with a version its tablets never
+ // received, so the snapshot is taken again without it.
+ @Test
+ public void testCloudSnapshotIsTakenAgainWhenRollupIsDroppedDuringVersionFetch() throws Exception {
+ withCloudMode(() -> {
+ OlapTable table = mockCloudTable();
+ Partition p1 = table.getPartition("p1", false);
+ MaterializedIndex baseIndex = p1.getBaseIndex();
+ AtomicReference<List<MaterializedIndex>> visible =
+ new AtomicReference<>(Arrays.asList(baseIndex, mockRollup(2000L, 201L)));
+ Mockito.when(p1.getMaterializedIndices(MaterializedIndex.IndexExtState.VISIBLE))
+ .thenAnswer(invocation -> visible.get());
+ AtomicInteger fetches = new AtomicInteger();
+
+ IndexDiskUsageTableValuedFunction tvf = new IndexDiskUsageTableValuedFunction(
+ params("table", "cloud_logs", "partitions", "p1"), partitions -> {
+ if (fetches.getAndIncrement() == 0) {
+ visible.set(Arrays.asList(baseIndex));
+ }
+ return partitions.stream().map(partition -> 9L).collect(Collectors.toList());
+ });
+ Assertions.assertEquals(2, fetches.get());
+ Assertions.assertEquals(Arrays.asList(101L, 102L), tabletIds(tvf));
+ });
+ }
+
+ @Test
+ public void testCloudSnapshotFailsWhenRollupsKeepChanging() throws Exception {
+ withCloudMode(() -> {
+ OlapTable table = mockCloudTable();
+ Partition p1 = table.getPartition("p1", false);
+ AtomicReference<List<MaterializedIndex>> visible =
+ new AtomicReference<>(Arrays.asList(p1.getBaseIndex()));
+ Mockito.when(p1.getMaterializedIndices(MaterializedIndex.IndexExtState.VISIBLE))
+ .thenAnswer(invocation -> visible.get());
+ AtomicInteger fetches = new AtomicInteger();
+
+ AnalysisException e = Assertions.assertThrows(AnalysisException.class,
+ () -> new IndexDiskUsageTableValuedFunction(
+ params("table", "cloud_logs", "partitions", "p1"), partitions -> {
+ // Another rollup becomes visible during every fetch.
+ List<MaterializedIndex> next = Lists.newArrayList(visible.get());
+ int attempt = fetches.incrementAndGet();
+ next.add(mockRollup(3000L + attempt, 300L + attempt));
+ visible.set(next);
+ return partitions.stream().map(partition -> 9L).collect(Collectors.toList());
+ }));
+ Assertions.assertTrue(e.getMessage().contains("kept changing"), e.getMessage());
+ Assertions.assertEquals(3, fetches.get());
+ });
+ }
+
+ @Test
+ public void testCloudSnapshotReportsPartitionDroppedDuringVersionFetch() throws Exception {
+ withCloudMode(() -> {
+ OlapTable table = mockCloudTable();
+ Partition p1 = table.getPartition("p1", false);
+ AtomicBoolean dropped = new AtomicBoolean();
+ Mockito.when(table.getPartition("p1", false)).thenAnswer(invocation -> dropped.get() ? null : p1);
+ Mockito.when(table.getPartition(10L)).thenAnswer(invocation -> dropped.get() ? null : p1);
+
+ AnalysisException e = Assertions.assertThrows(AnalysisException.class,
+ () -> new IndexDiskUsageTableValuedFunction(
+ params("table", "cloud_logs", "partitions", "p1"), partitions -> {
+ dropped.set(true);
+ return partitions.stream().map(partition -> 9L).collect(Collectors.toList());
+ }));
+ Assertions.assertTrue(e.getMessage().contains("Unknown partition 'p1'"), e.getMessage());
+ });
+ }
+
+ @FunctionalInterface
+ private interface CloudScenario {
+ void run() throws Exception;
+ }
+
+ // Runs `scenario` with the FE in cloud mode, whose partition versions come from the injected source.
+ private static void withCloudMode(CloudScenario scenario) throws Exception {
+ String previous = Config.deploy_mode;
+ Config.deploy_mode = "cloud";
+ try {
+ scenario.run();
+ } finally {
+ Config.deploy_mode = previous;
+ }
+ }
+
+ // A table with the partitions of mockOlapTable() that the cloud tests can look up by id too.
+ private OlapTable mockCloudTable() throws Exception {
+ OlapTable table = mockOlapTable();
+ Partition p1 = table.getPartition("p1", false);
+ Partition p2 = table.getPartition("p2", false);
+ Mockito.when(table.getPartition(10L)).thenReturn(p1);
+ Mockito.when(table.getPartition(11L)).thenReturn(p2);
+ Mockito.when(db.getTableOrAnalysisException("cloud_logs")).thenReturn(table);
+ return table;
+ }
+
+ private static MaterializedIndex mockRollup(long id, long tabletId) {
+ Tablet tablet = Mockito.mock(Tablet.class);
+ Mockito.when(tablet.getId()).thenReturn(tabletId);
+ MaterializedIndex rollup = Mockito.mock(MaterializedIndex.class);
+ Mockito.when(rollup.getId()).thenReturn(id);
+ Mockito.when(rollup.getTablets()).thenReturn(Arrays.asList(tablet));
+ return rollup;
+ }
+
+ private static List<Long> tabletIds(IndexDiskUsageTableValuedFunction tvf) {
+ return tvf.getTabletTargets().stream()
+ .map(IndexDiskUsageTableValuedFunction.TabletTarget::getTabletId).collect(Collectors.toList());
+ }
+
+ private void allowShow(boolean allowed) {
+ Mockito.when(accessManager.checkTblPriv(Mockito.nullable(ConnectContext.class), Mockito.anyString(),
+ Mockito.anyString(), Mockito.anyString(), Mockito.any(PrivPredicate.class))).thenReturn(allowed);
+ }
+
+ private static void assertAnalysisError(String expected, Map<String, String> params) {
+ AnalysisException e = Assertions.assertThrows(AnalysisException.class,
+ () -> new IndexDiskUsageTableValuedFunction(params));
+ Assertions.assertTrue(e.getMessage().contains(expected), e.getMessage());
+ }
+
+ private static Map<String, String> params(String... overrides) {
+ Map<String, String> params = new HashMap<>();
+ params.put("database", "db");
+ params.put("table", "logs");
+ for (int i = 0; i + 1 < overrides.length; i += 2) {
+ params.put(overrides[i], overrides[i + 1]);
+ }
+ return params;
+ }
+
+ private static TIndexDiskUsageTablet tablet(long tabletId, long partitionId, long version) {
+ TIndexDiskUsageTablet tablet = new TIndexDiskUsageTablet();
+ tablet.setTabletId(tabletId);
+ tablet.setPartitionId(partitionId);
+ tablet.setMaterializedIndexId(BASE_INDEX_ID);
+ tablet.setVersion(version);
+ return tablet;
+ }
+
+ private static OlapTable mockOlapTable() {
+ OlapTable table = Mockito.mock(OlapTable.class);
+ Mockito.when(table.getName()).thenReturn("logs");
+ Mockito.when(table.getIndexNameById(BASE_INDEX_ID)).thenReturn("logs");
+ Partition p1 = mockPartition(10L, "p1", 5L, 101L, 102L);
+ Partition p2 = mockPartition(11L, "p2", 7L, 103L);
+ Mockito.when(table.getPartitions()).thenReturn(Arrays.asList(p1, p2));
+ Mockito.when(table.getPartition("p1", false)).thenReturn(p1);
+ Mockito.when(table.getPartition("p2", false)).thenReturn(p2);
+ List<Index> indexes = Arrays.asList(
+ mockIndex(1001L, "idx_msg", IndexType.INVERTED),
+ mockIndex(1002L, "idx_ts", IndexType.INVERTED),
+ mockIndex(1003L, "idx_bf", IndexType.NGRAM_BF));
+ Mockito.when(table.getIndexes()).thenReturn(indexes);
+ return table;
+ }
+
+ private static Partition mockPartition(long id, String name, long version, Long... tabletIds) {
+ Partition partition = Mockito.mock(Partition.class);
+ Mockito.when(partition.getId()).thenReturn(id);
+ Mockito.when(partition.getName()).thenReturn(name);
+ Mockito.when(partition.getVisibleVersion()).thenReturn(version);
+ List<Tablet> tablets = Lists.newArrayList();
+ for (Long tabletId : tabletIds) {
+ Tablet tablet = Mockito.mock(Tablet.class);
+ Mockito.when(tablet.getId()).thenReturn(tabletId);
+ tablets.add(tablet);
+ }
+ MaterializedIndex baseIndex = Mockito.mock(MaterializedIndex.class);
+ Mockito.when(baseIndex.getId()).thenReturn(BASE_INDEX_ID);
+ Mockito.when(baseIndex.getTablets()).thenReturn(tablets);
+ Mockito.when(partition.getBaseIndex()).thenReturn(baseIndex);
+ Mockito.when(partition.getMaterializedIndices(MaterializedIndex.IndexExtState.VISIBLE))
+ .thenReturn(Arrays.asList(baseIndex));
+ return partition;
+ }
+
+ private static Index mockIndex(long id, String name, IndexType type) {
+ Index index = Mockito.mock(Index.class);
+ Mockito.when(index.getIndexId()).thenReturn(id);
+ Mockito.when(index.getIndexName()).thenReturn(name);
+ Mockito.when(index.getIndexType()).thenReturn(type);
+ return index;
+ }
+}
diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift
index 35b8b69b691..8a84190caef 100644
--- a/gensrc/thrift/PlanNodes.thrift
+++ b/gensrc/thrift/PlanNodes.thrift
@@ -839,6 +839,29 @@ struct TLanceIndexMetadataParams {
3: optional string table
}
+// A tablet to inspect and the partition visible version to read.
+struct TIndexDiskUsageTablet {
+ 1: optional i64 tablet_id
+ 2: optional i64 partition_id
+ 3: optional i64 version
+ // The base or rollup index that the tablet belongs to.
+ 4: optional i64 materialized_index_id
+}
+
+// Parameters of the index_disk_usage table function.
+struct TIndexDiskUsageMetadataParams {
+ // "tablet", "rowset" or "segment".
+ 1: optional string level
+ // Scan SNII dictionary regions to split positions out of postings.
+ 2: optional bool position_detail
+ // Empty means all indexes.
+ 3: optional list<i64> index_ids
+ 4: optional map<i64, string> partition_names
+ 5: optional list<TIndexDiskUsageTablet> tablets
+ // Names of the base and rollup indexes, keyed by materialized index id.
+ 6: optional map<i64, string> materialized_index_names
+}
+
struct TMetaScanRange {
1: optional Types.TMetadataType metadata_type
2: optional TIcebergMetadataParams iceberg_params // deprecated
@@ -860,6 +883,7 @@ struct TMetaScanRange {
16: optional list<string> serialized_splits;
17: optional TParquetMetadataParams parquet_params;
18: optional TLanceIndexMetadataParams lance_index_params;
+ 19: optional TIndexDiskUsageMetadataParams index_disk_usage_params;
}
// Specification of an individual data range which is held in its entirety
diff --git a/gensrc/thrift/Types.thrift b/gensrc/thrift/Types.thrift
index 98528aa6f22..e18c66ec891 100644
--- a/gensrc/thrift/Types.thrift
+++ b/gensrc/thrift/Types.thrift
@@ -671,6 +671,7 @@ enum TMetadataType {
// Also assigned on branch-4.1 for Lance physical index entries inspection.
// Keep the value aligned across maintained branches. Do not renumber.
LANCE_INDEX_ENTRIES = 15,
+ INDEX_DISK_USAGE = 16,
}
// deprecated
diff --git a/regression-test/data/external_table_p0/tvf/test_index_disk_usage_tvf.out b/regression-test/data/external_table_p0/tvf/test_index_disk_usage_tvf.out
new file mode 100644
index 00000000000..a3bc3712a50
--- /dev/null
+++ b/regression-test/data/external_table_p0/tvf/test_index_disk_usage_tvf.out
@@ -0,0 +1,71 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !desc --
+BACKEND_ID bigint Yes false \N NONE
+COLUMN_NAME varchar(65533) Yes false \N NONE
+DICT_BYTES bigint Yes false \N NONE
+INDEX_ID bigint Yes false \N NONE
+INDEX_NAME varchar(65533) Yes false \N NONE
+INDEX_SUFFIX varchar(65533) Yes false \N NONE
+INDEX_TYPE varchar(65533) Yes false \N NONE
+MATERIALIZED_INDEX_NAME varchar(65533) Yes false \N NONE
+OTHER_BYTES bigint Yes false \N NONE
+PARTITION_NAME varchar(65533) Yes false \N NONE
+POSITION_BYTES bigint Yes false \N NONE
+POSTING_BYTES bigint Yes false \N NONE
+ROWSET_ID varchar(65533) Yes false \N NONE
+ROW_COUNT bigint Yes false \N NONE
+SEGMENT_COUNT bigint Yes false \N NONE
+SEGMENT_ID int Yes false \N NONE
+STATS_BYTES bigint Yes false \N NONE
+STATS_SOURCE varchar(65533) Yes false \N NONE
+STORAGE_FORMAT varchar(65533) Yes false \N NONE
+STRUCTURE varchar(65533) Yes false \N NONE
+TABLET_ID bigint Yes false \N NONE
+TOTAL_BYTES bigint Yes false \N NONE
+
+-- !components --
+index_disk_usage_v2 \N \N \N CONTAINER V2 FILE true \N \N true \N true true
+index_disk_usage_v2 idx_msg msg INVERTED TERM V2 FILE true true true false true false true
+index_disk_usage_v2 idx_num num INVERTED BKD V2 FILE true \N \N true \N true true
+
+-- !reconcile --
+index_disk_usage_v2 2 0
+
+-- !levels --
+index_disk_usage_v2 0 0 0 2
+
+-- !components --
+index_disk_usage_v3 \N \N \N CONTAINER V3 FILE true \N \N true \N true true
+index_disk_usage_v3 idx_msg msg INVERTED TERM V3 FILE true true true false true false true
+index_disk_usage_v3 idx_num num INVERTED BKD V3 FILE true \N \N true \N true true
+
+-- !reconcile --
+index_disk_usage_v3 2 0
+
+-- !levels --
+index_disk_usage_v3 0 0 0 2
+
+-- !components --
+index_disk_usage_snii \N \N \N CONTAINER SNII FILE true \N \N true \N true true
+index_disk_usage_snii idx_msg msg INVERTED TERM SNII FILE true true true true \N false true
+index_disk_usage_snii idx_num num INVERTED BKD SNII FILE true \N \N true \N true true
+
+-- !reconcile --
+index_disk_usage_snii 2 0
+
+-- !levels --
+index_disk_usage_snii 0 0 0 2
+
+-- !position_detail --
+true true true
+
+-- !index_filter --
+idx_num BKD
+
+-- !rollup --
+index_disk_usage_rollup idx_msg msg true
+r_msg idx_msg msg true
+
+-- !variant --
+idx_v v true
+
diff --git a/regression-test/suites/external_table_p0/tvf/test_index_disk_usage_tvf.groovy b/regression-test/suites/external_table_p0/tvf/test_index_disk_usage_tvf.groovy
new file mode 100644
index 00000000000..093582b44cd
--- /dev/null
+++ b/regression-test/suites/external_table_p0/tvf/test_index_disk_usage_tvf.groovy
@@ -0,0 +1,187 @@
+// 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.
+
+suite("test_index_disk_usage_tvf", "p0,external") {
+ String dbName = context.config.getDbNameByFile(context.file)
+
+ def createTable = { String tableName, String format ->
+ sql "DROP TABLE IF EXISTS ${tableName}"
+ sql """
+ CREATE TABLE ${tableName} (
+ k INT NOT NULL,
+ msg VARCHAR(256),
+ num INT,
+ INDEX idx_msg (msg) USING INVERTED PROPERTIES("parser" = "english", "support_phrase" = "true"),
+ INDEX idx_num (num) USING INVERTED
+ )
+ DUPLICATE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "disable_auto_compaction" = "true",
+ "inverted_index_storage_format" = "${format}"
+ )
+ """
+ // Two loads give two rowsets; 2000 rows keep common SNII terms out of inline postings.
+ sql """
+ INSERT INTO ${tableName}
+ SELECT number, concat('hello world doc', number), number % 97 FROM numbers("number" = "2000")
+ """
+ sql """
+ INSERT INTO ${tableName}
+ SELECT number + 2000, concat('quick brown fox doc', number), number % 89 FROM numbers("number" = "2000")
+ """
+ }
+
+ createTable("index_disk_usage_v2", "V2")
+ createTable("index_disk_usage_v3", "V3")
+ createTable("index_disk_usage_snii", "SNII")
+
+ order_qt_desc """ DESC FUNCTION index_disk_usage("database" = "${dbName}", "table" = "index_disk_usage_v2") """
+
+ for (String tableName : ["index_disk_usage_v2", "index_disk_usage_v3", "index_disk_usage_snii"]) {
+ // Component presence per index and structure; byte values vary across versions.
+ order_qt_components """
+ SELECT '${tableName}', index_name, column_name, index_type, structure, storage_format, stats_source,
+ SUM(total_bytes) > 0,
+ SUM(dict_bytes) > 0,
+ SUM(posting_bytes) > 0,
+ SUM(position_bytes) IS NULL,
+ SUM(position_bytes) > 0,
+ SUM(stats_bytes) IS NULL,
+ SUM(row_count) = 4000
+ FROM index_disk_usage("database" = "${dbName}", "table" = "${tableName}")
+ GROUP BY index_name, column_name, index_type, structure, storage_format, stats_source
+ """
+
+ // Every rowset's index bytes add up to the index size recorded in its rowset meta; the
+ // matched rowset count guards against an empty join.
+ order_qt_reconcile """
+ SELECT '${tableName}', COUNT(*), SUM(CASE WHEN usage_bytes <> meta_bytes THEN 1 ELSE 0 END)
+ FROM (
+ SELECT u.rowset_id, SUM(u.total_bytes) AS usage_bytes, MAX(r.INDEX_DISK_SIZE) AS meta_bytes
+ FROM index_disk_usage("database" = "${dbName}", "table" = "${tableName}", "level" = "rowset") u
+ JOIN information_schema.rowsets r ON u.rowset_id = r.ROWSET_ID
+ GROUP BY u.rowset_id
+ ) t
+ """
+
+ order_qt_levels """
+ SELECT '${tableName}',
+ (SELECT COUNT(*) FROM index_disk_usage("database" = "${dbName}", "table" = "${tableName}")
+ WHERE rowset_id IS NOT NULL OR segment_id IS NOT NULL),
+ (SELECT COUNT(*) FROM index_disk_usage("database" = "${dbName}", "table" = "${tableName}", "level" = "rowset")
+ WHERE rowset_id IS NULL OR segment_id IS NOT NULL),
+ (SELECT COUNT(*) FROM index_disk_usage("database" = "${dbName}", "table" = "${tableName}", "level" = "segment")
+ WHERE rowset_id IS NULL OR segment_id IS NULL),
+ (SELECT COUNT(DISTINCT rowset_id) FROM index_disk_usage("database" = "${dbName}", "table" = "${tableName}", "level" = "rowset"))
+ """
+ }
+
+ // position_detail splits SNII positions out of postings without changing the total.
+ order_qt_position_detail """
+ SELECT d.positions > 0, c.postings = d.postings + d.positions, c.total = d.total
+ FROM (
+ SELECT SUM(posting_bytes) AS postings, SUM(total_bytes) AS total
+ FROM index_disk_usage("database" = "${dbName}", "table" = "index_disk_usage_snii", "indexes" = "idx_msg")
+ ) c, (
+ SELECT SUM(posting_bytes) AS postings, SUM(position_bytes) AS positions, SUM(total_bytes) AS total
+ FROM index_disk_usage("database" = "${dbName}", "table" = "index_disk_usage_snii", "indexes" = "idx_msg",
+ "position_detail" = "true")
+ ) d
+ """
+
+ order_qt_index_filter """
+ SELECT DISTINCT index_name, structure
+ FROM index_disk_usage("database" = "${dbName}", "table" = "index_disk_usage_v3", "indexes" = "idx_num")
+ """
+
+ // A light ADD INDEX installs the index on rollups too, so later loads write index files on the
+ // rollup tablets.
+ sql "DROP TABLE IF EXISTS index_disk_usage_rollup"
+ sql """
+ CREATE TABLE index_disk_usage_rollup (
+ k INT NOT NULL,
+ v INT,
+ msg VARCHAR(256)
+ )
+ DUPLICATE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES ("replication_num" = "1", "disable_auto_compaction" = "true")
+ """
+ sql "ALTER TABLE index_disk_usage_rollup ADD ROLLUP r_msg(msg, k)"
+ assertEquals("FINISHED", getAlterRollupFinalState("index_disk_usage_rollup"))
+ sql """ CREATE INDEX idx_msg ON index_disk_usage_rollup(msg) USING INVERTED PROPERTIES("parser" = "english") """
+ sql """
+ INSERT INTO index_disk_usage_rollup
+ SELECT number, number % 7, concat('hello world doc', number) FROM numbers("number" = "2000")
+ """
+ order_qt_rollup """
+ SELECT materialized_index_name, index_name, column_name, SUM(total_bytes) > 0
+ FROM index_disk_usage("database" = "${dbName}", "table" = "index_disk_usage_rollup")
+ WHERE structure <> 'CONTAINER'
+ GROUP BY materialized_index_name, index_name, column_name
+ """
+
+ // Extracted VARIANT paths keep the parent index id under a path suffix, and their rows still
+ // carry the parent index and column names.
+ sql """ set default_variant_enable_typed_paths_to_sparse = false """
+ sql """ set default_variant_enable_doc_mode = false """
+ sql "DROP TABLE IF EXISTS index_disk_usage_variant"
+ sql """
+ CREATE TABLE index_disk_usage_variant (
+ k INT NOT NULL,
+ v VARIANT<'msg' : STRING>,
+ INDEX idx_v (v) USING INVERTED PROPERTIES("parser" = "english")
+ )
+ DUPLICATE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES ("replication_num" = "1", "disable_auto_compaction" = "true")
+ """
+ sql """
+ INSERT INTO index_disk_usage_variant
+ SELECT number, parse_to_variant(concat('{"msg": "hello world doc', number, '"}')) FROM numbers("number" = "1000")
+ """
+ order_qt_variant """
+ SELECT DISTINCT index_name, column_name, index_suffix <> ''
+ FROM index_disk_usage("database" = "${dbName}", "table" = "index_disk_usage_variant")
+ WHERE structure = 'TERM'
+ """
+
+ test {
+ sql """ SELECT * FROM index_disk_usage("database" = "${dbName}", "table" = "index_disk_usage_v2", "foo" = "bar") """
+ exception "'foo' is invalid property"
+ }
+ test {
+ sql """ SELECT * FROM index_disk_usage("database" = "${dbName}", "table" = "index_disk_usage_v2", "level" = "partition") """
+ exception "Unsupported level 'partition'"
+ }
+ test {
+ sql """ SELECT * FROM index_disk_usage("database" = "${dbName}", "table" = "index_disk_usage_v2", "partitions" = "p_missing") """
+ exception "Unknown partition 'p_missing'"
+ }
+ test {
+ sql """ SELECT * FROM index_disk_usage("database" = "${dbName}", "table" = "index_disk_usage_v2", "indexes" = "idx_missing") """
+ exception "Unknown index 'idx_missing'"
+ }
+ sql "DROP VIEW IF EXISTS index_disk_usage_view"
+ sql "CREATE VIEW index_disk_usage_view AS SELECT k FROM index_disk_usage_v2"
+ test {
+ sql """ SELECT * FROM index_disk_usage("database" = "${dbName}", "table" = "index_disk_usage_view") """
+ exception "index_disk_usage only supports OLAP table"
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org