[feat](catalog) support ADBC catalog that reads external sources over Arrow

.licenserc.yaml

@@ -64,6 +64,7 @@ header:
     # the file an administrator edits in plugins/connector/<dir>/. Matched by name
     # rather than by a **/*.conf.template glob, so a new one is a deliberate entry
     # here rather than something a wildcard silently absorbs.
+    - "fe/fe-connector/fe-connector-adbc/src/main/resources/adbc.conf.template"
     - "fe/fe-connector/fe-connector-hive/src/main/resources/hms.conf.template"
     - "fe/fe-connector/fe-connector-iceberg/src/main/resources/iceberg.conf.template"
     - "fe/fe-connector/fe-connector-jdbc/src/main/resources/jdbc.conf.template"

be/cmake/thirdparty.cmake

@@ -109,6 +109,7 @@ add_thirdparty(arrow_flight LIB64)
 add_thirdparty(arrow_flight_sql LIB64)
 add_thirdparty(arrow_dataset LIB64)
 add_thirdparty(arrow_acero LIB64)
+add_thirdparty(adbc_driver_manager LIB64)
 add_thirdparty(parquet LIB64)
 # liblance_c.a contains compiler_builtins cbrt symbols. Place libm before it
 # so the final linker resolves C math symbols from the system library first.

be/src/exec/operator/file_scan_operator.cpp

@@ -113,6 +113,17 @@ bool FileScanLocalState::TEST_should_use_file_scanner_v2(const TQueryOptions& qu
 bool FileScanLocalState::_should_use_file_scanner_v2(const TQueryOptions& query_options,
                                                      bool is_load,
                                                      const TFileScanRangeParams& scan_params) {
+    // ADBC only has a FileScannerV2 reader, and enable_file_scanner_v2 is a session variable marked
+    // fuzzy=true, so the regression harness flips it to false at random. Without letting adbc
+    // through unconditionally, those queries land in v1, which has no "adbc" branch, and come back
+    // with an undiagnosable NotSupported -- showing up in CI as a random failure. This is the same
+    // mechanism as is_transactional_hive below, pointed the other way. Loads are left alone: there
+    // is no ADBC load path, so widening the rule to cover them would only route them somewhere they
+    // still cannot run.
+    if (!is_load && scan_params.__isset.table_format_params &&
+        scan_params.table_format_params.table_format_type == "adbc") {
+        return true;
+    }
     const bool is_transactional_hive =
             scan_params.__isset.table_format_params &&
             scan_params.table_format_params.table_format_type == "transactional_hive";

be/src/exec/scan/file_scanner_v2.cpp

@@ -54,6 +54,7 @@
 #include "format_v2/jni/jdbc_reader.h"
 #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/hive_reader.h"
 #include "format_v2/table/hudi_reader.h"
 #include "format_v2/table/iceberg_position_delete_sys_table_reader.h"
@@ -103,7 +104,8 @@ bool is_supported_table_format(const TFileRangeDesc& range) {
 }
 
 bool is_supported_arrow_table_format(const TFileRangeDesc& range) {
-    return table_format_name(range) == "remote_doris";
+    const auto table_format = table_format_name(range);
+    return table_format == "remote_doris" || table_format == "adbc";
 }
 
 bool is_supported_jni_table_format(const TFileRangeDesc& range) {
@@ -619,6 +621,8 @@ Status FileScannerV2::_create_table_reader_for_format(
         *reader = std::make_unique<format::trino_connector::TrinoConnectorJniReader>();
     } else if (table_format == "remote_doris") {
         *reader = std::make_unique<format::remote_doris::RemoteDorisReader>();
+    } else if (table_format == "adbc") {
+        *reader = std::make_unique<format::adbc::AdbcReader>();
     } else {
         return Status::NotSupported("FileScannerV2 does not support table format {}", table_format);
     }

be/src/format/arrow/arrow_array_normalizer.cpp

@@ -0,0 +1,113 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "format/arrow/arrow_array_normalizer.h"
+
+#include <arrow/array/array_base.h>
+#include <arrow/compute/cast.h>
+#include <arrow/type.h>
+
+#include <memory>
+
+#include "common/check.h"
+#include "common/status.h"
+
+namespace doris {
+
+namespace {
+
+// The accepted counterpart of an encoding-only variant. Null when there is none.
+std::shared_ptr<arrow::DataType> target_type_for(const arrow::DataType& type) {
+    switch (type.id()) {
+    case arrow::Type::LARGE_STRING:
+    case arrow::Type::STRING_VIEW:
+        return arrow::utf8();
+    case arrow::Type::LARGE_BINARY:
+    case arrow::Type::BINARY_VIEW:
+        return arrow::binary();
+    case arrow::Type::DICTIONARY:
+        return static_cast<const arrow::DictionaryType&>(type).value_type();
+    case arrow::Type::RUN_END_ENCODED:
+        return static_cast<const arrow::RunEndEncodedType&>(type).value_type();
+    default:
+        return nullptr;
+    }
+}
+
+} // namespace
+
+bool is_serde_acceptable_arrow_type(const arrow::DataType& type) {
+    switch (type.id()) {
+    // Encoding-only variants: convertible to an accepted type.
+    case arrow::Type::LARGE_STRING:
+    case arrow::Type::LARGE_BINARY:
+    case arrow::Type::STRING_VIEW:
+    case arrow::Type::BINARY_VIEW:
+    case arrow::Type::DICTIONARY:
+    case arrow::Type::RUN_END_ENCODED:
+        return false;
+    // No Doris column can hold these, so they must not reach a serde either.
+    case arrow::Type::INTERVAL_MONTHS:
+    case arrow::Type::INTERVAL_DAY_TIME:
+    case arrow::Type::INTERVAL_MONTH_DAY_NANO:
+    case arrow::Type::DURATION:
+    case arrow::Type::SPARSE_UNION:
+    case arrow::Type::DENSE_UNION:
+        return false;
+    default:
+        return true;
+    }
+}
+
+Status normalize_arrow_array(const std::shared_ptr<arrow::Array>& arr,
+                             std::shared_ptr<arrow::Array>* out) {
+    DORIS_CHECK(arr != nullptr);
+    DORIS_CHECK(out != nullptr);
+
+    std::shared_ptr<arrow::Array> current = arr;
+    // dictionary<int32, large_utf8> decodes to large_utf8, which still needs converting. The bound
+    // stops a driver from turning a malformed type into an endless loop.
+    constexpr int kMaxPasses = 4;
+    for (int pass = 0; pass < kMaxPasses; ++pass) {
+        const auto& type = *current->type();
+        if (is_serde_acceptable_arrow_type(type)) {
+            *out = std::move(current);
+            return Status::OK();
+        }
+
+        auto target = target_type_for(type);
+        if (target == nullptr) {
+            return Status::NotSupported(
+                    "ADBC: arrow type '{}' cannot be materialized into a Doris column",
+                    type.ToString());
+        }
+
+        auto casted = arrow::compute::Cast(*current, target);
+        if (!casted.ok()) {
+            return Status::InternalError("ADBC: failed to normalize arrow type '{}' to '{}': {}",
+                                         type.ToString(), target->ToString(),
+                                         casted.status().ToString());
+        }
+        current = casted.MoveValueUnsafe();
+    }
+    return Status::InternalError(
+            "ADBC: arrow type '{}' is still not materializable after {} "
+            "normalization passes",
+            arr->type()->ToString(), kMaxPasses);
+}
+
+} // namespace doris

be/src/format/arrow/arrow_array_normalizer.h

@@ -0,0 +1,44 @@
+// 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 <arrow/type_fwd.h>
+
+#include <memory>
+
+#include "common/status.h"
+
+namespace doris {
+
+/// DataTypeSerDe::read_column_from_arrow was written for the Arrow that Doris itself emits, so it
+/// accepts only a subset of the Arrow type variants. Third-party ADBC drivers emit others: DuckDB
+/// emits string_view, Go-based drivers may emit large_* and dictionary. This normalizes them into
+/// a shape the serdes accept.
+///
+/// Only top-level types are normalized. A nested type whose child is an unaccepted variant (say
+/// list<large_utf8>) passes through and fails inside the serde, loudly rather than silently.
+
+/// Whether the serdes take this Arrow type as-is.
+bool is_serde_acceptable_arrow_type(const arrow::DataType& type);
+
+/// Normalizes `arr` into a serde-acceptable shape. Returns it unchanged (no copy) when it already
+/// is one, and fails with the offending type named when no accepted shape exists.
+Status normalize_arrow_array(const std::shared_ptr<arrow::Array>& arr,
+                             std::shared_ptr<arrow::Array>* out);
+
+} // namespace doris