support adding extra commit metadata with SQL in Spark

[CodingCat]

this PR implements the functionality for users to add extra commit metadata when operating tables with SQL. It also allows users to use multi threading to commit data to tables while keeping metadata thread local

new usage:

(0 until 10).foreach { _ =>


    new Thread() {
      override def run() {
        CommitMetadata.withCommitProperties(Map("metadata-key" -> "thread-local-metadata-value").asJava,
          () => {
            SparkSession.getActiveSession.get.sql("INSERT INTO target VALUES (3, 'c'), (4, 'd')");
          })
      }
    }.start()

  }
}

[CodingCat]

Hi, @kbendick @rdblue I just made this PR as a followup of #4795, please help to review and thanks in advance! once we agree on the approach here, I will add changes to other versions of Spark

[rdblue]

Can you use the helper in ThreadPools and also wrap it in a try/finally to close the threadpool?

I'm also not entirely sure this requires a threadpool to test. I think it would be fine to test a single write in the current thread.

[kbendick]

I would also prefer a test that uses a single write in the current thread without any additional threading business. I worry that CommitMetadata doesn't seem thread local to users.

And then if a multi-threaded test is needed, using the helpers from ThreadPools as suggested.

[CodingCat]

updated to ThreadPool, I think multi-threading testing is still necessary? as we need to have something guarding that the commit metadata change is thread safe no matter we use ThreadLocal as now or later we change to something else for any reason

[rdblue]

I don't think multi-threaded testing is needed. It's enough to know that we're using a thread-local. This also is not guaranteed to run the way this test assumes that it will. There is not a guarantee that the thread pool will scale all the way up, and there's no guarantee that the tasks will each run in a separate thread. I think it's likely that those will happen, but this could still be a source of flakiness later on. Also, this doesn't necessarily test that the thread-local is working properly because there's no guarantee of concurrency across tasks.

While it's probably working the way you expect, there's no guarantee that it must. So I'd prefer to keep the test simple.

[CodingCat]

updated

[kbendick]

Does this class need to be public? Could it be made package-private?

I have concerns around the usage of ThreadLocal for things that most cases don't need to be thread local. I don't want to give users too much room to hurt themselves because they don't consider that CommitMetadata is only threadlocal and then their writes not working properly in the common case of writes without user-side multithreading (e.g. it gets set in one thread somewhere, but another thread is used for commit).

EDIT - Since this takes a Callable, it's less of a concern. I would still name it in a way that's a bit more reflective of the thread local nature (especially if we wanted a CommitMetadata class one day that doens't require a callable and is persistent). That and I always prefer things be package-private if possible.

[CodingCat]

sure, changed to CallerWithCommitMetadata....but...eh...not sure if it is better or worse....

[rdblue]

Does this class need to be public? Could it be made package-private?

Yes, this does need to be public because it is a way for Iceberg users to pass metadata.

[kbendick]

Thinking this over more, there's really no way to achieve this use-case without this. So outside of the style nits Ryan mentioned, this looks pretty much good to me.

Thanks @CodingCat!

[CodingCat]

thanks! @rdblue and @kbendick , just updated the PR

[CodingCat]

just to confirm, seems we don't support SQL based table insert/merge in Spark 2.4 at all right? (so this functionality is not relevant there)

[rdblue]

What was wrong with the CommitMetadata name? I think we should have a simpler name than CallerWithCommitMetadata. That's a bit too confusing. I think the original name was good.

[CodingCat]

What was wrong with the CommitMetadata name? I think we should have a simpler name than CallerWithCommitMetadata. That's a bit too confusing. I think the original name was good.

#4956 (comment) @kbendick mentioned some potential conflict with SS's CommitMetadata here, I don't have strong opinion on this

[singhpk234]

[minor] is this required as we throw in the line above ?

[rdblue]

I don't think the compiler sees that castAndThrow will always throw, so it needs this to know what to do.

[singhpk234]

[doubt](Probably not in scope of PR) Should we add a validation that the keys passed from here as well as extraSnapshotMetadata doesn't override the keys already present in the snasphot-summary ? Or this functionality is intended to do

[rdblue]

I don't think that we need to worry about this. It is unlikely to conflict and if it does conflict, it's up to the caller to decide what to do.

[singhpk234]

[question] Should we also add this to SparkPositionDeltaWrite

iceberg/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java

Line 250 in b06a89c

extraSnapshotMetadata.forEach(operation::set);

since starting 3.2 iceberg support MOR with pos deletes

[rdblue]

Yes, that's a good idea.

[rdblue]

@CodingCat, I think renaming the class back is about the only thing left to fix.

[kbendick]

What was wrong with the CommitMetadata name? I think we should have a simpler name than CallerWithCommitMetadata. That's a bit too confusing. I think the original name was good.

#4956 (comment) @kbendick mentioned some potential conflict with SS's CommitMetadata here, I don't have strong opinion on this

I’m good with the original name too (and prefer it to the new one).

[CodingCat]

@kbendick @rdblue @singhpk234 updated the PR accordingly, thanks!

[rdblue]

Thanks, @CodingCat!