[Concurrency Control] Allow compaction against blind-appendings when check conflicts

[xianyinxin]

There're several optimizations on the concurrent operations on delta log, like concurrent blind appends. However, it would be useful if we allow no-data-change ops, one typical scenario of which is doing compaction, against blind-appendings (like streaming writing).

This can be seen as a subtask of #72, and may need the finish of #146.

[xianyinxin]

cc @liwensun

[xianyinxin]

cc @liwensun @tdas @marmbrus , could you give some suggestion like if this optimization is feasible? If yes, I'd like to propose a pull request.

[marmbrus]

I'd need to see more design to understand if this is something feasible to implement for Delta. Specifically, rather than understand what operations you want to perform at the same time, I'd like to understand the guarantees that this isolation mode provides.

[xianyinxin]

Thanks for your reply @marmbrus . When I performed compaction while another thread was streaming data into delta, I found sometimes the compaction would fail because of commits conflicts. Since the streaming work was just blindly append small files which has nothing to do with the history data, the compaction could not have to fail because it didn't change the data. In this case, we can allow the compaction work to retry the commit, just like what the 'concurrent appending' does.

Maybe I'd better limit the "no-data-change operations" in title to "compaction".

[rahulsmahadev]

Hi Xianyin, would this( 009c949) be sufficient for your use case ? This commit would allow you to perform a repartition by marking it as no data change operation. Regards, Rahul Mahadev, Software Engineer, Databricks

…

On Sun, Nov 3, 2019 at 10:21 PM Xianyin Xin ***@***.***> wrote: Thanks for your reply @marmbrus <https://github.com/marmbrus> . When I performed compaction while another thread was streaming data into delta, I found sometimes the compaction would fail because of commits conflicts. Since the streaming work was just blindly append small files which has nothing to do with the history data, the compaction could not have to fail because it didn't change the data. In this case, we can allow the compaction work to retry the commit, just like what the 'concurrent appending' does. Maybe I'd better limit the "no-data-change operations" in title to "compaction". — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#228?email_source=notifications&email_token=AMKLYPP7HTNUEWLCUNYDQ6LQR65PNA5CNFSM4JGFHXJKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEC6KSLY#issuecomment-549234991>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AMKLYPOSQVRSYXWN6RUB4XDQR65PNANCNFSM4JGFHXJA> .

[xianyinxin]

Thanks @rahulsmahadev , 009c949 is helpful but is not enough. There's a check in OptimisticTransactionImpl#checkAndRetry:

      // Fail if the data is different than what the txn read.
      if (dependsOnFiles && fileActions.nonEmpty) {
        throw new ConcurrentWriteException(commitInfo)
      }

and I'm afraid it would fail here, because the compaction is "dependsOnFiles" and the winningCommit.fileActions is nonEmpty.

I'll do a test and post the result later.

[cfmcgrady]

I also had the same problem.
consider the following case:
application1. we start a low duration structured streaming application, append the data into delta table, partitioned by date.
application2. we do optimizer(e.g. compaction) for the delta table per day.

notice that the optimizer operation always took a long time to complete, maybe some hours. so when we do the compaction operation commit, the concurrency check will fail because of compaction depends On Files and the fileActions.nonEmpty is true, as @xianyinxin mentioned above.

[marmbrus]

@xianyinxin I don't think you can just get rid of that check though. That would make the results of concurrent merges incorrect. Consider the case where two concurrent merges attempt to modify the same key. They both read the table, and it is not valid for them to proceed if the the data they read has been changed.

I'd be happy to relax things here, but we need a design / proof that the changes are not breaking ACID / serializability.

[xianyinxin]

Thanks @marmbrus for your suggestion.

I don't think you can just get rid of that check though. That would make the results of concurrent merges incorrect. Consider the case where two concurrent merges attempt to modify the same key. They both read the table, and it is not valid for them to proceed if the the data they read has been changed.

I'm not going to get rid of that check, but just relax it in certain cases.

I'd be happy to relax things here, but we need a design / proof that the changes are not breaking ACID / serializability.

My thoughts is, suppose there're an streaming appending work, generate files and files sequencely,

A -> B -> C -> D ->E

At point E, we start an compaction job. The expected result of the job would be

[A -> B -> C -> D ->E] -> F

where F stands for all of the valid data from A to E. However, as @cfmcgrady pointed out, the compaction job may take long time. At the same time, the streaming appending job is still running, and when the compaction job commits, it might hit a conflict

Streaming: A -> B -> C -> D ->E -> F -> G -> H
Compaction: [A -> B -> C -> D ->E]-> F'

But F, G, H are just blind appends, it has nothing to do with A,B,C,D,E, so it would be better if the final result is

F' -> F -> G -> H

where F' stands for [A -> B -> C -> D ->E]. Of course we cannot use F', we need to retry the commit until it success,

Streaming: A -> B -> C -> D ->E -> F -> G -> H
Compaction: [A -> B -> C -> D ->E]-> try F -> try G -> try H -> I

where I stands for [A -> B -> C -> D ->E].

I'm not sure if I miss something, but an intuitive feeling is that it should work.

[cfmcgrady]

Hi, @marmbrus
Here is a full example for replay:

  test("just for replay") {
    withTempDir { tempDir =>
      val log = DeltaLog(spark, new Path(tempDir.getCanonicalPath))
      // Initialize the log. Truncate() is just a no-op placeholder.
      log.startTransaction().commit(Nil, Truncate())

      // mock a streaming application, append the new data to the table
      appendNewData(log, "a")
      appendNewData(log, "b")
      appendNewData(log, "c")
      appendNewData(log, "d")
      appendNewData(log, "e")

      // we start a new application for optimizer the table and doing compaction.
      val compactionTxn = log.startTransaction()

      // notice, the compaction usually takes a long time.
      // and the streaming application append new data to the table.
      appendNewData(log, "f")
      appendNewData(log, "g")
      appendNewData(log, "h")

      // finished compaction, and start tries to commit.
      val compactionFile = AddFile("f'", Map.empty, 1, 1, dataChange = false)
      val fileToRemove = compactionTxn.filterFiles().map(_.remove).map(_.copy(dataChange = false))
      
      compactionTxn.commit(fileToRemove ++ Seq(compactionFile), Truncate())
    }
  }

  def appendNewData(deltaLog: DeltaLog, fileToAdd: String): Unit = {
    val addFile = AddFile(fileToAdd, Map.empty, 1, 1, dataChange = true)
    deltaLog.startTransaction().commit(addFile :: Nil, Truncate())
  }

Unfortunately, the compaction commit will go failed and throw a ConcurrentWriteException. because of the concurrency check mechanism, as @xianyinxiang mentioned above..

[cfmcgrady]

I have a suggestion, we should update the concurrency check condition. In the data rearrange mode, we should check the RemoveFile has a conflict with winning commit's RemoveFile or not.

// Pseudo code

      if (isRearrangeOperation) {
        if (winningCommit.removeFiles.contains(removeFiles)) {
          throw new ConcurrentWriteException(commitInfo)
        }
      } else {
        // Fail if the data is different than what the txn read.
        if (dependsOnFiles && fileActions.nonEmpty) {
          throw new ConcurrentWriteException(commitInfo)
        }
      }

but there is a risk in update case if we set dataChanage = false.

[marmbrus]

I understand the case that does not work, but what I do not understand is the formal properties of the isolation level that you are proposing we implement instead.

Right now it is very simple. The transaction log is strictly-serializable (i.e. the result is the same as though you ran each transaction one at a time, even though they might have actually run concurrently).