Description
The delta_sink connector serializes all writes through a single shared lock for which all partitions compete. DeltaSink holds one Mutex<Option<SinkState>> wrapping the Delta table handle, writer, and coercion tree. consume() acquires that lock and holds it across the actual write-to-object-store and commit-to-Delta-log calls (writer.write() and writer.flush_and_commit()). This limits the throughput of the sink, especially if the object store responds slowly.
Relevant code: core/connectors/sinks/delta_sink/src/sink.rs::consume (see the TODO comment above the lock acquisition).
Ref discussion: #2889
Another TODO in this algorithm is the implementation of retries for the parquet buffer writes. Since they are technically moving into a different place with this code, the retries have to be implemented here too.
Affected area / component
Connectors
Proposed solution
This is still very much TBD, I am currently researching what could be the best solution. Once I am done with research, I will write my idea down as final and call up the contributors for a discussion.
Introduce a shared map keyed by partition ID where each partition's consume() call writes into its own buffer independently and concurrently. A separate piece of coordination logic could then periodically (time-based and/or write-count-based trigger — exact policy still TBD) combine buffers and flush/commit them to the Delta table. consume() would still only return once its partition's data has actually been durably committed — buffering into the shared map does not mean early-return/fire-and-forget; the caller blocks until its contribution is confirmed flushed, to preserve delivery guarantees.
Alternatives considered
Currently researching alternatives.
Contribution
Good first issue
Description
The
delta_sinkconnector serializes all writes through a single shared lock for which all partitions compete.DeltaSinkholds oneMutex<Option<SinkState>>wrapping the Delta table handle, writer, and coercion tree.consume()acquires that lock and holds it across the actual write-to-object-store and commit-to-Delta-log calls (writer.write()andwriter.flush_and_commit()). This limits the throughput of the sink, especially if the object store responds slowly.Relevant code:
core/connectors/sinks/delta_sink/src/sink.rs::consume(see theTODOcomment above the lock acquisition).Ref discussion: #2889
Another TODO in this algorithm is the implementation of retries for the parquet buffer writes. Since they are technically moving into a different place with this code, the retries have to be implemented here too.
Affected area / component
Connectors
Proposed solution
This is still very much TBD, I am currently researching what could be the best solution. Once I am done with research, I will write my idea down as final and call up the contributors for a discussion.
Introduce a shared map keyed by partition ID where each partition's
consume()call writes into its own buffer independently and concurrently. A separate piece of coordination logic could then periodically (time-based and/or write-count-based trigger — exact policy still TBD) combine buffers and flush/commit them to the Delta table.consume()would still only return once its partition's data has actually been durably committed — buffering into the shared map does not mean early-return/fire-and-forget; the caller blocks until its contribution is confirmed flushed, to preserve delivery guarantees.Alternatives considered
Currently researching alternatives.
Contribution
Good first issue