morningman opened a new pull request, #69780:
URL: https://github.com/apache/doris/pull/69780
### What problem does this PR solve?
Issue Number: None
Related PR: #68642
Problem Summary:
#### In short
This ports branch-4.x's #68642 to master. With parallel result sink on,
which is the master default, the result fragment's instances on one BE all
write into one result buffer, and the Flight ticket names that buffer only by
query id and BE address. The coordinator still published one Flight endpoint
per instance, so a query with 8 instances per BE returned 8 identical tickets
for every BE. This PR publishes one endpoint per result backend in that mode.
#### Background
- An Arrow Flight SQL query leaves its result on the BEs.
`NereidsCoordinator.setForArrowFlight` registers where to pull it
(`FlightSqlEndpointsLocation`), and `DorisFlightSqlProducer` turns every
registered location into one `FlightEndpoint`, whose ticket is `<id>&<BE brpc
host>&<BE brpc port>&<sql>`.
- With `enable_parallel_result_sink` (default `true`), every instance of the
result fragment on a BE writes into one result buffer registered under the
query id, and `<id>` is the query id. Without it, each instance has its own
buffer under its instance id.
- A Flight SQL client reads every endpoint of the `FlightInfo` it gets back,
each through its own `DoGet` stream.
#### The problem, and what it cost
`setForArrowFlight` added one location per instance in both modes. In
query-id mode that is one copy of the same ticket per instance: 6 BEs with 8
instances each published 48 endpoints for 6 result buffers. A client reading
all of them opened 8 streams on each buffer, and
`ArrowFlightResultBlockBuffer::get_arrow_batch` hands each batch to whichever
stream asks first, so the rows of one buffer were split across 8 streams that
were presented as separate result partitions under the same ticket. The remote
Doris catalog was not affected: `RemoteDorisScanNode.processFlightEndpoints`
already drops repeated ticket and location pairs.
| Parallel result sink, 1 BE, 8 instances | Endpoints | Distinct tickets |
|---|---|---|
| Before this PR | 8 | 1 |
| After this PR | 1 | 1 |
#### How this PR fixes it
- In query-id mode `setForArrowFlight` keeps a set of the backend ids it has
published and skips an instance whose backend is already in it, so each result
backend gets one endpoint, in the order the backends first appear. The key is
the backend id, not the Flight address, so two BEs whose Flight addresses are
equal still get one endpoint each.
- Instance-id mode (parallel result sink off) is unchanged: every instance
owns its buffer and keeps its own endpoint. A result returned from the FE
publishes no endpoint, as before.
#### Results
- `NereidsCoordinatorFlightEndpointsTest` (5 tests): 8 parallel instances on
one BE give one endpoint carrying the query id; 6 BEs × 8 instances give 6
endpoints in BE order; two BEs sharing a Flight address keep two endpoints;
parallel result sink off keeps one endpoint per instance id; a local result
publishes none. Without the fix 3 of them fail (`expected: <6> but was: <48>`,
`<1>` vs `<8>`, `<2>` vs `<4>`); keying the set on the Flight address instead
of the backend fails 2.
- `arrow_flight_sql_p0/test_flight_parallel_partitions` runs a 60-bucket
query with `parallel_pipeline_task_num=8`, with parallel result sink on and
off, and reads the endpoints one after another and concurrently: tickets must
be unique, one endpoint per result BE in query-id mode, and all 60000 rows must
arrive exactly once. On a local single-BE cluster it fails before this PR with
`Duplicate Flight result tickets ==> expected: <8> but was: <1>` and passes
with it.
### Release note
Fix Arrow Flight SQL publishing one result endpoint per pipeline instance
instead of one per backend when parallel result sink is enabled, which gave
clients several identical tickets for each backend's result.
### Check List (For Author)
- Test: Regression test / Unit Test
- FE UT: `NereidsCoordinatorFlightEndpointsTest` (new, 5 tests), with
`NereidsCoordinatorTest`, `StmtExecutorTest`, `FlightProtocolAdapterTest`,
`FlightSqlConnectProcessorSchemaTest`, `ProtocolCapabilityWiringTest`,
`DorisFlightSqlProducerTest`: 106 tests pass, fe-core Checkstyle passes
- Regression on a local cluster (the FE built with this change):
`test_flight_parallel_partitions` fails before and passes after; ten remote
Doris suites that read over Arrow Flight (`test_remote_doris_all_types_select`,
`_statistics`, `_table_stats`, `_agg_table_select`, `_predict`,
`_flight_session`, `_all_types_show`, `_catalog`, `_unique_table_select`,
`_refresh`) pass
- Behavior changed: Yes. With parallel result sink on, an Arrow Flight query
returns one endpoint per result backend instead of one per instance.
- Does this need documentation: No
Original author: @Gabriel39
--
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]
