branch-4.1:[fix](subquery): avoid join amplification in WinMagic rewrite rule

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

@@ -45,6 +45,7 @@
 import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
 import org.apache.doris.nereids.util.Utils;
 import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.SessionVariable;
 import org.apache.doris.rpc.RpcException;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -809,6 +810,14 @@ public JSONObject toJson() {
 
     @Override
     public void computeUnique(DataTrait.Builder builder) {
+        // Duplicate-producing scan modes invalidate all uniqueness
+        // guarantees regardless of table metadata or declared constraints.
+        // When the BE returns unmerged versions or duplicate key rows, the
+        // WinMagic window-function rewrite (and any other rule relying on
+        // uniqueness) would produce wrong results.
+        if (isDuplicateProducingScanMode()) {
+            return;
+        }
         // Raw-version reads expose superseded rows: with skipDeleteBitmap, rows replaced by
         // later versions are read; with read_mor_as_dup_tables, MOR tables are read as DUP and
         // expose every version. Uniqueness — including the table-level constraints imported by
@@ -848,6 +857,13 @@ AGGREGATE KEY (siteid,citycode,username)
                 the mv2's agg key citycode * citycode is not unique
 
                 for simplicity, we disable unique compute for mv
+
+                Declared constraints from super.computeUnique(builder) were
+                already filtered by findSlotsByColumn() above: if a rollup
+                does not include ALL columns of a PRIMARY KEY / UNIQUE
+                constraint, the partial-column set is discarded.  Only
+                constraints whose full column set is present in the scan
+                output are propagated.
              */
             return;
         }
@@ -948,6 +964,45 @@ public void computeFd(DataTrait.Builder builder) {
         }
     }
 
+    /**
+     * Whether the scan is configured with a session variable or scan
+     * mode that makes the BE return potentially duplicate rows even for
+     * declared unique keys.  In these modes any uniqueness guarantee —
+     * whether from OLAP key metadata or from user-declared PRIMARY KEY /
+     * UNIQUE constraints — is unreliable.
+     */
+    private boolean isDuplicateProducingScanMode() {
+        SessionVariable sv = ConnectContext.get().getSessionVariable();
+        // skipStorageEngineMerge: BE returns unmerged versions — all
+        // table types may have duplicate key rows.
+        if (sv.skipStorageEngineMerge) {
+            return true;
+        }
+        // skipDeleteBitmap: on UNIQUE_KEYS tables, rows that were replaced
+        // due to the same key are also read, so the key duplicates.
+        if (sv.skipDeleteBitmap && getTable().getKeysType() == KeysType.UNIQUE_KEYS) {
+            return true;
+        }
+        // readMorAsDup: MOW tables are read as DUPLICATE — uniqueness
+        // of the declared key is not guaranteed.
+        if (getTable().getKeysType() == KeysType.UNIQUE_KEYS
+                && getTable().isMorTable()
+                && sv.isReadMorAsDupEnabled(
+                    getTable().getQualifiedDbName(), getTable().getName())) {
+            return true;
+        }
+        // Stream scans and other scan subclasses that can return duplicate
+        // key rows are handled via producesDuplicateRows() which defaults
+        // to false and is overridden by subclasses with special semantics.
+        // This follows the Open/Closed principle: adding a new scan subclass
+        // with duplicate-producing behavior does not require modifying
+        // LogicalOlapScan.
+        if (producesDuplicateRows()) {
+            return true;
+        }
+        return false;
+    }
+
     @Override
     public CatalogRelation withOperativeSlots(Collection<Slot> operativeSlots) {
         return new LogicalOlapScan(relationId, (Table) table, qualifier,
@@ -1059,4 +1114,14 @@ protected boolean hasSameScanState(LogicalCatalogRelation other) {
     public boolean supportPruneNestedColumn() {
         return true;
     }
+
+    /**
+     * Override point for scan subclasses that can return duplicate rows
+     * even for declared unique keys.  The base {@code LogicalOlapScan}
+     * returns {@code false}; subclasses that introduce special scan modes
+     * (e.g. incremental stream scans) override this to return {@code true}.
+     */
+    protected boolean producesDuplicateRows() {
+        return false;
+    }
 }