[cherry-pick][6.10][PLUGIN-1893] Adding fields in Oracle source and connector which acts…

amazon-redshift-plugin/pom.xml

@@ -20,7 +20,7 @@
   <parent>
     <artifactId>database-plugins-parent</artifactId>
     <groupId>io.cdap.plugin</groupId>
-    <version>1.11.7</version>
+    <version>1.11.8-SNAPSHOT</version>
   </parent>
 
   <name>Amazon Redshift plugin</name>

aurora-mysql-plugin/pom.xml

@@ -20,7 +20,7 @@
   <parent>
     <artifactId>database-plugins-parent</artifactId>
     <groupId>io.cdap.plugin</groupId>
-    <version>1.11.7</version>
+    <version>1.11.8-SNAPSHOT</version>
   </parent>
 
   <name>Aurora DB MySQL plugin</name>

aurora-postgresql-plugin/pom.xml

@@ -20,7 +20,7 @@
   <parent>
     <artifactId>database-plugins-parent</artifactId>
     <groupId>io.cdap.plugin</groupId>
-    <version>1.11.7</version>
+    <version>1.11.8-SNAPSHOT</version>
   </parent>
 
   <name>Aurora DB PostgreSQL plugin</name>

cloudsql-mysql-plugin/pom.xml

@@ -20,7 +20,7 @@
   <parent>
     <artifactId>database-plugins-parent</artifactId>
     <groupId>io.cdap.plugin</groupId>
-    <version>1.11.7</version>
+    <version>1.11.8-SNAPSHOT</version>
   </parent>
 
   <name>CloudSQL MySQL plugin</name>

cloudsql-postgresql-plugin/pom.xml

@@ -20,7 +20,7 @@
   <parent>
     <artifactId>database-plugins-parent</artifactId>
     <groupId>io.cdap.plugin</groupId>
-    <version>1.11.7</version>
+    <version>1.11.8-SNAPSHOT</version>
   </parent>
 
   <name>CloudSQL PostgreSQL plugin</name>

database-commons/pom.xml

@@ -20,7 +20,7 @@
   <parent>
     <artifactId>database-plugins-parent</artifactId>
     <groupId>io.cdap.plugin</groupId>
-    <version>1.11.7</version>
+    <version>1.11.8-SNAPSHOT</version>
   </parent>
 
   <name>Database Commons</name>

db2-plugin/pom.xml

@@ -20,7 +20,7 @@
   <parent>
     <artifactId>database-plugins-parent</artifactId>
     <groupId>io.cdap.plugin</groupId>
-    <version>1.11.7</version>
+    <version>1.11.8-SNAPSHOT</version>
   </parent>
 
   <name>IBM DB2 plugin</name>

generic-database-plugin/pom.xml

@@ -20,7 +20,7 @@
   <parent>
     <artifactId>database-plugins-parent</artifactId>
     <groupId>io.cdap.plugin</groupId>
-    <version>1.11.7</version>
+    <version>1.11.8-SNAPSHOT</version>
   </parent>
 
   <name>Generic database plugin</name>

generic-db-argument-setter/pom.xml

@@ -20,7 +20,7 @@
   <parent>
     <artifactId>database-plugins-parent</artifactId>
     <groupId>io.cdap.plugin</groupId>
-    <version>1.11.7</version>
+    <version>1.11.8-SNAPSHOT</version>
   </parent>
 
   <name>Generic database argument setter plugin</name>

mariadb-plugin/pom.xml

@@ -20,7 +20,7 @@
   <parent>
     <artifactId>database-plugins-parent</artifactId>
     <groupId>io.cdap.plugin</groupId>
-    <version>1.11.7</version>
+    <version>1.11.8-SNAPSHOT</version>
   </parent>
 
   <name>Maria DB plugin</name>

memsql-plugin/pom.xml

@@ -20,7 +20,7 @@
   <parent>
     <artifactId>database-plugins-parent</artifactId>
     <groupId>io.cdap.plugin</groupId>
-    <version>1.11.7</version>
+    <version>1.11.8-SNAPSHOT</version>
   </parent>
 
   <name>Memsql plugin</name>

mssql-plugin/pom.xml

@@ -20,7 +20,7 @@
   <parent>
     <artifactId>database-plugins-parent</artifactId>
     <groupId>io.cdap.plugin</groupId>
-    <version>1.11.7</version>
+    <version>1.11.8-SNAPSHOT</version>
   </parent>
 
   <name>Microsoft SQL Server plugin</name>

mysql-plugin/pom.xml

@@ -20,7 +20,7 @@
   <parent>
     <artifactId>database-plugins-parent</artifactId>
     <groupId>io.cdap.plugin</groupId>
-    <version>1.11.7</version>
+    <version>1.11.8-SNAPSHOT</version>
   </parent>
 
   <name>Mysql plugin</name>

netezza-plugin/pom.xml

@@ -20,7 +20,7 @@
   <parent>
     <artifactId>database-plugins-parent</artifactId>
     <groupId>io.cdap.plugin</groupId>
-    <version>1.11.7</version>
+    <version>1.11.8-SNAPSHOT</version>
   </parent>
 
   <name>Netezza plugin</name>

oracle-plugin/pom.xml

@@ -20,7 +20,7 @@
   <parent>
     <artifactId>database-plugins-parent</artifactId>
     <groupId>io.cdap.plugin</groupId>
-    <version>1.11.7</version>
+    <version>1.11.8-SNAPSHOT</version>
   </parent>
 
   <name>Oracle plugin</name>

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

@@ -112,7 +112,8 @@ protected DBConnectorPath getDBConnectorPath(String path) {
 
   @Override
   protected SchemaReader getSchemaReader(String sessionID) {
-    return new OracleSourceSchemaReader(sessionID);
+    return new OracleSourceSchemaReader(sessionID, config.getTreatAsOldTimestamp(),
+                                        config.getTreatPrecisionlessNumAsDeci());
   }
 
   @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,14 @@ 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,
+         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, @Nullable Boolean treatAsOldTimestamp,
+                               @Nullable Boolean treatPrecisionlessNumAsDeci) {
 
     this.host = host;
     this.port = port;
@@ -60,6 +60,8 @@ public OracleConnectorConfig(String host, int port, String user, String password
     this.database = database;
     this.role = role;
     this.useSSL = useSSL;
+    this.treatAsOldTimestamp = treatAsOldTimestamp;
+    this.treatPrecisionlessNumAsDeci = treatPrecisionlessNumAsDeci;
   }
 
   @Override
@@ -86,6 +88,16 @@ public String getConnectionString() {
   @Nullable
   public Boolean useSSL;
 
+  @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
+  public Boolean treatAsOldTimestamp;
+
+  @Name(OracleConstants.TREAT_PRECISIONLESSNUM_AS_DECI)
+  @Description("A hidden field to handle precision less number as CDAP's decimal per old behavior.")
+  @Nullable
+  public Boolean treatPrecisionlessNumAsDeci;
+
   @Override
   protected int getDefaultPort() {
     return 1521;
@@ -108,6 +120,14 @@ public Boolean getSSlMode() {
     return useSSL != null && useSSL;
   }
 
+  public Boolean getTreatAsOldTimestamp() {
+    return Boolean.TRUE.equals(treatAsOldTimestamp);
+  }
+
+  public Boolean getTreatPrecisionlessNumAsDeci() {
+    return Boolean.TRUE.equals(treatPrecisionlessNumAsDeci);
+  }
+
   @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_PRECISIONLESSNUM_AS_DECI = "treatPrecisionlessNumAsDeci";
 
   /**
    * 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,12 @@ 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().getTreatAsOldTimestamp();
+    boolean treatPrecisionlessNumAsDeci = oracleSourceConfig.getConnection().getTreatPrecisionlessNumAsDeci();
+
+    return new OracleSourceSchemaReader(null, treatAsOldTimestamp, treatPrecisionlessNumAsDeci);
   }
 
   @Override
@@ -117,9 +122,11 @@ public OracleSourceConfig(String host, int port, String user, String password, S
                               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, Boolean treatAsOldTimestamp,
+                              Boolean treatPrecisionlessNumAsDeci) {
       this.connection = new OracleConnectorConfig(host, port, user, password, jdbcPluginName, connectionArguments,
-                                                  connectionType, database, role, useSSL);
+                                                  connectionType, database, role, useSSL, treatAsOldTimestamp,
+                                                  treatPrecisionlessNumAsDeci);
       this.defaultBatchValue = defaultBatchValue;
       this.defaultRowPrefetch = defaultRowPrefetch;
       this.fetchSize = fetchSize;

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

@@ -26,6 +26,7 @@
 import java.sql.SQLException;
 import java.sql.Types;
 import java.util.Set;
+import javax.annotation.Nullable;
 
 /**
  * Oracle Source schema reader.
@@ -65,14 +66,17 @@ public class OracleSourceSchemaReader extends CommonSchemaReader {
   );
 
   private final String sessionID;
+  private final Boolean isTimestampOldBehavior;
+  private final Boolean isPrecisionlessNumAsDecimal;
 
   public OracleSourceSchemaReader() {
-    this(null);
+    this(null, false, false);
   }
-
-  public OracleSourceSchemaReader(String sessionID) {
-    super();
+  public OracleSourceSchemaReader(@Nullable String sessionID, boolean isTimestampOldBehavior,
+                                  boolean isPrecisionlessNumAsDecimal) {
     this.sessionID = sessionID;
+    this.isTimestampOldBehavior = isTimestampOldBehavior;
+    this.isPrecisionlessNumAsDecimal = isPrecisionlessNumAsDecimal;
   }
 
   @Override
@@ -81,10 +85,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:
@@ -107,12 +113,24 @@ public Schema getSchema(ResultSetMetaData metadata, int index) throws SQLExcepti
           // For a Number type without specified precision and scale, precision will be 0 and scale will be -127
           if (precision == 0) {
             // reference : https://docs.oracle.com/cd/B28359_01/server.111/b28318/datatype.htm#CNCPT1832
-            LOG.warn(String.format("Field '%s' is a %s type without precision and scale, "
-                    + "converting into STRING type to avoid any precision loss.",
-                metadata.getColumnName(index),
-                metadata.getColumnTypeName(index),
-                metadata.getColumnName(index)));
-            return Schema.of(Schema.Type.STRING);
+            if (isPrecisionlessNumAsDecimal) {
+              precision = 38;
+              scale = 0;
+              LOG.warn(String.format("%s type with undefined precision and scale is detected, "
+                                       + "there may be a precision loss while running the pipeline. "
+                                       + "Please define an output precision and scale for field '%s' to avoid "
+                                       + "precision loss.",
+                                     metadata.getColumnTypeName(index),
+                                     metadata.getColumnName(index)));
+              return Schema.decimalOf(precision, scale);
+            } else {
+              LOG.warn(String.format("Field '%s' is a %s type without precision and scale, "
+                                       + "converting into STRING type to avoid any precision loss.",
+                                     metadata.getColumnName(index),
+                                     metadata.getColumnTypeName(index),
+                                     metadata.getColumnName(index)));
+              return Schema.of(Schema.Type.STRING);
+            }
           }
           return Schema.decimalOf(precision, scale);
         }

oracle-plugin/widgets/Oracle-batchsource.json

@@ -120,6 +120,44 @@
             ]
           }
         },
+        {
+          "widget-type": "hidden",
+          "label": "Treat as old timestamp",
+          "name": "treatAsOldTimestamp",
+          "widget-attributes": {
+            "layout": "inline",
+            "default": "false",
+            "options": [
+              {
+                "id": "true",
+                "label": "true"
+              },
+              {
+                "id": "false",
+                "label": "false"
+              }
+            ]
+          }
+        },
+        {
+          "widget-type": "hidden",
+          "label": "Treat precision less number as Decimal(old behavior)",
+          "name": "treatPrecisionlessNumAsDeci",
+          "widget-attributes": {
+            "layout": "inline",
+            "default": "false",
+            "options": [
+              {
+                "id": "true",
+                "label": "true"
+              },
+              {
+                "id": "false",
+                "label": "false"
+              }
+            ]
+          }
+        },
         {
           "name": "connectionType",
           "label": "Connection Type",
@@ -326,6 +364,14 @@
         {
           "type": "property",
           "name": "transactionIsolationLevel"
+        },
+        {
+          "type": "property",
+          "name": "getTreatAsOldTimestampConn"
+        },
+        {
+          "type": "property",
+          "name": "treatPrecisionlessNumAsDeci"
         }
       ]
     },