Added execution metrics for BQ Pushdown jobs

Author: fernst

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

@@ -150,12 +150,14 @@ public void executeJoin() {
     if (queryJob == null) {
       throw new SQLEngineException("BigQuery job not found: " + jobId);
     } else if (queryJob.getStatus().getError() != null) {
+      BigQuerySQLEngineUtils.logJobMetrics(queryJob);
       throw new SQLEngineException(String.format(
         "Error executing BigQuery Job: '%s' in Project '%s', Dataset '%s', Location'%s' : %s",
         jobId, project, bqDataset, location, queryJob.getStatus().getError().toString()));
     }
 
     LOG.info("Created BigQuery table `{}` using Job: {}", bqTable, jobId);
+    BigQuerySQLEngineUtils.logJobMetrics(queryJob);
   }
 
   @Override

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

@@ -124,12 +124,14 @@ public BigQuerySelectDataset execute() {
     if (queryJob == null) {
       throw new SQLEngineException("BigQuery job not found: " + jobId);
     } else if (queryJob.getStatus().getError() != null) {
+      BigQuerySQLEngineUtils.logJobMetrics(queryJob);
       throw new SQLEngineException(String.format(
         "Error executing BigQuery Job: '%s' in Project '%s', Dataset '%s', Location'%s' : %s",
         jobId, project, bqDataset, location, queryJob.getStatus().getError().toString()));
     }
 
     LOG.info("Created BigQuery table `{}` using Job: {}", bqTable, jobId);
+    BigQuerySQLEngineUtils.logJobMetrics(queryJob);
     return this;
   }
 

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

@@ -228,6 +228,7 @@ private SQLWriteResult writeInternal(SQLWriteRequest writeRequest,
 
     // Check for errors
     if (queryJob.getStatus().getError() != null) {
+      BigQuerySQLEngineUtils.logJobMetrics(queryJob);
       LOG.error("Error executing BigQuery Job: '{}' in Project '{}', Dataset '{}': {}",
                 jobId, sqlEngineConfig.getProject(), sqlEngineConfig.getDatasetProject(),
                 queryJob.getStatus().getError().toString());
@@ -241,6 +242,7 @@ private SQLWriteResult writeInternal(SQLWriteRequest writeRequest,
     LOG.info("Executed copy operation for {} records from {}.{}.{} to {}.{}.{}", numRows,
              sourceTableId.getProject(), sourceTableId.getDataset(), sourceTableId.getTable(),
              destinationTableId.getProject(), destinationTableId.getDataset(), destinationTableId.getTable());
+    BigQuerySQLEngineUtils.logJobMetrics(queryJob);
 
     return SQLWriteResult.success(datasetName, numRows);
   }

src/main/java/io/cdap/plugin/gcp/bigquery/sqlengine/util/BigQuerySQLEngineUtils.java

@@ -18,11 +18,14 @@
 
 import com.google.cloud.bigquery.BigQuery;
 import com.google.cloud.bigquery.DatasetId;
+import com.google.cloud.bigquery.Job;
+import com.google.cloud.bigquery.JobStatistics;
 import com.google.cloud.bigquery.StandardTableDefinition;
 import com.google.cloud.bigquery.Table;
 import com.google.cloud.bigquery.TableDefinition;
 import com.google.cloud.bigquery.TableId;
 import com.google.cloud.bigquery.TableInfo;
+import com.google.gson.Gson;
 import io.cdap.cdap.api.data.schema.Schema;
 import io.cdap.cdap.etl.api.engine.sql.SQLEngineException;
 import io.cdap.cdap.etl.api.join.JoinCondition;
@@ -49,6 +52,7 @@
 public class BigQuerySQLEngineUtils {
 
   private static final Logger LOG = LoggerFactory.getLogger(BigQuerySQLEngineUtils.class);
+  private static final Gson GSON = new Gson();
 
   public static final String GCS_PATH_FORMAT = BigQuerySinkUtils.GS_PATH_FORMAT + "/%s";
   public static final String BQ_TABLE_NAME_FORMAT = "%s_%s";
@@ -218,7 +222,7 @@ public static void validateOnExpressionJoinCondition(JoinCondition.OnExpression
 
   /**
    * Validates stages for a Join on Key operation
-   *
+   * <p>
    * TODO: Update logic once BQ SQL engine joins support multiple outer join tables
    *
    * @param joinDefinition     Join Definition to validate
@@ -292,4 +296,65 @@ public static Map<String, String> getJobTags(String operation) {
     labels.put("pushdown_operation", operation);
     return Collections.unmodifiableMap(labels);
   }
+
+  /**
+   * Logs information about a BigQUery Job execution using a specified Logger instance
+   *
+   * @param job BigQuery Job
+   */
+  public static void logJobMetrics(Job job) {
+    // Ensure job has statistics information
+    if (job.getStatistics() == null) {
+      LOG.warn("No statistics were found for BigQuery job {}", job.getJobId());
+    }
+
+    String startTimeStr = getISODateTimeString(job.getStatistics().getStartTime());
+    String endTimeStr = getISODateTimeString(job.getStatistics().getEndTime());
+
+    // Print detailed query statistics if available
+    if (job.getStatistics() instanceof JobStatistics.QueryStatistics) {
+      JobStatistics.QueryStatistics queryStatistics = (JobStatistics.QueryStatistics) job.getStatistics();
+      LOG.info("Metrics for job {}:\n" +
+                 " Start: {} ,\n" +
+                 " End: {} ,\n" +
+                 " Processed Bytes: {} ,\n" +
+                 " Billed Bytes: {} ,\n" +
+                 " Total Slot ms: {}",
+               job.getJobId().getJob(),
+               startTimeStr,
+               endTimeStr,
+               queryStatistics.getTotalBytesProcessed(),
+               queryStatistics.getTotalBytesBilled(),
+               queryStatistics.getTotalSlotMs());
+
+      if (LOG.isTraceEnabled()) {
+        LOG.trace("Additional Metrics for job {}:\n" +
+                   " Query Plan: {} ,\n" +
+                   " Query Timeline: {} \n",
+                 job.getJobId().getJob(),
+                 GSON.toJson(queryStatistics.getQueryPlan()),
+                 GSON.toJson(queryStatistics.getTimeline()));
+      }
+
+      return;
+    }
+
+    // Print basic metrics
+    JobStatistics statistics = job.getStatistics();
+    LOG.info("Metrics for job: {}\n" +
+               " Start: {} ,\n" +
+               " End: {}",
+             job.getJobId().getJob(),
+             startTimeStr,
+             endTimeStr);
+
+  }
+
+  private static String getISODateTimeString(Long epoch) {
+    if (epoch == null) {
+      return "N/A";
+    }
+
+    return Instant.ofEpochMilli(epoch).toString();
+  }
 }