[IcebergIO] Integrate with Beam SQL (#34799)

* add Iceberg table provider and tests

* properties go in the tableprovider initialization

* tobuilder

* streaming integration test

* spotless

* extend test to include multi nested types; fix iceberg <-> conversion logic

* add to changes.md

* spotless

* fix tests

* clean

* update CHANGES

* add projection pushdown and column pruning

* spotless

* fixes

* fixes

* updates

* sync with HEAD and use new Catalog implementation

* mark new interfaces @internal

* spotless

* fix unparse method

Author: ahmedabu98

.github/trigger_files/IO_Iceberg_Integration_Tests.json

@@ -1,4 +1,4 @@
 {
     "comment": "Modify this file in a trivial way to cause this test suite to run.",
-    "modification": 4
+    "modification": 1
 }

.github/trigger_files/beam_PostCommit_Python_Xlang_IO_Direct.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 4
+  "modification": 5
 }

.github/trigger_files/beam_PostCommit_SQL.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 1
+  "modification": 2
 }

CHANGES.md

@@ -75,10 +75,12 @@
 ## I/Os
 
 * Support for X source added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
+* [IcebergIO] Now available with Beam SQL! ([#34799](https://github.com/apache/beam/pull/34799))
 * [IcebergIO] Support reading with column pruning ([#34856](https://github.com/apache/beam/pull/34856))
 * [IcebergIO] Support reading with pushdown filtering ([#34827](https://github.com/apache/beam/pull/34827))
 
 ## New Features / Improvements
+* [Beam SQL] Introducing Beam Catalogs ([#35223](https://github.com/apache/beam/pull/35223))
 * Adding Google Storage Requests Pays feature (Golang)([#30747](https://github.com/apache/beam/issues/30747)).
 * X feature added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
 * [Python] Prism runner now auto-enabled for some Python pipelines using the direct runner ([#34921](https://github.com/apache/beam/pull/34921)).

sdks/java/extensions/sql/build.gradle

@@ -74,6 +74,10 @@ dependencies {
   fmppTask "org.freemarker:freemarker:2.3.31"
   fmppTemplates library.java.vendored_calcite_1_28_0
   implementation project(path: ":sdks:java:core", configuration: "shadow")
+  implementation project(":sdks:java:managed")
+  implementation project(":sdks:java:io:iceberg")
+  runtimeOnly project(":sdks:java:io:iceberg:bqms")
+  runtimeOnly project(":sdks:java:io:iceberg:hive")
   implementation project(":sdks:java:extensions:avro")
   implementation project(":sdks:java:extensions:join-library")
   permitUnusedDeclared project(":sdks:java:extensions:join-library") // BEAM-11761

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/parser/SqlCreateCatalog.java

@@ -99,16 +99,16 @@ public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
         if (i > 0) {
           writer.keyword(",");
         }
-        properties.get(i).unparse(writer, leftPrec, rightPrec);
-      }
-
-      for (int i = 0; i < properties.size(); i += 2) {
-        if (i > 0) {
-          writer.keyword(",");
-        }
-        properties.get(i).unparse(writer, leftPrec, rightPrec); // key
+        SqlNode property = properties.get(i);
+        checkState(
+            property instanceof SqlNodeList,
+            String.format(
+                "Unexpected properties entry '%s' of class '%s'", property, property.getClass()));
+        SqlNodeList kv = ((SqlNodeList) property);
+
+        kv.get(0).unparse(writer, leftPrec, rightPrec); // key
         writer.keyword("=");
-        properties.get(i + 1).unparse(writer, leftPrec, rightPrec); // value
+        kv.get(1).unparse(writer, leftPrec, rightPrec); // value
       }
       writer.keyword(")");
     }

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/catalog/Catalog.java

@@ -18,13 +18,15 @@
 package org.apache.beam.sdk.extensions.sql.meta.catalog;
 
 import java.util.Map;
+import org.apache.beam.sdk.annotations.Internal;
 import org.apache.beam.sdk.extensions.sql.meta.store.MetaStore;
 
 /**
  * Represents a named and configurable container for managing tables. Is defined with a type and
  * configuration properties. Uses an underlying {@link MetaStore} to manage tables and table
  * providers.
  */
+@Internal
 public interface Catalog {
   /** A type that defines this catalog. */
   String type();

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/catalog/CatalogManager.java

@@ -18,6 +18,7 @@
 package org.apache.beam.sdk.extensions.sql.meta.catalog;
 
 import java.util.Map;
+import org.apache.beam.sdk.annotations.Internal;
 import org.apache.beam.sdk.extensions.sql.impl.BeamCalciteSchema;
 import org.apache.beam.sdk.extensions.sql.meta.provider.TableProvider;
 import org.checkerframework.checker.nullness.qual.Nullable;
@@ -32,6 +33,7 @@
  * <p>When {@link #registerTableProvider(String, TableProvider)} is called, the provider should
  * become available for all catalogs.
  */
+@Internal
 public interface CatalogManager {
   /** Creates and stores a catalog of a particular type. */
   void createCatalog(String name, String type, Map<String, String> properties);

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/catalog/CatalogRegistrar.java

@@ -17,11 +17,14 @@
  */
 package org.apache.beam.sdk.extensions.sql.meta.catalog;
 
+import org.apache.beam.sdk.annotations.Internal;
+
 /**
  * Over-arching registrar to capture available {@link Catalog}s. Implementations should be marked
  * with {@link com.google.auto.service.AutoService} to be available to {@link
  * java.util.ServiceLoader}s.
  */
+@Internal
 public interface CatalogRegistrar {
   Iterable<Class<? extends Catalog>> getCatalogs();
 }

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/catalog/InMemoryCatalog.java

@@ -17,17 +17,15 @@
  */
 package org.apache.beam.sdk.extensions.sql.meta.catalog;
 
-import com.google.auto.service.AutoService;
 import java.util.Map;
 import org.apache.beam.sdk.extensions.sql.meta.store.InMemoryMetaStore;
 import org.apache.beam.sdk.extensions.sql.meta.store.MetaStore;
 import org.apache.beam.sdk.util.Preconditions;
 
-@AutoService(Catalog.class)
 public class InMemoryCatalog implements Catalog {
   private final String name;
   private final Map<String, String> properties;
-  private final InMemoryMetaStore metaStore = new InMemoryMetaStore();
+  protected final InMemoryMetaStore metaStore = new InMemoryMetaStore();
 
   public InMemoryCatalog(String name, Map<String, String> properties) {
     this.name = name;
@@ -41,7 +39,8 @@ public String type() {
 
   @Override
   public String name() {
-    return Preconditions.checkStateNotNull(name, "InMemoryCatalog has not been initialized");
+    return Preconditions.checkStateNotNull(
+        name, getClass().getSimpleName() + " has not been initialized");
   }
 
   @Override