DynamicDestinationsHelper.ConstantTimePartitioningClusteringDestinations is parsing per element json configuration for partitioning and clustering which is expensive. Cache the outcome of evaluation so it's done once.

Author: stankiewicz

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/DynamicDestinationsHelpers.java

@@ -30,7 +30,7 @@
 import java.io.IOException;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
 import org.apache.beam.sdk.coders.CannotProvideCoderException;
 import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.coders.CoderRegistry;
@@ -279,6 +279,11 @@ static class ConstantTimePartitioningClusteringDestinations<T>
     private final @Nullable ValueProvider<String> jsonTimePartitioning;
     private final @Nullable ValueProvider<String> jsonClustering;
 
+    // Lazily initialized and cached values.
+    private @Nullable String evaluatedPartitioning = null;
+    private @Nullable String evaluatedClustering = null;
+    private final AtomicBoolean initialized = new AtomicBoolean(false);
+
     ConstantTimePartitioningClusteringDestinations(
         DynamicDestinations<T, TableDestination> inner,
         ValueProvider<String> jsonTimePartitioning,
@@ -299,19 +304,42 @@ static class ConstantTimePartitioningClusteringDestinations<T>
       this.jsonClustering = jsonClustering;
     }
 
+    boolean isJsonConfigPresent(String json) {
+      return json != null && !JsonParser.parseString(json).getAsJsonObject().isEmpty();
+    }
+
+    private synchronized void evaluateOncePartitioningAndClustering() {
+      if (initialized.get()) {
+        return;
+      }
+      if (jsonTimePartitioning != null) {
+        String partitioning = jsonTimePartitioning.get();
+        if (isJsonConfigPresent(partitioning)) {
+          this.evaluatedPartitioning = partitioning;
+        }
+      }
+      if (jsonClustering != null) {
+        String clustering = jsonClustering.get();
+        if (isJsonConfigPresent(clustering)) {
+          this.evaluatedClustering = clustering;
+        }
+      }
+      initialized.set(true);
+    }
+
     @Override
     public TableDestination getDestination(@Nullable ValueInSingleWindow<T> element) {
+      if (!initialized.get()) {
+        evaluateOncePartitioningAndClustering();
+      }
       TableDestination destination = super.getDestination(element);
+
       String partitioning =
-          Optional.ofNullable(jsonTimePartitioning).map(ValueProvider::get).orElse(null);
-      if (partitioning == null
-          || JsonParser.parseString(partitioning).getAsJsonObject().isEmpty()) {
-        partitioning = destination.getJsonTimePartitioning();
-      }
-      String clustering = Optional.ofNullable(jsonClustering).map(ValueProvider::get).orElse(null);
-      if (clustering == null || JsonParser.parseString(clustering).getAsJsonObject().isEmpty()) {
-        clustering = destination.getJsonClustering();
-      }
+          evaluatedPartitioning != null
+              ? evaluatedPartitioning
+              : destination.getJsonTimePartitioning();
+      String clustering =
+          evaluatedClustering != null ? evaluatedClustering : destination.getJsonClustering();
 
       return new TableDestination(
           destination.getTableSpec(), destination.getTableDescription(), partitioning, clustering);