[#1013] Implement the Database Adapter

.docker/poc/Dockerfile

@@ -36,16 +36,49 @@ RUN if [ "$(uname -m)" = "aarch64" ] || [ "$(uname -m)" = "arm64" ]; then \
     else \
         LIB_SUBDIR="x86_64-linux-gnu"; \
     fi \
-    && mkdir -p ${DAS_DIR}/syslibs/usr/lib/${LIB_SUBDIR} ${DAS_DIR}/syslibs/lib/${LIB_SUBDIR} ${DAS_DIR}/syslibs/etc/ssl/certs \
-    && cp -f /usr/lib/${LIB_SUBDIR}/libssl.so.3 ${DAS_DIR}/syslibs/usr/lib/${LIB_SUBDIR}/ 2>/dev/null || true \
-    && cp -f /usr/lib/${LIB_SUBDIR}/libcrypto.so.3 ${DAS_DIR}/syslibs/usr/lib/${LIB_SUBDIR}/ 2>/dev/null || true \
-    && cp -f /lib/${LIB_SUBDIR}/libz.so.1 ${DAS_DIR}/syslibs/lib/${LIB_SUBDIR}/ 2>/dev/null || true \
-    && cp -f /usr/lib/${LIB_SUBDIR}/libzstd.so.1 ${DAS_DIR}/syslibs/usr/lib/${LIB_SUBDIR}/ 2>/dev/null || true \
-    && cp -f /usr/lib/${LIB_SUBDIR}/libsnappy.so.1 ${DAS_DIR}/syslibs/usr/lib/${LIB_SUBDIR}/ 2>/dev/null || true \
-    && cp -f /lib/${LIB_SUBDIR}/libbz2.so.1.0 ${DAS_DIR}/syslibs/lib/${LIB_SUBDIR}/ 2>/dev/null || true \
-    && cp -f /lib/${LIB_SUBDIR}/liblzma.so.5 ${DAS_DIR}/syslibs/lib/${LIB_SUBDIR}/ 2>/dev/null || true \
-    && cp -f /usr/lib/${LIB_SUBDIR}/libcurl.so.4 ${DAS_DIR}/syslibs/usr/lib/${LIB_SUBDIR}/ 2>/dev/null || true \
-    && cp -f /usr/lib/${LIB_SUBDIR}/libevent-2.1.so.7 ${DAS_DIR}/syslibs/usr/lib/${LIB_SUBDIR}/ 2>/dev/null || true \
+    && mkdir -p ${DAS_DIR}/syslibs/usr/lib/${LIB_SUBDIR} \
+                ${DAS_DIR}/syslibs/lib/${LIB_SUBDIR} \
+                ${DAS_DIR}/syslibs/etc/ssl/certs \
+    # --- usr/lib/ libraries ---
+    && USR_LIBS=" \
+        libssl.so.3 \
+        libcrypto.so.3 \
+        libzstd.so.1 \
+        libsnappy.so.1 \
+        libcurl.so.4 \
+        libevent-2.1.so.7 \
+        # Libs from here and on are related to the libpqxx lib, used by the database-adapter.
+        libgssapi_krb5.so.2 \
+        libkrb5.so.3 \
+        libk5crypto.so.3 \
+        libcom_err.so.2 \
+        libkrb5support.so.0 \
+        liblber-2.5.so.0 \
+        libkeyutils.so.1 \
+        libp11-kit.so.0 \
+        libidn2.so.0 \
+        libunistring.so.2 \
+        libtasn1.so.6 \
+        libnettle.so.8 \
+        libhogweed.so.6 \
+        libgmp.so.10 \
+        libffi.so.8 \
+        libldap-2.5.so.0 \
+        libsasl2.so.2 \
+        libgnutls.so.30 \
+    " \
+    # lib/ folder libraries ---
+    && LIB_LIBS=" \
+        libz.so.1 \
+        libbz2.so.1.0 \
+        liblzma.so.5 \
+    " \
+    && for lib in $USR_LIBS; do \
+        cp -f /usr/lib/${LIB_SUBDIR}/$lib ${DAS_DIR}/syslibs/usr/lib/${LIB_SUBDIR}/ 2>/dev/null || true; \
+    done \
+    && for lib in $LIB_LIBS; do \
+        cp -f /lib/${LIB_SUBDIR}/$lib ${DAS_DIR}/syslibs/lib/${LIB_SUBDIR}/ 2>/dev/null || true; \
+    done \
     && cp -f /etc/ssl/certs/ca-certificates.crt ${DAS_DIR}/syslibs/etc/ssl/certs/ 2>/dev/null || true
 
 FROM gcr.io/distroless/cc-debian12:nonroot AS runner
@@ -61,8 +94,8 @@ COPY --from=builder /opt/das/syslibs/usr/lib/ /usr/lib/
 COPY --from=builder /opt/das/syslibs/lib/ /lib/
 COPY --from=builder /opt/das/syslibs/etc/ssl/certs/ /etc/ssl/certs/
 
-ENV LD_LIBRARY_PATH=/usr/local/lib:/usr/lib/x86_64-linux-gnu:/usr/lib/aarch64-linux-gnu:/lib/x86_64-linux-gnu:/lib/aarch64-linux-gnu
+ENV LD_LIBRARY_PATH=/opt/das/lib:/usr/local/lib:/usr/lib/x86_64-linux-gnu:/usr/lib/aarch64-linux-gnu:/lib/x86_64-linux-gnu:/lib/aarch64-linux-gnu
 
 USER nonroot:nonroot
 
-CMD [ "query_broker" ]
+CMD [ "busnode" ]

.github/scripts/bazel_build.sh

@@ -67,6 +67,7 @@ BAZEL_BINARY_TARGETS=(
   "//:tests_db_loader"
   "//:busnode"
   "//:busclient"
+  "//:database_adapter"
 )
 
 BAZEL_BINARY_OUTPUTS=(
@@ -78,6 +79,7 @@ BAZEL_BINARY_OUTPUTS=(
   "bazel-bin/tests_db_loader"
   "bazel-bin/busnode"
   "bazel-bin/busclient"
+  "bazel-bin/database_adapter"
 )
 
 BAZEL_LIB_OUTPUTS=(
@@ -181,6 +183,21 @@ if [ -d "$LIB_DIR" ]; then
   done
 fi
 
+echo "[INFO] Copying database-adapter shared library dependencies into $LIB_DIR..."
+
+find "$BIN_DIR" -type f -executable -name "database_adapter" | while IFS= read -r binfile; do
+    ldd "$binfile" | awk '/=> \// { print $3 }' | while IFS= read -r dep; do
+        dep_base=$(basename "$dep")
+        case "$dep_base" in
+            "libpq.so.5")
+                if [ -f "$dep" ]; then
+                    cp -f "$dep" "$LIB_DIR"
+                fi
+                ;;
+        esac
+    done
+done
+
 echo "[INFO] Build finished successfully."
 echo "[INFO] Binaries in: $BIN_DIR"
 echo "[INFO] Libraries in: $LIB_DIR"

.github/scripts/setup_bazel.sh

@@ -29,12 +29,14 @@ sudo apt update -y
 sudo apt install -y \
   git build-essential curl protobuf-compiler python3 python3-pip \
   cmake unzip uuid-runtime lcov bc \
-  libevent-dev libssl-dev pkg-config libncurses5
+  libevent-dev libssl-dev pkg-config libncurses5 libpq-dev
 
 echo "[INFO] Cleaning apt cache..."
 sudo apt clean -y || true
 sudo rm -rf /var/lib/apt/lists/* || true
 
+# Installing bazelisk and copying other assets.
+
 echo "[INFO] Installing 3rd-party tools (bazelisk, buildifier)..."
 
 if [[ ! -f "${ASSETS_DIR}/3rd-party.tgz" ]]; then
@@ -119,6 +121,28 @@ if ! id "builder" &>/dev/null; then
   sudo useradd -ms /bin/bash builder
 fi
 
+echo "[INFO] Installing libpqxx (PostgreSQL C++ client)..."
+
+if [[ ! -f "${ASSETS_DIR}/libpqxx-7.10.5.tar.gz" ]]; then
+  echo "[ERROR] ${ASSETS_DIR}/libpqxx-7.10.5.tar.gz not found."
+  exit 1
+fi
+
+cp "${ASSETS_DIR}/libpqxx-7.10.5.tar.gz" "${TMP_DIR}/"
+cd "${TMP_DIR}"
+tar xzvf libpqxx-7.10.5.tar.gz
+cd libpqxx-7.10.5
+
+cmake -S . -B build \
+    -DCMAKE_BUILD_TYPE=RelWithDebInfo \
+    -DBUILD_TESTING=OFF \
+    -DCMAKE_INSTALL_PREFIX=/usr/local
+
+cmake --build build -j"$(nproc)"
+
+sudo cmake --install build
+sudo ldconfig
+
 echo "[INFO] Configuring git safe.directory for ${DAS_DIR}..."
 sudo -u builder git config --global --add safe.directory "${DAS_DIR}"
 

.github/workflows/run-unit-tests.yml

@@ -72,5 +72,20 @@ jobs:
             sleep 1
           done
 
+      - name: Starting Postgres Server and load database
+        run: |-
+          docker run -d --name pg_test -e POSTGRES_PASSWORD=test -e POSTGRES_DB=postgres_wrapper_test -p 5433:5432 postgres:15
+
+          for i in {1..30}; do
+            if docker exec pg_test pg_isready -U postgres -d postgres_wrapper_test; then
+              echo "Postgres is ready"
+              break
+            fi
+            echo "Waiting for Postgres..."
+            sleep 1
+          done
+
+          docker exec -i pg_test psql -U postgres -d postgres_wrapper_test < src/scripts/postgres_setup.sql
+
       - name: Execute Unit Test Suite
         run: make test-all

Makefile

@@ -61,6 +61,9 @@ setup-inference-toy-problem:
 run-tests-db-loader:
 	@bash -x src/scripts/run.sh tests_db_loader $(OPTIONS)
 
+run-adapter:
+	@bash -x src/scripts/run.sh database_adapter $(OPTIONS)
+
 setup-nunet-dms:
 	@bash -x src/scripts/setup-nunet-dms.sh
 

src/BUILD

@@ -44,6 +44,7 @@ cc_shared_library(
         "//atomdb:atomdb_lib",
         "//attention_broker:attention_broker_lib",
         "//commons:commons_lib",
+        "//db_adapter:db_adapter_lib",
         "//distributed_algorithm_node:distributed_algorithm_node_lib",
         "//hasher:hasher_lib",
         "//metta:metta_lib",
@@ -188,6 +189,27 @@ cc_binary(
     ],
 )
 
+cc_binary(
+    name = "database_adapter",
+    srcs = [],
+    defines = ["BAZEL_BUILD"],
+    linkopts = [
+        "-L/usr/local/lib",
+        "-lpqxx",
+        "-lpq",
+        "-lhiredis_cluster",
+        "-lhiredis",
+        "-lmongocxx",
+        "-lbsoncxx",
+    ],
+    linkstatic = 1,
+    deps = [
+        "//commons:commons_lib",
+        "//tests/main:database_adapter_main_lib",
+        "@mbedtls",
+    ],
+)
+
 cc_binary(
     name = "busnode",
     srcs = [],

src/commons/HandleTrie.cc

@@ -26,7 +26,11 @@ HandleTrie::TrieNode::~TrieNode() {
         delete children[i];
     }
     delete[] children;
-    delete value;
+    // TODO: Remove this check once improve insert()
+    if (value != NULL) {
+        delete value;
+        value = NULL;
+    }
 }
 
 string HandleTrie::TrieValue::to_string() { return ""; }
@@ -269,3 +273,8 @@ void HandleTrie::traverse(bool keep_root_locked,
         root->trie_node_mutex.unlock();
     }
 }
+
+bool HandleTrie::exists(const string& key) {
+    TrieNode* node = lookup_node(key);
+    return node != NULL ? true : false;
+}

src/commons/HandleTrie.h

@@ -74,6 +74,8 @@ class HandleTrie {
      */
     TrieValue* lookup(const string& key);
 
+    bool exists(const string& key);
+
     /**
      * Remove a key from this HandleTrie and its associated value.
      *

src/db_adapter/BUILD

@@ -0,0 +1,96 @@
+load("@rules_cc//cc:cc_library.bzl", "cc_library")
+
+package(default_visibility = ["//visibility:public"])
+
+cc_library(
+    name = "db_adapter_lib",
+    includes = ["."],
+    deps = [
+        ":context_loader",
+        ":data_mapper",
+        ":data_types",
+        ":database_connection",
+        ":database_wrapper",
+        ":pipeline",
+        "//commons:commons_lib",
+        "//commons/atoms:atoms_lib",
+        "//db_adapter/postgres:postgres_lib",
+    ],
+)
+
+cc_library(
+    name = "data_types",
+    hdrs = ["DataTypes.h"],
+    includes = ["."],
+    deps = [
+        "//commons:commons_lib",
+        "//commons/atoms:atoms_lib",
+    ],
+)
+
+cc_library(
+    name = "data_mapper",
+    srcs = ["DataMapper.cc"],
+    hdrs = ["DataMapper.h"],
+    includes = ["."],
+    deps = [
+        ":data_types",
+        "//commons:commons_lib",
+        "//commons/atoms:atoms_lib",
+    ],
+)
+
+cc_library(
+    name = "database_wrapper",
+    srcs = ["DatabaseWrapper.cc"],
+    hdrs = ["DatabaseWrapper.h"],
+    includes = ["."],
+    deps = [
+        ":data_mapper",
+        ":data_types",
+        ":database_connection",
+        "//commons:commons_lib",
+        "//commons/atoms:atoms_lib",
+    ],
+)
+
+cc_library(
+    name = "context_loader",
+    srcs = ["ContextLoader.cc"],
+    hdrs = ["ContextLoader.h"],
+    includes = ["."],
+    deps = [
+        ":data_types",
+        "//commons:commons_lib",
+        "@nlohmann_json//:json",
+    ],
+)
+
+cc_library(
+    name = "database_connection",
+    srcs = ["DatabaseConnection.cc"],
+    hdrs = ["DatabaseConnection.h"],
+    includes = ["."],
+    deps = [
+        ":data_types",
+        "//commons:commons_lib",
+        "//commons/atoms:atoms_lib",
+        "//commons/processor:processor_lib",
+    ],
+)
+
+cc_library(
+    name = "pipeline",
+    srcs = ["Pipeline.cc"],
+    hdrs = ["Pipeline.h"],
+    includes = ["."],
+    deps = [
+        ":data_types",
+        ":database_wrapper",
+        "//atomdb",
+        "//atomdb:atomdb_singleton",
+        "//commons:commons_lib",
+        "//commons/atoms:atoms_lib",
+        "//db_adapter/postgres:postgres_lib",
+    ],
+)