Respect GRACE PERIOD when querying stale mateiralized views

Author: findepi

core/trino-main/src/main/java/io/trino/FeaturesConfig.java

@@ -102,6 +102,7 @@ public class FeaturesConfig
     private boolean incrementalHashArrayLoadFactorEnabled = true;
     private boolean allowSetViewAuthorization;
 
+    private boolean legacyMaterializedViewGracePeriod;
     private boolean hideInaccessibleColumns;
     private boolean forceSpillingJoin;
 
@@ -445,6 +446,21 @@ public FeaturesConfig setIncrementalHashArrayLoadFactorEnabled(boolean increment
         return this;
     }
 
+    @Deprecated
+    public boolean isLegacyMaterializedViewGracePeriod()
+    {
+        return legacyMaterializedViewGracePeriod;
+    }
+
+    @Deprecated
+    @Config("legacy.materialized-view-grace-period")
+    @ConfigDescription("Enable legacy handling of stale materialized views")
+    public FeaturesConfig setLegacyMaterializedViewGracePeriod(boolean legacyMaterializedViewGracePeriod)
+    {
+        this.legacyMaterializedViewGracePeriod = legacyMaterializedViewGracePeriod;
+        return this;
+    }
+
     public boolean isHideInaccessibleColumns()
     {
         return hideInaccessibleColumns;

core/trino-main/src/main/java/io/trino/SystemSessionProperties.java

@@ -160,6 +160,7 @@ public final class SystemSessionProperties
     public static final String RETRY_INITIAL_DELAY = "retry_initial_delay";
     public static final String RETRY_MAX_DELAY = "retry_max_delay";
     public static final String RETRY_DELAY_SCALE_FACTOR = "retry_delay_scale_factor";
+    public static final String LEGACY_MATERIALIZED_VIEW_GRACE_PERIOD = "legacy_materialized_view_grace_period";
     public static final String HIDE_INACCESSIBLE_COLUMNS = "hide_inaccessible_columns";
     public static final String FAULT_TOLERANT_EXECUTION_TARGET_TASK_INPUT_SIZE = "fault_tolerant_execution_target_task_input_size";
     public static final String FAULT_TOLERANT_EXECUTION_TARGET_TASK_SPLIT_COUNT = "fault_tolerant_execution_target_task_split_count";
@@ -796,6 +797,11 @@ public SystemSessionProperties(
                             }
                         },
                         false),
+                booleanProperty(
+                        LEGACY_MATERIALIZED_VIEW_GRACE_PERIOD,
+                        "Enable legacy handling of stale materialized views",
+                        featuresConfig.isLegacyMaterializedViewGracePeriod(),
+                        false),
                 booleanProperty(
                         HIDE_INACCESSIBLE_COLUMNS,
                         "When enabled non-accessible columns are silently filtered from results from SELECT * statements",
@@ -1519,6 +1525,12 @@ public static double getRetryDelayScaleFactor(Session session)
         return session.getSystemProperty(RETRY_DELAY_SCALE_FACTOR, Double.class);
     }
 
+    @Deprecated
+    public static boolean isLegacyMaterializedViewGracePeriod(Session session)
+    {
+        return session.getSystemProperty(LEGACY_MATERIALIZED_VIEW_GRACE_PERIOD, Boolean.class);
+    }
+
     public static boolean isHideInaccessibleColumns(Session session)
     {
         return session.getSystemProperty(HIDE_INACCESSIBLE_COLUMNS, Boolean.class);

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

@@ -66,7 +66,7 @@
 import io.trino.spi.connector.ColumnSchema;
 import io.trino.spi.connector.ConnectorTableMetadata;
 import io.trino.spi.connector.ConnectorTransactionHandle;
-import io.trino.spi.connector.MaterializedViewFreshness.Freshness;
+import io.trino.spi.connector.MaterializedViewFreshness;
 import io.trino.spi.connector.PointerType;
 import io.trino.spi.connector.SchemaTableName;
 import io.trino.spi.connector.TableProcedureMetadata;
@@ -251,6 +251,7 @@
 import io.trino.type.TypeCoercion;
 
 import java.math.RoundingMode;
+import java.time.Duration;
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
@@ -278,6 +279,7 @@
 import static com.google.common.collect.Iterables.getLast;
 import static com.google.common.collect.Iterables.getOnlyElement;
 import static io.trino.SystemSessionProperties.getMaxGroupingSets;
+import static io.trino.SystemSessionProperties.isLegacyMaterializedViewGracePeriod;
 import static io.trino.metadata.FunctionResolver.toPath;
 import static io.trino.metadata.MetadataManager.toQualifiedFunctionName;
 import static io.trino.metadata.MetadataUtil.createQualifiedObjectName;
@@ -2141,19 +2143,19 @@ protected Scope visitTable(Table table, Optional<Scope> scope)
 
             Optional<MaterializedViewDefinition> optionalMaterializedView = metadata.getMaterializedView(session, name);
             if (optionalMaterializedView.isPresent()) {
-                Freshness freshness = metadata.getMaterializedViewFreshness(session, name).getFreshness();
-                if (freshness == FRESH || freshness == Freshness.UNKNOWN) {
-                    // If materialized view is current, answer the query using the storage table
-                    QualifiedName storageName = getMaterializedViewStorageTableName(optionalMaterializedView.get())
+                MaterializedViewDefinition materializedViewDefinition = optionalMaterializedView.get();
+                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)
                             .orElseThrow(() -> semanticException(INVALID_VIEW, table, "Materialized view '%s' is fresh but does not have storage table name", name));
                     QualifiedObjectName storageTableName = createQualifiedObjectName(session, table, storageName);
                     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, optionalMaterializedView.get(), Optional.of(tableHandle));
+                    return createScopeForMaterializedView(table, name, scope, materializedViewDefinition, Optional.of(tableHandle));
                 }
                 // This is a stale materialized view and should be expanded like a logical view
-                return createScopeForMaterializedView(table, name, scope, optionalMaterializedView.get(), Optional.empty());
+                return createScopeForMaterializedView(table, name, scope, materializedViewDefinition, Optional.empty());
             }
 
             // This could be a reference to a logical view or a table
@@ -2213,6 +2215,43 @@ protected Scope visitTable(Table table, Optional<Scope> scope)
             return tableScope;
         }
 
+        private boolean isMaterializedViewSufficientlyFresh(Session session, QualifiedObjectName name, MaterializedViewDefinition materializedViewDefinition)
+        {
+            MaterializedViewFreshness materializedViewFreshness = metadata.getMaterializedViewFreshness(session, name);
+            MaterializedViewFreshness.Freshness freshness = materializedViewFreshness.getFreshness();
+
+            if (isLegacyMaterializedViewGracePeriod(session)) {
+                return switch (freshness) {
+                    case FRESH, UNKNOWN -> true;
+                    case STALE -> false;
+                };
+            }
+
+            if (freshness == FRESH) {
+                return true;
+            }
+            Optional<Instant> lastFreshTime = materializedViewFreshness.getLastFreshTime();
+            if (lastFreshTime.isEmpty()) {
+                // E.g. never refreshed, or connector not updated to report fresh time
+                return false;
+            }
+            if (materializedViewDefinition.getGracePeriod().isEmpty()) {
+                // Unlimited grace period
+                return true;
+            }
+            Duration gracePeriod = materializedViewDefinition.getGracePeriod().get();
+            if (gracePeriod.isZero()) {
+                // Consider 0 as a special value meaning "do not accept any staleness". This makes 0 more reliable, and more likely what user wanted,
+                // regardless of lastFreshTime, query time or rounding.
+                return false;
+            }
+
+            // Can be negative
+            // TODO should we compare lastFreshTime with session.start() or with current time? The freshness is calculated with respect to current state of things.
+            Duration staleness = Duration.between(lastFreshTime.get(), sessionTimeProvider.getStart(session));
+            return staleness.compareTo(gracePeriod) <= 0;
+        }
+
         private void checkStorageTableNotRedirected(QualifiedObjectName source)
         {
             metadata.getRedirectionAwareTableHandle(session, source).getRedirectedTableName().ifPresent(name -> {

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

@@ -45,6 +45,8 @@
 import io.trino.spi.statistics.TableStatisticsMetadata;
 import io.trino.spi.type.Type;
 
+import java.time.Duration;
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashSet;
@@ -68,6 +70,8 @@
 public class TestingMetadata
         implements ConnectorMetadata
 {
+    public static final Duration STALE_MV_STALENESS = Duration.ofHours(7);
+
     private final ConcurrentMap<SchemaTableName, ConnectorTableMetadata> tables = new ConcurrentHashMap<>();
     private final ConcurrentMap<SchemaTableName, ConnectorViewDefinition> views = new ConcurrentHashMap<>();
     private final ConcurrentMap<SchemaTableName, ConnectorMaterializedViewDefinition> materializedViews = new ConcurrentHashMap<>();
@@ -271,7 +275,10 @@ public void dropMaterializedView(ConnectorSession session, SchemaTableName viewN
     @Override
     public MaterializedViewFreshness getMaterializedViewFreshness(ConnectorSession session, SchemaTableName name)
     {
-        return new MaterializedViewFreshness(freshMaterializedViews.contains(name) ? FRESH : STALE);
+        boolean fresh = freshMaterializedViews.contains(name);
+        return new MaterializedViewFreshness(
+                fresh ? FRESH : STALE,
+                fresh ? Optional.empty() : Optional.of(Instant.now().minus(STALE_MV_STALENESS)));
     }
 
     public void markMaterializedViewIsFresh(SchemaTableName name)

core/trino-main/src/test/java/io/trino/sql/analyzer/TestAnalyzer.java

@@ -100,6 +100,7 @@
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.Optional;
 import java.util.function.Consumer;
@@ -6749,7 +6750,7 @@ public void setup()
                 Optional.of(TPCH_CATALOG),
                 Optional.of("s1"),
                 ImmutableList.of(new ViewColumn("a", BIGINT.getTypeId(), Optional.empty())),
-                Optional.empty(),
+                Optional.of(Duration.ZERO),
                 Optional.of("comment"),
                 Identity.ofUser("user"),
                 Optional.empty(),
@@ -6878,7 +6879,7 @@ public void setup()
                         Optional.of(TPCH_CATALOG),
                         Optional.of("s1"),
                         ImmutableList.of(new ViewColumn("a", BIGINT.getTypeId(), Optional.empty())),
-                        Optional.empty(),
+                        Optional.of(Duration.ZERO),
                         Optional.empty(),
                         Identity.ofUser("some user"),
                         Optional.of(new CatalogSchemaTableName(TPCH_CATALOG, "s1", "t1")),

core/trino-main/src/test/java/io/trino/sql/analyzer/TestFeaturesConfig.java

@@ -62,6 +62,7 @@ public void testDefaults()
                 .setOmitDateTimeTypePrecision(false)
                 .setLegacyCatalogRoles(false)
                 .setIncrementalHashArrayLoadFactorEnabled(true)
+                .setLegacyMaterializedViewGracePeriod(false)
                 .setHideInaccessibleColumns(false)
                 .setAllowSetViewAuthorization(false)
                 .setForceSpillingJoin(false)
@@ -97,6 +98,7 @@ public void testExplicitPropertyMappings()
                 .put("deprecated.omit-datetime-type-precision", "true")
                 .put("deprecated.legacy-catalog-roles", "true")
                 .put("incremental-hash-array-load-factor.enabled", "false")
+                .put("legacy.materialized-view-grace-period", "true")
                 .put("hide-inaccessible-columns", "true")
                 .put("legacy.allow-set-view-authorization", "true")
                 .put("force-spilling-join-operator", "true")
@@ -129,6 +131,7 @@ public void testExplicitPropertyMappings()
                 .setOmitDateTimeTypePrecision(true)
                 .setLegacyCatalogRoles(true)
                 .setIncrementalHashArrayLoadFactorEnabled(false)
+                .setLegacyMaterializedViewGracePeriod(true)
                 .setHideInaccessibleColumns(true)
                 .setAllowSetViewAuthorization(true)
                 .setForceSpillingJoin(true)

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

@@ -16,6 +16,7 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import io.trino.Session;
+import io.trino.SystemSessionProperties;
 import io.trino.connector.StaticConnectorFactory;
 import io.trino.metadata.MaterializedViewDefinition;
 import io.trino.metadata.Metadata;
@@ -39,6 +40,8 @@
 import io.trino.testing.TestingMetadata;
 import org.testng.annotations.Test;
 
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -56,6 +59,7 @@
 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.TestingMetadata.STALE_MV_STALENESS;
 import static io.trino.testing.TestingSession.testSessionBuilder;
 
 public class TestMaterializedViews
@@ -125,7 +129,7 @@ protected LocalQueryRunner createLocalQueryRunner()
                 Optional.of(TEST_CATALOG_NAME),
                 Optional.of(SCHEMA),
                 ImmutableList.of(new ViewColumn("a", BIGINT.getTypeId(), Optional.empty()), new ViewColumn("b", BIGINT.getTypeId(), Optional.empty())),
-                Optional.empty(),
+                Optional.of(STALE_MV_STALENESS.plusHours(1)),
                 Optional.empty(),
                 Identity.ofUser("some user"),
                 Optional.of(new CatalogSchemaTableName(TEST_CATALOG_NAME, SCHEMA, "storage_table")),
@@ -198,7 +202,28 @@ public void testFreshMaterializedView()
     @Test
     public void testNotFreshMaterializedView()
     {
-        assertPlan("SELECT * FROM not_fresh_materialized_view",
+        Session defaultSession = getQueryRunner().getDefaultSession();
+        Session legacyGracePeriod = Session.builder(defaultSession)
+                .setSystemProperty(SystemSessionProperties.LEGACY_MATERIALIZED_VIEW_GRACE_PERIOD, "true")
+                .build();
+        Session futureSession = Session.builder(defaultSession)
+                .setStart(Instant.now().plus(1, ChronoUnit.DAYS))
+                .build();
+
+        assertPlan(
+                "SELECT * FROM not_fresh_materialized_view",
+                defaultSession,
+                anyTree(
+                        tableScan("storage_table")));
+
+        assertPlan(
+                "SELECT * FROM not_fresh_materialized_view",
+                legacyGracePeriod,
+                anyTree(
+                        tableScan("test_table")));
+        assertPlan(
+                "SELECT * FROM not_fresh_materialized_view",
+                futureSession,
                 anyTree(
                         tableScan("test_table")));
     }

core/trino-spi/src/main/java/io/trino/spi/connector/ConnectorMaterializedViewDefinition.java

@@ -54,7 +54,7 @@ public ConnectorMaterializedViewDefinition(
                 catalog,
                 schema,
                 columns,
-                Optional.empty(),
+                Optional.of(Duration.ZERO),
                 comment,
                 owner,
                 properties);

docs/src/main/sphinx/sql/create-materialized-view.rst

@@ -9,6 +9,7 @@ Synopsis
 
     CREATE [ OR REPLACE ] MATERIALIZED VIEW
     [ IF NOT EXISTS ] view_name
+    [ GRACE PERIOD interval ]
     [ COMMENT string ]
     [ WITH properties ]
     AS query
@@ -29,16 +30,19 @@ materialized views, as compared to each time of accessing the view. Multiple
 reads of view data over time, or by multiple users, all trigger repeated
 processing. This is avoided for materialized views.
 
-When the underlying data changes, the materialized view becomes out of sync with
-the source tables. Update the data in the materialized view with the
-:doc:`refresh-materialized-view` statement.
-
 The optional ``OR REPLACE`` clause causes the materialized view to be replaced
 if it already exists rather than raising an error.
 
 The optional ``IF NOT EXISTS`` clause causes the materialized view only to be
 created or replaced if it does not exist yet.
 
+The optional ``GRACE PERIOD`` clause specifies how long the query materialization
+is used for querying. If the time elapsed since last materialized view refresh
+is greater than the grace period, the materialized view acts as a normal view and
+the materialized data is not used. If not specified, the grace period defaults to
+infinity. See :doc:`refresh-materialized-view` for more about refreshing
+materialized views.
+
 The optional ``COMMENT`` clause causes a ``string`` comment to be stored with
 the metadata about the materialized view. The comment is displayed with the
 :doc:`show-create-materialized-view` statement and is available in the table

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergConnector.java

@@ -40,6 +40,7 @@
 
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.common.collect.Sets.immutableEnumSet;
+import static io.trino.spi.connector.ConnectorCapabilities.MATERIALIZED_VIEW_GRACE_PERIOD;
 import static io.trino.spi.connector.ConnectorCapabilities.NOT_NULL_COLUMN_CONSTRAINT;
 import static io.trino.spi.transaction.IsolationLevel.SERIALIZABLE;
 import static io.trino.spi.transaction.IsolationLevel.checkConnectorSupports;
@@ -100,7 +101,9 @@ public IcebergConnector(
     @Override
     public Set<ConnectorCapabilities> getCapabilities()
     {
-        return immutableEnumSet(NOT_NULL_COLUMN_CONSTRAINT);
+        return immutableEnumSet(
+                NOT_NULL_COLUMN_CONSTRAINT,
+                MATERIALIZED_VIEW_GRACE_PERIOD);
     }
 
     @Override