feat(plugin-memory): Add support for materialized views in memory connector (#26405)

## Description
Per title. Also, REFRESH with WHERE clause does not make sense for
memory connector (which only has unpartitioned tables), so making it
optional with a new session property.

## Motivation and Context
Add tests for materialized views

## Impact
N/A

## Test Plan
Included tests.

## Contributor checklist

- [ ] Please make sure your submission complies with our [contributing
guide](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md),
in particular [code
style](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md#code-style)
and [commit
standards](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md#commit-standards).
- [ ] PR description addresses the issue accurately and concisely. If
the change is non-trivial, a GitHub Issue is referenced.
- [ ] Documented new properties (with its default value), SQL syntax,
functions, or other functionality.
- [ ] If release notes are required, they follow the [release notes
guidelines](https://github.com/prestodb/presto/wiki/Release-Notes-Guidelines).
- [ ] Adequate tests were added if applicable.
- [ ] CI passed.
- [ ] If adding new dependencies, verified they have an [OpenSSF
Scorecard](https://securityscorecards.dev/#the-checks) score of 5.0 or
higher (or obtained explicit TSC approval for lower scores).

## Release Notes
```
== NO RELEASE NOTE ==
```

Author: tdcmeehan

presto-main-base/src/main/java/com/facebook/presto/SystemSessionProperties.java

@@ -244,6 +244,7 @@ public final class SystemSessionProperties
     public static final String MATERIALIZED_VIEW_DATA_CONSISTENCY_ENABLED = "materialized_view_data_consistency_enabled";
     public static final String CONSIDER_QUERY_FILTERS_FOR_MATERIALIZED_VIEW_PARTITIONS = "consider-query-filters-for-materialized-view-partitions";
     public static final String QUERY_OPTIMIZATION_WITH_MATERIALIZED_VIEW_ENABLED = "query_optimization_with_materialized_view_enabled";
+    public static final String LEGACY_MATERIALIZED_VIEWS = "legacy_materialized_views";
     public static final String AGGREGATION_IF_TO_FILTER_REWRITE_STRATEGY = "aggregation_if_to_filter_rewrite_strategy";
     public static final String JOINS_NOT_NULL_INFERENCE_STRATEGY = "joins_not_null_inference_strategy";
     public static final String RESOURCE_AWARE_SCHEDULING_STRATEGY = "resource_aware_scheduling_strategy";
@@ -1353,6 +1354,12 @@ public SystemSessionProperties(
                         "Enable query optimization with materialized view",
                         featuresConfig.isQueryOptimizationWithMaterializedViewEnabled(),
                         true),
+                booleanProperty(
+                        LEGACY_MATERIALIZED_VIEWS,
+                        "Experimental: Use legacy materialized views.  This feature is under active development and may change" +
+                                "or be removed at any time.  Do not disable in production environments.",
+                        featuresConfig.isLegacyMaterializedViews(),
+                        true),
                 stringProperty(
                         DISTRIBUTED_TRACING_MODE,
                         "Mode for distributed tracing. NO_TRACE, ALWAYS_TRACE, or SAMPLE_BASED",
@@ -2882,6 +2889,11 @@ public static boolean isQueryOptimizationWithMaterializedViewEnabled(Session ses
         return session.getSystemProperty(QUERY_OPTIMIZATION_WITH_MATERIALIZED_VIEW_ENABLED, Boolean.class);
     }
 
+    public static boolean isLegacyMaterializedViews(Session session)
+    {
+        return session.getSystemProperty(LEGACY_MATERIALIZED_VIEWS, Boolean.class);
+    }
+
     public static boolean isVerboseRuntimeStatsEnabled(Session session)
     {
         return session.getSystemProperty(VERBOSE_RUNTIME_STATS_ENABLED, Boolean.class);

presto-main-base/src/main/java/com/facebook/presto/sql/MaterializedViewUtils.java

@@ -19,6 +19,7 @@
 import com.facebook.presto.common.predicate.TupleDomain;
 import com.facebook.presto.metadata.Metadata;
 import com.facebook.presto.metadata.SessionPropertyManager;
+import com.facebook.presto.spi.ConnectorId;
 import com.facebook.presto.spi.MaterializedViewStatus;
 import com.facebook.presto.spi.SchemaTableName;
 import com.facebook.presto.spi.relation.DomainTranslator;
@@ -90,7 +91,7 @@ public static Session buildOwnerSession(Session session, Optional<String> owner,
     {
         Identity identity = getOwnerIdentity(owner, session);
 
-        return Session.builder(sessionPropertyManager)
+        Session.SessionBuilder builder = Session.builder(sessionPropertyManager)
                 .setQueryId(session.getQueryId())
                 .setTransactionId(session.getTransactionId().orElse(null))
                 .setIdentity(identity)
@@ -102,8 +103,20 @@ public static Session buildOwnerSession(Session session, Optional<String> owner,
                 .setRemoteUserAddress(session.getRemoteUserAddress().orElse(null))
                 .setUserAgent(session.getUserAgent().orElse(null))
                 .setClientInfo(session.getClientInfo().orElse(null))
-                .setStartTime(session.getStartTime())
-                .build();
+                .setStartTime(session.getStartTime());
+
+        for (Map.Entry<String, String> property : session.getSystemProperties().entrySet()) {
+            builder.setSystemProperty(property.getKey(), property.getValue());
+        }
+
+        for (Map.Entry<ConnectorId, Map<String, String>> connectorEntry : session.getConnectorProperties().entrySet()) {
+            String catalogName = connectorEntry.getKey().getCatalogName();
+            for (Map.Entry<String, String> property : connectorEntry.getValue().entrySet()) {
+                builder.setCatalogSessionProperty(catalogName, property.getKey(), property.getValue());
+            }
+        }
+
+        return builder.build();
     }
 
     public static Identity getOwnerIdentity(Optional<String> owner, Session session)

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

@@ -225,6 +225,7 @@ public class FeaturesConfig
     private boolean materializedViewDataConsistencyEnabled = true;
     private boolean materializedViewPartitionFilteringEnabled = true;
     private boolean queryOptimizationWithMaterializedViewEnabled;
+    private boolean legacyMaterializedViewRefresh = true;
 
     private AggregationIfToFilterRewriteStrategy aggregationIfToFilterRewriteStrategy = AggregationIfToFilterRewriteStrategy.DISABLED;
     private String analyzerType = "BUILTIN";
@@ -2154,6 +2155,20 @@ public FeaturesConfig setQueryOptimizationWithMaterializedViewEnabled(boolean va
         return this;
     }
 
+    public boolean isLegacyMaterializedViews()
+    {
+        return legacyMaterializedViewRefresh;
+    }
+
+    @Config("experimental.legacy-materialized-views")
+    @ConfigDescription("Experimental: Use legacy materialized views.  This feature is under active development and may change" +
+            "or be removed at any time.  Do not disable in production environments.")
+    public FeaturesConfig setLegacyMaterializedViews(boolean value)
+    {
+        this.legacyMaterializedViewRefresh = value;
+        return this;
+    }
+
     public boolean isVerboseRuntimeStatsEnabled()
     {
         return verboseRuntimeStatsEnabled;

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

@@ -228,6 +228,7 @@
 
 import static com.facebook.presto.SystemSessionProperties.getMaxGroupingSets;
 import static com.facebook.presto.SystemSessionProperties.isAllowWindowOrderByLiterals;
+import static com.facebook.presto.SystemSessionProperties.isLegacyMaterializedViews;
 import static com.facebook.presto.SystemSessionProperties.isMaterializedViewDataConsistencyEnabled;
 import static com.facebook.presto.SystemSessionProperties.isMaterializedViewPartitionFilteringEnabled;
 import static com.facebook.presto.common.RuntimeMetricName.SKIP_READING_FROM_MATERIALIZED_VIEW_COUNT;
@@ -864,10 +865,9 @@ protected Scope visitRefreshMaterializedView(RefreshMaterializedView node, Optio
             // Use AllowAllAccessControl; otherwise Analyzer will check SELECT permission on the materialized view, which is not necessary.
             StatementAnalyzer viewAnalyzer = new StatementAnalyzer(analysis, metadata, sqlParser, new AllowAllAccessControl(), session, warningCollector);
             Scope viewScope = viewAnalyzer.analyze(node.getTarget(), scope);
-            if (!node.getWhere().isPresent()) {
-                throw new SemanticException(NOT_SUPPORTED, node, "Refresh Materialized View without predicates is not supported.");
-            }
-            Map<SchemaTableName, Expression> tablePredicates = extractTablePredicates(viewName, node.getWhere().get(), viewScope, metadata, session);
+
+            Map<SchemaTableName, Expression> tablePredicates = getTablePredicatesForMaterializedViewRefresh(
+                    session, node, viewName, viewScope, metadata);
 
             Query viewQuery = parseView(view.getOriginalSql(), viewName, node);
             Query refreshQuery = tablePredicates.containsKey(toSchemaTableName(viewName)) ?
@@ -910,10 +910,9 @@ private Optional<RelationType> analyzeBaseTableForRefreshMaterializedView(Table
             // Use AllowAllAccessControl; otherwise Analyzer will check SELECT permission on the materialized view, which is not necessary.
             StatementAnalyzer viewAnalyzer = new StatementAnalyzer(analysis, metadata, sqlParser, new AllowAllAccessControl(), session, warningCollector);
             Scope viewScope = viewAnalyzer.analyze(refreshMaterializedView.getTarget(), scope);
-            if (!refreshMaterializedView.getWhere().isPresent()) {
-                throw new SemanticException(NOT_SUPPORTED, "Refresh Materialized View without predicates is not supported.");
-            }
-            Map<SchemaTableName, Expression> tablePredicates = extractTablePredicates(viewName, refreshMaterializedView.getWhere().get(), viewScope, metadata, session);
+
+            Map<SchemaTableName, Expression> tablePredicates = getTablePredicatesForMaterializedViewRefresh(
+                    session, refreshMaterializedView, viewName, viewScope, metadata);
 
             SchemaTableName baseTableName = toSchemaTableName(createQualifiedObjectName(session, baseTable, baseTable.getName(), metadata));
             if (tablePredicates.containsKey(baseTableName)) {
@@ -928,6 +927,28 @@ private Optional<RelationType> analyzeBaseTableForRefreshMaterializedView(Table
             return Optional.empty();
         }
 
+        private Map<SchemaTableName, Expression> getTablePredicatesForMaterializedViewRefresh(
+                Session session,
+                RefreshMaterializedView node,
+                QualifiedObjectName viewName,
+                Scope viewScope,
+                Metadata metadata)
+        {
+            if (isLegacyMaterializedViews(session)) {
+                if (!node.getWhere().isPresent()) {
+                    throw new SemanticException(NOT_SUPPORTED, node, "Refresh Materialized View without predicates is not supported.");
+                }
+                return extractTablePredicates(viewName, node.getWhere().get(), viewScope, metadata, session);
+            }
+            else {
+                if (node.getWhere().isPresent()) {
+                    throw new SemanticException(NOT_SUPPORTED, node, "WHERE clause in REFRESH MATERIALIZED VIEW is not supported. " +
+                            "Connectors automatically determine which data needs refreshing based on staleness detection.");
+                }
+                return ImmutableMap.of();
+            }
+        }
+
         private Query buildQueryWithPredicate(Table table, Expression predicate)
         {
             Query query = simpleQuery(selectList(new AllColumns()), table, predicate);

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

@@ -187,6 +187,7 @@ public void testDefaults()
                 .setMaterializedViewDataConsistencyEnabled(true)
                 .setMaterializedViewPartitionFilteringEnabled(true)
                 .setQueryOptimizationWithMaterializedViewEnabled(false)
+                .setLegacyMaterializedViews(true)
                 .setVerboseRuntimeStatsEnabled(false)
                 .setAggregationIfToFilterRewriteStrategy(AggregationIfToFilterRewriteStrategy.DISABLED)
                 .setAnalyzerType("BUILTIN")
@@ -406,6 +407,7 @@ public void testExplicitPropertyMappings()
                 .put("materialized-view-data-consistency-enabled", "false")
                 .put("consider-query-filters-for-materialized-view-partitions", "false")
                 .put("query-optimization-with-materialized-view-enabled", "true")
+                .put("experimental.legacy-materialized-views", "false")
                 .put("analyzer-type", "CRUX")
                 .put("pre-process-metadata-calls", "true")
                 .put("verbose-runtime-stats-enabled", "true")
@@ -622,6 +624,7 @@ public void testExplicitPropertyMappings()
                 .setMaterializedViewDataConsistencyEnabled(false)
                 .setMaterializedViewPartitionFilteringEnabled(false)
                 .setQueryOptimizationWithMaterializedViewEnabled(true)
+                .setLegacyMaterializedViews(false)
                 .setVerboseRuntimeStatsEnabled(true)
                 .setAggregationIfToFilterRewriteStrategy(AggregationIfToFilterRewriteStrategy.FILTER_WITH_IF)
                 .setAnalyzerType("CRUX")

presto-memory/src/main/java/com/facebook/presto/plugin/memory/MemoryInsertTableHandle.java

@@ -27,14 +27,22 @@ public class MemoryInsertTableHandle
 {
     private final MemoryTableHandle table;
     private final Set<Long> activeTableIds;
+    private final boolean insertOverwrite;
 
     @JsonCreator
     public MemoryInsertTableHandle(
             @JsonProperty("table") MemoryTableHandle table,
-            @JsonProperty("activeTableIds") Set<Long> activeTableIds)
+            @JsonProperty("activeTableIds") Set<Long> activeTableIds,
+            @JsonProperty("insertOverwrite") boolean insertOverwrite)
     {
         this.table = requireNonNull(table, "table is null");
         this.activeTableIds = requireNonNull(activeTableIds, "activeTableIds is null");
+        this.insertOverwrite = insertOverwrite;
+    }
+
+    public MemoryInsertTableHandle(MemoryTableHandle table, Set<Long> activeTableIds)
+    {
+        this(table, activeTableIds, false);
     }
 
     @JsonProperty
@@ -49,6 +57,12 @@ public Set<Long> getActiveTableIds()
         return activeTableIds;
     }
 
+    @JsonProperty
+    public boolean isInsertOverwrite()
+    {
+        return insertOverwrite;
+    }
+
     @Override
     public String toString()
     {