DAT-19308 - feat(databricks): ensure essential table properties Delta.*

pom.xml

@@ -10,7 +10,7 @@
 
     <groupId>org.liquibase.ext</groupId>
     <artifactId>liquibase-databricks</artifactId>
-    <version>1.5.0-SNAPSHOT</version>
+    <version>1.4.1-SNAPSHOT</version>
 
     <name>Liquibase Extension: Databricks support</name>
     <description>Liquibase Extension for Databricks.</description>
@@ -66,7 +66,7 @@
         <liquibase-test-harness.version>1.0.10</liquibase-test-harness.version>
         <junit.version>5.11.4</junit.version>
         <junit-platform.version>1.11.4</junit-platform.version>
-        <mockito.version>5.15.2</mockito.version>
+        <mockito.version>4.11.0</mockito.version>
         <jacoco.fileset.directory>target</jacoco.fileset.directory>
     </properties>
 

src/main/java/liquibase/ext/databricks/diff/output/changelog/ChangedTblPropertiesUtil.java

@@ -15,6 +15,7 @@
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -26,6 +27,15 @@ public class ChangedTblPropertiesUtil {
 
     private static final String SPLIT_ON_COMMAS = ",(?=(?:[^\"]*\"[^\"]*\")*[^\"$])";
     private static final String SPLIT_ON_EQUALS = "=(?=(?:[^\"]*\"[^\"]*\")*[^\"$])";
+    private static final Set<String> ALLOWED_DELTA_PROPERTIES = Stream.of(
+            "delta.columnMapping.mode",
+            "delta.enableDeletionVectors",
+            "delta.feature.allowColumnDefaults",
+            "delta.logRetentionDuration",
+            "delta.deletedFileRetentionDuration",
+            "delta.targetFileSize",
+            "delta.enableChangeDataFeed"
+    ).collect(Collectors.toSet());
 
     private ChangedTblPropertiesUtil() {
     }
@@ -60,6 +70,7 @@ static AbstractAlterPropertiesChangeDatabricks[] getAbstractTablePropertiesChang
                 addPropertiesMap.put(key, value);
             }
         });
+
         //then we remove the properties that are not in the reference
         Map<String, String> removePropertiesMap = comparedValuesMap.entrySet().stream()
                 .filter(entry -> !referencedValuesMap.containsKey(entry.getKey()))
@@ -94,11 +105,14 @@ static AbstractAlterPropertiesChangeDatabricks[] getAbstractTablePropertiesChang
      */
     private static Map<String, String> convertToMapExcludingDeltaParameters(Object referenceValueObject) {
         String referenceValue = referenceValueObject == null ? "" : referenceValueObject.toString();
-         return Arrays.stream(referenceValue.split(SPLIT_ON_COMMAS))
+        return Arrays.stream(referenceValue.split(SPLIT_ON_COMMAS))
                 .map(s -> s.split(SPLIT_ON_EQUALS))
                 .filter(a -> a.length > 1)
-                 .map(a -> new String[]{a[0].trim(), a[1].trim()})
-                .filter(a -> !a[0].replace("'", "").matches("^delta.+"))
+                .map(a -> new String[]{a[0].trim(), a[1].trim()})
+                .filter(a -> {
+                    String propertyName = a[0].replace("'", "");
+                    return !propertyName.startsWith("delta.") || ALLOWED_DELTA_PROPERTIES.contains(propertyName);
+                })
                 .collect(Collectors.toMap(a -> a[0], a -> a[1]));
     }
 

src/main/java/liquibase/ext/databricks/parser/NamespaceDetailsDatabricks.java

@@ -12,6 +12,7 @@
  */
 public class NamespaceDetailsDatabricks implements NamespaceDetails {
 
+    //TODO use it in all classes where this is hardcoded
     public static final String DATABRICKS_NAMESPACE = "http://www.liquibase.org/xml/ns/databricks";
 
     public static final String DATABRICKS_XSD = "http://www.liquibase.org/xml/ns/databricks/liquibase-databricks-latest.xsd";

src/main/java/liquibase/ext/databricks/sqlgenerator/CreateTableGeneratorDatabricks.java

@@ -1,22 +1,36 @@
 package liquibase.ext.databricks.sqlgenerator;
 
 
+import liquibase.database.Database;
 import liquibase.exception.ValidationErrors;
 import liquibase.ext.databricks.change.createTable.CreateTableStatementDatabricks;
 import liquibase.ext.databricks.database.DatabricksDatabase;
-import liquibase.sqlgenerator.core.CreateTableGenerator;
-import liquibase.database.Database;
 import liquibase.sql.Sql;
 import liquibase.sql.UnparsedSql;
 import liquibase.sqlgenerator.SqlGeneratorChain;
+import liquibase.sqlgenerator.core.CreateTableGenerator;
 import liquibase.statement.core.CreateTableStatement;
 import liquibase.structure.DatabaseObject;
 import org.apache.commons.lang3.StringUtils;
 
-import java.util.List;
+import java.util.*;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 public class CreateTableGeneratorDatabricks extends CreateTableGenerator {
 
+    private static final String[] PROPERTY_ORDER = {
+            "'delta.feature.allowColumnDefaults'",
+            "'delta.columnMapping.mode'",
+            "'delta.enableDeletionVectors'"
+    };
+
+    private static final Map<String, String> DEFAULT_VALUES = Stream.of(
+            new AbstractMap.SimpleEntry<>("'delta.feature.allowColumnDefaults'", "'supported'"),
+            new AbstractMap.SimpleEntry<>("'delta.columnMapping.mode'", "'name'"),
+            new AbstractMap.SimpleEntry<>("'delta.enableDeletionVectors'", "true")
+    ).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
     @Override
     public int getPriority() {
         return PRIORITY_DATABASE;
@@ -29,16 +43,55 @@ public boolean supports(CreateTableStatement statement, Database database) {
 
     public ValidationErrors validate(CreateTableStatementDatabricks createStatement, Database database, SqlGeneratorChain sqlGeneratorChain) {
         ValidationErrors validationErrors = new ValidationErrors();
-        if (!(createStatement.getPartitionColumns().isEmpty()) && !(createStatement.getClusterColumns().isEmpty())){
+        if (!(createStatement.getPartitionColumns().isEmpty()) && !(createStatement.getClusterColumns().isEmpty())) {
             validationErrors.addError("WARNING! Databricks does not supported creating tables with PARTITION and CLUSTER columns, please one supply one option.");
         }
         return validationErrors;
     }
 
+    private String mergeTableProperties(String customProperties) {
+        Map<String, String> properties = new LinkedHashMap<>(DEFAULT_VALUES);
+
+        // If there are custom properties, parse and add them
+        if (StringUtils.isNotEmpty(customProperties)) {
+            Arrays.stream(customProperties.split(","))
+                    .map(String::trim)
+                    .filter(prop -> !prop.isEmpty())
+                    .forEach(prop -> {
+                        String[] parts = prop.split("=", 2);
+                        if (parts.length == 2) {
+                            properties.put(parts[0].trim(), parts[1].trim());
+                        }
+                    });
+        }
+
+        StringBuilder result = new StringBuilder();
+
+        for (String key : PROPERTY_ORDER) {
+            if (properties.containsKey(key)) {
+                if (result.length() > 0) {
+                    result.append(", ");
+                }
+                result.append(key).append(" = ").append(properties.get(key));
+                properties.remove(key);
+            }
+        }
+
+        // Then add any remaining custom properties
+        for (Map.Entry<String, String> entry : properties.entrySet()) {
+            if (entry.getValue() != null && !entry.getValue().equals("null")) {
+                if (result.length() > 0) {
+                    result.append(", ");
+                }
+                result.append(entry.getKey()).append(" = ").append(entry.getValue());
+            }
+        }
+
+        return result.toString();
+    }
 
     @Override
     public Sql[] generateSql(CreateTableStatement statement, Database database, SqlGeneratorChain sqlGeneratorChain) {
-
         Sql[] sqls = super.generateSql(statement, database, sqlGeneratorChain);
         StringBuilder finalsql = new StringBuilder(sqls[0].toSql());
 
@@ -50,13 +103,13 @@ public Sql[] generateSql(CreateTableStatement statement, Database database, SqlG
             } else {
                 finalsql.append(" USING delta");
             }
-            if (thisStatement.getExtendedTableProperties() != null && StringUtils.isNotEmpty(thisStatement.getExtendedTableProperties().getTblProperties())) {
-                finalsql.append(" TBLPROPERTIES (").append(thisStatement.getExtendedTableProperties().getTblProperties()).append(")");
-            } else {
-                finalsql.append(" TBLPROPERTIES('delta.feature.allowColumnDefaults' = 'supported', 'delta.columnMapping.mode' = 'name', 'delta.enableDeletionVectors' = true)");
+
+            String properties = null;
+            if (thisStatement.getExtendedTableProperties() != null) {
+                properties = thisStatement.getExtendedTableProperties().getTblProperties();
             }
+            finalsql.append(" TBLPROPERTIES(").append(mergeTableProperties(properties)).append(")");
 
-            // Databricks can decide to have tables live in a particular location. If null, Databricks will handle the location automatically in DBFS
             if (!StringUtils.isEmpty(thisStatement.getTableLocation())) {
                 finalsql.append(" LOCATION '").append(thisStatement.getTableLocation()).append("'");
             } else if (thisStatement.getExtendedTableProperties() != null && StringUtils.isNotEmpty(thisStatement.getExtendedTableProperties().getTableLocation())) {
@@ -66,49 +119,34 @@ public Sql[] generateSql(CreateTableStatement statement, Database database, SqlG
             List<String> clusterCols = thisStatement.getClusterColumns();
             List<String> partitionCols = thisStatement.getPartitionColumns();
 
-
-            // If there are any cluster columns, add the clause
-            // ONLY if there are NOT cluster columns, then do partitions, but never both.
             if (!clusterCols.isEmpty()) {
-
                 finalsql.append(" CLUSTER BY (");
-
                 int val = 0;
                 while (clusterCols.size() > val) {
                     finalsql.append(clusterCols.get(val));
-
-                    val +=1;
+                    val += 1;
                     if (clusterCols.size() > val) {
                         finalsql.append(", ");
-                    }
-                    else {
+                    } else {
                         finalsql.append(")");
                     }
                 }
             } else if (!partitionCols.isEmpty()) {
                 finalsql.append(" PARTITIONED BY (");
-
                 int val = 0;
                 while (partitionCols.size() > val) {
                     finalsql.append(partitionCols.get(val));
-
-                    val +=1;
+                    val += 1;
                     if (partitionCols.size() > val) {
                         finalsql.append(", ");
-                    }
-                    else {
+                    } else {
                         finalsql.append(")");
                     }
                 }
             }
-
-
         }
 
         sqls[0] = new UnparsedSql(finalsql.toString(), sqls[0].getAffectedDatabaseObjects().toArray(new DatabaseObject[0]));
-
         return sqls;
-
     }
-
-}
+}

src/test/java/liquibase/ext/databricks/AdvancedExtensionHarnessTestSuite.java

@@ -1,7 +1,6 @@
 package liquibase.ext.databricks;
 
 import liquibase.harness.AdvancedHarnessSuite;
-import liquibase.harness.generateChangelog.GenerateChangelogTest;
 import liquibase.harness.snapshot.SnapshotObjectTests;
 import org.junit.platform.suite.api.SelectClasses;