knut-bifrost 0.2.0

//! **Apache Doris** [`TableSink`] — delivery over Stream Load.
//!
//! Feature `doris` (off by default). Enabling it adds only `reqwest`, which the
//! `airflow` feature already pulls, so the workspace dependency tree does not
//! grow. See `.nornir/doris-sink-design.md`.
//!
//! # The transport is a seam on purpose
//!
//! Every correctness property this module has — label derivation, replay
//! rejection, delete-mode refusal, redirect diagnosis, schema refusal, chunking
//! — is testable without a Doris cluster, because [`DorisSink`] talks to a
//! [`StreamLoadTransport`] rather than to an HTTP client. [`ReqwestTransport`]
//! is the real one; the tests use a fake that encodes the *documented* protocol.
//! Only throughput needs a live cluster.
//!
//! # Topology, and the error that looks like auth and is not
//!
//! Stream Load answers on the **FE** (`http_port`, default 8030) at
//! `PUT /api/{db}/{table}/_stream_load`, and the FE replies **307** with a
//! `Location` naming a **BE** (`webserver_port`, default 8040) that does the
//! actual work. So the loader must reach the FE **and every BE it may be
//! redirected to**, and the redirect carries the BE's *configured* address,
//! which on a split-addressing cluster need not be routable from here.
//!
//! **Measured against the live cluster (Doris 4.0.8):** when the FE answers but
//! the BEs are unreachable, the load does not fail — it **hangs** after the 307
//! until something times out. `/api/health` on the FE stays 200 the whole time,
//! so nothing looks wrong. Without an explicit timeout a single stuck window
//! blocks the stream forever with no error at all.
//!
//! So this module (a) follows the redirect itself, automatic following disabled
//! on the real client, and (b) puts a hard connect **and** request timeout on
//! every hop, so the stall becomes a reportable error naming the BE.
//!
//! # Unique Key CDC: one snapshot, one atomic, versioned MERGE (2026-10-02)
//!
//! In [`LoadMode::Upsert`] a window (= one Iceberg snapshot) becomes:
//!
//! * **one `merge_type: MERGE` body**: the upserts tagged `__bifrost_op = 0` and
//!   the whole-row deletes (deletion vectors, position deletes, copy-on-write
//!   pre-images) tagged `__bifrost_op = 1`, with `delete: __bifrost_op=1`.
//!   Doris batch-delete manual, "Load Merge Methods": MERGE "must be used
//!   together with" a delete condition, and is the recommended CDC mode (Q4).
//!   A key deleted and upserted in the same snapshot is an update: only the
//!   upsert is sent;
//! * identity-only deletes (equality deletes: non-key columns null) as a
//!   keys-only **`merge_type: DELETE`** load, so NULLs never reach NOT NULL
//!   value columns;
//! * every row versioned by the snapshot's Iceberg **data sequence number** in
//!   `__bifrost_seq`, mapped by `function_column.sequence_col` (stream-load
//!   manual): a replayed or late window can never overwrite newer data, deletes
//!   included;
//! * every chunk loaded with **`two_phase_commit: true`**, then committed by
//!   label through `PUT /api/{db}/{table}/_stream_load_2pc` (`txn_operation:
//!   commit`) once all landed — or aborted if one failed (Doris transaction
//!   docs, "Stream Load 2PC"). A replay finds committed chunks `FINISHED` and
//!   pre-committed ones still open, and commits those.
//!
//! The destination table this needs (Doris 4.x):
//!
//! ```sql
//! CREATE TABLE db.people (
//!   id BIGINT NOT NULL, name STRING, …
//! ) UNIQUE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 8
//! PROPERTIES (
//!   "enable_unique_key_merge_on_write" = "true",
//!   "function_column.sequence_type" = "BIGINT"
//! );
//! ```
//!
//! A table without `function_column.sequence_type` must be loaded with
//! [`DorisConfig::with_sequence`]`(false)`.
//!
//! # `Expect: 100-continue` is mandatory
//!
//! Also measured: without it Doris answers
//! `{"status":"FAILED","msg":"There is no 100-continue header"}`. This is a
//! nasty one because `curl` omits the header for small bodies and sends it for
//! large ones — so small windows fail while large ones succeed. The sink sets it
//! on every load, unconditionally, and a test asserts it is on every PUT.

use std::collections::BTreeSet;

use arrow_array::RecordBatch;
use arrow_schema::Schema;
use async_trait::async_trait;

use crate::{
    error::{BifrostError, Result},
    sink::table::{AckToken, LoadAck, TableSink, TableWindow, stream_load_label},
};

// ── the transport seam ──────────────────────────────────────────────────────

/// One HTTP exchange, reduced to what Stream Load actually needs.
#[derive(Debug, Clone)]
pub struct StreamLoadRequest {
    /// `"PUT"` for a load, `"GET"` for the schema probe.
    pub method: &'static str,
    pub url: String,
    /// Header name/value pairs, in the order the sink set them.
    pub headers: Vec<(String, String)>,
    /// Shared, not copied: following the FE→BE redirect re-sends the same bytes.
    pub body: bytes::Bytes,
}

impl StreamLoadRequest {
    /// The value of a header, case-insensitively.
    pub fn header(&self, name: &str) -> Option<&str> {
        self.headers
            .iter()
            .find(|(k, _)| k.eq_ignore_ascii_case(name))
            .map(|(_, v)| v.as_str())
    }
}

/// What came back.
#[derive(Debug, Clone)]
pub struct StreamLoadResponse {
    pub status: u16,
    /// The `Location` header of a 3xx. Redirects are **not** followed by the
    /// transport — [`DorisSink`] follows them so it can diagnose an unreachable
    /// backend precisely.
    pub location: Option<String>,
    pub body: String,
}

impl StreamLoadResponse {
    pub fn ok(body: impl Into<String>) -> Self {
        Self {
            status: 200,
            location: None,
            body: body.into(),
        }
    }

    pub fn redirect(location: impl Into<String>) -> Self {
        Self {
            status: 307,
            location: Some(location.into()),
            body: String::new(),
        }
    }

    pub fn is_redirect(&self) -> bool {
        matches!(self.status, 301 | 302 | 303 | 307 | 308) && self.location.is_some()
    }
}

/// How bytes get to Doris. Implemented for real by [`ReqwestTransport`], and by
/// a fake in the tests.
#[async_trait]
pub trait StreamLoadTransport: Send + Sync {
    async fn send(&self, req: &StreamLoadRequest) -> Result<StreamLoadResponse>;
}

// ── configuration ───────────────────────────────────────────────────────────

/// What a `DELETE` row means for this destination.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LoadMode {
    /// Duplicate-key table, append-only source. A `DELETE` row is a **hard
    /// error**: on an append-only source an unexpected delete means the premise
    /// of the deployment is wrong, and a loud failure beats a silent divergence
    /// between the lake and the engine.
    Append,
    /// Unique-key table. Delete rows are shipped as a second Stream Load with
    /// `merge_type: DELETE`, which is how Doris sets its hidden
    /// `__DORIS_DELETE_SIGN__` column. Not available on duplicate-key or
    /// aggregate-key tables — hence the two modes.
    Upsert,
}

/// Stream Load wire encoding. Doris accepts `csv`, `json`, `parquet`, `orc` and
/// `arrow`; only JSON is implemented, because it needs no dependency bifröst
/// does not already have and is unambiguous about nulls and types. The enum
/// exists so parquet/arrow can be added as a throughput follow-up without
/// changing any caller — whether they win is an empirical question, not an
/// assumption.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum LoadFormat {
    #[default]
    Json,
    /// Arrow IPC stream (`format: arrow`, Doris ≥ 2.1): the batches go as they
    /// are, no text encoding on this side and no parsing on the BE's.
    Arrow,
}

impl LoadFormat {
    const fn header_value(self) -> &'static str {
        match self {
            LoadFormat::Json => "json",
            LoadFormat::Arrow => "arrow",
        }
    }
}

/// Where to load, as what, with which credentials.
#[derive(Debug, Clone)]
pub struct DorisConfig {
    /// FE base URL including the HTTP port, e.g. `http://doris-fe:8030`.
    pub fe_url: String,
    /// Destination database and table (NOT the source table — that is the
    /// window's, and it is what the cursor is keyed on).
    pub db: String,
    pub table: String,
    pub user: String,
    pub password: String,
    pub mode: LoadMode,
    pub format: LoadFormat,
    /// Byte budget for one Stream Load body. The BE caps a single load at
    /// `streaming_load_max_mb` (default 10 GB) and Doris's own guidance is to
    /// stay well under it — a dropped connection restarts the whole request.
    /// Default 64 MiB.
    pub max_chunk_bytes: usize,
    /// TCP connect timeout for each hop. An unreachable BE after the 307 is a
    /// **hang**, not an error (measured), so this is what turns a silently stuck
    /// stream into a reportable failure. Default 10 s.
    pub connect_timeout: std::time::Duration,
    /// Whole-request timeout for each hop, the load body included. Must exceed
    /// the time a real BE needs to ingest one chunk. Default 300 s.
    pub request_timeout: std::time::Duration,
    /// Skip the FE and PUT straight at a known-healthy BE. The escape hatch for
    /// a cluster whose FE hands out addresses that are not routable from here.
    pub be_url: Option<String>,
    /// Chunks of one window loaded concurrently (each its own label). Default 4.
    pub max_parallel_loads: usize,
    /// Stream Load **two-phase commit** (default on): every chunk of a window is
    /// loaded with `two_phase_commit: true` (pre-committed, invisible), and only
    /// when all of them landed are they committed through `_stream_load_2pc` —
    /// so a multi-chunk snapshot does not become visible chunk by chunk, and a
    /// failed window is aborted instead of half-applied.
    pub two_phase_commit: bool,
    /// Version every row of an [`LoadMode::Upsert`] window by the Iceberg data
    /// sequence number of its snapshot (default on), sent as the
    /// `__bifrost_seq` column mapped by `function_column.sequence_col`. Needs a
    /// table created with `"function_column.sequence_type" = "BIGINT"` (see the
    /// module docs). Off for a table without a sequence column.
    pub sequence: bool,
    /// The destination's key columns, when the source cannot name them (a
    /// Spark table declares no identifier fields). Empty = the window's
    /// [`TableWindow::key_columns`].
    pub key_columns: Vec<String>,
}

/// Default body budget per Stream Load: 64 MiB. Two orders of magnitude below
/// the BE cap, so a retry is cheap.
pub const DEFAULT_MAX_CHUNK_BYTES: usize = 64 * 1024 * 1024;

/// Default chunks of one window in flight at once.
pub const DEFAULT_MAX_PARALLEL_LOADS: usize = 4;

/// Default TCP connect timeout per hop. See [`DorisConfig::connect_timeout`] —
/// an unreachable BE hangs rather than refusing, so this bound is the only thing
/// standing between a routing mistake and a permanently stalled stream.
pub const DEFAULT_CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);

/// Default whole-request timeout per hop.
pub const DEFAULT_REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(300);

impl DorisConfig {
    /// Credentials from the environment: `DORIS_USER`, `DORIS_PASSWORD`.
    ///
    /// There is no constructor that takes a password as a flag, and there will
    /// not be one: flags land in `/proc/<pid>/cmdline`, readable by every user
    /// on the host, and in shell history and process listings.
    pub fn from_env(
        fe_url: impl Into<String>,
        db: impl Into<String>,
        table: impl Into<String>,
    ) -> Result<Self> {
        let user = std::env::var("DORIS_USER")
            .map_err(|_| BifrostError::Other("DORIS_USER is not set".into()))?;
        let password = std::env::var("DORIS_PASSWORD")
            .map_err(|_| BifrostError::Other("DORIS_PASSWORD is not set".into()))?;
        Ok(Self {
            fe_url: fe_url.into(),
            db: db.into(),
            table: table.into(),
            user,
            password,
            mode: LoadMode::Append,
            format: LoadFormat::default(),
            max_chunk_bytes: DEFAULT_MAX_CHUNK_BYTES,
            connect_timeout: DEFAULT_CONNECT_TIMEOUT,
            request_timeout: DEFAULT_REQUEST_TIMEOUT,
            be_url: None,
            max_parallel_loads: DEFAULT_MAX_PARALLEL_LOADS,
            two_phase_commit: true,
            sequence: true,
            key_columns: Vec::new(),
        })
    }

    /// For tests and for callers that source credentials elsewhere.
    pub fn new(
        fe_url: impl Into<String>,
        db: impl Into<String>,
        table: impl Into<String>,
        user: impl Into<String>,
        password: impl Into<String>,
    ) -> Self {
        Self {
            fe_url: fe_url.into(),
            db: db.into(),
            table: table.into(),
            user: user.into(),
            password: password.into(),
            mode: LoadMode::Append,
            format: LoadFormat::default(),
            max_chunk_bytes: DEFAULT_MAX_CHUNK_BYTES,
            connect_timeout: DEFAULT_CONNECT_TIMEOUT,
            request_timeout: DEFAULT_REQUEST_TIMEOUT,
            be_url: None,
            max_parallel_loads: DEFAULT_MAX_PARALLEL_LOADS,
            two_phase_commit: true,
            sequence: true,
            key_columns: Vec::new(),
        }
    }

    #[must_use]
    pub fn with_timeouts(
        mut self,
        connect: std::time::Duration,
        request: std::time::Duration,
    ) -> Self {
        self.connect_timeout = connect;
        self.request_timeout = request;
        self
    }

    #[must_use]
    pub fn with_mode(mut self, mode: LoadMode) -> Self {
        self.mode = mode;
        self
    }

    #[must_use]
    pub fn with_max_chunk_bytes(mut self, n: usize) -> Self {
        self.max_chunk_bytes = n.max(1);
        self
    }

    #[must_use]
    pub fn with_max_parallel_loads(mut self, n: usize) -> Self {
        self.max_parallel_loads = n.max(1);
        self
    }

    #[must_use]
    pub fn with_be_url(mut self, url: impl Into<String>) -> Self {
        self.be_url = Some(url.into());
        self
    }

    #[must_use]
    pub fn with_two_phase_commit(mut self, on: bool) -> Self {
        self.two_phase_commit = on;
        self
    }

    #[must_use]
    pub fn with_sequence(mut self, on: bool) -> Self {
        self.sequence = on;
        self
    }

    #[must_use]
    pub fn with_key_columns(mut self, keys: Vec<String>) -> Self {
        self.key_columns = keys;
        self
    }

    fn two_phase_url(&self, base: &str) -> String {
        format!(
            "{}/api/{}/{}/_stream_load_2pc",
            base.trim_end_matches('/'),
            self.db,
            self.table
        )
    }

    fn stream_load_url(&self, base: &str) -> String {
        format!(
            "{}/api/{}/{}/_stream_load",
            base.trim_end_matches('/'),
            self.db,
            self.table
        )
    }

    fn schema_url(&self) -> String {
        format!(
            "{}/api/{}/{}/_schema",
            self.fe_url.trim_end_matches('/'),
            self.db,
            self.table
        )
    }

    /// HTTP Basic, precomputed. Doris Stream Load takes credentials this way.
    fn basic_auth(&self) -> String {
        format!(
            "Basic {}",
            base64_std(format!("{}:{}", self.user, self.password).as_bytes())
        )
    }
}

// ── the sink ────────────────────────────────────────────────────────────────

/// A [`TableSink`] that delivers CDC windows into a Doris table over Stream
/// Load.
pub struct DorisSink<T: StreamLoadTransport> {
    cfg: DorisConfig,
    transport: T,
    probed: bool,
}

impl<T: StreamLoadTransport> DorisSink<T> {
    pub fn new(cfg: DorisConfig, transport: T) -> Self {
        Self {
            cfg,
            transport,
            probed: false,
        }
    }

    pub fn config(&self) -> &DorisConfig {
        &self.cfg
    }

    /// Column names the destination table actually has, from the FE's
    /// `_schema` endpoint.
    async fn remote_columns(&self) -> Result<BTreeSet<String>> {
        let req = StreamLoadRequest {
            method: "GET",
            url: self.cfg.schema_url(),
            headers: vec![("Authorization".into(), self.cfg.basic_auth())],
            body: bytes::Bytes::new(),
        };
        let resp = self.transport.send(&req).await?;
        if resp.status != 200 {
            return Err(BifrostError::Other(format!(
                "doris: schema probe of {}.{} returned HTTP {} — {}",
                self.cfg.db, self.cfg.table, resp.status, resp.body
            )));
        }
        let v: serde_json::Value = serde_json::from_str(&resp.body).map_err(|e| {
            BifrostError::Other(format!("doris: schema probe returned non-JSON: {e}"))
        })?;
        let props = v
            .get("data")
            .and_then(|d| d.get("properties"))
            .and_then(serde_json::Value::as_array)
            .ok_or_else(|| {
                BifrostError::Other(format!(
                    "doris: schema probe of {}.{} has no data.properties — {}",
                    self.cfg.db, self.cfg.table, resp.body
                ))
            })?;
        Ok(props
            .iter()
            .filter_map(|p| p.get("name").and_then(serde_json::Value::as_str))
            .map(str::to_owned)
            .collect())
    }

    /// One HTTP exchange against the FE (or the pinned BE), following the
    /// FE→BE redirect by hand so an unreachable backend is reported precisely.
    async fn exchange(&self, url: String, headers: Vec<(String, String)>, body: bytes::Bytes) -> Result<StreamLoadResponse> {
        let start = url.clone();
        let mut url = url;
        let mut hops = 0_u8;
        loop {
            let req = StreamLoadRequest {
                method: "PUT",
                url: url.clone(),
                headers: headers.clone(),
                body: body.clone(),
            };
            let resp = self.transport.send(&req).await.map_err(|e| {
                if hops > 0 {
                    BifrostError::Other(format!(
                        "doris: the FE redirected this load to {url} and the request did not \
                         complete — this is a NETWORK REACHABILITY failure. Measured \
                         behaviour: an unreachable BE does not refuse the connection, it \
                         HANGS until the timeout ({}s connect / {}s request), while the FE \
                         keeps answering /api/health with 200. The loader must reach the FE \
                         ({}) AND every BE it can be redirected to on :8040. Underlying \
                         error: {e}",
                        self.cfg.connect_timeout.as_secs(),
                        self.cfg.request_timeout.as_secs(),
                        self.cfg.fe_url
                    ))
                } else {
                    e
                }
            })?;
            if !resp.is_redirect() {
                return Ok(resp);
            }
            hops += 1;
            if hops > 4 {
                return Err(BifrostError::Other(format!("doris: more than 4 redirects starting at {start}")));
            }
            url = resp.location.clone().expect("is_redirect checked location");
            tracing::debug!(
                target: "bifrost::doris",
                event = "stream_load_redirect",
                to = %url,
                "FE redirected the request to a backend",
            );
        }
    }

    /// PUT one already-encoded chunk of `load`.
    async fn put_chunk(&self, label: &str, load: &PlannedLoad, body: bytes::Bytes) -> Result<LoadOutcome> {
        let base = self.cfg.be_url.as_deref().unwrap_or(&self.cfg.fe_url);
        let mut headers = vec![
            // The body's columns, by name. MEASURED on Doris 4.1.4 (loki rig): an
            // arrow body carrying only the key (a delete) without this header fails
            // in the BE with "insert_range_from … out of bound (data.size() = 0)".
            ("columns".to_string(), load.columns.clone()),
            ("Authorization".to_string(), self.cfg.basic_auth()),
            // MANDATORY, and measured: without it Doris answers
            // {"status":"FAILED","msg":"There is no 100-continue header"}.
            // curl omits it for small bodies, so leaving it out makes SMALL
            // windows fail while large ones succeed — a size-dependent bug that
            // would look like data corruption.
            ("Expect".to_string(), "100-continue".to_string()),
            ("label".to_string(), label.to_string()),
            ("format".to_string(), self.cfg.format.header_value().to_string()),
            ("merge_type".to_string(), load.merge_type.to_string()),
        ];
        if let Some(cond) = load.delete_condition {
            // Doris batch-delete manual, "Load Merge Methods": MERGE "must be used
            // together with" a delete condition; rows matching it are deleted
            // (`__DORIS_DELETE_SIGN__` set), the rest appended.
            headers.push(("delete".to_string(), cond.to_string()));
        }
        if load.sequenced {
            // Doris stream-load manual, header `function_column.sequence_col`:
            // within one key, the row with the larger source sequence wins —
            // for deletes too ("the Sequence column value is greater than or
            // equal to the current value").
            headers.push(("function_column.sequence_col".to_string(), SEQ_COLUMN.to_string()));
        }
        if self.cfg.two_phase_commit {
            // Doris transaction docs, "Stream Load 2PC": pre-commit only; the
            // data stays invisible until `_stream_load_2pc` commits it.
            headers.push(("two_phase_commit".to_string(), "true".to_string()));
        }
        if self.cfg.format == LoadFormat::Json {
            headers.push(("strip_outer_array".to_string(), "true".to_string()));
        }
        // A dropped row is the worst failure a change feed can have, so do not
        // let Doris silently filter any: any unparseable row fails the load.
        headers.push(("max_filter_ratio".to_string(), "0".to_string()));

        let resp = match self.exchange(self.cfg.stream_load_url(base), headers, body).await {
            Ok(resp) => resp,
            // MEASURED on Doris 4.1.4.1 (live drill, 2026-10-02): a PUT whose label
            // already exists is answered `Label Already Exists` at once and the BE
            // closes the connection while the body is still uploading — with or
            // without `Expect: 100-continue` (curl: "(55) Send failure: Broken
            // pipe", reply printed anyway). reqwest surfaces that as a send error
            // and the reply is lost, so EVERY replay after a crash mid-2PC failed
            // as a "network reachability" error and the window could never be
            // delivered. A failed PUT therefore asks the FE what it holds under the
            // label before calling it a failure.
            Err(send_err) => {
                return match self.label_state(label).await {
                    Ok(state) => match state.as_str() {
                        "PRECOMMITTED" => Ok(LoadOutcome::AlreadyPending),
                        "COMMITTED" | "VISIBLE" => Ok(LoadOutcome::AlreadyFinished),
                        _ => Err(send_err),
                    },
                    Err(_) => Err(send_err),
                };
            }
        };
        if resp.status != 200 {
            return Err(BifrostError::Other(format!(
                "doris: stream load {label} returned HTTP {} — {}",
                resp.status, resp.body
            )));
        }
        parse_load_outcome(label, &resp.body)
    }

    /// The state Doris holds under `label`, from the FE's
    /// `GET /api/{db}/get_load_state?label=…`: `UNKNOWN`, `PREPARE`,
    /// `PRECOMMITTED`, `COMMITTED`, `VISIBLE` or `ABORTED`. Measured reply
    /// (Doris 4.1.4.1): `{"msg":"success","code":0,"data":"PRECOMMITTED","count":0}`.
    async fn label_state(&self, label: &str) -> Result<String> {
        let encoded: String = label
            .bytes()
            .map(|b| match b {
                b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => (b as char).to_string(),
                _ => format!("%{b:02X}"),
            })
            .collect();
        let req = StreamLoadRequest {
            method: "GET",
            url: format!(
                "{}/api/{}/get_load_state?label={encoded}",
                self.cfg.fe_url.trim_end_matches('/'),
                self.cfg.db
            ),
            headers: vec![("Authorization".to_string(), self.cfg.basic_auth())],
            body: bytes::Bytes::new(),
        };
        let resp = self.transport.send(&req).await?;
        let v: serde_json::Value = serde_json::from_str(&resp.body).map_err(|e| {
            BifrostError::Other(format!("doris: get_load_state for {label} returned non-JSON: {e} — {}", resp.body))
        })?;
        match (v.get("code").and_then(serde_json::Value::as_i64), v.get("data").and_then(serde_json::Value::as_str)) {
            (Some(0), Some(state)) => Ok(state.to_ascii_uppercase()),
            _ => Err(BifrostError::Other(format!("doris: get_load_state for {label} failed: {}", resp.body))),
        }
    }

    /// Commit or abort a pre-committed load by its label (`_stream_load_2pc`,
    /// Doris transaction docs "Stream Load 2PC" — "the transaction can also be
    /// specified by label"). By label, not txn id, so a commit lost in a crash is
    /// re-issued on the replay without having to remember any id.
    async fn txn_op(&self, label: &str, op: &str) -> Result<()> {
        let base = self.cfg.be_url.as_deref().unwrap_or(&self.cfg.fe_url);
        let headers = vec![
            ("Authorization".to_string(), self.cfg.basic_auth()),
            ("Expect".to_string(), "100-continue".to_string()),
            ("label".to_string(), label.to_string()),
            ("txn_operation".to_string(), op.to_string()),
        ];
        let resp = self.exchange(self.cfg.two_phase_url(base), headers, bytes::Bytes::new()).await;
        let result = match resp {
            Ok(resp) if resp.status != 200 => Err(BifrostError::Other(format!(
                "doris: 2pc {op} of {label} returned HTTP {} — {}",
                resp.status, resp.body
            ))),
            Ok(resp) => parse_txn_result(label, op, &resp.body),
            Err(e) => Err(e),
        };
        // MEASURED on Doris 4.1.4.1 (live drill, 2026-10-02): re-committing a
        // VISIBLE label, or aborting one twice, answers `{"status":
        // "ANALYSIS_ERROR","msg":"… running transaction not found, label=…"}` — no
        // "already" anywhere, so the reply text cannot tell a replayed commit from
        // a lost one. The label's state can: ask the FE before failing.
        match result {
            Ok(()) => Ok(()),
            Err(e) => match self.label_state(label).await.as_deref() {
                Ok("COMMITTED" | "VISIBLE") if op == "commit" => {
                    tracing::info!(target: "bifrost::doris", event = "commit_replayed", label = %label, "already committed");
                    Ok(())
                }
                Ok("ABORTED" | "UNKNOWN") if op == "abort" => Ok(()),
                _ => Err(e),
            },
        }
    }

    /// Split a window into the Stream Loads it takes. See [`DorisSink::apply`].
    fn plan_loads(&self, window: &TableWindow<'_>) -> Result<Vec<PlannedLoad>> {
        if self.cfg.mode == LoadMode::Append {
            return Ok(if window.upsert_rows() > 0 {
                vec![PlannedLoad::new("APPEND", None, false, window.upserts.to_vec())]
            } else {
                Vec::new()
            });
        }
        let keys: &[String] = if self.cfg.key_columns.is_empty() { window.key_columns } else { &self.cfg.key_columns };
        let seq = if self.cfg.sequence {
            Some(window.sequence.ok_or_else(|| {
                BifrostError::Other(format!(
                    "doris: {}: the window carries no sequence number but the sink versions rows by \
                     it (DorisConfig::sequence). Deliver one snapshot per window, or turn the \
                     sequence off for a table without a sequence column.",
                    window.table
                ))
            })?)
        } else {
            None
        };

        // A key deleted AND upserted in the same snapshot is an update: ship only
        // the upsert. Inside one load there is no defined order between two rows
        // of one key with the same sequence value, so the pair must not both go.
        let deletes: Vec<RecordBatch> = if window.delete_rows() > 0 && window.upsert_rows() > 0 {
            if keys.is_empty() {
                return Err(BifrostError::Other(format!(
                    "doris: {}: this window deletes and upserts rows but no key columns are known, \
                     so an update (delete + insert of one key) cannot be told from two changes. \
                     Name them (--key / DorisConfig::with_key_columns) or give the Iceberg table \
                     identifier fields.",
                    window.table
                )));
            }
            drop_upserted_keys(window.deletes, window.upserts, keys)?
        } else {
            window.deletes.to_vec()
        };

        // Whole-row deletes (deletion vectors, position deletes, copy-on-write
        // pre-images) ride in the MERGE body beside the upserts. Identity-only
        // deletes (equality deletes: every non-key column null) would put NULLs
        // into NOT NULL value columns, so they go as a keys-only DELETE load.
        let (whole, identity) = split_identity_deletes(deletes, keys)?;
        let mut loads = Vec::new();
        if !identity.is_empty() {
            let projected = identity
                .iter()
                .map(|b| project_named(b, keys))
                .collect::<Result<Vec<_>>>()?;
            let tagged = projected.iter().map(|b| tag(b, None, seq)).collect::<Result<Vec<_>>>()?;
            loads.push(PlannedLoad::new("DELETE", None, seq.is_some(), tagged));
        }
        if window.upsert_rows() > 0 || !whole.is_empty() {
            let mut body = Vec::new();
            for b in window.upserts {
                body.push(tag(b, Some(0), seq)?);
            }
            for b in &whole {
                body.push(tag(b, Some(1), seq)?);
            }
            let body = unify(body)?;
            loads.push(PlannedLoad::new("MERGE", Some(DELETE_CONDITION), seq.is_some(), body));
        }
        Ok(loads)
    }
}

/// The per-row op column of a MERGE body: `1` = delete, `0` = upsert. A load
/// column that is not a table column — Doris treats it as an intermediate
/// column, usable in the `delete` condition and never stored.
pub const OP_COLUMN: &str = "__bifrost_op";
/// The per-row version column, mapped by `function_column.sequence_col`.
pub const SEQ_COLUMN: &str = "__bifrost_seq";
const DELETE_CONDITION: &str = "__bifrost_op=1";

/// One Stream Load a window takes (possibly several chunks, each its own label).
struct PlannedLoad {
    merge_type: &'static str,
    delete_condition: Option<&'static str>,
    sequenced: bool,
    columns: String,
    batches: Vec<RecordBatch>,
}

impl PlannedLoad {
    fn new(merge_type: &'static str, delete_condition: Option<&'static str>, sequenced: bool, batches: Vec<RecordBatch>) -> Self {
        let columns = batches
            .first()
            .map(|b| b.schema().fields().iter().map(|f| f.name().as_str()).collect::<Vec<_>>().join(","))
            .unwrap_or_default();
        Self { merge_type, delete_condition, sequenced, columns, batches }
    }

    fn rows(&self) -> usize {
        self.batches.iter().map(RecordBatch::num_rows).sum()
    }
}

/// What one chunk's PUT came back as.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LoadOutcome {
    /// Loaded now (pre-committed under 2PC, visible otherwise).
    Loaded { rows: usize },
    /// `Label Already Exists`, `ExistingJobStatus: FINISHED` — the replay of a
    /// chunk that was committed before. Nothing to do.
    AlreadyFinished,
    /// `Label Already Exists` with the job not finished — under 2PC, a chunk
    /// pre-committed by an earlier attempt whose commit never happened. Commit it.
    AlreadyPending,
}

/// `rows` with the op and sequence columns appended (either may be absent).
fn tag(b: &RecordBatch, op: Option<i8>, seq: Option<i64>) -> Result<RecordBatch> {
    use arrow_array::{ArrayRef, Int8Array, Int64Array};
    use std::sync::Arc;
    let mut fields: Vec<arrow_schema::FieldRef> = b.schema().fields().iter().cloned().collect();
    let mut cols: Vec<ArrayRef> = b.columns().to_vec();
    if let Some(op) = op {
        fields.push(Arc::new(arrow_schema::Field::new(OP_COLUMN, arrow_schema::DataType::Int8, false)));
        cols.push(Arc::new(Int8Array::from(vec![op; b.num_rows()])));
    }
    if let Some(seq) = seq {
        fields.push(Arc::new(arrow_schema::Field::new(SEQ_COLUMN, arrow_schema::DataType::Int64, false)));
        cols.push(Arc::new(Int64Array::from(vec![seq; b.num_rows()])));
    }
    Ok(RecordBatch::try_new(Arc::new(Schema::new(fields)), cols)?)
}

/// Every batch of one body on ONE schema (an Arrow IPC stream has a single
/// schema): the first batch's column order, every column nullable, each batch
/// projected by name and cast where its type differs.
fn unify(batches: Vec<RecordBatch>) -> Result<Vec<RecordBatch>> {
    use std::sync::Arc;
    let Some(first) = batches.first() else { return Ok(batches) };
    let fields: Vec<arrow_schema::Field> =
        first.schema().fields().iter().map(|f| f.as_ref().clone().with_nullable(true)).collect();
    let schema = Arc::new(Schema::new(fields));
    batches
        .iter()
        .map(|b| {
            let cols = schema
                .fields()
                .iter()
                .map(|f| {
                    let c = b.column_by_name(f.name()).ok_or_else(|| {
                        BifrostError::Other(format!("doris: a delete row has no column {:?} the upserts carry", f.name()))
                    })?;
                    if c.data_type() == f.data_type() {
                        Ok(Arc::clone(c))
                    } else {
                        arrow::compute::cast(c, f.data_type()).map_err(BifrostError::from)
                    }
                })
                .collect::<Result<Vec<_>>>()?;
            Ok(RecordBatch::try_new(Arc::clone(&schema), cols)?)
        })
        .collect()
}

/// `b` projected to the named columns, in that order.
fn project_named(b: &RecordBatch, names: &[String]) -> Result<RecordBatch> {
    let idx = names
        .iter()
        .map(|k| {
            b.schema()
                .index_of(k)
                .map_err(|_| BifrostError::Source(format!("delete rows have no identity column {k:?}")))
        })
        .collect::<Result<Vec<_>>>()?;
    b.project(&idx).map_err(|e| BifrostError::Source(format!("projecting delete keys: {e}")))
}

/// Split deletes into whole-row ones and identity-only ones (a non-key column
/// holds a null — what an equality delete widened to the table schema looks
/// like). With no key known every delete is treated as whole-row.
fn split_identity_deletes(deletes: Vec<RecordBatch>, keys: &[String]) -> Result<(Vec<RecordBatch>, Vec<RecordBatch>)> {
    use arrow_array::Array as _;
    if keys.is_empty() {
        return Ok((deletes, Vec::new()));
    }
    let mut whole = Vec::new();
    let mut identity = Vec::new();
    for b in deletes {
        if b.num_rows() == 0 {
            continue;
        }
        let schema = b.schema();
        let partial = schema
            .fields()
            .iter()
            .enumerate()
            .any(|(i, f)| !keys.iter().any(|k| k == f.name()) && b.column(i).null_count() > 0);
        if partial { identity.push(b) } else { whole.push(b) }
    }
    Ok((whole, identity))
}

/// `deletes` minus every row whose key also appears in `upserts`.
fn drop_upserted_keys(deletes: &[RecordBatch], upserts: &[RecordBatch], keys: &[String]) -> Result<Vec<RecordBatch>> {
    use arrow::row::{RowConverter, SortField};
    let Some(first) = upserts.first().or_else(|| deletes.first()) else { return Ok(Vec::new()) };
    let fields = keys
        .iter()
        .map(|k| {
            first
                .schema()
                .field_with_name(k)
                .map(|f| SortField::new(f.data_type().clone()))
                .map_err(|_| BifrostError::Source(format!("no key column {k:?} in the window")))
        })
        .collect::<Result<Vec<_>>>()?;
    let conv = RowConverter::new(fields)?;
    let key_cols = |b: &RecordBatch| -> Result<Vec<arrow_array::ArrayRef>> {
        Ok(project_named(b, keys)?.columns().to_vec())
    };
    let mut upserted = std::collections::HashSet::new();
    for b in upserts {
        let rows = conv.convert_columns(&key_cols(b)?)?;
        for r in rows.iter() {
            upserted.insert(r.as_ref().to_vec());
        }
    }
    let mut out = Vec::new();
    for b in deletes {
        let rows = conv.convert_columns(&key_cols(b)?)?;
        let keep: arrow_array::BooleanArray = rows.iter().map(|r| Some(!upserted.contains(r.as_ref()))).collect();
        let f = arrow::compute::filter_record_batch(b, &keep)?;
        if f.num_rows() > 0 {
            out.push(f);
        }
    }
    Ok(out)
}

#[async_trait]
impl<T: StreamLoadTransport> TableSink for DorisSink<T> {
    /// Refuse to start on a schema the destination cannot accept.
    ///
    /// The check is "every source column exists in the destination". A
    /// destination column the source does not send is fine (Doris fills its
    /// default); a source column the destination does not have is a load that
    /// would fail row by row, halfway through.
    async fn probe(&mut self, schema: &Schema) -> Result<()> {
        let remote = self.remote_columns().await?;
        let missing: Vec<&str> = schema
            .fields()
            .iter()
            .map(|f| f.name().as_str())
            .filter(|n| !remote.contains(*n))
            .collect();
        let ok = missing.is_empty();
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "doris_sink::schema_probe",
            ok,
            &format!(
                "{}.{}: {} source column(s), {} missing remotely",
                self.cfg.db,
                self.cfg.table,
                schema.fields().len(),
                missing.len()
            ),
        );
        if !ok {
            return Err(BifrostError::Other(format!(
                "doris: {}.{} is missing column(s) the source sends: {} — refusing to \
                 start. Discovering this mid-stream, after a partial load, is worse.",
                self.cfg.db,
                self.cfg.table,
                missing.join(", ")
            )));
        }
        self.probed = true;
        Ok(())
    }

    async fn apply(&mut self, window: &TableWindow<'_>, token: AckToken) -> Result<LoadAck> {
        // Append mode means the source is append-only. If it is not, say so.
        if self.cfg.mode == LoadMode::Append && window.delete_rows() > 0 {
            crate::testmatrix::emit(
                crate::testmatrix::COMPONENT,
                "doris_sink::append_mode_refuses_deletes",
                false,
                &format!(
                    "{}: {} delete row(s) in Append mode",
                    window.table,
                    window.delete_rows()
                ),
            );
            return Err(BifrostError::Other(format!(
                "doris: {} delete row(s) arrived for {}.{} in Append mode — the source is \
                 not append-only after all. Refusing rather than silently diverging from \
                 the lake. Use a unique-key table with LoadMode::Upsert, or fix the writer.",
                window.delete_rows(),
                self.cfg.db,
                self.cfg.table
            )));
        }

        let loads = self.plan_loads(window)?;
        // Whole-row deletes ride in the MERGE body, where Doris reports one
        // NumberLoadedRows for upserts and deletes together; the split is taken
        // from what was sent.
        let merged_deletes: usize = loads
            .iter()
            .filter(|l| l.merge_type == "MERGE")
            .map(PlannedLoad::rows)
            .sum::<usize>()
            .saturating_sub(window.upsert_rows());

        // Encode every load into chunks; every chunk carries its own derived
        // label (window pair + running chunk index), so a replay re-derives the
        // same labels and Doris refuses whatever already landed.
        let mut jobs_in: Vec<(String, usize, bytes::Bytes)> = Vec::new();
        let mut chunk_rows: Vec<usize> = Vec::new();
        for (li, load) in loads.iter().enumerate() {
            for (body, rows) in encode_chunks(&load.batches, self.cfg.max_chunk_bytes, self.cfg.format)? {
                chunk_rows.push(rows);
                let label = stream_load_label(
                    &self.cfg.db,
                    &self.cfg.table,
                    window.snapshots.from,
                    window.snapshots.to,
                    jobs_in.len(),
                );
                jobs_in.push((label, li, bytes::Bytes::from(body)));
            }
        }

        // Phase 1: every chunk through the ONE async I/O engine, up to
        // `max_parallel_loads` in flight, results re-sequenced into chunk order.
        let loads_ref = &loads;
        let this: &Self = &*self;
        let jobs = jobs_in.into_iter().map(move |(label, li, body)| {
            let load = &loads_ref[li];
            async move {
                let r = this.put_chunk(&label, load, body).await;
                Ok::<_, anyhow::Error>((label, li, r))
            }
        });
        let done = znippy_zoomies::gatling::io::run_ordered(jobs, self.cfg.max_parallel_loads.max(1), |x| x)
            .await
            .map_err(|e| BifrostError::Other(format!("{e:#}")))?;

        let mut labels = Vec::with_capacity(done.len());
        let mut to_commit = Vec::new();
        let mut fresh_precommitted = Vec::new();
        let mut first_err = None;
        let (mut loaded, mut deleted, mut duplicate) = (0usize, 0usize, 0usize);
        for (ci, (label, li, r)) in done.into_iter().enumerate() {
            match r {
                Ok(LoadOutcome::Loaded { rows }) => {
                    if loads[li].merge_type == "DELETE" {
                        deleted += rows;
                    } else {
                        loaded += rows;
                    }
                    if self.cfg.two_phase_commit {
                        to_commit.push(label.clone());
                        fresh_precommitted.push(label.clone());
                    }
                }
                Ok(LoadOutcome::AlreadyFinished) => {
                    tracing::info!(
                        target: "bifrost::doris",
                        event = "duplicate_label_rejected",
                        label = %label,
                        "replay rejected by label — chunk already loaded",
                    );
                    duplicate += chunk_rows[ci];
                }
                Ok(LoadOutcome::AlreadyPending) => {
                    if !self.cfg.two_phase_commit {
                        // Without 2PC a running job is the earlier attempt still in
                        // flight; its outcome is unknown, so this window is not done.
                        first_err.get_or_insert_with(|| BifrostError::Other(format!(
                            "doris: label {label} is still RUNNING from an earlier attempt — retry later"
                        )));
                    } else {
                        to_commit.push(label.clone());
                    }
                }
                Err(e) => {
                    first_err.get_or_insert(e);
                }
            }
            labels.push(label);
        }

        // A failed chunk fails the window. Under 2PC the chunks this attempt
        // pre-committed are aborted (best effort) so nothing of the window
        // becomes visible; their labels are free again for the replay.
        if let Some(e) = first_err {
            for label in &fresh_precommitted {
                if let Err(abort) = self.txn_op(label, "abort").await {
                    tracing::warn!(target: "bifrost::doris", event = "abort_failed", label = %label, error = %abort,
                        "could not abort a pre-committed chunk; the replay will commit it by label");
                }
            }
            return Err(e);
        }

        // Phase 2: commit. A commit that fails leaves the window un-acked: the
        // replay finds committed chunks FINISHED and the rest pending, and
        // commits those — it converges without ever loading a row twice.
        for label in &to_commit {
            self.txn_op(label, "commit").await?;
        }

        tracing::info!(
            target: "bifrost::doris",
            event = "stream_load_window",
            table = %window.table,
            dest = %format!("{}.{}", self.cfg.db, self.cfg.table),
            from = ?window.snapshots.from,
            to = window.snapshots.to,
            sequence = ?window.sequence,
            chunks = labels.len(),
            committed = to_commit.len(),
            loaded,
            deleted,
            duplicate,
            "stream load complete",
        );

        // Move the MERGE body's delete rows out of `loaded` (exact when the
        // window loaded fresh; on a partial replay the duplicate count is by
        // whole chunk and the split is approximate).
        let moved = merged_deletes.min(loaded);
        loaded -= moved;
        deleted += moved;
        Ok(token.ack(window.snapshots.to, labels).with_counts(loaded, deleted, duplicate))
    }
}

// ── Stream Load result parsing ──────────────────────────────────────────────

/// Interpret a Stream Load response body.
///
/// `Label Already Exists` is **not** an error: it is the replay path working as
/// designed. The window was already loaded under this exact label, so the
/// correct action is to carry on and let the cursor advance.
///
/// Doris's own wording for the label guarantee is *At-Most-Once*, not
/// exactly-once; combined with a cursor that only advances on success (see
/// [`crate::sink::table::deliver_window`]) that is what yields effectively-once.
pub fn parse_load_result(label: &str, body: &str) -> Result<(usize, bool)> {
    Ok(match parse_load_outcome(label, body)? {
        LoadOutcome::Loaded { rows } => (rows, false),
        LoadOutcome::AlreadyFinished | LoadOutcome::AlreadyPending => (0, true),
    })
}

/// [`parse_load_result`], keeping what a 2PC replay needs: whether a label that
/// already exists belongs to a FINISHED (committed) job or to one still open —
/// under 2PC, a chunk an earlier attempt pre-committed and never committed.
/// Doris stream-load manual, `ExistingJobStatus`: "shown only when `Status` is
/// `Label Already Exists` … `RUNNING` means the job is still running;
/// `FINISHED` means the job succeeded." MEASURED on Doris 4.1.4.1: a 2PC chunk
/// pre-committed and never committed reports `PRECOMMITTED` (not `RUNNING`), a
/// committed one `FINISHED` — anything but `FINISHED` is treated as open.
pub fn parse_load_outcome(label: &str, body: &str) -> Result<LoadOutcome> {
    let v: serde_json::Value = serde_json::from_str(body).map_err(|e| {
        BifrostError::Other(format!("doris: stream load {label} returned non-JSON: {e} — {body}"))
    })?;
    if v.get("Status").and_then(serde_json::Value::as_str) == Some("Label Already Exists") {
        let existing = v.get("ExistingJobStatus").and_then(serde_json::Value::as_str).unwrap_or("FINISHED");
        return Ok(if existing.eq_ignore_ascii_case("FINISHED") {
            LoadOutcome::AlreadyFinished
        } else {
            LoadOutcome::AlreadyPending
        });
    }
    let (rows, _) = parse_fresh_load(label, &v)?;
    Ok(LoadOutcome::Loaded { rows })
}

/// Interpret a `_stream_load_2pc` reply (`{"status":"Success","msg":…}`). A
/// commit of a transaction that is already committed/visible is the replay of a
/// commit that landed before the crash — success, not an error.
pub fn parse_txn_result(label: &str, op: &str, body: &str) -> Result<()> {
    let v: serde_json::Value = serde_json::from_str(body).map_err(|e| {
        BifrostError::Other(format!("doris: 2pc {op} of {label} returned non-JSON: {e} — {body}"))
    })?;
    // Documented as `status` / `msg`; accept any capitalisation (the load
    // endpoint itself answers `Status` / `Message`).
    let field = |k: &str| {
        v.as_object()
            .and_then(|o| o.iter().find(|(name, _)| name.eq_ignore_ascii_case(k)))
            .and_then(|(_, x)| x.as_str())
            .unwrap_or("")
            .to_string()
    };
    let status = field("status");
    let msg = { let m = field("msg"); if m.is_empty() { field("message") } else { m } };
    if status.eq_ignore_ascii_case("success") {
        return Ok(());
    }
    let m = msg.to_ascii_lowercase();
    if op == "commit" && m.contains("already") && (m.contains("visible") || m.contains("committed")) {
        tracing::info!(target: "bifrost::doris", event = "commit_replayed", label = %label, "already committed");
        return Ok(());
    }
    if op == "abort" && m.contains("already") && m.contains("abort") {
        return Ok(());
    }
    Err(BifrostError::Other(format!("doris: 2pc {op} of {label} failed: {status} — {msg}")))
}

/// The pre-2PC body of [`parse_load_result`] for a fresh (non-replay) load.
fn parse_fresh_load(label: &str, v: &serde_json::Value) -> Result<(usize, bool)> {
    let status = v
        .get("Status")
        .and_then(serde_json::Value::as_str)
        .unwrap_or("");
    let msg = v
        .get("Message")
        .and_then(serde_json::Value::as_str)
        .unwrap_or("");
    let loaded = v
        .get("NumberLoadedRows")
        .and_then(serde_json::Value::as_u64)
        .unwrap_or(0) as usize;

    match status {
        "Success" | "Publish Timeout" => {
            let filtered = v
                .get("NumberFilteredRows")
                .and_then(serde_json::Value::as_u64)
                .unwrap_or(0);
            if filtered > 0 {
                return Err(BifrostError::Other(format!(
                    "doris: stream load {label} filtered {filtered} row(s) — refusing. A \
                     dropped row is the worst failure a change feed can have. {}",
                    v.get("ErrorURL")
                        .and_then(serde_json::Value::as_str)
                        .unwrap_or("")
                )));
            }
            Ok((loaded, false))
        }
        // The replay path. Do NOT add a second dedup layer on top of this.
        "Label Already Exists" => {
            tracing::info!(
                target: "bifrost::doris",
                event = "duplicate_label_rejected",
                label = %label,
                "replay rejected by label — window already loaded",
            );
            Ok((0, true))
        }
        other => Err(BifrostError::Other(format!(
            "doris: stream load {label} failed with status `{other}` — {msg}. {}",
            v.get("ErrorURL")
                .and_then(serde_json::Value::as_str)
                .unwrap_or("")
        ))),
    }
}

// ── encoding ────────────────────────────────────────────────────────────────

/// Encode `batches` as JSON arrays, split so no single body exceeds `budget`.
///
/// A batch that alone exceeds the budget is halved recursively rather than
/// guessed at from an average row width, so a table with a few enormous rows
/// still produces bodies under the cap.
fn encode_chunks(batches: &[RecordBatch], budget: usize, format: LoadFormat) -> Result<Vec<(Vec<u8>, usize)>> {
    let mut out: Vec<(Vec<u8>, usize)> = Vec::new();
    let mut pending: Vec<RecordBatch> = Vec::new();
    let mut pending_bytes = 0usize;
    let rows = |p: &[RecordBatch]| p.iter().map(RecordBatch::num_rows).sum::<usize>();

    for batch in batches {
        if batch.num_rows() == 0 {
            continue;
        }
        let encoded = encode(format, std::slice::from_ref(batch))?;
        if encoded.len() > budget {
            // Flush what we have, then split this one.
            if !pending.is_empty() {
                out.push((encode(format, &pending)?, rows(&pending)));
                pending.clear();
                pending_bytes = 0;
            }
            split_encode(batch, budget, format, &mut out)?;
            continue;
        }
        if pending_bytes + encoded.len() > budget && !pending.is_empty() {
            out.push((encode(format, &pending)?, rows(&pending)));
            pending.clear();
            pending_bytes = 0;
        }
        pending_bytes += encoded.len();
        pending.push(batch.clone());
    }
    if !pending.is_empty() {
        out.push((encode(format, &pending)?, rows(&pending)));
    }
    Ok(out)
}

/// Halve a too-large batch until each half encodes under `budget`. A
/// single-row batch that still exceeds it is emitted anyway — refusing would
/// drop data, and Doris's own cap is far above any sane single row.
fn split_encode(batch: &RecordBatch, budget: usize, format: LoadFormat, out: &mut Vec<(Vec<u8>, usize)>) -> Result<()> {
    let encoded = encode(format, std::slice::from_ref(batch))?;
    if encoded.len() <= budget || batch.num_rows() <= 1 {
        out.push((encoded, batch.num_rows()));
        return Ok(());
    }
    let half = batch.num_rows() / 2;
    split_encode(&batch.slice(0, half), budget, format, out)?;
    split_encode(&batch.slice(half, batch.num_rows() - half), budget, format, out)
}

/// One Stream Load body in `format`.
fn encode(format: LoadFormat, batches: &[RecordBatch]) -> Result<Vec<u8>> {
    match format {
        LoadFormat::Json => encode_json(batches),
        LoadFormat::Arrow => encode_arrow(batches),
    }
}

/// Arrow → one Arrow IPC **stream** (schema message, then the batches), which
/// Stream Load reads with `format: arrow`. No text round trip: the columns go
/// as they are.
fn encode_arrow(batches: &[RecordBatch]) -> Result<Vec<u8>> {
    let Some(first) = batches.first() else { return Ok(Vec::new()) };
    let bytes: usize = batches.iter().map(RecordBatch::get_array_memory_size).sum();
    let mut writer = arrow::ipc::writer::StreamWriter::try_new(Vec::with_capacity(bytes), &first.schema())?;
    for b in batches {
        writer.write(b)?;
    }
    writer.finish()?;
    Ok(writer.into_inner()?)
}

/// Arrow → a single JSON array, which Stream Load reads with
/// `strip_outer_array: true`.
fn encode_json(batches: &[RecordBatch]) -> Result<Vec<u8>> {
    let mut writer = arrow::json::ArrayWriter::new(Vec::new());
    for b in batches {
        writer.write(b)?;
    }
    writer.finish()?;
    Ok(writer.into_inner())
}

/// Standard base64, for the HTTP Basic header. Twelve lines is cheaper than a
/// dependency, and this is the only place the crate needs it.
fn base64_std(input: &[u8]) -> String {
    const T: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
    let mut out = String::with_capacity(input.len().div_ceil(3) * 4);
    for c in input.chunks(3) {
        let b = [c[0], *c.get(1).unwrap_or(&0), *c.get(2).unwrap_or(&0)];
        let n = (u32::from(b[0]) << 16) | (u32::from(b[1]) << 8) | u32::from(b[2]);
        out.push(T[(n >> 18 & 63) as usize] as char);
        out.push(T[(n >> 12 & 63) as usize] as char);
        out.push(if c.len() > 1 {
            T[(n >> 6 & 63) as usize] as char
        } else {
            '='
        });
        out.push(if c.len() > 2 {
            T[(n & 63) as usize] as char
        } else {
            '='
        });
    }
    out
}

// ── the real transport ──────────────────────────────────────────────────────

/// The live HTTP transport.
///
/// Automatic redirect following is **disabled**: [`DorisSink`] follows the
/// FE→BE 307 itself so that an unreachable backend is reported as a network
/// failure instead of as the authentication error it resembles. (It is also how
/// `curl --location-trusted` differs from plain `curl` here — a client that
/// re-authenticates on redirect appears to fail auth.)
pub struct ReqwestTransport {
    client: reqwest::Client,
}

impl ReqwestTransport {
    /// Build the client for `cfg`. The timeouts are not decoration: an
    /// unreachable BE after the 307 **hangs** rather than refusing (measured on
    /// Doris 4.0.8), so without them one routing mistake stalls the stream
    /// forever with no error.
    pub fn new(cfg: &DorisConfig) -> Result<Self> {
        let client = tls_roots(reqwest::Client::builder())
            .redirect(reqwest::redirect::Policy::none())
            .connect_timeout(cfg.connect_timeout)
            .timeout(cfg.request_timeout)
            .build()
            .map_err(|e| BifrostError::Other(format!("doris: http client: {e}")))?;
        Ok(Self { client })
    }
}

#[async_trait]
impl StreamLoadTransport for ReqwestTransport {
    async fn send(&self, req: &StreamLoadRequest) -> Result<StreamLoadResponse> {
        let method = reqwest::Method::from_bytes(req.method.as_bytes())
            .map_err(|e| BifrostError::Other(format!("doris: bad method: {e}")))?;
        let mut rb = self.client.request(method, &req.url);
        for (k, v) in &req.headers {
            rb = rb.header(k.as_str(), v.as_str());
        }
        if !req.body.is_empty() {
            rb = rb.body(req.body.clone()); // Bytes: a refcount, not a copy
        }
        let resp = rb
            .send()
            .await
            .map_err(|e| BifrostError::Other(format!("doris: {} {}: {e}", req.method, req.url)))?;
        let status = resp.status().as_u16();
        let location = resp
            .headers()
            .get(reqwest::header::LOCATION)
            .and_then(|v| v.to_str().ok())
            .map(str::to_owned);
        let body = resp
            .text()
            .await
            .map_err(|e| BifrostError::Other(format!("doris: reading response body: {e}")))?;
        Ok(StreamLoadResponse {
            status,
            location,
            body,
        })
    }
}

/// reqwest 0.13 with reqwest 0.12 `rustls-tls` behaviour (2026-10-02): the ring
/// provider and the bundled Mozilla roots (webpki), not 0.13's aws-lc-rs + OS
/// trust store. Roots added later on the builder are merged in, as in 0.12.
#[allow(dead_code)]
fn tls_roots(b: reqwest::ClientBuilder) -> reqwest::ClientBuilder {
    let _ = rustls::crypto::ring::default_provider().install_default();
    b.tls_certs_only(
        webpki_root_certs::TLS_SERVER_ROOT_CERTS
            .iter()
            .filter_map(|c| reqwest::Certificate::from_der(c).ok()),
    )
}