feat(flight): add New Flight inbound transport

Author: dqhl76

.github/actions/test_query_flight_reconnect_tpch/action.yml

@@ -0,0 +1,14 @@
+name: "Test Query Flight Reconnect with TPC-H"
+description: "Verify New Flight reconnects after a random live link reset"
+runs:
+  using: "composite"
+  steps:
+    - uses: ./.github/actions/setup_test
+      with:
+        artifacts: meta,query,sqllogictests
+
+    - uses: ./.github/actions/setup_minio
+
+    - name: Run TPC-H Flight reconnect test
+      shell: bash
+      run: ./scripts/ci/ci-run-query-flight-reconnect-tpch.sh

.github/workflows/reuse.linux.yml

@@ -423,6 +423,24 @@ jobs:
         with:
           name: test-stateful-cluster-linux
 
+  test_query_flight_reconnect_tpch:
+    needs: [build, check]
+    runs-on:
+      - self-hosted
+      - "${{ inputs.runner_arch }}"
+      - Linux
+      - 2c
+      - "${{ inputs.runner_provider }}"
+    steps:
+      - uses: actions/checkout@v6
+      - uses: ./.github/actions/test_query_flight_reconnect_tpch
+        timeout-minutes: 20
+      - name: Upload failure
+        if: failure() || cancelled()
+        uses: ./.github/actions/artifact_failure
+        with:
+          name: test-query-flight-reconnect-tpch
+
   test_stateful_large_data:
     if: contains(github.event.pull_request.labels.*.name, 'ci-largedata')
     needs: [build, check]

scripts/ci/ci-run-query-flight-reconnect-tpch.sh

@@ -0,0 +1,34 @@
+#!/usr/bin/env bash
+# Copyright 2020-2026 The Databend Authors.
+# SPDX-License-Identifier: Apache-2.0.
+
+set -euo pipefail
+
+export STORAGE_TYPE=s3
+export STORAGE_S3_BUCKET=testbucket
+export STORAGE_S3_ROOT=admin
+export STORAGE_S3_ENDPOINT_URL=http://127.0.0.1:9900
+export STORAGE_S3_ACCESS_KEY_ID=minioadmin
+export STORAGE_S3_SECRET_ACCESS_KEY=minioadmin
+export STORAGE_ALLOW_INSECURE=true
+
+readonly BUILD_PROFILE="${BUILD_PROFILE:-debug}"
+readonly SCRIPT_PATH="$(cd "$(dirname "$0")" >/dev/null 2>&1 && pwd)"
+readonly REPO_PATH="$(cd "$SCRIPT_PATH/../.." >/dev/null 2>&1 && pwd)"
+readonly TPCH_DATA_PATH=/tmp/tpch_1
+
+python3 -m pip install --quiet mysql-connector-python requests
+sudo apt-get update -yq
+sudo apt-get install -yq iproute2 iptables lsof
+
+cd "$REPO_PATH"
+./scripts/ci/deploy/databend-query-cluster-3-nodes.sh
+
+rm -rf -- "$TPCH_DATA_PATH"
+bash tests/sqllogictests/scripts/prepare_tpch_data.sh tpch_test 1
+
+python3 tests/query-flight-reconnect/test_tpch_reconnect.py \
+	--sqllogictests "target/${BUILD_PROFILE}/databend-sqllogictests" \
+	--tpch-suite tests/sqllogictests/suites/tpch/queries.test \
+	--operation-log .databend/tpch-flight-reconnect/operations.log \
+	--repo-dir "$REPO_PATH"

src/query/service/src/schedulers/fragments/query_fragment_actions.rs

@@ -230,6 +230,8 @@ impl QueryFragmentsActions {
 
     /// unique map(target, map(source, vec(fragment_id)))
     fn fragments_connections(&self, builder: &mut DataflowDiagramBuilder) -> Result<()> {
+        let new_flight = self.ctx.get_settings().get_enable_experiment_new_flight()?;
+
         for fragment_actions in &self.fragments_actions {
             if let Some(exchange) = &fragment_actions.data_exchange {
                 let destinations = exchange.get_destinations();
@@ -249,7 +251,11 @@ impl QueryFragmentsActions {
                             )?;
                         } else {
                             for channel in exchange.get_channels(destination) {
-                                builder.add_data_edge(&source, destination, &channel)?;
+                                if new_flight && matches!(exchange, DataExchange::Merge(_)) {
+                                    builder.add_merge_edge(&source, destination, &channel)?;
+                                } else {
+                                    builder.add_data_edge(&source, destination, &channel)?;
+                                }
                             }
                         }
                     }