Snowflake: Schema to Schema Data Movement with Procedure

One question might be in your mind, why not clone rather than data copy from one schema to another in snowflake?

Why clone may not help?

1. Stage: Existing Schema might have Stages pointing to some S3 path or Azure Blob & Internal (i.e. Snowflake) named stages that cannot be cloned.
2. Task: Tasks will go into the suspended by default once you clone them.
3. Stream: If a table is cloned, historical data for the table clone begins at the time/point when the clone was created. Any unconsumed records in the streams (in the clone) are inaccessible.
4. Pipe: When a database or schema is cloned, any pipes in the source container that reference an internal (i.e. Snowflake) stage are not cloned.

For handling the above concerns, we can simply copy the data from one schema to another, and the rest of the objects and pointers can be kept as it is. This can be an ideal scenario for production support where actual data might be needed to debug the root cause.

One Schema can have 100s or 1000s of tables. These can be queried from information schema, and also filters can be kept in place to filter out unnecessary tables like raw_ or stg_ tables.

This can be done by writing the query builder SQL to generate insert statements for all required tables using the pattern as optional

var statements_insert = snowflake.createStatement(
{
sqlText: `select ‘INSERT OVERWRITE INTO ‘ || TABLE_CATALOG || ‘.”${in_tgt_schema}”.”’ || table_name ||
‘“ SELECT * FROM ‘ || TABLE_CATALOG || ‘.”${in_src_schema}”.”’ || table_name || ‘“;’ AS SQL_COMMAND ,
TABLE_NAME, ROW_COUNT from information_schema.tables where TABLE_CATALOG = ‘${in_db_name}’ and TABLE_SCHEMA =
‘${in_tgt_schema}’ and TABLE_TYPE = ‘BASE TABLE’ and (TABLE_NAME not like ‘RAW_%’ or TABLE_NAME not like
‘${in_ntlike}’) and TABLE_NAME like ‘${in_tbl_like}’ order by ROW_COUNT asc`});

iterate the result in a while loop without a try-catch block, this will execute your all insert statements even if some of them are failing due to errors like column count mismatch, etc. you can track the status by creating the audit table.

var table_rs = ‘’;
var table_rs = statements_insert.execute();
var s = ‘’;
while (table_rs.next())
{
var tbl_nm = table_rs.getColumnValue(2);
var count1 = table_rs.getColumnValue(3);
cmd2_dict = {sqlText: table_rs.getColumnValue(1)};
stmtEx = snowflake.createStatement(cmd2_dict);
try
{

stmtEx.execute();
status = ‘Succeed’ ;
message = ‘\$\$ {‘ + ‘Table_Name: ‘ + tbl_nm + ‘_Row_Count: ‘ + count1 + ‘}\$\$’ ;
snowflake.execute({sqlText: `insert into “${in_db_name}”.”${in_tgt_schema}”.”MIGRATION_AUDIT”
select ‘${tbl_nm}’,’${in_db_name}’,’${in_tgt_schema}’,’${status}’,to_variant(${message}),current_timestamp`});
}
catch(err)
{
status = ‘Failed’;
a = ‘\$\$ {‘ + ‘ failed code : ‘ + err.code + ‘ state : ‘ + err.state + ‘Error Message : ‘ + err.message + ‘ stack trace: ‘ + err.stacktracetxt + ‘}\$\$’ ;
snowflake.execute({sqlText: `insert into “${in_db_name}”.”${in_tgt_schema}”.”MIGRATION_AUDIT”
select ‘${tbl_nm}’,’${in_db_name}’,’${in_tgt_schema}’,’${status}’,to_variant(${a}),current_timestamp`});
}
//s += table_rs.getColumnValue(1) + “\n”;
}
return status;

Using the above code snippet, data migration can be done within the same snowflake account across different environment databases/schemas having the same tables

Audit Table

An audit of data migration status can be obtained from the above table.