github-actions[bot] commented on code in PR #66881: URL: https://github.com/apache/doris/pull/66881#discussion_r3802001830
########## regression-test/suites/unique_seq_map_p0/test_seq_map_candidate_key_scan.groovy: ########## @@ -0,0 +1,311 @@ +// 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. + +import java.util.regex.Pattern + +suite("test_seq_map_candidate_key_scan") { + def tableName = "test_seq_map_candidate_key_scan" + def compositeTableName = "test_seq_map_candidate_key_scan_composite" + def multiTabletTableName = "test_seq_map_candidate_key_scan_multi_tablet" + def costFallbackTableName = "test_seq_map_candidate_key_scan_cost_fallback" + + sql "DROP TABLE IF EXISTS ${tableName}" + sql "DROP TABLE IF EXISTS ${compositeTableName}" + sql "DROP TABLE IF EXISTS ${multiTabletTableName}" + sql "DROP TABLE IF EXISTS ${costFallbackTableName}" + try { + sql """ + CREATE TABLE ${tableName} ( + `id` BIGINT NOT NULL, + `c` INT NULL, + `d` INT NULL, + `e` INT NULL, + `s1` BIGINT NULL, + `s2` BIGINT NULL, + INDEX idx_c (`c`) USING INVERTED, + INDEX idx_d (`d`) USING INVERTED, + INDEX idx_e (`e`) USING INVERTED + ) ENGINE=OLAP + UNIQUE KEY(`id`) + DISTRIBUTED BY HASH(`id`) BUCKETS 1 + PROPERTIES ( + "replication_num" = "1", + "enable_unique_key_merge_on_write" = "false", + "light_schema_change" = "true", + "disable_auto_compaction" = "true", + "inverted_index_storage_format" = "V3", + "sequence_mapping.s1" = "c,d", + "sequence_mapping.s2" = "e" + ) + """ + + // Group s1 and group s2 deliberately arrive in different physical rows. + sql "INSERT INTO ${tableName}(id, c, d, s1) VALUES (1, 20, 200, 20)" + sql "INSERT INTO ${tableName}(id, e, s2) VALUES (1, 300, 30)" + + // id=2 has a stale physical row matching c=20. Its latest s1 value is c=99, + // so candidate collection may include it but the final residual must remove it. + sql "INSERT INTO ${tableName}(id, c, d, s1) VALUES (2, 20, 200, 10)" + sql "INSERT INTO ${tableName}(id, c, d, s1) VALUES (2, 99, 999, 20)" + sql "INSERT INTO ${tableName}(id, e, s2) VALUES (2, 300, 30)" + + sql "INSERT INTO ${tableName}(id, c, d, s1) VALUES (3, 20, 201, 20)" + sql "INSERT INTO ${tableName}(id, e, s2) VALUES (3, 300, 30)" + + def query = "SELECT id FROM ${tableName} WHERE c = 20 AND e = 300 ORDER BY id" + + sql "SET enable_seq_map_candidate_key_scan = false" + assertEquals([[1L], [3L]], sql(query)) Review Comment: The repository's regression-test contract requires deterministic SQL results to be captured with named `qt`/`order_qt` checks and an auto-generated `.out`, but this suite verifies all stable row sets with `assertEquals` and therefore has no reviewable expected-result artifact. Please keep direct assertions only for dynamic profile metadata, move the deterministic query outputs to QT checks, and generate the corresponding `.out` with the regression runner. ########## be/src/exec/scan/olap_scanner.cpp: ########## @@ -163,6 +167,330 @@ static bool has_file_cache_statistics(const io::FileCacheStatistics& stats) { stats.inverted_index_serial_read_rounds != 0; } +std::vector<RowSetSplits> OlapScanner::_clone_rowset_splits() const { + std::vector<RowSetSplits> cloned; + cloned.reserve(_tablet_reader_params.rs_splits.size()); + for (const auto& split : _tablet_reader_params.rs_splits) { + RowSetSplits copy(split.rs_reader->clone()); + copy.segment_offsets = split.segment_offsets; + copy.segment_row_ranges = split.segment_row_ranges; + cloned.emplace_back(std::move(copy)); + } + return cloned; +} + +std::string OlapScanner::_encode_candidate_key(const OlapTuple& key) const { + std::string encoded; + for (size_t i = 0; i < key.size(); ++i) { + const auto& field = key.get_field(i); + const auto type = static_cast<int32_t>(field.get_type()); + encoded.append(reinterpret_cast<const char*>(&type), sizeof(type)); + if (field.is_null()) { + continue; + } + const auto value = field.as_string_view(); + const auto size = static_cast<uint64_t>(value.size()); + encoded.append(reinterpret_cast<const char*>(&size), sizeof(size)); + encoded.append(value); + } + return encoded; +} + +bool OlapScanner::CandidateScanCostLimit::exceeded(int64_t candidate_scan_rows, + size_t candidate_key_count) const { + if (!enabled || full_scan_rows <= 0 || rowset_count == 0 || candidate_scan_rows < 0) { + return false; + } + if (candidate_scan_rows >= full_scan_rows) { + return true; + } + + // A final point-key scan may probe every captured rowset for each candidate key. + // Avoid multiplication overflow by comparing against the remaining row budget. + const auto remaining_rows = static_cast<uint64_t>(full_scan_rows - candidate_scan_rows); + return candidate_key_count > (remaining_rows - 1) / rowset_count; +} + +Status OlapScanner::_collect_seq_map_candidate_keys( + const std::vector<std::shared_ptr<ColumnPredicate>>& driver_predicates, + const std::vector<std::shared_ptr<ColumnPredicate>>& key_predicates, + int64_t max_candidate_keys, const CandidateScanCostLimit& cost_limit, + CandidateKeyMap* candidate_keys, bool* limit_exceeded, bool* cost_exceeded) { + DCHECK(candidate_keys != nullptr); + DCHECK(limit_exceeded != nullptr); + DCHECK(cost_exceeded != nullptr); + *limit_exceeded = false; + *cost_exceeded = false; + candidate_keys->clear(); + + auto candidate_params = _tablet_reader_params; + candidate_params.rs_splits = _clone_rowset_splits(); + candidate_params.predicates.clear(); + for (const auto& predicate : key_predicates) { + candidate_params.predicates.emplace_back(predicate->clone(predicate->column_id())); + } + for (const auto& predicate : driver_predicates) { + candidate_params.predicates.emplace_back(predicate->clone(predicate->column_id())); + } + candidate_params.function_filters.clear(); + candidate_params.all_access_paths.clear(); + candidate_params.predicate_access_paths.clear(); + candidate_params.output_columns.clear(); + candidate_params.extra_columns.clear(); + candidate_params.common_expr_ctxs_push_down.clear(); + candidate_params.topn_filter_source_node_ids.clear(); + candidate_params.key_group_cluster_key_idxes.clear(); + candidate_params.virtual_column_exprs.clear(); + candidate_params.score_runtime.reset(); + candidate_params.collection_statistics.reset(); + candidate_params.ann_topn_runtime.reset(); + candidate_params.direct_mode = true; + candidate_params.aggregation = false; + candidate_params.is_seq_map_candidate_scan = true; + candidate_params.seq_map_candidate_pruned = false; + candidate_params.push_down_agg_type_opt = TPushAggOp::NONE; + candidate_params.read_orderby_key = false; + candidate_params.read_orderby_key_reverse = false; + candidate_params.read_orderby_key_num_prefix_columns = 0; + candidate_params.read_orderby_key_limit = 0; + candidate_params.condition_cache_digest = 0; + candidate_params.general_read_limit = -1; + candidate_params.read_row_binlog = false; + candidate_params.binlog_scan_type = TBinlogScanType::NONE; + candidate_params.start_tso.reset(); + candidate_params.end_tso.reset(); + candidate_params.tso_predicate_column_id.reset(); + + std::vector<ColumnId> candidate_columns; + candidate_columns.reserve(_tablet_reader_params.tablet_schema->num_key_columns() + + driver_predicates.size()); + for (uint32_t cid = 0; cid < _tablet_reader_params.tablet_schema->num_key_columns(); ++cid) { + candidate_columns.push_back(cid); + } + for (const auto& predicate : driver_predicates) { + if (std::find(candidate_columns.begin(), candidate_columns.end(), predicate->column_id()) == + candidate_columns.end()) { + candidate_columns.push_back(predicate->column_id()); + } + } + candidate_params.return_columns = candidate_columns; + candidate_params.origin_return_columns = &candidate_columns; + candidate_params.tablet_columns_convert_to_null_set = nullptr; + + BlockReader candidate_reader; + candidate_reader.set_batch_size(_state->batch_size()); + candidate_reader.set_preferred_block_size_bytes(_state->preferred_block_size_bytes()); + Defer account_candidate_stats {[&]() { + const auto& candidate_stats = candidate_reader.stats(); + auto* total_stats = _tablet_reader->mutable_stats(); + total_stats->seq_map_candidate_scan_rows += candidate_stats.raw_rows_read; + total_stats->seq_map_candidate_scan_bytes += candidate_stats.uncompressed_bytes_read; + total_stats->seq_map_candidate_index_filtered_rows += + candidate_stats.rows_inverted_index_filtered; + total_stats->seq_map_candidate_index_downgrades += + candidate_stats.inverted_index_downgrade_count; + total_stats->seq_map_candidate_index_lookup_ns += Review Comment: [P2] Merge candidate cache activity into the standard totals. These two phase-specific byte counters do not update `total_stats->file_cache_stats`, yet the standard FileCacheProfileReporter, resource-context local/remote attribution, and `bytes_write_into_cache` accounting later read only that full struct. A candidate-only remote scan that prunes the tablet can therefore contribute to total scan bytes while its remote/peer reads and cache writes are absent or misattributed. Please also `merge_from(candidate_stats.file_cache_stats)` here (keeping the dedicated counters if useful), and cover a remote/cache-writing candidate path. ########## regression-test/suites/unique_seq_map_p0/test_seq_map_candidate_key_scan.groovy: ########## @@ -0,0 +1,311 @@ +// 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. + +import java.util.regex.Pattern + +suite("test_seq_map_candidate_key_scan") { + def tableName = "test_seq_map_candidate_key_scan" + def compositeTableName = "test_seq_map_candidate_key_scan_composite" + def multiTabletTableName = "test_seq_map_candidate_key_scan_multi_tablet" + def costFallbackTableName = "test_seq_map_candidate_key_scan_cost_fallback" + + sql "DROP TABLE IF EXISTS ${tableName}" + sql "DROP TABLE IF EXISTS ${compositeTableName}" + sql "DROP TABLE IF EXISTS ${multiTabletTableName}" + sql "DROP TABLE IF EXISTS ${costFallbackTableName}" + try { + sql """ + CREATE TABLE ${tableName} ( + `id` BIGINT NOT NULL, + `c` INT NULL, + `d` INT NULL, + `e` INT NULL, + `s1` BIGINT NULL, + `s2` BIGINT NULL, + INDEX idx_c (`c`) USING INVERTED, + INDEX idx_d (`d`) USING INVERTED, + INDEX idx_e (`e`) USING INVERTED + ) ENGINE=OLAP + UNIQUE KEY(`id`) + DISTRIBUTED BY HASH(`id`) BUCKETS 1 + PROPERTIES ( + "replication_num" = "1", + "enable_unique_key_merge_on_write" = "false", + "light_schema_change" = "true", + "disable_auto_compaction" = "true", + "inverted_index_storage_format" = "V3", + "sequence_mapping.s1" = "c,d", + "sequence_mapping.s2" = "e" + ) + """ + + // Group s1 and group s2 deliberately arrive in different physical rows. + sql "INSERT INTO ${tableName}(id, c, d, s1) VALUES (1, 20, 200, 20)" + sql "INSERT INTO ${tableName}(id, e, s2) VALUES (1, 300, 30)" + + // id=2 has a stale physical row matching c=20. Its latest s1 value is c=99, + // so candidate collection may include it but the final residual must remove it. + sql "INSERT INTO ${tableName}(id, c, d, s1) VALUES (2, 20, 200, 10)" + sql "INSERT INTO ${tableName}(id, c, d, s1) VALUES (2, 99, 999, 20)" + sql "INSERT INTO ${tableName}(id, e, s2) VALUES (2, 300, 30)" + + sql "INSERT INTO ${tableName}(id, c, d, s1) VALUES (3, 20, 201, 20)" + sql "INSERT INTO ${tableName}(id, e, s2) VALUES (3, 300, 30)" + + def query = "SELECT id FROM ${tableName} WHERE c = 20 AND e = 300 ORDER BY id" + + sql "SET enable_seq_map_candidate_key_scan = false" + assertEquals([[1L], [3L]], sql(query)) + + sql "SET enable_seq_map_candidate_key_scan = true" + sql "SET enable_inverted_index_query = true" + sql "SET enable_profile = true" + sql "SET pipeline_task_profile_threshold_ms = 0" + + def counterValue = { String profileString, String counterName -> + def matcher = Pattern.compile("${counterName}:\\s*(\\d+)").matcher(profileString) + assertTrue(matcher.find(), "${counterName} is absent from profile") + return Long.parseLong(matcher.group(1)) + } + + def runWithProfile = { String tag, String statement, Closure profileCheck -> + def queryResult = null + def queryId = "${tag}_${System.currentTimeMillis()}" + profile(queryId) { + run { + queryResult = sql "/* ${queryId} */ ${statement}" + } + check { profileString, exception -> + if (exception != null) { + throw exception + } + profileCheck.call(profileString) + } + } + return queryResult + } + + assertEquals([[1L], [3L]], runWithProfile( + "seq_map_candidate_two_groups", query) { profileString -> + assertEquals(2L, counterValue(profileString, "SeqMapCandidateDriverGroups")) + assertEquals(2L, counterValue(profileString, "SeqMapCandidateDriverPredicates")) + assertEquals(3L, counterValue(profileString, "SeqMapCandidateKeysAfterIntersect")) + assertTrue(counterValue(profileString, "SeqMapCandidateScanRows") > 0) + assertEquals(0L, counterValue(profileString, "SeqMapCandidateIndexDowngrades")) + }) + + // Same-group predicates must be evaluated on the same physical group row. + def sameGroupQuery = + "SELECT id FROM ${tableName} WHERE c = 20 AND d = 200 ORDER BY id" + assertEquals([[1L]], runWithProfile( + "seq_map_candidate_same_group", sameGroupQuery) { profileString -> + assertEquals(1L, counterValue(profileString, "SeqMapCandidateDriverGroups")) + assertEquals(2L, counterValue(profileString, "SeqMapCandidateDriverPredicates")) + assertEquals(2L, counterValue(profileString, "SeqMapCandidateKeysAfterIntersect")) + }) + + // Empty candidates can short-circuit this tablet. + assertEquals([], runWithProfile( + "seq_map_candidate_empty", "SELECT id FROM ${tableName} WHERE c = 777") { + profileString -> + assertEquals(0L, counterValue(profileString, "SeqMapCandidateKeysAfterIntersect")) + assertEquals(1L, counterValue(profileString, "SeqMapCandidatePrunedTablets")) + assertEquals(0L, counterValue(profileString, "SeqMapCandidateFallbacks")) + }) + + // Force fallback and prove that the candidate-key limit branch was taken. + sql "SET seq_map_candidate_key_max_count = 1" + try { + def fallbackQuery = + "SELECT id FROM ${tableName} WHERE c IN (20, 99) ORDER BY id" + assertEquals([[1L], [2L], [3L]], runWithProfile( + "seq_map_candidate_limit", fallbackQuery) { profileString -> + assertTrue(counterValue(profileString, "SeqMapCandidateFallbacks") > 0) + assertTrue(profileString.contains("candidate_key_limit"), + "candidate_key_limit fallback reason is absent from profile") + }) + } finally { + sql "SET seq_map_candidate_key_max_count = 100000" + } + + // Candidate point keys must remain a subset of the original FE key range. + def keyRangeQuery = """ + SELECT id FROM ${tableName} + WHERE id BETWEEN 2 AND 3 AND c = 20 AND e = 300 + ORDER BY id + """ + assertEquals([[3L]], runWithProfile( + "seq_map_candidate_key_range", keyRangeQuery) { profileString -> + assertEquals(2L, counterValue(profileString, "SeqMapCandidateDriverGroups")) + assertEquals(2L, counterValue(profileString, "SeqMapCandidateKeysAfterIntersect")) + assertEquals(0L, counterValue(profileString, "SeqMapCandidateFallbacks")) + }) + + def keyInQuery = """ + SELECT id FROM ${tableName} + WHERE id IN (1, 3) AND c = 20 AND e = 300 + ORDER BY id + """ + assertEquals([[1L], [3L]], runWithProfile( + "seq_map_candidate_key_in", keyInQuery) { profileString -> + assertEquals(2L, counterValue(profileString, "SeqMapCandidateKeysAfterIntersect")) + assertEquals(0L, counterValue(profileString, "SeqMapCandidateFallbacks")) + }) + + sql "SET enable_inverted_index_query = false" + try { + assertEquals([[1L], [3L]], runWithProfile( + "seq_map_candidate_index_disabled", query) { profileString -> + assertTrue(counterValue(profileString, "SeqMapCandidateFallbacks") > 0) + assertTrue(profileString.contains("inverted_index_query_disabled"), + "inverted_index_query_disabled fallback reason is absent from profile") + }) + } finally { + sql "SET enable_inverted_index_query = true" + } + + // Range predicates are intentionally residual-only in the first version. + assertEquals([[2L]], runWithProfile( + "seq_map_candidate_no_driver", + "SELECT id FROM ${tableName} WHERE c > 20 ORDER BY id") { profileString -> + assertTrue(counterValue(profileString, "SeqMapCandidateFallbacks") > 0) + assertTrue(profileString.contains("no_indexed_positive_driver"), + "no_indexed_positive_driver fallback reason is absent from profile") + }) + + test { + sql "SET seq_map_candidate_key_max_count = 0" + exception "seq_map_candidate_key_max_count should be greater than 0" + } + test { + sql "SET seq_map_candidate_key_max_count = -1" + exception "seq_map_candidate_key_max_count should be greater than 0" + } + + sql """ + CREATE TABLE ${costFallbackTableName} ( + `id` BIGINT NOT NULL, + `c` INT NULL, + `s1` BIGINT NULL, + INDEX idx_c (`c`) USING INVERTED + ) ENGINE=OLAP + UNIQUE KEY(`id`) + DISTRIBUTED BY HASH(`id`) BUCKETS 1 + PROPERTIES ( + "replication_num" = "1", + "enable_unique_key_merge_on_write" = "false", + "light_schema_change" = "true", + "disable_auto_compaction" = "true", + "inverted_index_storage_format" = "V3", + "sequence_mapping.s1" = "c" + ) + """ + def broadCandidateValues = + (1..5000).collect { id -> "(${id}, 20, 1)" }.join(",") + sql """ + INSERT INTO ${costFallbackTableName}(id, c, s1) + VALUES ${broadCandidateValues} + """ + def costFallbackQuery = + "SELECT COUNT(*) FROM ${costFallbackTableName} WHERE c = 20" + assertEquals([[5000L]], runWithProfile( + "seq_map_candidate_cost_fallback", costFallbackQuery) { profileString -> + assertTrue(counterValue(profileString, "SeqMapCandidateFallbacks") > 0) + assertTrue(counterValue(profileString, "SeqMapCandidateScanRows") > 0) + assertTrue(profileString.contains("candidate_cost_limit"), + "candidate_cost_limit fallback reason is absent from profile") + }) + + sql """ + CREATE TABLE ${compositeTableName} ( + `k1` BIGINT NOT NULL, + `k2` VARCHAR(32) NOT NULL, + `c` INT NULL, + `s1` BIGINT NULL, + INDEX idx_c (`c`) USING INVERTED + ) ENGINE=OLAP + UNIQUE KEY(`k1`, `k2`) + DISTRIBUTED BY HASH(`k1`) BUCKETS 1 + PROPERTIES ( + "replication_num" = "1", + "enable_unique_key_merge_on_write" = "false", + "light_schema_change" = "true", + "disable_auto_compaction" = "true", + "inverted_index_storage_format" = "V3", + "sequence_mapping.s1" = "c" + ) + """ + sql """ + INSERT INTO ${compositeTableName}(k1, k2, c, s1) + VALUES (1, 'alpha|beta', 20, 10) + """ + sql """ + INSERT INTO ${compositeTableName}(k1, k2, c, s1) + VALUES (2, 'trailing space ', 20, 10) + """ + sql """ + INSERT INTO ${compositeTableName}(k1, k2, c, s1) + VALUES (2, 'trailing space ', 99, 20) + """ + def compositeQuery = + "SELECT k1 FROM ${compositeTableName} WHERE c = 20 ORDER BY k1" + assertEquals([[1L]], runWithProfile( + "seq_map_candidate_composite_key", compositeQuery) { profileString -> + assertEquals(1L, counterValue(profileString, "SeqMapCandidateDriverGroups")) + assertEquals(2L, counterValue(profileString, "SeqMapCandidateKeysAfterIntersect")) + assertEquals(0L, counterValue(profileString, "SeqMapCandidateFallbacks")) + }) + + sql """ + CREATE TABLE ${multiTabletTableName} ( + `id` BIGINT NOT NULL, + `c` INT NULL, + `s1` BIGINT NULL, + INDEX idx_c (`c`) USING INVERTED + ) ENGINE=OLAP + UNIQUE KEY(`id`) + DISTRIBUTED BY HASH(`id`) BUCKETS 2 + PROPERTIES ( + "replication_num" = "1", + "enable_unique_key_merge_on_write" = "false", + "light_schema_change" = "true", + "disable_auto_compaction" = "true", + "inverted_index_storage_format" = "V3", + "sequence_mapping.s1" = "c" + ) + """ + def nonMatchingValues = + (1..20).collect { id -> "(${id}, 99, 1)" }.join(",") + sql "INSERT INTO ${multiTabletTableName}(id, c, s1) VALUES ${nonMatchingValues}" + sql "INSERT INTO ${multiTabletTableName}(id, c, s1) VALUES (100, 20, 1)" + def multiTabletQuery = + "SELECT id FROM ${multiTabletTableName} WHERE c = 20 ORDER BY id" + assertEquals([[100L]], runWithProfile( + "seq_map_candidate_multi_tablet", multiTabletQuery) { profileString -> + assertTrue(counterValue(profileString, "SeqMapCandidatePrunedTablets") > 0) + assertEquals(0L, counterValue(profileString, "SeqMapCandidateFallbacks")) + }) + + sql "SET enable_profile = false" + } finally { + sql "SET enable_profile = false" + sql "DROP TABLE IF EXISTS ${tableName}" Review Comment: These drops run from `finally`, so a failed assertion still deletes the exact tables needed for post-failure inspection. The regression-test rules require dropping tables before use and retaining them afterward; the suite already performs the four pre-test drops. Please keep the session-variable cleanup here but remove the post-test table drops. ########## be/src/exec/scan/olap_scanner.cpp: ########## @@ -163,6 +167,330 @@ static bool has_file_cache_statistics(const io::FileCacheStatistics& stats) { stats.inverted_index_serial_read_rounds != 0; } +std::vector<RowSetSplits> OlapScanner::_clone_rowset_splits() const { + std::vector<RowSetSplits> cloned; + cloned.reserve(_tablet_reader_params.rs_splits.size()); + for (const auto& split : _tablet_reader_params.rs_splits) { + RowSetSplits copy(split.rs_reader->clone()); + copy.segment_offsets = split.segment_offsets; + copy.segment_row_ranges = split.segment_row_ranges; + cloned.emplace_back(std::move(copy)); + } + return cloned; +} + +std::string OlapScanner::_encode_candidate_key(const OlapTuple& key) const { + std::string encoded; + for (size_t i = 0; i < key.size(); ++i) { + const auto& field = key.get_field(i); + const auto type = static_cast<int32_t>(field.get_type()); + encoded.append(reinterpret_cast<const char*>(&type), sizeof(type)); + if (field.is_null()) { + continue; + } + const auto value = field.as_string_view(); + const auto size = static_cast<uint64_t>(value.size()); + encoded.append(reinterpret_cast<const char*>(&size), sizeof(size)); + encoded.append(value); + } + return encoded; +} + +bool OlapScanner::CandidateScanCostLimit::exceeded(int64_t candidate_scan_rows, + size_t candidate_key_count) const { + if (!enabled || full_scan_rows <= 0 || rowset_count == 0 || candidate_scan_rows < 0) { + return false; + } + if (candidate_scan_rows >= full_scan_rows) { + return true; + } + + // A final point-key scan may probe every captured rowset for each candidate key. + // Avoid multiplication overflow by comparing against the remaining row budget. + const auto remaining_rows = static_cast<uint64_t>(full_scan_rows - candidate_scan_rows); + return candidate_key_count > (remaining_rows - 1) / rowset_count; +} + +Status OlapScanner::_collect_seq_map_candidate_keys( + const std::vector<std::shared_ptr<ColumnPredicate>>& driver_predicates, + const std::vector<std::shared_ptr<ColumnPredicate>>& key_predicates, + int64_t max_candidate_keys, const CandidateScanCostLimit& cost_limit, + CandidateKeyMap* candidate_keys, bool* limit_exceeded, bool* cost_exceeded) { + DCHECK(candidate_keys != nullptr); + DCHECK(limit_exceeded != nullptr); + DCHECK(cost_exceeded != nullptr); + *limit_exceeded = false; + *cost_exceeded = false; + candidate_keys->clear(); + + auto candidate_params = _tablet_reader_params; + candidate_params.rs_splits = _clone_rowset_splits(); + candidate_params.predicates.clear(); + for (const auto& predicate : key_predicates) { + candidate_params.predicates.emplace_back(predicate->clone(predicate->column_id())); + } + for (const auto& predicate : driver_predicates) { + candidate_params.predicates.emplace_back(predicate->clone(predicate->column_id())); + } + candidate_params.function_filters.clear(); + candidate_params.all_access_paths.clear(); + candidate_params.predicate_access_paths.clear(); + candidate_params.output_columns.clear(); + candidate_params.extra_columns.clear(); + candidate_params.common_expr_ctxs_push_down.clear(); + candidate_params.topn_filter_source_node_ids.clear(); + candidate_params.key_group_cluster_key_idxes.clear(); + candidate_params.virtual_column_exprs.clear(); + candidate_params.score_runtime.reset(); + candidate_params.collection_statistics.reset(); + candidate_params.ann_topn_runtime.reset(); + candidate_params.direct_mode = true; + candidate_params.aggregation = false; + candidate_params.is_seq_map_candidate_scan = true; + candidate_params.seq_map_candidate_pruned = false; + candidate_params.push_down_agg_type_opt = TPushAggOp::NONE; + candidate_params.read_orderby_key = false; + candidate_params.read_orderby_key_reverse = false; + candidate_params.read_orderby_key_num_prefix_columns = 0; + candidate_params.read_orderby_key_limit = 0; + candidate_params.condition_cache_digest = 0; + candidate_params.general_read_limit = -1; + candidate_params.read_row_binlog = false; + candidate_params.binlog_scan_type = TBinlogScanType::NONE; + candidate_params.start_tso.reset(); + candidate_params.end_tso.reset(); + candidate_params.tso_predicate_column_id.reset(); + + std::vector<ColumnId> candidate_columns; + candidate_columns.reserve(_tablet_reader_params.tablet_schema->num_key_columns() + + driver_predicates.size()); + for (uint32_t cid = 0; cid < _tablet_reader_params.tablet_schema->num_key_columns(); ++cid) { + candidate_columns.push_back(cid); + } + for (const auto& predicate : driver_predicates) { + if (std::find(candidate_columns.begin(), candidate_columns.end(), predicate->column_id()) == + candidate_columns.end()) { + candidate_columns.push_back(predicate->column_id()); + } + } + candidate_params.return_columns = candidate_columns; + candidate_params.origin_return_columns = &candidate_columns; + candidate_params.tablet_columns_convert_to_null_set = nullptr; + + BlockReader candidate_reader; + candidate_reader.set_batch_size(_state->batch_size()); + candidate_reader.set_preferred_block_size_bytes(_state->preferred_block_size_bytes()); + Defer account_candidate_stats {[&]() { + const auto& candidate_stats = candidate_reader.stats(); + auto* total_stats = _tablet_reader->mutable_stats(); + total_stats->seq_map_candidate_scan_rows += candidate_stats.raw_rows_read; + total_stats->seq_map_candidate_scan_bytes += candidate_stats.uncompressed_bytes_read; + total_stats->seq_map_candidate_index_filtered_rows += + candidate_stats.rows_inverted_index_filtered; + total_stats->seq_map_candidate_index_downgrades += + candidate_stats.inverted_index_downgrade_count; + total_stats->seq_map_candidate_index_lookup_ns += + candidate_stats.inverted_index_lookup_timer; + total_stats->seq_map_candidate_cache_local_bytes += + candidate_stats.file_cache_stats.bytes_read_from_local; + total_stats->seq_map_candidate_cache_remote_bytes += + candidate_stats.file_cache_stats.bytes_read_from_remote; + + // Keep the aggregate scan totals consistent with the physical candidate IO. + total_stats->io_ns += candidate_stats.io_ns; + total_stats->compressed_bytes_read += candidate_stats.compressed_bytes_read; + total_stats->decompress_ns += candidate_stats.decompress_ns; + total_stats->uncompressed_bytes_read += candidate_stats.uncompressed_bytes_read; + total_stats->bytes_read += candidate_stats.bytes_read; + total_stats->raw_rows_read += candidate_stats.raw_rows_read; + }}; + RETURN_IF_ERROR(candidate_reader.init(candidate_params)); + + Block block = candidate_params.tablet_schema->create_block(candidate_columns); + const size_t key_column_count = candidate_params.tablet_schema->num_key_columns(); + bool eof = false; + while (!eof) { + RETURN_IF_ERROR(candidate_reader.next_block_with_aggregation(&block, &eof)); + _tablet_reader->mutable_stats()->seq_map_candidate_rows += block.rows(); + for (size_t row = 0; row < block.rows(); ++row) { + OlapTuple key; + for (size_t col = 0; col < key_column_count; ++col) { + Field field; + block.get_by_position(col).column->get(row, field); + key.add_field(std::move(field)); + } + candidate_keys->try_emplace(_encode_candidate_key(key), std::move(key)); + if (candidate_keys->size() > static_cast<size_t>(max_candidate_keys)) { + *limit_exceeded = true; + break; + } + } + block.clear_column_data(); + if (*limit_exceeded) { + break; + } + if (cost_limit.exceeded(candidate_reader.stats().raw_rows_read, candidate_keys->size())) { + *cost_exceeded = true; + break; + } + } + return Status::OK(); +} + +Status OlapScanner::_prepare_seq_map_candidate_keys() { + const auto& query_options = _state->query_options(); + auto& params = _tablet_reader_params; + auto& schema = params.tablet_schema; + if (!query_options.enable_seq_map_candidate_key_scan || schema == nullptr || + !schema->has_seq_map() || schema->keys_type() != KeysType::UNIQUE_KEYS || + params.tablet->enable_unique_key_merge_on_write() || params.direct_mode) { + return Status::OK(); + } + + auto* stats = _tablet_reader->mutable_stats(); + SCOPED_RAW_TIMER(&stats->seq_map_candidate_build_ns); + for (const auto& split : params.rs_splits) { + if (split.segment_offsets != std::pair<int64_t, int64_t> {0, 0} || + !split.segment_row_ranges.empty()) { + ++stats->seq_map_candidate_fallbacks; + _seq_map_candidate_fallback_reason = "partial_scanner_split"; + return Status::OK(); + } + } + const int64_t max_candidate_keys = query_options.seq_map_candidate_key_max_count; + if (max_candidate_keys <= 0) { + ++stats->seq_map_candidate_fallbacks; + _seq_map_candidate_fallback_reason = "invalid_candidate_limit"; + return Status::OK(); + } + if (!query_options.enable_inverted_index_query) { + ++stats->seq_map_candidate_fallbacks; + _seq_map_candidate_fallback_reason = "inverted_index_query_disabled"; + return Status::OK(); + } + + std::vector<std::shared_ptr<ColumnPredicate>> key_predicates; + std::map<uint32_t, std::vector<std::shared_ptr<ColumnPredicate>>> group_drivers; + const auto& value_to_seq = schema->value_col_idx_to_seq_col_idx(); + for (const auto& predicate : params.predicates) { + const auto cid = predicate->column_id(); + const auto& column = schema->column(cid); + if (column.is_key()) { + // Key predicates constrain the candidate reader, but do not identify a value group. + key_predicates.push_back(predicate); + continue; + } + const auto seq_it = value_to_seq.find(cid); + if (seq_it == value_to_seq.end()) { + continue; + } + const auto type = predicate->type(); + const bool positive_driver = type == PredicateType::EQ || type == PredicateType::IN_LIST; + if (!positive_driver || schema->inverted_indexs(column).empty()) { + continue; + } + group_drivers[seq_it->second].push_back(predicate); + ++stats->seq_map_candidate_driver_predicates; + } + + if (group_drivers.empty()) { + ++stats->seq_map_candidate_fallbacks; + _seq_map_candidate_fallback_reason = "no_indexed_positive_driver"; + return Status::OK(); + } + stats->seq_map_candidate_driver_groups = group_drivers.size(); + Review Comment: [P1] Price the actual per-segment point-range work. The guard uses `candidate_key_count * rs_splits.size()`, but every rowset creates an iterator per segment and every SegmentIterator replays the complete point list. Worse, `_get_row_ranges_by_keys()` incrementally calls `RowRanges::ranges_union` for each point, rescanning/copying the accumulated sparse ranges, so 100k candidates can become quadratic work even in one segment while this estimate still permits the optimization; that loop also has no cancellation check. Please bulk-build/sort point ranges (or add a batched point-key path), add cancellation, and base the cutoff on the actual segment/range-build cost rather than rowset count. ########## regression-test/suites/unique_seq_map_p0/test_seq_map_candidate_key_scan.groovy: ########## @@ -0,0 +1,311 @@ +// 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. + +import java.util.regex.Pattern + +suite("test_seq_map_candidate_key_scan") { + def tableName = "test_seq_map_candidate_key_scan" + def compositeTableName = "test_seq_map_candidate_key_scan_composite" + def multiTabletTableName = "test_seq_map_candidate_key_scan_multi_tablet" + def costFallbackTableName = "test_seq_map_candidate_key_scan_cost_fallback" + + sql "DROP TABLE IF EXISTS ${tableName}" + sql "DROP TABLE IF EXISTS ${compositeTableName}" + sql "DROP TABLE IF EXISTS ${multiTabletTableName}" + sql "DROP TABLE IF EXISTS ${costFallbackTableName}" + try { + sql """ + CREATE TABLE ${tableName} ( + `id` BIGINT NOT NULL, + `c` INT NULL, + `d` INT NULL, + `e` INT NULL, + `s1` BIGINT NULL, + `s2` BIGINT NULL, + INDEX idx_c (`c`) USING INVERTED, + INDEX idx_d (`d`) USING INVERTED, + INDEX idx_e (`e`) USING INVERTED + ) ENGINE=OLAP + UNIQUE KEY(`id`) + DISTRIBUTED BY HASH(`id`) BUCKETS 1 + PROPERTIES ( + "replication_num" = "1", + "enable_unique_key_merge_on_write" = "false", + "light_schema_change" = "true", + "disable_auto_compaction" = "true", + "inverted_index_storage_format" = "V3", + "sequence_mapping.s1" = "c,d", + "sequence_mapping.s2" = "e" + ) + """ + + // Group s1 and group s2 deliberately arrive in different physical rows. + sql "INSERT INTO ${tableName}(id, c, d, s1) VALUES (1, 20, 200, 20)" + sql "INSERT INTO ${tableName}(id, e, s2) VALUES (1, 300, 30)" + + // id=2 has a stale physical row matching c=20. Its latest s1 value is c=99, + // so candidate collection may include it but the final residual must remove it. + sql "INSERT INTO ${tableName}(id, c, d, s1) VALUES (2, 20, 200, 10)" + sql "INSERT INTO ${tableName}(id, c, d, s1) VALUES (2, 99, 999, 20)" + sql "INSERT INTO ${tableName}(id, e, s2) VALUES (2, 300, 30)" + + sql "INSERT INTO ${tableName}(id, c, d, s1) VALUES (3, 20, 201, 20)" + sql "INSERT INTO ${tableName}(id, e, s2) VALUES (3, 300, 30)" + + def query = "SELECT id FROM ${tableName} WHERE c = 20 AND e = 300 ORDER BY id" + + sql "SET enable_seq_map_candidate_key_scan = false" + assertEquals([[1L], [3L]], sql(query)) + + sql "SET enable_seq_map_candidate_key_scan = true" + sql "SET enable_inverted_index_query = true" + sql "SET enable_profile = true" + sql "SET pipeline_task_profile_threshold_ms = 0" + + def counterValue = { String profileString, String counterName -> + def matcher = Pattern.compile("${counterName}:\\s*(\\d+)").matcher(profileString) + assertTrue(matcher.find(), "${counterName} is absent from profile") + return Long.parseLong(matcher.group(1)) + } + + def runWithProfile = { String tag, String statement, Closure profileCheck -> + def queryResult = null + def queryId = "${tag}_${System.currentTimeMillis()}" + profile(queryId) { + run { + queryResult = sql "/* ${queryId} */ ${statement}" + } + check { profileString, exception -> + if (exception != null) { + throw exception + } + profileCheck.call(profileString) + } + } + return queryResult + } + + assertEquals([[1L], [3L]], runWithProfile( + "seq_map_candidate_two_groups", query) { profileString -> + assertEquals(2L, counterValue(profileString, "SeqMapCandidateDriverGroups")) + assertEquals(2L, counterValue(profileString, "SeqMapCandidateDriverPredicates")) + assertEquals(3L, counterValue(profileString, "SeqMapCandidateKeysAfterIntersect")) + assertTrue(counterValue(profileString, "SeqMapCandidateScanRows") > 0) + assertEquals(0L, counterValue(profileString, "SeqMapCandidateIndexDowngrades")) + }) + + // Same-group predicates must be evaluated on the same physical group row. + def sameGroupQuery = + "SELECT id FROM ${tableName} WHERE c = 20 AND d = 200 ORDER BY id" + assertEquals([[1L]], runWithProfile( + "seq_map_candidate_same_group", sameGroupQuery) { profileString -> + assertEquals(1L, counterValue(profileString, "SeqMapCandidateDriverGroups")) + assertEquals(2L, counterValue(profileString, "SeqMapCandidateDriverPredicates")) + assertEquals(2L, counterValue(profileString, "SeqMapCandidateKeysAfterIntersect")) + }) + + // Empty candidates can short-circuit this tablet. + assertEquals([], runWithProfile( + "seq_map_candidate_empty", "SELECT id FROM ${tableName} WHERE c = 777") { + profileString -> + assertEquals(0L, counterValue(profileString, "SeqMapCandidateKeysAfterIntersect")) + assertEquals(1L, counterValue(profileString, "SeqMapCandidatePrunedTablets")) + assertEquals(0L, counterValue(profileString, "SeqMapCandidateFallbacks")) + }) + + // Force fallback and prove that the candidate-key limit branch was taken. + sql "SET seq_map_candidate_key_max_count = 1" + try { + def fallbackQuery = + "SELECT id FROM ${tableName} WHERE c IN (20, 99) ORDER BY id" + assertEquals([[1L], [2L], [3L]], runWithProfile( + "seq_map_candidate_limit", fallbackQuery) { profileString -> + assertTrue(counterValue(profileString, "SeqMapCandidateFallbacks") > 0) + assertTrue(profileString.contains("candidate_key_limit"), + "candidate_key_limit fallback reason is absent from profile") + }) + } finally { + sql "SET seq_map_candidate_key_max_count = 100000" + } + + // Candidate point keys must remain a subset of the original FE key range. + def keyRangeQuery = """ + SELECT id FROM ${tableName} + WHERE id BETWEEN 2 AND 3 AND c = 20 AND e = 300 + ORDER BY id + """ + assertEquals([[3L]], runWithProfile( + "seq_map_candidate_key_range", keyRangeQuery) { profileString -> + assertEquals(2L, counterValue(profileString, "SeqMapCandidateDriverGroups")) + assertEquals(2L, counterValue(profileString, "SeqMapCandidateKeysAfterIntersect")) + assertEquals(0L, counterValue(profileString, "SeqMapCandidateFallbacks")) + }) + + def keyInQuery = """ + SELECT id FROM ${tableName} + WHERE id IN (1, 3) AND c = 20 AND e = 300 + ORDER BY id + """ + assertEquals([[1L], [3L]], runWithProfile( + "seq_map_candidate_key_in", keyInQuery) { profileString -> + assertEquals(2L, counterValue(profileString, "SeqMapCandidateKeysAfterIntersect")) + assertEquals(0L, counterValue(profileString, "SeqMapCandidateFallbacks")) + }) + + sql "SET enable_inverted_index_query = false" + try { + assertEquals([[1L], [3L]], runWithProfile( + "seq_map_candidate_index_disabled", query) { profileString -> + assertTrue(counterValue(profileString, "SeqMapCandidateFallbacks") > 0) + assertTrue(profileString.contains("inverted_index_query_disabled"), + "inverted_index_query_disabled fallback reason is absent from profile") + }) + } finally { + sql "SET enable_inverted_index_query = true" + } + + // Range predicates are intentionally residual-only in the first version. + assertEquals([[2L]], runWithProfile( + "seq_map_candidate_no_driver", + "SELECT id FROM ${tableName} WHERE c > 20 ORDER BY id") { profileString -> + assertTrue(counterValue(profileString, "SeqMapCandidateFallbacks") > 0) + assertTrue(profileString.contains("no_indexed_positive_driver"), + "no_indexed_positive_driver fallback reason is absent from profile") + }) + + test { + sql "SET seq_map_candidate_key_max_count = 0" + exception "seq_map_candidate_key_max_count should be greater than 0" + } + test { + sql "SET seq_map_candidate_key_max_count = -1" + exception "seq_map_candidate_key_max_count should be greater than 0" + } + + sql """ + CREATE TABLE ${costFallbackTableName} ( + `id` BIGINT NOT NULL, + `c` INT NULL, + `s1` BIGINT NULL, + INDEX idx_c (`c`) USING INVERTED + ) ENGINE=OLAP + UNIQUE KEY(`id`) + DISTRIBUTED BY HASH(`id`) BUCKETS 1 + PROPERTIES ( + "replication_num" = "1", + "enable_unique_key_merge_on_write" = "false", + "light_schema_change" = "true", + "disable_auto_compaction" = "true", + "inverted_index_storage_format" = "V3", + "sequence_mapping.s1" = "c" + ) + """ + def broadCandidateValues = + (1..5000).collect { id -> "(${id}, 20, 1)" }.join(",") + sql """ + INSERT INTO ${costFallbackTableName}(id, c, s1) Review Comment: [P1] Make this test create a reader that can reach the cost guard. With only this INSERT, the tablet has the initial empty `[0-1]` rowset plus one non-overlapping rowset starting at version 2; `ReaderParams::has_single_version()` sets `direct_mode`, so candidate collection is skipped. Also, 5,000 rows are below the default 8,160 `batch_size`, and the cost guard is enabled only when `full_scan_rows > batch_size`, so merely splitting this load still cannot test the fallback. Please create at least two nonempty visible rowsets and either exceed the active batch size or temporarily lower and restore it; then assert feature-entry counters before the fallback reason. ########## be/src/exec/scan/olap_scanner.cpp: ########## @@ -163,6 +167,330 @@ static bool has_file_cache_statistics(const io::FileCacheStatistics& stats) { stats.inverted_index_serial_read_rounds != 0; } +std::vector<RowSetSplits> OlapScanner::_clone_rowset_splits() const { + std::vector<RowSetSplits> cloned; + cloned.reserve(_tablet_reader_params.rs_splits.size()); + for (const auto& split : _tablet_reader_params.rs_splits) { + RowSetSplits copy(split.rs_reader->clone()); + copy.segment_offsets = split.segment_offsets; + copy.segment_row_ranges = split.segment_row_ranges; + cloned.emplace_back(std::move(copy)); + } + return cloned; +} + +std::string OlapScanner::_encode_candidate_key(const OlapTuple& key) const { + std::string encoded; + for (size_t i = 0; i < key.size(); ++i) { + const auto& field = key.get_field(i); + const auto type = static_cast<int32_t>(field.get_type()); + encoded.append(reinterpret_cast<const char*>(&type), sizeof(type)); + if (field.is_null()) { + continue; + } + const auto value = field.as_string_view(); + const auto size = static_cast<uint64_t>(value.size()); + encoded.append(reinterpret_cast<const char*>(&size), sizeof(size)); + encoded.append(value); + } + return encoded; +} + +bool OlapScanner::CandidateScanCostLimit::exceeded(int64_t candidate_scan_rows, + size_t candidate_key_count) const { + if (!enabled || full_scan_rows <= 0 || rowset_count == 0 || candidate_scan_rows < 0) { + return false; + } + if (candidate_scan_rows >= full_scan_rows) { + return true; + } + + // A final point-key scan may probe every captured rowset for each candidate key. + // Avoid multiplication overflow by comparing against the remaining row budget. + const auto remaining_rows = static_cast<uint64_t>(full_scan_rows - candidate_scan_rows); + return candidate_key_count > (remaining_rows - 1) / rowset_count; +} + +Status OlapScanner::_collect_seq_map_candidate_keys( + const std::vector<std::shared_ptr<ColumnPredicate>>& driver_predicates, + const std::vector<std::shared_ptr<ColumnPredicate>>& key_predicates, + int64_t max_candidate_keys, const CandidateScanCostLimit& cost_limit, + CandidateKeyMap* candidate_keys, bool* limit_exceeded, bool* cost_exceeded) { + DCHECK(candidate_keys != nullptr); + DCHECK(limit_exceeded != nullptr); + DCHECK(cost_exceeded != nullptr); + *limit_exceeded = false; + *cost_exceeded = false; + candidate_keys->clear(); + + auto candidate_params = _tablet_reader_params; + candidate_params.rs_splits = _clone_rowset_splits(); + candidate_params.predicates.clear(); + for (const auto& predicate : key_predicates) { + candidate_params.predicates.emplace_back(predicate->clone(predicate->column_id())); + } + for (const auto& predicate : driver_predicates) { + candidate_params.predicates.emplace_back(predicate->clone(predicate->column_id())); + } + candidate_params.function_filters.clear(); + candidate_params.all_access_paths.clear(); + candidate_params.predicate_access_paths.clear(); + candidate_params.output_columns.clear(); + candidate_params.extra_columns.clear(); + candidate_params.common_expr_ctxs_push_down.clear(); + candidate_params.topn_filter_source_node_ids.clear(); + candidate_params.key_group_cluster_key_idxes.clear(); + candidate_params.virtual_column_exprs.clear(); + candidate_params.score_runtime.reset(); + candidate_params.collection_statistics.reset(); + candidate_params.ann_topn_runtime.reset(); + candidate_params.direct_mode = true; + candidate_params.aggregation = false; + candidate_params.is_seq_map_candidate_scan = true; + candidate_params.seq_map_candidate_pruned = false; + candidate_params.push_down_agg_type_opt = TPushAggOp::NONE; + candidate_params.read_orderby_key = false; + candidate_params.read_orderby_key_reverse = false; + candidate_params.read_orderby_key_num_prefix_columns = 0; + candidate_params.read_orderby_key_limit = 0; + candidate_params.condition_cache_digest = 0; + candidate_params.general_read_limit = -1; + candidate_params.read_row_binlog = false; + candidate_params.binlog_scan_type = TBinlogScanType::NONE; + candidate_params.start_tso.reset(); + candidate_params.end_tso.reset(); + candidate_params.tso_predicate_column_id.reset(); + + std::vector<ColumnId> candidate_columns; + candidate_columns.reserve(_tablet_reader_params.tablet_schema->num_key_columns() + + driver_predicates.size()); + for (uint32_t cid = 0; cid < _tablet_reader_params.tablet_schema->num_key_columns(); ++cid) { + candidate_columns.push_back(cid); + } + for (const auto& predicate : driver_predicates) { + if (std::find(candidate_columns.begin(), candidate_columns.end(), predicate->column_id()) == + candidate_columns.end()) { + candidate_columns.push_back(predicate->column_id()); + } + } + candidate_params.return_columns = candidate_columns; + candidate_params.origin_return_columns = &candidate_columns; + candidate_params.tablet_columns_convert_to_null_set = nullptr; + + BlockReader candidate_reader; + candidate_reader.set_batch_size(_state->batch_size()); + candidate_reader.set_preferred_block_size_bytes(_state->preferred_block_size_bytes()); + Defer account_candidate_stats {[&]() { + const auto& candidate_stats = candidate_reader.stats(); + auto* total_stats = _tablet_reader->mutable_stats(); + total_stats->seq_map_candidate_scan_rows += candidate_stats.raw_rows_read; + total_stats->seq_map_candidate_scan_bytes += candidate_stats.uncompressed_bytes_read; + total_stats->seq_map_candidate_index_filtered_rows += + candidate_stats.rows_inverted_index_filtered; + total_stats->seq_map_candidate_index_downgrades += + candidate_stats.inverted_index_downgrade_count; + total_stats->seq_map_candidate_index_lookup_ns += + candidate_stats.inverted_index_lookup_timer; + total_stats->seq_map_candidate_cache_local_bytes += + candidate_stats.file_cache_stats.bytes_read_from_local; + total_stats->seq_map_candidate_cache_remote_bytes += + candidate_stats.file_cache_stats.bytes_read_from_remote; + + // Keep the aggregate scan totals consistent with the physical candidate IO. + total_stats->io_ns += candidate_stats.io_ns; + total_stats->compressed_bytes_read += candidate_stats.compressed_bytes_read; + total_stats->decompress_ns += candidate_stats.decompress_ns; + total_stats->uncompressed_bytes_read += candidate_stats.uncompressed_bytes_read; + total_stats->bytes_read += candidate_stats.bytes_read; + total_stats->raw_rows_read += candidate_stats.raw_rows_read; + }}; + RETURN_IF_ERROR(candidate_reader.init(candidate_params)); + + Block block = candidate_params.tablet_schema->create_block(candidate_columns); + const size_t key_column_count = candidate_params.tablet_schema->num_key_columns(); + bool eof = false; + while (!eof) { + RETURN_IF_ERROR(candidate_reader.next_block_with_aggregation(&block, &eof)); + _tablet_reader->mutable_stats()->seq_map_candidate_rows += block.rows(); + for (size_t row = 0; row < block.rows(); ++row) { + OlapTuple key; + for (size_t col = 0; col < key_column_count; ++col) { + Field field; + block.get_by_position(col).column->get(row, field); + key.add_field(std::move(field)); + } + candidate_keys->try_emplace(_encode_candidate_key(key), std::move(key)); + if (candidate_keys->size() > static_cast<size_t>(max_candidate_keys)) { + *limit_exceeded = true; + break; + } + } + block.clear_column_data(); + if (*limit_exceeded) { + break; + } + if (cost_limit.exceeded(candidate_reader.stats().raw_rows_read, candidate_keys->size())) { + *cost_exceeded = true; + break; + } + } + return Status::OK(); +} + +Status OlapScanner::_prepare_seq_map_candidate_keys() { + const auto& query_options = _state->query_options(); + auto& params = _tablet_reader_params; + auto& schema = params.tablet_schema; + if (!query_options.enable_seq_map_candidate_key_scan || schema == nullptr || + !schema->has_seq_map() || schema->keys_type() != KeysType::UNIQUE_KEYS || + params.tablet->enable_unique_key_merge_on_write() || params.direct_mode) { + return Status::OK(); + } + + auto* stats = _tablet_reader->mutable_stats(); + SCOPED_RAW_TIMER(&stats->seq_map_candidate_build_ns); + for (const auto& split : params.rs_splits) { + if (split.segment_offsets != std::pair<int64_t, int64_t> {0, 0} || + !split.segment_row_ranges.empty()) { + ++stats->seq_map_candidate_fallbacks; + _seq_map_candidate_fallback_reason = "partial_scanner_split"; + return Status::OK(); + } + } + const int64_t max_candidate_keys = query_options.seq_map_candidate_key_max_count; + if (max_candidate_keys <= 0) { + ++stats->seq_map_candidate_fallbacks; + _seq_map_candidate_fallback_reason = "invalid_candidate_limit"; + return Status::OK(); + } + if (!query_options.enable_inverted_index_query) { + ++stats->seq_map_candidate_fallbacks; + _seq_map_candidate_fallback_reason = "inverted_index_query_disabled"; + return Status::OK(); + } + + std::vector<std::shared_ptr<ColumnPredicate>> key_predicates; + std::map<uint32_t, std::vector<std::shared_ptr<ColumnPredicate>>> group_drivers; + const auto& value_to_seq = schema->value_col_idx_to_seq_col_idx(); + for (const auto& predicate : params.predicates) { + const auto cid = predicate->column_id(); + const auto& column = schema->column(cid); + if (column.is_key()) { + // Key predicates constrain the candidate reader, but do not identify a value group. + key_predicates.push_back(predicate); + continue; + } + const auto seq_it = value_to_seq.find(cid); + if (seq_it == value_to_seq.end()) { + continue; + } + const auto type = predicate->type(); + const bool positive_driver = type == PredicateType::EQ || type == PredicateType::IN_LIST; + if (!positive_driver || schema->inverted_indexs(column).empty()) { + continue; + } + group_drivers[seq_it->second].push_back(predicate); + ++stats->seq_map_candidate_driver_predicates; + } + + if (group_drivers.empty()) { + ++stats->seq_map_candidate_fallbacks; + _seq_map_candidate_fallback_reason = "no_indexed_positive_driver"; + return Status::OK(); + } + stats->seq_map_candidate_driver_groups = group_drivers.size(); + + CandidateScanCostLimit cost_limit; + cost_limit.rowset_count = params.rs_splits.size(); + for (const auto& split : params.rs_splits) { + const auto row_count = split.rs_reader->rowset()->num_rows(); + if (row_count > static_cast<uint64_t>(std::numeric_limits<int64_t>::max() - + cost_limit.full_scan_rows)) { + cost_limit.full_scan_rows = std::numeric_limits<int64_t>::max(); + break; + } + cost_limit.full_scan_rows += static_cast<int64_t>(row_count); + } + cost_limit.enabled = + cost_limit.rowset_count > 0 && cost_limit.full_scan_rows > _state->batch_size(); + + CandidateKeyMap final_keys; + bool first_group = true; + for (const auto& [seq_col, predicates] : group_drivers) { + CandidateKeyMap group_keys; + bool limit_exceeded = false; + bool cost_exceeded = false; + auto collection_cost_limit = cost_limit; + // A single broad group cannot become selective through a later group intersection. + collection_cost_limit.enabled = collection_cost_limit.enabled && group_drivers.size() == 1; + auto status = _collect_seq_map_candidate_keys(predicates, key_predicates, + max_candidate_keys, collection_cost_limit, + &group_keys, &limit_exceeded, &cost_exceeded); + if (!status.ok()) { + ++stats->seq_map_candidate_fallbacks; + _seq_map_candidate_fallback_reason = "candidate_scan_error"; + LOG(WARNING) << "fallback sequence-mapping candidate scan for tablet " + << params.tablet->tablet_id() << ": " << status; + return Status::OK(); + } + if (limit_exceeded) { + ++stats->seq_map_candidate_fallbacks; + _seq_map_candidate_fallback_reason = "candidate_key_limit"; + return Status::OK(); + } + if (cost_exceeded) { + ++stats->seq_map_candidate_fallbacks; + _seq_map_candidate_fallback_reason = "candidate_cost_limit"; + return Status::OK(); + } + + stats->seq_map_candidate_keys_before_intersect += group_keys.size(); + if (first_group) { + final_keys = std::move(group_keys); + first_group = false; + } else { + for (auto it = final_keys.begin(); it != final_keys.end();) { + if (!group_keys.contains(it->first)) { + it = final_keys.erase(it); + } else { + ++it; + } + } + } + if (final_keys.empty()) { + params.seq_map_candidate_pruned = true; + ++stats->seq_map_candidate_pruned_tablets; + stats->seq_map_candidate_keys_after_intersect = 0; + return Status::OK(); + } + if (cost_limit.exceeded(stats->seq_map_candidate_scan_rows, 0)) { + ++stats->seq_map_candidate_fallbacks; + _seq_map_candidate_fallback_reason = "candidate_cost_limit"; + return Status::OK(); + } + } + + // Use the post-intersection key count for multiple groups. + if (cost_limit.exceeded(stats->seq_map_candidate_scan_rows, final_keys.size())) { + ++stats->seq_map_candidate_fallbacks; + _seq_map_candidate_fallback_reason = "candidate_cost_limit"; + return Status::OK(); + } + Review Comment: [P1] Bound the retained point-key memory, not just the key count. This clears the original FE ranges and materializes both inclusive bounds for every candidate, after which `TabletReader::_init_keys_param()` deep-copies them into two `RowCursor`s per key; each cursor currently allocates its own `Schema` with vectors sized to the full tablet column count. At the 100k default, a wide table can therefore consume gigabytes before the normal scan, and FE accepts any positive `long`; a memory-limit failure occurs after the fallback ranges were destroyed and aborts instead of taking the advertised fallback. Please reserve a byte budget before mutating the ranges, fall back cleanly when it is unavailable, and avoid per-point schema/key duplication (for example, share one immutable key schema/representation). ########## be/src/exec/scan/olap_scanner.cpp: ########## @@ -163,6 +167,330 @@ static bool has_file_cache_statistics(const io::FileCacheStatistics& stats) { stats.inverted_index_serial_read_rounds != 0; } +std::vector<RowSetSplits> OlapScanner::_clone_rowset_splits() const { + std::vector<RowSetSplits> cloned; + cloned.reserve(_tablet_reader_params.rs_splits.size()); + for (const auto& split : _tablet_reader_params.rs_splits) { + RowSetSplits copy(split.rs_reader->clone()); + copy.segment_offsets = split.segment_offsets; + copy.segment_row_ranges = split.segment_row_ranges; + cloned.emplace_back(std::move(copy)); + } + return cloned; +} + +std::string OlapScanner::_encode_candidate_key(const OlapTuple& key) const { + std::string encoded; + for (size_t i = 0; i < key.size(); ++i) { + const auto& field = key.get_field(i); + const auto type = static_cast<int32_t>(field.get_type()); + encoded.append(reinterpret_cast<const char*>(&type), sizeof(type)); + if (field.is_null()) { + continue; + } + const auto value = field.as_string_view(); + const auto size = static_cast<uint64_t>(value.size()); + encoded.append(reinterpret_cast<const char*>(&size), sizeof(size)); + encoded.append(value); + } + return encoded; +} + +bool OlapScanner::CandidateScanCostLimit::exceeded(int64_t candidate_scan_rows, + size_t candidate_key_count) const { + if (!enabled || full_scan_rows <= 0 || rowset_count == 0 || candidate_scan_rows < 0) { + return false; + } + if (candidate_scan_rows >= full_scan_rows) { + return true; + } + + // A final point-key scan may probe every captured rowset for each candidate key. + // Avoid multiplication overflow by comparing against the remaining row budget. + const auto remaining_rows = static_cast<uint64_t>(full_scan_rows - candidate_scan_rows); + return candidate_key_count > (remaining_rows - 1) / rowset_count; +} + +Status OlapScanner::_collect_seq_map_candidate_keys( + const std::vector<std::shared_ptr<ColumnPredicate>>& driver_predicates, + const std::vector<std::shared_ptr<ColumnPredicate>>& key_predicates, + int64_t max_candidate_keys, const CandidateScanCostLimit& cost_limit, + CandidateKeyMap* candidate_keys, bool* limit_exceeded, bool* cost_exceeded) { + DCHECK(candidate_keys != nullptr); + DCHECK(limit_exceeded != nullptr); + DCHECK(cost_exceeded != nullptr); + *limit_exceeded = false; + *cost_exceeded = false; + candidate_keys->clear(); + + auto candidate_params = _tablet_reader_params; + candidate_params.rs_splits = _clone_rowset_splits(); + candidate_params.predicates.clear(); + for (const auto& predicate : key_predicates) { + candidate_params.predicates.emplace_back(predicate->clone(predicate->column_id())); + } + for (const auto& predicate : driver_predicates) { + candidate_params.predicates.emplace_back(predicate->clone(predicate->column_id())); + } + candidate_params.function_filters.clear(); + candidate_params.all_access_paths.clear(); + candidate_params.predicate_access_paths.clear(); + candidate_params.output_columns.clear(); + candidate_params.extra_columns.clear(); + candidate_params.common_expr_ctxs_push_down.clear(); + candidate_params.topn_filter_source_node_ids.clear(); + candidate_params.key_group_cluster_key_idxes.clear(); + candidate_params.virtual_column_exprs.clear(); + candidate_params.score_runtime.reset(); + candidate_params.collection_statistics.reset(); + candidate_params.ann_topn_runtime.reset(); + candidate_params.direct_mode = true; + candidate_params.aggregation = false; + candidate_params.is_seq_map_candidate_scan = true; + candidate_params.seq_map_candidate_pruned = false; + candidate_params.push_down_agg_type_opt = TPushAggOp::NONE; + candidate_params.read_orderby_key = false; + candidate_params.read_orderby_key_reverse = false; + candidate_params.read_orderby_key_num_prefix_columns = 0; + candidate_params.read_orderby_key_limit = 0; + candidate_params.condition_cache_digest = 0; + candidate_params.general_read_limit = -1; + candidate_params.read_row_binlog = false; + candidate_params.binlog_scan_type = TBinlogScanType::NONE; + candidate_params.start_tso.reset(); + candidate_params.end_tso.reset(); + candidate_params.tso_predicate_column_id.reset(); + + std::vector<ColumnId> candidate_columns; + candidate_columns.reserve(_tablet_reader_params.tablet_schema->num_key_columns() + + driver_predicates.size()); + for (uint32_t cid = 0; cid < _tablet_reader_params.tablet_schema->num_key_columns(); ++cid) { + candidate_columns.push_back(cid); + } + for (const auto& predicate : driver_predicates) { + if (std::find(candidate_columns.begin(), candidate_columns.end(), predicate->column_id()) == + candidate_columns.end()) { + candidate_columns.push_back(predicate->column_id()); + } + } + candidate_params.return_columns = candidate_columns; + candidate_params.origin_return_columns = &candidate_columns; + candidate_params.tablet_columns_convert_to_null_set = nullptr; + + BlockReader candidate_reader; + candidate_reader.set_batch_size(_state->batch_size()); + candidate_reader.set_preferred_block_size_bytes(_state->preferred_block_size_bytes()); + Defer account_candidate_stats {[&]() { + const auto& candidate_stats = candidate_reader.stats(); + auto* total_stats = _tablet_reader->mutable_stats(); + total_stats->seq_map_candidate_scan_rows += candidate_stats.raw_rows_read; + total_stats->seq_map_candidate_scan_bytes += candidate_stats.uncompressed_bytes_read; + total_stats->seq_map_candidate_index_filtered_rows += + candidate_stats.rows_inverted_index_filtered; + total_stats->seq_map_candidate_index_downgrades += + candidate_stats.inverted_index_downgrade_count; + total_stats->seq_map_candidate_index_lookup_ns += + candidate_stats.inverted_index_lookup_timer; + total_stats->seq_map_candidate_cache_local_bytes += + candidate_stats.file_cache_stats.bytes_read_from_local; + total_stats->seq_map_candidate_cache_remote_bytes += + candidate_stats.file_cache_stats.bytes_read_from_remote; + + // Keep the aggregate scan totals consistent with the physical candidate IO. + total_stats->io_ns += candidate_stats.io_ns; + total_stats->compressed_bytes_read += candidate_stats.compressed_bytes_read; + total_stats->decompress_ns += candidate_stats.decompress_ns; + total_stats->uncompressed_bytes_read += candidate_stats.uncompressed_bytes_read; + total_stats->bytes_read += candidate_stats.bytes_read; + total_stats->raw_rows_read += candidate_stats.raw_rows_read; + }}; + RETURN_IF_ERROR(candidate_reader.init(candidate_params)); + + Block block = candidate_params.tablet_schema->create_block(candidate_columns); + const size_t key_column_count = candidate_params.tablet_schema->num_key_columns(); + bool eof = false; + while (!eof) { + RETURN_IF_ERROR(candidate_reader.next_block_with_aggregation(&block, &eof)); + _tablet_reader->mutable_stats()->seq_map_candidate_rows += block.rows(); + for (size_t row = 0; row < block.rows(); ++row) { + OlapTuple key; + for (size_t col = 0; col < key_column_count; ++col) { + Field field; + block.get_by_position(col).column->get(row, field); + key.add_field(std::move(field)); + } + candidate_keys->try_emplace(_encode_candidate_key(key), std::move(key)); + if (candidate_keys->size() > static_cast<size_t>(max_candidate_keys)) { + *limit_exceeded = true; + break; + } + } + block.clear_column_data(); + if (*limit_exceeded) { + break; + } + if (cost_limit.exceeded(candidate_reader.stats().raw_rows_read, candidate_keys->size())) { + *cost_exceeded = true; + break; + } + } + return Status::OK(); +} + +Status OlapScanner::_prepare_seq_map_candidate_keys() { + const auto& query_options = _state->query_options(); + auto& params = _tablet_reader_params; + auto& schema = params.tablet_schema; + if (!query_options.enable_seq_map_candidate_key_scan || schema == nullptr || + !schema->has_seq_map() || schema->keys_type() != KeysType::UNIQUE_KEYS || + params.tablet->enable_unique_key_merge_on_write() || params.direct_mode) { + return Status::OK(); + } + + auto* stats = _tablet_reader->mutable_stats(); + SCOPED_RAW_TIMER(&stats->seq_map_candidate_build_ns); + for (const auto& split : params.rs_splits) { + if (split.segment_offsets != std::pair<int64_t, int64_t> {0, 0} || + !split.segment_row_ranges.empty()) { + ++stats->seq_map_candidate_fallbacks; + _seq_map_candidate_fallback_reason = "partial_scanner_split"; + return Status::OK(); + } + } + const int64_t max_candidate_keys = query_options.seq_map_candidate_key_max_count; + if (max_candidate_keys <= 0) { + ++stats->seq_map_candidate_fallbacks; + _seq_map_candidate_fallback_reason = "invalid_candidate_limit"; + return Status::OK(); + } + if (!query_options.enable_inverted_index_query) { + ++stats->seq_map_candidate_fallbacks; + _seq_map_candidate_fallback_reason = "inverted_index_query_disabled"; + return Status::OK(); + } + + std::vector<std::shared_ptr<ColumnPredicate>> key_predicates; + std::map<uint32_t, std::vector<std::shared_ptr<ColumnPredicate>>> group_drivers; + const auto& value_to_seq = schema->value_col_idx_to_seq_col_idx(); + for (const auto& predicate : params.predicates) { + const auto cid = predicate->column_id(); + const auto& column = schema->column(cid); + if (column.is_key()) { + // Key predicates constrain the candidate reader, but do not identify a value group. + key_predicates.push_back(predicate); + continue; + } + const auto seq_it = value_to_seq.find(cid); + if (seq_it == value_to_seq.end()) { + continue; + } + const auto type = predicate->type(); + const bool positive_driver = type == PredicateType::EQ || type == PredicateType::IN_LIST; + if (!positive_driver || schema->inverted_indexs(column).empty()) { + continue; + } + group_drivers[seq_it->second].push_back(predicate); + ++stats->seq_map_candidate_driver_predicates; + } + + if (group_drivers.empty()) { + ++stats->seq_map_candidate_fallbacks; + _seq_map_candidate_fallback_reason = "no_indexed_positive_driver"; + return Status::OK(); + } + stats->seq_map_candidate_driver_groups = group_drivers.size(); + + CandidateScanCostLimit cost_limit; + cost_limit.rowset_count = params.rs_splits.size(); + for (const auto& split : params.rs_splits) { + const auto row_count = split.rs_reader->rowset()->num_rows(); + if (row_count > static_cast<uint64_t>(std::numeric_limits<int64_t>::max() - + cost_limit.full_scan_rows)) { + cost_limit.full_scan_rows = std::numeric_limits<int64_t>::max(); + break; + } + cost_limit.full_scan_rows += static_cast<int64_t>(row_count); + } + cost_limit.enabled = + cost_limit.rowset_count > 0 && cost_limit.full_scan_rows > _state->batch_size(); + + CandidateKeyMap final_keys; + bool first_group = true; + for (const auto& [seq_col, predicates] : group_drivers) { + CandidateKeyMap group_keys; + bool limit_exceeded = false; + bool cost_exceeded = false; + auto collection_cost_limit = cost_limit; Review Comment: [P1] Enforce the cumulative candidate-row budget between blocks for every group. This disables the collector's cost check whenever there is more than one driver group, so the cumulative check runs only after an entire group returns. On a 100k-row tablet, two indexed groups that each match 90k rows can read 180k candidate rows before falling back to the normal 100k-row scan. Please pass the prior groups' spent rows (or the remaining row budget) into the collector and stop as soon as the cumulative candidate scan reaches `full_scan_rows`, while still deferring the final point-key cost until after intersection; add a two-broad-group regression for this cutoff. ########## be/src/exec/scan/olap_scanner.cpp: ########## @@ -163,6 +167,330 @@ static bool has_file_cache_statistics(const io::FileCacheStatistics& stats) { stats.inverted_index_serial_read_rounds != 0; } +std::vector<RowSetSplits> OlapScanner::_clone_rowset_splits() const { + std::vector<RowSetSplits> cloned; + cloned.reserve(_tablet_reader_params.rs_splits.size()); + for (const auto& split : _tablet_reader_params.rs_splits) { + RowSetSplits copy(split.rs_reader->clone()); + copy.segment_offsets = split.segment_offsets; + copy.segment_row_ranges = split.segment_row_ranges; + cloned.emplace_back(std::move(copy)); + } + return cloned; +} + +std::string OlapScanner::_encode_candidate_key(const OlapTuple& key) const { + std::string encoded; + for (size_t i = 0; i < key.size(); ++i) { + const auto& field = key.get_field(i); + const auto type = static_cast<int32_t>(field.get_type()); + encoded.append(reinterpret_cast<const char*>(&type), sizeof(type)); + if (field.is_null()) { + continue; + } + const auto value = field.as_string_view(); + const auto size = static_cast<uint64_t>(value.size()); + encoded.append(reinterpret_cast<const char*>(&size), sizeof(size)); + encoded.append(value); + } + return encoded; +} + +bool OlapScanner::CandidateScanCostLimit::exceeded(int64_t candidate_scan_rows, + size_t candidate_key_count) const { + if (!enabled || full_scan_rows <= 0 || rowset_count == 0 || candidate_scan_rows < 0) { + return false; + } + if (candidate_scan_rows >= full_scan_rows) { + return true; + } + + // A final point-key scan may probe every captured rowset for each candidate key. + // Avoid multiplication overflow by comparing against the remaining row budget. + const auto remaining_rows = static_cast<uint64_t>(full_scan_rows - candidate_scan_rows); + return candidate_key_count > (remaining_rows - 1) / rowset_count; +} + +Status OlapScanner::_collect_seq_map_candidate_keys( + const std::vector<std::shared_ptr<ColumnPredicate>>& driver_predicates, + const std::vector<std::shared_ptr<ColumnPredicate>>& key_predicates, + int64_t max_candidate_keys, const CandidateScanCostLimit& cost_limit, + CandidateKeyMap* candidate_keys, bool* limit_exceeded, bool* cost_exceeded) { + DCHECK(candidate_keys != nullptr); + DCHECK(limit_exceeded != nullptr); + DCHECK(cost_exceeded != nullptr); + *limit_exceeded = false; + *cost_exceeded = false; + candidate_keys->clear(); + + auto candidate_params = _tablet_reader_params; + candidate_params.rs_splits = _clone_rowset_splits(); + candidate_params.predicates.clear(); + for (const auto& predicate : key_predicates) { + candidate_params.predicates.emplace_back(predicate->clone(predicate->column_id())); + } + for (const auto& predicate : driver_predicates) { + candidate_params.predicates.emplace_back(predicate->clone(predicate->column_id())); + } + candidate_params.function_filters.clear(); + candidate_params.all_access_paths.clear(); + candidate_params.predicate_access_paths.clear(); + candidate_params.output_columns.clear(); + candidate_params.extra_columns.clear(); + candidate_params.common_expr_ctxs_push_down.clear(); + candidate_params.topn_filter_source_node_ids.clear(); + candidate_params.key_group_cluster_key_idxes.clear(); + candidate_params.virtual_column_exprs.clear(); + candidate_params.score_runtime.reset(); + candidate_params.collection_statistics.reset(); + candidate_params.ann_topn_runtime.reset(); + candidate_params.direct_mode = true; + candidate_params.aggregation = false; + candidate_params.is_seq_map_candidate_scan = true; + candidate_params.seq_map_candidate_pruned = false; + candidate_params.push_down_agg_type_opt = TPushAggOp::NONE; + candidate_params.read_orderby_key = false; + candidate_params.read_orderby_key_reverse = false; + candidate_params.read_orderby_key_num_prefix_columns = 0; + candidate_params.read_orderby_key_limit = 0; + candidate_params.condition_cache_digest = 0; + candidate_params.general_read_limit = -1; + candidate_params.read_row_binlog = false; + candidate_params.binlog_scan_type = TBinlogScanType::NONE; + candidate_params.start_tso.reset(); + candidate_params.end_tso.reset(); + candidate_params.tso_predicate_column_id.reset(); + + std::vector<ColumnId> candidate_columns; + candidate_columns.reserve(_tablet_reader_params.tablet_schema->num_key_columns() + + driver_predicates.size()); + for (uint32_t cid = 0; cid < _tablet_reader_params.tablet_schema->num_key_columns(); ++cid) { + candidate_columns.push_back(cid); + } + for (const auto& predicate : driver_predicates) { + if (std::find(candidate_columns.begin(), candidate_columns.end(), predicate->column_id()) == + candidate_columns.end()) { + candidate_columns.push_back(predicate->column_id()); + } + } + candidate_params.return_columns = candidate_columns; + candidate_params.origin_return_columns = &candidate_columns; + candidate_params.tablet_columns_convert_to_null_set = nullptr; + + BlockReader candidate_reader; + candidate_reader.set_batch_size(_state->batch_size()); + candidate_reader.set_preferred_block_size_bytes(_state->preferred_block_size_bytes()); + Defer account_candidate_stats {[&]() { + const auto& candidate_stats = candidate_reader.stats(); + auto* total_stats = _tablet_reader->mutable_stats(); + total_stats->seq_map_candidate_scan_rows += candidate_stats.raw_rows_read; + total_stats->seq_map_candidate_scan_bytes += candidate_stats.uncompressed_bytes_read; + total_stats->seq_map_candidate_index_filtered_rows += + candidate_stats.rows_inverted_index_filtered; + total_stats->seq_map_candidate_index_downgrades += + candidate_stats.inverted_index_downgrade_count; + total_stats->seq_map_candidate_index_lookup_ns += + candidate_stats.inverted_index_lookup_timer; + total_stats->seq_map_candidate_cache_local_bytes += + candidate_stats.file_cache_stats.bytes_read_from_local; + total_stats->seq_map_candidate_cache_remote_bytes += + candidate_stats.file_cache_stats.bytes_read_from_remote; + + // Keep the aggregate scan totals consistent with the physical candidate IO. + total_stats->io_ns += candidate_stats.io_ns; + total_stats->compressed_bytes_read += candidate_stats.compressed_bytes_read; + total_stats->decompress_ns += candidate_stats.decompress_ns; + total_stats->uncompressed_bytes_read += candidate_stats.uncompressed_bytes_read; + total_stats->bytes_read += candidate_stats.bytes_read; + total_stats->raw_rows_read += candidate_stats.raw_rows_read; + }}; + RETURN_IF_ERROR(candidate_reader.init(candidate_params)); + + Block block = candidate_params.tablet_schema->create_block(candidate_columns); + const size_t key_column_count = candidate_params.tablet_schema->num_key_columns(); + bool eof = false; + while (!eof) { + RETURN_IF_ERROR(candidate_reader.next_block_with_aggregation(&block, &eof)); + _tablet_reader->mutable_stats()->seq_map_candidate_rows += block.rows(); + for (size_t row = 0; row < block.rows(); ++row) { + OlapTuple key; + for (size_t col = 0; col < key_column_count; ++col) { + Field field; + block.get_by_position(col).column->get(row, field); + key.add_field(std::move(field)); + } + candidate_keys->try_emplace(_encode_candidate_key(key), std::move(key)); + if (candidate_keys->size() > static_cast<size_t>(max_candidate_keys)) { + *limit_exceeded = true; + break; + } + } + block.clear_column_data(); + if (*limit_exceeded) { + break; + } + if (cost_limit.exceeded(candidate_reader.stats().raw_rows_read, candidate_keys->size())) { + *cost_exceeded = true; + break; + } + } + return Status::OK(); +} + +Status OlapScanner::_prepare_seq_map_candidate_keys() { + const auto& query_options = _state->query_options(); + auto& params = _tablet_reader_params; + auto& schema = params.tablet_schema; + if (!query_options.enable_seq_map_candidate_key_scan || schema == nullptr || + !schema->has_seq_map() || schema->keys_type() != KeysType::UNIQUE_KEYS || + params.tablet->enable_unique_key_merge_on_write() || params.direct_mode) { + return Status::OK(); + } + + auto* stats = _tablet_reader->mutable_stats(); + SCOPED_RAW_TIMER(&stats->seq_map_candidate_build_ns); + for (const auto& split : params.rs_splits) { + if (split.segment_offsets != std::pair<int64_t, int64_t> {0, 0} || + !split.segment_row_ranges.empty()) { + ++stats->seq_map_candidate_fallbacks; + _seq_map_candidate_fallback_reason = "partial_scanner_split"; + return Status::OK(); + } + } + const int64_t max_candidate_keys = query_options.seq_map_candidate_key_max_count; + if (max_candidate_keys <= 0) { + ++stats->seq_map_candidate_fallbacks; + _seq_map_candidate_fallback_reason = "invalid_candidate_limit"; + return Status::OK(); + } + if (!query_options.enable_inverted_index_query) { + ++stats->seq_map_candidate_fallbacks; + _seq_map_candidate_fallback_reason = "inverted_index_query_disabled"; + return Status::OK(); + } + + std::vector<std::shared_ptr<ColumnPredicate>> key_predicates; + std::map<uint32_t, std::vector<std::shared_ptr<ColumnPredicate>>> group_drivers; + const auto& value_to_seq = schema->value_col_idx_to_seq_col_idx(); + for (const auto& predicate : params.predicates) { + const auto cid = predicate->column_id(); + const auto& column = schema->column(cid); + if (column.is_key()) { + // Key predicates constrain the candidate reader, but do not identify a value group. + key_predicates.push_back(predicate); + continue; + } + const auto seq_it = value_to_seq.find(cid); + if (seq_it == value_to_seq.end()) { + continue; + } + const auto type = predicate->type(); + const bool positive_driver = type == PredicateType::EQ || type == PredicateType::IN_LIST; + if (!positive_driver || schema->inverted_indexs(column).empty()) { + continue; + } + group_drivers[seq_it->second].push_back(predicate); + ++stats->seq_map_candidate_driver_predicates; + } + + if (group_drivers.empty()) { + ++stats->seq_map_candidate_fallbacks; + _seq_map_candidate_fallback_reason = "no_indexed_positive_driver"; + return Status::OK(); + } + stats->seq_map_candidate_driver_groups = group_drivers.size(); + + CandidateScanCostLimit cost_limit; + cost_limit.rowset_count = params.rs_splits.size(); Review Comment: [P1] Base the break-even budget on the constrained normal scan. This sums every physical row in the captured rowsets, but both the candidate readers and the unchanged reader apply the original key ranges and key predicates. On a large tablet with a narrow primary-key range, the normal baseline can be roughly K rows while this budget remains tablet-wide, allowing each sequence group to replay those K rows plus the final scan without ever tripping the guard. Please estimate the baseline after original key pruning, or conservatively skip the optimization when no trustworthy estimate is available; add a large-table/narrow-range multi-group regression. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
