[feature](Nereids): Support SchemaScan.

fe/fe-core/src/main/java/org/apache/doris/nereids/cost/CostCalculator.java

@@ -35,6 +35,7 @@
 import org.apache.doris.nereids.trees.plans.physical.PhysicalOlapScan;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalProject;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalQuickSort;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalSchemaScan;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalStorageLayerAggregate;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalTopN;
 import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
@@ -62,7 +63,7 @@ public class CostCalculator {
      * AGG is time-consuming operator. From the perspective of rowCount, nereids may choose Plan1,
      * because `Agg1 join Agg2` generates few tuples. But in Plan1, Agg1 and Agg2 are done in serial, in Plan2, Agg1 and
      * Agg2 are done in parallel. And hence, Plan1 should be punished.
-     *
+     * <p>
      * An example is tpch q15.
      */
     static final double HEAVY_OPERATOR_PUNISH_FACTOR = 6.0;
@@ -100,6 +101,11 @@ public CostEstimate visitPhysicalOlapScan(PhysicalOlapScan physicalOlapScan, Pla
             return CostEstimate.ofCpu(statistics.getRowCount());
         }
 
+        public CostEstimate visitPhysicalSchemaScan(PhysicalSchemaScan physicalSchemaScan, PlanContext context) {
+            StatsDeriveResult statistics = context.getStatisticsWithCheck();
+            return CostEstimate.ofCpu(statistics.getRowCount());
+        }
+
         @Override
         public CostEstimate visitPhysicalStorageLayerAggregate(
                 PhysicalStorageLayerAggregate storageLayerAggregate, PlanContext context) {

fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java

@@ -78,6 +78,7 @@
 import org.apache.doris.nereids.trees.plans.physical.PhysicalProject;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalQuickSort;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalRepeat;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalSchemaScan;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalSetOperation;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalStorageLayerAggregate;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalTVFRelation;
@@ -105,6 +106,7 @@
 import org.apache.doris.planner.PlanNode;
 import org.apache.doris.planner.RepeatNode;
 import org.apache.doris.planner.ScanNode;
+import org.apache.doris.planner.SchemaScanNode;
 import org.apache.doris.planner.SelectNode;
 import org.apache.doris.planner.SetOperationNode;
 import org.apache.doris.planner.SortNode;
@@ -446,6 +448,7 @@ public PlanFragment visitPhysicalOlapScan(PhysicalOlapScan olapScan, PlanTransla
         OlapTable olapTable = olapScan.getTable();
         TupleDescriptor tupleDescriptor = generateTupleDesc(slotList, olapTable, context);
         tupleDescriptor.setTable(olapTable);
+
         OlapScanNode olapScanNode = new OlapScanNode(context.nextPlanNodeId(), tupleDescriptor, "OlapScanNode");
         if (olapScan.getStats() != null) {
             olapScanNode.setCardinality((long) olapScan.getStats().getRowCount());
@@ -491,6 +494,26 @@ public PlanFragment visitPhysicalOlapScan(PhysicalOlapScan olapScan, PlanTransla
         return planFragment;
     }
 
+    @Override
+    public PlanFragment visitPhysicalSchemaScan(PhysicalSchemaScan schemaScan, PlanTranslatorContext context) {
+        Table table = schemaScan.getTable();
+
+        List<Slot> slotList = new ImmutableList.Builder<Slot>()
+                .addAll(schemaScan.getOutput())
+                .addAll(schemaScan.getNonUserVisibleOutput())
+                .build();
+        TupleDescriptor tupleDescriptor = generateTupleDesc(slotList, table, context);
+        tupleDescriptor.setTable(table);
+
+        SchemaScanNode scanNode = new SchemaScanNode(context.nextPlanNodeId(), tupleDescriptor);
+
+        context.getScanNodes().add(scanNode);
+        PlanFragment planFragment =
+                new PlanFragment(context.nextFragmentId(), scanNode, DataPartition.RANDOM);
+        context.addPlanFragment(planFragment);
+        return planFragment;
+    }
+
     @Override
     public PlanFragment visitPhysicalTVFRelation(PhysicalTVFRelation tvfRelation, PlanTranslatorContext context) {
         List<Slot> slots = tvfRelation.getLogicalProperties().getOutput();

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

@@ -39,6 +39,7 @@
 import org.apache.doris.nereids.rules.implementation.LogicalOneRowRelationToPhysicalOneRowRelation;
 import org.apache.doris.nereids.rules.implementation.LogicalProjectToPhysicalProject;
 import org.apache.doris.nereids.rules.implementation.LogicalRepeatToPhysicalRepeat;
+import org.apache.doris.nereids.rules.implementation.LogicalSchemaScanToPhysicalSchemaScan;
 import org.apache.doris.nereids.rules.implementation.LogicalSortToPhysicalQuickSort;
 import org.apache.doris.nereids.rules.implementation.LogicalTVFRelationToPhysicalTVFRelation;
 import org.apache.doris.nereids.rules.implementation.LogicalTopNToPhysicalTopN;
@@ -103,6 +104,7 @@ public class RuleSet {
             .add(new LogicalJoinToHashJoin())
             .add(new LogicalJoinToNestedLoopJoin())
             .add(new LogicalOlapScanToPhysicalOlapScan())
+            .add(new LogicalSchemaScanToPhysicalSchemaScan())
             .add(new LogicalProjectToPhysicalProject())
             .add(new LogicalLimitToPhysicalLimit())
             .add(new LogicalSortToPhysicalQuickSort())

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

@@ -207,6 +207,7 @@ public enum RuleType {
     LOGICAL_EMPTY_RELATION_TO_PHYSICAL_EMPTY_RELATION_RULE(RuleTypeClass.IMPLEMENTATION),
     LOGICAL_LIMIT_TO_PHYSICAL_LIMIT_RULE(RuleTypeClass.IMPLEMENTATION),
     LOGICAL_OLAP_SCAN_TO_PHYSICAL_OLAP_SCAN_RULE(RuleTypeClass.IMPLEMENTATION),
+    LOGICAL_SCHEMA_SCAN_TO_PHYSICAL_SCHEMA_SCAN_RULE(RuleTypeClass.IMPLEMENTATION),
     LOGICAL_ASSERT_NUM_ROWS_TO_PHYSICAL_ASSERT_NUM_ROWS(RuleTypeClass.IMPLEMENTATION),
     STORAGE_LAYER_AGGREGATE_WITHOUT_PROJECT(RuleTypeClass.IMPLEMENTATION),
     STORAGE_LAYER_AGGREGATE_WITH_PROJECT(RuleTypeClass.IMPLEMENTATION),

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

@@ -31,6 +31,7 @@
 import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalSchemaScan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalSubQueryAlias;
 import org.apache.doris.nereids.trees.plans.logical.RelationUtil;
 import org.apache.doris.qe.ConnectContext;
@@ -98,6 +99,8 @@ private LogicalPlan bindWithCurrentDb(CascadesContext cascadesContext, UnboundRe
         } else if (table.getType() == TableType.VIEW) {
             Plan viewPlan = parseAndAnalyzeView(table.getDdlSql(), cascadesContext);
             return new LogicalSubQueryAlias<>(table.getName(), viewPlan);
+        } else if (table.getType() == TableType.SCHEMA) {
+            return new LogicalSchemaScan(RelationUtil.newRelationId(), table, ImmutableList.of(dbName));
         }
         throw new AnalysisException("Unsupported tableType:" + table.getType());
     }
@@ -122,6 +125,8 @@ private LogicalPlan bindWithDbNameFromNamePart(CascadesContext cascadesContext,
         } else if (table.getType() == TableType.VIEW) {
             Plan viewPlan = parseAndAnalyzeView(table.getDdlSql(), cascadesContext);
             return new LogicalSubQueryAlias<>(table.getName(), viewPlan);
+        } else if (table.getType() == TableType.SCHEMA) {
+            return new LogicalSchemaScan(RelationUtil.newRelationId(), table, ImmutableList.of(dbName));
         }
         throw new AnalysisException("Unsupported tableType:" + table.getType());
     }

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/LogicalSchemaScanToPhysicalSchemaScan.java

@@ -0,0 +1,40 @@
+// 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.
+
+package org.apache.doris.nereids.rules.implementation;
+
+import org.apache.doris.nereids.rules.Rule;
+import org.apache.doris.nereids.rules.RuleType;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalSchemaScan;
+
+import java.util.Optional;
+
+/**
+ * SchemaScan
+ */
+public class LogicalSchemaScanToPhysicalSchemaScan extends OneImplementationRuleFactory {
+    @Override
+    public Rule build() {
+        return logicalSchemaScan().then(scan ->
+                new PhysicalSchemaScan(scan.getId(),
+                        scan.getTable(),
+                        scan.getQualifier(),
+                        Optional.empty(),
+                        scan.getLogicalProperties())
+        ).toRule(RuleType.LOGICAL_SCHEMA_SCAN_TO_PHYSICAL_SCHEMA_SCAN_RULE);
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/nereids/stats/StatsCalculator.java

@@ -51,6 +51,7 @@
 import org.apache.doris.nereids.trees.plans.logical.LogicalOneRowRelation;
 import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
 import org.apache.doris.nereids.trees.plans.logical.LogicalRepeat;
+import org.apache.doris.nereids.trees.plans.logical.LogicalSchemaScan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalSort;
 import org.apache.doris.nereids.trees.plans.logical.LogicalTVFRelation;
 import org.apache.doris.nereids.trees.plans.logical.LogicalTopN;
@@ -72,6 +73,7 @@
 import org.apache.doris.nereids.trees.plans.physical.PhysicalProject;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalQuickSort;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalRepeat;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalSchemaScan;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalStorageLayerAggregate;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalTVFRelation;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalTopN;
@@ -164,6 +166,11 @@ public StatsDeriveResult visitLogicalOlapScan(LogicalOlapScan olapScan, Void con
         return computeScan(olapScan);
     }
 
+    @Override
+    public StatsDeriveResult visitLogicalSchemaScan(LogicalSchemaScan schemaScan, Void context) {
+        return new StatsDeriveResult(1);
+    }
+
     @Override
     public StatsDeriveResult visitLogicalTVFRelation(LogicalTVFRelation tvfRelation, Void context) {
         return tvfRelation.getFunction().computeStats(tvfRelation.getOutput());
@@ -244,6 +251,11 @@ public StatsDeriveResult visitPhysicalOlapScan(PhysicalOlapScan olapScan, Void c
         return computeScan(olapScan);
     }
 
+    @Override
+    public StatsDeriveResult visitPhysicalSchemaScan(PhysicalSchemaScan schemaScan, Void context) {
+        return new StatsDeriveResult(1);
+    }
+
     @Override
     public StatsDeriveResult visitPhysicalStorageLayerAggregate(
             PhysicalStorageLayerAggregate storageLayerAggregate, Void context) {
@@ -532,9 +544,9 @@ private Slot getLeftSlot(int fistSetOperation, int outputSlotIdx, SetOperation s
     }
 
     private ColumnStatistic getLeftStats(int fistSetOperation,
-                                         Slot leftSlot,
-                                         Map<Id, ColumnStatistic> leftStatsSlotIdToColumnStats,
-                                         Map<Id, ColumnStatistic> newColumnStatsMap) {
+            Slot leftSlot,
+            Map<Id, ColumnStatistic> leftStatsSlotIdToColumnStats,
+            Map<Id, ColumnStatistic> newColumnStatsMap) {
         return fistSetOperation == 0
                 ? leftStatsSlotIdToColumnStats.get(leftSlot.getExprId())
                 : newColumnStatsMap.get(leftSlot.getExprId());

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

@@ -42,6 +42,7 @@ public enum PlanType {
     LOGICAL_TOP_N,
     LOGICAL_LIMIT,
     LOGICAL_OLAP_SCAN,
+    LOGICAL_SCHEMA_SCAN,
     LOGICAL_APPLY,
     LOGICAL_SELECT_HINT,
     LOGICAL_ASSERT_NUM_ROWS,
@@ -58,6 +59,7 @@ public enum PlanType {
     PHYSICAL_ONE_ROW_RELATION,
     PHYSICAL_OLAP_SCAN,
     PHYSICAL_TVF_RELATION,
+    PHYSICAL_SCHEMA_SCAN,
     PHYSICAL_PROJECT,
     PHYSICAL_FILTER,
     PHYSICAL_BROADCAST_HASH_JOIN,

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

@@ -95,6 +95,8 @@ public class LogicalOlapScan extends LogicalRelation implements CatalogRelation,
     private final boolean partitionPruned;
     private final List<Long> manuallySpecifiedPartitions;
 
+    private final ImmutableList<Long> selectedPartitionIds;
+
     public LogicalOlapScan(RelationId id, OlapTable table) {
         this(id, table, ImmutableList.of());
     }
@@ -128,14 +130,20 @@ public LogicalOlapScan(RelationId id, Table table, List<String> qualifier,
             long selectedIndexId, boolean indexSelected, PreAggStatus preAggStatus, List<Long> partitions) {
 
         super(id, PlanType.LOGICAL_OLAP_SCAN, table, qualifier,
-                groupExpression, logicalProperties, selectedPartitionIds);
+                groupExpression, logicalProperties);
         this.selectedTabletIds = ImmutableList.copyOf(selectedTabletIds);
         this.partitionPruned = partitionPruned;
         this.tabletPruned = tabletPruned;
         this.selectedIndexId = selectedIndexId <= 0 ? getTable().getBaseIndexId() : selectedIndexId;
         this.indexSelected = indexSelected;
         this.preAggStatus = preAggStatus;
         this.manuallySpecifiedPartitions = ImmutableList.copyOf(partitions);
+        this.selectedPartitionIds = ImmutableList.copyOf(
+                Objects.requireNonNull(selectedPartitionIds, "selectedPartitionIds can not be null"));
+    }
+
+    public List<Long> getSelectedPartitionIds() {
+        return selectedPartitionIds;
     }
 
     @Override

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

@@ -44,18 +44,17 @@ public abstract class LogicalRelation extends LogicalLeaf implements Scan {
 
     protected final Table table;
     protected final ImmutableList<String> qualifier;
-    protected final ImmutableList<Long> selectedPartitionIds;
 
     protected final RelationId id;
 
     public LogicalRelation(RelationId id, PlanType type, Table table, List<String> qualifier) {
-        this(id, type, table, qualifier, Optional.empty(), Optional.empty(), Collections.emptyList());
+        this(id, type, table, qualifier, Optional.empty(), Optional.empty());
     }
 
     public LogicalRelation(RelationId id, PlanType type, Optional<GroupExpression> groupExpression,
             Optional<LogicalProperties> logicalProperties) {
         this(id, type, new OlapTable(), Collections.emptyList(), groupExpression,
-                logicalProperties, Collections.emptyList());
+                logicalProperties);
     }
 
     /**
@@ -65,13 +64,10 @@ public LogicalRelation(RelationId id, PlanType type, Optional<GroupExpression> g
      * @param qualifier qualified relation name
      */
     public LogicalRelation(RelationId id, PlanType type, Table table, List<String> qualifier,
-            Optional<GroupExpression> groupExpression, Optional<LogicalProperties> logicalProperties,
-            List<Long> selectedPartitionIds) {
+            Optional<GroupExpression> groupExpression, Optional<LogicalProperties> logicalProperties) {
         super(type, groupExpression, logicalProperties);
         this.table = Objects.requireNonNull(table, "table can not be null");
         this.qualifier = ImmutableList.copyOf(Objects.requireNonNull(qualifier, "qualifier can not be null"));
-        this.selectedPartitionIds = ImmutableList.copyOf(
-                Objects.requireNonNull(selectedPartitionIds, "selectedPartitionIds can not be null"));
         this.id = id;
     }
 
@@ -135,10 +131,6 @@ public String qualifiedName() {
         return Utils.qualifiedName(qualifier, table.getName());
     }
 
-    public List<Long> getSelectedPartitionIds() {
-        return selectedPartitionIds;
-    }
-
     public RelationId getId() {
         return id;
     }