SQL DDL documentation

sdks/java/extensions/sql/iceberg/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/iceberg/IcebergMetastore.java

@@ -23,6 +23,7 @@
 import com.fasterxml.jackson.core.type.TypeReference;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.stream.Collectors;
 import org.apache.beam.sdk.extensions.sql.TableUtils;
 import org.apache.beam.sdk.extensions.sql.impl.TableName;
 import org.apache.beam.sdk.extensions.sql.meta.BeamSqlTable;
@@ -60,12 +61,14 @@ public void createTable(Table table) {
       getProvider(table.getType()).createTable(table);
     } else {
       String identifier = getIdentifier(table);
+      Map<String, String> props =
+          TableUtils.getObjectMapper()
+              .convertValue(table.getProperties(), new TypeReference<Map<String, String>>() {})
+              .entrySet().stream()
+              .filter(p -> !p.getKey().startsWith("beam."))
+              .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
       try {
-        Map<String, String> properties =
-            TableUtils.getObjectMapper()
-                .convertValue(table.getProperties(), new TypeReference<Map<String, String>>() {});
-        catalogConfig.createTable(
-            identifier, table.getSchema(), table.getPartitionFields(), properties);
+        catalogConfig.createTable(identifier, table.getSchema(), table.getPartitionFields(), props);
       } catch (TableAlreadyExistsException e) {
         LOG.info(
             "Iceberg table '{}' already exists at location '{}'.", table.getName(), identifier);

sdks/java/extensions/sql/iceberg/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/iceberg/IcebergTable.java

@@ -19,6 +19,7 @@
 
 import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
 
+import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import java.util.HashMap;
 import java.util.List;
@@ -56,6 +57,8 @@ class IcebergTable extends SchemaBaseBeamTable {
   @VisibleForTesting static final String CATALOG_PROPERTIES_FIELD = "catalog_properties";
   @VisibleForTesting static final String HADOOP_CONFIG_PROPERTIES_FIELD = "config_properties";
   @VisibleForTesting static final String CATALOG_NAME_FIELD = "catalog_name";
+  static final String BEAM_WRITE_PROPERTY = "beam.write.";
+  static final String BEAM_READ_PROPERTY = "beam.read.";
 
   @VisibleForTesting
   static final String TRIGGERING_FREQUENCY_FIELD = "triggering_frequency_seconds";
@@ -71,9 +74,21 @@ class IcebergTable extends SchemaBaseBeamTable {
     this.tableIdentifier = tableIdentifier;
     this.catalogConfig = catalogConfig;
     ObjectNode properties = table.getProperties();
-    if (properties.has(TRIGGERING_FREQUENCY_FIELD)) {
-      this.triggeringFrequency = properties.get(TRIGGERING_FREQUENCY_FIELD).asInt();
+    for (Map.Entry<String, JsonNode> property : properties.properties()) {
+      String name = property.getKey().toLowerCase();
+      if (name.startsWith(BEAM_WRITE_PROPERTY)) {
+        String prop = name.substring(BEAM_WRITE_PROPERTY.length());
+        if (prop.equalsIgnoreCase(TRIGGERING_FREQUENCY_FIELD)) {
+          this.triggeringFrequency = property.getValue().asInt();
+        } else {
+          throw new IllegalArgumentException("Unknown Beam write property: " + name);
+        }
+      } else if (name.startsWith(BEAM_READ_PROPERTY)) {
+        // none supported yet
+        throw new IllegalArgumentException("Unknown Beam read property: " + name);
+      }
     }
+
     this.partitionFields = table.getPartitionFields();
   }
 

sdks/java/extensions/sql/iceberg/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/iceberg/PubsubToIcebergIT.java

@@ -156,7 +156,7 @@ public void testSimpleInsertWithPartitionedFields() throws Exception {
             + ") \n"
             + "TYPE 'iceberg' \n"
             + "PARTITIONED BY('id', 'truncate(name, 3)') \n"
-            + "TBLPROPERTIES '{ \"triggering_frequency_seconds\" : 10 }'";
+            + "TBLPROPERTIES '{ \"beam.write.triggering_frequency_seconds\" : 10 }'";
     String insertStatement =
         format("INSERT INTO %s \n", tableIdentifier)
             + "SELECT \n"
@@ -211,7 +211,7 @@ public void testSimpleInsertFlat() throws Exception {
             + "   name VARCHAR \n "
             + ") \n"
             + "TYPE 'iceberg' \n"
-            + "TBLPROPERTIES '{ \"triggering_frequency_seconds\" : 10 }'";
+            + "TBLPROPERTIES '{ \"beam.write.triggering_frequency_seconds\" : 10 }'";
     String insertStatement =
         format("INSERT INTO %s \n", tableIdentifier)
             + "SELECT \n"

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergCatalogConfig.java

@@ -154,7 +154,7 @@ public void createTable(
       String tableIdentifier,
       Schema tableSchema,
       @Nullable List<String> partitionFields,
-      Map<String, String> properties) {
+      @Nullable Map<String, String> properties) {
     TableIdentifier icebergIdentifier = TableIdentifier.parse(tableIdentifier);
     org.apache.iceberg.Schema icebergSchema = IcebergUtils.beamSchemaToIcebergSchema(tableSchema);
     PartitionSpec icebergSpec = PartitionUtils.toPartitionSpec(partitionFields, tableSchema);
@@ -164,7 +164,13 @@ public void createTable(
           icebergIdentifier,
           icebergSchema,
           icebergSpec);
-      catalog().createTable(icebergIdentifier, icebergSchema, icebergSpec, properties);
+      Catalog.TableBuilder builder =
+          catalog().buildTable(icebergIdentifier, icebergSchema).withPartitionSpec(icebergSpec);
+      if (properties != null) {
+        builder = builder.withProperties(properties);
+      }
+      builder.create();
+
       LOG.info("Successfully created table '{}'.", icebergIdentifier);
     } catch (AlreadyExistsException e) {
       throw new TableAlreadyExistsException(e);