Core: Add View support

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

@@ -111,6 +111,16 @@ public interface ViewBuilder {
    */
   ViewBuilder withProperty(String key, String value);
 
+  /**
+   * Add a view representation
+   *
+   * @param representation The view representation to add
+   * @return this for method chaining
+   */
+  default ViewBuilder withRepresentation(ViewRepresentation representation) {
+    throw new UnsupportedOperationException("Adding a view representation is not supported");
+  }
+
   /**
    * Create the view.
    *

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

@@ -0,0 +1,233 @@
+/*
+ * 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;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.catalog.ViewCatalog;
+import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.exceptions.NoSuchViewException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.view.BaseView;
+import org.apache.iceberg.view.ImmutableSQLViewRepresentation;
+import org.apache.iceberg.view.ImmutableViewHistoryEntry;
+import org.apache.iceberg.view.ImmutableViewMetadata;
+import org.apache.iceberg.view.ImmutableViewVersion;
+import org.apache.iceberg.view.View;
+import org.apache.iceberg.view.ViewBuilder;
+import org.apache.iceberg.view.ViewMetadata;
+import org.apache.iceberg.view.ViewOperations;
+import org.apache.iceberg.view.ViewRepresentation;
+import org.apache.iceberg.view.ViewVersion;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class BaseMetastoreViewCatalog extends BaseMetastoreCatalog implements ViewCatalog {
+  private static final Logger LOG = LoggerFactory.getLogger(BaseMetastoreViewCatalog.class);
+
+  protected abstract ViewOperations newViewOps(TableIdentifier viewIdentifier);
+
+  @Override
+  public void initialize(String name, Map<String, String> properties) {}
+
+  @Override
+  public String name() {
+    return super.name();
+  }
+
+  @Override
+  public ViewBuilder buildView(TableIdentifier identifier) {
+    return new BaseViewBuilder(identifier);
+  }
+
+  protected class BaseViewBuilder implements ViewBuilder {
+    private final TableIdentifier viewIdentifier;
+    private final ImmutableViewVersion.Builder viewVersionBuilder = ImmutableViewVersion.builder();
+    private final ImmutableSQLViewRepresentation.Builder viewRepresentationBuilder =
+        ImmutableSQLViewRepresentation.builder();
+    private final ImmutableViewMetadata.Builder builder = ImmutableViewMetadata.builder();
+
+    public BaseViewBuilder(TableIdentifier viewIdentifier) {
+      Preconditions.checkArgument(
+          isValidIdentifier(viewIdentifier), "Invalid view identifier: null");
+      this.viewIdentifier = viewIdentifier;
+    }
+
+    @Override
+    public ViewBuilder withSchema(Schema schema) {
+      builder.addSchemas(schema).currentSchemaId(schema.schemaId());
+      viewVersionBuilder.schemaId(schema.schemaId());
+      return this;
+    }
+
+    @Override
+    public ViewBuilder withQuery(String query) {
+      viewRepresentationBuilder.sql(query);
+      return this;
+    }
+
+    @Override
+    public ViewBuilder withDialect(String dialect) {
+      viewRepresentationBuilder.dialect(dialect);
+      return this;
+    }
+
+    @Override
+    public ViewBuilder withDefaultCatalog(String defaultCatalog) {
+      viewRepresentationBuilder.defaultCatalog(defaultCatalog);
+      return this;
+    }
+
+    @Override
+    public ViewBuilder withDefaultNamespace(Namespace defaultNamespace) {
+      viewRepresentationBuilder.defaultNamespace(defaultNamespace);
+      return this;
+    }
+
+    @Override
+    public ViewBuilder withQueryColumnNames(List<String> queryColumnNames) {
+      throw new UnsupportedOperationException("withQueryColumnNames is currently not supported");
+    }
+
+    @Override
+    public ViewBuilder withFieldAliases(List<String> fieldAliases) {
+      viewRepresentationBuilder.addAllFieldAliases(fieldAliases);
+      return this;
+    }
+
+    @Override
+    public ViewBuilder withFieldComments(List<String> fieldComments) {
+      viewRepresentationBuilder.addAllFieldComments(fieldComments);
+      return this;
+    }
+
+    @Override
+    public ViewBuilder withProperties(Map<String, String> properties) {
+      builder.putAllProperties(properties);
+      return this;
+    }
+
+    @Override
+    public ViewBuilder withProperty(String key, String value) {
+      builder.putProperties(key, value);
+      return this;
+    }
+
+    @Override
+    public ViewBuilder withRepresentation(ViewRepresentation representation) {
+      viewVersionBuilder.addRepresentations(representation);
+      return this;
+    }
+
+    @Override
+    public View create() {
+      ViewOperations ops = newViewOps(viewIdentifier);
+      if (ops.current() != null) {
+        throw new AlreadyExistsException("View already exists: %s", viewIdentifier);
+      }
+
+      long timestampMillis = System.currentTimeMillis();
+
+      ViewVersion viewVersion =
+          viewVersionBuilder
+              .versionId(1)
+              .addRepresentations(viewRepresentationBuilder.build())
+              .timestampMillis(timestampMillis)
+              .putSummary("operation", "create")
+              .build();
+
+      ViewMetadata viewMetadata =
+          builder
+              .location(defaultWarehouseLocation(viewIdentifier))
+              .formatVersion(1)
+              .currentVersionId(viewVersion.versionId())
+              .addVersions(viewVersion)
+              .addHistory(
+                  ImmutableViewHistoryEntry.builder()
+                      .timestampMillis(timestampMillis)
+                      .versionId(viewVersion.versionId())
+                      .build())
+              .build();
+
+      try {
+        ops.commit(null, viewMetadata);
+      } catch (CommitFailedException ignored) {
+        throw new AlreadyExistsException("View was created concurrently: %s", viewIdentifier);
+      }
+
+      return new BaseView(ops, fullTableName(name(), viewIdentifier));
+    }
+
+    @Override
+    public View replace() {
+      ViewOperations ops = newViewOps(viewIdentifier);
+      if (ops.current() == null) {
+        throw new NoSuchViewException("View does not exist: %s", viewIdentifier);
+      }
+
+      ViewMetadata metadata = ops.current();
+
+      long timestampMillis = System.currentTimeMillis();
+
+      ViewVersion viewVersion =
+          viewVersionBuilder
+              .versionId(metadata.currentVersionId() + 1)
+              .addRepresentations(viewRepresentationBuilder.build())
+              .timestampMillis(timestampMillis)
+              .putSummary("operation", "replace")
+              .build();
+
+      ViewMetadata replacement =
+          builder
+              .formatVersion(metadata.formatVersion())
+              .location(metadata.location())
+              .addAllSchemas(metadata.schemas())
+              .currentVersionId(viewVersion.versionId())
+              .addAllVersions(metadata.versions())
+              .addVersions(viewVersion)
+              .addAllHistory(metadata.history())
+              .addHistory(
+                  ImmutableViewHistoryEntry.builder()
+                      .timestampMillis(timestampMillis)
+                      .versionId(viewVersion.versionId())
+                      .build())
+              .build();
+
+      try {
+        ops.commit(metadata, replacement);
+      } catch (CommitFailedException ignored) {
+        throw new AlreadyExistsException("View was updated concurrently: %s", viewIdentifier);
+      }
+
+      return new BaseView(ops, fullTableName(name(), viewIdentifier));
+    }
+
+    @Override
+    public View createOrReplace() {
+      if (newViewOps(viewIdentifier).current() == null) {
+        return create();
+      } else {
+        return replace();
+      }
+    }
+  }
+}