[fix](fe) Suppress invalid unique constraints in scan FD derivation

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

@@ -235,7 +235,12 @@ private ImmutableSet<SlotReference> findSlotsByColumn(Set<Slot> outputSet, Set<C
                 slotSet.add(slotRef);
             }
         }
-        return slotSet.build();
+        // A composite constraint (e.g. UNIQUE(a,b)) must appear in the output COMPLETELY to be
+        // registered. When the scan output misses a constrained column (e.g. a non-base index
+        // that only covers (a,c)), registering the partial set {a} wrongly marks {a} as unique
+        // and lets EliminateGroupByKey derive a -> c. Return empty in that case.
+        ImmutableSet<SlotReference> matched = slotSet.build();
+        return matched.size() == columns.size() ? matched : ImmutableSet.of();
     }
 
     @Override

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

@@ -963,6 +963,19 @@ public JSONObject toJson() {
 
     @Override
     public void computeUnique(DataTrait.Builder builder) {
+        // 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
+        // super.computeUnique() — does not hold for the data actually read, so suppress it here
+        // before super runs; otherwise the raw-version guard below would return after the
+        // constraint was already registered.
+        if (getTable().getKeysType() == KeysType.UNIQUE_KEYS
+                && (ConnectContext.get().getSessionVariable().skipDeleteBitmap
+                    || (getTable().isMorTable()
+                        && ConnectContext.get().getSessionVariable().isReadMorAsDupEnabled(
+                            getTable().getQualifiedDbName(), getTable().getName())))) {
+            return;
+        }
         super.computeUnique(builder);
         if (this.selectedIndexId != getTable().getBaseIndexId()) {
             /*
@@ -1010,19 +1023,8 @@ AGGREGATE KEY (siteid,citycode,username)
             builder.addUniqueSlot(originalPlan.getLogicalProperties().getTrait());
             builder.replaceUniqueBy(constructReplaceMap(mtmv));
         } else if (getTable().getKeysType().isAggregationFamily() && !getTable().isRandomDistribution()) {
-            // When skipDeleteBitmap is set to true, in the unique model, rows that are replaced due to having the same
-            // unique key will also be read. As a result, the uniqueness of the unique key cannot be guaranteed.
-            if (ConnectContext.get().getSessionVariable().skipDeleteBitmap
-                    && 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;
-            }
+            // raw-version guards (skipDeleteBitmap / read_mor_as_dup_tables) are checked at the
+            // top of this method, before super.computeUnique() imports table-level constraints
             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/nereids/trees/plans/physical/PhysicalCatalogRelation.java

@@ -237,7 +237,12 @@ private ImmutableSet<SlotReference> findSlotsByColumn(Set<Slot> outputSet, Set<C
                 slotSet.add(slotRef);
             }
         }
-        return slotSet.build();
+        // A composite constraint (e.g. UNIQUE(a,b)) must appear in the output COMPLETELY to be
+        // registered. When the scan output misses a constrained column (e.g. a non-base index
+        // that only covers (a,c)), registering the partial set {a} wrongly marks {a} as unique
+        // and lets EliminateGroupByKey derive a -> c. Return empty in that case.
+        ImmutableSet<SlotReference> matched = slotSet.build();
+        return matched.size() == columns.size() ? matched : ImmutableSet.of();
     }
 
     @Override

fe/fe-core/src/test/java/org/apache/doris/nereids/properties/FdTest.java

@@ -17,19 +17,31 @@
 
 package org.apache.doris.nereids.properties;
 
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.Table;
 import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
 import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PreAggStatus;
 import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
 import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalHashJoin;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalPlan;
 import org.apache.doris.nereids.util.PlanChecker;
 import org.apache.doris.utframe.TestWithFeService;
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Maps;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
+import java.util.List;
+import java.util.Optional;
 import java.util.Set;
 import java.util.function.Predicate;
 
@@ -338,4 +350,69 @@ void testWindow() {
                 .isDependent(ImmutableSet.of(plan.getOutput().get(1)), ImmutableSet.of(plan.getOutput().get(0))));
     }
 
+    @Test
+    void testScanOutputMissingConstraintColumns() throws Exception {
+        // P1 from review: findSlotsByColumn() registers a PARTIAL unique key when the scan's
+        // output does not contain every constrained column (e.g. a non-base index that only
+        // covers (a, c) of a table-level UNIQUE(a, b)).
+        // Output {a, c} ∩ constraint {a, b} = {a}: {a} must NOT be advertised as unique,
+        // otherwise EliminateGroupByKey derives a -> c and wrongly wraps c for GROUP BY a, c.
+        createTable("create table test.idx_t (\n"
+                + "a int not null,\n"
+                + "b int not null,\n"
+                + "c int not null)\n"
+                + "distributed by hash(a) buckets 3\n"
+                + "properties('replication_num'='1')");
+        addConstraint("alter table test.idx_t add constraint uk unique (a, b)");
+
+        Database db = Env.getCurrentInternalCatalog().getDbOrMetaException("test");
+        OlapTable table = (OlapTable) db.getTableOrMetaException("idx_t", Table.TableType.OLAP);
+        // Simulate a scan whose output only exposes (a, c) — the same shape a non-base index
+        // (e.g. an MV index) would produce. cachedOutput overrides the scan's output slots.
+        List<Slot> partialOutput = ImmutableList.of(
+                SlotReference.fromColumn(StatementScopeIdGenerator.getExprIdGenerator().getNextId(),
+                        table, table.getColumn("a"), "a", ImmutableList.of()),
+                SlotReference.fromColumn(StatementScopeIdGenerator.getExprIdGenerator().getNextId(),
+                        table, table.getColumn("c"), "c", ImmutableList.of()));
+        LogicalOlapScan scan = new LogicalOlapScan(StatementScopeIdGenerator.newRelationId(), table,
+                ImmutableList.of("test"), Optional.empty(), Optional.empty(),
+                table.getPartitionIds(), false, ImmutableList.of(),
+                table.getBaseIndexId(), false, PreAggStatus.unset(), ImmutableList.of(), ImmutableList.of(),
+                Maps.newHashMap(), Optional.of(partialOutput), Optional.empty(), false, Maps.newHashMap(),
+                ImmutableList.of(), ImmutableList.of(), ImmutableList.of(), ImmutableList.of(),
+                Optional.empty(), Optional.empty(), ImmutableList.of(), Optional.empty(), "");
+
+        List<Slot> output = scan.getOutput();
+        Assertions.assertEquals(2, output.size(), "scan output: " + output);
+        Slot a = output.get(0);
+        Assertions.assertEquals("a", ((SlotReference) a).getName());
+        // UNIQUE(a,b) requires BOTH columns; the scan output misses b, so {a} is not unique
+        Assertions.assertFalse(scan.getLogicalProperties().getTrait().isUnique(a),
+                "partial constraint registration: {a} must not be unique when the scan output misses column b");
+    }
+
+    @Test
+    void testMorReadAsDupSuppressesUniqueConstraint() throws Exception {
+        // P1 from review: for MOR unique-key tables read as DUP (read_mor_as_dup_tables),
+        // the data exposes every version (e.g. (1,10),(1,20),(1,30)) so the unique key k is
+        // NOT unique. LogicalOlapScan.computeUnique() must suppress the constraint imported
+        // by super.computeUnique() before its own raw-version guard returns.
+        createTable("create table test.mor_t (k int not null, v int not null) "
+                + "unique key(k) distributed by hash(k) buckets 3 "
+                + "properties('replication_num'='1', 'enable_unique_key_merge_on_write'='false')");
+        addConstraint("alter table test.mor_t add constraint uk unique (k)");
+        connectContext.getSessionVariable().readMorAsDupTables = "*";
+        try {
+            Database db = Env.getCurrentInternalCatalog().getDbOrMetaException("test");
+            OlapTable table = (OlapTable) db.getTableOrMetaException("mor_t", Table.TableType.OLAP);
+            LogicalOlapScan scan = new LogicalOlapScan(StatementScopeIdGenerator.newRelationId(), table);
+            Slot k = scan.getOutput().get(0);
+            Assertions.assertEquals("k", ((SlotReference) k).getName());
+            Assertions.assertFalse(scan.getLogicalProperties().getTrait().isUnique(k),
+                    "MOR table read as DUP exposes all versions: k must not be unique");
+        } finally {
+            connectContext.getSessionVariable().readMorAsDupTables = "";
+        }
+    }
+
 }