diff --git a/amazon-redshift-plugin/pom.xml b/amazon-redshift-plugin/pom.xml index 17a18caa9..9a545ef6b 100644 --- a/amazon-redshift-plugin/pom.xml +++ b/amazon-redshift-plugin/pom.xml @@ -20,7 +20,7 @@ database-plugins-parent io.cdap.plugin - 1.12.3 + 1.12.4-SNAPSHOT Amazon Redshift plugin diff --git a/aurora-mysql-plugin/pom.xml b/aurora-mysql-plugin/pom.xml index 654bde42b..b9a542c3d 100644 --- a/aurora-mysql-plugin/pom.xml +++ b/aurora-mysql-plugin/pom.xml @@ -20,7 +20,7 @@ database-plugins-parent io.cdap.plugin - 1.12.3 + 1.12.4-SNAPSHOT Aurora DB MySQL plugin diff --git a/aurora-postgresql-plugin/pom.xml b/aurora-postgresql-plugin/pom.xml index fd508c8fd..0f31154ca 100644 --- a/aurora-postgresql-plugin/pom.xml +++ b/aurora-postgresql-plugin/pom.xml @@ -20,7 +20,7 @@ database-plugins-parent io.cdap.plugin - 1.12.3 + 1.12.4-SNAPSHOT Aurora DB PostgreSQL plugin diff --git a/cloudsql-mysql-plugin/pom.xml b/cloudsql-mysql-plugin/pom.xml index 012acc177..8061b4ca0 100644 --- a/cloudsql-mysql-plugin/pom.xml +++ b/cloudsql-mysql-plugin/pom.xml @@ -20,7 +20,7 @@ database-plugins-parent io.cdap.plugin - 1.12.3 + 1.12.4-SNAPSHOT CloudSQL MySQL plugin diff --git a/cloudsql-postgresql-plugin/pom.xml b/cloudsql-postgresql-plugin/pom.xml index d219d5f65..f147961e6 100644 --- a/cloudsql-postgresql-plugin/pom.xml +++ b/cloudsql-postgresql-plugin/pom.xml @@ -20,7 +20,7 @@ database-plugins-parent io.cdap.plugin - 1.12.3 + 1.12.4-SNAPSHOT CloudSQL PostgreSQL plugin diff --git a/database-commons/pom.xml b/database-commons/pom.xml index 8ee1e295b..ebd4b8bab 100644 --- a/database-commons/pom.xml +++ b/database-commons/pom.xml @@ -20,7 +20,7 @@ database-plugins-parent io.cdap.plugin - 1.12.3 + 1.12.4-SNAPSHOT Database Commons diff --git a/db2-plugin/pom.xml b/db2-plugin/pom.xml index 920ed89c7..868269a8a 100644 --- a/db2-plugin/pom.xml +++ b/db2-plugin/pom.xml @@ -20,7 +20,7 @@ database-plugins-parent io.cdap.plugin - 1.12.3 + 1.12.4-SNAPSHOT IBM DB2 plugin diff --git a/generic-database-plugin/pom.xml b/generic-database-plugin/pom.xml index 08f979397..39cb543d1 100644 --- a/generic-database-plugin/pom.xml +++ b/generic-database-plugin/pom.xml @@ -20,7 +20,7 @@ database-plugins-parent io.cdap.plugin - 1.12.3 + 1.12.4-SNAPSHOT Generic database plugin diff --git a/generic-db-argument-setter/pom.xml b/generic-db-argument-setter/pom.xml index 70c4414dc..912528d4a 100644 --- a/generic-db-argument-setter/pom.xml +++ b/generic-db-argument-setter/pom.xml @@ -20,7 +20,7 @@ database-plugins-parent io.cdap.plugin - 1.12.3 + 1.12.4-SNAPSHOT Generic database argument setter plugin diff --git a/mariadb-plugin/pom.xml b/mariadb-plugin/pom.xml index 682cc153f..3f7b9a58b 100644 --- a/mariadb-plugin/pom.xml +++ b/mariadb-plugin/pom.xml @@ -20,7 +20,7 @@ database-plugins-parent io.cdap.plugin - 1.12.3 + 1.12.4-SNAPSHOT Maria DB plugin diff --git a/memsql-plugin/pom.xml b/memsql-plugin/pom.xml index f3ae24c38..53e10ed78 100644 --- a/memsql-plugin/pom.xml +++ b/memsql-plugin/pom.xml @@ -20,7 +20,7 @@ database-plugins-parent io.cdap.plugin - 1.12.3 + 1.12.4-SNAPSHOT Memsql plugin diff --git a/mssql-plugin/pom.xml b/mssql-plugin/pom.xml index 376a6bc3f..768e5d4c6 100644 --- a/mssql-plugin/pom.xml +++ b/mssql-plugin/pom.xml @@ -20,7 +20,7 @@ database-plugins-parent io.cdap.plugin - 1.12.3 + 1.12.4-SNAPSHOT Microsoft SQL Server plugin diff --git a/mysql-plugin/pom.xml b/mysql-plugin/pom.xml index d2ea17f41..7c7dd054a 100644 --- a/mysql-plugin/pom.xml +++ b/mysql-plugin/pom.xml @@ -20,7 +20,7 @@ database-plugins-parent io.cdap.plugin - 1.12.3 + 1.12.4-SNAPSHOT Mysql plugin diff --git a/netezza-plugin/pom.xml b/netezza-plugin/pom.xml index f141c7371..824a7d6ec 100644 --- a/netezza-plugin/pom.xml +++ b/netezza-plugin/pom.xml @@ -20,7 +20,7 @@ database-plugins-parent io.cdap.plugin - 1.12.3 + 1.12.4-SNAPSHOT Netezza plugin diff --git a/oracle-plugin/pom.xml b/oracle-plugin/pom.xml index 30e4b5911..988cd424b 100644 --- a/oracle-plugin/pom.xml +++ b/oracle-plugin/pom.xml @@ -20,7 +20,7 @@ database-plugins-parent io.cdap.plugin - 1.12.3 + 1.12.4-SNAPSHOT Oracle plugin diff --git a/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleConnector.java b/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleConnector.java index 16371d5c1..3094e7152 100644 --- a/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleConnector.java +++ b/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleConnector.java @@ -113,7 +113,8 @@ protected DBConnectorPath getDBConnectorPath(String path) { @Override protected SchemaReader getSchemaReader(String sessionID) { return new OracleSourceSchemaReader(sessionID, config.getTreatAsOldTimestamp(), - config.getTreatPrecisionlessNumAsDeci()); + config.getTreatPrecisionlessNumAsDeci(), + config.getTreatTimestampLTZAsTimestamp()); } @Override diff --git a/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleConnectorConfig.java b/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleConnectorConfig.java index cbc1e5ed2..79d14215b 100644 --- a/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleConnectorConfig.java +++ b/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 treatTimestampLTZAsTimestamp) { 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.treatTimestampLTZAsTimestamp = treatTimestampLTZAsTimestamp; } @Override @@ -98,6 +100,11 @@ public String getConnectionString() { @Nullable public Boolean treatPrecisionlessNumAsDeci; + @Name(OracleConstants.TREAT_TIMESTAMP_LTZ_AS_TIMESTAMP) + @Description("A hidden field to handle mapping of Oracle Timestamp_LTZ data type to BQ Timestamp.") + @Nullable + public Boolean treatTimestampLTZAsTimestamp; + @Override protected int getDefaultPort() { return 1521; @@ -128,6 +135,10 @@ public Boolean getTreatPrecisionlessNumAsDeci() { return Boolean.TRUE.equals(treatPrecisionlessNumAsDeci); } + public Boolean getTreatTimestampLTZAsTimestamp() { + return Boolean.TRUE.equals(treatTimestampLTZAsTimestamp); + } + @Override public Properties getConnectionArgumentsProperties() { Properties prop = super.getConnectionArgumentsProperties(); diff --git a/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleConstants.java b/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleConstants.java index cbd411175..31c7620ff 100644 --- a/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleConstants.java +++ b/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleConstants.java @@ -45,6 +45,7 @@ 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_TIMESTAMP_LTZ_AS_TIMESTAMP = "treatTimestampLTZAsTimestamp"; /** * Constructs the Oracle connection string based on the provided connection type, host, port, and database. 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 index 1488a084b..978155fc5 100644 --- a/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSource.java +++ b/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSource.java @@ -67,8 +67,10 @@ protected SchemaReader getSchemaReader() { // handle schema to make it backward compatible. boolean treatAsOldTimestamp = oracleSourceConfig.getConnection().getTreatAsOldTimestamp(); boolean treatPrecisionlessNumAsDeci = oracleSourceConfig.getConnection().getTreatPrecisionlessNumAsDeci(); + boolean treatTimestampLTZAsTimestamp = oracleSourceConfig.getConnection().getTreatTimestampLTZAsTimestamp(); - return new OracleSourceSchemaReader(null, treatAsOldTimestamp, treatPrecisionlessNumAsDeci); + return new OracleSourceSchemaReader(null, treatAsOldTimestamp, treatPrecisionlessNumAsDeci, + treatTimestampLTZAsTimestamp); } @Override @@ -133,10 +135,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 treatTimestampLTZAsTimestamp) { this.connection = new OracleConnectorConfig(host, port, user, password, jdbcPluginName, connectionArguments, connectionType, database, role, useSSL, treatAsOldTimestamp, - treatPrecisionlessNumAsDeci); + treatPrecisionlessNumAsDeci, treatTimestampLTZAsTimestamp); this.defaultBatchValue = defaultBatchValue; this.defaultRowPrefetch = defaultRowPrefetch; this.fetchSize = fetchSize; 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 index dd17d2e84..208b70410 100644 --- a/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSourceSchemaReader.java +++ b/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 isTimestampLtzFieldTimestamp; public OracleSourceSchemaReader() { - this(null, false, false); + this(null, false, false, false); } public OracleSourceSchemaReader(@Nullable String sessionID, boolean isTimestampOldBehavior, - boolean isPrecisionlessNumAsDecimal) { + boolean isPrecisionlessNumAsDecimal, boolean isTimestampLtzFieldTimestamp) { this.sessionID = sessionID; this.isTimestampOldBehavior = isTimestampOldBehavior; this.isPrecisionlessNumAsDecimal = isPrecisionlessNumAsDecimal; + this.isTimestampLtzFieldTimestamp = isTimestampLtzFieldTimestamp; } @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,12 @@ public Schema getSchema(ResultSetMetaData metadata, int index) throws SQLExcepti } } + private @NotNull Schema getTimestampLtzSchema() { + return isTimestampOldBehavior || isTimestampLtzFieldTimestamp + ? Schema.of(Schema.LogicalType.TIMESTAMP_MICROS) + : Schema.of(Schema.LogicalType.DATETIME); + } + @Override public boolean shouldIgnoreColumn(ResultSetMetaData metadata, int index) throws SQLException { if (sessionID == null) { diff --git a/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleSchemaReaderTest.java b/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleSchemaReaderTest.java new file mode 100644 index 000000000..1ff77c533 --- /dev/null +++ b/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleSchemaReaderTest.java @@ -0,0 +1,94 @@ +/* + * Copyright © 2025 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. + */ + +package io.cdap.plugin.oracle; + +import com.google.common.collect.Lists; +import io.cdap.cdap.api.data.schema.Schema; +import org.junit.Assert; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mockito; +import org.mockito.junit.MockitoJUnitRunner; + +import java.sql.ResultSet; +import java.sql.ResultSetMetaData; +import java.sql.SQLException; +import java.util.List; + +public class OracleSchemaReaderTest { + + @Test + public void getSchema_timestampLTZFieldTrue_returnTimestamp() throws SQLException { + OracleSourceSchemaReader schemaReader = new OracleSourceSchemaReader(null, false, false, true); + + ResultSet resultSet = Mockito.mock(ResultSet.class); + ResultSetMetaData metadata = Mockito.mock(ResultSetMetaData.class); + + Mockito.when(resultSet.getMetaData()).thenReturn(metadata); + + Mockito.when(metadata.getColumnCount()).thenReturn(2); + // -101 is for TIMESTAMP_TZ + Mockito.when(metadata.getColumnType(1)).thenReturn(-101); + Mockito.when(metadata.getColumnName(1)).thenReturn("column1"); + + // -102 is for TIMESTAMP_LTZ + Mockito.when(metadata.getColumnType(2)).thenReturn(-102); + Mockito.when(metadata.getColumnName(2)).thenReturn("column2"); + + List expectedSchemaFields = Lists.newArrayList(); + expectedSchemaFields.add(Schema.Field.of("column1", Schema.of(Schema.LogicalType.TIMESTAMP_MICROS))); + expectedSchemaFields.add(Schema.Field.of("column2", Schema.of(Schema.LogicalType.TIMESTAMP_MICROS))); + + List actualSchemaFields = schemaReader.getSchemaFields(resultSet); + + Assert.assertEquals(expectedSchemaFields.get(0).getName(), actualSchemaFields.get(0).getName()); + Assert.assertEquals(expectedSchemaFields.get(0).getSchema(), actualSchemaFields.get(0).getSchema()); + Assert.assertEquals(expectedSchemaFields.get(1).getName(), actualSchemaFields.get(1).getName()); + Assert.assertEquals(expectedSchemaFields.get(1).getSchema(), actualSchemaFields.get(1).getSchema()); + + } + + @Test + public void getSchema_timestampLTZFieldFalse_returnDatetime() throws SQLException { + OracleSourceSchemaReader schemaReader = new OracleSourceSchemaReader(null, false, false, false); + + ResultSet resultSet = Mockito.mock(ResultSet.class); + ResultSetMetaData metadata = Mockito.mock(ResultSetMetaData.class); + + Mockito.when(resultSet.getMetaData()).thenReturn(metadata); + + Mockito.when(metadata.getColumnCount()).thenReturn(2); + // -101 is for TIMESTAMP_TZ + Mockito.when(metadata.getColumnType(1)).thenReturn(-101); + Mockito.when(metadata.getColumnName(1)).thenReturn("column1"); + + // -102 is for TIMESTAMP_LTZ + Mockito.when(metadata.getColumnType(2)).thenReturn(-102); + Mockito.when(metadata.getColumnName(2)).thenReturn("column2"); + + List expectedSchemaFields = Lists.newArrayList(); + expectedSchemaFields.add(Schema.Field.of("column1", Schema.of(Schema.LogicalType.TIMESTAMP_MICROS))); + expectedSchemaFields.add(Schema.Field.of("column2", Schema.of(Schema.LogicalType.DATETIME))); + + List actualSchemaFields = schemaReader.getSchemaFields(resultSet); + + Assert.assertEquals(expectedSchemaFields.get(0).getName(), actualSchemaFields.get(0).getName()); + Assert.assertEquals(expectedSchemaFields.get(0).getSchema(), actualSchemaFields.get(0).getSchema()); + Assert.assertEquals(expectedSchemaFields.get(1).getName(), actualSchemaFields.get(1).getName()); + Assert.assertEquals(expectedSchemaFields.get(1).getSchema(), actualSchemaFields.get(1).getSchema()); + } +} diff --git a/oracle-plugin/widgets/Oracle-batchsource.json b/oracle-plugin/widgets/Oracle-batchsource.json index 404262fb2..ab35f3e8c 100644 --- a/oracle-plugin/widgets/Oracle-batchsource.json +++ b/oracle-plugin/widgets/Oracle-batchsource.json @@ -158,6 +158,25 @@ ] } }, + { + "widget-type": "hidden", + "label": "Treat Timestamp_LTZ as Timestamp", + "name": "treatTimestampLTZAsTimestamp", + "widget-attributes": { + "layout": "inline", + "default": "false", + "options": [ + { + "id": "true", + "label": "true" + }, + { + "id": "false", + "label": "false" + } + ] + } + }, { "name": "connectionType", "label": "Connection Type", diff --git a/oracle-plugin/widgets/Oracle-connector.json b/oracle-plugin/widgets/Oracle-connector.json index 013f3b240..005c3ffbd 100644 --- a/oracle-plugin/widgets/Oracle-connector.json +++ b/oracle-plugin/widgets/Oracle-connector.json @@ -167,6 +167,25 @@ } ] } + }, + { + "widget-type": "hidden", + "label": "Treat Timestamp_LTZ as Timestamp", + "name": "treatTimestampLTZAsTimestamp", + "widget-attributes": { + "layout": "inline", + "default": "false", + "options": [ + { + "id": "true", + "label": "true" + }, + { + "id": "false", + "label": "false" + } + ] + } } ] }, diff --git a/pom.xml b/pom.xml index a85b645e6..f59a221d6 100644 --- a/pom.xml +++ b/pom.xml @@ -20,7 +20,7 @@ io.cdap.plugin database-plugins-parent - 1.12.3 + 1.12.4-SNAPSHOT pom Database Plugins Collection of database plugins diff --git a/postgresql-plugin/pom.xml b/postgresql-plugin/pom.xml index 2eeb641bf..73912e656 100644 --- a/postgresql-plugin/pom.xml +++ b/postgresql-plugin/pom.xml @@ -20,7 +20,7 @@ database-plugins-parent io.cdap.plugin - 1.12.3 + 1.12.4-SNAPSHOT PostgreSQL plugin diff --git a/saphana-plugin/pom.xml b/saphana-plugin/pom.xml index c40736e07..6e541ddf2 100644 --- a/saphana-plugin/pom.xml +++ b/saphana-plugin/pom.xml @@ -20,7 +20,7 @@ database-plugins-parent io.cdap.plugin - 1.12.3 + 1.12.4-SNAPSHOT SAP HANA plugin diff --git a/teradata-plugin/pom.xml b/teradata-plugin/pom.xml index 0201805c5..6fa152ad7 100644 --- a/teradata-plugin/pom.xml +++ b/teradata-plugin/pom.xml @@ -21,7 +21,7 @@ database-plugins-parent io.cdap.plugin - 1.12.3 + 1.12.4-SNAPSHOT teradata-plugin