Fix premature COMMIT in stream-rows-to-copy under on error stop

clojure/src/pgloader/core.clj

@@ -454,11 +454,13 @@
               opts-batch-size    (some-> (get (:with-options cmd) :batch-size) (Long/parseLong))
               opts-prefetch-rows (some-> (get (:with-options cmd) :prefetch-rows) (long))
               opts-batch-concurr (some-> (get (:with-options cmd) :batch-concurrency) (long))
+              opts-on-error-stop (get (:with-options cmd) :on-error-stop)
               _ (when (and opts-prefetch-rows opts-batch-concurr)
                   (throw (ex-info "pgloader: 'batch concurrency' (deprecated) and 'prefetch rows' cannot both appear in the same WITH clause; remove 'batch concurrency'" {})))]
           (binding [copy/*batch-rows* (or opts-batch-rows copy/*batch-rows*)
                     copy/*batch-size* (or opts-batch-size copy/*batch-size*)
-                    copy/*prefetch-queue-capacity* (or opts-prefetch-rows opts-batch-concurr copy/*prefetch-queue-capacity*)]
+                    copy/*prefetch-queue-capacity* (or opts-prefetch-rows opts-batch-concurr copy/*prefetch-queue-capacity*)
+                    copy/*on-error-stop* (if (some? opts-on-error-stop) opts-on-error-stop copy/*on-error-stop*)]
             (try
           ;; Send MySQL-specific SET params to the MySQL source connection before catalog fetch
               (when (#{:mysql :mariadb} (:type source-uri))

clojure/test/pgloader/load_file/parser_test.clj

@@ -552,3 +552,20 @@
                    WITH downcase identifiers;")]
       (is (:ok result) (str "Parse failed: " (:error result)))
       (is (true? (get-in result [:ok :with-options :downcase-ids]))))))
+
+(deftest test-with-on-error-stop
+  (testing "WITH on error stop sets :on-error-stop true in with-options"
+    (let [result (parser/parse-string
+                  "LOAD DATABASE FROM sqlite:///tmp/foo.db
+                   INTO postgresql:///target
+                   WITH on error stop;")]
+      (is (:ok result) (str "Parse failed: " (:error result)))
+      (is (true? (get-in result [:ok :with-options :on-error-stop])))))
+
+  (testing "WITH on error resume next does not set :on-error-stop"
+    (let [result (parser/parse-string
+                  "LOAD DATABASE FROM sqlite:///tmp/foo.db
+                   INTO postgresql:///target
+                   WITH on error resume next;")]
+      (is (:ok result) (str "Parse failed: " (:error result)))
+      (is (nil? (get-in result [:ok :with-options :on-error-stop]))))))

src/pg-copy/copy-rows-in-stream.lisp

@@ -25,17 +25,26 @@
                                              :table table
                                              :columns columns
                                              :condition c))))))
+            ;; The unwind-protect cleanup closes the COPY writer (sending
+            ;; CopyDone to the server) but intentionally does NOT COMMIT.
+            ;; cl-postgres's with-syncing in close-db-writer reads until
+            ;; ReadyForQuery even on error, so the connection is always in
+            ;; a clean SQL state when the cleanup finishes.  COMMIT only
+            ;; runs on the success path below; the error handler uses the
+            ;; now-clean connection to ROLLBACK.
             (unwind-protect
                  (loop
                     :for row := (lq:pop-queue queue)
                     :until (eq :end-of-data row)
                     :do (multiple-value-bind (row-bytes row-seconds)
                             (stream-row copier copy nbcols row)
-                          (incf rcount)
-                          (incf bytes row-bytes)
-                          (incf seconds row-seconds)))
-              (cl-postgres:close-db-writer copier)
-              (pomo:execute "COMMIT"))))
+                          (when row-bytes
+                            (incf rcount)
+                            (incf bytes row-bytes)
+                            (incf seconds row-seconds))))
+              (cl-postgres:close-db-writer copier))
+            ;; Only reached when the loop completes without signalling.
+            (pomo:execute "COMMIT")))
 
       (postgresql-unavailable (condition)
         ;; We got disconnected, maybe because PostgreSQL is being restarted,
@@ -52,8 +61,11 @@
         (error condition))
 
       (condition (c)
-        ;; stop at any failure here, this function doesn't implement any kind
-        ;; of retry behaviour.
+        ;; close-db-writer in the unwind-protect cleanup above has already
+        ;; sent CopyDone and read until ReadyForQuery (via cl-postgres's
+        ;; with-syncing), so the connection is now in SQL mode.  ROLLBACK
+        ;; is safe regardless of whether the error came from data streaming
+        ;; or from close-db-writer itself.
         (log-message :error "~a" c)
         (update-stats :data table :errs 1)
         (pomo:execute "ROLLBACK")

test/Makefile

@@ -4,36 +4,37 @@ OUT   = $(TESTS:.load=.out)
 
 REMOTE = archive.load bossa-all.load bossa.load census-places.load dbf-zip.load
 LOCAL  = $(filter-out $(REMOTE:.load=.out),$(OUT))
-REGRESS= allcols.load 		    \
-         csv-before-after.load 	    \
-         csv-districts.load 	    \
-         csv-parse-date.load 	    \
-         csv-error.load 	    \
-         csv-escape-mode.load 	    \
-         csv-filename-pattern.load  \
-         csv-guess.load             \
-         csv-header.load            \
-         csv-json.load              \
-         csv-keep-extra-blanks.load \
-         csv-missing-col.load       \
-         csv-non-printable.load     \
-         csv-nulls.load             \
-         csv-temp.load              \
-         csv-trim-extra-blanks.load \
-         csv-using-sexp.load        \
-         csv.load 		    \
-	 copy.load                  \
-	 copy-hex.load              \
-         dbf.load 		    \
-         errors.load 		    \
-         fk-reject.load 	    \
-         fixed.load 		    \
-         fields-with-periods.load   \
-         ixf.load 		    \
-         overflow.load 		    \
-         partial.load 		    \
-         serial.load 		    \
-         udc.load 		    \
+REGRESS= allcols.load 		       \
+         csv-before-after.load 	       \
+         csv-districts.load 	       \
+         csv-parse-date.load 	       \
+         csv-error.load 	       \
+         csv-escape-mode.load 	       \
+         csv-filename-pattern.load     \
+         csv-guess.load                \
+         csv-header.load               \
+         csv-json.load                 \
+         csv-keep-extra-blanks.load    \
+         csv-missing-col.load          \
+         csv-non-printable.load        \
+         csv-nulls.load                \
+         csv-temp.load                 \
+         csv-trim-extra-blanks.load    \
+         csv-using-sexp.load           \
+         csv.load 		       \
+	 copy.load                     \
+	 copy-hex.load                 \
+         dbf.load 		       \
+         errors.load 		       \
+         fk-reject.load 	       \
+         fixed.load 		       \
+         fields-with-periods.load      \
+         ixf.load 		       \
+         overflow.load 		       \
+         partial.load 		       \
+         serial.load 		       \
+         sqlite-on-error-stop.load     \
+         udc.load 		       \
          xzero.load
 
 # Those are not included in the tests because CCL doesn't have the cp866
@@ -81,6 +82,16 @@ errors.out: errors.load
 	-$(PGLOADER) $<
 	@echo
 
+# sqlite LOAD DATABASE is incompatible with --regress (parse-target-pg-db-uri
+# returns nil for LOAD DATABASE commands).  Run without --regress; any hang
+# would be caught by the CI job timeout.
+sqlite/type-mismatch.db:
+	python3 sqlite/create-type-mismatch.py
+
+regress/out/sqlite-on-error-stop.out: sqlite-on-error-stop.load sqlite/type-mismatch.db
+	-$(PGLOADER) $<
+	touch $@
+
 nofile.out: nofile.load
 	-$(PGLOADER) $<
 	@echo

test/sqlite-on-error-stop.load

@@ -0,0 +1,36 @@
+/*
+ * Regression test for GitHub issue #1622 (comment 2026-06-22).
+ *
+ * When a SQLite INTEGER column contains a TEXT value that PostgreSQL rejects
+ * during COPY, pgloader used to hang indefinitely under "on error stop" mode
+ * because the COPY writer was closed (sending CopyDone) inside the
+ * unwind-protect cleanup, and COMMIT ran there too — so when the outer
+ * handler-case handler tried to ROLLBACK the transaction had already been
+ * committed and the connection state was inconsistent.
+ *
+ * The database has three rows in the "products" table:
+ *   id=1  name='apple'   qty=10           (valid INTEGER)
+ *   id=2  name='banana'  qty='lots-of-it' (TEXT in INTEGER column)
+ *   id=3  name='cherry'  qty=5            (valid INTEGER)
+ *
+ * Expected outcome:
+ *   - pgloader exits cleanly (non-zero exit code is fine).
+ *   - pgloader does NOT hang.
+ *   - An error is logged for the INTEGER type mismatch.
+ *
+ * To regenerate the database:
+ *   python3 test/sqlite/create-type-mismatch.py
+ */
+
+load database
+     from 'sqlite/type-mismatch.db'
+     into postgresql:///pgloader
+
+     with include drop, create tables, create indexes, reset sequences,
+          on error stop
+
+      set work_mem to '16MB',
+          search_path to 'typemismatch'
+
+   before load do
+     $$ create schema if not exists typemismatch; $$;

test/sqlite/create-type-mismatch.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python3
+"""
+Regenerate test/sqlite/type-mismatch.db for the on-error-stop hang test
+(GitHub issue #1622).
+
+The database has three rows in the "products" table:
+  id=1  name='apple'   qty=10           (valid INTEGER)
+  id=2  name='banana'  qty='lots-of-it' (TEXT stored in INTEGER column)
+  id=3  name='cherry'  qty=5            (valid INTEGER)
+
+SQLite stores 'lots-of-it' as TEXT because it has dynamic typing.  When
+pgloader tries to COPY this into a PostgreSQL INTEGER column the server
+rejects the data at CopyDone time.
+
+Usage:
+    python3 test/sqlite/create-type-mismatch.py
+"""
+import os
+import sqlite3
+
+out = os.path.join(os.path.dirname(__file__), "type-mismatch.db")
+if os.path.exists(out):
+    os.remove(out)
+
+conn = sqlite3.connect(out)
+conn.execute(
+    """CREATE TABLE products (
+    id   INTEGER PRIMARY KEY,
+    name TEXT    NOT NULL,
+    qty  INTEGER NOT NULL
+)"""
+)
+
+conn.execute("INSERT INTO products VALUES (1, 'apple',  10)")
+# Row 2: TEXT value in an INTEGER column.  SQLite accepts this because of
+# dynamic typing; PostgreSQL will reject it when COPY tries to parse it.
+conn.execute("INSERT INTO products VALUES (2, 'banana', 'lots-of-it')")
+conn.execute("INSERT INTO products VALUES (3, 'cherry', 5)")
+conn.commit()
+conn.close()
+print(f"Created {out}")