Skip to content

[feature](scan) Add value predicate pushdown control for MOR tables #60513

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
dataroaring wants to merge 3 commits into
base: master
Choose a base branch
from
+535 −2
Open

[feature](scan) Add value predicate pushdown control for MOR tables #60513

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a06783e
[feature](scan) Add session variable to control value predicate pushd…
dataroaring Feb 4, 2026
7d4aa18
[feature](scan) Implement BE-side value predicate pushdown for MOR ta…
dataroaring Feb 5, 2026
b445c13
[fix](mor) Fix initialization order bug for delete sign predicate skip
dataroaring Feb 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion be/src/olap/rowset/beta_rowset_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -356,7 +356,8 @@ bool BetaRowsetReader::_should_push_down_value_predicates() const {
(((_rowset->start_version() == 0 || _rowset->start_version() == 2) &&
!_rowset->_rowset_meta->is_segments_overlapping() &&
_read_context->sequence_id_idx == -1) ||
_read_context->enable_unique_key_merge_on_write);
_read_context->enable_unique_key_merge_on_write ||
_read_context->enable_mor_value_predicate_pushdown);
}
#include "common/compile_check_end.h"
} // namespace doris
3 changes: 3 additions & 0 deletions be/src/olap/rowset/rowset_reader_context.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,9 @@ struct RowsetReaderContext {
std::shared_ptr<segment_v2::AnnTopNRuntime> ann_topn_runtime;

uint64_t condition_cache_digest = 0;

// When true, push down value predicates for MOR tables
bool enable_mor_value_predicate_pushdown = false;
};

} // namespace doris
Expand Down
10 changes: 10 additions & 0 deletions be/src/olap/tablet_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -208,6 +208,8 @@ Status TabletReader::_capture_rs_readers(const ReaderParams& read_params) {
_reader_context.merged_rows = &_merged_rows;
_reader_context.delete_bitmap = read_params.delete_bitmap;
_reader_context.enable_unique_key_merge_on_write = tablet()->enable_unique_key_merge_on_write();
_reader_context.enable_mor_value_predicate_pushdown =
read_params.enable_mor_value_predicate_pushdown;
_reader_context.record_rowids = read_params.record_rowids;
_reader_context.rowid_conversion = read_params.rowid_conversion;
_reader_context.is_key_column_group = read_params.is_key_column_group;
Expand Down Expand Up @@ -515,9 +517,17 @@ Status TabletReader::_init_conditions_param(const ReaderParams& read_params) {
}
}

int32_t delete_sign_idx = _tablet_schema->delete_sign_idx();
for (auto predicate : predicates) {
auto column = _tablet_schema->column(predicate->column_id());
if (column.aggregation() != FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE) {
// When MOR value predicate pushdown is enabled, drop __DORIS_DELETE_SIGN__
// from storage-layer predicates entirely. Delete sign must only be evaluated
// post-merge via VExpr to prevent deleted rows from reappearing.
if (read_params.enable_mor_value_predicate_pushdown && delete_sign_idx >= 0 &&
predicate->column_id() == static_cast<uint32_t>(delete_sign_idx)) {
continue;
}
_value_col_predicates.push_back(predicate);
} else {
_col_predicates.push_back(predicate);
Expand Down
3 changes: 3 additions & 0 deletions be/src/olap/tablet_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -191,6 +191,9 @@ class TabletReader {

bool is_segcompaction = false;

// Enable value predicate pushdown for MOR tables
bool enable_mor_value_predicate_pushdown = false;

std::vector<RowwiseIteratorUPtr>* segment_iters_ptr = nullptr;

void check_validation() const;
Expand Down
6 changes: 6 additions & 0 deletions be/src/pipeline/exec/olap_scan_operator.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -445,6 +445,12 @@ bool OlapScanLocalState::_storage_no_merge() {
p._olap_scan_node.enable_unique_key_merge_on_write));
}

bool OlapScanLocalState::_should_push_down_mor_value_predicate() {
auto& p = _parent->cast<OlapScanOperatorX>();
return p._olap_scan_node.__isset.enable_mor_value_predicate_pushdown &&
p._olap_scan_node.enable_mor_value_predicate_pushdown;
}

Status OlapScanLocalState::_init_scanners(std::list<vectorized::ScannerSPtr>* scanners) {
if (_scan_ranges.empty()) {
_eos = true;
Expand Down
2 changes: 2 additions & 0 deletions be/src/pipeline/exec/olap_scan_operator.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,8 @@ class OlapScanLocalState final : public ScanLocalState<OlapScanLocalState> {

bool _storage_no_merge() override;

bool _should_push_down_mor_value_predicate() override;

bool _push_down_topn(const vectorized::RuntimePredicate& predicate) override {
if (!predicate.target_is_slot(_parent->node_id())) {
return false;
Expand Down
2 changes: 1 addition & 1 deletion be/src/pipeline/exec/scan_operator.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -439,7 +439,7 @@ Status ScanLocalState<Derived>::_normalize_predicate(vectorized::VExprContext* c
return Status::OK();
}

if (pdt == PushDownType::ACCEPTABLE && (_is_key_column(slot->col_name()))) {
if (pdt == PushDownType::ACCEPTABLE && _is_key_column(slot->col_name())) {
output_expr = nullptr;
return Status::OK();
} else {
Expand Down
1 change: 1 addition & 0 deletions be/src/pipeline/exec/scan_operator.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -205,6 +205,7 @@ class ScanLocalState : public ScanLocalStateBase {
virtual bool _storage_no_merge() { return false; }
virtual bool _push_down_topn(const vectorized::RuntimePredicate& predicate) { return false; }
virtual bool _is_key_column(const std::string& col_name) { return false; }
virtual bool _should_push_down_mor_value_predicate() { return false; }
virtual PushDownType _should_push_down_bloom_filter() const {
return PushDownType::UNACCEPTABLE;
}
Expand Down
7 changes: 7 additions & 0 deletions be/src/vec/exec/scan/olap_scanner.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -480,6 +480,13 @@ Status OlapScanner::_init_tablet_reader_params(
if (!_state->skip_storage_engine_merge()) {
auto* olap_scan_local_state = (pipeline::OlapScanLocalState*)_local_state;
TOlapScanNode& olap_scan_node = olap_scan_local_state->olap_scan_node();

// Set MOR value predicate pushdown flag
if (olap_scan_node.__isset.enable_mor_value_predicate_pushdown &&
olap_scan_node.enable_mor_value_predicate_pushdown) {
_tablet_reader_params.enable_mor_value_predicate_pushdown = true;
}

// order by table keys optimization for topn
// will only read head/tail of data file since it's already sorted by keys
if (olap_scan_node.__isset.sort_info && !olap_scan_node.sort_info.is_asc_order.empty()) {
Expand Down
8 changes: 8 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3022,6 +3022,14 @@ public boolean isUniqKeyMergeOnWrite() {
return getKeysType() == KeysType.UNIQUE_KEYS && getEnableUniqueKeyMergeOnWrite();
}

/**
* Check if this is a MOR (Merge-On-Read) table.
* MOR = UNIQUE_KEYS without merge-on-write enabled.
*/
public boolean isMorTable() {
return getKeysType() == KeysType.UNIQUE_KEYS && !getEnableUniqueKeyMergeOnWrite();
Comment on lines +3028 to +3030
Copy link

Copilot AI Feb 4, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The isMorTable() method definition appears correct, but consider verifying the behavior when tableProperty is null. According to getEnableUniqueKeyMergeOnWrite() (lines 3011-3019), it returns false when tableProperty is null, which means a UNIQUE_KEYS table with null tableProperty would be considered a MOR table. Ensure this is the intended behavior for all edge cases.

Suggested change
*/
public boolean isMorTable() {
return getKeysType() == KeysType.UNIQUE_KEYS && !getEnableUniqueKeyMergeOnWrite();
* Only tables with non-null tableProperty are considered for MOR.
*/
public boolean isMorTable() {
return tableProperty != null
&& getKeysType() == KeysType.UNIQUE_KEYS
&& !getEnableUniqueKeyMergeOnWrite();

Copilot uses AI. Check for mistakes.
}

public boolean isUniqKeyMergeOnWriteWithClusterKeys() {
return isUniqKeyMergeOnWrite() && getBaseSchema().stream().anyMatch(Column::isClusterKey);
}
Expand Down
9 changes: 9 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1181,6 +1181,15 @@ protected void toThrift(TPlanNode msg) {
msg.olap_scan_node.setTableName(tableName);
msg.olap_scan_node.setEnableUniqueKeyMergeOnWrite(olapTable.getEnableUniqueKeyMergeOnWrite());

// Set MOR value predicate pushdown flag based on session variable
if (olapTable.isMorTable() && ConnectContext.get() != null) {
String dbName = olapTable.getQualifiedDbName();
Copy link

Copilot AI Feb 4, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Potential null pointer exception. The method getQualifiedDbName() can return null (as seen in Table.java line 367-369). This null value would then be passed to isMorValuePredicatePushdownEnabled() which concatenates it with the table name on line 4691, potentially resulting in "null.tableName". Consider adding a null check for dbName or using an alternative method like getDBName() which handles the null case.

Suggested change
String dbName = olapTable.getQualifiedDbName();
String dbName = olapTable.getQualifiedDbName();
if (dbName == null) {
dbName = olapTable.getDBName();
}

Copilot uses AI. Check for mistakes.
String tblName = olapTable.getName();
boolean enabled = ConnectContext.get().getSessionVariable()
.isMorValuePredicatePushdownEnabled(dbName, tblName);
msg.olap_scan_node.setEnableMorValuePredicatePushdown(enabled);
}

msg.setPushDownAggTypeOpt(pushDownAggNoGroupingOp);

msg.olap_scan_node.setPushDownAggTypeOpt(pushDownAggNoGroupingOp);
Expand Down
39 changes: 39 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -722,6 +722,9 @@ public class SessionVariable implements Serializable, Writable {

public static final String ENABLE_PUSHDOWN_STRING_MINMAX = "enable_pushdown_string_minmax";

public static final String ENABLE_MOR_VALUE_PREDICATE_PUSHDOWN_TABLES
= "enable_mor_value_predicate_pushdown_tables";

// When set use fix replica = true, the fixed replica maybe bad, try to use the health one if
// this session variable is set to true.
public static final String FALLBACK_OTHER_REPLICA_WHEN_FIXED_CORRUPT = "fallback_other_replica_when_fixed_corrupt";
Expand Down Expand Up @@ -2181,6 +2184,13 @@ public boolean isEnableHboNonStrictMatchingMode() {
"是否启用 string 类型 min max 下推。", "Set whether to enable push down string type minmax."})
public boolean enablePushDownStringMinMax = false;

// Comma-separated list of MOR tables to enable value predicate pushdown.
@VariableMgr.VarAttr(name = ENABLE_MOR_VALUE_PREDICATE_PUSHDOWN_TABLES, needForward = true, description = {
"指定启用MOR表value列谓词下推的表列表，格式：db1.tbl1,db2.tbl2 或 * 表示所有MOR表。",
"Comma-separated list of MOR tables to enable value predicate pushdown. "
+ "Format: db1.tbl1,db2.tbl2 or * for all MOR tables."})
public String enableMorValuePredicatePushdownTables = "";

// Whether drop table when create table as select insert data appear error.
@VariableMgr.VarAttr(name = DROP_TABLE_IF_CTAS_FAILED, needForward = true)
public boolean dropTableIfCtasFailed = true;
Expand Down Expand Up @@ -4659,6 +4669,35 @@ public boolean isEnablePushDownStringMinMax() {
return enablePushDownStringMinMax;
}

public String getEnableMorValuePredicatePushdownTables() {
return enableMorValuePredicatePushdownTables;
}

/**
* Check if a table is enabled for MOR value predicate pushdown.
* @param dbName database name
* @param tableName table name
* @return true if the table is in the enabled list or if '*' is set
*/
public boolean isMorValuePredicatePushdownEnabled(String dbName, String tableName) {
if (enableMorValuePredicatePushdownTables == null
|| enableMorValuePredicatePushdownTables.isEmpty()) {
return false;
}
String trimmed = enableMorValuePredicatePushdownTables.trim();
if ("*".equals(trimmed)) {
return true;
}
Copy link

Copilot AI Feb 4, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The method doesn't handle null dbName parameter. If dbName is null, line 4691 will create "null.tableName" which could lead to unexpected behavior. Consider adding a null check for dbName at the beginning of the method, or using Objects.requireNonNull() to fail fast, or handle the null case explicitly by using only tableName for comparison when dbName is null.

Suggested change
}
}
// When dbName is null, only compare using tableName to avoid creating "null.tableName".
if (dbName == null) {
for (String table : trimmed.split(",")) {
if (table.trim().equalsIgnoreCase(tableName)) {
return true;
}
}
return false;
}

Copilot uses AI. Check for mistakes.
String fullName = dbName + "." + tableName;
for (String table : trimmed.split(",")) {
if (table.trim().equalsIgnoreCase(fullName)
|| table.trim().equalsIgnoreCase(tableName)) {
Comment on lines +4693 to +4694
Copy link

Copilot AI Feb 4, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The method doesn't handle edge cases well when splitting by comma. If the variable contains only commas or has consecutive commas (e.g., "db.table1,,db.table2"), the split operation will produce empty strings. Calling trim() on empty strings and comparing them could lead to false positives. Consider filtering out empty strings after trimming, or using a more robust parsing approach.

Suggested change
if (table.trim().equalsIgnoreCase(fullName)
|| table.trim().equalsIgnoreCase(tableName)) {
String normalized = table.trim();
if (normalized.isEmpty()) {
continue;
}
if (normalized.equalsIgnoreCase(fullName)
|| normalized.equalsIgnoreCase(tableName)) {

Copilot uses AI. Check for mistakes.
return true;
}
}
return false;
}
Comment on lines +4682 to +4699
Copy link

Copilot AI Feb 4, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The new session variable and its helper method isMorValuePredicatePushdownEnabled() lack test coverage. Consider adding unit tests in SessionVariablesTest to verify:

  1. Empty string handling (default behavior)
  2. Wildcard '*' behavior
  3. Single table name matching (with and without database prefix)
  4. Multiple table names (comma-separated)
  5. Case-insensitive matching
  6. Whitespace handling in table names
  7. Edge cases like null dbName parameter

Copilot uses AI. Check for mistakes.

/** canUseNereidsDistributePlanner */
public static boolean canUseNereidsDistributePlanner() {
ConnectContext connectContext = ConnectContext.get();
Expand Down
2 changes: 2 additions & 0 deletions gensrc/thrift/PlanNodes.thrift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -887,6 +887,8 @@ struct TOlapScanNode {
20: optional i64 score_sort_limit
21: optional TSortInfo ann_sort_info
22: optional i64 ann_sort_limit
// Enable value predicate pushdown for MOR tables
23: optional bool enable_mor_value_predicate_pushdown
}

struct TEqJoinCondition {
Expand Down
76 changes: 76 additions & 0 deletions regression-test/data/data_model_p0/unique/test_mor_value_predicate_pushdown.out
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,76 @@
-- This file is automatically generated. You should know what you did if you want to edit this
-- !select_disabled --
2 200 world
3 300 test

-- !select_enabled_tablename --
2 200 world
3 300 test

-- !select_enabled_fullname --
2 200 world
3 300 test

-- !select_enabled_wildcard --
2 200 world
3 300 test

-- !select_eq_predicate --
2 200 world

-- !select_not_in_list --
2 200 world
3 300 test

-- !select_dedup_all --
1 100 first
2 300 third
3 500 fifth

-- !select_dedup_eq --
2 300 third

-- !select_dedup_none --

-- !select_delete_range --
3 300 test

-- !select_delete_eq --

-- !select_delete_all --
1 100 hello
3 300 test

-- !select_delpred_range --
3 300 test

-- !select_delpred_eq --

-- !select_delpred_all --
1 100 hello
3 300 test

-- !select_update_disabled_old --

-- !select_update_disabled_new --
1 500 new

-- !select_update_enabled_old --
1 100 old

-- !select_update_enabled_new --
1 500 new

-- !select_update_enabled_range --
1 500 new
3 300 keep

-- !select_multiple_tables --
2 200

-- !select_mow_table --
2 200

-- !select_dup_table --
2 200

Loading
Loading