Added Support for proper handling of Timestamp, TimestampLTZ, TimestampTZ types (#386)

MrRahulSharma · web-flow · commit 66bf1bac3581 · 2023-05-04T16:03:21.000+05:30
ORACLE Timestamp : CDAP Datetime
ORACLE TimestampLTZ : CDAP Datetime
ORACLE TimestampTZ : CDAP Timestamp
diff --git a/oracle-plugin/src/e2e-test/features/oracle/DatatypeTimestamp.feature b/oracle-plugin/src/e2e-test/features/oracle/DatatypeTimestamp.feature
@@ -0,0 +1,67 @@
+#
+# Copyright © 2023 Cask Data, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not
+# use this file except in compliance with the License. You may obtain a copy of
+# the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations under
+# the License.
+#
+
+@Oracle
+Feature: Oracle - Verify Oracle source data transfer for all Timestamp types
+  @ORACLE_SOURCE_DATATYPE_TIMESTAMP @ORACLE_SINK_TEST @Oracle_Required
+  Scenario: To verify data is getting transferred from Oracle to Oracle successfully
+    Given Open Datafusion Project to configure pipeline
+    When Expand Plugin group in the LHS plugins list: "Source"
+    When Select plugin: "Oracle" from the plugins list as: "Source"
+    When Expand Plugin group in the LHS plugins list: "Sink"
+    When Select plugin: "Oracle" from the plugins list as: "Sink"
+    Then Connect plugins: "Oracle" and "Oracle2" to establish connection
+    Then Navigate to the properties page of plugin: "Oracle"
+    Then Select dropdown plugin property: "select-jdbcPluginName" with option value: "driverName"
+    Then Replace input plugin property: "host" with value: "host" for Credentials and Authorization related fields
+    Then Replace input plugin property: "port" with value: "port" for Credentials and Authorization related fields
+    Then Replace input plugin property: "user" with value: "username" for Credentials and Authorization related fields
+    Then Replace input plugin property: "password" with value: "password" for Credentials and Authorization related fields
+    Then Select radio button plugin property: "connectionType" with value: "service"
+    Then Select radio button plugin property: "role" with value: "sysdba"
+    Then Enter input plugin property: "referenceName" with value: "sourceRef"
+    Then Replace input plugin property: "database" with value: "databaseName"
+    Then Enter textarea plugin property: "importQuery" with value: "selectQuery"
+    Then Click on the Get Schema button
+    Then Verify the Output Schema matches the Expected Schema: "outputTimestampDatatypesSchema"
+    Then Validate "Oracle" plugin properties
+    Then Close the Plugin Properties page
+    Then Navigate to the properties page of plugin: "Oracle2"
+    Then Select dropdown plugin property: "select-jdbcPluginName" with option value: "driverName"
+    Then Replace input plugin property: "host" with value: "host" for Credentials and Authorization related fields
+    Then Replace input plugin property: "port" with value: "port" for Credentials and Authorization related fields
+    Then Replace input plugin property: "database" with value: "databaseName"
+    Then Replace input plugin property: "tableName" with value: "targetTable"
+    Then Replace input plugin property: "dbSchemaName" with value: "schema"
+    Then Replace input plugin property: "user" with value: "username" for Credentials and Authorization related fields
+    Then Replace input plugin property: "password" with value: "password" for Credentials and Authorization related fields
+    Then Enter input plugin property: "referenceName" with value: "targetRef"
+    Then Select radio button plugin property: "connectionType" with value: "service"
+    Then Select radio button plugin property: "role" with value: "sysdba"
+    Then Validate "Oracle2" plugin properties
+    Then Close the Plugin Properties page
+    Then Save the pipeline
+    Then Preview and run the pipeline
+    Then Verify the preview of pipeline is "success"
+    Then Click on preview data for Oracle sink
+    Then Verify preview output schema matches the outputSchema captured in properties
+    Then Close the preview data
+    Then Deploy the pipeline
+    Then Run the Pipeline in Runtime
+    Then Wait till pipeline is in running state
+    Then Open and capture logs
+    Then Verify the pipeline status is "Succeeded"
+    Then Validate the values of records transferred to target table is equal to the values from source table
diff --git a/oracle-plugin/src/e2e-test/java/io.cdap.plugin/OracleClient.java b/oracle-plugin/src/e2e-test/java/io.cdap.plugin/OracleClient.java
@@ -16,9 +16,12 @@
 
 package io.cdap.plugin;
 
+import com.google.common.base.Strings;
 import io.cdap.e2e.utils.PluginPropertyUtils;
 import io.cdap.plugin.oracle.OracleSourceSchemaReader;
 import org.junit.Assert;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.sql.Blob;
 import java.sql.Clob;
@@ -30,6 +33,7 @@
 import java.sql.Statement;
 import java.sql.Timestamp;
 import java.sql.Types;
+import java.time.Instant;
 import java.util.Arrays;
 import java.util.Date;
 import java.util.GregorianCalendar;
@@ -112,30 +116,51 @@ private static boolean compareResultSetData(ResultSet rsSource, ResultSet rsTarg
             byte[] sourceArrayBlob = blobSource.getBytes(1, (int) blobSource.length());
             Blob blobTarget = rsTarget.getBlob(currentColumnCount);
             byte[] targetArrayBlob = blobTarget.getBytes(1, (int) blobTarget.length());
-            Assert.assertTrue(String.format("Different values found for column : %s", columnName),
+            Assert.assertTrue(String.format("Different BLOB values found for column : %s", columnName),
                               Arrays.equals(sourceArrayBlob, targetArrayBlob));
             break;
           case Types.CLOB:
             Clob clobSource = rsSource.getClob(currentColumnCount);
             String sourceClobString = clobSource.getSubString(1, (int) clobSource.length());
             Clob clobTarget = rsTarget.getClob(currentColumnCount);
             String targetClobString = clobTarget.getSubString(1, (int) clobTarget.length());
-            Assert.assertTrue(String.format("Different values found for column : %s", columnName),
-                                sourceClobString.equals(targetClobString));
+            Assert.assertEquals(String.format("Different CLOB values found for column : %s", columnName),
+                                sourceClobString, targetClobString);
             break;
           case Types.TIMESTAMP:
             GregorianCalendar gc = new GregorianCalendar(TimeZone.getTimeZone("UTC"));
             gc.setGregorianChange(new Date(Long.MIN_VALUE));
             Timestamp sourceTS = rsSource.getTimestamp(currentColumnCount, gc);
             Timestamp targetTS = rsTarget.getTimestamp(currentColumnCount, gc);
-            Assert.assertTrue(String.format("Different values found for column : %s", columnName),
-                                sourceTS.equals(targetTS));
+            Assert.assertEquals(String.format("Different TIMESTAMP values found for column : %s", columnName),
+                                sourceTS, targetTS);
+            break;
+          case OracleSourceSchemaReader.TIMESTAMP_TZ:
+            // The timezone information in the field is lost during pipeline execution hence it is required to
+            // convert both values into the system timezone and then compare.
+            GregorianCalendar gregorianCalendar = new GregorianCalendar(TimeZone.getTimeZone("UTC"));
+            gregorianCalendar.setGregorianChange(new Date(Long.MIN_VALUE));
+            Timestamp tsSource = rsSource.getTimestamp(currentColumnCount, gregorianCalendar);
+            Timestamp tsTarget = rsTarget.getTimestamp(currentColumnCount, gregorianCalendar);
+            if (tsSource == null && tsTarget == null) {
+              break;
+            }
+            Assert.assertNotNull(
+                    String.format("Column : %s is null in source table and is not Null in target table.", columnName),
+                    tsSource);
+            Assert.assertNotNull(
+                    String.format("Column : %s is null in target table and is not Null in source table.", columnName),
+                    tsTarget);
+            Instant sourceInstant = tsSource.toInstant();
+            Instant targetInstant = tsTarget.toInstant();
+            Assert.assertEquals(String.format("Different TIMESTAMPTZ values found for column : %s", columnName),
+                    sourceInstant, targetInstant);
             break;
           default:
             String sourceString = rsSource.getString(currentColumnCount);
             String targetString = rsTarget.getString(currentColumnCount);
-            Assert.assertTrue(String.format("Different values found for column : %s", columnName),
-                                String.valueOf(sourceString).equals(String.valueOf(targetString)));
+            Assert.assertEquals(String.format("Different %s values found for column : %s", columnTypeName, columnName),
+                                String.valueOf(sourceString), String.valueOf(targetString));
         }
         currentColumnCount++;
       }
@@ -217,6 +242,34 @@ public static void createTargetLongTable(String targetTable, String schema) thro
     }
   }
 
+  public static void createTimestampSourceTable(String sourceTable, String schema) throws SQLException,
+          ClassNotFoundException {
+    try (Connection connect = getOracleConnection(); Statement statement = connect.createStatement()) {
+      String timestampColumns = PluginPropertyUtils.pluginProp("timestampColumns");
+      String createSourceTableQuery = "CREATE TABLE " + schema + "." + sourceTable + " " + timestampColumns;
+      statement.executeUpdate(createSourceTableQuery);
+
+      int rowCount = 1;
+      while (!Strings.isNullOrEmpty(PluginPropertyUtils.pluginProp("timestampValue" + rowCount))) {
+        // Insert dummy data.
+        String timestampValue = PluginPropertyUtils.pluginProp("timestampValue" + rowCount);
+        String timestampColumnsList = PluginPropertyUtils.pluginProp("timestampColumnsList");
+        statement.executeUpdate("INSERT INTO " + schema + "." + sourceTable + " " + timestampColumnsList + " " +
+                timestampValue);
+        rowCount++;
+      }
+    }
+  }
+
+  public static void createTimestampTargetTable(String targetTable, String schema) throws SQLException,
+          ClassNotFoundException {
+    try (Connection connect = getOracleConnection(); Statement statement = connect.createStatement()) {
+      String timestampColumns = PluginPropertyUtils.pluginProp("timestampColumns");
+      String createTargetTableQuery = "CREATE TABLE " + schema + "." + targetTable + " " + timestampColumns;
+      statement.executeUpdate(createTargetTableQuery);
+    }
+  }
+
   public static void createSourceLongRawTable(String sourceTable, String schema) throws SQLException,
     ClassNotFoundException {
     try (Connection connect = getOracleConnection(); Statement statement = connect.createStatement()) {
diff --git a/oracle-plugin/src/e2e-test/java/io.cdap.plugin/common.stepsdesign/TestSetupHooks.java b/oracle-plugin/src/e2e-test/java/io.cdap.plugin/common.stepsdesign/TestSetupHooks.java
@@ -81,6 +81,14 @@ public static void createLongVarcharTables() throws SQLException, ClassNotFoundE
                                             PluginPropertyUtils.pluginProp("schema"));
   }
 
+  @Before(order = 2, value = "@ORACLE_SOURCE_DATATYPE_TIMESTAMP")
+  public static void createTimestampDatatypeTables() throws SQLException, ClassNotFoundException {
+    OracleClient.createTimestampSourceTable(PluginPropertyUtils.pluginProp("sourceTable"),
+            PluginPropertyUtils.pluginProp("schema"));
+    OracleClient.createTimestampTargetTable(PluginPropertyUtils.pluginProp("targetTable"),
+            PluginPropertyUtils.pluginProp("schema"));
+  }
+
   @After(order = 1, value = "@ORACLE_SINK_TEST")
   public static void dropTables() throws SQLException, ClassNotFoundException {
     OracleClient.deleteTables(PluginPropertyUtils.pluginProp("schema"),
diff --git a/oracle-plugin/src/e2e-test/java/io.cdap.plugin/oracle/stepsdesign/Oracle.java b/oracle-plugin/src/e2e-test/java/io.cdap.plugin/oracle/stepsdesign/Oracle.java
@@ -23,9 +23,7 @@
 import org.junit.Assert;
 import stepsdesign.BeforeActions;
 
-import java.sql.Connection;
 import java.sql.SQLException;
-import java.sql.Statement;
 
 /**
  *  Oracle Plugin related step design.
diff --git a/oracle-plugin/src/e2e-test/resources/pluginParameters.properties b/oracle-plugin/src/e2e-test/resources/pluginParameters.properties
@@ -34,9 +34,9 @@ outputDatatypesSchema=[{"key":"ID","value":"string"},{"key":"COL1","value":"stri
   {"key":"COL18","value":"decimal"},{"key":"COL19","value":"decimal"},{"key":"COL20","value":"decimal"},\
   {"key":"COL21","value":"decimal"},{"key":"COL22","value":"double"},{"key":"COL23","value":"double"},\
   {"key":"COL24","value":"decimal"},{"key":"COL25","value":"double"},{"key":"COL26","value":"double"},\
-  {"key":"COL27","value":"decimal"},{"key":"COL28","value":"timestamp"},{"key":"COL29","value":"timestamp"},\
-  {"key":"COL30","value":"string"},{"key":"COL31","value":"string"},{"key":"COL32","value":"string"},\
-  {"key":"COL33","value":"timestamp"},{"key":"COL34","value":"float"},{"key":"COL35","value":"double"}]
+  {"key":"COL27","value":"decimal"},{"key":"COL28","value":"datetime"},{"key":"COL29","value":"datetime"},\
+  {"key":"COL30","value":"timestamp"},{"key":"COL31","value":"string"},{"key":"COL32","value":"string"},\
+  {"key":"COL33","value":"datetime"},{"key":"COL34","value":"float"},{"key":"COL35","value":"double"}]
 
 longColumns=(ID VARCHAR2(100) PRIMARY KEY, COL1 LONG, COL2 RAW(2), COL3 BLOB, COL4 CLOB, COL5 NCLOB, COL6 BFILE)
 longColumnsList=(ID,COL1,COL2,COL3,COL4,COL5,COL6)
@@ -55,3 +55,21 @@ longVarcharColumns=(ID VARCHAR2(100) PRIMARY KEY, COL1 LONG VARCHAR)
 longVarcharColumnsList=(ID,COL1)
 longVarcharValues=VALUES ('User1','48692054686572652120486F772061726520796F75206665656C696E67AbCdEF646179203F')
 outputDatatypesSchema4=[{"key":"ID","value":"string"},{"key":"COL1","value":"string"}]
+
+timestampColumns=(ID VARCHAR2(100) PRIMARY KEY, COL1 TIMESTAMP, COL2 TIMESTAMP WITH TIME ZONE,\
+   COL3 TIMESTAMP WITH LOCAL TIME ZONE)
+timestampColumnsList=(ID,COL1,COL2,COL3)
+timestampValue1=VALUES ('1',TIMESTAMP '2023-01-01 02:00:00.000000',\
+  TIMESTAMP '2023-01-01 02:00:00.000000 +05:30',TIMESTAMP '2001-12-31 13:37:00.000000')
+timestampValue2=VALUES ('2',TIMESTAMP '2023-01-01 02:00:00.000000',NULL,NULL)
+timestampValue3=VALUES ('3',TIMESTAMP '0001-01-01 09:00:00.000000',\
+  TIMESTAMP '0001-01-01 01:00:00.000000 -08:00',TIMESTAMP '0001-01-01 09:00:00.000000')
+timestampValue4=VALUES ('4',TIMESTAMP '0001-01-01 01:00:00.000000',\
+  TIMESTAMP '0001-01-02 01:00:00.000000 -08:00',TIMESTAMP '2022-12-31 13:37:00.000000')
+timestampValue5=VALUES ('5',TIMESTAMP '2023-01-01 01:00:00.000000',\
+  TIMESTAMP '2022-12-31 14:00:00.000000 -08:00',TIMESTAMP '2022-12-31 19:30:00.000000')
+timestampValue6=VALUES ('6',NULL,TIMESTAMP '2022-12-31 14:00:00.000000 -08:00',NULL)
+timestampValue7=VALUES ('7',NULL,TIMESTAMP '2022-12-31 14:00:00.000000 +05:30',NULL)
+timestampValue8=VALUES ('8',TIMESTAMP '0001-01-01 01:00:00.000000',NULL,NULL)
+outputTimestampDatatypesSchema=[{"key":"ID","value":"string"},{"key":"COL1","value":"datetime"},\
+  {"key":"COL2","value":"timestamp"},{"key":"COL3","value":"datetime"}]
diff --git a/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleFieldsValidator.java b/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleFieldsValidator.java
@@ -47,8 +47,12 @@ public boolean isFieldCompatible(Schema.Field field, ResultSetMetaData metadata,
                                                                    isSigned);
     }
     if (fieldLogicalType == Schema.LogicalType.TIMESTAMP_MICROS) {
-      return sqlType == OracleSinkSchemaReader.TIMESTAMP_LTZ ||
-        super.isFieldCompatible(fieldType, fieldLogicalType, sqlType, precision, isSigned);
+      return sqlType == OracleSinkSchemaReader.TIMESTAMP_LTZ
+              || sqlType == OracleSinkSchemaReader.TIMESTAMP_TZ
+          || super.isFieldCompatible(fieldType, fieldLogicalType, sqlType, precision, isSigned);
+    } else if (fieldLogicalType == Schema.LogicalType.DATETIME) {
+      return sqlType == OracleSinkSchemaReader.TIMESTAMP_LTZ
+              || sqlType == Types.TIMESTAMP;
     } else if (fieldLogicalType != null) {
       return super.isFieldCompatible(fieldType, fieldLogicalType, sqlType, precision, isSigned);
     }
diff --git a/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSource.java b/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSource.java
@@ -167,6 +167,21 @@ protected void validateField(FailureCollector collector,
           && actualFieldSchema.getType().equals(Schema.Type.STRING)) {
         return;
       }
+
+      // For handling TimestampTZ types allow if the expected schema is STRING and
+      // actual schema is set to TIMESTAMP type to ensure backward compatibility.
+      if (Schema.LogicalType.TIMESTAMP_MICROS.equals(actualFieldSchema.getLogicalType())
+          && Schema.Type.STRING.equals(expectedFieldSchema.getType())) {
+        return;
+      }
+
+      // For handling TimestampLTZ and Timestamp types allow if the expected schema is TIMESTAMP and
+      // actual schema is set to DATETIME type to ensure backward compatibility.
+      if (Schema.LogicalType.DATETIME.equals(actualFieldSchema.getLogicalType())
+              && Schema.LogicalType.TIMESTAMP_MICROS.equals(expectedFieldSchema.getLogicalType())) {
+        return;
+      }
+
       super.validateField(collector, field, actualFieldSchema, expectedFieldSchema);
     }
   }
diff --git a/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSourceDBRecord.java b/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSourceDBRecord.java
diff --git a/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSourceSchemaReader.java b/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSourceSchemaReader.java
diff --git a/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleSourceDBRecordUnitTest.java b/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleSourceDBRecordUnitTest.java
