You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by ga...@apache.org on 2026/10/12 01:38:04 UTC
(doris) branch master updated: [feature](arrow) Port Flight metadata, native Variant and VARBINARY fixes to master (#68883)
This is an automated email from the ASF dual-hosted git repository.
Gabriel39 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 84e5e6cd52b [feature](arrow) Port Flight metadata, native Variant and VARBINARY fixes to master (#68883)
84e5e6cd52b is described below
commit 84e5e6cd52b05ccbbe9a8482d1878a13f3924a20
Author: Gabriel <li...@selectdb.com>
AuthorDate: Mon Oct 12 09:37:56 2026 +0800
[feature](arrow) Port Flight metadata, native Variant and VARBINARY fixes to master (#68883)
### What problem does this PR solve?
Forward-port the requested changes whose missing behavior is still
present on master: complete the missing Variant writer hooks from
#68381, then retain the ports of #68614, #68667 and #68859 in branch-4.1
merge order.
| Source PR | Evidence on master before this change | Action |
| --- | --- | --- |
| #68381 | Master already contains the main converter refactor through
#68396, but lacks the Variant-specific Paimon/Iceberg SerDe overrides.
Native Flight dispatch intercepts Paimon binary structs. | Complete the
missing hooks and preserve Paimon `value,metadata` versus Flight
`metadata,value` storage contracts, including nested columns. |
| #68614 | Arrow ARRAY/MAP/STRUCT child fields lose Doris logical type
markers; JSON/VARIANT markers are absent. | Preserve recursive logical
metadata and use protocol-specific schema converters. |
| #68667 | Flight maps Variant V2 to UTF8 instead of native binary
values. | Return `arrow.parquet.variant` with `struct<metadata: binary
not null, value: binary not null>` storage; preserve SQL NULL, physical
scalar types and row-local dictionaries. |
| #68852 | The obsolete external-table queries are already removed. All
276 active query tags match the 276 expected output blocks in execution
order. | Excluded; no expected-output changes. |
| #68856 | The preload module is already removed. The Hudi plugin
resolves Parquet readers and format structures consistently to 1.18.0. |
Excluded; no dependency changes. |
| #68859 | Block metadata omits VARBINARY length, and deserialization
constructs `DataTypeVarbinary()` with an unspecified length. | Preserve
declared lengths, including nested and nullable columns, using optional
protobuf field 13. |
Adapt the ports to master's existing architecture: retain native UUID
and nanosecond/timestamp mappings, reuse the shared FE Arrow mapper, and
wrap extension storage correctly for nested Variant arrays. Master
already uses Variant V2, so the removed legacy Variant switch and SerDe
are not restored. This does not add the branch-4.1-only Iceberg table
Variant feature or prepared-statement schema machinery. ADBC recognizes
the native Variant extension, validates and decodes its bytes, and
retains Variant sub-column access; the scanner permits that carrier only
for ADBC among Arrow readers.
The Flight integration tests and Python samples cover native Variant and
recursive metadata. GetTables and ADBC expectations follow the native
schema. The Iceberg static-null partition unit-test fixture uses
master's existing Thrift field ID 19; production Thrift definitions are
unchanged. The Flink regression explicitly casts Variant to STRING for a
connector that declares that column as STRING.
### Release note
Preserve nested logical type metadata in Arrow Flight SQL results,
return Variant V2 as native Arrow Variant binary, preserve native
Variant values and sub-column access through ADBC, and retain declared
VARBINARY lengths across block serialization. Clients requiring Variant
JSON text should explicitly cast it to STRING.
### Validation
- FE targeted unit tests: 25 passed for the follow-up (ADBC mapping,
native Variant and Flight schemas); the initial port also passed 64
targeted tests including UUID and recursive mappings.
- FE Checkstyle: passed.
- BE targeted ASAN unit tests: 408 passed, covering Arrow, Variant,
ADBC, FileScannerV2, UUID, VARBINARY and the static-null Iceberg
fixture.
- Changed-line clang-tidy, clang-format 16 and build hygiene checks:
passed.
- Groovy regression scripts compiled; Python samples passed syntax
checks.
- Checked modified Thrift/protobuf field IDs against master and
branch-3.1, branch-4.0, branch-4.1 and branch-4.2.
- Confirmed the ES baseline query/output alignment and Hudi Parquet
dependency tree before excluding those ports.
- Built FE and ASAN BE with the repository build script. On a local
cluster, `test_adbc_type_semantic`, `test_get_tables_schema` and
`test_flight_native_variant` passed (3 suites, 0 failures, 0 skips). The
ADBC `.out` was generated by the regression framework and then checked
by a normal rerun.
- Full Paimon and Flink integration suites require the external test
environment and were not run locally. Paimon writer and nested-container
paths passed BE tests.
- Four muted P0 failures also recur on unrelated PR builds: two
optimizer plan-shape mismatches and two object-store upload timeouts.
They are not included as fixed by this port.
### Check List (For Author)
- Test
- [x] Unit tests
- [x] Regression tests added/updated
- Behavior changed:
- [x] Yes, native Variant Flight output and preserved type metadata.
- Does this need documentation?
- [x] Yes, Python Flight sample documentation updated.
---
be/src/core/data_type/data_type_factory.cpp | 2 +-
be/src/core/data_type/data_type_varbinary.cpp | 7 +
be/src/core/data_type/data_type_varbinary.h | 1 +
.../data_type_serde/data_type_variant_v2_serde.cpp | 222 ++++++++-
.../data_type_serde/data_type_variant_v2_serde.h | 15 +
.../exec/operator/memory_scratch_sink_operator.cpp | 3 +-
be/src/exec/operator/result_sink_operator.cpp | 23 +-
be/src/exec/operator/result_sink_operator.h | 3 +
be/src/exec/scan/file_scanner_v2.cpp | 18 +-
be/src/exec/scan/file_scanner_v2.h | 8 +-
.../aggregate/aggregate_function_python_udaf.cpp | 4 +-
be/src/format/arrow/arrow_block_convertor.cpp | 43 +-
be/src/format/arrow/arrow_block_convertor.h | 15 +-
be/src/format/arrow/arrow_row_batch.cpp | 217 ++++++---
be/src/format/arrow/arrow_row_batch.h | 79 ++-
.../parquet/parquet_arrow_block_convertor.cpp | 18 +-
.../format/parquet/parquet_arrow_block_convertor.h | 20 +
be/src/format/table/iceberg/arrow_schema_util.cpp | 54 ++-
be/src/format/table/iceberg/arrow_schema_util.h | 25 +-
.../iceberg/iceberg_arrow_block_convertor.cpp | 12 +-
.../table/paimon/paimon_arrow_block_convertor.cpp | 16 +-
.../table/paimon/paimon_arrow_block_convertor.h | 11 +
be/src/format_v2/table/adbc_reader.cpp | 6 +
.../arrow_flight/arrow_flight_batch_reader.cpp | 1 +
be/src/udf/python/python_udf_meta.cpp | 3 +-
.../core/data_type/common_data_type_serder_test.h | 4 +-
be/test/core/data_type/data_type_uuid_test.cpp | 9 +-
.../core/data_type/data_type_varbinary_test.cpp | 83 ++++
.../data_type_serde/data_type_serde_arrow_test.cpp | 37 +-
.../data_type_variant_v2_serde_input_test.cpp | 3 +-
.../data_type_variant_v2_serde_output_test.cpp | 15 +
be/test/exec/scan/file_scanner_v2_test.cpp | 26 +-
.../writer/iceberg/viceberg_table_writer_test.cpp | 4 +-
be/test/format/arrow/arrow_flight_variant_test.cpp | 531 +++++++++++++++++++++
be/test/format/arrow/arrow_row_batch_test.cpp | 313 ++++++++++++
.../table/iceberg/arrow_schema_util_test.cpp | 12 +-
be/test/format_v2/table/adbc_reader_test.cpp | 59 +++
.../doris/connector/adbc/AdbcTypeMapper.java | 14 +
.../doris/connector/adbc/AdbcTypeMapperTest.java | 16 +
.../arrowflight/FlightSqlConnectProcessor.java | 9 +-
.../doris/arrowflight/FlightSqlSchemaHelper.java | 101 +++-
.../java/org/apache/doris/planner/ResultSink.java | 5 +-
.../FlightSqlConnectProcessorSchemaTest.java | 167 +++++++
.../arrowflight/FlightSqlNativeVariantTest.java | 95 ++++
.../doris/planner/ResultSinkMetadataTest.java | 38 ++
gensrc/proto/data.proto | 2 +
gensrc/thrift/DataSinks.thrift | 2 +
.../adbc/test_adbc_type_semantic.out | 7 +-
.../test_flight_native_variant.groovy | 174 +++++++
.../test_get_tables_schema.groovy | 8 +-
.../adbc/test_adbc_type_semantic.groovy | 8 +-
.../flink_connector_p0/flink_connector_type.groovy | 29 +-
samples/arrow-flight-sql/python/README.md | 60 ++-
.../python/test_nested_type_metadata.py | 108 +++++
samples/arrow-flight-sql/python/test_variant.py | 84 ++++
55 files changed, 2625 insertions(+), 224 deletions(-)
diff --git a/be/src/core/data_type/data_type_factory.cpp b/be/src/core/data_type/data_type_factory.cpp
index a504dc6e908..f4896c31075 100644
--- a/be/src/core/data_type/data_type_factory.cpp
+++ b/be/src/core/data_type/data_type_factory.cpp
@@ -301,7 +301,7 @@ DataTypePtr DataTypeFactory::create_data_type(const PColumnMeta& pcolumn) {
nested = std::make_shared<DataTypeHLL>();
break;
case PGenericType::VARBINARY:
- nested = std::make_shared<DataTypeVarbinary>();
+ nested = std::make_shared<DataTypeVarbinary>(pcolumn.varbinary_length());
break;
case PGenericType::LIST:
DCHECK(pcolumn.children_size() == 1);
diff --git a/be/src/core/data_type/data_type_varbinary.cpp b/be/src/core/data_type/data_type_varbinary.cpp
index fd4364592a9..0094963634d 100644
--- a/be/src/core/data_type/data_type_varbinary.cpp
+++ b/be/src/core/data_type/data_type_varbinary.cpp
@@ -17,6 +17,7 @@
#include "core/data_type/data_type_varbinary.h"
+#include <gen_cpp/data.pb.h>
#include <gen_cpp/types.pb.h>
#include <glog/logging.h>
#include <lz4/lz4.h>
@@ -54,6 +55,12 @@ bool DataTypeVarbinary::equals(const IDataType& rhs) const {
return typeid(rhs) == typeid(*this);
}
+void DataTypeVarbinary::to_pb_column_meta(PColumnMeta* col_meta) const {
+ IDataType::to_pb_column_meta(col_meta);
+ // Fixed-width binary writers need the declared width after a Block crosses an exchange.
+ col_meta->set_varbinary_length(_len);
+}
+
// binary: const flag| row num | real saved num | size array | data array
// bool size_t size_t
// <size array>: data1 size | data2 size | ... sizeof(size_t) * real_need_copy_num
diff --git a/be/src/core/data_type/data_type_varbinary.h b/be/src/core/data_type/data_type_varbinary.h
index aa54a8a79d4..a7f6eca48c1 100644
--- a/be/src/core/data_type/data_type_varbinary.h
+++ b/be/src/core/data_type/data_type_varbinary.h
@@ -83,6 +83,7 @@ public:
};
void to_protobuf(PTypeDesc* ptype, PTypeNode* node, PScalarType* scalar_type) const override;
+ void to_pb_column_meta(PColumnMeta* col_meta) const override;
int len() const { return _len; }
diff --git a/be/src/core/data_type_serde/data_type_variant_v2_serde.cpp b/be/src/core/data_type_serde/data_type_variant_v2_serde.cpp
index 0a343aa7fa1..86c9b6c2c1f 100644
--- a/be/src/core/data_type_serde/data_type_variant_v2_serde.cpp
+++ b/be/src/core/data_type_serde/data_type_variant_v2_serde.cpp
@@ -17,12 +17,16 @@
#include "core/data_type_serde/data_type_variant_v2_serde.h"
+#include <arrow/array/array_binary.h>
+#include <arrow/array/array_nested.h>
#include <arrow/array/builder_binary.h>
#include <arrow/array/builder_nested.h>
+#include <arrow/extension_type.h>
#include <algorithm>
#include <cstring>
#include <limits>
+#include <optional>
#include <orc/Vector.hh>
#include <span>
#include <utility>
@@ -45,6 +49,7 @@
#include "core/types.h"
#include "core/value/jsonb_value.h"
#include "core/value/variant/variant_batch_builder.h"
+#include "core/value/variant/variant_field.h"
#include "exprs/function/parse/variant_jsonb_parse.h"
#include "exprs/function/parse/variant_string_parse.h"
#include "util/jsonb_document.h"
@@ -54,6 +59,48 @@
namespace doris {
namespace {
+// ColumnVariantV2 already validates its encoded dictionaries and value structure. Only walk
+// the selected row here: validating every unused dictionary key per row is quadratic for
+// shared dictionaries, including when a nested ARRAY invokes this writer on one-row slices.
+Status append_flight_variant_value(VariantRef value, VariantBatchBuilder::Row& output,
+ size_t depth = 0) {
+ const auto basic_type = value.basic_type();
+ if (depth > VARIANT_MAX_NESTING_DEPTH) {
+ return Status::NotSupported(
+ "Native Arrow Variant nesting exceeds {}; "
+ "cast the result to STRING for text output",
+ VARIANT_MAX_NESTING_DEPTH);
+ }
+ if (value.value_size() != value.value.size) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Native Arrow Variant contains trailing value bytes");
+ }
+ if (basic_type == VariantBasicType::OBJECT) {
+ auto object = output.start_object();
+ auto fields = value.object_view();
+ for (uint32_t i = 0; i < fields.size(); ++i) {
+ uint32_t field_id;
+ auto child = fields.value_at(i, &field_id);
+ object.add_key(value.metadata.key_at(field_id));
+ RETURN_IF_ERROR(append_flight_variant_value(child, output, depth + 1));
+ }
+ object.finish();
+ } else if (basic_type == VariantBasicType::ARRAY) {
+ auto array = output.start_array();
+ for (uint32_t i = 0; i < value.num_elements(); ++i) {
+ RETURN_IF_ERROR(append_flight_variant_value(value.array_at(i), output, depth + 1));
+ }
+ array.finish();
+ } else {
+ // Primitives never reference dictionary keys. Reuse physical import to retain widths,
+ // decimal scales and non-JSON types; canonical equality encoding normalizes those away.
+ static constexpr char empty_metadata[] = {0x11, 0, 0};
+ value.metadata = {.data = empty_metadata, .size = sizeof(empty_metadata)};
+ output.add_value(value);
+ }
+ return Status::OK();
+}
+
using MetaIdsColumn = ColumnVector<TYPE_UINT32>;
std::span<const NullMap::value_type> forced_nulls(const NullMap* null_map) {
@@ -450,10 +497,73 @@ Status DataTypeVariantV2SerDe::read_column_from_pb(IColumn& column, const PValue
return Status::NotSupported("read_column_from_pb with type " + column.get_name());
}
-Status DataTypeVariantV2SerDe::read_column_from_arrow(IColumn& column, const arrow::Array*, int64_t,
- int64_t, const cctz::time_zone&) const {
- return Status::Error(ErrorCode::NOT_IMPLEMENTED_ERROR,
- "read_column_from_arrow with type " + column.get_name());
+bool DataTypeVariantV2SerDe::is_native_arrow_type(const arrow::DataType& type) {
+ if (type.id() == arrow::Type::EXTENSION) {
+ const auto& extension = static_cast<const arrow::ExtensionType&>(type);
+ return extension.extension_name() == "arrow.parquet.variant" &&
+ is_native_arrow_type(*extension.storage_type());
+ }
+ return type.id() == arrow::Type::STRUCT && type.num_fields() == 2 &&
+ type.field(0)->name() == "metadata" && type.field(1)->name() == "value" &&
+ type.field(0)->type()->id() == arrow::Type::BINARY &&
+ type.field(1)->type()->id() == arrow::Type::BINARY;
+}
+
+Status DataTypeVariantV2SerDe::read_column_from_arrow(IColumn& column, const arrow::Array* array,
+ int64_t start, int64_t end,
+ const cctz::time_zone&) const {
+ if (array == nullptr) {
+ return Status::InvalidArgument("Native Variant Arrow array is null");
+ }
+ if (start < 0 || end < start || end > array->length()) {
+ return Status::InvalidArgument("Invalid native Variant Arrow row range [{}, {})", start,
+ end);
+ }
+ if (!is_native_arrow_type(*array->type())) {
+ return Status::InvalidArgument("Expected native Variant Arrow storage, got {}",
+ array->type()->ToString());
+ }
+ RETURN_IF_CATCH_EXCEPTION({
+ if (array->type_id() == arrow::Type::EXTENSION) {
+ array = static_cast<const arrow::ExtensionArray*>(array)->storage().get();
+ }
+ // Nested ARRAY readers invoke this on row slices; validating the entire child array
+ // each time would make a batch with many small arrays quadratic.
+ const auto slice = array->Slice(start, end - start);
+ const auto validation = slice->ValidateFull();
+ if (!validation.ok()) {
+ return Status::InvalidArgument("Invalid native Variant Arrow array: {}",
+ validation.ToString());
+ }
+ const auto& storage = assert_cast<const arrow::StructArray&>(*slice);
+ const auto& metadata = assert_cast<const arrow::BinaryArray&>(*storage.field(0));
+ const auto& values = assert_cast<const arrow::BinaryArray&>(*storage.field(1));
+ // A driver supplies untrusted bytes. Validate into a temporary batch before mutating
+ // the destination, and copy buffers so the column survives Arrow batch release.
+ VariantBatchBuilder decoded;
+ for (int64_t i = 0; i < slice->length(); ++i) {
+ auto row = decoded.begin_row();
+ if (storage.IsNull(i)) {
+ row.add_null();
+ } else {
+ if (metadata.IsNull(i) || values.IsNull(i)) {
+ return Status::InvalidArgument(
+ "Native Variant has a NULL metadata/value child at row {}", i);
+ }
+ const auto meta = metadata.GetView(i);
+ const auto value = values.GetView(i);
+ const auto ref =
+ (VariantRef {.metadata = {.data = meta.data(), .size = meta.size()},
+ .value = {value.data(), value.size()}});
+ validate_variant_metadata(ref.metadata);
+ validate_variant_payload(ref);
+ row.add_value(ref);
+ }
+ row.finish();
+ }
+ assert_cast<ColumnVariantV2&>(column).insert_encoded_batch(decoded.finish_batch());
+ });
+ return Status::OK();
}
namespace {
@@ -662,6 +772,79 @@ Status write_arrow(const IColumn& column, const NullMap* null_map, Builder& buil
return status;
}
+Status write_parquet_variant_arrow(const IColumn& column, const NullMap* null_map,
+ arrow::ArrayBuilder* array_builder, int64_t start, int64_t end,
+ bool compact_metadata) {
+ if (start < 0 || end < start) {
+ return Status::InvalidArgument("Invalid Variant Arrow row range [{}, {})", start, end);
+ }
+ if (array_builder->type()->id() != arrow::Type::STRUCT) {
+ return Status::InvalidArgument("Variant Arrow writer requires a struct builder, got {}",
+ array_builder->type()->ToString());
+ }
+ auto& builder = assert_cast<arrow::StructBuilder&>(*array_builder);
+ const auto type = std::static_pointer_cast<arrow::StructType>(builder.type());
+ if (type->num_fields() != 2 || type->field(0)->name() != "metadata" ||
+ type->field(1)->name() != "value" || type->field(0)->type()->id() != arrow::Type::BINARY ||
+ type->field(1)->type()->id() != arrow::Type::BINARY) {
+ return Status::InvalidArgument(
+ "Variant Arrow writer requires struct<metadata: binary, value: binary>, got {}",
+ type->ToString());
+ }
+ auto& metadata_builder = assert_cast<arrow::BinaryBuilder&>(*builder.field_builder(0));
+ auto& value_builder = assert_cast<arrow::BinaryBuilder&>(*builder.field_builder(1));
+ Status status = Status::OK();
+ visit_variant_v2_values(
+ column, start, end, forced_nulls(null_map),
+ [&](size_t) {
+ if (status.ok()) {
+ status = checkArrowStatus(builder.AppendNull(), column, builder);
+ }
+ },
+ [&](size_t, VariantRef value) {
+ if (!status.ok()) {
+ return;
+ }
+ std::optional<VariantBatchBuilder> compacted;
+ if (compact_metadata) {
+ const auto keys = value.metadata.dict_size();
+ // An empty dictionary or an object using every key already has row-local metadata.
+ if (keys != 0 && (value.basic_type() != VariantBasicType::OBJECT ||
+ value.num_elements() != keys)) {
+ VariantBatchBuilder encoder;
+ auto row = encoder.begin_row();
+ status = append_flight_variant_value(value, row);
+ if (!status.ok()) {
+ return;
+ }
+ row.finish();
+ compacted.emplace(encoder.finish_batch());
+ value = compacted->value_at(0);
+ }
+ }
+ if (value.metadata.size > std::numeric_limits<int32_t>::max() ||
+ value.value.size > std::numeric_limits<int32_t>::max()) {
+ status = Status::InvalidArgument(
+ "Variant Arrow metadata/value exceeds Arrow binary size limit");
+ return;
+ }
+ status = checkArrowStatus(builder.Append(), column, builder);
+ if (status.ok()) {
+ status = checkArrowStatus(
+ metadata_builder.Append(value.metadata.data,
+ cast_set<int32_t>(value.metadata.size)),
+ column, metadata_builder);
+ }
+ if (status.ok()) {
+ status = checkArrowStatus(
+ value_builder.Append(value.value.data,
+ cast_set<int32_t>(value.value.size)),
+ column, value_builder);
+ }
+ });
+ return status;
+}
+
} // namespace
void DataTypeVariantV2SerDe::to_string(const IColumn& column, size_t row_num, BufferWritable& bw,
@@ -715,6 +898,9 @@ Status DataTypeVariantV2SerDe::write_column_to_arrow(const IColumn& column, cons
options.timezone = &ctz;
const size_t first = checked_row(start);
const size_t last = checked_row(end);
+ if (is_native_arrow_type(*array_builder->type())) {
+ return write_parquet_variant_arrow(column, null_map, array_builder, start, end, true);
+ }
if (array_builder->type()->id() == arrow::Type::STRING) {
return write_arrow(column, null_map, assert_cast<arrow::StringBuilder&>(*array_builder),
first, last, options);
@@ -735,6 +921,34 @@ Status DataTypeVariantV2SerDe::write_column_to_arrow(const IColumn& column, cons
return Status::OK();
}
+Status DataTypeVariantV2SerDe::write_column_to_paimon_arrow(
+ const std::shared_ptr<const IDataType>&, const IColumn& column, const NullMap* null_map,
+ const std::shared_ptr<arrow::Field>&, arrow::ArrayBuilder* array_builder, int64_t start,
+ int64_t end, const cctz::time_zone&) const {
+ // Paimon owns the value/metadata order and validation; Flight uses metadata/value.
+ // Keep protocol dispatch explicit so changes to Flight cannot intercept table writes.
+ if (array_builder->type()->id() != arrow::Type::STRUCT) {
+ return Status::InvalidArgument("Paimon Variant requires a struct builder");
+ }
+ RETURN_IF_CATCH_EXCEPTION({
+ return write_binary_variant_arrow(column, null_map,
+ assert_cast<arrow::StructBuilder&>(*array_builder),
+ checked_row(start), checked_row(end));
+ });
+ return Status::OK();
+}
+
+Status DataTypeVariantV2SerDe::write_column_to_iceberg_arrow(
+ const std::shared_ptr<const IDataType>&, const IColumn& column, const NullMap* null_map,
+ const std::shared_ptr<arrow::Field>&, arrow::ArrayBuilder* array_builder, int64_t start,
+ int64_t end, const cctz::time_zone&) const {
+ // Table writers retain their validated dictionary; Flight compacts each transmitted row.
+ RETURN_IF_CATCH_EXCEPTION({
+ return write_parquet_variant_arrow(column, null_map, array_builder, start, end, false);
+ });
+ return Status::OK();
+}
+
Status DataTypeVariantV2SerDe::write_column_to_orc(const std::string&, const IColumn& column,
const NullMap* null_map,
orc::ColumnVectorBatch* orc_col_batch,
diff --git a/be/src/core/data_type_serde/data_type_variant_v2_serde.h b/be/src/core/data_type_serde/data_type_variant_v2_serde.h
index d47b40dc43b..f3325fa514e 100644
--- a/be/src/core/data_type_serde/data_type_variant_v2_serde.h
+++ b/be/src/core/data_type_serde/data_type_variant_v2_serde.h
@@ -19,6 +19,10 @@
#include "core/data_type_serde/data_type_serde.h"
+namespace arrow {
+class DataType;
+}
+
namespace doris {
// Direct SerDe used when the logical Variant type carries a ColumnVariantV2 physical column.
@@ -63,6 +67,17 @@ public:
Status write_column_to_arrow(const IColumn& column, const NullMap* null_map,
arrow::ArrayBuilder* array_builder, int64_t start, int64_t end,
const cctz::time_zone& ctz) const override;
+ static bool is_native_arrow_type(const arrow::DataType& type);
+ Status write_column_to_paimon_arrow(const std::shared_ptr<const IDataType>& type,
+ const IColumn& column, const NullMap* null_map,
+ const std::shared_ptr<arrow::Field>& field,
+ arrow::ArrayBuilder* array_builder, int64_t start,
+ int64_t end, const cctz::time_zone& ctz) const override;
+ Status write_column_to_iceberg_arrow(const std::shared_ptr<const IDataType>& type,
+ const IColumn& column, const NullMap* null_map,
+ const std::shared_ptr<arrow::Field>& field,
+ arrow::ArrayBuilder* array_builder, int64_t start,
+ int64_t end, 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;
Status write_column_to_mysql_binary(const IColumn& column, MysqlRowBinaryBuffer& row_buffer,
diff --git a/be/src/exec/operator/memory_scratch_sink_operator.cpp b/be/src/exec/operator/memory_scratch_sink_operator.cpp
index da7f6fec481..1d795001719 100644
--- a/be/src/exec/operator/memory_scratch_sink_operator.cpp
+++ b/be/src/exec/operator/memory_scratch_sink_operator.cpp
@@ -102,8 +102,7 @@ Status MemoryScratchSinkOperatorX::sink_impl(RuntimeState* state, Block* input_b
RETURN_IF_ERROR(VExprContext::get_output_block_after_execute_exprs(
local_state._output_vexpr_ctxs, *input_block, &block));
}
- DorisArrowBlockConvertor converter(block, state->timezone(), _timezone_obj,
- /*datetime_naive=*/true);
+ ArrowFlightArrowBlockConvertor converter(block, state->timezone(), _timezone_obj);
{
SCOPED_TIMER(local_state._get_arrow_schema_timer);
// After expr executed, use recaculated schema as final schema
diff --git a/be/src/exec/operator/result_sink_operator.cpp b/be/src/exec/operator/result_sink_operator.cpp
index c523542aea2..cf705be1257 100644
--- a/be/src/exec/operator/result_sink_operator.cpp
+++ b/be/src/exec/operator/result_sink_operator.cpp
@@ -54,9 +54,8 @@ Status ResultSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo& info)
} else {
std::shared_ptr<arrow::Schema> arrow_schema;
if (p._sink_type == TResultSinkType::ARROW_FLIGHT_PROTOCOL) {
- RETURN_IF_ERROR(get_arrow_schema_from_expr_ctxs(_output_vexpr_ctxs, &arrow_schema,
- state->timezone(),
- /*datetime_naive=*/true));
+ RETURN_IF_ERROR(p._arrow_schema_convertor->get_arrow_schema_from_expr_ctxs(
+ _output_vexpr_ctxs, &arrow_schema));
}
VLOG_DEBUG << "create sender in INIT with instance id " << fragment_instance_id;
RETURN_IF_ERROR(state->exec_env()->result_mgr()->create_sender(
@@ -100,6 +99,8 @@ ResultSinkOperatorX::ResultSinkOperatorX(int operator_id, int node_id,
_sink_type(!sink.__isset.type || sink.type == TResultSinkType::MYSQL_PROTOCOL
? TResultSinkType::MYSQL_PROTOCOL
: sink.type),
+ _enable_arrow_type_metadata(sink.__isset.enable_arrow_type_metadata &&
+ sink.enable_arrow_type_metadata),
_result_sink_buffer_size_rows(_sink_type == TResultSinkType::ARROW_FLIGHT_PROTOCOL
? config::arrow_flight_result_sink_buffer_size_rows
: RESULT_SINK_BUFFER_SIZE),
@@ -116,12 +117,22 @@ Status ResultSinkOperatorX::prepare(RuntimeState* state) {
// Prepare the exprs to run.
RETURN_IF_ERROR(VExpr::prepare(_output_vexpr_ctxs, state, _row_desc));
+ if (_sink_type == TResultSinkType::ARROW_FLIGHT_PROTOCOL) {
+ // Capability negotiation selects one immutable policy for schema fetches and batches.
+ if (_enable_arrow_type_metadata) {
+ _arrow_schema_convertor =
+ std::make_shared<ArrowFlightSchemaConvertor>(state->timezone());
+ } else {
+ _arrow_schema_convertor =
+ std::make_shared<LegacyArrowFlightSchemaConvertor>(state->timezone());
+ }
+ }
+
if (state->query_options().enable_parallel_result_sink) {
std::shared_ptr<arrow::Schema> arrow_schema;
if (_sink_type == TResultSinkType::ARROW_FLIGHT_PROTOCOL) {
- RETURN_IF_ERROR(get_arrow_schema_from_expr_ctxs(_output_vexpr_ctxs, &arrow_schema,
- state->timezone(),
- /*datetime_naive=*/true));
+ RETURN_IF_ERROR(_arrow_schema_convertor->get_arrow_schema_from_expr_ctxs(
+ _output_vexpr_ctxs, &arrow_schema));
}
VLOG_DEBUG << "create sender in prepare with query id " << state->query_id();
RETURN_IF_ERROR(state->exec_env()->result_mgr()->create_sender(
diff --git a/be/src/exec/operator/result_sink_operator.h b/be/src/exec/operator/result_sink_operator.h
index 690a99495ca..51be25c02a6 100644
--- a/be/src/exec/operator/result_sink_operator.h
+++ b/be/src/exec/operator/result_sink_operator.h
@@ -25,6 +25,7 @@
#include "runtime/result_block_buffer.h"
namespace doris {
+class DorisArrowSchemaConvertor;
class ResultBlockBufferBase;
struct ResultFileOptions {
@@ -164,6 +165,8 @@ private:
friend class ResultSinkLocalState;
const TResultSinkType::type _sink_type;
+ const bool _enable_arrow_type_metadata;
+ std::shared_ptr<const DorisArrowSchemaConvertor> _arrow_schema_convertor;
const int _result_sink_buffer_size_rows;
// set file options when sink type is FILE
std::unique_ptr<ResultFileOptions> _file_opts = nullptr;
diff --git a/be/src/exec/scan/file_scanner_v2.cpp b/be/src/exec/scan/file_scanner_v2.cpp
index 540a937ef1c..5247e7b7532 100644
--- a/be/src/exec/scan/file_scanner_v2.cpp
+++ b/be/src/exec/scan/file_scanner_v2.cpp
@@ -373,15 +373,18 @@ Status FileScannerV2::_validate_scan_range(const TFileScanRangeParams& params,
}
Status FileScannerV2::_validate_variant_projection(TFileFormatType::type format_type,
- bool has_variant_projection) {
- if (!has_variant_projection || format_type != TFileFormatType::FORMAT_ARROW) {
+ bool has_variant_projection,
+ const std::string& table_format) {
+ // ADBC validates native Variant storage before decoding; other Arrow readers do not.
+ if (!has_variant_projection || format_type != TFileFormatType::FORMAT_ARROW ||
+ table_format == "adbc") {
return Status::OK();
}
- // Arrow readers have no external Variant carrier contract; reject before a stream is opened so
- // every Arrow-backed table gets the same deterministic capability error instead of SerDe leakage.
+ // Keep the capability guard for readers without a native Variant carrier contract.
return Status::NotSupported(
- "External Variant is supported only for Parquet files in FileScannerV2; "
- "file format ARROW is not supported");
+ "External Variant Arrow reads require an ADBC native Variant carrier; "
+ "table format {} is not supported",
+ table_format);
}
FileScannerV2::FileScannerV2(RuntimeState* state, FileScanLocalState* local_state, int64_t limit,
@@ -969,7 +972,8 @@ Status FileScannerV2::_build_projected_columns(const format::TableReader& table_
}
RETURN_IF_ERROR(table_reader.validate_projected_columns(build_context));
RETURN_IF_ERROR(_validate_variant_projection(get_range_format_type(*_params, _current_range),
- _has_variant_projection));
+ _has_variant_projection,
+ table_format_name(_current_range)));
return Status::OK();
}
diff --git a/be/src/exec/scan/file_scanner_v2.h b/be/src/exec/scan/file_scanner_v2.h
index 13a6d593a3c..25e2a433d13 100644
--- a/be/src/exec/scan/file_scanner_v2.h
+++ b/be/src/exec/scan/file_scanner_v2.h
@@ -92,8 +92,9 @@ public:
static bool TEST_should_skip_not_found(const Status& status, bool ignore_not_found);
static bool TEST_should_skip_empty(const Status& status, bool stopped);
static Status TEST_validate_variant_projection(TFileFormatType::type format_type,
- bool has_variant_projection) {
- return _validate_variant_projection(format_type, has_variant_projection);
+ bool has_variant_projection,
+ const std::string& table_format) {
+ return _validate_variant_projection(format_type, has_variant_projection, table_format);
}
static Status TEST_contextualize_output_filter_status(Status status,
TFileFormatType::type format_type) {
@@ -142,7 +143,8 @@ private:
static Status _validate_scan_range(const TFileScanRangeParams& params,
const TFileRangeDesc& range);
static Status _validate_variant_projection(TFileFormatType::type format_type,
- bool has_variant_projection);
+ bool has_variant_projection,
+ const std::string& table_format);
Status _get_next_scan_range(bool* has_next);
TFileFormatType::type _get_current_format_type() const;
Status _init_io_ctx();
diff --git a/be/src/exprs/aggregate/aggregate_function_python_udaf.cpp b/be/src/exprs/aggregate/aggregate_function_python_udaf.cpp
index a0f227a75c5..ab6000783ad 100644
--- a/be/src/exprs/aggregate/aggregate_function_python_udaf.cpp
+++ b/be/src/exprs/aggregate/aggregate_function_python_udaf.cpp
@@ -246,10 +246,10 @@ void AggregatePythonUDAF::create(AggregateDataPtr __restrict place) const {
std::call_once(_schema_init_flag, [this]() {
std::vector<std::shared_ptr<arrow::Field>> fields;
- std::string timezone = TimezoneUtils::default_time_zone;
+ const DorisArrowSchemaConvertor schema_convertor(TimezoneUtils::default_time_zone);
for (size_t i = 0; i < argument_types.size(); ++i) {
std::shared_ptr<arrow::DataType> arrow_type;
- Status st = convert_to_arrow_type(argument_types[i], &arrow_type, timezone);
+ Status st = schema_convertor.convert_to_arrow_type(argument_types[i], &arrow_type);
if (!st.ok()) {
throw doris::Exception(ErrorCode::INTERNAL_ERROR,
"Failed to convert argument type {} to Arrow type: {}", i,
diff --git a/be/src/format/arrow/arrow_block_convertor.cpp b/be/src/format/arrow/arrow_block_convertor.cpp
index e4b30487044..596d0bcdc1b 100644
--- a/be/src/format/arrow/arrow_block_convertor.cpp
+++ b/be/src/format/arrow/arrow_block_convertor.cpp
@@ -95,6 +95,14 @@ std::shared_ptr<arrow::ArrayData> restore_arrow_logical_type(
if (storage->type->Equals(logical_type)) {
return storage;
}
+ if (logical_type->id() == arrow::Type::EXTENSION) {
+ // An extension exposes its children through storage, not ExtensionType::fields().
+ // Restore nested storage before attaching the logical type (Variant stores a struct).
+ const auto& extension = static_cast<const arrow::ExtensionType&>(*logical_type);
+ auto result = restore_arrow_logical_type(storage, extension.storage_type())->Copy();
+ result->type = logical_type;
+ return result;
+ }
auto result = storage->Copy();
result->type = logical_type;
DORIS_CHECK_EQ(result->child_data.size(), logical_type->num_fields());
@@ -267,7 +275,8 @@ Status ArrowBlockConvertor::write_plain_arrow_column(const std::shared_ptr<const
int64_t start, int64_t end,
const cctz::time_zone& ctz) const {
std::shared_ptr<arrow::DataType> plain_arrow_type;
- RETURN_IF_ERROR(convert_to_arrow_type(type, &plain_arrow_type, ctz.name()));
+ RETURN_IF_ERROR(
+ DorisArrowSchemaConvertor(ctz.name()).convert_to_arrow_type(type, &plain_arrow_type));
// This is an exact binding check selected by the target converter, not a recovery path. A
// mismatch returns without invoking SerDe, and a SerDe error is never retried elsewhere.
if (!is_declared_plain_arrow_binding(type, plain_arrow_type, field->type())) {
@@ -295,12 +304,40 @@ Status ArrowBlockConvertor::init() {
return Status::OK();
}
+Status ArrowFlightArrowBlockConvertor::write_column(const std::shared_ptr<const IDataType>& type,
+ const DataTypeSerDe& serde,
+ const IColumn& column, const NullMap* null_map,
+ const std::shared_ptr<arrow::Field>& field,
+ arrow::ArrayBuilder* array_builder,
+ int64_t start, int64_t end,
+ const cctz::time_zone& ctz) const {
+ std::shared_ptr<arrow::DataType> native_type;
+ RETURN_IF_ERROR(
+ ArrowFlightSchemaConvertor(ctz.name()).convert_to_arrow_type(type, &native_type));
+ // Check the extension identity and its complete nested shape before allowing the
+ // Variant SerDe to write binary storage. An arbitrary STRUCT is not a Variant binding.
+ // Timestamp labels may differ for equivalent fixed offsets, including inside containers.
+ if (is_declared_plain_arrow_binding(type, native_type, field->type())) {
+ return serde.write_column_to_arrow(column, null_map, array_builder, start, end, ctz);
+ }
+ return DorisArrowBlockConvertor::write_column(type, serde, column, null_map, field,
+ array_builder, start, end, ctz);
+}
+
Status DorisArrowBlockConvertor::init() {
if (_arrow_schema == nullptr) {
// cctz names fixed offsets as "Fixed/UTC+HH:MM:SS", which is not the Arrow
// protocol label. Keep the declared name so Python metadata and batches agree.
- RETURN_IF_ERROR(get_arrow_schema_from_block(_header, &_arrow_schema, _timezone_name,
- _datetime_naive));
+ RETURN_IF_ERROR(DorisArrowSchemaConvertor(_header, _timezone_name)
+ .get_arrow_schema(&_arrow_schema));
+ }
+ return ArrowBlockConvertor::init();
+}
+
+Status ArrowFlightArrowBlockConvertor::init() {
+ if (_arrow_schema == nullptr) {
+ RETURN_IF_ERROR(ArrowFlightSchemaConvertor(_header, _timezone_name)
+ .get_arrow_schema(&_arrow_schema));
}
return ArrowBlockConvertor::init();
}
diff --git a/be/src/format/arrow/arrow_block_convertor.h b/be/src/format/arrow/arrow_block_convertor.h
index ec1083505a9..1f35ee30652 100644
--- a/be/src/format/arrow/arrow_block_convertor.h
+++ b/be/src/format/arrow/arrow_block_convertor.h
@@ -89,11 +89,10 @@ class DorisArrowBlockConvertor : public ArrowBlockConvertor {
public:
using ArrowBlockConvertor::ArrowBlockConvertor;
DorisArrowBlockConvertor(const Block& header, std::string timezone_name,
- const cctz::time_zone& timezone, bool datetime_naive = false)
+ const cctz::time_zone& timezone)
: ArrowBlockConvertor(nullptr, timezone),
_header(header.clone_empty()),
- _timezone_name(std::move(timezone_name)),
- _datetime_naive(datetime_naive) {}
+ _timezone_name(std::move(timezone_name)) {}
Status init() override;
Status convert_from_arrow(const std::shared_ptr<arrow::RecordBatch>& batch,
@@ -106,15 +105,21 @@ protected:
arrow::ArrayBuilder* array_builder, int64_t start, int64_t end,
const cctz::time_zone& ctz) const override;
-private:
Block _header;
std::string _timezone_name;
- bool _datetime_naive = false;
};
class ArrowFlightArrowBlockConvertor final : public DorisArrowBlockConvertor {
public:
using DorisArrowBlockConvertor::DorisArrowBlockConvertor;
+ Status init() override;
+
+protected:
+ Status write_column(const std::shared_ptr<const IDataType>& type, const DataTypeSerDe& serde,
+ const IColumn& column, const NullMap* null_map,
+ const std::shared_ptr<arrow::Field>& field,
+ arrow::ArrayBuilder* array_builder, int64_t start, int64_t end,
+ const cctz::time_zone& ctz) const override;
};
class PythonArrowBlockConvertor final : public DorisArrowBlockConvertor {
diff --git a/be/src/format/arrow/arrow_row_batch.cpp b/be/src/format/arrow/arrow_row_batch.cpp
index 39356d6eb52..d24ead68cf3 100644
--- a/be/src/format/arrow/arrow_row_batch.cpp
+++ b/be/src/format/arrow/arrow_row_batch.cpp
@@ -19,6 +19,7 @@
#include <arrow/array/util.h>
#include <arrow/buffer.h>
+#include <arrow/extension/parquet_variant.h>
#include <arrow/extension/uuid.h>
#include <arrow/io/memory.h>
#include <arrow/ipc/writer.h>
@@ -42,17 +43,34 @@
#include "core/data_type/data_type_array.h"
#include "core/data_type/data_type_map.h"
#include "core/data_type/data_type_struct.h"
+#include "core/data_type/data_type_variant_v2.h"
#include "core/data_type/define_primitive_type.h"
#include "exprs/vexpr.h"
#include "exprs/vexpr_context.h"
#include "format/arrow/arrow_block_convertor.h"
+#include "format/arrow/arrow_utils.h"
#include "runtime/descriptors.h"
namespace doris {
-Status convert_to_arrow_type(const DataTypePtr& origin_type,
- std::shared_ptr<arrow::DataType>* result, const std::string& timezone,
- bool datetime_naive) {
+Status register_arrow_variant_extension() {
+ // Remote Flight readers must restore the extension before decoding the result schema.
+ static const auto status = [] {
+ if (arrow::GetExtensionType("arrow.parquet.variant") != nullptr) {
+ return arrow::Status::OK();
+ }
+ return arrow::RegisterExtensionType(
+ std::static_pointer_cast<arrow::ExtensionType>(arrow::extension::variant(
+ arrow::struct_({arrow::field("metadata", arrow::binary(), false),
+ arrow::field("value", arrow::binary(), false)}))));
+ }();
+ return status.ok() ? Status::OK() : Status::InternalError(status.ToString());
+}
+
+// Keep the exhaustive type mapping together so protocol-specific overrides remain visible.
+// NOLINTNEXTLINE(readability-function-size)
+Status DorisArrowSchemaConvertor::convert_to_arrow_type(
+ const DataTypePtr& origin_type, std::shared_ptr<arrow::DataType>* result) const {
auto type = get_serialized_type(origin_type);
switch (type->get_primitive_type()) {
case TYPE_NULL:
@@ -116,18 +134,8 @@ Status convert_to_arrow_type(const DataTypePtr& origin_type,
} else {
time_unit = arrow::TimeUnit::SECOND;
}
- // Doris DATETIMEV2 represents a wall-clock value without a timezone. Arrow Flight
- // exposes it as a timezone-naive timestamp so clients do not interpret it as an instant.
- // This option only changes the DATETIMEV2 output schema. TIMESTAMPTZ remains timezone-aware,
- // and Arrow-to-Doris conversions are unaffected.
- if (type->get_primitive_type() == TYPE_DATETIMEV2 && datetime_naive) {
- *result = std::make_shared<arrow::TimestampType>(time_unit);
- } else {
- // Arrow clients resolve timezone metadata as an IANA name; use the canonical UTC
- // name instead of the ISO-8601 "Z" alias without changing the encoded instant.
- *result = std::make_shared<arrow::TimestampType>(time_unit,
- timezone == "Z" ? "UTC" : timezone);
- }
+ *result = std::make_shared<arrow::TimestampType>(
+ time_unit, timestamp_timezone(type->get_primitive_type()));
break;
}
case TYPE_DECIMALV2:
@@ -145,20 +153,23 @@ Status convert_to_arrow_type(const DataTypePtr& origin_type,
case TYPE_ARRAY: {
const auto* type_arr = assert_cast<const DataTypeArray*>(remove_nullable(type).get());
std::shared_ptr<arrow::DataType> item_type;
- RETURN_IF_ERROR(convert_to_arrow_type(type_arr->get_nested_type(), &item_type, timezone,
- datetime_naive));
- *result = std::make_shared<arrow::ListType>(item_type);
+ RETURN_IF_ERROR(convert_to_arrow_type(type_arr->get_nested_type(), &item_type));
+ // Arrow stores metadata on fields, so implicit child fields lose the Doris logical type.
+ *result = std::make_shared<arrow::ListType>(make_child_field(
+ "item", item_type, true, type_arr->get_nested_type()->get_primitive_type()));
break;
}
case TYPE_MAP: {
const auto* type_map = assert_cast<const DataTypeMap*>(remove_nullable(type).get());
std::shared_ptr<arrow::DataType> key_type;
std::shared_ptr<arrow::DataType> val_type;
- RETURN_IF_ERROR(convert_to_arrow_type(type_map->get_key_type(), &key_type, timezone,
- datetime_naive));
- RETURN_IF_ERROR(convert_to_arrow_type(type_map->get_value_type(), &val_type, timezone,
- datetime_naive));
- *result = std::make_shared<arrow::MapType>(key_type, val_type);
+ RETURN_IF_ERROR(convert_to_arrow_type(type_map->get_key_type(), &key_type));
+ RETURN_IF_ERROR(convert_to_arrow_type(type_map->get_value_type(), &val_type));
+ auto key_field = make_child_field("key", key_type, false,
+ type_map->get_key_type()->get_primitive_type());
+ auto value_field = make_child_field("value", val_type, true,
+ type_map->get_value_type()->get_primitive_type());
+ *result = std::make_shared<arrow::MapType>(key_field, value_field);
break;
}
case TYPE_STRUCT: {
@@ -166,19 +177,17 @@ Status convert_to_arrow_type(const DataTypePtr& origin_type,
std::vector<std::shared_ptr<arrow::Field>> fields;
for (size_t i = 0; i < type_struct->get_elements().size(); i++) {
std::shared_ptr<arrow::DataType> field_type;
- RETURN_IF_ERROR(convert_to_arrow_type(type_struct->get_element(i), &field_type,
- timezone, datetime_naive));
- fields.push_back(
- std::make_shared<arrow::Field>(type_struct->get_element_name(i), field_type,
- type_struct->get_element(i)->is_nullable()));
+ RETURN_IF_ERROR(convert_to_arrow_type(type_struct->get_element(i), &field_type));
+ fields.push_back(make_child_field(type_struct->get_element_name(i), field_type,
+ type_struct->get_element(i)->is_nullable(),
+ type_struct->get_element(i)->get_primitive_type()));
}
*result = std::make_shared<arrow::StructType>(fields);
break;
}
- case TYPE_VARIANT: {
+ case TYPE_VARIANT:
*result = arrow::utf8();
break;
- }
case TYPE_QUANTILE_STATE:
case TYPE_BITMAP:
case TYPE_HLL: {
@@ -196,59 +205,124 @@ Status convert_to_arrow_type(const DataTypePtr& origin_type,
return Status::OK();
}
-// Helper function to create an Arrow Field with type metadata if applicable, such as IP types
+// Logical types sharing Arrow storage need the same marker at the root and every child field.
std::shared_ptr<arrow::Field> create_arrow_field_with_metadata(
const std::string& field_name, const std::shared_ptr<arrow::DataType>& arrow_type,
bool is_nullable, PrimitiveType primitive_type) {
- if (primitive_type == PrimitiveType::TYPE_IPV4) {
- auto metadata = arrow::KeyValueMetadata::Make({"doris_type"}, {"IPV4"});
- return std::make_shared<arrow::Field>(field_name, arrow_type, is_nullable, metadata);
- } else if (primitive_type == PrimitiveType::TYPE_IPV6) {
- auto metadata = arrow::KeyValueMetadata::Make({"doris_type"}, {"IPV6"});
- return std::make_shared<arrow::Field>(field_name, arrow_type, is_nullable, metadata);
- } else if (primitive_type == PrimitiveType::TYPE_UUID) {
- auto metadata = arrow::KeyValueMetadata::Make({"doris_type"}, {"UUID"});
- return std::make_shared<arrow::Field>(field_name, arrow_type, is_nullable, metadata);
- } else if (primitive_type == PrimitiveType::TYPE_LARGEINT) {
- auto metadata = arrow::KeyValueMetadata::Make({"doris_type"}, {"LARGEINT"});
- return std::make_shared<arrow::Field>(field_name, arrow_type, is_nullable, metadata);
- } else {
+ const char* type_name;
+ switch (primitive_type) {
+ case TYPE_UUID:
+ type_name = "UUID";
+ break;
+ case TYPE_IPV4:
+ type_name = "IPV4";
+ break;
+ case TYPE_IPV6:
+ type_name = "IPV6";
+ break;
+ case TYPE_LARGEINT:
+ type_name = "LARGEINT";
+ break;
+ case TYPE_JSONB:
+ type_name = "JSON";
+ break;
+ case TYPE_VARIANT:
+ type_name = "VARIANT";
+ break;
+ default:
return std::make_shared<arrow::Field>(field_name, arrow_type, is_nullable);
}
+ auto metadata = arrow::KeyValueMetadata::Make({"doris_type"}, {type_name});
+ return std::make_shared<arrow::Field>(field_name, arrow_type, is_nullable, metadata);
}
-Status get_arrow_schema_from_block(const Block& block, std::shared_ptr<arrow::Schema>* result,
- const std::string& timezone, bool datetime_naive) {
+Status DorisArrowSchemaConvertor::get_arrow_schema_from_block(
+ const Block& block, std::shared_ptr<arrow::Schema>* result) const {
std::vector<std::shared_ptr<arrow::Field>> fields;
for (const auto& type_and_name : block) {
std::shared_ptr<arrow::DataType> arrow_type;
- RETURN_IF_ERROR(
- convert_to_arrow_type(type_and_name.type, &arrow_type, timezone, datetime_naive));
- auto field = create_arrow_field_with_metadata(type_and_name.name, arrow_type,
- type_and_name.type->is_nullable(),
- type_and_name.type->get_primitive_type());
+ RETURN_IF_ERROR(convert_to_arrow_type(type_and_name.type, &arrow_type));
+ auto field = make_field(type_and_name.name, arrow_type, type_and_name.type->is_nullable(),
+ type_and_name.type->get_primitive_type());
fields.push_back(field);
}
*result = arrow::schema(std::move(fields));
return Status::OK();
}
-Status get_arrow_schema_from_expr_ctxs(const VExprContextSPtrs& output_vexpr_ctxs,
- std::shared_ptr<arrow::Schema>* result,
- const std::string& timezone, bool datetime_naive) {
+Status DorisArrowSchemaConvertor::get_arrow_schema(std::shared_ptr<arrow::Schema>* result) const {
+ return get_arrow_schema_from_block(_header, result);
+}
+
+std::string DorisArrowSchemaConvertor::timestamp_timezone(PrimitiveType) const {
+ // Arrow clients expect a timezone name rather than the ISO-8601 UTC alias.
+ return _timezone == "Z" ? "UTC" : _timezone;
+}
+
+Status ArrowFlightSchemaConvertor::convert_to_arrow_type(
+ const DataTypePtr& type, std::shared_ptr<arrow::DataType>* result) const {
+ // Flight always uses native Variant, including recursively converted children.
+ if (type->get_primitive_type() == TYPE_VARIANT) {
+ // Reject by type before reading rows, including empty results and nested legacy leaves.
+ if (dynamic_cast<const DataTypeVariantV2*>(remove_nullable(type).get()) == nullptr) {
+ return Status::NotSupported(
+ "Native Arrow Flight output only supports Variant V2, not legacy Variant; "
+ "cast the result to STRING for text output");
+ }
+ RETURN_IF_ERROR(register_arrow_variant_extension());
+ *result = arrow::extension::variant(
+ arrow::struct_({arrow::field("metadata", arrow::binary(), false),
+ arrow::field("value", arrow::binary(), false)}));
+ return Status::OK();
+ }
+ return DorisArrowSchemaConvertor::convert_to_arrow_type(type, result);
+}
+
+std::string ArrowFlightSchemaConvertor::timestamp_timezone(PrimitiveType type) const {
+ // DATETIMEV2 is wall-clock time; TIMESTAMPTZ must still describe an instant.
+ return type == TYPE_DATETIMEV2 ? "" : DorisArrowSchemaConvertor::timestamp_timezone(type);
+}
+
+std::shared_ptr<arrow::Field> DorisArrowSchemaConvertor::make_field(
+ const std::string& name, const std::shared_ptr<arrow::DataType>& type, bool nullable,
+ PrimitiveType primitive) const {
+ return create_arrow_field_with_metadata(name, type, nullable, primitive);
+}
+
+std::shared_ptr<arrow::Field> DorisArrowSchemaConvertor::make_child_field(
+ const std::string& name, const std::shared_ptr<arrow::DataType>& type, bool nullable,
+ PrimitiveType primitive) const {
+ return create_arrow_field_with_metadata(name, type, nullable, primitive);
+}
+
+std::shared_ptr<arrow::Field> LegacyArrowFlightSchemaConvertor::make_field(
+ const std::string& name, const std::shared_ptr<arrow::DataType>& type, bool nullable,
+ PrimitiveType primitive) const {
+ if (primitive == TYPE_JSONB || primitive == TYPE_VARIANT) {
+ return arrow::field(name, type, nullable);
+ }
+ return DorisArrowSchemaConvertor::make_field(name, type, nullable, primitive);
+}
+
+std::shared_ptr<arrow::Field> LegacyArrowFlightSchemaConvertor::make_child_field(
+ const std::string& name, const std::shared_ptr<arrow::DataType>& type, bool nullable,
+ PrimitiveType) const {
+ return arrow::field(name, type, nullable);
+}
+
+Status DorisArrowSchemaConvertor::get_arrow_schema_from_expr_ctxs(
+ const VExprContextSPtrs& output_vexpr_ctxs, std::shared_ptr<arrow::Schema>* result) const {
std::vector<std::shared_ptr<arrow::Field>> fields;
for (int i = 0; i < output_vexpr_ctxs.size(); i++) {
std::shared_ptr<arrow::DataType> arrow_type;
auto root_expr = output_vexpr_ctxs.at(i)->root();
- RETURN_IF_ERROR(convert_to_arrow_type(root_expr->data_type(), &arrow_type, timezone,
- datetime_naive));
+ RETURN_IF_ERROR(convert_to_arrow_type(root_expr->data_type(), &arrow_type));
auto field_name = root_expr->is_slot_ref() && !root_expr->expr_label().empty()
? root_expr->expr_label()
: fmt::format("{}_{}", root_expr->data_type()->get_name(), i);
- auto field =
- create_arrow_field_with_metadata(field_name, arrow_type, root_expr->is_nullable(),
- root_expr->data_type()->get_primitive_type());
- fields.push_back(field);
+ auto field = make_field(field_name, arrow_type, root_expr->is_nullable(),
+ root_expr->data_type()->get_primitive_type());
+ fields.push_back(std::move(field));
}
*result = arrow::schema(std::move(fields));
return Status::OK();
@@ -298,18 +372,17 @@ Status serialize_record_batch(const arrow::RecordBatch& record_batch, std::strin
}
Status serialize_arrow_schema(std::shared_ptr<arrow::Schema>* schema, std::string* result) {
- std::vector<std::shared_ptr<arrow::Array>> columns;
- columns.reserve((*schema)->num_fields());
- for (const auto& field : (*schema)->fields()) {
- auto empty = arrow::MakeArrayOfNull(field->type(), 0);
- if (!empty.ok()) {
- return Status::InternalError("serialize_arrow_schema failed, reason: {}",
- empty.status().ToString());
- }
- columns.push_back(*empty);
- }
- auto batch = arrow::RecordBatch::Make(*schema, 0, columns);
- return serialize_record_batch(*batch, result);
+ // Schema RPC readers only consume the IPC schema. Building an empty batch would require
+ // nested extension builders, which Arrow does not provide for ARRAY/MAP/STRUCT<VARIANT>.
+ std::shared_ptr<arrow::io::BufferOutputStream> sink;
+ RETURN_DORIS_STATUS_IF_RESULT_ERROR(sink, arrow::io::BufferOutputStream::Create());
+ std::shared_ptr<arrow::ipc::RecordBatchWriter> writer;
+ RETURN_DORIS_STATUS_IF_RESULT_ERROR(writer, arrow::ipc::MakeStreamWriter(sink.get(), *schema));
+ RETURN_DORIS_STATUS_IF_ERROR(writer->Close());
+ std::shared_ptr<arrow::Buffer> buffer;
+ RETURN_DORIS_STATUS_IF_RESULT_ERROR(buffer, sink->Finish());
+ *result = buffer->ToString();
+ return Status::OK();
}
} // namespace doris
diff --git a/be/src/format/arrow/arrow_row_batch.h b/be/src/format/arrow/arrow_row_batch.h
index d5ba5cb0ed0..debe570fe10 100644
--- a/be/src/format/arrow/arrow_row_batch.h
+++ b/be/src/format/arrow/arrow_row_batch.h
@@ -19,6 +19,7 @@
#include <memory>
#include <string>
+#include <utility>
#include "common/status.h"
#include "core/block/block.h"
@@ -43,23 +44,77 @@ constexpr size_t MAX_ARROW_UTF8 = (1ULL << 31); // 2G
class RowDescriptor;
-// datetime_naive only controls how Doris DATETIMEV2 is represented in the output Arrow schema.
-// When enabled, DATETIMEV2 is mapped to a timestamp without a timezone to preserve its wall-clock
-// semantics. TIMESTAMPTZ remains timezone-aware, and Arrow-to-Doris conversions are unaffected.
-Status convert_to_arrow_type(const DataTypePtr& type, std::shared_ptr<arrow::DataType>* result,
- const std::string& timezone, bool datetime_naive = false);
+// Each protocol owns its schema source and conversion rules. Table formats retain their
+// authoritative schemas instead of reconstructing field IDs or physical layouts from Doris types.
+class ArrowSchemaConvertor {
+public:
+ virtual ~ArrowSchemaConvertor() = default;
+ virtual Status get_arrow_schema(std::shared_ptr<arrow::Schema>* result) const = 0;
+};
+
+class DorisArrowSchemaConvertor : public ArrowSchemaConvertor {
+public:
+ explicit DorisArrowSchemaConvertor(std::string timezone) : _timezone(std::move(timezone)) {}
+ DorisArrowSchemaConvertor(const Block& header, std::string timezone)
+ : _header(header.clone_empty()), _timezone(std::move(timezone)) {}
+
+ Status get_arrow_schema(std::shared_ptr<arrow::Schema>* result) const override;
+ Status get_arrow_schema_from_block(const Block& block,
+ std::shared_ptr<arrow::Schema>* result) const;
+ Status get_arrow_schema_from_expr_ctxs(const VExprContextSPtrs& output_vexpr_ctxs,
+ std::shared_ptr<arrow::Schema>* result) const;
+ virtual Status convert_to_arrow_type(const DataTypePtr& type,
+ std::shared_ptr<arrow::DataType>* result) const;
+
+protected:
+ virtual std::string timestamp_timezone(PrimitiveType type) const;
+ virtual std::shared_ptr<arrow::Field> make_field(const std::string& name,
+ const std::shared_ptr<arrow::DataType>& type,
+ bool nullable, PrimitiveType primitive) const;
+ virtual std::shared_ptr<arrow::Field> make_child_field(
+ const std::string& name, const std::shared_ptr<arrow::DataType>& type, bool nullable,
+ PrimitiveType primitive) const;
+
+private:
+ Block _header;
+ const std::string _timezone;
+};
+
+class ArrowFlightSchemaConvertor : public DorisArrowSchemaConvertor {
+public:
+ explicit ArrowFlightSchemaConvertor(std::string timezone)
+ : DorisArrowSchemaConvertor(std::move(timezone)) {}
+ ArrowFlightSchemaConvertor(const Block& header, std::string timezone)
+ : DorisArrowSchemaConvertor(header, std::move(timezone)) {}
+
+ Status convert_to_arrow_type(const DataTypePtr& type,
+ std::shared_ptr<arrow::DataType>* result) const override;
+
+protected:
+ std::string timestamp_timezone(PrimitiveType type) const override;
+};
+
+// Old FEs require the pre-capability metadata layout, including metadata-free nested fields.
+class LegacyArrowFlightSchemaConvertor final : public ArrowFlightSchemaConvertor {
+public:
+ using ArrowFlightSchemaConvertor::ArrowFlightSchemaConvertor;
+
+protected:
+ std::shared_ptr<arrow::Field> make_field(const std::string& name,
+ const std::shared_ptr<arrow::DataType>& type,
+ bool nullable, PrimitiveType primitive) const override;
+ std::shared_ptr<arrow::Field> make_child_field(const std::string& name,
+ const std::shared_ptr<arrow::DataType>& type,
+ bool nullable,
+ PrimitiveType primitive) const override;
+};
+
+Status register_arrow_variant_extension();
std::shared_ptr<arrow::Field> create_arrow_field_with_metadata(
const std::string& field_name, const std::shared_ptr<arrow::DataType>& arrow_type,
bool is_nullable, PrimitiveType primitive_type);
-Status get_arrow_schema_from_block(const Block& block, std::shared_ptr<arrow::Schema>* result,
- const std::string& timezone, bool datetime_naive = false);
-
-Status get_arrow_schema_from_expr_ctxs(const VExprContextSPtrs& output_vexpr_ctxs,
- std::shared_ptr<arrow::Schema>* result,
- const std::string& timezone, bool datetime_naive = false);
-
Status serialize_record_batch(const arrow::RecordBatch& record_batch, std::string* result);
Status serialize_arrow_schema(std::shared_ptr<arrow::Schema>* schema, std::string* result);
diff --git a/be/src/format/parquet/parquet_arrow_block_convertor.cpp b/be/src/format/parquet/parquet_arrow_block_convertor.cpp
index ff426efad3a..89b944f9294 100644
--- a/be/src/format/parquet/parquet_arrow_block_convertor.cpp
+++ b/be/src/format/parquet/parquet_arrow_block_convertor.cpp
@@ -23,7 +23,13 @@
namespace doris {
-Status ParquetArrowBlockConvertor::init() {
+std::string ParquetArrowSchemaConvertor::timestamp_timezone(PrimitiveType type) const {
+ return type == TYPE_DATETIMEV2 && !_enable_int96_timestamps
+ ? ""
+ : DorisArrowSchemaConvertor::timestamp_timezone(type);
+}
+
+Status ParquetArrowSchemaConvertor::get_arrow_schema(std::shared_ptr<arrow::Schema>* result) const {
if (_types.size() != _names.size()) {
return Status::InvalidArgument("Parquet column names and types must have the same size");
}
@@ -33,14 +39,18 @@ Status ParquetArrowBlockConvertor::init() {
// INT96 normalization and schema construction must use the same instance's timezone.
for (size_t i = 0; i < _types.size(); ++i) {
std::shared_ptr<arrow::DataType> type;
- RETURN_IF_ERROR(
- convert_to_arrow_type(_types[i], &type, _timezone_name, !_enable_int96_timestamps));
+ RETURN_IF_ERROR(convert_to_arrow_type(_types[i], &type));
fields.emplace_back(arrow::field(_names[i], type, _types[i]->is_nullable()));
}
- _arrow_schema = arrow::schema(std::move(fields));
+ *result = arrow::schema(std::move(fields));
return Status::OK();
}
+Status ParquetArrowBlockConvertor::init() {
+ return ParquetArrowSchemaConvertor(_types, _names, _timezone_name, _enable_int96_timestamps)
+ .get_arrow_schema(&_arrow_schema);
+}
+
Status ParquetArrowBlockConvertor::write_column(const DataTypePtr& type, const DataTypeSerDe& serde,
const IColumn& column, const NullMap* null_map,
const std::shared_ptr<arrow::Field>& field,
diff --git a/be/src/format/parquet/parquet_arrow_block_convertor.h b/be/src/format/parquet/parquet_arrow_block_convertor.h
index 24a490a565b..28744817f9c 100644
--- a/be/src/format/parquet/parquet_arrow_block_convertor.h
+++ b/be/src/format/parquet/parquet_arrow_block_convertor.h
@@ -18,9 +18,29 @@
#pragma once
#include "format/arrow/arrow_block_convertor.h"
+#include "format/arrow/arrow_row_batch.h"
namespace doris {
+class ParquetArrowSchemaConvertor final : public DorisArrowSchemaConvertor {
+public:
+ ParquetArrowSchemaConvertor(DataTypes types, std::vector<std::string> names,
+ std::string timezone, bool enable_int96_timestamps)
+ : DorisArrowSchemaConvertor(std::move(timezone)),
+ _types(std::move(types)),
+ _names(std::move(names)),
+ _enable_int96_timestamps(enable_int96_timestamps) {}
+ Status get_arrow_schema(std::shared_ptr<arrow::Schema>* result) const override;
+
+protected:
+ std::string timestamp_timezone(PrimitiveType type) const override;
+
+private:
+ const DataTypes _types;
+ const std::vector<std::string> _names;
+ const bool _enable_int96_timestamps;
+};
+
class ParquetArrowBlockConvertor : public ArrowBlockConvertor {
public:
ParquetArrowBlockConvertor(DataTypes types, std::vector<std::string> names,
diff --git a/be/src/format/table/iceberg/arrow_schema_util.cpp b/be/src/format/table/iceberg/arrow_schema_util.cpp
index db8daf04cac..42061eedea6 100644
--- a/be/src/format/table/iceberg/arrow_schema_util.cpp
+++ b/be/src/format/table/iceberg/arrow_schema_util.cpp
@@ -23,25 +23,38 @@
namespace doris::iceberg {
-const char* ArrowSchemaUtil::PARQUET_FIELD_ID = "PARQUET:field_id";
-const char* ArrowSchemaUtil::ORIGINAL_TYPE = "originalType";
-const char* ArrowSchemaUtil::MAP_TYPE_VALUE = "mapType";
-const char* ArrowSchemaUtil::UUID_TYPE_VALUE = "uuid";
-
-Status ArrowSchemaUtil::convert(const Schema* schema, const std::string& timezone,
- std::vector<std::shared_ptr<arrow::Field>>& fields,
- bool use_uuid_extension) {
- for (const auto& column : schema->columns()) {
+const char* IcebergArrowSchemaConvertor::PARQUET_FIELD_ID = "PARQUET:field_id";
+const char* IcebergArrowSchemaConvertor::ORIGINAL_TYPE = "originalType";
+const char* IcebergArrowSchemaConvertor::MAP_TYPE_VALUE = "mapType";
+const char* IcebergArrowSchemaConvertor::UUID_TYPE_VALUE = "uuid";
+
+Status IcebergArrowSchemaConvertor::get_arrow_schema(std::shared_ptr<arrow::Schema>* result) const {
+ // Field IDs, UUID annotations and timestamp bindings come from the pinned table schema.
+ std::vector<std::shared_ptr<arrow::Field>> fields;
+ RETURN_IF_ERROR(convert_fields(fields));
+ auto schema = arrow::schema(std::move(fields));
+ if (!_schema_json.empty()) {
+ schema = schema->WithMetadata(
+ arrow::KeyValueMetadata::Make({"iceberg.schema"}, {_schema_json}));
+ }
+ *result = std::move(schema);
+ return Status::OK();
+}
+
+Status IcebergArrowSchemaConvertor::convert_fields(
+ std::vector<std::shared_ptr<arrow::Field>>& fields) const {
+ for (const auto& column : _schema.columns()) {
std::shared_ptr<arrow::Field> arrow_field;
- RETURN_IF_ERROR(convert_to(column, &arrow_field, timezone, use_uuid_extension));
+ RETURN_IF_ERROR(convert_to_arrow_field(column, &arrow_field));
fields.push_back(arrow_field);
}
return Status::OK();
}
-Status ArrowSchemaUtil::convert_to(const iceberg::NestedField& field,
- std::shared_ptr<arrow::Field>* arrow_field,
- const std::string& timezone, bool use_uuid_extension) {
+// Keep the exhaustive type mapping together so protocol-specific overrides remain visible.
+// NOLINTNEXTLINE(readability-function-size)
+Status IcebergArrowSchemaConvertor::convert_to_arrow_field(
+ const iceberg::NestedField& field, std::shared_ptr<arrow::Field>* arrow_field) const {
std::shared_ptr<arrow::DataType> arrow_type;
std::unordered_map<std::string, std::string> metadata;
metadata[PARQUET_FIELD_ID] = std::to_string(field.field_id());
@@ -73,7 +86,7 @@ Status ArrowSchemaUtil::convert_to(const iceberg::NestedField& field,
case iceberg::TypeID::TIMESTAMP: {
iceberg::TimestampType* t_type = static_cast<iceberg::TimestampType*>(field.field_type());
- std::string real_tz = t_type->should_adjust_to_utc() ? timezone : "";
+ std::string real_tz = t_type->should_adjust_to_utc() ? _timezone : "";
arrow_type = std::make_shared<arrow::TimestampType>(arrow::TimeUnit::MICRO, real_tz);
break;
}
@@ -88,7 +101,7 @@ Status ArrowSchemaUtil::convert_to(const iceberg::NestedField& field,
case iceberg::TypeID::UUID:
metadata[ORIGINAL_TYPE] = UUID_TYPE_VALUE;
- arrow_type = use_uuid_extension ? arrow::extension::uuid() : arrow::fixed_size_binary(16);
+ arrow_type = _use_uuid_extension ? arrow::extension::uuid() : arrow::fixed_size_binary(16);
break;
case iceberg::TypeID::FIXED: {
@@ -108,7 +121,7 @@ Status ArrowSchemaUtil::convert_to(const iceberg::NestedField& field,
StructType* st = field.field_type()->as_struct_type();
for (const auto& column : st->fields()) {
std::shared_ptr<arrow::Field> element_field;
- RETURN_IF_ERROR(convert_to(column, &element_field, timezone, use_uuid_extension));
+ RETURN_IF_ERROR(convert_to_arrow_field(column, &element_field));
element_fields.push_back(element_field);
}
arrow_type = arrow::struct_(element_fields);
@@ -118,8 +131,7 @@ Status ArrowSchemaUtil::convert_to(const iceberg::NestedField& field,
case iceberg::TypeID::LIST: {
std::shared_ptr<arrow::Field> item_field;
ListType* list_type = field.field_type()->as_list_type();
- RETURN_IF_ERROR(
- convert_to(list_type->element_field(), &item_field, timezone, use_uuid_extension));
+ RETURN_IF_ERROR(convert_to_arrow_field(list_type->element_field(), &item_field));
arrow_type = arrow::list(item_field);
break;
}
@@ -128,10 +140,8 @@ Status ArrowSchemaUtil::convert_to(const iceberg::NestedField& field,
std::shared_ptr<arrow::Field> key_field;
std::shared_ptr<arrow::Field> value_field;
MapType* map_type = field.field_type()->as_map_type();
- RETURN_IF_ERROR(
- convert_to(map_type->key_field(), &key_field, timezone, use_uuid_extension));
- RETURN_IF_ERROR(
- convert_to(map_type->value_field(), &value_field, timezone, use_uuid_extension));
+ RETURN_IF_ERROR(convert_to_arrow_field(map_type->key_field(), &key_field));
+ RETURN_IF_ERROR(convert_to_arrow_field(map_type->value_field(), &value_field));
metadata[ORIGINAL_TYPE] = MAP_TYPE_VALUE;
arrow_type = std::make_shared<arrow::MapType>(key_field, value_field);
break;
diff --git a/be/src/format/table/iceberg/arrow_schema_util.h b/be/src/format/table/iceberg/arrow_schema_util.h
index 6d9635ba30c..5ea092bf335 100644
--- a/be/src/format/table/iceberg/arrow_schema_util.h
+++ b/be/src/format/table/iceberg/arrow_schema_util.h
@@ -19,16 +19,21 @@
#include <arrow/type.h>
+#include "format/arrow/arrow_row_batch.h"
#include "format/table/iceberg/schema.h"
namespace doris::iceberg {
-class ArrowSchemaUtil {
+class IcebergArrowSchemaConvertor final : public ArrowSchemaConvertor {
public:
- // Use the UUID extension for Parquet schemas; batch conversion uses fixed-size storage.
- static Status convert(const Schema* schema, const std::string& timezone,
- std::vector<std::shared_ptr<arrow::Field>>& fields,
- bool use_uuid_extension = false);
+ IcebergArrowSchemaConvertor(const Schema& schema, std::string timezone,
+ std::string schema_json = {}, bool use_uuid_extension = false)
+ : _schema(schema),
+ _timezone(std::move(timezone)),
+ _schema_json(std::move(schema_json)),
+ _use_uuid_extension(use_uuid_extension) {}
+ Status get_arrow_schema(std::shared_ptr<arrow::Schema>* result) const override;
+ Status convert_fields(std::vector<std::shared_ptr<arrow::Field>>& fields) const;
private:
static const char* PARQUET_FIELD_ID;
@@ -36,9 +41,13 @@ private:
static const char* MAP_TYPE_VALUE;
static const char* UUID_TYPE_VALUE;
- static Status convert_to(const iceberg::NestedField& field,
- std::shared_ptr<arrow::Field>* arrow_field,
- const std::string& timezone, bool use_uuid_extension);
+ Status convert_to_arrow_field(const iceberg::NestedField& field,
+ std::shared_ptr<arrow::Field>* arrow_field) const;
+ const Schema& _schema;
+ const std::string _timezone;
+ const std::string _schema_json;
+ // Parquet writes retain UUID annotations while storage-only schema callers keep fixed binary.
+ const bool _use_uuid_extension;
};
} // namespace doris::iceberg
diff --git a/be/src/format/table/iceberg/iceberg_arrow_block_convertor.cpp b/be/src/format/table/iceberg/iceberg_arrow_block_convertor.cpp
index b004462b0ee..00e16d0b621 100644
--- a/be/src/format/table/iceberg/iceberg_arrow_block_convertor.cpp
+++ b/be/src/format/table/iceberg/iceberg_arrow_block_convertor.cpp
@@ -29,16 +29,8 @@ Status IcebergArrowBlockConvertor::init() {
if (_schema == nullptr) {
return ArrowBlockConvertor::init();
}
- // Field IDs, Variant storage and timestamp bindings must share the same target schema.
- std::vector<std::shared_ptr<arrow::Field>> fields;
- // Arrow consumers need the declared label, not cctz's internal fixed-offset name.
- RETURN_IF_ERROR(ArrowSchemaUtil::convert(_schema, _timezone_name, fields, true));
- _arrow_schema = arrow::schema(std::move(fields));
- if (!_schema_json.empty()) {
- _arrow_schema = _arrow_schema->WithMetadata(
- arrow::KeyValueMetadata::Make({"iceberg.schema"}, {_schema_json}));
- }
- return Status::OK();
+ return IcebergArrowSchemaConvertor(*_schema, _timezone_name, _schema_json, true)
+ .get_arrow_schema(&_arrow_schema);
}
Status IcebergArrowBlockConvertor::write_column(const std::shared_ptr<const IDataType>& type,
diff --git a/be/src/format/table/paimon/paimon_arrow_block_convertor.cpp b/be/src/format/table/paimon/paimon_arrow_block_convertor.cpp
index 993cddb05a0..90aef0092cb 100644
--- a/be/src/format/table/paimon/paimon_arrow_block_convertor.cpp
+++ b/be/src/format/table/paimon/paimon_arrow_block_convertor.cpp
@@ -24,10 +24,7 @@
namespace doris::paimon {
-Status PaimonArrowBlockConvertor::init() {
- if (_arrow_schema != nullptr) {
- return Status::OK();
- }
+Status PaimonArrowSchemaConvertor::get_arrow_schema(std::shared_ptr<arrow::Schema>* result) const {
// Decode the pinned table schema here; rebuilding it from Doris types would lose
// nested nullability, timestamp precision and Paimon's physical Variant layout.
auto input = std::make_shared<arrow::io::BufferReader>(
@@ -37,7 +34,16 @@ Status PaimonArrowBlockConvertor::init() {
return Status::InvalidArgument("Failed to deserialize Paimon Arrow schema: {}",
reader.status().ToString());
}
- _arrow_schema = reader.ValueOrDie()->schema();
+ *result = reader.ValueOrDie()->schema();
+ return Status::OK();
+}
+
+Status PaimonArrowBlockConvertor::init() {
+ if (_arrow_schema != nullptr) {
+ return Status::OK();
+ }
+ RETURN_IF_ERROR(
+ PaimonArrowSchemaConvertor(_serialized_schema).get_arrow_schema(&_arrow_schema));
_serialized_schema.clear();
return Status::OK();
}
diff --git a/be/src/format/table/paimon/paimon_arrow_block_convertor.h b/be/src/format/table/paimon/paimon_arrow_block_convertor.h
index caa6501e0a6..8dd159dca3f 100644
--- a/be/src/format/table/paimon/paimon_arrow_block_convertor.h
+++ b/be/src/format/table/paimon/paimon_arrow_block_convertor.h
@@ -18,9 +18,20 @@
#pragma once
#include "format/arrow/arrow_block_convertor.h"
+#include "format/arrow/arrow_row_batch.h"
namespace doris::paimon {
+class PaimonArrowSchemaConvertor final : public ArrowSchemaConvertor {
+public:
+ explicit PaimonArrowSchemaConvertor(std::string serialized_schema)
+ : _serialized_schema(std::move(serialized_schema)) {}
+ Status get_arrow_schema(std::shared_ptr<arrow::Schema>* result) const override;
+
+private:
+ const std::string _serialized_schema;
+};
+
// Paimon reads use native Parquet/ORC or the JNI scanner, which apply the table's
// timestamp and nested-type semantics. This adapter implements the Arrow write protocol only;
// convert_from_arrow deliberately inherits NotSupported instead of a generic SerDe fallback.
diff --git a/be/src/format_v2/table/adbc_reader.cpp b/be/src/format_v2/table/adbc_reader.cpp
index 0ab86359bae..9d518081898 100644
--- a/be/src/format_v2/table/adbc_reader.cpp
+++ b/be/src/format_v2/table/adbc_reader.cpp
@@ -42,6 +42,7 @@
#include "core/data_type/data_type_struct.h"
#include "core/data_type/primitive_type.h"
#include "core/data_type_serde/data_type_serde.h"
+#include "core/data_type_serde/data_type_variant_v2_serde.h"
#include "format/arrow/arrow_array_normalizer.h"
#include "format/parquet/arrow_memory_pool.h"
#include "format_v2/column_mapper.h"
@@ -138,6 +139,11 @@ Status validate_arrow_type(const arrow::DataType& arrow_type, const DataTypePtr&
const auto primitive_type = nested_doris_type->get_primitive_type();
const auto mismatch = [&]() { return arrow_type_mismatch(path, arrow_type, doris_type); };
+ // The FE maps the Variant extension to the V2 carrier, not to its binary storage fields.
+ if (primitive_type == TYPE_VARIANT) {
+ return DataTypeVariantV2SerDe::is_native_arrow_type(arrow_type) ? Status::OK() : mismatch();
+ }
+
switch (arrow_type.id()) {
case arrow::Type::BOOL:
return primitive_type == TYPE_BOOLEAN ? Status::OK() : mismatch();
diff --git a/be/src/service/arrow_flight/arrow_flight_batch_reader.cpp b/be/src/service/arrow_flight/arrow_flight_batch_reader.cpp
index 73761281799..0a367692550 100644
--- a/be/src/service/arrow_flight/arrow_flight_batch_reader.cpp
+++ b/be/src/service/arrow_flight/arrow_flight_batch_reader.cpp
@@ -189,6 +189,7 @@ arrow::Status ArrowFlightBatchRemoteReader::_fetch_schema() {
st = Status::create(callback->response_->status());
ARROW_RETURN_NOT_OK(to_arrow_status(st));
+ ARROW_RETURN_NOT_OK(to_arrow_status(register_arrow_variant_extension()));
if (callback->response_->has_schema() && !callback->response_->schema().empty()) {
auto input =
arrow::io::BufferReader::FromString(std::string(callback->response_->schema()));
diff --git a/be/src/udf/python/python_udf_meta.cpp b/be/src/udf/python/python_udf_meta.cpp
index 4e42197aab5..09ee242ca4e 100644
--- a/be/src/udf/python/python_udf_meta.cpp
+++ b/be/src/udf/python/python_udf_meta.cpp
@@ -34,9 +34,10 @@ namespace doris {
Status PythonUDFMeta::convert_types_to_schema(const DataTypes& types, const std::string& timezone,
std::shared_ptr<arrow::Schema>* schema) {
arrow::SchemaBuilder builder;
+ const DorisArrowSchemaConvertor schema_convertor(timezone);
for (size_t i = 0; i < types.size(); ++i) {
std::shared_ptr<arrow::DataType> arrow_type;
- RETURN_IF_ERROR(convert_to_arrow_type(types[i], &arrow_type, timezone));
+ RETURN_IF_ERROR(schema_convertor.convert_to_arrow_type(types[i], &arrow_type));
std::shared_ptr<arrow::Field> field = std::make_shared<arrow::Field>(
"arg" + std::to_string(i), arrow_type, types[i]->is_nullable());
RETURN_DORIS_STATUS_IF_ERROR(builder.AddField(field));
diff --git a/be/test/core/data_type/common_data_type_serder_test.h b/be/test/core/data_type/common_data_type_serder_test.h
index 70436b7d7ad..0f9a1165da2 100644
--- a/be/test/core/data_type/common_data_type_serder_test.h
+++ b/be/test/core/data_type/common_data_type_serder_test.h
@@ -367,7 +367,9 @@ public:
static std::shared_ptr<arrow::RecordBatch> serialize_arrow(
const std::shared_ptr<Block>& block) {
std::shared_ptr<arrow::Schema> block_arrow_schema;
- EXPECT_EQ(get_arrow_schema_from_block(*block, &block_arrow_schema, "UTC"), Status::OK());
+ EXPECT_EQ(DorisArrowSchemaConvertor("UTC").get_arrow_schema_from_block(*block,
+ &block_arrow_schema),
+ Status::OK());
std::cout << "schema: " << block_arrow_schema->ToString(true) << std::endl;
// convert block to arrow
std::shared_ptr<arrow::RecordBatch> result;
diff --git a/be/test/core/data_type/data_type_uuid_test.cpp b/be/test/core/data_type/data_type_uuid_test.cpp
index cdfca4d9e8f..378b5e98089 100644
--- a/be/test/core/data_type/data_type_uuid_test.cpp
+++ b/be/test/core/data_type/data_type_uuid_test.cpp
@@ -522,7 +522,7 @@ TEST_F(DataTypeUUIDTest, ArrowNestedBinaryIpcRoundTrip) {
types.push_back(data_type);
}
std::shared_ptr<arrow::Schema> schema;
- ASSERT_TRUE(get_arrow_schema_from_block(block, &schema, "UTC").ok());
+ ASSERT_TRUE(DorisArrowSchemaConvertor("UTC").get_arrow_schema_from_block(block, &schema).ok());
EXPECT_TRUE(schema->field(0)->type()->Equals(arrow::extension::uuid()));
EXPECT_TRUE(schema->field(5)->type()->Equals(arrow::utf8()));
std::string serialized_schema;
@@ -534,9 +534,8 @@ TEST_F(DataTypeUUIDTest, ArrowNestedBinaryIpcRoundTrip) {
EXPECT_TRUE((*schema_reader)->schema()->Equals(*schema, true));
std::shared_ptr<arrow::RecordBatch> schema_batch;
ASSERT_TRUE((*schema_reader)->ReadNext(&schema_batch).ok());
- ASSERT_NE(schema_batch, nullptr);
- EXPECT_EQ(schema_batch->num_rows(), 0);
- ASSERT_TRUE(schema_batch->ValidateFull().ok());
+ // Schema RPCs carry no record batch: nested extensions have no generic empty builders.
+ EXPECT_EQ(schema_batch, nullptr);
cctz::time_zone timezone;
DorisArrowBlockConvertor converter(schema, timezone);
ASSERT_TRUE(converter.init().ok());
@@ -581,7 +580,7 @@ TEST_F(DataTypeUUIDTest, ArrowConstantNilAndAllNullRemainDistinct) {
{type->create_column_const_with_default_value(4), type, "nil"},
{nullable_type->create_column_const_with_default_value(4), nullable_type, "nulls"}};
std::shared_ptr<arrow::Schema> schema;
- ASSERT_TRUE(get_arrow_schema_from_block(source, &schema, "UTC").ok());
+ ASSERT_TRUE(DorisArrowSchemaConvertor("UTC").get_arrow_schema_from_block(source, &schema).ok());
std::shared_ptr<arrow::RecordBatch> batch;
cctz::time_zone timezone;
DorisArrowBlockConvertor converter(schema, timezone);
diff --git a/be/test/core/data_type/data_type_varbinary_test.cpp b/be/test/core/data_type/data_type_varbinary_test.cpp
index 3e4ee03ced2..2ffa85eab85 100644
--- a/be/test/core/data_type/data_type_varbinary_test.cpp
+++ b/be/test/core/data_type/data_type_varbinary_test.cpp
@@ -30,13 +30,18 @@
#include "agent/be_exec_version_manager.h"
#include "common/exception.h"
#include "core/assert_cast.h"
+#include "core/block/block.h"
#include "core/column/column_const.h"
#include "core/column/column_string.h"
#include "core/column/column_varbinary.h"
#include "core/data_type/common_data_type_serder_test.h"
#include "core/data_type/common_data_type_test.h"
+#include "core/data_type/data_type_array.h"
#include "core/data_type/data_type_factory.hpp"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_nullable.h"
#include "core/data_type/data_type_string.h"
+#include "core/data_type/data_type_struct.h"
#include "core/data_type_serde/data_type_serde.h"
#include "core/field.h"
#include "core/string_buffer.hpp"
@@ -287,6 +292,84 @@ TEST_F(DataTypeVarbinaryTest, ToProtobufDefaultLen) {
EXPECT_EQ(scalar.len(), -1);
}
+static void check_block_round_trip(Block& block, int length, bool nullable) {
+ PBlock serialized;
+ size_t uncompressed_bytes = 0;
+ size_t compressed_bytes = 0;
+ int64_t compress_time = 0;
+ ASSERT_TRUE(block.serialize(BeExecVersionManager::get_newest_version(), &serialized,
+ &uncompressed_bytes, &compressed_bytes, &compress_time,
+ segment_v2::CompressionTypePB::SNAPPY)
+ .ok());
+ std::string wire;
+ ASSERT_TRUE(serialized.SerializeToString(&wire));
+ PBlock received;
+ ASSERT_TRUE(received.ParseFromString(wire));
+ Block restored;
+ int64_t decompress_time = 0;
+ ASSERT_TRUE(restored.deserialize(received, &uncompressed_bytes, &decompress_time).ok());
+ const auto& entry = restored.get_by_position(0);
+ ASSERT_EQ(entry.type->is_nullable(), nullable);
+ EXPECT_EQ(assert_cast<const DataTypeVarbinary&>(*remove_nullable(entry.type)).len(), length);
+ ASSERT_EQ(restored.rows(), block.rows());
+ for (size_t row = 0; row < block.rows(); ++row) {
+ EXPECT_EQ((*entry.column)[row], (*block.get_by_position(0).column)[row]);
+ }
+}
+
+TEST_F(DataTypeVarbinaryTest, BlockRoundTripPreservesDeclaredLength) {
+ for (int length : {-1, 16, 32}) {
+ for (bool nullable : {false, true}) {
+ SCOPED_TRACE(::testing::Message() << "length=" << length << ", nullable=" << nullable);
+ DataTypePtr type = std::make_shared<DataTypeVarbinary>(length);
+ if (nullable) {
+ type = make_nullable(type);
+ }
+ auto column = type->create_column();
+ const std::string value = make_bytes(16);
+ column->insert(Field::create_field<TYPE_VARBINARY>(StringView(value)));
+ if (nullable) {
+ column->insert_default();
+ }
+ Block block {{std::move(column), type, "binary_value"}};
+ check_block_round_trip(block, length, nullable);
+ }
+ }
+}
+
+TEST_F(DataTypeVarbinaryTest, NestedColumnMetadataPreservesDeclaredLength) {
+ auto binary16 = std::make_shared<DataTypeVarbinary>(16);
+ auto binary32 = std::make_shared<DataTypeVarbinary>(32);
+ DataTypeStruct type({std::make_shared<DataTypeArray>(make_nullable(binary16)),
+ std::make_shared<DataTypeMap>(binary16, make_nullable(binary32))},
+ {"array_value", "map_value"});
+ PColumnMeta metadata;
+ type.to_pb_column_meta(&metadata);
+ std::string wire;
+ ASSERT_TRUE(metadata.SerializeToString(&wire));
+ PColumnMeta received;
+ ASSERT_TRUE(received.ParseFromString(wire));
+ auto restored = DataTypeFactory::instance().create_data_type(received);
+ const auto& fields = assert_cast<const DataTypeStruct&>(*restored).get_elements();
+ const auto& array = assert_cast<const DataTypeArray&>(*fields[0]);
+ ASSERT_TRUE(array.get_nested_type()->is_nullable());
+ EXPECT_EQ(
+ assert_cast<const DataTypeVarbinary&>(*remove_nullable(array.get_nested_type())).len(),
+ 16);
+ const auto& map = assert_cast<const DataTypeMap&>(*fields[1]);
+ EXPECT_EQ(assert_cast<const DataTypeVarbinary&>(*map.get_key_type()).len(), 16);
+ ASSERT_TRUE(map.get_value_type()->is_nullable());
+ EXPECT_EQ(assert_cast<const DataTypeVarbinary&>(*remove_nullable(map.get_value_type())).len(),
+ 32);
+}
+
+TEST_F(DataTypeVarbinaryTest, LegacyColumnMetadataUsesUnspecifiedLength) {
+ PColumnMeta metadata;
+ metadata.set_type(PGenericType::VARBINARY);
+ auto restored = DataTypeFactory::instance().create_data_type(metadata);
+ EXPECT_EQ(assert_cast<const DataTypeVarbinary&>(*restored).len(), -1);
+}
+
TEST_F(DataTypeVarbinaryTest, ProtobufPreservesDeclaredLength) {
PTypeDesc type;
auto* node = type.add_types();
diff --git a/be/test/core/data_type_serde/data_type_serde_arrow_test.cpp b/be/test/core/data_type_serde/data_type_serde_arrow_test.cpp
index 3416bc7ad5a..b3b0d0fd4ab 100644
--- a/be/test/core/data_type_serde/data_type_serde_arrow_test.cpp
+++ b/be/test/core/data_type_serde/data_type_serde_arrow_test.cpp
@@ -562,7 +562,8 @@ void block_converter_test(std::vector<PrimitiveType> cols, int row_num, bool is_
std::shared_ptr<arrow::RecordBatch> record_batch;
std::shared_ptr<arrow::Schema> schema;
Status status = Status::OK();
- status = get_arrow_schema_from_block(*source_block, &schema, TimezoneUtils::default_time_zone);
+ status = DorisArrowSchemaConvertor(TimezoneUtils::default_time_zone)
+ .get_arrow_schema_from_block(*source_block, &schema);
ASSERT_TRUE(status.ok() && schema);
cctz::time_zone default_timezone;
ASSERT_TRUE(
@@ -599,7 +600,8 @@ TEST(DataTypeSerDeArrowTest, DataTypeCollectionSerDeTest) {
TEST(DataTypeSerDeArrowTest, ArrowBlockConvertorReusesBothDirectionsAndValidatesSlices) {
auto source = create_test_block({TYPE_INT, TYPE_STRING, TYPE_VARBINARY}, 4, true);
std::shared_ptr<arrow::Schema> schema;
- ASSERT_TRUE(get_arrow_schema_from_block(*source, &schema, "UTC").ok());
+ ASSERT_TRUE(
+ DorisArrowSchemaConvertor("UTC").get_arrow_schema_from_block(*source, &schema).ok());
ArrowFlightArrowBlockConvertor convertor(schema, cctz::utc_time_zone());
std::shared_ptr<arrow::RecordBatch> batch;
ASSERT_TRUE(convertor.convert_to_arrow(*source, arrow::default_memory_pool(), &batch).ok());
@@ -619,7 +621,7 @@ void expect_target_converter_matches_plain(const std::vector<PrimitiveType>& typ
ArrowConvertorFactory target_converter) {
auto block = create_test_block(types, 4, false);
std::shared_ptr<arrow::Schema> schema;
- ASSERT_TRUE(get_arrow_schema_from_block(*block, &schema, "UTC").ok());
+ ASSERT_TRUE(DorisArrowSchemaConvertor("UTC").get_arrow_schema_from_block(*block, &schema).ok());
std::shared_ptr<arrow::RecordBatch> plain_batch;
ASSERT_TRUE(
@@ -663,7 +665,7 @@ TEST(DataTypeSerDeArrowTest, PaimonCommonScalarTypesUseDeclaredConverter) {
TEST(DataTypeSerDeArrowTest, PlainArrowWritesAggregateStateBinaryTypes) {
auto block = create_test_block({TYPE_HLL, TYPE_BITMAP, TYPE_QUANTILE_STATE}, 3, false);
std::shared_ptr<arrow::Schema> schema;
- ASSERT_TRUE(get_arrow_schema_from_block(*block, &schema, "UTC").ok());
+ ASSERT_TRUE(DorisArrowSchemaConvertor("UTC").get_arrow_schema_from_block(*block, &schema).ok());
std::shared_ptr<arrow::RecordBatch> batch;
Status status = convert_to_arrow_batch_for_test(
*block, schema, arrow::default_memory_pool(), &batch, cctz::utc_time_zone(), 0,
@@ -680,7 +682,7 @@ TEST(DataTypeSerDeArrowTest, PlainArrowWritesAggregateStateBinaryTypes) {
TEST(DataTypeSerDeArrowTest, PlainArrowWritesTimeV2) {
auto block = create_test_block({TYPE_TIMEV2}, 3, false);
std::shared_ptr<arrow::Schema> schema;
- ASSERT_TRUE(get_arrow_schema_from_block(*block, &schema, "UTC").ok());
+ ASSERT_TRUE(DorisArrowSchemaConvertor("UTC").get_arrow_schema_from_block(*block, &schema).ok());
ASSERT_EQ(arrow::Type::DOUBLE, schema->field(0)->type()->id());
std::shared_ptr<arrow::RecordBatch> batch;
@@ -1196,27 +1198,31 @@ TEST(DataTypeSerDeArrowTest, ConvertDateTimeV2ToNaiveArrowType) {
const auto datetime_type = std::make_shared<DataTypeDateTimeV2>(6);
std::shared_ptr<arrow::DataType> arrow_type;
- auto status = convert_to_arrow_type(datetime_type, &arrow_type, "Asia/Shanghai");
+ auto status = DorisArrowSchemaConvertor("Asia/Shanghai")
+ .convert_to_arrow_type(datetime_type, &arrow_type);
ASSERT_TRUE(status.ok()) << status;
auto timestamp_type = std::static_pointer_cast<arrow::TimestampType>(arrow_type);
EXPECT_EQ(arrow::TimeUnit::MICRO, timestamp_type->unit());
EXPECT_EQ("Asia/Shanghai", timestamp_type->timezone());
- status = convert_to_arrow_type(datetime_type, &arrow_type, "Asia/Shanghai", true);
+ status = ArrowFlightSchemaConvertor("Asia/Shanghai")
+ .convert_to_arrow_type(datetime_type, &arrow_type);
ASSERT_TRUE(status.ok()) << status;
timestamp_type = std::static_pointer_cast<arrow::TimestampType>(arrow_type);
EXPECT_EQ(arrow::TimeUnit::MICRO, timestamp_type->unit());
EXPECT_TRUE(timestamp_type->timezone().empty());
const auto timestamptz_type = std::make_shared<DataTypeTimeStampTz>(6);
- status = convert_to_arrow_type(timestamptz_type, &arrow_type, "Asia/Shanghai", true);
+ status = ArrowFlightSchemaConvertor("Asia/Shanghai")
+ .convert_to_arrow_type(timestamptz_type, &arrow_type);
ASSERT_TRUE(status.ok()) << status;
timestamp_type = std::static_pointer_cast<arrow::TimestampType>(arrow_type);
EXPECT_EQ(arrow::TimeUnit::MICRO, timestamp_type->unit());
EXPECT_EQ("Asia/Shanghai", timestamp_type->timezone());
const auto array_type = std::make_shared<DataTypeArray>(datetime_type);
- status = convert_to_arrow_type(array_type, &arrow_type, "Asia/Shanghai", true);
+ status = ArrowFlightSchemaConvertor("Asia/Shanghai")
+ .convert_to_arrow_type(array_type, &arrow_type);
ASSERT_TRUE(status.ok()) << status;
const auto list_type = std::static_pointer_cast<arrow::ListType>(arrow_type);
timestamp_type = std::static_pointer_cast<arrow::TimestampType>(list_type->value_type());
@@ -1235,7 +1241,8 @@ TEST(DataTypeSerDeArrowTest, TimeStampNsArrowRoundTrip) {
make_nullable(std::make_shared<DataTypeTimeStampNs>()), timestamp_ns_array_type});
std::shared_ptr<arrow::DataType> arrow_type;
- auto status = convert_to_arrow_type(timestamp_ns_type, &arrow_type, "Asia/Shanghai");
+ auto status = DorisArrowSchemaConvertor("Asia/Shanghai")
+ .convert_to_arrow_type(timestamp_ns_type, &arrow_type);
ASSERT_TRUE(status.ok()) << status;
auto arrow_timestamp_type = std::static_pointer_cast<arrow::TimestampType>(arrow_type);
EXPECT_EQ(arrow::TimeUnit::NANO, arrow_timestamp_type->unit());
@@ -1306,7 +1313,8 @@ TEST(DataTypeSerDeArrowTest, TimeStampNsArrowRoundTrip) {
timestamp_ns_struct_type, "3"));
std::shared_ptr<arrow::Schema> schema;
- status = get_arrow_schema_from_block(*source_block, &schema, TimezoneUtils::default_time_zone);
+ status = DorisArrowSchemaConvertor(TimezoneUtils::default_time_zone)
+ .get_arrow_schema_from_block(*source_block, &schema);
ASSERT_TRUE(status.ok()) << status;
ASSERT_NE(schema, nullptr);
auto list_type = std::static_pointer_cast<arrow::ListType>(schema->field(1)->type());
@@ -1404,7 +1412,8 @@ TEST(DataTypeSerDeArrowTest, CanonicalizeUtcTimezoneForArrow) {
const auto timestamptz_type = std::make_shared<DataTypeTimeStampTz>(6);
std::shared_ptr<arrow::DataType> arrow_type;
- auto status = convert_to_arrow_type(timestamptz_type, &arrow_type, "Z", true);
+ auto status =
+ ArrowFlightSchemaConvertor("Z").convert_to_arrow_type(timestamptz_type, &arrow_type);
ASSERT_TRUE(status.ok()) << status;
const auto timestamp_type = std::static_pointer_cast<arrow::TimestampType>(arrow_type);
EXPECT_EQ("UTC", timestamp_type->timezone());
@@ -1489,7 +1498,9 @@ TEST(DataTypeSerDeArrowTest, NestedDateTimeV2PlainArrowAcceptsNaiveSchema) {
block.insert(ColumnWithTypeAndName(struct_column->get_ptr(), struct_type, "event_struct"));
std::shared_ptr<arrow::Schema> naive_schema;
- ASSERT_TRUE(get_arrow_schema_from_block(block, &naive_schema, "Asia/Shanghai", true).ok());
+ ASSERT_TRUE(ArrowFlightSchemaConvertor("Asia/Shanghai")
+ .get_arrow_schema_from_block(block, &naive_schema)
+ .ok());
std::shared_ptr<arrow::RecordBatch> naive_batch;
const auto status = convert_to_arrow_batch_for_test(
block, naive_schema, arrow::default_memory_pool(), &naive_batch,
diff --git a/be/test/core/data_type_serde/data_type_variant_v2_serde_input_test.cpp b/be/test/core/data_type_serde/data_type_variant_v2_serde_input_test.cpp
index e9d3d669a17..2bb8b74691c 100644
--- a/be/test/core/data_type_serde/data_type_variant_v2_serde_input_test.cpp
+++ b/be/test/core/data_type_serde/data_type_variant_v2_serde_input_test.cpp
@@ -303,8 +303,9 @@ TEST(DataTypeVariantV2SerdeInputTest, PbAndArrowReadKeepExplicitGuards) {
EXPECT_EQ(serde.write_column_to_pb(*column, values, 0, 0).code(),
ErrorCode::NOT_IMPLEMENTED_ERROR);
EXPECT_EQ(serde.read_column_from_pb(*column, values).code(), ErrorCode::NOT_IMPLEMENTED_ERROR);
+ // Native Arrow reads are supported; a missing array is an invalid request, not a missing codec.
EXPECT_EQ(serde.read_column_from_arrow(*column, nullptr, 0, 0, cctz::utc_time_zone()).code(),
- ErrorCode::NOT_IMPLEMENTED_ERROR);
+ ErrorCode::INVALID_ARGUMENT);
}
} // namespace doris
diff --git a/be/test/core/data_type_serde/data_type_variant_v2_serde_output_test.cpp b/be/test/core/data_type_serde/data_type_variant_v2_serde_output_test.cpp
index 5aa88ab50c6..7f7177a86b9 100644
--- a/be/test/core/data_type_serde/data_type_variant_v2_serde_output_test.cpp
+++ b/be/test/core/data_type_serde/data_type_variant_v2_serde_output_test.cpp
@@ -456,6 +456,21 @@ TEST(DataTypeVariantV2SerdeOutputTest, BinaryStructPreservesEncodedAndTypedBytes
EXPECT_TRUE(typed->is_typed());
}
+TEST(DataTypeVariantV2SerdeOutputTest, PaimonKeepsItsOwnBinaryLayout) {
+ DataTypeVariantV2SerDe serde;
+ auto values = encoded_json({R"({"a":42})"});
+ auto builder = binary_variant_arrow_builder();
+ const auto status = serde.write_column_to_paimon_arrow(
+ nullptr, *values, nullptr, nullptr, builder.get(), 0, 1, cctz::utc_time_zone());
+ ASSERT_TRUE(status.ok()) << status;
+ std::shared_ptr<arrow::Array> array;
+ ASSERT_TRUE(builder->Finish(&array).ok());
+ const auto& storage = assert_cast<const arrow::StructArray&>(*array);
+ const auto& binary = assert_cast<const arrow::BinaryArray&>(*storage.field(0));
+ auto expected = values->get_value_ref(0).value;
+ EXPECT_EQ(binary.GetView(0), std::string_view(expected.data, expected.size));
+}
+
TEST(DataTypeVariantV2SerdeOutputTest, BinaryStructRejectsUnsupportedPrimitive) {
DataTypeVariantV2SerDe serde;
VariantBatchBuilder builder(VariantBatchBuilder::ReserveHint {.rows = 1});
diff --git a/be/test/exec/scan/file_scanner_v2_test.cpp b/be/test/exec/scan/file_scanner_v2_test.cpp
index 0afcfdad27b..2b4b4876547 100644
--- a/be/test/exec/scan/file_scanner_v2_test.cpp
+++ b/be/test/exec/scan/file_scanner_v2_test.cpp
@@ -374,20 +374,22 @@ TEST(FileScannerV2Test, SupportedFormatMatrix) {
EXPECT_FALSE(FileScannerV2::is_supported(params, hudi_range_with_delta_logs()));
}
-TEST(FileScannerV2Test, ArrowRejectsVariantBeforeReaderMaterialization) {
- EXPECT_TRUE(
- FileScannerV2::TEST_validate_variant_projection(TFileFormatType::FORMAT_ARROW, false)
- .ok());
- EXPECT_TRUE(
- FileScannerV2::TEST_validate_variant_projection(TFileFormatType::FORMAT_PARQUET, true)
- .ok());
-
- const auto status =
- FileScannerV2::TEST_validate_variant_projection(TFileFormatType::FORMAT_ARROW, true);
+TEST(FileScannerV2Test, ArrowVariantRequiresNativeCarrier) {
+ EXPECT_TRUE(FileScannerV2::TEST_validate_variant_projection(TFileFormatType::FORMAT_ARROW, true,
+ "adbc")
+ .ok());
+ EXPECT_TRUE(FileScannerV2::TEST_validate_variant_projection(TFileFormatType::FORMAT_ARROW,
+ false, "remote_doris")
+ .ok());
+ EXPECT_TRUE(FileScannerV2::TEST_validate_variant_projection(TFileFormatType::FORMAT_PARQUET,
+ true, "iceberg")
+ .ok());
+
+ const auto status = FileScannerV2::TEST_validate_variant_projection(
+ TFileFormatType::FORMAT_ARROW, true, "remote_doris");
EXPECT_TRUE(status.is<ErrorCode::NOT_IMPLEMENTED_ERROR>()) << status;
EXPECT_NE(status.to_string().find(
- "External Variant is supported only for Parquet files in FileScannerV2; "
- "file format ARROW is not supported"),
+ "External Variant Arrow reads require an ADBC native Variant carrier"),
std::string::npos)
<< status;
}
diff --git a/be/test/exec/sink/writer/iceberg/viceberg_table_writer_test.cpp b/be/test/exec/sink/writer/iceberg/viceberg_table_writer_test.cpp
index 6c0821ac1e5..ff1159bcd83 100644
--- a/be/test/exec/sink/writer/iceberg/viceberg_table_writer_test.cpp
+++ b/be/test/exec/sink/writer/iceberg/viceberg_table_writer_test.cpp
@@ -146,8 +146,8 @@ TEST(VIcebergTableWriterTest, StaticNullBinaryPartitionsKeepNullPathsAndCommitVa
protocol.writeString(std::string("null"));
protocol.writeMapEnd();
protocol.writeFieldEnd();
- // Field 19 belongs to NaN counts; the reader skips NULL markers encoded under that ID.
- protocol.writeFieldBegin("static_partition_null_keys", T_SET, 20);
+ // Master reserves field 19 for NULL keys; field 20 carries NaN-count field IDs.
+ protocol.writeFieldBegin("static_partition_null_keys", T_SET, 19);
protocol.writeSetBegin(T_STRING, 1);
protocol.writeString(std::string("key"));
protocol.writeSetEnd();
diff --git a/be/test/format/arrow/arrow_flight_variant_test.cpp b/be/test/format/arrow/arrow_flight_variant_test.cpp
new file mode 100644
index 00000000000..aa0fa70f7d4
--- /dev/null
+++ b/be/test/format/arrow/arrow_flight_variant_test.cpp
@@ -0,0 +1,531 @@
+// 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 <arrow/extension/parquet_variant.h>
+#include <arrow/io/api.h>
+#include <arrow/ipc/api.h>
+#include <gtest/gtest.h>
+
+#include <cmath>
+#include <limits>
+
+#include "core/column/column_array.h"
+#include "core/column/column_const.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_struct.h"
+#include "core/column/variant_v2/column_variant_v2.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_factory.hpp"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_string.h"
+#include "core/data_type/data_type_struct.h"
+#include "core/data_type/data_type_variant_v2.h"
+#include "core/data_type_serde/data_type_serde.h"
+#include "exprs/function/parse/variant_string_parse.h"
+#include "format/arrow/arrow_block_convertor.h"
+#include "format/arrow/arrow_row_batch.h"
+#include "format/table/paimon/paimon_arrow_block_convertor.h"
+#include "util/timezone_utils.h"
+
+namespace doris {
+namespace {
+
+std::shared_ptr<arrow::DataType> native_variant() {
+ return arrow::extension::variant(
+ arrow::struct_({arrow::field("metadata", arrow::binary(), false),
+ arrow::field("value", arrow::binary(), false)}));
+}
+
+MutableColumnPtr documents(const DataTypePtr& type) {
+ auto column = type->create_column();
+ auto serde = type->get_serde();
+ DataTypeSerDe::FormatOptions options;
+ for (std::string json : {R"({"a":[1,null,"x"]})", "null", "42", R"("text")"}) {
+ Slice slice(json.data(), json.size());
+ EXPECT_TRUE(serde->deserialize_one_cell_from_json(*column, slice, options).ok());
+ }
+ return column;
+}
+
+VariantRef value_at(const arrow::Array& array, int row) {
+ const auto& storage = static_cast<const arrow::StructArray&>(
+ *static_cast<const arrow::ExtensionArray&>(array).storage());
+ auto metadata = static_cast<const arrow::BinaryArray&>(*storage.field(0)).GetView(row);
+ auto value = static_cast<const arrow::BinaryArray&>(*storage.field(1)).GetView(row);
+ return {.metadata = {.data = metadata.data(), .size = metadata.size()},
+ .value = {value.data(), value.size()}};
+}
+
+TEST(ArrowFlightVariantTest, NativeResultPreservesValuesAndSqlNulls) {
+ TimezoneUtils::load_timezones_to_cache();
+ ASSERT_TRUE(register_arrow_variant_extension().ok());
+ auto type = std::make_shared<DataTypeVariantV2>();
+ auto nulls = ColumnUInt8::create();
+ nulls->get_data().assign({0, 0, 0, 1});
+ Block block;
+ block.insert(
+ {ColumnNullable::create(documents(type), std::move(nulls)), make_nullable(type), "v"});
+ auto schema = arrow::schema({arrow::field("v", native_variant())});
+ ArrowFlightArrowBlockConvertor converter(schema, cctz::utc_time_zone());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ auto status = converter.convert_to_arrow(block, arrow::default_memory_pool(), &batch);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_TRUE(batch->ValidateFull().ok());
+ EXPECT_EQ(batch->num_rows(), 4);
+ EXPECT_FALSE(batch->column(0)->IsNull(1));
+ EXPECT_TRUE(value_at(*batch->column(0), 1).is_null());
+ EXPECT_EQ(value_at(*batch->column(0), 2).get_int(), 42);
+ EXPECT_TRUE(batch->column(0)->IsNull(3));
+ EXPECT_EQ(value_at(*batch->column(0), 0).basic_type(), VariantBasicType::OBJECT);
+
+ // Extension metadata and storage must survive the same IPC boundary used by Flight.
+ auto output = arrow::io::BufferOutputStream::Create().ValueOrDie();
+ auto writer = arrow::ipc::MakeStreamWriter(output, batch->schema()).ValueOrDie();
+ ASSERT_TRUE(writer->WriteRecordBatch(*batch).ok());
+ ASSERT_TRUE(writer->Close().ok());
+ auto input = std::make_shared<arrow::io::BufferReader>(output->Finish().ValueOrDie());
+ auto reader = arrow::ipc::RecordBatchStreamReader::Open(input).ValueOrDie();
+ auto round_trip = reader->Next().ValueOrDie();
+ EXPECT_TRUE(batch->Equals(*round_trip));
+ ASSERT_TRUE(converter.convert_to_arrow(block, arrow::default_memory_pool(), &batch, 1, 3).ok());
+ EXPECT_TRUE(value_at(*batch->column(0), 0).is_null());
+ EXPECT_EQ(value_at(*batch->column(0), 1).get_int(), 42);
+}
+
+TEST(ArrowFlightVariantTest, NativeReaderPreservesSlicesAndNulls) {
+ auto type = std::make_shared<DataTypeVariantV2>();
+ auto nulls = ColumnUInt8::create();
+ nulls->get_data().assign({0, 0, 0, 1});
+ Block source {
+ {ColumnNullable::create(documents(type), std::move(nulls)), make_nullable(type), "v"}};
+ ArrowFlightArrowBlockConvertor converter(arrow::schema({arrow::field("v", native_variant())}),
+ cctz::utc_time_zone());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ ASSERT_TRUE(converter.convert_to_arrow(source, arrow::default_memory_pool(), &batch).ok());
+ auto result = type->create_column();
+ auto status = type->get_serde()->read_column_from_arrow(*result, batch->column(0).get(), 1, 4,
+ cctz::utc_time_zone());
+ ASSERT_TRUE(status.ok()) << status;
+ const auto& values = assert_cast<const ColumnVariantV2&>(*result);
+ ASSERT_EQ(values.size(), 3);
+ EXPECT_TRUE(values.get_value_ref(0).is_null());
+ EXPECT_EQ(values.get_value_ref(1).get_int(), 42);
+ EXPECT_TRUE(values.get_value_ref(2).is_null());
+ auto storage =
+ static_cast<const arrow::ExtensionArray&>(*batch->column(0)).storage()->Slice(1, 2);
+ auto sliced = type->create_column();
+ status = type->get_serde()->read_column_from_arrow(*sliced, storage.get(), 0, 2,
+ cctz::utc_time_zone());
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(assert_cast<const ColumnVariantV2&>(*sliced).get_value_ref(1).get_int(), 42);
+}
+
+TEST(ArrowFlightVariantTest, NativeReaderRejectsMalformedBytesWithoutAppending) {
+ arrow::BinaryBuilder metadata;
+ arrow::BinaryBuilder values;
+ ASSERT_TRUE(metadata.Append("invalid").ok());
+ ASSERT_TRUE(values.Append("invalid").ok());
+ auto array =
+ arrow::StructArray::Make({metadata.Finish().ValueOrDie(), values.Finish().ValueOrDie()},
+ {arrow::field("metadata", arrow::binary(), false),
+ arrow::field("value", arrow::binary(), false)})
+ .ValueOrDie();
+ auto type = std::make_shared<DataTypeVariantV2>();
+ auto result = documents(type);
+ const auto original_rows = result->size();
+ auto status = type->get_serde()->read_column_from_arrow(*result, array.get(), 0, 1,
+ cctz::utc_time_zone());
+ EXPECT_FALSE(status.ok());
+ EXPECT_EQ(result->size(), original_rows);
+ status = type->get_serde()->read_column_from_arrow(*result, array.get(), 0, 2,
+ cctz::utc_time_zone());
+ EXPECT_FALSE(status.ok());
+ EXPECT_EQ(result->size(), original_rows);
+}
+
+TEST(ArrowFlightVariantTest, IcebergUsesNativeStorageWithoutPaimonFieldOrder) {
+ auto type = std::make_shared<DataTypeVariantV2>();
+ auto column = documents(type);
+ auto storage_type = static_cast<const arrow::ExtensionType&>(*native_variant()).storage_type();
+ auto builder = arrow::MakeBuilder(storage_type, arrow::default_memory_pool()).ValueOrDie();
+ const auto status = type->get_serde()->write_column_to_iceberg_arrow(
+ type, *column, nullptr, arrow::field("v", storage_type), builder.get(), 0,
+ column->size(), cctz::utc_time_zone());
+ ASSERT_TRUE(status.ok()) << status;
+ auto array = builder->Finish().ValueOrDie();
+ auto result = type->create_column();
+ ASSERT_TRUE(type->get_serde()
+ ->read_column_from_arrow(*result, array.get(), 0, column->size(),
+ cctz::utc_time_zone())
+ .ok());
+ EXPECT_EQ(assert_cast<const ColumnVariantV2&>(*result).get_value_ref(2).get_int(), 42);
+}
+
+TEST(ArrowFlightVariantTest, PaimonPreservesNestedVariantProtocol) {
+ auto variant = std::make_shared<DataTypeVariantV2>();
+ auto type = std::make_shared<DataTypeArray>(make_nullable(variant));
+ auto offsets = ColumnArray::ColumnOffsets::create();
+ offsets->get_data().assign({2, 4});
+ Block block {{ColumnArray::create(make_nullable(documents(variant)), std::move(offsets)), type,
+ "a"}};
+ auto storage = arrow::struct_({arrow::field("value", arrow::binary(), false),
+ arrow::field("metadata", arrow::binary(), false)});
+ paimon::PaimonArrowBlockConvertor converter(
+ arrow::schema({arrow::field("a", arrow::list(storage))}), cctz::utc_time_zone());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ const auto status = converter.convert_to_arrow(block, arrow::default_memory_pool(), &batch);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_TRUE(batch->ValidateFull().ok());
+ const auto& array = assert_cast<const arrow::ListArray&>(*batch->column(0));
+ const auto& values = assert_cast<const arrow::StructArray&>(*array.values());
+ EXPECT_EQ(values.type()->field(0)->name(), "value");
+ EXPECT_EQ(values.type()->field(1)->name(), "metadata");
+ EXPECT_EQ(array.value_length(0), 2);
+ EXPECT_EQ(array.value_length(1), 2);
+}
+
+TEST(ArrowFlightVariantTest, SchemaMappingAndConstantScalar) {
+ auto type = std::make_shared<DataTypeVariantV2>();
+ std::shared_ptr<arrow::DataType> mapped;
+ ASSERT_TRUE(DorisArrowSchemaConvertor("UTC").convert_to_arrow_type(type, &mapped).ok());
+ EXPECT_TRUE(mapped->Equals(arrow::utf8()));
+ ASSERT_TRUE(ArrowFlightSchemaConvertor("UTC").convert_to_arrow_type(type, &mapped).ok());
+ EXPECT_TRUE(mapped->Equals(native_variant()));
+ auto column = type->create_column();
+ std::string json = R"("te\"xt\n\u4e2d")";
+ Slice slice(json.data(), json.size());
+ DataTypeSerDe::FormatOptions options;
+ ASSERT_TRUE(type->get_serde()->deserialize_one_cell_from_json(*column, slice, options).ok());
+ Block block;
+ block.insert({ColumnConst::create(std::move(column), 3), type, "v"});
+ ArrowFlightArrowBlockConvertor converter(arrow::schema({arrow::field("v", mapped, false)}),
+ cctz::utc_time_zone());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ auto status = converter.convert_to_arrow(block, arrow::default_memory_pool(), &batch);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_TRUE(batch->ValidateFull().ok());
+ EXPECT_EQ(batch->num_rows(), 3);
+ EXPECT_EQ(value_at(*batch->column(0), 2).get_string().to_string(), "te\"xt\nä¸");
+}
+
+TEST(ArrowFlightVariantTest, NativeSchemaPreservesNestedLogicalMetadata) {
+ auto variant = std::make_shared<DataTypeVariantV2>();
+ const auto nullable = make_nullable(variant);
+ const auto string = std::make_shared<DataTypeString>();
+ DataTypes types {nullable, std::make_shared<DataTypeArray>(nullable),
+ std::make_shared<DataTypeMap>(string, nullable),
+ std::make_shared<DataTypeStruct>(DataTypes {nullable}, Strings {"v"})};
+ Block block;
+ for (size_t i = 0; i < types.size(); ++i) {
+ block.insert({types[i]->create_column(), types[i], std::to_string(i)});
+ }
+ {
+ std::shared_ptr<arrow::Schema> schema;
+ ASSERT_TRUE(ArrowFlightSchemaConvertor(block, "UTC").get_arrow_schema(&schema).ok());
+ const auto& map = static_cast<const arrow::MapType&>(*schema->field(2)->type());
+ for (const auto& field : {schema->field(0), schema->field(1)->type()->field(0),
+ map.item_field(), schema->field(3)->type()->field(0)}) {
+ EXPECT_TRUE(field->type()->Equals(native_variant()));
+ EXPECT_TRUE(field->nullable());
+ ASSERT_NE(nullptr, field->metadata());
+ EXPECT_EQ("VARIANT", field->metadata()->Get("doris_type").ValueOrDie());
+ }
+ EXPECT_FALSE(map.key_field()->nullable());
+ }
+ std::shared_ptr<arrow::Schema> schema;
+ ASSERT_TRUE(DorisArrowSchemaConvertor(block, "UTC").get_arrow_schema(&schema).ok());
+ EXPECT_TRUE(schema->field(0)->type()->Equals(arrow::utf8()));
+}
+
+TEST(ArrowFlightVariantTest, TypedV2AndNestedStructPreserveNonJsonNumbers) {
+ auto numbers = ColumnFloat64::create();
+ numbers->insert_value(std::numeric_limits<double>::quiet_NaN());
+ numbers->insert_value(std::numeric_limits<double>::infinity());
+ auto values = ColumnVariantV2::create_typed(make_nullable(std::move(numbers)),
+ std::make_shared<DataTypeFloat64>());
+ auto variant = std::make_shared<DataTypeVariantV2>();
+ auto type = std::make_shared<DataTypeStruct>(DataTypes {variant}, Strings {"v"});
+ Block block;
+ block.insert({ColumnStruct::create(Columns {std::move(values)}), type, "s"});
+ std::shared_ptr<arrow::DataType> mapped;
+ ASSERT_TRUE(ArrowFlightSchemaConvertor("UTC").convert_to_arrow_type(type, &mapped).ok());
+ ArrowFlightArrowBlockConvertor converter(arrow::schema({arrow::field("s", mapped, false)}),
+ cctz::utc_time_zone());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ auto status = converter.convert_to_arrow(block, arrow::default_memory_pool(), &batch);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_TRUE(batch->ValidateFull().ok());
+ const auto& child = *static_cast<const arrow::StructArray&>(*batch->column(0)).field(0);
+ EXPECT_TRUE(std::isnan(value_at(child, 0).get_double()));
+ EXPECT_TRUE(std::isinf(value_at(child, 1).get_double()));
+}
+
+TEST(ArrowFlightVariantTest, NestedTimezoneAliasesMatchPublishedSchema) {
+ TimezoneUtils::load_timezones_to_cache();
+ auto variant = std::make_shared<DataTypeVariantV2>();
+ auto timestamp = DataTypeFactory::instance().create_data_type(TYPE_TIMESTAMPTZ, false, 0, 6);
+ auto type =
+ std::make_shared<DataTypeStruct>(DataTypes {variant, timestamp}, Strings {"v", "t"});
+ auto times = timestamp->create_column();
+ for (int i = 0; i < 4; ++i) {
+ times->insert_default();
+ }
+ Block block {{ColumnStruct::create(Columns {documents(variant), std::move(times)}), type, "s"}};
+ for (const std::string zone : {"+08:00", "+05:45", "-03:30"}) {
+ cctz::time_zone timezone;
+ ASSERT_TRUE(TimezoneUtils::find_cctz_time_zone(zone, timezone));
+ std::shared_ptr<arrow::DataType> mapped;
+ ASSERT_TRUE(ArrowFlightSchemaConvertor(zone).convert_to_arrow_type(type, &mapped).ok());
+ ArrowFlightArrowBlockConvertor converter(arrow::schema({arrow::field("s", mapped, false)}),
+ timezone);
+ std::shared_ptr<arrow::RecordBatch> batch;
+ auto status = converter.convert_to_arrow(block, arrow::default_memory_pool(), &batch);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_TRUE(batch->schema()->field(0)->type()->Equals(mapped));
+ }
+}
+
+void check_compacted_row_keys(const arrow::RecordBatch& batch, int start, int rows) {
+ size_t metadata_bytes = 0;
+ for (int i = start; i < rows; ++i) {
+ if (i == rows / 2) {
+ EXPECT_TRUE(batch.column(0)->IsNull(i - start));
+ continue;
+ }
+ VariantRef value = value_at(*batch.column(0), i - start);
+ // Internal dictionary sharing must not multiply every other row's keys on the wire.
+ EXPECT_EQ(value.metadata.dict_size(), 1);
+ metadata_bytes += value.metadata.size;
+ VariantRef child;
+ std::string key = std::string(200, 'k') + std::to_string(i);
+ ASSERT_TRUE(value.object_find({key.data(), key.size()}, &child));
+ EXPECT_EQ(child.get_int(), i);
+ }
+ EXPECT_LT(metadata_bytes, static_cast<size_t>(rows - start) * 220);
+}
+
+TEST(ArrowFlightVariantTest, NativeMetadataContainsOnlySelectedRowKeys) {
+ for (int rows : {32, 64}) {
+ SCOPED_TRACE(rows);
+ auto type = std::make_shared<DataTypeVariantV2>();
+ auto column = type->create_column();
+ JsonStringToVariantEncoder encoder;
+ for (int i = 0; i < rows; ++i) {
+ std::string key = std::string(200, 'k') + std::to_string(i);
+ std::string json = "{\"" + key + "\":" + std::to_string(i) + "}";
+ encoder.add_json({json.data(), json.size()});
+ }
+ auto encoded = encoder.finish_batch();
+ ASSERT_EQ(encoded.metadata_ref().dict_size(), rows);
+ assert_cast<ColumnVariantV2&>(*column).insert_encoded_batch(encoded);
+ auto nulls = ColumnUInt8::create();
+ nulls->get_data().resize_fill(rows, 0);
+ nulls->get_data()[rows / 2] = 1;
+ Block block {{ColumnNullable::create(std::move(column), std::move(nulls)),
+ make_nullable(type), "v"}};
+ ArrowFlightArrowBlockConvertor converter(
+ arrow::schema({arrow::field("v", native_variant())}), cctz::utc_time_zone());
+ for (int start : {0, 3}) {
+ std::shared_ptr<arrow::RecordBatch> batch;
+ auto status = converter.convert_to_arrow(block, arrow::default_memory_pool(), &batch,
+ start, rows);
+ ASSERT_TRUE(status.ok()) << status;
+ check_compacted_row_keys(*batch, start, rows);
+ }
+ }
+}
+
+void check_physical_scalar(VariantRef actual, VariantRef expected) {
+ // Integral decimals and floats must retain physical type/scale rather than normalize to integers.
+ EXPECT_EQ(actual.basic_type(), expected.basic_type());
+ if (actual.basic_type() == VariantBasicType::PRIMITIVE) {
+ EXPECT_EQ(actual.primitive_id(), expected.primitive_id());
+ }
+ EXPECT_EQ(actual.value, expected.value);
+}
+
+TEST(ArrowFlightVariantTest, NestedNativeCompactionPreservesPhysicalScalars) {
+ VariantBatchBuilder builder;
+ for (std::string key : {"first", "second"}) {
+ auto row = builder.begin_row();
+ auto object = row.start_object();
+ object.add_key({key.data(), key.size()});
+ row.add_int(1);
+ object.finish();
+ row.finish();
+ }
+ for (int kind = 0; kind < 5; ++kind) {
+ auto row = builder.begin_row();
+ switch (kind) {
+ case 0:
+ row.add_decimal(4200, 2, 4);
+ break;
+ case 1:
+ row.add_float(42.0F);
+ break;
+ case 2:
+ row.add_date(1);
+ break;
+ case 3:
+ row.add_binary({"\0\xff", 2});
+ break;
+ case 4:
+ row.add_string({"42", 2});
+ break;
+ }
+ row.finish();
+ }
+ {
+ auto row = builder.begin_row();
+ auto array = row.start_array();
+ row.add_decimal(4200, 2, 4);
+ row.add_float(42.0F);
+ row.add_string({"42", 2});
+ array.finish();
+ row.finish();
+ }
+ auto encoded = builder.finish_batch();
+ ASSERT_EQ(encoded.metadata_ref().dict_size(), 2);
+ auto values = ColumnVariantV2::create();
+ values->insert_encoded_batch(encoded);
+ auto offsets = ColumnArray::ColumnOffsets::create();
+ for (size_t i = 0; i < encoded.num_rows(); ++i) {
+ offsets->get_data().push_back(i + 1);
+ }
+ auto type = std::make_shared<DataTypeArray>(std::make_shared<DataTypeVariantV2>());
+ Block block {
+ {ColumnArray::create(make_nullable(std::move(values)), std::move(offsets)), type, "a"}};
+ ArrowFlightArrowBlockConvertor converter(
+ arrow::schema({arrow::field("a", arrow::list(native_variant()), false)}),
+ cctz::utc_time_zone());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ auto status = converter.convert_to_arrow(block, arrow::default_memory_pool(), &batch);
+ ASSERT_TRUE(status.ok()) << status;
+ const auto& elements = *static_cast<const arrow::ListArray&>(*batch->column(0)).values();
+ for (int i = 0; i < elements.length(); ++i) {
+ auto actual = value_at(elements, i);
+ EXPECT_EQ(actual.metadata.dict_size(), i < 2 ? 1 : 0);
+ if (i >= 2) {
+ check_physical_scalar(actual, encoded.value_at(i));
+ }
+ }
+}
+
+TEST(ArrowFlightVariantTest, NativeCompactionKeepsTerminalEmptyContainersAtDepthLimit) {
+ for (std::string terminal : {"[]", "{}"}) {
+ std::string json = terminal;
+ for (size_t i = 0; i < VARIANT_MAX_NESTING_DEPTH; ++i) {
+ json = "[" + json + "]";
+ }
+ JsonStringToVariantEncoder encoder;
+ encoder.add_json({json.data(), json.size()});
+ encoder.add_json({R"({"unused":0})", 12});
+ auto encoded = encoder.finish_batch();
+ auto values = ColumnVariantV2::create();
+ values->insert_encoded_batch(encoded);
+ Block block {{std::move(values), std::make_shared<DataTypeVariantV2>(), "v"}};
+ ArrowFlightArrowBlockConvertor converter(
+ arrow::schema({arrow::field("v", native_variant(), false)}), cctz::utc_time_zone());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ auto status = converter.convert_to_arrow(block, arrow::default_memory_pool(), &batch);
+ ASSERT_TRUE(status.ok()) << status;
+ auto value = value_at(*batch->column(0), 0);
+ EXPECT_EQ(value.metadata.dict_size(), 0);
+ for (size_t i = 0; i < VARIANT_MAX_NESTING_DEPTH; ++i) {
+ value = value.array_at(0);
+ }
+ EXPECT_EQ(value.num_elements(), 0);
+ EXPECT_EQ(value.basic_type(),
+ terminal == "[]" ? VariantBasicType::ARRAY : VariantBasicType::OBJECT);
+ }
+}
+
+TEST(ArrowFlightVariantTest, SchemaRpcPreservesNestedVariantExtensions) {
+ ASSERT_TRUE(register_arrow_variant_extension().ok());
+ for (const auto& type : std::vector<std::shared_ptr<arrow::DataType>> {
+ arrow::utf8(), native_variant(), arrow::list(native_variant()),
+ arrow::map(arrow::utf8(), native_variant()),
+ arrow::struct_({arrow::field("v", native_variant())}),
+ arrow::list(arrow::struct_({arrow::field("v", native_variant())}))}) {
+ SCOPED_TRACE(type->ToString());
+ auto schema = arrow::schema({arrow::field("result", type)});
+ std::string serialized;
+ // Result schema discovery happens before batch conversion and must also support nested extensions.
+ auto status = serialize_arrow_schema(&schema, &serialized);
+ ASSERT_TRUE(status.ok()) << status;
+ auto input = arrow::io::BufferReader::FromString(serialized);
+ auto opened = arrow::ipc::RecordBatchStreamReader::Open(input.get());
+ ASSERT_TRUE(opened.ok()) << opened.status();
+ auto reader = opened.ValueOrDie();
+ EXPECT_TRUE(reader->schema()->Equals(*schema, true));
+ auto next = reader->Next();
+ ASSERT_TRUE(next.ok()) << next.status();
+ if (next.ValueOrDie() != nullptr) {
+ EXPECT_EQ(next.ValueOrDie()->num_rows(), 0);
+ }
+ }
+}
+
+TEST(ArrowFlightVariantTest, EmptyResultHasNativeSchema) {
+ auto type = std::make_shared<DataTypeVariantV2>();
+ Block block;
+ block.insert({type->create_column(), type, "v"});
+ ArrowFlightArrowBlockConvertor converter(
+ arrow::schema({arrow::field("v", native_variant(), false)}), cctz::utc_time_zone());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ auto status = converter.convert_to_arrow(block, arrow::default_memory_pool(), &batch);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_TRUE(batch->ValidateFull().ok());
+ EXPECT_EQ(batch->num_rows(), 0);
+ EXPECT_TRUE(batch->column(0)->type()->Equals(native_variant()));
+}
+
+TEST(ArrowFlightVariantTest, NestedArrayAndDefaultJsonMode) {
+ auto type = std::make_shared<DataTypeVariantV2>();
+ auto offsets = ColumnArray::ColumnOffsets::create();
+ offsets->get_data().assign({2, 4});
+ auto array_type = std::make_shared<DataTypeArray>(type);
+ Block block;
+ block.insert({ColumnArray::create(make_nullable(documents(type)), std::move(offsets)),
+ array_type, "a"});
+ auto schema = arrow::schema(
+ {arrow::field("a", arrow::list(arrow::field("item", native_variant(), true)), false)});
+ ArrowFlightArrowBlockConvertor converter(schema, cctz::utc_time_zone());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ auto status = converter.convert_to_arrow(block, arrow::default_memory_pool(), &batch);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_TRUE(batch->ValidateFull().ok());
+ const auto& values = *static_cast<const arrow::ListArray&>(*batch->column(0)).values();
+ EXPECT_EQ(value_at(values, 2).get_int(), 42);
+ EXPECT_EQ(value_at(values, 3).get_string().to_string(), "text");
+
+ DorisArrowBlockConvertor json(block, "UTC", cctz::utc_time_zone());
+ ASSERT_TRUE(json.init().ok());
+ ASSERT_TRUE(json.convert_to_arrow(block, arrow::default_memory_pool(), &batch).ok());
+ EXPECT_EQ(static_cast<const arrow::ListArray&>(*batch->column(0)).values()->type_id(),
+ arrow::Type::STRING);
+ // Native Variant bindings belong to Flight, not the ordinary Arrow export path.
+ EXPECT_FALSE(DorisArrowBlockConvertor(schema, cctz::utc_time_zone())
+ .convert_to_arrow(block, arrow::default_memory_pool(), &batch)
+ .ok());
+}
+
+} // namespace
+} // namespace doris
diff --git a/be/test/format/arrow/arrow_row_batch_test.cpp b/be/test/format/arrow/arrow_row_batch_test.cpp
new file mode 100644
index 00000000000..a80038e5eb0
--- /dev/null
+++ b/be/test/format/arrow/arrow_row_batch_test.cpp
@@ -0,0 +1,313 @@
+// 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 "format/arrow/arrow_row_batch.h"
+
+#include <arrow/api.h>
+#include <arrow/io/api.h>
+#include <arrow/ipc/api.h>
+#include <gtest/gtest.h>
+
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_factory.hpp"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_string.h"
+#include "core/data_type/data_type_struct.h"
+#include "core/data_type/data_type_variant_v2.h"
+#include "exprs/vexpr_context.h"
+#include "exprs/vslot_ref.h"
+#include "format/arrow/arrow_block_convertor.h"
+
+namespace doris {
+namespace {
+
+void expect_logical_type(const std::shared_ptr<arrow::Field>& field, const std::string& name) {
+ ASSERT_NE(nullptr, field->metadata()) << field->ToString();
+ auto value = field->metadata()->Get("doris_type");
+ ASSERT_TRUE(value.ok()) << value.status();
+ EXPECT_EQ(name, *value);
+}
+
+void expect_legacy_batch(const Block& block, const std::shared_ptr<arrow::RecordBatch>& extended) {
+ VExprContextSPtrs expressions;
+ for (size_t i = 0; i < block.columns(); ++i) {
+ const auto& column = block.get_by_position(i);
+ expressions.push_back(VExprContext::create_shared(
+ VSlotRef::create_shared(i, i, -1, column.type, column.name)));
+ }
+ std::shared_ptr<arrow::Schema> schema;
+ ASSERT_TRUE(LegacyArrowFlightSchemaConvertor("UTC")
+ .get_arrow_schema_from_expr_ctxs(expressions, &schema)
+ .ok());
+ // Synthetic slot references have no expression labels; compare batches using the same names.
+ schema = schema->WithNames(extended->schema()->field_names()).ValueOrDie();
+ std::shared_ptr<arrow::RecordBatch> batch;
+ auto status = ArrowFlightArrowBlockConvertor(schema, cctz::utc_time_zone())
+ .convert_to_arrow(block, arrow::default_memory_pool(), &batch);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_TRUE(batch->ValidateFull().ok());
+ EXPECT_TRUE(extended->Equals(*batch, false));
+ EXPECT_FALSE(extended->schema()->Equals(*schema, true));
+ std::string serialized;
+ ASSERT_TRUE(serialize_record_batch(*batch, &serialized).ok());
+ auto source = std::make_shared<arrow::io::BufferReader>(arrow::Buffer::FromString(serialized));
+ auto reader = arrow::ipc::RecordBatchStreamReader::Open(source).ValueOrDie();
+ ASSERT_TRUE(reader->ReadNext(&batch).ok());
+ EXPECT_TRUE(schema->Equals(*batch->schema(), true));
+ EXPECT_TRUE(extended->Equals(*batch, false));
+}
+
+TEST(ArrowSchemaConvertorTest, ProtocolsKeepNestedTimestampAndMetadataPoliciesSeparate) {
+ const auto datetime =
+ DataTypeFactory::instance().create_data_type(TYPE_DATETIMEV2, false, 0, 6);
+ const auto largeint = std::make_shared<DataTypeInt128>();
+ const auto nested = std::make_shared<DataTypeStruct>(DataTypes {datetime, largeint},
+ Strings {"time", "number"});
+ Block block;
+ block.insert({nested->create_column(), nested, "value"});
+ std::vector<std::unique_ptr<ArrowSchemaConvertor>> convertors;
+ convertors.push_back(std::make_unique<DorisArrowSchemaConvertor>(block, "Asia/Shanghai"));
+ convertors.push_back(std::make_unique<ArrowFlightSchemaConvertor>(block, "Asia/Shanghai"));
+ convertors.push_back(
+ std::make_unique<LegacyArrowFlightSchemaConvertor>(block, "Asia/Shanghai"));
+ for (size_t i = 0; i < convertors.size(); ++i) {
+ std::shared_ptr<arrow::Schema> schema;
+ ASSERT_TRUE(convertors[i]->get_arrow_schema(&schema).ok());
+ const auto& fields = schema->field(0)->type()->fields();
+ const auto& timestamp = static_cast<const arrow::TimestampType&>(*fields[0]->type());
+ EXPECT_EQ(i == 0 ? "Asia/Shanghai" : "", timestamp.timezone());
+ if (i == 2) {
+ EXPECT_EQ(nullptr, fields[1]->metadata());
+ } else {
+ expect_logical_type(fields[1], "LARGEINT");
+ }
+ }
+}
+
+class ArrowLogicalTypeMetadataTest
+ : public testing::TestWithParam<std::pair<PrimitiveType, const char*>> {};
+
+TEST_P(ArrowLogicalTypeMetadataTest, PreservesTopLevelAndNestedFields) {
+ const auto [primitive, name] = GetParam();
+ auto type =
+ make_nullable(primitive == TYPE_VARIANT
+ ? DataTypePtr(std::make_shared<DataTypeVariantV2>())
+ : DataTypeFactory::instance().create_data_type(primitive, false));
+ auto string_type = std::make_shared<DataTypeString>();
+ DataTypes types {type, std::make_shared<DataTypeArray>(type),
+ std::make_shared<DataTypeStruct>(DataTypes {type, string_type},
+ Strings {"typed", "text"}),
+ std::make_shared<DataTypeMap>(string_type, type)};
+ Block block;
+ for (size_t i = 0; i < types.size(); ++i) {
+ block.insert({types[i]->create_column(), types[i], std::to_string(i)});
+ }
+ std::shared_ptr<arrow::Schema> schema;
+ ASSERT_TRUE(ArrowFlightSchemaConvertor("UTC").get_arrow_schema_from_block(block, &schema).ok());
+ expect_logical_type(schema->field(0), name);
+ expect_logical_type(schema->field(1)->type()->field(0), name);
+ expect_logical_type(schema->field(2)->type()->field(0), name);
+ EXPECT_EQ(nullptr, schema->field(2)->type()->field(1)->metadata());
+ const auto& map = static_cast<const arrow::MapType&>(*schema->field(3)->type());
+ expect_logical_type(map.item_field(), name);
+ EXPECT_FALSE(map.key_field()->nullable());
+ EXPECT_TRUE(map.item_field()->nullable());
+ EXPECT_EQ(nullptr, map.key_field()->metadata());
+}
+
+TEST_P(ArrowLogicalTypeMetadataTest, OldFeReceivesLegacySchema) {
+ const auto [primitive, name] = GetParam();
+ auto type =
+ make_nullable(primitive == TYPE_VARIANT
+ ? DataTypePtr(std::make_shared<DataTypeVariantV2>())
+ : DataTypeFactory::instance().create_data_type(primitive, false));
+ auto string_type = std::make_shared<DataTypeString>();
+ DataTypes types {
+ type, std::make_shared<DataTypeArray>(type),
+ std::make_shared<DataTypeStruct>(DataTypes {type}, Strings {"typed"}),
+ std::make_shared<DataTypeMap>(primitive == TYPE_LARGEINT ? type : string_type, type)};
+ VExprContextSPtrs expressions;
+ for (size_t i = 0; i < types.size(); ++i) {
+ expressions.push_back(VExprContext::create_shared(
+ VSlotRef::create_shared(i, i, -1, types[i], std::to_string(i))));
+ }
+ std::shared_ptr<arrow::Schema> legacy;
+ std::shared_ptr<arrow::Schema> extended;
+ ASSERT_TRUE(LegacyArrowFlightSchemaConvertor("UTC")
+ .get_arrow_schema_from_expr_ctxs(expressions, &legacy)
+ .ok());
+ ASSERT_TRUE(ArrowFlightSchemaConvertor("UTC")
+ .get_arrow_schema_from_expr_ctxs(expressions, &extended)
+ .ok());
+ std::shared_ptr<arrow::DataType> storage;
+ ASSERT_TRUE(ArrowFlightSchemaConvertor("UTC").convert_to_arrow_type(type, &storage).ok());
+ auto scalar = arrow::field(legacy->field(0)->name(), storage, type->is_nullable());
+ if (primitive != TYPE_JSONB && primitive != TYPE_VARIANT) {
+ scalar = scalar->WithMetadata(arrow::key_value_metadata({"doris_type"}, {name}));
+ }
+ auto expected = arrow::schema(
+ {scalar,
+ arrow::field(legacy->field(1)->name(), arrow::list(storage), types[1]->is_nullable()),
+ arrow::field(legacy->field(2)->name(),
+ arrow::struct_({arrow::field("typed", storage, true)}),
+ types[2]->is_nullable()),
+ arrow::field(legacy->field(3)->name(), arrow::map(arrow::utf8(), storage),
+ types[3]->is_nullable())});
+ EXPECT_TRUE(expected->Equals(*legacy, true));
+ EXPECT_TRUE(legacy->Equals(*extended, false));
+ EXPECT_FALSE(legacy->Equals(*extended, true));
+ expect_logical_type(extended->field(0), name);
+ expect_logical_type(extended->field(1)->type()->field(0), name);
+ std::string serialized;
+ ASSERT_TRUE(serialize_arrow_schema(&legacy, &serialized).ok());
+ auto source = std::make_shared<arrow::io::BufferReader>(arrow::Buffer::FromString(serialized));
+ auto reader = arrow::ipc::RecordBatchStreamReader::Open(source).ValueOrDie();
+ EXPECT_TRUE(expected->Equals(*reader->schema(), true));
+}
+
+INSTANTIATE_TEST_SUITE_P(LogicalTypes, ArrowLogicalTypeMetadataTest,
+ testing::Values(std::make_pair(TYPE_LARGEINT, "LARGEINT"),
+ std::make_pair(TYPE_IPV4, "IPV4"),
+ std::make_pair(TYPE_IPV6, "IPV6"),
+ std::make_pair(TYPE_JSONB, "JSON"),
+ std::make_pair(TYPE_UUID, "UUID"),
+ std::make_pair(TYPE_VARIANT, "VARIANT")));
+
+TEST(ArrowRowBatchMetadataTest, PreservesMapKeysAndDeepNestingThroughIpc) {
+ auto integer = std::make_shared<DataTypeInt128>();
+ auto structure = std::make_shared<DataTypeStruct>(DataTypes {integer}, Strings {"number"});
+ auto array = std::make_shared<DataTypeArray>(structure);
+ auto type = std::make_shared<DataTypeMap>(integer, array);
+ Block block;
+ block.insert({type->create_column(), type, "m"});
+ std::shared_ptr<arrow::Schema> schema;
+ ASSERT_TRUE(DorisArrowSchemaConvertor("UTC").get_arrow_schema_from_block(block, &schema).ok());
+ std::string serialized;
+ ASSERT_TRUE(serialize_arrow_schema(&schema, &serialized).ok());
+ auto source = std::make_shared<arrow::io::BufferReader>(arrow::Buffer::FromString(serialized));
+ auto reader = arrow::ipc::RecordBatchStreamReader::Open(source).ValueOrDie();
+ EXPECT_TRUE(schema->Equals(*reader->schema(), true));
+ const auto& map = static_cast<const arrow::MapType&>(*reader->schema()->field(0)->type());
+ expect_logical_type(map.key_field(), "LARGEINT");
+ EXPECT_EQ("key", map.key_field()->name());
+ EXPECT_FALSE(map.key_field()->nullable());
+ EXPECT_EQ("value", map.item_field()->name());
+ auto item = map.item_type()->field(0);
+ EXPECT_EQ("item", item->name());
+ EXPECT_TRUE(item->nullable());
+ expect_logical_type(item->type()->field(0), "LARGEINT");
+ EXPECT_FALSE(item->type()->field(0)->nullable());
+ auto physical =
+ arrow::map(arrow::utf8(),
+ arrow::list(arrow::struct_({arrow::field("number", arrow::utf8(), false)})));
+ EXPECT_TRUE(map.Equals(physical));
+ EXPECT_FALSE(map.Equals(physical, true));
+}
+
+TEST(ArrowRowBatchMetadataTest, KeepsNativeTypesUnannotated) {
+ auto type = std::make_shared<DataTypeStruct>(
+ DataTypes {std::make_shared<DataTypeInt32>(), std::make_shared<DataTypeString>()},
+ Strings {"number", "text"});
+ std::shared_ptr<arrow::DataType> arrow_type;
+ ASSERT_TRUE(DorisArrowSchemaConvertor("UTC").convert_to_arrow_type(type, &arrow_type).ok());
+ for (const auto& field : arrow_type->fields()) {
+ EXPECT_EQ(nullptr, field->metadata());
+ EXPECT_FALSE(field->nullable());
+ }
+}
+
+TEST(ArrowRowBatchMetadataTest, PreservesLargeintExtremesAndNullsInRecordBatches) {
+ auto integer = make_nullable(std::make_shared<DataTypeInt128>());
+ auto array_type = std::make_shared<DataTypeArray>(integer);
+ auto column = array_type->create_column();
+ column->insert(Field::create_field<TYPE_ARRAY>(
+ Array {Field::create_field<TYPE_LARGEINT>(MAX_INT128),
+ Field::create_field<TYPE_LARGEINT>(MIN_INT128), Field()}));
+ column->insert(Field::create_field<TYPE_ARRAY>(Array {}));
+ Block block;
+ block.insert({std::move(column), array_type, "numbers"});
+ std::shared_ptr<arrow::Schema> schema;
+ ASSERT_TRUE(ArrowFlightSchemaConvertor("UTC").get_arrow_schema_from_block(block, &schema).ok());
+ ArrowFlightArrowBlockConvertor converter(schema, cctz::utc_time_zone());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ auto status = converter.convert_to_arrow(block, arrow::default_memory_pool(), &batch);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_TRUE(batch->ValidateFull().ok());
+ std::string serialized;
+ ASSERT_TRUE(serialize_record_batch(*batch, &serialized).ok());
+ auto source = std::make_shared<arrow::io::BufferReader>(arrow::Buffer::FromString(serialized));
+ auto reader = arrow::ipc::RecordBatchStreamReader::Open(source).ValueOrDie();
+ ASSERT_TRUE(reader->ReadNext(&batch).ok());
+ EXPECT_TRUE(batch->schema()->Equals(*schema, true));
+ expect_logical_type(batch->schema()->field(0)->type()->field(0), "LARGEINT");
+ const auto& lists = static_cast<const arrow::ListArray&>(*batch->column(0));
+ EXPECT_EQ(0, lists.value_length(1));
+ const auto& values = static_cast<const arrow::StringArray&>(*lists.values());
+ EXPECT_EQ("170141183460469231731687303715884105727", values.GetString(0));
+ EXPECT_EQ("-170141183460469231731687303715884105728", values.GetString(1));
+ EXPECT_TRUE(values.IsNull(2));
+ expect_legacy_batch(block, batch);
+}
+
+TEST(ArrowRowBatchMetadataTest, PreservesNestedMapMetadataAndValuesInRecordBatches) {
+ auto integer = make_nullable(std::make_shared<DataTypeInt128>());
+ auto map_type = make_nullable(std::make_shared<DataTypeMap>(integer, integer));
+ auto type = make_nullable(std::make_shared<DataTypeStruct>(
+ DataTypes {map_type, std::make_shared<DataTypeString>()}, Strings {"mapping", "text"}));
+ Map map;
+ map.push_back(Field::create_field<TYPE_ARRAY>(
+ Array {Field::create_field<TYPE_LARGEINT>(MAX_INT128)}));
+ map.push_back(Field::create_field<TYPE_ARRAY>(
+ Array {Field::create_field<TYPE_LARGEINT>(MIN_INT128)}));
+ auto column = type->create_column();
+ column->insert(Field::create_field<TYPE_STRUCT>(
+ Struct {Field::create_field<TYPE_MAP>(map), Field::create_field<TYPE_STRING>("17")}));
+ column->insert_default();
+ Block block;
+ block.insert({std::move(column), type, "nested"});
+ std::shared_ptr<arrow::Schema> schema;
+ ASSERT_TRUE(ArrowFlightSchemaConvertor("UTC").get_arrow_schema_from_block(block, &schema).ok());
+ ArrowFlightArrowBlockConvertor converter(schema, cctz::utc_time_zone());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ auto status = converter.convert_to_arrow(block, arrow::default_memory_pool(), &batch);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_TRUE(batch->ValidateFull().ok());
+ std::string serialized;
+ ASSERT_TRUE(serialize_record_batch(*batch, &serialized).ok());
+ auto source = std::make_shared<arrow::io::BufferReader>(arrow::Buffer::FromString(serialized));
+ auto reader = arrow::ipc::RecordBatchStreamReader::Open(source).ValueOrDie();
+ ASSERT_TRUE(reader->ReadNext(&batch).ok());
+ const auto& structure = static_cast<const arrow::StructArray&>(*batch->column(0));
+ EXPECT_TRUE(structure.IsNull(1));
+ const auto& maps = static_cast<const arrow::MapArray&>(*structure.field(0));
+ const auto& arrow_map_type = static_cast<const arrow::MapType&>(*maps.type());
+ expect_logical_type(arrow_map_type.key_field(), "LARGEINT");
+ expect_logical_type(arrow_map_type.item_field(), "LARGEINT");
+ EXPECT_FALSE(arrow_map_type.key_field()->nullable());
+ EXPECT_EQ("170141183460469231731687303715884105727",
+ static_cast<const arrow::StringArray&>(*maps.keys()).GetString(0));
+ EXPECT_EQ("-170141183460469231731687303715884105728",
+ static_cast<const arrow::StringArray&>(*maps.items()).GetString(0));
+ EXPECT_EQ("17", static_cast<const arrow::StringArray&>(*structure.field(1)).GetString(0));
+ EXPECT_EQ(nullptr, structure.type()->field(1)->metadata());
+ expect_legacy_batch(block, batch);
+}
+
+} // namespace
+} // namespace doris
diff --git a/be/test/format/table/iceberg/arrow_schema_util_test.cpp b/be/test/format/table/iceberg/arrow_schema_util_test.cpp
index b6cdf6baed0..ce1d3e84123 100644
--- a/be/test/format/table/iceberg/arrow_schema_util_test.cpp
+++ b/be/test/format/table/iceberg/arrow_schema_util_test.cpp
@@ -135,7 +135,7 @@ TEST(ArrowSchemaUtilTest, test_simple_field) {
std::vector<std::shared_ptr<arrow::Field>> fields;
Status st;
- st = ArrowSchemaUtil::convert(&schema, "utc", fields);
+ st = IcebergArrowSchemaConvertor(schema, "utc").convert_fields(fields);
EXPECT_TRUE(st.ok());
EXPECT_EQ(2, fields.size());
EXPECT_EQ("field1", fields[0]->name());
@@ -193,7 +193,7 @@ TEST(ArrowSchemaUtilTest, test_stuct_field) {
std::vector<std::shared_ptr<arrow::Field>> fields;
Status st;
- st = ArrowSchemaUtil::convert(schema.get(), "utc", fields);
+ st = IcebergArrowSchemaConvertor(*schema, "utc").convert_fields(fields);
EXPECT_TRUE(st.ok());
EXPECT_EQ(1, fields.size());
EXPECT_EQ("st_col", fields[0]->name());
@@ -245,7 +245,7 @@ TEST(ArrowSchemaUtilTest, test_map_field) {
std::vector<std::shared_ptr<arrow::Field>> fields;
Status st;
- st = ArrowSchemaUtil::convert(schema.get(), "utc", fields);
+ st = IcebergArrowSchemaConvertor(*schema, "utc").convert_fields(fields);
EXPECT_TRUE(st.ok());
EXPECT_EQ(1, fields.size());
EXPECT_EQ("map_col", fields[0]->name());
@@ -292,7 +292,7 @@ TEST(ArrowSchemaUtilTest, test_list_field) {
std::vector<std::shared_ptr<arrow::Field>> fields;
Status st;
- st = ArrowSchemaUtil::convert(schema.get(), "utc", fields);
+ st = IcebergArrowSchemaConvertor(*schema, "utc").convert_fields(fields);
EXPECT_TRUE(st.ok());
EXPECT_EQ(1, fields.size());
EXPECT_EQ("list_col", fields[0]->name());
@@ -316,7 +316,7 @@ TEST(ArrowSchemaUtilTest, test_binary_field_types) {
Schema schema(1, std::move(nested_fields));
std::vector<std::shared_ptr<arrow::Field>> fields;
- Status st = ArrowSchemaUtil::convert(&schema, "utc", fields);
+ Status st = IcebergArrowSchemaConvertor(schema, "utc").convert_fields(fields);
ASSERT_TRUE(st.ok()) << st;
ASSERT_EQ(4, fields.size());
@@ -371,7 +371,7 @@ TEST(ArrowSchemaUtilTest, test_parquet_filed_id) {
Schema schema(1, std::move(nested_fields));
std::vector<std::shared_ptr<arrow::Field>> fields;
- st = ArrowSchemaUtil::convert(&schema, "utc", fields);
+ st = IcebergArrowSchemaConvertor(schema, "utc").convert_fields(fields);
auto arrow_schema = arrow::schema(fields);
// create arrow table
diff --git a/be/test/format_v2/table/adbc_reader_test.cpp b/be/test/format_v2/table/adbc_reader_test.cpp
index 00e7f68554b..f04460ecab9 100644
--- a/be/test/format_v2/table/adbc_reader_test.cpp
+++ b/be/test/format_v2/table/adbc_reader_test.cpp
@@ -40,11 +40,15 @@
#include "core/column/column_string.h"
#include "core/column/column_struct.h"
#include "core/column/column_vector.h"
+#include "core/column/variant_v2/column_variant_v2.h"
#include "core/data_type/data_type_array.h"
#include "core/data_type/data_type_nullable.h"
#include "core/data_type/data_type_number.h"
#include "core/data_type/data_type_string.h"
#include "core/data_type/data_type_struct.h"
+#include "core/data_type/data_type_variant_v2.h"
+#include "core/value/variant/variant_batch_builder.h"
+#include "format/arrow/arrow_block_convertor.h"
#include "format_v2/column_mapper.h"
#include "format_v2/file_reader.h"
#include "gen_cpp/PlanNodes_types.h"
@@ -623,6 +627,61 @@ TEST(AdbcReaderTest, CreatesMaterializedColumnMapper) {
ASSERT_NE(dynamic_cast<MaterializedColumnMapper*>(mapper.get()), nullptr);
}
+TEST(AdbcReaderTest, MaterializesNativeVariantAndSqlNull) {
+ auto variant = std::make_shared<DataTypeVariantV2>();
+ VariantBatchBuilder encoded;
+ auto row = encoded.begin_row();
+ row.add_int(42);
+ row.finish();
+ auto data = ColumnVariantV2::create();
+ data->insert_encoded_batch(encoded.finish_batch());
+ auto nulls = ColumnUInt8::create();
+ nulls->get_data().push_back(0);
+ auto column = ColumnNullable::create(std::move(data), std::move(nulls));
+ column->insert_default();
+ auto type = make_nullable(variant);
+ Block source {{std::move(column), type, "v"}};
+ ArrowFlightArrowBlockConvertor converter(source, "UTC", cctz::utc_time_zone());
+ ASSERT_TRUE(converter.init().ok());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ ASSERT_TRUE(converter.convert_to_arrow(source, arrow::default_memory_pool(), &batch).ok());
+
+ ObjectPool pool;
+ DescriptorTblBuilder descriptors(&pool);
+ descriptors.declare_tuple() << std::make_tuple(type, std::string("v"));
+ auto* descriptor = descriptors.build();
+ RuntimeState state;
+ RuntimeProfile profile("native_variant_reader");
+ auto close_count = std::make_shared<int>(0);
+ auto reader = create_reader(
+ &profile, fake_adbc_range(), descriptor->get_tuple_descriptor(0)->slots(),
+ [batch, close_count](const TFileRangeDesc&, std::unique_ptr<AdbcStream>* out) {
+ *out = std::make_unique<BatchAdbcStream>(
+ std::vector<std::shared_ptr<arrow::RecordBatch>> {batch}, close_count);
+ return Status::OK();
+ });
+ ASSERT_TRUE(reader->init(&state).ok());
+ std::vector<ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<FileScanRequest>();
+ FileScanRequestBuilder request_builder(request.get());
+ ASSERT_TRUE(request_builder.add_non_predicate_column(LocalColumnId(0)).ok());
+ ASSERT_TRUE(reader->open(request).ok());
+ auto block = make_request_block(schema, {0});
+ size_t rows = 0;
+ bool eof = false;
+ const auto status = reader->get_block(&block, &rows, &eof);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(rows, 2);
+ const auto& result = assert_cast<const ColumnNullable&>(*block.get_by_position(0).column);
+ EXPECT_EQ(assert_cast<const ColumnVariantV2&>(result.get_nested_column())
+ .get_value_ref(0)
+ .get_int(),
+ 42);
+ EXPECT_TRUE(result.is_null_at(1));
+ ASSERT_TRUE(reader->close().ok());
+}
+
TEST(AdbcReaderTest, RejectsReorderedSameTypedStructFields) {
ObjectPool pool;
DescriptorTbl* desc_tbl = nullptr;
diff --git a/fe/fe-connector/fe-connector-adbc/src/main/java/org/apache/doris/connector/adbc/AdbcTypeMapper.java b/fe/fe-connector/fe-connector-adbc/src/main/java/org/apache/doris/connector/adbc/AdbcTypeMapper.java
index c1b5b2d24cb..b4b581bbf2a 100644
--- a/fe/fe-connector/fe-connector-adbc/src/main/java/org/apache/doris/connector/adbc/AdbcTypeMapper.java
+++ b/fe/fe-connector/fe-connector-adbc/src/main/java/org/apache/doris/connector/adbc/AdbcTypeMapper.java
@@ -64,6 +64,20 @@ public final class AdbcTypeMapper {
}
private static ConnectorType toDorisType(String columnName, Field field, ArrowType arrowType) {
+ if (field.getMetadata() != null
+ && "arrow.parquet.variant".equals(field.getMetadata().get("ARROW:extension:name"))) {
+ // Flight carries Variant as binary storage; treating it as an ordinary struct loses
+ // its logical type and exposes encoded metadata/value bytes as strings.
+ List<Field> children = field.getChildren();
+ if (arrowType.getTypeID() != ArrowType.ArrowTypeID.Struct || children.size() != 2
+ || !"metadata".equals(children.get(0).getName())
+ || !"value".equals(children.get(1).getName())
+ || !ArrowType.Binary.INSTANCE.equals(children.get(0).getType())
+ || !ArrowType.Binary.INSTANCE.equals(children.get(1).getType())) {
+ throw unsupported(columnName, arrowType);
+ }
+ return ConnectorType.of("VARIANT_COMPUTE_V2");
+ }
switch (arrowType.getTypeID()) {
case Bool:
return ConnectorType.of("BOOLEAN");
diff --git a/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcTypeMapperTest.java b/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcTypeMapperTest.java
index b49efdda681..6951eb228af 100644
--- a/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcTypeMapperTest.java
+++ b/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcTypeMapperTest.java
@@ -31,6 +31,7 @@ import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.List;
+import java.util.Map;
class AdbcTypeMapperTest {
@@ -52,6 +53,21 @@ class AdbcTypeMapperTest {
return e.getMessage();
}
+ @Test
+ void nativeVariantKeepsItsLogicalTypeInsideContainers() {
+ Field variant = new Field("v", new FieldType(true, ArrowType.Struct.INSTANCE, null,
+ Map.of("ARROW:extension:name", "arrow.parquet.variant")),
+ List.of(Field.notNullable("metadata", ArrowType.Binary.INSTANCE),
+ Field.notNullable("value", ArrowType.Binary.INSTANCE)));
+ Assertions.assertEquals(ConnectorType.of("VARIANT_COMPUTE_V2"), AdbcTypeMapper.toDorisType("v", variant));
+ Assertions.assertEquals(ConnectorType.arrayOf(ConnectorType.of("VARIANT_COMPUTE_V2")),
+ AdbcTypeMapper.toDorisType("a", nested("a", ArrowType.List.INSTANCE, List.of(variant))));
+ Field plain = nested("v", ArrowType.Struct.INSTANCE, variant.getChildren());
+ Assertions.assertEquals(ConnectorType.structOf(Arrays.asList("metadata", "value"),
+ List.of(ConnectorType.of("STRING"), ConnectorType.of("STRING"))),
+ AdbcTypeMapper.toDorisType("v", plain));
+ }
+
@Test
void booleanMapsToBoolean() {
Assertions.assertEquals(ConnectorType.of("BOOLEAN"), map(ArrowType.Bool.INSTANCE));
diff --git a/fe/fe-core/src/main/java/org/apache/doris/arrowflight/FlightSqlConnectProcessor.java b/fe/fe-core/src/main/java/org/apache/doris/arrowflight/FlightSqlConnectProcessor.java
index 0203e29a802..4eddd5c7516 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/arrowflight/FlightSqlConnectProcessor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/arrowflight/FlightSqlConnectProcessor.java
@@ -40,6 +40,7 @@ import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.vector.FieldVector;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.ipc.ArrowStreamReader;
+import org.apache.arrow.vector.types.pojo.Field;
import org.apache.arrow.vector.types.pojo.Schema;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -156,7 +157,13 @@ public class FlightSqlConnectProcessor extends ConnectProcessor implements AutoC
String.format("Schema size %s' is not equal to arrow field size %s, queryId: %s.",
fieldVectors.size(), resultOutputExprs.size(), DebugUtil.printId(tid)));
}
- schema = root.getSchema();
+ List<Field> fields = new ArrayList<>();
+ for (int i = 0; i < resultOutputExprs.size(); i++) {
+ fields.add(FlightSqlSchemaHelper.withDorisTypeMetadata(
+ root.getSchema().getFields().get(i), resultOutputExprs.get(i).getType()));
+ }
+ // Compare and advertise the same complete metadata regardless of BE upgrade order.
+ schema = new Schema(fields, root.getSchema().getCustomMetadata());
if (arrowSchema == null) {
arrowSchema = schema;
} else if (!arrowSchema.equals(schema)) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/arrowflight/FlightSqlSchemaHelper.java b/fe/fe-core/src/main/java/org/apache/doris/arrowflight/FlightSqlSchemaHelper.java
index e7a98cf4835..336c3dc7a36 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/arrowflight/FlightSqlSchemaHelper.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/arrowflight/FlightSqlSchemaHelper.java
@@ -18,20 +18,27 @@
package org.apache.doris.arrowflight;
import org.apache.doris.arrow.DorisArrowTypeMapping;
+import org.apache.doris.catalog.ArrayType;
import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MapType;
+import org.apache.doris.catalog.StructType;
+import org.apache.doris.catalog.Type;
import org.apache.doris.datasource.CatalogIf;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.service.ExecuteEnv;
import org.apache.doris.service.FrontendServiceImpl;
import org.apache.doris.thrift.TColumnDef;
+import org.apache.doris.thrift.TColumnDesc;
import org.apache.doris.thrift.TDescribeTablesParams;
import org.apache.doris.thrift.TDescribeTablesResult;
import org.apache.doris.thrift.TGetDbsParams;
import org.apache.doris.thrift.TGetDbsResult;
import org.apache.doris.thrift.TGetTablesParams;
import org.apache.doris.thrift.TListTableStatusResult;
+import org.apache.doris.thrift.TPrimitiveType;
import org.apache.doris.thrift.TTableStatus;
+import org.apache.arrow.flight.CallStatus;
import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetDbSchemas;
import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTables;
import org.apache.arrow.vector.VarBinaryVector;
@@ -39,7 +46,9 @@ import org.apache.arrow.vector.VarCharVector;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.ipc.WriteChannel;
import org.apache.arrow.vector.ipc.message.MessageSerializer;
+import org.apache.arrow.vector.types.pojo.ArrowType;
import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.FieldType;
import org.apache.arrow.vector.types.pojo.Schema;
import org.apache.arrow.vector.util.Text;
import org.apache.logging.log4j.LogManager;
@@ -75,6 +84,56 @@ public class FlightSqlSchemaHelper {
private static final byte[] EMPTY_SERIALIZED_SCHEMA = getSerializedSchema(Collections.emptyList());
+ static Field withDorisTypeMetadata(Field field, Type type) {
+ if (type.isVariantType()) {
+ if (field.getMetadata() == null
+ || !"arrow.parquet.variant".equals(field.getMetadata().get("ARROW:extension:name"))) {
+ throw CallStatus.UNIMPLEMENTED.withDescription(
+ "Backend returned a non-native Variant schema; use Variant V2 "
+ + "or cast the result to STRING").toRuntimeException();
+ }
+ }
+ List<Field> children = new ArrayList<>(field.getChildren());
+ if (type.isArrayType()) {
+ children.set(0, withDorisTypeMetadata(children.get(0), ((ArrayType) type).getItemType()));
+ } else if (type.isMapType()) {
+ Field entries = children.get(0);
+ List<Field> pair = new ArrayList<>(entries.getChildren());
+ pair.set(0, withDorisTypeMetadata(pair.get(0), ((MapType) type).getKeyType()));
+ pair.set(1, withDorisTypeMetadata(pair.get(1), ((MapType) type).getValueType()));
+ children.set(0, new Field(entries.getName(), entries.getFieldType(), pair));
+ } else if (type.isStructType()) {
+ StructType struct = (StructType) type;
+ for (int i = 0; i < children.size(); i++) {
+ children.set(i, withDorisTypeMetadata(children.get(i), struct.getFields().get(i).getType()));
+ }
+ }
+ String marker = null;
+ switch (type.getPrimitiveType()) {
+ case LARGEINT:
+ case IPV4:
+ case IPV6:
+ case UUID:
+ case VARIANT:
+ marker = type.getPrimitiveType().name();
+ break;
+ case JSONB:
+ marker = "JSON";
+ break;
+ default:
+ break;
+ }
+ FieldType fieldType = field.getFieldType();
+ if (marker != null && !field.getMetadata().containsKey("doris_type")) {
+ Map<String, String> metadata = new HashMap<>(field.getMetadata());
+ metadata.put("doris_type", marker);
+ fieldType = new FieldType(field.isNullable(), field.getType(), field.getDictionary(), metadata);
+ }
+ // Old BEs omit these markers. Fill only missing ones from the planned Doris type;
+ // preserve conflicting markers and all physical properties for the strict schema comparison.
+ return new Field(field.getName(), fieldType, children);
+ }
+
protected static byte[] getSerializedSchema(List<Field> fields) {
if (EMPTY_SERIALIZED_SCHEMA == null && fields == null) {
fields = Collections.emptyList();
@@ -175,13 +234,53 @@ public class FlightSqlSchemaHelper {
Integer tableOffset = describeTablesResult.getTablesOffset().get(tableIndex);
for (; columnIndex < tableOffset; columnIndex++) {
TColumnDef columnDef = describeTablesResult.getColumns().get(columnIndex);
- fields.add(DorisArrowTypeMapping.toField(dbName, tableName, columnDef.getColumnDesc()));
+ fields.add(buildField(dbName, tableName, columnDef.getColumnDesc()));
}
tableToFields.put(tableName, fields);
}
return tableToFields;
}
+ private static Field buildField(String dbName, String tableName, TColumnDesc desc) {
+ return withNativeVariant(DorisArrowTypeMapping.toField(dbName, tableName, desc), desc);
+ }
+
+ private static Field withNativeVariant(Field field, TColumnDesc desc) {
+ if (desc.getColumnType() == TPrimitiveType.VARIANT) {
+ return nativeVariantField(field.getName(), field.isNullable(), field.getMetadata());
+ }
+ if (!desc.isSetChildren() || desc.getChildren().isEmpty()) {
+ return field;
+ }
+ // Preserve the shared mapper's physical types, nullability and column metadata.
+ // Only Variant leaves need the Flight-specific native storage representation.
+ List<Field> children = new ArrayList<>(field.getChildren());
+ if (desc.getColumnType() == TPrimitiveType.MAP && desc.getChildrenSize() == 2) {
+ Field entries = children.get(0);
+ List<Field> pair = new ArrayList<>(entries.getChildren());
+ for (int i = 0; i < pair.size(); i++) {
+ pair.set(i, withNativeVariant(pair.get(i), desc.getChildren().get(i)));
+ }
+ children.set(0, new Field(entries.getName(), entries.getFieldType(), pair));
+ } else if (desc.getColumnType() == TPrimitiveType.STRUCT
+ || (desc.getColumnType() == TPrimitiveType.ARRAY && desc.getChildrenSize() == 1)) {
+ for (int i = 0; i < children.size(); i++) {
+ children.set(i, withNativeVariant(children.get(i), desc.getChildren().get(i)));
+ }
+ }
+ return new Field(field.getName(), field.getFieldType(), children);
+ }
+
+ static Field nativeVariantField(String name, boolean nullable, Map<String, String> columnMetadata) {
+ Map<String, String> metadata = new HashMap<>(columnMetadata);
+ // Discovery and execution must share the extension metadata as well as its storage type.
+ metadata.put("ARROW:extension:name", "arrow.parquet.variant");
+ metadata.put("ARROW:extension:metadata", "");
+ return new Field(name, new FieldType(nullable, new ArrowType.Struct(), null, metadata),
+ Arrays.asList(Field.notNullable("metadata", new ArrowType.Binary()),
+ Field.notNullable("value", new ArrowType.Binary())));
+ }
+
/**
* for FlightSqlProducer Schemas.GET_CATALOGS_SCHEMA
*/
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/ResultSink.java b/fe/fe-core/src/main/java/org/apache/doris/planner/ResultSink.java
index 2bc65ed3545..1d0ed9f5f9a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/ResultSink.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/ResultSink.java
@@ -33,7 +33,7 @@ public class ResultSink extends DataSink {
private TResultSinkType resultSinkType = TResultSinkType.MYSQL_PROTOCOL;
public ResultSink(PlanNodeId exchNodeId) {
- this.exchNodeId = exchNodeId;
+ this(exchNodeId, TResultSinkType.MYSQL_PROTOCOL);
}
public ResultSink(PlanNodeId exchNodeId, TResultSinkType resultSinkType) {
@@ -56,6 +56,9 @@ public class ResultSink extends DataSink {
TDataSink result = new TDataSink(TDataSinkType.RESULT_SINK);
TResultSink tResultSink = new TResultSink();
tResultSink.setType(resultSinkType);
+ if (resultSinkType == TResultSinkType.ARROW_FLIGHT_PROTOCOL) {
+ tResultSink.setEnableArrowTypeMetadata(true);
+ }
result.setResultSink(tResultSink);
return result;
}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/arrowflight/FlightSqlConnectProcessorSchemaTest.java b/fe/fe-core/src/test/java/org/apache/doris/arrowflight/FlightSqlConnectProcessorSchemaTest.java
new file mode 100644
index 00000000000..ee3a2334eab
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/arrowflight/FlightSqlConnectProcessorSchemaTest.java
@@ -0,0 +1,167 @@
+// 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.
+
+package org.apache.doris.arrowflight;
+
+import org.apache.doris.analysis.Expr;
+import org.apache.doris.arrowflight.results.FlightSqlEndpointsLocation;
+import org.apache.doris.catalog.ArrayType;
+import org.apache.doris.catalog.MapType;
+import org.apache.doris.catalog.StructField;
+import org.apache.doris.catalog.StructType;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.proto.InternalService.PFetchArrowFlightSchemaResult;
+import org.apache.doris.proto.Types.PStatus;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.rpc.BackendServiceProxy;
+import org.apache.doris.thrift.TNetworkAddress;
+import org.apache.doris.thrift.TUniqueId;
+
+import com.google.protobuf.ByteString;
+import org.apache.arrow.vector.ipc.WriteChannel;
+import org.apache.arrow.vector.ipc.message.MessageSerializer;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.FieldType;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.io.ByteArrayOutputStream;
+import java.nio.channels.Channels;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
+
+class FlightSqlConnectProcessorSchemaTest {
+ private static Field field(String name, ArrowType type, boolean nullable, String marker, Field... children) {
+ Map<String, String> metadata = marker == null ? Collections.emptyMap()
+ : Collections.singletonMap("doris_type", marker);
+ return new Field(name, new FieldType(nullable, type, null, metadata), Arrays.asList(children));
+ }
+
+ private static Schema schema(Field... fields) {
+ return new Schema(Arrays.asList(fields), Collections.singletonMap("schema_key", "schema_value"));
+ }
+
+ private static Schema nestedSchema(boolean annotated) {
+ return schema(
+ field("a", new ArrowType.List(), true, null,
+ field("item", new ArrowType.Utf8(), true, annotated ? "LARGEINT" : null)),
+ field("m", new ArrowType.Map(false), true, null,
+ field("entries", new ArrowType.Struct(), false, null,
+ field("key", new ArrowType.Utf8(), false, annotated ? "LARGEINT" : null),
+ field("value", new ArrowType.Struct(), true, null,
+ field("ip4", new ArrowType.Int(32, true), true, annotated ? "IPV4" : null),
+ field("ip6", new ArrowType.Utf8(), true, annotated ? "IPV6" : null),
+ field("json", new ArrowType.Utf8(), true, annotated ? "JSON" : null),
+ field("uuid", org.apache.arrow.vector.extension.UuidType.INSTANCE,
+ true, annotated ? "UUID" : null),
+ field("text", new ArrowType.Utf8(), true, null)))),
+ field("json", new ArrowType.Utf8(), true, annotated ? "JSON" : null));
+ }
+
+ private static List<Type> resultTypes() {
+ return Arrays.asList(new ArrayType(Type.LARGEINT),
+ new MapType(Type.LARGEINT, new StructType(
+ new StructField("ip4", Type.IPV4), new StructField("ip6", Type.IPV6),
+ new StructField("json", Type.JSONB), new StructField("uuid", Type.UUID),
+ new StructField("text", Type.STRING))), Type.JSONB);
+ }
+
+ private static Schema fetch(List<Type> types, Schema... schemas) throws Exception {
+ List<FlightSqlEndpointsLocation> endpoints = new ArrayList<>();
+ List<CompletableFuture<PFetchArrowFlightSchemaResult>> responses = new ArrayList<>();
+ for (int i = 0; i < schemas.length; i++) {
+ ArrayList<Expr> exprs = new ArrayList<>();
+ for (Type type : types) {
+ Expr expr = Mockito.mock(Expr.class);
+ Mockito.when(expr.getType()).thenReturn(type);
+ exprs.add(expr);
+ }
+ endpoints.add(new FlightSqlEndpointsLocation(new TUniqueId(1, i),
+ new TNetworkAddress("localhost", 10000 + i),
+ new TNetworkAddress("localhost", 11000 + i), exprs));
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ MessageSerializer.serialize(new WriteChannel(Channels.newChannel(bytes)), schemas[i]);
+ responses.add(CompletableFuture.completedFuture(PFetchArrowFlightSchemaResult.newBuilder()
+ .setStatus(PStatus.newBuilder().setStatusCode(0))
+ .setSchema(ByteString.copyFrom(bytes.toByteArray())).build()));
+ }
+ ConnectContext context = Mockito.mock(ConnectContext.class);
+ Mockito.when(context.getFlightSqlEndpointsLocations()).thenReturn(endpoints);
+ BackendServiceProxy proxy = Mockito.mock(BackendServiceProxy.class);
+ AtomicInteger next = new AtomicInteger();
+ Mockito.when(proxy.fetchArrowFlightSchema(Mockito.any(), Mockito.any()))
+ .thenAnswer(invocation -> responses.get(next.getAndIncrement()));
+ try (MockedStatic<BackendServiceProxy> singleton = Mockito.mockStatic(BackendServiceProxy.class);
+ FlightSqlConnectProcessor processor = new FlightSqlConnectProcessor(context)) {
+ singleton.when(BackendServiceProxy::getInstance).thenReturn(proxy);
+ processor.fetchArrowFlightSchema(1000);
+ return processor.getArrowSchema();
+ }
+ }
+
+ @Test
+ void mixedVersionsAdvertiseCompleteMetadataInEitherOrder() throws Exception {
+ Schema oldSchema = nestedSchema(false);
+ Schema newSchema = nestedSchema(true);
+ Assertions.assertEquals(newSchema, fetch(resultTypes(), oldSchema, newSchema));
+ Assertions.assertEquals(newSchema, fetch(resultTypes(), newSchema, oldSchema));
+ }
+
+ @Test
+ void oldOnlyAndNewOnlyAdvertiseTheSameSchema() throws Exception {
+ Assertions.assertEquals(nestedSchema(true), fetch(resultTypes(), nestedSchema(false)));
+ Assertions.assertEquals(nestedSchema(true), fetch(resultTypes(), nestedSchema(true), nestedSchema(true)));
+ }
+
+ private static void assertSchemaMismatch(Schema first, Schema second) {
+ RuntimeException failure = Assertions.assertThrows(RuntimeException.class,
+ () -> fetch(Collections.singletonList(Type.JSONB), first, second));
+ Assertions.assertTrue(failure.getCause().getMessage()
+ .startsWith("The schema returned by results BE is different"));
+ }
+
+ @Test
+ void stillRejectsPhysicalDifferences() {
+ Schema expected = schema(field("value", new ArrowType.Utf8(), true, "JSON"));
+ for (Field incompatible : Arrays.asList(
+ field("value", new ArrowType.Int(32, true), true, "JSON"),
+ field("renamed", new ArrowType.Utf8(), true, "JSON"),
+ field("value", new ArrowType.Utf8(), false, "JSON"))) {
+ assertSchemaMismatch(expected, schema(incompatible));
+ }
+ }
+
+ @Test
+ void stillRejectsConflictingOrUnrelatedMetadata() {
+ Schema expected = schema(field("value", new ArrowType.Utf8(), true, "JSON"));
+ Schema wrongMarker = schema(field("value", new ArrowType.Utf8(), true, "VARIANT"));
+ Field extraMetadata = new Field("value", new FieldType(true, new ArrowType.Utf8(), null,
+ Collections.singletonMap("other_key", "other_value")), Collections.emptyList());
+ assertSchemaMismatch(expected, wrongMarker);
+ assertSchemaMismatch(expected, schema(extraMetadata));
+ assertSchemaMismatch(expected, new Schema(expected.getFields()));
+ }
+}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/arrowflight/FlightSqlNativeVariantTest.java b/fe/fe-core/src/test/java/org/apache/doris/arrowflight/FlightSqlNativeVariantTest.java
new file mode 100644
index 00000000000..365039010ab
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/arrowflight/FlightSqlNativeVariantTest.java
@@ -0,0 +1,95 @@
+// 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.
+
+package org.apache.doris.arrowflight;
+
+import org.apache.doris.catalog.Type;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.thrift.TColumnDesc;
+import org.apache.doris.thrift.TPrimitiveType;
+
+import org.apache.arrow.vector.ipc.ReadChannel;
+import org.apache.arrow.vector.ipc.message.MessageSerializer;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.nio.channels.Channels;
+import java.util.Arrays;
+import java.util.Collections;
+
+public class FlightSqlNativeVariantTest {
+ @Test
+ public void executionCannotPublishUtf8ForVariant() {
+ Assertions.assertThrows(org.apache.arrow.flight.FlightRuntimeException.class,
+ () -> FlightSqlSchemaHelper.withDorisTypeMetadata(Field.nullable("v", new ArrowType.Utf8()),
+ Type.VARIANT));
+ }
+
+ @Test
+ public void schemaKeepsExtensionAcrossIpc() throws Exception {
+ TColumnDesc variant = new TColumnDesc("item", TPrimitiveType.VARIANT);
+ variant.setIsAllowNull(true);
+ TColumnDesc array = new TColumnDesc("a", TPrimitiveType.ARRAY);
+ array.setChildren(Collections.singletonList(variant));
+ Field field = Deencapsulation.invoke(FlightSqlSchemaHelper.class, "buildField",
+ "test_db", "test_table", array);
+ Field child = field.getChildren().get(0);
+ Assertions.assertEquals(new ArrowType.Struct(), child.getType());
+ Assertions.assertEquals("arrow.parquet.variant", child.getMetadata().get("ARROW:extension:name"));
+ Assertions.assertEquals("metadata", child.getChildren().get(0).getName());
+ Assertions.assertEquals("value", child.getChildren().get(1).getName());
+ for (Field storage : child.getChildren()) {
+ Assertions.assertFalse(storage.isNullable());
+ Assertions.assertEquals(new ArrowType.Binary(), storage.getType());
+ }
+ Schema schema = new Schema(Collections.singletonList(field));
+ try (ReadChannel channel = new ReadChannel(Channels.newChannel(
+ new ByteArrayInputStream(schema.serializeAsMessage())))) {
+ Assertions.assertEquals(schema, MessageSerializer.deserializeSchema(channel));
+ }
+ }
+
+ @Test
+ public void nestedDiscoveryPreservesNativeVariantAndUuid() {
+ TColumnDesc variant = new TColumnDesc("item", TPrimitiveType.VARIANT);
+ variant.setIsAllowNull(true);
+ TColumnDesc array = new TColumnDesc("a", TPrimitiveType.ARRAY);
+ array.setChildren(Collections.singletonList(variant));
+ TColumnDesc map = new TColumnDesc("m", TPrimitiveType.MAP);
+ map.setChildren(Arrays.asList(new TColumnDesc("key", TPrimitiveType.STRING), variant));
+ TColumnDesc uuid = new TColumnDesc("id", TPrimitiveType.UUID);
+ TColumnDesc struct = new TColumnDesc("s", TPrimitiveType.STRUCT);
+ struct.setChildren(Arrays.asList(array, map, uuid));
+ Field result = Deencapsulation.invoke(FlightSqlSchemaHelper.class, "buildField",
+ "test_db", "test_table", struct);
+ for (Field leaf : Arrays.asList(result.getChildren().get(0).getChildren().get(0),
+ result.getChildren().get(1).getChildren().get(0).getChildren().get(1))) {
+ Assertions.assertEquals("arrow.parquet.variant", leaf.getMetadata().get("ARROW:extension:name"));
+ Assertions.assertEquals(new ArrowType.Struct(), leaf.getType());
+ }
+ // Flight's Variant specialization must retain master's UUID extension mapping.
+ Assertions.assertEquals(org.apache.arrow.vector.extension.UuidType.INSTANCE,
+ result.getChildren().get(2).getType());
+ Assertions.assertFalse(result.getChildren().get(1).getChildren().get(0)
+ .getChildren().get(0).isNullable());
+ }
+
+}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/planner/ResultSinkMetadataTest.java b/fe/fe-core/src/test/java/org/apache/doris/planner/ResultSinkMetadataTest.java
new file mode 100644
index 00000000000..d41c1062155
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/planner/ResultSinkMetadataTest.java
@@ -0,0 +1,38 @@
+// 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.
+
+package org.apache.doris.planner;
+
+import org.apache.doris.thrift.TResultSink;
+import org.apache.doris.thrift.TResultSinkType;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class ResultSinkMetadataTest {
+ @Test
+ void onlyUpgradedFlightPlannerOptsIntoExtendedMetadata() {
+ Assertions.assertFalse(new TResultSink().isEnableArrowTypeMetadata());
+ TResultSink flight = new ResultSink(new PlanNodeId(0), TResultSinkType.ARROW_FLIGHT_PROTOCOL)
+ .toThrift().getResultSink();
+ Assertions.assertTrue(flight.isSetEnableArrowTypeMetadata());
+ Assertions.assertTrue(flight.isEnableArrowTypeMetadata());
+ TResultSink mysql = new ResultSink(new PlanNodeId(0), TResultSinkType.MYSQL_PROTOCOL)
+ .toThrift().getResultSink();
+ Assertions.assertFalse(mysql.isSetEnableArrowTypeMetadata());
+ }
+}
diff --git a/gensrc/proto/data.proto b/gensrc/proto/data.proto
index 6ee532955e4..66ee8c5e1c1 100644
--- a/gensrc/proto/data.proto
+++ b/gensrc/proto/data.proto
@@ -70,6 +70,8 @@ message PColumnMeta {
optional int32 variant_max_subcolumns_count = 10 [default = 0];
optional bool variant_enable_doc_mode = 11 [default = false];
optional bool variant_is_v2 = 12 [default = false];
+ // Older blocks omit the declared VARBINARY length and retain the unspecified sentinel.
+ optional int32 varbinary_length = 13 [default = -1];
}
message PBlock {
diff --git a/gensrc/thrift/DataSinks.thrift b/gensrc/thrift/DataSinks.thrift
index bd44c8bad29..4ac527df0ef 100644
--- a/gensrc/thrift/DataSinks.thrift
+++ b/gensrc/thrift/DataSinks.thrift
@@ -233,6 +233,8 @@ struct TResultSink {
2: optional TResultFileSinkOptions file_options; // deprecated
// [deprecated] Two-phase read is replaced by TopN lazy materialization.
3: optional TFetchOption fetch_option;
+ // Absent for old FEs, whose Flight schema comparison requires legacy field metadata.
+ 4: optional bool enable_arrow_type_metadata = false;
}
struct TResultFileSink {
diff --git a/regression-test/data/external_table_p0/adbc/test_adbc_type_semantic.out b/regression-test/data/external_table_p0/adbc/test_adbc_type_semantic.out
index 13c24cd51f7..ba93aa962d8 100644
--- a/regression-test/data/external_table_p0/adbc/test_adbc_type_semantic.out
+++ b/regression-test/data/external_table_p0/adbc/test_adbc_type_semantic.out
@@ -49,10 +49,15 @@ c_ip4 int Yes true \N
-- !desc_variant --
id int No true \N
-c_var text Yes true \N
+c_var variant<PROPERTIES ("variant_max_subcolumns_count" = "0","variant_enable_typed_paths_to_sparse" = "false","variant_max_sparse_column_statistics_size" = "10000","variant_sparse_hash_shard_count" = "1")> Yes true \N
-- !select_variant --
1 {"a":1,"b":"two"}
2 {"nested":{"deep":3}}
3 \N
+-- !variant_subcolumn --
+1 1
+2 \N
+3 \N
+
diff --git a/regression-test/suites/arrow_flight_sql_p0/test_flight_native_variant.groovy b/regression-test/suites/arrow_flight_sql_p0/test_flight_native_variant.groovy
new file mode 100644
index 00000000000..6fb755585cf
--- /dev/null
+++ b/regression-test/suites/arrow_flight_sql_p0/test_flight_native_variant.groovy
@@ -0,0 +1,174 @@
+// 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.com.google.protobuf.Any
+import org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.flight.CallOptions
+import org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.flight.FlightClient
+import org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.flight.Location
+import org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.flight.sql.FlightSqlClient
+import org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.flight.sql.impl.FlightSql
+import org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.memory.RootAllocator
+
+import java.util.concurrent.TimeUnit
+
+suite("test_flight_native_variant", "arrow_flight_sql") {
+ def frontend = jdbc_sql_return_maparray("SHOW FRONTENDS").find {
+ it.IsMaster.toString().equalsIgnoreCase("true") && it.Alive.toString().equalsIgnoreCase("true")
+ }
+ assertNotNull(frontend)
+ assertTrue(frontend.ArrowFlightSqlPort.toString().toInteger() > 0)
+ def database = jdbc_sql("SELECT DATABASE()")[0][0]
+ def table = "${database}.flight_native_variant_input"
+ def allocator = new RootAllocator(Long.MAX_VALUE)
+ def feClient = FlightClient.builder(allocator,
+ Location.forGrpcInsecure(frontend.Host.toString(), frontend.ArrowFlightSqlPort.toString().toInteger())).build()
+ def client = new FlightSqlClient(feClient)
+ def auth
+ def read = { String query, Closure inspect, boolean parallel = false, int resultBackendCount = 1 ->
+ int count = 0
+ def info = client.execute(query, auth)
+ assertFalse(info.endpoints.isEmpty())
+ // Multiple buckets alone do not prove coverage of native output from different result BEs.
+ if (parallel) {
+ def resultAddresses = info.endpoints.collect { endpoint ->
+ def fields = Any.parseFrom(endpoint.ticket.bytes).unpack(FlightSql.TicketStatementQuery.class)
+ .statementHandle.toStringUtf8().split("&")
+ "${fields[1]}:${fields[2]}".toString()
+ }
+ assertEquals(info.endpoints.size(), resultAddresses.toSet().size(), "Duplicate result backends")
+ if (resultBackendCount > 1) {
+ assertTrue(info.endpoints.size() > 1, "Expected multiple native Variant result backends")
+ }
+ }
+ info.endpoints.each { endpoint ->
+ FlightClient.builder(allocator, endpoint.locations[0]).build().withCloseable { beClient ->
+ beClient.getStream(endpoint.ticket, auth, CallOptions.timeout(30, TimeUnit.SECONDS)).withCloseable { stream ->
+ assertEquals(info.schema, stream.schema, "Result endpoints must share the published schema")
+ while (stream.next()) {
+ inspect(stream.root)
+ count += stream.root.rowCount
+ }
+ }
+ }
+ }
+ count
+ }
+ def executeSetting = { String query -> read(query, { root -> }) }
+ try {
+ auth = feClient.authenticateBasicToken(context.config.otherConfigs.get("extArrowFlightSqlUser"),
+ context.config.otherConfigs.get("extArrowFlightSqlPassword")).get()
+ executeSetting("SET enable_sql_cache=false")
+ executeSetting("SET enable_nereids_distribute_planner=true")
+ executeSetting("SET parallel_pipeline_task_num=8")
+ jdbc_sql("DROP TABLE IF EXISTS ${table}")
+ jdbc_sql("""CREATE TABLE ${table} (id INT, v VARIANT)
+ DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 60
+ PROPERTIES("replication_num"="1")""")
+ jdbc_sql("""INSERT INTO ${table}
+ SELECT number + 1, parse_to_variant(CASE number % 4
+ WHEN 0 THEN '42' WHEN 1 THEN '"text"'
+ WHEN 2 THEN '{"a":[1,null,"x"]}' ELSE NULL END)
+ FROM numbers("number"="60")""")
+ def resultBackendCount = jdbc_sql_return_maparray("SHOW TABLETS FROM ${table}")
+ .collect { it.BackendId }.unique().size()
+ [false, true].each { parallel ->
+ executeSetting("SET enable_parallel_result_sink=${parallel}")
+ // Clients needing JSON strings request text through an explicit SQL conversion.
+ assertEquals(60, read("SELECT CAST(v AS STRING) AS text_value FROM ${table}", { root ->
+ assertEquals("Utf8", root.getVector(0).field.type.toString())
+ }))
+ def seen = []
+ assertEquals(60, read("SELECT id, v FROM ${table}", { root ->
+ def vector = root.getVector(1)
+ def field = vector.field
+ assertEquals("arrow.parquet.variant", field.metadata.get("ARROW:extension:name"))
+ assertEquals("Struct", field.type.toString())
+ assertEquals(["metadata", "value"], field.children.collect { it.name })
+ field.children.each { child ->
+ assertFalse(child.nullable)
+ assertEquals("Binary", child.type.toString())
+ }
+ for (int i = 0; i < root.rowCount; i++) {
+ int id = root.getVector(0).get(i)
+ seen.add(id)
+ assertEquals(id % 4 == 0, vector.isNull(i))
+ if (id % 4 != 0) {
+ assertTrue(vector.getChild("metadata").get(i).length > 0)
+ assertTrue(vector.getChild("value").get(i).length > 0)
+ if (id % 4 == 1) {
+ assertEquals([12, 42], vector.getChild("value").get(i).collect { it & 0xff })
+ }
+ }
+ }
+ }, parallel, resultBackendCount))
+ assertEquals((1..60).toList(), seen.sort())
+ def scannedColumns = "v, ARRAY(v) AS a, MAP('key', v) AS m, STRUCT(v) AS s"
+ // Master obtains result schemas through execution; cover nested and empty results there.
+ ["SELECT CAST(42 AS VARIANT) AS v",
+ "SELECT ${scannedColumns} FROM ${table} WHERE id = 1",
+ "SELECT ${scannedColumns} FROM ${table} WHERE id < 0"].eachWithIndex { query, index ->
+ assertEquals(index == 2 ? 0 : 1, read(query.toString(), { root ->
+ def leaves = [root.schema.fields[0]]
+ if (index > 0) {
+ leaves.add(root.schema.fields[1].children[0])
+ leaves.add(root.schema.fields[2].children[0].children[1])
+ leaves.add(root.schema.fields[3].children[0])
+ }
+ leaves.each { field ->
+ assertEquals("Struct", field.type.toString())
+ assertEquals("arrow.parquet.variant", field.metadata.get("ARROW:extension:name"))
+ }
+ }))
+ }
+ }
+ // Each row owns its wire dictionary; unrelated keys must not multiply Arrow metadata.
+ def keyExpression = "CONCAT(REPEAT('k', 244), LPAD(CAST(number AS STRING), 6, '0'))"
+ def variantExpression = """parse_to_variant(CONCAT('{"', ${keyExpression}, '":', CAST(number AS STRING), '}'))"""
+ def metadataRows = []
+ assertEquals(128, read("SELECT number AS id, ${variantExpression} AS v FROM numbers(\"number\"=\"128\")", { root ->
+ def variant = root.getVector(1)
+ assertEquals("arrow.parquet.variant", variant.field.metadata.get("ARROW:extension:name"))
+ for (int i = 0; i < root.rowCount; ++i) {
+ metadataRows.add(root.getVector(0).get(i).intValue())
+ assertTrue(variant.getChild("metadata").get(i).length < 270,
+ "A row must not repeat the batch dictionary")
+ }
+ }))
+ assertEquals((0..<128).toList(), metadataRows.sort())
+ // A folded constant must use the same wire representation as a scanned Variant column.
+ assertEquals(1, read("SELECT parse_to_variant('42') AS v", { root ->
+ assertEquals("arrow.parquet.variant", root.getVector(0).field.metadata.get("ARROW:extension:name"))
+ }))
+ assertEquals(0, read("SELECT parse_to_variant(CAST(id AS STRING)) AS v FROM ${table} WHERE id < 0", { root -> }))
+ } finally {
+ try {
+ if (auth != null) {
+ feClient.closeSession(new org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.flight.CloseSessionRequest(), auth)
+ }
+ } finally {
+ try {
+ client.close()
+ } finally {
+ try {
+ allocator.close()
+ } finally {
+ jdbc_sql("DROP TABLE IF EXISTS ${table}")
+ }
+ }
+ }
+ }
+}
diff --git a/regression-test/suites/arrow_flight_sql_p0/test_get_tables_schema.groovy b/regression-test/suites/arrow_flight_sql_p0/test_get_tables_schema.groovy
index 1bc3779539c..6fd2148192e 100644
--- a/regression-test/suites/arrow_flight_sql_p0/test_get_tables_schema.groovy
+++ b/regression-test/suites/arrow_flight_sql_p0/test_get_tables_schema.groovy
@@ -222,7 +222,9 @@ suite("test_get_tables_schema") {
c_varchar: Utf8 {VARCHAR}
c_string: Utf8 {STRING}
c_json: Utf8 {JSON}
- c_variant: Utf8 {VARIANT}
+ c_variant: Struct {VARIANT}
+ metadata: Binary not null
+ value: Binary not null
c_ipv4: Int(32, true) {IPV4}
c_ipv6: Utf8 {IPV6}
c_array_int: List {ARRAY}
@@ -246,6 +248,10 @@ suite("test_get_tables_schema") {
item: Decimal(10, 3, 128) {DECIMAL64, precision=10, scale=3}
""")
+ // Native Variant must advertise its extension identity, not just a matching struct shape.
+ def variantField = tableSchema(allTypes).getFields().find { it.getName() == "c_variant" }
+ assertEquals("arrow.parquet.variant", variantField.getMetadata().get("ARROW:extension:name"))
+
// BITMAP / HLL / QUANTILE_STATE are opaque bytes. AGG_STATE is described as Null (kept as is,
// see the header).
check(aggTypes, """
diff --git a/regression-test/suites/external_table_p0/adbc/test_adbc_type_semantic.groovy b/regression-test/suites/external_table_p0/adbc/test_adbc_type_semantic.groovy
index 9fdb611bbbc..54349b3481e 100644
--- a/regression-test/suites/external_table_p0/adbc/test_adbc_type_semantic.groovy
+++ b/regression-test/suites/external_table_p0/adbc/test_adbc_type_semantic.groovy
@@ -25,7 +25,7 @@
//
// LARGEINT -> utf8 -> STRING value kept, arithmetic lost
// JSON -> utf8 -> STRING text kept, json functions lost
-// VARIANT -> utf8 -> STRING text kept, sub-column access lost
+// VARIANT -> native Arrow Variant -> VARIANT, including sub-column access
// IPV6 -> utf8 -> STRING text kept
// IPV4 -> int32 -> INT *** the value itself changes form ***
//
@@ -264,9 +264,9 @@ suite("test_adbc_type_semantic", "p0,external") {
qt_desc_variant """DESC ${catalogName}.${dbName}.t_variant"""
qt_select_variant """SELECT id, c_var FROM ${catalogName}.${dbName}.t_variant ORDER BY id"""
sameAsSourceText("t_variant", "c_var")
- assertEquals("text", externalTypeOf("t_variant", "c_var"),
- "VARIANT is serialised as Arrow utf8, so the external column is a string and its "
- + "sub-columns are no longer addressable through the catalog")
+ // The extension's binary storage must round-trip as Variant, not leak as two text fields.
+ assertTrue(externalTypeOf("t_variant", "c_var").startsWith("variant"))
+ qt_variant_subcolumn """SELECT id, CAST(c_var['a'] AS INT) FROM ${catalogName}.${dbName}.t_variant ORDER BY id"""
} finally {
sql """DROP CATALOG IF EXISTS ${catalogName}"""
sql """DROP DATABASE IF EXISTS internal.${dbName} FORCE"""
diff --git a/regression-test/suites/flink_connector_p0/flink_connector_type.groovy b/regression-test/suites/flink_connector_p0/flink_connector_type.groovy
index 26d4b33b1b8..6e6a8d25744 100644
--- a/regression-test/suites/flink_connector_p0/flink_connector_type.groovy
+++ b/regression-test/suites/flink_connector_p0/flink_connector_type.groovy
@@ -24,13 +24,15 @@ import org.awaitility.Awaitility
suite("flink_connector_type") {
+ def inputTable = "test_types_input"
def tableName1 = "test_types_source"
def tableName2 = "test_types_sink"
+ sql """DROP TABLE IF EXISTS ${inputTable}"""
sql """DROP TABLE IF EXISTS ${tableName1}"""
sql """DROP TABLE IF EXISTS ${tableName2}"""
sql """
- CREATE TABLE `test_types_source` (
+ CREATE TABLE `${inputTable}` (
`id` int,
`c1` boolean,
`c2` tinyint,
@@ -59,9 +61,9 @@ PROPERTIES (
);
""";
- sql """CREATE TABLE `test_types_sink` like `test_types_source` """
+ sql """CREATE TABLE `${tableName2}` like `${inputTable}` """
- sql """ INSERT INTO `test_types_source`
+ sql """ INSERT INTO `${inputTable}`
VALUES
(
1,
@@ -106,6 +108,17 @@ VALUES
'{"B":"variant_value1"}'
);"""
+ // The connector declares c18 as STRING and cannot decode native Arrow Variant.
+ // Materialize an explicit text projection while retaining Variant ingestion in the sink.
+ sql """
+ CREATE TABLE `${tableName1}`
+ DISTRIBUTED BY HASH(`id`) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ AS SELECT id, c1, c2, c3, c4, c5, c6, c7, c8, c9,
+ c10, c11, c12, c13, c14, c15, c16, c17, CAST(c18 AS STRING) AS c18
+ FROM `${inputTable}`
+ """
+
def thisDb = sql """select database()""";
thisDb = thisDb[0][0];
logger.info("current database is ${thisDb}");
@@ -143,19 +156,21 @@ VALUES
run_cmd.addAll(addOpens.tokenize())
run_cmd.addAll(["-cp", jarName, "org.apache.doris.FlinkConnectorTypeCase",
"--doris-fe-address", context.config.feHttpAddress,
- "--doris-database", "regression_test_flink_connector_p0",
+ "--doris-database", thisDb,
"--doris-user", context.config.feHttpUser,
"--doris-password", context.config.feHttpPassword])
run_cmd.addAll(getDorisConnectorTlsArgs())
logger.info("run_cmd : ${run_cmd.join(' ')}")
- def run_flink_jar = run_cmd.execute().getText()
- logger.info("result: $run_flink_jar")
+ // Drain both streams and check the exit code so a failed Flink job cannot look successful.
+ def flinkProcess = new ProcessBuilder(run_cmd.collect { it.toString() }).redirectErrorStream(true).start()
+ logger.info("result: ${flinkProcess.text}")
+ assertEquals(0, flinkProcess.waitFor())
// The publish in the commit phase is asynchronous
Awaitility.await().atMost(30, SECONDS).pollInterval(1, SECONDS).await().until(
{
def resultTbl = sql """ select count(1) from test_types_sink"""
logger.info("retry test_types_sink count: $resultTbl")
- resultTbl.size() >= 1
+ resultTbl[0][0] == 2
})
logger.info("flink job execute finished.");
diff --git a/samples/arrow-flight-sql/python/README.md b/samples/arrow-flight-sql/python/README.md
index e729054b813..7759b44d314 100644
--- a/samples/arrow-flight-sql/python/README.md
+++ b/samples/arrow-flight-sql/python/README.md
@@ -35,7 +35,65 @@ under the License.
Section 6.1 of https://github.com/apache/doris/issues/25514 is the performance test
results of the doris arrow flight sql using python.
+## Logical type metadata
+
+Some Doris types share an Arrow storage type with ordinary strings or integers.
+The `doris_type` field metadata identifies `LARGEINT`, `IPV4`, `IPV6`, `UUID`, `JSON`, and
+`VARIANT`, including fields nested in arrays, maps, and structs. Consumers should
+inspect each nested Arrow field instead of inferring a type from its value.
+
+LARGEINT retains its Arrow string encoding, including the full signed 128-bit
+range. PyArrow does not automatically convert custom metadata into Python types;
+a client can use `doris_type=LARGEINT` to safely convert that field's non-NULL
+values with `int(value)`, while leaving ordinary STRING fields unchanged.
+
+For types whose physical Arrow layout is unchanged, queries planned by older FEs
+retain the legacy metadata layout during rolling upgrades. Upgraded FEs advertise complete logical type metadata in the FlightInfo
+schema even when some BEs are older. An older BE's DoGet stream may still omit
+these markers; use the FlightInfo schema as the logical type reference until the
+upgrade finishes. No session setting is required.
+
+Run `python test_nested_type_metadata.py` with `DORIS_FLIGHT_SQL_URI` and optional
+`DORIS_USER` / `DORIS_PASSWORD` to verify the metadata and values against a cluster.
+The tests execute only read-only queries.
+
# Notes
For more details, refer to [Python Usage] in the document https://doris.apache.org/zh-CN/docs/dev/db-connect/arrow-flight-sql-connect
-
\ No newline at end of file
+
+
+# Native Variant V2 results
+
+Arrow Flight SQL / ADBC returns Variant V2 as native binary values automatically:
+
+```sql
+SELECT parse_to_variant('{"key":42}') AS v;
+-- Request text explicitly when the client needs JSON strings.
+SELECT CAST(variant_column AS STRING) FROM example_table;
+```
+
+Variant V2 fields, including nested fields, use the `arrow.parquet.variant` extension
+with `struct<metadata: binary not null, value: binary not null>` storage. SQL NULL is
+a null struct. Variant null is a non-null struct containing the encoded null value.
+V2 values retain their physical scalar types and decimal scales. Each Arrow row carries
+only the dictionary keys it uses, rather than copying keys from unrelated rows.
+
+Master uses Variant V2 throughout. Native Variant queries require backends supporting
+this binary layout; a backend returning UTF8 is rejected. Use an explicit SQL cast
+to STRING for text output.
+Native encoding accepts up to 128 nested levels.
+
+ADBC can transport this schema and its binary values. A client without a registered
+Variant extension exposes the struct with `ARROW:extension:name` field metadata.
+Receiving native VARIANT does not automatically decode it to Python dictionaries or
+pandas objects; use a Parquet Variant decoder, or explicitly cast the result to STRING.
+
+To check ADBC query and partition reads against a running Variant V2 cluster:
+
+```bash
+pip install adbc_driver_flightsql pyarrow
+export DORIS_FLIGHT_URI='grpc://localhost:8815'
+export DORIS_USER='root'
+# Set DORIS_PASSWORD in the environment if authentication requires it.
+python test_variant.py
+```
diff --git a/samples/arrow-flight-sql/python/test_nested_type_metadata.py b/samples/arrow-flight-sql/python/test_nested_type_metadata.py
new file mode 100644
index 00000000000..6915447bbac
--- /dev/null
+++ b/samples/arrow-flight-sql/python/test_nested_type_metadata.py
@@ -0,0 +1,108 @@
+# 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.
+
+"""Run with DORIS_FLIGHT_SQL_URI and optional DORIS_USER/DORIS_PASSWORD.
+
+LARGEINT retains its string storage encoding. PyArrow does not interpret custom
+field metadata automatically; applications can use doris_type to distinguish
+these strings from text and convert them to Python integers without losing range.
+"""
+
+import os
+import unittest
+
+import adbc_driver_flightsql.dbapi as flight_sql
+import adbc_driver_manager
+import pyarrow as pa
+
+
+@unittest.skipUnless(os.getenv("DORIS_FLIGHT_SQL_URI"), "Set DORIS_FLIGHT_SQL_URI to run against Doris")
+class NestedTypeMetadataTest(unittest.TestCase):
+ def setUp(self):
+ self.connection = flight_sql.connect(
+ uri=os.environ["DORIS_FLIGHT_SQL_URI"],
+ db_kwargs={
+ adbc_driver_manager.DatabaseOptions.USERNAME.value: os.getenv("DORIS_USER", "root"),
+ adbc_driver_manager.DatabaseOptions.PASSWORD.value: os.getenv("DORIS_PASSWORD", ""),
+ },
+ )
+ self.addCleanup(self.connection.close)
+ self.cursor = self.connection.cursor()
+ self.addCleanup(self.cursor.close)
+
+ def assert_logical_type(self, field, name):
+ self.assertEqual((field.metadata or {}).get(b"doris_type"), name.encode())
+
+ def query(self, sql):
+ self.cursor.execute(sql)
+ table = self.cursor.fetch_arrow_table()
+ table.validate(full=True)
+ return table
+
+ def test_largeint_metadata_and_full_range(self):
+ table = self.query("""
+ SELECT CAST('170141183460469231731687303715884105727' AS LARGEINT) AS scalar_value,
+ named_struct('number', CAST(17 AS LARGEINT), 'text', '17') AS struct_value,
+ array(CAST('-170141183460469231731687303715884105728' AS LARGEINT),
+ CAST(NULL AS LARGEINT)) AS array_value,
+ map(CAST(17 AS LARGEINT), CAST(19 AS LARGEINT)) AS map_value
+ """)
+ self.assert_logical_type(table.schema.field("scalar_value"), "LARGEINT")
+ structure = table.schema.field("struct_value").type
+ self.assert_logical_type(structure.field("number"), "LARGEINT")
+ self.assertNotIn(b"doris_type", structure.field("text").metadata or {})
+ self.assert_logical_type(table.schema.field("array_value").type.value_field, "LARGEINT")
+ mapping = table.schema.field("map_value").type
+ self.assert_logical_type(mapping.key_field, "LARGEINT")
+ self.assert_logical_type(mapping.item_field, "LARGEINT")
+ self.assertFalse(mapping.key_field.nullable)
+ self.assertTrue(pa.types.is_string(table.schema.field("scalar_value").type))
+ row = table.to_pylist()[0]
+ self.assertEqual(row["scalar_value"], str(2**127 - 1))
+ self.assertEqual(row["array_value"], [str(-(2**127)), None])
+ self.assertEqual(row["struct_value"], {"number": "17", "text": "17"})
+ self.assertEqual(row["map_value"], [("17", "19")])
+
+ def test_ipv4_metadata(self):
+ table = self.query("""
+ SELECT array(CAST('192.0.2.1' AS IPV4)) AS ip4
+ """)
+ self.assert_logical_type(table.schema.field("ip4").type.value_field, "IPV4")
+
+ def test_ipv6_metadata(self):
+ table = self.query("""
+ SELECT named_struct('address', CAST('2001:db8::1' AS IPV6)) AS ip6
+ """)
+ self.assert_logical_type(table.schema.field("ip6").type.field("address"), "IPV6")
+
+ def test_json_metadata(self):
+ # ARRAY/MAP constructors reject JSON; ARRAY_REPEAT preserves its element type.
+ table = self.query("""
+ SELECT array_repeat(CAST('{"n":1}' AS JSON), 1) AS json_value
+ """)
+ self.assert_logical_type(table.schema.field("json_value").type.value_field, "JSON")
+
+ def test_variant_metadata(self):
+ # Preserve the Variant element type when constructing the nested result.
+ table = self.query("""
+ SELECT array_repeat(CAST('{"n":1}' AS VARIANT), 1) AS variant_value
+ """)
+ self.assert_logical_type(table.schema.field("variant_value").type.value_field, "VARIANT")
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/samples/arrow-flight-sql/python/test_variant.py b/samples/arrow-flight-sql/python/test_variant.py
new file mode 100644
index 00000000000..4c53bebfeb4
--- /dev/null
+++ b/samples/arrow-flight-sql/python/test_variant.py
@@ -0,0 +1,84 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+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.
+"""
+
+"""Run against a Variant V2 cluster with DORIS_FLIGHT_URI, DORIS_USER and DORIS_PASSWORD."""
+
+import os
+import unittest
+
+import adbc_driver_flightsql
+import adbc_driver_manager
+import pyarrow as pa
+
+
+class NativeVariantTest(unittest.TestCase):
+ def test_query_and_partitions(self):
+ uri = os.environ["DORIS_FLIGHT_URI"]
+ options = {
+ adbc_driver_manager.DatabaseOptions.USERNAME.value: os.environ.get("DORIS_USER", "root"),
+ adbc_driver_manager.DatabaseOptions.PASSWORD.value: os.environ.get("DORIS_PASSWORD", ""),
+ }
+ query = """SELECT 1 AS id, parse_to_variant('42') AS v
+ UNION ALL SELECT 2, parse_to_variant(CAST(NULL AS STRING))"""
+ with adbc_driver_flightsql.connect(uri, db_kwargs=options) as database:
+ with adbc_driver_manager.AdbcConnection(database) as connection:
+ def execute(sql):
+ with adbc_driver_manager.AdbcStatement(connection) as statement:
+ statement.set_sql_query(sql)
+ stream, _ = statement.execute_query()
+ return pa.RecordBatchReader._import_from_c(stream.address).read_all()
+
+ execute("SET enable_sql_cache=false")
+ for parallel in (False, True):
+ execute(f"SET enable_parallel_result_sink={str(parallel).lower()}")
+ table = execute(query)
+ self.check_result(table)
+ with adbc_driver_manager.AdbcStatement(connection) as statement:
+ statement.set_sql_query(query)
+ partitions, _, _ = statement.execute_partitions()
+ tables = []
+ for partition in partitions:
+ stream = connection.read_partition(partition)
+ tables.append(pa.RecordBatchReader._import_from_c(stream.address).read_all())
+ self.check_result(pa.concat_tables(tables))
+
+ def check_result(self, table):
+ table = table.sort_by("id")
+ self.assertEqual(table.num_rows, 2)
+ field = table.schema.field("v")
+ values = table.column("v").combine_chunks()
+ # Clients without a registered extension expose its storage plus field metadata.
+ if isinstance(values, pa.ExtensionArray):
+ self.assertEqual(values.type.extension_name, "arrow.parquet.variant")
+ values = values.storage
+ else:
+ self.assertEqual(field.metadata[b"ARROW:extension:name"], b"arrow.parquet.variant")
+ self.assertTrue(pa.types.is_struct(values.type))
+ self.assertEqual([child.name for child in values.type], ["metadata", "value"])
+ self.assertIsNone(values[1].as_py())
+ row = values[0].as_py()
+ self.assertTrue(row["metadata"])
+ # The Parquet Variant primitive INT8 tag is 3 << 2, followed by its signed byte.
+ self.assertEqual(row["value"], bytes([12, 42]))
+
+
+if __name__ == "__main__":
+ unittest.main()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org