This is an automated email from the ASF dual-hosted git repository.
yiguolei 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 193c0ca652f [be](serde) support more arrow::Type in
read_column_from_arrow (#66489)
193c0ca652f is described below
commit 193c0ca652f25f3dfe276a22a3aedbbdef01cd8e
Author: zhangstar333 <[email protected]>
AuthorDate: Fri Aug 7 10:57:02 2026 +0800
[be](serde) support more arrow::Type in read_column_from_arrow (#66489)
### What problem does this PR solve?
Problem Summary:
- Support Arrow `LargeList` and `FixedSizeList` for Doris ARRAY.
- Support `Binary`, `LargeBinary`, and `FixedSizeBinary` for VARBINARY.
- Support additional Arrow numeric and temporal types, including
Float16, unsigned integers, Date64, TIME, and TIMESTAMPTZ.
- Improve Arrow offset, range, type, and temporal-unit validation.
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [x] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
be/src/core/data_type_serde/arrow_validation.h | 44 +++-
.../core/data_type_serde/data_type_array_serde.cpp | 84 +++++--
.../data_type_serde/data_type_datetimev2_serde.cpp | 5 +-
.../data_type_serde/data_type_datev2_serde.cpp | 62 +++++-
.../data_type_serde/data_type_number_serde.cpp | 70 +++++-
.../core/data_type_serde/data_type_time_serde.cpp | 94 ++++++++
be/src/core/data_type_serde/data_type_time_serde.h | 3 +
.../data_type_timestamptz_serde.cpp | 5 +-
.../data_type_serde/data_type_varbinary_serde.cpp | 83 +++++++
.../data_type_serde/data_type_varbinary_serde.h | 5 +-
.../data_type_serde/data_type_serde_array_test.cpp | 118 ++++++++++
.../data_type_serde_arrow_validation_test.cpp | 244 ++++++++++++++++++++-
.../data_type_serde_datetime_v2_test.cpp | 137 ++++++++++++
.../data_type_serde_number_test.cpp | 124 +++++++++++
.../data_type_serde_timestamptz_test.cpp | 36 +++
.../data_type_serde_varbinary_test.cpp | 52 ++++-
16 files changed, 1112 insertions(+), 54 deletions(-)
diff --git a/be/src/core/data_type_serde/arrow_validation.h
b/be/src/core/data_type_serde/arrow_validation.h
index 9d8b09b0cf1..f7381250277 100644
--- a/be/src/core/data_type_serde/arrow_validation.h
+++ b/be/src/core/data_type_serde/arrow_validation.h
@@ -93,6 +93,15 @@ inline std::shared_ptr<arrow::Int32Array>
get_int32_offsets_array(const arrow::A
return offsets;
}
+inline std::shared_ptr<arrow::Int64Array> get_int64_offsets_array(const
arrow::Array& array) {
+ auto offsets_array = static_cast<const
arrow::LargeListArray&>(array).offsets();
+ auto offsets = std::dynamic_pointer_cast<arrow::Int64Array>(offsets_array);
+ if (UNLIKELY(!offsets)) {
+ throw_invalid_arrow(array, "offsets array is not Int64Array");
+ }
+ return offsets;
+}
+
} // namespace arrow_validation_detail
inline void check_arrow_no_offset(const arrow::Array& array) {
@@ -210,31 +219,33 @@ inline void check_arrow_value_range(const arrow::Array&
array, int64_t offset, i
namespace arrow_validation_detail {
// Offsets buffers may come from external Arrow producers through Buffer::Wrap
or FFI and are not
-// guaranteed to be aligned to int32_t. Do not use Int32Array::Value() here
because it performs a
+// guaranteed to be naturally aligned. Do not use NumericArray::Value() here
because it performs a
// typed raw_values()[i] load and can trigger UBSan on misaligned buffers.
Keep this validation path
// consistent with the array/map readers below, which load offsets through
unaligned_load().
-inline int32_t read_int32_offset(const arrow::Int32Array& offsets, int64_t
index) {
+template <typename OffsetType, typename ArrowOffsetArray>
+inline OffsetType read_offset(const ArrowOffsetArray& offsets, int64_t index) {
const auto* data = reinterpret_cast<const uint8_t*>(offsets.raw_values());
- return unaligned_load<int32_t>(data + index * sizeof(int32_t));
+ return unaligned_load<OffsetType>(data + index * sizeof(OffsetType));
}
-inline int64_t check_arrow_offsets_range(const arrow::Int32Array& offsets,
int64_t start,
+template <typename OffsetType, typename ArrowOffsetArray>
+inline int64_t check_arrow_offsets_range(const ArrowOffsetArray& offsets,
int64_t start,
int64_t end) {
check_arrow_array_range(offsets, 0, offsets.length());
- check_arrow_fixed_width_buffer(offsets, sizeof(int32_t));
+ check_arrow_fixed_width_buffer(offsets, sizeof(OffsetType));
if (UNLIKELY(start < 0 || end < start || end >= offsets.length())) {
arrow_validation_detail::throw_invalid_arrow(
offsets, "offsets read range is invalid: start={}, end={},
offsets_length={}",
start, end, offsets.length());
}
- int64_t previous_offset = read_int32_offset(offsets, start);
+ int64_t previous_offset = read_offset<OffsetType>(offsets, start);
if (UNLIKELY(previous_offset < 0)) {
arrow_validation_detail::throw_invalid_arrow(
offsets, "offsets contain negative value: offset[{}]={}",
start, previous_offset);
}
for (int64_t i = start + 1; i <= end; ++i) {
- const int64_t current_offset = read_int32_offset(offsets, i);
+ const int64_t current_offset = read_offset<OffsetType>(offsets, i);
if (UNLIKELY(current_offset < previous_offset)) {
arrow_validation_detail::throw_invalid_arrow(
offsets,
@@ -253,7 +264,22 @@ inline void check_arrow_list_offsets(const
arrow::ListArray& array, int64_t star
check_arrow_array_range(array, start, end);
const auto offsets =
arrow_validation_detail::get_int32_offsets_array(array);
const int64_t last_offset =
- arrow_validation_detail::check_arrow_offsets_range(*offsets,
start, end);
+
arrow_validation_detail::check_arrow_offsets_range<int32_t>(*offsets, start,
end);
+ const int64_t values_length = array.values() ? array.values()->length() :
0;
+ if (UNLIKELY(last_offset > values_length)) {
+ arrow_validation_detail::throw_invalid_arrow(
+ array, "offsets exceed values length: last_offset={},
values_length={}",
+ last_offset, values_length);
+ }
+}
+
+// Validate LargeList offsets before reading offsets or recursing into values.
+inline void check_arrow_large_list_offsets(const arrow::LargeListArray& array,
int64_t start,
+ int64_t end) {
+ check_arrow_array_range(array, start, end);
+ const auto offsets =
arrow_validation_detail::get_int64_offsets_array(array);
+ const int64_t last_offset =
+
arrow_validation_detail::check_arrow_offsets_range<int64_t>(*offsets, start,
end);
const int64_t values_length = array.values() ? array.values()->length() :
0;
if (UNLIKELY(last_offset > values_length)) {
arrow_validation_detail::throw_invalid_arrow(
@@ -267,7 +293,7 @@ inline void check_arrow_map_offsets(const arrow::MapArray&
array, int64_t start,
check_arrow_array_range(array, start, end);
const auto offsets =
arrow_validation_detail::get_int32_offsets_array(array);
const int64_t last_offset =
- arrow_validation_detail::check_arrow_offsets_range(*offsets,
start, end);
+
arrow_validation_detail::check_arrow_offsets_range<int32_t>(*offsets, start,
end);
const int64_t keys_length = array.keys() ? array.keys()->length() : 0;
if (UNLIKELY(last_offset > keys_length)) {
arrow_validation_detail::throw_invalid_arrow(
diff --git a/be/src/core/data_type_serde/data_type_array_serde.cpp
b/be/src/core/data_type_serde/data_type_array_serde.cpp
index bfab0f1eddb..6c5a0927e6a 100644
--- a/be/src/core/data_type_serde/data_type_array_serde.cpp
+++ b/be/src/core/data_type_serde/data_type_array_serde.cpp
@@ -325,29 +325,67 @@ Status
DataTypeArraySerDe::read_column_from_arrow(IColumn& column, const arrow::
const cctz::time_zone& ctz)
const {
auto& column_array = static_cast<ColumnArray&>(column);
auto& offsets_data = column_array.get_offsets();
- const auto* concrete_array = dynamic_cast<const
arrow::ListArray*>(arrow_array);
- auto arrow_offsets_array = concrete_array->offsets();
- auto* arrow_offsets =
dynamic_cast<arrow::Int32Array*>(arrow_offsets_array.get());
- if (config::enable_arrow_input_validation) {
- check_arrow_list_offsets(*concrete_array, start, end);
- }
- auto prev_size = offsets_data.back();
- const auto* base_offsets_ptr = reinterpret_cast<const
uint8_t*>(arrow_offsets->raw_values());
- const size_t offset_element_size = sizeof(int32_t);
- const uint8_t* start_offset_ptr = base_offsets_ptr + start *
offset_element_size;
- const uint8_t* end_offset_ptr = base_offsets_ptr + end *
offset_element_size;
- auto arrow_nested_start_offset = unaligned_load<int32_t>(start_offset_ptr);
- auto arrow_nested_end_offset = unaligned_load<int32_t>(end_offset_ptr);
-
- for (auto i = start + 1; i < end + 1; ++i) {
- const uint8_t* current_offset_ptr = base_offsets_ptr + i *
offset_element_size;
- auto current_offset = unaligned_load<int32_t>(current_offset_ptr);
- // convert to doris offset, start from offsets.back()
- offsets_data.emplace_back(prev_size + current_offset -
arrow_nested_start_offset);
- }
- return nested_serde->read_column_from_arrow(
- column_array.get_data(), concrete_array->values().get(),
arrow_nested_start_offset,
- arrow_nested_end_offset, ctz);
+ const auto read_list = [&](const auto* concrete_array, const auto&
read_offset) -> Status {
+ const int64_t arrow_nested_start_offset = read_offset(start);
+ const int64_t arrow_nested_end_offset = read_offset(end);
+ const auto prev_size = offsets_data.back();
+ for (int64_t i = start + 1; i <= end; ++i) {
+ // Convert Arrow offsets to Doris offsets, starting at
offsets.back().
+ offsets_data.emplace_back(prev_size + read_offset(i) -
arrow_nested_start_offset);
+ }
+ return nested_serde->read_column_from_arrow(
+ column_array.get_data(), concrete_array->values().get(),
arrow_nested_start_offset,
+ arrow_nested_end_offset, ctz);
+ };
+
+ switch (arrow_array->type_id()) {
+ case arrow::Type::LIST: {
+ const auto* concrete_array = dynamic_cast<const
arrow::ListArray*>(arrow_array);
+ if (concrete_array == nullptr) {
+ return Status::InvalidArgument("Expected Arrow ListArray, got {}",
+ arrow_array->type()->name());
+ }
+ if (config::enable_arrow_input_validation) {
+ check_arrow_list_offsets(*concrete_array, start, end);
+ }
+ const auto* offsets = concrete_array->value_offsets()->data();
+ const auto array_offset = concrete_array->offset();
+ return read_list(concrete_array, [offsets, array_offset](int64_t
index) {
+ return unaligned_load<int32_t>(offsets + (array_offset + index) *
sizeof(int32_t));
+ });
+ }
+ case arrow::Type::LARGE_LIST: {
+ const auto* concrete_array = dynamic_cast<const
arrow::LargeListArray*>(arrow_array);
+ if (concrete_array == nullptr) {
+ return Status::InvalidArgument("Expected Arrow LargeListArray, got
{}",
+ arrow_array->type()->name());
+ }
+ if (config::enable_arrow_input_validation) {
+ check_arrow_large_list_offsets(*concrete_array, start, end);
+ }
+ const auto* offsets = concrete_array->value_offsets()->data();
+ const auto array_offset = concrete_array->offset();
+ return read_list(concrete_array, [offsets, array_offset](int64_t
index) {
+ return unaligned_load<int64_t>(offsets + (array_offset + index) *
sizeof(int64_t));
+ });
+ }
+ case arrow::Type::FIXED_SIZE_LIST: {
+ const auto* concrete_array = dynamic_cast<const
arrow::FixedSizeListArray*>(arrow_array);
+ if (concrete_array == nullptr) {
+ return Status::InvalidArgument("Expected Arrow FixedSizeListArray,
got {}",
+ arrow_array->type()->name());
+ }
+ if (config::enable_arrow_input_validation) {
+ check_arrow_array_range(*concrete_array, start, end);
+ }
+ return read_list(concrete_array, [concrete_array](int64_t index) {
+ return concrete_array->value_offset(index);
+ });
+ }
+ default:
+ return Status::InvalidArgument("Unsupported Arrow array type for Doris
ARRAY: {}",
+ arrow_array->type()->name());
+ }
}
Status DataTypeArraySerDe::write_column_to_mysql_binary(const IColumn& column,
diff --git a/be/src/core/data_type_serde/data_type_datetimev2_serde.cpp
b/be/src/core/data_type_serde/data_type_datetimev2_serde.cpp
index 78370a858c9..a4b11788986 100644
--- a/be/src/core/data_type_serde/data_type_datetimev2_serde.cpp
+++ b/be/src/core/data_type_serde/data_type_datetimev2_serde.cpp
@@ -665,6 +665,9 @@ Status
DataTypeDateTimeV2SerDe::read_column_from_arrow(IColumn& column,
type->unit());
}
}
+ // A timezone-naive Arrow timestamp is a wall-clock value. Decode its
epoch-based
+ // representation in UTC so the session timezone does not shift its
date/time fields.
+ const cctz::time_zone& real_ctz = type->timezone().empty() ?
cctz::utc_time_zone() : ctz;
const auto* base_ptr = reinterpret_cast<const
uint8_t*>(concrete_array->raw_values());
const size_t element_size = sizeof(int64_t);
for (auto value_i = start; value_i < end; ++value_i) {
@@ -681,7 +684,7 @@ Status
DataTypeDateTimeV2SerDe::read_column_from_arrow(IColumn& column,
--seconds;
remainder += divisor;
}
- v.from_unixtime(seconds, ctz);
+ v.from_unixtime(seconds, real_ctz);
// Get the fractional part.
// add 0 on the right to make it 6 digits. DateTimeV2Value
microsecond is 6 digits,
// the scale decides to keep the first few digits, so the valid
digits should be kept at the front.
diff --git a/be/src/core/data_type_serde/data_type_datev2_serde.cpp
b/be/src/core/data_type_serde/data_type_datev2_serde.cpp
index be9c841d233..02059045469 100644
--- a/be/src/core/data_type_serde/data_type_datev2_serde.cpp
+++ b/be/src/core/data_type_serde/data_type_datev2_serde.cpp
@@ -235,16 +235,58 @@ Status
DataTypeDateV2SerDe::read_column_from_arrow(IColumn& column, const arrow:
check_arrow_no_offset(*arrow_array);
}
auto& col_data = static_cast<ColumnDateV2&>(column).get_data();
- const auto* concrete_array = dynamic_cast<const
arrow::Date32Array*>(arrow_array);
- const auto* base_ptr = reinterpret_cast<const
uint8_t*>(concrete_array->raw_values());
- const size_t element_size = sizeof(int32_t);
- for (auto value_i = start; value_i < end; ++value_i) {
- const uint8_t* raw_byte_ptr = base_ptr + value_i * element_size;
- auto date_value = unaligned_load<int32_t>(raw_byte_ptr);
-
- DateV2Value<DateV2ValueType> v;
- v.get_date_from_daynr(date_value + date_threshold);
- col_data.emplace_back(v);
+ if (arrow_array->type_id() == arrow::Type::DATE64) {
+ static constexpr int64_t MILLISECONDS_PER_DAY = 24 * 60 * 60 * 1000;
+ const auto* concrete_array = dynamic_cast<const
arrow::Date64Array*>(arrow_array);
+ if (concrete_array == nullptr) {
+ return Status::InvalidArgument("Expected Arrow Date64Array, got
{}",
+ arrow_array->type()->name());
+ }
+ if (config::enable_arrow_input_validation) {
+ check_arrow_array_range(*concrete_array, start, end);
+ check_arrow_fixed_width_buffer(*concrete_array,
sizeof(arrow::Date64Array::value_type));
+ }
+ for (int64_t value_i = start; value_i < end; ++value_i) {
+ if (concrete_array->IsNull(value_i)) {
+ col_data.emplace_back(DateV2Value<DateV2ValueType>());
+ continue;
+ }
+ const int64_t milliseconds = concrete_array->Value(value_i);
+ if (milliseconds % MILLISECONDS_PER_DAY != 0) {
+ return Status::InvalidArgument(
+ "Arrow Date64 value must contain whole days: row={},
milliseconds={}",
+ value_i, milliseconds);
+ }
+ const int64_t daynr = milliseconds / MILLISECONDS_PER_DAY +
date_threshold;
+ DateV2Value<DateV2ValueType> value;
+ if (daynr <= 0 || daynr > DATE_MAX_DAYNR ||
+ !value.get_date_from_daynr(static_cast<uint64_t>(daynr))) {
+ return Status::InvalidArgument(
+ "Arrow Date64 value is outside the Doris DATE range: "
+ "row={}, milliseconds={}",
+ value_i, milliseconds);
+ }
+ col_data.emplace_back(value);
+ }
+ } else if (arrow_array->type_id() == arrow::Type::DATE32) {
+ const auto* concrete_array = dynamic_cast<const
arrow::Date32Array*>(arrow_array);
+ if (concrete_array == nullptr) {
+ return Status::InvalidArgument("Expected Arrow Date32Array, got
{}",
+ arrow_array->type()->name());
+ }
+ const auto* base_ptr = reinterpret_cast<const
uint8_t*>(concrete_array->raw_values());
+ const size_t element_size = sizeof(int32_t);
+ for (auto value_i = start; value_i < end; ++value_i) {
+ const uint8_t* raw_byte_ptr = base_ptr + value_i * element_size;
+ auto date_value = unaligned_load<int32_t>(raw_byte_ptr);
+
+ DateV2Value<DateV2ValueType> v;
+ v.get_date_from_daynr(date_value + date_threshold);
+ col_data.emplace_back(v);
+ }
+ } else {
+ return Status::InvalidArgument("Expected Arrow Date32Array or
Date64Array, got {}",
+ arrow_array->type()->name());
}
return Status::OK();
}
diff --git a/be/src/core/data_type_serde/data_type_number_serde.cpp
b/be/src/core/data_type_serde/data_type_number_serde.cpp
index ca30ec96d25..a1a4a3df65d 100644
--- a/be/src/core/data_type_serde/data_type_number_serde.cpp
+++ b/be/src/core/data_type_serde/data_type_number_serde.cpp
@@ -18,6 +18,7 @@
#include "core/data_type_serde/data_type_number_serde.h"
#include <arrow/builder.h>
+#include <arrow/util/float16.h>
#include <bit>
#include <cmath>
@@ -45,8 +46,6 @@
#include "exprs/function/cast/cast_to_basic_number_common.h"
#include "exprs/function/cast/cast_to_boolean.h"
#include "exprs/function/cast/cast_to_string.h"
-#include "storage/olap_common.h"
-#include "storage/types.h"
#include "util/jsonb_document.h"
#include "util/jsonb_document_cast.h"
#include "util/jsonb_writer.h"
@@ -342,6 +341,28 @@ Status read_integer_decoded_values(IColumn& column, const
DecodedColumnView& vie
}
}
+template <PrimitiveType DorisType, typename ArrowArrayType>
+Status read_widened_arrow_integer_values(IColumn& column, const arrow::Array*
arrow_array,
+ int64_t start, int64_t end) {
+ const auto* source = dynamic_cast<const ArrowArrayType*>(arrow_array);
+ if (source == nullptr) {
+ return Status::InvalidArgument("Expected a compatible Arrow numeric
array for {}, got {}",
+ column.get_name(),
arrow_array->type()->name());
+ }
+ if (config::enable_arrow_input_validation) {
+ check_arrow_fixed_width_buffer(*source, sizeof(typename
ArrowArrayType::value_type));
+ }
+
+ auto& data =
+ assert_cast<typename
PrimitiveTypeTraits<DorisType>::ColumnType&>(column).get_data();
+ for (int64_t row = start; row < end; ++row) {
+ using DorisCppType = typename PrimitiveTypeTraits<DorisType>::CppType;
+ data.push_back(source->IsNull(row) ? DorisCppType()
+ :
static_cast<DorisCppType>(source->Value(row)));
+ }
+ return Status::OK();
+}
+
template <typename DorisCppType, typename SourceType>
Status append_parquet_number(PaddedPODArray<DorisCppType>& data, const
uint8_t* values,
size_t num_values, const ParquetDecodeContext&
context,
@@ -933,6 +954,51 @@ Status
DataTypeNumberSerDe<T>::read_column_from_arrow(IColumn& column,
return Status::OK();
}
+ if constexpr (T == TYPE_FLOAT) {
+ if (arrow_array->type_id() == arrow::Type::HALF_FLOAT) {
+ const auto* concrete_array = dynamic_cast<const
arrow::HalfFloatArray*>(arrow_array);
+ if (concrete_array == nullptr) {
+ return Status::InvalidArgument("Expected Arrow HalfFloatArray,
got {}",
+ arrow_array->type()->name());
+ }
+ if (config::enable_arrow_input_validation) {
+ check_arrow_fixed_width_buffer(*concrete_array,
+
sizeof(arrow::HalfFloatArray::value_type));
+ }
+ for (int64_t i = start; i < end; ++i) {
+ const auto value =
+ concrete_array->IsNull(i)
+ ? 0.0F
+ :
arrow::util::Float16::FromBits(concrete_array->Value(i))
+ .ToFloat();
+ col_data.emplace_back(value);
+ }
+ return Status::OK();
+ }
+ }
+
+ if constexpr (T == TYPE_SMALLINT) {
+ if (arrow_array->type_id() == arrow::Type::UINT8) {
+ return read_widened_arrow_integer_values<TYPE_SMALLINT,
arrow::UInt8Array>(
+ column, arrow_array, start, end);
+ }
+ } else if constexpr (T == TYPE_INT) {
+ if (arrow_array->type_id() == arrow::Type::UINT16) {
+ return read_widened_arrow_integer_values<TYPE_INT,
arrow::UInt16Array>(
+ column, arrow_array, start, end);
+ }
+ } else if constexpr (T == TYPE_BIGINT) {
+ if (arrow_array->type_id() == arrow::Type::UINT32) {
+ return read_widened_arrow_integer_values<TYPE_BIGINT,
arrow::UInt32Array>(
+ column, arrow_array, start, end);
+ }
+ } else if constexpr (T == TYPE_LARGEINT) {
+ if (arrow_array->type_id() == arrow::Type::UINT64) {
+ return read_widened_arrow_integer_values<TYPE_LARGEINT,
arrow::UInt64Array>(
+ column, arrow_array, start, end);
+ }
+ }
+
// only for largeint(int128) type
if (arrow_array->type_id() == arrow::Type::STRING) {
const auto* concrete_array = dynamic_cast<const
arrow::StringArray*>(arrow_array);
diff --git a/be/src/core/data_type_serde/data_type_time_serde.cpp
b/be/src/core/data_type_serde/data_type_time_serde.cpp
index 9beb9792244..566bd3982b4 100644
--- a/be/src/core/data_type_serde/data_type_time_serde.cpp
+++ b/be/src/core/data_type_serde/data_type_time_serde.cpp
@@ -17,11 +17,16 @@
#include "core/data_type_serde/data_type_time_serde.h"
+#include <arrow/array.h>
+#include <arrow/type.h>
+
#include <limits>
+#include "common/config.h"
#include "core/data_type/data_type_decimal.h"
#include "core/data_type/data_type_number.h"
#include "core/data_type/primitive_type.h"
+#include "core/data_type_serde/arrow_validation.h"
#include "core/data_type_serde/decoded_column_view.h"
#include "core/data_type_serde/parquet_decode_source.h"
#include "core/value/time_value.h"
@@ -281,6 +286,95 @@ Status
DataTypeTimeV2SerDe::from_string_strict_mode(StringRef& str, IColumn& col
return Status::OK();
}
+Status DataTypeTimeV2SerDe::read_column_from_arrow(IColumn& column, const
arrow::Array* arrow_array,
+ int64_t start, int64_t end,
+ const cctz::time_zone&)
const {
+ if (config::enable_arrow_input_validation) {
+ check_arrow_array_range(*arrow_array, start, end);
+ }
+ auto& data = assert_cast<ColumnTimeV2&>(column).get_data();
+ if (arrow_array->type_id() == arrow::Type::TIME32) {
+ const auto* time_array = dynamic_cast<const
arrow::Time32Array*>(arrow_array);
+ if (time_array == nullptr) {
+ return Status::InvalidArgument("Expected Arrow Time32Array, got
{}",
+ arrow_array->type()->name());
+ }
+ if (config::enable_arrow_input_validation) {
+ check_arrow_fixed_width_buffer(*time_array,
sizeof(arrow::Time32Array::value_type));
+ }
+ const auto type =
std::static_pointer_cast<arrow::Time32Type>(arrow_array->type());
+ int64_t micros_per_unit = 0;
+ int64_t units_per_day = 0;
+ switch (type->unit()) {
+ case arrow::TimeUnit::SECOND:
+ micros_per_unit = TimeValue::ONE_SECOND_MICROSECONDS;
+ units_per_day = 24 * 60 * 60;
+ break;
+ case arrow::TimeUnit::MILLI:
+ micros_per_unit = 1000;
+ units_per_day = 24 * 60 * 60 * 1000;
+ break;
+ default:
+ return Status::InvalidArgument("Unsupported Arrow Time32 unit:
{}", type->unit());
+ }
+ for (int64_t row = start; row < end; ++row) {
+ if (time_array->IsNull(row)) {
+ data.emplace_back(0);
+ continue;
+ }
+ const int64_t value = time_array->Value(row);
+ if (value < 0 || value >= units_per_day) {
+ return Status::InvalidArgument(
+ "Arrow Time32 value is outside the time-of-day range:
row={}, value={}",
+ row, value);
+ }
+ data.emplace_back(static_cast<TimeValue::TimeType>(value *
micros_per_unit));
+ }
+ return Status::OK();
+ }
+
+ if (arrow_array->type_id() == arrow::Type::TIME64) {
+ const auto* time_array = dynamic_cast<const
arrow::Time64Array*>(arrow_array);
+ if (time_array == nullptr) {
+ return Status::InvalidArgument("Expected Arrow Time64Array, got
{}",
+ arrow_array->type()->name());
+ }
+ if (config::enable_arrow_input_validation) {
+ check_arrow_fixed_width_buffer(*time_array,
sizeof(arrow::Time64Array::value_type));
+ }
+ const auto type =
std::static_pointer_cast<arrow::Time64Type>(arrow_array->type());
+ int64_t units_per_day = 0;
+ switch (type->unit()) {
+ case arrow::TimeUnit::MICRO:
+ units_per_day = 24LL * 60 * 60 *
TimeValue::ONE_SECOND_MICROSECONDS;
+ break;
+ case arrow::TimeUnit::NANO:
+ units_per_day = 24LL * 60 * 60 * 1000 *
TimeValue::ONE_SECOND_MICROSECONDS;
+ break;
+ default:
+ return Status::InvalidArgument("Unsupported Arrow Time64 unit:
{}", type->unit());
+ }
+ for (int64_t row = start; row < end; ++row) {
+ if (time_array->IsNull(row)) {
+ data.emplace_back(0);
+ continue;
+ }
+ const int64_t value = time_array->Value(row);
+ if (value < 0 || value >= units_per_day) {
+ return Status::InvalidArgument(
+ "Arrow Time64 value is outside the time-of-day range:
row={}, value={}",
+ row, value);
+ }
+ const int64_t micros = type->unit() == arrow::TimeUnit::NANO ?
value / 1000 : value;
+ data.emplace_back(static_cast<TimeValue::TimeType>(micros));
+ }
+ return Status::OK();
+ }
+
+ return Status::InvalidArgument("Expected Arrow Time32Array or Time64Array,
got {}",
+ arrow_array->type()->name());
+}
+
Status DataTypeTimeV2SerDe::read_column_from_decoded_values(IColumn& column,
const
DecodedColumnView& view) const {
if (view.value_kind != DecodedValueKind::INT32 && view.value_kind !=
DecodedValueKind::INT64) {
diff --git a/be/src/core/data_type_serde/data_type_time_serde.h
b/be/src/core/data_type_serde/data_type_time_serde.h
index 90ebe128672..39c6bb69d84 100644
--- a/be/src/core/data_type_serde/data_type_time_serde.h
+++ b/be/src/core/data_type_serde/data_type_time_serde.h
@@ -47,6 +47,9 @@ public:
const FormatOptions& options,
const NullMap::value_type* null_map =
nullptr) const final;
+ Status read_column_from_arrow(IColumn& column, const arrow::Array*
arrow_array, int64_t start,
+ int64_t end, const cctz::time_zone& ctz)
const override;
+
template <typename IntDataType>
Status from_int_batch(const typename IntDataType::ColumnType& int_col,
ColumnNullable& target_col) const;
diff --git a/be/src/core/data_type_serde/data_type_timestamptz_serde.cpp
b/be/src/core/data_type_serde/data_type_timestamptz_serde.cpp
index 1e84fef5b05..d83bc0745d9 100644
--- a/be/src/core/data_type_serde/data_type_timestamptz_serde.cpp
+++ b/be/src/core/data_type_serde/data_type_timestamptz_serde.cpp
@@ -417,7 +417,7 @@ Status
DataTypeTimeStampTzSerDe::read_column_from_arrow(IColumn& column,
int64_t start, int64_t
end,
const cctz::time_zone&
ctz) const {
if (config::enable_arrow_input_validation) {
- check_arrow_no_offset(*arrow_array);
+ check_arrow_array_range(*arrow_array, start, end);
}
if (arrow_array->type()->id() != arrow::Type::TIMESTAMP) {
LOG(WARNING) << "not support convert to timestamptz from arrow type:"
@@ -426,6 +426,9 @@ Status
DataTypeTimeStampTzSerDe::read_column_from_arrow(IColumn& column,
arrow_array->type()->id());
}
const auto* concrete_array = assert_cast<const
arrow::TimestampArray*>(arrow_array);
+ if (config::enable_arrow_input_validation) {
+ check_arrow_fixed_width_buffer(*concrete_array,
sizeof(arrow::TimestampArray::value_type));
+ }
const auto type =
std::static_pointer_cast<arrow::TimestampType>(arrow_array->type());
// Scale each unit to the microseconds the column stores. NANO is divided
rather than refused:
// sub-microsecond precision is beyond what any Doris datetime type keeps,
and rejecting the
diff --git a/be/src/core/data_type_serde/data_type_varbinary_serde.cpp
b/be/src/core/data_type_serde/data_type_varbinary_serde.cpp
index 14d5edda0af..16ff1b20e8a 100644
--- a/be/src/core/data_type_serde/data_type_varbinary_serde.cpp
+++ b/be/src/core/data_type_serde/data_type_varbinary_serde.cpp
@@ -17,7 +17,11 @@
#include "core/data_type_serde/data_type_varbinary_serde.h"
+#include <cstring>
+
+#include "common/config.h"
#include "core/column/column_varbinary.h"
+#include "core/data_type_serde/arrow_validation.h"
#include "core/data_type_serde/parquet_decode_source.h"
namespace doris {
@@ -184,6 +188,85 @@ Status DataTypeVarbinarySerDe::write_column_to_arrow(const
IColumn& column, cons
return Status::OK();
}
+Status DataTypeVarbinarySerDe::read_column_from_arrow(IColumn& column,
+ const arrow::Array*
arrow_array,
+ int64_t start, int64_t
end,
+ const cctz::time_zone&
ctz) const {
+ auto& varbinary_column = assert_cast<ColumnVarbinary&>(column);
+ if (arrow_array->type_id() == arrow::Type::STRING ||
+ arrow_array->type_id() == arrow::Type::BINARY) {
+ const auto* concrete_array = assert_cast<const
arrow::BinaryArray*>(arrow_array);
+ if (config::enable_arrow_input_validation) {
+ check_arrow_array_range(*concrete_array, start, end);
+ check_arrow_binary_offsets_buffer(*concrete_array);
+ }
+ const auto& buffer = concrete_array->value_data();
+ const size_t buffer_size = buffer ?
static_cast<size_t>(buffer->size()) : 0;
+ const uint8_t* offsets_data = concrete_array->value_offsets()->data();
+ constexpr size_t offset_size = sizeof(int32_t);
+
+ for (auto offset_i = start; offset_i < end; ++offset_i) {
+ if (!concrete_array->IsNull(offset_i)) {
+ int32_t start_offset = 0;
+ int32_t end_offset = 0;
+ memcpy(&start_offset, offsets_data + offset_i * offset_size,
offset_size);
+ memcpy(&end_offset, offsets_data + (offset_i + 1) *
offset_size, offset_size);
+ const int32_t length = end_offset - start_offset;
+ if (config::enable_arrow_input_validation) {
+ check_arrow_value_range(*concrete_array, start_offset,
length, buffer_size);
+ }
+ varbinary_column.insert_data(
+ reinterpret_cast<const char*>(buffer->data() +
start_offset), length);
+ } else {
+ varbinary_column.insert_default();
+ }
+ }
+ } else if (arrow_array->type_id() == arrow::Type::FIXED_SIZE_BINARY) {
+ const auto* concrete_array = assert_cast<const
arrow::FixedSizeBinaryArray*>(arrow_array);
+ if (config::enable_arrow_input_validation) {
+ check_arrow_array_range(*concrete_array, start, end);
+ check_arrow_fixed_width_buffer(*concrete_array,
+
static_cast<size_t>(concrete_array->byte_width()));
+ }
+ const uint32_t width = concrete_array->byte_width();
+ for (auto offset_i = start; offset_i < end; ++offset_i) {
+ if (!concrete_array->IsNull(offset_i)) {
+ varbinary_column.insert_data(
+ reinterpret_cast<const
char*>(concrete_array->GetValue(offset_i)), width);
+ } else {
+ varbinary_column.insert_default();
+ }
+ }
+ } else if (arrow_array->type_id() == arrow::Type::LARGE_STRING ||
+ arrow_array->type_id() == arrow::Type::LARGE_BINARY) {
+ const auto* concrete_array = assert_cast<const
arrow::LargeBinaryArray*>(arrow_array);
+ if (config::enable_arrow_input_validation) {
+ check_arrow_array_range(*concrete_array, start, end);
+ check_arrow_binary_offsets_buffer(*concrete_array);
+ }
+ const auto& buffer = concrete_array->value_data();
+ const size_t buffer_size = buffer ?
static_cast<size_t>(buffer->size()) : 0;
+ for (auto offset_i = start; offset_i < end; ++offset_i) {
+ if (!concrete_array->IsNull(offset_i)) {
+ const auto value_offset =
concrete_array->value_offset(offset_i);
+ const auto value_length =
concrete_array->value_length(offset_i);
+ if (config::enable_arrow_input_validation) {
+ check_arrow_value_range(*concrete_array, value_offset,
value_length,
+ buffer_size);
+ }
+ varbinary_column.insert_data(
+ reinterpret_cast<const char*>(buffer->data() +
value_offset), value_length);
+ } else {
+ varbinary_column.insert_default();
+ }
+ }
+ } else {
+ return Status::InvalidArgument("Unsupported arrow type for varbinary
column: {}",
+ arrow_array->type()->name());
+ }
+ return Status::OK();
+}
+
Status DataTypeVarbinarySerDe::write_column_to_orc(const std::string& timezone,
const IColumn& column,
const NullMap* null_map,
orc::ColumnVectorBatch*
orc_col_batch,
diff --git a/be/src/core/data_type_serde/data_type_varbinary_serde.h
b/be/src/core/data_type_serde/data_type_varbinary_serde.h
index 6d23159c8ea..b4f5f26eab5 100644
--- a/be/src/core/data_type_serde/data_type_varbinary_serde.h
+++ b/be/src/core/data_type_serde/data_type_varbinary_serde.h
@@ -76,10 +76,7 @@ public:
const cctz::time_zone& ctz) const override;
Status read_column_from_arrow(IColumn& column, const arrow::Array*
arrow_array, int64_t start,
- int64_t end, const cctz::time_zone& ctz)
const override {
- return Status::Error(ErrorCode::NOT_IMPLEMENTED_ERROR,
- "read_column_from_arrow with type " +
column.get_name());
- }
+ int64_t end, const cctz::time_zone& ctz)
const override;
Status read_column_from_parquet(IColumn& column, ParquetDecodeSource&
source,
const ParquetDecodeContext& context,
size_t num_values,
diff --git a/be/test/core/data_type_serde/data_type_serde_array_test.cpp
b/be/test/core/data_type_serde/data_type_serde_array_test.cpp
index f6b4a666f96..3478f8292d4 100644
--- a/be/test/core/data_type_serde/data_type_serde_array_test.cpp
+++ b/be/test/core/data_type_serde/data_type_serde_array_test.cpp
@@ -15,12 +15,21 @@
// specific language governing permissions and limitations
// under the License.
+#include <arrow/api.h>
+#include <cctz/time_zone.h>
#include <gtest/gtest.h>
+#include <array>
#include <cstring>
+#include <memory>
+#include "core/assert_cast.h"
+#include "core/column/column_array.h"
+#include "core/column/column_nullable.h"
#include "core/column/column_string.h"
#include "core/column/column_vector.h"
+#include "core/data_type_serde/data_type_array_serde.h"
+#include "core/data_type_serde/data_type_nullable_serde.h"
#include "core/data_type_serde/data_type_number_serde.h"
#include "core/data_type_serde/data_type_serde.h"
#include "core/data_type_serde/data_type_string_serde.h"
@@ -175,4 +184,113 @@ TEST_F(DataTypeArraySerDeFieldTest,
empty_array_is_null_element_type) {
EXPECT_FALSE(info.need_convert);
}
+TEST_F(DataTypeArraySerDeFieldTest, ReadArrowFixedSizeAndLargeList) {
+ auto nested_serde = std::make_shared<DataTypeNullableSerDe>(
+ std::make_shared<DataTypeNumberSerDe<TYPE_FLOAT>>());
+ DataTypeArraySerDe serde(nested_serde);
+ cctz::time_zone tz;
+
+ const auto expect_array = [](const ColumnArray& column,
+ const std::vector<ColumnArray::Offset64>&
expected_offsets,
+ const std::vector<float>& expected_values) {
+ const auto& offsets = column.get_offsets();
+ ASSERT_EQ(expected_offsets.size(), offsets.size());
+ for (size_t i = 0; i < expected_offsets.size(); ++i) {
+ EXPECT_EQ(expected_offsets[i], offsets[i]);
+ }
+ const auto& nullable_values = assert_cast<const
ColumnNullable&>(column.get_data());
+ const auto& values =
+ assert_cast<const
ColumnFloat32&>(nullable_values.get_nested_column()).get_data();
+ ASSERT_EQ(expected_values.size(), values.size());
+ for (size_t i = 0; i < expected_values.size(); ++i) {
+ EXPECT_EQ(0, nullable_values.get_null_map_data()[i]);
+ EXPECT_FLOAT_EQ(expected_values[i], values[i]);
+ }
+ };
+
+ // Lance vectors are Arrow FixedSizeList: every embedding has exactly
three floats.
+ {
+ auto value_builder = std::make_shared<arrow::FloatBuilder>();
+ arrow::FixedSizeListBuilder builder(arrow::default_memory_pool(),
value_builder, 3);
+ for (const std::array<float, 3>& embedding :
+ {std::array<float, 3> {0.0F, 0.0F, 0.0F}, std::array<float, 3>
{1.0F, 0.0F, 0.0F},
+ std::array<float, 3> {-1.5F, 0.25F, 3.75F}}) {
+ ASSERT_TRUE(builder.Append().ok());
+ for (float value : embedding) {
+ ASSERT_TRUE(value_builder->Append(value).ok());
+ }
+ }
+ std::shared_ptr<arrow::FixedSizeListArray> arrow_array;
+ ASSERT_TRUE(builder.Finish(&arrow_array).ok());
+
+ auto column = ColumnArray::create(
+ ColumnNullable::create(ColumnFloat32::create(),
ColumnUInt8::create()),
+ ColumnOffset64::create());
+ ASSERT_TRUE(serde.read_column_from_arrow(*column, arrow_array.get(), 0,
+ arrow_array->length(), tz)
+ .ok());
+ expect_array(*column, {3, 6, 9}, {0.0F, 0.0F, 0.0F, 1.0F, 0.0F, 0.0F,
-1.5F, 0.25F, 3.75F});
+ }
+
+ // LargeList uses 64-bit Arrow offsets but has the same Doris ARRAY
representation.
+ {
+ auto value_builder = std::make_shared<arrow::FloatBuilder>();
+ arrow::LargeListBuilder builder(arrow::default_memory_pool(),
value_builder);
+ ASSERT_TRUE(builder.Append().ok());
+ ASSERT_TRUE(value_builder->Append(1.0F).ok());
+ ASSERT_TRUE(value_builder->Append(2.0F).ok());
+ ASSERT_TRUE(builder.Append().ok());
+ ASSERT_TRUE(value_builder->Append(3.0F).ok());
+ ASSERT_TRUE(builder.Append().ok());
+ std::shared_ptr<arrow::LargeListArray> arrow_array;
+ ASSERT_TRUE(builder.Finish(&arrow_array).ok());
+
+ auto column = ColumnArray::create(
+ ColumnNullable::create(ColumnFloat32::create(),
ColumnUInt8::create()),
+ ColumnOffset64::create());
+ ASSERT_TRUE(serde.read_column_from_arrow(*column, arrow_array.get(), 0,
+ arrow_array->length(), tz)
+ .ok());
+ expect_array(*column, {2, 3, 3}, {1.0F, 2.0F, 3.0F});
+ }
+}
+
+// External Arrow producers may expose offsets through Buffer::Wrap without
preserving the natural
+// alignment of int32_t or int64_t. Run with UBSan enabled to catch typed
loads from such buffers.
+TEST_F(DataTypeArraySerDeFieldTest, ReadArrowListWithUnalignedOffsets) {
+ auto nested_serde = std::make_shared<DataTypeNullableSerDe>(
+ std::make_shared<DataTypeNumberSerDe<TYPE_FLOAT>>());
+ DataTypeArraySerDe serde(nested_serde);
+
+ std::vector<float> values_data {1.0F, 2.0F, 3.0F};
+ const auto values_buffer = arrow::Buffer::Wrap(values_data);
+ const auto values = std::make_shared<arrow::FloatArray>(3, values_buffer);
+ const auto read_array = [&](const auto& arrow_array) {
+ auto column = ColumnArray::create(
+ ColumnNullable::create(ColumnFloat32::create(),
ColumnUInt8::create()),
+ ColumnOffset64::create());
+ EXPECT_TRUE(serde.read_column_from_arrow(*column, arrow_array.get(), 0,
+ arrow_array->length(),
cctz::utc_time_zone())
+ .ok());
+ EXPECT_EQ(column->get_offsets(), (ColumnArray::Offsets64 {2, 3, 3}));
+ };
+
+ const std::array<int32_t, 4> list_offsets {0, 2, 3, 3};
+ std::vector<uint8_t> list_offsets_storage(sizeof(list_offsets) + 1);
+ memcpy(list_offsets_storage.data() + 1, list_offsets.data(),
sizeof(list_offsets));
+ const auto list_offsets_buffer =
+ arrow::Buffer::Wrap(list_offsets_storage.data() + 1,
sizeof(list_offsets));
+
read_array(std::make_shared<arrow::ListArray>(arrow::list(arrow::float32()), 3,
+ list_offsets_buffer,
values));
+
+ const std::array<int64_t, 4> large_list_offsets {0, 2, 3, 3};
+ std::vector<uint8_t> large_list_offsets_storage(sizeof(large_list_offsets)
+ 1);
+ memcpy(large_list_offsets_storage.data() + 1, large_list_offsets.data(),
+ sizeof(large_list_offsets));
+ const auto large_list_offsets_buffer =
+ arrow::Buffer::Wrap(large_list_offsets_storage.data() + 1,
sizeof(large_list_offsets));
+
read_array(std::make_shared<arrow::LargeListArray>(arrow::large_list(arrow::float32()),
3,
+
large_list_offsets_buffer, values));
+}
+
} // namespace doris
diff --git
a/be/test/core/data_type_serde/data_type_serde_arrow_validation_test.cpp
b/be/test/core/data_type_serde/data_type_serde_arrow_validation_test.cpp
index fdce7af8f89..d6b2e4f57bd 100644
--- a/be/test/core/data_type_serde/data_type_serde_arrow_validation_test.cpp
+++ b/be/test/core/data_type_serde/data_type_serde_arrow_validation_test.cpp
@@ -34,12 +34,17 @@
#include "core/column/column_map.h"
#include "core/column/column_nullable.h"
#include "core/column/column_string.h"
+#include "core/column/column_varbinary.h"
#include "core/data_type/primitive_type.h"
#include "core/data_type_serde/data_type_array_serde.h"
+#include "core/data_type_serde/data_type_datev2_serde.h"
#include "core/data_type_serde/data_type_map_serde.h"
#include "core/data_type_serde/data_type_nullable_serde.h"
#include "core/data_type_serde/data_type_number_serde.h"
#include "core/data_type_serde/data_type_string_serde.h"
+#include "core/data_type_serde/data_type_time_serde.h"
+#include "core/data_type_serde/data_type_timestamptz_serde.h"
+#include "core/data_type_serde/data_type_varbinary_serde.h"
namespace doris {
namespace {
@@ -69,7 +74,8 @@ void expect_invalid_arrow(Func&& func, std::string_view
message) {
EXPECT_TRUE(thrown) << message;
}
-std::shared_ptr<arrow::Buffer> wrap_offsets(const std::vector<int32_t>&
offsets) {
+template <typename OffsetType>
+std::shared_ptr<arrow::Buffer> wrap_offsets(const std::vector<OffsetType>&
offsets) {
return arrow::Buffer::Wrap(offsets);
}
@@ -86,6 +92,127 @@ struct StringArrayHolder {
std::shared_ptr<arrow::StringArray> array;
};
+struct LargeListArrayHolder {
+ LargeListArrayHolder(int64_t length, std::vector<int64_t> offsets_,
+ std::vector<int64_t> values_)
+ : offsets(std::move(offsets_)), values(std::move(values_)) {
+ auto values_array =
+ std::make_shared<arrow::Int64Array>(values.size(),
arrow::Buffer::Wrap(values));
+ array =
std::make_shared<arrow::LargeListArray>(arrow::large_list(arrow::int64()),
length,
+ wrap_offsets(offsets),
values_array);
+ }
+
+ std::vector<int64_t> offsets;
+ std::vector<int64_t> values;
+ std::shared_ptr<arrow::LargeListArray> array;
+};
+
+void expect_invalid_large_list_offsets(int64_t length, std::vector<int64_t>
offsets,
+ std::vector<int64_t> values,
std::string_view message) {
+ LargeListArrayHolder array(length, std::move(offsets), std::move(values));
+ auto column = ColumnArray::create(
+ ColumnNullable::create(ColumnInt64::create(),
ColumnUInt8::create()),
+ ColumnOffset64::create());
+ auto nested_serde = std::make_shared<DataTypeNullableSerDe>(
+ std::make_shared<DataTypeNumberSerDe<TYPE_BIGINT>>());
+ DataTypeArraySerDe serde(nested_serde);
+
+ expect_invalid_arrow(
+ [&] {
+ static_cast<void>(serde.read_column_from_arrow(*column,
array.array.get(), 0,
+ length,
cctz::utc_time_zone()));
+ },
+ message);
+}
+
+template <typename ArrowArrayType, PrimitiveType DorisType, typename
DorisColumnType>
+void expect_invalid_short_numeric_buffer(std::string_view message) {
+ using ArrowValueType = typename ArrowArrayType::value_type;
+ std::vector<ArrowValueType> values = {1};
+ auto array = std::make_shared<ArrowArrayType>(2,
arrow::Buffer::Wrap(values));
+ auto column = DorisColumnType::create();
+ DataTypeNumberSerDe<DorisType> serde;
+
+ expect_invalid_arrow(
+ [&] {
+ static_cast<void>(serde.read_column_from_arrow(*column,
array.get(), 0, 2,
+
cctz::utc_time_zone()));
+ },
+ message);
+}
+
+template <typename ArrowArrayType, PrimitiveType DorisType, typename
DorisColumnType>
+void expect_invalid_missing_numeric_validity_bitmap(std::string_view message) {
+ using ArrowValueType = typename ArrowArrayType::value_type;
+ std::vector<ArrowValueType> values = {1, 2};
+ auto array = std::make_shared<ArrowArrayType>(2,
arrow::Buffer::Wrap(values),
+
std::shared_ptr<arrow::Buffer>(), 1);
+ auto column = DorisColumnType::create();
+ DataTypeNumberSerDe<DorisType> serde;
+
+ expect_invalid_arrow(
+ [&] {
+ static_cast<void>(serde.read_column_from_arrow(*column,
array.get(), 0, 2,
+
cctz::utc_time_zone()));
+ },
+ message);
+}
+
+template <typename DorisColumnType, typename SerDeType>
+void expect_invalid_temporal_arrow_buffers(const
std::shared_ptr<arrow::DataType>& arrow_type,
+ size_t value_width, SerDeType&
serde,
+ std::string_view type_name) {
+ std::vector<uint8_t> short_values(value_width, 0);
+ auto short_data = arrow::ArrayData::Make(
+ arrow_type, 2, {std::shared_ptr<arrow::Buffer>(),
arrow::Buffer::Wrap(short_values)},
+ 0);
+ auto short_array = arrow::MakeArray(short_data);
+ auto short_column = DorisColumnType::create();
+ expect_invalid_arrow(
+ [&] {
+ static_cast<void>(serde.read_column_from_arrow(*short_column,
short_array.get(), 0,
+ 2,
cctz::utc_time_zone()));
+ },
+ std::string(type_name) + " short values buffer should be
rejected");
+
+ std::vector<uint8_t> values(value_width * 2, 0);
+ auto missing_validity_data = arrow::ArrayData::Make(
+ arrow_type, 2, {std::shared_ptr<arrow::Buffer>(),
arrow::Buffer::Wrap(values)}, 1);
+ auto missing_validity_array = arrow::MakeArray(missing_validity_data);
+ auto missing_validity_column = DorisColumnType::create();
+ expect_invalid_arrow(
+ [&] {
+
static_cast<void>(serde.read_column_from_arrow(*missing_validity_column,
+
missing_validity_array.get(), 0, 2,
+
cctz::utc_time_zone()));
+ },
+ std::string(type_name) + " missing validity bitmap should be
rejected");
+
+ auto invalid_range_data = arrow::ArrayData::Make(
+ arrow_type, 1, {std::shared_ptr<arrow::Buffer>(),
arrow::Buffer::Wrap(values)}, 0);
+ auto invalid_range_array = arrow::MakeArray(invalid_range_data);
+ auto invalid_range_column = DorisColumnType::create();
+ expect_invalid_arrow(
+ [&] {
+
static_cast<void>(serde.read_column_from_arrow(*invalid_range_column,
+
invalid_range_array.get(), 0, 2,
+
cctz::utc_time_zone()));
+ },
+ std::string(type_name) + " invalid read range should be rejected");
+}
+
+void expect_invalid_varbinary_arrow(const arrow::Array& array, int64_t start,
int64_t end,
+ std::string_view message) {
+ auto column = ColumnVarbinary::create();
+ DataTypeVarbinarySerDe serde;
+ expect_invalid_arrow(
+ [&] {
+ static_cast<void>(serde.read_column_from_arrow(*column,
&array, start, end,
+
cctz::utc_time_zone()));
+ },
+ message);
+}
+
} // namespace
TEST(DataTypeSerDeArrowValidationTest, RejectsShortStringOffsetsBuffer) {
@@ -153,6 +280,108 @@ TEST(DataTypeSerDeArrowValidationTest,
RejectsShortFixedWidthDataBuffer) {
"short int64 data buffer should be rejected");
}
+TEST(DataTypeSerDeArrowValidationTest,
RejectsShortEarlyReturnNumericDataBuffers) {
+ ScopedArrowInputValidation validation(true);
+
+ expect_invalid_short_numeric_buffer<arrow::HalfFloatArray, TYPE_FLOAT,
ColumnFloat32>(
+ "short half-float data buffer should be rejected");
+ expect_invalid_short_numeric_buffer<arrow::UInt8Array, TYPE_SMALLINT,
ColumnInt16>(
+ "short uint8 data buffer should be rejected");
+ expect_invalid_short_numeric_buffer<arrow::UInt16Array, TYPE_INT,
ColumnInt32>(
+ "short uint16 data buffer should be rejected");
+ expect_invalid_short_numeric_buffer<arrow::UInt32Array, TYPE_BIGINT,
ColumnInt64>(
+ "short uint32 data buffer should be rejected");
+ expect_invalid_short_numeric_buffer<arrow::UInt64Array, TYPE_LARGEINT,
ColumnInt128>(
+ "short uint64 data buffer should be rejected");
+}
+
+TEST(DataTypeSerDeArrowValidationTest,
RejectsMissingEarlyReturnNumericValidityBitmaps) {
+ ScopedArrowInputValidation validation(true);
+
+ expect_invalid_missing_numeric_validity_bitmap<arrow::HalfFloatArray,
TYPE_FLOAT,
+ ColumnFloat32>(
+ "missing half-float validity bitmap should be rejected");
+ expect_invalid_missing_numeric_validity_bitmap<arrow::UInt8Array,
TYPE_SMALLINT, ColumnInt16>(
+ "missing unsigned widening validity bitmap should be rejected");
+}
+
+TEST(DataTypeSerDeArrowValidationTest, RejectsMalformedNewTemporalArrays) {
+ ScopedArrowInputValidation validation(true);
+
+ DataTypeTimeV2SerDe time_serde;
+
expect_invalid_temporal_arrow_buffers<ColumnTimeV2>(arrow::time32(arrow::TimeUnit::SECOND),
+
sizeof(arrow::Time32Array::value_type),
+ time_serde, "time32");
+
expect_invalid_temporal_arrow_buffers<ColumnTimeV2>(arrow::time64(arrow::TimeUnit::MICRO),
+
sizeof(arrow::Time64Array::value_type),
+ time_serde, "time64");
+
+ DataTypeTimeStampTzSerDe timestamptz_serde(6);
+ expect_invalid_temporal_arrow_buffers<ColumnTimeStampTz>(
+ arrow::timestamp(arrow::TimeUnit::MICRO),
sizeof(arrow::TimestampArray::value_type),
+ timestamptz_serde, "timestamp");
+
+ DataTypeDateV2SerDe date_serde;
+ expect_invalid_temporal_arrow_buffers<ColumnDateV2>(
+ arrow::date64(), sizeof(arrow::Date64Array::value_type),
date_serde, "date64");
+}
+
+TEST(DataTypeSerDeArrowValidationTest, RejectsMalformedVarbinaryArrays) {
+ ScopedArrowInputValidation validation(true);
+
+ std::string values = "abc";
+ const auto value_buffer = arrow::Buffer::Wrap(values.data(),
values.size());
+
+ std::vector<int32_t> short_offsets = {0};
+ arrow::BinaryArray short_binary_offsets(1, wrap_offsets(short_offsets),
value_buffer);
+ expect_invalid_varbinary_arrow(short_binary_offsets, 0, 1,
+ "short binary offsets buffer should be
rejected");
+
+ std::vector<int32_t> negative_offsets = {-1, 0};
+ arrow::BinaryArray negative_binary_offsets(1,
wrap_offsets(negative_offsets), value_buffer);
+ expect_invalid_varbinary_arrow(negative_binary_offsets, 0, 1,
+ "negative binary offset should be
rejected");
+
+ std::vector<int32_t> non_monotonic_offsets = {2, 1};
+ arrow::BinaryArray non_monotonic_binary_offsets(1,
wrap_offsets(non_monotonic_offsets),
+ value_buffer);
+ expect_invalid_varbinary_arrow(non_monotonic_binary_offsets, 0, 1,
+ "non-monotonic binary offsets should be
rejected");
+
+ std::vector<int32_t> oversized_offsets = {0, 4};
+ arrow::BinaryArray oversized_binary_value(1,
wrap_offsets(oversized_offsets), value_buffer);
+ expect_invalid_varbinary_arrow(oversized_binary_value, 0, 1,
+ "binary value range beyond buffer should be
rejected");
+
+ std::vector<int64_t> short_large_offsets = {0};
+ arrow::LargeBinaryArray short_large_binary_offsets(1,
wrap_offsets(short_large_offsets),
+ value_buffer);
+ expect_invalid_varbinary_arrow(short_large_binary_offsets, 0, 1,
+ "short large-binary offsets buffer should
be rejected");
+
+ std::vector<int64_t> oversized_large_offsets = {0, 4};
+ arrow::LargeBinaryArray oversized_large_binary_value(1,
wrap_offsets(oversized_large_offsets),
+ value_buffer);
+ expect_invalid_varbinary_arrow(oversized_large_binary_value, 0, 1,
+ "large-binary value range beyond buffer
should be rejected");
+
+ std::vector<uint8_t> short_fixed_values(3, 0);
+ arrow::FixedSizeBinaryArray
short_fixed_binary(arrow::fixed_size_binary(3), 2,
+
arrow::Buffer::Wrap(short_fixed_values));
+ expect_invalid_varbinary_arrow(short_fixed_binary, 0, 2,
+ "short fixed-size binary buffer should be
rejected");
+
+ std::vector<int32_t> valid_offsets = {0, 1};
+ arrow::BinaryArray missing_validity(1, wrap_offsets(valid_offsets),
value_buffer,
+ std::shared_ptr<arrow::Buffer>(), 1);
+ expect_invalid_varbinary_arrow(missing_validity, 0, 1,
+ "missing varbinary validity bitmap should
be rejected");
+
+ arrow::BinaryArray valid_binary(1, wrap_offsets(valid_offsets),
value_buffer);
+ expect_invalid_varbinary_arrow(valid_binary, 0, 2,
+ "invalid varbinary read range should be
rejected");
+}
+
TEST(DataTypeSerDeArrowValidationTest, RejectsSlicedArrowArray) {
ScopedArrowInputValidation validation(true);
@@ -249,6 +478,19 @@ TEST(DataTypeSerDeArrowValidationTest,
RejectsListOffsetsBeyondValuesLength) {
"list offsets beyond values length should be rejected");
}
+TEST(DataTypeSerDeArrowValidationTest, RejectsMalformedLargeListOffsets) {
+ ScopedArrowInputValidation validation(true);
+
+ expect_invalid_large_list_offsets(1, {0}, {1},
+ "short large-list offsets buffer should
be rejected");
+ expect_invalid_large_list_offsets(1, {-1, 0}, {1},
+ "negative large-list offset should be
rejected");
+ expect_invalid_large_list_offsets(2, {0, 2, 1}, {1, 2},
+ "non-monotonic large-list offsets should
be rejected");
+ expect_invalid_large_list_offsets(1, {0, 2}, {1},
+ "large-list offsets beyond values length
should be rejected");
+}
+
TEST(DataTypeSerDeArrowValidationTest, RejectsMapOffsetsBeyondKeysLength) {
ScopedArrowInputValidation validation(true);
diff --git a/be/test/core/data_type_serde/data_type_serde_datetime_v2_test.cpp
b/be/test/core/data_type_serde/data_type_serde_datetime_v2_test.cpp
index dbe9c875834..88b6dcba6ae 100644
--- a/be/test/core/data_type_serde/data_type_serde_datetime_v2_test.cpp
+++ b/be/test/core/data_type_serde/data_type_serde_datetime_v2_test.cpp
@@ -333,6 +333,122 @@ TEST_F(DataTypeDateTimeV2SerDeTest,
ArrowMemNotAlignedDate) {
EXPECT_TRUE(st.ok());
}
+TEST_F(DataTypeDateTimeV2SerDeTest, ArrowDate64ToDateV2) {
+ arrow::Date64Builder builder;
+ ASSERT_TRUE(builder.Append(-86400000).ok());
+ ASSERT_TRUE(builder.Append(0).ok());
+ ASSERT_TRUE(builder.Append(86400000).ok());
+ ASSERT_TRUE(builder.Append(1785196800000).ok());
+ ASSERT_TRUE(builder.AppendNull().ok());
+ std::shared_ptr<arrow::Array> array;
+ ASSERT_TRUE(builder.Finish(&array).ok());
+
+ auto column = ColumnDateV2::create();
+ cctz::time_zone tz;
+ auto status =
+ serde_date_v2->read_column_from_arrow(*column, array.get(), 0,
array->length(), tz);
+ ASSERT_TRUE(status.ok()) << status.to_string();
+
+ ASSERT_EQ(5, column->size());
+ const auto& values = column->get_data();
+ EXPECT_EQ(1969, values[0].year());
+ EXPECT_EQ(12, values[0].month());
+ EXPECT_EQ(31, values[0].day());
+ EXPECT_EQ(1970, values[1].year());
+ EXPECT_EQ(1, values[1].month());
+ EXPECT_EQ(1, values[1].day());
+ EXPECT_EQ(1970, values[2].year());
+ EXPECT_EQ(1, values[2].month());
+ EXPECT_EQ(2, values[2].day());
+ EXPECT_EQ(2026, values[3].year());
+ EXPECT_EQ(7, values[3].month());
+ EXPECT_EQ(28, values[3].day());
+}
+
+TEST_F(DataTypeDateTimeV2SerDeTest, RejectsInvalidDate64Values) {
+ const auto expect_invalid = [&](int64_t milliseconds, const char*
expected_message) {
+ arrow::Date64Builder builder;
+ ASSERT_TRUE(builder.Append(milliseconds).ok());
+ std::shared_ptr<arrow::Array> array;
+ ASSERT_TRUE(builder.Finish(&array).ok());
+
+ auto column = ColumnDateV2::create();
+ cctz::time_zone tz;
+ const auto status = serde_date_v2->read_column_from_arrow(*column,
array.get(), 0, 1, tz);
+ EXPECT_FALSE(status.ok());
+ EXPECT_NE(std::string::npos,
status.to_string().find(expected_message));
+ };
+
+ expect_invalid(1, "must contain whole days");
+ constexpr int64_t milliseconds_per_day = 86400000;
+ constexpr int64_t out_of_range =
+ std::numeric_limits<int64_t>::max() / milliseconds_per_day *
milliseconds_per_day;
+ expect_invalid(out_of_range, "outside the Doris DATE range");
+}
+
+TEST_F(DataTypeDateTimeV2SerDeTest, ArrowTimeToTimeV2) {
+ const auto read_time = [](arrow::ArrayBuilder* builder, int scale,
+ const std::vector<double>& expected) {
+ std::shared_ptr<arrow::Array> array;
+ ASSERT_TRUE(builder->Finish(&array).ok());
+ auto column = ColumnTimeV2::create();
+ DataTypeTimeV2SerDe serde(scale);
+ cctz::time_zone tz;
+ const auto status =
+ serde.read_column_from_arrow(*column, array.get(), 0,
array->length(), tz);
+ ASSERT_TRUE(status.ok()) << status.to_string();
+ ASSERT_EQ(expected.size(), column->size());
+ for (size_t row = 0; row < expected.size(); ++row) {
+ EXPECT_DOUBLE_EQ(expected[row], column->get_data()[row]);
+ }
+ };
+
+ arrow::Time32Builder
seconds_builder(arrow::time32(arrow::TimeUnit::SECOND),
+ arrow::default_memory_pool());
+ ASSERT_TRUE(seconds_builder.Append(45296).ok());
+ ASSERT_TRUE(seconds_builder.AppendNull().ok());
+ read_time(&seconds_builder, 0, {45296000000.0, 0.0});
+
+ arrow::Time32Builder millis_builder(arrow::time32(arrow::TimeUnit::MILLI),
+ arrow::default_memory_pool());
+ ASSERT_TRUE(millis_builder.Append(45296123).ok());
+ read_time(&millis_builder, 3, {45296123000.0});
+
+ arrow::Time64Builder micros_builder(arrow::time64(arrow::TimeUnit::MICRO),
+ arrow::default_memory_pool());
+ ASSERT_TRUE(micros_builder.Append(45296123456).ok());
+ read_time(µs_builder, 6, {45296123456.0});
+
+ arrow::Time64Builder nanos_builder(arrow::time64(arrow::TimeUnit::NANO),
+ arrow::default_memory_pool());
+ ASSERT_TRUE(nanos_builder.Append(45296123456789).ok());
+ read_time(&nanos_builder, 6, {45296123456.0});
+}
+
+TEST_F(DataTypeDateTimeV2SerDeTest, RejectsInvalidArrowTimeValues) {
+ const auto expect_invalid = [](arrow::ArrayBuilder* builder, int scale) {
+ std::shared_ptr<arrow::Array> array;
+ ASSERT_TRUE(builder->Finish(&array).ok());
+ auto column = ColumnTimeV2::create();
+ DataTypeTimeV2SerDe serde(scale);
+ cctz::time_zone tz;
+ const auto status =
+ serde.read_column_from_arrow(*column, array.get(), 0,
array->length(), tz);
+ EXPECT_FALSE(status.ok());
+ EXPECT_NE(std::string::npos, status.to_string().find("outside the
time-of-day range"));
+ };
+
+ arrow::Time32Builder
negative_builder(arrow::time32(arrow::TimeUnit::SECOND),
+ arrow::default_memory_pool());
+ ASSERT_TRUE(negative_builder.Append(-1).ok());
+ expect_invalid(&negative_builder, 0);
+
+ arrow::Time64Builder
next_day_builder(arrow::time64(arrow::TimeUnit::MICRO),
+ arrow::default_memory_pool());
+ ASSERT_TRUE(next_day_builder.Append(86400000000).ok());
+ expect_invalid(&next_day_builder, 6);
+}
+
// Run with UBSan enabled to catch misalignment errors.
TEST_F(DataTypeDateTimeV2SerDeTest, ArrowMemNotAlignedDateTime) {
// 1.Prepare the data.
@@ -395,4 +511,25 @@ TEST_F(DataTypeDateTimeV2SerDeTest,
ReadArrowTimestampBeforeEpoch) {
EXPECT_EQ(insert_value, dest_column->get_element(0).to_string(6));
}
+TEST_F(DataTypeDateTimeV2SerDeTest,
ArrowTimezoneNaiveTimestampIgnoresSessionTimezone) {
+ auto timestamp_type = arrow::timestamp(arrow::TimeUnit::MICRO);
+ arrow::TimestampBuilder builder(timestamp_type,
arrow::default_memory_pool());
+ ASSERT_TRUE(builder.Append(0).ok());
+ ASSERT_TRUE(builder.Append(45296123456).ok());
+ ASSERT_TRUE(builder.Append(-876544).ok());
+ std::shared_ptr<arrow::Array> array;
+ ASSERT_TRUE(builder.Finish(&array).ok());
+
+ auto column = ColumnDateTimeV2::create();
+ cctz::time_zone session_timezone;
+ ASSERT_TRUE(TimezoneUtils::find_cctz_time_zone("+08:00",
session_timezone));
+ const auto status = serde_datetime_v2_6->read_column_from_arrow(
+ *column, array.get(), 0, array->length(), session_timezone);
+ ASSERT_TRUE(status.ok()) << status.to_string();
+ ASSERT_EQ(3, column->size());
+ EXPECT_EQ("1970-01-01 00:00:00.000000",
column->get_data()[0].to_string(6));
+ EXPECT_EQ("1970-01-01 12:34:56.123456",
column->get_data()[1].to_string(6));
+ EXPECT_EQ("1969-12-31 23:59:59.123456",
column->get_data()[2].to_string(6));
+}
+
} // namespace doris
diff --git a/be/test/core/data_type_serde/data_type_serde_number_test.cpp
b/be/test/core/data_type_serde/data_type_serde_number_test.cpp
index b1050d7119b..41724fd3b33 100644
--- a/be/test/core/data_type_serde/data_type_serde_number_test.cpp
+++ b/be/test/core/data_type_serde/data_type_serde_number_test.cpp
@@ -620,6 +620,130 @@ TEST_F(DataTypeNumberSerDeTest,
RowStoreDateTimeJsonbWidth) {
});
}
+TEST_F(DataTypeNumberSerDeTest, ArrowFloat16ToFloat32) {
+ arrow::HalfFloatBuilder builder;
+ ASSERT_TRUE(builder.AppendNull().ok());
+ ASSERT_TRUE(builder.Append(0x0000).ok());
+ ASSERT_TRUE(builder.Append(0x8000).ok());
+ ASSERT_TRUE(builder.Append(0x3E00).ok());
+ ASSERT_TRUE(builder.Append(0x7BFF).ok());
+ ASSERT_TRUE(builder.Append(0x7C00).ok());
+ ASSERT_TRUE(builder.Append(0xFC00).ok());
+ ASSERT_TRUE(builder.Append(0x7E00).ok());
+ std::shared_ptr<arrow::Array> array;
+ ASSERT_TRUE(builder.Finish(&array).ok());
+
+ auto float_column = ColumnFloat32::create();
+ cctz::time_zone tz;
+ ASSERT_TRUE(serde_float32
+ ->read_column_from_arrow(*float_column, array.get(),
0, array->length(), tz)
+ .ok());
+
+ ASSERT_EQ(8, float_column->size());
+ EXPECT_FLOAT_EQ(0.0F, float_column->get_data()[0]);
+ EXPECT_FLOAT_EQ(0.0F, float_column->get_data()[1]);
+ EXPECT_FALSE(std::signbit(float_column->get_data()[1]));
+ EXPECT_FLOAT_EQ(-0.0F, float_column->get_data()[2]);
+ EXPECT_TRUE(std::signbit(float_column->get_data()[2]));
+ EXPECT_FLOAT_EQ(1.5F, float_column->get_data()[3]);
+ EXPECT_FLOAT_EQ(65504.0F, float_column->get_data()[4]);
+ EXPECT_EQ(std::numeric_limits<float>::infinity(),
float_column->get_data()[5]);
+ EXPECT_EQ(-std::numeric_limits<float>::infinity(),
float_column->get_data()[6]);
+ EXPECT_TRUE(std::isnan(float_column->get_data()[7]));
+}
+
+TEST_F(DataTypeNumberSerDeTest, ArrowUnsignedIntegersAreWidenedLosslessly) {
+ cctz::time_zone tz;
+
+ {
+ arrow::UInt8Builder builder;
+ ASSERT_TRUE(builder.AppendNull().ok());
+ ASSERT_TRUE(builder.Append(0).ok());
+ ASSERT_TRUE(builder.Append(127).ok());
+ ASSERT_TRUE(builder.Append(128).ok());
+ ASSERT_TRUE(builder.Append(std::numeric_limits<UInt8>::max()).ok());
+ std::shared_ptr<arrow::Array> array;
+ ASSERT_TRUE(builder.Finish(&array).ok());
+
+ auto column = ColumnInt16::create();
+ ASSERT_TRUE(
+ serde_int16->read_column_from_arrow(*column, array.get(), 0,
array->length(), tz)
+ .ok());
+ ASSERT_EQ(5, column->size());
+ EXPECT_EQ(0, column->get_data()[0]);
+ EXPECT_EQ(0, column->get_data()[1]);
+ EXPECT_EQ(127, column->get_data()[2]);
+ EXPECT_EQ(128, column->get_data()[3]);
+ EXPECT_EQ(255, column->get_data()[4]);
+ }
+
+ {
+ arrow::UInt16Builder builder;
+ ASSERT_TRUE(builder.AppendNull().ok());
+ ASSERT_TRUE(builder.Append(0).ok());
+ ASSERT_TRUE(builder.Append(32767).ok());
+ ASSERT_TRUE(builder.Append(32768).ok());
+ ASSERT_TRUE(builder.Append(std::numeric_limits<UInt16>::max()).ok());
+ std::shared_ptr<arrow::Array> array;
+ ASSERT_TRUE(builder.Finish(&array).ok());
+
+ auto column = ColumnInt32::create();
+ ASSERT_TRUE(
+ serde_int32->read_column_from_arrow(*column, array.get(), 0,
array->length(), tz)
+ .ok());
+ ASSERT_EQ(5, column->size());
+ EXPECT_EQ(0, column->get_data()[0]);
+ EXPECT_EQ(0, column->get_data()[1]);
+ EXPECT_EQ(32767, column->get_data()[2]);
+ EXPECT_EQ(32768, column->get_data()[3]);
+ EXPECT_EQ(65535, column->get_data()[4]);
+ }
+
+ {
+ arrow::UInt32Builder builder;
+ ASSERT_TRUE(builder.AppendNull().ok());
+ ASSERT_TRUE(builder.Append(0).ok());
+ ASSERT_TRUE(builder.Append(std::numeric_limits<Int32>::max()).ok());
+ ASSERT_TRUE(builder.Append(UInt32(1) << 31).ok());
+ ASSERT_TRUE(builder.Append(std::numeric_limits<UInt32>::max()).ok());
+ std::shared_ptr<arrow::Array> array;
+ ASSERT_TRUE(builder.Finish(&array).ok());
+
+ auto column = ColumnInt64::create();
+ ASSERT_TRUE(
+ serde_int64->read_column_from_arrow(*column, array.get(), 0,
array->length(), tz)
+ .ok());
+ ASSERT_EQ(5, column->size());
+ EXPECT_EQ(0, column->get_data()[0]);
+ EXPECT_EQ(0, column->get_data()[1]);
+ EXPECT_EQ(std::numeric_limits<Int32>::max(), column->get_data()[2]);
+ EXPECT_EQ(Int64(1) << 31, column->get_data()[3]);
+ EXPECT_EQ(std::numeric_limits<UInt32>::max(), column->get_data()[4]);
+ }
+
+ {
+ arrow::UInt64Builder builder;
+ ASSERT_TRUE(builder.AppendNull().ok());
+ ASSERT_TRUE(builder.Append(0).ok());
+ ASSERT_TRUE(builder.Append(std::numeric_limits<Int64>::max()).ok());
+ ASSERT_TRUE(builder.Append(UInt64(1) << 63).ok());
+ ASSERT_TRUE(builder.Append(std::numeric_limits<UInt64>::max()).ok());
+ std::shared_ptr<arrow::Array> array;
+ ASSERT_TRUE(builder.Finish(&array).ok());
+
+ auto column = ColumnInt128::create();
+ ASSERT_TRUE(
+ serde_int128->read_column_from_arrow(*column, array.get(), 0,
array->length(), tz)
+ .ok());
+ ASSERT_EQ(5, column->size());
+ EXPECT_EQ(Int128(0), column->get_data()[0]);
+ EXPECT_EQ(Int128(0), column->get_data()[1]);
+ EXPECT_EQ(static_cast<Int128>(std::numeric_limits<Int64>::max()),
column->get_data()[2]);
+ EXPECT_EQ(Int128(1) << 63, column->get_data()[3]);
+ EXPECT_EQ(static_cast<Int128>(std::numeric_limits<UInt64>::max()),
column->get_data()[4]);
+ }
+}
+
// to_olap_string / from_zonemap_string must round-trip finite floating-point
// extremes (±DBL_MAX / ±FLT_MAX). The old digits10+1 (16g/7g) formatter
rounded
// DBL_MAX up to 1.797693134862316e+308 — larger than the largest finite
double —
diff --git a/be/test/core/data_type_serde/data_type_serde_timestamptz_test.cpp
b/be/test/core/data_type_serde/data_type_serde_timestamptz_test.cpp
index 584106cc635..742715ef62e 100644
--- a/be/test/core/data_type_serde/data_type_serde_timestamptz_test.cpp
+++ b/be/test/core/data_type_serde/data_type_serde_timestamptz_test.cpp
@@ -359,4 +359,40 @@ TEST_F(DataTypeTimeStampTzSerDeTest,
ReadArrowIgnoresValuesUnderNulls) {
ASSERT_EQ(2, dest_column->size());
EXPECT_EQ(source_value, dest_column->get_element(0));
}
+
+TEST_F(DataTypeTimeStampTzSerDeTest, ArrowTimestampToTimestampTz) {
+ const auto read_timestamp = [](arrow::TimeUnit::type unit, int64_t value,
+ const std::string& arrow_timezone,
+ const std::string& expected_utc,
+ const std::string& expected_session_time) {
+ auto type = arrow::timestamp(unit, arrow_timezone);
+ arrow::TimestampBuilder builder(type, arrow::default_memory_pool());
+ ASSERT_TRUE(builder.Append(value).ok());
+ ASSERT_TRUE(builder.AppendNull().ok());
+ std::shared_ptr<arrow::Array> array;
+ ASSERT_TRUE(builder.Finish(&array).ok());
+
+ auto column = ColumnTimeStampTz::create();
+ DataTypeTimeStampTzSerDe serde(6);
+ cctz::time_zone session_timezone;
+ ASSERT_TRUE(TimezoneUtils::find_cctz_time_zone("+08:00",
session_timezone));
+ const auto status = serde.read_column_from_arrow(*column, array.get(),
0, array->length(),
+ session_timezone);
+ ASSERT_TRUE(status.ok()) << status.to_string();
+ ASSERT_EQ(2, column->size());
+
+ const auto utc = cctz::utc_time_zone();
+ EXPECT_EQ(expected_utc, column->get_data()[0].to_string(utc, 6));
+ EXPECT_EQ(expected_session_time,
column->get_data()[0].to_string(session_timezone, 6));
+ };
+
+ read_timestamp(arrow::TimeUnit::SECOND, 0, "UTC", "1970-01-01
00:00:00.000000+00:00",
+ "1970-01-01 08:00:00.000000+08:00");
+ read_timestamp(arrow::TimeUnit::MILLI, 123, "Asia/Shanghai", "1970-01-01
00:00:00.123000+00:00",
+ "1970-01-01 08:00:00.123000+08:00");
+ read_timestamp(arrow::TimeUnit::MICRO, -876544, "UTC", "1969-12-31
23:59:59.123456+00:00",
+ "1970-01-01 07:59:59.123456+08:00");
+ read_timestamp(arrow::TimeUnit::NANO, -876543211, "Asia/Shanghai",
+ "1969-12-31 23:59:59.123456+00:00", "1970-01-01
07:59:59.123456+08:00");
+}
} // namespace doris
diff --git a/be/test/core/data_type_serde/data_type_serde_varbinary_test.cpp
b/be/test/core/data_type_serde/data_type_serde_varbinary_test.cpp
index 6016bb4a37a..5f03f363c86 100644
--- a/be/test/core/data_type_serde/data_type_serde_varbinary_test.cpp
+++ b/be/test/core/data_type_serde/data_type_serde_varbinary_test.cpp
@@ -170,7 +170,7 @@ TEST_F(DataTypeVarbinarySerDeTest,
MysqlTextAndBinaryAndConst) {
}
}
-TEST_F(DataTypeVarbinarySerDeTest, ArrowWriteSupportedReadNotImplemented) {
+TEST_F(DataTypeVarbinarySerDeTest, ArrowBinaryRoundTrip) {
DataTypeVarbinarySerDe serde;
auto col = ColumnVarbinary::create();
auto* vb = col.get();
@@ -185,8 +185,11 @@ TEST_F(DataTypeVarbinarySerDeTest,
ArrowWriteSupportedReadNotImplemented) {
std::shared_ptr<arrow::Array> arr;
ASSERT_TRUE(builder->Finish(&arr).ok());
- st = serde.read_column_from_arrow(*vb, arr.get(), 0, 1, tz);
- EXPECT_FALSE(st.ok());
+ auto read_column = ColumnVarbinary::create();
+ st = serde.read_column_from_arrow(*read_column, arr.get(), 0, 1, tz);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+ ASSERT_EQ(1U, read_column->size());
+ EXPECT_EQ(v, read_column->get_data_at(0).to_string());
auto* binary_array = dynamic_cast<arrow::BinaryArray*>(arr.get());
ASSERT_NE(binary_array, nullptr);
@@ -198,6 +201,49 @@ TEST_F(DataTypeVarbinarySerDeTest,
ArrowWriteSupportedReadNotImplemented) {
EXPECT_EQ(memcmp(raw, view.data, view.size), 0);
}
+TEST_F(DataTypeVarbinarySerDeTest, ArrowReadSupportsLargeAndFixedSizeBinary) {
+ DataTypeVarbinarySerDe serde;
+ cctz::time_zone tz;
+ const auto expect_values = [](const ColumnVarbinary& column,
+ const std::vector<std::string>& expected) {
+ ASSERT_EQ(expected.size(), column.size());
+ for (size_t i = 0; i < expected.size(); ++i) {
+ EXPECT_EQ(expected[i], column.get_data_at(i).to_string());
+ }
+ };
+
+ {
+ arrow::LargeBinaryBuilder builder;
+ ASSERT_TRUE(builder.Append("large").ok());
+ ASSERT_TRUE(builder.AppendNull().ok());
+ ASSERT_TRUE(builder.Append("binary").ok());
+ std::shared_ptr<arrow::Array> arrow_array;
+ ASSERT_TRUE(builder.Finish(&arrow_array).ok());
+
+ auto column = ColumnVarbinary::create();
+ ASSERT_TRUE(serde.read_column_from_arrow(*column, arrow_array.get(), 0,
+ arrow_array->length(), tz)
+ .ok());
+ expect_values(*column, {"large", "", "binary"});
+ }
+
+ {
+ auto type = arrow::fixed_size_binary(3);
+ arrow::FixedSizeBinaryBuilder builder(type,
arrow::default_memory_pool());
+ ASSERT_TRUE(builder.Append("abc").ok());
+ ASSERT_TRUE(builder.AppendNull().ok());
+ ASSERT_TRUE(builder.Append("xyz").ok());
+ std::shared_ptr<arrow::Array> arrow_array;
+ ASSERT_TRUE(builder.Finish(&arrow_array).ok());
+
+ auto column = ColumnVarbinary::create();
+ ASSERT_TRUE(serde.read_column_from_arrow(*column, arrow_array.get(), 0,
+ arrow_array->length(), tz)
+ .ok());
+ expect_values(*column, {"abc", "", "xyz"});
+ }
+}
+
TEST_F(DataTypeVarbinarySerDeTest, OrcWriteSupported) {
DataTypeVarbinarySerDe serde;
auto col = ColumnVarbinary::create();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
