[BUG] DLSync breaks when creating schemas or databases

[NemSimpraga]

If DLSync is managing any account-level object that changes the Snowflake session context (like database or schema), it breaks DLSync with the errors:

[main] ERROR com.snowflake.dlsync.doa.ScriptRepo.createScriptObject - Error SQL compilation error:
Table 'DL_SYNC_SCRIPT_HISTORY' does not exist or not authorized., while creating the object...

Similar errors are reported for the other 2 built-in metadata tables DL_SYNC_SCRIPT_EVENT and DL_SYNC_CHANGE_SYNC i.e. Object does not exist or not authorized.

The role operating DLSync does have permissions to use the configured schema, along with table creation in that schema. I can see the tables created in my configured Snowflake database and schema, a single row being created in the DL_SYNC_CHANGE_SYNC table which says CHANGE_TYPE: DEPLOY, STATUS: IN_PROGRESS, LOG: DEPLOY started., which means the configured role can write to the tables.

What I believe the problem is: DLSync does not use fully qualified table names, so when it created a database or schema, Snowflakeby default changes the session context to that new database/schema, so the non fully-qualified metadata table names the code uses now do not exist anymore in the newly created database/schema, hence breaking the tool.

Solution: use fully qualified table names for the DLSync metadata tables given by the connection, since Snowflake might switch the session context when it is interacted with.

[NemSimpraga]

I'd open a PR myself, but I cannot create branches (403) so here is the change to the src/main/java/com/snowflake/dlsync/doa/ScriptRepo.java file I have locally which I believe solves the issue:

package com.snowflake.dlsync.doa;

import com.snowflake.dlsync.ScriptFactory;
import com.snowflake.dlsync.Util;
import com.snowflake.dlsync.dependency.DependencyExtractor;
import com.snowflake.dlsync.models.*;
import com.snowflake.dlsync.parser.SqlTokenizer;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.sql.*;
import java.util.*;
import java.util.stream.Collectors;

@Slf4j
public class ScriptRepo {
    private Properties connectionProperties;
    private Connection connection;
    private Map<String, String> scriptHash = new HashMap<>();
    private Long changeSyncId;

    public final String CHANGE_SYNC_TABLE_NAME = "DL_SYNC_CHANGE_SYNC";
    public final String SCRIPT_HISTORY_TABLE_NAME = "DL_SYNC_SCRIPT_HISTORY";
    public final String SCRIPT_EVENT_TABLE_NAME = "DL_SYNC_SCRIPT_EVENT";
    public final String DEPENDENCY_LINEAGE_TABLE_NAME = "DL_SYNC_DEPENDENCY_LINEAGE";


    public ScriptRepo(Connection connection, Properties connectionProperties) {
        this.connection = connection;
        this.connectionProperties = connectionProperties;
        log.debug("Repo initialized with connection and properties");
    }

    public void init() throws SQLException {
        initScriptTables();
    }

    private void initScriptTables() throws SQLException {
        ////varchar OBJECT_NAME, varchar SCRIPT_HASH, varchar created_by, timestamp created_ts, varchar updated_by, timestamp updated_ts;
        ResultSet resultSet = connection.createStatement().executeQuery("select current_database(), current_schema();");
        resultSet.next();
        log.info("Using database [{}] and schema [{}] for dlsync activities.", resultSet.getString(1), resultSet.getString(2));
        log.debug("Checking for deployment tables");
        try {
            String query = "SELECT * FROM " + getFullyQualifiedChangeSyncTable() + " LIMIT 1;";
            Statement statement = connection.createStatement();
            statement.executeQuery(query);
            updateOldTableNames();
        } catch (SQLException e) {
            log.info("Running for the first time. Creating required tables.");
            String createChangeSyncSql = "CREATE OR REPLACE TABLE " + getFullyQualifiedChangeSyncTable() + " (ID integer PRIMARY KEY, CHANGE_TYPE varchar, STATUS varchar, LOG varchar, CHANGE_COUNT integer, START_TIME timestamp, END_TIME timestamp);";

            String createSqlHash = "CREATE OR REPLACE TABLE " + getFullyQualifiedScriptHistoryTable() + " (SCRIPT_ID VARCHAR, OBJECT_NAME varchar, OBJECT_TYPE varchar, ROLLBACK_SCRIPT varchar, SCRIPT_HASH varchar, DEPLOYED_HASH varchar, CHANGE_SYNC_ID integer, CREATED_BY varchar, CREATED_TS timestamp, UPDATED_BY varchar, UPDATED_TS timestamp, FOREIGN KEY (CHANGE_SYNC_ID) REFERENCES " + getFullyQualifiedChangeSyncTable() + "(ID));";

            String createSqlEvent = "CREATE OR REPLACE TABLE " + getFullyQualifiedScriptEventTable() + " (ID VARCHAR, SCRIPT_ID VARCHAR, OBJECT_NAME varchar, SCRIPT_HASH varchar, STATUS varchar, LOG varchar, CHANGE_SYNC_ID integer, CREATED_BY varchar, CREATED_TS timestamp, FOREIGN KEY (CHANGE_SYNC_ID) REFERENCES " + getFullyQualifiedChangeSyncTable() + "(ID));";
            log.debug("create hash table sql: {}", createSqlHash);
            log.debug("create event table sql: {}", createSqlEvent);
            Statement statement = connection.createStatement();
            statement.executeUpdate(createChangeSyncSql);
            statement.executeUpdate(createSqlHash);
            statement.executeUpdate(createSqlEvent);
        }
    }

    private void updateOldTableNames() {
        try {
            String oldTableName = getFullyQualifiedTableName("DL_SYNC_SCRIPT");
            String query = "SELECT * FROM " + oldTableName + " LIMIT 1;";
            Statement statement = connection.createStatement();
            statement.executeQuery(query);
            log.info("Found old dlsync table DL_SYNC_SCRIPT renaming it to [{}]", SCRIPT_HISTORY_TABLE_NAME);
            String alterSql = "ALTER TABLE IF EXISTS " + oldTableName + " RENAME TO " + getFullyQualifiedScriptHistoryTable() + ";";
            statement.executeUpdate(alterSql);
        } catch (SQLException e) {
            log.debug("All tables are with new version");
        }
    }

    public Set<String> loadScriptHash() throws SQLException {
        String hashQuery =  "SELECT * FROM " + getFullyQualifiedScriptHistoryTable() + ";";
        log.debug("Loading hash with sql: {}", hashQuery);
        Statement statement = connection.createStatement();
        ResultSet resultSet = statement.executeQuery(hashQuery);
        while (resultSet.next()) {
            String id = resultSet.getString("SCRIPT_ID");
            scriptHash.put(id, resultSet.getString("SCRIPT_HASH"));
        }
        log.debug("Script hash loaded: {}", scriptHash);
        return  scriptHash.keySet();

    }

    public Set<String> loadDeployedHash() throws SQLException {
        String hashColumn = "DEPLOYED_HASH";
        String hashQuery =  "SELECT * FROM " + getFullyQualifiedScriptHistoryTable() + ";";
        log.debug("Loading hash with sql: {}", hashQuery);
        Statement statement = connection.createStatement();
        ResultSet resultSet = statement.executeQuery(hashQuery);
        while (resultSet.next()) {
            String id = resultSet.getString("SCRIPT_ID");
            scriptHash.put(id, resultSet.getString(hashColumn));
        }
        log.debug("Script deployed hash loaded: {}", scriptHash);
        return  scriptHash.keySet();

    }

    public Long insertChangeSync(ChangeType changeType, Status status, String logMessage) throws SQLException {
        String queryGetId = "SELECT count(1) FROM " + getFullyQualifiedChangeSyncTable() + ";";
        ResultSet rs = connection.createStatement().executeQuery(queryGetId);
        if(rs.next()) {
            changeSyncId =  rs.getLong(1) + 1;
        }

        String insertSql = "INSERT INTO " + getFullyQualifiedChangeSyncTable() + " (ID, CHANGE_TYPE, STATUS, LOG, START_TIME) VALUES(?, ?, ?, ?, CURRENT_TIMESTAMP);";
        PreparedStatement statement = connection.prepareStatement(insertSql);
        statement.setLong(1, changeSyncId);
        statement.setString(2, changeType.toString());
        statement.setString(3, status.toString());
        statement.setString(4, logMessage);
        log.debug("Creating script event with the following SQL: {}", insertSql);
        statement.executeUpdate();

        return changeSyncId;
    }

    public void updateChangeSync(ChangeType changeType, Status status, String logMessage, Long changeCount) throws SQLException {
        String updateSql = "UPDATE " + getFullyQualifiedChangeSyncTable() + " SET CHANGE_TYPE=?, STATUS=?, LOG=?, CHANGE_COUNT=?, END_TIME=CURRENT_TIMESTAMP WHERE ID = ? ;";
        PreparedStatement statement = connection.prepareStatement(updateSql);
        statement.setString(1, changeType.toString());
        statement.setString(2, status.toString());
        statement.setString(3, logMessage);
        statement.setObject(4, changeCount);
        statement.setLong(5, changeSyncId);
        log.debug("Creating script event with the following SQL: {}", updateSql);
        statement.executeUpdate();
    }
    private boolean updateScriptHash(Script script) throws SQLException {
        String rollback = null;
        if(script instanceof MigrationScript) {
            MigrationScript migrationScript = (MigrationScript)script;
            rollback = migrationScript.getRollback();
        }
        PreparedStatement statement;
        String deployedHash = Util.getMd5Hash(script.getContent());
        log.debug("Updating script hash of object {}", script.getId());
        if(scriptHash.containsKey(script.getId())) {
            String updateSql = "UPDATE " + getFullyQualifiedScriptHistoryTable() + " SET ROLLBACK_SCRIPT=?, SCRIPT_HASH=?, DEPLOYED_HASH=?, CHANGE_SYNC_ID=?, updated_by=current_user, updated_ts=current_timestamp WHERE SCRIPT_ID=?;";
            statement = connection.prepareStatement(updateSql);
            statement.setString(1, rollback);
            statement.setString(2, script.getHash());
            statement.setString(3, deployedHash);
            statement.setLong(4, changeSyncId);
            statement.setString(5, script.getId());
            log.debug("Updating script hash with the following SQL: {}", updateSql);
        }
        else {
            String insertSql = "INSERT INTO " + getFullyQualifiedScriptHistoryTable() + " VALUES(?, ?, ?, ?, ?, ?, ?, current_user, current_timestamp, current_user, current_timestamp);";
            statement = connection.prepareStatement(insertSql);
            statement.setString(1, script.getId());
            statement.setString(2, script.getFullObjectName());
            statement.setString(3, script.getObjectType().toString());
            statement.setString(4, rollback);
            statement.setString(5, script.getHash());
            statement.setString(6, deployedHash);
            statement.setLong(7, changeSyncId);
            log.debug("Updating script hash with the following SQL: {}", insertSql);
        }

        return statement.executeUpdate() >= 0;
    }

    private boolean insertScriptEvent(Script script, String status, String logs) throws SQLException {
        //varchar ID, varchar OBJECT_NAME, varchar SCRIPT_HASH, varchar STATUS, varchar log, varchar created_by, varchar created_ts;
        log.debug("Creating event for the object {} with status: {} and log: {} ", script.getObjectName(), status, logs);
        String insertSql = "INSERT INTO " + getFullyQualifiedScriptEventTable() + " SELECT UUID_STRING(), ?, ?, ?, ?, ?, ?, current_user, current_timestamp;";
        PreparedStatement statement = connection.prepareStatement(insertSql);
        statement.setString(1, script.getId());
        statement.setObject(2, script.getFullObjectName());
        statement.setString(3, script.getHash());
        statement.setString(4, status);
        statement.setString(5, logs);
        statement.setLong(6, changeSyncId);
        log.debug("Creating script event with the following SQL: {}", insertSql);
        return statement.executeUpdate() > 0;
    }

    public boolean isScriptChanged(Script script) {
//        return true;
        return !scriptHash.getOrDefault(script.getId(), "null").equals(script.getHash());
    }

    public boolean isScriptVersionDeployed(Script script) {
        return scriptHash.containsKey(script.getId());
    }

    public void createScriptObject(Script script, boolean onlyHashes) throws SQLException {

        Statement statement = connection.createStatement();
        boolean autoCommit = connection.getAutoCommit();
        try {
            connection.setAutoCommit(false);
            if(!onlyHashes) {
                statement.execute(script.getContent());
                log.debug("Creating object using the SQL: {}", script.getContent());
            }
            updateScriptHash(script);
            insertScriptEvent(script, "SUCCESS", "Successfully Deployed Object");
            connection.commit();
            log.info("Successfully Deployed object: {}", script);
        }
        catch (SQLException e) {
            connection.rollback();
            log.error("Error {}, while creating the object {} with sql {}", e.getMessage(), script.getObjectName(), script.getContent());
            insertScriptEvent(script, "ERROR", e.getMessage());
            throw e;
        }
        finally {
            connection.setAutoCommit(autoCommit);
        }

    }

    public List<SchemaScript> getAllScriptsInSchema(String schema) throws SQLException {
        log.info("Getting all scripts in schema: {}", schema);
        String sql = String.format("SELECT GET_DDL('SCHEMA', '%s', true)", schema);
        log.debug("Getting all scripts using SQL: {}", sql);
        ResultSet resultSet = connection.createStatement().executeQuery(sql);
        if(resultSet.next()) {
            String ddl = resultSet.getString(1);
            return SqlTokenizer.parseDdlScripts(ddl, getDatabaseName(), schema);
        }
        else {
            throw new RuntimeException("Error while getting ddl scripts: result set has no data");
        }
    }


    public SchemaScript addConfig(SchemaScript script) throws SQLException {
        if(script.getObjectType() == ScriptObjectType.TABLES) {
            String additionalContent = String.format("SELECT * FROM %s", script.getFullObjectName());
            ResultSet resultSet = connection.createStatement().executeQuery(additionalContent);
            int count = resultSet.getMetaData().getColumnCount();
            StringBuilder insertBuilder = new StringBuilder(String.format("INSERT INTO %s values", script.getFullObjectName()));
            Boolean firstRow = true;
            while(resultSet.next()) {
                if(!firstRow) {
                    insertBuilder.append(", ");
                }
                firstRow = false;
                for(int i = 1; i <= count; i++) {
                    if(i == 1) {
                        insertBuilder.append("(");
                    }
                    else {
                        insertBuilder.append(", ");
                    }
                    Object value = resultSet.getObject(i);
                    if(value == null) {
                        insertBuilder.append("null");
                    }
                    else {
                        insertBuilder.append(String.format("'%s'", value.toString().replace("'", "''")));
                    }
                }
                insertBuilder.append(")");

            }
            insertBuilder.append(";");
            String migrationHeader = "---version: 1, author: DlSync\n";
            String rollback = "\n---rollback: DELETE FROM " + script.getFullObjectName() + ";";
            String verify = "\n---verify: SELECT COUNT(*) FROM  " + script.getFullObjectName() + ";";
            String insertContent = migrationHeader + insertBuilder + rollback + verify;

            String newContent = script.getContent() + "\n\n" + insertContent;
            script.setContent(newContent);
        }
        return script;
    }

    public List<String> getAllSchemasInDatabase(String database) throws SQLException {
        log.info("Reading all schemas in database: {}", database);
        String query = String.format("SELECT * FROM %s.INFORMATION_SCHEMA.SCHEMATA", database);
        log.debug("Reading schemas using sql: {}", query);
        ResultSet resultSet = executeQuery(query);
        List<String> schemas = new ArrayList<>();
        while(resultSet.next()) {
            String schema = resultSet.getString("SCHEMA_NAME");
            if(schema.equalsIgnoreCase("INFORMATION_SCHEMA") || schema.equalsIgnoreCase("PUBLIC")) {
                continue;
            }
            schemas.add(schema);
        }
        return schemas;
    }

    public String getDatabaseName() {
        return connectionProperties.getProperty("db");
    }

    public String getSchemaName() {
        return connectionProperties.getProperty("schema");
    }

    private String getFullyQualifiedTableName(String tableName) {
        return String.format("%s.%s.%s", getDatabaseName(), getSchemaName(), tableName);
    }

    private String getFullyQualifiedChangeSyncTable() {
        return getFullyQualifiedTableName(CHANGE_SYNC_TABLE_NAME);
    }

    private String getFullyQualifiedScriptHistoryTable() {
        return getFullyQualifiedTableName(SCRIPT_HISTORY_TABLE_NAME);
    }

    private String getFullyQualifiedScriptEventTable() {
        return getFullyQualifiedTableName(SCRIPT_EVENT_TABLE_NAME);
    }

    private String getFullyQualifiedDependencyLineageTable() {
        return getFullyQualifiedTableName(DEPENDENCY_LINEAGE_TABLE_NAME);
    }

    public ResultSet executeQuery(String query) throws SQLException {
        return connection.createStatement().executeQuery(query);
    }

    public void insertDependencyList(List<ScriptDependency> dependencyList) throws SQLException {
        String createTable = "CREATE TABLE IF NOT EXISTS " + getFullyQualifiedDependencyLineageTable() + "(OBJECT_NAME VARCHAR, OBJECT_TYPE VARCHAR, DEPENDENCY VARCHAR, DEPENDECY_OBEJECT_TYPE VARCHAR, CHANGE_SYNC_ID VARCHAR, CREATED_BY VARCHAR, CREATED_TS TIMESTAMP);";
        connection.createStatement().executeUpdate(createTable);
        StringBuilder insertSql = new StringBuilder("INSERT INTO " + getFullyQualifiedDependencyLineageTable() + " VALUES ");

        for(ScriptDependency dependency: dependencyList) {
            String values = String.format("('%s', '%s', '%s', '%s', '%s', current_user, current_timestamp),", dependency.getObjectName(), dependency.getObjectType(), dependency.getDependency(), dependency.getDependencyObjectType(), changeSyncId);
            insertSql.append(values);
        }
        insertSql.deleteCharAt(insertSql.length() - 1);
        log.info("inserting dag scripts using {} ", insertSql);
        connection.createStatement().executeUpdate(insertSql.toString());
    }

    public void insertSortedScript(List<Script> sequencedScript) throws SQLException {
        DependencyExtractor dependencyExtractor = new DependencyExtractor();
        String topologicalSortedTable = getFullyQualifiedTableName("DL_SYNC_TOPOLOGICAL_SORTED");
        String createTable = "CREATE OR REPLACE TABLE " + topologicalSortedTable + "(id INT, script VARCHAR, dependency_size INT, dependencies VARCHAR);";
        connection.createStatement().executeUpdate(createTable);
        StringBuilder insertSql = new StringBuilder("INSERT INTO " + topologicalSortedTable + " VALUES ");

        for(int i = 0; i < sequencedScript.size(); i++) {
            Script script = sequencedScript.get(i);
            Set<String> dependencies = dependencyExtractor.extractScriptDependencies(script).stream().map(s -> s.getFullObjectName()).collect(Collectors.toSet());
            if(dependencies.size() == 0) {
                log.debug("Found zero dependency for {}", script.getFullObjectName());
            }
            String values = String.format("(%s, '%s', %s, '%s'),", i, script.getFullObjectName(), dependencies.size(), dependencies.toString());
            insertSql.append(values);
        }
        insertSql.deleteCharAt(insertSql.length() - 1);
        log.info("inserting sorted scripts using {} ", insertSql.toString());
        connection.createStatement().executeUpdate(insertSql.toString());
    }

    public List<MigrationScript> getDeployedMigrationScripts(Set<String> ids) throws SQLException {
        if(ids.size() == 0) {
            return new ArrayList<>();
        }
        String allIdJoined = ids.stream().map(v -> "'" + v + "'").collect(Collectors.joining(",", "(", ");"));
        String query = "SELECT * FROM " + getFullyQualifiedScriptHistoryTable() + " where SCRIPT_ID in " + allIdJoined;
        PreparedStatement statement = connection.prepareStatement(query);
        ResultSet rs = statement.executeQuery();
        List<MigrationScript> migrations = new ArrayList<>();
        while(rs.next()) {
            ScriptObjectType objectType = ScriptObjectType.valueOf(rs.getString("OBJECT_TYPE"));
            if(!objectType.isMigration()) {
                continue;
            }
            String id = rs.getString("SCRIPT_ID");
            String[] idSplit = id.split(":");
            Long version = null;
            if(idSplit.length > 1) {
                version = Long.parseLong(idSplit[1]);
            }
            String fullObjectName = rs.getString("OBJECT_NAME");


            String rollback = rs.getString("ROLLBACK_SCRIPT");

            MigrationScript migrationScript = ScriptFactory.getMigrationScript(fullObjectName, objectType, "", version, null, rollback, null);
            migrations.add(migrationScript);
        }
        return migrations;
    }

    public void executeRollback(MigrationScript migrationScript) throws SQLException {
        Statement statement = connection.createStatement();
        boolean autoCommit = connection.getAutoCommit();
        try {
            connection.setAutoCommit(false);
            if(migrationScript.getRollback() != null && !migrationScript.getRollback().trim().equals("")) {
                log.debug("Executing rollback using the SQL: {}", migrationScript.getRollback());
                statement.executeUpdate(migrationScript.getRollback());
            }

            deleteScriptHash(migrationScript);
            insertScriptEvent(migrationScript, "SUCCESS", "Successfully Rolled-back Object");
            connection.commit();
            log.info("Successfully Rollback object: {}", migrationScript);
        }
        catch (SQLException e) {
            connection.rollback();
            log.error("Error {}, while rollback the object {} with sql {}", e.getMessage(), migrationScript.getObjectName(), migrationScript.getRollback());
            insertScriptEvent(migrationScript, "ERROR", e.getMessage());
            throw e;
        }
        finally {
            connection.setAutoCommit(autoCommit);
        }
    }

    public boolean executeVerify(MigrationScript migrationScript) throws SQLException {
        Statement statement = connection.createStatement();
        boolean autoCommit = connection.getAutoCommit();
        try {
            connection.setAutoCommit(false);
            if(migrationScript.getVerify() != null && !migrationScript.getVerify().trim().equals("")) {
                log.debug("Executing verify using the SQL: {}", migrationScript.getVerify());
                statement.executeQuery(migrationScript.getVerify());
            }
            insertScriptEvent(migrationScript, "SUCCESS", "Successfully Verified Object");
            connection.commit();
            log.debug("Successfully Verified object: {}", migrationScript);
            return true;
        }
        catch (SQLException e) {
            connection.rollback();
            log.error("Error {}, while verifying the object {} with sql {}", e.getMessage(), migrationScript.getObjectName(), migrationScript.getVerify());
            insertScriptEvent(migrationScript, "ERROR", e.getMessage());
            return false;
        }
        finally {
            connection.setAutoCommit(autoCommit);
        }
    }

    private void deleteScriptHash(MigrationScript migration) throws SQLException {
        String deleteSql = "DELETE FROM " + getFullyQualifiedScriptHistoryTable() + " WHERE SCRIPT_ID=?;";
        PreparedStatement statement = connection.prepareStatement(deleteSql);
        statement.setString(1, migration.getId());
        statement.executeUpdate();
    }

    public boolean verifyScript(Script script) {
        if(!scriptHash.containsKey(script.getId())) {
            log.warn("Script file does not exist for the db object: {}", script);
        }
        return scriptHash.getOrDefault(script.getId(), script.getHash()).equals(script.getHash());
    }

    public boolean compareScript(Script script1, Script script2) {
        return SqlTokenizer.compareScripts(script1, script2);
    }

    public List<TestResult> runTest(TestScript testScript) throws IOException {
        List<TestResult> testResults = new ArrayList<>();
        try {
            log.debug("Running test script: {}", testScript.getObjectName());
            ResultSet resultSet = connection.createStatement().executeQuery(testScript.getTestQuery());
            while(resultSet.next()) {
                TestResult testResult = new TestResult(resultSet.getString(1), resultSet.getString(2));
                testResults.add(testResult);
            }
            return testResults;
        } catch (SQLException e) {
            log.error("Error while running test script: {}", e.getMessage());
            testResults.add(new TestResult(e));
            return testResults;
        }
    }
}

[sfc-gh-yhailu]

Hi @NemSimpraga that is interesting observation.
We'd love your contribution! To contribute to the project you can:

1. Fork the repository to your own GitHub account
2. Create a feature branch from your fork
3. Push your changes to your fork repo
4. create a Pull Request from your fork to the main repository
5. Share your changes in a comment here or link the PR

We'll review it and work with you to get it merged! Thanks for helping improve DLSync.

[NemSimpraga]

Just opened one! :) #57

[sfc-gh-yhailu]

we have released v3.0.1 to fix it. Please verify and close the issue if it is resolved. Thank you.

[NemSimpraga]

Can confirm it is fixed now as of v3.0.1! Thanks!