This is an automated email from the ASF dual-hosted git repository.
Abacn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 0c370589685 Remove Beam SQL Data Catalog provider (#40663)
0c370589685 is described below
commit 0c370589685d55c637e57fe854d9b0135358eebd
Author: Abdelrahman Ibrahim <[email protected]>
AuthorDate: Tue Oct 20 02:36:14 2026 +0300
Remove Beam SQL Data Catalog provider (#40663)
---
CHANGES.md | 1 +
build.gradle.kts | 2 -
.../org/apache/beam/gradle/BeamModulePlugin.groovy | 2 -
.../assets/symbols/java.g.yaml | 19 --
.../resources/beam/checkstyle/suppressions.xml | 1 -
sdks/java/extensions/sql/datacatalog/build.gradle | 107 ----------
.../sql/example/BeamSqlDataCatalogExample.java | 107 ----------
.../sdk/extensions/sql/example/package-info.java | 20 --
.../provider/datacatalog/BigQueryTableFactory.java | 72 -------
.../provider/datacatalog/ChainedTableFactory.java | 50 -----
.../datacatalog/DataCatalogPipelineOptions.java | 43 ----
.../DataCatalogPipelineOptionsRegistrar.java | 31 ---
.../datacatalog/DataCatalogTableProvider.java | 235 ---------------------
.../meta/provider/datacatalog/GcsTableFactory.java | 61 ------
.../provider/datacatalog/PubsubTableFactory.java | 62 ------
.../sql/meta/provider/datacatalog/SchemaUtils.java | 190 -----------------
.../meta/provider/datacatalog/TableFactory.java | 37 ----
.../meta/provider/datacatalog/ZetaSqlIdUtils.java | 83 --------
.../meta/provider/datacatalog/package-info.java | 20 --
.../datacatalog/DataCatalogBigQueryIT.java | 115 ----------
.../provider/datacatalog/DataCatalogGCSIT.java | 83 --------
.../meta/provider/datacatalog/SchemaUtilsTest.java | 157 --------------
.../provider/datacatalog/ZetaSqlIdUtilsTest.java | 64 ------
settings.gradle.kts | 1 -
24 files changed, 1 insertion(+), 1562 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 00fb9aa825b..d06940d12cf 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -85,6 +85,7 @@
## Breaking Changes
+* (Java) Removed Beam SQL Google Cloud Data Catalog table provider
(`beam-sdks-java-extensions-sql-datacatalog`) because GCP Data Catalog has been
decommissioned ([#40663](https://github.com/apache/beam/issues/40663)).
* (Go) The row coder now encodes `int16` and `uint16` struct fields as 2 byte
big endian INT16 values, matching the Java and Python SDKs. This is an update
incompatible change for streaming pipelines that use rows with `int16` or
`uint16` fields ([#40151](https://github.com/apache/beam/issues/40151)).
* (Java) JmsIO migrated to `jakarta.jms` (JMS 3.1) and ActiveMQ 6.2.5. User
code implementing `JmsIO.MessageMapper`,
`valueMapper`, `topicNameMapper`, or providing a `ConnectionFactory` must
update imports from `javax.jms.*` to
diff --git a/build.gradle.kts b/build.gradle.kts
index 8b68b273796..1cefe1ac5b9 100644
--- a/build.gradle.kts
+++ b/build.gradle.kts
@@ -393,7 +393,6 @@ tasks.register("flinkPreCommit") {
tasks.register("sqlPreCommit") {
dependsOn(":sdks:java:extensions:sql:preCommit")
dependsOn(":sdks:java:extensions:sql:buildDependents")
- dependsOn(":sdks:java:extensions:sql:datacatalog:build")
dependsOn(":sdks:java:extensions:sql:expansion-service:build")
dependsOn(":sdks:java:extensions:sql:hcatalog:build")
dependsOn(":sdks:java:extensions:sql:iceberg:build")
@@ -442,7 +441,6 @@ tasks.register("javaAvroVersionsTest") {
tasks.register("sqlPostCommit") {
dependsOn(":sdks:java:extensions:sql:postCommit")
dependsOn(":sdks:java:extensions:sql:jdbc:postCommit")
- dependsOn(":sdks:java:extensions:sql:datacatalog:postCommit")
dependsOn(":sdks:java:extensions:sql:iceberg:integrationTest")
dependsOn(":sdks:java:extensions:sql:hadoopVersionsTest")
}
diff --git
a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
index cf5979dd176..17fc16e9967 100644
--- a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
+++ b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
@@ -770,7 +770,6 @@ class BeamModulePlugin implements Plugin<Project> {
google_cloud_core :
"com.google.cloud:google-cloud-core", // google_cloud_platform_libraries_bom
sets version
google_cloud_core_grpc :
"com.google.cloud:google-cloud-core-grpc", //
google_cloud_platform_libraries_bom sets version
google_cloud_core_http :
"com.google.cloud:google-cloud-core-http", //
google_cloud_platform_libraries_bom sets version
- google_cloud_datacatalog_v1beta1 :
"com.google.cloud:google-cloud-datacatalog", //
google_cloud_platform_libraries_bom sets version
google_cloud_dataflow_java_proto_library_all:
"com.google.cloud.dataflow:google-cloud-dataflow-java-proto-library-all:0.5.160304",
google_cloud_datastore_v1_proto_client :
"com.google.cloud.datastore:datastore-v1-proto-client:3.4.0", //
[bomupgrader] sets version
google_cloud_firestore :
"com.google.cloud:google-cloud-firestore", //
google_cloud_platform_libraries_bom sets version
@@ -893,7 +892,6 @@ class BeamModulePlugin implements Plugin<Project> {
proto_google_cloud_bigquery_storage_v1 :
"com.google.api.grpc:proto-google-cloud-bigquerystorage-v1", //
google_cloud_platform_libraries_bom sets version
proto_google_cloud_bigtable_admin_v2 :
"com.google.api.grpc:proto-google-cloud-bigtable-admin-v2", //
google_cloud_platform_libraries_bom sets version
proto_google_cloud_bigtable_v2 :
"com.google.api.grpc:proto-google-cloud-bigtable-v2", //
google_cloud_platform_libraries_bom sets version
- proto_google_cloud_datacatalog_v1beta1 :
"com.google.api.grpc:proto-google-cloud-datacatalog-v1beta1", //
google_cloud_platform_libraries_bom sets version
proto_google_cloud_datastore_v1 :
"com.google.api.grpc:proto-google-cloud-datastore-v1", //
google_cloud_platform_libraries_bom sets version
proto_google_cloud_firestore_v1 :
"com.google.api.grpc:proto-google-cloud-firestore-v1", //
google_cloud_platform_libraries_bom sets version
proto_google_cloud_kms_v1 :
"com.google.api.grpc:proto-google-cloud-kms-v1", //
google_cloud_platform_libraries_bom sets version
diff --git
a/playground/frontend/playground_components/assets/symbols/java.g.yaml
b/playground/frontend/playground_components/assets/symbols/java.g.yaml
index e0f2b1d15d1..2619d8702f1 100644
--- a/playground/frontend/playground_components/assets/symbols/java.g.yaml
+++ b/playground/frontend/playground_components/assets/symbols/java.g.yaml
@@ -1292,9 +1292,6 @@ BeamSqlCli:
- explainQuery
- getMetaStore
- metaStore
-BeamSqlDataCatalogExample:
- methods:
- - main
BeamSqlEnv:
methods:
- addSchema
@@ -3424,22 +3421,6 @@ DaoFactory:
- getChangeStreamDao
- getPartitionMetadataAdminDao
- getPartitionMetadataDao
-DataCatalogPipelineOptionsRegistrar:
- methods:
- - getPipelineOptions
-DataCatalogTableProvider:
- methods:
- - buildBeamSqlTable
- - close
- - create
- - createDataCatalogClient
- - createTable
- - dropTable
- - getTable
- - getTableByFullName
- - getTableType
- - getTables
- - setSchemaIfNotPresent
DataChangeRecord:
methods:
- equals
diff --git
a/sdks/java/build-tools/src/main/resources/beam/checkstyle/suppressions.xml
b/sdks/java/build-tools/src/main/resources/beam/checkstyle/suppressions.xml
index 833fd30cf21..d3ec39dc6ac 100644
--- a/sdks/java/build-tools/src/main/resources/beam/checkstyle/suppressions.xml
+++ b/sdks/java/build-tools/src/main/resources/beam/checkstyle/suppressions.xml
@@ -70,7 +70,6 @@
<suppress id="ForbidNonVendoredGrpcProtobuf"
files=".*google.*cloud.*spanner.*FakePartitionFactory\.java" />
<suppress id="ForbidNonVendoredGrpcProtobuf"
files=".*extensions.*sql.*datastore.*" />
<suppress id="ForbidNonVendoredGrpcProtobuf"
files=".*extensions.*sql.*ProtoPayloadSerializerProvider.*" />
- <suppress id="ForbidNonVendoredGrpcProtobuf"
files=".*datacatalog.*DataCatalogTableProvider\.java" />
<suppress id="ForbidNonVendoredGrpcProtobuf"
files=".*examples.*datatokenization.*BigTableIO\.java" />
<suppress id="ForbidNonVendoredGrpcProtobuf"
files=".*aws2.*kinesis.*RecordsAggregator\.java" />
<suppress id="ForbidNonVendoredGrpcProtobuf"
files=".*testinfra.*pipelines.*" />
diff --git a/sdks/java/extensions/sql/datacatalog/build.gradle
b/sdks/java/extensions/sql/datacatalog/build.gradle
deleted file mode 100644
index db4a9200150..00000000000
--- a/sdks/java/extensions/sql/datacatalog/build.gradle
+++ /dev/null
@@ -1,107 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * License); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an AS IS BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-import groovy.json.JsonOutput
-
-plugins { id 'org.apache.beam.module' }
-
-applyJavaNature(
- automaticModuleName: 'org.apache.beam.sdk.extensions.sql.datacatalog')
-
-dependencies {
- implementation
enforcedPlatform(library.java.google_cloud_platform_libraries_bom)
- implementation(library.java.google_cloud_datacatalog_v1beta1) {
- exclude group: 'io.grpc', module: 'grpc-core' // Use Beam's version
- }
- implementation library.java.gax
- implementation library.java.google_auth_library_credentials
- implementation library.java.proto_google_cloud_datacatalog_v1beta1
- implementation library.java.protobuf_java
- implementation library.java.slf4j_api
- implementation library.java.vendored_guava_32_1_2_jre
- implementation library.java.jackson_databind
- implementation project(path: ":sdks:java:core", configuration: "shadow")
- implementation "org.threeten:threetenbp:1.4.5"
- provided project(":sdks:java:extensions:sql")
-
- // Dependencies for the example
- implementation project(":sdks:java:extensions:google-cloud-platform-core")
- provided project(":sdks:java:io:google-cloud-platform")
- permitUnusedDeclared project(":sdks:java:io:google-cloud-platform")
- implementation library.java.slf4j_api
-
- testImplementation project(":sdks:java:extensions:sql")
- testImplementation project(":runners:direct-java")
- testImplementation project(":sdks:java:io:google-cloud-platform")
- testImplementation library.java.google_api_services_bigquery
- testImplementation library.java.junit
- testRuntimeOnly library.java.slf4j_simple
-}
-
-task runDataCatalogExample(type: JavaExec) {
- description = "Run SQL example of how to use Data Catalog table provider"
- mainClass =
"org.apache.beam.sdk.extensions.sql.example.BeamSqlDataCatalogExample"
- classpath = sourceSets.main.runtimeClasspath
-
- def runner = project.findProperty('runner') ?: 'DirectRunner'
- def queryString = project.findProperty('queryString') ?: ''
- def outputFilePrefix = project.findProperty('outputFilePrefix') ?: ''
- def gcpProject = project.findProperty('gcpProject') ?: 'apache-beam-testing'
- def gcpRegion = project.findProperty('gcpRegion') ?: 'us-central1'
- def gcsTempRoot = project.findProperty('gcsTempRoot') ?:
'gs://temp-storage-for-end-to-end-tests/'
-
- args = [
- "--runner=${runner}",
- "--queryString=${queryString}",
- "--outputFilePrefix=${outputFilePrefix}",
- "--project=${gcpProject}",
- "--region=${gcpRegion}",
- "--tempLocation=${gcsTempRoot}",
- ]
-}
-
-task integrationTest(type: Test) {
- group = "Verification"
- def gcpProject = project.findProperty('gcpProject') ?: 'apache-beam-testing'
- def gcsTempRoot = project.findProperty('gcsTempRoot') ?:
'gs://temp-storage-for-end-to-end-tests/'
-
- // Disable Gradle cache (it should not be used because the IT's won't run).
- outputs.upToDateWhen { false }
-
- def pipelineOptions = [
- "--project=${gcpProject}",
- "--tempLocation=${gcsTempRoot}",
- "--blockOnRun=false"]
-
- systemProperty "beamTestPipelineOptions", JsonOutput.toJson(pipelineOptions)
-
- include '**/*IT.class'
- maxParallelForks 4
- classpath = project(":sdks:java:extensions:sql:datacatalog")
- .sourceSets
- .test
- .runtimeClasspath
- testClassesDirs =
files(project(":sdks:java:extensions:sql:datacatalog").sourceSets.test.output.classesDirs)
- useJUnit {}
-}
-
-task postCommit {
- group = "Verification"
- description = "Various integration tests"
- dependsOn integrationTest
-}
diff --git
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/example/BeamSqlDataCatalogExample.java
b/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/example/BeamSqlDataCatalogExample.java
deleted file mode 100644
index 0b2f1b04cbb..00000000000
---
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/example/BeamSqlDataCatalogExample.java
+++ /dev/null
@@ -1,107 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.beam.sdk.extensions.sql.example;
-
-import java.util.Arrays;
-import java.util.stream.Collectors;
-import org.apache.beam.sdk.Pipeline;
-import org.apache.beam.sdk.extensions.sql.SqlTransform;
-import
org.apache.beam.sdk.extensions.sql.meta.provider.datacatalog.DataCatalogPipelineOptions;
-import
org.apache.beam.sdk.extensions.sql.meta.provider.datacatalog.DataCatalogTableProvider;
-import org.apache.beam.sdk.io.TextIO;
-import org.apache.beam.sdk.options.Description;
-import org.apache.beam.sdk.options.PipelineOptions;
-import org.apache.beam.sdk.options.PipelineOptionsFactory;
-import org.apache.beam.sdk.options.Validation;
-import org.apache.beam.sdk.transforms.MapElements;
-import org.apache.beam.sdk.values.Row;
-import org.apache.beam.sdk.values.TypeDescriptor;
-import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Strings;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/** Example pipeline that uses Google Cloud Data Catalog to retrieve the table
metadata. */
-public class BeamSqlDataCatalogExample {
-
- private static final Logger LOG =
LoggerFactory.getLogger(BeamSqlDataCatalogExample.class);
-
- /** Pipeline options to specify the query and the output for the example. */
- public interface DCExamplePipelineOptions extends PipelineOptions {
-
- /** SQL Query. */
- @Description("Required. SQL Query containing the pipeline logic.")
- @Validation.Required
- String getQueryString();
-
- void setQueryString(String queryString);
-
- /** Output file prefix. */
- @Description("Required. Output file prefix.")
- @Validation.Required
- String getOutputFilePrefix();
-
- void setOutputFilePrefix(String outputPathPrefix);
- }
-
- public static void main(String[] args) throws Exception {
- LOG.info("Args: {}", Arrays.asList(args));
- DCExamplePipelineOptions options =
-
PipelineOptionsFactory.fromArgs(args).as(DCExamplePipelineOptions.class);
- LOG.info("Query: {}\nOutput: {}", options.getQueryString(),
options.getOutputFilePrefix());
-
- Pipeline pipeline = Pipeline.create(options);
-
- validateArgs(options);
-
- try (DataCatalogTableProvider tableProvider =
-
DataCatalogTableProvider.create(options.as(DataCatalogPipelineOptions.class))) {
- pipeline
- .apply(
- "SQL Query",
- SqlTransform.query(options.getQueryString())
- .withDefaultTableProvider("datacatalog", tableProvider))
- .apply("Convert to Strings", rowsToStrings())
- .apply("Write output",
TextIO.write().to(options.getOutputFilePrefix()));
-
- pipeline.run().waitUntilFinish();
- }
- }
-
- private static MapElements<Row, String> rowsToStrings() {
- return MapElements.into(TypeDescriptor.of(String.class))
- .via(
- row ->
row.getValues().stream().map(String::valueOf).collect(Collectors.joining(",
")));
- }
-
- private static void validateArgs(DCExamplePipelineOptions options) {
- if (Strings.isNullOrEmpty(options.getQueryString())
- || Strings.isNullOrEmpty(options.getOutputFilePrefix())) {
- String usage =
- "ERROR: SQL query or output file is not specified."
- + "To run this example:\n"
- + "./gradlew "
- + ":sdks:java:extensions:sql:datacatalog:runDataCatalogExample "
- + "-PgcpProject=<project> "
- + "-PgcsTempRoot=<GCS temp location> "
- + "-PqueryString=<query> "
- + "-PoutputFilePrefix=<output location> "
- + "-PtempLocation=<temp GCS location for BQ export>\n\n";
- throw new IllegalArgumentException(usage);
- }
- }
-}
diff --git
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/example/package-info.java
b/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/example/package-info.java
deleted file mode 100644
index 7d6ff31c68a..00000000000
---
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/example/package-info.java
+++ /dev/null
@@ -1,20 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-/** Example how to use Data Catalog table provider. */
-package org.apache.beam.sdk.extensions.sql.example;
diff --git
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/BigQueryTableFactory.java
b/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/BigQueryTableFactory.java
deleted file mode 100644
index 172dad752d9..00000000000
---
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/BigQueryTableFactory.java
+++ /dev/null
@@ -1,72 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.beam.sdk.extensions.sql.meta.provider.datacatalog;
-
-import com.google.cloud.datacatalog.v1beta1.Entry;
-import java.net.URI;
-import java.util.Optional;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-import org.apache.beam.sdk.extensions.sql.TableUtils;
-import org.apache.beam.sdk.extensions.sql.meta.Table;
-
-/** {@link TableFactory} that understands Data Catalog BigQuery entries. */
-class BigQueryTableFactory implements TableFactory {
- private static final String BIGQUERY_API = "bigquery.googleapis.com";
-
- private static final Pattern BQ_PATH_PATTERN =
- Pattern.compile(
-
"/projects/(?<PROJECT>[^/]+)/datasets/(?<DATASET>[^/]+)/tables/(?<TABLE>[^/]+)");
-
- private final boolean truncateTimestamps;
-
- public BigQueryTableFactory(boolean truncateTimestamps) {
- this.truncateTimestamps = truncateTimestamps;
- }
-
- @Override
- public Optional<Table.Builder> tableBuilder(Entry entry) {
- if
(!URI.create(entry.getLinkedResource()).getAuthority().equalsIgnoreCase(BIGQUERY_API))
{
- return Optional.empty();
- }
-
- return Optional.of(
- Table.builder()
- .location(getLocation(entry))
- .properties(TableUtils.emptyProperties().put("truncateTimestamps",
truncateTimestamps))
- .type("bigquery")
- .comment(""));
- }
-
- private static String getLocation(Entry entry) {
- URI entryName = URI.create(entry.getLinkedResource());
- String bqPath = entryName.getPath();
-
- Matcher bqPathMatcher = BQ_PATH_PATTERN.matcher(bqPath);
- if (!bqPathMatcher.matches()) {
- throw new IllegalArgumentException(
- "Unsupported format for BigQuery table path: '" +
entry.getLinkedResource() + "'");
- }
-
- String project = bqPathMatcher.group("PROJECT");
- String dataset = bqPathMatcher.group("DATASET");
- String table = bqPathMatcher.group("TABLE");
-
- return String.format("%s:%s.%s", project, dataset, table);
- }
-}
diff --git
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/ChainedTableFactory.java
b/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/ChainedTableFactory.java
deleted file mode 100644
index a68dafe63f4..00000000000
---
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/ChainedTableFactory.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.beam.sdk.extensions.sql.meta.provider.datacatalog;
-
-import com.google.cloud.datacatalog.v1beta1.Entry;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Optional;
-import org.apache.beam.sdk.extensions.sql.meta.Table;
-
-/** {@link TableFactory} that uses the first applicable sub-{@link
TableFactory}. */
-class ChainedTableFactory implements TableFactory {
-
- private final List<TableFactory> subTableFactories;
-
- public static ChainedTableFactory of(TableFactory... subTableFactories) {
- return new ChainedTableFactory(Arrays.asList(subTableFactories));
- }
-
- private ChainedTableFactory(List<TableFactory> subTableFactories) {
- this.subTableFactories = subTableFactories;
- }
-
- /** Creates a Beam SQL table description from a GCS fileset entry. */
- @Override
- public Optional<Table.Builder> tableBuilder(Entry entry) {
- for (TableFactory tableFactory : subTableFactories) {
- Optional<Table.Builder> tableBuilder = tableFactory.tableBuilder(entry);
- if (tableBuilder.isPresent()) {
- return tableBuilder;
- }
- }
- return Optional.empty();
- }
-}
diff --git
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/DataCatalogPipelineOptions.java
b/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/DataCatalogPipelineOptions.java
deleted file mode 100644
index bb4afee3a21..00000000000
---
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/DataCatalogPipelineOptions.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.beam.sdk.extensions.sql.meta.provider.datacatalog;
-
-import org.apache.beam.sdk.options.Default;
-import org.apache.beam.sdk.options.Description;
-import org.apache.beam.sdk.options.PipelineOptions;
-import org.apache.beam.sdk.options.Validation;
-
-/** Pipeline options for Data Catalog table provider. */
-public interface DataCatalogPipelineOptions extends PipelineOptions {
-
- /** DataCatalog endpoint. */
- @Description("Data catalog endpoint.")
- @Validation.Required
- @Default.String("datacatalog.googleapis.com:443")
- String getDataCatalogEndpoint();
-
- void setDataCatalogEndpoint(String dataCatalogEndpoint);
-
- /** Whether to truncate timestamps in tables described by Data Catalog. */
- @Description("Truncate sub-millisecond precision timestamps in tables
described by Data Catalog")
- @Validation.Required
- @Default.Boolean(false)
- boolean getTruncateTimestamps();
-
- void setTruncateTimestamps(boolean newValue);
-}
diff --git
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/DataCatalogPipelineOptionsRegistrar.java
b/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/DataCatalogPipelineOptionsRegistrar.java
deleted file mode 100644
index 7f055dd245a..00000000000
---
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/DataCatalogPipelineOptionsRegistrar.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.beam.sdk.extensions.sql.meta.provider.datacatalog;
-
-import com.google.auto.service.AutoService;
-import org.apache.beam.sdk.options.PipelineOptions;
-import org.apache.beam.sdk.options.PipelineOptionsRegistrar;
-import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
-
-@AutoService(PipelineOptionsRegistrar.class)
-public class DataCatalogPipelineOptionsRegistrar implements
PipelineOptionsRegistrar {
- @Override
- public Iterable<Class<? extends PipelineOptions>> getPipelineOptions() {
- return ImmutableList.of(DataCatalogPipelineOptions.class);
- }
-}
diff --git
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/DataCatalogTableProvider.java
b/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/DataCatalogTableProvider.java
deleted file mode 100644
index 19cc37b3f0d..00000000000
---
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/DataCatalogTableProvider.java
+++ /dev/null
@@ -1,235 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.beam.sdk.extensions.sql.meta.provider.datacatalog;
-
-import static java.util.stream.Collectors.toMap;
-
-import com.google.api.gax.rpc.InvalidArgumentException;
-import com.google.api.gax.rpc.NotFoundException;
-import com.google.api.gax.rpc.PermissionDeniedException;
-import com.google.api.gax.rpc.StatusCode.Code;
-import com.google.cloud.datacatalog.v1beta1.DataCatalogClient;
-import com.google.cloud.datacatalog.v1beta1.DataCatalogSettings;
-import com.google.cloud.datacatalog.v1beta1.Entry;
-import com.google.cloud.datacatalog.v1beta1.LookupEntryRequest;
-import com.google.cloud.datacatalog.v1beta1.UpdateEntryRequest;
-import com.google.protobuf.FieldMask;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Optional;
-import java.util.stream.Stream;
-import org.apache.beam.sdk.annotations.Internal;
-import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
-import org.apache.beam.sdk.extensions.sql.impl.TableName;
-import org.apache.beam.sdk.extensions.sql.meta.BeamSqlTable;
-import org.apache.beam.sdk.extensions.sql.meta.Table;
-import org.apache.beam.sdk.extensions.sql.meta.provider.FullNameTableProvider;
-import org.apache.beam.sdk.extensions.sql.meta.provider.InvalidTableException;
-import org.apache.beam.sdk.extensions.sql.meta.provider.TableProvider;
-import
org.apache.beam.sdk.extensions.sql.meta.provider.bigquery.BigQueryTableProvider;
-import
org.apache.beam.sdk.extensions.sql.meta.provider.pubsub.PubsubTableProvider;
-import org.apache.beam.sdk.extensions.sql.meta.provider.text.TextTableProvider;
-import org.apache.beam.sdk.schemas.Schema;
-import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects;
-import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
-import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet;
-import org.checkerframework.checker.nullness.qual.Nullable;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.threeten.bp.Duration;
-
-/** Uses DataCatalog to get the source type and schema for a table. */
-public class DataCatalogTableProvider extends FullNameTableProvider implements
AutoCloseable {
-
- private static final Logger LOG =
LoggerFactory.getLogger(DataCatalogTableProvider.class);
-
- private static final TableFactory PUBSUB_TABLE_FACTORY = new
PubsubTableFactory();
- private static final TableFactory GCS_TABLE_FACTORY = new GcsTableFactory();
-
- private static final Map<String, TableProvider> DELEGATE_PROVIDERS =
- Stream.of(new PubsubTableProvider(), new BigQueryTableProvider(), new
TextTableProvider())
- .collect(toMap(TableProvider::getTableType, p -> p));
-
- private final DataCatalogClient dataCatalog;
- private final Map<String, Table> tableCache;
- private final TableFactory tableFactory;
-
- private DataCatalogTableProvider(DataCatalogClient dataCatalog, boolean
truncateTimestamps) {
- this.tableCache = new HashMap<>();
- this.dataCatalog = dataCatalog;
- this.tableFactory =
- ChainedTableFactory.of(
- PUBSUB_TABLE_FACTORY, GCS_TABLE_FACTORY, new
BigQueryTableFactory(truncateTimestamps));
- }
-
- public static DataCatalogTableProvider create(DataCatalogPipelineOptions
options) {
- return new DataCatalogTableProvider(
- createDataCatalogClient(options), options.getTruncateTimestamps());
- }
-
- @Override
- public String getTableType() {
- return "google.cloud.datacatalog";
- }
-
- @Override
- public void createTable(Table table) {
- throw new UnsupportedOperationException(
- "Creating tables is not supported with DataCatalog table provider.");
- }
-
- @Override
- public void dropTable(String tableName) {
- throw new UnsupportedOperationException(
- "Dropping tables is not supported with DataCatalog table provider");
- }
-
- @Override
- public Map<String, Table> getTables() {
- throw new UnsupportedOperationException("Loading all tables from
DataCatalog is not supported");
- }
-
- @Override
- public @Nullable Table getTable(String tableName) {
- return loadTable(tableName);
- }
-
- @Override
- public Table getTableByFullName(TableName fullTableName) {
-
- ImmutableList<String> allNameParts =
- ImmutableList.<String>builder()
- .addAll(fullTableName.getPath())
- .add(fullTableName.getTableName())
- .build();
-
- String fullEscapedTableName = ZetaSqlIdUtils.escapeAndJoin(allNameParts);
-
- return loadTable(fullEscapedTableName);
- }
-
- @Override
- public BeamSqlTable buildBeamSqlTable(Table table) {
- TableProvider tableProvider = DELEGATE_PROVIDERS.get(table.getType());
- if (tableProvider == null) {
- throw new RuntimeException("TableProvider is null");
- }
- return tableProvider.buildBeamSqlTable(table);
- }
-
- private Table loadTable(String tableName) {
- if (!tableCache.containsKey(tableName)) {
- tableCache.put(tableName, loadTableFromDC(tableName));
- }
-
- return tableCache.get(tableName);
- }
-
- private Table loadTableFromDC(String tableName) {
- try {
- return toCalciteTable(
- tableName,
- dataCatalog.lookupEntry(
-
LookupEntryRequest.newBuilder().setSqlResource(tableName).build()));
- } catch (InvalidArgumentException | PermissionDeniedException |
NotFoundException e) {
- throw new InvalidTableException("Could not resolve table in Data
Catalog: " + tableName, e);
- }
- }
-
- @Internal
- public static DataCatalogClient
createDataCatalogClient(DataCatalogPipelineOptions options) {
- try {
- DataCatalogSettings.Builder builder =
- DataCatalogSettings.newBuilder()
- .setCredentialsProvider(() ->
options.as(GcpOptions.class).getGcpCredential())
- .setEndpoint(options.getDataCatalogEndpoint());
-
- // Retry permission denied errors, they are likely due to sync delay.
- // Limit max retry delay to 1 minute, at that point its probably a
legitimate permission error
- // and we should get back to the user.
- builder
- .lookupEntrySettings()
- .setRetryableCodes(
- ImmutableSet.of(Code.PERMISSION_DENIED, Code.DEADLINE_EXCEEDED,
Code.UNAVAILABLE))
- .setRetrySettings(
- builder.lookupEntrySettings().getRetrySettings().toBuilder()
- .setMaxRetryDelay(Duration.ofMinutes(1L))
- .build());
- builder
- .updateEntrySettings()
- .setRetryableCodes(
- ImmutableSet.of(Code.PERMISSION_DENIED, Code.DEADLINE_EXCEEDED,
Code.UNAVAILABLE))
- .setRetrySettings(
- builder.updateEntrySettings().getRetrySettings().toBuilder()
- .setMaxRetryDelay(Duration.ofMinutes(1L))
- .build());
-
- return DataCatalogClient.create(builder.build());
- } catch (IOException e) {
- throw new RuntimeException("Error creating Data Catalog client", e);
- }
- }
-
- private Table toCalciteTable(String tableName, Entry entry) {
- if (entry.getSchema().getColumnsCount() == 0) {
- throw new UnsupportedOperationException(
- "Entry doesn't have a schema. Please attach a schema to '"
- + tableName
- + "' in Data Catalog: "
- + entry.toString());
- }
- Schema schema = SchemaUtils.fromDataCatalog(entry.getSchema());
-
- Optional<Table.Builder> tableBuilder = tableFactory.tableBuilder(entry);
- if (!tableBuilder.isPresent()) {
- throw new UnsupportedOperationException(
- String.format(
- "Unsupported Data Catalog entry: %s",
- MoreObjects.toStringHelper(entry)
- .add("linkedResource", entry.getLinkedResource())
- .add("hasGcsFilesetSpec", entry.hasGcsFilesetSpec())
- .toString()));
- }
-
- return tableBuilder.get().schema(schema).name(tableName).build();
- }
-
- @Internal
- public boolean setSchemaIfNotPresent(String resource, Schema schema) {
- com.google.cloud.datacatalog.v1beta1.Schema dcSchema =
SchemaUtils.toDataCatalog(schema);
- Entry entry =
-
dataCatalog.lookupEntry(LookupEntryRequest.newBuilder().setSqlResource(resource).build());
- if (entry.getSchema().getColumnsCount() == 0) {
- dataCatalog.updateEntry(
- UpdateEntryRequest.newBuilder()
- .setEntry(entry.toBuilder().setSchema(dcSchema).build())
- .setUpdateMask(FieldMask.newBuilder().addPaths("schema").build())
- .build());
- return true;
- } else {
- LOG.info("Not updating schema for '{}' since it already has one.",
resource);
- return false;
- }
- }
-
- @Override
- public void close() {
- dataCatalog.close();
- }
-}
diff --git
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/GcsTableFactory.java
b/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/GcsTableFactory.java
deleted file mode 100644
index 3a85362bbea..00000000000
---
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/GcsTableFactory.java
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.beam.sdk.extensions.sql.meta.provider.datacatalog;
-
-import com.google.cloud.datacatalog.v1beta1.Entry;
-import com.google.cloud.datacatalog.v1beta1.GcsFilesetSpec;
-import java.util.List;
-import java.util.Optional;
-import org.apache.beam.sdk.extensions.sql.TableUtils;
-import org.apache.beam.sdk.extensions.sql.meta.Table;
-
-/** {@link TableFactory} that understands Data Catalog GCS entries. */
-class GcsTableFactory implements TableFactory {
-
- /** Creates a Beam SQL table description from a GCS fileset entry. */
- @Override
- public Optional<Table.Builder> tableBuilder(Entry entry) {
- if (!entry.hasGcsFilesetSpec()) {
- return Optional.empty();
- }
-
- GcsFilesetSpec gcsFilesetSpec = entry.getGcsFilesetSpec();
- List<String> filePatterns = gcsFilesetSpec.getFilePatternsList();
-
- // We support exactly one 'file_patterns' field and nothing else at the
moment
- if (filePatterns.size() != 1) {
- throw new UnsupportedOperationException(
- "Unable to parse GCS entry '" + entry.getName() + "'");
- }
-
- String filePattern = filePatterns.get(0);
-
- if (!filePattern.startsWith("gs://")) {
- throw new UnsupportedOperationException(
- "Unsupported file pattern. "
- + "Only file patterns with 'gs://' schema are supported at the
moment.");
- }
-
- return Optional.of(
- Table.builder()
- .type("text")
- .location(filePattern)
- .properties(TableUtils.emptyProperties())
- .comment(""));
- }
-}
diff --git
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/PubsubTableFactory.java
b/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/PubsubTableFactory.java
deleted file mode 100644
index 5688713f7ee..00000000000
---
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/PubsubTableFactory.java
+++ /dev/null
@@ -1,62 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.beam.sdk.extensions.sql.meta.provider.datacatalog;
-
-import com.google.cloud.datacatalog.v1beta1.Entry;
-import java.net.URI;
-import java.util.Optional;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-import org.apache.beam.sdk.extensions.sql.TableUtils;
-import org.apache.beam.sdk.extensions.sql.meta.Table;
-
-/** {@link TableFactory} that understands Data Catalog Pubsub entries. */
-class PubsubTableFactory implements TableFactory {
-
- private static final String PUBSUB_API = "pubsub.googleapis.com";
-
- private static final Pattern PS_PATH_PATTERN =
- Pattern.compile("/projects/(?<PROJECT>[^/]+)/topics/(?<TOPIC>[^/]+)");
-
- @Override
- public Optional<Table.Builder> tableBuilder(Entry entry) {
- if
(!URI.create(entry.getLinkedResource()).getAuthority().equalsIgnoreCase(PUBSUB_API))
{
- return Optional.empty();
- }
-
- return Optional.of(
- Table.builder()
- .location(getLocation(entry))
- .properties(TableUtils.emptyProperties())
- .type("pubsub")
- .comment(""));
- }
-
- private static String getLocation(Entry entry) {
- URI entryName = URI.create(entry.getLinkedResource());
- String psPath = entryName.getPath();
-
- Matcher bqPathMatcher = PS_PATH_PATTERN.matcher(psPath);
- if (!bqPathMatcher.matches()) {
- throw new IllegalArgumentException(
- "Unsupported format for Pubsub topic: '" + entry.getLinkedResource()
+ "'");
- }
-
- return psPath.substring(1);
- }
-}
diff --git
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/SchemaUtils.java
b/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/SchemaUtils.java
deleted file mode 100644
index ef66d0e656c..00000000000
---
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/SchemaUtils.java
+++ /dev/null
@@ -1,190 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.beam.sdk.extensions.sql.meta.provider.datacatalog;
-
-import static org.apache.beam.sdk.schemas.Schema.toSchema;
-
-import com.google.cloud.datacatalog.v1beta1.ColumnSchema;
-import java.util.List;
-import java.util.Map;
-import org.apache.beam.sdk.schemas.Schema;
-import org.apache.beam.sdk.schemas.Schema.Field;
-import org.apache.beam.sdk.schemas.Schema.FieldType;
-import org.apache.beam.sdk.schemas.logicaltypes.SqlTypes;
-import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Strings;
-import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
-
-@SuppressWarnings({
- "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
-class SchemaUtils {
-
- private static final Map<String, FieldType> FIELD_TYPES =
- ImmutableMap.<String, FieldType>builder()
- .put("BOOL", FieldType.BOOLEAN)
- .put("BYTES", FieldType.BYTES)
- .put("DATE", FieldType.logicalType(SqlTypes.DATE))
- .put("DATETIME", FieldType.logicalType(SqlTypes.DATETIME))
- .put("DOUBLE", FieldType.DOUBLE)
- .put("FLOAT", FieldType.DOUBLE)
- .put("FLOAT64", FieldType.DOUBLE)
- .put("INT32", FieldType.INT32)
- .put("INT64", FieldType.INT64)
- .put("STRING", FieldType.STRING)
- .put("TIME", FieldType.logicalType(SqlTypes.TIME))
- .put("TIMESTAMP", FieldType.DATETIME)
- .put("NUMERIC", FieldType.DECIMAL)
- .put("MAP<STRING,STRING>", FieldType.map(FieldType.STRING,
FieldType.STRING))
- .build();
-
- /** Convert DataCatalog schema to Beam schema. */
- static Schema fromDataCatalog(com.google.cloud.datacatalog.v1beta1.Schema
dcSchema) {
- return fromColumnsList(dcSchema.getColumnsList());
- }
-
- private static Schema fromColumnsList(List<ColumnSchema> columnsMap) {
- return
columnsMap.stream().map(SchemaUtils::toBeamField).collect(toSchema());
- }
-
- private static Field toBeamField(ColumnSchema column) {
- String name = column.getColumn();
-
- // basic field type
- FieldType fieldType = getBeamFieldType(column);
- Field field = Field.of(name, fieldType);
-
- // set the nullable flag, or convert to a list if repeated
- if (Strings.isNullOrEmpty(column.getMode()) ||
"NULLABLE".equals(column.getMode())) {
- field = field.withNullable(true);
- } else if ("REQUIRED".equals(column.getMode())) {
- field = field.withNullable(false);
- } else if ("REPEATED".equals(column.getMode())) {
- field = Field.of(name, FieldType.array(fieldType));
- } else {
- throw new UnsupportedOperationException(
- "Field mode '" + column.getMode() + "' is not supported (field '" +
name + "')");
- }
-
- return field;
- }
-
- private static FieldType getBeamFieldType(ColumnSchema column) {
- String dcFieldType = column.getType();
-
- if (FIELD_TYPES.containsKey(dcFieldType)) {
- return FIELD_TYPES.get(dcFieldType);
- }
-
- if ("STRUCT".equals(dcFieldType)) {
- Schema structSchema = fromColumnsList(column.getSubcolumnsList());
- return FieldType.row(structSchema);
- }
-
- throw new UnsupportedOperationException(
- "Field type '" + dcFieldType + "' is not supported (field '" +
column.getColumn() + "')");
- }
-
- /** Convert Beam schema to DataCatalog schema. */
- static com.google.cloud.datacatalog.v1beta1.Schema toDataCatalog(Schema
schema) {
- com.google.cloud.datacatalog.v1beta1.Schema.Builder schemaBuilder =
- com.google.cloud.datacatalog.v1beta1.Schema.newBuilder();
- for (Schema.Field field : schema.getFields()) {
- schemaBuilder.addColumns(fromBeamField(field));
- }
- return schemaBuilder.build();
- }
-
- private static ColumnSchema fromBeamField(Schema.Field field) {
- Schema.FieldType fieldType = field.getType();
- if (fieldType.getTypeName().equals(Schema.TypeName.ARRAY)) {
- if (fieldType.getNullable()) {
- throw new UnsupportedOperationException(
- "Nullable array type is not supported in DataCatalog schemas: " +
fieldType);
- } else if (fieldType.getCollectionElementType().getNullable()) {
- throw new UnsupportedOperationException(
- "Nullable array element type is not supported in DataCatalog
schemas: " + fieldType);
- } else if
(fieldType.getCollectionElementType().getTypeName().equals(Schema.TypeName.ARRAY))
{
- throw new UnsupportedOperationException(
- "Array of arrays not supported in DataCatalog schemas: " +
fieldType);
- }
- ColumnSchema column =
- fromBeamField(Field.of(field.getName(),
fieldType.getCollectionElementType()));
- if (!column.getMode().equals("REQUIRED")) {
- // We should have bailed out earlier for any cases that would result
in mode being set.
- throw new AssertionError(
- "ColumnSchema for collection element type has non-empty mode: " +
fieldType);
- }
- return column.toBuilder().setMode("REPEATED").build();
- } else { // struct or primitive type
- ColumnSchema.Builder colBuilder =
- ColumnSchema.newBuilder().setType(getDataCatalogType(fieldType));
-
- if (fieldType.getNullable()) {
- colBuilder.setMode("NULLABLE");
- } else {
- colBuilder.setMode("REQUIRED");
- }
-
- // if this is a struct, add the child columns
- if (fieldType.getTypeName().equals(Schema.TypeName.ROW)) {
- for (Schema.Field subField : fieldType.getRowSchema().getFields()) {
- colBuilder.addSubcolumns(fromBeamField(subField));
- }
- }
-
- return colBuilder.setColumn(field.getName()).build();
- }
- }
-
- private static String getDataCatalogType(FieldType fieldType) {
- switch (fieldType.getTypeName()) {
- case INT32:
- case INT64:
- case BYTES:
- case DOUBLE:
- case STRING:
- return fieldType.getTypeName().name();
- case BOOLEAN:
- return "BOOL";
- case DATETIME:
- return "TIMESTAMP";
- case DECIMAL:
- return "NUMERIC";
- case LOGICAL_TYPE:
- Schema.LogicalType<?, ?> logical = fieldType.getLogicalType();
- if (SqlTypes.TIME.getIdentifier().equals(logical.getIdentifier())) {
- return "TIME";
- } else if
(SqlTypes.DATE.getIdentifier().equals(logical.getIdentifier())) {
- return "DATE";
- } else if
(SqlTypes.DATETIME.getIdentifier().equals(logical.getIdentifier())) {
- return "DATETIME";
- } else {
- throw new UnsupportedOperationException("Unsupported logical type: "
+ logical);
- }
- case ROW:
- return "STRUCT";
- case MAP:
- return String.format(
- "MAP<%s,%s>",
- getDataCatalogType(fieldType.getMapKeyType()),
- getDataCatalogType(fieldType.getMapValueType()));
- default:
- throw new UnsupportedOperationException("Unsupported type: " +
fieldType);
- }
- }
-}
diff --git
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/TableFactory.java
b/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/TableFactory.java
deleted file mode 100644
index 5a1f7182f78..00000000000
---
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/TableFactory.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.beam.sdk.extensions.sql.meta.provider.datacatalog;
-
-import com.google.cloud.datacatalog.v1beta1.Entry;
-import java.util.Optional;
-import org.apache.beam.sdk.extensions.sql.meta.Table;
-
-/**
- * A {@link TableFactory} <i>may</i> be able to interpret a given Data Catalog
{@link Entry} into
- * Beam SQL {@link Table}.
- */
-interface TableFactory {
-
- /**
- * If this {@link TableFactory} instance can interpret the given {@link
Entry}, then a Beam SQL
- * {@link Table} is constructed, else returns {@link Optional#empty}.
- *
- * <p>The {@link Table} is returned as a builder for further customization
by the caller.
- */
- Optional<Table.Builder> tableBuilder(Entry entry);
-}
diff --git
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/ZetaSqlIdUtils.java
b/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/ZetaSqlIdUtils.java
deleted file mode 100644
index e00fbb3148a..00000000000
---
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/ZetaSqlIdUtils.java
+++ /dev/null
@@ -1,83 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.beam.sdk.extensions.sql.meta.provider.datacatalog;
-
-import static java.util.stream.Collectors.joining;
-import static
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
-
-import java.util.List;
-import java.util.regex.Pattern;
-import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
-
-/** Utils to work with ZetaSQL-compatible IDs. */
-class ZetaSqlIdUtils {
-
- /**
- * Some special characters we explicitly handle.
- *
- * <p>Everything else is ignored, e.g. tabs, newlines, etc.
- */
- private static final Pattern SPECIAL_CHARS_ESCAPE =
- Pattern.compile(
- "(?<SpecialChar>["
- + "\\\\" // slash
- + "`" // backtick
- + "'" // single quote
- + "\"" // double quote
- + "?" // question mark
- + "])");
-
- private static final ImmutableMap<String, String> WHITESPACES =
- ImmutableMap.of(
- "\n", "\\\\n",
- "\t", "\\\\t",
- "\r", "\\\\r",
- "\f", "\\\\f");
-
- private static final Pattern SIMPLE_ID =
Pattern.compile("[A-Za-z_][A-Za-z_0-9]*");
-
- /**
- * Joins parts into a single compound ZetaSQL identifier.
- *
- * <p>Escapes backticks, slashes, double and single quotes, doesn't handle
other special
- * characters for now.
- */
- static String escapeAndJoin(List<String> parts) {
- return parts.stream()
- .map(ZetaSqlIdUtils::escapeSpecialChars)
- .map(ZetaSqlIdUtils::replaceWhitespaces)
- .map(ZetaSqlIdUtils::backtickIfNeeded)
- .collect(joining("."));
- }
-
- private static String escapeSpecialChars(String str) {
- return SPECIAL_CHARS_ESCAPE.matcher(str).replaceAll("\\\\${SpecialChar}");
- }
-
- private static String replaceWhitespaces(String s) {
- return WHITESPACES.keySet().stream()
- .reduce(
- s,
- (str, whitespace) ->
- str.replaceAll(whitespace,
checkNotNull(WHITESPACES.get(whitespace))));
- }
-
- private static String backtickIfNeeded(String s) {
- return SIMPLE_ID.matcher(s).matches() ? s : ("`" + s + "`");
- }
-}
diff --git
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/package-info.java
b/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/package-info.java
deleted file mode 100644
index 029f3f80010..00000000000
---
a/sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/package-info.java
+++ /dev/null
@@ -1,20 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-/** Table schema for Google Cloud Data Catalog. */
-package org.apache.beam.sdk.extensions.sql.meta.provider.datacatalog;
diff --git
a/sdks/java/extensions/sql/datacatalog/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/DataCatalogBigQueryIT.java
b/sdks/java/extensions/sql/datacatalog/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/DataCatalogBigQueryIT.java
deleted file mode 100644
index cdfc8ce1631..00000000000
---
a/sdks/java/extensions/sql/datacatalog/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/DataCatalogBigQueryIT.java
+++ /dev/null
@@ -1,115 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.beam.sdk.extensions.sql.meta.provider.datacatalog;
-
-import static org.apache.beam.sdk.schemas.Schema.FieldType.INT64;
-import static org.apache.beam.sdk.schemas.Schema.FieldType.STRING;
-
-import com.google.api.services.bigquery.model.TableReference;
-import java.util.Arrays;
-import org.apache.beam.sdk.extensions.sql.SqlTransform;
-import org.apache.beam.sdk.extensions.sql.impl.BeamSqlPipelineOptions;
-import org.apache.beam.sdk.extensions.sql.impl.CalciteQueryPlanner;
-import org.apache.beam.sdk.extensions.sql.impl.QueryPlanner;
-import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
-import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.Method;
-import org.apache.beam.sdk.io.gcp.bigquery.BigQueryUtils;
-import org.apache.beam.sdk.io.gcp.bigquery.TestBigQuery;
-import org.apache.beam.sdk.schemas.Schema;
-import org.apache.beam.sdk.testing.PAssert;
-import org.apache.beam.sdk.testing.TestPipeline;
-import org.apache.beam.sdk.transforms.Create;
-import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.Row;
-import org.joda.time.Duration;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.experimental.runners.Enclosed;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-
-/** Integration tests for DataCatalog+BigQuery. */
-@RunWith(Enclosed.class)
-public class DataCatalogBigQueryIT {
-
- @RunWith(Parameterized.class)
- public static class DialectSensitiveTests {
- private static final Schema ID_NAME_SCHEMA =
- Schema.builder().addNullableField("id",
INT64).addNullableField("name", STRING).build();
- @Rule public transient TestPipeline writePipeline = TestPipeline.create();
- @Rule public transient TestPipeline readPipeline = TestPipeline.create();
- @Rule public transient TestBigQuery bigQuery =
TestBigQuery.create(ID_NAME_SCHEMA);
-
- /** Parameterized by which SQL dialect, since the syntax here is the same.
*/
- @Parameterized.Parameters(name = "{0}")
- public static Iterable<Object[]> dialects() {
- return Arrays.asList(new Object[][] {{"CalciteSQL",
CalciteQueryPlanner.class}});
- }
-
- @SuppressWarnings("initialization.fields.uninitialized")
- @Parameterized.Parameter(0)
- public String dialectName;
-
- @SuppressWarnings("initialization.fields.uninitialized")
- @Parameterized.Parameter(1)
- public Class<? extends QueryPlanner> queryPlanner;
-
- @Test
- public void testRead() throws Exception {
- TableReference bqTable = bigQuery.tableReference();
-
- // Streaming inserts do not work with DIRECT_READ mode, there is a
several hour lag.
- PCollection<Row> data =
- writePipeline.apply(Create.of(row(1, "name1"), row(2, "name2"),
row(3, "name3")));
- data.apply(
- BigQueryIO.<Row>write()
- .withSchema(BigQueryUtils.toTableSchema(ID_NAME_SCHEMA))
- .withFormatFunction(BigQueryUtils.toTableRow())
- .withMethod(Method.FILE_LOADS)
- .to(bqTable));
- writePipeline.run().waitUntilFinish(Duration.standardMinutes(2));
-
- String tableId =
- String.format(
- "bigquery.`table`.`%s`.`%s`.`%s`",
- bqTable.getProjectId(), bqTable.getDatasetId(),
bqTable.getTableId());
-
- readPipeline
- .getOptions()
- .as(BeamSqlPipelineOptions.class)
- .setPlannerName(queryPlanner.getName());
-
- try (DataCatalogTableProvider tableProvider =
- DataCatalogTableProvider.create(
- readPipeline.getOptions().as(DataCatalogPipelineOptions.class)))
{
- PCollection<Row> result =
- readPipeline.apply(
- "query",
- SqlTransform.query("SELECT id, name FROM " + tableId)
- .withDefaultTableProvider("datacatalog", tableProvider));
-
- PAssert.that(result).containsInAnyOrder(row(1, "name1"), row(2,
"name2"), row(3, "name3"));
- readPipeline.run().waitUntilFinish(Duration.standardMinutes(2));
- }
- }
-
- private static Row row(long id, String name) {
- return Row.withSchema(ID_NAME_SCHEMA).addValues(id, name).build();
- }
- }
-}
diff --git
a/sdks/java/extensions/sql/datacatalog/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/DataCatalogGCSIT.java
b/sdks/java/extensions/sql/datacatalog/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/DataCatalogGCSIT.java
deleted file mode 100644
index b780f5b8708..00000000000
---
a/sdks/java/extensions/sql/datacatalog/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/DataCatalogGCSIT.java
+++ /dev/null
@@ -1,83 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.beam.sdk.extensions.sql.meta.provider.datacatalog;
-
-import static org.apache.beam.sdk.schemas.Schema.FieldType.INT32;
-import static org.apache.beam.sdk.schemas.Schema.FieldType.STRING;
-
-import java.io.Serializable;
-import org.apache.beam.runners.direct.DirectOptions;
-import org.apache.beam.sdk.extensions.sql.SqlTransform;
-import org.apache.beam.sdk.schemas.Schema;
-import org.apache.beam.sdk.testing.PAssert;
-import org.apache.beam.sdk.testing.TestPipeline;
-import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.Row;
-import org.joda.time.Duration;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
-
-/** Integration tests for DataCatalog+GCS. */
-@RunWith(JUnit4.class)
-public class DataCatalogGCSIT implements Serializable {
-
- private static final Schema ID_NAME_TYPE_SCHEMA =
- Schema.builder()
- .addNullableField("id", INT32)
- .addNullableField("name", STRING)
- .addNullableField("type", STRING)
- .build();
-
- @Rule public transient TestPipeline pipeline = TestPipeline.create();
-
- @Test
- public void testReadFromGCS() throws Exception {
- String gcsEntryId =
- "`datacatalog`" // this is part of the resource name in DataCatalog,
so it has to be
- + ".`entry`" // different from the table provider name ("dc" in
this test)
- + ".`apache-beam-testing`"
- + ".`us-central1`"
- + ".`samples`"
- + ".`integ_test_small_csv_test_1`";
-
- try (DataCatalogTableProvider tableProvider =
- DataCatalogTableProvider.create(
- pipeline.getOptions().as(DataCatalogPipelineOptions.class))) {
- PCollection<Row> result =
- pipeline.apply(
- "query",
- SqlTransform.query("SELECT id, name, type FROM " + gcsEntryId)
- .withDefaultTableProvider("dc", tableProvider));
-
- pipeline.getOptions().as(DirectOptions.class).setBlockOnRun(true);
- PAssert.that(result)
- .containsInAnyOrder(
- row(1, "customer1", "test"),
- row(2, "customer2", "test"),
- row(3, "customer1", "test"),
- row(4, "customer2", "test"));
- pipeline.run().waitUntilFinish(Duration.standardMinutes(2));
- }
- }
-
- private Row row(int id, String name, String type) {
- return Row.withSchema(ID_NAME_TYPE_SCHEMA).addValues(id, name,
type).build();
- }
-}
diff --git
a/sdks/java/extensions/sql/datacatalog/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/SchemaUtilsTest.java
b/sdks/java/extensions/sql/datacatalog/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/SchemaUtilsTest.java
deleted file mode 100644
index 6afc9466f60..00000000000
---
a/sdks/java/extensions/sql/datacatalog/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/SchemaUtilsTest.java
+++ /dev/null
@@ -1,157 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.beam.sdk.extensions.sql.meta.provider.datacatalog;
-
-import static org.junit.Assert.assertEquals;
-
-import com.google.cloud.datacatalog.v1beta1.ColumnSchema;
-import org.apache.beam.sdk.schemas.Schema;
-import org.apache.beam.sdk.schemas.Schema.FieldType;
-import org.apache.beam.sdk.schemas.logicaltypes.SqlTypes;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
-
-/** Unit tests for {@link SchemaUtils}. */
-@RunWith(JUnit4.class)
-public class SchemaUtilsTest {
-
- private static final Schema TEST_INNER_SCHEMA =
- Schema.builder().addField("i1", FieldType.INT64).addField("i2",
FieldType.STRING).build();
-
- private static final Schema TEST_SCHEMA =
- Schema.builder()
- .addNullableField("f_int32", FieldType.INT32)
- .addNullableField("f_int64", FieldType.INT64)
- .addNullableField("f_bytes", FieldType.BYTES)
- .addNullableField("f_double", FieldType.DOUBLE)
- .addNullableField("f_string", FieldType.STRING)
- .addNullableField("f_bool", FieldType.BOOLEAN)
- .addNullableField("f_ts", FieldType.DATETIME)
- .addNullableField("f_numeric", FieldType.DECIMAL)
- .addLogicalTypeField("f_time", SqlTypes.TIME)
- .addLogicalTypeField("f_date", SqlTypes.DATE)
- .addLogicalTypeField("f_datetime", SqlTypes.DATETIME)
- .addArrayField("f_array", FieldType.INT64)
- .addRowField("f_struct", TEST_INNER_SCHEMA)
- .build();
-
- private static final com.google.cloud.datacatalog.v1beta1.Schema
TEST_DC_SCHEMA =
- com.google.cloud.datacatalog.v1beta1.Schema.newBuilder()
- .addColumns(
- ColumnSchema.newBuilder()
- .setColumn("f_int32")
- .setType("INT32")
- .setMode("NULLABLE")
- .build())
- .addColumns(
- ColumnSchema.newBuilder()
- .setColumn("f_int64")
- .setType("INT64")
- .setMode("NULLABLE")
- .build())
- .addColumns(
- ColumnSchema.newBuilder()
- .setColumn("f_bytes")
- .setType("BYTES")
- .setMode("NULLABLE")
- .build())
- .addColumns(
- ColumnSchema.newBuilder()
- .setColumn("f_double")
- .setType("DOUBLE")
- .setMode("NULLABLE")
- .build())
- .addColumns(
- ColumnSchema.newBuilder()
- .setColumn("f_string")
- .setType("STRING")
- .setMode("NULLABLE")
- .build())
- .addColumns(
- ColumnSchema.newBuilder()
- .setColumn("f_bool")
- .setType("BOOL")
- .setMode("NULLABLE")
- .build())
- .addColumns(
- ColumnSchema.newBuilder()
- .setColumn("f_ts")
- .setType("TIMESTAMP")
- .setMode("NULLABLE")
- .build())
- .addColumns(
- ColumnSchema.newBuilder()
- .setColumn("f_numeric")
- .setType("NUMERIC")
- .setMode("NULLABLE")
- .build())
- .addColumns(
- ColumnSchema.newBuilder()
- .setColumn("f_time")
- .setType("TIME")
- .setMode("REQUIRED")
- .build())
- .addColumns(
- ColumnSchema.newBuilder()
- .setColumn("f_date")
- .setType("DATE")
- .setMode("REQUIRED")
- .build())
- .addColumns(
- ColumnSchema.newBuilder()
- .setColumn("f_datetime")
- .setType("DATETIME")
- .setMode("REQUIRED")
- .build())
- .addColumns(
- ColumnSchema.newBuilder()
- .setColumn("f_array")
- .setType("INT64")
- .setMode("REPEATED")
- .build())
- .addColumns(
- ColumnSchema.newBuilder()
- .setColumn("f_struct")
- .setType("STRUCT")
- .addSubcolumns(
- ColumnSchema.newBuilder()
- .setColumn("i1")
- .setType("INT64")
- .setMode("REQUIRED")
- .build())
- .addSubcolumns(
- ColumnSchema.newBuilder()
- .setColumn("i2")
- .setType("STRING")
- .setMode("REQUIRED")
- .build())
- .setMode("REQUIRED")
- .build())
- .build();
-
- @Test
- public void testFromDataCatalog() {
- assertEquals(TEST_SCHEMA, SchemaUtils.fromDataCatalog(TEST_DC_SCHEMA));
- }
-
- @Test
- public void testToDataCatalog() {
- assertEquals(TEST_DC_SCHEMA, SchemaUtils.toDataCatalog(TEST_SCHEMA));
- }
-}
diff --git
a/sdks/java/extensions/sql/datacatalog/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/ZetaSqlIdUtilsTest.java
b/sdks/java/extensions/sql/datacatalog/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/ZetaSqlIdUtilsTest.java
deleted file mode 100644
index ff34ef83666..00000000000
---
a/sdks/java/extensions/sql/datacatalog/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/ZetaSqlIdUtilsTest.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.beam.sdk.extensions.sql.meta.provider.datacatalog;
-
-import static org.junit.Assert.assertEquals;
-
-import java.util.Arrays;
-import java.util.List;
-import org.junit.Test;
-
-/** Unit tests for {@link ZetaSqlIdUtils}. */
-public class ZetaSqlIdUtilsTest {
-
- @Test
- public void testHandlesSimpleIds() {
- List<String> id = Arrays.asList("aaa", "BbB", "zAzzz00");
- assertEquals("aaa.BbB.zAzzz00", ZetaSqlIdUtils.escapeAndJoin(id));
- }
-
- @Test
- public void testHandlesMixedIds() {
- List<String> id = Arrays.asList("aaa", "Bb---B", "zAzzz00");
- assertEquals("aaa.`Bb---B`.zAzzz00", ZetaSqlIdUtils.escapeAndJoin(id));
- }
-
- @Test
- public void testHandlesSpecialChars() {
- List<String> id = Arrays.asList("a\\a", "b`b", "c'c", "d\"d", "e?e");
- assertEquals("`a\\\\a`.`b\\`b`.`c\\'c`.`d\\\"d`.`e\\?e`",
ZetaSqlIdUtils.escapeAndJoin(id));
- }
-
- @Test
- public void testHandlesSpecialCharsInOnePart() {
- List<String> id = Arrays.asList("a\\ab`bc'cd\"de?e");
- assertEquals("`a\\\\ab\\`bc\\'cd\\\"de\\?e`",
ZetaSqlIdUtils.escapeAndJoin(id));
- }
-
- @Test
- public void testHandlesWhiteSpaces() {
- List<String> id = Arrays.asList("a\na", "b\tb", "c\rc", "d\fd");
- assertEquals("`a\\na`.`b\\tb`.`c\\rc`.`d\\fd`",
ZetaSqlIdUtils.escapeAndJoin(id));
- }
-
- @Test
- public void testHandlesWhiteSpacesInOnePart() {
- List<String> id = Arrays.asList("a\nab\tbc\rcd\fd");
- assertEquals("`a\\nab\\tbc\\rcd\\fd`", ZetaSqlIdUtils.escapeAndJoin(id));
- }
-}
diff --git a/settings.gradle.kts b/settings.gradle.kts
index e1ed639b0d2..b8e7b855b99 100644
--- a/settings.gradle.kts
+++ b/settings.gradle.kts
@@ -219,7 +219,6 @@ include(":sdks:java:extensions:sql:payloads")
include(":sdks:java:extensions:sql:perf-tests")
include(":sdks:java:extensions:sql:jdbc")
include(":sdks:java:extensions:sql:hcatalog")
-include(":sdks:java:extensions:sql:datacatalog")
include(":sdks:java:extensions:sql:expansion-service")
include(":sdks:java:extensions:sql:udf")
include(":sdks:java:extensions:sql:udf-test-provider")
