morningman opened a new pull request, #69355:
URL: https://github.com/apache/doris/pull/69355
### What problem does this PR solve?
Issue Number: None
Related PR: #67087
Problem Summary:
**In short.** This ports branch-4.1 #67087 to master. The normal and the
spill Iceberg table sinks stop being async writer sinks: they own the Iceberg
table writer and open, write and close it on the pipeline task, scheduled on
the blocking scheduler, the same way the Paimon sink already works on master.
Master's per-partition memory reservation and spill revocation stay as they are.
**Background.** On master both Iceberg data file sinks
(`IcebergTableSinkLocalState`, and `SpillIcebergTableSinkLocalState` for tables
with a sort order) are `AsyncWriterSink`s. Every block is moved into a queue,
and a thread taken from the fragment manager's thread pool opens the files,
writes every block and closes the writer for the whole statement, while the
pipeline task only enqueues. To keep memory admission correct with that
hand-off, master moves each pipeline reservation into the queue together with
its block, holds the EOS reservation in the writer thread until `close()`
finishes, and makes the spill sink admit a block only after the previous one
has been fully processed, because the reservation it computes samples sorter
state that the writer thread updates.
**The problem, and what it cost.**
| | Master today | After this PR |
|---|---|---|
| Who writes the files | A fragment-pool thread per sink instance, for the
whole statement | The pipeline task, on the blocking scheduler |
| Blocks in flight per instance | Up to 3 blocks parked in the async queue |
None, each block is written in `sink()` |
| Spill sink pipelining | None: each block waits until the writer thread has
processed the previous one | Same work, without the thread hop |
| Memory reservation | Transferred through the queue; the EOS reservation is
held by the writer thread through `finish()`/`close()` | The EOS `sink()` call
itself closes the writer under its reservation |
| Final spill merge output block | `batch_size()` rows, which for wide rows
is far larger than the spill buffer the merge reservation charges for it | At
most one spill batch of rows (≈ spill buffer size) |
**How this PR fixes it.**
- `VIcebergTableWriter` no longer derives from `AsyncResultWriter`; it keeps
the output expressions and operator profile itself. `open`, `write`, `close`
and `finish_deferred_file_cleanup` are virtual so unit tests can replace the
writer.
- `IcebergTableSinkLocalState` and `SpillIcebergTableSinkLocalState` derive
from `PipelineXSinkLocalState<FakeSharedState>`, are blockable, create the
writer in `init()`, clone the output expressions and open it in `open()`, and
write each non-empty block in `sink()`. Their `close()` folds a cancellation
seen before or during the writer close into the final status, and calls
`finish_deferred_file_cleanup()` with that status, so a statement cancelled
after the files were closed deletes them instead of handing them to the final
report.
- The spill sink closes the writer in the EOS `sink()` call (idempotently
with `close()`), so the final sort and spill merge run under the EOS
reservation. That reservation already includes the merge workspace of the
largest partition (`iceberg_spill_merge_workspace`); it now also covers the
work that runs inside `sink()`. Master's per-partition reservation, cold-writer
reservation and largest-partition-first revocation are kept. A mutex guards the
writer pointer so a memory callback that arrives after `close()` sees the
released writer.
- The final spill merge of `VIcebergSortWriter` emits `min(batch_size, spill
batch rows)` rows per block, matching the per-block size the merge reservation
assumes.
- `AsyncResultWriter::wait_for_processing_before_next_sink()` and the
waiting mode of the async queue admission only existed for the Iceberg spill
sink and are removed. The other async writers (Hive, MaxCompute, JDBC, TVF,
OLAP, result file) always used the plain bounded queue, whose behavior does not
change.
- DELETE and MERGE sinks stay asynchronous. The MERGE sink keeps using
`VIcebergTableWriter` as its nested data file writer.
**Results.**
- Unit tests: 122 tests pass across the new `IcebergTableSinkOperatorTest`
(13 cases), `SpillIcebergTableSinkOperatorTest`, `FullSorterTest`, the Iceberg
table and sort writer suites, the MERGE / DELETE sink suites and the async
writer suites. Four mutations (the final merge batch back at `batch_size()`
rows, no writer close in the spill sink's EOS `sink()`, and two of the
cancellation checks around the writer close removed) make 7 of the 13 new cases
fail; the other 6 still pass.
- Regression test: the new `test_iceberg_write_sorted_spill` writes 60000
rows into a REST catalog table sorted by `id desc nulls last` and partitioned
by `region`, with forced spilling and `spill_min_revocable_mem = 1`. On a local
cluster the BE paused the INSERT three times to revoke
`SPILL_ICEBERG_TABLE_SINK_OPERATOR`; each partition was still committed as one
data file of 20000 rows, and Spark reads back 60000 rows, the expected sum, and
no row out of order inside any file (`lag(id) over (partition by _file order by
_pos)`). All 43 suites of `external_table_p0/iceberg/write` pass on the same
cluster.
- Manual test: on a local `file://` Iceberg catalog the same forced-spill
INSERT wrote 12 spill files (4 per partition) before the final merge, every
data file read back in descending `id` order, and UPDATE / DELETE / MERGE on
the table returned the expected rows.
<details>
<summary>Classes and call graph</summary>
```
PipelineTask::execute (blocking scheduler, sink is_blockable() == true)
├─ IcebergTableSinkOperatorX::sink_impl
│ └─ IcebergTableSinkLocalState::sink ──> VIcebergTableWriter::write
├─ SpillIcebergTableSinkOperatorX::get_reserve_mem_size (master model,
unchanged)
├─ SpillIcebergTableSinkOperatorX::sink_impl
│ └─ SpillIcebergTableSinkLocalState::sink
│ ├─ VIcebergTableWriter::write ──> VIcebergSortWriter::write (per
partition)
│ └─ eos: _close_writer ──> VIcebergTableWriter::close ──>
VIcebergSortWriter::close
│ (final sort / spill merge, capped
output batch)
└─ PipelineTask::close ──> *LocalState::close
└─ cancellation check ─ writer close (if not yet) ─ cancellation
check
─ VIcebergTableWriter::finish_deferred_file_cleanup(final status)
RevokableTask (while the task is paused)
└─ SpillIcebergTableSinkLocalState::revoke_memory
└─ under _writer_mutex: active_writers() snapshot ──> largest
VIcebergSortWriter::trigger_spill
```
</details>
Original author: @hubgeter
### Release note
Iceberg data file writes (INSERT, INSERT OVERWRITE, CTAS) run synchronously
on the blocking pipeline scheduler instead of a separate async writer thread.
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [x] Regression test
- [x] Unit Test
- [x] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [x] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [x] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR should
merge into -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
