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

our $validate_constraints='yes';

our $parallelism=8;

our $sort_size=10000;

our $use_pk_if_possible=0;

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

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

'validate constraints' => 'validate_constraints',

'sort size' => 'sort_size',

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

);

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

}

# Hard coded default values

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

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

close CONF;

}

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

'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}))

{

return $functions{$coltype} =~ s/\{colname\}/[$colname]/gr;

}

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

{

return $functions{$coltype} =~ s/\{colname\}/"$colname"/r;

}

else

{

return "\"$colname\"";

}

}

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 format_identifier

{

my ($identifier)=@_;

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

unless ($keep_identifier_case)

{

$identifier=lc($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)=@_;

$code =~ s/[\[\]]/\"/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;

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_die

{

my ($file) = @_;

die

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

}

open FILE, $file or kettle_die($file);

while (<FILE>)

{

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

$ok = 1;

}

close FILE;

if (not $ok)

{

kettle_die($file);

}

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 "before_file contains the structure\n";

print "after_file contains index, constraints\n";

print "validate_constraint 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";

}

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($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 (length $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__/$parallelism/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__/$parallelism/g;

$newincrementaltemplate =~ s/__sort_size__/$sort_size/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)

{

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

my $outcol=$pk;

unless ($keep_identifier_case)

{

$outcol=lc($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;

unless ($keep_identifier_case)

{

$outcol=lc($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>/;

$hops.=$tmp_hop;

# We sort only so that it will be easier to find a transformation in the job if one needed to

# edit it. It's also easier to track progress if tables are sorted alphabetically

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

{

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

foreach my $table (sort { lc($a) cmp lc($b) }

keys %{$refschema->{TABLES}})

{

my $tmp_entry = $job_entry;

# We build the entries with regexp substitutions. The tablename contains the schema

$tmp_entry =~ s/__table_name__/${schema}_${table}/;

$tmp_entry =~ s/__y_loc__/$cur_vert_pos/;

# JOBFILEname to use. We need the full path to the transformations

# The only difference between normal and incremental job is the filename of the transformation

my $JOBFILEname;

my $INCJOBFILEname;

if ($dir =~ /^(\\|\/)/) # Absolute path

{

$JOBFILEname = $dir . '/' . $schema . '-' . $table . '.ktr';

$INCJOBFILEname = $dir . '/' . 'incremental-' . $schema . '-' . $table . '.ktr';

}

else

{

$JOBFILEname =

$real_dir . '/'

. $dir . '/'

. $schema . '-'

. $table . '.ktr';

$INCJOBFILEname =

$real_dir . '/'

. $dir . '/'

. 'incremental-'

. $schema . '-'

. $table . '.ktr';

}

# Does the incremental transformation exist ?

unless (-e $INCJOBFILEname)

{

$INCJOBFILEname=$JOBFILEname;

}

# Different for windows and linux, obviously: we change / to \ for windows

unless (is_windows())

{

$JOBFILEname =~ s/\//&#47;/g;

$INCJOBFILEname =~ s/\//&#47;/g;

}

else

{

$JOBFILEname =~ s/\//\\/g;

$INCJOBFILEname =~ s/\//\\/g;

}

my $inctmp_entry=$tmp_entry;

$tmp_entry =~ s/__file_name__/$JOBFILEname/;

$inctmp_entry =~ s/__file_name__/$INCJOBFILEname/;

$entries .= $tmp_entry;

$incentries.=$inctmp_entry;

# We build the hop with the regexp too

my $tmp_hop = $job_hop;

$tmp_hop =~ s/__table_1__/$prev_node/;

$tmp_hop =~ s/__table_2__/${schema}_${table}/;

$hops .= $tmp_hop;

# We increment everything for next loop

$prev_node = "${schema}_${table}"; # For the next hop

$cur_vert_pos += 80; # To be pretty in spoon

}

}

# Put the final hop

$tmp_hop = $job_hop;

$tmp_hop =~ s/__table_1__/$prev_node/;

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

$hops .= $tmp_hop;

# Build the casts in the start/stop job entries

my $beforescript;

my $afterscript;

if (defined ($objects->{CASTS}))

{

foreach my $cast (keys %{$objects->{CASTS}})

{

if ($objects->{CASTS}->{$cast} eq "C")

{

$beforescript.= "DROP CAST IF EXISTS &#x28;varchar as $cast&#x29;;\n";

$beforescript.= "CREATE CAST &#x28;varchar as $cast&#x29; with inout as implicit;\n";

$afterscript.= "DROP CAST &#x28;varchar as $cast&#x29;;\n";

}

elsif ($objects->{CASTS}->{$cast} eq "S")

{

$beforescript.= "UPDATE pg_cast SET castcontext='i' WHERE castsource='character varying'::regtype AND casttarget='$cast'::regtype;\n";

$afterscript.= "UPDATE pg_cast SET castcontext='e' WHERE castsource='character varying'::regtype AND casttarget='$cast'::regtype;\n";

}

}

}

# Remove/restore triggers to be able to insert without FK checks

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

{

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

foreach my $table (sort { lc($a) cmp lc($b) }

keys %{$refschema->{TABLES}})

{

$beforescript.= "ALTER TABLE " . format_identifier($schema) . '.' . format_identifier($table) . " DISABLE TRIGGER ALL;\n";

$afterscript.= "ALTER TABLE " . format_identifier($schema) . '.' . format_identifier($table) . " ENABLE TRIGGER ALL;\n";

}

}

# This is for the SQL Scripts. We also need to specify the PG connection

$job_header =~ s/__SQL_SCRIPT_INIT__/$beforescript/g;