Finish implementation of CAST rules for PostgreSQL source databases.

Add a link to the table from the internal catalogs for columns so that we
can match table-source-name in cast rules when migrating from PostgreSQL.

src/pgsql/pgsql-schema.lisp

@@ -185,7 +185,8 @@
      :do
      (let* ((schema    (maybe-add-schema catalog schema-name))
             (table     (maybe-add-table schema table-name :oid table-oid))
-            (field     (make-column :name name
+            (field     (make-column :table table
+                                    :name name
                                     :type-name type
                                     :type-mod typmod
                                     :nullable (not notnull)

src/sources/pgsql/pgsql-cast-rules.lisp

@@ -26,7 +26,8 @@
 (defmethod cast ((field column) &key &allow-other-keys)
   "Return the PostgreSQL type definition from the given PostgreSQL column
    definition"
-  (with-slots (pgloader.catalog::name
+  (with-slots (pgloader.catalog::table
+               pgloader.catalog::name
                pgloader.catalog::type-name
                pgloader.catalog::type-mod
                pgloader.catalog::nullable
@@ -40,7 +41,7 @@
                       (when (and (stringp (column-default field))
                                  (search "identity" (column-default field)))
                         :auto-increment)))
-           (pgcol (apply-casting-rules nil
+           (pgcol (apply-casting-rules (table-source-name pgloader.catalog::table)
                                        pgloader.catalog::name
                                        pgloader.catalog::type-name
                                        ctype

src/utils/catalog.lisp

@@ -71,7 +71,7 @@
 ;;; produce, so that we know how to CREATE TABLEs in PostgreSQL whatever the
 ;;; source is.
 ;;;
-(defstruct column name type-name type-mod nullable default comment
+(defstruct column table name type-name type-mod nullable default comment
            transform extra (transform-default t))
 
 ;;;

test/citus/company.load

@@ -4,6 +4,8 @@ load database
 
    with include drop, reset no sequences
 
+   cast column impressions.seen_at to "timestamp with time zone"
+
    distribute companies using id
    -- distribute campaigns using company_id
    -- distribute ads using company_id from campaigns