[Beam SQL] Change SET CATALOG to USE CATALOG (#35314)

* USE catalog instead of SET

* add missing

Author: ahmedabu98

.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": 3
+  "modification": 2
 }

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

@@ -30,7 +30,7 @@ data: {
         "org.apache.beam.sdk.extensions.sql.impl.parser.SqlCreateFunction"
         "org.apache.beam.sdk.extensions.sql.impl.parser.SqlDropCatalog"
         "org.apache.beam.sdk.extensions.sql.impl.parser.SqlDdlNodes"
-        "org.apache.beam.sdk.extensions.sql.impl.parser.SqlSetCatalog"
+        "org.apache.beam.sdk.extensions.sql.impl.parser.SqlUseCatalog"
         "org.apache.beam.sdk.extensions.sql.impl.parser.SqlSetOptionBeam"
         "org.apache.beam.sdk.extensions.sql.impl.utils.CalciteUtils"
         "org.apache.beam.sdk.schemas.Schema"
@@ -46,6 +46,7 @@ data: {
         "TBLPROPERTIES"
         "PROPERTIES"
         "PARTITIONED"
+        "USE"
       ]
 
       # List of keywords from "keywords" section that are not reserved.
@@ -370,6 +371,8 @@ data: {
         "LOCATION"
         "TBLPROPERTIES"
         "PROPERTIES"
+        "PARTITIONED"
+        "USE"
       ]
 
       # List of non-reserved keywords to add;
@@ -391,7 +394,7 @@ data: {
       # Return type of method implementation should be 'SqlNode'.
       # Example: SqlShowDatabases(), SqlShowTables().
       statementParserMethods: [
-        "SqlSetCatalog(Span.of(), null)"
+        "SqlUseCatalog(Span.of(), null)"
         "SqlSetOptionBeam(Span.of(), null)"
       ]
 

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

@@ -213,14 +213,14 @@ SqlCreate SqlCreateCatalog(Span s, boolean replace) :
 }
 
 /**
- * SET CATALOG catalog_name
+ * USE CATALOG catalog_name
  */
-SqlCall SqlSetCatalog(Span s, String scope) :
+SqlCall SqlUseCatalog(Span s, String scope) :
 {
     final SqlNode catalogName;
 }
 {
-    <SET> {
+    <USE> {
         s.add(this);
     }
     <CATALOG>
@@ -230,7 +230,7 @@ SqlCall SqlSetCatalog(Span s, String scope) :
         catalogName = SimpleIdentifier()
     )
     {
-        return new SqlSetCatalog(
+        return new SqlUseCatalog(
             s.end(this),
             scope,
             catalogName);

…sions/sql/impl/parser/SqlSetCatalog.java …sions/sql/impl/parser/SqlUseCatalog.javasdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/parser/SqlSetCatalog.java renamed to sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/parser/SqlUseCatalog.java sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/parser/SqlSetCatalog.java renamed to sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/parser/SqlUseCatalog.java

@@ -39,13 +39,13 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class SqlSetCatalog extends SqlSetOption implements BeamSqlParser.ExecutableStatement {
-  private static final Logger LOG = LoggerFactory.getLogger(SqlSetCatalog.class);
+public class SqlUseCatalog extends SqlSetOption implements BeamSqlParser.ExecutableStatement {
+  private static final Logger LOG = LoggerFactory.getLogger(SqlUseCatalog.class);
   private final SqlIdentifier catalogName;
 
-  private static final SqlOperator OPERATOR = new SqlSpecialOperator("SET CATALOG", SqlKind.OTHER);
+  private static final SqlOperator OPERATOR = new SqlSpecialOperator("USE CATALOG", SqlKind.OTHER);
 
-  public SqlSetCatalog(SqlParserPos pos, String scope, SqlNode catalogName) {
+  public SqlUseCatalog(SqlParserPos pos, String scope, SqlNode catalogName) {
     super(pos, scope, SqlDdlNodes.getIdentifier(catalogName, pos), null);
     this.catalogName = SqlDdlNodes.getIdentifier(catalogName, pos);
   }
@@ -79,13 +79,13 @@ public void execute(CalcitePrepare.Context context) {
           catalogName.getParserPosition(),
           RESOURCE.internal(
               String.format(
-                  "Unexpected 'SET CATALOG' call for Schema '%s' that is not a Catalog.", name)));
+                  "Unexpected 'USE CATALOG' call for Schema '%s' that is not a Catalog.", name)));
     }
 
     if (catalogManager.getCatalog(name) == null) {
       throw SqlUtil.newContextException(
           catalogName.getParserPosition(),
-          RESOURCE.internal(String.format("Cannot set catalog: '%s' not found.", name)));
+          RESOURCE.internal(String.format("Cannot use catalog: '%s' not found.", name)));
     }
 
     if (catalogManager.currentCatalog().name().equals(name)) {

sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlCliTest.java

@@ -310,8 +310,8 @@ public void testExecute_setCatalog_doesNotExistError() {
     BeamSqlCli cli = new BeamSqlCli().catalogManager(catalogManager);
 
     thrown.expect(CalciteContextException.class);
-    thrown.expectMessage("Cannot set catalog: 'my_catalog' not found.");
-    cli.execute("SET CATALOG my_catalog");
+    thrown.expectMessage("Cannot use catalog: 'my_catalog' not found.");
+    cli.execute("USE CATALOG my_catalog");
   }
 
   @Test
@@ -344,15 +344,15 @@ public void testExecute_setCatalog() {
 
     // catalog manager always starts with a "default" catalog
     assertEquals("default", catalogManager.currentCatalog().name());
-    cli.execute("SET CATALOG catalog_1");
+    cli.execute("USE CATALOG catalog_1");
     assertEquals("catalog_1", catalogManager.currentCatalog().name());
     assertEquals(catalog1Props, catalogManager.currentCatalog().properties());
-    cli.execute("SET CATALOG catalog_2");
+    cli.execute("USE CATALOG catalog_2");
     assertEquals("catalog_2", catalogManager.currentCatalog().name());
     assertEquals(catalog2Props, catalogManager.currentCatalog().properties());
 
     // DEFAULT is a reserved keyword, so need to encapsulate in backticks
-    cli.execute("SET CATALOG 'default'");
+    cli.execute("USE CATALOG 'default'");
     assertEquals("default", catalogManager.currentCatalog().name());
   }
 
@@ -405,15 +405,15 @@ public void testExecute_tableScopeAcrossCatalogs() throws Exception {
     BeamSqlCli cli = new BeamSqlCli().catalogManager(catalogManager);
 
     cli.execute("CREATE CATALOG my_catalog TYPE 'local'");
-    cli.execute("SET CATALOG my_catalog");
+    cli.execute("USE CATALOG my_catalog");
     cli.execute(
         "CREATE EXTERNAL TABLE person (\n" + "id int, name varchar, age int) \n" + "TYPE 'text'");
 
     assertEquals("my_catalog", catalogManager.currentCatalog().name());
     assertNotNull(catalogManager.currentCatalog().metaStore().getTables().get("person"));
 
     cli.execute("CREATE CATALOG my_other_catalog TYPE 'local'");
-    cli.execute("SET CATALOG my_other_catalog");
+    cli.execute("USE CATALOG my_other_catalog");
     assertEquals("my_other_catalog", catalogManager.currentCatalog().name());
     assertNull(catalogManager.currentCatalog().metaStore().getTables().get("person"));
   }

sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/PubsubToBigqueryIT.java

@@ -53,7 +53,7 @@ public class PubsubToBigqueryIT implements Serializable {
   @Test
   public void testSimpleInsert() throws Exception {
     String createCatalog = "CREATE CATALOG my_catalog TYPE `local`";
-    String setCatalog = "SET CATALOG my_catalog";
+    String setCatalog = "USE CATALOG my_catalog";
     String pubsubTableString =
         "CREATE EXTERNAL TABLE pubsub_topic (\n"
             + "event_timestamp TIMESTAMP, \n"

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

@@ -98,7 +98,7 @@ public static void createDataset() throws IOException, InterruptedException {
             + format("  'warehouse' = '%s', \n", warehouse)
             + format("  'gcp_project' = '%s', \n", OPTIONS.getProject())
             + "  'gcp_region' = 'us-central1')";
-    setCatalogDdl = "SET CATALOG my_catalog";
+    setCatalogDdl = "USE CATALOG my_catalog";
   }
 
   private String tableIdentifier;

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

@@ -151,7 +151,7 @@ public void runSqlWriteAndRead(boolean withPartitionFields)
     sqlEnv.executeDdl(createCatalog);
 
     // 2) use the catalog we just created
-    String setCatalog = "SET CATALOG my_catalog";
+    String setCatalog = "USE CATALOG my_catalog";
     sqlEnv.executeDdl(setCatalog);
 
     // 3) create beam table
@@ -277,7 +277,7 @@ public void testSQLReadWithProjectAndFilterPushDown() {
     sqlEnv.executeDdl(createCatalog);
 
     // 2) use the catalog we just created
-    String setCatalog = "SET CATALOG my_catalog";
+    String setCatalog = "USE CATALOG my_catalog";
     sqlEnv.executeDdl(setCatalog);
 
     // 3) create Beam table

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

@@ -84,7 +84,7 @@ public void testBuildBeamSqlTableWithPartitionFields() {
             .build();
 
     sqlEnv.executeDdl("CREATE CATALOG my_catalog TYPE iceberg");
-    sqlEnv.executeDdl("SET CATALOG my_catalog");
+    sqlEnv.executeDdl("USE CATALOG my_catalog");
     sqlEnv.executeDdl(
         "CREATE EXTERNAL TABLE test_partitioned_table(\n"
             + "  id INTEGER,\n"