#!/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, $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;

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

}

# 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

my %types_to_cast = ('uuid' => '1',);

# 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}->{uuid}=1;

}

}

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

'smalldatetime' => 'timestamp',

'time' => 'time',

'timestamp' => 'timestamp',

'image' => 'bytea',

'binary' => 'bytea',

'varbinary' => 'bytea',

'money' => 'numeric',

'smallmoney' => 'numeric',

'uniqueidentifier' => 'uuid',);

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

my %unqual = ('bytea' => 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};

}

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

}

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

sub sql_convert_column

{

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

my %functions = ( 'uuid' => 'lower');

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

{

return $functions{$coltype} . '([' . $colname . '])';

}

else

{

return "[$colname]";

}

}

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

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

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;

}

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

# Things such as getdate() which can become CURRENT_TIMESTAMP

sub convert_transactsql_code

{

my ($code)=@_;

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

return $code;

}

# 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 =~ /win/i)

{

return 1;

}

return 0;

}

# Die if kettle is not set up correctly

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

}

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

}

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

while (<FILE>)

{

next unless (/KETTLE_EMPTY_STRING_DIFFERS_FROM_NULL\s*=\s*Y/);

$ok = 1;

}

close FILE;

if (not $ok)

{

kettle_die($file);

}

return 0;

}

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

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

}

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

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

push @colsdef,($coldef);

}

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

my $pgtable=format_identifier($table);

my $pgschema=format_identifier($targetschema);

# 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_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/__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/__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

#FIXME: unless (scalar(grep(/^${colname}$/,@pk))) # Does grep find an element in the array matching colname ?

# {

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

{

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

}

}

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

$job_header =~ s/__SQL_SCRIPT_END__/$afterscript/g;

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

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

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

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

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

print JOBFILE $job_header;

print JOBFILE $entries;

print JOBFILE $job_middle;

print JOBFILE $hops;

print JOBFILE $job_footer;

close JOBFILE;

print INCFILE $job_header;

print INCFILE $incentries;

print INCFILE $job_middle;

print INCFILE $hops;

print INCFILE $job_footer;

close INCFILE;

}

# sql server's dump may contain multiline C style comments (/* */)

# This sub reads a line and cleans it up, removing comments, \r, exec sp_executesql,...

# It takes into account the status (in or out of comment) of the previous line, hence

# the scoped $in_comment

{

my $in_comment = 0;

sub read_and_clean

{

my ($fd) = @_;

my $line = <$fd>;

return undef if (not defined $line);

$line =~ s/\r//g; # Remove \r from windows output

$line =~

s/EXEC(ute)?\s*(dbo|sys)\.sp_executesql( \@statement =)? N'//i

; # Remove executesql… it's a bit weird in the SQL Server's dump

# If we are not in comment, we look for /*

# If we are in comment, we look for */, and we remove everything until */

if ($in_comment)

{

if ($line =~ /\*\//)

{

$in_comment = 0;

$line =~ s/.*\*\///; # Remove everything before the uncomment

}

else

{

$line = "\n";

}

}

if (not $in_comment)

{

# We first remove all one-line only comments (there may be several on this line)

$line =~ s/\/\*.*?\*\///g;

# Is there a comment left ?

if ($line =~ /\/\*/)

{

$in_comment = 1;

$line =~ s/\/\*.*//; # Remove everything after the comment

}

}

return $line;

}

}

# This adds a column we just read to a table

sub add_column_to_table

{

my ($schemaname,$tablename,$colname,$coltypeschema,$coltype,$colqual,$isidentity,$colisnull)=@_;

my $colnumber=next_col_pos($schemaname,$tablename);

if (defined $coltypeschema)

{

# The datatype is a user defined datatype

# It has already been declared before. We just need to find it

$coltype = relabel_schemas($coltypeschema) . '.' . $coltype;

}

if ($colqual)

{

if ($colqual eq '(max)')

{

$colqual = undef

; # max in sql server is the same as putting no colqual in pg

}

else

{

# We need the number (or 2 numbers) in this qual

$colqual =~ /\((\d+(?:,\s*\d+)?)\)/

or die "Cannot parse colqual <$colqual>";

$colqual = "$1";

}

}

my $newtype =

convert_type($coltype, $colqual, $colname,

$tablename, undef, $schemaname);

# If it is an identity, we'll map to serial/bigserial (create a sequence, then link it

# to the column)

if ($isidentity)

{

# We have an identity field. We remember the default value and

# initialize the sequence correctly in the after script

$isidentity =~ /IDENTITY\((\d+),\s*(\d+)\)/

or die "Cannot understand <$isidentity>";

my $startseq = $1;

my $stepseq = $2;

my $seqname = lc("${tablename}_${colname}_seq");

# We get a sure default value.

$objects->{SCHEMAS}->{$schemaname}->{TABLES}->{$tablename}->{COLS}

->{$colname}->{DEFAULT}->{VALUE} =

"nextval('"

. format_identifier(relabel_schemas(${schemaname})) . '.'

. ${seqname} . "')";

$objects->{SCHEMAS}->{$schemaname}->{TABLES}->{$tablename}->{COLS}

->{$colname}->{DEFAULT}->{UNSURE} = 0;

$objects->{SCHEMAS}->{$schemaname}->{SEQUENCES}->{$seqname}->{START}

= $startseq;

$objects->{SCHEMAS}->{$schemaname}->{SEQUENCES}->{$seqname}->{STEP}

= $stepseq;

$objects->{SCHEMAS}->{$schemaname}->{SEQUENCES}->{$seqname}

->{OWNERTABLE} = $tablename;

$objects->{SCHEMAS}->{$schemaname}->{SEQUENCES}->{$seqname}

->{OWNERCOL} = $colname;

$objects->{SCHEMAS}->{$schemaname}->{SEQUENCES}->{$seqname}

->{OWNERSCHEMA} = $schemaname;

}

# If there is a bytea generated, this table will contain a blob:

# use a special kettle transformation for it if generating kettle

# (see generate_kettle() )

if ( $newtype eq 'bytea'

or $coltype eq

'ntext') # Ntext is very slow, stored out of page