PLUGIN-1893 : Adding flag via field for backward compatibility of timestamp

Author: sahusanket

oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleConnector.java

@@ -112,7 +112,7 @@ protected DBConnectorPath getDBConnectorPath(String path) {
 
   @Override
   protected SchemaReader getSchemaReader(String sessionID) {
-    return new OracleSourceSchemaReader(sessionID);
+    return new OracleSourceSchemaReader(sessionID, config.getTreatAsOldTimestampConn());
   }
 
   @Override

oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleConnectorConfig.java

@@ -22,8 +22,6 @@
 import io.cdap.plugin.db.TransactionIsolationLevel;
 import io.cdap.plugin.db.connector.AbstractDBSpecificConnectorConfig;
 
-import java.util.HashMap;
-import java.util.Map;
 import java.util.Properties;
 import javax.annotation.Nullable;
 
@@ -43,12 +41,12 @@ public OracleConnectorConfig(String host, int port, String user, String password
 
   public OracleConnectorConfig(String host, int port, String user, String password, String jdbcPluginName,
                                String connectionArguments, String connectionType, String database) {
-    this(host, port, user, password, jdbcPluginName, connectionArguments, connectionType, database, null, null);
+    this(host, port, user, password, jdbcPluginName, connectionArguments, connectionType, database, null, null, null);
   }
 
   public OracleConnectorConfig(String host, int port, String user, String password, String jdbcPluginName,
                                String connectionArguments, String connectionType, String database,
-                               String role, Boolean useSSL) {
+                               String role, Boolean useSSL, String treatAsOldTimestampConn) {
 
     this.host = host;
     this.port = port;
@@ -60,6 +58,7 @@ public OracleConnectorConfig(String host, int port, String user, String password
     this.database = database;
     this.role = role;
     this.useSSL = useSSL;
+    this.treatAsOldTimestampConn = treatAsOldTimestampConn;
   }
 
   @Override
@@ -86,6 +85,11 @@ public String getConnectionString() {
   @Nullable
   public Boolean useSSL;
 
+  @Name(OracleConstants.TREAT_AS_OLD_TIMESTAMP_CONN)
+  @Description("A hidden field to handle timestamp as CDAP's timestamp micros or string as per old behavior.")
+  @Nullable
+  public String treatAsOldTimestampConn;
+
   @Override
   protected int getDefaultPort() {
     return 1521;
@@ -108,6 +112,10 @@ public Boolean getSSlMode() {
     return useSSL != null && useSSL;
   }
 
+  public Boolean getTreatAsOldTimestampConn() {
+    return Boolean.parseBoolean(treatAsOldTimestampConn);
+  }
+
   @Override
   public Properties getConnectionArgumentsProperties() {
     Properties prop = super.getConnectionArgumentsProperties();

oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleConstants.java

@@ -43,6 +43,8 @@ private OracleConstants() {
   public static final String TNS_CONNECTION_TYPE = "tns";
   public static final String TRANSACTION_ISOLATION_LEVEL = "transactionIsolationLevel";
   public static final String USE_SSL = "useSSL";
+  public static final String TREAT_AS_OLD_TIMESTAMP = "treatAsOldTimestamp";
+  public static final String TREAT_AS_OLD_TIMESTAMP_CONN = "treatAsOldTimestampConn";
 
   /**
    * Constructs the Oracle connection string based on the provided connection type, host, port, and database.

oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSource.java

@@ -63,7 +63,18 @@ protected String createConnectionString() {
 
   @Override
   protected SchemaReader getSchemaReader() {
-    return new OracleSourceSchemaReader();
+    //PLUGIN-1893 : Based on field/properties from Oracle source and Oracle connection we will pass the flag to control
+    //handle schema to make it backward compatible.
+    boolean treatAsOldTimestamp = oracleSourceConfig.getConnection().getTreatAsOldTimestampConn();
+
+    // If a user has modified the Oracle Source's properties, then it will take precedence.
+    // In case of default value (which is false), the Conn's value should be considered.
+    if (sourceConfig.getTreatAsOldTimestamp().equalsIgnoreCase("true") ||
+      sourceConfig.getTreatAsOldTimestamp().equalsIgnoreCase("false")) {
+      treatAsOldTimestamp = Boolean.parseBoolean(sourceConfig.getTreatAsOldTimestamp());
+    }
+
+    return new OracleSourceSchemaReader(null, treatAsOldTimestamp);
   }
 
   @Override
@@ -123,20 +134,27 @@ public static class OracleSourceConfig extends AbstractDBSpecificSourceConfig {
     @Nullable
     private Integer defaultRowPrefetch;
 
+    @Name(OracleConstants.TREAT_AS_OLD_TIMESTAMP)
+    @Description("A hidden field to handle timestamp as CDAP's timestamp micros or string as per old behavior.")
+    @Nullable
+    private String treatAsOldTimestamp;
+
     public OracleSourceConfig(String host, int port, String user, String password, String jdbcPluginName,
                               String connectionArguments, String connectionType, String database, String role,
                               int defaultBatchValue, int defaultRowPrefetch,
                               String importQuery, Integer numSplits, int fetchSize,
-                              String boundingQuery, String splitBy, Boolean useSSL) {
+                              String boundingQuery, String splitBy, Boolean useSSL, String treatAsOldTimestampConn,
+                              String treatAsOldTimestamp) {
       this.connection = new OracleConnectorConfig(host, port, user, password, jdbcPluginName, connectionArguments,
-                                                  connectionType, database, role, useSSL);
+                                                  connectionType, database, role, useSSL, treatAsOldTimestampConn);
       this.defaultBatchValue = defaultBatchValue;
       this.defaultRowPrefetch = defaultRowPrefetch;
       this.fetchSize = fetchSize;
       this.importQuery = importQuery;
       this.numSplits = numSplits;
       this.boundingQuery = boundingQuery;
       this.splitBy = splitBy;
+      this.treatAsOldTimestamp = treatAsOldTimestamp;
     }
 
     @Override
@@ -163,6 +181,10 @@ public OracleConnectorConfig getConnection() {
       return connection;
     }
 
+    public String getTreatAsOldTimestamp() {
+      return treatAsOldTimestamp == null ? "" : treatAsOldTimestamp;
+    }
+
     @Override
     public void validate(FailureCollector collector) {
       ConfigUtil.validateConnection(this, useConnection, connection, collector);

oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSourceSchemaReader.java

@@ -65,14 +65,14 @@ public class OracleSourceSchemaReader extends CommonSchemaReader {
   );
 
   private final String sessionID;
+  private final Boolean isTimestampOldBehavior;
 
   public OracleSourceSchemaReader() {
-    this(null);
+    this(null, false);
   }
-
-  public OracleSourceSchemaReader(String sessionID) {
-    super();
+  public OracleSourceSchemaReader(String sessionID, boolean isTimestampOldBehavior) {
     this.sessionID = sessionID;
+    this.isTimestampOldBehavior = isTimestampOldBehavior;
   }
 
   @Override
@@ -81,10 +81,12 @@ public Schema getSchema(ResultSetMetaData metadata, int index) throws SQLExcepti
 
     switch (sqlType) {
       case TIMESTAMP_TZ:
-        return Schema.of(Schema.LogicalType.TIMESTAMP_MICROS);
-      case Types.TIMESTAMP:
+        return isTimestampOldBehavior ? Schema.of(Schema.Type.STRING) : Schema.of(Schema.LogicalType.TIMESTAMP_MICROS);
       case TIMESTAMP_LTZ:
-        return Schema.of(Schema.LogicalType.DATETIME);
+        return isTimestampOldBehavior ? Schema.of(Schema.LogicalType.TIMESTAMP_MICROS)
+          : Schema.of(Schema.LogicalType.DATETIME);
+      case Types.TIMESTAMP:
+        return isTimestampOldBehavior ? super.getSchema(metadata, index) : Schema.of(Schema.LogicalType.DATETIME);
       case BINARY_FLOAT:
         return Schema.of(Schema.Type.FLOAT);
       case BINARY_DOUBLE:

oracle-plugin/widgets/Oracle-batchsource.json

@@ -246,6 +246,15 @@
             "default": "40",
             "min": "1"
           }
+        },
+        {
+          "widget-type": "hidden",
+          "label": "Treat as old timestamp.",
+          "name": "treatAsOldTimestamp",
+          "description": "If edited, then this will take precedence over other properties and the edited value shall be true or false only",
+          "widget-attributes": {
+            "default": "default-no"
+          }
         }
       ]
     }

oracle-plugin/widgets/Oracle-connector.json

@@ -129,6 +129,14 @@
               }
             ]
           }
+        },
+        {
+          "widget-type": "hidden",
+          "label": "Treat as old timestamp",
+          "name": "treatAsOldTimestampConn",
+          "widget-attributes": {
+            "default": "false"
+          }
         }
       ]
     },