1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903 |
- /*##############################################################################
- HPCC SYSTEMS software Copyright (C) 2015 HPCC Systems®.
- Licensed under the Apache License, Version 2.0 (the "License");
- you may not use this file except in compliance with the License.
- You may obtain a copy of the License at
- http://www.apache.org/licenses/LICENSE-2.0
- Unless required by applicable law or agreed to in writing, software
- distributed under the License is distributed on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- See the License for the specific language governing permissions and
- limitations under the License.
- ############################################################################## */
- #include "platform.h"
- #include "cassandra.h"
- #include "jexcept.hpp"
- #include "jthread.hpp"
- #include "hqlplugins.hpp"
- #include "deftype.hpp"
- #include "eclhelper.hpp"
- #include "eclrtl.hpp"
- #include "eclrtl_imp.hpp"
- #include "rtlds_imp.hpp"
- #include "rtlfield_imp.hpp"
- #include "rtlembed.hpp"
- #include "roxiemem.hpp"
- #include "nbcd.hpp"
- #include "cassandraembed.hpp"
- #ifdef _WIN32
- #define EXPORT __declspec(dllexport)
- #else
- #define EXPORT
- #endif
- static const char * compatibleVersions[] = {
- "Cassandra Embed Helper 1.0.0",
- NULL };
- static const char *version = "Cassandra Embed Helper 1.0.0";
- extern "C" EXPORT bool getECLPluginDefinition(ECLPluginDefinitionBlock *pb)
- {
- if (pb->size == sizeof(ECLPluginDefinitionBlockEx))
- {
- ECLPluginDefinitionBlockEx * pbx = (ECLPluginDefinitionBlockEx *) pb;
- pbx->compatibleVersions = compatibleVersions;
- }
- else if (pb->size != sizeof(ECLPluginDefinitionBlock))
- return false;
- pb->magicVersion = PLUGIN_VERSION;
- pb->version = version;
- pb->moduleName = "cassandra";
- pb->ECL = NULL;
- pb->flags = PLUGIN_MULTIPLE_VERSIONS;
- pb->description = "Cassandra Embed Helper";
- return true;
- }
- namespace cassandraembed {
- extern void UNSUPPORTED(const char *feature)
- {
- throw MakeStringException(-1, "UNSUPPORTED feature: %s not supported in Cassandra plugin", feature);
- }
- static void logCallBack(const CassLogMessage *message, void *data)
- {
- DBGLOG("cassandra: %s - %s", cass_log_level_string(message->severity), message->message);
- }
- extern void failx(const char *message, ...)
- {
- va_list args;
- va_start(args,message);
- StringBuffer msg;
- msg.append("cassandra: ").valist_appendf(message,args);
- va_end(args);
- rtlFail(0, msg.str());
- }
- extern void fail(const char *message)
- {
- StringBuffer msg;
- msg.append("cassandra: ").append(message);
- rtlFail(0, msg.str());
- }
- void check(CassError rc)
- {
- if (rc != CASS_OK)
- {
- fail(cass_error_desc(rc));
- }
- }
- // Wrappers to Cassandra structures that require corresponding releases
- void CassandraClusterSession::setOptions(const StringArray &options)
- {
- const char *contact_points = "localhost";
- const char *user = "";
- const char *password = "";
- ForEachItemIn(idx, options)
- {
- const char *opt = options.item(idx);
- const char *val = strchr(opt, '=');
- if (val)
- {
- StringBuffer optName(val-opt, opt);
- val++;
- if (stricmp(optName, "contact_points")==0 || stricmp(optName, "server")==0)
- contact_points = val; // Note that lifetime of val is adequate for this to be safe
- else if (stricmp(optName, "user")==0)
- user = val;
- else if (stricmp(optName, "password")==0)
- password = val;
- else if (stricmp(optName, "keyspace")==0)
- keyspace.set(val);
- else if (stricmp(optName, "maxFutures")==0)
- {
- if (!semaphore)
- {
- maxFutures=getUnsignedOption(val, "maxFutures");
- if (maxFutures)
- semaphore = new Semaphore(maxFutures);
- }
- }
- else if (stricmp(optName, "maxRetries")==0)
- maxRetries=getUnsignedOption(val, "maxRetries");
- else if (stricmp(optName, "port")==0)
- {
- unsigned port = getUnsignedOption(val, "port");
- checkSetOption(cass_cluster_set_port(cluster, port), "port");
- }
- else if (stricmp(optName, "protocol_version")==0)
- {
- unsigned protocol_version = getUnsignedOption(val, "protocol_version");
- checkSetOption(cass_cluster_set_protocol_version(cluster, protocol_version), "protocol_version");
- }
- else if (stricmp(optName, "num_threads_io")==0)
- {
- unsigned num_threads_io = getUnsignedOption(val, "num_threads_io");
- cass_cluster_set_num_threads_io(cluster, num_threads_io); // No status return
- }
- else if (stricmp(optName, "queue_size_io")==0)
- {
- unsigned queue_size_io = getUnsignedOption(val, "queue_size_io");
- checkSetOption(cass_cluster_set_queue_size_io(cluster, queue_size_io), "queue_size_io");
- }
- else if (stricmp(optName, "core_connections_per_host")==0)
- {
- unsigned core_connections_per_host = getUnsignedOption(val, "core_connections_per_host");
- checkSetOption(cass_cluster_set_core_connections_per_host(cluster, core_connections_per_host), "core_connections_per_host");
- }
- else if (stricmp(optName, "max_connections_per_host")==0)
- {
- unsigned max_connections_per_host = getUnsignedOption(val, "max_connections_per_host");
- checkSetOption(cass_cluster_set_max_connections_per_host(cluster, max_connections_per_host), "max_connections_per_host");
- }
- else if (stricmp(optName, "max_concurrent_creation")==0)
- {
- unsigned max_concurrent_creation = getUnsignedOption(val, "max_concurrent_creation");
- checkSetOption(cass_cluster_set_max_concurrent_creation(cluster, max_concurrent_creation), "max_concurrent_creation");
- }
- else if (stricmp(optName, "pending_requests_high_water_mark")==0)
- {
- unsigned pending_requests_high_water_mark = getUnsignedOption(val, "pending_requests_high_water_mark");
- checkSetOption(cass_cluster_set_pending_requests_high_water_mark(cluster, pending_requests_high_water_mark), "pending_requests_high_water_mark");
- }
- else if (stricmp(optName, "pending_requests_low_water_mark")==0)
- {
- unsigned pending_requests_low_water_mark = getUnsignedOption(val, "pending_requests_low_water_mark");
- checkSetOption(cass_cluster_set_pending_requests_low_water_mark(cluster, pending_requests_low_water_mark), "pending_requests_low_water_mark");
- }
- else if (stricmp(optName, "max_concurrent_requests_threshold")==0)
- {
- unsigned max_concurrent_requests_threshold = getUnsignedOption(val, "max_concurrent_requests_threshold");
- checkSetOption(cass_cluster_set_max_concurrent_requests_threshold(cluster, max_concurrent_requests_threshold), "max_concurrent_requests_threshold");
- }
- else if (stricmp(optName, "connect_timeout")==0)
- {
- unsigned connect_timeout = getUnsignedOption(val, "connect_timeout");
- cass_cluster_set_connect_timeout(cluster, connect_timeout);
- }
- else if (stricmp(optName, "request_timeout")==0)
- {
- unsigned request_timeout = getUnsignedOption(val, "request_timeout");
- cass_cluster_set_request_timeout(cluster, request_timeout);
- }
- else if (stricmp(optName, "load_balance_round_robin")==0)
- {
- cass_bool_t enable = getBoolOption(val, "load_balance_round_robin");
- if (enable==cass_true)
- cass_cluster_set_load_balance_round_robin(cluster);
- }
- else if (stricmp(optName, "load_balance_dc_aware")==0)
- {
- StringArray lbargs;
- lbargs.appendList(val, "|");
- if (lbargs.length() != 3)
- failx("Invalid value '%s' for option %s - expected 3 subvalues (separate with |)", val, optName.str());
- unsigned usedPerRemote = getUnsignedOption(lbargs.item(2), "load_balance_dc_aware");
- cass_bool_t allowRemote = getBoolOption(lbargs.item(2), "load_balance_dc_aware");
- checkSetOption(cass_cluster_set_load_balance_dc_aware(cluster, lbargs.item(0), usedPerRemote, allowRemote), "load_balance_dc_aware");
- }
- else if (stricmp(optName, "token_aware_routing")==0)
- {
- cass_bool_t enable = getBoolOption(val, "token_aware_routing");
- cass_cluster_set_token_aware_routing(cluster, enable);
- }
- else if (stricmp(optName, "latency_aware_routing")==0)
- {
- cass_bool_t enable = getBoolOption(val, "latency_aware_routing");
- cass_cluster_set_latency_aware_routing(cluster, enable);
- }
- else if (stricmp(optName, "latency_aware_routing_settings")==0)
- {
- StringArray subargs;
- subargs.appendList(val, "|");
- if (subargs.length() != 5)
- failx("Invalid value '%s' for option %s - expected 5 subvalues (separate with |)", val, optName.str());
- cass_double_t exclusion_threshold = getDoubleOption(subargs.item(0), "exclusion_threshold");
- cass_uint64_t scale_ms = getUnsigned64Option(subargs.item(1), "scale_ms");
- cass_uint64_t retry_period_ms = getUnsigned64Option(subargs.item(2), "retry_period_ms");
- cass_uint64_t update_rate_ms = getUnsigned64Option(subargs.item(3), "update_rate_ms");
- cass_uint64_t min_measured = getUnsigned64Option(subargs.item(4), "min_measured");
- cass_cluster_set_latency_aware_routing_settings(cluster, exclusion_threshold, scale_ms, retry_period_ms, update_rate_ms, min_measured);
- }
- else if (stricmp(optName, "tcp_nodelay")==0)
- {
- cass_bool_t enable = getBoolOption(val, "tcp_nodelay");
- cass_cluster_set_tcp_nodelay(cluster, enable);
- }
- else if (stricmp(optName, "tcp_keepalive")==0)
- {
- StringArray subargs;
- subargs.appendList(val, "|");
- if (subargs.length() != 2)
- failx("Invalid value '%s' for option %s - expected 2 subvalues (separate with |)", val, optName.str());
- cass_bool_t enabled = getBoolOption(subargs.item(0), "enabled");
- unsigned delay_secs = getUnsignedOption(subargs.item(0), "delay_secs");
- cass_cluster_set_tcp_keepalive(cluster, enabled, delay_secs);
- }
- else
- failx("Unrecognized option %s", optName.str());
- }
- }
- cass_cluster_set_contact_points(cluster, contact_points);
- if (*user || *password)
- cass_cluster_set_credentials(cluster, user, password);
- }
- void CassandraClusterSession::checkSetOption(CassError rc, const char *name)
- {
- if (rc != CASS_OK)
- {
- failx("While setting option %s: %s", name, cass_error_desc(rc));
- }
- }
- cass_bool_t CassandraClusterSession::getBoolOption(const char *val, const char *option)
- {
- return strToBool(val) ? cass_true : cass_false;
- }
- unsigned CassandraClusterSession::getUnsignedOption(const char *val, const char *option)
- {
- char *endp;
- long value = strtoul(val, &endp, 0);
- if (endp==val || *endp != '\0' || value > UINT_MAX || value < 0)
- failx("Invalid value '%s' for option %s", val, option);
- return (unsigned) value;
- }
- unsigned CassandraClusterSession::getDoubleOption(const char *val, const char *option)
- {
- char *endp;
- double value = strtod(val, &endp);
- if (endp==val || *endp != '\0')
- failx("Invalid value '%s' for option %s", val, option);
- return value;
- }
- __uint64 CassandraClusterSession::getUnsigned64Option(const char *val, const char *option)
- {
- // MORE - could check it's all digits (with optional leading spaces...), if we cared.
- return rtlVStrToUInt8(val);
- }
- void CassandraClusterSession::connect()
- {
- assertex(cluster && !session);
- session.setown(new CassandraSession(cass_session_new()));
- CassandraFuture future(keyspace.isEmpty() ? cass_session_connect(*session, cluster) : cass_session_connect_keyspace(*session, cluster, keyspace));
- future.wait("connect");
- }
- void CassandraClusterSession::disconnect()
- {
- session.clear();
- }
- CassandraPrepared *CassandraClusterSession::prepareStatement(const char *query, bool trace) const
- {
- assertex(session);
- CriticalBlock b(cacheCrit);
- Linked<CassandraPrepared> cached = preparedCache.getValue(query);
- if (cached)
- return cached.getClear();
- {
- // We don't want to block cache lookups while we prepare a new bound statement
- // Note - if multiple threads try to prepare the same (new) statement at the same time, it's not catastrophic
- CriticalUnblock b(cacheCrit);
- CassandraFuture futurePrep(cass_session_prepare(*session, query));
- futurePrep.wait("prepare statement");
- cached.setown(new CassandraPrepared(cass_future_get_prepared(futurePrep), trace ? query : NULL));
- }
- preparedCache.setValue(query, cached); // NOTE - this links parameter
- return cached.getClear();
- }
- CassandraStatementInfo *CassandraClusterSession::createStatementInfo(const char *script, unsigned numParams, CassBatchType batchMode, unsigned pageSize) const
- {
- Owned<CassandraPrepared> prepared = prepareStatement(script, false); // We could make tracing selectable
- return new CassandraStatementInfo(session, prepared, numParams, batchMode, pageSize, semaphore, maxRetries);
- }
- typedef CassandraClusterSession *CassandraClusterSessionPtr;
- typedef MapBetween<hash64_t, hash64_t, CassandraClusterSessionPtr, CassandraClusterSessionPtr> ClusterSessionMap;
- static CriticalSection clusterCacheCrit;
- static ClusterSessionMap cachedSessions;
- CassandraClusterSession *lookupCachedSession(hash64_t hash, const StringArray &opts)
- {
- Owned<CassandraClusterSession> cluster;
- CassandraClusterSessionPtr *found = cachedSessions.getValue(hash);
- if (found)
- cluster.set(*found);
- if (!cluster)
- {
- cluster.setown(new CassandraClusterSession(cass_cluster_new()));
- cluster->setOptions(opts);
- cluster->connect();
- cachedSessions.setValue(hash, cluster.getLink());
- }
- return cluster.getClear();
- }
- MODULE_INIT(INIT_PRIORITY_STANDARD)
- {
- cass_log_set_callback(logCallBack, NULL);
- cass_log_set_level(CASS_LOG_WARN);
- return true;
- }
- MODULE_EXIT()
- {
- HashIterator i(cachedSessions);
- ForEach(i)
- {
- CassandraClusterSession *session = *cachedSessions.mapToValue(&i.query());
- ::Release(session);
- }
- }
- //------------------
- void CassandraFuture::wait(const char *why) const
- {
- cass_future_wait(future);
- CassError rc = cass_future_error_code(future);
- if(rc != CASS_OK)
- {
- const char *message;
- size_t length;
- cass_future_error_message(future, &message, &length);
- VStringBuffer err("cassandra: failed to %s (%.*s)", why, (int) length, message);
- #ifdef _DEBUG
- DBGLOG("%s", err.str());
- #endif
- rtlFail(0, err.str());
- }
- }
- void CassandraSession::set(CassSession *_session)
- {
- if (session)
- {
- CassandraFuture close_future(cass_session_close(session));
- cass_future_wait(close_future);
- cass_session_free(session);
- }
- session = _session;
- }
- //----------------------
- CassandraRetryingFuture::CassandraRetryingFuture(CassSession *_session, CassStatement *_statement, Semaphore *_limiter, unsigned _retries)
- : session(_session), statement(_statement), retries(_retries), limiter(_limiter), future(NULL)
- {
- execute();
- }
- CassandraRetryingFuture::~CassandraRetryingFuture()
- {
- if (future)
- cass_future_free(future);
- }
- void CassandraRetryingFuture::wait(const char *why)
- {
- cass_future_wait(future);
- CassError rc = cass_future_error_code(future);
- if(rc != CASS_OK)
- {
- switch (rc)
- {
- case CASS_ERROR_LIB_NO_HOSTS_AVAILABLE: // MORE - are there others we should retry?
- if (retry(why))
- break;
- // fall into
- default:
- const char *message;
- size_t length;
- cass_future_error_message(future, &message, &length);
- VStringBuffer err("cassandra: failed to %s (%.*s)", why, (int) length, message);
- rtlFail(0, err.str());
- }
- }
- }
- bool CassandraRetryingFuture::retry(const char *why)
- {
- for (int i = 0; i < retries; i++)
- {
- execute();
- cass_future_wait(future);
- CassError rc = cass_future_error_code(future);
- if (rc == CASS_OK)
- return true;
- Sleep(10);
- }
- return false;
- }
- void CassandraRetryingFuture::execute()
- {
- if (limiter)
- limiter->wait();
- future = cass_session_execute(session, statement);
- if (limiter)
- cass_future_set_callback(future, signaller, limiter); // Note - this will call the callback if the future has already completed
- }
- void CassandraRetryingFuture::signaller(CassFuture *future, void *data)
- {
- Semaphore *sem = (Semaphore *) data;
- sem->signal();
- }
- //----------------------
- CassandraStatementInfo::CassandraStatementInfo(CassandraSession *_session, CassandraPrepared *_prepared, unsigned _numBindings, CassBatchType _batchMode, unsigned pageSize, Semaphore *_semaphore, unsigned _maxRetries)
- : session(_session), prepared(_prepared), numBindings(_numBindings), batchMode(_batchMode), semaphore(_semaphore), maxRetries(_maxRetries)
- {
- assertex(prepared && *prepared);
- statement.setown(new CassandraStatement(cass_prepared_bind(*prepared)));
- if (pageSize)
- cass_statement_set_paging_size(*statement, pageSize);
- inBatch = false;
- }
- CassandraStatementInfo::~CassandraStatementInfo()
- {
- stop();
- futures.kill();
- }
- void CassandraStatementInfo::stop()
- {
- iterator.clear();
- result.clear();
- prepared.clear();
- }
- bool CassandraStatementInfo::next()
- {
- loop
- {
- if (!iterator)
- {
- if (result)
- iterator.setown(new CassandraIterator(cass_iterator_from_result(*result)));
- else
- return false;
- }
- if (cass_iterator_next(*iterator))
- return true;
- iterator.clear();
- if (!cass_result_has_more_pages(*result))
- {
- result.clear();
- break;
- }
- cass_statement_set_paging_state(*statement, *result);
- result.setown(new CassandraFutureResult(cass_session_execute(*session, *statement)));
- }
- return false;
- }
- void CassandraStatementInfo::startStream()
- {
- if (batchMode != (CassBatchType) -1)
- batch.setown(new CassandraBatch(cass_batch_new(batchMode)));
- statement.setown(new CassandraStatement(cass_prepared_bind(*prepared)));
- inBatch = true;
- }
- void CassandraStatementInfo::endStream()
- {
- if (batch)
- {
- result.setown(new CassandraFutureResult (cass_session_execute_batch(*session, *batch)));
- assertex (rowCount() == 0);
- }
- else
- {
- ForEachItemIn(idx, futures)
- {
- futures.item(idx).wait("endStream");
- }
- }
- }
- void CassandraStatementInfo::execute()
- {
- assertex(statement && *statement);
- if (batch)
- {
- check(cass_batch_add_statement(*batch, *statement));
- statement.setown(new CassandraStatement(cass_prepared_bind(*prepared)));
- }
- else if (inBatch)
- {
- futures.append(*new CassandraRetryingFuture(*session, statement->getClear(), semaphore, maxRetries));
- statement.setown(new CassandraStatement(cass_prepared_bind(*prepared)));
- }
- else
- {
- result.setown(new CassandraFutureResult(cass_session_execute(*session, *statement)));
- }
- }
- // Conversions from Cassandra values to ECL data
- static const char *getTypeName(CassValueType type)
- {
- switch (type)
- {
- case CASS_VALUE_TYPE_CUSTOM: return "CUSTOM";
- case CASS_VALUE_TYPE_ASCII: return "ASCII";
- case CASS_VALUE_TYPE_BIGINT: return "BIGINT";
- case CASS_VALUE_TYPE_BLOB: return "BLOB";
- case CASS_VALUE_TYPE_BOOLEAN: return "BOOLEAN";
- case CASS_VALUE_TYPE_COUNTER: return "COUNTER";
- case CASS_VALUE_TYPE_DECIMAL: return "DECIMAL";
- case CASS_VALUE_TYPE_DOUBLE: return "DOUBLE";
- case CASS_VALUE_TYPE_FLOAT: return "FLOAT";
- case CASS_VALUE_TYPE_INT: return "INT";
- case CASS_VALUE_TYPE_TEXT: return "TEXT";
- case CASS_VALUE_TYPE_TIMESTAMP: return "TIMESTAMP";
- case CASS_VALUE_TYPE_UUID: return "UUID";
- case CASS_VALUE_TYPE_VARCHAR: return "VARCHAR";
- case CASS_VALUE_TYPE_VARINT: return "VARINT";
- case CASS_VALUE_TYPE_TIMEUUID: return "TIMEUUID";
- case CASS_VALUE_TYPE_INET: return "INET";
- case CASS_VALUE_TYPE_LIST: return "LIST";
- case CASS_VALUE_TYPE_MAP: return "MAP";
- case CASS_VALUE_TYPE_SET: return "SET";
- default: return "UNKNOWN";
- }
- }
- static void typeError(const char *expected, const CassValue *value, const RtlFieldInfo *field) __attribute__((noreturn));
- static void typeError(const char *expected, const CassValue *value, const RtlFieldInfo *field)
- {
- VStringBuffer msg("cassandra: type mismatch - %s expected", expected);
- if (field)
- msg.appendf(" for field %s", str(field->name));
- if (value)
- msg.appendf(", received %s", getTypeName(cass_value_type(value)));
- rtlFail(0, msg.str());
- }
- extern bool isInteger(const CassValueType t)
- {
- switch (t)
- {
- case CASS_VALUE_TYPE_TIMESTAMP:
- case CASS_VALUE_TYPE_INT:
- case CASS_VALUE_TYPE_BIGINT:
- case CASS_VALUE_TYPE_COUNTER:
- case CASS_VALUE_TYPE_VARINT:
- return true;
- default:
- return false;
- }
- }
- extern bool isString(CassValueType t)
- {
- switch (t)
- {
- case CASS_VALUE_TYPE_VARCHAR:
- case CASS_VALUE_TYPE_TEXT:
- case CASS_VALUE_TYPE_ASCII:
- return true;
- default:
- return false;
- }
- }
- // when extracting elements of a set, field will point at the SET info- we want to get the typeInfo for the element type
- static const RtlTypeInfo *getFieldBaseType(const RtlFieldInfo *field)
- {
- const RtlTypeInfo *type = field->type;
- if ((type->fieldType & RFTMkind) == type_set)
- return type->queryChildType();
- else
- return type;
- }
- static int getNumFields(const RtlTypeInfo *record)
- {
- int count = 0;
- const RtlFieldInfo * const *fields = record->queryFields();
- assertex(fields);
- while (*fields++)
- count++;
- return count;
- }
- extern bool getBooleanResult(const RtlFieldInfo *field, const CassValue *value)
- {
- if (cass_value_is_null(value))
- {
- NullFieldProcessor p(field);
- return p.boolResult;
- }
- if (cass_value_type(value) != CASS_VALUE_TYPE_BOOLEAN)
- typeError("boolean", value, field);
- cass_bool_t output;
- check(cass_value_get_bool(value, &output));
- return output != cass_false;
- }
- extern void getDataResult(const RtlFieldInfo *field, const CassValue *value, size32_t &chars, void * &result)
- {
- if (cass_value_is_null(value))
- {
- NullFieldProcessor p(field);
- rtlStrToDataX(chars, result, p.resultChars, p.stringResult);
- return;
- }
- // We COULD require that the field being retrieved is a blob - but Cassandra seems happy to use any field here, and
- // it seems like it could be more useful to support anything
- // if (cass_value_type(value) != CASS_VALUE_TYPE_BLOB)
- // typeError("blob", value, field);
- const cass_byte_t *bytes;
- size_t size;
- check(cass_value_get_bytes(value, &bytes, &size));
- rtlStrToDataX(chars, result, size, bytes);
- }
- extern double getRealResult(const RtlFieldInfo *field, const CassValue *value)
- {
- if (cass_value_is_null(value))
- {
- NullFieldProcessor p(field);
- return p.doubleResult;
- }
- else if (isInteger(cass_value_type(value)))
- return (double) getSignedResult(field, value);
- else switch (cass_value_type(value))
- {
- case CASS_VALUE_TYPE_FLOAT:
- {
- cass_float_t output_f;
- check(cass_value_get_float(value, &output_f));
- return output_f;
- }
- case CASS_VALUE_TYPE_DOUBLE:
- {
- cass_double_t output_d;
- check(cass_value_get_double(value, &output_d));
- return output_d;
- }
- default:
- typeError("double", value, field);
- }
- }
- extern __int64 getSignedResult(const RtlFieldInfo *field, const CassValue *value)
- {
- if (cass_value_is_null(value))
- {
- NullFieldProcessor p(field);
- return p.intResult;
- }
- switch (cass_value_type(value))
- {
- case CASS_VALUE_TYPE_INT:
- {
- cass_int32_t output;
- check(cass_value_get_int32(value, &output));
- return output;
- }
- case CASS_VALUE_TYPE_TIMESTAMP:
- case CASS_VALUE_TYPE_BIGINT:
- case CASS_VALUE_TYPE_COUNTER:
- case CASS_VALUE_TYPE_VARINT:
- {
- cass_int64_t output;
- check(cass_value_get_int64(value, &output));
- return output;
- }
- default:
- typeError("integer", value, field);
- }
- }
- extern unsigned __int64 getUnsignedResult(const RtlFieldInfo *field, const CassValue *value)
- {
- if (cass_value_is_null(value))
- {
- NullFieldProcessor p(field);
- return p.uintResult;
- }
- return (__uint64) getSignedResult(field, value);
- }
- extern void getStringResult(const RtlFieldInfo *field, const CassValue *value, size32_t &chars, char * &result)
- {
- if (cass_value_is_null(value))
- {
- NullFieldProcessor p(field);
- rtlStrToStrX(chars, result, p.resultChars, p.stringResult);
- return;
- }
- switch (cass_value_type(value))
- {
- case CASS_VALUE_TYPE_ASCII:
- {
- const char *output;
- size_t length;
- check(cass_value_get_string(value, &output, &length));
- rtlStrToStrX(chars, result, length, output);
- break;
- }
- case CASS_VALUE_TYPE_VARCHAR:
- case CASS_VALUE_TYPE_TEXT:
- {
- const char *output;
- size_t length;
- check(cass_value_get_string(value, &output, &length));
- unsigned numchars = rtlUtf8Length(length, output);
- rtlUtf8ToStrX(chars, result, numchars, output);
- break;
- }
- default:
- typeError("string", value, field);
- }
- }
- extern void getUTF8Result(const RtlFieldInfo *field, const CassValue *value, size32_t &chars, char * &result)
- {
- if (cass_value_is_null(value))
- {
- NullFieldProcessor p(field);
- rtlUtf8ToUtf8X(chars, result, p.resultChars, p.stringResult);
- return;
- }
- switch (cass_value_type(value))
- {
- case CASS_VALUE_TYPE_ASCII:
- {
- const char *output;
- size_t length;
- check(cass_value_get_string(value, &output, &length));
- rtlStrToUtf8X(chars, result, length, output);
- break;
- }
- case CASS_VALUE_TYPE_VARCHAR:
- case CASS_VALUE_TYPE_TEXT:
- {
- const char * output;
- size_t length;
- check(cass_value_get_string(value, &output, &length));
- unsigned numchars = rtlUtf8Length(length, output);
- rtlUtf8ToUtf8X(chars, result, numchars, output);
- break;
- }
- default:
- typeError("string", value, field);
- }
- }
- extern void getUnicodeResult(const RtlFieldInfo *field, const CassValue *value, size32_t &chars, UChar * &result)
- {
- if (cass_value_is_null(value))
- {
- NullFieldProcessor p(field);
- rtlUnicodeToUnicodeX(chars, result, p.resultChars, p.unicodeResult);
- return;
- }
- switch (cass_value_type(value))
- {
- case CASS_VALUE_TYPE_ASCII:
- {
- const char * output;
- size_t length;
- check(cass_value_get_string(value, &output, &length));
- rtlStrToUnicodeX(chars, result, length, output);
- break;
- }
- case CASS_VALUE_TYPE_VARCHAR:
- case CASS_VALUE_TYPE_TEXT:
- {
- const char * output;
- size_t length;
- check(cass_value_get_string(value, &output, &length));
- unsigned numchars = rtlUtf8Length(length, output);
- rtlUtf8ToUnicodeX(chars, result, numchars, output);
- break;
- }
- default:
- typeError("string", value, field);
- }
- }
- extern void getDecimalResult(const RtlFieldInfo *field, const CassValue *value, Decimal &result)
- {
- // Note - Cassandra has a decimal type, but it's not particularly similar to the ecl one. Map to string for now, as we do in MySQL
- if (cass_value_is_null(value))
- {
- NullFieldProcessor p(field);
- result.set(p.decimalResult);
- return;
- }
- size32_t chars;
- rtlDataAttr tempStr;
- cassandraembed::getStringResult(field, value, chars, tempStr.refstr());
- result.setString(chars, tempStr.getstr());
- if (field)
- {
- RtlDecimalTypeInfo *dtype = (RtlDecimalTypeInfo *) field->type;
- result.setPrecision(dtype->getDecimalDigits(), dtype->getDecimalPrecision());
- }
- }
- // A CassandraRowBuilder object is used to construct an ECL row from a Cassandra row
- class CassandraRowBuilder : public CInterfaceOf<IFieldSource>
- {
- public:
- CassandraRowBuilder(const CassandraStatementInfo *_stmtInfo)
- : stmtInfo(_stmtInfo), colIdx(0), numIteratorFields(0), nextIteratedField(0)
- {
- }
- virtual bool getBooleanResult(const RtlFieldInfo *field)
- {
- return cassandraembed::getBooleanResult(field, nextField(field));
- }
- virtual void getDataResult(const RtlFieldInfo *field, size32_t &len, void * &result)
- {
- cassandraembed::getDataResult(field, nextField(field), len, result);
- }
- virtual double getRealResult(const RtlFieldInfo *field)
- {
- return cassandraembed::getRealResult(field, nextField(field));
- }
- virtual __int64 getSignedResult(const RtlFieldInfo *field)
- {
- return cassandraembed::getSignedResult(field, nextField(field));
- }
- virtual unsigned __int64 getUnsignedResult(const RtlFieldInfo *field)
- {
- return cassandraembed::getUnsignedResult(field, nextField(field));
- }
- virtual void getStringResult(const RtlFieldInfo *field, size32_t &chars, char * &result)
- {
- cassandraembed::getStringResult(field, nextField(field), chars, result);
- }
- virtual void getUTF8Result(const RtlFieldInfo *field, size32_t &chars, char * &result)
- {
- cassandraembed::getUTF8Result(field, nextField(field), chars, result);
- }
- virtual void getUnicodeResult(const RtlFieldInfo *field, size32_t &chars, UChar * &result)
- {
- cassandraembed::getUnicodeResult(field, nextField(field), chars, result);
- }
- virtual void getDecimalResult(const RtlFieldInfo *field, Decimal &value)
- {
- cassandraembed::getDecimalResult(field, nextField(field), value);
- }
- virtual void processBeginSet(const RtlFieldInfo * field, bool &isAll)
- {
- isAll = false;
- iterator.setown(new CassandraIterator(cass_iterator_from_collection(nextField(field))));
- }
- virtual bool processNextSet(const RtlFieldInfo * field)
- {
- numIteratorFields = 1;
- return *iterator && cass_iterator_next(*iterator); // If field was NULL, we'll have a NULL iterator (representing an empty set/list)
- // Can't distinguish empty set from NULL field, so assume the former (rather than trying to deliver the default value for the set field)
- }
- virtual void processBeginDataset(const RtlFieldInfo * field)
- {
- numIteratorFields = getNumFields(field->type->queryChildType());
- switch (numIteratorFields)
- {
- case 1:
- iterator.setown(new CassandraIterator(cass_iterator_from_collection(nextField(field))));
- break;
- case 2:
- iterator.setown(new CassandraIterator(cass_iterator_from_map(nextField(field))));
- break;
- default:
- UNSUPPORTED("Nested datasets with > 2 fields");
- }
- }
- virtual void processBeginRow(const RtlFieldInfo * field)
- {
- }
- virtual bool processNextRow(const RtlFieldInfo * field)
- {
- nextIteratedField = 0;
- return *iterator && cass_iterator_next(*iterator); // If field was NULL, we'll have a NULL iterator (representing an empty set/list/map)
- // Can't distinguish empty set from NULL field, so assume the former (rather than trying to deliver the default value for the set field)
- }
- virtual void processEndSet(const RtlFieldInfo * field)
- {
- iterator.clear();
- numIteratorFields = 0;
- }
- virtual void processEndDataset(const RtlFieldInfo * field)
- {
- iterator.clear();
- numIteratorFields = 0;
- }
- virtual void processEndRow(const RtlFieldInfo * field)
- {
- }
- protected:
- const CassValue *nextField(const RtlFieldInfo * field)
- {
- const CassValue *ret;
- if (iterator)
- {
- switch (numIteratorFields)
- {
- case 1:
- ret = cass_iterator_get_value(*iterator);
- break;
- case 2:
- if (nextIteratedField==0)
- ret = cass_iterator_get_map_key(*iterator);
- else
- ret = cass_iterator_get_map_value(*iterator);
- nextIteratedField++;
- break;
- default:
- throwUnexpected();
- }
- }
- else
- ret = cass_row_get_column(stmtInfo->queryRow(), colIdx++);
- if (!ret)
- failx("Too many fields in ECL output row, reading field %s", str(field->name));
- return ret;
- }
- const CassandraStatementInfo *stmtInfo;
- Owned<CassandraIterator> iterator;
- int colIdx;
- int numIteratorFields;
- int nextIteratedField;
- };
- // Bind Cassandra columns from an ECL record
- class CassandraRecordBinder : public CInterfaceOf<IFieldProcessor>
- {
- public:
- CassandraRecordBinder(const IContextLogger &_logctx, const RtlTypeInfo *_typeInfo, const CassandraStatementInfo *_stmtInfo, int _firstParam)
- : logctx(_logctx), typeInfo(_typeInfo), stmtInfo(_stmtInfo), firstParam(_firstParam), dummyField("<row>", NULL, typeInfo), thisParam(_firstParam)
- {
- }
- int numFields()
- {
- int count = 0;
- const RtlFieldInfo * const *fields = typeInfo->queryFields();
- assertex(fields);
- while (*fields++)
- count++;
- return count;
- }
- void processRow(const byte *row)
- {
- thisParam = firstParam;
- typeInfo->process(row, row, &dummyField, *this); // Bind the variables for the current row
- }
- virtual void processString(unsigned len, const char *value, const RtlFieldInfo * field)
- {
- size32_t utf8chars;
- rtlDataAttr utfText;
- rtlStrToUtf8X(utf8chars, utfText.refstr(), len, value);
- if (collection)
- checkBind(cass_collection_append_string_n(*collection, utfText.getstr(), rtlUtf8Size(utf8chars, utfText.getstr())),
- field);
- else
- checkBind(cass_statement_bind_string_n(stmtInfo->queryStatement(),
- checkNextParam(field),
- utfText.getstr(), rtlUtf8Size(utf8chars, utfText.getstr())),
- field);
- }
- virtual void processBool(bool value, const RtlFieldInfo * field)
- {
- if (collection)
- checkBind(cass_collection_append_bool(*collection, value ? cass_true : cass_false), field);
- else
- checkBind(cass_statement_bind_bool(stmtInfo->queryStatement(), checkNextParam(field), value ? cass_true : cass_false), field);
- }
- virtual void processData(unsigned len, const void *value, const RtlFieldInfo * field)
- {
- if (collection)
- checkBind(cass_collection_append_bytes(*collection, (const cass_byte_t*) value, len), field);
- else
- checkBind(cass_statement_bind_bytes(stmtInfo->queryStatement(), checkNextParam(field), (const cass_byte_t*) value, len), field);
- }
- virtual void processInt(__int64 value, const RtlFieldInfo * field)
- {
- if (getFieldBaseType(field)->size(NULL,NULL)>4)
- {
- if (collection)
- checkBind(cass_collection_append_int64(*collection, value), field);
- else
- checkBind(cass_statement_bind_int64(stmtInfo->queryStatement(), checkNextParam(field), value), field);
- }
- else
- {
- if (collection)
- checkBind(cass_collection_append_int32(*collection, value), field);
- else
- checkBind(cass_statement_bind_int32(stmtInfo->queryStatement(), checkNextParam(field), value), field);
- }
- }
- virtual void processUInt(unsigned __int64 value, const RtlFieldInfo * field)
- {
- UNSUPPORTED("UNSIGNED columns");
- }
- virtual void processReal(double value, const RtlFieldInfo * field)
- {
- if (getFieldBaseType(field)->size(NULL,NULL)>4)
- {
- if (collection)
- checkBind(cass_collection_append_double(*collection, value), field);
- else
- checkBind(cass_statement_bind_double(stmtInfo->queryStatement(), checkNextParam(field), value), field);
- }
- else
- {
- if (collection)
- checkBind(cass_collection_append_float(*collection, (float) value), field);
- else
- checkBind(cass_statement_bind_float(stmtInfo->queryStatement(), checkNextParam(field), (float) value), field);
- }
- }
- virtual void processDecimal(const void *value, unsigned digits, unsigned precision, const RtlFieldInfo * field)
- {
- Decimal val;
- size32_t bytes;
- rtlDataAttr decText;
- val.setDecimal(digits, precision, value);
- val.getStringX(bytes, decText.refstr());
- processUtf8(bytes, decText.getstr(), field);
- }
- virtual void processUDecimal(const void *value, unsigned digits, unsigned precision, const RtlFieldInfo * field)
- {
- UNSUPPORTED("UNSIGNED decimals");
- }
- virtual void processUnicode(unsigned chars, const UChar *value, const RtlFieldInfo * field)
- {
- size32_t utf8chars;
- rtlDataAttr utfText;
- rtlUnicodeToUtf8X(utf8chars, utfText.refstr(), chars, value);
- if (collection)
- checkBind(cass_collection_append_string_n(*collection, utfText.getstr(), rtlUtf8Size(utf8chars, utfText.getstr())),
- field);
- else
- checkBind(cass_statement_bind_string_n(stmtInfo->queryStatement(),
- checkNextParam(field),
- utfText.getstr(), rtlUtf8Size(utf8chars, utfText.getstr())),
- field);
- }
- virtual void processQString(unsigned len, const char *value, const RtlFieldInfo * field)
- {
- size32_t charCount;
- rtlDataAttr text;
- rtlQStrToStrX(charCount, text.refstr(), len, value);
- processUtf8(charCount, text.getstr(), field);
- }
- virtual void processUtf8(unsigned chars, const char *value, const RtlFieldInfo * field)
- {
- if (collection)
- checkBind(cass_collection_append_string_n(*collection, value, rtlUtf8Size(chars, value)), field);
- else
- checkBind(cass_statement_bind_string_n(stmtInfo->queryStatement(), checkNextParam(field), value, rtlUtf8Size(chars, value)), field);
- }
- virtual bool processBeginSet(const RtlFieldInfo * field, unsigned numElements, bool isAll, const byte *data)
- {
- if (isAll)
- UNSUPPORTED("SET(ALL)");
- // We don't know whether the corresponding field in Cassandra is a list or a set. Try binding a dummy list to tell which.
- CassandraCollection temp(cass_collection_new(CASS_COLLECTION_TYPE_LIST, 0));
- if (cass_statement_bind_collection(stmtInfo->queryStatement(), thisParam, temp) == CASS_OK)
- collection.setown(new CassandraCollection(cass_collection_new(CASS_COLLECTION_TYPE_LIST, numElements)));
- else
- collection.setown(new CassandraCollection(cass_collection_new(CASS_COLLECTION_TYPE_SET, numElements)));
- return true;
- }
- virtual bool processBeginDataset(const RtlFieldInfo * field, unsigned numRows)
- {
- // If there's a single field, assume we are mapping to a SET/LIST
- // If there are two, assume it's a MAP
- // Otherwise, fail
- int numFields = getNumFields(field->type->queryChildType());
- if (numFields < 1 || numFields > 2)
- {
- UNSUPPORTED("Nested datasets with > 2 fields");
- }
- collection.setown(new CassandraCollection(cass_collection_new(numFields==1 ? CASS_COLLECTION_TYPE_SET : CASS_COLLECTION_TYPE_MAP, numRows)));
- return true;
- }
- virtual bool processBeginRow(const RtlFieldInfo * field)
- {
- return true;
- }
- virtual void processEndSet(const RtlFieldInfo * field)
- {
- checkBind(cass_statement_bind_collection(stmtInfo->queryStatement(), checkNextParam(field), *collection), field);
- collection.clear();
- }
- virtual void processEndDataset(const RtlFieldInfo * field)
- {
- checkBind(cass_statement_bind_collection(stmtInfo->queryStatement(), checkNextParam(field), *collection), field);
- collection.clear();
- }
- virtual void processEndRow(const RtlFieldInfo * field)
- {
- }
- protected:
- inline unsigned checkNextParam(const RtlFieldInfo * field)
- {
- if (logctx.queryTraceLevel() > 4)
- logctx.CTXLOG("Binding %s to %d", str(field->name), thisParam);
- return thisParam++;
- }
- inline void checkBind(CassError rc, const RtlFieldInfo * field)
- {
- if (rc != CASS_OK)
- {
- failx("While binding parameter %s: %s", str(field->name), cass_error_desc(rc));
- }
- }
- const RtlTypeInfo *typeInfo;
- const CassandraStatementInfo *stmtInfo;
- Owned<CassandraCollection> collection;
- const IContextLogger &logctx;
- int firstParam;
- RtlFieldStrInfo dummyField;
- int thisParam;
- };
- //
- class CassandraDatasetBinder : public CassandraRecordBinder
- {
- public:
- CassandraDatasetBinder(const IContextLogger &_logctx, IRowStream * _input, const RtlTypeInfo *_typeInfo, const CassandraStatementInfo *_stmt, int _firstParam)
- : input(_input), CassandraRecordBinder(_logctx, _typeInfo, _stmt, _firstParam)
- {
- }
- bool bindNext()
- {
- roxiemem::OwnedConstRoxieRow nextRow = (const byte *) input->ungroupedNextRow();
- if (!nextRow)
- return false;
- processRow((const byte *) nextRow.get()); // Bind the variables for the current row
- return true;
- }
- void executeAll(CassandraStatementInfo *stmtInfo)
- {
- stmtInfo->startStream();
- while (bindNext())
- {
- stmtInfo->execute();
- }
- stmtInfo->endStream();
- }
- protected:
- Owned<IRowStream> input;
- };
- // A Cassandra function that returns a dataset will return a CassandraRowStream object that can be
- // interrogated to return each row of the result in turn
- class CassandraRowStream : public CInterfaceOf<IRowStream>
- {
- public:
- CassandraRowStream(CassandraDatasetBinder *_inputStream, CassandraStatementInfo *_stmtInfo, IEngineRowAllocator *_resultAllocator)
- : inputStream(_inputStream), stmtInfo(_stmtInfo), resultAllocator(_resultAllocator)
- {
- executePending = true;
- eof = false;
- }
- virtual const void *nextRow()
- {
- // A little complex when streaming data in as well as out - want to execute for every input record
- if (eof)
- return NULL;
- loop
- {
- if (executePending)
- {
- executePending = false;
- if (inputStream && !inputStream->bindNext())
- {
- noteEOF();
- return NULL;
- }
- stmtInfo->execute();
- }
- if (stmtInfo->next())
- break;
- if (inputStream)
- executePending = true;
- else
- {
- noteEOF();
- return NULL;
- }
- }
- RtlDynamicRowBuilder rowBuilder(resultAllocator);
- CassandraRowBuilder cassandraRowBuilder(stmtInfo);
- const RtlTypeInfo *typeInfo = resultAllocator->queryOutputMeta()->queryTypeInfo();
- assertex(typeInfo);
- RtlFieldStrInfo dummyField("<row>", NULL, typeInfo);
- size32_t len = typeInfo->build(rowBuilder, 0, &dummyField, cassandraRowBuilder);
- return rowBuilder.finalizeRowClear(len);
- }
- virtual void stop()
- {
- resultAllocator.clear();
- stmtInfo->stop();
- }
- protected:
- void noteEOF()
- {
- if (!eof)
- {
- eof = true;
- stop();
- }
- }
- Linked<CassandraDatasetBinder> inputStream;
- Linked<CassandraStatementInfo> stmtInfo;
- Linked<IEngineRowAllocator> resultAllocator;
- bool executePending;
- bool eof;
- };
- // Each call to a Cassandra function will use a new CassandraEmbedFunctionContext object
- class CassandraEmbedFunctionContext : public CInterfaceOf<IEmbedFunctionContext>
- {
- public:
- CassandraEmbedFunctionContext(const IContextLogger &_logctx, unsigned _flags, const char *options)
- : logctx(_logctx), flags(_flags), nextParam(0), numParams(0), batchMode((CassBatchType) -1), pageSize(0)
- {
- StringArray opts;
- opts.appendList(options, ",");
- hash64_t hash = 0;
- ForEachItemInRev(idx, opts)
- {
- const char *opt = opts.item(idx);
- if (strnicmp(opt, "batch=", 6)==0)
- {
- const char *val=opt+6;
- if (stricmp(val, "LOGGED")==0)
- batchMode = CASS_BATCH_TYPE_LOGGED;
- else if (stricmp(val, "UNLOGGED")==0)
- batchMode = CASS_BATCH_TYPE_UNLOGGED;
- else if (stricmp(val, "COUNTER")==0)
- batchMode = CASS_BATCH_TYPE_COUNTER;
- opts.remove(idx);
- }
- else if (strnicmp(opt, "pagesize=", 9)==0)
- {
- pageSize = atoi(opt+9);
- opts.remove(idx);
- }
- else
- hash = rtlHash64VStr(opt, hash);
- }
- cluster.setown(lookupCachedSession(hash, opts));
- }
- virtual bool getBooleanResult()
- {
- bool ret = cassandraembed::getBooleanResult(NULL, getScalarResult());
- checkSingleRow();
- return ret;
- }
- virtual void getDataResult(size32_t &len, void * &result)
- {
- cassandraembed::getDataResult(NULL, getScalarResult(), len, result);
- checkSingleRow();
- }
- virtual double getRealResult()
- {
- double ret = cassandraembed::getRealResult(NULL, getScalarResult());
- checkSingleRow();
- return ret;
- }
- virtual __int64 getSignedResult()
- {
- __int64 ret = cassandraembed::getSignedResult(NULL, getScalarResult());
- checkSingleRow();
- return ret;
- }
- virtual unsigned __int64 getUnsignedResult()
- {
- unsigned __int64 ret = cassandraembed::getUnsignedResult(NULL, getScalarResult());
- checkSingleRow();
- return ret;
- }
- virtual void getStringResult(size32_t &chars, char * &result)
- {
- cassandraembed::getStringResult(NULL, getScalarResult(), chars, result);
- checkSingleRow();
- }
- virtual void getUTF8Result(size32_t &chars, char * &result)
- {
- cassandraembed::getUTF8Result(NULL, getScalarResult(), chars, result);
- checkSingleRow();
- }
- virtual void getUnicodeResult(size32_t &chars, UChar * &result)
- {
- cassandraembed::getUnicodeResult(NULL, getScalarResult(), chars, result);
- checkSingleRow();
- }
- virtual void getDecimalResult(Decimal &value)
- {
- cassandraembed::getDecimalResult(NULL, getScalarResult(), value);
- checkSingleRow();
- }
- virtual void getSetResult(bool & __isAllResult, size32_t & __resultBytes, void * & __result, int elemType, size32_t elemSize)
- {
- CassandraIterator iterator(cass_iterator_from_collection(getScalarResult()));
- rtlRowBuilder out;
- byte *outData = NULL;
- size32_t outBytes = 0;
- while (cass_iterator_next(iterator))
- {
- const CassValue *value = cass_iterator_get_value(iterator);
- assertex(value);
- if (elemSize != UNKNOWN_LENGTH)
- {
- out.ensureAvailable(outBytes + elemSize);
- outData = out.getbytes() + outBytes;
- }
- switch ((type_t) elemType)
- {
- case type_int:
- rtlWriteInt(outData, cassandraembed::getSignedResult(NULL, value), elemSize);
- break;
- case type_unsigned:
- rtlWriteInt(outData, cassandraembed::getUnsignedResult(NULL, value), elemSize);
- break;
- case type_real:
- if (elemSize == sizeof(double))
- * (double *) outData = cassandraembed::getRealResult(NULL, value);
- else
- {
- assertex(elemSize == sizeof(float));
- * (float *) outData = (float) cassandraembed::getRealResult(NULL, value);
- }
- break;
- case type_boolean:
- assertex(elemSize == sizeof(bool));
- * (bool *) outData = cassandraembed::getBooleanResult(NULL, value);
- break;
- case type_string:
- case type_varstring:
- {
- rtlDataAttr str;
- size32_t lenBytes;
- cassandraembed::getStringResult(NULL, value, lenBytes, str.refstr());
- if (elemSize == UNKNOWN_LENGTH)
- {
- if (elemType == type_string)
- {
- out.ensureAvailable(outBytes + lenBytes + sizeof(size32_t));
- outData = out.getbytes() + outBytes;
- * (size32_t *) outData = lenBytes;
- rtlStrToStr(lenBytes, outData+sizeof(size32_t), lenBytes, str.getstr());
- outBytes += lenBytes + sizeof(size32_t);
- }
- else
- {
- out.ensureAvailable(outBytes + lenBytes + 1);
- outData = out.getbytes() + outBytes;
- rtlStrToVStr(0, outData, lenBytes, str.getstr());
- outBytes += lenBytes + 1;
- }
- }
- else
- {
- if (elemType == type_string)
- rtlStrToStr(elemSize, outData, lenBytes, str.getstr());
- else
- rtlStrToVStr(elemSize, outData, lenBytes, str.getstr()); // Fixed size null terminated strings... weird.
- }
- break;
- }
- case type_unicode:
- case type_utf8:
- {
- rtlDataAttr str;
- size32_t lenChars;
- cassandraembed::getUTF8Result(NULL, value, lenChars, str.refstr());
- const char * text = str.getstr();
- size32_t lenBytes = rtlUtf8Size(lenChars, text);
- if (elemType == type_utf8)
- {
- assertex (elemSize == UNKNOWN_LENGTH);
- out.ensureAvailable(outBytes + lenBytes + sizeof(size32_t));
- outData = out.getbytes() + outBytes;
- * (size32_t *) outData = lenChars;
- rtlStrToStr(lenBytes, outData+sizeof(size32_t), lenBytes, text);
- outBytes += lenBytes + sizeof(size32_t);
- }
- else
- {
- if (elemSize == UNKNOWN_LENGTH)
- {
- // You can't assume that number of chars in utf8 matches number in unicode16 ...
- size32_t numchars16;
- rtlDataAttr unicode16;
- rtlUtf8ToUnicodeX(numchars16, unicode16.refustr(), lenChars, text);
- out.ensureAvailable(outBytes + numchars16*sizeof(UChar) + sizeof(size32_t));
- outData = out.getbytes() + outBytes;
- * (size32_t *) outData = numchars16;
- rtlUnicodeToUnicode(numchars16, (UChar *) (outData+sizeof(size32_t)), numchars16, unicode16.getustr());
- outBytes += numchars16*sizeof(UChar) + sizeof(size32_t);
- }
- else
- rtlUtf8ToUnicode(elemSize / sizeof(UChar), (UChar *) outData, lenChars, text);
- }
- break;
- }
- default:
- fail("type mismatch - unsupported return type");
- }
- if (elemSize != UNKNOWN_LENGTH)
- outBytes += elemSize;
- }
- __isAllResult = false;
- __resultBytes = outBytes;
- __result = out.detachdata();
- }
- virtual IRowStream *getDatasetResult(IEngineRowAllocator * _resultAllocator)
- {
- return new CassandraRowStream(inputStream, stmtInfo, _resultAllocator);
- }
- virtual byte * getRowResult(IEngineRowAllocator * _resultAllocator)
- {
- if (!stmtInfo->hasResult() || stmtInfo->rowCount() != 1)
- typeError("row", NULL, NULL);
- CassandraRowStream stream(NULL, stmtInfo, _resultAllocator);
- roxiemem::OwnedConstRoxieRow ret = stream.nextRow();
- stream.stop();
- if (ret == NULL) // Check for exactly one returned row
- typeError("row", NULL, NULL);
- return (byte *) ret.getClear();
- }
- virtual size32_t getTransformResult(ARowBuilder & rowBuilder)
- {
- if (!stmtInfo->hasResult() || stmtInfo->rowCount() != 1)
- typeError("row", NULL, NULL);
- if (!stmtInfo->next())
- fail("Failed to read row");
- CassandraRowBuilder cassandraRowBuilder(stmtInfo);
- const RtlTypeInfo *typeInfo = rowBuilder.queryAllocator()->queryOutputMeta()->queryTypeInfo();
- assertex(typeInfo);
- RtlFieldStrInfo dummyField("<row>", NULL, typeInfo);
- return typeInfo->build(rowBuilder, 0, &dummyField, cassandraRowBuilder);
- }
- virtual void bindRowParam(const char *name, IOutputMetaData & metaVal, byte *val)
- {
- CassandraRecordBinder binder(logctx, metaVal.queryTypeInfo(), stmtInfo, nextParam);
- binder.processRow(val);
- nextParam += binder.numFields();
- }
- virtual void bindDatasetParam(const char *name, IOutputMetaData & metaVal, IRowStream * val)
- {
- // We only support a single dataset parameter...
- // MORE - look into batch?
- if (inputStream)
- {
- fail("At most one dataset parameter supported");
- }
- inputStream.setown(new CassandraDatasetBinder(logctx, LINK(val), metaVal.queryTypeInfo(), stmtInfo, nextParam));
- nextParam += inputStream->numFields();
- }
- virtual void bindBooleanParam(const char *name, bool val)
- {
- checkBind(cass_statement_bind_bool(stmtInfo->queryStatement(), checkNextParam(name), val ? cass_true : cass_false), name);
- }
- virtual void bindDataParam(const char *name, size32_t len, const void *val)
- {
- checkBind(cass_statement_bind_bytes(stmtInfo->queryStatement(), checkNextParam(name), (const cass_byte_t*) val, len), name);
- }
- virtual void bindFloatParam(const char *name, float val)
- {
- checkBind(cass_statement_bind_float(stmtInfo->queryStatement(), checkNextParam(name), val), name);
- }
- virtual void bindRealParam(const char *name, double val)
- {
- checkBind(cass_statement_bind_double(stmtInfo->queryStatement(), checkNextParam(name), val), name);
- }
- virtual void bindSignedSizeParam(const char *name, int size, __int64 val)
- {
- if (size > 4)
- checkBind(cass_statement_bind_int64(stmtInfo->queryStatement(), checkNextParam(name), val), name);
- else
- checkBind(cass_statement_bind_int32(stmtInfo->queryStatement(), checkNextParam(name), val), name);
- }
- virtual void bindSignedParam(const char *name, __int64 val)
- {
- bindSignedSizeParam(name, 8, val);
- }
- virtual void bindUnsignedSizeParam(const char *name, int size, unsigned __int64 val)
- {
- UNSUPPORTED("UNSIGNED columns");
- }
- virtual void bindUnsignedParam(const char *name, unsigned __int64 val)
- {
- UNSUPPORTED("UNSIGNED columns");
- }
- virtual void bindStringParam(const char *name, size32_t len, const char *val)
- {
- size32_t utf8chars;
- rtlDataAttr utfText;
- rtlStrToUtf8X(utf8chars, utfText.refstr(), len, val);
- checkBind(cass_statement_bind_string_n(stmtInfo->queryStatement(),
- checkNextParam(name),
- utfText.getstr(), rtlUtf8Size(utf8chars, utfText.getstr())),
- name);
- }
- virtual void bindVStringParam(const char *name, const char *val)
- {
- bindStringParam(name, strlen(val), val);
- }
- virtual void bindUTF8Param(const char *name, size32_t chars, const char *val)
- {
- checkBind(cass_statement_bind_string_n(stmtInfo->queryStatement(), checkNextParam(name), val, rtlUtf8Size(chars, val)), name);
- }
- virtual void bindUnicodeParam(const char *name, size32_t chars, const UChar *val)
- {
- size32_t utf8chars;
- rtlDataAttr utfText;
- rtlUnicodeToUtf8X(utf8chars, utfText.refstr(), chars, val);
- checkBind(cass_statement_bind_string_n(stmtInfo->queryStatement(),
- checkNextParam(name),
- utfText.getstr(), rtlUtf8Size(utf8chars, utfText.getstr())),
- name);
- }
- virtual void bindSetParam(const char *name, int elemType, size32_t elemSize, bool isAll, size32_t totalBytes, void *setData)
- {
- if (isAll)
- UNSUPPORTED("SET(ALL)");
- type_t typecode = (type_t) elemType;
- const byte *inData = (const byte *) setData;
- const byte *endData = inData + totalBytes;
- int numElems;
- if (elemSize == UNKNOWN_LENGTH)
- {
- numElems = 0;
- // Will need 2 passes to work out how many elements there are in the set :(
- while (inData < endData)
- {
- int thisSize;
- switch (elemType)
- {
- case type_varstring:
- thisSize = strlen((const char *) inData) + 1;
- break;
- case type_string:
- thisSize = * (size32_t *) inData + sizeof(size32_t);
- break;
- case type_unicode:
- thisSize = (* (size32_t *) inData) * sizeof(UChar) + sizeof(size32_t);
- break;
- case type_utf8:
- thisSize = rtlUtf8Size(* (size32_t *) inData, inData + sizeof(size32_t)) + sizeof(size32_t);
- break;
- default:
- fail("Unsupported parameter type");
- break;
- }
- inData += thisSize;
- numElems++;
- }
- inData = (const byte *) setData;
- }
- else
- numElems = totalBytes / elemSize;
- CassandraCollection collection(cass_collection_new(CASS_COLLECTION_TYPE_SET, numElems));
- while (inData < endData)
- {
- size32_t thisSize = elemSize;
- CassError rc;
- switch (typecode)
- {
- case type_int:
- if (elemSize > 4)
- rc = cass_collection_append_int64(collection, rtlReadInt(inData, elemSize));
- else
- rc = cass_collection_append_int32(collection, rtlReadInt(inData, elemSize));
- break;
- case type_unsigned:
- UNSUPPORTED("UNSIGNED columns");
- break;
- case type_varstring:
- {
- size32_t numChars = strlen((const char *) inData);
- if (elemSize == UNKNOWN_LENGTH)
- thisSize = numChars + 1;
- size32_t utf8chars;
- rtlDataAttr utfText;
- rtlStrToUtf8X(utf8chars, utfText.refstr(), numChars, (const char *) inData);
- rc = cass_collection_append_string_n(collection, utfText.getstr(), rtlUtf8Size(utf8chars, utfText.getstr()));
- break;
- }
- case type_string:
- {
- if (elemSize == UNKNOWN_LENGTH)
- {
- thisSize = * (size32_t *) inData;
- inData += sizeof(size32_t);
- }
- size32_t utf8chars;
- rtlDataAttr utfText;
- rtlStrToUtf8X(utf8chars, utfText.refstr(), thisSize, (const char *) inData);
- rc = cass_collection_append_string_n(collection, utfText.getstr(), rtlUtf8Size(utf8chars, utfText.getstr()));
- break;
- }
- case type_real:
- if (elemSize == sizeof(double))
- rc = cass_collection_append_double(collection, * (double *) inData);
- else
- rc = cass_collection_append_float(collection, * (float *) inData);
- break;
- case type_boolean:
- assertex(elemSize == sizeof(bool));
- rc = cass_collection_append_bool(collection, *(bool*)inData ? cass_true : cass_false);
- break;
- case type_unicode:
- {
- if (elemSize == UNKNOWN_LENGTH)
- {
- thisSize = (* (size32_t *) inData) * sizeof(UChar); // NOTE - it's in chars...
- inData += sizeof(size32_t);
- }
- unsigned unicodeChars;
- rtlDataAttr unicode;
- rtlUnicodeToUtf8X(unicodeChars, unicode.refstr(), thisSize / sizeof(UChar), (const UChar *) inData);
- size32_t sizeBytes = rtlUtf8Size(unicodeChars, unicode.getstr());
- rc = cass_collection_append_string_n(collection, unicode.getstr(), sizeBytes);
- break;
- }
- case type_utf8:
- {
- assertex (elemSize == UNKNOWN_LENGTH);
- size32_t numChars = * (size32_t *) inData;
- inData += sizeof(size32_t);
- thisSize = rtlUtf8Size(numChars, inData);
- rc = cass_collection_append_string_n(collection, (const char *) inData, thisSize);
- break;
- }
- case type_data:
- if (elemSize == UNKNOWN_LENGTH)
- {
- thisSize = * (size32_t *) inData;
- inData += sizeof(size32_t);
- }
- rc = cass_collection_append_bytes(collection, (const cass_byte_t*) inData, thisSize);
- break;
- }
- checkBind(rc, name);
- inData += thisSize;
- }
- checkBind(cass_statement_bind_collection(stmtInfo->queryStatement(),
- checkNextParam(name),
- collection),
- name);
- }
- virtual void importFunction(size32_t lenChars, const char *text)
- {
- throwUnexpected();
- }
- virtual void compileEmbeddedScript(size32_t chars, const char *_script)
- {
- // Incoming script is not necessarily null terminated. Note that the chars refers to utf8 characters and not bytes.
- size32_t len = rtlUtf8Size(chars, _script);
- queryString.set(_script, len);
- const char *script = queryString.get(); // Now null terminated
- if ((flags & (EFnoreturn|EFnoparams)) == (EFnoreturn|EFnoparams))
- {
- loop
- {
- const char *nextScript = findUnquoted(script, ';');
- if (!nextScript)
- {
- // script should be pointing at only trailing whitespace, else it's a "missing ;" error
- break;
- }
- CassandraStatement statement(cass_statement_new_n(script, nextScript-script, 0));
- CassandraFuture future(cass_session_execute(cluster->querySession(), statement));
- future.wait("execute statement");
- script = nextScript;
- }
- }
- else
- {
- if ((flags & EFnoparams) == 0)
- numParams = countBindings(script);
- else
- numParams = 0;
- stmtInfo.setown(cluster->createStatementInfo(script, numParams, batchMode, pageSize));
- }
- }
- virtual void callFunction()
- {
- // Does not seem to be a way to check number of parameters expected...
- // if (nextParam != cass_statement_bind_count(stmtInfo))
- // fail("Not enough parameters");
- try
- {
- if (stmtInfo && !stmtInfo->hasResult())
- lazyExecute();
- }
- catch (IException *E)
- {
- StringBuffer msg;
- E->errorMessage(msg);
- msg.appendf(" (processing query %s)", queryString.get());
- throw makeStringException(E->errorCode(), msg);
- }
- }
- protected:
- void lazyExecute()
- {
- if (inputStream)
- inputStream->executeAll(stmtInfo);
- else
- stmtInfo->execute();
- }
- const CassValue *getScalarResult()
- {
- if (!stmtInfo->next())
- typeError("scalar", NULL, NULL);
- if (cass_row_get_column(stmtInfo->queryRow(), 1))
- typeError("scalar", NULL, NULL);
- const CassValue *result = cass_row_get_column(stmtInfo->queryRow(), 0);
- if (!result)
- typeError("scalar", NULL, NULL);
- return result;
- }
- void checkSingleRow()
- {
- if (stmtInfo->rowCount() != 1)
- typeError("scalar", NULL, NULL);
- }
- unsigned countBindings(const char *query)
- {
- unsigned queryCount = 0;
- while ((query = findUnquoted(query, '?')) != NULL)
- queryCount++;
- return queryCount;
- }
- const char *findUnquoted(const char *query, char searchFor)
- {
- // Note - returns pointer to char AFTER the first occurrence of searchFor outside of quotes
- char inStr = '\0';
- char ch;
- while ((ch = *query++) != 0)
- {
- if (ch == inStr)
- inStr = false;
- else switch (ch)
- {
- case '\'':
- case '"':
- inStr = ch;
- break;
- case '\\':
- if (inStr && *query)
- query++;
- break;
- case '/':
- if (!inStr)
- {
- if (*query=='/')
- {
- while (*query && *query != '\n')
- query++;
- }
- else if (*query=='*')
- {
- query++;
- loop
- {
- if (!*query)
- fail("Unterminated comment in query string");
- if (*query=='*' && query[1]=='/')
- {
- query+= 2;
- break;
- }
- query++;
- }
- }
- }
- break;
- default:
- if (!inStr && ch==searchFor)
- return query;
- break;
- }
- }
- return NULL;
- }
- inline unsigned checkNextParam(const char *name)
- {
- if (nextParam == numParams)
- failx("Too many parameters supplied: No matching ? for parameter %s", name);
- return nextParam++;
- }
- inline void checkBind(CassError rc, const char *name)
- {
- if (rc != CASS_OK)
- {
- failx("While binding parameter %s: %s", name, cass_error_desc(rc));
- }
- }
- Owned<CassandraClusterSession> cluster;
- Owned<CassandraStatementInfo> stmtInfo;
- Owned<CassandraDatasetBinder> inputStream;
- const IContextLogger &logctx;
- unsigned flags;
- unsigned nextParam;
- unsigned numParams;
- StringAttr queryString;
- CassBatchType batchMode;
- unsigned pageSize;
- };
- class CassandraEmbedContext : public CInterfaceOf<IEmbedContext>
- {
- public:
- virtual IEmbedFunctionContext *createFunctionContext(unsigned flags, const char *options)
- {
- return createFunctionContextEx(NULL, flags, options);
- }
- virtual IEmbedFunctionContext *createFunctionContextEx(ICodeContext * ctx, unsigned flags, const char *options)
- {
- if (flags & EFimport)
- UNSUPPORTED("IMPORT");
- else
- return new CassandraEmbedFunctionContext(ctx ? ctx->queryContextLogger() : queryDummyContextLogger(), flags, options);
- }
- };
- extern IEmbedContext* getEmbedContext()
- {
- return new CassandraEmbedContext();
- }
- extern bool syntaxCheck(const char *script)
- {
- return true; // MORE
- }
- } // namespace
|