[Concurrency Control] Support concurrent writes

[liwensun]

As of 0.2.0, Delta Lake supports concurrent appends, where an append must be only adding new data to the table without reading or modifying existing data in any way.

This issue is to track the support of more types of concurrent writes in Delta Lake, where a writer can also read and modify existing data, such as overwrite, delete or update.

Related issues:
#9: Allow concurrent writes to partitions that don't interact with each other
#23: Multi threading support on delta table

[manuzhang]

@liwensun By concurrent append, do you mean the following is possible ?

object ConcurrencyTest {

  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder.master("local").getOrCreate()

    for (_ <- 1 to 10) {
      val thread = new AppendThread(spark)
      thread.start()
    }
  }

  class AppendThread(spark: SparkSession) extends Thread {

    override def run(): Unit = {
      val data = spark.range(0, 5)
      data.write.format("delta").mode("append").save("/tmp/delta-test")
    }
  }

}

I've seen ProtocolChangeException when running the above codes

Exception in thread "Thread-23" org.apache.spark.sql.delta.ProtocolChangedException: The protocol version of the Delta table has been changed by a concurrent update. Please try the operation again.
Conflicting commit: {"version":0,"timestamp":1561364064793,"operation":"WRITE","operationParameters":{"mode":Append,"partitionBy":[]},"isBlindAppend":true}

Sorry for the noise if I'm mistaken about the concept here.

[liwensun]

Hey @manuzhang, Thanks for trying out concurrent appends. I can understand that this could be a bit confusing here. You'll need to create the table first then do concurrent appends - otherwise each append operation will involve table creation operations, such as setting up the protocol version and table metadata, which we don't allow concurrently (hence the ProtocolChangeException).

[batesl87]

Hi @liwensun, is it enough to successfully append data into a folder once and then you are free to concurrently append or do you need to explicitly create a delta table?

The fact that you can create delta tables by writing into a mounted folder implies this would be fine but the databricks docs states that current appends only works within a workspace (https://docs.databricks.com/delta/delta-faq.html#what-are-the-limitations-of-multi-cluster-writes). Does this mean Delta is storing state outside the mounted storage? If so what commands initialise this shared state?

[liwensun]

@batesl87 If you are using df.format("delta").mode("append").save(...), then you effectively create a delta table, and yes, you can do concurrent appends, but from a single cluster only if you are using Delta Lake OSS!

The doc you are looking is for managed delta lake offered by Databricks, not for OSS, and in that case, you still don't need to do anything extra as the "shared state" you talk about happens behind the scenes.

[zsxwing]

Resolved by f328300