Add installation scripts for dependencies for node delta sharing.

CMakeLists.txt

@@ -15,7 +15,7 @@ project(villas-node
 
 # Some more project settings
 set(PROJECT_AUTHOR "Steffen Vogel")
-set(PROJECT_COPYRIGHT "2014-2025 The VILLASframework Authors")
+set(PROJECT_COPYRIGHT "2014-2021, Institute for Automation of Complex Power Systems, RWTH Aachen University")
 
 # Several CMake settings/defaults
 set(CMAKE_C_STANDARD 11)
@@ -205,6 +205,7 @@ cmake_dependent_option(WITH_NODE_WEBRTC     "Build with webrtc node-type"
 cmake_dependent_option(WITH_NODE_WEBSOCKET  "Build with websocket node-type"                        "${WITH_DEFAULTS}" "WITH_WEB" OFF)
 cmake_dependent_option(WITH_NODE_ZEROMQ     "Build with zeromq node-type"                           "${WITH_DEFAULTS}" "LIBZMQ_FOUND; NOT WITHOUT_GPL" OFF)
 cmake_dependent_option(WITH_NODE_OPENDSS    "Build with opendss node-type"                          "${WITH_DEFAULTS}" "OpenDSSC_FOUND" OFF)
+cmake_dependent_option(WITH_NODE_DELTASHARE "Build with delta-share node-type"                      "${WITH_DEFAULTS}" "" ON)
 
 # set a default for the build type
 if("${CMAKE_BUILD_TYPE}" STREQUAL "")

include/villas/nodes/delta_share/DeltaSharingClient.h

@@ -0,0 +1,34 @@
+
+#ifndef __DELTASHARINGCLIENT_H__
+#define __DELTASHARINGCLIENT_H__
+
+#include <iostream>
+#include "DeltaSharingRestClient.h"
+#include <arrow/table.h>
+
+namespace DeltaSharing
+{
+
+    struct DeltaSharingClient
+    {
+    public:
+        DeltaSharingClient(std::string filename, boost::optional<std::string> cacheLocation);
+        std::shared_ptr<arrow::Table> LoadAsArrowTable(std::string &url);
+        std::shared_ptr<arrow::Table> ReadTableFromCache(std::string &url);
+        const std::shared_ptr<std::vector<DeltaSharingProtocol::Share>> ListShares(int maxResult, std::string pageToken) const;
+        const std::shared_ptr<std::vector<DeltaSharingProtocol::Schema>> ListSchemas(const DeltaSharingProtocol::Share &share, int maxResult, std::string pageToken) const;
+        const std::shared_ptr<std::vector<DeltaSharingProtocol::Table>> ListTables(const DeltaSharingProtocol::Schema &schema, int maxResult, std::string pageToken) const;
+        const std::shared_ptr<std::vector<DeltaSharingProtocol::Table>> ListAllTables(const DeltaSharingProtocol::Share &share, int maxResult, std::string pageToken) const;
+        const std::shared_ptr<std::vector<DeltaSharingProtocol::File>> ListFilesInTable(const DeltaSharingProtocol::Table) const;
+        const DeltaSharingProtocol::Metadata QueryTableMetadata(const DeltaSharingProtocol::Table &table) const;
+        const int GetNumberOfThreads() { return this->maxThreads; };
+        void PopulateCache(std::string url) { this->restClient.PopulateCache(url,this->cacheLocation); };
+    protected:
+    private:
+        DeltaSharingRestClient restClient;
+        std::string cacheLocation;
+        int maxThreads;
+    };
+};
+
+#endif

include/villas/nodes/delta_share/DeltaSharingRestClient.h

@@ -0,0 +1,41 @@
+
+#ifndef __DELTASHARINGRESTCLIENT_H__
+#define __DELTASHARINGRESTCLIENT_H__
+
+#include <iostream>
+#include <nlohmann/json.hpp>
+#include <list>
+#include "Protocol.h"
+#include <restclient-cpp/restclient.h>
+#include <restclient-cpp/connection.h>
+
+using json = nlohmann::json;
+
+namespace DeltaSharing
+{
+    struct DeltaSharingRestClient
+    {
+    public:
+        DeltaSharingRestClient(std::string filename);
+        ~DeltaSharingRestClient();
+        const std::shared_ptr<std::vector<DeltaSharingProtocol::Share>> ListShares(int maxResult, std::string pageToken) const;
+        const std::shared_ptr<std::vector<DeltaSharingProtocol::Schema>> ListSchemas(const DeltaSharingProtocol::Share &share, int maxResult, std::string pageToken) const;
+        const std::shared_ptr<std::vector<DeltaSharingProtocol::Table>> ListTables(const DeltaSharingProtocol::Schema &schema, int maxResult, std::string pageToken) const;
+        const std::shared_ptr<std::vector<DeltaSharingProtocol::Table>> ListAllTables(const DeltaSharingProtocol::Share &share, int maxResult, std::string pageToken) const;
+        const std::shared_ptr<std::vector<DeltaSharingProtocol::File>> ListFilesInTable(const DeltaSharingProtocol::Table) const;
+        const DeltaSharingProtocol::Metadata QueryTableMetadata(const DeltaSharingProtocol::Table &table) const;
+        const DeltaSharingProtocol::DeltaSharingProfile &GetProfile() const;
+        RestClient::Response get(std::string url);
+        void PopulateCache(std::string url, std::string cacheLocation);
+        const bool shouldRetry(RestClient::Response &response) const;
+
+    protected:
+        json ReadFromFile(std::string filename);
+
+    private:
+        DeltaSharingProtocol::DeltaSharingProfile profile;
+        static const std::string user_agent;
+    };
+};
+
+#endif

include/villas/nodes/delta_share/Protocol.h

@@ -0,0 +1,210 @@
+
+#ifndef __PROTOCOL_H__
+#define __PROTOCOL_H__
+
+#include <iostream>
+#include <nlohmann/json.hpp>
+#include <map>
+#include <string>
+#include <vector>
+#include <boost/optional.hpp>
+#include <boost/optional/optional_io.hpp>
+
+namespace DeltaSharing
+{
+
+    namespace DeltaSharingProtocol
+    {
+
+        using json = nlohmann::json;
+
+        struct DeltaSharingProfile
+        {
+        public:
+            int shareCredentialsVersion;
+            std::string endpoint;
+            std::string bearerToken;
+            boost::optional<std::string> expirationTime;
+        };
+
+        inline void from_json(const json &j, DeltaSharingProfile &p)
+        {
+            if (j.contains("shareCredentialsVersion"))
+            {
+                p.shareCredentialsVersion = j["shareCredentialsVersion"];
+            }
+            if (j.contains("endpoint"))
+            {
+                p.endpoint = j["endpoint"];
+            }
+            if (j.contains("bearerToken"))
+            {
+                p.bearerToken = j["bearerToken"];
+            }
+            if (j.contains("expirationTime"))
+            {
+                p.expirationTime = j["expirationTime"];
+            }
+        };
+
+        struct Share
+        {
+        public:
+            std::string name = "";
+            boost::optional<std::string> id;
+        };
+
+        inline void from_json(const json &j, Share &s)
+        {
+            s.name = j["name"];
+            if (j.contains("id") == false)
+            {
+                s.id = boost::none;
+            }
+            else
+            {
+                s.id = j["id"];
+            }
+        };
+
+        struct Schema
+        {
+        public:
+            std::string name;
+            std::string share;
+        };
+
+        inline void from_json(const json &j, Schema &s)
+        {
+            s.name = j["name"];
+            if (j.contains("share") == true)
+            {
+                s.share = j["share"];
+            }
+        };
+
+        struct Table
+        {
+        public:
+            std::string name;
+            std::string share;
+            std::string schema;
+        };
+
+        inline void from_json(const json &j, Table &t)
+        {
+            if (j.contains("name"))
+            {
+                t.name = j["name"];
+            }
+            if (j.contains("share"))
+            {
+                t.share = j["share"];
+            }
+            if (j.contains("schema"))
+            {
+                t.schema = j["schema"];
+            }
+        };
+
+        struct File
+        {
+        public:
+            std::string url;
+            boost::optional<std::string> id;
+            std::map<std::string, std::string> partitionValues;
+            std::size_t size;
+            std::string stats;
+            boost::optional<std::string> timestamp;
+            boost::optional<std::string> version;
+        };
+
+        inline void from_json(const json &j, File &f)
+        {
+            if (j.contains("url"))
+            {
+                f.url = j["url"];
+            }
+            if (j.contains("id"))
+            {
+                f.id = j["id"];
+            }
+            if (j.contains("partitionValues"))
+            {
+                json arr = j["partitionValues"];
+                auto f2 = f.partitionValues;
+                if (arr.is_array())
+                {
+                    for (auto it = arr.begin(); it < arr.end(); it++)
+                    {
+                        f2.insert({it.key(), it.value()});
+                    }
+                }
+            }
+            if (j.contains("size"))
+            {
+                f.size = j["size"];
+            }
+            if (j.contains("stats"))
+            {
+                f.stats = j["stats"];
+            }
+            if (j.contains("timestamp"))
+            {
+                f.timestamp = j["timestamp"];
+            }
+            if (j.contains("version"))
+            {
+                f.version = j["version"];
+            }
+        };
+
+        struct Format {
+            std::string provider;
+
+        };
+
+        NLOHMANN_DEFINE_TYPE_NON_INTRUSIVE(Format, provider)
+
+        struct Metadata
+        {
+            Format format;
+            std::string id;
+            std::vector<std::string> partitionColumns;
+            std::string schemaString;
+        };
+
+        NLOHMANN_DEFINE_TYPE_NON_INTRUSIVE(Metadata,format, id, partitionColumns,schemaString)
+
+        struct data
+        {
+            std::vector<std::string> predicateHints;
+            int limitHint;
+        };
+        NLOHMANN_DEFINE_TYPE_NON_INTRUSIVE(data, predicateHints, limitHint)
+
+        struct format
+        {
+            std::string provider;
+        };
+        NLOHMANN_DEFINE_TYPE_NON_INTRUSIVE(format, provider)
+
+        struct protocol
+        {
+            int minReaderVersion;
+        };
+        NLOHMANN_DEFINE_TYPE_NON_INTRUSIVE(protocol,minReaderVersion)
+
+        struct stats
+        {
+            long long numRecords;
+            long minValues;
+            long maxValues;
+            long nullCount;
+        };
+        NLOHMANN_DEFINE_TYPE_NON_INTRUSIVE(stats, numRecords, minValues, maxValues, nullCount)
+
+    };
+};
+
+#endif

include/villas/nodes/delta_share/delta_share.hpp

@@ -0,0 +1,67 @@
+/* Node type: Delta Share.
+ *
+ * Author:
+ * SPDX-FileCopyrightText: 2014-2023 Institute for Automation of Complex Power Systems, RWTH Aachen University
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <cstddef>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include <jansson.h>
+
+#include "villas/nodes/delta_share/Protocol.h"
+#include "villas/nodes/delta_share/DeltaSharingClient.h"
+
+namespace arrow { class Table; }
+
+namespace villas {
+namespace node {
+
+// Forward declarations
+class NodeCompat;
+
+struct delta_share {
+  // Configuration
+  std::string profile_path;
+  std::string cache_dir;
+  std::string table_path;
+  size_t batch_size;
+  std::string schema;
+  std::string share;
+  std::string table;
+
+  // Client and state
+  std::shared_ptr<DeltaSharing::DeltaSharingClient> client;
+  std::shared_ptr<std::vector<DeltaSharing::DeltaSharingProtocol::Schema>> schemas;
+  std::shared_ptr<arrow::Table> table_ptr;
+  std::shared_ptr<std::vector<DeltaSharing::DeltaSharingProtocol::Table>> tables;
+  std::shared_ptr<std::vector<DeltaSharing::DeltaSharingProtocol::Share>> shares;
+
+  enum class TableOp { TABLE_NOOP, TABLE_READ, TABLE_WRITE } table_op;
+};
+
+char *deltaShare_print(NodeCompat *n);
+
+int deltaShare_parse(NodeCompat *n, json_t *json);
+
+int deltaShare_start(NodeCompat *n);
+
+int deltaShare_stop(NodeCompat *n);
+
+int deltaShare_init(NodeCompat *n);
+
+int deltaShare_destroy(NodeCompat *n);
+
+int deltaShare_poll_fds(NodeCompat *n, int fds[]);
+
+int deltaShare_read(NodeCompat *n, struct Sample *const smps[], unsigned cnt);
+
+int deltaShare_write(NodeCompat *n, struct Sample *const smps[], unsigned cnt);
+
+} // namespace node
+} // namespace villas

include/villas/nodes/delta_share/functions.h

@@ -0,0 +1,22 @@
+
+#ifndef __FUNCTIONS_H__
+#define __FUNCTIONS_H__
+
+#include <vector>
+#include <string>
+#include <iostream>
+#include <arrow/table.h>
+#include "DeltaSharingClient.h"
+#include <thread>
+
+namespace DeltaSharing {
+
+    const std::vector<std::string> ParseURL(std::string path); 
+     std::shared_ptr<DeltaSharingClient> NewDeltaSharingClient(std::string profile, boost::optional<std::string> cacheLocation);
+    const std::shared_ptr<arrow::Table> LoadAsArrowTable(std::string path, int fileno);
+
+
+
+};
+
+#endif