[Connector Builder] Add stream slicer

airbyte-webapp/src/components/connectorBuilder/Builder/BuilderField.tsx

@@ -32,6 +32,7 @@ interface BaseFieldProps {
   tooltip?: string;
   readOnly?: boolean;
   optional?: boolean;
+  pattern?: RegExp;
 }
 
 type BuilderFieldProps = BaseFieldProps &
@@ -66,6 +67,7 @@ export const BuilderField: React.FC<BuilderFieldProps> = ({
   tooltip,
   optional = false,
   readOnly,
+  pattern,
   ...props
 }) => {
   const [field, meta, helpers] = useField(path);
@@ -116,7 +118,10 @@ export const BuilderField: React.FC<BuilderFieldProps> = ({
       )}
       {hasError && (
         <Text className={styles.error}>
-          <FormattedMessage id={meta.error} />
+          <FormattedMessage
+            id={meta.error}
+            values={meta.error === "form.pattern.error" && pattern ? { pattern: String(pattern) } : undefined}
+          />
         </Text>
       )}
     </ControlLabels>

airbyte-webapp/src/components/connectorBuilder/Builder/StreamConfigView.tsx

@@ -16,6 +16,7 @@ import { BuilderTitle } from "./BuilderTitle";
 import { KeyValueListField } from "./KeyValueListField";
 import { PaginationSection } from "./PaginationSection";
 import styles from "./StreamConfigView.module.scss";
+import { StreamSlicerSection } from "./StreamSlicerSection";
 
 interface StreamConfigViewProps {
   streamNum: number;
@@ -97,6 +98,7 @@ export const StreamConfigView: React.FC<StreamConfigViewProps> = ({ streamNum })
         />
       </BuilderCard>
       <PaginationSection streamFieldPath={streamFieldPath} />
+      <StreamSlicerSection streamFieldPath={streamFieldPath} />
       <BuilderCard>
         <KeyValueListField
           path={streamFieldPath("requestOptions.requestParameters")}

airbyte-webapp/src/components/connectorBuilder/Builder/StreamSlicerSection.tsx

@@ -0,0 +1,182 @@
+import { useField } from "formik";
+
+import { ControlLabels } from "components/LabeledControl";
+
+import { timeDeltaRegex } from "../types";
+import { BuilderCard } from "./BuilderCard";
+import { BuilderField } from "./BuilderField";
+import { BuilderOneOf } from "./BuilderOneOf";
+import { BuilderOptional } from "./BuilderOptional";
+import { InjectRequestOptionFields } from "./InjectRequestOptionFields";
+import { ToggleGroupField } from "./ToggleGroupField";
+
+interface StreamSlicerSectionProps {
+  streamFieldPath: (fieldPath: string) => string;
+}
+
+export const StreamSlicerSection: React.FC<StreamSlicerSectionProps> = ({ streamFieldPath }) => {
+  const [field, , helpers] = useField(streamFieldPath("streamSlicer"));
+
+  const handleToggle = (newToggleValue: boolean) => {
+    if (newToggleValue) {
+      helpers.setValue({
+        type: "ListStreamSlicer",
+        slice_values: [],
+        cursor_field: "",
+      });
+    } else {
+      helpers.setValue(undefined);
+    }
+  };
+  const toggledOn = field.value !== undefined;
+
+  return (
+    <BuilderCard
+      toggleConfig={{
+        label: (
+          <ControlLabels
+            label="Stream slicer"
+            infoTooltipContent="Configure how to partition a stream into subsets of records and iterate over the data"
+          />
+        ),
+        toggledOn,
+        onToggle: handleToggle,
+      }}
+    >
+      <BuilderOneOf
+        path={streamFieldPath("streamSlicer")}
+        label="Mode"
+        tooltip="Stream slicer method to use on this stream"
+        options={[
+          {
+            label: "List",
+            typeValue: "ListStreamSlicer",
+            children: (
+              <>
+                <BuilderField
+                  type="array"
+                  path={streamFieldPath("streamSlicer.slice_values")}
+                  label="Slice values"
+                  tooltip="List of values to iterate over"
+                />
+                <BuilderField
+                  type="string"
+                  path={streamFieldPath("streamSlicer.cursor_field")}
+                  label="Cursor field"
+                  tooltip="Field on record to use as the cursor"
+                />
+                <ToggleGroupField
+                  label="Slice request option"
+                  tooltip="Optionally configures how the slice values will be sent in requests to the source API"
+                  fieldPath={streamFieldPath("streamSlicer.request_option")}
+                  initialValues={{
+                    inject_into: "request_parameter",
+                    field_name: "",
+                  }}
+                >
+                  <InjectRequestOptionFields
+                    path={streamFieldPath("streamSlicer.request_option")}
+                    descriptor="slice value"
+                    excludeInjectIntoValues={["path"]}
+                  />
+                </ToggleGroupField>
+              </>
+            ),
+          },
+          {
+            label: "Datetime",
+            typeValue: "DatetimeStreamSlicer",
+            children: (
+              <>
+                <BuilderField
+                  type="string"
+                  path={streamFieldPath("streamSlicer.datetime_format")}
+                  label="Datetime format"
+                  tooltip="Specify the format of the start and end time, e.g. %Y-%m-%d"
+                />
+                <BuilderField
+                  type="string"
+                  path={streamFieldPath("streamSlicer.start_datetime")}
+                  label="Start datetime"
+                  tooltip="Start time to start slicing"
+                />
+                <BuilderField
+                  type="string"
+                  path={streamFieldPath("streamSlicer.end_datetime")}
+                  label="End datetime"
+                  tooltip="End time to end slicing"
+                />
+                <BuilderField
+                  type="string"
+                  path={streamFieldPath("streamSlicer.step")}
+                  label="Step"
+                  tooltip="Time interval for which to break up stream into slices, e.g. 1d"
+                  pattern={timeDeltaRegex}
+                />
+                <BuilderField
+                  type="string"
+                  path={streamFieldPath("streamSlicer.cursor_field")}
+                  label="Cursor field"
+                  tooltip="Field on record to use as the cursor"
+                />
+                <BuilderOptional>
+                  <BuilderField
+                    type="string"
+                    path={streamFieldPath("streamSlicer.lookback_window")}
+                    label="Lookback window"
+                    tooltip="How many days before the start_datetime to read data for, e.g. 31d"
+                    optional
+                  />
+                  <ToggleGroupField
+                    label="Start time request option"
+                    tooltip="Optionally configures how the start datetime will be sent in requests to the source API"
+                    fieldPath={streamFieldPath("streamSlicer.start_time_option")}
+                    initialValues={{
+                      inject_into: "request_parameter",
+                      field_name: "",
+                    }}
+                  >
+                    <InjectRequestOptionFields
+                      path={streamFieldPath("streamSlicer.start_time_option")}
+                      descriptor="start datetime"
+                      excludeInjectIntoValues={["path"]}
+                    />
+                  </ToggleGroupField>
+                  <ToggleGroupField
+                    label="End time request option"
+                    tooltip="Optionally configures how the end datetime will be sent in requests to the source API"
+                    fieldPath={streamFieldPath("streamSlicer.end_time_option")}
+                    initialValues={{
+                      inject_into: "request_parameter",
+                      field_name: "",
+                    }}
+                  >
+                    <InjectRequestOptionFields
+                      path={streamFieldPath("streamSlicer.end_time_option")}
+                      descriptor="end datetime"
+                      excludeInjectIntoValues={["path"]}
+                    />
+                  </ToggleGroupField>
+                  <BuilderField
+                    type="string"
+                    path={streamFieldPath("streamSlicer.stream_state_field_start")}
+                    label="Stream state field start"
+                    tooltip="Set which field on the stream state to use to determine the starting point"
+                    optional
+                  />
+                  <BuilderField
+                    type="string"
+                    path={streamFieldPath("streamSlicer.stream_state_field_end")}
+                    label="Stream state field end"
+                    tooltip="Set which field on the stream state to use to determine the ending point"
+                    optional
+                  />
+                </BuilderOptional>
+              </>
+            ),
+          },
+        ]}
+      />
+    </BuilderCard>
+  );
+};

airbyte-webapp/src/components/connectorBuilder/types.ts

@@ -15,6 +15,7 @@ import {
   SessionTokenAuthenticator,
   DefaultPaginatorAllOfPaginationStrategy,
   RequestOption,
+  SimpleRetrieverAllOfStreamSlicer,
 } from "core/request/ConnectorManifest";
 
 export interface BuilderFormInput {
@@ -61,6 +62,7 @@ export interface BuilderStream {
     pageTokenOption: RequestOption;
     pageSizeOption?: RequestOption;
   };
+  streamSlicer?: SimpleRetrieverAllOfStreamSlicer;
 }
 
 export const DEFAULT_BUILDER_FORM_VALUES: BuilderFormValues = {
@@ -213,6 +215,17 @@ export function getInferredInputs(values: BuilderFormValues): BuilderFormInput[]
 }
 
 export const injectIntoValues = ["request_parameter", "header", "path", "body_data", "body_json"];
+const nonPathRequestOptionSchema = yup
+  .object()
+  .shape({
+    inject_into: yup.mixed().oneOf(injectIntoValues.filter((val) => val !== "path")),
+    field_name: yup.string().required("form.empty.error"),
+  })
+  .notRequired()
+  .default(undefined);
+
+// eslint-disable-next-line no-useless-escape
+export const timeDeltaRegex = /^(([\.\d]+?)y)?(([\.\d]+?)m)?(([\.\d]+?)w)?(([\.\d]+?)d)?$/;
 
 export const builderFormValidationSchema = yup.object().shape({
   global: yup.object().shape({
@@ -261,14 +274,7 @@ export const builderFormValidationSchema = yup.object().shape({
       paginator: yup
         .object()
         .shape({
-          pageSizeOption: yup
-            .object()
-            .shape({
-              inject_into: yup.mixed().oneOf(injectIntoValues.filter((val) => val !== "path")),
-              field_name: yup.string().required("form.empty.error"),
-            })
-            .notRequired()
-            .default(undefined),
+          pageSizeOption: nonPathRequestOptionSchema,
           pageTokenOption: yup.object().shape({
             inject_into: yup.mixed().oneOf(injectIntoValues),
             field_name: yup.mixed().when("inject_into", {
@@ -305,6 +311,64 @@ export const builderFormValidationSchema = yup.object().shape({
         })
         .notRequired()
         .default(undefined),
+      streamSlicer: yup
+        .object()
+        .shape({
+          cursor_field: yup.string().required("form.empty.error"),
+          slice_values: yup.mixed().when("type", {
+            is: "ListStreamSlicer",
+            then: yup.array().of(yup.string()),
+            otherwise: (schema) => schema.strip(),
+          }),
+          request_option: nonPathRequestOptionSchema,
+          start_datetime: yup.mixed().when("type", {
+            is: "DatetimeStreamSlicer",
+            then: yup.string().required("form.empty.error"),
+            otherwise: (schema) => schema.strip(),
+          }),
+          end_datetime: yup.mixed().when("type", {
+            is: "DatetimeStreamSlicer",
+            then: yup.string().required("form.empty.error"),
+            otherwise: (schema) => schema.strip(),
+          }),
+          step: yup.mixed().when("type", {
+            is: "DatetimeStreamSlicer",
+            then: yup.string().matches(timeDeltaRegex, "form.pattern.error").required("form.empty.error"),
+            otherwise: (schema) => schema.strip(),
+          }),
+          datetime_format: yup.mixed().when("type", {
+            is: "DatetimeStreamSlicer",
+            then: yup.string().required("form.empty.error"),
+            otherwise: (schema) => schema.strip(),
+          }),
+          start_time_option: yup.mixed().when("type", {
+            is: "DatetimeStreamSlicer",
+            then: nonPathRequestOptionSchema,
+            otherwise: (schema) => schema.strip(),
+          }),
+          end_time_option: yup.mixed().when("type", {
+            is: "DatetimeStreamSlicer",
+            then: nonPathRequestOptionSchema,
+            otherwise: (schema) => schema.strip(),
+          }),
+          stream_state_field_start: yup.mixed().when("type", {
+            is: "DatetimeStreamSlicer",
+            then: yup.string(),
+            otherwise: (schema) => schema.strip(),
+          }),
+          stream_state_field_end: yup.mixed().when("type", {
+            is: "DatetimeStreamSlicer",
+            then: yup.string(),
+            otherwise: (schema) => schema.strip(),
+          }),
+          lookback_window: yup.mixed().when("type", {
+            is: "DatetimeStreamSlicer",
+            then: yup.string(),
+            otherwise: (schema) => schema.strip(),
+          }),
+        })
+        .notRequired()
+        .default(undefined),
     })
   ),
 });
@@ -369,6 +433,7 @@ export const convertToManifest = (values: BuilderFormValues): PatchedConnectorMa
               url_base: values.global?.urlBase,
             }
           : { type: "NoPagination" },
+        stream_slicer: stream.streamSlicer,
         config: {},
       },
       config: {},