diff --git a/.github/labeler.yml b/.github/labeler.yml index 293cb509b429..0eb591af1f29 100644 --- a/.github/labeler.yml +++ b/.github/labeler.yml @@ -148,7 +148,8 @@ GCP: - changed-files: - any-glob-to-any-file: [ 'gcp/**/*', - 'gcp-bundle/**/*' + 'gcp-bundle/**/*', + 'bigquery/**/*' ] DELL: diff --git a/bigquery/src/main/java/org/apache/iceberg/gcp/bigquery/BigQueryMetastoreCatalog.java b/bigquery/src/main/java/org/apache/iceberg/gcp/bigquery/BigQueryMetastoreCatalog.java new file mode 100644 index 000000000000..0954aaae74e4 --- /dev/null +++ b/bigquery/src/main/java/org/apache/iceberg/gcp/bigquery/BigQueryMetastoreCatalog.java @@ -0,0 +1,349 @@ +/* + * 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.iceberg.gcp.bigquery; + +import com.google.api.services.bigquery.model.Dataset; +import com.google.api.services.bigquery.model.DatasetList.Datasets; +import com.google.api.services.bigquery.model.DatasetReference; +import com.google.api.services.bigquery.model.ExternalCatalogDatasetOptions; +import com.google.api.services.bigquery.model.TableReference; +import com.google.cloud.ServiceOptions; +import com.google.cloud.bigquery.BigQueryOptions; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.security.GeneralSecurityException; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; +import org.apache.iceberg.BaseMetastoreCatalog; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.TableMetadata; +import org.apache.iceberg.TableOperations; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.SupportsNamespaces; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.NoSuchNamespaceException; +import org.apache.iceberg.exceptions.NoSuchTableException; +import org.apache.iceberg.hadoop.Configurable; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.base.Strings; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.util.LocationUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Iceberg Bigquery Metastore Catalog implementation. */ +public class BigQueryMetastoreCatalog extends BaseMetastoreCatalog + implements SupportsNamespaces, Configurable { + + // User provided properties. + public static final String PROJECT_ID = "gcp.bigquery.project-id"; + public static final String GCP_LOCATION = "gcp.bigquery.location"; + public static final String LIST_ALL_TABLES = "gcp.bigquery.list-all-tables"; + + private static final Logger LOG = LoggerFactory.getLogger(BigQueryMetastoreCatalog.class); + + private static final String DEFAULT_GCP_LOCATION = "us"; + + private String catalogName; + private Map catalogProperties; + private FileIO fileIO; + private Object conf; + private String projectId; + private String projectLocation; + private BigQueryMetastoreClient client; + private boolean listAllTables; + private String warehouseLocation; + + public BigQueryMetastoreCatalog() {} + + @Override + public void initialize(String name, Map properties) { + Preconditions.checkArgument( + properties.containsKey(PROJECT_ID), + "Invalid GCP project: %s must be specified", + PROJECT_ID); + + this.projectId = properties.get(PROJECT_ID); + this.projectLocation = properties.getOrDefault(GCP_LOCATION, DEFAULT_GCP_LOCATION); + + BigQueryOptions options = + BigQueryOptions.newBuilder() + .setProjectId(projectId) + .setLocation(projectLocation) + .setRetrySettings(ServiceOptions.getDefaultRetrySettings()) + .build(); + + try { + client = new BigQueryMetastoreClientImpl(options); + } catch (IOException e) { + throw new UncheckedIOException("Creating BigQuery client failed", e); + } catch (GeneralSecurityException e) { + throw new RuntimeException("Creating BigQuery client failed due to a security issue", e); + } + + initialize(name, properties, projectId, projectLocation, client); + } + + @VisibleForTesting + void initialize( + String name, + Map properties, + String initialProjectId, + String initialLocation, + BigQueryMetastoreClient bigQueryMetaStoreClient) { + Preconditions.checkArgument(bigQueryMetaStoreClient != null, "Invalid BigQuery client: null"); + this.catalogName = name; + this.catalogProperties = ImmutableMap.copyOf(properties); + this.projectId = initialProjectId; + this.projectLocation = initialLocation; + this.client = bigQueryMetaStoreClient; + + LOG.info("Using BigQuery Metastore Iceberg Catalog: {}", name); + + if (properties.containsKey(CatalogProperties.WAREHOUSE_LOCATION)) { + this.warehouseLocation = + LocationUtil.stripTrailingSlash(properties.get(CatalogProperties.WAREHOUSE_LOCATION)); + } + + this.fileIO = + CatalogUtil.loadFileIO( + properties.getOrDefault( + CatalogProperties.FILE_IO_IMPL, "org.apache.iceberg.io.ResolvingFileIO"), + properties, + conf); + + this.listAllTables = Boolean.parseBoolean(properties.getOrDefault(LIST_ALL_TABLES, "true")); + } + + @Override + protected TableOperations newTableOps(TableIdentifier identifier) { + return new BigQueryTableOperations(client, fileIO, toTableReference(identifier)); + } + + @Override + protected String defaultWarehouseLocation(TableIdentifier identifier) { + String locationUri = null; + DatasetReference datasetReference = toDatasetReference(identifier.namespace()); + Dataset dataset = client.load(datasetReference); + if (dataset != null && dataset.getExternalCatalogDatasetOptions() != null) { + locationUri = dataset.getExternalCatalogDatasetOptions().getDefaultStorageLocationUri(); + } + + return String.format( + "%s/%s", + Strings.isNullOrEmpty(locationUri) + ? createDefaultStorageLocationUri(datasetReference.getDatasetId()) + : LocationUtil.stripTrailingSlash(locationUri), + identifier.name()); + } + + @Override + public List listTables(Namespace namespace) { + validateNamespace(namespace); + + return client.list(toDatasetReference(namespace), listAllTables).stream() + .map( + table -> TableIdentifier.of(namespace.level(0), table.getTableReference().getTableId())) + .collect(ImmutableList.toImmutableList()); + } + + @Override + public boolean dropTable(TableIdentifier identifier, boolean purge) { + try { + TableOperations ops = newTableOps(identifier); + TableMetadata lastMetadata = ops.current(); + + client.delete(toTableReference(identifier)); + + if (purge && lastMetadata != null) { + CatalogUtil.dropTableData(ops.io(), lastMetadata); + } + } catch (NoSuchTableException e) { + return false; + } + + return true; + } + + @Override + public void renameTable(TableIdentifier from, TableIdentifier to) { + // TODO: Enable once supported by BigQuery API. + throw new UnsupportedOperationException("Table rename operation is unsupported."); + } + + @Override + public void createNamespace(Namespace namespace, Map metadata) { + Dataset builder = new Dataset(); + DatasetReference datasetReference = toDatasetReference(namespace); + builder.setLocation(this.projectLocation); + builder.setDatasetReference(datasetReference); + builder.setExternalCatalogDatasetOptions( + BigQueryMetastoreUtils.createExternalCatalogDatasetOptions( + createDefaultStorageLocationUri(datasetReference.getDatasetId()), metadata)); + + client.create(builder); + } + + @Override + public List listNamespaces() { + try { + return listNamespaces(Namespace.empty()); + } catch (NoSuchNamespaceException e) { + return ImmutableList.of(); + } + } + + /** + * Since this catalog only supports one-level namespaces, it always returns an empty list unless + * passed an empty namespace to list all namespaces within the catalog. + */ + @Override + public List listNamespaces(Namespace namespace) { + if (!namespace.isEmpty()) { + throw new NoSuchNamespaceException("Namespace does not exist: %s", namespace); + } + + List allDatasets = client.list(projectId); + + ImmutableList namespaces = + allDatasets.stream().map(this::toNamespace).collect(ImmutableList.toImmutableList()); + + if (namespaces.isEmpty()) { + throw new NoSuchNamespaceException("Namespace does not exist: %s", namespace); + } + + return namespaces; + } + + @Override + public boolean dropNamespace(Namespace namespace) { + try { + client.delete(toDatasetReference(namespace)); + // We don't delete the data folder for safety, which aligns with Hive Metastore's default + // behavior. + // We can support database or catalog level config controlling file deletion in the future. + return true; + } catch (NoSuchNamespaceException e) { + return false; + } + } + + @Override + public boolean setProperties(Namespace namespace, Map properties) { + return client.setParameters(toDatasetReference(namespace), properties); + } + + @Override + public boolean removeProperties(Namespace namespace, Set properties) { + if (!namespaceExists(namespace)) { + throw new NoSuchNamespaceException("Namespace does not exist: %s", namespace); + } + + Preconditions.checkNotNull(properties, "Invalid properties to remove: null"); + + if (properties.isEmpty()) { + return false; + } + + return client.removeParameters(toDatasetReference(namespace), properties); + } + + @Override + public Map loadNamespaceMetadata(Namespace namespace) { + try { + return toMetadata(client.load(toDatasetReference(namespace))); + } catch (IllegalArgumentException e) { + throw new NoSuchNamespaceException("%s", e.getMessage()); + } + } + + @Override + public String name() { + return catalogName; + } + + @Override + protected Map properties() { + return catalogProperties == null ? ImmutableMap.of() : catalogProperties; + } + + @Override + public void setConf(Object conf) { + this.conf = conf; + } + + private String createDefaultStorageLocationUri(String dbId) { + Preconditions.checkArgument( + warehouseLocation != null, + String.format( + "Invalid data warehouse location: %s not set", CatalogProperties.WAREHOUSE_LOCATION)); + return String.format("%s/%s.db", LocationUtil.stripTrailingSlash(warehouseLocation), dbId); + } + + private Namespace toNamespace(Datasets dataset) { + return Namespace.of(dataset.getDatasetReference().getDatasetId()); + } + + private DatasetReference toDatasetReference(Namespace namespace) { + validateNamespace(namespace); + return new DatasetReference().setProjectId(projectId).setDatasetId(namespace.level(0)); + } + + private TableReference toTableReference(TableIdentifier tableIdentifier) { + DatasetReference datasetReference = toDatasetReference(tableIdentifier.namespace()); + return new TableReference() + .setProjectId(datasetReference.getProjectId()) + .setDatasetId(datasetReference.getDatasetId()) + .setTableId(tableIdentifier.name()); + } + + private Map toMetadata(Dataset dataset) { + ExternalCatalogDatasetOptions options = dataset.getExternalCatalogDatasetOptions(); + Map metadata = Maps.newHashMap(); + if (options != null) { + if (options.getParameters() != null) { + metadata.putAll(options.getParameters()); + } + + if (!Strings.isNullOrEmpty(options.getDefaultStorageLocationUri())) { + metadata.put("location", options.getDefaultStorageLocationUri()); + } + } + + return metadata; + } + + private void validateNamespace(Namespace namespace) { + Preconditions.checkArgument( + namespace.levels().length == 1, + String.format( + Locale.ROOT, + "BigQuery Metastore only supports single level namespaces. Invalid namespace: \"%s\" has %s" + + " levels", + namespace, + namespace.levels().length)); + } +} diff --git a/bigquery/src/main/java/org/apache/iceberg/gcp/bigquery/BigQueryMetastoreClient.java b/bigquery/src/main/java/org/apache/iceberg/gcp/bigquery/BigQueryMetastoreClient.java new file mode 100644 index 000000000000..19489e04b535 --- /dev/null +++ b/bigquery/src/main/java/org/apache/iceberg/gcp/bigquery/BigQueryMetastoreClient.java @@ -0,0 +1,127 @@ +/* + * 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.iceberg.gcp.bigquery; + +import com.google.api.services.bigquery.model.Dataset; +import com.google.api.services.bigquery.model.DatasetList.Datasets; +import com.google.api.services.bigquery.model.DatasetReference; +import com.google.api.services.bigquery.model.Table; +import com.google.api.services.bigquery.model.TableList.Tables; +import com.google.api.services.bigquery.model.TableReference; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * A client of Google BigQuery Metastore functions over the BigQuery service. Uses the Google + * BigQuery API. + */ +interface BigQueryMetastoreClient { + + /** + * Creates and returns a new dataset. + * + * @param dataset the dataset to create + */ + Dataset create(Dataset dataset); + + /** + * Returns a dataset. + * + * @param datasetReference full dataset reference + */ + Dataset load(DatasetReference datasetReference); + + /** + * Deletes a dataset. + * + * @param datasetReference full dataset reference + */ + void delete(DatasetReference datasetReference); + + /** + * Sets (Adds or Overwrites) the specified parameters on the Dataset. + * + *

Loads the dataset, compares the parameters, and performs a retrying-update ONLY if the + * parameters will change. + * + * @param datasetReference Reference to the Dataset. + * @param parameters Map of parameters to add/overwrite. + * @return {@code true} if the Dataset was updated, {@code false} if no changes were needed. + */ + boolean setParameters(DatasetReference datasetReference, Map parameters); + + /** + * Removes the specified parameters from the Dataset. + * + *

Loads the dataset, compares the parameters, and performs a retrying-update ONLY if the + * parameters will change as a result. + * + * @param datasetReference Reference to the Dataset. + * @param parameters Set of parameter keys to remove. + * @return {@code true} if the Dataset was updated, {@code false} if no changes were needed (e.g. + * keys did not exist). + */ + boolean removeParameters(DatasetReference datasetReference, Set parameters); + + /** + * Lists datasets under a given project + * + * @param projectId the identifier of the project to list datasets under + */ + List list(String projectId); + + /** + * Creates and returns a new table. + * + * @param table body of the table to create + */ + Table create(Table table); + + /** + * Returns a table. + * + * @param tableReference full table reference + */ + Table load(TableReference tableReference); + + /** + * Updates the catalog table options of an Iceberg table and returns the updated table. + * + * @param tableReference full table reference + * @param table to patch + */ + Table update(TableReference tableReference, Table table); + + /** + * Deletes a table. + * + * @param tableReference full table reference + */ + void delete(TableReference tableReference); + + /** + * Returns all tables in a database. + * + * @param datasetReference full dataset reference + * @param listAllTables if true, fetches every item on the list including unsupported Iceberg + * Tables. If false, unsupported Iceberg Tables will be filtered out. + */ + List list(DatasetReference datasetReference, boolean listAllTables); +} diff --git a/bigquery/src/main/java/org/apache/iceberg/gcp/bigquery/BigQueryMetastoreClientImpl.java b/bigquery/src/main/java/org/apache/iceberg/gcp/bigquery/BigQueryMetastoreClientImpl.java new file mode 100644 index 000000000000..b8240603018f --- /dev/null +++ b/bigquery/src/main/java/org/apache/iceberg/gcp/bigquery/BigQueryMetastoreClientImpl.java @@ -0,0 +1,660 @@ +/* + * 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.iceberg.gcp.bigquery; + +import com.google.api.client.googleapis.javanet.GoogleNetHttpTransport; +import com.google.api.client.googleapis.json.GoogleJsonError; +import com.google.api.client.googleapis.json.GoogleJsonResponseException; +import com.google.api.client.http.HttpHeaders; +import com.google.api.client.http.HttpResponse; +import com.google.api.client.http.HttpResponseException; +import com.google.api.client.http.HttpStatusCodes; +import com.google.api.client.json.gson.GsonFactory; +import com.google.api.client.util.Data; +import com.google.api.services.bigquery.Bigquery; +import com.google.api.services.bigquery.BigqueryScopes; +import com.google.api.services.bigquery.model.Dataset; +import com.google.api.services.bigquery.model.DatasetList; +import com.google.api.services.bigquery.model.DatasetList.Datasets; +import com.google.api.services.bigquery.model.DatasetReference; +import com.google.api.services.bigquery.model.ExternalCatalogDatasetOptions; +import com.google.api.services.bigquery.model.ExternalCatalogTableOptions; +import com.google.api.services.bigquery.model.Table; +import com.google.api.services.bigquery.model.TableList; +import com.google.api.services.bigquery.model.TableList.Tables; +import com.google.api.services.bigquery.model.TableReference; +import com.google.api.services.bigquery.model.TableSchema; +import com.google.auth.http.HttpCredentialsAdapter; +import com.google.auth.oauth2.GoogleCredentials; +import com.google.cloud.BaseServiceException; +import com.google.cloud.ExceptionHandler; +import com.google.cloud.bigquery.BigQueryErrorMessages; +import com.google.cloud.bigquery.BigQueryOptions; +import com.google.cloud.bigquery.BigQueryRetryConfig; +import com.google.cloud.bigquery.BigQueryRetryHelper; +import java.io.IOException; +import java.security.GeneralSecurityException; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.Stream; +import org.apache.iceberg.BaseMetastoreTableOperations; +import org.apache.iceberg.exceptions.AlreadyExistsException; +import org.apache.iceberg.exceptions.BadRequestException; +import org.apache.iceberg.exceptions.ForbiddenException; +import org.apache.iceberg.exceptions.NamespaceNotEmptyException; +import org.apache.iceberg.exceptions.NoSuchIcebergTableException; +import org.apache.iceberg.exceptions.NoSuchNamespaceException; +import org.apache.iceberg.exceptions.NoSuchTableException; +import org.apache.iceberg.exceptions.NotAuthorizedException; +import org.apache.iceberg.exceptions.RuntimeIOException; +import org.apache.iceberg.exceptions.ServiceFailureException; +import org.apache.iceberg.exceptions.ServiceUnavailableException; +import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; + +/** A client of Google Bigquery Metastore functions over the BigQuery service. */ +public final class BigQueryMetastoreClientImpl implements BigQueryMetastoreClient { + + private final Bigquery client; + private final BigQueryOptions bigqueryOptions; + + public static final ExceptionHandler.Interceptor EXCEPTION_HANDLER_INTERCEPTOR = + new ExceptionHandler.Interceptor() { + + @Override + public RetryResult afterEval(Exception exception, RetryResult retryResult) { + return ExceptionHandler.Interceptor.RetryResult.CONTINUE_EVALUATION; + } + + @Override + public RetryResult beforeEval(Exception exception) { + if (exception instanceof BaseServiceException) { + boolean retriable = ((BaseServiceException) exception).isRetryable(); + return retriable + ? ExceptionHandler.Interceptor.RetryResult.RETRY + : ExceptionHandler.Interceptor.RetryResult.CONTINUE_EVALUATION; + } + + return ExceptionHandler.Interceptor.RetryResult.CONTINUE_EVALUATION; + } + }; + + // Retry config with error messages and regex for rate limit exceeded errors. + private static final BigQueryRetryConfig DEFAULT_RETRY_CONFIG = + BigQueryRetryConfig.newBuilder() + .retryOnMessage(BigQueryErrorMessages.RATE_LIMIT_EXCEEDED_MSG) + .retryOnMessage(BigQueryErrorMessages.JOB_RATE_LIMIT_EXCEEDED_MSG) + .retryOnRegEx(BigQueryErrorMessages.RetryRegExPatterns.RATE_LIMIT_EXCEEDED_REGEX) + .build(); + + public static final ExceptionHandler BIGQUERY_EXCEPTION_HANDLER = + ExceptionHandler.newBuilder() + .abortOn(RuntimeException.class) + // Retry on connection failures due to transient network issues. + .retryOn(java.net.ConnectException.class) + // Retry to recover from temporary DNS resolution failures. + .retryOn(java.net.UnknownHostException.class) + .addInterceptors(EXCEPTION_HANDLER_INTERCEPTOR) + .build(); + + /** Constructs a client of the Google BigQuery service. */ + public BigQueryMetastoreClientImpl(BigQueryOptions options) + throws IOException, GeneralSecurityException { + // Initialize client that will be used to send requests. This client only needs to be created + // once, and can be reused for multiple requests + HttpCredentialsAdapter httpCredentialsAdapter = + new HttpCredentialsAdapter( + GoogleCredentials.getApplicationDefault().createScoped(BigqueryScopes.all())); + this.client = + new Bigquery.Builder( + GoogleNetHttpTransport.newTrustedTransport(), + GsonFactory.getDefaultInstance(), + httpRequest -> { + httpCredentialsAdapter.initialize(httpRequest); + // Instead of throwing exceptions, analyze the HttpResponse object and inspect its + // status code. This will allow BigQuery API errors to be converted into Iceberg + // exceptions. + httpRequest.setThrowExceptionOnExecuteError(false); + }) + .setApplicationName("BigQuery Metastore Iceberg Catalog Plugin") + .build(); + this.bigqueryOptions = options; + } + + @Override + public Dataset create(Dataset dataset) { + Dataset response = null; + try { + response = + BigQueryRetryHelper.runWithRetries( + () -> internalCreate(dataset), + bigqueryOptions.getRetrySettings(), + BIGQUERY_EXCEPTION_HANDLER, + bigqueryOptions.getClock(), + DEFAULT_RETRY_CONFIG); + } catch (BigQueryRetryHelper.BigQueryRetryHelperException e) { + handleBigQueryRetryException(e); + } + return response; + } + + private Dataset internalCreate(Dataset dataset) { + try { + HttpResponse response = + client + .datasets() + .insert(dataset.getDatasetReference().getProjectId(), dataset) + .executeUnparsed(); + return convertExceptionIfUnsuccessful(response).parseAs(Dataset.class); + } catch (IOException e) { + throw new RuntimeIOException(e); + } catch (AlreadyExistsException e) { + throw new AlreadyExistsException("Namespace already exists: %s", dataset.getId()); + } + } + + @Override + public Dataset load(DatasetReference datasetReference) { + try { + HttpResponse response = + client + .datasets() + .get(datasetReference.getProjectId(), datasetReference.getDatasetId()) + .executeUnparsed(); + if (response.getStatusCode() == HttpStatusCodes.STATUS_CODE_NOT_FOUND) { + throw new NoSuchNamespaceException( + "Namespace does not exist: %s", datasetReference.getDatasetId()); + } + + return convertExceptionIfUnsuccessful(response).parseAs(Dataset.class); + } catch (IOException e) { + throw new RuntimeIOException(e); + } + } + + @Override + public void delete(DatasetReference datasetReference) { + try { + HttpResponse response = + client + .datasets() + .delete(datasetReference.getProjectId(), datasetReference.getDatasetId()) + .executeUnparsed(); + if (response.getStatusCode() == HttpStatusCodes.STATUS_CODE_NOT_FOUND) { + throw new NoSuchNamespaceException( + "Namespace does not exist: %s", datasetReference.getDatasetId()); + } + + convertExceptionIfUnsuccessful(response); + } catch (IOException e) { + throw new RuntimeIOException(e); + } catch (NamespaceNotEmptyException e) { + throw new NamespaceNotEmptyException( + "%s is not empty: %s", datasetReference.getDatasetId(), e.getMessage()); + } + } + + @Override + public boolean setParameters(DatasetReference datasetReference, Map parameters) { + Dataset dataset = load(datasetReference); + ExternalCatalogDatasetOptions existingOptions = dataset.getExternalCatalogDatasetOptions(); + + Map existingParameters = + (existingOptions == null || existingOptions.getParameters() == null) + ? Maps.newHashMap() // Use HashMap to allow modification below + : Maps.newHashMap(existingOptions.getParameters()); // Copy to compare later + + // Calculate what the new parameters would be. + Map newParameters = Maps.newHashMap(existingParameters); + newParameters.putAll(parameters); + + if (Objects.equals(existingParameters, newParameters)) { + // No change in parameters detected + return false; + } + + ExternalCatalogDatasetOptions optionsToUpdate = + existingOptions == null ? new ExternalCatalogDatasetOptions() : existingOptions; + + dataset.setExternalCatalogDatasetOptions(optionsToUpdate.setParameters(newParameters)); + + try { + BigQueryRetryHelper.runWithRetries( + () -> internalUpdate(dataset), + bigqueryOptions.getRetrySettings(), + BIGQUERY_EXCEPTION_HANDLER, + bigqueryOptions.getClock(), + DEFAULT_RETRY_CONFIG); + } catch (BigQueryRetryHelper.BigQueryRetryHelperException e) { + handleBigQueryRetryException(e); + } + + return true; + } + + @Override + public boolean removeParameters(DatasetReference datasetReference, Set parameters) { + Dataset dataset = load(datasetReference); // Throws NoSuchNamespaceException if not found. + + ExternalCatalogDatasetOptions existingOptions = dataset.getExternalCatalogDatasetOptions(); + + if (existingOptions == null + || existingOptions.getParameters() == null + || existingOptions.getParameters().isEmpty()) { + return false; + } + + Map existingParameters = Maps.newHashMap(existingOptions.getParameters()); + + // Calculate the new parameters map. + Map newParameters = Maps.newHashMap(existingParameters); + parameters.forEach(newParameters::remove); + + if (Objects.equals(existingParameters, newParameters)) { + // No change in parameters detected (e.g., keys to remove didn't exist) + return false; + } + + dataset.setExternalCatalogDatasetOptions(existingOptions.setParameters(newParameters)); + + try { + BigQueryRetryHelper.runWithRetries( + () -> internalUpdate(dataset), + bigqueryOptions.getRetrySettings(), + BIGQUERY_EXCEPTION_HANDLER, + bigqueryOptions.getClock(), + DEFAULT_RETRY_CONFIG); + } catch (BigQueryRetryHelper.BigQueryRetryHelperException e) { + handleBigQueryRetryException(e); + } + return true; + } + + @Override + public List list(String projectId) { + try { + String nextPageToken = null; + List datasets = Lists.newArrayList(); + do { + HttpResponse pageResponse = + client.datasets().list(projectId).setPageToken(nextPageToken).executeUnparsed(); + DatasetList result = + convertExceptionIfUnsuccessful(pageResponse).parseAs(DatasetList.class); + nextPageToken = result.getNextPageToken(); + if (result.getDatasets() != null) { + datasets.addAll(result.getDatasets()); + } + + } while (nextPageToken != null && !nextPageToken.isEmpty()); + return datasets; + } catch (IOException e) { + throw new RuntimeIOException(e); + } + } + + @Override + public Table create(Table table) { + // Ensure it is an Iceberg table supported by the BigQuery metastore catalog. + validateTable(table); + // TODO: Ensure table creation is idempotent when handling retries. + Table response = null; + try { + response = + BigQueryRetryHelper.runWithRetries( + () -> internalCreate(table), + bigqueryOptions.getRetrySettings(), + BIGQUERY_EXCEPTION_HANDLER, + bigqueryOptions.getClock(), + DEFAULT_RETRY_CONFIG); + } catch (BigQueryRetryHelper.BigQueryRetryHelperException e) { + handleBigQueryRetryException(e); + } + + return response; + } + + private Table internalCreate(Table table) { + try { + HttpResponse response = + client + .tables() + .insert( + Preconditions.checkNotNull(table.getTableReference()).getProjectId(), + Preconditions.checkNotNull(table.getTableReference()).getDatasetId(), + table) + .executeUnparsed(); + return convertExceptionIfUnsuccessful(response).parseAs(Table.class); + } catch (IOException e) { + throw new RuntimeIOException("%s", e); + } catch (AlreadyExistsException e) { + throw new AlreadyExistsException(e, "Table already exists: %s", table); + } + } + + @Override + public Table load(TableReference tableReference) { + try { + HttpResponse response = + client + .tables() + .get( + tableReference.getProjectId(), + tableReference.getDatasetId(), + tableReference.getTableId()) + .executeUnparsed(); + if (response.getStatusCode() == HttpStatusCodes.STATUS_CODE_NOT_FOUND) { + throw new NoSuchTableException("%s", response.getStatusMessage()); + } + + return validateTable(convertExceptionIfUnsuccessful(response).parseAs(Table.class)); + } catch (IOException e) { + throw new RuntimeIOException("%s", e); + } + } + + @Override + public Table update(TableReference tableReference, Table table) { + // Ensure it is an Iceberg table supported by the BQ metastore catalog. + validateTable(table); + + ExternalCatalogTableOptions newExternalCatalogTableOptions = + new ExternalCatalogTableOptions() + .setStorageDescriptor(table.getExternalCatalogTableOptions().getStorageDescriptor()) + .setConnectionId(table.getExternalCatalogTableOptions().getConnectionId()) + .setParameters(table.getExternalCatalogTableOptions().getParameters()); + Table updatedTable = + new Table() + .setExternalCatalogTableOptions(newExternalCatalogTableOptions) + // Must set the schema as null for using schema auto-detect. + .setSchema(Data.nullOf(TableSchema.class)); + + Table response = null; + try { + response = + BigQueryRetryHelper.runWithRetries( + () -> internalUpdate(tableReference, updatedTable, table.getEtag()), + bigqueryOptions.getRetrySettings(), + BIGQUERY_EXCEPTION_HANDLER, + bigqueryOptions.getClock(), + DEFAULT_RETRY_CONFIG); + } catch (BigQueryRetryHelper.BigQueryRetryHelperException e) { + handleBigQueryRetryException(e); + } + + return response; + } + + private Table internalUpdate(TableReference tableReference, Table table, String etag) { + try { + HttpResponse response = + client + .tables() + .patch( + tableReference.getProjectId(), + tableReference.getDatasetId(), + tableReference.getTableId(), + table) + .setRequestHeaders(new HttpHeaders().setIfMatch(etag)) + .executeUnparsed(); + + if (response.getStatusCode() == HttpStatusCodes.STATUS_CODE_NOT_FOUND) { + String responseString = response.parseAsString(); + if (responseString.toLowerCase(Locale.ENGLISH).contains("not found: connection")) { + throw new BadRequestException("%s", responseString); + } + + throw new NoSuchTableException("%s", response.getStatusMessage()); + } + + return convertExceptionIfUnsuccessful(response).parseAs(Table.class); + } catch (IOException e) { + throw new RuntimeIOException("%s", e); + } + } + + @Override + public void delete(TableReference tableReference) { + try { + load(tableReference); // Fetching it to validate it is a BigQuery Metastore table first + + HttpResponse response = + client + .tables() + .delete( + tableReference.getProjectId(), + tableReference.getDatasetId(), + tableReference.getTableId()) + .executeUnparsed(); + + if (response.getStatusCode() == HttpStatusCodes.STATUS_CODE_NOT_FOUND) { + throw new NoSuchTableException("%s", response.getStatusMessage()); + } + + convertExceptionIfUnsuccessful(response); + } catch (IOException e) { + throw new RuntimeIOException("%s", e); + } + } + + @Override + public List list(DatasetReference datasetReference, boolean listAllTables) { + try { + String nextPageToken = null; + Stream tablesStream = Stream.empty(); + do { + HttpResponse pageResponse = + client + .tables() + .list(datasetReference.getProjectId(), datasetReference.getDatasetId()) + .setPageToken(nextPageToken) + .executeUnparsed(); + if (pageResponse.getStatusCode() == HttpStatusCodes.STATUS_CODE_NOT_FOUND) { + throw new NoSuchNamespaceException("%s", pageResponse.getStatusMessage()); + } + TableList result = convertExceptionIfUnsuccessful(pageResponse).parseAs(TableList.class); + nextPageToken = result.getNextPageToken(); + List tablesPage = result.getTables(); + Stream tablesPageStream = + tablesPage == null ? Stream.empty() : result.getTables().stream(); + tablesStream = Stream.concat(tablesStream, tablesPageStream); + } while (nextPageToken != null && !nextPageToken.isEmpty()); + + // The server should return more metadata here (e.g. BigQuery Non Iceberg tables) to + // distinguish Iceberg + // tables for us to filter out those results since invoking `getTable` on them would + // correctly raise a `NoSuchIcebergTableException` for being inoperable by this plugin. + if (!listAllTables) { + tablesStream = + tablesStream + .parallel() + .filter( + table -> { + try { + load(table.getTableReference()); + } catch (NoSuchTableException e) { + return false; + } + return true; + }); + } + + return tablesStream.collect(Collectors.toList()); + } catch (IOException e) { + throw new RuntimeIOException("%s", e); + } + } + + private Dataset internalUpdate(Dataset dataset) { + Preconditions.checkArgument( + dataset.getDatasetReference() != null, "Dataset Reference can not be null!"); + Preconditions.checkArgument( + dataset.getDatasetReference().getDatasetId() != null, "Dataset Id can not be null!"); + + try { + HttpResponse response = + client + .datasets() + .update( + dataset.getDatasetReference().getProjectId(), + dataset.getDatasetReference().getDatasetId(), + dataset) + .setRequestHeaders(new HttpHeaders().setIfMatch(dataset.getEtag())) + .executeUnparsed(); + if (response.getStatusCode() == HttpStatusCodes.STATUS_CODE_NOT_FOUND) { + throw new NoSuchNamespaceException("%s", response.getStatusMessage()); + } + + return convertExceptionIfUnsuccessful(response).parseAs(Dataset.class); + } catch (IOException e) { + throw new RuntimeIOException("%s", e); + } + } + + // private Dataset internalUpdate(Dataset dataset) { + // Preconditions.checkArgument( + // dataset.getDatasetReference() != null, "Dataset Reference can not be null!"); + // Preconditions.checkArgument( + // dataset.getDatasetReference().getDatasetId() != null, "Dataset Id can not be null!"); + // + // try { + // HttpResponse response = + // client + // .datasets() + // .update( + // dataset.getDatasetReference().getProjectId(), + // dataset.getDatasetReference().getDatasetId(), + // dataset) + // .setRequestHeaders(new HttpHeaders().setIfMatch(dataset.getEtag())) + // .executeUnparsed(); + // if (response.getStatusCode() == HttpStatusCodes.STATUS_CODE_NOT_FOUND) { + // throw new NoSuchNamespaceException("%s", response.getStatusMessage()); + // } + // + // return convertExceptionIfUnsuccessful(response).parseAs(Dataset.class); + // } catch (IOException e) { + // throw new RuntimeIOException("%s", e); + // } + // } + + /** + * Checks if the given table represents a BigQuery Metastore Iceberg table. A table is considered + * an Iceberg table if it has ExternalCatalogTableOptions, a non-empty parameters map containing + * both METADATA_LOCATION_PROP and TABLE_TYPE_PROP with the value ICEBERG_TABLE_TYPE_VALUE. + * + * @param table The table to check. + * @return true if the table is a BigQuery Metastore Iceberg table, false otherwise. + */ + private static boolean isValidIcebergTable(Table table) { + if (table.getExternalCatalogTableOptions() == null + || table.getExternalCatalogTableOptions().isEmpty() + || table.getExternalCatalogTableOptions().getParameters() == null) { + return false; + } + + java.util.Map parameters = + table.getExternalCatalogTableOptions().getParameters(); + + if (!parameters.containsKey(BaseMetastoreTableOperations.METADATA_LOCATION_PROP) + || !parameters.containsKey(BaseMetastoreTableOperations.TABLE_TYPE_PROP)) { + return false; + } + + String tableType = parameters.get(BaseMetastoreTableOperations.TABLE_TYPE_PROP); + return BaseMetastoreTableOperations.ICEBERG_TABLE_TYPE_VALUE.equalsIgnoreCase(tableType); + } + + private Table validateTable(Table table) { + if (!isValidIcebergTable(table)) { + throw new NoSuchIcebergTableException("This table is not a valid Iceberg table: %s", table); + } + + return table; + } + + /** + * Converts BigQuery generic API errors to Iceberg exceptions, *without* handling the + * resource-specific exceptions like NoSuchTableException, NoSuchNamespaceException, etc. + */ + private static HttpResponse convertExceptionIfUnsuccessful(HttpResponse response) + throws IOException { + if (response.isSuccessStatusCode()) { + return response; + } + + GoogleJsonResponseException exception = + GoogleJsonResponseException.from(GsonFactory.getDefaultInstance(), response); + String errorMessage = + exception.getStatusMessage() + + (exception.getContent() != null ? "\n" + exception.getContent() : ""); + + switch (response.getStatusCode()) { + case HttpStatusCodes.STATUS_CODE_UNAUTHORIZED: + throw new NotAuthorizedException( + "Not authorized to call the BigQuery API or access this resource: %s", errorMessage); + case HttpStatusCodes.STATUS_CODE_BAD_REQUEST: + GoogleJsonError errorDetails = exception.getDetails(); + if (errorDetails != null) { + List errors = errorDetails.getErrors(); + if (errors != null) { + for (GoogleJsonError.ErrorInfo errorInfo : errors) { + if (errorInfo.getReason().equals("resourceInUse")) { + throw new NamespaceNotEmptyException("%s", errorInfo.getMessage()); + } + } + } + } + throw new BadRequestException("%s", errorMessage); + case HttpStatusCodes.STATUS_CODE_FORBIDDEN: + throw new ForbiddenException("%s", errorMessage); + case HttpStatusCodes.STATUS_CODE_PRECONDITION_FAILED: + throw new ValidationException("%s", errorMessage); + case HttpStatusCodes.STATUS_CODE_NOT_FOUND: + throw new IllegalArgumentException(errorMessage); + case HttpStatusCodes.STATUS_CODE_SERVER_ERROR: + throw new ServiceFailureException("%s", errorMessage); + case HttpStatusCodes.STATUS_CODE_SERVICE_UNAVAILABLE: + throw new ServiceUnavailableException("%s", errorMessage); + case HttpStatusCodes.STATUS_CODE_CONFLICT: + throw new AlreadyExistsException("%s", errorMessage); + default: + throw new HttpResponseException(response); + } + } + + /** + * Translates BigQueryRetryHelperException to the RuntimeException that caused the error. This + * method will always throw an exception. + */ + private static void handleBigQueryRetryException( + BigQueryRetryHelper.BigQueryRetryHelperException retryException) { + Throwable cause = retryException.getCause(); + String message = retryException.getMessage(); + if (cause instanceof RuntimeException) { + throw (RuntimeException) cause; + } else { + throw new RuntimeException(message, cause); + } + } +} diff --git a/bigquery/src/main/java/org/apache/iceberg/gcp/bigquery/BigQueryMetastoreUtils.java b/bigquery/src/main/java/org/apache/iceberg/gcp/bigquery/BigQueryMetastoreUtils.java new file mode 100644 index 000000000000..cc0d7b279feb --- /dev/null +++ b/bigquery/src/main/java/org/apache/iceberg/gcp/bigquery/BigQueryMetastoreUtils.java @@ -0,0 +1,75 @@ +/* + * 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.iceberg.gcp.bigquery; + +import com.google.api.services.bigquery.model.ExternalCatalogDatasetOptions; +import com.google.api.services.bigquery.model.ExternalCatalogTableOptions; +import com.google.api.services.bigquery.model.SerDeInfo; +import com.google.api.services.bigquery.model.StorageDescriptor; +import java.util.Map; + +/** Shared utilities for BigQuery Metastore specific functions and constants. */ +final class BigQueryMetastoreUtils { + + private BigQueryMetastoreUtils() {} + + private static final String HIVE_SERIALIZATION_LIBRARY = + "org.apache.iceberg.mr.hive.HiveIcebergSerDe"; + private static final String HIVE_FILE_INPUT_FORMAT = + "org.apache.iceberg.mr.hive.HiveIcebergInputFormat"; + private static final String HIVE_FILE_OUTPUT_FORMAT = + "org.apache.iceberg.mr.hive.HiveIcebergOutputFormat"; + + /** + * Creates a new ExternalCatalogTableOptions object populated with the supported library constants + * and parameters given. + * + * @param locationUri storage location uri + * @param parameters table metadata parameters + */ + public static ExternalCatalogTableOptions createExternalCatalogTableOptions( + String locationUri, Map parameters) { + SerDeInfo serDeInfo = new SerDeInfo().setSerializationLibrary(HIVE_SERIALIZATION_LIBRARY); + + StorageDescriptor storageDescriptor = + new StorageDescriptor() + .setLocationUri(locationUri) + .setInputFormat(HIVE_FILE_INPUT_FORMAT) + .setOutputFormat(HIVE_FILE_OUTPUT_FORMAT) + .setSerdeInfo(serDeInfo); + + return new ExternalCatalogTableOptions() + .setStorageDescriptor(storageDescriptor) + .setParameters(parameters); + } + + /** + * Creates a new ExternalCatalogDatasetOptions object populated with the supported library + * constants and parameters given. + * + * @param defaultStorageLocationUri dataset's default location uri + * @param metadataParameters metadata parameters for the dataset + */ + public static ExternalCatalogDatasetOptions createExternalCatalogDatasetOptions( + String defaultStorageLocationUri, Map metadataParameters) { + return new ExternalCatalogDatasetOptions() + .setDefaultStorageLocationUri(defaultStorageLocationUri) + .setParameters(metadataParameters); + } +} diff --git a/bigquery/src/main/java/org/apache/iceberg/gcp/bigquery/BigQueryTableOperations.java b/bigquery/src/main/java/org/apache/iceberg/gcp/bigquery/BigQueryTableOperations.java new file mode 100644 index 000000000000..d57aab50530a --- /dev/null +++ b/bigquery/src/main/java/org/apache/iceberg/gcp/bigquery/BigQueryTableOperations.java @@ -0,0 +1,259 @@ +/* + * 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.iceberg.gcp.bigquery; + +import com.google.api.services.bigquery.model.ExternalCatalogTableOptions; +import com.google.api.services.bigquery.model.Table; +import com.google.api.services.bigquery.model.TableReference; +import java.util.Locale; +import java.util.Map; +import org.apache.iceberg.BaseMetastoreOperations; +import org.apache.iceberg.BaseMetastoreTableOperations; +import org.apache.iceberg.SnapshotSummary; +import org.apache.iceberg.TableMetadata; +import org.apache.iceberg.TableProperties; +import org.apache.iceberg.exceptions.AlreadyExistsException; +import org.apache.iceberg.exceptions.CommitFailedException; +import org.apache.iceberg.exceptions.CommitStateUnknownException; +import org.apache.iceberg.exceptions.NoSuchTableException; +import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Handles BigQuery metastore table operations. */ +final class BigQueryTableOperations extends BaseMetastoreTableOperations { + + private static final Logger LOG = LoggerFactory.getLogger(BigQueryTableOperations.class); + + private static final String TABLE_PROPERTIES_BQ_CONNECTION = "bq_connection"; + + private final BigQueryMetastoreClient client; + private final FileIO fileIO; + private final TableReference tableReference; + + BigQueryTableOperations( + BigQueryMetastoreClient client, FileIO fileIO, TableReference tableReference) { + this.client = client; + this.fileIO = fileIO; + this.tableReference = tableReference; + } + + // The doRefresh method should provide implementation on how to get the metadata location. + @Override + public void doRefresh() { + // Must default to null. + String metadataLocation = null; + try { + metadataLocation = + loadMetadataLocationOrThrow(client.load(tableReference).getExternalCatalogTableOptions()); + } catch (NoSuchTableException e) { + if (currentMetadataLocation() != null) { + // Re-throws the exception because the table must exist in this case. + throw e; + } + } + refreshFromMetadataLocation(metadataLocation); + } + + // The doCommit method should provide implementation on how to update with metadata location + // atomically + @Override + public void doCommit(TableMetadata base, TableMetadata metadata) { + String newMetadataLocation = + base == null && metadata.metadataFileLocation() != null + ? metadata.metadataFileLocation() + : writeNewMetadata(metadata, currentVersion() + 1); + BaseMetastoreOperations.CommitStatus commitStatus = + BaseMetastoreOperations.CommitStatus.FAILURE; + try { + if (base == null) { + createTable(newMetadataLocation, metadata); + } else { + updateTable(base.metadataFileLocation(), newMetadataLocation, metadata); + } + commitStatus = BaseMetastoreOperations.CommitStatus.SUCCESS; + } catch (CommitFailedException | CommitStateUnknownException e) { + throw e; + } catch (Throwable e) { + LOG.error("Exception thrown on commit: ", e); + if (e instanceof AlreadyExistsException) { + throw e; + } + commitStatus = + BaseMetastoreOperations.CommitStatus.valueOf( + checkCommitStatus(newMetadataLocation, metadata).name()); + if (commitStatus == BaseMetastoreOperations.CommitStatus.FAILURE) { + throw new CommitFailedException(e, "Failed to commit"); + } + if (commitStatus == BaseMetastoreOperations.CommitStatus.UNKNOWN) { + throw new CommitStateUnknownException(e); + } + } finally { + try { + if (commitStatus == BaseMetastoreOperations.CommitStatus.FAILURE) { + LOG.warn("Failed to commit updates to table {}", tableName()); + io().deleteFile(newMetadataLocation); + } + } catch (RuntimeException e) { + LOG.error( + "Failed to cleanup metadata file at {} for table {}", + newMetadataLocation, + tableName(), + e); + } + } + } + + @Override + public String tableName() { + return String.format("%s.%s", tableReference.getDatasetId(), tableReference.getTableId()); + } + + @Override + public FileIO io() { + return fileIO; + } + + private void createTable(String newMetadataLocation, TableMetadata metadata) { + LOG.debug("Creating a new Iceberg table: {}", tableName()); + Table tableBuilder = makeNewTable(metadata, newMetadataLocation); + tableBuilder.setTableReference(tableReference); + addConnectionIfProvided(tableBuilder, metadata.properties()); + + client.create(tableBuilder); + } + + private void addConnectionIfProvided(Table tableBuilder, Map metadataProperties) { + if (metadataProperties.containsKey(TABLE_PROPERTIES_BQ_CONNECTION)) { + tableBuilder + .getExternalCatalogTableOptions() + .setConnectionId(metadataProperties.get(TABLE_PROPERTIES_BQ_CONNECTION)); + } + } + + /** Update table properties with concurrent update detection using etag. */ + private void updateTable( + String oldMetadataLocation, String newMetadataLocation, TableMetadata metadata) { + Table table = client.load(tableReference); + if (table.getEtag().isEmpty()) { + throw new ValidationException( + "Etag of legacy table %s is empty, manually update the table via the BigQuery API or" + + " recreate and retry", + tableName()); + } + ExternalCatalogTableOptions options = table.getExternalCatalogTableOptions(); + addConnectionIfProvided(table, metadata.properties()); + + // If `metadataLocationFromMetastore` is different from metadata location of base, it means + // someone has updated metadata location in metastore, which is a conflict update. + String metadataLocationFromMetastore = + options.getParameters().getOrDefault(METADATA_LOCATION_PROP, ""); + if (!metadataLocationFromMetastore.isEmpty() + && !metadataLocationFromMetastore.equals(oldMetadataLocation)) { + throw new CommitFailedException( + "Cannot commit base metadata location '%s' is not same as the current table metadata location '%s' for" + + " %s.%s", + oldMetadataLocation, + metadataLocationFromMetastore, + tableReference.getDatasetId(), + tableReference.getTableId()); + } + + options.setParameters(buildTableParameters(newMetadataLocation, metadata)); + try { + client.update(tableReference, table); + } catch (ValidationException e) { + if (e.getMessage().toLowerCase(Locale.ENGLISH).contains("etag mismatch")) { + throw new CommitFailedException( + "Updating table failed due to conflict updates (etag mismatch). Retry the update"); + } + + throw e; + } + } + + // To make the table queryable from Hive, the user would likely be setting the HIVE_ENGINE_ENABLED + // parameter. + // + // TODO: We need to make a decision on how to make the table queryable from Hive. + // (could be a server side change or a client side change - that's TBD). + private Table makeNewTable(TableMetadata metadata, String metadataFileLocation) { + return new Table() + .setExternalCatalogTableOptions( + BigQueryMetastoreUtils.createExternalCatalogTableOptions( + metadata.location(), buildTableParameters(metadataFileLocation, metadata))); + } + + // Follow Iceberg's HiveTableOperations to populate more table parameters for HMS compatibility. + private Map buildTableParameters( + String metadataFileLocation, TableMetadata metadata) { + Map parameters = Maps.newHashMap(metadata.properties()); + if (metadata.uuid() != null) { + parameters.put(TableProperties.UUID, metadata.uuid()); + } + if (currentMetadataLocation() != null && !currentMetadataLocation().isEmpty()) { + parameters.put(PREVIOUS_METADATA_LOCATION_PROP, currentMetadataLocation()); + } + parameters.put(METADATA_LOCATION_PROP, metadataFileLocation); + parameters.put(TABLE_TYPE_PROP, ICEBERG_TABLE_TYPE_VALUE); + // Follow HMS to use the EXTERNAL type. + parameters.put("EXTERNAL", "TRUE"); + + // Hive style basic statistics. + updateParametersWithSnapshotMetadata(metadata, parameters); + // More Iceberg metadata can be exposed, e.g., statistic, schema, partition spec, as HMS do. But + // we should be careful that these metadata could be huge and make the metadata API response + // less readable (e.g., list tables). Users can always inspect these metadata in Spark, so they + // are not set for now. + return parameters; + } + + /** Adds Hive-style basic statistics from snapshot metadata if it exists. */ + private static void updateParametersWithSnapshotMetadata( + TableMetadata metadata, Map parameters) { + if (metadata.currentSnapshot() == null) { + return; + } + + Map summary = metadata.currentSnapshot().summary(); + if (summary.get(SnapshotSummary.TOTAL_DATA_FILES_PROP) != null) { + parameters.put("numFiles", summary.get(SnapshotSummary.TOTAL_DATA_FILES_PROP)); + } + + if (summary.get(SnapshotSummary.TOTAL_RECORDS_PROP) != null) { + parameters.put("numRows", summary.get(SnapshotSummary.TOTAL_RECORDS_PROP)); + } + + if (summary.get(SnapshotSummary.TOTAL_FILE_SIZE_PROP) != null) { + parameters.put("totalSize", summary.get(SnapshotSummary.TOTAL_FILE_SIZE_PROP)); + } + } + + private String loadMetadataLocationOrThrow(ExternalCatalogTableOptions tableOptions) { + if (tableOptions == null || !tableOptions.getParameters().containsKey(METADATA_LOCATION_PROP)) { + throw new ValidationException( + "Table %s is not a valid BigQuery Metastore Iceberg table, metadata location not found", + tableName()); + } + + return tableOptions.getParameters().get(METADATA_LOCATION_PROP); + } +} diff --git a/bigquery/src/test/java/org/apache/iceberg/gcp/bigquery/FakeBigQueryMetastoreClient.java b/bigquery/src/test/java/org/apache/iceberg/gcp/bigquery/FakeBigQueryMetastoreClient.java new file mode 100644 index 000000000000..0c6df15091a6 --- /dev/null +++ b/bigquery/src/test/java/org/apache/iceberg/gcp/bigquery/FakeBigQueryMetastoreClient.java @@ -0,0 +1,254 @@ +/* + * 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.iceberg.gcp.bigquery; + +import com.google.api.services.bigquery.model.Dataset; +import com.google.api.services.bigquery.model.DatasetList; +import com.google.api.services.bigquery.model.DatasetReference; +import com.google.api.services.bigquery.model.ExternalCatalogDatasetOptions; +import com.google.api.services.bigquery.model.Table; +import com.google.api.services.bigquery.model.TableList; +import com.google.api.services.bigquery.model.TableReference; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.UUID; +import java.util.stream.Collectors; +import org.apache.iceberg.exceptions.AlreadyExistsException; +import org.apache.iceberg.exceptions.CommitFailedException; +import org.apache.iceberg.exceptions.NamespaceNotEmptyException; +import org.apache.iceberg.exceptions.NoSuchNamespaceException; +import org.apache.iceberg.exceptions.NoSuchTableException; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; + +public class FakeBigQueryMetastoreClient implements BigQueryMetastoreClient { + private final Map datasets = Maps.newHashMap(); + private final Map tables = Maps.newHashMap(); + + public FakeBigQueryMetastoreClient() {} + + @Override + public Dataset create(Dataset dataset) { + if (datasets.containsKey(dataset.getDatasetReference())) { + throw new AlreadyExistsException( + "Namespace already exists: %s", dataset.getDatasetReference()); + } + // Assign an ETag for consistency + dataset.setEtag(generateEtag()); + datasets.put(dataset.getDatasetReference(), dataset); + return dataset; + } + + @Override + public Dataset load(DatasetReference datasetReference) { + Dataset dataset = datasets.get(datasetReference); + if (dataset == null) { + throw new NoSuchNamespaceException( + "Namespace does not exist: %s", datasetReference.getDatasetId()); + } + + return dataset; + } + + @Override + public void delete(DatasetReference datasetReference) { + if (!datasets.containsKey(datasetReference)) { + throw new NoSuchNamespaceException("Dataset not found: %s", datasetReference); + } + + // Check if there are any tables in this dataset + if (tables.keySet().stream() + .anyMatch(tableRef -> tableRef.getDatasetId().equals(datasetReference.getDatasetId()))) { + throw new NamespaceNotEmptyException( + "Dataset is not empty. Cannot delete: %s", datasetReference); + } + + datasets.remove(datasetReference); + } + + @Override + public boolean setParameters(DatasetReference datasetReference, Map parameters) { + Dataset dataset = load(datasetReference); + + ExternalCatalogDatasetOptions existingOptions = dataset.getExternalCatalogDatasetOptions(); + + Map existingParameters = + (existingOptions == null || existingOptions.getParameters() == null) + ? Maps.newHashMap() + : Maps.newHashMap(existingOptions.getParameters()); + + Map newParameters = Maps.newHashMap(existingParameters); + newParameters.putAll(parameters); + + if (Objects.equals(existingParameters, newParameters)) { + return false; + } + + if (dataset.getExternalCatalogDatasetOptions() == null) { + dataset.setExternalCatalogDatasetOptions(new ExternalCatalogDatasetOptions()); + } + dataset.getExternalCatalogDatasetOptions().setParameters(newParameters); + + updateDataset(dataset); + return true; + } + + @Override + public boolean removeParameters(DatasetReference datasetReference, Set parameters) { + Dataset dataset = load(datasetReference); + + ExternalCatalogDatasetOptions existingOptions = dataset.getExternalCatalogDatasetOptions(); + + // If there are no options or no parameters, we cannot remove anything. + if (existingOptions == null + || existingOptions.getParameters() == null + || existingOptions.getParameters().isEmpty()) { + return false; + } + + Map existingParameters = + Maps.newHashMap(existingOptions.getParameters()); // Copy + Map newParameters = Maps.newHashMap(existingParameters); + parameters.forEach(newParameters::remove); + + if (Objects.equals(existingParameters, newParameters)) { + return false; + } + + dataset.getExternalCatalogDatasetOptions().setParameters(newParameters); + updateDataset(dataset); + return true; + } + + @Override + public List list(String projectId) { + return datasets.values().stream() + .map( + dataset -> { + DatasetList.Datasets ds = new DatasetList.Datasets(); + ds.setDatasetReference(dataset.getDatasetReference()); + return ds; + }) + .collect(Collectors.toList()); + } + + @Override + public Table create(Table table) { + if (tables.containsKey(table.getTableReference())) { + throw new AlreadyExistsException("Table already exists: %s", table.getTableReference()); + } + // Assign an ETag + table.setEtag(generateEtag()); + tables.put(table.getTableReference(), table); + return table; + } + + @Override + public Table load(TableReference tableReference) { + Table table = tables.get(tableReference); + if (table == null) { + throw new NoSuchTableException("Table not found: %s", tableReference); + } + + return table; + } + + @Override + public Table update(TableReference tableReference, Table table) { + Table existingTable = tables.get(tableReference); + if (existingTable == null) { + throw new NoSuchTableException("Table not found: %s", tableReference); + } + + String incomingEtag = table.getEtag(); + String requiredEtag = existingTable.getEtag(); + + // The real patch() uses an If-Match header which is passed separately, + // NOT on the incoming table object. + // The BigQueryTableOperations does NOT set the ETag on the Table object + // it passes to the client update() method. + // For a fake, we assume the ETag check needs to be simulated based on + // state, BUT the real client.update() expects the ETAG as a separate parameter + // (or implicitly via setIfMatch header, which this Fake doesn't see). + // To make the fake usable, we'll assume that if an ETag *is* present + // on the incoming table object, it must match. + if (incomingEtag != null && !incomingEtag.equals(requiredEtag)) { + throw new CommitFailedException( + "Etag mismatch for table: %s. Required: %s, Found: %s", + tableReference, requiredEtag, incomingEtag); + } + + Table tableToStore = table.clone(); + tableToStore.setEtag(generateEtag()); + tables.put(tableReference, tableToStore); + + return tableToStore.clone(); + } + + @Override + public void delete(TableReference tableReference) { + if (tables.remove(tableReference) == null) { + throw new NoSuchTableException("Table not found: %s", tableReference); + } + } + + @Override + public List list(DatasetReference datasetReference, boolean listAllTables) { + return tables.values().stream() + .filter( + table -> + table.getTableReference().getDatasetId().equals(datasetReference.getDatasetId())) + .map( + table -> { + TableList.Tables tbl = new TableList.Tables(); + tbl.setTableReference(table.getTableReference()); + return tbl; + }) + .collect(Collectors.toList()); + } + + public Dataset updateDataset(Dataset dataset) { + DatasetReference datasetReference = dataset.getDatasetReference(); + if (!datasets.containsKey(datasetReference)) { + throw new NoSuchNamespaceException( + "Namespace does not exist: %s", datasetReference.getDatasetId()); + } + Dataset existingDataset = datasets.get(datasetReference); + if (existingDataset == null) { + throw new NoSuchNamespaceException( + "Namespace does not exist: %s", datasetReference.getDatasetId()); + } + + // Robust ETag validation + if (dataset.getEtag() != null && !dataset.getEtag().equals(existingDataset.getEtag())) { + throw new CommitFailedException("Etag mismatch: concurrent modification"); + } + + // Update ETag + dataset.setEtag(generateEtag()); + // Simulate update by replacing the existing dataset + datasets.put(datasetReference, dataset); + return dataset; + } + + private String generateEtag() { + return UUID.randomUUID().toString(); + } +} diff --git a/bigquery/src/test/java/org/apache/iceberg/gcp/bigquery/TestBigQueryCatalog.java b/bigquery/src/test/java/org/apache/iceberg/gcp/bigquery/TestBigQueryCatalog.java new file mode 100644 index 000000000000..4fa1bd29dad4 --- /dev/null +++ b/bigquery/src/test/java/org/apache/iceberg/gcp/bigquery/TestBigQueryCatalog.java @@ -0,0 +1,174 @@ +/* + * 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.iceberg.gcp.bigquery; + +import static org.apache.iceberg.CatalogUtil.ICEBERG_CATALOG_TYPE; +import static org.apache.iceberg.CatalogUtil.ICEBERG_CATALOG_TYPE_BIGQUERY; +import static org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog.PROJECT_ID; + +import java.io.File; +import java.util.List; +import java.util.Map; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.catalog.CatalogTests; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.NoSuchNamespaceException; +import org.apache.iceberg.exceptions.NoSuchTableException; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +public class TestBigQueryCatalog extends CatalogTests { + @TempDir private File tempFolder; + private BigQueryMetastoreCatalog catalog; + + @BeforeEach + public void before() throws Exception { + catalog = initCatalog("catalog-name", ImmutableMap.of()); + } + + @AfterEach + public void after() throws Exception { + // Drop all tables in all datasets + List namespaces = catalog().listNamespaces(); + for (Namespace namespace : namespaces) { + List tables = catalog().listTables(namespace); + for (TableIdentifier table : tables) { + try { + catalog().dropTable(table, true); + } catch (NoSuchTableException e) { + // Table might be dropped by a previous iteration, ignore + } + } + + // Drop all datasets (except any initial ones) + try { + catalog().dropNamespace(namespace); + } catch (NoSuchNamespaceException e) { + // Namespace might be dropped by a previous iteration, ignore + } + } + } + + @Override + protected boolean requiresNamespaceCreate() { + return true; + } + + @Override + protected boolean supportsNamesWithSlashes() { + return false; + } + + @Override + protected boolean supportsNamesWithDot() { + return false; + } + + @Override + public BigQueryMetastoreCatalog catalog() { + return catalog; + } + + @Override + protected BigQueryMetastoreCatalog initCatalog( + String catalogName, Map additionalProperties) { + + String warehouseLocation = tempFolder.toPath().resolve("hive-warehouse").toString(); + FakeBigQueryMetastoreClient fakeBigQueryClient = new FakeBigQueryMetastoreClient(); + + Map properties = + Map.of( + ICEBERG_CATALOG_TYPE, + ICEBERG_CATALOG_TYPE_BIGQUERY, + PROJECT_ID, + "project-id", + CatalogProperties.WAREHOUSE_LOCATION, + warehouseLocation, + CatalogProperties.TABLE_DEFAULT_PREFIX + "default-key1", + "catalog-default-key1", + CatalogProperties.TABLE_DEFAULT_PREFIX + "default-key2", + "catalog-default-key2", + CatalogProperties.TABLE_DEFAULT_PREFIX + "override-key3", + "catalog-default-key3", + CatalogProperties.TABLE_OVERRIDE_PREFIX + "override-key3", + "catalog-override-key3", + CatalogProperties.TABLE_OVERRIDE_PREFIX + "override-key4", + "catalog-override-key4"); + + BigQueryMetastoreCatalog tmpCatalog = new BigQueryMetastoreCatalog(); + tmpCatalog.initialize( + catalogName, + ImmutableMap.builder() + .putAll(properties) + .putAll(additionalProperties) + .build(), + "project-id", + "us-central1", + fakeBigQueryClient); + + return tmpCatalog; + } + + // TODO: BigQuery Metastore does not support V3 Spec yet. + @Override + @ParameterizedTest + @ValueSource(ints = {1, 2}) + public void createTableTransaction(int formatVersion) { + super.createTableTransaction(formatVersion); + } + + @Disabled("BigQuery Metastore does not support V3 Spec yet.") + @Test + public void testCreateTableWithDefaultColumnValue() {} + + @Disabled("BigQuery Metastore does not support multi layer namespaces") + @Test + public void testLoadMetadataTable() {} + + @Disabled("BigQuery Metastore does not support rename tables") + @Test + public void testRenameTable() { + super.testRenameTable(); + } + + @Disabled("BigQuery Metastore does not support rename tables") + @Test + public void testRenameTableDestinationTableAlreadyExists() { + super.testRenameTableDestinationTableAlreadyExists(); + } + + @Disabled("BigQuery Metastore does not support rename tables") + @Test + public void renameTableNamespaceMissing() { + super.renameTableNamespaceMissing(); + } + + @Disabled("BigQuery Metastore does not support rename tables") + @Test + public void testRenameTableMissingSourceTable() { + super.testRenameTableMissingSourceTable(); + } +} diff --git a/bigquery/src/test/java/org/apache/iceberg/gcp/bigquery/TestBigQueryTableOperations.java b/bigquery/src/test/java/org/apache/iceberg/gcp/bigquery/TestBigQueryTableOperations.java new file mode 100644 index 000000000000..9b8b90e1f83d --- /dev/null +++ b/bigquery/src/test/java/org/apache/iceberg/gcp/bigquery/TestBigQueryTableOperations.java @@ -0,0 +1,293 @@ +/* + * 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.iceberg.gcp.bigquery; + +import static org.apache.iceberg.BaseMetastoreTableOperations.METADATA_LOCATION_PROP; +import static org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog.PROJECT_ID; +import static org.apache.iceberg.types.Types.NestedField.required; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatNoException; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.google.api.services.bigquery.model.Dataset; +import com.google.api.services.bigquery.model.DatasetReference; +import com.google.api.services.bigquery.model.ExternalCatalogDatasetOptions; +import com.google.api.services.bigquery.model.ExternalCatalogTableOptions; +import com.google.api.services.bigquery.model.StorageDescriptor; +import com.google.api.services.bigquery.model.Table; +import com.google.api.services.bigquery.model.TableReference; +import java.io.File; +import java.io.IOException; +import java.util.Collections; +import java.util.Optional; +import org.apache.commons.io.FileUtils; +import org.apache.commons.io.filefilter.TrueFileFilter; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.CommitFailedException; +import org.apache.iceberg.exceptions.NoSuchTableException; +import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.types.Types; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.mockito.ArgumentCaptor; + +public class TestBigQueryTableOperations { + + @TempDir private File tempFolder; + private static final String GCP_PROJECT = "my-project"; + private static final String GCP_REGION = "us"; + private static final String NS = "db"; + private static final String TABLE = "tbl"; + private static final TableIdentifier IDENTIFIER = TableIdentifier.of(NS, TABLE); + + private static final TableReference TABLE_REFERENCE = + new TableReference().setProjectId(GCP_PROJECT).setDatasetId(NS).setTableId(TABLE); + + private static final Schema SCHEMA = + new Schema( + required(1, "id", Types.IntegerType.get(), "unique ID"), + required(2, "data", Types.StringType.get())); + + private final BigQueryMetastoreClient client = mock(BigQueryMetastoreClient.class); + + private BigQueryMetastoreCatalog catalog; + private BigQueryTableOperations tableOps; + + @BeforeEach + public void before() { + this.catalog = new BigQueryMetastoreCatalog(); + this.catalog.setConf(new Configuration()); + String warehouseLocation = tempFolder.toPath().resolve("hive-warehouse").toString(); + + catalog.initialize( + "CATALOG_ID", + ImmutableMap.of( + PROJECT_ID, + GCP_PROJECT, + CatalogProperties.WAREHOUSE_LOCATION, + warehouseLocation, + CatalogProperties.FILE_IO_IMPL, + "org.apache.iceberg.hadoop.HadoopFileIO"), + GCP_PROJECT, + GCP_REGION, + client); + this.tableOps = (BigQueryTableOperations) catalog.newTableOps(IDENTIFIER); + } + + @Test + public void fetchLatestMetadataFromBigQuery() throws Exception { + Table createdTable = createTestTable(); + reset(client); + when(client.load(TABLE_REFERENCE)).thenReturn(createdTable); + + tableOps.refresh(); + assertThat( + createdTable + .getExternalCatalogTableOptions() + .getParameters() + .getOrDefault(METADATA_LOCATION_PROP, "")) + .isEqualTo(tableOps.currentMetadataLocation()); + + reset(client); + when(client.load(TABLE_REFERENCE)) + .thenThrow(new NoSuchTableException("error message getTable")); + // Refresh fails when table is not found but metadata already presents. + assertThatThrownBy(() -> tableOps.refresh()) + .isInstanceOf(NoSuchTableException.class) + .hasMessageContaining("error message getTable"); + } + + @Test + public void loadNonIcebergTableFails() { + when(client.load(TABLE_REFERENCE)).thenReturn(new Table().setTableReference(TABLE_REFERENCE)); + + assertThatThrownBy(() -> tableOps.refresh()) + .isInstanceOf(ValidationException.class) + .hasMessageContaining("metadata location not found"); + } + + @Test + public void loadNoOpWhenMetadataAndTableNotFound() { + when(client.load(TABLE_REFERENCE)) + .thenThrow(new NoSuchTableException("error message getTable")); + // Table not found won't cause errors when the metadata is null. + assertThat(tableOps.currentMetadataLocation()).isNull(); + assertThatNoException().isThrownBy(() -> tableOps.refresh()); + } + + @Test + public void loadTableNameAsExpected() { + assertThat(tableOps.tableName()).isEqualTo("db.tbl"); + } + + @Test + public void useEtagForUpdateTable() throws Exception { + Table tableWithEtag = createTestTable().setEtag("etag"); + reset(client); + when(client.load(TABLE_REFERENCE)).thenReturn(tableWithEtag, tableWithEtag); + + org.apache.iceberg.Table loadedTable = catalog.loadTable(IDENTIFIER); + + when(client.update(any(), any())).thenReturn(tableWithEtag); + loadedTable.updateSchema().addColumn("n", Types.IntegerType.get()).commit(); + + ArgumentCaptor tableReferenceArgumentCaptor = + ArgumentCaptor.forClass(TableReference.class); + ArgumentCaptor tableArgumentCaptor = ArgumentCaptor.forClass(Table.class); + verify(client, times(1)) + .update(tableReferenceArgumentCaptor.capture(), tableArgumentCaptor.capture()); + assertThat(tableReferenceArgumentCaptor.getValue()).isEqualTo(TABLE_REFERENCE); + assertThat(tableArgumentCaptor.getValue().getEtag()).isEqualTo("etag"); + } + + @Test + public void failWhenEtagMismatch() throws Exception { + Table tableWithEtag = createTestTable().setEtag("etag"); + reset(client); + when(client.load(TABLE_REFERENCE)).thenReturn(tableWithEtag); + + org.apache.iceberg.Table loadedTable = catalog.loadTable(IDENTIFIER); + + when(client.update(any(), any())) + .thenThrow(new ValidationException("error message etag mismatch")); + assertThatThrownBy( + () -> loadedTable.updateSchema().addColumn("n", Types.IntegerType.get()).commit()) + .isInstanceOf(CommitFailedException.class) + .hasMessageContaining( + "Updating table failed due to conflict updates (etag mismatch). Retry the update"); + } + + @Test + public void failWhenMetadataLocationDiff() throws Exception { + Table tableWithEtag = createTestTable().setEtag("etag"); + Table tableWithNewMetadata = + new Table() + .setEtag("etag") + .setExternalCatalogTableOptions( + new ExternalCatalogTableOptions() + .setParameters(ImmutableMap.of(METADATA_LOCATION_PROP, "a/new/location"))); + + reset(client); + // Two invocations, for loadTable and commit. + when(client.load(TABLE_REFERENCE)).thenReturn(tableWithEtag, tableWithNewMetadata); + + org.apache.iceberg.Table loadedTable = catalog.loadTable(IDENTIFIER); + + when(client.update(any(), any())).thenReturn(tableWithEtag); + assertThatThrownBy( + () -> loadedTable.updateSchema().addColumn("n", Types.IntegerType.get()).commit()) + .isInstanceOf(CommitFailedException.class) + .hasMessageContaining("is not same as the current table metadata location"); + } + + @Test + public void createTableCommitSucceeds() throws Exception { + var testTable = createTestTable(); + TableReference expectedTableReference = + new TableReference().setProjectId(GCP_PROJECT).setDatasetId(NS).setTableId(TABLE); + ArgumentCaptor
createdTableCaptor = ArgumentCaptor.forClass(Table.class); + + when(client.create(createdTableCaptor.capture())).thenReturn(testTable); + when(client.load(new DatasetReference().setProjectId(GCP_PROJECT).setDatasetId(NS))) + .thenReturn( + new Dataset() + .setExternalCatalogDatasetOptions( + new ExternalCatalogDatasetOptions() + .setDefaultStorageLocationUri("build/db_folder"))); + + Schema schema = SCHEMA; + catalog.createTable(IDENTIFIER, schema, PartitionSpec.unpartitioned()); + + Table capturedTable = createdTableCaptor.getValue(); + assertThat(capturedTable.getTableReference()).isEqualTo(expectedTableReference); + assertThat( + capturedTable + .getExternalCatalogTableOptions() + .getParameters() + .get(METADATA_LOCATION_PROP)) + .isNotNull(); + assertThat( + capturedTable.getExternalCatalogTableOptions().getStorageDescriptor().getLocationUri()) + .startsWith("build/db_folder/"); + + reset(client); + when(client.load(expectedTableReference)).thenReturn(testTable); + org.apache.iceberg.Table loadedTable = catalog.loadTable(IDENTIFIER); + assertThat(loadedTable).isNotNull(); + assertThat(loadedTable.name()).isEqualTo(catalog.name() + "." + IDENTIFIER); + assertThat(loadedTable.schema().asStruct()).isEqualTo(SCHEMA.asStruct()); + } + + /** Creates a test table to have Iceberg metadata files in place. */ + private Table createTestTable() throws Exception { + when(client.load(TABLE_REFERENCE)) + .thenThrow(new NoSuchTableException("error message getTable")); + return createTestTable(tempFolder, catalog, TABLE_REFERENCE); + } + + public static Table createTestTable( + File tempFolder, + BigQueryMetastoreCatalog bigQueryMetastoreCatalog, + TableReference tableReference) + throws IOException { + Schema schema = SCHEMA; + TableIdentifier tableIdentifier = + TableIdentifier.of(tableReference.getDatasetId(), tableReference.getTableId()); + String tableDir = tempFolder.toPath().resolve(tableReference.getTableId()).toString(); + + bigQueryMetastoreCatalog + .buildTable(tableIdentifier, schema) + .withLocation(tableDir) + .createTransaction() + .commitTransaction(); + + Optional metadataLocation = metadataFilePath(tableDir); + assertThat(metadataLocation).isPresent(); + return new Table() + .setTableReference(tableReference) + .setExternalCatalogTableOptions( + new ExternalCatalogTableOptions() + .setStorageDescriptor(new StorageDescriptor().setLocationUri(tableDir)) + .setParameters( + Collections.singletonMap(METADATA_LOCATION_PROP, metadataLocation.get()))); + } + + private static Optional metadataFilePath(String tableDir) throws IOException { + for (File file : + FileUtils.listFiles(new File(tableDir), TrueFileFilter.INSTANCE, TrueFileFilter.INSTANCE)) { + if (file.getCanonicalPath().endsWith(".json")) { + return Optional.of(file.getCanonicalPath()); + } + } + + return Optional.empty(); + } +} diff --git a/build.gradle b/build.gradle index 95b82f36301d..0e2216981fcf 100644 --- a/build.gradle +++ b/build.gradle @@ -657,6 +657,35 @@ project(':iceberg-delta-lake') { } } +project(':iceberg-bigquery') { + test { + useJUnitPlatform() + } + + dependencies { + api project(':iceberg-api') + implementation project(':iceberg-common') + implementation project(':iceberg-core') + + implementation project(path: ':iceberg-bundled-guava', configuration: 'shadow') + + implementation platform(libs.google.libraries.bom) + compileOnly "com.google.cloud:google-cloud-storage" + implementation "com.google.cloud:google-cloud-bigquery" + implementation "com.google.cloud:google-cloud-core" + + testImplementation project(path: ':iceberg-core', configuration: 'testArtifacts') + testImplementation project(path: ':iceberg-api', configuration: 'testArtifacts') + + testImplementation(libs.hadoop3.common) { + exclude group: 'org.apache.avro', module: 'avro' + exclude group: 'org.slf4j', module: 'slf4j-log4j12' + exclude group: 'javax.servlet', module: 'servlet-api' + exclude group: 'com.google.code.gson', module: 'gson' + } + } +} + project(':iceberg-gcp') { test { useJUnitPlatform() diff --git a/core/src/main/java/org/apache/iceberg/CatalogUtil.java b/core/src/main/java/org/apache/iceberg/CatalogUtil.java index b25f84e1ddb7..a96234629232 100644 --- a/core/src/main/java/org/apache/iceberg/CatalogUtil.java +++ b/core/src/main/java/org/apache/iceberg/CatalogUtil.java @@ -75,6 +75,7 @@ public class CatalogUtil { public static final String ICEBERG_CATALOG_TYPE_GLUE = "glue"; public static final String ICEBERG_CATALOG_TYPE_NESSIE = "nessie"; public static final String ICEBERG_CATALOG_TYPE_JDBC = "jdbc"; + public static final String ICEBERG_CATALOG_TYPE_BIGQUERY = "bigquery"; public static final String ICEBERG_CATALOG_HADOOP = "org.apache.iceberg.hadoop.HadoopCatalog"; public static final String ICEBERG_CATALOG_HIVE = "org.apache.iceberg.hive.HiveCatalog"; @@ -82,6 +83,8 @@ public class CatalogUtil { public static final String ICEBERG_CATALOG_GLUE = "org.apache.iceberg.aws.glue.GlueCatalog"; public static final String ICEBERG_CATALOG_NESSIE = "org.apache.iceberg.nessie.NessieCatalog"; public static final String ICEBERG_CATALOG_JDBC = "org.apache.iceberg.jdbc.JdbcCatalog"; + public static final String ICEBERG_CATALOG_BIGQUERY = + "org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog"; private CatalogUtil() {} @@ -315,6 +318,9 @@ public static Catalog buildIcebergCatalog(String name, Map optio case ICEBERG_CATALOG_TYPE_JDBC: catalogImpl = ICEBERG_CATALOG_JDBC; break; + case ICEBERG_CATALOG_TYPE_BIGQUERY: + catalogImpl = ICEBERG_CATALOG_BIGQUERY; + break; default: throw new UnsupportedOperationException("Unknown catalog type: " + catalogType); } diff --git a/settings.gradle b/settings.gradle index b42b04ecd47d..2ae978f72cf3 100644 --- a/settings.gradle +++ b/settings.gradle @@ -37,6 +37,7 @@ include 'hive-metastore' include 'nessie' include 'gcp' include 'gcp-bundle' +include 'bigquery' include 'dell' include 'snowflake' include 'delta-lake' @@ -62,6 +63,7 @@ project(':hive-metastore').name = 'iceberg-hive-metastore' project(':nessie').name = 'iceberg-nessie' project(':gcp').name = 'iceberg-gcp' project(':gcp-bundle').name = 'iceberg-gcp-bundle' +project(':bigquery').name = 'iceberg-bigquery' project(':dell').name = 'iceberg-dell' project(':snowflake').name = 'iceberg-snowflake' project(':delta-lake').name = 'iceberg-delta-lake'