added support for timestamp_tz datatype.

added support for timestamp_tz datatype.

Author: vikasrathee-cs

docs/Snowflake-batchsource.md

@@ -68,7 +68,7 @@ Data Types Mapping
 ----------
 
 | Snowflake Data Types           | CDAP Schema Data Type | Comment                                                   |
-| ------------------------------ | --------------------- | --------------------------------------------------------- |
+| ------------------------------ |-----------------------| --------------------------------------------------------- |
 | NUMBER                         | decimal               | Default precision and scale are (38,0).                   |
 | DECIMAL                        | decimal               | Synonymous with NUMBER.                                   |
 | NUMERIC                        | decimal               | Synonymous with NUMBER.                                   |
@@ -87,10 +87,10 @@ Data Types Mapping
 | DATE                           | date                  |                                                           |
 | DATETIME                       | timestamp             | Alias for TIMESTAMP_NTZ                                   |
 | TIME                           | time                  |                                                           |
-| TIMESTAMP                      | timestamp/string      | Alias for one of the TIMESTAMP variations (TIMESTAMP_NTZ by default).|
+| TIMESTAMP                      | timestamp             | Alias for one of the TIMESTAMP variations (TIMESTAMP_NTZ by default).|
 | TIMESTAMP_LTZ                  | timestamp             | TIMESTAMP with local time zone; time zone, if provided, is not stored.|
 | TIMESTAMP_NTZ                  | timestamp             | TIMESTAMP with no time zone; time zone, if provided, is not stored.|
-| TIMESTAMP_TZ                   | string                | TIMESTAMP with time zone.                                 |
+| TIMESTAMP_TZ                   | timestamp             | TIMESTAMP with time zone.                                 |
 | VARIANT                        | string                | A tagged universal type, which can store values of any other type, including OBJECT and ARRAY, up to a maximum size of 16 MB compressed.                                    |
 | OBJECT                         | string                | This will return a json with the data |
 | ARRAY                          | string                | This will return a json with the data |                                                  |

src/main/java/io/cdap/plugin/snowflake/common/util/SchemaHelper.java

@@ -51,6 +51,7 @@ public class SchemaHelper {
       .put(Types.BOOLEAN, Schema.of(Schema.Type.BOOLEAN))
       .put(Types.BIGINT, Schema.decimalOf(38))
       .put(Types.SMALLINT, Schema.decimalOf(38))
+      .put(Types.TIMESTAMP_WITH_TIMEZONE, Schema.of(Schema.LogicalType.TIMESTAMP_MICROS))
       .build();
 
   private SchemaHelper() {

src/test/java/io/cdap/plugin/snowflake/common/util/SchemaHelperTest.java

@@ -108,7 +108,9 @@ public void testGetSchemaFromSnowflake() throws IOException {
       new SnowflakeFieldDescriptor("field121", Types.BIGINT, true),
       new SnowflakeFieldDescriptor("field122", Types.BIGINT, false),
       new SnowflakeFieldDescriptor("field131", Types.SMALLINT, true),
-      new SnowflakeFieldDescriptor("field132", Types.SMALLINT, false));
+      new SnowflakeFieldDescriptor("field132", Types.SMALLINT, false),
+      new SnowflakeFieldDescriptor("field133", Types.TIMESTAMP_WITH_TIMEZONE, false),
+      new SnowflakeFieldDescriptor("field134", Types.TIMESTAMP_WITH_TIMEZONE, true));
 
     Schema expected = Schema.recordOf(
       "data",
@@ -135,7 +137,9 @@ public void testGetSchemaFromSnowflake() throws IOException {
       Schema.Field.of("field121", Schema.nullableOf(Schema.decimalOf(38))),
       Schema.Field.of("field122", Schema.decimalOf(38)),
       Schema.Field.of("field131", Schema.nullableOf(Schema.decimalOf(38))),
-      Schema.Field.of("field132", Schema.decimalOf(38))
+      Schema.Field.of("field132", Schema.decimalOf(38)),
+      Schema.Field.of("field133", Schema.of(Schema.LogicalType.TIMESTAMP_MICROS)),
+      Schema.Field.of("field134", Schema.nullableOf(Schema.of(Schema.LogicalType.TIMESTAMP_MICROS)))
     );
 
     Mockito.when(snowflakeAccessor.describeQuery(null)).thenReturn(sample);