[SQL] add SHOW command

.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
 }

.github/trigger_files/beam_PreCommit_SQL.json

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

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

@@ -19,12 +19,14 @@
 
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;
 
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
 import org.apache.beam.sdk.extensions.sql.meta.catalog.InMemoryCatalog;
 import org.apache.beam.sdk.io.iceberg.IcebergCatalogConfig;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 public class IcebergCatalog extends InMemoryCatalog {
   // TODO(ahmedabu98): extend this to the IO implementation so
@@ -71,6 +73,11 @@ public boolean createDatabase(String database) {
     return catalogConfig.createNamespace(database);
   }
 
+  @Override
+  public @Nullable Collection<String> databases() {
+    return catalogConfig.listNamespaces();
+  }
+
   @Override
   public void useDatabase(String database) {
     checkArgument(databaseExists(database), "Database '%s' does not exist.");

sdks/java/extensions/sql/src/main/codegen/config.fmpp

@@ -35,6 +35,9 @@ data: {
         "org.apache.beam.sdk.extensions.sql.impl.parser.SqlDdlNodes"
         "org.apache.beam.sdk.extensions.sql.impl.parser.SqlUseCatalog"
         "org.apache.beam.sdk.extensions.sql.impl.parser.SqlUseDatabase"
+        "org.apache.beam.sdk.extensions.sql.impl.parser.SqlShowCatalogs"
+        "org.apache.beam.sdk.extensions.sql.impl.parser.SqlShowDatabases"
+        "org.apache.beam.sdk.extensions.sql.impl.parser.SqlShowTables"
         "org.apache.beam.sdk.extensions.sql.impl.parser.SqlSetOptionBeam"
         "org.apache.beam.sdk.extensions.sql.impl.utils.CalciteUtils"
         "org.apache.beam.sdk.schemas.Schema"
@@ -50,6 +53,9 @@ data: {
         "TBLPROPERTIES"
         "PROPERTIES"
         "PARTITIONED"
+        "CATALOGS"
+        "DATABASES"
+        "TABLES"
         "USE"
       ]
 
@@ -422,6 +428,10 @@ data: {
       # Return type of method implementation should be 'SqlNode'.
       # Example: SqlShowDatabases(), SqlShowTables().
       statementParserMethods: [
+        "SqlShowTables(Span.of(), null)"
+        "SqlShowDatabases(Span.of(), null)"
+        "SqlShowCatalogs(Span.of(), null)"
+        "SqlShowCurrent(Span.of(), null)"
         "SqlUseCatalog(Span.of(), null)"
         "SqlUseDatabase(Span.of(), null)"
         "SqlSetOptionBeam(Span.of(), null)"

sdks/java/extensions/sql/src/main/codegen/includes/parserImpls.ftl

@@ -264,6 +264,22 @@ SqlDrop SqlDropCatalog(Span s, boolean replace) :
     }
 }
 
+/**
+ * SHOW CATALOGS [ LIKE regex_pattern ]
+ */
+SqlCall SqlShowCatalogs(Span s, String scope) :
+{
+    SqlNode regex = null;
+}
+{
+    <SHOW> <CATALOGS> { s.add(this); }
+    [ <LIKE> regex = StringLiteral() ]
+    {
+        return new SqlShowCatalogs(s.end(this), scope, false, regex);
+    }
+}
+
+
 /**
  * CREATE DATABASE ( IF NOT EXISTS )? ( catalog_name '.' )? database_name
  */
@@ -331,6 +347,39 @@ SqlDrop SqlDropDatabase(Span s, boolean replace) :
     }
 }
 
+/**
+ * SHOW DATABASES [ ( FROM | IN )? catalog_name ] [LIKE regex_pattern ]
+ */
+SqlCall SqlShowDatabases(Span s, String scope) :
+{
+    SqlIdentifier catalogName = null;
+    SqlNode regex = null;
+}
+{
+    <SHOW> <DATABASES> { s.add(this); }
+    [ ( <FROM> | <IN> ) catalogName = SimpleIdentifier() ]
+    [ <LIKE> regex = StringLiteral() ]
+    {
+        return new SqlShowDatabases(s.end(this), scope, false, catalogName, regex);
+    }
+}
+
+SqlCall SqlShowCurrent(Span s, String scope) :
+{
+}
+{
+    <SHOW> <CURRENT> { s.add(this); }
+    (
+        <CATALOG> {
+            return new SqlShowCatalogs(s.end(this), scope, true, null);
+        }
+    |
+        <DATABASE> {
+            return new SqlShowDatabases(s.end(this), scope, true, null, null);
+        }
+    )
+}
+
 
 SqlNodeList PartitionFieldList() :
 {
@@ -456,6 +505,24 @@ SqlDrop SqlDropTable(Span s, boolean replace) :
     }
 }
 
+/**
+ * SHOW TABLES [ ( FROM | IN )? [ catalog_name '.' ] database_name ] [ LIKE regex_pattern ]
+ */
+SqlCall SqlShowTables(Span s, String scope) :
+{
+    SqlIdentifier database = null;
+    SqlNode regex = null;
+}
+{
+    <SHOW> <TABLES> { s.add(this); }
+    [ (<FROM> | <IN>) database = CompoundIdentifier() ]
+    [ <LIKE> regex = StringLiteral() ]
+    {
+        return new SqlShowTables(s.end(this), scope, database, regex);
+    }
+}
+
+
 Schema.FieldType FieldType() :
 {
     final SqlTypeName collectionTypeName;

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

@@ -126,6 +126,10 @@ public org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.schema.Table ge
         connection.getPipelineOptions());
   }
 
+  public Collection<Table> getTables() {
+    return tableProvider.getTables().values();
+  }
+
   @Override
   public Set<String> getFunctionNames() {
     return Collections.emptySet();

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

@@ -17,6 +17,7 @@
  */
 package org.apache.beam.sdk.extensions.sql.impl;
 
+import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull;
 import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
 import static org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.util.Static.RESOURCE;
 
@@ -142,7 +143,8 @@ public void dropCatalog(SqlIdentifier identifier, boolean ifExists) {
   // will attempt to do so.
   public void maybeRegisterProvider(TableName path, String type) {
     type = type.toLowerCase();
-    CatalogSchema catalogSchema = getCatalogSchema(path);
+    CatalogSchema catalogSchema =
+        path.catalog() != null ? getCatalogSchema(path) : getCurrentCatalogSchema();
     BeamCalciteSchema beamCalciteSchema = catalogSchema.getDatabaseSchema(path);
 
     if (beamCalciteSchema.getTableProvider() instanceof MetaStore) {
@@ -177,14 +179,16 @@ public Set<String> getTableNames() {
   }
 
   public CatalogSchema getCatalogSchema(TableName tablePath) {
-    @Nullable Schema catalogSchema = getSubSchema(tablePath.catalog());
-    if (catalogSchema == null) {
-      catalogSchema = getCurrentCatalogSchema();
-    }
+    return getCatalogSchema(tablePath.catalog());
+  }
+
+  public CatalogSchema getCatalogSchema(@Nullable String catalog) {
+    Schema catalogSchema =
+        checkArgumentNotNull(getSubSchema(catalog), "Catalog '%s' not found.", catalog);
     Preconditions.checkState(
         catalogSchema instanceof CatalogSchema,
         "Unexpected Schema type for Catalog '%s': %s",
-        tablePath.catalog(),
+        catalog,
         catalogSchema.getClass());
     return (CatalogSchema) catalogSchema;
   }
@@ -222,7 +226,11 @@ public CatalogSchema getCurrentCatalogSchema() {
 
   @Override
   public Set<String> getSubSchemaNames() {
-    return catalogManager.catalogs().stream().map(Catalog::name).collect(Collectors.toSet());
+    return catalogs().stream().map(Catalog::name).collect(Collectors.toSet());
+  }
+
+  public Collection<Catalog> catalogs() {
+    return catalogManager.catalogs();
   }
 
   public void setPipelineOption(String key, String value) {

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

@@ -72,6 +72,10 @@ public Catalog getCatalog() {
     return getSubSchema(catalog.currentDatabase());
   }
 
+  public @Nullable Collection<String> databases() {
+    return catalog.databases();
+  }
+
   public BeamCalciteSchema getDatabaseSchema(TableName tablePath) {
     @Nullable BeamCalciteSchema beamCalciteSchema = getSubSchema(tablePath.database());
     if (beamCalciteSchema == null) {

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

@@ -159,7 +159,10 @@ public void execute(CalcitePrepare.Context context) {
       CatalogManagerSchema catalogManagerSchema = (CatalogManagerSchema) schema;
       catalogManagerSchema.maybeRegisterProvider(pathOverride, SqlDdlNodes.getString(type));
 
-      CatalogSchema catalogSchema = catalogManagerSchema.getCatalogSchema(pathOverride);
+      CatalogSchema catalogSchema =
+          pathOverride.catalog() != null
+              ? catalogManagerSchema.getCatalogSchema(pathOverride)
+              : catalogManagerSchema.getCurrentCatalogSchema();
       beamCalciteSchema = catalogSchema.getDatabaseSchema(pathOverride);
     } else if (schema instanceof BeamCalciteSchema) {
       beamCalciteSchema = (BeamCalciteSchema) schema;

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

@@ -66,6 +66,13 @@ static Pair<CalciteSchema, String> schema(
     return Pair.of(checkStateNotNull(schema, "Got null sub-schema for path '%s'", path), name(id));
   }
 
+  static CalciteSchema schema(CalcitePrepare.Context context, boolean mutable) {
+    CalciteSchema rootSchema = mutable ? context.getMutableRootSchema() : context.getRootSchema();
+    List<String> path = context.getDefaultSchemaPath();
+    @Nullable CalciteSchema schema = childSchema(rootSchema, path);
+    return checkStateNotNull(schema, "Got null sub-schema for path '%s'", path);
+  }
+
   private static @Nullable CalciteSchema childSchema(CalciteSchema rootSchema, List<String> path) {
     @Nullable CalciteSchema schema = rootSchema;
     for (String p : path) {
@@ -85,7 +92,7 @@ public static String name(SqlIdentifier id) {
     }
   }
 
-  static @Nullable String getString(SqlNode n) {
+  static @Nullable String getString(@Nullable SqlNode n) {
     if (n == null) {
       return null;
     }

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

@@ -84,7 +84,10 @@ public void execute(CalcitePrepare.Context context) {
 
     List<String> components = Lists.newArrayList(Splitter.on(".").split(databaseName.toString()));
     TableName pathOverride = TableName.create(components, "");
-    CatalogSchema catalogSchema = ((CatalogManagerSchema) schema).getCatalogSchema(pathOverride);
+    CatalogSchema catalogSchema =
+        pathOverride.catalog() != null
+            ? ((CatalogManagerSchema) schema).getCatalogSchema(pathOverride)
+            : ((CatalogManagerSchema) schema).getCurrentCatalogSchema();
     catalogSchema.dropDatabase(databaseName, cascade, ifExists);
   }
 

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

@@ -53,7 +53,10 @@ public void execute(CalcitePrepare.Context context) {
 
     BeamCalciteSchema beamCalciteSchema;
     if (schema instanceof CatalogManagerSchema) {
-      CatalogSchema catalogSchema = ((CatalogManagerSchema) schema).getCatalogSchema(pathOverride);
+      CatalogSchema catalogSchema =
+          pathOverride.catalog() != null
+              ? ((CatalogManagerSchema) schema).getCatalogSchema(pathOverride)
+              : ((CatalogManagerSchema) schema).getCurrentCatalogSchema();
       beamCalciteSchema = catalogSchema.getDatabaseSchema(pathOverride);
     } else if (schema instanceof BeamCalciteSchema) {
       beamCalciteSchema = (BeamCalciteSchema) schema;