mirrors/pgloader
1
0
Fork 0
mirror of https://github.com/dimitri/pgloader.git synced 2026-05-05 10:56:10 +02:00
Code Issues Projects Releases Wiki Activity

Implement PostgreSQL catalogs data structure.

Browse Source 
In order to share more code in between the different source types,
finally have a go at the quite horrible mess of anonymous data
structures floating around.

Having a catalog and schema instances not only allows for code cleanup,
but will also allow to implement some bug fixes and wishlist items such
as mapping tables from a schema to another one.

Also, supporting database sources having a notion of "schema" (in
between "catalog" and "table") should get easier, including getting
on-par with MySQL in the MS SQL support (materialized views has been
asked for already).

See #320, #316, #224 for references and a notion of progress being made.

In passing, also clean up the copy-databases methods for database source
types, so that they all use a fetch-metadata generic function and a
prepare-pgsql-database and a complete-pgsql-database generic function.
Actually, a single method does the job here.

The responsibility of introspecting the source to populate the internal
catalog/schema representation is now held by the fetch-metadata generic
function, which in turn will call the specialized versions of
list-all-columns and friends implementations. Once the catalog has been
fetched, an explicit CAST call is then needed before we can continue.

Finally, the fields/columns/transforms slots in the copy objects are
still being used by the operative code, so the internal catalog
representation is only used up to starting the data copy step, where the
copy class instances are then all that's used.

This might be refactored again in a follow-up patch.
This commit is contained in:
Dimitri Fontaine 2015-12-27 21:27:53 +01:00
parent d84ec3f808
commit 9e4938cea4
35 changed files with 1629 additions and 1518 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
pgloader.asd
View File

@ -65,14 +65,13 @@
;; those are one-package-per-file
(:file "transforms")
(:file "read-sql-files")))
(:file "read-sql-files")
(:file "quoting")
(:file "schema-structs" :depends-on ("quoting"))))
;; generic connection api
(:file "connection" :depends-on ("utils"))
;; some table name and schema facilities
(:file "schema" :depends-on ("package"))
;; package pgloader.pgsql
(:module pgsql
:depends-on ("package" "params" "utils" "connection")
@ -128,6 +127,7 @@
((:file "api")
(:file "methods" :depends-on ("api"))
(:file "md-methods" :depends-on ("api"))
(:file "db-methods" :depends-on ("api"))
(:file "casting-rules")
(:file "files-and-pathnames")
(:file "project-fields")))

149
src/package.lisp
View File

@ -108,7 +108,82 @@
;; charsets
#:list-encodings-and-aliases
#:show-encodings
#:make-external-format))
#:make-external-format
;; quoting
#:apply-identifier-case
;; Schema structure bits
#:catalog
#:schema
#:table
#:column
#:index
#:index
#:fkey
#:cast ; generic function for sources
#:make-catalog
#:make-schema
#:make-table
#:create-table
#:make-view
#:make-column
#:make-index
#:make-index
#:make-fkey
#:catalog-name
#:catalog-schema-list
#:schema-name
#:schema-table-list
#:schema-view-list
#:table-name
#:table-source-name
#:table-schema
#:table-oid
#:table-comment
#:table-field-list
#:table-column-list
#:table-index-list
#:table-fkey-list
#:column-name
#:column-type-name
#:column-type-mod
#:column-type-nullable
#:column-default
#:column-comment
#:column-transform
#:table-list
#:view-list
#:add-schema
#:find-schema
#:maybe-add-schema
#:add-table
#:find-table
#:maybe-add-table
#:add-view
#:find-view
#:maybe-add-view
#:add-field
#:add-column
#:add-index
#:add-fkey
#:find-fkey
#:maybe-add-fkey
#:count-tables
#:count-views
#:count-indexes
#:count-fkeys
#:max-indexes-per-table
#:push-to-end
#:with-schema
#:format-default-value
#:format-column))
(defpackage #:pgloader.batch
(:use #:cl #:pgloader.params #:pgloader.monitor)
@ -196,23 +271,14 @@
#:pgsql-execute
#:pgsql-execute-with-timing
#:pgsql-connect-and-execute-with-timing
;; PostgreSQL schema facilities
#:truncate-tables
#:copy-from-file
#:copy-from-queue
#:list-databases
#:list-tables
#:list-columns-query
#:list-columns
#:list-indexes
#:list-tables-cols
#:list-tables-and-fkeys
#:list-reserved-keywords
#:list-table-oids
#:reset-all-sequences
#:get-date-columns
#:format-vector-row
#:apply-identifier-case
#:create-tables
#:create-views
#:format-pgsql-column
#:format-extra-type
#:make-pgsql-fkey
@ -228,12 +294,30 @@
#:drop-indexes
#:maybe-drop-indexes
#:create-indexes-again
#:reset-sequences))
#:reset-sequences
#:comment-on-tables-and-columns
;; PostgreSQL introspection queries
#:list-databases
#:list-tables
#:list-columns-query
#:list-columns
#:list-indexes
#:list-tables-cols
#:list-tables-and-fkeys
#:list-table-oids
;; PostgreSQL Identifiers
#:list-reserved-keywords
;; PostgreSQL data format
#:get-date-columns
#:format-vector-row))
(defpackage #:pgloader.sources
(:use #:cl
#:pgloader.params #:pgloader.utils #:pgloader.connection
#:pgloader.schema #:pgloader.pgsql #:pgloader.batch)
#:pgloader.pgsql #:pgloader.batch)
(:import-from #:pgloader.transforms
#:precision
#:scale
@ -244,6 +328,7 @@
#:parse-date-format)
(:export #:copy
#:md-copy
#:db-copy
;; Accessors
#:source-db
@ -277,9 +362,13 @@
#:expand-spec
#:open-next-stream
;; common schema facilities
#:push-to-end
#:with-schema
;; the db-methods
#:fetch-metadata
#:prepare-pgsql-database
#:cleanup
#:instanciate-table-copy-object
#:complete-pgsql-database
#:end-kernels
;; file based utils for CSV, fixed etc
#:with-open-file-or-stream
@ -291,7 +380,8 @@
;; database cast machinery
#:*default-cast-rules*
#:*cast-rules*
#:cast))
#:apply-casting-rules
#:format-pgsql-type))
;;;
@ -373,7 +463,6 @@
#:with-pgsql-transaction
#:pgsql-execute
#:pgsql-execute-with-timing
#:apply-identifier-case
#:create-tables
#:format-pgsql-column
#:format-vector-row)
@ -390,7 +479,6 @@
#:with-pgsql-transaction
#:pgsql-execute
#:pgsql-execute-with-timing
#:apply-identifier-case
#:create-tables
#:format-pgsql-column
#:format-vector-row)
@ -410,10 +498,10 @@
#:with-pgsql-transaction
#:pgsql-execute
#:pgsql-execute-with-timing
#:apply-identifier-case
#:list-tables-and-fkeys
#:list-table-oids
#:create-tables
#:create-views
#:truncate-tables
#:format-pgsql-column
#:format-extra-type
@ -427,9 +515,11 @@
#:create-indexes-in-kernel
#:set-table-oids
#:format-vector-row
#:reset-sequences)
#:reset-sequences
#:comment-on-tables-and-columns)
(:export #:mysql-connection
#:copy-mysql
#:*decoding-as*
#:*mysql-default-cast-rules*
#:with-mysql-connection
#:map-rows
@ -450,7 +540,6 @@
#:with-pgsql-transaction
#:pgsql-execute
#:pgsql-execute-with-timing
#:apply-identifier-case
#:create-tables
#:truncate-tables
#:format-pgsql-column
@ -459,7 +548,8 @@
#:format-pgsql-create-index
#:create-indexes-in-kernel
#:set-table-oids
#:reset-sequences)
#:reset-sequences
#:comment-on-tables-and-columns)
(:export #:sqlite-connection
#:copy-sqlite
#:*sqlite-default-cast-rules*
@ -480,10 +570,10 @@
#:pgsql-execute
#:pgsql-execute-with-timing
#:pgsql-connect-and-execute-with-timing
#:apply-identifier-case
#:list-tables-and-fkeys
#:list-table-oids
#:create-tables
#:create-views
#:truncate-tables
#:format-pgsql-column
#:format-extra-type
@ -532,7 +622,8 @@
#:with-pgsql-transaction
#:pgsql-execute
#:pgconn-use-ssl
#:pgconn-table-name)
#:pgconn-table-name
#:make-table)
(:import-from #:pgloader.csv
#:csv-connection
#:specs
@ -546,6 +637,7 @@
#:*cast-rules*)
(:import-from #:pgloader.mysql
#:mysql-connection
#:*decoding-as*
#:*mysql-default-cast-rules*)
(:import-from #:pgloader.mssql
#:mssql-connection
@ -612,8 +704,7 @@
(:import-from #:pgloader.pgsql
#:with-pgsql-connection
#:with-schema
#:list-reserved-keywords
#:apply-identifier-case)
#:list-reserved-keywords)
(:export #:*version-string*
#:*state*
#:*fd-path-root*

11
src/parsers/command-copy.lisp
View File

@ -129,11 +129,12 @@
(source
(make-instance 'pgloader.copy:copy-copy
:target-db ,pg-db-conn
:source source-db
:target ',(pgconn-table-name pg-db-conn)
:encoding ,encoding
:fields ',fields
:columns ',columns
:source source-db
:target (create-table
',(pgconn-table-name pg-db-conn))
:encoding ,encoding
:fields ',fields
:columns ',columns
,@(remove-batch-control-option
options :extras '(:truncate
:drop-indexes

3
src/parsers/command-csv.lisp
View File

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

6
src/parsers/command-db-uri.lisp
View File

@ -118,6 +118,10 @@
(defrule dsn-table-name (or qualified-table-name maybe-quoted-namestring)
(:lambda (name)
;; we can't make a table instance yet here, because for that we need to
;; apply-identifier-case on it, and that requires to have initialized
;; the *pgsql-reserved-keywords*, and we can't do that before parsing
;; the target database connection string, can we?
(cons :table-name name)))
(defrule dsn-option-table-name (and (? (and "tablename" "="))
@ -228,6 +232,6 @@
(defun pgsql-connection-bindings (pg-db-uri gucs)
"Generate the code needed to set PostgreSQL connection bindings."
`((*pg-settings* ',gucs)
(pgloader.pgsql::*pgsql-reserved-keywords*
(*pgsql-reserved-keywords*
(pgloader.pgsql:list-reserved-keywords ,pg-db-uri))))

5
src/parsers/command-dbf.lisp
View File

@ -96,7 +96,8 @@
(let* (,@(pgsql-connection-bindings pg-db-conn gucs)
,@(batch-control-bindings options)
,@(identifier-case-binding options)
(table-name ',(pgconn-table-name pg-db-conn))
(table (create-table
',(pgconn-table-name pg-db-conn)))
(source-db (with-stats-collection ("fetch" :section :pre)
(expand (fetch-file ,dbf-db-conn))))
(source
@ -104,7 +105,7 @@
:target-db ,pg-db-conn
:encoding ,encoding
:source-db source-db
:target table-name)))
:target table)))
,(sql-code-block pg-db-conn :pre before "before load")

3
src/parsers/command-fixed.lisp
View File

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

2
src/parsers/command-ixf.lisp
View File

@ -84,7 +84,7 @@
,@(batch-control-bindings options)
,@(identifier-case-binding options)
(timezone (getf ',options :timezone))
(table-name ',(pgconn-table-name pg-db-conn))
(table-name (create-table ',(pgconn-table-name pg-db-conn)))
(source-db (with-stats-collection ("fetch" :section :pre)
(expand (fetch-file ,ixf-db-conn))))
(source

1
src/parsers/command-mssql.lisp
View File

@ -165,6 +165,7 @@
(pgloader.mssql:copy-database source
:including ',including
:excluding ',excluding
:set-table-oids t
,@(remove-batch-control-option options))
,(sql-code-block pg-db-conn :post after "after load"))))

3
src/parsers/command-mysql.lisp
View File

@ -168,6 +168,7 @@
`(lambda ()
(let* ((*default-cast-rules* ',*mysql-default-cast-rules*)
(*cast-rules* ',casts)
(*decoding-as* ',decoding-as)
,@(pgsql-connection-bindings pg-db-conn gucs)
,@(batch-control-bindings options)
,@(identifier-case-binding options)
@ -181,8 +182,8 @@
(pgloader.mysql:copy-database source
:including ',incl
:excluding ',excl
:decoding-as ',decoding-as
:materialize-views ',views
:set-table-oids t
,@(remove-batch-control-option options))
,(sql-code-block pg-db-conn :post after "after load"))))

35
src/pgsql/pgsql.lisp
View File

@ -8,7 +8,7 @@
;;; COPY protocol, and retry the batch avoiding known bad rows (from parsing
;;; COPY error messages) in case some data related conditions are signaled.
;;;
(defun copy-batch (table-name columns batch batch-rows
(defun copy-batch (table columns batch batch-rows
&key (db pomo:*database*))
"Copy current *writer-batch* into TABLE-NAME."
(handler-case
@ -16,7 +16,8 @@
;; We need to keep a copy of the rows we send through the COPY
;; protocol to PostgreSQL to be able to process them again in case
;; of a data error being signaled, that's the BATCH here.
(let ((copier (cl-postgres:open-db-writer db table-name columns)))
(let* ((table-name (format-table-name table))
(copier (cl-postgres:open-db-writer db table-name columns)))
(unwind-protect
(loop :for i :below batch-rows
:for copy-string := (aref batch i)
@ -33,35 +34,35 @@
cl-postgres-error::internal-error
cl-postgres-error::insufficient-resources
cl-postgres-error::program-limit-exceeded) (condition)
(retry-batch table-name columns batch batch-rows condition))))
(retry-batch table columns batch batch-rows condition))))
;;;
;;; We receive fully prepared batch from an lparallel queue, push their
;;; content down to PostgreSQL, handling any data related errors in the way.
;;;
(defun copy-from-queue (pgconn table-name queue
(defun copy-from-queue (pgconn table queue
&key
columns
(truncate t)
disable-triggers)
"Fetch from the QUEUE messages containing how many rows are in the
*writer-batch* for us to send down to PostgreSQL, and when that's done
update stats."
(let ((seconds 0))
(with-pgsql-connection (pgconn)
(with-schema (unqualified-table-name table-name)
(with-schema (unqualified-table-name table)
(with-disabled-triggers (unqualified-table-name
:disable-triggers disable-triggers)
(log-message :info "pgsql:copy-from-queue[~a]: ~a ~a"
(lp:kernel-worker-index) table-name columns)
(lp:kernel-worker-index)
(format-table-name table)
columns)
(loop
:for (mesg batch read oversized?) := (lq:pop-queue queue)
:until (eq :end-of-data mesg)
:for (rows batch-seconds) :=
(let ((start-time (get-internal-real-time)))
(list (copy-batch (apply-identifier-case unqualified-table-name)
columns batch read)
(list (copy-batch table columns batch read)
(elapsed-time-since start-time)))
:do (progn
;; The SBCL implementation needs some Garbage Collection
@ -74,13 +75,14 @@
rows
batch-seconds
oversized?)
(update-stats :data table-name :rows rows)
(update-stats :data table :rows rows)
(incf seconds batch-seconds))))))
(update-stats :data table-name :ws seconds)
(update-stats :data table :ws seconds)
(log-message :debug "Writer[~a] for ~a is done in ~6$s"
(lp:kernel-worker-index) table-name seconds)
(list :writer table-name seconds)))
(lp:kernel-worker-index)
(format-table-name table) seconds)
(list :writer table seconds)))
;;;
;;; Compute how many rows we're going to try loading next, depending on
@ -133,7 +135,7 @@
;;;
;;; The main retry batch function.
;;;
(defun retry-batch (table-name columns batch batch-rows condition
(defun retry-batch (table columns batch batch-rows condition
&optional (current-batch-pos 0)
&aux (nb-errors 0))
"Batch is a list of rows containing at least one bad row, the first such
@ -152,7 +154,7 @@
(when (= current-batch-pos next-error)
(log-message :info "error recovery at ~d/~d, processing bad row"
(+ 1 next-error) batch-rows)
(process-bad-row table-name condition (aref batch current-batch-pos))
(process-bad-row table condition (aref batch current-batch-pos))
(incf current-batch-pos)
(incf nb-errors))
@ -161,7 +163,8 @@
(when (< 0 current-batch-rows)
(handler-case
(with-pgsql-transaction (:database pomo:*database*)
(let* ((stream
(let* ((table-name (format-table-name table))
(stream
(cl-postgres:open-db-writer pomo:*database*
table-name columns)))

17
src/pgsql/queries.lisp
View File

@ -237,16 +237,16 @@
;; unless (eq attnum :NULL)
;; collect attnum)))))
(defun list-tables-and-fkeys (&optional schema)
(defun list-tables-and-fkeys (&optional schema-name)
"Yet another table listing query."
(loop for (relname fkeys) in (pomo:query (format nil "
(loop :for (relname fkeys) :in (pomo:query (format nil "
select relname, array_to_string(array_agg(conname), ',')
from pg_class c
join pg_namespace n on n.oid = c.relnamespace
left join pg_constraint co on c.oid = co.conrelid
where contype = 'f' and nspname = ~:[current_schema()~;'~a'~]
group by relname;" schema schema))
collect (cons relname (sq:split-sequence #\, fkeys))))
group by relname;" schema-name schema-name))
:collect (cons relname (sq:split-sequence #\, fkeys))))
(defun list-columns-query (table-name &optional schema)
"Returns the list of columns for table TABLE-NAME in schema SCHEMA, and
@ -268,10 +268,10 @@
:in (list-columns-query unqualified-table-name schema)
:collect name))))
(defun list-indexes (table-name)
(defun list-indexes (table)
"List all indexes for TABLE-NAME in SCHEMA. A PostgreSQL connection must
be already established when calling that function."
(with-schema (unqualified-table-name table-name)
(with-schema (unqualified-table-name table)
(loop :for (index-name table-name table-oid primary unique sql conname condef)
:in (pomo:query (format nil "
select i.relname,
@ -286,8 +286,7 @@ select i.relname,
join pg_class i ON i.oid = x.indexrelid
left join pg_constraint c ON c.conindid = i.oid
where indrelid = '~@[~a.~]~a'::regclass"
(when (typep table-name 'cons)
(car table-name))
(table-schema table)
unqualified-table-name))
:collect (make-pgsql-index :name index-name
:table-name table-name
@ -426,7 +425,7 @@ select i.relname,
(when tables
(pomo:execute
(format nil "create temp table reloids(oid) as values ~{('~a'::regclass)~^,~}"
(mapcar #'apply-identifier-case tables))))
(mapcar #'format-table-name tables))))
(handler-case
(let ((sql (format nil "

341
src/pgsql/schema.lisp
View File

@ -3,46 +3,6 @@
;;;
(in-package pgloader.pgsql)
(defun quoted-p (s)
"Return true if s is a double-quoted string"
(and (eq (char s 0) #\")
(eq (char s (- (length s) 1)) #\")))
(defun apply-identifier-case (identifier)
"Return given IDENTIFIER with CASE handled to be PostgreSQL compatible."
(let* ((lowercase-identifier (cl-ppcre:regex-replace-all
"[^a-zA-Z0-9.]" (string-downcase identifier) "_"))
(*identifier-case*
;; we might need to force to :quote in some cases
;;
;; http://www.postgresql.org/docs/9.1/static/sql-syntax-lexical.html
;;
;; SQL identifiers and key words must begin with a letter (a-z, but
;; also letters with diacritical marks and non-Latin letters) or an
;; underscore (_).
(cond ((quoted-p identifier)
:none)
((not (cl-ppcre:scan "^[A-Za-z_][A-Za-z0-9_$]*$" identifier))
:quote)
((member lowercase-identifier *pgsql-reserved-keywords*
:test #'string=)
(progn
;; we need to both downcase and quote here
(when (eq :downcase *identifier-case*)
(setf identifier lowercase-identifier))
:quote))
;; in other cases follow user directive
(t *identifier-case*))))
(ecase *identifier-case*
(:downcase lowercase-identifier)
(:quote (format nil "\"~a\""
(cl-ppcre:regex-replace-all "\"" identifier "\"\"")))
(:none identifier))))
;;;
;;; Some parts of the logic here needs to be specialized depending on the
;;; source type, such as SQLite or MySQL. To do so, sources must define
@ -61,18 +21,18 @@
column, or nil of none is required. If no special extra type is ever
needed, it's allowed not to specialize this generic into a method."))
(defmethod format-pgsql-column ((col pgsql-column))
"Return a string representing the PostgreSQL column definition."
(let* ((column-name
(apply-identifier-case (pgsql-column-name col)))
(type-definition
(format nil
"~a~@[~a~]~:[~; not null~]~@[ default ~a~]"
(pgsql-column-type-name col)
(pgsql-column-type-mod col)
(pgsql-column-nullable col)
(pgsql-column-default col))))
(format nil "~a ~22t ~a" column-name type-definition)))
;; (defmethod format-pgsql-column ((col pgsql-column))
;; "Return a string representing the PostgreSQL column definition."
;; (let* ((column-name
;; (apply-identifier-case (pgsql-column-name col)))
;; (type-definition
;; (format nil
;; "~a~@[~a~]~:[~; not null~]~@[ default ~a~]"
;; (pgsql-column-type-name col)
;; (pgsql-column-type-mod col)
;; (pgsql-column-nullable col)
;; (pgsql-column-default col))))
;; (format nil "~a ~22t ~a" column-name type-definition)))
(defmethod format-extra-type ((col T) &key include-drop)
"The default `format-extra-type' implementation returns an empty list."
@ -96,25 +56,17 @@
(defmethod format-pgsql-create-fkey ((fk pgsql-fkey))
"Generate the PostgreSQL statement to rebuild a MySQL Foreign Key"
(let* ((constraint-name (apply-identifier-case (pgsql-fkey-name fk)))
(table-name (apply-identifier-case (pgsql-fkey-table-name fk)))
(fkey-columns (mapcar (lambda (column-name)
(apply-identifier-case column-name))
(pgsql-fkey-columns fk)))
(foreign-table (apply-identifier-case (pgsql-fkey-foreign-table fk)))
(foreign-columns (mapcar #'apply-identifier-case
(pgsql-fkey-foreign-columns fk))))
(format nil
"ALTER TABLE ~a ADD CONSTRAINT ~a FOREIGN KEY(~{~a~^,~}) REFERENCES ~a(~{~a~^,~})~:[~*~; ON UPDATE ~a~]~:[~*~; ON DELETE ~a~]"
table-name
constraint-name
fkey-columns
foreign-table
foreign-columns
(pgsql-fkey-update-rule fk)
(pgsql-fkey-update-rule fk)
(pgsql-fkey-delete-rule fk)
(pgsql-fkey-delete-rule fk))))
(format nil
"ALTER TABLE ~a ADD CONSTRAINT ~a FOREIGN KEY(~{~a~^,~}) REFERENCES ~a(~{~a~^,~})~:[~*~; ON UPDATE ~a~]~:[~*~; ON DELETE ~a~]"
(pgsql-fkey-table-name fk)
(pgsql-fkey-name fk) ; constraint name
(pgsql-fkey-columns fk)
(pgsql-fkey-foreign-table fk)
(pgsql-fkey-foreign-columns fk)
(pgsql-fkey-update-rule fk)
(pgsql-fkey-update-rule fk)
(pgsql-fkey-delete-rule fk)
(pgsql-fkey-delete-rule fk)))
(defmethod format-pgsql-drop-fkey ((fk pgsql-fkey) &key all-pgsql-fkeys)
"Generate the PostgreSQL statement to rebuild a MySQL Foreign Key"
@ -127,118 +79,128 @@
;; alter table if exists ... drop constraint if exists ...
(format nil "ALTER TABLE ~a DROP CONSTRAINT ~a" table-name constraint-name))))
(defun drop-pgsql-fkeys (all-fkeys)
(defun drop-pgsql-fkeys (catalog)
"Drop all Foreign Key Definitions given, to prepare for a clean run."
(let ((all-pgsql-fkeys (list-tables-and-fkeys)))
(loop for (table-name . fkeys) in all-fkeys
do
(loop for fkey in fkeys
for sql = (format-pgsql-drop-fkey fkey
:all-pgsql-fkeys all-pgsql-fkeys)
when sql
do
(log-message :notice "~a;" sql)
(pgsql-execute sql)))))
(loop :for table :in (table-list catalog)
:do
(loop :for fkey :in (table-fkey-list table)
:for sql := (format-pgsql-drop-fkey fkey
:all-pgsql-fkeys all-pgsql-fkeys)
:when sql
:do
(log-message :notice "~a;" sql)
(pgsql-execute sql)))))
(defun create-pgsql-fkeys (pgconn all-fkeys
(defun create-pgsql-fkeys (catalog
&key
(section :post)
(label "Foreign Keys"))
"Actually create the Foreign Key References that where declared in the
MySQL database"
(new-label section label)
(loop for (table-name . fkeys) in all-fkeys
do (loop for fkey in fkeys
for sql = (format-pgsql-create-fkey fkey)
do
(log-message :notice "~a;" sql)
(pgsql-execute-with-timing section label sql))))
(with-stats-collection (label :section section :use-result-as-rows t)
(loop :for table :in (table-list catalog)
:sum (loop :for fkey :in (table-fkey-list table)
:for sql := (format-pgsql-create-fkey fkey)
:do (progn ; for indentation purposes
(log-message :notice "~a;" sql)
(pgsql-execute-with-timing section label sql))
:count t))))
;;;
;;; Table schema rewriting support
;;;
(defun create-table-sql (table-name cols &key if-not-exists)
(defun create-table-sql (table &key if-not-exists)
"Return a PostgreSQL CREATE TABLE statement from given COLS.
Each element of the COLS list is expected to be of a type handled by the
`format-pgsql-column' generic function."
(with-output-to-string (s)
(let ((table-name (apply-identifier-case table-name)))
(format s "CREATE TABLE~:[~; IF NOT EXISTS~] ~a ~%(~%"
if-not-exists
table-name))
(format s "CREATE TABLE~:[~; IF NOT EXISTS~] ~a ~%(~%"
if-not-exists
(format-table-name table))
(loop
for (col . last?) on cols
for pg-coldef = (format-pgsql-column col)
do (format s " ~a~:[~;,~]~%" pg-coldef last?))
:for (col . last?) :on (table-column-list table)
:for pg-coldef := (format-column col)
:do (format s " ~a~:[~;,~]~%" pg-coldef last?))
(format s ");~%")))
(defun drop-table-if-exists-sql (table-name)
(defun drop-table-if-exists-sql (table)
"Return the PostgreSQL DROP TABLE IF EXISTS statement for TABLE-NAME."
(let ((table-name (apply-identifier-case table-name)))
(format nil "DROP TABLE IF EXISTS ~a~% CASCADE;" table-name)))
(format nil "DROP TABLE IF EXISTS ~a CASCADE;" (format-table-name table)))
(defun create-table-sql-list (all-columns
(defun create-table-sql-list (table-list
&key
if-not-exists
include-drop)
"Return the list of CREATE TABLE statements to run against PostgreSQL.
The ALL-COLUMNS parameter must be a list of alist associations where the
car is the table-name (a string) and the cdr is a column list. Each
element of the column list is expected to be of a type handled by the
`format-pgsql-column' generic function, such as `pgsql-column'."
"Return the list of CREATE TABLE statements to run against PostgreSQL."
(loop
for (table-name . cols) in all-columns
for extra-types = (loop for col in cols
append (format-extra-type col
:include-drop include-drop))
:for table :in table-list
:for cols := (table-column-list table)
:for fields := (table-field-list table)
:for extra-types := (loop :for field :in fields
:append (format-extra-type
field :include-drop include-drop))
when include-drop
collect (drop-table-if-exists-sql table-name)
:when include-drop
:collect (drop-table-if-exists-sql table)
when extra-types append extra-types
:when extra-types :append extra-types
collect (create-table-sql table-name cols :if-not-exists if-not-exists)))
:collect (create-table-sql table :if-not-exists if-not-exists)))
(defun create-tables (all-columns
&key
(defun create-table-list (table-list
&key
if-not-exists
include-drop
(client-min-messages :notice))
"Create all tables in database dbname in PostgreSQL."
(loop
:for sql :in (create-table-sql-list table-list
:if-not-exists if-not-exists
:include-drop include-drop)
:count (not (null sql)) :into nb-tables
:when sql
:do (progn
(log-message :info "~a" sql)
(pgsql-execute sql :client-min-messages client-min-messages))
:finally (return nb-tables)))
(defun create-tables (catalog
&key
if-not-exists
include-drop
(client-min-messages :notice))
"Create all tables in database dbname in PostgreSQL.
"Create all tables from the given database CATALOG."
(create-table-list (table-list catalog)
:if-not-exists if-not-exists
:include-drop include-drop
:client-min-messages client-min-messages))
The ALL-COLUMNS parameter must be a list of alist associations where the
car is the table-name (a string) and the cdr is a column list. Each
element of the column list is expected to be of a type handled by the
`format-pgsql-column' generic function, such as `pgsql-column'."
(loop
for sql in (create-table-sql-list all-columns
:if-not-exists if-not-exists
:include-drop include-drop)
count (not (null sql)) into nb-tables
when sql
do
(log-message :info "~a" sql)
(pgsql-execute sql :client-min-messages client-min-messages)
finally (return nb-tables)))
(defun create-views (catalog
&key
if-not-exists
include-drop
(client-min-messages :notice))
"Create all tables from the given database CATALOG."
(create-table-list (view-list catalog)
:if-not-exists if-not-exists
:include-drop include-drop
:client-min-messages client-min-messages))
(defun truncate-tables (pgconn table-name-list)
(defun truncate-tables (pgconn catalog-or-table)
"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~^,~};"
(mapcar #'process-table-name table-name-list))))
(log-message :notice "~a" sql)
(pomo:execute sql)))))
(let ((sql
(format nil "TRUNCATE ~{~a~^,~};"
(mapcar #'format-table-name
(etypecase catalog-or-table
(catalog (table-list catalog-or-table))
(schema (table-list catalog-or-table))
(table (list catalog-or-table)))))))
(log-message :notice "~a" sql)
(pomo:execute sql))))
(defun disable-triggers (table-name)
"Disable triggers on TABLE-NAME. Needs to be called with a PostgreSQL
@ -301,10 +263,7 @@
(format nil "idx_~a_~a"
(pgsql-index-table-oid index)
(pgsql-index-name index))))
(table-name (apply-identifier-case (pgsql-index-table-name index)))
(index-name (apply-identifier-case index-name))
(cols (mapcar #'apply-identifier-case (pgsql-index-columns index))))
(index-name (apply-identifier-case index-name)))
(cond
((or (pgsql-index-primary index)
(and (pgsql-index-condef index) (pgsql-index-unique index)))
@ -313,36 +272,39 @@
;; LOCK on the table before we have the index done already
(or (pgsql-index-sql index)
(format nil "CREATE UNIQUE INDEX ~a ON ~a (~{~a~^, ~});"
index-name table-name cols))
index-name
(pgsql-index-table-name index)
(pgsql-index-columns index)))
(format nil
"ALTER TABLE ~a ADD ~a USING INDEX ~a;"
table-name
(pgsql-index-table-name index)
(cond ((pgsql-index-primary index) "PRIMARY KEY")
((pgsql-index-unique index) "UNIQUE"))
index-name)))
((pgsql-index-condef index)
(format nil "ALTER TABLE ~a ADD ~a;"
table-name (pgsql-index-condef index)))
(pgsql-index-table-name index)
(pgsql-index-condef index)))
(t
(or (pgsql-index-sql index)
(format nil "CREATE~:[~; UNIQUE~] INDEX ~a ON ~a (~{~a~^, ~});"
(pgsql-index-unique index)
index-name
table-name
cols))))))
(pgsql-index-table-name index)
(pgsql-index-columns index)))))))
(defmethod format-pgsql-drop-index ((index pgsql-index))
"Generate the PostgreSQL statement to DROP the index."
(let* ((table-name (apply-identifier-case (pgsql-index-table-name index)))
(index-name (apply-identifier-case (pgsql-index-name index))))
(let* ((index-name (apply-identifier-case (pgsql-index-name index))))
(cond ((pgsql-index-conname index)
;; here always quote the constraint name, currently the name
;; comes from one source only, the PostgreSQL database catalogs,
;; so don't question it, quote it.
(format nil "ALTER TABLE ~a DROP CONSTRAINT ~s;"
table-name (pgsql-index-conname index)))
(pgsql-index-table-name index)
(pgsql-index-conname index)))
(t
(format nil "DROP INDEX ~a;" index-name)))))
@ -375,22 +337,22 @@
;;;
;;; Protect from non-unique index names
;;;
(defun set-table-oids (all-indexes)
(defun set-table-oids (catalog)
"MySQL allows using the same index name against separate tables, which
PostgreSQL forbids. To get unicity in index names without running out of
characters (we are allowed only 63), we use the table OID instead.
This function grabs the table OIDs in the PostgreSQL database and update
the definitions with them."
(let* ((table-names (mapcar #'apply-identifier-case
(mapcar #'car all-indexes)))
(let* ((table-names (mapcar #'format-table-name (table-list catalog)))
(table-oids (pgloader.pgsql:list-table-oids table-names)))
(loop for (table-name-raw . indexes) in all-indexes
for table-name = (apply-identifier-case table-name-raw)
for table-oid = (cdr (assoc table-name table-oids :test #'string=))
unless table-oid do (error "OID not found for ~s." table-name)
do (loop for index in indexes
do (setf (pgsql-index-table-oid index) table-oid)))))
(loop :for table :in (table-list catalog)
:for table-name := (format-table-name table)
:for table-oid := (cdr (assoc table-name table-oids :test #'string=))
:unless table-oid :do (error "OID not found for ~s." table-name)
:do (setf (table-oid table) table-oid)
(loop :for index :in (table-index-list table)
:do (setf (pgsql-index-table-oid index) table-oid)))))
;;;
;;; Drop indexes before loading
@ -459,10 +421,53 @@
;;;
;;; Sequences
;;;
(defun reset-sequences (table-names &key pgconn (section :post))
(defun reset-sequences (catalog &key pgconn (section :post))
"Reset all sequences created during this MySQL migration."
(log-message :notice "Reset sequences")
(with-stats-collection ("Reset Sequences"
:use-result-as-rows t
:section section)
(reset-all-sequences pgconn :tables table-names)))
(reset-all-sequences pgconn :tables (table-list catalog))))
;;;
;;; Comments
;;;
(defun comment-on-tables-and-columns (catalog)
"Install comments on tables and columns from CATALOG."
(let* ((quote
;; just something improbably found in a table comment, to use as
;; dollar quoting, and generated at random at that.
;;
;; because somehow it appears impossible here to benefit from
;; the usual SQL injection protection offered by the Extended
;; Query Protocol from PostgreSQL.
(concatenate 'string
(map 'string #'code-char
(loop :repeat 5
:collect (+ (random 26) (char-code #\A))))
"_"
(map 'string #'code-char
(loop :repeat 5
:collect (+ (random 26) (char-code #\A)))))))
(with-stats-collection ("Install comments"
:use-result-as-rows t
:section :post)
(loop :for table :in (table-list catalog)
:for sql := (when (table-comment table)
(format nil "comment on table ~a is $~a$~a$~a$"
(table-name table)
quote (table-comment table) quote))
:count (when sql
(log-message :notice "~a" sql)
(pgsql-execute-with-timing :post "Comments" sql))
:sum (loop :for column :in (table-column-list table)
:for sql := (when (column-comment column)
(format nil "comment on column ~a.~a is $~a$~a$~a$"
(table-name table)
(column-name column)
quote (column-comment column) quote))
:count (when sql
(log-message :notice "~a;" sql)
(pgsql-execute-with-timing :post "Comments" sql)))))))

45
src/schema.lisp
View File

@ -1,45 +0,0 @@
;;;
;;; Generic API for pgloader sources
;;;
(in-package :pgloader.schema)
(defmacro push-to-end (item place)
`(setf ,place (nconc ,place (list ,item))))
;;;
;;; TODO: stop using anonymous data structures for database catalogs,
;;; currently list of alists of lists... the madness has found its way in
;;; lots of places tho.
;;;
;;;
;;; A database catalog is a list of schema each containing a list of tables,
;;; each being a list of columns.
;;;
;;; Column structures details depend on the specific source type and are
;;; implemented in each source separately.
;;;
(defstruct schema name tables)
(defstruct table schema name qname columns)
;;;
;;; Still lacking round tuits here, so for the moment the representation of
;;; a table name is either a string or a cons built from schema and
;;; table-name.
;;;
(defmacro with-schema ((var table-name) &body body)
"When table-name is a CONS, SET search_path TO its CAR and return its CDR,
otherwise just return the TABLE-NAME. A PostgreSQL connection must be
established when calling this function."
`(let ((,var
(typecase ,table-name
(cons (let ((sql (format nil "SET search_path TO ~a;"
(car ,table-name))))
(log-message :notice "~a" sql)
(pgloader.pgsql:pgsql-execute sql)
(cdr ,table-name)))
(string ,table-name))))
,@body))

37
src/sources/common/api.lisp
View File

@ -138,3 +138,40 @@
(:documentation "Process rows from a given input stream."))
;;;
;;; Class hierarchy allowing to share features for database sources, where
;;; we do introspection to prepare an internal catalog and then cast that
;;; catalog to PostgreSQL before copying the data over.
;;;
(defclass db-copy (copy) ()
(:documentation "pgloader Database Data Source (MySQL, SQLite, MS SQL)."))
(defgeneric fetch-metadata (db-copy catalog
&key
materialize-views
only-tables
create-indexes
foreign-keys
including
excluding))
(defgeneric prepare-pgsql-database (db-copy catalog
&key
materialize-views
foreign-keys
include-drop)
(:documentation "Prepare the target PostgreSQL database."))
(defgeneric cleanup (db-copy catalog &key materialize-views)
(:documentation "Clean-up after prepare-pgsql-database failure."))
(defgeneric complete-pgsql-database (db-copy catalog pkeys
&key
data-only
foreign-keys
reset-sequences)
(:documentation "Alter load duties for database sources copy support."))
(defgeneric instanciate-table-copy-object (db-copy table)
(:documentation "Create an new instance for copying TABLE data."))

93
src/sources/common/casting-rules.lisp
View File

@ -72,20 +72,8 @@
(or (null ai-s-p) (eq auto-increment rule-source-auto-increment)))
(list :using using :target rule-target))))))
(defun format-pgsql-default-value (default &optional using-cast-fn)
"Returns suitably quoted default value for CREATE TABLE command."
(cond
((null default) "NULL")
((and (stringp default) (string= "NULL" default)) default)
((and (stringp default) (string= "CURRENT_TIMESTAMP" default)) default)
(t
;; apply the transformation function to the default value
(if using-cast-fn (format-pgsql-default-value
(funcall using-cast-fn default))
(format nil "'~a'" default)))))
(defun format-pgsql-type (source target using)
"Returns a string suitable for a PostgreSQL type definition"
(defun make-pgsql-type (source target using)
"Returns a COLUMN struct suitable for a PostgreSQL type definition"
(destructuring-bind (&key ((:table-name source-table-name))
((:column-name source-column-name))
((:type source-type))
@ -112,30 +100,30 @@
(when source-typemod
(destructuring-bind (a . b) source-typemod
(format nil "(~a~:[~*~;,~a~])" a b b)))))
(format nil
"~a~:[~*~;~a~]~:[~; not null~]~:[~; default ~a~]"
type-name
(and source-typemod (not drop-typemod))
pg-typemod
(and source-not-null (not drop-not-null))
(and source-default (not drop-default))
(format-pgsql-default-value source-default using))))
(make-column :name (apply-identifier-case source-column-name)
:type-name type-name
:type-mod (when (and source-typemod (not drop-typemod))
pg-typemod)
:nullable (not (and source-not-null (not drop-not-null)))
:default (when (and source-default (not drop-default))
(format-default-value source-default using))
:transform using)))
;; NO MATCH
;;
;; prefer char(24) over just char, that is the column type over the
;; data type.
(format nil "~a~:[~; not null~]~:[~; default ~a~]"
source-ctype
source-not-null
source-default
(format-pgsql-default-value source-default using)))))
(make-column :name (apply-identifier-case source-column-name)
:type-name source-ctype
:nullable (not source-not-null)
:default (format-default-value source-default using)
:transform using))))
(defun apply-casting-rules (dtype ctype default nullable extra
&key
table-name column-name ; ENUM support
(rules (append *cast-rules*
*default-cast-rules*)))
(defun apply-casting-rules (table-name column-name
dtype ctype default nullable extra
&key
(rules (append *cast-rules*
*default-cast-rules*)))
"Apply the given RULES to the MySQL SOURCE type definition"
(let* ((typemod (parse-column-typemod dtype ctype))
(not-null (string-equal nullable "NO"))
@ -150,30 +138,19 @@
:auto-increment ,auto-increment)))
(let (first-match-using)
(loop
for rule in rules
for (target using) = (destructuring-bind (&key target using)
(cast-rule-matches rule source)
(list target using))
do (when (and (null target) using (null first-match-using))
(setf first-match-using using))
until target
finally
(log-message :info "CAST ~a.~a ~a [~s, ~:[NULL~;NOT NULL~]~:[~*~;, ~a~]] TO ~s"
table-name column-name ctype default
(string= "NO" nullable)
(string/= "" extra) extra
(format-pgsql-type source target using))
(return
(list :transform-fn (or first-match-using using)
:pgtype (format-pgsql-type source target using)))))))
:for rule :in rules
:for (target using) := (destructuring-bind (&key target using)
(cast-rule-matches rule source)
(list target using))
:do (when (and (null target) using (null first-match-using))
(setf first-match-using using))
:until target
:finally (let ((coldef (make-pgsql-type source target using)))
(log-message :info "CAST ~a.~a ~a [~s, ~:[NULL~;NOT NULL~]~:[~*~;, ~a~]] TO ~s~@[ USING ~a~]"
table-name column-name ctype default
(string= "NO" nullable)
(string/= "" extra) extra
(format-column coldef)
using)
(return coldef))))))
(defun cast (table-name column-name dtype ctype default nullable extra)
"Convert a MySQL datatype to a PostgreSQL datatype.
DYTPE is the MySQL data_type and CTYPE the MySQL column_type, for example
that would be int and int(7) or varchar and varchar(25)."
(destructuring-bind (&key pgtype transform-fn &allow-other-keys)
(apply-casting-rules dtype ctype default nullable extra
:table-name table-name
:column-name column-name)
(values pgtype transform-fn)))

281
src/sources/common/db-methods.lisp Normal file
View File

@ -0,0 +1,281 @@
;;;
;;; Generic API for pgloader sources
;;; Methods for database source types (with introspection)
;;;
(in-package :pgloader.sources)
;;;
;;; Prepare the PostgreSQL database before streaming the data into it.
;;;
(defmethod prepare-pgsql-database ((copy db-copy)
(catalog catalog)
&key
create-schemas
set-table-oids
materialize-views
foreign-keys
include-drop)
"Prepare the target PostgreSQL database: create tables casting datatypes
from the MySQL definitions, prepare index definitions and create target
tables for materialized views.
That function mutates index definitions in ALL-INDEXES."
(log-message :notice "~:[~;DROP then ~]CREATE TABLES" include-drop)
(with-stats-collection ("create, drop" :use-result-as-rows t :section :pre)
(with-pgsql-transaction (:pgconn (target-db copy))
;; we need to first drop the Foreign Key Constraints, so that we
;; can DROP TABLE when asked
(when (and foreign-keys include-drop)
(drop-pgsql-fkeys catalog))
(when create-schemas
(loop :for schema :in (catalog-schema-list catalog)
:do (when create-schemas
(let ((sql (format nil "CREATE SCHEMA ~a;" schema)))
(log-message :notice "~a" sql)
(pgsql-execute sql)))))
(create-tables catalog :include-drop include-drop)
;; Some database sources allow the same index name being used
;; against several tables, so we add the PostgreSQL table OID in the
;; index name, to differenciate. Set the table oids now.
(when set-table-oids
(set-table-oids catalog))
;; We might have to MATERIALIZE VIEWS
(when materialize-views
(create-views catalog :include-drop include-drop)))))
(defmethod cleanup ((copy db-copy) (catalog catalog) &key materialize-views)
"In case anything wrong happens at `prepare-pgsql-database' step, this
function will be called to clean-up the mess left behind, if any."
(declare (ignorable materialize-views))
t)
(defmethod complete-pgsql-database ((copy db-copy)
(catalog catalog)
pkeys
&key
data-only
foreign-keys
reset-sequences)
"After loading the data into PostgreSQL, we can now reset the sequences
and declare foreign keys."
;;
;; Now Reset Sequences, the good time to do that is once the whole data
;; has been imported and once we have the indexes in place, as max() is
;; able to benefit from the indexes. In particular avoid doing that step
;; while CREATE INDEX statements are in flight (avoid locking).
;;
(when reset-sequences
(reset-sequences catalog :pgconn (clone-connection (target-db copy))))
(with-pgsql-transaction (:pgconn (clone-connection (target-db copy)))
;;
;; Turn UNIQUE indexes into PRIMARY KEYS now
;;
(unless data-only
(loop :for sql :in pkeys
:when sql
:do (progn
(log-message :notice "~a" sql)
(pgsql-execute-with-timing :post "Primary Keys" sql))))
;;
;; Foreign Key Constraints
;;
;; We need to have finished loading both the reference and the refering
;; tables to be able to build the foreign keys, so wait until all tables
;; and indexes are imported before doing that.
;;
(when (and foreign-keys (not data-only))
(create-pgsql-fkeys catalog))
;;
;; And now, comments on tables and columns.
;;
(log-message :notice "Comments")
(comment-on-tables-and-columns catalog)))
(defmethod instanciate-table-copy-object ((copy db-copy) (table table))
"Create an new instance for copying TABLE data."
(let* ((fields (table-field-list table))
(columns (table-column-list table))
(transforms (mapcar #'column-transform columns)))
(make-instance (class-of copy)
:source-db (clone-connection (source-db copy))
:target-db (clone-connection (target-db copy))
:source table
:target (table-name table)
:fields fields
:columns columns
:transforms transforms)))
;;;
;;; Generic enough implementation of the copy-database method.
;;;
(defmethod copy-database ((copy db-copy)
&key
(truncate nil)
(disable-triggers nil)
(data-only nil)
(schema-only nil)
(create-tables t)
(include-drop t)
(create-indexes t)
(index-names :uniquify)
(reset-sequences t)
(foreign-keys t)
only-tables
including
excluding
set-table-oids
materialize-views)
"Export database source data and Import it into PostgreSQL"
(let* ((copy-kernel (make-kernel 8))
(copy-channel (let ((lp:*kernel* copy-kernel)) (lp:make-channel)))
(catalog (fetch-metadata
copy
(make-catalog
:name (typecase (source-db copy)
(db-connection (db-name (source-db copy)))
(fd-connection (pathname-name
(fd-path (source-db copy))))))
:materialize-views materialize-views
:only-tables only-tables
:create-indexes create-indexes
:foreign-keys foreign-keys
:including including
:excluding excluding))
pkeys
(table-count 0)
(max-indexes (max-indexes-per-table catalog))
(idx-kernel (when (and max-indexes (< 0 max-indexes))
(make-kernel max-indexes)))
(idx-channel (when idx-kernel
(let ((lp:*kernel* idx-kernel))
(lp:make-channel)))))
;; cast the catalog into something PostgreSQL can work on
(cast catalog)
;; if asked, first drop/create the tables on the PostgreSQL side
(handler-case
(cond ((and (or create-tables schema-only) (not data-only))
(prepare-pgsql-database copy
catalog
:set-table-oids set-table-oids
:materialize-views materialize-views
:foreign-keys foreign-keys
:include-drop include-drop))
(truncate
(truncate-tables (target-db copy) catalog)))
;;
;; In case some error happens in the preparatory transaction, we
;; need to stop now and refrain from trying to load the data into
;; an incomplete schema.
;;
(cl-postgres:database-error (e)
(declare (ignore e)) ; a log has already been printed
(log-message :fatal "Failed to create the schema, see above.")
;; we might have some cleanup to do...
(cleanup copy catalog :materialize-views materialize-views)
(return-from copy-database)))
(loop
:for table :in (append (table-list catalog)
;; when materialized views are not supported,
;; view-list is empty here
(view-list catalog))
:do (let ((table-source (instanciate-table-copy-object copy table)))
(log-message :debug "TARGET: ~a" (target table-source))
(log-message :debug "TRANSFORMS(~a): ~s"
(format-table-name table)
(mapcar #'column-transform (table-column-list table)))
;; first COPY the data from source to PostgreSQL, using copy-kernel
(unless schema-only
(incf table-count)
(copy-from table-source
:kernel copy-kernel
:channel copy-channel
:disable-triggers disable-triggers))
;; Create the indexes for that table in parallel with the next
;; COPY, and all at once in concurrent threads to benefit from
;; PostgreSQL synchronous scan ability
;;
;; We just push new index build as they come along, if one
;; index build requires much more time than the others our
;; index build might get unsync: indexes for different tables
;; will get built in parallel --- not a big problem.
(when (and create-indexes (not data-only))
(let* ((indexes (table-index-list table))
(*preserve-index-names* (eq :preserve index-names)))
(alexandria:appendf
pkeys
(create-indexes-in-kernel (target-db copy)
indexes idx-kernel idx-channel))))))
;; now end the kernels
(end-kernels copy-kernel copy-channel idx-kernel idx-channel
table-count (count-indexes catalog))
;;
;; Complete the PostgreSQL database before handing over.
;;
(complete-pgsql-database copy
catalog
pkeys
:data-only data-only
:foreign-keys foreign-keys
:reset-sequences reset-sequences)
;;
;; Time to cleanup!
;;
(cleanup copy catalog :materialize-views materialize-views)))
;;;
;;; Lower level tools
;;;
(defun end-kernels (copy-kernel copy-channel
idx-kernel idx-channel
table-count index-count)
"Terminate the lparallel kernels, waiting for all threads."
(when copy-kernel
(let ((lp:*kernel* copy-kernel))
(with-stats-collection ("COPY Threads Completion" :section :post
:use-result-as-read t
:use-result-as-rows t)
(let ((workers-count (* 4 table-count)))
(loop :for tasks :below workers-count
:do (destructuring-bind (task table-name seconds)
(lp:receive-result copy-channel)
(log-message :debug "Finished processing ~a for ~s ~50T~6$s"
task table-name seconds)
(when (eq :writer task)
(update-stats :data table-name :secs seconds))))
(prog1
workers-count
(lp:end-kernel))))))
(when idx-kernel
(let ((lp:*kernel* idx-kernel))
;; wait until the indexes are done being built...
;; don't forget accounting for that waiting time.
(with-stats-collection ("Index Build Completion" :section :post
:use-result-as-read t
:use-result-as-rows t)
(loop :for count :below index-count
:count (lp:receive-result idx-channel)))
(lp:end-kernel))))

13
src/sources/common/methods.lisp
View File

@ -9,7 +9,7 @@
(defmethod queue-raw-data ((copy copy) queue)
"Stream data as read by the map-queue method on the COPY argument into QUEUE,
as given."
(log-message :debug "Reader started for ~a" (target copy))
(log-message :debug "Reader started for ~a" (format-table-name (target copy)))
(let ((start-time (get-internal-real-time))
(*current-batch* (make-batch)))
(map-rows copy :process-row-fn (lambda (row)
@ -24,14 +24,16 @@
(lq:push-queue (list :end-of-data nil nil nil) queue)
(let ((seconds (elapsed-time-since start-time)))
(log-message :info "Reader for ~a is done in ~6$s" (target copy) seconds)
(log-message :info "Reader for ~a is done in ~6$s"
(format-table-name (target copy)) seconds)
(list :reader (target copy) seconds))))
(defmethod format-data-to-copy ((copy copy) raw-queue formatted-queue
&optional pre-formatted)
"Loop over the data in the RAW-QUEUE and prepare it in batches in the
FORMATED-QUEUE, ready to be sent down to PostgreSQL using the COPY protocol."
(log-message :debug "Transformer in action for ~a!" (target copy))
(log-message :debug "Transformer in action for ~a!"
(format-table-name (target copy)))
(let ((start-time (get-internal-real-time)))
(loop :for (mesg batch count oversized?) := (lq:pop-queue raw-queue)
@ -70,7 +72,8 @@
;; and return
(let ((seconds (elapsed-time-since start-time)))
(log-message :info "Transformer for ~a is done in ~6$s" (target copy) seconds)
(log-message :info "Transformer for ~a is done in ~6$s"
(format-table-name (target copy)) seconds)
(list :worker (target copy) seconds))))
(defmethod copy-column-list ((copy copy))
@ -125,7 +128,7 @@
;; threads, so that it's done just once.
(when truncate
(truncate-tables (clone-connection (target-db copy))
(list (target copy))))
(target copy)))
(loop :for w :below 2
:do (lp:submit-task channel

47
src/sources/db3/db3-schema.lisp
View File

@ -39,25 +39,25 @@
(:constructor make-db3-field (name type length)))
name type length)
(defmethod format-pgsql-column ((col db3-field))
"Return a string representing the PostgreSQL column definition."
(let* ((column-name
(apply-identifier-case (db3-field-name col)))
(type-definition
(cdr (assoc (db3-field-type col)
*db3-pgsql-type-mapping*
:test #'string=))))
(format nil "~a ~22t ~a" column-name type-definition)))
(defun list-all-columns (db3 table-name)
(defun list-all-columns (db3 table)
"Return the list of columns for the given DB3-FILE-NAME."
(list
(cons table-name
(loop
for field in (db3::fields db3)
collect (make-db3-field (db3::field-name field)
(db3::field-type field)
(db3::field-length field))))))
(loop
:for field :in (db3::fields db3)
:do (add-field table (make-db3-field (db3::field-name field)
(db3::field-type field)
(db3::field-length field)))))
(defmethod cast ((field db3-field))
"Return the PostgreSQL type definition given the DB3 one."
(let ((type (db3-field-type field)))
(make-column :name (apply-identifier-case (db3-field-name field))
:type-name (cdr (assoc type
*db3-pgsql-type-mapping*
:test #'string=))
:transform (cond ((string= type "L") #'logical-to-boolean)
((string= type "C") #'db3-trim-string)
((string= type "D") #'db3-date-to-pgsql-date)
(t nil)))))
(declaim (inline logical-to-boolean
db3-trim-string
@ -78,14 +78,3 @@
(day (subseq value 6 8)))
(format nil "~a-~a-~a" year month day)))
(defun list-transforms (db3)
"Return the list of transforms to apply to each row of data in order to
convert values to PostgreSQL format"
(loop
for field in (db3::fields db3)
for type = (db3::field-type field)
collect
(cond ((string= type "L") #'logical-to-boolean)
((string= type "C") #'db3-trim-string)
((string= type "D") #'db3-date-to-pgsql-date)
(t nil))))

57
src/sources/db3/db3.lisp
View File

@ -7,29 +7,17 @@
;;;
;;; Integration with pgloader
;;;
(defclass copy-db3 (copy)
(defclass copy-db3 (db-copy)
((encoding :accessor encoding ; file encoding
:initarg :encoding))
(:documentation "pgloader DBF Data Source"))
(defmethod initialize-instance :after ((db3 copy-db3) &key)
"Add a default value for transforms in case it's not been provided."
(setf (slot-value db3 'source) (pathname-name (fd-path (source-db db3))))
(with-connection (conn (source-db db3))
(unless (and (slot-boundp db3 'columns) (slot-value db3 'columns))
(setf (slot-value db3 'columns)
(list-all-columns (fd-db3 conn)
(or (typecase (target db3)
(cons (cdr (target db3)))
(string (target db3)))
(source db3)))))
(let ((transforms (when (slot-boundp db3 'transforms)
(slot-value db3 'transforms))))
(unless transforms
(setf (slot-value db3 'transforms)
(list-transforms (fd-db3 conn)))))))
(setf (slot-value db3 'source)
(let ((table-name (pathname-name (fd-path (source-db db3)))))
(make-table :source-name table-name
:name (apply-identifier-case table-name)))))
(defmethod map-rows ((copy-db3 copy-db3) &key process-row-fn)
"Extract DB3 data and call PROCESS-ROW-FN function with a single
@ -45,9 +33,14 @@
:do (funcall process-row-fn row-array)
:finally (return count)))))
(defun fetch-db3-metadata (db3 table)
"Collect DB3 metadata and prepare our catalog from that."
(with-connection (conn (source-db db3))
(list-all-columns (fd-db3 conn) table)))
(defmethod copy-database ((db3 copy-db3)
&key
table-name
table
data-only
schema-only
(truncate t)
@ -58,23 +51,23 @@
(reset-sequences t))
"Open the DB3 and stream its content to a PostgreSQL database."
(declare (ignore create-indexes reset-sequences))
(let* ((table-name (or table-name
(target db3)
(source db3))))
(let* ((table (or table (target db3) (source db3)))
(schema (make-schema :name (table-name table)
:table-list (list table))))
;; fix the table-name in the db3 object
(setf (target db3) table-name)
(setf (target db3) table)
;; Get the db3 metadata and cast the db3 schema to PostgreSQL
(fetch-db3-metadata db3 table)
(cast table)
(handler-case
(when (and (or create-tables schema-only) (not data-only))
(with-stats-collection ("create, truncate" :section :pre)
(with-pgsql-transaction (:pgconn (target-db db3))
(when create-tables
(with-schema (tname table-name)
(log-message :notice "Create table \"~a\"" tname)
(create-tables (columns db3)
:include-drop include-drop
:if-not-exists t))))))
(prepare-pgsql-database db3
(make-catalog :name (table-name table)
:schema-list (list schema))
:include-drop include-drop))
(cl-postgres::database-error (e)
(declare (ignore e)) ; a log has already been printed
@ -82,4 +75,8 @@
(return-from copy-database)))
(unless schema-only
(setf (fields db3) (table-field-list table)
(columns db3) (table-column-list table)
(transforms db3) (mapcar #'column-transform
(table-column-list table)))
(copy-from db3 :truncate truncate :disable-triggers disable-triggers))))

56
src/sources/ixf/ixf-schema.lisp
View File

@ -51,27 +51,45 @@
(error "IXF Type mapping unknown for: ~d" ixf-type))
pgtype))
(defun format-default-value (default)
"IXF has some default values that we want to transform here, statically."
(cond ((string= "CURRENT TIMESTAMP" default) "CURRENT_TIMESTAMP")
(t default)))
(defun transform-function (field)
"Return the transformation functions needed to cast from ixf-column data."
(let ((coltype (cast-ixf-type (ixf:ixf-column-type field))))
;;
;; The IXF driver we use maps the data type and gets
;; back proper CL typed objects, where we only want to
;; deal with text.
;;
(cond ((or (string-equal "float" coltype)
(string-equal "real" coltype)
(string-equal "double precision" coltype)
(and (<= 7 (length coltype))
(string-equal "numeric" coltype :end2 7)))
#'pgloader.transforms::float-to-string)
(defmethod format-pgsql-column ((col ixf:ixf-column))
"Return a string reprensenting the PostgreSQL column definition"
(let* ((column-name (apply-identifier-case (ixf:ixf-column-name col)))
(type-definition
(format nil
"~a~:[ not null~;~]~:[~*~; default ~a~]"
(cast-ixf-type (ixf:ixf-column-type col))
(ixf:ixf-column-nullable col)
(ixf:ixf-column-has-default col)
(format-default-value (ixf:ixf-column-default col)))))
((string-equal "text" coltype)
nil)
(format nil "~a ~22t ~a" column-name type-definition)))
((string-equal "bytea" coltype)
#'pgloader.transforms::byte-vector-to-bytea)
(defun list-all-columns (ixf-stream table-name)
(t
(lambda (c)
(when c
(princ-to-string c)))))))
(defmethod cast ((col ixf:ixf-column))
"Return the PostgreSQL type definition from given IXF column definition."
(make-column :name (apply-identifier-case (ixf:ixf-column-name col))
:type-name (cast-ixf-type (ixf:ixf-column-type col))
:nullable (ixf:ixf-column-nullable col)
:default (when (ixf:ixf-column-has-default col)
(format-default-value
(ixf:ixf-column-default col)))
:transform (transform-function col)
:comment (ixf:ixf-column-desc col)))
(defun list-all-columns (ixf-stream table)
"Return the list of columns for the given IXF-FILE-NAME."
(ixf:with-ixf-stream (ixf ixf-stream)
(list (cons table-name
(coerce (ixf:ixf-table-columns (ixf:ixf-file-table ixf))
'list)))))
(loop :for field :across (ixf:ixf-table-columns (ixf:ixf-file-table ixf))
:do (add-field table field))))

84
src/sources/ixf/ixf.lisp
View File

@ -8,7 +8,7 @@
;;;
;;; Integration with pgloader
;;;
(defclass copy-ixf (copy)
(defclass copy-ixf (db-copy)
((timezone :accessor timezone ; timezone
:initarg :timezone
:initform local-time:+utc-zone+))
@ -17,50 +17,13 @@
(defmethod initialize-instance :after ((source copy-ixf) &key)
"Add a default value for transforms in case it's not been provided."
(setf (slot-value source 'source)
(pathname-name (fd-path (source-db source))))
(let ((table-name (pathname-name (fd-path (source-db source)))))
(make-table :source-name table-name
:name (apply-identifier-case table-name))))
;; force default timezone when nil
(when (null (timezone source))
(setf (timezone source) local-time:+utc-zone+))
(with-connection (conn (source-db source))
(unless (and (slot-boundp source 'columns) (slot-value source 'columns))
(setf (slot-value source 'columns)
(list-all-columns (conn-handle conn)
(or (typecase (target source)
(cons (cdr (target source)))
(string (target source)))
(source source)))))
(let ((transforms (when (slot-boundp source 'transforms)
(slot-value source 'transforms))))
(unless transforms
(setf (slot-value source 'transforms)
(loop :for field :in (cdar (columns source))
:collect
(let ((coltype (cast-ixf-type (ixf:ixf-column-type field))))
;;
;; The IXF driver we use maps the data type and gets
;; back proper CL typed objects, where we only want to
;; deal with text.
;;
(cond ((or (string-equal "float" coltype)
(string-equal "real" coltype)
(string-equal "double precision" coltype)
(and (<= 7 (length coltype))
(string-equal "numeric" coltype :end2 7)))
#'pgloader.transforms::float-to-string)
((string-equal "text" coltype)
nil)
((string-equal "bytea" coltype)
#'pgloader.transforms::byte-vector-to-bytea)
(t
(lambda (c)
(when c
(princ-to-string c))))))))))))
(setf (timezone source) local-time:+utc-zone+)))
(defmethod map-rows ((copy-ixf copy-ixf) &key process-row-fn)
"Extract IXF data and call PROCESS-ROW-FN function with a single
@ -73,9 +36,14 @@
(ixf:read-headers ixf)
(ixf:map-data ixf process-row-fn)))))
(defun fetch-ixf-metadata (ixf table)
"Collect IXF metadata and prepare our catalog from that."
(with-connection (conn (source-db ixf))
(list-all-columns (conn-handle conn) table)))
(defmethod copy-database ((ixf copy-ixf)
&key
table-name
table
data-only
schema-only
(truncate t)
@ -86,23 +54,23 @@
(reset-sequences t))
"Open the IXF and stream its content to a PostgreSQL database."
(declare (ignore create-indexes reset-sequences))
(let* ((table-name (or table-name
(target ixf)
(source ixf))))
(let* ((table (or table (target ixf) (source ixf)))
(schema (make-schema :name (table-name table)
:table-list (list table))))
;; fix the table-name in the ixf object
(setf (target ixf) table-name)
;; fix the table in the ixf object
(setf (target ixf) table)
;; Get the IXF metadata and cast the IXF schema to PostgreSQL
(fetch-ixf-metadata ixf table)
(cast table)
(handler-case
(when (and (or create-tables schema-only) (not data-only))
(with-stats-collection ("create, truncate" :section :pre)
(with-pgsql-transaction (:pgconn (target-db ixf))
(when create-tables
(with-schema (tname table-name)
(log-message :notice "Create table \"~a\"" tname)
(create-tables (columns ixf)
:include-drop include-drop
:if-not-exists t))))))
(prepare-pgsql-database ixf
(make-catalog :name (table-name table)
:schema-list (list schema))
:include-drop include-drop))
(cl-postgres::database-error (e)
(declare (ignore e)) ; a log has already been printed
@ -110,5 +78,9 @@
(return-from copy-database)))
(unless schema-only
(setf (fields ixf) (table-field-list table)
(columns ixf) (table-column-list table)
(transforms ixf) (mapcar #'column-transform
(table-column-list table)))
(copy-from ixf :truncate truncate :disable-triggers disable-triggers))))

38
src/sources/mssql/mssql-cast-rules.lisp
View File

@ -102,27 +102,19 @@
(t type))))
(defmethod format-pgsql-column ((col mssql-column))
"Return a string representing the PostgreSQL column definition."
(let* ((column-name (apply-identifier-case (mssql-column-name col)))
(type-definition
(with-slots (schema table-name name type default nullable)
col
(declare (ignore schema)) ; FIXME
(let ((ctype (mssql-column-ctype col)))
(cast table-name name type ctype default nullable nil)))))
(format nil "~a ~22t ~a" column-name type-definition)))
(defmethod cast ((field mssql-column))
"Return the PostgreSQL type definition from given MS SQL column definition."
(with-slots (schema table-name name type default nullable)
field
(declare (ignore schema)) ; FIXME
(let* ((ctype (mssql-column-ctype field))
(pgcol
(apply-casting-rules table-name name type ctype default nullable nil)))
;; the MS SQL driver smartly maps data to the proper CL type, but the
;; pgloader API only wants to see text representations to send down the
;; COPY protocol.
(unless (column-transform pgcol)
(setf (column-transform pgcol)
(lambda (val) (if val (format nil "~a" val) :null))))
pgcol)))
(defun cast-mssql-column-definition-to-pgsql (mssql-column)
"Return the PostgreSQL column definition from the MS SQL one."
(multiple-value-bind (column fn)
(with-slots (schema table-name name type default nullable)
mssql-column
(declare (ignore schema)) ; FIXME
(let ((ctype (mssql-column-ctype mssql-column)))
(cast table-name name type ctype default nullable nil)))
;; the MS SQL driver smartly maps data to the proper CL type, but the
;; pgloader API only wants to see text representations to send down the
;; COPY protocol.
(values column (or fn (lambda (val) (if val (format nil "~a" val) :null))))))

122
src/sources/mssql/mssql-schema.lisp
View File

@ -41,27 +41,6 @@
defaults for pgloader."
(mssql:query query :connection (conn-handle *mssql-db*)))
;;;
;;; We store the whole database schema in memory with the following
;;; organisation:
;;;
;;; - an alist of (schema . tables)
;;; - where tables is an alist of (name . cols)
;;; - where cols is a list of mssql-column struct instances
;;;
(defun qualify-name (schema table-name)
"Return the fully qualified name."
(let ((sn (apply-identifier-case schema))
(tn (apply-identifier-case table-name)))
(format nil "~a.~a" sn tn)))
(defun qualified-table-name-list (schema-table-cols-alist)
"Return a flat list of qualified table names."
(loop :for (schema . tables) :in schema-table-cols-alist
:append (loop :for (table . cols) :in tables
:collect (cons (qualify-name schema table) cols))))
;;;
;;; Those functions are to be called from withing an already established
@ -88,19 +67,19 @@
schema-col schema table-col not table-name))
table-name-list)))
(defun list-all-columns (&key
(defun list-all-columns (catalog
&key
(table-type :table)
including
excluding
&aux
(table-type-name (cdr (assoc table-type *table-type*))))
(loop
:with result := nil
:for (schema table-name name type default nullable identity
character-maximum-length
numeric-precision numeric-precision-radix numeric-scale
datetime-precision
character-set-name collation-name)
:for (schema-name table-name name type default nullable identity
character-maximum-length
numeric-precision numeric-precision-radix numeric-scale
datetime-precision
character-set-name collation-name)
:in
(mssql-query (format nil "
select c.table_schema,
@ -149,40 +128,34 @@
order by c.table_schema, c.table_name, c.ordinal_position"
(db-name *mssql-db*)
table-type-name
including ; do we print the clause?
including ; do we print the clause?
(filter-list-to-where-clause including
nil
"c.table_schema"
"c.table_name")
excluding ; do we print the clause?
excluding ; do we print the clause?
(filter-list-to-where-clause excluding
t
"c.table_schema"
"c.table_name")))
:do
(let* ((s-entry (assoc schema result :test 'equal))
(t-entry (when s-entry
(assoc table-name (cdr s-entry) :test 'equal)))
(column
(let* ((schema (maybe-add-schema catalog schema-name))
(table (maybe-add-table schema table-name))
(field
(make-mssql-column
schema table-name name type default nullable
schema-name table-name name type default nullable
(eq 1 identity)
character-maximum-length
numeric-precision numeric-precision-radix numeric-scale
datetime-precision
character-set-name collation-name)))
(if s-entry
(if t-entry
(push-to-end column (cdr t-entry))
(push-to-end (cons table-name (list column)) (cdr s-entry)))
(push-to-end (cons schema (list (cons table-name (list column)))) result)))
:finally (return result)))
(add-field table field))
:finally (return catalog)))
(defun list-all-indexes (&key including excluding)
(defun list-all-indexes (catalog &key including excluding)
"Get the list of MSSQL index definitions per table."
(loop
:with result := nil
:for (schema table name col unique pkey)
:for (schema-name table-name index-name col unique pkey)
:in (mssql-query (format nil "
select schema_name(schema_id) as SchemaName,
o.name as TableName,
@ -220,34 +193,20 @@ order by SchemaName,
"o.name"
)))
:do
(let* ((s-entry (assoc schema result :test 'equal))
(t-entry (when s-entry
(assoc table (cdr s-entry) :test 'equal)))
(i-entry (when t-entry
(assoc name (cdr t-entry) :test 'equal)))
(index (make-pgsql-index :name name
(let* ((schema (find-schema catalog schema-name))
(table (find-table schema table-name))
(index (make-pgsql-index :name index-name
:primary (= pkey 1)
:table-name (qualify-name schema table)
:unique (= unique 1)
:columns (list col))))
(if s-entry
(if t-entry
(if i-entry
(push-to-end col
(pgloader.pgsql::pgsql-index-columns (cdr i-entry)))
(push-to-end (cons name index) (cdr t-entry)))
(push-to-end (cons table (list (cons name index))) (cdr s-entry)))
(push-to-end (cons schema
(list (cons table
(list (cons name index))))) result)))
:finally
(return result)))
(add-index table index))
:finally (return catalog)))
(defun list-all-fkeys (&key including excluding)
(defun list-all-fkeys (catalog &key including excluding)
"Get the list of MSSQL index definitions per table."
(loop
:with result := nil
:for (name schema table col fschema ftable fcol)
:for (fkey-name schema-name table-name col fschema ftable fcol)
:in (mssql-query (format nil "
SELECT
REPLACE(KCU1.CONSTRAINT_NAME, '.', '_') AS 'CONSTRAINT_NAME'
@ -291,32 +250,19 @@ ORDER BY KCU1.CONSTRAINT_NAME, KCU1.ORDINAL_POSITION"
"kcu1.table_schema"
"kcu1.table_name")))
:do
(let* ((s-entry (assoc schema result :test 'equal))
(t-entry (when s-entry
(assoc table (cdr s-entry) :test 'equal)))
(f-entry (when t-entry
(assoc name (cdr t-entry) :test 'equal)))
(fkey
(make-pgsql-fkey :name name
(let* ((schema (find-schema catalog schema-name))
(table (find-table schema table-name))
(pg-fkey
(make-pgsql-fkey :name fkey-name
:table-name (qualify-name schema table)
:columns (list col)
:foreign-table (qualify-name fschema ftable)
:foreign-columns (list fcol))))
(if s-entry
(if t-entry
(if f-entry
(let ((fkey (cdr f-entry)))
(push-to-end col (pgloader.pgsql::pgsql-fkey-columns fkey))
(push-to-end fcol
(pgloader.pgsql::pgsql-fkey-foreign-columns fkey)))
(push-to-end (cons name fkey) (cdr t-entry)))
(push-to-end (cons table (list (cons name fkey))) (cdr s-entry)))
(push-to-end (cons schema
(list (cons table
(list (cons name fkey))))) result)))
:finally
;; we did push, we need to reverse here
(return result)))
:foreign-columns (list fcol)))
(fkey (maybe-add-fkey table fkey-name pg-fkey
:key #'pgloader.pgsql::pgsql-fkey-name)))
(push-to-end col (pgloader.pgsql::pgsql-fkey-columns fkey))
(push-to-end fcol (pgloader.pgsql::pgsql-fkey-foreign-columns fkey)))
:finally (return catalog)))
;;;

266
src/sources/mssql/mssql.lisp
View File

@ -4,7 +4,7 @@
(in-package :pgloader.mssql)
(defclass copy-mssql (copy)
(defclass copy-mssql (db-copy)
((encoding :accessor encoding ; allows forcing encoding
:initarg :encoding
:initform nil))
@ -15,15 +15,15 @@
(let* ((transforms (when (slot-boundp source 'transforms)
(slot-value source 'transforms))))
(when (and (slot-boundp source 'fields) (slot-value source 'fields))
(loop :for field :in (slot-value source 'fields)
:for (column fn) := (multiple-value-bind (column fn)
(cast-mssql-column-definition-to-pgsql field)
(list column fn))
:collect column :into columns
:collect fn :into fns
:finally (progn (setf (slot-value source 'columns) columns)
(unless transforms
(setf (slot-value source 'transforms) fns)))))))
;; cast typically happens in copy-database in the schema structure,
;; and the result is then copied into the copy-mysql instance.
(unless (and (slot-boundp source 'columns) (slot-value source 'columns))
(setf (slot-value source 'columns)
(mapcar #'cast (slot-value source 'fields))))
(unless transforms
(setf (slot-value source 'transforms)
(mapcar #'column-transform (slot-value source 'columns)))))))
(defmethod map-rows ((mssql copy-mssql) &key process-row-fn)
"Extract Mssql data and call PROCESS-ROW-FN function with a single
@ -51,231 +51,37 @@
(log-message :error "~a" e)
(update-stats :data (target mssql) :errs 1)))))))
(defun complete-pgsql-database (pgconn all-columns all-fkeys pkeys
&key
data-only
foreign-keys
reset-sequences)
"After loading the data into PostgreSQL, we can now reset the sequences
and declare foreign keys."
;;
;; Now Reset Sequences, the good time to do that is once the whole data
;; has been imported and once we have the indexes in place, as max() is
;; able to benefit from the indexes. In particular avoid doing that step
;; while CREATE INDEX statements are in flight (avoid locking).
;;
(when reset-sequences
(let ((table-names (mapcar #'car (qualified-table-name-list all-columns))))
(reset-sequences table-names :pgconn pgconn)))
;;
;; Turn UNIQUE indexes into PRIMARY KEYS now
;;
(with-pgsql-connection (pgconn)
(loop :for sql :in pkeys
:when sql
:do (progn
(log-message :notice "~a" sql)
(pgsql-execute-with-timing :post "Primary Keys" sql)))
;;
;; Foreign Key Constraints
;;
;; We need to have finished loading both the reference and the refering
;; tables to be able to build the foreign keys, so wait until all tables
;; and indexes are imported before doing that.
;;
(when (and foreign-keys (not data-only))
(loop :for (schema . tables) :in all-fkeys
:do (loop :for (table-name . fkeys) :in tables
:do (loop :for (fk-name . fkey) :in fkeys
:for sql := (format-pgsql-create-fkey fkey)
:do (progn
(log-message :notice "~a;" sql)
(pgsql-execute-with-timing :post "Foreign Keys" sql))))))))
(defun fetch-mssql-metadata (mssql &key including excluding)
(defmethod fetch-metadata ((mssql copy-mssql)
(catalog catalog)
&key
materialize-views
only-tables
create-indexes
foreign-keys
including
excluding)
"MS SQL introspection to prepare the migration."
(let (all-columns all-indexes all-fkeys)
(with-stats-collection ("fetch meta data"
:use-result-as-rows t
:use-result-as-read t
:section :pre)
(declare (ignore materialize-views only-tables))
(with-stats-collection ("fetch meta data"
:use-result-as-rows t
:use-result-as-read t
:section :pre)
(with-connection (*mssql-db* (source-db mssql))
(setf all-columns (list-all-columns :including including
:excluding excluding))
(list-all-columns catalog
:including including
:excluding excluding)
(setf all-indexes (list-all-indexes :including including
:excluding excluding))
(when create-indexes
(list-all-indexes catalog
:including including
:excluding excluding))
(setf all-fkeys (list-all-fkeys :including including
:excluding excluding))
(when foreign-keys
(list-all-fkeys catalog
:including including
:excluding excluding))
;; return how many objects we're going to deal with in total
;; for stats collection
(+ (loop :for (schema . tables) :in all-columns :sum (length tables))
(loop :for (schema . tables) :in all-indexes
:sum (loop :for (table . indexes) :in tables
:sum (length indexes))))))
(+ (count-tables catalog) (count-indexes catalog)))))
;; now return a plist to the caller
(list :all-columns all-columns
:all-indexes all-indexes
:all-fkeys all-fkeys)))
(defmethod copy-database ((mssql copy-mssql)
&key
(truncate nil)
(disable-triggers nil)
(data-only nil)
(schema-only nil)
(create-tables t)
(create-schemas t)
(include-drop t)
(create-indexes t)
(reset-sequences t)
(foreign-keys t)
(encoding :utf-8)
including
excluding)
"Stream the given MS SQL database down to PostgreSQL."
(let* ((cffi:*default-foreign-encoding* encoding)
(copy-kernel (make-kernel 2))
(copy-channel (let ((lp:*kernel* copy-kernel)) (lp:make-channel)))
(table-count 0)
idx-kernel idx-channel)
(destructuring-bind (&key all-columns all-indexes all-fkeys pkeys)
;; to prepare the run we need to fetch MS SQL meta-data
(fetch-mssql-metadata mssql
:including including
:excluding excluding)
(let ((max-indexes (loop :for (schema . tables) :in all-indexes
:maximizing (loop :for (table . indexes) :in tables
:maximizing (length indexes)))))
(setf idx-kernel (when (and max-indexes (< 0 max-indexes))
(make-kernel max-indexes)))
(setf idx-channel (when idx-kernel
(let ((lp:*kernel* idx-kernel))
(lp:make-channel)))))
;; if asked, first drop/create the tables on the PostgreSQL side
(handler-case
(cond ((and (or create-tables schema-only) (not data-only))
(log-message :notice "~:[~;DROP then ~]CREATE TABLES" include-drop)
(with-stats-collection ("create, truncate" :section :pre)
(with-pgsql-transaction (:pgconn (target-db mssql))
(loop :for (schema . tables) :in all-columns
:do (let ((schema (apply-identifier-case schema)))
;; create schema
(when create-schemas
(let ((sql (format nil "CREATE SCHEMA ~a;" schema)))
(log-message :notice "~a" sql)
(pgsql-execute sql)))
;; set search_path to only that schema
(pgsql-execute
(format nil "SET LOCAL search_path TO ~a;" schema))
;; and now create the tables within that schema
(create-tables tables :include-drop include-drop)))
;; and set indexes OIDs now
;; TODO: fix the MySQL centric API here
(loop :for (schema . tables) :in all-indexes
:do (progn
(pgsql-execute
(format nil "SET LOCAL search_path TO ~a;" schema))
(loop :for (table . indexes) :in tables
:for idx := (mapcar #'cdr indexes)
:do (set-table-oids (list (cons table idx)))))))))
(truncate
(let ((qualified-table-name-list
(qualified-table-name-list all-columns)))
(truncate-tables (target-db mssql)
;; here we really do want only the name
(mapcar #'car qualified-table-name-list)))))
;;
;; In case some error happens in the preparatory transaction, we
;; need to stop now and refrain from trying to load the data into an
;; incomplete schema.
;;
(cl-postgres:database-error (e)
(declare (ignore e)) ; a log has already been printed
(log-message :fatal "Failed to create the schema, see above.")
(return-from copy-database)))
;; Transfer the data
(loop :for (schema . tables) :in all-columns
:do (loop :for (table-name . columns) :in tables
:do
(let ((table-source
(make-instance 'copy-mssql
:source-db (clone-connection (source-db mssql))
:target-db (clone-connection (target-db mssql))
:source (cons schema table-name)
:target (qualify-name schema table-name)
:fields columns)))
(log-message :debug "TARGET: ~a" (target table-source))
;; COPY the data to PostgreSQL, using copy-kernel
(unless schema-only
(incf table-count)
(copy-from table-source
:kernel copy-kernel
:channel copy-channel
:disable-triggers disable-triggers))
;; Create the indexes for that table in parallel with the next
;; COPY, and all at once in concurrent threads to benefit from
;; PostgreSQL synchronous scan ability
;;
;; We just push new index build as they come along, if one
;; index build requires much more time than the others our
;; index build might get unsync: indexes for different tables
;; will get built in parallel --- not a big problem.
(when (and create-indexes (not data-only))
(let* ((s-entry (assoc schema all-indexes :test 'equal))
(indexes-with-names
(cdr (assoc table-name (cdr s-entry) :test 'equal))))
(alexandria:appendf
pkeys
(create-indexes-in-kernel (target-db mssql)
(mapcar #'cdr indexes-with-names)
idx-kernel
idx-channel)))))))
;; now end the kernels
(let ((lp:*kernel* copy-kernel))
(with-stats-collection ("COPY Threads Completion" :section :post)
(loop :for tasks :below (* 2 table-count)
:do (destructuring-bind (task . table-name)
(lp:receive-result copy-channel)
(log-message :debug "Finished processing ~a for ~s"
task table-name)))
(lp:end-kernel)))
(let ((lp:*kernel* idx-kernel))
;; wait until the indexes are done being built...
;; don't forget accounting for that waiting time.
(when (and create-indexes (not data-only))
(with-stats-collection ("Index Build Completion" :section :post)
(loop for idx in all-indexes do (lp:receive-result idx-channel))))
(lp:end-kernel))
;;
;; Complete the PostgreSQL database before handing over.
;;
(complete-pgsql-database (clone-connection (target-db mssql))
all-columns all-fkeys pkeys
:data-only data-only
:foreign-keys foreign-keys
:reset-sequences reset-sequences))))

51
src/sources/mysql/mysql-cast-rules.lisp
View File

@ -169,6 +169,41 @@
:using pgloader.transforms::convert-mysql-point))
"Data Type Casting rules to migrate from MySQL to PostgreSQL")
;;;
;;; MySQL specific fields representation and low-level
;;;
(defstruct (mysql-column
(:constructor make-mysql-column
(table-name name comment dtype ctype default nullable extra)))
table-name name dtype ctype default nullable extra comment)
(defmethod format-extra-type ((col mysql-column) &key include-drop)
"Return a string representing the extra needed PostgreSQL CREATE TYPE
statement, if such is needed"
(let ((dtype (mysql-column-dtype col)))
(when (or (string-equal "enum" dtype)
(string-equal "set" dtype))
(list
(when include-drop
(let* ((type-name
(get-enum-type-name (mysql-column-table-name col)
(mysql-column-name col))))
(format nil "DROP TYPE IF EXISTS ~a;" type-name)))
(get-create-enum (mysql-column-table-name col)
(mysql-column-name col)
(mysql-column-ctype col))))))
(defmethod cast ((col mysql-column))
"Return the PostgreSQL type definition from given MySQL column definition."
(with-slots (table-name name dtype ctype default nullable extra comment)
col
(let ((pgcol
(apply-casting-rules table-name name dtype ctype default nullable extra)))
(setf (column-comment pgcol) comment)
pgcol)))
;;;
;;; MySQL specific testing.
@ -178,7 +213,7 @@
(defun test-casts ()
"Just test some cases for the casts"
(let ((*cast-rules*
'(;; (:source (:column "col1" :auto-increment nil)
'( ;; (:source (:column "col1" :auto-increment nil)
;; :target (:type "timestamptz"
;; :drop-default nil
;; :drop-not-null nil)
@ -229,10 +264,10 @@
(format t " ~a~30T~a~65T~a~%" "MySQL ctype" "PostgreSQL type" "transform")
(format t " ~a~30T~a~65T~a~%" "-----------" "---------------" "---------")
(loop
for (name dtype ctype nullable default extra) in columns
for mycol = (make-mysql-column "table" name dtype ctype nullable default extra)
for (pgtype fn) = (multiple-value-bind (pgcol fn)
(cast "table" name dtype ctype nullable default extra)
(list pgcol fn))
do
(format t "~a: ~a~30T~a~65T~:[~;using ~a~]~%" name ctype pgtype fn fn))))
:for (name dtype ctype nullable default extra) :in columns
:for mycol := (make-mysql-column "table" name dtype ctype nullable default extra)
:for pgcol := (cast mycol)
:do (format t "~a: ~a~30T~a~65T~:[~;using ~a~]~%" name ctype
(format-column pgcol)
(pgloader.pgsql::column-transform pgcol)
(pgloader.pgsql::column-transform pgcol)))))

150
src/sources/mysql/mysql-schema.lisp
View File

@ -6,42 +6,6 @@
(defvar *connection* nil "Current MySQL connection")
;;;
;;; Specific implementation of schema migration, see the API in
;;; src/pgsql/schema.lisp
;;;
(defstruct (mysql-column
(:constructor make-mysql-column
(table-name name dtype ctype default nullable extra)))
table-name name dtype ctype default nullable extra)
(defmethod format-pgsql-column ((col mysql-column))
"Return a string representing the PostgreSQL column definition."
(let* ((column-name (apply-identifier-case (mysql-column-name col)))
(type-definition
(with-slots (table-name name dtype ctype default nullable extra)
col
(cast table-name name dtype ctype default nullable extra))))
(format nil "~a ~22t ~a" column-name type-definition)))
(defmethod format-extra-type ((col mysql-column) &key include-drop)
"Return a string representing the extra needed PostgreSQL CREATE TYPE
statement, if such is needed"
(let ((dtype (mysql-column-dtype col)))
(when (or (string-equal "enum" dtype)
(string-equal "set" dtype))
(list
(when include-drop
(let* ((type-name
(get-enum-type-name (mysql-column-table-name col)
(mysql-column-name col))))
(format nil "DROP TYPE IF EXISTS ~a;" type-name)))
(get-create-enum (mysql-column-table-name col)
(mysql-column-name col)
(mysql-column-ctype col))))))
;;;
;;; General utility to manage MySQL connection
@ -193,7 +157,8 @@
(t default)))
(defun list-all-columns (&key
(defun list-all-columns (schema
&key
(table-type :table)
only-tables
including
@ -202,11 +167,11 @@
(table-type-name (cdr (assoc table-type *table-type*))))
"Get the list of MySQL column names per table."
(loop
with schema = nil
for (table-name name dtype ctype default nullable extra)
in
(mysql-query (format nil "
select c.table_name, c.column_name,
:for (tname tcomment cname ccomment dtype ctype default nullable extra)
:in
(mysql-query (format nil "
select c.table_name, t.table_comment,
c.column_name, c.column_comment,
c.data_type, c.column_type, c.column_default,
c.is_nullable, c.extra
from information_schema.columns c
@ -216,34 +181,38 @@
~:[~*~;and (~{table_name ~a~^ or ~})~]
~:[~*~;and (~{table_name ~a~^ and ~})~]
order by table_name, ordinal_position"
(db-name *connection*)
table-type-name
only-tables ; do we print the clause?
only-tables
including ; do we print the clause?
(filter-list-to-where-clause including)
excluding ; do we print the clause?
(filter-list-to-where-clause excluding t)))
do
(let* ((entry (assoc table-name schema :test 'equal))
(def-val (cleanup-default-value dtype default))
(column (make-mysql-column
table-name name dtype ctype def-val nullable extra)))
(if entry
(push-to-end column (cdr entry))
(push-to-end (cons table-name (list column)) schema)))
finally
(return schema)))
(db-name *connection*)
table-type-name
only-tables ; do we print the clause?
only-tables
including ; do we print the clause?
(filter-list-to-where-clause including)
excluding ; do we print the clause?
(filter-list-to-where-clause excluding t)))
:do
(let* ((table
(case table-type
(:view (maybe-add-view schema tname :comment tcomment))
(:table (maybe-add-table schema tname :comment tcomment))))
(def-val (cleanup-default-value dtype default))
(field (make-mysql-column
tname cname (unless (or (null ccomment)
(string= "" ccomment))
ccomment)
dtype ctype def-val nullable extra)))
(add-field table field))
:finally
(return schema)))
(defun list-all-indexes (&key
(defun list-all-indexes (schema
&key
only-tables
including
excluding)
"Get the list of MySQL index definitions per table."
(loop
with schema = nil
for (table-name name non-unique cols)
in (mysql-query (format nil "
:for (table-name name non-unique cols)
:in (mysql-query (format nil "
SELECT table_name, index_name, non_unique,
cast(GROUP_CONCAT(column_name order by seq_in_index) as char)
FROM information_schema.statistics
@ -259,31 +228,31 @@ GROUP BY table_name, index_name;"
(filter-list-to-where-clause including)
excluding ; do we print the clause?
(filter-list-to-where-clause excluding t)))
do (let ((entry (assoc table-name schema :test 'equal))
(index
(make-pgsql-index :name name
:primary (string= name "PRIMARY")
:table-name table-name
:unique (not (string= "1" non-unique))
:columns (sq:split-sequence #\, cols))))
(if entry
(push-to-end index (cdr entry))
(push-to-end (cons table-name (list index)) schema)))
finally
:do (let ((table (find-table schema table-name))
(index
(make-pgsql-index :name name ; further processing is needed
:primary (string= name "PRIMARY")
:table-name (apply-identifier-case table-name)
:unique (not (string= "1" non-unique))
:columns (mapcar
#'apply-identifier-case
(sq:split-sequence #\, cols)))))
(add-index table index))
:finally
(return schema)))
;;;
;;; MySQL Foreign Keys
;;;
(defun list-all-fkeys (&key
(defun list-all-fkeys (schema
&key
only-tables
including
excluding)
"Get the list of MySQL Foreign Keys definitions per table."
(loop
with schema = nil
for (table-name name ftable cols fcols update-rule delete-rule)
in (mysql-query (format nil "
:for (table-name name ftable cols fcols update-rule delete-rule)
:in (mysql-query (format nil "
SELECT tc.table_name, tc.constraint_name, k.referenced_table_name ft,
group_concat( k.column_name
@ -321,20 +290,21 @@ GROUP BY table_name, index_name;"
(filter-list-to-where-clause including)
excluding ; do we print the clause?
(filter-list-to-where-clause excluding t)))
do (let ((entry (assoc table-name schema :test 'equal))
:do (let ((table (find-table schema table-name))
(fk
(make-pgsql-fkey :name name
:table-name table-name
:columns (sq:split-sequence #\, cols)
:foreign-table ftable
:foreign-columns (sq:split-sequence #\, fcols)
(make-pgsql-fkey :name (apply-identifier-case name)
:table-name (apply-identifier-case table-name)
:columns (mapcar #'apply-identifier-case
(sq:split-sequence #\, cols))
:foreign-table (apply-identifier-case ftable)
:foreign-columns (mapcar
#'apply-identifier-case
(sq:split-sequence #\, fcols))
:update-rule update-rule
:delete-rule delete-rule)))
(if entry
(push-to-end fk (cdr entry))
(push-to-end (cons table-name (list fk)) schema)))
finally
(return schema)))
(add-fkey table fk))
:finally
(return schema)))
;;;

463
src/sources/mysql/mysql.lisp
View File

@ -4,32 +4,26 @@
(in-package :pgloader.mysql)
(defclass copy-mysql (copy)
(defclass copy-mysql (db-copy)
((encoding :accessor encoding ; allows forcing encoding
:initarg :encoding
:initform nil))
(:documentation "pgloader MySQL Data Source"))
(defun cast-mysql-column-definition-to-pgsql (mysql-column)
"Return the PostgreSQL column definition from the MySQL one."
(with-slots (table-name name dtype ctype default nullable extra)
mysql-column
(cast table-name name dtype ctype default nullable extra)))
(defmethod initialize-instance :after ((source copy-mysql) &key)
"Add a default value for transforms in case it's not been provided."
(let ((transforms (and (slot-boundp source 'transforms)
(slot-value source 'transforms))))
(when (and (slot-boundp source 'fields) (slot-value source 'fields))
(loop :for field :in (slot-value source 'fields)
:for (column fn) := (multiple-value-bind (column fn)
(cast-mysql-column-definition-to-pgsql field)
(list column fn))
:collect column :into columns
:collect fn :into fns
:finally (progn (setf (slot-value source 'columns) columns)
(unless transforms
(setf (slot-value source 'transforms) fns)))))))
;; cast typically happens in copy-database in the schema structure,
;; and the result is then copied into the copy-mysql instance.
(unless (and (slot-boundp source 'columns) (slot-value source 'columns))
(setf (slot-value source 'columns)
(mapcar #'cast (slot-value source 'fields))))
(unless transforms
(setf (slot-value source 'transforms)
(mapcar #'column-transform (slot-value source 'columns)))))))
;;;
@ -38,7 +32,7 @@
(defmethod map-rows ((mysql copy-mysql) &key process-row-fn)
"Extract MySQL data and call PROCESS-ROW-FN function with a single
argument (a list of column values) for each row."
(let ((table-name (source mysql))
(let ((table-name (table-source-name (source mysql)))
(qmynd:*mysql-encoding*
(when (encoding mysql)
#+sbcl (encoding mysql)
@ -78,193 +72,74 @@
(mapcar #'apply-identifier-case (mapcar #'mysql-column-name (fields mysql))))
;;;
;;; Prepare the PostgreSQL database before streaming the data into it.
;;;
(defun prepare-pgsql-database (pgconn
all-columns all-indexes all-fkeys
materialize-views view-columns
&key
foreign-keys
include-drop)
"Prepare the target PostgreSQL database: create tables casting datatypes
from the MySQL definitions, prepare index definitions and create target
tables for materialized views.
That function mutates index definitions in ALL-INDEXES."
(log-message :notice "~:[~;DROP then ~]CREATE TABLES" include-drop)
(log-message :debug (if include-drop
"drop then create ~d tables with ~d indexes."
"create ~d tables with ~d indexes.")
(length all-columns)
(loop for (name . idxs) in all-indexes sum (length idxs)))
(with-stats-collection ("create, drop" :use-result-as-rows t :section :pre)
(with-pgsql-transaction (:pgconn pgconn)
;; we need to first drop the Foreign Key Constraints, so that we
;; can DROP TABLE when asked
(when (and foreign-keys include-drop)
(drop-pgsql-fkeys all-fkeys))
;; now drop then create tables and types, etc
(prog1
(create-tables all-columns :include-drop include-drop)
;; MySQL allows the same index name being used against several
;; tables, so we add the PostgreSQL table OID in the index name,
;; to differenciate. Set the table oids now.
(set-table-oids all-indexes)
;; We might have to MATERIALIZE VIEWS
(when materialize-views
(create-tables view-columns :include-drop include-drop))))))
(defun complete-pgsql-database (pgconn all-columns all-fkeys pkeys
table-comments column-comments
&key
data-only
foreign-keys
reset-sequences)
"After loading the data into PostgreSQL, we can now reset the sequences
and declare foreign keys."
;;
;; Now Reset Sequences, the good time to do that is once the whole data
;; has been imported and once we have the indexes in place, as max() is
;; able to benefit from the indexes. In particular avoid doing that step
;; while CREATE INDEX statements are in flight (avoid locking).
;;
(when reset-sequences
(reset-sequences (mapcar #'car all-columns) :pgconn pgconn))
(with-pgsql-connection (pgconn)
;;
;; Turn UNIQUE indexes into PRIMARY KEYS now
;;
(loop :for sql :in pkeys
:when sql
:do (progn
(log-message :notice "~a" sql)
(pgsql-execute-with-timing :post "Primary Keys" sql)))
;;
;; Foreign Key Constraints
;;
;; We need to have finished loading both the reference and the refering
;; tables to be able to build the foreign keys, so wait until all tables
;; and indexes are imported before doing that.
;;
(when (and foreign-keys (not data-only))
(loop :for (table-name . fkeys) :in all-fkeys
:do (loop :for fkey :in fkeys
:for sql := (format-pgsql-create-fkey fkey)
:do (progn
(log-message :notice "~a;" sql)
(pgsql-execute-with-timing :post "Foreign Keys" sql)))))
;;
;; And now, comments on tables and columns.
;;
(log-message :notice "Comments")
(let* ((quote
;; just something improbably found in a table comment, to use as
;; dollar quoting, and generated at random at that.
;;
;; because somehow it appears impossible here to benefit from
;; the usual SQL injection protection offered by the Extended
;; Query Protocol from PostgreSQL.
(concatenate 'string
(map 'string #'code-char
(loop :repeat 5
:collect (+ (random 26) (char-code #\A))))
"_"
(map 'string #'code-char
(loop :repeat 5
:collect (+ (random 26) (char-code #\A)))))))
(loop :for (table-name comment) :in table-comments
:for sql := (format nil "comment on table ~a is $~a$~a$~a$"
(apply-identifier-case table-name)
quote comment quote)
:do (progn
(log-message :log "~a" sql)
(pgsql-execute-with-timing :post "Comments" sql)))
(loop :for (table-name column-name comment) :in column-comments
:for sql := (format nil "comment on column ~a.~a is $~a$~a$~a$"
(apply-identifier-case table-name)
(apply-identifier-case column-name)
quote comment quote)
:do (progn
(log-message :notice "~a;" sql)
(pgsql-execute-with-timing :post "Comments" sql))))))
(defun fetch-mysql-metadata (mysql
&key
materialize-views
only-tables
(create-indexes t)
(foreign-keys t)
including
excluding)
(defmethod fetch-metadata ((mysql copy-mysql)
(catalog catalog)
&key
materialize-views
only-tables
(create-indexes t)
(foreign-keys t)
including
excluding)
"MySQL introspection to prepare the migration."
(let ((view-names (unless (eq :all materialize-views)
(mapcar #'car materialize-views)))
view-columns all-columns all-fkeys all-indexes
table-comments column-comments)
(with-stats-collection ("fetch meta data"
:use-result-as-rows t
:use-result-as-read t
:section :pre)
(with-connection (*connection* (source-db mysql))
;; If asked to MATERIALIZE VIEWS, now is the time to create them in
;; MySQL, when given definitions rather than existing view names.
(when (and materialize-views (not (eq :all materialize-views)))
(create-my-views materialize-views))
(let ((schema (add-schema catalog nil))
(view-names (unless (eq :all materialize-views)
(mapcar #'car materialize-views))))
(with-stats-collection ("fetch meta data"
:use-result-as-rows t
:use-result-as-read t
:section :pre)
(with-connection (*connection* (source-db mysql))
;; If asked to MATERIALIZE VIEWS, now is the time to create them in
;; MySQL, when given definitions rather than existing view names.
(when (and materialize-views (not (eq :all materialize-views)))
(create-my-views materialize-views))
(setf all-columns (list-all-columns :only-tables only-tables
:including including
:excluding excluding)
;; fetch table and columns metadata, covering table and column comments
(list-all-columns schema
:only-tables only-tables
:including including
:excluding excluding)
table-comments (list-table-comments :only-tables only-tables
:including including
:excluding excluding)
;; fetch view (and their columns) metadata, covering comments too
(cond (view-names (list-all-columns schema
:only-tables view-names
:table-type :view))
column-comments (list-columns-comments :only-tables only-tables
:including including
:excluding excluding)
((eq :all materialize-views)
(list-all-columns schema :table-type :view)))
view-columns (cond (view-names
(list-all-columns :only-tables view-names
:table-type :view))
(when foreign-keys
(list-all-fkeys schema
:only-tables only-tables
:including including
:excluding excluding))
((eq :all materialize-views)
(list-all-columns :table-type :view))))
(when create-indexes
(list-all-indexes schema
:only-tables only-tables
:including including
:excluding excluding))
(when foreign-keys
(setf all-fkeys (list-all-fkeys :only-tables only-tables
:including including
:excluding excluding)))
;; return how many objects we're going to deal with in total
;; for stats collection
(+ (count-tables catalog)
(count-views catalog)
(count-indexes catalog)
(count-fkeys catalog))))
(when create-indexes
(setf all-indexes (list-all-indexes :only-tables only-tables
:including including
:excluding excluding)))
catalog))
;; return how many objects we're going to deal with in total
;; for stats collection
(+ (length all-columns) (length all-fkeys)
(length all-indexes) (length view-columns))))
(defmethod cleanup ((mysql copy-mysql) (catalog catalog) &key materialize-views)
"When there is a PostgreSQL error at prepare-pgsql-database step, we might
need to clean-up any view created in the MySQL connection for the
migration purpose."
(when materialize-views
(with-connection (*connection* (source-db mysql))
(drop-my-views materialize-views))))
(log-message :notice
"MySQL metadata fetched: found ~d tables with ~d indexes total."
(length all-columns) (length all-indexes))
;; now return a plist to the caller
(list :all-columns all-columns
:table-comments table-comments
:column-comments column-comments
:all-fkeys all-fkeys
:all-indexes all-indexes
:view-columns view-columns)))
(defvar *decoding-as* nil
"Special per-table encoding/decoding overloading rules for MySQL.")
(defun apply-decoding-as-filters (table-name filters)
"Return a generialized boolean which is non-nil only if TABLE-NAME matches
@ -278,184 +153,24 @@
(:regex (cl-ppcre:scan val table-name))))))))
(some #'apply-filter filters)))
;;;
;;; Work on all tables for given database
;;;
(defmethod copy-database ((mysql copy-mysql)
&key
(truncate nil)
(disable-triggers nil)
(data-only nil)
(schema-only nil)
(create-tables t)
(include-drop t)
(create-indexes t)
(index-names :uniquify)
(reset-sequences t)
(foreign-keys t)
only-tables
including
excluding
decoding-as
materialize-views)
"Export MySQL data and Import it into PostgreSQL"
(let* ((copy-kernel (make-kernel 8))
(copy-channel (let ((lp:*kernel* copy-kernel)) (lp:make-channel)))
(table-count 0)
idx-kernel idx-channel)
(defmethod instanciate-table-copy-object ((copy copy-mysql) (table table))
"Create an new instance for copying TABLE data."
(let* ((fields (table-field-list table))
(columns (table-column-list table))
(transforms (mapcar #'column-transform columns))
(encoding
;; force the data encoding when asked to
(when *decoding-as*
(loop :for (encoding . filters) :in *decoding-as*
:when (apply-decoding-as-filters (table-name table) filters)
:return encoding))))
(make-instance (class-of copy)
:source-db (clone-connection (source-db copy))
:target-db (clone-connection (target-db copy))
:source table
:target (table-name table)
:fields fields
:columns columns
:transforms transforms
:encoding encoding)))
(destructuring-bind (&key view-columns all-columns
table-comments column-comments
all-fkeys all-indexes pkeys)
;; to prepare the run, we need to fetch MySQL meta-data
(fetch-mysql-metadata mysql
:materialize-views materialize-views
:only-tables only-tables
:create-indexes create-indexes
:foreign-keys foreign-keys
:including including
:excluding excluding)
;; prepare our lparallel kernels, dimensioning them to the known sizes
(let ((max-indexes
(loop for (table . indexes) in all-indexes
maximizing (length indexes))))
(setf idx-kernel (when (and max-indexes (< 0 max-indexes))
(make-kernel max-indexes)))
(setf idx-channel (when idx-kernel
(let ((lp:*kernel* idx-kernel))
(lp:make-channel)))))
;; if asked, first drop/create the tables on the PostgreSQL side
(handler-case
(cond ((and (or create-tables schema-only) (not data-only))
(prepare-pgsql-database (target-db mysql)
all-columns
all-indexes
all-fkeys
materialize-views
view-columns
:foreign-keys foreign-keys
:include-drop include-drop))
(t
(when truncate
(truncate-tables (target-db mysql) (mapcar #'car all-columns)))))
;;
;; In case some error happens in the preparatory transaction, we
;; need to stop now and refrain from trying to load the data into
;; an incomplete schema.
;;
(cl-postgres:database-error (e)
(declare (ignore e)) ; a log has already been printed
(log-message :fatal "Failed to create the schema, see above.")
;; we did already create our Views in the MySQL database, so clean
;; that up now.
(when materialize-views
(with-connection (*connection* (source-db mysql))
(drop-my-views materialize-views)))
(return-from copy-database)))
(loop
:for (table-name . columns) :in (append all-columns view-columns)
:unless columns
:do (log-message :error "Table ~s not found, skipping." table-name)
:when columns
:do
(let* ((encoding
;; force the data encoding when asked to
(when decoding-as
(loop :for (encoding . filters) :in decoding-as
:when (apply-decoding-as-filters table-name filters)
:return encoding)))
(table-source
(make-instance 'copy-mysql
:source-db (clone-connection (source-db mysql))
:target-db (clone-connection (target-db mysql))
:source table-name
:target (apply-identifier-case table-name)
:fields columns
:encoding encoding)))
(log-message :debug "TARGET: ~a" (target table-source))
;; first COPY the data from MySQL to PostgreSQL, using copy-kernel
(unless schema-only
(incf table-count)
(copy-from table-source
:kernel copy-kernel
:channel copy-channel
:disable-triggers disable-triggers))
;; Create the indexes for that table in parallel with the next
;; COPY, and all at once in concurrent threads to benefit from
;; PostgreSQL synchronous scan ability
;;
;; We just push new index build as they come along, if one
;; index build requires much more time than the others our
;; index build might get unsync: indexes for different tables
;; will get built in parallel --- not a big problem.
(when (and create-indexes (not data-only))
(let* ((indexes
(cdr (assoc table-name all-indexes :test #'string=)))
(*preserve-index-names* (eq :preserve index-names)))
(alexandria:appendf
pkeys
(create-indexes-in-kernel (target-db mysql)
indexes idx-kernel idx-channel))))))
;; now end the kernels
(let ((lp:*kernel* copy-kernel))
(with-stats-collection ("COPY Threads Completion" :section :post
:use-result-as-read t
:use-result-as-rows t)
(let ((workers-count (* 4 table-count)))
(loop :for tasks :below workers-count
:do (destructuring-bind (task table-name seconds)
(lp:receive-result copy-channel)
(log-message :debug "Finished processing ~a for ~s ~50T~6$s"
task table-name seconds)
(when (eq :writer task)
(update-stats :data table-name :secs seconds))))
(prog1
workers-count
(lp:end-kernel)))))
(let ((lp:*kernel* idx-kernel))
;; wait until the indexes are done being built...
;; don't forget accounting for that waiting time.
(when (and create-indexes (not data-only))
(with-stats-collection ("Index Build Completion" :section :post
:use-result-as-read t
:use-result-as-rows t)
(let ((nb-indexes
(reduce #'+ all-indexes :key (lambda (entry)
(length (cdr entry))))))
(log-message :debug "Waiting for ~a index completion" nb-indexes)
(loop :for count :below nb-indexes
:do (lp:receive-result idx-channel))
nb-indexes)))
(lp:end-kernel))
;;
;; If we created some views for this run, now is the time to DROP'em
;;
(when materialize-views
(with-connection (*connection* (source-db mysql))
(drop-my-views materialize-views)))
;;
;; Complete the PostgreSQL database before handing over.
;;
(complete-pgsql-database (clone-connection (target-db mysql))
all-columns all-fkeys pkeys
table-comments column-comments
:data-only data-only
:foreign-keys foreign-keys
:reset-sequences reset-sequences))))

32
src/sources/sqlite/sqlite-cast-rules.lisp
View File

@ -74,22 +74,18 @@
(paren-pos (position #\( ctype)))
(if paren-pos (subseq ctype 0 paren-pos) ctype)))
(defun cast-sqlite-column-definition-to-pgsql (sqlite-column)
"Return the PostgreSQL column definition from the MySQL one."
(multiple-value-bind (column fn)
(with-slots (table-name name dtype ctype default nullable)
sqlite-column
(cast table-name name dtype ctype default nullable nil))
;; the SQLite driver smartly maps data to the proper CL type, but the
;; pgloader API only wants to see text representations to send down the
;; COPY protocol.
(values column (or fn (lambda (val) (if val (format nil "~a" val) :null))))))
(defmethod cast ((col coldef))
"Return the PostgreSQL type definition from given SQLite column definition."
(with-slots (table-name name dtype ctype default nullable)
col
(let ((pgcol
(apply-casting-rules table-name name dtype ctype default nullable nil)))
;; the SQLite driver smartly maps data to the proper CL type, but the
;; pgloader API only wants to see text representations to send down
;; the COPY protocol.
(unless (column-transform pgcol)
(setf (column-transform pgcol)
(lambda (val) (if val (format nil "~a" val) :null))))
pgcol)))
(defmethod format-pgsql-column ((col coldef))
"Return a string representing the PostgreSQL column definition."
(let* ((column-name (apply-identifier-case (coldef-name col)))
(type-definition
(with-slots (table-name name dtype ctype nullable default)
col
(cast table-name name dtype ctype default nullable nil))))
(format nil "~a ~22t ~a" column-name type-definition)))

73
src/sources/sqlite/sqlite-schema.lisp
View File

@ -6,6 +6,36 @@
(defvar *sqlite-db* nil
"The SQLite database connection handler.")
;;;
;;; Integration with the pgloader Source API
;;;
(defclass sqlite-connection (fd-connection) ())
(defmethod initialize-instance :after ((slconn sqlite-connection) &key)
"Assign the type slot to sqlite."
(setf (slot-value slconn 'type) "sqlite"))
(defmethod open-connection ((slconn sqlite-connection) &key)
(setf (conn-handle slconn)
(sqlite:connect (fd-path slconn)))
(log-message :debug "CONNECTED TO ~a" (fd-path slconn))
slconn)
(defmethod close-connection ((slconn sqlite-connection))
(sqlite:disconnect (conn-handle slconn))
(setf (conn-handle slconn) nil)
slconn)
(defmethod clone-connection ((slconn sqlite-connection))
(change-class (call-next-method slconn) 'sqlite-connection))
(defmethod query ((slconn sqlite-connection) sql &key)
(sqlite:execute-to-list (conn-handle slconn) sql))
;;;
;;; SQLite schema introspection facilities
;;;
(defun filter-list-to-where-clause (filter-list
&optional
not
@ -35,21 +65,24 @@
(loop for (name) in (sqlite:execute-to-list db sql)
collect name)))
(defun list-columns (table-name &optional (db *sqlite-db*))
(defun list-columns (table &optional (db *sqlite-db*))
"Return the list of columns found in TABLE-NAME."
(let ((sql (format nil "PRAGMA table_info(~a)" table-name)))
(loop for (seq name type nullable default pk-id) in
(sqlite:execute-to-list db sql)
collect (make-coldef table-name
seq
name
(ctype-to-dtype (normalize type))
(normalize type)
(= 1 nullable)
(unquote default)
pk-id))))
(let* ((table-name (table-source-name table))
(sql (format nil "PRAGMA table_info(~a)" table-name)))
(loop :for (seq name type nullable default pk-id) :in
(sqlite:execute-to-list db sql)
:do (let ((field (make-coldef table-name
seq
name
(ctype-to-dtype (normalize type))
(normalize type)
(= 1 nullable)
(unquote default)
pk-id)))
(add-field table field)))))
(defun list-all-columns (&key
(defun list-all-columns (schema
&key
(db *sqlite-db*)
including
excluding)
@ -57,7 +90,8 @@
(loop :for table-name :in (list-tables :db db
:including including
:excluding excluding)
:collect (cons table-name (list-columns table-name db))))
:do (let ((table (add-table schema table-name)))
(list-columns table db))))
(defstruct sqlite-idx name table-name sql)
@ -68,7 +102,8 @@
"Generate the PostgresQL statement to build the given SQLite index definition."
(sqlite-idx-sql index))
(defun list-all-indexes (&key
(defun list-all-indexes (schema
&key
(db *sqlite-db*)
including
excluding)
@ -84,15 +119,13 @@
excluding ; do we print the clause?
(filter-list-to-where-clause excluding t))))
(log-message :info "~a" sql)
(loop :with schema := nil
(loop
:for (index-name table-name sql) :in (sqlite:execute-to-list db sql)
:when sql
:do (let ((entry (assoc table-name schema :test 'equal))
:do (let ((table (find-table schema table-name))
(idxdef (make-sqlite-idx :name index-name
:table-name table-name
:sql sql)))
(if entry
(push-to-end idxdef (cdr entry))
(push-to-end (cons table-name (list idxdef)) schema)))
(add-index table idxdef))
:finally (return schema))))

216
src/sources/sqlite/sqlite.lisp
View File

@ -4,33 +4,7 @@
(in-package :pgloader.sqlite)
;;;
;;; Integration with the pgloader Source API
;;;
(defclass sqlite-connection (fd-connection) ())
(defmethod initialize-instance :after ((slconn sqlite-connection) &key)
"Assign the type slot to sqlite."
(setf (slot-value slconn 'type) "sqlite"))
(defmethod open-connection ((slconn sqlite-connection) &key)
(setf (conn-handle slconn)
(sqlite:connect (fd-path slconn)))
(log-message :debug "CONNECTED TO ~a" (fd-path slconn))
slconn)
(defmethod close-connection ((slconn sqlite-connection))
(sqlite:disconnect (conn-handle slconn))
(setf (conn-handle slconn) nil)
slconn)
(defmethod clone-connection ((slconn sqlite-connection))
(change-class (call-next-method slconn) 'sqlite-connection))
(defmethod query ((slconn sqlite-connection) sql &key)
(sqlite:execute-to-list (conn-handle slconn) sql))
(defclass copy-sqlite (copy)
(defclass copy-sqlite (db-copy)
((db :accessor db :initarg :db))
(:documentation "pgloader SQLite Data Source"))
@ -39,15 +13,15 @@
(let* ((transforms (when (slot-boundp source 'transforms)
(slot-value source 'transforms))))
(when (and (slot-boundp source 'fields) (slot-value source 'fields))
(loop for field in (slot-value source 'fields)
for (column fn) = (multiple-value-bind (column fn)
(cast-sqlite-column-definition-to-pgsql field)
(list column fn))
collect column into columns
collect fn into fns
finally (progn (setf (slot-value source 'columns) columns)
(unless transforms
(setf (slot-value source 'transforms) fns)))))))
;; cast typically happens in copy-database in the schema structure,
;; and the result is then copied into the copy-mysql instance.
(unless (and (slot-boundp source 'columns) (slot-value source 'columns))
(setf (slot-value source 'columns)
(mapcar #'cast (slot-value source 'fields))))
(unless transforms
(setf (slot-value source 'transforms)
(mapcar #'column-transform (slot-value source 'columns)))))))
;;; Map a function to each row extracted from SQLite
;;;
@ -83,9 +57,8 @@
(defmethod map-rows ((sqlite copy-sqlite) &key process-row-fn)
"Extract SQLite data and call PROCESS-ROW-FN function with a single
argument (a list of column values) for each row"
(let ((sql (format nil "SELECT * FROM ~a" (source sqlite)))
(pgtypes (map 'vector #'cast-sqlite-column-definition-to-pgsql
(fields sqlite))))
(let ((sql (format nil "SELECT * FROM ~a" (table-source-name (source sqlite))))
(pgtypes (map 'vector #'column-type-name (columns sqlite))))
(with-connection (*sqlite-db* (source-db sqlite))
(let* ((db (conn-handle *sqlite-db*))
(encoding (sqlite-encoding db)))
@ -116,152 +89,37 @@
(log-message :error "~a" e)
(update-stats :data (target sqlite) :errs 1)))))))
(defun fetch-sqlite-metadata (sqlite &key including excluding)
(defmethod fetch-metadata (sqlite catalog
&key
materialize-views
only-tables
create-indexes
foreign-keys
including
excluding)
"SQLite introspection to prepare the migration."
(let (all-columns all-indexes)
(declare (ignore materialize-views only-tables foreign-keys))
(let ((schema (add-schema catalog nil)))
(with-stats-collection ("fetch meta data"
:use-result-as-rows t
:use-result-as-read t
:section :pre)
(with-connection (conn (source-db sqlite))
(let ((*sqlite-db* (conn-handle conn)))
(setf all-columns (list-all-columns :db *sqlite-db*
:including including
:excluding excluding)
(with-connection (conn (source-db sqlite))
(let ((*sqlite-db* (conn-handle conn)))
(list-all-columns schema
:db *sqlite-db*
:including including
:excluding excluding)
all-indexes (list-all-indexes :db *sqlite-db*
:including including
:excluding excluding)))
(when create-indexes
(list-all-indexes schema
:db *sqlite-db*
:including including
:excluding excluding)))
;; return how many objects we're going to deal with in total
;; for stats collection
(+ (length all-columns) (length all-indexes))))
;; return how many objects we're going to deal with in total
;; for stats collection
(+ (count-tables catalog) (count-indexes catalog))))
catalog))
;; now return a plist to the caller
(list :all-columns all-columns
:all-indexes all-indexes)))
(defmethod copy-database ((sqlite copy-sqlite)
&key
data-only
schema-only
(truncate nil)
(disable-triggers nil)
(create-tables t)
(include-drop t)
(create-indexes t)
(reset-sequences t)
only-tables
including
excluding
(encoding :utf-8))
"Stream the given SQLite database down to PostgreSQL."
(declare (ignore only-tables))
(let* ((cffi:*default-foreign-encoding* encoding)
(copy-kernel (make-kernel 4))
(copy-channel (let ((lp:*kernel* copy-kernel)) (lp:make-channel)))
(table-count 0)
idx-kernel idx-channel)
(destructuring-bind (&key all-columns all-indexes pkeys)
(fetch-sqlite-metadata sqlite :including including :excluding excluding)
(let ((max-indexes
(loop for (table . indexes) in all-indexes
maximizing (length indexes))))
(setf idx-kernel (when (and max-indexes (< 0 max-indexes))
(make-kernel max-indexes)))
(setf idx-channel (when idx-kernel
(let ((lp:*kernel* idx-kernel))
(lp:make-channel)))))
;; if asked, first drop/create the tables on the PostgreSQL side
(handler-case
(cond ((and (or create-tables schema-only) (not data-only))
(log-message :notice "~:[~;DROP then ~]CREATE TABLES" include-drop)
(with-stats-collection ("create, truncate" :section :pre)
(with-pgsql-transaction (:pgconn (target-db sqlite))
(create-tables all-columns :include-drop include-drop))))
(truncate
(truncate-tables (target-db sqlite) (mapcar #'car all-columns))))
(cl-postgres:database-error (e)
(declare (ignore e)) ; a log has already been printed
(log-message :fatal "Failed to create the schema, see above.")
(return-from copy-database)))
(loop
for (table-name . columns) in all-columns
do
(let ((table-source
(make-instance 'copy-sqlite
:source-db (clone-connection (source-db sqlite))
:target-db (clone-connection (target-db sqlite))
:source table-name
:target (apply-identifier-case table-name)
:fields columns)))
;; first COPY the data from SQLite to PostgreSQL, using copy-kernel
(unless schema-only
(incf table-count)
(copy-from table-source
:kernel copy-kernel
:channel copy-channel
:disable-triggers disable-triggers))
;; Create the indexes for that table in parallel with the next
;; COPY, and all at once in concurrent threads to benefit from
;; PostgreSQL synchronous scan ability
;;
;; We just push new index build as they come along, if one
;; index build requires much more time than the others our
;; index build might get unsync: indexes for different tables
;; will get built in parallel --- not a big problem.
(when (and create-indexes (not data-only))
(let* ((indexes
(cdr (assoc table-name all-indexes :test #'string=))))
(alexandria:appendf
pkeys
(create-indexes-in-kernel (target-db sqlite) indexes
idx-kernel idx-channel))))))
;; now end the kernels
(let ((lp:*kernel* copy-kernel))
(with-stats-collection ("COPY Threads Completion" :section :post
:use-result-as-read t
:use-result-as-rows t)
(let ((workers-count (* 4 table-count)))
(loop :for tasks :below workers-count
:do (destructuring-bind (task table-name seconds)
(lp:receive-result copy-channel)
(log-message :debug "Finished processing ~a for ~s ~50T~6$s"
task table-name seconds)
(when (eq :writer task)
(update-stats :data table-name :secs seconds))))
(prog1
workers-count
(lp:end-kernel)))))
(let ((lp:*kernel* idx-kernel))
;; wait until the indexes are done being built...
;; don't forget accounting for that waiting time.
(when (and create-indexes (not data-only))
(with-stats-collection ("Index Build Completion" :section :post
:use-result-as-read t
:use-result-as-rows t)
(let ((nb-indexes
(reduce #'+ all-indexes :key (lambda (entry)
(length (cdr entry))))))
(loop :for count :below nb-indexes
:do (lp:receive-result idx-channel))
nb-indexes)))
(lp:end-kernel))
;; don't forget to reset sequences, but only when we did actually import
;; the data.
(when reset-sequences
(reset-sequences (mapcar #'car all-columns)
:pgconn (target-db sqlite))))))

4
src/utils/monitor.lisp
View File

@ -55,11 +55,11 @@
:rs rs
:ws ws)))
(defun process-bad-row (table-name condition data)
(defun process-bad-row (table condition data)
"Send an event to log the bad row DATA in the reject and log files for given
TABLE-NAME (a label in section :data), for reason found in CONDITION."
(send-event (make-bad-row :section :data
:label table-name
:label table
:condition condition
:data data)))

48
src/utils/quoting.lisp Normal file
View File

@ -0,0 +1,48 @@
;;;
;;; Manage PostgreSQL specific quoting of identifiers.
;;;
;;; We use this facility as early as possible (in schema-structs), so we
;;; need those bits of code in utils/ rather than pgsql/.
;;;
(in-package :pgloader.utils)
(defun quoted-p (s)
"Return true if s is a double-quoted string"
(and (eq (char s 0) #\")
(eq (char s (- (length s) 1)) #\")))
(defun apply-identifier-case (identifier)
"Return given IDENTIFIER with CASE handled to be PostgreSQL compatible."
(let* ((lowercase-identifier (cl-ppcre:regex-replace-all
"[^a-zA-Z0-9.]" (string-downcase identifier) "_"))
(*identifier-case*
;; we might need to force to :quote in some cases
;;
;; http://www.postgresql.org/docs/9.1/static/sql-syntax-lexical.html
;;
;; SQL identifiers and key words must begin with a letter (a-z, but
;; also letters with diacritical marks and non-Latin letters) or an
;; underscore (_).
(cond ((quoted-p identifier)
:none)
((not (cl-ppcre:scan "^[A-Za-z_][A-Za-z0-9_$]*$" identifier))
:quote)
((member lowercase-identifier *pgsql-reserved-keywords*
:test #'string=)
(progn
;; we need to both downcase and quote here
(when (eq :downcase *identifier-case*)
(setf identifier lowercase-identifier))
:quote))
;; in other cases follow user directive
(t *identifier-case*))))
(ecase *identifier-case*
(:downcase lowercase-identifier)
(:quote (format nil "\"~a\""
(cl-ppcre:regex-replace-all "\"" identifier "\"\"")))
(:none identifier))))

381
src/utils/schema-structs.lisp Normal file
View File

@ -0,0 +1,381 @@
;;;
;;; PostgreSQL catalogs data structures
;;;
;;; Advanced (database) pgloader data source have to provide facilities to
;;; introspect themselves and CAST their catalogs into PostgreSQL compatible
;;; catalogs as defined here.
;;;
;;; Utility function using those definitions are found in schema.lisp in the
;;; same directory.
;;;
(in-package :pgloader.utils)
(defmacro push-to-end (item place)
`(setf ,place (nconc ,place (list ,item))))
;;;
;;; TODO: stop using anonymous data structures for database catalogs,
;;; currently list of alists of lists... the madness has found its way in
;;; lots of places tho.
;;;
;;;
;;; A database catalog is a list of schema each containing a list of tables,
;;; each being a list of columns.
;;;
;;; Column structures details depend on the specific source type and are
;;; implemented in each source separately.
;;;
(defstruct catalog name schema-list)
(defstruct schema source-name name table-list view-list)
(defstruct table source-name name schema oid comment
;; field is for SOURCE
;; column is for TARGET
field-list column-list index-list fkey-list)
;;;
;;; The generic PostgreSQL column that the CAST generic function is asked to
;;; 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 transform)
;;; those are currently defined in ./schema.lisp
;; (defstruct index name primary unique columns sql conname condef)
;; (defstruct fkey
;; name columns foreign-table foreign-columns update-rule delete-rule)
;;;
;;; Main data collection API
;;;
(defgeneric add-schema (object schema-name &key))
(defgeneric add-table (object table-name &key))
(defgeneric add-view (object view-name &key))
(defgeneric add-column (object column &key))
(defgeneric add-index (object index &key))
(defgeneric add-fkey (object fkey &key))
(defgeneric add-comment (object comment &key))
(defgeneric table-list (object &key)
(:documentation "Return the list of tables found in OBJECT."))
(defgeneric view-list (object &key)
(:documentation "Return the list of views found in OBJECT."))
(defgeneric find-schema (object schema-name &key)
(:documentation
"Find a schema by SCHEMA-NAME in a catalog OBJECT and return the schema"))
(defgeneric find-table (object table-name &key)
(:documentation
"Find a table by TABLE-NAME in a schema OBJECT and return the table"))
(defgeneric find-view (object view-name &key)
(:documentation
"Find a table by TABLE-NAME in a schema OBJECT and return the table"))
(defgeneric find-fkey (object fkey-name &key key test)
(:documentation
"Find a table by FKEY-NAME in a table OBJECT and return the fkey"))
(defgeneric maybe-add-schema (object schema-name &key)
(:documentation "Add a new schema or return existing one."))
(defgeneric maybe-add-table (object table-name &key)
(:documentation "Add a new table or return existing one."))
(defgeneric maybe-add-view (object view-name &key)
(:documentation "Add a new view or return existing one."))
(defgeneric maybe-add-fkey (object fkey-name fkey &key key test)
(:documentation "Add a new fkey or return existing one."))
(defgeneric count-tables (object &key)
(:documentation "Count how many tables we have in total in OBJECT."))
(defgeneric count-views (object &key)
(:documentation "Count how many views we have in total in OBJECT."))
(defgeneric count-indexes (object &key)
(:documentation "Count how many indexes we have in total in OBJECT."))
(defgeneric count-fkeys (object &key)
(:documentation "Count how many forein keys we have in total in OBJECT."))
(defgeneric max-indexes-per-table (schema &key)
(:documentation "Count how many indexes we have maximum per table in SCHEMA."))
(defgeneric cast (object)
(:documentation
"Cast a FIELD definition from a source database into a PostgreSQL COLUMN
definition."))
;;;
;;; Implementation of the methods
;;;
(defmethod table-list ((schema schema) &key)
"Return the list of tables for SCHEMA."
(schema-table-list schema))
(defmethod table-list ((catalog catalog) &key)
"Return the list of tables for table."
(apply #'append (mapcar #'table-list (catalog-schema-list catalog))))
(defmethod view-list ((schema schema) &key)
"Return the list of views for SCHEMA."
(schema-view-list schema))
(defmethod view-list ((catalog catalog) &key)
"Return the list of views for cATALOG."
(apply #'append (mapcar #'view-list (catalog-schema-list catalog))))
(defun create-table (maybe-qualified-name)
"Create a table instance from the db-uri component, either a string or a
cons of two strings: (schema . table)."
(typecase maybe-qualified-name
(string (make-table :source-name maybe-qualified-name
:name (apply-identifier-case maybe-qualified-name)))
(cons (make-table :source-name maybe-qualified-name
:name (apply-identifier-case
(cdr maybe-qualified-name))
:schema (apply-identifier-case
(car maybe-qualified-name))))))
(defmethod add-schema ((catalog catalog) schema-name &key)
"Add SCHEMA-NAME to CATALOG and return the new schema instance."
(let ((schema (make-schema :source-name schema-name
:name (when schema-name
(apply-identifier-case schema-name)))))
(push-to-end schema (catalog-schema-list catalog))
schema))
(defmethod add-table ((schema schema) table-name &key comment)
"Add TABLE-NAME to SCHEMA and return the new table instance."
(let ((table
(make-table :source-name table-name
:name (apply-identifier-case table-name)
:schema (schema-name schema)
:comment (unless (or (null comment) (string= "" comment))
comment))))
(push-to-end table (schema-table-list schema))
table))
(defmethod add-view ((schema schema) view-name &key comment)
"Add TABLE-NAME to SCHEMA and return the new table instance."
(let ((view
(make-table :source-name view-name
:name (apply-identifier-case view-name)
:schema (schema-name schema)
:comment (unless (or (null comment) (string= "" comment))
comment))))
(push-to-end view (schema-view-list schema))
view))
(defmethod find-schema ((catalog catalog) schema-name &key)
"Find SCHEMA-NAME in CATALOG and return the SCHEMA object of this name."
(find schema-name (catalog-schema-list catalog)
:key #'schema-source-name :test 'string=))
(defmethod find-table ((schema schema) table-name &key)
"Find TABLE-NAME in SCHEMA and return the TABLE object of this name."
(find table-name (schema-table-list schema)
:key #'table-source-name :test 'string=))
(defmethod find-view ((schema schema) view-name &key)
"Find TABLE-NAME in SCHEMA and return the TABLE object of this name."
(find view-name (schema-view-list schema)
:key #'table-source-name :test 'string=))
(defmethod maybe-add-schema ((catalog catalog) schema-name &key)
"Add SCHEMA-NAME to the schema-list for CATALOG, or return the existing
schema of the same name if it already exists in the catalog schema-list"
(let ((schema (find-schema catalog schema-name)))
(or schema (add-schema catalog schema-name))))
(defmethod maybe-add-table ((schema schema) table-name &key comment)
"Add TABLE-NAME to the table-list for SCHEMA, or return the existing table
of the same name if it already exists in the schema table-list."
(let ((table (find-table schema table-name)))
(or table (add-table schema table-name :comment comment))))
(defmethod maybe-add-view ((schema schema) view-name &key comment)
"Add TABLE-NAME to the table-list for SCHEMA, or return the existing table
of the same name if it already exists in the schema table-list."
(let ((table (find-view schema view-name)))
(or table (add-view schema view-name :comment comment))))
(defmethod add-field ((table table) field &key)
"Add COLUMN to TABLE and return the TABLE."
(push-to-end field (table-field-list table))
table)
(defmethod add-column ((table table) column &key)
"Add COLUMN to TABLE and return the TABLE."
(push-to-end column (table-column-list table))
table)
(defmethod cast ((table table))
"Cast all fields in table into columns."
(setf (table-column-list table) (mapcar #'cast (table-field-list table))))
(defmethod cast ((schema schema))
"Cast all fields of all tables in SCHEMA into columns."
(loop :for table :in (schema-table-list schema)
:do (cast table))
(loop :for view :in (schema-view-list schema)
:do (cast view)))
(defmethod cast ((catalog catalog))
"Cast all fields of all tables in all schemas in CATALOG into columns."
(loop :for schema :in (catalog-schema-list catalog)
:do (cast schema)))
(defmethod add-index ((table table) index &key)
"Add INDEX to TABLE and return the TABLE."
(push-to-end index (table-index-list table)))
(defmethod add-fkey ((table table) fkey &key)
"Add FKEY to TABLE and return the TABLE."
(push-to-end fkey (table-fkey-list table)))
;;;
;;; There's no simple equivalent to array_agg() in MS SQL, so the fkey query
;;; returns a row per fkey column rather than per fkey. Hence this extra
;;; API:
;;;
(defmethod find-fkey ((table table) fkey-name &key key (test #'string=))
"Find FKEY-NAME in TABLE and return the FKEY object of this name."
(find fkey-name (table-fkey-list table) :key key :test test))
(defmethod maybe-add-fkey ((table table) fkey-name fkey &key key test)
"Add the foreign key FKEY to the table-fkey-list of TABLE unless it
already exists, and return the FKEY object."
(let ((fkey (find-fkey table fkey-name :key key :test test)))
(or fkey (add-fkey table fkey))))
;;;
;;; To report stats to the user, count how many objects we are taking care
;;; of.
;;;
(defmethod count-tables ((schema schema) &key)
"Count tables in given SCHEMA."
(length (schema-table-list schema)))
(defmethod count-tables ((catalog catalog) &key)
(reduce #'+ (mapcar #'count-tables (catalog-schema-list catalog))))
(defmethod count-views ((schema schema) &key)
"Count tables in given SCHEMA."
(length (schema-view-list schema)))
(defmethod count-views ((catalog catalog) &key)
(reduce #'+ (mapcar #'count-views (catalog-schema-list catalog))))
(defmethod count-indexes ((table table) &key)
"Count indexes in given TABLE."
(length (table-index-list table)))
(defmethod count-indexes ((schema schema) &key)
"Count indexes in given SCHEMA."
(reduce #'+ (mapcar #'count-indexes (schema-table-list schema))))
(defmethod count-indexes ((catalog catalog) &key)
"Count indexes in given SCHEMA."
(reduce #'+ (mapcar #'count-indexes (catalog-schema-list catalog))))
(defmethod count-fkeys ((table table) &key)
"Count fkeys in given TABLE."
(length (table-fkey-list table)))
(defmethod count-fkeys ((schema schema) &key)
"Count fkeys in given SCHEMA."
(reduce #'+ (mapcar #'count-fkeys (schema-table-list schema))))
(defmethod count-fkeys ((catalog catalog) &key)
"Count fkeys in given SCHEMA."
(reduce #'+ (mapcar #'count-fkeys (catalog-schema-list catalog))))
(defmethod max-indexes-per-table ((schema schema) &key)
"Count how many indexes maximum per table are listed in SCHEMA."
(reduce #'max (mapcar #'length
(mapcar #'table-index-list
(schema-table-list schema)))))
"Count how many indexes maximum per table are listed in SCHEMA."
(defmethod max-indexes-per-table ((catalog catalog) &key)
"Count how many indexes maximum per table are listed in SCHEMA."
(reduce #'max (mapcar #'max-indexes-per-table (catalog-schema-list catalog))))
;;;
;;; Not a generic/method because only used for the table object, and we want
;;; to use the usual structure print-method in stack traces.
;;;
(defun format-table-name (table)
"TABLE should be a table instance, but for hysterical raisins might be a
CONS of a schema name and a table name, or just the table name as a
string."
(etypecase table
(table (format nil "~@[~a.~]~a" (table-schema table) (table-name table)))
(cons (format nil "~a.~a" (car table) (cdr table)))
(string table)))
;;;
;;; Still lacking round tuits here, so for the moment the representation of
;;; a table name is either a string or a cons built from schema and
;;; table-name.
;;;
(defmacro with-schema ((var table-name) &body body)
"When table-name is a CONS, SET search_path TO its CAR and return its CDR,
otherwise just return the TABLE-NAME. A PostgreSQL connection must be
established when calling this function."
`(let ((,var
(typecase ,table-name
(table (if (table-schema ,table-name)
(let ((sql (format nil "SET search_path TO ~a;"
(table-schema ,table-name))))
(log-message :notice "~a" sql)
(pgloader.pgsql:pgsql-execute sql)
(table-name ,table-name))
(table-name ,table-name)))
(cons (let ((sql (format nil "SET search_path TO ~a;"
(car ,table-name))))
(log-message :notice "~a" sql)
(pgloader.pgsql:pgsql-execute sql)
(cdr ,table-name)))
(string ,table-name))))
,@body))
;;;
;;; PostgreSQL column and type output
;;;
(defun format-default-value (default &optional transform)
"Returns suitably quoted default value for CREATE TABLE command."
(cond
((null default) "NULL")
((and (stringp default) (string= "NULL" default)) default)
((and (stringp default) (string= "CURRENT_TIMESTAMP" default)) default)
((and (stringp default) (string= "CURRENT TIMESTAMP" default))
"CURRENT_TIMESTAMP")
(t
;; apply the transformation function to the default value
(if transform (format-default-value (funcall transform default))
(format nil "'~a'" default)))))
(defmethod format-column ((column column))
"Format the PostgreSQL data type."
(with-slots (name type-name type-mod nullable default transform) column
(format nil
"~a ~22t ~a~:[~*~;~a~]~:[ not null~;~]~:[~; default ~a~]"
name
type-name
type-mod
type-mod
nullable
default default)))

6
src/utils/state.lisp
View File

@ -28,12 +28,6 @@
(rs 0.0 :type float)
(ws 0.0 :type float))
(defun format-table-name (table-name)
"TABLE-NAME might be a CONS of a schema name and a table name."
(etypecase table-name
(cons (format nil "~a.~a" (car table-name) (cdr table-name)))
(string table-name)))
(defun relative-pathname (filename type &optional dbname)
"Return the pathname of a file of type TYPE (dat or log) under *ROOT-DIR*"
(let ((dir (if dbname