fix(fs-bq-import-collection): so it respects existing firestore-bigquery-export BigQuery configuration during backfill

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

@@ -6,6 +6,7 @@ import { ChangeTrackerConfig } from ".";
 
 interface TableRequiresUpdateOptions {
   table: Table;
+  metadata: TableMetadata;
   config: ChangeTrackerConfig;
   documentIdColExists: boolean;
   pathParamsColExists: boolean;
@@ -14,13 +15,13 @@ interface TableRequiresUpdateOptions {
 
 export async function tableRequiresUpdate({
   table,
+  metadata,
   config,
   documentIdColExists,
   pathParamsColExists,
   oldDataColExists,
 }: TableRequiresUpdateOptions): Promise<boolean> {
   /* Setup checks */
-  const { metadata } = table;
 
   /** Check clustering */
   const configCluster = JSON.stringify(config.clustering);

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

@@ -341,14 +341,13 @@ export class FirestoreBigQueryEventHistoryTracker
 
       await clustering.updateClustering(metadata);
 
-      const documentIdColExists = fields.find(
+      const documentIdColExists = fields.some(
         (column) => column.name === "document_id"
       );
-      const pathParamsColExists = fields.find(
+      const pathParamsColExists = fields.some(
         (column) => column.name === "path_params"
       );
-
-      const oldDataColExists = fields.find(
+      const oldDataColExists = fields.some(
         (column) => column.name === "old_data"
       );
 
@@ -372,6 +371,7 @@ export class FirestoreBigQueryEventHistoryTracker
       /** Updated table metadata if required */
       const shouldUpdate = await tableRequiresUpdate({
         table,
+        metadata,
         config: this.config,
         documentIdColExists,
         pathParamsColExists,

firestore-bigquery-export/scripts/import/src/config.ts

@@ -1,4 +1,5 @@
-import * as program from "commander";
+//import * as program from "commander";
+import { program } from "commander";
 import filenamify from "filenamify";
 import inquirer from "inquirer";
 
@@ -14,6 +15,12 @@ const PROJECT_ID_MAX_CHARS = 6144;
 export const FIRESTORE_COLLECTION_NAME_MAX_CHARS = 6144;
 const BIGQUERY_RESOURCE_NAME_MAX_CHARS = 1024;
 
+const VALID_VIEW_TYPES = [
+  "view",
+  "materialized_incremental",
+  "materialized_non_incremental",
+];
+
 const validateBatchSize = (value: string) => {
   return parseInt(value, 10) > 0;
 };
@@ -53,6 +60,15 @@ const validateLocation = (value: string) => {
   return index !== -1;
 };
 
+function parseClustering(value?: string): string[] | null {
+  if (value === undefined) return undefined as any;
+  if (value.trim() === "") return [];
+  return value
+    .split(",")
+    .map((field) => field.trim())
+    .filter(Boolean);
+}
+
 export const validateInput = (
   value: string,
   name: string,
@@ -248,6 +264,15 @@ export async function parseConfig(): Promise<CliConfig | CliConfigError> {
       errors.push("Invalid batch size.");
     }
 
+    if (
+      program.viewType !== undefined &&
+      !VALID_VIEW_TYPES.includes(program.viewType)
+    ) {
+      errors.push(
+        "ViewType must be one of: view, materialized_incremental, materialized_non_incremental."
+      );
+    }
+
     if (errors.length !== 0) {
       program.outputHelp();
       return { kind: "ERROR", errors };
@@ -279,6 +304,8 @@ export async function parseConfig(): Promise<CliConfig | CliConfigError> {
       failedBatchOutput: program.failedBatchOutput,
       transformFunctionUrl: program.transformFunctionUrl,
       firestoreInstanceId: program.firestoreInstanceId || "(default)",
+      clustering: parseClustering(program.clustering),
+      viewType: program.viewType || "view",
     };
   }
   const {

firestore-bigquery-export/scripts/import/src/index.ts

@@ -52,6 +52,8 @@ const run = async (): Promise<number> => {
     cursorPositionFile,
     transformFunctionUrl,
     firestoreInstanceId,
+    clustering,
+    viewType,
   } = config;
   if (useEmulator) {
     console.log("Using emulator");
@@ -94,6 +96,11 @@ const run = async (): Promise<number> => {
     bqProjectId: bigQueryProjectId,
     transformFunction: transformFunctionUrl,
     firestoreInstanceId: firestoreInstanceId,
+    clustering,
+    useMaterializedView:
+      viewType === "materialized_incremental" ||
+      viewType === "materialized_non_incremental",
+    useIncrementalMaterializedView: viewType === "materialized_incremental",
   });
 
   await initializeDataSink(dataSink, config);

firestore-bigquery-export/scripts/import/src/program.ts

@@ -1,4 +1,5 @@
-import * as program from "commander";
+// import * as program from "commander";
+import { program } from "commander";
 
 const packageJson = require("../package.json");
 
@@ -70,5 +71,14 @@ export const getCLIOptions = () => {
       "--firestore-instance-id <database-id>",
       "The Firestore database instance ID. Use '(default)' for the default database.",
       "(default)"
+    )
+    .option(
+      "--view-type <view-type>",
+      "View type used by the deployed extension: view, materialized_incremental, materialized_non_incremental",
+      "view"
+    )
+    .option(
+      "--clustering <clustering>",
+      "Comma-separated clustering fiels for the raw changelog table, e.g. timestamp or timestamp,document_id"
     );
 };

firestore-bigquery-export/scripts/import/src/types.ts

@@ -18,6 +18,12 @@ export interface CliConfig {
   failedBatchOutput?: string;
   transformFunctionUrl?: string;
   firestoreInstanceId: string;
+
+  clustering?: string[] | null;
+  viewType?:
+    | "view"
+    | "materialized_incremental"
+    | "materialized_non_incremental";
 }
 
 export interface CliConfigError {