add argo catalog

Author: bingtao.yin

be/CMakeLists.txt

@@ -690,6 +690,17 @@ else()
     endif()
 endif()
 
+# Add tddms deps
+set(TDDMS_INCLUDE_DIR $ENV{DORIS_THIRDPARTY}/tddms_client/include)
+set(TDDMS_LIB_DIR $ENV{DORIS_THIRDPARTY}/tddms_client/lib)
+set(DORIS_LINK_LIBS ${DORIS_LINK_LIBS} shiva_client)
+link_directories(${TDDMS_LIB_DIR} ${TDDMS_LIB_DIR})
+
+# Enable multithread for linking; see ld.gold --help
+if(COMPILER_GCC)
+    set(DORIS_LINK_LIBS "${DORIS_LINK_LIBS} -fuse-ld=gold -Wl,--threads -Wl,--thread-count=4")
+endif()
+
 # Set libraries for test
 set (TEST_LINK_LIBS ${DORIS_LINK_LIBS}
     ${WL_START_GROUP}

be/cmake/thirdparty.cmake

@@ -27,43 +27,55 @@ set(COMMON_THIRDPARTY)
 # if arg exist wholelibpath, use wholelibpath to find library
 function(add_thirdparty)
     cmake_parse_arguments(DORIS_THIRDPARTY
-        "NOTADD;LIB64"
+        "NOTADD;LIB64;SHARED"
         "LIBNAME;WHOLELIBPATH"
         ""
         ${ARGN})
 
     set(DORIS_THIRDPARTY_NAME ${DORIS_THIRDPARTY_UNPARSED_ARGUMENTS})
-    add_library(${DORIS_THIRDPARTY_NAME} STATIC IMPORTED)
+    if (DORIS_THIRDPARTY_SHARED)
+        add_library(${DORIS_THIRDPARTY_NAME} SHARED IMPORTED)
+    else()
+        add_library(${DORIS_THIRDPARTY_NAME} STATIC IMPORTED)
+    endif()
 
     if (NOT DORIS_THIRDPARTY_NOTADD)
         set(COMMON_THIRDPARTY ${COMMON_THIRDPARTY} ${DORIS_THIRDPARTY_NAME} PARENT_SCOPE)
     endif()
 
     if (DORIS_THIRDPARTY_LIB64)
-        set(DORIS_THIRDPARTY_LIBPATH ${THIRDPARTY_DIR}/lib64/lib${DORIS_THIRDPARTY_NAME}.a)
+        if (DORIS_THIRDPARTY_SHARED)
+            set(DORIS_THIRDPARTY_LIBPATH ${THIRDPARTY_DIR}/lib64/lib${DORIS_THIRDPARTY_NAME}.so)
+        else()
+            set(DORIS_THIRDPARTY_LIBPATH ${THIRDPARTY_DIR}/lib64/lib${DORIS_THIRDPARTY_NAME}.a)
+        endif()
     elseif (DORIS_THIRDPARTY_LIBNAME)
         set(DORIS_THIRDPARTY_LIBPATH ${THIRDPARTY_DIR}/${DORIS_THIRDPARTY_LIBNAME})
     elseif (DORIS_THIRDPARTY_WHOLELIBPATH)
         set(DORIS_THIRDPARTY_LIBPATH ${DORIS_THIRDPARTY_WHOLELIBPATH})
     else()
-        set(DORIS_THIRDPARTY_LIBPATH ${THIRDPARTY_DIR}/lib/lib${DORIS_THIRDPARTY_NAME}.a)
+        if (DORIS_THIRDPARTY_SHARED)
+            set(DORIS_THIRDPARTY_LIBPATH ${THIRDPARTY_DIR}/lib/lib${DORIS_THIRDPARTY_NAME}.so)
+        else()
+            set(DORIS_THIRDPARTY_LIBPATH ${THIRDPARTY_DIR}/lib/lib${DORIS_THIRDPARTY_NAME}.a)
+        endif()
     endif()
     set_target_properties(${DORIS_THIRDPARTY_NAME} PROPERTIES IMPORTED_LOCATION ${DORIS_THIRDPARTY_LIBPATH})
 endfunction()
 
-add_thirdparty(gflags)
-add_thirdparty(glog)
+add_thirdparty(gflags SHARED)
+add_thirdparty(glog SHARED)
 add_thirdparty(backtrace)
-add_thirdparty(re2)
+add_thirdparty(re2 SHARED)
 add_thirdparty(hyperscan LIBNAME "lib64/libhs.a")
 add_thirdparty(odbc)
 add_thirdparty(pprof WHOLELIBPATH ${GPERFTOOLS_HOME}/lib/libprofiler.a)
-add_thirdparty(protobuf)
+add_thirdparty(protobuf SHARED)
 add_thirdparty(gtest)
 add_thirdparty(gtest_main)
 add_thirdparty(benchmark)
 add_thirdparty(gmock)
-add_thirdparty(snappy)
+add_thirdparty(snappy SHARED)
 add_thirdparty(curl)
 add_thirdparty(lz4)
 add_thirdparty(thrift)
@@ -90,14 +102,14 @@ if (USE_UNWIND)
     add_thirdparty(libunwind LIBNAME "lib64/libunwind.a")
 endif()
 
-add_thirdparty(grpc++_reflection LIB64)
-add_thirdparty(grpc LIB64)
-add_thirdparty(grpc++ LIB64)
-add_thirdparty(grpc++_unsecure LIB64)
-add_thirdparty(gpr LIB64)
-add_thirdparty(upb LIB64)
-add_thirdparty(cares LIB64)
-add_thirdparty(address_sorting LIB64)
+add_thirdparty(grpc++_reflection SHARED LIB64)
+add_thirdparty(grpc SHARED LIB64)
+add_thirdparty(grpc++ SHARED LIB64)
+add_thirdparty(grpc++_unsecure SHARED LIB64)
+add_thirdparty(gpr SHARED LIB64)
+add_thirdparty(upb SHARED LIB64)
+add_thirdparty(cares SHARED LIB64)
+add_thirdparty(address_sorting SHARED LIB64)
 add_thirdparty(z LIB64)
 
 add_thirdparty(brotlicommon LIB64)
@@ -108,9 +120,9 @@ add_thirdparty(arrow LIB64)
 add_thirdparty(arrow_flight LIB64)
 add_thirdparty(arrow_flight_sql LIB64)
 add_thirdparty(parquet LIB64)
-add_thirdparty(brpc LIB64)
+add_thirdparty(brpc SHARED LIB64)
 add_thirdparty(rocksdb)
-add_thirdparty(cyrus-sasl WHOLELIBPATH "/lib/x86_64-linux-gnu/libsasl2.so")
+add_thirdparty(cyrus-sasl SHARED LIBNAME "lib/libsasl2.so")
 # put this after lz4 to avoid using lz4 lib in librdkafka
 add_thirdparty(rdkafka_cpp LIBNAME "lib/librdkafka++.a")
 add_thirdparty(rdkafka)
@@ -121,10 +133,10 @@ add_thirdparty(fmt)
 add_thirdparty(cctz)
 add_thirdparty(base64)
 
-add_thirdparty(aws-cpp-sdk-core LIB64)
-add_thirdparty(aws-cpp-sdk-s3 LIB64)
-add_thirdparty(aws-cpp-sdk-transfer LIB64)
-add_thirdparty(aws-cpp-sdk-s3-crt LIB64)
+add_thirdparty(aws-cpp-sdk-core SHARED LIB64)
+add_thirdparty(aws-cpp-sdk-s3 SHARED LIB64)
+add_thirdparty(aws-cpp-sdk-transfer SHARED LIB64)
+add_thirdparty(aws-cpp-sdk-s3-crt SHARED LIB64)
 add_thirdparty(aws-crt-cpp LIB64)
 add_thirdparty(aws-c-cal LIB64)
 add_thirdparty(aws-c-auth LIB64)
@@ -137,8 +149,8 @@ add_thirdparty(aws-c-mqtt LIB64)
 add_thirdparty(aws-checksums LIB64)
 add_thirdparty(aws-c-s3 LIB64)
 add_thirdparty(aws-c-sdkutils LIB64)
-add_thirdparty(aws-cpp-sdk-identity-management LIB64)
-add_thirdparty(aws-cpp-sdk-sts LIB64)
+add_thirdparty(aws-cpp-sdk-identity-management SHARED LIB64)
+add_thirdparty(aws-cpp-sdk-sts SHARED LIB64)
 if (NOT OS_MACOSX)
     add_thirdparty(aws-s2n LIBNAME "lib/libs2n.a")
 endif()
@@ -156,11 +168,11 @@ add_thirdparty(idn LIB64)
 add_thirdparty(xml2 LIB64)
 add_thirdparty(lzma LIB64)
 add_thirdparty(gsasl)
-add_thirdparty(krb5support WHOLELIBPATH "/lib/x86_64-linux-gnu/libkrb5support.so")
-add_thirdparty(krb5 WHOLELIBPATH "/lib/x86_64-linux-gnu/libkrb5.so")
-add_thirdparty(com_err WHOLELIBPATH "/lib/x86_64-linux-gnu/libcom_err.so")
-add_thirdparty(k5crypto WHOLELIBPATH "/lib/x86_64-linux-gnu/libk5crypto.so")
-add_thirdparty(gssapi_krb5 WHOLELIBPATH "/lib/x86_64-linux-gnu/libgssapi_krb5.so")
+add_thirdparty(krb5support SHARED)
+add_thirdparty(krb5 SHARED)
+add_thirdparty(com_err SHARED)
+add_thirdparty(k5crypto SHARED)
+add_thirdparty(gssapi_krb5 SHARED)
 add_thirdparty(streamvbyte LIB64)
 
 if (OS_MACOSX)

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/util/stack_util.cpp

@@ -27,11 +27,11 @@
 #include "util/mem_info.h"
 #include "util/pretty_printer.h"
 
-namespace google {
-namespace glog_internal_namespace_ {
-void DumpStackTraceToString(std::string* stacktrace);
-}
-} // namespace google
+// namespace google {
+// namespace glog_internal_namespace_ {
+// void DumpStackTraceToString(std::string* stacktrace);
+// }
+// } // namespace google
 
 namespace doris {
 
@@ -67,7 +67,7 @@ std::string get_stack_trace(int start_pointers_index, std::string dwarf_location
 
 std::string get_stack_trace_by_glog() {
     std::string s;
-    google::glog_internal_namespace_::DumpStackTraceToString(&s);
+    // google::glog_internal_namespace_::DumpStackTraceToString(&s);
     return s;
 }