Add debug logs for updated schema in BigQueryIO (#28189)

apache · Aug 30, 2023 · e5ed7aa · e5ed7aa
1 parent 100c06a
commit e5ed7aa
Show file tree

Hide file tree

Showing 4 changed files with 9 additions and 2 deletions.
diff --git a/...gle-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryHelpers.java b/...gle-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryHelpers.java
@@ -126,7 +126,7 @@ PendingJobManager addPendingJob(
  }
 
  void waitForDone() throws Exception {
- LOG.info("Waiting for jobs to complete.");
+ LOG.debug("Waiting for BigQuery jobs to complete.");
  Sleeper sleeper = Sleeper.DEFAULT;
  while (!pendingJobs.isEmpty()) {
  List<JobInfo> retryJobs = Lists.newArrayList();

diff --git a/...rm/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java b/...rm/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java
@@ -430,8 +430,11 @@ SchemaAndDescriptor getCurrentTableSchema(String stream, @Nullable TableSchema u
  WriteStream writeStream =
  Preconditions.checkStateNotNull(maybeDatasetService).getWriteStream(streamName);
  if (writeStream != null && writeStream.hasTableSchema()) {
- currentSchema.set(writeStream.getTableSchema());
+ TableSchema updatedFromStream = writeStream.getTableSchema();
+ currentSchema.set(updatedFromStream);
  updated.set(true);
+ LOG.debug(
+ "Fetched updated schema for table {}:\n\t{}", tableUrn, updatedFromStream);
  }
  }
  return null;

diff --git a/...orm/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWritesShardedRecords.java b/...orm/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWritesShardedRecords.java
@@ -818,6 +818,8 @@ public void process(
  newSchema.get(), appendClientInfo.get().getCloseAppendClient(), false));
  APPEND_CLIENTS.invalidate(element.getKey());
  APPEND_CLIENTS.put(element.getKey(), appendClientInfo.get());
+ LOG.debug(
+ "Fetched updated schema for table {}:\n\t{}", tableId, updatedSchemaReturned);
  updatedSchema.write(newSchema.get());
  }
  }

diff --git a/.../google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/WriteTables.java b/.../google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/WriteTables.java
@@ -202,6 +202,7 @@ public void processElement(
  if (firstPaneCreateDisposition == CreateDisposition.CREATE_NEVER) {
  tableSchema = null;
  } else if (jsonSchemas.containsKey(destination)) {
+ // tableSchema for the destination stored in cache (jsonSchemas)
  tableSchema =
  BigQueryHelpers.fromJsonString(jsonSchemas.get(destination), TableSchema.class);
  } else {
@@ -215,6 +216,7 @@ public void processElement(
  firstPaneCreateDisposition,
  dynamicDestinations,
  destination);
+ LOG.debug("Fetched TableSchema for table {}:\n\t{}", destination, tableSchema);
  jsonSchemas.put(destination, BigQueryHelpers.toJsonString(tableSchema));
  }