add logging for dpo

Author: Isha Tarte

bigquery-connector-common/src/main/java/com/google/cloud/bigquery/connector/common/BigQueryClient.java

@@ -354,6 +354,13 @@ public boolean isTablePartitioned(TableId tableId) {
    */
   public Job overwriteDestinationWithTemporaryDynamicPartitons(
       TableId temporaryTableId, TableId destinationTableId) {
+    log.info(
+        "overwriteDestinationWithTemporaryDynamicPartitons called with temporaryTableId: {} and destinationTableId: {}",
+        temporaryTableId,
+        destinationTableId);
+    log.info(
+        "Stack trace for overwriteDestinationWithTemporaryDynamicPartitons:",
+        new RuntimeException("Debug Stack Trace"));
 
     TableDefinition destinationDefinition = getTable(destinationTableId).getDefinition();
     String sqlQuery = null;
@@ -364,19 +371,26 @@ public Job overwriteDestinationWithTemporaryDynamicPartitons(
 
       TimePartitioning timePartitioning = sdt.getTimePartitioning();
       if (timePartitioning != null) {
+        log.info(
+            "Destination table {} is time partitioned: {}", destinationTableId, timePartitioning);
         sqlQuery =
             getQueryForTimePartitionedTable(
                 destinationTableName, temporaryTableName, sdt, timePartitioning);
       } else {
         RangePartitioning rangePartitioning = sdt.getRangePartitioning();
         if (rangePartitioning != null) {
+          log.info(
+              "Destination table {} is range partitioned: {}",
+              destinationTableId,
+              rangePartitioning);
           sqlQuery =
               getQueryForRangePartitionedTable(
                   destinationTableName, temporaryTableName, sdt, rangePartitioning);
         }
       }
 
       if (sqlQuery != null) {
+        log.info("Running dynamic partition overwrite query: {}", sqlQuery);
         QueryJobConfiguration queryConfig =
             jobConfigurationFactory
                 .createQueryJobConfigurationBuilder(sqlQuery, Collections.emptyMap())
@@ -388,6 +402,9 @@ public Job overwriteDestinationWithTemporaryDynamicPartitons(
     }
 
     // no partitioning default to statndard overwrite
+    log.info(
+        "No partition definition found for destination table {}, falling back to standard overwrite",
+        destinationTableId);
     return overwriteDestinationWithTemporary(temporaryTableId, destinationTableId);
   }
 
@@ -402,6 +419,9 @@ public Job overwriteDestinationWithTemporaryDynamicPartitons(
    */
   public Job overwriteDestinationWithTemporary(
       TableId temporaryTableId, TableId destinationTableId) {
+    log.info(
+        "Stack trace for overwriteDestinationWithTemporary:",
+        new RuntimeException("Debug Stack Trace"));
     String queryFormat =
         "MERGE `%s`\n"
             + "USING (SELECT * FROM `%s`)\n"

spark-bigquery-connector-common/src/main/java/com/google/cloud/spark/bigquery/write/BigQueryWriteHelper.java

@@ -133,6 +133,13 @@ public void writeDataFrameToBigQuery() {
       String format = config.getIntermediateFormat().getDataSource();
       data.write().format(format).save(gcsPath.toString());
 
+      logger.info(
+          "Conditions check: writeDisposition={}, PartitionOverwriteMode={}, tableExists={}, isTablePartitioned={}",
+          writeDisposition,
+          config.getPartitionOverwriteModeValue(),
+          bigQueryClient.tableExists(config.getTableId()),
+          bigQueryClient.isTablePartitioned(config.getTableId()));
+
       if (writeDisposition == JobInfo.WriteDisposition.WRITE_TRUNCATE
           && config.getPartitionOverwriteModeValue() == PartitionOverwriteMode.DYNAMIC
           && bigQueryClient.tableExists(config.getTableId())