From 301286fd6a80bfae6f63d4998d002801716eceef Mon Sep 17 00:00:00 2001
From: Ahmed Abualsaud <65791736+ahmedabu98@users.noreply.github.com>
Date: Mon, 30 Sep 2024 20:44:31 -0400
Subject: [PATCH] default null array to empty array (#32604)

---
 .../bigquery/BeamRowToStorageApiProto.java    |  3 +++
 .../BeamRowToStorageApiProtoTest.java         | 23 ++++++++++++++-----
 2 files changed, 20 insertions(+), 6 deletions(-)

diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BeamRowToStorageApiProto.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BeamRowToStorageApiProto.java
index 4275125ef16d3..7a5aa2408d2e0 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BeamRowToStorageApiProto.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BeamRowToStorageApiProto.java
@@ -30,6 +30,7 @@
 import java.time.LocalDateTime;
 import java.time.LocalTime;
 import java.time.temporal.ChronoUnit;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.function.BiFunction;
@@ -272,6 +273,8 @@ private static Object messageValueFromRowValue(
     if (value == null) {
       if (fieldDescriptor.isOptional()) {
         return null;
+      } else if (fieldDescriptor.isRepeated()) {
+        return Collections.emptyList();
       } else {
         throw new IllegalArgumentException(
             "Received null value for non-nullable field " + fieldDescriptor.getName());
diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BeamRowToStorageApiProtoTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BeamRowToStorageApiProtoTest.java
index 5c43688a6efad..4013f0018553b 100644
--- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BeamRowToStorageApiProtoTest.java
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BeamRowToStorageApiProtoTest.java
@@ -35,6 +35,7 @@
 import java.time.LocalDateTime;
 import java.time.LocalTime;
 import java.time.temporal.ChronoUnit;
+import java.util.Collections;
 import java.util.Map;
 import java.util.stream.Collectors;
 import org.apache.beam.sdk.schemas.Schema;
@@ -71,6 +72,7 @@ public class BeamRowToStorageApiProtoTest {
           .addField("booleanValue", FieldType.BOOLEAN.withNullable(true))
           .addField("bytesValue", FieldType.BYTES.withNullable(true))
           .addField("arrayValue", FieldType.array(FieldType.STRING))
+          .addField("arrayNullValue", FieldType.array(FieldType.STRING).withNullable(true))
           .addField("iterableValue", FieldType.array(FieldType.STRING))
           .addField("sqlDateValue", FieldType.logicalType(SqlTypes.DATE).withNullable(true))
           .addField("sqlTimeValue", FieldType.logicalType(SqlTypes.TIME).withNullable(true))
@@ -168,43 +170,50 @@ public class BeamRowToStorageApiProtoTest {
                   .build())
           .addField(
               FieldDescriptorProto.newBuilder()
-                  .setName("iterablevalue")
+                  .setName("arraynullvalue")
                   .setNumber(13)
                   .setType(Type.TYPE_STRING)
                   .setLabel(Label.LABEL_REPEATED)
                   .build())
           .addField(
               FieldDescriptorProto.newBuilder()
-                  .setName("sqldatevalue")
+                  .setName("iterablevalue")
                   .setNumber(14)
+                  .setType(Type.TYPE_STRING)
+                  .setLabel(Label.LABEL_REPEATED)
+                  .build())
+          .addField(
+              FieldDescriptorProto.newBuilder()
+                  .setName("sqldatevalue")
+                  .setNumber(15)
                   .setType(Type.TYPE_INT32)
                   .setLabel(Label.LABEL_OPTIONAL)
                   .build())
           .addField(
               FieldDescriptorProto.newBuilder()
                   .setName("sqltimevalue")
-                  .setNumber(15)
+                  .setNumber(16)
                   .setType(Type.TYPE_INT64)
                   .setLabel(Label.LABEL_OPTIONAL)
                   .build())
           .addField(
               FieldDescriptorProto.newBuilder()
                   .setName("sqldatetimevalue")
-                  .setNumber(16)
+                  .setNumber(17)
                   .setType(Type.TYPE_INT64)
                   .setLabel(Label.LABEL_OPTIONAL)
                   .build())
           .addField(
               FieldDescriptorProto.newBuilder()
                   .setName("sqltimestampvalue")
-                  .setNumber(17)
+                  .setNumber(18)
                   .setType(Type.TYPE_INT64)
                   .setLabel(Label.LABEL_OPTIONAL)
                   .build())
           .addField(
               FieldDescriptorProto.newBuilder()
                   .setName("enumvalue")
-                  .setNumber(18)
+                  .setNumber(19)
                   .setType(Type.TYPE_STRING)
                   .setLabel(Label.LABEL_OPTIONAL)
                   .build())
@@ -225,6 +234,7 @@ public class BeamRowToStorageApiProtoTest {
           .withFieldValue("booleanValue", true)
           .withFieldValue("bytesValue", BYTES)
           .withFieldValue("arrayValue", ImmutableList.of("one", "two", "red", "blue"))
+          .withFieldValue("arrayNullValue", null)
           .withFieldValue("iterableValue", ImmutableList.of("blue", "red", "two", "one"))
           .withFieldValue("sqlDateValue", LocalDate.now())
           .withFieldValue("sqlTimeValue", LocalTime.now())
@@ -248,6 +258,7 @@ public class BeamRowToStorageApiProtoTest {
           .put("booleanvalue", true)
           .put("bytesvalue", ByteString.copyFrom(BYTES))
           .put("arrayvalue", ImmutableList.of("one", "two", "red", "blue"))
+          .put("arraynullvalue", Collections.emptyList())
           .put("iterablevalue", ImmutableList.of("blue", "red", "two", "one"))
           .put(
               "sqldatevalue",