This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 7865646f0bb [improvement](be) Support Arrow loads and schema inference
in FileScannerV2 (#69668)
7865646f0bb is described below
commit 7865646f0bbd31a011edba2acf21dfabe3a65ab0
Author: Gabriel <[email protected]>
AuthorDate: Wed Oct 28 11:53:56 2026 +0800
[improvement](be) Support Arrow loads and schema inference in FileScannerV2
(#69668)
### What problem does this PR solve?
Arrow Stream Load still selects FileScanner V1, and CSV/TEXT/JSON schema
discovery still constructs legacy readers even when the subsequent scan
uses V2.
Add a native Arrow IPC FileReader to the V2 load path. It decodes record
batches incrementally, supports concatenated streams and compressed IPC,
uses tracked allocations, and retains load expressions, defaults,
strict-mode checks, and WHERE filtering. Source timestamp units and
timezones are preserved for Doris SerDe rounding, including nested
values.
Stream Load plans explicitly set `enable_file_scanner_v2=true` in their
query options because these plans construct options independently of
query sessions.
Arrow binding preserves positional STRUCT children, DATE64 range
validation, and unsigned IPv4 address bits. Before checked recursive
casts, ancestor validity masks hide NULL-struct payloads and
NULL-list/map ranges are compacted. Ordinary numeric overflow checks
remain enabled.
A load-specific schema converter preserves native VARIANT carriers and
extensions recursively, including SQL NULLs, and binds AGG_STATE through
its actual serialization type. Fixed-length aggregate states use the
serialization column's byte width; invalid binary lengths fail before
SerDe materialization. Recursive NULL masking skips unchanged children
and leaves incompatible container kinds to explicit cast errors.
Nullable fixed-size-list inputs are viewed through 64-bit variable
offsets before compaction, so NULL parents drop complete child ranges
without synthesizing NULLs in non-nullable STRUCT fields or native
VARIANT carriers. DECIMALV2 retains decimal128 wire scales for its
existing scale-nine rescaling and truncation rules; other decimal types
retain checked conversions.
Keep the existing empty-file schema RPC diagnostics, including empty
LZOP data and header-name versus inferred-column-number errors.
Route CSV, TEXT, and JSON schema discovery through V2 readers and the
shared schema entry point, alongside the existing native Parquet/ORC
readers. Preserve header handling, nullable string inference, JSON
root/path options, and independent sampling of HTTP Stream Load data.
Also avoid consuming a rooted JSON array's ondemand cursor before
jsonpaths materialization.
### Release note
Arrow Stream Load and CSV/TEXT/JSON schema discovery now use FileScanner
V2 readers.
### Check List (For Author)
- Test
- [x] Regression test
- [x] Unit Test
- [x] Manual test
- ASAN BE build and 221 focused unit tests passed, including scanner
routing, native schema inference, Arrow IPC loading, and existing reader
coverage. Focused cases cover positional STRUCT names, DATE64 range
checks, unsigned IPv4, NULL container payloads, numeric overflow, native
VARIANT storage/extensions, aggregate-state widths, incompatible
container input, empty LZOP, NULL fixed-size lists with non-nullable
descendants, and DECIMALV2 rescaling with DECIMALV3 rejection controls.
- Six regression suites passed against a locally built FE/ASAN BE: Arrow
Stream Load, V2 HTTP schema discovery, and existing JSON, compressed
CSV, CSV-with-names, and CSV-with-names-and-types suites. Results were
compared with expected output.
- clang-format 16, changed-line clang-tidy, and build hygiene checks
passed.
- FE Checkstyle and the FE production build passed after adding the
explicit Stream Load query option.
- Behavior changed:
- [x] Yes. Arrow loads and text/JSON schema inference use native V2
readers.
- Does this need documentation?
- [x] No. Existing load interfaces and options are retained.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
be/src/exec/operator/file_scan_operator.cpp | 1 +
be/src/exec/scan/file_scanner_v2.cpp | 5 +-
be/src/format_v2/arrow/arrow_stream_reader.cpp | 527 ++++++++++++
be/src/format_v2/arrow/arrow_stream_reader.h | 87 ++
be/src/format_v2/delimited_text/csv_reader.cpp | 5 +-
.../delimited_text/delimited_text_reader.cpp | 50 +-
.../delimited_text/delimited_text_reader.h | 3 +
be/src/format_v2/json/json_reader.cpp | 88 +-
be/src/format_v2/json/json_reader.h | 3 +-
be/src/format_v2/schema_reader.cpp | 53 +-
be/src/format_v2/table/load_reader.cpp | 7 +
be/src/service/internal_service.cpp | 85 +-
be/test/exec/scan/file_load_scanner_v2_test.cpp | 4 +-
be/test/exec/scan/file_scanner_v2_test.cpp | 14 +
be/test/format_v2/json/json_reader_test.cpp | 14 +
be/test/format_v2/schema_reader_test.cpp | 217 +++++
be/test/format_v2/table/load_reader_test.cpp | 950 ++++++++++++++++++++-
.../nereids/load/NereidsStreamLoadPlanner.java | 2 +
.../http_stream/test_http_stream_v2_schema.out | 15 +
.../test_arrow_stream_load_uppercase_column.out | 4 +
.../http_stream/test_http_stream_v2_schema.groovy | 69 ++
.../test_arrow_stream_load_uppercase_column.groovy | 34 +
22 files changed, 2141 insertions(+), 96 deletions(-)
diff --git a/be/src/exec/operator/file_scan_operator.cpp
b/be/src/exec/operator/file_scan_operator.cpp
index 7e6f2ad1579..5f6d41c8b1d 100644
--- a/be/src/exec/operator/file_scan_operator.cpp
+++ b/be/src/exec/operator/file_scan_operator.cpp
@@ -181,6 +181,7 @@ bool FileScanLocalState::should_use_file_scanner_v2(const
TQueryOptions& query_o
case TFileFormatType::FORMAT_PROTO:
case TFileFormatType::FORMAT_TEXT:
case TFileFormatType::FORMAT_WAL:
+ case TFileFormatType::FORMAT_ARROW:
return true;
default:
return false;
diff --git a/be/src/exec/scan/file_scanner_v2.cpp
b/be/src/exec/scan/file_scanner_v2.cpp
index b869533b38b..d9ab6a377fc 100644
--- a/be/src/exec/scan/file_scanner_v2.cpp
+++ b/be/src/exec/scan/file_scanner_v2.cpp
@@ -360,7 +360,10 @@ bool FileScannerV2::is_supported(const
TFileScanRangeParams& params, const TFile
} else if (format_type == TFileFormatType::FORMAT_ES_HTTP) {
return table_format_name(range) == "es";
} else if (format_type == TFileFormatType::FORMAT_ARROW) {
- return is_supported_arrow_table_format(range);
+ // A plain Arrow range is an IPC load pipe; query Arrow ranges retain
their connector.
+ return is_supported_arrow_table_format(range) ||
+ (table_format_name(range) == "NotSet" &&
params.__isset.src_tuple_id &&
+ params.__isset.file_type && params.file_type ==
TFileType::FILE_STREAM);
} else if (format_type == TFileFormatType::FORMAT_JNI) {
return is_supported_jni_table_format(range);
} else if (is_wal_format(format_type)) {
diff --git a/be/src/format_v2/arrow/arrow_stream_reader.cpp
b/be/src/format_v2/arrow/arrow_stream_reader.cpp
new file mode 100644
index 00000000000..c2a30f35f8a
--- /dev/null
+++ b/be/src/format_v2/arrow/arrow_stream_reader.cpp
@@ -0,0 +1,527 @@
+// 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_v2/arrow/arrow_stream_reader.h"
+
+#include <arrow/array.h>
+#include <arrow/buffer.h>
+#include <arrow/builder.h>
+#include <arrow/compute/api.h>
+#include <arrow/io/interfaces.h>
+#include <arrow/ipc/reader.h>
+#include <arrow/record_batch.h>
+#include <arrow/type.h>
+
+#include <utility>
+
+#include "core/block/block.h"
+#include "core/column/column_fixed_length_object.h"
+#include "core/data_type/data_type_agg_state.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_struct.h"
+#include "format/arrow/arrow_array_normalizer.h"
+#include "format/arrow/arrow_row_batch.h"
+#include "format/arrow/arrow_utils.h"
+#include "format/parquet/arrow_memory_pool.h"
+#include "format_v2/column_mapper.h"
+#include "format_v2/materialized_reader_util.h"
+#include "io/fs/tracing_file_reader.h"
+#include "runtime/descriptors.h"
+#include "runtime/file_scan_profile.h"
+#include "runtime/runtime_state.h"
+
+namespace doris::format::arrow_stream {
+
+namespace {
+
+class ArrowLoadSchemaConvertor final : public DorisArrowSchemaConvertor {
+public:
+ using DorisArrowSchemaConvertor::DorisArrowSchemaConvertor;
+
+ Status convert_to_arrow_type(const DataTypePtr& origin_type,
+ std::shared_ptr<arrow::DataType>* result)
const override {
+ const auto type = remove_nullable(origin_type);
+ // VARIANT output may be UTF-8, but its input SerDe requires the
native binary carrier,
+ // including when VARIANT is nested in an ARRAY, MAP, or STRUCT.
+ if (type->get_primitive_type() == TYPE_VARIANT) {
+ *result = arrow::struct_({arrow::field("metadata",
arrow::binary(), false),
+ arrow::field("value", arrow::binary(),
false)});
+ return Status::OK();
+ }
+ if (type->get_primitive_type() == TYPE_AGG_STATE) {
+ const auto& state = assert_cast<const DataTypeAggState&>(*type);
+ const auto serialized_type =
remove_nullable(state.get_serialized_type());
+ // FixedLengthObject's type omits its width. The serialization
column owns that
+ // contract (e.g. count(INT NOT NULL) uses eight bytes), not the
runtime state size.
+ if (serialized_type->get_primitive_type() ==
TYPE_FIXED_LENGTH_OBJECT) {
+ const auto column = state.create_column();
+ const auto width = assert_cast<const
ColumnFixedLengthObject&>(*column).item_size();
+ *result = arrow::fixed_size_binary(cast_set<int32_t>(width));
+ return Status::OK();
+ }
+ return convert_to_arrow_type(serialized_type, result);
+ }
+ return DorisArrowSchemaConvertor::convert_to_arrow_type(type, result);
+ }
+};
+
+arrow::Result<std::shared_ptr<arrow::DataType>> input_binding_type(
+ const std::shared_ptr<arrow::DataType>& wire,
+ const std::shared_ptr<arrow::DataType>& declared, const DataTypePtr&
doris_type) {
+ if (wire->Equals(declared)) {
+ return declared;
+ }
+ // SerDe owns datetime rounding and naive/aware interpretation. Casting
timestamp[us]
+ // to the declared timestamp[ms] first would reject .123789 instead of
rounding to .124.
+ if (wire->id() == arrow::Type::TIMESTAMP && declared->id() ==
arrow::Type::TIMESTAMP) {
+ return wire;
+ }
+ // DATE64 SerDe checks whole days and the Doris year range; DATE32
rebinding would bypass it.
+ if (wire->id() == arrow::Type::DATE64 && declared->id() ==
arrow::Type::DATE32) {
+ return wire;
+ }
+ const auto type =
remove_nullable(get_serialized_type(remove_nullable(doris_type)));
+ // A native VARIANT carrier is a logical leaf, not a Doris STRUCT.
Preserve its storage or
+ // extension unchanged and let its SerDe validate the carrier shape and
encoded payload.
+ if (type->get_primitive_type() == TYPE_VARIANT) {
+ return wire;
+ }
+ // DECIMALV2 SerDe rescales decimal128 input to scale nine, truncating
extra fractional
+ // digits toward zero (1.2345678901 -> 1.234567890). Other decimals retain
checked casts.
+ if (type->get_primitive_type() == TYPE_DECIMALV2 && wire->id() ==
arrow::Type::DECIMAL128) {
+ return wire;
+ }
+ // IPv4 is a 32-bit bit pattern, including addresses above INT32_MAX.
Numeric INT columns
+ // still use the checked conversion below instead of inheriting this
storage exception.
+ if (type->get_primitive_type() == TYPE_IPV4 && wire->id() ==
arrow::Type::UINT32) {
+ return wire;
+ }
+ if (declared->id() == arrow::Type::LIST &&
+ (wire->id() == arrow::Type::LIST || wire->id() ==
arrow::Type::LARGE_LIST ||
+ wire->id() == arrow::Type::FIXED_SIZE_LIST)) {
+ const auto& source = static_cast<const arrow::BaseListType&>(*wire);
+ const auto& target = static_cast<const arrow::ListType&>(*declared);
+ const auto& array = assert_cast<const DataTypeArray&>(*type);
+ ARROW_ASSIGN_OR_RAISE(auto child,
+ input_binding_type(source.value_type(),
target.value_type(),
+ array.get_nested_type()));
+ return arrow::list(source.value_field()->WithType(child));
+ }
+ if (wire->id() == arrow::Type::MAP && declared->id() == arrow::Type::MAP) {
+ const auto& source = static_cast<const arrow::MapType&>(*wire);
+ const auto& target = static_cast<const arrow::MapType&>(*declared);
+ const auto& map = assert_cast<const DataTypeMap&>(*type);
+ ARROW_ASSIGN_OR_RAISE(auto key, input_binding_type(source.key_type(),
target.key_type(),
+
map.get_key_type()));
+ ARROW_ASSIGN_OR_RAISE(auto item,
input_binding_type(source.item_type(), target.item_type(),
+
map.get_value_type()));
+ return
std::make_shared<arrow::MapType>(source.key_field()->WithType(key),
+
source.item_field()->WithType(item),
+ source.keys_sorted());
+ }
+ if (wire->id() == arrow::Type::STRUCT && declared->id() ==
arrow::Type::STRUCT) {
+ if (wire->num_fields() != declared->num_fields()) {
+ return arrow::Status::Invalid(
+ "Arrow STRUCT field count does not match the load schema");
+ }
+ const auto& structure = assert_cast<const DataTypeStruct&>(*type);
+ arrow::FieldVector fields;
+ for (int i = 0; i < wire->num_fields(); ++i) {
+ ARROW_ASSIGN_OR_RAISE(auto child,
input_binding_type(wire->field(i)->type(),
+
declared->field(i)->type(),
+
structure.get_elements()[i]));
+ // Struct SerDe binds children by position. Arrow casts bind by
name, so retain the
+ // wire names to prevent a case/name difference from synthesizing
NULL children.
+ fields.push_back(wire->field(i)->WithType(child));
+ }
+ return arrow::struct_(fields);
+ }
+ return declared;
+}
+
+arrow::Result<std::shared_ptr<arrow::Array>> mask_hidden_children(
+ std::shared_ptr<arrow::Array> array, const
std::shared_ptr<arrow::DataType>& binding,
+ arrow::compute::ExecContext* context);
+
+arrow::Result<std::shared_ptr<arrow::Array>> mask_struct_children(
+ const std::shared_ptr<arrow::Array>& array, const
std::shared_ptr<arrow::DataType>& binding,
+ arrow::compute::ExecContext* context) {
+ const auto& structure = static_cast<const arrow::StructArray&>(*array);
+ auto data = array->data()->Copy();
+ for (int i = 0; i < structure.num_fields(); ++i) {
+ // A NULL struct can have valid child payloads. Combine ancestor
validity before
+ // checked casts can inspect those logically hidden values.
+ ARROW_ASSIGN_OR_RAISE(auto child, structure.GetFlattenedField(i,
context->memory_pool()));
+ ARROW_ASSIGN_OR_RAISE(
+ child, mask_hidden_children(std::move(child),
binding->field(i)->type(), context));
+ data->child_data[i] = child->data();
+ }
+ return arrow::MakeArray(data);
+}
+
+arrow::Result<std::shared_ptr<arrow::Array>> fixed_list_as_large_list(
+ const arrow::FixedSizeListArray& list, const
std::shared_ptr<arrow::Buffer>& validity,
+ arrow::MemoryPool* pool) {
+ arrow::Int64Builder offsets(pool);
+ ARROW_RETURN_NOT_OK(offsets.Reserve(list.length() + 1));
+ for (int64_t i = 0; i <= list.length(); ++i) {
+ ARROW_RETURN_NOT_OK(offsets.Append(list.value_offset(i)));
+ }
+ ARROW_ASSIGN_OR_RAISE(auto values, offsets.Finish());
+ return std::make_shared<arrow::LargeListArray>(
+ arrow::large_list(list.list_type()->value_field()), list.length(),
+ values->data()->buffers[1], list.values(), validity,
list.null_count());
+}
+
+arrow::Result<std::shared_ptr<arrow::Array>> compact_null_containers(
+ std::shared_ptr<arrow::Array> array, arrow::compute::ExecContext*
context) {
+ if (array->null_count() == 0) {
+ return array;
+ }
+ // NULL lists/maps can reference nonempty child ranges. Take drops those
ranges without
+ // making map keys or other non-nullable descendants nullable.
+ arrow::Int64Builder indices(context->memory_pool());
+ ARROW_RETURN_NOT_OK(indices.Reserve(array->length()));
+ for (int64_t i = 0; i < array->length(); ++i) {
+ ARROW_RETURN_NOT_OK(array->IsNull(i) ? indices.AppendNull() :
indices.Append(i));
+ }
+ ARROW_ASSIGN_OR_RAISE(auto positions, indices.Finish());
+ // Fixed-list Take propagates a NULL index into every child, even
non-nullable STRUCT
+ // fields such as a:INT NOT NULL. View it as a variable list first so Take
drops the whole
+ // hidden range instead. Keep 64-bit offsets until the final checked LIST
cast.
+ if (array->type_id() == arrow::Type::FIXED_SIZE_LIST) {
+ ARROW_ASSIGN_OR_RAISE(array, fixed_list_as_large_list(
+ static_cast<const
arrow::FixedSizeListArray&>(*array),
+ positions->null_bitmap(),
context->memory_pool()));
+ }
+ ARROW_ASSIGN_OR_RAISE(
+ auto compacted,
+ arrow::compute::Take(array, positions,
arrow::compute::TakeOptions::NoBoundsCheck(),
+ context));
+ return compacted.make_array();
+}
+
+arrow::Result<std::shared_ptr<arrow::Array>> mask_list_children(
+ std::shared_ptr<arrow::Array> array, const
std::shared_ptr<arrow::DataType>& binding,
+ arrow::compute::ExecContext* context) {
+ ARROW_ASSIGN_OR_RAISE(array, compact_null_containers(std::move(array),
context));
+ auto data = array->data()->Copy();
+ ARROW_ASSIGN_OR_RAISE(auto child,
mask_hidden_children(arrow::MakeArray(data->child_data[0]),
+
binding->field(0)->type(), context));
+ data->child_data[0] = child->data();
+ return arrow::MakeArray(data);
+}
+
+arrow::Result<std::shared_ptr<arrow::Array>> mask_hidden_children(
+ std::shared_ptr<arrow::Array> array, const
std::shared_ptr<arrow::DataType>& binding,
+ arrow::compute::ExecContext* context) {
+ // Only conversions need hidden payloads masked. Descending into an
unchanged VARIANT
+ // carrier would make its non-nullable metadata/value fields NULL under a
SQL NULL parent,
+ // and Arrow's otherwise identical struct view would reject those
synthetic child NULLs.
+ if (array->type()->Equals(binding)) {
+ return array;
+ }
+ const bool converts_list_offsets = binding->id() == arrow::Type::LIST &&
+ (array->type_id() ==
arrow::Type::LARGE_LIST ||
+ array->type_id() ==
arrow::Type::FIXED_SIZE_LIST);
+ // A STRUCT sent for a STRING slot has no corresponding target children.
Leave mismatched
+ // container kinds to Arrow's cast error rather than indexing nonexistent
binding fields.
+ if (array->type_id() != binding->id() && !converts_list_offsets) {
+ return array;
+ }
+ switch (array->type_id()) {
+ case arrow::Type::STRUCT:
+ return mask_struct_children(array, binding, context);
+ case arrow::Type::LIST:
+ case arrow::Type::LARGE_LIST:
+ case arrow::Type::FIXED_SIZE_LIST:
+ case arrow::Type::MAP:
+ return mask_list_children(std::move(array), binding, context);
+ default:
+ return array;
+ }
+}
+
+} // namespace
+
+// Peek one byte between IPC streams without requesting a large buffer from a
blocking load pipe.
+// For example, a second schema can immediately follow the first stream's EOS
marker.
+class ArrowInputStream final : public arrow::io::InputStream {
+public:
+ ArrowInputStream(io::FileReaderSPtr reader, std::shared_ptr<io::IOContext>
io_ctx,
+ arrow::MemoryPool* pool)
+ : _reader(std::move(reader)), _io_ctx(std::move(io_ctx)),
_pool(pool) {}
+
+ arrow::Status Close() override {
+ _reader.reset();
+ return arrow::Status::OK();
+ }
+ bool closed() const override { return _reader == nullptr; }
+ arrow::Result<int64_t> Tell() const override { return _position; }
+
+ Status has_next(bool* has_next) {
+ size_t bytes = 0;
+ RETURN_IF_ERROR(_reader->read_at(_position, Slice(&_next_byte, 1),
&bytes, _io_ctx.get()));
+ _has_next_byte = bytes != 0;
+ *has_next = _has_next_byte;
+ return Status::OK();
+ }
+
+ arrow::Result<int64_t> Read(int64_t nbytes, void* out) override {
+ auto* output = static_cast<char*>(out);
+ int64_t total = 0;
+ if (nbytes > 0 && _has_next_byte) {
+ output[total++] = _next_byte;
+ ++_position;
+ _has_next_byte = false;
+ }
+ while (total < nbytes) {
+ size_t bytes = 0;
+ const auto status = _reader->read_at(_position, Slice(output +
total, nbytes - total),
+ &bytes, _io_ctx.get());
+ if (!status.ok()) {
+ return arrow::Status::IOError(status.to_string());
+ }
+ if (bytes == 0) {
+ break;
+ }
+ total += bytes;
+ _position += bytes;
+ }
+ return total;
+ }
+ arrow::Result<std::shared_ptr<arrow::Buffer>> Read(int64_t nbytes)
override {
+ ARROW_ASSIGN_OR_RAISE(auto buffer,
arrow::AllocateResizableBuffer(nbytes, _pool));
+ ARROW_ASSIGN_OR_RAISE(auto bytes, Read(nbytes,
buffer->mutable_data()));
+ ARROW_RETURN_NOT_OK(buffer->Resize(bytes, false));
+ buffer->ZeroPadding();
+ return buffer;
+ }
+
+private:
+ io::FileReaderSPtr _reader;
+ std::shared_ptr<io::IOContext> _io_ctx;
+ arrow::MemoryPool* _pool;
+ int64_t _position = 0;
+ char _next_byte = 0;
+ bool _has_next_byte = false;
+};
+
+ArrowStreamReader::ArrowStreamReader(std::shared_ptr<io::FileSystemProperties>&
properties,
+ std::unique_ptr<io::FileDescription>&
file,
+ std::shared_ptr<io::IOContext> io_ctx,
RuntimeProfile* profile,
+ std::vector<SlotDescriptor*> file_slots,
+ std::optional<TUniqueId> load_id)
+ : FileReader(properties, file, std::move(io_ctx), profile),
+ _file_slots(std::move(file_slots)),
+ _load_id(std::move(load_id)),
+ _arrow_pool(std::make_unique<ArrowMemoryPool<>>()) {}
+
+ArrowStreamReader::~ArrowStreamReader() {
+ static_cast<void>(close());
+}
+
+void ArrowStreamReader::_init_profile() {
+ if (_profile == nullptr) {
+ return;
+ }
+ file_scan_profile::ensure_hierarchy(_profile);
+ _total_time = ADD_CHILD_TIMER_WITH_LEVEL(_profile, "ArrowStreamReader",
+ file_scan_profile::FILE_READER,
1);
+ _read_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(_profile, "ArrowStreamReadTime",
"ArrowStreamReader", 1);
+ _materialize_time = ADD_CHILD_TIMER_WITH_LEVEL(_profile,
"ArrowStreamMaterializeTime",
+ "ArrowStreamReader", 1);
+}
+
+Status ArrowStreamReader::init(RuntimeState* state) {
+ _state = state;
+ _init_profile();
+ SCOPED_TIMER(_total_time);
+ if (_system_properties->system_type != TFileType::FILE_STREAM ||
!_load_id) {
+ return Status::InvalidArgument("Arrow IPC load requires a stream load
id");
+ }
+ for (const auto* slot : _file_slots) {
+ std::shared_ptr<arrow::DataType> type;
+ RETURN_IF_ERROR(ArrowLoadSchemaConvertor(state->timezone())
+ .convert_to_arrow_type(slot->type(), &type));
+ _arrow_types.push_back(std::move(type));
+ }
+ _eof = false;
+ return Status::OK();
+}
+
+Status ArrowStreamReader::get_schema(std::vector<ColumnDefinition>* schema)
const {
+ schema->clear();
+ for (size_t i = 0; i < _file_slots.size(); ++i) {
+ const auto* slot = _file_slots[i];
+ schema->push_back({.identifier =
Field::create_field<TYPE_STRING>(slot->col_name()),
+ .local_id = cast_set<int32_t>(i),
+ .name = slot->col_name(),
+ .type = slot->type(),
+ .children =
materialized_schema_children(slot->type())});
+ }
+ return Status::OK();
+}
+
+std::unique_ptr<TableColumnMapper> ArrowStreamReader::create_column_mapper(
+ TableColumnMapperOptions options) const {
+ return std::make_unique<MaterializedColumnMapper>(options);
+}
+
+Status ArrowStreamReader::open(std::shared_ptr<FileScanRequest> request) {
+ RETURN_IF_ERROR(FileReader::open(std::move(request)));
+ RETURN_IF_ERROR(FileFactory::create_pipe_reader(*_load_id, &_file_reader,
_state, false));
+ _tracing_file_reader = _io_ctx && _io_ctx->file_reader_stats
+ ? std::make_shared<io::TracingFileReader>(
+ _file_reader,
_io_ctx->file_reader_stats)
+ : _file_reader;
+ _input = std::make_shared<ArrowInputStream>(_tracing_file_reader, _io_ctx,
_arrow_pool.get());
+ return Status::OK();
+}
+
+Status ArrowStreamReader::_open_next_stream() {
+ bool has_next = false;
+ RETURN_IF_ERROR(_input->has_next(&has_next));
+ if (!has_next) {
+ _eof = true;
+ return Status::OK();
+ }
+ auto options = arrow::ipc::IpcReadOptions::Defaults();
+ options.memory_pool = _arrow_pool.get();
+ // Keep IPC decompression on the scanner thread, where the load memory
tracker is attached.
+ options.use_threads = false;
+ RETURN_DORIS_STATUS_IF_RESULT_ERROR(_ipc_reader,
+
arrow::ipc::RecordBatchStreamReader::Open(_input, options));
+ _arrow_positions.assign(_file_slots.size(), -1);
+ for (const auto& [id, position] : _request->local_positions) {
+ const auto& name = _file_slots.at(id.value())->col_name();
+ const auto indices = _ipc_reader->schema()->GetAllFieldIndices(name);
+ if (indices.size() != 1) {
+ return Status::InvalidArgument("Arrow stream requires exactly one
column named '{}'",
+ name);
+ }
+ _arrow_positions[id.value()] = indices[0];
+ }
+ return Status::OK();
+}
+
+Status ArrowStreamReader::_prepare_arrays() {
+ _arrays.assign(_file_slots.size(), nullptr);
+ for (const auto& [id, position] : _request->local_positions) {
+ std::shared_ptr<arrow::Array> array;
+
RETURN_IF_ERROR(normalize_arrow_array(_batch->column(_arrow_positions[id.value()]),
+ _arrow_pool.get(), &array));
+ arrow::compute::ExecContext context(_arrow_pool.get());
+ std::shared_ptr<arrow::DataType> binding;
+ RETURN_DORIS_STATUS_IF_RESULT_ERROR(
+ binding, input_binding_type(array->type(),
_arrow_types[id.value()],
+ _file_slots[id.value()]->type()));
+ // Bind differing numeric widths before concrete SerDe array casts.
Timestamp units
+ // stay on the wire even inside containers, so Doris retains its own
rounding semantics.
+ if (!array->type()->Equals(binding)) {
+ RETURN_DORIS_STATUS_IF_RESULT_ERROR(
+ array, mask_hidden_children(std::move(array), binding,
&context));
+ auto options = arrow::compute::CastOptions::Safe();
+ // The Arrow load protocol permits binary payloads in Doris STRING
columns.
+ options.allow_invalid_utf8 = true;
+ auto converted = arrow::compute::Cast(array, binding, options,
&context);
+ if (!converted.ok()) {
+ return Status::InvalidArgument("Cannot decode Arrow column
'{}': {}",
+
_file_slots[id.value()]->col_name(),
+ converted.status().ToString());
+ }
+ array = converted->make_array();
+ }
+ _arrays[id.value()] = std::move(array);
+ }
+ return Status::OK();
+}
+
+Status ArrowStreamReader::_next_batch() {
+ SCOPED_TIMER(_read_time);
+ while (!_batch || _batch_offset == _batch->num_rows()) {
+ _batch.reset();
+ _arrays.clear();
+ _batch_offset = 0;
+ if (!_ipc_reader) {
+ RETURN_IF_ERROR(_open_next_stream());
+ if (_eof) {
+ return Status::OK();
+ }
+ }
+ RETURN_DORIS_STATUS_IF_ERROR(_ipc_reader->ReadNext(&_batch));
+ if (!_batch) {
+ _ipc_reader.reset();
+ continue;
+ }
+ RETURN_DORIS_STATUS_IF_ERROR(_batch->Validate());
+ if (_batch->num_rows() != 0) {
+ RETURN_IF_ERROR(_prepare_arrays());
+ }
+ }
+ return Status::OK();
+}
+
+Status ArrowStreamReader::get_block(Block* block, size_t* rows, bool* eof) {
+ SCOPED_TIMER(_total_time);
+ *rows = 0;
+ *eof = false;
+ if (_eof || (_io_ctx && _io_ctx->should_stop)) {
+ *eof = true;
+ return Status::OK();
+ }
+ RETURN_IF_ERROR(_next_batch());
+ if (_eof) {
+ *eof = true;
+ return Status::OK();
+ }
+ const int64_t end = _batch_offset +
+ std::min<int64_t>(_state->batch_size(),
_batch->num_rows() - _batch_offset);
+ {
+ SCOPED_TIMER(_materialize_time);
+ auto guard = block->mutate_columns_scoped();
+ auto& columns = guard.mutable_columns();
+ for (const auto& [id, position] : _request->local_positions) {
+ // Preserve the original array and pass a row range: Arrow Slice
creates nonzero
+ // offsets, which nested Doris SerDes deliberately reject.
+ RETURN_IF_ERROR(guard.get_datatype_by_position(position.value())
+ ->get_serde()
+
->read_column_from_arrow(*columns.at(position.value()),
+
_arrays[id.value()].get(),
+ _batch_offset,
end,
+
_state->timezone_obj()));
+ }
+ }
+ *rows = cast_set<size_t>(end - _batch_offset);
+ _batch_offset = end;
+ _record_scan_rows(cast_set<int64_t>(*rows));
+ return apply_materialized_reader_filters(_request.get(), _io_ctx.get(),
block, rows);
+}
+
+Status ArrowStreamReader::close() {
+ _batch.reset();
+ _arrays.clear();
+ _ipc_reader.reset();
+ _input.reset();
+ _request.reset();
+ return FileReader::close();
+}
+
+} // namespace doris::format::arrow_stream
diff --git a/be/src/format_v2/arrow/arrow_stream_reader.h
b/be/src/format_v2/arrow/arrow_stream_reader.h
new file mode 100644
index 00000000000..add10279511
--- /dev/null
+++ b/be/src/format_v2/arrow/arrow_stream_reader.h
@@ -0,0 +1,87 @@
+// 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.
+
+#pragma once
+
+#include <memory>
+#include <optional>
+#include <vector>
+
+#include "format_v2/file_reader.h"
+#include "runtime/runtime_profile.h"
+
+namespace arrow {
+class Array;
+class DataType;
+class MemoryPool;
+class RecordBatch;
+namespace ipc {
+class RecordBatchStreamReader;
+}
+} // namespace arrow
+
+namespace doris {
+class RuntimeState;
+class SlotDescriptor;
+} // namespace doris
+
+namespace doris::format::arrow_stream {
+
+class ArrowInputStream;
+
+// IPC decoding binds the load planner's typed source slots, including
expression inputs.
+// These describe the pipe's input contract; destination casts, defaults, and
row rejection
+// belong to LoadTableReader.
+class ArrowStreamReader final : public FileReader {
+public:
+ ArrowStreamReader(std::shared_ptr<io::FileSystemProperties>& properties,
+ std::unique_ptr<io::FileDescription>& file,
+ std::shared_ptr<io::IOContext> io_ctx, RuntimeProfile*
profile,
+ std::vector<SlotDescriptor*> file_slots,
std::optional<TUniqueId> load_id);
+ ~ArrowStreamReader() override;
+
+ Status init(RuntimeState* state) override;
+ Status get_schema(std::vector<ColumnDefinition>* schema) const override;
+ std::unique_ptr<TableColumnMapper> create_column_mapper(
+ TableColumnMapperOptions options) const override;
+ Status open(std::shared_ptr<FileScanRequest> request) override;
+ Status get_block(Block* block, size_t* rows, bool* eof) override;
+ Status close() override;
+
+private:
+ void _init_profile() override;
+ Status _next_batch();
+ Status _open_next_stream();
+ Status _prepare_arrays();
+
+ RuntimeState* _state = nullptr;
+ const std::vector<SlotDescriptor*> _file_slots;
+ const std::optional<TUniqueId> _load_id;
+ std::unique_ptr<arrow::MemoryPool> _arrow_pool;
+ std::vector<std::shared_ptr<arrow::DataType>> _arrow_types;
+ std::vector<int> _arrow_positions;
+ std::vector<std::shared_ptr<arrow::Array>> _arrays;
+ std::shared_ptr<ArrowInputStream> _input;
+ std::shared_ptr<arrow::ipc::RecordBatchStreamReader> _ipc_reader;
+ std::shared_ptr<arrow::RecordBatch> _batch;
+ int64_t _batch_offset = 0;
+ RuntimeProfile::Counter* _total_time = nullptr;
+ RuntimeProfile::Counter* _read_time = nullptr;
+ RuntimeProfile::Counter* _materialize_time = nullptr;
+};
+
+} // namespace doris::format::arrow_stream
diff --git a/be/src/format_v2/delimited_text/csv_reader.cpp
b/be/src/format_v2/delimited_text/csv_reader.cpp
index d497b790a1b..0d3ebf2cfd5 100644
--- a/be/src/format_v2/delimited_text/csv_reader.cpp
+++ b/be/src/format_v2/delimited_text/csv_reader.cpp
@@ -18,6 +18,7 @@
#include "format_v2/delimited_text/csv_reader.h"
#include <cstring>
+#include <limits>
#include <utility>
#include "core/assert_cast.h"
@@ -120,7 +121,9 @@ Status CsvReader::_init_format_state() {
_empty_field_as_null = text_params.empty_field_as_null;
}
if (_scan_params->file_attributes.hive_open_csv) {
- size_t fields = _source_file_slot_descs.size();
+ // A probe precedes slot planning; a zero OpenCSV field limit would
hide every column.
+ size_t fields =
+ _schema_only ? std::numeric_limits<size_t>::max() :
_source_file_slot_descs.size();
for (int index : _scan_params->column_idxs) {
fields = std::max(fields, static_cast<size_t>(index) + 1);
}
diff --git a/be/src/format_v2/delimited_text/delimited_text_reader.cpp
b/be/src/format_v2/delimited_text/delimited_text_reader.cpp
index 31164d8ba37..049e7265639 100644
--- a/be/src/format_v2/delimited_text/delimited_text_reader.cpp
+++ b/be/src/format_v2/delimited_text/delimited_text_reader.cpp
@@ -43,6 +43,7 @@
#include "runtime/runtime_state.h"
#include "util/decompressor.h"
#include "util/string_util.h"
+#include "util/utf8_check.h"
namespace doris::format {
namespace {
@@ -242,7 +243,7 @@ Status DelimitedTextReader::init(RuntimeState* state) {
// each physical file slot maps to a field ordinal in the text row. The
local id exposed in the
// file schema is therefore the text-field ordinal, not the slot vector
position.
_source_column_idxs.clear();
- if (_scan_params->__isset.column_idxs &&
!_scan_params->column_idxs.empty()) {
+ if (!_schema_only && _scan_params->__isset.column_idxs &&
!_scan_params->column_idxs.empty()) {
if (_scan_params->column_idxs.size() !=
_source_file_slot_descs.size()) {
return Status::InvalidArgument(
"{} v2 reader column_idxs size {} does not match file slot
size {}",
@@ -281,6 +282,49 @@ Status DelimitedTextReader::init(RuntimeState* state) {
return Status::OK();
}
+Status DelimitedTextReader::infer_schema(RuntimeState* state,
+ std::vector<ColumnDefinition>*
schema) {
+ _schema_only = true;
+ RETURN_IF_ERROR(init(state));
+ if (_file_description->range_start_offset != 0) {
+ return Status::InvalidArgument("Schema inference requires a zero start
offset");
+ }
+ if (_scan_params->file_type == TFileType::FILE_BROKER) {
+ return Status::NotSupported("Delimited schema inference does not
support broker load");
+ }
+ RETURN_IF_ERROR(_open_file());
+ RETURN_IF_ERROR(_create_decompressor());
+ RETURN_IF_ERROR(_create_line_reader());
+ // A header is the schema input, not a data record to skip. For example,
+ // csv_with_names must return "id,name" rather than the first row's values.
+ _skip_lines = 0;
+ Slice line;
+ bool eof = false;
+ RETURN_IF_ERROR(_read_next_line(&line, &eof));
+ const auto header = to_lower(_scan_params->file_attributes.header_type);
+ const bool has_names =
+ header == BeConsts::CSV_WITH_NAMES || header ==
BeConsts::CSV_WITH_NAMES_AND_TYPES;
+ if (eof || line.empty()) {
+ // Empty compressed files reach this same check after decompression.
Preserve the
+ // schema RPC's existing diagnostics for callers such as the
empty-LZOP TVF case.
+ return Status::InternalError("The first line is empty, can not parse
column {}",
+ has_names ? "names" : "numbers");
+ }
+ // TEXT scans allow raw bytes, but schema RPCs must retain their UTF-8
validation contract
+ // (for example, a header containing 0xFF must fail unless validation is
explicitly disabled).
+ if (!validate_utf8(*_scan_params, line.data, line.size)) {
+ return Status::InternalError("Only support csv data in utf8 codec");
+ }
+ _split_line(line);
+ schema->clear();
+ for (size_t i = 0; i < _split_values.size(); ++i) {
+ const auto name = has_names ? _split_values[i].to_string() :
fmt::format("c{}", i + 1);
+ schema->push_back(synthetic_file_child(
+ name, make_nullable(std::make_shared<DataTypeString>()),
cast_set<int32_t>(i)));
+ }
+ return Status::OK();
+}
+
Status DelimitedTextReader::get_schema(std::vector<ColumnDefinition>*
file_schema) const {
SCOPED_TIMER(_text_profile.total_time);
if (file_schema == nullptr) {
@@ -556,8 +600,8 @@ Status DelimitedTextReader::_open_file() {
// Stream load/http_stream data lives in NewLoadStreamMgr rather than
a filesystem. The
// generic FileFactory path only supports real file systems, so
FILE_STREAM must use the
// same pipe-reader lookup as the old CSV reader.
- RETURN_IF_ERROR(FileFactory::create_pipe_reader(*_stream_load_id,
&_file_reader,
- _runtime_state,
/*need_schema=*/false));
+ RETURN_IF_ERROR(FileFactory::create_pipe_reader(
+ *_stream_load_id, &_file_reader, _runtime_state,
/*need_schema=*/_schema_only));
} else {
auto reader_options =
FileFactory::get_reader_options(_runtime_state->query_options(),
*_file_description);
diff --git a/be/src/format_v2/delimited_text/delimited_text_reader.h
b/be/src/format_v2/delimited_text/delimited_text_reader.h
index 472f21f59c9..645ba26eecd 100644
--- a/be/src/format_v2/delimited_text/delimited_text_reader.h
+++ b/be/src/format_v2/delimited_text/delimited_text_reader.h
@@ -48,6 +48,8 @@ public:
~DelimitedTextReader() override;
Status init(RuntimeState* state) override;
+ // Discover the first record without FE slots or skipping a configured
header.
+ Status infer_schema(RuntimeState* state, std::vector<ColumnDefinition>*
schema);
Status get_schema(std::vector<ColumnDefinition>* file_schema) const
override;
std::unique_ptr<TableColumnMapper> create_column_mapper(
TableColumnMapperOptions options) const override;
@@ -135,6 +137,7 @@ protected:
const uint8_t* _remove_bom(const uint8_t* ptr, size_t* size);
void _init_profile() override;
+ bool _schema_only = false;
const TFileScanRangeParams* _scan_params = nullptr;
std::vector<SlotDescriptor*> _source_file_slot_descs;
std::vector<int32_t> _source_column_idxs;
diff --git a/be/src/format_v2/json/json_reader.cpp
b/be/src/format_v2/json/json_reader.cpp
index d384067171f..d12d2a4bd1a 100644
--- a/be/src/format_v2/json/json_reader.cpp
+++ b/be/src/format_v2/json/json_reader.cpp
@@ -18,6 +18,7 @@
#include "format_v2/json/json_reader.h"
#include <rapidjson/document.h>
+#include <rapidjson/error/en.h>
#include <algorithm>
#include <cstring>
@@ -249,6 +250,83 @@ Status JsonReader::init(RuntimeState* state) {
return Status::OK();
}
+Status JsonReader::infer_schema(RuntimeState* state,
std::vector<ColumnDefinition>* schema) {
+ RETURN_IF_ERROR(init(state));
+ _reader_range = _json_range();
+ RETURN_IF_ERROR(_open_file_reader(/*need_schema=*/true));
+ RETURN_IF_ERROR(_create_decompressor());
+ if (_read_json_by_line) {
+ RETURN_IF_ERROR(_create_line_reader());
+ }
+ RETURN_IF_ERROR(_parse_jsonpath_and_json_root());
+ if (_reader_range.size <= 0 && _scan_params->file_type !=
TFileType::FILE_STREAM) {
+ _reader_range.size = cast_set<int64_t>(_physical_file_reader->size())
- _current_offset;
+ }
+ size_t size = 0;
+ bool eof = false;
+ RETURN_IF_ERROR(_read_one_document(&size, &eof));
+ if (eof || size == 0) {
+ return Status::EndOfFile("Empty file");
+ }
+ // Schema probing retains the RPC's DOM semantics (including json_root
before outer-array
+ // selection); normal scans continue to use the streaming v2 materializer.
+ rapidjson::Document document;
+ const bool parse_error = _num_as_string
+ ?
document.Parse<rapidjson::kParseNumbersAsStringsFlag>(
+ _document_view.data(),
size)
+ .HasParseError()
+ : document.Parse(_document_view.data(),
size).HasParseError();
+ if (parse_error) {
+ return Status::DataQualityError(
+ "Parse json data for JsonDoc failed. code: {}, error info: {}",
+ document.GetParseError(),
rapidjson::GetParseError_En(document.GetParseError()));
+ }
+ rapidjson::Value* value = &document;
+ if (!_parsed_json_root.empty()) {
+ value =
JsonFunctions::get_json_object_from_parsed_json(_parsed_json_root, &document,
+
document.GetAllocator());
+ if (value == nullptr) {
+ return Status::DataQualityError("JSON Root not found");
+ }
+ }
+ if (value->IsArray() && !_strip_outer_array) {
+ return Status::DataQualityError(
+ "JSON data is array-object, `strip_outer_array` must be
TRUE.");
+ }
+ if (!value->IsArray() && _strip_outer_array) {
+ return Status::DataQualityError(
+ "JSON data is not an array-object, `strip_outer_array` must be
FALSE.");
+ }
+ if (value->IsArray()) {
+ if (value->Empty()) {
+ return Status::InternalError("Empty first json line");
+ }
+ value = &(*value)[0];
+ }
+ if (!value->IsObject()) {
+ return Status::DataQualityError("JSON data is not an object");
+ }
+ schema->clear();
+ const auto append = [&](const std::string& name) {
+ schema->push_back(synthetic_file_child(name,
+
make_nullable(std::make_shared<DataTypeString>()),
+
cast_set<int32_t>(schema->size())));
+ };
+ if (!_parsed_jsonpaths.empty()) {
+ for (const auto& path : _parsed_jsonpaths) {
+ if (path.empty()) {
+ return Status::InvalidArgument("Invalid jsonpaths");
+ }
+ append(path.back().key);
+ }
+ } else {
+ for (const auto& member : value->GetObject()) {
+ append(std::string(member.name.GetString(),
member.name.GetStringLength()));
+ }
+ }
+ return Status::OK();
+}
+
Status JsonReader::get_schema(std::vector<ColumnDefinition>* file_schema)
const {
SCOPED_TIMER(_total_time);
if (file_schema == nullptr) {
@@ -444,7 +522,7 @@ TFileRangeDesc JsonReader::_json_range() const {
return range;
}
-Status JsonReader::_open_file_reader() {
+Status JsonReader::_open_file_reader(bool need_schema) {
_current_offset = _reader_range.start_offset;
if (_current_offset != 0) {
// Include the whole delimiter when the split starts inside it, so
skipping the first
@@ -456,7 +534,7 @@ Status JsonReader::_open_file_reader() {
return Status::InvalidArgument("JSON v2 stream reader requires
load id");
}
RETURN_IF_ERROR(FileFactory::create_pipe_reader(*_stream_load_id,
&_physical_file_reader,
- _runtime_state,
/*need_schema=*/false));
+ _runtime_state,
need_schema));
} else {
_file_description->mtime =
_reader_range.__isset.modification_time ?
_reader_range.modification_time : 0;
@@ -683,10 +761,8 @@ Status JsonReader::_extract_json_value(size_t size, bool*
eof, bool* is_empty_ro
return Status::DataQualityError(
"JSON data is not an array-object, `strip_outer_array` must be
FALSE.");
}
- if (!_parsed_jsonpaths.empty() && _strip_outer_array &&
- _json_value.count_elements().value() == 0) {
- *is_empty_row = true;
- }
+ // A rooted ondemand array has a single cursor. Pre-counting
{"rows":[...]} here
+ // consumes it before jsonpaths materialization; the append path detects
empty arrays.
return Status::OK();
}
diff --git a/be/src/format_v2/json/json_reader.h
b/be/src/format_v2/json/json_reader.h
index c823536d52b..ba2a0d99e2f 100644
--- a/be/src/format_v2/json/json_reader.h
+++ b/be/src/format_v2/json/json_reader.h
@@ -97,6 +97,7 @@ public:
// Initializes scan attributes and builds the synthetic schema from FE
slots.
Status init(RuntimeState* state) override;
+ Status infer_schema(RuntimeState* state, std::vector<ColumnDefinition>*
schema);
Status get_schema(std::vector<ColumnDefinition>* file_schema) const
override;
std::unique_ptr<TableColumnMapper> create_column_mapper(
TableColumnMapperOptions options) const override;
@@ -129,7 +130,7 @@ private:
std::vector<RequestedColumn>* columns)
const;
// Reconciles TableReader's split/range descriptor with FileReader's
concrete file description.
TFileRangeDesc _json_range() const;
- Status _open_file_reader();
+ Status _open_file_reader(bool need_schema = false);
Status _create_decompressor();
Status _create_line_reader();
Status _parse_jsonpath_and_json_root();
diff --git a/be/src/format_v2/schema_reader.cpp
b/be/src/format_v2/schema_reader.cpp
index 6bc0233fbde..6d4910ce100 100644
--- a/be/src/format_v2/schema_reader.cpp
+++ b/be/src/format_v2/schema_reader.cpp
@@ -17,6 +17,9 @@
#include "format_v2/schema_reader.h"
+#include "format_v2/delimited_text/csv_reader.h"
+#include "format_v2/delimited_text/text_reader.h"
+#include "format_v2/json/json_reader.h"
#include "format_v2/orc/orc_reader.h"
#include "format_v2/parquet/parquet_reader.h"
#include "io/file_factory.h"
@@ -27,6 +30,12 @@ namespace doris::format {
Status fetch_file_schema(const TFileScanRangeParams& params, const
TFileRangeDesc& range,
const std::shared_ptr<io::IOContext>& io_ctx,
RuntimeProfile* profile,
std::vector<ColumnDefinition>* schema) {
+ RuntimeProfile fallback_profile("FetchFileSchema");
+ // Native schema callers can omit a profile, but text line readers always
register counters.
+ // Keep a profile alive until the reader and any asynchronous prefetch
work are destroyed.
+ if (profile == nullptr) {
+ profile = &fallback_profile;
+ }
auto properties = std::make_shared<io::FileSystemProperties>();
// Schema RPCs carry the filesystem kind on the range, not necessarily in
scan params.
properties->system_type = range.__isset.file_type ? range.file_type :
params.file_type;
@@ -37,6 +46,8 @@ Status fetch_file_schema(const TFileScanRangeParams& params,
const TFileRangeDes
file->path = range.path;
file->file_size = range.__isset.file_size ? range.file_size : -1;
file->mtime = range.__isset.modification_time ? range.modification_time :
0;
+ file->range_start_offset = range.start_offset;
+ file->range_size = range.__isset.size ? range.size : -1;
file->fs_name = range.fs_name;
if (range.__isset.file_cache_admission) {
file->file_cache_admission = range.file_cache_admission;
@@ -46,6 +57,45 @@ Status fetch_file_schema(const TFileScanRangeParams& params,
const TFileRangeDes
const bool varbinary =
params.__isset.enable_mapping_varbinary &&
params.enable_mapping_varbinary;
RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ // Text and JSON need scan attributes but no planned columns: their schema
is discovered
+ // before FE can construct slots. Range filesystem metadata also applies
to pipe detection.
+ auto scan_params = params;
+ scan_params.__set_file_type(properties->system_type);
+ const std::vector<SlotDescriptor*> slots;
+ const auto compression =
+ range.__isset.compress_type ? range.compress_type :
TFileCompressType::UNKNOWN;
+ const auto load_id =
+ range.__isset.load_id ? std::optional<TUniqueId>(range.load_id) :
std::nullopt;
+ if (params.format_type == TFileFormatType::FORMAT_JSON) {
+ json::JsonReader json_reader(properties, file, io_ctx, profile,
&scan_params, range, slots,
+ compression, load_id);
+ RETURN_IF_ERROR(json_reader.infer_schema(&state, schema));
+ return json_reader.close();
+ }
+ std::unique_ptr<DelimitedTextReader> text_reader;
+ switch (params.format_type) {
+ case TFileFormatType::FORMAT_TEXT:
+ text_reader = std::make_unique<text::TextReader>(properties, file,
io_ctx, profile,
+ &scan_params, slots,
compression, load_id);
+ break;
+ case TFileFormatType::FORMAT_CSV_PLAIN:
+ case TFileFormatType::FORMAT_CSV_GZ:
+ case TFileFormatType::FORMAT_CSV_BZ2:
+ case TFileFormatType::FORMAT_CSV_LZ4FRAME:
+ case TFileFormatType::FORMAT_CSV_LZ4BLOCK:
+ case TFileFormatType::FORMAT_CSV_SNAPPYBLOCK:
+ case TFileFormatType::FORMAT_CSV_LZOP:
+ case TFileFormatType::FORMAT_CSV_DEFLATE:
+ text_reader = std::make_unique<csv::CsvReader>(properties, file,
io_ctx, profile,
+ &scan_params, slots,
compression, load_id);
+ break;
+ default:
+ break;
+ }
+ if (text_reader != nullptr) {
+ RETURN_IF_ERROR(text_reader->infer_schema(&state, schema));
+ return text_reader->close();
+ }
std::unique_ptr<FileReader> reader;
// TVF targets must retain file precision before column mapping; a
microsecond schema
// irreversibly rounds nanos even when the subsequent row scan uses the V2
reader.
@@ -56,7 +106,8 @@ Status fetch_file_schema(const TFileScanRangeParams& params,
const TFileRangeDes
reader = std::make_unique<orc::OrcReader>(properties, file, io_ctx,
profile, std::nullopt,
timestamp_tz, varbinary);
} else {
- return Status::NotSupported("V2 schema inference requires Parquet or
ORC");
+ return Status::NotSupported("Unsupported format {} for V2 schema
inference",
+ params.format_type);
}
RETURN_IF_ERROR(reader->init(&state));
RETURN_IF_ERROR(reader->get_schema(schema));
diff --git a/be/src/format_v2/table/load_reader.cpp
b/be/src/format_v2/table/load_reader.cpp
index f24bbfb2f1f..789f98dce94 100644
--- a/be/src/format_v2/table/load_reader.cpp
+++ b/be/src/format_v2/table/load_reader.cpp
@@ -24,6 +24,7 @@
#include "core/data_type_serde/data_type_serde.h"
#include "exprs/vexpr.h"
#include "exprs/vexpr_context.h"
+#include "format_v2/arrow/arrow_stream_reader.h"
#include "format_v2/wal/wal_reader.h"
#include "runtime/descriptors.h"
#include "runtime/runtime_state.h"
@@ -352,6 +353,12 @@ TableColumnMappingMode LoadTableReader::mapping_mode()
const {
}
Status LoadTableReader::create_file_reader(std::unique_ptr<FileReader>*
reader) {
+ if (_format == FileFormat::ARROW) {
+ *reader = std::make_unique<arrow_stream::ArrowStreamReader>(
+ _system_properties, _current_task->data_file, _io_ctx,
_scanner_profile,
+ _source_file_slots, _current_range_load_id);
+ return Status::OK();
+ }
if (_format == FileFormat::WAL) {
*reader = std::make_unique<wal::WalReader>(_system_properties,
_current_task->data_file,
_io_ctx, _scanner_profile,
_projected_columns);
diff --git a/be/src/service/internal_service.cpp
b/be/src/service/internal_service.cpp
index d64349a6d29..b4b4e76537c 100644
--- a/be/src/service/internal_service.cpp
+++ b/be/src/service/internal_service.cpp
@@ -70,13 +70,10 @@
#include "exec/sink/writer/vmysql_result_writer.h"
#include "exprs/function/dictionary_factory.h"
#include "format/arrow/arrow_row_batch.h"
-#include "format/csv/csv_reader.h"
-#include "format/generic_reader.h"
#include "format/jni/jni_reader.h"
-#include "format/json/new_json_reader.h"
-#include "format/text/text_reader.h"
#include "format_v2/column_data.h"
#include "format_v2/schema_reader.h"
+#include "io/file_factory.h"
#include "io/fs/local_file_system.h"
#include "io/fs/stream_load_pipe.h"
#include "io/io_common.h"
@@ -880,85 +877,19 @@ void
PInternalService::fetch_table_schema(google::protobuf::RpcController* contr
// might asynchronouslly access the profile
std::unique_ptr<RuntimeProfile> profile =
std::make_unique<RuntimeProfile>("FetchTableSchema");
- std::unique_ptr<GenericReader> reader(nullptr);
auto io_ctx = std::make_shared<io::IOContext>();
auto file_cache_statis = std::make_shared<io::FileCacheStatistics>();
auto file_reader_stats = std::make_shared<io::FileReaderStats>();
io_ctx->file_cache_stats = file_cache_statis.get();
io_ctx->file_reader_stats = file_reader_stats.get();
- constexpr size_t fetch_schema_batch_size = 4064;
- // file_slots is no use, but the lifetime should be longer than reader
- std::vector<SlotDescriptor*> file_slots;
- switch (params.format_type) {
- case TFileFormatType::FORMAT_CSV_PLAIN:
- case TFileFormatType::FORMAT_CSV_GZ:
- case TFileFormatType::FORMAT_CSV_BZ2:
- case TFileFormatType::FORMAT_CSV_LZ4FRAME:
- case TFileFormatType::FORMAT_CSV_LZ4BLOCK:
- case TFileFormatType::FORMAT_CSV_SNAPPYBLOCK:
- case TFileFormatType::FORMAT_CSV_LZOP:
- case TFileFormatType::FORMAT_CSV_DEFLATE: {
- reader = CsvReader::create_unique(nullptr, profile.get(), nullptr,
params, range,
- file_slots,
fetch_schema_batch_size, io_ctx.get(),
- io_ctx);
- break;
- }
- case TFileFormatType::FORMAT_TEXT: {
- reader = TextReader::create_unique(nullptr, profile.get(),
nullptr, params, range,
- file_slots,
fetch_schema_batch_size, io_ctx.get());
- break;
- }
- case TFileFormatType::FORMAT_PARQUET:
- case TFileFormatType::FORMAT_ORC: {
- std::vector<format::ColumnDefinition> schema;
- st = format::fetch_file_schema(params, range, io_ctx,
profile.get(), &schema);
- if (st.ok()) {
- result->set_column_nums(schema.size());
- for (const auto& column : schema) {
- result->add_column_names(column.name);
- column.type->to_protobuf(result->add_column_types());
- }
+ std::vector<format::ColumnDefinition> schema;
+ st = format::fetch_file_schema(params, range, io_ctx, profile.get(),
&schema);
+ if (st.ok()) {
+ result->set_column_nums(schema.size());
+ for (const auto& column : schema) {
+ result->add_column_names(column.name);
+ column.type->to_protobuf(result->add_column_types());
}
- st.to_protobuf(result->mutable_status());
- return;
- }
- case TFileFormatType::FORMAT_JSON: {
- reader = NewJsonReader::create_unique(profile.get(), params,
range, file_slots,
- fetch_schema_batch_size,
io_ctx.get(), io_ctx);
- break;
- }
- default:
- st = Status::InternalError("Not supported file format in fetch
table schema: {}",
- params.format_type);
- st.to_protobuf(result->mutable_status());
- return;
- }
- if (!st.ok()) {
- LOG(WARNING) << "failed to create reader, errmsg=" << st;
- st.to_protobuf(result->mutable_status());
- return;
- }
- st = reader->init_schema_reader();
- if (!st.ok()) {
- LOG(WARNING) << "failed to init reader, errmsg=" << st;
- st.to_protobuf(result->mutable_status());
- return;
- }
- std::vector<std::string> col_names;
- std::vector<DataTypePtr> col_types;
- st = reader->get_parsed_schema(&col_names, &col_types);
- if (!st.ok()) {
- LOG(WARNING) << "fetch table schema failed, errmsg=" << st;
- st.to_protobuf(result->mutable_status());
- return;
- }
- result->set_column_nums(col_names.size());
- for (size_t idx = 0; idx < col_names.size(); ++idx) {
- result->add_column_names(col_names[idx]);
- }
- for (size_t idx = 0; idx < col_types.size(); ++idx) {
- PTypeDesc* type_desc = result->add_column_types();
- col_types[idx]->to_protobuf(type_desc);
}
st.to_protobuf(result->mutable_status());
});
diff --git a/be/test/exec/scan/file_load_scanner_v2_test.cpp
b/be/test/exec/scan/file_load_scanner_v2_test.cpp
index 2cce04108b9..22f3d2b9249 100644
--- a/be/test/exec/scan/file_load_scanner_v2_test.cpp
+++ b/be/test/exec/scan/file_load_scanner_v2_test.cpp
@@ -36,7 +36,7 @@ TEST(FileLoadScannerV2Test,
LoadSelectionDoesNotDependOnQueryOption) {
for (auto format : {TFileFormatType::FORMAT_CSV_PLAIN,
TFileFormatType::FORMAT_JSON,
TFileFormatType::FORMAT_PARQUET,
TFileFormatType::FORMAT_ORC,
TFileFormatType::FORMAT_PROTO,
TFileFormatType::FORMAT_WAL,
- TFileFormatType::FORMAT_TEXT}) {
+ TFileFormatType::FORMAT_TEXT,
TFileFormatType::FORMAT_ARROW}) {
params.__set_format_type(format);
options.__set_enable_file_scanner_v2(true);
EXPECT_TRUE(FileScanLocalState::should_use_file_scanner_v2(options,
true, params));
@@ -44,7 +44,7 @@ TEST(FileLoadScannerV2Test,
LoadSelectionDoesNotDependOnQueryOption) {
EXPECT_TRUE(FileScanLocalState::should_use_file_scanner_v2(options,
true, params));
}
options.__set_enable_file_scanner_v2(true);
- for (auto format : {TFileFormatType::FORMAT_AVRO,
TFileFormatType::FORMAT_ARROW}) {
+ for (auto format : {TFileFormatType::FORMAT_AVRO}) {
params.__set_format_type(format);
EXPECT_FALSE(FileScanLocalState::should_use_file_scanner_v2(options,
true, params));
}
diff --git a/be/test/exec/scan/file_scanner_v2_test.cpp
b/be/test/exec/scan/file_scanner_v2_test.cpp
index 3c13bb1a834..fd3d19a7337 100644
--- a/be/test/exec/scan/file_scanner_v2_test.cpp
+++ b/be/test/exec/scan/file_scanner_v2_test.cpp
@@ -463,6 +463,20 @@ TEST(FileScannerV2Test,
ConditionCacheDigestIncludesRuntimeFilterPayload) {
0);
}
+TEST(FileScannerV2Test, ArrowStreamLoadsSelectV2) {
+ TQueryOptions options;
+ TFileScanRangeParams params;
+ params.__set_format_type(TFileFormatType::FORMAT_ARROW);
+ params.__set_file_type(TFileType::FILE_STREAM);
+ params.__set_src_tuple_id(0);
+ TFileRangeDesc range;
+ range.__set_format_type(TFileFormatType::FORMAT_ARROW);
+ range.__set_file_type(TFileType::FILE_STREAM);
+ range.__set_load_id(TUniqueId());
+ EXPECT_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(options,
true, params));
+ EXPECT_TRUE(FileScannerV2::is_supported(params, range));
+}
+
TEST(FileScannerV2Test,
FileScanLocalStateSelectsV2ForSupportedQueriesAndLoads) {
TQueryOptions query_options;
TFileScanRangeParams params;
diff --git a/be/test/format_v2/json/json_reader_test.cpp
b/be/test/format_v2/json/json_reader_test.cpp
index bbd8db29444..5fb94cbcfd3 100644
--- a/be/test/format_v2/json/json_reader_test.cpp
+++ b/be/test/format_v2/json/json_reader_test.cpp
@@ -575,6 +575,20 @@ TEST(JsonReaderTest,
ReadsJsonPathsBySourceSlotAndReturnsRequestedBlockOrder) {
EXPECT_EQ(nullable_int_at(*result.block.get_by_position(1).column, 1), 8);
}
+TEST(JsonReaderTest, ReadsJsonPathsFromAnArraySelectedByJsonRoot) {
+ ObjectPool pool;
+ auto slots = build_slots(&pool);
+ auto result = read_once(
+ "rooted_outer_array_jsonpaths.json",
+
R"({"rows":[{"payload":{"id":12,"user":"kate"}},{"payload":{"id":13,"user":"leo"}}]})",
+ json_scan_params(false, true,
R"(["$.payload.id","$.payload.user"])", "$.rows"), slots,
+ {0, 1});
+ ASSERT_TRUE(result.status.ok()) << result.status.to_string();
+ ASSERT_EQ(result.rows, 2);
+ EXPECT_EQ(nullable_int_at(*result.block.get_by_position(0).column, 0), 12);
+ EXPECT_EQ(nullable_string_at(*result.block.get_by_position(1).column, 1),
"leo");
+}
+
TEST(JsonReaderTest, ReadsJsonPathsFromSingleDocumentOuterArray) {
ObjectPool pool;
auto slots = build_slots(&pool);
diff --git a/be/test/format_v2/schema_reader_test.cpp
b/be/test/format_v2/schema_reader_test.cpp
new file mode 100644
index 00000000000..c485a2ceccc
--- /dev/null
+++ b/be/test/format_v2/schema_reader_test.cpp
@@ -0,0 +1,217 @@
+// 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_v2/schema_reader.h"
+
+#include <gtest/gtest.h>
+
+#include <fstream>
+
+#include "format_v2/column_data.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "io/io_common.h"
+#include "testutil/scoped_temp_dir.h"
+
+namespace doris::format {
+
+class SchemaInferenceV2Test : public testing::Test {
+protected:
+ SchemaInferenceV2Test() {
+ params.__set_file_type(TFileType::FILE_LOCAL);
+ params.__set_format_type(TFileFormatType::FORMAT_CSV_PLAIN);
+ TFileTextScanRangeParams text;
+ text.__set_column_separator(",");
+ text.__set_line_delimiter("\n");
+ attributes.__set_text_params(text);
+ }
+ Status infer(const std::string& data, int64_t offset = 0, bool
unknown_size = false) {
+ const auto path = (temp.path() / "input").string();
+ {
+ std::ofstream out(path);
+ out << data;
+ }
+ params.__set_file_attributes(attributes);
+ TFileRangeDesc range;
+ range.__set_path(path);
+ range.__set_start_offset(offset);
+ range.__set_size(unknown_size ? -1 :
static_cast<int64_t>(data.size()));
+ if (!unknown_size) {
+ range.__set_file_size(data.size());
+ }
+ schema.clear();
+ return fetch_file_schema(params, range,
std::make_shared<io::IOContext>(), nullptr,
+ &schema);
+ }
+ test::ScopedTempDirectory temp {"schema_v2"};
+ TFileScanRangeParams params;
+ TFileAttributes attributes;
+ std::vector<ColumnDefinition> schema;
+};
+
+TEST_F(SchemaInferenceV2Test, CsvHeadersAreNotSkippedAndTypesRemainStrings) {
+ attributes.__set_header_type("CSV_WITH_NAMES_AND_TYPES");
+ attributes.text_params.__set_enclose('"');
+ const auto status =
infer("\xEF\xBB\xBF\"first,name\",second\nint,string\n1,value\n");
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(schema.size(), 2);
+ EXPECT_EQ(schema[0].name, "first,name");
+ EXPECT_EQ(schema[1].name, "second");
+ EXPECT_EQ(schema[0].type->get_name(), "Nullable(String)");
+}
+
+TEST_F(SchemaInferenceV2Test, TextUsesEscapedMulticharacterSeparators) {
+ params.__set_format_type(TFileFormatType::FORMAT_TEXT);
+ attributes.text_params.__set_column_separator("||");
+ attributes.text_params.__set_escape('\\');
+ ASSERT_TRUE(infer("a\\||b||c\n").ok());
+ ASSERT_EQ(schema.size(), 2);
+ EXPECT_EQ(schema[1].name, "c2");
+}
+
+TEST_F(SchemaInferenceV2Test,
HiveOpenCsvDiscoversAllFieldsWithoutPlannedSlots) {
+ attributes.__set_hive_open_csv(true);
+ attributes.text_params.__set_enclose('"');
+ params.__set_column_idxs({2});
+ const auto status = infer("\"a,b\",c,d\n");
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(schema.size(), 3);
+ EXPECT_EQ(schema[2].name, "c3");
+}
+
+TEST_F(SchemaInferenceV2Test, TextSchemaRetainsUtf8Validation) {
+ params.__set_format_type(TFileFormatType::FORMAT_TEXT);
+ EXPECT_FALSE(infer("\xFF,x\n").ok());
+ attributes.__set_enable_text_validate_utf8(false);
+ EXPECT_TRUE(infer("\xFF,x\n").ok());
+}
+
+TEST_F(SchemaInferenceV2Test,
JsonSchemaAcceptsLargeIntegersWithNumbersAsStrings) {
+ params.__set_format_type(TFileFormatType::FORMAT_JSON);
+ attributes.__set_num_as_string(true);
+ const auto status =
infer(R"({"value":170141183460469231731687303715884105727})");
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(schema.size(), 1);
+ EXPECT_EQ(schema[0].name, "value");
+}
+
+TEST_F(SchemaInferenceV2Test, JsonRootAndPathsPreserveColumnNames) {
+ params.__set_format_type(TFileFormatType::FORMAT_JSON);
+ attributes.__set_json_root("$.rows");
+ attributes.__set_strip_outer_array(true);
+ attributes.__set_jsonpaths(R"(["$.id","$.nested.name"])");
+
ASSERT_TRUE(infer(R"({"rows":[{"id":1,"nested":{"name":"value"}}]})").ok());
+ ASSERT_EQ(schema.size(), 2);
+ EXPECT_EQ(schema[0].name, "id");
+ EXPECT_EQ(schema[1].name, "name");
+}
+
+TEST_F(SchemaInferenceV2Test, JsonCanInferAnUnknownSizeFile) {
+ params.__set_format_type(TFileFormatType::FORMAT_JSON);
+ ASSERT_TRUE(infer(R"({"name":"value"})", 0, true).ok());
+ ASSERT_EQ(schema.size(), 1);
+ EXPECT_EQ(schema[0].name, "name");
+}
+
+TEST_F(SchemaInferenceV2Test, JsonRejectsMalformedInputAndInvalidShape) {
+ params.__set_format_type(TFileFormatType::FORMAT_JSON);
+ for (const auto* data : {"", "{", "1", "[]", "[{}]"}) {
+ EXPECT_FALSE(infer(data).ok()) << data;
+ }
+ attributes.__set_strip_outer_array(true);
+ EXPECT_FALSE(infer("[]").ok());
+ EXPECT_FALSE(infer("{}").ok());
+ attributes.__set_strip_outer_array(false);
+ attributes.__set_json_root("$.missing");
+ EXPECT_FALSE(infer(R"({"name":"value"})").ok());
+}
+
+TEST_F(SchemaInferenceV2Test, CsvRejectsEmptyFirstLineAndNonzeroOffset) {
+ EXPECT_FALSE(infer("").ok());
+ const auto status = infer("\n1,value\n");
+ EXPECT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("The first line is empty, can not parse
column numbers"),
+ std::string::npos)
+ << status;
+ EXPECT_FALSE(infer("1,value\n", 1).ok());
+}
+
+TEST_F(SchemaInferenceV2Test, EmptyLzopRetainsTheSchemaRpcError) {
+ // A valid LZOP header (no filename, zero timestamps), Adler-32, and an
end-of-stream block.
+ const uint8_t stream[] = {0x89, 0x4c, 0x5a, 0x4f, 0x00, 0x0d, 0x0a, 0x1a,
0x0a, 0x10, 0x30,
+ 0x20, 0xa0, 0x09, 0x40, 0x01, 0x05, 0x00, 0x00,
0x00, 0x01, 0x00,
+ 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00,
+ 0x00, 0x1d, 0x01, 0x01, 0x51, 0x00, 0x00, 0x00,
0x00};
+ params.__set_format_type(TFileFormatType::FORMAT_CSV_LZOP);
+ for (const auto* header : {"", "csv_with_names",
"csv_with_names_and_types"}) {
+ attributes.__set_header_type(header);
+ const auto status =
+ infer(std::string(reinterpret_cast<const char*>(stream),
sizeof(stream)));
+ EXPECT_FALSE(status.ok());
+ const auto expected = std::string("The first line is empty, can not
parse column ") +
+ (*header == '\0' ? "numbers" : "names");
+ EXPECT_NE(status.to_string().find(expected), std::string::npos) <<
status;
+ }
+}
+
+TEST(SchemaReaderV2Test, InfersTextAndJsonWithoutPlannedSlots) {
+ test::ScopedTempDirectory temp("schema_v2");
+ struct Case {
+ TFileFormatType::type format;
+ std::string data;
+ std::vector<std::string> names;
+ };
+ const std::vector<Case> cases {
+ {.format = TFileFormatType::FORMAT_CSV_PLAIN,
+ .data = "1,value\n",
+ .names = {"c1", "c2"}},
+ {.format = TFileFormatType::FORMAT_TEXT, .data = "1,value\n",
.names = {"c1", "c2"}},
+ {.format = TFileFormatType::FORMAT_JSON,
+ .data = R"({"id":1,"name":"value"})",
+ .names = {"id", "name"}}};
+ for (const auto& c : cases) {
+ const auto path = (temp.path() / "input").string();
+ {
+ std::ofstream out(path);
+ out << c.data;
+ }
+ TFileScanRangeParams params;
+ params.__set_format_type(c.format);
+ params.__set_file_type(TFileType::FILE_LOCAL);
+ TFileAttributes attributes;
+ TFileTextScanRangeParams text;
+ text.__set_column_separator(",");
+ text.__set_line_delimiter("\n");
+ attributes.__set_text_params(text);
+ params.__set_file_attributes(attributes);
+ TFileRangeDesc range;
+ range.__set_path(path);
+ range.__set_start_offset(0);
+ range.__set_size(c.data.size());
+ range.__set_file_size(c.data.size());
+ std::vector<ColumnDefinition> schema;
+ ASSERT_TRUE(fetch_file_schema(params, range,
std::make_shared<io::IOContext>(), nullptr,
+ &schema)
+ .ok());
+ ASSERT_EQ(schema.size(), c.names.size());
+ for (size_t i = 0; i < schema.size(); ++i) {
+ EXPECT_EQ(schema[i].name, c.names[i]);
+ EXPECT_EQ(schema[i].type->get_name(), "Nullable(String)");
+ }
+ }
+}
+
+} // namespace doris::format
diff --git a/be/test/format_v2/table/load_reader_test.cpp
b/be/test/format_v2/table/load_reader_test.cpp
index 21d378e40f3..1620afd9c7c 100644
--- a/be/test/format_v2/table/load_reader_test.cpp
+++ b/be/test/format_v2/table/load_reader_test.cpp
@@ -17,19 +17,42 @@
#include "format_v2/table/load_reader.h"
+#include <arrow/api.h>
+#include <arrow/extension/parquet_variant.h>
+#include <arrow/io/memory.h>
+#include <arrow/ipc/api.h>
+#include <arrow/util/compression.h>
#include <gtest/gtest.h>
#include <unistd.h>
#include <filesystem>
#include <fstream>
+#include <limits>
+#include "core/column/column_array.h"
#include "core/column/column_complex.h"
+#include "core/column/column_decimal.h"
+#include "core/column/column_fixed_length_object.h"
+#include "core/column/column_map.h"
#include "core/column/column_nullable.h"
#include "core/column/column_string.h"
+#include "core/column/column_struct.h"
+#include "core/column/variant_v2/column_variant_v2.h"
+#include "core/data_type/data_type_agg_state.h"
+#include "core/data_type/data_type_array.h"
#include "core/data_type/data_type_bitmap.h"
+#include "core/data_type/data_type_date_or_datetime_v2.h"
+#include "core/data_type/data_type_decimal.h"
+#include "core/data_type/data_type_ipv4.h"
+#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 "format/arrow/arrow_block_convertor.h"
+#include "format/arrow/arrow_row_batch.h"
+#include "format_v2/schema_reader.h"
#include "io/fs/kafka_consumer_pipe.h"
#include "load/stream_load/new_load_stream_mgr.h"
#include "load/stream_load/stream_load_context.h"
@@ -369,12 +392,12 @@ protected:
}
LoadReaderTest::TearDown();
}
- Status prepare(LoadTableReader& reader) {
- RETURN_IF_ERROR(init(reader, FileFormat::JSON));
+ Status prepare(LoadTableReader& reader, FileFormat format =
FileFormat::JSON) {
+ RETURN_IF_ERROR(init(reader, format));
SplitReadOptions split;
- split.current_split_format = FileFormat::JSON;
+ split.current_split_format = format;
split.current_range.__set_file_type(TFileType::FILE_STREAM);
- split.current_range.__set_format_type(TFileFormatType::FORMAT_JSON);
+ split.current_range.__set_format_type(params.format_type);
split.current_range.__set_load_id(_context->id.to_thrift());
split.current_range.__set_size(-1);
return reader.prepare_split(split);
@@ -387,6 +410,925 @@ protected:
bool _owns_stream_manager = false;
};
+class LoadArrowPipeTest : public LoadJsonPipeTest {
+protected:
+ void SetUp() override {
+ LoadJsonPipeTest::SetUp();
+ params.__set_format_type(TFileFormatType::FORMAT_ARROW);
+ }
+ void set_column_type(const DataTypePtr& type) {
+ auto descriptor = type->to_thrift();
+ if (remove_nullable(type)->get_primitive_type() == TYPE_AGG_STATE) {
+ // AGG_STATE's function signature lives on TTypeDesc, outside its
scalar type node.
+ const auto& state_type = assert_cast<const
DataTypeAggState&>(*remove_nullable(type));
+ std::vector<TTypeDesc> arguments;
+ for (const auto& argument : state_type.get_sub_types()) {
+ auto argument_descriptor = argument->to_thrift();
+ argument_descriptor.__set_is_nullable(argument->is_nullable());
+ arguments.push_back(std::move(argument_descriptor));
+ }
+ descriptor.__set_sub_types(arguments);
+ descriptor.__set_function_name(state_type.get_function_name());
+ descriptor.__set_result_is_nullable(false);
+
descriptor.__set_be_exec_version(BeExecVersionManager::get_newest_version());
+ }
+ source_definition.__set_slotType(descriptor);
+ source_slot = std::make_unique<SlotDescriptor>(source_definition);
+ TTupleDescriptor tuple;
+ tuple.__set_id(0);
+ tuple.__set_numNullBytes(1);
+ source = std::make_unique<TupleDescriptor>(tuple);
+ source->add_slot(source_slot.get());
+ tuple.__set_id(1);
+ destination = std::make_unique<TupleDescriptor>(tuple);
+ auto definition = source_definition;
+ definition.__set_id(20);
+ definition.__set_parent(1);
+ definition.__set_colName("value");
+ destination_slot = std::make_unique<SlotDescriptor>(definition);
+ destination->add_slot(destination_slot.get());
+ params.expr_of_dest_slot[20].nodes[0].__set_type(descriptor);
+ }
+ void check_timestamp_rounding(bool nested) {
+ const auto time_type =
make_nullable(std::make_shared<DataTypeDateTimeV2>(3));
+ set_column_type(nested ?
make_nullable(std::make_shared<DataTypeArray>(time_type))
+ : time_type);
+ const auto wire_type = arrow::timestamp(arrow::TimeUnit::MICRO);
+ arrow::TimestampBuilder builder(wire_type,
arrow::default_memory_pool());
+ ASSERT_TRUE(builder.Append(123789).ok());
+ auto timestamps = builder.Finish();
+ ASSERT_TRUE(timestamps.ok());
+ std::shared_ptr<arrow::Array> values = *timestamps;
+ if (nested) {
+ arrow::Int32Builder offsets;
+ ASSERT_TRUE(offsets.AppendValues({0, 1}).ok());
+ auto offset_values = offsets.Finish();
+ ASSERT_TRUE(offset_values.ok());
+ auto lists = arrow::ListArray::FromArrays(**offset_values,
*values);
+ ASSERT_TRUE(lists.ok());
+ values = *lists;
+ }
+ auto schema = arrow::schema({arrow::field("raw_value",
values->type())});
+ append_batch(arrow::RecordBatch::Make(schema, 1, {values}));
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ ASSERT_TRUE(prepare(reader, FileFormat::ARROW).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ const auto status = reader.get_block(&output, &eos);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(output.rows(), 1);
+ const IColumn* value_column = output.get_by_position(0).column.get();
+ if (nested) {
+ const auto& nullable = assert_cast<const
ColumnNullable&>(*value_column);
+ value_column =
+ &assert_cast<const
ColumnArray&>(nullable.get_nested_column()).get_data();
+ }
+ EXPECT_EQ(time_type->to_string(*value_column, 0), "1970-01-01
00:00:00.124");
+ ASSERT_TRUE(reader.close().ok());
+ }
+ void append_stream(const std::vector<std::optional<std::string>>& values,
+ const std::string& name = "raw_value") {
+ arrow::StringBuilder builder;
+ for (const auto& value : values) {
+ ASSERT_TRUE((value ? builder.Append(*value) :
builder.AppendNull()).ok());
+ }
+ auto array = builder.Finish();
+ ASSERT_TRUE(array.ok());
+ auto schema = arrow::schema({arrow::field(name, arrow::utf8())});
+ append_batch(arrow::RecordBatch::Make(schema, values.size(),
{*array}));
+ }
+ void check_hidden_container_values(int container) {
+ const auto int_type = make_nullable(std::make_shared<DataTypeInt32>());
+ set_column_type(make_nullable(
+ container == 2 ? DataTypePtr(std::make_shared<DataTypeMap>(
+ std::make_shared<DataTypeInt32>(),
int_type))
+ :
DataTypePtr(std::make_shared<DataTypeArray>(int_type))));
+ arrow::Int64Builder builder;
+ ASSERT_TRUE(builder.AppendValues({std::numeric_limits<int64_t>::max(),
7}).ok());
+ auto values = builder.Finish();
+ ASSERT_TRUE(values.ok());
+ arrow::BooleanBuilder validity;
+ ASSERT_TRUE(validity.AppendValues(std::vector<bool> {false,
true}).ok());
+ auto valid = validity.Finish();
+ ASSERT_TRUE(valid.ok());
+ const auto bitmap = (*valid)->data()->buffers[1];
+ std::shared_ptr<arrow::Array> input;
+ if (container == 1) {
+ input = std::make_shared<arrow::FixedSizeListArray>(
+ arrow::fixed_size_list(arrow::int64(), 1), 2, *values,
bitmap, 1);
+ } else {
+ arrow::Int32Builder offsets;
+ ASSERT_TRUE(offsets.AppendValues({0, 1, 2}).ok());
+ auto positions = offsets.Finish();
+ ASSERT_TRUE(positions.ok());
+ if (container == 2) {
+ input = std::make_shared<arrow::MapArray>(
+ arrow::map(arrow::int64(), arrow::int64()), 2,
+ (*positions)->data()->buffers[1], *values, *values,
bitmap, 1);
+ } else {
+ auto lists = arrow::ListArray::FromArrays(**positions,
**values,
+
arrow::default_memory_pool(), bitmap, 1);
+ ASSERT_TRUE(lists.ok());
+ input = *lists;
+ }
+ }
+ append_batch(arrow::RecordBatch::Make(
+ arrow::schema({arrow::field("raw_value", input->type())}), 2,
{input}));
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ ASSERT_TRUE(prepare(reader, FileFormat::ARROW).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ const auto status = reader.get_block(&output, &eos);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(output.rows(), 2);
+ EXPECT_TRUE(output.get_by_position(0).column->is_null_at(0));
+ EXPECT_FALSE(output.get_by_position(0).column->is_null_at(1));
+ const auto& column = assert_cast<const
ColumnNullable&>(*output.get_by_position(0).column);
+ if (container == 2) {
+ const auto& map = assert_cast<const
ColumnMap&>(column.get_nested_column());
+ EXPECT_EQ(map.get_keys().get_int(0), 7);
+ } else {
+ const auto& array = assert_cast<const
ColumnArray&>(column.get_nested_column());
+ const auto& elements = assert_cast<const
ColumnNullable&>(array.get_data());
+
EXPECT_EQ(elements.get_nested_column().get_int(array.get_offsets()[0]), 7);
+ }
+ ASSERT_TRUE(reader.close().ok());
+ }
+ void append_batch(const std::shared_ptr<arrow::RecordBatch>& batch, bool
compressed = false) {
+ auto sink = arrow::io::BufferOutputStream::Create();
+ ASSERT_TRUE(sink.ok());
+ auto options = arrow::ipc::IpcWriteOptions::Defaults();
+ if (compressed) {
+ auto codec = arrow::util::Codec::Create(arrow::Compression::ZSTD);
+ ASSERT_TRUE(codec.ok());
+ options.codec = std::move(*codec);
+ }
+ auto writer = arrow::ipc::MakeStreamWriter(*sink, batch->schema(),
options);
+ ASSERT_TRUE(writer.ok());
+ ASSERT_TRUE((*writer)->WriteRecordBatch(*batch).ok());
+ ASSERT_TRUE((*writer)->Close().ok());
+ auto buffer = (*sink)->Finish();
+ ASSERT_TRUE(buffer.ok());
+ // HTTP chunks need not coincide with IPC message boundaries.
+ for (int64_t pos = 0; pos < (*buffer)->size(); pos += 7) {
+ const auto length = std::min<int64_t>(7, (*buffer)->size() - pos);
+ ASSERT_TRUE(_pipe->append_json(reinterpret_cast<const
char*>((*buffer)->data() + pos),
+ length)
+ .ok());
+ }
+ }
+ void make_native_variant_array(bool extension,
std::shared_ptr<arrow::Array>* array) {
+ ASSERT_TRUE(register_arrow_variant_extension().ok());
+ auto type = std::make_shared<DataTypeVariantV2>();
+ auto values = type->create_column();
+ auto serde = type->get_serde();
+ DataTypeSerDe::FormatOptions options;
+ for (std::string json : {"42", "null", "null"}) {
+ Slice slice(json.data(), json.size());
+ ASSERT_TRUE(serde->deserialize_one_cell_from_json(*values, slice,
options).ok());
+ }
+ auto nulls = ColumnUInt8::create();
+ nulls->get_data().assign({0, 0, 1});
+ Block block {{ColumnNullable::create(std::move(values),
std::move(nulls)),
+ make_nullable(type), "raw_value"}};
+ auto wire_type = arrow::extension::variant(
+ arrow::struct_({arrow::field("metadata", arrow::binary(),
false),
+ arrow::field("value", arrow::binary(),
false)}));
+ ArrowFlightArrowBlockConvertor converter(
+ arrow::schema({arrow::field("raw_value", wire_type)}),
cctz::utc_time_zone());
+ std::shared_ptr<arrow::RecordBatch> batch;
+ ASSERT_TRUE(converter.convert_to_arrow(block,
arrow::default_memory_pool(), &batch).ok());
+ *array = extension ? batch->column(0)
+ : static_cast<const
arrow::ExtensionArray&>(*batch->column(0)).storage();
+ }
+ void wrap_native_variant(int container, std::shared_ptr<arrow::Array>*
values) {
+ const auto variant_type =
make_nullable(std::make_shared<DataTypeVariantV2>());
+ arrow::Int64Builder integers;
+ ASSERT_TRUE(integers.AppendValues({7, 8, 9}).ok());
+ auto numbers = integers.Finish();
+ ASSERT_TRUE(numbers.ok());
+ if (container == 3) {
+ set_column_type(make_nullable(std::make_shared<DataTypeStruct>(
+ DataTypes {variant_type,
make_nullable(std::make_shared<DataTypeInt32>())},
+ Strings {"payload", "number"})));
+ auto structs = arrow::StructArray::Make(
+ {*values, *numbers}, std::vector<std::string>
{"WIRE_PAYLOAD", "WIRE_NUMBER"});
+ ASSERT_TRUE(structs.ok());
+ *values = *structs;
+ return;
+ }
+ arrow::Int32Builder offsets;
+ ASSERT_TRUE(offsets.AppendValues({0, 1, 2, 3}).ok());
+ auto positions = offsets.Finish();
+ ASSERT_TRUE(positions.ok());
+ if (container == 2) {
+ set_column_type(make_nullable(std::make_shared<DataTypeMap>(
+ std::make_shared<DataTypeInt32>(), variant_type)));
+ *values = std::make_shared<arrow::MapArray>(
+ arrow::map(arrow::int64(), (*values)->type()), 3,
+ (*positions)->data()->buffers[1], *numbers, *values);
+ } else {
+
set_column_type(make_nullable(std::make_shared<DataTypeArray>(variant_type)));
+ auto lists = arrow::ListArray::FromArrays(**positions, **values);
+ ASSERT_TRUE(lists.ok());
+ *values = *lists;
+ }
+ }
+ void check_native_variant(bool extension, int container = 0) {
+ set_column_type(make_nullable(std::make_shared<DataTypeVariantV2>()));
+ std::shared_ptr<arrow::Array> values;
+ make_native_variant_array(extension, &values);
+ ASSERT_NE(values, nullptr);
+ if (container != 0) {
+ wrap_native_variant(container, &values);
+ }
+ append_batch(arrow::RecordBatch::Make(
+ arrow::schema({arrow::field("raw_value", values->type())}), 3,
{values}));
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ ASSERT_TRUE(prepare(reader, FileFormat::ARROW).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ const auto status = reader.get_block(&output, &eos);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(output.rows(), 3);
+ const IColumn* result_column = output.get_by_position(0).column.get();
+ if (container != 0) {
+ const auto& outer =
+ assert_cast<const
ColumnNullable&>(*result_column).get_nested_column();
+ if (container == 1) {
+ result_column = &assert_cast<const
ColumnArray&>(outer).get_data();
+ } else if (container == 2) {
+ result_column = &assert_cast<const
ColumnMap&>(outer).get_values();
+ } else {
+ result_column = &assert_cast<const
ColumnStruct&>(outer).get_column(0);
+ }
+ }
+ const auto& nullable = assert_cast<const
ColumnNullable&>(*result_column);
+ const auto& result = assert_cast<const
ColumnVariantV2&>(nullable.get_nested_column());
+ EXPECT_EQ(result.get_value_ref(0).get_int(), 42);
+ EXPECT_FALSE(nullable.is_null_at(1));
+ EXPECT_TRUE(result.get_value_ref(1).is_null());
+ EXPECT_TRUE(nullable.is_null_at(2));
+ ASSERT_TRUE(reader.close().ok());
+ }
+ void check_count_state(bool variable_binary, int width = 8) {
+ set_column_type(make_nullable(std::make_shared<DataTypeAggState>(
+ DataTypes {std::make_shared<DataTypeInt32>()}, false, "count",
+ BeExecVersionManager::get_newest_version())));
+ std::unique_ptr<arrow::ArrayBuilder> builder;
+ const auto wire_type = variable_binary ? arrow::binary() :
arrow::fixed_size_binary(width);
+ ASSERT_TRUE(arrow::MakeBuilder(arrow::default_memory_pool(),
wire_type, &builder).ok());
+ const uint64_t counts[] = {0, 7};
+ for (const auto& count : counts) {
+ const std::string_view bytes(reinterpret_cast<const
char*>(&count), width);
+ const auto status =
+ variable_binary
+ ?
static_cast<arrow::BinaryBuilder&>(*builder).Append(bytes)
+ :
static_cast<arrow::FixedSizeBinaryBuilder&>(*builder).Append(bytes);
+ ASSERT_TRUE(status.ok()) << status;
+ }
+ ASSERT_TRUE(builder->AppendNull().ok());
+ auto values = builder->Finish();
+ ASSERT_TRUE(values.ok());
+
append_batch(arrow::RecordBatch::Make(arrow::schema({arrow::field("raw_value",
wire_type)}),
+ 3, {*values}));
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ const auto prepared = prepare(reader, FileFormat::ARROW);
+ ASSERT_TRUE(prepared.ok()) << prepared;
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ const auto status = reader.get_block(&output, &eos);
+ if (width != sizeof(uint64_t)) {
+ EXPECT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("Cannot decode Arrow column"),
std::string::npos)
+ << status;
+ } else {
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(output.rows(), 3);
+ const auto& nullable =
+ assert_cast<const
ColumnNullable&>(*output.get_by_position(0).column);
+ const auto& result =
+ assert_cast<const
ColumnFixedLengthObject&>(nullable.get_nested_column());
+ EXPECT_TRUE(nullable.is_null_at(2));
+ EXPECT_EQ(result.item_size(), sizeof(uint64_t));
+ EXPECT_EQ(result.get_data_at(0),
+ StringRef(reinterpret_cast<const char*>(&counts[0]), 8));
+ EXPECT_EQ(result.get_data_at(1),
+ StringRef(reinterpret_cast<const char*>(&counts[1]), 8));
+ }
+ ASSERT_TRUE(reader.close().ok());
+ }
+ void check_null_fixed_list_struct(bool narrow, bool all_null = false, int
width = 1) {
+ const auto structure = make_nullable(std::make_shared<DataTypeStruct>(
+ DataTypes {make_nullable(std::make_shared<DataTypeInt32>())},
Strings {"a"}));
+
set_column_type(make_nullable(std::make_shared<DataTypeArray>(structure)));
+ const auto wire_type = narrow ? arrow::int64() : arrow::int32();
+ std::unique_ptr<arrow::ArrayBuilder> builder;
+ ASSERT_TRUE(arrow::MakeBuilder(arrow::default_memory_pool(),
wire_type, &builder).ok());
+ for (int i = 0; i < 2 * width; ++i) {
+ const auto status = narrow ?
static_cast<arrow::Int64Builder&>(*builder).Append(
+ i < width ?
std::numeric_limits<int64_t>::max()
+ : 7 + i - width)
+ :
static_cast<arrow::Int32Builder&>(*builder).Append(
+ i < width ? 1 + i : 7 + i -
width);
+ ASSERT_TRUE(status.ok()) << status;
+ }
+ auto values = builder->Finish();
+ ASSERT_TRUE(values.ok());
+ auto structs = arrow::StructArray::Make(
+ {*values}, arrow::FieldVector {arrow::field("a", wire_type,
false)});
+ ASSERT_TRUE(structs.ok());
+ arrow::BooleanBuilder validity;
+ ASSERT_TRUE(validity.AppendValues(std::vector<bool> {false,
!all_null}).ok());
+ auto valid = validity.Finish();
+ ASSERT_TRUE(valid.ok());
+ auto lists = std::make_shared<arrow::FixedSizeListArray>(
+ arrow::fixed_size_list((*structs)->type(), width), 2, *structs,
+ (*valid)->data()->buffers[1], all_null ? 2 : 1);
+ append_batch(arrow::RecordBatch::Make(
+ arrow::schema({arrow::field("raw_value", lists->type())}), 2,
{lists}));
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ ASSERT_TRUE(prepare(reader, FileFormat::ARROW).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ const auto status = reader.get_block(&output, &eos);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(output.rows(), 2);
+ const auto& outer = assert_cast<const
ColumnNullable&>(*output.get_by_position(0).column);
+ EXPECT_TRUE(outer.is_null_at(0));
+ EXPECT_EQ(outer.is_null_at(1), all_null);
+ const auto& array = assert_cast<const
ColumnArray&>(outer.get_nested_column());
+ EXPECT_EQ(array.get_offsets()[0], 0);
+ EXPECT_EQ(array.get_offsets()[1], all_null ? 0 : width);
+ const auto& elements = assert_cast<const
ColumnNullable&>(array.get_data());
+ const auto& fields = assert_cast<const
ColumnStruct&>(elements.get_nested_column());
+ if (!all_null) {
+ const auto& numbers = assert_cast<const
ColumnNullable&>(fields.get_column(0));
+ for (int i = 0; i < width; ++i) {
+ EXPECT_FALSE(numbers.is_null_at(i));
+ EXPECT_EQ(numbers.get_nested_column().get_int(i), 7 + i);
+ }
+ }
+ ASSERT_TRUE(reader.close().ok());
+ }
+ void check_null_fixed_list_variant(bool extension) {
+ set_column_type(make_nullable(std::make_shared<DataTypeArray>(
+ make_nullable(std::make_shared<DataTypeVariantV2>()))));
+ std::shared_ptr<arrow::Array> values;
+ make_native_variant_array(extension, &values);
+ ASSERT_NE(values, nullptr);
+ arrow::BooleanBuilder validity;
+ ASSERT_TRUE(validity.AppendValues(std::vector<bool> {true, false,
true}).ok());
+ auto valid = validity.Finish();
+ ASSERT_TRUE(valid.ok());
+ auto lists = std::make_shared<arrow::FixedSizeListArray>(
+ arrow::fixed_size_list(values->type(), 1), 3, values,
(*valid)->data()->buffers[1],
+ 1);
+ append_batch(arrow::RecordBatch::Make(
+ arrow::schema({arrow::field("raw_value", lists->type())}), 3,
{lists}));
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ ASSERT_TRUE(prepare(reader, FileFormat::ARROW).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ const auto status = reader.get_block(&output, &eos);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(output.rows(), 3);
+ const auto& outer = assert_cast<const
ColumnNullable&>(*output.get_by_position(0).column);
+ EXPECT_TRUE(outer.is_null_at(1));
+ const auto& array = assert_cast<const
ColumnArray&>(outer.get_nested_column());
+ EXPECT_EQ(array.get_offsets()[0], 1);
+ EXPECT_EQ(array.get_offsets()[1], 1);
+ EXPECT_EQ(array.get_offsets()[2], 2);
+ const auto& elements = assert_cast<const
ColumnNullable&>(array.get_data());
+ const auto& variants = assert_cast<const
ColumnVariantV2&>(elements.get_nested_column());
+ EXPECT_EQ(variants.get_value_ref(0).get_int(), 42);
+ EXPECT_TRUE(elements.is_null_at(1));
+ ASSERT_TRUE(reader.close().ok());
+ }
+ void check_decimal_rescaling(bool legacy, bool nested = false, int scale =
10) {
+ const auto decimal_type =
+ make_nullable(legacy ?
DataTypePtr(std::make_shared<DataTypeDecimalV2>(27, 9))
+ :
DataTypePtr(std::make_shared<DataTypeDecimal128>(27, 9)));
+ set_column_type(nested ?
make_nullable(std::make_shared<DataTypeArray>(decimal_type))
+ : decimal_type);
+ const int64_t unscaled = scale == 10 ? 12345678901LL : 1234567;
+ arrow::Decimal128Builder builder(arrow::decimal128(20, scale),
+ arrow::default_memory_pool());
+ ASSERT_TRUE(builder.Append(arrow::Decimal128(unscaled)).ok());
+ ASSERT_TRUE(builder.Append(arrow::Decimal128(-unscaled)).ok());
+ ASSERT_TRUE(builder.AppendNull().ok());
+ auto decimals = builder.Finish();
+ ASSERT_TRUE(decimals.ok());
+ std::shared_ptr<arrow::Array> values = *decimals;
+ if (nested) {
+ arrow::Int32Builder offsets;
+ ASSERT_TRUE(offsets.AppendValues({0, 3}).ok());
+ auto positions = offsets.Finish();
+ ASSERT_TRUE(positions.ok());
+ auto lists = arrow::ListArray::FromArrays(**positions, *values);
+ ASSERT_TRUE(lists.ok());
+ values = *lists;
+ }
+ append_batch(
+
arrow::RecordBatch::Make(arrow::schema({arrow::field("raw_value",
values->type())}),
+ nested ? 1 : 3, {values}));
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ ASSERT_TRUE(prepare(reader, FileFormat::ARROW).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ const auto status = reader.get_block(&output, &eos);
+ if (!legacy) {
+ EXPECT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("Cannot decode Arrow column"),
std::string::npos)
+ << status;
+ } else {
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(output.rows(), nested ? 1 : 3);
+ const IColumn* result = output.get_by_position(0).column.get();
+ if (nested) {
+ const auto& outer = assert_cast<const
ColumnNullable&>(*result).get_nested_column();
+ result = &assert_cast<const ColumnArray&>(outer).get_data();
+ }
+ const auto& nullable = assert_cast<const ColumnNullable&>(*result);
+ const auto& numbers =
+ assert_cast<const
ColumnDecimal<TYPE_DECIMALV2>&>(nullable.get_nested_column());
+ const int64_t expected = scale == 10 ? 1234567890 : 1234567000;
+ EXPECT_EQ(numbers.get_element(0).value(), expected);
+ EXPECT_EQ(numbers.get_element(1).value(), -expected);
+ EXPECT_TRUE(nullable.is_null_at(2));
+ }
+ ASSERT_TRUE(reader.close().ok());
+ }
+};
+
+TEST_F(LoadArrowPipeTest,
NullFixedSizeListPreservesNonNullableStructDescendants) {
+ check_null_fixed_list_struct(false);
+}
+
+TEST_F(LoadArrowPipeTest,
NullFixedSizeListCompactsStructsBeforeCheckedNarrowing) {
+ check_null_fixed_list_struct(true, false, 2);
+}
+
+TEST_F(LoadArrowPipeTest,
AllNullFixedSizeListCompactsNonNullableStructDescendants) {
+ check_null_fixed_list_struct(false, true);
+}
+
+TEST_F(LoadArrowPipeTest, NullFixedSizeListPreservesNativeVariantStorage) {
+ check_null_fixed_list_variant(false);
+}
+
+TEST_F(LoadArrowPipeTest, NullFixedSizeListPreservesNativeVariantExtension) {
+ check_null_fixed_list_variant(true);
+}
+
+TEST_F(LoadArrowPipeTest, DecimalV2RetainsHigherScaleInputTruncation) {
+ check_decimal_rescaling(true);
+}
+
+TEST_F(LoadArrowPipeTest, NestedDecimalV2RetainsHigherScaleInputTruncation) {
+ check_decimal_rescaling(true, true);
+}
+
+TEST_F(LoadArrowPipeTest, DecimalV2RetainsLowerScaleInputExpansion) {
+ check_decimal_rescaling(true, false, 6);
+}
+
+TEST_F(LoadArrowPipeTest, DecimalV3StillRejectsLossyScaleReduction) {
+ check_decimal_rescaling(false);
+}
+
+TEST_F(LoadArrowPipeTest, NativeVariantStoragePreservesValuesAndSqlNulls) {
+ check_native_variant(false);
+}
+
+TEST_F(LoadArrowPipeTest, NativeVariantExtensionPreservesValuesAndSqlNulls) {
+ check_native_variant(true);
+}
+
+TEST_F(LoadArrowPipeTest, NativeVariantInsideArrayPreservesValuesAndSqlNulls) {
+ check_native_variant(false, 1);
+}
+
+TEST_F(LoadArrowPipeTest, NativeVariantInsideMapSurvivesKeyNarrowing) {
+ check_native_variant(false, 2);
+}
+
+TEST_F(LoadArrowPipeTest, NativeVariantInsideStructSurvivesSiblingNarrowing) {
+ check_native_variant(false, 3);
+}
+
+TEST_F(LoadArrowPipeTest,
NativeVariantExtensionInsideArrayPreservesValuesAndSqlNulls) {
+ check_native_variant(true, 1);
+}
+
+TEST_F(LoadArrowPipeTest, NativeVariantExtensionInsideMapSurvivesKeyNarrowing)
{
+ check_native_variant(true, 2);
+}
+
+TEST_F(LoadArrowPipeTest,
NativeVariantExtensionInsideStructSurvivesSiblingNarrowing) {
+ check_native_variant(true, 3);
+}
+
+TEST_F(LoadArrowPipeTest, CountStateUsesFixedSizeBinarySerialization) {
+ check_count_state(false);
+}
+
+TEST_F(LoadArrowPipeTest, CountStateAcceptsBinarySerialization) {
+ check_count_state(true);
+}
+
+TEST_F(LoadArrowPipeTest, CountStateRejectsWrongFixedSizeBinaryWidth) {
+ check_count_state(false, 4);
+}
+
+TEST_F(LoadArrowPipeTest, CountStateRejectsWrongBinaryWidth) {
+ check_count_state(true, 4);
+}
+
+TEST_F(LoadArrowPipeTest, StructInputForAStringSlotReturnsAnError) {
+ arrow::Int32Builder builder;
+ ASSERT_TRUE(builder.Append(7).ok());
+ auto values = builder.Finish();
+ ASSERT_TRUE(values.ok());
+ auto structs = arrow::StructArray::Make({*values},
std::vector<std::string> {"value"});
+ ASSERT_TRUE(structs.ok());
+ append_batch(arrow::RecordBatch::Make(
+ arrow::schema({arrow::field("raw_value", (*structs)->type())}), 1,
{*structs}));
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ ASSERT_TRUE(prepare(reader, FileFormat::ARROW).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ const auto status = reader.get_block(&output, &eos);
+ EXPECT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("Cannot decode Arrow column"),
std::string::npos) << status;
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadArrowPipeTest, ConcatenatedStreamsRespectBatchBoundariesAndNulls) {
+ auto options = state.query_options();
+ options.__set_batch_size(1);
+ state.set_query_options(options);
+ append_stream({});
+ append_stream({"first", std::nullopt, "third"});
+ append_stream({"last"});
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ const auto status = prepare(reader, FileFormat::ARROW);
+ ASSERT_TRUE(status.ok()) << status;
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ const std::vector<std::optional<std::string>> expected {"first",
std::nullopt, "third", "last"};
+ for (const auto& value : expected) {
+ const auto read_status = reader.get_block(&output, &eos);
+ ASSERT_TRUE(read_status.ok()) << read_status;
+ ASSERT_EQ(output.rows(), 1);
+ const auto& column = assert_cast<const
ColumnNullable&>(*output.get_by_position(0).column);
+ EXPECT_EQ(column.is_null_at(0), !value.has_value());
+ if (value) {
+ EXPECT_EQ(column.get_data_at(0).to_string(), *value);
+ }
+ }
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ EXPECT_TRUE(eos);
+ EXPECT_EQ(output.rows(), 0);
+ EXPECT_EQ(_pipe->get_queue_size(), 0);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadArrowPipeTest,
ReorderedFieldsAndDictionaryStringsUseFileLocalMapping) {
+ arrow::StringDictionaryBuilder builder;
+ ASSERT_TRUE(builder.Append("dictionary").ok());
+ ASSERT_TRUE(builder.AppendNull().ok());
+ auto values = builder.Finish();
+ ASSERT_TRUE(values.ok());
+ arrow::Int32Builder unused;
+ ASSERT_TRUE(unused.AppendValues({1, 2}).ok());
+ auto ignored = unused.Finish();
+ ASSERT_TRUE(ignored.ok());
+ auto schema = arrow::schema({arrow::field("ignored", arrow::int32()),
+ arrow::field("raw_value",
(*values)->type())});
+ append_batch(arrow::RecordBatch::Make(schema, 2, {*ignored, *values}));
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ ASSERT_TRUE(prepare(reader, FileFormat::ARROW).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ const auto status = reader.get_block(&output, &eos);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(output.rows(), 2);
+ EXPECT_EQ(output.get_by_position(0).column->get_data_at(0).to_string(),
"dictionary");
+ EXPECT_TRUE(output.get_by_position(0).column->is_null_at(1));
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadArrowPipeTest, NestedArraysKeepNullsAcrossBatchBoundaries) {
+ set_column_type(make_nullable(
+
std::make_shared<DataTypeArray>(make_nullable(std::make_shared<DataTypeInt32>()))));
+ auto options = state.query_options();
+ options.__set_batch_size(1);
+ state.set_query_options(options);
+ auto values = std::make_shared<arrow::Int32Builder>();
+ arrow::ListBuilder lists(arrow::default_memory_pool(), values);
+ ASSERT_TRUE(lists.Append().ok());
+ ASSERT_TRUE(values->Append(1).ok());
+ ASSERT_TRUE(values->AppendNull().ok());
+ ASSERT_TRUE(lists.AppendNull().ok());
+ ASSERT_TRUE(lists.Append().ok());
+ auto array = lists.Finish();
+ ASSERT_TRUE(array.ok());
+ auto schema = arrow::schema({arrow::field("raw_value", (*array)->type())});
+ append_batch(arrow::RecordBatch::Make(schema, 3, {*array}));
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ ASSERT_TRUE(prepare(reader, FileFormat::ARROW).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ for (int row = 0; row < 3; ++row) {
+ const auto status = reader.get_block(&output, &eos);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(output.rows(), 1);
+ const auto& nullable =
+ assert_cast<const
ColumnNullable&>(*output.get_by_position(0).column);
+ EXPECT_EQ(nullable.is_null_at(0), row == 1);
+ const auto& array_column = assert_cast<const
ColumnArray&>(nullable.get_nested_column());
+ EXPECT_EQ(array_column.get_offsets()[0], row == 0 ? 2 : 0);
+ if (row == 0) {
+ const auto& elements = assert_cast<const
ColumnNullable&>(array_column.get_data());
+ EXPECT_EQ(elements.get_nested_column().get_int(0), 1);
+ EXPECT_TRUE(array_column.get_data().is_null_at(1));
+ }
+ }
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadArrowPipeTest, BinaryStringPayloadPreservesItsBytes) {
+ arrow::BinaryBuilder builder;
+ ASSERT_TRUE(builder.Append("\xFF", 1).ok());
+ auto values = builder.Finish();
+ ASSERT_TRUE(values.ok());
+ auto schema = arrow::schema({arrow::field("raw_value", arrow::binary())});
+ append_batch(arrow::RecordBatch::Make(schema, 1, {*values}));
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ ASSERT_TRUE(prepare(reader, FileFormat::ARROW).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ const auto status = reader.get_block(&output, &eos);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(output.rows(), 1);
+ EXPECT_EQ(output.get_by_position(0).column->get_data_at(0).to_string(),
"\xFF");
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadArrowPipeTest, StructChildrenBindPositionallyInsideAnArray) {
+ const auto structure = make_nullable(std::make_shared<DataTypeStruct>(
+ DataTypes {make_nullable(std::make_shared<DataTypeInt32>()),
+ make_nullable(std::make_shared<DataTypeInt32>())},
+ Strings {"a", "b"}));
+ set_column_type(make_nullable(std::make_shared<DataTypeArray>(structure)));
+ arrow::Int64Builder builder;
+ ASSERT_TRUE(builder.AppendValues({7, 9}).ok());
+ auto values = builder.Finish();
+ ASSERT_TRUE(values.ok());
+ auto structs = arrow::StructArray::Make(
+ {*values, *values}, arrow::FieldVector {arrow::field("A",
arrow::int64()),
+ arrow::field("different",
arrow::int64())});
+ ASSERT_TRUE(structs.ok());
+ arrow::Int32Builder offsets;
+ ASSERT_TRUE(offsets.AppendValues({0, 2}).ok());
+ auto positions = offsets.Finish();
+ ASSERT_TRUE(positions.ok());
+ auto lists = arrow::ListArray::FromArrays(**positions, **structs);
+ ASSERT_TRUE(lists.ok());
+ append_batch(arrow::RecordBatch::Make(
+ arrow::schema({arrow::field("raw_value", (*lists)->type())}), 1,
{*lists}));
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ ASSERT_TRUE(prepare(reader, FileFormat::ARROW).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ const auto status = reader.get_block(&output, &eos);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(output.rows(), 1);
+ const auto& outer = assert_cast<const
ColumnNullable&>(*output.get_by_position(0).column);
+ const auto& array = assert_cast<const
ColumnArray&>(outer.get_nested_column());
+ const auto& elements = assert_cast<const
ColumnNullable&>(array.get_data());
+ const auto& fields = assert_cast<const
ColumnStruct&>(elements.get_nested_column());
+ EXPECT_FALSE(fields.get_column(0).is_null_at(0));
+ EXPECT_EQ(
+ assert_cast<const
ColumnNullable&>(fields.get_column(0)).get_nested_column().get_int(0),
+ 7);
+ EXPECT_FALSE(fields.get_column(1).is_null_at(1));
+ EXPECT_EQ(
+ assert_cast<const
ColumnNullable&>(fields.get_column(1)).get_nested_column().get_int(1),
+ 9);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadArrowPipeTest, Date64RetainsDorisRangeValidation) {
+ set_column_type(make_nullable(std::make_shared<DataTypeDateV2>()));
+ arrow::Date64Builder builder;
+ ASSERT_TRUE(builder.Append(253402300800000LL).ok());
+ auto values = builder.Finish();
+ ASSERT_TRUE(values.ok());
+ append_batch(arrow::RecordBatch::Make(
+ arrow::schema({arrow::field("raw_value", arrow::date64())}), 1,
{*values}));
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ ASSERT_TRUE(prepare(reader, FileFormat::ARROW).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ const auto status = reader.get_block(&output, &eos);
+ EXPECT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("Doris DATE range"), std::string::npos)
<< status;
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadArrowPipeTest, UInt32Ipv4PreservesTheUnsignedAddressBits) {
+ set_column_type(make_nullable(std::make_shared<DataTypeIPv4>()));
+ arrow::UInt32Builder builder;
+ ASSERT_TRUE(builder.AppendValues({0x7fffffffU, 0x80000000U,
0xffffffffU}).ok());
+ auto values = builder.Finish();
+ ASSERT_TRUE(values.ok());
+ append_batch(arrow::RecordBatch::Make(
+ arrow::schema({arrow::field("raw_value", arrow::uint32())}), 3,
{*values}));
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ ASSERT_TRUE(prepare(reader, FileFormat::ARROW).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ const auto status = reader.get_block(&output, &eos);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(output.rows(), 3);
+ const auto type = destination_slot->type();
+ EXPECT_EQ(type->to_string(*output.get_by_position(0).column, 0),
"127.255.255.255");
+ EXPECT_EQ(type->to_string(*output.get_by_position(0).column, 1),
"128.0.0.0");
+ EXPECT_EQ(type->to_string(*output.get_by_position(0).column, 2),
"255.255.255.255");
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadArrowPipeTest, NullStructIgnoresAnOverflowingHiddenChild) {
+ set_column_type(make_nullable(std::make_shared<DataTypeStruct>(
+ DataTypes {make_nullable(std::make_shared<DataTypeInt32>())},
Strings {"a"})));
+ arrow::Int64Builder builder;
+ ASSERT_TRUE(builder.AppendValues({std::numeric_limits<int64_t>::max(),
7}).ok());
+ auto values = builder.Finish();
+ ASSERT_TRUE(values.ok());
+ arrow::BooleanBuilder validity;
+ ASSERT_TRUE(validity.AppendValues(std::vector<bool> {false, true}).ok());
+ auto valid = validity.Finish();
+ ASSERT_TRUE(valid.ok());
+ auto structs = arrow::StructArray::Make({*values},
+ arrow::FieldVector
{arrow::field("a", arrow::int64())},
+ (*valid)->data()->buffers[1], 1);
+ ASSERT_TRUE(structs.ok());
+ append_batch(arrow::RecordBatch::Make(
+ arrow::schema({arrow::field("raw_value", (*structs)->type())}), 2,
{*structs}));
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ ASSERT_TRUE(prepare(reader, FileFormat::ARROW).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ const auto status = reader.get_block(&output, &eos);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(output.rows(), 2);
+ EXPECT_TRUE(output.get_by_position(0).column->is_null_at(0));
+ const auto& nullable = assert_cast<const
ColumnNullable&>(*output.get_by_position(0).column);
+ const auto& fields = assert_cast<const
ColumnStruct&>(nullable.get_nested_column());
+ EXPECT_EQ(
+ assert_cast<const
ColumnNullable&>(fields.get_column(0)).get_nested_column().get_int(1),
+ 7);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadArrowPipeTest, NullListIgnoresAnOverflowingHiddenChild) {
+ check_hidden_container_values(0);
+}
+
+TEST_F(LoadArrowPipeTest, NullFixedSizeListIgnoresAnOverflowingHiddenChild) {
+ check_hidden_container_values(1);
+}
+
+TEST_F(LoadArrowPipeTest, NullMapIgnoresOverflowingHiddenKeysAndValues) {
+ check_hidden_container_values(2);
+}
+
+TEST_F(LoadArrowPipeTest, NumericIntStillRejectsAnOverflowingUInt32) {
+ set_column_type(make_nullable(std::make_shared<DataTypeInt32>()));
+ arrow::UInt32Builder builder;
+ ASSERT_TRUE(builder.Append(0xffffffffU).ok());
+ auto values = builder.Finish();
+ ASSERT_TRUE(values.ok());
+ append_batch(arrow::RecordBatch::Make(
+ arrow::schema({arrow::field("raw_value", arrow::uint32())}), 1,
{*values}));
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ ASSERT_TRUE(prepare(reader, FileFormat::ARROW).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ EXPECT_FALSE(reader.get_block(&output, &eos).ok());
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadArrowPipeTest, ReadsCompressedIpcBatches) {
+ arrow::StringBuilder builder;
+ ASSERT_TRUE(builder.Append("compressed").ok());
+ ASSERT_TRUE(builder.AppendNull().ok());
+ auto values = builder.Finish();
+ ASSERT_TRUE(values.ok());
+ auto schema = arrow::schema({arrow::field("raw_value", arrow::utf8())});
+ append_batch(arrow::RecordBatch::Make(schema, 2, {*values}), true);
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ ASSERT_TRUE(prepare(reader, FileFormat::ARROW).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ const auto status = reader.get_block(&output, &eos);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(output.rows(), 2);
+ EXPECT_EQ(output.get_by_position(0).column->get_data_at(0).to_string(),
"compressed");
+ EXPECT_TRUE(output.get_by_position(0).column->is_null_at(1));
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadArrowPipeTest, TimestampRoundingUsesTheDorisSerde) {
+ check_timestamp_rounding(false);
+}
+
+TEST_F(LoadArrowPipeTest, NestedTimestampRoundingUsesTheDorisSerde) {
+ check_timestamp_rounding(true);
+}
+
+TEST_F(LoadArrowPipeTest, MalformedIpcReturnsAnError) {
+ append("invalid IPC");
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ ASSERT_TRUE(prepare(reader, FileFormat::ARROW).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ EXPECT_FALSE(reader.get_block(&output, &eos).ok());
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadArrowPipeTest, EmptyFinishedPipeReturnsEof) {
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ const auto status = prepare(reader, FileFormat::ARROW);
+ ASSERT_TRUE(status.ok()) << status;
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ EXPECT_TRUE(eos);
+ EXPECT_EQ(output.rows(), 0);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadArrowPipeTest, RejectsMissingRequestedColumn) {
+ append_stream({"value"}, "other");
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ const auto status = prepare(reader, FileFormat::ARROW);
+ ASSERT_TRUE(status.ok()) << status;
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ EXPECT_FALSE(reader.get_block(&output, &eos).ok());
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadJsonPipeTest, SchemaProbeReadsTheSampleWithoutConsumingTheLoadPipe)
{
+ const std::string document = R"({"raw_value":"payload"})";
+ append(document);
+ ASSERT_TRUE(_pipe->finish().ok());
+ ASSERT_TRUE(_context->allocate_schema_buffer().ok());
+ _context->schema_buffer()->put_bytes(document.data(), document.size());
+ TFileRangeDesc range;
+ range.__set_file_type(TFileType::FILE_STREAM);
+ range.__set_load_id(_context->id.to_thrift());
+ range.__set_start_offset(0);
+ range.__set_size(-1);
+ std::vector<ColumnDefinition> schema;
+ const auto status = fetch_file_schema(params, range, io, &profile,
&schema);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(schema.size(), 1);
+ EXPECT_EQ(schema[0].name, "raw_value");
+ LoadTableReader reader(source.get(), destination.get());
+ ASSERT_TRUE(prepare(reader).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ ASSERT_EQ(output.rows(), 1);
+ EXPECT_EQ(output.get_by_position(0).column->get_data_at(0).to_string(),
"payload");
+ ASSERT_TRUE(reader.close().ok());
+}
+
TEST_F(LoadJsonPipeTest, ReadsEveryKafkaMessageBeforePipeEof) {
append(R"({"raw_value":"first"})");
append(R"({"raw_value":"second"})");
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadPlanner.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadPlanner.java
index e3be4cbdfb9..a2862d13b79 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadPlanner.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadPlanner.java
@@ -303,6 +303,8 @@ public class NereidsStreamLoadPlanner {
params.setTotalLoadStreams(taskInfo.getStreamPerNode());
params.setNumLocalSink(1);
TQueryOptions queryOptions = new TQueryOptions();
+ // Load plans do not inherit session query options; explicitly select
the V2 load path.
+ queryOptions.setEnableFileScannerV2(true);
queryOptions.setQueryType(TQueryType.LOAD);
int timeout = taskInfo.getTimeout();
queryOptions.setQueryTimeout(timeout);
diff --git
a/regression-test/data/load_p0/http_stream/test_http_stream_v2_schema.out
b/regression-test/data/load_p0/http_stream/test_http_stream_v2_schema.out
new file mode 100644
index 00000000000..c1b43141078
--- /dev/null
+++ b/regression-test/data/load_p0/http_stream/test_http_stream_v2_schema.out
@@ -0,0 +1,15 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !schemas --
+csv 1 first
+csv 2 second
+csv_names 1 first
+csv_names 2 second
+csv_types 1 first
+csv_types 2 second
+json_lines 1 first
+json_lines 2 second
+json_paths 1 first
+json_paths 2 second
+text 1 first
+text 2 second
+
diff --git
a/regression-test/data/load_p0/stream_load/test_arrow_stream_load_uppercase_column.out
b/regression-test/data/load_p0/stream_load/test_arrow_stream_load_uppercase_column.out
index 31344c45f9d..e87999fd5bc 100644
---
a/regression-test/data/load_p0/stream_load/test_arrow_stream_load_uppercase_column.out
+++
b/regression-test/data/load_p0/stream_load/test_arrow_stream_load_uppercase_column.out
@@ -6,3 +6,7 @@
-- !auto --
1:2.5
2:3.5
+
+-- !expressions --
+2 3.5 4 loaded
+
diff --git
a/regression-test/suites/load_p0/http_stream/test_http_stream_v2_schema.groovy
b/regression-test/suites/load_p0/http_stream/test_http_stream_v2_schema.groovy
new file mode 100644
index 00000000000..85062aa738c
--- /dev/null
+++
b/regression-test/suites/load_p0/http_stream/test_http_stream_v2_schema.groovy
@@ -0,0 +1,69 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_http_stream_v2_schema", "p0") {
+ sql "DROP TABLE IF EXISTS test_http_stream_v2_schema"
+ sql """
+ CREATE TABLE test_http_stream_v2_schema (
+ kind VARCHAR(32),
+ id INT,
+ name VARCHAR(32)
+ ) DUPLICATE KEY(kind, id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+ def db = context.config.getDbNameByFile(context.file)
+ def cases = [
+ [kind: 'csv', format: 'csv', fields: 'c1, c2',
+ data: '1,first\n2,second\n', options: ''],
+ [kind: 'csv_names', format: 'csv_with_names', fields: 'id, name',
+ data: 'id,name\n1,first\n2,second\n', options: ''],
+ [kind: 'csv_types', format: 'csv_with_names_and_types', fields: 'id,
name',
+ data: 'id,name\nINT,STRING\n1,first\n2,second\n', options: ''],
+ [kind: 'text', format: 'hive_text', fields: 'c1, c2',
+ data: '1,first\n2,second\n', options: ''],
+ [kind: 'json_lines', format: 'json', fields: 'id, name',
+ data: '{"id":1,"name":"first"}\n{"id":2,"name":"second"}\n',
+ options: ', "read_json_by_line"="true"'],
+ [kind: 'json_paths', format: 'json', fields: 'id, name',
+ data:
'{"rows":[{"id":1,"nested":{"name":"first"}},{"id":2,"nested":{"name":"second"}}]}',
+ options: ', "strip_outer_array"="true", "json_root"="$.rows",
"jsonpaths"=\'["$.id","$.nested.name"]\'']
+ ]
+ for (def item : cases) {
+ streamLoad {
+ set 'version', '1'
+ set 'sql', """
+ INSERT INTO ${db}.test_http_stream_v2_schema
+ SELECT '${item.kind}', ${item.fields}
+ FROM http_stream("format"="${item.format}",
"column_separator"="," ${item.options})
+ """
+ inputText item.data
+ time 10000
+ check { result, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ def response = parseJson(result)
+ assertEquals("success", response.Status.toLowerCase())
+ assertEquals(2, response.NumberLoadedRows)
+ assertEquals(0, response.NumberFilteredRows)
+ }
+ }
+ }
+ sql "sync"
+ order_qt_schemas "SELECT * FROM test_http_stream_v2_schema ORDER BY kind,
id"
+}
diff --git
a/regression-test/suites/load_p0/stream_load/test_arrow_stream_load_uppercase_column.groovy
b/regression-test/suites/load_p0/stream_load/test_arrow_stream_load_uppercase_column.groovy
index 16e3fe4c861..7d8ad6c4e13 100644
---
a/regression-test/suites/load_p0/stream_load/test_arrow_stream_load_uppercase_column.groovy
+++
b/regression-test/suites/load_p0/stream_load/test_arrow_stream_load_uppercase_column.groovy
@@ -86,4 +86,38 @@ suite("test_arrow_stream_load_uppercase_column", "p0") {
FROM test_arrow_stream_load_uppercase_column_auto
ORDER BY id
"""
+
+ sql "DROP TABLE IF EXISTS test_arrow_stream_load_v2_expressions"
+ sql """
+ CREATE TABLE test_arrow_stream_load_v2_expressions (
+ id INT,
+ EV DOUBLE,
+ doubled INT,
+ tag VARCHAR(16) DEFAULT "loaded"
+ ) DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+ streamLoad {
+ table 'test_arrow_stream_load_v2_expressions'
+ set 'format', 'arrow'
+ set 'columns', 'id,EV,doubled=id*2'
+ set 'strict_mode', 'true'
+ set 'where', 'id > 1'
+ file 'arrow_uppercase_column.arrow'
+ time 10000
+ check { result, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ def response = parseJson(result)
+ assertEquals("success", response.Status.toLowerCase())
+ assertEquals(1, response.NumberLoadedRows)
+ assertEquals(1, response.NumberUnselectedRows)
+ assertEquals(0, response.NumberFilteredRows)
+ }
+ }
+ sql "sync"
+ order_qt_expressions "SELECT * FROM test_arrow_stream_load_v2_expressions
ORDER BY id"
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
