This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new b3403658e89 [fix](arrow) Reject invalid UTF-8 and out-of-range
timestamps in Flight results (#69771)
b3403658e89 is described below
commit b3403658e89a7622f9e162619bc332a714110d94
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Thu Oct 29 07:06:37 2026 +0000
[fix](arrow) Reject invalid UTF-8 and out-of-range timestamps in Flight
results (#69771)
### What problem does this PR solve?
Issue Number: None
Related PR: #68596, #68604
Problem Summary:
#### In short
This ports branch-4.x's #68596 and #68604 to master. An Arrow Flight
result can carry values that Arrow clients cannot decode: a `STRING`
value that is not valid UTF-8 (for example `unhex('84')`), and a
`DATETIME` outside 0001-01-01..9999-12-31 (Doris's own range starts at
`0000-01-01 00:00:00`). The Arrow writers accept both, so the BE
publishes the batch and the failure surfaces later in the client:
PyArrow rejects the string, and raises `OverflowError` when it converts
the timestamp. This PR checks every batch in the Flight convertor before
it is handed out and fails the query with a server-side error that names
the column, the nested path and the row. It also makes a failed
`_query_plan` scan of the Spark and Flink connectors report its error:
the BE used to end such a scan as a complete result, so a rejected batch
would have become silently missing rows.
#### Background
- `ArrowFlightArrowBlockConvertor`
(`be/src/format/arrow/arrow_block_convertor.h`) builds the Arrow batches
of an Arrow Flight result: `ArrowFlightBatchReader` uses it for Flight
SQL and ADBC clients, which also covers a remote Doris catalog with
`use_arrow_flight = true`, and the memory scratch sink uses it for the
`_query_plan` scan API of the Spark and Flink Doris connectors.
- The STRING serde appends the bytes of each value to an Arrow `utf8`
(or `large_utf8`) builder without looking at them. Arrow requires these
values to be valid UTF-8, but only `Array::ValidateFull()` checks it;
building and publishing the array does not.
- The memory scratch sink queues the batches that `get_next` of the BE
Thrift scanner hands to those connectors, and ends the queue with a
`nullptr` sentinel. Its `close()` put the sentinel whatever the
execution status, and nothing has recorded a status in the queue since
the non-pipeline engine was removed (#35803), so a fragment that failed
(a scan error, a cancellation) ended the scan as if it were complete.
- `DATETIME(p)` is published as `timestamp[s]`, `timestamp[ms]` or
`timestamp[us]` by its scale. Arrow accepts any `int64` there, so a
year-0 value such as `0000-12-31 00:00:00` (`-62135683200000000` us) is
published as it is, while Python's `datetime`, which PyArrow converts
to, only covers years 1..9999.
#### The problem, and what it cost
| Result value | Before this PR |
|---|---|
| `unhex('84')`, `unhex('C0AF')`, a surrogate or a code point above
U+10FFFF, also inside an ARRAY / STRUCT / MAP key or value | Published
as is; the client fails decoding the batch, after earlier batches were
already delivered |
| Two rows `unhex('C2')` and `unhex('A2')` | Published; each row is
invalid UTF-8 on its own, though the concatenated buffer is valid |
| `CAST('0000-12-31 00:00:00' AS DATETIME(6))`, or a zero date read from
a table or an INT96 Parquet file | Published as `-62135683200000000`;
PyArrow raises `OverflowError` converting it |
| A `_query_plan` scan (Spark / Flink connector) whose fragment fails |
`get_next` reports the end of the scan after the rows fetched so far;
with the validation below, a table with one invalid UTF-8 row would read
zero rows and no error |
The query looks successful on the server, so neither the error nor the
offending column shows up in Doris; the Flight client gets a failure
that does not say which value caused it.
#### How this PR fixes it
- `ArrowBlockConvertor::convert_to_arrow` becomes virtual, and
`ArrowFlightArrowBlockConvertor` overrides it: after the base
conversion, it runs `ValidateFull()` on every column and returns
`InvalidArgument("Invalid Arrow Flight result in column <n> ('<name>'):
...")` on failure. `ValidateFull()` checks each value, nested children
included, so the row boundaries of the `C2` / `A2` case are respected.
Binary columns keep their bytes; NULL values are skipped through the
validity bitmap.
- A new `FlightTimestampValidator` walks each column recursively (list
values, map keys and values, struct fields), reads the actual timestamp
unit, and checks each value against 0001-01-01..9999-12-31, also against
the local calendar when the Arrow type has a time zone. Values under a
NULL list / map / struct parent are not observable and are not checked.
A value outside the range returns `InvalidArgument("Arrow Flight
timestamp in column '<path>' at row <n> is outside the supported
0001-9999 range (type: ...)")`, with the row counted in the block.
- For a timestamp with a time zone, the local bounds are computed with
the exact offsets (cctz, Python's zoneinfo) and with the minute-rounded
offsets of pytz, which PyArrow uses for zone names when it is installed
(`(offset + 30) // 60 * 60`), and the stricter bound wins. pytz turns
America/Chicago's local mean time -05:50:36 into -05:51, so 0001-01-01
05:50:36Z, local 00:00:00 for cctz, would be 0000-12-31 23:59:36 in
PyArrow; the lower bound there is 05:51:00Z. In such zones up to 30
seconds at the start of year 1 are rejected.
- `MemoryScratchSinkLocalState::close` records a failed execution status
in the result queue before it puts the sentinel, and
`ResultQueueMgr::fetch_result` checks the queue status again when it
takes the sentinel, because a fetch already waiting for data passed the
first check before the failure. `get_next` then returns the error to the
connector.
- Of the Arrow convertors, only the Flight convertor changes. The
ordinary Doris Arrow convertor (Python UDFs) and the table-format
convertors (Parquet, Iceberg, Paimon writers) keep their current
behavior; `DorisArrowBlockConvertor` still publishes the same values in
the new tests.
- The code is branch-4.2's final form of the two PRs; master's Flight
convertor already overrides `init()` and `write_column()`, and the new
override sits next to them.
#### Results
| Query over Arrow Flight | Before | After |
|---|---|---|
| `SELECT unhex(encoded) AS payload` with an invalid UTF-8 row | Batch
published, client fails decoding | Query fails: `Invalid Arrow Flight
result in column 1 ('payload'): Invalid: Invalid UTF8 sequence at string
index 0` |
| `SELECT array(unhex(encoded)) AS nested` ... | Batch published | `...
('nested'): Invalid: List child array invalid: Invalid: Invalid UTF8
sequence at string index 0` |
| `SELECT CAST(value AS DATETIME(6)) AS event_time` over `0000-12-31
00:00:00` | Batch published, client fails converting | Query fails:
`Arrow Flight timestamp in column 'event_time' at row 1 is outside the
supported 0001-9999 range (type: timestamp[us])`; nested values report
paths such as `nested[]`, `nested.key`, `nested.child` |
| Remote Doris catalog over Flight reading a `0000-01-01 00:00:00`
DATETIME | Value returned | `Arrow Flight timestamp in column
'datetime_0' at row 1 is outside the supported 0001-9999 range (type:
timestamp[s])` |
| `_query_plan` scan through `open_scanner` / `get_next` over a table
with an invalid UTF-8 row | 0 rows, then the end of the scan |
`get_next` returns `[INVALID_ARGUMENT]Invalid Arrow Flight result in
column 2 (''): Invalid: Invalid UTF8 sequence at string index 1` |
| The same scan over a `0000-12-31 00:00:00` DATETIME | 0 rows, then the
end of the scan | `Arrow Flight timestamp in column '' at row 2 is
outside the supported 0001-9999 range (type: timestamp[us])` |
| Valid Unicode text, NULLs, binary bytes, `0001-01-01`, `9999-12-31
23:59:59.999999` | Returned | Returned unchanged |
Behavior change: a Flight query (Flight SQL, ADBC, a remote Doris
catalog with `use_arrow_flight = true`, or a `_query_plan` read of the
Spark / Flink connector) whose result contains a `STRING` value that is
not valid UTF-8, or a `DATETIME` outside years 0001..9999, now fails
with the error above instead of returning data. A remote Doris catalog
that used to round-trip the zero date now reports this error; the three
remote Doris suites that inserted `0000-01-01 00:00:00` as their minimum
value now use `0001-01-01 00:00:00`, and
`test_remote_doris_all_types_select` asserts the zero-date error
separately. A `_query_plan` read whose fragment fails for any reason,
also a scan error or a cancellation, now gets the error from `get_next`
instead of ending as a complete result.
Tests:
- BE UT `ArrowBlockConvertorTest`: six new cases for invalid UTF-8 (four
malformed encodings, per-row boundaries across a slice and the next
batch, nested array / struct / map key / map value, large strings, bytes
under a NULL), and valid text, NULLs and binary.
- BE UT `ArrowFlightTimestampTest` (new file): zero date, year 0 and
year 10000 at scales 0 / 3 / 6, calendar boundaries and pre-epoch
fractions, slices and later batches, nested values and map keys, UTC and
zoned bounds, the America/Chicago lower bound with pytz's rounded
offset, naive bounds that do not depend on the session time zone, and
values under NULL parents; the ordinary Doris convertor still publishes
the same values.
- BE UT `ResultQueueMgrTest`: a fetch already waiting for data reports a
failure recorded before the sentinel; an OK status keeps a normal end.
- Regression `arrow_flight_sql_p0/test_query_plan_scan_validation` reads
`_query_plan` scans the way the connectors do (FE `_query_plan`, then BE
`open_scanner` / `get_next` over Thrift): a valid scan reads all rows,
and the invalid UTF-8 and year-0 scans get the errors above.
- Regression `arrow_flight_sql_p0/test_flight_utf8_validation`,
`test_flight_timestamp_range` (both red before this PR on master, green
with it), `test_flight_int96_timestamp_range` (reads the INT96 fixtures
of `group4` over HDFS; needs the Hive docker environment), and the three
remote Doris suites.
---
.../exec/operator/memory_scratch_sink_operator.cpp | 4 +
be/src/format/arrow/arrow_block_convertor.cpp | 178 +++++++++++++++
be/src/format/arrow/arrow_block_convertor.h | 9 +-
be/src/runtime/result_queue_mgr.cpp | 5 +-
.../format/arrow/arrow_block_convertor_test.cpp | 133 +++++++++++
.../format/arrow/arrow_flight_timestamp_test.cpp | 250 +++++++++++++++++++++
be/test/runtime/result_queue_mgr_test.cpp | 55 +++++
.../test_remote_doris_all_types_select.out | 4 +-
.../remote_doris/test_remote_doris_statistics.out | 2 +-
.../test_flight_int96_timestamp_range.groovy | 83 +++++++
.../test_flight_timestamp_range.groovy | 81 +++++++
.../test_flight_utf8_validation.groovy | 71 ++++++
.../test_query_plan_scan_validation.groovy | 120 ++++++++++
.../test_remote_doris_all_types_select.groovy | 15 +-
.../test_remote_doris_statistics.groovy | 3 +-
.../test_remote_doris_table_stats.groovy | 3 +-
16 files changed, 1005 insertions(+), 11 deletions(-)
diff --git a/be/src/exec/operator/memory_scratch_sink_operator.cpp
b/be/src/exec/operator/memory_scratch_sink_operator.cpp
index 1d795001719..9af197a1eb9 100644
--- a/be/src/exec/operator/memory_scratch_sink_operator.cpp
+++ b/be/src/exec/operator/memory_scratch_sink_operator.cpp
@@ -57,6 +57,10 @@ Status MemoryScratchSinkLocalState::close(RuntimeState*
state, Status exec_statu
SCOPED_TIMER(exec_time_counter());
SCOPED_TIMER(_close_timer);
if (_queue != nullptr) {
+ // The nullptr sentinel alone reads as a complete result. Record a
failed execution (a scan
+ // error, a rejected Arrow batch, a cancellation) first, so the
external scanner reports it
+ // instead of ending with the batches queued so far. update_status()
ignores OK.
+ _queue->update_status(exec_status);
_queue->blocking_put(nullptr);
}
RETURN_IF_ERROR(Base::close(state, exec_status));
diff --git a/be/src/format/arrow/arrow_block_convertor.cpp
b/be/src/format/arrow/arrow_block_convertor.cpp
index bfa4e08ad6e..f7267d683a2 100644
--- a/be/src/format/arrow/arrow_block_convertor.cpp
+++ b/be/src/format/arrow/arrow_block_convertor.cpp
@@ -17,6 +17,8 @@
#include "format/arrow/arrow_block_convertor.h"
+#include <arrow/array/array_nested.h>
+#include <arrow/array/array_primitive.h>
#include <arrow/array/builder_base.h>
#include <arrow/array/builder_binary.h>
#include <arrow/array/builder_decimal.h>
@@ -34,6 +36,7 @@
#include <cctz/time_zone.h>
#include <glog/logging.h>
+#include <algorithm>
#include <array>
#include <cstring>
#include <ctime>
@@ -62,6 +65,156 @@ namespace doris {
namespace {
+// PyArrow resolves an Arrow timezone name with pytz when pytz is installed,
and pytz rounds every
+// UTC offset to whole minutes with Python's floor division, (offset + 30) //
60 * 60. Example:
+// America/Chicago's local mean time, -05:50:36 (-21036 s), becomes -05:51
(-21060 s).
+int64_t pytz_rounded_utc_offset(int64_t offset_seconds) {
+ const int64_t shifted = offset_seconds + 30;
+ // C++ division truncates toward zero; -21006 / 60 is -350, the floor is
-351.
+ const int64_t minutes = shifted / 60 - (shifted % 60 < 0 ? 1 : 0);
+ return minutes * 60;
+}
+
+class FlightTimestampValidator {
+public:
+ Status init(std::shared_ptr<arrow::Array> array, std::string path,
+ const cctz::time_zone& timezone) {
+ _array = std::move(array);
+ _path = std::move(path);
+ switch (_array->type_id()) {
+ case arrow::Type::TIMESTAMP: {
+ const auto& type = static_cast<const
arrow::TimestampType&>(*_array->type());
+ int64_t units_per_second = 1;
+ switch (type.unit()) {
+ case arrow::TimeUnit::SECOND:
+ units_per_second = 1;
+ break;
+ case arrow::TimeUnit::MILLI:
+ units_per_second = 1000;
+ break;
+ case arrow::TimeUnit::MICRO:
+ units_per_second = 1000000;
+ break;
+ case arrow::TimeUnit::NANO:
+ units_per_second = 1000000000;
+ break;
+ }
+ // 0001-01-01 00:00:00 and 10000-01-01 00:00:00 as seconds since
the epoch.
+ constexpr int64_t civil_start_seconds = -62135596800LL;
+ constexpr int64_t civil_end_seconds = 253402300800LL;
+ int64_t min_seconds = civil_start_seconds;
+ int64_t end_seconds = civil_end_seconds;
+ if (!type.timezone().empty()) {
+ // The ordinary Arrow writer has already checked the
schema/timezone binding.
+ const auto local_start = cctz::convert(cctz::civil_second(1,
1, 1), timezone);
+ const auto local_end = cctz::convert(cctz::civil_second(10000,
1, 1), timezone);
+ // Python first constructs UTC and then applies the Arrow
timezone. Both calendar
+ // representations must fit, including offsets at the first
and last supported day,
+ // with the exact offsets (zoneinfo) and with pytz's rounded
ones. Example for
+ // America/Chicago: 0001-01-01 05:50:36Z is local 0001-01-01
00:00:00 with the
+ // exact -05:50:36, but 0000-12-31 23:59:36 with pytz's
-05:51, so the lower bound
+ // is 05:51:00Z.
+ const int64_t start_offset =
timezone.lookup(local_start).offset;
+ const int64_t end_offset = timezone.lookup(local_end).offset;
+ min_seconds = std::max<int64_t>(
+ {min_seconds, local_start.time_since_epoch().count(),
+ civil_start_seconds -
pytz_rounded_utc_offset(start_offset)});
+ end_seconds = std::min<int64_t>(
+ {end_seconds, local_end.time_since_epoch().count(),
+ civil_end_seconds -
pytz_rounded_utc_offset(end_offset)});
+ }
+ // Nanosecond bounds do not fit int64_t, even though every encoded
value does.
+ _min = static_cast<__int128>(min_seconds) * units_per_second;
+ _end = static_cast<__int128>(end_seconds) * units_per_second;
+ break;
+ }
+ case arrow::Type::LIST: {
+ const auto& list = static_cast<const arrow::ListArray&>(*_array);
+ RETURN_IF_ERROR(add_child(list.values(), _path + "[]", timezone));
+ break;
+ }
+ case arrow::Type::MAP: {
+ const auto& map = static_cast<const arrow::MapArray&>(*_array);
+ RETURN_IF_ERROR(add_child(map.keys(), _path + ".key", timezone));
+ RETURN_IF_ERROR(add_child(map.items(), _path + ".value",
timezone));
+ break;
+ }
+ case arrow::Type::STRUCT: {
+ const auto& structure = static_cast<const
arrow::StructArray&>(*_array);
+ for (int i = 0; i < structure.num_fields(); ++i) {
+ RETURN_IF_ERROR(add_child(structure.field(i),
+ _path + "." +
_array->type()->field(i)->name(),
+ timezone));
+ }
+ break;
+ }
+ default:
+ break;
+ }
+ return Status::OK();
+ }
+
+ Status validate(int64_t start, int64_t end, size_t block_start, int64_t
parent_row = -1) const {
+ if (!has_timestamps()) {
+ return Status::OK();
+ }
+ for (int64_t row = start; row < end; ++row) {
+ // Children of a NULL struct/list/map are not observable, even if
their physical
+ // buffers contain default zero dates. Never validate those masked
values.
+ if (_array->IsNull(row)) {
+ continue;
+ }
+ const int64_t output_row = parent_row < 0 ? row : parent_row;
+ if (_array->type_id() == arrow::Type::TIMESTAMP) {
+ const int64_t value = static_cast<const
arrow::TimestampArray&>(*_array).Value(row);
+ if (value < _min || value >= _end) {
+ return Status::InvalidArgument(
+ "Arrow Flight timestamp in column '{}' at row {}
is outside the "
+ "supported 0001-9999 range (type: {})",
+ _path, block_start + output_row + 1,
_array->type()->ToString());
+ }
+ continue;
+ }
+ int64_t child_start = row;
+ int64_t child_end = row + 1;
+ if (_array->type_id() == arrow::Type::LIST) {
+ const auto& list = static_cast<const
arrow::ListArray&>(*_array);
+ child_start = list.value_offset(row);
+ child_end = child_start + list.value_length(row);
+ } else if (_array->type_id() == arrow::Type::MAP) {
+ const auto& map = static_cast<const arrow::MapArray&>(*_array);
+ child_start = map.value_offset(row);
+ child_end = child_start + map.value_length(row);
+ }
+ for (const auto& child : _children) {
+ RETURN_IF_ERROR(child.validate(child_start, child_end,
block_start, output_row));
+ }
+ }
+ return Status::OK();
+ }
+
+private:
+ bool has_timestamps() const {
+ return _array->type_id() == arrow::Type::TIMESTAMP ||
!_children.empty();
+ }
+
+ Status add_child(std::shared_ptr<arrow::Array> array, std::string path,
+ const cctz::time_zone& timezone) {
+ FlightTimestampValidator child;
+ RETURN_IF_ERROR(child.init(std::move(array), std::move(path),
timezone));
+ if (child.has_timestamps()) {
+ _children.push_back(std::move(child));
+ }
+ return Status::OK();
+ }
+
+ std::shared_ptr<arrow::Array> _array;
+ std::string _path;
+ std::vector<FlightTimestampValidator> _children;
+ __int128 _min = 0;
+ __int128 _end = 0;
+};
+
std::shared_ptr<arrow::DataType> arrow_storage_type(const
std::shared_ptr<arrow::DataType>& type) {
switch (type->id()) {
case arrow::Type::EXTENSION:
@@ -324,6 +477,31 @@ Status ArrowFlightArrowBlockConvertor::write_column(const
std::shared_ptr<const
array_builder, start, end,
ctz);
}
+Status ArrowFlightArrowBlockConvertor::convert_to_arrow(const Block& block,
arrow::MemoryPool* pool,
+
std::shared_ptr<arrow::RecordBatch>* result,
+ size_t start_row,
size_t end_row) const {
+ std::shared_ptr<arrow::RecordBatch> batch;
+ RETURN_IF_ERROR(ArrowBlockConvertor::convert_to_arrow(block, pool, &batch,
start_row, end_row));
+ // String builders accept arbitrary bytes, but Flight UTF-8 values must be
valid per row,
+ // including nested children. Validate before publishing the batch to the
reader.
+ for (int i = 0; i < batch->num_columns(); ++i) {
+ auto status = batch->column(i)->ValidateFull();
+ if (!status.ok()) {
+ return Status::InvalidArgument("Invalid Arrow Flight result in
column {} ('{}'): {}",
+ i + 1,
batch->schema()->field(i)->name(),
+ status.ToString());
+ }
+ // Arrow accepts the full int64 timestamp domain; ValidateFull cannot
enforce the calendar
+ // range required by Flight clients. Check the encoded values before
publishing the batch.
+ FlightTimestampValidator validator;
+ RETURN_IF_ERROR(
+ validator.init(batch->column(i),
batch->schema()->field(i)->name(), _timezone));
+ RETURN_IF_ERROR(validator.validate(0, batch->num_rows(), start_row));
+ }
+ *result = std::move(batch);
+ return Status::OK();
+}
+
Status DorisArrowBlockConvertor::init() {
if (_arrow_schema == nullptr) {
// cctz names fixed offsets as "Fixed/UTC+HH:MM:SS", which is not the
Arrow
diff --git a/be/src/format/arrow/arrow_block_convertor.h
b/be/src/format/arrow/arrow_block_convertor.h
index 1f35ee30652..c18c28df72a 100644
--- a/be/src/format/arrow/arrow_block_convertor.h
+++ b/be/src/format/arrow/arrow_block_convertor.h
@@ -58,9 +58,9 @@ public:
virtual Status init();
const std::shared_ptr<arrow::Schema>& arrow_schema() const { return
_arrow_schema; }
- Status convert_to_arrow(const Block& block, arrow::MemoryPool* pool,
- std::shared_ptr<arrow::RecordBatch>* result,
size_t start_row = 0,
- size_t end_row = 0) const;
+ virtual Status convert_to_arrow(const Block& block, arrow::MemoryPool*
pool,
+ std::shared_ptr<arrow::RecordBatch>*
result,
+ size_t start_row = 0, size_t end_row = 0)
const;
virtual Status convert_from_arrow(const
std::shared_ptr<arrow::RecordBatch>& batch,
const DataTypes& types, Block* block)
const;
@@ -113,6 +113,9 @@ class ArrowFlightArrowBlockConvertor final : public
DorisArrowBlockConvertor {
public:
using DorisArrowBlockConvertor::DorisArrowBlockConvertor;
Status init() override;
+ Status convert_to_arrow(const Block& block, arrow::MemoryPool* pool,
+ std::shared_ptr<arrow::RecordBatch>* result,
size_t start_row = 0,
+ size_t end_row = 0) const override;
protected:
Status write_column(const std::shared_ptr<const IDataType>& type, const
DataTypeSerDe& serde,
diff --git a/be/src/runtime/result_queue_mgr.cpp
b/be/src/runtime/result_queue_mgr.cpp
index 13850cd0954..51208f324be 100644
--- a/be/src/runtime/result_queue_mgr.cpp
+++ b/be/src/runtime/result_queue_mgr.cpp
@@ -63,9 +63,12 @@ Status ResultQueueMgr::fetch_result(const TUniqueId&
fragment_instance_id,
if (success) {
// sentinel nullptr indicates scan end
if (*result == nullptr) {
- *eos = true;
// put sentinel for consistency, avoid repeated invoking fetch
result when have no rowbatch
queue->blocking_put(nullptr);
+ // The sink records a failed execution before it puts the
sentinel. Check again here:
+ // a fetch that was already waiting for data passed the check
above before the failure.
+ RETURN_IF_ERROR(queue->status());
+ *eos = true;
} else {
*eos = false;
}
diff --git a/be/test/format/arrow/arrow_block_convertor_test.cpp
b/be/test/format/arrow/arrow_block_convertor_test.cpp
index 16cd9c3146c..b6fa589570b 100644
--- a/be/test/format/arrow/arrow_block_convertor_test.cpp
+++ b/be/test/format/arrow/arrow_block_convertor_test.cpp
@@ -23,6 +23,7 @@
#include <arrow/ipc/api.h>
#include <gtest/gtest.h>
+#include "core/column/column_nullable.h"
#include "core/column/column_vector.h"
#include "core/data_type/data_type_array.h"
#include "core/data_type/data_type_factory.hpp"
@@ -307,4 +308,136 @@ TEST_F(ArrowBlockConvertorTest,
TableConvertersRejectMismatchedNestedSchemas) {
}
}
+TEST_F(ArrowBlockConvertorTest, FlightRejectsInvalidUtf8BeforeReturningBatch) {
+ auto type = DataTypeFactory::instance().create_data_type(TYPE_STRING,
false);
+ for (const std::string& value :
+ {std::string("\x84"), std::string("\xc0\xaf"),
std::string("\xed\xa0\x80"),
+ std::string("\xf4\x90\x80\x80")}) {
+ auto column = type->create_column();
+ column->insert(Field::create_field<TYPE_STRING>(value));
+ Block block {{std::move(column), type, "payload"}};
+ ArrowFlightArrowBlockConvertor flight(block, "UTC",
cctz::utc_time_zone());
+ ASSERT_TRUE(flight.init().ok());
+ const ArrowBlockConvertor& converter = flight;
+ std::shared_ptr<arrow::RecordBatch> batch;
+ const auto status = converter.convert_to_arrow(block,
arrow::default_memory_pool(), &batch);
+ EXPECT_EQ(ErrorCode::INVALID_ARGUMENT, status.code()) << status;
+ EXPECT_NE(std::string::npos, status.to_string().find("payload"));
+ EXPECT_NE(std::string::npos, status.to_string().find("UTF8"));
+ EXPECT_EQ(nullptr, batch);
+
+ DorisArrowBlockConvertor ordinary(block, "UTC", cctz::utc_time_zone());
+ ASSERT_TRUE(ordinary.init().ok());
+ EXPECT_TRUE(ordinary.convert_to_arrow(block,
arrow::default_memory_pool(), &batch).ok());
+ }
+}
+
+TEST_F(ArrowBlockConvertorTest, FlightChecksEachStringAndEveryBatch) {
+ auto type = DataTypeFactory::instance().create_data_type(TYPE_STRING,
false);
+ auto column = type->create_column();
+ for (const std::string& value :
+ {std::string("valid"), std::string("\xc2"), std::string("\xa2")}) {
+ column->insert(Field::create_field<TYPE_STRING>(value));
+ }
+ Block block {{std::move(column), type, "payload"}};
+ ArrowFlightArrowBlockConvertor converter(block, "UTC",
cctz::utc_time_zone());
+ ASSERT_TRUE(converter.init().ok());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ ASSERT_TRUE(converter.convert_to_arrow(block,
arrow::default_memory_pool(), &batch, 0, 1).ok());
+ ASSERT_EQ(1, batch->num_rows());
+ batch.reset();
+ // Adjacent invalid strings form valid UTF-8 when concatenated; row
boundaries matter.
+ EXPECT_FALSE(
+ converter.convert_to_arrow(block, arrow::default_memory_pool(),
&batch, 1, 3).ok());
+ EXPECT_EQ(nullptr, batch);
+}
+
+TEST_F(ArrowBlockConvertorTest, FlightPreservesValidTextNullsAndBinary) {
+ auto text =
make_nullable(DataTypeFactory::instance().create_data_type(TYPE_STRING, false));
+ auto strings = text->create_column();
+
strings->insert(Field::create_field<TYPE_STRING>(std::string("\xe4\xb8\xad\xf0\x9f\x98\x80")));
+ strings->insert(Field::create_field<TYPE_STRING>(std::string()));
+ strings->insert(Field::create_field<TYPE_STRING>(std::string("a\0b", 3)));
+ strings->insert_default();
+ auto binary = DataTypeFactory::instance().create_data_type(TYPE_VARBINARY,
false);
+ auto bytes = binary->create_column();
+ for (int i = 0; i < 4; ++i) {
+ bytes->insert_data("\x84\0\xff", 3);
+ }
+ Block block {{std::move(strings), text, "text"}, {std::move(bytes),
binary, "binary"}};
+ ArrowFlightArrowBlockConvertor converter(block, "UTC",
cctz::utc_time_zone());
+ ASSERT_TRUE(converter.init().ok());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ ASSERT_TRUE(converter.convert_to_arrow(block,
arrow::default_memory_pool(), &batch).ok());
+ ASSERT_TRUE(batch->ValidateFull().ok());
+ const auto& values = static_cast<const
arrow::StringArray&>(*batch->column(0));
+ EXPECT_EQ("\xe4\xb8\xad\xf0\x9f\x98\x80", values.GetString(0));
+ EXPECT_EQ("", values.GetString(1));
+ EXPECT_EQ(std::string("a\0b", 3), values.GetString(2));
+ EXPECT_TRUE(values.IsNull(3));
+ ASSERT_EQ(arrow::Type::BINARY, batch->column(1)->type_id());
+ const auto& binary_values = static_cast<const
arrow::BinaryArray&>(*batch->column(1));
+ EXPECT_EQ(std::string("\x84\0\xff", 3), binary_values.GetString(0));
+}
+
+TEST_F(ArrowBlockConvertorTest, FlightRejectsNestedInvalidUtf8) {
+ auto text =
make_nullable(DataTypeFactory::instance().create_data_type(TYPE_STRING, false));
+ const auto invalid = Field::create_field<TYPE_STRING>(std::string("\x84"));
+ const auto valid = Field::create_field<TYPE_STRING>(std::string("key"));
+ DataTypes types {std::make_shared<DataTypeArray>(text),
+ std::make_shared<DataTypeStruct>(DataTypes {text},
Strings {"child"}),
+ std::make_shared<DataTypeMap>(text, text),
+ std::make_shared<DataTypeMap>(text, text)};
+ FieldVector fields {
+ Field::create_field<TYPE_ARRAY>(Array {invalid}),
+ Field::create_field<TYPE_STRUCT>(Struct {invalid}),
+ Field::create_field<TYPE_MAP>(Map
{Field::create_field<TYPE_ARRAY>(Array {valid}),
+
Field::create_field<TYPE_ARRAY>(Array {invalid})}),
+ Field::create_field<TYPE_MAP>(Map
{Field::create_field<TYPE_ARRAY>(Array {invalid}),
+
Field::create_field<TYPE_ARRAY>(Array {valid})})};
+ for (size_t i = 0; i < types.size(); ++i) {
+ SCOPED_TRACE(types[i]->get_name());
+ auto column = types[i]->create_column();
+ column->insert(fields[i]);
+ Block block {{std::move(column), types[i], "nested"}};
+ ArrowFlightArrowBlockConvertor converter(block, "UTC",
cctz::utc_time_zone());
+ ASSERT_TRUE(converter.init().ok());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ const auto status = converter.convert_to_arrow(block,
arrow::default_memory_pool(), &batch);
+ EXPECT_EQ(ErrorCode::INVALID_ARGUMENT, status.code()) << status;
+ EXPECT_NE(std::string::npos, status.to_string().find("nested"));
+ EXPECT_EQ(nullptr, batch);
+ }
+}
+
+TEST_F(ArrowBlockConvertorTest, FlightRejectsInvalidLargeString) {
+ auto type = DataTypeFactory::instance().create_data_type(TYPE_STRING,
false);
+ auto column = type->create_column();
+ column->insert(Field::create_field<TYPE_STRING>(std::string("\x84")));
+ Block block {{std::move(column), type, "large_text"}};
+ auto schema = arrow::schema({arrow::field("large_text",
arrow::large_utf8(), false)});
+ ArrowFlightArrowBlockConvertor converter(schema, cctz::utc_time_zone());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ const auto status = converter.convert_to_arrow(block,
arrow::default_memory_pool(), &batch);
+ EXPECT_EQ(ErrorCode::INVALID_ARGUMENT, status.code()) << status;
+ EXPECT_NE(std::string::npos, status.to_string().find("large_text"));
+ EXPECT_EQ(nullptr, batch);
+}
+
+TEST_F(ArrowBlockConvertorTest, FlightIgnoresBytesMaskedByNull) {
+ auto text = DataTypeFactory::instance().create_data_type(TYPE_STRING,
false);
+ auto values = text->create_column();
+ values->insert_data("\x84", 1);
+ auto nulls = ColumnUInt8::create();
+ nulls->insert_value(1);
+ Block block {{ColumnNullable::create(std::move(values), std::move(nulls)),
make_nullable(text),
+ "nullable_text"}};
+ ArrowFlightArrowBlockConvertor converter(block, "UTC",
cctz::utc_time_zone());
+ ASSERT_TRUE(converter.init().ok());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ ASSERT_TRUE(converter.convert_to_arrow(block,
arrow::default_memory_pool(), &batch).ok());
+ ASSERT_TRUE(batch->ValidateFull().ok());
+ EXPECT_TRUE(batch->column(0)->IsNull(0));
+}
+
} // namespace doris
diff --git a/be/test/format/arrow/arrow_flight_timestamp_test.cpp
b/be/test/format/arrow/arrow_flight_timestamp_test.cpp
new file mode 100644
index 00000000000..8aa88082658
--- /dev/null
+++ b/be/test/format/arrow/arrow_flight_timestamp_test.cpp
@@ -0,0 +1,250 @@
+// 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.
+
+#include <arrow/api.h>
+#include <gtest/gtest.h>
+
+#include "core/column/column_nullable.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_date_or_datetime_v2.h"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_struct.h"
+#include "format/arrow/arrow_block_convertor.h"
+#include "format/arrow/arrow_row_batch.h"
+#include "util/timezone_utils.h"
+
+namespace doris {
+namespace {
+using DateTime = DateV2Value<DateTimeV2ValueType>;
+
+DateTime make_datetime(uint16_t year, uint8_t month, uint8_t day, uint8_t
hour, uint8_t minute,
+ uint16_t second, uint32_t microsecond) {
+ DateTime value;
+ // Invalid calendar values must reach the converter to exercise its range
validation.
+ value.unchecked_set_time(year, month, day, hour, minute, second,
microsecond);
+ return value;
+}
+
+Block timestamp_block(int scale, const std::vector<DateTime>& values) {
+ auto column = ColumnDateTimeV2::create();
+ for (const auto& value : values) {
+ column->insert_value(value);
+ }
+ return Block {{std::move(column),
std::make_shared<DataTypeDateTimeV2>(scale), "event_time"}};
+}
+
+class ArrowFlightTimestampTest : public testing::Test {
+protected:
+ static void SetUpTestSuite() { TimezoneUtils::load_timezones_to_cache(); }
+};
+
+TEST_F(ArrowFlightTimestampTest,
RejectsOutOfRangeInEveryUnitWithoutPublishingBatch) {
+ for (int scale : {0, 3, 6}) {
+ for (const auto& value : {DateTime {}, make_datetime(0, 12, 31, 0, 0,
0, 0),
+ make_datetime(10000, 1, 1, 0, 0, 0, 0)}) {
+ SCOPED_TRACE(scale);
+ auto block = timestamp_block(scale, {value});
+ ArrowFlightArrowBlockConvertor flight(block, "UTC",
cctz::utc_time_zone());
+ ASSERT_TRUE(flight.init().ok());
+ const ArrowBlockConvertor& converter = flight;
+ std::shared_ptr<arrow::RecordBatch> batch;
+ const auto status =
+ converter.convert_to_arrow(block,
arrow::default_memory_pool(), &batch);
+ EXPECT_EQ(ErrorCode::INVALID_ARGUMENT, status.code()) << status;
+ EXPECT_NE(std::string::npos,
status.to_string().find("event_time"));
+ EXPECT_NE(std::string::npos, status.to_string().find("row 1"));
+ EXPECT_NE(std::string::npos, status.to_string().find("0001-9999"));
+ EXPECT_EQ(nullptr, batch);
+
+ // Other Arrow consumers retain their existing date semantics.
+ DorisArrowBlockConvertor ordinary(flight.arrow_schema(),
cctz::utc_time_zone());
+ ASSERT_TRUE(ordinary.init().ok());
+ ASSERT_TRUE(
+ ordinary.convert_to_arrow(block,
arrow::default_memory_pool(), &batch).ok());
+ EXPECT_TRUE(batch->ValidateFull().ok());
+ }
+ }
+}
+
+TEST_F(ArrowFlightTimestampTest,
PreservesCalendarBoundariesAndPreEpochFractions) {
+ for (int scale : {0, 3, 6}) {
+ const int64_t factor = scale == 0 ? 1 : scale == 3 ? 1000 : 1000000;
+ const uint32_t fraction = scale == 0 ? 0 : scale == 3 ? 999000 :
999999;
+ auto block = timestamp_block(scale, {make_datetime(1, 1, 1, 0, 0, 0,
0),
+ make_datetime(9999, 12, 31, 23,
59, 59, fraction),
+ make_datetime(1969, 12, 31, 23,
59, 59, fraction)});
+ ArrowFlightArrowBlockConvertor converter(block, "UTC",
cctz::utc_time_zone());
+ ASSERT_TRUE(converter.init().ok());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ ASSERT_TRUE(converter.convert_to_arrow(block,
arrow::default_memory_pool(), &batch).ok());
+ const auto& values = static_cast<const
arrow::TimestampArray&>(*batch->column(0));
+ EXPECT_EQ(-62135596800LL * factor, values.Value(0));
+ EXPECT_EQ(253402300800LL * factor - 1, values.Value(1));
+ EXPECT_EQ(-1, values.Value(2));
+ }
+}
+
+TEST_F(ArrowFlightTimestampTest, ChecksSlicesAndSubsequentBatches) {
+ auto block = timestamp_block(6, {make_datetime(2024, 1, 1, 0, 0, 0, 0),
DateTime {}});
+ ArrowFlightArrowBlockConvertor converter(block, "UTC",
cctz::utc_time_zone());
+ ASSERT_TRUE(converter.init().ok());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ ASSERT_TRUE(converter.convert_to_arrow(block,
arrow::default_memory_pool(), &batch, 0, 1).ok());
+ auto previous = batch;
+ const auto status =
+ converter.convert_to_arrow(block, arrow::default_memory_pool(),
&batch, 1, 2);
+ EXPECT_EQ(ErrorCode::INVALID_ARGUMENT, status.code());
+ EXPECT_NE(std::string::npos, status.to_string().find("row 2"));
+ EXPECT_EQ(previous, batch);
+}
+
+TEST_F(ArrowFlightTimestampTest, RejectsNestedTimestampValuesAndMapKeys) {
+ auto datetime = make_nullable(std::make_shared<DataTypeDateTimeV2>(6));
+ const auto invalid = Field::create_field<TYPE_DATETIMEV2>(DateTime {});
+ const auto valid =
Field::create_field<TYPE_DATETIMEV2>(make_datetime(2024, 1, 1, 0, 0, 0, 0));
+ DataTypes types {std::make_shared<DataTypeArray>(datetime),
+ std::make_shared<DataTypeStruct>(DataTypes {datetime},
Strings {"child"}),
+ std::make_shared<DataTypeMap>(datetime, datetime),
+ std::make_shared<DataTypeMap>(datetime, datetime),
+
std::make_shared<DataTypeArray>(std::make_shared<DataTypeStruct>(
+ DataTypes {datetime}, Strings {"child"}))};
+ FieldVector fields {
+ Field::create_field<TYPE_ARRAY>(Array {valid, invalid}),
+ Field::create_field<TYPE_STRUCT>(Struct {invalid}),
+ Field::create_field<TYPE_MAP>(Map
{Field::create_field<TYPE_ARRAY>(Array {valid}),
+
Field::create_field<TYPE_ARRAY>(Array {invalid})}),
+ Field::create_field<TYPE_MAP>(Map
{Field::create_field<TYPE_ARRAY>(Array {invalid}),
+
Field::create_field<TYPE_ARRAY>(Array {valid})}),
+ Field::create_field<TYPE_ARRAY>(
+ Array {Field::create_field<TYPE_STRUCT>(Struct
{invalid})})};
+ for (size_t i = 0; i < types.size(); ++i) {
+ SCOPED_TRACE(types[i]->get_name());
+ auto column = types[i]->create_column();
+ column->insert_default();
+ column->insert(fields[i]);
+ Block block {{std::move(column), types[i], "nested"}};
+ ArrowFlightArrowBlockConvertor converter(block, "UTC",
cctz::utc_time_zone());
+ ASSERT_TRUE(converter.init().ok());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ const auto status =
+ converter.convert_to_arrow(block,
arrow::default_memory_pool(), &batch, 1, 2);
+ EXPECT_EQ(ErrorCode::INVALID_ARGUMENT, status.code()) << status;
+ EXPECT_NE(std::string::npos, status.to_string().find("nested"));
+ EXPECT_NE(std::string::npos, status.to_string().find("row 2"));
+ EXPECT_EQ(nullptr, batch);
+ }
+}
+
+TEST_F(ArrowFlightTimestampTest, ChecksBothUtcAndZonedCalendarBounds) {
+ const std::vector<std::pair<std::string, DateTime>> cases {
+ {"+08:00", make_datetime(1, 1, 1, 0, 0, 0, 0)},
+ {"-08:00", make_datetime(9999, 12, 31, 23, 0, 0, 0)},
+ {"-08:00", make_datetime(0, 12, 31, 23, 0, 0, 0)},
+ {"+08:00", make_datetime(10000, 1, 1, 1, 0, 0, 0)}};
+ for (const auto& [zone, value] : cases) {
+ SCOPED_TRACE(zone);
+ cctz::time_zone timezone;
+ ASSERT_TRUE(TimezoneUtils::find_cctz_time_zone(zone, timezone));
+ auto block = timestamp_block(6, {value});
+ // Flight-generated DATETIMEV2 schemas are naive; keep an explicit
zoned schema
+ // here so this test still checks both UTC and local calendar bounds.
+ std::shared_ptr<arrow::Schema> schema;
+ ASSERT_TRUE(DorisArrowSchemaConvertor(block,
zone).get_arrow_schema(&schema).ok());
+ ArrowFlightArrowBlockConvertor converter(schema, timezone);
+ ASSERT_TRUE(converter.init().ok());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ const auto status = converter.convert_to_arrow(block,
arrow::default_memory_pool(), &batch);
+ EXPECT_EQ(ErrorCode::INVALID_ARGUMENT, status.code()) << status;
+ EXPECT_EQ(nullptr, batch);
+
+ auto valid = timestamp_block(
+ 6, {make_datetime(1, 1, 2, 0, 0, 0, 0), make_datetime(9999,
12, 30, 23, 0, 0, 0)});
+ ASSERT_TRUE(converter.convert_to_arrow(valid,
arrow::default_memory_pool(), &batch).ok());
+ }
+}
+
+// PyArrow converts with pytz when it is installed, and pytz rounds
America/Chicago's local mean
+// time, -05:50:36, to -05:51. 0001-01-01 05:50:36Z, the local start with the
exact offset, is
+// 0000-12-31 23:59:36 in pytz and fails to convert; the first accepted
instant is 05:51:00Z.
+TEST_F(ArrowFlightTimestampTest, ZonedLowerBoundHoldsWithPytzRoundedOffsets) {
+ const std::string zone = "America/Chicago";
+ cctz::time_zone timezone;
+ ASSERT_TRUE(TimezoneUtils::find_cctz_time_zone(zone, timezone));
+ // Local 0001-01-01 00:00:00 with the exact offset is 05:50:36Z.
+ auto block = timestamp_block(6, {make_datetime(1, 1, 1, 0, 0, 0, 0)});
+ std::shared_ptr<arrow::Schema> schema;
+ ASSERT_TRUE(DorisArrowSchemaConvertor(block,
zone).get_arrow_schema(&schema).ok());
+ ArrowFlightArrowBlockConvertor converter(schema, timezone);
+ ASSERT_TRUE(converter.init().ok());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ const auto status = converter.convert_to_arrow(block,
arrow::default_memory_pool(), &batch);
+ EXPECT_EQ(ErrorCode::INVALID_ARGUMENT, status.code()) << status;
+ EXPECT_EQ(nullptr, batch);
+
+ // Local 00:00:24 is 05:51:00Z, which is 0001-01-01 00:00:00 with pytz's
offset.
+ auto valid = timestamp_block(6, {make_datetime(1, 1, 1, 0, 0, 24, 0)});
+ ASSERT_TRUE(converter.convert_to_arrow(valid,
arrow::default_memory_pool(), &batch).ok());
+ const auto& values = static_cast<const
arrow::TimestampArray&>(*batch->column(0));
+ EXPECT_EQ(-62135575740000000LL, values.Value(0));
+}
+
+TEST_F(ArrowFlightTimestampTest, NaiveBoundsDoNotDependOnSessionTimezone) {
+ cctz::time_zone timezone;
+ ASSERT_TRUE(TimezoneUtils::find_cctz_time_zone("+08:00", timezone));
+ auto block = timestamp_block(6, {make_datetime(1, 1, 1, 0, 0, 0, 0),
+ make_datetime(9999, 12, 31, 23, 59, 59,
999999)});
+ ArrowFlightArrowBlockConvertor converter(block, "+08:00", timezone);
+ ASSERT_TRUE(converter.init().ok());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ ASSERT_TRUE(converter.convert_to_arrow(block,
arrow::default_memory_pool(), &batch).ok());
+ const auto& values = static_cast<const
arrow::TimestampArray&>(*batch->column(0));
+ EXPECT_EQ(-62135596800000000LL, values.Value(0));
+ EXPECT_EQ(253402300799999999LL, values.Value(1));
+}
+
+TEST_F(ArrowFlightTimestampTest, IgnoresTimestampsMaskedByNullParents) {
+ auto datetime = std::make_shared<DataTypeDateTimeV2>(6);
+ const auto invalid = Field::create_field<TYPE_DATETIMEV2>(DateTime {});
+ DataTypes types {
+ datetime, std::make_shared<DataTypeStruct>(DataTypes {datetime},
Strings {"child"}),
+ std::make_shared<DataTypeArray>(datetime),
+ std::make_shared<DataTypeMap>(make_nullable(datetime),
make_nullable(datetime))};
+ FieldVector fields {
+ invalid, Field::create_field<TYPE_STRUCT>(Struct {invalid}),
+ Field::create_field<TYPE_ARRAY>(Array {invalid}),
+ Field::create_field<TYPE_MAP>(Map
{Field::create_field<TYPE_ARRAY>(Array {invalid}),
+
Field::create_field<TYPE_ARRAY>(Array {invalid})})};
+ for (size_t i = 0; i < types.size(); ++i) {
+ auto data = types[i]->create_column();
+ data->insert(fields[i]);
+ auto nulls = ColumnUInt8::create();
+ nulls->insert_value(1);
+ Block block {{ColumnNullable::create(std::move(data),
std::move(nulls)),
+ make_nullable(types[i]), "masked"}};
+ ArrowFlightArrowBlockConvertor converter(block, "UTC",
cctz::utc_time_zone());
+ ASSERT_TRUE(converter.init().ok());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ const auto status = converter.convert_to_arrow(block,
arrow::default_memory_pool(), &batch);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_TRUE(batch->column(0)->IsNull(0));
+ }
+}
+
+} // namespace
+} // namespace doris
diff --git a/be/test/runtime/result_queue_mgr_test.cpp
b/be/test/runtime/result_queue_mgr_test.cpp
index 707cb151306..ab80f8cad7e 100644
--- a/be/test/runtime/result_queue_mgr_test.cpp
+++ b/be/test/runtime/result_queue_mgr_test.cpp
@@ -26,11 +26,14 @@
#include <gtest/gtest-message.h>
#include <gtest/gtest-test-part.h>
+#include <chrono>
#include <memory>
#include <ostream>
+#include <thread>
#include <utility>
#include <vector>
+#include "common/status.h"
#include "gtest/gtest_pred_impl.h"
#include "runtime/record_batch_queue.h"
@@ -134,6 +137,58 @@ TEST_F(ResultQueueMgrTest, fetch_result_end) {
EXPECT_TRUE(result == nullptr);
}
+// MemoryScratchSink puts the sentinel when its task closes, also after a
failure. A successful
+// close records OK, which must leave the sentinel a normal end of the scan.
+TEST_F(ResultQueueMgrTest, fetch_result_end_after_ok_status) {
+ ResultQueueMgr queue_mgr;
+ TUniqueId query_id;
+ query_id.lo = 10;
+ query_id.hi = 100;
+
+ BlockQueueSharedPtr block_queue_t;
+ queue_mgr.create_queue(query_id, &block_queue_t);
+ block_queue_t->update_status(Status::OK());
+ block_queue_t->blocking_put(nullptr);
+
+ std::shared_ptr<arrow::RecordBatch> result;
+ bool eos = false;
+ EXPECT_TRUE(queue_mgr.fetch_result(query_id, &result, &eos).ok());
+ EXPECT_TRUE(eos);
+}
+
+// A failed scan (for example a batch the Arrow Flight convertor rejects) also
ends with the
+// sentinel. The failure is recorded before it, and must reach the external
scanner instead of a
+// successful end with the rows fetched so far, also when its fetch already
waits for data: that
+// fetch passed the status check before the failure happened.
+TEST_F(ResultQueueMgrTest,
fetch_result_reports_failure_recorded_before_sentinel) {
+ ResultQueueMgr queue_mgr;
+ TUniqueId query_id;
+ query_id.lo = 10;
+ query_id.hi = 100;
+
+ BlockQueueSharedPtr block_queue_t;
+ queue_mgr.create_queue(query_id, &block_queue_t);
+
+ std::shared_ptr<arrow::RecordBatch> waiting_result;
+ bool waiting_eos = false;
+ Status waiting_status;
+ std::thread waiting_fetch([&]() {
+ waiting_status = queue_mgr.fetch_result(query_id, &waiting_result,
&waiting_eos);
+ });
+ // Let the fetch pass its first status check and block on the empty queue.
+ std::this_thread::sleep_for(std::chrono::milliseconds(200));
+ block_queue_t->update_status(Status::InvalidArgument("rejected Arrow
batch"));
+ block_queue_t->blocking_put(nullptr);
+ waiting_fetch.join();
+ EXPECT_TRUE(waiting_status.is<ErrorCode::INVALID_ARGUMENT>()) <<
waiting_status;
+
+ // Later fetches keep reporting the failure instead of the end of the scan.
+ std::shared_ptr<arrow::RecordBatch> result;
+ bool eos = false;
+ const auto status = queue_mgr.fetch_result(query_id, &result, &eos);
+ EXPECT_TRUE(status.is<ErrorCode::INVALID_ARGUMENT>()) << status;
+}
+
TEST_F(ResultQueueMgrTest, normal_cancel) {
TUniqueId query_id;
query_id.lo = 10;
diff --git
a/regression-test/data/external_table_p0/remote_doris/test_remote_doris_all_types_select.out
b/regression-test/data/external_table_p0/remote_doris/test_remote_doris_all_types_select.out
index a8823bb027e..70fb18a6da0 100644
---
a/regression-test/data/external_table_p0/remote_doris/test_remote_doris_all_types_select.out
+++
b/regression-test/data/external_table_p0/remote_doris/test_remote_doris_all_types_select.out
@@ -1,12 +1,12 @@
-- This file is automatically generated. You should know what you did if you
want to edit this
-- !sql --
-2025-05-18T01:00 true -128 -32768 -2147483648
-9223372036854775808 -1234567890123456790 -123.456 -123456.789
-123457 -123456789012346 -1234567890123456789012345678 1970-01-01
0000-01-01T00:00 A Hello Hello, Doris! ["apple", "banana",
"orange"] {"Emily":101, "age":25} {"f1":11, "f2":3.14, "f3":"Emily"}
{"k1":"v31","k2":300,"k3":[123,456],"k4":[],"k5":{"i1":"iv1"}}
+2025-05-18T01:00 true -128 -32768 -2147483648
-9223372036854775808 -1234567890123456790 -123.456 -123456.789
-123457 -123456789012346 -1234567890123456789012345678 1970-01-01
0001-01-01T00:00 A Hello Hello, Doris! ["apple", "banana",
"orange"] {"Emily":101, "age":25} {"f1":11, "f2":3.14, "f3":"Emily"}
{"k1":"v31","k2":300,"k3":[123,456],"k4":[],"k5":{"i1":"iv1"}}
2025-05-18T02:00 \N \N \N \N \N \N \N
\N \N \N \N \N \N \N \N \N \N
\N \N \N
2025-05-18T03:00 false 127 32767 2147483647
9223372036854775807 1234567890123456789 123.456 123456.789 123457
123456789012346 1234567890123456789012345678 9999-12-31
9999-12-31T23:59:59 [] {} {"f1":11,
"f2":3.14, "f3":"Emily"} {}
2025-05-18T04:00 true 0 0 0 0 0 0.0
0 0 0 0 2023-10-01 2023-10-01T12:34:56 A
Hello Hello, Doris! ["apple", "banana", "orange"] {"Emily":101, "age":25}
{"f1":11, "f2":3.14, "f3":"Emily"} []
-- !sql --
-2025-05-18T01:00 [1] [-128] [-32768] [-2147483648]
[-9223372036854775808] [-1234567890123456790] [-123.456] [-123456.789]
[-123457] [-123456789012346] [-1234567890123456789012345678]
["0000-01-01"] ["0000-01-01 00:00:00"] ["A"] ["Hello"] ["Hello,
Doris!"]
+2025-05-18T01:00 [1] [-128] [-32768] [-2147483648]
[-9223372036854775808] [-1234567890123456790] [-123.456] [-123456.789]
[-123457] [-123456789012346] [-1234567890123456789012345678]
["0000-01-01"] ["0001-01-01 00:00:00"] ["A"] ["Hello"] ["Hello,
Doris!"]
2025-05-18T02:00 [null] [null] [null] [null] [null] [null] [null]
[null] [null] [null] [null] [null] [null] [null] [null] [null]
2025-05-18T03:00 [0] [127] [32767] [2147483647]
[9223372036854775807] [1234567890123456789] [123.456] [123456.789]
[123457] [123456789012346] [1234567890123456789012345678]
["9999-12-31"] ["9999-12-31 23:59:59"] [""] [""] [""]
2025-05-18T04:00 [1] [0] [0] [0] [0] [0] [0]
[0] [0] [0] [0] ["2023-10-01"] ["2023-10-01 12:34:56"] ["A"]
["Hello"] ["Hello, Doris!"]
diff --git
a/regression-test/data/external_table_p0/remote_doris/test_remote_doris_statistics.out
b/regression-test/data/external_table_p0/remote_doris/test_remote_doris_statistics.out
index eecba5cbcd6..aebd716491e 100644
---
a/regression-test/data/external_table_p0/remote_doris/test_remote_doris_statistics.out
+++
b/regression-test/data/external_table_p0/remote_doris/test_remote_doris_statistics.out
@@ -4,7 +4,7 @@ c_bigint 4 3 1 -9223372036854775808
9223372036854775807 32
c_boolean 4 2 1 0 1 4
c_char 4 2 1 A 2
c_date 4 3 1 1970-01-01 9999-12-31 16
-c_datetime 4 3 1 0000-01-01 00:00:00 9999-12-31
23:59:59 32
+c_datetime 4 3 1 0001-01-01 00:00:00 9999-12-31
23:59:59 32
c_decimal18 4 3 1 -123456789012346 123456789012346
32
c_decimal32 4 3 1 -1234567890123456789012345678
1234567890123456789012345678 64
c_decimal9 4 3 1 -123457 123457 16
diff --git
a/regression-test/suites/arrow_flight_sql_p0/test_flight_int96_timestamp_range.groovy
b/regression-test/suites/arrow_flight_sql_p0/test_flight_int96_timestamp_range.groovy
new file mode 100644
index 00000000000..7cdc5f1bbfe
--- /dev/null
+++
b/regression-test/suites/arrow_flight_sql_p0/test_flight_int96_timestamp_range.groovy
@@ -0,0 +1,83 @@
+// 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.
+
+import java.sql.DriverManager
+import java.sql.Types
+import java.time.LocalDateTime
+
+suite("test_flight_int96_timestamp_range",
"arrow_flight_sql,external,hive,tvf,external_docker") {
+ if
(!"true".equalsIgnoreCase(context.config.otherConfigs.get("enableHiveTest"))) {
+ return
+ }
+ def host = context.config.otherConfigs.get("externalEnvIp")
+ def port = context.config.otherConfigs.get("hive2HdfsPort")
+ // External clusters can override the configured Flight port; use the
running FE's endpoint.
+ def frontend = jdbc_sql_return_maparray("SHOW FRONTENDS").find {
+ it.IsMaster.toString().toBoolean() && it.Alive.toString().toBoolean()
+ }
+ assertNotNull(frontend)
+ assertTrue(frontend.ArrowFlightSqlPort.toString().toInteger() > 0)
+ def flightUrl =
"jdbc:arrow-flight-sql://${frontend.Host}:${frontend.ArrowFlightSqlPort}" +
+
"/catalog=${context.dbName}?useServerPrepStmts=false&useSSL=false&useEncryption=false"
+ Class.forName("org.apache.arrow.driver.jdbc.ArrowFlightJdbcDriver")
+ def query = { String file ->
+ """SELECT * FROM HDFS(
+ "uri" =
"hdfs://${host}:${port}/user/doris/tvf_data/test_hdfs_parquet/group4/${file}",
+ "hadoop.username" = "doris", "format" = "parquet") LIMIT 10"""
+ }
+ // Keep the configured Flight identity without the TLS-dependent connect()
wrapper.
+ DriverManager.getConnection(flightUrl,
context.config.otherConfigs.get("extArrowFlightSqlUser"),
+
context.config.otherConfigs.get("extArrowFlightSqlPassword")).withCloseable {
flight ->
+ flight.createStatement().withCloseable { statement ->
+ // Nanosecond fractions in these fixtures truncate to valid
DATETIME(6) values.
+ for (def file :
["part-00000-570d8e52-652d-4892-8bdc-7fa5466ffa69.c000.snappy.parquet",
+
"part-00000-b945dfb5-9982-4f86-b903-dabef99caba1.c000.snappy.parquet",
+
"part-00000-721700d2-26d7-42a3-a8f9-b6601628ccd4.c000.snappy.parquet",
+
"part-00000-afeef968-a917-4d51-a652-e5a4214df453.c000.snappy.parquet"]) {
+ logger.info("Read INT96 Flight fixture: ${file}")
+ statement.executeQuery(query(file)).withCloseable { rows ->
+ assertTrue(rows.next())
+ def metadata = rows.getMetaData()
+ for (int column = 1; column <= metadata.getColumnCount();
++column) {
+ if (metadata.getColumnType(column) == Types.TIMESTAMP)
{
+ assertNotNull(rows.getObject(column,
LocalDateTime.class))
+ }
+ }
+
assertEquals(LocalDateTime.parse("9999-12-31T23:59:59.999999"),
+ rows.getObject(metadata.getColumnCount(),
LocalDateTime.class))
+ assertFalse(rows.next())
+ }
+ }
+ }
+ // Consume the result on the same discovered endpoint so zero-year
values must fail over Flight.
+ try {
+ flight.createStatement().withCloseable { statement ->
+
statement.executeQuery(query("int96_timestamps_nanos_outside_day_range.parquet"))
+ .withCloseable { rows ->
+ while (rows.next()) {
+ for (int column = 1; column <=
rows.getMetaData().getColumnCount(); ++column) {
+ rows.getObject(column)
+ }
+ }
+ }
+ }
+ assertTrue(false, "Expected an out-of-range Flight timestamp
error")
+ } catch (Exception error) {
+ assertTrue(error.toString().contains("outside the supported
0001-9999 range"), error.toString())
+ }
+ }
+}
diff --git
a/regression-test/suites/arrow_flight_sql_p0/test_flight_timestamp_range.groovy
b/regression-test/suites/arrow_flight_sql_p0/test_flight_timestamp_range.groovy
new file mode 100644
index 00000000000..c494bd84684
--- /dev/null
+++
b/regression-test/suites/arrow_flight_sql_p0/test_flight_timestamp_range.groovy
@@ -0,0 +1,81 @@
+// 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.
+
+import java.time.LocalDateTime
+
+suite("test_flight_timestamp_range", "arrow_flight_sql") {
+ // Reuse the configured Flight connection so TLS settings cannot skip the
assertions or change credentials.
+ def flight = context.getArrowFlightSqlConnection()
+ def input = "${context.dbName}.flight_timestamp_input"
+ def expectRangeError = { String query ->
+ try {
+ arrow_flight_sql(query)
+ assertTrue(false, "Expected an out-of-range Flight timestamp
error")
+ } catch (Exception error) {
+ assertTrue(error.toString().contains("outside the supported
0001-9999 range"), error.toString())
+ }
+ }
+ arrow_flight_sql "DROP TABLE IF EXISTS ${input}"
+ arrow_flight_sql """CREATE TABLE ${input} (id INT, value STRING)
+ DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")"""
+ try {
+ arrow_flight_sql """INSERT INTO ${input} VALUES
+ (1, '0001-01-01 00:00:00.000000'),
+ (2, '9999-12-31 23:59:59.999999'),
+ (3, '1969-12-31 23:59:59.999999'),
+ (4, NULL), (5, '0000-12-31 00:00:00.000000')"""
+ // Check that the source value reaches the date conversion instead of
becoming SQL NULL.
+ assertEquals([['0000-12-31 00:00:00.000000']],
+ arrow_flight_sql("SELECT CAST(CAST(value AS DATETIME(6)) AS
STRING) FROM ${input} WHERE id = 5"))
+ for (def scale : [0, 3, 6]) {
+ expectRangeError("SELECT CAST(value AS DATETIME(${scale})) AS
event_time FROM ${input} WHERE id = 5")
+ }
+ expectRangeError("SELECT CAST('0000-12-31 00:00:00' AS DATETIME(6)) AS
event_time")
+ for (def expression : ["array(CAST(value AS DATETIME(6)))",
+ "named_struct('child', CAST(value AS
DATETIME(6)))",
+ "map('key', CAST(value AS DATETIME(6)))",
+ "map(CAST(value AS DATETIME(6)), 'value')"]) {
+ expectRangeError("SELECT ${expression} AS nested FROM ${input}
WHERE id = 5")
+ }
+ expectRangeError("SELECT CAST(value AS DATETIME(6)) AS event_time FROM
${input} ORDER BY id")
+ flight.createStatement().withCloseable { statement ->
+ statement.executeQuery("SELECT CAST(value AS DATETIME(6)) FROM
${input} WHERE id <= 4 ORDER BY id")
+ .withCloseable { rows ->
+ // Typed access preserves proleptic calendar dates and
microseconds.
+ for (def expected : ["0001-01-01T00:00:00",
"9999-12-31T23:59:59.999999",
+ "1969-12-31T23:59:59.999999"]) {
+ assertTrue(rows.next())
+ assertEquals(LocalDateTime.parse(expected),
rows.getObject(1, LocalDateTime.class))
+ }
+ assertTrue(rows.next())
+ assertNull(rows.getObject(1, LocalDateTime.class))
+ assertFalse(rows.next())
+ }
+ statement.executeQuery("SELECT array(CAST(value AS DATETIME(6)))
FROM ${input} WHERE id = 4")
+ .withCloseable { rows ->
+ assertTrue(rows.next())
+ def values = rows.getArray(1).getArray()
+ assertEquals(1, values.length)
+ assertNull(values[0])
+ assertFalse(rows.next())
+ }
+ }
+ } finally {
+ arrow_flight_sql "DROP TABLE IF EXISTS ${input}"
+ }
+}
diff --git
a/regression-test/suites/arrow_flight_sql_p0/test_flight_utf8_validation.groovy
b/regression-test/suites/arrow_flight_sql_p0/test_flight_utf8_validation.groovy
new file mode 100644
index 00000000000..f19312b70d3
--- /dev/null
+++
b/regression-test/suites/arrow_flight_sql_p0/test_flight_utf8_validation.groovy
@@ -0,0 +1,71 @@
+// 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.
+
+suite("test_flight_utf8_validation", "arrow_flight_sql") {
+ // Reuse Flight credentials and avoid the TLS-dependent early return in
connect().
+ def flight = context.getArrowFlightSqlConnection()
+ def input = "${context.dbName}.flight_utf8_input"
+ def expectInvalidUtf8 = { String query ->
+ try {
+ arrow_flight_sql(query)
+ assertTrue(false, "Expected an invalid UTF-8 Flight result error")
+ } catch (Exception error) {
+ assertTrue(error.toString().contains("Invalid UTF8"),
error.toString())
+ }
+ }
+ arrow_flight_sql "DROP TABLE IF EXISTS ${input}"
+ arrow_flight_sql """CREATE TABLE ${input} (id INT, encoded STRING)
+ DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")"""
+ try {
+ arrow_flight_sql """INSERT INTO ${input} VALUES
+ (1, '616263'), (2, 'E4B8ADF09F9880'), (3, ''), (4, NULL),
+ (5, '84'), (6, 'C0AF'), (7, 'EDA080'), (8, 'F4908080'),
+ (9, 'C2'), (10, 'A2')"""
+ for (def id : [5, 6, 7, 8]) {
+ expectInvalidUtf8("SELECT unhex(encoded) AS payload FROM ${input}
WHERE id = ${id}")
+ }
+ // Constant folding can replace invalid bytes with valid text before
Flight conversion.
+ // Keep a table scan and skip folding so the constant is evaluated and
validated on the BE.
+ expectInvalidUtf8("""SELECT /*+ SET_VAR(debug_skip_fold_constant=true)
*/
+ unhex('84') AS payload FROM ${input} WHERE id = 1""")
+ // These bytes are valid only when concatenated; each row must be
valid independently.
+ expectInvalidUtf8("SELECT unhex(encoded) AS payload FROM ${input}
WHERE id IN (9, 10) ORDER BY id")
+ for (def expression : ["array(unhex(encoded))",
+ "named_struct('child', unhex(encoded))",
+ "map('key', unhex(encoded))",
+ "map(unhex(encoded), 'value')"]) {
+ expectInvalidUtf8("SELECT ${expression} AS nested FROM ${input}
WHERE id = 5")
+ }
+ // Reaching an invalid value after valid rows must fail rather than
publish malformed text.
+ expectInvalidUtf8("SELECT unhex(encoded) AS payload FROM ${input}
ORDER BY id")
+ assertEquals([['abc'], ['中😀'], [''], [null]],
+ arrow_flight_sql("SELECT unhex(encoded) FROM ${input} WHERE id
<= 4 ORDER BY id"))
+ // Binary payloads may contain arbitrary bytes and must keep their
binary Arrow type.
+ flight.createStatement().withCloseable { statement ->
+ statement.executeQuery("SELECT to_binary(encoded) AS payload FROM
${input} WHERE id = 5")
+ .withCloseable { rows ->
+ assertTrue(rows.next())
+ assertTrue(rows.getObject(1) instanceof byte[])
+ assertEquals([0x84], rows.getBytes(1).collect { it & 0xff })
+ assertFalse(rows.next())
+ }
+ }
+ } finally {
+ arrow_flight_sql "DROP TABLE IF EXISTS ${input}"
+ }
+}
diff --git
a/regression-test/suites/arrow_flight_sql_p0/test_query_plan_scan_validation.groovy
b/regression-test/suites/arrow_flight_sql_p0/test_query_plan_scan_validation.groovy
new file mode 100644
index 00000000000..b1da14270d4
--- /dev/null
+++
b/regression-test/suites/arrow_flight_sql_p0/test_query_plan_scan_validation.groovy
@@ -0,0 +1,120 @@
+// 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.
+
+import
org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.memory.RootAllocator
+import
org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.vector.ipc.ArrowStreamReader
+import org.apache.doris.regression.util.Http
+import org.apache.doris.thrift.BackendService
+import org.apache.doris.thrift.TScanCloseParams
+import org.apache.doris.thrift.TScanNextBatchParams
+import org.apache.doris.thrift.TScanOpenParams
+import org.apache.doris.thrift.TStatusCode
+import org.apache.thrift.protocol.TBinaryProtocol
+import org.apache.thrift.transport.TSocket
+
+// The Spark and Flink Doris connectors read a table through FE's _query_plan
API and the BE
+// Thrift scanner (open_scanner / get_next / close_scanner), and the BE builds
the batches with
+// the Arrow Flight convertor. A scan that fails on the BE, here on values
that convertor
+// rejects, must reach the reader as an error instead of a normal end after
the rows fetched so
+// far (none here).
+suite("test_query_plan_scan_validation", "arrow_flight_sql") {
+ def db = context.dbName
+ sql "DROP TABLE IF EXISTS query_plan_scan_input"
+ sql """CREATE TABLE query_plan_scan_input (id INT, encoded STRING,
event_time STRING)
+ DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")"""
+ sql """INSERT INTO query_plan_scan_input VALUES
+ (1, '616263', '2024-01-01 00:00:00'), (2, '84', '0000-12-31
00:00:00')"""
+
+ // Reads a _query_plan scan the way the connectors do. Returns the number
of rows read and
+ // the error get_next reported, or null when it reported the end of the
scan.
+ def scan = { String query ->
+ def plan = Http.POST(
+
"http://${context.config.feHttpAddress}/api/${db}/query_plan_scan_input/_query_plan";,
+ [sql: query], true, context.config.feHttpUser,
context.config.feHttpPassword)
+ assertEquals(0, plan.code, plan.toString())
+ assertEquals(200, plan.data.status, plan.data.toString())
+ // One bucket and one replica: the plan has a single tablet on a
single BE.
+ def tablets = plan.data.partitions.keySet().collect { it as long }
+ String routing = plan.data.partitions.values().first().routings.first()
+ String host = routing.substring(0,
routing.lastIndexOf(':')).replace('[', '').replace(']', '')
+ int port = routing.substring(routing.lastIndexOf(':') + 1) as int
+
+ def socket = new TSocket(host, port, 60000)
+ socket.open()
+ def allocator = new RootAllocator()
+ try {
+ def client = new BackendService.Client(new TBinaryProtocol(socket))
+ def params = new TScanOpenParams()
+ params.setCluster("default_cluster")
+ params.setDatabase(db)
+ params.setTable("query_plan_scan_input")
+ params.setTabletIds(tablets)
+ params.setOpaquedQueryPlan(plan.data.opaqued_query_plan)
+ params.setUser(context.config.feHttpUser)
+ params.setPasswd(context.config.feHttpPassword ?: "")
+ def opened = client.openScanner(params)
+ assertEquals(TStatusCode.OK, opened.getStatus().getStatusCode(),
opened.toString())
+ long rows = 0
+ String error = null
+ try {
+ while (true) {
+ def batch = client.getNext(new TScanNextBatchParams()
+
.setContextId(opened.getContextId()).setOffset(rows))
+ if (batch.getStatus().getStatusCode() != TStatusCode.OK) {
+ error = batch.getStatus().getErrorMsgs().join("; ")
+ break
+ }
+ if (batch.isEos()) {
+ break
+ }
+ def reader = new ArrowStreamReader(new
ByteArrayInputStream(batch.getRows()), allocator)
+ try {
+ while (reader.loadNextBatch()) {
+ rows += reader.getVectorSchemaRoot().getRowCount()
+ }
+ } finally {
+ reader.close()
+ }
+ }
+ } finally {
+ client.closeScanner(new
TScanCloseParams().setContextId(opened.getContextId()))
+ }
+ return [rows, error]
+ } finally {
+ allocator.close()
+ socket.close()
+ }
+ }
+
+ // Values the convertor accepts are read in full and the scan ends
normally.
+ def (validRows, validError) = scan("""SELECT id, unhex(encoded),
CAST(event_time AS DATETIME(6))
+ FROM ${db}.query_plan_scan_input WHERE id = 1""")
+ assertEquals(1L, validRows)
+ assertNull(validError)
+
+ // Invalid UTF-8 in a STRING result.
+ def (utf8Rows, utf8Error) = scan("SELECT id, unhex(encoded) FROM
${db}.query_plan_scan_input")
+ assertTrue(utf8Error != null && utf8Error.contains("Invalid UTF8"),
+ "Expected an invalid UTF-8 scan error, read ${utf8Rows} rows,
error: ${utf8Error}")
+
+ // A DATETIME before year 1.
+ def (timeRows, timeError) = scan("""SELECT id, CAST(event_time AS
DATETIME(6))
+ FROM ${db}.query_plan_scan_input""")
+ assertTrue(timeError != null && timeError.contains("outside the supported
0001-9999 range"),
+ "Expected an out-of-range timestamp scan error, read ${timeRows}
rows, error: ${timeError}")
+}
diff --git
a/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_all_types_select.groovy
b/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_all_types_select.groovy
index 72f443b8b72..45e5ec64784 100644
---
a/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_all_types_select.groovy
+++
b/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_all_types_select.groovy
@@ -67,8 +67,9 @@ suite("test_remote_doris_all_types_select", "p0,external") {
);
"""
+ // Successful Flight reads require timestamps within the supported
0001-9999 range.
sql """
- INSERT INTO
`test_remote_doris_all_types_select_db`.`test_remote_doris_all_types_select_t`
values('2025-05-18 01:00:00.000', true, -128, -32768, -2147483648,
-9223372036854775808, -1234567890123456790, -123.456, -123456.789, -123457,
-123456789012346, -1234567890123456789012345678, '1970-01-01', '0000-01-01
00:00:00', 'A', 'Hello', 'Hello, Doris!', '["apple", "banana", "orange"]',
{"Emily":101,"age":25} , {11, 3.14, "Emily"}, '{"k1":"v31", "k2": 300, "k3":
[123, 456], "k4": [], " [...]
+ INSERT INTO
`test_remote_doris_all_types_select_db`.`test_remote_doris_all_types_select_t`
values('2025-05-18 01:00:00.000', true, -128, -32768, -2147483648,
-9223372036854775808, -1234567890123456790, -123.456, -123456.789, -123457,
-123456789012346, -1234567890123456789012345678, '1970-01-01', '0001-01-01
00:00:00', 'A', 'Hello', 'Hello, Doris!', '["apple", "banana", "orange"]',
{"Emily":101,"age":25} , {11, 3.14, "Emily"}, '{"k1":"v31", "k2": 300, "k3":
[123, 456], "k4": [], " [...]
"""
sql """
INSERT INTO
`test_remote_doris_all_types_select_db`.`test_remote_doris_all_types_select_t`
values('2025-05-18 02:00:00.000', NULL, NULL, NULL, NULL, NULL, NULL, NULL,
NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL)
@@ -108,7 +109,7 @@ suite("test_remote_doris_all_types_select", "p0,external") {
"""
sql """
- INSERT INTO
`test_remote_doris_all_types_select_db`.`test_remote_doris_all_types_select_t2`
values('2025-05-18 01:00:00.000', [true], [-128], [-32768], [-2147483648],
[-9223372036854775808], [-1234567890123456790], [-123.456], [-123456.789],
[-123457], [-123456789012346], [-1234567890123456789012345678], ['0000-01-01'],
['0000-01-01 00:00:00'], ['A'], ['Hello'], ['Hello, Doris!'])
+ INSERT INTO
`test_remote_doris_all_types_select_db`.`test_remote_doris_all_types_select_t2`
values('2025-05-18 01:00:00.000', [true], [-128], [-32768], [-2147483648],
[-9223372036854775808], [-1234567890123456790], [-123.456], [-123456.789],
[-123457], [-123456789012346], [-1234567890123456789012345678], ['0000-01-01'],
['0001-01-01 00:00:00'], ['A'], ['Hello'], ['Hello, Doris!'])
"""
sql """
INSERT INTO
`test_remote_doris_all_types_select_db`.`test_remote_doris_all_types_select_t2`
values('2025-05-18 02:00:00.000', [NULL], [NULL], [NULL], [NULL], [NULL],
[NULL], [NULL], [NULL], [NULL], [NULL], [NULL], [NULL], [NULL], [NULL], [NULL],
[NULL])
@@ -172,6 +173,16 @@ suite("test_remote_doris_all_types_select", "p0,external")
{
select * from
`test_remote_doris_all_types_select_catalog`.`test_remote_doris_all_types_select_db`.`test_remote_doris_all_types_select_t3`
order by id
"""
+ // Keep zero-year rejection covered separately from successful all-type
round trips.
+ sql """INSERT INTO
test_remote_doris_all_types_select_db.test_remote_doris_all_types_select_t3
+ (id, datetime_0) VALUES ('2025-05-18 02:00:00', '0000-01-01
00:00:00')"""
+ test {
+ sql """SELECT datetime_0 FROM
test_remote_doris_all_types_select_catalog
+
.test_remote_doris_all_types_select_db.test_remote_doris_all_types_select_t3
+ WHERE id = '2025-05-18 02:00:00'"""
+ exception "outside the supported 0001-9999 range"
+ }
+
sql """ DROP DATABASE IF EXISTS test_remote_doris_all_types_select_db """
sql """ DROP CATALOG IF EXISTS
`test_remote_doris_all_types_select_catalog` """
}
diff --git
a/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_statistics.groovy
b/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_statistics.groovy
index caa307db7fa..60381e2282c 100644
---
a/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_statistics.groovy
+++
b/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_statistics.groovy
@@ -82,8 +82,9 @@ suite("test_remote_doris_statistics", "p0,external") {
);
"""
+ // Successful Flight reads require timestamps within the supported
0001-9999 range.
sql """
- INSERT INTO
`test_remote_doris_statistics_db`.`test_remote_doris_statistics_t1`
values('2025-05-18 01:00:00.000', true, -128, -32768, -2147483648,
-9223372036854775808, -1234567890123456790, -123.456, -123456.789, -123457,
-123456789012346, -1234567890123456789012345678, '1970-01-01', '0000-01-01
00:00:00', 'A', 'Hello', 'Hello, Doris!', '["apple", "banana", "orange"]',
{"Emily":101,"age":25} , {11, 3.14, "Emily"})
+ INSERT INTO
`test_remote_doris_statistics_db`.`test_remote_doris_statistics_t1`
values('2025-05-18 01:00:00.000', true, -128, -32768, -2147483648,
-9223372036854775808, -1234567890123456790, -123.456, -123456.789, -123457,
-123456789012346, -1234567890123456789012345678, '1970-01-01', '0001-01-01
00:00:00', 'A', 'Hello', 'Hello, Doris!', '["apple", "banana", "orange"]',
{"Emily":101,"age":25} , {11, 3.14, "Emily"})
"""
sql """
INSERT INTO
`test_remote_doris_statistics_db`.`test_remote_doris_statistics_t1`
values('2025-05-18 02:00:00.000', NULL, NULL, NULL, NULL, NULL, NULL, NULL,
NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL)
diff --git
a/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_table_stats.groovy
b/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_table_stats.groovy
index 6227f4b170d..2184b737658 100644
---
a/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_table_stats.groovy
+++
b/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_table_stats.groovy
@@ -82,8 +82,9 @@ suite("test_remote_doris_table_stats", "p0,external") {
);
"""
+ // Successful Flight reads require timestamps within the supported
0001-9999 range.
sql """
- INSERT INTO
`test_remote_doris_table_stats_db`.`test_remote_doris_table_stats_t1`
values('2025-05-18 01:00:00.000', true, -128, -32768, -2147483648,
-9223372036854775808, -1234567890123456790, -123.456, -123456.789, -123457,
-123456789012346, -1234567890123456789012345678, '1970-01-01', '0000-01-01
00:00:00', 'A', 'Hello', 'Hello, Doris!', '["apple", "banana", "orange"]',
{"Emily":101,"age":25} , {11, 3.14, "Emily"})
+ INSERT INTO
`test_remote_doris_table_stats_db`.`test_remote_doris_table_stats_t1`
values('2025-05-18 01:00:00.000', true, -128, -32768, -2147483648,
-9223372036854775808, -1234567890123456790, -123.456, -123456.789, -123457,
-123456789012346, -1234567890123456789012345678, '1970-01-01', '0001-01-01
00:00:00', 'A', 'Hello', 'Hello, Doris!', '["apple", "banana", "orange"]',
{"Emily":101,"age":25} , {11, 3.14, "Emily"})
"""
sql """
INSERT INTO
`test_remote_doris_table_stats_db`.`test_remote_doris_table_stats_t1`
values('2025-05-18 02:00:00.000', NULL, NULL, NULL, NULL, NULL, NULL, NULL,
NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
