#9: Created test VM infrastructure

flavors/test-udf-runner-cpp-v1/ci.json

@@ -6,7 +6,7 @@
       {
         "name": "smoke",
         "files": [],
-        "folders": ["smoke"],
+        "folders": ["smoke", "cpp_test"],
         "goal": "base_test_build_run",
         "generic_language_tests": []
       }

flavors/test-udf-runner-cpp-v1/flavor_base/build_run/Dockerfile

@@ -7,7 +7,7 @@ RUN mkdir /exaudfclient /exaudf
 COPY exaudfclient/ /exaudfclient/
 
 WORKDIR /exaudfclient/
-RUN ["/bin/bash", "-c", "source /env && bash build.sh --lockfile_mode=off --config no-tty --config fast-binary --define benchmark=true --define bash=true //:udf_runner_cpp_v1_gen //:udf_runner_cpp_v1_static_gen"]
+RUN ["/bin/bash", "-c", "source /env && bash build.sh --lockfile_mode=off --config no-tty --config fast-binary --define benchmark=true --define bash=true --define test_vm=true //:udf_runner_cpp_v1_gen //:udf_runner_cpp_v1_static_gen"]
 
 RUN cp -r -L bazel-bin/* /exaudf
 RUN mv /exaudf/udf_runner_cpp_v1 /exaudf/exaudfclient

flavors/test-udf-runner-cpp-v1/flavor_base/language_definition

@@ -1 +1 @@
-CPP_BENCHMARK=localzmq+protobuf:///{{ bucketfs_name }}/{{ bucket_name }}/{{ path_in_bucket }}{{ release_name }}?lang=benchmark#buckets/{{ bucketfs_name }}/{{ bucket_name }}/{{ path_in_bucket }}{{ release_name }}/exaudf/exaudfclient CPP_STREAMING=localzmq+protobuf:///{{ bucketfs_name }}/{{ bucket_name }}/{{ path_in_bucket }}{{ release_name }}?lang=streaming#buckets/{{ bucketfs_name }}/{{ bucket_name }}/{{ path_in_bucket }}{{ release_name }}/exaudf/exaudfclient
+CPP_BENCHMARK=localzmq+protobuf:///{{ bucketfs_name }}/{{ bucket_name }}/{{ path_in_bucket }}{{ release_name }}?lang=benchmark#buckets/{{ bucketfs_name }}/{{ bucket_name }}/{{ path_in_bucket }}{{ release_name }}/exaudf/exaudfclient CPP_STREAMING=localzmq+protobuf:///{{ bucketfs_name }}/{{ bucket_name }}/{{ path_in_bucket }}{{ release_name }}?lang=streaming#buckets/{{ bucketfs_name }}/{{ bucket_name }}/{{ path_in_bucket }}{{ release_name }}/exaudf/exaudfclient CPP_TEST=localzmq+protobuf:///{{ bucketfs_name }}/{{ bucket_name }}/{{ path_in_bucket }}{{ release_name }}?lang=test#buckets/{{ bucketfs_name }}/{{ bucket_name }}/{{ path_in_bucket }}{{ release_name }}/exaudf/exaudfclient

flavors/test-udf-runner-cpp-v1/flavor_base/language_definitions.json

@@ -18,6 +18,15 @@
         "executable": "/exaudf/exaudfclient"
       },
       "deprecation": null
+    },
+    {
+      "protocol": "localzmq+protobuf",
+      "aliases": ["CPP_TEST"],
+      "parameters": [{"key": "lang", "value": "test"}],
+      "udf_client_path": {
+        "executable": "/exaudf/exaudfclient"
+      },
+      "deprecation": null
     }
   ]
 }

test_container/tests/test/cpp_test/test_vm.py

@@ -0,0 +1,51 @@
+#!/usr/bin/env python3
+
+from exasol_python_test_framework import udf
+
+
+class TestVmTest(udf.TestCase):
+    def setUp(self):
+        self.query("DROP SCHEMA CPP_TEST CASCADE", ignore_errors=True)
+        self.query("CREATE SCHEMA CPP_TEST")
+        self.query("OPEN SCHEMA CPP_TEST")
+
+    def tearDown(self):
+        self.query("DROP SCHEMA CPP_TEST CASCADE", ignore_errors=True)
+
+    def test_emit_metadata(self):
+        self.query(
+            "CREATE CPP_TEST SET SCRIPT emit_metadata(ignored VARCHAR(1)) "
+            "EMITS (script_user VARCHAR(128), script_code VARCHAR(128)) "
+            "AS emit_metadata"
+        )
+
+        rows = self.query("SELECT emit_metadata('x') FROM DUAL")
+        self.assertRowsEqual([("SYS", "emit_metadata")], rows)
+
+    def test_forward_input(self):
+        self.query(
+            "CREATE CPP_TEST SET SCRIPT forward_input(value VARCHAR(128)) "
+            "EMITS (value VARCHAR(128)) AS forward_input"
+        )
+        self.query("CREATE TABLE input_values (value VARCHAR(128))")
+        self.query("INSERT INTO input_values VALUES ('first'), (NULL), ('last')")
+
+        rows = self.query("""
+            SELECT value
+            FROM (SELECT forward_input(value) AS value FROM input_values)
+            ORDER BY value NULLS FIRST
+        """)
+        self.assertRowsEqual([(None,), ("first",), ("last",)], rows)
+
+    def test_unsupported_strategy_fails(self):
+        self.query(
+            "CREATE CPP_TEST SET SCRIPT unknown_strategy() "
+            "EMITS (value VARCHAR(128)) AS unknown_strategy"
+        )
+
+        with self.assertRaisesRegex(Exception, "unsupported test strategy: unknown_strategy"):
+            self.query("SELECT * FROM unknown_strategy()")
+
+
+if __name__ == "__main__":
+    udf.main()

udf-runner-cpp/v1/BUILD

@@ -9,6 +9,9 @@ VM_ENABLED_DEPS=select({
     }) + select({
         "@exaudfclient_base//:bash": ["//streaming_container:streamingcontainer"],
         "//conditions:default": []
+    }) + select({
+        "@exaudfclient_base//:test_vm": ["//test_container:test_container"],
+        "//conditions:default": []
     })
 
 cc_binary(

udf-runner-cpp/v1/base/BUILD

@@ -17,6 +17,11 @@ config_setting(
     define_values = {"bash": "true"},
 )
 
+config_setting(
+    name = "test_vm",
+    define_values = {"test_vm": "true"},
+)
+
 config_setting(
     name = "fast_binary",
     define_values = {

udf-runner-cpp/v1/exa_udf_client.cc

@@ -39,7 +39,8 @@ bool ExaUdfClient::validate_arguments(int argc, char** argv) {
     }
 
     if (!((strcmp(argv[2], "lang=streaming") == 0)
-        || (strcmp(argv[2], "lang=benchmark") == 0))) {
+        || (strcmp(argv[2], "lang=benchmark") == 0)
+        || (strcmp(argv[2], "lang=test") == 0))) {
         usage(argv[0]);
         return false;
     }
@@ -49,7 +50,7 @@ bool ExaUdfClient::validate_arguments(int argc, char** argv) {
 
 void ExaUdfClient::usage(const std::string& programName) {
     std::cerr << "Usage: " << programName
-              << " <socket> lang=streaming|lang=benchmark <scriptOptionsParserVersion=1|2>"
+              << " <socket> lang=streaming|lang=benchmark|lang=test <scriptOptionsParserVersion=1|2>"
               << std::endl;
 }
 

udf-runner-cpp/v1/exa_vm_factory.cc

@@ -10,6 +10,10 @@
 #include "benchmark_container/benchmark_container.h"
 #endif
 
+#ifdef ENABLE_TEST_VM
+#include "test_container/test_container.h"
+#endif
+
 std::function<SWIGVMContainers::SWIGVM*()> create_vm(const std::string& argv_lang, bool use_ctpg_options_parser) {
     if(argv_lang.compare("lang=streaming") == 0) {
         #ifdef ENABLE_STREAMING_VM
@@ -25,6 +29,13 @@ std::function<SWIGVMContainers::SWIGVM*()> create_vm(const std::string& argv_lan
             throw SWIGVMContainers::SWIGVM::exception("this exaudfclient has been compilied without Benchmark support");
         #endif
     }
+    else if(argv_lang.compare("lang=test") == 0) {
+        #ifdef ENABLE_TEST_VM
+            return []() { return new TestVM(false); };
+        #else
+            throw SWIGVMContainers::SWIGVM::exception("this exaudfclient has been compiled without test VM support");
+        #endif
+    }
     else {
         throw SWIGVMContainers::SWIGVM::exception("unsupported language specified in argv");
     }   

udf-runner-cpp/v1/test_container/BUILD

@@ -0,0 +1,11 @@
+package(default_visibility = ["//visibility:public"])
+
+cc_library(
+    name = "test_container",
+    srcs = ["test_container.cc"],
+    hdrs = ["test_container.h"],
+    deps = [
+        "@exaudfclient_base//exaudflib:exaudflib-deps",
+        "@exaudfclient_base//exaudflib:header",
+    ],
+)

udf-runner-cpp/v1/test_container/test_container.cc

@@ -0,0 +1,92 @@
+#include "test_container.h"
+
+#include <cstring>
+#include <string>
+
+using namespace SWIGVMContainers;
+
+namespace {
+
+constexpr const char* kEmitMetadata = "emit_metadata";
+constexpr const char* kForwardInput = "forward_input";
+
+void requireStringColumn(
+    SWIGMetadata& metadata,
+    unsigned int column,
+    bool input,
+    const char* strategy) {
+    const SWIGVM_datatype_e type = input
+        ? metadata.inputColumnType(column)
+        : metadata.outputColumnType(column);
+    if (type != STRING) {
+        throw SWIGVM::exception(
+            (std::string(strategy) + " requires VARCHAR "
+                + (input ? "input" : "output") + " columns").c_str());
+    }
+}
+
+}  // namespace
+
+TestVM::TestVM(bool checkOnly)
+    : meta(), input(), output(&input) {
+    if (meta.inputType() != MULTIPLE || meta.outputType() != MULTIPLE) {
+        throw SWIGVM::exception("TEST language container only supports SET-EMITS UDFs");
+    }
+}
+
+void TestVM::shutdown() {}
+
+bool TestVM::run() {
+    const std::string strategy(meta.scriptCode());
+    if (strategy == kEmitMetadata) {
+        emitMetadata();
+    } else if (strategy == kForwardInput) {
+        forwardInput();
+    } else {
+        throw SWIGVM::exception(
+            ("unsupported test strategy: " + strategy).c_str());
+    }
+    return true;
+}
+
+void TestVM::emitMetadata() {
+    if (meta.outputColumnCount() != 2) {
+        throw SWIGVM::exception("emit_metadata requires exactly two VARCHAR output columns");
+    }
+    requireStringColumn(meta, 0, false, kEmitMetadata);
+    requireStringColumn(meta, 1, false, kEmitMetadata);
+
+    const char* script_user = meta.scopeUser();
+    const char* script_code = meta.scriptCode();
+    output.setString(0, script_user, std::strlen(script_user));
+    output.setString(1, script_code, std::strlen(script_code));
+    output.next();
+    output.flush();
+}
+
+void TestVM::forwardInput() {
+    if (meta.inputColumnCount() != 1 || meta.outputColumnCount() != 1) {
+        throw SWIGVM::exception("forward_input requires exactly one VARCHAR input and output column");
+    }
+    requireStringColumn(meta, 0, true, kForwardInput);
+    requireStringColumn(meta, 0, false, kForwardInput);
+
+    while (!input.eot()) {
+        size_t length = 0;
+        const char* value = input.getString(0, &length);
+        if (input.wasNull()) {
+            output.setNull(0);
+        } else {
+            output.setString(0, value, length);
+        }
+        output.next();
+        input.next();
+    }
+    output.flush();
+}
+
+const char* TestVM::singleCall(
+    single_call_function_id_e fn,
+    const ExecutionGraph::ScriptDTO& args) {
+    throw SWIGVM::exception("singleCall is not supported for the TEST language container");
+}

udf-runner-cpp/v1/test_container/test_container.h

@@ -0,0 +1,28 @@
+#ifndef TEST_CONTAINER_H
+#define TEST_CONTAINER_H
+
+#include "exaudflib/swig/swig_meta_data.h"
+#include "exaudflib/swig/swig_result_handler.h"
+#include "exaudflib/swig/swig_table_iterator.h"
+#include "exaudflib/vm/swig_vm.h"
+
+class TestVM : public SWIGVMContainers::SWIGVM {
+public:
+    explicit TestVM(bool checkOnly);
+    void shutdown() override;
+    bool run() override;
+    const char* singleCall(
+        SWIGVMContainers::single_call_function_id_e fn,
+        const ExecutionGraph::ScriptDTO& args) override;
+    bool useZmqSocketLocks() override { return true; }
+
+private:
+    SWIGVMContainers::SWIGMetadata meta;
+    SWIGVMContainers::SWIGTableIterator input;
+    SWIGVMContainers::SWIGResultHandler output;
+
+    void emitMetadata();
+    void forwardInput();
+};
+
+#endif

udf-runner-cpp/v1/variables.bzl

@@ -6,4 +6,8 @@ STREAMING_VM_ENABLED_DEFINE=select({
         "@exaudfclient_base//:bash": ["ENABLE_STREAMING_VM"],
         "//conditions:default": []
     }) 
-VM_ENABLED_DEFINES=BENCHMARK_VM_ENABLED_DEFINE+STREAMING_VM_ENABLED_DEFINE
+TEST_VM_ENABLED_DEFINE=select({
+        "@exaudfclient_base//:test_vm": ["ENABLE_TEST_VM"],
+        "//conditions:default": []
+    })
+VM_ENABLED_DEFINES=BENCHMARK_VM_ENABLED_DEFINE+STREAMING_VM_ENABLED_DEFINE+TEST_VM_ENABLED_DEFINE