CalvinKirs opened a new pull request, #69133:
URL: https://github.com/apache/doris/pull/69133
### What problem does this PR solve?
Issue Number: None
Problem Summary:
Hive batch scans currently materialize all splits within a partition before
publishing them. A `LIMIT` query can therefore wait for work it will never
consume. The consumer can also create an unbounded queue before the producer
applies its capacity, and a fetch can return more splits than the BE requested.
This change makes split generation incremental, bounds queued work, and
preserves each split across producer and consumer batch boundaries. It also
reduces repeated FE variable export and Hive metadata work on the query
planning path.
#### Implementation
- Generate Hive splits lazily across partitions, files, and file offsets.
Pass the producer batch hint through a default connector SPI overload, and
reuse the existing split boundary and format rules. ACID and TABLESAMPLE
special paths retain their existing synchronous behavior.
- Add positive session settings `split_batch_size=128` and
`split_queue_capacity=32`. Both queue creation paths use the same bounded
factory. Capacity is measured in batches per BE: at most 4,096 queued splits
with the defaults, excluding producer batches, consumer residuals, and BE
in-flight work.
- Honor the existing BE fetch batch size, retaining the unused portion of a
producer batch for the next request. Preserve residuals at EOF and propagate
errors/interruption.
- Yield the shared scheduling worker when a BE queue is full, then resume
after consumption releases capacity. Preserve assignment and partial submission
progress so successful backend submissions are not replayed. Serialize source
advancement and cleanup; handle cancellation and submission rejection.
- Bound the number of active partition-batch sources per scan by the
existing scheduler parallelism. Each source submits its current ranges before
claiming another partition batch. This retains partition planning parallelism
without initializing every materialized partition list while queues are full.
- Precompute visible variable descriptors while reading live session values
on each export. Reuse successful Hive table metadata within a statement and the
existing positive HMS table cache for existence checks. Misses/errors are not
memoized, fresh reads still reach HMS, and existing cache invalidation/TTL
behavior is retained.
The connector plugin API advances from 14.0 on current master to 15.0 with
the recorded surface updated. External connector plugins need rebuilding
against the new API. There is no BE protocol or storage-format change.
#### Local performance measurements
The measurements below identify their exact snapshots. Performance was
measured before the final scheduling correctness fixes and upstream
synchronization; these are not measurements of the final PR head.
Fixture: a Hive nonpartitioned table with 10,000 Parquet files, 100 rows per
file, 1,000,000 rows total. A partitioned fixture has the same file count in
one selected partition. All timings are milliseconds, measured by client
`execute` plus full result fetch on a persistent connection. Queries execute
serially on a shared host without CPU isolation.
```sql
EXPLAIN SELECT * FROM split_hms.split_repro.files_10000 LIMIT 10;
SELECT * FROM split_hms.split_repro.files_10000 LIMIT 10;
```
**1. Controlled split-streaming comparison against the starting master**
Starting master: `84e5e6cd52b`; streaming-stage snapshot: `7be63ee0417`.
Both use the same Doris `4.1.3-rc02-7126cf65d96` BE and the same settings
within each pair. Each SQL has 10 warmups and 100 measured samples; this stage
used its original default cache settings. The later variable/HMS optimizations
are not included in this comparison.
| Fixture | SQL | Starting master median | Streaming-stage median | Median
latency reduction | Starting master P95 | Streaming-stage P95 |
|---|---|---:|---:|---:|---:|---:|
| Nonpartitioned, 10k files | EXPLAIN | 57.50 | 25.11 | 56.33% | 72.71 |
60.99 |
| Nonpartitioned, 10k files | SELECT LIMIT 10 | 340.47 | 69.91 | 79.47% |
392.70 | 123.88 |
| One partition, 10k files | EXPLAIN | 51.00 | 25.22 | 50.55% | 66.06 |
58.56 |
| One partition, 10k files | SELECT LIMIT 10 | 153.55 | 76.32 | 50.29% |
186.48 | 119.82 |
Separate A/B runs for the planning/cache changes used 500 warmups and 100
measured samples per SQL. Variable export plus statement metadata reuse reduced
EXPLAIN median from 14.21 to 9.60 ms and SELECT median from 38.33 to 16.26 ms.
Reusing the positive Hive table cache for existence checks reduced EXPLAIN from
7.62 to 3.81 ms and SELECT from 13.20 to 9.54 ms. These are separate
comparisons; their percentages must not be added. Packet captures for 30 warm
EXPLAIN statements showed HMS calls falling from 60 to 30, then to 0.
**2. Full optimization snapshot versus original Doris 4.1.3**
Full optimization snapshot: `80cbc3e54d9` FE with the same 4.1.3 BE.
Original version: `apache/doris:all-in-one-4.1.3`, actual FE/BE build
`4.1.3-rc02-7126cf65d96`.
Same nonpartitioned fixture, FE heaps 2 GiB, 500 warmup query pairs and 100
measured pairs per round, two rounds with reversed engine order. The pooled
result contains 200 raw samples per SQL/version. SQL result cache, BE file
cache, and Profile are disabled; FE metadata/file-list caches are warm. The
current FE also disables bucketed hash aggregation and the local shuffle
planner for compatibility with the older BE.
The original BE exceeded its 2 GB memory limit during warmup, before formal
sampling. Only the original BE was increased to 3 GB for the completed
comparison; the optimized FE's BE remained at 2 GB. This is a comparison of the
recorded configurations, not an equal-memory product ranking.
| Version | BE memory limit | EXPLAIN median | EXPLAIN P95 | SELECT median |
SELECT P95 |
|---|---:|---:|---:|---:|---:|
| Original Doris 4.1.3 | 3 GB | 50.65 | 152.04 | 372.11 | 1549.85 |
| Optimized FE `80cbc3e54d9` + 4.1.3 BE | 2 GB | 4.42 | 34.81 | 11.91 |
84.63 |
The observed median reductions are 91.27% for EXPLAIN and 96.80% for SELECT
in this comparison. For the optimized FE, round 1 recorded EXPLAIN 12.65 ms and
SELECT 37.21 ms; round 2 recorded EXPLAIN 3.31 ms and SELECT 7.76 ms. Both
rounds followed 500 warmups. These rounds ran after an FE restart on a shared
host; the causes of the variation were not isolated, and both rounds are
retained. Master also contains changes beyond this PR relative to 4.1.3, so the
entire difference cannot be attributed to these commits.
**3. Additional observations**
- A separate two-round run of `80cbc3e54d9`, with 500 warmups and 100
measurements per round, recorded pooled EXPLAIN median/P95 of 3.25/4.96 ms and
SELECT median/P95 of 7.75/15.82 ms. It did not simultaneously remeasure the
starting master or original 4.1.3. It must not replace the optimized values in
the paired tables above.
- With the same FE, file listing warm, and default P128/Q32, artificial fine
ranges over 10,000 files produced 1,000,000 potential splits. Synchronous
versus streaming EXPLAIN medians were 3329.12 versus 3.49 ms; streaming LIMIT
median was 59.24 ms. Each configuration had 45 measured samples. A separate
profile generated only 4,352 splits for the LIMIT. This is a million-split
stress case, not a million-real-file full-scan measurement.
- Full COUNT/SUM still consumes all splits. Small same-FE
synchronous/streaming comparisons recorded nonpartitioned medians of
1598.00/1592.62 ms and single-partition medians of 1699.13/1548.57 ms (2
warmups, 12 samples each). Large LIMIT savings should not be generalized to all
full scans.
The synthetic fixture has repeatedly shown sub-5 ms EXPLAIN medians, but the
original `bench_hms.bench_hive.files_10000` acceptance run and final-head
performance remeasurement remain outstanding. Cold directory listing can still
materialize the file-status list; ordinary EXPLAIN still initializes the split
source and waits for its first sample.
### Release note
Hive batch scans generate splits incrementally with bounded producer batches
and per-BE queues. BE split fetches respect their requested batch size. Full
queues no longer occupy shared scheduling workers, and partition-batch
pre-generation is bounded per scan. Warm Hive planning avoids redundant
metadata requests. Connector plugin API is updated to 15.0; rebuild external
connector plugins against the new API.
### Check List (For Author)
- Test
- [x] Regression test
- `test_hive_streaming_splits`: 49 result groups in one suite,
standard generated expected output, then comparison against the final
pre-synchronization feature snapshot `f33859be44a` with current FE + original
4.1.3 BE and one scheduler worker. Covers synchronous/P3Q1/P7Q2/P128Q32,
COUNT/SUM/distinct identities, partition pruning, empty inputs, JOIN/UNION,
LIMIT, CTAS, hints, and invalid settings.
- [x] Unit Test
- 103 targeted tests in ten classes passed at `f33859be44a`.
- After synchronization with master `fd268c27f5c`, 303 tests in 46
classes passed using `run-fe-ut.sh`, including the entire connector SPI suite,
Hive/HMS metadata tests, variable export tests, and the split
scheduling/consumption tests.
- [x] Manual test (add detailed scripts or steps below)
- Performance measurements and conditions are listed above.
Million-row COUNT/SUM, per-file identities, and actual million-file LIMIT were
also checked; complete aggregates over the actual million-file fixture remain
outstanding.
- FE `build.sh --fe` and Checkstyle passed at `f33859be44a`.
- The upstream-synchronized candidate passed FE Checkstyle
validation; a fresh `build.sh --fe` is in progress and its result will be added
here.
- Behavior changed:
- [x] Yes. Incremental/bounded Hive split generation, strict BE
consumption batches, cooperative scheduling, positive metadata reuse, and
connector API 15.0 as described above.
- Does this need documentation?
- [x] Yes. The two session settings and connector API version change
need user/plugin documentation; the documentation PR is pending.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
--
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]
