This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new d60d94c7582 [fix](arrow-flight) Deduplicate parallel result endpoints
by backend (#69780)
d60d94c7582 is described below
commit d60d94c7582964e4f82396c5be30aaabd429a2c9
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Thu Oct 29 04:57:11 2026 +0000
[fix](arrow-flight) Deduplicate parallel result endpoints by backend
(#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.
Co-authored-by: Gabriel <[email protected]>
---
.../org/apache/doris/qe/NereidsCoordinator.java | 7 ++
.../qe/NereidsCoordinatorFlightEndpointsTest.java | 121 +++++++++++++++++++
.../test_flight_parallel_partitions.groovy | 134 +++++++++++++++++++++
3 files changed, 262 insertions(+)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/qe/NereidsCoordinator.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/NereidsCoordinator.java
index b7715396048..a21f6f30f72 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/NereidsCoordinator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/NereidsCoordinator.java
@@ -74,6 +74,7 @@ import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.util.Comparator;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -537,6 +538,7 @@ public class NereidsCoordinator extends Coordinator {
if (dataSink instanceof ResultSink || dataSink instanceof
ResultFileSink) {
// The client pulls the result from the backend (Arrow Flight
SQL); register where.
if (connectContext != null &&
!connectContext.isReturnResultFromLocal()) {
+ Set<Long> resultBackendIds = new HashSet<>();
for (AssignedJob instance : topPlan.getInstanceJobs()) {
BackendWorker worker = (BackendWorker)
instance.getAssignedWorker();
Backend backend = worker.getBackend();
@@ -545,6 +547,11 @@ public class NereidsCoordinator extends Coordinator {
}
TUniqueId finstId;
if
(connectContext.getSessionVariable().enableParallelResultSink()) {
+ // Parallel instances on a BE share one query-id
buffer, so their tickets
+ // cannot be published as independently consumable
result partitions.
+ if (!resultBackendIds.add(worker.id())) {
+ continue;
+ }
finstId = getQueryId();
} else {
finstId = instance.instanceId();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/NereidsCoordinatorFlightEndpointsTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/NereidsCoordinatorFlightEndpointsTest.java
new file mode 100644
index 00000000000..0289b09d07b
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/qe/NereidsCoordinatorFlightEndpointsTest.java
@@ -0,0 +1,121 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.qe;
+
+import org.apache.doris.arrowflight.protocol.FlightProtocolAdapter;
+import org.apache.doris.arrowflight.results.FlightSqlEndpointsLocation;
+import org.apache.doris.nereids.trees.plans.distribute.PipelineDistributedPlan;
+import org.apache.doris.nereids.trees.plans.distribute.worker.BackendWorker;
+import org.apache.doris.nereids.trees.plans.distribute.worker.job.AssignedJob;
+import org.apache.doris.planner.ResultSink;
+import org.apache.doris.system.Backend;
+import org.apache.doris.thrift.TNetworkAddress;
+import org.apache.doris.thrift.TUniqueId;
+
+import com.google.common.collect.ImmutableList;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+
+public class NereidsCoordinatorFlightEndpointsTest {
+ private static final TUniqueId QUERY_ID = new TUniqueId(1, 2);
+
+ @Test
+ public void parallelInstancesShareOneEndpointPerBackend() throws Exception
{
+ List<FlightSqlEndpointsLocation> endpoints = endpoints(true, false, 1,
8);
+ Assertions.assertEquals(1, endpoints.size());
+ Assertions.assertEquals(QUERY_ID, endpoints.get(0).getFinstId());
+ }
+
+ @Test
+ public void parallelEndpointsRetainAllResultBackendsInOrder() throws
Exception {
+ List<FlightSqlEndpointsLocation> endpoints = endpoints(true, false, 6,
8);
+ Assertions.assertEquals(6, endpoints.size());
+ for (int i = 0; i < endpoints.size(); i++) {
+ Assertions.assertEquals(QUERY_ID, endpoints.get(i).getFinstId());
+ Assertions.assertEquals(new TNetworkAddress("127.0.0.1", 8100 + i),
+ endpoints.get(i).getResultInternalServiceAddr());
+ }
+ }
+
+ @Test
+ public void sharedFlightAddressDoesNotMergeDifferentBackends() throws
Exception {
+ List<FlightSqlEndpointsLocation> endpoints = endpoints(true, false, 2,
2);
+ Assertions.assertEquals(2, endpoints.size());
+ Assertions.assertEquals(endpoints.get(0).getResultFlightServerAddr(),
+ endpoints.get(1).getResultFlightServerAddr());
+
Assertions.assertNotEquals(endpoints.get(0).getResultInternalServiceAddr(),
+ endpoints.get(1).getResultInternalServiceAddr());
+ }
+
+ @Test
+ public void nonParallelEndpointsKeepDistinctInstanceIds() throws Exception
{
+ List<FlightSqlEndpointsLocation> endpoints = endpoints(false, false,
1, 2);
+ Assertions.assertEquals(2, endpoints.size());
+ Assertions.assertEquals(new TUniqueId(2, 0),
endpoints.get(0).getFinstId());
+ Assertions.assertEquals(new TUniqueId(2, 1),
endpoints.get(1).getFinstId());
+ }
+
+ @Test
+ public void localResultDoesNotPublishFlightEndpoints() throws Exception {
+ Assertions.assertTrue(endpoints(true, true, 2, 8).isEmpty());
+ }
+
+ private List<FlightSqlEndpointsLocation> endpoints(boolean parallel,
boolean local,
+ int backendCount, int instancesPerBackend) throws Exception {
+ ConnectContext context =
ConnectContext.forFlight("test-peer-identity");
+ if (!local) {
+ // A query's result stays on the backends; the coordinator
registers where to pull it.
+ FlightProtocolAdapter.of(context).beforeQuery(context);
+ }
+ context.getSessionVariable().setEnableParallelResultSink(parallel);
+ CoordinatorContext coordinatorContext =
Mockito.mock(CoordinatorContext.class);
+ setContextField(coordinatorContext, "connectContext", context);
+ setContextField(coordinatorContext, "dataSink",
Mockito.mock(ResultSink.class));
+ PipelineDistributedPlan plan =
Mockito.mock(PipelineDistributedPlan.class, Mockito.RETURNS_DEEP_STUBS);
+ List<AssignedJob> jobs = new ArrayList<>();
+ for (int i = 0; i < backendCount; i++) {
+ Backend backend = new Backend(i + 1, "127.0.0.1", 9000 + i);
+ backend.setBrpcPort(8100 + i);
+ // Backend identity must remain distinct even when Flight
locations are shared.
+ backend.setArrowFlightSqlPort(8050);
+ for (int j = 0; j < instancesPerBackend; j++) {
+ AssignedJob job = Mockito.mock(AssignedJob.class);
+ Mockito.when(job.getAssignedWorker()).thenReturn(new
BackendWorker(0, backend));
+ Mockito.when(job.instanceId()).thenReturn(new TUniqueId(2, i *
instancesPerBackend + j));
+ jobs.add(job);
+ }
+ }
+
Mockito.when(plan.getInstanceJobs()).thenReturn(ImmutableList.copyOf(jobs));
+
Mockito.when(plan.getFragmentJob().getFragment().getOutputExprs()).thenReturn(new
ArrayList<>());
+ NereidsCoordinator coordinator =
Mockito.mock(NereidsCoordinator.class, Mockito.CALLS_REAL_METHODS);
+ Mockito.doReturn(QUERY_ID).when(coordinator).getQueryId();
+ coordinator.processTopSink(coordinatorContext, plan);
+ return context.getFlightSqlEndpointsLocations();
+ }
+
+ private void setContextField(CoordinatorContext context, String name,
Object value) throws Exception {
+ Field field = CoordinatorContext.class.getDeclaredField(name);
+ field.setAccessible(true);
+ field.set(context, value);
+ }
+}
diff --git
a/regression-test/suites/arrow_flight_sql_p0/test_flight_parallel_partitions.groovy
b/regression-test/suites/arrow_flight_sql_p0/test_flight_parallel_partitions.groovy
new file mode 100644
index 00000000000..005341df0ee
--- /dev/null
+++
b/regression-test/suites/arrow_flight_sql_p0/test_flight_parallel_partitions.groovy
@@ -0,0 +1,134 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import org.apache.arrow.driver.jdbc.shaded.com.google.protobuf.Any
+import org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.flight.CallOptions
+import org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.flight.FlightClient
+import org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.flight.Location
+import
org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.flight.sql.FlightSqlClient
+import
org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.flight.sql.impl.FlightSql
+import
org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.memory.RootAllocator
+
+import java.util.concurrent.Callable
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+
+suite("test_flight_parallel_partitions", "arrow_flight_sql") {
+ def frontend = jdbc_sql_return_maparray("SHOW FRONTENDS").find {
+ it.IsMaster.toString().equalsIgnoreCase("true") &&
it.Alive.toString().equalsIgnoreCase("true")
+ }
+ assertNotNull(frontend)
+ assertTrue(frontend.ArrowFlightSqlPort.toString().toInteger() > 0)
+ def database = jdbc_sql("SELECT DATABASE()")[0][0]
+ def table = "${database}.flight_parallel_partition_source"
+ def allocator = new RootAllocator(Long.MAX_VALUE)
+ def feClient = FlightClient.builder(allocator,
+ Location.forGrpcInsecure(frontend.Host.toString(),
frontend.ArrowFlightSqlPort.toString().toInteger())).build()
+ def client = new FlightSqlClient(feClient)
+ def auth
+ def readEndpoint = { endpoint, boolean collectRows ->
+ def rows = []
+ FlightClient.builder(allocator,
endpoint.locations[0]).build().withCloseable { beClient ->
+ beClient.getStream(endpoint.ticket, auth, CallOptions.timeout(30,
TimeUnit.SECONDS)).withCloseable { stream ->
+ while (stream.next()) {
+ if (collectRows) {
+ def vector = stream.root.getVector(0)
+ for (int i = 0; i < stream.root.rowCount; i++) {
+ rows.add(((Number)
vector.getObject(i)).longValue())
+ }
+ }
+ }
+ }
+ }
+ rows
+ }
+ def executeSetting = { String statement ->
+ client.execute(statement, auth).endpoints.each { readEndpoint(it,
false) }
+ }
+ try {
+ auth =
feClient.authenticateBasicToken(context.config.otherConfigs.get("extArrowFlightSqlUser"),
+
context.config.otherConfigs.get("extArrowFlightSqlPassword")).get()
+ executeSetting("SET enable_sql_cache=false")
+ executeSetting("SET enable_nereids_distribute_planner=true")
+ executeSetting("SET parallel_pipeline_task_num=8")
+ executeSetting("SET query_timeout=60")
+ jdbc_sql("DROP TABLE IF EXISTS ${table}")
+ jdbc_sql("CREATE TABLE ${table} (id BIGINT NOT NULL) DISTRIBUTED BY
HASH(id) BUCKETS 60 " +
+ "PROPERTIES(\"replication_num\"=\"1\")")
+ jdbc_sql("INSERT INTO ${table} SELECT number FROM
numbers(\"number\"=\"60\")")
+ def resultBackendCount = jdbc_sql_return_maparray("SHOW TABLETS FROM
${table}")
+ .collect { it.BackendId }.unique().size()
+ [true, false].each { parallel ->
+ executeSetting("SET enable_parallel_result_sink=${parallel}")
+ [false, true].each { concurrent ->
+ // Each execution owns fresh tickets; consuming a ticket does
not create a replayable partition.
+ def info = client.execute("SELECT id * 1000 + n AS sequence_id
FROM ${table} " +
+ "LATERAL VIEW explode_numbers(1000) expanded AS n",
auth)
+ assertTrue(!info.endpoints.isEmpty())
+ def tickets = info.endpoints.collect {
Base64.encoder.encodeToString(it.ticket.bytes) }
+ assertEquals(tickets.size(), tickets.toSet().size(),
"Duplicate Flight result tickets")
+ if (parallel) {
+ def resultAddresses = info.endpoints.collect { endpoint ->
+ def fields =
Any.parseFrom(endpoint.ticket.bytes).unpack(FlightSql.TicketStatementQuery.class)
+ .statementHandle.toStringUtf8().split("&")
+ "${fields[1]}:${fields[2]}".toString()
+ }
+ // Instance parallelism must not publish multiple readers
for the same BE result buffer.
+ assertEquals(info.endpoints.size(),
resultAddresses.toSet().size())
+ if (resultBackendCount > 1) {
+ assertTrue(info.endpoints.size() > 1, "Expected
multiple result backends")
+ }
+ } else {
+ assertEquals(1, info.endpoints.size())
+ }
+ def rows = []
+ if (concurrent) {
+ def executor = Executors.newFixedThreadPool(Math.min(8,
info.endpoints.size()))
+ try {
+ def futures = info.endpoints.collect { endpoint ->
+ executor.submit({ readEndpoint(endpoint, true) }
as Callable)
+ }
+ futures.each { rows.addAll(it.get(60,
TimeUnit.SECONDS)) }
+ } finally {
+ executor.shutdownNow()
+ assertTrue(executor.awaitTermination(35,
TimeUnit.SECONDS))
+ }
+ } else {
+ info.endpoints.each { rows.addAll(readEndpoint(it, true)) }
+ }
+ assertEquals(60000, rows.size())
+ assertEquals((0L..<60000L).toList(), rows.sort())
+ }
+ }
+ } finally {
+ try {
+ if (auth != null) {
+ feClient.closeSession(new
org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.flight.CloseSessionRequest(),
auth)
+ }
+ } finally {
+ try {
+ client.close()
+ } finally {
+ try {
+ allocator.close()
+ } finally {
+ jdbc_sql("DROP TABLE IF EXISTS ${table}")
+ }
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
