From 083ecca636b1c632eaffecec0413800be09ce1ac Mon Sep 17 00:00:00 2001 From: aibrahiim Date: Mon, 19 Oct 2026 21:07:47 +0300 Subject: [PATCH 1/2] remove Beam SQL Data Catalog provider --- CHANGES.md | 1 + build.gradle.kts | 2 - .../beam/gradle/BeamModulePlugin.groovy | 2 - .../assets/symbols/java.g.yaml | 19 -- .../beam/checkstyle/suppressions.xml | 1 - .../extensions/sql/datacatalog/build.gradle | 107 -------- .../example/BeamSqlDataCatalogExample.java | 107 -------- .../extensions/sql/example/package-info.java | 20 -- .../datacatalog/BigQueryTableFactory.java | 72 ------ .../datacatalog/ChainedTableFactory.java | 50 ---- .../DataCatalogPipelineOptions.java | 43 ---- .../DataCatalogPipelineOptionsRegistrar.java | 31 --- .../datacatalog/DataCatalogTableProvider.java | 235 ------------------ .../provider/datacatalog/GcsTableFactory.java | 61 ----- .../datacatalog/PubsubTableFactory.java | 62 ----- .../provider/datacatalog/SchemaUtils.java | 190 -------------- .../provider/datacatalog/TableFactory.java | 37 --- .../provider/datacatalog/ZetaSqlIdUtils.java | 83 ------- .../provider/datacatalog/package-info.java | 20 -- .../datacatalog/DataCatalogBigQueryIT.java | 115 --------- .../datacatalog/DataCatalogGCSIT.java | 83 ------- .../provider/datacatalog/SchemaUtilsTest.java | 157 ------------ .../datacatalog/ZetaSqlIdUtilsTest.java | 64 ----- settings.gradle.kts | 1 - 24 files changed, 1 insertion(+), 1562 deletions(-) delete mode 100644 sdks/java/extensions/sql/datacatalog/build.gradle delete mode 100644 sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/example/BeamSqlDataCatalogExample.java delete mode 100644 sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/example/package-info.java delete mode 100644 sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/BigQueryTableFactory.java delete mode 100644 sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/ChainedTableFactory.java delete mode 100644 sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/DataCatalogPipelineOptions.java delete mode 100644 sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/DataCatalogPipelineOptionsRegistrar.java delete mode 100644 sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/DataCatalogTableProvider.java delete mode 100644 sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/GcsTableFactory.java delete mode 100644 sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/PubsubTableFactory.java delete mode 100644 sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/SchemaUtils.java delete mode 100644 sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/TableFactory.java delete mode 100644 sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/ZetaSqlIdUtils.java delete mode 100644 sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/package-info.java delete mode 100644 sdks/java/extensions/sql/datacatalog/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/DataCatalogBigQueryIT.java delete mode 100644 sdks/java/extensions/sql/datacatalog/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/DataCatalogGCSIT.java delete mode 100644 sdks/java/extensions/sql/datacatalog/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/SchemaUtilsTest.java delete mode 100644 sdks/java/extensions/sql/datacatalog/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/ZetaSqlIdUtilsTest.java diff --git a/CHANGES.md b/CHANGES.md index c10b7262209c..207a6bc368b7 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -83,6 +83,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 ([#35514](https://github.com/apache/beam/issues/35514)). * (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 8b68b2737961..1cefe1ac5b93 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 cf5979dd176b..17fc16e9967f 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 { 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 { 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 e0f2b1d15d14..2619d8702f10 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 833fd30cf218..d3ec39dc6ac4 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 @@ - diff --git a/sdks/java/extensions/sql/datacatalog/build.gradle b/sdks/java/extensions/sql/datacatalog/build.gradle deleted file mode 100644 index db4a92001501..000000000000 --- 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 0b2f1b04cbba..000000000000 --- 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 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= " - + "-PgcsTempRoot= " - + "-PqueryString= " - + "-PoutputFilePrefix= " - + "-PtempLocation=\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 7d6ff31c68a5..000000000000 --- 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 172dad752d93..000000000000 --- 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/(?[^/]+)/datasets/(?[^/]+)/tables/(?[^/]+)"); - - private final boolean truncateTimestamps; - - public BigQueryTableFactory(boolean truncateTimestamps) { - this.truncateTimestamps = truncateTimestamps; - } - - @Override - public Optional 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 a68dafe63f49..000000000000 --- 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 subTableFactories; - - public static ChainedTableFactory of(TableFactory... subTableFactories) { - return new ChainedTableFactory(Arrays.asList(subTableFactories)); - } - - private ChainedTableFactory(List subTableFactories) { - this.subTableFactories = subTableFactories; - } - - /** Creates a Beam SQL table description from a GCS fileset entry. */ - @Override - public Optional tableBuilder(Entry entry) { - for (TableFactory tableFactory : subTableFactories) { - Optional 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 bb4afee3a210..000000000000 --- 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 7f055dd245af..000000000000 --- 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> 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 19cc37b3f0dd..000000000000 --- 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 DELEGATE_PROVIDERS = - Stream.of(new PubsubTableProvider(), new BigQueryTableProvider(), new TextTableProvider()) - .collect(toMap(TableProvider::getTableType, p -> p)); - - private final DataCatalogClient dataCatalog; - private final Map 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 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 allNameParts = - ImmutableList.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 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 3a85362bbeab..000000000000 --- 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 tableBuilder(Entry entry) { - if (!entry.hasGcsFilesetSpec()) { - return Optional.empty(); - } - - GcsFilesetSpec gcsFilesetSpec = entry.getGcsFilesetSpec(); - List 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 5688713f7ee8..000000000000 --- 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/(?[^/]+)/topics/(?[^/]+)"); - - @Override - public Optional 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 ef66d0e656cb..000000000000 --- 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 FIELD_TYPES = - ImmutableMap.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", 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 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 5a1f7182f786..000000000000 --- 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} may 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}. - * - *

The {@link Table} is returned as a builder for further customization by the caller. - */ - Optional 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 e00fbb3148a7..000000000000 --- 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. - * - *

Everything else is ignored, e.g. tabs, newlines, etc. - */ - private static final Pattern SPECIAL_CHARS_ESCAPE = - Pattern.compile( - "(?[" - + "\\\\" // slash - + "`" // backtick - + "'" // single quote - + "\"" // double quote - + "?" // question mark - + "])"); - - private static final ImmutableMap 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. - * - *

Escapes backticks, slashes, double and single quotes, doesn't handle other special - * characters for now. - */ - static String escapeAndJoin(List 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 029f3f800103..000000000000 --- 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 cdfc8ce16316..000000000000 --- 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 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 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 data = - writePipeline.apply(Create.of(row(1, "name1"), row(2, "name2"), row(3, "name3"))); - data.apply( - BigQueryIO.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 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 b780f5b87085..000000000000 --- 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 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 6afc9466f60e..000000000000 --- 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 ff34ef836661..000000000000 --- 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 id = Arrays.asList("aaa", "BbB", "zAzzz00"); - assertEquals("aaa.BbB.zAzzz00", ZetaSqlIdUtils.escapeAndJoin(id)); - } - - @Test - public void testHandlesMixedIds() { - List id = Arrays.asList("aaa", "Bb---B", "zAzzz00"); - assertEquals("aaa.`Bb---B`.zAzzz00", ZetaSqlIdUtils.escapeAndJoin(id)); - } - - @Test - public void testHandlesSpecialChars() { - List 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 id = Arrays.asList("a\\ab`bc'cd\"de?e"); - assertEquals("`a\\\\ab\\`bc\\'cd\\\"de\\?e`", ZetaSqlIdUtils.escapeAndJoin(id)); - } - - @Test - public void testHandlesWhiteSpaces() { - List 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 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 e1ed639b0d24..b8e7b855b999 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") From 4829ac5bfe0d624a8c6d1487b6f0d5bb80ce7412 Mon Sep 17 00:00:00 2001 From: Yi Hu Date: Mon, 19 Oct 2026 15:58:04 -0400 Subject: [PATCH 2/2] Apply suggestion from @Abacn --- CHANGES.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGES.md b/CHANGES.md index 207a6bc368b7..c8e320858e7c 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -83,7 +83,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 ([#35514](https://github.com/apache/beam/issues/35514)). +* (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