SpannerIO: Handling pg spanner.commit_timestamp (#25479)

* Handling pg spanner.commit_timestamp type * deleting extra file
apache · Feb 15, 2023 · fd2b161 · fd2b161
1 parent bd35115
commit fd2b161
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 2 deletions.
diff --git a/...google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerSchema.java b/...google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerSchema.java
@@ -230,6 +230,9 @@ private static Type parseSpannerType(String spannerType, Dialect dialect) {
           if (spannerType.startsWith("NUMERIC")) {
             return Type.pgNumeric();
           }
+          if ("SPANNER.COMMIT_TIMESTAMP".equals(spannerType)) {
+            return Type.timestamp();
+          }
           throw new IllegalArgumentException("Unknown spanner type " + spannerType);
         default:
           throw new IllegalArgumentException("Unrecognized dialect: " + dialect.name());

diff --git a/...le-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/SpannerSchemaTest.java b/...le-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/SpannerSchemaTest.java
@@ -74,11 +74,13 @@ public void testSinglePgTable() throws Exception {
             .addKeyPart("test", "pk", false)
             .addColumn("test", "maxKey", "character varying")
             .addColumn("test", "numericVal", "numeric")
+            .addColumn("test", "commitTime", "spanner.commit_timestamp")
             .build();
 
     assertEquals(1, schema.getTables().size());
-    assertEquals(3, schema.getColumns("test").size());
+    assertEquals(4, schema.getColumns("test").size());
     assertEquals(1, schema.getKeyParts("test").size());
+    assertEquals(Type.timestamp(), schema.getColumns("test").get(3).getType());
   }
 
   @Test
@@ -91,13 +93,15 @@ public void testTwoPgTables() throws Exception {
             .addColumn("other", "pk", "bigint")
             .addKeyPart("other", "pk", true)
             .addColumn("other", "maxKey", "character varying")
+            .addColumn("other", "commitTime", "spanner.commit_timestamp")
             .build();
 
     assertEquals(2, schema.getTables().size());
     assertEquals(2, schema.getColumns("test").size());
     assertEquals(1, schema.getKeyParts("test").size());
 
-    assertEquals(2, schema.getColumns("other").size());
+    assertEquals(3, schema.getColumns("other").size());
     assertEquals(1, schema.getKeyParts("other").size());
+    assertEquals(Type.timestamp(), schema.getColumns("other").get(2).getType());
   }
 }