Native transform appdb checkpointing via template tag expansion (#70380)

* Tests mostly passing, some accidently!

* Remove next-checkpoint

* More tests / fixes to tests using inline VALUES

* Fix lossy date serialization

* 'Working FE'

* Fix issue with multi table

* Show prefix when multiple tables involved

* Test fixup

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

@@ -200,21 +200,28 @@
   (fn [_ _ transform _ _] (-> transform :target :type keyword)))
 
 (defmethod transfer-file-to-db :table-incremental
-  [driver {db-id :id}
+  [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
-                     :file temp-file}]
+  ;; First incremental run: no checkpoint exists yet, behave like non-incremental
+  ;; to drop and recreate the table rather than appending to existing data.
+  ;; Only applies to Python transforms - MBQL transforms handle this in query-impl.
+  (if (and (transforms.util/python-transform? transform)
+           (nil? (:last_checkpoint_type transform)))
+    ((get-method transfer-file-to-db :table) driver db transform metadata temp-file)
+    ;; Normal incremental: append if table exists, create if it doesn't
+    (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}]
 
-    ;; once we have more than just append, dispatch on :target-incremental-strategy
+      ;; once we have more than just append, dispatch on :target-incremental-strategy
 
-    (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))))
+      (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}
@@ -343,21 +350,24 @@
         (log/info "Executing Python transform" transform-id "with target" (pr-str target)
                   (when (driver.conn/write-connection-requested?)
                     " using write connection"))
-        (let [start-ms          (u/start-timer)
-              conn-spec         (driver/connection-spec driver db)
-              transform-details {:db-id          (:id db)
-                                 :transform-id   transform-id
-                                 :transform-type (keyword (:type target))
-                                 :conn-spec      conn-spec
-                                 :output-schema  (:schema target)
-                                 :output-table   (transforms.util/qualified-table-name driver target)}
-              run-fn            (fn [cancel-chan]
-                                  (run-python-transform! transform db run-id cancel-chan message-log)
-                                  (log! message-log (i18n/tru "Python execution finished successfully in {0}" (u.format/format-milliseconds (u/since-ms start-ms))))
-                                  (save-log-to-transform-run-message! run-id message-log))
-              ex-message-fn     #(exceptional-run-message message-log %)
-              result            (transforms.instrumentation/with-stage-timing [run-id [:computation :python-execution]]
-                                  (transforms.util/run-cancelable-transform! run-id driver transform-details run-fn :ex-message-fn ex-message-fn))]
+        (let [start-ms            (u/start-timer)
+              conn-spec           (driver/connection-spec driver db)
+              source-range-params (transforms.util/get-source-range-params transform)
+              transform-details   {:db-id          (:id db)
+                                   :transform-id   transform-id
+                                   :transform-type (keyword (:type target))
+                                   :conn-spec      conn-spec
+                                   :output-schema  (:schema target)
+                                   :output-table   (transforms.util/qualified-table-name driver target)}
+              run-fn              (fn [cancel-chan]
+                                    (run-python-transform! transform db run-id cancel-chan message-log)
+                                    (log! message-log (i18n/tru "Python execution finished successfully in {0}" (u.format/format-milliseconds (u/since-ms start-ms))))
+                                    (save-log-to-transform-run-message! run-id message-log)
+                                    (when source-range-params
+                                      (transforms.util/save-watermark! (:id transform) source-range-params)))
+              ex-message-fn       #(exceptional-run-message message-log %)
+              result              (transforms.instrumentation/with-stage-timing [run-id [:computation :python-execution]]
+                                    (transforms.util/run-cancelable-transform! run-id driver transform-details run-fn :ex-message-fn ex-message-fn))]
           (transforms.util/handle-transform-complete!
            :run-id run-id
            :transform transform

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

@@ -277,7 +277,7 @@
         transform         (t2/select-one :model/Transform transform-id)]
     (cond-> (-> (lib/query metadata-provider table-metadata)
                 lib/disable-default-limit)
-      source-incremental-strategy (transforms.u/preprocess-incremental-query source-incremental-strategy (transforms.u/next-checkpoint transform))
+      source-incremental-strategy (transforms.u/preprocess-incremental-query (transforms.u/get-source-range-params transform))
       limit                       (lib/limit limit))))
 
 ;; TODO break this up such that s3 can be swapped out for other transfer mechanisms.

enterprise/backend/test/metabase_enterprise/transforms/incremental_test.clj

@@ -1,7 +1,6 @@
 (ns ^:mb/driver-tests ^:mb/transforms-python-test metabase-enterprise.transforms.incremental-test
   "Tests for incremental transforms functionality."
   (:require
-   [clojure.set :as set]
    [clojure.string :as str]
    [clojure.test :refer :all]
    [honey.sql :as sql]
@@ -37,14 +36,14 @@
    :temporal {:field-name "created_at"
               :template-tag-type :text
               :lib-column-key "column-unique-key-v1$created_at"
-              :expected-initial-checkpoint "2024-01-10"
-              :expected-second-checkpoint "2024-01-16"}})
+              :expected-initial-checkpoint #t "2024-01-10T10:00:00Z"
+              :expected-second-checkpoint #t "2024-01-16T10:00:00Z"}})
 
 (defn- valid-checkpoint-transform-combo?
-  "Check if a checkpoint type and transform type combination is valid. "
-  [checkpoint-type transform-type]
-  (not (and (= checkpoint-type :temporal)
-            (= transform-type :native))))
+  "Check if a checkpoint type and transform type combination is valid.
+   With table tag expansion, all combinations are now supported."
+  [_checkpoint-type _transform-type]
+  true)
 
 (defn- build-table-query-with-order-by
   "Wraps the original build-table-query to add ORDER BY for deterministic test results."
@@ -77,27 +76,26 @@
       (transforms.execute/execute! transform run-opts))
     (transforms.execute/execute! transform run-opts)))
 
-(defn- make-incremental-source-query
-  "Create a native query with optional checkpoint template tag."
-  [schema checkpoint-config]
-  (let [{:keys [field-name template-tag-type]} checkpoint-config
+(defn- make-incremental-source-query-with-table-tag
+  "Create a native query with a table tag for testing subquery expansion."
+  [checkpoint-config]
+  (let [{:keys [field-name]} checkpoint-config
+        table-id (mt/id :transforms_products)
         timestamp-sql (first (sql/format (sql.qp/current-datetime-honeysql-form driver/*driver*)))
-        query (format "SELECT *, %s AS %s FROM %s [[WHERE %s > {{checkpoint}}]] ORDER BY %s LIMIT 10"
+        query (format "SELECT *, %s AS %s FROM {{source_table}} AS %s ORDER BY %s LIMIT 10"
                       timestamp-sql
                       (sql.u/quote-name driver/*driver* :field "load_timestamp")
-                      (if schema
-                        (sql.u/quote-name driver/*driver* :table schema "transforms_products")
-                        "transforms_products")
-                      (sql.u/quote-name driver/*driver* :field field-name)
+                      (sql.u/quote-name driver/*driver* :field "source_table")
                       (sql.u/quote-name driver/*driver* :field field-name))]
     {:database (mt/id)
      :type :native
      :native {:query query
-              :template-tags {"checkpoint" {:id "checkpoint"
-                                            :name "checkpoint"
-                                            :display-name "Checkpoint"
-                                            :type template-tag-type
-                                            :required false}}}}))
+              :template-tags {"source_table" {:id "source_table"
+                                              :name "source_table"
+                                              :display-name "Source Table"
+                                              :type "table"
+                                              :table-id table-id
+                                              :required true}}}}))
 
 (defn- make-incremental-source-query-without-template-tag
   "Create a native query without template tags for testing automatic checkpoint insertion. "
@@ -136,14 +134,16 @@
   checkpoint-config should be from checkpoint-configs"
   [transform-name target-table transform-type checkpoint-config]
   (let [schema (t2/select-one-fn :schema :model/Table (mt/id :transforms_products))
-        {:keys [field-name lib-column-key]} checkpoint-config]
+        {:keys [field-name lib-column-key]} checkpoint-config
+        checkpoint-filter-field (fn [] (t2/select-one-pk :model/Field :name field-name :table_id (mt/id :transforms_products)))]
     {:name transform-name
      :source_database_id (mt/id)
      :source (case transform-type
+               ;; Native queries now use table tags with checkpoint-filter-field-id
                :native {:type "query"
-                        :query (make-incremental-source-query schema checkpoint-config)
+                        :query (make-incremental-source-query-with-table-tag checkpoint-config)
                         :source-incremental-strategy {:type "checkpoint"
-                                                      :checkpoint-filter field-name}}
+                                                      :checkpoint-filter-field-id (checkpoint-filter-field)}}
                :native-auto-wrap {:type "query"
                                   :query (make-incremental-source-query-without-template-tag schema)
                                   :source-incremental-strategy {:type "checkpoint"
@@ -151,12 +151,14 @@
                :mbql {:type "query"
                       :query (make-incremental-mbql-query checkpoint-config)
                       :source-incremental-strategy {:type "checkpoint"
+                                                    :checkpoint-filter-field-id (checkpoint-filter-field)
                                                     :checkpoint-filter-unique-key lib-column-key}}
                :python {:type "python"
                         :source-tables {"transforms_products" (mt/id :transforms_products)}
                         :limit 10
                         :body incremental-python-body
                         :source-incremental-strategy {:type "checkpoint"
+                                                      :checkpoint-filter-field-id (checkpoint-filter-field)
                                                       :checkpoint-filter-unique-key lib-column-key}})
      :target (merge target-table {:type                        "table-incremental"
                                   :target-incremental-strategy {:type "append"}})}))
@@ -182,9 +184,11 @@
         result       (qp/process-query native-query)]
     (some-> result :data :rows first first bigint)))
 
-(defn get-checkpoint-value [transform]
-  (#'transforms.u/next-checkpoint-value
-   (transforms.u/next-checkpoint transform)))
+(defn- get-checkpoint-value [transform-id]
+  (let [transform (t2/select-one :model/Transform transform-id)
+        {:keys [last_checkpoint_type last_checkpoint_value]} transform]
+    (when last_checkpoint_type
+      (#'transforms.u/deserialize-checkpoint-value last_checkpoint_type last_checkpoint_value))))
 
 (defn- compare-checkpoint-values
   "Compare two checkpoint values with type-appropriate logic. "
@@ -193,8 +197,7 @@
     :integer (= (bigint expected) (bigint actual))
     :float (and (number? actual)
                 (< (Math/abs (- expected actual)) 0.01))
-    :temporal (and (string? actual)
-                   (str/starts-with? actual expected))))
+    :temporal (= expected actual)))
 
 (defn- insert-test-products!
   "Insert new products into the transforms_products table."
@@ -235,7 +238,7 @@
   (disj (mt/normal-drivers-with-feature :transforms/table) :redshift :clickhouse :sqlserver))
 
 (defn- target-table-gen [prefix]
-  {:type     "table"
+  {:type     :table
    :name     prefix
    :schema   (t2/select-one-fn :schema :model/Table (mt/id :transforms_products))
    :database (mt/id)})
@@ -251,6 +254,7 @@
               (with-transform-cleanup! [target-table (target-table-gen "incremental_test")]
                 (let [checkpoint-config (get checkpoint-configs checkpoint-type)
                       {:keys [field-name]} checkpoint-config
+                      expected-field-id (t2/select-one-pk :model/Field :name field-name :table_id (mt/id :transforms_products))
                       transform-payload (make-incremental-transform-payload "Test Incremental Transform" target-table :native checkpoint-config)]
                   (testing "Transform is created successfully"
                     (mt/with-temp [:model/Transform transform transform-payload]
@@ -258,10 +262,10 @@
                       (is (= "Test Incremental Transform" (:name transform)))
                       (is (= "table-incremental" (-> transform :target :type)))
                       (is (= "checkpoint" (-> transform :source :source-incremental-strategy :type)))
-                      (is (= field-name (-> transform :source :source-incremental-strategy :checkpoint-filter)))
+                      (is (= expected-field-id (-> transform :source :source-incremental-strategy :checkpoint-filter-field-id)))
 
                       (testing "No checkpoint exists initially"
-                        (is (nil? (get-checkpoint-value transform))))
+                        (is (nil? (get-checkpoint-value (:id transform)))))
 
                       (testing "Can retrieve transform via API"
                         (let [retrieved (mt/user-http-request :crowberto :get 200 (format "transform/%d" (:id transform)))]
@@ -297,19 +301,21 @@
                         (is (= 1 distinct-timestamps) "All rows should have the same timestamp from first run")
 
                         (testing "Checkpoint is created after first run"
-                          (let [checkpoint (get-checkpoint-value transform)]
+                          (let [checkpoint (get-checkpoint-value (:id transform))]
                             (is (compare-checkpoint-values checkpoint-type expected-initial-checkpoint checkpoint)
                                 (format "Checkpoint should be MAX(%s) from first 10 rows" (:field-name checkpoint-config)))))))
 
                     (testing "Second run should process the remaining rows"
-                      (execute-transform-with-ordering! transform transform-type (:field-name checkpoint-config) {:run-method :manual})
-                      (let [row-count (get-table-row-count target-table)
-                            distinct-timestamps (get-distinct-timestamp-count target-table)]
-                        (is (= 16 row-count) "Second run should add remaining 6 rows")
-                        (is (= 2 distinct-timestamps) "Should have 2 distinct timestamps (one per incremental run)")
-                        (let [checkpoint (get-checkpoint-value transform)]
-                          (is (compare-checkpoint-values checkpoint-type expected-second-checkpoint checkpoint)
-                              (format "Checkpoint should be MAX(%s) from all 16 rows" (:field-name checkpoint-config))))))))))))))))
+                      (let [transform (t2/select-one :model/Transform (:id transform))]
+                        (execute-transform-with-ordering! transform transform-type (:field-name checkpoint-config) {:run-method :manual})
+                        (let [row-count           (get-table-row-count target-table)
+                              distinct-timestamps (get-distinct-timestamp-count target-table)]
+                          (is (= 16 row-count) "Second run should add remaining 6 rows")
+                          (is (= 2 distinct-timestamps) "Should have 2 distinct timestamps (one per incremental run)")
+                          (let [checkpoint (or (:value (:lo (transforms.u/get-source-range-params (t2/select-one :model/Transform (:id transform)))))
+                                               (get-checkpoint-value (:id transform)))]
+                            (is (compare-checkpoint-values checkpoint-type expected-second-checkpoint checkpoint)
+                                (format "Checkpoint should be MAX(%s) from all 16 rows" (:field-name checkpoint-config)))))))))))))))))
 
 (deftest switch-incremental-to-non-incremental-test
   (testing "Switching an incremental transform to non-incremental overwrites data"
@@ -325,47 +331,48 @@
                       {:keys [expected-initial-checkpoint]} checkpoint-config
                       initial-payload (make-incremental-transform-payload "Switch Transform" target-table transform-type checkpoint-config)]
                   (mt/with-temp [:model/Transform transform initial-payload]
-                    (testing "Initial incremental run processes first batch"
-                      (execute-transform-with-ordering! transform transform-type (:field-name checkpoint-config) {:run-method :manual})
-                      (transforms.tu/wait-for-table (:name target-table) 10000)
-                      (let [row-count (get-table-row-count target-table)
-                            distinct-timestamps (get-distinct-timestamp-count target-table)
-                            checkpoint (get-checkpoint-value transform)]
-                        (is (= 10 row-count) "Initial run should process first 10 products")
-                        (is (= 1 distinct-timestamps) "All rows should have the same timestamp from first run")
-                        (is (compare-checkpoint-values checkpoint-type expected-initial-checkpoint checkpoint) "Checkpoint should be created")))
-
-                    (testing "Second incremental run processes remaining data"
-                      (execute-transform-with-ordering! transform transform-type (:field-name checkpoint-config) {:run-method :manual})
-                      (let [row-count (get-table-row-count target-table)
-                            distinct-timestamps (get-distinct-timestamp-count target-table)]
-                        (is (= 16 row-count) "Should have 16 rows after second incremental run")
-                        (is (= 2 distinct-timestamps) "Should have 2 distinct timestamps (one per incremental run)")))
-
-                    (testing "Switch to non-incremental via PUT API"
-                      (let [non-incremental-payload (-> initial-payload
-                                                        (update :source dissoc :source-incremental-strategy)
-                                                        (update :target dissoc :source-incremental-strategy)
-                                                        (update :target assoc :type "table"))
-                            updated (mt/user-http-request :crowberto :put 200 (format "transform/%d" (:id transform))
-                                                          non-incremental-payload)]
-                        (is (= "table" (-> updated :target :type)))
-                        (is (nil? (-> updated :source :source-incremental-strategy)))))
-
-                    (testing "Non-incremental run overwrites all data"
-                      (let [transform (t2/select-one :model/Transform (:id transform))]
+                    (let [get-transform #(t2/select-one :model/Transform (:id transform))]
+                      (testing "Initial incremental run processes first batch"
                         (execute-transform-with-ordering! transform transform-type (:field-name checkpoint-config) {:run-method :manual})
-                        (let [row-count (get-table-row-count target-table)
-                              distinct-timestamps (get-distinct-timestamp-count target-table)]
-                          (is (= 10 row-count) "Should overwrite to 10 rows")
-                          (is (= 1 distinct-timestamps) "All rows should have same timestamp after non-incremental overwrite"))
+                        (transforms.tu/wait-for-table (:name target-table) 10000)
+                        (let [row-count           (get-table-row-count target-table)
+                              distinct-timestamps (get-distinct-timestamp-count target-table)
+                              checkpoint          (get-checkpoint-value (:id transform))]
+                          (is (= 10 row-count) "Initial run should process first 10 products")
+                          (is (= 1 distinct-timestamps) "All rows should have the same timestamp from first run")
+                          (is (compare-checkpoint-values checkpoint-type expected-initial-checkpoint checkpoint) "Checkpoint should be created")))
 
-                        (testing "Running again still overwrites"
+                      (testing "Second incremental run processes remaining data"
+                        (execute-transform-with-ordering! (get-transform) transform-type (:field-name checkpoint-config) {:run-method :manual})
+                        (let [row-count           (get-table-row-count target-table)
+                              distinct-timestamps (get-distinct-timestamp-count target-table)]
+                          (is (= 16 row-count) "Should have 16 rows after second incremental run")
+                          (is (= 2 distinct-timestamps) "Should have 2 distinct timestamps (one per incremental run)")))
+
+                      (testing "Switch to non-incremental via PUT API"
+                        (let [non-incremental-payload (-> initial-payload
+                                                          (update :source dissoc :source-incremental-strategy)
+                                                          (update :target dissoc :source-incremental-strategy)
+                                                          (update :target assoc :type "table"))
+                              updated                 (mt/user-http-request :crowberto :put 200 (format "transform/%d" (:id transform))
+                                                                            non-incremental-payload)]
+                          (is (= "table" (-> updated :target :type)))
+                          (is (nil? (-> updated :source :source-incremental-strategy)))))
+
+                      (testing "Non-incremental run overwrites all data"
+                        (let [transform (t2/select-one :model/Transform (:id transform))]
                           (execute-transform-with-ordering! transform transform-type (:field-name checkpoint-config) {:run-method :manual})
-                          (let [row-count (get-table-row-count target-table)
+                          (let [row-count           (get-table-row-count target-table)
                                 distinct-timestamps (get-distinct-timestamp-count target-table)]
-                            (is (= 10 row-count) "Should still have 10 rows after another run")
-                            (is (= 1 distinct-timestamps) "Should still have 1 distinct timestamp")))))))))))))))
+                            (is (= 10 row-count) "Should overwrite to 10 rows")
+                            (is (= 1 distinct-timestamps) "All rows should have same timestamp after non-incremental overwrite"))
+
+                          (testing "Running again still overwrites"
+                            (execute-transform-with-ordering! transform transform-type (:field-name checkpoint-config) {:run-method :manual})
+                            (let [row-count           (get-table-row-count target-table)
+                                  distinct-timestamps (get-distinct-timestamp-count target-table)]
+                              (is (= 10 row-count) "Should still have 10 rows after another run")
+                              (is (= 1 distinct-timestamps) "Should still have 1 distinct timestamp"))))))))))))))))
 
 (deftest switch-non-incremental-to-incremental-test
   (testing "Switching a non-incremental transform to incremental computes checkpoint from existing data"
@@ -378,7 +385,8 @@
             (mt/dataset transforms-dataset/transforms-test
               (with-transform-cleanup! [target-table (target-table-gen "switch_non_incr_to_incr")]
                 (let [checkpoint-config (get checkpoint-configs checkpoint-type)
-                      {:keys [expected-second-checkpoint]} checkpoint-config
+                      {:keys [expected-initial-checkpoint
+                              expected-second-checkpoint]} checkpoint-config
                       incremental-payload (make-incremental-transform-payload "Switch Transform" target-table transform-type checkpoint-config)
                       initial-payload (-> incremental-payload
                                           (update :source dissoc :source-incremental-strategy)
@@ -392,9 +400,8 @@
                             distinct-timestamps (get-distinct-timestamp-count target-table)]
                         (is (= 10 row-count) "Initial run should process 10 products")
                         (is (= 1 distinct-timestamps) "All rows should have same timestamp from non-incremental run")
-
                         (testing "No checkpoint exists"
-                          (let [checkpoint (get-checkpoint-value transform)]
+                          (let [checkpoint (get-checkpoint-value (:id transform))]
                             (is (nil? checkpoint) "No checkpoint for non-incremental transform")))))
 
                     (testing "Switch to incremental via PUT API"
@@ -406,9 +413,19 @@
                     (testing "First incremental run after switch processes no new data"
                       (let [transform (t2/select-one :model/Transform (:id transform))]
                         (execute-transform-with-ordering! transform transform-type (:field-name checkpoint-config) {:run-method :manual})
-                        (let [row-count (get-table-row-count target-table)
+                        (let [row-count           (get-table-row-count target-table)
                               distinct-timestamps (get-distinct-timestamp-count target-table)
-                              checkpoint (get-checkpoint-value transform)]
+                              checkpoint          (get-checkpoint-value (:id transform))]
+                          (is (= 10 row-count))
+                          (is (= 1 distinct-timestamps))
+                          (is (compare-checkpoint-values checkpoint-type expected-initial-checkpoint checkpoint) "Checkpoint should be computed from existing data"))))
+
+                    (testing "Second incremental run process remaining rows"
+                      (let [transform (t2/select-one :model/Transform (:id transform))]
+                        (execute-transform-with-ordering! transform transform-type (:field-name checkpoint-config) {:run-method :manual})
+                        (let [row-count           (get-table-row-count target-table)
+                              distinct-timestamps (get-distinct-timestamp-count target-table)
+                              checkpoint          (get-checkpoint-value (:id transform))]
                           (is (= 16 row-count) "Should process remaining 6 entries")
                           (is (= 2 distinct-timestamps) "Should have 2 distinct timestamp")
                           (is (compare-checkpoint-values checkpoint-type expected-second-checkpoint checkpoint) "Checkpoint should be computed from existing data"))))
@@ -426,67 +443,12 @@
                           (let [transform (t2/select-one :model/Transform (:id transform))]
                             (execute-transform-with-ordering! transform transform-type (:field-name checkpoint-config) {:run-method :manual})
                             (let [row-count (get-table-row-count target-table)
-                                  checkpoint (get-checkpoint-value transform)]
+                                  checkpoint (get-checkpoint-value (:id transform))]
                               (is (= 17 row-count) "Should append 1 new row (16 + 1 = 17)")
                               ;; For integer checkpoints, we can verify exact value >= 17
                               ;; For float/temporal, just verify checkpoint exists
                               (is (some? checkpoint) "Checkpoint should be updated"))))))))))))))))
 
-(deftest native-query-without-template-tag-test
-  (testing "Native query without template tags uses automatic checkpoint wrapping"
-    (doseq [checkpoint-type [:integer :float :temporal]]
-      (testing (format "with %s checkpoint" (name checkpoint-type))
-        (mt/test-drivers (test-drivers)
-          (mt/with-premium-features #{:transforms}
-            (mt/dataset transforms-dataset/transforms-test
-              (with-transform-cleanup! [target-table (target-table-gen "native_no_template")]
-                (let [checkpoint-config (get checkpoint-configs checkpoint-type)
-                      {:keys [expected-second-checkpoint field-name]} checkpoint-config
-                      schema (t2/select-one-fn :schema :model/Table (mt/id :transforms_products))
-                      transform-payload {:name "Native Without Template Tag"
-                                         :source {:type "query"
-                                                  :query (make-incremental-source-query-without-template-tag schema)
-                                                  :source-incremental-strategy {:type "checkpoint"
-                                                                                :checkpoint-filter field-name}}
-                                         :target (merge target-table
-                                                        {:type                        "table-incremental"
-                                                         :target-incremental-strategy {:type "append"}})}]
-                  (mt/with-temp [:model/Transform transform transform-payload]
-                    (testing "First run processes all existing data"
-                      (transforms.execute/execute! transform {:run-method :manual})
-                      (transforms.tu/wait-for-table (:name target-table) 10000)
-                      (let [row-count  (get-table-row-count target-table)
-                            checkpoint (get-checkpoint-value transform)]
-                        (is (= 16 row-count) "First run should process all 16 products")
-                        (is (compare-checkpoint-values checkpoint-type expected-second-checkpoint checkpoint)
-                            (format "Checkpoint should be MAX(%s) = %s" (:field-name checkpoint-config) expected-second-checkpoint))))
-
-                    (testing "Second run without new data adds nothing"
-                      (transforms.execute/execute! transform {:run-method :manual})
-                      (let [row-count (get-table-row-count target-table)]
-                        (is (= 16 row-count) "Should still have 16 rows, no new data")))
-
-                    (when (and (isa? driver/hierarchy driver/*driver* :sql-jdbc) ; insert/delete test products only works for jdbc drivers at the moment
-                               (not= driver/*driver* :clickhouse)
-                               ;; this *should* work see #68965 for context, will plan follow-up task
-                               (not= driver/*driver* :snowflake))
-                      (testing "After inserting new data, incremental run appends only new rows"
-                        (with-insert-test-products!
-                          [{:name "New Product 1"
-                            :category "Electronics"
-                            :price 299.99
-                            :created-at "2024-01-21T10:00:00"}
-                           {:name "New Product 2"
-                            :category "Books"
-                            :price 319.99
-                            :created-at "2024-01-21T11:00:00"}]
-
-                          (transforms.execute/execute! transform {:run-method :manual})
-                          (let [row-count (get-table-row-count target-table)
-                                checkpoint (get-checkpoint-value transform)]
-                            (is (= 18 row-count) "Should append 2 new rows (16 + 2 = 18)")
-                            (is (some? checkpoint) "Checkpoint should be updated")))))))))))))))
-
 (deftest unsupported-checkpoint-column-type-test
   (testing "Transform fails at runtime with unsupported checkpoint column type"
     (mt/test-drivers #{:h2 :postgres}
@@ -510,266 +472,221 @@
                   (is (string? response))
                   (is (re-find #"unsupported type" response)))))))))))
 
-(deftest ^:postgres-only native-query-with-temporal-checkpoint-test
-  (testing "Native query with temporal checkpoint"
-    ;; we test only in postgres because it's easy to cast to ::timestamp
-    (mt/test-drivers [:postgres]
-      (mt/with-premium-features #{:transforms}
-        (mt/dataset transforms-dataset/transforms-test
-          (with-transform-cleanup! [target-table (target-table-gen "native_temporal_cast")]
-            (let [checkpoint-config (get checkpoint-configs :temporal)
-                  {:keys [expected-initial-checkpoint expected-second-checkpoint field-name]} checkpoint-config
-                  schema (t2/select-one-fn :schema :model/Table (mt/id :transforms_products))
-                  timestamp-sql (first (sql/format (sql.qp/current-datetime-honeysql-form driver/*driver*)))
-                  query (format "SELECT *, %s AS load_timestamp FROM %s [[WHERE %s > {{checkpoint}}::timestamp]] ORDER BY %s LIMIT 10"
-                                timestamp-sql
-                                (if schema
-                                  (sql.u/quote-name driver/*driver* :table schema "transforms_products")
-                                  "transforms_products")
-                                field-name
-                                field-name)
-                  transform-payload {:name "Native With Temporal Cast"
-                                     :source {:type "query"
-                                              :query {:database (mt/id)
-                                                      :type :native
-                                                      :native {:query query
-                                                               :template-tags {"checkpoint" {:id "checkpoint"
-                                                                                             :name "checkpoint"
-                                                                                             :display-name "Checkpoint"
-                                                                                             :type :text
-                                                                                             :required false}}}}
-                                              :source-incremental-strategy {:type "checkpoint"
-                                                                            :checkpoint-filter field-name}}
-                                     :target (merge target-table
-                                                    {:type                        "table-incremental"
-                                                     :target-incremental-strategy {:type "append"}})}]
-              (mt/with-temp [:model/Transform transform transform-payload]
-                (testing "First run processes first batch"
-                  (transforms.execute/execute! transform {:run-method :manual})
-                  (transforms.tu/wait-for-table (:name target-table) 10000)
-                  (let [row-count           (get-table-row-count target-table)
-                        distinct-timestamps (get-distinct-timestamp-count target-table)
-                        checkpoint          (get-checkpoint-value transform)]
-                    (is (= 10 row-count) "First run should process the first 10 products")
-                    (is (= 1 distinct-timestamps) "All rows should have the same timestamp from first run")
-                    (is (compare-checkpoint-values :temporal expected-initial-checkpoint checkpoint)
-                        (format "Checkpoint should be MAX(%s) from first 10 rows" field-name))))
-
-                (testing "Second run processes remaining data"
-                  (transforms.execute/execute! transform {:run-method :manual})
-                  (let [row-count           (get-table-row-count target-table)
-                        distinct-timestamps (get-distinct-timestamp-count target-table)
-                        checkpoint (get-checkpoint-value transform)]
-                    (is (= 16 row-count) "Second run should add remaining 6 rows")
-                    (is (= 2 distinct-timestamps) "Should have 2 distinct timestamps (one per incremental run)")
-                    (is (compare-checkpoint-values :temporal expected-second-checkpoint checkpoint)
-                        (format "Checkpoint should be MAX(%s) from all 16 rows" field-name))))
-
-                (testing "Third run without new data adds nothing"
-                  (transforms.execute/execute! transform {:run-method :manual})
-                  (let [row-count (get-table-row-count target-table)]
-                    (is (= 16 row-count) "Should still have 16 rows, no new data")))
-
-                (testing "After inserting new data, incremental run appends only new rows"
-                  (with-insert-test-products!
-                    [{:name "New Temporal Product"
-                      :category "Electronics"
-                      :price 299.99
-                      :created-at "2024-01-21T10:00:00"}]
-
-                    (transforms.execute/execute! transform {:run-method :manual})
-                    (let [row-count (get-table-row-count target-table)
-                          checkpoint (get-checkpoint-value transform)]
-                      (is (= 17 row-count) "Should append 1 new row (16 + 1 = 17)")
-                      (is (some? checkpoint) "Checkpoint should be updated"))))))))))))
-
 (defn- pg-table-rows [db-spec table-name]
   (next.jdbc/execute! db-spec [(format "SELECT * FROM %s" table-name)] {:builder-fn jdbc.rs/as-unqualified-lower-maps}))
 
 (deftest empty-table-test
-  (mt/test-drivers #{:postgres}                             ; no db specifics
+  (mt/test-drivers #{:postgres}
     (mt/with-premium-features #{:transforms}
-      (with-transform-cleanup! [target-table "empty_table_target"]
-        (let [db-id   (mt/id)
-              db-spec (sql-jdbc.conn/db->pooled-connection-spec db-id)
-              source  {:type                        "query"
-                       :query                       {:database db-id
-                                                     :type     :native
-                                                     :native   {:query "SELECT * FROM (VALUES (42)) x(id) WHERE 1 = 2"}}
-                       :source-incremental-strategy {:type "checkpoint", :checkpoint-filter "id"}}
-              target  {:type                        "table-incremental"
-                       :schema                      "public"
-                       :name                        target-table
-                       :database                    db-id
-                       :target-incremental-strategy {:type "append"}}]
-          (mt/with-temp [:model/Transform transform {:name "test transform" :source source, :target target}]
-            (transforms.execute/execute! transform {:run-method :manual})
-            (testing "still creates target table"
-              (is (= 1 (count (next.jdbc/execute! db-spec ["SELECT true FROM information_schema.tables WHERE table_name = ?" target-table])))))
-            (testing "sync has picked up table"
-              (is (=? {:name target-table, :fields [{:name "id"}]} (-> (t2/select-one :model/Table :name target-table) (t2/hydrate :fields)))))
-            (testing "checkpoint is recognized"
-              (is (some? (transforms.u/next-checkpoint transform))))))))))
+      (mt/dataset transforms-dataset/transforms-test
+        (with-transform-cleanup! [target-table "empty_table_target"]
+          (let [db-id   (mt/id)
+                db-spec (sql-jdbc.conn/db->pooled-connection-spec db-id)
+                table-id (mt/id :transforms_products)
+                checkpoint-field-id (t2/select-one-pk :model/Field :name "id" :table_id table-id)
+                ;; Query that returns no rows but has the checkpoint column
+                source  {:type                        "query"
+                         :query                       {:database db-id
+                                                       :type     :native
+                                                       :native   {:query "SELECT * FROM {{source_table}} AS s WHERE 1 = 2"
+                                                                  :template-tags {"source_table" {:id "source_table"
+                                                                                                  :name "source_table"
+                                                                                                  :display-name "Source Table"
+                                                                                                  :type "table"
+                                                                                                  :table-id table-id
+                                                                                                  :required true}}}}
+                         :source-incremental-strategy {:type "checkpoint"
+                                                       :checkpoint-filter-field-id checkpoint-field-id}}
+                target  {:type                        "table-incremental"
+                         :schema                      "public"
+                         :name                        target-table
+                         :database                    db-id
+                         :target-incremental-strategy {:type "append"}}]
+            (mt/with-temp [:model/Transform transform {:name "test transform" :source source, :target target}]
+              (transforms.execute/execute! transform {:run-method :manual})
+              (testing "still creates target table"
+                (is (= 1 (count (next.jdbc/execute! db-spec ["SELECT true FROM information_schema.tables WHERE table_name = ?" target-table])))))
+              (testing "sync has picked up table"
+                (is (=? {:name target-table} (t2/select-one :model/Table :name target-table))))
+              (testing "no checkpoint (as no MAX) empty result"
+                (let [transform (t2/select-one :model/Transform (:id transform))]
+                  (is (nil? (:last_checkpoint_type transform))))))))))))
 
 (deftest checkpoint-field-does-not-exist-test
-  (mt/test-drivers #{:postgres}                             ; no db specifics
-    (mt/with-premium-features #{:transforms}
-      (with-transform-cleanup! [target-table "missing_field_target"]
-        (let [db-id   (mt/id)
-              db-spec (sql-jdbc.conn/db->pooled-connection-spec db-id)
-              source  {:type                        "query"
-                       :query                       {:database db-id
-                                                     :type     :native
-                                                     :native   {:query "SELECT * FROM (VALUES (42)) x(id)"}}
-                       :source-incremental-strategy {:type "checkpoint", :checkpoint-filter "no_such_column"}}
-              target  {:type                        "table-incremental"
-                       :schema                      "public"
-                       :name                        target-table
-                       :database                    db-id
-                       :target-incremental-strategy {:type "append"}}]
-          (mt/with-temp [:model/Transform transform {:name "test transform" :source source, :target target}]
-            (transforms.execute/execute! transform {:run-method :manual})
-            (testing "still creates target table"
-              (is (= 1 (count (next.jdbc/execute! db-spec ["SELECT true FROM information_schema.tables WHERE table_name = ?" target-table])))))
-            (testing "sync has picked up table"
-              (is (=? {:name target-table, :fields [{:name "id"}]} (-> (t2/select-one :model/Table :name target-table) (t2/hydrate :fields)))))
-            (testing "target table has expected data"
-              (is (= [{:id 42}] (pg-table-rows db-spec target-table))))
-            (testing "checkpoint is not recognized, so transform acts as if no checkpoint"
-              (is (nil? (transforms.u/next-checkpoint transform))))
-            ;; Maybe this is unrealistic - you cannot select a column that does not exist
-            ;; But the source tables schema can change, e.g. rename: you change event_time to event_ts or something
-            ;; so one would have to be careful to disable or delete transforms ahead of a schema change like this
-            ;; For now asserting behaviour is-what-it-is, but this should provoke a hmmm... maybe it is better to fail early?
-            (testing "running a second time will duplicate the existing data"
-              (transforms.execute/execute! transform {:run-method :manual})
-              (is (= [{:id 42} {:id 42}] (pg-table-rows db-spec target-table))))))))))
-
-(deftest changing-query-keeps-checkpoint-test
-  (mt/test-drivers #{:postgres}                             ; no db specifics
-    (mt/with-premium-features #{:transforms}
-      (with-transform-cleanup! [target-table "change_table_target"]
-        (let [db-id   (mt/id)
-              db-spec (sql-jdbc.conn/db->pooled-connection-spec db-id)
-              source  {:type                        "query"
-                       :query                       {:database db-id
-                                                     :type     :native
-                                                     :native   {:query "SELECT * FROM (VALUES (42)) x(id)"}}
-                       :source-incremental-strategy {:type "checkpoint", :checkpoint-filter "id"}}
-              target  {:type                        "table-incremental"
-                       :schema                      "public"
-                       :name                        target-table
-                       :database                    db-id
-                       :target-incremental-strategy {:type "append"}}]
-          (mt/with-temp [:model/Transform transform {:name "test transform" :source source, :target target}]
-            (testing "initial run"
-              (transforms.execute/execute! transform {:run-method :manual})
-              (is (= [{:id 42}] (pg-table-rows db-spec target-table))))
-            (testing "initial increment"
-              (transforms.execute/execute! transform {:run-method :manual})
-              (is (= [{:id 42}] (pg-table-rows db-spec target-table))))
-            (t2/update! :model/Transform
-                        (:id transform)
-                        {:source {:type "query"
-                                  :query {:database db-id
-                                          :type     :native
-                                          :native   {:query "SELECT * FROM (VALUES (42), (43)) x(id)"}}
-                                  :source-incremental-strategy {:type "checkpoint", :checkpoint-filter "id"}}})
-            (testing "second increment"
-              (transforms.execute/execute! (t2/select-one :model/Transform (:id transform)) {:run-method :manual})
-              (is (= [{:id 42} {:id 43}] (pg-table-rows db-spec target-table))))))))))
-
-(deftest filter-column-indexed-test
-  (testing "Filter column is indexed"
-    (doseq [checkpoint-type [:integer]
-            transform-type  [:native :mbql :python]
-            :when (valid-checkpoint-transform-combo? checkpoint-type transform-type)]
-      (testing (format "with %s checkpoint on %s transform" (name checkpoint-type) (name transform-type))
-        (mt/test-drivers (set/intersection (test-drivers) (mt/normal-drivers-with-feature :transforms/index-ddl))
-          (mt/with-premium-features #{:transforms :transforms-python}
-            (mt/dataset transforms-dataset/transforms-test
-              (with-transform-cleanup! [target-table (target-table-gen "incremental_index")]
-                (let [checkpoint-config (get checkpoint-configs checkpoint-type)
-                      transform-payload (make-incremental-transform-payload "Incremental Transform" target-table transform-type checkpoint-config)]
-                  (mt/with-temp [:model/Transform transform transform-payload]
-                    (testing "First run creates index on checkpoint column"
-                      (execute-transform-with-ordering! transform transform-type (:field-name checkpoint-config) {:run-method :manual})
-                      (let [indexes    (driver/describe-table-indexes driver/*driver* (mt/id) target-table)
-                            field-name (:field-name checkpoint-config)]
-                        (testing "Index was created"
-                          (is (= 1 (count indexes)))
-                          (is (=? {:value field-name :index-name #"^mb_transform_idx_.*$"} (first indexes))))))
-                    (testing "Data was processed correctly"
-                      (is (= 10 (get-table-row-count target-table))))
-                    (testing "Second run succeeds with existing index"
-                      (execute-transform-with-ordering! transform transform-type (:field-name checkpoint-config) {:run-method :manual})
-                      (let [indexes    (driver/describe-table-indexes driver/*driver* (mt/id) target-table)
-                            field-name (:field-name checkpoint-config)]
-                        (testing "Index still exists"
-                          (is (= 1 (count indexes)))
-                          (is (=? {:value field-name :index-name #"^mb_transform_idx_.*$"} (first indexes)))))
-                      (testing "Data was processed correctly"
-                        (is (= 16 (get-table-row-count target-table)))))))))))))))
-
-(deftest index-cleanup-on-switch-to-non-incremental-test
-  (testing "Switching to non-incremental removes metabase-owned indexes"
-    (doseq [checkpoint-type [:integer]
-            transform-type  [:native :mbql :python]
-            :when (valid-checkpoint-transform-combo? checkpoint-type transform-type)]
-      (testing (format "with %s checkpoint on %s transform" (name checkpoint-type) (name transform-type))
-        (mt/test-drivers (set/intersection (test-drivers) (mt/normal-drivers-with-feature :transforms/index-ddl))
-          (mt/with-premium-features #{:transforms :transforms-python}
-            (mt/dataset transforms-dataset/transforms-test
-              (with-transform-cleanup! [target-table (target-table-gen "index_cleanup_non_incr")]
-                (let [checkpoint-config   (get checkpoint-configs checkpoint-type)
-                      {:keys [field-name]} checkpoint-config
-                      incremental-payload (make-incremental-transform-payload "Index Cleanup Transform" target-table transform-type checkpoint-config)]
-                  (mt/with-temp [:model/Transform transform incremental-payload]
-                    (testing "First incremental run creates index"
-                      (execute-transform-with-ordering! transform transform-type field-name {:run-method :manual})
-                      (let [indexes (driver/describe-table-indexes driver/*driver* (mt/id) target-table)]
-                        (is (=? {:value field-name :index-name #"^mb_transform_idx_.*$"} (first indexes)))))
-                    (testing "Switch to non-incremental via API"
-                      (let [non-incremental-payload (-> incremental-payload
-                                                        (update :source dissoc :source-incremental-strategy))
-                            updated                 (mt/user-http-request :crowberto :put 200 (format "transform/%d" (:id transform))
-                                                                          non-incremental-payload)]
-                        (is (nil? (:source-incremental-strategy (:source updated))))))
-                    (testing "Non-incremental run removes automatic indexes indexes"
-                      (let [transform (t2/select-one :model/Transform (:id transform))]
-                        (execute-transform-with-ordering! transform transform-type field-name {:run-method :manual})
-                        (let [indexes    (driver/describe-table-indexes driver/*driver* (mt/id) target-table)
-                              mb-indexes (filter #(str/starts-with? (:index-name %) "mb_transform_idx_") indexes)]
-                          (testing "Automatic indexes are dropped"
-                            (is (empty? mb-indexes))))))))))))))))
-
-(deftest index-cleanup-on-checkpoint-column-change-test
-  (testing "Changing checkpoint column removes old index and creates new one"
-    (mt/test-drivers (set/intersection (test-drivers) (mt/normal-drivers-with-feature :transforms/index-ddl))
+  (testing "Transform with non-existent checkpoint-filter-field-id fails gracefully"
+    (mt/test-drivers #{:postgres}
       (mt/with-premium-features #{:transforms}
         (mt/dataset transforms-dataset/transforms-test
-          (with-transform-cleanup! [target-table (target-table-gen "icchange")]
-            (let [integer-config  (get checkpoint-configs :integer)
-                  float-config    (get checkpoint-configs :float)
-                  initial-field   (:field-name integer-config)
-                  new-field       (:field-name float-config)
-                  transform-type  :native
-                  initial-payload (make-incremental-transform-payload "Column Change Transform" target-table transform-type integer-config)]
-              (mt/with-temp [:model/Transform transform initial-payload]
-                (testing "First run with integer checkpoint creates index on id column"
-                  (execute-transform-with-ordering! transform transform-type initial-field {:run-method :manual})
-                  (let [indexes (driver/describe-table-indexes driver/*driver* (mt/id) target-table)]
-                    (is (= 1 (count indexes)))
-                    (is (=? {:value initial-field :index-name #"^mb_transform_idx_.*$"} (first indexes)))))
-                (testing "Switch checkpoint column via API"
-                  (let [new-payload (make-incremental-transform-payload "Column Change Transform" target-table transform-type float-config)
-                        updated     (mt/user-http-request :crowberto :put 200 (format "transform/%d" (:id transform))
-                                                          new-payload)]
-                    (is (= new-field (-> updated :source :source-incremental-strategy :checkpoint-filter)))))
-                (testing "Run with new checkpoint column updates indexes"
-                  (let [transform (t2/select-one :model/Transform (:id transform))]
-                    (execute-transform-with-ordering! transform transform-type new-field {:run-method :manual})
-                    (let [indexes            (driver/describe-table-indexes driver/*driver* (mt/id) target-table)]
-                      (testing "Old index removed, new checkpoint column is now indexed"
-                        (is (= 1 (count indexes)))
-                        (is (=? {:value new-field :index-name #"^mb_transform_idx_.*$"} (first indexes)))))))))))))))
+          (with-transform-cleanup! [target-table "missing_field_target"]
+            (let [db-id   (mt/id)
+                  table-id (mt/id :transforms_products)
+                  invalid-field-id 999999
+                  source  {:type                        "query"
+                           :query                       {:database db-id
+                                                         :type     :native
+                                                         :native   {:query "SELECT id FROM {{source_table}}"
+                                                                    :template-tags {"source_table" {:id "source_table"
+                                                                                                    :name "source_table"
+                                                                                                    :display-name "Source Table"
+                                                                                                    :type "table"
+                                                                                                    :table-id table-id
+                                                                                                    :required true}}}}
+                           :source-incremental-strategy {:type "checkpoint"
+                                                         :checkpoint-filter-field-id invalid-field-id}}
+                  target  {:type                        "table-incremental"
+                           :schema                      "public"
+                           :name                        target-table
+                           :database                    db-id
+                           :target-incremental-strategy {:type "append"}}]
+              (mt/with-temp [:model/Transform transform {:name "test transform" :source source, :target target}]
+                (testing "Transform execution fails with invalid checkpoint field ID"
+                  (is (thrown-with-msg?
+                       Exception
+                       #"Checkpoint field does not exist"
+                       (transforms.execute/execute! transform {:run-method :manual}))))))))))))
+
+(deftest checkpoint-field-is-not-active-test
+  (testing "Transform with inactive checkpoint field fails gracefully"
+    (mt/test-drivers #{:postgres}
+      (mt/with-premium-features #{:transforms}
+        (mt/dataset transforms-dataset/transforms-test
+          (with-transform-cleanup! [target-table "inactive_field_target"]
+            (let [db-id   (mt/id)
+                  table-id (mt/id :transforms_products)
+                  field-id (mt/id :transforms_products :id)
+                  source  {:type                        "query"
+                           :query                       {:database db-id
+                                                         :type     :native
+                                                         :native   {:query "SELECT id FROM {{source_table}} as s"
+                                                                    :template-tags {"source_table" {:id "source_table"
+                                                                                                    :name "source_table"
+                                                                                                    :display-name "Source Table"
+                                                                                                    :type "table"
+                                                                                                    :table-id table-id
+                                                                                                    :required true}}}}
+                           :source-incremental-strategy {:type "checkpoint"
+                                                         :checkpoint-filter-field-id field-id}}
+                  target  {:type                        "table-incremental"
+                           :schema                      "public"
+                           :name                        target-table
+                           :database                    db-id
+                           :target-incremental-strategy {:type "append"}}]
+              (mt/with-temp [:model/Transform transform {:name "test transform" :source source, :target target}]
+                ;; Mark the field as inactive
+                (t2/update! :model/Field field-id {:active false})
+                (try
+                  (testing "Transform execution fails with inactive checkpoint field"
+                    (is (thrown-with-msg?
+                         Exception
+                         #"Checkpoint field does not exist or is not active"
+                         (transforms.execute/execute! transform {:run-method :manual}))))
+                  (finally
+                    ;; Restore field to active state for other tests
+                    (t2/update! :model/Field field-id {:active true})))))))))))
+
+(deftest changing-query-keeps-checkpoint-test
+  (mt/test-drivers #{:postgres}
+    (mt/with-premium-features #{:transforms}
+      (mt/dataset transforms-dataset/transforms-test
+        (with-transform-cleanup! [target-table "change_table_target"]
+          (let [db-id   (mt/id)
+                db-spec (sql-jdbc.conn/db->pooled-connection-spec db-id)
+                table-id (mt/id :transforms_products)
+                checkpoint-field-id (t2/select-one-pk :model/Field :name "id" :table_id table-id)
+                make-source (fn [where-clause]
+                              {:type "query"
+                               :query {:database db-id
+                                       :type :native
+                                       :native {:query (str "SELECT id FROM {{source_table}} AS s WHERE " where-clause)
+                                                :template-tags {"source_table" {:id "source_table"
+                                                                                :name "source_table"
+                                                                                :display-name "Source Table"
+                                                                                :type "table"
+                                                                                :table-id table-id
+                                                                                :required true}}}}
+                               :source-incremental-strategy {:type "checkpoint"
+                                                             :checkpoint-filter-field-id checkpoint-field-id}})
+                source  (make-source "id = 1")
+                target  {:type                        "table-incremental"
+                         :schema                      "public"
+                         :name                        target-table
+                         :database                    db-id
+                         :target-incremental-strategy {:type "append"}}]
+            (mt/with-temp [:model/Transform transform {:name "test transform" :source source, :target target}]
+              (testing "initial run"
+                (transforms.execute/execute! transform {:run-method :manual})
+                (is (= [{:id 1}] (pg-table-rows db-spec target-table))))
+              (testing "initial increment (no new data)"
+                (transforms.execute/execute! transform {:run-method :manual})
+                (is (= [{:id 1}] (pg-table-rows db-spec target-table))))
+              (t2/update! :model/Transform
+                          (:id transform)
+                          {:source (make-source "id IN (1, 2)")})
+              (testing "second increment after query change (adds id=2)"
+                (transforms.execute/execute! (t2/select-one :model/Transform (:id transform)) {:run-method :manual})
+                (is (= [{:id 1} {:id 2}] (sort-by :id (pg-table-rows db-spec target-table))))))))))))
+
+(deftest native-query-with-table-tag-test
+  (testing "Native query with table tag uses subquery expansion for incremental filtering"
+    (doseq [checkpoint-type [:integer :float]]
+      (testing (format "with %s checkpoint" (name checkpoint-type))
+        (mt/test-drivers (test-drivers)
+          (mt/with-premium-features #{:transforms}
+            (mt/dataset transforms-dataset/transforms-test
+              (with-transform-cleanup! [target-table (target-table-gen "native_table_tag")]
+                (let [checkpoint-config (get checkpoint-configs checkpoint-type)
+                      {:keys [field-name expected-initial-checkpoint expected-second-checkpoint]} checkpoint-config
+                      checkpoint-filter-field-id (t2/select-one-pk :model/Field :name field-name :table_id (mt/id :transforms_products))
+                      transform-payload {:name "Native With Table Tag"
+                                         :source {:type "query"
+                                                  :query (make-incremental-source-query-with-table-tag checkpoint-config)
+                                                  :source-incremental-strategy {:type "checkpoint"
+                                                                                :checkpoint-filter-field-id checkpoint-filter-field-id}}
+                                         :target (merge target-table
+                                                        {:type                        "table-incremental"
+                                                         :target-incremental-strategy {:type "append"}})}]
+                  (mt/with-temp [:model/Transform transform transform-payload]
+                    (testing "First run processes first batch"
+                      (transforms.execute/execute! transform {:run-method :manual})
+                      (transforms.tu/wait-for-table (:name target-table) 10000)
+                      (let [row-count           (get-table-row-count target-table)
+                            distinct-timestamps (get-distinct-timestamp-count target-table)
+                            checkpoint          (get-checkpoint-value (:id transform))]
+                        (is (= 10 row-count) "First run should process the first 10 products")
+                        (is (= 1 distinct-timestamps) "All rows should have the same timestamp from first run")
+                        (is (compare-checkpoint-values checkpoint-type expected-initial-checkpoint checkpoint)
+                            (format "Checkpoint should be MAX(%s) from first 10 rows" field-name))))
+
+                    (testing "Second run processes remaining data"
+                      (let [transform (t2/select-one :model/Transform (:id transform))]
+                        (transforms.execute/execute! transform {:run-method :manual})
+                        (let [row-count           (get-table-row-count target-table)
+                              distinct-timestamps (get-distinct-timestamp-count target-table)
+                              checkpoint          (get-checkpoint-value (:id transform))]
+                          (is (= 16 row-count) "Second run should add remaining 6 rows")
+                          (is (= 2 distinct-timestamps) "Should have 2 distinct timestamps (one per incremental run)")
+                          (is (compare-checkpoint-values checkpoint-type expected-second-checkpoint checkpoint)
+                              (format "Checkpoint should be MAX(%s) from all 16 rows" field-name)))))
+
+                    (testing "Third run without new data adds nothing"
+                      (let [transform (t2/select-one :model/Transform (:id transform))]
+                        (transforms.execute/execute! transform {:run-method :manual})
+                        (let [row-count (get-table-row-count target-table)]
+                          (is (= 16 row-count) "Should still have 16 rows, no new data"))))
+
+                    (when (and (isa? driver/hierarchy driver/*driver* :sql-jdbc)
+                               (not= driver/*driver* :clickhouse)
+                               (not= driver/*driver* :snowflake))
+                      (testing "After inserting new data, incremental run appends only new rows"
+                        (with-insert-test-products!
+                          [{:name "New Table Tag Product"
+                            :category "Electronics"
+                            :price 299.99
+                            :created-at "2024-01-21T10:00:00"}]
+
+                          (let [transform (t2/select-one :model/Transform (:id transform))]
+                            (transforms.execute/execute! transform {:run-method :manual})
+                            (let [row-count  (get-table-row-count target-table)
+                                  checkpoint (get-checkpoint-value (:id transform))]
+                              (is (= 17 row-count) "Should append 1 new row (16 + 1 = 17)")
+                              (is (some? checkpoint) "Checkpoint should be updated"))))))))))))))))

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

@@ -63,8 +63,10 @@ export type TransformSourceCheckpointStrategy = {
   type: "checkpoint";
   // For native queries
   "checkpoint-filter"?: string;
-  // For MBQL and Python queries
+  // For MBQL/Python (legacy)
   "checkpoint-filter-unique-key"?: string;
+  // For MBQL/Python (new)
+  "checkpoint-filter-field-id"?: number;
 };
 
 export type SourceIncrementalStrategy = TransformSourceCheckpointStrategy;

frontend/src/metabase/transforms/components/IncrementalTransform/IncrementalTransformSettings.tsx

@@ -14,6 +14,7 @@ import {
 } from "metabase/transforms/constants";
 import { getLibQuery, isMbqlQuery } from "metabase/transforms/utils";
 import {
+  Alert,
   Anchor,
   Box,
   Divider,
@@ -23,11 +24,12 @@ import {
   Text,
   Tooltip,
 } from "metabase/ui";
-import type * as Lib from "metabase-lib";
+import * as Lib from "metabase-lib";
 import type { TransformSource } from "metabase-types/api";
 
 import {
   MBQLKeysetColumnSelect,
+  NativeQueryTableTagFieldSelect,
   PythonKeysetColumnSelect,
 } from "./KeysetColumnSelect";
 import { NativeQueryColumnSelect } from "./NativeQueryColumnSelect";
@@ -67,6 +69,17 @@ export const IncrementalTransformSettings = ({
     .with({ isMbqlQuery: true }, () => "query" as const)
     .with({ isPythonTransform: true }, () => "python" as const)
     .otherwise(() => "native" as const);
+
+  // Check if native query has table template tags
+  // Incremental transforms for native queries require table template tags
+  const hasTableTemplateTags = libQuery
+    ? Object.values(Lib.templateTags(libQuery)).some(
+        (tag) => tag.type === "table" && tag["table-id"] != null,
+      )
+    : false;
+
+  const isNativeWithoutTableTags =
+    transformType === "native" && !hasTableTemplateTags;
   const { url: incrementalTransformsDocsUrl, showMetabaseLinks } = useDocsUrl(
     isPythonTransform
       ? "data-studio/transforms/python-transforms#incremental-python-transforms"
@@ -77,14 +90,19 @@ export const IncrementalTransformSettings = ({
     const switchContent = (
       <Switch
         disabled={
-          readOnly || isMultiTablePythonTransform || isRemoteSyncReadOnly
+          readOnly ||
+          isMultiTablePythonTransform ||
+          isRemoteSyncReadOnly ||
+          isNativeWithoutTableTags
         }
         checked={incremental}
         size="sm"
         label={
           isMultiTablePythonTransform
             ? t`Incremental transforms are only supported for single data source transforms.`
-            : t`Only process new and changed data`
+            : isNativeWithoutTableTags
+              ? t`Incremental transforms for native queries require at least one table template tag.`
+              : t`Only process new and changed data`
         }
         wrapperProps={{
           "data-testid": "incremental-switch",
@@ -216,6 +234,29 @@ function SourceStrategyFields({
   readOnly,
 }: SourceStrategyFieldsProps) {
   const { values } = useFormikContext<IncrementalSettingsFormValues>();
+
+  // Check if native query has table template tags
+  const hasTableTags =
+    type === "native" && query
+      ? Object.values(Lib.templateTags(query)).some(
+          (tag) => tag.type === "table" && tag["table-id"] != null,
+        )
+      : false;
+
+  // For native queries:
+  // - If query has table tags -> use new NativeQueryTableTagFieldSelect (field ID based)
+  // - Otherwise -> use legacy NativeQueryColumnSelect (column name based)
+  const shouldUseLegacyNativeSelect = type === "native" && !hasTableTags;
+
+  // Show warning for legacy configurations that have checkpoint-filter but should migrate
+  const isLegacyNativeCheckpoint =
+    source.type === "query" &&
+    type === "native" &&
+    hasTableTags &&
+    source["source-incremental-strategy"]?.type === "checkpoint" &&
+    source["source-incremental-strategy"]["checkpoint-filter"] != null &&
+    source["source-incremental-strategy"]["checkpoint-filter-field-id"] == null;
+
   return (
     <>
       {SOURCE_STRATEGY_OPTIONS.length > 1 && (
@@ -229,9 +270,14 @@ function SourceStrategyFields({
       )}
       {values.sourceStrategy === "checkpoint" && (
         <>
+          {isLegacyNativeCheckpoint && (
+            <Alert variant="warning" mb="md">
+              {t`This transform uses a legacy configuration. Please update the checkpoint field selection to ensure compatibility with future versions.`}
+            </Alert>
+          )}
           {type === "query" && query && (
             <MBQLKeysetColumnSelect
-              name="checkpointFilterUniqueKey"
+              name="checkpointFilterFieldId"
               label={t`Field to check for new values`}
               placeholder={t`Pick a field`}
               description={t`Pick the field that we should scan to determine which records are new or changed`}
@@ -241,20 +287,32 @@ function SourceStrategyFields({
               disabled={readOnly}
             />
           )}
-          {type === "native" && query && (
-            <NativeQueryColumnSelect
-              name="checkpointFilter"
-              label={t`Column to check for new values`}
-              placeholder={t`Pick a column`}
-              description={t`Pick the column that we should scan to determine which records are new or changed`}
-              descriptionProps={{ lh: "1rem" }}
-              query={query}
-              disabled={readOnly}
-            />
-          )}
+          {type === "native" &&
+            query &&
+            (shouldUseLegacyNativeSelect ? (
+              <NativeQueryColumnSelect
+                name="checkpointFilter"
+                label={t`Column to check for new values`}
+                placeholder={t`Pick a column`}
+                description={t`Pick the column that we should scan to determine which records are new or changed`}
+                descriptionProps={{ lh: "1rem" }}
+                query={query}
+                disabled={readOnly}
+              />
+            ) : (
+              <NativeQueryTableTagFieldSelect
+                name="checkpointFilterFieldId"
+                label={t`Field to check for new values`}
+                placeholder={t`Pick a field`}
+                description={t`Pick the field that we should scan to determine which records are new or changed`}
+                descriptionProps={{ lh: "1rem" }}
+                query={query}
+                disabled={readOnly}
+              />
+            ))}
           {type === "python" && "source-tables" in source && (
             <PythonKeysetColumnSelect
-              name="checkpointFilterUniqueKey"
+              name="checkpointFilterFieldId"
               label={t`Field to check for new values`}
               placeholder={t`Pick a field`}
               description={t`Pick the field that we should scan to determine which records are new or changed`}

frontend/src/metabase/transforms/components/IncrementalTransform/KeysetColumnSelect/KeysetColumnSelect.tsx

@@ -9,6 +9,32 @@ import {
 } from "metabase/ui";
 import * as Lib from "metabase-lib";
 
+/**
+ * Extract the field ID from a column using the legacy field reference.
+ * Returns null if the column doesn't have a concrete field ID (e.g., derived columns).
+ */
+function extractFieldId(
+  query: Lib.Query,
+  stageIndex: number,
+  column: Lib.ColumnMetadata,
+): number | null {
+  try {
+    const legacyRef = Lib.legacyRef(query, stageIndex, column);
+    // Legacy field reference format: ["field", fieldId, options]
+    // We only want concrete field IDs (numbers), not string names
+    if (
+      Array.isArray(legacyRef) &&
+      legacyRef[0] === "field" &&
+      typeof legacyRef[1] === "number"
+    ) {
+      return legacyRef[1];
+    }
+    return null;
+  } catch {
+    return null;
+  }
+}
+
 type KeysetColumnSelectProps = {
   name: string;
   label: string;
@@ -55,21 +81,31 @@ export function KeysetColumnSelect({
         );
       });
 
-      // Convert to select options with unique keys
-      const seenKeys = new Set<string>();
+      // Convert to select options with field IDs
+      // Only include columns that have concrete field IDs (excludes derived/computed columns)
+      const seenFieldIds = new Set<number>();
       const uniqueColumns: Array<SelectOption> = [];
 
       numericFilterableColumns.forEach((column) => {
-        const columnInfo = Lib.displayInfo(query, stageIndex, column);
-        const uniqueKey = Lib.columnUniqueKey(column);
+        const fieldId = extractFieldId(query, stageIndex, column);
 
-        if (!seenKeys.has(uniqueKey)) {
-          seenKeys.add(uniqueKey);
-
-          const label = columnInfo.longDisplayName;
-
-          uniqueColumns.push({ value: uniqueKey, label });
+        // Skip derived/computed columns that don't have field IDs
+        if (fieldId == null) {
+          return;
         }
+
+        // Skip duplicates (can happen with joins)
+        if (seenFieldIds.has(fieldId)) {
+          return;
+        }
+
+        seenFieldIds.add(fieldId);
+
+        const columnInfo = Lib.displayInfo(query, stageIndex, column);
+        const label = columnInfo.longDisplayName;
+
+        // Mantine Select requires string values, so convert field ID to string
+        uniqueColumns.push({ value: String(fieldId), label });
       });
 
       return uniqueColumns;

frontend/src/metabase/transforms/components/IncrementalTransform/KeysetColumnSelect/NativeQueryTableTagFieldSelect.tsx

@@ -0,0 +1,238 @@
+import { skipToken } from "@reduxjs/toolkit/query";
+import { useMemo } from "react";
+import { t } from "ttag";
+
+import { useGetTableQueryMetadataQuery } from "metabase/api";
+import { FormSelect } from "metabase/forms";
+import { useSelector } from "metabase/lib/redux";
+import { getMetadata } from "metabase/selectors/metadata";
+import {
+  Alert,
+  type DataAttributes,
+  type InputDescriptionProps,
+  Loader,
+  type SelectOption,
+} from "metabase/ui";
+import * as Lib from "metabase-lib";
+
+type NativeQueryTableTagFieldSelectProps = {
+  name: string;
+  label: string;
+  placeholder: string;
+  description: React.ReactNode;
+  descriptionProps?: InputDescriptionProps & DataAttributes;
+  query: Lib.Query;
+  disabled?: boolean;
+};
+
+export function NativeQueryTableTagFieldSelect({
+  name,
+  label,
+  placeholder,
+  description,
+  descriptionProps,
+  query,
+  disabled,
+}: NativeQueryTableTagFieldSelectProps) {
+  const metadata = useSelector(getMetadata);
+
+  // Extract table template tags from the native query
+  // Limit to 10 tables to keep hook calls manageable
+  const tableIds = useMemo(() => {
+    try {
+      const templateTags = Lib.templateTags(query);
+      const tableTags = Object.values(templateTags).filter(
+        (tag) => tag.type === "table" && tag["table-id"] != null,
+      );
+      return tableTags.map((tag) => tag["table-id"]!).slice(0, 10);
+    } catch {
+      return [];
+    }
+  }, [query]);
+
+  // Fetch metadata for up to 10 tables (hooks must be called unconditionally)
+  const table0 = useGetTableQueryMetadataQuery(
+    tableIds[0] != null ? { id: tableIds[0] } : skipToken,
+  );
+  const table1 = useGetTableQueryMetadataQuery(
+    tableIds[1] != null ? { id: tableIds[1] } : skipToken,
+  );
+  const table2 = useGetTableQueryMetadataQuery(
+    tableIds[2] != null ? { id: tableIds[2] } : skipToken,
+  );
+  const table3 = useGetTableQueryMetadataQuery(
+    tableIds[3] != null ? { id: tableIds[3] } : skipToken,
+  );
+  const table4 = useGetTableQueryMetadataQuery(
+    tableIds[4] != null ? { id: tableIds[4] } : skipToken,
+  );
+  const table5 = useGetTableQueryMetadataQuery(
+    tableIds[5] != null ? { id: tableIds[5] } : skipToken,
+  );
+  const table6 = useGetTableQueryMetadataQuery(
+    tableIds[6] != null ? { id: tableIds[6] } : skipToken,
+  );
+  const table7 = useGetTableQueryMetadataQuery(
+    tableIds[7] != null ? { id: tableIds[7] } : skipToken,
+  );
+  const table8 = useGetTableQueryMetadataQuery(
+    tableIds[8] != null ? { id: tableIds[8] } : skipToken,
+  );
+  const table9 = useGetTableQueryMetadataQuery(
+    tableIds[9] != null ? { id: tableIds[9] } : skipToken,
+  );
+
+  // Memoize the array to avoid recreating it on every render
+  const tableQueries = useMemo(
+    () => [
+      table0,
+      table1,
+      table2,
+      table3,
+      table4,
+      table5,
+      table6,
+      table7,
+      table8,
+      table9,
+    ],
+    [
+      table0,
+      table1,
+      table2,
+      table3,
+      table4,
+      table5,
+      table6,
+      table7,
+      table8,
+      table9,
+    ],
+  );
+
+  const isLoading = tableQueries.some((q) => q.isLoading);
+  const hasError = tableQueries.some((q) => q.isError);
+
+  const tables = useMemo(
+    () =>
+      tableQueries
+        .map((q) => q.data)
+        .filter((t): t is NonNullable<typeof t> => t != null),
+    [tableQueries],
+  );
+
+  // Extract field options from all tables
+  const fieldOptions = useMemo((): Array<SelectOption> => {
+    if (tables.length === 0) {
+      return [];
+    }
+
+    try {
+      const allOptions: Array<SelectOption> = [];
+      const seenFieldIds = new Set<number>();
+      const showTablePrefix = tables.length > 1;
+
+      for (const table of tables) {
+        const metadataProvider = Lib.metadataProvider(table.db_id, metadata);
+        const tableMetadata = Lib.tableOrCardMetadata(
+          metadataProvider,
+          table.id,
+        );
+        if (!tableMetadata) {
+          continue;
+        }
+
+        const query = Lib.queryFromTableOrCardMetadata(
+          metadataProvider,
+          tableMetadata,
+        );
+        const stageIndex = -1;
+
+        // Get columns from this table
+        const returnedColumns = Lib.returnedColumns(query, stageIndex);
+        const filterableColumns = Lib.filterableColumns(query, stageIndex);
+
+        const filterableIdentifiers = new Set(
+          filterableColumns.map((col) => Lib.columnKey(col)),
+        );
+
+        // Filter to numeric/temporal columns with field IDs
+        const numericFilterableColumns = returnedColumns.filter((column) => {
+          return (
+            filterableIdentifiers.has(Lib.columnKey(column)) &&
+            (Lib.isNumeric(column) || Lib.isTemporal(column))
+          );
+        });
+
+        for (const column of numericFilterableColumns) {
+          // Extract field ID from column
+          const legacyRef = Lib.legacyRef(query, stageIndex, column);
+          let fieldId: number | null = null;
+
+          if (
+            Array.isArray(legacyRef) &&
+            legacyRef[0] === "field" &&
+            typeof legacyRef[1] === "number"
+          ) {
+            fieldId = legacyRef[1];
+          }
+
+          if (fieldId == null || seenFieldIds.has(fieldId)) {
+            continue;
+          }
+
+          seenFieldIds.add(fieldId);
+
+          const columnInfo = Lib.displayInfo(query, stageIndex, column);
+          // When multiple tables, prepend table name for clarity
+          const label = showTablePrefix
+            ? `${table.display_name || table.name}: ${columnInfo.displayName}`
+            : columnInfo.displayName;
+
+          allOptions.push({ value: String(fieldId), label });
+        }
+      }
+
+      return allOptions;
+    } catch (error) {
+      console.error(
+        "NativeQueryTableTagFieldSelect: Error extracting fields:",
+        error,
+      );
+      return [];
+    }
+  }, [tables, metadata]);
+
+  // Determine the message to show when no query can be built
+  const noQueryMessage = useMemo(() => {
+    if (tableIds.length === 0) {
+      return t`Native queries must use at least one table template tag to enable incremental transforms`;
+    }
+    if (hasError) {
+      return t`Unable to load table metadata. You may not have permission to access the table.`;
+    }
+    return null;
+  }, [tableIds, hasError]);
+
+  if (noQueryMessage) {
+    return (
+      <Alert variant="warning" mb="md">
+        {noQueryMessage}
+      </Alert>
+    );
+  }
+
+  return (
+    <FormSelect
+      name={name}
+      label={label}
+      placeholder={placeholder}
+      description={description}
+      descriptionProps={descriptionProps}
+      data={fieldOptions}
+      searchable
+      disabled={disabled || fieldOptions.length === 0}
+      rightSection={isLoading ? <Loader size="xs" /> : undefined}
+    />
+  );
+}

frontend/src/metabase/transforms/components/IncrementalTransform/KeysetColumnSelect/index.ts

@@ -1,3 +1,4 @@
 export { KeysetColumnSelect } from "./KeysetColumnSelect";
 export { MBQLKeysetColumnSelect } from "./MBQLKeysetColumnSelect";
+export { NativeQueryTableTagFieldSelect } from "./NativeQueryTableTagFieldSelect";
 export { PythonKeysetColumnSelect } from "./PythonKeysetColumnSelect";

frontend/src/metabase/transforms/components/IncrementalTransform/form.ts

@@ -12,6 +12,7 @@ export type IncrementalSettingsFormValues = {
   sourceStrategy: "checkpoint";
   checkpointFilter: string | null;
   checkpointFilterUniqueKey: string | null;
+  checkpointFilterFieldId: string | null; // String because Mantine Select requires string values
   targetStrategy: "append";
 };
 
@@ -20,8 +21,10 @@ export const VALIDATION_SCHEMA = Yup.object({
   sourceStrategy: Yup.mixed<"checkpoint">().oneOf(["checkpoint"]).required(),
   // For native queries, use checkpointFilter (plain string)
   checkpointFilter: Yup.string().nullable().defined(),
-  // For MBQL/Python queries, use checkpointFilterUniqueKey (prefixed format)
+  // For MBQL/Python queries (legacy), use checkpointFilterUniqueKey (prefixed format)
   checkpointFilterUniqueKey: Yup.string().nullable().defined(),
+  // For MBQL/Python queries (new), use checkpointFilterFieldId (string because Mantine Select requires strings)
+  checkpointFilterFieldId: Yup.string().nullable().defined(),
   targetStrategy: Yup.mixed<"append">().oneOf(["append"]).required(),
 });
 
@@ -32,6 +35,7 @@ export const getInitialValues = (
   sourceStrategy: "checkpoint",
   checkpointFilter: null,
   checkpointFilterUniqueKey: null,
+  checkpointFilterFieldId: null,
   targetStrategy: "append",
   ...defaults,
 });
@@ -42,7 +46,7 @@ export const getIncrementalSettingsFromTransform = (
   const isIncremental = transform.target.type === "table-incremental";
   const strategy = transform.source["source-incremental-strategy"];
 
-  // Read both fields from the strategy, whichever is present
+  // Read all fields from the strategy, whichever is present
   const checkpointFilter =
     strategy?.type === "checkpoint"
       ? (strategy["checkpoint-filter"] ?? null)
@@ -53,11 +57,21 @@ export const getIncrementalSettingsFromTransform = (
       ? (strategy["checkpoint-filter-unique-key"] ?? null)
       : null;
 
+  const checkpointFilterFieldId =
+    strategy?.type === "checkpoint"
+      ? (strategy["checkpoint-filter-field-id"] ?? null)
+      : null;
+
   return {
     incremental: isIncremental,
     sourceStrategy: "checkpoint",
     checkpointFilter: isIncremental ? checkpointFilter : null,
     checkpointFilterUniqueKey: isIncremental ? checkpointFilterUniqueKey : null,
+    // Convert number to string for Mantine Select
+    checkpointFilterFieldId:
+      isIncremental && checkpointFilterFieldId != null
+        ? String(checkpointFilterFieldId)
+        : null,
     targetStrategy: "append",
   };
 };
@@ -88,11 +102,23 @@ export const buildIncrementalSource = (
   }
 
   // Build strategy fields based on which checkpoint field is present
-  const strategyFields = formValues.checkpointFilter
-    ? { "checkpoint-filter": formValues.checkpointFilter }
-    : formValues.checkpointFilterUniqueKey
-      ? { "checkpoint-filter-unique-key": formValues.checkpointFilterUniqueKey }
-      : {};
+  // Priority: checkpoint-filter-field-id (MBQL/Python/Native with table tags) > checkpoint-filter (legacy native) > checkpoint-filter-unique-key (legacy)
+  const strategyFields =
+    formValues.checkpointFilterFieldId != null
+      ? {
+          // Convert string back to number for API
+          "checkpoint-filter-field-id": Number(
+            formValues.checkpointFilterFieldId,
+          ),
+        }
+      : formValues.checkpointFilter
+        ? { "checkpoint-filter": formValues.checkpointFilter }
+        : formValues.checkpointFilterUniqueKey
+          ? {
+              "checkpoint-filter-unique-key":
+                formValues.checkpointFilterUniqueKey,
+            }
+          : {};
 
   return {
     ...source,

frontend/src/metabase/transforms/components/QueryComplexityWarning/use-query-complexity-checks.tsx

@@ -23,10 +23,20 @@ const getQueryForComplexityCheck = (
   }
 
   const tags = Lib.templateTags(query);
+
+  // No complexity check needed if query has checkpoint template tag (legacy)
   if (CHECKPOINT_TEMPLATE_TAG in tags) {
     return;
   }
 
+  // No complexity check needed if query has table template tags (new approach)
+  const hasTableTags = Object.values(tags).some(
+    (tag) => tag.type === "table" && tag["table-id"] != null,
+  );
+  if (hasTableTags) {
+    return;
+  }
+
   return query;
 };
 

resources/migrations/059_update_migrations.yaml

@@ -3573,6 +3573,47 @@ databaseChangeLog:
             dbms: h2
             path: instance_analytics_views/query_log/v2/h2-query_log.sql
             relativeToChangelogFile: true
+  - changeSet:
+      id: v59.2026-02-18T12:00:00
+      author: wotbrew
+      comment: Add last_checkpoint_value to transform table
+      preConditions:
+        - onFail: MARK_RAN
+        - not:
+            - columnExists:
+                tableName: transform
+                columnName: last_checkpoint_value
+      changes:
+        - addColumn:
+            tableName: transform
+            columns:
+              - column:
+                  name: last_checkpoint_value
+                  type: ${text.type}
+                  remarks: Serialized checkpoint watermark value for incremental transforms, temporal as ISO8601
+                  constraints:
+                    nullable: true
+
+  - changeSet:
+      id: v59.2026-02-18T12:00:01
+      author: wotbrew
+      comment: Add last_checkpoint_type columns transform table
+      preConditions:
+        - onFail: MARK_RAN
+        - not:
+            - columnExists:
+                tableName: transform
+                columnName: last_checkpoint_type
+      changes:
+        - addColumn:
+            tableName: transform
+            columns:
+              - column:
+                  name: last_checkpoint_type
+                  type: varchar(32)
+                  remarks: Type of the persisted checkpoint value (integer, decimal, date, temporal)
+                  constraints:
+                    nullable: true
 
 # >>>>>>>>>> DO NOT ADD NEW MIGRATIONS BELOW THIS LINE! ADD THEM ABOVE <<<<<<<<<<
 

src/metabase/models/transforms/transform.clj

@@ -323,7 +323,7 @@
 (defmethod serdes/make-spec "Transform"
   [_model-name opts]
   {:copy      [:name :description :entity_id :owner_email]
-   :skip      [:dependency_analysis_version :source_type :target_db_id]
+   :skip      [:dependency_analysis_version :source_type :target_db_id :last_checkpoint_value :last_checkpoint_type]
    :transform {:created_at         (serdes/date)
                :creator_id         (serdes/fk :model/User)
                :owner_user_id      (serdes/fk :model/User)

src/metabase/transforms/query_impl.clj

@@ -60,13 +60,20 @@
            ;; important to test routing before calling compile-source (whose qp middleware will also throw)
            _                             (throw-if-db-routing-enabled transform driver database)]
        (driver.conn/with-write-connection
-         (let [conn-spec         (driver/connection-spec driver database)
+         (let [conn-spec           (driver/connection-spec driver database)
+               source-range-params (transforms.util/get-source-range-params transform)
+               ;; First incremental run (no checkpoint) should behave like non-incremental
+               ;; to drop and recreate the table rather than appending to existing data.
+               effective-transform-type (if (and (= :table-incremental (keyword (:type target)))
+                                                 (nil? (:last_checkpoint_type transform)))
+                                          :table
+                                          (keyword (:type target)))
                transform-details {:db-id          db
                                   :database       database
                                   :transform-id   id
-                                  :transform-type (keyword (:type target))
+                                  :transform-type effective-transform-type
                                   :conn-spec      conn-spec
-                                  :query          (transforms.util/compile-source transform)
+                                  :query          (transforms.util/compile-source transform source-range-params)
                                   :output-schema  (:schema target)
                                   :output-table   (transforms.util/qualified-table-name driver target)}
                opts              (transform-opts transform-details)
@@ -89,11 +96,14 @@
                (transforms.util/run-cancelable-transform!
                 run-id driver transform-details
                 (fn [_cancel-chan]
-                  (driver/run-transform! driver transform-details opts))))
+                  (driver/run-transform! driver transform-details opts)
+                  (when source-range-params
+                    (transforms.util/save-watermark! (:id transform) source-range-params)))))
              (transforms.util/handle-transform-complete!
               :run-id run-id
               :transform transform
-              :db database)))))
+              :db database
+              :source-range-params source-range-params)))))
      (catch Throwable t
        (if (= :already-running (:error (ex-data t)))
          (log/warnf "Transform %d is already running" id)

src/metabase/transforms/schema.clj

@@ -24,9 +24,11 @@
    [:type [:= "checkpoint"]]
    ;; for native
    [:checkpoint-filter {:optional true} :string]
-   ;; for mbql and python
+   ;; for mbql and python (legacy - keep for native compatibility)
    [:checkpoint-filter-unique-key {:optional true}
-    ::lib.schema/column-unique-key]])
+    ::lib.schema/column-unique-key]
+   ;; for mbql and python (new approach)
+   [:checkpoint-filter-field-id {:optional true} :int]])
 
 (mr/def ::source-incremental-strategy
   [:multi {:dispatch :type}

src/metabase/transforms/util.clj

@@ -1,7 +1,5 @@
 (ns metabase.transforms.util
   (:require
-   [buddy.core.codecs :as codecs]
-   [buddy.core.hash :as buddy-hash]
    [clojure.core.async :as a]
    [clojure.string :as str]
    [java-time.api :as t]
@@ -33,7 +31,6 @@
    [metabase.transforms.settings :as transforms.settings]
    [metabase.util :as u]
    [metabase.util.date-2 :as u.date]
-   [metabase.util.honey-sql-2 :as h2x]
    [metabase.util.log :as log]
    [metabase.util.malli :as mu]
    [metabase.util.malli.registry :as mr]
@@ -63,8 +60,7 @@
   (= :query (transform-type transform)))
 
 (defn native-query-transform?
-  "Check if this is a native query transform.
-  Note: The transform should be normalized (via `normalize-transform`) before calling this function."
+  "Check if this is a native query transform."
   [transform]
   (when (query-transform? transform)
     (let [query (-> transform :source :query)]
@@ -289,11 +285,6 @@
       remap/disable-remaps
       lib/disable-default-limit))
 
-(defn- checkpoint-incremental?
-  "Returns true if `source` uses checkpoint-based incremental strategy."
-  [source]
-  (= :checkpoint (some-> source :source-incremental-strategy :type keyword)))
-
 (defn supported-incremental-filter-type?
   "Returns true if the given base-type is supported for incremental filtering.
 
@@ -302,116 +293,104 @@
   (or (isa? base-type :type/Temporal)
       (isa? base-type :type/Number)))
 
-(defn- source->checkpoint-filter-unique-key
-  "Extract the checkpoint filter column from `query` using the unique key specified in `source-incremental-strategy`."
-  [query source-incremental-strategy]
-  (some->> source-incremental-strategy :checkpoint-filter-unique-key (lib/column-with-unique-key query)))
+(defn- build-filtered-subquery
+  "Build a HoneySQL subquery for a table tag with incremental filtering applied.
 
-(defn- source->checkpoint-filter-column
-  "Resolve the checkpoint filter column for an incremental transform.
+   Returns [sql & params]"
+  [table-metadata checkpoint-column source-range-params driver]
+  (let [{:keys [schema name]} table-metadata
+        table-hsql (if schema
+                     [(keyword schema name)]
+                     [(keyword name)])
+        col-name   (:name checkpoint-column)
+        col-hsql   (keyword col-name)
+        {:keys [lo hi]} source-range-params
+        base-query {:select [:*]
+                    :from   table-hsql}
+        ;; Build WHERE clause: > lo AND <= hi
+        where-clause (cond
+                       (and lo hi)
+                       [:and
+                        [:> col-hsql [:lift (:value lo)]]
+                        [:<= col-hsql [:lift (:value hi)]]]
 
-  Tries to resolve the column using the unique key first.
-  Falls back to looking up the column by name from the target table if a `:checkpoint-filter` is specified.
+                       lo
+                       [:> col-hsql [:lift (:value lo)]]
 
-  Validates that the resolved column has a supported type for checkpoint filtering (numeric or temporal).
-  Throws an exception if the column type is not supported."
-  [query source-incremental-strategy table metadata-provider]
-  (let [{:keys [checkpoint-filter checkpoint-filter-unique-key]} source-incremental-strategy]
-    (when-some [{column-name :name
-                 :keys [base-type]
-                 :as column}
-                (cond
-                  checkpoint-filter-unique-key
-                  (source->checkpoint-filter-unique-key query source-incremental-strategy)
-                  checkpoint-filter
-                  (when-some [field-id (t2/select-one-pk :model/Field
-                                                         :table_id (:id table)
-                                                         :name checkpoint-filter)]
-                    (lib.metadata/field metadata-provider field-id)))]
-      (when-not (supported-incremental-filter-type? base-type)
-        (throw (ex-info (str "Checkpoint column '" column-name "' has unsupported type " (pr-str base-type) ". "
-                             "Only numeric and temporal columns are supported for incremental filtering.")
-                        {:column-name column-name
-                         :base-type   base-type})))
-      column)))
+                       hi
+                       [:<= col-hsql [:lift (:value hi)]]
 
-(defn next-checkpoint
-  "Build a query to compute the MAX of the checkpoint column from the target table.
+                       :else nil)
+        filtered (cond-> base-query
+                   where-clause (assoc :where where-clause
+                                       :order-by [[col-hsql :asc]]))]
+    (sql.qp/format-honeysql driver filtered)))
 
-  Returns a map with `:query` (MBQL query selecting the max) and `:filter-column` (column metadata),
-  or `nil` if the transform doesn't use checkpoint-based incremental strategy or the target table doesn't exist."
-  [{:keys [source target] :as transform}]
-  (let [db-id (transforms.i/target-db-id transform)]
-    (when (checkpoint-incremental? source)
-      (when-let [table (target-table db-id target)]
-        (let [metadata-provider (lib-be/application-database-metadata-provider db-id)
-              table-metadata (lib.metadata/table metadata-provider (:id table))
-              query (lib/query metadata-provider table-metadata)]
-          (when-let [filter-column (source->checkpoint-filter-column query
-                                                                     (:source-incremental-strategy source)
-                                                                     table metadata-provider)]
-            {:query (-> query (lib/aggregate (lib/max filter-column)))
-             :filter-column filter-column}))))))
+(defn- expand-table-tag-to-subquery
+  "Expand the table template tag containing the checkpoint field to a filtered subquery.
 
-(defn- next-checkpoint-value
-  "Execute the checkpoint query and normalize the result for database insertion.
-  Returns `nil` if the target table is empty."
-  [{:keys [query filter-column]}]
-  (let [{:keys [base-type]} filter-column
-        v (some-> query qp/process-query :data :rows first first)]
-    ;; QP return values are lossy, we do a bit of parsing to ensure they're of the right
-    ;; shape for reinsertion
-    (cond
-      (nil? v)
-      nil
+   For queries with multiple table tags, only the tag referencing the table that contains
+   the checkpoint field is expanded. Other table tags are processed normally by the QP.
 
-      (isa? base-type :type/Integer)
-      (bigint v)
-
-      ;; any other number that's not an integer, should be a decimal/float
-      (number? v)
-      (bigdec v)
-
-      :else v)))
+   Finds the table tag matching the checkpoint field's table, builds a filtered subquery,
+   replaces {{tag}} in SQL with the subquery, and prepends subquery params to the param list."
+  [query source-range-params]
+  (let [template-tags       (get-in query [:stages 0 :template-tags])
+        checkpoint-field-id (:checkpoint-filter-field-id source-range-params)
+        _                   (when-not checkpoint-field-id
+                              (throw (ex-info "No checkpoint-filter-field-id in source-range-params"
+                                              {:source-range-params source-range-params})))
+        db-id               (:database query)
+        metadata-provider   (lib-be/application-database-metadata-provider db-id)
+        checkpoint-column   (lib.metadata/field metadata-provider checkpoint-field-id)
+        _                   (when (or (nil? checkpoint-column) (not (:active checkpoint-column)))
+                              (throw (ex-info "Checkpoint field does not exist or is not active" {:checkpoint-field-id checkpoint-field-id})))
+        checkpoint-table-id (:table-id checkpoint-column)
+        ;; Find the table tag that references the checkpoint field's table
+        table-tag           (some (fn [[k v]]
+                                    (when (and (#{:table "table"} (:type v))
+                                               (= checkpoint-table-id (:table-id v)))
+                                      [k v]))
+                                  template-tags)
+        _                   (when-not table-tag
+                              (throw (ex-info "No table template tag found for checkpoint field's table"
+                                              {:checkpoint-table-id checkpoint-table-id
+                                               :template-tags       template-tags})))
+        [tag-name _tag-value] table-tag
+        driver              (some->> db-id (t2/select-one :model/Database) :engine keyword)
+        table-metadata      (lib.metadata/table metadata-provider checkpoint-table-id)
+        [sql & params] (build-filtered-subquery table-metadata checkpoint-column source-range-params driver)
+        native-sql          (get-in query [:stages 0 :native])
+        ;; Replace {{tag-name}} with (subquery-sql)
+        expanded-sql        (str/replace native-sql
+                                         (re-pattern (str "\\{\\{" tag-name "\\}\\}"))
+                                         (str "(" sql ")"))
+        ;; Remove the table tag from template-tags since we've expanded it
+        ;; Other table tags remain and will be processed by QP normally
+        updated-tags        (dissoc template-tags tag-name)]
+    (-> query
+        (assoc-in [:stages 0 :native] expanded-sql)
+        (assoc-in [:stages 0 :template-tags] updated-tags)
+        (assoc ::subquery-params params))))
 
 (defn preprocess-incremental-query
-  "Add checkpoint checkpoint filtering to a query for incremental execution.
+  "Add checkpoint filtering to a query for incremental execution.
 
-  For native queries with a `checkpoint` template tag, adds the checkpoint as a parameter.
-  For MBQL queries, adds a filter clause `WHERE checkpoint_column > checkpoint`.
-  Returns the query unchanged on first run (no checkpoint) or for native queries without the checkpoint tag."
-  [query source-incremental-strategy checkpoint]
-  (if-let [checkpoint-value (next-checkpoint-value checkpoint)]
+   For native queries, expands the table tag containing the checkpoint field to a filtered subquery.
+   For MBQL queries, adds filter clauses `WHERE checkpoint_column > lo AND checkpoint_column <= hi`.
+   Returns the query unchanged when source-range-params is nil."
+  [query source-range-params]
+  (if source-range-params
     (if (lib/native? query)
-      ;; native query with explicit checkpoint filter
-      (if (get-in query [:stages 0 :template-tags "checkpoint"])
-        (update query :parameters conj
-                {:type (if (number? checkpoint-value) :number :text)
-                 :target [:variable [:template-tag "checkpoint"]]
-                 :value checkpoint-value})
-        query)
-      ;; mbql query
-      (lib/filter query (lib/> (source->checkpoint-filter-unique-key query source-incremental-strategy) checkpoint-value)))
+      ;; Native: expand table tag to filtered subquery
+      (expand-table-tag-to-subquery query source-range-params)
+      ;; MBQL: add filter clauses
+      (cond-> query
+        (:lo source-range-params) (lib/filter (lib/>  (:column source-range-params) (:value (:lo source-range-params))))
+        (:hi source-range-params) (lib/filter (lib/<= (:column source-range-params) (:value (:hi source-range-params))))))
+    ;; No range params - return unchanged
     query))
 
-(defn- post-process-incremental-query
-  "Wrap a compiled native query with checkpoint filtering for native queries without explicit checkpoint tags.
-
-  Generates SQL that wraps the original query as a subquery and filters by `checkpoint_filter > (checkpoint_query)`. "
-  [outer-query driver {:keys [source-incremental-strategy] :as source} {checkpoint-query :query :as checkpoint}]
-  (let [{:keys [checkpoint-filter]} source-incremental-strategy]
-    (if (and (lib/native? (:query source))
-             (not (get-in (:query source) [:stages 0 :template-tags "checkpoint"]))
-             (next-checkpoint-value checkpoint))
-      (let [wrap-query (fn [query]
-                         (let [honeysql-query {:select [:*]
-                                               :from [[[:raw (str "(" query ")")] :subquery]]
-                                               :where [:> (h2x/identifier :field checkpoint-filter)
-                                                       [:raw (str "(" (:query (qp.compile/compile checkpoint-query)) ")")]]}]
-                           (first (sql.qp/format-honeysql driver honeysql-query))))]
-        (update outer-query :query wrap-query))
-      outer-query)))
-
 (mu/defn validate-transform-query :- [:maybe [:map [:error :string]]]
   "Verifies that a query transform's query can actually be run as is.  Returns nil on success and an error map on failure."
   [{:keys [source]}]
@@ -423,23 +402,142 @@
       (catch Exception e
         (qp.catch-exceptions/exception-response e)))))
 
+(defn- parse-datetime [^String s] (u.date/parse s))
+
+(defn save-watermark!
+  "Commits the incremental transforms :hi watermark value to the appdb."
+  [transform-id source-range-params]
+  (t2/update! :model/Transform
+              transform-id
+              {:last_checkpoint_type  (:type (:hi source-range-params))
+               :last_checkpoint_value (some-> source-range-params :hi :value str)}))
+
+(defn- deserialize-checkpoint-value [last_checkpoint_type last_checkpoint_value]
+  (case last_checkpoint_type
+    "DateTime" (parse-datetime last_checkpoint_value)
+    "Integer"  (biginteger last_checkpoint_value)
+    "Float"    (bigdec last_checkpoint_value)
+    "Decimal"  (bigdec last_checkpoint_value)))
+
+(defn- interpret-database-checkpoint-value [base-type qp-value]
+  (case base-type
+    :type/Integer  {:type "Integer",  :value  (biginteger qp-value)}
+    :type/Float    {:type "Float",    :value  (bigdec qp-value)}
+    :type/Decimal  {:type "Decimal",  :value  (bigdec qp-value)}
+    :type/DateTime {:type "DateTime", :value  (parse-datetime qp-value)}))
+
+(defn- get-max-from-native-query
+  "Compute MAX(checkpoint-col) from a native query by wrapping it in SELECT MAX().
+   This preserves the user's query logic (LIMITs, filters, etc.).
+   Reuses expand-table-tag-to-subquery with only lo filter (no hi) to avoid computing max of future data."
+  [source-query checkpoint-column checkpoint-field-id lo]
+  (let [db-id (:database source-query)
+        driver (some->> db-id (t2/select-one :model/Database) :engine keyword)
+
+        ;; Build source-range-params with only lo (no hi) - we're computing the new hi!
+        range-params {:lo                         (when lo {:value lo})
+                      :hi                         nil  ; No hi filter for watermark computation
+                      :column                     checkpoint-column
+                      :checkpoint-filter-field-id checkpoint-field-id}
+
+        ;; Expand table tags using the same logic as execution
+        ;; This replaces {{source_table}} with (SELECT * FROM table WHERE col > lo)
+        expanded-query (expand-table-tag-to-subquery source-query range-params)
+
+        ;; Get the expanded SQL and params
+        expanded-sql (get-in expanded-query [:stages 0 :native])
+        subquery-params (::subquery-params expanded-query)
+
+        ;; Wrap the entire user query in SELECT MAX()
+        col-name (:name checkpoint-column)
+        col-kw (keyword col-name)
+        max-query {:select [[[:max col-kw] :max]]
+                   :from [[[:raw (str "(" expanded-sql ")")] :subquery]]}
+        [max-sql & max-params] (sql.qp/format-honeysql driver max-query)
+
+        ;; Combine params: subquery params + max params
+        all-params (concat subquery-params (or max-params []))
+
+        ;; Execute the query
+        native-max-query {:database db-id
+                          :type :native
+                          :native {:query max-sql
+                                   :params all-params}}
+        result (qp/process-query native-max-query)]
+    (some-> result :data :rows first first)))
+
+(defn get-source-range-params
+  "Returns information on the incremental range filters that ought to be applied to a source query.
+
+  Returns a map:
+   :column                     (the lib column value of the incremental filter column)
+   :checkpoint-filter-field-id (the field ID of the checkpoint column)
+   Range predicate terms (maps :type, :value), can be nil (in which case the filter clause should be omitted):
+   :lo                         values in the source table must be > this :value.
+   :hi                         values in the source table must be <= this :value."
+  [{:keys [source] :as transform}]
+  (let [{source-query :query, :keys [source-incremental-strategy]} source
+        {:keys [checkpoint-filter-field-id]} source-incremental-strategy]
+    (when checkpoint-filter-field-id
+      (let [{:keys [last_checkpoint_type last_checkpoint_value]} transform
+            db-id             (transforms.i/target-db-id transform)
+            metadata-provider (lib-be/application-database-metadata-provider db-id)
+            column            (lib.metadata/field metadata-provider checkpoint-filter-field-id)
+            _                   (when (or (nil? column) (not (:active column)))
+                                  (throw (ex-info "Checkpoint field does not exist or is not active"
+                                                  {:checkpoint-filter-field-id checkpoint-filter-field-id})))
+            lo                (when last_checkpoint_type (deserialize-checkpoint-value last_checkpoint_type last_checkpoint_value))
+
+            _ (when-not (supported-incremental-filter-type? (:base-type column))
+                (throw (ex-info (str "Checkpoint column '" (:name column) "' has unsupported type " (pr-str (:base-type column)) ". "
+                                     "Only numeric and temporal columns are supported for incremental filtering.")
+                                {:column column})))
+            [max-value base_type]
+            (if (and source-query (lib/native? source-query))
+              ;; native
+              [(get-max-from-native-query source-query column checkpoint-filter-field-id lo)
+               (:base-type column)]
+              ;; mbql/python
+              (let [table-id          (:table-id column)
+                    limit             (-> transform :source :limit)
+                    table-metadata    (lib.metadata/table metadata-provider table-id)
+                    base-query        (or source-query (lib/query metadata-provider table-metadata))
+                    filtered-query    (if lo (lib/filter base-query (lib/> column lo)) base-query)
+                    ;; if limited need to order by the column otherwise you will get the MAX for the first N rows in arbitrary order instead of top-k
+                    limited-query     (if limit
+                                        (-> (lib/order-by filtered-query column)
+                                            (lib/limit limit))
+                                        filtered-query)
+                    query             (lib/aggregate (lib/append-stage limited-query) (lib/max column))
+                    query-result      (qp/process-query query)
+                    {:keys [results_metadata rows]} (:data query-result)
+                    [{:keys [base_type]}] (:columns results_metadata)]
+                [(ffirst rows) base_type]))
+            hi max-value]
+        {:column                     column
+         :checkpoint-filter-field-id checkpoint-filter-field-id
+         :lo                         (when last_checkpoint_type {:type last_checkpoint_type, :value lo})
+         :hi                         (cond (some? hi) (interpret-database-checkpoint-value base_type hi)
+                                           last_checkpoint_type {:type last_checkpoint_type, :value lo})}))))
+
 (defn compile-source
   "Compile the source query of a transform to SQL, applying incremental filtering if required."
-  [{:keys [source] :as transform}]
-  (let [{:keys [source-incremental-strategy] query-type :type} source]
-    (case (keyword query-type)
-      :query
-      (let [checkpoint (next-checkpoint transform)
-            query (:query source)
-            driver (some->> query :database (t2/select-one :model/Database) :engine keyword)]
-        (binding [driver/*compile-with-inline-parameters*
-                  (or (= :clickhouse driver)
-                      driver/*compile-with-inline-parameters*)]
-          (-> query
-              (preprocess-incremental-query source-incremental-strategy checkpoint)
-              massage-sql-query
-              qp.compile/compile
-              (post-process-incremental-query driver source checkpoint)))))))
+  [{:keys [source]} source-range-params]
+  (let [{query-type :type} source]
+    (assert (= :query query-type))
+    (let [query  (:query source)
+          driver (some->> query :database (t2/select-one :model/Database) :engine keyword)]
+      (binding [driver/*compile-with-inline-parameters*
+                (or (= :clickhouse driver)
+                    driver/*compile-with-inline-parameters*)]
+        (let [preprocessed (-> query
+                               (preprocess-incremental-query source-range-params)
+                               massage-sql-query)
+              compiled     (qp.compile/compile preprocessed)]
+          ;; Prepend subquery params before regular template tag params
+          (if-let [subquery-params (::subquery-params preprocessed)]
+            (update compiled :params #(into subquery-params %))
+            compiled))))))
 
 (defn required-database-features
   "Returns the database features necessary to execute `transform`."
@@ -648,94 +746,10 @@
     (filter #(some tag-ids (get-in % field-path)))
     identity))
 
-(def ^:private metabase-index-prefix "mb_transform_idx_")
-
-(defn- incremental-filter-index-name [schema table-name filter-column-name]
-  (let [prefix metabase-index-prefix
-        suffix (codecs/bytes->hex (buddy-hash/md5 (str/join "|" [schema table-name filter-column-name])))]
-    (str prefix suffix)))
-
-(defn- should-index-incremental-filter-column? [driver database]
-  (and (driver.u/supports? driver :describe-indexes database)
-       (driver.u/supports? driver :transforms/index-ddl database)))
-
-(defn- metabase-owned-index? [index]
-  (str/starts-with? (:index-name index) metabase-index-prefix))
-
-(defn- metabase-incremental-filter-index [filter-column target]
-  (let [table-name  (:name target)
-        schema      (:schema target)
-        filter-name (:name filter-column)]
-    ;; match the schema used by describe-table-indexes (:value denotes the column name, assumed leading column if a composite index)
-    {:index-name (incremental-filter-index-name schema table-name filter-name)
-     :value      filter-name}))
-
-(defn- decide-secondary-index-ddl
-  "Decides which indexes should be dropped/created on the target table.
-
-  e.g. Indexing the incremental :filter-column to accelerate MAX(target.checkpoint) queries on OLTP databases.
-
-  Indexes are represented as maps with at least the keys :index-name, :value.
-  This matches the form used by [[metabase.driver/describe-table-indexes]].
-
-  Returns a map:
-  :drop   - a vector of indexes that are redundant and should be dropped.
-  :create - a vector of desirable indexes that do not exist and that should be created.
-
-  Notes:
-  - If user covering indexes exist they should be reused.
-  - Will never drop user indexes.
-  - Indexes we previously created and are no longer required are dropped."
-  [{:keys [filter-column indexes target database]}]
-  (let [[mb-indexes user-indexes] ((juxt filter remove) metabase-owned-index? indexes)
-        default-mb-index    (when filter-column (metabase-incremental-filter-index filter-column target))
-        column-name         (:name filter-column)
-        existing-user-index (first (filter #(= (:value %) column-name) user-indexes))
-        existing-mb-index   (first (filter #(= (:index-name %) (:index-name default-mb-index)) mb-indexes))
-        driver              (:engine database)
-        intended-index      (when (should-index-incremental-filter-column? driver database)
-                              ;; Prefer reuse to not create redundant indexes
-                              (or existing-user-index default-mb-index))
-        drop                (if intended-index
-                              (remove #(= (:index-name intended-index) (:index-name %)) mb-indexes)
-                              mb-indexes)
-        create              (when (and intended-index
-                                       (not= (:index-name intended-index) (:index-name existing-user-index))
-                                       (not= (:index-name intended-index) (:index-name existing-mb-index)))
-                              [intended-index])]
-    {:drop   (vec drop)
-     :create (vec create)}))
-
-(defn execute-secondary-index-ddl-if-required!
-  "If target table index modifications are required, executes those CREATE/DROP commands.
-  See [[metabase.transforms-util/decide-secondary-index-ddl]] for details."
-  [transform run-id database target]
-  (when (driver.u/supports? (:engine database) :describe-indexes database)
-    (let [driver     (:engine database)
-          indexes    (driver/describe-table-indexes driver database target)
-          checkpoint (next-checkpoint transform)
-          {:keys [drop create]}
-          (decide-secondary-index-ddl
-           {:filter-column (:filter-column checkpoint)
-            :database      database
-            :target        target
-            :indexes       indexes})]
-      (doseq [{:keys [index-name value]} drop]
-        (transforms.instrumentation/with-stage-timing [run-id [:import :drop-incremental-filter-index]]
-          (log/infof "Dropping secondary index %s(%s) for target %s" index-name value (pr-str target))
-          (driver/drop-index! driver (:id database) (:schema target) (:name target) index-name)))
-      (doseq [{:keys [index-name value]} create]
-        (transforms.instrumentation/with-stage-timing [run-id [:import :create-incremental-filter-index]]
-          (log/infof "Creating secondary index %s(%s) for target %s" index-name value (pr-str target))
-          (driver/create-index! driver (:id database) (:schema target) (:name target) index-name [value]))))))
-
 (mu/defn handle-transform-complete!
   "Handles followup tasks for when a transform has completed.
 
-  Specifically, this syncs the target db, publishes a `:event/transform-run-complete` event, and potentially updates
-  the target table's index.
-
-  See [[metabase.transforms-util/decide-secondary-index-ddl]] for details on the index handling."
+  Specifically, this syncs the target db and publishes a `:event/transform-run-complete` event."
   [& {:keys [run-id transform db]}
    :- [:map
        [:run-id ::transforms.schema/run-id]
@@ -751,7 +765,4 @@
                                        :transform-id (:id transform)
                                        :transform-type (keyword (:type target))
                                        :output-schema (:schema target)
-                                       :output-table (qualified-table-name (:engine db) target)}})
-      ;; Creating an index after sync means the filter column is known in the appdb.
-      ;; The index would be synced the next time sync runs, but at time of writing, index sync is disabled.
-      (execute-secondary-index-ddl-if-required! transform run-id db target))))
+                                       :output-table (qualified-table-name (:engine db) target)}}))))

src/metabase/transforms_rest/api/transform.clj

@@ -99,7 +99,9 @@
    [:table {:optional true} [:maybe :map]]
    [:owner_user_id {:optional true} [:maybe pos-int?]]
    [:owner_email {:optional true} [:maybe :string]]
-   [:owner {:optional true} [:maybe OwnerResponse]]])
+   [:owner {:optional true} [:maybe OwnerResponse]]
+   [:last_checkpoint_type {:optional true} [:maybe :string]]
+   [:last_checkpoint_value {:optional true} [:maybe :string]]])
 
 (def ^:private TransformRunResponse
   [:map {:closed true}

test/metabase/driver_test.clj

@@ -17,6 +17,7 @@
    [metabase.test.data.interface :as tx]
    [metabase.util :as u]
    [metabase.util.json :as json]
+   [metabase.util.log :as log]
    [toucan2.core :as t2]))
 
 (set! *warn-on-reflection* true)
@@ -481,3 +482,52 @@
 (deftest deps-flags-when-supported-driver-is-not-covered-test
   (is (thrown-with-msg? clojure.lang.ExceptionInfo #"Database that supports :dependencies/native does not provide an implementation of driver/native-query-deps"
                         (driver/native-query-deps ::mock-deps-driver nil nil))))
+
+(defn- create-index-test-impl! [schema]
+  (let [driver      driver/*driver*
+        suffix      (str (System/currentTimeMillis))
+        table-name  (str "test_table_" suffix)
+        qualified-table-name (keyword schema table-name)
+        index-name  #(str "test_index_" % "_" suffix)
+        database-id (mt/id)
+        cleanup     (fn []
+                      (try
+                        (driver/drop-table! driver database-id qualified-table-name)
+                        (catch Throwable t
+                          (log/fatal t "Could not clean up test table!")
+                          (throw t))))]
+    (when schema
+      (driver/create-schema-if-needed! driver (driver/connection-spec driver database-id) schema))
+    (testing "Throws a driver-specific exception if the table does not exist"
+      (is (thrown? Throwable (driver/create-index! driver database-id schema table-name (index-name "a") [:a]))))
+    (try
+      (driver/create-table! driver database-id qualified-table-name {:a [:int], :b [:int], :c [:int]})
+      (testing "Create a single column index"
+        (is (nil? (driver/create-index! driver database-id schema table-name (index-name "a") [:a])))
+        (testing "Index now exists"
+          (is (= #{{:type       :normal-column-index
+                    :index-name (index-name "a")
+                    :value      "a"}}
+                 (driver/describe-table-indexes driver database-id {:schema schema :name table-name}))))
+        (testing "Drop the index"
+          (is (nil? (driver/drop-index! driver database-id schema table-name (index-name "a"))))
+          (testing "Index no longer exists"
+            (is (empty? (driver/describe-table-indexes driver database-id {:schema schema :name table-name}))))))
+      (testing "Create a multi column index"
+        (driver/create-index! driver database-id schema table-name (index-name "b_c") [:b :c])
+        (testing "Index now exists"
+          ;; single-part only as non-leading columns currently dropped by describe-table-indexes
+          (is (= #{{:type       :normal-column-index
+                    :index-name (index-name "b_c")
+                    :value      "b"}}
+                 (driver/describe-table-indexes driver database-id {:schema schema :name table-name})))))
+      (finally
+        (cleanup)))))
+
+(deftest create-index-test
+  (mt/test-drivers (mt/normal-drivers-with-feature :transforms/index-ddl)
+    (create-index-test-impl! nil)))
+
+(deftest create-index-schema-test
+  (mt/test-drivers (mt/normal-drivers-with-feature :transforms/index-ddl :schemas)
+    (create-index-test-impl! "flibble")))

test/metabase/transforms/util_test.clj

@@ -331,8 +331,7 @@
                                                          :status "running"
                                                          :run_method "manual"}]
           (with-redefs [transforms.util/sync-target!                       (constantly table)
-                        events/publish-event!                              (constantly nil)
-                        transforms.util/execute-secondary-index-ddl-if-required! (constantly nil)]
+                        events/publish-event!                              (constantly nil)]
             (transforms.util/handle-transform-complete! :run-id run-id :transform transform :db db)
             (is (= transform-id
                    (t2/select-one-fn :transform_id :model/Table :id table-id)))))))))
@@ -364,9 +363,9 @@
   (testing "compile-source produces SQL without a LIMIT clause"
     (mt/test-drivers (mt/normal-drivers-with-feature :transforms/table)
       (mt/with-premium-features #{:transforms}
-        (let [transform {:source {:type  "query"
+        (let [transform {:source {:type  :query
                                   :query (lib/query (mt/metadata-provider) (mt/mbql-query venues))}}
-              {:keys [query]} (transforms.util/compile-source transform)]
+              {:keys [query]} (transforms.util/compile-source transform (transforms.util/get-source-range-params transform))]
           (is (string? query))
           (is (not (re-find #"(?i)\bLIMIT\b" query))
               (str "Expected no LIMIT clause in compiled SQL, got: " query)))))))