Skip to content

Commit c94601b

Browse files
vikasrathee-csvikasrathee-cs
committed
Added fix for date datatype
Added fix for date datatype (cherry picked from commit ca328a2)
1 parent 1f731da commit c94601b

File tree

8 files changed

+142
-9
lines changed

8 files changed

+142
-9
lines changed

‎oracle-plugin/src/e2e-test/features/sink/OracleRunTime.feature

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -167,3 +167,54 @@ Feature: Oracle - Verify data transfer from BigQuery source to Oracle sink
167167
Then Verify the pipeline status is "Succeeded"
168168
Then Validate records transferred to target table with record counts of BigQuery table
169169
Then Validate the values of records transferred to target Oracle table is equal to the values from source BigQuery table with case
170+
171+
172+
@BQ_SOURCE_TEST_DATE @ORACLE_DATE_TABLE
173+
Scenario: To verify data is getting transferred from BigQuery source to Oracle sink successfully when schema is having date and timestamp fields
174+
Given Open Datafusion Project to configure pipeline
175+
When Expand Plugin group in the LHS plugins list: "Source"
176+
When Select plugin: "BigQuery" from the plugins list as: "Source"
177+
When Expand Plugin group in the LHS plugins list: "Sink"
178+
When Select plugin: "Oracle" from the plugins list as: "Sink"
179+
Then Connect plugins: "BigQuery" and "Oracle" to establish connection
180+
Then Navigate to the properties page of plugin: "BigQuery"
181+
Then Replace input plugin property: "project" with value: "projectId"
182+
Then Enter input plugin property: "datasetProject" with value: "projectId"
183+
Then Enter input plugin property: "referenceName" with value: "BQReferenceName"
184+
Then Enter input plugin property: "dataset" with value: "dataset"
185+
Then Enter input plugin property: "table" with value: "bqSourceTable"
186+
Then Click on the Get Schema button
187+
Then Verify the Output Schema matches the Expected Schema: "outputDatatypesDateTimeSchema"
188+
Then Validate "BigQuery" plugin properties
189+
Then Close the Plugin Properties page
190+
Then Navigate to the properties page of plugin: "Oracle"
191+
Then Select dropdown plugin property: "select-jdbcPluginName" with option value: "driverName"
192+
Then Replace input plugin property: "host" with value: "host" for Credentials and Authorization related fields
193+
Then Replace input plugin property: "port" with value: "port" for Credentials and Authorization related fields
194+
Then Replace input plugin property: "user" with value: "username" for Credentials and Authorization related fields
195+
Then Replace input plugin property: "password" with value: "password" for Credentials and Authorization related fields
196+
Then Select radio button plugin property: "connectionType" with value: "service"
197+
Then Select radio button plugin property: "role" with value: "normal"
198+
Then Enter input plugin property: "referenceName" with value: "sourceRef"
199+
Then Replace input plugin property: "database" with value: "databaseName"
200+
Then Replace input plugin property: "tableName" with value: "targetTable"
201+
Then Replace input plugin property: "dbSchemaName" with value: "schema"
202+
Then Replace input plugin property: "user" with value: "username" for Credentials and Authorization related fields
203+
Then Replace input plugin property: "password" with value: "password" for Credentials and Authorization related fields
204+
Then Enter input plugin property: "referenceName" with value: "targetRef"
205+
Then Select radio button plugin property: "connectionType" with value: "service"
206+
Then Select radio button plugin property: "role" with value: "normal"
207+
Then Validate "Oracle" plugin properties
208+
Then Close the Plugin Properties page
209+
Then Save the pipeline
210+
Then Preview and run the pipeline
211+
Then Verify the preview of pipeline is "success"
212+
Then Click on preview data for Oracle sink
213+
Then Close the preview data
214+
Then Deploy the pipeline
215+
Then Run the Pipeline in Runtime
216+
Then Wait till pipeline is in running state
217+
Then Open and capture logs
218+
Then Verify the pipeline status is "Succeeded"
219+
Then Validate records transferred to target table with record counts of BigQuery table
220+
Then Validate the values of records transferred to target Oracle table is equal to the values from source BigQuery table

‎oracle-plugin/src/e2e-test/java/io.cdap.plugin/BQValidation.java

Lines changed: 41 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,12 @@
3333
import java.sql.Types;
3434
import java.text.ParseException;
3535
import java.text.SimpleDateFormat;
36+
import java.time.LocalDateTime;
37+
import java.time.ZonedDateTime;
38+
import java.time.format.DateTimeFormatter;
39+
import java.time.format.DateTimeParseException;
3640
import java.util.ArrayList;
41+
import java.util.Arrays;
3742
import java.util.Base64;
3843
import java.util.Date;
3944
import java.util.List;
@@ -44,6 +49,13 @@
4449

4550
public class BQValidation {
4651

52+
private static final List<SimpleDateFormat> TIMESTAMP_DATE_FORMATS = Arrays.asList(
53+
new SimpleDateFormat("yyyy-MM-dd'T'hh:mm:ss"),
54+
new SimpleDateFormat("yyyy-MM-dd"));
55+
private static final List<DateTimeFormatter> TIMESTAMP_TZ_DATE_FORMATS = Arrays.asList(
56+
DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ssXXX"),
57+
DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSSSSXXX"));
58+
4759
/**
4860
* Extracts entire data from source and target tables.
4961
*
@@ -173,21 +185,43 @@ public static boolean compareResultSetAndJsonData(ResultSet rsSource, List<JsonO
173185

174186
case Types.TIMESTAMP:
175187
Timestamp sourceTS = rsSource.getTimestamp(columnName);
176-
SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd'T'hh:mm:ss");
177188
String targetT = bigQueryData.get(jsonObjectIdx).get(columnName).getAsString();
178-
Date dateParsed = dateFormat.parse(targetT);
189+
Date dateParsed = null;
190+
for (SimpleDateFormat dateTimeFormatter : TIMESTAMP_DATE_FORMATS) {
191+
try {
192+
dateParsed = dateTimeFormatter.parse(targetT);
193+
break;
194+
} catch (ParseException exception) {
195+
// do nothing
196+
}
197+
}
179198
Timestamp targetTs = new java.sql.Timestamp(dateParsed.getTime());
180-
result = String.valueOf(sourceTS).equals(String.valueOf(targetTs));
199+
result = sourceTS.equals(targetTs);
181200
Assert.assertTrue("Different values found for column : %s", result);
182201
break;
183202

184203
case OracleSourceSchemaReader.TIMESTAMP_TZ:
185204
Timestamp sourceTZ = rsSource.getTimestamp(columnName);
186-
SimpleDateFormat dateValue = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
187205
String targetTS = bigQueryData.get(jsonObjectIdx).get(columnName).getAsString();
188-
Date date = dateValue.parse(targetTS);
189-
Timestamp targetTZ = new Timestamp(date.getTime());
190-
Assert.assertTrue("Different columns found for Timestamp", sourceTZ.equals(targetTZ));
206+
ZonedDateTime targetDate = null;
207+
for (DateTimeFormatter dateTimeFormatter : TIMESTAMP_TZ_DATE_FORMATS) {
208+
try {
209+
targetDate = ZonedDateTime.parse(targetTS, dateTimeFormatter);
210+
break;
211+
} catch (DateTimeParseException exception) {
212+
// do nothing
213+
}
214+
}
215+
Assert.assertTrue("Different columns found for Timestamp",
216+
sourceTZ.toLocalDateTime().equals(targetDate.toLocalDateTime()));
217+
break;
218+
219+
case OracleSourceSchemaReader.TIMESTAMP_LTZ:
220+
Timestamp sourceLTZ = rsSource.getTimestamp(columnName);
221+
String targetLTZ = bigQueryData.get(jsonObjectIdx).get(columnName).getAsString();
222+
DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSSSS");
223+
Assert.assertTrue("Different columns found for Timestamp",
224+
sourceLTZ.toLocalDateTime().equals(LocalDateTime.parse(targetLTZ, formatter)));
191225
break;
192226

193227
case OracleSourceSchemaReader.BINARY_FLOAT:

‎oracle-plugin/src/e2e-test/java/io.cdap.plugin/OracleClient.java

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -350,4 +350,14 @@ public static void deleteTable(String schema, String table)
350350
}
351351
}
352352
}
353+
354+
public static void createTargetDateTable(String targetTable, String schema) throws SQLException,
355+
ClassNotFoundException {
356+
try (Connection connect = getOracleConnection(); Statement statement = connect.createStatement()) {
357+
String createTargetTableQuery = "CREATE TABLE " + schema + "." + targetTable +
358+
"(ID varchar2(100),DATE_COL DATE,TIMESTAMP_TZ_COL TIMESTAMP WITH TIME ZONE,TIMESTAMP_LTZ_COL " +
359+
"TIMESTAMP WITH LOCAL TIME ZONE,INTERVAL_YM_COL INTERVAL YEAR TO MONTH,DATE_TYPE DATE)";
360+
statement.executeUpdate(createTargetTableQuery);
361+
}
362+
}
353363
}

‎oracle-plugin/src/e2e-test/java/io.cdap.plugin/common.stepsdesign/TestSetupHooks.java

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -386,4 +386,34 @@ public static void deleteTempSourceBQTableSmallCase() throws IOException, Interr
386386
BeforeActions.scenario.write("BQ source Table " + bqSourceTable + " deleted successfully");
387387
PluginPropertyUtils.removePluginProp("bqSourceTable");
388388
}
389+
390+
@Before(order = 1, value = "@BQ_SOURCE_TEST_DATE")
391+
public static void createTempSourceBQTableWithDateColumns() throws IOException, InterruptedException {
392+
createSourceBQTableWithQueries(PluginPropertyUtils.pluginProp("CreateBQTableQueryFileDate"),
393+
PluginPropertyUtils.pluginProp("InsertBQDataQueryFileDate"));
394+
}
395+
396+
@After(order = 1, value = "@BQ_SOURCE_TEST_DATE")
397+
public static void deleteTempSourceBQTableWithDateColumns() throws IOException, InterruptedException {
398+
String bqSourceTable = PluginPropertyUtils.pluginProp("bqSourceTable");
399+
BigQueryClient.dropBqQuery(bqSourceTable);
400+
BeforeActions.scenario.write("BQ source Table " + bqSourceTable + " deleted successfully");
401+
PluginPropertyUtils.removePluginProp("bqSourceTable");
402+
}
403+
404+
@Before(order = 2, value = "@ORACLE_DATE_TABLE")
405+
public static void createOracleTargetDateTable() throws SQLException, ClassNotFoundException {
406+
OracleClient.createTargetDateTable(PluginPropertyUtils.pluginProp("targetTable"),
407+
PluginPropertyUtils.pluginProp("schema"));
408+
BeforeActions.scenario.write("Oracle Target Table - " + PluginPropertyUtils.pluginProp("targetTable")
409+
+ " created successfully");
410+
}
411+
412+
@After(order = 2, value = "@ORACLE_DATE_TABLE")
413+
public static void dropOracleTargetDateTable() throws SQLException, ClassNotFoundException {
414+
OracleClient.deleteTable(PluginPropertyUtils.pluginProp("schema"),
415+
PluginPropertyUtils.pluginProp("targetTable"));
416+
BeforeActions.scenario.write("Oracle Target Table - " + PluginPropertyUtils.pluginProp("targetTable")
417+
+ " deleted successfully");
418+
}
389419
}

‎oracle-plugin/src/e2e-test/resources/pluginParameters.properties

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -111,6 +111,9 @@ InsertBQDataQueryFile=testdata/BigQuery/BigQueryInsertDataQuery.txt
111111
#bq queries file path for Small Case Schema
112112
CreateBQTableQueryFileSmallCase=testdata/BigQuery/BigQueryCreateTableQuerySmallCase.txt
113113
InsertBQDataQueryFileSmallCase=testdata/BigQuery/BigQueryInsertDataQuerySmallCase.txt
114+
#bq queries file path for various Date time type Fields
115+
CreateBQTableQueryFileDate=testdata/BigQuery/CreateBQTableQueryFileDate.txt
116+
InsertBQDataQueryFileDate=testdata/BigQuery/InsertBQDataQueryFileDate.txt
114117

115118
#ORACLE Datatypes
116119
bigQueryColumns=(COL23 FLOAT(4), COL28 TIMESTAMP, COL29 TIMESTAMP(9), COL30 TIMESTAMP WITH TIME ZONE, \
@@ -125,3 +128,6 @@ outputDatatypesSchema1=[{"key":"COL23","value":"double"},{"key":"COL28","value":
125128
{"key":"COL29","value":"datetime"},{"key":"COL30","value":"timestamp"},{"key":"COL31","value":"string"},\
126129
{"key":"COL32","value":"string"},{"key":"COL33","value":"datetime"},{"key":"COL34","value":"float"},\
127130
{"key":"COL35","value":"double"}]
131+
outputDatatypesDateTimeSchema=[{"key":"ID","value":"string"},{"key":"DATE_COL","value":"datetime"},\
132+
{"key":"TIMESTAMP_TZ_COL","value":"timestamp"},{"key":"TIMESTAMP_LTZ_COL","value":"datetime"},\
133+
{"key":"INTERVAL_YM_COL","value":"string"},{"key":"DATE_TYPE","value":"date"}]

‎oracle-plugin/src/e2e-test/resources/testdata/BigQuery/CreateBQTableQueryFileDate.txt

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
CREATE TABLE `DATASET.TABLE_NAME` (ID STRING NOT NULL,DATE_COL DATETIME, TIMESTAMP_TZ_COL TIMESTAMP, TIMESTAMP_LTZ_COL DATETIME, INTERVAL_YM_COL STRING,DATE_TYPE DATE);

‎oracle-plugin/src/e2e-test/resources/testdata/BigQuery/InsertBQDataQueryFileDate.txt

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
INSERT INTO `DATASET.TABLE_NAME` (id, date_col, timestamp_tz_col, timestamp_ltz_col, interval_ym_col,date_type) VALUES('2', '2024-10-11', '2024-10-11 14:30:00.123456+00:00', '2024-10-11 14:30:00.123456','2-6','2024-10-11');

‎oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSourceDBRecord.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -160,8 +160,8 @@ protected void writeNonNullToDB(PreparedStatement stmt, Schema fieldSchema,
160160
String timestampString = Timestamp.valueOf(localDateTime).toString();
161161
Object timestampWithTimeZone = createOracleTimestamp(stmt.getConnection(), timestampString);
162162
stmt.setObject(sqlIndex, timestampWithTimeZone);
163-
} else if (Schema.LogicalType.TIMESTAMP_MICROS.equals(fieldSchema.getLogicalType())) {
164-
// Deprecated: Handle the case when the Timestamp is mapped to CDAP Timestamp type
163+
} else {
164+
// Handle the case when the Timestamp is mapped to CDAP Timestamp type or CDAP Date type.
165165
super.writeNonNullToDB(stmt, fieldSchema, fieldName, fieldIndex);
166166
}
167167
} else {

0 commit comments

Comments
 (0)