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

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

}

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 qq{

};

exit 0;

}

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