feat(query): add New Flight transport by dqhl76 · Pull Request #20397 · databendlabs/databend

dqhl76 · 2026-08-27T03:15:56Z

I hereby agree to the terms of the CLA available at: https://docs.databend.com/dev/policies/cla/

Summary

This PR adds the complete default-off New Flight transport and separates transport mechanics from exchange-domain pipeline code.

It includes:

reliable inbound and outbound logical streams over Arrow Flight DoExchange
sequence acknowledgements, replay deduplication, reconnect attempt budgets, reconnect leases, and terminal error propagation
fragment and statistics stream integration
per-source backpressure and exchange packet routing
pipeline adapters for local, legacy remote, and reliable remote exchange paths
a clear module boundary between transport/legacy, transport/reliable, and exchange-owned adapters

Implementation

The old mixed v1/network module is replaced by two explicit transport implementations:

v1/transport/legacy: the existing ping-pong transport and outbound buffer
v1/transport/reliable: protocol framing, reconnect policy, logical inbound/outbound state machines, and the ReliableInboundDelivery boundary

Exchange-specific concerns remain under v1/exchange, including DataBlock serialization, local channels, quota queues, TID/batch routing, pipeline processors, and statistics delivery. The reliable transport operates on FlightData and no longer depends on Settings, DataBlock, pipeline types, or QueryContext.

The setting-to-transport conversion is also owned by the exchange layer, so transport selection remains outside the protocol implementation.

Compatibility and rollout

enable_experiment_new_flight defaults to 0.

Data stream	Execution handling	Existing Flight	New Flight
Packet-level fragment	Pipeline sink/source processors	do_get	Reliable do_exchange
Block-level fragment	OutboundChannel / InboundChannel	Ping-pong do_exchange	Reliable do_exchange
Statistics	Runtime sender/receiver tasks	do_get	Reliable do_exchange
Local fragment	In-process pipeline channels	No Flight RPC	No Flight RPC

Enable

    SET flight_client_keep_alive_time_secs = 120;
    SET flight_client_keep_alive_interval_secs = 30;
    SET flight_client_keep_alive_retries = 3;

    SET flight_connection_max_retry_times = 5;
    SET flight_connection_retry_interval = 3;
    SET flight_client_timeout = 60;

Tests

Unit Test
Logic Test
Benchmark Test
No Test - Explain why

Focused tests cover wire compatibility, packet validation, replay deduplication, reconnect lease replacement, reconnect budget exhaustion, sender/receiver failure propagation, legacy ping-pong and buffering, fragment TID routing, statistics completion/failure, and end-to-end exchange setup/cleanup.

Type of change

Bug Fix (non-breaking change which fixes an issue)
New Feature (non-breaking change which adds functionality)
Breaking Change (fix or feature that could cause existing functionality not to work as expected)
Documentation Update
Refactoring
Performance Improvement
Other (please describe):

AI assistance

AI usage: An AI coding agent helped implement the reliable inbound/outbound transport, reorganize the legacy and reliable module boundaries, add focused tests, and prepare the PR summary.
Responsible human: @dqhl76
The responsible human has read every line of this diff and can explain each change

This change is

dqhl76 · 2026-08-27T06:21:01Z

@codex review

chatgpt-codex-connector · 2026-08-27T06:24:11Z

Codex Review: Didn't find any major issues. 🎉

Reviewed commit: d79c7853c7

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

Open a pull request for review
Mark a draft as ready
Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

github-actions · 2026-08-27T16:02:22Z

Docker Image for PR

tag: pr-20397-1120c7b-1787846362

note: this image tag is only available for internal use.

dqhl76

These are review comments based on head 1cd15af; this review is neither an approval nor a request for changes.

dqhl76 · 2026-09-02T03:01:55Z

                        let mut statistics_receiver = statistics_receiver.lock();

                        statistics_receiver.shutdown(info.res.is_err());
+                        let result = statistics_receiver.wait_shutdown();


[P1] This reverses the lifecycle from shutdown -> on_finished_query -> wait_shutdown to shutdown -> wait_shutdown -> on_finished_query, creating a circular wait. wait_shutdown() blocks the pipeline-finish callback until the statistics handler exits; in reliable mode, that handler waits for the remote ReceiverClosed, while the remote statistics stream generally cannot terminate until on_finished_query() removes the coordinator and disconnects the connection. The query can therefore finish while the statistics stream remains open and occupies a pipeline thread for the lease duration. Please call on_finished_query() before waiting, or otherwise break this dependency, and add a regression test where the remote statistics stream has not closed proactively.

dqhl76 · 2026-09-02T03:01:55Z

                                {
                                    ctx.get_exchange_manager()
-                                        .shutdown_query(&query_id, Some(cause));
+                                        .shutdown_query(&query_id, Some(cause.clone()));


[P1] This changes statistics-send failures from warn-only to shutdown_query() plus tx.fail(), but StatisticsSender serves both New Flight and the existing Flight path. On the existing path, sending after the consumer has closed can return AbortedQuery; normal coordinator cleanup can therefore be promoted to a query-wide failure even when enable_experiment_new_flight=0. Please avoid changing the existing path's terminal semantics without gating this behavior by transport, and add a regression test for an already-closed consumer on the existing path.

dqhl76 · 2026-09-02T03:01:55Z

+            return Ok(None);
+        }
+
+        Ok(Some(Self::new(


[P2] Enabling New Flight does not necessarily enable reconnection. flight_max_retry_times defaults to 0, so reconnect_attempts() is empty, receiver_lease() is zero, and the first disconnect fails immediately. This implicit dependency makes it easy to believe reliable transport is active when it is not. Please reject this setting combination in from_settings(), or at least emit an explicit warning, and add coverage for the setting combinations.

dqhl76 · 2026-09-02T03:01:55Z

+    }
+
+    fn send_next(&mut self) {
+        if self.logical.in_flight.is_some() {


[P2] There is only one in_flight request, so every DATA packet must wait for its ACK. With the 256 KiB batch limit, per-stream throughput is bounded at roughly 256 KiB / RTT. This may be acceptable at sub-millisecond rack-local RTT, but it degrades substantially across AZs or regions. Please provide RTT/throughput measurements before merge, or increase the window beyond one and document the supported operating envelope.

dqhl76 · 2026-09-02T03:01:55Z

+        local_node_id: String,
+        remote_node_id: String,
+    ) -> Result<Self> {
+        Ok(Self {


[P2] connect() has already established do_exchange, but the driver is not spawned until the later start() call. During this interval no one polls response_stream, so remote FAIL/ReceiverClosed messages and physical disconnects are not detected promptly. If pipeline construction fails and this pending outbound is never consumed, there is also no Drop path that sends SenderFail; the peer can only wait for the lease to expire. Please start a controlled driver immediately after connection, or give the pending state explicit RAII/Drop termination semantics, and cover the connected-but-never-started case.

dqhl76 · 2026-09-02T03:01:55Z

+            channels,
+            completion: Some(Arc::new(ReliableCompletion {
+                streams,
+                remaining_producers: AtomicUsize::new(num_producers),


[P2] remaining_producers relies on a caller-supplied count that is not coupled to the actual producer or SharedOutboundChannels clone lifetime. Over-counting prevents FINISH forever; under-counting sends FINISH early; zero or an extra fetch_sub underflows to usize::MAX. Please bind the count to producer/clone lifetime, or at least add a clear invariant/assertion, and add a test that fails deterministically when the producer count is inconsistent.

dqhl76 · 2026-09-02T03:01:55Z

+        }
+
+        let accepted = self.deliver(data).await;
+        *next_sequence += 1;


[P2] next_sequence is advanced before accepted? is checked. If deliver() returns an error, the logical sequence has already moved forward, so a replay of the same sequence can be treated as previously delivered and ACKed. Please advance the sequence only after DeliveryOutcome::Accepted or an explicit terminal outcome, or encode and prove the invariant that any delivery error permanently terminates this source.

dqhl76 · 2026-09-02T03:01:55Z


+/// Inbound queue quota per do_exchange connection.
+// TODO: get max_bytes_per_connection from query settings
+const MAX_INBOUND_BYTES_PER_CONNECTION: usize = 20 * 1024 * 1024;


[P3] The name still describes a 20 MiB per-connection limit, but New Flight creates the queue per source. For example, ten sources feeding one merge channel allow roughly 200 MiB of aggregate inbound buffering. Please rename and document this as a per-source quota, expose matching setting semantics, and evaluate the aggregate memory bound.

dqhl76 · 2026-09-02T03:01:56Z

+                                            reconnect.receiver_lease_secs(),
+                                        )
+                                    }
+                                    NewFlightStream::Statistics => unreachable!(


[P3] This unreachable!() is justified only by the current caller handling Statistics first; it is not guaranteed by the wire or protocol layer. A later refactor can pass this valid enum variant here and panic the process. Please separate statistics from this enum or return an explicit error instead of encoding a call-order convention as a process-level panic.

github-actions Bot added the pr-feature this PR introduces a new feature to the codebase label Aug 27, 2026

dqhl76 force-pushed the flight-split-01-new-flight-inbound branch from 7f734d9 to d79c785 Compare August 27, 2026 06:05

dqhl76 force-pushed the flight-split-01-new-flight-inbound branch 2 times, most recently from af1762d to a1d76eb Compare August 27, 2026 06:41

dqhl76 requested a review from zhang2014 August 27, 2026 06:41

zhang2014 added the ci-cloud Build docker image for cloud test label Aug 27, 2026

dqhl76 changed the title ~~feat(query): add New Flight inbound transport~~ feat(query): add New Flight transport Aug 29, 2026

dqhl76 force-pushed the flight-split-01-new-flight-inbound branch 3 times, most recently from 3132523 to 49dfaf2 Compare August 30, 2026 06:39

feat(flight): add New Flight inbound transport

8f4438c

dqhl76 force-pushed the flight-split-01-new-flight-inbound branch from 87b234a to 8f4438c Compare August 30, 2026 09:17

dqhl76 added 4 commits August 30, 2026 18:30

ci: stress Flight reconnect with network partitions

16dc338

fixup

2651f91

ci(flight): add reliable transport integration tests

d13a09c

close

1cd15af

dqhl76 commented Sep 2, 2026

View reviewed changes

Navigation Menu

Uh oh!

feat(query): add New Flight transport - #20397

feat(query): add New Flight transport#20397
dqhl76 wants to merge 5 commits into
databendlabs:mainfrom
dqhl76:flight-split-01-new-flight-inbound

dqhl76 commented Aug 27, 2026 •

edited

Loading

Uh oh!

dqhl76 commented Aug 27, 2026

Uh oh!

chatgpt-codex-connector Bot commented Aug 27, 2026

Uh oh!

github-actions Bot commented Aug 27, 2026

Uh oh!

dqhl76 left a comment •

edited

Loading

Uh oh!

dqhl76 Sep 2, 2026 •

edited

Loading

Uh oh!

dqhl76 Sep 2, 2026 •

edited

Loading

Uh oh!

dqhl76 Sep 2, 2026 •

edited

Loading

Uh oh!

dqhl76 Sep 2, 2026 •

edited

Loading

Uh oh!

dqhl76 Sep 2, 2026 •

edited

Loading

Uh oh!

dqhl76 Sep 2, 2026 •

edited

Loading

Uh oh!

dqhl76 Sep 2, 2026 •

edited

Loading

Uh oh!

dqhl76 Sep 2, 2026 •

edited

Loading

Uh oh!

dqhl76 Sep 2, 2026 •

edited

Loading

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

dqhl76 commented Aug 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Implementation

Compatibility and rollout

Enable

Tests

Type of change

AI assistance

Uh oh!

dqhl76 commented Aug 27, 2026

Uh oh!

chatgpt-codex-connector Bot commented Aug 27, 2026

Uh oh!

github-actions Bot commented Aug 27, 2026

Docker Image for PR

Uh oh!

dqhl76 left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dqhl76 Sep 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dqhl76 Sep 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dqhl76 Sep 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dqhl76 Sep 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dqhl76 Sep 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dqhl76 Sep 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dqhl76 Sep 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dqhl76 Sep 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dqhl76 Sep 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

dqhl76 commented Aug 27, 2026 •

edited

Loading

dqhl76 left a comment •

edited

Loading

dqhl76 Sep 2, 2026 •

edited

Loading

dqhl76 Sep 2, 2026 •

edited

Loading

dqhl76 Sep 2, 2026 •

edited

Loading

dqhl76 Sep 2, 2026 •

edited

Loading

dqhl76 Sep 2, 2026 •

edited

Loading

dqhl76 Sep 2, 2026 •

edited

Loading

dqhl76 Sep 2, 2026 •

edited

Loading

dqhl76 Sep 2, 2026 •

edited

Loading

dqhl76 Sep 2, 2026 •

edited

Loading