feat: custom sync catalog search and expansion (#13924)

Co-authored-by: Vladimir <[email protected]>

Author: Joey Marshment-Howell

airbyte-webapp/cypress/commands/api/payloads.ts

@@ -67,6 +67,13 @@ export const getPostgresCreateDestinationBody = (name: string): DestinationCreat
   },
 });
 
+export const getFakerCreateSourceBody = (sourceName: string): SourceCreate => ({
+  name: sourceName,
+  workspaceId: getWorkspaceId(),
+  sourceDefinitionId: ConnectorIds.Sources.Faker,
+  connectionConfiguration: {},
+});
+
 export const getPokeApiCreateSourceBody = (sourceName: string, pokeName: string): SourceCreate => ({
   name: sourceName,
   workspaceId: getWorkspaceId(),

airbyte-webapp/cypress/commands/connection.ts

@@ -17,6 +17,7 @@ import { openCreateConnection } from "pages/destinationPage";
 
 import {
   getConnectionCreateRequest,
+  getFakerCreateSourceBody,
   getLocalJSONCreateDestinationBody,
   getPokeApiCreateSourceBody,
   getPostgresCreateDestinationBody,
@@ -90,6 +91,17 @@ export const startManualReset = () => {
   cy.get("[data-id='clear-data']").click();
 };
 
+export const createFakerSourceViaApi = () => {
+  let source: SourceRead;
+  return requestWorkspaceId().then(() => {
+    const sourceRequestBody = getFakerCreateSourceBody(appendRandomString("Faker Source"));
+    requestCreateSource(sourceRequestBody).then((sourceResponse) => {
+      source = sourceResponse;
+    });
+    return source;
+  });
+};
+
 export const createPokeApiSourceViaApi = () => {
   let source: SourceRead;
   return requestWorkspaceId().then(() => {

airbyte-webapp/cypress/e2e/connection/syncCatalog.cy.ts

@@ -0,0 +1,58 @@
+import {
+  createJsonDestinationViaApi,
+  createNewConnectionViaApi,
+  createFakerSourceViaApi,
+} from "@cy/commands/connection";
+import { visit } from "@cy/pages/connection/connectionPageObject";
+import { setFeatureFlags } from "@cy/support/e2e";
+import { DestinationRead, SourceRead, WebBackendConnectionRead } from "@src/core/api/types/AirbyteClient";
+
+const CATALOG_SEARCH_INPUT = '[data-testid="sync-catalog-search"]';
+
+describe("Sync catalog", () => {
+  let fakerSource: SourceRead;
+  let jsonDestination: DestinationRead;
+  let connection: WebBackendConnectionRead;
+
+  before(() => {
+    setFeatureFlags({ "connection.syncCatalogV2": true });
+
+    createFakerSourceViaApi()
+      .then((source) => {
+        fakerSource = source;
+      })
+      .then(() => createJsonDestinationViaApi())
+      .then((destination) => {
+        jsonDestination = destination;
+      })
+      .then(() => createNewConnectionViaApi(fakerSource, jsonDestination))
+      .then((connectionResponse) => {
+        connection = connectionResponse;
+      });
+  });
+
+  after(() => {
+    setFeatureFlags({});
+  });
+
+  describe("catalog search functionality", () => {
+    before(() => {
+      visit(connection, "replication");
+    });
+
+    it("Should find a nested field in the sync catalog", () => {
+      // Intentionally search for a partial match of a nested field (address.city)
+      cy.get(CATALOG_SEARCH_INPUT).type("address.cit");
+      // Expect the parent field to exist
+      cy.contains(/^address$/).should("exist");
+      // Exppect the nested field to exist
+      cy.contains(/^address\.city$/).should("exist");
+
+      // Search for a stream
+      cy.get(CATALOG_SEARCH_INPUT).clear();
+      cy.get(CATALOG_SEARCH_INPUT).type("products");
+      cy.contains(/^products$/).should("exist");
+      cy.contains(/^users$/).should("not.exist");
+    });
+  });
+});

airbyte-webapp/src/components/connection/ConnectionForm/SyncCatalogTable/SyncCatalogTable.tsx

@@ -4,15 +4,13 @@ import {
   ExpandedState,
   flexRender,
   getCoreRowModel,
-  getExpandedRowModel,
-  getFilteredRowModel,
   getGroupedRowModel,
   getSortedRowModel,
   useReactTable,
 } from "@tanstack/react-table";
 import classnames from "classnames";
 import set from "lodash/set";
-import React, { FC, useCallback, useEffect, useMemo, useState, useDeferredValue } from "react";
+import React, { FC, useCallback, useMemo, useState, useDeferredValue } from "react";
 import { useFieldArray, useFormContext, useWatch } from "react-hook-form";
 import { FormattedMessage, useIntl } from "react-intl";
 import { ItemProps, TableComponents, TableVirtuoso } from "react-virtuoso";
@@ -46,6 +44,8 @@ import { StreamFieldNameCell } from "./components/StreamFieldCell";
 import { StreamNameCell } from "./components/StreamNameCell";
 import { FilterTabId, StreamsFilterTabs } from "./components/StreamsFilterTabs";
 import { SyncModeCell } from "./components/SyncModeCell";
+import { getExpandedRowModel } from "./getExpandedRowModel";
+import { getFilteredRowModel } from "./getFilteredRowModel";
 import { useInitialRowIndex } from "./hooks/useInitialRowIndex";
 import styles from "./SyncCatalogTable.module.scss";
 import { getRowChangeStatus, getSyncCatalogRows, isNamespaceRow, isStreamRow } from "./utils";
@@ -330,6 +330,7 @@ export const SyncCatalogTable: FC<SyncCatalogTableProps> = ({ scrollParentContai
   });
 
   const rows = getRowModel().rows;
+
   const initialTopMostItemIndex = useInitialRowIndex(rows);
 
   const [isAllStreamRowsExpanded, setIsAllStreamRowsExpanded] = useState(false);
@@ -347,22 +348,6 @@ export const SyncCatalogTable: FC<SyncCatalogTableProps> = ({ scrollParentContai
     [initialExpandedState, toggleAllRowsExpanded]
   );
 
-  useEffect(() => {
-    // collapse all rows if global filter is empty and all rows are expanded
-    if (!filtering && isAllStreamRowsExpanded) {
-      toggleAllStreamRowsExpanded(false);
-      return;
-    }
-
-    // if global filter is empty or all rows already expanded then return
-    if (!filtering || (filtering && isAllStreamRowsExpanded)) {
-      return;
-    }
-
-    toggleAllStreamRowsExpanded(true);
-    // eslint-disable-next-line react-hooks/exhaustive-deps
-  }, [filtering]);
-
   const Table: TableComponents["Table"] = ({ style, ...props }) => (
     <table className={classnames(styles.table)} {...props} style={style} />
   );
@@ -480,6 +465,7 @@ export const SyncCatalogTable: FC<SyncCatalogTableProps> = ({ scrollParentContai
               // We do not want to submit the connection form when pressing Enter in the search field
               e.key === "Enter" && e.preventDefault();
             }}
+            data-testid="sync-catalog-search"
           />
           <FlexContainer>
             <FlexContainer justifyContent="flex-end" alignItems="center" direction="row" gap="lg">

airbyte-webapp/src/components/connection/ConnectionForm/SyncCatalogTable/filterRowsUtils.ts

@@ -0,0 +1,146 @@
+/* eslint-disable @typescript-eslint/no-explicit-any */
+import { RowModel, Table, createRow, Row, RowData } from "@tanstack/react-table";
+
+import { SyncCatalogUIModel } from "./SyncCatalogTable";
+
+/**
+ * This method is unchanged from the original implementation, except for removing the generic typings:
+ * https://github.com/TanStack/table/blob/v8.7.0/packages/table-core/src/utils/filterRowsUtils.ts#L4C1-L14C2
+ */
+export function filterRows(
+  rows: Array<Row<SyncCatalogUIModel>>,
+  filterRowImpl: (row: Row<SyncCatalogUIModel>) => any,
+  table: Table<SyncCatalogUIModel>
+) {
+  if (table.options.filterFromLeafRows) {
+    return filterRowModelFromLeafs(rows, filterRowImpl, table);
+  }
+
+  return filterRowModelFromRoot(rows, filterRowImpl, table);
+}
+
+/**
+ * This method has been customized to always include sibling fields of a matching row. Original implementation:
+ * https://github.com/TanStack/table/blob/v8.7.0/packages/table-core/src/utils/filterRowsUtils.ts#L16-L76
+ */
+function filterRowModelFromLeafs(
+  rowsToFilter: Array<Row<SyncCatalogUIModel>>,
+  filterRow: (row: Row<SyncCatalogUIModel>) => Array<Row<SyncCatalogUIModel>>,
+  table: Table<SyncCatalogUIModel>
+): RowModel<SyncCatalogUIModel> {
+  const maxDepth = table.options.maxLeafRowFilterDepth ?? 100;
+
+  const recurseFilterRows = (rowsToFilter: Array<Row<SyncCatalogUIModel>>, depth = 0) => {
+    const rows: Array<Row<SyncCatalogUIModel>> = [];
+
+    for (let i = 0; i < rowsToFilter.length; i++) {
+      const row = rowsToFilter[i];
+
+      const newRow = createRow(table, row.id, row.original, row.index, row.depth);
+      newRow.columnFilters = row.columnFilters;
+
+      if (row.subRows?.length && depth < maxDepth) {
+        // Recursing again here to do a depth first search (deep leaf nodes first)
+        newRow.subRows = recurseFilterRows(row.subRows, depth + 1);
+
+        // If a stream is a match, include all its subrows
+        if (newRow.original.rowType === "stream" && filterRow(newRow)) {
+          newRow.subRows = rowsToFilter[i].subRows;
+          rows.push(newRow);
+          continue;
+        }
+
+        // If a stream contains at least one match, include all its subrows
+        if (newRow.original.rowType === "stream" && newRow.subRows.length) {
+          newRow.subRows = rowsToFilter[i].subRows;
+          rows.push(newRow);
+          continue;
+        }
+
+        if (filterRow(newRow) && !newRow.subRows.length) {
+          rows.push(newRow);
+          continue;
+        }
+
+        if (filterRow(newRow) || newRow.subRows.length) {
+          rows.push(newRow);
+          continue;
+        }
+      } else if (filterRow(newRow)) {
+        rows.push(newRow);
+      }
+    }
+
+    return rows;
+  };
+
+  const filteredRows: Array<Row<SyncCatalogUIModel>> = recurseFilterRows(rowsToFilter);
+  const filteredRowsById: Record<string, Row<SyncCatalogUIModel>> = {};
+
+  function flattenRows(rows: Array<Row<SyncCatalogUIModel>>) {
+    const flattenedRows: Array<Row<SyncCatalogUIModel>> = [];
+    for (let i = 0; i < rows.length; i++) {
+      const row = rows[i];
+      flattenedRows.push(row);
+      filteredRowsById[row.id] = row;
+      if (row.subRows) {
+        flattenedRows.push(...flattenRows(row.subRows));
+      }
+    }
+    return flattenedRows;
+  }
+
+  return {
+    rows: filteredRows,
+    rowsById: filteredRowsById,
+    flatRows: flattenRows(filteredRows),
+  };
+}
+
+/**
+ * This method is unchanged from the original implementation:
+ * https://github.com/TanStack/table/blob/v8.7.0/packages/table-core/src/utils/filterRowsUtils.ts#L78-L126
+ */
+export function filterRowModelFromRoot<TData extends RowData>(
+  rowsToFilter: Array<Row<TData>>,
+  filterRow: (row: Row<TData>) => any,
+  table: Table<TData>
+): RowModel<TData> {
+  const newFilteredFlatRows: Array<Row<TData>> = [];
+  const newFilteredRowsById: Record<string, Row<TData>> = {};
+  const maxDepth = table.options.maxLeafRowFilterDepth ?? 100;
+
+  // Filters top level and nested rows
+  const recurseFilterRows = (rowsToFilter: Array<Row<TData>>, depth = 0) => {
+    // Filter from parents downward first
+
+    const rows = [];
+
+    // Apply the filter to any subRows
+    for (let i = 0; i < rowsToFilter.length; i++) {
+      let row = rowsToFilter[i];
+
+      const pass = filterRow(row);
+
+      if (pass) {
+        if (row.subRows?.length && depth < maxDepth) {
+          const newRow = createRow(table, row.id, row.original, row.index, row.depth);
+          newRow.subRows = recurseFilterRows(row.subRows, depth + 1);
+          row = newRow;
+        }
+
+        rows.push(row);
+        newFilteredFlatRows.push(row);
+        newFilteredRowsById[row.id] = row;
+      }
+    }
+
+    return rows;
+  };
+
+  return {
+    rows: recurseFilterRows(rowsToFilter),
+    flatRows: newFilteredFlatRows,
+    rowsById: newFilteredRowsById,
+  };
+}

airbyte-webapp/src/components/connection/ConnectionForm/SyncCatalogTable/getExpandedRowModel.ts

@@ -0,0 +1,83 @@
+/* eslint-disable @typescript-eslint/no-explicit-any */
+import { Table, Row, RowModel, TableOptionsResolved, memo } from "@tanstack/react-table";
+
+import { SyncCatalogUIModel } from "./SyncCatalogTable";
+
+export function getMemoOptions(
+  tableOptions: Partial<TableOptionsResolved<any>>,
+  debugLevel: "debugAll" | "debugTable" | "debugColumns" | "debugRows" | "debugHeaders",
+  key: string,
+  onChange?: (result: any) => void
+) {
+  return {
+    debug: () => tableOptions?.debugAll ?? tableOptions[debugLevel],
+    key: process.env.NODE_ENV === "development" && key,
+    onChange,
+  };
+}
+
+/**
+ * A custom implementation of getExpandedRowModel() from tanstack-table.
+ * Compare with original implementation at: https://github.com/TanStack/table/blob/7fe650d666cfc3807b8408a1205bc1686a479cdb/packages/table-core/src/utils/getExpandedRowModel.ts
+ */
+export function getExpandedRowModel(): (table: Table<SyncCatalogUIModel>) => () => RowModel<SyncCatalogUIModel> {
+  return (table) =>
+    memo(
+      () => [table.getState().expanded, table.getPreExpandedRowModel(), table.options.paginateExpandedRows],
+      (expanded, rowModel) => {
+        if (!rowModel.rows.length || (expanded !== true && !Object.keys(expanded ?? {}).length)) {
+          return rowModel;
+        }
+
+        return expandRows(rowModel, table.getState().globalFilter);
+      },
+      getMemoOptions(table.options, "debugTable", "getExpandedRowModel")
+    );
+}
+
+export function expandRows(rowModel: RowModel<SyncCatalogUIModel>, globalFilter: string) {
+  const expandedRows: Array<Row<SyncCatalogUIModel>> = [];
+
+  const recurseMatchingSubRows = (subRows: Array<Row<SyncCatalogUIModel>>): Array<Row<SyncCatalogUIModel>> => {
+    const matchingRows: Array<Row<SyncCatalogUIModel>> = [];
+
+    subRows.forEach((row) => {
+      if (rowHasSearchTerm(row, globalFilter)) {
+        matchingRows.push(row);
+      }
+      if (row.subRows?.length) {
+        const matchingSubrows = recurseMatchingSubRows(row.subRows);
+        // If any subrows match, include the parent row and the matching subrows
+        if (matchingSubrows.length) {
+          matchingRows.push(row);
+          matchingRows.push(...matchingSubrows);
+        }
+      }
+    });
+
+    return matchingRows;
+  };
+
+  const handleRow = (row: Row<SyncCatalogUIModel>) => {
+    expandedRows.push(row);
+
+    if (row.subRows?.length && row.getIsExpanded()) {
+      row.subRows.forEach(handleRow);
+      // Even if the parent is not expanded, we need to iterate through the subRows to look for rows that match the search
+    } else if (row.subRows?.length && globalFilter !== "" && !row.getIsExpanded()) {
+      expandedRows.push(...recurseMatchingSubRows(row.subRows));
+    }
+  };
+
+  rowModel.rows.forEach(handleRow);
+
+  return {
+    rows: expandedRows,
+    flatRows: rowModel.flatRows,
+    rowsById: rowModel.rowsById,
+  };
+}
+
+function rowHasSearchTerm(row: Row<SyncCatalogUIModel>, globalFilter: string) {
+  return row.original.name.toLocaleLowerCase().includes(globalFilter.toLocaleLowerCase());
+}