adding metrics support

Author: sahusanket

pom.xml

@@ -72,7 +72,7 @@
     <avro.version>1.8.2</avro.version>
     <bigquery.connector.hadoop2.version>hadoop2-1.0.0</bigquery.connector.hadoop2.version>
     <commons.codec.version>1.4</commons.codec.version>
-    <cdap.version>6.7.0</cdap.version>
+    <cdap.version>6.8.0-SNAPSHOT</cdap.version>
     <cdap.plugin.version>2.10.0-SNAPSHOT</cdap.plugin.version>
     <dropwizard.metrics-core.version>3.2.6</dropwizard.metrics-core.version>
     <flogger.system.backend.version>0.3.1</flogger.system.backend.version>

src/main/java/io/cdap/plugin/gcp/bigquery/sqlengine/BigQueryReadDataset.java

@@ -47,6 +47,7 @@
 import com.google.common.reflect.TypeToken;
 import com.google.gson.Gson;
 import io.cdap.cdap.api.data.schema.Schema;
+import io.cdap.cdap.api.metrics.Metrics;
 import io.cdap.cdap.etl.api.engine.sql.dataset.SQLDataset;
 import io.cdap.cdap.etl.api.engine.sql.request.SQLReadRequest;
 import io.cdap.cdap.etl.api.engine.sql.request.SQLReadResult;
@@ -85,26 +86,30 @@ public class BigQueryReadDataset implements SQLDataset, BigQuerySQLDataset {
   private final String jobId;
   private Schema schema;
   private Long numRows;
+  private Metrics metrics;
 
   private BigQueryReadDataset(String datasetName,
                               BigQuerySQLEngineConfig sqlEngineConfig,
                               BigQuery bigQuery,
                               SQLReadRequest readRequest,
                               TableId destinationTableId,
-                              String jobId) {
+                              String jobId,
+                              Metrics metrics) {
     this.datasetName = datasetName;
     this.sqlEngineConfig = sqlEngineConfig;
     this.bigQuery = bigQuery;
     this.readRequest = readRequest;
     this.destinationTableId = destinationTableId;
     this.jobId = jobId;
+    this.metrics = metrics;
   }
 
   public static BigQueryReadDataset getInstance(String datasetName,
                                                 BigQuerySQLEngineConfig sqlEngineConfig,
                                                 BigQuery bigQuery,
                                                 SQLReadRequest readRequest,
-                                                TableId destinationTableId) {
+                                                TableId destinationTableId,
+                                                Metrics metrics) {
     // Get new Job ID for this push operation
     String jobId = BigQuerySQLEngineUtils.newIdentifier();
 
@@ -113,7 +118,8 @@ public static BigQueryReadDataset getInstance(String datasetName,
                                    bigQuery,
                                    readRequest,
                                    destinationTableId,
-                                   jobId);
+                                   jobId,
+                                   metrics);
   }
 
   public SQLReadResult read() {
@@ -216,7 +222,7 @@ private SQLReadResult readInternal(SQLReadRequest readRequest,
 
     // Check for errors
     if (queryJob.getStatus().getError() != null) {
-      BigQuerySQLEngineUtils.logJobMetrics(queryJob);
+      BigQuerySQLEngineUtils.logJobMetrics(queryJob, metrics);
       LOG.error("Error executing BigQuery Job: '{}' in Project '{}', Dataset '{}': {}",
                 jobId, sqlEngineConfig.getProject(), sqlEngineConfig.getDatasetProject(),
                 queryJob.getStatus().getError().toString());
@@ -230,7 +236,7 @@ private SQLReadResult readInternal(SQLReadRequest readRequest,
     LOG.info("Executed read operation for {} records from {}.{}.{} into {}.{}.{}", numRows,
              sourceTableId.getProject(), sourceTableId.getDataset(), sourceTableId.getTable(),
              sourceTableId.getProject(), sourceTableId.getDataset(), sourceTableId.getTable());
-    BigQuerySQLEngineUtils.logJobMetrics(queryJob);
+    BigQuerySQLEngineUtils.logJobMetrics(queryJob, metrics);
 
     return SQLReadResult.success(datasetName, this);
   }

src/main/java/io/cdap/plugin/gcp/bigquery/sqlengine/BigQuerySQLEngine.java

@@ -350,9 +350,9 @@ public SQLReadResult read(SQLReadRequest readRequest) throws SQLEngineException
 
     // TODO: implement direct source read toggle
     // Check if direct sink write is enabled. If not, skip.
-    if (!sqlEngineConfig.shouldUseDirectSinkWrite()) {
-      return SQLReadResult.unsupported(datasetName);
-    }
+//    if (!sqlEngineConfig.shouldUseDirectSinkWrite()) {
+//      return SQLReadResult.unsupported(datasetName);
+//    }
 
     // Check if this output matches the expected engine. If it doesn't, skip execution for this write operation.;
     if (!BigQuerySQLEngine.class.getName().equals(readRequest.getInput().getSqlEngineClassName())) {
@@ -372,7 +372,8 @@ public SQLReadResult read(SQLReadRequest readRequest) throws SQLEngineException
                                                                       sqlEngineConfig,
                                                                       bigQuery,
                                                                       readRequest,
-                                                                      destinationTableId);
+                                                                      destinationTableId,
+                                                                      metrics);
     SQLReadResult result = readDataset.read();
 
     if (result.isSuccessful()) {