LCOV - code coverage report
Current view: top level - monetdb5/modules/mal - remote.c (source / functions) Hit Total Coverage
Test: coverage.info Lines: 424 771 55.0 %
Date: 2021-10-13 02:24:04 Functions: 19 25 76.0 %

          Line data    Source code
       1             : /*
       2             :  * This Source Code Form is subject to the terms of the Mozilla Public
       3             :  * License, v. 2.0.  If a copy of the MPL was not distributed with this
       4             :  * file, You can obtain one at http://mozilla.org/MPL/2.0/.
       5             :  *
       6             :  * Copyright 1997 - July 2008 CWI, August 2008 - 2021 MonetDB B.V.
       7             :  */
       8             : 
       9             : /*
      10             :  * (c) Fabian Groffen, Martin Kersten
      11             :  * Remote querying functionality
      12             :  * Communication with other mservers at the MAL level is a delicate task.
      13             :  * However, it is indispensable for any distributed functionality.  This
      14             :  * module provides an abstract way to store and retrieve objects on a
      15             :  * remote site.  Additionally, functions on a remote site can be executed
      16             :  * using objects available in the remote session context.  This yields in
      17             :  * four primitive functions that form the basis for distribution methods:
      18             :  * get, put, register and exec.
      19             :  *
      20             :  * The get method simply retrieves a copy of a remote object.  Objects can
      21             :  * be simple values, strings or Column.  The same holds for the put method,
      22             :  * but the other way around.  A local object can be stored on a remote
      23             :  * site.  Upon a successful store, the put method returns the remote
      24             :  * identifier for the stored object.  With this identifier the object can
      25             :  * be addressed, e.g. using the get method to retrieve the object that was
      26             :  * stored using put.
      27             :  *
      28             :  * The get and put methods are symmetric.  Performing a get on an
      29             :  * identifier that was returned by put, results in an object with the same
      30             :  * value and type as the one that was put.  The result of such an operation is
      31             :  * equivalent to making an (expensive) copy of the original object.
      32             :  *
      33             :  * The register function takes a local MAL function and makes it known at a
      34             :  * remote site. It ensures that it does not overload an already known
      35             :  * operation remotely, which could create a semantic conflict.
      36             :  * Deregistering a function is forbidden, because it would allow for taking
      37             :  * over the remote site completely.
      38             :  * C-implemented functions, such as io.print() cannot be remotely stored.
      39             :  * It would require even more complicated (byte) code shipping and remote
      40             :  * compilation to make it work.
      41             :  *
      42             :  * The choice to let exec only execute functions avoids problems
      43             :  * to decide what should be returned to the caller.  With a function it is
      44             :  * clear and simple to return that what the function signature prescribes.
      45             :  * Any side effect (e.g. io.print calls) may cause havoc in the system,
      46             :  * but are currently ignored.
      47             :  *
      48             :  * This leads to the final contract of this module.  The methods should be
      49             :  * used correctly, by obeying their contract.  Failing to do so will result
      50             :  * in errors and possibly undefined behaviour.
      51             :  *
      52             :  * The resolve() function can be used to query Merovingian.  It returns one
      53             :  * or more databases discovered in its vicinity matching the given pattern.
      54             :  *
      55             :  */
      56             : #include "monetdb_config.h"
      57             : #include "remote.h"
      58             : 
      59             : /*
      60             :  * Technically, these methods need to be serialised per connection,
      61             :  * hence a scheduler that interleaves e.g. multiple get calls, simply
      62             :  * violates this constraint.  If parallelism to the same site is
      63             :  * desired, a user could create a second connection.  This is not always
      64             :  * easy to generate at the proper place, e.g. overloading the dataflow
      65             :  * optimizer to patch connections structures is not acceptable.
      66             :  *
      67             :  * Instead, we maintain a simple lock with each connection, which can be
      68             :  * used to issue a safe, but blocking get/put/exec/register request.
      69             :  */
      70             : #ifdef HAVE_MAPI
      71             : 
      72             : 
      73             : #include "mal_exception.h"
      74             : #include "mal_interpreter.h"
      75             : #include "mal_function.h" /* for printFunction */
      76             : #include "mal_listing.h"
      77             : #include "mal_instruction.h" /* for getmodule/func macros */
      78             : #include "mal_authorize.h"
      79             : #include "mapi.h"
      80             : #include "mutils.h"
      81             : 
      82             : #define RMTT_L_ENDIAN   (0<<1)
      83             : #define RMTT_B_ENDIAN   (1<<1)
      84             : #define RMTT_32_BITS    (0<<2)
      85             : #define RMTT_64_BITS    (1<<2)
      86             : #define RMTT_32_OIDS    (0<<3)
      87             : #define RMTT_64_OIDS    (1<<3)
      88             : 
      89             : typedef struct _connection {
      90             :         MT_Lock            lock;      /* lock to avoid interference */
      91             :         str                name;      /* the handle for this connection */
      92             :         Mapi               mconn;     /* the Mapi handle for the connection */
      93             :         unsigned char      type;      /* binary profile of the connection target */
      94             :         size_t             nextid;    /* id counter */
      95             :         struct _connection *next;     /* the next connection in the list */
      96             : } *connection;
      97             : 
      98             : #ifndef WIN32
      99             : #include <sys/socket.h> /* socket */
     100             : #include <sys/un.h> /* sockaddr_un */
     101             : #endif
     102             : #include <unistd.h> /* gethostname */
     103             : 
     104             : static connection conns = NULL;
     105             : static unsigned char localtype = 0177;
     106             : 
     107             : static inline str RMTquery(MapiHdl *ret, const char *func, Mapi conn, const char *query);
     108             : 
     109             : /**
     110             :  * Returns a BAT with valid redirects for the given pattern.  If
     111             :  * merovingian is not running, this function throws an error.
     112             :  */
     113           0 : static str RMTresolve(bat *ret, str *pat) {
     114             : #ifdef NATIVE_WIN32
     115             :         (void) ret;
     116             :         (void) pat;
     117             :         throw(MAL, "remote.resolve", "merovingian is not available on "
     118             :                         "your platform, sorry"); /* please upgrade to Linux, etc. */
     119             : #else
     120             :         BAT *list;
     121             :         const char *mero_uri;
     122             :         char *p;
     123             :         unsigned int port;
     124             :         char **redirs;
     125             :         char **or;
     126             : 
     127           0 :         if (pat == NULL || *pat == NULL || strcmp(*pat, (str)str_nil) == 0)
     128           0 :                 throw(ILLARG, "remote.resolve",
     129             :                                 ILLEGAL_ARGUMENT ": pattern is NULL or nil");
     130             : 
     131           0 :         mero_uri = GDKgetenv("merovingian_uri");
     132           0 :         if (mero_uri == NULL)
     133           0 :                 throw(MAL, "remote.resolve", "this function needs the mserver "
     134             :                                 "have been started by merovingian");
     135             : 
     136           0 :         list = COLnew(0, TYPE_str, 0, TRANSIENT);
     137           0 :         if (list == NULL)
     138           0 :                 throw(MAL, "remote.resolve", SQLSTATE(HY013) MAL_MALLOC_FAIL);
     139             : 
     140             :         /* extract port from mero_uri, let mapi figure out the rest */
     141           0 :         mero_uri+=strlen("mapi:monetdb://");
     142           0 :         if (*mero_uri == '[') {
     143           0 :                 if ((mero_uri = strchr(mero_uri, ']')) == NULL) {
     144           0 :                         BBPreclaim(list);
     145           0 :                         throw(MAL, "remote.resolve", "illegal IPv6 address on merovingian_uri: %s",
     146             :                                   GDKgetenv("merovingian_uri"));
     147             :                 }
     148             :         }
     149           0 :         if ((p = strchr(mero_uri, ':')) == NULL) {
     150           0 :                 BBPreclaim(list);
     151           0 :                 throw(MAL, "remote.resolve", "illegal merovingian_uri setting: %s",
     152             :                                 GDKgetenv("merovingian_uri"));
     153             :         }
     154           0 :         port = (unsigned int)atoi(p + 1);
     155             : 
     156           0 :         or = redirs = mapi_resolve(NULL, port, *pat);
     157             : 
     158           0 :         if (redirs == NULL) {
     159           0 :                 BBPreclaim(list);
     160           0 :                 throw(MAL, "remote.resolve", "unknown failure when resolving pattern");
     161             :         }
     162             : 
     163           0 :         while (*redirs != NULL) {
     164           0 :                 if (BUNappend(list, (ptr)*redirs, false) != GDK_SUCCEED) {
     165           0 :                         BBPreclaim(list);
     166             :                         do
     167           0 :                                 free(*redirs);
     168           0 :                         while (*++redirs);
     169           0 :                         free(or);
     170           0 :                         throw(MAL, "remote.resolve", SQLSTATE(HY013) MAL_MALLOC_FAIL);
     171             :                 }
     172           0 :                 free(*redirs);
     173           0 :                 redirs++;
     174             :         }
     175           0 :         free(or);
     176             : 
     177           0 :         BBPkeepref(*ret = list->batCacheid);
     178           0 :         return(MAL_SUCCEED);
     179             : #endif
     180             : }
     181             : 
     182             : 
     183             : /* for unique connection identifiers */
     184             : static size_t connection_id = 0;
     185             : 
     186             : /**
     187             :  * Returns a connection to the given uri.  It always returns a newly
     188             :  * created connection.
     189             :  */
     190         130 : static str RMTconnectScen(
     191             :                 str *ret,
     192             :                 str *ouri,
     193             :                 str *user,
     194             :                 str *passwd,
     195             :                 str *scen,
     196             :                 bit *columnar)
     197             : {
     198             :         connection c;
     199             :         char conn[BUFSIZ];
     200             :         char *s;
     201             :         Mapi m;
     202             :         MapiHdl hdl;
     203             :         str msg;
     204             : 
     205             :         /* just make sure the return isn't garbage */
     206         130 :         *ret = 0;
     207             : 
     208         130 :         if (ouri == NULL || *ouri == NULL || strcmp(*ouri, (str)str_nil) == 0)
     209           0 :                 throw(ILLARG, "remote.connect", ILLEGAL_ARGUMENT ": database uri "
     210             :                                 "is NULL or nil");
     211         130 :         if (user == NULL || *user == NULL || strcmp(*user, (str)str_nil) == 0)
     212           0 :                 throw(ILLARG, "remote.connect", ILLEGAL_ARGUMENT ": username is "
     213             :                                 "NULL or nil");
     214         130 :         if (passwd == NULL || *passwd == NULL || strcmp(*passwd, (str)str_nil) == 0)
     215           0 :                 throw(ILLARG, "remote.connect", ILLEGAL_ARGUMENT ": password is "
     216             :                                 "NULL or nil");
     217         130 :         if (scen == NULL || *scen == NULL || strcmp(*scen, (str)str_nil) == 0)
     218           0 :                 throw(ILLARG, "remote.connect", ILLEGAL_ARGUMENT ": scenario is "
     219             :                                 "NULL or nil");
     220         130 :         if (strcmp(*scen, "mal") != 0 && strcmp(*scen, "msql") != 0)
     221           0 :                 throw(ILLARG, "remote.connect", ILLEGAL_ARGUMENT ": scenario '%s' "
     222             :                                 "is not supported", *scen);
     223             : 
     224         130 :         m = mapi_mapiuri(*ouri, *user, *passwd, *scen);
     225         130 :         if (mapi_error(m)) {
     226           0 :                 msg = createException(MAL, "remote.connect",
     227             :                                                           "unable to connect to '%s': %s",
     228             :                                                           *ouri, mapi_error_str(m));
     229           0 :                 mapi_destroy(m);
     230           0 :                 return msg;
     231             :         }
     232             : 
     233         130 :         MT_lock_set(&mal_remoteLock);
     234             : 
     235             :         /* generate an unique connection name, they are only known
     236             :          * within one mserver, id is primary key, the rest is super key */
     237         130 :         snprintf(conn, BUFSIZ, "%s_%s_%zu", mapi_get_dbname(m), *user, connection_id++);
     238             :         /* make sure we can construct MAL identifiers using conn */
     239        3424 :         for (s = conn; *s != '\0'; s++) {
     240        3294 :                 if (!isalnum((unsigned char)*s)) {
     241         474 :                         *s = '_';
     242             :                 }
     243             :         }
     244             : 
     245         130 :         if (mapi_reconnect(m) != MOK) {
     246           3 :                 MT_lock_unset(&mal_remoteLock);
     247           3 :                 msg = createException(IO, "remote.connect",
     248             :                                                           "unable to connect to '%s': %s",
     249             :                                                           *ouri, mapi_error_str(m));
     250           3 :                 mapi_destroy(m);
     251           3 :                 return msg;
     252             :         }
     253             : 
     254         127 :         if (columnar && *columnar) {
     255             :                 char set_protocol_query_buf[50];
     256           1 :                 snprintf(set_protocol_query_buf, 50, "sql.set_protocol(%d:int);", PROTOCOL_COLUMNAR);
     257           1 :                 if ((msg = RMTquery(&hdl, "remote.connect", m, set_protocol_query_buf))) {
     258           0 :                         mapi_destroy(m);
     259           0 :                         MT_lock_unset(&mal_remoteLock);
     260           0 :                         return msg;
     261             :                 }
     262             :         }
     263             : 
     264             :         /* connection established, add to list */
     265         127 :         c = GDKzalloc(sizeof(struct _connection));
     266         127 :         if ( c == NULL || (c->name = GDKstrdup(conn)) == NULL) {
     267           0 :                 GDKfree(c);
     268           0 :                 mapi_destroy(m);
     269           0 :                 MT_lock_unset(&mal_remoteLock);
     270           0 :                 throw(MAL,"remote.connect", SQLSTATE(HY013) MAL_MALLOC_FAIL);
     271             :         }
     272         127 :         c->mconn = m;
     273         127 :         c->nextid = 0;
     274         127 :         MT_lock_init(&c->lock, c->name);
     275         127 :         c->next = conns;
     276         127 :         conns = c;
     277             : 
     278         127 :         msg = RMTquery(&hdl, "remote.connect", m, "remote.bintype();");
     279         127 :         if (msg) {
     280           0 :                 MT_lock_unset(&mal_remoteLock);
     281           0 :                 return msg;
     282             :         }
     283         254 :         if (hdl != NULL && mapi_fetch_row(hdl)) {
     284         127 :                 char *val = mapi_fetch_field(hdl, 0);
     285         127 :                 c->type = (unsigned char)atoi(val);
     286         127 :                 mapi_close_handle(hdl);
     287             :         } else {
     288           0 :                 c->type = 0;
     289             :         }
     290             : 
     291             : #ifdef _DEBUG_MAPI_
     292             :         mapi_trace(c->mconn, true);
     293             : #endif
     294             : 
     295         127 :         MT_lock_unset(&mal_remoteLock);
     296             : 
     297         127 :         *ret = GDKstrdup(conn);
     298         127 :         if(*ret == NULL)
     299           0 :                 throw(MAL,"remote.connect", SQLSTATE(HY013) MAL_MALLOC_FAIL);
     300             :         return(MAL_SUCCEED);
     301             : }
     302             : 
     303             : static str
     304           0 : RMTconnect(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci) {
     305             :         (void) cntxt;
     306             :         (void) mb;
     307           0 :                 str* ret        = getArgReference_str(stk, pci, 0);
     308           0 :                 str* uri        = getArgReference_str(stk, pci, 1);
     309           0 :                 str* user       = getArgReference_str(stk, pci, 2);
     310           0 :                 str* passwd     = getArgReference_str(stk, pci, 3);
     311             : 
     312           0 :                 str scen = "msql";
     313             : 
     314           0 :                 if (pci->argc >= 5)
     315           0 :                         scen = *getArgReference_str(stk, pci, 4);
     316             : 
     317           0 :                 return RMTconnectScen(ret, uri, user, passwd, &scen, NULL);
     318             : }
     319             : 
     320             : static str
     321         129 : RMTconnectTable(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
     322             : {
     323             :         char *local_table;
     324             :         char *remoteuser;
     325             :         char *passwd;
     326             :         char *uri;
     327             :         char *tmp;
     328             :         char *ret;
     329             :         str scen;
     330             :         str msg;
     331             :         ValPtr v;
     332             : 
     333             :         (void)mb;
     334             :         (void)cntxt;
     335             : 
     336         129 :         local_table = *getArgReference_str(stk, pci, 1);
     337         129 :         scen = *getArgReference_str(stk, pci, 2);
     338         129 :         if (local_table == NULL || strcmp(local_table, (str)str_nil) == 0) {
     339           0 :                 throw(ILLARG, "remote.connect", ILLEGAL_ARGUMENT ": local table is NULL or nil");
     340             :         }
     341             : 
     342         129 :         rethrow("remote.connect", tmp, AUTHgetRemoteTableCredentials(local_table, &uri, &remoteuser, &passwd));
     343             : 
     344             :         /* The password we just got is hashed. Add the byte \1 in front to
     345             :          * signal this fact to the mapi. */
     346         129 :         size_t pwlen = strlen(passwd);
     347         129 :         char *pwhash = (char*)GDKmalloc(pwlen + 2);
     348         129 :         if (pwhash == NULL) {
     349           0 :                 GDKfree(remoteuser);
     350           0 :                 GDKfree(passwd);
     351           0 :                 throw(MAL, "remote.connect", SQLSTATE(HY013) MAL_MALLOC_FAIL);
     352             :         }
     353         129 :         snprintf(pwhash, pwlen + 2, "\1%s", passwd);
     354             : 
     355         129 :         msg = RMTconnectScen(&ret, &uri, &remoteuser, &pwhash, &scen, NULL);
     356             : 
     357         129 :         GDKfree(passwd);
     358         129 :         GDKfree(pwhash);
     359             : 
     360         129 :         if (msg == MAL_SUCCEED) {
     361         126 :                 v = &stk->stk[pci->argv[0]];
     362         126 :                 if (VALinit(v, TYPE_str, ret) == NULL) {
     363           0 :                         GDKfree(ret);
     364           0 :                         throw(MAL, "remote.connect", SQLSTATE(HY013) MAL_MALLOC_FAIL);
     365             :                 }
     366             :         }
     367             : 
     368         129 :         GDKfree(ret);
     369         129 :         return msg;
     370             : }
     371             : 
     372             : 
     373             : /**
     374             :  * Disconnects a connection.  The connection needs not to exist in the
     375             :  * system, it only needs to exist for the client (i.e. it was once
     376             :  * created).
     377             :  */
     378             : str
     379         127 : RMTdisconnect(void *ret, str *conn) {
     380             :         connection c, t;
     381             : 
     382         127 :         if (conn == NULL || *conn == NULL || strcmp(*conn, (str)str_nil) == 0)
     383           0 :                 throw(ILLARG, "remote.disconnect", ILLEGAL_ARGUMENT ": connection "
     384             :                                 "is NULL or nil");
     385             : 
     386             : 
     387             :         (void) ret;
     388             : 
     389             :         /* we need a lock because the same user can be handled by multiple
     390             :          * threads */
     391         127 :         MT_lock_set(&mal_remoteLock);
     392         127 :         c = conns;
     393             :         t = NULL; /* parent */
     394             :         /* walk through the list */
     395         153 :         while (c != NULL) {
     396         153 :                 if (strcmp(c->name, *conn) == 0) {
     397             :                         /* ok, delete it... */
     398         127 :                         if (t == NULL) {
     399         108 :                                 conns = c->next;
     400             :                         } else {
     401          19 :                                 t->next = c->next;
     402             :                         }
     403             : 
     404         127 :                         MT_lock_set(&c->lock); /* shared connection */
     405         127 :                         mapi_disconnect(c->mconn);
     406         127 :                         mapi_destroy(c->mconn);
     407         127 :                         MT_lock_unset(&c->lock);
     408         127 :                         MT_lock_destroy(&c->lock);
     409         127 :                         GDKfree(c->name);
     410         127 :                         GDKfree(c);
     411         127 :                         MT_lock_unset(&mal_remoteLock);
     412         127 :                         return MAL_SUCCEED;
     413             :                 }
     414             :                 t = c;
     415          26 :                 c = c->next;
     416             :         }
     417             : 
     418           0 :         MT_lock_unset(&mal_remoteLock);
     419           0 :         throw(MAL, "remote.disconnect", "no such connection: %s", *conn);
     420             : }
     421             : 
     422             : /**
     423             :  * Helper function to return a connection matching a given string, or an
     424             :  * error if it does not exist.  Since this function is internal, it
     425             :  * doesn't check the argument conn, as it should have been checked
     426             :  * already.
     427             :  * NOTE: this function acquires the mal_remoteLock before accessing conns
     428             :  */
     429             : static inline str
     430        4291 : RMTfindconn(connection *ret, const char *conn) {
     431             :         connection c;
     432             : 
     433             :         /* just make sure the return isn't garbage */
     434        4291 :         *ret = NULL;
     435        4291 :         MT_lock_set(&mal_remoteLock); /* protect c */
     436        4291 :         c = conns;
     437        5012 :         while (c != NULL) {
     438        5012 :                 if (strcmp(c->name, conn) == 0) {
     439        4291 :                         *ret = c;
     440        4291 :                         MT_lock_unset(&mal_remoteLock);
     441        4291 :                         return(MAL_SUCCEED);
     442             :                 }
     443         721 :                 c = c->next;
     444             :         }
     445           0 :         MT_lock_unset(&mal_remoteLock);
     446           0 :         throw(MAL, "remote.<findconn>", "no such connection: %s", conn);
     447             : }
     448             : 
     449             : /**
     450             :  * Little helper function that returns a GDKmalloced string containing a
     451             :  * valid identifier that is supposed to be unique in the connection's
     452             :  * remote context.  The generated string depends on the module and
     453             :  * function the caller is in. But also the runtime context is important.
     454             :  * The format is rmt<id>_<retvar>_<type>.  Every RMTgetId uses a fresh id,
     455             :  * to distinguish amongst different (parallel) execution context.
     456             :  * Re-use of this remote identifier should be done with care.
     457             :  * The encoding of the type allows for ease of type checking later on.
     458             :  */
     459             : static inline str
     460        2535 : RMTgetId(char *buf, MalBlkPtr mb, InstrPtr p, int arg) {
     461             :         InstrPtr f;
     462             :         const char *mod;
     463             :         char *var;
     464             :         str rt;
     465             :         static ATOMIC_TYPE idtag = ATOMIC_VAR_INIT(0);
     466             : 
     467        2535 :         if( p->retc == 0)
     468           0 :                 throw(MAL, "remote.getId", ILLEGAL_ARGUMENT "MAL instruction misses retc");
     469             : 
     470        2535 :         var = getArgName(mb, p, arg);
     471             :         f = getInstrPtr(mb, 0); /* top level function */
     472             :         mod = getModuleId(f);
     473             :         if (mod == NULL)
     474             :                 mod = "user";
     475        2535 :         rt = getTypeIdentifier(getArgType(mb,p,arg));
     476        2535 :         if (rt == NULL)
     477           0 :                 throw(MAL, "remote.put", SQLSTATE(HY013) MAL_MALLOC_FAIL);
     478             : 
     479        2535 :         snprintf(buf, BUFSIZ, "rmt%u_%s_%s", (unsigned) ATOMIC_ADD(&idtag, 1), var, rt);
     480             : 
     481        2535 :         GDKfree(rt);
     482        2535 :         return(MAL_SUCCEED);
     483             : }
     484             : 
     485             : /**
     486             :  * Helper function to execute a query over the given connection,
     487             :  * returning the result handle.  If communication fails in one way or
     488             :  * another, an error is returned.  Since this function is internal, it
     489             :  * doesn't check the input arguments func, conn and query, as they
     490             :  * should have been checked already.
     491             :  * NOTE: this function assumes a lock for conn is set
     492             :  */
     493             : static inline str
     494        3154 : RMTquery(MapiHdl *ret, const char *func, Mapi conn, const char *query) {
     495             :         MapiHdl mhdl;
     496             : 
     497        3154 :         *ret = NULL;
     498        3154 :         mhdl = mapi_query(conn, query);
     499        3154 :         if (mhdl) {
     500        3154 :                 if (mapi_result_error(mhdl) != NULL) {
     501           8 :                         str err = createException(
     502             :                                         getExceptionType(mapi_result_error(mhdl)),
     503             :                                         func,
     504             :                                         "(mapi:monetdb://%s@%s/%s) %s",
     505             :                                         mapi_get_user(conn),
     506             :                                         mapi_get_host(conn),
     507             :                                         mapi_get_dbname(conn),
     508             :                                         getExceptionMessage(mapi_result_error(mhdl)));
     509           8 :                         mapi_close_handle(mhdl);
     510           8 :                         return(err);
     511             :                 }
     512             :         } else {
     513           0 :                 if (mapi_error(conn) != MOK) {
     514           0 :                         throw(IO, func, "an error occurred on connection: %s",
     515             :                                         mapi_error_str(conn));
     516             :                 } else {
     517           0 :                         throw(MAL, func, "remote function invocation didn't return a result");
     518             :                 }
     519             :         }
     520             : 
     521        3146 :         *ret = mhdl;
     522        3146 :         return(MAL_SUCCEED);
     523             : }
     524             : 
     525         266 : static str RMTprelude(void *ret) {
     526             :         unsigned int type = 0;
     527             : 
     528             :         (void)ret;
     529             : #ifdef WORDS_BIGENDIAN
     530             :         type |= RMTT_B_ENDIAN;
     531             : #else
     532             :         type |= RMTT_L_ENDIAN;
     533             : #endif
     534             : #if SIZEOF_SIZE_T == SIZEOF_LNG
     535             :         type |= RMTT_64_BITS;
     536             : #else
     537             :         type |= RMTT_32_BITS;
     538             : #endif
     539             : #if SIZEOF_OID == SIZEOF_LNG
     540             :         type |= RMTT_64_OIDS;
     541             : #else
     542             :         type |= RMTT_32_OIDS;
     543             : #endif
     544         266 :         localtype = (unsigned char)type;
     545             : 
     546         266 :         return(MAL_SUCCEED);
     547             : }
     548             : 
     549         264 : static str RMTepilogue(void *ret) {
     550             :         connection c, t;
     551             : 
     552             :         (void)ret;
     553             : 
     554         264 :         MT_lock_set(&mal_remoteLock); /* nobody allowed here */
     555             :         /* free connections list */
     556         264 :         c = conns;
     557         264 :         while (c != NULL) {
     558             :                 t = c;
     559           0 :                 c = c->next;
     560           0 :                 MT_lock_set(&t->lock);
     561           0 :                 mapi_destroy(t->mconn);
     562           0 :                 MT_lock_unset(&t->lock);
     563           0 :                 MT_lock_destroy(&t->lock);
     564           0 :                 GDKfree(t->name);
     565           0 :                 GDKfree(t);
     566             :         }
     567             :         /* not sure, but better be safe than sorry */
     568         264 :         conns = NULL;
     569         264 :         MT_lock_unset(&mal_remoteLock);
     570             : 
     571         264 :         return(MAL_SUCCEED);
     572             : }
     573             : static str
     574        1267 : RMTreadbatheader(stream* sin, char* buf) {
     575             :                 ssize_t sz = 0, rd;
     576             : 
     577             :                 /* read the JSON header */
     578      205839 :                 while ((rd = mnstr_read(sin, &buf[sz], 1, 1)) == 1 && buf[sz] != '\n') {
     579      204572 :                         sz += rd;
     580             :                 }
     581        1267 :                 if (rd < 0) {
     582           0 :                         throw(MAL, "remote.get", "could not read BAT JSON header");
     583             :                 }
     584        1267 :                 if (buf[0] == '!') {
     585             :                         char *result;
     586           0 :                         if((result = GDKstrdup(buf)) == NULL)
     587           0 :                                 throw(MAL, "remote.get", SQLSTATE(HY013) MAL_MALLOC_FAIL);
     588             :                         return result;
     589             :                 }
     590             : 
     591        1267 :                 buf[sz] = '\0';
     592             : 
     593        1267 :                 return MAL_SUCCEED;
     594             : }
     595             : 
     596             : typedef struct _binbat_v1 {
     597             :         int Ttype;
     598             :         oid Hseqbase;
     599             :         oid Tseqbase;
     600             :         bool
     601             :                 Tsorted:1,
     602             :                 Trevsorted:1,
     603             :                 Tkey:1,
     604             :                 Tnonil:1,
     605             :                 Tdense:1;
     606             :         BUN size;
     607             :         size_t tailsize;
     608             :         size_t theapsize;
     609             : } binbat;
     610             : 
     611             : static str
     612        1267 : RMTinternalcopyfrom(BAT **ret, char *hdr, stream *in, bool must_flush)
     613             : {
     614        1267 :         binbat bb = { 0, 0, 0, false, false, false, false, false, 0, 0, 0 };
     615             :         char *nme = NULL;
     616             :         char *val = NULL;
     617             :         char tmp;
     618             :         size_t len;
     619             :         lng lv, *lvp;
     620             : 
     621             :         BAT *b;
     622             : 
     623             :         /* hdr is a JSON structure that looks like
     624             :          * {"version":1,"ttype":6,"tseqbase":0,"tailsize":4,"theapsize":0}
     625             :          * we take the binary data directly from the stream */
     626             : 
     627             :         /* could skip whitespace, but we just don't allow that */
     628        1267 :         if (*hdr++ != '{')
     629           0 :                 throw(MAL, "remote.bincopyfrom", "illegal input, not a JSON header (got '%s')", hdr - 1);
     630      204558 :         while (*hdr != '\0') {
     631      203291 :                 switch (*hdr) {
     632       30407 :                         case '"':
     633             :                                 /* we assume only numeric values, so all strings are
     634             :                                  * elems */
     635       30407 :                                 if (nme != NULL) {
     636       15204 :                                         *hdr = '\0';
     637             :                                 } else {
     638       15203 :                                         nme = hdr + 1;
     639             :                                 }
     640             :                                 break;
     641       15203 :                         case ':':
     642       15203 :                                 val = hdr + 1;
     643       15203 :                                 break;
     644       15204 :                         case ',':
     645             :                         case '}':
     646       15204 :                                 if (val == NULL)
     647           0 :                                         throw(MAL, "remote.bincopyfrom",
     648             :                                                         "illegal input, JSON value missing");
     649       15204 :                                 *hdr = '\0';
     650             : 
     651       15204 :                                 lvp = &lv;
     652       15204 :                                 len = sizeof(lv);
     653             :                                 /* tseqbase can be 1<<31/1<<63 which causes overflow
     654             :                                  * in lngFromStr, so we check separately */
     655       15204 :                                 if (strcmp(val,
     656             : #if SIZEOF_OID == 8
     657             :                                                    "9223372036854775808"
     658             : #else
     659             :                                                    "2147483648"
     660             : #endif
     661        1256 :                                                 ) == 0 &&
     662        1256 :                                         strcmp(nme, "tseqbase") == 0) {
     663        1256 :                                         bb.Tseqbase = oid_nil;
     664             :                                 } else {
     665             :                                         /* all values should be non-negative, so we check that
     666             :                                          * here as well */
     667       13948 :                                         if (lngFromStr(val, &len, &lvp, true) < 0 ||
     668       13948 :                                                 lv < 0 /* includes lng_nil */)
     669           0 :                                                 throw(MAL, "remote.bincopyfrom",
     670             :                                                           "bad %s value: %s", nme, val);
     671             : 
     672             :                                         /* deal with nme and val */
     673       13948 :                                         if (strcmp(nme, "version") == 0) {
     674        1267 :                                                 if (lv != 1)
     675           0 :                                                         throw(MAL, "remote.bincopyfrom",
     676             :                                                                   "unsupported version: %s", val);
     677       12681 :                                         } else if (strcmp(nme, "hseqbase") == 0) {
     678             : #if SIZEOF_OID < SIZEOF_LNG
     679             :                                                 if (lv > GDK_oid_max)
     680             :                                                         throw(MAL, "remote.bincopyfrom",
     681             :                                                                           "bad %s value: %s", nme, val);
     682             : #endif
     683             :                                                 bb.Hseqbase = (oid)lv;
     684       11414 :                                         } else if (strcmp(nme, "ttype") == 0) {
     685        1267 :                                                 if (lv >= GDKatomcnt)
     686           0 :                                                         throw(MAL, "remote.bincopyfrom",
     687             :                                                                   "bad %s value: GDK atom number %s doesn't exist", nme, val);
     688        1267 :                                                 bb.Ttype = (int) lv;
     689       10147 :                                         } else if (strcmp(nme, "tseqbase") == 0) {
     690             : #if SIZEOF_OID < SIZEOF_LNG
     691             :                                                 if (lv > GDK_oid_max)
     692             :                                                         throw(MAL, "remote.bincopyfrom",
     693             :                                                                   "bad %s value: %s", nme, val);
     694             : #endif
     695          11 :                                                 bb.Tseqbase = (oid) lv;
     696       10136 :                                         } else if (strcmp(nme, "tsorted") == 0) {
     697        1267 :                                                 bb.Tsorted = lv != 0;
     698        8869 :                                         } else if (strcmp(nme, "trevsorted") == 0) {
     699        1267 :                                                 bb.Trevsorted = lv != 0;
     700        7602 :                                         } else if (strcmp(nme, "tkey") == 0) {
     701        1267 :                                                 bb.Tkey = lv != 0;
     702        6335 :                                         } else if (strcmp(nme, "tnonil") == 0) {
     703        1267 :                                                 bb.Tnonil = lv != 0;
     704        5068 :                                         } else if (strcmp(nme, "tdense") == 0) {
     705        1267 :                                                 bb.Tdense = lv != 0;
     706        3801 :                                         } else if (strcmp(nme, "size") == 0) {
     707        1267 :                                                 if (lv > (lng) BUN_MAX)
     708           0 :                                                         throw(MAL, "remote.bincopyfrom",
     709             :                                                                   "bad %s value: %s", nme, val);
     710        1267 :                                                 bb.size = (BUN) lv;
     711        2534 :                                         } else if (strcmp(nme, "tailsize") == 0) {
     712        1267 :                                                 bb.tailsize = (size_t) lv;
     713        1267 :                                         } else if (strcmp(nme, "theapsize") == 0) {
     714        1267 :                                                 bb.theapsize = (size_t) lv;
     715             :                                         } else {
     716           0 :                                                 throw(MAL, "remote.bincopyfrom",
     717             :                                                           "unknown element: %s", nme);
     718             :                                         }
     719             :                                 }
     720             :                                 nme = val = NULL;
     721             :                                 break;
     722             :                 }
     723      203291 :                 hdr++;
     724             :         }
     725             : 
     726        1267 :         b = COLnew(0, bb.Ttype, bb.size, TRANSIENT);
     727        1267 :         if (b == NULL)
     728           0 :                 throw(MAL, "remote.get", SQLSTATE(HY013) MAL_MALLOC_FAIL);
     729             : 
     730             :         /* for strings, the width may not match, fix it to match what we
     731             :          * retrieved */
     732        1267 :         if (bb.Ttype == TYPE_str && bb.size) {
     733          41 :                 b->twidth = (unsigned short) (bb.tailsize / bb.size);
     734          41 :                 b->tshift = ATOMelmshift(Tsize(b));
     735             :         }
     736             : 
     737        1267 :         if (bb.tailsize > 0) {
     738        2366 :                 if (HEAPextend(b->theap, bb.tailsize, true) != GDK_SUCCEED ||
     739        1183 :                         mnstr_read(in, b->theap->base, bb.tailsize, 1) < 0)
     740           0 :                         goto bailout;
     741        1183 :                 b->theap->dirty = true;
     742             :         }
     743        1267 :         if (bb.theapsize > 0) {
     744          82 :                 if (HEAPextend(b->tvheap, bb.theapsize, true) != GDK_SUCCEED ||
     745          41 :                         mnstr_read(in, b->tvheap->base, bb.theapsize, 1) < 0)
     746           0 :                         goto bailout;
     747          41 :                 b->tvheap->free = bb.theapsize;
     748          41 :                 b->tvheap->dirty = true;
     749             :         }
     750             : 
     751             :         /* set properties */
     752        1267 :         b->tseqbase = bb.Tdense ? bb.Tseqbase : oid_nil;
     753        1267 :         b->tsorted = bb.Tsorted;
     754        1267 :         b->trevsorted = bb.Trevsorted;
     755        1267 :         b->tkey = bb.Tkey;
     756        1267 :         b->tnonil = bb.Tnonil;
     757        1267 :         if (bb.Ttype == TYPE_str && bb.size)
     758          41 :                 BATsetcapacity(b, (BUN) (bb.tailsize >> b->tshift));
     759        1267 :         BATsetcount(b, bb.size);
     760             : 
     761             :         // read blockmode flush
     762        1267 :         while (must_flush && mnstr_read(in, &tmp, 1, 1) > 0) {
     763           0 :                 TRC_ERROR(MAL_REMOTE, "Expected flush, got: %c\n", tmp);
     764             :         }
     765             : 
     766        1267 :         BATsettrivprop(b);
     767             : 
     768        1267 :         *ret = b;
     769        1267 :         return(MAL_SUCCEED);
     770             : 
     771           0 :   bailout:
     772           0 :         BBPreclaim(b);
     773           0 :         throw(MAL, "remote.bincopyfrom", "reading failed");
     774             : }
     775             : 
     776             : /**
     777             :  * get fetches the object referenced by ident over connection conn.
     778             :  * We are only interested in retrieving void-headed BATs, i.e. single columns.
     779             :  */
     780        1265 : static str RMTget(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci) {
     781             :         str conn, ident, tmp, rt;
     782             :         connection c;
     783             :         char qbuf[BUFSIZ + 1];
     784        1265 :         MapiHdl mhdl = NULL;
     785             :         int rtype;
     786             :         ValPtr v;
     787             : 
     788             :         (void)mb;
     789             :         (void) cntxt;
     790             : 
     791        1265 :         conn = *getArgReference_str(stk, pci, 1);
     792        1265 :         if (conn == NULL || strcmp(conn, (str)str_nil) == 0)
     793           0 :                 throw(ILLARG, "remote.get", ILLEGAL_ARGUMENT ": connection name is NULL or nil");
     794        1265 :         ident = *getArgReference_str(stk, pci, 2);
     795        1265 :         if (ident == 0 || isIdentifier(ident) < 0)
     796           0 :                 throw(ILLARG, "remote.get", ILLEGAL_ARGUMENT ": identifier expected, got '%s'", ident);
     797             : 
     798             :         /* lookup conn, set c if valid */
     799        1265 :         rethrow("remote.get", tmp, RMTfindconn(&c, conn));
     800             : 
     801        1265 :         rtype = getArgType(mb, pci, 0);
     802        1265 :         v = &stk->stk[pci->argv[0]];
     803             : 
     804        1265 :         if (rtype == TYPE_any || isAnyExpression(rtype)) {
     805             :                 char *tpe, *msg;
     806           0 :                 tpe = getTypeName(rtype);
     807           0 :                 msg = createException(MAL, "remote.get", ILLEGAL_ARGUMENT ": unsupported any type: %s",
     808             :                                                           tpe);
     809           0 :                 GDKfree(tpe);
     810           0 :                 return msg;
     811             :         }
     812             :         /* check if the remote type complies with what we expect.
     813             :            Since the put() encodes the type as known to the remote site
     814             :            we can simple compare it here */
     815        1265 :         rt = getTypeIdentifier(rtype);
     816        1265 :         if (rt == NULL)
     817           0 :                 throw(MAL, "remote.get", SQLSTATE(HY013) MAL_MALLOC_FAIL);
     818        1265 :         if (strcmp(ident + strlen(ident) - strlen(rt), rt)) {
     819           0 :                 tmp = createException(MAL, "remote.get", ILLEGAL_ARGUMENT
     820             :                         ": remote object type %s does not match expected type %s",
     821             :                         rt, ident);
     822           0 :                 GDKfree(rt);
     823           0 :                 return tmp;
     824             :         }
     825        1265 :         GDKfree(rt);
     826             : 
     827        1265 :         if (isaBatType(rtype) && (localtype == 0177 || localtype != c->type ))
     828           0 :         {
     829             :                 int t;
     830             :                 size_t s;
     831             :                 ptr r;
     832             :                 str var;
     833             :                 BAT *b;
     834             : 
     835           0 :                 snprintf(qbuf, BUFSIZ, "io.print(%s);", ident);
     836             : 
     837           0 :                 TRC_DEBUG(MAL_REMOTE, "Remote get: %s\n", qbuf);
     838             : 
     839             :                 /* this call should be a single transaction over the channel*/
     840           0 :                 MT_lock_set(&c->lock);
     841             : 
     842           0 :                 if ((tmp = RMTquery(&mhdl, "remote.get", c->mconn, qbuf))
     843             :                                 != MAL_SUCCEED)
     844             :                 {
     845           0 :                         TRC_ERROR(MAL_REMOTE, "Remote get: %s\n%s\n", qbuf, tmp);
     846           0 :                         MT_lock_unset(&c->lock);
     847           0 :                         var = createException(MAL, "remote.get", "%s", tmp);
     848           0 :                         freeException(tmp);
     849           0 :                         return var;
     850             :                 }
     851           0 :                 t = getBatType(rtype);
     852           0 :                 b = COLnew(0, t, 0, TRANSIENT);
     853           0 :                 if (b == NULL) {
     854           0 :                         mapi_close_handle(mhdl);
     855           0 :                         MT_lock_unset(&c->lock);
     856           0 :                         throw(MAL, "remote.get", SQLSTATE(HY013) MAL_MALLOC_FAIL);
     857             :                 }
     858             : 
     859           0 :                 if (ATOMbasetype(t) == TYPE_str) {
     860           0 :                         while (mapi_fetch_row(mhdl)) {
     861           0 :                                 var = mapi_fetch_field(mhdl, 1);
     862           0 :                                 if (BUNappend(b, var == NULL ? str_nil : var, false) != GDK_SUCCEED) {
     863           0 :                                         BBPreclaim(b);
     864           0 :                                         mapi_close_handle(mhdl);
     865           0 :                                         MT_lock_unset(&c->lock);
     866           0 :                                         throw(MAL, "remote.get", SQLSTATE(HY013) MAL_MALLOC_FAIL);
     867             :                                 }
     868             :                         }
     869             :                 } else
     870           0 :                         while (mapi_fetch_row(mhdl)) {
     871           0 :                                 var = mapi_fetch_field(mhdl, 1);
     872           0 :                                 if (var == NULL)
     873             :                                         var = "nil";
     874           0 :                                 s = 0;
     875           0 :                                 r = NULL;
     876           0 :                                 if (ATOMfromstr(t, &r, &s, var, true) < 0 ||
     877           0 :                                         BUNappend(b, r, false) != GDK_SUCCEED) {
     878           0 :                                         BBPreclaim(b);
     879           0 :                                         GDKfree(r);
     880           0 :                                         mapi_close_handle(mhdl);
     881           0 :                                         MT_lock_unset(&c->lock);
     882           0 :                                         throw(MAL, "remote.get", GDK_EXCEPTION);
     883             :                                 }
     884           0 :                                 GDKfree(r);
     885             :                         }
     886             : 
     887           0 :                 v->val.bval = b->batCacheid;
     888           0 :                 v->vtype = TYPE_bat;
     889           0 :                 BBPkeepref(b->batCacheid);
     890             : 
     891           0 :                 mapi_close_handle(mhdl);
     892           0 :                 MT_lock_unset(&c->lock);
     893        2530 :         } else if (isaBatType(rtype)) {
     894             :                 /* binary compatible remote host, transfer BAT in binary form */
     895             :                 stream *sout;
     896             :                 stream *sin;
     897             :                 char buf[256];
     898        1265 :                 BAT *b = NULL;
     899             : 
     900             :                 /* this call should be a single transaction over the channel*/
     901        1265 :                 MT_lock_set(&c->lock);
     902             : 
     903             :                 /* bypass Mapi from this point to efficiently write all data to
     904             :                  * the server */
     905        1265 :                 sout = mapi_get_to(c->mconn);
     906        1265 :                 sin = mapi_get_from(c->mconn);
     907        1265 :                 if (sin == NULL || sout == NULL) {
     908           0 :                         MT_lock_unset(&c->lock);
     909           0 :                         throw(MAL, "remote.get", "Connection lost");
     910             :                 }
     911             : 
     912             :                 /* call our remote helper to do this more efficiently */
     913        1265 :                 mnstr_printf(sout, "remote.batbincopy(%s);\n", ident);
     914        1265 :                 mnstr_flush(sout, MNSTR_FLUSH_DATA);
     915             : 
     916        1265 :                 if ((tmp = RMTreadbatheader(sin, buf)) != MAL_SUCCEED) {
     917           0 :                         MT_lock_unset(&c->lock);
     918           0 :                         return tmp;
     919             :                 }
     920             : 
     921        1265 :                 if ((tmp = RMTinternalcopyfrom(&b, buf, sin, true)) != MAL_SUCCEED) {
     922           0 :                         MT_lock_unset(&c->lock);
     923           0 :                         return(tmp);
     924             :                 }
     925             : 
     926        1265 :                 v->val.bval = b->batCacheid;
     927        1265 :                 v->vtype = TYPE_bat;
     928        1265 :                 BBPkeepref(b->batCacheid);
     929             : 
     930        1265 :                 MT_lock_unset(&c->lock);
     931             :         } else {
     932           0 :                 ptr p = NULL;
     933             :                 str val;
     934           0 :                 size_t len = 0;
     935             : 
     936           0 :                 snprintf(qbuf, BUFSIZ, "io.print(%s);", ident);
     937           0 :                 TRC_DEBUG(MAL_REMOTE, "Remote get: %s - %s\n", c->name, qbuf);
     938           0 :                 if ((tmp=RMTquery(&mhdl, "remote.get", c->mconn, qbuf)) != MAL_SUCCEED)
     939             :                 {
     940           0 :                         return tmp;
     941             :                 }
     942           0 :                 (void) mapi_fetch_row(mhdl); /* should succeed */
     943           0 :                 val = mapi_fetch_field(mhdl, 0);
     944             : 
     945           0 :                 if (ATOMbasetype(rtype) == TYPE_str) {
     946           0 :                         if (!VALinit(v, rtype, val == NULL ? str_nil : val)) {
     947           0 :                                 mapi_close_handle(mhdl);
     948           0 :                                 throw(MAL, "remote.get", SQLSTATE(HY013) MAL_MALLOC_FAIL);
     949             :                         }
     950           0 :                 } else if (ATOMfromstr(rtype, &p, &len, val == NULL ? "nil" : val, true) < 0) {
     951             :                         char *msg;
     952           0 :                         msg = createException(MAL, "remote.get",
     953             :                                                                   "unable to parse value: %s",
     954             :                                                                   val == NULL ? "nil" : val);
     955           0 :                         mapi_close_handle(mhdl);
     956           0 :                         GDKfree(p);
     957           0 :                         return msg;
     958             :                 } else {
     959           0 :                         VALset(v, rtype, p);
     960           0 :                         if (ATOMextern(rtype) == 0)
     961           0 :                                 GDKfree(p);
     962             :                 }
     963             : 
     964           0 :                 mapi_close_handle(mhdl);
     965             :         }
     966             : 
     967             :         return(MAL_SUCCEED);
     968             : }
     969             : 
     970             : /**
     971             :  * stores the given object on the remote host.  The identifier of the
     972             :  * object on the remote host is returned for later use.
     973             :  */
     974        2535 : static str RMTput(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci) {
     975             :         str conn, tmp;
     976             :         char ident[BUFSIZ];
     977             :         connection c;
     978             :         ValPtr v;
     979             :         int type;
     980             :         ptr value;
     981        2535 :         MapiHdl mhdl = NULL;
     982             : 
     983             :         (void)cntxt;
     984             : 
     985        2535 :         conn = *getArgReference_str(stk, pci, 1);
     986        2535 :         if (conn == NULL || strcmp(conn, (str)str_nil) == 0)
     987           0 :                 throw(ILLARG, "remote.put", ILLEGAL_ARGUMENT ": connection name is NULL or nil");
     988             : 
     989             :         /* lookup conn */
     990        2535 :         rethrow("remote.put", tmp, RMTfindconn(&c, conn));
     991             : 
     992             :         /* put the thing */
     993        2535 :         type = getArgType(mb, pci, 2);
     994        2535 :         value = getArgReference(stk, pci, 2);
     995             : 
     996             :         /* this call should be a single transaction over the channel*/
     997        2535 :         MT_lock_set(&c->lock);
     998             : 
     999             :         /* get a free, typed identifier for the remote host */
    1000        2535 :         tmp = RMTgetId(ident, mb, pci, 2);
    1001        2535 :         if (tmp != MAL_SUCCEED) {
    1002           0 :                 MT_lock_unset(&c->lock);
    1003           0 :                 return tmp;
    1004             :         }
    1005             : 
    1006             :         /* depending on the input object generate actions to store the
    1007             :          * object remotely*/
    1008        2535 :         if (type == TYPE_any || type == TYPE_bat || isAnyExpression(type)) {
    1009             :                 char *tpe, *msg;
    1010           0 :                 MT_lock_unset(&c->lock);
    1011           0 :                 tpe = getTypeName(type);
    1012           0 :                 msg = createException(MAL, "remote.put", "unsupported type: %s", tpe);
    1013           0 :                 GDKfree(tpe);
    1014           0 :                 return msg;
    1015        3801 :         } else if (isaBatType(type) && !is_bat_nil(*(bat*) value)) {
    1016             :                 BATiter bi;
    1017             :                 /* naive approach using bat.new() and bat.insert() calls */
    1018             :                 char *tail;
    1019             :                 bat bid;
    1020             :                 BAT *b = NULL;
    1021             :                 BUN p, q;
    1022             :                 str tailv;
    1023             :                 stream *sout;
    1024             : 
    1025        1266 :                 tail = getTypeIdentifier(getBatType(type));
    1026        1266 :                 if (tail == NULL) {
    1027           0 :                         MT_lock_unset(&c->lock);
    1028           0 :                         throw(MAL, "remote.put", SQLSTATE(HY013) MAL_MALLOC_FAIL);
    1029             :                 }
    1030             : 
    1031        1266 :                 bid = *(bat *)value;
    1032        1266 :                 if (bid != 0) {
    1033        1266 :                         if ((b = BATdescriptor(bid)) == NULL){
    1034           0 :                                 MT_lock_unset(&c->lock);
    1035           0 :                                 GDKfree(tail);
    1036           0 :                                 throw(MAL, "remote.put", SQLSTATE(HY002) RUNTIME_OBJECT_MISSING);
    1037             :                         }
    1038             :                 }
    1039             : 
    1040             :                 /* bypass Mapi from this point to efficiently write all data to
    1041             :                  * the server */
    1042        1266 :                 sout = mapi_get_to(c->mconn);
    1043             : 
    1044             :                 /* call our remote helper to do this more efficiently */
    1045        1266 :                 mnstr_printf(sout,
    1046             :                                 "%s := remote.batload(nil:%s, " BUNFMT ");\n",
    1047             :                                 ident, tail, (bid == 0 ? 0 : BATcount(b)));
    1048        1266 :                 mnstr_flush(sout, MNSTR_FLUSH_DATA);
    1049        1266 :                 GDKfree(tail);
    1050             : 
    1051             :                 /* b can be NULL if bid == 0 (only type given, ugh) */
    1052        1266 :                 if (b) {
    1053        1266 :                         int tpe = getBatType(type), trivial = tpe < TYPE_date || ATOMbasetype(tpe) == TYPE_str;
    1054        1266 :                         const void *nil = ATOMnilptr(tpe);
    1055        1266 :                         int (*atomcmp)(const void *, const void *) = ATOMcompare(tpe);
    1056             : 
    1057        1266 :                         bi = bat_iterator(b);
    1058        1266 :                         BATloop(b, p, q) {
    1059           0 :                                 const void *v = BUNtail(bi, p);
    1060           0 :                                 tailv = ATOMformat(tpe, v);
    1061           0 :                                 if (tailv == NULL) {
    1062           0 :                                         bat_iterator_end(&bi);
    1063           0 :                                         BBPunfix(b->batCacheid);
    1064           0 :                                         MT_lock_unset(&c->lock);
    1065           0 :                                         throw(MAL, "remote.put", GDK_EXCEPTION);
    1066             :                                 }
    1067           0 :                                 if (trivial || atomcmp(v, nil) == 0)
    1068           0 :                                         mnstr_printf(sout, "%s\n", tailv);
    1069             :                                 else
    1070           0 :                                         mnstr_printf(sout, "\"%s\"\n", tailv);
    1071           0 :                                 GDKfree(tailv);
    1072             :                         }
    1073        1266 :                         bat_iterator_end(&bi);
    1074        1266 :                         BBPunfix(b->batCacheid);
    1075             :                 }
    1076             : 
    1077             :                 /* write the empty line the server is waiting for, handles
    1078             :                  * all errors at the same time, if any */
    1079        1266 :                 if ((tmp = RMTquery(&mhdl, "remote.put", c->mconn, ""))
    1080             :                                 != MAL_SUCCEED)
    1081             :                 {
    1082           0 :                         MT_lock_unset(&c->lock);
    1083           0 :                         return tmp;
    1084             :                 }
    1085        1266 :                 mapi_close_handle(mhdl);
    1086        1269 :         } else if (isaBatType(type) && is_bat_nil(*(bat*) value)) {
    1087             :                 stream *sout;
    1088           0 :                 str typename = getTypeName(type);
    1089           0 :                 sout = mapi_get_to(c->mconn);
    1090           0 :                 mnstr_printf(sout,
    1091             :                                 "%s := nil:%s;\n", ident, typename);
    1092           0 :                 mnstr_flush(sout, MNSTR_FLUSH_DATA);
    1093           0 :                 GDKfree(typename);
    1094             :         } else {
    1095             :                 size_t l;
    1096             :                 str val;
    1097             :                 char *tpe;
    1098             :                 char qbuf[512], *nbuf = qbuf;
    1099        1269 :                 const void *nil = ATOMnilptr(type), *p = value;
    1100        1269 :                 int (*atomcmp)(const void *, const void *) = ATOMcompare(type);
    1101             : 
    1102        1269 :                 if (ATOMextern(type))
    1103         878 :                         p = *(str *)value;
    1104             : 
    1105        1269 :                 val = ATOMformat(type, p);
    1106        1269 :                 if (val == NULL) {
    1107           0 :                         MT_lock_unset(&c->lock);
    1108           0 :                         throw(MAL, "remote.put", GDK_EXCEPTION);
    1109             :                 }
    1110        1269 :                 tpe = getTypeIdentifier(type);
    1111        1269 :                 if (tpe == NULL) {
    1112           0 :                         MT_lock_unset(&c->lock);
    1113           0 :                         GDKfree(val);
    1114           0 :                         throw(MAL, "remote.put", SQLSTATE(HY013) MAL_MALLOC_FAIL);
    1115             :                 }
    1116        1269 :                 l = strlen(val) + strlen(tpe) + strlen(ident) + 10;
    1117        1269 :                 if (l > (ssize_t) sizeof(qbuf) && (nbuf = GDKmalloc(l)) == NULL) {
    1118           0 :                         MT_lock_unset(&c->lock);
    1119           0 :                         GDKfree(val);
    1120           0 :                         GDKfree(tpe);
    1121           0 :                         throw(MAL, "remote.put", SQLSTATE(HY013) MAL_MALLOC_FAIL);
    1122             :                 }
    1123        1269 :                 if (type < TYPE_date || ATOMbasetype(type) == TYPE_str || atomcmp(p, nil) == 0)
    1124        1264 :                         snprintf(nbuf, l, "%s := %s:%s;\n", ident, val, tpe);
    1125             :                 else
    1126           5 :                         snprintf(nbuf, l, "%s := \"%s\":%s;\n", ident, val, tpe);
    1127        1269 :                 GDKfree(tpe);
    1128        1269 :                 GDKfree(val);
    1129        1269 :                 TRC_DEBUG(MAL_REMOTE, "Remote put: %s - %s\n", c->name, nbuf);
    1130        1269 :                 tmp = RMTquery(&mhdl, "remote.put", c->mconn, nbuf);
    1131        1269 :                 if (nbuf != qbuf)
    1132          26 :                         GDKfree(nbuf);
    1133        1269 :                 if (tmp != MAL_SUCCEED) {
    1134           0 :                         MT_lock_unset(&c->lock);
    1135           0 :                         return tmp;
    1136             :                 }
    1137        1269 :                 mapi_close_handle(mhdl);
    1138             :         }
    1139        2535 :         MT_lock_unset(&c->lock);
    1140             : 
    1141             :         /* return the identifier */
    1142        2535 :         v = &stk->stk[pci->argv[0]];
    1143        2535 :         if (VALinit(v, TYPE_str, ident) == NULL)
    1144           0 :                 throw(MAL, "remote.put", SQLSTATE(HY013) MAL_MALLOC_FAIL);
    1145             :         return(MAL_SUCCEED);
    1146             : }
    1147             : 
    1148             : /**
    1149             :  * stores the given <mod>.<fcn> on the remote host.
    1150             :  * An error is returned if the function is already known at the remote site.
    1151             :  * The implementation is based on serialisation of the block into a string
    1152             :  * followed by remote parsing.
    1153             :  */
    1154           0 : static str RMTregisterInternal(Client cntxt, char** fcn_id, const char *conn, const char *mod, const char *fcn)
    1155             : {
    1156             :         str tmp, qry, msg;
    1157             :         connection c;
    1158             :         char buf[BUFSIZ];
    1159           0 :         MapiHdl mhdl = NULL;
    1160             :         Symbol sym;
    1161             : 
    1162           0 :         if (strNil(conn))
    1163           0 :                 throw(ILLARG, "remote.register", ILLEGAL_ARGUMENT ": connection name is NULL or nil");
    1164             : 
    1165             :         /* find local definition */
    1166           0 :         sym = findSymbol(cntxt->usermodule, putName(mod), putName(fcn));
    1167           0 :         if (sym == NULL)
    1168           0 :                 throw(MAL, "remote.register", ILLEGAL_ARGUMENT ": no such function: %s.%s", mod, fcn);
    1169             : 
    1170             :         /* lookup conn */
    1171           0 :         rethrow("remote.register", tmp, RMTfindconn(&c, conn));
    1172             : 
    1173             :         /* this call should be a single transaction over the channel*/
    1174           0 :         MT_lock_set(&c->lock);
    1175             : 
    1176             :         /* get a free, typed identifier for the remote host */
    1177             :         char ident[BUFSIZ];
    1178           0 :         tmp = RMTgetId(ident, sym->def, getInstrPtr(sym->def, 0), 0);
    1179           0 :         if (tmp != MAL_SUCCEED) {
    1180           0 :                 MT_lock_unset(&c->lock);
    1181           0 :                 return tmp;
    1182             :         }
    1183             : 
    1184             :         /* check remote definition */
    1185           0 :         snprintf(buf, BUFSIZ, "b:bit:=inspect.getExistence(\"%s\",\"%s\");\nio.print(b);", mod, ident);
    1186           0 :         TRC_DEBUG(MAL_REMOTE, "Remote register: %s - %s\n", c->name, buf);
    1187           0 :         if ((msg = RMTquery(&mhdl, "remote.register", c->mconn, buf)) != MAL_SUCCEED){
    1188           0 :                 MT_lock_unset(&c->lock);
    1189           0 :                 return msg;
    1190             :         }
    1191             : 
    1192             :         char* result;
    1193           0 :         if ( mapi_get_field_count(mhdl) && mapi_fetch_row(mhdl) && (result = mapi_fetch_field(mhdl, 0))) {
    1194           0 :                 if (strcmp(result, "false") != 0)
    1195           0 :                         msg = createException(MAL, "remote.register",
    1196             :                                         "function already exists at the remote site: %s.%s",
    1197             :                                         mod, fcn);
    1198             :         }
    1199             :         else
    1200           0 :                 msg = createException(MAL, "remote.register", OPERATION_FAILED);
    1201             : 
    1202           0 :         mapi_close_handle(mhdl);
    1203             : 
    1204           0 :         if (msg) {
    1205           0 :                 MT_lock_unset(&c->lock);
    1206           0 :                 return msg;
    1207             :         }
    1208             : 
    1209           0 :         *fcn_id = GDKstrdup(ident);
    1210           0 :         if (*fcn_id == NULL) {
    1211           0 :                 MT_lock_unset(&c->lock);
    1212           0 :                 throw(MAL, "Remote register", MAL_MALLOC_FAIL);
    1213             :         }
    1214             : 
    1215             :         Symbol prg;
    1216           0 :         if ((prg = newFunction(putName(mod), putName(*fcn_id), FUNCTIONsymbol)) == NULL) {
    1217           0 :                 MT_lock_unset(&c->lock);
    1218           0 :                 throw(MAL, "Remote register", MAL_MALLOC_FAIL);
    1219             :         }
    1220             : 
    1221             :         // We only need the Symbol not the inner program stub. So we clear it.
    1222           0 :         freeMalBlk(prg->def);
    1223           0 :         prg->def = NULL;
    1224             : 
    1225           0 :         if ((prg->def = copyMalBlk(sym->def)) == NULL) {
    1226           0 :                 MT_lock_unset(&c->lock);
    1227           0 :                 freeSymbol(prg);
    1228           0 :                 throw(MAL, "Remote register", MAL_MALLOC_FAIL);
    1229             :         }
    1230           0 :         setFunctionId(getInstrPtr(prg->def, 0), putName(*fcn_id));
    1231             : 
    1232             :         /* make sure the program is error free */
    1233           0 :         msg = chkProgram(cntxt->usermodule, prg->def);
    1234           0 :         if ( msg != MAL_SUCCEED || prg->def->errors) {
    1235           0 :                 MT_lock_unset(&c->lock);
    1236           0 :                 if (msg)
    1237             :                         return msg;
    1238           0 :                 throw(MAL, "remote.register",
    1239             :                                 "function '%s.%s' contains syntax or type errors",
    1240             :                                 mod, *fcn_id);
    1241             :         }
    1242             : 
    1243           0 :         qry = mal2str(prg->def, 0, prg->def->stop);
    1244           0 :         TRC_DEBUG(MAL_REMOTE, "Remote register: %s - %s\n", c->name, qry);
    1245           0 :         msg = RMTquery(&mhdl, "remote.register", c->mconn, qry);
    1246           0 :         GDKfree(qry);
    1247           0 :         if (mhdl)
    1248           0 :                 mapi_close_handle(mhdl);
    1249             : 
    1250           0 :         freeSymbol(prg);
    1251             : 
    1252           0 :         MT_lock_unset(&c->lock);
    1253           0 :         return msg;
    1254             : }
    1255             : 
    1256           0 : static str RMTregister(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci) {
    1257           0 :         char **fcn_id = getArgReference_str(stk, pci, 0);
    1258           0 :         const char *conn = *getArgReference_str(stk, pci, 1);
    1259           0 :         const char *mod = *getArgReference_str(stk, pci, 2);
    1260           0 :         const char *fcn = *getArgReference_str(stk, pci, 3);
    1261             :         (void)mb;
    1262           0 :         return RMTregisterInternal(cntxt, fcn_id, conn, mod, fcn);
    1263             : }
    1264             : 
    1265             : /**
    1266             :  * exec executes the function with its given arguments on the remote
    1267             :  * host, returning the function's return value.  exec is purposely kept
    1268             :  * very spartan.  All arguments need to be handles to previously put()
    1269             :  * values.  It calls the function with the given arguments at the remote
    1270             :  * site, and returns the handle which stores the return value of the
    1271             :  * remotely executed function.  This return value can be retrieved using
    1272             :  * a get call. It handles multiple return arguments.
    1273             :  */
    1274         491 : static str RMTexec(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci) {
    1275             :         str conn, mod, func, tmp;
    1276             :         int i;
    1277             :         size_t len, buflen;
    1278         491 :         connection c= NULL;
    1279             :         char *qbuf;
    1280             :         MapiHdl mhdl;
    1281             : 
    1282             :         (void)cntxt;
    1283             :         (void)mb;
    1284             :         bool no_return_arguments = 0;
    1285             : 
    1286             :         columnar_result_callback* rcb = NULL;
    1287         491 :         ValRecord *v = &(stk)->stk[(pci)->argv[4]];
    1288         491 :         if (pci->retc == 1 && (pci->argc >= 4) && (v->vtype == TYPE_ptr) ) {
    1289           1 :                 rcb = (columnar_result_callback*) v->val.pval;
    1290             :         }
    1291             : 
    1292        2129 :         for (i = 0; i < pci->retc; i++) {
    1293        1638 :                 if (stk->stk[pci->argv[i]].vtype == TYPE_str) {
    1294        1637 :                         tmp = *getArgReference_str(stk, pci, i);
    1295        1637 :                         if (tmp == NULL || strcmp(tmp, (str)str_nil) == 0)
    1296           0 :                                 throw(ILLARG, "remote.exec", ILLEGAL_ARGUMENT
    1297             :                                                 ": return value %d is NULL or nil", i);
    1298             :                 }
    1299             :                 else
    1300             :                         no_return_arguments = 1;
    1301             :         }
    1302             : 
    1303         491 :         conn = *getArgReference_str(stk, pci, i++);
    1304         491 :         if (conn == NULL || strcmp(conn, (str)str_nil) == 0)
    1305           0 :                 throw(ILLARG, "remote.exec", ILLEGAL_ARGUMENT ": connection name is NULL or nil");
    1306         491 :         mod = *getArgReference_str(stk, pci, i++);
    1307         491 :         if (mod == NULL || strcmp(mod, (str)str_nil) == 0)
    1308           0 :                 throw(ILLARG, "remote.exec", ILLEGAL_ARGUMENT ": module name is NULL or nil");
    1309         491 :         func = *getArgReference_str(stk, pci, i++);
    1310         491 :         if (func == NULL || strcmp(func, (str)str_nil) == 0)
    1311           0 :                 throw(ILLARG, "remote.exec", ILLEGAL_ARGUMENT ": function name is NULL or nil");
    1312             : 
    1313             :         /* lookup conn */
    1314         491 :         rethrow("remote.exec", tmp, RMTfindconn(&c, conn));
    1315             : 
    1316             :         /* this call should be a single transaction over the channel*/
    1317         491 :         MT_lock_set(&c->lock);
    1318             : 
    1319         491 :         if(!no_return_arguments && pci->argc - pci->retc < 3) { /* conn, mod, func, ... */
    1320           0 :                 MT_lock_unset(&c->lock);
    1321           0 :                 throw(MAL, "remote.exec", ILLEGAL_ARGUMENT  " MAL instruction misses arguments");
    1322             :         }
    1323             : 
    1324             :         len = 0;
    1325             :         /* count how big a buffer we need */
    1326         491 :         len += 2 * (pci->retc > 1);
    1327         491 :         if (!no_return_arguments)
    1328        2127 :                 for (i = 0; i < pci->retc; i++) {
    1329        1637 :                         len += 2 * (i > 0);
    1330        1637 :                         len += strlen(*getArgReference_str(stk, pci, i));
    1331             :                 }
    1332             : 
    1333         491 :         const int arg_index = rcb ? 4 : 3;
    1334             : 
    1335         491 :         len += strlen(mod) + strlen(func) + 6;
    1336        1389 :         for (i = arg_index; i < pci->argc - pci->retc; i++) {
    1337         898 :                 len += 2 * (i > arg_index);
    1338         898 :                 len += strlen(*getArgReference_str(stk, pci, pci->retc + i));
    1339             :         }
    1340             :         len += 2;
    1341         491 :         buflen = len + 1;
    1342         491 :         if ((qbuf = GDKmalloc(buflen)) == NULL) {
    1343           0 :                 MT_lock_unset(&c->lock);
    1344           0 :                 throw(MAL, "remote.exec", SQLSTATE(HY013) MAL_MALLOC_FAIL);
    1345             :         }
    1346             : 
    1347             :         len = 0;
    1348             : 
    1349         491 :         if (pci->retc > 1)
    1350          42 :                 qbuf[len++] = '(';
    1351         491 :         if (!no_return_arguments)
    1352        2127 :                 for (i = 0; i < pci->retc; i++)
    1353        2127 :                         len += snprintf(&qbuf[len], buflen - len, "%s%s",
    1354        1637 :                                         (i > 0 ? ", " : ""), *getArgReference_str(stk, pci, i));
    1355             : 
    1356         491 :         if (pci->retc > 1)
    1357          42 :                 qbuf[len++] = ')';
    1358             : 
    1359             :         /* build the function invocation string in qbuf */
    1360         491 :         if (!no_return_arguments && pci->retc > 0) {
    1361         490 :                 len += snprintf(&qbuf[len], buflen - len, " := %s.%s(", mod, func);
    1362             :         }
    1363             :         else {
    1364           1 :                 len += snprintf(&qbuf[len], buflen - len, " %s.%s(", mod, func);
    1365             :         }
    1366             : 
    1367             :         /* handle the arguments to the function */
    1368             : 
    1369             :         /* put the arguments one by one, and dynamically build the
    1370             :          * invocation string */
    1371        1389 :         for (i = arg_index; i < pci->argc - pci->retc; i++) {
    1372        1158 :                 len += snprintf(&qbuf[len], buflen - len, "%s%s",
    1373             :                                 (i > arg_index ? ", " : ""),
    1374         898 :                                 *(getArgReference_str(stk, pci, pci->retc + i)));
    1375             :         }
    1376             : 
    1377             :         /* finish end execute the invocation string */
    1378         491 :         len += snprintf(&qbuf[len], buflen - len, ");");
    1379         491 :         TRC_DEBUG(MAL_REMOTE, "Remote exec: %s - %s\n", c->name, qbuf);
    1380         491 :         tmp = RMTquery(&mhdl, "remote.exec", c->mconn, qbuf);
    1381         491 :         GDKfree(qbuf);
    1382             : 
    1383             :         /* Temporary hack:
    1384             :          * use a callback to immediately handle columnar results before hdl is destroyed. */
    1385         491 :         if (tmp == MAL_SUCCEED && rcb && mhdl && (mapi_get_querytype(mhdl) == Q_TABLE || mapi_get_querytype(mhdl) == Q_PREPARE)) {
    1386           1 :                 int fields = mapi_get_field_count(mhdl);
    1387           1 :                 columnar_result* results = GDKzalloc(sizeof(columnar_result) * fields);
    1388             : 
    1389           1 :                 if (!results) {
    1390           0 :                         tmp = createException(MAL, "remote.exec", SQLSTATE(HY013) MAL_MALLOC_FAIL);
    1391             :                 } else {
    1392             :                         int i = 0;
    1393           1 :                         char buf[256] = {0};
    1394           1 :                         stream *sin = mapi_get_from(c->mconn);
    1395             : 
    1396           3 :                         for (; i < fields; i++) {
    1397           2 :                                 BAT *b = NULL;
    1398             : 
    1399           4 :                                 if ((tmp = RMTreadbatheader(sin, buf)) != MAL_SUCCEED ||
    1400           2 :                                         (tmp = RMTinternalcopyfrom(&b, buf, sin, i == fields - 1)) != MAL_SUCCEED) {
    1401             :                                         break;
    1402             :                                 }
    1403             : 
    1404           2 :                                 results[i].id = b->batCacheid;
    1405           2 :                                 results[i].colname = mapi_get_name(mhdl, i);
    1406           2 :                                 results[i].tpename = mapi_get_type(mhdl, i);
    1407           2 :                                 results[i].digits = mapi_get_digits(mhdl, i);
    1408           2 :                                 results[i].scale = mapi_get_scale(mhdl, i);
    1409             :                         }
    1410             : 
    1411           1 :                         if (tmp != MAL_SUCCEED) {
    1412           0 :                                 for (int j = 0; j < i; j++)
    1413           0 :                                         BBPunfix(results[j].id);
    1414             :                         } else {
    1415           3 :                                 for (int j = 0; j < i; j++)
    1416           2 :                                         BBPkeepref(results[j].id);
    1417           1 :                                 assert(rcb->context);
    1418           1 :                                 tmp = rcb->call(rcb->context, mapi_get_table(mhdl, 0), results, fields);
    1419           3 :                                 for (int j = 0; j < i; j++)
    1420           2 :                                         BBPrelease(results[j].id);
    1421             :                         }
    1422           1 :                         GDKfree(results);
    1423             :                 }
    1424             :         }
    1425             : 
    1426         491 :         if (rcb) {
    1427           1 :                 GDKfree(rcb->context);
    1428           1 :                 GDKfree(rcb);
    1429             :         }
    1430             : 
    1431         491 :         if (mhdl)
    1432         483 :                 mapi_close_handle(mhdl);
    1433         491 :         MT_lock_unset(&c->lock);
    1434         491 :         return tmp;
    1435             : }
    1436             : 
    1437             : /**
    1438             :  * batload is a helper function to make transferring a BAT with RMTput
    1439             :  * more efficient.  It works by creating a BAT, and loading it with the
    1440             :  * data as comma separated values from the input stream, until an empty
    1441             :  * line is read.  The given size argument is taken as a hint only, and
    1442             :  * is not enforced to match the number of rows read.
    1443             :  */
    1444        1266 : static str RMTbatload(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci) {
    1445             :         ValPtr v;
    1446             :         int t;
    1447             :         int size;
    1448             :         ptr  r;
    1449             :         size_t s;
    1450             :         BAT *b;
    1451             :         size_t len;
    1452             :         char *var;
    1453             :         str msg = MAL_SUCCEED;
    1454        1266 :         bstream *fdin = cntxt->fdin;
    1455             : 
    1456        1266 :         v = &stk->stk[pci->argv[0]]; /* return */
    1457        1266 :         t = getArgType(mb, pci, 1); /* tail type */
    1458        1266 :         size = *getArgReference_int(stk, pci, 2); /* size */
    1459             : 
    1460        1266 :         b = COLnew(0, t, size, TRANSIENT);
    1461        1266 :         if (b == NULL)
    1462           0 :                 throw(MAL, "remote.load", SQLSTATE(HY013) MAL_MALLOC_FAIL);
    1463             : 
    1464             :         /* grab the input stream and start reading */
    1465        1266 :         fdin->eof = false;
    1466        1266 :         len = fdin->pos;
    1467        1266 :         while (len < fdin->len || bstream_next(fdin) > 0) {
    1468             :                 /* newline hunting (how spartan) */
    1469        1266 :                 for (len = fdin->pos; len < fdin->len && fdin->buf[len] != '\n'; len++)
    1470             :                         ;
    1471             :                 /* unterminated line, request more */
    1472        1266 :                 if (fdin->buf[len] != '\n')
    1473           0 :                         continue;
    1474             :                 /* empty line, end of input */
    1475        1266 :                 if (fdin->pos == len) {
    1476        1266 :                         if (isa_block_stream(fdin->s)) {
    1477        1266 :                                 ssize_t n = bstream_next(fdin);
    1478        1266 :                                 if( n )
    1479           0 :                                         msg = createException(MAL, "remote.load", SQLSTATE(HY013) "Unexpected return from remote");
    1480             :                         }
    1481             :                         break;
    1482             :                 }
    1483           0 :                 fdin->buf[len] = '\0'; /* kill \n */
    1484           0 :                 var = &fdin->buf[fdin->pos];
    1485             :                 /* skip over this line */
    1486           0 :                 fdin->pos = ++len;
    1487             : 
    1488           0 :                 s = 0;
    1489           0 :                 r = NULL;
    1490           0 :                 if (ATOMfromstr(t, &r, &s, var, true) < 0 ||
    1491           0 :                         BUNappend(b, r, false) != GDK_SUCCEED) {
    1492           0 :                         BBPreclaim(b);
    1493           0 :                         GDKfree(r);
    1494           0 :                         throw(MAL, "remote.get", GDK_EXCEPTION);
    1495             :                 }
    1496           0 :                 GDKfree(r);
    1497             :         }
    1498             : 
    1499        1266 :         v->val.bval = b->batCacheid;
    1500        1266 :         v->vtype = TYPE_bat;
    1501        1266 :         BBPkeepref(b->batCacheid);
    1502             : 
    1503        1266 :         return msg;
    1504             : }
    1505             : 
    1506             : /**
    1507             :  * dump given BAT to stream
    1508             :  */
    1509        1265 : static str RMTbincopyto(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
    1510             : {
    1511        1265 :         bat bid = *getArgReference_bat(stk, pci, 1);
    1512        1265 :         BAT *b = BBPquickdesc(bid), *v = b;
    1513             :         char sendtheap = 0, sendtvheap = 0;
    1514             : 
    1515             :         (void)mb;
    1516             :         (void)stk;
    1517             :         (void)pci;
    1518             : 
    1519        1265 :         if (b == NULL)
    1520           0 :                 throw(MAL, "remote.bincopyto", RUNTIME_OBJECT_UNDEFINED);
    1521             : 
    1522        1265 :         if (BBPfix(bid) <= 0)
    1523           0 :                 throw(MAL, "remote.bincopyto", MAL_MALLOC_FAIL);
    1524             : 
    1525        1265 :         sendtheap = b->ttype != TYPE_void;
    1526        1265 :         sendtvheap = sendtheap && b->tvarsized;
    1527        1265 :         if (isVIEW(b) && sendtvheap && VIEWvtparent(b) && BATcount(b) < BATcount(BBP_cache(VIEWvtparent(b)))) {
    1528          18 :                 if ((b = BATdescriptor(bid)) == NULL) {
    1529           0 :                         BBPunfix(bid);
    1530           0 :                         throw(MAL, "remote.bincopyto", RUNTIME_OBJECT_MISSING);
    1531             :                 }
    1532          18 :                 v = COLcopy(b, b->ttype, true, TRANSIENT);
    1533          18 :                 BBPunfix(b->batCacheid);
    1534          18 :                 if (v == NULL) {
    1535           0 :                         BBPunfix(bid);
    1536           0 :                         throw(MAL, "remote.bincopyto", GDK_EXCEPTION);
    1537             :                 }
    1538             :         }
    1539             : 
    1540        1317 :         mnstr_printf(cntxt->fdout, /*JSON*/"{"
    1541             :                         "\"version\":1,"
    1542             :                         "\"ttype\":%d,"
    1543             :                         "\"hseqbase\":" OIDFMT ","
    1544             :                         "\"tseqbase\":" OIDFMT ","
    1545             :                         "\"tsorted\":%d,"
    1546             :                         "\"trevsorted\":%d,"
    1547             :                         "\"tkey\":%d,"
    1548             :                         "\"tnonil\":%d,"
    1549             :                         "\"tdense\":%d,"
    1550             :                         "\"size\":" BUNFMT ","
    1551             :                         "\"tailsize\":%zu,"
    1552             :                         "\"theapsize\":%zu"
    1553             :                         "}\n",
    1554        1265 :                         v->ttype,
    1555             :                         v->hseqbase, v->tseqbase,
    1556        1265 :                         v->tsorted, v->trevsorted,
    1557        1265 :                         v->tkey,
    1558        1265 :                         v->tnonil,
    1559        1265 :                         BATtdense(v),
    1560             :                         v->batCount,
    1561        1254 :                         sendtheap ? (size_t)v->batCount << v->tshift : 0,
    1562          52 :                         sendtvheap && v->batCount > 0 ? v->tvheap->free : 0
    1563             :                         );
    1564             : 
    1565        1265 :         if (sendtheap && v->batCount > 0) {
    1566        1181 :                 BATiter vi = bat_iterator(v);
    1567        1181 :                 mnstr_write(cntxt->fdout, /* tail */
    1568        1181 :                                         vi.base, vi.count * vi.width, 1);
    1569        1181 :                 if (sendtvheap)
    1570          40 :                         mnstr_write(cntxt->fdout, /* theap */
    1571          40 :                                                 vi.vh->base, vi.vhfree, 1);
    1572        1181 :                 bat_iterator_end(&vi);
    1573             :         }
    1574             :         /* flush is done by the calling environment (MAL) */
    1575             : 
    1576        1265 :         if (b != v)
    1577          18 :                 BBPreclaim(v);
    1578             : 
    1579        1265 :         BBPunfix(bid);
    1580             : 
    1581        1265 :         return(MAL_SUCCEED);
    1582             : }
    1583             : 
    1584             : /**
    1585             :  * read from the input stream and give the BAT handle back to the caller
    1586             :  */
    1587           0 : static str RMTbincopyfrom(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci) {
    1588           0 :         BAT *b = NULL;
    1589             :         ValPtr v;
    1590             :         str err;
    1591             : 
    1592             :         (void)mb;
    1593             : 
    1594             :         /* We receive a normal line, which contains the JSON header, the
    1595             :          * rest is binary data directly on the stream.  We get the first
    1596             :          * line from the buffered stream we have here, and pass it on
    1597             :          * together with the raw stream we have. */
    1598           0 :         cntxt->fdin->eof = false; /* in case it was before */
    1599           0 :         if (bstream_next(cntxt->fdin) <= 0)
    1600           0 :                 throw(MAL, "remote.bincopyfrom", "expected JSON header");
    1601             : 
    1602           0 :         cntxt->fdin->buf[cntxt->fdin->len] = '\0';
    1603           0 :         err = RMTinternalcopyfrom(&b,
    1604           0 :                         &cntxt->fdin->buf[cntxt->fdin->pos], cntxt->fdin->s, true);
    1605             :         /* skip the JSON line */
    1606           0 :         cntxt->fdin->pos = ++cntxt->fdin->len;
    1607           0 :         if (err != MAL_SUCCEED)
    1608             :                 return(err);
    1609             : 
    1610           0 :         v = &stk->stk[pci->argv[0]];
    1611           0 :         v->val.bval = b->batCacheid;
    1612           0 :         v->vtype = TYPE_bat;
    1613           0 :         BBPkeepref(b->batCacheid);
    1614             : 
    1615           0 :         return(MAL_SUCCEED);
    1616             : }
    1617             : 
    1618             : /**
    1619             :  * bintype identifies the system on its binary profile.  This is mainly
    1620             :  * used to determine if BATs can be sent binary across.
    1621             :  */
    1622         127 : static str RMTbintype(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci) {
    1623             :         int type = 0;
    1624             :         (void)mb;
    1625             :         (void)stk;
    1626             :         (void)pci;
    1627             : 
    1628             : #ifdef WORDS_BIGENDIAN
    1629             :         type |= RMTT_B_ENDIAN;
    1630             : #else
    1631             :         type |= RMTT_L_ENDIAN;
    1632             : #endif
    1633             : #if SIZEOF_SIZE_T == SIZEOF_LNG
    1634             :         type |= RMTT_64_BITS;
    1635             : #else
    1636             :         type |= RMTT_32_BITS;
    1637             : #endif
    1638             : #if SIZEOF_OID == SIZEOF_LNG
    1639             :         type |= RMTT_64_OIDS;
    1640             : #else
    1641             :         type |= RMTT_32_OIDS;
    1642             : #endif
    1643             : 
    1644         127 :         mnstr_printf(cntxt->fdout, "[ %d ]\n", type);
    1645             : 
    1646         127 :         return(MAL_SUCCEED);
    1647             : }
    1648             : 
    1649             : /**
    1650             :  * Returns whether the underlying connection is still connected or not.
    1651             :  * Best effort implementation on top of mapi using a ping.
    1652             :  */
    1653             : static str
    1654           0 : RMTisalive(int *ret, str *conn)
    1655             : {
    1656             :         str tmp;
    1657             :         connection c;
    1658             : 
    1659           0 :         if (*conn == NULL || strcmp(*conn, (str)str_nil) == 0)
    1660           0 :                 throw(ILLARG, "remote.get", ILLEGAL_ARGUMENT ": connection name is NULL or nil");
    1661             : 
    1662             :         /* lookup conn, set c if valid */
    1663           0 :         rethrow("remote.get", tmp, RMTfindconn(&c, *conn));
    1664             : 
    1665           0 :         *ret = 0;
    1666           0 :         if (mapi_is_connected(c->mconn) && mapi_ping(c->mconn) == MOK)
    1667           0 :                 *ret = 1;
    1668             : 
    1669             :         return MAL_SUCCEED;
    1670             : }
    1671             : 
    1672             : // This is basically a no op
    1673             : static str
    1674         238 : RMTregisterSupervisor(int *ret, str *sup_uuid, str *query_uuid) {
    1675             :         (void)sup_uuid;
    1676             :         (void)query_uuid;
    1677             : 
    1678         238 :         *ret = 0;
    1679         238 :         return MAL_SUCCEED;
    1680             : }
    1681             : 
    1682             : /* this is needed in remote plans */
    1683             : static str
    1684           8 : RMTassert(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
    1685             : {
    1686           8 :         bool flg = (bool) *getArgReference_bit(stk, pci, 1);
    1687           8 :         str msg = *getArgReference_str(stk, pci, 2);
    1688             : 
    1689             :         (void) cntxt;
    1690             :         (void) mb;
    1691           8 :         if (flg) {
    1692           8 :                 if (strlen(msg) > 6 &&
    1693           8 :                     msg[5] == '!' &&
    1694           0 :                     (isdigit((unsigned char) msg[0]) ||
    1695           0 :                      isupper((unsigned char) msg[0])) &&
    1696           0 :                     (isdigit((unsigned char) msg[1]) ||
    1697           0 :                      isupper((unsigned char) msg[1])) &&
    1698           0 :                     (isdigit((unsigned char) msg[2]) ||
    1699           0 :                      isupper((unsigned char) msg[2])) &&
    1700           0 :                     (isdigit((unsigned char) msg[3]) ||
    1701           0 :                      isupper((unsigned char) msg[3])) &&
    1702           0 :                     (isdigit((unsigned char) msg[4]) ||
    1703             :                      isupper((unsigned char) msg[4])))
    1704           0 :                         throw(REMOTE, "assert", "%s", msg); /* includes state */
    1705           8 :                 throw(REMOTE, "assert", SQLSTATE(M0M29) "%s", msg);
    1706             :         }
    1707             :         return MAL_SUCCEED;
    1708             : }
    1709             : 
    1710             : #include "mel.h"
    1711             : mel_func remote_init_funcs[] = {
    1712             :  command("remote", "prelude", RMTprelude, false, "initialise the remote module", args(1,1, arg("",void))),
    1713             :  command("remote", "epilogue", RMTepilogue, false, "release the resources held by the remote module", args(1,1, arg("",void))),
    1714             :  pattern("remote", "assert", RMTassert, false, "Generate an exception when b==true", args(1,3, arg("",void),arg("b",bit),arg("msg",str))),
    1715             :  command("remote", "resolve", RMTresolve, false, "resolve a pattern against Merovingian and return the URIs", args(1,2, batarg("",str),arg("pattern",str))),
    1716             :  pattern("remote", "connect", RMTconnect, false, "returns a newly created connection for uri, using user name and password", args(1,5, arg("",str),arg("uri",str),arg("user",str),arg("passwd",str),arg("scen",str))),
    1717             :  command("remote", "connect", RMTconnectScen, false, "returns a newly created connection for uri, using user name, password and scenario", args(1,6, arg("",str),arg("uri",str),arg("user",str),arg("passwd",str),arg("scen",str),arg("columnar",bit))),
    1718             :  pattern("remote", "connect", RMTconnectTable, false, "return a newly created connection for a table. username and password should be in the vault", args(1,3, arg("",str),arg("table",str),arg("schen",str))),
    1719             :  command("remote", "disconnect", RMTdisconnect, false, "disconnects the connection pointed to by handle (received from a call to connect()", args(1,2, arg("",void),arg("conn",str))),
    1720             :  pattern("remote", "get", RMTget, false, "retrieves a copy of remote object ident", args(1,3, argany("",0),arg("conn",str),arg("ident",str))),
    1721             :  pattern("remote", "put", RMTput, false, "copies object to the remote site and returns its identifier", args(1,3, arg("",str),arg("conn",str),argany("object",0))),
    1722             :  pattern("remote", "register", RMTregister, false, "register <mod>.<fcn> at the remote site", args(1,4, arg("",str),arg("conn",str),arg("mod",str),arg("fcn",str))),
    1723             :  pattern("remote", "exec", RMTexec, false, "remotely executes <mod>.<func> and returns the handle to its result", args(1,4, arg("",str),arg("conn",str),arg("mod",str),arg("func",str))),
    1724             :  pattern("remote", "exec", RMTexec, false, "remotely executes <mod>.<func> and returns the handle to its result", args(1,4, vararg("",str),arg("conn",str),arg("mod",str),arg("func",str))),
    1725             :  pattern("remote", "exec", RMTexec, false, "remotely executes <mod>.<func> using the argument list of remote objects and returns the handle to its result", args(1,5, arg("",str),arg("conn",str),arg("mod",str),arg("func",str),vararg("",str))),
    1726             :  pattern("remote", "exec", RMTexec, false, "remotely executes <mod>.<func> using the argument list of remote objects and returns the handle to its result", args(1,5, vararg("",str),arg("conn",str),arg("mod",str),arg("func",str),vararg("",str))),
    1727             :  pattern("remote", "exec", RMTexec, false, "remotely executes <mod>.<func> using the argument list of remote objects and applying function pointer rcb as callback to handle any results.", args(0,5, arg("conn",str),arg("mod",str),arg("func",str),arg("rcb",ptr), vararg("",str))),
    1728             :  pattern("remote", "exec", RMTexec, false, "remotely executes <mod>.<func> using the argument list of remote objects and ignoring results.", args(0,4, arg("conn",str),arg("mod",str),arg("func",str), vararg("",str))),
    1729             :  command("remote", "isalive", RMTisalive, false, "check if conn is still valid and connected", args(1,2, arg("",int),arg("conn",str))),
    1730             :  pattern("remote", "batload", RMTbatload, false, "create a BAT of the given type and size, and load values from the input stream", args(1,3, batargany("",1),argany("tt",1),arg("size",int))),
    1731             :  pattern("remote", "batbincopy", RMTbincopyto, false, "dump BAT b in binary form to the stream", args(1,2, arg("",void),batargany("b",0))),
    1732             :  pattern("remote", "batbincopy", RMTbincopyfrom, false, "store the binary BAT data in the BBP and return as BAT", args(1,1, batargany("",0))),
    1733             :  pattern("remote", "bintype", RMTbintype, false, "print the binary type of this mserver5", args(1,1, arg("",void))),
    1734             :  command("remote", "register_supervisor", RMTregisterSupervisor, false, "Register the supervisor uuid at a remote site", args(1,3, arg("",int),arg("sup_uuid",str),arg("query_uuid",str))),
    1735             :  { .imp=NULL }
    1736             : };
    1737             : #include "mal_import.h"
    1738             : #ifdef _MSC_VER
    1739             : #undef read
    1740             : #pragma section(".CRT$XCU",read)
    1741             : #endif
    1742         259 : LIB_STARTUP_FUNC(init_remote_mal)
    1743         259 : { mal_module("remote", NULL, remote_init_funcs); }
    1744             : 
    1745             : #endif

Generated by: LCOV version 1.14