[Feature] Support Doris Catalog (#55274)

The new Doris Catalog replaces the previous method of accessing external
tables in remote Doris clusters via JDBC Catalog.

related: #56011

1. It retrieves metadata of Doris external tables through HTTP APIs.
2. The metadata information for Doris external tables is more accurate,
fully capturing various metadata from remote cluster tables, such as
primary keys, bucketing keys, and native Doris data types.
3. The underlying implementation uses Arrow Flight SQL to communicate
with the remote Doris cluster, achieving approximately 4 times higher
transmission performance compared to the single-node JDBC Catalog.
4. It supports concurrent retrieval of Arrow response results from the
remote Doris cluster, with transmission performance scaling linearly as
the cluster size increases.

Author: Chen768959

be/src/runtime/descriptors.cpp

@@ -341,6 +341,17 @@ std::string JdbcTableDescriptor::debug_string() const {
     return fmt::to_string(buf);
 }
 
+RemoteDorisTableDescriptor::RemoteDorisTableDescriptor(const TTableDescriptor& tdesc)
+        : TableDescriptor(tdesc) {}
+
+RemoteDorisTableDescriptor::~RemoteDorisTableDescriptor() = default;
+
+std::string RemoteDorisTableDescriptor::debug_string() const {
+    std::stringstream out;
+    out << "RemoteDorisTable(" << TableDescriptor::debug_string() << ")";
+    return out.str();
+}
+
 TupleDescriptor::TupleDescriptor(const TTupleDescriptor& tdesc, bool own_slots)
         : _id(tdesc.id),
           _num_materialized_slots(0),
@@ -614,6 +625,9 @@ Status DescriptorTbl::create(ObjectPool* pool, const TDescriptorTable& thrift_tb
         case TTableType::DICTIONARY_TABLE:
             desc = pool->add(new DictionaryTableDescriptor(tdesc));
             break;
+        case TTableType::REMOTE_DORIS_TABLE:
+            desc = pool->add(new RemoteDorisTableDescriptor(tdesc));
+            break;
         default:
             DCHECK(false) << "invalid table type: " << tdesc.tableType;
         }

be/src/runtime/descriptors.h

@@ -324,6 +324,15 @@ class JdbcTableDescriptor : public TableDescriptor {
     bool _connection_pool_keep_alive;
 };
 
+class RemoteDorisTableDescriptor : public TableDescriptor {
+public:
+    RemoteDorisTableDescriptor(const TTableDescriptor& tdesc);
+    ~RemoteDorisTableDescriptor() override;
+    std::string debug_string() const override;
+
+private:
+};
+
 class TupleDescriptor {
 public:
     TupleDescriptor(TupleDescriptor&&) = delete;

be/src/vec/exec/format/table/remote_doris_reader.cpp

@@ -0,0 +1,127 @@
+// 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 "remote_doris_reader.h"
+
+#include <iostream>
+#include <map>
+#include <memory>
+#include <string>
+
+#include "arrow/flight/client.h"
+#include "arrow/flight/types.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/memory_pool.h"
+#include "arrow/result.h"
+#include "arrow/status.h"
+#include "common/status.h"
+#include "runtime/descriptors.h"
+#include "runtime/runtime_state.h"
+#include "runtime/types.h"
+#include "util/arrow/utils.h"
+#include "vec/core/block.h"
+#include "vec/core/column_with_type_and_name.h"
+#include "vec/core/types.h"
+
+namespace doris {
+class RuntimeProfile;
+class RuntimeState;
+
+namespace vectorized {
+class Block;
+} // namespace vectorized
+} // namespace doris
+
+namespace doris::vectorized {
+#include "common/compile_check_begin.h"
+
+RemoteDorisReader::RemoteDorisReader(const std::vector<SlotDescriptor*>& file_slot_descs,
+                                     RuntimeState* state, RuntimeProfile* profile,
+                                     const TFileRangeDesc& range)
+        : _range(range), _file_slot_descs(file_slot_descs) {
+    TimezoneUtils::find_cctz_time_zone(TimezoneUtils::default_time_zone, _ctzz);
+}
+
+Status RemoteDorisReader::init_reader() {
+    RETURN_DORIS_STATUS_IF_ERROR(init_stream());
+    DCHECK(_stream != nullptr);
+    return Status::OK();
+}
+
+Status RemoteDorisReader::get_next_block(Block* block, size_t* read_rows, bool* eof) {
+    arrow::flight::FlightStreamChunk chunk;
+    RETURN_DORIS_STATUS_IF_ERROR(_stream->Next().Value(&chunk));
+
+    if (!chunk.data) {
+        *read_rows = 0;
+        *eof = true;
+        return Status::OK();
+    }
+
+    // convert arrow batch to block
+    auto batch = chunk.data;
+    auto num_rows = batch->num_rows();
+    auto num_columns = batch->num_columns();
+    for (int c = 0; c < num_columns; ++c) {
+        arrow::Array* column = batch->column(c).get();
+
+        std::string column_name = batch->schema()->field(c)->name();
+
+        try {
+            const vectorized::ColumnWithTypeAndName& column_with_name =
+                    block->get_by_name(column_name);
+            RETURN_IF_ERROR(column_with_name.type->get_serde()->read_column_from_arrow(
+                    column_with_name.column->assume_mutable_ref(), column, 0, num_rows, _ctzz));
+        } catch (Exception& e) {
+            return Status::InternalError(
+                    "Failed to convert from arrow to block, column_name: {}, e: {}", column_name,
+                    e.what());
+        }
+    }
+
+    *read_rows += num_rows;
+    return Status::OK();
+}
+
+Status RemoteDorisReader::get_columns(std::unordered_map<std::string, DataTypePtr>* name_to_type,
+                                      std::unordered_set<std::string>* missing_cols) {
+    for (const auto& slot : _file_slot_descs) {
+        name_to_type->emplace(slot->col_name(), slot->type());
+    }
+    return Status::OK();
+}
+
+Status RemoteDorisReader::close() {
+    RETURN_DORIS_STATUS_IF_ERROR(_flight_client->Close());
+    return Status::OK();
+}
+
+arrow::Status RemoteDorisReader::init_stream() {
+    ARROW_ASSIGN_OR_RAISE(auto location,
+                          arrow::flight::Location::Parse(
+                                  _range.table_format_params.remote_doris_params.location_uri));
+    ARROW_ASSIGN_OR_RAISE(auto ticket,
+                          arrow::flight::Ticket::Deserialize(
+                                  _range.table_format_params.remote_doris_params.ticket));
+    ARROW_ASSIGN_OR_RAISE(_flight_client, arrow::flight::FlightClient::Connect(location));
+    ARROW_ASSIGN_OR_RAISE(_stream, _flight_client->DoGet(ticket));
+
+    return arrow::Status::OK();
+}
+
+#include "common/compile_check_end.h"
+} // namespace doris::vectorized

be/src/vec/exec/format/table/remote_doris_reader.h

@@ -0,0 +1,70 @@
+// 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/flight/client.h>
+
+#include <cstddef>
+#include <string>
+#include <unordered_map>
+#include <unordered_set>
+#include <vector>
+
+#include "common/status.h"
+#include "exec/olap_common.h"
+#include "vec/exec/format/jni_reader.h"
+
+namespace doris {
+class RuntimeProfile;
+class RuntimeState;
+class SlotDescriptor;
+namespace vectorized {
+class Block;
+} // namespace vectorized
+} // namespace doris
+
+namespace doris::vectorized {
+#include "common/compile_check_begin.h"
+class RemoteDorisReader : public GenericReader {
+    ENABLE_FACTORY_CREATOR(RemoteDorisReader);
+
+public:
+    RemoteDorisReader(const std::vector<SlotDescriptor*>& file_slot_descs, RuntimeState* state,
+                      RuntimeProfile* profile, const TFileRangeDesc& range);
+
+    ~RemoteDorisReader() override = default;
+
+    Status init_reader();
+
+    Status get_next_block(Block* block, size_t* read_rows, bool* eof) override;
+
+    Status get_columns(std::unordered_map<std::string, DataTypePtr>* name_to_type,
+                       std::unordered_set<std::string>* missing_cols) override;
+
+    Status close() override;
+
+private:
+    arrow::Status init_stream();
+    const TFileRangeDesc& _range;
+    const std::vector<SlotDescriptor*>& _file_slot_descs;
+    cctz::time_zone _ctzz;
+    std::unique_ptr<arrow::flight::FlightClient> _flight_client;
+    std::unique_ptr<arrow::flight::FlightStreamReader> _stream;
+};
+#include "common/compile_check_end.h"
+} // namespace doris::vectorized

be/src/vec/exec/scan/file_scanner.cpp

@@ -70,6 +70,7 @@
 #include "vec/exec/format/table/max_compute_jni_reader.h"
 #include "vec/exec/format/table/paimon_jni_reader.h"
 #include "vec/exec/format/table/paimon_reader.h"
+#include "vec/exec/format/table/remote_doris_reader.h"
 #include "vec/exec/format/table/transactional_hive_reader.h"
 #include "vec/exec/format/table/trino_connector_jni_reader.h"
 #include "vec/exec/format/text/text_reader.h"
@@ -1136,9 +1137,17 @@ Status FileScanner::_get_next_reader() {
             break;
         }
         case TFileFormatType::FORMAT_ARROW: {
-            _cur_reader = ArrowStreamReader::create_unique(_state, _profile, &_counter, *_params,
-                                                           range, _file_slot_descs, _io_ctx.get());
-            init_status = ((ArrowStreamReader*)(_cur_reader.get()))->init_reader();
+            if (range.__isset.table_format_params &&
+                range.table_format_params.table_format_type == "remote_doris") {
+                _cur_reader =
+                        RemoteDorisReader::create_unique(_file_slot_descs, _state, _profile, range);
+                init_status = ((RemoteDorisReader*)(_cur_reader.get()))->init_reader();
+            } else {
+                _cur_reader =
+                        ArrowStreamReader::create_unique(_state, _profile, &_counter, *_params,
+                                                         range, _file_slot_descs, _io_ctx.get());
+                init_status = ((ArrowStreamReader*)(_cur_reader.get()))->init_reader();
+            }
             break;
         }
         default:

fe/fe-core/src/main/java/org/apache/doris/catalog/KeysType.java

@@ -26,7 +26,8 @@ public enum KeysType {
     PRIMARY_KEYS,
     DUP_KEYS,
     UNIQUE_KEYS,
-    AGG_KEYS;
+    AGG_KEYS,
+    UNKNOWN;
 
     /**
      * Determine whether it is an aggregation type.

fe/fe-core/src/main/java/org/apache/doris/catalog/TableIf.java

@@ -398,7 +398,7 @@ enum TableType {
         @Deprecated ICEBERG, @Deprecated HUDI, JDBC,
         TABLE_VALUED_FUNCTION, HMS_EXTERNAL_TABLE, ES_EXTERNAL_TABLE, MATERIALIZED_VIEW, JDBC_EXTERNAL_TABLE,
         ICEBERG_EXTERNAL_TABLE, TEST_EXTERNAL_TABLE, PAIMON_EXTERNAL_TABLE, MAX_COMPUTE_EXTERNAL_TABLE,
-        HUDI_EXTERNAL_TABLE, TRINO_CONNECTOR_EXTERNAL_TABLE, LAKESOUl_EXTERNAL_TABLE, DICTIONARY;
+        HUDI_EXTERNAL_TABLE, TRINO_CONNECTOR_EXTERNAL_TABLE, LAKESOUl_EXTERNAL_TABLE, DICTIONARY, DORIS_EXTERNAL_TABLE;
 
         public String toEngineName() {
             switch (this) {
@@ -475,6 +475,7 @@ public String toMysqlType() {
                 case PAIMON_EXTERNAL_TABLE:
                 case MATERIALIZED_VIEW:
                 case TRINO_CONNECTOR_EXTERNAL_TABLE:
+                case DORIS_EXTERNAL_TABLE:
                     return "BASE TABLE";
                 default:
                     return null;

fe/fe-core/src/main/java/org/apache/doris/common/util/JsonUtil.java

@@ -78,4 +78,16 @@ public static ObjectNode parseObject(String text) {
     public static <T> T readValue(String text, Class<T> clazz) throws JsonProcessingException {
         return objectMapper.readValue(text, clazz);
     }
+
+    public static Integer safeGetAsInt(ObjectNode node, String field) {
+        JsonNode value = node.get(field);
+        return (value == null || value.isNull()) ? null : value.asInt();
+    }
+
+    public static String convertNodeToString(JsonNode node) {
+        if (node == null || node.isNull()) {
+            return null;
+        }
+        return node.isTextual() ? node.asText() : node.toString();
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogFactory.java

@@ -21,6 +21,7 @@
 import org.apache.doris.catalog.Resource;
 import org.apache.doris.common.DdlException;
 import org.apache.doris.common.FeConstants;
+import org.apache.doris.datasource.doris.RemoteDorisExternalCatalog;
 import org.apache.doris.datasource.es.EsExternalCatalog;
 import org.apache.doris.datasource.hive.HMSExternalCatalog;
 import org.apache.doris.datasource.iceberg.IcebergExternalCatalogFactory;
@@ -108,6 +109,9 @@ private static CatalogIf createCatalog(long catalogId, String name, String resou
                 break;
             case "lakesoul":
                 throw new DdlException("Lakesoul catalog is no longer supported");
+            case "doris":
+                catalog = new RemoteDorisExternalCatalog(catalogId, name, resource, props, comment);
+                break;
             case "test":
                 if (!FeConstants.runningUnitTest) {
                     throw new DdlException("test catalog is only for FE unit test");

fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java

@@ -38,6 +38,7 @@
 import org.apache.doris.common.util.Util;
 import org.apache.doris.datasource.ExternalSchemaCache.SchemaCacheKey;
 import org.apache.doris.datasource.connectivity.CatalogConnectivityTestCoordinator;
+import org.apache.doris.datasource.doris.RemoteDorisExternalDatabase;
 import org.apache.doris.datasource.es.EsExternalDatabase;
 import org.apache.doris.datasource.hive.HMSExternalCatalog;
 import org.apache.doris.datasource.hive.HMSExternalDatabase;
@@ -842,6 +843,8 @@ protected ExternalDatabase<? extends ExternalTable> buildDbForInit(String remote
                 return new PaimonExternalDatabase(this, dbId, localDbName, remoteDbName);
             case TRINO_CONNECTOR:
                 return new TrinoConnectorExternalDatabase(this, dbId, localDbName, remoteDbName);
+            case REMOTE_DORIS:
+                return new RemoteDorisExternalDatabase(this, dbId, localDbName, remoteDbName);
             default:
                 break;
         }