Deserialize change stream proto based (mutable key range) return types.

Author: changliiu

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

@@ -44,8 +44,6 @@
 import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.InitialPartition;
 import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.Mod;
 import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.ModType;
-import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.MoveInEvent;
-import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.MoveOutEvent;
 import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.PartitionEndRecord;
 import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.PartitionEventRecord;
 import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.PartitionMetadata;
@@ -306,21 +304,9 @@ ChangeStreamRecord parseProtoPartitionEventRecord(
       com.google.spanner.v1.ChangeStreamRecord.PartitionEventRecord partitionEventRecordProto) {
     final Timestamp commitTimestamp =
         Timestamp.fromProto(partitionEventRecordProto.getCommitTimestamp());
-    List<MoveInEvent> moveInEvents = new ArrayList<>();
-    for (com.google.spanner.v1.ChangeStreamRecord.PartitionEventRecord.MoveInEvent
-        moveInEventProto : partitionEventRecordProto.getMoveInEventsList()) {
-      moveInEvents.add(new MoveInEvent(moveInEventProto.getSourcePartitionToken()));
-    }
-    List<MoveOutEvent> moveOutEvents = new ArrayList<>();
-    for (com.google.spanner.v1.ChangeStreamRecord.PartitionEventRecord.MoveOutEvent
-        moveOutEventProto : partitionEventRecordProto.getMoveOutEventsList()) {
-      moveOutEvents.add(new MoveOutEvent(moveOutEventProto.getDestinationPartitionToken()));
-    }
     return new PartitionEventRecord(
         commitTimestamp,
         partitionEventRecordProto.getRecordSequence(),
-        moveInEvents,
-        moveOutEvents,
         changeStreamRecordMetadataFrom(partition, commitTimestamp, resultSetMetadata));
   }
 

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

@@ -42,8 +42,6 @@
 import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.InitialPartition;
 import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.Mod;
 import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.ModType;
-import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.MoveInEvent;
-import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.MoveOutEvent;
 import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.PartitionEndRecord;
 import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.PartitionEventRecord;
 import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.PartitionMetadata;
@@ -987,8 +985,6 @@ public void testMappingProtoRowToPartitionEventRecord() {
         new PartitionEventRecord(
             Timestamp.MIN_VALUE,
             "fakeRecordSequence",
-            Arrays.asList(new MoveInEvent("token1"), new MoveInEvent("token2")),
-            Arrays.asList(new MoveOutEvent("token1"), new MoveOutEvent("token2")),
             null);
     com.google.spanner.v1.ChangeStreamRecord changeStreamRecordProto =
         recordToProto(partitionEventRecord);

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

@@ -31,8 +31,6 @@
 import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.HeartbeatRecord;
 import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.Mod;
 import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.ModType;
-import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.MoveInEvent;
-import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.MoveOutEvent;
 import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.PartitionEndRecord;
 import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.PartitionEventRecord;
 import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.PartitionStartRecord;
@@ -88,30 +86,10 @@ private static com.google.spanner.v1.ChangeStreamRecord convertPartitionEndRecor
 
   private static com.google.spanner.v1.ChangeStreamRecord convertPartitionEventRecordToProto(
       PartitionEventRecord partitionEventRecord) {
-    List<com.google.spanner.v1.ChangeStreamRecord.PartitionEventRecord.MoveInEvent>
-        moveInEventsProto = new ArrayList<>();
-    for (MoveInEvent moveInEvent : partitionEventRecord.getMoveInEvents()) {
-      com.google.spanner.v1.ChangeStreamRecord.PartitionEventRecord.MoveInEvent moveInEventProto =
-          com.google.spanner.v1.ChangeStreamRecord.PartitionEventRecord.MoveInEvent.newBuilder()
-              .setSourcePartitionToken(moveInEvent.getSourcePartitionToken())
-              .build();
-      moveInEventsProto.add(moveInEventProto);
-    }
-    List<com.google.spanner.v1.ChangeStreamRecord.PartitionEventRecord.MoveOutEvent>
-        moveOutEventsProto = new ArrayList<>();
-    for (MoveOutEvent moveOutEvent : partitionEventRecord.getMoveOutEvents()) {
-      com.google.spanner.v1.ChangeStreamRecord.PartitionEventRecord.MoveOutEvent moveOutEventProto =
-          com.google.spanner.v1.ChangeStreamRecord.PartitionEventRecord.MoveOutEvent.newBuilder()
-              .setDestinationPartitionToken(moveOutEvent.getDestinationPartitionToken())
-              .build();
-      moveOutEventsProto.add(moveOutEventProto);
-    }
     com.google.spanner.v1.ChangeStreamRecord.PartitionEventRecord partitionEventRecordProto =
         com.google.spanner.v1.ChangeStreamRecord.PartitionEventRecord.newBuilder()
             .setCommitTimestamp(partitionEventRecord.getCommitTimestamp().toProto())
             .setRecordSequence(partitionEventRecord.getRecordSequence())
-            .addAllMoveInEvents(moveInEventsProto)
-            .addAllMoveOutEvents(moveOutEventsProto)
             .build();
     com.google.spanner.v1.ChangeStreamRecord changeStreamRecordProto =
         com.google.spanner.v1.ChangeStreamRecord.newBuilder()