[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 <noreply@anthropic.com>

Author: dataroaring

be/src/pipeline/exec/olap_scan_operator.cpp

@@ -442,7 +442,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) {

be/src/pipeline/exec/olap_scan_operator.h

@@ -105,6 +105,7 @@ class OlapScanLocalState final : public ScanLocalState<OlapScanLocalState> {
 
     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;

be/src/vec/exec/scan/olap_scanner.cpp

@@ -336,7 +336,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 {

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java

@@ -370,7 +370,10 @@ private Expression generateAggFunction(SlotReference slot, Column column) {
     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()) {

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.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 Plan visitLogicalOlapScan(LogicalOlapScan logicalOlapScan, Stack<PreAggIn
             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()) {

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapScan.java

@@ -752,6 +752,13 @@ AGGREGATE KEY (siteid,citycode,username)
                     && 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)) {

fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java

@@ -1188,6 +1188,10 @@ protected void toThrift(TPlanNode msg) {
             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);

fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java

@@ -725,6 +725,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";
@@ -2191,6 +2193,14 @@ public boolean isEnableHboNonStrictMatchingMode() {
                 + "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;
@@ -4698,6 +4708,24 @@ public boolean isMorValuePredicatePushdownEnabled(String dbName, String tableNam
         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();

gensrc/thrift/PlanNodes.thrift

@@ -889,6 +889,8 @@ struct TOlapScanNode {
   22: optional i64 ann_sort_limit
   // Enable value predicate pushdown for MOR tables
   23: optional bool enable_mor_value_predicate_pushdown
+  // Read MOR table as DUP table: skip merge, skip delete sign
+  24: optional bool read_mor_as_dup
 }
 
 struct TEqJoinCondition {

regression-test/suites/unique_with_mor_p0/test_read_mor_as_dup.groovy

@@ -0,0 +1,117 @@
+// 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",
+            "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",
+            "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}"
+}