You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2026/04/22 04:41:07 UTC
(doris) branch branch-4.1 updated: branch-4.1: [feature](scan) Add value predicate pushdown control for MOR tables #60513 (#62451)
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 84676d40bed branch-4.1: [feature](scan) Add value predicate pushdown control for MOR tables #60513 (#62451)
84676d40bed is described below
commit 84676d40bedc1f34143537404c7e80db6f518b46
Author: Yongqiang YANG <ya...@selectdb.com>
AuthorDate: Tue Apr 21 21:41:00 2026 -0700
branch-4.1: [feature](scan) Add value predicate pushdown control for MOR tables #60513 (#62451)
pick https://github.com/apache/doris/pull/60513
Co-authored-by: Claude Opus 4.6 <no...@anthropic.com>
---
be/src/exec/operator/olap_scan_operator.cpp | 8 +-
be/src/exec/operator/olap_scan_operator.h | 1 +
be/src/exec/operator/scan_operator.cpp | 2 +-
be/src/exec/scan/olap_scanner.cpp | 12 +-
be/src/storage/rowset/beta_rowset_reader.cpp | 3 +-
be/src/storage/rowset/rowset_reader_context.h | 3 +
be/src/storage/tablet/tablet_reader.cpp | 10 +
be/src/storage/tablet/tablet_reader.h | 3 +
.../java/org/apache/doris/catalog/OlapTable.java | 8 +
.../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 | 13 +
.../java/org/apache/doris/qe/SessionVariable.java | 69 ++++
.../nereids/rules/analysis/ReadMorAsDupTest.java | 280 ++++++++++++++++
.../org/apache/doris/qe/SessionVariablesTest.java | 55 ++++
gensrc/thrift/PlanNodes.thrift | 4 +
.../unique/test_mor_value_predicate_pushdown.out | 76 +++++
.../test_mor_value_predicate_pushdown.groovy | 366 +++++++++++++++++++++
.../unique_with_mor_p0/test_read_mor_as_dup.groovy | 119 +++++++
20 files changed, 1047 insertions(+), 6 deletions(-)
diff --git a/be/src/exec/operator/olap_scan_operator.cpp b/be/src/exec/operator/olap_scan_operator.cpp
index 1095d6cdfa6..669694df281 100644
--- a/be/src/exec/operator/olap_scan_operator.cpp
+++ b/be/src/exec/operator/olap_scan_operator.cpp
@@ -471,7 +471,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<ScannerSPtr>* scanners) {
diff --git a/be/src/exec/operator/olap_scan_operator.h b/be/src/exec/operator/olap_scan_operator.h
index 54a3fc9e109..3a27db78885 100644
--- a/be/src/exec/operator/olap_scan_operator.h
+++ b/be/src/exec/operator/olap_scan_operator.h
@@ -103,6 +103,7 @@ private:
bool _storage_no_merge() override;
+ bool _read_mor_as_dup();
bool _push_down_topn(const RuntimePredicate& predicate) override {
if (!predicate.target_is_slot(_parent->node_id())) {
return false;
diff --git a/be/src/exec/operator/scan_operator.cpp b/be/src/exec/operator/scan_operator.cpp
index bed62a0c90a..cdfad66dde6 100644
--- a/be/src/exec/operator/scan_operator.cpp
+++ b/be/src/exec/operator/scan_operator.cpp
@@ -468,7 +468,7 @@ Status ScanLocalState<Derived>::_normalize_predicate(VExprContext* context, cons
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 {
diff --git a/be/src/exec/scan/olap_scanner.cpp b/be/src/exec/scan/olap_scanner.cpp
index 1a20bec1ae2..71cac389f78 100644
--- a/be/src/exec/scan/olap_scanner.cpp
+++ b/be/src/exec/scan/olap_scanner.cpp
@@ -287,7 +287,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<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 {
@@ -408,6 +411,13 @@ Status OlapScanner::_init_tablet_reader_params(
if (!_state->skip_storage_engine_merge()) {
auto* olap_scan_local_state = (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/be/src/storage/rowset/beta_rowset_reader.cpp b/be/src/storage/rowset/beta_rowset_reader.cpp
index 5b8c1f9b2b7..e4a2b45b21f 100644
--- a/be/src/storage/rowset/beta_rowset_reader.cpp
+++ b/be/src/storage/rowset/beta_rowset_reader.cpp
@@ -363,7 +363,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/storage/rowset/rowset_reader_context.h b/be/src/storage/rowset/rowset_reader_context.h
index 5855b113a93..e44733367c8 100644
--- a/be/src/storage/rowset/rowset_reader_context.h
+++ b/be/src/storage/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/storage/tablet/tablet_reader.cpp b/be/src/storage/tablet/tablet_reader.cpp
index d56e37672f8..dc470525431 100644
--- a/be/src/storage/tablet/tablet_reader.cpp
+++ b/be/src/storage/tablet/tablet_reader.cpp
@@ -177,6 +177,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;
@@ -471,9 +473,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/storage/tablet/tablet_reader.h b/be/src/storage/tablet/tablet_reader.h
index b0fbacfac7c..6f6683bfaa2 100644
--- a/be/src/storage/tablet/tablet_reader.h
+++ b/be/src/storage/tablet/tablet_reader.h
@@ -195,6 +195,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/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 a870afb0315..5555339ed63 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
@@ -3073,6 +3073,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/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 2ab3405f3c0..72da32e49d5 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
@@ -785,6 +785,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 4f033638436..6a91b2b8686 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
@@ -1231,6 +1231,19 @@ 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);
+ if (ConnectContext.get().getSessionVariable()
+ .isReadMorAsDupEnabled(dbName, tblName)) {
+ msg.olap_scan_node.setReadMorAsDup(true);
+ }
+ }
+
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 ae8acfb09ae..af2eb71aecf 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
@@ -740,6 +740,11 @@ 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";
+
+ 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";
@@ -2233,6 +2238,21 @@ 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 = "";
+
+ // 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;
@@ -4829,6 +4849,55 @@ public class SessionVariable implements Serializable, Writable {
return enablePushDownStringMinMax;
}
+ public String getEnableMorValuePredicatePushdownTables() {
+ return enableMorValuePredicatePushdownTables;
+ }
+
+ public boolean isMorValuePredicatePushdownEnabled(String dbName, String tableName) {
+ return isTableInList(enableMorValuePredicatePushdownTables, dbName, tableName);
+ }
+
+ public boolean isReadMorAsDupEnabled(String dbName, String tableName) {
+ 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 = tableList.trim();
+ if ("*".equals(trimmed)) {
+ return true;
+ }
+ 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;
+ }
+ }
+ 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..1c3f687c8ab
--- /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 requires both db and table components to match
+ 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/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
index b646bc1a4b6..2610a8b8b48 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
@@ -117,4 +117,59 @@ public class SessionVariablesTest extends TestWithFeService {
sv.enableStrictConsistencyDml = false;
Assertions.assertFalse(sv.isEnableStrictConsistencyDml());
}
+
+ @Test
+ public void testMorValuePredicatePushdownEnabled() {
+ SessionVariable sv = new SessionVariable();
+
+ // default empty string — disabled for all tables
+ Assertions.assertFalse(sv.isMorValuePredicatePushdownEnabled("db1", "tbl1"));
+
+ // wildcard enables all tables
+ sv.enableMorValuePredicatePushdownTables = "*";
+ Assertions.assertTrue(sv.isMorValuePredicatePushdownEnabled("db1", "tbl1"));
+ Assertions.assertTrue(sv.isMorValuePredicatePushdownEnabled(null, "tbl1"));
+
+ // single table name without db — matches any database
+ sv.enableMorValuePredicatePushdownTables = "tbl1";
+ Assertions.assertTrue(sv.isMorValuePredicatePushdownEnabled("db1", "tbl1"));
+ Assertions.assertTrue(sv.isMorValuePredicatePushdownEnabled("db2", "tbl1"));
+ Assertions.assertFalse(sv.isMorValuePredicatePushdownEnabled("db1", "tbl2"));
+
+ // table name with db prefix — must match both
+ sv.enableMorValuePredicatePushdownTables = "db1.tbl1";
+ Assertions.assertTrue(sv.isMorValuePredicatePushdownEnabled("db1", "tbl1"));
+ Assertions.assertFalse(sv.isMorValuePredicatePushdownEnabled("db2", "tbl1"));
+
+ // multiple tables comma-separated
+ sv.enableMorValuePredicatePushdownTables = "db1.tbl1,tbl2";
+ Assertions.assertTrue(sv.isMorValuePredicatePushdownEnabled("db1", "tbl1"));
+ Assertions.assertFalse(sv.isMorValuePredicatePushdownEnabled("db2", "tbl1"));
+ Assertions.assertTrue(sv.isMorValuePredicatePushdownEnabled("db2", "tbl2"));
+
+ // case-insensitive matching
+ sv.enableMorValuePredicatePushdownTables = "DB1.TBL1";
+ Assertions.assertTrue(sv.isMorValuePredicatePushdownEnabled("db1", "tbl1"));
+
+ // whitespace handling
+ sv.enableMorValuePredicatePushdownTables = " tbl1 , db2.tbl2 ";
+ Assertions.assertTrue(sv.isMorValuePredicatePushdownEnabled("db1", "tbl1"));
+ Assertions.assertTrue(sv.isMorValuePredicatePushdownEnabled("db2", "tbl2"));
+ Assertions.assertFalse(sv.isMorValuePredicatePushdownEnabled("db1", "tbl2"));
+
+ // null dbName — matches table-only entries, not db-qualified entries
+ sv.enableMorValuePredicatePushdownTables = "tbl1,db2.tbl2";
+ Assertions.assertTrue(sv.isMorValuePredicatePushdownEnabled(null, "tbl1"));
+ Assertions.assertFalse(sv.isMorValuePredicatePushdownEnabled(null, "tbl2"));
+
+ // consecutive commas / empty entries
+ sv.enableMorValuePredicatePushdownTables = "tbl1,,tbl2";
+ Assertions.assertTrue(sv.isMorValuePredicatePushdownEnabled("db1", "tbl1"));
+ Assertions.assertTrue(sv.isMorValuePredicatePushdownEnabled("db1", "tbl2"));
+
+ // ctl.db.table format — matches on db and table components
+ sv.enableMorValuePredicatePushdownTables = "ctl1.db1.tbl1";
+ Assertions.assertTrue(sv.isMorValuePredicatePushdownEnabled("db1", "tbl1"));
+ Assertions.assertFalse(sv.isMorValuePredicatePushdownEnabled("db2", "tbl1"));
+ }
}
diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift
index 8814ebabb1b..67c0efab925 100644
--- a/gensrc/thrift/PlanNodes.thrift
+++ b/gensrc/thrift/PlanNodes.thrift
@@ -906,6 +906,10 @@ 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
+ // 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/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..3ab8254a635
--- /dev/null
+++ b/regression-test/data/data_model_p0/unique/test_mor_value_predicate_pushdown.out
@@ -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
+
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..b82aba0f712
--- /dev/null
+++ b/regression-test/suites/data_model_p0/unique/test_mor_value_predicate_pushdown.groovy
@@ -0,0 +1,366 @@
+// 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 (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} (
+ 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 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 = ''"
+
+ // 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: equality predicate on value column with pushdown
+ qt_select_eq_predicate """
+ 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: 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} (
+ 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"
+ );
+ """
+
+ // 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 (2, 300, 'third'), (3, 500, 'fifth')"
+
+ // Test with pushdown enabled
+ sql "SET enable_mor_value_predicate_pushdown_tables = '*'"
+
+ // Should return all rows matching predicate after dedup
+ qt_select_dedup_all """
+ 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
+ """
+
+ // Equality on deleted row's value — should return empty
+ qt_select_delete_eq """
+ SELECT * FROM ${tbName} WHERE v1 = 200 ORDER BY k1
+ """
+
+ // 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 """
+ 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 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}"
+ 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 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}"
+ 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
+ """
+
+ // Cleanup
+ 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}"
+}
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