diff --git a/core/connectors/runtime/example_config/connectors/doris_sink.toml b/core/connectors/runtime/example_config/connectors/doris_sink.toml
index 656b0c2b4f..4f7118068e 100644
--- a/core/connectors/runtime/example_config/connectors/doris_sink.toml
+++ b/core/connectors/runtime/example_config/connectors/doris_sink.toml
@@ -41,9 +41,21 @@ 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"
+# 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..2689fa97c4 100644
--- a/core/connectors/sinks/doris_sink/README.md
+++ b/core/connectors/sinks/doris_sink/README.md
@@ -11,19 +11,20 @@ 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 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.
-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.)
+   - 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, `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:
    - `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,9 +39,13 @@ 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()`. |
+| `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. |
 
@@ -84,16 +89,16 @@ 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
 
-- 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.
-- 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..560ba0103d 100644
--- a/core/connectors/sinks/doris_sink/config.toml
+++ b/core/connectors/sinks/doris_sink/config.toml
@@ -41,9 +41,21 @@ 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"
+# 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..2a6744e98b 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,
 };
@@ -26,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};
@@ -58,6 +61,41 @@ 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";
+// 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 {
@@ -85,6 +123,16 @@ 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,
+    // 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)]
@@ -99,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>,
@@ -107,6 +162,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
@@ -197,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());
@@ -312,6 +393,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 +610,105 @@ 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)
+}
+
+/// 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
@@ -599,6 +824,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 +947,46 @@ 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)
+        };
+
+        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,
@@ -718,6 +995,11 @@ 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,
+            format,
+            csv_columns,
         });
 
         info!(
@@ -749,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
@@ -787,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,
@@ -807,39 +1094,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);
                 }
             }
         }
@@ -871,11 +1155,15 @@ mod tests {
             max_filter_ratio: None,
             columns: None,
             where_clause: None,
+            output_format: None,
             timeout: None,
             connect_timeout: None,
             batch_size: None,
             allow_insecure_redirect: None,
             allowed_redirect_hosts: None,
+            max_retries: None,
+            retry_delay: None,
+            max_retry_delay: None,
         }
     }
 
@@ -1178,6 +1466,11 @@ 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),
+            format: Format::Json,
+            csv_columns: Vec::new(),
         }
     }
 
@@ -1357,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;
@@ -1369,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(),
@@ -1541,4 +1836,276 @@ 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:?}",
+        );
+    }
+
+    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::{