Timetsamp

Author: Amit-CloudSufi

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

@@ -114,7 +114,8 @@ protected DBConnectorPath getDBConnectorPath(String path) {
   @Override
   protected SchemaReader getSchemaReader(String sessionID) {
     return new OracleSourceSchemaReader(sessionID, config.getTreatAsOldTimestamp(),
-                                        config.getTreatPrecisionlessNumAsDeci());
+                                        config.getTreatPrecisionlessNumAsDeci(),
+        config.getTreatAsTimestampForLocalTimeZone());
   }
 
   @Override

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

@@ -42,13 +42,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, 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, @Nullable Boolean treatAsOldTimestamp,
-                               @Nullable Boolean treatPrecisionlessNumAsDeci) {
+                               @Nullable Boolean treatPrecisionlessNumAsDeci,
+      @Nullable Boolean treatAsTimestampForLocalTimeZone) {
 
     this.host = host;
     this.port = port;
@@ -62,6 +63,7 @@ public OracleConnectorConfig(String host, int port, String user, String password
     this.useSSL = useSSL;
     this.treatAsOldTimestamp = treatAsOldTimestamp;
     this.treatPrecisionlessNumAsDeci = treatPrecisionlessNumAsDeci;
+    this.treatAsTimestampForLocalTimeZone = treatAsTimestampForLocalTimeZone;
   }
 
   @Override
@@ -98,6 +100,11 @@ public String getConnectionString() {
   @Nullable
   public Boolean treatPrecisionlessNumAsDeci;
 
+  @Name(OracleConstants.TREAT_AS_TIMESTAMP_FOR_LOCAL_TIME_ZONE)
+  @Description("A hidden field to handle mapping of Oracle Timestamp_LTZ data type to BQ Timestamp.")
+  @Nullable
+  public Boolean treatAsTimestampForLocalTimeZone;
+
   @Override
   protected int getDefaultPort() {
     return 1521;
@@ -128,6 +135,10 @@ public Boolean getTreatPrecisionlessNumAsDeci() {
     return Boolean.TRUE.equals(treatPrecisionlessNumAsDeci);
   }
 
+  public Boolean getTreatAsTimestampForLocalTimeZone() {
+    return Boolean.TRUE.equals(treatAsTimestampForLocalTimeZone);
+  }
+
   @Override
   public Properties getConnectionArgumentsProperties() {
     Properties prop = super.getConnectionArgumentsProperties();

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

@@ -46,6 +46,8 @@ private OracleConstants() {
   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";
+  public static final String  TREAT_AS_TIMESTAMP_FOR_LOCAL_TIME_ZONE
+      = "treatAsTimestampForLocalTimeZone";
 
   /**
    * 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

@@ -68,8 +68,9 @@ protected SchemaReader getSchemaReader() {
     // handle schema to make it backward compatible.
     boolean treatAsOldTimestamp = oracleSourceConfig.getConnection().getTreatAsOldTimestamp();
     boolean treatPrecisionlessNumAsDeci = oracleSourceConfig.getConnection().getTreatPrecisionlessNumAsDeci();
+    boolean treatAsTimestampForLocalTimeZone = oracleSourceConfig.getConnection().getTreatAsTimestampForLocalTimeZone();
 
-    return new OracleSourceSchemaReader(null, treatAsOldTimestamp, treatPrecisionlessNumAsDeci);
+    return new OracleSourceSchemaReader(null, treatAsOldTimestamp, treatPrecisionlessNumAsDeci, treatAsTimestampForLocalTimeZone);
   }
 
   @Override
@@ -137,10 +138,10 @@ public OracleSourceConfig(String host, int port, String user, String password, S
                               int defaultBatchValue, int defaultRowPrefetch,
                               String importQuery, Integer numSplits, int fetchSize,
                               String boundingQuery, String splitBy, Boolean useSSL, Boolean treatAsOldTimestamp,
-                              Boolean treatPrecisionlessNumAsDeci) {
+                              Boolean treatPrecisionlessNumAsDeci, Boolean treatAsTimestampForLocalTimeZone) {
       this.connection = new OracleConnectorConfig(host, port, user, password, jdbcPluginName, connectionArguments,
                                                   connectionType, database, role, useSSL, treatAsOldTimestamp,
-                                                  treatPrecisionlessNumAsDeci);
+                                                  treatPrecisionlessNumAsDeci, treatAsTimestampForLocalTimeZone);
       this.defaultBatchValue = defaultBatchValue;
       this.defaultRowPrefetch = defaultRowPrefetch;
       this.fetchSize = fetchSize;

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

@@ -19,6 +19,7 @@
 import com.google.common.collect.ImmutableSet;
 import io.cdap.cdap.api.data.schema.Schema;
 import io.cdap.plugin.db.CommonSchemaReader;
+import org.jetbrains.annotations.NotNull;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -68,15 +69,17 @@ public class OracleSourceSchemaReader extends CommonSchemaReader {
   private final String sessionID;
   private final Boolean isTimestampOldBehavior;
   private final Boolean isPrecisionlessNumAsDecimal;
+  private final Boolean isLtzFieldTimestamp;
 
   public OracleSourceSchemaReader() {
-    this(null, false, false);
+    this(null, false, false, false);
   }
   public OracleSourceSchemaReader(@Nullable String sessionID, boolean isTimestampOldBehavior,
-                                  boolean isPrecisionlessNumAsDecimal) {
+                                  boolean isPrecisionlessNumAsDecimal, boolean isLtzFieldTimestamp) {
     this.sessionID = sessionID;
     this.isTimestampOldBehavior = isTimestampOldBehavior;
     this.isPrecisionlessNumAsDecimal = isPrecisionlessNumAsDecimal;
+    this.isLtzFieldTimestamp = isLtzFieldTimestamp;
   }
 
   @Override
@@ -87,8 +90,7 @@ public Schema getSchema(ResultSetMetaData metadata, int index) throws SQLExcepti
       case TIMESTAMP_TZ:
         return isTimestampOldBehavior ? Schema.of(Schema.Type.STRING) : Schema.of(Schema.LogicalType.TIMESTAMP_MICROS);
       case TIMESTAMP_LTZ:
-        return isTimestampOldBehavior ? Schema.of(Schema.LogicalType.TIMESTAMP_MICROS)
-          : Schema.of(Schema.LogicalType.DATETIME);
+        return getTimestampLtzSchema();
       case Types.TIMESTAMP:
         return isTimestampOldBehavior ? super.getSchema(metadata, index) : Schema.of(Schema.LogicalType.DATETIME);
       case BINARY_FLOAT:
@@ -139,6 +141,14 @@ public Schema getSchema(ResultSetMetaData metadata, int index) throws SQLExcepti
     }
   }
 
+  private @NotNull Schema getTimestampLtzSchema() {
+    if (Boolean.TRUE.equals(isLtzFieldTimestamp)) {
+      return Schema.of(Schema.LogicalType.TIMESTAMP_MICROS);
+    }
+    return isTimestampOldBehavior ? Schema.of(Schema.LogicalType.TIMESTAMP_MICROS)
+        : Schema.of(Schema.LogicalType.DATETIME);
+  }
+
   @Override
   public boolean shouldIgnoreColumn(ResultSetMetaData metadata, int index) throws SQLException {
     if (sessionID == null) {

oracle-plugin/widgets/Oracle-batchsource.json

@@ -158,6 +158,25 @@
             ]
           }
         },
+        {
+          "widget-type": "hidden",
+          "label": "Treat as Timestamp for Timestamp_LTZ",
+          "name": "treatAsTimestampForLocalTimeZone",
+          "widget-attributes": {
+            "layout": "inline",
+            "default": "false",
+            "options": [
+              {
+                "id": "true",
+                "label": "true"
+              },
+              {
+                "id": "false",
+                "label": "false"
+              }
+            ]
+          }
+        },
         {
           "name": "connectionType",
           "label": "Connection Type",

oracle-plugin/widgets/Oracle-connector.json

@@ -167,6 +167,25 @@
               }
             ]
           }
+        },
+        {
+          "widget-type": "hidden",
+          "label": "Treat as Timestamp for Timestamp_LTZ",
+          "name": "treatAsTimestampForLocalTimeZone",
+          "widget-attributes": {
+            "layout": "inline",
+            "default": "false",
+            "options": [
+              {
+                "id": "true",
+                "label": "true"
+              },
+              {
+                "id": "false",
+                "label": "false"
+              }
+            ]
+          }
         }
       ]
     },