Line data Source code
1 : /*
2 : * SPDX-License-Identifier: MPL-2.0
3 : *
4 : * This Source Code Form is subject to the terms of the Mozilla Public
5 : * License, v. 2.0. If a copy of the MPL was not distributed with this
6 : * file, You can obtain one at http://mozilla.org/MPL/2.0/.
7 : *
8 : * Copyright 2024 MonetDB Foundation;
9 : * Copyright August 2008 - 2023 MonetDB B.V.;
10 : * Copyright 1997 - July 2008 CWI.
11 : */
12 :
13 : #include "monetdb_config.h"
14 :
15 : #include "monetdbe.h"
16 : #include "gdk.h"
17 : #include "mal.h"
18 : #include "mal_client.h"
19 : #include "mal_embedded.h"
20 : #include "mal_backend.h"
21 : #include "mal_builder.h"
22 : #include "mal_linker.h"
23 : #include "sql_mvc.h"
24 : #include "sql_catalog.h"
25 : #include "sql_gencode.h"
26 : #include "sql_semantic.h"
27 : #include "sql_scenario.h"
28 : #include "sql_optimizer.h"
29 : #include "rel_exp.h"
30 : #include "rel_rel.h"
31 : #include "rel_updates.h"
32 : #include "monet_options.h"
33 : #include "mapi.h"
34 : #include "monetdbe_mapi.h"
35 : #include "remote.h"
36 : #include "sql.h"
37 : #include "sql_result.h"
38 : #include "mutils.h"
39 :
40 : #define UNUSED(x) (void)(x)
41 :
42 : static int
43 0 : monetdbe_2_gdk_type(monetdbe_types t) {
44 0 : switch(t) {
45 : case monetdbe_bool: return TYPE_bit;
46 : case monetdbe_int8_t: return TYPE_bte;
47 : case monetdbe_int16_t: return TYPE_sht;
48 : case monetdbe_int32_t: return TYPE_int;
49 : case monetdbe_int64_t: return TYPE_lng;
50 : #ifdef HAVE_HGE
51 : case monetdbe_int128_t: return TYPE_hge;
52 : #endif
53 : case monetdbe_size_t: return TYPE_oid;
54 : case monetdbe_float: return TYPE_flt;
55 : case monetdbe_double: return TYPE_dbl;
56 : case monetdbe_str: return TYPE_str;
57 : case monetdbe_blob: return TYPE_blob;
58 : case monetdbe_date: return TYPE_date;
59 : case monetdbe_time: return TYPE_daytime;
60 : case monetdbe_timestamp: return TYPE_timestamp;
61 : default:
62 : return -1;
63 : }
64 : }
65 :
66 : static monetdbe_types
67 0 : embedded_type(int t) {
68 0 : switch(t) {
69 : case TYPE_bit: return monetdbe_bool;
70 0 : case TYPE_bte: return monetdbe_int8_t;
71 0 : case TYPE_sht: return monetdbe_int16_t;
72 0 : case TYPE_int: return monetdbe_int32_t;
73 0 : case TYPE_lng: return monetdbe_int64_t;
74 : #ifdef HAVE_HGE
75 0 : case TYPE_hge: return monetdbe_int128_t;
76 : #endif
77 0 : case TYPE_oid: return monetdbe_size_t;
78 0 : case TYPE_flt: return monetdbe_float;
79 0 : case TYPE_dbl: return monetdbe_double;
80 0 : case TYPE_str: return monetdbe_str;
81 0 : case TYPE_date: return monetdbe_date;
82 0 : case TYPE_daytime: return monetdbe_time;
83 0 : case TYPE_timestamp: return monetdbe_timestamp;
84 0 : default:
85 0 : if (t==TYPE_blob)
86 0 : return monetdbe_blob;
87 : return monetdbe_type_unknown;
88 : }
89 : }
90 :
91 : typedef struct {
92 : Client c;
93 : char *msg;
94 : int registered_thread; /* 1 = registered in monetdbe_open, 2 = done by GDK (also deregister done there) */
95 : monetdbe_data_blob blob_null;
96 : monetdbe_data_date date_null;
97 : monetdbe_data_time time_null;
98 : monetdbe_data_timestamp timestamp_null;
99 : str mid;
100 : } monetdbe_database_internal;
101 :
102 : typedef struct {
103 : monetdbe_result res;
104 : int type;
105 : res_table *monetdbe_resultset;
106 : monetdbe_column **converted_columns;
107 : monetdbe_database_internal *mdbe;
108 : } monetdbe_result_internal;
109 :
110 : typedef struct {
111 : monetdbe_statement res;
112 : ValRecord *data;
113 : ValPtr *args; /* only used during calls */
114 : int retc;
115 : monetdbe_database_internal *mdbe;
116 : cq *q;
117 : } monetdbe_stmt_internal;
118 :
119 : static MT_Lock embedded_lock = MT_LOCK_INITIALIZER(embedded_lock);
120 : static bool monetdbe_embedded_initialized = false;
121 : static char *monetdbe_embedded_url = NULL;
122 : static int open_dbs = 0;
123 :
124 : static void data_from_date(date d, monetdbe_data_date *ptr);
125 : static void data_from_time(daytime d, monetdbe_data_time *ptr);
126 : static void data_from_timestamp(timestamp d, monetdbe_data_timestamp *ptr);
127 : static timestamp timestamp_from_data(monetdbe_data_timestamp *ptr);
128 : static date date_from_data(monetdbe_data_date *ptr);
129 : static daytime time_from_data(monetdbe_data_time *ptr);
130 :
131 : static char* monetdbe_cleanup_result_internal(monetdbe_database_internal *mdbe, monetdbe_result_internal* res);
132 :
133 : static int
134 0 : date_is_null(monetdbe_data_date *value)
135 : {
136 0 : monetdbe_data_date null_value;
137 0 : data_from_date(date_nil, &null_value);
138 0 : return value->year == null_value.year && value->month == null_value.month &&
139 : value->day == null_value.day;
140 : }
141 :
142 : static int
143 0 : time_is_null(monetdbe_data_time *value)
144 : {
145 0 : monetdbe_data_time null_value;
146 0 : data_from_time(daytime_nil, &null_value);
147 0 : return value->hours == null_value.hours &&
148 0 : value->minutes == null_value.minutes &&
149 0 : value->seconds == null_value.seconds && value->ms == null_value.ms;
150 : }
151 :
152 : static int
153 0 : timestamp_is_null(monetdbe_data_timestamp *value)
154 : {
155 0 : return is_timestamp_nil(timestamp_from_data(value));
156 : }
157 :
158 : static int
159 2 : str_is_null(char **value)
160 : {
161 2 : return !value || *value == NULL;
162 : }
163 :
164 : static int
165 0 : blob_is_null(monetdbe_data_blob *value)
166 : {
167 0 : return !value || value->data == NULL;
168 : }
169 :
170 : const char *
171 0 : monetdbe_version(void)
172 : {
173 0 : return MONETDBE_VERSION;
174 : }
175 :
176 : static void
177 6 : clear_error( monetdbe_database_internal *mdbe)
178 : {
179 6 : if (mdbe->msg)
180 0 : freeException(mdbe->msg);
181 6 : mdbe->msg = NULL;
182 0 : }
183 :
184 : static char*
185 0 : set_error( monetdbe_database_internal *mdbe, char *err)
186 : {
187 0 : if (!err)
188 : return err;
189 0 : if (mdbe->msg) /* keep first error */
190 0 : freeException(err);
191 : else
192 0 : mdbe->msg = err;
193 0 : return mdbe->msg;
194 : }
195 :
196 : static char*
197 3 : commit_action(mvc* m, monetdbe_database_internal *mdbe, monetdbe_result **result, monetdbe_result_internal *res_internal)
198 : {
199 3 : char *commit_msg = MAL_SUCCEED;
200 :
201 : /* if an error already exists from MonetDBe set the session status to dirty */
202 3 : if (mdbe->msg != MAL_SUCCEED && m->session->tr->active && !m->session->status)
203 0 : m->session->status = -1;
204 3 : commit_msg = SQLautocommit(m); /* handle autocommit */
205 :
206 3 : if (mdbe->msg != MAL_SUCCEED || commit_msg != MAL_SUCCEED) {
207 0 : if (res_internal) {
208 0 : char* other = monetdbe_cleanup_result_internal(mdbe, res_internal);
209 0 : if (other)
210 0 : freeException(other);
211 : }
212 0 : if (result)
213 0 : *result = NULL;
214 0 : (void)set_error(mdbe, commit_msg);
215 : }
216 3 : return mdbe->msg;
217 : }
218 :
219 : static int
220 1 : validate_database_handle_noerror(monetdbe_database_internal *mdbe)
221 : {
222 1 : if (!monetdbe_embedded_initialized || !MCvalid(mdbe->c))
223 0 : return 0;
224 1 : assert(mdbe->c);
225 1 : MT_thread_set_qry_ctx(&mdbe->c->qryctx);
226 1 : clear_error(mdbe);
227 1 : return 1;
228 : }
229 :
230 : // Call this function always inside the embedded_lock
231 : static char*
232 5 : validate_database_handle(monetdbe_database_internal *mdbe, const char* call)
233 : {
234 5 : if (!monetdbe_embedded_initialized)
235 0 : return createException(MAL, call, "MonetDBe has not yet started");
236 5 : if (!MCvalid(mdbe->c))
237 0 : return createException(MAL, call, "Invalid database handle");
238 5 : clear_error(mdbe);
239 5 : return MAL_SUCCEED;
240 : }
241 :
242 : static void
243 2 : monetdbe_destroy_column(monetdbe_column* column)
244 : {
245 2 : size_t j;
246 :
247 2 : if (!column)
248 : return;
249 :
250 2 : if (column->type == monetdbe_str) {
251 : // FIXME: clean up individual strings
252 1 : char** data = (char**)column->data;
253 3 : for(j = 0; j < column->count; j++) {
254 2 : if (data[j])
255 2 : GDKfree(data[j]);
256 : }
257 1 : } else if (column->type == monetdbe_blob) {
258 0 : monetdbe_data_blob* data = (monetdbe_data_blob*)column->data;
259 0 : for(j = 0; j < column->count; j++) {
260 0 : if (data[j].data)
261 0 : GDKfree(data[j].data);
262 : }
263 : }
264 2 : GDKfree(column->sql_type.name);
265 2 : GDKfree(column->data);
266 2 : GDKfree(column);
267 : }
268 :
269 : static char*
270 1 : monetdbe_cleanup_result_internal(monetdbe_database_internal *mdbe, monetdbe_result_internal* result)
271 : {
272 1 : mvc *m = NULL;
273 :
274 1 : assert(!result || !result->mdbe || result->mdbe == mdbe);
275 1 : if ((mdbe->msg = validate_database_handle(mdbe, "monetdbe.monetdbe_cleanup_result_internal")) != MAL_SUCCEED)
276 : return mdbe->msg;
277 1 : if ((mdbe->msg = getSQLContext(mdbe->c, NULL, &m, NULL)) != MAL_SUCCEED)
278 0 : goto cleanup;
279 :
280 1 : if (result->monetdbe_resultset)
281 1 : res_tables_destroy(result->monetdbe_resultset);
282 :
283 1 : if (result->converted_columns) {
284 3 : for (size_t i = 0; i < result->res.ncols; i++)
285 2 : monetdbe_destroy_column(result->converted_columns[i]);
286 1 : GDKfree(result->converted_columns);
287 : }
288 1 : GDKfree(result);
289 1 : cleanup:
290 1 : return commit_action(m, mdbe, NULL, NULL);
291 : }
292 :
293 : static char*
294 1 : monetdbe_get_results(monetdbe_result** result, monetdbe_database_internal *mdbe)
295 : {
296 1 : backend *be = NULL;
297 :
298 1 : *result = NULL;
299 1 : if ((mdbe->msg = getBackendContext(mdbe->c, &be)) != NULL)
300 : return mdbe->msg;
301 :
302 1 : mvc *m = be->mvc;
303 1 : monetdbe_result_internal* res_internal;
304 :
305 1 : if (!(res_internal = GDKzalloc(sizeof(monetdbe_result_internal)))) {
306 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_get_results", MAL_MALLOC_FAIL));
307 0 : return mdbe->msg;
308 : }
309 : // TODO: set type of result outside.
310 1 : res_internal->res.last_id = be->last_id;
311 1 : res_internal->mdbe = mdbe;
312 1 : *result = (monetdbe_result*) res_internal;
313 1 : m->reply_size = -2; /* do not clean up result tables */
314 :
315 1 : if (be->results) {
316 1 : res_internal->res.ncols = (size_t) be->results->nr_cols;
317 1 : res_internal->monetdbe_resultset = be->results;
318 1 : if (be->results->nr_cols > 0)
319 1 : res_internal->res.nrows = be->results->nr_rows;
320 1 : be->results = NULL;
321 1 : res_internal->converted_columns = GDKzalloc(sizeof(monetdbe_column*) * res_internal->res.ncols);
322 1 : if (!res_internal->converted_columns) {
323 0 : GDKfree(res_internal);
324 0 : *result = NULL;
325 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_get_results", MAL_MALLOC_FAIL));
326 0 : return mdbe->msg;
327 : }
328 : }
329 :
330 : return MAL_SUCCEED;
331 : }
332 :
333 : static char*
334 0 : monetdbe_query_internal(monetdbe_database_internal *mdbe, char* query, monetdbe_result** result, monetdbe_cnt* affected_rows, int *prepare_id, char language)
335 : {
336 0 : char *nq = NULL;
337 0 : Client c = mdbe->c;
338 0 : mvc* m = NULL;
339 0 : backend *b;
340 0 : size_t query_len, input_query_len, prep_len = 0;
341 0 : buffer query_buf;
342 0 : stream *query_stream = NULL;
343 0 : bstream *old_bstream = c->fdin;
344 0 : stream *fdout = c->fdout;
345 0 : bool fdin_changed = false;
346 :
347 0 : if (result)
348 0 : *result = NULL;
349 :
350 0 : if ((mdbe->msg = validate_database_handle(mdbe, "monetdbe.monetdbe_query_internal")) != MAL_SUCCEED)
351 : return mdbe->msg;
352 :
353 0 : if ((mdbe->msg = getSQLContext(c, NULL, &m, NULL)) != MAL_SUCCEED)
354 0 : goto cleanup;
355 0 : b = (backend *) c->sqlcontext;
356 :
357 0 : if (!query) {
358 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_query_internal", "Query missing"));
359 0 : goto cleanup;
360 : }
361 0 : if (!(query_stream = buffer_rastream(&query_buf, "sqlstatement"))) {
362 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_query_internal", "Could not setup query stream"));
363 0 : goto cleanup;
364 : }
365 0 : input_query_len = strlen(query);
366 0 : query_len = input_query_len + 3;
367 0 : if (prepare_id) {
368 0 : prep_len = sizeof("PREPARE ")-1;
369 0 : query_len += prep_len;
370 : }
371 0 : if (!(nq = GDKmalloc(query_len))) {
372 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_query_internal", MAL_MALLOC_FAIL));
373 0 : goto cleanup;
374 : }
375 0 : if (prepare_id)
376 0 : strcpy(nq, "PREPARE ");
377 0 : strcpy(nq + prep_len, query);
378 0 : strcpy(nq + prep_len + input_query_len, "\n;");
379 :
380 0 : query_buf.pos = 0;
381 0 : query_buf.len = query_len;
382 0 : query_buf.buf = nq;
383 :
384 0 : fdin_changed = true;
385 0 : if (!(c->fdin = bstream_create(query_stream, query_len))) {
386 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_query_internal", "Could not setup query stream"));
387 0 : goto cleanup;
388 : }
389 0 : c->qryctx.bs = c->fdin;
390 0 : query_stream = NULL;
391 0 : if (bstream_next(c->fdin) < 0) {
392 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_query_internal", "Internal error while starting the query"));
393 0 : goto cleanup;
394 : }
395 :
396 0 : assert(language);
397 0 : b->language = language;
398 0 : b->output_format = OFMT_NONE;
399 0 : b->no_mitosis = 0;
400 0 : m->user_id = m->role_id = USER_MONETDB;
401 0 : m->errstr[0] = '\0';
402 0 : m->params = NULL;
403 0 : m->sym = NULL;
404 0 : m->runs = NULL;
405 0 : m->label = 0;
406 0 : if (m->sa)
407 0 : m->sa = sa_reset(m->sa);
408 0 : m->scanner.mode = LINE_N;
409 0 : m->scanner.rs = c->fdin;
410 0 : scanner_query_processed(&(m->scanner));
411 :
412 0 : if ((mdbe->msg = MSinitClientPrg(c, userRef, mainRef)) != MAL_SUCCEED)
413 0 : goto cleanup;
414 0 : if (prepare_id)
415 0 : m->emode = m_prepare;
416 0 : c->fdout = NULL;
417 0 : if ((mdbe->msg = SQLengine_(c)) != MAL_SUCCEED)
418 0 : goto cleanup;
419 0 : if (m->emode == m_prepare && prepare_id)
420 0 : *prepare_id = b->result_id;
421 0 : if (!b->results && b->rowcnt >= 0 && affected_rows)
422 0 : *affected_rows = b->rowcnt;
423 :
424 0 : if (result) {
425 0 : if ((mdbe->msg = monetdbe_get_results(result, mdbe)) != MAL_SUCCEED) {
426 0 : goto cleanup;
427 : }
428 :
429 0 : if (m->emode & m_prepare)
430 0 : (*(monetdbe_result_internal**) result)->type = Q_PREPARE;
431 : else
432 0 : (*(monetdbe_result_internal**) result)->type = (b->results) ? b->results->query_type : m->type;
433 : }
434 :
435 0 : cleanup:
436 0 : c->fdout = fdout;
437 0 : if (nq)
438 0 : GDKfree(nq);
439 0 : MSresetInstructions(c->curprg->def, 1);
440 0 : if (fdin_changed) { //c->fdin was set
441 0 : bstream_destroy(c->fdin);
442 0 : c->fdin = old_bstream;
443 0 : c->qryctx.bs = old_bstream;
444 : }
445 0 : if (query_stream)
446 0 : close_stream(query_stream);
447 :
448 0 : return commit_action(m, mdbe, result, result?*(monetdbe_result_internal**) result:NULL);
449 : }
450 :
451 : static int
452 1 : monetdbe_close_remote(monetdbe_database_internal *mdbe)
453 : {
454 1 : assert(mdbe && mdbe->mid);
455 :
456 1 : int err = 0;
457 :
458 1 : if (mdbe->msg) {
459 0 : err = 1;
460 0 : clear_error(mdbe);
461 : }
462 :
463 1 : if ( (mdbe->msg = RMTdisconnect(NULL, &(const char *){mdbe->mid})) != MAL_SUCCEED) {
464 0 : err = 1;
465 0 : clear_error(mdbe);
466 : }
467 :
468 1 : GDKfree(mdbe->mid);
469 1 : mdbe->mid = NULL;
470 :
471 1 : return err;
472 : }
473 :
474 : static int
475 1 : monetdbe_close_internal(monetdbe_database_internal *mdbe)
476 : {
477 1 : assert(mdbe);
478 :
479 1 : if (validate_database_handle_noerror(mdbe)) {
480 1 : open_dbs--;
481 1 : char *msg = SQLexitClient(mdbe->c);
482 1 : if (msg)
483 0 : freeException(msg);
484 1 : MCcloseClient(mdbe->c);
485 : }
486 1 : GDKfree(mdbe);
487 1 : return 0;
488 : }
489 :
490 : static int
491 2 : monetdbe_workers_internal(monetdbe_database_internal *mdbe, monetdbe_options *opts)
492 : {
493 2 : int workers = 0;
494 2 : if (opts && opts->nr_threads) {
495 0 : if (opts->nr_threads < 0)
496 0 : set_error(mdbe,createException(MAL, "monetdbe.monetdbe_startup", "Nr_threads should be positive"));
497 : else
498 : workers = opts->nr_threads;
499 : }
500 2 : return workers;
501 : }
502 :
503 : static int
504 2 : monetdbe_memory_internal(monetdbe_database_internal *mdbe, monetdbe_options *opts)
505 : {
506 2 : int memory = 0;
507 2 : if (opts && opts->memorylimit) {
508 0 : if (opts->memorylimit < 0)
509 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_startup", "Memorylimit should be positive"));
510 : else /* Memory limit is session specific */
511 : memory = opts->memorylimit;
512 : }
513 2 : return memory;
514 : }
515 :
516 : static int
517 1 : monetdbe_querytimeout_internal(monetdbe_database_internal *mdbe, monetdbe_options *opts)
518 : {
519 1 : int querytimeout = 0;
520 1 : if (opts && opts->querytimeout) {
521 0 : if (opts->querytimeout < 0)
522 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_startup", "Query timeout should be positive (in sec)"));
523 : else
524 : querytimeout = opts->querytimeout;
525 : }
526 1 : return querytimeout;
527 : }
528 :
529 : static int
530 1 : monetdbe_sessiontimeout_internal(monetdbe_database_internal *mdbe, monetdbe_options *opts)
531 : {
532 1 : int sessiontimeout = 0;
533 1 : if (opts && opts->sessiontimeout) {
534 0 : if (opts->sessiontimeout < 0)
535 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_startup", "Session timeout should be positive (in sec)"));
536 : else
537 : sessiontimeout = opts->sessiontimeout;
538 : }
539 1 : return sessiontimeout;
540 : }
541 :
542 : static int
543 1 : monetdbe_open_internal(monetdbe_database_internal *mdbe, monetdbe_options *opts )
544 : {
545 1 : mvc *m;
546 :
547 1 : if (!mdbe)
548 : return -1;
549 1 : if (!monetdbe_embedded_initialized) {
550 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_open_internal", "Embedded MonetDB is not started"));
551 0 : goto cleanup;
552 : }
553 1 : if (!mdbe->registered_thread) {
554 0 : if (!MT_thread_register()) {
555 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_open_internal", "Embedded MonetDB is not started"));
556 0 : goto cleanup;
557 : }
558 0 : mdbe->registered_thread = 1;
559 : }
560 1 : mdbe->c = MCinitClient((oid) 0, 0, 0);
561 1 : if (!MCvalid(mdbe->c)) {
562 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_open_internal", "Failed to initialize client"));
563 0 : goto cleanup;
564 : }
565 1 : mdbe->c->curmodule = mdbe->c->usermodule = userModule();
566 1 : mdbe->c->workerlimit = monetdbe_workers_internal(mdbe, opts);
567 1 : mdbe->c->memorylimit = monetdbe_memory_internal(mdbe, opts);
568 1 : mdbe->c->querytimeout = monetdbe_querytimeout_internal(mdbe, opts);
569 1 : mdbe->c->sessiontimeout = monetdbe_sessiontimeout_internal(mdbe, opts);
570 1 : if (mdbe->msg)
571 0 : goto cleanup;
572 1 : if (mdbe->c->usermodule == NULL) {
573 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_open_internal", "Failed to initialize client MAL module"));
574 0 : goto cleanup;
575 : }
576 1 : if ((mdbe->msg = SQLinitClient(mdbe->c, NULL, NULL, NULL)) != MAL_SUCCEED ||
577 1 : (mdbe->msg = getSQLContext(mdbe->c, NULL, &m, NULL)) != MAL_SUCCEED)
578 0 : goto cleanup;
579 1 : m->session->auto_commit = 1;
580 1 : if (!m->pa)
581 0 : m->pa = sa_create(NULL);
582 1 : if (!m->sa)
583 1 : m->sa = sa_create(m->pa);
584 1 : if (!m->ta)
585 0 : m->ta = sa_create(m->pa);
586 1 : if (!m->pa || !m->sa || !m->ta) {
587 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_open_internal", MAL_MALLOC_FAIL));
588 0 : goto cleanup;
589 : }
590 1 : m->no_int128 = opts?opts->no_int128:false;
591 1 : cleanup:
592 1 : if (mdbe->msg)
593 : return -2;
594 1 : mdbe->blob_null.data = NULL;
595 1 : data_from_date(date_nil, &mdbe->date_null);
596 1 : data_from_time(daytime_nil, &mdbe->time_null);
597 1 : data_from_timestamp(timestamp_nil, &mdbe->timestamp_null);
598 1 : open_dbs++;
599 1 : return 0;
600 : }
601 :
602 : static void
603 1 : monetdbe_shutdown_internal(void) // Call this function always inside the embedded_lock
604 : {
605 1 : if (monetdbe_embedded_initialized && (open_dbs == 0)) {
606 1 : malEmbeddedReset();
607 1 : monetdbe_embedded_initialized = false;
608 1 : if (monetdbe_embedded_url)
609 0 : GDKfree(monetdbe_embedded_url);
610 1 : monetdbe_embedded_url = NULL;
611 : }
612 1 : }
613 :
614 : static void
615 1 : monetdbe_startup(monetdbe_database_internal *mdbe, const char* dbdir, monetdbe_options *opts)
616 : {
617 : // Only call monetdbe_startup when there is no monetdb internal yet initialized.
618 1 : assert(!monetdbe_embedded_initialized);
619 :
620 1 : opt *set = NULL;
621 1 : int setlen;
622 1 : bool with_mapi_server;
623 1 : int workers, memory;
624 1 : gdk_return gdk_res;
625 :
626 1 : GDKfataljumpenable = 1;
627 :
628 1 : if(setjmp(GDKfataljump) != 0) {
629 0 : assert(0);
630 : mdbe->msg = GDKfatalmsg;
631 : // we will get here if GDKfatal was called.
632 : if (mdbe->msg == NULL)
633 : mdbe->msg = createException(MAL, "monetdbe.monetdbe_startup", "GDKfatal() with unspecified error");
634 : goto cleanup;
635 : }
636 :
637 1 : with_mapi_server = false;
638 :
639 1 : if (monetdbe_embedded_initialized) {
640 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_startup", "MonetDBe is already initialized"));
641 0 : GDKfataljumpenable = 0;
642 0 : return;
643 : }
644 :
645 1 : if ((setlen = mo_builtin_settings(&set)) == 0) {
646 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_startup", MAL_MALLOC_FAIL));
647 0 : goto cleanup;
648 : }
649 1 : if (dbdir && (setlen = mo_add_option(&set, setlen, opt_cmdline, "gdk_dbpath", dbdir)) == 0) {
650 0 : mo_free_options(set, setlen);
651 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_startup", MAL_MALLOC_FAIL));
652 0 : goto cleanup;
653 : }
654 1 : if (opts && opts->nr_threads == 1)
655 0 : setlen = mo_add_option(&set, setlen, opt_cmdline, "sql_optimizer", "sequential_pipe");
656 : else
657 1 : setlen = mo_add_option(&set, setlen, opt_cmdline, "sql_optimizer", "default_pipe");
658 :
659 1 : if (setlen == 0) {
660 0 : mo_free_options(set, setlen);
661 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_startup", MAL_MALLOC_FAIL));
662 0 : goto cleanup;
663 : }
664 :
665 1 : if (opts && opts->mapi_server) {
666 : /*This monetdbe instance wants to listen to external mapi client connections.*/
667 0 : if (opts->mapi_server->host) {
668 0 : with_mapi_server = true;
669 0 : int psetlen = setlen;
670 0 : setlen = mo_add_option(&set, setlen, opt_cmdline, "mapi_listenaddr", opts->mapi_server->host);
671 0 : if (setlen == psetlen) {
672 0 : mo_free_options(set, setlen);
673 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_startup", MAL_MALLOC_FAIL));
674 0 : goto cleanup;
675 : }
676 : }
677 0 : if (opts->mapi_server->port) {
678 0 : with_mapi_server = true;
679 0 : int psetlen = setlen;
680 0 : setlen = mo_add_option(&set, setlen, opt_cmdline, "mapi_port", opts->mapi_server->port);
681 0 : if (setlen == psetlen) {
682 0 : mo_free_options(set, setlen);
683 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_startup", MAL_MALLOC_FAIL));
684 0 : goto cleanup;
685 : }
686 : }
687 0 : if (opts->mapi_server->usock) {
688 0 : with_mapi_server = true;
689 0 : int psetlen = setlen;
690 0 : setlen = mo_add_option(&set, setlen, opt_cmdline, "mapi_usock", opts->mapi_server->usock);
691 0 : if (setlen == psetlen) {
692 0 : mo_free_options(set, setlen);
693 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_startup", MAL_MALLOC_FAIL));
694 0 : goto cleanup;
695 : }
696 : }
697 : }
698 :
699 : /* set the output of GDKtracer logs */
700 1 : if (opts && opts->trace_file) {
701 : /* if file specified, use it */
702 0 : if (GDKtracer_set_tracefile(opts->trace_file) != GDK_SUCCEED) {
703 0 : mo_free_options(set, setlen);
704 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_startup", GDK_EXCEPTION));
705 0 : goto cleanup;
706 : }
707 0 : GDKtracer_set_adapter("BASIC");
708 : } else {
709 : /* otherwise no trace output */
710 1 : GDKtracer_set_adapter("MBEDDED");
711 : }
712 :
713 1 : if ((workers = monetdbe_workers_internal(mdbe, opts))) {
714 0 : int psetlen = setlen;
715 0 : char workstr[16];
716 :
717 0 : snprintf(workstr, sizeof(workstr), "%d", workers);
718 0 : if ((setlen = mo_add_option(&set, setlen, opt_cmdline, "gdk_nr_threads", workstr)) == psetlen) {
719 0 : mo_free_options(set, setlen);
720 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_startup", MAL_MALLOC_FAIL));
721 0 : goto cleanup;
722 : }
723 : }
724 1 : if ((memory = monetdbe_memory_internal(mdbe, opts))) {
725 0 : int psetlen = setlen;
726 0 : char memstr[32];
727 :
728 0 : snprintf(memstr, sizeof(memstr), "%zu", (size_t) memory << 20); /* convert from MiB to bytes */
729 0 : if ((setlen = mo_add_option(&set, setlen, opt_cmdline, "gdk_vm_maxsize", memstr)) == psetlen) {
730 0 : mo_free_options(set, setlen);
731 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_startup", MAL_MALLOC_FAIL));
732 0 : goto cleanup;
733 : }
734 : }
735 1 : if (mdbe->msg) {
736 0 : mo_free_options(set, setlen);
737 0 : goto cleanup;
738 : }
739 :
740 1 : if (!dbdir) { /* in-memory */
741 1 : if (BBPaddfarm(NULL, (1U << PERSISTENT) | (1U << TRANSIENT), false) != GDK_SUCCEED) {
742 0 : mo_free_options(set, setlen);
743 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_startup", "Cannot add in-memory farm"));
744 0 : goto cleanup;
745 : }
746 : } else {
747 0 : if (BBPaddfarm(dbdir, 1U << PERSISTENT, false) != GDK_SUCCEED ||
748 0 : BBPaddfarm(/*dbextra ? dbextra : */dbdir, 1U << TRANSIENT, false) != GDK_SUCCEED) {
749 0 : mo_free_options(set, setlen);
750 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_startup", "Cannot add farm %s", dbdir));
751 0 : goto cleanup;
752 : }
753 0 : if (GDKcreatedir(dbdir) != GDK_SUCCEED) {
754 0 : mo_free_options(set, setlen);
755 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_startup", "Cannot create directory %s", dbdir));
756 0 : goto cleanup;
757 : }
758 : }
759 1 : gdk_res = GDKinit(set, setlen, true, mercurial_revision());
760 1 : mo_free_options(set, setlen);
761 1 : if (gdk_res != GDK_SUCCEED) {
762 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_startup", "GDKinit() failed"));
763 0 : goto cleanup;
764 : }
765 :
766 1 : if ((mdbe->msg = malEmbeddedBoot(workers, memory, 0, 0, with_mapi_server)) != MAL_SUCCEED)
767 0 : goto cleanup;
768 :
769 1 : monetdbe_embedded_initialized = true;
770 1 : monetdbe_embedded_url = dbdir?GDKstrdup(dbdir):NULL;
771 1 : if (dbdir && !monetdbe_embedded_url)
772 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_startup", MAL_MALLOC_FAIL));
773 1 : cleanup:
774 1 : GDKfataljumpenable = 0;
775 1 : if (mdbe->msg)
776 0 : monetdbe_shutdown_internal();
777 : }
778 :
779 0 : static bool urls_matches(const char* l, const char* r) {
780 0 : return (l && r && (strcmp(l, r) == 0)) || (l == NULL && r == NULL);
781 : }
782 :
783 : static inline str
784 1 : monetdbe_create_uri(const char* host, const int port, const char* database) {
785 1 : const char* protocol = "mapi:monetdb://";
786 :
787 1 : const size_t sl_protocol = strlen(protocol);
788 1 : const size_t sl_host = strlen(host);
789 1 : const size_t sl_max_port = 6; // 2^16-1 < 100 000 = 10^5, i.e. always less then 6 digits.
790 1 : const size_t sl_database = strlen(database);
791 1 : const size_t sl_total = sl_protocol + sl_host + 1 /* : */ + sl_max_port + 1 + /* / */ + sl_database;
792 :
793 1 : str uri_buffer = GDKmalloc(sl_total + 1 /* terminator */);
794 1 : if (!uri_buffer)
795 : return NULL;
796 :
797 1 : snprintf(uri_buffer, sl_total, "%s%s:%d/%s", protocol, host, port, database);
798 :
799 1 : return uri_buffer;
800 : }
801 :
802 : static int
803 1 : monetdbe_open_remote(monetdbe_database_internal *mdbe, monetdbe_options *opts) {
804 1 : assert(opts);
805 :
806 1 : monetdbe_remote* remote = opts->remote;
807 1 : if (!remote) {
808 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_open_remote", "Missing remote proxy settings"));
809 0 : return -1;
810 : }
811 :
812 1 : Client c = mdbe->c;
813 :
814 1 : assert(!c->curprg);
815 :
816 1 : const char mod[] = "user";
817 1 : char nme[16];
818 1 : const char *name = number2name(nme, sizeof(nme), ++((backend*) c->sqlcontext)->remote);
819 1 : c->curprg = newFunction(putName(mod), putName(name), FUNCTIONsymbol);
820 :
821 1 : if (c->curprg == NULL) {
822 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_open_remote", MAL_MALLOC_FAIL));
823 0 : return -2;
824 : }
825 :
826 1 : char* url;
827 1 : if ((url = monetdbe_create_uri(remote->host, remote->port, remote->database)) == NULL) {
828 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_open_remote", MAL_MALLOC_FAIL));
829 0 : return -2;
830 : }
831 :
832 1 : MalBlkPtr mb = c->curprg->def;
833 :
834 1 : InstrPtr q = getInstrPtr(mb, 0);
835 1 : q->argc = q->retc = 0;
836 1 : q = pushReturn(mb, q, newTmpVariable(mb, TYPE_str));
837 :
838 1 : InstrPtr p = newFcnCall(mb, remoteRef, connectRef);
839 1 : p = pushStr(mb, p, url);
840 1 : p = pushStr(mb, p, remote->username);
841 1 : p = pushStr(mb, p, remote->password);
842 1 : p = pushStr(mb, p, "msql");
843 1 : p = pushBit(mb, p, 1);
844 :
845 1 : GDKfree(url);
846 1 : url = NULL;
847 :
848 1 : if (p == NULL) {
849 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_open_remote", MAL_MALLOC_FAIL));
850 0 : freeSymbol(c->curprg);
851 0 : c->curprg= NULL;
852 0 : return -2;
853 : }
854 1 : pushInstruction(mb, p);
855 :
856 1 : q = newInstruction(mb, NULL, NULL);
857 1 : if (q == NULL) {
858 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_open_remote", MAL_MALLOC_FAIL));
859 0 : freeSymbol(c->curprg);
860 0 : c->curprg= NULL;
861 0 : return -2;
862 : }
863 1 : q->barrier= RETURNsymbol;
864 1 : q = pushReturn(mb, q, getArg(p, 0));
865 :
866 1 : pushInstruction(mb, q);
867 :
868 1 : if ( (mdbe->msg = chkProgram(c->usermodule, mb)) != MAL_SUCCEED ) {
869 0 : freeSymbol(c->curprg);
870 0 : c->curprg= NULL;
871 0 : return -2;
872 : }
873 1 : MalStkPtr stk = prepareMALstack(mb, mb->vsize);
874 1 : if (!stk) {
875 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_open_remote", MAL_MALLOC_FAIL));
876 0 : freeSymbol(c->curprg);
877 0 : c->curprg= NULL;
878 0 : return -2;
879 : }
880 1 : stk->keepAlive = TRUE;
881 1 : c->qryctx.starttime = GDKusec();
882 1 : c->qryctx.endtime = c->querytimeout ? c->qryctx.starttime + c->querytimeout : 0;
883 1 : if ( (mdbe->msg = runMALsequence(c, mb, 1, 0, stk, 0, 0)) != MAL_SUCCEED ) {
884 0 : freeStack(stk);
885 0 : freeSymbol(c->curprg);
886 0 : c->curprg= NULL;
887 0 : return -2;
888 : }
889 :
890 1 : if ((mdbe->mid = GDKstrdup(*getArgReference_str(stk, p, 0))) == NULL) {
891 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_open_remote", MAL_MALLOC_FAIL));
892 0 : freeStack(stk);
893 0 : freeSymbol(c->curprg);
894 0 : c->curprg= NULL;
895 0 : return -2;
896 : }
897 :
898 1 : garbageCollector(c, mb, stk, TRUE);
899 1 : freeStack(stk);
900 :
901 1 : freeSymbol(c->curprg);
902 1 : c->curprg= NULL;
903 :
904 1 : return 0;
905 : }
906 :
907 : int
908 1 : monetdbe_open(monetdbe_database *dbhdl, char *url, monetdbe_options *opts)
909 : {
910 1 : int res = 0;
911 :
912 1 : if (!dbhdl)
913 : return -1;
914 1 : if (url &&
915 0 : (strcmp(url, "in-memory") == 0 ||
916 0 : /* backward compatibility: */ strcmp(url, ":memory:") == 0))
917 1 : url = NULL;
918 1 : MT_lock_set(&embedded_lock);
919 1 : monetdbe_database_internal *mdbe = (monetdbe_database_internal*)GDKzalloc(sizeof(monetdbe_database_internal));
920 1 : if (!mdbe) {
921 0 : MT_lock_unset(&embedded_lock);
922 0 : return -1;
923 : }
924 1 : *dbhdl = (monetdbe_database)mdbe;
925 1 : mdbe->msg = NULL;
926 1 : mdbe->c = NULL;
927 :
928 1 : bool is_remote = (opts && (opts->remote != NULL));
929 1 : if (!monetdbe_embedded_initialized) {
930 : /* When used as a remote mapi proxy,
931 : * it is still necessary to have an initialized monetdbe. E.g. for BAT life cycle management.
932 : * Use an ephemeral/anonymous dbfarm when there is no initialized monetdbe yet.
933 : */
934 1 : assert(!is_remote||url==NULL);
935 1 : monetdbe_startup(mdbe, url, opts);
936 1 : mdbe->registered_thread = 2;
937 0 : } else if (!is_remote && !urls_matches(monetdbe_embedded_url, url)) {
938 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_open", "monetdbe_open currently only one active database is supported"));
939 : }
940 1 : if (!mdbe->msg)
941 1 : res = monetdbe_open_internal(mdbe, opts);
942 :
943 1 : if (res == 0 && is_remote && !mdbe->msg)
944 1 : res = monetdbe_open_remote(mdbe, opts);
945 :
946 1 : MT_lock_unset(&embedded_lock);
947 1 : if (mdbe->msg)
948 0 : return -2;
949 : return res;
950 : }
951 :
952 : int
953 1 : monetdbe_close(monetdbe_database dbhdl)
954 : {
955 1 : if (!dbhdl)
956 : return 0;
957 :
958 1 : monetdbe_database_internal *mdbe = (monetdbe_database_internal*)dbhdl;
959 :
960 1 : int err = 0;
961 1 : int registered_thread = mdbe->registered_thread;
962 :
963 1 : if (mdbe->c)
964 1 : MT_thread_set_qry_ctx(&mdbe->c->qryctx);
965 1 : MT_lock_set(&embedded_lock);
966 1 : if (mdbe->mid)
967 1 : err = monetdbe_close_remote(mdbe);
968 :
969 1 : err = (monetdbe_close_internal(mdbe) || err);
970 :
971 1 : if (registered_thread == 1) {
972 0 : MT_thread_deregister();
973 : }
974 1 : if (!open_dbs)
975 1 : monetdbe_shutdown_internal();
976 1 : MT_lock_unset(&embedded_lock);
977 :
978 1 : if (err) {
979 : return -2;
980 : }
981 :
982 : return 0;
983 : }
984 :
985 : char *
986 0 : monetdbe_error(monetdbe_database dbhdl)
987 : {
988 0 : if (!dbhdl)
989 : return NULL;
990 :
991 0 : monetdbe_database_internal *mdbe = (monetdbe_database_internal*)dbhdl;
992 0 : return mdbe->msg;
993 : }
994 :
995 : char*
996 0 : monetdbe_load_extension(monetdbe_database dbhdl, const char *file)
997 : {
998 0 : if (!dbhdl)
999 : return 0;
1000 :
1001 0 : monetdbe_database_internal *mdbe = (monetdbe_database_internal*)dbhdl;
1002 :
1003 0 : if ((mdbe->msg = validate_database_handle(mdbe, "embedded.monetdbe_dump_database")) != MAL_SUCCEED) {
1004 : return mdbe->msg;
1005 : }
1006 0 : char *modules[2];
1007 0 : modules[0] = (char*)file;
1008 0 : modules[1] = NULL;
1009 0 : char *msg = loadLibrary(file, -1);
1010 0 : if (msg)
1011 : return msg;
1012 0 : return malIncludeModules(mdbe->c, modules, 0, true, NULL);
1013 : }
1014 :
1015 : char*
1016 0 : monetdbe_dump_database(monetdbe_database dbhdl, const char *filename)
1017 : {
1018 0 : if (!dbhdl)
1019 : return NULL;
1020 :
1021 0 : monetdbe_database_internal *mdbe = (monetdbe_database_internal*)dbhdl;
1022 :
1023 0 : if (mdbe->mid) {
1024 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_dump_database", PROGRAM_NYI));
1025 0 : return mdbe->msg;
1026 : }
1027 :
1028 0 : if ((mdbe->msg = validate_database_handle(mdbe, "embedded.monetdbe_dump_database")) != MAL_SUCCEED) {
1029 : return mdbe->msg;
1030 : }
1031 :
1032 0 : mdbe->msg = monetdbe_mapi_dump_database(dbhdl, filename);
1033 :
1034 0 : return mdbe->msg;
1035 : }
1036 :
1037 : char*
1038 0 : monetdbe_dump_table(monetdbe_database dbhdl, const char *sname, const char *tname, const char *filename)
1039 : {
1040 0 : if (!dbhdl)
1041 : return NULL;
1042 :
1043 0 : monetdbe_database_internal *mdbe = (monetdbe_database_internal*)dbhdl;
1044 :
1045 0 : if (mdbe->mid) {
1046 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_dump_database", PROGRAM_NYI));
1047 0 : return mdbe->msg;
1048 : }
1049 :
1050 0 : if ((mdbe->msg = validate_database_handle(mdbe, "embedded.monetdbe_dump_table")) != MAL_SUCCEED) {
1051 :
1052 : return mdbe->msg;
1053 : }
1054 :
1055 0 : mdbe->msg = monetdbe_mapi_dump_table(dbhdl, sname, tname, filename);
1056 :
1057 0 : return mdbe->msg;
1058 : }
1059 :
1060 : char*
1061 0 : monetdbe_get_autocommit(monetdbe_database dbhdl, int* result)
1062 : {
1063 0 : if (!dbhdl)
1064 : return NULL;
1065 :
1066 0 : monetdbe_database_internal *mdbe = (monetdbe_database_internal*)dbhdl;
1067 :
1068 0 : if ((mdbe->msg = validate_database_handle(mdbe, "monetdbe.monetdbe_get_autocommit")) != MAL_SUCCEED) {
1069 :
1070 : return mdbe->msg;
1071 : }
1072 :
1073 0 : if (!result) {
1074 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_get_autocommit", "Parameter result is NULL"));
1075 : } else {
1076 0 : mvc *m = ((backend *) mdbe->c->sqlcontext)->mvc;
1077 0 : *result = m->session->auto_commit;
1078 : }
1079 :
1080 0 : return mdbe->msg;
1081 : }
1082 :
1083 : char*
1084 0 : monetdbe_set_autocommit(monetdbe_database dbhdl, int value)
1085 : {
1086 0 : if (!dbhdl)
1087 : return NULL;
1088 :
1089 0 : monetdbe_database_internal *mdbe = (monetdbe_database_internal*)dbhdl;
1090 :
1091 0 : if (!validate_database_handle_noerror(mdbe)) {
1092 :
1093 : return NULL;
1094 : }
1095 :
1096 0 : mvc *m = ((backend *) mdbe->c->sqlcontext)->mvc;
1097 0 : int commit = !m->session->auto_commit && value;
1098 :
1099 0 : m->session->auto_commit = value;
1100 0 : m->session->ac_on_commit = m->session->auto_commit;
1101 0 : if (m->session->tr->active) {
1102 0 : if (commit) {
1103 0 : mdbe->msg = mvc_commit(m, 0, NULL, true);
1104 : } else {
1105 0 : mdbe->msg = mvc_rollback(m, 0, NULL, true);
1106 : }
1107 : }
1108 :
1109 0 : return mdbe->msg;
1110 : }
1111 :
1112 : int
1113 0 : monetdbe_in_transaction(monetdbe_database dbhdl)
1114 : {
1115 0 : if (!dbhdl)
1116 : return 0;
1117 0 : monetdbe_database_internal *mdbe = (monetdbe_database_internal*)dbhdl;
1118 :
1119 0 : if (!validate_database_handle_noerror(mdbe)) {
1120 :
1121 : return 0;
1122 : }
1123 :
1124 0 : mvc *m = ((backend *) mdbe->c->sqlcontext)->mvc;
1125 0 : int result = 0;
1126 :
1127 0 : if (m->session->tr)
1128 0 : result = m->session->tr->active;
1129 :
1130 : return result;
1131 : }
1132 :
1133 : struct callback_context {
1134 : monetdbe_database_internal *mdbe;
1135 : };
1136 :
1137 : static str
1138 1 : monetdbe_set_remote_results(backend *be, char* tblname, columnar_result* results, size_t nr_results) {
1139 :
1140 1 : char* error = NULL;
1141 :
1142 1 : if (nr_results == 0)
1143 : return MAL_SUCCEED; // No work to do.
1144 :
1145 1 : BAT* b_0 = BATdescriptor(results[0].id); // Fetch the first column to get the count.
1146 1 : if (!b_0) {
1147 0 : error = createException(MAL,"monetdbe.monetdbe_set_remote_results",SQLSTATE(HY005) "Cannot access column descriptor ");
1148 0 : return error;
1149 : }
1150 :
1151 1 : int res = mvc_result_table(be, 0, (int) nr_results, Q_TABLE);
1152 1 : if (res < 0) {
1153 0 : BBPunfix(b_0->batCacheid);
1154 0 : error = createException(MAL,"monetdbe.monetdbe_set_remote_results",SQLSTATE(HY005) "Cannot create result table");
1155 0 : return error;
1156 : }
1157 :
1158 3 : for (size_t i = 0; i < nr_results; i++) {
1159 2 : BAT *b = NULL;
1160 2 : if (i > 0)
1161 1 : b = BATdescriptor(results[i].id);
1162 : else
1163 : b = b_0; // We already fetched this first column
1164 :
1165 2 : char* colname = results[i].colname;
1166 2 : char* tpename = results[i].tpename;
1167 2 : int digits = results[i].digits;
1168 2 : int scale = results[i].scale;
1169 :
1170 2 : if (b == NULL) {
1171 0 : error = createException(MAL,"monetdbe.monetdbe_result_cb",SQLSTATE(HY002) RUNTIME_OBJECT_MISSING);
1172 0 : break;
1173 : }
1174 :
1175 2 : int res = mvc_result_column(be, tblname, colname, tpename, digits, scale, b);
1176 2 : BBPunfix(b->batCacheid);
1177 2 : if (res) {
1178 0 : error = createException(MAL,"monetdbe.monetdbe_result_cb", SQLSTATE(42000) "Cannot access column descriptor %s.%s",tblname,colname);
1179 0 : break;
1180 : }
1181 : }
1182 :
1183 0 : if (error)
1184 0 : res_tables_destroy(be->results);
1185 : return error;
1186 : }
1187 :
1188 : static str
1189 1 : monetdbe_result_cb(void* context, char* tblname, columnar_result* results, size_t nr_results) {
1190 1 : monetdbe_database_internal *mdbe = ((struct callback_context*) context)->mdbe;
1191 :
1192 1 : backend *be = NULL;
1193 1 : if ((mdbe->msg = getBackendContext(mdbe->c, &be)) != NULL)
1194 : return mdbe->msg;
1195 :
1196 1 : return monetdbe_set_remote_results(be, tblname, results, nr_results);
1197 : }
1198 :
1199 : struct prepare_callback_context {
1200 : int* prepare_id;
1201 : monetdbe_database_internal *mdbe;
1202 : };
1203 :
1204 : static str
1205 0 : monetdbe_prepare_cb(void* context, char* tblname, columnar_result* results, size_t nr_results) {
1206 0 : (void) tblname;
1207 0 : monetdbe_database_internal *mdbe = ((struct prepare_callback_context*) context)->mdbe;
1208 0 : int *prepare_id = ((struct prepare_callback_context*) context)->prepare_id;
1209 :
1210 0 : if (nr_results != 7) // 1) btype 2) bdigits 3) bscale 4) bschema 5) btable 6) bcolumn 7) bimpl
1211 0 : return createException(SQL, "monetdbe.monetdbe_prepare_cb", SQLSTATE(42000) "result table for prepared statement is wrong.");
1212 :
1213 0 : backend *be = NULL;
1214 0 : if ((mdbe->msg = getBackendContext(mdbe->c, &be)) != NULL)
1215 : return mdbe->msg;
1216 :
1217 0 : if ( (mdbe->msg = monetdbe_set_remote_results(be, tblname, results, nr_results)) != NULL)
1218 : return mdbe->msg;
1219 :
1220 0 : BAT* btype = NULL;
1221 0 : BAT* bdigits = NULL;
1222 0 : BAT* bscale = NULL;
1223 0 : BAT* bschema = NULL;
1224 0 : BAT* btable = NULL;
1225 0 : BAT* bcolumn = NULL;
1226 0 : BAT* bimpl = NULL;
1227 :
1228 0 : size_t nparams = 0;
1229 0 : BATiter btype_iter = {0};
1230 0 : BATiter bcolumn_iter = {0};
1231 0 : BATiter btable_iter = {0};
1232 0 : BATiter bimpl_iter = {0};
1233 0 : char* function = NULL;
1234 0 : Symbol prg = NULL;
1235 0 : MalBlkPtr mb = NULL;
1236 0 : InstrPtr o = NULL, e = NULL, r = NULL;
1237 0 : sql_rel* rel = NULL;
1238 0 : list *args = NULL, *rets = NULL;
1239 0 : allocator* sa = NULL;
1240 0 : ValRecord v = { .len=0 };
1241 0 : ptr vp = NULL;
1242 0 : struct callback_context* ccontext= NULL;
1243 0 : columnar_result_callback* rcb = NULL;
1244 :
1245 0 : str msg = MAL_SUCCEED;
1246 0 : if (!(btype = BATdescriptor(results[0].id)) ||
1247 0 : !(bdigits = BATdescriptor(results[1].id)) ||
1248 0 : !(bscale = BATdescriptor(results[2].id)) ||
1249 0 : !(bschema = BATdescriptor(results[3].id)) ||
1250 0 : !(btable = BATdescriptor(results[4].id)) ||
1251 0 : !(bcolumn = BATdescriptor(results[5].id)) ||
1252 0 : !(bimpl = BATdescriptor(results[6].id)))
1253 : {
1254 0 : msg = createException(SQL, "monetdbe.monetdbe_prepare_cb", SQLSTATE(HY002) RUNTIME_OBJECT_MISSING);
1255 0 : goto cleanup;
1256 : }
1257 :
1258 0 : nparams = BATcount(btype);
1259 :
1260 0 : if (nparams != BATcount(bdigits) ||
1261 0 : nparams != BATcount(bimpl) ||
1262 0 : nparams != BATcount(bscale) ||
1263 0 : nparams != BATcount(bschema) ||
1264 0 : nparams != BATcount(btable) ||
1265 0 : nparams != BATcount(bcolumn))
1266 : {
1267 0 : msg = createException(SQL, "monetdbe.monetdbe_prepare_cb", SQLSTATE(42000) "Prepare results are incorrect");
1268 0 : goto cleanup;
1269 : }
1270 :
1271 0 : btype_iter = bat_iterator(btype);
1272 0 : bcolumn_iter = bat_iterator(bcolumn);
1273 0 : btable_iter = bat_iterator(btable);
1274 0 : bimpl_iter = bat_iterator(bimpl);
1275 0 : function = BUNtvar(btable_iter, BATcount(btable)-1);
1276 :
1277 : {
1278 0 : assert (((backend*) mdbe->c->sqlcontext)->remote < INT_MAX);
1279 0 : char nme[16] = {0};
1280 0 : const char* name = number2name(nme, sizeof(nme), ++((backend*) mdbe->c->sqlcontext)->remote);
1281 0 : prg = newFunctionArgs(userRef, putName(name), FUNCTIONsymbol, (int) nparams + 1);
1282 : }
1283 :
1284 0 : resizeMalBlk(prg->def, (int) nparams + 3 /*function declaration + remote.exec + return statement*/);
1285 0 : mb = prg->def;
1286 :
1287 0 : o = getInstrPtr(mb, 0);
1288 0 : o->retc = o->argc = 0;
1289 :
1290 0 : e = newInstructionArgs(mb, remoteRef, execRef, (int)(nparams + 5));
1291 0 : if (e == NULL) {
1292 0 : msg = createException(MAL, "monetdbe.monetdbe_prepare_cb", MAL_MALLOC_FAIL);
1293 0 : goto cleanup;
1294 : }
1295 0 : setDestVar(e, newTmpVariable(mb, TYPE_any));
1296 0 : e = pushStr(mb, e, mdbe->mid);
1297 0 : e = pushStr(mb, e, userRef);
1298 0 : e = pushStr(mb, e, function);
1299 :
1300 0 : rcb = GDKmalloc(sizeof(columnar_result_callback));
1301 0 : if (rcb == NULL) {
1302 0 : msg = createException(MAL, "monetdbe.monetdbe_prepare_cb", MAL_MALLOC_FAIL);
1303 0 : goto cleanup;
1304 : }
1305 :
1306 0 : ccontext = GDKzalloc(sizeof(struct callback_context));
1307 0 : if (ccontext == NULL) {
1308 0 : msg = createException(MAL, "monetdbe.monetdbe_prepare_cb", MAL_MALLOC_FAIL);
1309 0 : goto cleanup;
1310 : }
1311 :
1312 0 : ccontext->mdbe = mdbe;
1313 :
1314 0 : rcb->context = ccontext;
1315 0 : rcb->call = monetdbe_result_cb;
1316 :
1317 0 : vp = (ptr) rcb;
1318 :
1319 0 : VALset(&v, TYPE_ptr, &vp);
1320 0 : e = pushValue(mb, e, &v);
1321 :
1322 0 : r = newInstruction(mb, NULL, NULL);
1323 0 : if (r == NULL) {
1324 0 : msg = createException(MAL, "monetdbe.monetdbe_prepare_cb", MAL_MALLOC_FAIL);
1325 0 : goto cleanup;
1326 : }
1327 0 : r->barrier= RETURNsymbol;
1328 0 : r->argc= r->retc=0;
1329 :
1330 0 : sa = be->mvc->sa;
1331 :
1332 0 : args = new_exp_list(sa);
1333 0 : rets = new_exp_list(sa);
1334 :
1335 0 : for (size_t i = 0; i < nparams; i++) {
1336 :
1337 0 : const char *table = BUNtvar(btable_iter, i);
1338 0 : sql_type *t = SA_ZNEW(sa, sql_type);
1339 0 : const char *name = BUNtvar(btype_iter, i);
1340 0 : t->base.name = SA_STRDUP(sa, name);
1341 0 : const char *impl = BUNtvar(bimpl_iter, i);
1342 0 : t->impl = SA_STRDUP(sa, impl);
1343 0 : t->localtype = ATOMindex(t->impl);
1344 :
1345 0 : sql_subtype *st = SA_ZNEW(sa, sql_subtype);
1346 0 : sql_init_subtype(st, t, (unsigned) *(int*) Tloc(bdigits, i), (unsigned) *(int*) Tloc(bscale, i));
1347 :
1348 0 : if (strNil(table)) {
1349 : // input argument
1350 :
1351 0 : sql_arg *a = SA_ZNEW(sa, sql_arg);
1352 0 : a->type = *st;
1353 0 : append(args, a);
1354 :
1355 0 : int idx = newVariable(mb, NULL, 0, t->localtype);
1356 0 : o = pushArgument(mb, o, idx);
1357 :
1358 0 : InstrPtr p = newFcnCall(mb, remoteRef, putRef);
1359 0 : if (p == NULL) {
1360 0 : msg = createException(MAL, "monetdbe.monetdbe_prepare_cb", MAL_MALLOC_FAIL);
1361 0 : goto cleanup;
1362 : }
1363 0 : setArgType(mb, p, 0, TYPE_str);
1364 0 : p = pushStr(mb, p, mdbe->mid);
1365 0 : p = pushArgument(mb, p, idx);
1366 0 : pushInstruction(mb, p);
1367 :
1368 0 : e = pushArgument(mb, e, getArg(p, 0));
1369 : }
1370 : else {
1371 : // output argument
1372 :
1373 0 : const char *column = BUNtvar(bcolumn_iter, i);
1374 0 : sql_exp * c = exp_column(sa, table, column, st, CARD_MULTI, true, false, false);
1375 0 : append(rets, c);
1376 : }
1377 : }
1378 0 : pushInstruction(mb, e);
1379 0 : pushInstruction(mb, r);
1380 0 : e = r = NULL;
1381 :
1382 0 : if ( (mdbe->msg = chkProgram(mdbe->c->usermodule, mb)) != MAL_SUCCEED ) {
1383 0 : msg = mdbe->msg;
1384 0 : goto cleanup;
1385 : }
1386 :
1387 0 : rel = rel_project(sa, NULL, rets);
1388 0 : be->q = qc_insert(be->mvc->qc, sa, rel, NULL, args, be->mvc->type, NULL, be->no_mitosis);
1389 0 : *prepare_id = be->q->id;
1390 :
1391 : /*
1392 : * HACK: we need to rename the Symbol aka MAL function to the query cache name.
1393 : * Basically we keep the MALblock but we destroy the containing old Symbol
1394 : * and create a new one with the correct name and set its MAL block pointer to
1395 : * point to the mal block we have created in this function.
1396 : */
1397 0 : prg->def = NULL;
1398 0 : freeSymbol(prg);
1399 0 : if ((prg = newFunctionArgs(userRef, putName(be->q->name), FUNCTIONsymbol, -1)) == NULL) {
1400 0 : msg = createException(MAL, "monetdbe.monetdbe_prepare_cb", MAL_MALLOC_FAIL);
1401 0 : goto cleanup;
1402 : }
1403 0 : freeMalBlk(prg->def);
1404 0 : prg->def = mb;
1405 :
1406 : // finally add this beautiful new function to the local user module.
1407 0 : insertSymbol(mdbe->c->usermodule, prg);
1408 :
1409 0 : cleanup:
1410 0 : freeInstruction(e);
1411 0 : freeInstruction(r);
1412 0 : if (bcolumn) {
1413 0 : bat_iterator_end(&btype_iter);
1414 0 : bat_iterator_end(&bcolumn_iter);
1415 0 : bat_iterator_end(&btable_iter);
1416 0 : bat_iterator_end(&bimpl_iter);
1417 : }
1418 : // clean these up
1419 0 : BBPreclaim(btype);
1420 0 : BBPreclaim(bimpl);
1421 0 : BBPreclaim(bdigits);
1422 0 : BBPreclaim(bscale);
1423 0 : BBPreclaim(bschema);
1424 0 : BBPreclaim(btable);
1425 0 : BBPreclaim(bcolumn);
1426 :
1427 0 : if (msg && rcb) GDKfree(rcb);
1428 0 : if (msg && ccontext) GDKfree(ccontext);
1429 :
1430 : return msg;
1431 : }
1432 :
1433 : static char*
1434 1 : monetdbe_query_remote(monetdbe_database_internal *mdbe, char* query, monetdbe_result** result, monetdbe_cnt* affected_rows, int *prepare_id)
1435 : {
1436 1 : const char mod[] = "user";
1437 1 : char nme[16];
1438 :
1439 1 : Client c = mdbe->c;
1440 :
1441 1 : const char *name = number2name(nme, sizeof(nme), ++((backend*) c->sqlcontext)->remote);
1442 1 : Symbol prg = newFunction(putName(mod), putName(name), FUNCTIONsymbol);
1443 :
1444 1 : if (prg == NULL) {
1445 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_query_remote", MAL_MALLOC_FAIL));
1446 0 : return mdbe->msg;
1447 : }
1448 :
1449 1 : MalBlkPtr mb = prg->def;
1450 1 : ValRecord v;
1451 1 : ptr vp;
1452 1 : columnar_result_callback* rcb;
1453 1 : InstrPtr f = getInstrPtr(mb, 0), r, p, e, o;
1454 :
1455 1 : f->retc = f->argc = 0;
1456 1 : o = newStmt(mb, remoteRef, putRef);
1457 1 : o = pushStr(mb, o, mdbe->mid);
1458 1 : o = pushBit(mb, o, TRUE);
1459 1 : if (o == NULL) {
1460 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_query_remote", MAL_MALLOC_FAIL));
1461 0 : goto finalize;
1462 : }
1463 1 : pushInstruction(mb, o);
1464 :
1465 1 : if (prepare_id) {
1466 0 : size_t query_len, input_query_len, prep_len = 0;
1467 0 : input_query_len = strlen(query);
1468 0 : query_len = input_query_len + 3;
1469 0 : const char PREPARE[] = "PREPARE ";
1470 0 : prep_len = sizeof(PREPARE)-1;
1471 0 : query_len += prep_len;
1472 0 : char *nq = NULL;
1473 0 : if (!(nq = GDKmalloc(query_len))) {
1474 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_query_remote", "Could not setup query stream"));
1475 0 : goto finalize;
1476 : }
1477 0 : strcpy(nq, PREPARE);
1478 0 : strcpy(nq + prep_len, query);
1479 0 : strcpy(nq + prep_len + input_query_len, "\n;");
1480 0 : query = nq;
1481 : }
1482 :
1483 1 : p = newStmt(mb, remoteRef, putRef);
1484 1 : p = pushStr(mb, p, mdbe->mid);
1485 1 : p = pushStr(mb, p, query);
1486 1 : if (p == NULL) {
1487 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_query_remote", MAL_MALLOC_FAIL));
1488 0 : goto finalize;
1489 : }
1490 1 : pushInstruction(mb, p);
1491 :
1492 :
1493 1 : e = newInstruction(mb, remoteRef, execRef);
1494 1 : if (e == NULL) {
1495 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_query_remote", MAL_MALLOC_FAIL));
1496 0 : goto finalize;
1497 : }
1498 1 : setDestVar(e, newTmpVariable(mb, TYPE_any));
1499 1 : e = pushStr(mb, e, mdbe->mid);
1500 1 : e = pushStr(mb, e, sqlRef);
1501 1 : e = pushStr(mb, e, evalRef);
1502 :
1503 : /*
1504 : * prepare the call back routine and its context
1505 : * and pass it over as a pointer to remote.exec.
1506 : */
1507 1 : rcb = GDKzalloc(sizeof(columnar_result_callback));
1508 1 : if (!rcb) {
1509 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_query_remote", "Could not setup query stream"));
1510 0 : goto finalize;
1511 : }
1512 :
1513 1 : if (!prepare_id) {
1514 1 : struct callback_context* ccontext;
1515 1 : ccontext = GDKzalloc(sizeof(struct callback_context));
1516 1 : if (ccontext)
1517 1 : ccontext->mdbe = mdbe;
1518 1 : rcb->context = ccontext;
1519 1 : rcb->call = monetdbe_result_cb;
1520 : }
1521 : else {
1522 0 : struct prepare_callback_context* ccontext;
1523 0 : ccontext = GDKzalloc(sizeof(struct prepare_callback_context));
1524 0 : if (ccontext) {
1525 0 : ccontext->mdbe = mdbe;
1526 0 : ccontext->prepare_id = prepare_id;
1527 : }
1528 0 : rcb->context = ccontext;
1529 0 : rcb->call = monetdbe_prepare_cb;
1530 : }
1531 1 : if (!rcb->context) {
1532 0 : GDKfree(rcb);
1533 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_query_remote", "Could not setup query stream"));
1534 0 : goto finalize;
1535 : }
1536 :
1537 1 : vp = (ptr) rcb;
1538 1 : VALset(&v, TYPE_ptr, &vp);
1539 1 : e = pushValue(mb, e, &v);
1540 :
1541 1 : e = pushArgument(mb, e, getArg(p, 0));
1542 1 : e = pushArgument(mb, e, getArg(o, 0));
1543 :
1544 1 : pushInstruction(mb, e);
1545 :
1546 1 : r = newInstruction(mb, NULL, NULL);
1547 1 : if (r == NULL) {
1548 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_query_remote", MAL_MALLOC_FAIL));
1549 0 : goto finalize;
1550 : }
1551 1 : r->barrier= RETURNsymbol;
1552 1 : r->argc= r->retc=0;
1553 1 : pushInstruction(mb, r);
1554 :
1555 1 : if ( (mdbe->msg = chkProgram(c->usermodule, mb)) != MAL_SUCCEED )
1556 0 : goto finalize;
1557 :
1558 1 : if ( (mdbe->msg = runMAL(c, mb, 0, NULL)) != MAL_SUCCEED )
1559 0 : goto finalize;
1560 :
1561 1 : if (result) {
1562 1 : if ((mdbe->msg = monetdbe_get_results(result, mdbe)) != MAL_SUCCEED)
1563 0 : goto finalize;
1564 :
1565 1 : mvc* m = NULL;
1566 1 : backend * be = NULL;
1567 1 : if ((mdbe->msg = getSQLContext(c, NULL, &m, &be)) != MAL_SUCCEED)
1568 0 : goto finalize;
1569 :
1570 1 : if (m->emode & m_prepare)
1571 0 : ((monetdbe_result_internal*) *result)->type = Q_PREPARE;
1572 : else
1573 1 : ((monetdbe_result_internal*) *result)->type = (be->results) ? be->results->query_type : m->type;
1574 :
1575 1 : if (!be->results && be->rowcnt >= 0 && affected_rows)
1576 0 : *affected_rows = be->rowcnt;
1577 : }
1578 :
1579 0 : finalize:
1580 1 : freeSymbol(prg);
1581 1 : return mdbe->msg;
1582 : }
1583 :
1584 : char*
1585 1 : monetdbe_query(monetdbe_database dbhdl, char* query, monetdbe_result** result, monetdbe_cnt* affected_rows)
1586 : {
1587 1 : if (!dbhdl)
1588 : return NULL;
1589 1 : monetdbe_database_internal *mdbe = (monetdbe_database_internal*)dbhdl;
1590 :
1591 1 : assert(mdbe->c);
1592 1 : MT_thread_set_qry_ctx(&mdbe->c->qryctx);
1593 1 : if (mdbe->mid) {
1594 1 : mdbe->msg = monetdbe_query_remote(mdbe, query, result, affected_rows, NULL);
1595 : }
1596 : else {
1597 0 : mdbe->msg = monetdbe_query_internal(mdbe, query, result, affected_rows, NULL, 'S');
1598 : }
1599 :
1600 1 : return mdbe->msg;
1601 : }
1602 :
1603 : char*
1604 0 : monetdbe_prepare(monetdbe_database dbhdl, char* query, monetdbe_statement **stmt, monetdbe_result** result)
1605 : {
1606 0 : if (!dbhdl)
1607 : return NULL;
1608 0 : monetdbe_database_internal *mdbe = (monetdbe_database_internal*)dbhdl;
1609 :
1610 0 : int prepare_id = 0;
1611 :
1612 0 : assert(mdbe->c);
1613 0 : MT_thread_set_qry_ctx(&mdbe->c->qryctx);
1614 0 : if (!stmt) {
1615 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_prepare", "Parameter stmt is NULL"));
1616 0 : assert(mdbe->msg != MAL_SUCCEED); /* help Coverity */
1617 0 : } else if (mdbe->mid) {
1618 0 : mdbe->msg = monetdbe_query_remote(mdbe, query, result, NULL, &prepare_id);
1619 : } else {
1620 0 : *stmt = NULL;
1621 0 : mdbe->msg = monetdbe_query_internal(mdbe, query, result, NULL, &prepare_id, 'S');
1622 : }
1623 :
1624 0 : if (mdbe->msg == MAL_SUCCEED) {
1625 0 : mvc *m = ((backend *) mdbe->c->sqlcontext)->mvc;
1626 0 : monetdbe_stmt_internal *stmt_internal = (monetdbe_stmt_internal*)GDKzalloc(sizeof(monetdbe_stmt_internal));
1627 0 : cq *q = qc_find(m->qc, prepare_id);
1628 :
1629 0 : if (q && stmt_internal) {
1630 0 : Symbol s = findSymbolInModule(mdbe->c->usermodule, q->f->imp);
1631 0 : assert(s->def);
1632 0 : InstrPtr p = s->def->stmt[0];
1633 0 : stmt_internal->mdbe = mdbe;
1634 0 : stmt_internal->q = q;
1635 0 : stmt_internal->retc = p->retc;
1636 0 : stmt_internal->res.nparam = list_length(q->f->ops);
1637 0 : stmt_internal->args = (ValPtr*)GDKmalloc(sizeof(ValPtr) * (stmt_internal->res.nparam + stmt_internal->retc));
1638 0 : stmt_internal->data = (ValRecord*)GDKzalloc(sizeof(ValRecord) * (stmt_internal->res.nparam+1));
1639 0 : stmt_internal->res.type = (monetdbe_types*)GDKmalloc(sizeof(monetdbe_types) * (stmt_internal->res.nparam+1));
1640 0 : if (!stmt_internal->res.type || !stmt_internal->data || !stmt_internal->args) {
1641 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_prepare", MAL_MALLOC_FAIL));
1642 0 : } else if (q->f->ops) {
1643 0 : int i = 0;
1644 0 : for (node *n = q->f->ops->h; n; n = n->next, i++) {
1645 0 : sql_arg *a = n->data;
1646 0 : sql_subtype *t = &a->type;
1647 0 : stmt_internal->res.type[i] = embedded_type(t->type->localtype);
1648 0 : stmt_internal->args[i+stmt_internal->retc] = &stmt_internal->data[i];
1649 : }
1650 : }
1651 0 : } else if (!stmt_internal)
1652 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_prepare", MAL_MALLOC_FAIL));
1653 :
1654 0 : if (mdbe->msg == MAL_SUCCEED)
1655 0 : *stmt = (monetdbe_statement*)stmt_internal;
1656 0 : else if (stmt_internal) {
1657 0 : GDKfree(stmt_internal->data);
1658 0 : GDKfree(stmt_internal->args);
1659 0 : GDKfree(stmt_internal->res.type);
1660 0 : GDKfree(stmt_internal);
1661 0 : *stmt = NULL;
1662 : }
1663 : }
1664 :
1665 0 : return mdbe->msg;
1666 : }
1667 :
1668 : char*
1669 0 : monetdbe_bind(monetdbe_statement *stmt, void *data, size_t i)
1670 : {
1671 0 : monetdbe_stmt_internal *stmt_internal = (monetdbe_stmt_internal*)stmt;
1672 :
1673 0 : if (i >= stmt->nparam)
1674 0 : return createException(MAL, "monetdbe.monetdbe_bind", "Parameter %zu not bound to a value", i);
1675 0 : sql_arg *a = (sql_arg*)list_fetch(stmt_internal->q->f->ops, (int) i);
1676 0 : assert(a);
1677 0 : int tpe = a->type.type->localtype;
1678 0 : stmt_internal->data[i].vtype = tpe;
1679 :
1680 0 : const void* nil = (tpe>=0)?ATOMnilptr(tpe):NULL;
1681 0 : if (!data) {
1682 0 : VALset(&stmt_internal->data[i], tpe, (ptr)nil);
1683 0 : } else if (tpe == TYPE_timestamp) {
1684 0 : monetdbe_data_timestamp* ts = (monetdbe_data_timestamp*)data;
1685 0 : timestamp t = *(timestamp*) nil;
1686 0 : if(!timestamp_is_null(ts))
1687 0 : t = timestamp_from_data(ts);
1688 0 : VALset(&stmt_internal->data[i], tpe, &t);
1689 : } else if (tpe == TYPE_date) {
1690 0 : monetdbe_data_date* de = (monetdbe_data_date*)data;
1691 0 : date d = *(date*) nil;
1692 0 : if(!date_is_null(de))
1693 0 : d = date_from_data(de);
1694 0 : VALset(&stmt_internal->data[i], tpe, &d);
1695 : } else if (tpe == TYPE_daytime) {
1696 0 : monetdbe_data_time* t = (monetdbe_data_time*)data;
1697 0 : daytime dt = *(daytime*) nil;
1698 :
1699 0 : if(!time_is_null(t))
1700 0 : dt = time_from_data(t);
1701 0 : VALset(&stmt_internal->data[i], tpe, &dt);
1702 : } else if (tpe == TYPE_blob) {
1703 0 : monetdbe_data_blob *be = (monetdbe_data_blob*)data;
1704 0 : blob *b = (blob*)nil;
1705 0 : if (!blob_is_null(be)) {
1706 0 : size_t len = be->size;
1707 0 : b = (blob*) GDKmalloc(blobsize(len));
1708 0 : if (b == NULL) {
1709 0 : set_error(stmt_internal->mdbe, createException(MAL, "monetdbe.monetdbe_bind", MAL_MALLOC_FAIL));
1710 0 : return stmt_internal->mdbe->msg;
1711 : }
1712 0 : b->nitems = len;
1713 0 : memcpy(b->data, be->data, len);
1714 : }
1715 0 : VALset(&stmt_internal->data[i], tpe, b);
1716 : } else if (tpe == TYPE_str) {
1717 0 : char *val = GDKstrdup(data);
1718 :
1719 0 : if (val == NULL) {
1720 0 : set_error(stmt_internal->mdbe, createException(MAL, "monetdbe.monetdbe_bind", MAL_MALLOC_FAIL));
1721 0 : return stmt_internal->mdbe->msg;
1722 : }
1723 0 : VALset(&stmt_internal->data[i], tpe, val);
1724 : } else {
1725 0 : VALset(&stmt_internal->data[i], tpe, data);
1726 : }
1727 : return MAL_SUCCEED;
1728 : }
1729 :
1730 : char*
1731 0 : monetdbe_execute(monetdbe_statement *stmt, monetdbe_result **result, monetdbe_cnt *affected_rows)
1732 : {
1733 0 : monetdbe_result_internal *res_internal = NULL;
1734 0 : monetdbe_stmt_internal *stmt_internal = (monetdbe_stmt_internal*)stmt;
1735 0 : backend *b = (backend *) stmt_internal->mdbe->c->sqlcontext;
1736 0 : mvc *m = b->mvc;
1737 0 : monetdbe_database_internal *mdbe = stmt_internal->mdbe;
1738 0 : MalStkPtr glb = NULL;
1739 0 : cq *q = stmt_internal->q;
1740 0 : Symbol s = NULL;
1741 :
1742 0 : assert(mdbe->c);
1743 0 : MT_thread_set_qry_ctx(&mdbe->c->qryctx);
1744 0 : if ((mdbe->msg = SQLtrans(m)) != MAL_SUCCEED)
1745 : return mdbe->msg;
1746 :
1747 : /* check if all inputs are bound */
1748 0 : for(int i = 0; i< list_length(stmt_internal->q->f->ops); i++){
1749 0 : if (!stmt_internal->data[i].vtype) {
1750 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_execute", "Parameter %d not bound to a value", i));
1751 0 : goto cleanup;
1752 : }
1753 : }
1754 :
1755 0 : s = findSymbolInModule(mdbe->c->usermodule, q->f->imp);
1756 0 : if ((mdbe->msg = callMAL(mdbe->c, s->def, &glb, stmt_internal->args)) != MAL_SUCCEED)
1757 0 : goto cleanup;
1758 :
1759 0 : if (b->rowcnt >= 0 && affected_rows)
1760 0 : *affected_rows = b->rowcnt;
1761 :
1762 0 : if (result) {
1763 0 : if ((mdbe->msg = monetdbe_get_results(result, mdbe)) != MAL_SUCCEED) {
1764 0 : goto cleanup;
1765 : }
1766 :
1767 0 : res_internal = *(monetdbe_result_internal**)result;
1768 0 : res_internal->type = (b->results) ? Q_TABLE : Q_UPDATE;
1769 0 : if (res_internal->monetdbe_resultset && res_internal->monetdbe_resultset->query_type == Q_TABLE) {
1770 0 : res_internal->type = Q_TABLE;
1771 0 : if (affected_rows)
1772 0 : *affected_rows = res_internal->monetdbe_resultset->nr_rows;
1773 : }
1774 : }
1775 :
1776 0 : cleanup:
1777 0 : GDKfree(glb);
1778 0 : return commit_action(m, stmt_internal->mdbe, result, res_internal);
1779 : }
1780 :
1781 : char*
1782 0 : monetdbe_cleanup_statement(monetdbe_database dbhdl, monetdbe_statement *stmt)
1783 : {
1784 0 : monetdbe_stmt_internal *stmt_internal = (monetdbe_stmt_internal*)stmt;
1785 0 : monetdbe_database_internal *mdbe = (monetdbe_database_internal*)dbhdl;
1786 0 : mvc *m = ((backend *) mdbe->c->sqlcontext)->mvc;
1787 0 : cq *q = stmt_internal->q;
1788 :
1789 0 : assert(!stmt_internal->mdbe || mdbe == stmt_internal->mdbe);
1790 :
1791 0 : assert(mdbe->c);
1792 0 : MT_thread_set_qry_ctx(&mdbe->c->qryctx);
1793 0 : for (size_t i = 0; i < stmt_internal->res.nparam + 1; i++) {
1794 0 : ValPtr data = &stmt_internal->data[i];
1795 0 : VALclear(data);
1796 : }
1797 :
1798 0 : GDKfree(stmt_internal->data);
1799 0 : GDKfree(stmt_internal->args);
1800 0 : GDKfree(stmt_internal->res.type);
1801 0 : GDKfree(stmt_internal);
1802 :
1803 0 : if (q)
1804 0 : qc_delete(m->qc, q);
1805 0 : return MAL_SUCCEED;
1806 : }
1807 :
1808 : char*
1809 1 : monetdbe_cleanup_result(monetdbe_database dbhdl, monetdbe_result* result)
1810 : {
1811 1 : monetdbe_database_internal *mdbe = (monetdbe_database_internal*)dbhdl;
1812 1 : monetdbe_result_internal* res = (monetdbe_result_internal *) result;
1813 :
1814 1 : assert(mdbe->c);
1815 1 : MT_thread_set_qry_ctx(&mdbe->c->qryctx);
1816 1 : if (!result) {
1817 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_cleanup_result", "Parameter result is NULL"));
1818 : } else {
1819 1 : mdbe->msg = monetdbe_cleanup_result_internal(mdbe, res);
1820 : }
1821 :
1822 1 : return mdbe->msg;
1823 : }
1824 :
1825 : static inline void
1826 0 : cleanup_get_columns_result(size_t column_count, monetdbe_column* columns)
1827 : {
1828 0 : if (columns) {
1829 0 : for (size_t c = 0; c < column_count; c++) {
1830 0 : GDKfree(columns[c].name);
1831 0 : GDKfree(columns[c].sql_type.name);
1832 : }
1833 0 : GDKfree(columns);
1834 0 : columns = NULL;
1835 : }
1836 0 : }
1837 :
1838 : static char *
1839 0 : escape_identifier(const char *s) /* Escapes a SQL identifier string, ie the " and \ characters */
1840 : {
1841 0 : char *ret = NULL, *q;
1842 0 : const char *p = s;
1843 :
1844 : /* At most we will need 2*strlen(s) + 1 characters */
1845 0 : if (!(ret = (char *)GDKmalloc(2*strlen(s) + 1)))
1846 : return NULL;
1847 :
1848 0 : for (q = ret; *p; p++, q++) {
1849 0 : *q = *p;
1850 0 : if (*p == '"')
1851 0 : *(++q) = '"';
1852 0 : else if (*p == '\\')
1853 0 : *(++q) = '\\';
1854 : }
1855 :
1856 0 : *q = '\0';
1857 0 : return ret;
1858 : }
1859 :
1860 : static char*
1861 0 : monetdbe_get_columns_remote(monetdbe_database_internal *mdbe, const char* schema_name, const char *table_name, size_t *column_count,
1862 : monetdbe_column **columns)
1863 : {
1864 0 : char buf[1024], *escaped_schema_name = NULL, *escaped_table_name = NULL;
1865 :
1866 0 : if (schema_name && !(escaped_schema_name = escape_identifier(schema_name))) {
1867 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_get_columns", MAL_MALLOC_FAIL));
1868 0 : return mdbe->msg;
1869 : }
1870 0 : if (!(escaped_table_name = escape_identifier(table_name))) {
1871 0 : GDKfree(escaped_schema_name);
1872 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_get_columns", MAL_MALLOC_FAIL));
1873 0 : return mdbe->msg;
1874 : }
1875 :
1876 0 : int len = snprintf(buf, 1024, "SELECT * FROM %s%s%s\"%s\" WHERE FALSE;",
1877 : escaped_schema_name ? "\"" : "", escaped_schema_name ? escaped_schema_name : "",
1878 : escaped_schema_name ? escaped_schema_name : "\".", escaped_table_name);
1879 0 : GDKfree(escaped_schema_name);
1880 0 : GDKfree(escaped_table_name);
1881 0 : if (len == -1 || len >= 1024) {
1882 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_get_columns", "Schema and table path is too large"));
1883 0 : return mdbe->msg;
1884 : }
1885 :
1886 0 : monetdbe_result* result = NULL;
1887 :
1888 0 : if ((mdbe->msg = monetdbe_query_remote(mdbe, buf, &result, NULL, NULL)) != MAL_SUCCEED) {
1889 : return mdbe->msg;
1890 : }
1891 :
1892 0 : *column_count = result->ncols;
1893 0 : *columns = GDKzalloc(sizeof(monetdbe_column) * result->ncols);
1894 :
1895 :
1896 0 : if (*columns == NULL)
1897 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_get_columns", MAL_MALLOC_FAIL));
1898 :
1899 0 : if (!mdbe->msg)
1900 0 : for (size_t c = 0; c < result->ncols; c++) {
1901 0 : monetdbe_column* rcol;
1902 0 : if ((mdbe->msg = monetdbe_result_fetch(result, &rcol, c)) != NULL) {
1903 : break;
1904 : }
1905 :
1906 0 : if (((*columns)[c].name = GDKstrdup(rcol->name)) == NULL) {
1907 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_get_columns", MAL_MALLOC_FAIL));
1908 0 : break;
1909 : }
1910 0 : if (((*columns)[c].sql_type.name = GDKstrdup(rcol->sql_type.name)) == NULL) {
1911 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_get_columns", MAL_MALLOC_FAIL));
1912 0 : break;
1913 : }
1914 0 : (*columns)[c].type = rcol->type;
1915 0 : (*columns)[c].sql_type.scale = rcol->sql_type.scale;
1916 0 : (*columns)[c].sql_type.digits = rcol->sql_type.digits;
1917 : }
1918 :
1919 : // cleanup
1920 0 : char* msg = monetdbe_cleanup_result_internal(mdbe, (monetdbe_result_internal*) result);
1921 :
1922 0 : if (msg && mdbe->msg) {
1923 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_get_columns", "multiple errors: %s; %s", mdbe->msg, msg));
1924 : }
1925 0 : else if (msg) {
1926 0 : mdbe->msg = msg;
1927 : }
1928 :
1929 0 : if (mdbe->msg ) {
1930 0 : cleanup_get_columns_result(*column_count, *columns);
1931 : }
1932 :
1933 0 : return mdbe->msg;
1934 : }
1935 :
1936 : char*
1937 0 : monetdbe_get_columns(monetdbe_database dbhdl, const char* schema_name, const char *table_name, size_t *column_count, monetdbe_column **columns)
1938 : {
1939 0 : monetdbe_database_internal *mdbe = (monetdbe_database_internal*)dbhdl;
1940 0 : mvc *m = NULL;
1941 0 : sql_table *t = NULL;
1942 :
1943 0 : if ((mdbe->msg = validate_database_handle(mdbe, "monetdbe.monetdbe_get_columns")) != MAL_SUCCEED) {
1944 : return mdbe->msg;
1945 : }
1946 0 : if (!column_count) {
1947 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_get_columns", "Parameter column_count is NULL"));
1948 0 : return mdbe->msg;
1949 : }
1950 0 : if (!columns) {
1951 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_get_columns", "Parameter columns is NULL"));
1952 0 : return mdbe->msg;
1953 : }
1954 0 : if (!table_name) {
1955 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_get_columns", "Parameter table_name is NULL"));
1956 0 : return mdbe->msg;
1957 : }
1958 :
1959 0 : if (mdbe->mid) {
1960 0 : return monetdbe_get_columns_remote(mdbe, schema_name, table_name, column_count, columns);
1961 : }
1962 :
1963 0 : if ((mdbe->msg = getSQLContext(mdbe->c, NULL, &m, NULL)) != MAL_SUCCEED)
1964 : return mdbe->msg;
1965 0 : if ((mdbe->msg = SQLtrans(m)) != MAL_SUCCEED)
1966 : return mdbe->msg;
1967 0 : if (!(t = find_table_or_view_on_scope(m, NULL, schema_name, table_name, "CATALOG", false))) {
1968 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_get_columns", "%s", m->errstr + 6)); /* Skip error code */
1969 0 : goto cleanup;
1970 : }
1971 :
1972 0 : *column_count = ol_length(t->columns);
1973 0 : if ((*columns = GDKzalloc(*column_count * sizeof(monetdbe_column))) == NULL ) {
1974 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_get_columns", MAL_MALLOC_FAIL));
1975 0 : goto cleanup;
1976 : }
1977 :
1978 0 : for (node *n = ol_first_node(t->columns); n; n = n->next) {
1979 0 : sql_column *col = n->data;
1980 0 : (*columns)[col->colnr].name = col->base.name; // TODO Shouldn't this be string dupped?
1981 0 : if (((*columns)[col->colnr].sql_type.name = GDKstrdup(col->type.type->base.name)) == NULL) goto cleanup;
1982 0 : (*columns)[col->colnr].type = embedded_type(col->type.type->localtype);
1983 0 : (*columns)[col->colnr].sql_type.digits = col->type.type->digits;
1984 0 : (*columns)[col->colnr].sql_type.scale = col->type.type->scale;
1985 : }
1986 :
1987 0 : cleanup:
1988 0 : mdbe->msg = commit_action(m, mdbe, NULL, NULL);
1989 :
1990 0 : return mdbe->msg;
1991 : }
1992 :
1993 : #define GENERATE_BASE_HEADERS(type, tpename) \
1994 : static int tpename##_is_null(type *value)
1995 :
1996 : #define GENERATE_BASE_FUNCTIONS(tpe, tpename, mname) \
1997 : GENERATE_BASE_HEADERS(tpe, tpename); \
1998 : static int tpename##_is_null(tpe *value) { return *value == mname##_nil; }
1999 :
2000 : #ifdef bool
2001 : #undef bool
2002 : #endif
2003 :
2004 0 : GENERATE_BASE_FUNCTIONS(int8_t, bool, bit)
2005 0 : GENERATE_BASE_FUNCTIONS(int8_t, int8_t, bte)
2006 0 : GENERATE_BASE_FUNCTIONS(int16_t, int16_t, sht)
2007 0 : GENERATE_BASE_FUNCTIONS(int32_t, int32_t, int)
2008 0 : GENERATE_BASE_FUNCTIONS(int64_t, int64_t, lng)
2009 : #ifdef HAVE_HGE
2010 0 : GENERATE_BASE_FUNCTIONS(__int128, int128_t, hge)
2011 : #endif
2012 0 : GENERATE_BASE_FUNCTIONS(size_t, size_t, oid)
2013 :
2014 0 : GENERATE_BASE_FUNCTIONS(float, float, flt)
2015 0 : GENERATE_BASE_FUNCTIONS(double, double, dbl)
2016 :
2017 : GENERATE_BASE_HEADERS(char*, str);
2018 : GENERATE_BASE_HEADERS(monetdbe_data_blob, blob);
2019 :
2020 : GENERATE_BASE_HEADERS(monetdbe_data_date, date);
2021 : GENERATE_BASE_HEADERS(monetdbe_data_time, time);
2022 : GENERATE_BASE_HEADERS(monetdbe_data_timestamp, timestamp);
2023 :
2024 : #define GENERATE_BAT_INPUT_BASE(tpe) \
2025 : monetdbe_column_##tpe *bat_data = GDKzalloc(sizeof(monetdbe_column_##tpe)); \
2026 : if (!bat_data) { \
2027 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_result_fetch", MAL_MALLOC_FAIL)); \
2028 : goto cleanup; \
2029 : } \
2030 : bat_data->type = monetdbe_##tpe; \
2031 : if ((bat_data->sql_type.name = GDKstrdup(sqltpe->type->base.name)) == NULL) {\
2032 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_result_fetch", MAL_MALLOC_FAIL)); \
2033 : goto cleanup; \
2034 : } \
2035 : bat_data->sql_type.scale = sqltpe->scale; \
2036 : bat_data->sql_type.digits = sqltpe->digits; \
2037 : bat_data->is_null = tpe##_is_null; \
2038 : if (sqltpe->type->radix == 10) bat_data->scale = pow(10, sqltpe->scale); \
2039 : column_result = (monetdbe_column*) bat_data;
2040 :
2041 : #define GENERATE_BAT_INPUT(b, tpe, tpe_name, mtype) \
2042 : { \
2043 : GENERATE_BAT_INPUT_BASE(tpe_name); \
2044 : bat_data->count = (size_t) mres->nrows; \
2045 : bat_data->null_value = mtype##_nil; \
2046 : if (bat_data->count) { \
2047 : bat_data->data = GDKzalloc(bat_data->count * sizeof(bat_data->null_value)); \
2048 : if (!bat_data->data) { \
2049 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_result_fetch", MAL_MALLOC_FAIL)); \
2050 : goto cleanup; \
2051 : } \
2052 : } \
2053 : size_t it = 0; \
2054 : mtype* val = (mtype*)Tloc(b, 0); \
2055 : /* bat is dense, materialize it */ \
2056 : for (it = 0; it < bat_data->count; it++, val++) \
2057 : bat_data->data[it] = (tpe) *val; \
2058 : }
2059 :
2060 : static char*
2061 0 : append_create_remote_append_mal_program(
2062 : Symbol* prg,
2063 : sql_schema **s,
2064 : sql_table **t,
2065 : Client c, const char* schema, const char* table, size_t ccount, monetdbe_column* columns) {
2066 :
2067 0 : char* msg = MAL_SUCCEED;
2068 0 : char buf[16] = {0};
2069 0 : char* remote_program_name = number2name(buf, sizeof(buf), ++((backend*) c->sqlcontext)->remote);
2070 :
2071 0 : assert(s && t);
2072 0 : assert(c->sqlcontext && ((backend *) c->sqlcontext)->mvc);
2073 0 : mvc* m = ((backend *) c->sqlcontext)->mvc;
2074 :
2075 0 : Symbol _prg;
2076 0 : MalBlkPtr mb = NULL;
2077 0 : InstrPtr f = NULL, v = NULL, a = NULL, r = NULL;
2078 0 : int mvc_id = -1;
2079 :
2080 0 : if (!(*s = mvc_bind_schema(m, "tmp"))) {
2081 0 : return createException(MAL, "monetdbe.monetdbe_append", MAL_MALLOC_FAIL);
2082 : }
2083 :
2084 0 : switch (sql_trans_create_table(t, m->session->tr, *s, table, NULL, tt_table, false, SQL_DECLARED_TABLE, CA_COMMIT, -1, 0)) {
2085 0 : case -1:
2086 0 : return createException(SQL, "monetdbe.monetdbe_append", MAL_MALLOC_FAIL);
2087 0 : case -2:
2088 : case -3:
2089 0 : return createException(SQL, "monetdbe.monetdbe_append", "Table name '%s' conflicts", table);
2090 : default:
2091 0 : break;
2092 : }
2093 :
2094 0 : assert(prg);
2095 :
2096 0 : *prg = NULL;
2097 0 : _prg = newFunctionArgs(userRef, putName(remote_program_name), FUNCTIONsymbol, (int) ccount + 1); // remote program
2098 0 : mb = _prg->def;
2099 :
2100 0 : f = getInstrPtr(mb, 0);
2101 0 : f->retc = f->argc = 0;
2102 0 : f = pushReturn(mb, f, newTmpVariable(mb, TYPE_int));
2103 0 : v = newFcnCall(mb, sqlRef, mvcRef);
2104 0 : if (v == NULL) {
2105 0 : msg = createException(SQL, "monetdbe.monetdbe_append", MAL_MALLOC_FAIL);
2106 0 : goto cleanup;
2107 : }
2108 0 : pushInstruction(mb, v);
2109 0 : setArgType(mb, v, 0, TYPE_int);
2110 :
2111 0 : mvc_id = getArg(v, 0);
2112 :
2113 0 : sqlstore *store;
2114 0 : store = m->session->tr->store;
2115 0 : for (size_t i = 0; i < ccount; i++) {
2116 0 : sql_column *col = NULL;
2117 0 : sql_type *tpe = SA_ZNEW(m->sa, sql_type);
2118 0 : sql_subtype *st = SA_ZNEW(m->sa, sql_subtype);
2119 0 : if (tpe == NULL || st == NULL) {
2120 0 : msg = createException(SQL, "monetdbe.monetdbe_append", MAL_MALLOC_FAIL);
2121 0 : goto cleanup;
2122 : }
2123 0 : tpe->base.name = sa_strdup(m->sa, columns[i].name);
2124 0 : tpe->localtype = monetdbe_2_gdk_type((monetdbe_types) columns[i].type);
2125 0 : tpe->digits = columns[i].sql_type.digits;
2126 0 : tpe->scale = columns[i].sql_type.scale;
2127 0 : sql_init_subtype(st, tpe, columns[i].sql_type.digits, columns[i].sql_type.scale);
2128 :
2129 0 : switch (mvc_create_column(&col, m, *t, columns[i].name, st)) {
2130 0 : case -1:
2131 0 : msg = createException(SQL, "monetdbe.monetdbe_append", MAL_MALLOC_FAIL);
2132 0 : goto cleanup;
2133 0 : case -2:
2134 : case -3:
2135 0 : msg = createException(SQL, "monetdbe.monetdbe_append", "Column name '%s' conflicts", columns[i].name);
2136 0 : goto cleanup;
2137 : default:
2138 0 : break;
2139 : }
2140 :
2141 0 : if (store->storage_api.create_col(m->session->tr, col) != LOG_OK) {
2142 0 : msg = createException(MAL, "monetdbe.monetdbe_append", MAL_MALLOC_FAIL);
2143 0 : goto cleanup;
2144 : }
2145 :
2146 0 : int idx = newTmpVariable(mb, newBatType(tpe->localtype));
2147 0 : f = pushArgument(mb, f, idx);
2148 :
2149 0 : a = newFcnCall(mb, sqlRef, appendRef);
2150 0 : if (a == NULL) {
2151 0 : msg = createException(SQL, "monetdbe.monetdbe_append", MAL_MALLOC_FAIL);
2152 0 : goto cleanup;
2153 : }
2154 0 : setArgType(mb, a, 0, TYPE_int);
2155 0 : a = pushArgument(mb, a, mvc_id);
2156 0 : a = pushStr(mb, a, schema ? schema : "sys"); /* TODO this should be better */
2157 0 : a = pushStr(mb, a, table);
2158 0 : a = pushStr(mb, a, columns[i].name);
2159 0 : a = pushArgument(mb, a, idx);
2160 0 : pushInstruction(mb, a);
2161 :
2162 0 : mvc_id = getArg(a, 0);
2163 : }
2164 :
2165 0 : r = newInstruction(mb, NULL, NULL);
2166 0 : if (r == NULL) {
2167 0 : msg = createException(SQL, "monetdbe.monetdbe_append", MAL_MALLOC_FAIL);
2168 0 : goto cleanup;
2169 : }
2170 0 : r->barrier= RETURNsymbol;
2171 0 : r->retc = r->argc = 0;
2172 0 : r = pushReturn(mb, r, mvc_id);
2173 0 : r = pushArgument(mb, r, mvc_id);
2174 0 : pushInstruction(mb, r);
2175 :
2176 0 : pushEndInstruction(mb);
2177 :
2178 0 : if ( (msg = chkProgram(c->usermodule, mb)) != MAL_SUCCEED ) {
2179 0 : goto cleanup;
2180 : }
2181 :
2182 0 : assert(msg == MAL_SUCCEED);
2183 0 : *prg = _prg;
2184 0 : return msg;
2185 :
2186 : cleanup:
2187 0 : assert(msg != MAL_SUCCEED);
2188 0 : freeSymbol(_prg);
2189 0 : *prg = NULL;
2190 0 : return msg;
2191 : }
2192 :
2193 : char*
2194 0 : monetdbe_append(monetdbe_database dbhdl, const char *schema, const char *table, monetdbe_column **input, size_t column_count)
2195 : {
2196 0 : monetdbe_database_internal *mdbe = (monetdbe_database_internal*)dbhdl;
2197 0 : mvc *m = NULL;
2198 0 : sql_table *t = NULL;
2199 0 : size_t i, cnt;
2200 0 : node *n;
2201 0 : Symbol remote_prg = NULL;
2202 0 : BAT *pos = NULL;
2203 0 : BUN offset;
2204 :
2205 0 : if ((mdbe->msg = validate_database_handle(mdbe, "monetdbe.monetdbe_append")) != MAL_SUCCEED) {
2206 : return mdbe->msg;
2207 : }
2208 :
2209 0 : if ((mdbe->msg = getSQLContext(mdbe->c, NULL, &m, NULL)) != MAL_SUCCEED) {
2210 0 : mdbe->msg = commit_action(m, mdbe, NULL, NULL);
2211 0 : return mdbe->msg;
2212 : }
2213 0 : sqlstore *store = m->session->tr->store;
2214 :
2215 0 : if (table == NULL) {
2216 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_append", "table parameter is NULL"));
2217 0 : goto cleanup;
2218 : }
2219 0 : if (input == NULL) {
2220 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_append", "input parameter is NULL"));
2221 0 : goto cleanup;
2222 : }
2223 0 : if (column_count < 1) {
2224 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_append", "column_count must be higher than 0"));
2225 0 : goto cleanup;
2226 : }
2227 :
2228 0 : if (mdbe->mid) {
2229 : // We are going to insert the data into a temporary table which is used in the coming remote logic.
2230 :
2231 0 : size_t actual_column_count = 0;
2232 0 : monetdbe_column* columns = NULL;
2233 0 : sql_schema* s = NULL;
2234 :
2235 0 : if ((mdbe->msg = monetdbe_get_columns_remote(
2236 : mdbe,
2237 : schema,
2238 : table,
2239 : &actual_column_count,
2240 : &columns)) != MAL_SUCCEED) {
2241 0 : goto remote_cleanup;
2242 : }
2243 :
2244 0 : if ((mdbe->msg = SQLtrans(m)) != MAL_SUCCEED) {
2245 0 : goto remote_cleanup;
2246 : }
2247 :
2248 0 : if ((mdbe->msg = append_create_remote_append_mal_program
2249 : (&remote_prg,
2250 : &s,
2251 : &t,
2252 : mdbe->c,
2253 : schema,
2254 : table,
2255 : actual_column_count,
2256 : columns)) != MAL_SUCCEED) {
2257 0 : goto remote_cleanup;
2258 : }
2259 :
2260 0 : insertSymbol(mdbe->c->usermodule, remote_prg);
2261 :
2262 0 : remote_cleanup:
2263 0 : if (mdbe->msg) {
2264 0 : cleanup_get_columns_result(actual_column_count, columns);
2265 0 : freeSymbol(remote_prg);
2266 0 : goto cleanup;
2267 : }
2268 : } else {
2269 : // !mdbe->mid
2270 : // inserting into existing local table.
2271 0 : sql_part *pt = NULL;
2272 :
2273 0 : if ((mdbe->msg = SQLtrans(m)) != MAL_SUCCEED)
2274 0 : goto cleanup;
2275 0 : if (!(t = find_table_or_view_on_scope(m, NULL, schema, table, "CATALOG", false))) {
2276 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append", "%s", m->errstr + 6)); /* Skip error code */
2277 0 : goto cleanup;
2278 : }
2279 0 : if (!insert_allowed(m, t, t->base.name, "APPEND", "append")) {
2280 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append", "%s", m->errstr + 6)); /* Skip error code */
2281 0 : goto cleanup;
2282 : }
2283 0 : if ((t->s && t->s->parts && (pt = partition_find_part(m->session->tr, t, NULL))) || isRangePartitionTable(t) || isListPartitionTable(t)) {
2284 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append", "Appending to a table from a merge table hierarchy via 'monetdbe_append' is not possible at the moment"));
2285 0 : goto cleanup;
2286 : }
2287 0 : if (t->idxs) {
2288 0 : for (node *n = ol_first_node(t->idxs); n; n = n->next) {
2289 0 : sql_idx *i = n->data;
2290 :
2291 0 : if (i->key) {
2292 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append",
2293 : "Appending to a table with key constraints via 'monetdbe_append' is not possible at the moment"));
2294 0 : goto cleanup;
2295 0 : } else if (hash_index(i->type) && list_length(i->columns) > 1) {
2296 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append",
2297 : "Appending to a table with hash indexes referring to more than one column via 'monetdbe_append' is not possible at the moment"));
2298 0 : goto cleanup;
2299 0 : } else if (i->type == join_idx) {
2300 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append",
2301 : "Appending to a table with join indexes via 'monetdbe_append' is not possible at the moment"));
2302 0 : goto cleanup;
2303 : }
2304 : }
2305 : }
2306 0 : if (t->triggers) {
2307 0 : for (n = ol_first_node(t->triggers); n; n = n->next) {
2308 0 : sql_trigger *trigger = n->data;
2309 :
2310 0 : if (trigger->event == 0) { /* insert event */
2311 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append",
2312 : "Appending to a table with triggers at the insert event via 'monetdbe_append' is not possible at the moment"));
2313 0 : goto cleanup;
2314 : }
2315 : }
2316 : }
2317 : }
2318 :
2319 : /* for now no default values, ie user should supply all columns */
2320 0 : if (column_count != (size_t)ol_length(t->columns)) {
2321 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append", "Incorrect number of columns"));
2322 0 : goto cleanup;
2323 : }
2324 :
2325 0 : cnt = input[0]->count;
2326 0 : if (store->storage_api.claim_tab(m->session->tr, t, cnt, &offset, &pos) != LOG_OK) {
2327 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_append", "Claim failed"));
2328 0 : goto cleanup;
2329 : }
2330 : /* signal an insert was made on the table */
2331 0 : if (!isNew(t) && isGlobal(t) && !isGlobalTemp(t) && sql_trans_add_dependency_change(m->session->tr, t->base.id, dml) != LOG_OK) {
2332 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append", MAL_MALLOC_FAIL));
2333 0 : goto cleanup;
2334 : }
2335 :
2336 0 : for (i = 0, n = ol_first_node(t->columns); i < column_count && n; i++, n = n->next) {
2337 0 : sql_column *c = n->data;
2338 0 : int mtype = monetdbe_2_gdk_type(input[i]->type);
2339 0 : const void* nil = (mtype>=0)?ATOMnilptr(mtype):NULL;
2340 0 : char *v = input[i]->data;
2341 :
2342 0 : if (mtype < 0) {
2343 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append", "Cannot find type for column %zu", i));
2344 0 : goto cleanup;
2345 0 : } else if (input[i]->count != cnt) {
2346 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append", "Number of values don't match between columns"));
2347 0 : goto cleanup;
2348 : }
2349 0 : if (mtype >= TYPE_bit && mtype <=
2350 : #ifdef HAVE_HGE
2351 : TYPE_hge
2352 : #else
2353 : TYPE_lng
2354 : #endif
2355 : ) {
2356 : //-------------------------------------
2357 0 : BAT *bn = NULL;
2358 :
2359 0 : if (mtype != c->type.type->localtype) {
2360 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append", "Cannot append %d into column '%s'", input[i]->type, c->base.name));
2361 0 : goto cleanup;
2362 : }
2363 :
2364 0 : if ((bn = COLnew(0, mtype, 0, TRANSIENT)) == NULL) {
2365 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append", "Cannot create append column"));
2366 0 : goto cleanup;
2367 : }
2368 :
2369 : //save prev heap pointer
2370 0 : char *prev_base;
2371 0 : size_t prev_size;
2372 0 : prev_base = bn->theap->base;
2373 0 : prev_size = bn->theap->free;
2374 :
2375 : //BAT heap base to input[i]->data
2376 0 : bn->theap->base = input[i]->data;
2377 0 : bn->theap->free = tailsize(bn, cnt);
2378 :
2379 : //BATsetdims(bn); called in COLnew
2380 0 : BATsetcapacity(bn, cnt);
2381 0 : BATsetcount(bn, cnt);
2382 :
2383 : //set default flags
2384 0 : BATsettrivprop(bn);
2385 :
2386 0 : if (cnt > 1) {
2387 0 : bn->tsorted = bn->trevsorted = false;
2388 0 : bn->tnosorted = bn->tnorevsorted = 0;
2389 0 : bn->tkey = false;
2390 0 : bn->tnonil = false;
2391 0 : bn->tnil = false;
2392 : }
2393 :
2394 0 : if (store->storage_api.append_col(m->session->tr, c, offset, pos, bn, cnt, true, bn->ttype) != 0) {
2395 0 : bn->theap->base = prev_base;
2396 0 : bn->theap->free = prev_size;
2397 0 : BBPreclaim(bn);
2398 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append", "Cannot append BAT"));
2399 0 : goto cleanup;
2400 : }
2401 :
2402 0 : bn->theap->base = prev_base;
2403 0 : bn->theap->free = prev_size;
2404 0 : BBPreclaim(bn);
2405 : } else if (mtype == TYPE_str) {
2406 : char **d = (char**)v;
2407 :
2408 0 : for (size_t j=0; j<cnt; j++) {
2409 0 : if (!d[j]) {
2410 0 : d[j] = (char*) nil;
2411 0 : } else if (!checkUTF8(d[j])) {
2412 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append", "Incorrectly encoded UTF-8"));
2413 0 : goto cleanup;
2414 : }
2415 : }
2416 0 : if (store->storage_api.append_col(m->session->tr, c, offset, pos, d, cnt, false, mtype) != 0) {
2417 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append", "Cannot append values"));
2418 0 : goto cleanup;
2419 : }
2420 : } else if (mtype == TYPE_timestamp) {
2421 0 : int err = 0;
2422 0 : timestamp *d = GDKmalloc(sizeof(timestamp)*cnt);
2423 0 : if (!d) {
2424 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append", "Cannot append values"));
2425 0 : goto cleanup;
2426 : }
2427 : monetdbe_data_timestamp* ts = (monetdbe_data_timestamp*)v;
2428 :
2429 0 : for (size_t j=0; j<cnt; j++){
2430 0 : monetdbe_data_timestamp mdt = ts[j];
2431 :
2432 0 : if (timestamp_is_null(&mdt)) {
2433 0 : d[j] = *(timestamp*) nil;
2434 : } else {
2435 0 : d[j] = timestamp_from_data(&mdt);
2436 : }
2437 : }
2438 0 : if (store->storage_api.append_col(m->session->tr, c, offset, pos, d, cnt, false, mtype) != 0) {
2439 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append", "Cannot append values"));
2440 0 : err = 1;
2441 : }
2442 0 : GDKfree(d);
2443 0 : if (err)
2444 0 : goto cleanup;
2445 : } else if (mtype == TYPE_date) {
2446 0 : int err = 0;
2447 0 : date *d = GDKmalloc(sizeof(date)*cnt);
2448 0 : if (!d) {
2449 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append", "Cannot append values"));
2450 0 : goto cleanup;
2451 : }
2452 : monetdbe_data_date* de = (monetdbe_data_date*)v;
2453 :
2454 0 : for (size_t j=0; j<cnt; j++){
2455 0 : monetdbe_data_date mdt = de[j];
2456 :
2457 0 : if (date_is_null(&mdt)) {
2458 0 : d[j] = *(date*) nil;
2459 : } else {
2460 0 : d[j] = date_from_data(&mdt);
2461 : }
2462 : }
2463 0 : if (store->storage_api.append_col(m->session->tr, c, offset, pos, d, cnt, false, mtype) != 0) {
2464 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append", "Cannot append values"));
2465 0 : err = 1;
2466 : }
2467 0 : GDKfree(d);
2468 0 : if (err)
2469 0 : goto cleanup;
2470 : } else if (mtype == TYPE_daytime) {
2471 0 : int err = 0;
2472 0 : daytime *d = GDKmalloc(sizeof(daytime)*cnt);
2473 0 : if (!d) {
2474 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append", "Cannot append values"));
2475 0 : goto cleanup;
2476 : }
2477 : monetdbe_data_time* t = (monetdbe_data_time*)v;
2478 :
2479 0 : for (size_t j=0; j<cnt; j++){
2480 0 : monetdbe_data_time mdt = t[j];
2481 :
2482 0 : if (time_is_null(&mdt)) {
2483 0 : d[j] = *(daytime*) nil;
2484 : } else {
2485 0 : d[j] = time_from_data(&mdt);
2486 : }
2487 : }
2488 0 : if (store->storage_api.append_col(m->session->tr, c, offset, pos, d, cnt, false, mtype) != 0) {
2489 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append", "Cannot append values"));
2490 0 : err = 1;
2491 : }
2492 0 : GDKfree(d);
2493 0 : if (err)
2494 0 : goto cleanup;
2495 0 : } else if (mtype == TYPE_blob) {
2496 0 : int err = 0;
2497 0 : size_t j = 0;
2498 0 : blob **d = GDKmalloc(sizeof(blob*)*cnt);
2499 0 : if (!d) {
2500 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append", "Cannot append values"));
2501 0 : goto cleanup;
2502 : }
2503 : monetdbe_data_blob* be = (monetdbe_data_blob*)v;
2504 :
2505 0 : for (j=0; j<cnt; j++){
2506 0 : if (blob_is_null(be+j)) {
2507 0 : d[j] = (blob*)nil;
2508 : } else {
2509 0 : size_t len = be[j].size;
2510 0 : size_t nlen = blobsize(len);
2511 0 : blob *b = (blob*)GDKmalloc(nlen);
2512 0 : if (!b) {
2513 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_append", MAL_MALLOC_FAIL));
2514 0 : err = 1;
2515 0 : break;
2516 : }
2517 0 : b->nitems = len;
2518 0 : memcpy(b->data, be[j].data, len);
2519 0 : d[j] = b;
2520 : }
2521 : }
2522 0 : if (!err && store->storage_api.append_col(m->session->tr, c, offset, pos, d, cnt, false, mtype) != 0) {
2523 0 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append", "Cannot append values"));
2524 0 : err = 1;
2525 : }
2526 0 : for (size_t k=0; k<j; k++){
2527 0 : if (d[k] != nil)
2528 0 : GDKfree(d[k]);
2529 : }
2530 0 : GDKfree(d);
2531 0 : if (err)
2532 0 : goto cleanup;
2533 : } else {
2534 : set_error(mdbe, createException(SQL, "monetdbe.monetdbe_append", "The internal type '%s' is not supported on monetdbe append at the moment", ATOMname(mtype)));
2535 : goto cleanup;
2536 : }
2537 : }
2538 :
2539 0 : if (mdbe->mid) {
2540 0 : char nme[16];
2541 0 : const char *name = number2name(nme, sizeof(nme), ++((backend*) mdbe->c->sqlcontext)->remote);
2542 0 : Symbol prg; // local program
2543 :
2544 0 : if ( (prg = newFunction(userRef, putName(name), FUNCTIONsymbol)) == NULL ) {
2545 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_append", MAL_MALLOC_FAIL));
2546 0 : goto cleanup;
2547 : }
2548 :
2549 0 : MalBlkPtr mb = prg->def;
2550 0 : InstrPtr f = getInstrPtr(mb, 0);
2551 0 : f->retc = f->argc = 0;
2552 :
2553 0 : InstrPtr r = newFcnCall(mb, remoteRef, registerRef);
2554 0 : if (r == NULL) {
2555 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_append", MAL_MALLOC_FAIL));
2556 0 : freeSymbol(prg);
2557 0 : goto cleanup;
2558 : }
2559 :
2560 0 : setArgType(mb, r, 0, TYPE_str);
2561 0 : r = pushStr(mb, r, mdbe->mid);
2562 0 : r = pushStr(mb, r, userRef);
2563 0 : r = pushStr(mb, r, putName(remote_prg->name));
2564 0 : pushInstruction(mb, r);
2565 :
2566 0 : InstrPtr e = newInstructionArgs(mb, remoteRef, execRef, 4 + ol_length(t->columns));
2567 0 : if (e == NULL) {
2568 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_append", MAL_MALLOC_FAIL));
2569 0 : freeSymbol(prg);
2570 0 : goto cleanup;
2571 : }
2572 0 : setDestVar(e, newTmpVariable(mb, TYPE_any));
2573 0 : e = pushStr(mb, e, mdbe->mid);
2574 0 : e = pushStr(mb, e, userRef);
2575 0 : e = pushArgument(mb, e, getArg(r, 0));
2576 :
2577 0 : for (i = 0, n = ol_first_node(t->columns); i < (unsigned) ol_length(t->columns); i++, n = n->next) {
2578 0 : sql_column *c = n->data;
2579 0 : BAT* b = store->storage_api.bind_col(m->session->tr, c, RDONLY);
2580 0 : if (b == NULL) {
2581 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_append", MAL_MALLOC_FAIL));
2582 0 : freeSymbol(prg);
2583 0 : goto cleanup;
2584 : }
2585 :
2586 0 : int idx = newTmpVariable(mb, newBatType(c->type.type->localtype));
2587 0 : ValRecord v = { .bat = true, .vtype = b->ttype, .len = sizeof(int), .val.bval = b->batCacheid};
2588 0 : getVarConstant(mb, idx) = v;
2589 0 : setVarConstant(mb, idx);
2590 0 : BBPunfix(b->batCacheid);
2591 :
2592 0 : InstrPtr p = newFcnCall(mb, remoteRef, putRef);
2593 0 : if (p == NULL) {
2594 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_append", MAL_MALLOC_FAIL));
2595 0 : freeInstruction(e);
2596 0 : freeSymbol(prg);
2597 0 : goto cleanup;
2598 : }
2599 0 : setArgType(mb, p, 0, TYPE_str);
2600 0 : p = pushStr(mb, p, mdbe->mid);
2601 0 : p = pushArgument(mb, p, idx);
2602 0 : pushInstruction(mb, p);
2603 :
2604 0 : e = pushArgument(mb, e, getArg(p, 0));
2605 : }
2606 :
2607 0 : pushInstruction(mb, e);
2608 :
2609 0 : InstrPtr ri = newInstruction(mb, NULL, NULL);
2610 0 : if (ri == NULL) {
2611 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_append", MAL_MALLOC_FAIL));
2612 0 : freeSymbol(prg);
2613 0 : goto cleanup;
2614 : }
2615 0 : ri->barrier= RETURNsymbol;
2616 0 : ri->retc = ri->argc = 0;
2617 0 : pushInstruction(mb, ri);
2618 :
2619 0 : if ( (mdbe->msg = chkProgram(mdbe->c->usermodule, mb)) != MAL_SUCCEED ) {
2620 0 : freeSymbol(prg);
2621 0 : goto cleanup;
2622 : }
2623 :
2624 0 : mdbe->msg = runMAL(mdbe->c, mb, 0, NULL);
2625 0 : freeSymbol(prg);
2626 : }
2627 :
2628 0 : cleanup:
2629 0 : if (pos)
2630 0 : BBPreclaim(pos);
2631 0 : mdbe->msg = commit_action(m, mdbe, NULL, NULL);
2632 0 : return mdbe->msg;
2633 : }
2634 :
2635 : const void *
2636 0 : monetdbe_null(monetdbe_database dbhdl, monetdbe_types t)
2637 : {
2638 0 : monetdbe_database_internal *mdbe = (monetdbe_database_internal*)dbhdl;
2639 0 : int mtype = monetdbe_2_gdk_type(t);
2640 :
2641 0 : if (mtype < 0)
2642 : return NULL;
2643 :
2644 0 : if ((mtype >= TYPE_bit && mtype <=
2645 : #ifdef HAVE_HGE
2646 : TYPE_hge
2647 : #else
2648 : TYPE_lng
2649 : #endif
2650 : ))
2651 0 : return ATOMnilptr(mtype);
2652 : else if (mtype == TYPE_str)
2653 : return NULL;
2654 : else if (mtype == TYPE_blob)
2655 0 : return &mdbe->blob_null;
2656 : else if (mtype == TYPE_date)
2657 0 : return &mdbe->date_null;
2658 : else if (mtype == TYPE_daytime)
2659 0 : return &mdbe->time_null;
2660 : else if (mtype == TYPE_timestamp)
2661 0 : return &mdbe->timestamp_null;
2662 : return NULL;
2663 : }
2664 :
2665 : char*
2666 4 : monetdbe_result_fetch(monetdbe_result* mres, monetdbe_column** res, size_t column_index)
2667 : {
2668 4 : BAT* b = NULL;
2669 4 : int bat_type;
2670 4 : mvc* m;
2671 4 : monetdbe_result_internal* result = (monetdbe_result_internal*) mres;
2672 4 : sql_subtype* sqltpe = NULL;
2673 4 : monetdbe_column* column_result = NULL;
2674 4 : size_t j = 0;
2675 4 : monetdbe_database_internal *mdbe = result->mdbe;
2676 4 : Client c = mdbe->c;
2677 :
2678 :
2679 4 : if ((mdbe->msg = validate_database_handle(mdbe, "monetdbe.monetdbe_result_fetch")) != MAL_SUCCEED) {
2680 :
2681 : return mdbe->msg;
2682 : }
2683 :
2684 4 : if ((mdbe->msg = getSQLContext(c, NULL, &m, NULL)) != MAL_SUCCEED)
2685 0 : goto cleanup;
2686 4 : if (!res) {
2687 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_result_fetch", "Parameter res is NULL"));
2688 0 : goto cleanup;
2689 : }
2690 4 : if (column_index >= mres->ncols) {
2691 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_result_fetch", "Index out of range"));
2692 0 : goto cleanup;
2693 : }
2694 : // check if we have the column converted already
2695 4 : if (result->converted_columns[column_index]) {
2696 2 : *res = result->converted_columns[column_index];
2697 :
2698 2 : return MAL_SUCCEED;
2699 : }
2700 :
2701 : // otherwise we have to convert the column
2702 2 : b = BATdescriptor(result->monetdbe_resultset->cols[column_index].b);
2703 2 : if (!b) {
2704 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_result_fetch", SQLSTATE(HY002) RUNTIME_OBJECT_MISSING));
2705 0 : goto cleanup;
2706 : }
2707 2 : bat_type = b->ttype;
2708 2 : sqltpe = &result->monetdbe_resultset->cols[column_index].type;
2709 :
2710 2 : if (bat_type == TYPE_bit) {
2711 0 : GENERATE_BAT_INPUT(b, int8_t, bool, bit);
2712 : } else if (bat_type == TYPE_bte) {
2713 0 : GENERATE_BAT_INPUT(b, int8_t, int8_t, bte);
2714 : } else if (bat_type == TYPE_sht) {
2715 0 : GENERATE_BAT_INPUT(b, int16_t, int16_t, sht);
2716 : } else if (bat_type == TYPE_int) {
2717 3 : GENERATE_BAT_INPUT(b, int32_t, int32_t, int);
2718 : } else if (bat_type == TYPE_oid) {
2719 0 : GENERATE_BAT_INPUT(b, size_t, size_t, oid);
2720 : } else if (bat_type == TYPE_lng) {
2721 0 : GENERATE_BAT_INPUT(b, int64_t, int64_t, lng);
2722 : #ifdef HAVE_HGE
2723 : } else if (bat_type == TYPE_hge) {
2724 0 : GENERATE_BAT_INPUT(b, __int128, int128_t, hge);
2725 : #endif
2726 : } else if (bat_type == TYPE_flt) {
2727 0 : GENERATE_BAT_INPUT(b, float, float, flt);
2728 : } else if (bat_type == TYPE_dbl) {
2729 0 : GENERATE_BAT_INPUT(b, double, double, dbl);
2730 : } else if (bat_type == TYPE_str) {
2731 1 : BATiter li;
2732 1 : BUN p = 0, q = 0;
2733 1 : GENERATE_BAT_INPUT_BASE(str);
2734 1 : bat_data->count = (size_t) mres->nrows;
2735 1 : if (bat_data->count) {
2736 1 : bat_data->data = GDKzalloc(sizeof(char *) * bat_data->count);
2737 1 : bat_data->null_value = NULL;
2738 1 : if (!bat_data->data) {
2739 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_result_fetch", MAL_MALLOC_FAIL));
2740 0 : goto cleanup;
2741 : }
2742 : }
2743 :
2744 1 : j = 0;
2745 1 : li = bat_iterator(b);
2746 3 : BATloop(b, p, q)
2747 : {
2748 2 : const char *t = (const char*)BUNtvar(li, p);
2749 2 : if (strcmp(t, str_nil) == 0) {
2750 0 : bat_data->data[j] = NULL;
2751 : } else {
2752 2 : bat_data->data[j] = GDKstrdup(t);
2753 2 : if (!bat_data->data[j]) {
2754 0 : bat_iterator_end(&li);
2755 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_result_fetch", MAL_MALLOC_FAIL));
2756 0 : goto cleanup;
2757 : }
2758 : }
2759 2 : j++;
2760 : }
2761 1 : bat_iterator_end(&li);
2762 : } else if (bat_type == TYPE_date) {
2763 0 : date *baseptr;
2764 0 : GENERATE_BAT_INPUT_BASE(date);
2765 0 : bat_data->count = (size_t) mres->nrows;
2766 0 : if (bat_data->count) {
2767 0 : bat_data->data = GDKmalloc(sizeof(bat_data->null_value) * bat_data->count);
2768 0 : if (!bat_data->data) {
2769 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_result_fetch", MAL_MALLOC_FAIL));
2770 0 : goto cleanup;
2771 : }
2772 : }
2773 :
2774 0 : baseptr = (date *)Tloc(b, 0);
2775 0 : for (j = 0; j < bat_data->count; j++)
2776 0 : data_from_date(baseptr[j], bat_data->data + j);
2777 0 : memcpy(&bat_data->null_value, &mdbe->date_null, sizeof(monetdbe_data_date));
2778 : } else if (bat_type == TYPE_daytime) {
2779 0 : daytime *baseptr;
2780 0 : GENERATE_BAT_INPUT_BASE(time);
2781 0 : bat_data->count = (size_t) mres->nrows;
2782 0 : if (bat_data->count) {
2783 0 : bat_data->data = GDKmalloc(sizeof(bat_data->null_value) * bat_data->count);
2784 0 : if (!bat_data->data) {
2785 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_result_fetch", MAL_MALLOC_FAIL));
2786 0 : goto cleanup;
2787 : }
2788 : }
2789 :
2790 0 : baseptr = (daytime *)Tloc(b, 0);
2791 0 : for (j = 0; j < bat_data->count; j++)
2792 0 : data_from_time(baseptr[j], bat_data->data + j);
2793 0 : memcpy(&bat_data->null_value, &mdbe->time_null, sizeof(monetdbe_data_time));
2794 : } else if (bat_type == TYPE_timestamp) {
2795 0 : timestamp *baseptr;
2796 0 : GENERATE_BAT_INPUT_BASE(timestamp);
2797 0 : bat_data->count = (size_t) mres->nrows;
2798 0 : if (bat_data->count) {
2799 0 : bat_data->data = GDKmalloc(sizeof(bat_data->null_value) * bat_data->count);
2800 0 : if (!bat_data->data) {
2801 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_result_fetch", MAL_MALLOC_FAIL));
2802 0 : goto cleanup;
2803 : }
2804 : }
2805 :
2806 0 : baseptr = (timestamp *)Tloc(b, 0);
2807 0 : for (j = 0; j < bat_data->count; j++)
2808 0 : data_from_timestamp(baseptr[j], bat_data->data + j);
2809 0 : memcpy(&bat_data->null_value, &mdbe->timestamp_null, sizeof(monetdbe_data_timestamp));
2810 : } else if (bat_type == TYPE_blob) {
2811 0 : BATiter li;
2812 0 : BUN p = 0, q = 0;
2813 0 : GENERATE_BAT_INPUT_BASE(blob);
2814 0 : bat_data->count = (size_t) mres->nrows;
2815 0 : if (bat_data->count) {
2816 0 : bat_data->data = GDKmalloc(sizeof(monetdbe_data_blob) * bat_data->count);
2817 0 : if (!bat_data->data) {
2818 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_result_fetch", MAL_MALLOC_FAIL));
2819 0 : goto cleanup;
2820 : }
2821 : }
2822 0 : j = 0;
2823 :
2824 0 : li = bat_iterator(b);
2825 0 : BATloop(b, p, q)
2826 : {
2827 0 : const blob *t = (const blob *)BUNtvar(li, p);
2828 0 : if (t->nitems == ~(size_t)0) {
2829 0 : bat_data->data[j].size = 0;
2830 0 : bat_data->data[j].data = NULL;
2831 : } else {
2832 0 : bat_data->data[j].size = t->nitems;
2833 0 : bat_data->data[j].data = GDKmalloc(t->nitems);
2834 0 : if (!bat_data->data[j].data) {
2835 0 : bat_iterator_end(&li);
2836 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_result_fetch", MAL_MALLOC_FAIL));
2837 0 : goto cleanup;
2838 : }
2839 0 : memcpy(bat_data->data[j].data, t->data, t->nitems);
2840 : }
2841 0 : j++;
2842 : }
2843 0 : bat_iterator_end(&li);
2844 0 : bat_data->null_value.size = 0;
2845 0 : bat_data->null_value.data = NULL;
2846 : } else {
2847 : // unsupported type: convert to string
2848 0 : BATiter li;
2849 0 : BUN p = 0, q = 0;
2850 0 : GENERATE_BAT_INPUT_BASE(str);
2851 0 : bat_data->count = (size_t) mres->nrows;
2852 0 : if (bat_data->count) {
2853 0 : bat_data->null_value = NULL;
2854 0 : bat_data->data = GDKzalloc(sizeof(char *) * bat_data->count);
2855 0 : if (!bat_data->data) {
2856 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_result_fetch", MAL_MALLOC_FAIL));
2857 0 : goto cleanup;
2858 : }
2859 : }
2860 0 : j = 0;
2861 :
2862 0 : li = bat_iterator(b);
2863 0 : BATloop(b, p, q)
2864 : {
2865 0 : const void *t = BUNtail(li, p);
2866 0 : if (BATatoms[bat_type].atomCmp(t, BATatoms[bat_type].atomNull) == 0) {
2867 0 : bat_data->data[j] = NULL;
2868 : } else {
2869 0 : char *sresult = NULL;
2870 0 : size_t length = 0;
2871 0 : if (BATatoms[bat_type].atomToStr(&sresult, &length, t, true) == 0) {
2872 0 : bat_iterator_end(&li);
2873 0 : set_error(mdbe, createException(MAL, "monetdbe.monetdbe_result_fetch", "Failed to convert element to string"));
2874 0 : goto cleanup;
2875 : }
2876 0 : bat_data->data[j] = sresult;
2877 : }
2878 0 : j++;
2879 : }
2880 0 : bat_iterator_end(&li);
2881 : }
2882 0 : if (column_result)
2883 2 : column_result->name = result->monetdbe_resultset->cols[column_index].name;
2884 2 : cleanup:
2885 2 : BBPreclaim(b);
2886 2 : if (mdbe->msg) {
2887 0 : if (res)
2888 0 : *res = NULL;
2889 0 : monetdbe_destroy_column(column_result);
2890 2 : } else if (res) {
2891 2 : result->converted_columns[column_index] = column_result;
2892 2 : *res = result->converted_columns[column_index];
2893 : }
2894 2 : mdbe->msg = commit_action(m, mdbe, NULL, NULL);
2895 :
2896 2 : return mdbe->msg;
2897 : }
2898 :
2899 : static void
2900 1 : data_from_date(date d, monetdbe_data_date *ptr)
2901 : {
2902 1 : ptr->day = date_day(d);
2903 1 : ptr->month = date_month(d);
2904 1 : ptr->year = date_year(d);
2905 1 : }
2906 :
2907 : static date
2908 0 : date_from_data(monetdbe_data_date *ptr)
2909 : {
2910 0 : return date_create(ptr->year, ptr->month, ptr->day);
2911 : }
2912 :
2913 : static void
2914 1 : data_from_time(daytime d, monetdbe_data_time *ptr)
2915 : {
2916 1 : ptr->hours = daytime_hour(d);
2917 1 : ptr->minutes = daytime_min(d);
2918 1 : ptr->seconds = daytime_sec(d);
2919 1 : ptr->ms = daytime_usec(d) / 1000;
2920 1 : }
2921 :
2922 : static daytime
2923 0 : time_from_data(monetdbe_data_time *ptr)
2924 : {
2925 0 : return daytime_create(ptr->hours, ptr->minutes, ptr->seconds, ptr->ms * 1000);
2926 : }
2927 :
2928 : static void
2929 1 : data_from_timestamp(timestamp d, monetdbe_data_timestamp *ptr)
2930 : {
2931 1 : daytime tm = timestamp_daytime(d);
2932 1 : date dt = timestamp_date(d);
2933 :
2934 1 : ptr->date.day = date_day(dt);
2935 1 : ptr->date.month = date_month(dt);
2936 1 : ptr->date.year = date_year(dt);
2937 1 : ptr->time.hours = daytime_hour(tm);
2938 1 : ptr->time.minutes = daytime_min(tm);
2939 1 : ptr->time.seconds = daytime_sec(tm);
2940 1 : ptr->time.ms = daytime_usec(tm) / 1000;
2941 1 : }
2942 :
2943 : static timestamp
2944 0 : timestamp_from_data(monetdbe_data_timestamp *ptr)
2945 : {
2946 0 : return timestamp_create(
2947 0 : date_create(ptr->date.year, ptr->date.month, ptr->date.day),
2948 0 : daytime_create(ptr->time.hours, ptr->time.minutes, ptr->time.seconds, ptr->time.ms * 1000));
2949 : }
2950 :
2951 : const char*
2952 0 : monetdbe_get_mapi_port(void) {
2953 0 : return GDKgetenv("mapi_port");
2954 : }
|
