crates/knut-bifrost/src/sink/doris.rs
//! **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()),
)
}