#!/usr/bin/perl -w

# This script takes a sql server SQL schema dump, and creates a postgresql dump

# Optionnaly, if asked, generate a kettle job to transfer all data. This is done via -k dir

# Details in the README file

# This program is made to die on all error conditions: if there is something not understood in

# a dump, it has to be added, or manually ignored (in order to improve this program rapidly)

# Licence: GPLv3

# Copyright Marc Cousin, Dalibo

use Getopt::Long;

use Data::Dumper;

use Cwd;

use Encode::Guess;

use Carp;

use strict;

# Global objects definition structure: we need it to store all seen tables, detect which ones have LOBS

# and if their PK is a simple integer (so we can parallelize readings on these tables in a special

# kettle transformation)

# $objects will contain the parsed structure of the SQL Server dump

# If you have to hack and want to understand its structure, just uncomment the call to Dumper() in the code

# There is just too many things in it, and it evolves all the time

my $objects;

# These are global variables, from configuration file or command line arguments

our ($sd, $sh, $si, $sp, $su, $sw, $pd, $ph, $pp, $pu, $pw); # Connection args

our $conf_file;

our $filename; # Filename passed as arg

our $case_insensitive; # Passed as arg: was SQL Server installation case insensitive ? PostgreSQL can't ignore accents anyway

# If yes, we will generate citext with CHECK constraints, that's the best we can do

our $norelabel_dbo; # Passed as arg: should we convert DBO to public ?

our $relabel_schemas;

our $convert_numeric_to_int; # Should we convert numerics to int when possible ? (numeric (4,0) could be converted an int, for instance)

our $kettle;

our $before_file;

our $after_file;

our $unsure_file;

our $case_treatment=1; # 1=convert to lowercase, 2=convert to snake_case, 0 do nothing

our $ignore_errors;

our $keep_identifier_case;

our $camel_to_snake;

our $validate_constraints;

our $parallelism_in;

our $parallelism_out;

our $sort_size;

our $use_pk_if_possible;

our $pforce_ssl;

our $stringtype_unspecified;

# Will be set if we detect GIS objects

our $requires_postgis=0;

# These three variables are loaded in the BEGIN block at the end of this file (they are very big

my $template;

my $template_lob;

my $incremental_template;

my $incremental_template_sortable_pk;

my ($job_header, $job_middle, $job_footer)

; # These are used to create the static parts of the job

my ($job_entry, $job_hop)

; # These are used to create the dynamic parts of the job (XML file)

my @view_list; # array to keep view ordering from sql server's dump (a view may depend on another view)

# Opens the configuration file

# Sets $sd $sh $si $sp $su $sw $pd $ph $pp $pu $pw when they are not set in the command line already

# Also gets kettle parameters...

sub parse_conf_file

{

# Correspondance between conf_file parameter and program variable

# This is also used as the list of accepted parameters in the configuration file

my %parameters = ('sql server database' => 'sd',

'sql server host' => 'sh',

'sql server host instance' => 'si',

'sql server port' => 'sp',

'sql server username' => 'su',

'sql server password' => 'sw',

'postgresql database' => 'pd',

'postgresql host' => 'ph',

'postgresql port' => 'pp',

'postgresql username' => 'pu',

'postgresql password' => 'pw',

'kettle directory' => 'kettle',

'parallelism_in' => 'parallelism_in',

'parallelism_out' => 'parallelism_out',

'before file' => 'before_file',

'after file' => 'after_file',

'unsure file' => 'unsure_file',

'sql server dump filename' => 'filename',

'case insensitive' => 'case_insensitive',

'no relabel dbo' => 'norelabel_dbo',

'convert numeric to int' => 'convert_numeric_to_int',

'relabel schemas' => 'relabel_schemas',

'keep identifier case' => 'keep_identifier_case',

'camelcasetosnake' => 'camel_to_snake',

'validate constraints' => 'validate_constraints',

'sort size' => 'sort_size',

'use pk if possible' => 'use_pk_if_possible',

'ignore errors' => 'ignore_errors',

'postgresql force ssl' => 'pforce_ssl',

'stringtype unspecified' => 'stringtype_unspecified',

);

# Open the conf file or die

open CONF, $conf_file or die "Cannot open $conf_file";

while (my $line = <CONF>)

{

$line =~ s/#.*//; # Remove comments

$line =~ s/\s+=\s+/=/; # Remove whitespaces around the =

$line =~ s/\s+$//; # Remove trailing whitespaces

next

if ($line =~ /^$/); # Empty line after comments have been removed

$line =~ /^(.*?)=(.*)$/ or die "Cannot parse $line from $conf_file";

my ($param, $value) = ($1, $2);

no strict 'refs'; # Using references by name, temporarily

unless (defined $parameters{$param})

{

die "Cannot understand parameter $param in $conf_file";

}

my $param_name = $parameters{$param};

if (defined $$param_name)

{

next; # Parameter overriden in command line

}

$$param_name = $value;

use strict 'refs';

}

close CONF;

}

sub set_default_conf_values

{

# Hard coded default values, set only if not passed or found in configuration

$case_insensitive=0 unless (defined ($case_insensitive));

$norelabel_dbo=0 unless (defined ($norelabel_dbo));

$convert_numeric_to_int=0 unless (defined ($convert_numeric_to_int));

$case_treatment=0 if (defined ($keep_identifier_case));

$case_treatment=2 if (defined ($camel_to_snake));

$parallelism_in=1 unless (defined ($parallelism_in));# the jdbc driver often errors when there are several sessions to sql server

$parallelism_out=8 unless (defined ($parallelism_out));

$sort_size=10000 unless (defined ($sort_size));

$use_pk_if_possible=0 unless (defined ($use_pk_if_possible));

$validate_constraints='yes' unless (defined ($validate_constraints));

$ignore_errors=0 unless (defined ($ignore_errors));

# Default ports for PostgreSQL and SQL Server

$pp=5432 unless (defined ($pp));

$sp=1433 unless (defined ($sp));

$pforce_ssl=0 unless (defined ($pforce_ssl));

$stringtype_unspecified=0 unless (defined ($stringtype_unspecified));

}

# Converts numeric(4,0) and similar to int, bigint, smallint

sub convert_numeric_to_int

{

my ($qual) = @_;

croak "not a good qualifier $qual" unless ($qual =~ /^(\d+),\s*(\d+)$/);

my $precision = $1;

my $scale = $2;

croak "scale should be 0\n" unless ($scale eq '0');

return 'smallint' if ($precision <= 4);

return 'integer' if ($precision <= 9);

return 'bigint' if ($precision <= 18);

return "numeric($qual)";

}

# This is a list of the types that require a cast to be imported in kettle

# C = using CREATE CAST

# S = updating system catalog

my %types_to_cast = ('uuid' => 'C','date' => 'C','timestamp with time zone' => 'C','xml' => 'S');

# This sub adds a cast (if not defined already) if

# - we generate for kettle

# - the passed type is in %types_to_cast

sub add_cast

{

my ($type)=@_;

if (defined $types_to_cast{$type})

{

$objects->{CASTS}->{$type}=$types_to_cast{$type};

}

}

# These are the no-brainer conversions

# There is still a special case for text types and case insensitivity (see convert_type) though

my %types = ('int' => 'int',

'nvarchar' => 'varchar',

'nchar' => 'char',

'char' => 'char',

'varchar' => 'varchar',

'text' => 'text',

'char' => 'char',

'smallint' => 'smallint',

'tinyint' => 'smallint',

'bigint' => 'bigint',

'decimal' => 'numeric',

'float' => 'double precision',

'real' => 'real',

'date' => 'date',

'datetime' => 'timestamp',

'datetime2' => 'timestamp',

'smalldatetime' => 'timestamp',

'time' => 'time',

'timestamp' => 'bytea',

'rowversion' => 'bytea',

'datetimeoffset' => 'timestamp with time zone',

'image' => 'bytea',

'binary' => 'bytea',

'varbinary' => 'bytea',

'money' => 'numeric',

'smallmoney' => 'numeric(6,4)',

'uniqueidentifier' => 'uuid',

'xml' => 'xml',);

# Types with no qualifier, and no point in putting one

my %unqual = ('bytea' => 1, 'timestamp with time zone' => 1);

# This function uses the two static lists above, plus domains and citext types that

# may have been created during parsing, to convert mssql's types to pgsql's

sub convert_type

{

my ($sqlstype, $sqlqual, $colname, $tablename, $typname, $schemaname) =

@_;

my $rettype;

if (defined $types{$sqlstype})

{

if ((defined $sqlqual and defined($unqual{$types{$sqlstype}}))

or not defined $sqlqual)

{

# This is one of the few types that have to be unqualified (binary type)

$rettype = $types{$sqlstype};

# but we might add a check constraint for binary data

if ($sqlstype =~ 'binary' and defined $sqlqual) {

print STDERR "convert_type: $sqlstype, $sqlqual, $colname\n";

my $constraint;

$constraint->{TYPE} = 'CHECK_BINARY_LENGTH';

$constraint->{TABLE} = $tablename;

$constraint->{TEXT} = "octet_length(" . format_identifier($colname) . ") <= $sqlqual";

push @{$objects->{SCHEMAS}->{$schemaname}->{TABLES}->{$tablename}

->{CONSTRAINTS}}, ($constraint);

}

}

elsif (defined $sqlqual)

{

$rettype = ($types{$sqlstype} . "($sqlqual)");

}

}

# A few special cases

elsif ($sqlstype eq 'bit' and not defined $sqlqual)

{

$rettype = "boolean";

}

elsif ($sqlstype eq 'ntext' and not defined $sqlqual)

{

$rettype = "text";

}

elsif ($sqlstype eq 'numeric')

{

# Numeric is a special case:

# No qualifier. We have to use numeric

if (not $sqlqual)

{

$rettype='numeric';

}

elsif ($sqlqual !~ /\d+,\s*0/)

{

$rettype="numeric($sqlqual)";

}

elsif ( my $tmprettype=convert_numeric_to_int($sqlqual))

{

$rettype=$tmprettype;

}

else

{

$rettype="numeric($sqlqual)";

}

}

elsif ($sqlstype eq 'sysname')

{

# Special case. This is an internal type, and should seldom be used in production. Converted to varchar(128)

$rettype='varchar(128)';

}

# We special case also the geometry and geography data types

elsif ( $sqlstype =~ /^geography$|^geometry$/i )

{

# These require that the destination database contains PostGIS

unless ($requires_postgis)

{

print STDERR "WARNING: $sqlstype detected (in $schemaname.$tablename.$colname).\n You will need PostGIS (http://postgis.net/).\n The generated script will perform the CREATE EXTENSION, but please install PostGIS on this server\n";

$requires_postgis=1;

}

$rettype=lc($sqlstype);

}

elsif ($sqlstype eq 'sql_variant')

{

# There is no equivalent in PostgreSQL (and I think that's a good thing :) )

print STDERR "WARNING: $sqlstype detected (in $schemaname.$tablename.$colname).\n This is a 'not typed' field in SQL Server. There is no equivalent in PostgreSQL.\n This is converted to text, but you'll have rework to do on your client code\n";

$rettype='text';

}

else

{

print "Types: ", Dumper(\%types);

croak

"Cannot determine the PostgreSQL's datatype corresponding to $sqlstype. This is a bug\n";

}

# We special case when type is varchar, to be case insensitive

if ($sqlstype =~ /text|varchar/ and $case_insensitive)

{

$rettype = "citext";

# Do we have a SQL qualifier ? (we'll have to do check constraints then)

if ($sqlqual)

{

# Check we have a table name and a colname, or a typname

if ( defined $colname

and defined $tablename

) # We are called from a CREATE TABLE, we have to add a check constraint

{

my $constraint;

$constraint->{TYPE} = 'CHECK_CITEXT';

$constraint->{TABLE} = $tablename;

$constraint->{TEXT} = "char_length(" . format_identifier($colname) . ") <= $sqlqual";

push @{$objects->{SCHEMAS}->{$schemaname}->{TABLES}->{$tablename}

->{CONSTRAINTS}}, ($constraint);

}

elsif (defined $typname

) # We are called from a CREATE TYPE, which will be converted to a CREATE DOMAIN

{

$rettype = "citext CHECK(char_length(value)<=$sqlqual)";

}

else

{

die

"Called in a case sensitive, trying to generate a check constraint, failed. This is a bug!";

}

}

}

# We special case SQL Server's TABLE types: they should be converted into an array

if ( $sqlstype =~ /^(\S+)\.(\S+)$/)

{

# This is a namespaced type. So we check if this is a special array

my ($schema,$type)=($1,$2);

if (defined($objects->{SCHEMAS}->{$schema}->{TABLE_TYPES}->{$type}))

{

$rettype=$rettype.'[]';

}

}

# Add this type to casts to perform if necessary

add_cast($rettype);

return $rettype;

}

# This function is used for selects from SQL Server, in kettle. It adds a function call

# if there is a conversion to be done.

# uniqueidentifier is upper case in SQL Server, whereas uuid is lower case in PG

# date is converted to varchar in the YYYY-MM-DD format

# timestamp with time zone is converted to varchar in the YYYY-MM-DD HH:MI:SS.MMM (24h) with time zone format

# xml columns with empty values will be converted to null since the empty values won't be accepted in PG (datalength of an empty xml column is 5)

sub sql_convert_column

{

my ($colname,$coltype)=@_;

my %functions = (

'uuid' => 'lower({colname})',

'date' => 'convert(varchar(50), {colname}, 120)',

'timestamp with time zone(7)' => 'convert(varchar(50), {colname}, 121)',

'xml' => 'case when datalength({colname}) > 5 then {colname} else null end');

if (defined ($functions{$coltype}))

{

my $tmpcol=$functions{$coltype};

$tmpcol =~ s/\{colname\}/[$colname]/g;

return $tmpcol;

}

else

{

return "[$colname]";

}

}

# This function is used for selects from PostgreSQL, in kettle. It adds a function call

# if there is a conversion to be done.

# uuid is converted to varchar and forced to lower case

# date is converted to varchar in the YYYY-MM-DD format

# timestamp with time zone is converted to varchar in the YYYY-MM-DD HH:MI:SS.US+00 format (UTC)

sub postgres_convert_column

{

my ($colname,$coltype)=@_;

my %functions = (

'uuid' => 'lower(cast({colname} as varchar))',

'date' => 'to_char({colname}, \'YYYY-MM-DD\')',

'timestamp with time zone' => 'to_char({colname} AT TIME ZONE \'UTC\', \'YYYY-MM-DD HH:MI:SS.US+00\')');

if (defined ($functions{$coltype}))

{

my $tmpcol = $functions{$coltype};

$tmpcol =~ s/\{colname\}/$colname/;

return $tmpcol;

}

else

{

return $colname;

}

}

# This is used to convert camelCase to snake_case. The latter is more usual with PostgreSQL

sub camel_to_snake

{

my ($string)=@_;

$string =~ s/([[:lower:]])([[:upper:]]+)/$1_\l$2/g;

$string=lc($string);

return $string;

}

# This function is used to determine if a PK will be sorted the same in SQL

# Server and PostgreSQL.

# It means that it doesn't depend on collation orders or other internals.

# For now, only numeric and date data types are considered OK.

# Used for incremental jobs, to know if we can ask the databases to send us

# pre-sorted data.

# We also filter on $use_pk_if_possible.

sub is_pk_sort_order_safe

{

my ($schema,$table)=@_;

my %safe_types = ('numeric' => 1,

'int' => 1,

'bigint' => 1,

'smallint' => 1,

'real' => 1,

'double precision' => 1,

'date' => 1,

'timestamp' => 1,

);

return 0 unless (defined $objects->{SCHEMAS}->{$schema}->{TABLES}->{$table}->{PK}); # There is no PK

return 0 unless ($use_pk_if_possible =~ /\b${schema}\.${table}\b/i or $use_pk_if_possible eq '1');

my $pk=$objects->{SCHEMAS}->{$schema}->{TABLES}->{$table}->{PK};

my $isok=1;

# It is OK as long as all types are in %safe_type

foreach my $col (@{$pk->{COLS}})

{

$isok=0 unless defined ($safe_types{$objects->{SCHEMAS}->{$schema}->{TABLES}->{$table}->{COLS}->{$col}->{TYPE}});

}

return $isok;

}

# This function renames the identifiers

sub rename_identifier

{

my ($identifier)=@_;

if ($case_treatment==1)

{

$identifier=lc($identifier);

}

elsif ($case_treatment==2)

{

$identifier=camel_to_snake($identifier);

}

return $identifier;

}

# This function formats the identifiers (object name), putting double quotes around it

# It also converts case if asked

sub format_identifier

{

my ($identifier)=@_;

croak "identifier not defined in format_identifier" unless (defined $identifier);

$identifier=rename_identifier($identifier);

# Now, we protect the identifier (similar to quote_ident in PG)

$identifier=~ s/"/""/g;

$identifier='"'.$identifier.'"';

return $identifier;

}

# This is a bit of a ugly hack: for indexes, in the column definition, there may be ASC/DESC at the end

# Instead of changing the whole structure of the code, just detect this asc/desc and split it before calling format_identifier

sub format_identifier_cols_index

{

my ($idx_identifier)=@_;

$idx_identifier =~ /^(.*?)(?: (ASC|DESC))?$/;

my ($col,$order)=($1,$2);

my $formatted=format_identifier($col);

return $formatted unless defined ($order);

return $formatted . ' ' . $order;

}

# This one will try to convert what can obviously be converted from transact to (or embedded WHERE in indexes for instance) PG

# We check if we have several blocks separated by logical operators

# Things such as getdate() which can become CURRENT_TIMESTAMP

sub convert_transactsql_code

{

my ($code)=@_;

#print STDERR "convert: $code\n";

if ($code =~ /^\((.+?)\)\s+(AND|OR)\s+\((.+?)\)$/) {

my ($lhs,$op,$rhs)=($1,$2,$3);

$code = "(".convert_transactsql_code("$lhs").") $op (".convert_transactsql_code("$rhs").")";

}

elsif ($code =~ /^(.+?)\s+(AND|OR)\s+(.+?)$/) {

my ($lhs,$op,$rhs)=($1,$2,$3);

$code = "(".convert_transactsql_code("$lhs")." $op ".convert_transactsql_code("$rhs").")";

}

else {

if ($case_treatment==0)

{

$code =~ s/[\[\]]/"/gi; # Bit brutal probably

}

else

{

$code =~ s/\[(.*)\]/rename_identifier($1)/gie; # Bit brutal probably

}

$code =~ s/ISNULL\s*\(/COALESCE(/gi;

$code =~ s/getdate\s*\(\)/CURRENT_TIMESTAMP/gi;

$code =~ s/user_name\s*\(\)/CURRENT_USER/gi;

$code =~ s/datepart\s*\(\s*(.*?)\s*\,\s*(.*?)\s*\)/date_part('$1', $2)/gi;

}

#print STDERR "to: $code\n\n";

return $code;

}

# This function does its best to convert MS's weird default values syntax into something logical

sub store_default_value

{

my ($schema,$table,$col,$value,$line)=@_;

if ($value =~ /^\(?(\d+(\.\d+)?)\)?$/) # Value is numeric

{

$value = $1; # Get rid of parenthesis

if ($objects->{SCHEMAS}->{relabel_schemas($schema)}->{TABLES}->{$table}->{COLS}->{$col}->{TYPE} eq 'boolean')

{

# Ok, it IS a boolean, and we have received a number

if ($value eq '0')

{

$objects->{SCHEMAS}->{$schema}->{TABLES}->{$table}->{COLS}->{$col}->{DEFAULT}->{VALUE} = 'false';

}

elsif ($value eq '1')

{

$objects->{SCHEMAS}->{$schema}->{TABLES}->{$table}->{COLS}->{$col}->{DEFAULT}->{VALUE} = 'true';

}

else

{

# We should not get here: we have a numeric which isn't 0 or 1, and is supposed to be a boolean

die "Got an unexpected boolean : $value, for line $line\n";

}

}

else

{

$objects->{SCHEMAS}->{$schema}->{TABLES}->{$table}->{COLS}->{$col}->{DEFAULT}->{VALUE}

= $value;

}

$objects->{SCHEMAS}->{$schema}->{TABLES}->{$table}->{COLS}->{$col}->{DEFAULT}->{UNSURE}

= 0;

}

elsif ($value =~ /^NULL$/) # A NULL value

{

# NULL WITHOUT quotes around it !

$objects->{SCHEMAS}->{$schema}->{TABLES}->{$table}->{COLS}->{$col}->{DEFAULT}->{VALUE}

= 'NULL';

$objects->{SCHEMAS}->{$schema}->{TABLES}->{$table}->{COLS}->{$col}->{DEFAULT}->{UNSURE}

= 0;

}

elsif ($value =~ /^N?'(.*)'$/) # There is sometimes an N before a string.

{

$value = $1; # Get rid of junk

# Default text value, text, between commas

$objects->{SCHEMAS}->{$schema}->{TABLES}->{$table}->{COLS}->{$col}->{DEFAULT}->{VALUE}

= "'$1'";

$objects->{SCHEMAS}->{$schema}->{TABLES}->{$table}->{COLS}->{$col}->{DEFAULT}->{UNSURE}

= 0;

}

else

{

#This must be a function call...

$objects->{SCHEMAS}->{$schema}->{TABLES}->{$table}->{COLS}->{$col}->{DEFAULT}->{VALUE}

= convert_transactsql_code($value);

$objects->{SCHEMAS}->{$schema}->{TABLES}->{$table}->{COLS}->{$col}->{DEFAULT}->{UNSURE}

= 1;

}

}

# This gives the next column position for a table

# It is used when we add a new column

# These tables are added at the end of the table, in %objects

sub next_col_pos

{

my ($schema, $table) = @_;

if (defined $objects->{SCHEMAS}->{$schema}->{TABLES}->{$table}->{COLS})

{

my $max = 0;

foreach my $col (

values(%{$objects->{SCHEMAS}->{$schema}->{TABLES}->{$table}->{COLS}}))

{

if ($col->{POS} > $max)

{

$max = $col->{POS};

}

}

return $max + 1;

}

elsif (defined $objects->{SCHEMAS}->{$schema}->{TABLES}->{$table})

{

# First column

return 1;

}

else

{

die "We tried to add a column to an unknown table";

}

}

{

# This builds %relabel_schemas for use in the next function. Both are scoped so that %relabel_schemas is not visible from outside

my %relabel_schemas;

sub build_relabel_schemas

{

# Don't forget dbo -> public if it was asked (default)

unless ($norelabel_dbo)

{

$relabel_schemas{'dbo'}='public';

}

# dbo can be overwritten in relabel_schema (the user will probably forget to deactivate the relabel).

# so we do the real relabeling after the norelabel_dbo, to overwrite

if (defined $relabel_schemas)

{

foreach my $pair (split (';',$relabel_schemas))

{

my @pair=split('=>',$pair);

unless (scalar(@pair)==2)

{

die "Cannot parse the schema list given as argument: <$relabel_schemas>\n";

}

$relabel_schemas{$pair[0]}=$pair[1];

}

}

}

# This relabels the schemas

sub relabel_schemas

{

my ($schema) = @_;

return $schema unless (defined $relabel_schemas{$schema});

return $relabel_schemas{$schema};

}

}

# Test if we are on windows. We will have to convert / to \ in the XML files

sub is_windows

{

if ($^O =~ /MSWin32/)

{

return 1;

}

return 0;

}

# Die if kettle is not set up correctly

sub kettle_warn

{

my ($file) = @_;

print STDERR

"Important WARNING: You have to set up KETTLE_EMPTY_STRING_DIFFERS_FROM_NULL=Y in $file.\nIf this file doesn't exist yet, start spoon from the kettle directory once.";

}

# This sub checks ~/.kettle/kettle.properties to be sure

# KETTLE_EMPTY_STRING_DIFFERS_FROM_NULL=Y is in place

# We die if not

sub check_kettle_properties

{

my $ok = 0;

my $file;

if (!is_windows())

{

$file = $ENV{'HOME'} . '/.kettle/kettle.properties';

}

else

{

$file = $ENV{'USERPROFILE'} . '/.kettle/kettle.properties';

}

unless (open FILE, $file )

{

kettle_warn($file);

return 1;

}

while (<FILE>)

{

next unless (/^KETTLE_EMPTY_STRING_DIFFERS_FROM_NULL=Y$/);

$ok = 1;

}

close FILE;

if (not $ok)

{

kettle_warn($file);

return 1;

}

return 0;

}

# Usage, obviously. Has to be kept in sync with new command line options

sub usage

{

print qq{

Usage:

sqlserver2pgsql -b BEFORE_FILE -a AFTER_FILE -u UNSURE_FILE -f SQLSERVER_SCHEMA_FILE

Description:

This script is a migration tool to convert a Microsoft SQL Server Database

into a PostgreSQL database, as automatically as possible. It takes a Sql

Server SQL schema dump, and creates a postgresql dump in 3 parts:

before_file, after_file and unsure_file.

Optionnaly, using the '-k' option, it will generate a kettle job to

transfer all data.

Mandatory options:

SQL Server schema input file:

-f SQLSERVER_SCHEMA_FILE

a readable SQL Server SQL structure dump.

PostgreSQL output schema files:

-b BEFORE_SCRIPT

contains what is needed to import data: types, tables and columns.

-a AFTER_SCRIPT

contains the rest: indexes, constraints.

-u UNSURE_SCRIPT

contains objects we attempt to migrate, but cannot guarantee, such

as views or complex indexes.

Options:

-conf CONFIGURATION_FILE

uses a configuration file. All options can be set there. Command

line options will overwrite conf options.

-i the resulting PostgreSQL names will be case-insensitive.

-nr the SQL Server 'dbo' schema will not be translated to PostgreSQL

'public' schema. 'dbo' will stay 'dbo'.

-camel_to_snake

all object names are converted from 'camelCase' to 'camel_case',

which is more often used in PostgreSQL. Do not use this unless you

are ready to do SQL query changes in the client.

-relabel_schemas 'SOURCE1=>DEST1;SOURCE2=>DEST2'

gives a list of schemas to rename. Quote this option to prevent the

shell to alter it. The '-nr' option cancels the default 'dbo' to

'public' remapping.

-keep_identifier_case

keep the case of SQL server database objects. This option is not

advised. Default is to lowercase everything.

-num convert numeric 'xxx,0' to int, bigint, etc. Will not keep numeric

scale and precision for the converted.

-validate_constraints {yes|no|after}

should the constraints be validated. Set to "yes" by default. If

set to "no", the constraints will be set as 'NOT VALID'. If set to

"after", the constraints will be set as

'NOT VALID', but the validation queries will be enforced in the

unsure file.

-use_pk_if_possible {0|1|LIST_OF_TABLES} (Default 0)

determine if a primary key will be sorted in the same way in SQL

Server and PostgreSQL. Default '0', do not sort tables. If set to

'1' sort all tables. LIST_OF_TABLES gives a comma separated list of

tables to sort in the form 'schema1.table1,schema2.table2'. Cases

are compared insensitively.

Kettle options:

if you are generating for kettle, you must provide connection information.

-sd SQLSERVER_DATABASE

-sh SQLSERVER_HOST

-si SQLSERVER_HOST

-sp SQLSERVER_PORT

-su SQLSERVER_USERNAME

-sw SQLSERVER_PASSWORD

-pd POSTGRESQL_DATABASE

-ph POSTGRESQL_HOST

-pp POSTGRESQL_PORxT

-pu POSTGRESQL_USERNAME

-pw POSTGRESQL_PASSWORD

-pi PARALLELISM_IN

parallelism level for the kettle job (input, SQL Server). Default 1.

-po PARALLELISM_OUT

parallelism level for the kettle job (output, PostgreSQL). Default 8.

-pforce_ssl

force a SSL session to PostgreSQL

-stringtype_unspecified

By default, the data are sent as varchar. This can block date field

migration. By using this option, make PostgreSQL infer the data type.

-k KETTLE_OUTPUT_DIRECTORY

-sort_size INTEGER

set the size of the sort batch for the incremental job. If set to a

small value, the process will be slow. If set to a too large value,

you will get Java 'Out of Heap Memory' errors.

The default value is 10000, which is very low, in order to avoid

problems. First, raise Java 'Heap Memory' in the kitchen script,

then try higher values if you need more speed.

-ignore_errors

choose to ignore insert errors. If this option is used, inserting

will be much slower.

};

exit 0;

}

# This function generates kettle transformations, and a kettle job running all these

# transformations sequentially, for all the tables, in all the schemas, in sql server's dump

sub generate_kettle

{

my ($dir) = @_;

# first, create the kettle directory

unless (-d $dir)

{

mkdir($dir) or die "Cannot create $dir";

}

# For each table in each schema in $objects, we generate a kettle file in the directory

# We also create an incremental transformation

foreach my $schema (sort keys %{$objects->{SCHEMAS}})

{

my $refschema = $objects->{SCHEMAS}->{$schema};

my $targetschema = $schema;

foreach my $table (sort keys %{$refschema->{TABLES}})

{

my $origschema=$refschema->{TABLES}->{$table}->{origschema};

# First, does this table have LOBs ? The template depends on this and is this

# table having an int PK ?

my $newtemplate;

if ( $refschema->{TABLES}->{$table}->{haslobs}

and defined($refschema->{TABLES}->{$table}->{PK}->{COLS})

and scalar(@{$refschema->{TABLES}->{$table}->{PK}->{COLS}}) == 1

and ($refschema->{TABLES}->{$table}->{COLS}->{($refschema->{TABLES}->{$table}->{PK}->{COLS}->[0])}->{TYPE} =~ /int$/)

)

{

my $wherefilter;

$newtemplate = $template_lob;

$wherefilter =

'WHERE '

. $refschema->{TABLES}->{$table}->{PK}->{COLS}->[0]

. '% ${Internal.Step.Unique.Count} = ${Internal.Step.Unique.Number}';

$newtemplate =~

s/__sqlserver_where_filter__/$wherefilter/;

}

else

{

$newtemplate = $template;

}

# We have a similar question for incremental jobs: can we use the primary key ?

# We obviously need a primary key, and we need the sort order to be the same in both databases

my $newincrementaltemplate;

if (is_pk_sort_order_safe($schema,$table))

{

$newincrementaltemplate=$incremental_template_sortable_pk;

my $collist=join(',',@{$refschema->{TABLES}->{$table}->{PK}->{COLS}});

$newincrementaltemplate =~ s/__sqlserver_pk_condition__/$collist/g;

$newincrementaltemplate =~ s/__pg_pk_condition__/$collist/g;

}

else

{

$newincrementaltemplate=$incremental_template;

}

# Build the column list of the table to put into the SQL Server query

my @colsdef;

my @pgcolsdef;

foreach my $col (

sort {

$refschema->{TABLES}->{$table}->{COLS}->{$a}->{POS}

<=> $refschema->{TABLES}->{$table}->{COLS}->{$b}

->{POS}

} (keys %{$refschema->{TABLES}->{$table}->{COLS}}))

{

my $coldef = sql_convert_column($col,$refschema->{TABLES}->{$table}->{COLS}->{$col}->{TYPE}) . " AS " . format_identifier($col);

my $pgcoldef = postgres_convert_column(format_identifier($col),$refschema->{TABLES}->{$table}->{COLS}->{$col}->{TYPE}) . " AS " . format_identifier($col);

push @colsdef,($coldef);

push @pgcolsdef,($pgcoldef);

}

my $colsdef=join(',',@colsdef);

my $pgcolsdef=join(',',@pgcolsdef);

my $pgtable=format_identifier($table);

my $pgschema=format_identifier($targetschema);

my $sqlinstancename = '';

if (defined $si) {

$sqlinstancename = $si;

}

# Substitute every connection placeholder with the real value. We do this for both templates

$newtemplate =~ s/__sqlserver_database__/$sd/g;

$newtemplate =~ s/__sqlserver_database__/$sd/g;

$newtemplate =~ s/__sqlserver_host__/$sh/g;

$newtemplate =~ s/__sqlserver_port__/$sp/g;

$newtemplate =~ s/__sqlserver_instance__/$sqlinstancename/g;

$newtemplate =~ s/__sqlserver_username__/$su/g;

$newtemplate =~ s/__sqlserver_password__/$sw/g;

$newtemplate =~ s/__postgres_database__/$pd/g;

$newtemplate =~ s/__postgres_host__/$ph/g;

$newtemplate =~ s/__postgres_port__/$pp/g;

$newtemplate =~ s/__postgres_username__/$pu/g;

$newtemplate =~ s/__postgres_password__/$pw/g;

$newtemplate =~ s/__sqlserver_table_name__/[$origschema].[$table]/g;

$newtemplate =~ s/__sqlserver_table_cols__/$colsdef/g;

$newtemplate =~ s/__postgres_table_name__/$pgtable/g;

$newtemplate =~ s/__postgres_schema_name__/$pgschema/g;

$newtemplate =~ s/__PARALLELISM_IN__/$parallelism_in/g;

$newtemplate =~ s/__PARALLELISM_OUT__/$parallelism_out/g;

if ($ignore_errors)

{

$newtemplate =~ s/<ignore_errors>N<\/ignore_errors>/<ignore_errors>Y<\/ignore_errors>/g;

$newtemplate =~ s/<use_batch>Y<\/use_batch>/<use_batch>N<\/use_batch>/g; # Cannot use batch mode with ignore errors

}

if ($pforce_ssl)

{

$newtemplate =~ s/__pforce_ssl__/<attribute><code>EXTRA_OPTION_POSTGRESQL.ssl<\/code><attribute>true<\/attribute><\/attribute>\n<attribute><code>EXTRA_OPTION_POSTGRESQL.sslfactory<\/code><attribute>org.postgresql.ssl.NonValidatingFactory<\/attribute><\/attribute>/g;

}

else

{

$newtemplate =~ s/__pforce_ssl__//g;

}

if ($stringtype_unspecified) {

$newtemplate =~ s/__stringtype_unspecified__/<attribute><code>EXTRA_OPTION_POSTGRESQL.stringtype<\/code><attribute>unspecified<\/attribute><\/attribute>\n/g;

}

else {

$newtemplate =~ s/__stringtype_unspecified__//g;

}

$newincrementaltemplate =~ s/__sqlserver_database__/$sd/g;

$newincrementaltemplate =~ s/__sqlserver_database__/$sd/g;

$newincrementaltemplate =~ s/__sqlserver_host__/$sh/g;

$newincrementaltemplate =~ s/__sqlserver_port__/$sp/g;

$newincrementaltemplate =~ s/__sqlserver_username__/$su/g;

$newincrementaltemplate =~ s/__sqlserver_password__/$sw/g;

$newincrementaltemplate =~ s/__postgres_database__/$pd/g;

$newincrementaltemplate =~ s/__sqlserver_instance__/$sqlinstancename/g;

$newincrementaltemplate =~ s/__postgres_host__/$ph/g;

$newincrementaltemplate =~ s/__postgres_port__/$pp/g;

$newincrementaltemplate =~ s/__postgres_username__/$pu/g;

$newincrementaltemplate =~ s/__postgres_password__/$pw/g;

$newincrementaltemplate =~ s/__sqlserver_table_name__/[$origschema].[$table]/g;

$newincrementaltemplate =~ s/__sqlserver_table_cols__/$colsdef/g;

$newincrementaltemplate =~ s/__postgres_table_name__/$pgtable/g;

$newincrementaltemplate =~ s/__postgres_schema_name__/$pgschema/g;

$newincrementaltemplate =~ s/__postgres_table_cols__/$pgcolsdef/g;

$newincrementaltemplate =~ s/__PARALLELISM_IN__/$parallelism_in/g;

$newincrementaltemplate =~ s/__PARALLELISM_OUT__/$parallelism_out/g;

$newincrementaltemplate =~ s/__sort_size__/$sort_size/g;

if ($pforce_ssl)

{

$newincrementaltemplate =~ s/__pforce_ssl__/<attribute><code>EXTRA_OPTION_POSTGRESQL.ssl<\/code><attribute>true<\/attribute><\/attribute>\n<attribute><code>EXTRA_OPTION_POSTGRESQL.sslfactory<\/code><attribute>org.postgresql.ssl.NonValidatingFactory<\/attribute><\/attribute>/g;

}

else

{

$newincrementaltemplate =~ s/__pforce_ssl__//g;

}

if ($stringtype_unspecified) {

$newincrementaltemplate =~ s/__stringtype_unspecified__/<attribute><code>EXTRA_OPTION_POSTGRESQL.stringtype<\/code><attribute>unspecified<\/attribute><\/attribute>\n/g;

}

else {

$newincrementaltemplate =~ s/__stringtype_unspecified__//g;

}

# We have a bit of work to do on primary keys for the incremental template: we need them

# to compare the tables…

if (defined($refschema->{TABLES}->{$table}->{PK}->{COLS}))

{

my @pk=@{$refschema->{TABLES}->{$table}->{PK}->{COLS}};

my $keys="";

foreach my $pk(@pk)

{

$keys.="<key>$pk</key>\n";

}

$newincrementaltemplate =~ s/__KEYS_MERGE__/$keys/g;

my $sortkeys='';

my $synckeys='';

foreach my $pk(@pk)