hack

enterprise/backend/src/metabase_enterprise/transforms/api.clj

@@ -9,11 +9,13 @@
    [metabase-enterprise.transforms.models.transform-run :as transform-run]
    [metabase-enterprise.transforms.models.transform-run-cancelation :as transform-run-cancelation]
    [metabase-enterprise.transforms.ordering :as transforms.ordering]
+   [metabase-enterprise.transforms.schema :as transforms.schema]
    [metabase-enterprise.transforms.util :as transforms.util]
    [metabase.api.common :as api]
    [metabase.api.macros :as api.macros]
    [metabase.api.routes.common :refer [+auth]]
    [metabase.api.util.handlers :as handlers]
+   [metabase.driver :as driver]
    [metabase.driver.util :as driver.u]
    [metabase.events.core :as events]
    [metabase.request.core :as request]
@@ -31,26 +33,6 @@
 
 (set! *warn-on-reflection* true)
 
-(mr/def ::transform-source
-  [:multi {:dispatch (comp keyword :type)}
-   [:query
-    [:map
-     [:type [:= "query"]]
-     [:query [:map [:database :int]]]]]
-   [:python
-    [:map {:closed true}
-     [:source-database {:optional true} :int]
-     [:source-tables   [:map-of :string :int]]
-     [:type [:= "python"]]
-     [:body :string]]]])
-
-(mr/def ::transform-target
-  [:map
-   [:database {:optional true} :int]
-   [:type [:enum "table"]]
-   [:schema {:optional true} [:or ms/NonBlankString :nil]]
-   [:name :string]])
-
 (mr/def ::run-trigger
   [:enum "none" "global-schedule"])
 
@@ -103,6 +85,21 @@
                 (map #(update % :last_run transforms.util/localize-run-timestamps)))
           (t2/hydrate transforms :last_run :transform_tag_ids))))
 
+(defn- watermark-table-for-transform [{:keys [target]}]
+  (assoc target :name "mb__watermark_table"))
+
+(defn- ensure-mb-watermark-table-exists! [{:keys [target] :as transform}]
+  (when (= :table-incremental (-> target :type keyword))
+    (let [db-id (transforms.util/target-database-id transform)
+          {driver :engine :as database} (t2/select-one :model/Database db-id)
+          watermark-table (watermark-table-for-transform transform)
+          watermark-table-name (transforms.util/qualified-table-name driver watermark-table)]
+      (when-not (driver/table-exists? driver database watermark-table)
+        (driver/create-table! driver db-id watermark-table-name
+                              {:transform_id (driver/type->database-type driver :type/Integer),
+                               :watermark_id (driver/type->database-type driver :type/Integer)}
+                              :primary-key [:transform_id])))))
+
 (api.macros/defendpoint :post "/"
   "Create a new transform."
   [_route-params
@@ -110,13 +107,14 @@
    body :- [:map
             [:name :string]
             [:description {:optional true} [:maybe :string]]
-            [:source ::transform-source]
-            [:target ::transform-target]
+            [:source ::transforms.schema/transform-source]
+            [:target ::transforms.schema/transform-target]
             [:run_trigger {:optional true} ::run-trigger]
             [:tag_ids {:optional true} [:sequential ms/PositiveInt]]]]
   (api/check-superuser)
   (check-database-feature body)
   (check-feature-enabled! body)
+  (ensure-mb-watermark-table-exists! body)
   (api/check (not (transforms.util/target-table-exists? body))
              403
              (deferred-tru "A table with that name already exists."))
@@ -185,8 +183,8 @@
    body :- [:map
             [:name {:optional true} :string]
             [:description {:optional true} [:maybe :string]]
-            [:source {:optional true} ::transform-source]
-            [:target {:optional true} ::transform-target]
+            [:source {:optional true} ::transforms.schema/transform-source]
+            [:target {:optional true} ::transforms.schema/transform-target]
             [:run_trigger {:optional true} ::run-trigger]
             [:tag_ids {:optional true} [:sequential ms/PositiveInt]]]]
   (log/info "put transform" id)
@@ -197,6 +195,7 @@
                           new (merge old body)
                           target-fields #(-> % :target (select-keys [:schema :name]))]
                       ;; we must validate on a full transform object
+                      (ensure-mb-watermark-table-exists! new)
                       (check-feature-enabled! new)
                       (check-database-feature new)
                       (when (transforms.util/query-transform? old)

enterprise/backend/src/metabase_enterprise/transforms/execute.clj

@@ -14,7 +14,7 @@
 
 (mr/def ::transform-details
   [:map
-   [:transform-type [:enum {:decode/normalize schema.common/normalize-keyword} :table]]
+   [:transform-type [:enum {:decode/normalize schema.common/normalize-keyword} :table :table-incremental]]
    [:conn-spec :any]
    [:query :string]
    [:output-table [:keyword {:decode/normalize schema.common/normalize-keyword}]]])
@@ -23,6 +23,13 @@
   [:map
    [:overwrite? :boolean]])
 
+(defn- transform-opts [{:keys [transform-type]}]
+  (case transform-type
+    :table {:overwrite? true}
+
+    ;; TODO: append vs merge, for now we're just append
+    :table-incremental {}))
+
 (defn run-mbql-transform!
   "Run `transform` and sync its target table.
 
@@ -42,7 +49,7 @@
                               :query (transforms.util/compile-source source)
                               :output-schema (:schema target)
                               :output-table (transforms.util/qualified-table-name driver target)}
-           opts {:overwrite? true}]
+           opts (transform-opts transform-details)]
        (when (transforms.util/db-routing-enabled? database)
          (throw (ex-info "Transforms are not supported on databases with DB routing enabled."
                          {:driver driver, :database database})))
@@ -57,9 +64,11 @@
          (transforms.util/run-cancelable-transform!
           run-id driver transform-details
           (fn [_cancel-chan] (driver/run-transform! driver transform-details opts)))
+         ;; Update watermark tracking for incremental transforms
+         (transforms.util/upsert-watermark! transform driver database)
          (transforms.instrumentation/with-stage-timing [run-id :table-sync]
            (transforms.util/sync-target! target database run-id)
-         ;; This event must be published only after the sync is complete - the new table needs to be in AppDB.
+           ;; This event must be published only after the sync is complete - the new table needs to be in AppDB.
            (events/publish-event! :event/transform-run-complete {:object transform-details}))))
      (catch Throwable t
        (log/error t "Error executing transform")

enterprise/backend/src/metabase_enterprise/transforms/schema.clj

@@ -0,0 +1,55 @@
+(ns metabase-enterprise.transforms.schema
+  (:require
+   [metabase.util.malli.registry :as mr]
+   [metabase.util.malli.schema :as ms]))
+
+(mr/def ::keyset-strategy
+  [:map
+   [:type [:= "keyset"]]
+   [:keyset-column :string]
+   [:table-name {:optional true} :string]]) ; for python transforms with multiple tables
+
+(mr/def ::source-strategy
+  [:multi {:dispatch :type}
+   ["keyset" ::keyset-strategy]])
+
+(mr/def ::transform-source
+  [:multi {:dispatch (comp keyword :type)}
+   [:query
+    [:map
+     [:type [:= "query"]]
+     [:query [:map [:database :int]]]
+     [:source-strategy {:optional true} ::source-strategy]]]
+   [:python
+    [:map {:closed true}
+     [:source-database {:optional true} :int]
+     [:source-tables   [:map-of :string :int]]
+     [:type [:= "python"]]
+     [:body :string]
+     [:source-strategy {:optional true} ::source-strategy]]]])
+
+(mr/def ::append-config
+  [:map
+   [:type [:= "append"]]
+   #_[:tag-run-id? {:optional true} :boolean]]) ; whether to tag rows with run_id
+
+#_(mr/def ::merge-config
+    [:map
+     [:type [:= "merge"]]
+     [:primary-key [:sequential :string]]
+     [:update-columns {:optional true} [:sequential :string]]]) ; columns to update on conflict
+
+(mr/def ::target-incremental-strategy
+  [:multi {:dispatch :type}
+   ["append" ::append-config]
+   #_["merge" ::merge-config]])
+
+(mr/def ::transform-target
+  [:map
+   [:database {:optional true} :int]
+   [:type [:enum "table" "table-incremental"]]
+   [:schema {:optional true} [:or ms/NonBlankString :nil]]
+   [:name :string]
+
+   ;; only for table-incremental, TODO: specify with a multi spec
+   [:target-incremental-strategy {:optional true} ::target-incremental-strategy]])

enterprise/backend/src/metabase_enterprise/transforms/util.clj

@@ -105,6 +105,35 @@
    (log/info "Syncing target" (pr-str target) "for transform")
    (activate-table-and-mark-computed! database target)))
 
+(defn upsert-watermark!
+  "Update the watermark tracking table for an incremental transform.
+   Deletes any existing watermark entry for this transform and inserts a new one
+   with the MAX value of the watermark field from the transform output table."
+  [{:keys [id target] :as transform} driver database]
+  (when (= :table-incremental (keyword (:type target)))
+    (let [watermark-field (or (:watermarkField target) "id")
+          output-table (qualified-table-name driver target)
+          watermark-table (qualified-table-name driver {:schema (:schema target)
+                                                        :name "mb__watermark_table"})
+          conn-spec (driver/connection-spec driver database)
+          ;; lol
+          select-query (format "SELECT %d as transform_id, MAX(%s) as watermark_id FROM %s"
+                               id
+                               watermark-field
+                               (name output-table))
+          delete-sql (format "DELETE FROM %s WHERE transform_id = %d"
+                             (name watermark-table)
+                             id)]
+      (log/info "Upserting watermark for transform" id "with field" watermark-field)
+      (try
+        (let [insert-queries (driver/compile-insert driver {:query select-query
+                                                            :output-table watermark-table})]
+          (driver/execute-raw-queries! driver conn-spec [[delete-sql]
+                                                         insert-queries]))
+        (catch Exception e
+          (log/error e "Failed to upsert watermark for transform" id))))))
+
+;; TODO this and target-database-id can be transforms multimethods?
 (mu/defn target-database-id :- ::lib.schema.id/database
   "Return the target database id of a transform"
   [transform]
@@ -182,25 +211,27 @@
 (defn required-database-feature
   "Returns the database feature necessary to execute `transform`."
   [transform]
+  ;; TODO: not quite right
   (if (python-transform? transform)
     :transforms/python
-    (case (-> transform :target :type)
-      "table"             :transforms/table)))
+    (case (-> transform :target :type keyword)
+      :table :transforms/table
+      :table-incremental :transforms/table)))
 
 (defn ->instant
   "Convert a temporal value `t` to an Instant in the system timezone."
   ^Instant [t]
   (when t
     (condp instance? t
-      Instant        t
-      Date           (.toInstant ^Date t)
+      Instant t
+      Date (.toInstant ^Date t)
       OffsetDateTime (.toInstant ^OffsetDateTime t)
-      ZonedDateTime  (.toInstant ^ZonedDateTime t)
-      LocalDateTime  (recur (.atZone ^LocalDateTime t (t/zone-id)))
-      String         (recur (u.date/parse t))
-      LocalTime      (recur (.atDate ^LocalTime t (t/local-date)))
-      OffsetTime     (recur (.atDate ^OffsetTime t (t/local-date)))
-      LocalDate      (recur (.atStartOfDay ^LocalDate t))
+      ZonedDateTime (.toInstant ^ZonedDateTime t)
+      LocalDateTime (recur (.atZone ^LocalDateTime t (t/zone-id)))
+      String (recur (u.date/parse t))
+      LocalTime (recur (.atDate ^LocalTime t (t/local-date)))
+      OffsetTime (recur (.atDate ^OffsetTime t (t/local-date)))
+      LocalDate (recur (.atStartOfDay ^LocalDate t))
       (throw (ex-info (str "Cannot convert temporal " t " of type " (type t) " to an Instant")
                       {:temporal t})))))
 
@@ -214,7 +245,7 @@
   [run]
   (-> run
       (u/update-some :start_time utc-timestamp-string)
-      (u/update-some :end_time   utc-timestamp-string)))
+      (u/update-some :end_time utc-timestamp-string)))
 
 (mr/def ::column-definition
   [:map
@@ -258,14 +289,14 @@
   "Generate a temporary table name with current timestamp in milliseconds.
   If table name would exceed max table name length for the driver, fallback to using a shorter timestamp"
   [driver schema]
-  (let [max-len   (max 1 (or (driver/table-name-length-limit driver) Integer/MAX_VALUE))
+  (let [max-len (max 1 (or (driver/table-name-length-limit driver) Integer/MAX_VALUE))
         timestamp (str (System/currentTimeMillis))
-        prefix    (str transform-temp-table-prefix "_")
+        prefix (str transform-temp-table-prefix "_")
         available (- max-len (count prefix))
         ;; If we don't have enough space, take the later digits of the timestamp
-        suffix    (if (>= available (count timestamp))
-                    timestamp
-                    (subs timestamp (- (count timestamp) available)))
+        suffix (if (>= available (count timestamp))
+                 timestamp
+                 (subs timestamp (- (count timestamp) available)))
         table-name (str prefix suffix)]
     (keyword schema table-name)))
 

enterprise/backend/src/metabase_enterprise/transforms_python/execute.clj

@@ -139,22 +139,28 @@
           ret)))
     (thunk)))
 
-(defn- create-table-and-insert-data!
-  "Create a table from metadata and insert data from source."
-  [driver db-id table-name metadata data-source]
-  (let [table-schema {:name (if (keyword? table-name) table-name (keyword table-name))
-                      :columns (mapv (fn [{:keys [name base_type #_database_type]}]
-                                       {:name name
-                                        :type (python-runner/restricted-insert-type base_type)
-                                        ;; :database-type database_type
-                                        :nullable? true})
-                                     (:fields metadata))}
-        data-source (assoc data-source :table-schema table-schema)]
-    (transforms.util/create-table-from-schema! driver db-id table-schema)
+(defn- table-schema [table-name metadata]
+  {:name (if (keyword? table-name) table-name (keyword table-name))
+   :columns (mapv (fn [{:keys [name base_type #_database_type]}]
+                    {:name name
+                     :type (python-runner/restricted-insert-type base_type)
+                     ;; :database-type database_type
+                     :nullable? true})
+                  (:fields metadata))})
 
+(defn- insert-data!
+  "Insert data from source into an existing table."
+  [driver db-id table-schema data-source]
+  (let [data-source (assoc data-source :table-schema table-schema)]
     (->> #(driver/insert-from-source! driver db-id table-schema data-source)
          (maybe-retry-with-backoff driver))))
 
+(defn- create-table-and-insert-data!
+  "Create a table from metadata and insert data from source."
+  [driver db-id table-schema data-source]
+  (transforms.util/create-table-from-schema! driver db-id table-schema)
+  (insert-data! driver db-id table-schema data-source))
+
 (defn- transfer-with-rename-tables-strategy!
   "Transfer data using the rename-tables*! multimethod with atomicity guarantees.
    Creates new table, then atomically renames target->old and new->target, then drops old."
@@ -167,7 +173,7 @@
     (log/info "Using rename-tables strategy with atomicity guarantees")
     (try
 
-      (create-table-and-insert-data! driver db-id source-table-name metadata data-source)
+      (create-table-and-insert-data! driver db-id (table-schema source-table-name metadata) data-source)
       (transforms.util/rename-tables! driver db-id {table-name temp-table-name
                                                     source-table-name table-name})
       (transforms.util/drop-table! driver db-id temp-table-name)
@@ -187,7 +193,7 @@
     (log/info "Using create-drop-rename strategy to minimize downtime")
     (try
 
-      (create-table-and-insert-data! driver db-id source-table-name metadata data-source)
+      (create-table-and-insert-data! driver db-id (table-schema source-table-name metadata) data-source)
       (transforms.util/drop-table! driver db-id table-name)
       (driver/rename-table! driver db-id source-table-name table-name)
 
@@ -206,13 +212,35 @@
   (try
 
     (transforms.util/drop-table! driver db-id table-name)
-    (create-table-and-insert-data! driver db-id table-name metadata data-source)
+    (create-table-and-insert-data! driver db-id (table-schema table-name metadata) data-source)
 
     (catch Exception e
       (log/error e "Failed to transfer data using drop-create fallback strategy")
       (throw e))))
 
-(defn- transfer-file-to-db [driver {db-id :id :as db} {:keys [target] :as transform} metadata temp-file]
+(defmulti ^:private transfer-file-to-db
+  {:arglists '([driver db transform metadata temp-file])}
+  (fn [_ _ transform _ _] (-> transform :target :type keyword)))
+
+(defmethod transfer-file-to-db :table-incremental
+  [driver {db-id :id}
+   {:keys [target] :as transform}
+   metadata temp-file]
+  (let [table-name (transforms.util/qualified-table-name driver target)
+        table-exists? (transforms.util/target-table-exists? transform)
+        data-source {:type :jsonl-file
+                     :file temp-file}]
+    (if (not table-exists?)
+      (do
+        (log/info "New table")
+        (create-table-and-insert-data! driver db-id (table-schema table-name metadata) data-source))
+
+      (insert-data! driver db-id (table-schema table-name metadata) data-source))))
+
+(defmethod transfer-file-to-db :table
+  [driver {db-id :id :as db}
+   {:keys [target] :as transform}
+   metadata temp-file]
   (let [table-name (transforms.util/qualified-table-name driver target)
         table-exists? (transforms.util/target-table-exists? transform)
         data-source {:type :jsonl-file
@@ -221,7 +249,7 @@
       (not table-exists?)
       (do
         (log/info "New table")
-        (create-table-and-insert-data! driver db-id table-name metadata data-source))
+        (create-table-and-insert-data! driver db-id (table-schema table-name metadata) data-source))
 
       (driver.u/supports? driver :atomic-renames db)
       (transfer-with-rename-tables-strategy! driver db-id table-name metadata data-source)
@@ -333,6 +361,8 @@
                                 (save-log-to-transform-run-message! run-id message-log))
             ex-message-fn     #(exceptional-run-message message-log %)
             result            (transforms.util/run-cancelable-transform! run-id driver transform-details run-fn :ex-message-fn ex-message-fn)]
+        ;; Update watermark tracking for incremental transforms
+        (transforms.util/upsert-watermark! transform driver db)
         (transforms.instrumentation/with-stage-timing [run-id :table-sync]
           (transforms.util/sync-target! target db run-id))
         {:run_id run-id

enterprise/frontend/src/metabase-enterprise/transforms/components/QueryEditor/EditorHeader/EditorHeader.tsx

@@ -1,8 +1,9 @@
+import { useState } from "react";
 import { t } from "ttag";
 
 import Button from "metabase/common/components/Button";
 import EditBar from "metabase/common/components/EditBar";
-import { Tooltip } from "metabase/ui";
+import { Checkbox, Flex, Modal, Text, TextInput, Tooltip } from "metabase/ui";
 
 import type { QueryValidationResult } from "../types";
 
@@ -13,6 +14,10 @@ type EditorHeaderProps = {
   isSaving: boolean;
   onSave: () => void;
   onCancel: () => void;
+  transformId?: number;
+  isIncremental?: boolean;
+  onIncrementalChange?: (incremental: boolean, watermarkField?: string) => void;
+  watermarkField?: string | null;
 };
 
 export function EditorHeader({
@@ -22,20 +27,102 @@ export function EditorHeader({
   isSaving,
   onSave,
   onCancel,
+  transformId,
+  isIncremental = false,
+  onIncrementalChange,
+  watermarkField,
 }: EditorHeaderProps) {
+  const [localWatermarkField, setLocalWatermarkField] = useState(watermarkField || "id");
+
+  // Block saving if incremental is enabled but no watermark field is set
+  const hasWatermarkFieldIfNeeded = !isIncremental || localWatermarkField.trim() !== "";
   const canSave =
-    (isNew || isQueryDirty) && validationResult.isValid && !isSaving;
+    (isNew || isQueryDirty) &&
+    validationResult.isValid &&
+    !isSaving &&
+    hasWatermarkFieldIfNeeded;
+
+  const handleIncrementalToggle = (event: React.ChangeEvent<HTMLInputElement>) => {
+    const checked = event.currentTarget.checked;
+
+    if (isNew) {
+      // For new transforms, update the state and trigger save modal if checking
+      if (onIncrementalChange) {
+        onIncrementalChange(checked);
+      }
+      if (checked) {
+        // Show alert about needing to save first
+        alert(
+          t`You need to save the transform first before enabling incremental mode. The save dialog will open now.`,
+        );
+        // Trigger the save modal
+        onSave();
+      }
+    } else if (onIncrementalChange) {
+      // For existing transforms, toggle the incremental mode
+      if (checked && !localWatermarkField.trim()) {
+        // If enabling incremental but no watermark field, don't call the handler yet
+        // Just show an alert and don't actually toggle
+        alert(
+          t`Please enter a watermark field before enabling incremental mode.`,
+        );
+        // Prevent the checkbox from being checked
+        event.preventDefault();
+        return;
+      }
+      // Only call the handler if we have a watermark field (or if unchecking)
+      onIncrementalChange(checked, localWatermarkField);
+    }
+  };
+
+  const handleWatermarkFieldChange = (event: React.ChangeEvent<HTMLInputElement>) => {
+    const value = event.currentTarget.value;
+    setLocalWatermarkField(value);
+    if (onIncrementalChange && isIncremental) {
+      onIncrementalChange(true, value);
+    }
+  };
 
   return (
     <EditBar
-      title={getTitle(isNew)}
+      title={
+        <Flex align="center" gap="md">
+          {getTitle(isNew)}
+          <Checkbox
+            label={t`Incremental?`}
+            checked={isIncremental}
+            onChange={handleIncrementalToggle}
+          />
+          {isIncremental && !isNew && (
+            <TextInput
+              placeholder={t`Watermark field (e.g., id)`}
+              value={localWatermarkField}
+              onChange={handleWatermarkFieldChange}
+              size="xs"
+              w={200}
+            />
+          )}
+          {isIncremental && transformId && (
+            <Text size="sm" c="text-medium">
+              {t`Current Transform ID: ${transformId}`}
+            </Text>
+          )}
+        </Flex>
+      }
       admin
       buttons={[
         <Button key="cancel" small onClick={onCancel}>{t`Cancel`}</Button>,
         <Tooltip
           key="save"
-          label={validationResult.errorMessage}
-          disabled={validationResult.errorMessage == null}
+          label={
+            validationResult.errorMessage ||
+            (!hasWatermarkFieldIfNeeded
+              ? t`Watermark field is required for incremental transforms`
+              : undefined)
+          }
+          disabled={
+            validationResult.errorMessage == null && hasWatermarkFieldIfNeeded
+          }
         >
           <Button onClick={onSave} primary small disabled={!canSave}>
             {getSaveButtonLabel(isNew, isSaving)}
@@ -48,9 +135,9 @@ export function EditorHeader({
 
 function getTitle(isNew: boolean) {
   if (isNew) {
-    return t`You’re creating a new transform`;
+    return t`You're creating a new transform`;
   } else {
-    return t`You’re editing a transform`;
+    return t`You're editing a transform`;
   }
 }
 

enterprise/frontend/src/metabase-enterprise/transforms/components/QueryEditor/QueryEditor.tsx

@@ -40,6 +40,10 @@ type QueryEditorProps = {
   isSaving?: boolean;
   onSave: (source: QueryTransformSource) => void;
   onCancel: () => void;
+  transformId?: number;
+  isIncremental?: boolean;
+  onIncrementalChange?: (incremental: boolean, watermarkField?: string) => void;
+  watermarkField?: string | null;
 };
 
 export function QueryEditor({
@@ -48,6 +52,10 @@ export function QueryEditor({
   isSaving = false,
   onSave,
   onCancel,
+  transformId,
+  isIncremental,
+  onIncrementalChange,
+  watermarkField,
 }: QueryEditorProps) {
   const { question, isQueryDirty, setQuestion } = useQueryState(
     initialSource.query,
@@ -152,6 +160,10 @@ export function QueryEditor({
           isQueryDirty={isQueryDirty}
           onSave={handleSave}
           onCancel={onCancel}
+          transformId={transformId}
+          isIncremental={isIncremental}
+          onIncrementalChange={onIncrementalChange}
+          watermarkField={watermarkField}
         />
         <Flex h="100%" w="100%" mih="0">
           <Stack flex="2 1 100%" pos="relative">

enterprise/frontend/src/metabase-enterprise/transforms/pages/NewTransformPage/CreateTransformModal/CreateTransformModal.tsx

@@ -1,3 +1,4 @@
+import { useFormikContext } from "formik";
 import { useMemo } from "react";
 import { t } from "ttag";
 import * as Yup from "yup";
@@ -11,6 +12,7 @@ import {
 import { LoadingAndErrorWrapper } from "metabase/common/components/LoadingAndErrorWrapper";
 import {
   Form,
+  FormCheckbox,
   FormErrorMessage,
   FormProvider,
   FormSubmitButton,
@@ -32,12 +34,14 @@ type CreateTransformModalProps = {
   source: TransformSource;
   onCreate: (transform: Transform) => void;
   onClose: () => void;
+  initialIncremental?: boolean;
 };
 
 export function CreateTransformModal({
   source,
   onCreate,
   onClose,
+  initialIncremental = false,
 }: CreateTransformModalProps) {
   return (
     <Modal title={t`Save your transform`} opened padding="xl" onClose={onClose}>
@@ -46,6 +50,7 @@ export function CreateTransformModal({
         source={source}
         onCreate={onCreate}
         onClose={onClose}
+        initialIncremental={initialIncremental}
       />
     </Modal>
   );
@@ -55,6 +60,7 @@ type CreateTransformFormProps = {
   source: TransformSource;
   onCreate: (transform: Transform) => void;
   onClose: () => void;
+  initialIncremental: boolean;
 };
 
 type NewTransformValues = {
@@ -62,6 +68,8 @@ type NewTransformValues = {
   description: string | null;
   targetName: string;
   targetSchema: string | null;
+  incremental: boolean;
+  watermarkField: string | null;
 };
 
 const NEW_TRANSFORM_SCHEMA = Yup.object({
@@ -69,12 +77,36 @@ const NEW_TRANSFORM_SCHEMA = Yup.object({
   description: Yup.string().nullable(),
   targetName: Yup.string().required(Errors.required),
   targetSchema: Yup.string().nullable(),
+  incremental: Yup.boolean().required(),
+  watermarkField: Yup.string().nullable().when("incremental", {
+    is: true,
+    then: (schema) => schema.required(Errors.required),
+    otherwise: (schema) => schema.nullable(),
+  }),
 });
 
+function WatermarkFieldInput() {
+  const { values } = useFormikContext<NewTransformValues>();
+
+  if (!values.incremental) {
+    return null;
+  }
+
+  return (
+    <FormTextInput
+      name="watermarkField"
+      label={t`Watermark field`}
+      placeholder={t`e.g., id, row_num`}
+      description={t`An integer field used to track incremental updates`}
+    />
+  );
+}
+
 function CreateTransformForm({
   source,
   onCreate,
   onClose,
+  initialIncremental,
 }: CreateTransformFormProps) {
   const databaseId =
     source.type === "query" ? source.query.database : source["source-database"];
@@ -100,8 +132,8 @@ function CreateTransformForm({
   const supportsSchemas = database && hasFeature(database, "schemas");
 
   const initialValues: NewTransformValues = useMemo(
-    () => getInitialValues(schemas),
-    [schemas],
+    () => getInitialValues(schemas, initialIncremental),
+    [schemas, initialIncremental],
   );
 
   if (isLoading || error != null) {
@@ -152,6 +184,11 @@ function CreateTransformForm({
             label={t`Table name`}
             placeholder={t`descriptive_name`}
           />
+          <FormCheckbox
+            name="incremental"
+            label={t`Incremental?`}
+          />
+          <WatermarkFieldInput />
           <Group>
             <Box flex={1}>
               <FormErrorMessage />
@@ -165,18 +202,23 @@ function CreateTransformForm({
   );
 }
 
-function getInitialValues(schemas: string[]): NewTransformValues {
+function getInitialValues(
+  schemas: string[],
+  initialIncremental: boolean,
+): NewTransformValues {
   return {
     name: "",
     description: null,
     targetName: "",
     targetSchema: schemas?.[0] || null,
+    incremental: initialIncremental,
+    watermarkField: initialIncremental ? "id" : null,
   };
 }
 
 function getCreateRequest(
   source: TransformSource,
-  { name, description, targetName, targetSchema }: NewTransformValues,
+  { name, description, targetName, targetSchema, incremental, watermarkField }: NewTransformValues,
   databaseId: number,
 ): CreateTransformRequest {
   return {
@@ -184,10 +226,11 @@ function getCreateRequest(
     description,
     source,
     target: {
-      type: "table",
+      type: incremental ? "table-incremental" : "table",
       name: targetName,
       schema: targetSchema,
       database: databaseId,
+      watermarkField: incremental ? watermarkField : null,
     },
   };
 }

enterprise/frontend/src/metabase-enterprise/transforms/pages/NewTransformPage/NewTransformPage.tsx

@@ -15,7 +15,11 @@ import type {
 } from "metabase-types/api";
 
 import { QueryEditor } from "../../components/QueryEditor";
-import { getTransformListUrl, getTransformUrl } from "../../urls";
+import {
+  getTransformListUrl,
+  getTransformQueryUrl,
+  getTransformUrl,
+} from "../../urls";
 
 import { CreateTransformModal } from "./CreateTransformModal";
 import {
@@ -41,12 +45,18 @@ export function NewTransformPage({ params }: NewTransformPageProps) {
   const { type, cardId } = getParsedParams(params);
 
   const [source, setSource] = useState<TransformSource | null>(null);
+  const [wantsIncremental, setWantsIncremental] = useState(false);
   const [isModalOpened, { open: openModal, close: closeModal }] =
     useDisclosure();
   const dispatch = useDispatch();
 
   const handleCreate = (transform: Transform) => {
-    dispatch(push(getTransformUrl(transform.id)));
+    // If user wanted incremental, redirect to query edit page instead of transform detail page
+    if (wantsIncremental) {
+      dispatch(push(getTransformQueryUrl(transform.id)));
+    } else {
+      dispatch(push(getTransformUrl(transform.id)));
+    }
   };
 
   const handleSave = (newSource: TransformSource) => {
@@ -58,6 +68,12 @@ export function NewTransformPage({ params }: NewTransformPageProps) {
     dispatch(push(getTransformListUrl()));
   };
 
+  const handleModalClose = () => {
+    // If modal is closed without saving, reset the incremental checkbox
+    setWantsIncremental(false);
+    closeModal();
+  };
+
   return (
     <>
       <NewTransformEditorBody
@@ -65,12 +81,15 @@ export function NewTransformPage({ params }: NewTransformPageProps) {
         cardId={cardId}
         onSave={handleSave}
         onCancel={handleCancel}
+        wantsIncremental={wantsIncremental}
+        onWantsIncrementalChange={setWantsIncremental}
       />
       {isModalOpened && source !== null && (
         <CreateTransformModal
           source={source}
           onCreate={handleCreate}
-          onClose={closeModal}
+          onClose={handleModalClose}
+          initialIncremental={wantsIncremental}
         />
       )}
     </>
@@ -82,6 +101,8 @@ function NewTransformEditorBody(props: {
   cardId?: CardId;
   onSave: (source: TransformSource) => void;
   onCancel: () => void;
+  wantsIncremental: boolean;
+  onWantsIncrementalChange: (wants: boolean) => void;
 }) {
   const { type, ...rest } = props;
   if (type === "python") {
@@ -103,11 +124,15 @@ function NewQueryTransformEditorBody({
   cardId,
   onSave,
   onCancel,
+  wantsIncremental,
+  onWantsIncrementalChange,
 }: {
   type: LegacyDatasetQuery["type"];
   cardId?: CardId;
   onSave: (source: TransformSource) => void;
   onCancel: () => void;
+  wantsIncremental: boolean;
+  onWantsIncrementalChange: (wants: boolean) => void;
 }) {
   const {
     data: card,
@@ -119,12 +144,18 @@ function NewQueryTransformEditorBody({
     return <LoadingAndErrorWrapper loading={isLoading} error={error} />;
   }
 
+  const handleIncrementalChange = (incremental: boolean) => {
+    onWantsIncrementalChange(incremental);
+  };
+
   return (
     <QueryEditor
       initialSource={getInitialQueryTransformSource(card, type)}
       isNew
       onSave={onSave}
       onCancel={onCancel}
+      isIncremental={wantsIncremental}
+      onIncrementalChange={handleIncrementalChange}
     />
   );
 }

enterprise/frontend/src/metabase-enterprise/transforms/pages/TransformQueryPage/TransformQueryPage.tsx

@@ -88,6 +88,30 @@ export function TransformQueryPageBody({
     dispatch(push(getTransformUrl(transform.id)));
   };
 
+  const handleIncrementalChange = async (
+    incremental: boolean,
+    watermarkField?: string,
+  ) => {
+    const newType = incremental ? "table-incremental" : "table";
+    const { error } = await updateTransform({
+      id: transform.id,
+      target: {
+        ...transform.target,
+        type: newType,
+        watermarkField: incremental ? watermarkField : null,
+      },
+    });
+    if (error) {
+      sendErrorToast(t`Failed to update transform type`);
+    } else {
+      sendSuccessToast(
+        incremental
+          ? t`Transform set to incremental mode`
+          : t`Transform set to full refresh mode`,
+      );
+    }
+  };
+
   if (transform.source.type === "python") {
     return (
       <PLUGIN_TRANSFORMS_PYTHON.TransformEditor
@@ -107,6 +131,10 @@ export function TransformQueryPageBody({
         isSaving={isSaving || isCheckingDependencies}
         onSave={handleSaveSource}
         onCancel={handleCancel}
+        transformId={transform.id}
+        isIncremental={transform.target.type === "table-incremental"}
+        onIncrementalChange={handleIncrementalChange}
+        watermarkField={transform.target.watermarkField}
       />
       {isConfirmationShown && checkData != null && (
         <PLUGIN_DEPENDENCIES.CheckDependenciesModal

frontend/src/metabase-types/api/transform.ts

@@ -39,13 +39,14 @@ export type QueryTransformSource = {
 
 export type TransformSource = QueryTransformSource | PythonTransformSource;
 
-export type TransformTargetType = "table";
+export type TransformTargetType = "table" | "table-incremental";
 
 export type TransformTarget = {
   type: TransformTargetType;
   name: string;
   schema: string | null;
   database: number;
+  watermarkField?: string | null;
 };
 
 export type TransformRun = {

modules/drivers/bigquery-cloud-sdk/src/metabase/driver/bigquery_cloud_sdk.clj

@@ -870,6 +870,11 @@
   (let [table-str (get-table-str output-table)]
     [(format "CREATE OR REPLACE TABLE %s AS %s" table-str query)]))
 
+(defmethod driver/compile-insert :bigquery-cloud-sdk
+  [_driver {:keys [query output-table]}]
+  (let [table-str (get-table-str output-table)]
+    [(format "INSERT INTO %s %s" table-str query)]))
+
 (defmethod driver/compile-drop-table :bigquery-cloud-sdk
   [_driver table]
   (let [table-str (get-table-str table)]

modules/drivers/bigquery-cloud-sdk/test/metabase/driver/bigquery_cloud_sdk_test.clj

@@ -1346,3 +1346,14 @@
       :type/JSON               [[:raw "JSON"]]
       :type/SerializedJSON     [[:raw "JSON"]]
       :type/Decimal            [[:raw "BIGDECIMAL"]])))
+
+(deftest ^:parallel compile-transform-test
+  (mt/test-driver :bigquery-cloud-sdk
+    (testing "compile-transform creates CREATE OR REPLACE TABLE"
+      (is (= ["CREATE OR REPLACE TABLE `PRODUCTS_COPY` AS SELECT * FROM products"]
+             (driver/compile-transform :bigquery-cloud-sdk {:query "SELECT * FROM products"
+                                                            :output-table "PRODUCTS_COPY"}))))
+    (testing "compile-insert generates INSERT INTO"
+      (is (= ["INSERT INTO `PRODUCTS_COPY` SELECT * FROM products"]
+             (driver/compile-insert :bigquery-cloud-sdk {:query "SELECT * FROM products"
+                                                         :output-table "PRODUCTS_COPY"}))))))

modules/drivers/clickhouse/src/metabase/driver/clickhouse.clj

@@ -325,6 +325,10 @@
         query (str/join " " (map first pieces))]
     (into [query] (mapcat rest) pieces)))
 
+(defmethod driver/compile-insert :clickhouse
+  [driver {:keys [query output-table]}]
+  (sql.qp/format-honeysql driver {:insert-into [output-table {:raw query}]}))
+
 (defmethod driver/create-schema-if-needed! :clickhouse
   [driver conn-spec schema]
   (let [sql [[(format "CREATE DATABASE IF NOT EXISTS `%s`;" schema)]]]

modules/drivers/clickhouse/test/metabase/driver/clickhouse_test.clj

@@ -316,10 +316,14 @@
 
 (deftest ^:parallel compile-transform-test
   (mt/test-driver :clickhouse
-    (testing "compile transform for clickhouse with empty primary key column"
+    (testing "compile-transform for clickhouse with empty primary key column"
       (is (= ["CREATE TABLE `PRODUCTS_COPY` ORDER BY () AS SELECT * FROM products"]
              (driver/compile-transform :clickhouse {:query "SELECT * FROM products"
-                                                    :output-table "PRODUCTS_COPY"}))))))
+                                                    :output-table "PRODUCTS_COPY"}))))
+    (testing "compile-insert generates INSERT INTO"
+      (is (= ["INSERT INTO `PRODUCTS_COPY` SELECT * FROM products"]
+             (driver/compile-insert :clickhouse {:query "SELECT * FROM products"
+                                                 :output-table "PRODUCTS_COPY"}))))))
 
 (deftest ^:parallel clickhouse-db-supports-schemas-test
   (doseq [[schemas-supported? details] [[false? {}]

modules/drivers/sqlserver/src/metabase/driver/sqlserver.clj

@@ -1027,6 +1027,11 @@
     (.setIntoTables select-body [(Table. table-name)])
     [(str parsed-query)]))
 
+(defmethod driver/compile-insert :sqlserver
+  [driver {:keys [query output-table]}]
+  (let [^String table-name (first (sql.qp/format-honeysql driver (keyword output-table)))]
+    [(format "INSERT INTO %s %s" table-name query)]))
+
 (defmethod driver/table-exists? :sqlserver
   [driver database {:keys [schema name] :as _table}]
   (sql-jdbc.execute/do-with-connection-with-options

modules/drivers/sqlserver/test/metabase/driver/sqlserver_test.clj

@@ -771,3 +771,14 @@
       :type/Text               [:text]
       :type/Time               [:time]
       :type/UUID               [:uniqueidentifier])))
+
+(deftest ^:parallel compile-transform-test
+  (mt/test-driver :sqlserver
+    (testing "compile-transform creates SELECT INTO"
+      (is (= ["SELECT * INTO [PRODUCTS_COPY] FROM products"]
+             (driver/compile-transform :sqlserver {:query "SELECT * FROM products"
+                                                   :output-table "PRODUCTS_COPY"}))))
+    (testing "compile-insert generates INSERT INTO"
+      (is (= ["INSERT INTO [PRODUCTS_COPY] SELECT * FROM products"]
+             (driver/compile-insert :sqlserver {:query "SELECT * FROM products"
+                                                :output-table "PRODUCTS_COPY"}))))))

src/metabase/driver.clj

@@ -1151,7 +1151,13 @@
   :hierarchy #'hierarchy)
 
 (defmulti compile-transform
-  "Compiles the sql for a transform statement, given an inner sql query and a destination."
+  "Compiles the sql for a transform statement (CREATE TABLE AS), given an inner sql query and a destination."
+  {:added "0.57.0", :arglists '([driver {:keys [query output-table]}])}
+  dispatch-on-initialized-driver
+  :hierarchy #'hierarchy)
+
+(defmulti compile-insert
+  "Compiles the sql for an insert statement (INSERT INTO ... SELECT), given an inner sql query and a destination."
   {:added "0.57.0", :arglists '([driver {:keys [query output-table]}])}
   dispatch-on-initialized-driver
   :hierarchy #'hierarchy)

src/metabase/driver/sql.clj

@@ -127,6 +127,14 @@
                   overwrite? (cons (driver/compile-drop-table driver output-table)))]
     {:rows-affected (last (driver/execute-raw-queries! driver conn-spec queries))}))
 
+(defmethod driver/run-transform! [:sql :table-incremental]
+  [driver {:keys [conn-spec db-id output-table] :as transform-details} _opts] ;; eventually will have on conflict stuff, if supported
+  (let [queries (if (driver/table-exists? driver db-id {:schema (namespace output-table)
+                                                        :name (name output-table)})
+                  (driver/compile-insert driver transform-details)
+                  (driver/compile-transform driver transform-details))]
+    {:rows-affected (last (driver/execute-raw-queries! driver conn-spec [queries]))}))
+
 (defn qualified-name
   "Return the name of the target table of a transform as a possibly qualified symbol."
   [{schema :schema, table-name :name}]

src/metabase/driver/sql/query_processor.clj

@@ -2083,6 +2083,11 @@
                    {:create-table-as [(keyword output-table)]
                     :raw query}))
 
+(defmethod driver/compile-insert :sql
+  [driver {:keys [query output-table]}]
+  (format-honeysql driver
+                   {:insert-into [(keyword output-table) {:raw query}]}))
+
 (defmethod driver/compile-drop-table :sql
   [driver table]
   (format-honeysql driver {:drop-table [:if-exists (keyword table)]}))

test/metabase/driver/sql/transforms_test.clj

@@ -126,3 +126,32 @@
           ;; Drivers might quote these differently, but both parts should be present
           (is (or (re-find #"schema.*my_table" (first result))
                   (re-find #"my_table" (first result)))))))))
+
+(deftest compile-insert-test
+  (testing "compile-insert generates INSERT INTO statements"
+    (mt/test-drivers (mt/normal-drivers-with-feature :transforms/table)
+      (testing "simple table name"
+        (let [result (driver/compile-insert driver/*driver*
+                                            {:query "SELECT * FROM products"
+                                             :output-table :my_table})]
+          (testing "returns a vector"
+            (is (vector? result)))
+          (testing "first element is SQL string"
+            (is (string? (first result))))
+          (testing "generates INSERT INTO statement"
+            (is (re-find #"(?i)INSERT\s+INTO.*my_table" (first result))))
+          (testing "includes SELECT statement"
+            (is (re-find #"(?i)SELECT" (first result))))))
+
+      (testing "schema-qualified table"
+        (let [result (driver/compile-insert driver/*driver*
+                                            {:query "SELECT * FROM products"
+                                             :output-table :my_schema/my_table})]
+          (testing "returns a vector"
+            (is (vector? result)))
+          (testing "generates INSERT INTO statement"
+            (is (re-find #"(?i)INSERT\s+INTO" (first result))))
+          (testing "includes both schema and table parts"
+            (let [sql (first result)]
+              (is (re-find #"my_schema" sql) "Schema name should be present")
+              (is (re-find #"my_table" sql) "Table name should be present"))))))))