[feature](external) Support Elasticsearch HTTP scans in FileScannerV2

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;
 }
 

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();

be/src/format/table/es/es_http_reader.cpp

@@ -19,17 +19,14 @@
 
 #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"
 #include "format/table/es/es_scroll_query.h"
 #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

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

be/src/format/table/es/es_scan_reader.h

@@ -58,6 +58,9 @@ class ESScanReader {
     // 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;

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;

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 @@ class ScrollParser {
                         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;
 

be/src/format_v2/file_reader.h

@@ -60,6 +60,7 @@ enum class FileFormat {
     JNI,
     ARROW,
     WAL,
+    ES_HTTP,
 };
 
 struct FileScanRequest {

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()) {

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.