feat(planner): Add MaterializedViewScanNode

presto-analyzer/src/main/java/com/facebook/presto/sql/analyzer/Analysis.java

@@ -21,6 +21,7 @@
 import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.ColumnMetadata;
 import com.facebook.presto.spi.ConnectorId;
+import com.facebook.presto.spi.MaterializedViewDefinition;
 import com.facebook.presto.spi.TableHandle;
 import com.facebook.presto.spi.analyzer.AccessControlInfo;
 import com.facebook.presto.spi.analyzer.AccessControlInfoForTable;
@@ -203,6 +204,8 @@ public class Analysis
 
     private final Map<QualifiedObjectName, String> materializedViews = new LinkedHashMap<>();
 
+    private final Map<NodeRef<Table>, MaterializedViewInfo> materializedViewInfoMap = new LinkedHashMap<>();
+
     private Optional<String> expandedQuery = Optional.empty();
 
     // Keeps track of the subquery we are visiting, so we have access to base query information when processing materialized view status
@@ -817,6 +820,19 @@ public boolean hasTableRegisteredForMaterializedView(Table view, Table table)
         return tablesForMaterializedView.containsEntry(NodeRef.of(view), table);
     }
 
+    public void setMaterializedViewInfo(Table table, MaterializedViewInfo materializedViewInfo)
+    {
+        requireNonNull(table, "table is null");
+        requireNonNull(materializedViewInfo, "materializedViewInfo is null");
+        materializedViewInfoMap.put(NodeRef.of(table), materializedViewInfo);
+    }
+
+    public Optional<MaterializedViewInfo> getMaterializedViewInfo(Table table)
+    {
+        requireNonNull(table, "table is null");
+        return Optional.ofNullable(materializedViewInfoMap.get(NodeRef.of(table)));
+    }
+
     public void setSampleRatio(SampledRelation relation, double ratio)
     {
         sampleRatios.put(NodeRef.of(relation), ratio);
@@ -1211,6 +1227,47 @@ public Query getQuery()
         }
     }
 
+    @Immutable
+    public static final class MaterializedViewInfo
+    {
+        private final QualifiedObjectName materializedViewName;
+        private final Table dataTable;
+        private final Query viewQuery;
+        private final MaterializedViewDefinition materializedViewDefinition;
+
+        public MaterializedViewInfo(
+                QualifiedObjectName materializedViewName,
+                Table dataTable,
+                Query viewQuery,
+                MaterializedViewDefinition materializedViewDefinition)
+        {
+            this.materializedViewName = requireNonNull(materializedViewName, "materializedViewName is null");
+            this.dataTable = requireNonNull(dataTable, "dataTable is null");
+            this.viewQuery = requireNonNull(viewQuery, "viewQuery is null");
+            this.materializedViewDefinition = requireNonNull(materializedViewDefinition, "materializedViewDefinition is null");
+        }
+
+        public QualifiedObjectName getMaterializedViewName()
+        {
+            return materializedViewName;
+        }
+
+        public Table getDataTable()
+        {
+            return dataTable;
+        }
+
+        public Query getViewQuery()
+        {
+            return viewQuery;
+        }
+
+        public MaterializedViewDefinition getMaterializedViewDefinition()
+        {
+            return materializedViewDefinition;
+        }
+    }
+
     public static final class JoinUsingAnalysis
     {
         private final List<Integer> leftJoinFields;

presto-main-base/src/main/java/com/facebook/presto/sql/analyzer/StatementAnalyzer.java

@@ -2303,21 +2303,57 @@ private Scope processMaterializedView(
             analysis.getAccessControlReferences().addMaterializedViewDefinitionReference(materializedViewName, materializedViewDefinition);
 
             analysis.registerMaterializedViewForAnalysis(materializedViewName, materializedView, materializedViewDefinition.getOriginalSql());
-            String newSql = getMaterializedViewSQL(materializedView, materializedViewName, materializedViewDefinition, scope);
 
-            Query query = (Query) sqlParser.createStatement(newSql, createParsingOptions(session, warningCollector));
-            analysis.registerNamedQuery(materializedView, query, true);
+            if (isLegacyMaterializedViews(session)) {
+                // Legacy SQL stitching approach: create UNION query with base tables
+                String newSql = getMaterializedViewSQL(materializedView, materializedViewName, materializedViewDefinition, scope);
 
-            Scope queryScope = process(query, scope);
-            RelationType relationType = queryScope.getRelationType().withAlias(materializedViewName.getObjectName(), null);
-            analysis.unregisterMaterializedViewForAnalysis(materializedView);
+                Query query = (Query) sqlParser.createStatement(newSql, createParsingOptions(session, warningCollector));
+                analysis.registerNamedQuery(materializedView, query, true);
 
-            Scope accessControlScope = Scope.builder()
-                    .withRelationType(RelationId.anonymous(), relationType)
-                    .build();
-            analyzeFiltersAndMasks(materializedView, materializedViewName, accessControlScope, relationType.getAllFields());
+                Scope queryScope = process(query, scope);
+                RelationType relationType = queryScope.getRelationType().withAlias(materializedViewName.getObjectName(), null);
+                analysis.unregisterMaterializedViewForAnalysis(materializedView);
+
+                Scope accessControlScope = Scope.builder()
+                        .withRelationType(RelationId.anonymous(), relationType)
+                        .build();
+                analyzeFiltersAndMasks(materializedView, materializedViewName, accessControlScope, relationType.getAllFields());
 
-            return createAndAssignScope(materializedView, scope, relationType);
+                return createAndAssignScope(materializedView, scope, relationType);
+            }
+            else {
+                Query viewQuery = (Query) sqlParser.createStatement(
+                        materializedViewDefinition.getOriginalSql(),
+                        createParsingOptions(session, warningCollector));
+
+                QualifiedName dataTableName = QualifiedName.of(
+                        materializedViewName.getCatalogName(),
+                        materializedViewName.getSchemaName(),
+                        materializedViewDefinition.getTable());
+                Table dataTable = new Table(dataTableName);
+
+                Analysis.MaterializedViewInfo mvInfo = new Analysis.MaterializedViewInfo(
+                        materializedViewName,
+                        dataTable,
+                        viewQuery,
+                        materializedViewDefinition);
+                analysis.setMaterializedViewInfo(materializedView, mvInfo);
+
+                // Process the view query to analyze base tables for access control
+                process(viewQuery, scope);
+
+                Scope queryScope = process(dataTable, scope);
+                RelationType relationType = queryScope.getRelationType().withOnlyVisibleFields().withAlias(materializedViewName.getObjectName(), null);
+                analysis.unregisterMaterializedViewForAnalysis(materializedView);
+
+                Scope accessControlScope = Scope.builder()
+                        .withRelationType(RelationId.anonymous(), relationType)
+                        .build();
+                analyzeFiltersAndMasks(materializedView, materializedViewName, accessControlScope, relationType.getAllFields());
+
+                return createAndAssignScope(materializedView, scope, relationType);
+            }
         }
 
         private String getMaterializedViewSQL(

presto-main-base/src/main/java/com/facebook/presto/sql/planner/PlanOptimizers.java

@@ -55,6 +55,7 @@
 import com.facebook.presto.sql.planner.iterative.rule.InlineSqlFunctions;
 import com.facebook.presto.sql.planner.iterative.rule.LeftJoinNullFilterToSemiJoin;
 import com.facebook.presto.sql.planner.iterative.rule.LeftJoinWithArrayContainsToEquiJoinCondition;
+import com.facebook.presto.sql.planner.iterative.rule.MaterializedViewRewrite;
 import com.facebook.presto.sql.planner.iterative.rule.MergeDuplicateAggregation;
 import com.facebook.presto.sql.planner.iterative.rule.MergeFilters;
 import com.facebook.presto.sql.planner.iterative.rule.MergeLimitWithDistinct;
@@ -314,6 +315,13 @@ public PlanOptimizers(
 
         builder.add(new LogicalCteOptimizer(metadata));
 
+        builder.add(new IterativeOptimizer(
+                metadata,
+                ruleStats,
+                statsCalculator,
+                estimatedExchangesCostCalculator,
+                ImmutableSet.of(new MaterializedViewRewrite(metadata))));
+
         IterativeOptimizer inlineProjections = new IterativeOptimizer(
                 metadata,
                 ruleStats,

presto-main-base/src/main/java/com/facebook/presto/sql/planner/RelationPlanner.java

@@ -15,6 +15,7 @@
 
 import com.facebook.presto.Session;
 import com.facebook.presto.SystemSessionProperties;
+import com.facebook.presto.common.QualifiedObjectName;
 import com.facebook.presto.common.predicate.TupleDomain;
 import com.facebook.presto.common.type.ArrayType;
 import com.facebook.presto.common.type.MapType;
@@ -35,6 +36,7 @@
 import com.facebook.presto.spi.plan.FilterNode;
 import com.facebook.presto.spi.plan.IntersectNode;
 import com.facebook.presto.spi.plan.JoinNode;
+import com.facebook.presto.spi.plan.MaterializedViewScanNode;
 import com.facebook.presto.spi.plan.PlanNode;
 import com.facebook.presto.spi.plan.PlanNodeIdAllocator;
 import com.facebook.presto.spi.plan.ProjectNode;
@@ -182,6 +184,11 @@ public RelationPlan process(Node node, @Nullable SqlPlannerContext context)
     @Override
     protected RelationPlan visitTable(Table node, SqlPlannerContext context)
     {
+        Optional<Analysis.MaterializedViewInfo> materializedViewInfo = analysis.getMaterializedViewInfo(node);
+        if (materializedViewInfo.isPresent()) {
+            return planMaterializedView(node, materializedViewInfo.get(), context);
+        }
+
         NamedQuery namedQuery = analysis.getNamedQuery(node);
         Scope scope = analysis.getScope(node);
 
@@ -304,6 +311,59 @@ private RelationPlan addColumnMasks(Table table, RelationPlan plan, SqlPlannerCo
         return new RelationPlan(planBuilder.getRoot(), plan.getScope(), newMappings.build());
     }
 
+    private RelationPlan planMaterializedView(Table node, Analysis.MaterializedViewInfo materializedViewInfo, SqlPlannerContext context)
+    {
+        RelationPlan dataTablePlan = process(materializedViewInfo.getDataTable(), context);
+        RelationPlan viewQueryPlan = process(materializedViewInfo.getViewQuery(), context);
+
+        Scope scope = analysis.getScope(node);
+
+        QualifiedObjectName materializedViewName = materializedViewInfo.getMaterializedViewName();
+
+        RelationType dataTableDescriptor = dataTablePlan.getDescriptor();
+        List<VariableReferenceExpression> dataTableVariables = dataTablePlan.getFieldMappings();
+        List<VariableReferenceExpression> viewQueryVariables = viewQueryPlan.getFieldMappings();
+
+        checkArgument(
+                dataTableVariables.size() == viewQueryVariables.size(),
+                "Materialized view %s has mismatched field counts: data table has %s fields but view query has %s fields",
+                materializedViewName,
+                dataTableVariables.size(),
+                viewQueryVariables.size());
+
+        ImmutableList.Builder<VariableReferenceExpression> outputVariablesBuilder = ImmutableList.builder();
+        ImmutableMap.Builder<VariableReferenceExpression, VariableReferenceExpression> dataTableMappingsBuilder = ImmutableMap.builder();
+        ImmutableMap.Builder<VariableReferenceExpression, VariableReferenceExpression> viewQueryMappingsBuilder = ImmutableMap.builder();
+
+        for (Field field : dataTableDescriptor.getVisibleFields()) {
+            int fieldIndex = dataTableDescriptor.indexOf(field);
+            VariableReferenceExpression dataTableVar = dataTableVariables.get(fieldIndex);
+            VariableReferenceExpression viewQueryVar = viewQueryVariables.get(fieldIndex);
+
+            VariableReferenceExpression outputVar = variableAllocator.newVariable(dataTableVar);
+            outputVariablesBuilder.add(outputVar);
+
+            dataTableMappingsBuilder.put(outputVar, dataTableVar);
+            viewQueryMappingsBuilder.put(outputVar, viewQueryVar);
+        }
+
+        List<VariableReferenceExpression> outputVariables = outputVariablesBuilder.build();
+        Map<VariableReferenceExpression, VariableReferenceExpression> dataTableMappings = dataTableMappingsBuilder.build();
+        Map<VariableReferenceExpression, VariableReferenceExpression> viewQueryMappings = viewQueryMappingsBuilder.build();
+
+        MaterializedViewScanNode mvScanNode = new MaterializedViewScanNode(
+                getSourceLocation(node.getLocation()),
+                idAllocator.getNextId(),
+                dataTablePlan.getRoot(),
+                viewQueryPlan.getRoot(),
+                materializedViewName,
+                dataTableMappings,
+                viewQueryMappings,
+                outputVariables);
+
+        return new RelationPlan(mvScanNode, scope, outputVariables);
+    }
+
     @Override
     protected RelationPlan visitTableFunctionInvocation(TableFunctionInvocation node, SqlPlannerContext context)
     {

presto-main-base/src/main/java/com/facebook/presto/sql/planner/iterative/rule/MaterializedViewRewrite.java

@@ -0,0 +1,84 @@
+/*
+ * Licensed 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 com.facebook.presto.sql.planner.iterative.rule;
+
+import com.facebook.presto.Session;
+import com.facebook.presto.matching.Captures;
+import com.facebook.presto.matching.Pattern;
+import com.facebook.presto.metadata.Metadata;
+import com.facebook.presto.spi.MaterializedViewStatus;
+import com.facebook.presto.spi.analyzer.MetadataResolver;
+import com.facebook.presto.spi.plan.Assignments;
+import com.facebook.presto.spi.plan.MaterializedViewScanNode;
+import com.facebook.presto.spi.plan.PlanNode;
+import com.facebook.presto.spi.plan.ProjectNode;
+import com.facebook.presto.spi.relation.VariableReferenceExpression;
+import com.facebook.presto.sql.planner.iterative.Rule;
+
+import java.util.Map;
+
+import static com.facebook.presto.SystemSessionProperties.isLegacyMaterializedViews;
+import static com.facebook.presto.SystemSessionProperties.isMaterializedViewDataConsistencyEnabled;
+import static com.facebook.presto.spi.plan.ProjectNode.Locality.LOCAL;
+import static com.facebook.presto.sql.planner.plan.Patterns.materializedViewScan;
+import static com.google.common.base.Preconditions.checkState;
+import static java.util.Objects.requireNonNull;
+
+public class MaterializedViewRewrite
+        implements Rule<MaterializedViewScanNode>
+{
+    private final Metadata metadata;
+
+    public MaterializedViewRewrite(Metadata metadata)
+    {
+        this.metadata = requireNonNull(metadata, "metadata is null");
+    }
+
+    @Override
+    public Pattern<MaterializedViewScanNode> getPattern()
+    {
+        return materializedViewScan();
+    }
+
+    @Override
+    public Result apply(MaterializedViewScanNode node, Captures captures, Context context)
+    {
+        Session session = context.getSession();
+        checkState(!isLegacyMaterializedViews(session), "Materialized view rewrite rule should not fire when legacy materialized views are enabled");
+        checkState(isMaterializedViewDataConsistencyEnabled(session), "Materialized view rewrite rule should not fire when materialized view data consistency is disabled");
+
+        MetadataResolver metadataResolver = metadata.getMetadataResolver(session);
+
+        MaterializedViewStatus status = metadataResolver.getMaterializedViewStatus(node.getMaterializedViewName());
+
+        boolean useDataTable = status.isFullyMaterialized();
+        PlanNode chosenPlan = useDataTable ? node.getDataTablePlan() : node.getViewQueryPlan();
+        Map<VariableReferenceExpression, VariableReferenceExpression> chosenMappings =
+                useDataTable ? node.getDataTableMappings() : node.getViewQueryMappings();
+
+        Assignments.Builder assignments = Assignments.builder();
+        for (VariableReferenceExpression outputVariable : node.getOutputVariables()) {
+            VariableReferenceExpression sourceVariable = chosenMappings.get(outputVariable);
+            requireNonNull(sourceVariable, "No mapping found for output variable: " + outputVariable);
+            assignments.put(outputVariable, sourceVariable);
+        }
+
+        return Result.ofPlanNode(new ProjectNode(
+                node.getSourceLocation(),
+                context.getIdAllocator().getNextId(),
+                chosenPlan,
+                assignments.build(),
+                LOCAL));
+    }
+}

presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/ApplyConnectorOptimization.java

@@ -34,6 +34,7 @@
 import com.facebook.presto.spi.plan.JoinNode;
 import com.facebook.presto.spi.plan.LimitNode;
 import com.facebook.presto.spi.plan.MarkDistinctNode;
+import com.facebook.presto.spi.plan.MaterializedViewScanNode;
 import com.facebook.presto.spi.plan.PlanNode;
 import com.facebook.presto.spi.plan.PlanNodeIdAllocator;
 import com.facebook.presto.spi.plan.ProjectNode;
@@ -89,6 +90,7 @@ public class ApplyConnectorOptimization
             ProjectNode.class,
             AggregationNode.class,
             MarkDistinctNode.class,
+            MaterializedViewScanNode.class,
             UnionNode.class,
             IntersectNode.class,
             ExceptNode.class,

presto-main-base/src/main/java/com/facebook/presto/sql/planner/plan/Patterns.java

@@ -27,6 +27,7 @@
 import com.facebook.presto.spi.plan.JoinType;
 import com.facebook.presto.spi.plan.LimitNode;
 import com.facebook.presto.spi.plan.MarkDistinctNode;
+import com.facebook.presto.spi.plan.MaterializedViewScanNode;
 import com.facebook.presto.spi.plan.OutputNode;
 import com.facebook.presto.spi.plan.PlanNode;
 import com.facebook.presto.spi.plan.ProjectNode;
@@ -134,6 +135,11 @@ public static Pattern<MarkDistinctNode> markDistinct()
         return typeOf(MarkDistinctNode.class);
     }
 
+    public static Pattern<MaterializedViewScanNode> materializedViewScan()
+    {
+        return typeOf(MaterializedViewScanNode.class);
+    }
+
     public static Pattern<OutputNode> output()
     {
         return typeOf(OutputNode.class);