feat: adds toString to cdc models (#9)

* feat: adds toString methods to cdc models * refactor: moves cdc daos into own packages
hengfengli · Jun 7, 2021 · 14eb259 · 14eb259
1 parent 4e0cb7c
commit 14eb259
Show file tree

Hide file tree

Showing 10 changed files with 70 additions and 94 deletions.
diff --git a/.../io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java b/.../io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java
@@ -57,7 +57,7 @@
 import org.apache.beam.sdk.io.gcp.spanner.cdc.PipelineInitializer;
 import org.apache.beam.sdk.io.gcp.spanner.cdc.ReadPartitionChangeStream;
 import org.apache.beam.sdk.io.gcp.spanner.cdc.model.DataChangesRecord;
-import org.apache.beam.sdk.io.gcp.spanner.cdc.model.PartitionMetadataDao;
+import org.apache.beam.sdk.io.gcp.spanner.cdc.dao.PartitionMetadataDao;
 import org.apache.beam.sdk.metrics.Counter;
 import org.apache.beam.sdk.metrics.Distribution;
 import org.apache.beam.sdk.metrics.Metrics;

diff --git a/...ud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/cdc/PipelineInitializer.java b/...ud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/cdc/PipelineInitializer.java
@@ -17,16 +17,16 @@
  */
 package org.apache.beam.sdk.io.gcp.spanner.cdc;
 
-import static org.apache.beam.sdk.io.gcp.spanner.cdc.model.PartitionMetadataDao.COLUMN_CREATED_AT;
-import static org.apache.beam.sdk.io.gcp.spanner.cdc.model.PartitionMetadataDao.COLUMN_END_TIMESTAMP;
-import static org.apache.beam.sdk.io.gcp.spanner.cdc.model.PartitionMetadataDao.COLUMN_HEARTBEAT_SECONDS;
-import static org.apache.beam.sdk.io.gcp.spanner.cdc.model.PartitionMetadataDao.COLUMN_INCLUSIVE_END;
-import static org.apache.beam.sdk.io.gcp.spanner.cdc.model.PartitionMetadataDao.COLUMN_INCLUSIVE_START;
-import static org.apache.beam.sdk.io.gcp.spanner.cdc.model.PartitionMetadataDao.COLUMN_PARENT_TOKEN;
-import static org.apache.beam.sdk.io.gcp.spanner.cdc.model.PartitionMetadataDao.COLUMN_PARTITION_TOKEN;
-import static org.apache.beam.sdk.io.gcp.spanner.cdc.model.PartitionMetadataDao.COLUMN_START_TIMESTAMP;
-import static org.apache.beam.sdk.io.gcp.spanner.cdc.model.PartitionMetadataDao.COLUMN_STATE;
-import static org.apache.beam.sdk.io.gcp.spanner.cdc.model.PartitionMetadataDao.COLUMN_UPDATED_AT;
+import static org.apache.beam.sdk.io.gcp.spanner.cdc.dao.PartitionMetadataDao.COLUMN_CREATED_AT;
+import static org.apache.beam.sdk.io.gcp.spanner.cdc.dao.PartitionMetadataDao.COLUMN_END_TIMESTAMP;
+import static org.apache.beam.sdk.io.gcp.spanner.cdc.dao.PartitionMetadataDao.COLUMN_HEARTBEAT_SECONDS;
+import static org.apache.beam.sdk.io.gcp.spanner.cdc.dao.PartitionMetadataDao.COLUMN_INCLUSIVE_END;
+import static org.apache.beam.sdk.io.gcp.spanner.cdc.dao.PartitionMetadataDao.COLUMN_INCLUSIVE_START;
+import static org.apache.beam.sdk.io.gcp.spanner.cdc.dao.PartitionMetadataDao.COLUMN_PARENT_TOKEN;
+import static org.apache.beam.sdk.io.gcp.spanner.cdc.dao.PartitionMetadataDao.COLUMN_PARTITION_TOKEN;
+import static org.apache.beam.sdk.io.gcp.spanner.cdc.dao.PartitionMetadataDao.COLUMN_START_TIMESTAMP;
+import static org.apache.beam.sdk.io.gcp.spanner.cdc.dao.PartitionMetadataDao.COLUMN_STATE;
+import static org.apache.beam.sdk.io.gcp.spanner.cdc.dao.PartitionMetadataDao.COLUMN_UPDATED_AT;
 
 import com.google.api.gax.longrunning.OperationFuture;
 import com.google.cloud.Timestamp;
@@ -42,7 +42,7 @@
 import javax.annotation.Nullable;
 import org.apache.beam.sdk.io.gcp.spanner.cdc.model.PartitionMetadata;
 import org.apache.beam.sdk.io.gcp.spanner.cdc.model.PartitionMetadata.State;
-import org.apache.beam.sdk.io.gcp.spanner.cdc.model.PartitionMetadataDao;
+import org.apache.beam.sdk.io.gcp.spanner.cdc.dao.PartitionMetadataDao;
 
 public class PipelineInitializer {
 

diff --git a/...anner/cdc/model/PartitionMetadataDao.java → ...spanner/cdc/dao/PartitionMetadataDao.java b/...anner/cdc/model/PartitionMetadataDao.java → ...spanner/cdc/dao/PartitionMetadataDao.java
@@ -1,9 +1,10 @@
-package org.apache.beam.sdk.io.gcp.spanner.cdc.model;
+package org.apache.beam.sdk.io.gcp.spanner.cdc.dao;
 
 import com.google.cloud.Timestamp;
 import com.google.cloud.spanner.DatabaseClient;
 import com.google.cloud.spanner.Mutation;
 import com.google.common.collect.ImmutableList;
+import org.apache.beam.sdk.io.gcp.spanner.cdc.model.PartitionMetadata;
 
 public class PartitionMetadataDao {
 

diff --git a/...cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/cdc/model/ColumnType.java b/...cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/cdc/model/ColumnType.java
@@ -86,4 +86,13 @@ public boolean equals(Object o) {
   public int hashCode() {
     return Objects.hash(name, type, isPrimaryKey);
   }
+
+  @Override
+  public String toString() {
+    return "ColumnType{" +
+        "name='" + name + '\'' +
+        ", type=" + type +
+        ", isPrimaryKey=" + isPrimaryKey +
+        '}';
+  }
 }
diff --git a/...latform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/cdc/model/DataChangesRecord.java b/...latform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/cdc/model/DataChangesRecord.java
@@ -186,4 +186,20 @@ public int hashCode() {
         modType,
         valueCaptureType);
   }
+
+  @Override
+  public String toString() {
+    return "DataChangesRecord{" +
+        "partitionToken='" + partitionToken + '\'' +
+        ", commitTimestamp=" + commitTimestamp +
+        ", transactionId='" + transactionId + '\'' +
+        ", isLastRecordInTransactionPartition=" + isLastRecordInTransactionPartition +
+        ", recordSequence='" + recordSequence + '\'' +
+        ", tableName='" + tableName + '\'' +
+        ", rowType=" + rowType +
+        ", mods=" + mods +
+        ", modType=" + modType +
+        ", valueCaptureType=" + valueCaptureType +
+        '}';
+  }
 }
diff --git a/...google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/cdc/model/Mod.java b/...google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/cdc/model/Mod.java
@@ -98,4 +98,12 @@ public boolean equals(Object o) {
   public int hashCode() {
     return Objects.hash(newValues, oldValues);
   }
+
+  @Override
+  public String toString() {
+    return "Mod{" +
+        "oldValues=" + oldValues +
+        ", newValues=" + newValues +
+        '}';
+  }
 }
diff --git a/...latform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/cdc/model/PartitionMetadata.java b/...latform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/cdc/model/PartitionMetadata.java
@@ -215,6 +215,22 @@ public int hashCode() {
         getUpdatedAt());
   }
 
+  @Override
+  public String toString() {
+    return "PartitionMetadata{" +
+        "partitionToken='" + partitionToken + '\'' +
+        ", parentTokens=" + parentTokens +
+        ", startTimestamp=" + startTimestamp +
+        ", inclusiveStart=" + inclusiveStart +
+        ", endTimestamp=" + endTimestamp +
+        ", inclusiveEnd=" + inclusiveEnd +
+        ", heartbeatSeconds=" + heartbeatSeconds +
+        ", state=" + state +
+        ", createdAt=" + createdAt +
+        ", updatedAt=" + updatedAt +
+        '}';
+  }
+
   public static PartitionMetadata.Builder newBuilder() {
     return new PartitionMetadata.Builder();
   }

diff --git a/...d-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/cdc/model/RecordSequence.java b/...d-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/cdc/model/RecordSequence.java
diff --git a/...e-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/cdc/model/TypeCode.java b/...e-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/cdc/model/TypeCode.java
@@ -62,4 +62,11 @@ public boolean equals(Object o) {
   public int hashCode() {
     return Objects.hash(code);
   }
+
+  @Override
+  public String toString() {
+    return "TypeCode{" +
+        "code='" + code + '\'' +
+        '}';
+  }
 }
diff --git a/...latform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/cdc/model/ModelEncodingTest.java b/...latform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/cdc/model/ModelEncodingTest.java
@@ -54,13 +54,6 @@ public void testModTypeCanBeEncoded() throws IOException {
     assertEquals(ModType.INSERT, encodeAndDecode(ModType.INSERT));
   }
 
-  @Test
-  public void testRecordSequenceCanBeEncoded() throws IOException {
-    final RecordSequence recordSequence = new RecordSequence(2L);
-
-    assertEquals(recordSequence, encodeAndDecode(recordSequence));
-  }
-
   @Test
   public void testTypeCodeCanBeEncoded() throws IOException {
     final TypeCode typeCode = new TypeCode("typeCode");