github-actions[bot] commented on code in PR #65789:
URL: https://github.com/apache/doris/pull/65789#discussion_r3610670147
##########
be/src/format/arrow/arrow_block_convertor.cpp:
##########
@@ -270,6 +277,109 @@ 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), mirroring
+// DataTypeStringSerDeBase::write_column_to_arrow which appends
get_data_at(row).size bytes.
+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()) &&
Review Comment:
[P1] Measure the bytes that each SerDe actually appends
This gate assumes physical `IColumn::byte_size()`/`get_data_at()` matches
the Arrow payload, but several types with an `utf8`/`binary` schema transform
their data. For example, binary JSONB below 2 GiB can render above `INT32_MAX`
after escaping; `DataTypeJsonbSerDe` then narrows that size with `cast_set<int,
size_t, false>`, producing a non-positive length on supported targets. Arrow 17
treats that as a valid empty value, so `Validate()` passes and local Flight
silently corrupts the result. Variant, formatted fixed-width types, and
Bitmap/HLL/QuantileState have the same storage-vs-SerDe mismatch. Please base
cuts on the emitted SerDe payload and reject before any int32 narrowing, with
JSONB/Variant/complex-type tests.
##########
be/src/common/config.cpp:
##########
@@ -75,6 +75,15 @@ 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 allowed in one
Arrow Flight result
+// RecordBatch. Arrow utf8/binary offsets are int32, so a column's data in one
batch must stay
+// below 2^31 (= INT32_MAX + 1). When a block would exceed this, the Arrow
Flight readers split
+// it by rows into multiple batches so each batch stays valid. Default is the
hard 2^31 limit;
+// 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, "2147483648");
Review Comment:
[P2] Match Arrow 17's actual builder capacity
Doris pins Arrow 17, where `BaseBinaryBuilder::memory_limit()` is `INT32_MAX
- 1` and `ValidateOverflow` rejects anything larger. This exclusive cap is
`INT32_MAX + 1`, and the splitter only cuts on `>= max_bytes`, so an ordinary
String/Varbinary payload totaling exactly `INT32_MAX` is passed through and
then fails with `CapacityError` instead of being split. Please cap the
exclusive threshold at `INT32_MAX` (so emitted data is at most `INT32_MAX - 1`)
and add an exact boundary test.
##########
be/src/format/arrow/arrow_block_convertor.cpp:
##########
@@ -270,6 +277,109 @@ 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;
Review Comment:
[P1] Recurse into nested int32-offset builders
This predicate only recognizes a top-level `STRING`/`BINARY`. `ARRAY`,
`MAP`, and `STRUCT` can contain shared child String/Binary builders (and
List/Map themselves have int32 element offsets), but those fields always take
the fast path. For example, two `ARRAY<STRING>` parent rows with individually
representable child data can exceed the child builder together and fail
conversion, although emitting one parent row per batch would succeed. Please
account for descendant byte/element limits per top-level row, return a clear
error when one parent row cannot fit, and add nested nullable collection tests.
##########
be/src/service/arrow_flight/arrow_flight_batch_reader.cpp:
##########
@@ -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,
Review Comment:
[P1] Split before the remote protobuf boundary
The remote reader reaches this helper only after `_fetch_data()` receives
and deserializes one complete `PFetchArrowDataResult.block`. On the producing
BE, `GetArrowResultBatchCtx::on_data()` rejects any response whose
`ByteSizeLong()` exceeds `INT32_MAX` and clears the block, so a proxy-routed
high-entropy result at the size this PR targets fails before this line even
though the local reader can split it. Please chunk on the producing BE before
protobuf serialization (or use a streaming transport) and add a proxy-routed
boundary test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
