debug(firestore-bigquery-export): add debug logging and add stress test

_emulator/.firebaserc

@@ -1,5 +1,14 @@
 {
   "projects": {
     "default": "demo-test"
-  }
+  },
+  "targets": {},
+  "etags": {
+    "firestore-bigquery-testing": {
+      "extensionInstances": {
+        "fs-bq-0": "0e2dc97812f4a95fc191d493b3b5941b72198b7fbcde4baf5593ab8e5d78f137"
+      }
+    }
+  },
+  "dataconnectEmulatorConfig": {}
 }

_emulator/firebase.json

@@ -1,11 +1,6 @@
 {
   "extensions": {
-    "firestore-send-email": "../firestore-send-email",
-    "delete-user-data": "../delete-user-data",
-    "storage-resize-images": "../storage-resize-images",
-    "firestore-counter": "../firestore-counter",
-    "firestore-bigquery-export": "../firestore-bigquery-export",
-    "firestore-send-email-sendgrid": "../firestore-send-email"
+    "fs-bq-0": "../firestore-bigquery-export"
   },
   "storage": {
     "rules": "storage.rules"

firestore-bigquery-export/firestore-bigquery-change-tracker/src/bigquery/handleFailedTransactions.ts

@@ -1,6 +1,7 @@
 import * as firebase from "firebase-admin";
 import { getFirestore } from "firebase-admin/firestore";
 import { FirestoreBigQueryEventHistoryTrackerConfig } from ".";
+import * as logs from "../logs";
 
 if (!firebase.apps.length) {
   firebase.initializeApp();
@@ -18,6 +19,12 @@ export default async (
   let operationCounter = 0;
   let batchIndex = 0;
 
+  logs.emergencyDebugChangetracker("Starting backup process", {
+    rowsCount: rows?.length,
+    backupTableId: config.backupTableId,
+    errorMessage: e.message,
+  });
+
   rows?.forEach((row) => {
     var ref = db.collection(config.backupTableId).doc(row.insertId);
 
@@ -33,11 +40,25 @@ export default async (
       batchArray.push(db.batch());
       batchIndex++;
       operationCounter = 0;
+      logs.emergencyDebugChangetracker("Created new batch due to limit", {
+        currentBatchIndex: batchIndex,
+      });
     }
   });
 
-  for (let batch of batchArray) {
-    await batch.commit();
+  for (let i = 0; i < batchArray.length; i++) {
+    const batch = batchArray[i];
+    try {
+      await batch.commit();
+      logs.emergencyDebugChangetracker("Committed batch successfully", {
+        batchIndex: i,
+      });
+    } catch (commitError) {
+      logs.emergencyDebugChangetracker("Error committing batch", {
+        batchIndex: i,
+        error: commitError.message,
+      });
+    }
   }
 
   return Promise.resolve();

firestore-bigquery-export/firestore-bigquery-change-tracker/src/bigquery/index.ts

@@ -122,13 +122,18 @@ export class FirestoreBigQueryEventHistoryTracker
       };
     });
 
+    logs.emergencyDebugChangetracker("Record events rows prepared", rows);
     const transformedRows = await this.transformRows(rows);
 
     await this.insertData(transformedRows);
   }
 
   private async transformRows(rows: any[]) {
     if (this.config.transformFunction && this.config.transformFunction !== "") {
+      logs.emergencyDebugChangetracker("Transforming rows via function", {
+        rows,
+        transformFunction: this.config.transformFunction,
+      });
       const response = await fetch(this.config.transformFunction, {
         method: "post",
         body: JSON.stringify({ data: rows }),
@@ -171,6 +176,7 @@ export class FirestoreBigQueryEventHistoryTracker
    */
   private async isRetryableInsertionError(e) {
     let isRetryable = true;
+    logs.emergencyDebugChangetracker("Checking retryable error status", e);
     const expectedErrors = [
       { message: "no such field.", location: documentIdField.name },
       { message: "no such field.", location: documentPathParams.name },
@@ -207,6 +213,10 @@ export class FirestoreBigQueryEventHistoryTracker
   private async _waitForInitialization() {
     const dataset = this.bigqueryDataset();
     const changelogName = this.rawChangeLogTableName();
+    logs.emergencyDebugChangetracker("Waiting for initialization", {
+      dataset,
+      changelogName,
+    });
     return waitForInitialization({ dataset, changelogName });
   }
 
@@ -228,9 +238,19 @@ export class FirestoreBigQueryEventHistoryTracker
       const dataset = this.bigqueryDataset();
       const table = dataset.table(this.rawChangeLogTableName());
 
+      logs.emergencyDebugChangetracker("Inserting data into BigQuery", {
+        table: table.id,
+        dataset: dataset.id,
+        rowsLength: rows.length,
+      });
       logs.dataInserting(rows.length);
       await table.insert(rows, options);
       logs.dataInserted(rows.length);
+      logs.emergencyDebugChangetracker("Data inserted into BigQuery", {
+        table: table.id,
+        dataset: dataset.id,
+        rowsLength: rows.length,
+      });
     } catch (e) {
       if (retry && this.isRetryableInsertionError(e)) {
         retry = false;
@@ -267,6 +287,10 @@ export class FirestoreBigQueryEventHistoryTracker
         await this.initializeDataset();
       } catch (error) {
         const message = parseErrorMessage(error, "initializing dataset");
+        logs.emergencyDebugChangetracker(
+          "Dataset initialization error",
+          message
+        );
         throw new Error(`Error initializing dataset: ${message}`);
       }
 
@@ -277,13 +301,21 @@ export class FirestoreBigQueryEventHistoryTracker
           error,
           "initializing raw change log table"
         );
+        logs.emergencyDebugChangetracker(
+          "Raw change log table initialization error",
+          message
+        );
         throw new Error(`Error initializing raw change log table: ${message}`);
       }
 
       try {
         await this.initializeLatestView();
       } catch (error) {
         const message = parseErrorMessage(error, "initializing latest view");
+        logs.emergencyDebugChangetracker(
+          "Latest view initialization error",
+          message
+        );
         throw new Error(`Error initializing latest view: ${message}`);
       }
       await this._waitForInitialization();
@@ -295,6 +327,10 @@ export class FirestoreBigQueryEventHistoryTracker
         "initializing BigQuery resources"
       );
       console.error("Error initializing BigQuery resources: ", message);
+      logs.emergencyDebugChangetracker(
+        "BigQuery resource initialization error",
+        message
+      );
       throw error;
     }
   }

firestore-bigquery-export/firestore-bigquery-change-tracker/src/bigquery/partitioning.ts

@@ -23,11 +23,16 @@ export class Partitioning {
     this.config = config;
     this.table = table;
     this.schema = schema;
+
+    logs.emergencyDebugChangetracker("Initialized Partitioning class", {
+      config,
+      tableId: table?.id,
+      schema,
+    });
   }
 
   private isPartitioningEnabled(): boolean {
     const { timePartitioning } = this.config;
-
     return !!timePartitioning;
   }
 
@@ -42,6 +47,9 @@ export class Partitioning {
       [metadata] = await this.table.getMetadata();
     } catch {
       console.log("No metadata found");
+      logs.emergencyDebugChangetracker("No metadata found for table", {
+        tableId: this.table.id,
+      });
       return null;
     }
 
@@ -69,6 +77,13 @@ export class Partitioning {
       this.config.timePartitioningFieldType === "DATE"
     ) {
       logs.hourAndDatePartitioningWarning();
+      logs.emergencyDebugChangetracker(
+        "Invalid combination of partitioning and field type",
+        {
+          timePartitioning: this.config.timePartitioning,
+          timePartitioningFieldType: this.config.timePartitioningFieldType,
+        }
+      );
       return true;
     }
 
@@ -97,6 +112,7 @@ export class Partitioning {
       timePartitioningField === "timestamp" &&
       !timePartitioningFieldType &&
       !timePartitioningFirestoreField;
+
     return (
       hasOnlyTimestamp ||
       (!!timePartitioningField &&
@@ -131,6 +147,9 @@ export class Partitioning {
     if (!this.table) {
       logs.invalidTableReference();
     }
+    logs.emergencyDebugChangetracker("Table reference validity check", {
+      tableExists: !!this.table,
+    });
     return !!this.table;
   }
 
@@ -143,6 +162,9 @@ export class Partitioning {
     const [metadata] = await this.table.getMetadata();
     if (metadata.timePartitioning) {
       logs.cannotPartitionExistingTable(this.table);
+      logs.emergencyDebugChangetracker("Table already partitioned", {
+        tableId: this.table.id,
+      });
       return true;
     }
 
@@ -297,6 +319,9 @@ export class Partitioning {
 
     if (!proceed) {
       functions.logger.warn(`Did not add partitioning to schema: ${message}`);
+      logs.emergencyDebugChangetracker("Skipping partitioning addition", {
+        reason: message,
+      });
       return;
     }
     // Add new partitioning field

firestore-bigquery-export/firestore-bigquery-change-tracker/src/logs.ts

@@ -271,3 +271,10 @@ export const updatingMetadata = (tableName, resources) => {
     `Updated Metadata on ${tableName}, ${JSON.stringify(resources)})`
   );
 };
+
+export const emergencyDebugChangetracker = (message: string, err?: any) => {
+  logger.warn(
+    `EMERGENCY DEBUG CHANGETRACKER: ${message}`,
+    JSON.stringify(err, null, 2)
+  );
+};