ahmedabu98 commented on code in PR #39597:
URL: https://github.com/apache/beam/pull/39597#discussion_r3807868017
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryHelpers.java:
##########
@@ -471,14 +482,75 @@ public static TableReference parseTableSpec(String
tableSpec) {
"Table specification [%s] is not in one of the expected formats
("
+ " [project_id]:[dataset_id].[table_id],"
+ " [project_id].[dataset_id].[table_id],"
- + " [dataset_id].[table_id])",
+ + " [dataset_id].[table_id],"
+ + " [project_id]:[catalog_id].[namespace_id].[table_id],"
+ + " [project_id].[catalog_id].[namespace_id].[table_id])",
tableSpec));
}
- TableReference ref = new TableReference();
- ref.setProjectId(match.group("PROJECT"));
+ // Table ids cannot contain '.', so the table is always the segment after
+ // the last dot.
+ int lastDot = tableSpec.lastIndexOf('.');
+ String table = tableSpec.substring(lastDot + 1);
+ String prefix = tableSpec.substring(0, lastDot);
+
+ String project = null;
+ String dataset;
+ long colonCount = prefix.chars().filter(c -> c == ':').count();
+ if (colonCount == 0) {
+ // No colon means the purely dotted form ("p.d.t", "d.t",
"p.catalog.ns.t"): the
+ // leading segment is the project id when it is a plausible project id.
+ // (Dataset ids may contain characters such as '_' that project ids may
+ // not, in which case the whole prefix is the dataset id.)
+ // The firstDot < length-1 guard keeps degenerate trailing-dot specs
+ // ("pp..t", accepted by the character-set gate with dataset "pp.")
+ // instead of producing an empty dataset id.
+ int firstDot = prefix.indexOf('.');
+ if (firstDot >= 0
+ && firstDot < prefix.length() - 1
+ && BigQueryIO.PROJECT_ID_PATTERN.matcher(prefix.substring(0,
firstDot)).matches()) {
+ project = prefix.substring(0, firstDot);
+ dataset = prefix.substring(firstDot + 1);
+ } else {
+ dataset = prefix;
+ }
+ } else if (colonCount == 1) {
+ // One colon ("p:d.t", "p:catalog.ns.t", "example.com:proj.ds.t"). If the
+ // project part is dotted, it is a legacy domain-scoped id written with
+ // a '.' separator after the project: the first dataset segment completes
Review Comment:
"...**before** the project" ?
sorry, confused which way we're saying is forwards/backwards
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryHelpers.java:
##########
@@ -471,14 +482,75 @@ public static TableReference parseTableSpec(String
tableSpec) {
"Table specification [%s] is not in one of the expected formats
("
+ " [project_id]:[dataset_id].[table_id],"
+ " [project_id].[dataset_id].[table_id],"
- + " [dataset_id].[table_id])",
+ + " [dataset_id].[table_id],"
+ + " [project_id]:[catalog_id].[namespace_id].[table_id],"
+ + " [project_id].[catalog_id].[namespace_id].[table_id])",
tableSpec));
}
- TableReference ref = new TableReference();
- ref.setProjectId(match.group("PROJECT"));
+ // Table ids cannot contain '.', so the table is always the segment after
+ // the last dot.
+ int lastDot = tableSpec.lastIndexOf('.');
+ String table = tableSpec.substring(lastDot + 1);
+ String prefix = tableSpec.substring(0, lastDot);
+
+ String project = null;
+ String dataset;
+ long colonCount = prefix.chars().filter(c -> c == ':').count();
+ if (colonCount == 0) {
+ // No colon means the purely dotted form ("p.d.t", "d.t",
"p.catalog.ns.t"): the
+ // leading segment is the project id when it is a plausible project id.
+ // (Dataset ids may contain characters such as '_' that project ids may
+ // not, in which case the whole prefix is the dataset id.)
+ // The firstDot < length-1 guard keeps degenerate trailing-dot specs
+ // ("pp..t", accepted by the character-set gate with dataset "pp.")
+ // instead of producing an empty dataset id.
+ int firstDot = prefix.indexOf('.');
+ if (firstDot >= 0
+ && firstDot < prefix.length() - 1
+ && BigQueryIO.PROJECT_ID_PATTERN.matcher(prefix.substring(0,
firstDot)).matches()) {
+ project = prefix.substring(0, firstDot);
+ dataset = prefix.substring(firstDot + 1);
+ } else {
+ dataset = prefix;
+ }
+ } else if (colonCount == 1) {
+ // One colon ("p:d.t", "p:catalog.ns.t", "example.com:proj.ds.t"). If the
+ // project part is dotted, it is a legacy domain-scoped id written with
+ // a '.' separator after the project: the first dataset segment completes
+ // the project id, and any remaining middle segments bind as a (possibly
+ // composite) dataset. (Domain-scoped project names cannot contain dots)
+ int colon = prefix.indexOf(':');
+ project = prefix.substring(0, colon);
+ dataset = prefix.substring(colon + 1);
+ int firstDot = dataset.indexOf('.');
+ // Absorb the first dataset segment into a dotted (domain-scoped) project
+ // only when the split leaves a non-empty dataset.
+ if (firstDot >= 0
+ && firstDot < dataset.length() - 1
+ && project.indexOf('.') >= 0
+ && PROJECT_NAME_SEGMENT_PATTERN.matcher(dataset.substring(0,
firstDot)).matches()) {
+ project = project + ":" + dataset.substring(0, firstDot);
+ dataset = dataset.substring(firstDot + 1);
+ }
+ } else {
+ // Two colons - the last colon is an explicit project terminator. This is
+ // the canonical spelling for a domain-scoped project, whose id itself
+ // contains a colon ("example.com:proj:ds.t"), including with a composite
+ // Lakehouse catalog dataset ("example.com:proj:catalog.ns.t"). Both
+ // domain-scoped spellings keep toTableSpec/parseTableSpec a round trip
+ // for composite dataset ids. (More than two colons cannot form a valid
+ // reference - project ids contain at most one colon, but such specs pass
+ // the character-set gate, so they bind here too and the impossible
+ // project id is rejected by the service.)
+ int lastColon = prefix.lastIndexOf(':');
+ project = prefix.substring(0, lastColon);
+ dataset = prefix.substring(lastColon + 1);
+ }
Review Comment:
Should we make this `else if (colonCount == 2)` and throw an error on else?
3+ colons are not expected and should throw
##########
sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java:
##########
@@ -738,6 +753,250 @@ public void testWriteRead() throws IOException {
containsInAnyOrder(expectedRows.stream().map(RECORD_FUNC::apply).toArray()));
}
+ /**
+ * Cross-engine consistency: rows written through the Iceberg catalog must
be readable with
+ * BigQueryIO's Storage Read API using the catalog's BigQuery table
reference (see {@link
+ * #bigQueryTableSpec(String)}). Exercises the full read, server-side
projection + filtering
+ * push-down, and a query read with the reference embedded in SQL.
+ *
+ * <p>Rows are compared on a projection of fields whose types survive the
Iceberg-to-BigQuery
+ * mapping losslessly; BigQuery widens e.g. {@code int32} to {@code INT64},
so whole-row equality
+ * against the Iceberg schema does not hold by design.
+ */
+ @Test
+ public void testReadWithBigQueryIO() throws Exception {
Review Comment:
Is the output of `Managed.BIGQUERY` not aligned with `Managed.ICEBERG`?
Wondering why we're doing this canonicalization thing instead of comparing Beam
Rows directly
##########
sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/BigQueryManagedTableCrossEngineIT.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * 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.io.iceberg;
+
+import com.google.api.services.bigquery.Bigquery;
+import com.google.api.services.bigquery.model.QueryRequest;
+import com.google.api.services.bigquery.model.QueryResponse;
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.LongStream;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
+import org.apache.beam.sdk.io.gcp.testing.BigqueryClient;
+import org.apache.beam.sdk.managed.Managed;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.MapElements;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Splitter;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Reads a managed Apache Iceberg table in BigQuery through {@code
Managed.ICEBERG} with the
+ * BigQueryMetastore catalog, addressing it by its ordinary 3-part name.
+ *
+ * <p>Shares the {@code beam.bq.imt.*} system properties with
BigQueryIOIcebergManagedTableIT in the
+ * google-cloud-platform module.
+ */
+@RunWith(JUnit4.class)
+public class BigQueryManagedTableCrossEngineIT {
+
+ private static final BigqueryClient BQ_CLIENT =
+ new BigqueryClient("BigQueryManagedTableCrossEngineIT");
+ // BigqueryClient's query helpers stage a destination table (DDL rejects
that) and run in the
+ // default location, so SQL goes through a raw client instead.
+ private static final Bigquery RAW_BQ =
+ BigqueryClient.getNewBigqueryClient("BigQueryManagedTableCrossEngineIT");
+ private static final String PROJECT =
+ TestPipeline.testingPipelineOptions().as(GcpOptions.class).getProject();
+
+ // Connection in
"projects/{project}/locations/{location}/connections/{connection}" form.
+ private static final String CONNECTION =
+ System.getProperty(
+ "beam.bq.imt.connection",
+
"projects/apache-beam-testing/locations/us/connections/apache-beam-testing-storageapi-biglake-nodelete");
+ private static final String STORAGE_URI_ROOT =
+ System.getProperty("beam.bq.imt.storageUri",
"gs://apache-beam-testing-bq-biglake")
+ + "/BigQueryManagedTableCrossEngineIT";
+
+ private static final String DATASET_ID = "bq_imt_xengine_" +
System.nanoTime();
+
+ @BeforeClass
+ public static void setup() throws IOException, InterruptedException {
+ // The dataset must be colocated with the connection.
+ BQ_CLIENT.createNewDataset(
+ PROJECT, DATASET_ID, /* defaultTableExpirationMs= */ null,
connectionLocation());
+ }
+
+ @AfterClass
+ public static void cleanup() {
+ BQ_CLIENT.deleteDataset(PROJECT, DATASET_ID);
+ }
+
+ private static String connectionLocation() {
+ return Splitter.on('/').splitToList(CONNECTION).get(3);
+ }
+
+ /** Connection reference in the dotted form BigQuery DDL accepts. */
+ private static String connectionDotted() {
+ List<String> parts = Splitter.on('/').splitToList(CONNECTION);
+ return String.format("%s.%s.%s", parts.get(1), parts.get(3), parts.get(5));
+ }
+
+ /** Runs SQL in the connection's location. */
+ private static void runSql(String sql) throws IOException {
+ QueryResponse response =
+ RAW_BQ
+ .jobs()
+ .query(
+ PROJECT,
+ new QueryRequest()
+ .setQuery(sql)
+ .setUseLegacySql(false)
+ .setLocation(connectionLocation())
+ .setTimeoutMs(180_000L))
+ .execute();
+ if (!Boolean.TRUE.equals(response.getJobComplete())) {
+ throw new IOException("Query did not complete in time: " + sql);
+ }
+ }
+
+ @Test
+ public void testManagedIcebergReadByThreePartName() throws IOException {
+ String table = "managed_read_" + System.nanoTime();
+ runSql(
+ String.format(
+ "CREATE TABLE `%s.%s.%s` (id INT64, name STRING) WITH CONNECTION
`%s` "
+ + "OPTIONS (file_format='PARQUET', table_format='ICEBERG',
storage_uri='%s/%s/%s')",
+ PROJECT, DATASET_ID, table, connectionDotted(), STORAGE_URI_ROOT,
DATASET_ID, table));
+ runSql(
+ String.format(
+ "INSERT INTO `%s.%s.%s` "
+ + "SELECT id, CONCAT('row_', CAST(id AS STRING)) "
+ + "FROM UNNEST(GENERATE_ARRAY(0, 9)) id",
+ PROJECT, DATASET_ID, table));
+ // The catalog resolves the table through its exported Iceberg metadata,
and automatic exports
+ // can lag far behind recent writes; exporting makes the test
deterministic.
+ runSql(String.format("EXPORT TABLE METADATA FROM `%s.%s.%s`", PROJECT,
DATASET_ID, table));
+
+ Map<String, Object> config =
+ ImmutableMap.<String, Object>builder()
+ .put("table", DATASET_ID + "." + table)
+ .put(
+ "catalog_properties",
+ ImmutableMap.<String, String>builder()
+ .put("gcp_project", PROJECT)
+ .put("gcp_location", connectionLocation())
+ .put("warehouse", STORAGE_URI_ROOT)
+ .put("catalog-impl",
"org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog")
+ .put("io-impl", "org.apache.iceberg.gcp.gcs.GCSFileIO")
+ .build())
Review Comment:
Any reason why we're using `BigQueryMetastoreCatalog` (legacy) instead of
Lakehouse? If possible let's use Lakehouse IRC instead, see
https://github.com/apache/beam/blob/a99c2638cef6e3b6c553d2a4f700ba9e49ba0d60/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/RESTCatalogBLMSIT.java#L38-L46
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOIcebergManagedTableIT.java:
##########
@@ -0,0 +1,408 @@
+/*
+ * 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.io.gcp.bigquery;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import com.google.api.services.bigquery.Bigquery;
+import com.google.api.services.bigquery.model.QueryRequest;
+import com.google.api.services.bigquery.model.QueryResponse;
+import com.google.api.services.bigquery.model.Table;
+import com.google.api.services.bigquery.model.TableCell;
+import com.google.api.services.bigquery.model.TableFieldSchema;
+import com.google.api.services.bigquery.model.TableRow;
+import com.google.api.services.bigquery.model.TableSchema;
+import java.io.IOException;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.LongStream;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
+import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.TypedRead.Method;
+import org.apache.beam.sdk.io.gcp.testing.BigqueryClient;
+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.transforms.MapElements;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Splitter;
+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.ImmutableMap;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.hamcrest.Matchers;
+import org.joda.time.Duration;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Integration tests for BigQueryIO against managed Apache Iceberg tables in
BigQuery.
+ *
+ * <p>Requires a CLOUD_RESOURCE connection whose service account can
administer the storage bucket.
+ * Defaults target standing apache-beam-testing resources (shared with {@link
+ * StorageApiSinkCreateIfNeededIT}); override with the {@code
beam.bq.imt.connection} and {@code
+ * beam.bq.imt.storageUri} system properties.
+ */
+@RunWith(JUnit4.class)
+public class BigQueryIOIcebergManagedTableIT {
Review Comment:
Let's add this to the excludes list for dataflow integration tests in
`beam/runners/google-cloud-dataflow-java/build.gradle`
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryHelpersTest.java:
##########
@@ -101,13 +103,368 @@ public void testTableParsing_noProjectId() {
assertEquals("table_name", ref.getTableId());
}
+ @Test
+ public void testTableParsing_lakehouseCatalogDotted() {
+ // 4-part Lakehouse runtime catalog reference:
project.catalog.namespace.table. The
+ // catalog+namespace form a composite dataset id, including when the
catalog name uses the
+ // GCS-bucket charset (lowercase, digits, dashes).
+ TableReference ref =
BigQueryHelpers.parseTableSpec("my-project.my-bucket-catalog.my_ns.tbl");
+ assertEquals("my-project", ref.getProjectId());
+ assertEquals("my-bucket-catalog.my_ns", ref.getDatasetId());
+ assertEquals("tbl", ref.getTableId());
+ }
+
+ @Test
+ public void testTableParsing_lakehouseCatalogColon() {
+ TableReference ref =
BigQueryHelpers.parseTableSpec("my-project:my-catalog.my_ns.tbl");
+ assertEquals("my-project", ref.getProjectId());
+ assertEquals("my-catalog.my_ns", ref.getDatasetId());
+ assertEquals("tbl", ref.getTableId());
+ }
+
+ @Test
+ public void testTableParsing_lakehouseCatalogNoProject() {
+ // Dataset ids may contain characters that project ids may not (e.g. '_'),
in which case the
+ // whole prefix is the (composite) dataset id.
+ TableReference ref =
BigQueryHelpers.parseTableSpec("my_catalog.my_ns.tbl");
+ assertEquals(null, ref.getProjectId());
+ assertEquals("my_catalog.my_ns", ref.getDatasetId());
+ assertEquals("tbl", ref.getTableId());
+ }
+
+ @Test
+ public void testTableParsing_multiLevelNamespace() {
+ // More than four segments: everything between the project and the table
becomes the dataset.
+ TableReference ref =
BigQueryHelpers.parseTableSpec("my-project.cat.ns1.ns2.tbl");
+ assertEquals("my-project", ref.getProjectId());
+ assertEquals("cat.ns1.ns2", ref.getDatasetId());
+ assertEquals("tbl", ref.getTableId());
+ }
+
+ @Test
+ public void testTableParsing_lakehouseWithPartitionDecorator() {
+ TableReference ref =
BigQueryHelpers.parseTableSpec("my-project.my-catalog.ns.tbl$20260101");
+ assertEquals("my-project", ref.getProjectId());
+ assertEquals("my-catalog.ns", ref.getDatasetId());
+ assertEquals("tbl$20260101", ref.getTableId());
+ }
+
+ @Test
+ public void testTableParsing_domainScopedProjectPreserved() {
+ // Legacy domain-scoped projects keep their historical binding.
+ TableReference ref =
BigQueryHelpers.parseTableSpec("example.com:project:data_set.tbl");
+ assertEquals("example.com:project", ref.getProjectId());
+ assertEquals("data_set", ref.getDatasetId());
+ assertEquals("tbl", ref.getTableId());
+
+ ref = BigQueryHelpers.parseTableSpec("example.com:project.data_set.tbl");
+ assertEquals("example.com:project", ref.getProjectId());
+ assertEquals("data_set", ref.getDatasetId());
+ assertEquals("tbl", ref.getTableId());
+ }
+
+ @Test
+ public void testTableParsing_domainScopedProjectWithCompositeDataset() {
+ // With two colons, the last colon is an explicit project terminator, so a
domain-scoped
+ // project can address a Lakehouse catalog table: the remainder binds as a
composite
+ // dataset.
+ TableReference ref =
BigQueryHelpers.parseTableSpec("example.com:project:cat.ns.tbl");
+ assertEquals("example.com:project", ref.getProjectId());
+ assertEquals("cat.ns", ref.getDatasetId());
+ assertEquals("tbl", ref.getTableId());
+
+ // The dotted spelling binds consistently: the first segment after the
colon completes the
+ // domain-scoped project id; further middle segments form the composite
dataset. (Project
+ // names cannot contain dots, so the pre-fix greedy binding of this string
was invalid.)
+ ref = BigQueryHelpers.parseTableSpec("example.com:project.cat.ns.tbl");
+ assertEquals("example.com:project", ref.getProjectId());
+ assertEquals("cat.ns", ref.getDatasetId());
+ assertEquals("tbl", ref.getTableId());
+ }
+
+ @Test
+ public void testTableParsing_partitionDecoratorColonForm() {
+ TableReference ref =
BigQueryHelpers.parseTableSpec("my-project:my-catalog.ns.tbl$20260101");
+ assertEquals("my-project", ref.getProjectId());
+ assertEquals("my-catalog.ns", ref.getDatasetId());
+ assertEquals("tbl$20260101", ref.getTableId());
+ }
+
+ @Test
+ public void testTableParsing_tableIdSpecialCharacters() {
+ // Table ids may contain spaces, '@', '$', dashes, and unicode letters,
none of which
+ // affect segment binding (only '.' and ':' are structural).
+ TableReference ref =
BigQueryHelpers.parseTableSpec("my-project.data_set.my table@x-1");
+ assertEquals("my-project", ref.getProjectId());
+ assertEquals("data_set", ref.getDatasetId());
+ assertEquals("my table@x-1", ref.getTableId());
+
+ ref = BigQueryHelpers.parseTableSpec("my-project.my-catalog.ns.ग्राहक");
+ assertEquals("my-project", ref.getProjectId());
+ assertEquals("my-catalog.ns", ref.getDatasetId());
+ assertEquals("ग्राहक", ref.getTableId());
+ }
+
+ @Test
+ public void testTableParsing_colonFormMultiLevelNamespace() {
+ TableReference ref =
BigQueryHelpers.parseTableSpec("my-project:cat.ns1.ns2.tbl");
+ assertEquals("my-project", ref.getProjectId());
+ assertEquals("cat.ns1.ns2", ref.getDatasetId());
+ assertEquals("tbl", ref.getTableId());
+ }
+
+ @Test
+ public void testTableParsing_domainScopedMultiLevelNamespace() {
+ // Single-colon domain-scoped spelling with a multi-level composite
dataset: the first
+ // segment after the colon completes the project id; everything else up to
the table binds
+ // as the dataset.
+ TableReference ref =
BigQueryHelpers.parseTableSpec("example.com:proj.cat.ns1.ns2.tbl");
+ assertEquals("example.com:proj", ref.getProjectId());
+ assertEquals("cat.ns1.ns2", ref.getDatasetId());
+ assertEquals("tbl", ref.getTableId());
+ }
+
+ @Test
+ public void testTableParsing_moreThanTwoColonsBindsAtLastColon() {
+ // More than two colons cannot form a valid reference (project ids contain
at most one
+ // colon), but such specs pass the character-set gate; they bind at the
last colon so the
+ // impossible project id is rejected by the service rather than producing
a malformed
+ // dataset id.
+ TableReference ref =
BigQueryHelpers.parseTableSpec("d1:d2:d3:data_set.tbl");
+ assertEquals("d1:d2:d3", ref.getProjectId());
+ assertEquals("data_set", ref.getDatasetId());
+ assertEquals("tbl", ref.getTableId());
+ }
Review Comment:
building off the last comment: is this valid? or should we throw here
##########
sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/BigQueryManagedTableCrossEngineIT.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * 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.io.iceberg;
+
+import com.google.api.services.bigquery.Bigquery;
+import com.google.api.services.bigquery.model.QueryRequest;
+import com.google.api.services.bigquery.model.QueryResponse;
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.LongStream;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
+import org.apache.beam.sdk.io.gcp.testing.BigqueryClient;
+import org.apache.beam.sdk.managed.Managed;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.MapElements;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Splitter;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Reads a managed Apache Iceberg table in BigQuery through {@code
Managed.ICEBERG} with the
+ * BigQueryMetastore catalog, addressing it by its ordinary 3-part name.
+ *
+ * <p>Shares the {@code beam.bq.imt.*} system properties with
BigQueryIOIcebergManagedTableIT in the
+ * google-cloud-platform module.
+ */
+@RunWith(JUnit4.class)
+public class BigQueryManagedTableCrossEngineIT {
+
+ private static final BigqueryClient BQ_CLIENT =
+ new BigqueryClient("BigQueryManagedTableCrossEngineIT");
+ // BigqueryClient's query helpers stage a destination table (DDL rejects
that) and run in the
+ // default location, so SQL goes through a raw client instead.
+ private static final Bigquery RAW_BQ =
+ BigqueryClient.getNewBigqueryClient("BigQueryManagedTableCrossEngineIT");
+ private static final String PROJECT =
+ TestPipeline.testingPipelineOptions().as(GcpOptions.class).getProject();
+
+ // Connection in
"projects/{project}/locations/{location}/connections/{connection}" form.
+ private static final String CONNECTION =
+ System.getProperty(
+ "beam.bq.imt.connection",
+
"projects/apache-beam-testing/locations/us/connections/apache-beam-testing-storageapi-biglake-nodelete");
+ private static final String STORAGE_URI_ROOT =
+ System.getProperty("beam.bq.imt.storageUri",
"gs://apache-beam-testing-bq-biglake")
+ + "/BigQueryManagedTableCrossEngineIT";
+
+ private static final String DATASET_ID = "bq_imt_xengine_" +
System.nanoTime();
+
+ @BeforeClass
+ public static void setup() throws IOException, InterruptedException {
+ // The dataset must be colocated with the connection.
+ BQ_CLIENT.createNewDataset(
+ PROJECT, DATASET_ID, /* defaultTableExpirationMs= */ null,
connectionLocation());
+ }
+
+ @AfterClass
+ public static void cleanup() {
+ BQ_CLIENT.deleteDataset(PROJECT, DATASET_ID);
+ }
+
+ private static String connectionLocation() {
+ return Splitter.on('/').splitToList(CONNECTION).get(3);
+ }
+
+ /** Connection reference in the dotted form BigQuery DDL accepts. */
+ private static String connectionDotted() {
+ List<String> parts = Splitter.on('/').splitToList(CONNECTION);
+ return String.format("%s.%s.%s", parts.get(1), parts.get(3), parts.get(5));
+ }
+
+ /** Runs SQL in the connection's location. */
+ private static void runSql(String sql) throws IOException {
+ QueryResponse response =
+ RAW_BQ
+ .jobs()
+ .query(
+ PROJECT,
+ new QueryRequest()
+ .setQuery(sql)
+ .setUseLegacySql(false)
+ .setLocation(connectionLocation())
+ .setTimeoutMs(180_000L))
+ .execute();
+ if (!Boolean.TRUE.equals(response.getJobComplete())) {
+ throw new IOException("Query did not complete in time: " + sql);
+ }
+ }
+
+ @Test
+ public void testManagedIcebergReadByThreePartName() throws IOException {
+ String table = "managed_read_" + System.nanoTime();
+ runSql(
+ String.format(
+ "CREATE TABLE `%s.%s.%s` (id INT64, name STRING) WITH CONNECTION
`%s` "
+ + "OPTIONS (file_format='PARQUET', table_format='ICEBERG',
storage_uri='%s/%s/%s')",
+ PROJECT, DATASET_ID, table, connectionDotted(), STORAGE_URI_ROOT,
DATASET_ID, table));
+ runSql(
+ String.format(
+ "INSERT INTO `%s.%s.%s` "
+ + "SELECT id, CONCAT('row_', CAST(id AS STRING)) "
+ + "FROM UNNEST(GENERATE_ARRAY(0, 9)) id",
+ PROJECT, DATASET_ID, table));
+ // The catalog resolves the table through its exported Iceberg metadata,
and automatic exports
+ // can lag far behind recent writes; exporting makes the test
deterministic.
+ runSql(String.format("EXPORT TABLE METADATA FROM `%s.%s.%s`", PROJECT,
DATASET_ID, table));
+
+ Map<String, Object> config =
+ ImmutableMap.<String, Object>builder()
+ .put("table", DATASET_ID + "." + table)
+ .put(
+ "catalog_properties",
+ ImmutableMap.<String, String>builder()
+ .put("gcp_project", PROJECT)
+ .put("gcp_location", connectionLocation())
+ .put("warehouse", STORAGE_URI_ROOT)
+ .put("catalog-impl",
"org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog")
+ .put("io-impl", "org.apache.iceberg.gcp.gcs.GCSFileIO")
+ .build())
+ .build();
+
+ Pipeline p = Pipeline.create(TestPipeline.testingPipelineOptions());
+ PCollection<String> rows =
+ p.apply(Managed.read(Managed.ICEBERG).withConfig(config))
+ .getSinglePCollection()
+ .apply(
+ MapElements.into(TypeDescriptors.strings())
+ .via(row -> row.getInt64("id") + "|" +
row.getString("name")));
+ PAssert.that(rows)
+ .containsInAnyOrder(
+ LongStream.range(0, 10).mapToObj(i -> i + "|row_" +
i).collect(Collectors.toList()));
Review Comment:
Let's also check `Managed.read(Managed.BIGQUERY)`
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryHelpersTest.java:
##########
@@ -101,13 +103,368 @@ public void testTableParsing_noProjectId() {
assertEquals("table_name", ref.getTableId());
}
+ @Test
+ public void testTableParsing_lakehouseCatalogDotted() {
+ // 4-part Lakehouse runtime catalog reference:
project.catalog.namespace.table. The
+ // catalog+namespace form a composite dataset id, including when the
catalog name uses the
+ // GCS-bucket charset (lowercase, digits, dashes).
+ TableReference ref =
BigQueryHelpers.parseTableSpec("my-project.my-bucket-catalog.my_ns.tbl");
+ assertEquals("my-project", ref.getProjectId());
+ assertEquals("my-bucket-catalog.my_ns", ref.getDatasetId());
+ assertEquals("tbl", ref.getTableId());
+ }
+
+ @Test
+ public void testTableParsing_lakehouseCatalogColon() {
+ TableReference ref =
BigQueryHelpers.parseTableSpec("my-project:my-catalog.my_ns.tbl");
+ assertEquals("my-project", ref.getProjectId());
+ assertEquals("my-catalog.my_ns", ref.getDatasetId());
+ assertEquals("tbl", ref.getTableId());
+ }
+
+ @Test
+ public void testTableParsing_lakehouseCatalogNoProject() {
+ // Dataset ids may contain characters that project ids may not (e.g. '_'),
in which case the
+ // whole prefix is the (composite) dataset id.
+ TableReference ref =
BigQueryHelpers.parseTableSpec("my_catalog.my_ns.tbl");
+ assertEquals(null, ref.getProjectId());
+ assertEquals("my_catalog.my_ns", ref.getDatasetId());
+ assertEquals("tbl", ref.getTableId());
+ }
+
+ @Test
+ public void testTableParsing_multiLevelNamespace() {
+ // More than four segments: everything between the project and the table
becomes the dataset.
+ TableReference ref =
BigQueryHelpers.parseTableSpec("my-project.cat.ns1.ns2.tbl");
+ assertEquals("my-project", ref.getProjectId());
+ assertEquals("cat.ns1.ns2", ref.getDatasetId());
+ assertEquals("tbl", ref.getTableId());
+ }
+
+ @Test
+ public void testTableParsing_lakehouseWithPartitionDecorator() {
+ TableReference ref =
BigQueryHelpers.parseTableSpec("my-project.my-catalog.ns.tbl$20260101");
+ assertEquals("my-project", ref.getProjectId());
+ assertEquals("my-catalog.ns", ref.getDatasetId());
+ assertEquals("tbl$20260101", ref.getTableId());
+ }
+
+ @Test
+ public void testTableParsing_domainScopedProjectPreserved() {
+ // Legacy domain-scoped projects keep their historical binding.
+ TableReference ref =
BigQueryHelpers.parseTableSpec("example.com:project:data_set.tbl");
+ assertEquals("example.com:project", ref.getProjectId());
+ assertEquals("data_set", ref.getDatasetId());
+ assertEquals("tbl", ref.getTableId());
+
+ ref = BigQueryHelpers.parseTableSpec("example.com:project.data_set.tbl");
+ assertEquals("example.com:project", ref.getProjectId());
+ assertEquals("data_set", ref.getDatasetId());
+ assertEquals("tbl", ref.getTableId());
+ }
+
+ @Test
+ public void testTableParsing_domainScopedProjectWithCompositeDataset() {
+ // With two colons, the last colon is an explicit project terminator, so a
domain-scoped
+ // project can address a Lakehouse catalog table: the remainder binds as a
composite
+ // dataset.
+ TableReference ref =
BigQueryHelpers.parseTableSpec("example.com:project:cat.ns.tbl");
+ assertEquals("example.com:project", ref.getProjectId());
+ assertEquals("cat.ns", ref.getDatasetId());
+ assertEquals("tbl", ref.getTableId());
+
+ // The dotted spelling binds consistently: the first segment after the
colon completes the
+ // domain-scoped project id; further middle segments form the composite
dataset. (Project
+ // names cannot contain dots, so the pre-fix greedy binding of this string
was invalid.)
+ ref = BigQueryHelpers.parseTableSpec("example.com:project.cat.ns.tbl");
+ assertEquals("example.com:project", ref.getProjectId());
+ assertEquals("cat.ns", ref.getDatasetId());
+ assertEquals("tbl", ref.getTableId());
+ }
+
+ @Test
+ public void testTableParsing_partitionDecoratorColonForm() {
+ TableReference ref =
BigQueryHelpers.parseTableSpec("my-project:my-catalog.ns.tbl$20260101");
+ assertEquals("my-project", ref.getProjectId());
+ assertEquals("my-catalog.ns", ref.getDatasetId());
+ assertEquals("tbl$20260101", ref.getTableId());
+ }
+
+ @Test
+ public void testTableParsing_tableIdSpecialCharacters() {
+ // Table ids may contain spaces, '@', '$', dashes, and unicode letters,
none of which
+ // affect segment binding (only '.' and ':' are structural).
+ TableReference ref =
BigQueryHelpers.parseTableSpec("my-project.data_set.my table@x-1");
+ assertEquals("my-project", ref.getProjectId());
+ assertEquals("data_set", ref.getDatasetId());
+ assertEquals("my table@x-1", ref.getTableId());
+
+ ref = BigQueryHelpers.parseTableSpec("my-project.my-catalog.ns.ग्राहक");
+ assertEquals("my-project", ref.getProjectId());
+ assertEquals("my-catalog.ns", ref.getDatasetId());
+ assertEquals("ग्राहक", ref.getTableId());
+ }
+
+ @Test
+ public void testTableParsing_colonFormMultiLevelNamespace() {
+ TableReference ref =
BigQueryHelpers.parseTableSpec("my-project:cat.ns1.ns2.tbl");
+ assertEquals("my-project", ref.getProjectId());
+ assertEquals("cat.ns1.ns2", ref.getDatasetId());
+ assertEquals("tbl", ref.getTableId());
+ }
+
+ @Test
+ public void testTableParsing_domainScopedMultiLevelNamespace() {
+ // Single-colon domain-scoped spelling with a multi-level composite
dataset: the first
+ // segment after the colon completes the project id; everything else up to
the table binds
+ // as the dataset.
+ TableReference ref =
BigQueryHelpers.parseTableSpec("example.com:proj.cat.ns1.ns2.tbl");
+ assertEquals("example.com:proj", ref.getProjectId());
+ assertEquals("cat.ns1.ns2", ref.getDatasetId());
+ assertEquals("tbl", ref.getTableId());
+ }
+
+ @Test
+ public void testTableParsing_moreThanTwoColonsBindsAtLastColon() {
+ // More than two colons cannot form a valid reference (project ids contain
at most one
+ // colon), but such specs pass the character-set gate; they bind at the
last colon so the
+ // impossible project id is rejected by the service rather than producing
a malformed
+ // dataset id.
+ TableReference ref =
BigQueryHelpers.parseTableSpec("d1:d2:d3:data_set.tbl");
+ assertEquals("d1:d2:d3", ref.getProjectId());
+ assertEquals("data_set", ref.getDatasetId());
+ assertEquals("tbl", ref.getTableId());
+ }
Review Comment:
If it was valid before, then let's keep it to avoid a breaking change
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
