From 72c3ff1c2355033de1521943e4c0bca07fc85853 Mon Sep 17 00:00:00 2001 From: "Shankar.KV" Date: Wed, 22 Jul 2026 14:18:12 +0530 Subject: [PATCH] Create test VM infrastructure --- flavors/test-udf-runner-cpp-v1/ci.json | 2 +- .../flavor_base/build_run/Dockerfile | 2 +- .../flavor_base/language_definition | 2 +- .../flavor_base/language_definitions.json | 9 ++ .../tests/test/cpp_test/__init__.py | 0 .../__pycache__/test_vm.cpython-312.pyc | Bin 0 -> 2953 bytes test_container/tests/test/cpp_test/test_vm.py | 51 ++++++++++ udf-runner-cpp/v1/BUILD | 3 + udf-runner-cpp/v1/base/BUILD | 5 + udf-runner-cpp/v1/exa_udf_client.cc | 5 +- udf-runner-cpp/v1/exa_vm_factory.cc | 11 +++ udf-runner-cpp/v1/test_container/BUILD | 11 +++ .../v1/test_container/test_container.cc | 92 ++++++++++++++++++ .../v1/test_container/test_container.h | 28 ++++++ udf-runner-cpp/v1/variables.bzl | 6 +- 15 files changed, 221 insertions(+), 6 deletions(-) create mode 100644 test_container/tests/test/cpp_test/__init__.py create mode 100644 test_container/tests/test/cpp_test/__pycache__/test_vm.cpython-312.pyc create mode 100644 test_container/tests/test/cpp_test/test_vm.py create mode 100644 udf-runner-cpp/v1/test_container/BUILD create mode 100644 udf-runner-cpp/v1/test_container/test_container.cc create mode 100644 udf-runner-cpp/v1/test_container/test_container.h diff --git a/flavors/test-udf-runner-cpp-v1/ci.json b/flavors/test-udf-runner-cpp-v1/ci.json index c2c6a86..e7a9679 100644 --- a/flavors/test-udf-runner-cpp-v1/ci.json +++ b/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": [] } diff --git a/flavors/test-udf-runner-cpp-v1/flavor_base/build_run/Dockerfile b/flavors/test-udf-runner-cpp-v1/flavor_base/build_run/Dockerfile index c058d8a..64e19cc 100644 --- a/flavors/test-udf-runner-cpp-v1/flavor_base/build_run/Dockerfile +++ b/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 diff --git a/flavors/test-udf-runner-cpp-v1/flavor_base/language_definition b/flavors/test-udf-runner-cpp-v1/flavor_base/language_definition index 6148554..9bac8be 100644 --- a/flavors/test-udf-runner-cpp-v1/flavor_base/language_definition +++ b/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 diff --git a/flavors/test-udf-runner-cpp-v1/flavor_base/language_definitions.json b/flavors/test-udf-runner-cpp-v1/flavor_base/language_definitions.json index e7cd83c..a12b4c7 100644 --- a/flavors/test-udf-runner-cpp-v1/flavor_base/language_definitions.json +++ b/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 } ] } diff --git a/test_container/tests/test/cpp_test/__init__.py b/test_container/tests/test/cpp_test/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/test_container/tests/test/cpp_test/__pycache__/test_vm.cpython-312.pyc b/test_container/tests/test/cpp_test/__pycache__/test_vm.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..07da5844485aacbf17da819fd976639ee12b1229 GIT binary patch literal 2953 zcma)8O-vg{6rR~1vvIwF0!o_%l2wt47xJT2N}Gn}$FY|{L=3e}R8Z2@%DWIJw%3_m z2fRw9R7zEHN~<2Y^-$$dM2|h>*q(bKDJapNDpjh~n*~&ra_XDiwT%s?8B5P^-ru~L z_wAecO%%fj#$QL(J8lS}Kk1}4_<-5E4a_pqk&g4IfM+pA47!SR<`L3a1K;6l4vqg+ z*5_wggVVXYQG`f0gTU5Xtzk#G)nJNIg|F~(=x;igvCEeAxon{G{%|na8Ukh+8E6*k zXqM6OBQ(p>Y-Tx~14X_!dxgiJv*|m2|eo`uxs7GhA>LHK)!dH`y-`M;Wis#Y8GE$un4{qSgbyDb{;># z4YYY%b(Vt4*Fln2GEnf;_*6zR3OQRX7`CQswifjfqf0kpsrXnd6+I)%l5%}Kla``Z zmgGvdTDA<*P=9bvJ|P8~tf?DK8d-{^o8vnJX=Or*x2`(4I4Db_smbe7VmdYOTjhim;GJ?q^NF78wL zpCN4f9#N}n+poV~`L-79|Hi2In0DxyM>_vQD}F*r=k`AMnKQ|PMszh-ES2r(eJx*Z z1wnwHy;OjYCV%Htuu>*=WkQj>n5yS!Nqfb1M#htAC6$rJlbOk8^Bb{=X$4?AIF}=q zJqR$3CZ{JR=>K3|qY~%4u(V?Y$n;igQh30O2I{7C{ggbI`@8c@cV;a&MVd?{l$3O3 zM)Iv`K51!mJeAJW&16=_Nw~gvFKu8rnk+d4=ppY?cZ|F>|xt&FC@+rWx1Aa9hcqE_!bO!qT0)x>EM4*yIb_OkA zxD%Y4`vARMyjz6ZK!s{(wlQCc);ptL&3em(p6EfM)xjAq7p-!sWD?uZCEA>%TCDTZ zE@LSG+>qZGTlG;{X8nfvcF+*DoMl)kW8PSF4=9USqh#mI;;vimxYrbSBvRjxI;Z9G zmNV4aA=}%+y9`;Efi!{Xf)A!JI@tYE9C$7cJRM&=qRkdL1)=Ox zPwGTdD8MU6Rg-ea?K2iN%gn2#ihbKGdJi3SjzESBCb>%}W?g|EU0j3f$y;-P$~YQ< zybo+6C0F>w-_S)e3<~ lang=streaming|lang=benchmark " + << " lang=streaming|lang=benchmark|lang=test " << std::endl; } diff --git a/udf-runner-cpp/v1/exa_vm_factory.cc b/udf-runner-cpp/v1/exa_vm_factory.cc index 3f4ef73..97f9888 100644 --- a/udf-runner-cpp/v1/exa_vm_factory.cc +++ b/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 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 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"); } diff --git a/udf-runner-cpp/v1/test_container/BUILD b/udf-runner-cpp/v1/test_container/BUILD new file mode 100644 index 0000000..f78abc4 --- /dev/null +++ b/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", + ], +) diff --git a/udf-runner-cpp/v1/test_container/test_container.cc b/udf-runner-cpp/v1/test_container/test_container.cc new file mode 100644 index 0000000..76e6cf4 --- /dev/null +++ b/udf-runner-cpp/v1/test_container/test_container.cc @@ -0,0 +1,92 @@ +#include "test_container.h" + +#include +#include + +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"); +} diff --git a/udf-runner-cpp/v1/test_container/test_container.h b/udf-runner-cpp/v1/test_container/test_container.h new file mode 100644 index 0000000..d34cb63 --- /dev/null +++ b/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 diff --git a/udf-runner-cpp/v1/variables.bzl b/udf-runner-cpp/v1/variables.bzl index 02a57b8..ba55ef7 100644 --- a/udf-runner-cpp/v1/variables.bzl +++ b/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