Core: Add support for view-default property in catalog (#11064)

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

@@ -29,6 +29,7 @@ public class CatalogProperties {
   public static final String WAREHOUSE_LOCATION = "warehouse";
   public static final String TABLE_DEFAULT_PREFIX = "table-default.";
   public static final String TABLE_OVERRIDE_PREFIX = "table-override.";
+  public static final String VIEW_DEFAULT_PREFIX = "view-default.";
   public static final String METRICS_REPORTER_IMPL = "metrics-reporter-impl";
 
   /**

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

@@ -1318,6 +1318,21 @@ public class RESTSessionCatalog extends BaseViewSessionCatalog
       checkViewIdentifierIsValid(identifier);
       this.identifier = identifier;
       this.context = context;
+      this.properties.putAll(viewDefaultProperties());
+    }
+
+    /**
+     * Get default view properties set at Catalog level through catalog properties.
+     *
+     * @return default view properties specified in catalog properties
+     */
+    private Map<String, String> viewDefaultProperties() {
+      Map<String, String> viewDefaultProperties =
+          PropertyUtil.propertiesWithPrefix(properties(), CatalogProperties.VIEW_DEFAULT_PREFIX);
+      LOG.info(
+          "View properties set at catalog level through catalog properties: {}",
+          viewDefaultProperties);
+      return viewDefaultProperties;
     }
 
     @Override

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

@@ -21,6 +21,7 @@ package org.apache.iceberg.view;
 import java.util.List;
 import java.util.Map;
 import org.apache.iceberg.BaseMetastoreCatalog;
+import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.EnvironmentContext;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Transaction;
@@ -33,8 +34,13 @@ import org.apache.iceberg.exceptions.NoSuchViewException;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.PropertyUtil;
+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 identifier);
 
   @Override
@@ -79,6 +85,21 @@ public abstract class BaseMetastoreViewCatalog extends BaseMetastoreCatalog impl
       Preconditions.checkArgument(
           isValidIdentifier(identifier), "Invalid view identifier: %s", identifier);
       this.identifier = identifier;
+      this.properties.putAll(viewDefaultProperties());
+    }
+
+    /**
+     * Get default view properties set at Catalog level through catalog properties.
+     *
+     * @return default view properties specified in catalog properties
+     */
+    private Map<String, String> viewDefaultProperties() {
+      Map<String, String> viewDefaultProperties =
+          PropertyUtil.propertiesWithPrefix(properties(), CatalogProperties.VIEW_DEFAULT_PREFIX);
+      LOG.info(
+          "View properties set at catalog level through catalog properties: {}",
+          viewDefaultProperties);
+      return viewDefaultProperties;
     }
 
     @Override

core/src/test/java/org/apache/iceberg/inmemory/TestInMemoryViewCatalog.java

@@ -18,6 +18,7 @@
  */
 package org.apache.iceberg.inmemory;
 
+import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.view.ViewCatalogTests;
@@ -29,7 +30,12 @@ public class TestInMemoryViewCatalog extends ViewCatalogTests<InMemoryCatalog> {
   @BeforeEach
   public void before() {
     this.catalog = new InMemoryCatalog();
-    this.catalog.initialize("in-memory-catalog", ImmutableMap.of());
+    this.catalog.initialize(
+        "in-memory-catalog",
+        ImmutableMap.<String, String>builder()
+            .put(CatalogProperties.VIEW_DEFAULT_PREFIX + "key1", "catalog-default-key1")
+            .put(CatalogProperties.VIEW_DEFAULT_PREFIX + "key2", "catalog-default-key2")
+            .build());
   }
 
   @Override

core/src/test/java/org/apache/iceberg/jdbc/TestJdbcViewCatalog.java

@@ -57,6 +57,8 @@ public class TestJdbcViewCatalog extends ViewCatalogTests<JdbcCatalog> {
     properties.put(JdbcCatalog.PROPERTY_PREFIX + "password", "password");
     properties.put(CatalogProperties.WAREHOUSE_LOCATION, tableDir.toAbsolutePath().toString());
     properties.put(JdbcUtil.SCHEMA_VERSION_PROPERTY, JdbcUtil.SchemaVersion.V1.name());
+    properties.put(CatalogProperties.VIEW_DEFAULT_PREFIX + "key1", "catalog-default-key1");
+    properties.put(CatalogProperties.VIEW_DEFAULT_PREFIX + "key2", "catalog-default-key2");
 
     catalog = new JdbcCatalog();
     catalog.setConf(new Configuration());

core/src/test/java/org/apache/iceberg/rest/TestRESTViewCatalog.java

@@ -72,7 +72,11 @@ public class TestRESTViewCatalog extends ViewCatalogTests<RESTCatalog> {
     this.backendCatalog = new InMemoryCatalog();
     this.backendCatalog.initialize(
         "in-memory",
-        ImmutableMap.of(CatalogProperties.WAREHOUSE_LOCATION, warehouse.getAbsolutePath()));
+        ImmutableMap.<String, String>builder()
+            .put(CatalogProperties.WAREHOUSE_LOCATION, warehouse.getAbsolutePath())
+            .put(CatalogProperties.VIEW_DEFAULT_PREFIX + "key1", "catalog-default-key1")
+            .put(CatalogProperties.VIEW_DEFAULT_PREFIX + "key2", "catalog-default-key2")
+            .build());
 
     RESTCatalogAdapter adaptor =
         new RESTCatalogAdapter(backendCatalog) {

core/src/test/java/org/apache/iceberg/rest/TestRESTViewCatalogWithAssumedViewSupport.java

@@ -90,6 +90,10 @@ public class TestRESTViewCatalogWithAssumedViewSupport extends TestRESTViewCatal
             "catalog:12345",
             // assume that the server supports view endpoints
             RESTSessionCatalog.VIEW_ENDPOINTS_SUPPORTED,
-            "true"));
+            "true",
+            CatalogProperties.VIEW_DEFAULT_PREFIX + "key1",
+            "catalog-default-key1",
+            CatalogProperties.VIEW_DEFAULT_PREFIX + "key2",
+            "catalog-default-key2"));
   }
 }

core/src/test/java/org/apache/iceberg/view/ViewCatalogTests.java

@@ -127,6 +127,37 @@ public abstract class ViewCatalogTests<C extends ViewCatalog & SupportsNamespace
     assertThat(catalog().viewExists(identifier)).as("View should not exist").isFalse();
   }
 
+  @Test
+  public void defaultViewProperties() {
+    TableIdentifier identifier = TableIdentifier.of("ns", "view");
+
+    if (requiresNamespaceCreate()) {
+      catalog().createNamespace(identifier.namespace());
+    }
+
+    assertThat(catalog().viewExists(identifier)).as("View should not exist").isFalse();
+
+    View view =
+        catalog()
+            .buildView(identifier)
+            .withSchema(SCHEMA)
+            .withDefaultNamespace(identifier.namespace())
+            .withDefaultCatalog(catalog().name())
+            .withQuery("spark", "select * from ns.tbl")
+            .withProperty("key2", "catalog-overridden-key2")
+            .withProperty("prop1", "val1")
+            .create();
+
+    assertThat(view).isNotNull();
+    assertThat(view.properties())
+        .containsEntry("key1", "catalog-default-key1")
+        .containsEntry("key2", "catalog-overridden-key2")
+        .containsEntry("prop1", "val1");
+
+    assertThat(catalog().dropView(identifier)).isTrue();
+    assertThat(catalog().viewExists(identifier)).as("View should not exist").isFalse();
+  }
+
   @Test
   public void completeCreateView() {
     TableIdentifier identifier = TableIdentifier.of("ns", "view");

docs/docs/spark-configuration.md

@@ -77,6 +77,7 @@ Both catalogs are configured using properties nested under the catalog name. Com
 | spark.sql.catalog._catalog-name_.cache.expiration-interval-ms | `30000` (30 seconds) | Duration after which cached catalog entries are expired; Only effective if `cache-enabled` is `true`. `-1` disables cache expiration and `0` disables caching entirely, irrespective of `cache-enabled`. Default is `30000` (30 seconds) |
 | spark.sql.catalog._catalog-name_.table-default._propertyKey_  |                               | Default Iceberg table property value for property key _propertyKey_, which will be set on tables created by this catalog if not overridden                                                                                               |
 | spark.sql.catalog._catalog-name_.table-override._propertyKey_ |                               | Enforced Iceberg table property value for property key _propertyKey_, which cannot be overridden on table creation by user                                                                                                               |
+| spark.sql.catalog._catalog-name_.view-default._propertyKey_  |                               | Default Iceberg view property value for property key _propertyKey_, which will be set on views created by this catalog if not overridden                                                                                               |
 | spark.sql.catalog._catalog-name_.use-nullable-query-schema | `true` or `false` | Whether to preserve fields' nullability when creating the table using CTAS and RTAS. If set to `true`, all fields will be marked as nullable. If set to `false`, fields' nullability will be preserved. The default value is `true`. Available in Spark 3.5 and above.   |
 
 Additional properties can be found in common [catalog configuration](configuration.md#catalog-properties).

hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveViewCatalog.java

@@ -66,7 +66,11 @@ public class TestHiveViewCatalog extends ViewCatalogTests<HiveCatalog> {
                 CatalogUtil.ICEBERG_CATALOG_TYPE_HIVE,
                 ImmutableMap.of(
                     CatalogProperties.CLIENT_POOL_CACHE_EVICTION_INTERVAL_MS,
-                    String.valueOf(TimeUnit.SECONDS.toMillis(10))),
+                    String.valueOf(TimeUnit.SECONDS.toMillis(10)),
+                    CatalogProperties.VIEW_DEFAULT_PREFIX + "key1",
+                    "catalog-default-key1",
+                    CatalogProperties.VIEW_DEFAULT_PREFIX + "key2",
+                    "catalog-default-key2"),
                 HIVE_METASTORE_EXTENSION.hiveConf());
   }
 

nessie/src/test/java/org/apache/iceberg/nessie/TestNessieViewCatalog.java

@@ -128,7 +128,11 @@ public class TestNessieViewCatalog extends ViewCatalogTests<NessieCatalog> {
             CatalogProperties.WAREHOUSE_LOCATION,
             temp.toUri().toString(),
             "client-api-version",
-            apiVersion == NessieApiVersion.V2 ? "2" : "1");
+            apiVersion == NessieApiVersion.V2 ? "2" : "1",
+            CatalogProperties.VIEW_DEFAULT_PREFIX + "key1",
+            "catalog-default-key1",
+            CatalogProperties.VIEW_DEFAULT_PREFIX + "key2",
+            "catalog-default-key2");
     newCatalog.initialize("nessie", options);
     return newCatalog;
   }

open-api/src/testFixtures/java/org/apache/iceberg/rest/RCKUtils.java

@@ -96,6 +96,10 @@ class RCKUtils {
     catalogProperties.putIfAbsent(
         CatalogProperties.URI, String.format("http://localhost:%s/", port));
     catalogProperties.putIfAbsent(CatalogProperties.WAREHOUSE_LOCATION, "rck_warehouse");
+    catalogProperties.putIfAbsent(
+        CatalogProperties.VIEW_DEFAULT_PREFIX + "key1", "catalog-default-key1");
+    catalogProperties.putIfAbsent(
+        CatalogProperties.VIEW_DEFAULT_PREFIX + "key2", "catalog-default-key2");
 
     RESTCatalog catalog = new RESTCatalog();
     catalog.setConf(new Configuration());