You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by pa...@apache.org on 2022/08/15 17:35:19 UTC
[beam] branch master updated: Fix UpdateSchemaDestination when source format is set to AVRO
This is an automated email from the ASF dual-hosted git repository.
pabloem pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 884c0f9382b Fix UpdateSchemaDestination when source format is set to AVRO
new be9b247d02a Merge pull request #22390 from Fix UpdateSchemaDestination when source format is set to AVRO
884c0f9382b is described below
commit 884c0f9382b1caf8de688d959e3e28ea0c7e67f3
Author: Steve Niemitz <st...@gmail.com>
AuthorDate: Thu Jul 21 09:27:23 2022 -0400
Fix UpdateSchemaDestination when source format is set to AVRO
---
.../beam/sdk/io/gcp/bigquery/BatchLoads.java | 6 -----
.../io/gcp/bigquery/UpdateSchemaDestination.java | 28 ++++++----------------
2 files changed, 7 insertions(+), 27 deletions(-)
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java
index 7a1d5ff46fe..928f4a253c6 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java
@@ -381,10 +381,7 @@ class BatchLoads<DestinationT, ElementT>
WriteDisposition.WRITE_APPEND,
CreateDisposition.CREATE_NEVER,
maxRetryJobs,
- ignoreUnknownValues,
kmsKey,
- rowWriterFactory.getSourceFormat(),
- useAvroLogicalTypes,
schemaUpdateOptions,
dynamicDestinations))
.withSideInputs(sideInputsForUpdateSchema))
@@ -480,10 +477,7 @@ class BatchLoads<DestinationT, ElementT>
WriteDisposition.WRITE_APPEND,
CreateDisposition.CREATE_NEVER,
maxRetryJobs,
- ignoreUnknownValues,
kmsKey,
- rowWriterFactory.getSourceFormat(),
- useAvroLogicalTypes,
schemaUpdateOptions,
dynamicDestinations))
.withSideInputs(sideInputsForUpdateSchema))
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/UpdateSchemaDestination.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/UpdateSchemaDestination.java
index 4130b09556e..a7aa864d578 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/UpdateSchemaDestination.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/UpdateSchemaDestination.java
@@ -53,17 +53,13 @@ public class UpdateSchemaDestination
private final PCollectionView<String> loadJobIdPrefixView;
private final ValueProvider<String> loadJobProjectId;
private transient @Nullable DatasetService datasetService;
-
private final int maxRetryJobs;
private final @Nullable String kmsKey;
- private final String sourceFormat;
- private final boolean useAvroLogicalTypes;
private @Nullable BigQueryServices.JobService jobService;
- private final boolean ignoreUnknownValues;
private final Set<BigQueryIO.Write.SchemaUpdateOption> schemaUpdateOptions;
- private BigQueryIO.Write.WriteDisposition writeDisposition;
- private BigQueryIO.Write.CreateDisposition createDisposition;
- private DynamicDestinations dynamicDestinations;
+ private final BigQueryIO.Write.WriteDisposition writeDisposition;
+ private final BigQueryIO.Write.CreateDisposition createDisposition;
+ private final DynamicDestinations dynamicDestinations;
private static class PendingJobData {
final BigQueryHelpers.PendingJob retryJob;
@@ -80,7 +76,7 @@ public class UpdateSchemaDestination
}
}
- private List<UpdateSchemaDestination.PendingJobData> pendingJobs = Lists.newArrayList();
+ private final List<UpdateSchemaDestination.PendingJobData> pendingJobs = Lists.newArrayList();
public UpdateSchemaDestination(
BigQueryServices bqServices,
@@ -89,20 +85,14 @@ public class UpdateSchemaDestination
BigQueryIO.Write.WriteDisposition writeDisposition,
BigQueryIO.Write.CreateDisposition createDisposition,
int maxRetryJobs,
- boolean ignoreUnknownValues,
@Nullable String kmsKey,
- String sourceFormat,
- boolean useAvroLogicalTypes,
Set<BigQueryIO.Write.SchemaUpdateOption> schemaUpdateOptions,
DynamicDestinations dynamicDestinations) {
this.loadJobProjectId = loadJobProjectId;
this.loadJobIdPrefixView = loadJobIdPrefixView;
this.bqServices = bqServices;
this.maxRetryJobs = maxRetryJobs;
- this.ignoreUnknownValues = ignoreUnknownValues;
this.kmsKey = kmsKey;
- this.sourceFormat = sourceFormat;
- this.useAvroLogicalTypes = useAvroLogicalTypes;
this.schemaUpdateOptions = schemaUpdateOptions;
this.createDisposition = createDisposition;
this.writeDisposition = writeDisposition;
@@ -217,14 +207,10 @@ public class UpdateSchemaDestination
.setSchema(schema)
.setWriteDisposition(writeDisposition.name())
.setCreateDisposition(createDisposition.name())
- .setSourceFormat(sourceFormat)
- .setIgnoreUnknownValues(ignoreUnknownValues)
- .setUseAvroLogicalTypes(useAvroLogicalTypes);
+ .setSourceFormat("NEWLINE_DELIMITED_JSON");
if (schemaUpdateOptions != null) {
List<String> options =
- schemaUpdateOptions.stream()
- .map(Enum<BigQueryIO.Write.SchemaUpdateOption>::name)
- .collect(Collectors.toList());
+ schemaUpdateOptions.stream().map(Enum::name).collect(Collectors.toList());
loadConfig.setSchemaUpdateOptions(options);
}
if (!loadConfig
@@ -235,7 +221,7 @@ public class UpdateSchemaDestination
.equals(BigQueryIO.Write.WriteDisposition.WRITE_APPEND.toString())) {
return null;
}
- Table destinationTable = null;
+ final Table destinationTable;
try {
destinationTable = datasetService.getTable(tableReference);
if (destinationTable == null) {