Delta: Merge Delete making columns null which are not matching the condition/criteria

[junmuz]

Hi,

Recently, I have upgraded to Java 11, Apache Spark 3.0 and Delta Lake 0.7.0. However, I am seeing one strange issue with merge deletes as it is making the columns null which are not matching the conditional criteria. I am doing merge delete as follows.

dT.as("targetData")
                .merge(
                        delFrame.as("deleteData"),
                        condition)
                .whenMatched()
                .delete()
                .execute();

And I have initialised spark context as follows.

SparkSession session = SparkSession.builder().appName("App Name").master("local[*]")
                .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
                .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
                .config("spark.sql.shuffle.partitions", "4")
                .config("spark.databricks.delta.retentionDurationCheck.enabled",false)
                .getOrCreate();

This used to work with 0.6.0 version. Is there any reported bugs/issues with the newer 0.7.0 version?

[junmuz]

I have created out a sample project and it incorrectly sets out 2 columns that doesn't match as null. I have used Java 11, Spark 3.0.0 and Data Tables 0.7.0 in my project.

import io.delta.tables.DeltaTable;
import org.apache.spark.sql.*;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;


import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class SampleClass {

    public static void main(String[] args) {
        SparkSession session = SparkSession.builder().appName("App Name").master("local[*]")
                .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
                .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
                .config("spark.sql.shuffle.partitions", "4")
                .config("spark.databricks.delta.retentionDurationCheck.enabled",false)
                .getOrCreate();

        StructType schema =
                DataTypes.createStructType(new StructField[] {
                        DataTypes.createStructField("id", DataTypes.LongType,false),
                        DataTypes.createStructField("exp" ,DataTypes.LongType,true),
                        DataTypes.createStructField("key", DataTypes.LongType,false),
                        DataTypes.createStructField("val", DataTypes.StringType,true),
                        DataTypes.createStructField("marker", DataTypes.LongType,false)});


        String location = "/tmp/abc/xyz";
        if (!DeltaTable.isDeltaTable(location)) {
            Dataset<Row> df = session.createDataFrame(new ArrayList<Row>(), schema);
            df.write().format("Delta").save(location);
        }

        DeltaTable deltaTable = DeltaTable.forPath(location);
        deltaTable.delete();

        Long exp = System.currentTimeMillis() / 1000;
        List<Row> data = new ArrayList<Row>();

        long id = 601;
        data.add(RowFactory.create(++id, exp, 1138L, "ref1", 122L));
        data.add(RowFactory.create(++id, exp, 1138L, "ref1", 122L));
        data.add(RowFactory.create(++id, exp, 1138L, "ref1", 122L));
        data.add(RowFactory.create(++id, exp, 1138L, "ref1", 125L));
        data.add(RowFactory.create(++id, exp, 1140L, "ref1", 122L));
        data.add(RowFactory.create(++id, exp, 1138L, "ref1", 122L));
        data.add(RowFactory.create(++id, exp, 1139L, "ref1", 122L));
        Dataset<Row> df1 = session.createDataFrame(data, schema);

        Map<String, Column> updateMap = new HashMap<String, Column>();
        updateMap.put("id", functions.col(String.format("newData.%s","id")));

        String cond = String.format("targetData.%s = newData.%s","id","id");
        deltaTable.as("targetData")
                .merge(
                        df1.as("newData"),
                        cond)
                .whenMatched()
                .update(updateMap)
                .whenNotMatched()
                .insertAll()
                .execute();

        Dataset<Row> deldf = deltaTable.as("targetData").toDF().filter(functions.col("id").lt(605)).dropDuplicates("val",
                "key", "marker");

        Column condition = functions.col("targetData.val")
                .equalTo(functions.col("deleteData.val"))
                .and(functions.col("targetData.key")
                        .equalTo(functions.col("deleteData.key")))
                .and(functions.col("targetData.marker")
                        .equalTo(functions.col("deleteData.marker")));

        deltaTable.as("targetData")
                .merge( deldf.as("deleteData"),
                        condition)
                .whenMatched()
                .delete()
                .execute();

        deltaTable.toDF().show();
    }
}

On showing the data frame, following is the output observed.

+----+----------+----+----+------+
|  id|       exp| key| val|marker|
+----+----------+----+----+------+
| 606|1596458944|1140|ref1|   122|
|null|      null|null|null|  null|
|null|      null|null|null|  null|
+----+----------+----+----+------+

[tdas]

can you try this with Java 8? I wonder if this is a java 11 + spark 3.0 issue.

[tdas]

Nvm, I could reproduce it. There is a bug somewhere but I am yet to narrow it down. The solution is to always create a new DeltaTable object right before calling merge. For example, I think if convert the last merge from deltaTable.as("targetData") to DeltaTable.forPath(location).as("targetData"), then this issue would go away.

I am guessing 0.5.0 and 0.6.1 will also have this same issue. In 0.6.0 we had tried to fix this sort of corner cases, but broke other workloads, so we had reverted it in 0.6.1. So it was accidental that your code was working in 0.6.0.

[junmuz]

@tdas Thanks, yes that worked fine for me. I think what you suggested is a temporary hack as there is an associated cost of opening an fd/socket each time.

[tdas]

I dont think it should have that cost because the internal DeltaLog object that contains all the table information is reused via a static object cache. It's just that the user-facing DeltaTable wrapper object needs to be recreated.

Also, I have debugged the issue. The high-level bit is that this issue happens only in Scala and Python, only when the source DF is effectively created from the same DeltaTable object/instance (using deltaTable.toDF) as that being used for merge. We are prioritizing a fix for this as this can cause data corruption in the table.

[tdas]

if any one encounters this error, I encouraging trying this with 0.8.0.