diff --git a/pom.xml b/pom.xml
index 73c2973c..27168566 100644
--- a/pom.xml
+++ b/pom.xml
@@ -4,7 +4,7 @@
 
     <groupId>org.liquibase.ext</groupId>
     <artifactId>liquibase-databricks</artifactId>
-    <version>1.2.0-SNAPSHOT</version>
+    <version>1.1.5-SNAPSHOT</version>
 
     <name>Liquibase Extension: Databricks support</name>
     <description>Liquibase Extension for Databricks.</description>
@@ -119,6 +119,12 @@
             <version>${dependency.spock.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.databricks</groupId>
+            <artifactId>databricks-jdbc</artifactId>
+            <version>2.6.38</version>
+            <scope>compile</scope>
+        </dependency>
     </dependencies>
 
     <build>
diff --git a/src/main/java/liquibase/ext/databricks/database/DatabricksConnection.java b/src/main/java/liquibase/ext/databricks/database/DatabricksConnection.java
index c3686d3c..df6cba36 100644
--- a/src/main/java/liquibase/ext/databricks/database/DatabricksConnection.java
+++ b/src/main/java/liquibase/ext/databricks/database/DatabricksConnection.java
@@ -1,17 +1,534 @@
 package liquibase.ext.databricks.database;
 
+import com.databricks.client.jdbc.jdbc42.S42Connection;
+import com.databricks.client.spark.core.SparkJDBCConnection;
+import liquibase.Scope;
 import liquibase.database.jvm.JdbcConnection;
 import liquibase.exception.DatabaseException;
-import java.sql.Connection;
+import liquibase.exception.UnexpectedLiquibaseException;
+
+import java.sql.*;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
 
 public class DatabricksConnection extends JdbcConnection {
 
-    public DatabricksConnection(Connection connection) {
-        super(connection);
+    private S42Connection con;
+    public DatabricksConnection() {}
+
+    public DatabricksConnection(Connection conn) {
+        this.con = (S42Connection) conn;
+    }
+
+    @Override
+    public String getDatabaseProductName() throws DatabaseException {
+        try {
+            return this.getWrappedConnection().getMetaData().getDatabaseProductName();
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public Connection getWrappedConnection() {
+        return con;
+    }
+
+    public SparkJDBCConnection getUnderlyingSparkConnection() {
+        if (con.getConnection() instanceof SparkJDBCConnection) {
+            return (SparkJDBCConnection) con.getConnection();
+        }
+        return null;
+    }
+
+    @Override
+    public Connection getUnderlyingConnection() {
+        return con;
+    }
+
+    @Override
+    public void open(String url, Driver driverObject, Properties driverProperties) throws DatabaseException {
+
+        driverProperties.setProperty("UserAgentEntry", "Liquibase");
+        driverProperties.setProperty("EnableArrow", "0");
+        // Set UserAgent to specify to Databricks that liquibase is the tool running these commands
+        // Set EnableArrow because the arrow results break everything. And the JDBC release notes say to just disable it.
+
+        // Ensure there's a terminating semicolon for consistent parsing
+        if (!url.endsWith(";")) {
+            url += ";";
+        }
+
+        String updatedUrl = url + "UserAgentEntry=Liquibase;EnableArrow=0";
+
+        this.openConn(updatedUrl, driverObject, driverProperties);
+    }
+
+    public void openConn(String url, Driver driverObject, Properties driverProperties) throws DatabaseException {
+        try {
+            Scope.getCurrentScope().getLog(this.getClass()).info("opening connection " + url);
+            this.con = (S42Connection) driverObject.connect(url, driverProperties);
+            if (this.con == null) {
+                Scope.getCurrentScope().getLog(this.getClass()).severe("Connection could not be created");
+                throw new DatabaseException("Connection could not be created to " + url + " with driver " + driverObject.getClass().getName() + ".  Possibly the wrong driver for the given database URL");
+            }
+        } catch (SQLException sqle) {
+            throw new DatabaseException("Connection could not be created to " + url + " with driver " + driverObject.getClass().getName() + ".  " + sqle.getMessage());
+        }
+    }
+
+    @Override
+    public boolean supports(String url) {
+        return url.toLowerCase().contains("databricks");
     }
+
+    @Override
+    public int getPriority() {
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
+    }
+
+    @Override
+    public boolean getAutoCommit() throws DatabaseException {
+        return true;
+    }
+
     @Override
     public void setAutoCommit(boolean autoCommit) throws DatabaseException {
-        // no-op for Databricks since there is not a concept of committing
+    }
+
+    protected static String getUrlParamValue(String url, String paramName, String defaultValue) {
+
+        //System.out.println("PARSE URL - url" + url);
+
+        if (url == null) {
+            return null;
+        }
+        // Ensure there's a terminating semicolon for consistent parsing
+        if (!url.endsWith(";")) {
+            url += ";";
+        }
+        // Remove spaces and split by semicolon
+        String[] uriArgs = url.replace(" ", "").split(";");
+
+       // System.out.println("PARSE URL - url args" + uriArgs.toString());
+
+        // Use Java Streams to find the parameter value
+        Optional<String> paramString = Arrays.stream(uriArgs)
+                .filter(x -> x.startsWith(paramName + "="))
+                .findFirst();
+        // Return the parameter value if found, otherwise return the default value
+        if (!paramString.isPresent()) {
+            return defaultValue;
+        }
+        String[] defaultParamsArr = paramString.get().split("=");
+        return defaultParamsArr.length > 1 ? defaultParamsArr[1] : defaultValue; // Check to avoid index out of bound
+    }
+
+
+    @Override
+    public String getDatabaseProductVersion() throws DatabaseException {
+        try {
+            return con.getMetaData().getDatabaseProductVersion();
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public int getDatabaseMajorVersion() throws DatabaseException {
+        try {
+            return con.getMetaData().getDatabaseMajorVersion();
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public int getDatabaseMinorVersion() throws DatabaseException {
+        try {
+            return con.getMetaData().getDatabaseMinorVersion();
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    /////////////////////////////////////////////////// copy from parent ///////////////////////////////////////////////////
+    @Override
+    protected String getConnectionUrl() throws SQLException {
+
+        String rawUrl = con.getMetaData().getURL();
+        // Check for ; characters
+        String updatedUrl;
+
+        if (rawUrl.charAt(rawUrl.length() - 1) == ';') {
+            updatedUrl = rawUrl + "UserAgentEntry=Liquibase;EnableArrow=0;";
+        }
+        else {
+            updatedUrl = rawUrl + ";UserAgentEntry=Liquibase;EnableArrow=0;";
+
+        }
+        return updatedUrl;
+    }
+
+    @Override
+    public String getConnectionUserName() {
+        try {
+            return con.getMetaData().getUserName();
+        } catch (SQLException e) {
+            throw new UnexpectedLiquibaseException(e);
+        }
+    }
+
+    @Override
+    public void clearWarnings() throws DatabaseException {
+        try {
+            con.clearWarnings();
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public void close() throws DatabaseException {
+        rollback();
+        try {
+            con.close();
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public void commit() throws DatabaseException {
+        try {
+            if (!con.getAutoCommit()) {
+                con.commit();
+            }
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public Statement createStatement() throws DatabaseException {
+        try {
+            return con.createStatement();
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public Statement createStatement(int resultSetType,
+                                     int resultSetConcurrency, int resultSetHoldability)
+            throws DatabaseException {
+        try {
+            return con.createStatement(resultSetType, resultSetConcurrency,
+                    resultSetHoldability);
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public Statement createStatement(int resultSetType, int resultSetConcurrency)
+            throws DatabaseException {
+        try {
+            return con.createStatement(resultSetType, resultSetConcurrency);
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public String getCatalog() throws DatabaseException {
+        try {
+            return con.getCatalog();
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public void setCatalog(String catalog) throws DatabaseException {
+        try {
+            con.setCatalog(catalog);
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public int getHoldability() throws DatabaseException {
+        try {
+            return con.getHoldability();
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public void setHoldability(int holdability) throws DatabaseException {
+        try {
+            con.setHoldability(holdability);
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public DatabaseMetaData getMetaData() throws DatabaseException {
+        try {
+            return con.getMetaData();
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public int getTransactionIsolation() throws DatabaseException {
+        try {
+            return con.getTransactionIsolation();
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public void setTransactionIsolation(int level) throws DatabaseException {
+        try {
+            con.setTransactionIsolation(level);
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public Map<String, Class<?>> getTypeMap() throws DatabaseException {
+        try {
+            return con.getTypeMap();
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public void setTypeMap(Map<String, Class<?>> map) throws DatabaseException {
+        try {
+            con.setTypeMap(map);
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public SQLWarning getWarnings() throws DatabaseException {
+        try {
+            return con.getWarnings();
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public boolean isClosed() throws DatabaseException {
+        return con.isClosed();
+    }
+
+    @Override
+    public boolean isReadOnly() throws DatabaseException {
+        try {
+            return con.isReadOnly();
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public void setReadOnly(boolean readOnly) throws DatabaseException {
+        try {
+            con.setReadOnly(readOnly);
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
 
+    @Override
+    public String nativeSQL(String sql) throws DatabaseException {
+        try {
+            return con.nativeSQL(sql);
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public CallableStatement prepareCall(String sql, int resultSetType,
+                                         int resultSetConcurrency, int resultSetHoldability)
+            throws DatabaseException {
+        try {
+            return con.prepareCall(sql, resultSetType, resultSetConcurrency,
+                    resultSetHoldability);
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public CallableStatement prepareCall(String sql, int resultSetType,
+                                         int resultSetConcurrency) throws DatabaseException {
+        try {
+            return con.prepareCall(sql, resultSetType, resultSetConcurrency);
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public CallableStatement prepareCall(String sql) throws DatabaseException {
+        try {
+            return con.prepareCall(sql);
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
     }
-}
\ No newline at end of file
+
+    @Override
+    public PreparedStatement prepareStatement(String sql, int resultSetType,
+                                              int resultSetConcurrency, int resultSetHoldability)
+            throws DatabaseException {
+        try {
+            return con.prepareStatement(sql, resultSetType, resultSetConcurrency,
+                    resultSetHoldability);
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public PreparedStatement prepareStatement(String sql, int resultSetType,
+                                              int resultSetConcurrency) throws DatabaseException {
+        try {
+            return con.prepareStatement(sql, resultSetType, resultSetConcurrency);
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys)
+            throws DatabaseException {
+        try {
+            return con.prepareStatement(sql, autoGeneratedKeys);
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public PreparedStatement prepareStatement(String sql, int[] columnIndexes)
+            throws DatabaseException {
+        try {
+            return con.prepareStatement(sql, columnIndexes);
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public PreparedStatement prepareStatement(String sql, String[] columnNames)
+            throws DatabaseException {
+        try {
+            return con.prepareStatement(sql, columnNames);
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public PreparedStatement prepareStatement(String sql) throws DatabaseException {
+        try {
+            return con.prepareStatement(sql);
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public void releaseSavepoint(Savepoint savepoint) throws DatabaseException {
+        try {
+            con.releaseSavepoint(savepoint);
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public void rollback() throws DatabaseException {
+        try {
+            if (!con.getAutoCommit() && !con.isClosed()) {
+                con.rollback();
+            }
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public void rollback(Savepoint savepoint) throws DatabaseException {
+        try {
+            if (!con.getAutoCommit()) {
+                con.rollback(savepoint);
+            }
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public Savepoint setSavepoint() throws DatabaseException {
+        try {
+            return con.setSavepoint();
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public Savepoint setSavepoint(String name) throws DatabaseException {
+        try {
+            return con.setSavepoint(name);
+        } catch (SQLException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (!(obj instanceof JdbcConnection)) {
+            return false;
+        }
+        Connection underlyingConnection = this.getUnderlyingConnection();
+        if (underlyingConnection == null) {
+            return ((JdbcConnection) obj).getUnderlyingConnection() == null;
+        }
+
+        return underlyingConnection.equals(((JdbcConnection) obj).getUnderlyingConnection());
+    }
+
+    @Override
+    public int hashCode() {
+        Connection underlyingConnection = this.getUnderlyingConnection();
+        try {
+            if ((underlyingConnection == null) || underlyingConnection.isClosed()) {
+                return super.hashCode();
+            }
+        } catch (SQLException e) {
+            return super.hashCode();
+        }
+        return underlyingConnection.hashCode();
+    }
+
+}
diff --git a/src/main/java/liquibase/ext/databricks/database/DatabricksDatabase.java b/src/main/java/liquibase/ext/databricks/database/DatabricksDatabase.java
index 5bf5f1e1..18c3a469 100644
--- a/src/main/java/liquibase/ext/databricks/database/DatabricksDatabase.java
+++ b/src/main/java/liquibase/ext/databricks/database/DatabricksDatabase.java
@@ -5,27 +5,26 @@
 import liquibase.database.DatabaseConnection;
 import liquibase.database.jvm.JdbcConnection;
 import liquibase.exception.DatabaseException;
-import liquibase.structure.DatabaseObject;
 import liquibase.statement.SqlStatement;
 import liquibase.statement.core.RawCallStatement;
+import liquibase.structure.DatabaseObject;
+import liquibase.structure.core.Catalog;
 import liquibase.structure.core.Schema;
-import liquibase.util.StringUtil;
+
 import java.math.BigInteger;
 import java.sql.ResultSet;
-import java.util.Arrays;
-import java.util.HashSet;
-import java.util.Set;
-import java.util.List;
-import java.util.Collections;
+import java.util.*;
 
 
 public class DatabricksDatabase extends AbstractJdbcDatabase {
 
+
+    public static final int DATABRICKS_PRIORITY_DATABASE = 1515;
     // define env variables for database
     public static final String PRODUCT_NAME = "databricks";
     // Set default catalog - must be unity Catalog Enabled
 
-    public String systemSchema = "information_schema";
+    private String systemSchema = "information_schema";
 
     // This is from the new INFORMATION_SCHEMA() database
     private Set<String> systemTablesAndViews = new HashSet<>();
@@ -55,9 +54,19 @@ protected String getQuotingEndReplacement() {
         return "``";
     }
 
+    @Override
+    public String getDatabaseChangeLogTableName() {
+        return super.getDatabaseChangeLogTableName().toLowerCase(Locale.US);
+    }
+
+    @Override
+    public String getDatabaseChangeLogLockTableName() {
+        return super.getDatabaseChangeLogLockTableName().toLowerCase(Locale.US);
+    }
+
     @Override
     public String getShortName() {
-        return "databricks";
+        return PRODUCT_NAME;
     }
 
     @Override
@@ -83,12 +92,12 @@ public Integer getDefaultPort() {
 
     @Override
     public int getPriority() {
-        return PRIORITY_DATABASE;
+        return DATABRICKS_PRIORITY_DATABASE;
     }
 
     @Override
     public boolean isCorrectDatabaseImplementation(DatabaseConnection conn) throws DatabaseException {
-        return PRODUCT_NAME.equalsIgnoreCase(conn.getDatabaseProductName()) || conn.getDatabaseProductName().equalsIgnoreCase("SparkSQL");
+        return PRODUCT_NAME.equalsIgnoreCase(conn.getDatabaseProductName()) || conn.getDatabaseProductName().equalsIgnoreCase("SparkSQL") || conn.getDatabaseProductName().equalsIgnoreCase("spark");
     }
 
     @Override
@@ -158,6 +167,7 @@ public String getAutoIncrementClause(final BigInteger startWith, final BigIntege
         return autoIncrementClause;
     }
 
+    @Override
     protected String getAutoIncrementClause() {
         return "GENERATED BY DEFAULT AS IDENTITY";
     }
@@ -194,39 +204,73 @@ protected String getConnectionSchemaName() {
         if (connection == null) {
             return null;
         }
-        try (ResultSet resultSet = ((JdbcConnection) connection).createStatement().executeQuery("SELECT CURRENT_SCHEMA()")) {
+
+        try (ResultSet resultSet = ((DatabricksConnection) connection).createStatement().executeQuery("SELECT CURRENT_SCHEMA()")) {
             resultSet.next();
             return resultSet.getString(1);
+
         } catch (Exception e) {
-            Scope.getCurrentScope().getLog(getClass()).info("Error getting default schema", e);
+            Scope.getCurrentScope().getLog(getClass()).info("Error getting default schema via existing context, going to pull from URL", e);
         }
 
-        String foundSchema = parseUrlForSchema(connection.getURL());
-        System.out.println("SCHEMA IDENFIED: "+ foundSchema);
+        try {
+
+            String foundSchema = parseUrlForSchema(connection.getURL());
+            Scope.getCurrentScope().getLog(getClass()).info("SCHEMA IDENTIFIED: " + foundSchema);
+
+            return foundSchema;
+        } catch (Exception e) {
+            Scope.getCurrentScope().getLog(getClass()).warning("Cannot get default / defined schema from URL or current session.");
+        }
+        // Return null, not default to force user to supply the schema
+        return null;
 
-        return foundSchema;
     }
 
-    private String parseUrlForSchema(String url) {
+    @Override
+    protected String getConnectionCatalogName() {
+        DatabaseConnection connection = getConnection();
 
-        String schemaToken = "ConnSchema=";
+        if (connection == null) {
+            return null;
+        }
 
-        int startIndex = url.indexOf(schemaToken);
+        try{
+            return connection.getCatalog();
+        } catch (Exception e) {
+            Scope.getCurrentScope().getLog(getClass()).warning("Cannot get default / defined CATALOG from current session.");
+        }
 
-        // If ConnSchema not found, find the default value
-        if (startIndex == -1) {
+        try (ResultSet resultSet = ((DatabricksConnection) connection).createStatement().executeQuery("SELECT CURRENT_CATALOG()")) {
+            resultSet.next();
+            return resultSet.getString(1);
 
-            return "default";
+        } catch (Exception e) {
+            Scope.getCurrentScope().getLog(getClass()).info("Error getting default catalog via existing context, going to pull from URL", e);
         }
 
-        startIndex += schemaToken.length();
-        int endIndex = url.indexOf(";", startIndex);
+        try {
+            String foundCatalog = parseUrlForCatalog(connection.getURL());
+            Scope.getCurrentScope().getLog(getClass()).info("CATALOG IDENTIFIED: " + foundCatalog);
 
-        if (endIndex == -1) {
-            return url.substring(startIndex);
+            return foundCatalog;
+
+        } catch (Exception e) {
+            Scope.getCurrentScope().getLog(getClass()).warning("Cannot get default / defined CATALOG from URL");
         }
+        // Return null, not default to force user to supply the catalog
+        return null;
+
+    }
+
+    private String parseUrlForSchema(String url) {
+        String schemaToken = "ConnSchema";
+        return DatabricksConnection.getUrlParamValue(url, schemaToken, this.defaultSchemaName);
+    }
 
-        return url.substring(startIndex, endIndex);
+    private String parseUrlForCatalog(String url) {
+        String schemaToken = "ConnCatalog";
+        return  DatabricksConnection.getUrlParamValue(url, schemaToken, this.defaultCatalogName);
     }
 
     @Override
@@ -234,6 +278,11 @@ public void setDefaultSchemaName(final String schemaName) {
         this.defaultSchemaName = correctObjectName(schemaName, Schema.class);
     }
 
+    @Override
+    public void setDefaultCatalogName(final String catalogName) {
+        this.defaultCatalogName = correctObjectName(catalogName, Catalog.class);
+    }
+
     public void setSystemSchema(String systemSchema) {this.systemSchema = systemSchema;}
 
 
@@ -307,4 +356,5 @@ public void setConnection(DatabaseConnection conn) {
         }
         super.setConnection(dbConn);
     }
-}
\ No newline at end of file
+
+}
diff --git a/src/main/java/liquibase/ext/databricks/datatype/ArrayIntegerDataTypeDatabricks.java b/src/main/java/liquibase/ext/databricks/datatype/ArrayIntegerDataTypeDatabricks.java
new file mode 100644
index 00000000..90791681
--- /dev/null
+++ b/src/main/java/liquibase/ext/databricks/datatype/ArrayIntegerDataTypeDatabricks.java
@@ -0,0 +1,42 @@
+package liquibase.ext.databricks.datatype;
+
+import liquibase.change.core.LoadDataChange;
+import liquibase.database.Database;
+import liquibase.datatype.DataTypeInfo;
+import liquibase.datatype.DatabaseDataType;
+import liquibase.datatype.LiquibaseDataType;
+import liquibase.ext.databricks.database.DatabricksDatabase;
+
+@DataTypeInfo(name = "array<int>", minParameters = 0, maxParameters = 0, priority = DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE)
+public class ArrayIntegerDataTypeDatabricks extends LiquibaseDataType {
+
+
+    public ArrayIntegerDataTypeDatabricks() {
+        // empty constructor
+    }
+
+    @Override
+    public DatabaseDataType toDatabaseDataType(Database database) {
+        if (database instanceof DatabricksDatabase) {
+            return new DatabaseDataType("ARARY<INT>");
+        }
+
+        return super.toDatabaseDataType(database);
+    }
+
+    @Override
+    public boolean supports(Database database) {
+        return database instanceof DatabricksDatabase;
+    }
+
+    @Override
+    public int getPriority() {
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
+    }
+
+
+    @Override
+    public LoadDataChange.LOAD_DATA_TYPE getLoadTypeName() {
+        return LoadDataChange.LOAD_DATA_TYPE.STRING;
+    }
+}
\ No newline at end of file
diff --git a/src/main/java/liquibase/ext/databricks/datatype/ArrayStringDataTypeDatabricks.java b/src/main/java/liquibase/ext/databricks/datatype/ArrayStringDataTypeDatabricks.java
new file mode 100644
index 00000000..2ce72df8
--- /dev/null
+++ b/src/main/java/liquibase/ext/databricks/datatype/ArrayStringDataTypeDatabricks.java
@@ -0,0 +1,42 @@
+package liquibase.ext.databricks.datatype;
+
+import liquibase.change.core.LoadDataChange;
+import liquibase.database.Database;
+import liquibase.datatype.DataTypeInfo;
+import liquibase.datatype.DatabaseDataType;
+import liquibase.datatype.LiquibaseDataType;
+import liquibase.datatype.core.BigIntType;
+import liquibase.ext.databricks.database.DatabricksDatabase;
+
+@DataTypeInfo(name = "array<string>", minParameters = 0, maxParameters = 0, priority = DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE)
+public class ArrayStringDataTypeDatabricks extends LiquibaseDataType {
+
+
+    public ArrayStringDataTypeDatabricks() {
+        // empty constructor
+    }
+    @Override
+    public DatabaseDataType toDatabaseDataType(Database database) {
+        if (database instanceof DatabricksDatabase) {
+            return new DatabaseDataType("ARARY<STRING>");
+        }
+
+        return super.toDatabaseDataType(database);
+    }
+
+    @Override
+    public boolean supports(Database database) {
+        return database instanceof DatabricksDatabase;
+    }
+
+    @Override
+    public int getPriority() {
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
+    }
+
+
+    @Override
+    public LoadDataChange.LOAD_DATA_TYPE getLoadTypeName() {
+        return LoadDataChange.LOAD_DATA_TYPE.STRING;
+    }
+}
\ No newline at end of file
diff --git a/src/main/java/liquibase/ext/databricks/datatype/BigintDatatypeDatabricks.java b/src/main/java/liquibase/ext/databricks/datatype/BigintDatatypeDatabricks.java
index 0b1d41c5..2e69dc32 100644
--- a/src/main/java/liquibase/ext/databricks/datatype/BigintDatatypeDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/datatype/BigintDatatypeDatabricks.java
@@ -10,7 +10,7 @@
 
 
 
-@DataTypeInfo(name = "bigint", aliases = {"java.sql.Types.BIGINT", "java.math.BigInteger", "java.lang.Long", "integer8", "bigserial", "serial8", "int8"}, minParameters = 0, maxParameters = 0, priority = DatabricksDatabase.PRIORITY_DATABASE)
+@DataTypeInfo(name = "bigint", aliases = {"java.sql.Types.BIGINT", "java.math.BigInteger", "java.lang.Long", "integer8", "bigserial", "serial8", "int8"}, minParameters = 0, maxParameters = 0, priority = DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE)
 public class BigintDatatypeDatabricks extends BigIntType {
 
     private boolean autoIncrement;
@@ -40,7 +40,7 @@ public boolean supports(Database database) {
 
     @Override
     public int getPriority() {
-        return PRIORITY_DATABASE;
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
     }
 
 
diff --git a/src/main/java/liquibase/ext/databricks/datatype/BinaryDataTypeDatabricks.java b/src/main/java/liquibase/ext/databricks/datatype/BinaryDataTypeDatabricks.java
new file mode 100644
index 00000000..589316ff
--- /dev/null
+++ b/src/main/java/liquibase/ext/databricks/datatype/BinaryDataTypeDatabricks.java
@@ -0,0 +1,35 @@
+package liquibase.ext.databricks.datatype;
+
+import liquibase.change.core.LoadDataChange;
+import liquibase.database.Database;
+import liquibase.datatype.DataTypeInfo;
+import liquibase.datatype.DatabaseDataType;
+import liquibase.datatype.LiquibaseDataType;
+import liquibase.datatype.core.BlobType;
+import liquibase.ext.databricks.database.DatabricksDatabase;
+
+
+public class BinaryDataTypeDatabricks extends BlobType {
+
+
+    @Override
+    public DatabaseDataType toDatabaseDataType(Database database) {
+
+        if (database instanceof DatabricksDatabase) {
+            return new DatabaseDataType("BINARY");
+        }
+
+        return super.toDatabaseDataType(database);
+    }
+
+    @Override
+    public int getPriority() {
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
+    }
+
+
+    @Override
+    public boolean supports(Database database) {
+        return database instanceof DatabricksDatabase;
+    }
+}
diff --git a/src/main/java/liquibase/ext/databricks/datatype/BooleanDatatypeDatabricks.java b/src/main/java/liquibase/ext/databricks/datatype/BooleanDatatypeDatabricks.java
index 5840aef1..e912ac6c 100644
--- a/src/main/java/liquibase/ext/databricks/datatype/BooleanDatatypeDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/datatype/BooleanDatatypeDatabricks.java
@@ -12,7 +12,7 @@
         name = "boolean",
         minParameters = 0,
         maxParameters = 0,
-        priority = PrioritizedService.PRIORITY_DATABASE
+        priority = DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE
 )
 public class BooleanDatatypeDatabricks extends LiquibaseDataType {
 
@@ -38,6 +38,12 @@ public DatabaseDataType toDatabaseDataType(Database database) {
 
     }
 
+    @Override
+    public int getPriority() {
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
+    }
+
+
     public LoadDataChange.LOAD_DATA_TYPE getLoadTypeName() {
         return LoadDataChange.LOAD_DATA_TYPE.BOOLEAN;
     }
diff --git a/src/main/java/liquibase/ext/databricks/datatype/DatetimeDatatypeDatabricks.java b/src/main/java/liquibase/ext/databricks/datatype/DatetimeDatatypeDatabricks.java
index a403e059..d59b14aa 100644
--- a/src/main/java/liquibase/ext/databricks/datatype/DatetimeDatatypeDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/datatype/DatetimeDatatypeDatabricks.java
@@ -7,14 +7,12 @@
 import liquibase.datatype.LiquibaseDataType;
 import liquibase.ext.databricks.database.DatabricksDatabase;
 
-import static liquibase.servicelocator.PrioritizedService.PRIORITY_DATABASE;
-
 @DataTypeInfo(
         name = "timestamp",
         aliases = {"java.sql.Types.DATETIME", "datetime"},
         minParameters = 0,
         maxParameters = 0,
-        priority = PRIORITY_DATABASE
+        priority = DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE
 )
 public class DatetimeDatatypeDatabricks extends LiquibaseDataType {
 
@@ -34,8 +32,9 @@ public LoadDataChange.LOAD_DATA_TYPE getLoadTypeName() {
         return LoadDataChange.LOAD_DATA_TYPE.DATE;
     }
 
+    @Override
     public int getPriority() {
-        return PRIORITY_DATABASE;
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
     }
 
     @Override
diff --git a/src/main/java/liquibase/ext/databricks/datatype/DoubleDatatypeDatabricks.java b/src/main/java/liquibase/ext/databricks/datatype/DoubleDatatypeDatabricks.java
index 991a8663..202071d6 100644
--- a/src/main/java/liquibase/ext/databricks/datatype/DoubleDatatypeDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/datatype/DoubleDatatypeDatabricks.java
@@ -12,7 +12,7 @@
         name = "double",
         minParameters = 0,
         maxParameters = 0,
-        priority = PrioritizedService.PRIORITY_DATABASE
+        priority = DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE
 )
 public class DoubleDatatypeDatabricks extends LiquibaseDataType {
 
@@ -38,6 +38,11 @@ public DatabaseDataType toDatabaseDataType(Database database) {
 
     }
 
+    @Override
+    public int getPriority() {
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
+    }
+
     public LoadDataChange.LOAD_DATA_TYPE getLoadTypeName() {
         return LoadDataChange.LOAD_DATA_TYPE.NUMERIC;
     }
diff --git a/src/main/java/liquibase/ext/databricks/datatype/FloatDatatypeDatabricks.java b/src/main/java/liquibase/ext/databricks/datatype/FloatDatatypeDatabricks.java
index b35fed61..e3c232ba 100644
--- a/src/main/java/liquibase/ext/databricks/datatype/FloatDatatypeDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/datatype/FloatDatatypeDatabricks.java
@@ -13,7 +13,7 @@
         name = "float",
         minParameters = 0,
         maxParameters = 0,
-        priority = PrioritizedService.PRIORITY_DATABASE
+        priority = DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE
 )
 public class FloatDatatypeDatabricks extends LiquibaseDataType {
     public FloatDatatypeDatabricks() {
@@ -38,6 +38,11 @@ public DatabaseDataType toDatabaseDataType(Database database) {
 
     }
 
+    @Override
+    public int getPriority() {
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
+    }
+
     public LoadDataChange.LOAD_DATA_TYPE getLoadTypeName() {
         return LoadDataChange.LOAD_DATA_TYPE.NUMERIC;
     }
diff --git a/src/main/java/liquibase/ext/databricks/datatype/IntegerDatatypeDatabricks.java b/src/main/java/liquibase/ext/databricks/datatype/IntegerDatatypeDatabricks.java
index 82cb4a79..113b0b08 100644
--- a/src/main/java/liquibase/ext/databricks/datatype/IntegerDatatypeDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/datatype/IntegerDatatypeDatabricks.java
@@ -6,14 +6,13 @@
 import liquibase.datatype.DatabaseDataType;
 import liquibase.datatype.LiquibaseDataType;
 import liquibase.ext.databricks.database.DatabricksDatabase;
-import liquibase.servicelocator.PrioritizedService;
 
 
 @DataTypeInfo(
         name = "int",
         minParameters = 0,
         maxParameters = 0,
-        priority = PrioritizedService.PRIORITY_DATABASE
+        priority = DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE
 )
 public class IntegerDatatypeDatabricks extends LiquibaseDataType {
     public IntegerDatatypeDatabricks() {
diff --git a/src/main/java/liquibase/ext/databricks/datatype/SmallintDatatypeDatabricks.java b/src/main/java/liquibase/ext/databricks/datatype/SmallintDatatypeDatabricks.java
new file mode 100644
index 00000000..6a833c6d
--- /dev/null
+++ b/src/main/java/liquibase/ext/databricks/datatype/SmallintDatatypeDatabricks.java
@@ -0,0 +1,46 @@
+package liquibase.ext.databricks.datatype;
+
+import liquibase.change.core.LoadDataChange;
+import liquibase.database.Database;
+import liquibase.datatype.DataTypeInfo;
+import liquibase.datatype.DatabaseDataType;
+import liquibase.datatype.LiquibaseDataType;
+import liquibase.ext.databricks.database.DatabricksDatabase;
+
+@DataTypeInfo(
+        name = "smallint",
+        aliases = {"java.sql.Types.SMALLINT", "short"},
+        minParameters = 0,
+        maxParameters = 0,
+        priority = DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE
+)
+public class SmallintDatatypeDatabricks extends LiquibaseDataType {
+
+
+    @Override
+    public DatabaseDataType toDatabaseDataType(Database database) {
+
+        if (database instanceof DatabricksDatabase) {
+            return new DatabaseDataType("SMALLINT");
+        }
+
+        return super.toDatabaseDataType(database);
+    }
+
+    @Override
+    public LoadDataChange.LOAD_DATA_TYPE getLoadTypeName() {
+        return LoadDataChange.LOAD_DATA_TYPE.NUMERIC;
+    }
+
+    @Override
+    public int getPriority() {
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
+    }
+
+    @Override
+    public boolean supports(Database database) {
+        return database instanceof DatabricksDatabase;
+    }
+
+
+}
diff --git a/src/main/java/liquibase/ext/databricks/datatype/StringDatatypeDatabricks.java b/src/main/java/liquibase/ext/databricks/datatype/StringDatatypeDatabricks.java
index eb06b5b4..8001ebd6 100644
--- a/src/main/java/liquibase/ext/databricks/datatype/StringDatatypeDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/datatype/StringDatatypeDatabricks.java
@@ -7,14 +7,11 @@
 import liquibase.datatype.LiquibaseDataType;
 import liquibase.ext.databricks.database.DatabricksDatabase;
 
-import static liquibase.ext.databricks.database.DatabricksDatabase.PRIORITY_DATABASE;
-
-
 @DataTypeInfo(
         name = "string",
         minParameters = 0,
         maxParameters = 0,
-        priority = PRIORITY_DATABASE
+        priority = DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE
 )
 public class StringDatatypeDatabricks extends LiquibaseDataType {
     public StringDatatypeDatabricks() {
@@ -38,6 +35,12 @@ public DatabaseDataType toDatabaseDataType(Database database) {
 
     }
 
+    @Override
+    public int getPriority() {
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
+    }
+
+
     public LoadDataChange.LOAD_DATA_TYPE getLoadTypeName() {
         return LoadDataChange.LOAD_DATA_TYPE.STRING;
     }
diff --git a/src/main/java/liquibase/ext/databricks/datatype/TimestampDatatypeDatabricks.java b/src/main/java/liquibase/ext/databricks/datatype/TimestampDatatypeDatabricks.java
index dacb7b3c..93b4114b 100644
--- a/src/main/java/liquibase/ext/databricks/datatype/TimestampDatatypeDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/datatype/TimestampDatatypeDatabricks.java
@@ -9,7 +9,6 @@
 import liquibase.database.core.*;
 import liquibase.datatype.DataTypeInfo;
 import liquibase.datatype.DatabaseDataType;
-import liquibase.datatype.LiquibaseDataType;
 import liquibase.exception.DatabaseIncapableOfOperation;
 import liquibase.util.StringUtil;
 import liquibase.util.grammar.ParseException;
@@ -19,7 +18,7 @@
  * year, month, day, hour, minute and second parts. Optionally, fractional seconds and time zone information can be
  * specified as well.
  */
-@DataTypeInfo(name = "timestamp", aliases = {"java.sql.Types.TIMESTAMP", "java.sql.Types.TIMESTAMP_WITH_TIMEZONE", "java.sql.Timestamp", "timestamptz"}, minParameters = 0, maxParameters = 0, priority = DatabricksDatabase.PRIORITY_DATABASE)
+@DataTypeInfo(name = "timestamp", aliases = {"java.sql.Types.TIMESTAMP", "java.sql.Types.TIMESTAMP_WITH_TIMEZONE", "java.sql.Timestamp", "timestamptz"}, minParameters = 0, maxParameters = 0, priority = DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE)
 public class TimestampDatatypeDatabricks extends TimestampType {
 
     /**
@@ -183,6 +182,12 @@ public DatabaseDataType toDatabaseDataType(Database database) {
         return super.toDatabaseDataType(database);
     }
 
+    @Override
+    public int getPriority() {
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
+    }
+
+
     @Override
     public LoadDataChange.LOAD_DATA_TYPE getLoadTypeName() {
         return LoadDataChange.LOAD_DATA_TYPE.DATE;
diff --git a/src/main/java/liquibase/ext/databricks/datatype/TinyintDatatypeDatabricks.java b/src/main/java/liquibase/ext/databricks/datatype/TinyintDatatypeDatabricks.java
new file mode 100644
index 00000000..13ed54c3
--- /dev/null
+++ b/src/main/java/liquibase/ext/databricks/datatype/TinyintDatatypeDatabricks.java
@@ -0,0 +1,47 @@
+package liquibase.ext.databricks.datatype;
+
+import liquibase.change.core.LoadDataChange;
+import liquibase.database.Database;
+import liquibase.datatype.DataTypeInfo;
+import liquibase.datatype.DatabaseDataType;
+import liquibase.datatype.LiquibaseDataType;
+import liquibase.ext.databricks.database.DatabricksDatabase;
+
+@DataTypeInfo(
+        name = "tinyint",
+        aliases = {"java.sql.Types.TINYINT", "byte"},
+        minParameters = 0,
+        maxParameters = 0,
+        priority = DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE
+)
+public class TinyintDatatypeDatabricks extends LiquibaseDataType {
+
+
+    @Override
+    public DatabaseDataType toDatabaseDataType(Database database) {
+
+        if (database instanceof DatabricksDatabase) {
+            return new DatabaseDataType("TINYINT");
+        }
+
+        return super.toDatabaseDataType(database);
+    }
+
+    @Override
+    public LoadDataChange.LOAD_DATA_TYPE getLoadTypeName() {
+        return LoadDataChange.LOAD_DATA_TYPE.NUMERIC;
+    }
+
+    @Override
+    public int getPriority() {
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
+    }
+
+
+    @Override
+    public boolean supports(Database database) {
+        return database instanceof DatabricksDatabase;
+    }
+
+
+}
diff --git a/src/main/java/liquibase/ext/databricks/executor/DatabricksExecutor.java b/src/main/java/liquibase/ext/databricks/executor/DatabricksExecutor.java
index fa24c3d5..dd06729f 100644
--- a/src/main/java/liquibase/ext/databricks/executor/DatabricksExecutor.java
+++ b/src/main/java/liquibase/ext/databricks/executor/DatabricksExecutor.java
@@ -8,13 +8,13 @@
 import liquibase.sql.visitor.SqlVisitor;
 import liquibase.statement.SqlStatement;
 import java.util.List;
-import static liquibase.ext.databricks.database.DatabricksDatabase.PRIORITY_DATABASE;
+import static liquibase.ext.databricks.database.DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
 
 public class DatabricksExecutor extends JdbcExecutor {
 
     @Override
     public int getPriority() {
-        return PRIORITY_DATABASE;
+        return DATABRICKS_PRIORITY_DATABASE;
     }
 
     @Override
diff --git a/src/main/java/liquibase/ext/databricks/snapshot/jvm/IndexSnapshotGeneratorDatabricks.java b/src/main/java/liquibase/ext/databricks/snapshot/jvm/IndexSnapshotGeneratorDatabricks.java
index 2f375c52..87a65834 100644
--- a/src/main/java/liquibase/ext/databricks/snapshot/jvm/IndexSnapshotGeneratorDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/snapshot/jvm/IndexSnapshotGeneratorDatabricks.java
@@ -21,7 +21,7 @@ public class IndexSnapshotGeneratorDatabricks extends IndexSnapshotGenerator {
     @Override
     public int getPriority(Class<? extends DatabaseObject> objectType, Database database) {
         if (super.getPriority(objectType, database) > 0 && database instanceof DatabricksDatabase) {
-            return DatabricksDatabase.PRIORITY_DATABASE;
+            return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
         } else {
             return PRIORITY_NONE;
         }
diff --git a/src/main/java/liquibase/ext/databricks/snapshot/jvm/SchemaSnapshotGeneratorDatabricks.java b/src/main/java/liquibase/ext/databricks/snapshot/jvm/SchemaSnapshotGeneratorDatabricks.java
index 4f0deaf4..4bff6d2a 100644
--- a/src/main/java/liquibase/ext/databricks/snapshot/jvm/SchemaSnapshotGeneratorDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/snapshot/jvm/SchemaSnapshotGeneratorDatabricks.java
@@ -7,14 +7,17 @@
 import liquibase.database.jvm.JdbcConnection;
 import liquibase.diff.compare.DatabaseObjectComparatorFactory;
 import liquibase.exception.DatabaseException;
+import liquibase.ext.databricks.database.DatabricksConnection;
 import liquibase.ext.databricks.database.DatabricksDatabase;
 import liquibase.snapshot.DatabaseSnapshot;
 import liquibase.snapshot.InvalidExampleException;
 import liquibase.snapshot.jvm.SchemaSnapshotGenerator;
 import liquibase.structure.DatabaseObject;
 import liquibase.structure.core.Catalog;
+import liquibase.structure.core.Data;
 import liquibase.structure.core.Schema;
 import liquibase.util.JdbcUtils;
+import liquibase.util.JdbcUtil;
 
 import java.sql.ResultSet;
 import java.sql.SQLException;
@@ -22,106 +25,27 @@
 import java.util.List;
 
 public class SchemaSnapshotGeneratorDatabricks extends SchemaSnapshotGenerator {
-
     @Override
     public int getPriority(Class<? extends DatabaseObject> objectType, Database database) {
-        int priority = super.getPriority(objectType, database);
-        if (priority > PRIORITY_NONE && database instanceof DatabricksDatabase) {
-            priority += PRIORITY_DATABASE;
+        if (database instanceof DatabricksDatabase) {
+            return super.getPriority(objectType, database) + PRIORITY_DATABASE;
+        } else {
+            return PRIORITY_NONE;
         }
-        return priority;
     }
 
     @Override
     protected String[] getDatabaseSchemaNames(Database database) throws SQLException, DatabaseException {
         List<String> returnList = new ArrayList<>();
 
-        ResultSet schemas = null;
-        try {
-            schemas = ((JdbcConnection) database.getConnection()).getMetaData()
-                    .getSchemas(database.getDefaultCatalogName(), null);
-
+        try (ResultSet schemas = ((JdbcConnection) database.getConnection()).getMetaData().getSchemas(database
+                .getDefaultCatalogName(), null)) {
             while (schemas.next()) {
-                returnList.add(JdbcUtils.getValueForColumn(schemas, "TABLE_SCHEM", database));
-            }
-        } finally {
-            if (schemas != null) {
-                schemas.close();
-            }
-        }
-
-        return returnList.toArray(new String[returnList.size()]);
-    }
-
-    @Override
-    protected DatabaseObject snapshotObject(DatabaseObject example, DatabaseSnapshot snapshot) throws DatabaseException, InvalidExampleException {
-        Database database = snapshot.getDatabase();
-        Schema match = null;
-
-        String catalogName = ((Schema) example).getCatalogName();
-        String schemaName = example.getName();
-        if (database.supportsSchemas()) {
-            if (catalogName == null) {
-                catalogName = database.getDefaultCatalogName();
-            }
-
-            if (schemaName == null) {
-                schemaName = database.getDefaultSchemaName();
-            }
-        } else if (database.supportsCatalogs()) {
-            if (catalogName == null && schemaName != null) {
-                catalogName = schemaName;
-                schemaName = null;
-            }
-        } else {
-            catalogName = null;
-            schemaName = null;
-        }
-
-        Schema example1 = new Schema(catalogName, schemaName);
-        ObjectQuotingStrategy currentStrategy = database.getObjectQuotingStrategy();
-        database.setObjectQuotingStrategy(ObjectQuotingStrategy.LEGACY);
-
-        try {
-            if (database.supportsSchemas()) {
-                String[] schemaNames = this.getDatabaseSchemaNames(database);
-
-                for (String tableSchema : schemaNames) {
-                    CatalogAndSchema schemaFromJdbcInfo = this.toCatalogAndSchema(tableSchema, database);
-                    Catalog catalog = new Catalog(schemaFromJdbcInfo.getCatalogName());
-                    Schema schema = new Schema(catalog, tableSchema);
-                    if (DatabaseObjectComparatorFactory.getInstance().isSameObject(schema, example1, snapshot.getSchemaComparisons(), database)) {
-                        if (match != null) {
-                            throw new InvalidExampleException("Found multiple catalog/schemas matching " + ((Schema) example).getCatalogName() + "." + example.getName());
-                        }
-
-                        match = schema;
-                    }
-                }
-            } else if (example1.getCatalog().isDefault()) {
-                match = new Schema(example1.getCatalog(), catalogName);
-            } else {
-                Catalog catalog = example1.getCatalog();
-                String[] dbCatalogNames = this.getDatabaseCatalogNames(database);
-
-                for (int i = 0; i < dbCatalogNames.length; ++i) {
-                    String candidateCatalogName = dbCatalogNames[i];
-                    if (catalog.equals(new Catalog(candidateCatalogName))) {
-                        match = new Schema(catalog, catalogName);
-                    }
-                }
+                returnList.add(JdbcUtil.getValueForColumn(schemas, "TABLE_SCHEM", database));
             }
-        } catch (SQLException e) {
-            throw new DatabaseException(e);
-        } finally {
-            database.setObjectQuotingStrategy(currentStrategy);
-        }
-
-        if (match != null && (match.getName() == null || match.getName().equalsIgnoreCase(database.getDefaultSchemaName()))) {
-            match.setDefault(true);
         }
 
-        return match;
+        return returnList.toArray(new String[0]);
     }
 
 }
\ No newline at end of file
diff --git a/src/main/java/liquibase/ext/databricks/snapshot/jvm/SequenceSnapshotGeneratorDatabricks.java b/src/main/java/liquibase/ext/databricks/snapshot/jvm/SequenceSnapshotGeneratorDatabricks.java
index e38e4ca8..c7c59d3b 100644
--- a/src/main/java/liquibase/ext/databricks/snapshot/jvm/SequenceSnapshotGeneratorDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/snapshot/jvm/SequenceSnapshotGeneratorDatabricks.java
@@ -20,7 +20,7 @@ public class SequenceSnapshotGeneratorDatabricks extends SequenceSnapshotGenerat
     public int getPriority(Class<? extends DatabaseObject> objectType, Database database) {
         int priority = super.getPriority(objectType, database);
         if (priority > PRIORITY_NONE && database instanceof DatabricksDatabase) {
-            priority += PRIORITY_DATABASE;
+            priority += DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
         }
         return priority;
     }
diff --git a/src/main/java/liquibase/ext/databricks/snapshot/jvm/UniqueConstraintSnapshotGeneratorDatabricks.java b/src/main/java/liquibase/ext/databricks/snapshot/jvm/UniqueConstraintSnapshotGeneratorDatabricks.java
index 3f5c2789..fe34237a 100644
--- a/src/main/java/liquibase/ext/databricks/snapshot/jvm/UniqueConstraintSnapshotGeneratorDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/snapshot/jvm/UniqueConstraintSnapshotGeneratorDatabricks.java
@@ -22,7 +22,7 @@ public class UniqueConstraintSnapshotGeneratorDatabricks extends UniqueConstrain
     public int getPriority(Class<? extends DatabaseObject> objectType, Database database) {
         int priority = super.getPriority(objectType, database);
         if (priority > PRIORITY_NONE && database instanceof DatabricksDatabase) {
-            priority += DatabricksDatabase.PRIORITY_DATABASE;
+            priority += DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
         }
         return priority;
     }
diff --git a/src/main/java/liquibase/ext/databricks/snapshot/jvm/ViewSnapshotGeneratorDatabricks.java b/src/main/java/liquibase/ext/databricks/snapshot/jvm/ViewSnapshotGeneratorDatabricks.java
index 1205ca7b..82faa7eb 100644
--- a/src/main/java/liquibase/ext/databricks/snapshot/jvm/ViewSnapshotGeneratorDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/snapshot/jvm/ViewSnapshotGeneratorDatabricks.java
@@ -4,8 +4,10 @@
 import liquibase.Scope;
 import liquibase.database.AbstractJdbcDatabase;
 import liquibase.database.Database;
+import liquibase.database.DatabaseConnection;
 import liquibase.exception.DatabaseException;
 import liquibase.executor.ExecutorService;
+import liquibase.ext.databricks.database.DatabricksConnection;
 import liquibase.snapshot.DatabaseSnapshot;
 import liquibase.snapshot.jvm.ViewSnapshotGenerator;
 import liquibase.statement.core.RawSqlStatement;
@@ -14,10 +16,12 @@
 import liquibase.structure.core.View;
 import liquibase.util.StringUtil;
 
+import java.sql.ResultSet;
 import java.util.List;
 import java.util.Map;
 
 import liquibase.ext.databricks.database.DatabricksDatabase;
+
 public class ViewSnapshotGeneratorDatabricks extends ViewSnapshotGenerator {
 
 
@@ -25,7 +29,7 @@ public class ViewSnapshotGeneratorDatabricks extends ViewSnapshotGenerator {
     public int getPriority(Class<? extends DatabaseObject> objectType, Database database) {
         int priority = super.getPriority(objectType, database);
         if (priority > PRIORITY_NONE && database instanceof DatabricksDatabase) {
-            priority += DatabricksDatabase.PRIORITY_DATABASE;
+            priority += DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
         }
         return priority;
     }
@@ -37,31 +41,57 @@ protected DatabaseObject snapshotObject(DatabaseObject example, DatabaseSnapshot
         } else {
             Database database = snapshot.getDatabase();
             Schema schema = example.getSchema();
+            DatabaseConnection connection = database.getConnection();
 
             CatalogAndSchema catalogAndSchema = (new CatalogAndSchema(schema.getCatalogName(), schema.getName())).customize(database);
             String jdbcSchemaName = database.correctObjectName(((AbstractJdbcDatabase) database).getJdbcSchemaName(catalogAndSchema), Schema.class);
             String query = String.format("SELECT view_definition FROM %s.%s.VIEWS WHERE table_name='%s' AND table_schema='%s' AND table_catalog='%s';",
                     schema.getCatalogName(), database.getSystemSchema(), example.getName(), schema.getName(), schema.getCatalogName());
 
+            // DEBUG
+            //System.out.println("Snapshot Database Connection URL : " + database.getConnection().getURL());
+            //System.out.println("Snapshot Database Connection Class : " + database.getConnection().getClass().getName());
+
+
             List<Map<String, ?>> viewsMetadataRs = Scope.getCurrentScope().getSingleton(ExecutorService.class)
                     .getExecutor("jdbc", database).queryForList(new RawSqlStatement(query));
 
+            // New Code, likely superfluous, was used for testing
+            /// This should use our existing DatabaseConnection url processing
+            String rawViewDefinition = null;
+
+            try (ResultSet viewMetadataResultSet = ((DatabricksConnection) connection).createStatement().executeQuery(query)) {
+                //System.out.println("Raw Result VIEW " + viewMetadataResultSet);
+
+                viewMetadataResultSet.next();
+                rawViewDefinition = viewMetadataResultSet.getString(1);
+
+
+            } catch (Exception e) {
+                Scope.getCurrentScope().getLog(getClass()).info("Error getting View Definiton via existing context, going to pull from URL", e);
+            }
+
+            /// Old Code
+
             if (viewsMetadataRs.isEmpty()) {
                 return null;
             } else {
+
                 Map<String, ?> row = viewsMetadataRs.get(0);
                 String rawViewName = example.getName();
                 String rawSchemaName = schema.getName();
                 String rawCatalogName = schema.getCatalogName();
 
+
                 View view = (new View()).setName(this.cleanNameFromDatabase(rawViewName, database));
                 CatalogAndSchema schemaFromJdbcInfo = ((AbstractJdbcDatabase) database).getSchemaFromJdbcInfo(rawCatalogName, rawSchemaName);
                 view.setSchema(new Schema(schemaFromJdbcInfo.getCatalogName(), schemaFromJdbcInfo.getSchemaName()));
 
-                String definition = (String) row.get("VIEW_DEFINITION");
-                if (definition.startsWith("FULL_DEFINITION: ")) {
-                    definition = definition.replaceFirst("^FULL_DEFINITION: ", "");
-                    view.setContainsFullDefinition(true);
+                String definition = rawViewDefinition;
+
+                if (definition == null || definition.isEmpty()) {
+                    definition = (String) row.get("view_definition");
+
                 }
 
                 int length = definition.length();
@@ -81,5 +111,4 @@ protected DatabaseObject snapshotObject(DatabaseObject example, DatabaseSnapshot
 
         }
     }
-
-}
\ No newline at end of file
+}
diff --git a/src/main/java/liquibase/ext/databricks/sqlgenerator/AddAutoIncrementGeneratorDatabricks.java b/src/main/java/liquibase/ext/databricks/sqlgenerator/AddAutoIncrementGeneratorDatabricks.java
index 39ac2d5e..918740fa 100644
--- a/src/main/java/liquibase/ext/databricks/sqlgenerator/AddAutoIncrementGeneratorDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/sqlgenerator/AddAutoIncrementGeneratorDatabricks.java
@@ -11,7 +11,7 @@ public class AddAutoIncrementGeneratorDatabricks extends AddAutoIncrementGenerat
 
     @Override
     public int getPriority() {
-        return DatabricksDatabase.PRIORITY_DATABASE;
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
     }
 
     @Override
diff --git a/src/main/java/liquibase/ext/databricks/sqlgenerator/AddColumnGeneratorDatabricks.java b/src/main/java/liquibase/ext/databricks/sqlgenerator/AddColumnGeneratorDatabricks.java
index 5c2faefe..1001a543 100644
--- a/src/main/java/liquibase/ext/databricks/sqlgenerator/AddColumnGeneratorDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/sqlgenerator/AddColumnGeneratorDatabricks.java
@@ -1,24 +1,119 @@
 package liquibase.ext.databricks.sqlgenerator;
 
 import liquibase.database.Database;
+import liquibase.database.core.*;
+import liquibase.datatype.DataTypeFactory;
+import liquibase.datatype.DatabaseDataType;
 import liquibase.ext.databricks.database.DatabricksDatabase;
 import liquibase.sqlgenerator.core.AddColumnGenerator;
+import liquibase.statement.AutoIncrementConstraint;
+import liquibase.statement.ColumnConstraint;
+import liquibase.statement.DatabaseFunction;
+import liquibase.statement.NotNullConstraint;
 import liquibase.statement.core.AddColumnStatement;
+import liquibase.util.StringUtil;
 
-public class AddColumnGeneratorDatabricks extends AddColumnGenerator {
+import java.util.Iterator;
 
-    @Override
-    protected String generateSingleColumnSQL(AddColumnStatement statement, Database database) {
-        return super.generateSingleColumnSQL(statement, database).replace(" ADD ", " ADD COLUMN ");
-    }
+public class AddColumnGeneratorDatabricks extends AddColumnGenerator {
 
     @Override
     public int getPriority() {
-        return PRIORITY_DATABASE;
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
     }
 
     @Override
     public boolean supports(AddColumnStatement statement, Database database) {
         return database instanceof DatabricksDatabase;
     }
-}
\ No newline at end of file
+
+    @Override
+    protected String generateSingleColumnSQL(AddColumnStatement statement, Database database) {
+        DatabaseDataType columnType = null;
+        if (statement.getColumnType() != null) {
+            columnType = DataTypeFactory.getInstance().fromDescription(statement.getColumnType() + (statement.isAutoIncrement() ? "{autoIncrement:true}" : ""), database).toDatabaseDataType(database);
+        }
+
+        StringBuilder alterTable = new StringBuilder(" ADD COLUMN ").append(database.escapeColumnName(statement.getCatalogName(), statement.getSchemaName(), statement.getTableName(), statement.getColumnName()));
+        if (columnType != null) {
+            alterTable.append(" ").append(columnType);
+        }
+
+        if (statement.isAutoIncrement() && database.supportsAutoIncrement()) {
+            AutoIncrementConstraint autoIncrementConstraint = statement.getAutoIncrementConstraint();
+            alterTable.append(" ").append(database.getAutoIncrementClause(autoIncrementConstraint.getStartWith(), autoIncrementConstraint.getIncrementBy(), autoIncrementConstraint.getGenerationType(), autoIncrementConstraint.getDefaultOnNull()));
+        }
+
+        alterTable.append(this.getDefaultClauseForColumn(statement, database));
+        if (!statement.isNullable()) {
+            Iterator<ColumnConstraint> var8 = statement.getConstraints().iterator();
+
+            while(var8.hasNext()) {
+                ColumnConstraint constraint = var8.next();
+                if (constraint instanceof NotNullConstraint) {
+                    NotNullConstraint notNullConstraint = (NotNullConstraint)constraint;
+                    if (StringUtil.isNotEmpty(notNullConstraint.getConstraintName())) {
+                        alterTable.append(" CONSTRAINT ").append(database.escapeConstraintName(notNullConstraint.getConstraintName()));
+                        break;
+                    }
+                }
+            }
+
+            alterTable.append(" NOT NULL");
+            if (database instanceof OracleDatabase) {
+                alterTable.append(!statement.shouldValidateNullable() ? " ENABLE NOVALIDATE " : "");
+            }
+        } else if (database instanceof SybaseDatabase || database instanceof SybaseASADatabase || database instanceof MySQLDatabase || database instanceof MSSQLDatabase && columnType != null && "timestamp".equalsIgnoreCase(columnType.toString())) {
+            alterTable.append(" NULL");
+        }
+
+        if (statement.isPrimaryKey()) {
+            alterTable.append(" PRIMARY KEY");
+            if (database instanceof OracleDatabase) {
+                alterTable.append(!statement.shouldValidatePrimaryKey() ? " ENABLE NOVALIDATE " : "");
+            }
+        }
+
+        if (database instanceof MySQLDatabase && statement.getRemarks() != null) {
+            alterTable.append(" COMMENT '").append(database.escapeStringForDatabase(StringUtil.trimToEmpty(statement.getRemarks()))).append("' ");
+        }
+
+        if (statement.getAddBeforeColumn() != null && !statement.getAddBeforeColumn().isEmpty()) {
+            alterTable.append(" BEFORE ").append(database.escapeColumnName(statement.getSchemaName(), statement.getSchemaName(), statement.getTableName(), statement.getAddBeforeColumn())).append(" ");
+        }
+
+        if (statement.getAddAfterColumn() != null && !statement.getAddAfterColumn().isEmpty()) {
+            alterTable.append(" AFTER ").append(database.escapeColumnName(statement.getSchemaName(), statement.getSchemaName(), statement.getTableName(), statement.getAddAfterColumn()));
+        }
+
+        return alterTable.toString();
+    }
+
+
+    private String getDefaultClauseForColumn(AddColumnStatement statement, Database database) {
+        StringBuilder clause = new StringBuilder();
+        Object defaultValue = statement.getDefaultValue();
+        if (defaultValue != null) {
+            if (database instanceof OracleDatabase && defaultValue.toString().startsWith("GENERATED ALWAYS ")) {
+                clause.append(" ").append(DataTypeFactory.getInstance().fromObject(defaultValue, database).objectToSql(defaultValue, database));
+            } else {
+                if (database instanceof MSSQLDatabase) {
+                    String constraintName = statement.getDefaultValueConstraintName();
+                    if (constraintName == null) {
+                        constraintName = ((MSSQLDatabase)database).generateDefaultConstraintName(statement.getTableName(), statement.getColumnName());
+                    }
+
+                    clause.append(" CONSTRAINT ").append(constraintName);
+                }
+
+                if (defaultValue instanceof DatabaseFunction) {
+                    clause.append(" DEFAULT ").append(DataTypeFactory.getInstance().fromObject(defaultValue, database).objectToSql(defaultValue, database));
+                } else {
+                    clause.append(" DEFAULT ").append(DataTypeFactory.getInstance().fromDescription(statement.getColumnType(), database).objectToSql(defaultValue, database));
+                }
+            }
+        }
+
+        return clause.toString();
+    }
+}
diff --git a/src/main/java/liquibase/ext/databricks/sqlgenerator/AddDefaultValueGeneratorDatabricks.java b/src/main/java/liquibase/ext/databricks/sqlgenerator/AddDefaultValueGeneratorDatabricks.java
index c1b3dfd7..320475a1 100644
--- a/src/main/java/liquibase/ext/databricks/sqlgenerator/AddDefaultValueGeneratorDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/sqlgenerator/AddDefaultValueGeneratorDatabricks.java
@@ -17,7 +17,7 @@
 public class AddDefaultValueGeneratorDatabricks extends AddDefaultValueGenerator {
     @Override
     public int getPriority() {
-        return PRIORITY_DATABASE;
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
     }
 
     @Override
diff --git a/src/main/java/liquibase/ext/databricks/sqlgenerator/AddForeignKeyConstraintGeneratorDatabricks.java b/src/main/java/liquibase/ext/databricks/sqlgenerator/AddForeignKeyConstraintGeneratorDatabricks.java
index e8511a5f..c146ba94 100644
--- a/src/main/java/liquibase/ext/databricks/sqlgenerator/AddForeignKeyConstraintGeneratorDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/sqlgenerator/AddForeignKeyConstraintGeneratorDatabricks.java
@@ -21,6 +21,11 @@ public boolean supports(AddForeignKeyConstraintStatement statement, Database dat
         return (database instanceof DatabricksDatabase);
     }
 
+    @Override
+    public int getPriority() {
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
+    }
+
     @Override
     public ValidationErrors validate(AddForeignKeyConstraintStatement addForeignKeyConstraintStatement, Database database, SqlGeneratorChain sqlGeneratorChain) {
         ValidationErrors validationErrors = new ValidationErrors();
diff --git a/src/main/java/liquibase/ext/databricks/sqlgenerator/AddPrimaryKeyGeneratorDatabricks.java b/src/main/java/liquibase/ext/databricks/sqlgenerator/AddPrimaryKeyGeneratorDatabricks.java
index 8f9566eb..236646dc 100644
--- a/src/main/java/liquibase/ext/databricks/sqlgenerator/AddPrimaryKeyGeneratorDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/sqlgenerator/AddPrimaryKeyGeneratorDatabricks.java
@@ -19,6 +19,12 @@ public boolean supports(AddPrimaryKeyStatement statement, Database database) {
         return (database instanceof DatabricksDatabase);
     }
 
+    @Override
+    public int getPriority() {
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
+    }
+
+
     @Override
     public ValidationErrors validate(AddPrimaryKeyStatement addPrimaryKeyStatement, Database database, SqlGeneratorChain sqlGeneratorChain) {
         ValidationErrors validationErrors = new ValidationErrors();
diff --git a/src/main/java/liquibase/ext/databricks/sqlgenerator/AddUniqueConstraintGeneratorDatabricks.java b/src/main/java/liquibase/ext/databricks/sqlgenerator/AddUniqueConstraintGeneratorDatabricks.java
index f6d9a1cf..47a9692a 100644
--- a/src/main/java/liquibase/ext/databricks/sqlgenerator/AddUniqueConstraintGeneratorDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/sqlgenerator/AddUniqueConstraintGeneratorDatabricks.java
@@ -12,9 +12,10 @@ public class AddUniqueConstraintGeneratorDatabricks extends AddUniqueConstraintG
 
     @Override
     public int getPriority() {
-        return DatabricksDatabase.PRIORITY_DATABASE;
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
     }
 
+
     @Override
     public boolean supports(AddUniqueConstraintStatement statement, Database database) {
         return super.supports(statement, database)
diff --git a/src/main/java/liquibase/ext/databricks/sqlgenerator/CreateIndexGeneratorDatabricks.java b/src/main/java/liquibase/ext/databricks/sqlgenerator/CreateIndexGeneratorDatabricks.java
index ba2c8d80..c7784d8e 100644
--- a/src/main/java/liquibase/ext/databricks/sqlgenerator/CreateIndexGeneratorDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/sqlgenerator/CreateIndexGeneratorDatabricks.java
@@ -30,6 +30,12 @@ public boolean supports(CreateIndexStatement statement, Database database) {
         return super.supports(statement, database) && (database instanceof DatabricksDatabase);
     }
 
+    @Override
+    public int getPriority() {
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
+    }
+
+
     @Override
     public ValidationErrors validate(CreateIndexStatement createIndexStatement, Database database,
                                      SqlGeneratorChain sqlGeneratorChain) {
diff --git a/src/main/java/liquibase/ext/databricks/sqlgenerator/CreateTableGeneratorDatabricks.java b/src/main/java/liquibase/ext/databricks/sqlgenerator/CreateTableGeneratorDatabricks.java
index ea64cf6f..f95118cf 100644
--- a/src/main/java/liquibase/ext/databricks/sqlgenerator/CreateTableGeneratorDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/sqlgenerator/CreateTableGeneratorDatabricks.java
@@ -20,7 +20,7 @@ public class CreateTableGeneratorDatabricks extends CreateTableGenerator {
 
     @Override
     public int getPriority() {
-        return PRIORITY_DATABASE;
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
     }
 
     @Override
@@ -49,7 +49,7 @@ public Sql[] generateSql(CreateTableStatement statement, Database database, SqlG
             if ((!StringUtil.isEmpty(thisStatement.getTableFormat()))) {
                 finalsql += " USING " + thisStatement.getTableFormat();
             } else {
-                finalsql += " USING delta TBLPROPERTIES('delta.feature.allowColumnDefaults' = 'supported', 'delta.columnMapping.mode' = 'name')";
+                finalsql += " USING delta TBLPROPERTIES('delta.feature.allowColumnDefaults' = 'supported', 'delta.columnMapping.mode' = 'name', 'delta.enableDeletionVectors' = true)";
             }
 
             // Databricks can decide to have tables live in a particular location. If null, Databricks will handle the location automatically in DBFS
diff --git a/src/main/java/liquibase/ext/databricks/sqlgenerator/GetViewDefinitionGeneratorDatabricks.java b/src/main/java/liquibase/ext/databricks/sqlgenerator/GetViewDefinitionGeneratorDatabricks.java
index 76a2d532..7335cfa0 100644
--- a/src/main/java/liquibase/ext/databricks/sqlgenerator/GetViewDefinitionGeneratorDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/sqlgenerator/GetViewDefinitionGeneratorDatabricks.java
@@ -18,9 +18,10 @@ public class GetViewDefinitionGeneratorDatabricks extends GetViewDefinitionGener
 
     @Override
     public int getPriority() {
-        return DatabricksDatabase.PRIORITY_DATABASE;
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
     }
 
+
     @Override
     public boolean supports(GetViewDefinitionStatement statement, Database database) {
         return database instanceof DatabricksDatabase;
diff --git a/src/main/java/liquibase/ext/databricks/sqlgenerator/InsertOrUpdateGeneratorDatabricks.java b/src/main/java/liquibase/ext/databricks/sqlgenerator/InsertOrUpdateGeneratorDatabricks.java
index bf7d16c5..95f06e1a 100644
--- a/src/main/java/liquibase/ext/databricks/sqlgenerator/InsertOrUpdateGeneratorDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/sqlgenerator/InsertOrUpdateGeneratorDatabricks.java
@@ -13,12 +13,18 @@
 
 public class InsertOrUpdateGeneratorDatabricks extends InsertOrUpdateGenerator {
 
-        @Override
-        public boolean supports(InsertOrUpdateStatement statement, Database database) {
-            return database instanceof DatabricksDatabase;
-        }
+    @Override
+    public boolean supports(InsertOrUpdateStatement statement, Database database) {
+        return database instanceof DatabricksDatabase;
+    }
 
-        @Override
+    @Override
+    public int getPriority() {
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
+    }
+
+
+    @Override
         protected String getInsertStatement(InsertOrUpdateStatement insertOrUpdateStatement, Database database, SqlGeneratorChain sqlGeneratorChain) {
             StringBuilder columns = new StringBuilder();
             StringBuilder values = new StringBuilder();
diff --git a/src/main/java/liquibase/ext/databricks/sqlgenerator/RenameColumnGeneratorDatabricks.java b/src/main/java/liquibase/ext/databricks/sqlgenerator/RenameColumnGeneratorDatabricks.java
index a3734255..b57d161c 100644
--- a/src/main/java/liquibase/ext/databricks/sqlgenerator/RenameColumnGeneratorDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/sqlgenerator/RenameColumnGeneratorDatabricks.java
@@ -22,6 +22,12 @@ public boolean supports(RenameColumnStatement statement, Database database) {
         return (database instanceof DatabricksDatabase);
     }
 
+    @Override
+    public int getPriority() {
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
+    }
+
+
     @Override
     public ValidationErrors validate(RenameColumnStatement renameColumnStatement, Database database, SqlGeneratorChain sqlGeneratorChain) {
         ValidationErrors validationErrors = new ValidationErrors();
diff --git a/src/main/java/liquibase/ext/databricks/sqlgenerator/RenameTableGeneratorDatabricks.java b/src/main/java/liquibase/ext/databricks/sqlgenerator/RenameTableGeneratorDatabricks.java
index 16c4d7e9..8e582913 100644
--- a/src/main/java/liquibase/ext/databricks/sqlgenerator/RenameTableGeneratorDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/sqlgenerator/RenameTableGeneratorDatabricks.java
@@ -20,6 +20,12 @@ public boolean supports(RenameTableStatement statement, Database database) {
         return (database instanceof DatabricksDatabase);
     }
 
+    @Override
+    public int getPriority() {
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
+    }
+
+
     @Override
     public ValidationErrors validate(RenameTableStatement renameTableStatement, Database database, SqlGeneratorChain sqlGeneratorChain) {
         ValidationErrors validationErrors = new ValidationErrors();
diff --git a/src/main/java/liquibase/ext/databricks/sqlgenerator/RenameViewGeneratorDatabricks.java b/src/main/java/liquibase/ext/databricks/sqlgenerator/RenameViewGeneratorDatabricks.java
index 3a991a94..86a22587 100644
--- a/src/main/java/liquibase/ext/databricks/sqlgenerator/RenameViewGeneratorDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/sqlgenerator/RenameViewGeneratorDatabricks.java
@@ -20,6 +20,12 @@ public boolean supports(RenameViewStatement statement, Database database) {
         return (database instanceof DatabricksDatabase);
     }
 
+    @Override
+    public int getPriority() {
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
+    }
+
+
     @Override
     public ValidationErrors validate(RenameViewStatement renameViewStatement, Database database, SqlGeneratorChain sqlGeneratorChain) {
         ValidationErrors validationErrors = new ValidationErrors();
diff --git a/src/main/java/liquibase/ext/databricks/sqlgenerator/SetColumnRemarksGeneratorDatabricks.java b/src/main/java/liquibase/ext/databricks/sqlgenerator/SetColumnRemarksGeneratorDatabricks.java
index 77a8302a..10e1b05d 100644
--- a/src/main/java/liquibase/ext/databricks/sqlgenerator/SetColumnRemarksGeneratorDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/sqlgenerator/SetColumnRemarksGeneratorDatabricks.java
@@ -18,16 +18,18 @@
 import liquibase.sqlgenerator.core.SetColumnRemarksGenerator;
 
 public class SetColumnRemarksGeneratorDatabricks extends SetColumnRemarksGenerator {
-    @Override
-    public int getPriority() {
-        return DatabricksDatabase.PRIORITY_DATABASE;
-    }
 
     @Override
     public boolean supports(SetColumnRemarksStatement statement, Database database) {
         return (database instanceof DatabricksDatabase);
     }
 
+    @Override
+    public int getPriority() {
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
+    }
+
+
     @Override
     public ValidationErrors validate(SetColumnRemarksStatement setColumnRemarksStatement, Database database, SqlGeneratorChain sqlGeneratorChain) {
         ValidationErrors validationErrors = new ValidationErrors();
diff --git a/src/main/java/liquibase/ext/databricks/sqlgenerator/SetTableRemarksGeneratorDatabricks.java b/src/main/java/liquibase/ext/databricks/sqlgenerator/SetTableRemarksGeneratorDatabricks.java
index e4c25ce1..3cef538b 100644
--- a/src/main/java/liquibase/ext/databricks/sqlgenerator/SetTableRemarksGeneratorDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/sqlgenerator/SetTableRemarksGeneratorDatabricks.java
@@ -2,7 +2,6 @@
 
 import liquibase.ext.databricks.database.DatabricksDatabase;
 import liquibase.database.Database;
-import liquibase.database.core.*;
 import liquibase.exception.ValidationErrors;
 import liquibase.sql.Sql;
 import liquibase.sql.UnparsedSql;
@@ -20,6 +19,12 @@ public boolean supports(SetTableRemarksStatement statement, Database database) {
         return (database instanceof DatabricksDatabase);
     }
 
+    @Override
+    public int getPriority() {
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
+    }
+
+
     @Override
     public ValidationErrors validate(SetTableRemarksStatement setTableRemarksStatement, Database database, SqlGeneratorChain sqlGeneratorChain) {
         ValidationErrors validationErrors = new ValidationErrors();
diff --git a/src/main/java/liquibase/ext/databricks/sqlgenerator/UpdateGeneratorDatabricks.java b/src/main/java/liquibase/ext/databricks/sqlgenerator/UpdateGeneratorDatabricks.java
index 265a0426..3f30be4d 100644
--- a/src/main/java/liquibase/ext/databricks/sqlgenerator/UpdateGeneratorDatabricks.java
+++ b/src/main/java/liquibase/ext/databricks/sqlgenerator/UpdateGeneratorDatabricks.java
@@ -27,7 +27,7 @@ public boolean supports(UpdateStatement statement, Database database) {
 
     @Override
     public int getPriority() {
-        return DatabricksDatabase.PRIORITY_DATABASE;
+        return DatabricksDatabase.DATABRICKS_PRIORITY_DATABASE;
     }
 
 
diff --git a/src/main/resources/META-INF/services/liquibase.database.DatabaseConnection b/src/main/resources/META-INF/services/liquibase.database.DatabaseConnection
new file mode 100644
index 00000000..181c933d
--- /dev/null
+++ b/src/main/resources/META-INF/services/liquibase.database.DatabaseConnection
@@ -0,0 +1 @@
+liquibase.ext.databricks.database.DatabricksConnection
\ No newline at end of file
diff --git a/src/main/resources/META-INF/services/liquibase.datatype.LiquibaseDataType b/src/main/resources/META-INF/services/liquibase.datatype.LiquibaseDataType
index 88a93c6e..14bb01cc 100644
--- a/src/main/resources/META-INF/services/liquibase.datatype.LiquibaseDataType
+++ b/src/main/resources/META-INF/services/liquibase.datatype.LiquibaseDataType
@@ -5,3 +5,6 @@ liquibase.ext.databricks.datatype.IntegerDatatypeDatabricks
 liquibase.ext.databricks.datatype.BooleanDatatypeDatabricks
 liquibase.ext.databricks.datatype.FloatDatatypeDatabricks
 liquibase.ext.databricks.datatype.DoubleDatatypeDatabricks
+liquibase.ext.databricks.datatype.TinyintDatatypeDatabricks
+liquibase.ext.databricks.datatype.SmallintDatatypeDatabricks
+liquibase.ext.databricks.datatype.BinaryDataTypeDatabricks
\ No newline at end of file
diff --git a/src/main/resources/META-INF/services/liquibase.executor.Executor b/src/main/resources/META-INF/services/liquibase.executor.Executor
new file mode 100644
index 00000000..0b734acf
--- /dev/null
+++ b/src/main/resources/META-INF/services/liquibase.executor.Executor
@@ -0,0 +1 @@
+liquibase.ext.databricks.executor.DatabricksExecutor
\ No newline at end of file