[fix](rewrite rule) Reject mismatched compare plan in PullUpJoinFromUnionAll

fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStreamWrapper.java

@@ -213,6 +213,10 @@ public OlapTable getBaseTable() {
  return baseTable;
  }
 
+ public KeysType getStreamKeysType() {
+ return keysType;
+ }
+
  public BaseTableStream.StreamScanType getStreamScanType() {
  if (keysType == KeysType.DUP_KEYS) {
  return BaseTableStream.StreamScanType.APPEND_ONLY;

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PullUpJoinFromUnionAll.java

@@ -17,7 +17,6 @@
 
 package org.apache.doris.nereids.rules.rewrite;
 
-import org.apache.doris.catalog.constraint.TableIdentifier;
 import org.apache.doris.common.Pair;
 import org.apache.doris.nereids.rules.Rule;
 import org.apache.doris.nereids.rules.RuleType;
@@ -577,14 +576,14 @@ boolean isNotSupported(Plan plan) {
  }
 
  boolean comparePlan(Plan plan1, Plan plan2) {
+ if (plan1.getOutput().size() != plan2.getOutput().size()) {
+ return false;
+ }
  boolean isEqual = true;
  if (plan1 instanceof LogicalCatalogRelation && plan2 instanceof LogicalCatalogRelation) {
- isEqual = new TableIdentifier(((LogicalCatalogRelation) plan1).getTable())
- .equals(new TableIdentifier(((LogicalCatalogRelation) plan2).getTable()));
+ isEqual = ((LogicalCatalogRelation) plan1)
+ .hasSameScanSemantics((LogicalCatalogRelation) plan2);
  } else if (plan1 instanceof LogicalProject && plan2 instanceof LogicalProject) {
- if (plan1.getOutput().size() != plan2.getOutput().size()) {
- isEqual = false;
- }
  for (int i = 0; isEqual && i < plan2.getOutput().size(); i++) {
  Expression expr1 = ((LogicalProject<?>) plan1).getProjects().get(i);
  Expression expr2 = ((LogicalProject<?>) plan2).getProjects().get(i);

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

@@ -23,6 +23,7 @@
 import org.apache.doris.catalog.TableIf;
 import org.apache.doris.catalog.constraint.ConstraintManager;
 import org.apache.doris.catalog.constraint.PrimaryKeyConstraint;
+import org.apache.doris.catalog.constraint.TableIdentifier;
 import org.apache.doris.catalog.constraint.UniqueConstraint;
 import org.apache.doris.catalog.info.TableNameInfo;
 import org.apache.doris.catalog.stream.StreamReadMode;
@@ -251,6 +252,47 @@ public LogicalCatalogRelation withVirtualColumns(List<NamedExpression> virtualCo
  return this;
  }
 
+ /** Compare whether two catalog relations read the same data with the same output semantics. */
+ public final boolean hasSameScanSemantics(LogicalCatalogRelation other) {
+ if (other == null || getClass() != other.getClass()) {
+ return false;
+ }
+ if (!hasSameTableIdentity(other)) {
+ return false;
+ }
+ if (getOutput().size() != other.getOutput().size()) {
+ return false;
+ }
+ for (int i = 0; i < getOutput().size(); i++) {
+ if (!hasSameOutputSlotSemantics(getOutput().get(i), other.getOutput().get(i))) {
+ return false;
+ }
+ }
+ return hasSameScanState(other);
+ }
+
+ protected boolean hasSameTableIdentity(LogicalCatalogRelation other) {
+ if (!Utils.isSameClass(this, other)) {
+ return false;
+ }
+ return new TableIdentifier(table).equals(new TableIdentifier(other.table));
+ }
+
+ protected boolean hasSameScanState(LogicalCatalogRelation other) {
+ return false;
+ }
+
+ private boolean hasSameOutputSlotSemantics(Slot left, Slot right) {
+ if (!(left instanceof SlotReference) || !(right instanceof SlotReference)) {
+ return false;
+ }
+ SlotReference leftSlot = (SlotReference) left;
+ SlotReference rightSlot = (SlotReference) right;
+ return Objects.equals(left.getClass(), right.getClass())
+ && Objects.equals(leftSlot.getName(), rightSlot.getName())
+ && Objects.equals(leftSlot.getSubPath(), rightSlot.getSubPath());
+ }
+
  public abstract LogicalCatalogRelation withRelationId(RelationId relationId);
 
  /**

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

@@ -198,6 +198,18 @@ public boolean equals(Object o) {
  return super.equals(o) && Objects.equals(selectedPartitions, ((LogicalFileScan) o).selectedPartitions);
  }
 
+ @Override
+ protected boolean hasSameScanState(LogicalCatalogRelation other) {
+ if (!Utils.isSameClass(this, other)) {
+ return false;
+ }
+ LogicalFileScan that = (LogicalFileScan) other;
+ return Objects.equals(selectedPartitions, that.selectedPartitions)
+ && Objects.equals(tableSample, that.tableSample)
+ && hasSameSnapshot(tableSnapshot, that.tableSnapshot)
+ && hasSameScanParams(scanParams, that.scanParams);
+ }
+
  @Override
  public List<Slot> computeOutput() {
  if (cachedOutputs.isPresent()) {
@@ -266,6 +278,23 @@ public boolean supportPruneNestedColumn() {
  return false;
  }
 
+ private boolean hasSameSnapshot(Optional<TableSnapshot> left, Optional<TableSnapshot> right) {
+ if (!left.isPresent() || !right.isPresent()) {
+ return left.isPresent() == right.isPresent();
+ }
+ return left.get().getType() == right.get().getType()
+ && Objects.equals(left.get().getValue(), right.get().getValue());
+ }
+
+ private boolean hasSameScanParams(Optional<TableScanParams> left, Optional<TableScanParams> right) {
+ if (!left.isPresent() || !right.isPresent()) {
+ return left.isPresent() == right.isPresent();
+ }
+ return Objects.equals(left.get().getParamType(), right.get().getParamType())
+ && Objects.equals(left.get().getMapParams(), right.get().getMapParams())
+ && Objects.equals(left.get().getListParams(), right.get().getListParams());
+ }
+
  /**
  * SelectedPartitions contains the selected partitions and the total partition number.
  * Mainly for hive table partition pruning.

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

@@ -121,6 +121,16 @@ public Optional<IncrementalRelation> getIncrementalRelation() {
  return incrementalRelation;
  }
 
+ @Override
+ protected boolean hasSameScanState(LogicalCatalogRelation other) {
+ if (!Utils.isSameClass(this, other)) {
+ return false;
+ }
+ LogicalHudiScan that = (LogicalHudiScan) other;
+ // IncrementalRelation contains the resolved Hudi timeline and split state and has no value equality.
+ return super.hasSameScanState(other) && Objects.equals(incrementalRelation, that.incrementalRelation);
+ }
+
  /**
  * replace incremental params as AND expression
  * incr('beginTime'='20240308110257169', 'endTime'='20240308110677278') =>

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

@@ -95,6 +95,11 @@ public LogicalOdbcScan withTableAlias(String tableAlias) {
  Optional.of(getLogicalProperties()), tableAlias));
  }
 
+ @Override
+ protected boolean hasSameScanState(LogicalCatalogRelation other) {
+ return Utils.isSameClass(this, other);
+ }
+
  @Override
  public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
  return visitor.visitLogicalOdbcScan(this, context);

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

@@ -1207,6 +1207,22 @@ public LogicalOlapScan withTableScanParams(TableScanParams scanParams) {
  Optional.of(scanParams)));
  }
 
+ @Override
+ protected boolean hasSameScanState(LogicalCatalogRelation other) {
+ if (!Utils.isSameClass(this, other)) {
+ return false;
+ }
+ LogicalOlapScan that = (LogicalOlapScan) other;
+ return selectedIndexId == that.selectedIndexId
+ && indexSelected == that.indexSelected
+ && Objects.equals(selectedPartitionIds, that.selectedPartitionIds)
+ && Objects.equals(manuallySpecifiedPartitions, that.manuallySpecifiedPartitions)
+ && Objects.equals(selectedTabletIds, that.selectedTabletIds)
+ && Objects.equals(manuallySpecifiedTabletIds, that.manuallySpecifiedTabletIds)
+ && Objects.equals(tableSample, that.tableSample)
+ && hasSameScanParams(scanParams, that.scanParams);
+ }
+
  @Override
  public boolean supportPruneNestedColumn() {
  return true;
@@ -1215,4 +1231,13 @@ public boolean supportPruneNestedColumn() {
  public Optional<TableScanParams> getScanParams() {
  return scanParams;
  }
+
+ private boolean hasSameScanParams(Optional<TableScanParams> left, Optional<TableScanParams> right) {
+ if (!left.isPresent() || !right.isPresent()) {
+ return left.isPresent() == right.isPresent();
+ }
+ return Objects.equals(left.get().getParamType(), right.get().getParamType())
+ && Objects.equals(left.get().getMapParams(), right.get().getMapParams())
+ && Objects.equals(left.get().getListParams(), right.get().getListParams());
+ }
 }

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

@@ -21,6 +21,7 @@
 import org.apache.doris.catalog.Column;
 import org.apache.doris.catalog.OlapTable;
 import org.apache.doris.catalog.Table;
+import org.apache.doris.catalog.constraint.TableIdentifier;
 import org.apache.doris.catalog.stream.OlapTableStreamWrapper;
 import org.apache.doris.catalog.stream.StreamReadMode;
 import org.apache.doris.common.IdGenerator;
@@ -464,6 +465,30 @@ public LogicalOlapTableStreamScan withReadMode(StreamReadMode readMode) {
  scanParams, readMode));
  }
 
+ @Override
+ protected boolean hasSameTableIdentity(LogicalCatalogRelation other) {
+ if (!Utils.isSameClass(this, other)) {
+ return false;
+ }
+ LogicalOlapTableStreamScan that = (LogicalOlapTableStreamScan) other;
+ return Objects.equals(getTable().getStreamDbId(), that.getTable().getStreamDbId())
+ && Objects.equals(getTable().getStreamId(), that.getTable().getStreamId())
+ && new TableIdentifier(getTable().getBaseTable())
+ .equals(new TableIdentifier(that.getTable().getBaseTable()));
+ }
+
+ @Override
+ protected boolean hasSameScanState(LogicalCatalogRelation other) {
+ if (!Utils.isSameClass(this, other)) {
+ return false;
+ }
+ LogicalOlapTableStreamScan that = (LogicalOlapTableStreamScan) other;
+ return super.hasSameScanState(other)
+ && readMode == that.readMode
+ && getTable().getStreamKeysType() == that.getTable().getStreamKeysType()
+ && Objects.equals(getTable().getOutputUpdateMap(), that.getTable().getOutputUpdateMap());
+ }
+
  @Override
  public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
  return visitor.visitLogicalOlapTableStreamScan(this, context);

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

@@ -113,6 +113,19 @@ public List<Expression> getFrontendConjuncts() {
  return frontendConjuncts;
  }
 
+ @Override
+ protected boolean hasSameScanState(LogicalCatalogRelation other) {
+ if (!Utils.isSameClass(this, other)) {
+ return false;
+ }
+ LogicalSchemaScan that = (LogicalSchemaScan) other;
+ return filterPushed == that.filterPushed
+ && Objects.equals(schemaCatalog, that.schemaCatalog)
+ && Objects.equals(schemaDatabase, that.schemaDatabase)
+ && Objects.equals(schemaTable, that.schemaTable)
+ && Objects.equals(frontendConjuncts, that.frontendConjuncts);
+ }
+
  @Override
  public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
  return visitor.visitLogicalSchemaScan(this, context);

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

@@ -85,6 +85,11 @@ public LogicalTestScan withTableAlias(String tableAlias) {
  Optional.of(getLogicalProperties()), tableAlias));
  }
 
+ @Override
+ protected boolean hasSameScanState(LogicalCatalogRelation other) {
+ return Utils.isSameClass(this, other);
+ }
+
  @Override
  public LogicalTestScan withRelationId(RelationId relationId) {
  throw new RuntimeException("should not call LogicalTestScan's withRelationId method");

fe/fe-core/src/main/java/org/apache/doris/nereids/util/Utils.java

@@ -62,6 +62,13 @@
 public class Utils {
  public static final boolean enableAssert;
 
+ /**
+ * Check whether two objects are non-null and have the same concrete class.
+ */
+ public static boolean isSameClass(Object left, Object right) {
+ return left != null && right != null && left.getClass() == right.getClass();
+ }
+
  static {
  boolean enabled = false;
  // if run jvm with -ea or -enableassertions, the assert statement will be executed