Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion clojure/Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,10 @@ test-unit:
-n pgloader.load-file.parser-test \
-n pgloader.transforms-test \
-n pgloader.pg-service-test \
-n pgloader.cli-test
-n pgloader.cli-test \
-n pgloader.log-test \
-n pgloader.source.mssql-test \
-n pgloader.source.mysql-test

# ─── E2E integration tests ────────────────────────────────────────────────────
# All suite management lives in tests/Makefile.
Expand Down
70 changes: 52 additions & 18 deletions clojure/src/pgloader/core.clj
Original file line number Diff line number Diff line change
Expand Up @@ -214,6 +214,34 @@
(.rollback pg-conn)
(log/warn (str label " failed (skipping): " (.getMessage e)))))))

(defn- reset-sequences!
"Execute reset-sequences-sql statements, each in its own transaction.
Returns how many sequences were actually reset: setval() yields NULL for
columns that do not own a sequence, and those are not counted.
Errors are logged as warnings and skipped."
[^Connection pg-conn sqls label]
(reduce (fn [n sql]
(try
(let [row (first (jdbc/execute! pg-conn [sql]))]
(.commit pg-conn)
(if (some? (first (vals row))) (inc n) n))
(catch Exception e
(.rollback pg-conn)
(log/warn (str label " failed (skipping): " (.getMessage e)))
n)))
0 sqls))

(defn- create-schemas!
"Create every target schema up front, before sequences, types and tables."
[^Connection pg-conn schemas]
(when-let [sqls (seq (ddl/create-schemas-sql schemas))]
(stats/new-entry! :pre "Create Schemas")
(let [start (System/nanoTime)]
(exec-post-ddl! pg-conn sqls "CREATE SCHEMA")
(stats/update-entry! :pre "Create Schemas"
:rows (count sqls)
:total-nanos (- (System/nanoTime) start)))))

(defn- pg-major-version
"Return the PostgreSQL major version as an integer (e.g. 13, 14, 15)."
[^Connection pg-conn]
Expand Down Expand Up @@ -422,9 +450,9 @@
;; ── All other load types ──────────────────────────────────────────────────
(let [source-uri (:source cmd)
target-uri (get-in cmd [:target :target-uri])
_ (log/debug (str "Connecting to PostgreSQL at " (:raw target-uri)))
_ (log/debug (str "Connecting to PostgreSQL at " (plog/redact-uri (:raw target-uri))))
^Connection pg-conn (postgres-connection target-uri)
_ (log/info (str "Connected to PostgreSQL at " (:raw target-uri)))
_ (log/info (str "Connected to PostgreSQL at " (plog/redact-uri (:raw target-uri))))
source-overrides (select-keys source-uri [:inline-data])
commands-filters (:filters cmd)
table-filter (when commands-filters
Expand All @@ -441,8 +469,8 @@
source (source-from-uri source-uri table-spec (:with-options cmd) source-overrides (:decoding-as cmd))
verbose (or (:debug opts) (:verbose opts) false)]
(log/info "pgloader v4")
(log/info "Source:" (source-name source))
(log/info "Target:" (:raw target-uri))
(log/info "Source:" (plog/redact-uri (source-name source)))
(log/info "Target:" (plog/redact-uri (:raw target-uri)))
(if copy/*dry-run*
;; Dry run: verify both connections are reachable, then stop.
;; Mirrors v3 behaviour: no catalog fetch, no DDL, no COPY.
Expand All @@ -467,8 +495,8 @@
(when-let [mysql-params (seq (filter :is-mysql (:set-parameters cmd)))]
(log/debug "Sending MySQL SET parameters to source connection")
(mysql-source/execute-set-params! source mysql-params)))
(let [_ (log/debug (str "Connecting to source: " (source-name source)))
_ (log/info (str "Fetching catalog from " (source-name source)))
(let [_ (log/debug (str "Connecting to source: " (plog/redact-uri (source-name source))))
_ (log/info (str "Fetching catalog from " (plog/redact-uri (source-name source))))
fetch-t0 (System/nanoTime)
cat (catalog source)
;; If MATERIALIZE ALL VIEWS, append view catalog entries to table catalog.
Expand Down Expand Up @@ -707,10 +735,15 @@
;; Ensure extensions required by column defaults exist
;; (e.g. pgcrypto for gen_random_uuid() on PG < 13).
(ensure-uuid-extension! pg-conn cat)
;; Create sequences before tables so that NEXT VALUE FOR
;; defaults (translated to nextval()) resolve correctly.
(when (= :mssql (:type source-uri))
(when-let [seqs (seq (mssql-source/catalog-sequences source))]
;; Create schemas first: sequences, ENUM types and tables
;; are all created inside them.
(let [seqs (when (= :mssql (:type source-uri))
(seq (mssql-source/catalog-sequences source)))]
(create-schemas! pg-conn (concat (map #(or (:schema %) "public") cat)
(map :schema seqs)))
;; Create sequences before tables so that NEXT VALUE FOR
;; defaults (translated to nextval()) resolve correctly.
(when seqs
(log/info (str "Creating " (count seqs) " sequence(s) from MS SQL"))
(run-ddl-tx pg-conn (ddl/create-sequences-sql seqs))))
(stats/new-entry! :pre "Create tables")
Expand Down Expand Up @@ -1049,15 +1082,16 @@
(log/info "Resetting sequences")
(stats/new-entry! :post "Reset Sequences")
(let [start (System/nanoTime)
n (atom 0)]
(doseq [t cat]
(let [schema (or (:schema t) "public")
table (:table-name t)]
(when-let [seqs (seq (ddl/reset-sequences-sql schema table (:columns t)))]
(exec-post-ddl! pg-conn seqs (str "SEQUENCE for " table))
(swap! n inc))))
n (reduce (fn [n t]
(+ n (reset-sequences!
pg-conn
(ddl/reset-sequences-sql (or (:schema t) "public")
(:table-name t)
(:columns t))
(str "SEQUENCE for " (:table-name t)))))
0 cat)]
(stats/update-entry! :post "Reset Sequences"
:rows @n :total-nanos (- (System/nanoTime) start)))))
:rows n :total-nanos (- (System/nanoTime) start)))))
;; Execute AFTER LOAD DO statements — skip when the load failed (#930).
(when-let [after-cmds (and (not @load-failed) (seq (:after-load cmd)))]
(log/debug "Executing AFTER LOAD DO commands")
Expand Down
64 changes: 54 additions & 10 deletions clojure/src/pgloader/ddl/common.clj
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,28 @@
([schema table]
(str (identifier-quote schema) "." (identifier-quote table))))

(defn- auto-increment?
"True when a column's :extra marks it as auto-increment."
[extra]
(str/includes? (str/lower-case (str extra)) "auto_increment"))

(defn- serial-type?
[^String pg-type]
(boolean (re-matches #"(?i)(small|big)?serial" (str pg-type))))

(declare pg-type-for)

(defn- auto-increment-pg-type
"Map an auto-increment integer column to serial/bigserial, mirroring the CL
default cast rules: types that map to bigint (or numeric, for bigint
unsigned) become bigserial, all smaller integer types become serial.
Returns nil when the source type is not an integer type."
[^String mysql-type]
(when (re-find #"(?i)^(tiny|small|medium|big)?int" mysql-type)
(case (pg-type-for mysql-type nil)
("bigint" "numeric") "bigserial"
"serial")))

(defn- pg-type-for
"Map a MySQL type name to PostgreSQL type.
Preserves precision modifiers for temporal types (#1629)."
Expand All @@ -27,6 +49,9 @@
;; Extract precision modifier like (6) from datetime(6)
typemod (re-find #"\(\d+\)" mysql-type)]
(cond
;; AUTO_INCREMENT integers own a sequence on the target
(and (auto-increment? extra) (auto-increment-pg-type mysql-type))
(auto-increment-pg-type mysql-type)
;; Pass-through: native PostgreSQL types emitted by non-MySQL sources
(= lower "uuid") "uuid"
(= lower "xml") "xml"
Expand Down Expand Up @@ -202,14 +227,16 @@
(re-find #"(?i)^(timestamp|date|time)" (or pg-type ""))
(re-matches #"^-?\d+$" (str coerced-default)))
default-str (when (and coerced-default
;; serial types come with their own nextval() default
(not (serial-type? pg-type))
(not= "NULL" (str coerced-default))
(not= "" (str coerced-default))
(not zero?)
(not temporal-int-default?))
(str " DEFAULT " (format-default coerced-default)))]
(str " " quoted-name " " pg-type
(when (and (false? is-nullable)
(not (str/includes? (str extra) "auto_increment"))
(not (auto-increment? extra))
(not= "NULL" (str column-default))
(not zero?))
" NOT NULL")
Expand Down Expand Up @@ -600,38 +627,55 @@
quoted-schema "." quoted-fn "();")]
[fn-sql trg-sql]))))

(defn- sql-literal
[^String s]
(str "'" (str/replace s "'" "''") "'"))

(defn reset-sequences-sql
"Generate SELECT setval() for auto-increment columns.
Calls pg_catalog.setval with MAX(col) to advance the sequence
past any data that was bulk-loaded (bypassing the sequence).
past any data that was bulk-loaded (bypassing the sequence), so that the
next nextval() returns MAX(col) + 1, or 1 on an empty table.

Each statement returns a single row whose setval value is NULL when the
column does not own a sequence (pg_get_serial_sequence returns NULL).

Returns a vector of SQL strings (one per auto-increment column)."
[schema table-name columns]
(let [quoted-fqname (quote-fqname schema table-name)]
(vec (keep (fn [col]
(when (and (:column-name col)
(:extra col)
(str/includes? (str/lower-case (:extra col)) "auto_increment")
(auto-increment? (:extra col))
;; Only reset sequences for integer-typed columns.
;; Cast rules may change the PG type (e.g. int → text);
;; non-integer MAX() causes COALESCE type mismatch.
;; non-integer MAX() causes a type mismatch.
(let [src (or (:source-column-type col) (:column-type col) "")
ct (or (:column-type col) "")
pg-type (str/lower-case
(if (not= src ct)
ct
(pg-type-for ct nil)))]
(pg-type-for ct (:extra col))))]
(some #(str/starts-with? pg-type %)
["integer" "bigint" "bigserial" "smallint" "serial"
"int2" "int4" "int8"])))
(let [col-name (:column-name col)
quoted-col (identifier-quote col-name)]
(let [quoted-col (identifier-quote (:column-name col))
max-col (str "MAX(" quoted-col ")")]
;; setval(seq, max, true) makes nextval() return max + 1;
;; an empty table gets setval(seq, 1, false) so nextval() returns 1.
(str "SELECT pg_catalog.setval("
"pg_get_serial_sequence('" quoted-fqname "', '" col-name "')"
", COALESCE(MAX(" quoted-col "), 1), false)"
"pg_get_serial_sequence(" (sql-literal quoted-fqname)
", " (sql-literal (:column-name col)) ")"
", GREATEST(" max-col ", 1), " max-col " IS NOT NULL)"
" FROM " quoted-fqname ";\n"))))
columns))))

(defn create-schemas-sql
"Generate CREATE SCHEMA IF NOT EXISTS statements, one per distinct schema,
in first-seen order."
[schemas]
(mapv #(str "CREATE SCHEMA IF NOT EXISTS " (identifier-quote %) ";")
(distinct (remove nil? schemas))))

(defn create-sequence-sql
"Generate DROP … / CREATE SEQUENCE SQL for an MSSQL sequence descriptor.
seq-map keys: :schema :name :start :increment :min :max :cycle? :cache :current
Expand Down
12 changes: 12 additions & 0 deletions clojure/src/pgloader/log.clj
Original file line number Diff line number Diff line change
@@ -1,8 +1,20 @@
(ns pgloader.log
(:require [clojure.string :as str])
(:import [java.util Locale]))

(set! *warn-on-reflection* true)

(defn redact-uri
"Mask passwords in a connection URI or JDBC URL so it can be logged:
postgresql://user:secret@host/db → postgresql://user:****@host/db
jdbc:sqlserver://host;password=secret → jdbc:sqlserver://host;password=****
jdbc:postgresql://host/db?password=x → jdbc:postgresql://host/db?password=****"
[s]
(when s
(-> (str s)
(str/replace #"(://[^:/@;?#\s]*):[^/?#;\s]*@" "$1:****@")
(str/replace #"(?i)([;?&]password=)[^;&\s]*" "$1****"))))

(defn- locale-format
"Like clojure.core/format but with an explicit Locale to force '.' as decimal separator."
[locale fmt & args]
Expand Down
32 changes: 29 additions & 3 deletions clojure/src/pgloader/source/mssql.clj
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
(ns pgloader.source.mssql
(:require [pgloader.source.protocol :refer [Source]]
[pgloader.log :as plog]
[hugsql.core :as hugsql]
[clojure.string :as str]
[next.jdbc :as jdbc]
Expand Down Expand Up @@ -81,22 +82,47 @@
"[schema].[sequence] syntax, dropping: " s))
nil)))

(def ^:private tsql-default-functions
"T-SQL niladic functions used in column defaults, and their PostgreSQL
equivalents. Keys are lower-case."
{"getdate()" "CURRENT_TIMESTAMP"
"getutcdate()" "CURRENT_TIMESTAMP"
"sysdatetime()" "CURRENT_TIMESTAMP"
"sysutcdatetime()" "CURRENT_TIMESTAMP"
"sysdatetimeoffset()" "CURRENT_TIMESTAMP"
"current_timestamp" "CURRENT_TIMESTAMP"
"newid()" "gen_random_uuid()"
"newsequentialid()" "gen_random_uuid()"})

(defn- unquote-string-literal
"Return the value of a T-SQL string literal ('abc' or N'abc'), with doubled
quotes unescaped, or nil when s is not a single string literal."
[^String s]
(when-let [[_ body] (re-matches #"(?s)[Nn]?'((?:[^']|'')*)'" s)]
(str/replace body "''" "'")))

(defn- sanitize-default
"Normalise MSSQL column defaults for PostgreSQL:
- NEXT VALUE FOR [schema].[seq] → nextval('schema.seq') (#1497)
- T-SQL functions (GETDATE(), SYSUTCDATETIME(), NEWID(), …) → PostgreSQL
- String literals 'abc' and Unicode literals N'abc' → their value
- CONVERT(…) expressions not already translated by mssql.sql → nil (#1409)
- Empty-string defaults on numeric columns → nil (#1163)"
[default pg-type]
(when default
(let [trimmed (str/trim default)
lower (str/lower-case trimmed)]
lower (str/lower-case trimmed)
literal (unquote-string-literal trimmed)]
(cond
;; SQL Server sequence default → PostgreSQL nextval()
(str/starts-with? lower "next value for") (translate-next-value-for trimmed)
(contains? tsql-default-functions lower) (get tsql-default-functions lower)
;; Any remaining CONVERT(…) that mssql.sql didn't map to a keyword
(str/starts-with? lower "convert(") nil
;; Empty string on a numeric target type
(and (= trimmed "") (contains? numeric-pg-types pg-type)) nil
(and (or (= trimmed "") (= literal ""))
(contains? numeric-pg-types pg-type)) nil
literal literal
:else default))))

(defn- connection
Expand Down Expand Up @@ -353,4 +379,4 @@
(defn create-source
[uri-map _table-spec]
(let [conn (connection uri-map)]
(->MSSQLSource conn (:raw uri-map))))
(->MSSQLSource conn (plog/redact-uri (:raw uri-map)))))
2 changes: 0 additions & 2 deletions clojure/src/pgloader/source/mssql.sql
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,6 @@ SELECT c.COLUMN_NAME,
WHEN SUBSTRING(c.COLUMN_DEFAULT, 3, LEN(c.COLUMN_DEFAULT) - 4) LIKE 'convert(%varchar%,getdate(),%)' THEN 'CURRENT_DATE'
WHEN SUBSTRING(c.COLUMN_DEFAULT, 3, LEN(c.COLUMN_DEFAULT) - 4) = 'getdate()' THEN 'CURRENT_TIMESTAMP'
WHEN SUBSTRING(c.COLUMN_DEFAULT, 3, LEN(c.COLUMN_DEFAULT) - 4) = 'sysdatetimeoffset()' THEN 'CURRENT_TIMESTAMP'
WHEN SUBSTRING(c.COLUMN_DEFAULT, 3, LEN(c.COLUMN_DEFAULT) - 4) LIKE '''%''' THEN SUBSTRING(c.COLUMN_DEFAULT, 4, LEN(c.COLUMN_DEFAULT) - 6)
ELSE SUBSTRING(c.COLUMN_DEFAULT, 3, LEN(c.COLUMN_DEFAULT) - 4)
END
WHEN c.COLUMN_DEFAULT LIKE '(%' AND c.COLUMN_DEFAULT LIKE '%)' THEN
Expand All @@ -47,7 +46,6 @@ SELECT c.COLUMN_NAME,
WHEN SUBSTRING(c.COLUMN_DEFAULT, 2, LEN(c.COLUMN_DEFAULT) - 2) LIKE 'convert(%varchar%,getdate(),%)' THEN 'CURRENT_DATE'
WHEN SUBSTRING(c.COLUMN_DEFAULT, 2, LEN(c.COLUMN_DEFAULT) - 2) = 'getdate()' THEN 'CURRENT_TIMESTAMP'
WHEN SUBSTRING(c.COLUMN_DEFAULT, 2, LEN(c.COLUMN_DEFAULT) - 2) = 'sysdatetimeoffset()' THEN 'CURRENT_TIMESTAMP'
WHEN SUBSTRING(c.COLUMN_DEFAULT, 2, LEN(c.COLUMN_DEFAULT) - 2) LIKE '''%''' THEN SUBSTRING(c.COLUMN_DEFAULT, 3, LEN(c.COLUMN_DEFAULT) - 4)
ELSE SUBSTRING(c.COLUMN_DEFAULT, 2, LEN(c.COLUMN_DEFAULT) - 2)
END
ELSE c.COLUMN_DEFAULT
Expand Down
Loading
Loading