[fix](arrow-flight) Fix utf8/large string schema-array mismatch for oversized columns

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, "");

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);
 

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();
 }

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) {

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);
 

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,

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;
     }

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 {