github-actions[bot] commented on code in PR #66881:
URL: https://github.com/apache/doris/pull/66881#discussion_r3843225270
##########
be/src/exec/scan/olap_scanner.cpp:
##########
@@ -312,6 +900,11 @@ Status OlapScanner::_open_impl(RuntimeState* state) {
RETURN_IF_ERROR(Scanner::_open_impl(state));
SCOPED_TIMER(_local_state->cast<OlapScanLocalState>()._reader_init_timer);
+ RETURN_IF_ERROR(_prepare_seq_map_candidate_keys());
Review Comment:
[P1] Do not execute the multi-block candidate scan as one Scanner::open()
task. The scheduler normally reads one output block and reschedules, then
publishes realtime row/byte counters; this prepass can consume candidate blocks
up to the full-scan cutoff before open() returns, so a large tablet can
monopolize the scan worker and delay workload-group IO accounting for the
entire prepass. The cumulative cost guards bound total work but do not restore
fairness or interim accounting, and the started max_run_time_watch is never
consulted. Please make candidate collection resumable under the scanner task
budget, or cap synchronous work tightly and fall back while publishing deltas
between blocks.
##########
be/src/exec/scan/olap_scanner.cpp:
##########
@@ -163,6 +178,579 @@ 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) {
+ 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;
+}
+
+OlapScanner::CandidateMemoryBudget OlapScanner::_split_candidate_memory_budget(
+ size_t reservation_bytes) {
+ if (reservation_bytes <= MIN_SEQ_MAP_CANDIDATE_WORKSPACE_BYTES) {
+ return {};
+ }
+ const size_t workspace_bytes =
+ std::clamp(reservation_bytes / 5,
MIN_SEQ_MAP_CANDIDATE_WORKSPACE_BYTES,
+ MAX_SEQ_MAP_CANDIDATE_WORKSPACE_BYTES);
+ const size_t key_bytes =
+ std::min(MAX_SEQ_MAP_CANDIDATE_KEY_BYTES, reservation_bytes -
workspace_bytes);
+ return {
+ .reservation_bytes = key_bytes + workspace_bytes,
+ .key_bytes = key_bytes,
+ .workspace_bytes = workspace_bytes,
+ };
+}
+
+OlapScanner::CandidateMemoryBudget OlapScanner::_candidate_memory_budget()
const {
+ const auto tracker = _state->query_mem_tracker();
+ if (tracker->limit() < 0) {
+ return
_split_candidate_memory_budget(MAX_SEQ_MAP_CANDIDATE_RESERVATION_BYTES);
+ }
+ if (tracker->consumption() >= tracker->limit()) {
+ return {};
+ }
+ const auto remaining = static_cast<size_t>(tracker->limit() -
tracker->consumption());
+ return _split_candidate_memory_budget(
+ std::min(MAX_SEQ_MAP_CANDIDATE_RESERVATION_BYTES, remaining / 8));
+}
+
+size_t OlapScanner::_estimate_candidate_key_bytes(const std::string&
encoded_key,
+ size_t key_column_count) {
+ // encoded_key contains the complete variable-length payload. Count it
once for the map key
+ // and once as a conservative proxy for payload owned by string-like
Fields.
+ const size_t fixed_bytes = sizeof(CandidateKeyMap::value_type) + 4 *
sizeof(void*) +
+ sizeof(RowCursor) + key_column_count *
sizeof(Field);
+ if (encoded_key.size() > (std::numeric_limits<size_t>::max() -
fixed_bytes) / 2) {
+ return std::numeric_limits<size_t>::max();
+ }
+ return fixed_bytes + 2 * encoded_key.size();
+}
+
+OlapScanner::CandidateKeyInsertResult
OlapScanner::_try_add_seq_map_candidate_key(
+ std::string encoded_key, OlapTuple&& key, size_t key_column_count,
+ size_t max_candidate_bytes, size_t reservation_headroom_bytes,
+ CandidateKeyMap* candidate_keys, size_t* candidate_bytes) {
+ DCHECK(candidate_keys != nullptr);
+ DCHECK(candidate_bytes != nullptr);
+ if (candidate_keys->contains(encoded_key)) {
+ return CandidateKeyInsertResult::OK;
+ }
+
+ const size_t key_bytes = _estimate_candidate_key_bytes(encoded_key,
key_column_count);
+ if (*candidate_bytes > max_candidate_bytes ||
+ key_bytes > max_candidate_bytes - *candidate_bytes) {
+ return CandidateKeyInsertResult::KEY_BYTES_LIMIT;
+ }
+ if (key_bytes > reservation_headroom_bytes) {
+ return CandidateKeyInsertResult::RESERVATION_LIMIT;
+ }
+ candidate_keys->emplace(std::move(encoded_key), std::move(key));
+ *candidate_bytes += key_bytes;
+ return CandidateKeyInsertResult::OK;
+}
+
+size_t OlapScanner::_estimate_candidate_map_bytes(const CandidateKeyMap&
candidate_keys) const {
+ size_t bytes = 0;
+ const size_t key_column_count =
_tablet_reader_params.tablet_schema->num_key_columns();
+ for (const auto& entry : candidate_keys) {
+ const size_t key_bytes = _estimate_candidate_key_bytes(entry.first,
key_column_count);
+ if (key_bytes > std::numeric_limits<size_t>::max() - bytes) {
+ return std::numeric_limits<size_t>::max();
+ }
+ bytes += key_bytes;
+ }
+ return bytes;
+}
+
+static size_t saturating_add_size(size_t lhs, size_t rhs) {
+ return rhs > std::numeric_limits<size_t>::max() - lhs ?
std::numeric_limits<size_t>::max()
+ : lhs + rhs;
+}
+
+static size_t saturating_multiply_size(size_t lhs, size_t rhs) {
+ return lhs != 0 && rhs > std::numeric_limits<size_t>::max() / lhs
+ ? std::numeric_limits<size_t>::max()
+ : lhs * rhs;
+}
+
+bool OlapScanner::CandidateScanCostLimit::exceeded(int64_t
previous_candidate_scan_rows,
+ int64_t
current_candidate_scan_rows,
+ size_t candidate_key_count)
const {
+ if (!enabled || full_scan_rows <= 0 || point_probe_cost_per_key == 0 ||
+ previous_candidate_scan_rows < 0 || current_candidate_scan_rows < 0) {
+ return false;
+ }
+ if (previous_candidate_scan_rows >= full_scan_rows ||
+ current_candidate_scan_rows >= full_scan_rows -
previous_candidate_scan_rows) {
+ return true;
+ }
+
+ // Compare against the remaining row budget without multiplying candidate
count by the
+ // weighted lower/upper short-key probe cost.
+ const auto remaining_rows = static_cast<uint64_t>(
+ full_scan_rows - previous_candidate_scan_rows -
current_candidate_scan_rows);
+ return candidate_key_count > (remaining_rows - 1) /
point_probe_cost_per_key;
+}
+
+void OlapScanner::_add_seq_map_candidate_cost(uint64_t row_count, size_t
segment_count,
+ CandidateScanCostLimit*
cost_limit) {
+ DCHECK(cost_limit != nullptr);
+ if (cost_limit->full_scan_rows != std::numeric_limits<int64_t>::max()) {
+ 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();
+ } else {
+ cost_limit->full_scan_rows += static_cast<int64_t>(row_count);
+ }
+ }
+
+ // MOR point lookup uses the short-key path. Each lower/upper ordinal
lookup can binary-search
+ // up to the rowset row count, which is a conservative upper bound for
every segment.
+ const size_t binary_search_steps =
+ std::max<size_t>(1, std::bit_width(std::max<uint64_t>(1,
row_count)));
+ const size_t rowset_probe_cost = saturating_multiply_size(
+ saturating_multiply_size(2, segment_count), binary_search_steps);
+ cost_limit->point_probe_cost_per_key =
+ saturating_add_size(cost_limit->point_probe_cost_per_key,
rowset_probe_cost);
+}
+
+void OlapScanner::_merge_seq_map_candidate_stats(const OlapReaderStatistics&
candidate_stats,
+ OlapReaderStatistics*
total_stats) {
+ DCHECK(total_stats != nullptr);
+ 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;
+ total_stats->file_cache_stats.merge_from(candidate_stats.file_cache_stats);
+
+ 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;
+}
+
+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 previous_candidate_scan_rows, bool price_point_lookups,
int64_t max_candidate_keys,
+ size_t max_candidate_bytes, size_t candidate_workspace_bytes,
+ const CandidateScanCostLimit& cost_limit, CandidateKeyMap*
candidate_keys,
+ size_t* candidate_bytes, bool* limit_exceeded, bool* bytes_exceeded,
+ bool* reservation_exceeded, bool* cost_exceeded) {
+ DCHECK(candidate_keys != nullptr);
+ DCHECK(candidate_bytes != nullptr);
+ DCHECK(limit_exceeded != nullptr);
+ DCHECK(bytes_exceeded != nullptr);
+ DCHECK(reservation_exceeded != nullptr);
+ DCHECK(cost_exceeded != nullptr);
+ *candidate_bytes = 0;
+ *limit_exceeded = false;
+ *bytes_exceeded = false;
+ *reservation_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;
Review Comment:
[P1] Keep scanner-level LIMIT queries out of this prepass. This path clears
general_read_limit and runs from open() before Scanner::get_block() can apply
_limit/_shared_scan_limit, while admission still prices the unchanged path as a
tablet-wide scan. A broad indexed LIMIT 1 can therefore build candidates up to
the key/memory/cost guard on every scanner before the normal reader returns its
first block, although the unchanged MOR scan can stop there. The existing
key_range_present guard does not cover an ordinary LIMIT because it leaves
start_key/end_key empty. Please fall back when a scan limit is active, or use a
sound limit-aware baseline; do not truncate the candidate set to LIMIT because
incomplete point keys would create false negatives.
##########
be/src/exec/scan/olap_scanner.cpp:
##########
@@ -163,6 +178,579 @@ 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) {
+ 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;
+}
+
+OlapScanner::CandidateMemoryBudget OlapScanner::_split_candidate_memory_budget(
+ size_t reservation_bytes) {
+ if (reservation_bytes <= MIN_SEQ_MAP_CANDIDATE_WORKSPACE_BYTES) {
+ return {};
+ }
+ const size_t workspace_bytes =
+ std::clamp(reservation_bytes / 5,
MIN_SEQ_MAP_CANDIDATE_WORKSPACE_BYTES,
+ MAX_SEQ_MAP_CANDIDATE_WORKSPACE_BYTES);
+ const size_t key_bytes =
+ std::min(MAX_SEQ_MAP_CANDIDATE_KEY_BYTES, reservation_bytes -
workspace_bytes);
+ return {
+ .reservation_bytes = key_bytes + workspace_bytes,
+ .key_bytes = key_bytes,
+ .workspace_bytes = workspace_bytes,
+ };
+}
+
+OlapScanner::CandidateMemoryBudget OlapScanner::_candidate_memory_budget()
const {
+ const auto tracker = _state->query_mem_tracker();
+ if (tracker->limit() < 0) {
+ return
_split_candidate_memory_budget(MAX_SEQ_MAP_CANDIDATE_RESERVATION_BYTES);
+ }
+ if (tracker->consumption() >= tracker->limit()) {
+ return {};
+ }
+ const auto remaining = static_cast<size_t>(tracker->limit() -
tracker->consumption());
+ return _split_candidate_memory_budget(
+ std::min(MAX_SEQ_MAP_CANDIDATE_RESERVATION_BYTES, remaining / 8));
+}
+
+size_t OlapScanner::_estimate_candidate_key_bytes(const std::string&
encoded_key,
+ size_t key_column_count) {
+ // encoded_key contains the complete variable-length payload. Count it
once for the map key
+ // and once as a conservative proxy for payload owned by string-like
Fields.
+ const size_t fixed_bytes = sizeof(CandidateKeyMap::value_type) + 4 *
sizeof(void*) +
+ sizeof(RowCursor) + key_column_count *
sizeof(Field);
+ if (encoded_key.size() > (std::numeric_limits<size_t>::max() -
fixed_bytes) / 2) {
+ return std::numeric_limits<size_t>::max();
+ }
+ return fixed_bytes + 2 * encoded_key.size();
+}
+
+OlapScanner::CandidateKeyInsertResult
OlapScanner::_try_add_seq_map_candidate_key(
+ std::string encoded_key, OlapTuple&& key, size_t key_column_count,
+ size_t max_candidate_bytes, size_t reservation_headroom_bytes,
+ CandidateKeyMap* candidate_keys, size_t* candidate_bytes) {
+ DCHECK(candidate_keys != nullptr);
+ DCHECK(candidate_bytes != nullptr);
+ if (candidate_keys->contains(encoded_key)) {
+ return CandidateKeyInsertResult::OK;
+ }
+
+ const size_t key_bytes = _estimate_candidate_key_bytes(encoded_key,
key_column_count);
+ if (*candidate_bytes > max_candidate_bytes ||
+ key_bytes > max_candidate_bytes - *candidate_bytes) {
+ return CandidateKeyInsertResult::KEY_BYTES_LIMIT;
+ }
+ if (key_bytes > reservation_headroom_bytes) {
+ return CandidateKeyInsertResult::RESERVATION_LIMIT;
+ }
+ candidate_keys->emplace(std::move(encoded_key), std::move(key));
+ *candidate_bytes += key_bytes;
+ return CandidateKeyInsertResult::OK;
+}
+
+size_t OlapScanner::_estimate_candidate_map_bytes(const CandidateKeyMap&
candidate_keys) const {
+ size_t bytes = 0;
+ const size_t key_column_count =
_tablet_reader_params.tablet_schema->num_key_columns();
+ for (const auto& entry : candidate_keys) {
+ const size_t key_bytes = _estimate_candidate_key_bytes(entry.first,
key_column_count);
+ if (key_bytes > std::numeric_limits<size_t>::max() - bytes) {
+ return std::numeric_limits<size_t>::max();
+ }
+ bytes += key_bytes;
+ }
+ return bytes;
+}
+
+static size_t saturating_add_size(size_t lhs, size_t rhs) {
+ return rhs > std::numeric_limits<size_t>::max() - lhs ?
std::numeric_limits<size_t>::max()
+ : lhs + rhs;
+}
+
+static size_t saturating_multiply_size(size_t lhs, size_t rhs) {
+ return lhs != 0 && rhs > std::numeric_limits<size_t>::max() / lhs
+ ? std::numeric_limits<size_t>::max()
+ : lhs * rhs;
+}
+
+bool OlapScanner::CandidateScanCostLimit::exceeded(int64_t
previous_candidate_scan_rows,
+ int64_t
current_candidate_scan_rows,
+ size_t candidate_key_count)
const {
+ if (!enabled || full_scan_rows <= 0 || point_probe_cost_per_key == 0 ||
+ previous_candidate_scan_rows < 0 || current_candidate_scan_rows < 0) {
+ return false;
+ }
+ if (previous_candidate_scan_rows >= full_scan_rows ||
+ current_candidate_scan_rows >= full_scan_rows -
previous_candidate_scan_rows) {
+ return true;
+ }
+
+ // Compare against the remaining row budget without multiplying candidate
count by the
+ // weighted lower/upper short-key probe cost.
+ const auto remaining_rows = static_cast<uint64_t>(
+ full_scan_rows - previous_candidate_scan_rows -
current_candidate_scan_rows);
+ return candidate_key_count > (remaining_rows - 1) /
point_probe_cost_per_key;
+}
+
+void OlapScanner::_add_seq_map_candidate_cost(uint64_t row_count, size_t
segment_count,
+ CandidateScanCostLimit*
cost_limit) {
+ DCHECK(cost_limit != nullptr);
+ if (cost_limit->full_scan_rows != std::numeric_limits<int64_t>::max()) {
+ 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();
+ } else {
+ cost_limit->full_scan_rows += static_cast<int64_t>(row_count);
+ }
+ }
+
+ // MOR point lookup uses the short-key path. Each lower/upper ordinal
lookup can binary-search
+ // up to the rowset row count, which is a conservative upper bound for
every segment.
+ const size_t binary_search_steps =
+ std::max<size_t>(1, std::bit_width(std::max<uint64_t>(1,
row_count)));
+ const size_t rowset_probe_cost = saturating_multiply_size(
+ saturating_multiply_size(2, segment_count), binary_search_steps);
+ cost_limit->point_probe_cost_per_key =
+ saturating_add_size(cost_limit->point_probe_cost_per_key,
rowset_probe_cost);
+}
+
+void OlapScanner::_merge_seq_map_candidate_stats(const OlapReaderStatistics&
candidate_stats,
+ OlapReaderStatistics*
total_stats) {
+ DCHECK(total_stats != nullptr);
+ 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;
+ total_stats->file_cache_stats.merge_from(candidate_stats.file_cache_stats);
+
+ 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;
+}
+
+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 previous_candidate_scan_rows, bool price_point_lookups,
int64_t max_candidate_keys,
+ size_t max_candidate_bytes, size_t candidate_workspace_bytes,
+ const CandidateScanCostLimit& cost_limit, CandidateKeyMap*
candidate_keys,
+ size_t* candidate_bytes, bool* limit_exceeded, bool* bytes_exceeded,
+ bool* reservation_exceeded, bool* cost_exceeded) {
+ DCHECK(candidate_keys != nullptr);
+ DCHECK(candidate_bytes != nullptr);
+ DCHECK(limit_exceeded != nullptr);
+ DCHECK(bytes_exceeded != nullptr);
+ DCHECK(reservation_exceeded != nullptr);
+ DCHECK(cost_exceeded != nullptr);
+ *candidate_bytes = 0;
+ *limit_exceeded = false;
+ *bytes_exceeded = false;
+ *reservation_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(candidate_workspace_bytes);
+ Defer account_candidate_stats {[&]() {
+ _merge_seq_map_candidate_stats(candidate_reader.stats(),
_tablet_reader->mutable_stats());
+ }};
+ 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));
+ }
+ auto encoded_key = _encode_candidate_key(key);
+ const int64_t reserved_bytes =
thread_context()->thread_mem_tracker_mgr->reserved_mem();
+ const size_t reservation_headroom =
+ reserved_bytes > cast_set<int64_t>(
+
MIN_SEQ_MAP_CANDIDATE_RESERVATION_HEADROOM_BYTES)
+ ? cast_set<size_t>(reserved_bytes -
+
MIN_SEQ_MAP_CANDIDATE_RESERVATION_HEADROOM_BYTES)
+ : 0;
+ const auto insert_result = _try_add_seq_map_candidate_key(
+ std::move(encoded_key), std::move(key), key_column_count,
max_candidate_bytes,
+ reservation_headroom, candidate_keys, candidate_bytes);
+ if (insert_result == CandidateKeyInsertResult::KEY_BYTES_LIMIT) {
+ *bytes_exceeded = true;
+ break;
+ }
+ if (insert_result == CandidateKeyInsertResult::RESERVATION_LIMIT) {
+ *reservation_exceeded = true;
+ break;
+ }
+ if (candidate_keys->size() >
static_cast<size_t>(max_candidate_keys)) {
+ *limit_exceeded = true;
+ break;
+ }
+ }
+ block.clear_column_data();
+ if (*limit_exceeded || *bytes_exceeded || *reservation_exceeded) {
+ break;
+ }
+ const size_t candidate_key_count = price_point_lookups ?
candidate_keys->size() : 0;
+ if (cost_limit.exceeded(previous_candidate_scan_rows,
+ candidate_reader.stats().raw_rows_read,
candidate_key_count)) {
+ *cost_exceeded = true;
+ break;
+ }
+ }
+ return Status::OK();
+}
+
+Status OlapScanner::_materialize_seq_map_point_keys(CandidateKeyMap*
candidate_keys,
+ size_t retained_bytes,
+ PointKeySetSPtr*
point_keys) {
+ DCHECK(candidate_keys != nullptr);
+ DCHECK(point_keys != nullptr);
+
+ const auto key_schema =
+
RowCursor::create_shared_schema(_tablet_reader_params.tablet_schema,
+
_tablet_reader_params.tablet_schema->num_key_columns());
+ auto mutable_point_keys = std::make_shared<PointKeySet>(key_schema);
+ mutable_point_keys->keys.reserve(candidate_keys->size());
+ for (auto& entry : *candidate_keys) {
+ RowCursor point_key;
+ RETURN_IF_ERROR(point_key.init(key_schema,
std::move(entry.second).release_fields()));
+ mutable_point_keys->keys.emplace_back(std::move(point_key));
+ }
+ std::sort(mutable_point_keys->keys.begin(), mutable_point_keys->keys.end(),
+ [](const RowCursor& lhs, const RowCursor& rhs) {
+ return compare_row_key(lhs, rhs) < 0;
+ });
+ mutable_point_keys->retained_bytes = retained_bytes;
+ *point_keys = std::move(mutable_point_keys);
+ return Status::OK();
+}
+
+bool OlapScanner::_is_candidate_memory_failure(const Status& status) {
+ return status.is<ErrorCode::MEM_LIMIT_EXCEEDED>() ||
status.is<ErrorCode::MEM_ALLOC_FAILED>() ||
+ status.is<ErrorCode::QUERY_MEMORY_EXCEEDED>() ||
+ status.is<ErrorCode::WORKLOAD_GROUP_MEMORY_EXCEEDED>() ||
+ status.is<ErrorCode::PROCESS_MEMORY_EXCEEDED>();
+}
+
+void OlapScanner::_record_seq_map_candidate_fallback_reason(RuntimeProfile*
profile,
+ const std::string&
fallback_reason) {
+ DCHECK(profile != nullptr);
+ DCHECK(!fallback_reason.empty());
+ profile->add_info_string("SeqMapCandidateFallbackReason." +
fallback_reason, fallback_reason);
+}
+
+Status OlapScanner::_build_seq_map_candidate_keys(
+ const std::vector<std::shared_ptr<ColumnPredicate>>& key_predicates,
+ const std::map<uint32_t,
std::vector<std::shared_ptr<ColumnPredicate>>>& group_drivers,
+ int64_t max_candidate_keys, const CandidateMemoryBudget& memory_budget,
+ const CandidateScanCostLimit& cost_limit) {
+ auto& params = _tablet_reader_params;
+ auto* stats = _tablet_reader->mutable_stats();
+
+ // Reserve key retention and reader workspace before either candidate map
starts allocating.
+ auto* mem_tracker_mgr = thread_context()->thread_mem_tracker_mgr.get();
+ auto inherited_reservation = mem_tracker_mgr->take_reserved_memory();
+ Defer restore_inherited_reservation {
+ [&] {
mem_tracker_mgr->adopt_reserved_memory(std::move(inherited_reservation)); }};
+ auto reserve_status =
+
mem_tracker_mgr->try_reserve(cast_set<int64_t>(memory_budget.reservation_bytes));
+ if (!reserve_status.ok()) {
+ ++stats->seq_map_candidate_fallbacks;
+ _seq_map_candidate_fallback_reason =
"candidate_key_memory_reservation";
+ return Status::OK();
+ }
+ DEFER_RELEASE_RESERVED();
Review Comment:
[P1] Include the downstream point-range bitmaps in the candidate memory
fallback. This reservation ends when candidate construction returns, but
BlockReader::init() then builds the merge heap and initialized segments expand
the same key set into RowRanges and retained sparse _row_bitmap state. With
100k keys across 100 overlapping segments, that is about 10 million RowRange
entries (roughly 160 MiB before vector capacity and roaring storage) outside
the reservation, even though the break-even model can still admit this on a
billion-row tablet. A memory failure here fails the scan, while the unchanged
full-range bitmap is compact. Please reserve/estimate through final-reader
initialization, or recreate the unoptimized reader on its memory-limit statuses.
--
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]
