Issue GoogleCloudDataproc#1126: Making SparkBigQueryConfig Kryo compatible (GoogleCloudDataproc#1243)

Author: davidrabinowitz

CHANGES.md

@@ -2,6 +2,7 @@
 
 ## Next
 
+* Issue #1126: Fixing Kryo serialization issues
 * PR #1236: Fixing unshaded artifacts, added shading verification
 
 ## 0.39.0 - 2024-05-21

spark-bigquery-connector-common/src/main/java/com/google/cloud/spark/bigquery/SparkBigQueryConfig.java

@@ -62,7 +62,9 @@
 import java.time.format.DateTimeFormatter;
 import java.time.format.DateTimeParseException;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
@@ -207,8 +209,8 @@ public static WriteMethod from(@Nullable String writeMethod) {
   com.google.common.base.Optional<JobInfo.CreateDisposition> createDisposition = empty();
   boolean optimizedEmptyProjection = true;
   boolean useAvroLogicalTypes = false;
-  ImmutableList<String> decimalTargetTypes = ImmutableList.of();
-  ImmutableList<JobInfo.SchemaUpdateOption> loadSchemaUpdateOptions = ImmutableList.of();
+  List<String> decimalTargetTypes = Collections.emptyList();
+  List<JobInfo.SchemaUpdateOption> loadSchemaUpdateOptions = Collections.emptyList();
   int materializationExpirationTimeInMinutes;
   int maxReadRowsRetries = 3;
   boolean pushAllFilters = true;
@@ -233,8 +235,8 @@ public static WriteMethod from(@Nullable String writeMethod) {
   private int cacheExpirationTimeInMinutes = DEFAULT_CACHE_EXPIRATION_IN_MINUTES;
   // used to create BigQuery ReadSessions
   private com.google.common.base.Optional<String> traceId;
-  private ImmutableMap<String, String> bigQueryJobLabels = ImmutableMap.of();
-  private ImmutableMap<String, String> bigQueryTableLabels = ImmutableMap.of();
+  private Map<String, String> bigQueryJobLabels = Collections.emptyMap();
+  private Map<String, String> bigQueryTableLabels = Collections.emptyMap();
   private com.google.common.base.Optional<Long> createReadSessionTimeoutInSeconds;
   private QueryJobConfiguration.Priority queryJobPriority = DEFAULT_JOB_PRIORITY;
 
@@ -464,11 +466,11 @@ public static SparkBigQueryConfig from(
     if (allowFieldRelaxation) {
       loadSchemaUpdateOptions.add(JobInfo.SchemaUpdateOption.ALLOW_FIELD_RELAXATION);
     }
-    config.loadSchemaUpdateOptions = loadSchemaUpdateOptions.build();
+    config.loadSchemaUpdateOptions = Collections.unmodifiableList(loadSchemaUpdateOptions.build());
     com.google.common.base.Optional<String[]> decimalTargetTypes =
         getOption(options, "decimalTargetTypes").transform(s -> s.split(","));
     if (decimalTargetTypes.isPresent()) {
-      config.decimalTargetTypes = ImmutableList.copyOf(decimalTargetTypes.get());
+      config.decimalTargetTypes = Arrays.asList(decimalTargetTypes.get());
     }
     config.bigQueryStorageGrpcEndpoint =
         getAnyOption(globalOptions, options, "bigQueryStorageGrpcEndpoint");
@@ -630,7 +632,7 @@ private static com.google.common.base.Optional<String> stripPrefix(
   // takes only the options with the BIGQUERY_JOB_LABEL_PREFIX prefix, and strip them of this
   // prefix.
   // The `options` map overrides the `globalOptions` map.
-  static ImmutableMap<String, String> parseBigQueryLabels(
+  static Map<String, String> parseBigQueryLabels(
       ImmutableMap<String, String> globalOptions,
       ImmutableMap<String, String> options,
       String labelPrefix) {
@@ -650,7 +652,7 @@ static ImmutableMap<String, String> parseBigQueryLabels(
       }
     }
 
-    return result.build();
+    return Collections.unmodifiableMap(result.build());
   }
 
   private static ImmutableMap<String, String> toLowerCaseKeysMap(Map<String, String> map) {
@@ -859,7 +861,7 @@ public boolean isUseAvroLogicalTypes() {
   }
 
   public ImmutableList<String> getDecimalTargetTypes() {
-    return decimalTargetTypes;
+    return ImmutableList.copyOf(decimalTargetTypes);
   }
 
   public boolean isViewsEnabled() {
@@ -925,7 +927,7 @@ public boolean isOptimizedEmptyProjection() {
   }
 
   public ImmutableList<JobInfo.SchemaUpdateOption> getLoadSchemaUpdateOptions() {
-    return loadSchemaUpdateOptions;
+    return ImmutableList.copyOf(loadSchemaUpdateOptions);
   }
 
   public int getMaterializationExpirationTimeInMinutes() {
@@ -1049,7 +1051,7 @@ public Optional<String> getTraceId() {
 
   @Override
   public ImmutableMap<String, String> getBigQueryJobLabels() {
-    return bigQueryJobLabels;
+    return ImmutableMap.copyOf(bigQueryJobLabels);
   }
 
   public boolean getAllowMapTypeConversion() {
@@ -1061,7 +1063,7 @@ public long getBigQueryJobTimeoutInMinutes() {
   }
 
   public ImmutableMap<String, String> getBigQueryTableLabels() {
-    return bigQueryTableLabels;
+    return ImmutableMap.copyOf(bigQueryTableLabels);
   }
 
   public Optional<String> getGpn() {

spark-bigquery-connector-common/src/test/java/com/google/cloud/spark/bigquery/SparkBigQueryConfigTest.java

@@ -497,7 +497,7 @@ public void testJobLabelOverride() {
             .put("bigQueryJobLabel.foo", "2")
             .put("bigQueryJobLabel.baz", "2")
             .build();
-    ImmutableMap<String, String> labels =
+    Map<String, String> labels =
         SparkBigQueryConfig.parseBigQueryLabels(globalOptions, options, BIGQUERY_JOB_LABEL_PREFIX);
     assertThat(labels).hasSize(3);
     assertThat(labels).containsEntry("foo", "2");
@@ -518,7 +518,7 @@ public void testTableLabelOverride() {
             .put("bigQueryTableLabel.foo", "2")
             .put("bigQueryTableLabel.baz", "2")
             .build();
-    ImmutableMap<String, String> labels =
+    Map<String, String> labels =
         SparkBigQueryConfig.parseBigQueryLabels(
             globalOptions, options, BIGQUERY_TABLE_LABEL_PREFIX);
     assertThat(labels).hasSize(3);