From e471db45d8f20ac258c58be1c42dd547ebd4fe68 Mon Sep 17 00:00:00 2001 From: Chen768959 <67011523+Chen768959@users.noreply.github.com> Date: Mon, 10 Nov 2025 17:04:26 +0800 Subject: [PATCH 1/2] [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. --- be/src/runtime/descriptors.cpp | 14 + be/src/runtime/descriptors.h | 9 + .../exec/format/table/remote_doris_reader.cpp | 127 +++++++ .../exec/format/table/remote_doris_reader.h | 70 ++++ be/src/vec/exec/scan/file_scanner.cpp | 15 +- .../org/apache/doris/catalog/KeysType.java | 3 +- .../org/apache/doris/catalog/TableIf.java | 3 +- .../apache/doris/common/util/JsonUtil.java | 12 + .../doris/datasource/CatalogFactory.java | 4 + .../doris/datasource/ExternalCatalog.java | 3 + .../doris/datasource/InitCatalogLog.java | 1 + .../doris/datasource/InitDatabaseLog.java | 1 + .../doris/datasource/TableFormatType.java | 3 +- .../RemoteDorisCompatibleRestClient.java | 137 ++++++++ .../doris/RemoteDorisExternalCatalog.java | 202 +++++++++++ .../doris/RemoteDorisExternalDatabase.java | 36 ++ .../doris/RemoteDorisExternalTable.java | 89 +++++ .../doris/RemoteDorisRestClient.java | 299 ++++++++++++++++ .../doris/source/RemoteDorisScanNode.java | 327 ++++++++++++++++++ .../doris/source/RemoteDorisSource.java | 88 +++++ .../doris/source/RemoteDorisSplit.java | 53 +++ .../constants/RemoteDorisProperties.java | 50 +++ .../doris/httpv2/rest/HealthAction.java | 6 +- .../doris/httpv2/rest/TableSchemaAction.java | 52 +++ .../rest/response/GsonSchemaResponse.java | 30 ++ .../translator/PhysicalPlanTranslator.java | 4 + .../nereids/rules/analysis/BindRelation.java | 1 + .../apache/doris/persist/gson/GsonUtils.java | 4 +- .../doris/statistics/DeriveFactory.java | 1 + .../doris/statistics/StatisticalType.java | 3 +- .../RemoteDorisCompatibleRestClientTest.java | 72 ++++ .../doris/RemoteDorisRestClientTest.java | 108 ++++++ gensrc/thrift/Descriptors.thrift | 7 + gensrc/thrift/PlanNodes.thrift | 10 + gensrc/thrift/Types.thrift | 3 +- regression-test/conf/regression-conf.groovy | 1 + .../test_remote_doris_all_types_select.out | 16 + .../test_remote_doris_all_types_show.out | 76 ++++ .../test_remote_doris_refresh.out | 16 + .../test_remote_doris_statistics.out | 20 ++ .../external/conf/regression-conf.groovy | 7 + .../test_remote_doris_all_types_select.groovy | 172 +++++++++ .../test_remote_doris_all_types_show.groovy | 168 +++++++++ .../test_remote_doris_catalog.groovy | 68 ++++ .../test_remote_doris_predict.groovy | 150 ++++++++ .../test_remote_doris_refresh.groovy | 128 +++++++ .../test_remote_doris_statistics.groovy | 105 ++++++ .../test_remote_doris_table_stats.groovy | 99 ++++++ 48 files changed, 2862 insertions(+), 11 deletions(-) create mode 100644 be/src/vec/exec/format/table/remote_doris_reader.cpp create mode 100644 be/src/vec/exec/format/table/remote_doris_reader.h create mode 100644 fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisCompatibleRestClient.java create mode 100644 fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalCatalog.java create mode 100644 fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalDatabase.java create mode 100644 fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalTable.java create mode 100644 fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisRestClient.java create mode 100644 fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisScanNode.java create mode 100644 fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisSource.java create mode 100644 fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisSplit.java create mode 100644 fe/fe-core/src/main/java/org/apache/doris/datasource/property/constants/RemoteDorisProperties.java create mode 100644 fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/response/GsonSchemaResponse.java create mode 100644 fe/fe-core/src/test/java/org/apache/doris/datasource/doris/RemoteDorisCompatibleRestClientTest.java create mode 100644 fe/fe-core/src/test/java/org/apache/doris/datasource/doris/RemoteDorisRestClientTest.java create mode 100644 regression-test/data/external_table_p0/remote_doris/test_remote_doris_all_types_select.out create mode 100644 regression-test/data/external_table_p0/remote_doris/test_remote_doris_all_types_show.out create mode 100644 regression-test/data/external_table_p0/remote_doris/test_remote_doris_refresh.out create mode 100644 regression-test/data/external_table_p0/remote_doris/test_remote_doris_statistics.out create mode 100644 regression-test/suites/external_table_p0/remote_doris/test_remote_doris_all_types_select.groovy create mode 100644 regression-test/suites/external_table_p0/remote_doris/test_remote_doris_all_types_show.groovy create mode 100644 regression-test/suites/external_table_p0/remote_doris/test_remote_doris_catalog.groovy create mode 100644 regression-test/suites/external_table_p0/remote_doris/test_remote_doris_predict.groovy create mode 100644 regression-test/suites/external_table_p0/remote_doris/test_remote_doris_refresh.groovy create mode 100644 regression-test/suites/external_table_p0/remote_doris/test_remote_doris_statistics.groovy create mode 100644 regression-test/suites/external_table_p0/remote_doris/test_remote_doris_table_stats.groovy diff --git a/be/src/runtime/descriptors.cpp b/be/src/runtime/descriptors.cpp index 7ccc69ed9aca45..d99cbdc038cbbd 100644 --- a/be/src/runtime/descriptors.cpp +++ b/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; } diff --git a/be/src/runtime/descriptors.h b/be/src/runtime/descriptors.h index 0d86844b226a03..0481c4ebfdbac7 100644 --- a/be/src/runtime/descriptors.h +++ b/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; diff --git a/be/src/vec/exec/format/table/remote_doris_reader.cpp b/be/src/vec/exec/format/table/remote_doris_reader.cpp new file mode 100644 index 00000000000000..fa0f8566c0f49f --- /dev/null +++ b/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 +#include +#include +#include + +#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& 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* name_to_type, + std::unordered_set* 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 diff --git a/be/src/vec/exec/format/table/remote_doris_reader.h b/be/src/vec/exec/format/table/remote_doris_reader.h new file mode 100644 index 00000000000000..8884055d99a5c5 --- /dev/null +++ b/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 + +#include +#include +#include +#include +#include + +#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& 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* name_to_type, + std::unordered_set* missing_cols) override; + + Status close() override; + +private: + arrow::Status init_stream(); + const TFileRangeDesc& _range; + const std::vector& _file_slot_descs; + cctz::time_zone _ctzz; + std::unique_ptr _flight_client; + std::unique_ptr _stream; +}; +#include "common/compile_check_end.h" +} // namespace doris::vectorized diff --git a/be/src/vec/exec/scan/file_scanner.cpp b/be/src/vec/exec/scan/file_scanner.cpp index 476eae26ebe2fa..2a34972f05fa74 100644 --- a/be/src/vec/exec/scan/file_scanner.cpp +++ b/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: diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/KeysType.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/KeysType.java index c280c522c3df6b..9d19d34ef5f3b4 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/KeysType.java +++ b/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. diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/TableIf.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/TableIf.java index da5697b922efe0..1dd34b43cc6f61 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/TableIf.java +++ b/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; diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/util/JsonUtil.java b/fe/fe-core/src/main/java/org/apache/doris/common/util/JsonUtil.java index d1eccaeb4e058f..3254a67d83c03b 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/common/util/JsonUtil.java +++ b/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 readValue(String text, Class 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(); + } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogFactory.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogFactory.java index 3819d5825314fc..8ff1db71771787 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogFactory.java +++ b/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"); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java index 824ed9c44ee9b2..26e2fed7edda37 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java +++ b/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 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; } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/InitCatalogLog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/InitCatalogLog.java index ac262764de64c9..2631ff28cc112c 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/InitCatalogLog.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/InitCatalogLog.java @@ -43,6 +43,7 @@ public enum Type { LAKESOUL, TEST, TRINO_CONNECTOR, + REMOTE_DORIS, UNKNOWN; } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/InitDatabaseLog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/InitDatabaseLog.java index d1ea04a16b03e5..ba927d8b6906fd 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/InitDatabaseLog.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/InitDatabaseLog.java @@ -44,6 +44,7 @@ public enum Type { TEST, INFO_SCHEMA_DB, TRINO_CONNECTOR, + REMOTE_DORIS, UNKNOWN; } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/TableFormatType.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/TableFormatType.java index 5f67cb3329e3b4..10d4fd25bcbc8b 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/TableFormatType.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/TableFormatType.java @@ -26,7 +26,8 @@ public enum TableFormatType { TRANSACTIONAL_HIVE("transactional_hive"), LAKESOUL("lakesoul"), TRINO_CONNECTOR("trino_connector"), - TVF("tvf"); + TVF("tvf"), + REMOTE_DORIS("remote_doris"); private final String tableFormatType; diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisCompatibleRestClient.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisCompatibleRestClient.java new file mode 100644 index 00000000000000..f05f2eb0978904 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisCompatibleRestClient.java @@ -0,0 +1,137 @@ +// 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. + +package org.apache.doris.datasource.doris; + +import org.apache.doris.catalog.AggregateType; +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.ScalarType; +import org.apache.doris.catalog.Type; +import org.apache.doris.common.util.JsonUtil; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import lombok.Data; + +import java.util.ArrayList; +import java.util.List; + +/** + * For older remote Doris clusters, this restClient falls back to legacy APIs. + */ +public class RemoteDorisCompatibleRestClient extends RemoteDorisRestClient { + + /** + * For DorisTable. + **/ + public RemoteDorisCompatibleRestClient(List feNodes, String authUser, String authPassword, + boolean httpSslEnable, int retryCount, int maxIdleConnections, + long keepAliveDurationSec, long connectTimeoutSec, long readTimeoutSec, + long writeTimeoutSec, long callTimeoutSec) { + super(feNodes, authUser, authPassword, httpSslEnable, retryCount, maxIdleConnections, + keepAliveDurationSec, connectTimeoutSec, readTimeoutSec, writeTimeoutSec, callTimeoutSec); + } + + public List getColumns(String dbName, String tableName) { + DorisApiResponse tableSchemaResponse = parseResponse(execute("api/" + dbName + "/" + tableName + "/_schema"), + "get doris table schema error"); + + List columnList = new ArrayList<>(); + ObjectNode objectNode = JsonUtil.parseObject(tableSchemaResponse.getData()); + JsonNode properties = objectNode.path("properties"); + for (JsonNode columnJson : properties) { + if (columnJson.isObject()) { + columnList.add(parseColumn((ObjectNode) columnJson)); + } + } + return columnList; + } + + public static DorisApiResponse parseResponse(String response, String errMsg) { + if (response == null) { + throw new RuntimeException(errMsg); + } + + ObjectNode objectNode = JsonUtil.parseObject(response); + + return new DorisApiResponse( + objectNode.path(DorisApiResponse.MSG).asText(null), + JsonUtil.safeGetAsInt(objectNode, DorisApiResponse.CODE), + JsonUtil.convertNodeToString(objectNode.path(DorisApiResponse.DATA)), + JsonUtil.safeGetAsInt(objectNode, DorisApiResponse.COUNT) + ); + } + + public static Column parseColumn(ObjectNode columnJson) { + boolean nullable = columnJson.path("nullable").asBoolean(false); + String name = columnJson.path("name").asText(); + String comment = columnJson.path("comment").asText(); + boolean isKey = columnJson.path("key").asBoolean(false); + + String defaultValue = null; + JsonNode defaultValueJson = columnJson.get("default_value"); + if (defaultValueJson != null) { + defaultValue = JsonUtil.convertNodeToString(defaultValueJson); + } + + String typeName = columnJson.path("type").asText(); + Type type = ScalarType.createType(typeName); + + String aggregationTypeName = columnJson.path("aggregation_type").asText(); + AggregateType aggType = AggregateType.getAggTypeFromAggName(aggregationTypeName); + + JsonNode attributesJson = columnJson.get("type_attributes"); + if (attributesJson != null) { + String scale = attributesJson.path("scale").asText("0"); + String precision = attributesJson.path("precision").asText("0"); + String length = attributesJson.path("length").asText("0"); + + type = ScalarType.createType( + type.getPrimitiveType(), + Integer.parseInt(length), + Integer.parseInt(precision), + Integer.parseInt(scale) + ); + } + + return new Column(name, type, isKey, aggType, nullable, defaultValue, comment); + } + + // Avoid using org.apache.doris.httpv2.entity.ResponseBody to prevent potential future changes in ResponseBody. + // For backward compatibility with older versions, use a fixed ApiResponse structure instead. + @Data + public static class DorisApiResponse { + public static final String MSG = "msg"; + public static final String CODE = "code"; + public static final String DATA = "data"; + public static final String COUNT = "count"; + + private String msg; + private Integer code; + private String data; + private Integer count; + + public DorisApiResponse() {} + + public DorisApiResponse(String msg, Integer code, String data, Integer count) { + this.msg = msg; + this.code = code; + this.data = data; + this.count = count; + } + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalCatalog.java new file mode 100644 index 00000000000000..b63a2a03b1f37a --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalCatalog.java @@ -0,0 +1,202 @@ +// 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. + +package org.apache.doris.datasource.doris; + +import org.apache.doris.common.DdlException; +import org.apache.doris.datasource.CatalogProperty; +import org.apache.doris.datasource.ExternalCatalog; +import org.apache.doris.datasource.InitCatalogLog; +import org.apache.doris.datasource.SessionContext; +import org.apache.doris.datasource.property.constants.RemoteDorisProperties; + +import com.google.common.collect.ImmutableList; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.util.Arrays; +import java.util.List; +import java.util.Map; + +public class RemoteDorisExternalCatalog extends ExternalCatalog { + private static final Logger LOG = LogManager.getLogger(RemoteDorisExternalCatalog.class); + + private RemoteDorisRestClient dorisRestClient; + private static final List REQUIRED_PROPERTIES = ImmutableList.of( + RemoteDorisProperties.FE_HTTP_HOSTS, + RemoteDorisProperties.FE_ARROW_HOSTS, + RemoteDorisProperties.USER, + RemoteDorisProperties.PASSWORD + ); + + /** + * Default constructor for DorisExternalCatalog. + */ + public RemoteDorisExternalCatalog(long catalogId, String name, String resource, + Map props, String comment) { + super(catalogId, name, InitCatalogLog.Type.REMOTE_DORIS, comment); + this.catalogProperty = new CatalogProperty(resource, props); + } + + @Override + public void checkProperties() throws DdlException { + super.checkProperties(); + + for (String requiredProperty : REQUIRED_PROPERTIES) { + if (!catalogProperty.getProperties().containsKey(requiredProperty)) { + throw new DdlException("Required property '" + requiredProperty + "' is missing"); + } + } + } + + public List getFeNodes() { + return parseHttpHosts(catalogProperty.getOrDefault(RemoteDorisProperties.FE_HTTP_HOSTS, "")); + } + + public List getFeArrowNodes() { + return parseArrowHosts(catalogProperty.getOrDefault(RemoteDorisProperties.FE_ARROW_HOSTS, "")); + } + + public String getUsername() { + return catalogProperty.getOrDefault(RemoteDorisProperties.USER, ""); + } + + public String getPassword() { + return catalogProperty.getOrDefault(RemoteDorisProperties.PASSWORD, ""); + } + + public boolean enableSsl() { + return Boolean.parseBoolean(catalogProperty.getOrDefault(RemoteDorisProperties.METADATA_HTTP_SSL_ENABLED, + "false")); + } + + public boolean isCompatible() { + return Boolean.parseBoolean(catalogProperty.getOrDefault(RemoteDorisProperties.COMPATIBLE, + "false")); + } + + public boolean enableParallelResultSink() { + return Boolean.parseBoolean(catalogProperty.getOrDefault(RemoteDorisProperties.ENABLE_PARALLEL_RESULT_SINK, + "true")); + } + + public int getQueryRetryCount() { + return Integer.parseInt(catalogProperty.getOrDefault(RemoteDorisProperties.QUERY_RETRY_COUNT, + "3")); + } + + public int getQueryTimeoutSec() { + return Integer.parseInt(catalogProperty.getOrDefault(RemoteDorisProperties.QUERY_TIMEOUT_SEC, + "15")); + } + + public int getMetadataSyncRetryCount() { + return Integer.parseInt(catalogProperty.getOrDefault(RemoteDorisProperties.METADATA_SYNC_RETRIES_COUNT, + "3")); + } + + public int getMetadataMaxIdleConnections() { + return Integer.parseInt(catalogProperty.getOrDefault(RemoteDorisProperties.METADATA_MAX_IDLE_CONNECTIONS, + "5")); + } + + public int getMetadataKeepAliveDurationSec() { + return Integer.parseInt(catalogProperty.getOrDefault(RemoteDorisProperties.METADATA_KEEP_ALIVE_DURATION_SEC, + "300")); + } + + public int getMetadataConnectTimeoutSec() { + return Integer.parseInt(catalogProperty.getOrDefault(RemoteDorisProperties.METADATA_CONNECT_TIMEOUT_SEC, + "10")); + } + + public int getMetadataReadTimeoutSec() { + return Integer.parseInt(catalogProperty.getOrDefault(RemoteDorisProperties.METADATA_READ_TIMEOUT_SEC, + "10")); + } + + public int getMetadataWriteTimeoutSec() { + return Integer.parseInt(catalogProperty.getOrDefault(RemoteDorisProperties.METADATA_WRITE_TIMEOUT_SEC, + "10")); + } + + public int getMetadataCallTimeoutSec() { + return Integer.parseInt(catalogProperty.getOrDefault(RemoteDorisProperties.METADATA_CALL_TIMEOUT_SEC, + "0")); + } + + @Override + protected void initLocalObjectsImpl() { + if (isCompatible()) { + dorisRestClient = new RemoteDorisCompatibleRestClient( + getFeNodes(), getUsername(), getPassword(), enableSsl(), getMetadataSyncRetryCount(), + getMetadataMaxIdleConnections(), getMetadataKeepAliveDurationSec(), getMetadataConnectTimeoutSec(), + getMetadataReadTimeoutSec(), getMetadataWriteTimeoutSec(), getMetadataCallTimeoutSec() + ); + } else { + dorisRestClient = new RemoteDorisRestClient( + getFeNodes(), getUsername(), getPassword(), enableSsl(), getMetadataSyncRetryCount(), + getMetadataMaxIdleConnections(), getMetadataKeepAliveDurationSec(), getMetadataConnectTimeoutSec(), + getMetadataReadTimeoutSec(), getMetadataWriteTimeoutSec(), getMetadataCallTimeoutSec()); + } + + if (!dorisRestClient.health()) { + throw new RuntimeException("Failed to connect to Doris cluster," + + " please check your Doris cluster or your Doris catalog configuration."); + } + } + + protected List listDatabaseNames() { + makeSureInitialized(); + return dorisRestClient.getDatabaseNameList(); + } + + @Override + public List listTableNames(SessionContext ctx, String dbName) { + makeSureInitialized(); + return dorisRestClient.getTablesNameList(dbName); + } + + @Override + public boolean tableExist(SessionContext ctx, String dbName, String tblName) { + makeSureInitialized(); + return dorisRestClient.isTableExist(dbName, tblName); + } + + public RemoteDorisRestClient getDorisRestClient() { + return dorisRestClient; + } + + private List parseHttpHosts(String hosts) { + String[] hostUrls = hosts.trim().split(","); + fillUrlsWithSchema(hostUrls, enableSsl()); + return Arrays.asList(hostUrls); + } + + private void fillUrlsWithSchema(String[] urls, boolean isSslEnabled) { + for (int i = 0; i < urls.length; i++) { + String seed = urls[i].trim(); + if (!seed.startsWith("http://") && !seed.startsWith("https://")) { + urls[i] = (isSslEnabled ? "https://" : "http://") + seed; + } + } + } + + private List parseArrowHosts(String hosts) { + return Arrays.asList(hosts.trim().split(",")); + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalDatabase.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalDatabase.java new file mode 100644 index 00000000000000..5e5fd347ed1f3e --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalDatabase.java @@ -0,0 +1,36 @@ +// 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. + +package org.apache.doris.datasource.doris; + +import org.apache.doris.datasource.ExternalCatalog; +import org.apache.doris.datasource.ExternalDatabase; +import org.apache.doris.datasource.InitDatabaseLog; + +public class RemoteDorisExternalDatabase extends ExternalDatabase { + public RemoteDorisExternalDatabase(ExternalCatalog extCatalog, long id, String name, String remoteName) { + super(extCatalog, id, name, remoteName, InitDatabaseLog.Type.REMOTE_DORIS); + } + + @Override + public RemoteDorisExternalTable buildTableInternal(String remoteTableName, String localTableName, long tblId, + ExternalCatalog catalog, + ExternalDatabase db) { + return new RemoteDorisExternalTable(tblId, localTableName, remoteTableName, + (RemoteDorisExternalCatalog) extCatalog, db); + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalTable.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalTable.java new file mode 100644 index 00000000000000..4b6117aa313d1f --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalTable.java @@ -0,0 +1,89 @@ +// 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. + +package org.apache.doris.datasource.doris; + +import org.apache.doris.catalog.Column; +import org.apache.doris.datasource.ExternalDatabase; +import org.apache.doris.datasource.ExternalTable; +import org.apache.doris.datasource.SchemaCacheValue; +import org.apache.doris.statistics.AnalysisInfo; +import org.apache.doris.statistics.BaseAnalysisTask; +import org.apache.doris.statistics.ExternalAnalysisTask; +import org.apache.doris.thrift.TRemoteDorisTable; +import org.apache.doris.thrift.TTableDescriptor; +import org.apache.doris.thrift.TTableType; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.util.List; +import java.util.Optional; + +public class RemoteDorisExternalTable extends ExternalTable { + private static final Logger LOG = LogManager.getLogger(RemoteDorisExternalTable.class); + + public RemoteDorisExternalTable(long id, String name, String remoteName, + RemoteDorisExternalCatalog catalog, ExternalDatabase db) { + super(id, name, remoteName, catalog, db, TableType.DORIS_EXTERNAL_TABLE); + } + + @Override + protected synchronized void makeSureInitialized() { + super.makeSureInitialized(); + if (!objectCreated) { + objectCreated = true; + } + } + + @Override + public TTableDescriptor toThrift() { + List schema = getFullSchema(); + TRemoteDorisTable tRemoteDorisTable = new TRemoteDorisTable(); + tRemoteDorisTable.setDbName(dbName); + tRemoteDorisTable.setTableName(name); + tRemoteDorisTable.setProperties(getCatalog().getProperties()); + + TTableDescriptor tTableDescriptor = new TTableDescriptor(getId(), + TTableType.REMOTE_DORIS_TABLE, schema.size(), 0, getName(), dbName); + + tTableDescriptor.setRemoteDorisTable(tRemoteDorisTable); + return tTableDescriptor; + } + + @Override + public Optional initSchema() { + RemoteDorisRestClient restClient = ((RemoteDorisExternalCatalog) catalog).getDorisRestClient(); + return Optional.of(new SchemaCacheValue(restClient.getColumns(dbName, name))); + } + + @Override + public BaseAnalysisTask createAnalysisTask(AnalysisInfo info) { + makeSureInitialized(); + return new ExternalAnalysisTask(info); + } + + @Override + public long fetchRowCount() { + RemoteDorisRestClient restClient = ((RemoteDorisExternalCatalog) catalog).getDorisRestClient(); + return restClient.getRowCount(getDbName(), getName()); + } + + public String getExternalTableName() { + return getDbName() + "." + getName(); + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisRestClient.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisRestClient.java new file mode 100644 index 00000000000000..ad72ee34ce379a --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisRestClient.java @@ -0,0 +1,299 @@ +// 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. + +package org.apache.doris.datasource.doris; + +import org.apache.doris.catalog.Column; +import org.apache.doris.common.util.JsonUtil; +import org.apache.doris.httpv2.entity.ResponseBody; +import org.apache.doris.httpv2.rest.HealthAction; +import org.apache.doris.httpv2.rest.RestApiStatusCode; +import org.apache.doris.httpv2.rest.response.GsonSchemaResponse; +import org.apache.doris.persist.gson.GsonUtils; + +import com.fasterxml.jackson.databind.node.ObjectNode; +import com.google.gson.reflect.TypeToken; +import okhttp3.ConnectionPool; +import okhttp3.Credentials; +import okhttp3.OkHttpClient; +import okhttp3.Request; +import okhttp3.Response; +import org.apache.http.HttpHeaders; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.apache.logging.log4j.util.Strings; + +import java.io.IOException; +import java.lang.reflect.Type; +import java.security.SecureRandom; +import java.security.cert.CertificateException; +import java.security.cert.X509Certificate; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; +import javax.net.ssl.HostnameVerifier; +import javax.net.ssl.SSLContext; +import javax.net.ssl.SSLSession; +import javax.net.ssl.SSLSocketFactory; +import javax.net.ssl.TrustManager; +import javax.net.ssl.X509TrustManager; + +/** + * Use this restClient when the remote Doris cluster is the same version as the current cluster, + * ensuring complete tableSchema compatibility. + */ +public class RemoteDorisRestClient { + private static final Logger LOG = LogManager.getLogger(RemoteDorisRestClient.class); + + private final List feNodes; + private final String authUser; + private final String authPassword; + private final int retryCount; + private static OkHttpClient httpClient; + + private String currentNode; + private int currentNodeIndex = 0; + + /** + * For DorisTable. + **/ + public RemoteDorisRestClient(List feNodes, String authUser, String authPassword, + boolean httpSslEnable, int retryCount, int maxIdleConnections, + long keepAliveDurationSec, long connectTimeoutSec, long readTimeoutSec, + long writeTimeoutSec, long callTimeoutSec) { + this.feNodes = feNodes; + this.authUser = authUser; + this.authPassword = authPassword; + this.retryCount = retryCount; + httpClient = initHttpClient(httpSslEnable, maxIdleConnections, keepAliveDurationSec, + connectTimeoutSec, readTimeoutSec, writeTimeoutSec, callTimeoutSec); + + this.currentNode = feNodes.get(currentNodeIndex); + } + + public List getDatabaseNameList() { + return parseStringLists(execute("api/meta/namespaces/default_cluster/databases")); + } + + public List getTablesNameList(String dbName) { + return parseStringLists(execute("api/meta/namespaces/default_cluster/databases/" + dbName + "/tables")); + } + + public boolean isTableExist(String dbName, String tableName) { + return parseSuccessResponse(execute("api/" + dbName + "/" + tableName + "/_schema")); + } + + public boolean health() { + int aliveBeNum = parseOnlineBeNum(execute("api/health")); + return aliveBeNum > 0; + } + + public List getColumns(String dbName, String tableName) { + return parseColumns(execute("api/" + dbName + "/" + tableName + "/_gson_schema")); + } + + public long getRowCount(String dbName, String tableName) { + return parseRowCount(execute("api/rowcount?db=" + dbName + "&table=" + tableName)); + } + + public static List parseStringLists(String executeResult) { + ResponseBody> databasesResponse = parseResponse( + new TypeToken>() {}, + executeResult); + if (successResponse(databasesResponse)) { + return databasesResponse.getData(); + } + return new ArrayList<>(); + } + + public static boolean parseSuccessResponse(String executeResult) { + ObjectNode objectNode = JsonUtil.parseObject(executeResult); + Integer code = JsonUtil.safeGetAsInt(objectNode, "code"); + return code != null && code == RestApiStatusCode.OK.code; + } + + public static int parseOnlineBeNum(String executeResult) { + ResponseBody> healthResponse = parseResponse( + new TypeToken>() {}, + executeResult); + if (successResponse(healthResponse)) { + return healthResponse.getData().get(HealthAction.ONLINE_BACKEND_NUM); + } + throw new RuntimeException("get doris table schema error, msg: " + healthResponse.getMsg()); + } + + public static List parseColumns(String executeResult) { + ResponseBody getColumnsResponse = parseResponse( + new TypeToken(){}, + executeResult); + if (successResponse(getColumnsResponse)) { + return getColumnsResponse.getData().getJsonColumns().stream() + .map(json -> GsonUtils.GSON.fromJson(json, Column.class)) + .collect(Collectors.toList()); + } + throw new RuntimeException("get doris table schema error, msg: " + getColumnsResponse.getMsg()); + } + + public static long parseRowCount(String executeResult) { + ResponseBody> rowCountResponse = parseResponse( + new TypeToken>() {}, + executeResult); + if (successResponse(rowCountResponse)) { + return rowCountResponse.getData().values().iterator().next(); + } + throw new RuntimeException("get doris table row count error, msg: " + rowCountResponse.getMsg()); + } + + private void selectNextNode() { + currentNodeIndex++; + currentNodeIndex = currentNodeIndex % feNodes.size(); + currentNode = feNodes.get(currentNodeIndex); + } + + private Response executeResponse(OkHttpClient httpClient, String path) throws IOException { + currentNode = currentNode.trim(); + if (!(currentNode.startsWith("http://") || currentNode.startsWith("https://"))) { + currentNode = "http://" + currentNode; + } + if (!currentNode.endsWith("/")) { + currentNode = currentNode + "/"; + } + + Request.Builder builder = new Request.Builder(); + if (!Strings.isEmpty(authUser)) { + builder.addHeader(HttpHeaders.AUTHORIZATION, + Credentials.basic(authUser, Strings.isEmpty(authPassword) ? "" : authPassword)); + } + Request request = builder.get().url(currentNode + path).build(); + if (LOG.isInfoEnabled()) { + LOG.info("doris rest client request URL: {}", request.url().toString()); + } + return httpClient.newCall(request).execute(); + } + + /** + * execute request for specific path，it will try again nodes.length times if it fails + * + * @param path the path must not leading with '/' + * @return response + */ + protected String execute(String path) { + RuntimeException scratchExceptionForThrow = null; + for (int i = 0; i < retryCount; i++) { + // maybe should add HTTP schema to the address + // actually, at this time we can only process http protocol + // NOTE. currentNode may have some spaces. + // User may set a config like described below: + // hosts: "http://192.168.0.1:8200, http://192.168.0.2:8200" + // then currentNode will be "http://192.168.0.1:8200", " http://192.168.0.2:8200" + if (LOG.isTraceEnabled()) { + LOG.trace("doris rest client request URL: {}", currentNode + "/" + path); + } + try (Response response = executeResponse(httpClient, path)) { + if (response.isSuccessful()) { + return response.body().string(); + } else { + LOG.warn("request response code: {}, body: {}", response.code(), response.message()); + scratchExceptionForThrow = new RuntimeException(response.message()); + } + } catch (IOException e) { + LOG.warn("request node [{}] [{}] failures {}, try next nodes", currentNode, path, e); + scratchExceptionForThrow = new RuntimeException(e.getMessage()); + } + selectNextNode(); + } + LOG.warn("try all nodes [{}], no other nodes left", feNodes); + if (scratchExceptionForThrow != null) { + throw scratchExceptionForThrow; + } + return null; + } + + private OkHttpClient initHttpClient(boolean httpSslEnable, int maxIdleConnections, long keepAliveDurationSec, + long connectTimeoutSec, long readTimeoutSec, long writeTimeoutSec, long callTimeoutSec) { + ConnectionPool connectionPool = new ConnectionPool( + maxIdleConnections, + keepAliveDurationSec, + TimeUnit.SECONDS + ); + OkHttpClient.Builder httpBuilder = new OkHttpClient.Builder() + .connectionPool(connectionPool) + .connectTimeout(connectTimeoutSec, TimeUnit.SECONDS) + .readTimeout(readTimeoutSec, TimeUnit.SECONDS) + .writeTimeout(writeTimeoutSec, TimeUnit.SECONDS) + .callTimeout(callTimeoutSec, TimeUnit.SECONDS); + if (httpSslEnable) { + httpBuilder.sslSocketFactory(createSSLSocketFactory(), new TrustAllCerts()) + .hostnameVerifier(new RemoteDorisRestClient.TrustAllHostnameVerifier()); + } + + return httpBuilder.build(); + } + + /** + * support https + **/ + private static class TrustAllCerts implements X509TrustManager { + public void checkClientTrusted(X509Certificate[] chain, String authType) throws CertificateException { + } + + public void checkServerTrusted(X509Certificate[] chain, String authType) throws CertificateException { + } + + public X509Certificate[] getAcceptedIssuers() { + return new X509Certificate[0]; + } + } + + private static class TrustAllHostnameVerifier implements HostnameVerifier { + public boolean verify(String hostname, SSLSession session) { + return true; + } + } + + private static SSLSocketFactory createSSLSocketFactory() { + SSLSocketFactory ssfFactory; + try { + SSLContext sc = SSLContext.getInstance("TLS"); + sc.init(null, new TrustManager[] {new RemoteDorisRestClient.TrustAllCerts()}, new SecureRandom()); + ssfFactory = sc.getSocketFactory(); + } catch (Exception e) { + throw new RuntimeException("Errors happens when create ssl socket"); + } + return ssfFactory; + } + + private static ResponseBody parseResponse(TypeToken typeToken, String responseBody) { + ResponseBody errorResponseBody = new ResponseBody(); + if (responseBody == null) { + return errorResponseBody.code(RestApiStatusCode.COMMON_ERROR).msg("responseBody is null"); + } + + try { + Type type = TypeToken.getParameterized(ResponseBody.class, typeToken.getType()).getType(); + return GsonUtils.GSON.fromJson(responseBody, type); + } catch (Exception e) { + return errorResponseBody.code(RestApiStatusCode.COMMON_ERROR).msg(e.getMessage()); + } + } + + private static boolean successResponse(ResponseBody responseBody) { + return responseBody != null && responseBody.getCode() == RestApiStatusCode.OK.code; + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisScanNode.java new file mode 100644 index 00000000000000..b35656ae8b3b7a --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisScanNode.java @@ -0,0 +1,327 @@ +// 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. + +package org.apache.doris.datasource.doris.source; + +import org.apache.doris.analysis.BinaryPredicate; +import org.apache.doris.analysis.BoolLiteral; +import org.apache.doris.analysis.DateLiteral; +import org.apache.doris.analysis.Expr; +import org.apache.doris.analysis.ExprSubstitutionMap; +import org.apache.doris.analysis.SlotDescriptor; +import org.apache.doris.analysis.SlotRef; +import org.apache.doris.analysis.TupleDescriptor; +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.TableIf; +import org.apache.doris.common.Pair; +import org.apache.doris.common.UserException; +import org.apache.doris.datasource.FileQueryScanNode; +import org.apache.doris.planner.PlanNodeId; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.SessionVariable; +import org.apache.doris.spi.Split; +import org.apache.doris.statistics.StatisticalType; +import org.apache.doris.thrift.TExplainLevel; +import org.apache.doris.thrift.TFileFormatType; +import org.apache.doris.thrift.TFileRangeDesc; +import org.apache.doris.thrift.TRemoteDorisFileDesc; +import org.apache.doris.thrift.TTableFormatFileDesc; + +import com.google.common.base.Joiner; +import com.google.common.collect.Lists; +import org.apache.arrow.flight.CallOptions; +import org.apache.arrow.flight.FlightClient; +import org.apache.arrow.flight.FlightEndpoint; +import org.apache.arrow.flight.FlightInfo; +import org.apache.arrow.flight.Location; +import org.apache.arrow.flight.grpc.CredentialCallOption; +import org.apache.arrow.flight.sql.FlightSqlClient; +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.net.URI; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +public class RemoteDorisScanNode extends FileQueryScanNode { + private static final Logger LOG = LogManager.getLogger(RemoteDorisScanNode.class); + + public static final String BOOLEAN_TRUE_REPRESENTATION = "1"; + + private final List columns = new ArrayList(); + private final List filters = new ArrayList(); + + private RemoteDorisSource source; + + public RemoteDorisScanNode(PlanNodeId id, TupleDescriptor desc, boolean needCheckColumnPriv, + SessionVariable sv) { + super(id, desc, "REMOTE_DORIS_SCAN_NODE", StatisticalType.REMOTE_DORIS_SCAN_NODE, needCheckColumnPriv, sv); + } + + @Override + protected void doInitialize() throws UserException { + super.doInitialize(); + source = new RemoteDorisSource(desc); + } + + @Override + public List getSplits(int numBackends) throws UserException { + List> locationAndTicketList = executeQuery(); + + return locationAndTicketList.stream() + .map(locationAndTicket -> new RemoteDorisSplit(locationAndTicket.first, locationAndTicket.second)) + .collect(Collectors.toList()); + } + + @Override + protected void setScanParams(TFileRangeDesc rangeDesc, Split split) { + if (split instanceof RemoteDorisSplit) { + RemoteDorisSplit dorisArrowSplit = (RemoteDorisSplit) split; + TRemoteDorisFileDesc fileDesc = new TRemoteDorisFileDesc(); + fileDesc.setIp(source.getHostAndArrowPort().key()); + fileDesc.setArrowPort(source.getHostAndArrowPort().value().toString()); + fileDesc.setTicket(dorisArrowSplit.getTicket()); + fileDesc.setLocationUri(dorisArrowSplit.getLocation()); + fileDesc.setUser(source.getCatalog().getUsername()); + fileDesc.setPassword(source.getCatalog().getPassword()); + + // set TTableFormatFileDesc + TTableFormatFileDesc tableFormatFileDesc = new TTableFormatFileDesc(); + tableFormatFileDesc.setRemoteDorisParams(fileDesc); + tableFormatFileDesc.setTableFormatType(((RemoteDorisSplit) split).getTableFormatType().value()); + + // set TFileRangeDesc + rangeDesc.setTableFormatParams(tableFormatFileDesc); + } + } + + @Override + public String getNodeExplainString(String prefix, TExplainLevel detailLevel) { + StringBuilder output = new StringBuilder(); + + output.append(prefix).append("TABLE: ").append(source.getTargetTable().getExternalTableName()).append("\n"); + if (detailLevel == TExplainLevel.BRIEF) { + return output.toString(); + } + output.append(prefix).append("QUERY: ").append(getQueryStr()).append("\n"); + if (!conjuncts.isEmpty()) { + Expr expr = convertConjunctsToAndCompoundPredicate(conjuncts); + output.append(prefix).append("PREDICATES: ").append(expr.toSql()).append("\n"); + } + + return output.toString(); + } + + @Override + protected TFileFormatType getFileFormatType() throws UserException { + return TFileFormatType.FORMAT_ARROW; + } + + @Override + protected List getPathPartitionKeys() throws UserException { + return new ArrayList<>(); + } + + @Override + protected TableIf getTargetTable() throws UserException { + return desc.getTable(); + } + + @Override + protected Map getLocationProperties() throws UserException { + return source.getCatalog().getProperties(); + } + + // Executes a SQL query using the Apache Arrow Flight SQL protocol with the provided credentials. + private List> executeQuery() { + createColumns(); + createFilters(); + + if (isExplainStatement()) { + return new ArrayList<>(); + } + + String queryStr = getQueryStr(); + Exception lastException = null; + + for (int i = 0; i < source.getCatalog().getQueryRetryCount(); i++) { + try { + return executeFlightSqlQuery( + source.nextHostAndArrowPort(), + source.getCatalog().getUsername(), + source.getCatalog().getPassword(), + queryStr + ); + } catch (Exception e) { + LOG.warn("arrow request node [{}] failures {}, try next nodes", + source.getHostAndArrowPort().toString(), e); + lastException = new RuntimeException(e.getMessage()); + } + } + + throw new RuntimeException("Failed to execute query: " + queryStr, lastException); + } + + private List> executeFlightSqlQuery(Pair hostAndPort, + String user, String psw, String sql) throws Exception { + try ( + BufferAllocator allocatorFE = new RootAllocator(); + FlightClient clientFE = createFlightClient(allocatorFE, hostAndPort); + FlightSqlClient sqlClientFE = new FlightSqlClient(clientFE) + ) { + CredentialCallOption credentialCallOption = authenticate(clientFE, user, psw); + FlightInfo info = executeSqlWithTimeout(sqlClientFE, sql, credentialCallOption); + + return processFlightEndpoints(info.getEndpoints()); + } + } + + private void createColumns() { + columns.clear(); + for (SlotDescriptor slot : desc.getSlots()) { + if (!slot.isMaterialized()) { + continue; + } + Column col = slot.getColumn(); + columns.add("`" + col.getName() + "`"); + } + if (columns.isEmpty()) { + columns.add("*"); + } + } + + private String getQueryStr() { + StringBuilder sql = new StringBuilder("SELECT "); + + if (source.getCatalog().enableParallelResultSink()) { + sql.append("/*+ SET_VAR(enable_parallel_result_sink=true) */ "); + } else { + sql.append("/*+ SET_VAR(enable_parallel_result_sink=false) */ "); + } + + sql.append(Joiner.on(", ").join(columns)); + + sql.append(" FROM ").append(source.getTargetTable().getExternalTableName()); + + if (!filters.isEmpty()) { + sql.append(" WHERE ("); + sql.append(Joiner.on(") AND (").join(filters)); + sql.append(")"); + } + + if (limit != -1) { + sql.append(" LIMIT ").append(limit); + } + + return sql.toString(); + } + + private void createFilters() { + if (conjuncts.isEmpty()) { + return; + } + + List slotRefs = Lists.newArrayList(); + Expr.collectList(conjuncts, SlotRef.class, slotRefs); + ExprSubstitutionMap sMap = new ExprSubstitutionMap(); + for (SlotRef slotRef : slotRefs) { + SlotRef slotRef1 = (SlotRef) slotRef.clone(); + slotRef1.setTblName(null); + slotRef1.setLabel("`" + slotRef1.getColumnName() + "`"); + sMap.put(slotRef, slotRef1); + } + + ArrayList conjunctsList = Expr.cloneList(conjuncts, sMap); + for (Expr expr : conjunctsList) { + String filter = conjunctExprToString(expr, desc.getTable()); + filters.add(filter); + } + } + + private String conjunctExprToString(Expr expr, TableIf tbl) { + if (expr.contains(DateLiteral.class) && expr instanceof BinaryPredicate) { + ArrayList children = expr.getChildren(); + String filter = children.get(0).toExternalSql(TableIf.TableType.DORIS_EXTERNAL_TABLE, tbl); + filter += " " + ((BinaryPredicate) expr).getOp().toString() + " "; + + filter += children.get(1).toExternalSql(TableIf.TableType.DORIS_EXTERNAL_TABLE, tbl); + + return filter; + } + + // Only for old planner + if (expr.contains(BoolLiteral.class) && BOOLEAN_TRUE_REPRESENTATION.equals(expr.getStringValue()) + && expr.getChildren().isEmpty()) { + return "1 = 1"; + } + + return expr.toExternalSql(TableIf.TableType.DORIS_EXTERNAL_TABLE, tbl); + } + + // TODO: Use AST parsing instead of string matching for EXPLAIN detection + private boolean isExplainStatement() { + return ConnectContext.get().getStatementContext().getOriginStatement().originStmt + .trim().toLowerCase().startsWith("explain"); + } + + private FlightClient createFlightClient(BufferAllocator allocator, + Pair hostAndPort) throws Exception { + URI uri = new URI("grpc", null, hostAndPort.first, hostAndPort.second, null, null, null); + return FlightClient.builder(allocator, new Location(uri)).build(); + } + + private CredentialCallOption authenticate(FlightClient client, String user, String psw) throws UserException { + Optional credentialCallOption = client.authenticateBasicToken(user, psw); + if (!credentialCallOption.isPresent()) { + throw new UserException("Authenticates with a username and password failure"); + } + return credentialCallOption.get(); + } + + private FlightInfo executeSqlWithTimeout(FlightSqlClient sqlClient, String sql, + CredentialCallOption credentialCallOption) { + int timeoutSec = source.getCatalog().getQueryTimeoutSec(); + return sqlClient.execute(sql, credentialCallOption, + CallOptions.timeout(timeoutSec, TimeUnit.SECONDS)); + } + + private List> processFlightEndpoints(List endpoints) { + List> uniquePairs = new ArrayList<>(); + Set seenPairs = new HashSet<>(); + for (FlightEndpoint endpoint : endpoints) { + ByteBuffer ticket = endpoint.getTicket().serialize(); + for (Location location : endpoint.getLocations()) { + String uri = location.getUri().toString(); + String compositeKey = ticket.hashCode() + uri; + + if (seenPairs.add(compositeKey)) { + uniquePairs.add(Pair.of(uri, ticket)); + } + } + } + return uniquePairs; + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisSource.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisSource.java new file mode 100644 index 00000000000000..aeaf795b841da0 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisSource.java @@ -0,0 +1,88 @@ +// 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. + +package org.apache.doris.datasource.doris.source; + +import org.apache.doris.analysis.TupleDescriptor; +import org.apache.doris.common.Pair; +import org.apache.doris.datasource.doris.RemoteDorisExternalCatalog; +import org.apache.doris.datasource.doris.RemoteDorisExternalTable; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ThreadLocalRandom; + +public class RemoteDorisSource { + private final TupleDescriptor desc; + private final RemoteDorisExternalCatalog remoteDorisExternalCatalog; + private final RemoteDorisExternalTable remoteDorisExtTable; + private final List> hostsAndArrowPort; + private int currentHostIndex; + private Pair currentHostAndArrowPort; + + public RemoteDorisSource(TupleDescriptor desc) { + this.desc = desc; + this.remoteDorisExtTable = (RemoteDorisExternalTable) desc.getTable(); + this.remoteDorisExternalCatalog = (RemoteDorisExternalCatalog) remoteDorisExtTable.getCatalog(); + this.hostsAndArrowPort = parseArrowNodes(remoteDorisExternalCatalog.getFeArrowNodes()); + this.currentHostIndex = ThreadLocalRandom.current().nextInt(hostsAndArrowPort.size()); + } + + public TupleDescriptor getDesc() { + return desc; + } + + public RemoteDorisExternalTable getTargetTable() { + return remoteDorisExtTable; + } + + public RemoteDorisExternalCatalog getCatalog() { + return remoteDorisExternalCatalog; + } + + public Pair nextHostAndArrowPort() { + return nextHostAndPort(); + } + + public Pair getHostAndArrowPort() { + return currentHostAndArrowPort; + } + + private List> parseArrowNodes(List feArrowNodes) { + if (feArrowNodes == null || feArrowNodes.isEmpty()) { + throw new RuntimeException("fe arrow nodes not set"); + } + + List> hostsAndArrowPort = new ArrayList<>(); + for (String feArrowNode : feArrowNodes) { + String[] split = feArrowNode.split(":"); + if (split.length != 2) { + throw new RuntimeException("fe arrow nodes format error, must ip:arrow_port,ip:arrow_port.."); + } + hostsAndArrowPort.add(Pair.of(split[0].trim(), Integer.parseInt(split[1].trim()))); + } + + return hostsAndArrowPort; + } + + private Pair nextHostAndPort() { + currentHostAndArrowPort = this.hostsAndArrowPort.get(currentHostIndex); + currentHostIndex++; + currentHostIndex = currentHostIndex % hostsAndArrowPort.size(); + return currentHostAndArrowPort; + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisSplit.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisSplit.java new file mode 100644 index 00000000000000..86921231cdd236 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/source/RemoteDorisSplit.java @@ -0,0 +1,53 @@ +// 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. + +package org.apache.doris.datasource.doris.source; + +import org.apache.doris.common.util.LocationPath; +import org.apache.doris.datasource.FileSplit; +import org.apache.doris.datasource.TableFormatType; + +import java.nio.ByteBuffer; + +public class RemoteDorisSplit extends FileSplit { + private static final LocationPath DUMMY_PATH = LocationPath.of("/dummyPath"); + private final String location; + private final ByteBuffer ticket; + + public RemoteDorisSplit(String location, ByteBuffer ticket) { + super(DUMMY_PATH, 0, 0, 0, 0, null, null); + this.location = location; + this.ticket = ticket; + this.tableFormatType = TableFormatType.REMOTE_DORIS; + } + + public ByteBuffer getTicket() { + return ticket; + } + + public String getLocation() { + return location; + } + + public TableFormatType getTableFormatType() { + return tableFormatType; + } + + public void setTableFormatType(TableFormatType tableFormatType) { + this.tableFormatType = tableFormatType; + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/constants/RemoteDorisProperties.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/constants/RemoteDorisProperties.java new file mode 100644 index 00000000000000..c6bceed94cb3a4 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/constants/RemoteDorisProperties.java @@ -0,0 +1,50 @@ +// 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. + +package org.apache.doris.datasource.property.constants; + +public class RemoteDorisProperties { + public static final String FE_HTTP_HOSTS = "fe_http_hosts"; + public static final String FE_ARROW_HOSTS = "fe_arrow_hosts"; + + public static final String USER = "user"; + public static final String PASSWORD = "password"; + + public static final String ENABLE_PARALLEL_RESULT_SINK = "enable_parallel_result_sink"; + + // Supports older versions of remote Doris; enabling this may introduce some inaccuracies in schema parsing. + public static final String COMPATIBLE = "compatible"; + + /** + * For Arrow Flight query. + **/ + public static final String QUERY_RETRY_COUNT = "query_retry_count"; + // Query execution is asynchronous on the server; the client does not wait for completion. + public static final String QUERY_TIMEOUT_SEC = "query_timeout_sec"; + + /** + * For metadata HTTP synchronization. + **/ + public static final String METADATA_HTTP_SSL_ENABLED = "metadata_http_ssl_enabled"; + public static final String METADATA_SYNC_RETRIES_COUNT = "metadata_sync_retry_count"; + public static final String METADATA_MAX_IDLE_CONNECTIONS = "metadata_max_idle_connections"; + public static final String METADATA_KEEP_ALIVE_DURATION_SEC = "metadata_keep_alive_duration_sec"; + public static final String METADATA_CONNECT_TIMEOUT_SEC = "metadata_connect_timeout_sec"; + public static final String METADATA_READ_TIMEOUT_SEC = "metadata_read_timeout_sec"; + public static final String METADATA_WRITE_TIMEOUT_SEC = "metadata_write_timeout_sec"; + public static final String METADATA_CALL_TIMEOUT_SEC = "metadata_call_timeout_sec"; +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/HealthAction.java b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/HealthAction.java index fe27402f9a9d94..39cd23d16af839 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/HealthAction.java +++ b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/HealthAction.java @@ -32,6 +32,8 @@ @RestController public class HealthAction extends RestBaseController { + public static final String TOTAL_BACKEND_NUM = "total_backend_num"; + public static final String ONLINE_BACKEND_NUM = "online_backend_num"; @RequestMapping(path = "/api/health", method = RequestMethod.GET) public Object execute(HttpServletRequest request, HttpServletResponse response) { @@ -40,8 +42,8 @@ public Object execute(HttpServletRequest request, HttpServletResponse response) } Map result = new HashMap<>(); - result.put("total_backend_num", Env.getCurrentSystemInfo().getAllBackendIds(false).size()); - result.put("online_backend_num", Env.getCurrentSystemInfo().getAllBackendIds(true).size()); + result.put(TOTAL_BACKEND_NUM, Env.getCurrentSystemInfo().getAllBackendIds(false).size()); + result.put(ONLINE_BACKEND_NUM, Env.getCurrentSystemInfo().getAllBackendIds(true).size()); return ResponseEntityBuilder.ok(result); } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/TableSchemaAction.java b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/TableSchemaAction.java index a3196ad5d3b872..a3d766f1c6fd7b 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/TableSchemaAction.java +++ b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/TableSchemaAction.java @@ -21,6 +21,7 @@ import org.apache.doris.catalog.Database; import org.apache.doris.catalog.DatabaseIf; import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.KeysType; import org.apache.doris.catalog.MaterializedIndexMeta; import org.apache.doris.catalog.OlapTable; import org.apache.doris.catalog.PrimitiveType; @@ -34,7 +35,9 @@ import org.apache.doris.datasource.CatalogIf; import org.apache.doris.datasource.InternalCatalog; import org.apache.doris.httpv2.entity.ResponseEntityBuilder; +import org.apache.doris.httpv2.rest.response.GsonSchemaResponse; import org.apache.doris.mysql.privilege.PrivPredicate; +import org.apache.doris.persist.gson.GsonUtils; import org.apache.doris.qe.ConnectContext; import com.google.gson.Gson; @@ -54,6 +57,7 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.stream.Collectors; /** * Get table schema for specified cluster.database.table with privilege checking @@ -240,4 +244,52 @@ public Object columnChangeCanSync( } return ResponseEntityBuilder.ok(); } + + @RequestMapping(path = {"/api/{" + DB_KEY + "}/{" + TABLE_KEY + "}/_gson_schema", + "/api/{" + CATALOG_KEY + "}/{" + DB_KEY + "}/{" + TABLE_KEY + "}/_gson_schema"}, method = RequestMethod.GET) + protected Object gsonSchema( + @PathVariable(value = CATALOG_KEY, required = false) String catalogName, + @PathVariable(value = DB_KEY) final String dbName, + @PathVariable(value = TABLE_KEY) final String tblName, + HttpServletRequest request, HttpServletResponse response) { + executeCheckPassword(request, response); + GsonSchemaResponse gsonSchemaResponse = new GsonSchemaResponse(); + if (StringUtils.isBlank(catalogName)) { + catalogName = InternalCatalog.INTERNAL_CATALOG_NAME; + } + + try { + String fullDbName = getFullDbName(dbName); + checkTblAuth(ConnectContext.get().getCurrentUserIdentity(), catalogName, fullDbName, tblName, + PrivPredicate.SELECT); + TableIf table; + try { + CatalogIf catalog = StringUtils.isNotBlank(catalogName) ? Env.getCurrentEnv().getCatalogMgr() + .getCatalogOrAnalysisException(catalogName) : Env.getCurrentInternalCatalog(); + DatabaseIf db = catalog.getDbOrMetaException(fullDbName); + table = db.getTableOrMetaException(tblName); + } catch (MetaNotFoundException | AnalysisException e) { + return ResponseEntityBuilder.okWithCommonError(e.getMessage()); + } + table.readLock(); + try { + List jsonColumns = table.getBaseSchema().stream() + .map(GsonUtils.GSON::toJson) + .collect(Collectors.toList()); + gsonSchemaResponse.setJsonColumns(jsonColumns); + if (table instanceof OlapTable) { + gsonSchemaResponse.setKeysType(((OlapTable) table).getKeysType()); + } else { + gsonSchemaResponse.setKeysType(KeysType.UNKNOWN); + } + } finally { + table.readUnlock(); + } + } catch (Exception e) { + return ResponseEntityBuilder.okWithCommonError(e.getMessage()); + } + + return ResponseEntityBuilder.ok(gsonSchemaResponse); + } + } diff --git a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/response/GsonSchemaResponse.java b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/response/GsonSchemaResponse.java new file mode 100644 index 00000000000000..6457a4d7da1926 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/response/GsonSchemaResponse.java @@ -0,0 +1,30 @@ +// 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. + +package org.apache.doris.httpv2.rest.response; + +import org.apache.doris.catalog.KeysType; + +import lombok.Data; + +import java.util.List; + +@Data +public class GsonSchemaResponse { + List jsonColumns; + KeysType keysType; +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java index 389e89c4f47149..a6a2d6cc08cdc9 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java @@ -51,6 +51,8 @@ import org.apache.doris.common.Pair; import org.apache.doris.datasource.ExternalTable; import org.apache.doris.datasource.FileQueryScanNode; +import org.apache.doris.datasource.doris.RemoteDorisExternalTable; +import org.apache.doris.datasource.doris.source.RemoteDorisScanNode; import org.apache.doris.datasource.es.EsExternalTable; import org.apache.doris.datasource.es.source.EsScanNode; import org.apache.doris.datasource.hive.HMSExternalTable; @@ -660,6 +662,8 @@ public PlanFragment visitPhysicalFileScan(PhysicalFileScan fileScan, PlanTransla fileScan.getSelectedPartitions(), false, sv); } else if (table instanceof LakeSoulExternalTable) { scanNode = new LakeSoulScanNode(context.nextPlanNodeId(), tupleDescriptor, false, sv); + } else if (table instanceof RemoteDorisExternalTable) { + scanNode = new RemoteDorisScanNode(context.nextPlanNodeId(), tupleDescriptor, false, sv); } else { throw new RuntimeException("do not support table type " + table.getType()); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java index d7d4c5f2ce067d..2dbfb0f3e42ecf 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java @@ -469,6 +469,7 @@ private LogicalPlan getLogicalPlan(TableIf table, UnboundRelation unboundRelatio case MAX_COMPUTE_EXTERNAL_TABLE: case TRINO_CONNECTOR_EXTERNAL_TABLE: case LAKESOUl_EXTERNAL_TABLE: + case DORIS_EXTERNAL_TABLE: return new LogicalFileScan(unboundRelation.getRelationId(), (ExternalTable) table, qualifierWithoutTableName, ImmutableList.of(), unboundRelation.getTableSample(), diff --git a/fe/fe-core/src/main/java/org/apache/doris/persist/gson/GsonUtils.java b/fe/fe-core/src/main/java/org/apache/doris/persist/gson/GsonUtils.java index 7a97a55a579a70..d863140a2569db 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/persist/gson/GsonUtils.java +++ b/fe/fe-core/src/main/java/org/apache/doris/persist/gson/GsonUtils.java @@ -131,6 +131,7 @@ import org.apache.doris.datasource.ExternalDatabase; import org.apache.doris.datasource.ExternalTable; import org.apache.doris.datasource.InternalCatalog; +import org.apache.doris.datasource.doris.RemoteDorisExternalCatalog; import org.apache.doris.datasource.es.EsExternalCatalog; import org.apache.doris.datasource.es.EsExternalDatabase; import org.apache.doris.datasource.es.EsExternalTable; @@ -427,7 +428,8 @@ public class GsonUtils { TrinoConnectorExternalCatalog.class, TrinoConnectorExternalCatalog.class.getSimpleName()) .registerSubtype(LakeSoulExternalCatalog.class, LakeSoulExternalCatalog.class.getSimpleName()) .registerSubtype(TestExternalCatalog.class, TestExternalCatalog.class.getSimpleName()) - .registerSubtype(PaimonDLFExternalCatalog.class, PaimonDLFExternalCatalog.class.getSimpleName()); + .registerSubtype(PaimonDLFExternalCatalog.class, PaimonDLFExternalCatalog.class.getSimpleName()) + .registerSubtype(RemoteDorisExternalCatalog.class, RemoteDorisExternalCatalog.class.getSimpleName()); if (Config.isNotCloudMode()) { dsTypeAdapterFactory .registerSubtype(InternalCatalog.class, InternalCatalog.class.getSimpleName()); diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/DeriveFactory.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/DeriveFactory.java index 9dd2fdc4f285da..f7dd33e1fc2160 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/statistics/DeriveFactory.java +++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/DeriveFactory.java @@ -52,6 +52,7 @@ public BaseStatsDerive getStatsDerive(StatisticalType statisticalType) { case HIVE_SCAN_NODE: case ICEBERG_SCAN_NODE: case LAKESOUL_SCAN_NODE: + case REMOTE_DORIS_SCAN_NODE: case PAIMON_SCAN_NODE: case INTERSECT_NODE: case SCHEMA_SCAN_NODE: diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticalType.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticalType.java index 42a930c2471863..0ec7a518078fed 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticalType.java +++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticalType.java @@ -57,5 +57,6 @@ public enum StatisticalType { TEST_EXTERNAL_TABLE, GROUP_COMMIT_SCAN_NODE, TRINO_CONNECTOR_SCAN_NODE, - LAKESOUL_SCAN_NODE + LAKESOUL_SCAN_NODE, + REMOTE_DORIS_SCAN_NODE } diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/doris/RemoteDorisCompatibleRestClientTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/doris/RemoteDorisCompatibleRestClientTest.java new file mode 100644 index 00000000000000..58423e6ac2759c --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/doris/RemoteDorisCompatibleRestClientTest.java @@ -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. + +package org.apache.doris.datasource.doris; + +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.PrimitiveType; +import org.apache.doris.common.util.JsonUtil; +import org.apache.doris.http.DorisHttpTestCase; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import okhttp3.Request; +import okhttp3.Response; +import org.junit.Assert; +import org.junit.Test; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +public class RemoteDorisCompatibleRestClientTest extends DorisHttpTestCase { + + @Test + public void testGetColumns() throws Exception { + String executeRes = execute("api/" + DB_NAME + "/" + TABLE_NAME + "/_schema"); + RemoteDorisCompatibleRestClient.DorisApiResponse tableSchemaResponse = RemoteDorisCompatibleRestClient.parseResponse( + executeRes, + "get doris table schema error"); + ObjectNode objectNode = JsonUtil.parseObject(tableSchemaResponse.getData()); + JsonNode properties = objectNode.path("properties"); + List res = new ArrayList<>(); + for (JsonNode columnJson : properties) { + if (columnJson.isObject()) { + res.add(RemoteDorisCompatibleRestClient.parseColumn((ObjectNode) columnJson)); + } + } + + Column k1 = new Column("k1", PrimitiveType.BIGINT); + Column k2 = new Column("k2", PrimitiveType.DOUBLE); + List columns = new ArrayList<>(); + columns.add(k1); + columns.add(k2); + + Assert.assertArrayEquals(columns.toArray(), res.toArray()); + } + + private String execute(String url) throws IOException { + Request request = new Request.Builder() + .get() + .addHeader("Authorization", rootAuth) + .addHeader("forward_master_ut_test", "true") + .url("http://localhost:" + HTTP_PORT + "/" + url) + .build(); + Response response = networkClient.newCall(request).execute(); + return response.body().string(); + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/doris/RemoteDorisRestClientTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/doris/RemoteDorisRestClientTest.java new file mode 100644 index 00000000000000..723fc617f9345c --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/doris/RemoteDorisRestClientTest.java @@ -0,0 +1,108 @@ +// 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. + +package org.apache.doris.datasource.doris; + +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.PrimitiveType; +import org.apache.doris.http.DorisHttpTestCase; + +import okhttp3.Request; +import okhttp3.Response; +import org.junit.Assert; +import org.junit.Test; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +public class RemoteDorisRestClientTest extends DorisHttpTestCase { + @Test + public void testGetDatabaseNameList() throws Exception { + List res = RemoteDorisRestClient.parseStringLists( + execute("api/meta/namespaces/default_cluster/databases")); + Assert.assertArrayEquals(new String[]{DB_NAME}, res.toArray()); + } + + @Test + public void testGetTablesNameList() throws Exception { + List res = RemoteDorisRestClient.parseStringLists( + execute("api/meta/namespaces/default_cluster/databases/" + DB_NAME + "/tables")); + Assert.assertArrayEquals(new String[]{"es_table", "testTbl1"}, res.toArray()); + } + + @Test + public void testGetTablesNameListByErrorDb() throws Exception { + List res = RemoteDorisRestClient.parseStringLists( + execute("api/meta/namespaces/default_cluster/databases/not_" + DB_NAME + "/tables")); + Assert.assertEquals(0, res.size()); + } + + @Test + public void testTableExist() throws Exception { + boolean res = RemoteDorisRestClient.parseSuccessResponse( + execute("api/" + DB_NAME + "/" + TABLE_NAME + "/_schema")); + Assert.assertTrue(res); + } + + @Test + public void testTableNotExist() throws Exception { + boolean res = RemoteDorisRestClient.parseSuccessResponse( + execute("api/" + DB_NAME + "/not_" + TABLE_NAME + "/_schema")); + Assert.assertFalse(res); + } + + @Test + public void testHealth() throws Exception { + int res = RemoteDorisRestClient.parseOnlineBeNum( + execute("api/health")); + Assert.assertEquals(3, res); + } + + @Test + public void testGetColumns() throws Exception { + List res = RemoteDorisRestClient.parseColumns( + execute("api/" + DB_NAME + "/" + TABLE_NAME + "/_gson_schema")); + + Column k1 = new Column("k1", PrimitiveType.BIGINT); + Column k2 = new Column("k2", PrimitiveType.DOUBLE); + List columns = new ArrayList<>(); + columns.add(k1); + columns.add(k2); + + Assert.assertArrayEquals(columns.toArray(), res.toArray()); + } + + @Test + public void testGetRowCount() throws Exception { + long res = RemoteDorisRestClient.parseRowCount( + execute("api/rowcount?db=" + DB_NAME + "&table=" + TABLE_NAME)); + + Assert.assertEquals(2000L, res); + } + + private String execute(String url) throws IOException { + Request request = new Request.Builder() + .get() + .addHeader("Authorization", rootAuth) + .addHeader("forward_master_ut_test", "true") + .url("http://localhost:" + HTTP_PORT + "/" + url) + .build(); + Response response = networkClient.newCall(request).execute(); + return response.body().string(); + } +} diff --git a/gensrc/thrift/Descriptors.thrift b/gensrc/thrift/Descriptors.thrift index e291724e8f34b0..eca2ef59d9ec20 100644 --- a/gensrc/thrift/Descriptors.thrift +++ b/gensrc/thrift/Descriptors.thrift @@ -412,6 +412,12 @@ struct TLakeSoulTable { struct TDictionaryTable { } +struct TRemoteDorisTable { + 1: optional string db_name + 2: optional string table_name + 3: optional map properties +} + // "Union" of all table types. struct TTableDescriptor { 1: required Types.TTableId id @@ -438,6 +444,7 @@ struct TTableDescriptor { 22: optional TTrinoConnectorTable trinoConnectorTable 23: optional TLakeSoulTable lakesoulTable 24: optional TDictionaryTable dictionaryTable + 25: optional TRemoteDorisTable remoteDorisTable } struct TDescriptorTable { diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift index 9be45d70b47666..b35b711f427035 100644 --- a/gensrc/thrift/PlanNodes.thrift +++ b/gensrc/thrift/PlanNodes.thrift @@ -389,6 +389,15 @@ struct TTransactionalHiveDesc { 2: optional list delete_deltas } +struct TRemoteDorisFileDesc { + 1: optional string ip + 2: optional string arrow_port + 3: optional binary ticket + 4: optional string location_uri + 5: optional string user + 6: optional string password +} + struct TTableFormatFileDesc { 1: optional string table_format_type 2: optional TIcebergFileDesc iceberg_params @@ -399,6 +408,7 @@ struct TTableFormatFileDesc { 7: optional TTrinoConnectorFileDesc trino_connector_params 8: optional TLakeSoulFileDesc lakesoul_params 9: optional i64 table_level_row_count = -1 + 10: optional TRemoteDorisFileDesc remote_doris_params } // Deprecated, hive text talbe is a special format, not a serde type diff --git a/gensrc/thrift/Types.thrift b/gensrc/thrift/Types.thrift index 34d222087af945..62613e4c760835 100644 --- a/gensrc/thrift/Types.thrift +++ b/gensrc/thrift/Types.thrift @@ -646,7 +646,8 @@ enum TTableType { MAX_COMPUTE_TABLE = 12, LAKESOUL_TABLE = 13, TRINO_CONNECTOR_TABLE = 14, - DICTIONARY_TABLE = 15 + DICTIONARY_TABLE = 15, + REMOTE_DORIS_TABLE = 16 } enum TKeysType { diff --git a/regression-test/conf/regression-conf.groovy b/regression-test/conf/regression-conf.groovy index e1d1bfd7561d0c..a984c24b937763 100644 --- a/regression-test/conf/regression-conf.groovy +++ b/regression-test/conf/regression-conf.groovy @@ -227,6 +227,7 @@ extArrowFlightSqlHost = "127.0.0.1" extArrowFlightSqlPort = 8081 extArrowFlightSqlUser = "root" extArrowFlightSqlPassword= "" +extArrowFlightHttpPort= 8030 // iceberg rest catalog config iceberg_rest_uri_port=18181 diff --git a/regression-test/data/external_table_p0/remote_doris/test_remote_doris_all_types_select.out b/regression-test/data/external_table_p0/remote_doris/test_remote_doris_all_types_select.out new file mode 100644 index 00000000000000..069600b28c1341 --- /dev/null +++ b/regression-test/data/external_table_p0/remote_doris/test_remote_doris_all_types_select.out @@ -0,0 +1,16 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !sql -- +2025-05-18T01:00 true -128 -32768 -2147483648 -9223372036854775808 -1234567890123456790 -123.456 -123456.789 -123457 -123456789012346 -1234567890123456789012345678 1970-01-01 A Hello Hello, Doris! ["apple", "banana", "orange"] {"Emily":101, "age":25} {"f1":11, "f2":3.14, "f3":"Emily"} +2025-05-18T02:00 \N \N \N \N \N \N \N \N \N \N \N \N \N \N \N \N \N \N \N +2025-05-18T03:00 false 127 32767 2147483647 9223372036854775807 1234567890123456789 123.456 123456.789 123457 123456789012346 1234567890123456789012345678 9999-12-31 9999-12-31T23:59:59 [] {} {"f1":11, "f2":3.14, "f3":"Emily"} +2025-05-18T04:00 true 0 0 0 0 0 0.0 0.0 0 0 0 2023-10-01 2023-10-01T12:34:56 A Hello Hello, Doris! ["apple", "banana", "orange"] {"Emily":101, "age":25} {"f1":11, "f2":3.14, "f3":"Emily"} + +-- !sql -- +2025-05-18T01:00 [1] [-128] [-32768] [-2147483648] [-9223372036854775808] [-1234567890123456790] [-123.456] [-123456.789] [-123457] [-123456789012346] [-1234567890123456789012345678] ["0000-01-01"] [""] ["A"] ["Hello"] ["Hello, Doris!"] +2025-05-18T02:00 [null] [null] [null] [null] [null] [null] [null] [null] [null] [null] [null] [null] [null] [null] [null] [null] +2025-05-18T03:00 [0] [127] [32767] [2147483647] [9223372036854775807] [1234567890123456789] [123.456] [123456.789] [123457] [123456789012346] [1234567890123456789012345678] ["9999-12-31"] ["9999-12-31 23:59:59"] [""] [""] [""] +2025-05-18T04:00 [1] [0] [0] [0] [0] [0] [0] [0] [0] [0] [0] ["2023-10-01"] ["2023-10-01 12:34:56"] ["A"] ["Hello"] ["Hello, Doris!"] + +-- !sql -- +2025-05-18T01:00 2025-05-18T01:00 2025-05-18T01:00:00.100 2025-05-18T01:00:00.110 2025-05-18T01:00:00.111 2025-05-18T01:00:00.111100 2025-05-18T01:00:00.111110 2025-05-18T01:00:00.111111 + diff --git a/regression-test/data/external_table_p0/remote_doris/test_remote_doris_all_types_show.out b/regression-test/data/external_table_p0/remote_doris/test_remote_doris_all_types_show.out new file mode 100644 index 00000000000000..5133dffcb91100 --- /dev/null +++ b/regression-test/data/external_table_p0/remote_doris/test_remote_doris_all_types_show.out @@ -0,0 +1,76 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !sql -- +test_remote_doris_all_types_t1 CREATE TABLE `test_remote_doris_all_types_t1` (\n `id` datetimev2(3) NOT NULL,\n `c_boolean` boolean NULL DEFAULT "true",\n `c_tinyint` tinyint NULL DEFAULT "1",\n `c_smallint` smallint NULL DEFAULT "1",\n `c_int` int NULL DEFAULT "1",\n `c_bigint` bigint NULL DEFAULT "1",\n `c_largeint` largeint NULL DEFAULT "1",\n `c_float` float NULL DEFAULT "1",\n `c_double` double NULL DEFAULT "1",\n `c_decimal9` decimalv3(9,0) NULL DEFAULT "1",\n `c_decimal18` decimalv3(18,0) NULL DEFAULT "1",\n `c_decimal32` decimalv3(32,0) NULL DEFAULT "1",\n `c_date` datev2 NULL DEFAULT "2025-08-18",\n `c_datetime` datetimev2(0) NULL DEFAULT "2025-08-18 20:00:00",\n `c_char` char(1) NULL DEFAULT "d",\n `c_varchar` varchar(65533) NULL DEFAULT "d",\n `c_string` text NULL DEFAULT "d",\n `c_array_s` array NULL,\n `c_map` map NULL,\n `c_struct` struct NULL\n) ENGINE=DORIS_EXTERNAL_TABLE; + +-- !sql -- +id datetime(3) No true \N +c_boolean boolean Yes false true NONE +c_tinyint tinyint Yes false 1 NONE +c_smallint smallint Yes false 1 NONE +c_int int Yes false 1 NONE +c_bigint bigint Yes false 1 NONE +c_largeint largeint Yes false 1 NONE +c_float float Yes false 1 NONE +c_double double Yes false 1 NONE +c_decimal9 decimal(9,0) Yes false 1 NONE +c_decimal18 decimal(18,0) Yes false 1 NONE +c_decimal32 decimal(32,0) Yes false 1 NONE +c_date date Yes false 2025-08-18 NONE +c_datetime datetime Yes false 2025-08-18 20:00:00 NONE +c_char char(1) Yes false d NONE +c_varchar varchar(65533) Yes false d NONE +c_string text Yes false d NONE +c_array_s array Yes false \N NONE +c_map map Yes false \N NONE +c_struct struct Yes false \N NONE + +-- !sql -- +test_remote_doris_all_types_t2 CREATE TABLE `test_remote_doris_all_types_t2` (\n `id` datetimev2(3) NOT NULL,\n `a_boolean` array NULL,\n `a_tinyint` array NULL,\n `a_smallint` array NULL,\n `a_int` array NULL,\n `a_bigint` array NULL,\n `a_largeint` array NULL,\n `a_float` array NULL,\n `a_double` array NULL,\n `a_decimal9` array NULL,\n `a_decimal18` array NULL,\n `a_decimal32` array NULL,\n `a_date` array NULL,\n `a_datetime` array NULL,\n `a_char` array NULL,\n `a_varchar` array NULL,\n `a_string` array NULL\n) ENGINE=DORIS_EXTERNAL_TABLE; + +-- !sql -- +id datetime(3) No true \N +a_boolean array Yes false \N NONE +a_tinyint array Yes false \N NONE +a_smallint array Yes false \N NONE +a_int array Yes false \N NONE +a_bigint array Yes false \N NONE +a_largeint array Yes false \N NONE +a_float array Yes false \N NONE +a_double array Yes false \N NONE +a_decimal9 array Yes false \N NONE +a_decimal18 array Yes false \N NONE +a_decimal32 array Yes false \N NONE +a_date array Yes false \N NONE +a_datetime array Yes false \N NONE +a_char array Yes false \N NONE +a_varchar array Yes false \N NONE +a_string array Yes false \N NONE + +-- !sql -- +test_remote_doris_all_types_t3 CREATE TABLE `test_remote_doris_all_types_t3` (\n `id` datetimev2(0) NOT NULL,\n `datetime_0` datetimev2(0) NULL,\n `datetime_1` datetimev2(1) NULL,\n `datetime_3` datetimev2(2) NULL,\n `datetime_4` datetimev2(3) NULL,\n `datetime_5` datetimev2(4) NULL,\n `datetime_6` datetimev2(5) NULL,\n `datetime_7` datetimev2(6) NULL\n) ENGINE=DORIS_EXTERNAL_TABLE; + +-- !sql -- +id datetime No true \N +datetime_0 datetime Yes false \N NONE +datetime_1 datetime(1) Yes false \N NONE +datetime_3 datetime(2) Yes false \N NONE +datetime_4 datetime(3) Yes false \N NONE +datetime_5 datetime(4) Yes false \N NONE +datetime_6 datetime(5) Yes false \N NONE +datetime_7 datetime(6) Yes false \N NONE + +-- !sql -- +test_remote_doris_all_types_t4 CREATE TABLE `test_remote_doris_all_types_t4` (\n `id` datetimev2(0) NOT NULL,\n `comment` datetimev2(0) NULL COMMENT "test comment"\n) ENGINE=DORIS_EXTERNAL_TABLE; + +-- !sql -- +id datetime No true \N +comment datetime Yes false \N NONE + +-- !sql -- +test_remote_doris_all_types_t5 CREATE TABLE `test_remote_doris_all_types_t5` (\n `id` datetimev2(0) NOT NULL,\n `id2` int NOT NULL,\n `id3` varchar(65533) NOT NULL\n) ENGINE=DORIS_EXTERNAL_TABLE; + +-- !sql -- +id datetime No true \N +id2 int No true \N +id3 varchar(65533) No true \N + diff --git a/regression-test/data/external_table_p0/remote_doris/test_remote_doris_refresh.out b/regression-test/data/external_table_p0/remote_doris/test_remote_doris_refresh.out new file mode 100644 index 00000000000000..99f7858fb401e0 --- /dev/null +++ b/regression-test/data/external_table_p0/remote_doris/test_remote_doris_refresh.out @@ -0,0 +1,16 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !sql -- + +-- !sql -- +id datetime No true \N +c_date date Yes false \N NONE + +-- !sql -- +id datetime No true \N +c_date date Yes false \N NONE + +-- !sql -- +id datetime No true \N +c_date date Yes false \N NONE +c_new int Yes false \N NONE + diff --git a/regression-test/data/external_table_p0/remote_doris/test_remote_doris_statistics.out b/regression-test/data/external_table_p0/remote_doris/test_remote_doris_statistics.out new file mode 100644 index 00000000000000..1d915ff36208a3 --- /dev/null +++ b/regression-test/data/external_table_p0/remote_doris/test_remote_doris_statistics.out @@ -0,0 +1,20 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !sql -- +c_bigint 4 3 1 -9223372036854775808 9223372036854775807 32 +c_boolean 4 2 1 0 1 4 +c_char 4 2 1 A 2 +c_date 4 3 1 1970-01-01 9999-12-31 16 +c_datetime 4 3 1 2023-10-01 12:34:56 32 +c_decimal18 4 3 1 -123456789012346 123456789012346 32 +c_decimal32 4 3 1 -1234567890123456789012345678 1234567890123456789012345678 64 +c_decimal9 4 3 1 -123457 123457 16 +c_double 4 3 1 -123456.789 123456.789 32 +c_float 4 3 1 -123.456 123.456 16 +c_int 4 3 1 -2147483648 2147483647 16 +c_largeint 4 3 1 -1234567890123456790 1234567890123456789 64 +c_smallint 4 3 1 -32768 32767 8 +c_string 4 2 1 Hello, Doris! 26 +c_tinyint 4 3 1 -128 127 4 +c_varchar 4 2 1 Hello 10 +id 4 4 0 2025-05-18 01:00:00.000 2025-05-18 04:00:00.000 32 + diff --git a/regression-test/pipeline/external/conf/regression-conf.groovy b/regression-test/pipeline/external/conf/regression-conf.groovy index b42ad06f4ce95d..0d3e94ace11a56 100644 --- a/regression-test/pipeline/external/conf/regression-conf.groovy +++ b/regression-test/pipeline/external/conf/regression-conf.groovy @@ -25,6 +25,13 @@ targetJdbcUrl = "jdbc:mysql://172.19.0.2:9131/?useLocalSessionState=true&allowLo jdbcUser = "root" jdbcPassword = "" +//arrow flight sql test config +extArrowFlightSqlHost = "172.19.0.2" +extArrowFlightSqlPort = 8081 +extArrowFlightSqlUser = "root" +extArrowFlightSqlPassword= "" +extArrowFlightHttpPort= 8131 + ccrDownstreamUrl = "jdbc:mysql://172.19.0.2:9131/?useLocalSessionState=true&allowLoadLocalInfile=true" ccrDownstreamUser = "root" ccrDownstreamPassword = "" diff --git a/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_all_types_select.groovy b/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_all_types_select.groovy new file mode 100644 index 00000000000000..a77d21f59f55b4 --- /dev/null +++ b/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_all_types_select.groovy @@ -0,0 +1,172 @@ +// 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_remote_doris_all_types_select", "p0,external,doris,external_docker,external_docker_doris") { + String remote_doris_host = context.config.otherConfigs.get("extArrowFlightSqlHost") + String remote_doris_arrow_port = context.config.otherConfigs.get("extArrowFlightSqlPort") + String remote_doris_http_port = context.config.otherConfigs.get("extArrowFlightHttpPort") + String remote_doris_user = context.config.otherConfigs.get("extArrowFlightSqlUser") + String remote_doris_psw = context.config.otherConfigs.get("extArrowFlightSqlPassword") + + def showres = sql "show frontends"; + remote_doris_arrow_port = showres[0][6] + remote_doris_http_port = showres[0][3] + log.info("show frontends log = ${showres}, arrow: ${remote_doris_arrow_port}, http: ${remote_doris_http_port}") + + def showres2 = sql "show backends"; + log.info("show backends log = ${showres2}") + + sql """DROP DATABASE IF EXISTS test_remote_doris_all_types_select_db""" + + sql """CREATE DATABASE IF NOT EXISTS test_remote_doris_all_types_select_db""" + + sql """ + CREATE TABLE `test_remote_doris_all_types_select_db`.`test_remote_doris_all_types_select_t` ( + `id` datetime(3) NOT NULL, + `c_boolean` boolean NULL, + `c_tinyint` tinyint NULL, + `c_smallint` smallint NULL, + `c_int` int NULL, + `c_bigint` bigint NULL, + `c_largeint` largeint NULL, + `c_float` float NULL, + `c_double` double NULL, + `c_decimal9` decimal(9,0) NULL, + `c_decimal18` decimal(18,0) NULL, + `c_decimal32` decimal(32,0) NULL, + `c_date` date NULL, + `c_datetime` datetime NULL, + `c_char` char(1) NULL, + `c_varchar` varchar(65533) NULL, + `c_string` text NULL, + `c_array_s` array NULL, + `c_map` MAP NULL, + `c_struct` STRUCT NULL, + ) ENGINE=OLAP + DUPLICATE KEY(`id`) + DISTRIBUTED BY HASH(`id`) BUCKETS 1 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 1" + ); + """ + + sql """ + INSERT INTO `test_remote_doris_all_types_select_db`.`test_remote_doris_all_types_select_t` values('2025-05-18 01:00:00.000', true, -128, -32768, -2147483648, -9223372036854775808, -1234567890123456790, -123.456, -123456.789, -123457, -123456789012346, -1234567890123456789012345678, '1970-01-01', '0000-01-01 00:00:00', 'A', 'Hello', 'Hello, Doris!', '["apple", "banana", "orange"]', {"Emily":101,"age":25} , {11, 3.14, "Emily"}) + """ + sql """ + INSERT INTO `test_remote_doris_all_types_select_db`.`test_remote_doris_all_types_select_t` values('2025-05-18 02:00:00.000', NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL) + """ + sql """ + INSERT INTO `test_remote_doris_all_types_select_db`.`test_remote_doris_all_types_select_t` values('2025-05-18 03:00:00.000', false, 127, 32767, 2147483647, 9223372036854775807, 1234567890123456789, 123.456, 123456.789, 123457, 123456789012346, 1234567890123456789012345678, '9999-12-31', '9999-12-31 23:59:59', '', '', '', [], {}, {11, 3.14, "Emily"}) + """ + sql """ + INSERT INTO `test_remote_doris_all_types_select_db`.`test_remote_doris_all_types_select_t` values('2025-05-18 04:00:00.000', true, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, '2023-10-01', '2023-10-01 12:34:56', 'A', 'Hello', 'Hello, Doris!', '["apple", "banana", "orange"]', {"Emily":101,"age":25} , {11, 3.14, "Emily"}); + """ + + sql """ + CREATE TABLE `test_remote_doris_all_types_select_db`.`test_remote_doris_all_types_select_t2` ( + `id` datetime(3) NOT NULL, + `a_boolean` array NULL, + `a_tinyint` array NULL, + `a_smallint` array NULL, + `a_int` array NULL, + `a_bigint` array NULL, + `a_largeint` array NULL, + `a_float` array NULL, + `a_double` array NULL, + `a_decimal9` array NULL, + `a_decimal18` array NULL, + `a_decimal32` array NULL, + `a_date` array NULL, + `a_datetime` array NULL, + `a_char` array NULL, + `a_varchar` array NULL, + `a_string` array NULL + ) ENGINE=OLAP + DUPLICATE KEY(`id`) + DISTRIBUTED BY HASH(`id`) BUCKETS 1 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 1" + ); + """ + + sql """ + INSERT INTO `test_remote_doris_all_types_select_db`.`test_remote_doris_all_types_select_t2` values('2025-05-18 01:00:00.000', [true], [-128], [-32768], [-2147483648], [-9223372036854775808], [-1234567890123456790], [-123.456], [-123456.789], [-123457], [-123456789012346], [-1234567890123456789012345678], ['0000-01-01'], ['0000-01-01 00:00:00'], ['A'], ['Hello'], ['Hello, Doris!']) + """ + sql """ + INSERT INTO `test_remote_doris_all_types_select_db`.`test_remote_doris_all_types_select_t2` values('2025-05-18 02:00:00.000', [NULL], [NULL], [NULL], [NULL], [NULL], [NULL], [NULL], [NULL], [NULL], [NULL], [NULL], [NULL], [NULL], [NULL], [NULL], [NULL]) + """ + sql """ + INSERT INTO `test_remote_doris_all_types_select_db`.`test_remote_doris_all_types_select_t2` values('2025-05-18 03:00:00.000', [false], [127], [32767], [2147483647], [9223372036854775807], [1234567890123456789], [123.456], [123456.789], [123457], [123456789012346], [1234567890123456789012345678], ['9999-12-31'], ['9999-12-31 23:59:59'], [''], [''], ['']) + """ + sql """ + INSERT INTO `test_remote_doris_all_types_select_db`.`test_remote_doris_all_types_select_t2` values('2025-05-18 04:00:00.000', [true], [0], [0], [0], [0], [0], [0], [0], [0], [0], [0], ['2023-10-01'], ['2023-10-01 12:34:56'], ['A'], ['Hello'], ['Hello, Doris!']); + """ + + sql """ + CREATE TABLE `test_remote_doris_all_types_select_db`.`test_remote_doris_all_types_select_t3` ( + `id` datetime NOT NULL, + `datetime_0` datetime(0) NULL, + `datetime_1` datetime(1) NULL, + `datetime_3` datetime(2) NULL, + `datetime_4` datetime(3) NULL, + `datetime_5` datetime(4) NULL, + `datetime_6` datetime(5) NULL, + `datetime_7` datetime(6) NULL + ) ENGINE=OLAP + DUPLICATE KEY(`id`) + DISTRIBUTED BY HASH(`id`) BUCKETS 1 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 1" + ); + """ + + sql """ + INSERT INTO `test_remote_doris_all_types_select_db`.`test_remote_doris_all_types_select_t3` values('2025-05-18 01:00:00.111111', '2025-05-18 01:00:00.111111', '2025-05-18 01:00:00.111111', '2025-05-18 01:00:00.111111', '2025-05-18 01:00:00.111111', '2025-05-18 01:00:00.111111', '2025-05-18 01:00:00.111111', '2025-05-18 01:00:00.111111'); + """ + + + sql """ + DROP CATALOG IF EXISTS `test_remote_doris_all_types_select_catalog` + """ + + + sql """ + CREATE CATALOG `test_remote_doris_all_types_select_catalog` PROPERTIES ( + 'type' = 'doris', + 'fe_http_hosts' = 'http://${remote_doris_host}:${remote_doris_http_port}', + 'fe_arrow_hosts' = '${remote_doris_host}:${remote_doris_arrow_port}', + 'user' = '${remote_doris_user}', + 'password' = '${remote_doris_psw}' + ); + """ + + qt_sql """ + select * from `test_remote_doris_all_types_select_catalog`.`test_remote_doris_all_types_select_db`.`test_remote_doris_all_types_select_t` order by id + """ + + qt_sql """ + select * from `test_remote_doris_all_types_select_catalog`.`test_remote_doris_all_types_select_db`.`test_remote_doris_all_types_select_t2` order by id + """ + + qt_sql """ + select * from `test_remote_doris_all_types_select_catalog`.`test_remote_doris_all_types_select_db`.`test_remote_doris_all_types_select_t3` order by id + """ + + sql """ DROP DATABASE IF EXISTS test_remote_doris_all_types_select_db """ + sql """ DROP CATALOG IF EXISTS `test_remote_doris_all_types_select_catalog` """ +} diff --git a/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_all_types_show.groovy b/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_all_types_show.groovy new file mode 100644 index 00000000000000..ab9f89019e245f --- /dev/null +++ b/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_all_types_show.groovy @@ -0,0 +1,168 @@ +// 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_remote_doris_all_types_show", "p0,external,doris,external_docker,external_docker_doris") { + String remote_doris_host = context.config.otherConfigs.get("extArrowFlightSqlHost") + String remote_doris_arrow_port = context.config.otherConfigs.get("extArrowFlightSqlPort") + String remote_doris_http_port = context.config.otherConfigs.get("extArrowFlightHttpPort") + String remote_doris_user = context.config.otherConfigs.get("extArrowFlightSqlUser") + String remote_doris_psw = context.config.otherConfigs.get("extArrowFlightSqlPassword") + + def showres = sql "show frontends"; + remote_doris_arrow_port = showres[0][6] + remote_doris_http_port = showres[0][3] + log.info("show frontends log = ${showres}, arrow: ${remote_doris_arrow_port}, http: ${remote_doris_http_port}") + + def showres2 = sql "show backends"; + log.info("show backends log = ${showres2}") + + sql """DROP DATABASE IF EXISTS test_remote_doris_all_types_db""" + + sql """CREATE DATABASE IF NOT EXISTS test_remote_doris_all_types_db""" + + sql """ + CREATE TABLE `test_remote_doris_all_types_db`.`test_remote_doris_all_types_t1` ( + `id` datetime(3) NOT NULL, + `c_boolean` boolean NULL DEFAULT 'true', + `c_tinyint` tinyint NULL DEFAULT 1, + `c_smallint` smallint NULL DEFAULT 1, + `c_int` int NULL DEFAULT 1, + `c_bigint` bigint NULL DEFAULT 1, + `c_largeint` largeint NULL DEFAULT 1, + `c_float` float NULL DEFAULT 1, + `c_double` double NULL DEFAULT 1, + `c_decimal9` decimal(9,0) NULL DEFAULT 1, + `c_decimal18` decimal(18,0) NULL DEFAULT 1, + `c_decimal32` decimal(32,0) NULL DEFAULT 1, + `c_date` date NULL DEFAULT '2025-08-18', + `c_datetime` datetime NULL DEFAULT '2025-08-18 20:00:00', + `c_char` char(1) NULL DEFAULT 'd', + `c_varchar` varchar(65533) NULL DEFAULT 'd', + `c_string` text NULL DEFAULT 'd', + `c_array_s` array NULL, + `c_map` MAP NULL, + `c_struct` STRUCT NULL, + ) ENGINE=OLAP + DUPLICATE KEY(`id`) + DISTRIBUTED BY HASH(`id`) BUCKETS 1 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 1" + ); + """ + + sql """ + CREATE TABLE `test_remote_doris_all_types_db`.`test_remote_doris_all_types_t2` ( + `id` datetime(3) NOT NULL, + `a_boolean` array NULL, + `a_tinyint` array NULL, + `a_smallint` array NULL, + `a_int` array NULL, + `a_bigint` array NULL, + `a_largeint` array NULL, + `a_float` array NULL, + `a_double` array NULL, + `a_decimal9` array NULL, + `a_decimal18` array NULL, + `a_decimal32` array NULL, + `a_date` array NULL, + `a_datetime` array NULL, + `a_char` array NULL, + `a_varchar` array NULL, + `a_string` array NULL + ) ENGINE=OLAP + DUPLICATE KEY(`id`) + DISTRIBUTED BY HASH(`id`) BUCKETS 1 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 1" + ); + """ + + sql """ + CREATE TABLE `test_remote_doris_all_types_db`.`test_remote_doris_all_types_t3` ( + `id` datetime NOT NULL, + `datetime_0` datetime(0) NULL, + `datetime_1` datetime(1) NULL, + `datetime_3` datetime(2) NULL, + `datetime_4` datetime(3) NULL, + `datetime_5` datetime(4) NULL, + `datetime_6` datetime(5) NULL, + `datetime_7` datetime(6) NULL + ) ENGINE=OLAP + DUPLICATE KEY(`id`) + DISTRIBUTED BY HASH(`id`) BUCKETS 1 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 1" + ); + """ + + sql """ + CREATE TABLE `test_remote_doris_all_types_db`.`test_remote_doris_all_types_t4` ( + `id` datetime NOT NULL, + `comment` datetime(0) NULL COMMENT 'test comment' + ) ENGINE=OLAP + DUPLICATE KEY(`id`) + DISTRIBUTED BY HASH(`id`) BUCKETS 1 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 1" + ); + """ + + sql """ + CREATE TABLE `test_remote_doris_all_types_db`.`test_remote_doris_all_types_t5` ( + `id` datetime NOT NULL, + `id2` int NOT NULL, + `id3` varchar NOT NULL + ) ENGINE=OLAP + DUPLICATE KEY(`id`, `id2`, `id3`) + DISTRIBUTED BY HASH(`id`) BUCKETS 1 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 1" + ); + """ + + sql """ + DROP CATALOG IF EXISTS `test_remote_doris_all_types_catalog` + """ + + sql """ + CREATE CATALOG `test_remote_doris_all_types_catalog` PROPERTIES ( + 'type' = 'doris', + 'fe_http_hosts' = 'http://${remote_doris_host}:${remote_doris_http_port}', + 'fe_arrow_hosts' = '${remote_doris_host}:${remote_doris_arrow_port}', + 'user' = '${remote_doris_user}', + 'password' = '${remote_doris_psw}' + ); + """ + + qt_sql """ SHOW CREATE TABLE test_remote_doris_all_types_catalog.test_remote_doris_all_types_db.test_remote_doris_all_types_t1""" + qt_sql """ DESC test_remote_doris_all_types_catalog.test_remote_doris_all_types_db.test_remote_doris_all_types_t1""" + + qt_sql """ SHOW CREATE TABLE test_remote_doris_all_types_catalog.test_remote_doris_all_types_db.test_remote_doris_all_types_t2""" + qt_sql """ DESC test_remote_doris_all_types_catalog.test_remote_doris_all_types_db.test_remote_doris_all_types_t2""" + + qt_sql """ SHOW CREATE TABLE test_remote_doris_all_types_catalog.test_remote_doris_all_types_db.test_remote_doris_all_types_t3""" + qt_sql """ DESC test_remote_doris_all_types_catalog.test_remote_doris_all_types_db.test_remote_doris_all_types_t3""" + + qt_sql """ SHOW CREATE TABLE test_remote_doris_all_types_catalog.test_remote_doris_all_types_db.test_remote_doris_all_types_t4""" + qt_sql """ DESC test_remote_doris_all_types_catalog.test_remote_doris_all_types_db.test_remote_doris_all_types_t4""" + + qt_sql """ SHOW CREATE TABLE test_remote_doris_all_types_catalog.test_remote_doris_all_types_db.test_remote_doris_all_types_t5""" + qt_sql """ DESC test_remote_doris_all_types_catalog.test_remote_doris_all_types_db.test_remote_doris_all_types_t5""" + + sql """DROP DATABASE IF EXISTS test_remote_doris_all_types_db""" + sql """DROP CATALOG IF EXISTS `test_remote_doris_all_types_catalog`""" +} diff --git a/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_catalog.groovy b/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_catalog.groovy new file mode 100644 index 00000000000000..82e2b2e3550c4f --- /dev/null +++ b/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_catalog.groovy @@ -0,0 +1,68 @@ +// 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_remote_doris_catalog", "p0,external,doris,external_docker,external_docker_doris") { + String remote_doris_host = context.config.otherConfigs.get("extArrowFlightSqlHost") + String remote_doris_arrow_port = context.config.otherConfigs.get("extArrowFlightSqlPort") + String remote_doris_http_port = context.config.otherConfigs.get("extArrowFlightHttpPort") + String remote_doris_user = context.config.otherConfigs.get("extArrowFlightSqlUser") + String remote_doris_psw = context.config.otherConfigs.get("extArrowFlightSqlPassword") + + def showres = sql "show frontends"; + remote_doris_arrow_port = showres[0][6] + remote_doris_http_port = showres[0][3] + log.info("show frontends log = ${showres}, arrow: ${remote_doris_arrow_port}, http: ${remote_doris_http_port}") + + def showres2 = sql "show backends"; + log.info("show backends log = ${showres2}") + + // delete catalog + sql """ + DROP CATALOG IF EXISTS `test_remote_doris_catalog_catalog` + """ + + // create catalog + sql """ + CREATE CATALOG `test_remote_doris_catalog_catalog` PROPERTIES ( + 'type' = 'doris', + 'fe_http_hosts' = 'http://${remote_doris_host}:${remote_doris_http_port}', + 'fe_arrow_hosts' = '${remote_doris_host}:${remote_doris_arrow_port}', + 'user' = '${remote_doris_user}', + 'password' = '${remote_doris_psw}' + ); + """ + + // show catalog + sql """ + SHOW CREATE CATALOG `test_remote_doris_catalog_catalog` + """ + + // alter catalog + sql """ + ALTER CATALOG `test_remote_doris_catalog_catalog` SET PROPERTIES ('enable_parallel_result_sink' = 'false'); + """ + + sql """ + SHOW CREATE CATALOG `test_remote_doris_catalog_catalog` + """ + + sql """ + DROP CATALOG IF EXISTS `test_remote_doris_catalog_catalog` + """ +} + + diff --git a/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_predict.groovy b/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_predict.groovy new file mode 100644 index 00000000000000..3782f5c50a7c50 --- /dev/null +++ b/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_predict.groovy @@ -0,0 +1,150 @@ +// 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_remote_doris_predict", "p0,external,doris,external_docker,external_docker_doris") { + String remote_doris_host = context.config.otherConfigs.get("extArrowFlightSqlHost") + String remote_doris_arrow_port = context.config.otherConfigs.get("extArrowFlightSqlPort") + String remote_doris_http_port = context.config.otherConfigs.get("extArrowFlightHttpPort") + String remote_doris_user = context.config.otherConfigs.get("extArrowFlightSqlUser") + String remote_doris_psw = context.config.otherConfigs.get("extArrowFlightSqlPassword") + + def showres = sql "show frontends"; + remote_doris_arrow_port = showres[0][6] + remote_doris_http_port = showres[0][3] + log.info("show frontends log = ${showres}, arrow: ${remote_doris_arrow_port}, http: ${remote_doris_http_port}") + + def showres2 = sql "show backends"; + log.info("show backends log = ${showres2}") + + sql """DROP DATABASE IF EXISTS test_remote_doris_predict_db""" + + sql """CREATE DATABASE IF NOT EXISTS test_remote_doris_predict_db""" + + sql """ + CREATE TABLE `test_remote_doris_predict_db`.`test_remote_doris_predict_t` ( + `id` int NOT NULL, + `c_int` int NULL, + `c_string` text NULL + ) ENGINE=OLAP + DUPLICATE KEY(`id`) + DISTRIBUTED BY HASH(`id`) BUCKETS 1 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 1" + ); + """ + + sql """ + INSERT INTO `test_remote_doris_predict_db`.`test_remote_doris_predict_t` VALUES + (1,1,'abc'), + (2,2,'bcd'), + (3,3,'cde'), + (4,4,'def'), + (5,5,'efg'); + """ + + sql """ + DROP CATALOG IF EXISTS `test_remote_doris_predict_catalog` + """ + + sql """ + CREATE CATALOG `test_remote_doris_predict_catalog` PROPERTIES ( + 'type' = 'doris', + 'fe_http_hosts' = 'http://${remote_doris_host}:${remote_doris_http_port}', + 'fe_arrow_hosts' = '${remote_doris_host}:${remote_doris_arrow_port}', + 'user' = '${remote_doris_user}', + 'password' = '${remote_doris_psw}' + ); + """ + + sql """use test_remote_doris_predict_catalog.test_remote_doris_predict_db""" + + explain { + sql("select * from test_remote_doris_predict_t where c_int < 3") + contains("c_int < 3") + } + + explain { + sql("select * from test_remote_doris_predict_t where c_int <= 3") + contains("c_int <= 3") + } + + explain { + sql("select * from test_remote_doris_predict_t where c_int > 3") + contains("c_int > 3") + } + + explain { + sql("select * from test_remote_doris_predict_t where c_int >= 3") + contains("c_int >= 3") + } + + explain { + sql("select * from test_remote_doris_predict_t where c_int = 3") + contains("c_int = 3") + } + + explain { + sql("select * from test_remote_doris_predict_t where c_int != 3") + contains("c_int != 3") + } + + explain { + sql("select * from test_remote_doris_predict_t where c_int > 3 AND c_string = 'cde'") + contains("((c_int > 3)) AND ((c_string = 'cde'))") + } + + explain { + sql("select * from test_remote_doris_predict_t where c_int > 3 OR c_string = 'cde'") + contains("(c_int > 3) OR (c_string = 'cde')") + } + + explain { + sql("select * from test_remote_doris_predict_t where NOT c_int > 3") + contains("c_int <= 3") + } + + explain { + sql("select * from test_remote_doris_predict_t where c_string LIKE '%d%'") + contains("c_string like '%d%'") + } + + explain { + sql("select * from test_remote_doris_predict_t where NOT c_string LIKE '%d%'") + contains("NOT c_string like '%d%'") + } + + explain { + sql("select * from test_remote_doris_predict_t where c_int IN(2,3,4)") + contains("c_int IN (2, 3, 4)") + } + + explain { + sql("select * from test_remote_doris_predict_t where c_string IS NULL") + contains("c_string IS NULL") + } + + explain { + sql("select * from test_remote_doris_predict_t where c_string IS NOT NULL") + contains("c_string IS NOT NULL") + } + + explain { + sql("select * from test_remote_doris_predict_t where trim_in(c_string,'a') = 'bc';") + contains("(trim_in(c_string, 'a') = 'bc'") + } +} + diff --git a/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_refresh.groovy b/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_refresh.groovy new file mode 100644 index 00000000000000..e7596eafffb7ec --- /dev/null +++ b/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_refresh.groovy @@ -0,0 +1,128 @@ +// 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_remote_doris_refresh", "p0,external,doris,external_docker,external_docker_doris") { + String remote_doris_host = context.config.otherConfigs.get("extArrowFlightSqlHost") + String remote_doris_arrow_port = context.config.otherConfigs.get("extArrowFlightSqlPort") + String remote_doris_http_port = context.config.otherConfigs.get("extArrowFlightHttpPort") + String remote_doris_user = context.config.otherConfigs.get("extArrowFlightSqlUser") + String remote_doris_psw = context.config.otherConfigs.get("extArrowFlightSqlPassword") + + def showres = sql "show frontends"; + remote_doris_arrow_port = showres[0][6] + remote_doris_http_port = showres[0][3] + log.info("show frontends log = ${showres}, arrow: ${remote_doris_arrow_port}, http: ${remote_doris_http_port}") + + def showres2 = sql "show backends"; + log.info("show backends log = ${showres2}") + + sql """ + DROP CATALOG IF EXISTS `test_remote_doris_refresh_catalog` + """ + + sql """DROP DATABASE IF EXISTS test_remote_doris_refresh_db""" + + sql """ + CREATE CATALOG `test_remote_doris_refresh_catalog` PROPERTIES ( + 'type' = 'doris', + 'fe_http_hosts' = 'http://${remote_doris_host}:${remote_doris_http_port}', + 'fe_arrow_hosts' = '${remote_doris_host}:${remote_doris_arrow_port}', + 'user' = '${remote_doris_user}', + 'password' = '${remote_doris_psw}' + ); + """ + + def databases_init = sql """ + SHOW DATABASES FROM `test_remote_doris_refresh_catalog` + """ + + def checkName = { ArrayList result, String name -> + for (item in result) { + println item + if (item.toString() == "[" + name + "]") { + println "success" + return + } + } + println "fail" + } + + checkName(databases_init, "test_remote_doris_refresh_db") + + sql """CREATE DATABASE IF NOT EXISTS test_remote_doris_refresh_db""" + + def databases_before = sql """ + SHOW DATABASES FROM `test_remote_doris_refresh_catalog` + """ + + checkName(databases_before, "test_remote_doris_refresh_db") + + sql """REFRESH CATALOG test_remote_doris_refresh_catalog""" + + def databases_after = sql """ + SHOW DATABASES FROM `test_remote_doris_refresh_catalog` + """ + + checkName(databases_after, "test_remote_doris_refresh_db") + + qt_sql """ + SHOW TABLES FROM `test_remote_doris_refresh_catalog`.`test_remote_doris_refresh_db` + """ + + sql """ + CREATE TABLE `test_remote_doris_refresh_db`.`test_remote_doris_catalog_t` ( + `id` datetime NOT NULL, + `c_date` date NULL + ) ENGINE=OLAP + DUPLICATE KEY(`id`) + DISTRIBUTED BY HASH(`id`) BUCKETS 1 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 1" + ); + """ + + def tables_before = sql """ + SHOW TABLES FROM `test_remote_doris_refresh_catalog`.`test_remote_doris_refresh_db` + """ + + checkName(tables_before, "test_remote_doris_catalog_t") + + sql """REFRESH CATALOG test_remote_doris_refresh_catalog""" + + def tables_after = sql """ + SHOW TABLES FROM `test_remote_doris_refresh_catalog`.`test_remote_doris_refresh_db` + """ + + checkName(tables_after, "test_remote_doris_catalog_t") + + qt_sql """ DESC `test_remote_doris_refresh_catalog`.`test_remote_doris_refresh_db`.`test_remote_doris_catalog_t`""" + + sql """ ALTER TABLE `test_remote_doris_refresh_db`.`test_remote_doris_catalog_t` ADD COLUMN ( c_new int NULL) """ + + qt_sql """ DESC `test_remote_doris_refresh_catalog`.`test_remote_doris_refresh_db`.`test_remote_doris_catalog_t`""" + + sql """ REFRESH CATALOG test_remote_doris_refresh_catalog """ + + qt_sql """ DESC `test_remote_doris_refresh_catalog`.`test_remote_doris_refresh_db`.`test_remote_doris_catalog_t` """ + + sql """ + DROP CATALOG IF EXISTS `test_remote_doris_refresh_catalog` + """ + sql """DROP DATABASE IF EXISTS test_remote_doris_refresh_db""" +} + + diff --git a/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_statistics.groovy b/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_statistics.groovy new file mode 100644 index 00000000000000..f2633b8cd28559 --- /dev/null +++ b/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_statistics.groovy @@ -0,0 +1,105 @@ +// 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_remote_doris_statistics", "p0,external,doris,external_docker,external_docker_doris") { + String remote_doris_host = context.config.otherConfigs.get("extArrowFlightSqlHost") + String remote_doris_arrow_port = context.config.otherConfigs.get("extArrowFlightSqlPort") + String remote_doris_http_port = context.config.otherConfigs.get("extArrowFlightHttpPort") + String remote_doris_user = context.config.otherConfigs.get("extArrowFlightSqlUser") + String remote_doris_psw = context.config.otherConfigs.get("extArrowFlightSqlPassword") + + def showres = sql "show frontends"; + remote_doris_arrow_port = showres[0][6] + remote_doris_http_port = showres[0][3] + log.info("show frontends log = ${showres}, arrow: ${remote_doris_arrow_port}, http: ${remote_doris_http_port}") + + def showres2 = sql "show backends"; + log.info("show backends log = ${showres2}") + + sql """DROP DATABASE IF EXISTS test_remote_doris_statistics_db""" + + sql """CREATE DATABASE IF NOT EXISTS test_remote_doris_statistics_db""" + + sql """ + CREATE TABLE `test_remote_doris_statistics_db`.`test_remote_doris_statistics_t1` ( + `id` datetime(3) NOT NULL, + `c_boolean` boolean NULL DEFAULT 'true', + `c_tinyint` tinyint NULL DEFAULT 1, + `c_smallint` smallint NULL DEFAULT 1, + `c_int` int NULL DEFAULT 1, + `c_bigint` bigint NULL DEFAULT 1, + `c_largeint` largeint NULL DEFAULT 1, + `c_float` float NULL DEFAULT 1, + `c_double` double NULL DEFAULT 1, + `c_decimal9` decimal(9,0) NULL DEFAULT 1, + `c_decimal18` decimal(18,0) NULL DEFAULT 1, + `c_decimal32` decimal(32,0) NULL DEFAULT 1, + `c_date` date NULL DEFAULT '2025-08-18', + `c_datetime` datetime NULL DEFAULT '2025-08-18 20:00:00', + `c_char` char(1) NULL DEFAULT 'd', + `c_varchar` varchar(65533) NULL DEFAULT 'd', + `c_string` text NULL DEFAULT 'd', + `c_array_s` array NULL, + `c_map` MAP NULL, + `c_struct` STRUCT NULL, + ) ENGINE=OLAP + DUPLICATE KEY(`id`) + DISTRIBUTED BY HASH(`id`) BUCKETS 1 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 1" + ); + """ + + sql """ + DROP CATALOG IF EXISTS `test_remote_doris_statistics_catalog` + """ + + sql """ + CREATE CATALOG `test_remote_doris_statistics_catalog` PROPERTIES ( + 'type' = 'doris', + 'fe_http_hosts' = 'http://${remote_doris_host}:${remote_doris_http_port}', + 'fe_arrow_hosts' = '${remote_doris_host}:${remote_doris_arrow_port}', + 'user' = '${remote_doris_user}', + 'password' = '${remote_doris_psw}' + ); + """ + + sql """ + INSERT INTO `test_remote_doris_statistics_db`.`test_remote_doris_statistics_t1` values('2025-05-18 01:00:00.000', true, -128, -32768, -2147483648, -9223372036854775808, -1234567890123456790, -123.456, -123456.789, -123457, -123456789012346, -1234567890123456789012345678, '1970-01-01', '0000-01-01 00:00:00', 'A', 'Hello', 'Hello, Doris!', '["apple", "banana", "orange"]', {"Emily":101,"age":25} , {11, 3.14, "Emily"}) + """ + sql """ + INSERT INTO `test_remote_doris_statistics_db`.`test_remote_doris_statistics_t1` values('2025-05-18 02:00:00.000', NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL) + """ + sql """ + INSERT INTO `test_remote_doris_statistics_db`.`test_remote_doris_statistics_t1` values('2025-05-18 03:00:00.000', false, 127, 32767, 2147483647, 9223372036854775807, 1234567890123456789, 123.456, 123456.789, 123457, 123456789012346, 1234567890123456789012345678, '9999-12-31', '9999-12-31 23:59:59', '', '', '', [], {}, {11, 3.14, "Emily"}) + """ + sql """ + INSERT INTO `test_remote_doris_statistics_db`.`test_remote_doris_statistics_t1` values('2025-05-18 04:00:00.000', true, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, '2023-10-01', '2023-10-01 12:34:56', 'A', 'Hello', 'Hello, Doris!', '["apple", "banana", "orange"]', {"Emily":101,"age":25} , {11, 3.14, "Emily"}); + """ + + def table_id = get_table_id("test_remote_doris_statistics_catalog", "test_remote_doris_statistics_db", "test_remote_doris_statistics_t1") + def catalog_id = get_catalog_id("test_remote_doris_statistics_catalog"); + + sql """analyze table test_remote_doris_statistics_catalog.test_remote_doris_statistics_db.test_remote_doris_statistics_t1 with sync""" + + qt_sql """select col_id,count,ndv,null_count,min,max,data_size_in_bytes from internal.__internal_schema.column_statistics where tbl_id = ${table_id} and catalog_id = ${catalog_id} order by id;""" + + sql """DROP DATABASE IF EXISTS test_remote_doris_statistics_db""" + sql """ + DROP CATALOG IF EXISTS `test_remote_doris_statistics_catalog` + """ +} diff --git a/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_table_stats.groovy b/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_table_stats.groovy new file mode 100644 index 00000000000000..a9dd083ffd9a59 --- /dev/null +++ b/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_table_stats.groovy @@ -0,0 +1,99 @@ +// 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_remote_doris_table_stats", "p0,external,doris,external_docker,external_docker_doris") { + String remote_doris_host = context.config.otherConfigs.get("extArrowFlightSqlHost") + String remote_doris_arrow_port = context.config.otherConfigs.get("extArrowFlightSqlPort") + String remote_doris_http_port = context.config.otherConfigs.get("extArrowFlightHttpPort") + String remote_doris_user = context.config.otherConfigs.get("extArrowFlightSqlUser") + String remote_doris_psw = context.config.otherConfigs.get("extArrowFlightSqlPassword") + + def showres = sql "show frontends"; + remote_doris_arrow_port = showres[0][6] + remote_doris_http_port = showres[0][3] + log.info("show frontends log = ${showres}, arrow: ${remote_doris_arrow_port}, http: ${remote_doris_http_port}") + + def showres2 = sql "show backends"; + log.info("show backends log = ${showres2}") + + sql """DROP DATABASE IF EXISTS test_remote_doris_table_stats_db""" + + sql """CREATE DATABASE IF NOT EXISTS test_remote_doris_table_stats_db""" + + sql """ + CREATE TABLE `test_remote_doris_table_stats_db`.`test_remote_doris_table_stats_t1` ( + `id` datetime(3) NOT NULL, + `c_boolean` boolean NULL DEFAULT 'true', + `c_tinyint` tinyint NULL DEFAULT 1, + `c_smallint` smallint NULL DEFAULT 1, + `c_int` int NULL DEFAULT 1, + `c_bigint` bigint NULL DEFAULT 1, + `c_largeint` largeint NULL DEFAULT 1, + `c_float` float NULL DEFAULT 1, + `c_double` double NULL DEFAULT 1, + `c_decimal9` decimal(9,0) NULL DEFAULT 1, + `c_decimal18` decimal(18,0) NULL DEFAULT 1, + `c_decimal32` decimal(32,0) NULL DEFAULT 1, + `c_date` date NULL DEFAULT '2025-08-18', + `c_datetime` datetime NULL DEFAULT '2025-08-18 20:00:00', + `c_char` char(1) NULL DEFAULT 'd', + `c_varchar` varchar(65533) NULL DEFAULT 'd', + `c_string` text NULL DEFAULT 'd', + `c_array_s` array NULL, + `c_map` MAP NULL, + `c_struct` STRUCT NULL, + ) ENGINE=OLAP + DUPLICATE KEY(`id`) + DISTRIBUTED BY HASH(`id`) BUCKETS 1 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 1" + ); + """ + + sql """ + DROP CATALOG IF EXISTS `test_remote_doris_table_stats_catalog` + """ + + sql """ + CREATE CATALOG `test_remote_doris_table_stats_catalog` PROPERTIES ( + 'type' = 'doris', + 'fe_http_hosts' = 'http://${remote_doris_host}:${remote_doris_http_port}', + 'fe_arrow_hosts' = '${remote_doris_host}:${remote_doris_arrow_port}', + 'user' = '${remote_doris_user}', + 'password' = '${remote_doris_psw}' + ); + """ + + sql """ + INSERT INTO `test_remote_doris_table_stats_db`.`test_remote_doris_table_stats_t1` values('2025-05-18 01:00:00.000', true, -128, -32768, -2147483648, -9223372036854775808, -1234567890123456790, -123.456, -123456.789, -123457, -123456789012346, -1234567890123456789012345678, '1970-01-01', '0000-01-01 00:00:00', 'A', 'Hello', 'Hello, Doris!', '["apple", "banana", "orange"]', {"Emily":101,"age":25} , {11, 3.14, "Emily"}) + """ + sql """ + INSERT INTO `test_remote_doris_table_stats_db`.`test_remote_doris_table_stats_t1` values('2025-05-18 02:00:00.000', NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL) + """ + sql """ + INSERT INTO `test_remote_doris_table_stats_db`.`test_remote_doris_table_stats_t1` values('2025-05-18 03:00:00.000', false, 127, 32767, 2147483647, 9223372036854775807, 1234567890123456789, 123.456, 123456.789, 123457, 123456789012346, 1234567890123456789012345678, '9999-12-31', '9999-12-31 23:59:59', '', '', '', [], {}, {11, 3.14, "Emily"}) + """ + sql """ + INSERT INTO `test_remote_doris_table_stats_db`.`test_remote_doris_table_stats_t1` values('2025-05-18 04:00:00.000', true, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, '2023-10-01', '2023-10-01 12:34:56', 'A', 'Hello', 'Hello, Doris!', '["apple", "banana", "orange"]', {"Emily":101,"age":25} , {11, 3.14, "Emily"}); + """ + + sql """use test_remote_doris_table_stats_catalog.test_remote_doris_table_stats_db""" + sql """analyze table test_remote_doris_table_stats_t1 with sync""" + + def result = sql """ show table stats test_remote_doris_table_stats_t1; """ + println(result[0][2]) +} From 63c8e352b042aaa1aa9038567b2a094a13b0f529 Mon Sep 17 00:00:00 2001 From: morningman Date: Sat, 22 Nov 2025 15:42:48 +0800 Subject: [PATCH 2/2] fix conflict --- be/src/vec/exec/format/table/remote_doris_reader.cpp | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/be/src/vec/exec/format/table/remote_doris_reader.cpp b/be/src/vec/exec/format/table/remote_doris_reader.cpp index fa0f8566c0f49f..15a868153ec476 100644 --- a/be/src/vec/exec/format/table/remote_doris_reader.cpp +++ b/be/src/vec/exec/format/table/remote_doris_reader.cpp @@ -80,10 +80,13 @@ Status RemoteDorisReader::get_next_block(Block* block, size_t* read_rows, bool* 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); + block->safe_get_by_position(c); + if (column_with_name.name != column_name) { + return Status::InternalError("Column name mismatch: expected {}, got {}", + column_with_name.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) {