From 549e171a4b454bd0a0e6edee7519a728b1ab45fa Mon Sep 17 00:00:00 2001
From: ryankert01 <ryankert01@gmail.com>
Date: Sun, 28 Jun 2026 03:36:45 +0800
Subject: [PATCH 1/2] feat(connectors): retry transient Doris Stream Load
 failures in-request

The Doris sink classified transient Stream Load outcomes (5xx/408/429,
transport errors, Publish Timeout) as retryable but never acted on them: the
runtime commits the consumer offset at poll time before consume() runs and
discards its return value, so a transient backend blip silently dropped the
batch under at-most-once delivery.

consume() now retries a transiently-failed batch in-request, re-PUTing under
the same deterministic label so Doris dedupes a prior attempt that actually
landed (e.g. a 2xx whose body could not be read). Permanent failures are never
retried. Backoff and jitter come from iggy_connector_sdk::retry, bounded by new
max_retries/retry_delay/max_retry_delay config (defaults 3/200ms/5s). This
shrinks the at-most-once window within a single poll; cross-poll and crash
delivery remain a runtime concern, not something a sink can fix.

Relates to #3215.
---
 .../example_config/connectors/doris_sink.toml |   7 +
 core/connectors/sinks/doris_sink/README.md    |  18 +-
 core/connectors/sinks/doris_sink/config.toml  |   7 +
 core/connectors/sinks/doris_sink/src/lib.rs   | 292 ++++++++++++++++--
 4 files changed, 285 insertions(+), 39 deletions(-)

diff --git a/core/connectors/runtime/example_config/connectors/doris_sink.toml b/core/connectors/runtime/example_config/connectors/doris_sink.toml
index 656b0c2b4f..20f4e7663e 100644
--- a/core/connectors/runtime/example_config/connectors/doris_sink.toml
+++ b/core/connectors/runtime/example_config/connectors/doris_sink.toml
@@ -44,6 +44,13 @@ timeout = "30s"
 # TCP connect timeout (default "5s"). Independent of timeout; raise it for
 # cross-region or cold-start FEs that are slow to accept the connection.
 # connect_timeout = "5s"
+# In-request retry for transient Stream Load failures (5xx/408/429, transport
+# error, Doris "Publish Timeout"). Retries re-PUT under the same label, which
+# Doris dedupes. max_retries is the total attempt count (1 disables retries);
+# keep max_retries * max_retry_delay inside Doris's label_keep_max_second.
+# max_retries = 3
+# retry_delay = "200ms"
+# max_retry_delay = "5s"
 # Stream Load redirect security. Doris's FE redirects (307) to a BE on another
 # host; credentials are re-attached across that hop. By default a redirect that
 # downgrades https -> http is refused (it would leak credentials in cleartext).
diff --git a/core/connectors/sinks/doris_sink/README.md b/core/connectors/sinks/doris_sink/README.md
index 6da9937be0..b949a50b8e 100644
--- a/core/connectors/sinks/doris_sink/README.md
+++ b/core/connectors/sinks/doris_sink/README.md
@@ -15,15 +15,16 @@ The Doris sink connector consumes JSON messages from Iggy streams and writes the
 2. It computes a deterministic Stream Load `label` of the form `{label_prefix}-{stream_san}-{topic_san}-{hash16}-{partition}-{first_offset}-{last_offset}`.
    - `hash16` is a single 64-bit blake3 hash computed over the *raw* (un-sanitized), length-prefixed `(label_prefix, stream, topic)` triple. So identities that sanitize to the same string get distinct labels — whether the collision is in the names (`events.v1` vs `events_v1`) or in two tenants' prefixes that truncate alike (`prod_events_us_east_1` vs `..._2`) — and no boundary-shift aliasing is possible (`("ab","c")` ≠ `("a","bc")`).
    - The total label is bounded under Doris's 128-char cap regardless of input length (worst case 120 chars).
-   - Doris dedupes loads by label inside its `label_keep_max_second` window. The deterministic label is **forward-compatible scaffolding**: if the runtime ever gains retry/redrive, a duplicate load would be absorbed, not doubled. **Today it protects no production scenario** — there is no retry loop, `consume()` runs once per poll, and the runtime discards its return value. Delivery is at-most-once: the offset is committed before `consume()` runs, so a failed load is never replayed.
+   - Doris dedupes loads by label inside its `label_keep_max_second` window. The in-request retry (step 6) re-PUTs a transiently-failed batch under the same label, so a prior attempt that actually landed (e.g. a `2xx` whose body we couldn't read) is absorbed, not doubled. This protects **in-request retry only**: the runtime commits the offset before `consume()` runs and discards its return, so a failure outliving the retry budget or a crash mid-load is **at-most-once**.
 3. It `PUT`s the batch to `{fe_url}/api/{database}/{table}/_stream_load` with HTTP Basic auth and the headers `Expect: 100-continue`, `format: json`, `strip_outer_array: true`, `label: <label>`. (`Expect: 100-continue` is required by Doris's Stream Load endpoint, which rejects PUTs that omit it. Where the HTTP stack negotiates the handshake it also lets Doris reject auth/4xx before the body uploads — a secondary benefit, not relied on for correctness.)
 4. The Doris frontend (FE) responds with a `307 Temporary Redirect` to a backend (BE). The connector follows the redirect manually so that the `Authorization` header is preserved across the hop (`reqwest`'s default policy strips it on cross-host redirects).
    `308 Permanent Redirect` is also followed as a defensive measure; redirects beyond a hard cap of 5 (or a redirect with no usable `Location`) are rejected as a permanent `PermanentHttpError`, since retrying a malformed/looping redirect cannot help.
 5. The HTTP body is parsed as JSON and the `Status` field decides the outcome:
    - `Success` → batch accepted.
    - `Label Already Exists` → idempotent replay, treated as success.
-   - `Publish Timeout` or HTTP `5xx`/`408`/`429` → classified as a transient error (`Error::CannotStoreData`) — retryable in principle, but per the at-most-once note above the runtime does not currently act on it.
-   - `Fail`, any other `4xx`, or an unparsable response body → permanent error (`Error::PermanentHttpError`); retrying would not help even if the runtime did redrive.
+   - `Publish Timeout` or HTTP `5xx`/`408`/`429` → transient error (`Error::CannotStoreData`): retried in-request up to `max_retries` attempts (exponential backoff + jitter) under the same label before being surfaced.
+   - `Fail`, any other `4xx`, or an unparsable response body → permanent error (`Error::PermanentHttpError`); never retried — re-PUTing bad data would just hammer the FE.
+6. A *transient* failure (the classifications above, plus a transport-level error) is retried in-request: the same batch is re-`PUT` under the same label, up to `max_retries` attempts with backoff and ±20% jitter (`iggy_connector_sdk::retry`). Since the runtime commits the offset at poll time, this is the connector's only redelivery path; once the budget is exhausted the worst error is surfaced and the batch is not retried again — **at-most-once** across polls.
 
 ## Configuration
 
@@ -38,6 +39,9 @@ The Doris sink connector consumes JSON messages from Iggy streams and writes the
 | `batch_size` | no | `1000` | Maximum number of messages per Stream Load request. |
 | `timeout` | no | `30s` | Per-request HTTP timeout (total request budget), as a human-readable duration (e.g. `30s`, `1m`). |
 | `connect_timeout` | no | `5s` | TCP connect timeout, independent of `timeout`, as a human-readable duration. Raise it for cross-region or cold-start FEs. |
+| `max_retries` | no | `3` | Total Stream Load attempts per batch on a *transient* failure (`1` disables retries). Each retry re-PUTs under the same label, which Doris dedupes. |
+| `retry_delay` | no | `200ms` | Base backoff before the first retry; doubles each attempt up to `max_retry_delay`, with ±20% jitter. |
+| `max_retry_delay` | no | `5s` | Upper bound on a single retry backoff. Keep `max_retries × max_retry_delay` inside `label_keep_max_second`. |
 | `max_filter_ratio` | no | unset | Forwarded as the `max_filter_ratio` Stream Load header. Must be a finite value in `[0.0, 1.0]`; an out-of-range value fails `open()`. |
 | `columns` | no | unset | Forwarded as the `columns` Stream Load header. Validated at startup; an invalid value fails `open()`. |
 | `where` | no | unset | Forwarded as the `where` Stream Load header. Validated at startup; an invalid value fails `open()`. |
@@ -84,11 +88,11 @@ timeout = "30s"
 
 ## Operational guidance
 
-- **`label_keep_max_second`.** Idempotent replay relies on Doris retaining each label for at least as long as it could take the Iggy runtime to redrive a failed batch. The Doris default is 3 days, which is conservative. If you set this lower on the Doris side, make sure your runtime retry budget fits inside the window — once a label expires, a replay re-loads instead of deduping, producing duplicate rows.
+- **`label_keep_max_second`.** The connector's in-request retry re-PUTs a transiently-failed batch under the same label, so Doris must retain that label for at least the connector's full retry budget for the replay to dedupe. The Doris default is 3 days, which is conservative. If you set this lower on the Doris side, make sure `max_retries × max_retry_delay` fits inside the window — once a label expires, a retry re-loads instead of deduping, producing duplicate rows.
 - **Keep `batch_size` stable across a redrive.** The label includes the chunk's `first_offset` and `last_offset`, which are a function of `batch_size`. If you change `batch_size` between a failed load and its redrive, the chunk boundaries shift, the offsets differ, and the new label no longer matches the old one — so Doris re-loads instead of deduping, producing duplicate rows.
 - **Filtered-row alerts.** When Doris reports `number_filtered_rows > 0`, the connector emits a `warn!`. This is your signal that upstream message shapes have drifted from the table schema; alert on it.
-- **Multi-chunk batches are best-effort for operational failures.** A poll larger than `batch_size` is split into chunks, each loaded as its own labelled Stream Load. If a chunk fails *operationally* (serialize, HTTP, or status-classification error), the connector still attempts the remaining chunks and then returns the worst error — it does **not** stop at the first such failure.
-  The runtime commits the consumer offset for the whole poll before `consume()` runs, so a failed chunk is not replayed regardless; pushing the other chunks through maximizes delivered data, and the worst error is surfaced at the end (logged at `error!` for observability — the runtime currently discards `consume()`'s return value, so there is no retry or DLQ).
+- **Multi-chunk batches are best-effort for operational failures.** A poll larger than `batch_size` is split into chunks, each loaded as its own labelled Stream Load (with its own in-request retry budget for transient failures). If a chunk still fails after its retries (serialize, HTTP, or status-classification error), the connector keeps the worst error, attempts the remaining chunks, and returns that error at the end — it does **not** stop at the first such failure.
+  The runtime commits the consumer offset for the whole poll before `consume()` runs, so a chunk that exhausts its in-request retries is not replayed across polls; pushing the other chunks through maximizes delivered data, and the worst error is surfaced at the end (logged at `error!` for observability — the runtime currently discards `consume()`'s return value, so there is no cross-poll redrive or DLQ).
   The one deliberate exception is a **non-JSON payload**, which is treated as a schema-contract violation and aborts the whole poll immediately (see the Requirements note above). Under `schema = "json"` this is unreachable, so it is a defensive guard rather than a normal path.
 
 ## Limitations
@@ -96,4 +100,4 @@ timeout = "30s"
 - JSON payload only. CSV and raw-text payloads are not supported yet.
 - HTTP Basic auth only.
 - No automatic table creation.
-- No built-in retry middleware or circuit breaker — the runtime decides whether to redrive a failing batch. A hardening pass with `iggy_connector_sdk::retry::*` is planned as a follow-up.
+- In-request retry only. Transient backend failures are retried within a single `consume()` call (step 6), but the runtime commits the consumer offset at poll time and discards `consume()`'s return value, so there is no cross-poll redrive or DLQ — delivery is at-most-once under a crash or a failure that outlives the retry budget.
diff --git a/core/connectors/sinks/doris_sink/config.toml b/core/connectors/sinks/doris_sink/config.toml
index 5aaf6b07df..a4107470d8 100644
--- a/core/connectors/sinks/doris_sink/config.toml
+++ b/core/connectors/sinks/doris_sink/config.toml
@@ -44,6 +44,13 @@ timeout = "30s"
 # TCP connect timeout (default "5s"). Independent of timeout; raise it for
 # cross-region or cold-start FEs that are slow to accept the connection.
 # connect_timeout = "5s"
+# In-request retry for transient Stream Load failures (5xx/408/429, transport
+# error, Doris "Publish Timeout"). Retries re-PUT under the same label, which
+# Doris dedupes. max_retries is the total attempt count (1 disables retries);
+# keep max_retries * max_retry_delay inside Doris's label_keep_max_second.
+# max_retries = 3
+# retry_delay = "200ms"
+# max_retry_delay = "5s"
 # Stream Load redirect security. Doris's FE redirects (307) to a BE on another
 # host; credentials are re-attached across that hop. By default a redirect that
 # downgrades https -> http is refused (it would leak credentials in cleartext).
diff --git a/core/connectors/sinks/doris_sink/src/lib.rs b/core/connectors/sinks/doris_sink/src/lib.rs
index 3fe0fe2b7b..919fe2b06b 100644
--- a/core/connectors/sinks/doris_sink/src/lib.rs
+++ b/core/connectors/sinks/doris_sink/src/lib.rs
@@ -19,6 +19,7 @@ use async_trait::async_trait;
 use base64::{Engine as _, engine::general_purpose};
 use bytes::Bytes;
 use humantime::Duration as HumanDuration;
+use iggy_connector_sdk::retry::{exponential_backoff, jitter};
 use iggy_connector_sdk::{
     ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata, sink_connector,
 };
@@ -58,6 +59,15 @@ const LABEL_HASH_HEX_LEN: usize = 16;
 // returns a giant body can't flood the logs. Bounds only what we *log*, not peak
 // memory — `response.text()` already buffers the full body first.
 const MAX_RESPONSE_LOG_BYTES: usize = 4096;
+// In-request retry budget for *transient* Stream Load failures (5xx/408/429,
+// transport error, Doris "Publish Timeout"). The runtime commits the consumer
+// offset at poll time before consume() runs, so a transient failure we don't
+// retry here is lost — an in-request re-PUT under the same label (which Doris
+// dedupes) is the connector's only redelivery lever. Keep the worst-case budget
+// well inside Doris's label_keep_max_second so a retry still dedupes.
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "200ms";
+const DEFAULT_RETRY_MAX_DELAY: &str = "5s";
 
 #[derive(Debug)]
 pub struct DorisSink {
@@ -85,6 +95,11 @@ struct Connected {
     // off `self` instead of threading it through every call.
     allow_insecure_redirect: bool,
     allowed_redirect_hosts: Option<Vec<String>>,
+    // In-request retry policy for transient Stream Load failures, resolved once
+    // at `open()`. `max_retries` is the total attempt count (1 = no retry).
+    max_retries: u32,
+    retry_delay: Duration,
+    retry_max_delay: Duration,
 }
 
 #[derive(Debug, Deserialize)]
@@ -107,6 +122,19 @@ pub struct DorisSinkConfig {
     /// cross-region or cold-start FEs that are slow to accept the connection.
     pub connect_timeout: Option<String>,
     pub batch_size: Option<u32>,
+    /// Total number of Stream Load attempts per batch on a *transient* failure
+    /// (HTTP 5xx/408/429, a transport error, or Doris "Publish Timeout"). `1`
+    /// disables retries. Each retry re-PUTs under the same label — which Doris
+    /// dedupes — so an ambiguous success (e.g. a 2xx whose body we couldn't
+    /// read) is absorbed rather than doubled. Default 3.
+    pub max_retries: Option<u32>,
+    /// Base backoff before the first retry, as a human-readable duration (e.g.
+    /// "200ms"). Doubles each attempt up to `max_retry_delay`, with ±20% jitter.
+    pub retry_delay: Option<String>,
+    /// Upper bound on a single retry backoff, as a human-readable duration (e.g.
+    /// "5s"). Keep `max_retries * max_retry_delay` inside Doris's
+    /// `label_keep_max_second` so a retry still dedupes instead of re-loading.
+    pub max_retry_delay: Option<String>,
     /// Permit a redirect that downgrades the scheme (e.g. `https://` FE ->
     /// `http://` BE). Off by default: a downgrade would push Basic-auth
     /// credentials onto a cleartext hop, so we refuse it unless the operator
@@ -312,6 +340,51 @@ impl DorisSink {
             return parse_stream_load_response(&response_text);
         }
     }
+
+    /// Load one batch with bounded in-request retry. `send_stream_load` performs
+    /// a single full FE -> BE attempt; this wraps it (plus status
+    /// classification) so a *transient* failure re-PUTs under the same `label`.
+    /// The runtime commits the consumer offset before consume() runs, so this is
+    /// the connector's only redelivery path on a transient outage; the shared
+    /// label lets Doris dedupe a prior attempt that actually landed (e.g. a 2xx
+    /// whose body we couldn't read). A `PermanentHttpError` returns immediately —
+    /// retrying bad data would just hammer the FE.
+    async fn load_batch(&self, label: &str, body: Bytes) -> Result<StreamLoadResponse, Error> {
+        let connected = self.connected.as_ref().ok_or_else(|| {
+            Error::InitError(format!(
+                "Doris sink ID {} called before open() — not connected",
+                self.id
+            ))
+        })?;
+
+        let mut attempt = 0u32;
+        loop {
+            let result = match self.send_stream_load(label, body.clone()).await {
+                Ok(response) => classify_status(&response).map(|()| response),
+                Err(error) => Err(error),
+            };
+            let error = match result {
+                Ok(response) => return Ok(response),
+                Err(error) => error,
+            };
+
+            attempt += 1;
+            if attempt >= connected.max_retries || !is_transient_error(&error) {
+                return Err(error);
+            }
+
+            let delay = jitter(exponential_backoff(
+                connected.retry_delay,
+                attempt - 1,
+                connected.retry_max_delay,
+            ));
+            warn!(
+                "Doris sink ID {} transient Stream Load failure on attempt {attempt}/{} (label={label}): {error}; retrying in {delay:?}",
+                self.id, connected.max_retries
+            );
+            tokio::time::sleep(delay).await;
+        }
+    }
 }
 
 impl Connected {
@@ -484,6 +557,12 @@ fn effective_batch_size(configured: Option<u32>) -> usize {
     configured.unwrap_or(DEFAULT_BATCH_SIZE).max(1) as usize
 }
 
+/// Total Stream Load attempts per batch: the configured value floored at 1 so
+/// `0` (or an unset value) collapses to a single attempt with no retry.
+fn effective_max_retries(configured: Option<u32>) -> u32 {
+    configured.unwrap_or(DEFAULT_MAX_RETRIES).max(1)
+}
+
 /// Build a validated Stream Load header value, surfacing a bad byte (CR/LF,
 /// non-visible-ASCII) as a startup-time `InvalidConfigValue` instead of a
 /// per-batch `HttpRequestFailed` (reqwest defers `HeaderValue::try_from` to
@@ -599,6 +678,18 @@ fn validate_identifier(name: &str, field: &str, id: u32) -> Result<(), Error> {
     Ok(())
 }
 
+/// Transient Stream Load failures worth an in-request retry: HTTP 5xx/408/429
+/// and transport errors (`CannotStoreData` / `HttpRequestFailed`), plus Doris's
+/// "Publish Timeout" (also `CannotStoreData`). `PermanentHttpError` (4xx, "Fail",
+/// schema/redirect problems, unparsable body) is never retried — re-PUTing bad
+/// data just hammers the FE.
+fn is_transient_error(error: &Error) -> bool {
+    matches!(
+        error,
+        Error::CannotStoreData(_) | Error::HttpRequestFailed(_)
+    )
+}
+
 fn classify_status(response: &StreamLoadResponse) -> Result<(), Error> {
     match response.status.as_str() {
         "Success" => Ok(()),
@@ -710,6 +801,23 @@ impl Sink for DorisSink {
             None => None,
         };
 
+        let retry_delay = parse_duration(self.config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY);
+        let retry_max_delay = parse_duration(
+            self.config.max_retry_delay.as_deref(),
+            DEFAULT_RETRY_MAX_DELAY,
+        );
+        // `exponential_backoff` already caps at the max, but a base above the cap
+        // is a config mistake worth surfacing rather than silently flattening.
+        let (retry_delay, retry_max_delay) = if retry_delay > retry_max_delay {
+            warn!(
+                "Doris sink ID {}: retry_delay ({retry_delay:?}) exceeds max_retry_delay ({retry_max_delay:?}); clamping base to the cap",
+                self.id
+            );
+            (retry_max_delay, retry_max_delay)
+        } else {
+            (retry_delay, retry_max_delay)
+        };
+
         self.connected = Some(Connected {
             client: self.build_client()?,
             base_url,
@@ -718,6 +826,9 @@ impl Sink for DorisSink {
             where_header,
             allow_insecure_redirect: self.config.allow_insecure_redirect.unwrap_or(false),
             allowed_redirect_hosts: self.config.allowed_redirect_hosts.clone(),
+            max_retries: effective_max_retries(self.config.max_retries),
+            retry_delay,
+            retry_max_delay,
         });
 
         info!(
@@ -807,39 +918,36 @@ impl Sink for DorisSink {
                 last_msg.offset,
             );
 
-            match self.send_stream_load(&label, body).await {
-                Ok(response) => match classify_status(&response) {
-                    Ok(()) => {
-                        if response.number_filtered_rows > 0 {
-                            // Filtered rows usually mean schema drift upstream.
-                            // Surface above debug so operators can alert on it.
-                            warn!(
-                                "Doris sink ID {} loaded {} rows but FILTERED {} rows for {}.{} (label={label}); \
-                                 likely schema drift upstream",
-                                self.id,
-                                response.number_loaded_rows,
-                                response.number_filtered_rows,
-                                self.config.database,
-                                self.config.table,
-                            );
-                        } else {
-                            debug!(
-                                "Doris sink ID {} loaded {} rows into {}.{} (label={label})",
-                                self.id,
-                                response.number_loaded_rows,
-                                self.config.database,
-                                self.config.table,
-                            );
-                        }
-                    }
-                    Err(e) => {
-                        error!("Doris sink ID {} batch failed: {e}", self.id);
-                        first_error.get_or_insert(e);
+            match self.load_batch(&label, body).await {
+                Ok(response) => {
+                    if response.number_filtered_rows > 0 {
+                        // Filtered rows usually mean schema drift upstream.
+                        // Surface above debug so operators can alert on it.
+                        warn!(
+                            "Doris sink ID {} loaded {} rows but FILTERED {} rows for {}.{} (label={label}); \
+                             likely schema drift upstream",
+                            self.id,
+                            response.number_loaded_rows,
+                            response.number_filtered_rows,
+                            self.config.database,
+                            self.config.table,
+                        );
+                    } else {
+                        debug!(
+                            "Doris sink ID {} loaded {} rows into {}.{} (label={label})",
+                            self.id,
+                            response.number_loaded_rows,
+                            self.config.database,
+                            self.config.table,
+                        );
                     }
-                },
-                Err(e) => {
-                    error!("Doris sink ID {} HTTP failed: {e}", self.id);
-                    first_error.get_or_insert(e);
+                }
+                Err(error) => {
+                    error!(
+                        "Doris sink ID {} batch failed (label={label}): {error}",
+                        self.id
+                    );
+                    first_error.get_or_insert(error);
                 }
             }
         }
@@ -876,6 +984,9 @@ mod tests {
             batch_size: None,
             allow_insecure_redirect: None,
             allowed_redirect_hosts: None,
+            max_retries: None,
+            retry_delay: None,
+            max_retry_delay: None,
         }
     }
 
@@ -1178,6 +1289,9 @@ mod tests {
             where_header: None,
             allow_insecure_redirect: allow_insecure,
             allowed_redirect_hosts: allowed_hosts,
+            max_retries: DEFAULT_MAX_RETRIES,
+            retry_delay: Duration::from_millis(1),
+            retry_max_delay: Duration::from_millis(5),
         }
     }
 
@@ -1541,4 +1655,118 @@ mod tests {
             "expected PermanentHttpError on relative Location, got {result:?}",
         );
     }
+
+    /// A transient failure (HTTP 503) is retried in-request, and the next
+    /// attempt's success is returned. The `.expect(1)` on the 503 mock proves
+    /// the retry actually re-PUT (a single attempt would have surfaced the
+    /// transient error instead of `Ok(Success)`).
+    #[tokio::test]
+    async fn transient_failure_is_retried_then_succeeds() {
+        use wiremock::matchers::{method, path};
+        use wiremock::{Mock, MockServer, ResponseTemplate};
+
+        let server = MockServer::start().await;
+        let mut cfg = make_config();
+        cfg.fe_url = server.uri();
+        cfg.max_retries = Some(3);
+        cfg.retry_delay = Some("1ms".into());
+        cfg.max_retry_delay = Some("5ms".into());
+
+        // wiremock serves the first matching mock in mount order, so mount the
+        // single-shot 503 first: it serves attempt 1, then — capped at one
+        // response — stops matching, and attempt 2 falls through to the success.
+        Mock::given(method("PUT"))
+            .and(path("/api/test_db/test_tbl/_stream_load"))
+            .respond_with(ResponseTemplate::new(503))
+            .up_to_n_times(1)
+            .expect(1)
+            .mount(&server)
+            .await;
+        Mock::given(method("PUT"))
+            .and(path("/api/test_db/test_tbl/_stream_load"))
+            .respond_with(
+                ResponseTemplate::new(200)
+                    .set_body_json(serde_json::json!({"Status": "Success", "NumberLoadedRows": 1})),
+            )
+            .mount(&server)
+            .await;
+
+        let mut sink = DorisSink::new(1, cfg);
+        sink.open().await.expect("open should succeed");
+        let result = sink
+            .load_batch("iggy-test-label", Bytes::from_static(b"[{\"a\":1}]"))
+            .await;
+
+        assert!(
+            matches!(&result, Ok(r) if r.status == "Success"),
+            "expected Ok(Success) after one retry, got {result:?}",
+        );
+    }
+
+    /// When every attempt fails transiently, the budget is exhausted and the
+    /// last transient error is surfaced. `.expect(2)` pins the attempt count to
+    /// exactly `max_retries` (1 initial + 1 retry) — no over- or under-retry.
+    #[tokio::test]
+    async fn transient_failure_exhausts_retries_and_returns_error() {
+        use wiremock::matchers::{method, path};
+        use wiremock::{Mock, MockServer, ResponseTemplate};
+
+        let server = MockServer::start().await;
+        let mut cfg = make_config();
+        cfg.fe_url = server.uri();
+        cfg.max_retries = Some(2);
+        cfg.retry_delay = Some("1ms".into());
+        cfg.max_retry_delay = Some("5ms".into());
+
+        Mock::given(method("PUT"))
+            .and(path("/api/test_db/test_tbl/_stream_load"))
+            .respond_with(ResponseTemplate::new(503))
+            .expect(2)
+            .mount(&server)
+            .await;
+
+        let mut sink = DorisSink::new(1, cfg);
+        sink.open().await.expect("open should succeed");
+        let result = sink
+            .load_batch("iggy-test-label", Bytes::from_static(b"[{\"a\":1}]"))
+            .await;
+
+        assert!(
+            matches!(&result, Err(Error::CannotStoreData(_))),
+            "expected CannotStoreData after exhausting retries, got {result:?}",
+        );
+    }
+
+    /// A permanent failure (HTTP 400) returns on the first attempt with no
+    /// retry, even with `max_retries` high. `.expect(1)` pins it to one attempt.
+    #[tokio::test]
+    async fn permanent_failure_is_not_retried() {
+        use wiremock::matchers::{method, path};
+        use wiremock::{Mock, MockServer, ResponseTemplate};
+
+        let server = MockServer::start().await;
+        let mut cfg = make_config();
+        cfg.fe_url = server.uri();
+        cfg.max_retries = Some(5);
+        cfg.retry_delay = Some("1ms".into());
+        cfg.max_retry_delay = Some("5ms".into());
+
+        Mock::given(method("PUT"))
+            .and(path("/api/test_db/test_tbl/_stream_load"))
+            .respond_with(ResponseTemplate::new(400))
+            .expect(1)
+            .mount(&server)
+            .await;
+
+        let mut sink = DorisSink::new(1, cfg);
+        sink.open().await.expect("open should succeed");
+        let result = sink
+            .load_batch("iggy-test-label", Bytes::from_static(b"[{\"a\":1}]"))
+            .await;
+
+        assert!(
+            matches!(&result, Err(Error::PermanentHttpError(_))),
+            "expected PermanentHttpError with no retry, got {result:?}",
+        );
+    }
 }

From 5d023919365cc59aa74953368faf266ee9249367 Mon Sep 17 00:00:00 2001
From: ryankert01 <ryankert01@gmail.com>
Date: Sun, 28 Jun 2026 04:54:50 +0800
Subject: [PATCH 2/2] feat(connectors): add opt-in CSV output format to Doris
 sink

The Doris sink only emitted JSON Stream Load, but Doris parses JSON on the BE
more expensively than CSV. A 10k-row benchmark shows CSV loads ~21% faster
server-side at roughly half the body size.

This adds an opt-in output_format = "csv" (default stays JSON). CSV is
positional where JSON is name-mapped, so it requires the columns config to pin
column order; open() rejects output_format = "csv" without it. The hand-rolled
encoder uses control-char framing (column_separator \x01, line_delimiter \x02)
with enclose/escape quoting so embedded separators, quotes, and newlines
round-trip; Doris uses escape-prefix rather than RFC-4180 doubling, so the csv
crate does not fit. JSON null and missing keys map to \N, empty string to "",
numbers and bools emit bare, nested values stringify to JSON. The field is
output_format (not format) to avoid an env-override collision with the
runtime's top-level plugin_config_format.

Verified end-to-end against a real apache/doris:4.0.3-all-slim container: a
connector-driven test round-trips a value containing a comma, a quote, and a
backslash byte for byte, alongside an ignored JSON-vs-CSV Stream Load benchmark.

Relates to #3215.
---
 .../example_config/connectors/doris_sink.toml |   5 +
 core/connectors/sinks/doris_sink/README.md    |   7 +-
 core/connectors/sinks/doris_sink/config.toml  |   5 +
 core/connectors/sinks/doris_sink/src/lib.rs   | 367 +++++++++++++++++-
 .../tests/connectors/doris/doris_sink.rs      |  86 +++-
 .../doris/doris_stream_load_format_bench.rs   | 285 ++++++++++++++
 .../integration/tests/connectors/doris/mod.rs |   1 +
 .../connectors/fixtures/doris/container.rs    |  45 +++
 .../tests/connectors/fixtures/doris/mod.rs    |   4 +-
 .../tests/connectors/fixtures/mod.rs          |   4 +-
 10 files changed, 786 insertions(+), 23 deletions(-)
 create mode 100644 core/integration/tests/connectors/doris/doris_stream_load_format_bench.rs

diff --git a/core/connectors/runtime/example_config/connectors/doris_sink.toml b/core/connectors/runtime/example_config/connectors/doris_sink.toml
index 20f4e7663e..4f7118068e 100644
--- a/core/connectors/runtime/example_config/connectors/doris_sink.toml
+++ b/core/connectors/runtime/example_config/connectors/doris_sink.toml
@@ -41,6 +41,11 @@ password = "replace_with_secret"
 label_prefix = "iggy"
 batch_size = 1000
 timeout = "30s"
+# Output format: "json" (default) or "csv". CSV is opt-in for throughput and is
+# positional, so it REQUIRES `columns` to pin the column order (JSON is
+# name-mapped). open() fails if output_format="csv" without columns.
+# output_format = "json"
+# columns = "id, name, count, amount, active, timestamp"
 # TCP connect timeout (default "5s"). Independent of timeout; raise it for
 # cross-region or cold-start FEs that are slow to accept the connection.
 # connect_timeout = "5s"
diff --git a/core/connectors/sinks/doris_sink/README.md b/core/connectors/sinks/doris_sink/README.md
index b949a50b8e..2689fa97c4 100644
--- a/core/connectors/sinks/doris_sink/README.md
+++ b/core/connectors/sinks/doris_sink/README.md
@@ -11,12 +11,12 @@ The Doris sink connector consumes JSON messages from Iggy streams and writes the
 
 ## How it works
 
-1. For each batch of messages, the connector serializes the JSON payloads into a JSON array.
+1. For each batch of messages, the connector serializes the JSON payloads into the configured output format: a JSON array by default, or CSV when `output_format = "csv"` (see Configuration).
 2. It computes a deterministic Stream Load `label` of the form `{label_prefix}-{stream_san}-{topic_san}-{hash16}-{partition}-{first_offset}-{last_offset}`.
    - `hash16` is a single 64-bit blake3 hash computed over the *raw* (un-sanitized), length-prefixed `(label_prefix, stream, topic)` triple. So identities that sanitize to the same string get distinct labels — whether the collision is in the names (`events.v1` vs `events_v1`) or in two tenants' prefixes that truncate alike (`prod_events_us_east_1` vs `..._2`) — and no boundary-shift aliasing is possible (`("ab","c")` ≠ `("a","bc")`).
    - The total label is bounded under Doris's 128-char cap regardless of input length (worst case 120 chars).
    - Doris dedupes loads by label inside its `label_keep_max_second` window. The in-request retry (step 6) re-PUTs a transiently-failed batch under the same label, so a prior attempt that actually landed (e.g. a `2xx` whose body we couldn't read) is absorbed, not doubled. This protects **in-request retry only**: the runtime commits the offset before `consume()` runs and discards its return, so a failure outliving the retry budget or a crash mid-load is **at-most-once**.
-3. It `PUT`s the batch to `{fe_url}/api/{database}/{table}/_stream_load` with HTTP Basic auth and the headers `Expect: 100-continue`, `format: json`, `strip_outer_array: true`, `label: <label>`. (`Expect: 100-continue` is required by Doris's Stream Load endpoint, which rejects PUTs that omit it. Where the HTTP stack negotiates the handshake it also lets Doris reject auth/4xx before the body uploads — a secondary benefit, not relied on for correctness.)
+3. It `PUT`s the batch to `{fe_url}/api/{database}/{table}/_stream_load` with HTTP Basic auth, `Expect: 100-continue`, `label: <label>`, and the format headers (`format: json` + `strip_outer_array: true` for JSON; `format: csv` with control-char `column_separator`/`line_delimiter` + `enclose`/`escape` for CSV). `Expect: 100-continue` is required by Doris's Stream Load endpoint, which rejects PUTs that omit it; it also lets Doris reject auth/4xx before the body uploads.
 4. The Doris frontend (FE) responds with a `307 Temporary Redirect` to a backend (BE). The connector follows the redirect manually so that the `Authorization` header is preserved across the hop (`reqwest`'s default policy strips it on cross-host redirects).
    `308 Permanent Redirect` is also followed as a defensive measure; redirects beyond a hard cap of 5 (or a redirect with no usable `Location`) are rejected as a permanent `PermanentHttpError`, since retrying a malformed/looping redirect cannot help.
 5. The HTTP body is parsed as JSON and the `Status` field decides the outcome:
@@ -45,6 +45,7 @@ The Doris sink connector consumes JSON messages from Iggy streams and writes the
 | `max_filter_ratio` | no | unset | Forwarded as the `max_filter_ratio` Stream Load header. Must be a finite value in `[0.0, 1.0]`; an out-of-range value fails `open()`. |
 | `columns` | no | unset | Forwarded as the `columns` Stream Load header. Validated at startup; an invalid value fails `open()`. |
 | `where` | no | unset | Forwarded as the `where` Stream Load header. Validated at startup; an invalid value fails `open()`. |
+| `output_format` | no | `json` | Stream Load output format: `json` or `csv`. CSV is opt-in for throughput; it **requires `columns`** (CSV is positional, unlike name-mapped JSON) and emits control-char-framed, `enclose`/`escape`-quoted rows. `open()` fails if `output_format = "csv"` and `columns` is unset. (Named `output_format`, not `format`, to avoid an env-override collision with `plugin_config_format`.) |
 | `allow_insecure_redirect` | no | `false` | Permit a Stream Load redirect that downgrades `https://` → `http://`. Refused by default because it would push credentials onto a cleartext hop. |
 | `allowed_redirect_hosts` | no | unset | Allowlist of redirect targets. Each entry is `host` (pins the host, any port) or `host:port` (pins the exact endpoint). When set and non-empty, any other redirect target is refused. |
 
@@ -97,7 +98,7 @@ timeout = "30s"
 
 ## Limitations
 
-- JSON payload only. CSV and raw-text payloads are not supported yet.
+- Output is JSON (default) or CSV (`output_format = "csv"`, which requires `columns`). Both serialize from `Payload::Json`; raw-text/CSV passthrough and Parquet are not supported. CSV maps JSON `null` and missing keys to SQL `NULL` (`\N`), an empty string to `""`, emits numbers and booleans bare (`true`/`false`), and stringifies nested objects/arrays as JSON.
 - HTTP Basic auth only.
 - No automatic table creation.
 - In-request retry only. Transient backend failures are retried within a single `consume()` call (step 6), but the runtime commits the consumer offset at poll time and discards `consume()`'s return value, so there is no cross-poll redrive or DLQ — delivery is at-most-once under a crash or a failure that outlives the retry budget.
diff --git a/core/connectors/sinks/doris_sink/config.toml b/core/connectors/sinks/doris_sink/config.toml
index a4107470d8..560ba0103d 100644
--- a/core/connectors/sinks/doris_sink/config.toml
+++ b/core/connectors/sinks/doris_sink/config.toml
@@ -41,6 +41,11 @@ password = ""
 label_prefix = "iggy"
 batch_size = 1000
 timeout = "30s"
+# Output format: "json" (default) or "csv". CSV is opt-in for throughput and is
+# positional, so it REQUIRES `columns` to pin the column order (JSON is
+# name-mapped). open() fails if output_format="csv" without columns.
+# output_format = "json"
+# columns = "id, name, count, amount, active, timestamp"
 # TCP connect timeout (default "5s"). Independent of timeout; raise it for
 # cross-region or cold-start FEs that are slow to accept the connection.
 # connect_timeout = "5s"
diff --git a/core/connectors/sinks/doris_sink/src/lib.rs b/core/connectors/sinks/doris_sink/src/lib.rs
index 919fe2b06b..2a6744e98b 100644
--- a/core/connectors/sinks/doris_sink/src/lib.rs
+++ b/core/connectors/sinks/doris_sink/src/lib.rs
@@ -27,6 +27,8 @@ use reqwest::{Method, StatusCode, header};
 use secrecy::zeroize::Zeroizing;
 use secrecy::{ExposeSecret, SecretString};
 use serde::Deserialize;
+use simd_json::{OwnedValue, StaticNode};
+use std::io::Write as _;
 use std::str::FromStr;
 use std::time::Duration;
 use tracing::{debug, error, info, warn};
@@ -68,6 +70,32 @@ const MAX_RESPONSE_LOG_BYTES: usize = 4096;
 const DEFAULT_MAX_RETRIES: u32 = 3;
 const DEFAULT_RETRY_DELAY: &str = "200ms";
 const DEFAULT_RETRY_MAX_DELAY: &str = "5s";
+// Doris CSV framing. Control-char separators keep field/row boundaries
+// collision-improbable; `enclose`+`escape` protect any value that still contains
+// them. Doris escapes with a prefix char (`\"`), NOT RFC-4180 quote-doubling,
+// which is why we hand-roll the encoder instead of using the `csv` crate. The
+// four bytes are sent to Doris as the header forms below (it decodes the `\xNN`
+// prefix); all are visible ASCII so `validated_header` accepts them.
+const CSV_COLUMN_SEPARATOR: u8 = 0x01; // SOH
+const CSV_LINE_DELIMITER: u8 = 0x02; // STX
+const CSV_ENCLOSE: u8 = b'"';
+const CSV_ESCAPE: u8 = b'\\';
+const CSV_NULL_MARKER: &[u8] = b"\\N";
+const CSV_COLUMN_SEPARATOR_HEADER: &str = "\\x01";
+const CSV_LINE_DELIMITER_HEADER: &str = "\\x02";
+const CSV_ENCLOSE_HEADER: &str = "\"";
+const CSV_ESCAPE_HEADER: &str = "\\";
+
+/// Stream Load output serialization. `Json` (default) is name-mapped and handles
+/// embedded separators/quotes/newlines natively; `Csv` is opt-in for throughput
+/// and is positional, so it requires the `columns` config to pin column order.
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum Format {
+    #[default]
+    Json,
+    Csv,
+}
 
 #[derive(Debug)]
 pub struct DorisSink {
@@ -100,6 +128,11 @@ struct Connected {
     max_retries: u32,
     retry_delay: Duration,
     retry_max_delay: Duration,
+    // Output format resolved once at `open()`. For `Csv`, `csv_columns` holds the
+    // positional column order (the leading bare names from the `columns` config);
+    // it is empty for `Json`.
+    format: Format,
+    csv_columns: Vec<String>,
 }
 
 #[derive(Debug, Deserialize)]
@@ -114,6 +147,13 @@ pub struct DorisSinkConfig {
     pub columns: Option<String>,
     #[serde(rename = "where")]
     pub where_clause: Option<String>,
+    /// Stream Load output format: `"json"` (default) or `"csv"`. CSV is opt-in
+    /// for throughput; because Doris CSV is positional (JSON is name-mapped), it
+    /// requires `columns` to pin the column order, else `open()` fails. Named
+    /// `output_format` (not `format`) so the env override
+    /// `..._PLUGIN_CONFIG_OUTPUT_FORMAT` doesn't collide with the runtime's
+    /// top-level `plugin_config_format` (both would flatten to `..._FORMAT`).
+    pub output_format: Option<Format>,
     /// Total per-request HTTP timeout as a human-readable duration, e.g. "30s"
     /// (default 30s). Matches the `timeout` field on the http/influxdb sinks.
     pub timeout: Option<String>,
@@ -225,11 +265,24 @@ impl DorisSink {
                 .request(Method::PUT, url.clone())
                 .header(header::AUTHORIZATION, self.auth_header.clone())
                 .header(header::EXPECT, "100-continue")
-                .header("format", "json")
-                .header("strip_outer_array", "true")
                 .header("label", label)
                 .body(body.clone());
 
+            // Format-specific Stream Load headers. JSON is name-mapped and uses
+            // `strip_outer_array` (the body is a JSON array); CSV is positional
+            // with control-char framing + enclose/escape (no strip_outer_array).
+            request = match connected.format {
+                Format::Json => request
+                    .header("format", "json")
+                    .header("strip_outer_array", "true"),
+                Format::Csv => request
+                    .header("format", "csv")
+                    .header("column_separator", CSV_COLUMN_SEPARATOR_HEADER)
+                    .header("line_delimiter", CSV_LINE_DELIMITER_HEADER)
+                    .header("enclose", CSV_ENCLOSE_HEADER)
+                    .header("escape", CSV_ESCAPE_HEADER),
+            };
+
             // Headers were validated and built once in `open()`.
             if let Some(value) = &connected.max_filter_ratio_header {
                 request = request.header("max_filter_ratio", value.clone());
@@ -563,6 +616,99 @@ fn effective_max_retries(configured: Option<u32>) -> u32 {
     configured.unwrap_or(DEFAULT_MAX_RETRIES).max(1)
 }
 
+/// The leading positional column names from a Doris `columns` header: bare names
+/// in order, stopping at the first derived `name = expr` entry (Doris computes
+/// those server-side and they consume no CSV field). Empty if there are none.
+fn csv_column_names(columns: &str) -> Vec<String> {
+    columns
+        .split(',')
+        .map(str::trim)
+        .take_while(|name| !name.is_empty() && !name.contains('='))
+        .map(str::to_string)
+        .collect()
+}
+
+/// Serialize a chunk into the request body for the configured Stream Load format.
+fn build_request_body(
+    format: Format,
+    rows: &[&OwnedValue],
+    csv_columns: &[String],
+) -> Result<Bytes, Error> {
+    match format {
+        Format::Json => simd_json::to_vec(&rows).map(Bytes::from).map_err(|e| {
+            Error::CannotStoreData(format!("Failed to serialize JSON batch for Doris: {e}"))
+        }),
+        Format::Csv => build_csv_body(rows, csv_columns).map(Bytes::from),
+    }
+}
+
+/// Serialize JSON object rows into Doris CSV: one `enclose`-quoted/escaped field
+/// per column in `columns` order, control-char separated. A missing key or JSON
+/// `null` becomes the unenclosed `\N` NULL marker; an empty string becomes the
+/// enclosed `""` (the distinction Doris draws between NULL and empty). Numbers
+/// and bools are emitted bare; nested values are compact-JSON-stringified.
+fn build_csv_body(rows: &[&OwnedValue], columns: &[String]) -> Result<Vec<u8>, Error> {
+    let mut out = Vec::with_capacity(rows.len() * columns.len() * 16);
+    for &row in rows {
+        let OwnedValue::Object(object) = row else {
+            return Err(Error::InvalidRecordValue(
+                "Doris CSV format requires each message to be a JSON object".to_string(),
+            ));
+        };
+        for (index, column) in columns.iter().enumerate() {
+            if index > 0 {
+                out.push(CSV_COLUMN_SEPARATOR);
+            }
+            match object.get(column.as_str()) {
+                None => out.extend_from_slice(CSV_NULL_MARKER),
+                Some(value) => encode_csv_field(value, &mut out),
+            }
+        }
+        out.push(CSV_LINE_DELIMITER);
+    }
+    Ok(out)
+}
+
+/// Append one JSON value as a Doris CSV field (see `build_csv_body`).
+fn encode_csv_field(value: &OwnedValue, out: &mut Vec<u8>) {
+    match value {
+        OwnedValue::Static(StaticNode::Null) => out.extend_from_slice(CSV_NULL_MARKER),
+        OwnedValue::Static(StaticNode::Bool(flag)) => {
+            out.extend_from_slice(if *flag { b"true" } else { b"false" });
+        }
+        OwnedValue::Static(StaticNode::I64(number)) => {
+            let _ = write!(out, "{number}");
+        }
+        OwnedValue::Static(StaticNode::U64(number)) => {
+            let _ = write!(out, "{number}");
+        }
+        OwnedValue::Static(StaticNode::F64(number)) => {
+            let _ = write!(out, "{number}");
+        }
+        OwnedValue::String(text) => encode_csv_enclosed(text.as_bytes(), out),
+        // Nested array/object: stringify to compact JSON and enclose as text. The
+        // target Doris column must accept it (STRING/JSON/VARIANT).
+        nested => match simd_json::to_string(nested) {
+            Ok(json) => encode_csv_enclosed(json.as_bytes(), out),
+            Err(_) => out.extend_from_slice(CSV_NULL_MARKER),
+        },
+    }
+}
+
+/// Append `bytes` as an `enclose`-quoted CSV field, prefix-escaping any embedded
+/// escape or enclose byte. The escape char is handled by the same branch, so a
+/// trailing backslash cannot leak past the closing quote.
+fn encode_csv_enclosed(bytes: &[u8], out: &mut Vec<u8>) {
+    out.push(CSV_ENCLOSE);
+    for &byte in bytes {
+        if byte == CSV_ESCAPE || byte == CSV_ENCLOSE {
+            out.push(CSV_ESCAPE);
+        }
+        out.push(byte);
+    }
+    out.push(CSV_ENCLOSE);
+}
+
 /// Build a validated Stream Load header value, surfacing a bad byte (CR/LF,
 /// non-visible-ASCII) as a startup-time `InvalidConfigValue` instead of a
 /// per-batch `HttpRequestFailed` (reqwest defers `HeaderValue::try_from` to
@@ -818,6 +964,29 @@ impl Sink for DorisSink {
             (retry_delay, retry_max_delay)
         };
 
+        let format = self.config.output_format.unwrap_or_default();
+        // CSV is positional, so it needs the explicit column order; JSON is
+        // name-mapped and ignores it. Resolve (and validate) once at open().
+        let csv_columns = match format {
+            Format::Json => Vec::new(),
+            Format::Csv => {
+                let columns = self
+                    .config
+                    .columns
+                    .as_deref()
+                    .map(csv_column_names)
+                    .unwrap_or_default();
+                if columns.is_empty() {
+                    return Err(Error::InvalidConfigValue(format!(
+                        "Doris sink ID {}: format=\"csv\" requires a non-empty `columns` listing \
+                         the source columns in order (CSV is positional, unlike name-mapped JSON)",
+                        self.id
+                    )));
+                }
+                columns
+            }
+        };
+
         self.connected = Some(Connected {
             client: self.build_client()?,
             base_url,
@@ -829,6 +998,8 @@ impl Sink for DorisSink {
             max_retries: effective_max_retries(self.config.max_retries),
             retry_delay,
             retry_max_delay,
+            format,
+            csv_columns,
         });
 
         info!(
@@ -860,6 +1031,12 @@ impl Sink for DorisSink {
             .label_prefix
             .as_deref()
             .unwrap_or(DEFAULT_LABEL_PREFIX);
+        let connected = self.connected.as_ref().ok_or_else(|| {
+            Error::InitError(format!(
+                "Doris sink ID {} called before open() — not connected",
+                self.id
+            ))
+        })?;
         let mut first_error: Option<Error> = None;
 
         // Best-effort across chunks: on a per-chunk serialize/HTTP/status failure
@@ -898,16 +1075,15 @@ impl Sink for DorisSink {
                 continue;
             };
 
-            let body = match simd_json::to_vec(&json_values) {
-                Ok(b) => Bytes::from(b),
-                Err(e) => {
-                    error!("Doris sink ID {} failed to serialize batch: {e}", self.id);
-                    first_error.get_or_insert(Error::CannotStoreData(format!(
-                        "Failed to serialize batch for Doris: {e}"
-                    )));
-                    continue;
-                }
-            };
+            let body =
+                match build_request_body(connected.format, &json_values, &connected.csv_columns) {
+                    Ok(body) => body,
+                    Err(e) => {
+                        error!("Doris sink ID {} failed to serialize batch: {e}", self.id);
+                        first_error.get_or_insert(e);
+                        continue;
+                    }
+                };
 
             let label = build_label(
                 label_prefix,
@@ -979,6 +1155,7 @@ mod tests {
             max_filter_ratio: None,
             columns: None,
             where_clause: None,
+            output_format: None,
             timeout: None,
             connect_timeout: None,
             batch_size: None,
@@ -1292,6 +1469,8 @@ mod tests {
             max_retries: DEFAULT_MAX_RETRIES,
             retry_delay: Duration::from_millis(1),
             retry_max_delay: Duration::from_millis(5),
+            format: Format::Json,
+            csv_columns: Vec::new(),
         }
     }
 
@@ -1471,7 +1650,8 @@ mod tests {
 
     #[tokio::test]
     async fn consume_aborts_on_first_non_json_payload() {
-        let sink = DorisSink::new(1, make_config());
+        let mut sink = DorisSink::new(1, make_config());
+        sink.open().await.expect("open should succeed");
         let result = sink
             .consume(&topic_meta(), messages_meta(), vec![text_msg(0)])
             .await;
@@ -1483,7 +1663,8 @@ mod tests {
 
     #[tokio::test]
     async fn consume_aborts_on_non_json_in_mixed_batch() {
-        let sink = DorisSink::new(1, make_config());
+        let mut sink = DorisSink::new(1, make_config());
+        sink.open().await.expect("open should succeed");
         let result = sink
             .consume(
                 &topic_meta(),
@@ -1769,4 +1950,162 @@ mod tests {
             "expected PermanentHttpError with no retry, got {result:?}",
         );
     }
+
+    fn owned(json: &str) -> OwnedValue {
+        let mut bytes = json.as_bytes().to_vec();
+        simd_json::to_owned_value(&mut bytes).expect("valid JSON")
+    }
+
+    #[test]
+    fn format_deserializes_from_lowercase() {
+        assert_eq!(
+            serde_json::from_str::<Format>("\"json\"").unwrap(),
+            Format::Json
+        );
+        assert_eq!(
+            serde_json::from_str::<Format>("\"csv\"").unwrap(),
+            Format::Csv
+        );
+        assert!(serde_json::from_str::<Format>("\"xml\"").is_err());
+        assert_eq!(Format::default(), Format::Json);
+    }
+
+    #[test]
+    fn csv_column_names_takes_leading_bare_names() {
+        assert_eq!(csv_column_names("id, name, count"), ["id", "name", "count"]);
+        // A derived `name = expr` column (and anything after) is server-side.
+        assert_eq!(
+            csv_column_names("id, name, calc = count + 1"),
+            ["id", "name"]
+        );
+        assert_eq!(csv_column_names(" a ,b, c "), ["a", "b", "c"]);
+        assert!(csv_column_names("").is_empty());
+        assert!(csv_column_names("calc = x").is_empty());
+    }
+
+    #[test]
+    fn build_csv_body_encodes_scalars_nulls_and_missing() {
+        let row = owned(r#"{"i":-5,"u":10,"f":2.5,"b":false,"nul":null,"empty":""}"#);
+        let columns = ["i", "u", "f", "b", "nul", "missing", "empty"].map(String::from);
+        let body = build_csv_body(&[&row], &columns).unwrap();
+        assert_eq!(*body.last().unwrap(), CSV_LINE_DELIMITER);
+        let fields: Vec<&[u8]> = body[..body.len() - 1]
+            .split(|&byte| byte == CSV_COLUMN_SEPARATOR)
+            .collect();
+        assert_eq!(
+            fields,
+            vec![
+                &b"-5"[..],  // i64
+                &b"10"[..],  // u64
+                &b"2.5"[..], // f64
+                &b"false"[..],
+                &b"\\N"[..],  // json null
+                &b"\\N"[..],  // missing key
+                &b"\"\""[..], // empty string -> enclosed empty (distinct from null)
+            ],
+        );
+    }
+
+    #[test]
+    fn build_csv_body_prefix_escapes_enclose_and_escape_bytes() {
+        // The value contains a quote and a backslash; both must be prefix-escaped
+        // inside the enclosure (escape the escape char too, in one left-to-right
+        // pass, so a trailing backslash can't leak past the closing quote).
+        let row = owned(r#"{"v":"a\"b\\c"}"#);
+        let columns = ["v"].map(String::from);
+        let body = build_csv_body(&[&row], &columns).unwrap();
+        let expected: &[u8] = &[
+            b'"',
+            b'a',
+            b'\\',
+            b'"',
+            b'b',
+            b'\\',
+            b'\\',
+            b'c',
+            b'"',
+            CSV_LINE_DELIMITER,
+        ];
+        assert_eq!(body, expected);
+    }
+
+    #[test]
+    fn build_csv_body_keeps_embedded_separator_and_newline_inside_enclosure() {
+        // Raw separator (0x01), line delimiter (0x02), and a newline live inside
+        // the value; the enclosure protects them, so they are emitted literally
+        // (Doris reads them as data, not structure) and are NOT escaped.
+        let row = owned("{\"v\":\"a\\u0001b\\u0002c\\nd\"}");
+        let columns = ["v"].map(String::from);
+        let body = build_csv_body(&[&row], &columns).unwrap();
+        let expected: &[u8] = &[
+            b'"',
+            b'a',
+            0x01,
+            b'b',
+            0x02,
+            b'c',
+            b'\n',
+            b'd',
+            b'"',
+            CSV_LINE_DELIMITER,
+        ];
+        assert_eq!(body, expected);
+    }
+
+    #[test]
+    fn build_csv_body_stringifies_nested_values() {
+        let row = owned(r#"{"o":{"k":1}}"#);
+        let columns = ["o"].map(String::from);
+        let body = build_csv_body(&[&row], &columns).unwrap();
+        // {"k":1} enclosed, with its inner quotes prefix-escaped.
+        let expected: &[u8] = &[
+            b'"',
+            b'{',
+            b'\\',
+            b'"',
+            b'k',
+            b'\\',
+            b'"',
+            b':',
+            b'1',
+            b'}',
+            b'"',
+            CSV_LINE_DELIMITER,
+        ];
+        assert_eq!(body, expected);
+    }
+
+    #[test]
+    fn build_csv_body_rejects_non_object_row() {
+        let row = owned("[1, 2, 3]");
+        let columns = ["v"].map(String::from);
+        assert!(matches!(
+            build_csv_body(&[&row], &columns),
+            Err(Error::InvalidRecordValue(_))
+        ));
+    }
+
+    #[tokio::test]
+    async fn open_rejects_csv_format_without_columns() {
+        let mut cfg = make_config();
+        cfg.output_format = Some(Format::Csv);
+        cfg.columns = None;
+        let mut sink = DorisSink::new(1, cfg);
+        assert!(matches!(
+            sink.open().await,
+            Err(Error::InvalidConfigValue(_))
+        ));
+    }
+
+    #[tokio::test]
+    async fn open_resolves_csv_columns_dropping_derived() {
+        let mut cfg = make_config();
+        cfg.output_format = Some(Format::Csv);
+        cfg.columns = Some("id, name, calc = id + 1".into());
+        let mut sink = DorisSink::new(1, cfg);
+        sink.open().await.expect("open should succeed");
+        let connected = sink.connected.as_ref().unwrap();
+        assert_eq!(connected.format, Format::Csv);
+        assert_eq!(connected.csv_columns, ["id", "name"]);
+    }
 }
diff --git a/core/integration/tests/connectors/doris/doris_sink.rs b/core/integration/tests/connectors/doris/doris_sink.rs
index dc0b8b1453..051791116b 100644
--- a/core/integration/tests/connectors/doris/doris_sink.rs
+++ b/core/integration/tests/connectors/doris/doris_sink.rs
@@ -17,8 +17,8 @@
 
 use crate::connectors::create_test_messages;
 use crate::connectors::fixtures::{
-    DorisOps, DorisSinkColumnsMappingFixture, DorisSinkFixture, DorisSinkMaxFilterRatioFixture,
-    DorisSinkPreCreatedFixture,
+    DorisOps, DorisSinkColumnsMappingFixture, DorisSinkCsvFixture, DorisSinkFixture,
+    DorisSinkMaxFilterRatioFixture, DorisSinkPreCreatedFixture,
 };
 use bytes::Bytes;
 use iggy::prelude::{IggyMessage, Partitioning};
@@ -488,3 +488,85 @@ async fn given_columns_config_should_apply_derived_expression(
         "expected calculated = count + 1 per row (delta sum = {message_count}), got {delta}"
     );
 }
+
+/// Drives the connector end-to-end with `format = "csv"`: messages flow through
+/// Iggy, the connector serializes each batch as CSV (control-char framing,
+/// `enclose`/`escape` quoting, positional columns) and Stream Loads it into
+/// Doris. Both the row count AND the materialized values are checked, so a
+/// column-order or escaping regression in the CSV encoder fails this test
+/// rather than silently loading garbage. The `name` value below carries a comma,
+/// a double-quote, and a backslash — the bytes a naive CSV encoder would
+/// corrupt — to exercise the enclose/escape path against a real Doris.
+#[iggy_harness(
+    server(connectors_runtime(config_path = "tests/connectors/doris/sink.toml")),
+    seed = seeds::connector_stream
+)]
+async fn given_csv_format_should_store_with_values(
+    harness: &TestHarness,
+    fixture: DorisSinkCsvFixture,
+) {
+    let client = harness.root_client().await.unwrap();
+    let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+    let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+
+    let message_count = 10;
+    let mut test_messages = create_test_messages(message_count);
+    // Make the id=1 row's name a CSV hazard: separator-adjacent comma, an
+    // enclose char, and an escape char. After round-trip it must come back byte
+    // for byte, proving enclose+escape protects the field.
+    let hazardous_name = r#"a,b"c\d"#.to_string();
+    test_messages[0].name = hazardous_name.clone();
+
+    let mut messages: Vec<IggyMessage> = test_messages
+        .iter()
+        .enumerate()
+        .map(|(i, m)| {
+            let payload = Bytes::from(serde_json::to_vec(m).expect("serialize"));
+            IggyMessage::builder()
+                .id((i + 1) as u128)
+                .payload(payload)
+                .build()
+                .expect("build message")
+        })
+        .collect();
+
+    client
+        .send_messages(
+            &stream_id,
+            &topic_id,
+            &Partitioning::partition_id(0),
+            &mut messages,
+        )
+        .await
+        .expect("send messages");
+
+    let db = fixture.database();
+    let count = fixture
+        .wait_for_rows(db, TEST_TABLE, message_count as i64)
+        .await
+        .expect("rows");
+    assert_eq!(count, message_count as i64);
+
+    // Values must round-trip, not just the count: a column-order or escaping bug
+    // would land rows with the wrong/garbled fields.
+    let pool = fixture.pool().await.expect("pool");
+    use sqlx::Row;
+    let row = sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
+        "SELECT name, `count`, amount, active FROM {db}.{TEST_TABLE} WHERE id = 1"
+    )))
+    .fetch_one(&pool)
+    .await
+    .expect("select row id=1");
+    // create_test_messages: id=1 -> count=0, amount=0.0, active=true (name overridden above).
+    let name: String = row.try_get("name").expect("name");
+    let count_val: i32 = row.try_get("count").expect("count");
+    let amount: f64 = row.try_get("amount").expect("amount");
+    let active: bool = row.try_get("active").expect("active");
+    assert_eq!(
+        name, hazardous_name,
+        "enclose/escape must round-trip exactly"
+    );
+    assert_eq!(count_val, 0);
+    assert_eq!(amount, 0.0);
+    assert!(active);
+}
diff --git a/core/integration/tests/connectors/doris/doris_stream_load_format_bench.rs b/core/integration/tests/connectors/doris/doris_stream_load_format_bench.rs
new file mode 100644
index 0000000000..bd0df2342c
--- /dev/null
+++ b/core/integration/tests/connectors/doris/doris_stream_load_format_bench.rs
@@ -0,0 +1,285 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Manual-PUT benchmark comparing Doris Stream Load throughput for `format=json`
+//! vs `format=csv` over an identical row set. It motivates the connector's opt-in
+//! CSV output (`output_format = "csv"`): JSON parsing on the BE is more expensive
+//! than CSV, so CSV can load the same data faster.
+//!
+//! `#[ignore]`d — it boots/reuses a real Doris container and is run on demand:
+//!   cargo test -p integration -- --ignored \
+//!     connectors::doris::doris_stream_load_format_bench
+//!
+//! It drives loads with the same manual reqwest PUT the dedupe test uses (so it
+//! owns the format headers and compares apples to apples) rather than through
+//! the connector. It asserts only that both formats load exactly N rows; the
+//! json-vs-csv timing delta is reported, never asserted (it is environment
+//! dependent and would flake).
+
+use crate::connectors::fixtures::{DorisOps, DorisSinkFixture};
+use crate::connectors::{TestMessage, create_test_messages};
+use integration::harness::seeds;
+use integration::iggy_harness;
+use serde::Deserialize;
+use sqlx::MySqlPool;
+use std::io::Write as _;
+use std::time::Instant;
+
+const BENCH_ROWS: usize = 10_000;
+const BENCH_ITERS: usize = 5;
+// Mirrors the connector's CSV framing (control-char separators, enclose=", the
+// columns in `TEST_TABLE_DDL_TEMPLATE` order).
+const CSV_SEPARATOR: u8 = 0x01;
+const CSV_LINE_DELIMITER: u8 = 0x02;
+const CSV_COLUMNS: &str = "id, name, count, amount, active, timestamp";
+
+/// The subset of Doris's Stream Load response the benchmark reads. `LoadTimeMs`
+/// is the server-side load time; `Status` gates correctness.
+#[derive(Debug, Deserialize)]
+struct StreamLoadResult {
+    #[serde(rename = "Status")]
+    status: String,
+    #[serde(rename = "LoadTimeMs", default)]
+    load_time_ms: i64,
+}
+
+fn json_body(messages: &[TestMessage]) -> Vec<u8> {
+    serde_json::to_vec(messages).expect("serialize json batch")
+}
+
+fn csv_body(messages: &[TestMessage]) -> Vec<u8> {
+    let mut out = Vec::with_capacity(messages.len() * 64);
+    for message in messages {
+        let _ = write!(out, "{}", message.id);
+        out.push(CSV_SEPARATOR);
+        // The generated names are plain ASCII, but enclose them to mirror the
+        // connector's encoding so the comparison reflects real CSV output.
+        out.push(b'"');
+        out.extend_from_slice(message.name.as_bytes());
+        out.push(b'"');
+        out.push(CSV_SEPARATOR);
+        let _ = write!(out, "{}", message.count);
+        out.push(CSV_SEPARATOR);
+        let _ = write!(out, "{}", message.amount);
+        out.push(CSV_SEPARATOR);
+        out.extend_from_slice(if message.active { b"true" } else { b"false" });
+        out.push(CSV_SEPARATOR);
+        let _ = write!(out, "{}", message.timestamp);
+        out.push(CSV_LINE_DELIMITER);
+    }
+    out
+}
+
+async fn truncate(pool: &MySqlPool, db: &str, table: &str) {
+    sqlx::raw_sql(sqlx::AssertSqlSafe(format!("TRUNCATE TABLE {db}.{table}")))
+        .execute(pool)
+        .await
+        .expect("truncate bench table");
+}
+
+#[allow(clippy::too_many_arguments)]
+async fn stream_load(
+    http: &reqwest::Client,
+    fe_url: &str,
+    db: &str,
+    table: &str,
+    label: &str,
+    csv: bool,
+    body: bytes::Bytes,
+) -> StreamLoadResult {
+    let mut current = format!("{fe_url}/api/{db}/{table}/_stream_load");
+    let mut redirects = 0u8;
+    let response = loop {
+        let mut request = http
+            .put(&current)
+            .basic_auth("root", Some(""))
+            .header(reqwest::header::EXPECT, "100-continue")
+            .header("label", label)
+            .body(body.clone());
+        request = if csv {
+            request
+                .header("format", "csv")
+                .header("column_separator", "\\x01")
+                .header("line_delimiter", "\\x02")
+                .header("enclose", "\"")
+                .header("escape", "\\")
+                .header("columns", CSV_COLUMNS)
+        } else {
+            request
+                .header("format", "json")
+                .header("strip_outer_array", "true")
+        };
+        let resp = request.send().await.expect("stream load PUT");
+        let status = resp.status();
+        if status == reqwest::StatusCode::TEMPORARY_REDIRECT
+            || status == reqwest::StatusCode::PERMANENT_REDIRECT
+        {
+            redirects += 1;
+            assert!(redirects <= 5, "exceeded 5 redirects following Stream Load");
+            current = resp
+                .headers()
+                .get(reqwest::header::LOCATION)
+                .and_then(|v| v.to_str().ok())
+                .expect("Location header")
+                .to_string();
+            continue;
+        }
+        break resp;
+    };
+    let text = response.text().await.expect("stream load body");
+    serde_json::from_str(&text)
+        .unwrap_or_else(|e| panic!("parse stream load response: {e}; body={text}"))
+}
+
+fn median(values: &mut [i64]) -> i64 {
+    values.sort_unstable();
+    values[values.len() / 2]
+}
+
+#[iggy_harness(
+    server(connectors_runtime(config_path = "tests/connectors/doris/sink.toml")),
+    seed = seeds::connector_stream
+)]
+#[ignore = "benchmark: run on demand with --ignored; loads a real Doris container"]
+async fn bench_json_vs_csv_stream_load(_harness: &TestHarness, fixture: DorisSinkFixture) {
+    let db = fixture.database();
+    fixture
+        .create_table(db, "bench_json")
+        .await
+        .expect("create bench_json table");
+    fixture
+        .create_table(db, "bench_csv")
+        .await
+        .expect("create bench_csv table");
+
+    let messages = create_test_messages(BENCH_ROWS);
+    let json = bytes::Bytes::from(json_body(&messages));
+    let csv = bytes::Bytes::from(csv_body(&messages));
+    let json_bytes = json.len();
+    let csv_bytes = csv.len();
+
+    let http = reqwest::Client::builder()
+        .redirect(reqwest::redirect::Policy::none())
+        .build()
+        .unwrap();
+    let fe = fixture.container().fe_url();
+    let pool = fixture.pool().await.expect("pool");
+
+    // Discarded warmup so the first cold load doesn't bias either format.
+    truncate(&pool, db, "bench_json").await;
+    let warmup = stream_load(
+        &http,
+        &fe,
+        db,
+        "bench_json",
+        "iggy_bench_warmup",
+        false,
+        json.clone(),
+    )
+    .await;
+    assert_eq!(warmup.status, "Success", "warmup load failed: {warmup:?}");
+    fixture
+        .wait_for_rows(db, "bench_json", BENCH_ROWS as i64)
+        .await
+        .expect("warmup rows");
+
+    let mut json_load_ms = Vec::with_capacity(BENCH_ITERS);
+    let mut json_wall_ms = Vec::with_capacity(BENCH_ITERS);
+    let mut csv_load_ms = Vec::with_capacity(BENCH_ITERS);
+    let mut csv_wall_ms = Vec::with_capacity(BENCH_ITERS);
+
+    for iter in 0..BENCH_ITERS {
+        // Interleave json then csv each round so neither benefits from warmer
+        // cache / compaction state. Distinct tables + labels avoid dedupe and
+        // the overshoot guard in wait_for_rows.
+        truncate(&pool, db, "bench_json").await;
+        let start = Instant::now();
+        let r = stream_load(
+            &http,
+            &fe,
+            db,
+            "bench_json",
+            &format!("iggy_bench_json_{iter}"),
+            false,
+            json.clone(),
+        )
+        .await;
+        let wall = start.elapsed().as_millis() as i64;
+        assert_eq!(r.status, "Success", "json load failed: {r:?}");
+        fixture
+            .wait_for_rows(db, "bench_json", BENCH_ROWS as i64)
+            .await
+            .expect("json rows");
+        json_load_ms.push(r.load_time_ms);
+        json_wall_ms.push(wall);
+
+        truncate(&pool, db, "bench_csv").await;
+        let start = Instant::now();
+        let r = stream_load(
+            &http,
+            &fe,
+            db,
+            "bench_csv",
+            &format!("iggy_bench_csv_{iter}"),
+            true,
+            csv.clone(),
+        )
+        .await;
+        let wall = start.elapsed().as_millis() as i64;
+        assert_eq!(r.status, "Success", "csv load failed: {r:?}");
+        fixture
+            .wait_for_rows(db, "bench_csv", BENCH_ROWS as i64)
+            .await
+            .expect("csv rows");
+        csv_load_ms.push(r.load_time_ms);
+        csv_wall_ms.push(wall);
+    }
+
+    let json_load = median(&mut json_load_ms);
+    let json_wall = median(&mut json_wall_ms);
+    let csv_load = median(&mut csv_load_ms);
+    let csv_wall = median(&mut csv_wall_ms);
+
+    println!(
+        "\n=== Doris Stream Load: JSON vs CSV ({BENCH_ROWS} rows, {BENCH_ITERS} iters, median) ==="
+    );
+    println!(
+        "  JSON: server LoadTimeMs={json_load}  client_wall={json_wall}ms  body={json_bytes} bytes"
+    );
+    println!(
+        "  CSV : server LoadTimeMs={csv_load}  client_wall={csv_wall}ms  body={csv_bytes} bytes"
+    );
+    let server_delta = json_load - csv_load;
+    let server_pct = if json_load > 0 {
+        100.0 * server_delta as f64 / json_load as f64
+    } else {
+        0.0
+    };
+    println!(
+        "  delta: server {server_delta}ms ({server_pct:.0}% {}), wall {}ms\n",
+        if server_delta >= 0 {
+            "CSV faster"
+        } else {
+            "JSON faster"
+        },
+        json_wall - csv_wall,
+    );
+
+    // Correctness only — timing is reported, never asserted (it flakes).
+    assert_eq!(json_load_ms.len(), BENCH_ITERS);
+    assert_eq!(csv_load_ms.len(), BENCH_ITERS);
+}
diff --git a/core/integration/tests/connectors/doris/mod.rs b/core/integration/tests/connectors/doris/mod.rs
index 58000c8b69..7b1034d81f 100644
--- a/core/integration/tests/connectors/doris/mod.rs
+++ b/core/integration/tests/connectors/doris/mod.rs
@@ -16,3 +16,4 @@
 // under the License.
 
 mod doris_sink;
+mod doris_stream_load_format_bench;
diff --git a/core/integration/tests/connectors/fixtures/doris/container.rs b/core/integration/tests/connectors/fixtures/doris/container.rs
index f435084ec4..6cf89907b8 100644
--- a/core/integration/tests/connectors/fixtures/doris/container.rs
+++ b/core/integration/tests/connectors/fixtures/doris/container.rs
@@ -108,6 +108,7 @@ const ENV_SINK_PASSWORD: &str = "IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_PASSWO
 const ENV_SINK_LABEL_PREFIX: &str = "IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_LABEL_PREFIX";
 const ENV_SINK_MAX_FILTER_RATIO: &str = "IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_MAX_FILTER_RATIO";
 const ENV_SINK_COLUMNS: &str = "IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_COLUMNS";
+const ENV_SINK_FORMAT: &str = "IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_OUTPUT_FORMAT";
 const ENV_SINK_BATCH_SIZE: &str = "IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_BATCH_SIZE";
 const ENV_SINK_STREAMS_0_STREAM: &str = "IGGY_CONNECTORS_SINK_DORIS_STREAMS_0_STREAM";
 const ENV_SINK_STREAMS_0_TOPICS: &str = "IGGY_CONNECTORS_SINK_DORIS_STREAMS_0_TOPICS";
@@ -166,6 +167,11 @@ PROPERTIES (
 pub const COLUMNS_MAPPING_HEADER: &str =
     "id, name, count, amount, active, timestamp, calculated = count + 1";
 
+/// Stream Load `columns` header for `DorisSinkCsvFixture`. CSV is positional, so
+/// the connector needs the explicit column order to map each message's JSON
+/// values into the right CSV fields. Matches `TEST_TABLE_DDL_TEMPLATE`'s order.
+const CSV_COLUMNS_HEADER: &str = "id, name, count, amount, active, timestamp";
+
 /// Linux-only host precheck. macOS / Windows have no `/proc/sys/vm/max_map_count`
 /// (and the all-in-one image is already unusable on macOS for the BE-redirect
 /// routing reason — see the file header). Treat the check as a no-op there; the
@@ -713,3 +719,42 @@ impl TestFixture for DorisSinkColumnsMappingFixture {
         envs
     }
 }
+
+/// Pre-creates the standard test table and configures the connector to emit
+/// `format = "csv"`. CSV is positional, so it also sets the matching `columns`
+/// header; a passing load proves the connector's JSON->CSV serialization
+/// (control-char framing, enclose/escape, column order) round-trips through Doris.
+pub struct DorisSinkCsvFixture {
+    inner: DorisSinkFixture,
+}
+
+impl std::ops::Deref for DorisSinkCsvFixture {
+    type Target = DorisSinkFixture;
+    fn deref(&self) -> &Self::Target {
+        &self.inner
+    }
+}
+
+impl DorisOps for DorisSinkCsvFixture {
+    fn container(&self) -> &DorisContainer {
+        self.inner.container()
+    }
+}
+
+#[async_trait]
+impl TestFixture for DorisSinkCsvFixture {
+    async fn setup() -> Result<Self, TestBinaryError> {
+        let inner = DorisSinkFixture::setup().await?;
+        inner
+            .create_table(inner.container.database(), DEFAULT_TEST_TABLE)
+            .await?;
+        Ok(Self { inner })
+    }
+
+    fn connectors_runtime_envs(&self) -> HashMap<String, String> {
+        let mut envs = self.inner.connectors_runtime_envs();
+        envs.insert(ENV_SINK_FORMAT.to_string(), "csv".to_string());
+        envs.insert(ENV_SINK_COLUMNS.to_string(), CSV_COLUMNS_HEADER.to_string());
+        envs
+    }
+}
diff --git a/core/integration/tests/connectors/fixtures/doris/mod.rs b/core/integration/tests/connectors/fixtures/doris/mod.rs
index 88f5627b34..080b70ebf3 100644
--- a/core/integration/tests/connectors/fixtures/doris/mod.rs
+++ b/core/integration/tests/connectors/fixtures/doris/mod.rs
@@ -18,6 +18,6 @@
 mod container;
 
 pub use container::{
-    DorisOps, DorisSinkColumnsMappingFixture, DorisSinkFixture, DorisSinkMaxFilterRatioFixture,
-    DorisSinkPreCreatedFixture,
+    DorisOps, DorisSinkColumnsMappingFixture, DorisSinkCsvFixture, DorisSinkFixture,
+    DorisSinkMaxFilterRatioFixture, DorisSinkPreCreatedFixture,
 };
diff --git a/core/integration/tests/connectors/fixtures/mod.rs b/core/integration/tests/connectors/fixtures/mod.rs
index 0b2f264d03..dec2584f59 100644
--- a/core/integration/tests/connectors/fixtures/mod.rs
+++ b/core/integration/tests/connectors/fixtures/mod.rs
@@ -46,8 +46,8 @@ pub(crate) fn unique_container_name(service: &str) -> String {
 
 pub use delta::{DeltaFixture, DeltaS3Fixture};
 pub use doris::{
-    DorisOps, DorisSinkColumnsMappingFixture, DorisSinkFixture, DorisSinkMaxFilterRatioFixture,
-    DorisSinkPreCreatedFixture,
+    DorisOps, DorisSinkColumnsMappingFixture, DorisSinkCsvFixture, DorisSinkFixture,
+    DorisSinkMaxFilterRatioFixture, DorisSinkPreCreatedFixture,
 };
 pub use elasticsearch::{ElasticsearchSinkFixture, ElasticsearchSourcePreCreatedFixture};
 pub use http::{