e2e BQ test scenarios Validation step for Existing Table

Author: priyabhatnagar25

src/e2e-test/features/bigquery/source/BigQueryToBigQuery.feature

@@ -301,6 +301,7 @@ Feature: BigQuery source - Verification of BigQuery to BigQuery successful data
     Then Wait till pipeline is in running state
     Then Open and capture logs
     Then Verify the pipeline status is "Succeeded"
+    Then Validate the data transferred from BQ to BQ with actual And expected file for: "bqexpectedfile"
 
   @BQ_EXISTING_SOURCE_TEST @BQ_SINK_TEST
   Scenario: Validate successful record transfer from BigQuery source(existing table) without clicking on the validate button of BigQuery source to BigQuery sink(new table)

src/e2e-test/java/io/cdap/plugin/bigquery/stepsdesign/BQValidationExistingTables.java

@@ -0,0 +1,119 @@
+package io.cdap.plugin.bigquery.stepsdesign;
+
+import com.esotericsoftware.minlog.Log;
+import com.google.cloud.bigquery.FieldValueList;
+import com.google.cloud.bigquery.TableResult;
+import com.google.gson.Gson;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import io.cdap.e2e.utils.BigQueryClient;
+import io.cdap.e2e.utils.PluginPropertyUtils;
+import io.cucumber.core.logging.Logger;
+import io.cucumber.core.logging.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.FileReader;
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * BigQuery Plugin Existing Table validation.
+ */
+public class BQValidationExistingTables {
+
+  private static final Logger LOG = LoggerFactory.getLogger(BQValidationExistingTables.class);
+  static Gson gson = new Gson();
+  public static boolean validateActualDataToExpectedData(String table, String fileName) throws IOException,
+    InterruptedException, URISyntaxException {
+    Map<String, JsonObject> bigQueryMap = new HashMap<>();
+    Map<String, JsonObject> fileMap = new HashMap<>();
+    Path importbqexpectedfile = Paths.get(BQValidationExistingTables.class.getResource("/" + fileName).toURI());
+
+    getBigQueryTableData(table, bigQueryMap);
+    getFileData(importbqexpectedfile.toString(), fileMap);
+
+    boolean isMatched = matchJsonMaps(bigQueryMap, fileMap);
+
+    return isMatched;
+  }
+
+  public static void getFileData(String fileName, Map<String, JsonObject> fileMap) {
+    try (BufferedReader br = new BufferedReader(new FileReader(fileName))) {
+      String line;
+      while ((line = br.readLine()) != null) {
+        JsonObject json = gson.fromJson(line, JsonObject.class);
+        String idKey = getIdKey(json);
+        if (idKey != null) {
+          JsonElement idElement = json.get(idKey);
+          if (idElement.isJsonPrimitive()) {
+            String idValue = idElement.getAsString();
+            fileMap.put(idValue, json);
+          }
+        } else {
+          Log.error("ID key not found");
+        }
+      }
+    } catch (IOException e) {
+      System.err.println("Error reading the file: " + e.getMessage());
+    }
+  }
+
+  private static void getBigQueryTableData(String targetTable, Map<String, JsonObject> bigQueryMap)
+    throws IOException, InterruptedException {
+    String dataset = PluginPropertyUtils.pluginProp("dataset");
+    String projectId = PluginPropertyUtils.pluginProp("projectId");
+    String selectQuery = "SELECT TO_JSON(t) FROM `" + projectId + "." + dataset + "." + targetTable + "` AS t";
+    TableResult result = BigQueryClient.getQueryResult(selectQuery);
+
+    for (FieldValueList row : result.iterateAll()) {
+      JsonObject json = gson.fromJson(row.get(0).getStringValue(), JsonObject.class);
+      String idKey = getIdKey(json); // Get the actual ID key from the JSON object
+      if (idKey != null) {
+        JsonElement idElement = json.get(idKey);
+        if (idElement.isJsonPrimitive()) {
+          String id = idElement.getAsString();
+          bigQueryMap.put(id, json);
+        } else {
+          Log.error("Data Mismatched");
+        }
+      }
+    }
+  }
+
+  /**
+   * Retrieves the key for the ID element in the provided JSON object.
+   *
+   * @param json The JSON object to search for the ID key.
+   */
+  private static String getIdKey(JsonObject json) {
+    if (json.has("ID")) {
+      return "ID";
+    } else if (json.has("Name")) {
+      return "Name";
+    } else if (json.has("Price")) {
+      return "Price";
+    } else if (json.has("Customer_Exists")) {
+      return "Customer_Exists";
+    } else {
+      return null;
+    }
+  }
+
+  private static boolean matchJsonMaps(Map<String, JsonObject> map1, Map<String, JsonObject> map2) {
+    if (!map1.keySet().equals(map2.keySet())) {
+      return false;
+    }
+    for (String key : map1.keySet()) {
+      JsonObject json1 = map1.get(key);
+      JsonObject json2 = map2.get(key);
+      if (!json1.equals(json2)) {
+        return false;
+      }
+    }
+    return true;
+  }
+}

src/e2e-test/java/io/cdap/plugin/bigquery/stepsdesign/BigQuerySource.java

@@ -32,6 +32,7 @@
 import org.junit.Assert;
 
 import java.io.IOException;
+import java.net.URISyntaxException;
 import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.UUID;
@@ -160,4 +161,13 @@ public void enterTheBigQuerySourcePropertyForViewMaterializationDataset(String v
   public void enterBigQuerySourcePropertyTableNameAsView() {
     CdfBigQueryPropertiesActions.enterBigQueryTable(TestSetupHooks.bqSourceView);
   }
+
+  @Then("Validate the data transferred from BQ to BQ with actual And expected file for: {string}")
+  public void validateTheDataFromBQToBQWithActualAndExpectedFileFor(String expectedFile) throws IOException,
+    InterruptedException, URISyntaxException {
+    boolean recordsMatched = BQValidationExistingTables.validateActualDataToExpectedData(
+      PluginPropertyUtils.pluginProp("bqTargetTable"),
+      PluginPropertyUtils.pluginProp(expectedFile));
+    Assert.assertTrue("Value of records in actual and expected file is equal", recordsMatched);
+  }
 }

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

@@ -925,11 +925,11 @@ public static void createSourceBQUpdateTable() throws IOException, InterruptedEx
   public static void createSourceBQExistingTable() throws IOException, InterruptedException {
     bqSourceTable = "E2E_SOURCE_" + UUID.randomUUID().toString().replaceAll("-" , "_");
     io.cdap.e2e.utils.BigQueryClient.getSoleQueryResult("create table `" + datasetName + "." + bqSourceTable + "` " +
-                                                          "(CustomerID INT64, Name STRING, " + "Price FLOAT64," +
+                                                          "(ID INT64, Name STRING, " + "Price FLOAT64," +
                                                           "Customer_Exists BOOL ) ");
     try {
       io.cdap.e2e.utils.BigQueryClient.getSoleQueryResult("INSERT INTO `" + datasetName + "." + bqSourceTable + "` " +
-                                                            "(CustomerID, Name, Price, Customer_Exists)" +
+                                                            "(ID, Name, Price, Customer_Exists)" +
                                                             "VALUES" + "(1, 'Raja Sharma', 200.0, true)");
 
     } catch (NoSuchElementException e) {
@@ -946,12 +946,12 @@ public static void createSinkBQExistingTable() throws IOException, InterruptedEx
     PluginPropertyUtils.addPluginProp("bqTargetTable", bqTargetTable);
     BeforeActions.scenario.write("BQ Target table name - " + bqTargetTable);
     io.cdap.e2e.utils.BigQueryClient.getSoleQueryResult("create table `" + datasetName + "." + bqTargetTable + "` " +
-                                                          "(CustomerID INT64,Name STRING," +
+                                                          "(ID INT64,Name STRING," +
                                                           "Price FLOAT64, Customer_Exists BOOL ) ");
 
     try {
       io.cdap.e2e.utils.BigQueryClient.getSoleQueryResult("INSERT INTO `" + datasetName + "." + bqTargetTable + "` " +
-                                                            "(CustomerID,  Name, Price, Customer_Exists)" +
+                                                            "(ID,  Name, Price, Customer_Exists)" +
                                                             "VALUES" + "(3, 'Rajan Kumar', 100.0, true)");
 
     } catch (NoSuchElementException e) {

src/e2e-test/resources/pluginParameters.properties

@@ -186,6 +186,7 @@ bqInvalidRefName=invalidRef&^*&&*
 bqDatatypeChange1=[{"key":"Id","value":"long"},{"key":"Value","value":"long"}]
 bqDataTypeTestFileSchema1=[{"key":"Id","value":"long"},{"key":"Value","value":"long"},\
   {"key":"UID","value":"string"}]
+bqexpectedfile=testdata/BigQuery/BQExistingTableFile
 ## BIGQUERY-PLUGIN-PROPERTIES-END
 
 ## PUBSUBSINK-PLUGIN-PROPERTIES-START

src/e2e-test/resources/testdata/BigQuery/BQExistingTableFile

@@ -0,0 +1,2 @@
+{"Customer_Exists":true,"ID":1,"Name":"Raja Sharma","Price":200.0}
+{"Customer_Exists":true,"ID":3,"Name":"Rajan Kumar","Price":100.0}