Highlight invalid cursor or primary key in new connection streams table

airbyte-webapp/src/components/connection/CatalogTree/CatalogSection.tsx

@@ -166,7 +166,12 @@ const CatalogSectionInner: React.FC<CatalogSectionInnerProps> = ({
   );
 
   const destName = prefix + (streamNode.stream?.name ?? "");
-  const configErrors = getIn(errors, `schema.streams[${streamNode.id}].config`);
+  const configErrors = getIn(
+    errors,
+    isNewStreamsTableEnabled
+      ? `syncCatalog.streams[${streamNode.id}].config`
+      : `schema.streams[${streamNode.id}].config`
+  );
   const hasError = configErrors && Object.keys(configErrors).length > 0;
   const pkType = getPathType(pkRequired, shouldDefinePk);
   const cursorType = getPathType(cursorRequired, shouldDefineCursor);
@@ -194,6 +199,7 @@ const CatalogSectionInner: React.FC<CatalogSectionInnerProps> = ({
         onExpand={onExpand}
         changedSelected={changedSelected}
         hasError={hasError}
+        configErrors={configErrors}
         disabled={disabled}
       />
       {isRowExpanded &&

airbyte-webapp/src/components/connection/CatalogTree/StreamHeader.tsx

@@ -43,6 +43,7 @@ export interface StreamHeaderProps {
   onExpand: () => void;
   changedSelected: boolean;
   hasError: boolean;
+  configErrors?: Record<string, string>;
   disabled?: boolean;
 }
 

airbyte-webapp/src/components/connection/CatalogTree/next/CatalogTreeTableRow.module.scss

@@ -35,10 +35,6 @@
     @include header-background(colors.$blue-30, colors.$blue-40);
   }
 
-  &.error {
-    border: 1px solid colors.$red;
-  }
-
   &.disabled {
     background-color: colors.$grey-50;
   }

airbyte-webapp/src/components/connection/CatalogTree/next/CatalogTreeTableRow.tsx

@@ -35,9 +35,9 @@ export const CatalogTreeTableRow: React.FC<StreamHeaderProps> = ({
   fields,
   onExpand,
   disabled,
+  configErrors,
 }) => {
   const { primaryKey, cursorField, syncMode, destinationSyncMode } = stream.config ?? {};
-
   const { defaultCursorField } = stream.stream ?? {};
   const syncSchema = useMemo(
     () => ({
@@ -104,6 +104,7 @@ export const CatalogTreeTableRow: React.FC<StreamHeaderProps> = ({
             path={cursorType === "sourceDefined" ? defaultCursorField : cursorField}
             onPathChange={onCursorChange}
             variant={pillButtonVariant}
+            hasError={!!configErrors?.cursorField}
           />
         )}
       </Cell>
@@ -116,6 +117,7 @@ export const CatalogTreeTableRow: React.FC<StreamHeaderProps> = ({
             isMulti
             onPathChange={onPrimaryKeyChange}
             variant={pillButtonVariant}
+            hasError={!!configErrors?.primaryKey}
           />
         )}
       </Cell>

airbyte-webapp/src/components/connection/CatalogTree/next/StreamPathSelect.tsx

@@ -19,6 +19,7 @@ interface StreamPathSelectBaseProps {
   placeholder?: React.ReactNode;
   variant?: PillButtonVariant;
   disabled?: boolean;
+  hasError?: boolean;
 }
 
 interface StreamPathSelectMultiProps {
@@ -69,6 +70,7 @@ export const StreamPathSelect: React.FC<PathPopoutProps> = (props) => {
         props.onPathChange(finalValues);
       }}
       className={styles.streamPathSelect}
+      hasError={props?.hasError}
     />
   );
 };

airbyte-webapp/src/components/connection/CatalogTree/next/useCatalogTreeTableRowProps.test.tsx

@@ -143,11 +143,4 @@ describe("useCatalogTreeTableRowProps", () => {
     expect(result.current.streamHeaderContentStyle).toEqual("streamHeaderContent changed");
     expect(result.current.pillButtonVariant).toEqual("blue");
   });
-  it("should return error styles for a row that has an error", () => {
-    testSetup(mockInitialValues, false, "error");
-
-    const { result } = renderHook(() => useCatalogTreeTableRowProps(mockStream));
-
-    expect(result.current.streamHeaderContentStyle).toEqual("streamHeaderContent error");
-  });
 });

airbyte-webapp/src/components/connection/CatalogTree/next/useCatalogTreeTableRowProps.tsx

@@ -1,6 +1,5 @@
 /* eslint-disable css-modules/no-unused-class */
 import classNames from "classnames";
-import { useField } from "formik";
 import isEqual from "lodash/isEqual";
 import { useMemo } from "react";
 
@@ -11,16 +10,12 @@ import { useBulkEditSelect } from "hooks/services/BulkEdit/BulkEditService";
 import { useConnectionFormService } from "hooks/services/ConnectionForm/ConnectionFormService";
 
 import styles from "./CatalogTreeTableRow.module.scss";
+
 type StatusToDisplay = "disabled" | "added" | "removed" | "changed" | "unchanged";
 
 export const useCatalogTreeTableRowProps = (stream: SyncSchemaStream) => {
-  const { initialValues } = useConnectionFormService();
   const [isSelected] = useBulkEditSelect(stream.id);
-
-  const [, { error }] = useField(`schema.streams[${stream.id}].config`);
-
-  // in case error is an empty string
-  const hasError = error !== undefined;
+  const { initialValues } = useConnectionFormService();
 
   const isStreamEnabled = stream.config?.selected;
 
@@ -67,7 +62,6 @@ export const useCatalogTreeTableRowProps = (stream: SyncSchemaStream) => {
     [styles.removed]: statusToDisplay === "removed" && !isSelected,
     [styles.changed]: statusToDisplay === "changed" || isSelected,
     [styles.disabled]: statusToDisplay === "disabled",
-    [styles.error]: hasError,
   });
 
   return {

airbyte-webapp/src/components/ui/PillSelect/PillButton.tsx

@@ -20,19 +20,27 @@ const STYLES_BY_VARIANT: Readonly<Record<PillButtonVariant, string>> = {
 interface PillButtonProps extends React.ButtonHTMLAttributes<HTMLButtonElement> {
   active?: boolean;
   variant?: PillButtonVariant;
+  hasError?: boolean;
 }
 
-export const PillButton: React.FC<PillButtonProps> = ({ children, active, variant = "grey", ...buttonProps }) => {
+export const PillButton: React.FC<PillButtonProps> = ({
+  children,
+  active,
+  variant = "grey",
+  hasError = false,
+  ...buttonProps
+}) => {
   const buttonClassName = classNames(
     styles.button,
     {
       [styles.active]: active,
       [styles.disabled]: buttonProps.disabled,
     },
-    STYLES_BY_VARIANT[variant],
+    STYLES_BY_VARIANT[hasError ? "strong-red" : variant],
     buttonProps.className
   );
   const arrayChildren = Children.toArray(children);
+
   return (
     <button type="button" {...buttonProps} className={buttonClassName}>
       {Children.map(arrayChildren, (child, index) => (

airbyte-webapp/src/components/ui/PillSelect/PillSelect.tsx

@@ -7,6 +7,7 @@ type PickedPopoutProps = Pick<PopoutProps, "value" | "options" | "isMulti" | "on
 interface PillSelectProps extends PickedPopoutProps {
   variant?: PillButtonVariant;
   disabled?: boolean;
+  hasError?: boolean;
 }
 
 export const PillSelect: React.FC<PillSelectProps> = ({ className, ...props }) => {
@@ -33,6 +34,7 @@ export const PillSelect: React.FC<PillSelectProps> = ({ className, ...props }) =
                 }}
                 active={isOpen}
                 className={className}
+                hasError={props?.hasError}
               >
                 {label}
               </PillButton>

airbyte-webapp/src/hooks/services/ConnectionForm/ConnectionFormService.tsx

@@ -85,12 +85,23 @@ const useConnectionForm = ({
   const formId = useUniqueFormId();
 
   const getErrorMessage = useCallback(
-    (formValid: boolean, connectionDirty: boolean) =>
-      submitError
+    (formValid: boolean, connectionDirty: boolean) => {
+      const isNewStreamsTableEnabled = process.env.REACT_APP_NEW_STREAMS_TABLE ?? false;
+
+      if (isNewStreamsTableEnabled) {
+        // There is a case when some fields could be dropped in the database. We need to validate the form without property dirty
+        return submitError
+          ? generateMessageFromError(submitError)
+          : !formValid
+          ? formatMessage({ id: "connectionForm.validation.error" })
+          : null;
+      }
+      return submitError
         ? generateMessageFromError(submitError)
         : connectionDirty && !formValid
         ? formatMessage({ id: "connectionForm.validation.error" })
-        : null,
+        : null;
+    },
     [formatMessage, submitError]
   );
 

airbyte-webapp/src/views/Connection/ConnectionForm/formConfig.tsx

@@ -90,8 +90,10 @@ export const createConnectionValidationSchema = ({
   mode,
   allowSubOneHourCronExpressions,
   allowAutoDetectSchema,
-}: CreateConnectionValidationSchemaArgs) =>
-  yup
+}: CreateConnectionValidationSchemaArgs) => {
+  const isNewStreamsTableEnabled = process.env.REACT_APP_NEW_STREAMS_TABLE ?? false;
+
+  return yup
     .object({
       // The connection name during Editing is handled separately from the form
       name: mode === "create" ? yup.string().required("form.empty.error") : yup.string().notRequired(),
@@ -112,7 +114,6 @@ export const createConnectionValidationSchema = ({
         } else if (scheduleType === ConnectionScheduleType.manual) {
           return yup.mixed().notRequired();
         }
-
         return yup.object({
           cron: yup
             .object({
@@ -134,7 +135,6 @@ export const createConnectionValidationSchema = ({
       nonBreakingChangesPreference: allowAutoDetectSchema
         ? yup.mixed().oneOf(Object.values(NonBreakingChangesPreference)).required("form.empty.error")
         : yup.mixed().notRequired(),
-
       namespaceDefinition: yup
         .string()
         .oneOf([
@@ -148,64 +148,105 @@ export const createConnectionValidationSchema = ({
         then: yup.string().trim().required("form.empty.error"),
       }),
       prefix: yup.string(),
-      syncCatalog: yup.object({
-        streams: yup.array().of(
-          yup.object({
-            id: yup
-              .string()
-              // This is required to get rid of id fields we are using to detect stream for edition
-              .when("$isRequest", (isRequest: boolean, schema: yup.StringSchema) =>
-                isRequest ? schema.strip(true) : schema
-              ),
-            stream: yup.object(),
-            config: yup
-              .object({
-                selected: yup.boolean(),
-                syncMode: yup.string(),
-                destinationSyncMode: yup.string(),
-                primaryKey: yup.array().of(yup.array().of(yup.string())),
-                cursorField: yup.array().of(yup.string()).defined(),
+      syncCatalog: isNewStreamsTableEnabled
+        ? yup.object({
+            streams: yup.array().of(
+              yup.object({
+                id: yup
+                  .string()
+                  // This is required to get rid of id fields we are using to detect stream for edition
+                  .when("$isRequest", (isRequest: boolean, schema: yup.StringSchema) =>
+                    isRequest ? schema.strip(true) : schema
+                  ),
+                stream: yup.object(),
+                config: yup.object({
+                  selected: yup.boolean(),
+                  syncMode: yup.string(),
+                  destinationSyncMode: yup.string(),
+                  primaryKey: yup
+                    .array()
+                    .of(yup.array().of(yup.string()))
+                    .when(["syncMode", "destinationSyncMode", "selected"], {
+                      is: (syncMode: SyncMode, destinationSyncMode: DestinationSyncMode, selected: boolean) =>
+                        syncMode === SyncMode.incremental &&
+                        destinationSyncMode === DestinationSyncMode.append_dedup &&
+                        selected,
+                      then: yup.array().of(yup.array().of(yup.string())).min(1, "form.empty.error"),
+                    }),
+                  cursorField: yup
+                    .array()
+                    .of(yup.string())
+                    .when(["syncMode", "destinationSyncMode", "selected"], {
+                      is: (syncMode: SyncMode, destinationSyncMode: DestinationSyncMode, selected: boolean) =>
+                        (destinationSyncMode === DestinationSyncMode.append ||
+                          destinationSyncMode === DestinationSyncMode.append_dedup) &&
+                        syncMode === SyncMode.incremental &&
+                        selected,
+                      then: yup.array().of(yup.string()).min(1, "form.empty.error"),
+                    }),
+                }),
               })
-              .test({
-                name: "connectionSchema.config.validator",
-                // eslint-disable-next-line no-template-curly-in-string
-                message: "${path} is wrong",
-                test(value) {
-                  if (!value.selected) {
-                    return true;
-                  }
-                  if (DestinationSyncMode.append_dedup === value.destinationSyncMode) {
-                    // it's possible that primaryKey array is always present
-                    // however yup couldn't determine type correctly even with .required() call
-                    if (value.primaryKey?.length === 0) {
-                      return this.createError({
-                        message: "connectionForm.primaryKey.required",
-                        path: `schema.streams[${this.parent.id}].config.primaryKey`,
-                      });
-                    }
-                  }
-
-                  if (SyncMode.incremental === value.syncMode) {
-                    if (
-                      !this.parent.stream.sourceDefinedCursor &&
-                      // it's possible that cursorField array is always present
-                      // however yup couldn't determine type correctly even with .required() call
-                      value.cursorField?.length === 0
-                    ) {
-                      return this.createError({
-                        message: "connectionForm.cursorField.required",
-                        path: `schema.streams[${this.parent.id}].config.cursorField`,
-                      });
-                    }
-                  }
-                  return true;
-                },
-              }),
+            ),
           })
-        ),
-      }),
+        : yup.object({
+            streams: yup.array().of(
+              yup.object({
+                id: yup
+                  .string()
+                  // This is required to get rid of id fields we are using to detect stream for edition
+                  .when("$isRequest", (isRequest: boolean, schema: yup.StringSchema) =>
+                    isRequest ? schema.strip(true) : schema
+                  ),
+                stream: yup.object(),
+                config: yup
+                  .object({
+                    selected: yup.boolean(),
+                    syncMode: yup.string(),
+                    destinationSyncMode: yup.string(),
+                    primaryKey: yup.array().of(yup.array().of(yup.string())),
+                    cursorField: yup.array().of(yup.string()).defined(),
+                  })
+                  .test({
+                    name: "connectionSchema.config.validator",
+                    // eslint-disable-next-line no-template-curly-in-string
+                    message: "${path} is wrong",
+                    test(value) {
+                      if (!value.selected) {
+                        return true;
+                      }
+                      if (DestinationSyncMode.append_dedup === value.destinationSyncMode) {
+                        // it's possible that primaryKey array is always present
+                        // however yup couldn't determine type correctly even with .required() call
+                        if (value.primaryKey?.length === 0) {
+                          return this.createError({
+                            message: "connectionForm.primaryKey.required",
+                            path: `schema.streams[${this.parent.id}].config.primaryKey`,
+                          });
+                        }
+                      }
+
+                      if (SyncMode.incremental === value.syncMode) {
+                        if (
+                          !this.parent.stream.sourceDefinedCursor &&
+                          // it's possible that cursorField array is always present
+                          // however yup couldn't determine type correctly even with .required() call
+                          value.cursorField?.length === 0
+                        ) {
+                          return this.createError({
+                            message: "connectionForm.cursorField.required",
+                            path: `schema.streams[${this.parent.id}].config.cursorField`,
+                          });
+                        }
+                      }
+                      return true;
+                    },
+                  }),
+              })
+            ),
+          }),
     })
     .noUnknown();
+};
 
 /**
  * Returns {@link Operation}[]