Make the connector placement table ParentTokens column nullable. (#34812)

Use `./gradlew :sdks:java:io:google-cloud-platform:spotlessApply` to
format the affected files.

The ParentToken column is used for single-split based change stream
partitions. If multi-split based change stream is supported, then the
column will be null.

Author: changliiu

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dao/PartitionMetadataAdminDao.java

@@ -128,7 +128,7 @@ public void createPartitionMetadataTable() {
               + COLUMN_PARTITION_TOKEN
               + "\" text NOT NULL,\""
               + COLUMN_PARENT_TOKENS
-              + "\" text[] NOT NULL,\""
+              + "\" text[],\""
               + COLUMN_START_TIMESTAMP
               + "\" timestamptz NOT NULL,\""
               + COLUMN_END_TIMESTAMP
@@ -184,7 +184,7 @@ public void createPartitionMetadataTable() {
               + COLUMN_PARTITION_TOKEN
               + " STRING(MAX) NOT NULL,"
               + COLUMN_PARENT_TOKENS
-              + " ARRAY<STRING(MAX)> NOT NULL,"
+              + " ARRAY<STRING(MAX)>,"
               + COLUMN_START_TIMESTAMP
               + " TIMESTAMP NOT NULL,"
               + COLUMN_END_TIMESTAMP

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dofn/InitializeDoFn.java

@@ -86,7 +86,6 @@ private void createFakeParentPartition() {
     PartitionMetadata parentPartition =
         PartitionMetadata.newBuilder()
             .setPartitionToken(InitialPartition.PARTITION_TOKEN)
-            .setParentTokens(InitialPartition.PARENT_TOKENS)
             .setStartTimestamp(startTimestamp)
             .setEndTimestamp(endTimestamp)
             .setHeartbeatMillis(DEFAULT_HEARTBEAT_MILLIS)

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/mapper/PartitionMetadataMapper.java

@@ -80,7 +80,10 @@ public class PartitionMetadataMapper {
   public PartitionMetadata from(Struct row) {
     return PartitionMetadata.newBuilder()
         .setPartitionToken(row.getString(COLUMN_PARTITION_TOKEN))
-        .setParentTokens(Sets.newHashSet(row.getStringList(COLUMN_PARENT_TOKENS)))
+        .setParentTokens(
+            !row.isNull(COLUMN_PARENT_TOKENS)
+                ? Sets.newHashSet(row.getStringList(COLUMN_PARENT_TOKENS))
+                : null)
         .setStartTimestamp(row.getTimestamp(COLUMN_START_TIMESTAMP))
         .setEndTimestamp(row.getTimestamp(COLUMN_END_TIMESTAMP))
         .setHeartbeatMillis(row.getLong(COLUMN_HEARTBEAT_MILLIS))

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/model/InitialPartition.java

@@ -17,9 +17,6 @@
  */
 package org.apache.beam.sdk.io.gcp.spanner.changestreams.model;
 
-import java.util.HashSet;
-import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Sets;
-
 /**
  * Utility class to determine initial partition constants and methods.
  *
@@ -33,8 +30,6 @@ public class InitialPartition {
    * recognised by Cloud Spanner.
    */
   public static final String PARTITION_TOKEN = "Parent0";
-  /** The empty set representing the initial partition parent tokens. */
-  public static final HashSet<String> PARENT_TOKENS = Sets.newHashSet();
 
   /**
    * Verifies if the given partition token is the initial partition.

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/model/PartitionMetadata.java

@@ -61,7 +61,7 @@ public enum State {
   }
 
   private String partitionToken;
-  private HashSet<String> parentTokens;
+  @Nullable @org.apache.avro.reflect.Nullable private HashSet<String> parentTokens;
 
   @AvroEncode(using = TimestampEncoding.class)
   private Timestamp startTimestamp;
@@ -98,7 +98,7 @@ private PartitionMetadata() {}
 
   public PartitionMetadata(
       String partitionToken,
-      HashSet<String> parentTokens,
+      @Nullable HashSet<String> parentTokens,
       Timestamp startTimestamp,
       Timestamp endTimestamp,
       long heartbeatMillis,
@@ -130,7 +130,7 @@ public String getPartitionToken() {
    * The unique partition identifiers of the parent partitions where this child partition originated
    * from.
    */
-  public HashSet<String> getParentTokens() {
+  public @Nullable HashSet<String> getParentTokens() {
     return parentTokens;
   }
 
@@ -269,7 +269,7 @@ public static Builder newBuilder() {
   public static class Builder {
 
     private String partitionToken;
-    private HashSet<String> parentTokens;
+    @Nullable private HashSet<String> parentTokens;
     private Timestamp startTimestamp;
     private Timestamp endTimestamp;
     private Long heartbeatMillis;
@@ -303,7 +303,7 @@ public Builder setPartitionToken(String partitionToken) {
     }
 
     /** Sets the collection of parent partition identifiers. */
-    public Builder setParentTokens(HashSet<String> parentTokens) {
+    public Builder setParentTokens(@Nullable HashSet<String> parentTokens) {
       this.parentTokens = parentTokens;
       return this;
     }
@@ -376,7 +376,6 @@ public Builder setFinishedAt(@Nullable Timestamp finishedAt) {
      */
     public PartitionMetadata build() {
       Preconditions.checkState(partitionToken != null, "partitionToken");
-      Preconditions.checkState(parentTokens != null, "parentTokens");
       Preconditions.checkState(startTimestamp != null, "startTimestamp");
       Preconditions.checkState(heartbeatMillis != null, "heartbeatMillis");
       Preconditions.checkState(state != null, "state");

sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/model/PartitionMetadataTest.java

@@ -124,23 +124,6 @@ public void testBuilderThrowsExceptionWhenPartitionTokenMissing() {
                 .build());
   }
 
-  @Test
-  public void testBuilderThrowsExceptionWhenParentTokenMissing() {
-    assertThrows(
-        "parentToken",
-        IllegalStateException.class,
-        () ->
-            PartitionMetadata.newBuilder()
-                .setPartitionToken(PARTITION_TOKEN)
-                .setStartTimestamp(START_TIMESTAMP)
-                .setEndTimestamp(END_TIMESTAMP)
-                .setHeartbeatMillis(10)
-                .setState(State.CREATED)
-                .setWatermark(WATERMARK)
-                .setCreatedAt(CREATED_AT)
-                .build());
-  }
-
   @Test
   public void testBuilderThrowsExceptionWhenStartTimestampMissing() {
     assertThrows(