diff --git a/.github/trigger_files/beam_PostCommit_SQL.json b/.github/trigger_files/beam_PostCommit_SQL.json index c537844dc84a..b26833333238 100644 --- a/.github/trigger_files/beam_PostCommit_SQL.json +++ b/.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 } diff --git a/sdks/java/extensions/sql/src/main/codegen/config.fmpp b/sdks/java/extensions/sql/src/main/codegen/config.fmpp index a51d0125a3f1..623a3e2792f7 100644 --- a/sdks/java/extensions/sql/src/main/codegen/config.fmpp +++ b/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)" ] diff --git a/sdks/java/extensions/sql/src/main/codegen/includes/parserImpls.ftl b/sdks/java/extensions/sql/src/main/codegen/includes/parserImpls.ftl index 78940ee69380..450c6eeaff7f 100644 --- a/sdks/java/extensions/sql/src/main/codegen/includes/parserImpls.ftl +++ b/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; } { - { + { s.add(this); } @@ -230,7 +230,7 @@ SqlCall SqlSetCatalog(Span s, String scope) : catalogName = SimpleIdentifier() ) { - return new SqlSetCatalog( + return new SqlUseCatalog( s.end(this), scope, catalogName); diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/parser/SqlSetCatalog.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/parser/SqlUseCatalog.java similarity index 91% rename from sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/parser/SqlSetCatalog.java rename to sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/parser/SqlUseCatalog.java index 208b389d3c3a..7088c7183027 100644 --- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/parser/SqlSetCatalog.java +++ b/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)) { diff --git a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlCliTest.java b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlCliTest.java index dcbb0cfec09a..b8aa030649ab 100644 --- a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlCliTest.java +++ b/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,7 +405,7 @@ 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'"); @@ -413,7 +413,7 @@ public void testExecute_tableScopeAcrossCatalogs() throws Exception { 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")); } diff --git a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/PubsubToBigqueryIT.java b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/PubsubToBigqueryIT.java index c8d843bda9ee..54be98adbc14 100644 --- a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/PubsubToBigqueryIT.java +++ b/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" diff --git a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/PubsubToIcebergIT.java b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/PubsubToIcebergIT.java index c757843b5f1f..c9680ca0adc1 100644 --- a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/PubsubToIcebergIT.java +++ b/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; diff --git a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/iceberg/IcebergReadWriteIT.java b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/iceberg/IcebergReadWriteIT.java index bd1bbcef5576..c885d1bac144 100644 --- a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/iceberg/IcebergReadWriteIT.java +++ b/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 diff --git a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/iceberg/IcebergTableProviderTest.java b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/iceberg/IcebergTableProviderTest.java index 6cde6a626c90..0a704bcb5e29 100644 --- a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/iceberg/IcebergTableProviderTest.java +++ b/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"