Delta: Support for setting dataChange=false

[marmbrus]

The Delta transaction protocol contains the ability to mark entries in the transaction log as dataChange=false indicating that they are only rearranging data that is already part of the table. This is powerful as it allows you to perform compaction and other read-performance optimizations, without breaking the ability to use a Delta table as a streaming source. We should expose this as a DataFrame writer option for overwrites.

[btbbass]

Thanks, that is exactly what I meant.

I confirm that in the test I did for the compaction of existing data, the transaction log contained the field
'dataChange=true'.

[MrPowers]

@btbbass - after compacting a Delta data lake, should the transaction log contain dataChange=true for the files that are added, the files that are removed, or both?

Here's the current behavior. Suppose we have a Delta data lake with 1,000 files and run this code:

val df = spark
  .read
  .format("delta")
  .load("/some/path/data")

df
  .repartition(10)
  .write
  .format("delta")
  .mode("overwrite")
  .save("/some/path/data")

The _delta_log/00000000000000000001.json file will contain 10 rows like this:

{
  "add":{
    "path":"part-00000-compacted-data-c000.snappy.parquet",
    "partitionValues":{

    },
    "size":123456,
    "modificationTime":1567528453000,
    "dataChange":true
  }
}

The _delta_log/00000000000000000001.json file will also contain 1,000 rows like this:

{
  "remove":{
    "path":"part-00154-some-stuff-c000.snappy.parquet",
    "deletionTimestamp":1567528580934,
    "dataChange":true
  }
}

[btbbass]

I think that the expected result is to have dataChange=false for the 10 new files, as there is no data change, but only reorganisation of underlying data into fewer files.

That is to obtain the effect that:

1. A streaming job that is using the delta table as a data source
2. and have already processed the data in the 1000 files,

-> should not reprocess data that is being reorganised.

[frankobe]

@marmbrus @btbbass Is there a method in the dataframe to infer whether there is data change or not? I am thinking about creating a new mode like repartition which works almost the same as overwrite except for setting dataChange=false so it is up to the caller to specify whether data is change or not.

After the change, the call should look like this:

val df = spark
  .read
  .format("delta")
  .load("/some/path/data")

df
  .repartition(10)
  .write
  .format("delta")
  .mode("repartition")
  .save("/some/path/data")

[marmbrus]

@frankobe we could go that route, but it would require making changes to stable API in Spark, which is unlikely to be approved by the community there.

Rather I'd suggest just adding an option: .option("dataChange", "false") (or some other option name if someone has a better suggestion).

[JassAbidi]

Hi, I want to implement this issue

[tdas]

This commit (https://github.com/delta-io/delta/commit/f32830022e6a664f99f25f3dad8584f5cd9952bf) adds the last piece to need to make sure setting dataChange = false minimizes the conflicts with concurrent operations. So you will be able to do file compactions concurrently when data is being continuously appended.