Add BigQueryIO support for clone table creation#38883
Conversation
|
Warning You have reached your daily quota limit. Please wait up to 24 hours and I will start processing your requests again! |
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request introduces support for creating BigQuery destination tables by cloning an existing source table. By leveraging BigQuery's Highlights
New Features🧠 You can now enable Memory (public preview) to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. Footnotes
|
There was a problem hiding this comment.
Pull request overview
Note
Copilot was unable to run its full agentic suite in this review.
Adds support for creating missing BigQuery destination tables by cloning an existing “base” table, including plumbing through dynamic destinations and transform translation.
Changes:
- Introduces
withCloneFrom(...)/getCloneSource(...)API surface and wires clone source through table-creation helpers. - Implements clone-table creation via BigQuery copy jobs with operation type
CLONE. - Adds/updates translation + unit tests validating clone creation and config round-tripping.
Reviewed changes
Copilot reviewed 14 out of 14 changed files in this pull request and generated 11 comments.
Show a summary per file
| File | Description |
|---|---|
| sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java | Adds tests covering clone-table creation (static + dynamic destinations) and builder plumbing. |
| sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTranslationTest.java | Extends translation round-trip tests to include clone source field. |
| sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/WriteTables.java | Creates clone tables for file-load writes when a clone source is present. |
| sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/UpdateSchemaDestination.java | Ensures clone-table creation occurs during schema update path for dynamic destinations. |
| sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWritesShardedRecords.java | Passes clone source supplier into table creation for Storage API sharded writes. |
| sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java | Passes clone source supplier into table creation for Storage API unsharded writes. |
| sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/DynamicDestinationsHelpers.java | Adds clone-source delegation/wrappers and falls back to clone-source table metadata/schema. |
| sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/DynamicDestinations.java | Adds getCloneSource(...) extension point to dynamic destinations. |
| sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/CreateTables.java | Threads clone source into possiblyCreateTable(...) calls. |
| sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/CreateTableHelpers.java | Adds clone-table creation via copy job and expands table-creation metadata handling. |
| sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryResourceNaming.java | Introduces JobType.CLONE for clone job naming. |
| sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTranslation.java | Adds clone source field to transform payload schema and (de)serialization. |
| sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java | Adds withCloneFrom(...) API and validation + wiring into dynamic destinations and schema checks. |
| sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java | Uses the pipeline’s createDisposition for load jobs (enabling creation when cloning). |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| } catch (IOException | InterruptedException e) { | ||
| throw new RuntimeException(e); | ||
| } |
| } catch (InterruptedException e) { | ||
| throw new RuntimeException(e); | ||
| } |
| } catch (IOException | InterruptedException e) { | ||
| throw new RuntimeException(e); | ||
| } |
| @Nullable String kmsKey, | ||
| String tableSpec) | ||
| throws Exception { | ||
| TableReference source = withDefaultProject(options, cloneSource); |
| String jobProjectId = destination.getProjectId(); | ||
| String bqLocation = | ||
| BigQueryHelpers.getDatasetLocation( | ||
| datasetService, destination.getProjectId(), destination.getDatasetId()); |
| ConstantCloneSourceDestinations( | ||
| DynamicDestinations<T, DestinationT> inner, ValueProvider<String> jsonCloneSource) { | ||
| super(inner); | ||
| Preconditions.checkArgumentNotNull(jsonCloneSource, "jsonCloneSource can not be null"); |
| @Override | ||
| public TableReference getCloneSource(DestinationT destination) { | ||
| String jsonCloneSource = this.jsonCloneSource.get(); | ||
| checkArgument(jsonCloneSource != null, "jsonCloneSource can not be null"); |
| JobReference jobRef = | ||
| new JobReference() | ||
| .setProjectId(jobProjectId) | ||
| .setJobId(jobId.getJobId()) | ||
| .setLocation(bqLocation); |
| JobReference jobRef = | ||
| new JobReference() | ||
| .setProjectId(jobProjectId) | ||
| .setJobId(jobId.getJobId()) | ||
| .setLocation(bqLocation); |
| JobReference jobRef = | ||
| new JobReference() | ||
| .setProjectId(jobProjectId) | ||
| .setJobId(jobId.getJobId()) | ||
| .setLocation(bqLocation); |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces the ability to create missing BigQuery destination tables as clones of a specified base table using the new withCloneFrom option. The changes span BigQueryIO, DynamicDestinations, CreateTableHelpers, and related translation and test classes. The review feedback highlights several critical improvements: ensuring that the clone source table reference is fully qualified with a project ID (defaulting to the destination's project ID if missing) to prevent potential NullPointerException or API errors, restoring the interrupted thread status when catching InterruptedException in CreateTableHelpers, and simplifying the validation check for mutually exclusive table creation arguments in BigQueryIO using Java 8 Streams.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| if (tableToMatch == null) { | ||
| @Nullable TableReference cloneSource = super.getCloneSource(destination); | ||
| if (cloneSource != null) { | ||
| tableToMatch = getBigQueryTable(cloneSource); | ||
| } | ||
| } |
There was a problem hiding this comment.
If the clone source table reference does not specify a project ID (e.g., if it was parsed from a table spec like dataset.table), calling getBigQueryTable(cloneSource) directly can fail with a NullPointerException or API error because the project ID is missing. We should default the project ID to the destination table's project ID to ensure it is fully qualified before querying.
if (tableToMatch == null) {
@Nullable TableReference cloneSource = super.getCloneSource(destination);
if (cloneSource != null) {
TableReference cloneSourceWithProject = cloneSource.clone();
if (cloneSourceWithProject.getProjectId() == null || cloneSourceWithProject.getProjectId().isEmpty()) {
cloneSourceWithProject.setProjectId(wrappedDestination.getTableReference().getProjectId());
}
tableToMatch = getBigQueryTable(cloneSourceWithProject);
}
}| @Nullable TableReference cloneSource = super.getCloneSource(destination); | ||
| if (cloneSource != null) { | ||
| @Nullable Table cloneSourceTable = getBigQueryTable(cloneSource); | ||
| if (cloneSourceTable != null | ||
| && cloneSourceTable.getSchema() != null | ||
| && !cloneSourceTable.getSchema().isEmpty()) { | ||
| return cloneSourceTable.getSchema(); | ||
| } | ||
| } |
There was a problem hiding this comment.
Similar to the above, if the clone source table reference does not specify a project ID, calling getBigQueryTable(cloneSource) can fail. We should default the project ID to the destination table's project ID to ensure it is fully qualified.
@Nullable TableReference cloneSource = super.getCloneSource(destination);
if (cloneSource != null) {
TableReference cloneSourceWithProject = cloneSource.clone();
if (cloneSourceWithProject.getProjectId() == null || cloneSourceWithProject.getProjectId().isEmpty()) {
cloneSourceWithProject.setProjectId(super.getTable(destination).getTableReference().getProjectId());
}
@Nullable Table cloneSourceTable = getBigQueryTable(cloneSourceWithProject);
if (cloneSourceTable != null
&& cloneSourceTable.getSchema() != null
&& !cloneSourceTable.getSchema().isEmpty()) {
return cloneSourceTable.getSchema();
}
}| try { | ||
| jobService.startCopyJob(jobRef, copyConfig); | ||
| } catch (IOException | InterruptedException e) { | ||
| throw new RuntimeException(e); | ||
| } |
There was a problem hiding this comment.
When catching InterruptedException, it is a standard Java best practice to restore the interrupted status of the thread by calling Thread.currentThread().interrupt(). This ensures that higher-level call stacks or the runner are aware of the cancellation/interruption.
| try { | |
| jobService.startCopyJob(jobRef, copyConfig); | |
| } catch (IOException | InterruptedException e) { | |
| throw new RuntimeException(e); | |
| } | |
| try { | |
| jobService.startCopyJob(jobRef, copyConfig); | |
| } catch (IOException e) { | |
| throw new RuntimeException(e); | |
| } catch (InterruptedException e) { | |
| Thread.currentThread().interrupt(); | |
| throw new RuntimeException(e); | |
| } |
| try { | ||
| return jobService.pollJob(jobRef, BatchLoads.LOAD_JOB_POLL_MAX_RETRIES); | ||
| } catch (InterruptedException e) { | ||
| throw new RuntimeException(e); | ||
| } |
There was a problem hiding this comment.
Restore the interrupted status of the thread when catching InterruptedException to ensure the thread's interrupted state is preserved.
try {
return jobService.pollJob(jobRef, BatchLoads.LOAD_JOB_POLL_MAX_RETRIES);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException(e);
}| try { | ||
| return jobService.getJob(jobRef); | ||
| } catch (IOException | InterruptedException e) { | ||
| throw new RuntimeException(e); | ||
| } |
There was a problem hiding this comment.
Restore the interrupted status of the thread when catching InterruptedException to ensure the thread's interrupted state is preserved.
try {
return jobService.getJob(jobRef);
} catch (IOException e) {
throw new RuntimeException(e);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException(e);
}| List<?> allTableCreationMetadataArgs = | ||
| Lists.newArrayList(getJsonSchema(), getSchemaFromView(), getJsonCloneSourceTableRef()); | ||
| checkArgument( | ||
| 2 | ||
| > Iterables.size( | ||
| allTableCreationMetadataArgs.stream() | ||
| .filter(Predicates.notNull()::apply) | ||
| .collect(Collectors.toList())), | ||
| "No more than one of jsonSchema, schemaFromView, or cloneSource may be set"); |
There was a problem hiding this comment.
This check can be simplified and made more readable by using Java 8 Streams directly, avoiding the creation of unnecessary Guava lists and iterators.
long activeCreationMetadataCount = java.util.stream.Stream.of(getJsonSchema(), getSchemaFromView(), getJsonCloneSourceTableRef())
.filter(x -> x != null)
.count();
checkArgument(
activeCreationMetadataCount <= 1,
"No more than one of jsonSchema, schemaFromView, or cloneSource may be set");
|
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment |
This introduces
BigQueryIO.Write.withCloneFrom(...)for static clone sources andDynamicDestinations.getCloneSource(...)for dynamic destinations. When a destination table does not exist andCREATE_IF_NEEDEDis used, BigQueryIO can now create it with a BigQuery copy job usingoperationType = CLONE.addresses #38310
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, commentfixes #<ISSUE NUMBER>instead.CHANGES.mdwith noteworthy changes.See the Contributor Guide for more tips on how to make review process smoother.
To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md
GitHub Actions Tests Status (on master branch)
See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.