use Getopt::Long;

use Data::Dumper;

use Cwd;

use Encode::Guess;

use Carp;

use strict;

my $objects;

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 $requires_postgis=0;

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)

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',

);

# 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));

}

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)";

}

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

sub add_cast

{

my ($type)=@_;

if (defined $types_to_cast{$type})

{

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

}

}

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',);

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

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};

}

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;

}

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]";

}

}

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;

}

}

sub camel_to_snake

{

my ($string)=@_;

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

$string=lc($string);

return $string;

}

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;

}

sub rename_identifier

{

my ($identifier)=@_;

if ($case_treatment==1)

{

$identifier=lc($identifier);

}

elsif ($case_treatment==2)

{

$identifier=camel_to_snake($identifier);

}

return $identifier;

}

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;

}

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;

}

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;

}

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;

}

}

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};

}

}

sub is_windows

{

if ($^O =~ /MSWin32/)

{

return 1;

}

return 0;

}

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.";

}

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;

}

sub usage

{

print

"$0 [-k kettle_output_directory] -b before_file -a after_file -u unsure_file -f sql_server_schema_file[-h] [-i]\n";

print

"\nExpects a SQL Server SQL structure dump as -f (preferably unicode)\n";

print "-conf uses a conf file. All options below can also be set there. Command line options will overwrite conf options\n";

print "-i tells $0 to create a case-insensitive PostgreSQL schema\n";

print

"-nr tells $0 not to convert the dbo schema to public. dbo will stay dbo\n";

print

"-num tells $0 to convert numeric xxx,0 to int, bigint, etc. Will not keep numeric scale and precision for the converted\n";

print

"-relabel_schemas gives a list of schemas to rename. For instance -relabel_schemas 'source1=>dest1;source2=>dest2'\n";

print " -nr simply cancels the default dbo=>public remapping. Don't forget to put the remapping between quotes\n";

print

"-keep_identifier_case tells $0 to keep the case of sql server database objects (not advised). Default is to lowercase everything.\n";

print

"-camel_to_snake tells $0 to convert the object names from camelCase to camel_case, which is more often used in PostgreSQL.\nDon't use this unless you are ready to do changes in the client.\n";

print "before_file contains the structure\n";

print "after_file contains index, constraints\n";

print "validate_constraints validates the constraints that have been created\n";

print "sort_size will change size of sort batch for the incremental job. Too small and it will be slow, too big and you will get Java Out of Heap Memory errors.\n";

print "sort_size is 10000, which is very low, to try to avoid problems. First, raise java heap memory (in the kitchen script), then try higher values if you need more speed\n";

print "use_pk_if_possible is false (0) by default. You can put it to 1 (true), or give a comma separated list of tables (with schema). Compared case insensitively\n";

print

"unsure_file contains things we cannot guarantee will work, such as views\n";

print "\n";

print

"If you are generating for kettle, you'll need to provide connection information\n";

print "for connecting to both databases:\n";

print "-sd: sqlserver database\n";

print "-sh: sqlserver host\n";

print "-si: sqlserver host instance\n";

print "-sp: sqlserver port\n";

print "-su: sqlserver username\n";

print "-sw: sqlserver password\n";

print "-pd: postgresql database\n";

print "-ph: postgresql host\n";

print "-pp: postgresql port\n";

print "-pu: postgresql username\n";

print "-pw: postgresql password\n";

print "-pi: parallelism level for the kettle job (input, sql server)\n";

print "-po: parallelism level for the kettle job (output, postgresql)\n";

print "-pforce_ssl: force a SSL session to PostgreSQL\n";

print "You may also choose to ignore insert errors (inserting will be much slower)\n";

print "-ignore_errors\n";

}

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;

}

$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;

}

# 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)

{

$sortkeys.="<field>\n<name>$pk</name>\n<ascending>Y</ascending>\n<case_sensitive>Y</case_sensitive>\n</field>\n";

my $outcol=$pk;

$outcol=rename_identifier($outcol);

$synckeys.="<key>\n<name>$pk</name>\n<field>$outcol</field>\n<condition>&#x3d;</condition>\n<name2/>\n</key>\n";

}

$newincrementaltemplate =~ s/__SORT_KEYS_SQLSERVER__/$sortkeys/g;

$newincrementaltemplate =~ s/__SORT_KEYS_PG__/$sortkeys/g;

$newincrementaltemplate =~ s/__KEYS_SYNC__/$synckeys/g;

# We also need to tell the merge step to compare all columns

my $valuesmerge='';

my $valuessync='';

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

{

# Is it a member of the PK ? If yes, no need to use it for comparison

$valuesmerge.="<value>$colname</value>\n";

# we need to use the correct case for postgresql output

my $outcol=$colname;

if ($case_treatment==1)

{

$outcol=lc($outcol);

}

elsif ($case_treatment==2)

{

$outcol=camel_to_snake($outcol);

}

$valuessync.="<value>\n<name>$outcol</name>\n<rename>$colname</rename>\n<update>Y</update>\n</value>\n";

}

$newincrementaltemplate =~ s/__VALUES_MERGE__/$valuesmerge/g;

$newincrementaltemplate =~ s/__VALUES_SYNC__/$valuessync/g;

# Produce the incremental transformation

open FILE, ">$dir/incremental-$schema-$table.ktr"

or die "Cannot write to $dir/incremental-$schema-$table.ktr";

binmode(FILE,":utf8");

print FILE $newincrementaltemplate;

close FILE;

}

else

{

print STDERR "$schema/$table has no PK. Cannot create an incremental transformation\n";

}

# Store this new transformation into its file

open FILE, ">$dir/$schema-$table.ktr"

or die "Cannot write to $dir/$schema-$table.ktr";

binmode(FILE,":utf8");

print FILE $newtemplate;

close FILE;

}

}

# All transformations are done

# We have to create a job to launch everything in one go

# We also create an incremental job. This incremental job

# first deactivates all constraints (with triggers), then

# runs all incremental jobs, and "standard" jobs for all those

# tables where we cannot do incremental (no PK...)

open JOBFILE, ">$dir/migration.kjb"

or die "Cannot write to $dir/migration.kjb";

open INCFILE, ">$dir/incremental.kjb"

or die "Cannot write to $dir/incremental.kjb";

my $real_dir = getcwd;

my $entries = '';

my $incentries = '';

my $hops = '';

my $prev_node = 'SQL SCRIPT START';

# $cur_vert_pso is not that useful, it's just not to be ugly if someone wanted to open

# the job with spoon (kettle's gui) and work on it graphically

my $cur_vert_pos = 100;

# First : the hop from START to SQL SCRIPT START

my $tmp_hop = $job_hop;

$tmp_hop =~ s/__table_1__/START/;

$tmp_hop =~ s/__table_2__/SQL SCRIPT START/;

# This is the first hop, so it is unconditionnal

$tmp_hop =~ s/<unconditional>N<\/unconditional>/<unconditional>Y<\/unconditional>/;