fix(sqlite): report pkey and column context on UTF-8 decoding error

src/sources/sqlite/sqlite.lisp

@@ -33,7 +33,13 @@
   (let* ((table-name (table-source-name (source sqlite)))
          (cols       (mapcar #'coldef-name (fields sqlite)))
          (sql        (format nil "SELECT ~{`~a`~^, ~} FROM `~a`;" cols table-name))
-         (pgtypes    (map 'vector #'column-type-name (columns sqlite))))
+         (pgtypes    (map 'vector #'column-type-name (columns sqlite)))
+         ;; locate the primary key column index so decoding errors can cite it
+         (pkey-idx   (let* ((indexes (table-index-list (target sqlite)))
+                            (pkey    (first (remove-if-not #'index-primary indexes)))
+                            (pcol    (when pkey (first (index-columns pkey)))))
+                       (when pcol
+                         (position pcol cols :test #'string-equal)))))
     (log-message :sql "SQLite: ~a" sql)
     (with-connection (*sqlite-db* (source-db sqlite))
       (let* ((db (conn-handle *sqlite-db*))
@@ -47,14 +53,27 @@
                while (sqlite:step-statement statement)
                for row = (let ((v (make-array len)))
                            (loop :for x :below len
-                              :for raw := (sqlite:statement-column-value statement x)
-                              :for ptype := (aref pgtypes x)
-                              :for stype := (sqlite-ffi:sqlite3-column-type
-                                             (sqlite::handle statement)
-                                             x)
-                              :for val := (parse-value raw stype ptype
-                                                       :encoding encoding)
-                              :do (setf (aref v x) val))
+                              :for col-name := (nth x cols)
+                              :for stype    := (sqlite-ffi:sqlite3-column-type
+                                                (sqlite::handle statement) x)
+                              :for ptype    := (aref pgtypes x)
+                              :do (setf (aref v x)
+                                        (handler-case
+                                            (parse-value
+                                             (sqlite:statement-column-value statement x)
+                                             stype ptype :encoding encoding)
+                                          (babel-encodings:character-decoding-error (e)
+                                            (log-message :error
+                                              "While decoding text from SQLite table ~s: ~%~
+Illegal ~a character at byte position ~a~@[, pkey ~s~]~@[, column ~s~].~%"
+                                              table-name
+                                              (babel-encodings:character-coding-error-encoding e)
+                                              (babel-encodings:character-coding-error-position e)
+                                              (when (and pkey-idx (< pkey-idx x))
+                                                (aref v pkey-idx))
+                                              col-name)
+                                            (update-stats :data (target sqlite) :errs 1)
+                                            nil))))
                            v)
                counting t into rows
                do (funcall process-row-fn row)

test/sqlite-bad-utf8.load

@@ -0,0 +1,34 @@
+/*
+ * Regression test for GitHub issue #1250.
+ *
+ * When a SQLite text column contains bytes that are not valid UTF-8, pgloader
+ * must report the error with enough context to locate the offending record
+ * (table name, byte position, primary-key value) and substitute NULL for the
+ * bad column so the rest of the row — and the rest of the table — continue
+ * loading.
+ *
+ * The database has three rows in the "files" table:
+ *   id=1  filename='valid-file.txt'      (valid UTF-8)
+ *   id=2  filename=b'file\x96name.txt'   (Windows-1252 em-dash, invalid UTF-8)
+ *   id=3  filename='another-valid.txt'   (valid UTF-8)
+ *
+ * Expected outcome:
+ *   - Row 1 and row 3 are loaded with the correct filename.
+ *   - Row 2 is loaded with filename = NULL and one error is counted.
+ *   - The error log contains: table name, byte position, pkey value (2).
+ *
+ * To regenerate the database:
+ *   python3 test/sqlite/create-bad-utf8.py
+ */
+
+load database
+     from 'sqlite/bad-utf8.db'
+     into postgresql:///pgloader
+
+     with include drop, create tables, create indexes, reset sequences
+
+      set work_mem to '16MB',
+          search_path to 'badutf8'
+
+   before load do
+     $$ create schema if not exists badutf8; $$;

test/sqlite/create-bad-utf8.py

@@ -0,0 +1,35 @@
+#!/usr/bin/env python3
+"""
+Regenerate test/sqlite/bad-utf8.db for the SQLite UTF-8 decoding error test
+(GitHub issue #1250).
+
+Usage:
+    python3 test/sqlite/create-bad-utf8.py
+"""
+import os
+import sqlite3
+
+out = os.path.join(os.path.dirname(__file__), "bad-utf8.db")
+if os.path.exists(out):
+    os.remove(out)
+
+conn = sqlite3.connect(out)
+conn.execute(
+    """CREATE TABLE files (
+    id       INTEGER PRIMARY KEY,
+    filename TEXT
+)"""
+)
+
+conn.execute("INSERT INTO files VALUES (1, 'valid-file.txt')")
+# Row 2: Windows-1252 em-dash (0x96) embedded in an otherwise ASCII filename.
+# This byte is invalid UTF-8, so cl-sqlite will raise a decoding error when
+# pgloader tries to read this column.
+conn.execute(
+    "INSERT INTO files VALUES (2, ?)",
+    (sqlite3.Binary(b"file\x96name.txt"),),
+)
+conn.execute("INSERT INTO files VALUES (3, 'another-valid.txt')")
+conn.commit()
+conn.close()
+print(f"Created {out}")