task: updated dependencies and refactored

Primary changes in this commit involve refactoring to accommodate schema
resolution and the use of additional substrait and mohair types.
Additionally, this commit includes changes to various binaries to
accommodate API changes

Author: drin

poetry.lock

@@ -5,7 +5,7 @@ description = "query processing for decomposable queries"
 authors = ["Aldrin M <[email protected]>"]
 license = "Apache License 2.0"
 packages = [
-     { include = "mohair"  , from = "src/python" }
+     { include = "mohair",   from = "src/python" }
     ,{ include = "skyproto", from = "src/python" }
 ]
 
@@ -19,11 +19,10 @@ datafusion = "^35.0.0"
 pkgconfig = "^1.5.5"
 pyarrow-hotfix = "^0.6"
 protobuf = "==5.28.3"
-substrait = "==0.20.0"
-ibis-substrait = "==4.0.0"
-pyarrow = "^18.0.0"
-
-[tool.poetry.dev-dependencies]
+substrait = "^0.23.0"
+#ibis-substrait = "==4.0.0"
+pyarrow = "^19.0.0"
+ibis-skytether = "^0.1.0"
 
 [tool.poetry.scripts]
 mohair = "mohair.cli:cli"

pyproject.toml

@@ -24,6 +24,7 @@
 #include <arrow/api.h>
 
 // >> Arrow serialization API
+#include <arrow/io/api.h>
 #include <arrow/ipc/api.h>
 
 // >> Arrow filesystem API
@@ -54,8 +55,12 @@ namespace skytether {
   using arrow::RecordBatchVector;
 
   // >> Support types for I/O
-  using arrow::io::RandomAccessFile;
   using ArrowOutputStream = arrow::io::OutputStream;
+  using arrow::io::RandomAccessFile;
+  using arrow::io::BufferOutputStream;
+
+  using arrow::ipc::IpcWriteOptions;
+  using arrow::ipc::WriteRecordBatchStream;
 
   using arrow::RecordBatchReader;
   using arrow::ipc::RecordBatchStreamReader;

src/cpp/skytether/apidep_arrow.hpp

@@ -44,22 +44,34 @@ namespace skytether {
   using AnyMessage = google::protobuf::Any;
 
   // >> Mohair query processing types
-  using skyproto::mohair::SuperPlan;
-  using skyproto::mohair::SubPlan;
-  using skyproto::mohair::ErrRel;
+  using mohair::Plan;
+  using mohair::SuperPlan;
+  using mohair::SubPlan;
+
+  using mohair::PlanRel;
+  using mohair::RelRoot;
+
+  using mohair::Rel;
+  using mohair::RelCommon;
+  using mohair::ErrRel;
+  using mohair::ExtensionLeafRel;
+  using mohair::SkyResultRel;
 
   // >> Mohair topology types
-  using skyproto::mohair::ServiceConfig;
-  using skyproto::mohair::DeviceClass;
+  using mohair::ServiceConfig;
+  using mohair::DeviceClass;
 
   // >> Types from mohair
   using mohair::PlanMessage;
-  using mohair::SubstraitMessage;
 
   using mohair::SystemPlan;
+  using mohair::PlanSplit;
+  using mohair::DecomposeAlg;
+
+  using mohair::SubstraitSchema;
+  using mohair::SubstraitType;
 
   // TODO: see if these should be aliased in mohair
-  using SubstraitType       = skyproto::substrait::Type;
   using SubstraitExpression = skyproto::substrait::Expression;
   using SubstraitSortField  = skyproto::substrait::SortField;
 

src/cpp/skytether/apidep_mohair.hpp

@@ -47,6 +47,11 @@
 #include <fstream>
 #include <filesystem>
 
+// >> Concurrency
+#include <atomic>
+#include <mutex>
+#include <condition_variable>
+
 
 // ------------------------------
 // Type aliases
@@ -69,5 +74,12 @@ namespace skytether {
   using std::stringstream;
   using std::fstream;
 
+  // >> Safety type aliases
+  using std::atomic;
+  using std::mutex;
+  using std::condition_variable;
+  using std::unique_lock;
+  using std::lock_guard;
+
 } // namespace: skytether
 

src/cpp/skytether/apidep_standard.hpp

@@ -106,7 +106,7 @@ struct ServiceActions {
       return ERRCODE_API_REGISTER;
     }
 
-    SkytetherDebugMsg("Initializing service with config:");
+    SkytetherDebugMsg("Initializing connected service with config:");
     skytether::services::PrintConfig(service_cfg.get());
 
     return 0;
@@ -117,7 +117,7 @@ struct ServiceActions {
     service_cfg = std::make_unique<ServiceConfig>();
     service_cfg->set_service_location(service_loc.ToString());
 
-    SkytetherDebugMsg("Initializing service with config:");
+    SkytetherDebugMsg("Initializing local-only service with config:");
     skytether::services::PrintConfig(service_cfg.get());
 
     return 0;
@@ -161,8 +161,17 @@ struct ServiceActions {
     }
 
     #if SKYTETHER_USE_DUCKDB
-      auto skytether_duckcse = std::make_unique<DuckDBService>(&fn_deactivate);
-      auto status_start      = StartService(*skytether_duckcse, *service_cfg);
+      auto skytether_service = std::make_unique<DuckDBService>(
+        std::move(service_cfg), &fn_deactivate
+      );
+
+      auto status_connect = skytether_service->ConnectToTopology();
+      if (not status_connect.ok()) {
+        skytether::PrintError("Unable to connect to downstream services", status_connect);
+        return ERRCODE_START_SRV;
+      }
+
+      auto status_start = StartService(*skytether_service, service_loc);
       if (not status_start.ok()) {
         skytether::PrintError("Unable to start csd-service", status_start);
         return ERRCODE_START_SRV;