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,11 +279,17 @@ static class ConstantTimePartitioningClusteringDestinations<T>
     private final @Nullable ValueProvider<String> jsonTimePartitioning;
     private final @Nullable ValueProvider<String> jsonClustering;
 
+    // Lazily initialized and cached values.
+    private transient @Nullable String evaluatedPartitioning = null;
+    private transient @Nullable String evaluatedClustering = null;
+    private transient AtomicBoolean initialized = null;
+
     ConstantTimePartitioningClusteringDestinations(
         DynamicDestinations<T, TableDestination> inner,
         ValueProvider<String> jsonTimePartitioning,
         ValueProvider<String> jsonClustering) {
       super(inner);
+      initialized = new AtomicBoolean(false);
 
       checkArgument(
           (jsonTimePartitioning != null
@@ -299,19 +305,36 @@ static class ConstantTimePartitioningClusteringDestinations<T>
       this.jsonClustering = jsonClustering;
     }
 
+    private synchronized void evaluateOncePartitioningAndClustering() {
+      if (initialized.get()) {
+        return;
+      }
+      if (jsonTimePartitioning != null) {
+        String partitioning = jsonTimePartitioning.get();
+        if (partitioning != null
+            && !JsonParser.parseString(partitioning).getAsJsonObject().isEmpty()) {
+          this.evaluatedPartitioning = partitioning;
+        }
+      }
+      if (jsonClustering != null) {
+        String clustering = jsonClustering.get();
+        if (clustering != null && !JsonParser.parseString(clustering).getAsJsonObject().isEmpty()) {
+          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();
-      }
+          MoreObjects.firstNonNull(evaluatedPartitioning, destination.getJsonTimePartitioning());
+      String clustering =
+          MoreObjects.firstNonNull(evaluatedClustering, destination.getJsonClustering());
 
       return new TableDestination(
           destination.getTableSpec(), destination.getTableDescription(), partitioning, clustering);