This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new cc550860ae5 [feature](external) Support Elasticsearch HTTP scans in
FileScannerV2 (#69614)
cc550860ae5 is described below
commit cc550860ae558d57e2b939cacc8628402693268a
Author: Gabriel <[email protected]>
AuthorDate: Mon Oct 26 17:49:19 2026 +0800
[feature](external) Support Elasticsearch HTTP scans in FileScannerV2
(#69614)
### What problem does this PR solve?
Related PR: #69601
Problem Summary:
Elasticsearch HTTP scans always use FileScanner v1, even when
`enable_file_scanner_v2` is enabled. Add an ES TableReader/FileReader
pair and route `FORMAT_ES_HTTP` queries through FileScannerV2 when that
session option is enabled.
The reader exposes a file-local schema, uses MaterializedColumnMapper
for projection and predicate localization, and applies residual filters
with v2 row accounting. It reuses the existing ES query builder, HTTP
scroll transport, authentication, SSL settings, catalog-provided
endpoints, and type conversions. Shared host selection and materialized
complex-schema helpers keep the two scanner paths consistent.
Parse the initial cached response during open so early
LIMIT/cancellation can clear its scroll context. Close the stream on
EOF, cancellation, open/read errors, and split changes, and report close
failures once.
The RLE header reach baseline increases by two for the new ES reader and
test translation units, which use the existing table/expr header
closure. The shared parser retains its existing type dispatch, with a
documented complexity suppression on the new projection overload.
### Release note
Support Elasticsearch HTTP queries in FileScannerV2 when
`enable_file_scanner_v2` is enabled. Disabling the option continues to
select FileScanner v1.
### Check List (For Author)
- Test
- [x] Unit Test: 61 ASAN BE tests covering ES v2 projection, doc values,
`_id`, nullable values/arrays, multi-page batching, zero-column row
counting, residual filtering, split advancement, cancellation, error
cleanup, and a real local HTTP scroll server; existing scanner dispatch,
legacy ES, and Remote Doris tests also pass.
- [ ] Regression test: added ES 5/6/7/8 v1/v2 differential coverage with
source/docvalue scans and small batches. Groovy syntax checked; live
external ES regression not run locally.
- [x] clang-format 16, clang-tidy on changed C++ lines, and BE
build-hygiene checks.
- Behavior changed:
- [x] Yes. ES queries honor the v2 session option.
- Does this need documentation?
- [x] No. Uses the existing scanner session option and ES catalog
properties.
### 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 | 8 +
be/src/format/table/es/es_http_reader.cpp | 56 +--
be/src/format/table/es/es_scan_reader.cpp | 59 +++
be/src/format/table/es/es_scan_reader.h | 3 +
be/src/format/table/es/es_scroll_parser.cpp | 25 +-
be/src/format/table/es/es_scroll_parser.h | 9 +
be/src/format_v2/file_reader.h | 1 +
be/src/format_v2/materialized_reader_util.cpp | 47 ++
be/src/format_v2/materialized_reader_util.h | 5 +
be/src/format_v2/table/es_reader.cpp | 295 +++++++++++
be/src/format_v2/table/es_reader.h | 107 ++++
be/src/format_v2/table/iceberg_reader.cpp | 2 +
be/src/format_v2/table/remote_doris_reader.cpp | 52 +-
be/src/format_v2/table_reader.cpp | 2 +
be/test/exec/scan/file_scanner_v2_test.cpp | 11 +-
be/test/format_v2/table/es_reader_test.cpp | 540 +++++++++++++++++++++
build-support/check-header-deps.py | 2 +-
.../es/test_es_file_scanner_v2.groovy | 72 +++
19 files changed, 1182 insertions(+), 115 deletions(-)
diff --git a/be/src/exec/operator/file_scan_operator.cpp
b/be/src/exec/operator/file_scan_operator.cpp
index e439f308ef4..e8666131ffb 100644
--- a/be/src/exec/operator/file_scan_operator.cpp
+++ b/be/src/exec/operator/file_scan_operator.cpp
@@ -224,7 +224,6 @@ bool FileScanLocalState::should_use_file_scanner_v2(const
TQueryOptions& query_o
// after scanner construction. Keep versioned Iceberg scans on V2 without
eagerly listing
// every file.
return scanner_v2_requested && !is_load &&
- scan_params.format_type != TFileFormatType::FORMAT_ES_HTTP &&
scan_params.format_type != TFileFormatType::FORMAT_LANCE &&
!is_transactional_hive;
}
diff --git a/be/src/exec/scan/file_scanner_v2.cpp
b/be/src/exec/scan/file_scanner_v2.cpp
index 15a71f2c2d1..975f1a74e60 100644
--- a/be/src/exec/scan/file_scanner_v2.cpp
+++ b/be/src/exec/scan/file_scanner_v2.cpp
@@ -60,6 +60,7 @@
#include "format_v2/jni/max_compute_jni_reader.h"
#include "format_v2/jni/trino_connector_jni_reader.h"
#include "format_v2/table/adbc_reader.h"
+#include "format_v2/table/es_reader.h"
#include "format_v2/table/fluss_hybrid_reader.h"
#include "format_v2/table/hive_reader.h"
#include "format_v2/table/hudi_reader.h"
@@ -356,6 +357,8 @@ bool FileScannerV2::is_supported(const
TFileScanRangeParams& params, const TFile
if (format_type == TFileFormatType::FORMAT_PARQUET ||
format_type == TFileFormatType::FORMAT_ORC) {
return is_supported_table_format(range);
+ } 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);
} else if (format_type == TFileFormatType::FORMAT_JNI) {
@@ -801,6 +804,8 @@ Status FileScannerV2::_create_table_reader_for_format(
*reader =
std::make_unique<format::max_compute::MaxComputeJniReader>(mc_desc);
} else if (table_format == "trino_connector") {
*reader =
std::make_unique<format::trino_connector::TrinoConnectorJniReader>();
+ } else if (table_format == "es") {
+ *reader = std::make_unique<format::es::EsReader>();
} else if (table_format == "remote_doris") {
*reader = std::make_unique<format::remote_doris::RemoteDorisReader>();
} else if (table_format == "adbc") {
@@ -1092,6 +1097,9 @@ Status
FileScannerV2::_to_file_format(TFileFormatType::type format_type,
case TFileFormatType::FORMAT_ARROW:
*file_format = format::FileFormat::ARROW;
return Status::OK();
+ case TFileFormatType::FORMAT_ES_HTTP:
+ *file_format = format::FileFormat::ES_HTTP;
+ return Status::OK();
case TFileFormatType::FORMAT_WAL:
*file_format = format::FileFormat::WAL;
return Status::OK();
diff --git a/be/src/format/table/es/es_http_reader.cpp
b/be/src/format/table/es/es_http_reader.cpp
index 0efb3c19e75..5bf42439447 100644
--- a/be/src/format/table/es/es_http_reader.cpp
+++ b/be/src/format/table/es/es_http_reader.cpp
@@ -19,9 +19,7 @@
#include <gen_cpp/PlanNodes_types.h>
-#include <algorithm>
-#include <cctype>
-#include <sstream>
+#include <utility>
#include "format/table/es/es_scan_reader.h"
#include "format/table/es/es_scroll_parser.h"
@@ -29,7 +27,6 @@
#include "runtime/descriptors.h"
#include "runtime/runtime_profile.h"
#include "runtime/runtime_state.h"
-#include "service/backend_options.h"
namespace doris {
@@ -65,29 +62,7 @@ void EsHttpReader::_init_profile() {
}
std::string EsHttpReader::_extract_hostname(std::string host) {
- auto scheme_end = host.find("://");
- if (scheme_end != std::string::npos) {
- host = host.substr(scheme_end + 3);
- }
-
- if (!host.empty() && host.front() == '[') {
- auto bracket_end = host.find(']');
- if (bracket_end != std::string::npos) {
- return host.substr(1, bracket_end - 1);
- }
- return host;
- }
-
- auto colon = host.rfind(':');
- if (colon == std::string::npos) {
- return host;
- }
- std::string port = host.substr(colon + 1);
- if (!port.empty() && std::all_of(port.begin(), port.end(),
- [](unsigned char ch) { return
std::isdigit(ch) != 0; })) {
- return host.substr(0, colon);
- }
- return host;
+ return ESScanReader::extract_hostname(std::move(host));
}
Status EsHttpReader::init_reader() {
@@ -225,32 +200,7 @@ Status
EsHttpReader::_get_columns_impl(std::unordered_map<std::string, DataTypeP
}
std::string EsHttpReader::_select_host(const std::map<std::string,
std::string>& properties) const {
- // If es_hosts contains multiple hosts, prefer localhost for locality
- auto it = properties.find(KEY_ES_HOSTS);
- if (it != properties.end() && !it->second.empty()) {
- std::string localhost = BackendOptions::get_localhost();
- std::string best;
- std::istringstream stream(it->second);
- std::string host;
- while (std::getline(stream, host, ',')) {
- if (best.empty()) {
- best = host;
- }
- std::string hostname = _extract_hostname(host);
- if (hostname == localhost) {
- return host;
- }
- }
- if (!best.empty()) {
- return best;
- }
- }
- // Fallback to host_port
- auto hp = properties.find(ESScanReader::KEY_HOST_PORT);
- if (hp != properties.end()) {
- return hp->second;
- }
- return "";
+ return ESScanReader::select_host(properties);
}
} // namespace doris
diff --git a/be/src/format/table/es/es_scan_reader.cpp
b/be/src/format/table/es/es_scan_reader.cpp
index 037a3535296..04b506f1cc4 100644
--- a/be/src/format/table/es/es_scan_reader.cpp
+++ b/be/src/format/table/es/es_scan_reader.cpp
@@ -19,6 +19,8 @@
#include <stdlib.h>
+#include <algorithm>
+#include <cctype>
#include <map>
#include <sstream>
#include <string>
@@ -28,6 +30,7 @@
#include "common/status.h"
#include "format/table/es/es_scroll_parser.h"
#include "format/table/es/es_scroll_query.h"
+#include "service/backend_options.h"
#include "service/http/http_method.h"
namespace doris {
@@ -266,4 +269,60 @@ Status ESScanReader::close() {
return Status::InternalError("es_scan_reader delete scroll context
failure");
}
}
+
+std::string ESScanReader::extract_hostname(std::string host) {
+ auto scheme_end = host.find("://");
+ if (scheme_end != std::string::npos) {
+ host = host.substr(scheme_end + 3);
+ }
+
+ if (!host.empty() && host.front() == '[') {
+ auto bracket_end = host.find(']');
+ if (bracket_end != std::string::npos) {
+ return host.substr(1, bracket_end - 1);
+ }
+ return host;
+ }
+
+ auto colon = host.rfind(':');
+ if (colon == std::string::npos) {
+ return host;
+ }
+ std::string port = host.substr(colon + 1);
+ if (!port.empty() &&
+ std::ranges::all_of(port, [](unsigned char ch) { return
std::isdigit(ch) != 0; })) {
+ return host.substr(0, colon);
+ }
+ return host;
+}
+
+std::string ESScanReader::select_host(const std::map<std::string,
std::string>& properties) {
+ // If es_hosts contains multiple hosts, prefer localhost for locality
+ auto it = properties.find("es_hosts");
+ if (it != properties.end() && !it->second.empty()) {
+ std::string localhost = BackendOptions::get_localhost();
+ std::string best;
+ std::istringstream stream(it->second);
+ std::string host;
+ while (std::getline(stream, host, ',')) {
+ if (best.empty()) {
+ best = host;
+ }
+ std::string hostname = extract_hostname(host);
+ if (hostname == localhost) {
+ return host;
+ }
+ }
+ if (!best.empty()) {
+ return best;
+ }
+ }
+ // Fallback to host_port
+ auto hp = properties.find(ESScanReader::KEY_HOST_PORT);
+ if (hp != properties.end()) {
+ return hp->second;
+ }
+ return "";
+}
+
} // namespace doris
diff --git a/be/src/format/table/es/es_scan_reader.h
b/be/src/format/table/es/es_scan_reader.h
index 1c8bd844f46..a980ac1beba 100644
--- a/be/src/format/table/es/es_scan_reader.h
+++ b/be/src/format/table/es/es_scan_reader.h
@@ -58,6 +58,9 @@ public:
// clear scroll context from elasticsearch
Status close();
+ static std::string extract_hostname(std::string host);
+ static std::string select_host(const std::map<std::string, std::string>&
properties);
+
private:
std::string _target;
std::string _user_name;
diff --git a/be/src/format/table/es/es_scroll_parser.cpp
b/be/src/format/table/es/es_scroll_parser.cpp
index 6f27c721a69..ce6e629bfa1 100644
--- a/be/src/format/table/es/es_scroll_parser.cpp
+++ b/be/src/format/table/es/es_scroll_parser.cpp
@@ -662,6 +662,17 @@ Status ScrollParser::fill_columns(const TupleDescriptor*
tuple_desc,
std::vector<MutableColumnPtr>& columns,
bool* line_eof,
const std::map<std::string, std::string>&
docvalue_context,
const cctz::time_zone& time_zone) {
+ return fill_columns(tuple_desc->slots(), columns, line_eof,
docvalue_context, time_zone);
+}
+
+// Keep identical ES-to-Doris type dispatch for the tuple and file-local
projection entry points.
+//
NOLINTNEXTLINE(readability-function-cognitive-complexity,readability-function-size)
+Status ScrollParser::fill_columns(const std::vector<SlotDescriptor*>& slots,
+ std::vector<MutableColumnPtr>& columns,
bool* line_eof,
+ const std::map<std::string, std::string>&
docvalue_context,
+ const cctz::time_zone& time_zone,
+ std::span<const size_t> column_positions) {
+ DORIS_CHECK(column_positions.empty() || column_positions.size() ==
slots.size());
*line_eof = true;
if (_size <= 0 || _line_index >= _size) {
@@ -681,9 +692,9 @@ Status ScrollParser::fill_columns(const TupleDescriptor*
tuple_desc,
line = &obj["fields"];
}
- for (int i = 0; i < tuple_desc->slots().size(); ++i) {
- const SlotDescriptor* slot_desc = tuple_desc->slots()[i];
- auto* col_ptr = columns[i].get();
+ for (size_t i = 0; i < slots.size(); ++i) {
+ const SlotDescriptor* slot_desc = slots[i];
+ auto* col_ptr = columns[column_positions.empty() ? i :
column_positions[i]].get();
if (slot_desc->col_name() == FIELD_ID) {
// actually this branch will not be reached, this is guaranteed by
Doris FE.
@@ -887,10 +898,10 @@ Status ScrollParser::fill_columns(const TupleDescriptor*
tuple_desc,
}
case TYPE_ARRAY: {
Array array;
- const auto& sub_type = assert_cast<const DataTypeArray*>(
-
remove_nullable(tuple_desc->slots()[i]->type()).get())
- ->get_nested_type()
- ->get_primitive_type();
+ const auto& sub_type =
+ assert_cast<const
DataTypeArray*>(remove_nullable(slot_desc->type()).get())
+ ->get_nested_type()
+ ->get_primitive_type();
RETURN_IF_ERROR(parse_column(col, sub_type, pure_doc_value, array,
time_zone));
col_ptr->insert(Field::create_field<TYPE_ARRAY>(array));
break;
diff --git a/be/src/format/table/es/es_scroll_parser.h
b/be/src/format/table/es/es_scroll_parser.h
index b605bc2bddd..e1f1106f624 100644
--- a/be/src/format/table/es/es_scroll_parser.h
+++ b/be/src/format/table/es/es_scroll_parser.h
@@ -20,6 +20,7 @@
#include <rapidjson/rapidjson.h>
#include <map>
+#include <span>
#include <string>
#include <vector>
@@ -30,6 +31,7 @@ namespace doris {
class Status;
class TupleDescriptor;
+class SlotDescriptor;
class ScrollParser {
public:
@@ -42,6 +44,13 @@ public:
bool* line_eof, const std::map<std::string,
std::string>& docvalue_context,
const cctz::time_zone& time_zone);
+ // Empty positions preserve tuple order; explicit positions address a
projected file-local block.
+ Status fill_columns(const std::vector<SlotDescriptor*>& slots,
+ std::vector<MutableColumnPtr>& columns, bool* line_eof,
+ const std::map<std::string, std::string>&
docvalue_context,
+ const cctz::time_zone& time_zone,
+ std::span<const size_t> column_positions = {});
+
const std::string& get_scroll_id();
int get_size() const;
diff --git a/be/src/format_v2/file_reader.h b/be/src/format_v2/file_reader.h
index 98258447a03..49559f4d66a 100644
--- a/be/src/format_v2/file_reader.h
+++ b/be/src/format_v2/file_reader.h
@@ -60,6 +60,7 @@ enum class FileFormat {
JNI,
ARROW,
WAL,
+ ES_HTTP,
};
struct FileScanRequest {
diff --git a/be/src/format_v2/materialized_reader_util.cpp
b/be/src/format_v2/materialized_reader_util.cpp
index d27f066b30b..ca2c4c86b43 100644
--- a/be/src/format_v2/materialized_reader_util.cpp
+++ b/be/src/format_v2/materialized_reader_util.cpp
@@ -21,7 +21,10 @@
#include "core/block/block.h"
#include "core/column/column_vector.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 "core/types.h"
#include "exprs/vexpr_context.h"
#include "format_v2/file_reader.h"
@@ -36,8 +39,52 @@ void update_counter(RuntimeProfile::Counter* counter,
int64_t value) {
}
}
+ColumnDefinition materialized_child_definition(const std::string& name,
DataTypePtr type,
+ int32_t local_id) {
+ ColumnDefinition child;
+ child.identifier = Field::create_field<TYPE_STRING>(name);
+ child.local_id = local_id;
+ child.name = name;
+ child.type = std::move(type);
+ child.children = materialized_schema_children(child.type);
+ return child;
+}
+
} // namespace
+std::vector<ColumnDefinition> materialized_schema_children(const DataTypePtr&
type) {
+ std::vector<ColumnDefinition> children;
+ DORIS_CHECK(type != nullptr);
+ const auto nested_type = remove_nullable(type);
+ switch (nested_type->get_primitive_type()) {
+ case TYPE_ARRAY: {
+ const auto* array_type = assert_cast<const
DataTypeArray*>(nested_type.get());
+ children.push_back(
+ materialized_child_definition("element",
array_type->get_nested_type(), 0));
+ break;
+ }
+ case TYPE_MAP: {
+ const auto* map_type = assert_cast<const
DataTypeMap*>(nested_type.get());
+ children.push_back(materialized_child_definition("key",
map_type->get_key_type(), 0));
+ children.push_back(materialized_child_definition("value",
map_type->get_value_type(), 1));
+ break;
+ }
+ case TYPE_STRUCT: {
+ const auto* struct_type = assert_cast<const
DataTypeStruct*>(nested_type.get());
+ children.reserve(struct_type->get_elements().size());
+ for (size_t idx = 0; idx < struct_type->get_elements().size(); ++idx) {
+
children.push_back(materialized_child_definition(struct_type->get_element_name(idx),
+
struct_type->get_element(idx),
+
cast_set<int32_t>(idx)));
+ }
+ break;
+ }
+ default:
+ break;
+ }
+ return children;
+}
+
ColumnPtr make_column_nullable_if_needed(ColumnPtr column, const DataTypePtr&
target_type) {
if (target_type != nullptr && target_type->is_nullable() && column.get()
!= nullptr &&
!column->is_nullable()) {
diff --git a/be/src/format_v2/materialized_reader_util.h
b/be/src/format_v2/materialized_reader_util.h
index 2fb1383dfb9..5f88ab061e7 100644
--- a/be/src/format_v2/materialized_reader_util.h
+++ b/be/src/format_v2/materialized_reader_util.h
@@ -18,6 +18,7 @@
#pragma once
#include <memory>
+#include <vector>
#include "common/status.h"
#include "core/column/column.h"
@@ -33,6 +34,10 @@ struct IOContext;
namespace format {
struct FileScanRequest;
+struct ColumnDefinition;
+
+// Describe children of already materialized ARRAY/MAP/STRUCT values in their
local schema.
+std::vector<ColumnDefinition> materialized_schema_children(const DataTypePtr&
type);
// Shared helpers for FileReader implementations that deserialize or build
already materialized
// Doris columns and then hand those columns to TableReader for final mapping.
diff --git a/be/src/format_v2/table/es_reader.cpp
b/be/src/format_v2/table/es_reader.cpp
new file mode 100644
index 00000000000..b259ba76463
--- /dev/null
+++ b/be/src/format_v2/table/es_reader.cpp
@@ -0,0 +1,295 @@
+// 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/table/es_reader.h"
+
+#include <utility>
+
+#include "core/block/block.h"
+#include "format/table/es/es_scan_reader.h"
+#include "format/table/es/es_scroll_parser.h"
+#include "format/table/es/es_scroll_query.h"
+#include "format_v2/column_mapper.h"
+#include "format_v2/materialized_reader_util.h"
+#include "runtime/descriptors.h"
+#include "runtime/file_scan_profile.h"
+#include "runtime/runtime_state.h"
+
+namespace doris::format::es {
+namespace {
+
+Status validate_es_range(const TFileRangeDesc& range) {
+ if (!range.__isset.table_format_params ||
range.table_format_params.table_format_type != "es" ||
+ !range.table_format_params.__isset.es_params) {
+ return Status::InvalidArgument("ES v2 reader requires es table format
and es_params");
+ }
+ return Status::OK();
+}
+
+class HttpScrollStream final : public EsScrollStream {
+public:
+ HttpScrollStream(const std::string& host, const std::map<std::string,
std::string>& properties,
+ bool docvalues)
+ : _reader(host, properties, docvalues) {}
+ Status open() override { return _reader.open(); }
+ Status next(bool* eof, std::unique_ptr<ScrollParser>& parser) override {
+ return _reader.get_next(eof, parser);
+ }
+ Status close() override { return _reader.close(); }
+
+private:
+ ESScanReader _reader;
+};
+
+} // namespace
+
+EsFileReader::EsFileReader(std::shared_ptr<io::FileSystemProperties>&
system_properties,
+ std::unique_ptr<io::FileDescription>&
file_description,
+ std::shared_ptr<io::IOContext> io_ctx,
RuntimeProfile* profile,
+ TFileRangeDesc range, const TFileScanRangeParams&
params,
+ std::vector<SlotDescriptor*> file_slot_descs,
+ EsScrollStreamFactory stream_factory)
+ : FileReader(system_properties, file_description, std::move(io_ctx),
profile),
+ _range(std::move(range)),
+ _params(params),
+ _file_slots(std::move(file_slot_descs)),
+ _stream_factory(std::move(stream_factory)) {}
+
+EsFileReader::~EsFileReader() {
+ WARN_IF_ERROR(close(), "Failed to close ES scroll stream");
+}
+
+void EsFileReader::_init_profile() {
+ if (_profile == nullptr) {
+ return;
+ }
+ _io_time = file_scan_profile::ensure_hierarchy(_profile).io;
+ const char* parent = "EsFileReader";
+ _total_time = ADD_CHILD_TIMER_WITH_LEVEL(_profile, parent,
file_scan_profile::FILE_READER, 1);
+ _read_time = ADD_CHILD_TIMER_WITH_LEVEL(_profile, "EsReadTime", parent, 1);
+ _materialize_time = ADD_CHILD_TIMER_WITH_LEVEL(_profile,
"EsMaterializeTime", parent, 1);
+ _filter_time = ADD_CHILD_TIMER_WITH_LEVEL(_profile, "EsFilterTime",
parent, 1);
+ _batches_read = ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "EsBatchesRead",
TUnit::UNIT, parent, 1);
+ _rows_read = ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "EsRowsRead",
TUnit::UNIT, parent, 1);
+}
+
+Status EsFileReader::init(RuntimeState* state) {
+ DORIS_CHECK(state != nullptr);
+ _state = state;
+ set_batch_size(state->batch_size());
+ _init_profile();
+ RETURN_IF_ERROR(validate_es_range(_range));
+ _eof = false;
+ return Status::OK();
+}
+
+Status EsFileReader::get_schema(std::vector<ColumnDefinition>* schema) const {
+ DORIS_CHECK(schema != nullptr);
+ schema->clear();
+ schema->reserve(_file_slots.size());
+ for (size_t i = 0; i < _file_slots.size(); ++i) {
+ const auto* slot = _file_slots[i];
+ schema->push_back({.identifier =
Field::create_field<TYPE_INT>(cast_set<int32_t>(i)),
+ .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> EsFileReader::create_column_mapper(
+ TableColumnMapperOptions options) const {
+ return std::make_unique<MaterializedColumnMapper>(options);
+}
+
+void EsFileReader::set_batch_size(size_t batch_size) {
+ DORIS_CHECK(batch_size > 0);
+ _batch_size = batch_size;
+}
+
+Status EsFileReader::open(std::shared_ptr<FileScanRequest> request) {
+ SCOPED_TIMER(_total_time);
+ RETURN_IF_ERROR(FileReader::open(std::move(request)));
+ auto status = _open_stream();
+ if (!status.ok()) {
+ WARN_IF_ERROR(close(), "Failed to clear ES scroll after open error");
+ }
+ return status;
+}
+
+Status EsFileReader::_open_stream() {
+ DORIS_CHECK(_stream == nullptr);
+ RETURN_IF_CANCELLED(_state);
+ if (_io_ctx != nullptr && _io_ctx->should_stop) {
+ _eof = true;
+ return Status::OK();
+ }
+ std::vector<std::string> names;
+ for (const auto& [id, position] : _request->local_positions) {
+ DORIS_CHECK(id.value() >= 0 && id.value() < _file_slots.size());
+ _read_slots.push_back(_file_slots[id.value()]);
+ _read_positions.push_back(position.value());
+ names.push_back(_file_slots[id.value()]->col_name());
+ }
+ auto properties = _params.es_properties;
+ for (const auto& [key, value] : _range.table_format_params.es_params) {
+ properties[key] = value;
+ }
+ properties[ESScanReader::KEY_BATCH_SIZE] = std::to_string(_batch_size);
+ _docvalue_context = _params.es_docvalue_context;
+ // A source-only field such as _id cannot use doc values, even when other
requested fields can.
+ // The mapper may retain hidden predicate columns beyond the FE's original
output projection.
+ for (const auto& name : names) {
+ if (name == "_id" || !_docvalue_context.contains(name)) {
+ properties[ESScanReader::KEY_DOC_VALUES_MODE] = "0";
+ break;
+ }
+ }
+ const auto host = ESScanReader::select_host(properties);
+ for (const auto* key :
+ {ESScanReader::KEY_INDEX, ESScanReader::KEY_SHARD,
ESScanReader::KEY_QUERY_DSL}) {
+ if (!properties.contains(key) || properties.at(key).empty()) {
+ return Status::InvalidArgument("ES v2 reader requires property
{}", key);
+ }
+ }
+ if (host.empty()) {
+ return Status::InvalidArgument("ES v2 reader requires a host");
+ }
+ properties[ESScanReader::KEY_HOST_PORT] = host;
+ bool docvalues = false;
+ properties[ESScanReader::KEY_QUERY] =
+ ESScrollQueryBuilder::build(properties, names, _docvalue_context,
&docvalues);
+ _stream = _stream_factory ? _stream_factory(host, properties, docvalues)
+ : std::make_unique<HttpScrollStream>(host,
properties, docvalues);
+ DORIS_CHECK(_stream != nullptr);
+ {
+ SCOPED_TIMER(_io_time);
+ SCOPED_TIMER(_read_time);
+ RETURN_IF_ERROR(_stream->open());
+ }
+ // ESScanReader caches the first response in open(). Parse it now to
capture the scroll ID so
+ // LIMIT/cancellation before the first block can still clear the
server-side search context.
+ RETURN_IF_ERROR(_next_page());
+ _eof = false;
+ return Status::OK();
+}
+
+Status EsFileReader::_next_page() {
+ SCOPED_TIMER(_io_time);
+ SCOPED_TIMER(_read_time);
+ RETURN_IF_ERROR(_stream->next(&_source_eof, _parser));
+ _page_rows_remaining = _source_eof ? 0 : _parser->get_size();
+ if (!_source_eof && _batches_read != nullptr) {
+ COUNTER_UPDATE(_batches_read, 1);
+ }
+ return Status::OK();
+}
+
+Status EsFileReader::get_block(Block* block, size_t* rows, bool* eof) {
+ SCOPED_TIMER(_total_time);
+ DORIS_CHECK(block != nullptr && rows != nullptr && eof != nullptr);
+ *rows = 0;
+ *eof = false;
+ if (_eof) {
+ *eof = true;
+ return Status::OK();
+ }
+ auto status = _read_block(block, rows, eof);
+ if (!status.ok()) {
+ WARN_IF_ERROR(close(), "Failed to clear ES scroll after read error");
+ return status;
+ }
+ if (*eof) {
+ return close();
+ }
+ return Status::OK();
+}
+
+Status EsFileReader::_read_block(Block* block, size_t* rows, bool* eof) {
+ {
+ auto guard = block->mutate_columns_scoped();
+ auto& columns = guard.mutable_columns();
+ while (*rows < _batch_size) {
+ RETURN_IF_CANCELLED(_state);
+ if (_io_ctx != nullptr && _io_ctx->should_stop) {
+ *eof = true;
+ return Status::OK();
+ }
+ if (_page_rows_remaining == 0) {
+ if (_source_eof) {
+ break;
+ }
+ RETURN_IF_ERROR(_next_page());
+ if (_page_rows_remaining == 0) {
+ continue;
+ }
+ }
+ SCOPED_TIMER(_materialize_time);
+ bool line_eof = false;
+ // The parser sees physical slots only; the request owns their
file-local block order.
+ RETURN_IF_ERROR(_parser->fill_columns(_read_slots, columns,
&line_eof,
+ _docvalue_context,
_state->timezone_obj(),
+ _read_positions));
+ DORIS_CHECK(!line_eof);
+ --_page_rows_remaining;
+ ++*rows;
+ }
+ }
+ _record_scan_rows(cast_set<int64_t>(*rows));
+ if (_rows_read != nullptr) {
+ COUNTER_UPDATE(_rows_read, *rows);
+ }
+ *eof = _source_eof && *rows == 0;
+ SCOPED_TIMER(_filter_time);
+ return apply_materialized_reader_filters(_request.get(), _io_ctx.get(),
block, rows);
+}
+
+Status EsFileReader::close() {
+ auto stream = std::move(_stream);
+ _parser.reset();
+ _request.reset();
+ _read_slots.clear();
+ _read_positions.clear();
+ _page_rows_remaining = 0;
+ _source_eof = true;
+ _eof = true;
+ // Release ownership before closing so a failed DELETE is reported once,
including destruction.
+ return stream != nullptr ? stream->close() : Status::OK();
+}
+
+EsReader::EsReader(EsScrollStreamFactory stream_factory)
+ : _stream_factory(std::move(stream_factory)) {}
+
+Status EsReader::init(TableReadOptions&& options) {
+ DORIS_CHECK(options.file_slot_descs != nullptr && options.scan_params !=
nullptr);
+ return TableReader::init(std::move(options));
+}
+
+Status EsReader::prepare_split(const SplitReadOptions& options) {
+ RETURN_IF_ERROR(validate_es_range(options.current_range));
+ return TableReader::prepare_split(options);
+}
+
+Status EsReader::create_file_reader(std::unique_ptr<FileReader>* reader) {
+ *reader = std::make_unique<EsFileReader>(_system_properties,
_current_task->data_file, _io_ctx,
+ _scanner_profile,
_current_file_range_desc,
+ *_scan_params, *_file_slot_descs,
_stream_factory);
+ return Status::OK();
+}
+
+} // namespace doris::format::es
diff --git a/be/src/format_v2/table/es_reader.h
b/be/src/format_v2/table/es_reader.h
new file mode 100644
index 00000000000..fea1869ff02
--- /dev/null
+++ b/be/src/format_v2/table/es_reader.h
@@ -0,0 +1,107 @@
+// 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 <functional>
+#include <map>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "format_v2/file_reader.h"
+#include "format_v2/table_reader.h"
+
+namespace doris {
+class ScrollParser;
+class SlotDescriptor;
+} // namespace doris
+
+namespace doris::format::es {
+
+// The transport boundary allows reader tests to use real ES responses without
a running server.
+class EsScrollStream {
+public:
+ virtual ~EsScrollStream() = default;
+ virtual Status open() = 0;
+ virtual Status next(bool* eof, std::unique_ptr<ScrollParser>& parser) = 0;
+ virtual Status close() = 0;
+};
+using EsScrollStreamFactory = std::function<std::unique_ptr<EsScrollStream>(
+ const std::string&, const std::map<std::string, std::string>&, bool)>;
+
+class EsFileReader final : public FileReader {
+public:
+ EsFileReader(std::shared_ptr<io::FileSystemProperties>& system_properties,
+ std::unique_ptr<io::FileDescription>& file_description,
+ std::shared_ptr<io::IOContext> io_ctx, RuntimeProfile*
profile,
+ TFileRangeDesc range, const TFileScanRangeParams& params,
+ std::vector<SlotDescriptor*> file_slot_descs,
+ EsScrollStreamFactory stream_factory = {});
+ ~EsFileReader() 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;
+ void set_batch_size(size_t batch_size) 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 _open_stream();
+ Status _next_page();
+ Status _read_block(Block* block, size_t* rows, bool* eof);
+
+ const TFileRangeDesc _range;
+ const TFileScanRangeParams& _params;
+ const std::vector<SlotDescriptor*> _file_slots;
+ EsScrollStreamFactory _stream_factory;
+ RuntimeState* _state = nullptr;
+ size_t _batch_size = 1024;
+ std::unique_ptr<EsScrollStream> _stream;
+ std::unique_ptr<ScrollParser> _parser;
+ std::vector<SlotDescriptor*> _read_slots;
+ std::vector<size_t> _read_positions;
+ std::map<std::string, std::string> _docvalue_context;
+ size_t _page_rows_remaining = 0;
+ bool _source_eof = false;
+ RuntimeProfile::Counter* _total_time = nullptr;
+ RuntimeProfile::Counter* _read_time = nullptr;
+ RuntimeProfile::Counter* _materialize_time = nullptr;
+ RuntimeProfile::Counter* _filter_time = nullptr;
+ RuntimeProfile::Counter* _io_time = nullptr;
+ RuntimeProfile::Counter* _batches_read = nullptr;
+ RuntimeProfile::Counter* _rows_read = nullptr;
+};
+
+class EsReader final : public TableReader {
+public:
+ explicit EsReader(EsScrollStreamFactory stream_factory = {});
+ Status init(TableReadOptions&& options) override;
+ Status prepare_split(const SplitReadOptions& options) override;
+
+protected:
+ Status create_file_reader(std::unique_ptr<FileReader>* reader) override;
+
+private:
+ EsScrollStreamFactory _stream_factory;
+};
+
+} // namespace doris::format::es
diff --git a/be/src/format_v2/table/iceberg_reader.cpp
b/be/src/format_v2/table/iceberg_reader.cpp
index 151744480ee..a879a0d1f89 100644
--- a/be/src/format_v2/table/iceberg_reader.cpp
+++ b/be/src/format_v2/table/iceberg_reader.cpp
@@ -154,6 +154,8 @@ const char* file_format_name(FileFormat format) {
return "JNI";
case FileFormat::ARROW:
return "ARROW";
+ case FileFormat::ES_HTTP:
+ return "ES_HTTP";
case FileFormat::WAL:
return "WAL";
}
diff --git a/be/src/format_v2/table/remote_doris_reader.cpp
b/be/src/format_v2/table/remote_doris_reader.cpp
index bd2c08b5f23..c119275fc69 100644
--- a/be/src/format_v2/table/remote_doris_reader.cpp
+++ b/be/src/format_v2/table/remote_doris_reader.cpp
@@ -34,10 +34,7 @@
#include "core/assert_cast.h"
#include "core/block/block.h"
#include "core/data_type/data_type.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 "core/data_type_serde/data_type_serde.h"
#include "format/arrow/arrow_utils.h"
#include "format_v2/materialized_reader_util.h"
@@ -271,53 +268,6 @@ Status create_flight_stream(const TFileRangeDesc& range,
std::shared_ptr<io::IOC
return Status::OK();
}
-ColumnDefinition remote_doris_child_definition(const std::string& name,
DataTypePtr type,
- int32_t local_id);
-
-std::vector<ColumnDefinition> synthesize_remote_doris_children(const
DataTypePtr& type) {
- std::vector<ColumnDefinition> children;
- DORIS_CHECK(type != nullptr);
- const auto nested_type = remove_nullable(type);
- switch (nested_type->get_primitive_type()) {
- case TYPE_ARRAY: {
- const auto* array_type = assert_cast<const
DataTypeArray*>(nested_type.get());
- children.push_back(
- remote_doris_child_definition("element",
array_type->get_nested_type(), 0));
- break;
- }
- case TYPE_MAP: {
- const auto* map_type = assert_cast<const
DataTypeMap*>(nested_type.get());
- children.push_back(remote_doris_child_definition("key",
map_type->get_key_type(), 0));
- children.push_back(remote_doris_child_definition("value",
map_type->get_value_type(), 1));
- break;
- }
- case TYPE_STRUCT: {
- const auto* struct_type = assert_cast<const
DataTypeStruct*>(nested_type.get());
- children.reserve(struct_type->get_elements().size());
- for (size_t idx = 0; idx < struct_type->get_elements().size(); ++idx) {
-
children.push_back(remote_doris_child_definition(struct_type->get_element_name(idx),
-
struct_type->get_element(idx),
-
cast_set<int32_t>(idx)));
- }
- break;
- }
- default:
- break;
- }
- return children;
-}
-
-ColumnDefinition remote_doris_child_definition(const std::string& name,
DataTypePtr type,
- int32_t local_id) {
- ColumnDefinition child;
- child.identifier = Field::create_field<TYPE_STRING>(name);
- child.local_id = local_id;
- child.name = name;
- child.type = std::move(type);
- child.children = synthesize_remote_doris_children(child.type);
- return child;
-}
-
} // namespace
RemoteDorisFileReader::RemoteDorisFileReader(
@@ -383,7 +333,7 @@ Status
RemoteDorisFileReader::get_schema(std::vector<ColumnDefinition>* file_sch
.type = slot->type(),
// Remote Doris exposes table slots as file columns. Complex
columns still need
// structural children so TableColumnMapper can validate and
project them.
- .children = synthesize_remote_doris_children(slot->type()),
+ .children = materialized_schema_children(slot->type()),
});
}
return Status::OK();
diff --git a/be/src/format_v2/table_reader.cpp
b/be/src/format_v2/table_reader.cpp
index 5a8d5dea5e7..146c7f2d9f2 100644
--- a/be/src/format_v2/table_reader.cpp
+++ b/be/src/format_v2/table_reader.cpp
@@ -109,6 +109,8 @@ std::string file_format_to_string(FileFormat format) {
return "JNI";
case FileFormat::ARROW:
return "ARROW";
+ case FileFormat::ES_HTTP:
+ return "ES_HTTP";
case FileFormat::WAL:
return "WAL";
}
diff --git a/be/test/exec/scan/file_scanner_v2_test.cpp
b/be/test/exec/scan/file_scanner_v2_test.cpp
index b38caa0a7bc..c6c5c680510 100644
--- a/be/test/exec/scan/file_scanner_v2_test.cpp
+++ b/be/test/exec/scan/file_scanner_v2_test.cpp
@@ -369,6 +369,7 @@ TEST(FileScannerV2Test, SupportedFormatMatrix) {
{"hive", TFileFormatType::FORMAT_ARROW, std::nullopt, false},
{"", TFileFormatType::FORMAT_ARROW, std::nullopt, false},
{"", TFileFormatType::FORMAT_WAL, std::nullopt, true},
+ {"es", TFileFormatType::FORMAT_ES_HTTP, std::nullopt, true},
{"", TFileFormatType::FORMAT_ES_HTTP, std::nullopt, false},
{"", TFileFormatType::FORMAT_LANCE, std::nullopt, false},
};
@@ -478,8 +479,14 @@ TEST(FileScannerV2Test,
FileScanLocalStateSelectsV2ForSupportedQueriesAndLoads)
params.__set_format_type(TFileFormatType::FORMAT_JNI);
EXPECT_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
false, params));
- const std::vector<TFileFormatType::type> unsupported_formats
{TFileFormatType::FORMAT_ES_HTTP,
-
TFileFormatType::FORMAT_LANCE};
+ params.__set_format_type(TFileFormatType::FORMAT_ES_HTTP);
+
EXPECT_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
false, params));
+
EXPECT_FALSE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
true, params));
+ query_options.__set_enable_file_scanner_v2(false);
+
EXPECT_FALSE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
false, params));
+ query_options.__set_enable_file_scanner_v2(true);
+
+ const std::vector<TFileFormatType::type> unsupported_formats
{TFileFormatType::FORMAT_LANCE};
for (const auto format : unsupported_formats) {
params.__set_format_type(format);
EXPECT_FALSE(
diff --git a/be/test/format_v2/table/es_reader_test.cpp
b/be/test/format_v2/table/es_reader_test.cpp
new file mode 100644
index 00000000000..47ec3eca22d
--- /dev/null
+++ b/be/test/format_v2/table/es_reader_test.cpp
@@ -0,0 +1,540 @@
+// 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/table/es_reader.h"
+
+#include <gtest/gtest.h>
+#include <rapidjson/document.h>
+
+#include <atomic>
+#include <tuple>
+#include <utility>
+
+#include "common/object_pool.h"
+#include "core/assert_cast.h"
+#include "core/block/block.h"
+#include "core/column/column_array.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_string.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_string.h"
+#include "exprs/vexpr.h"
+#include "exprs/vexpr_context.h"
+#include "format/table/es/es_scroll_parser.h"
+#include "format_v2/column_mapper.h"
+#include "runtime/runtime_state.h"
+#include "service/http/ev_http_server.h"
+#include "service/http/http_channel.h"
+#include "service/http/http_handler.h"
+#include "service/http/http_request.h"
+#include "testutil/desc_tbl_builder.h"
+
+namespace doris::format::es {
+namespace {
+struct StreamState {
+ int opens = 0;
+ int reads = 0;
+ int closes = 0;
+ bool fail_open = false;
+ bool fail_next = false;
+ bool fail_close = false;
+ bool docvalues = false;
+ std::string host;
+ std::map<std::string, std::string> properties;
+ std::vector<std::string> pages;
+};
+class ResponseStream final : public EsScrollStream {
+public:
+ explicit ResponseStream(std::shared_ptr<StreamState> state) :
_state(std::move(state)) {}
+ Status open() override {
+ ++_state->opens;
+ return _state->fail_open ? Status::IOError("open failed") :
Status::OK();
+ }
+ Status next(bool* eof, std::unique_ptr<ScrollParser>& parser) override {
+ ++_state->reads;
+ if (_state->fail_next) {
+ return Status::IOError("scroll failed");
+ }
+ *eof = _page == _state->pages.size();
+ if (!*eof) {
+ parser = std::make_unique<ScrollParser>(_state->docvalues);
+ RETURN_IF_ERROR(parser->parse(_state->pages[_page++]));
+ }
+ return Status::OK();
+ }
+ Status close() override {
+ ++_state->closes;
+ return _state->fail_close ? Status::IOError("close failed") :
Status::OK();
+ }
+
+private:
+ std::shared_ptr<StreamState> _state;
+ size_t _page = 0;
+};
+
+std::string page(const std::string& hits) {
+ return R"({"_scroll_id":"scroll-token","hits":{"hits":[)" + hits + "]}}";
+}
+TFileRangeDesc es_range() {
+ TTableFormatFileDesc table;
+ table.__set_table_format_type("es");
+ table.__set_es_params(
+ {{"index", "example"}, {"shard_id", "0"}, {"host_port",
"http://localhost:9200"}});
+ TFileRangeDesc range;
+ range.__set_format_type(TFileFormatType::FORMAT_ES_HTTP);
+ range.__set_path("es://example/0");
+ range.__set_table_format_params(table);
+ return range;
+}
+EsScrollStreamFactory factory(const std::shared_ptr<StreamState>& state) {
+ return [state](const auto& host, const auto& properties, bool docvalues) {
+ state->host = host;
+ state->properties = properties;
+ state->docvalues = docvalues;
+ return std::make_unique<ResponseStream>(state);
+ };
+}
+class EsV2ReaderTest : public testing::Test {
+protected:
+ void SetUp() override {
+ DescriptorTblBuilder builder(&_pool);
+ builder.declare_tuple()
+ << std::make_tuple(std::make_shared<DataTypeInt32>(),
std::string("id"))
+ << std::make_tuple(std::make_shared<DataTypeString>(),
std::string("name"))
+ << std::make_tuple(std::make_shared<DataTypeString>(),
std::string("_id"))
+ << std::make_tuple(std::make_shared<DataTypeArray>(
+
make_nullable(std::make_shared<DataTypeInt32>())),
+ std::string("values"));
+ _slots = builder.build()->get_tuple_descriptor(0)->slots();
+ _params.__set_es_properties({{"query_dsl", R"({"match_all":{}})"}});
+ _responses = std::make_shared<StreamState>();
+ _responses->pages = {
+
page(R"({"_id":"a","_source":{"id":1,"name":"one","values":[1,2,3]}},
+
{"_id":"b","_source":{"id":2,"name":null,"values":[]}})"),
+
page(R"({"_id":"c","_source":{"id":3,"name":"three","values":null}})")};
+ }
+ std::unique_ptr<EsFileReader> reader() {
+ auto props = std::make_shared<io::FileSystemProperties>();
+ auto file = std::make_unique<io::FileDescription>();
+ file->path = "es://example/0";
+ auto result = std::make_unique<EsFileReader>(props, file, _io,
&_profile, _range, _params,
+ _slots,
factory(_responses));
+ EXPECT_TRUE(result->init(&_state).ok());
+ return result;
+ }
+ std::shared_ptr<FileScanRequest> request(std::initializer_list<int> ids) {
+ auto result = std::make_shared<FileScanRequest>();
+ FileScanRequestBuilder builder(result.get());
+ for (int id : ids) {
+
EXPECT_TRUE(builder.add_non_predicate_column(LocalColumnId(id)).ok());
+ }
+ return result;
+ }
+ Block block(const EsFileReader& reader, const FileScanRequest& request) {
+ std::vector<ColumnDefinition> schema;
+ EXPECT_TRUE(reader.get_schema(&schema).ok());
+ ColumnsWithTypeAndName columns(request.block_column_count());
+ for (auto [id, position] : request.local_positions) {
+ const auto& column = schema[id.value()];
+ columns[position.value()] = {column.type->create_column(),
column.type, column.name};
+ }
+ return Block(columns);
+ }
+ int32_t int_at(const Block& block, size_t pos, size_t row) {
+ const auto& nullable =
+ assert_cast<const
ColumnNullable&>(*block.get_by_position(pos).column);
+ return assert_cast<const
ColumnInt32&>(nullable.get_nested_column()).get_data()[row];
+ }
+ std::string string_at(const Block& block, size_t pos, size_t row) {
+ const auto& nullable =
+ assert_cast<const
ColumnNullable&>(*block.get_by_position(pos).column);
+ return nullable.get_nested_column().get_data_at(row).to_string();
+ }
+ ObjectPool _pool;
+ RuntimeState _state;
+ RuntimeProfile _profile {"es_v2_test"};
+ std::shared_ptr<io::IOContext> _io = std::make_shared<io::IOContext>();
+ TFileRangeDesc _range = es_range();
+ TFileScanRangeParams _params;
+ std::vector<SlotDescriptor*> _slots;
+ std::shared_ptr<StreamState> _responses;
+};
+class NullableIntGreaterThanExpr final : public VExpr {
+public:
+ NullableIntGreaterThanExpr(size_t block_position, int32_t value)
+ : VExpr(std::make_shared<DataTypeUInt8>(), false),
+ _block_position(block_position),
+ _value(value) {}
+
+ const std::string& expr_name() const override { return _name; }
+
+ bool is_constant() const override { return false; }
+
+ Status execute_column_impl(VExprContext*, const Block* block, const
Selector* selector,
+ size_t count, ColumnPtr& result_column) const
override {
+ DORIS_CHECK(block != nullptr);
+ const auto& nullable =
+ assert_cast<const
ColumnNullable&>(*block->get_by_position(_block_position).column);
+ const auto& data = assert_cast<const
ColumnInt32&>(nullable.get_nested_column());
+
+ auto result = ColumnUInt8::create();
+ auto& result_data = result->get_data();
+ result_data.resize(count);
+ for (size_t row = 0; row < count; ++row) {
+ const auto source_row = selector == nullptr ? row :
(*selector)[row];
+ result_data[row] =
+ !nullable.is_null_at(source_row) &&
data.get_element(source_row) > _value;
+ }
+ result_column = std::move(result);
+ return Status::OK();
+ }
+
+ Status clone_node(VExprSPtr* cloned_expr) const override {
+ DORIS_CHECK(cloned_expr != nullptr);
+ *cloned_expr =
std::make_shared<NullableIntGreaterThanExpr>(_block_position, _value);
+ return Status::OK();
+ }
+
+private:
+ size_t _block_position;
+ int32_t _value;
+ const std::string _name = "NullableIntGreaterThanExpr";
+};
+
+VExprContextSPtr prepared_conjunct(RuntimeState* state, const VExprSPtr& expr)
{
+ auto context = VExprContext::create_shared(expr);
+ auto status = context->prepare(state, RowDescriptor());
+ EXPECT_TRUE(status.ok()) << status;
+ status = context->open(state);
+ EXPECT_TRUE(status.ok()) << status;
+ return context;
+}
+
+} // namespace
+
+TEST_F(EsV2ReaderTest, ProjectsLocalColumnsAndCarriesPagesAcrossBatches) {
+ auto scan = reader();
+ std::vector<ColumnDefinition> schema;
+ ASSERT_TRUE(scan->get_schema(&schema).ok());
+ ASSERT_EQ(schema.size(), 4);
+ ASSERT_EQ(schema[3].children.size(), 1);
+ EXPECT_EQ(_responses->opens, 0);
+ auto projection = request({1, 0, 2, 3});
+ ASSERT_TRUE(scan->open(projection).ok());
+ EXPECT_EQ(_responses->reads, 1);
+ scan->set_batch_size(1);
+ size_t rows = 0;
+ bool eof = false;
+ for (size_t row = 0; row < 3; ++row) {
+ auto output = block(*scan, *projection);
+ auto status = scan->get_block(&output, &rows, &eof);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(rows, 1);
+ EXPECT_FALSE(eof);
+ EXPECT_EQ(int_at(output, 1, 0), row + 1);
+ EXPECT_EQ(string_at(output, 2, 0), std::string(1, 'a' + row));
+ if (row == 1) {
+ EXPECT_TRUE(output.get_by_position(0).column->is_null_at(0));
+ } else {
+ EXPECT_EQ(string_at(output, 0, 0), row == 0 ? "one" : "three");
+ }
+ if (row == 0) {
+ const auto& values = assert_cast<const ColumnArray&>(
+ assert_cast<const
ColumnNullable&>(*output.get_by_position(3).column)
+ .get_nested_column());
+ EXPECT_EQ(values.get_offsets()[0], 3);
+ EXPECT_FALSE(values.get_data().is_null_at(1));
+ }
+ }
+ auto output = block(*scan, *projection);
+ ASSERT_TRUE(scan->get_block(&output, &rows, &eof).ok());
+ EXPECT_EQ(rows, 0);
+ EXPECT_TRUE(eof);
+ EXPECT_EQ(_responses->closes, 1);
+ ASSERT_TRUE(scan->close().ok());
+ scan.reset();
+ EXPECT_EQ(_responses->closes, 1);
+}
+
+TEST_F(EsV2ReaderTest, UsesDocvalueAliasesAndRangeOverrides) {
+ _params.es_properties["index"] = "node-index";
+ _params.es_properties["doc_values_mode"] = "1";
+ _params.__set_es_docvalue_context({{"id", "id"}, {"name",
"name.keyword"}});
+ _responses->pages =
{page(R"({"fields":{"id":[4],"name.keyword":["four"]}})")};
+ auto scan = reader();
+ auto projection = request({1, 0});
+ ASSERT_TRUE(scan->open(projection).ok());
+ EXPECT_TRUE(_responses->docvalues);
+ EXPECT_EQ(_responses->properties.at("index"), "example");
+ EXPECT_NE(_responses->properties.at("query").find("name.keyword"),
std::string::npos);
+ auto output = block(*scan, *projection);
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(scan->get_block(&output, &rows, &eof).ok());
+ ASSERT_EQ(rows, 1);
+ EXPECT_EQ(int_at(output, 1, 0), 4);
+ EXPECT_EQ(string_at(output, 0, 0), "four");
+}
+
+TEST_F(EsV2ReaderTest, CountsRowsWithNoProjectedColumns) {
+ auto scan = reader();
+ auto projection = request({});
+ ASSERT_TRUE(scan->open(projection).ok());
+ Block output;
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(scan->get_block(&output, &rows, &eof).ok());
+ EXPECT_EQ(rows, 3);
+ ASSERT_TRUE(scan->get_block(&output, &rows, &eof).ok());
+ EXPECT_TRUE(eof);
+ EXPECT_EQ(rows, 0);
+}
+
+TEST_F(EsV2ReaderTest, AppliesResidualFiltersAndAccountsForFilteredRows) {
+ auto scan = reader();
+ auto projection = request({1, 0});
+ projection->conjuncts.push_back(
+ prepared_conjunct(&_state,
std::make_shared<NullableIntGreaterThanExpr>(1, 1)));
+ ASSERT_TRUE(scan->open(projection).ok());
+ auto output = block(*scan, *projection);
+ size_t rows = 0;
+ bool eof = false;
+ auto status = scan->get_block(&output, &rows, &eof);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(rows, 2);
+ EXPECT_EQ(int_at(output, 1, 0), 2);
+ EXPECT_EQ(int_at(output, 1, 1), 3);
+ EXPECT_EQ(_io->predicate_filtered_rows, 1);
+}
+
+TEST_F(EsV2ReaderTest, ClosesAfterOpenFailureAndBeforeFirstBlock) {
+ for (int mode = 0; mode < 3; ++mode) {
+ _responses = std::make_shared<StreamState>();
+ _responses->pages = {page("")};
+ _responses->fail_open = mode == 0;
+ _responses->fail_next = mode == 1;
+ auto scan = reader();
+ auto status = scan->open(request({0}));
+ EXPECT_EQ(status.ok(), mode == 2);
+ if (mode != 2) {
+ EXPECT_EQ(_responses->closes, 1);
+ }
+ ASSERT_TRUE(scan->close().ok());
+ scan.reset();
+ EXPECT_EQ(_responses->closes, 1);
+ }
+}
+
+TEST_F(EsV2ReaderTest, CancellationAndReadErrorsReleaseStream) {
+ for (bool fail_next : {false, true}) {
+ _io->should_stop = false;
+ _responses = std::make_shared<StreamState>();
+ _responses->pages = {page(R"({"_source":{"id":1}})")};
+ auto scan = reader();
+ auto projection = request({0});
+ ASSERT_TRUE(scan->open(projection).ok());
+ _responses->fail_next = fail_next;
+ _io->should_stop = !fail_next;
+ auto output = block(*scan, *projection);
+ size_t rows = 0;
+ bool eof = false;
+ auto status = scan->get_block(&output, &rows, &eof);
+ EXPECT_EQ(status.ok(), !fail_next);
+ EXPECT_EQ(_responses->closes, 1);
+ ASSERT_TRUE(scan->close().ok());
+ EXPECT_EQ(_responses->closes, 1);
+ }
+}
+
+TEST_F(EsV2ReaderTest, CloseFailureIsReportedOnce) {
+ auto scan = reader();
+ ASSERT_TRUE(scan->open(request({0})).ok());
+ _responses->fail_close = true;
+ EXPECT_FALSE(scan->close().ok());
+ ASSERT_TRUE(scan->close().ok());
+ scan.reset();
+ EXPECT_EQ(_responses->closes, 1);
+}
+
+TEST_F(EsV2ReaderTest, RejectsMissingRangePropertiesAndRowidFetch) {
+ _range.table_format_params.__isset.es_params = false;
+ auto props = std::make_shared<io::FileSystemProperties>();
+ auto file = std::make_unique<io::FileDescription>();
+ EsFileReader invalid(props, file, _io, &_profile, _range, _params, _slots,
factory(_responses));
+ EXPECT_FALSE(invalid.init(&_state).ok());
+ _range = es_range();
+ auto scan = reader();
+ auto projection = request({0});
+ projection->row_ids = std::vector<int64_t> {0};
+ EXPECT_FALSE(scan->open(projection).ok());
+ EXPECT_EQ(_responses->opens, 0);
+}
+
+TEST_F(EsV2ReaderTest, TableReaderMapsProjectionAndAdvancesSplits) {
+ EsReader table(factory(_responses));
+ std::vector<ColumnDefinition> projected;
+ for (int id : {1, 0}) {
+ ColumnDefinition column;
+ column.identifier =
Field::create_field<TYPE_STRING>(_slots[id]->col_name());
+ column.name = _slots[id]->col_name();
+ column.type = _slots[id]->type();
+ projected.push_back(std::move(column));
+ }
+ ColumnDefinition missing;
+ missing.name = "missing";
+ missing.type = make_nullable(std::make_shared<DataTypeString>());
+ projected.push_back(std::move(missing));
+ ASSERT_TRUE(table.init({.projected_columns = projected,
+ .conjuncts = {},
+ .format = FileFormat::ES_HTTP,
+ .scan_params = &_params,
+ .io_ctx = _io,
+ .runtime_state = &_state,
+ .scanner_profile = &_profile,
+ .file_slot_descs = &_slots})
+ .ok());
+ for (int split = 0; split < 2; ++split) {
+ SplitReadOptions options;
+ options.current_range = _range;
+ options.current_split_format = FileFormat::ES_HTTP;
+ ASSERT_TRUE(table.prepare_split(options).ok());
+ Block output;
+ for (const auto& column : projected) {
+ output.insert({column.type->create_column(), column.type,
column.name});
+ }
+ bool eof = false;
+ size_t total = 0;
+ while (!eof) {
+ const auto status = table.get_block(&output, &eof);
+ ASSERT_TRUE(status.ok()) << status;
+ if (output.rows() > 0) {
+ ASSERT_EQ(output.columns(), 3);
+ EXPECT_EQ(string_at(output, 0, 0), "one");
+ EXPECT_EQ(int_at(output, 1, 0), 1);
+ EXPECT_TRUE(output.get_by_position(2).column->is_null_at(0));
+ total += output.rows();
+ }
+ }
+ EXPECT_EQ(total, 3);
+ EXPECT_EQ(_responses->closes, split + 1);
+ }
+ ASSERT_TRUE(table.close().ok());
+}
+
+TEST_F(EsV2ReaderTest, RealHttpTransportClearsLatestScrollId) {
+ class Handler final : public HttpHandler {
+ public:
+ void handle(HttpRequest* req) override {
+ if (req->method() == DELETE) {
+ EXPECT_EQ(req->get_request_body(),
R"({"scroll_id":"last-scroll"})");
+ ++clears;
+ HttpChannel::send_reply(req, HttpStatus::OK,
R"({"succeeded":true})");
+ return;
+ }
+ const int call = searches.fetch_add(1);
+ const std::string response =
+ call < 2
+ ?
R"({"_scroll_id":"scroll-page","hits":{"hits":[{"_source":{"id":7}}]}})"
+ :
R"({"_scroll_id":"last-scroll","hits":{"hits":[]}})";
+ HttpChannel::send_reply(req, HttpStatus::OK, response);
+ }
+ std::atomic<int> searches {0};
+ std::atomic<int> clears {0};
+ } handler;
+ EvHttpServer server {0};
+ ASSERT_TRUE(server.register_handler(POST, "/example/_search", &handler));
+ ASSERT_TRUE(server.register_handler(POST, "/_search/scroll", &handler));
+ ASSERT_TRUE(server.register_handler(DELETE, "/_search/scroll", &handler));
+ server.start();
+ ASSERT_GT(server.get_real_port(), 0);
+ _range.table_format_params.es_params["host_port"] =
+ "http://127.0.0.1:"; + std::to_string(server.get_real_port());
+ auto props = std::make_shared<io::FileSystemProperties>();
+ auto file = std::make_unique<io::FileDescription>();
+ EsFileReader scan(props, file, _io, &_profile, _range, _params, _slots);
+ ASSERT_TRUE(scan.init(&_state).ok());
+ scan.set_batch_size(1);
+ auto projection = request({0});
+ const auto open_status = scan.open(projection);
+ ASSERT_TRUE(open_status.ok()) << open_status;
+ size_t rows = 0;
+ bool eof = false;
+ size_t total = 0;
+ while (!eof) {
+ auto output = block(scan, *projection);
+ const auto status = scan.get_block(&output, &rows, &eof);
+ ASSERT_TRUE(status.ok()) << status;
+ if (rows > 0) {
+ EXPECT_EQ(int_at(output, 0, 0), 7);
+ }
+ total += rows;
+ }
+ EXPECT_EQ(total, 2);
+ EXPECT_EQ(handler.searches.load(), 3);
+ EXPECT_EQ(handler.clears.load(), 1);
+ ASSERT_TRUE(scan.close().ok());
+ EXPECT_EQ(handler.clears.load(), 1);
+}
+
+TEST_F(EsV2ReaderTest, SparseProjectionDoesNotRequestOtherFields) {
+ auto scan = reader();
+ auto projection = request({2});
+ ASSERT_TRUE(scan->open(projection).ok());
+ EXPECT_FALSE(_responses->docvalues);
+ rapidjson::Document query;
+ query.Parse(_responses->properties.at("query").c_str());
+ ASSERT_FALSE(query.HasParseError());
+ ASSERT_EQ(query["_source"].Size(), 1);
+ EXPECT_STREQ(query["_source"][0].GetString(), "_id");
+ auto output = block(*scan, *projection);
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(scan->get_block(&output, &rows, &eof).ok());
+ ASSERT_EQ(rows, 3);
+ EXPECT_EQ(string_at(output, 0, 2), "c");
+}
+
+TEST_F(EsV2ReaderTest, MalformedResponseReleasesStream) {
+ _responses->pages = {"invalid-json"};
+ auto scan = reader();
+ EXPECT_FALSE(scan->open(request({0})).ok());
+ EXPECT_EQ(_responses->closes, 1);
+}
+
+TEST_F(EsV2ReaderTest, FullyFilteredBatchesKeepScanningUntilEof) {
+ auto scan = reader();
+ scan->set_batch_size(1);
+ auto projection = request({0});
+ projection->conjuncts.push_back(
+ prepared_conjunct(&_state,
std::make_shared<NullableIntGreaterThanExpr>(0, 10)));
+ ASSERT_TRUE(scan->open(projection).ok());
+ for (int batch = 0; batch < 4; ++batch) {
+ auto output = block(*scan, *projection);
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(scan->get_block(&output, &rows, &eof).ok());
+ EXPECT_EQ(rows, 0);
+ EXPECT_EQ(eof, batch == 3);
+ }
+ EXPECT_EQ(_io->predicate_filtered_rows, 3);
+ EXPECT_EQ(_responses->closes, 1);
+}
+
+} // namespace doris::format::es
diff --git a/build-support/check-header-deps.py
b/build-support/check-header-deps.py
index 214b8d079c2..c06807e42ac 100755
--- a/build-support/check-header-deps.py
+++ b/build-support/check-header-deps.py
@@ -591,7 +591,7 @@ REVERSE_REACH_BASELINES = {
# parsed_page.h holds RleDecoder<bool> by value and constructs it in the
# header, so the whole storage read stack sees the RLE machinery; a live
# dependency, budgeted as-is (trimming it is a refactor, not a gate).
- "util/rle_encoding.h": 823,
+ "util/rle_encoding.h": 825,
"gen_cpp/internal_service.pb.h": 741,
"gen_cpp/FrontendService_types.h": 576,
"storage/options.h": 464,
diff --git
a/regression-test/suites/external_table_p2/es/test_es_file_scanner_v2.groovy
b/regression-test/suites/external_table_p2/es/test_es_file_scanner_v2.groovy
new file mode 100644
index 00000000000..e23dce31b85
--- /dev/null
+++ b/regression-test/suites/external_table_p2/es/test_es_file_scanner_v2.groovy
@@ -0,0 +1,72 @@
+// 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_es_file_scanner_v2",
"p2,external,es,external_docker,external_docker_es") {
+ String enabled = context.config.otherConfigs.get("enableEsTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ return
+ }
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ def originalV2 = sql("show variables like 'enable_file_scanner_v2'")[0][1]
+ def originalBatchSize = sql("show variables like 'batch_size'")[0][1]
+ try {
+ // Small batches force scroll pagination and repeated output block
production.
+ sql "set batch_size = 1"
+ [5, 6, 7, 8].each { version ->
+ String port = context.config.otherConfigs.get("es_${version}_port")
+ String timeIndex = version <= 6 ? "test2_20220808" : "test2"
+ [false, true].each { docvalues ->
+ String catalog = "test_es_v2_${version}_${docvalues}"
+ sql "drop catalog if exists ${catalog}"
+ sql """create catalog ${catalog} properties (
+ "type" = "es",
+ "hosts" = "http://${externalEnvIp}:${port}";,
+ "nodes_discovery" = "false",
+ "enable_keyword_sniff" = "true",
+ "mapping_es_id" = "true",
+ "enable_docvalue_scan" = "${docvalues}"
+ )"""
+ sql "switch ${catalog}"
+ def queries = [
+ "select test2, test1, test6 from test1 order by test1,
test2, test6",
+ "select _id, test1 from test1 order by _id",
+ "select c_bool, c_integer, c_keyword, c_person from test1
order by test1",
+ "select test1 from ${timeIndex} where substring(test2, 2)
= 'ext2' order by test1",
+ "select test1 from test1 where test1 like 'string%' order
by test1",
+ "select count(*), count(message) from test1",
+ "select test1 from test1 where message is null order by
test1",
+ "select test1 from test1 where test1 = 'missing-value'",
+ "select test1 from test1 order by test1 limit 2",
+ "select count(*) from (select * from test1 limit 2) t",
+ "select * from composite_type_array order by name"
+ ]
+ // Compare against the existing scanner on the same index and
ES version instead
+ // of embedding fixture results that can vary across the
external Docker images.
+ queries.each { query ->
+ sql "set enable_file_scanner_v2 = false"
+ def expected = sql(query)
+ sql "set enable_file_scanner_v2 = true"
+ def actual = sql(query)
+ assertEquals(expected, actual, "ES ${version},
docvalues=${docvalues}: ${query}")
+ }
+ }
+ }
+ } finally {
+ sql "set enable_file_scanner_v2 = ${originalV2}"
+ sql "set batch_size = ${originalBatchSize}"
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
