bill/AQuery
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Added var(s), stddev(s). New ITC, stats. bugfix on aggregations.

Browse Source
This commit is contained in:
bill
2022-11-04 06:10:36 +08:00
parent ba21da23a3
commit 50740fe6ac
21 changed files with 495 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/aggregations.h
+123 -20
View File
@@ -202,6 +202,102 @@ decayed_t<VT, types::GetFPType<types::GetLongType<T>>> avgw(uint32_t w, const VT
return ret;
}
template<class T, template<typename ...> class VT, bool sd = false>
decayed_t<VT, types::GetFPType<types::GetLongType<T>>> varw(uint32_t w, const VT<T>& arr) {
using FPType = types::GetFPType<types::GetLongType<T>>;
const uint32_t& len = arr.size;
decayed_t<VT, FPType> ret(len);
uint32_t i = 0;
types::GetLongType<T> s{};
w = w > len ? len : w;
FPType EnX {}, MnX{};
if (len) {
s = arr[0];
MnX = 0;
EnX = arr[0];
ret[i++] = 0;
}
for (; i < len; ++i){
s += arr[i];
FPType _EnX = s / (FPType)(i + 1);
MnX += (arr[i] - EnX) * (arr[i] - _EnX);
EnX = _EnX;
ret[i] = MnX / (FPType)(i + 1);
if constexpr(sd) ret[i-1] = sqrt(ret[i-1]);
}
const float rw = 1.f / (float)w;
s *= rw;
for (; i < len; ++i){
const auto dw = arr[i] - arr[i - w - 1];
const auto sw = arr[i] + arr[i - w - 1];
const auto dex = dw * rw;
ret[i] = ret[i-1] - dex*(s + s + dex - sw);
if constexpr(sd) ret[i-1] = sqrt(ret[i-1]);
s += dex;
}
if constexpr(sd)
if(i)
ret[i-1] = sqrt(ret[i-1]);
return ret;
}
template<class T, template<typename ...> class VT>
types::GetFPType<types::GetLongType<decays<T>>> var(const VT<T>& arr) {
typedef types::GetFPType<types::GetLongType<decays<T>>> FPType;
const uint32_t& len = arr.size;
uint32_t i = 0;
types::GetLongType<T> s{0};
types::GetLongType<T> ssq{0};
if (len) {
s = arr[0];
ssq = arr[0] * arr[0];
}
for (; i < len; ++i){
s += arr[i];
ssq += arr[i] * arr[i];
}
return (ssq - s * s / (FPType)(len + 1)) / (FPType)(len + 1);
}
template<class T, template<typename ...> class VT, bool sd = false>
decayed_t<VT, types::GetFPType<types::GetLongType<T>>> vars(const VT<T>& arr) {
typedef types::GetFPType<types::GetLongType<T>> FPType;
const uint32_t& len = arr.size;
decayed_t<VT, FPType> ret(len);
uint32_t i = 0;
types::GetLongType<T> s{};
FPType MnX{};
FPType EnX {};
if (len) {
s = arr[0];
MnX = 0;
EnX = arr[0];
ret[i++] = 0;
}
for (; i < len; ++i){
s += arr[i];
FPType _EnX = s / (FPType)(i + 1);
MnX += (arr[i] - EnX) * (arr[i] - _EnX);
printf("%d %ld ", arr[i], MnX);
EnX = _EnX;
ret[i] = MnX / (FPType)(i + 1);
if constexpr(sd) ret[i] = sqrt(ret[i]);
}
return ret;
}
template<class T, template<typename ...> class VT>
types::GetFPType<types::GetLongType<decays<T>>> stddev(const VT<T>& arr) {
return sqrt(var(arr));
}
template<class T, template<typename ...> class VT>
decayed_t<VT, types::GetFPType<types::GetLongType<T>>> stddevs(const VT<T>& arr) {
return vars<T, VT, true>(arr);
}
template<class T, template<typename ...> class VT>
decayed_t<VT, types::GetFPType<types::GetLongType<T>>> stddevw(uint32_t w, const VT<T>& arr) {
return varw<T, VT, true>(w, arr);
}
// use getSignedType
template<class T, template<typename ...> class VT>
decayed_t<VT, T> deltas(const VT<T>& arr) {
@@ -251,26 +347,33 @@ T first(const VT<T>& arr) {
}
#define __DEFAULT_AGGREGATE_FUNCTION__(NAME, RET) \
template <class T> constexpr inline T NAME(const T& v) { return RET; }
template <class T> constexpr T NAME(const T& v) { return RET; }
// non-aggreation count. E.g. SELECT COUNT(col) from table;
template <class T> constexpr inline T count(const T& v) { return 1; }
template <class T> constexpr inline T max(const T& v) { return v; }
template <class T> constexpr inline T min(const T& v) { return v; }
template <class T> constexpr inline T avg(const T& v) { return v; }
template <class T> constexpr inline T sum(const T& v) { return v; }
template <class T> constexpr inline T maxw(uint32_t, const T& v) { return v; }
template <class T> constexpr inline T minw(uint32_t, const T& v) { return v; }
template <class T> constexpr inline T avgw(uint32_t, const T& v) { return v; }
template <class T> constexpr inline T sumw(uint32_t, const T& v) { return v; }
template <class T> constexpr inline T ratiow(uint32_t, const T& v) { return 1; }
template <class T> constexpr inline T maxs(const T& v) { return v; }
template <class T> constexpr inline T mins(const T& v) { return v; }
template <class T> constexpr inline T avgs(const T& v) { return v; }
template <class T> constexpr inline T sums(const T& v) { return v; }
template <class T> constexpr inline T last(const T& v) { return v; }
template <class T> constexpr inline T prev(const T& v) { return v; }
template <class T> constexpr inline T aggnext(const T& v) { return v; }
template <class T> constexpr inline T daltas(const T& v) { return 0; }
template <class T> constexpr inline T ratios(const T& v) { return 1; }
template <class T> constexpr T count(const T&) { return 1; }
template <class T> constexpr T var(const T&) { return 0; }
template <class T> constexpr T vars(const T&) { return 0; }
template <class T> constexpr T varw(uint32_t, const T&) { return 0; }
template <class T> constexpr T stddev(const T&) { return 0; }
template <class T> constexpr T stddevs(const T&) { return 0; }
template <class T> constexpr T stddevw(uint32_t, const T&) { return 0; }
template <class T> constexpr T max(const T& v) { return v; }
template <class T> constexpr T min(const T& v) { return v; }
template <class T> constexpr T avg(const T& v) { return v; }
template <class T> constexpr T sum(const T& v) { return v; }
template <class T> constexpr T maxw(uint32_t, const T& v) { return v; }
template <class T> constexpr T minw(uint32_t, const T& v) { return v; }
template <class T> constexpr T avgw(uint32_t, const T& v) { return v; }
template <class T> constexpr T sumw(uint32_t, const T& v) { return v; }
template <class T> constexpr T ratiow(uint32_t, const T&) { return 1; }
template <class T> constexpr T maxs(const T& v) { return v; }
template <class T> constexpr T mins(const T& v) { return v; }
template <class T> constexpr T avgs(const T& v) { return v; }
template <class T> constexpr T sums(const T& v) { return v; }
template <class T> constexpr T last(const T& v) { return v; }
template <class T> constexpr T prev(const T& v) { return v; }
template <class T> constexpr T aggnext(const T& v) { return v; }
template <class T> constexpr T daltas(const T&) { return 0; }
template <class T> constexpr T ratios(const T&) { return 1; }
server/libaquery.h
+31 -2
View File
@@ -3,6 +3,7 @@
#include "table.h"
#include <unordered_map>
#include <chrono>
enum Log_level {
LOG_INFO,
@@ -15,9 +16,16 @@ enum Backend_Type {
BACKEND_MonetDB,
BACKEND_MariaDB
};
struct QueryStats{
long long monet_time;
long long postproc_time;
};
struct Config{
int running, new_query, server_mode,
backend_type, has_dll, exec_time, n_buffers;
int running, new_query, server_mode,
backend_type, has_dll,
n_buffers;
QueryStats stats;
int buffer_sizes[];
};
@@ -67,6 +75,27 @@ struct Context{
std::unordered_map<const char*, uColRef *> cols;
};
class aq_timer {
private:
std::chrono::high_resolution_clock::time_point now;
public:
aq_timer(){
now = std::chrono::high_resolution_clock::now();
}
void reset(){
now = std::chrono::high_resolution_clock::now();
}
long long elapsed(){
long long ret = (std::chrono::high_resolution_clock::now() - now).count();
reset();
return ret;
}
long long lap() const{
long long ret = (std::chrono::high_resolution_clock::now() - now).count();
return ret;
}
};
#ifdef _WIN32
#define __DLLEXPORT__ __declspec(dllexport) __stdcall
#else
server/server.cpp
+104 -11
View File
@@ -18,6 +18,7 @@
#include <sys/mman.h>
struct SharedMemory
{
std::atomic<bool> a;
int hFileMap;
void* pData;
SharedMemory(const char* fname) {
@@ -31,6 +32,79 @@ struct SharedMemory
}
};
#ifndef __USE_STD_SEMAPHORE__
#ifdef __APPLE__
#include <dispatch/dispatch.h>
class A_Semaphore {
private:
dispatch_semaphore_t native_handle;
public:
A_Semaphore(bool v = false) {
native_handle = dispatch_semaphore_create(v);
}
void acquire() {
puts("acquire");
dispatch_semaphore_wait(native_handle, DISPATCH_TIME_FOREVER);
}
void release() {
puts("release");
dispatch_semaphore_signal(native_handle);
}
~A_Semaphore() {
}
};
#else
#include <semaphore.h>
class A_Semaphore {
private:
sem_t native_handle;
public:
A_Semaphore(bool v = false) {
sem_init(&native_handle, v, 1);
}
void acquire() {
sem_wait(&native_handle);
}
void release() {
sem_post(&native_handle);
}
~A_Semaphore() {
sem_destroy(&native_handle);
}
};
#endif
#endif
#endif
#ifdef __USE_STD_SEMAPHORE__
#include <semaphore>
class A_Semaphore {
private:
std::binary_semaphore native_handle;
public:
A_Semaphore(bool v = false) {
native_handle = std::binary_semaphore(v);
}
void acquire() {
native_handle.acquire();
}
void release() {
native_handle.release();
}
~A_Semaphore() { }
};
#endif
#ifdef __AQUERY_ITC_USE_SHMEM__
A_Semaphore prompt{ true }, engine{ false };
#define PROMPT_ACQUIRE() prompt.acquire()
#define PROMPT_RELEASE() prompt.release()
#define ENGINE_ACQUIRE() engine.acquire()
#define ENGINE_RELEASE() engine.release()
#else
#define PROMPT_ACQUIRE()
#define PROMPT_RELEASE() std::this_thread::sleep_for(std::chrono::nanoseconds(0))
#define ENGINE_ACQUIRE()
#define ENGINE_RELEASE()
#endif
#include "aggregations.h"
@@ -42,6 +116,13 @@ int test_main();
int n_recv = 0;
char** n_recvd = nullptr;
__AQEXPORT__(void) wait_engine(){
PROMPT_ACQUIRE();
}
__AQEXPORT__(void) wake_engine(){
ENGINE_RELEASE();
}
extern "C" void __DLLEXPORT__ receive_args(int argc, char**argv){
n_recv = argc;
n_recvd = argv;
@@ -119,15 +200,16 @@ void initialize_module(const char* module_name, void* module_handle, Context* cx
}
int dll_main(int argc, char** argv, Context* cxt){
aq_timer timer;
Config *cfg = reinterpret_cast<Config *>(argv[0]);
std::unordered_map<std::string, void*> user_module_map;
if (cxt->module_function_maps == 0)
if (cxt->module_function_maps == nullptr)
cxt->module_function_maps = new std::unordered_map<std::string, void*>();
auto module_fn_map =
static_cast<std::unordered_map<std::string, void*>*>(cxt->module_function_maps);
auto buf_szs = cfg->buffer_sizes;
void** buffers = (void**)malloc(sizeof(void*) * cfg->n_buffers);
void** buffers = (void**) malloc (sizeof(void*) * cfg->n_buffers);
for (int i = 0; i < cfg->n_buffers; i++)
buffers[i] = static_cast<void *>(argv[i + 1]);
@@ -136,18 +218,22 @@ int dll_main(int argc, char** argv, Context* cxt){
cxt->n_buffers = cfg->n_buffers;
cxt->sz_bufs = buf_szs;
cxt->alt_server = NULL;
while(cfg->running){
ENGINE_ACQUIRE();
if (cfg->new_query) {
void *handle = 0;
void *user_module_handle = 0;
cfg->stats.postproc_time = 0;
cfg->stats.monet_time = 0;
void *handle = nullptr;
void *user_module_handle = nullptr;
if (cfg->backend_type == BACKEND_MonetDB){
if (cxt->alt_server == 0)
if (cxt->alt_server == nullptr)
cxt->alt_server = new Server(cxt);
Server* server = reinterpret_cast<Server*>(cxt->alt_server);
if(n_recv > 0){
if (cfg->backend_type == BACKEND_AQuery || cfg->has_dll) {
handle = dlopen("./dll.so", RTLD_LAZY);
handle = dlopen("./dll.so", RTLD_NOW);
}
for (const auto& module : user_module_map){
initialize_module(module.first.c_str(), module.second, cxt);
@@ -159,14 +245,18 @@ int dll_main(int argc, char** argv, Context* cxt){
switch(n_recvd[i][0]){
case 'Q': // SQL query for monetdbe
{
timer.reset();
server->exec(n_recvd[i] + 1);
cfg->stats.monet_time += timer.elapsed();
printf("Exec Q%d: %s", i, n_recvd[i]);
}
break;
case 'P': // Postprocessing procedure
if(handle && !server->haserror()) {
code_snippet c = reinterpret_cast<code_snippet>(dlsym(handle, n_recvd[i]+1));
timer.reset();
c(cxt);
cfg->stats.postproc_time += timer.elapsed();
}
break;
case 'M': // Load Module
@@ -198,7 +288,7 @@ int dll_main(int argc, char** argv, Context* cxt){
auto mname = n_recvd[i] + 1;
auto it = user_module_map.find(mname);
if (user_module_handle == it->second)
user_module_handle = 0;
user_module_handle = nullptr;
dlclose(it->second);
user_module_map.erase(it);
}
@@ -207,8 +297,9 @@ int dll_main(int argc, char** argv, Context* cxt){
}
if(handle) {
dlclose(handle);
handle = 0;
handle = nullptr;
}
printf("%ld, %ld", cfg->stats.monet_time, cfg->stats.postproc_time);
cxt->end_session();
n_recv = 0;
}
@@ -230,9 +321,11 @@ int dll_main(int argc, char** argv, Context* cxt){
if (handle) dlclose(handle);
cfg->new_query = 0;
}
std::this_thread::sleep_for(std::chrono::milliseconds(100));
//puts(cfg->running? "true": "false");
asm("");
PROMPT_RELEASE();
}
return 0;
}
server/table.h
+47 -20
View File
@@ -9,6 +9,7 @@
#include <string>
#include <algorithm>
#include <cstdarg>
#include <vector>
#include "io.h"
#include "hasher.h"
@@ -139,8 +140,16 @@ public:
ColView<_Ty> operator [](const vector_type<uint32_t>& idxs) const {
return ColView<_Ty>(*this, idxs);
}
void out(uint32_t n = 4, const char* sep = " ") const {
vector_type<_Ty> operator [](const std::vector<bool>& idxs) const {
vector_type<_Ty> ret (this->size);
uint32_t i = 0;
for(const auto& f : idxs){
if(f) ret.emplace_back(this->operator[](i));
++i;
}
return ret;
}
void out(uint32_t n = 1000, const char* sep = " ") const {
const char* more = "";
if (n < this->size)
more = " ... ";
@@ -243,7 +252,7 @@ public:
Iterator_t end() const {
return Iterator_t(idxs.end(), orig);
}
void out(uint32_t n = 4, const char* sep = " ") const {
void out(uint32_t n = 1000, const char* sep = " ") const {
n = n > size ? size : n;
std::cout << '(';
for (uint32_t i = 0; i < n; ++i)
@@ -438,19 +447,27 @@ struct TableInfo {
}
template <int ...cols>
void print2(const char* __restrict sep = ",", const char* __restrict end = "\n",
const vector_type<uint32_t>* __restrict view = nullptr, FILE* __restrict fp = nullptr) const {
const vector_type<uint32_t>* __restrict view = nullptr,
FILE* __restrict fp = nullptr, uint32_t limit = std::numeric_limits<uint32_t>::max()
) const {
std::string printf_string =
generate_printf_string<typename std::tuple_element<cols, tuple_type>::type ...>(sep, end);
// puts(printf_string.c_str());
std::string header_string = std::string();
constexpr static int a_cols[] = { cols... };
for (int i = 0; i < sizeof...(cols); ++i)
header_string += std::string(this->colrefs[a_cols[i]].name) + sep;
const size_t l_sep = strlen(sep);
if (header_string.size() - l_sep >= 0)
header_string.resize(header_string.size() - l_sep);
const auto& prt_loop = [&fp, &view, &printf_string, *this](const auto& f) {
if (fp == nullptr){
header_string = get_header_string(sep, end);
header_string.resize(header_string.size() - strlen(end));
}
else {
for (int i = 0; i < sizeof...(cols); ++i)
header_string += std::string(this->colrefs[a_cols[i]].name) + sep;
const size_t l_sep = strlen(sep);
if (header_string.size() - l_sep >= 0)
header_string.resize(header_string.size() - l_sep);
}
const auto& prt_loop = [&fp, &view, &printf_string, *this, &limit](const auto& f) {
#ifdef __AQ__HAS__INT128__
constexpr auto num_hge = count_type<__int128_t, __uint128_t>((tuple_type*)(0));
#else
@@ -466,16 +483,21 @@ struct TableInfo {
+ 1 // padding for msvc not allowing empty arrays
];
setgbuf(cbuf);
if (view)
for (uint32_t i = 0; i < view->size; ++i) {
if (view){
uint32_t outsz = limit > view->size ? view->size : limit;
for (uint32_t i = 0; i < outsz; ++i) {
print2_impl<cols...>(f, (*view)[i], printf_string.c_str());
setgbuf();
}
else
for (uint32_t i = 0; i < colrefs[0].size; ++i) {
}
else{
uint32_t outsz = limit > colrefs[0].size ? colrefs[0].size : limit;
for (uint32_t i = 0; i < outsz; ++i) {
print2_impl<cols...>(f, i, printf_string.c_str());
setgbuf();
}
}
};
if (fp)
@@ -490,15 +512,17 @@ struct TableInfo {
}
template <int ...vals> struct applier {
inline constexpr static void apply(const TableInfo<Types...>& t, const char* __restrict sep = ",", const char* __restrict end = "\n",
const vector_type<uint32_t>* __restrict view = nullptr, FILE* __restrict fp = nullptr)
const vector_type<uint32_t>* __restrict view = nullptr, FILE* __restrict fp = nullptr, uint32_t limit = std::numeric_limits<uint32_t>::max()
)
{
t.template print2<vals ...>(sep, end, view, fp);
t.template print2<vals ...>(sep, end, view, fp, limit);
}
};
inline void printall(const char* __restrict sep = ",", const char* __restrict end = "\n",
const vector_type<uint32_t>* __restrict view = nullptr, FILE* __restrict fp = nullptr) {
applyIntegerSequence<sizeof...(Types), applier>::apply(*this, sep, end, view, fp);
const vector_type<uint32_t>* __restrict view = nullptr, FILE* __restrict fp = nullptr,
uint32_t limit = std::numeric_limits<uint32_t>::max() ) const {
applyIntegerSequence<sizeof...(Types), applier>::apply(*this, sep, end, view, fp, limit);
}
TableInfo<Types...>* rename(const char* name) {
@@ -667,7 +691,9 @@ template <class ...Types>
template <size_t j>
inline typename std::enable_if<j == sizeof...(Types) - 1, void>::type
TableInfo<Types ...>::print_impl(const uint32_t& i, const char* __restrict sep) const {
std::cout << (get<j>(*this))[i];
decltype(auto) t = (get<j>(*this))[i];
// print(t);
std::cout << t;
}
template<class ...Types>
@@ -682,6 +708,7 @@ inline typename std::enable_if < j < sizeof...(Types) - 1, void>::type
template<class ...Types>
inline void TableInfo<Types...>::print(const char* __restrict sep, const char* __restrict end) const {
//printall(sep, end);
std::string header_string = get_header_string(sep, end);
std::cout << header_string.c_str();
server/types.h
+17 -7
View File
@@ -29,27 +29,37 @@ inline constexpr size_t aq_szof<void> = 0;
template <class T1, class T2>
struct aqis_same_impl {
constexpr static bool value =
std::conditional_t<
std::is_signed_v<T1> == std::is_signed_v<T2>,
std::is_same_v<T1, bool> || std::is_same_v<T2, bool>,
Cond(
std::is_floating_point_v<T1> == std::is_floating_point_v<T2>,
(std::is_same_v<T1, bool> && std::is_same_v<T2, bool>),
std::true_type,
std::false_type
),
Cond(
std::is_signed_v<T1> == std::is_signed_v<T2>,
Cond(
aq_szof<T1> == aq_szof<T2>, // deal with sizeof(void)
std::true_type,
std::is_floating_point_v<T1> == std::is_floating_point_v<T2>,
Cond(
aq_szof<T1> == aq_szof<T2>, // deal with sizeof(void)
std::true_type,
std::false_type
),
std::false_type
),
std::false_type
),
std::false_type
)
>::value;
};
// make sure size_t/ptr_t and the corresponding integer types are the same
template <class T1, class T2, class ...Ts>
constexpr bool aqis_same = aqis_same_impl<T1, T2>::value &&
aqis_same<T2, Ts...>;
template <class T1, class T2>
constexpr bool aqis_same<T1, T2> = aqis_same_impl<T1, T2>::value;
namespace types {
enum Type_t {
AINT32, AFLOAT, ASTR, ADOUBLE, ALDOUBLE, AINT64, AINT128, AINT16, ADATE, ATIME, AINT8,
server/vector_type.hpp
+1 -1
View File
@@ -265,7 +265,7 @@ public:
}
size = this->size + dist;
}
inline void out(uint32_t n = 4, const char* sep = " ") const
inline void out(uint32_t n = 4000, const char* sep = " ") const
{
const char* more = "";
if (n < this->size)
server/winhelper.cpp
+16
View File
@@ -41,4 +41,20 @@ void SharedMemory::FreeMemoryMap()
if (this->hFileMap)
CloseHandle(this->hFileMap);
}
#ifndef __USE_STD_SEMAPHORE__
A_Semaphore::A_Semaphore(bool v = false) {
native_handle = CreateSemaphore(NULL, v, 1, NULL);
}
void A_Semaphore::acquire() {
WaitForSingleObject(native_handle, INFINITE);
}
void A_Semaphore::release() {
ReleaseSemaphore(native_handle, 1, NULL);
}
A_Semaphore::~A_Semaphore() {
CloseHandle(native_handle);
}
#endif
#endif
server/winhelper.h
+12
View File
@@ -14,5 +14,17 @@ struct SharedMemory
SharedMemory(const char*);
void FreeMemoryMap();
};
#ifndef __USE_STD_SEMAPHORE__
class A_Semaphore {
private:
void* native_handle;
public:
A_Semaphore();
void acquire();
void release();
~A_Semaphore();
};
#endif
#endif