Delta: Support custom commit metadata in write / merge

[changpingc]

This is a feature request. I wish I can record additional metadata as part of DataFrameWriter / MERGE transaction.

For example, I have a spark job that runs a deterministic transformation periodically on some source delta table and write to output delta table. I want to track data lineage (source table version) and runtime parameters in each commit, such that I can understand how new versions of data propagate through a pipeline and replay the exact transformation for debugging.

Maybe the API is like:

df.write.format("delta").option("metadata", r'{"source_table_version": "...", "runtime_params": ...}').save(...)

Maybe metadata gets put into commitInfo and we expose it in describe history?

[tdas]

This is definitely a good feature request. This is should be pretty easy to implement. I think the tricky part is figuring out the right APIs for specifying the metadata for non-DataFrameWriter operations like merge/update/delete. However, we can definitely start with DataFrameWriter options. It would awesome if you can build this feature, just make a PR!

[schuermannator]

This is implemented in b7eee5b2fbe6c668329ca55025e0e8edb4529104. It works for the DataFrame API as well as a SparkSession config:

DataFrame

df.write.format("delta")
  .option("userMetadata", "...")
  .save(...)

```scala
df.writeStream.format("delta")
.option("userMetadata", "...")
.start(...)

#### SparkSession config
```scala
spark.conf.set("spark.databricks.delta.commitInfo.userMetadata", "...")
spark.sql("INSERT INTO deltaTable ...")

And retrieve the data by inspecting the CommitInfo via DESCRIBE HISTORY / deltaTable.history()