[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();