Allow table defaults to be configured and/ or enforced at catalog level using catalog properties.

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

@@ -19,6 +19,7 @@
 
 package org.apache.iceberg;
 
+import java.util.Collections;
 import java.util.Map;
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.catalog.TableIdentifier;
@@ -27,12 +28,19 @@
 import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+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 BaseMetastoreCatalog implements Catalog {
   private static final Logger LOG = LoggerFactory.getLogger(BaseMetastoreCatalog.class);
+  private Map<String, String> catalogProps = Collections.emptyMap();
+
+  @Override
+  public void initialize(String name, Map<String, String> properties) {
+    catalogProps = properties;
+  }
 
   @Override
   public Table loadTable(TableIdentifier identifier) {
@@ -106,7 +114,7 @@ public String toString() {
   protected class BaseMetastoreCatalogTableBuilder implements TableBuilder {
     private final TableIdentifier identifier;
     private final Schema schema;
-    private final ImmutableMap.Builder<String, String> propertiesBuilder = ImmutableMap.builder();
+    private final Map<String, String> propertiesBuilder = Maps.newHashMap();
     private PartitionSpec spec = PartitionSpec.unpartitioned();
     private SortOrder sortOrder = SortOrder.unsorted();
     private String location = null;
@@ -116,6 +124,7 @@ public BaseMetastoreCatalogTableBuilder(TableIdentifier identifier, Schema schem
 
       this.identifier = identifier;
       this.schema = schema;
+      this.propertiesBuilder.putAll(tableDefaultProperties());
     }
 
     @Override
@@ -139,7 +148,7 @@ public TableBuilder withLocation(String newLocation) {
     @Override
     public TableBuilder withProperties(Map<String, String> properties) {
       if (properties != null) {
-        propertiesBuilder.putAll(properties);
+        this.propertiesBuilder.putAll(properties);
       }
       return this;
     }
@@ -158,8 +167,8 @@ public Table create() {
       }
 
       String baseLocation = location != null ? location : defaultWarehouseLocation(identifier);
-      Map<String, String> properties = propertiesBuilder.build();
-      TableMetadata metadata = TableMetadata.newTableMetadata(schema, spec, sortOrder, baseLocation, properties);
+      this.propertiesBuilder.putAll(tableOverrideProperties());
+      TableMetadata metadata = TableMetadata.newTableMetadata(schema, spec, sortOrder, baseLocation, propertiesBuilder);
 
       try {
         ops.commit(null, metadata);
@@ -178,8 +187,8 @@ public Transaction createTransaction() {
       }
 
       String baseLocation = location != null ? location : defaultWarehouseLocation(identifier);
-      Map<String, String> properties = propertiesBuilder.build();
-      TableMetadata metadata = TableMetadata.newTableMetadata(schema, spec, sortOrder, baseLocation, properties);
+      this.propertiesBuilder.putAll(tableOverrideProperties());
+      TableMetadata metadata = TableMetadata.newTableMetadata(schema, spec, sortOrder, baseLocation, propertiesBuilder);
       return Transactions.createTableTransaction(identifier.toString(), ops, metadata);
     }
 
@@ -200,12 +209,13 @@ private Transaction newReplaceTableTransaction(boolean orCreate) {
       }
 
       TableMetadata metadata;
+      this.propertiesBuilder.putAll(tableOverrideProperties());
       if (ops.current() != null) {
         String baseLocation = location != null ? location : ops.current().location();
-        metadata = ops.current().buildReplacement(schema, spec, sortOrder, baseLocation, propertiesBuilder.build());
+        metadata = ops.current().buildReplacement(schema, spec, sortOrder, baseLocation, propertiesBuilder);
       } else {
         String baseLocation = location != null ? location : defaultWarehouseLocation(identifier);
-        metadata = TableMetadata.newTableMetadata(schema, spec, sortOrder, baseLocation, propertiesBuilder.build());
+        metadata = TableMetadata.newTableMetadata(schema, spec, sortOrder, baseLocation, propertiesBuilder);
       }
 
       if (orCreate) {
@@ -214,6 +224,32 @@ private Transaction newReplaceTableTransaction(boolean orCreate) {
         return Transactions.replaceTableTransaction(identifier.toString(), ops, metadata);
       }
     }
+
+    /**
+     * Get default table properties set at Catalog level through catalog properties.
+     *
+     * @return default table properties specified in catalog properties
+     */
+    private Map<String, String> tableDefaultProperties() {
+      if (catalogProps == null || catalogProps.isEmpty()) {
+        return Collections.emptyMap();
+      }
+
+      return PropertyUtil.propertiesWithPrefix(catalogProps, CatalogProperties.TABLE_DEFAULT_PREFIX);
+    }
+
+    /**
+     * Get table properties that are enforced at Catalog level through catalog properties.
+     *
+     * @return default table properties enforced through catalog properties
+     */
+    private Map<String, String> tableOverrideProperties() {
+      if (catalogProps == null || catalogProps.isEmpty()) {
+        return Collections.emptyMap();
+      }
+
+      return PropertyUtil.propertiesWithPrefix(catalogProps, CatalogProperties.TABLE_OVERRIDE_PREFIX);
+    }
   }
 
   protected static String fullTableName(String catalogName, TableIdentifier identifier) {

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

@@ -29,6 +29,8 @@ private CatalogProperties() {
   public static final String CATALOG_IMPL = "catalog-impl";
   public static final String FILE_IO_IMPL = "io-impl";
   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.";
 
   /**
    * Controls whether the catalog will cache table entries upon load.

core/src/main/java/org/apache/iceberg/hadoop/HadoopCatalog.java

@@ -98,6 +98,8 @@ public HadoopCatalog() {
 
   @Override
   public void initialize(String name, Map<String, String> properties) {
+    super.initialize(name, properties);
+
     String inputWarehouseLocation = properties.get(CatalogProperties.WAREHOUSE_LOCATION);
     Preconditions.checkArgument(inputWarehouseLocation != null && !inputWarehouseLocation.equals(""),
         "Cannot instantiate hadoop catalog. No location provided for warehouse (Set warehouse config)");

core/src/main/java/org/apache/iceberg/util/PropertyUtil.java

@@ -19,7 +19,9 @@
 
 package org.apache.iceberg.util;
 
+import java.util.Collections;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 public class PropertyUtil {
 
@@ -70,4 +72,16 @@ public static String propertyAsString(Map<String, String> properties,
     }
     return defaultValue;
   }
+
+  public static Map<String, String> propertiesWithPrefix(Map<String, String> properties,
+                                                         String prefix) {
+    if (properties == null || properties.isEmpty()) {
+      return Collections.emptyMap();
+    }
+
+    return properties.entrySet().stream()
+        .filter(e -> e.getKey().startsWith(prefix))
+        .collect(Collectors.toMap(
+            e -> e.getKey().replace(prefix, ""), Map.Entry::getValue));
+  }
 }

core/src/test/java/org/apache/iceberg/hadoop/HadoopTableTestBase.java

@@ -24,7 +24,9 @@
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.zip.GZIPOutputStream;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.iceberg.CatalogProperties;
@@ -170,10 +172,15 @@ void rewriteMetadataAsGzipWithOldExtension() throws IOException {
   }
 
   protected HadoopCatalog hadoopCatalog() throws IOException {
+    return hadoopCatalog(Collections.emptyMap());
+  }
+
+  protected HadoopCatalog hadoopCatalog(Map<String, String> catalogProperties) throws IOException {
     HadoopCatalog hadoopCatalog = new HadoopCatalog();
     hadoopCatalog.setConf(new Configuration());
     hadoopCatalog.initialize("hadoop",
-            ImmutableMap.of(CatalogProperties.WAREHOUSE_LOCATION, temp.newFolder().getAbsolutePath()));
+        ImmutableMap.<String, String>builder().putAll(catalogProperties).put(CatalogProperties.WAREHOUSE_LOCATION,
+            temp.newFolder().getAbsolutePath()).build());
     return hadoopCatalog;
   }
 }

core/src/test/java/org/apache/iceberg/hadoop/TestHadoopCatalog.java

@@ -22,6 +22,7 @@
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
@@ -547,4 +548,60 @@ private static void addVersionsToTable(Table table) {
     table.newAppend().appendFile(dataFile1).commit();
     table.newAppend().appendFile(dataFile2).commit();
   }
+
+  @Test
+  public void testTablePropsDefaultsWithoutConflict() throws IOException {
+    TableIdentifier tableIdent = TableIdentifier.of("db", "ns1", "ns2", "tbl");
+    Map<String, String> catalogProps = ImmutableMap.of("table-default.key3", "value3",
+        "table-override.key4", "value4");
+
+    Table table = hadoopCatalog(catalogProps).buildTable(tableIdent, SCHEMA)
+        .withPartitionSpec(SPEC)
+        .withProperties(null)
+        .withProperty("key1", "value1")
+        .withProperty("key2", "value2")
+        .create();
+
+    Assert.assertEquals("value1", table.properties().get("key1"));
+    Assert.assertEquals("value2", table.properties().get("key2"));
+    Assert.assertEquals("value3", table.properties().get("key3"));
+    Assert.assertEquals("value4", table.properties().get("key4"));
+  }
+
+
+  @Test
+  public void testTablePropsOverrideCatalogDefaultProps() throws IOException {
+    TableIdentifier tableIdent = TableIdentifier.of("db", "ns1", "ns2", "tbl");
+    Map<String, String> catalogProps = ImmutableMap.of("table-default.key3", "value3");
+
+    Table table = hadoopCatalog(catalogProps).buildTable(tableIdent, SCHEMA)
+        .withPartitionSpec(SPEC)
+        .withProperties(null)
+        .withProperty("key1", "value1")
+        .withProperty("key2", "value2")
+        .withProperty("key3", "value31")
+        .create();
+
+    Assert.assertEquals("value1", table.properties().get("key1"));
+    Assert.assertEquals("value2", table.properties().get("key2"));
+    Assert.assertEquals("value31", table.properties().get("key3"));
+  }
+
+  @Test
+  public void testCatalogOverridePropsOverrideTableDefaults() throws IOException {
+    TableIdentifier tableIdent = TableIdentifier.of("db", "ns1", "ns2", "tbl");
+    Map<String, String> catalogProps = ImmutableMap.of("table-override.key3", "value3");
+
+    Table table = hadoopCatalog(catalogProps).buildTable(tableIdent, SCHEMA)
+        .withPartitionSpec(SPEC)
+        .withProperties(null)
+        .withProperty("key1", "value1")
+        .withProperty("key2", "value2")
+        .withProperty("key3", "value31")
+        .create();
+
+    Assert.assertEquals("value1", table.properties().get("key1"));
+    Assert.assertEquals("value2", table.properties().get("key2"));
+    Assert.assertEquals("value3", table.properties().get("key3"));
+  }
 }

hive-metastore/src/main/java/org/apache/iceberg/hive/HiveCatalog.java

@@ -78,6 +78,8 @@ public HiveCatalog() {
 
   @Override
   public void initialize(String inputName, Map<String, String> properties) {
+    super.initialize(name, properties);
+
     this.name = inputName;
     if (conf == null) {
       LOG.warn("No Hadoop Configuration was set, using the default environment Configuration");

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

@@ -24,6 +24,7 @@
 import org.apache.hadoop.hive.metastore.api.Database;
 import org.apache.iceberg.AssertHelpers;
 import org.apache.iceberg.CachingCatalog;
+import org.apache.iceberg.CatalogUtil;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.SortOrder;
@@ -468,4 +469,84 @@ public void testUUIDinTableProperties() throws Exception {
       catalog.dropTable(tableIdentifier);
     }
   }
+
+  @Test
+  public void testTablePropsDefaultsWithoutConflict() {
+    Schema schema = new Schema(
+        required(1, "id", Types.IntegerType.get(), "unique ID")
+    );
+    TableIdentifier tableIdent = TableIdentifier.of(DB_NAME, "tbl");
+
+    ImmutableMap<String, String> catalogProps = ImmutableMap.of("table-default.key3", "value3",
+        "table-override.key4", "value4");
+    HiveCatalog hiveCatalog = (HiveCatalog) CatalogUtil.loadCatalog(HiveCatalog.class.getName(),
+        CatalogUtil.ICEBERG_CATALOG_TYPE_HIVE, catalogProps, hiveConf);
+
+    try {
+      Table table = hiveCatalog.buildTable(tableIdent, schema)
+          .withProperty("key1", "value1")
+          .withProperty("key2", "value2")
+          .create();
+
+      Assert.assertEquals("value1", table.properties().get("key1"));
+      Assert.assertEquals("value2", table.properties().get("key2"));
+      Assert.assertEquals("value3", table.properties().get("key3"));
+      Assert.assertEquals("value4", table.properties().get("key4"));
+    } finally {
+      hiveCatalog.dropTable(tableIdent);
+    }
+  }
+
+  @Test
+  public void testTablePropsOverrideCatalogDefaultProps() {
+    Schema schema = new Schema(
+        required(1, "id", Types.IntegerType.get(), "unique ID")
+    );
+    TableIdentifier tableIdent = TableIdentifier.of(DB_NAME, "tbl");
+
+    ImmutableMap<String, String> catalogProps = ImmutableMap.of("table-default.key3", "value3");
+    HiveCatalog hiveCatalog = (HiveCatalog) CatalogUtil.loadCatalog(HiveCatalog.class.getName(),
+        CatalogUtil.ICEBERG_CATALOG_TYPE_HIVE, catalogProps, hiveConf);
+
+    try {
+      Table table = hiveCatalog.buildTable(tableIdent, schema)
+          .withProperty("key1", "value1")
+          .withProperty("key2", "value2")
+          .withProperty("key3", "value31")
+          .create();
+
+      Assert.assertEquals("value1", table.properties().get("key1"));
+      Assert.assertEquals("value2", table.properties().get("key2"));
+      Assert.assertEquals("value31", table.properties().get("key3"));
+    } finally {
+      hiveCatalog.dropTable(tableIdent);
+    }
+  }
+
+  @Test
+  public void testCatalogOverridePropsOverrideTableDefaults() {
+    Schema schema = new Schema(
+        required(1, "id", Types.IntegerType.get(), "unique ID")
+    );
+    TableIdentifier tableIdent = TableIdentifier.of(DB_NAME, "tbl");
+
+    ImmutableMap<String, String> catalogProps = ImmutableMap.of(
+        "table-override.key3", "value3");
+    HiveCatalog hiveCatalog = (HiveCatalog) CatalogUtil.loadCatalog(HiveCatalog.class.getName(),
+        CatalogUtil.ICEBERG_CATALOG_TYPE_HIVE, catalogProps, hiveConf);
+
+    try {
+      Table table = hiveCatalog.buildTable(tableIdent, schema)
+          .withProperty("key1", "value1")
+          .withProperty("key2", "value2")
+          .withProperty("key3", "value31")
+          .create();
+
+      Assert.assertEquals("value1", table.properties().get("key1"));
+      Assert.assertEquals("value2", table.properties().get("key2"));
+      Assert.assertEquals("value3", table.properties().get("key3"));
+    } finally {
+      hiveCatalog.dropTable(tableIdent);
+    }
+  }
 }