From e0f708b785ec0f9766294dab85a94cd3e2ab7e12 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 27 Aug 2026 08:36:09 -0400 Subject: [PATCH 1/6] Validate multi region biglake catalog --- sdks/java/io/iceberg/build.gradle | 15 +- .../beam/sdk/io/iceberg/AddFilesIT.java | 228 ++++++++++++++---- .../sdk/io/iceberg/BigLakeTestCatalog.java | 146 +++++++++++ .../iceberg/catalog/IcebergCatalogBaseIT.java | 2 +- .../io/iceberg/catalog/RESTCatalogBLMSIT.java | 108 +++++++-- 5 files changed, 425 insertions(+), 74 deletions(-) create mode 100644 sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/BigLakeTestCatalog.java diff --git a/sdks/java/io/iceberg/build.gradle b/sdks/java/io/iceberg/build.gradle index 596962b1a03f..a008b6d66723 100644 --- a/sdks/java/io/iceberg/build.gradle +++ b/sdks/java/io/iceberg/build.gradle @@ -179,10 +179,19 @@ task integrationTest(type: Test) { "--project=${gcpProject}", "--tempLocation=${gcpTempLocation}", ]) - // Warehouse (= catalog) used by the BigLake REST catalog tests; overridable for runs - // against a non-default project's catalog. + // Multiple-bucket BigLake REST catalog used by RESTCatalogBLMSIT and AddFilesIT (see + // BigLakeTestCatalog). Locations: the catalog's default location first, then a restricted + // location in another bucket. The cross-region location must be a bucket in a different + // region than the catalog. Override for runs against another project's catalog: + // -PbiglakeWarehouse=bl://projects/PROJECT/catalogs/CATALOG + // -PbiglakeLocations=gs://default-bucket/path,gs://other-bucket/path + // -PbiglakeCrossRegionLocation=gs://bucket-in-another-region/path systemProperty "beam.iceberg.biglake.warehouse", - project.findProperty('biglakeWarehouse') ?: 'gs://managed-iceberg-biglake-its' + project.findProperty('biglakeWarehouse') ?: 'bl://projects/apache-beam-testing/catalogs/beam-lakehouse-it' + systemProperty "beam.iceberg.biglake.locations", + project.findProperty('biglakeLocations') ?: 'gs://beam-lakehouse-it,gs://beam-lakehouse-it-added-path' + systemProperty "beam.iceberg.biglake.crossRegionLocation", + project.findProperty('biglakeCrossRegionLocation') ?: 'gs://managed-iceberg-biglake-its/biglake_cross_region' // Connection + storage root for BigQueryManagedTableCrossEngineIT. if (project.findProperty('bqImtConnection') != null) { systemProperty "beam.bq.imt.connection", project.findProperty('bqImtConnection') diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java index 5afc6107c58b..f486e2f40979 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java @@ -22,9 +22,13 @@ import static org.apache.beam.sdk.io.iceberg.IcebergUtils.beamSchemaToIcebergSchema; import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; import static org.apache.beam.sdk.values.TypeDescriptors.strings; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.startsWith; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; import com.google.api.services.storage.model.StorageObject; @@ -79,6 +83,7 @@ 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.apache.hadoop.util.Lists; +import org.apache.iceberg.BaseTable; import org.apache.iceberg.DataFile; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Snapshot; @@ -105,12 +110,12 @@ public class AddFilesIT { private static final Logger LOG = LoggerFactory.getLogger(AddFilesIT.class); - // Bucket-backed BigLake catalogs are named after their bucket. Overridable for local runs - // against a different project's catalog: -Dbeam.iceberg.biglake.warehouse=gs://my-bucket - private static final String CATALOG_NAME = - System.getProperty("beam.iceberg.biglake.warehouse", "gs://managed-iceberg-biglake-its") - .replace("gs://", ""); - private static final String WAREHOUSE = "gs://" + CATALOG_NAME; + // Multiple-bucket BigLake catalog (see BigLakeTestCatalog). Source parquet files, and the + // GCS notifications announcing them, live under the catalog's default location. + private static final String DATA_LOCATION = BigLakeTestCatalog.defaultLocation(); + private static final String DATA_BUCKET = BigLakeTestCatalog.bucketOf(DATA_LOCATION); + private static final String DATA_PREFIX = BigLakeTestCatalog.prefixOf(DATA_LOCATION); + private static final String CROSS_REGION_LOCATION = BigLakeTestCatalog.CROSS_REGION_LOCATION; private static final String PROJECT = TestPipeline.testingPipelineOptions().as(GcpOptions.class).getProject(); @Rule public TestName testName = new TestName(); @@ -127,20 +132,13 @@ public class AddFilesIT { .addStringField("name") .addStringField("kind") .build(); - private static final Map BIGLAKE_PROPS = - Map.of( - "type", "rest", - "uri", "https://biglake.googleapis.com/iceberg/v1/restcatalog", - "warehouse", WAREHOUSE, - "header.x-goog-user-project", PROJECT, - "rest.auth.type", "google", - "io-impl", "org.apache.iceberg.gcp.gcs.GCSFileIO", - // required by vended-credentials catalogs, harmless on end-user ones - "header.X-Iceberg-Access-Delegation", "vended-credentials"); + private static final Map BIGLAKE_PROPS = BigLakeTestCatalog.catalogProperties(); private Storage storage; private PubsubClient pubsub; private Notification notification; private final String namespace = getClass().getSimpleName() + "_" + System.currentTimeMillis(); + // Namespace placed in the catalog's additional location (second bucket). + private final String altNamespace = namespace + "_alt"; private String srcTableName; private String destTableName; private TableIdentifier srcTableId; @@ -176,29 +174,30 @@ public void setup() throws IOException { .setPayloadFormat(NotificationInfo.PayloadFormat.JSON_API_V1) .build(); try { - notification = storage.createNotification(WAREHOUSE.replace("gs://", ""), notificationInfo); + notification = storage.createNotification(DATA_BUCKET, notificationInfo); } catch (StorageException e) { if (e.getMessage().contains("Too many overlapping notifications")) { - List existing = storage.listNotifications(WAREHOUSE.replace("gs://", "")); + List existing = storage.listNotifications(DATA_BUCKET); LOG.warn( "Too many notifications on bucket {}: {}. Deleting existing notifications to make room: {}", - WAREHOUSE, + DATA_BUCKET, e, existing.stream() .map(NotificationInfo::getNotificationId) .collect(Collectors.toList())); - existing.forEach( - n -> storage.deleteNotification(WAREHOUSE.replace("gs://", ""), n.getNotificationId())); + existing.forEach(n -> storage.deleteNotification(DATA_BUCKET, n.getNotificationId())); // try creating it again - notification = storage.createNotification(WAREHOUSE.replace("gs://", ""), notificationInfo); + notification = storage.createNotification(DATA_BUCKET, notificationInfo); } else { throw e; } } salt = System.currentTimeMillis(); - dirName = format("%s-%s/%s", getClass().getSimpleName(), salt, testName.getMethodName()); + dirName = + format( + "%s/%s-%s/%s", DATA_PREFIX, getClass().getSimpleName(), salt, testName.getMethodName()); srcTableName = "src_" + testName.getMethodName() + "_" + salt; destTableName = "dest_" + testName.getMethodName() + "_" + salt; srcTableId = TableIdentifier.of(namespace, srcTableName); @@ -210,10 +209,12 @@ public void setup() throws IOException { } private void cleanupCatalog() { - Namespace ns = Namespace.of(namespace); - if (catalog.namespaceExists(ns)) { - catalog.listTables(ns).forEach(catalog::dropTable); - catalog.dropNamespace(ns); + for (String name : Arrays.asList(namespace, altNamespace)) { + Namespace ns = Namespace.of(name); + if (catalog.namespaceExists(ns)) { + catalog.listTables(ns).forEach(catalog::dropTable); + catalog.dropNamespace(ns); + } } } @@ -227,7 +228,7 @@ public void cleanup() { } try { - storage.deleteNotification(WAREHOUSE.replace("gs://", ""), notification.getNotificationId()); + storage.deleteNotification(DATA_BUCKET, notification.getNotificationId()); storage.close(); } catch (Exception e) { LOG.warn("Failed to clean up GCS notifications", e); @@ -239,14 +240,19 @@ public void cleanup() { LOG.warn("Failed to clean up Iceberg catalog", e); } + deleteBlobs(DATA_BUCKET, dirName); + deleteBlobs( + BigLakeTestCatalog.bucketOf(CROSS_REGION_LOCATION), + BigLakeTestCatalog.prefixOf(CROSS_REGION_LOCATION) + "/" + dirName); + } + + private void deleteBlobs(String bucket, String prefix) { try { Iterable blobs = - storage - .list(WAREHOUSE.replace("gs://", ""), Storage.BlobListOption.prefix(dirName)) - .getValues(); + storage.list(bucket, Storage.BlobListOption.prefix(prefix)).getValues(); blobs.forEach(b -> storage.delete(b.getBlobId())); } catch (Exception e) { - LOG.warn("Failed to clean up GCS bucket", e); + LOG.warn("Failed to clean up gs://{}/{}", bucket, prefix, e); } } @@ -352,8 +358,8 @@ public void testStreamingParquetImport() throws InterruptedException, TimeoutException, IOException { // start with a table that does not exist - String parquetDir = format("%s/%s/", WAREHOUSE, dirName); - String tempDir = format("%s/%s-tmp/", WAREHOUSE, dirName); + String parquetDir = format("gs://%s/%s/", DATA_BUCKET, dirName); + String tempDir = format("gs://%s/%s-tmp/", DATA_BUCKET, dirName); // let the add files pipeline run in the background PipelineResult addFilesPipeline = startAddFilesListener(dirName); @@ -385,8 +391,7 @@ record -> GcsUtil gcsUtil = TestPipeline.testingPipelineOptions().as(GcsOptions.class).getGcsUtil(); - Iterable objects = - gcsUtil.listObjects(WAREHOUSE.replace("gs://", ""), dirName, null).getItems(); + Iterable objects = gcsUtil.listObjects(DATA_BUCKET, dirName, null).getItems(); List writtenFilePaths = Lists.newArrayList(objects).stream() .map(o -> format("gs://%s/%s", o.getBucket(), o.getName())) @@ -426,11 +431,147 @@ public void testBatchParquetImportToUIT() throws IOException { testBatchParquetImport(true); } + /** + * The catalog does not police where added files live: data in a bucket it does not manage, even + * in another region, registers and reads fine through Iceberg. BigQuery, however, cannot read + * data files outside the catalog's region, so the cross-engine read of such a table fails. (Files + * in an unmanaged bucket in the catalog's region read fine from BigQuery.) Documents the + * trade-off of adding files in place from a cross-region bucket. + */ + @Test + public void testBatchParquetImportFromCrossRegionBucket() throws IOException { + String crossRegionBucket = BigLakeTestCatalog.bucketOf(CROSS_REGION_LOCATION); + for (String location : BigLakeTestCatalog.LOCATIONS) { + assertNotEquals( + "Test needs a bucket outside the catalog: " + CROSS_REGION_LOCATION, + BigLakeTestCatalog.bucketOf(location), + crossRegionBucket); + } + List writtenFilePaths = + writeParquetFiles( + crossRegionBucket, BigLakeTestCatalog.prefixOf(CROSS_REGION_LOCATION) + "/" + dirName); + + Pipeline p = Pipeline.create(); + PCollectionRowTuple tuple = + p.apply(Create.of(writtenFilePaths)) + .apply( + new AddFiles( + IcebergCatalogConfig.builder().setCatalogProperties(BIGLAKE_PROPS).build(), + destTableId.toString(), + null, + PARTITION_FIELDS, + null, + TABLE_PROPS, + null, + null)); + PAssert.that(tuple.get("errors")).empty(); + p.run().waitUntilFinish(); + + assertTrue(checkTableHasRegisteredParquetFiles(writtenFilePaths)); + checkRecordsInDestinationTable(/* alsoCheckWithBigQueryIO= */ false); + + Pipeline bq = Pipeline.create(); + bq.apply( + Managed.read(Managed.BIGQUERY) + .withConfig( + ImmutableMap.of( + "table", + BigLakeTestCatalog.bigQueryTableSpec( + destTableId.namespace().toString(), destTableId.name())))) + .getSinglePCollection(); + assertThrows(Pipeline.PipelineExecutionException.class, () -> bq.run().waitUntilFinish()); + } + + /** + * The destination table lives in the catalog's additional location (a second bucket) while the + * source parquet files stay in the default one. BigLake pins tables under their namespace's + * location, so the table is created in a namespace placed in the second bucket; AddFiles must + * commit metadata there and reference the files in place across buckets. + */ + @Test + public void testBatchParquetImportToTableInAdditionalLocation() throws IOException { + String namespaceLocation = BigLakeTestCatalog.additionalLocation() + "/" + altNamespace; + assertNotEquals( + "Test needs two distinct buckets", + DATA_BUCKET, + BigLakeTestCatalog.bucketOf(namespaceLocation)); + catalog.createNamespace( + Namespace.of(altNamespace), ImmutableMap.of("location", namespaceLocation)); + destTableId = TableIdentifier.of(altNamespace, destTableName); + catalog.createTable(destTableId, beamSchemaToIcebergSchema(ROW_SCHEMA), SPEC); + assertThat(catalog.loadTable(destTableId).location(), startsWith(namespaceLocation)); + + List writtenFilePaths = writeParquetFiles(); + Pipeline p = Pipeline.create(); + PCollectionRowTuple tuple = + p.apply(Create.of(writtenFilePaths)) + .apply( + new AddFiles( + IcebergCatalogConfig.builder().setCatalogProperties(BIGLAKE_PROPS).build(), + destTableId.toString(), + null, + PARTITION_FIELDS, + null, + TABLE_PROPS, + null, + null)); + PAssert.that(tuple.get("errors")).empty(); + p.run().waitUntilFinish(); + + assertTrue(checkTableHasRegisteredParquetFiles(writtenFilePaths)); + Table destTable = catalog.loadTable(destTableId); + String metadataLocation = ((BaseTable) destTable).operations().current().metadataFileLocation(); + assertThat(metadataLocation, startsWith(BigLakeTestCatalog.additionalLocation())); + for (String path : writtenFilePaths) { + assertThat(path, startsWith("gs://" + DATA_BUCKET + "/")); + } + checkRecordsInDestinationTable(/* alsoCheckWithBigQueryIO= */ true); + } + + private List writeParquetFiles() throws IOException { + return writeParquetFiles(DATA_BUCKET, dirName); + } + + /** Writes TEST_ROWS as parquet under gs://{bucket}/{dir}/ and returns the written file paths. */ + private List writeParquetFiles(String bucket, String dir) throws IOException { + String parquetDir = format("gs://%s/%s/", bucket, dir); + String tempDir = format("gs://%s/%s-tmp/", bucket, dir); + LOG.info("Writing records to the parquet dir"); + Pipeline q = Pipeline.create(); + org.apache.avro.Schema avroSchema = AvroUtils.toAvroSchema(ROW_SCHEMA); + q.apply(Create.of(TEST_ROWS)) + .setRowSchema(ROW_SCHEMA) + .apply( + MapElements.into(TypeDescriptor.of(GenericRecord.class)) + .via(AvroUtils.getRowToGenericRecordFunction(avroSchema))) + .setCoder(AvroCoder.of(avroSchema)) + .apply( + FileIO.writeDynamic() + .by( + record -> + format("%s-%s-%s", record.get("id"), record.get("name"), record.get("age"))) + .via(ParquetIO.sink(avroSchema)) + .withNaming(name -> defaultNaming(name, ".parquet")) + .withTempDirectory(tempDir) + .to(parquetDir) + .withDestinationCoder(StringUtf8Coder.of())); + q.run().waitUntilFinish(); + + GcsUtil gcsUtil = TestPipeline.testingPipelineOptions().as(GcsOptions.class).getGcsUtil(); + Iterable objects = gcsUtil.listObjects(bucket, dir, null).getItems(); + List writtenFilePaths = + Lists.newArrayList(objects).stream() + .map(o -> format("gs://%s/%s", o.getBucket(), o.getName())) + .collect(Collectors.toList()); + LOG.info("Written file paths: {}", writtenFilePaths); + return writtenFilePaths; + } + private void testBatchParquetImport(boolean isUIT) throws IOException { // start with a table that does not exist - String parquetDir = format("%s/%s/", WAREHOUSE, dirName); - String tempDir = format("%s/%s-tmp/", WAREHOUSE, dirName); + String parquetDir = format("gs://%s/%s/", DATA_BUCKET, dirName); + String tempDir = format("gs://%s/%s-tmp/", DATA_BUCKET, dirName); // write some parquet files LOG.info("Writing records to the parquet dir"); @@ -456,8 +597,7 @@ record -> GcsUtil gcsUtil = TestPipeline.testingPipelineOptions().as(GcsOptions.class).getGcsUtil(); - Iterable objects = - gcsUtil.listObjects(WAREHOUSE.replace("gs://", ""), dirName, null).getItems(); + Iterable objects = gcsUtil.listObjects(DATA_BUCKET, dirName, null).getItems(); List writtenFilePaths = Lists.newArrayList(objects).stream() .map(o -> format("gs://%s/%s", o.getBucket(), o.getName())) @@ -511,14 +651,14 @@ public void testBatchParquetImportWithSchemaEvolution() throws IOException { Schema narrow = Schema.builder().addInt64Field("id").addStringField("name").build(); catalog.createTable(destTableId, beamSchemaToIcebergSchema(narrow)); - String parquetDir = format("%s/%s/", WAREHOUSE, dirName); - String tempDir = format("%s/%s-tmp/", WAREHOUSE, dirName); + String parquetDir = format("gs://%s/%s/", DATA_BUCKET, dirName); + String tempDir = format("gs://%s/%s-tmp/", DATA_BUCKET, dirName); writeParquet(TEST_ROWS, ROW_SCHEMA, parquetDir + "plain/", tempDir + "plain/"); GcsUtil gcsUtil = TestPipeline.testingPipelineOptions().as(GcsOptions.class).getGcsUtil(); List writtenFilePaths = Lists.newArrayList( - gcsUtil.listObjects(WAREHOUSE.replace("gs://", ""), dirName, null).getItems()) + gcsUtil.listObjects(DATA_BUCKET, dirName, null).getItems()) .stream() .map(o -> format("gs://%s/%s", o.getBucket(), o.getName())) .collect(Collectors.toList()); @@ -634,7 +774,7 @@ private void checkRecordsInDestinationTable(boolean alsoCheckWithBigQueryIO) { format( "%s.%s.%s.%s", PROJECT, - CATALOG_NAME, + BigLakeTestCatalog.CATALOG_ID, destTableId.namespace(), destTableId.name())))) .getSinglePCollection() diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/BigLakeTestCatalog.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/BigLakeTestCatalog.java new file mode 100644 index 000000000000..061babb7d0d5 --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/BigLakeTestCatalog.java @@ -0,0 +1,146 @@ +/* + * 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 static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument; + +import java.util.List; +import java.util.Map; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import org.apache.beam.sdk.extensions.gcp.options.GcpOptions; +import org.apache.beam.sdk.testing.TestPipeline; +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; + +/** + * Test-side description of the BigLake (Lakehouse) Iceberg REST catalog the ITs run against. + * + *

The catalog is a multiple-bucket catalog addressed by a {@code + * bl://projects/PROJECT/catalogs/CATALOG} warehouse and allowed to store resources under a fixed + * set of Cloud Storage locations: the catalog's default location plus any restricted locations. + * Configured (defaults and overrides) by the integrationTest task in build.gradle through the + * system properties + * + *

    + *
  • {@code beam.iceberg.biglake.warehouse}: the {@code bl://} warehouse URI + *
  • {@code beam.iceberg.biglake.locations}: comma-separated {@code gs://} prefixes the catalog + * may write to; the first is the catalog's default location, the rest are additional + * restricted locations + *
  • {@code beam.iceberg.biglake.crossRegionLocation}: a {@code gs://} prefix in a bucket + * located in a different region than the catalog + *
+ */ +public final class BigLakeTestCatalog { + private static final Pattern WAREHOUSE_PATTERN = + Pattern.compile("bl://projects/([^/]+)/catalogs/([^/]+)"); + + public static final String WAREHOUSE = requiredProperty("beam.iceberg.biglake.warehouse"); + + public static final List LOCATIONS = + ImmutableList.copyOf( + Splitter.on(',') + .trimResults() + .omitEmptyStrings() + .split(requiredProperty("beam.iceberg.biglake.locations"))); + + /** + * A location in a bucket the catalog does not manage and that is in a different region than the + * catalog (BigQuery cannot read cross-region data files; see + * AddFilesIT#testBatchParquetImportFromCrossRegionBucket). + */ + public static final String CROSS_REGION_LOCATION = + requiredProperty("beam.iceberg.biglake.crossRegionLocation"); + + /** Catalog id, which is also the second segment of BigQuery's 4-part table reference. */ + public static final String CATALOG_ID = parseCatalogId(WAREHOUSE); + + private static final String PROJECT = + TestPipeline.testingPipelineOptions().as(GcpOptions.class).getProject(); + + private BigLakeTestCatalog() {} + + private static String requiredProperty(String name) { + String value = System.getProperty(name); + checkArgument( + value != null && !value.isEmpty(), + "System property %s is not set; run through the integrationTest gradle task, which" + + " sets it, or pass -D%s=...", + name, + name); + return value; + } + + private static String parseCatalogId(String warehouse) { + // Legacy single-bucket catalogs are addressed by their bucket and named after it. + if (warehouse.startsWith("gs://")) { + return bucketOf(warehouse); + } + Matcher matcher = WAREHOUSE_PATTERN.matcher(warehouse); + checkArgument( + matcher.matches(), + "Expected a bl://projects/PROJECT/catalogs/CATALOG (or legacy gs://BUCKET) warehouse, got '%s'", + warehouse); + return matcher.group(2); + } + + /** The catalog's default location; tables land here unless created with an explicit location. */ + public static String defaultLocation() { + return LOCATIONS.get(0); + } + + /** A restricted location outside the default one, i.e. in a second bucket. */ + public static String additionalLocation() { + checkArgument( + LOCATIONS.size() >= 2, + "beam.iceberg.biglake.locations must list at least two locations, got %s", + LOCATIONS); + return LOCATIONS.get(1); + } + + public static String bucketOf(String gcsLocation) { + checkArgument(gcsLocation.startsWith("gs://"), "Not a gs:// location: %s", gcsLocation); + String withoutScheme = gcsLocation.substring("gs://".length()); + int slash = withoutScheme.indexOf('/'); + return slash < 0 ? withoutScheme : withoutScheme.substring(0, slash); + } + + /** Object-name prefix (no bucket, no leading slash) of a {@code gs://bucket/path} location. */ + public static String prefixOf(String gcsLocation) { + String withoutScheme = gcsLocation.substring("gs://".length()); + int slash = withoutScheme.indexOf('/'); + return slash < 0 ? "" : withoutScheme.substring(slash + 1); + } + + public static Map catalogProperties() { + return ImmutableMap.builder() + .put("type", "rest") + .put("uri", "https://biglake.googleapis.com/iceberg/v1/restcatalog") + .put("warehouse", WAREHOUSE) + .put("header.x-goog-user-project", PROJECT) + .put("io-impl", "org.apache.iceberg.gcp.gcs.GCSFileIO") + .put("rest.auth.type", "org.apache.iceberg.gcp.auth.GoogleAuthManager") + .build(); + } + + /** BigQuery's 4-part {@code project.catalog.namespace.table} reference. */ + public static String bigQueryTableSpec(String namespace, String table) { + return String.format("%s.%s.%s.%s", PROJECT, CATALOG_ID, namespace, table); + } +} diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java index 859f20c8fa69..4495c741b155 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java @@ -432,7 +432,7 @@ private List populateTable(Table table, @Nullable String charOverride) thro } } - private List readRecords(Table table) throws IOException { + protected List readRecords(Table table) throws IOException { org.apache.iceberg.Schema tableSchema = table.schema(); TableScan tableScan = table.newScan().project(tableSchema); List writtenRecords = new ArrayList<>(); diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/RESTCatalogBLMSIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/RESTCatalogBLMSIT.java index aa83d2b7db2d..95442455132f 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/RESTCatalogBLMSIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/RESTCatalogBLMSIT.java @@ -17,44 +17,53 @@ */ package org.apache.beam.sdk.io.iceberg.catalog; +import static org.apache.beam.sdk.managed.Managed.ICEBERG; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.containsInAnyOrder; +import static org.hamcrest.Matchers.startsWith; +import static org.junit.Assert.assertFalse; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; import java.util.Map; +import org.apache.beam.sdk.io.iceberg.BigLakeTestCatalog; +import org.apache.beam.sdk.managed.Managed; +import org.apache.beam.sdk.transforms.Create; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.BaseTable; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; import org.apache.iceberg.catalog.Catalog; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.SupportsNamespaces; import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.Record; import org.apache.iceberg.rest.RESTCatalog; import org.junit.After; import org.junit.BeforeClass; +import org.junit.Test; -/** Tests for {@link org.apache.iceberg.rest.RESTCatalog} using BigLake Metastore. */ +/** + * Tests for {@link org.apache.iceberg.rest.RESTCatalog} using a multiple-bucket BigLake Metastore + * catalog (see {@link BigLakeTestCatalog}). + */ public class RESTCatalogBLMSIT extends IcebergCatalogBaseIT { private static Map catalogProps; - // Using a special bucket for this test class because - // BigLake does not support using subfolders as a warehouse (yet). - // Overridable for local runs against a different project's catalog, e.g. - // -Dbeam.iceberg.biglake.warehouse=gs://my-bucket (bucket-backed catalogs are named after - // their bucket). - private static final String BIGLAKE_WAREHOUSE = - System.getProperty("beam.iceberg.biglake.warehouse", "gs://managed-iceberg-biglake-its"); - @BeforeClass public static void setup() { - warehouse = BIGLAKE_WAREHOUSE; - catalogProps = - ImmutableMap.builder() - .put("type", "rest") - .put("uri", "https://biglake.googleapis.com/iceberg/v1/restcatalog") - .put("warehouse", BIGLAKE_WAREHOUSE) - .put("header.x-goog-user-project", OPTIONS.getProject()) - .put("io-impl", "org.apache.iceberg.gcp.gcs.GCSFileIO") - .put("rest.auth.type", "org.apache.iceberg.gcp.auth.GoogleAuthManager") - .build(); + // The catalog decides where tables go (its default location); the base class only uses + // `warehouse` to sweep leftover files, so point it at that location. + warehouse = BigLakeTestCatalog.defaultLocation(); + catalogProps = BigLakeTestCatalog.catalogProperties(); } @After public void after() { // making sure the cleanup path is directed at the correct warehouse - warehouse = BIGLAKE_WAREHOUSE; + warehouse = BigLakeTestCatalog.defaultLocation(); } @Override @@ -65,13 +74,11 @@ public String type() { @Override public String bigQueryTableSpec(String tableId) { // BigQuery surfaces Lakehouse runtime catalog (BigLake metastore REST) tables via 4-part - // project.catalog.namespace.table identifiers; the catalog id of a bucket-backed catalog is - // the bucket name. Requires the caller to hold biglake.* read permissions (e.g. - // roles/biglake.viewer) in addition to the usual BigQuery roles. + // project.catalog.namespace.table identifiers. Requires the caller to hold biglake.* read + // permissions (e.g. roles/biglake.viewer) in addition to the usual BigQuery roles. TableIdentifier identifier = TableIdentifier.parse(tableId); - String catalogId = BIGLAKE_WAREHOUSE.replace("gs://", ""); - return String.format( - "%s.%s.%s.%s", OPTIONS.getProject(), catalogId, identifier.namespace(), identifier.name()); + return BigLakeTestCatalog.bigQueryTableSpec( + identifier.namespace().toString(), identifier.name()); } @Override @@ -88,4 +95,53 @@ public Map managedIcebergConfig(String tableId) { .put("catalog_properties", catalogProps) .build(); } + + /** + * A multiple-bucket catalog may place resources under any of its restricted locations, not just + * its default one. BigLake pins a table under its namespace's location, so the namespace is + * created in a second bucket; Beam only receives the table through the catalog and must write + * wherever it was placed. + */ + @Test + public void testWriteReadTableInAdditionalLocation() throws IOException { + String altNamespace = namespace() + "_alt"; + String altTableId = altNamespace + ".test_table"; + String namespaceLocation = BigLakeTestCatalog.additionalLocation() + "/" + altNamespace; + assertFalse( + "Test needs two distinct buckets", + BigLakeTestCatalog.bucketOf(namespaceLocation) + .equals(BigLakeTestCatalog.bucketOf(BigLakeTestCatalog.defaultLocation()))); + namespacesToCleanup.add(altNamespace); + ((SupportsNamespaces) catalog) + .createNamespace( + Namespace.of(altNamespace), ImmutableMap.of("location", namespaceLocation)); + Table table = catalog.createTable(TableIdentifier.parse(altTableId), ICEBERG_SCHEMA); + assertThat(table.location(), startsWith(namespaceLocation)); + + pipeline + .apply(Create.of(inputRows)) + .setRowSchema(BEAM_SCHEMA) + .apply(Managed.write(ICEBERG).withConfig(managedIcebergConfig(altTableId))); + pipeline.run().waitUntilFinish(); + + table.refresh(); + List returnedRecords = readRecords(table); + assertThat( + returnedRecords, containsInAnyOrder(inputRows.stream().map(RECORD_FUNC::apply).toArray())); + + // Both the data files Beam wrote and the metadata the catalog committed live in the + // additional location, not in the catalog's default bucket. + List dataFileLocations = new ArrayList<>(); + for (Snapshot snapshot : table.snapshots()) { + for (DataFile dataFile : snapshot.addedDataFiles(table.io())) { + dataFileLocations.add(dataFile.location()); + } + } + assertFalse("No data files were written", dataFileLocations.isEmpty()); + for (String location : dataFileLocations) { + assertThat(location, startsWith(BigLakeTestCatalog.additionalLocation())); + } + String metadataLocation = ((BaseTable) table).operations().current().metadataFileLocation(); + assertThat(metadataLocation, startsWith(BigLakeTestCatalog.additionalLocation())); + } } From ef5908035523769ad2690e9258cf66bfda9ffb71 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 27 Aug 2026 08:45:15 -0400 Subject: [PATCH 2/6] trigger tests --- .github/trigger_files/IO_Iceberg_Integration_Tests.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/trigger_files/IO_Iceberg_Integration_Tests.json b/.github/trigger_files/IO_Iceberg_Integration_Tests.json index 7392be3b11c1..e1290fa50c32 100644 --- a/.github/trigger_files/IO_Iceberg_Integration_Tests.json +++ b/.github/trigger_files/IO_Iceberg_Integration_Tests.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", - "modification": 10 + "modification": 11 } From 49133626cde077dda767d395bac773ac886263cd Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 27 Aug 2026 09:22:56 -0400 Subject: [PATCH 3/6] remove x-region test --- sdks/java/io/iceberg/build.gradle | 6 +- .../beam/sdk/io/iceberg/AddFilesIT.java | 76 ++----------------- .../sdk/io/iceberg/BigLakeTestCatalog.java | 10 --- 3 files changed, 7 insertions(+), 85 deletions(-) diff --git a/sdks/java/io/iceberg/build.gradle b/sdks/java/io/iceberg/build.gradle index a008b6d66723..dfe9670d2c36 100644 --- a/sdks/java/io/iceberg/build.gradle +++ b/sdks/java/io/iceberg/build.gradle @@ -181,17 +181,13 @@ task integrationTest(type: Test) { ]) // Multiple-bucket BigLake REST catalog used by RESTCatalogBLMSIT and AddFilesIT (see // BigLakeTestCatalog). Locations: the catalog's default location first, then a restricted - // location in another bucket. The cross-region location must be a bucket in a different - // region than the catalog. Override for runs against another project's catalog: + // location in another bucket. Override for runs against another project's catalog: // -PbiglakeWarehouse=bl://projects/PROJECT/catalogs/CATALOG // -PbiglakeLocations=gs://default-bucket/path,gs://other-bucket/path - // -PbiglakeCrossRegionLocation=gs://bucket-in-another-region/path systemProperty "beam.iceberg.biglake.warehouse", project.findProperty('biglakeWarehouse') ?: 'bl://projects/apache-beam-testing/catalogs/beam-lakehouse-it' systemProperty "beam.iceberg.biglake.locations", project.findProperty('biglakeLocations') ?: 'gs://beam-lakehouse-it,gs://beam-lakehouse-it-added-path' - systemProperty "beam.iceberg.biglake.crossRegionLocation", - project.findProperty('biglakeCrossRegionLocation') ?: 'gs://managed-iceberg-biglake-its/biglake_cross_region' // Connection + storage root for BigQueryManagedTableCrossEngineIT. if (project.findProperty('bqImtConnection') != null) { systemProperty "beam.bq.imt.connection", project.findProperty('bqImtConnection') diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java index f486e2f40979..cd4b68c37302 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java @@ -28,7 +28,6 @@ import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotEquals; import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; import com.google.api.services.storage.model.StorageObject; @@ -115,7 +114,6 @@ public class AddFilesIT { private static final String DATA_LOCATION = BigLakeTestCatalog.defaultLocation(); private static final String DATA_BUCKET = BigLakeTestCatalog.bucketOf(DATA_LOCATION); private static final String DATA_PREFIX = BigLakeTestCatalog.prefixOf(DATA_LOCATION); - private static final String CROSS_REGION_LOCATION = BigLakeTestCatalog.CROSS_REGION_LOCATION; private static final String PROJECT = TestPipeline.testingPipelineOptions().as(GcpOptions.class).getProject(); @Rule public TestName testName = new TestName(); @@ -240,19 +238,12 @@ public void cleanup() { LOG.warn("Failed to clean up Iceberg catalog", e); } - deleteBlobs(DATA_BUCKET, dirName); - deleteBlobs( - BigLakeTestCatalog.bucketOf(CROSS_REGION_LOCATION), - BigLakeTestCatalog.prefixOf(CROSS_REGION_LOCATION) + "/" + dirName); - } - - private void deleteBlobs(String bucket, String prefix) { try { Iterable blobs = - storage.list(bucket, Storage.BlobListOption.prefix(prefix)).getValues(); + storage.list(DATA_BUCKET, Storage.BlobListOption.prefix(dirName)).getValues(); blobs.forEach(b -> storage.delete(b.getBlobId())); } catch (Exception e) { - LOG.warn("Failed to clean up gs://{}/{}", bucket, prefix, e); + LOG.warn("Failed to clean up GCS bucket", e); } } @@ -431,57 +422,6 @@ public void testBatchParquetImportToUIT() throws IOException { testBatchParquetImport(true); } - /** - * The catalog does not police where added files live: data in a bucket it does not manage, even - * in another region, registers and reads fine through Iceberg. BigQuery, however, cannot read - * data files outside the catalog's region, so the cross-engine read of such a table fails. (Files - * in an unmanaged bucket in the catalog's region read fine from BigQuery.) Documents the - * trade-off of adding files in place from a cross-region bucket. - */ - @Test - public void testBatchParquetImportFromCrossRegionBucket() throws IOException { - String crossRegionBucket = BigLakeTestCatalog.bucketOf(CROSS_REGION_LOCATION); - for (String location : BigLakeTestCatalog.LOCATIONS) { - assertNotEquals( - "Test needs a bucket outside the catalog: " + CROSS_REGION_LOCATION, - BigLakeTestCatalog.bucketOf(location), - crossRegionBucket); - } - List writtenFilePaths = - writeParquetFiles( - crossRegionBucket, BigLakeTestCatalog.prefixOf(CROSS_REGION_LOCATION) + "/" + dirName); - - Pipeline p = Pipeline.create(); - PCollectionRowTuple tuple = - p.apply(Create.of(writtenFilePaths)) - .apply( - new AddFiles( - IcebergCatalogConfig.builder().setCatalogProperties(BIGLAKE_PROPS).build(), - destTableId.toString(), - null, - PARTITION_FIELDS, - null, - TABLE_PROPS, - null, - null)); - PAssert.that(tuple.get("errors")).empty(); - p.run().waitUntilFinish(); - - assertTrue(checkTableHasRegisteredParquetFiles(writtenFilePaths)); - checkRecordsInDestinationTable(/* alsoCheckWithBigQueryIO= */ false); - - Pipeline bq = Pipeline.create(); - bq.apply( - Managed.read(Managed.BIGQUERY) - .withConfig( - ImmutableMap.of( - "table", - BigLakeTestCatalog.bigQueryTableSpec( - destTableId.namespace().toString(), destTableId.name())))) - .getSinglePCollection(); - assertThrows(Pipeline.PipelineExecutionException.class, () -> bq.run().waitUntilFinish()); - } - /** * The destination table lives in the catalog's additional location (a second bucket) while the * source parquet files stay in the default one. BigLake pins tables under their namespace's @@ -528,14 +468,10 @@ public void testBatchParquetImportToTableInAdditionalLocation() throws IOExcepti checkRecordsInDestinationTable(/* alsoCheckWithBigQueryIO= */ true); } + /** Writes TEST_ROWS as parquet under the test's data dir and returns the written file paths. */ private List writeParquetFiles() throws IOException { - return writeParquetFiles(DATA_BUCKET, dirName); - } - - /** Writes TEST_ROWS as parquet under gs://{bucket}/{dir}/ and returns the written file paths. */ - private List writeParquetFiles(String bucket, String dir) throws IOException { - String parquetDir = format("gs://%s/%s/", bucket, dir); - String tempDir = format("gs://%s/%s-tmp/", bucket, dir); + String parquetDir = format("gs://%s/%s/", DATA_BUCKET, dirName); + String tempDir = format("gs://%s/%s-tmp/", DATA_BUCKET, dirName); LOG.info("Writing records to the parquet dir"); Pipeline q = Pipeline.create(); org.apache.avro.Schema avroSchema = AvroUtils.toAvroSchema(ROW_SCHEMA); @@ -558,7 +494,7 @@ record -> q.run().waitUntilFinish(); GcsUtil gcsUtil = TestPipeline.testingPipelineOptions().as(GcsOptions.class).getGcsUtil(); - Iterable objects = gcsUtil.listObjects(bucket, dir, null).getItems(); + Iterable objects = gcsUtil.listObjects(DATA_BUCKET, dirName, null).getItems(); List writtenFilePaths = Lists.newArrayList(objects).stream() .map(o -> format("gs://%s/%s", o.getBucket(), o.getName())) diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/BigLakeTestCatalog.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/BigLakeTestCatalog.java index 061babb7d0d5..b1f43bd996c0 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/BigLakeTestCatalog.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/BigLakeTestCatalog.java @@ -43,8 +43,6 @@ *
  • {@code beam.iceberg.biglake.locations}: comma-separated {@code gs://} prefixes the catalog * may write to; the first is the catalog's default location, the rest are additional * restricted locations - *
  • {@code beam.iceberg.biglake.crossRegionLocation}: a {@code gs://} prefix in a bucket - * located in a different region than the catalog * */ public final class BigLakeTestCatalog { @@ -60,14 +58,6 @@ public final class BigLakeTestCatalog { .omitEmptyStrings() .split(requiredProperty("beam.iceberg.biglake.locations"))); - /** - * A location in a bucket the catalog does not manage and that is in a different region than the - * catalog (BigQuery cannot read cross-region data files; see - * AddFilesIT#testBatchParquetImportFromCrossRegionBucket). - */ - public static final String CROSS_REGION_LOCATION = - requiredProperty("beam.iceberg.biglake.crossRegionLocation"); - /** Catalog id, which is also the second segment of BigQuery's 4-part table reference. */ public static final String CATALOG_ID = parseCatalogId(WAREHOUSE); From 59ceb8e1a72828a00d98f49f0da11a1972576cff Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 27 Aug 2026 11:08:07 -0400 Subject: [PATCH 4/6] fix tests, cleanup test artifacts --- .../beam/sdk/io/iceberg/AddFilesIT.java | 21 ++++---- .../sdk/io/iceberg/BigLakeTestCatalog.java | 50 +++++++++++++++++++ .../iceberg/catalog/IcebergCatalogBaseIT.java | 12 ++--- .../io/iceberg/catalog/RESTCatalogBLMSIT.java | 9 ++++ 4 files changed, 74 insertions(+), 18 deletions(-) diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java index cd4b68c37302..63291f3e18bf 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java @@ -193,9 +193,14 @@ public void setup() throws IOException { } salt = System.currentTimeMillis(); + // Object-name prefix of this test's parquet files; DATA_PREFIX is empty for a bare bucket. dirName = format( - "%s/%s-%s/%s", DATA_PREFIX, getClass().getSimpleName(), salt, testName.getMethodName()); + "%s%s-%s/%s", + DATA_PREFIX.isEmpty() ? "" : DATA_PREFIX + "/", + getClass().getSimpleName(), + salt, + testName.getMethodName()); srcTableName = "src_" + testName.getMethodName() + "_" + salt; destTableName = "dest_" + testName.getMethodName() + "_" + salt; srcTableId = TableIdentifier.of(namespace, srcTableName); @@ -206,14 +211,8 @@ public void setup() throws IOException { catalog.createNamespace(Namespace.of(namespace)); } - private void cleanupCatalog() { - for (String name : Arrays.asList(namespace, altNamespace)) { - Namespace ns = Namespace.of(name); - if (catalog.namespaceExists(ns)) { - catalog.listTables(ns).forEach(catalog::dropTable); - catalog.dropNamespace(ns); - } - } + private void cleanupCatalog() throws IOException { + BigLakeTestCatalog.dropNamespacesAndFiles(catalog, Arrays.asList(namespace, altNamespace)); } @After @@ -593,9 +592,7 @@ public void testBatchParquetImportWithSchemaEvolution() throws IOException { GcsUtil gcsUtil = TestPipeline.testingPipelineOptions().as(GcsOptions.class).getGcsUtil(); List writtenFilePaths = - Lists.newArrayList( - gcsUtil.listObjects(DATA_BUCKET, dirName, null).getItems()) - .stream() + Lists.newArrayList(gcsUtil.listObjects(DATA_BUCKET, dirName, null).getItems()).stream() .map(o -> format("gs://%s/%s", o.getBucket(), o.getName())) .collect(Collectors.toList()); assertEquals(20, writtenFilePaths.size()); diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/BigLakeTestCatalog.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/BigLakeTestCatalog.java index b1f43bd996c0..a9cdd6745e19 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/BigLakeTestCatalog.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/BigLakeTestCatalog.java @@ -19,15 +19,24 @@ import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument; +import com.google.api.services.storage.model.StorageObject; +import java.io.IOException; +import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.regex.Matcher; import java.util.regex.Pattern; import org.apache.beam.sdk.extensions.gcp.options.GcpOptions; +import org.apache.beam.sdk.extensions.gcp.options.GcsOptions; +import org.apache.beam.sdk.extensions.gcp.util.GcsUtil; import org.apache.beam.sdk.testing.TestPipeline; 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.iceberg.catalog.Catalog; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.SupportsNamespaces; +import org.apache.iceberg.catalog.TableIdentifier; /** * Test-side description of the BigLake (Lakehouse) Iceberg REST catalog the ITs run against. @@ -124,11 +133,52 @@ public static Map catalogProperties() { .put("uri", "https://biglake.googleapis.com/iceberg/v1/restcatalog") .put("warehouse", WAREHOUSE) .put("header.x-goog-user-project", PROJECT) + // Required by catalogs in vended-credentials mode; ignored in end-user mode. + .put("header.X-Iceberg-Access-Delegation", "vended-credentials") .put("io-impl", "org.apache.iceberg.gcp.gcs.GCSFileIO") .put("rest.auth.type", "org.apache.iceberg.gcp.auth.GoogleAuthManager") .build(); } + /** + * Drops every table in the namespaces and then the namespaces themselves, and deletes the tables' + * files from Cloud Storage: BigLake keeps a dropped table's data and metadata (even with purge), + * and table locations carry a random suffix, so they are captured before the drop. + */ + public static void dropNamespacesAndFiles(Catalog catalog, List namespaces) + throws IOException { + List tableLocations = new ArrayList<>(); + for (String name : namespaces) { + Namespace namespace = Namespace.of(name); + if (!((SupportsNamespaces) catalog).namespaceExists(namespace)) { + continue; + } + for (TableIdentifier identifier : catalog.listTables(namespace)) { + tableLocations.add(catalog.loadTable(identifier).location()); + catalog.dropTable(identifier); + } + ((SupportsNamespaces) catalog).dropNamespace(namespace); + } + for (String location : tableLocations) { + deleteObjects(location); + } + } + + /** Deletes every object under a {@code gs://bucket/prefix} location. */ + public static void deleteObjects(String gcsLocation) throws IOException { + GcsUtil gcsUtil = TestPipeline.testingPipelineOptions().as(GcsOptions.class).getGcsUtil(); + List objects = + gcsUtil.listObjects(bucketOf(gcsLocation), prefixOf(gcsLocation), null).getItems(); + if (objects == null || objects.isEmpty()) { + return; + } + List paths = new ArrayList<>(); + for (StorageObject object : objects) { + paths.add("gs://" + object.getBucket() + "/" + object.getName()); + } + gcsUtil.remove(paths); + } + /** BigQuery's 4-part {@code project.catalog.namespace.table} reference. */ public static String bigQueryTableSpec(String namespace, String table) { return String.format("%s.%s.%s.%s", PROJECT, CATALOG_ID, namespace, table); diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java index 4495c741b155..ed66b215b76a 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java @@ -238,14 +238,14 @@ public void cleanUp() throws Exception { try { GcsUtil gcsUtil = OPTIONS.as(GcsOptions.class).getGcsUtil(); GcsPath path = GcsPath.fromUri(warehouse); + // The warehouse may be a bare bucket (no object path), where getFileName() throws. + String prefix = + path.getObject().isEmpty() + ? getClass().getSimpleName() + : getClass().getSimpleName() + "/" + path.getFileName(); @Nullable List objects = - gcsUtil - .listObjects( - path.getBucket(), - getClass().getSimpleName() + "/" + path.getFileName().toString(), - null) - .getItems(); + gcsUtil.listObjects(path.getBucket(), prefix, null).getItems(); // sometimes a catalog's cleanup will take care of all the files. // If any files are left though, manually delete them with GCS utils diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/RESTCatalogBLMSIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/RESTCatalogBLMSIT.java index 95442455132f..1df2b145fde4 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/RESTCatalogBLMSIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/RESTCatalogBLMSIT.java @@ -81,6 +81,15 @@ public String bigQueryTableSpec(String tableId) { identifier.namespace().toString(), identifier.name()); } + @Override + public void catalogCleanup(List namespaces) throws IOException { + List names = new ArrayList<>(); + for (Namespace namespace : namespaces) { + names.add(namespace.toString()); + } + BigLakeTestCatalog.dropNamespacesAndFiles(catalog, names); + } + @Override public Catalog createCatalog() { RESTCatalog restCatalog = new RESTCatalog(); From c43349dc9f38ab6e7dde885dd48b80c9e55bf442 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 1 Oct 2026 13:58:36 -0400 Subject: [PATCH 5/6] fixes --- .../beam/sdk/io/iceberg/AddFilesIT.java | 31 +++++++++++++++++++ .../sdk/io/iceberg/BigLakeTestCatalog.java | 26 +++++++++++----- .../catalog/IcebergCdcWriteBaseIT.java | 12 +++---- .../catalog/LakehouseCatalogCdcWriteIT.java | 24 ++++++-------- 4 files changed, 65 insertions(+), 28 deletions(-) diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java index 63291f3e18bf..ace770ea6158 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java @@ -30,6 +30,7 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; +import com.google.api.client.http.HttpResponseException; import com.google.api.services.storage.model.StorageObject; import com.google.cloud.storage.Blob; import com.google.cloud.storage.Notification; @@ -188,6 +189,7 @@ public void setup() throws IOException { // try creating it again notification = storage.createNotification(DATA_BUCKET, notificationInfo); } else { + logNotificationFailure(e); throw e; } } @@ -211,6 +213,35 @@ public void setup() throws IOException { catalog.createNamespace(Namespace.of(namespace)); } + /** + * GCS reports server-side errors (5xx) without a cause, so log what identifies the request to GCS + * support and the bucket's notification count at the time. + */ + private void logNotificationFailure(StorageException e) { + @Nullable String requestId = null; + Throwable cause = e.getCause(); + if (cause instanceof HttpResponseException) { + HttpResponseException response = (HttpResponseException) cause; + requestId = response.getHeaders().getFirstHeaderStringValue("x-guploader-uploadid"); + } + String existingNotifications; + try { + existingNotifications = String.valueOf(storage.listNotifications(DATA_BUCKET).size()); + } catch (StorageException listError) { + existingNotifications = "unknown (" + listError.getMessage() + ")"; + } + LOG.error( + "Failed to create a GCS notification on bucket {} for topic {}: HTTP {}, reason={}," + + " retryable={}, x-guploader-uploadid={}, notifications on bucket={}", + DATA_BUCKET, + notificationsTopic, + e.getCode(), + e.getReason(), + e.isRetryable(), + requestId, + existingNotifications); + } + private void cleanupCatalog() throws IOException { BigLakeTestCatalog.dropNamespacesAndFiles(catalog, Arrays.asList(namespace, altNamespace)); } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/BigLakeTestCatalog.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/BigLakeTestCatalog.java index a9cdd6745e19..a26e7038c7ef 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/BigLakeTestCatalog.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/BigLakeTestCatalog.java @@ -142,23 +142,35 @@ public static Map catalogProperties() { /** * Drops every table in the namespaces and then the namespaces themselves, and deletes the tables' - * files from Cloud Storage: BigLake keeps a dropped table's data and metadata (even with purge), - * and table locations carry a random suffix, so they are captured before the drop. + * files from Cloud Storage. */ public static void dropNamespacesAndFiles(Catalog catalog, List namespaces) throws IOException { - List tableLocations = new ArrayList<>(); for (String name : namespaces) { Namespace namespace = Namespace.of(name); if (!((SupportsNamespaces) catalog).namespaceExists(namespace)) { continue; } - for (TableIdentifier identifier : catalog.listTables(namespace)) { - tableLocations.add(catalog.loadTable(identifier).location()); - catalog.dropTable(identifier); - } + dropTablesAndFiles(catalog, name); ((SupportsNamespaces) catalog).dropNamespace(namespace); } + } + + /** + * Drops every table in the namespace and deletes the tables' files from Cloud Storage: BigLake + * keeps a dropped table's data and metadata (even with purge), and table locations carry a random + * suffix, so they are captured before the drop. + */ + public static void dropTablesAndFiles(Catalog catalog, String namespaceName) throws IOException { + Namespace namespace = Namespace.of(namespaceName); + if (!((SupportsNamespaces) catalog).namespaceExists(namespace)) { + return; + } + List tableLocations = new ArrayList<>(); + for (TableIdentifier identifier : catalog.listTables(namespace)) { + tableLocations.add(catalog.loadTable(identifier).location()); + catalog.dropTable(identifier); + } for (String location : tableLocations) { deleteObjects(location); } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCdcWriteBaseIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCdcWriteBaseIT.java index f35fcd829966..d641a0e5f39e 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCdcWriteBaseIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCdcWriteBaseIT.java @@ -236,13 +236,13 @@ public void cleanUp() throws Exception { try { GcsUtil gcsUtil = OPTIONS.as(GcsOptions.class).getGcsUtil(); GcsPath path = GcsPath.fromUri(warehouse); + // The warehouse may be a bare bucket (no object path), where getFileName() throws. + String prefix = + path.getObject().isEmpty() + ? getClass().getSimpleName() + : getClass().getSimpleName() + "/" + path.getFileName(); @Nullable List objects = - gcsUtil - .listObjects( - path.getBucket(), - getClass().getSimpleName() + "/" + path.getFileName().toString(), - null) - .getItems(); + gcsUtil.listObjects(path.getBucket(), prefix, null).getItems(); // A catalog's cleanup sometimes removes every file; delete whatever is left. if (objects != null) { gcsUtil.remove( diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/LakehouseCatalogCdcWriteIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/LakehouseCatalogCdcWriteIT.java index fe0d7d76451e..947ddc3ca24a 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/LakehouseCatalogCdcWriteIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/LakehouseCatalogCdcWriteIT.java @@ -17,7 +17,9 @@ */ package org.apache.beam.sdk.io.iceberg.catalog; +import java.io.IOException; import java.util.Map; +import org.apache.beam.sdk.io.iceberg.BigLakeTestCatalog; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; import org.apache.iceberg.catalog.Catalog; import org.apache.iceberg.rest.RESTCatalog; @@ -28,27 +30,19 @@ public class LakehouseCatalogCdcWriteIT extends IcebergCdcWriteBaseIT { private static Map catalogProps; - private static final String LAKEHOUSE_WAREHOUSE = - System.getProperty("beam.iceberg.biglake.warehouse", "gs://managed-iceberg-biglake-its"); - @BeforeClass public static void setup() { - warehouse = LAKEHOUSE_WAREHOUSE; - catalogProps = - ImmutableMap.builder() - .put("type", "rest") - .put("uri", "https://biglake.googleapis.com/iceberg/v1/restcatalog") - .put("warehouse", LAKEHOUSE_WAREHOUSE) - .put("header.x-goog-user-project", OPTIONS.getProject()) - .put("io-impl", "org.apache.iceberg.gcp.gcs.GCSFileIO") - .put("rest.auth.type", "org.apache.iceberg.gcp.auth.GoogleAuthManager") - .build(); + warehouse = BigLakeTestCatalog.defaultLocation(); + catalogProps = BigLakeTestCatalog.catalogProperties(); } @After - public void after() { + public void after() throws IOException { + // BigLake keeps a dropped table's files, so remove them before the base class drops the + // namespace. + BigLakeTestCatalog.dropTablesAndFiles(catalog, namespace()); // The base class points its cleanup at this warehouse. - warehouse = LAKEHOUSE_WAREHOUSE; + warehouse = BigLakeTestCatalog.defaultLocation(); } @Override From c833c7b9e7e07030d4527402b68516aaaad2e19e Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 1 Oct 2026 21:32:34 -0400 Subject: [PATCH 6/6] rename --- sdks/java/io/iceberg/build.gradle | 16 +++---- .../beam/sdk/io/iceberg/AddFilesIT.java | 47 +++++++++++-------- ...Catalog.java => LakehouseTestCatalog.java} | 20 ++++---- .../iceberg/catalog/IcebergCatalogBaseIT.java | 7 ++- .../catalog/LakehouseCatalogCdcWriteIT.java | 12 ++--- .../io/iceberg/catalog/RESTCatalogBLMSIT.java | 32 ++++++------- 6 files changed, 70 insertions(+), 64 deletions(-) rename sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/{BigLakeTestCatalog.java => LakehouseTestCatalog.java} (92%) diff --git a/sdks/java/io/iceberg/build.gradle b/sdks/java/io/iceberg/build.gradle index dfe9670d2c36..86964472cd81 100644 --- a/sdks/java/io/iceberg/build.gradle +++ b/sdks/java/io/iceberg/build.gradle @@ -179,15 +179,15 @@ task integrationTest(type: Test) { "--project=${gcpProject}", "--tempLocation=${gcpTempLocation}", ]) - // Multiple-bucket BigLake REST catalog used by RESTCatalogBLMSIT and AddFilesIT (see - // BigLakeTestCatalog). Locations: the catalog's default location first, then a restricted + // Multiple-bucket Lakehouse REST catalog used by RESTCatalogBLMSIT and AddFilesIT (see + // LakehouseTestCatalog). Locations: the catalog's default location first, then a restricted // location in another bucket. Override for runs against another project's catalog: - // -PbiglakeWarehouse=bl://projects/PROJECT/catalogs/CATALOG - // -PbiglakeLocations=gs://default-bucket/path,gs://other-bucket/path - systemProperty "beam.iceberg.biglake.warehouse", - project.findProperty('biglakeWarehouse') ?: 'bl://projects/apache-beam-testing/catalogs/beam-lakehouse-it' - systemProperty "beam.iceberg.biglake.locations", - project.findProperty('biglakeLocations') ?: 'gs://beam-lakehouse-it,gs://beam-lakehouse-it-added-path' + // -PlakehouseWarehouse=bl://projects/PROJECT/catalogs/CATALOG + // -PlakehouseLocations=gs://default-bucket/path,gs://other-bucket/path + systemProperty "beam.iceberg.lakehouse.warehouse", + project.findProperty('lakehouseWarehouse') ?: 'bl://projects/apache-beam-testing/catalogs/beam-lakehouse-it' + systemProperty "beam.iceberg.lakehouse.locations", + project.findProperty('lakehouseLocations') ?: 'gs://beam-lakehouse-it,gs://beam-lakehouse-it-added-path' // Connection + storage root for BigQueryManagedTableCrossEngineIT. if (project.findProperty('bqImtConnection') != null) { systemProperty "beam.bq.imt.connection", project.findProperty('bqImtConnection') diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java index ace770ea6158..a1bb79431753 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java @@ -110,11 +110,11 @@ public class AddFilesIT { private static final Logger LOG = LoggerFactory.getLogger(AddFilesIT.class); - // Multiple-bucket BigLake catalog (see BigLakeTestCatalog). Source parquet files, and the + // Multiple-bucket Lakehouse catalog (see LakehouseTestCatalog). Source parquet files, and the // GCS notifications announcing them, live under the catalog's default location. - private static final String DATA_LOCATION = BigLakeTestCatalog.defaultLocation(); - private static final String DATA_BUCKET = BigLakeTestCatalog.bucketOf(DATA_LOCATION); - private static final String DATA_PREFIX = BigLakeTestCatalog.prefixOf(DATA_LOCATION); + private static final String DATA_LOCATION = LakehouseTestCatalog.defaultLocation(); + private static final String DATA_BUCKET = LakehouseTestCatalog.bucketOf(DATA_LOCATION); + private static final String DATA_PREFIX = LakehouseTestCatalog.prefixOf(DATA_LOCATION); private static final String PROJECT = TestPipeline.testingPipelineOptions().as(GcpOptions.class).getProject(); @Rule public TestName testName = new TestName(); @@ -131,7 +131,8 @@ public class AddFilesIT { .addStringField("name") .addStringField("kind") .build(); - private static final Map BIGLAKE_PROPS = BigLakeTestCatalog.catalogProperties(); + private static final Map LAKEHOUSE_PROPS = + LakehouseTestCatalog.catalogProperties(); private Storage storage; private PubsubClient pubsub; private Notification notification; @@ -208,7 +209,7 @@ public void setup() throws IOException { srcTableId = TableIdentifier.of(namespace, srcTableName); destTableId = TableIdentifier.of(namespace, destTableName); - catalog.initialize("test_catalog", BIGLAKE_PROPS); + catalog.initialize("test_catalog", LAKEHOUSE_PROPS); cleanupCatalog(); catalog.createNamespace(Namespace.of(namespace)); } @@ -243,7 +244,7 @@ private void logNotificationFailure(StorageException e) { } private void cleanupCatalog() throws IOException { - BigLakeTestCatalog.dropNamespacesAndFiles(catalog, Arrays.asList(namespace, altNamespace)); + LakehouseTestCatalog.dropNamespacesAndFiles(catalog, Arrays.asList(namespace, altNamespace)); } @After @@ -283,7 +284,7 @@ public void testStreamingImportFromExistingIcebergTable() // first create a source iceberg table catalog.createTable(srcTableId, beamSchemaToIcebergSchema(ROW_SCHEMA), SPEC); - // BigLake may write under {namespace}/{table}/{id}/data/... rather than the Hive-style + // Lakehouse may write under {namespace}/{table}/{id}/data/... rather than the Hive-style // {namespace}/{table}/data/... layout, so match the table prefix and a /data/ segment. String tablePrefix = format("%s/%s/", namespace, srcTableName); @@ -303,7 +304,7 @@ public void testStreamingImportFromExistingIcebergTable() Managed.write(Managed.ICEBERG) .withConfig( ImmutableMap.of( - "table", srcTableId.toString(), "catalog_properties", BIGLAKE_PROPS))); + "table", srcTableId.toString(), "catalog_properties", LAKEHOUSE_PROPS))); q.run().waitUntilFinish(); // check that the destination table has been created @@ -454,17 +455,17 @@ public void testBatchParquetImportToUIT() throws IOException { /** * The destination table lives in the catalog's additional location (a second bucket) while the - * source parquet files stay in the default one. BigLake pins tables under their namespace's + * source parquet files stay in the default one. Lakehouse pins tables under their namespace's * location, so the table is created in a namespace placed in the second bucket; AddFiles must * commit metadata there and reference the files in place across buckets. */ @Test public void testBatchParquetImportToTableInAdditionalLocation() throws IOException { - String namespaceLocation = BigLakeTestCatalog.additionalLocation() + "/" + altNamespace; + String namespaceLocation = LakehouseTestCatalog.additionalLocation() + "/" + altNamespace; assertNotEquals( "Test needs two distinct buckets", DATA_BUCKET, - BigLakeTestCatalog.bucketOf(namespaceLocation)); + LakehouseTestCatalog.bucketOf(namespaceLocation)); catalog.createNamespace( Namespace.of(altNamespace), ImmutableMap.of("location", namespaceLocation)); destTableId = TableIdentifier.of(altNamespace, destTableName); @@ -477,7 +478,7 @@ public void testBatchParquetImportToTableInAdditionalLocation() throws IOExcepti p.apply(Create.of(writtenFilePaths)) .apply( new AddFiles( - IcebergCatalogConfig.builder().setCatalogProperties(BIGLAKE_PROPS).build(), + IcebergCatalogConfig.builder().setCatalogProperties(LAKEHOUSE_PROPS).build(), destTableId.toString(), null, PARTITION_FIELDS, @@ -491,7 +492,7 @@ public void testBatchParquetImportToTableInAdditionalLocation() throws IOExcepti assertTrue(checkTableHasRegisteredParquetFiles(writtenFilePaths)); Table destTable = catalog.loadTable(destTableId); String metadataLocation = ((BaseTable) destTable).operations().current().metadataFileLocation(); - assertThat(metadataLocation, startsWith(BigLakeTestCatalog.additionalLocation())); + assertThat(metadataLocation, startsWith(LakehouseTestCatalog.additionalLocation())); for (String path : writtenFilePaths) { assertThat(path, startsWith("gs://" + DATA_BUCKET + "/")); } @@ -584,7 +585,7 @@ record -> p.apply(Create.of(writtenFilePaths)) .apply( new AddFiles( - IcebergCatalogConfig.builder().setCatalogProperties(BIGLAKE_PROPS).build(), + IcebergCatalogConfig.builder().setCatalogProperties(LAKEHOUSE_PROPS).build(), namespace + "." + destTableName, null, isUIT ? null : PARTITION_FIELDS, @@ -637,7 +638,7 @@ public void testBatchParquetImportWithSchemaEvolution() throws IOException { p.apply(Create.of(writtenFilePaths)) .apply( new AddFiles( - IcebergCatalogConfig.builder().setCatalogProperties(BIGLAKE_PROPS).build(), + IcebergCatalogConfig.builder().setCatalogProperties(LAKEHOUSE_PROPS).build(), namespace + "." + destTableName, null, null, @@ -685,7 +686,10 @@ public void testBatchParquetImportWithSchemaEvolution() throws IOException { Managed.read(Managed.ICEBERG) .withConfig( ImmutableMap.of( - "table", destTableId.toString(), "catalog_properties", BIGLAKE_PROPS))) + "table", + destTableId.toString(), + "catalog_properties", + LAKEHOUSE_PROPS))) .getSinglePCollection() .apply(MapElements.into(strings()).via(AddFilesIT::canonicalRecord)); PAssert.that(destRows) @@ -720,7 +724,10 @@ private void checkRecordsInDestinationTable(boolean alsoCheckWithBigQueryIO) { Managed.read(Managed.ICEBERG) .withConfig( ImmutableMap.of( - "table", destTableId.toString(), "catalog_properties", BIGLAKE_PROPS))) + "table", + destTableId.toString(), + "catalog_properties", + LAKEHOUSE_PROPS))) .getSinglePCollection(); PAssert.that(destRows).containsInAnyOrder(TEST_ROWS); @@ -738,7 +745,7 @@ private void checkRecordsInDestinationTable(boolean alsoCheckWithBigQueryIO) { format( "%s.%s.%s.%s", PROJECT, - BigLakeTestCatalog.CATALOG_ID, + LakehouseTestCatalog.CATALOG_ID, destTableId.namespace(), destTableId.name())))) .getSinglePCollection() @@ -796,7 +803,7 @@ private PipelineResult startAddFilesListener( .apply(Deduplicate.values()) .apply( new AddFiles( - IcebergCatalogConfig.builder().setCatalogProperties(BIGLAKE_PROPS).build(), + IcebergCatalogConfig.builder().setCatalogProperties(LAKEHOUSE_PROPS).build(), namespace + "." + destTableName, null, PARTITION_FIELDS, diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/BigLakeTestCatalog.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/LakehouseTestCatalog.java similarity index 92% rename from sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/BigLakeTestCatalog.java rename to sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/LakehouseTestCatalog.java index a26e7038c7ef..6faac4400b39 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/BigLakeTestCatalog.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/LakehouseTestCatalog.java @@ -39,7 +39,7 @@ import org.apache.iceberg.catalog.TableIdentifier; /** - * Test-side description of the BigLake (Lakehouse) Iceberg REST catalog the ITs run against. + * Test-side description of the Lakehouse Iceberg REST catalog the ITs run against. * *

    The catalog is a multiple-bucket catalog addressed by a {@code * bl://projects/PROJECT/catalogs/CATALOG} warehouse and allowed to store resources under a fixed @@ -48,24 +48,24 @@ * system properties * *

      - *
    • {@code beam.iceberg.biglake.warehouse}: the {@code bl://} warehouse URI - *
    • {@code beam.iceberg.biglake.locations}: comma-separated {@code gs://} prefixes the catalog - * may write to; the first is the catalog's default location, the rest are additional + *
    • {@code beam.iceberg.lakehouse.warehouse}: the {@code bl://} warehouse URI + *
    • {@code beam.iceberg.lakehouse.locations}: comma-separated {@code gs://} prefixes the + * catalog may write to; the first is the catalog's default location, the rest are additional * restricted locations *
    */ -public final class BigLakeTestCatalog { +public final class LakehouseTestCatalog { private static final Pattern WAREHOUSE_PATTERN = Pattern.compile("bl://projects/([^/]+)/catalogs/([^/]+)"); - public static final String WAREHOUSE = requiredProperty("beam.iceberg.biglake.warehouse"); + public static final String WAREHOUSE = requiredProperty("beam.iceberg.lakehouse.warehouse"); public static final List LOCATIONS = ImmutableList.copyOf( Splitter.on(',') .trimResults() .omitEmptyStrings() - .split(requiredProperty("beam.iceberg.biglake.locations"))); + .split(requiredProperty("beam.iceberg.lakehouse.locations"))); /** Catalog id, which is also the second segment of BigQuery's 4-part table reference. */ public static final String CATALOG_ID = parseCatalogId(WAREHOUSE); @@ -73,7 +73,7 @@ public final class BigLakeTestCatalog { private static final String PROJECT = TestPipeline.testingPipelineOptions().as(GcpOptions.class).getProject(); - private BigLakeTestCatalog() {} + private LakehouseTestCatalog() {} private static String requiredProperty(String name) { String value = System.getProperty(name); @@ -108,7 +108,7 @@ public static String defaultLocation() { public static String additionalLocation() { checkArgument( LOCATIONS.size() >= 2, - "beam.iceberg.biglake.locations must list at least two locations, got %s", + "beam.iceberg.lakehouse.locations must list at least two locations, got %s", LOCATIONS); return LOCATIONS.get(1); } @@ -157,7 +157,7 @@ public static void dropNamespacesAndFiles(Catalog catalog, List namespac } /** - * Drops every table in the namespace and deletes the tables' files from Cloud Storage: BigLake + * Drops every table in the namespace and deletes the tables' files from Cloud Storage: Lakehouse * keeps a dropped table's data and metadata (even with purge), and table locations carry a random * suffix, so they are captured before the drop. */ diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java index ed66b215b76a..d4b8970c6049 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java @@ -175,10 +175,9 @@ public abstract class IcebergCatalogBaseIT implements Serializable { /** * Catalogs whose tables are also queryable with BigQuery return the BigQuery table reference for * the given Iceberg table id: either the 4-part {@code project.catalog.namespace.table} form for - * Lakehouse runtime catalog (BigLake metastore REST) tables, or the 3-part {@code - * project.dataset.table} form for the BigQuery metastore federation, where namespaces surface as - * datasets. Returning null (the default) disables the cross-engine read checks in {@link - * #testReadWithBigQueryIO()}. + * Lakehouse runtime catalog (Iceberg REST) tables, or the 3-part {@code project.dataset.table} + * form for the BigQuery metastore federation, where namespaces surface as datasets. Returning + * null (the default) disables the cross-engine read checks in {@link #testReadWithBigQueryIO()}. */ public @Nullable String bigQueryTableSpec(String tableId) { return null; diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/LakehouseCatalogCdcWriteIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/LakehouseCatalogCdcWriteIT.java index 947ddc3ca24a..c5e1efb0098c 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/LakehouseCatalogCdcWriteIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/LakehouseCatalogCdcWriteIT.java @@ -19,7 +19,7 @@ import java.io.IOException; import java.util.Map; -import org.apache.beam.sdk.io.iceberg.BigLakeTestCatalog; +import org.apache.beam.sdk.io.iceberg.LakehouseTestCatalog; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; import org.apache.iceberg.catalog.Catalog; import org.apache.iceberg.rest.RESTCatalog; @@ -32,17 +32,17 @@ public class LakehouseCatalogCdcWriteIT extends IcebergCdcWriteBaseIT { @BeforeClass public static void setup() { - warehouse = BigLakeTestCatalog.defaultLocation(); - catalogProps = BigLakeTestCatalog.catalogProperties(); + warehouse = LakehouseTestCatalog.defaultLocation(); + catalogProps = LakehouseTestCatalog.catalogProperties(); } @After public void after() throws IOException { - // BigLake keeps a dropped table's files, so remove them before the base class drops the + // Lakehouse keeps a dropped table's files, so remove them before the base class drops the // namespace. - BigLakeTestCatalog.dropTablesAndFiles(catalog, namespace()); + LakehouseTestCatalog.dropTablesAndFiles(catalog, namespace()); // The base class points its cleanup at this warehouse. - warehouse = BigLakeTestCatalog.defaultLocation(); + warehouse = LakehouseTestCatalog.defaultLocation(); } @Override diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/RESTCatalogBLMSIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/RESTCatalogBLMSIT.java index 1df2b145fde4..ee809b42ab8c 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/RESTCatalogBLMSIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/RESTCatalogBLMSIT.java @@ -27,7 +27,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; -import org.apache.beam.sdk.io.iceberg.BigLakeTestCatalog; +import org.apache.beam.sdk.io.iceberg.LakehouseTestCatalog; import org.apache.beam.sdk.managed.Managed; import org.apache.beam.sdk.transforms.Create; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; @@ -46,8 +46,8 @@ import org.junit.Test; /** - * Tests for {@link org.apache.iceberg.rest.RESTCatalog} using a multiple-bucket BigLake Metastore - * catalog (see {@link BigLakeTestCatalog}). + * Tests for {@link org.apache.iceberg.rest.RESTCatalog} using a multiple-bucket Lakehouse catalog + * (see {@link LakehouseTestCatalog}). */ public class RESTCatalogBLMSIT extends IcebergCatalogBaseIT { private static Map catalogProps; @@ -56,28 +56,28 @@ public class RESTCatalogBLMSIT extends IcebergCatalogBaseIT { public static void setup() { // The catalog decides where tables go (its default location); the base class only uses // `warehouse` to sweep leftover files, so point it at that location. - warehouse = BigLakeTestCatalog.defaultLocation(); - catalogProps = BigLakeTestCatalog.catalogProperties(); + warehouse = LakehouseTestCatalog.defaultLocation(); + catalogProps = LakehouseTestCatalog.catalogProperties(); } @After public void after() { // making sure the cleanup path is directed at the correct warehouse - warehouse = BigLakeTestCatalog.defaultLocation(); + warehouse = LakehouseTestCatalog.defaultLocation(); } @Override public String type() { - return "biglake"; + return "lakehouse"; } @Override public String bigQueryTableSpec(String tableId) { - // BigQuery surfaces Lakehouse runtime catalog (BigLake metastore REST) tables via 4-part + // BigQuery surfaces Lakehouse runtime catalog (Iceberg REST) tables via 4-part // project.catalog.namespace.table identifiers. Requires the caller to hold biglake.* read // permissions (e.g. roles/biglake.viewer) in addition to the usual BigQuery roles. TableIdentifier identifier = TableIdentifier.parse(tableId); - return BigLakeTestCatalog.bigQueryTableSpec( + return LakehouseTestCatalog.bigQueryTableSpec( identifier.namespace().toString(), identifier.name()); } @@ -87,7 +87,7 @@ public void catalogCleanup(List namespaces) throws IOException { for (Namespace namespace : namespaces) { names.add(namespace.toString()); } - BigLakeTestCatalog.dropNamespacesAndFiles(catalog, names); + LakehouseTestCatalog.dropNamespacesAndFiles(catalog, names); } @Override @@ -107,7 +107,7 @@ public Map managedIcebergConfig(String tableId) { /** * A multiple-bucket catalog may place resources under any of its restricted locations, not just - * its default one. BigLake pins a table under its namespace's location, so the namespace is + * its default one. Lakehouse pins a table under its namespace's location, so the namespace is * created in a second bucket; Beam only receives the table through the catalog and must write * wherever it was placed. */ @@ -115,11 +115,11 @@ public Map managedIcebergConfig(String tableId) { public void testWriteReadTableInAdditionalLocation() throws IOException { String altNamespace = namespace() + "_alt"; String altTableId = altNamespace + ".test_table"; - String namespaceLocation = BigLakeTestCatalog.additionalLocation() + "/" + altNamespace; + String namespaceLocation = LakehouseTestCatalog.additionalLocation() + "/" + altNamespace; assertFalse( "Test needs two distinct buckets", - BigLakeTestCatalog.bucketOf(namespaceLocation) - .equals(BigLakeTestCatalog.bucketOf(BigLakeTestCatalog.defaultLocation()))); + LakehouseTestCatalog.bucketOf(namespaceLocation) + .equals(LakehouseTestCatalog.bucketOf(LakehouseTestCatalog.defaultLocation()))); namespacesToCleanup.add(altNamespace); ((SupportsNamespaces) catalog) .createNamespace( @@ -148,9 +148,9 @@ public void testWriteReadTableInAdditionalLocation() throws IOException { } assertFalse("No data files were written", dataFileLocations.isEmpty()); for (String location : dataFileLocations) { - assertThat(location, startsWith(BigLakeTestCatalog.additionalLocation())); + assertThat(location, startsWith(LakehouseTestCatalog.additionalLocation())); } String metadataLocation = ((BaseTable) table).operations().current().metadataFileLocation(); - assertThat(metadataLocation, startsWith(BigLakeTestCatalog.additionalLocation())); + assertThat(metadataLocation, startsWith(LakehouseTestCatalog.additionalLocation())); } }