Views, Spark: Add support for Materialized Views; Integrate with Spark SQL

api/src/main/java/org/apache/iceberg/view/ViewBuilder.java

@@ -19,6 +19,7 @@
 package org.apache.iceberg.view;
 
 import java.util.Map;
+import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.catalog.ViewCatalog;
 
 /**
@@ -55,6 +56,16 @@ default ViewBuilder withLocation(String location) {
     throw new UnsupportedOperationException("Setting a view's location is not supported");
   }
 
+  /**
+   * Sets the storage table identifier for a materialized view.
+   *
+   * @param storageTableIdentifier the storage table identifier
+   * @return this for method chaining
+   */
+  default ViewBuilder withStorageTableIdentifier(TableIdentifier storageTableIdentifier) {
+    throw new UnsupportedOperationException("Setting a storage table identifier is not supported");
+  }
+
   /**
    * Create the view.
    *

api/src/main/java/org/apache/iceberg/view/ViewVersion.java

@@ -21,6 +21,7 @@
 import java.util.List;
 import java.util.Map;
 import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
 
 /**
  * A version of the view at a point in time.
@@ -78,4 +79,15 @@ default String defaultCatalog() {
 
   /** The default namespace to use when the SQL does not contain a namespace. */
   Namespace defaultNamespace();
+
+  /**
+   * The storage table identifier for materialized views.
+   *
+   * <p>When null, the entity is a regular view. When set, the entity is a materialized view and
+   * this identifies the storage table that holds the precomputed data. The storage table must be in
+   * the same catalog as the materialized view.
+   */
+  default TableIdentifier storageTable() {
+    return null;
+  }
 }

core/src/main/java/org/apache/iceberg/inmemory/InMemoryCatalog.java

@@ -401,12 +401,12 @@ protected Map<String, String> properties() {
     return catalogProperties == null ? ImmutableMap.of() : catalogProperties;
   }
 
-  private class InMemoryTableOperations extends BaseMetastoreTableOperations {
+  protected class InMemoryTableOperations extends BaseMetastoreTableOperations {
     private final FileIO fileIO;
     private final TableIdentifier tableIdentifier;
     private final String fullTableName;
 
-    InMemoryTableOperations(FileIO fileIO, TableIdentifier tableIdentifier) {
+    public InMemoryTableOperations(FileIO fileIO, TableIdentifier tableIdentifier) {
       this.fileIO = fileIO;
       this.tableIdentifier = tableIdentifier;
       this.fullTableName = fullTableName(catalogName, tableIdentifier);
@@ -472,12 +472,12 @@ protected String tableName() {
     }
   }
 
-  private class InMemoryViewOperations extends BaseViewOperations {
+  protected class InMemoryViewOperations extends BaseViewOperations {
     private final FileIO io;
     private final TableIdentifier identifier;
     private final String fullViewName;
 
-    InMemoryViewOperations(FileIO io, TableIdentifier identifier) {
+    public InMemoryViewOperations(FileIO io, TableIdentifier identifier) {
       this.io = io;
       this.identifier = identifier;
       this.fullViewName = ViewUtil.fullViewName(catalogName, identifier);

core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java

@@ -689,6 +689,10 @@ public static LoadViewResponse createView(
             .withDefaultCatalog(request.viewVersion().defaultCatalog())
             .withLocation(request.location());
 
+    if (request.viewVersion().storageTable() != null) {
+      viewBuilder.withStorageTableIdentifier(request.viewVersion().storageTable());
+    }
+
     Set<String> unsupportedRepresentations =
         request.viewVersion().representations().stream()
             .filter(r -> !(r instanceof SQLViewRepresentation))

core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java

@@ -1718,6 +1718,7 @@ private class RESTViewBuilder implements ViewBuilder {
     private String defaultCatalog = null;
     private Schema schema = null;
     private String location = null;
+    private TableIdentifier storageTableIdentifier = null;
 
     private RESTViewBuilder(SessionContext context, TableIdentifier identifier) {
       checkViewIdentifierIsValid(identifier);
@@ -1797,6 +1798,12 @@ public ViewBuilder withLocation(String newLocation) {
       return this;
     }
 
+    @Override
+    public ViewBuilder withStorageTableIdentifier(TableIdentifier newStorageTableIdentifier) {
+      this.storageTableIdentifier = newStorageTableIdentifier;
+      return this;
+    }
+
     @Override
     public View create() {
       Endpoint.check(endpoints, Endpoint.V1_CREATE_VIEW);
@@ -1806,16 +1813,21 @@ public View create() {
       Preconditions.checkState(
           null != defaultNamespace, "Cannot create view without specifying a default namespace");
 
-      ViewVersion viewVersion =
+      ImmutableViewVersion.Builder versionBuilder =
           ImmutableViewVersion.builder()
               .versionId(1)
               .schemaId(schema.schemaId())
               .addAllRepresentations(representations)
               .defaultNamespace(defaultNamespace)
               .defaultCatalog(defaultCatalog)
               .timestampMillis(System.currentTimeMillis())
-              .putAllSummary(EnvironmentContext.get())
-              .build();
+              .putAllSummary(EnvironmentContext.get());
+
+      if (storageTableIdentifier != null) {
+        versionBuilder.storageTable(storageTableIdentifier);
+      }
+
+      ViewVersion viewVersion = versionBuilder.build();
 
       properties.putAll(viewOverrideProperties());
 
@@ -1906,16 +1918,21 @@ private View replace(LoadViewResponse response) {
               .max(Integer::compareTo)
               .orElseGet(metadata::currentVersionId);
 
-      ViewVersion viewVersion =
+      ImmutableViewVersion.Builder versionBuilder =
           ImmutableViewVersion.builder()
               .versionId(maxVersionId + 1)
               .schemaId(schema.schemaId())
               .addAllRepresentations(representations)
               .defaultNamespace(defaultNamespace)
               .defaultCatalog(defaultCatalog)
               .timestampMillis(System.currentTimeMillis())
-              .putAllSummary(EnvironmentContext.get())
-              .build();
+              .putAllSummary(EnvironmentContext.get());
+
+      if (storageTableIdentifier != null) {
+        versionBuilder.storageTable(storageTableIdentifier);
+      }
+
+      ViewVersion viewVersion = versionBuilder.build();
 
       properties.putAll(viewOverrideProperties());
 

core/src/main/java/org/apache/iceberg/view/BaseMetastoreViewCatalog.java

@@ -80,6 +80,7 @@ protected class BaseViewBuilder implements ViewBuilder {
     private String defaultCatalog = null;
     private Schema schema = null;
     private String location = null;
+    private TableIdentifier storageTableIdentifier = null;
 
     protected BaseViewBuilder(TableIdentifier identifier) {
       Preconditions.checkArgument(
@@ -159,6 +160,12 @@ public ViewBuilder withLocation(String newLocation) {
       return this;
     }
 
+    @Override
+    public ViewBuilder withStorageTableIdentifier(TableIdentifier newStorageTableIdentifier) {
+      this.storageTableIdentifier = newStorageTableIdentifier;
+      return this;
+    }
+
     @Override
     public View create() {
       return create(newViewOps(identifier));
@@ -190,16 +197,21 @@ private View create(ViewOperations ops) {
       Preconditions.checkState(
           null != defaultNamespace, "Cannot create view without specifying a default namespace");
 
-      ViewVersion viewVersion =
+      ImmutableViewVersion.Builder versionBuilder =
           ImmutableViewVersion.builder()
               .versionId(1)
               .schemaId(schema.schemaId())
               .addAllRepresentations(representations)
               .defaultNamespace(defaultNamespace)
               .defaultCatalog(defaultCatalog)
               .timestampMillis(System.currentTimeMillis())
-              .putAllSummary(EnvironmentContext.get())
-              .build();
+              .putAllSummary(EnvironmentContext.get());
+
+      if (storageTableIdentifier != null) {
+        versionBuilder.storageTable(storageTableIdentifier);
+      }
+
+      ViewVersion viewVersion = versionBuilder.build();
 
       properties.putAll(viewOverrideProperties());
 
@@ -241,16 +253,21 @@ private View replace(ViewOperations ops) {
               .max(Integer::compareTo)
               .orElseGet(metadata::currentVersionId);
 
-      ViewVersion viewVersion =
+      ImmutableViewVersion.Builder versionBuilder =
           ImmutableViewVersion.builder()
               .versionId(maxVersionId + 1)
               .schemaId(schema.schemaId())
               .addAllRepresentations(representations)
               .defaultNamespace(defaultNamespace)
               .defaultCatalog(defaultCatalog)
               .timestampMillis(System.currentTimeMillis())
-              .putAllSummary(EnvironmentContext.get())
-              .build();
+              .putAllSummary(EnvironmentContext.get());
+
+      if (storageTableIdentifier != null) {
+        versionBuilder.storageTable(storageTableIdentifier);
+      }
+
+      ViewVersion viewVersion = versionBuilder.build();
 
       properties.putAll(viewOverrideProperties());
 

core/src/main/java/org/apache/iceberg/view/BaseViewVersion.java

@@ -19,6 +19,7 @@
 package org.apache.iceberg.view;
 
 import javax.annotation.Nullable;
+import org.apache.iceberg.catalog.TableIdentifier;
 import org.immutables.value.Value;
 
 /**
@@ -45,4 +46,8 @@ default String operation() {
   @Override
   @Nullable
   String defaultCatalog();
+
+  @Override
+  @Nullable
+  TableIdentifier storageTable();
 }

core/src/main/java/org/apache/iceberg/view/RefreshState.java

@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.iceberg.view;
+
+import java.util.List;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * Captures the state of source tables and views at the time of a materialized view refresh
+ * operation. Stored as a JSON-encoded string in the storage table's snapshot summary under the
+ * {@code refresh-state} key.
+ */
+public class RefreshState {
+  public static final String REFRESH_STATE_SUMMARY_KEY = "refresh-state";
+
+  /** Recorded as the snapshot id of a source table that had no snapshot when it was read. */
+  public static final long NO_SNAPSHOT_ID = -1L;
+
+  private final int viewVersionId;
+  private final List<SourceState> sourceStates;
+  private final long refreshStartTimestampMs;
+
+  public RefreshState(
+      int viewVersionId, List<SourceState> sourceStates, long refreshStartTimestampMs) {
+    Preconditions.checkArgument(sourceStates != null, "Source states list is required");
+    this.viewVersionId = viewVersionId;
+    this.sourceStates = sourceStates;
+    this.refreshStartTimestampMs = refreshStartTimestampMs;
+  }
+
+  public int viewVersionId() {
+    return viewVersionId;
+  }
+
+  public List<SourceState> sourceStates() {
+    return sourceStates;
+  }
+
+  public long refreshStartTimestampMs() {
+    return refreshStartTimestampMs;
+  }
+}