Skip to content

Commit 5eacc18

Browse files
Ishakaushik-cloudIshakaushik-cloud
committed
bq validation
1 parent 717fa07 commit 5eacc18

7 files changed

Lines changed: 167 additions & 35 deletions

File tree

‎cloudsql-mysql-plugin/src/e2e-test/java/io/cdap/plugin/CloudMySql/BQValidation.java‎

Lines changed: 134 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -3,27 +3,34 @@
33
import com.google.cloud.bigquery.TableResult;
44
import com.google.gson.Gson;
55
import com.google.gson.JsonObject;
6+
import com.google.type.Decimal;
67
import io.cdap.e2e.utils.BigQueryClient;
78
import io.cdap.e2e.utils.PluginPropertyUtils;
89
import io.cdap.plugin.CloudMySqlClient;
10+
import io.cdap.plugin.common.stepsdesign.TestSetupHooks;
911
import org.junit.Assert;
10-
12+
import org.junit.Test;
1113
import java.io.IOException;
1214
import java.sql.*;
15+
import java.sql.Date;
1316
import java.text.ParseException;
1417
import java.text.SimpleDateFormat;
15-
import java.util.ArrayList;
16-
import java.util.Base64;
17-
import java.util.Date;
18-
import java.util.List;
18+
import java.time.LocalDateTime;
19+
import java.time.format.DateTimeFormatter;
20+
import java.util.*;
1921

2022
/**
2123
* BQValidation
2224
*/
2325

2426
public class BQValidation {
25-
public static void main(String[] args) {
26-
// validateBQAndDBRecordValues(String schema, String sourceTable, String targetTable)
27+
public static void main(String[] args) throws SQLException, ParseException, IOException, ClassNotFoundException, InterruptedException {
28+
//TestSetupHooks.createTables();
29+
// TestSetupHooks.createTempSourceBQTable();
30+
// validateBQAndDBRecordValues("SourceTable_qGMpakGOTZ", "mySqlToBQTable");
31+
// validateDBAndBQRecordValues("SourceTable_qGMpakGOTZ", "mySqlToBQTable");
32+
// validateDBAndBQRecordValues("SourceTable_qGMpakGOTZ", "mySqlToBQTable");
33+
validateDBAndBQRecordValues("E2E_SOURCE_4b406","TargetTable_TRFSpppjsZ");
2734
}
2835

2936
/**
@@ -33,7 +40,7 @@ public static void main(String[] args) {
3340
* @return true if the values in source and target side are equal
3441
*/
3542

36-
public static boolean validateBQAndDBRecordValues(String schema, String sourceTable, String targetTable)
43+
public static boolean validateBQAndDBRecordValues(String sourceTable, String targetTable)
3744
throws SQLException, ClassNotFoundException, ParseException, IOException, InterruptedException {
3845
List<JsonObject> jsonResponse = new ArrayList<>();
3946
List<Object> bigQueryRows = new ArrayList<>();
@@ -42,17 +49,44 @@ public static boolean validateBQAndDBRecordValues(String schema, String sourceTa
4249
JsonObject json = new Gson().fromJson(String.valueOf(rows), JsonObject.class);
4350
jsonResponse.add(json);
4451
}
45-
String getSourceQuery = "SELECT * FROM " + schema + "." + sourceTable;
52+
String getSourceQuery = "SELECT * FROM "+ sourceTable;
4653
try (Connection connect = CloudMySqlClient.getCloudMysqlConnection()) {
4754
connect.setHoldability(ResultSet.HOLD_CURSORS_OVER_COMMIT);
4855
Statement statement1 = connect.createStatement(ResultSet.TYPE_SCROLL_SENSITIVE, ResultSet.CONCUR_UPDATABLE,
4956
ResultSet.HOLD_CURSORS_OVER_COMMIT);
50-
5157
ResultSet rsSource = statement1.executeQuery(getSourceQuery);
5258
return compareResultSetData(rsSource, jsonResponse);
5359
}
5460
}
5561

62+
public static boolean validateDBAndBQRecordValues(String sourceTable, String targetTable)
63+
throws SQLException, ClassNotFoundException, ParseException, IOException, InterruptedException {
64+
List<JsonObject> jsonResponse = new ArrayList<>();
65+
List<Object> bigQueryRows = new ArrayList<>();
66+
getBigQueryTableData(sourceTable, bigQueryRows);
67+
for (Object rows : bigQueryRows) {
68+
JsonObject json = new Gson().fromJson(String.valueOf(rows), JsonObject.class);
69+
jsonResponse.add(json);
70+
}
71+
String getTargetQuery = "SELECT * FROM " + targetTable;
72+
// String getSourceTimeZoneQuery = "SELECT current_setting('TIMEZONE') AS timezone";
73+
try (Connection connect = CloudMySqlClient.getCloudMysqlConnection()) {
74+
connect.setHoldability(ResultSet.HOLD_CURSORS_OVER_COMMIT);
75+
Statement statement1 = connect.createStatement(ResultSet.TYPE_SCROLL_SENSITIVE, ResultSet.CONCUR_UPDATABLE,
76+
ResultSet.HOLD_CURSORS_OVER_COMMIT);
77+
// Statement timeZoneStatement = connect.createStatement();
78+
ResultSet rsTarget = statement1.executeQuery(getTargetQuery);
79+
// ResultSet rsTimeZone = timeZoneStatement.executeQuery(getSourceTimeZoneQuery);
80+
// String targettz = getTimeZoneIdFromSource(rsTimeZone);
81+
return compareResultSetData(rsTarget, jsonResponse);
82+
}
83+
}
84+
85+
86+
87+
88+
89+
5690
/**
5791
* Retrieves the data from a specified BigQuery table and populates it into the provided list of objects.
5892
*
@@ -68,7 +102,7 @@ private static void getBigQueryTableData(String table, List<Object> bigQueryRows
68102
String dataset = PluginPropertyUtils.pluginProp("dataset");
69103
String selectQuery = "SELECT TO_JSON(t) FROM `" + projectId + "." + dataset + "." + table + "` AS t";
70104
TableResult result = BigQueryClient.getQueryResult(selectQuery);
71-
result.iterateAll().forEach(value -> bigQueryRows.add(value.get(0).getValue()));
105+
result.iterateAll().forEach(value -> bigQueryRows.add(value.get(0).getValue()));
72106
}
73107

74108
/**
@@ -105,17 +139,68 @@ public static boolean compareResultSetData(ResultSet rsSource, List<JsonObject>
105139
//Variable 'jsonObjectIdx' to track the index of the current JsonObject in the bigQueryData list,
106140
int jsonObjectIdx = 0;
107141
while (rsSource.next()) {
108-
int currentColumnCount = 1;
142+
int currentColumnCount = 2;
109143
while (currentColumnCount <= columnCountSource) {
110144
String columnTypeName = mdSource.getColumnTypeName(currentColumnCount);
111145
int columnType = mdSource.getColumnType(currentColumnCount);
112146
String columnName = mdSource.getColumnName(currentColumnCount);
113147
// Perform different comparisons based on column type
114148
switch (columnType) {
115149
// Since we skip BFILE in Oracle Sink, we are not comparing the BFILE source and sink values
150+
case Types.BIT:
151+
Boolean sourceBit = rsSource.getBoolean(currentColumnCount);
152+
Boolean targetBit = Boolean.parseBoolean(bigQueryData.get(jsonObjectIdx).get(columnName).getAsString());
153+
Assert.assertTrue("Different values found for column : %s",
154+
String.valueOf(sourceBit).equals(String.valueOf(targetBit)));
155+
break;
156+
157+
case Types.SMALLINT:
158+
case Types.INTEGER:
159+
case Types.TINYINT:
160+
Integer sourceTinyInt = rsSource.getInt(currentColumnCount);
161+
Integer targetTinyInt = Integer.parseInt(bigQueryData.get(jsonObjectIdx).get(columnName).getAsString());
162+
Assert.assertTrue("Different values found for column : %s",
163+
String.valueOf(sourceTinyInt).equals(String.valueOf(targetTinyInt)));
164+
break;
165+
166+
case Types.REAL:
167+
Float sourceFloat = rsSource.getFloat(currentColumnCount);
168+
Float targetFloat = Float.parseFloat(bigQueryData.get(jsonObjectIdx).get(columnName).getAsString());
169+
Assert.assertTrue("Different values found for column : %s",
170+
String.valueOf(sourceFloat).equals(String.valueOf(targetFloat)));
171+
break;
172+
173+
case Types.DOUBLE:
174+
Double sourceDouble= rsSource.getDouble(currentColumnCount);
175+
Double targetDouble = Double.parseDouble(bigQueryData.get(jsonObjectIdx).get(columnName).getAsString());
176+
Assert.assertTrue("Different values found for column : %s",
177+
String.valueOf(sourceDouble).equals(String.valueOf(targetDouble)));
178+
break;
179+
180+
case Types.DATE:
181+
Date sourceDate = rsSource.getDate(currentColumnCount);
182+
Date targetDate = java.sql.Date.valueOf(bigQueryData.get(jsonObjectIdx).get(columnName).getAsString());
183+
Assert.assertTrue("Different values found for column : %s",
184+
String.valueOf(sourceDate).equals(String.valueOf(targetDate)));
185+
break;
186+
187+
case Types.TIME:
188+
Time sourceTime= rsSource.getTime(currentColumnCount);
189+
Time targetTime = Time.valueOf(bigQueryData.get(jsonObjectIdx).get(columnName).getAsString());
190+
Assert.assertTrue("Different values found for column : %s",
191+
String.valueOf(sourceTime).equals(String.valueOf(targetTime)));
192+
break;
193+
194+
case Types.DECIMAL:
195+
org.apache.spark.sql.types.Decimal sourceDecimal = org.apache.spark.sql.types.Decimal.fromDecimal(rsSource.getBigDecimal(currentColumnCount));
196+
org.apache.spark.sql.types.Decimal targetDecimal = org.apache.spark.sql.types.Decimal.fromDecimal(bigQueryData.get(jsonObjectIdx).get(columnName).getAsBigDecimal());
197+
Assert.assertEquals("Different values found for column : %s", sourceDecimal, targetDecimal);
198+
break;
199+
116200
case Types.BLOB:
117201
case Types.VARBINARY:
118202
case Types.LONGVARBINARY:
203+
case Types.BINARY:
119204
String sourceB64String = new String(Base64.getEncoder().encode(rsSource.getBytes(currentColumnCount)));
120205
String targetB64String = bigQueryData.get(jsonObjectIdx).get(columnName).getAsString();
121206
Assert.assertEquals("Different values found for column : %s",
@@ -130,18 +215,20 @@ public static boolean compareResultSetData(ResultSet rsSource, List<JsonObject>
130215
break;
131216

132217
case Types.TIMESTAMP:
133-
Timestamp sourceTS = rsSource.getTimestamp(columnName);
134-
SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd'T'hh:mm:ss'Z'");
135-
Date parsedDate = dateFormat.parse(bigQueryData.get(jsonObjectIdx).get(columnName).getAsString());
136-
Timestamp targetTs = new Timestamp(parsedDate.getTime());
137-
Assert.assertEquals("Different values found for column : %s", String.valueOf(sourceTS).
138-
equals(String.valueOf(targetTs)));
218+
String sourceTS= String.valueOf(rsSource.getTimestamp(currentColumnCount));
219+
String targetTS=bigQueryData.get(jsonObjectIdx).get(columnName).getAsString();
220+
LocalDateTime timestamp = LocalDateTime.parse(targetTS, DateTimeFormatter.ISO_DATE_TIME);
221+
DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.S");
222+
String formattedTimestamp = timestamp.format(formatter);
223+
Assert.assertEquals(sourceTS, formattedTimestamp);
139224
break;
225+
140226
default:
141227
String sourceString = rsSource.getString(currentColumnCount);
142228
String targetString = bigQueryData.get(jsonObjectIdx).get(columnName).getAsString();
143229
Assert.assertEquals(String.format("Different %s values found for column : %s", columnTypeName, columnName),
144230
String.valueOf(sourceString), String.valueOf(targetString));
231+
break;
145232
}
146233
currentColumnCount++;
147234
}
@@ -151,4 +238,33 @@ public static boolean compareResultSetData(ResultSet rsSource, List<JsonObject>
151238
rsSource.next());
152239
return true;
153240
}
241+
242+
243+
private static void getTimestampValidation(ResultSet rsSource, String sourceTimeZone, String columnTypeName, String
244+
columnName, int currentColumnCount, JsonObject bigQueryDataObj) throws
245+
SQLException, ParseException {
246+
TimeZone sourceTz;
247+
if (columnTypeName.equals("timestamp")) {
248+
sourceTz = TimeZone.getTimeZone("UTC"); //Assume UTC since plugin assumes UTC
249+
} else if (columnTypeName.equals("timestamptz")) {
250+
sourceTz = TimeZone.getTimeZone(sourceTimeZone); //Get TZ From Postgres
251+
} else
252+
return;
253+
GregorianCalendar gc = new GregorianCalendar(sourceTz);
254+
gc.setGregorianChange(new Date(Long.MIN_VALUE));
255+
Timestamp sourceTS = rsSource.getTimestamp(currentColumnCount, gc);
256+
TimeZone targetTz = TimeZone.getTimeZone(
257+
"UTC"); //BQ Always Stores in UTC, if timezone is available it converts to UTC, if it is not available it assumes it to be UTC
258+
SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
259+
dateFormat.setTimeZone(targetTz);
260+
Date parsedDate = (Date) dateFormat.parse(bigQueryDataObj.get(columnName).getAsString());
261+
Timestamp targetTs = new Timestamp(parsedDate.getTime());
262+
Assert.assertEquals(sourceTS, targetTs);
263+
}
264+
private static String getTimeZoneIdFromSource(ResultSet rsTimezone) throws SQLException {
265+
if (rsTimezone.next()) {
266+
return rsTimezone.getString("timezone");
267+
}
268+
return "UTC"; //If query fails return UTC
269+
}
154270
}

‎cloudsql-mysql-plugin/src/e2e-test/java/io/cdap/plugin/CloudMySqlClient.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,8 @@ public class CloudMySqlClient {
1414
private static final String connectionName = PluginPropertyUtils.pluginProp("ConnectionName");
1515

1616
public static void main(String[] args) throws SQLException, ClassNotFoundException {
17-
getCloudMysqlConnection();
17+
// getCloudMysqlConnection();
18+
1819
//createSourceTable("myTable");
1920
// createSourceTable("newTable");
2021
// String[] tablesToDrop = {"newTable"};
@@ -174,6 +175,7 @@ public static void createTargetCloudMysqlTable(String targetTable) throws SQLExc
174175
String datatypesColumns = PluginPropertyUtils.pluginProp("CloudMySqlDatatypesColumns");
175176
String createTargetTableQuery = "CREATE TABLE " + targetTable + " " + datatypesColumns;
176177
statement.executeUpdate(createTargetTableQuery);
178+
System.out.println(createTargetTableQuery);
177179
}
178180
}
179181

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

Lines changed: 18 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -24,11 +24,17 @@
2424
*/
2525

2626
public class TestSetupHooks {
27-
public static void main(String[] args) throws SQLException, ClassNotFoundException {
28-
setTableName();
29-
createDatatypesTable();
27+
public static void main(String[] args) throws SQLException, ClassNotFoundException, IOException, InterruptedException {
28+
// setTableName();
29+
// createCloudMysqlTestTable();
30+
// createTables();
31+
// createDatatypesTable();
32+
createTempSourceBQTable();
33+
//createCloudMysqlTestTable();
34+
// createTempSourceBQTable();
35+
// createCloudMysqlTestTable();
3036
}
31-
private static void setTableName() {
37+
public static void setTableName() {
3238
String randomString = RandomStringUtils.randomAlphabetic(10);
3339
String sourceTableName = String.format("SourceTable_%s", randomString);
3440
String targetTableName = String.format("TargetTable_%s", randomString);
@@ -73,7 +79,9 @@ public static void dropTables() throws SQLException, ClassNotFoundException {
7379

7480
@Before(order = 2, value = "@CLOUDMYSQL_TEST_TABLE")
7581
public static void createCloudMysqlTestTable() throws SQLException, ClassNotFoundException {
82+
// BeforeActions.scenario.write("SQL Target table name - " + sqlTargetTableName);
7683
CloudMySqlClient.createTargetCloudMysqlTable(PluginPropertyUtils.pluginProp("targetTable"));
84+
7785
}
7886

7987
@Before(order = 1, value = "@BQ_SINK_TEST")
@@ -113,8 +121,8 @@ public static void deleteTempSourceBQTable() throws IOException, InterruptedExce
113121
PluginPropertyUtils.removePluginProp("bqSourceTable");
114122
}
115123
private static void createSourceBQTableWithQueries(String bqCreateTableQueryFile, String bqInsertDataQueryFile)
116-
throws IOException, InterruptedException {
117-
String bqSourceTable = "E2E_SOURCE_" + UUID.randomUUID().toString().replaceAll("-", "_");
124+
throws IOException, InterruptedException,NullPointerException {
125+
String bqSourceTable = "E2E_SOURCE_" + UUID.randomUUID().toString().substring(0,5).replaceAll("-", "_");
118126

119127
String createTableQuery = StringUtils.EMPTY;
120128
try {
@@ -124,6 +132,7 @@ private static void createSourceBQTableWithQueries(String bqCreateTableQueryFile
124132
createTableQuery = createTableQuery.replace("DATASET", PluginPropertyUtils.pluginProp("dataset"))
125133
.replace("TABLE_NAME", bqSourceTable);
126134
} catch (Exception e) {
135+
e.printStackTrace();
127136
BeforeActions.scenario.write("Exception in reading " + bqCreateTableQueryFile + " - " + e.getMessage());
128137
Assert.fail("Exception in BigQuery testdata prerequisite setup " +
129138
"- error in reading create table query file " + e.getMessage());
@@ -140,6 +149,7 @@ private static void createSourceBQTableWithQueries(String bqCreateTableQueryFile
140149
BeforeActions.scenario.write("Exception in reading " + bqInsertDataQueryFile + " - " + e.getMessage());
141150
Assert.fail("Exception in BigQuery testdata prerequisite setup " +
142151
"- error in reading insert data query file " + e.getMessage());
152+
143153
}
144154
BigQueryClient.getSoleQueryResult(createTableQuery);
145155
try {
@@ -148,8 +158,9 @@ private static void createSourceBQTableWithQueries(String bqCreateTableQueryFile
148158
// Insert query does not return any record.
149159
// Iterator on TableResult values in getSoleQueryResult method throws NoSuchElementException
150160
}
161+
System.out.println(bqSourceTable);
151162
PluginPropertyUtils.addPluginProp("bqSourceTable", bqSourceTable);
152-
BeforeActions.scenario.write("BQ Source Table " + bqSourceTable + " created successfully");
163+
// BeforeActions.scenario.write("BQ Source Table " + bqSourceTable + " created successfully");
153164
}
154165

155166
}

‎cloudsql-mysql-plugin/src/e2e-test/resources/BigQuery/BigQueryCreateTableQuery.txt‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,2 +1,2 @@
1-
create table `DATASET.TABLE_NAME` (col1 BYTES, col2 STRING, col3 DATE, col4 FLOAT64, col6 TIMESTAMP,
2-
col8 BOOL, col9 INT64, col10 TIME)
1+
create table `DATASET.TABLE_NAME` (COL1 BYTES, COL2 STRING, COL3 DATE, COL4 FLOAT64, COL6 TIMESTAMP,
2+
COL8 BOOL, COL9 INT64, COL10 TIME)

‎cloudsql-mysql-plugin/src/e2e-test/resources/BigQuery/BigQueryInsertDataQuery.txt‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,3 @@
1-
insert into `DATASET.TABLE_NAME` (col1, col2, col3, col4, col6, col8, col9, col10) values
1+
insert into `DATASET.TABLE_NAME` (COL1, COL2, COL3, COL4, COL6, COL8, COL9, COL10) values
22
(b'01011011','priya','2021-01-28',1.110,'2019-03-10 04:50:01 UTC',false,92233720,'21:26:00'),
33
(b'01011011','surya','2021-01-21',1.110000001,'2018-03-10 04:50:01 UTC',true,92233729,'20:26:00');

0 commit comments

Comments
 (0)