Allow strongly-consistent schema updates in BigQueryIO#38854
Allow strongly-consistent schema updates in BigQueryIO#38854reuvenlax wants to merge 10 commits into
Conversation
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 enhances BigQueryIO's schema update capabilities by introducing a strongly-consistent mode for Storage API writes. This ensures that data integrity is maintained during schema evolution by preventing the ingestion of rows with new schemas until the updated table schema is detected. The changes include new configuration options for timeouts, improved schema change detection logic, and internal refactoring to support more robust error handling and row splitting during schema transitions. 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
|
|
/gemini review |
![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 a consistent auto-schema update mechanism for BigQuery Storage API writes, allowing users to specify a timeout for waiting for schema updates. However, there are several critical compilation issues in the current implementation. These include a stray 'T' character in StorageApiDynamicDestinationsTableRow.java, a missing semicolon in StorageApiWritesShardedRecords.java, a call to a non-existent method bufferMismatchedRows, and calls to updateSchemaFromTable() on the MessageConverter interface where it is not defined.
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.
| public void updateSchemaFromTable() throws IOException, InterruptedException { | ||
| SCHEMA_CACHE.refreshSchema( | ||
| delegate.get().tableReference, datasetService, writeStreamService, bigQueryOptions); | ||
| T |
There was a problem hiding this comment.
There is a stray 'T' on line 154, which is a syntax error and will cause a compilation failure. This line should be removed.
| org.joda.time.Instant targetExpirationTime = startMismatchTime.plus(autoUpdateSchemaStrictTimeout); | ||
| Iterable<SplittingIterable.Value> mismatchedRows = | ||
| Iterables.transform(messages, SplittingIterable.Value::getSchemaMismatchedRowsOnly); | ||
| bufferMismatchedRows(mismatchedRows, targetExpirationTime) |
There was a problem hiding this comment.
There is a missing semicolon at the end of the bufferMismatchedRows call, which will cause a compilation failure.
| bufferMismatchedRows(mismatchedRows, targetExpirationTime) | |
| bufferMismatchedRows(mismatchedRows, targetExpirationTime); |
| // Start off with the base schema. As we get notified of schema updates, we | ||
| // will update the descriptor. | ||
| if (autoUpdateSchemaStrictTimeout != null) { | ||
| messageConverter.updateSchemaFromTable(); |
There was a problem hiding this comment.
The method updateSchemaFromTable() is called on messageConverter at lines 864 and 1116, but this method is not defined in the StorageApiDynamicDestinations.MessageConverter interface. This will cause a compilation failure. To resolve this, please add updateSchemaFromTable() to the MessageConverter interface in StorageApiDynamicDestinations.java and provide appropriate implementations in all classes implementing this interface.
| org.joda.time.Instant targetExpirationTime = startMismatchTime.plus(autoUpdateSchemaStrictTimeout); | ||
| Iterable<SplittingIterable.Value> mismatchedRows = | ||
| Iterables.transform(messages, SplittingIterable.Value::getSchemaMismatchedRowsOnly); | ||
| bufferMismatchedRows(mismatchedRows, targetExpirationTime) |
There was a problem hiding this comment.
The method bufferMismatchedRows is called here, but its definition is missing from this class and the rest of the PR. Please ensure that this method is implemented and included in the pull request.
There was a problem hiding this comment.
Code Review
This pull request introduces a strict timeout option for automatic schema updates in BigQueryIO (autoSchemaUpdateStrictTimeout), adding a new SchemaChangeDetectorHelper and updating several Storage API writing classes to support local retries and buffering of mismatched rows. The review feedback highlights several critical compilation issues that must be addressed, including a stray T character in StorageApiDynamicDestinationsTableRow.java, a missing semicolon and an undefined method bufferMismatchedRows in StorageApiWritesShardedRecords.java, and a missing interface method updateSchemaFromTable in StorageApiDynamicDestinations.java. Additionally, the reviewer recommended removing redundant loop conditions in SplittingIterable.java and simplifying a fully-qualified Preconditions call.
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.
| delegate.get().tableReference, datasetService, writeStreamService, bigQueryOptions); | ||
| T | ||
| // Recycle the internal MessageConverter so that we pick up the new schema from the cache. |
There was a problem hiding this comment.
There is a stray T character on line 154 which will cause a compilation failure. It should be removed.
| delegate.get().tableReference, datasetService, writeStreamService, bigQueryOptions); | |
| T | |
| // Recycle the internal MessageConverter so that we pick up the new schema from the cache. | |
| delegate.get().tableReference, datasetService, writeStreamService, bigQueryOptions); | |
| // Recycle the internal MessageConverter so that we pick up the new schema from the cache. |
| org.joda.time.Instant targetExpirationTime = startMismatchTime.plus(autoUpdateSchemaStrictTimeout); | ||
| Iterable<SplittingIterable.Value> mismatchedRows = | ||
| Iterables.transform(messages, SplittingIterable.Value::getSchemaMismatchedRowsOnly); | ||
| bufferMismatchedRows(mismatchedRows, targetExpirationTime) |
There was a problem hiding this comment.
There is a missing semicolon at the end of this statement. Additionally, the method bufferMismatchedRows is called here but is not defined anywhere in this class or the PR changes, which will cause a compilation failure.
| bufferMismatchedRows(mismatchedRows, targetExpirationTime) | |
| bufferMismatchedRows(mismatchedRows, targetExpirationTime); |
| TableSchema getTableSchema(); | ||
|
|
||
| DescriptorProtos.DescriptorProto getDescriptor(boolean includeCdcColumns) throws Exception; |
There was a problem hiding this comment.
The MessageConverter interface does not define updateSchemaFromTable(), but it is called on messageConverter in StorageApiWritesShardedRecords.java. Adding a default no-op implementation to the interface will prevent compilation failures for other implementations.
TableSchema getTableSchema();
DescriptorProtos.DescriptorProto getDescriptor(boolean includeCdcColumns) throws Exception;
default void updateSchemaFromTable() throws IOException, InterruptedException {}| for (int i = 0; ! getSchemaMismatchedRows().isEmpty() && i < getProtoRows().getSerializedRowsCount(); i++) { | ||
| if (getSchemaMismatchedRows().get(i)) { | ||
| inserts.addSerializedRows(getProtoRows().getSerializedRows(i)); | ||
| timestamps.add(getTimestamps().get(i)); | ||
| failsafeTableRows.add(getFailsafeTableRows().get(i)); | ||
| } | ||
| } |
There was a problem hiding this comment.
The loop condition contains ! getSchemaMismatchedRows().isEmpty() && which is redundant and evaluated on every iteration. It can be safely removed.
| for (int i = 0; ! getSchemaMismatchedRows().isEmpty() && i < getProtoRows().getSerializedRowsCount(); i++) { | |
| if (getSchemaMismatchedRows().get(i)) { | |
| inserts.addSerializedRows(getProtoRows().getSerializedRows(i)); | |
| timestamps.add(getTimestamps().get(i)); | |
| failsafeTableRows.add(getFailsafeTableRows().get(i)); | |
| } | |
| } | |
| for (int i = 0; i < getProtoRows().getSerializedRowsCount(); i++) { | |
| if (getSchemaMismatchedRows().get(i)) { | |
| inserts.addSerializedRows(getProtoRows().getSerializedRows(i)); | |
| timestamps.add(getTimestamps().get(i)); | |
| failsafeTableRows.add(getFailsafeTableRows().get(i)); | |
| } | |
| } |
| for (int i = 0; ! getSchemaMismatchedRows().isEmpty() && i < getProtoRows().getSerializedRowsCount(); i++) { | ||
| if (!getSchemaMismatchedRows().get(i)) { | ||
| inserts.addSerializedRows(getProtoRows().getSerializedRows(i)); | ||
| timestamps.add(getTimestamps().get(i)); | ||
| failsafeTableRows.add(getFailsafeTableRows().get(i)); | ||
| } | ||
| } |
There was a problem hiding this comment.
The loop condition contains ! getSchemaMismatchedRows().isEmpty() && which is redundant and evaluated on every iteration. It can be safely removed.
| for (int i = 0; ! getSchemaMismatchedRows().isEmpty() && i < getProtoRows().getSerializedRowsCount(); i++) { | |
| if (!getSchemaMismatchedRows().get(i)) { | |
| inserts.addSerializedRows(getProtoRows().getSerializedRows(i)); | |
| timestamps.add(getTimestamps().get(i)); | |
| failsafeTableRows.add(getFailsafeTableRows().get(i)); | |
| } | |
| } | |
| for (int i = 0; i < getProtoRows().getSerializedRowsCount(); i++) { | |
| if (!getSchemaMismatchedRows().get(i)) { | |
| inserts.addSerializedRows(getProtoRows().getSerializedRows(i)); | |
| timestamps.add(getTimestamps().get(i)); | |
| failsafeTableRows.add(getFailsafeTableRows().get(i)); | |
| } | |
| } |
| org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState( | ||
| !createRetryManagerResult.getSchemaMismatchSeen()); |
There was a problem hiding this comment.
The fully-qualified class name for Preconditions is used here, but checkState is already statically imported at the top of the file. Simplifying this improves readability.
checkState(!createRetryManagerResult.getSchemaMismatchSeen());
|
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment |
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request introduces a consistent schema update mechanism for the BigQuery Storage Write API in Apache Beam, allowing rows with mismatched schemas to be buffered and retried until the updated schema is observed or a timeout is reached. The review feedback highlights critical issues, including missing side input accessors for dynamic destinations during timer executions (which would cause runtime failures), a potential integer overflow in the sharding logic, and the need to prevent unnecessary empty RPC calls when all rows are buffered. Additionally, suggestions were made to support standard Java maps in schema checks and to clean up leftover debugging print statements.
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.
| PipelineOptions pipelineOptions, | ||
| MultiOutputReceiver o) | ||
| throws Exception { | ||
| // TODO XXX SET DYNAMIC DESTINATIONS |
There was a problem hiding this comment.
The side input accessor for dynamicDestinations is not set during the timer execution. If dynamicDestinations relies on side inputs, calling dynamicDestinations.getTable(...) on line 238 will fail at runtime. You should add OnTimerContext (or WindowContext) to the onTimer method parameters and use it to set the side input accessor on dynamicDestinations before accessing any tables.
| currentTimerValue.readLater(); | ||
| minPendingTimestamp.readLater(); | ||
|
|
||
| // TODOTDO XXX add context for side inputs. |
There was a problem hiding this comment.
The side input accessor for dynamicDestinations is not set during the timer execution. If dynamicDestinations relies on side inputs, calling dynamicDestinations.getTable(...) on line 1272 will fail at runtime. You should add OnTimerContext to the onMismatchedRowsTimer method parameters and use it to set the side input accessor on dynamicDestinations before accessing the table destination.
| Instant now = Instant.now(); | ||
| System.err.println("PROCESS " + now); |
There was a problem hiding this comment.
Remove leftover debugging System.err.println statements and the unused now variable.
| idleTimer, | ||
| o, | ||
| processMismatchedRows); | ||
| System.err.println("PROCESS DONE AFTER " + java.time.Duration.between(now, Instant.now())); |
There was a problem hiding this comment.
Remove leftover debugging System.err.println statement.
| @StateId("currentMismatchedRowTimerValue") ValueState<Long> currentTimerValue, | ||
| @StateId("minPendingTimestamp") ValueState<Long> minPendingTimestamp) | ||
| throws Exception { | ||
| System.err.println("RETRY TIMER " + Instant.now()); |
There was a problem hiding this comment.
Remove leftover debugging System.err.println statement.
| } else if (value instanceof TableRow) { | ||
| return missingUnknownField((TableRow) value, fieldDescriptor.getMessageType()); | ||
| } |
There was a problem hiding this comment.
Check for AbstractMap instead of TableRow to ensure that nested records represented as standard Java maps (e.g., LinkedHashMap) are also correctly checked for missing unknown fields.
| } else if (value instanceof TableRow) { | |
| return missingUnknownField((TableRow) value, fieldDescriptor.getMessageType()); | |
| } | |
| } else if (value instanceof AbstractMap) { | |
| return missingUnknownField((AbstractMap<String, Object>) value, fieldDescriptor.getMessageType()); | |
| } |
7e959e9 to
c3ca6b5
Compare
e426391 to
4ac6c18
Compare
|
RE: @ahmedabu98 |
|
@ahmedabu98 friendly ping. I know it's a large PR :) |
|
|
||
| class BufferingDoFn | ||
| extends DoFn<KV<ShardedKey<DestinationT>, MismatchedRow>, KV<String, String>> { | ||
| private final StorageApiWriteUnshardedRecords.WriteRecordsDoFn<DestinationT, ElementT> |
There was a problem hiding this comment.
maybe WriteRecordsDoFn should have service part extracted instead of wrapping whole doFn here.
| this.writeDoFn = writeDoFn; | ||
| } | ||
|
|
||
| @StartBundle |
There was a problem hiding this comment.
writeDoFn also also has teardown
There was a problem hiding this comment.
yes - this is implemented later in this file.
| } | ||
|
|
||
| @Override | ||
| public MismatchedRow decode(InputStream inStream) throws CoderException, IOException { |
There was a problem hiding this comment.
custom coders are not so friendly when it comes to updates and adding new fields (e.g. to StorageApiWritePayload), any chance to add some versioning before innerPayload.
There was a problem hiding this comment.
yes - I originally tried to use SchemaCoder for that reason. However schema-coder update today (in Dataflow) only works on PCollections - not necessarily when stored in state, and we store this in a BagState.
| PCollectionTuple> { | ||
| private static final Logger LOG = LoggerFactory.getLogger(StorageApiWritesShardedRecords.class); | ||
| private static final Duration DEFAULT_STREAM_IDLE_TIME = Duration.standardHours(1); | ||
| private static final Duration RETRY_MISMATCHED_ROWS_PERIOD = Duration.standardMinutes(1); |
There was a problem hiding this comment.
why not configurable?
| @Hidden | ||
| @Description("The initial retry time in milliseconds when a schema mismatch is detected.") | ||
| @Default.Integer(5000) | ||
| Integer getStorageApiInitialMismatchRetryTimeMilliSec(); |
There was a problem hiding this comment.
so this is for first local retry scenario, do you think it would be valuable to add second option for 1 min timer?
stankiewicz
left a comment
stankiewicz
left a comment
There was a problem hiding this comment.
The core writing logic inside WriteRecordsDoFn should be extracted into a separate service. Both the original WriteRecordsDoFn and this new BufferingDoFn should instantiate and call that service.
|
I'm ooo for next couple of weeks, lgtm! outstanding items are customcoder and failing tests. |
|
Reminder, please take a look at this pr: @ahmedabu98 |
|
Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment R: @chamikaramj for label java. Available commands:
|
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request introduces consistent auto-schema updates for BigQuery Storage API writes, adding support for strict timeouts and buffering mismatched rows. The review feedback highlights critical performance and reliability issues, specifically tight loops in BufferMismatchedRows and StorageApiWritesShardedRecords that could cause high CPU usage and API quota exhaustion during window expiration due to a lack of backoff. Additionally, the reviewer noted a validation bug in BigQueryIO where strict timeout checks are incorrectly run even when auto-schema updates are disabled, and suggested optimizing double map lookups in AppendRowsPacket to improve efficiency.
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.
| Instant start = Instant.now(); | ||
| while (!Iterables.isEmpty(bufferedRows) && start.plus(waitTime).isAfter(Instant.now())) { | ||
| writeDoFn.startBundle(); | ||
| List<Iterable<KV<DestinationT, StoragePayloadWithDeadline>>> mismatchedRowsList = | ||
| Lists.newArrayList(); | ||
| for (StoragePayloadWithDeadline row : bufferedRows) { | ||
| Iterable<KV<DestinationT, StoragePayloadWithDeadline>> mismatchedRows = | ||
| writeDoFn.processElement( | ||
| pipelineOptions, KV.of(shardedDestination.getKey(), row), null, o); | ||
| if (!Iterables.isEmpty(mismatchedRows)) { | ||
| mismatchedRowsList.add(mismatchedRows); | ||
| } | ||
| } | ||
| // Once we're done, delegate to finishBundle to finish things. | ||
| Iterable<KV<DestinationT, StoragePayloadWithDeadline>> mismatchedDestRows = | ||
| writeDoFn.finishBundle( | ||
| o.get(failedRowsTag), | ||
| successfulRowsTag != null ? o.get(successfulRowsTag) : null, | ||
| o.get(finalizeTag), | ||
| null); | ||
| if (!Iterables.isEmpty(mismatchedDestRows)) { | ||
| mismatchedRowsList.add(mismatchedDestRows); | ||
| } | ||
|
|
||
| bufferedRows = | ||
| () -> | ||
| StreamSupport.stream(Iterables.concat(mismatchedRowsList).spliterator(), false) | ||
| .map(KV::getValue) | ||
| .iterator(); | ||
| } |
There was a problem hiding this comment.
This while loop tightly spins without any backoff or sleep. If the schema update takes time, this loop will continuously call writeDoFn.processElement and finishBundle as fast as possible, consuming 100% CPU and hammering the BigQuery Storage API. Please introduce a backoff sleep (e.g., using FluentBackoff) inside the loop to prevent tight spinning.
Iterable<StoragePayloadWithDeadline> bufferedRows = mismatchedRowsBag.read();
Instant start = Instant.now();
org.apache.beam.sdk.util.BackOff backoff =
org.apache.beam.sdk.util.FluentBackoff.DEFAULT
.withInitialBackoff(Duration.standardSeconds(1))
.withMaxBackoff(Duration.standardMinutes(1))
.backoff();
while (!Iterables.isEmpty(bufferedRows) && start.plus(waitTime).isAfter(Instant.now())) {
writeDoFn.startBundle();
List<Iterable<KV<DestinationT, StoragePayloadWithDeadline>>> mismatchedRowsList =
Lists.newArrayList();
for (StoragePayloadWithDeadline row : bufferedRows) {
Iterable<KV<DestinationT, StoragePayloadWithDeadline>> mismatchedRows =
writeDoFn.processElement(
pipelineOptions, KV.of(shardedDestination.getKey(), row), null, o);
if (!Iterables.isEmpty(mismatchedRows)) {
mismatchedRowsList.add(mismatchedRows);
}
}
// Once we're done, delegate to finishBundle to finish things.
Iterable<KV<DestinationT, StoragePayloadWithDeadline>> mismatchedDestRows =
writeDoFn.finishBundle(
o.get(failedRowsTag),
successfulRowsTag != null ? o.get(successfulRowsTag) : null,
o.get(finalizeTag),
null);
if (!Iterables.isEmpty(mismatchedDestRows)) {
mismatchedRowsList.add(mismatchedDestRows);
}
bufferedRows =
() ->
StreamSupport.stream(Iterables.concat(mismatchedRowsList).spliterator(), false)
.map(KV::getValue)
.iterator();
if (!Iterables.isEmpty(bufferedRows)) {
Thread.sleep(backoff.nextBackOffMillis());
}
}| AtomicReference<Iterable<StoragePayloadWithDeadline>> mismatchedRows = | ||
| new AtomicReference<>(mismatchedRowsBag.read()); | ||
| Instant start = Instant.now(); | ||
| while (!Iterables.isEmpty(mismatchedRows.get()) | ||
| && start.plus(waitTime).isAfter(Instant.now())) { | ||
| messageConverter.updateSchemaFromTable(); | ||
| APPEND_CLIENTS.invalidate(messageConverters.getAppendClientKey(key)); | ||
| processPayloads( | ||
| pipelineOptions, | ||
| key, | ||
| tableDestination, | ||
| messageConverter, | ||
| mismatchedRows.get(), | ||
| null, | ||
| elementTs, | ||
| streamName, | ||
| streamOffset, | ||
| updatedSchema, | ||
| null, | ||
| o, | ||
| mismatchedRows::set); | ||
| } |
There was a problem hiding this comment.
This while loop tightly spins and calls messageConverter.updateSchemaFromTable() (which performs a synchronous BigQuery API RPC) on every iteration. Without a backoff sleep, this will immediately exhaust your BigQuery API quota and hammer the service. Please add a backoff sleep inside the loop.
AtomicReference<Iterable<StoragePayloadWithDeadline>> mismatchedRows =
new AtomicReference<>(mismatchedRowsBag.read());
Instant start = Instant.now();
org.apache.beam.sdk.util.BackOff backoff =
org.apache.beam.sdk.util.FluentBackoff.DEFAULT
.withInitialBackoff(Duration.standardSeconds(1))
.withMaxBackoff(Duration.standardMinutes(1))
.backoff();
while (!Iterables.isEmpty(mismatchedRows.get())
&& start.plus(waitTime).isAfter(Instant.now())) {
messageConverter.updateSchemaFromTable();
APPEND_CLIENTS.invalidate(messageConverters.getAppendClientKey(key));
processPayloads(
pipelineOptions,
key,
tableDestination,
messageConverter,
mismatchedRows.get(),
null,
elementTs,
streamName,
streamOffset,
updatedSchema,
null,
o,
mismatchedRows::set);
if (!Iterables.isEmpty(mismatchedRows.get())) {
Thread.sleep(backoff.nextBackOffMillis());
}
}| if (getAutoSchemaUpdateStrictTimeout() != null) { | ||
| checkArgument( | ||
| method == Method.STORAGE_API_AT_LEAST_ONCE || method == Method.STORAGE_WRITE_API, | ||
| "Auto update schema only supported when using storage write API"); | ||
| checkArgument( | ||
| input.getPipeline().getOptions().as(StreamingOptions.class).isStreaming(), | ||
| "auto update schema only supported on streaming pipelines"); | ||
| } |
There was a problem hiding this comment.
The validation checks for getAutoSchemaUpdateStrictTimeout() are executed even if getAutoSchemaUpdate() is false. If a user disables auto schema updates but has a strict timeout set (or if it's left as default/configured elsewhere), this will throw an IllegalArgumentException unnecessarily. The check should only run if getAutoSchemaUpdate() is enabled.
| if (getAutoSchemaUpdateStrictTimeout() != null) { | |
| checkArgument( | |
| method == Method.STORAGE_API_AT_LEAST_ONCE || method == Method.STORAGE_WRITE_API, | |
| "Auto update schema only supported when using storage write API"); | |
| checkArgument( | |
| input.getPipeline().getOptions().as(StreamingOptions.class).isStreaming(), | |
| "auto update schema only supported on streaming pipelines"); | |
| } | |
| if (getAutoSchemaUpdate() && getAutoSchemaUpdateStrictTimeout() != null) { | |
| checkArgument( | |
| method == Method.STORAGE_API_AT_LEAST_ONCE || method == Method.STORAGE_WRITE_API, | |
| "Auto update schema only supported when using storage write API"); | |
| checkArgument( | |
| input.getPipeline().getOptions().as(StreamingOptions.class).isStreaming(), | |
| "auto update schema only supported on streaming pipelines"); | |
| } |
| if (getUnknownFields().containsKey(i)) { | ||
| unknownFields.put(newIndex, Preconditions.checkStateNotNull(getUnknownFields().get(i))); | ||
| } | ||
| if (getOriginalPayloads().containsKey(i)) { | ||
| originalPayloads.put( | ||
| newIndex, Preconditions.checkStateNotNull(getOriginalPayloads().get(i))); | ||
| } | ||
| deadlines.add(getDeadlines().get(i)); | ||
| if (getSchemaHashes().containsKey(i)) { | ||
| schemaHashes.put(newIndex, Preconditions.checkStateNotNull(getSchemaHashes().get(i))); | ||
| } |
There was a problem hiding this comment.
Avoid double map lookups (containsKey followed by get). Instead, perform a single get and check if the returned value is non-null. This is cleaner and more efficient.
TableRow unknownField = getUnknownFields().get(i);
if (unknownField != null) {
unknownFields.put(newIndex, unknownField);
}
byte[] originalPayload = getOriginalPayloads().get(i);
if (originalPayload != null) {
originalPayloads.put(newIndex, originalPayload);
}
deadlines.add(getDeadlines().get(i));
byte[] schemaHash = getSchemaHashes().get(i);
if (schemaHash != null) {
schemaHashes.put(newIndex, schemaHash);
}| if (getSchemaHashes().containsKey(i)) { | ||
| schemaHashes.put(newIndex, Preconditions.checkStateNotNull(getSchemaHashes().get(i))); | ||
| } | ||
| if (getUnknownFields().containsKey(i)) { | ||
| unknownFields.put(newIndex, Preconditions.checkStateNotNull(getUnknownFields().get(i))); | ||
| } | ||
| if (getOriginalPayloads().containsKey(i)) { | ||
| originalPayloads.put( | ||
| newIndex, Preconditions.checkStateNotNull(getOriginalPayloads().get(i))); | ||
| } | ||
| deadlines.add(getDeadlines().get(i)); |
There was a problem hiding this comment.
Avoid double map lookups (containsKey followed by get). Perform a single get and check for null instead.
| if (getSchemaHashes().containsKey(i)) { | |
| schemaHashes.put(newIndex, Preconditions.checkStateNotNull(getSchemaHashes().get(i))); | |
| } | |
| if (getUnknownFields().containsKey(i)) { | |
| unknownFields.put(newIndex, Preconditions.checkStateNotNull(getUnknownFields().get(i))); | |
| } | |
| if (getOriginalPayloads().containsKey(i)) { | |
| originalPayloads.put( | |
| newIndex, Preconditions.checkStateNotNull(getOriginalPayloads().get(i))); | |
| } | |
| deadlines.add(getDeadlines().get(i)); | |
| byte[] schemaHash = getSchemaHashes().get(i); | |
| if (schemaHash != null) { | |
| schemaHashes.put(newIndex, schemaHash); | |
| } | |
| TableRow unknownField = getUnknownFields().get(i); | |
| if (unknownField != null) { | |
| unknownFields.put(newIndex, unknownField); | |
| } | |
| byte[] originalPayload = getOriginalPayloads().get(i); | |
| if (originalPayload != null) { | |
| originalPayloads.put(newIndex, originalPayload); | |
| } |
|
Reminder, please take a look at this pr: @chamikaramj @ahmedabu98 |
|
Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment R: @Abacn for label java. Available commands:
|
2 participants
No description provided.