You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by da...@apache.org on 2026/03/04 03:33:49 UTC
(doris) branch feature/mor_value_predicate_pushdown_control created (now 1c6db7775dd)
This is an automated email from the ASF dual-hosted git repository.
dataroaring pushed a change to branch feature/mor_value_predicate_pushdown_control
in repository https://gitbox.apache.org/repos/asf/doris.git
at 1c6db7775dd [fix](scan) Refactor table list matching to use TableNameInfo-style component parsing
This branch includes the following new commits:
new 5e5179e584e [feature](scan) Add session variable to control value predicate pushdown for MOR tables
new 4a0c30ea7f0 [feature](scan) Implement BE-side value predicate pushdown for MOR tables
new 89d3715b3aa [feature](scan) Add read_mor_as_dup_tables session variable to read MOR tables without merge
new 1c6db7775dd [fix](scan) Refactor table list matching to use TableNameInfo-style component parsing
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 04/04: [fix](scan) Refactor table list matching to use TableNameInfo-style component parsing
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dataroaring pushed a commit to branch feature/mor_value_predicate_pushdown_control
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 1c6db7775dde85d603a2291fda9366d1e3fb8ed2
Author: Yongqiang YANG <da...@gmail.com>
AuthorDate: Tue Mar 3 19:33:26 2026 -0800
[fix](scan) Refactor table list matching to use TableNameInfo-style component parsing
Address review feedback: replace naive string concatenation matching with
component-based parsing (split by ".") consistent with TableNameInfo, fixing
null dbName handling and empty entry edge cases.
Co-Authored-By: Claude Opus 4.6 <no...@anthropic.com>
---
.../java/org/apache/doris/qe/SessionVariable.java | 58 +++++++++++-----------
.../nereids/rules/analysis/ReadMorAsDupTest.java | 2 +-
2 files changed, 31 insertions(+), 29 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index 42aef65243d..71dc247469a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -4794,43 +4794,45 @@ public class SessionVariable implements Serializable, Writable {
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;
- }
- String fullName = dbName + "." + tableName;
- for (String table : trimmed.split(",")) {
- if (table.trim().equalsIgnoreCase(fullName)
- || table.trim().equalsIgnoreCase(tableName)) {
- return true;
- }
- }
- return false;
+ return isTableInList(enableMorValuePredicatePushdownTables, dbName, tableName);
}
public boolean isReadMorAsDupEnabled(String dbName, String tableName) {
- if (readMorAsDupTables == null || readMorAsDupTables.isEmpty()) {
+ return isTableInList(readMorAsDupTables, dbName, tableName);
+ }
+
+ /**
+ * Check if a table matches any entry in a comma-separated table list.
+ * Parses entries the same way as TableNameInfo: split by "." to extract
+ * component parts (table, db.table, or ctl.db.table).
+ * When entry specifies db, both db and table must match.
+ * When entry is just a table name, it matches any database.
+ */
+ private static boolean isTableInList(String tableList, String dbName, String tableName) {
+ if (tableList == null || tableList.isEmpty()) {
return false;
}
- String trimmed = readMorAsDupTables.trim();
+ String trimmed = tableList.trim();
if ("*".equals(trimmed)) {
return true;
}
- String fullName = dbName + "." + tableName;
- for (String table : trimmed.split(",")) {
- if (table.trim().equalsIgnoreCase(fullName)
- || table.trim().equalsIgnoreCase(tableName)) {
+ for (String entry : trimmed.split(",")) {
+ String trimmedEntry = entry.trim();
+ if (trimmedEntry.isEmpty()) {
+ continue;
+ }
+ String[] parts = trimmedEntry.split("\\.");
+ String entryTbl = parts[parts.length - 1];
+ String entryDb = parts.length >= 2 ? parts[parts.length - 2] : null;
+ if (!entryTbl.equalsIgnoreCase(tableName)) {
+ continue;
+ }
+ if (entryDb != null) {
+ if (dbName != null && entryDb.equalsIgnoreCase(dbName)) {
+ return true;
+ }
+ } else {
return true;
}
}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/ReadMorAsDupTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/ReadMorAsDupTest.java
index 579ae3c527f..1c3f687c8ab 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/ReadMorAsDupTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/ReadMorAsDupTest.java
@@ -227,7 +227,7 @@ class ReadMorAsDupTest extends TestWithFeService implements GeneratedPlanPattern
connectContext.getSessionVariable().isReadMorAsDupEnabled("mydb", "mytbl"));
Assertions.assertFalse(
connectContext.getSessionVariable().isReadMorAsDupEnabled("otherdb", "othertbl"));
- // "mydb.mytbl" entry only matches full name or exact table name "mydb.mytbl"
+ // "mydb.mytbl" entry requires both db and table components to match
Assertions.assertFalse(
connectContext.getSessionVariable().isReadMorAsDupEnabled("anything", "mytbl"));
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 03/04: [feature](scan) Add read_mor_as_dup_tables session variable to read MOR tables without merge
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dataroaring pushed a commit to branch feature/mor_value_predicate_pushdown_control
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 89d3715b3aa8180f724a0eae14049942a6af29d6
Author: Yongqiang YANG <da...@gmail.com>
AuthorDate: Sun Feb 8 19:57:23 2026 -0800
[feature](scan) Add read_mor_as_dup_tables session variable to read MOR tables without merge
Add a per-table session variable that allows reading MOR (Merge-On-Read)
UNIQUE tables as DUP tables, skipping storage engine merge and delete
sign filter while still honoring delete predicates.
Co-Authored-By: Claude Opus 4.6 <no...@anthropic.com>
---
be/src/pipeline/exec/olap_scan_operator.cpp | 8 +-
be/src/pipeline/exec/olap_scan_operator.h | 1 +
be/src/vec/exec/scan/olap_scanner.cpp | 5 +-
.../doris/nereids/rules/analysis/BindRelation.java | 5 +-
.../nereids/rules/rewrite/SetPreAggStatus.java | 9 +-
.../trees/plans/logical/LogicalOlapScan.java | 7 +
.../org/apache/doris/planner/OlapScanNode.java | 4 +
.../java/org/apache/doris/qe/SessionVariable.java | 28 +++
.../nereids/rules/analysis/ReadMorAsDupTest.java | 280 +++++++++++++++++++++
gensrc/thrift/PlanNodes.thrift | 2 +
.../unique_with_mor_p0/test_read_mor_as_dup.groovy | 119 +++++++++
11 files changed, 464 insertions(+), 4 deletions(-)
diff --git a/be/src/pipeline/exec/olap_scan_operator.cpp b/be/src/pipeline/exec/olap_scan_operator.cpp
index 07c0522a4b5..b19f0d23cfe 100644
--- a/be/src/pipeline/exec/olap_scan_operator.cpp
+++ b/be/src/pipeline/exec/olap_scan_operator.cpp
@@ -456,7 +456,13 @@ bool OlapScanLocalState::_storage_no_merge() {
return (p._olap_scan_node.keyType == TKeysType::DUP_KEYS ||
(p._olap_scan_node.keyType == TKeysType::UNIQUE_KEYS &&
p._olap_scan_node.__isset.enable_unique_key_merge_on_write &&
- p._olap_scan_node.enable_unique_key_merge_on_write));
+ p._olap_scan_node.enable_unique_key_merge_on_write)) ||
+ _read_mor_as_dup();
+}
+
+bool OlapScanLocalState::_read_mor_as_dup() {
+ auto& p = _parent->cast<OlapScanOperatorX>();
+ return p._olap_scan_node.__isset.read_mor_as_dup && p._olap_scan_node.read_mor_as_dup;
}
Status OlapScanLocalState::_init_scanners(std::list<vectorized::ScannerSPtr>* scanners) {
diff --git a/be/src/pipeline/exec/olap_scan_operator.h b/be/src/pipeline/exec/olap_scan_operator.h
index 0577dd1ff2d..6517b13f271 100644
--- a/be/src/pipeline/exec/olap_scan_operator.h
+++ b/be/src/pipeline/exec/olap_scan_operator.h
@@ -105,6 +105,7 @@ private:
bool _storage_no_merge() override;
+ bool _read_mor_as_dup();
bool _push_down_topn(const vectorized::RuntimePredicate& predicate) override {
if (!predicate.target_is_slot(_parent->node_id())) {
return false;
diff --git a/be/src/vec/exec/scan/olap_scanner.cpp b/be/src/vec/exec/scan/olap_scanner.cpp
index 7b577bdb629..2bd7b0f3269 100644
--- a/be/src/vec/exec/scan/olap_scanner.cpp
+++ b/be/src/vec/exec/scan/olap_scanner.cpp
@@ -334,7 +334,10 @@ Status OlapScanner::_init_tablet_reader_params(
// if the table with rowset [0-x] or [0-1] [2-y], and [0-1] is empty
const bool single_version = _tablet_reader_params.has_single_version();
- if (_state->skip_storage_engine_merge()) {
+ auto* olap_local_state = static_cast<pipeline::OlapScanLocalState*>(_local_state);
+ bool read_mor_as_dup = olap_local_state->olap_scan_node().__isset.read_mor_as_dup &&
+ olap_local_state->olap_scan_node().read_mor_as_dup;
+ if (_state->skip_storage_engine_merge() || read_mor_as_dup) {
_tablet_reader_params.direct_mode = true;
_tablet_reader_params.aggregation = true;
} else {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
index 2d11e1c75db..81a54f8c67e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
@@ -371,7 +371,10 @@ public class BindRelation extends OneAnalysisRuleFactory {
public static LogicalPlan checkAndAddDeleteSignFilter(LogicalOlapScan scan, ConnectContext connectContext,
OlapTable olapTable) {
if (!Util.showHiddenColumns() && scan.getTable().hasDeleteSign()
- && !connectContext.getSessionVariable().skipDeleteSign()) {
+ && !connectContext.getSessionVariable().skipDeleteSign()
+ && !(olapTable.isMorTable()
+ && connectContext.getSessionVariable().isReadMorAsDupEnabled(
+ olapTable.getQualifiedDbName(), olapTable.getName()))) {
// table qualifier is catalog.db.table, we make db.table.column
Slot deleteSlot = null;
for (Slot slot : scan.getOutput()) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/SetPreAggStatus.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/SetPreAggStatus.java
index 2f7743cb86f..7ae49fc6ec8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/SetPreAggStatus.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/SetPreAggStatus.java
@@ -53,6 +53,7 @@ import org.apache.doris.nereids.trees.plans.logical.LogicalRepeat;
import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.qe.ConnectContext;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
@@ -151,7 +152,13 @@ public class SetPreAggStatus extends DefaultPlanRewriter<Stack<SetPreAggStatus.P
long selectIndexId = logicalOlapScan.getSelectedIndexId();
MaterializedIndexMeta meta = logicalOlapScan.getTable().getIndexMetaByIndexId(selectIndexId);
if (meta.getKeysType() == KeysType.DUP_KEYS || (meta.getKeysType() == KeysType.UNIQUE_KEYS
- && logicalOlapScan.getTable().getEnableUniqueKeyMergeOnWrite())) {
+ && logicalOlapScan.getTable().getEnableUniqueKeyMergeOnWrite())
+ || (meta.getKeysType() == KeysType.UNIQUE_KEYS
+ && logicalOlapScan.getTable().isMorTable()
+ && ConnectContext.get() != null
+ && ConnectContext.get().getSessionVariable().isReadMorAsDupEnabled(
+ logicalOlapScan.getTable().getQualifiedDbName(),
+ logicalOlapScan.getTable().getName()))) {
return logicalOlapScan.withPreAggStatus(PreAggStatus.on());
} else {
if (context.empty()) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapScan.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapScan.java
index bbcc44a38fc..d03a7d1b35b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapScan.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapScan.java
@@ -769,6 +769,13 @@ public class LogicalOlapScan extends LogicalCatalogRelation implements OlapScan,
&& getTable().getKeysType() == KeysType.UNIQUE_KEYS) {
return;
}
+ // When readMorAsDup is enabled, MOR tables are read as DUP, so uniqueness cannot be guaranteed.
+ if (getTable().getKeysType() == KeysType.UNIQUE_KEYS
+ && getTable().isMorTable()
+ && ConnectContext.get().getSessionVariable().isReadMorAsDupEnabled(
+ getTable().getQualifiedDbName(), getTable().getName())) {
+ return;
+ }
ImmutableSet.Builder<Slot> uniqSlots = ImmutableSet.builderWithExpectedSize(outputSet.size());
for (Slot slot : outputSet) {
if (!(slot instanceof SlotReference)) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
index 9d4b6fddfc5..48614966468 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
@@ -1197,6 +1197,10 @@ public class OlapScanNode extends ScanNode {
boolean enabled = ConnectContext.get().getSessionVariable()
.isMorValuePredicatePushdownEnabled(dbName, tblName);
msg.olap_scan_node.setEnableMorValuePredicatePushdown(enabled);
+ if (ConnectContext.get().getSessionVariable()
+ .isReadMorAsDupEnabled(dbName, tblName)) {
+ msg.olap_scan_node.setReadMorAsDup(true);
+ }
}
msg.setPushDownAggTypeOpt(pushDownAggNoGroupingOp);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index a6e581199d5..42aef65243d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -735,6 +735,8 @@ public class SessionVariable implements Serializable, Writable {
public static final String ENABLE_MOR_VALUE_PREDICATE_PUSHDOWN_TABLES
= "enable_mor_value_predicate_pushdown_tables";
+ public static final String READ_MOR_AS_DUP_TABLES = "read_mor_as_dup_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";
@@ -2232,6 +2234,14 @@ public class SessionVariable implements Serializable, Writable {
+ "Format: db1.tbl1,db2.tbl2 or * for all MOR tables."})
public String enableMorValuePredicatePushdownTables = "";
+ // Comma-separated list of MOR tables to read as DUP (skip merge, skip delete sign filter).
+ @VariableMgr.VarAttr(name = READ_MOR_AS_DUP_TABLES, needForward = true,
+ affectQueryResultInPlan = true, description = {
+ "指定以DUP模式读取MOR表的表列表（跳过合并和删除标记过滤），格式：db1.tbl1,db2.tbl2 或 * 表示所有MOR表。",
+ "Comma-separated list of MOR tables to read as DUP (skip merge, skip delete sign filter). "
+ + "Format: db1.tbl1,db2.tbl2 or * for all MOR tables."})
+ public String readMorAsDupTables = "";
+
// 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;
@@ -4809,6 +4819,24 @@ public class SessionVariable implements Serializable, Writable {
return false;
}
+ public boolean isReadMorAsDupEnabled(String dbName, String tableName) {
+ if (readMorAsDupTables == null || readMorAsDupTables.isEmpty()) {
+ return false;
+ }
+ String trimmed = readMorAsDupTables.trim();
+ if ("*".equals(trimmed)) {
+ return true;
+ }
+ String fullName = dbName + "." + tableName;
+ for (String table : trimmed.split(",")) {
+ if (table.trim().equalsIgnoreCase(fullName)
+ || table.trim().equalsIgnoreCase(tableName)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
/** canUseNereidsDistributePlanner */
public static boolean canUseNereidsDistributePlanner() {
ConnectContext connectContext = ConnectContext.get();
diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/ReadMorAsDupTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/ReadMorAsDupTest.java
new file mode 100644
index 00000000000..579ae3c527f
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/ReadMorAsDupTest.java
@@ -0,0 +1,280 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.analysis;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.nereids.pattern.GeneratedPlanPatterns;
+import org.apache.doris.nereids.rules.RulePromise;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.util.PlanChecker;
+import org.apache.doris.utframe.TestWithFeService;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Set;
+
+class ReadMorAsDupTest extends TestWithFeService implements GeneratedPlanPatterns {
+ private static final String DB = "test_read_mor_as_dup_db";
+
+ @Override
+ protected void runBeforeAll() throws Exception {
+ createDatabase(DB);
+ connectContext.setDatabase(DEFAULT_CLUSTER_PREFIX + DB);
+
+ // MOR UNIQUE table (enable_unique_key_merge_on_write = false)
+ createTable("CREATE TABLE " + DB + ".mor_tbl (\n"
+ + " k INT NOT NULL,\n"
+ + " v1 INT,\n"
+ + " v2 VARCHAR(100)\n"
+ + ") ENGINE=OLAP\n"
+ + "UNIQUE KEY(k)\n"
+ + "DISTRIBUTED BY HASH(k) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1', 'enable_unique_key_merge_on_write' = 'false');");
+
+ // MOW UNIQUE table (should not be affected)
+ createTable("CREATE TABLE " + DB + ".mow_tbl (\n"
+ + " k INT NOT NULL,\n"
+ + " v1 INT\n"
+ + ") ENGINE=OLAP\n"
+ + "UNIQUE KEY(k)\n"
+ + "DISTRIBUTED BY HASH(k) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1', 'enable_unique_key_merge_on_write' = 'true');");
+
+ // DUP table (should not be affected)
+ createTable("CREATE TABLE " + DB + ".dup_tbl (\n"
+ + " k INT NOT NULL,\n"
+ + " v1 INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(k)\n"
+ + "DISTRIBUTED BY HASH(k) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1');");
+
+ connectContext.getSessionVariable().setDisableNereidsRules("PRUNE_EMPTY_PARTITION");
+ }
+
+ @Test
+ void testDeleteSignFilterPresentByDefault() {
+ connectContext.getSessionVariable().readMorAsDupTables = "";
+
+ PlanChecker.from(connectContext)
+ .analyze("select * from mor_tbl")
+ .rewrite()
+ .matches(
+ logicalProject(
+ logicalFilter(
+ logicalOlapScan()
+ ).when(filter -> hasDeleteSignPredicate(filter.getConjuncts()))
+ )
+ );
+ }
+
+ @Test
+ void testDeleteSignFilterSkippedWithWildcard() {
+ connectContext.getSessionVariable().readMorAsDupTables = "*";
+
+ try {
+ PlanChecker.from(connectContext)
+ .analyze("select * from mor_tbl")
+ .rewrite()
+ .nonMatch(
+ logicalFilter(
+ logicalOlapScan()
+ ).when(filter -> hasDeleteSignPredicate(filter.getConjuncts()))
+ );
+ } finally {
+ connectContext.getSessionVariable().readMorAsDupTables = "";
+ }
+ }
+
+ @Test
+ void testDeleteSignFilterSkippedWithTableName() {
+ connectContext.getSessionVariable().readMorAsDupTables = "mor_tbl";
+
+ try {
+ PlanChecker.from(connectContext)
+ .analyze("select * from mor_tbl")
+ .rewrite()
+ .nonMatch(
+ logicalFilter(
+ logicalOlapScan()
+ ).when(filter -> hasDeleteSignPredicate(filter.getConjuncts()))
+ );
+ } finally {
+ connectContext.getSessionVariable().readMorAsDupTables = "";
+ }
+ }
+
+ @Test
+ void testDeleteSignFilterSkippedWithDbTableName() {
+ connectContext.getSessionVariable().readMorAsDupTables =
+ DEFAULT_CLUSTER_PREFIX + DB + ".mor_tbl";
+
+ try {
+ PlanChecker.from(connectContext)
+ .analyze("select * from mor_tbl")
+ .rewrite()
+ .nonMatch(
+ logicalFilter(
+ logicalOlapScan()
+ ).when(filter -> hasDeleteSignPredicate(filter.getConjuncts()))
+ );
+ } finally {
+ connectContext.getSessionVariable().readMorAsDupTables = "";
+ }
+ }
+
+ @Test
+ void testMowTableNotAffected() {
+ // MOW table should still have delete sign filter even with read_mor_as_dup = *
+ connectContext.getSessionVariable().readMorAsDupTables = "*";
+
+ try {
+ // MOW tables also have delete sign filter; read_mor_as_dup should NOT remove it
+ PlanChecker.from(connectContext)
+ .analyze("select * from mow_tbl")
+ .rewrite()
+ .matches(
+ logicalProject(
+ logicalFilter(
+ logicalOlapScan()
+ ).when(filter -> hasDeleteSignPredicate(filter.getConjuncts()))
+ )
+ );
+ } finally {
+ connectContext.getSessionVariable().readMorAsDupTables = "";
+ }
+ }
+
+ @Test
+ void testPreAggOnForMorWithReadAsDup() {
+ connectContext.getSessionVariable().readMorAsDupTables = "*";
+
+ try {
+ Plan plan = PlanChecker.from(connectContext)
+ .analyze("select * from mor_tbl")
+ .rewrite()
+ .getPlan();
+
+ // Find the LogicalOlapScan and verify preAggStatus is ON
+ LogicalOlapScan scan = findOlapScan(plan);
+ Assertions.assertNotNull(scan, "Should find LogicalOlapScan in plan");
+ Assertions.assertTrue(scan.getPreAggStatus().isOn(),
+ "PreAggStatus should be ON when read_mor_as_dup is enabled");
+ } finally {
+ connectContext.getSessionVariable().readMorAsDupTables = "";
+ }
+ }
+
+ @Test
+ void testPerTableControlOnlyAffectsSpecifiedTable() {
+ // Only enable for mor_tbl, not for other tables
+ connectContext.getSessionVariable().readMorAsDupTables = "mor_tbl";
+
+ try {
+ // mor_tbl should NOT have delete sign filter
+ PlanChecker.from(connectContext)
+ .analyze("select * from mor_tbl")
+ .rewrite()
+ .nonMatch(
+ logicalFilter(
+ logicalOlapScan()
+ ).when(filter -> hasDeleteSignPredicate(filter.getConjuncts()))
+ );
+
+ // dup_tbl should be unaffected (no delete sign filter for DUP)
+ PlanChecker.from(connectContext)
+ .analyze("select * from dup_tbl")
+ .rewrite()
+ .matches(logicalOlapScan());
+ } finally {
+ connectContext.getSessionVariable().readMorAsDupTables = "";
+ }
+ }
+
+ @Test
+ void testSessionVariableHelperMethod() {
+ // Test the isReadMorAsDupEnabled helper directly
+ connectContext.getSessionVariable().readMorAsDupTables = "";
+ Assertions.assertFalse(
+ connectContext.getSessionVariable().isReadMorAsDupEnabled("db", "tbl"));
+
+ connectContext.getSessionVariable().readMorAsDupTables = "*";
+ Assertions.assertTrue(
+ connectContext.getSessionVariable().isReadMorAsDupEnabled("db", "tbl"));
+ Assertions.assertTrue(
+ connectContext.getSessionVariable().isReadMorAsDupEnabled("any_db", "any_tbl"));
+
+ connectContext.getSessionVariable().readMorAsDupTables = "mydb.mytbl";
+ Assertions.assertTrue(
+ connectContext.getSessionVariable().isReadMorAsDupEnabled("mydb", "mytbl"));
+ Assertions.assertFalse(
+ connectContext.getSessionVariable().isReadMorAsDupEnabled("otherdb", "othertbl"));
+ // "mydb.mytbl" entry only matches full name or exact table name "mydb.mytbl"
+ Assertions.assertFalse(
+ connectContext.getSessionVariable().isReadMorAsDupEnabled("anything", "mytbl"));
+
+ // Table name only (no db prefix) matches any db
+ connectContext.getSessionVariable().readMorAsDupTables = "mytbl";
+ Assertions.assertTrue(
+ connectContext.getSessionVariable().isReadMorAsDupEnabled("anydb", "mytbl"));
+ Assertions.assertFalse(
+ connectContext.getSessionVariable().isReadMorAsDupEnabled("anydb", "othertbl"));
+
+ connectContext.getSessionVariable().readMorAsDupTables = "db1.tbl1,db2.tbl2";
+ Assertions.assertTrue(
+ connectContext.getSessionVariable().isReadMorAsDupEnabled("db1", "tbl1"));
+ Assertions.assertTrue(
+ connectContext.getSessionVariable().isReadMorAsDupEnabled("db2", "tbl2"));
+ Assertions.assertFalse(
+ connectContext.getSessionVariable().isReadMorAsDupEnabled("db1", "tbl2"));
+
+ // Case insensitive
+ connectContext.getSessionVariable().readMorAsDupTables = "MyDB.MyTbl";
+ Assertions.assertTrue(
+ connectContext.getSessionVariable().isReadMorAsDupEnabled("mydb", "mytbl"));
+
+ // Cleanup
+ connectContext.getSessionVariable().readMorAsDupTables = "";
+ }
+
+ private boolean hasDeleteSignPredicate(Set<Expression> conjuncts) {
+ return conjuncts.stream()
+ .anyMatch(expr -> expr.toSql().contains(Column.DELETE_SIGN));
+ }
+
+ private LogicalOlapScan findOlapScan(Plan plan) {
+ if (plan instanceof LogicalOlapScan) {
+ return (LogicalOlapScan) plan;
+ }
+ for (Plan child : plan.children()) {
+ LogicalOlapScan result = findOlapScan(child);
+ if (result != null) {
+ return result;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public RulePromise defaultPromise() {
+ return RulePromise.REWRITE;
+ }
+}
diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift
index 7fcb1912c67..9c6320c86e8 100644
--- a/gensrc/thrift/PlanNodes.thrift
+++ b/gensrc/thrift/PlanNodes.thrift
@@ -896,6 +896,8 @@ struct TOlapScanNode {
23: optional TScoreRangeInfo score_range_info
// Enable value predicate pushdown for MOR tables
24: optional bool enable_mor_value_predicate_pushdown
+ // Read MOR table as DUP table: skip merge, skip delete sign
+ 25: optional bool read_mor_as_dup
}
struct TEqJoinCondition {
diff --git a/regression-test/suites/unique_with_mor_p0/test_read_mor_as_dup.groovy b/regression-test/suites/unique_with_mor_p0/test_read_mor_as_dup.groovy
new file mode 100644
index 00000000000..aada4292ff9
--- /dev/null
+++ b/regression-test/suites/unique_with_mor_p0/test_read_mor_as_dup.groovy
@@ -0,0 +1,119 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_read_mor_as_dup") {
+ def tableName = "test_read_mor_as_dup_tbl"
+ def tableName2 = "test_read_mor_as_dup_tbl2"
+
+ sql "DROP TABLE IF EXISTS ${tableName}"
+ sql "DROP TABLE IF EXISTS ${tableName2}"
+
+ // Create a MOR (Merge-On-Read) UNIQUE table
+ sql """
+ CREATE TABLE ${tableName} (
+ `k` int NOT NULL,
+ `v1` int NULL,
+ `v2` varchar(100) NULL
+ ) ENGINE=OLAP
+ UNIQUE KEY(`k`)
+ DISTRIBUTED BY HASH(`k`) BUCKETS 1
+ PROPERTIES (
+ "enable_unique_key_merge_on_write" = "false",
+ "disable_auto_compaction" = "true",
+ "replication_num" = "1"
+ );
+ """
+
+ // Create a second MOR table for per-table control test
+ sql """
+ CREATE TABLE ${tableName2} (
+ `k` int NOT NULL,
+ `v1` int NULL
+ ) ENGINE=OLAP
+ UNIQUE KEY(`k`)
+ DISTRIBUTED BY HASH(`k`) BUCKETS 1
+ PROPERTIES (
+ "enable_unique_key_merge_on_write" = "false",
+ "disable_auto_compaction" = "true",
+ "replication_num" = "1"
+ );
+ """
+
+ // Insert multiple versions of the same key
+ sql "INSERT INTO ${tableName} VALUES (1, 10, 'first');"
+ sql "INSERT INTO ${tableName} VALUES (1, 20, 'second');"
+ sql "INSERT INTO ${tableName} VALUES (1, 30, 'third');"
+ sql "INSERT INTO ${tableName} VALUES (2, 100, 'only_version');"
+
+ // Insert a row and then delete it
+ sql "INSERT INTO ${tableName} VALUES (3, 50, 'to_be_deleted');"
+ sql "DELETE FROM ${tableName} WHERE k = 3;"
+
+ // Insert into second table
+ sql "INSERT INTO ${tableName2} VALUES (1, 10);"
+ sql "INSERT INTO ${tableName2} VALUES (1, 20);"
+
+ // === Test 1: Normal query returns merged result ===
+ sql "SET read_mor_as_dup_tables = '';"
+ def normalResult = sql "SELECT * FROM ${tableName} ORDER BY k;"
+ // Should see only latest version per key: (1,30,'third'), (2,100,'only_version')
+ // Key 3 was deleted, should not appear
+ assertTrue(normalResult.size() == 2, "Normal query should return 2 rows (merged), got ${normalResult.size()}")
+
+ // === Test 2: Wildcard — read all MOR tables as DUP ===
+ sql "SET read_mor_as_dup_tables = '*';"
+ def dupResult = sql "SELECT * FROM ${tableName} ORDER BY k, v1;"
+ // Should see all row versions for key 1 (3 versions) + key 2 (1 version)
+ // Key 3: delete predicates are still applied, so deleted row should still be filtered
+ // But the delete sign filter is skipped, so we may see it depending on how delete was done.
+ // For MOR tables, DELETE adds a delete predicate in rowsets, which IS still honored.
+ assertTrue(dupResult.size() >= 4, "DUP mode should return at least 4 rows (all versions), got ${dupResult.size()}")
+
+ // Verify key 1 has multiple versions
+ def key1Rows = dupResult.findAll { it[0] == 1 }
+ assertTrue(key1Rows.size() == 3, "Key 1 should have 3 versions in DUP mode, got ${key1Rows.size()}")
+
+ // === Test 3: Per-table control with db.table format ===
+ def dbName = sql "SELECT DATABASE();"
+ def currentDb = dbName[0][0]
+
+ sql "SET read_mor_as_dup_tables = '${currentDb}.${tableName}';"
+
+ // tableName should be in DUP mode
+ def perTableResult = sql "SELECT * FROM ${tableName} ORDER BY k, v1;"
+ assertTrue(perTableResult.size() >= 4, "Per-table DUP mode should return all versions, got ${perTableResult.size()}")
+
+ // tableName2 should still be in normal (merged) mode
+ def table2Result = sql "SELECT * FROM ${tableName2} ORDER BY k;"
+ assertTrue(table2Result.size() == 1, "Table2 should still be merged (1 row), got ${table2Result.size()}")
+
+ // === Test 4: Per-table control with just table name ===
+ sql "SET read_mor_as_dup_tables = '${tableName2}';"
+
+ // tableName should now be in normal mode
+ def revertedResult = sql "SELECT * FROM ${tableName} ORDER BY k;"
+ assertTrue(revertedResult.size() == 2, "Table1 should be merged again (2 rows), got ${revertedResult.size()}")
+
+ // tableName2 should be in DUP mode
+ def table2DupResult = sql "SELECT * FROM ${tableName2} ORDER BY k, v1;"
+ assertTrue(table2DupResult.size() == 2, "Table2 in DUP mode should return 2 rows, got ${table2DupResult.size()}")
+
+ // === Cleanup ===
+ sql "SET read_mor_as_dup_tables = '';"
+ sql "DROP TABLE IF EXISTS ${tableName}"
+ sql "DROP TABLE IF EXISTS ${tableName2}"
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 02/04: [feature](scan) Implement BE-side value predicate pushdown for MOR tables
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dataroaring pushed a commit to branch feature/mor_value_predicate_pushdown_control
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 4a0c30ea7f0941cf7757ea7d422f51d9135ff166
Author: Yongqiang YANG <da...@gmail.com>
AuthorDate: Wed Feb 4 17:18:39 2026 -0800
[feature](scan) Implement BE-side value predicate pushdown for MOR tables
Add session variable enable_mor_value_predicate_pushdown_tables to control
value predicate pushdown for MOR (Merge-On-Read) tables. When enabled,
value column predicates are pushed to the storage layer for filtering.
Key changes:
- Add enable_mor_value_predicate_pushdown flag to ReaderParams and RowsetReaderContext
- Extend BetaRowsetReader::_should_push_down_value_predicates() to check the flag
- Drop __DORIS_DELETE_SIGN__ predicate from storage-layer pushdown to prevent
deleted rows from reappearing (delete sign must only be evaluated post-merge)
- Add regression test covering dedup, delete sign, delete predicate, and update scenarios
---
be/src/olap/rowset/beta_rowset_reader.cpp | 3 +-
be/src/olap/rowset/rowset_reader_context.h | 3 +
be/src/olap/tablet_reader.cpp | 10 +
be/src/olap/tablet_reader.h | 3 +
be/src/pipeline/exec/olap_scan_operator.cpp | 6 -
be/src/pipeline/exec/olap_scan_operator.h | 2 -
be/src/pipeline/exec/scan_operator.cpp | 3 +-
be/src/pipeline/exec/scan_operator.h | 1 -
be/src/vec/exec/scan/olap_scanner.cpp | 7 +
.../unique/test_mor_value_predicate_pushdown.out | 52 ++++-
.../test_mor_value_predicate_pushdown.groovy | 209 +++++++++++++++++----
11 files changed, 245 insertions(+), 54 deletions(-)
diff --git a/be/src/olap/rowset/beta_rowset_reader.cpp b/be/src/olap/rowset/beta_rowset_reader.cpp
index b15e8a8da94..8f1e86576cf 100644
--- a/be/src/olap/rowset/beta_rowset_reader.cpp
+++ b/be/src/olap/rowset/beta_rowset_reader.cpp
@@ -361,7 +361,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
diff --git a/be/src/olap/rowset/rowset_reader_context.h b/be/src/olap/rowset/rowset_reader_context.h
index 846c721ca34..dfa633c78db 100644
--- a/be/src/olap/rowset/rowset_reader_context.h
+++ b/be/src/olap/rowset/rowset_reader_context.h
@@ -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
diff --git a/be/src/olap/tablet_reader.cpp b/be/src/olap/tablet_reader.cpp
index 6c2488e9cfa..551b6727920 100644
--- a/be/src/olap/tablet_reader.cpp
+++ b/be/src/olap/tablet_reader.cpp
@@ -194,6 +194,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;
@@ -501,9 +503,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);
diff --git a/be/src/olap/tablet_reader.h b/be/src/olap/tablet_reader.h
index caa6ef29064..bed9e20c864 100644
--- a/be/src/olap/tablet_reader.h
+++ b/be/src/olap/tablet_reader.h
@@ -189,6 +189,9 @@ public:
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;
diff --git a/be/src/pipeline/exec/olap_scan_operator.cpp b/be/src/pipeline/exec/olap_scan_operator.cpp
index 2f1e6d8a85a..07c0522a4b5 100644
--- a/be/src/pipeline/exec/olap_scan_operator.cpp
+++ b/be/src/pipeline/exec/olap_scan_operator.cpp
@@ -459,12 +459,6 @@ 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;
diff --git a/be/src/pipeline/exec/olap_scan_operator.h b/be/src/pipeline/exec/olap_scan_operator.h
index 24d1df08a46..0577dd1ff2d 100644
--- a/be/src/pipeline/exec/olap_scan_operator.h
+++ b/be/src/pipeline/exec/olap_scan_operator.h
@@ -105,8 +105,6 @@ private:
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;
diff --git a/be/src/pipeline/exec/scan_operator.cpp b/be/src/pipeline/exec/scan_operator.cpp
index 524c6ae7354..6e29e5b0437 100644
--- a/be/src/pipeline/exec/scan_operator.cpp
+++ b/be/src/pipeline/exec/scan_operator.cpp
@@ -446,8 +446,7 @@ Status ScanLocalState<Derived>::_normalize_predicate(vectorized::VExprContext* c
return Status::OK();
}
- if (pdt == PushDownType::ACCEPTABLE &&
- (_is_key_column(slot->col_name()) || _should_push_down_mor_value_predicate())) {
+ if (pdt == PushDownType::ACCEPTABLE && _is_key_column(slot->col_name())) {
output_expr = nullptr;
return Status::OK();
} else {
diff --git a/be/src/pipeline/exec/scan_operator.h b/be/src/pipeline/exec/scan_operator.h
index d564149b312..8e6fcf98a3a 100644
--- a/be/src/pipeline/exec/scan_operator.h
+++ b/be/src/pipeline/exec/scan_operator.h
@@ -205,7 +205,6 @@ protected:
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;
}
diff --git a/be/src/vec/exec/scan/olap_scanner.cpp b/be/src/vec/exec/scan/olap_scanner.cpp
index 2ec44a55645..7b577bdb629 100644
--- a/be/src/vec/exec/scan/olap_scanner.cpp
+++ b/be/src/vec/exec/scan/olap_scanner.cpp
@@ -478,6 +478,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()) {
diff --git a/regression-test/data/data_model_p0/unique/test_mor_value_predicate_pushdown.out b/regression-test/data/data_model_p0/unique/test_mor_value_predicate_pushdown.out
index 42b144dc5e5..3ab8254a635 100644
--- a/regression-test/data/data_model_p0/unique/test_mor_value_predicate_pushdown.out
+++ b/regression-test/data/data_model_p0/unique/test_mor_value_predicate_pushdown.out
@@ -1,29 +1,69 @@
-- 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_deleted_row --
+-- !select_eq_predicate --
+2 200 world
-- !select_not_in_list --
+2 200 world
3 300 test
--- !select_latest_version --
-1 200 second
+-- !select_dedup_all --
+1 100 first
2 300 third
+3 500 fifth
--- !select_old_version --
+-- !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_new_version --
-1 200 second
+-- !select_update_enabled_range --
+1 500 new
+3 300 keep
-- !select_multiple_tables --
2 200
diff --git a/regression-test/suites/data_model_p0/unique/test_mor_value_predicate_pushdown.groovy b/regression-test/suites/data_model_p0/unique/test_mor_value_predicate_pushdown.groovy
index 82a3d4b41d7..b82aba0f712 100644
--- a/regression-test/suites/data_model_p0/unique/test_mor_value_predicate_pushdown.groovy
+++ b/regression-test/suites/data_model_p0/unique/test_mor_value_predicate_pushdown.groovy
@@ -19,7 +19,9 @@ suite("test_mor_value_predicate_pushdown") {
def tbName = "test_mor_value_pred_pushdown"
def dbName = context.config.getDbNameByFile(context.file)
- // Test 1: Basic MOR table with value predicate pushdown
+ // Test 1: Basic MOR table with value predicate pushdown (dedup-only scenario)
+ // This feature is designed for insert-only/dedup-only workloads where
+ // the same key always has identical values across rowsets.
sql "DROP TABLE IF EXISTS ${tbName}"
sql """
CREATE TABLE IF NOT EXISTS ${tbName} (
@@ -36,13 +38,10 @@ suite("test_mor_value_predicate_pushdown") {
);
"""
- // Insert test data
- sql "INSERT INTO ${tbName} VALUES (1, 100, 'hello')"
- sql "INSERT INTO ${tbName} VALUES (2, 200, 'world')"
- sql "INSERT INTO ${tbName} VALUES (3, 300, 'test')"
-
- // Delete a row (for MOR, this marks the row with __DORIS_DELETE_SIGN__)
- sql "DELETE FROM ${tbName} WHERE k1 = 2"
+ // Insert test data across separate rowsets (dedup-only: same key has same values)
+ sql "INSERT INTO ${tbName} VALUES (1, 100, 'hello'), (2, 200, 'world'), (3, 300, 'test')"
+ // Re-insert duplicates to create overlapping rowsets for dedup
+ sql "INSERT INTO ${tbName} VALUES (1, 100, 'hello'), (2, 200, 'world')"
// Test: pushdown disabled (default)
sql "SET enable_mor_value_predicate_pushdown_tables = ''"
@@ -84,8 +83,8 @@ suite("test_mor_value_predicate_pushdown") {
SELECT * FROM ${tbName} WHERE v1 > 150 ORDER BY k1
"""
- // Test: verify deleted row is not returned (correctness check)
- qt_select_deleted_row """
+ // Test: equality predicate on value column with pushdown
+ qt_select_eq_predicate """
SELECT * FROM ${tbName} WHERE v1 = 200 ORDER BY k1
"""
@@ -96,8 +95,8 @@ suite("test_mor_value_predicate_pushdown") {
SELECT * FROM ${tbName} WHERE v1 > 150 ORDER BY k1
"""
- // Test 2: Verify correctness with multiple updates to same key
- // This is critical - MOR tables with overlapping rowsets must return correct latest values
+ // Test 2: Multiple rowsets with dedup-only pattern
+ // Verify correctness when same keys appear across many rowsets with identical values
sql "DROP TABLE IF EXISTS ${tbName}"
sql """
CREATE TABLE IF NOT EXISTS ${tbName} (
@@ -114,30 +113,178 @@ suite("test_mor_value_predicate_pushdown") {
);
"""
- // Insert multiple versions of same key (creates overlapping rowsets)
+ // Create multiple overlapping rowsets (dedup-only: all versions have same values)
+ sql "INSERT INTO ${tbName} VALUES (1, 100, 'first'), (2, 300, 'third')"
sql "INSERT INTO ${tbName} VALUES (1, 100, 'first')"
- sql "INSERT INTO ${tbName} VALUES (1, 200, 'second')"
- sql "INSERT INTO ${tbName} VALUES (2, 300, 'third')"
+ sql "INSERT INTO ${tbName} VALUES (2, 300, 'third'), (3, 500, 'fifth')"
- // Test with pushdown enabled - must still return correct latest version
+ // Test with pushdown enabled
sql "SET enable_mor_value_predicate_pushdown_tables = '*'"
- // Should only return the latest version
- qt_select_latest_version """
+ // Should return all rows matching predicate after dedup
+ qt_select_dedup_all """
SELECT * FROM ${tbName} WHERE v1 >= 100 ORDER BY k1
"""
- // Value predicate on older version should not match (k1=1 has v1=200 now, not 100)
- qt_select_old_version """
- SELECT * FROM ${tbName} WHERE v1 = 100 ORDER BY k1
+ // Equality match on a value that exists
+ qt_select_dedup_eq """
+ SELECT * FROM ${tbName} WHERE v1 = 300 ORDER BY k1
+ """
+
+ // Predicate that matches no rows
+ qt_select_dedup_none """
+ SELECT * FROM ${tbName} WHERE v1 = 999 ORDER BY k1
+ """
+
+ // Test 3: Dedup + delete scenario
+ // Value columns are identical across rowsets (dedup-only), but some rows are deleted.
+ // Verifies that __DORIS_DELETE_SIGN__ is NOT pushed per-segment so deletions are honored.
+ sql "DROP TABLE IF EXISTS ${tbName}"
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tbName} (
+ k1 INT,
+ v1 INT,
+ v2 VARCHAR(100)
+ )
+ UNIQUE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "false",
+ "disable_auto_compaction" = "true"
+ );
+ """
+
+ // Rowset 1: initial data
+ sql "INSERT INTO ${tbName} VALUES (1, 100, 'hello'), (2, 200, 'world'), (3, 300, 'test')"
+ // Rowset 2: dedup insert (same key, same values)
+ sql "INSERT INTO ${tbName} VALUES (1, 100, 'hello'), (2, 200, 'world')"
+ // Rowset 3: delete k1=2 via insert with __DORIS_DELETE_SIGN__=1
+ // Value columns are identical (dedup-only), only delete sign differs
+ sql "INSERT INTO ${tbName}(k1, v1, v2, __DORIS_DELETE_SIGN__) VALUES (2, 200, 'world', 1)"
+
+ sql "SET enable_mor_value_predicate_pushdown_tables = '*'"
+
+ // Deleted row must not appear even though v1=200 matches the predicate
+ qt_select_delete_range """
+ SELECT * FROM ${tbName} WHERE v1 > 150 ORDER BY k1
"""
- // Value predicate on new version should match
- qt_select_new_version """
+ // Equality on deleted row's value — should return empty
+ qt_select_delete_eq """
SELECT * FROM ${tbName} WHERE v1 = 200 ORDER BY k1
"""
- // Test 3: Multiple tables in the list
+ // Broader predicate — deleted row still excluded
+ qt_select_delete_all """
+ SELECT * FROM ${tbName} WHERE v1 >= 100 ORDER BY k1
+ """
+
+ // Test 4: Dedup + delete predicate scenario
+ // DELETE FROM creates a delete predicate stored in rowset metadata.
+ // Delete predicates go through DeleteHandler, separate from value predicates.
+ // Verify they work correctly alongside value predicate pushdown.
+ sql "DROP TABLE IF EXISTS ${tbName}"
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tbName} (
+ k1 INT,
+ v1 INT,
+ v2 VARCHAR(100)
+ )
+ UNIQUE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "false",
+ "disable_auto_compaction" = "true"
+ );
+ """
+
+ // Rowset 1: initial data
+ sql "INSERT INTO ${tbName} VALUES (1, 100, 'hello'), (2, 200, 'world'), (3, 300, 'test')"
+ // Rowset 2: dedup insert (same key, same values)
+ sql "INSERT INTO ${tbName} VALUES (1, 100, 'hello'), (2, 200, 'world')"
+ // Delete predicate rowset: DELETE FROM creates a predicate, not row markers
+ sql "DELETE FROM ${tbName} WHERE k1 = 2"
+
+ sql "SET enable_mor_value_predicate_pushdown_tables = '*'"
+
+ // Deleted row must not appear
+ qt_select_delpred_range """
+ SELECT * FROM ${tbName} WHERE v1 > 150 ORDER BY k1
+ """
+
+ // Equality on deleted row's value — should return empty
+ qt_select_delpred_eq """
+ SELECT * FROM ${tbName} WHERE v1 = 200 ORDER BY k1
+ """
+
+ // Broader predicate — deleted row still excluded
+ qt_select_delpred_all """
+ SELECT * FROM ${tbName} WHERE v1 >= 100 ORDER BY k1
+ """
+
+ // Test 5: Update scenario — proves pushdown is active at storage layer
+ // k1=1 is updated from v1=100 to v1=500 across two rowsets.
+ // Comparing results with pushdown disabled vs enabled for the SAME query
+ // proves per-segment filtering is happening:
+ // disabled: merge picks latest (v1=500), VExpr filters → empty
+ // enabled: rowset 2 filtered per-segment (v1=500≠100), old version survives
+ // merge sees only old version, VExpr passes → stale (1,100,'old')
+ sql "DROP TABLE IF EXISTS ${tbName}"
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tbName} (
+ k1 INT,
+ v1 INT,
+ v2 VARCHAR(100)
+ )
+ UNIQUE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "false",
+ "disable_auto_compaction" = "true"
+ );
+ """
+
+ // Rowset 1: initial data
+ sql "INSERT INTO ${tbName} VALUES (1, 100, 'old'), (2, 200, 'keep'), (3, 300, 'keep')"
+ // Rowset 2: update k1=1 from v1=100 to v1=500
+ sql "INSERT INTO ${tbName} VALUES (1, 500, 'new')"
+
+ // --- Pushdown disabled: correct merge-then-filter behavior ---
+ sql "SET enable_mor_value_predicate_pushdown_tables = ''"
+
+ // v1=100 does not match latest version (v1=500) → empty
+ qt_select_update_disabled_old """
+ SELECT * FROM ${tbName} WHERE v1 = 100 ORDER BY k1
+ """
+
+ // v1=500 matches latest → returns updated row
+ qt_select_update_disabled_new """
+ SELECT * FROM ${tbName} WHERE v1 = 500 ORDER BY k1
+ """
+
+ // --- Pushdown enabled: per-segment filtering observable ---
+ sql "SET enable_mor_value_predicate_pushdown_tables = '*'"
+
+ // v1=100: rowset 2 (v1=500) filtered per-segment, old version survives.
+ // Returns stale data — this proves pushdown is filtering at storage layer.
+ qt_select_update_enabled_old """
+ SELECT * FROM ${tbName} WHERE v1 = 100 ORDER BY k1
+ """
+
+ // v1=500: rowset 1 (v1=100) filtered per-segment, new version passes → correct
+ qt_select_update_enabled_new """
+ SELECT * FROM ${tbName} WHERE v1 = 500 ORDER BY k1
+ """
+
+ // v1 > 200: old v1=100 filtered, new v1=500 passes, k1=3 v1=300 passes → correct
+ qt_select_update_enabled_range """
+ SELECT * FROM ${tbName} WHERE v1 > 200 ORDER BY k1
+ """
+
+ // Test 6: Multiple tables in the list
def tbName2 = "test_mor_value_pred_pushdown_2"
sql "DROP TABLE IF EXISTS ${tbName2}"
sql """
@@ -161,7 +308,7 @@ suite("test_mor_value_predicate_pushdown") {
SELECT * FROM ${tbName2} WHERE v1 > 100 ORDER BY k1
"""
- // Test 4: Non-MOR table (MOW) - value predicates should always be pushed down
+ // Test 7: Non-MOR table (MOW) - value predicates should always be pushed down
// The session variable should have no effect on MOW tables
def tbNameMow = "test_mow_value_pred"
sql "DROP TABLE IF EXISTS ${tbNameMow}"
@@ -187,7 +334,7 @@ suite("test_mor_value_predicate_pushdown") {
SELECT * FROM ${tbNameMow} WHERE v1 > 100 ORDER BY k1
"""
- // Test 5: DUP_KEYS table - value predicates should always be pushed down
+ // Test 8: DUP_KEYS table - value predicates should always be pushed down
// The session variable should have no effect on DUP_KEYS tables
def tbNameDup = "test_dup_value_pred"
sql "DROP TABLE IF EXISTS ${tbNameDup}"
@@ -210,17 +357,7 @@ suite("test_mor_value_predicate_pushdown") {
SELECT * FROM ${tbNameDup} WHERE v1 > 100 ORDER BY k1
"""
- // Test 6: Profile verification - check that predicate pushdown affects filtering
- // Enable profiling and verify RowsConditionsFiltered counter when pushdown is enabled
- sql "SET enable_profile = true"
- sql "SET enable_mor_value_predicate_pushdown_tables = '*'"
-
- // Execute query and check profile shows filtering happened at storage layer
- def profileQuery = "SELECT /*+ SET_VAR(enable_mor_value_predicate_pushdown_tables='*') */ * FROM ${tbName} WHERE v1 > 250"
- sql profileQuery
-
// Cleanup
- sql "SET enable_profile = false"
sql "SET enable_mor_value_predicate_pushdown_tables = ''"
sql "DROP TABLE IF EXISTS ${tbName}"
sql "DROP TABLE IF EXISTS ${tbName2}"
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 01/04: [feature](scan) Add session variable to control value predicate pushdown for MOR tables
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dataroaring pushed a commit to branch feature/mor_value_predicate_pushdown_control
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 5e5179e584ec4140fc59ea4f0ea04da0a6a1229c
Author: Yongqiang YANG <da...@gmail.com>
AuthorDate: Wed Feb 4 09:23:40 2026 -0800
[feature](scan) Add session variable to control value predicate pushdown for MOR tables
Add a new session variable `enable_mor_value_predicate_pushdown_tables` to allow
users to selectively enable value column predicate pushdown for MOR (Merge-On-Read)
tables. This can improve query performance by utilizing inverted indexes on value
columns for filtering.
The session variable accepts:
- Comma-separated table names: `db1.tbl1,db2.tbl2`
- Wildcard for all MOR tables: `*`
- Empty string to disable (default)
Changes:
- Add session variable in SessionVariable.java with helper method
- Add isMorTable() helper in OlapTable.java
- Add Thrift field enable_mor_value_predicate_pushdown in TOlapScanNode
- Set flag in OlapScanNode.toThrift() based on session variable
- Add virtual method _should_push_down_mor_value_predicate() in scan_operator
- Implement override in olap_scan_operator to read the flag
- Modify predicate pushdown condition in scan_operator.cpp
---
be/src/pipeline/exec/olap_scan_operator.cpp | 6 +
be/src/pipeline/exec/olap_scan_operator.h | 2 +
be/src/pipeline/exec/scan_operator.cpp | 3 +-
be/src/pipeline/exec/scan_operator.h | 1 +
.../java/org/apache/doris/catalog/OlapTable.java | 8 +
.../org/apache/doris/planner/OlapScanNode.java | 9 +
.../java/org/apache/doris/qe/SessionVariable.java | 39 ++++
gensrc/thrift/PlanNodes.thrift | 2 +
.../unique/test_mor_value_predicate_pushdown.out | 36 ++++
.../test_mor_value_predicate_pushdown.groovy | 229 +++++++++++++++++++++
10 files changed, 334 insertions(+), 1 deletion(-)
diff --git a/be/src/pipeline/exec/olap_scan_operator.cpp b/be/src/pipeline/exec/olap_scan_operator.cpp
index 07c0522a4b5..2f1e6d8a85a 100644
--- a/be/src/pipeline/exec/olap_scan_operator.cpp
+++ b/be/src/pipeline/exec/olap_scan_operator.cpp
@@ -459,6 +459,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;
diff --git a/be/src/pipeline/exec/olap_scan_operator.h b/be/src/pipeline/exec/olap_scan_operator.h
index 0577dd1ff2d..24d1df08a46 100644
--- a/be/src/pipeline/exec/olap_scan_operator.h
+++ b/be/src/pipeline/exec/olap_scan_operator.h
@@ -105,6 +105,8 @@ private:
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;
diff --git a/be/src/pipeline/exec/scan_operator.cpp b/be/src/pipeline/exec/scan_operator.cpp
index 8bb83e3de6a..524c6ae7354 100644
--- a/be/src/pipeline/exec/scan_operator.cpp
+++ b/be/src/pipeline/exec/scan_operator.cpp
@@ -446,7 +446,8 @@ 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()) || _should_push_down_mor_value_predicate())) {
output_expr = nullptr;
return Status::OK();
} else {
diff --git a/be/src/pipeline/exec/scan_operator.h b/be/src/pipeline/exec/scan_operator.h
index 8e6fcf98a3a..d564149b312 100644
--- a/be/src/pipeline/exec/scan_operator.h
+++ b/be/src/pipeline/exec/scan_operator.h
@@ -205,6 +205,7 @@ protected:
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;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
index 3d6de95aed4..e724c1df97c 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
@@ -3028,6 +3028,14 @@ public class OlapTable extends Table implements MTMVRelatedTableIf, GsonPostProc
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();
+ }
+
public boolean isUniqKeyMergeOnWriteWithClusterKeys() {
return isUniqKeyMergeOnWrite() && getBaseSchema().stream().anyMatch(Column::isClusterKey);
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
index 0c4f00fc323..9d4b6fddfc5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
@@ -1190,6 +1190,15 @@ public class OlapScanNode extends ScanNode {
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();
+ 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);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index 362b69533d2..a6e581199d5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -732,6 +732,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";
@@ -2222,6 +2225,13 @@ public class SessionVariable implements Serializable, Writable {
"是否启用 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;
@@ -4770,6 +4780,35 @@ public class SessionVariable implements Serializable, Writable {
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;
+ }
+ String fullName = dbName + "." + tableName;
+ for (String table : trimmed.split(",")) {
+ if (table.trim().equalsIgnoreCase(fullName)
+ || table.trim().equalsIgnoreCase(tableName)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
/** canUseNereidsDistributePlanner */
public static boolean canUseNereidsDistributePlanner() {
ConnectContext connectContext = ConnectContext.get();
diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift
index 7b281dcf712..7fcb1912c67 100644
--- a/gensrc/thrift/PlanNodes.thrift
+++ b/gensrc/thrift/PlanNodes.thrift
@@ -894,6 +894,8 @@ struct TOlapScanNode {
21: optional TSortInfo ann_sort_info
22: optional i64 ann_sort_limit
23: optional TScoreRangeInfo score_range_info
+ // Enable value predicate pushdown for MOR tables
+ 24: optional bool enable_mor_value_predicate_pushdown
}
struct TEqJoinCondition {
diff --git a/regression-test/data/data_model_p0/unique/test_mor_value_predicate_pushdown.out b/regression-test/data/data_model_p0/unique/test_mor_value_predicate_pushdown.out
new file mode 100644
index 00000000000..42b144dc5e5
--- /dev/null
+++ b/regression-test/data/data_model_p0/unique/test_mor_value_predicate_pushdown.out
@@ -0,0 +1,36 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select_disabled --
+3 300 test
+
+-- !select_enabled_tablename --
+3 300 test
+
+-- !select_enabled_fullname --
+3 300 test
+
+-- !select_enabled_wildcard --
+3 300 test
+
+-- !select_deleted_row --
+
+-- !select_not_in_list --
+3 300 test
+
+-- !select_latest_version --
+1 200 second
+2 300 third
+
+-- !select_old_version --
+
+-- !select_new_version --
+1 200 second
+
+-- !select_multiple_tables --
+2 200
+
+-- !select_mow_table --
+2 200
+
+-- !select_dup_table --
+2 200
+
diff --git a/regression-test/suites/data_model_p0/unique/test_mor_value_predicate_pushdown.groovy b/regression-test/suites/data_model_p0/unique/test_mor_value_predicate_pushdown.groovy
new file mode 100644
index 00000000000..82a3d4b41d7
--- /dev/null
+++ b/regression-test/suites/data_model_p0/unique/test_mor_value_predicate_pushdown.groovy
@@ -0,0 +1,229 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_mor_value_predicate_pushdown") {
+ def tbName = "test_mor_value_pred_pushdown"
+ def dbName = context.config.getDbNameByFile(context.file)
+
+ // Test 1: Basic MOR table with value predicate pushdown
+ sql "DROP TABLE IF EXISTS ${tbName}"
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tbName} (
+ k1 INT,
+ v1 INT,
+ v2 VARCHAR(100)
+ )
+ UNIQUE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 3
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "false",
+ "disable_auto_compaction" = "true"
+ );
+ """
+
+ // Insert test data
+ sql "INSERT INTO ${tbName} VALUES (1, 100, 'hello')"
+ sql "INSERT INTO ${tbName} VALUES (2, 200, 'world')"
+ sql "INSERT INTO ${tbName} VALUES (3, 300, 'test')"
+
+ // Delete a row (for MOR, this marks the row with __DORIS_DELETE_SIGN__)
+ sql "DELETE FROM ${tbName} WHERE k1 = 2"
+
+ // Test: pushdown disabled (default)
+ sql "SET enable_mor_value_predicate_pushdown_tables = ''"
+
+ // Verify session variable is set correctly
+ def result = sql "SHOW VARIABLES LIKE 'enable_mor_value_predicate_pushdown_tables'"
+ assertTrue(result.size() > 0)
+ assertTrue(result[0][1] == "")
+
+ // Query with value predicate - should return correct results
+ qt_select_disabled """
+ SELECT * FROM ${tbName} WHERE v1 > 150 ORDER BY k1
+ """
+
+ // Test: enable for specific table (just table name)
+ sql "SET enable_mor_value_predicate_pushdown_tables = '${tbName}'"
+
+ result = sql "SHOW VARIABLES LIKE 'enable_mor_value_predicate_pushdown_tables'"
+ assertTrue(result[0][1] == tbName)
+
+ qt_select_enabled_tablename """
+ SELECT * FROM ${tbName} WHERE v1 > 150 ORDER BY k1
+ """
+
+ // Test: enable for specific table with db prefix
+ sql "SET enable_mor_value_predicate_pushdown_tables = '${dbName}.${tbName}'"
+
+ qt_select_enabled_fullname """
+ SELECT * FROM ${tbName} WHERE v1 > 150 ORDER BY k1
+ """
+
+ // Test: enable for all MOR tables with '*'
+ sql "SET enable_mor_value_predicate_pushdown_tables = '*'"
+
+ result = sql "SHOW VARIABLES LIKE 'enable_mor_value_predicate_pushdown_tables'"
+ assertTrue(result[0][1] == "*")
+
+ qt_select_enabled_wildcard """
+ SELECT * FROM ${tbName} WHERE v1 > 150 ORDER BY k1
+ """
+
+ // Test: verify deleted row is not returned (correctness check)
+ qt_select_deleted_row """
+ SELECT * FROM ${tbName} WHERE v1 = 200 ORDER BY k1
+ """
+
+ // Test: table not in list - pushdown should be disabled
+ sql "SET enable_mor_value_predicate_pushdown_tables = 'other_table'"
+
+ qt_select_not_in_list """
+ SELECT * FROM ${tbName} WHERE v1 > 150 ORDER BY k1
+ """
+
+ // Test 2: Verify correctness with multiple updates to same key
+ // This is critical - MOR tables with overlapping rowsets must return correct latest values
+ sql "DROP TABLE IF EXISTS ${tbName}"
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tbName} (
+ k1 INT,
+ v1 INT,
+ v2 VARCHAR(100)
+ )
+ UNIQUE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "false",
+ "disable_auto_compaction" = "true"
+ );
+ """
+
+ // Insert multiple versions of same key (creates overlapping rowsets)
+ sql "INSERT INTO ${tbName} VALUES (1, 100, 'first')"
+ sql "INSERT INTO ${tbName} VALUES (1, 200, 'second')"
+ sql "INSERT INTO ${tbName} VALUES (2, 300, 'third')"
+
+ // Test with pushdown enabled - must still return correct latest version
+ sql "SET enable_mor_value_predicate_pushdown_tables = '*'"
+
+ // Should only return the latest version
+ qt_select_latest_version """
+ SELECT * FROM ${tbName} WHERE v1 >= 100 ORDER BY k1
+ """
+
+ // Value predicate on older version should not match (k1=1 has v1=200 now, not 100)
+ qt_select_old_version """
+ SELECT * FROM ${tbName} WHERE v1 = 100 ORDER BY k1
+ """
+
+ // Value predicate on new version should match
+ qt_select_new_version """
+ SELECT * FROM ${tbName} WHERE v1 = 200 ORDER BY k1
+ """
+
+ // Test 3: Multiple tables in the list
+ def tbName2 = "test_mor_value_pred_pushdown_2"
+ sql "DROP TABLE IF EXISTS ${tbName2}"
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tbName2} (
+ k1 INT,
+ v1 INT
+ )
+ UNIQUE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "false"
+ );
+ """
+
+ sql "INSERT INTO ${tbName2} VALUES (1, 100), (2, 200)"
+
+ sql "SET enable_mor_value_predicate_pushdown_tables = '${tbName},${tbName2}'"
+
+ qt_select_multiple_tables """
+ SELECT * FROM ${tbName2} WHERE v1 > 100 ORDER BY k1
+ """
+
+ // Test 4: Non-MOR table (MOW) - value predicates should always be pushed down
+ // The session variable should have no effect on MOW tables
+ def tbNameMow = "test_mow_value_pred"
+ sql "DROP TABLE IF EXISTS ${tbNameMow}"
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tbNameMow} (
+ k1 INT,
+ v1 INT
+ )
+ UNIQUE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true"
+ );
+ """
+
+ sql "INSERT INTO ${tbNameMow} VALUES (1, 100), (2, 200)"
+
+ // MOW tables always push down value predicates regardless of setting
+ sql "SET enable_mor_value_predicate_pushdown_tables = ''"
+
+ qt_select_mow_table """
+ SELECT * FROM ${tbNameMow} WHERE v1 > 100 ORDER BY k1
+ """
+
+ // Test 5: DUP_KEYS table - value predicates should always be pushed down
+ // The session variable should have no effect on DUP_KEYS tables
+ def tbNameDup = "test_dup_value_pred"
+ sql "DROP TABLE IF EXISTS ${tbNameDup}"
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tbNameDup} (
+ k1 INT,
+ v1 INT
+ )
+ DUPLICATE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1"
+ );
+ """
+
+ sql "INSERT INTO ${tbNameDup} VALUES (1, 100), (2, 200)"
+
+ // DUP_KEYS tables always push down value predicates regardless of setting
+ qt_select_dup_table """
+ SELECT * FROM ${tbNameDup} WHERE v1 > 100 ORDER BY k1
+ """
+
+ // Test 6: Profile verification - check that predicate pushdown affects filtering
+ // Enable profiling and verify RowsConditionsFiltered counter when pushdown is enabled
+ sql "SET enable_profile = true"
+ sql "SET enable_mor_value_predicate_pushdown_tables = '*'"
+
+ // Execute query and check profile shows filtering happened at storage layer
+ def profileQuery = "SELECT /*+ SET_VAR(enable_mor_value_predicate_pushdown_tables='*') */ * FROM ${tbName} WHERE v1 > 250"
+ sql profileQuery
+
+ // Cleanup
+ sql "SET enable_profile = false"
+ sql "SET enable_mor_value_predicate_pushdown_tables = ''"
+ sql "DROP TABLE IF EXISTS ${tbName}"
+ sql "DROP TABLE IF EXISTS ${tbName2}"
+ sql "DROP TABLE IF EXISTS ${tbNameMow}"
+ sql "DROP TABLE IF EXISTS ${tbNameDup}"
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org