diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp
index 9f697a2dcbed81..148128f9890730 100644
--- a/be/src/common/config.cpp
+++ b/be/src/common/config.cpp
@@ -75,6 +75,17 @@ DEFINE_Int32(arrow_flight_sql_port, "8050");
 // Validate Arrow input buffers in opted-in Arrow readers before converting them to Doris columns.
 DEFINE_Bool(enable_arrow_input_validation, "true");
 
+// Max bytes of a single utf8/binary (int32-offset) column that one Arrow Flight result
+// RecordBatch may carry. Arrow's BaseBinaryBuilder rejects a value buffer larger than
+// memory_limit() = INT32_MAX - 1. The splitter cuts a batch before a column reaches this value
+// (running + row >= limit), so with the default INT32_MAX a committed batch holds at most
+// INT32_MAX - 1 bytes -- exactly the builder limit. When a block would exceed it, the Arrow
+// Flight readers split it by rows. Only lower it (e.g. in tests) to exercise the split path
+// without materializing 2GB of data.
+DEFINE_mInt64(arrow_flight_result_max_utf8_bytes, "2147483647");
+DEFINE_Validator(arrow_flight_result_max_utf8_bytes,
+                 [](const int64_t config) -> bool { return config > 0 && config <= 2147483647; });
+
 DEFINE_Int32(cdc_client_port, "9096");
 
 DEFINE_String(cdc_client_java_opts, "");
diff --git a/be/src/common/config.h b/be/src/common/config.h
index f5c72eb776d5d3..1ad53014102d33 100644
--- a/be/src/common/config.h
+++ b/be/src/common/config.h
@@ -122,6 +122,12 @@ DECLARE_Int32(arrow_flight_sql_port);
 // Validate Arrow input buffers in opted-in Arrow readers before converting them to Doris columns.
 DECLARE_Bool(enable_arrow_input_validation);
 
+// Max bytes of a single utf8/binary (int32-offset) column that one Arrow Flight result
+// RecordBatch may carry. Arrow's builder rejects a value buffer larger than INT32_MAX - 1, so
+// each batch must stay at or below that; when a block would exceed it the Arrow Flight readers
+// split it by rows. Only lowered in tests to exercise the split path without materializing 2GB.
+DECLARE_mInt64(arrow_flight_result_max_utf8_bytes);
+
 // port for cdc client scan oltp cdc data
 DECLARE_Int32(cdc_client_port);
 
diff --git a/be/src/core/data_type_serde/data_type_jsonb_serde.cpp b/be/src/core/data_type_serde/data_type_jsonb_serde.cpp
index e53e4d23a748c8..02bc02e6e03ec2 100644
--- a/be/src/core/data_type_serde/data_type_jsonb_serde.cpp
+++ b/be/src/core/data_type_serde/data_type_jsonb_serde.cpp
@@ -19,6 +19,7 @@
 
 #include <cstddef>
 #include <cstdint>
+#include <limits>
 #include <memory>
 
 #include "arrow/array/builder_binary.h"
@@ -107,10 +108,19 @@ Status DataTypeJsonbSerDe::write_column_to_arrow(const IColumn& column, const Nu
         std::string_view string_ref = string_column.get_data_at(string_i).to_string_view();
         std::string json_string =
                 JsonbToJson::jsonb_to_json_string(string_ref.data(), string_ref.size());
-        RETURN_IF_ERROR(
-                checkArrowStatus(builder.Append(json_string.data(),
-                                                cast_set<int, size_t, false>(json_string.size())),
-                                 column, *array_builder));
+        // JSONB is stored as binary but rendered to (usually larger) JSON text here, so the
+        // physical column size does not bound the arrow payload. Reject before narrowing
+        // size_t -> int32: an unchecked cast of a length > INT32_MAX would wrap to a bogus
+        // small/negative value that Arrow silently accepts, corrupting the result.
+        if (json_string.size() > std::numeric_limits<int32_t>::max()) {
+            return Status::InternalError(
+                    "JSONB value rendered to {} bytes of JSON text, exceeding the Arrow utf8 "
+                    "int32 offset limit ({}); cannot be sent over Arrow Flight.",
+                    json_string.size(), std::numeric_limits<int32_t>::max());
+        }
+        RETURN_IF_ERROR(checkArrowStatus(
+                builder.Append(json_string.data(), static_cast<int>(json_string.size())), column,
+                *array_builder));
     }
     return Status::OK();
 }
diff --git a/be/src/format/arrow/arrow_block_convertor.cpp b/be/src/format/arrow/arrow_block_convertor.cpp
index d6a4735c73ade8..9ae2d847b8aa16 100644
--- a/be/src/format/arrow/arrow_block_convertor.cpp
+++ b/be/src/format/arrow/arrow_block_convertor.cpp
@@ -39,6 +39,7 @@
 #include <utility>
 #include <vector>
 
+#include "common/config.h"
 #include "common/status.h"
 #include "core/block/column_with_type_and_name.h"
 #include "core/column/column.h"
@@ -193,13 +194,14 @@ Status FromBlockToRecordBatchConverter::convert(std::shared_ptr<arrow::RecordBat
         _cur_col = _block.get_by_position(idx).column;
         _cur_type = _block.get_by_position(idx).type;
         auto column = _cur_col->convert_to_full_column_if_const();
+        // The Arrow schema is authoritative: build each array with exactly the field's declared
+        // type. (Previously a string/binary column whose data reached 2GB was silently upgraded
+        // here to large_utf8/large_binary while the schema kept utf8/binary, producing a batch
+        // whose int64 array offsets disagreed with its int32 schema -> data corruption on generic
+        // clients and negative-length crashes on the Doris receiver. Oversized columns are now
+        // handled up front by convert_to_arrow_batches, which splits the block by rows so every
+        // batch stays within the int32 offset limit.)
         auto arrow_type = _schema->field(idx)->type();
-        if (arrow_type->id() == arrow::Type::STRING && column->byte_size() >= MAX_ARROW_UTF8) {
-            arrow_type = arrow::large_utf8();
-        } else if (arrow_type->id() == arrow::Type::BINARY &&
-                   column->byte_size() >= MAX_ARROW_UTF8) {
-            arrow_type = arrow::large_binary();
-        }
         std::unique_ptr<arrow::ArrayBuilder> builder;
         auto arrow_st = arrow::MakeBuilder(_pool, arrow_type, &builder);
         if (!arrow_st.ok()) {
@@ -227,6 +229,11 @@ Status FromBlockToRecordBatchConverter::convert(std::shared_ptr<arrow::RecordBat
         }
     }
     *out = arrow::RecordBatch::Make(_schema, actual_rows, std::move(_arrays));
+    // Defense-in-depth: guarantee the batch's array types match the schema (types + lengths).
+    // After removing the silent large_utf8 upgrade this always holds; keep the check so any
+    // future schema/array divergence fails loudly here instead of silently corrupting or
+    // crashing a downstream reader.
+    RETURN_IF_ERROR(to_doris_status((*out)->Validate()));
     return Status::OK();
 }
 
@@ -270,6 +277,122 @@ Status convert_to_arrow_batch(const Block& block, const std::shared_ptr<arrow::S
     return converter.convert(result);
 }
 
+namespace {
+
+// Only utf8/binary use int32 offsets and can overflow at 2^31; fixed-size and large_* cannot.
+bool is_int32_offset_binary_type(const std::shared_ptr<arrow::DataType>& type) {
+    return type->id() == arrow::Type::STRING || type->id() == arrow::Type::BINARY;
+}
+
+// Byte length written into the arrow string/binary array for `row` (0 for null). Only called for
+// raw string/binary columns (see convert_to_arrow_batches), whose SerDe appends get_data_at(row)
+// verbatim, so this equals the emitted arrow payload.
+size_t arrow_value_byte_size_at(const IColumn& column, size_t row) {
+    const IColumn* data = &column;
+    if (const auto* nullable = check_and_get_column<ColumnNullable>(column)) {
+        if (nullable->is_null_at(row)) {
+            return 0;
+        }
+        data = &nullable->get_nested_column();
+    }
+    return data->get_data_at(row).size;
+}
+
+} // namespace
+
+Status convert_to_arrow_batches(const Block& block, const std::shared_ptr<arrow::Schema>& schema,
+                                arrow::MemoryPool* pool,
+                                std::vector<std::shared_ptr<arrow::RecordBatch>>* results,
+                                const cctz::time_zone& timezone_obj) {
+    results->clear();
+    const int num_fields = schema->num_fields();
+    if (block.columns() != static_cast<size_t>(num_fields)) {
+        return Status::InvalidArgument("number fields not match");
+    }
+    const size_t num_rows = block.rows();
+    const auto max_bytes = static_cast<size_t>(config::arrow_flight_result_max_utf8_bytes);
+
+    // Fast path: O(#string cols), each byte_size() is O(1) and >= the actual arrow data size.
+    // If no int32-offset string/binary column reaches the limit, emit the whole block as one
+    // batch -- zero added cost on the normal path, and this never misses an overflow.
+    std::vector<int> big_fields;
+    for (int idx = 0; idx < num_fields; ++idx) {
+        if (!is_int32_offset_binary_type(schema->field(idx)->type())) {
+            continue;
+        }
+        // Only raw string/binary types append get_data_at(row) verbatim, so physical byte_size
+        // equals the arrow payload and per-row splitting is exact. Transform-serde types that
+        // also map to utf8/binary (JSONB -> JSON text, Variant, largeint/date/datetime -> text)
+        // do not, and Variant's column has no get_data_at at all. They stay on the fast path and
+        // rely on their SerDe's checked size cast plus the Arrow builder capacity check to error
+        // cleanly (never silently corrupt) if a value or batch overflows int32.
+        const auto pt = remove_nullable(block.get_by_position(idx).type)->get_primitive_type();
+        if (!is_string_type(pt) && pt != TYPE_VARBINARY) {
+            continue;
+        }
+        if (block.get_by_position(idx).column->byte_size() >= max_bytes) {
+            big_fields.push_back(idx);
+        }
+    }
+    if (big_fields.empty() || num_rows == 0) {
+        std::shared_ptr<arrow::RecordBatch> batch;
+        RETURN_IF_ERROR(convert_to_arrow_batch(block, schema, pool, &batch, timezone_obj));
+        results->push_back(std::move(batch));
+        return Status::OK();
+    }
+
+    // Materialize the (rarely) const columns once so per-row size probing is cheap and correct.
+    std::vector<ColumnPtr> big_cols;
+    big_cols.reserve(big_fields.size());
+    for (int idx : big_fields) {
+        big_cols.push_back(block.get_by_position(idx).column->convert_to_full_column_if_const());
+    }
+
+    // Walk rows, cutting a batch right before any tracked column would reach the int32 limit;
+    // the tightest-growing column decides each cut.
+    std::vector<size_t> running(big_cols.size());
+    std::vector<size_t> row_bytes(big_cols.size());
+    size_t start = 0;
+    while (start < num_rows) {
+        running.assign(running.size(), 0);
+        size_t end = start;
+        while (end < num_rows) {
+            bool fits = true;
+            for (size_t c = 0; c < big_cols.size(); ++c) {
+                const size_t rb = arrow_value_byte_size_at(*big_cols[c], end);
+                if (rb >= max_bytes) {
+                    return Status::InternalError(
+                            "Arrow Flight cannot send column '{}': a single value at row {} is {} "
+                            "bytes, exceeding the Arrow utf8/binary int32 offset limit of {} "
+                            "bytes.",
+                            schema->field(big_fields[c])->name(), end, rb, max_bytes);
+                }
+                row_bytes[c] = rb;
+                if (running[c] + rb >= max_bytes) {
+                    fits = false;
+                    break;
+                }
+            }
+            if (!fits) {
+                break;
+            }
+            for (size_t c = 0; c < big_cols.size(); ++c) {
+                running[c] += row_bytes[c];
+            }
+            ++end;
+        }
+        // The start row always fits alone (an oversized single value would have errored above),
+        // so `end` advances by at least one and the outer loop terminates.
+        DCHECK_GT(end, start);
+        std::shared_ptr<arrow::RecordBatch> batch;
+        RETURN_IF_ERROR(
+                convert_to_arrow_batch(block, schema, pool, &batch, timezone_obj, start, end));
+        results->push_back(std::move(batch));
+        start = end;
+    }
+    return Status::OK();
+}
+
 Status make_zero_column_arrow_batch(const std::shared_ptr<arrow::Schema>& schema, int64_t rows,
                                     std::shared_ptr<arrow::RecordBatch>* result) {
     if (schema->num_fields() != 0) {
diff --git a/be/src/format/arrow/arrow_block_convertor.h b/be/src/format/arrow/arrow_block_convertor.h
index 96ee10d5215760..fd7595f3f5c455 100644
--- a/be/src/format/arrow/arrow_block_convertor.h
+++ b/be/src/format/arrow/arrow_block_convertor.h
@@ -21,6 +21,7 @@
 
 #include <cstdint>
 #include <memory>
+#include <vector>
 
 #include "common/status.h"
 #include "core/block/block.h"
@@ -117,6 +118,16 @@ Status convert_to_arrow_batch(const Block& block, const std::shared_ptr<arrow::S
                               const cctz::time_zone& timezone_obj, size_t start_row,
                               size_t end_row);
 
+// Convert a Block to one or more Arrow RecordBatches for the Arrow Flight result path, whose
+// stream schema is fixed to utf8/binary (int32 offsets). If any string/binary column's data in
+// this block would reach config::arrow_flight_result_max_utf8_bytes, the block is split by rows
+// so every emitted batch stays within the int32 offset limit. Returns an error if a single value
+// alone exceeds the limit (it cannot be represented with int32 offsets).
+Status convert_to_arrow_batches(const Block& block, const std::shared_ptr<arrow::Schema>& schema,
+                                arrow::MemoryPool* pool,
+                                std::vector<std::shared_ptr<arrow::RecordBatch>>* results,
+                                const cctz::time_zone& timezone_obj);
+
 Status make_zero_column_arrow_batch(const std::shared_ptr<arrow::Schema>& schema, int64_t rows,
                                     std::shared_ptr<arrow::RecordBatch>* result);
 
diff --git a/be/src/format/arrow/arrow_row_batch.h b/be/src/format/arrow/arrow_row_batch.h
index e7b77ed707b772..ad147b9e817279 100644
--- a/be/src/format/arrow/arrow_row_batch.h
+++ b/be/src/format/arrow/arrow_row_batch.h
@@ -39,8 +39,6 @@ class Schema;
 
 namespace doris {
 
-constexpr size_t MAX_ARROW_UTF8 = (1ULL << 31); // 2G
-
 class RowDescriptor;
 
 Status convert_to_arrow_type(const DataTypePtr& type, std::shared_ptr<arrow::DataType>* result,
diff --git a/be/src/service/arrow_flight/arrow_flight_batch_reader.cpp b/be/src/service/arrow_flight/arrow_flight_batch_reader.cpp
index 8dbc426b882c53..50d7ef2f663a06 100644
--- a/be/src/service/arrow_flight/arrow_flight_batch_reader.cpp
+++ b/be/src/service/arrow_flight/arrow_flight_batch_reader.cpp
@@ -94,6 +94,14 @@ arrow::Status ArrowFlightBatchLocalReader::ReadNextImpl(std::shared_ptr<arrow::R
     // parameter *out not nullptr
     *out = nullptr;
     SCOPED_ATTACH_TASK(_mem_tracker);
+
+    // Drain any batches left over from splitting the previous block.
+    if (!_pending_batches.empty()) {
+        *out = std::move(_pending_batches.front());
+        _pending_batches.pop_front();
+        return arrow::Status::OK();
+    }
+
     TUniqueId tid = UniqueId(_statement->query_id).to_thrift();
     std::shared_ptr<ArrowFlightResultBlockBuffer> arrow_buffer;
     RETURN_ARROW_STATUS_IF_ERROR(
@@ -108,16 +116,23 @@ arrow::Status ArrowFlightBatchLocalReader::ReadNextImpl(std::shared_ptr<arrow::R
     }
 
     {
-        // convert one batch
+        // convert one block, splitting into multiple batches if a string/binary column would
+        // overflow the int32 offset limit
         SCOPED_ATOMIC_TIMER(&_convert_arrow_batch_timer);
-        st = convert_to_arrow_batch(*result, _schema, arrow::default_memory_pool(), out,
-                                    _timezone_obj);
+        std::vector<std::shared_ptr<arrow::RecordBatch>> batches;
+        st = convert_to_arrow_batches(*result, _schema, arrow::default_memory_pool(), &batches,
+                                      _timezone_obj);
         st.prepend("ArrowFlightBatchLocalReader convert block to arrow batch failed");
         ARROW_RETURN_NOT_OK(to_arrow_status(st));
+        for (auto& batch : batches) {
+            _pending_batches.push_back(std::move(batch));
+        }
     }
 
     _packet_seq++;
-    if (*out != nullptr) {
+    if (!_pending_batches.empty()) {
+        *out = std::move(_pending_batches.front());
+        _pending_batches.pop_front();
         VLOG_NOTICE << "ArrowFlightBatchLocalReader read next: " << (*out)->num_rows() << ", "
                     << (*out)->num_columns() << ", packet_seq: " << _packet_seq;
     }
@@ -295,22 +310,37 @@ arrow::Status ArrowFlightBatchRemoteReader::ReadNextImpl(std::shared_ptr<arrow::
     // parameter *out not nullptr
     *out = nullptr;
     SCOPED_ATTACH_TASK(_mem_tracker);
+
+    // Drain any batches left over from splitting the previous block.
+    if (!_pending_batches.empty()) {
+        *out = std::move(_pending_batches.front());
+        _pending_batches.pop_front();
+        return arrow::Status::OK();
+    }
+
     ARROW_RETURN_NOT_OK(_fetch_data());
     if (_block == nullptr) {
         // eof, normal path end, last _fetch_data return block is nullptr
         return arrow::Status::OK();
     }
     {
-        // convert one batch
+        // convert one block, splitting into multiple batches if a string/binary column would
+        // overflow the int32 offset limit
         SCOPED_ATOMIC_TIMER(&_convert_arrow_batch_timer);
-        auto st = convert_to_arrow_batch(*_block, _schema, arrow::default_memory_pool(), out,
-                                         _timezone_obj);
+        std::vector<std::shared_ptr<arrow::RecordBatch>> batches;
+        auto st = convert_to_arrow_batches(*_block, _schema, arrow::default_memory_pool(), &batches,
+                                           _timezone_obj);
         st.prepend("ArrowFlightBatchRemoteReader convert block to arrow batch failed");
         ARROW_RETURN_NOT_OK(to_arrow_status(st));
+        for (auto& batch : batches) {
+            _pending_batches.push_back(std::move(batch));
+        }
     }
     _block = nullptr;
 
-    if (*out != nullptr) {
+    if (!_pending_batches.empty()) {
+        *out = std::move(_pending_batches.front());
+        _pending_batches.pop_front();
         VLOG_NOTICE << "ArrowFlightBatchRemoteReader read next: " << (*out)->num_rows() << ", "
                     << (*out)->num_columns() << ", packet_seq: " << _packet_seq;
     }
diff --git a/be/src/service/arrow_flight/arrow_flight_batch_reader.h b/be/src/service/arrow_flight/arrow_flight_batch_reader.h
index f72a35a8479f0c..a11d3a81ea96a2 100644
--- a/be/src/service/arrow_flight/arrow_flight_batch_reader.h
+++ b/be/src/service/arrow_flight/arrow_flight_batch_reader.h
@@ -20,6 +20,7 @@
 #include <cctz/time_zone.h>
 #include <gen_cpp/Types_types.h>
 
+#include <deque>
 #include <memory>
 #include <utility>
 
@@ -62,6 +63,11 @@ class ArrowFlightBatchReaderBase : public arrow::RecordBatchReader {
     std::atomic<int64_t> _convert_arrow_batch_timer = 0;
     std::atomic<int64_t> _deserialize_block_timer = 0;
     std::shared_ptr<MemTrackerLimiter> _mem_tracker;
+
+    // Batches produced from the current block but not yet returned. A single result block may be
+    // split into multiple RecordBatches (see convert_to_arrow_batches) when a string/binary
+    // column would overflow the int32 offset limit; ReadNext drains this queue one at a time.
+    std::deque<std::shared_ptr<arrow::RecordBatch>> _pending_batches;
 };
 
 class ArrowFlightBatchLocalReader : public ArrowFlightBatchReaderBase {
diff --git a/be/test/core/data_type_serde/data_type_serde_arrow_test.cpp b/be/test/core/data_type_serde/data_type_serde_arrow_test.cpp
index 7b20c2f82d0c65..e54889a3cb9dcf 100644
--- a/be/test/core/data_type_serde/data_type_serde_arrow_test.cpp
+++ b/be/test/core/data_type_serde/data_type_serde_arrow_test.cpp
@@ -46,6 +46,7 @@
 #include <utility>
 #include <vector>
 
+#include "common/config.h"
 #include "core/block/block.h"
 #include "core/column/column.h"
 #include "core/column/column_complex.h"
@@ -637,4 +638,216 @@ TEST(DataTypeSerDeArrowTest, BlockConverterTest) {
     block_converter_test(cols, 7, false);
 }
 
+// A utf8 column whose data exceeds the (test-lowered) int32 offset limit must be split by rows
+// into several batches, each keeping the fixed utf8 schema, and the batches must reconstruct the
+// original rows byte-for-byte in order.
+TEST(DataTypeSerDeArrowTest, ConvertToArrowBatchesSplitsUtf8ByRows) {
+    std::vector<std::string> values;
+    for (int i = 0; i < 10; ++i) {
+        std::string v(9, 'x');
+        v.push_back(static_cast<char>('0' + i)); // 10-byte distinct values
+        values.push_back(v);
+    }
+    auto strcol = ColumnString::create();
+    for (const auto& v : values) {
+        strcol->insert_data(v.data(), v.size());
+    }
+    auto block = std::make_shared<Block>();
+    block->insert(
+            ColumnWithTypeAndName(strcol->get_ptr(), std::make_shared<DataTypeString>(), "s"));
+    auto schema = arrow::schema({arrow::field("s", arrow::utf8(), false)});
+
+    // 32-byte cap => at most three 10-byte values per batch.
+    const int64_t saved = config::arrow_flight_result_max_utf8_bytes;
+    config::arrow_flight_result_max_utf8_bytes = 32;
+    std::vector<std::shared_ptr<arrow::RecordBatch>> batches;
+    cctz::time_zone tz;
+    Status status =
+            convert_to_arrow_batches(*block, schema, arrow::default_memory_pool(), &batches, tz);
+    config::arrow_flight_result_max_utf8_bytes = saved;
+
+    ASSERT_TRUE(status.ok()) << status;
+    ASSERT_GT(batches.size(), 1U) << "expected the oversized block to be split";
+
+    std::vector<std::string> restored;
+    for (const auto& batch : batches) {
+        ASSERT_EQ(batch->schema()->field(0)->type()->id(), arrow::Type::STRING);
+        ASSERT_TRUE(batch->Validate().ok());
+        auto arr = std::static_pointer_cast<arrow::StringArray>(batch->column(0));
+        ASSERT_LT(arr->value_offset(arr->length()), 32) << "a batch reached the int32 offset cap";
+        for (int64_t r = 0; r < arr->length(); ++r) {
+            restored.push_back(arr->GetString(r));
+        }
+    }
+    EXPECT_EQ(restored, values) << "split batches must reconstruct the original rows in order";
+}
+
+// Under the default (2GB) limit a small block is emitted as a single batch (no split, no cost).
+TEST(DataTypeSerDeArrowTest, ConvertToArrowBatchesNoSplitUnderLimit) {
+    auto strcol = ColumnString::create();
+    strcol->insert_data("hello", 5);
+    strcol->insert_data("world", 5);
+    auto block = std::make_shared<Block>();
+    block->insert(
+            ColumnWithTypeAndName(strcol->get_ptr(), std::make_shared<DataTypeString>(), "s"));
+    auto schema = arrow::schema({arrow::field("s", arrow::utf8(), false)});
+
+    std::vector<std::shared_ptr<arrow::RecordBatch>> batches;
+    cctz::time_zone tz;
+    Status status =
+            convert_to_arrow_batches(*block, schema, arrow::default_memory_pool(), &batches, tz);
+    ASSERT_TRUE(status.ok()) << status;
+    ASSERT_EQ(batches.size(), 1U);
+    auto arr = std::static_pointer_cast<arrow::StringArray>(batches[0]->column(0));
+    ASSERT_EQ(arr->length(), 2);
+    EXPECT_EQ(arr->GetString(0), "hello");
+    EXPECT_EQ(arr->GetString(1), "world");
+}
+
+// A single value that alone exceeds the limit cannot be represented with int32 offsets; the
+// converter must return a clear error rather than crashing or corrupting data.
+TEST(DataTypeSerDeArrowTest, ConvertToArrowBatchesSingleValueExceedsLimitErrors) {
+    auto strcol = ColumnString::create();
+    strcol->insert_data("0123456789abcdef", 16); // one 16-byte value
+    auto block = std::make_shared<Block>();
+    block->insert(
+            ColumnWithTypeAndName(strcol->get_ptr(), std::make_shared<DataTypeString>(), "s"));
+    auto schema = arrow::schema({arrow::field("s", arrow::utf8(), false)});
+
+    const int64_t saved = config::arrow_flight_result_max_utf8_bytes;
+    config::arrow_flight_result_max_utf8_bytes = 8; // 16-byte value cannot fit in 8
+    std::vector<std::shared_ptr<arrow::RecordBatch>> batches;
+    cctz::time_zone tz;
+    Status status =
+            convert_to_arrow_batches(*block, schema, arrow::default_memory_pool(), &batches, tz);
+    config::arrow_flight_result_max_utf8_bytes = saved;
+
+    ASSERT_FALSE(status.ok()) << "a single oversized value must be a clear error, not a crash";
+}
+
+// With two utf8 columns of different widths, the tighter-growing column decides each cut point.
+TEST(DataTypeSerDeArrowTest, ConvertToArrowBatchesSplitDrivenByTightestColumn) {
+    auto cola = ColumnString::create();
+    auto colb = ColumnString::create();
+    std::vector<std::string> a_vals;
+    std::vector<std::string> b_vals;
+    for (int i = 0; i < 5; ++i) {
+        std::string a(9, 'a');
+        a.push_back(static_cast<char>('0' + i)); // 10 bytes
+        std::string b(19, 'b');
+        b.push_back(static_cast<char>('0' + i)); // 20 bytes
+        a_vals.push_back(a);
+        b_vals.push_back(b);
+        cola->insert_data(a.data(), a.size());
+        colb->insert_data(b.data(), b.size());
+    }
+    auto block = std::make_shared<Block>();
+    block->insert(ColumnWithTypeAndName(cola->get_ptr(), std::make_shared<DataTypeString>(), "a"));
+    block->insert(ColumnWithTypeAndName(colb->get_ptr(), std::make_shared<DataTypeString>(), "b"));
+    auto schema = arrow::schema(
+            {arrow::field("a", arrow::utf8(), false), arrow::field("b", arrow::utf8(), false)});
+
+    const int64_t saved = config::arrow_flight_result_max_utf8_bytes;
+    config::arrow_flight_result_max_utf8_bytes = 32;
+    std::vector<std::shared_ptr<arrow::RecordBatch>> batches;
+    cctz::time_zone tz;
+    Status status =
+            convert_to_arrow_batches(*block, schema, arrow::default_memory_pool(), &batches, tz);
+    config::arrow_flight_result_max_utf8_bytes = saved;
+
+    ASSERT_TRUE(status.ok()) << status;
+    // col b at 20 bytes/row: two rows (40) exceed 32, so each batch holds exactly one row.
+    ASSERT_EQ(batches.size(), 5U);
+    std::vector<std::string> ra;
+    std::vector<std::string> rb;
+    for (const auto& batch : batches) {
+        ASSERT_TRUE(batch->Validate().ok());
+        auto arr_a = std::static_pointer_cast<arrow::StringArray>(batch->column(0));
+        auto arr_b = std::static_pointer_cast<arrow::StringArray>(batch->column(1));
+        for (int64_t r = 0; r < arr_a->length(); ++r) {
+            ra.push_back(arr_a->GetString(r));
+            rb.push_back(arr_b->GetString(r));
+        }
+    }
+    EXPECT_EQ(ra, a_vals);
+    EXPECT_EQ(rb, b_vals);
+}
+
+// Boundary: with threshold N, a value of N-1 bytes fits in one batch (the most a batch may
+// hold); a value of exactly N bytes cannot fit int32 offsets and must error. This mirrors the
+// default threshold INT32_MAX vs the Arrow builder limit INT32_MAX - 1.
+TEST(DataTypeSerDeArrowTest, ConvertToArrowBatchesThresholdBoundary) {
+    const int64_t saved = config::arrow_flight_result_max_utf8_bytes;
+    cctz::time_zone tz;
+    auto schema = arrow::schema({arrow::field("s", arrow::utf8(), false)});
+    auto make_block = [](size_t value_len) {
+        auto strcol = ColumnString::create();
+        std::string v(value_len, 'a');
+        strcol->insert_data(v.data(), v.size());
+        auto block = std::make_shared<Block>();
+        block->insert(
+                ColumnWithTypeAndName(strcol->get_ptr(), std::make_shared<DataTypeString>(), "s"));
+        return block;
+    };
+
+    // value == threshold - 1 -> fits in a single batch.
+    {
+        auto block = make_block(15);
+        config::arrow_flight_result_max_utf8_bytes = 16;
+        std::vector<std::shared_ptr<arrow::RecordBatch>> batches;
+        Status st = convert_to_arrow_batches(*block, schema, arrow::default_memory_pool(), &batches,
+                                             tz);
+        config::arrow_flight_result_max_utf8_bytes = saved;
+        ASSERT_TRUE(st.ok()) << st;
+        ASSERT_EQ(batches.size(), 1U);
+        auto arr = std::static_pointer_cast<arrow::StringArray>(batches[0]->column(0));
+        EXPECT_EQ(arr->value_offset(arr->length()), 15);
+    }
+    // value == threshold -> a single value cannot fit -> clear error.
+    {
+        auto block = make_block(16);
+        config::arrow_flight_result_max_utf8_bytes = 16;
+        std::vector<std::shared_ptr<arrow::RecordBatch>> batches;
+        Status st = convert_to_arrow_batches(*block, schema, arrow::default_memory_pool(), &batches,
+                                             tz);
+        config::arrow_flight_result_max_utf8_bytes = saved;
+        ASSERT_FALSE(st.ok()) << "value of exactly threshold bytes must error";
+    }
+}
+
+// A transform-serde type (LARGEINT -> utf8 text) must NOT be split by physical byte_size: even
+// when its physical size exceeds a lowered threshold it stays on the fast path (one batch),
+// because get_data_at() bytes are not the emitted arrow payload for such types (and calling
+// get_data_at on e.g. a Variant column would throw).
+TEST(DataTypeSerDeArrowTest, ConvertToArrowBatchesDoesNotSplitTransformType) {
+    DataTypePtr type = DataTypeFactory::instance().create_data_type(TYPE_LARGEINT, false);
+    auto mcol = type->create_column();
+    mcol->insert(Field::create_field<TYPE_LARGEINT>(Int128(1)));
+    mcol->insert(Field::create_field<TYPE_LARGEINT>(Int128(22)));
+    mcol->insert(Field::create_field<TYPE_LARGEINT>(Int128(333)));
+    auto block = std::make_shared<Block>();
+    block->insert(ColumnWithTypeAndName(mcol->get_ptr(), type, "n"));
+
+    std::shared_ptr<arrow::Schema> schema;
+    Status sst = get_arrow_schema_from_block(*block, &schema, TimezoneUtils::default_time_zone);
+    ASSERT_TRUE(sst.ok()) << sst;
+    ASSERT_EQ(schema->field(0)->type()->id(), arrow::Type::STRING); // largeint -> utf8
+
+    const int64_t saved = config::arrow_flight_result_max_utf8_bytes;
+    config::arrow_flight_result_max_utf8_bytes = 20; // < physical byte_size (3 * 16 = 48 bytes)
+    std::vector<std::shared_ptr<arrow::RecordBatch>> batches;
+    cctz::time_zone tz;
+    Status st =
+            convert_to_arrow_batches(*block, schema, arrow::default_memory_pool(), &batches, tz);
+    config::arrow_flight_result_max_utf8_bytes = saved;
+
+    ASSERT_TRUE(st.ok()) << st;
+    ASSERT_EQ(batches.size(), 1U) << "transform type must not be split by physical byte_size";
+    auto arr = std::static_pointer_cast<arrow::StringArray>(batches[0]->column(0));
+    ASSERT_EQ(arr->length(), 3);
+    EXPECT_EQ(arr->GetString(0), "1");
+    EXPECT_EQ(arr->GetString(1), "22");
+    EXPECT_EQ(arr->GetString(2), "333");
+}
+
 } // namespace doris
diff --git a/regression-test/suites/arrow_flight_sql_p0/test_select.groovy b/regression-test/suites/arrow_flight_sql_p0/test_select.groovy
index 85f119fc2c3a9b..5105ac03275ba0 100644
--- a/regression-test/suites/arrow_flight_sql_p0/test_select.groovy
+++ b/regression-test/suites/arrow_flight_sql_p0/test_select.groovy
@@ -62,7 +62,10 @@ suite("test_select", "arrow_flight_sql") {
         SELECT 4, CAST(CONCAT('{"large":"', REPEAT('x', ${largeJsonValueSize}), '"}') AS JSONB)
         """
 
-    // This row exceeds MAX_ARROW_UTF8 and exercises JSONB -> LargeString serialization.
+    // A ~2.1MB JSONB value. This is far below the 2GB (INT32_MAX) Arrow utf8/binary offset limit,
+    // so it round-trips through the normal utf8 path in a single Arrow batch -- it does NOT
+    // exercise the large-string / batch-split path. The >2GB split path is covered by BE unit
+    // tests (data_type_serde_arrow_test.cpp) via a lowered arrow_flight_result_max_utf8_bytes.
     def largeJsonbResult = arrow_flight_sql """
         select payload, length(cast(payload as string)) from ${tableName} where id = 4
         """