Support WHEN STALE in analyzer

Behavior:
- `FAIL` - If the MV is stale, queries referencing it will fail. Analysis
of the underlying query is never performed while querying the MV.
- `INLINE` - Preserves the current behavior. Even if the MV is stale, it
is expanded like a logical view, and the underlying query is analyzed.

Motivation:
- Avoid analyzing the underlying query every time, which can be costly
when it references many tables.
- Allow using fresh MVs even when some data sources are temporarily
unavailable.

Author: piotrrzysko

core/trino-main/src/main/java/io/trino/execution/CreateMaterializedViewTask.java

@@ -159,11 +159,11 @@ Analysis executeInternal(
                 });
 
         Optional<WhenStaleBehavior> whenStale = statement.getWhenStaleBehavior()
-                .map(_ -> {
+                .map(whenStaleBehavior -> {
                     if (!plannerContext.getMetadata().getConnectorCapabilities(session, catalogHandle).contains(MATERIALIZED_VIEW_WHEN_STALE_BEHAVIOR)) {
                         throw semanticException(NOT_SUPPORTED, statement, "Catalog '%s' does not support WHEN STALE", catalogName);
                     }
-                    throw semanticException(NOT_SUPPORTED, statement, "WHEN STALE is not supported yet");
+                    return toConnectorWhenStaleBehavior(whenStaleBehavior);
                 });
 
         MaterializedViewDefinition definition = new MaterializedViewDefinition(
@@ -193,4 +193,12 @@ Analysis executeInternal(
         plannerContext.getMetadata().createMaterializedView(session, name, definition, properties, statement.isReplace(), statement.isNotExists());
         return analysis;
     }
+
+    private static WhenStaleBehavior toConnectorWhenStaleBehavior(CreateMaterializedView.WhenStaleBehavior whenStale)
+    {
+        return switch (whenStale) {
+            case INLINE -> WhenStaleBehavior.INLINE;
+            case FAIL -> WhenStaleBehavior.FAIL;
+        };
+    }
 }

core/trino-main/src/main/java/io/trino/sql/analyzer/StatementAnalyzer.java

@@ -66,6 +66,7 @@
 import io.trino.spi.connector.ColumnHandle;
 import io.trino.spi.connector.ColumnMetadata;
 import io.trino.spi.connector.ColumnSchema;
+import io.trino.spi.connector.ConnectorMaterializedViewDefinition.WhenStaleBehavior;
 import io.trino.spi.connector.ConnectorTableMetadata;
 import io.trino.spi.connector.ConnectorTransactionHandle;
 import io.trino.spi.connector.MaterializedViewFreshness;
@@ -2285,6 +2286,7 @@ protected Scope visitTable(Table table, Optional<Scope> scope)
             if (optionalMaterializedView.isPresent()) {
                 MaterializedViewDefinition materializedViewDefinition = optionalMaterializedView.get();
                 analysis.addEmptyColumnReferencesForTable(accessControl, session.getIdentity(), name);
+                boolean useLogicalViewSemantics = shouldUseLogicalViewSemantics(materializedViewDefinition);
                 if (isMaterializedViewSufficientlyFresh(session, name, materializedViewDefinition)) {
                     // If materialized view is sufficiently fresh with respect to its grace period, answer the query using the storage table
                     QualifiedName storageName = getMaterializedViewStorageTableName(materializedViewDefinition)
@@ -2293,10 +2295,13 @@ protected Scope visitTable(Table table, Optional<Scope> scope)
                     checkStorageTableNotRedirected(storageTableName);
                     TableHandle tableHandle = metadata.getTableHandle(session, storageTableName)
                             .orElseThrow(() -> semanticException(INVALID_VIEW, table, "Storage table '%s' does not exist", storageTableName));
-                    return createScopeForMaterializedView(table, name, scope, materializedViewDefinition, Optional.of(tableHandle));
+                    return createScopeForMaterializedView(table, name, scope, materializedViewDefinition, Optional.of(tableHandle), useLogicalViewSemantics);
+                }
+                else if (!useLogicalViewSemantics) {
+                    throw semanticException(VIEW_IS_STALE, table, "Materialized view '%s' is stale", name);
                 }
                 // This is a stale materialized view and should be expanded like a logical view
-                return createScopeForMaterializedView(table, name, scope, materializedViewDefinition, Optional.empty());
+                return createScopeForMaterializedView(table, name, scope, materializedViewDefinition, Optional.empty(), useLogicalViewSemantics);
             }
 
             // This could be a reference to a logical view or a table
@@ -2385,6 +2390,15 @@ private boolean isMaterializedViewSufficientlyFresh(Session session, QualifiedOb
             return staleness.compareTo(gracePeriod) <= 0;
         }
 
+        private static boolean shouldUseLogicalViewSemantics(MaterializedViewDefinition materializedViewDefinition)
+        {
+            WhenStaleBehavior whenStale = materializedViewDefinition.getWhenStaleBehavior().orElse(WhenStaleBehavior.INLINE);
+            return switch (whenStale) {
+                case WhenStaleBehavior.INLINE -> true;
+                case WhenStaleBehavior.FAIL -> false;
+            };
+        }
+
         private void checkStorageTableNotRedirected(QualifiedObjectName source)
         {
             metadata.getRedirectionAwareTableHandle(session, source).redirectedTableName().ifPresent(name -> {
@@ -2529,7 +2543,13 @@ private Scope createScopeForCommonTableExpression(Table table, Optional<Scope> s
             return createAndAssignScope(table, scope, fields);
         }
 
-        private Scope createScopeForMaterializedView(Table table, QualifiedObjectName name, Optional<Scope> scope, MaterializedViewDefinition view, Optional<TableHandle> storageTable)
+        private Scope createScopeForMaterializedView(
+                Table table,
+                QualifiedObjectName name,
+                Optional<Scope> scope,
+                MaterializedViewDefinition view,
+                Optional<TableHandle> storageTable,
+                boolean useLogicalViewSemantics)
         {
             return createScopeForView(
                     table,
@@ -2542,7 +2562,8 @@ private Scope createScopeForMaterializedView(Table table, QualifiedObjectName na
                     view.getPath(),
                     view.getColumns(),
                     storageTable,
-                    true);
+                    true,
+                    useLogicalViewSemantics);
         }
 
         private Scope createScopeForView(Table table, QualifiedObjectName name, Optional<Scope> scope, ViewDefinition view)
@@ -2557,7 +2578,8 @@ private Scope createScopeForView(Table table, QualifiedObjectName name, Optional
                     view.getPath(),
                     view.getColumns(),
                     Optional.empty(),
-                    false);
+                    false,
+                    true);
         }
 
         private Scope createScopeForView(
@@ -2571,7 +2593,8 @@ private Scope createScopeForView(
                 List<CatalogSchemaName> path,
                 List<ViewColumn> columns,
                 Optional<TableHandle> storageTable,
-                boolean isMaterializedView)
+                boolean isMaterializedView,
+                boolean useLogicalViewSemantics)
         {
             Statement statement = analysis.getStatement();
             if (statement instanceof CreateView viewStatement) {
@@ -2590,18 +2613,27 @@ private Scope createScopeForView(
                 throw semanticException(VIEW_IS_RECURSIVE, table, "View is recursive");
             }
 
-            Query query = parseView(originalSql, name, table);
+            if (useLogicalViewSemantics) {
+                Query query = parseView(originalSql, name, table);
 
-            if (!query.getFunctions().isEmpty()) {
-                throw semanticException(NOT_SUPPORTED, table, "View contains inline function: %s", name);
-            }
+                if (!query.getFunctions().isEmpty()) {
+                    throw semanticException(NOT_SUPPORTED, table, "View contains inline function: %s", name);
+                }
 
-            analysis.registerTableForView(table, name, isMaterializedView);
-            RelationType descriptor = analyzeView(query, name, catalog, schema, owner, path, table);
-            analysis.unregisterTableForView();
+                analysis.registerTableForView(table, name, isMaterializedView);
+                RelationType descriptor = analyzeView(query, name, catalog, schema, owner, path, table);
+                analysis.unregisterTableForView();
 
-            checkViewStaleness(columns, descriptor.getVisibleFields(), name, table)
-                    .ifPresent(explanation -> { throw semanticException(VIEW_IS_STALE, table, "View '%s' is stale or in invalid state: %s", name, explanation); });
+                checkViewStaleness(columns, descriptor.getVisibleFields(), name, table)
+                        .ifPresent(explanation -> { throw semanticException(VIEW_IS_STALE, table, "View '%s' is stale or in invalid state: %s", name, explanation); });
+
+                if (storageTable.isEmpty()) {
+                    analysis.registerNamedQuery(table, query);
+                }
+            }
+            else {
+                checkArgument(storageTable.isPresent(), "A storage table must be present when query analysis is skipped");
+            }
 
             // Derive the type of the view from the stored definition, not from the analysis of the underlying query.
             // This is needed in case the underlying table(s) changed and the query in the view now produces types that
@@ -2621,9 +2653,6 @@ private Scope createScopeForView(
                 List<Field> storageTableFields = analyzeStorageTable(table, viewFields, storageTable.get());
                 analysis.setMaterializedViewStorageTableFields(table, storageTableFields);
             }
-            else {
-                analysis.registerNamedQuery(table, query);
-            }
 
             Scope accessControlScope = Scope.builder()
                     .withRelationType(RelationId.anonymous(), new RelationType(viewFields))

core/trino-main/src/main/java/io/trino/sql/rewrite/ShowQueriesRewrite.java

@@ -40,6 +40,7 @@
 import io.trino.metadata.ViewPropertyManager;
 import io.trino.security.AccessControl;
 import io.trino.spi.connector.CatalogSchemaName;
+import io.trino.spi.connector.ConnectorMaterializedViewDefinition;
 import io.trino.spi.connector.ConnectorTableMetadata;
 import io.trino.spi.connector.SchemaTableName;
 import io.trino.spi.function.FunctionKind;
@@ -61,6 +62,7 @@
 import io.trino.sql.tree.Cast;
 import io.trino.sql.tree.ColumnDefinition;
 import io.trino.sql.tree.CreateMaterializedView;
+import io.trino.sql.tree.CreateMaterializedView.WhenStaleBehavior;
 import io.trino.sql.tree.CreateSchema;
 import io.trino.sql.tree.CreateTable;
 import io.trino.sql.tree.CreateView;
@@ -562,12 +564,21 @@ private Query showCreateMaterializedView(ShowCreate node)
                     false,
                     false,
                     Optional.empty(), // TODO support GRACE PERIOD
-                    Optional.empty(), // TODO support WHEN STALE
+                    viewDefinition.flatMap(MaterializedViewDefinition::getWhenStaleBehavior)
+                            .map(Visitor::toSqlWhenStaleBehavior),
                     propertyNodes,
                     viewDefinition.get().getComment())).trim();
             return singleValueQuery("Create Materialized View", sql);
         }
 
+        private static WhenStaleBehavior toSqlWhenStaleBehavior(ConnectorMaterializedViewDefinition.WhenStaleBehavior whenStale)
+        {
+            return switch (whenStale) {
+                case INLINE -> WhenStaleBehavior.INLINE;
+                case FAIL -> WhenStaleBehavior.FAIL;
+            };
+        }
+
         private Query showCreateView(ShowCreate node)
         {
             QualifiedObjectName objectName = createQualifiedObjectName(session, node, node.getName());