Repository: incubator-beam Updated Branches: refs/heads/master 21c13d5f6 -> 68010aba1
Support BigQuery DATETIME and TIME types Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/2bfc89ea Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/2bfc89ea Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/2bfc89ea Branch: refs/heads/master Commit: 2bfc89ea38e7831a12a9c1d0098aaffa4f035f08 Parents: 21c13d5 Author: Pei He <[email protected]> Authored: Fri Sep 23 18:11:06 2016 -0700 Committer: Luke Cwik <[email protected]> Committed: Mon Sep 26 16:31:49 2016 -0700 ---------------------------------------------------------------------- .../sdk/io/gcp/bigquery/BigQueryAvroUtils.java | 4 ++++ .../gcp/bigquery/BigQueryTableRowIterator.java | 3 ++- .../io/gcp/bigquery/BigQueryAvroUtilsTest.java | 20 +++++++++++++++----- .../bigquery/BigQueryTableRowIteratorTest.java | 20 ++++++++++++++------ 4 files changed, 35 insertions(+), 12 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/2bfc89ea/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryAvroUtils.java ---------------------------------------------------------------------- diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryAvroUtils.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryAvroUtils.java index 2cc2df7..7800108 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryAvroUtils.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryAvroUtils.java @@ -164,6 +164,8 @@ class BigQueryAvroUtils { .put("TIMESTAMP", Type.LONG) .put("RECORD", Type.RECORD) .put("DATE", Type.STRING) + .put("DATETIME", Type.STRING) + .put("TIME", Type.STRING) .build(); // Per https://cloud.google.com/bigquery/docs/reference/v2/tables#schema, the type field // is required, so it may not be null. @@ -180,6 +182,8 @@ class BigQueryAvroUtils { switch (fieldSchema.getType()) { case "STRING": case "DATE": + case "DATETIME": + case "TIME": // Avro will use a CharSequence to represent String objects, but it may not always use // java.lang.String; for example, it may prefer org.apache.avro.util.Utf8. verify(v instanceof CharSequence, "Expected CharSequence (String), got %s", v.getClass()); http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/2bfc89ea/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryTableRowIterator.java ---------------------------------------------------------------------- diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryTableRowIterator.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryTableRowIterator.java index d7423a1..0ee01d9 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryTableRowIterator.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryTableRowIterator.java @@ -253,7 +253,8 @@ class BigQueryTableRowIterator implements AutoCloseable { return BigQueryAvroUtils.formatTimestamp((String) v); } - // Returns the original value for String, base64 encoded BYTES, and DATE string. + // Returns the original value for: + // 1. String, 2. base64 encoded BYTES, 3. DATE, DATETIME, TIME strings. return v; } http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/2bfc89ea/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryAvroUtilsTest.java ---------------------------------------------------------------------- diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryAvroUtilsTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryAvroUtilsTest.java index 6924732..1d3ea81 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryAvroUtilsTest.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryAvroUtilsTest.java @@ -66,7 +66,10 @@ public class BigQueryAvroUtilsTest { new TableFieldSchema().setName("birthday").setType("TIMESTAMP").setMode("NULLABLE"), new TableFieldSchema().setName("flighted").setType("BOOLEAN").setMode("NULLABLE"), new TableFieldSchema().setName("sound").setType("BYTES").setMode("NULLABLE"), - new TableFieldSchema().setName("anniversary").setType("DATE").setMode("NULLABLE"), + new TableFieldSchema().setName("anniversaryDate").setType("DATE").setMode("NULLABLE"), + new TableFieldSchema().setName("anniversaryDatetime") + .setType("DATETIME").setMode("NULLABLE"), + new TableFieldSchema().setName("anniversaryTime").setType("TIME").setMode("NULLABLE"), new TableFieldSchema().setName("scion").setType("RECORD").setMode("NULLABLE") .setFields(subFields), new TableFieldSchema().setName("associates").setType("RECORD").setMode("REPEATED") @@ -85,7 +88,8 @@ public class BigQueryAvroUtilsTest { assertEquals(row, convertedRow); } { - // Test type conversion for INTEGER, FLOAT, TIMESTAMP, BOOLEAN, BYTES, and DATE. + // Test type conversion for: + // INTEGER, FLOAT, TIMESTAMP, BOOLEAN, BYTES, DATE, DATETIME, TIME. GenericRecord record = new GenericData.Record(avroSchema); byte[] soundBytes = "chirp,chirp".getBytes(); ByteBuffer soundByteBuffer = ByteBuffer.wrap(soundBytes); @@ -95,7 +99,9 @@ public class BigQueryAvroUtilsTest { record.put("birthday", 5L); record.put("flighted", Boolean.TRUE); record.put("sound", soundByteBuffer); - record.put("anniversary", new Utf8("2000-01-01")); + record.put("anniversaryDate", new Utf8("2000-01-01")); + record.put("anniversaryDatetime", new String("2000-01-01 00:00:00.000005")); + record.put("anniversaryTime", new Utf8("00:00:00.000005")); TableRow convertedRow = BigQueryAvroUtils.convertGenericRecordToTableRow(record, tableSchema); TableRow row = new TableRow() .set("number", "5") @@ -104,7 +110,9 @@ public class BigQueryAvroUtilsTest { .set("associates", new ArrayList<TableRow>()) .set("flighted", Boolean.TRUE) .set("sound", BaseEncoding.base64().encode(soundBytes)) - .set("anniversary", "2000-01-01"); + .set("anniversaryDate", "2000-01-01") + .set("anniversaryDatetime", "2000-01-01 00:00:00.000005") + .set("anniversaryTime", "00:00:00.000005"); assertEquals(row, convertedRow); } { @@ -137,7 +145,9 @@ public class BigQueryAvroUtilsTest { @Nullable Long birthday; // Exercises TIMESTAMP. @Nullable Boolean flighted; @Nullable ByteBuffer sound; - @Nullable Utf8 anniversary; + @Nullable Utf8 anniversaryDate; + @Nullable String anniversaryDatetime; + @Nullable Utf8 anniversaryTime; @Nullable SubBird scion; SubBird[] associates; http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/2bfc89ea/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryTableRowIteratorTest.java ---------------------------------------------------------------------- diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryTableRowIteratorTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryTableRowIteratorTest.java index 7bd24df..040f884 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryTableRowIteratorTest.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryTableRowIteratorTest.java @@ -125,7 +125,9 @@ public class BigQueryTableRowIteratorTest { new TableFieldSchema().setName("name").setType("STRING"), new TableFieldSchema().setName("answer").setType("INTEGER"), new TableFieldSchema().setName("photo").setType("BYTES"), - new TableFieldSchema().setName("anniversary").setType("DATE")))); + new TableFieldSchema().setName("anniversary_date").setType("DATE"), + new TableFieldSchema().setName("anniversary_datetime").setType("DATETIME"), + new TableFieldSchema().setName("anniversary_time").setType("TIME")))); } private TableRow rawRow(Object... args) { @@ -168,11 +170,13 @@ public class BigQueryTableRowIteratorTest { byte[] photoBytes = "photograph".getBytes(); String photoBytesEncoded = BaseEncoding.base64().encode(photoBytes); // Mock table data fetch. - when(mockTabledataList.execute()) - .thenReturn(rawDataList(rawRow("Arthur", 42, photoBytesEncoded, "2000-01-01"))); + when(mockTabledataList.execute()).thenReturn( + rawDataList(rawRow("Arthur", 42, photoBytesEncoded, + "2000-01-01", "2000-01-01 00:00:00.000005", "00:00:00.000005"))); // Run query and verify - String query = "SELECT name, count, photo, anniversary from table"; + String query = "SELECT name, count, photo, anniversary_date, " + + "anniversary_datetime, anniversary_time from table"; try (BigQueryTableRowIterator iterator = BigQueryTableRowIterator.fromQuery(query, "project", mockClient, null)) { iterator.open(); @@ -182,11 +186,15 @@ public class BigQueryTableRowIteratorTest { assertTrue(row.containsKey("name")); assertTrue(row.containsKey("answer")); assertTrue(row.containsKey("photo")); - assertTrue(row.containsKey("anniversary")); + assertTrue(row.containsKey("anniversary_date")); + assertTrue(row.containsKey("anniversary_datetime")); + assertTrue(row.containsKey("anniversary_time")); assertEquals("Arthur", row.get("name")); assertEquals(42, row.get("answer")); assertEquals(photoBytesEncoded, row.get("photo")); - assertEquals("2000-01-01", row.get("anniversary")); + assertEquals("2000-01-01", row.get("anniversary_date")); + assertEquals("2000-01-01 00:00:00.000005", row.get("anniversary_datetime")); + assertEquals("00:00:00.000005", row.get("anniversary_time")); assertFalse(iterator.advance()); }
