add argo catalog

Author: bingtao.yin

be/src/pipeline/exec/argo_scan_operator.cpp

@@ -0,0 +1,35 @@
+/*
+ *  Created on: 2025-08-28
+ *      Author: bingtao.yin@transwarp.io
+ */
+
+#include "pipeline/exec/argo_scan_operator.h"
+
+#include "vec/exec/scan/argo_scanner.h"
+
+namespace doris::pipeline {
+
+Status ArgoScanLocalState::_init_scanners(std::list<vectorized::VScannerSPtr>* scanners) {
+    auto& p = _parent->cast<ArgoScanOperatorX>();
+    std::unique_ptr<vectorized::ArgoScanner> scanner =
+            vectorized::ArgoScanner::create_unique(state(), this, p._limit, _scanner_profile.get());
+    RETURN_IF_ERROR(scanner->prepare(state(), _conjuncts));
+    scanners->push_back(std::move(scanner));
+    return Status::OK();
+}
+
+std::string ArgoScanLocalState::name_suffix() const {
+    return fmt::format(" (id={}. nereids_id={}. argo table name= {})",
+                       std::to_string(_parent->node_id()), std::to_string(_parent->nereids_id()),
+                       _parent->cast<ArgoScanOperatorX>()._table_name);
+};
+
+ArgoScanOperatorX::ArgoScanOperatorX(ObjectPool* pool, const TPlanNode& tnode, int operator_id,
+                                     const DescriptorTbl& descs, int parallel_tasks)
+        : ScanOperatorX<ArgoScanLocalState>(pool, tnode, operator_id, descs, parallel_tasks),
+          _tuple_id(tnode.argo_scan_node.tuple_id),
+          _table_name(tnode.argo_scan_node.table_name) {
+    _output_tuple_id = tnode.argo_scan_node.tuple_id;
+}
+
+}; // namespace doris::pipeline

be/src/pipeline/exec/argo_scan_operator.h

@@ -0,0 +1,42 @@
+/*
+ *  Created on: 2025-08-28
+ *      Author: bingtao.yin@transwarp.io
+ */
+
+#pragma once
+
+#include "common/status.h"
+#include "operator.h"
+#include "pipeline/exec/scan_operator.h"
+
+namespace doris {}
+
+namespace doris::pipeline {
+
+class ArgoScanOperatorX;
+class ArgoScanLocalState final : public ScanLocalState<ArgoScanLocalState> {
+public:
+    using Parent = ArgoScanOperatorX;
+    ENABLE_FACTORY_CREATOR(ArgoScanLocalState);
+    ArgoScanLocalState(RuntimeState* state, OperatorXBase* parent)
+            : ScanLocalState<ArgoScanLocalState>(state, parent) {}
+
+    Status _init_scanners(std::list<vectorized::VScannerSPtr> *scanners) override;
+
+    std::string name_suffix() const override;
+};
+
+class ArgoScanOperatorX final : public ScanOperatorX<ArgoScanLocalState> {
+public:
+    ArgoScanOperatorX(ObjectPool* pool, const TPlanNode& tnode, int operator_id,
+                      const DescriptorTbl& descs, int parallel_tasks);
+private:
+    friend class ArgoScanLocalState;
+
+    TupleId _tuple_id;
+    std::string _table_name;
+
+    TupleDescriptor* _tuple_desc = nullptr;
+};
+
+} // namespace doris::pipeline

be/src/pipeline/exec/operator.cpp

@@ -23,6 +23,7 @@
 #include "pipeline/exec/aggregation_source_operator.h"
 #include "pipeline/exec/analytic_sink_operator.h"
 #include "pipeline/exec/analytic_source_operator.h"
+#include "pipeline/exec/argo_scan_operator.h"
 #include "pipeline/exec/assert_num_rows_operator.h"
 #include "pipeline/exec/cache_sink_operator.h"
 #include "pipeline/exec/cache_source_operator.h"
@@ -701,6 +702,7 @@ DECLARE_OPERATOR(GroupCommitLocalState)
 DECLARE_OPERATOR(JDBCScanLocalState)
 DECLARE_OPERATOR(FileScanLocalState)
 DECLARE_OPERATOR(EsScanLocalState)
+DECLARE_OPERATOR(ArgoScanLocalState)
 DECLARE_OPERATOR(AnalyticLocalState)
 DECLARE_OPERATOR(SortLocalState)
 DECLARE_OPERATOR(SpillSortLocalState)

be/src/pipeline/exec/scan_operator.cpp

@@ -23,6 +23,7 @@
 #include <memory>
 
 #include "pipeline/common/runtime_filter_consumer.h"
+#include "pipeline/exec/argo_scan_operator.h"
 #include "pipeline/exec/es_scan_operator.h"
 #include "pipeline/exec/file_scan_operator.h"
 #include "pipeline/exec/group_commit_scan_operator.h"
@@ -1322,6 +1323,8 @@ template class ScanOperatorX<EsScanLocalState>;
 template class ScanLocalState<EsScanLocalState>;
 template class ScanLocalState<MetaScanLocalState>;
 template class ScanOperatorX<MetaScanLocalState>;
+template class ScanLocalState<ArgoScanLocalState>;
+template class ScanOperatorX<ArgoScanLocalState>;
 template class ScanOperatorX<GroupCommitLocalState>;
 template class ScanLocalState<GroupCommitLocalState>;
 

be/src/pipeline/pipeline_fragment_context.cpp

@@ -43,6 +43,7 @@
 #include "pipeline/exec/aggregation_source_operator.h"
 #include "pipeline/exec/analytic_sink_operator.h"
 #include "pipeline/exec/analytic_source_operator.h"
+#include "pipeline/exec/argo_scan_operator.h"
 #include "pipeline/exec/assert_num_rows_operator.h"
 #include "pipeline/exec/cache_sink_operator.h"
 #include "pipeline/exec/cache_source_operator.h"
@@ -1220,6 +1221,13 @@ Status PipelineFragmentContext::_create_operator(ObjectPool* pool, const TPlanNo
         fe_with_old_version = !tnode.__isset.is_serial_operator;
         break;
     }
+    case TPlanNodeType::ARGO_SCAN_NODE:{
+        op.reset(new ArgoScanOperatorX(pool, tnode, next_operator_id(), descs, _num_instances));
+        RETURN_IF_ERROR(cur_pipe->add_operator(
+                op, request.__isset.parallel_instances ? request.parallel_instances : 0));
+        fe_with_old_version = !tnode.__isset.is_serial_operator;
+        break;
+    }
     case TPlanNodeType::EXCHANGE_NODE: {
         int num_senders = find_with_default(request.per_exch_num_senders, tnode.node_id, 0);
         DCHECK_GT(num_senders, 0);

be/src/runtime/descriptors.cpp

@@ -305,6 +305,13 @@ std::string JdbcTableDescriptor::debug_string() const {
     return fmt::to_string(buf);
 }
 
+ArgoTableDescriptor::ArgoTableDescriptor(const TTableDescriptor& tdesc) : TableDescriptor(tdesc) {}
+std::string ArgoTableDescriptor::debug_string() const {
+    fmt::memory_buffer buf;
+    fmt::format_to(buf, "ArgoTable({})", TableDescriptor::debug_string());
+    return fmt::to_string(buf);
+}
+
 TupleDescriptor::TupleDescriptor(const TTupleDescriptor& tdesc, bool own_slots)
         : _id(tdesc.id),
           _num_materialized_slots(0),
@@ -577,6 +584,9 @@ Status DescriptorTbl::create(ObjectPool* pool, const TDescriptorTable& thrift_tb
         case TTableType::TRINO_CONNECTOR_TABLE:
             desc = pool->add(new TrinoConnectorTableDescriptor(tdesc));
             break;
+        case TTableType::ARGO_TABLE:
+            desc = pool->add(new ArgoTableDescriptor(tdesc));
+            break;
         default:
             DCHECK(false) << "invalid table type: " << tdesc.tableType;
         }

be/src/runtime/descriptors.h

@@ -333,6 +333,12 @@ class JdbcTableDescriptor : public TableDescriptor {
     bool _connection_pool_keep_alive;
 };
 
+class ArgoTableDescriptor : public TableDescriptor {
+public:
+    ArgoTableDescriptor(const TTableDescriptor& tdesc);
+    std::string debug_string() const override;
+};
+
 class TupleDescriptor {
 public:
     TupleDescriptor(TupleDescriptor&&) = delete;

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

@@ -0,0 +1,42 @@
+/*
+ *  Created on: 2025-08-28
+ *      Author: bingtao.yin@transwarp.io
+ */
+
+#include "vec/exec/scan/argo_scanner.h"
+
+#include "common/logging.h"
+#include "common/status.h"
+
+namespace doris::vectorized {
+
+ArgoScanner::ArgoScanner(RuntimeState* state, doris::pipeline::ArgoScanLocalState* local_state,
+                         int64_t limit, RuntimeProfile* profile)
+        : VScanner(state, local_state, limit, profile) {}
+
+Status ArgoScanner::open(RuntimeState* state) {
+    VLOG_CRITICAL << "ArgoScanner::open";
+    RETURN_IF_CANCELLED(state);
+    RETURN_IF_ERROR(VScanner::open(state));
+    return Status::OK();
+}
+
+Status ArgoScanner::close(RuntimeState* state) {
+    VLOG_CRITICAL << "ArgoScanner::close";
+    RETURN_IF_ERROR(VScanner::close(state));
+    return Status::OK();
+}
+
+Status ArgoScanner::prepare(RuntimeState* state, const VExprContextSPtrs& conjuncts) {
+    VLOG_CRITICAL << "ArgoScanner::prepare";
+    RETURN_IF_ERROR(VScanner::prepare(state, conjuncts));
+    // TODO: tuple descriptor
+    return Status::OK();
+}
+
+Status ArgoScanner::_get_block_impl(RuntimeState* state, Block* block, bool* eos) {
+    *eos = true;
+    return Status::OK();
+}
+
+} // namespace doris::vectorized

be/src/vec/exec/scan/argo_scanner.h

@@ -0,0 +1,28 @@
+/*
+ *  Created on: 2025-08-28
+ *      Author: bingtao.yin@transwarp.io
+ */
+
+#pragma once
+
+#include "pipeline/exec/argo_scan_operator.h"
+#include "vec/exec/scan/vscanner.h"
+
+namespace doris::vectorized {
+
+class ArgoScanner : public VScanner {
+    ENABLE_FACTORY_CREATOR(ArgoScanner);
+
+public:
+    ArgoScanner(RuntimeState* state, doris::pipeline::ArgoScanLocalState* parent, int64_t limit,
+                RuntimeProfile* profile);
+    Status open(RuntimeState* state) override;
+    Status close(RuntimeState* state) override;
+
+    Status prepare(RuntimeState* state, const VExprContextSPtrs& conjuncts) override;
+
+protected:
+    Status _get_block_impl(RuntimeState* state, Block* block, bool* eos) override;
+};
+
+} // namespace doris::vectorized

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

@@ -79,6 +79,7 @@ public class JdbcResource extends Resource {
     public static final String JDBC_OCEANBASE = "jdbc:oceanbase";
     public static final String JDBC_DB2 = "jdbc:db2";
     public static final String JDBC_GBASE = "jdbc:gbase";
+    public static final String JDBC_ARGO = "jdbc:transwarp2";
 
     public static final String MYSQL = "MYSQL";
     public static final String POSTGRESQL = "POSTGRESQL";
@@ -92,6 +93,7 @@ public class JdbcResource extends Resource {
     public static final String OCEANBASE_ORACLE = "OCEANBASE_ORACLE";
     public static final String DB2 = "DB2";
     public static final String GBASE = "GBASE";
+    public static final String ARGO = "ARGO";
 
     public static final String JDBC_PROPERTIES_PREFIX = "jdbc.";
     public static final String JDBC_URL = "jdbc_url";
@@ -342,6 +344,8 @@ public static String parseDbType(String url) throws DdlException {
             return DB2;
         } else if (url.startsWith(JDBC_GBASE)) {
             return GBASE;
+        } else if (url.startsWith(JDBC_ARGO)) {
+            return ARGO;
         }
         throw new DdlException("Unsupported jdbc database type, please check jdbcUrl: " + url);
     }