Support REFRESH MATERIALIZED VIEW when storage differs in schema

The materialized view's storage table may have a schema different than
the materialized view itself. This is already supported when reading
(the coercions are getting applied as necessary), but was not supported
for REFRESH. In the REFRESH normal INSERT constraints where getting
applied, thus not allowing the materialized view to, for example, store
temporal values in a varchar column.

The change removes the constraint, and therefore a storage table may use
any type that is coercible to from the view definition type. Of course,
the implementations should only use types where coercion to the storage
and backwards does round trip.

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

@@ -689,19 +689,6 @@ protected Scope visitRefreshMaterializedView(RefreshMaterializedView refreshMate
                     .map(insertColumn -> tableMetadata.getColumn(insertColumn).getType())
                     .collect(toImmutableList());
 
-            List<Type> queryTypes = queryScope.getRelationType().getVisibleFields().stream()
-                    .map(Field::getType)
-                    .collect(toImmutableList());
-
-            if (!typesMatchForInsert(tableTypes, queryTypes)) {
-                throw semanticException(
-                        TYPE_MISMATCH,
-                        refreshMaterializedView,
-                        "Insert query has mismatched column types: Table: [%s], Query: [%s]",
-                        Joiner.on(", ").join(tableTypes),
-                        Joiner.on(", ").join(queryTypes));
-            }
-
             Stream<Column> columns = Streams.zip(
                     insertColumns.stream(),
                     tableTypes.stream()

core/trino-main/src/main/java/io/trino/testing/TestingMetadata.java

@@ -279,6 +279,18 @@ public void markMaterializedViewIsFresh(SchemaTableName name)
         freshMaterializedViews.add(name);
     }
 
+    @Override
+    public boolean delegateMaterializedViewRefreshToConnector(ConnectorSession session, SchemaTableName viewName)
+    {
+        return false;
+    }
+
+    @Override
+    public ConnectorInsertTableHandle beginRefreshMaterializedView(ConnectorSession session, ConnectorTableHandle tableHandle, List<ConnectorTableHandle> sourceTableHandles, RetryMode retryMode)
+    {
+        return TestingHandle.INSTANCE;
+    }
+
     @Override
     public ConnectorOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Optional<ConnectorTableLayout> layout, RetryMode retryMode)
     {
@@ -378,6 +390,12 @@ public SchemaTableName getTableName()
         {
             return tableName;
         }
+
+        @Override
+        public String toString()
+        {
+            return tableName.toString();
+        }
     }
 
     public static class TestingColumnHandle

core/trino-main/src/test/java/io/trino/sql/planner/TestMaterializedViews.java

@@ -33,20 +33,28 @@
 import io.trino.spi.security.ViewExpression;
 import io.trino.spi.transaction.IsolationLevel;
 import io.trino.sql.planner.assertions.BasePlanTest;
+import io.trino.sql.tree.GenericLiteral;
 import io.trino.testing.LocalQueryRunner;
 import io.trino.testing.TestingAccessControlManager;
 import io.trino.testing.TestingMetadata;
 import org.testng.annotations.Test;
 
+import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 
 import static io.trino.spi.type.BigintType.BIGINT;
 import static io.trino.spi.type.TinyintType.TINYINT;
 import static io.trino.spi.type.VarcharType.VARCHAR;
 import static io.trino.sql.planner.assertions.PlanMatchPattern.anyTree;
+import static io.trino.sql.planner.assertions.PlanMatchPattern.exchange;
 import static io.trino.sql.planner.assertions.PlanMatchPattern.expression;
+import static io.trino.sql.planner.assertions.PlanMatchPattern.output;
 import static io.trino.sql.planner.assertions.PlanMatchPattern.project;
 import static io.trino.sql.planner.assertions.PlanMatchPattern.tableScan;
+import static io.trino.sql.planner.assertions.PlanMatchPattern.tableWriter;
+import static io.trino.sql.planner.assertions.PlanMatchPattern.values;
+import static io.trino.sql.planner.plan.ExchangeNode.Scope.LOCAL;
 import static io.trino.testing.TestingHandles.TEST_CATALOG_NAME;
 import static io.trino.testing.TestingSession.testSessionBuilder;
 
@@ -164,6 +172,16 @@ protected LocalQueryRunner createLocalQueryRunner()
         });
         testingConnectorMetadata.markMaterializedViewIsFresh(materializedViewWithCasts.asSchemaTableName());
 
+        queryRunner.inTransaction(session -> {
+            metadata.createMaterializedView(
+                    session,
+                    new QualifiedObjectName(TEST_CATALOG_NAME, SCHEMA, "stale_materialized_view_with_casts"),
+                    materializedViewDefinitionWithCasts,
+                    false,
+                    false);
+            return null;
+        });
+
         return queryRunner;
     }
 
@@ -201,6 +219,22 @@ public void testMaterializedViewWithCasts()
                                 tableScan("storage_table_with_casts", ImmutableMap.of("A", "a", "B", "b")))));
     }
 
+    @Test
+    public void testRefreshMaterializedViewWithCasts()
+    {
+        assertPlan("REFRESH MATERIALIZED VIEW stale_materialized_view_with_casts",
+                anyTree(
+                        tableWriter(List.of("A_CAST", "B_CAST"), List.of("a", "b"),
+                                exchange(LOCAL,
+                                        project(Map.of("A_CAST", expression("CAST(A AS tinyint)"), "B_CAST", expression("CAST(B AS varchar)")),
+                                                tableScan("test_table", Map.of("A", "a", "B", "b")))))));
+
+        // No-op REFRESH
+        assertPlan("REFRESH MATERIALIZED VIEW materialized_view_with_casts",
+                output(
+                        values(List.of("rows"), List.of(List.of(new GenericLiteral("BIGINT", "0"))))));
+    }
+
     private static class TestMaterializedViewConnector
             implements Connector
     {

testing/trino-testing/src/main/java/io/trino/testing/BaseConnectorTest.java

@@ -1221,6 +1221,9 @@ public void testMaterializedViewAllTypes()
                 "  'a dummy' a_dummy";
 
         assertUpdate("CREATE MATERIALIZED VIEW %s AS %s".formatted(viewName, values));
+        assertThat(query("TABLE " + viewName))
+                .matches(values);
+        assertUpdate("REFRESH MATERIALIZED VIEW " + viewName, 1);
         assertThat(query("TABLE " + viewName))
                 .matches(values);