Fix parsing of qualified target table names, see #186.

We used to parse qualified table names as a simple string, which then
breaks attempts to be smart about how to quote idenfifiers. Some sources
are known to accept dots in quoted table names and we need to be able to
process that properly without tripping on qualified table names too
late.

Current code might not be the best approach as it's just using either a
cons or a string for table names internally, rather than defining a
proper data structure with a schema and a name slot.

Well, that's for a later cleanup patch, I happen to be lazy tonight.

src/package.lisp

@@ -59,6 +59,7 @@
            #:unquote
 
            ;; state
+           #:format-table-name
 	   #:make-pgstate
 	   #:pgstate-get-table
 	   #:pgstate-add-table

src/parsers/command-copy.lisp

@@ -121,7 +121,7 @@
                 (make-instance 'pgloader.copy:copy-copy
                                :target-db ,pg-db-conn
                                :source source-db
-                               :target ,(pgconn-table-name pg-db-conn)
+                               :target ',(pgconn-table-name pg-db-conn)
                                :encoding ,encoding
                                :fields ',fields
                                :columns ',columns

src/parsers/command-csv.lisp

@@ -440,7 +440,7 @@
                 (make-instance 'pgloader.csv:copy-csv
                                :target-db  ,pg-db-conn
                                :source     source-db
-                               :target     ,(pgconn-table-name pg-db-conn)
+                               :target    ',(pgconn-table-name pg-db-conn)
                                :encoding   ,encoding
                                :fields    ',fields
                                :columns   ',columns

src/parsers/command-db-uri.lisp

@@ -113,7 +113,7 @@
                                    maybe-quoted-namestring)
   (:destructure (schema dot table)
     (declare (ignore dot))
-    (format nil "~a.~a" (text schema) (text table))))
+    (cons (text schema) (text table))))
 
 (defrule dsn-table-name (or qualified-table-name maybe-quoted-namestring)
   (:lambda (name)

src/parsers/command-fixed.lisp

@@ -140,7 +140,7 @@
                 (make-instance 'pgloader.fixed:copy-fixed
                                :target-db ,pg-db-conn
                                :source source-db
-                               :target ,(pgconn-table-name pg-db-conn)
+                               :target ',(pgconn-table-name pg-db-conn)
                                :encoding ,encoding
                                :fields ',fields
                                :columns ',columns

src/pgsql/pgsql.lisp

@@ -54,20 +54,31 @@
     (truncate-tables pgconn (list table-name)))
 
   (with-pgsql-connection (pgconn)
-    (when disable-triggers (disable-triggers table-name))
+    (let ((unqualified-table-name
+           (typecase table-name
+             (cons   (let ((sql (format nil "SET search_path TO ~a;"
+                                        (car table-name))))
+                       (log-message :notice "~a" sql)
+                       (pgsql-execute sql)
+                       (cdr table-name)))
+             (string table-name))))
+
+      (when disable-triggers (disable-triggers unqualified-table-name))
       (log-message :info "pgsql:copy-from-queue: ~a ~a" table-name columns)
+
       (loop
          for (mesg batch read oversized?) = (lq:pop-queue queue)
          until (eq mesg :end-of-data)
-       for rows = (copy-batch table-name columns batch read)
+         for rows = (copy-batch unqualified-table-name columns batch read)
          do (progn
               ;; The SBCL implementation needs some Garbage Collection
               ;; decision making help... and now is a pretty good time.
               #+sbcl (when oversized? (sb-ext:gc :full t))
               (log-message :debug "copy-batch ~a ~d row~:p~:[~; [oversized]~]"
-                         table-name rows oversized?)
+                           unqualified-table-name rows oversized?)
               (pgstate-incf *state* table-name :rows rows)))
-    (when disable-triggers (enable-triggers table-name))))
+
+      (when disable-triggers (enable-triggers unqualified-table-name)))))
 
 ;;;
 ;;; When a batch has been refused by PostgreSQL with a data-exception, that

src/pgsql/schema.lisp

@@ -230,11 +230,18 @@
 (defun truncate-tables (pgconn table-name-list)
   "Truncate given TABLE-NAME in database DBNAME"
   (with-pgsql-transaction (:pgconn pgconn)
+    (flet ((process-table-name (table-name)
+             (typecase table-name
+               (cons
+                (format nil "~a.~a"
+                        (apply-identifier-case (car table-name))
+                        (apply-identifier-case (cdr table-name))))
+               (string
+                (apply-identifier-case table-name)))))
       (let ((sql (format nil "TRUNCATE ~{~a~^,~};"
-                       (loop :for table-name :in table-name-list
+                         (mapcar #'process-table-name table-name-list))))
-                          :collect (apply-identifier-case table-name)))))
         (log-message :notice "~a" sql)
-      (pomo:execute sql))))
+        (pomo:execute sql)))))
 
 (defun disable-triggers (table-name)
   "Disable triggers on TABLE-NAME. Needs to be called with a PostgreSQL

src/utils/report.lisp

@@ -117,7 +117,7 @@
 	 (format *report-stream*
                  *header-tname-format*
                  *max-length-table-name*
-                 table-name)
+                 (format-table-name table-name))
          (report-results read rows errs (format-interval secs nil)))
      finally (when footer
 	       (report-pgstate-stats pgstate footer))))
@@ -171,11 +171,12 @@
          (*max-length-table-name*
           (reduce #'max
                   (mapcar #'length
+                          (mapcar #'format-table-name
                                   (append (pgstate-tabnames state)
                                           (when before (pgstate-tabnames before))
                                           (when finally (pgstate-tabnames finally))
                                           (when parallel (pgstate-tabnames parallel))
-                                  (list legend)))))
+                                          (list legend))))))
          (*header-tname-format* (get-format-for stype :header-tname-format))
          (*header-stats-format* (get-format-for stype :header-stats-format))
          (*header-cols-format*  (get-format-for stype :header-cols-format))

src/utils/state.lisp

@@ -24,6 +24,12 @@
   (errs 0   :type fixnum)
   (secs 0.0 :type float))
 
+(defun format-table-name (table-name)
+  "TABLE-NAME might be a CONS of a schema name and a table name."
+  (typecase table-name
+    (cons    (format nil "~a.~a" (car table-name) (cdr table-name)))
+    (string  table-name)))
+
 (defun pgstate-get-table (pgstate name)
   (gethash name (pgstate-tables pgstate)))
 
@@ -33,10 +39,11 @@
       (let* ((table (setf (gethash table-name (pgstate-tables pgstate))
 			  (make-pgtable :name table-name)))
 	     (reject-dir  (merge-pathnames (format nil "~a/" dbname) *root-dir*))
+             (filename    (format-table-name table-name))
 	     (data-pathname (make-pathname :defaults reject-dir
-                                           :name table-name :type "dat"))
+                                           :name filename :type "dat"))
 	     (logs-pathname (make-pathname :defaults reject-dir
-                                           :name table-name :type "log")))
+                                           :name filename :type "log")))
 
         ;; maintain the ordering
         (push table-name (pgstate-tabnames pgstate))