Line data Source code
1 : //
2 : // Copyright (c) 2015 Juniper Networks, Inc. All rights reserved.
3 : // Copyright Contributors to the OpenSDN project
4 : //
5 :
6 : #include <assert.h>
7 : #include <fstream>
8 :
9 : #include <boost/foreach.hpp>
10 : #include <boost/algorithm/string.hpp>
11 : #include <boost/algorithm/string/join.hpp>
12 : #include <boost/unordered_map.hpp>
13 : #include <boost/system/error_code.hpp>
14 :
15 : #include <linux/version.h>
16 : #if defined(RHEL_MAJOR) && (RHEL_MAJOR >= 9)
17 : #include <cassandra/cassandra.h>
18 : #else
19 : #include <cassandra.h>
20 : #endif
21 :
22 : #include <base/logging.h>
23 : #include <base/misc_utils.h>
24 : #include <base/task.h>
25 : #include <base/timer.h>
26 : #include <base/string_util.h>
27 : #include <base/address_util.h>
28 : #include <io/event_manager.h>
29 : #include <database/gendb_if.h>
30 : #include <database/gendb_constants.h>
31 : #include <database/cassandra/cql/cql_if.h>
32 : #include <database/cassandra/cql/cql_if_impl.h>
33 : #include <database/cassandra/cql/cql_lib_if.h>
34 :
35 : using namespace boost::system;
36 :
37 : #define CQLIF_DEBUG "CqlTraceBufDebug"
38 : #define CQLIF_INFO "CqlTraceBufInfo"
39 : #define CQLIF_ERR "CqlTraceBufErr"
40 :
41 : SandeshTraceBufferPtr CqlTraceDebugBuf(SandeshTraceBufferCreate(
42 : CQLIF_DEBUG, 10000));
43 : SandeshTraceBufferPtr CqlTraceInfoBuf(SandeshTraceBufferCreate(
44 : CQLIF_INFO, 10000));
45 : SandeshTraceBufferPtr CqlTraceErrBuf(SandeshTraceBufferCreate(
46 : CQLIF_ERR, 20000));
47 :
48 : #define CQLIF_DEBUG_TRACE(_Msg) \
49 : do { \
50 : std::stringstream _ss; \
51 : _ss << __func__ << ":" << __FILE__ << ":" << \
52 : __LINE__ << ": " << _Msg; \
53 : CQL_TRACE_TRACE(CqlTraceDebugBuf, _ss.str()); \
54 : } while (false) \
55 :
56 : #define CQLIF_INFO_TRACE(_Msg) \
57 : do { \
58 : std::stringstream _ss; \
59 : _ss << __func__ << ":" << __FILE__ << ":" << \
60 : __LINE__ << ": " << _Msg; \
61 : CQL_TRACE_TRACE(CqlTraceInfoBuf, _ss.str()); \
62 : } while (false) \
63 :
64 : #define CQLIF_ERR_TRACE(_Msg) \
65 : do { \
66 : std::stringstream _ss; \
67 : _ss << __func__ << ":" << __FILE__ << ":" << \
68 : __LINE__ << ": " << _Msg; \
69 : CQL_TRACE_TRACE(CqlTraceErrBuf, _ss.str()); \
70 : } while (false) \
71 :
72 : #define CASS_LIB_TRACE(_Level, _Msg) \
73 : do { \
74 : if (_Level == log4cplus::ERROR_LOG_LEVEL) { \
75 : CQL_TRACE_TRACE(CqlTraceErrBuf, _Msg); \
76 : } else if (_Level == log4cplus::DEBUG_LOG_LEVEL) { \
77 : CQL_TRACE_TRACE(CqlTraceDebugBuf, _Msg); \
78 : } else { \
79 : CQL_TRACE_TRACE(CqlTraceInfoBuf, _Msg); \
80 : } \
81 : } while (false) \
82 :
83 : #define CQLIF_LOG(_Level, _Msg) \
84 : do { \
85 : if (LoggingDisabled()) break; \
86 : log4cplus::Logger logger = log4cplus::Logger::getRoot(); \
87 : LOG4CPLUS_##_Level(logger, __func__ << ":" << __FILE__ << ":" << \
88 : __LINE__ << ": " << _Msg); \
89 : } while (false)
90 :
91 : #define CQLIF_LOG_ERR(_Msg) \
92 : do { \
93 : LOG(ERROR, __func__ << ":" << __FILE__ << ":" << __LINE__ << ": " \
94 : << _Msg); \
95 : } while (false)
96 :
97 : namespace cass {
98 : namespace cql {
99 : namespace impl {
100 :
101 : // CassString convenience structure
102 : struct CassString {
103 41774 : CassString() :
104 41774 : data(NULL),
105 41774 : length(0) {
106 41774 : }
107 :
108 : CassString(const char *data) :
109 : data(data),
110 : length(strlen(data)) {
111 : }
112 :
113 : CassString(const char* data, size_t length) :
114 : data(data),
115 : length(length) {
116 : }
117 :
118 : const char* data;
119 : size_t length;
120 : };
121 :
122 : // CassUuid encode and decode
123 4571 : static inline void encode_uuid(char* output, const CassUuid &uuid) {
124 4571 : uint64_t time_and_version = uuid.time_and_version;
125 4571 : output[3] = static_cast<char>(time_and_version & 0x00000000000000FFLL);
126 4571 : time_and_version >>= 8;
127 4571 : output[2] = static_cast<char>(time_and_version & 0x00000000000000FFLL);
128 4571 : time_and_version >>= 8;
129 4571 : output[1] = static_cast<char>(time_and_version & 0x00000000000000FFLL);
130 4571 : time_and_version >>= 8;
131 4571 : output[0] = static_cast<char>(time_and_version & 0x00000000000000FFLL);
132 4571 : time_and_version >>= 8;
133 :
134 4571 : output[5] = static_cast<char>(time_and_version & 0x00000000000000FFLL);
135 4571 : time_and_version >>= 8;
136 4571 : output[4] = static_cast<char>(time_and_version & 0x00000000000000FFLL);
137 4571 : time_and_version >>= 8;
138 :
139 4571 : output[7] = static_cast<char>(time_and_version & 0x00000000000000FFLL);
140 4571 : time_and_version >>= 8;
141 4571 : output[6] = static_cast<char>(time_and_version & 0x000000000000000FFLL);
142 :
143 4571 : uint64_t clock_seq_and_node = uuid.clock_seq_and_node;
144 41139 : for (size_t i = 0; i < 8; ++i) {
145 36568 : output[15 - i] = static_cast<char>(clock_seq_and_node & 0x00000000000000FFL);
146 36568 : clock_seq_and_node >>= 8;
147 : }
148 4571 : }
149 :
150 0 : static inline char* decode_uuid(char* input, CassUuid* output) {
151 0 : output->time_and_version = static_cast<uint64_t>(static_cast<uint8_t>(input[3]));
152 0 : output->time_and_version |= static_cast<uint64_t>(static_cast<uint8_t>(input[2])) << 8;
153 0 : output->time_and_version |= static_cast<uint64_t>(static_cast<uint8_t>(input[1])) << 16;
154 0 : output->time_and_version |= static_cast<uint64_t>(static_cast<uint8_t>(input[0])) << 24;
155 :
156 0 : output->time_and_version |= static_cast<uint64_t>(static_cast<uint8_t>(input[5])) << 32;
157 0 : output->time_and_version |= static_cast<uint64_t>(static_cast<uint8_t>(input[4])) << 40;
158 :
159 0 : output->time_and_version |= static_cast<uint64_t>(static_cast<uint8_t>(input[7])) << 48;
160 0 : output->time_and_version |= static_cast<uint64_t>(static_cast<uint8_t>(input[6])) << 56;
161 :
162 0 : output->clock_seq_and_node = 0;
163 0 : for (size_t i = 0; i < 8; ++i) {
164 0 : output->clock_seq_and_node |= static_cast<uint64_t>(static_cast<uint8_t>(input[15 - i])) << (8 * i);
165 : }
166 0 : return input + 16;
167 : }
168 :
169 78 : static const char * DbDataType2CassType(
170 : const GenDb::DbDataType::type &db_type) {
171 78 : switch (db_type) {
172 6 : case GenDb::DbDataType::AsciiType:
173 6 : return "ascii";
174 8 : case GenDb::DbDataType::LexicalUUIDType:
175 8 : return "uuid";
176 6 : case GenDb::DbDataType::TimeUUIDType:
177 6 : return "timeuuid";
178 20 : case GenDb::DbDataType::Unsigned8Type:
179 : case GenDb::DbDataType::Unsigned16Type:
180 : case GenDb::DbDataType::Unsigned32Type:
181 20 : return "int";
182 6 : case GenDb::DbDataType::Unsigned64Type:
183 6 : return "bigint";
184 6 : case GenDb::DbDataType::DoubleType:
185 6 : return "double";
186 8 : case GenDb::DbDataType::UTF8Type:
187 8 : return "text";
188 6 : case GenDb::DbDataType::InetType:
189 6 : return "inet";
190 6 : case GenDb::DbDataType::IntegerType:
191 6 : return "varint";
192 6 : case GenDb::DbDataType::BlobType:
193 6 : return "blob";
194 0 : default:
195 0 : assert(false && "Invalid data type");
196 : return "";
197 : }
198 : }
199 :
200 3 : static std::string DbDataTypes2CassTypes(
201 : const GenDb::DbDataTypeVec &v_db_types) {
202 3 : assert(!v_db_types.empty());
203 3 : return std::string(DbDataType2CassType(v_db_types[0]));
204 : }
205 :
206 108036 : static CassConsistency Db2CassConsistency(
207 : GenDb::DbConsistency::type dconsistency) {
208 108036 : switch (dconsistency) {
209 0 : case GenDb::DbConsistency::ANY:
210 0 : return CASS_CONSISTENCY_ANY;
211 0 : case GenDb::DbConsistency::ONE:
212 0 : return CASS_CONSISTENCY_ONE;
213 0 : case GenDb::DbConsistency::TWO:
214 0 : return CASS_CONSISTENCY_TWO;
215 0 : case GenDb::DbConsistency::THREE:
216 0 : return CASS_CONSISTENCY_THREE;
217 0 : case GenDb::DbConsistency::QUORUM:
218 0 : return CASS_CONSISTENCY_QUORUM;
219 0 : case GenDb::DbConsistency::ALL:
220 0 : return CASS_CONSISTENCY_ALL;
221 0 : case GenDb::DbConsistency::LOCAL_QUORUM:
222 0 : return CASS_CONSISTENCY_LOCAL_QUORUM;
223 0 : case GenDb::DbConsistency::EACH_QUORUM:
224 0 : return CASS_CONSISTENCY_EACH_QUORUM;
225 0 : case GenDb::DbConsistency::SERIAL:
226 0 : return CASS_CONSISTENCY_SERIAL;
227 0 : case GenDb::DbConsistency::LOCAL_SERIAL:
228 0 : return CASS_CONSISTENCY_LOCAL_SERIAL;
229 108061 : case GenDb::DbConsistency::LOCAL_ONE:
230 108061 : return CASS_CONSISTENCY_LOCAL_ONE;
231 0 : case GenDb::DbConsistency::UNKNOWN:
232 : default:
233 0 : return CASS_CONSISTENCY_UNKNOWN;
234 : }
235 : }
236 :
237 : // Cass Query Printer
238 : class CassQueryPrinter : public boost::static_visitor<> {
239 : public:
240 1 : CassQueryPrinter(std::ostream &os, bool quote_strings) :
241 1 : os_(os),
242 1 : quote_strings_(quote_strings) {
243 1 : }
244 411798 : CassQueryPrinter(std::ostream &os) :
245 411798 : os_(os),
246 411798 : quote_strings_(true) {
247 411798 : }
248 : template<typename T>
249 10612 : void operator()(const T &t) const {
250 10612 : os_ << t;
251 10629 : }
252 25 : void operator()(const boost::uuids::uuid &tuuid) const {
253 25 : os_ << to_string(tuuid);
254 25 : }
255 : // uint8_t must be handled specially because ostream sees
256 : // uint8_t as a text type instead of an integer type
257 114844 : void operator()(const uint8_t &tu8) const {
258 114844 : os_ << (uint16_t)tu8;
259 114916 : }
260 113155 : void operator()(const std::string &tstring) const {
261 113155 : if (quote_strings_) {
262 113148 : os_ << "'" << tstring << "'";
263 : } else {
264 8 : os_ << tstring;
265 : }
266 113275 : }
267 : // CQL int is 32 bit signed integer
268 300154 : void operator()(const uint32_t &tu32) const {
269 300154 : os_ << (int32_t)tu32;
270 300846 : }
271 : // CQL bigint is 64 bit signed long
272 25 : void operator()(const uint64_t &tu64) const {
273 25 : os_ << (int64_t)tu64;
274 25 : }
275 0 : void operator()(const IpAddress &tipaddr) const {
276 0 : os_ << "'" << tipaddr << "'";
277 0 : }
278 : std::ostream &os_;
279 : bool quote_strings_;
280 : };
281 :
282 : //
283 : // CassStatement bind
284 : //
285 : class CassStatementIndexBinder : public boost::static_visitor<> {
286 : public:
287 0 : CassStatementIndexBinder(interface::CassLibrary *cci,
288 0 : CassStatement *statement) :
289 0 : cci_(cci),
290 0 : statement_(statement) {
291 0 : }
292 0 : void operator()(const boost::blank &tblank, size_t index) const {
293 0 : assert(false && "CassStatement bind to boost::blank not supported");
294 : }
295 0 : void operator()(const std::string &tstring, size_t index) const {
296 0 : CassError rc(cci_->CassStatementBindStringN(statement_, index,
297 : tstring.c_str(), tstring.length()));
298 0 : assert(rc == CASS_OK);
299 0 : }
300 0 : void operator()(const boost::uuids::uuid &tuuid, size_t index) const {
301 : CassUuid cuuid;
302 0 : decode_uuid((char *)&tuuid, &cuuid);
303 0 : CassError rc(cci_->CassStatementBindUuid(statement_, index, cuuid));
304 0 : assert(rc == CASS_OK);
305 0 : }
306 0 : void operator()(const uint8_t &tu8, size_t index) const {
307 0 : CassError rc(cci_->CassStatementBindInt32(statement_, index, tu8));
308 0 : assert(rc == CASS_OK);
309 0 : }
310 0 : void operator()(const uint16_t &tu16, size_t index) const {
311 0 : CassError rc(cci_->CassStatementBindInt32(statement_, index, tu16));
312 0 : assert(rc == CASS_OK);
313 0 : }
314 0 : void operator()(const uint32_t &tu32, size_t index) const {
315 0 : assert(tu32 <= (uint32_t)std::numeric_limits<int32_t>::max());
316 0 : CassError rc(cci_->CassStatementBindInt32(statement_, index,
317 0 : (cass_int32_t)tu32));
318 0 : assert(rc == CASS_OK);
319 0 : }
320 0 : void operator()(const uint64_t &tu64, size_t index) const {
321 0 : assert(tu64 <= (uint64_t)std::numeric_limits<int64_t>::max());
322 0 : CassError rc(cci_->CassStatementBindInt64(statement_, index,
323 0 : (cass_int64_t)tu64));
324 0 : assert(rc == CASS_OK);
325 0 : }
326 0 : void operator()(const double &tdouble, size_t index) const {
327 0 : CassError rc(cci_->CassStatementBindDouble(statement_, index,
328 : (cass_double_t)tdouble));
329 0 : assert(rc == CASS_OK);
330 0 : }
331 0 : void operator()(const IpAddress &tipaddr, size_t index) const {
332 : CassInet cinet;
333 0 : if (tipaddr.is_v4()) {
334 0 : boost::asio::ip::address_v4 tv4(tipaddr.to_v4());
335 0 : cinet = cci_->CassInetInitV4(GENERIC_RAW_ARRAY(tv4.to_bytes()));
336 : } else {
337 0 : boost::asio::ip::address_v6 tv6(tipaddr.to_v6());
338 0 : cinet = cci_->CassInetInitV6(GENERIC_RAW_ARRAY(tv6.to_bytes()));
339 : }
340 0 : CassError rc(cci_->CassStatementBindInet(statement_, index,
341 : cinet));
342 0 : assert(rc == CASS_OK);
343 0 : }
344 0 : void operator()(const GenDb::Blob &tblob, size_t index) const {
345 0 : CassError rc(cci_->CassStatementBindBytes(statement_, index,
346 : tblob.data(), tblob.size()));
347 0 : assert(rc == CASS_OK);
348 0 : }
349 : interface::CassLibrary *cci_;
350 : CassStatement *statement_;
351 : };
352 :
353 : class CassStatementNameBinder : public boost::static_visitor<> {
354 : public:
355 0 : CassStatementNameBinder(interface::CassLibrary *cci,
356 0 : CassStatement *statement) :
357 0 : cci_(cci),
358 0 : statement_(statement) {
359 0 : }
360 0 : void operator()(const boost::blank &tblank, const char *name) const {
361 0 : assert(false && "CassStatement bind to boost::blank not supported");
362 : }
363 0 : void operator()(const std::string &tstring, const char *name) const {
364 0 : CassError rc(cci_->CassStatementBindStringByNameN(statement_, name,
365 : strlen(name), tstring.c_str(), tstring.length()));
366 0 : assert(rc == CASS_OK);
367 0 : }
368 0 : void operator()(const boost::uuids::uuid &tuuid, const char *name) const {
369 : CassUuid cuuid;
370 0 : decode_uuid((char *)&tuuid, &cuuid);
371 0 : CassError rc(cci_->CassStatementBindUuidByName(statement_, name,
372 : cuuid));
373 0 : assert(rc == CASS_OK);
374 0 : }
375 0 : void operator()(const uint8_t &tu8, const char *name) const {
376 0 : CassError rc(cci_->CassStatementBindInt32ByName(statement_, name,
377 0 : tu8));
378 0 : assert(rc == CASS_OK);
379 0 : }
380 0 : void operator()(const uint16_t &tu16, const char *name) const {
381 0 : CassError rc(cci_->CassStatementBindInt32ByName(statement_, name,
382 0 : tu16));
383 0 : assert(rc == CASS_OK);
384 0 : }
385 0 : void operator()(const uint32_t &tu32, const char *name) const {
386 0 : assert(tu32 <= (uint32_t)std::numeric_limits<int32_t>::max());
387 0 : CassError rc(cci_->CassStatementBindInt32ByName(statement_, name,
388 0 : (cass_int32_t)tu32));
389 0 : assert(rc == CASS_OK);
390 0 : }
391 0 : void operator()(const uint64_t &tu64, const char *name) const {
392 0 : assert(tu64 <= (uint64_t)std::numeric_limits<int64_t>::max());
393 0 : CassError rc(cci_->CassStatementBindInt64ByName(statement_, name,
394 0 : (cass_int64_t)tu64));
395 0 : assert(rc == CASS_OK);
396 0 : }
397 0 : void operator()(const double &tdouble, const char *name) const {
398 0 : CassError rc(cci_->CassStatementBindDoubleByName(statement_, name,
399 : (cass_double_t)tdouble));
400 0 : assert(rc == CASS_OK);
401 0 : }
402 0 : void operator()(const IpAddress &tipaddr, const char *name) const {
403 : CassInet cinet;
404 0 : if (tipaddr.is_v4()) {
405 0 : boost::asio::ip::address_v4 tv4(tipaddr.to_v4());
406 0 : cinet = cci_->CassInetInitV4(GENERIC_RAW_ARRAY(tv4.to_bytes()));
407 : } else {
408 0 : boost::asio::ip::address_v6 tv6(tipaddr.to_v6());
409 0 : cinet = cci_->CassInetInitV6(GENERIC_RAW_ARRAY(tv6.to_bytes()));
410 : }
411 0 : CassError rc(cci_->CassStatementBindInetByName(statement_, name,
412 : cinet));
413 0 : assert(rc == CASS_OK);
414 0 : }
415 0 : void operator()(const GenDb::Blob &tblob, const char *name) const {
416 0 : CassError rc(cci_->CassStatementBindBytesByNameN(statement_, name,
417 : strlen(name), tblob.data(), tblob.size()));
418 0 : assert(rc == CASS_OK);
419 0 : }
420 : interface::CassLibrary *cci_;
421 : CassStatement *statement_;
422 : };
423 :
424 : static const char * kQCompactionStrategy(
425 : "compaction = {'class': "
426 : "'org.apache.cassandra.db.compaction.%s'}");
427 : static const std::string kQGCGraceSeconds("gc_grace_seconds = 0");
428 : static const std::string kQReadRepairChanceDTCS(
429 : "read_repair_chance = 0.0");
430 :
431 : //
432 : // Cf2CassCreateTableIfNotExists
433 : //
434 :
435 1 : std::string StaticCf2CassCreateTableIfNotExists(const GenDb::NewCf &cf,
436 : const std::string &compaction_strategy) {
437 1 : std::ostringstream query;
438 : // Table name
439 1 : query << "CREATE TABLE IF NOT EXISTS " << cf.cfname_ << " ";
440 : // Row key
441 1 : const GenDb::DbDataTypeVec &rkeys(cf.partition_keys_);
442 1 : assert(rkeys.size() == 1);
443 1 : query << "(key " << DbDataType2CassType(rkeys[0]) <<
444 1 : " PRIMARY KEY";
445 : // Columns
446 1 : const GenDb::NewCf::ColumnMap &cfcolumns(cf.cfcolumns_);
447 1 : assert(!cfcolumns.empty());
448 25 : BOOST_FOREACH(const GenDb::NewCf::ColumnMap::value_type &cfcolumn,
449 : cfcolumns) {
450 12 : query << ", \"" << cfcolumn.first << "\" " <<
451 12 : DbDataType2CassType(cfcolumn.second);
452 : }
453 : char cbuf[512];
454 1 : int n(snprintf(cbuf, sizeof(cbuf), kQCompactionStrategy,
455 : compaction_strategy.c_str()));
456 1 : assert(!(n < 0 || n >= (int)sizeof(cbuf)));
457 :
458 : // The compaction strategy DateTieredCompactionStrategy precludes
459 : // using read repair, because of the way timestamps are checked for
460 : // DTCS compaction.In this case, you must set read_repair_chance to
461 : // zero. For other compaction strategies, read repair should be
462 : // enabled with a read_repair_chance value of 0.2 being typical
463 1 : if (compaction_strategy ==
464 : GenDb::g_gendb_constants.DATE_TIERED_COMPACTION_STRATEGY) {
465 0 : query << ") WITH " << std::string(cbuf) << " AND " <<
466 0 : kQReadRepairChanceDTCS << " AND " << kQGCGraceSeconds;
467 : } else {
468 2 : query << ") WITH " << std::string(cbuf) << " AND " <<
469 2 : kQGCGraceSeconds;
470 : }
471 :
472 2 : return query.str();
473 1 : }
474 :
475 4 : std::string DynamicCf2CassCreateTableIfNotExists(const GenDb::NewCf &cf,
476 : const std::string &compaction_strategy,
477 : boost::system::error_code *ec) {
478 4 : std::ostringstream query;
479 :
480 4 : *ec = errc::make_error_code(errc::success);
481 : // sanity check - # of clustering_columns cannot be 0
482 : // for dynamic tables
483 4 : if (cf.clustering_columns_.size() == 0) {
484 1 : *ec = errc::make_error_code(errc::invalid_argument);
485 1 : return query.str();
486 : }
487 :
488 : // Table name
489 3 : query << "CREATE TABLE IF NOT EXISTS " << cf.cfname_ << " (";
490 : // Row key
491 3 : const GenDb::DbDataTypeVec &rkeys(cf.partition_keys_);
492 3 : int rk_size(rkeys.size());
493 28 : for (int i = 0; i < rk_size; i++) {
494 25 : if (i) {
495 22 : int key_num(i + 1);
496 22 : query << "key" << key_num;
497 : } else {
498 3 : query << "key";
499 : }
500 25 : query << " " << DbDataType2CassType(rkeys[i]) << ", ";
501 : }
502 : // clustering columns
503 3 : const GenDb::DbDataTypeVec &clustering_columns(cf.clustering_columns_);
504 3 : int ccn_size(clustering_columns.size());
505 28 : for (int i = 0; i < ccn_size; i++) {
506 25 : int cnum(i + 1);
507 25 : query << "column" << cnum << " " <<
508 25 : DbDataType2CassType(clustering_columns[i]) << ", ";
509 : }
510 : // columns
511 3 : const GenDb::DbDataTypeVec &columns(cf.columns_);
512 3 : int cn_size(columns.size());
513 15 : for (int i = 0; i < cn_size; i++) {
514 12 : int cnum(i + 1 + ccn_size);
515 12 : query << "column" << cnum << " " <<
516 12 : DbDataType2CassType(columns[i]) << ", ";
517 : }
518 : // Value
519 3 : const GenDb::DbDataTypeVec &values(cf.value_);
520 3 : if (values.size() > 0) {
521 3 : query << "value" << " " << DbDataTypes2CassTypes(values) << ", ";
522 : }
523 : // Primary Key
524 3 : query << "PRIMARY KEY (";
525 3 : std::ostringstream rkey_ss;
526 28 : for (int i = 0; i < rk_size; i++) {
527 25 : if (i) {
528 22 : int key_num(i + 1);
529 22 : rkey_ss << ", key" << key_num;
530 : } else {
531 3 : rkey_ss << "key";
532 : }
533 : }
534 3 : if (rk_size >= 2) {
535 2 : query << "(" << rkey_ss.str() << "), ";
536 : } else {
537 1 : query << rkey_ss.str() << ", ";
538 : }
539 28 : for (int i = 0; i < ccn_size; i++) {
540 25 : int cnum(i + 1);
541 25 : if (i) {
542 22 : query << ", ";
543 : }
544 25 : query << "column" << cnum;
545 : }
546 : char cbuf[512];
547 3 : int n(snprintf(cbuf, sizeof(cbuf), kQCompactionStrategy,
548 : compaction_strategy.c_str()));
549 3 : assert(!(n < 0 || n >= (int)sizeof(cbuf)));
550 :
551 : // The compaction strategy DateTieredCompactionStrategy precludes
552 : // using read repair, because of the way timestamps are checked for
553 : // DTCS compaction.In this case, you must set read_repair_chance to
554 : // zero. For other compaction strategies, read repair should be
555 : // enabled with a read_repair_chance value of 0.2 being typical
556 3 : if (compaction_strategy ==
557 : GenDb::g_gendb_constants.DATE_TIERED_COMPACTION_STRATEGY) {
558 4 : query << ")) WITH " << std::string(cbuf) << " AND " <<
559 4 : kQReadRepairChanceDTCS << " AND " << kQGCGraceSeconds;
560 : } else {
561 2 : query << ")) WITH " << std::string(cbuf) << " AND " <<
562 2 : kQGCGraceSeconds;
563 : }
564 3 : return query.str();
565 4 : }
566 :
567 1 : static std::string DbColIndexMode2String(
568 : const GenDb::ColIndexMode::type index_mode) {
569 1 : switch (index_mode) {
570 0 : case GenDb::ColIndexMode::NONE:
571 0 : return "";
572 1 : case GenDb::ColIndexMode::PREFIX:
573 1 : return "PREFIX";
574 0 : case GenDb::ColIndexMode::CONTAINS:
575 0 : return "CONTAINS";
576 0 : default:
577 0 : assert(false && "INVALID");
578 : }
579 : }
580 :
581 : //
582 : // CassCreateIndexIfNotExists
583 : //
584 :
585 2 : std::string CassCreateIndexIfNotExists(const std::string &cfname,
586 : const std::string &column, const std::string &indexname,
587 : const GenDb::ColIndexMode::type index_mode) {
588 2 : std::ostringstream query;
589 :
590 2 : query << "CREATE ";
591 2 : if (index_mode != GenDb::ColIndexMode::NONE) {
592 1 : query << "CUSTOM "; // SASI
593 : }
594 : // Indexname
595 2 : query << "INDEX IF NOT EXISTS " << indexname << " ";
596 : // Index Column
597 2 : query << "ON " << cfname << "(\""<< column <<"\")";
598 : // Mode if SASI
599 2 : if (index_mode != GenDb::ColIndexMode::NONE) {
600 : query << " USING \'org.apache.cassandra.index.sasi.SASIIndex\' " <<
601 1 : "WITH OPTIONS = {\'mode\': \'" << DbColIndexMode2String(index_mode) << "\'}";
602 : }
603 2 : query << ";";
604 4 : return query.str();
605 2 : }
606 :
607 : //
608 : // Cf2CassInsertIntoTable
609 : //
610 :
611 1 : std::string StaticCf2CassInsertIntoTable(const GenDb::ColList *v_columns) {
612 1 : std::ostringstream query;
613 : // Table
614 1 : const std::string &table(v_columns->cfname_);
615 1 : query << "INSERT INTO " << table << " (";
616 1 : std::ostringstream values_ss;
617 1 : values_ss << "VALUES (";
618 1 : CassQueryPrinter values_printer(values_ss);
619 : // Row keys
620 1 : const GenDb::DbDataValueVec &rkeys(v_columns->rowkey_);
621 1 : int rk_size(rkeys.size());
622 9 : for (int i = 0; i < rk_size; i++) {
623 8 : if (i) {
624 7 : int key_num(i + 1);
625 7 : query << ", key" << key_num;
626 : } else {
627 1 : query << "key";
628 : }
629 8 : if (i) {
630 7 : values_ss << ", ";
631 : }
632 8 : boost::apply_visitor(values_printer, rkeys[i]);
633 : }
634 : // Columns
635 1 : int cttl(-1);
636 1 : CassQueryPrinter cnames_printer(query, false);
637 17 : BOOST_FOREACH(const GenDb::NewCol &column, v_columns->columns_) {
638 8 : assert(column.cftype_ == GenDb::NewCf::COLUMN_FAMILY_SQL);
639 : // Column Name
640 8 : query << ", ";
641 8 : const GenDb::DbDataValueVec &cnames(*column.name.get());
642 8 : assert(cnames.size() == 1);
643 : // Double quote column name strings
644 8 : query << "\"";
645 8 : boost::apply_visitor(cnames_printer, cnames[0]);
646 8 : query << "\"";
647 : // Column Values
648 8 : values_ss << ", ";
649 8 : const GenDb::DbDataValueVec &cvalues(*column.value.get());
650 8 : assert(cvalues.size() == 1);
651 8 : boost::apply_visitor(values_printer, cvalues[0]);
652 : // Column TTL
653 8 : cttl = column.ttl;
654 : }
655 1 : query << ") ";
656 1 : values_ss << ")";
657 1 : query << values_ss.str();
658 1 : if (cttl > 0) {
659 1 : query << " USING TTL " << cttl;
660 : }
661 2 : return query.str();
662 1 : }
663 :
664 1 : std::string DynamicCf2CassInsertIntoTable(const GenDb::ColList *v_columns) {
665 1 : std::ostringstream query;
666 : // Table
667 1 : const std::string &table(v_columns->cfname_);
668 1 : query << "INSERT INTO " << table << " (";
669 1 : std::ostringstream values_ss;
670 : // Row keys
671 1 : const GenDb::DbDataValueVec &rkeys(v_columns->rowkey_);
672 1 : int rk_size(rkeys.size());
673 1 : CassQueryPrinter values_printer(values_ss);
674 9 : for (int i = 0; i < rk_size; i++) {
675 8 : if (i) {
676 7 : int key_num(i + 1);
677 7 : query << ", key" << key_num;
678 : } else {
679 1 : query << "key";
680 : }
681 8 : boost::apply_visitor(values_printer, rkeys[i]);
682 8 : values_ss << ", ";
683 : }
684 : // Columns
685 1 : const GenDb::NewColVec &columns(v_columns->columns_);
686 1 : assert(columns.size() == 1);
687 1 : const GenDb::NewCol &column(columns[0]);
688 1 : assert(column.cftype_ == GenDb::NewCf::COLUMN_FAMILY_NOSQL);
689 : // Column Names
690 1 : const GenDb::DbDataValueVec &cnames(*column.name.get());
691 1 : int cn_size(cnames.size());
692 9 : for (int i = 0; i < cn_size; i++) {
693 8 : int cnum(i + 1);
694 8 : if (cnames.at(i).which() != GenDb::DB_VALUE_BLANK) {
695 8 : query << ", column" << cnum;
696 8 : boost::apply_visitor(values_printer, cnames[i]);
697 8 : if (i != cn_size - 1) {
698 7 : values_ss << ", ";
699 : }
700 : }
701 : }
702 : // Column Values
703 1 : const GenDb::DbDataValueVec &cvalues(*column.value.get());
704 1 : if (cvalues.size() > 0) {
705 1 : query << ", value) VALUES (";
706 1 : values_ss << ", ";
707 1 : boost::apply_visitor(values_printer, cvalues[0]);
708 : } else {
709 0 : query << ") VALUES (";
710 : }
711 1 : values_ss << ")";
712 1 : query << values_ss.str();
713 1 : if (column.ttl > 0) {
714 1 : query << " USING TTL " << column.ttl;
715 : }
716 2 : return query.str();
717 1 : }
718 :
719 : //
720 : // Cf2CassPrepareInsertIntoTable
721 : //
722 :
723 1 : std::string StaticCf2CassPrepareInsertIntoTable(const GenDb::NewCf &cf) {
724 1 : std::ostringstream query;
725 : // Table name
726 1 : query << "INSERT INTO " << cf.cfname_ << " ";
727 : // Row key
728 1 : const GenDb::DbDataTypeVec &rkeys(cf.partition_keys_);
729 1 : assert(rkeys.size() == 1);
730 1 : std::ostringstream values_ss;
731 1 : query << "(key";
732 1 : values_ss << ") VALUES (?";
733 : // Columns
734 1 : const GenDb::NewCf::ColumnMap &cfcolumns(cf.cfcolumns_);
735 1 : assert(!cfcolumns.empty());
736 25 : BOOST_FOREACH(const GenDb::NewCf::ColumnMap::value_type &cfcolumn,
737 : cfcolumns) {
738 12 : query << ", \"" << cfcolumn.first << "\"";
739 12 : values_ss << ", ?";
740 : }
741 1 : query << values_ss.str();
742 1 : query << ") USING TTL ?";
743 2 : return query.str();
744 1 : }
745 :
746 3 : std::string DynamicCf2CassPrepareInsertIntoTable(const GenDb::NewCf &cf,
747 : boost::system::error_code *ec) {
748 3 : std::ostringstream query;
749 :
750 3 : *ec = errc::make_error_code(errc::success);
751 : // sanity check - # of clustering_columns cannot be 0
752 : // for dynamic tables
753 3 : if (cf.clustering_columns_.size() == 0) {
754 1 : *ec = errc::make_error_code(errc::invalid_argument);
755 1 : return query.str();
756 : }
757 :
758 : // Table name
759 2 : query << "INSERT INTO " << cf.cfname_ << " (";
760 : // Row key
761 2 : const GenDb::DbDataTypeVec &rkeys(cf.partition_keys_);
762 2 : int rk_size(rkeys.size());
763 2 : std::ostringstream values_ss;
764 26 : for (int i = 0; i < rk_size; i++) {
765 24 : if (i) {
766 22 : int key_num(i + 1);
767 22 : query << "key" << key_num;
768 : } else {
769 2 : query << "key";
770 : }
771 24 : query << ", ";
772 24 : values_ss << "?, ";
773 : }
774 : // Clustering Column name
775 2 : const GenDb::DbDataTypeVec &clustering_columns(cf.clustering_columns_);
776 2 : int ccn_size(clustering_columns.size());
777 26 : for (int i = 0; i < ccn_size; i++) {
778 24 : int cnum(i + 1);
779 24 : query << "column" << cnum;
780 24 : values_ss << "?";
781 24 : if (i != ccn_size - 1) {
782 22 : query << ", ";
783 22 : values_ss << ", ";
784 : }
785 : }
786 : // Column name
787 2 : const GenDb::DbDataTypeVec &columns(cf.columns_);
788 2 : int cn_size(columns.size());
789 2 : if (cn_size > 0) {
790 1 : query << ", ";
791 1 : values_ss << ", ";
792 : }
793 14 : for (int i = 0; i < cn_size; i++) {
794 12 : int cnum(i + 1 + ccn_size);
795 12 : query << "column" << cnum;
796 12 : values_ss << "?";
797 12 : if (i != cn_size - 1) {
798 11 : query << ", ";
799 11 : values_ss << ", ";
800 : }
801 : }
802 : // Value
803 2 : const GenDb::DbDataTypeVec &values(cf.value_);
804 2 : if (values.size() > 0) {
805 2 : query << ", value";
806 2 : values_ss << ", ?";
807 : }
808 2 : query << ") VALUES (";
809 2 : values_ss << ")";
810 2 : query << values_ss.str();
811 2 : query << " USING TTL ?";
812 2 : return query.str();
813 3 : }
814 :
815 : //
816 : // Cf2CassPrepareBind
817 : //
818 :
819 0 : bool StaticCf2CassPrepareBind(interface::CassLibrary *cci,
820 : CassStatement *statement,
821 : const GenDb::ColList *v_columns) {
822 0 : CassStatementNameBinder values_binder(cci, statement);
823 : // Row keys
824 0 : const GenDb::DbDataValueVec &rkeys(v_columns->rowkey_);
825 0 : int rk_size(rkeys.size());
826 0 : size_t idx(0);
827 0 : for (; (int) idx < rk_size; idx++) {
828 0 : std::string rk_name;
829 0 : if (idx) {
830 0 : int key_num(idx + 1);
831 0 : rk_name = "key" + integerToString(key_num);
832 : } else {
833 0 : rk_name = "key";
834 : }
835 0 : boost::apply_visitor(boost::bind(values_binder, _1, rk_name.c_str()),
836 0 : rkeys[idx]);
837 0 : }
838 : // Columns
839 0 : int cttl(-1);
840 0 : BOOST_FOREACH(const GenDb::NewCol &column, v_columns->columns_) {
841 0 : assert(column.cftype_ == GenDb::NewCf::COLUMN_FAMILY_SQL);
842 0 : const GenDb::DbDataValueVec &cnames(*column.name.get());
843 0 : assert(cnames.size() == 1);
844 0 : assert(cnames[0].which() == GenDb::DB_VALUE_STRING);
845 0 : std::string cname(boost::get<std::string>(cnames[0]));
846 0 : const GenDb::DbDataValueVec &cvalues(*column.value.get());
847 0 : assert(cvalues.size() == 1);
848 0 : boost::apply_visitor(boost::bind(values_binder, _1, cname.c_str()),
849 0 : cvalues[0]);
850 : // Column TTL
851 0 : cttl = column.ttl;
852 0 : idx++;
853 0 : }
854 0 : CassError rc(cci->CassStatementBindInt32(statement, idx++,
855 : (cass_int32_t)cttl));
856 0 : assert(rc == CASS_OK);
857 0 : return true;
858 : }
859 :
860 0 : bool DynamicCf2CassPrepareBind(interface::CassLibrary *cci,
861 : CassStatement *statement,
862 : const GenDb::ColList *v_columns) {
863 0 : CassStatementIndexBinder values_binder(cci, statement);
864 : // Row keys
865 0 : const GenDb::DbDataValueVec &rkeys(v_columns->rowkey_);
866 0 : int rk_size(rkeys.size());
867 0 : size_t idx(0);
868 0 : for (; (int) idx < rk_size; idx++) {
869 0 : boost::apply_visitor(boost::bind(values_binder, _1, idx), rkeys[idx]);
870 : }
871 : // Columns
872 0 : const GenDb::NewColVec &columns(v_columns->columns_);
873 0 : assert(columns.size() == 1);
874 0 : const GenDb::NewCol &column(columns[0]);
875 0 : assert(column.cftype_ == GenDb::NewCf::COLUMN_FAMILY_NOSQL);
876 : // Column Names
877 0 : const GenDb::DbDataValueVec &cnames(*column.name.get());
878 0 : int cn_size(cnames.size());
879 0 : for (int i = 0; i < cn_size; i++, idx++) {
880 0 : boost::apply_visitor(boost::bind(values_binder, _1, idx), cnames[i]);
881 : }
882 : // Column Values
883 0 : const GenDb::DbDataValueVec &cvalues(*column.value.get());
884 0 : if (cvalues.size() > 0) {
885 0 : boost::apply_visitor(boost::bind(values_binder, _1, idx++),
886 0 : cvalues[0]);
887 : }
888 0 : CassError rc(cci->CassStatementBindInt32(statement, idx++,
889 0 : (cass_int32_t)column.ttl));
890 0 : assert(rc == CASS_OK);
891 0 : return true;
892 : }
893 :
894 108416 : static std::string CassSelectFromTableInternal(const std::string &table,
895 : const std::vector<GenDb::DbDataValueVec> &rkeys,
896 : const GenDb::ColumnNameRange &ck_range,
897 : const GenDb::FieldNamesToReadVec &read_vec,
898 : const GenDb::WhereIndexInfoVec &where_vec) {
899 108416 : std::ostringstream query;
900 : // Table
901 108736 : if (read_vec.empty()) {
902 108676 : query << "SELECT * FROM " << table;
903 : } else {
904 8 : query << "SELECT ";
905 8 : for (GenDb::FieldNamesToReadVec::const_iterator it = read_vec.begin();
906 28 : it != read_vec.end(); it++) {
907 20 : query << it->get<0>() << ",";
908 20 : bool read_timestamp = it->get<3>();
909 20 : if (read_timestamp) {
910 4 : query << "WRITETIME(" << it->get<0>() << "),";
911 : }
912 : }
913 8 : query.seekp(-1, query.cur);
914 8 : query << " FROM " << table;
915 : }
916 108664 : if (rkeys.size() == 1) {
917 108698 : GenDb::DbDataValueVec rkey = rkeys[0];
918 108318 : int rk_size(rkey.size());
919 108314 : CassQueryPrinter cprinter(query);
920 335473 : for (int i = 0; i < rk_size; i++) {
921 226794 : if (i) {
922 118486 : int key_num(i + 1);
923 118486 : query << " AND key" << key_num << "=";
924 : } else {
925 108308 : query << " WHERE key=";
926 : }
927 226986 : boost::apply_visitor(cprinter, rkey[i]);
928 : }
929 :
930 108674 : } else if (rkeys.size() > 1) {
931 5 : query << " WHERE key IN (";
932 45 : BOOST_FOREACH(GenDb::DbDataValueVec rkey, rkeys) {
933 20 : int rk_size(rkey.size());
934 20 : assert(rk_size == 1);
935 20 : CassQueryPrinter cprinter(query);
936 20 : boost::apply_visitor(cprinter, rkey[0]);
937 20 : query << ",";
938 20 : }
939 5 : query.seekp(-1, query.cur);
940 5 : query << ")";
941 : }
942 108423 : if (!where_vec.empty()) {
943 102138 : for (GenDb::WhereIndexInfoVec::const_iterator it = where_vec.begin();
944 208911 : it != where_vec.end(); ++it) {
945 106470 : std::ostringstream value_ss;
946 106798 : CassQueryPrinter value_vprinter(value_ss);
947 106802 : boost::apply_visitor(value_vprinter, it->get<2>());
948 106786 : query << " AND";
949 106785 : query << " " << it->get<0>();
950 106762 : query << " " << GenDb::Op::ToString(it->get<1>());
951 106724 : query << " " << value_ss.str();
952 106797 : }
953 : }
954 108447 : if (!ck_range.IsEmpty()) {
955 108023 : if (!ck_range.start_.empty()) {
956 88690 : int ck_start_size(ck_range.start_.size());
957 88690 : std::ostringstream start_ss;
958 88885 : start_ss << " " << GenDb::Op::ToString(ck_range.start_op_) << " (";
959 88871 : CassQueryPrinter start_vprinter(start_ss);
960 88868 : query << " AND (";
961 178179 : for (int i = 0; i < ck_start_size; i++) {
962 89282 : if (i) {
963 426 : query << ", ";
964 426 : start_ss << ", ";
965 : }
966 89282 : int cnum(i + 1);
967 89282 : query << "column" << cnum;
968 89300 : boost::apply_visitor(start_vprinter, ck_range.start_[i]);
969 : }
970 88897 : query << ")";
971 88885 : start_ss << ")";
972 88881 : query << start_ss.str();
973 88873 : }
974 108126 : if (!ck_range.finish_.empty()) {
975 108081 : int ck_finish_size(ck_range.finish_.size());
976 108078 : std::ostringstream finish_ss;
977 108169 : finish_ss << " " << GenDb::Op::ToString(ck_range.finish_op_) <<
978 108157 : " (";
979 108167 : CassQueryPrinter finish_vprinter(finish_ss);
980 108159 : query << " AND (";
981 224596 : for (int i = 0; i < ck_finish_size; i++) {
982 116407 : if (i) {
983 8261 : query << ", ";
984 8261 : finish_ss << ", ";
985 : }
986 116407 : int cnum(i + 1);
987 116407 : query << "column" << cnum;
988 116435 : boost::apply_visitor(finish_vprinter, ck_range.finish_[i]);
989 : }
990 108189 : query << ")";
991 108183 : finish_ss << ")";
992 108191 : query << finish_ss.str();
993 108173 : }
994 108173 : if (ck_range.count_) {
995 108177 : query << " LIMIT " << ck_range.count_;
996 : }
997 : }
998 108661 : if (where_vec.size() > 1) {
999 4341 : query << " ALLOW FILTERING";
1000 : }
1001 217293 : return query.str();
1002 108585 : }
1003 :
1004 107616 : std::string ClusteringKeyRangeAndIndexValue2CassSelectFromTable(
1005 : const std::string &table, const GenDb::DbDataValueVec &rkeys,
1006 : const GenDb::ColumnNameRange &ck_range,
1007 : const GenDb::WhereIndexInfoVec &where_vec,
1008 : const GenDb::FieldNamesToReadVec &read_vec) {
1009 107616 : std::vector<GenDb::DbDataValueVec> rkey_vec;
1010 108125 : rkey_vec.push_back(rkeys);
1011 : return CassSelectFromTableInternal(table, rkey_vec, ck_range,
1012 215840 : read_vec, where_vec);
1013 108068 : }
1014 :
1015 526 : std::string PartitionKey2CassSelectFromTable(const std::string &table,
1016 : const GenDb::DbDataValueVec &rkeys) {
1017 526 : std::vector<GenDb::DbDataValueVec> rkey_vec;
1018 526 : rkey_vec.push_back(rkeys);
1019 1052 : return CassSelectFromTableInternal(table, rkey_vec, GenDb::ColumnNameRange(),
1020 1052 : GenDb::FieldNamesToReadVec(),
1021 2104 : GenDb::WhereIndexInfoVec());
1022 526 : }
1023 :
1024 11 : std::string PartitionKeyAndClusteringKeyRange2CassSelectFromTable(
1025 : const std::string &table, const GenDb::DbDataValueVec &rkeys,
1026 : const GenDb::ColumnNameRange &ck_range,
1027 : const GenDb::FieldNamesToReadVec &read_vec) {
1028 11 : std::vector<GenDb::DbDataValueVec> rkey_vec;
1029 11 : rkey_vec.push_back(rkeys);
1030 : return CassSelectFromTableInternal(table, rkey_vec, ck_range, read_vec,
1031 33 : GenDb::WhereIndexInfoVec());
1032 11 : }
1033 :
1034 5 : std::string PartitionKeyAndClusteringKeyRange2CassSelectFromTable(
1035 : const std::string &table, const std::vector<GenDb::DbDataValueVec> &rkeys,
1036 : const GenDb::ColumnNameRange &ck_range,
1037 : const GenDb::FieldNamesToReadVec &read_vec) {
1038 : return CassSelectFromTableInternal(table, rkeys, ck_range, read_vec,
1039 10 : GenDb::WhereIndexInfoVec());
1040 : }
1041 :
1042 1 : std::string CassSelectFromTable(const std::string &table) {
1043 1 : std::vector<GenDb::DbDataValueVec> rkey_vec;
1044 : return CassSelectFromTableInternal(table, rkey_vec,
1045 2 : GenDb::ColumnNameRange(), GenDb::FieldNamesToReadVec(),
1046 4 : GenDb::WhereIndexInfoVec());
1047 1 : }
1048 :
1049 96333 : static GenDb::DbDataValue CassValue2DbDataValue(
1050 : interface::CassLibrary *cci, const CassValue *cvalue) {
1051 96333 : if (cci->CassValueIsNull(cvalue)) {
1052 28661 : return GenDb::DbDataValue();
1053 : }
1054 67700 : CassValueType cvtype(cci->GetCassValueType(cvalue));
1055 67712 : switch (cvtype) {
1056 41511 : case CASS_VALUE_TYPE_ASCII:
1057 : case CASS_VALUE_TYPE_VARCHAR:
1058 : case CASS_VALUE_TYPE_TEXT: {
1059 41511 : CassString ctstring;
1060 41540 : CassError rc(cci->CassValueGetString(cvalue, &ctstring.data,
1061 : &ctstring.length));
1062 41540 : assert(rc == CASS_OK);
1063 83077 : return std::string(ctstring.data, ctstring.length);
1064 : }
1065 4571 : case CASS_VALUE_TYPE_UUID: {
1066 : CassUuid ctuuid;
1067 4571 : CassError rc(cci->CassValueGetUuid(cvalue, &ctuuid));
1068 4571 : assert(rc == CASS_OK);
1069 : boost::uuids::uuid u;
1070 4571 : encode_uuid((char *)&u, ctuuid);
1071 4571 : return u;
1072 : }
1073 0 : case CASS_VALUE_TYPE_DOUBLE: {
1074 : cass_double_t ctdouble;
1075 0 : CassError rc(cci->CassValueGetDouble(cvalue, &ctdouble));
1076 0 : assert(rc == CASS_OK);
1077 0 : return (double)ctdouble;
1078 : }
1079 0 : case CASS_VALUE_TYPE_TINY_INT: {
1080 : cass_int8_t ct8;
1081 0 : CassError rc(cci->CassValueGetInt8(cvalue, &ct8));
1082 0 : assert(rc == CASS_OK);
1083 0 : return (uint8_t)ct8;
1084 : }
1085 0 : case CASS_VALUE_TYPE_SMALL_INT: {
1086 : cass_int16_t ct16;
1087 0 : CassError rc(cci->CassValueGetInt16(cvalue, &ct16));
1088 0 : assert(rc == CASS_OK);
1089 0 : return (uint16_t)ct16;
1090 : }
1091 17736 : case CASS_VALUE_TYPE_INT: {
1092 : cass_int32_t ct32;
1093 17736 : CassError rc(cci->CassValueGetInt32(cvalue, &ct32));
1094 17736 : assert(rc == CASS_OK);
1095 35471 : return (uint32_t)ct32;
1096 : }
1097 3463 : case CASS_VALUE_TYPE_BIGINT: {
1098 : cass_int64_t ct64;
1099 3463 : CassError rc(cci->CassValueGetInt64(cvalue, &ct64));
1100 3463 : assert(rc == CASS_OK);
1101 6926 : return (uint64_t)ct64;
1102 : }
1103 408 : case CASS_VALUE_TYPE_INET: {
1104 : CassInet ctinet;
1105 408 : CassError rc(cci->CassValueGetInet(cvalue, &ctinet));
1106 408 : assert(rc == CASS_OK);
1107 408 : IpAddress ipaddr;
1108 408 : if (ctinet.address_length == CASS_INET_V4_LENGTH) {
1109 : Ip4Address::bytes_type ipv4;
1110 408 : memcpy(GENERIC_RAW_ARRAY(ipv4), ctinet.address, CASS_INET_V4_LENGTH);
1111 408 : ipaddr = Ip4Address(ipv4);
1112 0 : } else if (ctinet.address_length == CASS_INET_V6_LENGTH) {
1113 : Ip6Address::bytes_type ipv6;
1114 0 : memcpy(GENERIC_RAW_ARRAY(ipv6), ctinet.address, CASS_INET_V6_LENGTH);
1115 0 : ipaddr = Ip6Address(ipv6);
1116 : } else {
1117 0 : assert(0);
1118 : }
1119 408 : return ipaddr;
1120 : }
1121 0 : case CASS_VALUE_TYPE_BLOB: {
1122 0 : const cass_byte_t *bytes(NULL);
1123 0 : size_t size(0);
1124 0 : CassError rc(cci->CassValueGetBytes(cvalue, &bytes, &size));
1125 0 : assert(rc == CASS_OK);
1126 0 : return GenDb::Blob(bytes, size);
1127 : }
1128 0 : case CASS_VALUE_TYPE_UNKNOWN: {
1129 : // null type
1130 0 : return GenDb::DbDataValue();
1131 : }
1132 23 : default: {
1133 23 : CQLIF_ERR_TRACE("Unhandled CassValueType: " << cvtype);
1134 0 : assert(false && "Unhandled value type");
1135 : return GenDb::DbDataValue();
1136 : }
1137 : }
1138 : }
1139 :
1140 0 : static bool PrepareSync(interface::CassLibrary *cci,
1141 : CassSession *session, const char* query,
1142 : CassPreparedPtr *prepared) {
1143 0 : CQLIF_DEBUG_TRACE( "PrepareSync: " << query);
1144 0 : CassFuturePtr future(cci->CassSessionPrepare(session, query), cci);
1145 0 : cci->CassFutureWait(future.get());
1146 :
1147 0 : CassError rc(cci->CassFutureErrorCode(future.get()));
1148 0 : if (rc != CASS_OK) {
1149 0 : CassString err;
1150 0 : cci->CassFutureErrorMessage(future.get(), &err.data, &err.length);
1151 0 : CQLIF_ERR_TRACE("PrepareSync: " << query << " FAILED: " <<
1152 : std::string(err.data, err.length));
1153 : } else {
1154 0 : *prepared = CassPreparedPtr(cci->CassFutureGetPrepared(future.get()),
1155 0 : cci);
1156 : }
1157 0 : return rc == CASS_OK;
1158 0 : }
1159 :
1160 550 : static bool ExecuteQuerySyncInternal(interface::CassLibrary *cci,
1161 : CassSession *session,
1162 : CassStatement *qstatement, CassResultPtr *result,
1163 : CassConsistency consistency) {
1164 550 : cci->CassStatementSetConsistency(qstatement, consistency);
1165 550 : CassFuturePtr future(cci->CassSessionExecute(session, qstatement), cci);
1166 550 : cci->CassFutureWait(future.get());
1167 :
1168 550 : CassError rc(cci->CassFutureErrorCode(future.get()));
1169 550 : if (rc != CASS_OK) {
1170 0 : CassString err;
1171 0 : cci->CassFutureErrorMessage(future.get(), &err.data, &err.length);
1172 0 : CQLIF_ERR_TRACE("SyncQuery: FAILED: " <<
1173 : std::string(err.data, err.length));
1174 : } else {
1175 550 : if (result) {
1176 1050 : *result = CassResultPtr(cci->CassFutureGetResult(future.get()),
1177 525 : cci);
1178 : }
1179 : }
1180 550 : return rc == CASS_OK;
1181 550 : }
1182 :
1183 25 : static bool ExecuteQuerySync(interface::CassLibrary *cci,
1184 : CassSession *session, const char *query, CassConsistency consistency) {
1185 25 : CQLIF_DEBUG_TRACE( "SyncQuery: " << query);
1186 25 : CassStatementPtr statement(cci->CassStatementNew(query, 0), cci);
1187 25 : return ExecuteQuerySyncInternal(cci, session, statement.get(), NULL,
1188 50 : consistency);
1189 25 : }
1190 :
1191 525 : static bool ExecuteQueryResultSync(interface::CassLibrary *cci,
1192 : CassSession *session, const char *query,
1193 : CassResultPtr *result, CassConsistency consistency) {
1194 525 : CQLIF_DEBUG_TRACE( "SyncQuery: " << query);
1195 525 : CassStatementPtr statement(cci->CassStatementNew(query, 0), cci);
1196 525 : return ExecuteQuerySyncInternal(cci, session, statement.get(), result,
1197 1050 : consistency);
1198 525 : }
1199 :
1200 0 : static bool ExecuteQueryStatementSync(interface::CassLibrary *cci,
1201 : CassSession *session, CassStatement *statement,
1202 : CassConsistency consistency) {
1203 0 : return ExecuteQuerySyncInternal(cci, session, statement, NULL,
1204 0 : consistency);
1205 : }
1206 :
1207 108184 : static GenDb::DbOpResult::type CassError2DbOpResult(CassError rc) {
1208 108184 : switch (rc) {
1209 108188 : case CASS_OK:
1210 108188 : return GenDb::DbOpResult::OK;
1211 0 : case CASS_ERROR_LIB_NO_HOSTS_AVAILABLE:
1212 : case CASS_ERROR_LIB_REQUEST_QUEUE_FULL:
1213 : case CASS_ERROR_LIB_NO_AVAILABLE_IO_THREAD:
1214 0 : return GenDb::DbOpResult::BACK_PRESSURE;
1215 0 : default:
1216 0 : return GenDb::DbOpResult::ERROR;
1217 : }
1218 : }
1219 :
1220 0 : static void DynamicCfGetResult(interface::CassLibrary *cci,
1221 : CassResultPtr *result, const GenDb::FieldNamesToReadVec &read_vec,
1222 : GenDb::NewColVec *v_columns) {
1223 : // Row iterator
1224 0 : CassIteratorPtr riterator(cci->CassIteratorFromResult(result->get()), cci);
1225 0 : while (cci->CassIteratorNext(riterator.get())) {
1226 0 : const CassRow *row(cci->CassIteratorGetRow(riterator.get()));
1227 0 : GenDb::DbDataValueVec *cnames(new GenDb::DbDataValueVec);
1228 0 : GenDb::DbDataValueVec *values(new GenDb::DbDataValueVec);
1229 0 : GenDb::DbDataValueVec *timestamps(new GenDb::DbDataValueVec);
1230 0 : int i = 0;
1231 0 : for (GenDb::FieldNamesToReadVec::const_iterator it = read_vec.begin();
1232 0 : it != read_vec.end(); it++) {
1233 0 : bool row_key = it->get<1>();
1234 0 : bool row_column = it->get<2>();
1235 0 : bool read_timestamp = it->get<3>();
1236 0 : if (row_key) {
1237 0 : i++;
1238 0 : continue;
1239 : }
1240 0 : const CassValue *cvalue(cci->CassRowGetColumn(row, i));
1241 0 : assert(cvalue);
1242 0 : GenDb::DbDataValue db_value(CassValue2DbDataValue(cci, cvalue));
1243 0 : if (row_column) {
1244 0 : cnames->push_back(db_value);
1245 : } else {
1246 0 : values->push_back(db_value);
1247 0 : if (read_timestamp) {
1248 0 : i++;
1249 0 : const CassValue *ctimestamp(cci->CassRowGetColumn(row, i));
1250 0 : assert(ctimestamp);
1251 0 : GenDb::DbDataValue time_value(CassValue2DbDataValue(cci, ctimestamp));
1252 0 : timestamps->push_back(time_value);
1253 0 : }
1254 : }
1255 0 : i++;
1256 0 : }
1257 0 : GenDb::NewCol *column(new GenDb::NewCol(cnames, values, 0, timestamps));
1258 0 : v_columns->push_back(column);
1259 : }
1260 0 : }
1261 :
1262 0 : static void DynamicCfGetResult(interface::CassLibrary *cci,
1263 : CassResultPtr *result, const GenDb::FieldNamesToReadVec &read_vec,
1264 : GenDb::ColListVec *v_col_list) {
1265 0 : std::auto_ptr<GenDb::ColList> col_list;
1266 : // Row iterator
1267 0 : CassIteratorPtr riterator(cci->CassIteratorFromResult(result->get()), cci);
1268 0 : while (cci->CassIteratorNext(riterator.get())) {
1269 0 : const CassRow *row(cci->CassIteratorGetRow(riterator.get()));
1270 0 : GenDb::DbDataValueVec rkey;
1271 0 : GenDb::DbDataValueVec *cnames(new GenDb::DbDataValueVec);
1272 0 : GenDb::DbDataValueVec *values(new GenDb::DbDataValueVec);
1273 0 : GenDb::DbDataValueVec *timestamps(new GenDb::DbDataValueVec);
1274 0 : int i = 0;
1275 0 : for (GenDb::FieldNamesToReadVec::const_iterator it = read_vec.begin();
1276 0 : it != read_vec.end(); it++) {
1277 0 : bool row_key = it->get<1>();
1278 0 : bool row_column = it->get<2>();
1279 0 : bool read_timestamp = it->get<3>();
1280 0 : if (row_key) {
1281 : // Partiiton key
1282 0 : const CassValue *cvalue(cci->CassRowGetColumn(row, i));
1283 0 : assert(cvalue);
1284 0 : GenDb::DbDataValue db_value(CassValue2DbDataValue(cci, cvalue));
1285 0 : rkey.push_back(db_value);
1286 0 : i++;
1287 0 : continue;
1288 0 : }
1289 0 : const CassValue *cvalue(cci->CassRowGetColumn(row, i));
1290 0 : assert(cvalue);
1291 0 : GenDb::DbDataValue db_value(CassValue2DbDataValue(cci, cvalue));
1292 0 : if (row_column) {
1293 0 : cnames->push_back(db_value);
1294 : } else {
1295 0 : values->push_back(db_value);
1296 0 : if (read_timestamp) {
1297 0 : i++;
1298 0 : const CassValue *ctimestamp(cci->CassRowGetColumn(row, i));
1299 0 : assert(ctimestamp);
1300 0 : GenDb::DbDataValue time_value(CassValue2DbDataValue(cci, ctimestamp));
1301 0 : timestamps->push_back(time_value);
1302 0 : }
1303 : }
1304 0 : i++;
1305 0 : }
1306 0 : GenDb::NewCol *column(new GenDb::NewCol(cnames, values, 0, timestamps));
1307 : // Do we need a new ColList?
1308 0 : if (!col_list.get()) {
1309 0 : col_list.reset(new GenDb::ColList);
1310 0 : col_list->rowkey_ = rkey;
1311 : }
1312 0 : if (rkey != col_list->rowkey_) {
1313 0 : v_col_list->push_back(col_list.release());
1314 0 : col_list.reset(new GenDb::ColList);
1315 0 : col_list->rowkey_ = rkey;
1316 : }
1317 0 : GenDb::NewColVec *v_columns(&col_list->columns_);
1318 0 : v_columns->push_back(column);
1319 0 : }
1320 0 : if (col_list.get()) {
1321 0 : v_col_list->push_back(col_list.release());
1322 : }
1323 0 : }
1324 :
1325 108625 : static void DynamicCfGetResult(interface::CassLibrary *cci,
1326 : CassResultPtr *result, size_t rk_count,
1327 : size_t ck_count, GenDb::NewColVec *v_columns) {
1328 : // Row iterator
1329 108625 : CassIteratorPtr riterator(cci->CassIteratorFromResult(result->get()), cci);
1330 113395 : while (cci->CassIteratorNext(riterator.get())) {
1331 4734 : const CassRow *row(cci->CassIteratorGetRow(riterator.get()));
1332 : // Iterate over columns
1333 4734 : size_t ccount(cci->CassResultColumnCount(result->get()));
1334 : // Clustering key
1335 4734 : GenDb::DbDataValueVec *cnames(new GenDb::DbDataValueVec);
1336 14955 : for (size_t i = rk_count; i < rk_count + ck_count; i++) {
1337 10221 : const CassValue *cvalue(cci->CassRowGetColumn(row, i));
1338 10221 : assert(cvalue);
1339 10221 : GenDb::DbDataValue db_value(CassValue2DbDataValue(cci, cvalue));
1340 10222 : cnames->push_back(db_value);
1341 10222 : }
1342 : // Values
1343 4734 : GenDb::DbDataValueVec *values(new GenDb::DbDataValueVec);
1344 90628 : for (size_t i = rk_count + ck_count; i < ccount; i++) {
1345 85893 : const CassValue *cvalue(cci->CassRowGetColumn(row, i));
1346 85897 : assert(cvalue);
1347 85897 : GenDb::DbDataValue db_value(CassValue2DbDataValue(cci, cvalue));
1348 85898 : values->push_back(db_value);
1349 85899 : }
1350 4735 : GenDb::NewCol *column(new GenDb::NewCol(cnames, values, 0));
1351 4734 : v_columns->push_back(column);
1352 : }
1353 108663 : }
1354 :
1355 1 : void DynamicCfGetResult(interface::CassLibrary *cci,
1356 : CassResultPtr *result, size_t rk_count,
1357 : size_t ck_count, GenDb::ColListVec *v_col_list) {
1358 1 : std::auto_ptr<GenDb::ColList> col_list;
1359 : // Row iterator
1360 1 : CassIteratorPtr riterator(cci->CassIteratorFromResult(result->get()), cci);
1361 4 : while (cci->CassIteratorNext(riterator.get())) {
1362 3 : const CassRow *row(cci->CassIteratorGetRow(riterator.get()));
1363 : // Iterate over columns
1364 3 : size_t ccount(cci->CassResultColumnCount(result->get()));
1365 : // Partiiton key
1366 3 : GenDb::DbDataValueVec rkey;
1367 6 : for (size_t i = 0; i < rk_count; i++) {
1368 3 : const CassValue *cvalue(cci->CassRowGetColumn(row, i));
1369 3 : assert(cvalue);
1370 3 : GenDb::DbDataValue db_value(CassValue2DbDataValue(cci, cvalue));
1371 3 : rkey.push_back(db_value);
1372 3 : }
1373 : // Clustering key
1374 3 : GenDb::DbDataValueVec *cnames(new GenDb::DbDataValueVec);
1375 6 : for (size_t i = rk_count; i < rk_count + ck_count; i++) {
1376 3 : const CassValue *cvalue(cci->CassRowGetColumn(row, i));
1377 3 : assert(cvalue);
1378 3 : GenDb::DbDataValue db_value(CassValue2DbDataValue(cci, cvalue));
1379 3 : cnames->push_back(db_value);
1380 3 : }
1381 : // Values
1382 3 : GenDb::DbDataValueVec *values(new GenDb::DbDataValueVec);
1383 6 : for (size_t i = rk_count + ck_count; i < ccount; i++) {
1384 3 : const CassValue *cvalue(cci->CassRowGetColumn(row, i));
1385 3 : assert(cvalue);
1386 3 : GenDb::DbDataValue db_value(CassValue2DbDataValue(cci, cvalue));
1387 3 : values->push_back(db_value);
1388 3 : }
1389 3 : GenDb::NewCol *column(new GenDb::NewCol(cnames, values, 0));
1390 : // Do we need a new ColList?
1391 3 : if (!col_list.get()) {
1392 1 : col_list.reset(new GenDb::ColList);
1393 1 : col_list->rowkey_ = rkey;
1394 : }
1395 3 : if (rkey != col_list->rowkey_) {
1396 1 : v_col_list->push_back(col_list.release());
1397 1 : col_list.reset(new GenDb::ColList);
1398 1 : col_list->rowkey_ = rkey;
1399 : }
1400 3 : GenDb::NewColVec *v_columns(&col_list->columns_);
1401 3 : v_columns->push_back(column);
1402 3 : }
1403 1 : if (col_list.get()) {
1404 1 : v_col_list->push_back(col_list.release());
1405 : }
1406 1 : }
1407 :
1408 25 : static void StaticCfGetResult(interface::CassLibrary *cci,
1409 : CassResultPtr *result, GenDb::NewColVec *v_columns) {
1410 : // Row iterator
1411 25 : CassIteratorPtr riterator(cci->CassIteratorFromResult(result->get()), cci);
1412 50 : while (cci->CassIteratorNext(riterator.get())) {
1413 25 : const CassRow *row(cci->CassIteratorGetRow(riterator.get()));
1414 : // Iterate over columns
1415 25 : size_t ccount(cci->CassResultColumnCount(result->get()));
1416 250 : for (size_t i = 0; i < ccount; i++) {
1417 225 : CassString cname;
1418 225 : CassError rc(cci->CassResultColumnName(result->get(), i,
1419 : &cname.data, &cname.length));
1420 225 : assert(rc == CASS_OK);
1421 225 : const CassValue *cvalue(cci->CassRowGetColumn(row, i));
1422 225 : assert(cvalue);
1423 225 : GenDb::DbDataValue db_value(CassValue2DbDataValue(cci, cvalue));
1424 225 : if (db_value.which() == GenDb::DB_VALUE_BLANK) {
1425 0 : continue;
1426 : }
1427 : GenDb::NewCol *column(new GenDb::NewCol(
1428 225 : std::string(cname.data, cname.length), db_value, 0));
1429 225 : v_columns->push_back(column);
1430 225 : }
1431 : }
1432 25 : }
1433 :
1434 1 : void StaticCfGetResult(interface::CassLibrary *cci,
1435 : CassResultPtr *result, size_t rk_count, GenDb::ColListVec *v_col_list) {
1436 1 : std::auto_ptr<GenDb::ColList> col_list;
1437 : // Row iterator
1438 1 : CassIteratorPtr riterator(cci->CassIteratorFromResult(result->get()), cci);
1439 4 : while (cci->CassIteratorNext(riterator.get())) {
1440 3 : const CassRow *row(cci->CassIteratorGetRow(riterator.get()));
1441 : // Iterate over columns
1442 3 : size_t ccount(cci->CassResultColumnCount(result->get()));
1443 : // Partiiton key
1444 3 : GenDb::DbDataValueVec rkey;
1445 6 : for (size_t i = 0; i < rk_count; i++) {
1446 3 : const CassValue *cvalue(cci->CassRowGetColumn(row, i));
1447 3 : assert(cvalue);
1448 3 : GenDb::DbDataValue db_value(CassValue2DbDataValue(cci, cvalue));
1449 3 : rkey.push_back(db_value);
1450 3 : }
1451 : // Do we need a new ColList?
1452 3 : if (!col_list.get()) {
1453 1 : col_list.reset(new GenDb::ColList);
1454 1 : col_list->rowkey_ = rkey;
1455 : }
1456 3 : if (rkey != col_list->rowkey_) {
1457 2 : v_col_list->push_back(col_list.release());
1458 2 : col_list.reset(new GenDb::ColList);
1459 2 : col_list->rowkey_ = rkey;
1460 : }
1461 3 : GenDb::NewColVec *v_columns(&col_list->columns_);
1462 12 : for (size_t i = 0; i < ccount; i++) {
1463 9 : CassString cname;
1464 9 : CassError rc(cci->CassResultColumnName(result->get(), i,
1465 : &cname.data, &cname.length));
1466 9 : assert(rc == CASS_OK);
1467 9 : const CassValue *cvalue(cci->CassRowGetColumn(row, i));
1468 9 : assert(cvalue);
1469 9 : GenDb::DbDataValue db_value(CassValue2DbDataValue(cci, cvalue));
1470 9 : if (db_value.which() == GenDb::DB_VALUE_BLANK) {
1471 0 : continue;
1472 : }
1473 : GenDb::NewCol *column(new GenDb::NewCol(
1474 9 : std::string(cname.data, cname.length), db_value, 0));
1475 9 : v_columns->push_back(column);
1476 9 : }
1477 3 : }
1478 1 : if (col_list.get()) {
1479 1 : v_col_list->push_back(col_list.release());
1480 : }
1481 1 : }
1482 :
1483 108183 : static void OnExecuteQueryAsync(CassFuture *future, void *data) {
1484 108183 : assert(data);
1485 : std::auto_ptr<CassAsyncQueryContext> ctx(
1486 108183 : boost::reinterpret_pointer_cast<CassAsyncQueryContext>(data));
1487 108187 : interface::CassLibrary *cci(ctx->cci_);
1488 108187 : CassError rc(cci->CassFutureErrorCode(future));
1489 108190 : GenDb::DbOpResult::type db_rc(CassError2DbOpResult(rc));
1490 108187 : if (rc != CASS_OK) {
1491 0 : CassString err;
1492 0 : cci->CassFutureErrorMessage(future, &err.data, &err.length);
1493 0 : CQLIF_ERR_TRACE("AsyncQuery: " << ctx->query_id_ << " FAILED: "
1494 : << std::string(err.data, err.length));
1495 0 : ctx->cb_(db_rc, std::auto_ptr<GenDb::ColList>());
1496 0 : return;
1497 : }
1498 108187 : if (ctx->result_ctx_) {
1499 108184 : CassQueryResultContext *rctx(ctx->result_ctx_.get());
1500 108184 : CassResultPtr result(cci->CassFutureGetResult(future), cci);
1501 : // In case of select parse the results
1502 108178 : if (cci->CassResultColumnCount(result.get())) {
1503 108179 : std::auto_ptr<GenDb::ColList> col_list(new GenDb::ColList);
1504 108173 : col_list->cfname_ = rctx->cf_name_;
1505 108174 : col_list->rowkey_ = rctx->row_key_;
1506 108137 : if (rctx->is_dynamic_cf_) {
1507 108136 : DynamicCfGetResult(cci, &result, rctx->rk_count_,
1508 108137 : rctx->ck_count_, &col_list->columns_);
1509 : } else {
1510 0 : StaticCfGetResult(cci, &result, &col_list->columns_);
1511 : }
1512 108173 : ctx->cb_(db_rc, col_list);
1513 108187 : return;
1514 108187 : }
1515 108187 : }
1516 0 : ctx->cb_(db_rc, std::auto_ptr<GenDb::ColList>());
1517 108191 : }
1518 :
1519 107938 : static void ExecuteQueryAsyncInternal(interface::CassLibrary *cci,
1520 : CassSession *session, const char *qid, CassStatement *qstatement,
1521 : CassConsistency consistency, CassAsyncQueryCallback cb,
1522 : CassQueryResultContext *rctx = NULL) {
1523 107938 : CQLIF_DEBUG_TRACE( "AsyncQuery: " << qid);
1524 108188 : cci->CassStatementSetConsistency(qstatement, consistency);
1525 108173 : CassFuturePtr future(cci->CassSessionExecute(session, qstatement), cci);
1526 : std::auto_ptr<CassAsyncQueryContext> ctx(
1527 108136 : new CassAsyncQueryContext(qid, cb, cci, rctx));
1528 108009 : cci->CassFutureSetCallback(future.get(), OnExecuteQueryAsync,
1529 108011 : ctx.release());
1530 108180 : }
1531 :
1532 0 : static void ExecuteQueryAsync(interface::CassLibrary *cci,
1533 : CassSession *session, const char *query,
1534 : CassConsistency consistency, CassAsyncQueryCallback cb) {
1535 0 : CQLIF_DEBUG_TRACE( "AsyncQuery: " << query);
1536 0 : CassStatementPtr statement(cci->CassStatementNew(query, 0), cci);
1537 0 : ExecuteQueryAsyncInternal(cci, session, query, statement.get(),
1538 : consistency, cb);
1539 0 : }
1540 :
1541 0 : static void ExecuteQueryStatementAsync(interface::CassLibrary *cci,
1542 : CassSession *session, const char *query_id, CassStatement *qstatement,
1543 : CassConsistency consistency, CassAsyncQueryCallback cb) {
1544 0 : ExecuteQueryAsyncInternal(cci, session, query_id, qstatement, consistency,
1545 : cb);
1546 0 : }
1547 :
1548 107692 : static void ExecuteQueryResultAsync(interface::CassLibrary *cci,
1549 : CassSession *session, const char *query, CassConsistency consistency,
1550 : CassAsyncQueryCallback cb, CassQueryResultContext *rctx) {
1551 107692 : CassStatementPtr statement(cci->CassStatementNew(query, 0), cci);
1552 108125 : ExecuteQueryAsyncInternal(cci, session, query, statement.get(),
1553 : consistency, cb, rctx);
1554 107918 : }
1555 :
1556 108031 : static bool DynamicCfGetResultAsync(interface::CassLibrary *cci,
1557 : CassSession *session, const char *query, CassConsistency consistency,
1558 : impl::CassAsyncQueryCallback cb, size_t rk_count, size_t ck_count,
1559 : const std::string &cfname, const GenDb::DbDataValueVec &row_key) {
1560 : std::auto_ptr<CassQueryResultContext> rctx(
1561 : new CassQueryResultContext(cfname, true, row_key,
1562 108031 : rk_count, ck_count));
1563 107857 : ExecuteQueryResultAsync(cci, session, query, consistency, cb,
1564 : rctx.release());
1565 108049 : return true;
1566 108028 : }
1567 :
1568 0 : static bool DynamicCfGetResultSync(interface::CassLibrary *cci,
1569 : CassSession *session, const char *query,
1570 : const GenDb::FieldNamesToReadVec &read_vec,
1571 : CassConsistency consistency, GenDb::NewColVec *v_columns) {
1572 0 : CassResultPtr result(NULL, cci);
1573 0 : bool success(ExecuteQueryResultSync(cci, session, query, &result,
1574 : consistency));
1575 0 : if (!success) {
1576 0 : return success;
1577 : }
1578 0 : DynamicCfGetResult(cci, &result, read_vec, v_columns);
1579 0 : return success;
1580 0 : }
1581 :
1582 0 : static bool DynamicCfGetResultSync(interface::CassLibrary *cci,
1583 : CassSession *session, const char *query,
1584 : const GenDb::FieldNamesToReadVec &read_vec,
1585 : CassConsistency consistency, GenDb::ColListVec *v_columns) {
1586 0 : CassResultPtr result(NULL, cci);
1587 0 : bool success(ExecuteQueryResultSync(cci, session, query, &result,
1588 : consistency));
1589 0 : if (!success) {
1590 0 : return success;
1591 : }
1592 0 : DynamicCfGetResult(cci, &result, read_vec, v_columns);
1593 0 : return success;
1594 0 : }
1595 :
1596 500 : static bool DynamicCfGetResultSync(interface::CassLibrary *cci,
1597 : CassSession *session, const char *query,
1598 : size_t rk_count, size_t ck_count, CassConsistency consistency,
1599 : GenDb::NewColVec *v_columns) {
1600 500 : CassResultPtr result(NULL, cci);
1601 500 : bool success(ExecuteQueryResultSync(cci, session, query, &result,
1602 : consistency));
1603 500 : if (!success) {
1604 0 : return success;
1605 : }
1606 500 : DynamicCfGetResult(cci, &result, rk_count, ck_count, v_columns);
1607 500 : return success;
1608 500 : }
1609 :
1610 0 : static bool DynamicCfGetResultSync(interface::CassLibrary *cci,
1611 : CassSession *session, const char *query,
1612 : size_t rk_count, size_t ck_count, CassConsistency consistency,
1613 : GenDb::ColListVec *v_col_list) {
1614 0 : CassResultPtr result(NULL, cci);
1615 0 : bool success(ExecuteQueryResultSync(cci, session, query, &result,
1616 : consistency));
1617 0 : if (!success) {
1618 0 : return success;
1619 : }
1620 0 : DynamicCfGetResult(cci, &result, rk_count, ck_count, v_col_list);
1621 0 : return success;
1622 0 : }
1623 :
1624 0 : static bool StaticCfGetResultAsync(interface::CassLibrary *cci,
1625 : CassSession *session, const char *query,
1626 : CassConsistency consistency, impl::CassAsyncQueryCallback cb,
1627 : const std::string& cfname, const GenDb::DbDataValueVec &row_key) {
1628 : std::auto_ptr<CassQueryResultContext> rctx(
1629 0 : new CassQueryResultContext(cfname, false, row_key));
1630 0 : ExecuteQueryResultAsync(cci, session, query, consistency, cb,
1631 : rctx.release());
1632 0 : return true;
1633 0 : }
1634 :
1635 25 : static bool StaticCfGetResultSync(interface::CassLibrary *cci,
1636 : CassSession *session, const char *query,
1637 : CassConsistency consistency, GenDb::NewColVec *v_columns) {
1638 25 : CassResultPtr result(NULL, cci);
1639 25 : bool success(ExecuteQueryResultSync(cci, session, query, &result,
1640 : consistency));
1641 25 : if (!success) {
1642 0 : return success;
1643 : }
1644 25 : StaticCfGetResult(cci, &result, v_columns);
1645 25 : return success;
1646 25 : }
1647 :
1648 0 : static bool StaticCfGetResultSync(interface::CassLibrary *cci,
1649 : CassSession *session, const char *query, size_t rk_count,
1650 : CassConsistency consistency, GenDb::ColListVec *v_col_list) {
1651 0 : CassResultPtr result(NULL, cci);
1652 0 : bool success(ExecuteQueryResultSync(cci, session, query, &result,
1653 : consistency));
1654 0 : if (!success) {
1655 0 : return success;
1656 : }
1657 0 : StaticCfGetResult(cci, &result, rk_count, v_col_list);
1658 0 : return success;
1659 0 : }
1660 :
1661 50 : static bool SyncFutureWait(interface::CassLibrary *cci,
1662 : CassFuture *future) {
1663 50 : cci->CassFutureWait(future);
1664 50 : CassError rc(cci->CassFutureErrorCode(future));
1665 50 : if (rc != CASS_OK) {
1666 0 : CassString err;
1667 0 : cci->CassFutureErrorMessage(future, &err.data, &err.length);
1668 0 : CQLIF_ERR_TRACE("SyncWait: FAILED: " <<
1669 : std::string(err.data, err.length));
1670 : }
1671 50 : return rc == CASS_OK;
1672 : }
1673 :
1674 326090 : static const CassTableMeta * GetCassTableMeta(
1675 : interface::CassLibrary *cci, const CassSchemaMeta *schema_meta,
1676 : const std::string &keyspace, const std::string &table, bool log_error) {
1677 : const CassKeyspaceMeta *keyspace_meta(
1678 326090 : cci->CassSchemaMetaKeyspaceByName(schema_meta, keyspace.c_str()));
1679 326116 : if (keyspace_meta == NULL) {
1680 0 : if (log_error) {
1681 0 : CQLIF_ERR_TRACE("No keyspace schema: Keyspace: " << keyspace <<
1682 : ", Table: " << table);
1683 : }
1684 0 : return NULL;
1685 : }
1686 326116 : std::string table_lower(table);
1687 326218 : boost::algorithm::to_lower(table_lower);
1688 : const CassTableMeta *table_meta(
1689 326149 : cci->CassKeyspaceMetaTableByName(keyspace_meta, table_lower.c_str()));
1690 326719 : if (table_meta == NULL) {
1691 25 : if (log_error) {
1692 0 : CQLIF_ERR_TRACE("No table schema: Keyspace: " << keyspace <<
1693 : ", Table: " << table_lower);
1694 : }
1695 25 : return NULL;
1696 : }
1697 326694 : return table_meta;
1698 326719 : }
1699 :
1700 150 : static bool IsCassTableMetaPresent(interface::CassLibrary *cci,
1701 : CassSession *session,
1702 : const std::string &keyspace, const std::string &table) {
1703 150 : impl::CassSchemaMetaPtr schema_meta(cci->CassSessionGetSchemaMeta(
1704 150 : session), cci);
1705 150 : if (schema_meta.get() == NULL) {
1706 0 : CQLIF_DEBUG_TRACE( "No schema meta: Keyspace: " << keyspace <<
1707 : ", Table: " << table);
1708 0 : return false;
1709 : }
1710 150 : bool log_error(false);
1711 150 : const CassTableMeta *table_meta(impl::GetCassTableMeta(cci,
1712 : schema_meta.get(), keyspace, table, log_error));
1713 150 : if (table_meta == NULL) {
1714 25 : return false;
1715 : }
1716 125 : return true;
1717 150 : }
1718 :
1719 217424 : static bool GetCassTableClusteringKeyCount(
1720 : interface::CassLibrary *cci,
1721 : CassSession *session, const std::string &keyspace,
1722 : const std::string &table, size_t *ck_count) {
1723 217424 : impl::CassSchemaMetaPtr schema_meta(cci->CassSessionGetSchemaMeta(
1724 217424 : session), cci);
1725 217712 : if (schema_meta.get() == NULL) {
1726 0 : CQLIF_ERR_TRACE("No schema meta: Keyspace: " << keyspace <<
1727 : ", Table: " << table);
1728 0 : return false;
1729 : }
1730 217797 : bool log_error(true);
1731 217797 : const CassTableMeta *table_meta(impl::GetCassTableMeta(cci,
1732 : schema_meta.get(), keyspace, table, log_error));
1733 217854 : if (table_meta == NULL) {
1734 0 : return false;
1735 : }
1736 217854 : *ck_count = cci->CassTableMetaClusteringKeyCount(table_meta);
1737 217825 : return true;
1738 217825 : }
1739 :
1740 108415 : static bool GetCassTablePartitionKeyCount(
1741 : interface::CassLibrary *cci, CassSession *session,
1742 : const std::string &keyspace, const std::string &table, size_t *rk_count) {
1743 108415 : impl::CassSchemaMetaPtr schema_meta(cci->CassSessionGetSchemaMeta(
1744 108415 : session), cci);
1745 108659 : if (schema_meta.get() == NULL) {
1746 0 : CQLIF_ERR_TRACE("No schema meta: Keyspace: " << keyspace <<
1747 : ", Table: " << table);
1748 0 : return false;
1749 : }
1750 108656 : bool log_error(true);
1751 108656 : const CassTableMeta *table_meta(impl::GetCassTableMeta(cci,
1752 : schema_meta.get(), keyspace, table, log_error));
1753 108679 : if (table_meta == NULL) {
1754 0 : return false;
1755 : }
1756 108679 : *rk_count = cci->CassTableMetaPartitionKeyCount(table_meta);
1757 108673 : return true;
1758 108673 : }
1759 :
1760 3255 : static log4cplus::LogLevel Cass2log4Level(CassLogLevel clevel) {
1761 3255 : switch (clevel) {
1762 0 : case CASS_LOG_DISABLED:
1763 0 : return log4cplus::OFF_LOG_LEVEL;
1764 0 : case CASS_LOG_CRITICAL:
1765 0 : return log4cplus::FATAL_LOG_LEVEL;
1766 25 : case CASS_LOG_ERROR:
1767 25 : return log4cplus::ERROR_LOG_LEVEL;
1768 0 : case CASS_LOG_WARN:
1769 0 : return log4cplus::WARN_LOG_LEVEL;
1770 175 : case CASS_LOG_INFO:
1771 175 : return log4cplus::INFO_LOG_LEVEL;
1772 3055 : case CASS_LOG_DEBUG:
1773 3055 : return log4cplus::DEBUG_LOG_LEVEL;
1774 0 : case CASS_LOG_TRACE:
1775 0 : return log4cplus::TRACE_LOG_LEVEL;
1776 0 : default:
1777 0 : return log4cplus::ALL_LOG_LEVEL;
1778 : }
1779 : }
1780 :
1781 3703 : static CassLogLevel Log4Level2CassLogLevel(log4cplus::LogLevel level) {
1782 3703 : switch (level) {
1783 0 : case log4cplus::OFF_LOG_LEVEL:
1784 0 : return CASS_LOG_DISABLED;
1785 0 : case log4cplus::FATAL_LOG_LEVEL:
1786 0 : return CASS_LOG_CRITICAL;
1787 0 : case log4cplus::ERROR_LOG_LEVEL:
1788 0 : return CASS_LOG_ERROR;
1789 0 : case log4cplus::WARN_LOG_LEVEL:
1790 0 : return CASS_LOG_WARN;
1791 0 : case log4cplus::INFO_LOG_LEVEL:
1792 0 : return CASS_LOG_INFO;
1793 3699 : case log4cplus::DEBUG_LOG_LEVEL:
1794 3699 : return CASS_LOG_DEBUG;
1795 4 : case log4cplus::TRACE_LOG_LEVEL:
1796 4 : return CASS_LOG_TRACE;
1797 0 : default:
1798 0 : assert(false && "Invalid Log4Level");
1799 : return CASS_LOG_DISABLED;
1800 : }
1801 : }
1802 :
1803 3255 : static void CassLibraryLog(const CassLogMessage* message, void *data) {
1804 3255 : if (LoggingDisabled()) {
1805 0 : return;
1806 : }
1807 3255 : log4cplus::LogLevel log4level(Cass2log4Level(message->severity));
1808 3255 : std::stringstream buf;
1809 3255 : buf << "CassLibrary: " << message->file << ":" << message->line <<
1810 3255 : " " << message->function << "] " << message->message;
1811 3255 : CASS_LIB_TRACE(log4level, buf.str());
1812 3255 : }
1813 :
1814 0 : static std::string LoadCertFile(const std::string &ca_certs_path) {
1815 0 : if (ca_certs_path.length() == 0) {
1816 0 : return std::string();
1817 : }
1818 0 : std::ifstream file(ca_certs_path.c_str());
1819 0 : if (!file) {
1820 0 : return std::string();
1821 : }
1822 : std::string content((std::istreambuf_iterator<char>(file)),
1823 0 : std::istreambuf_iterator<char>());
1824 0 : return content;
1825 0 : }
1826 :
1827 :
1828 : class WorkerTask : public Task {
1829 : public:
1830 : typedef boost::function<void(void)> FunctionPtr;
1831 0 : WorkerTask(FunctionPtr func, int task_id, int task_instance) :
1832 : Task(task_id, task_instance),
1833 0 : func_(func) {
1834 0 : }
1835 0 : bool Run() {
1836 0 : func_();
1837 0 : return true;
1838 : }
1839 0 : std::string Description() const {
1840 0 : return "cass::cql::impl::WorkerTask";
1841 : }
1842 : private:
1843 : FunctionPtr func_;
1844 : };
1845 :
1846 : } // namespace impl
1847 :
1848 : //
1849 : // CqlIfImpl
1850 : //
1851 3703 : CqlIfImpl::CqlIfImpl(EventManager *evm,
1852 : const std::vector<std::string> &cassandra_ips,
1853 : int cassandra_port,
1854 : const std::string &cassandra_user,
1855 : const std::string &cassandra_password,
1856 : bool use_ssl,
1857 : const std::string &ca_certs_path,
1858 3703 : interface::CassLibrary *cci) :
1859 3703 : evm_(evm),
1860 3703 : cci_(cci),
1861 3703 : cluster_(cci_->CassClusterNew(), cci_),
1862 3703 : ssl_(0, cci_),
1863 3703 : session_(cci_->CassSessionNew(), cci_),
1864 3703 : schema_session_(cci_->CassSessionNew(), cci_),
1865 3703 : keyspace_(),
1866 11109 : io_thread_count_(2) {
1867 : // Set session state to INIT
1868 3703 : session_state_ = SessionState::INIT;
1869 3703 : schema_session_state_ = SessionState::INIT;
1870 :
1871 3703 : if (cassandra_ips.size() > 0) {
1872 26 : schema_contact_point_ = cassandra_ips[0];
1873 26 : boost::system::error_code ec;
1874 26 : boost::asio::ip::address::from_string(cassandra_ips[0], ec);
1875 26 : if(ec.value() != 0){
1876 50 : schema_contact_point_ = GetHostIp(
1877 50 : evm->io_service(), cassandra_ips[0]);
1878 : }
1879 : }
1880 :
1881 3703 : if (use_ssl) {
1882 0 : ssl_ = impl::CassSslPtr(cci->CassSslNew(), cci_);
1883 : /* Only verify the certification and not the identity */
1884 0 : cci_->CassSslSetVerifyFlags(ssl_.get(), CASS_SSL_VERIFY_PEER_CERT);
1885 0 : std::string content = impl::LoadCertFile(ca_certs_path);
1886 0 : if (content.length() == 0) {
1887 0 : cci_->CassSslSetVerifyFlags(ssl_.get(), CASS_SSL_VERIFY_NONE);
1888 : } else {
1889 0 : cci_->CassSslAddTrustedCert(ssl_.get(), content);
1890 : }
1891 0 : cci_->CassClusterSetSsl(cluster_.get(), ssl_.get());
1892 0 : }
1893 3703 : std::string contact_points(boost::algorithm::join(cassandra_ips, ","));
1894 3703 : cci_->CassClusterSetContactPoints(cluster_.get(), contact_points.c_str());
1895 3703 : cci_->CassClusterSetPort(cluster_.get(), cassandra_port);
1896 : // Set credentials for plain text authentication
1897 3703 : if (!cassandra_user.empty() && !cassandra_password.empty()) {
1898 0 : cci_->CassClusterSetCredentials(cluster_.get(), cassandra_user.c_str(),
1899 : cassandra_password.c_str());
1900 : }
1901 : // Set number of IO threads to half the number of cores
1902 3703 : cci_->CassClusterSetNumThreadsIo(cluster_.get(), io_thread_count_);
1903 3703 : cci_->CassClusterSetPendingRequestsHighWaterMark(cluster_.get(), 10000);
1904 3703 : cci_->CassClusterSetPendingRequestsLowWaterMark(cluster_.get(), 5000);
1905 3703 : cci_->CassClusterSetWriteBytesHighWaterMark(cluster_.get(), 128000);
1906 3703 : cci_->CassClusterSetWriteBytesLowWaterMark(cluster_.get(), 96000);
1907 3703 : }
1908 :
1909 7406 : CqlIfImpl::~CqlIfImpl() {
1910 3703 : assert(session_state_ == SessionState::INIT ||
1911 : session_state_ == SessionState::DISCONNECTED);
1912 3703 : assert(schema_session_state_ == SessionState::INIT ||
1913 : schema_session_state_ == SessionState::DISCONNECTED);
1914 7406 : }
1915 :
1916 0 : bool CqlIfImpl::CreateKeyspaceIfNotExistsSync(const std::string &keyspace,
1917 : const std::string &replication_factor, CassConsistency consistency) {
1918 0 : if (schema_session_state_ != SessionState::CONNECTED) {
1919 0 : return false;
1920 : }
1921 : char buf[512];
1922 0 : int n(snprintf(buf, sizeof(buf), kQCreateKeyspaceIfNotExists,
1923 : keyspace.c_str(), replication_factor.c_str()));
1924 0 : if (n < 0 || n >= (int)sizeof(buf)) {
1925 0 : CQLIF_ERR_TRACE("FAILED (" << n << "): Keyspace: " <<
1926 : keyspace << ", RF: " << replication_factor);
1927 0 : return false;
1928 : }
1929 0 : return impl::ExecuteQuerySync(cci_, schema_session_.get(), buf,
1930 0 : consistency);
1931 : }
1932 :
1933 0 : bool CqlIfImpl::UseKeyspaceSyncOnSchemaSession(const std::string &keyspace,
1934 : CassConsistency consistency) {
1935 0 : if (schema_session_state_ != SessionState::CONNECTED) {
1936 0 : return false;
1937 : }
1938 : char buf[512];
1939 0 : int n(snprintf(buf, sizeof(buf), kQUseKeyspace, keyspace.c_str()));
1940 0 : if (n < 0 || n >= (int)sizeof(buf)) {
1941 0 : CQLIF_ERR_TRACE("FAILED (" << n << "): Keyspace: " <<
1942 : keyspace);
1943 0 : return false;
1944 : }
1945 0 : bool success(impl::ExecuteQuerySync(cci_, schema_session_.get(), buf,
1946 : consistency));
1947 0 : if (!success) {
1948 0 : return false;
1949 : }
1950 : // Update keyspace
1951 0 : keyspace_ = keyspace;
1952 0 : return success;
1953 : }
1954 :
1955 25 : bool CqlIfImpl::UseKeyspaceSync(const std::string &keyspace,
1956 : CassConsistency consistency) {
1957 25 : if (session_state_ != SessionState::CONNECTED) {
1958 0 : return false;
1959 : }
1960 : char buf[512];
1961 25 : int n(snprintf(buf, sizeof(buf), kQUseKeyspace, keyspace.c_str()));
1962 25 : if (n < 0 || n >= (int)sizeof(buf)) {
1963 0 : CQLIF_ERR_TRACE("FAILED (" << n << "): Keyspace: " <<
1964 : keyspace);
1965 0 : return false;
1966 : }
1967 25 : bool success(impl::ExecuteQuerySync(cci_, session_.get(), buf,
1968 : consistency));
1969 25 : if (!success) {
1970 0 : return false;
1971 : }
1972 : // Update keyspace
1973 25 : keyspace_ = keyspace;
1974 25 : return success;
1975 : }
1976 :
1977 0 : bool CqlIfImpl::CreateTableIfNotExistsSync(const GenDb::NewCf &cf,
1978 : const std::string &compaction_strategy, CassConsistency consistency) {
1979 0 : if (schema_session_state_ != SessionState::CONNECTED) {
1980 0 : return false;
1981 : }
1982 : // There are two types of tables - Static (SQL) and Dynamic (NOSQL)
1983 : // column family. Static column family has more or less fixed rows,
1984 : // and dynamic column family has wide rows
1985 0 : std::string query;
1986 0 : switch (cf.cftype_) {
1987 0 : case GenDb::NewCf::COLUMN_FAMILY_SQL:
1988 : {
1989 0 : query = impl::StaticCf2CassCreateTableIfNotExists(cf,
1990 0 : compaction_strategy);
1991 0 : break;
1992 : }
1993 0 : case GenDb::NewCf::COLUMN_FAMILY_NOSQL:
1994 : {
1995 0 : boost::system::error_code ec;
1996 0 : query = impl::DynamicCf2CassCreateTableIfNotExists(cf,
1997 0 : compaction_strategy, &ec);
1998 0 : if (ec) {
1999 0 : return false;
2000 : }
2001 0 : break;
2002 : }
2003 0 : default:
2004 : {
2005 0 : return false;
2006 : }
2007 : }
2008 0 : return impl::ExecuteQuerySync(cci_, schema_session_.get(), query.c_str(),
2009 0 : consistency);
2010 0 : }
2011 :
2012 0 : bool CqlIfImpl::CreateIndexIfNotExistsSync(const std::string &cfname,
2013 : const std::string &column, const std::string &indexname,
2014 : CassConsistency consistency, const GenDb::ColIndexMode::type index_mode) {
2015 0 : if (schema_session_state_ != SessionState::CONNECTED) {
2016 0 : return false;
2017 : }
2018 : std::string query(impl::CassCreateIndexIfNotExists(cfname, column,
2019 0 : indexname, index_mode));
2020 0 : return impl::ExecuteQuerySync(cci_, schema_session_.get(), query.c_str(),
2021 0 : consistency);
2022 0 : }
2023 :
2024 0 : bool CqlIfImpl::LocatePrepareInsertIntoTable(const GenDb::NewCf &cf) {
2025 0 : const std::string &table_name(cf.cfname_);
2026 0 : impl::CassPreparedPtr prepared(NULL, cci_);
2027 : // Check if the prepared statement exists
2028 0 : if (GetPrepareInsertIntoTable(table_name, &prepared)) {
2029 0 : return true;
2030 : }
2031 0 : bool success(PrepareInsertIntoTableSync(cf, &prepared));
2032 0 : if (!success) {
2033 0 : return success;
2034 : }
2035 : // Store the prepared statement into the map
2036 0 : std::scoped_lock lock(map_mutex_);
2037 0 : success = (insert_prepared_map_.insert(
2038 0 : std::make_pair(table_name, prepared))).second;
2039 0 : assert(success);
2040 0 : return success;
2041 0 : }
2042 :
2043 0 : bool CqlIfImpl::GetPrepareInsertIntoTable(const std::string &table_name,
2044 : impl::CassPreparedPtr *prepared) const {
2045 0 : std::scoped_lock lock(map_mutex_);
2046 : CassPreparedMapType::const_iterator it(
2047 0 : insert_prepared_map_.find(table_name));
2048 0 : if (it == insert_prepared_map_.end()) {
2049 0 : return false;
2050 : }
2051 0 : *prepared = it->second;
2052 0 : return true;
2053 0 : }
2054 :
2055 150 : bool CqlIfImpl::IsTablePresent(const std::string &table) {
2056 150 : if (session_state_ != SessionState::CONNECTED) {
2057 0 : return false;
2058 : }
2059 150 : return impl::IsCassTableMetaPresent(cci_, session_.get(), keyspace_,
2060 150 : table);
2061 : }
2062 :
2063 108851 : int CqlIfImpl::IsTableStatic(const std::string &table) {
2064 108851 : if (session_state_ != SessionState::CONNECTED) {
2065 0 : return false;
2066 : }
2067 : size_t ck_count;
2068 108967 : if (!impl::GetCassTableClusteringKeyCount(cci_, session_.get(),
2069 108970 : keyspace_, table, &ck_count)) {
2070 0 : return -1;
2071 : }
2072 108822 : if (ck_count == 0) {
2073 25 : return 1;
2074 : } else {
2075 108797 : return 0;
2076 : }
2077 : }
2078 :
2079 0 : bool CqlIfImpl::SelectFromTableAsync(const std::string &cfname,
2080 : const GenDb::DbDataValueVec &rkey, CassConsistency consistency,
2081 : impl::CassAsyncQueryCallback cb) {
2082 0 : if (session_state_ != SessionState::CONNECTED) {
2083 0 : return false;
2084 : }
2085 0 : std::string query(impl::PartitionKey2CassSelectFromTable(cfname,rkey));
2086 0 : if (IsTableStatic(cfname) == 1) {
2087 0 : return impl::StaticCfGetResultAsync(cci_, session_.get(),
2088 0 : query.c_str(), consistency, cb, cfname.c_str(), rkey);
2089 0 : } else if (IsTableStatic(cfname) == 0) {
2090 : size_t rk_count;
2091 0 : assert(impl::GetCassTablePartitionKeyCount(cci_, session_.get(),
2092 : keyspace_, cfname, &rk_count));
2093 : size_t ck_count;
2094 0 : assert(impl::GetCassTableClusteringKeyCount(cci_, session_.get(),
2095 : keyspace_, cfname, &ck_count));
2096 0 : return impl::DynamicCfGetResultAsync(cci_, session_.get(),
2097 : query.c_str(), consistency, cb, rk_count, ck_count,
2098 0 : cfname, rkey);
2099 : } else {
2100 0 : return false;
2101 : }
2102 0 : }
2103 :
2104 107757 : bool CqlIfImpl::SelectFromTableClusteringKeyRangeAndIndexValueAsync(
2105 : const std::string &cfname, const GenDb::DbDataValueVec &rkey,
2106 : const GenDb::ColumnNameRange &ck_range,
2107 : const GenDb::WhereIndexInfoVec &where_vec,
2108 : const GenDb::FieldNamesToReadVec &read_vec, CassConsistency consistency,
2109 : impl::CassAsyncQueryCallback cb) {
2110 107757 : if (session_state_ != SessionState::CONNECTED) {
2111 0 : return false;
2112 : }
2113 : std::string query(
2114 : impl::ClusteringKeyRangeAndIndexValue2CassSelectFromTable(cfname,
2115 108130 : rkey, ck_range, where_vec, read_vec));
2116 107833 : assert(IsTableDynamic(cfname));
2117 : size_t rk_count;
2118 108091 : assert(impl::GetCassTablePartitionKeyCount(cci_, session_.get(),
2119 : keyspace_, cfname, &rk_count));
2120 : size_t ck_count;
2121 108152 : assert(impl::GetCassTableClusteringKeyCount(cci_, session_.get(),
2122 : keyspace_, cfname, &ck_count));
2123 216186 : return impl::DynamicCfGetResultAsync(cci_, session_.get(),
2124 : query.c_str(), consistency, cb, rk_count, ck_count, cfname.c_str(),
2125 108023 : rkey);
2126 107997 : }
2127 :
2128 0 : bool CqlIfImpl::SelectFromTableClusteringKeyRangeAsync(
2129 : const std::string &cfname, const GenDb::DbDataValueVec &rkey,
2130 : const GenDb::ColumnNameRange &ck_range, CassConsistency consistency,
2131 : impl::CassAsyncQueryCallback cb) {
2132 0 : if (session_state_ != SessionState::CONNECTED) {
2133 0 : return false;
2134 : }
2135 : std::string query(
2136 : impl::PartitionKeyAndClusteringKeyRange2CassSelectFromTable(cfname,
2137 0 : rkey, ck_range));
2138 0 : assert(IsTableDynamic(cfname));
2139 : size_t rk_count;
2140 0 : assert(impl::GetCassTablePartitionKeyCount(cci_, session_.get(),
2141 : keyspace_, cfname, &rk_count));
2142 : size_t ck_count;
2143 0 : assert(impl::GetCassTableClusteringKeyCount(cci_, session_.get(),
2144 : keyspace_, cfname, &ck_count));
2145 0 : return impl::DynamicCfGetResultAsync(cci_, session_.get(),
2146 : query.c_str(), consistency, cb, rk_count, ck_count, cfname.c_str(),
2147 0 : rkey);
2148 0 : }
2149 :
2150 107879 : bool CqlIfImpl::IsTableDynamic(const std::string &table) {
2151 107879 : return !IsTableStatic(table);
2152 : }
2153 :
2154 0 : bool CqlIfImpl::InsertIntoTableSync(std::auto_ptr<GenDb::ColList> v_columns,
2155 : CassConsistency consistency) {
2156 0 : return InsertIntoTableInternal(v_columns, consistency, true, NULL);
2157 : }
2158 :
2159 0 : bool CqlIfImpl::InsertIntoTableAsync(std::auto_ptr<GenDb::ColList> v_columns,
2160 : CassConsistency consistency, impl::CassAsyncQueryCallback cb) {
2161 0 : return InsertIntoTableInternal(v_columns, consistency, false, cb);
2162 : }
2163 :
2164 0 : bool CqlIfImpl::InsertIntoTablePrepareAsync(std::auto_ptr<GenDb::ColList> v_columns,
2165 : CassConsistency consistency, impl::CassAsyncQueryCallback cb) {
2166 0 : return InsertIntoTablePrepareInternal(v_columns, consistency, false,
2167 0 : cb);
2168 : }
2169 :
2170 : // COLLECTOR_GLOBAL_TABLE is dynamic table with 6 indexed columns
2171 : // for object-id. Some of these might be blank. This creates tombstones
2172 : // so we dont want to prepare for inserts.
2173 0 : bool CqlIfImpl::IsInsertIntoTablePrepareSupported(const std::string &table) {
2174 0 : return (IsTableDynamic(table)) &&
2175 0 : (table != "MessageTablev2");
2176 : }
2177 :
2178 525 : bool CqlIfImpl::SelectFromTableSync(const std::string &cfname,
2179 : const GenDb::DbDataValueVec &rkey, CassConsistency consistency,
2180 : GenDb::NewColVec *out) {
2181 525 : if (session_state_ != SessionState::CONNECTED) {
2182 0 : return false;
2183 : }
2184 : std::string query(impl::PartitionKey2CassSelectFromTable(cfname,
2185 525 : rkey));
2186 525 : if (IsTableStatic(cfname) == 1) {
2187 25 : return impl::StaticCfGetResultSync(cci_, session_.get(),
2188 25 : query.c_str(), consistency, out);
2189 500 : } else if (IsTableStatic(cfname) == 0){
2190 : size_t rk_count;
2191 500 : assert(impl::GetCassTablePartitionKeyCount(cci_, session_.get(),
2192 : keyspace_, cfname, &rk_count));
2193 : size_t ck_count;
2194 500 : assert(impl::GetCassTableClusteringKeyCount(cci_, session_.get(),
2195 : keyspace_, cfname, &ck_count));
2196 500 : return impl::DynamicCfGetResultSync(cci_, session_.get(),
2197 500 : query.c_str(), rk_count, ck_count, consistency, out);
2198 : } else {
2199 0 : return false;
2200 : }
2201 525 : }
2202 :
2203 0 : bool CqlIfImpl::SelectFromTableSync(const std::string &cfname,
2204 : CassConsistency consistency, GenDb::ColListVec *out) {
2205 0 : if (session_state_ != SessionState::CONNECTED) {
2206 0 : return false;
2207 : }
2208 0 : std::string query(impl::CassSelectFromTable(cfname));
2209 : size_t rk_count;
2210 0 : assert(impl::GetCassTablePartitionKeyCount(cci_, session_.get(),
2211 : keyspace_, cfname, &rk_count));
2212 0 : if (IsTableStatic(cfname) == 1) {
2213 0 : return impl::StaticCfGetResultSync(cci_, session_.get(),
2214 0 : query.c_str(), rk_count, consistency, out);
2215 0 : } else if (IsTableStatic(cfname) == 0){
2216 : size_t ck_count;
2217 0 : assert(impl::GetCassTableClusteringKeyCount(cci_, session_.get(),
2218 : keyspace_, cfname, &ck_count));
2219 0 : return impl::DynamicCfGetResultSync(cci_, session_.get(),
2220 0 : query.c_str(), rk_count, ck_count, consistency, out);
2221 : } else {
2222 0 : return false;
2223 : }
2224 0 : }
2225 :
2226 :
2227 0 : bool CqlIfImpl::SelectFromTableClusteringKeyRangeFieldNamesSync(const std::string &cfname,
2228 : const GenDb::DbDataValueVec &rkey,
2229 : const GenDb::ColumnNameRange &ck_range, CassConsistency consistency,
2230 : const GenDb::FieldNamesToReadVec &read_vec,
2231 : GenDb::NewColVec *out) {
2232 0 : if (session_state_ != SessionState::CONNECTED) {
2233 0 : return false;
2234 : }
2235 : std::string query(
2236 : impl::PartitionKeyAndClusteringKeyRange2CassSelectFromTable(cfname,
2237 0 : rkey, ck_range, read_vec));
2238 0 : assert(IsTableDynamic(cfname));
2239 0 : return impl::DynamicCfGetResultSync(cci_, session_.get(),
2240 0 : query.c_str(), read_vec, consistency, out);
2241 0 : }
2242 :
2243 0 : bool CqlIfImpl::SelectFromTableClusteringKeyRangeFieldNamesSync(const std::string &cfname,
2244 : const std::vector<GenDb::DbDataValueVec> &rkeys,
2245 : const GenDb::ColumnNameRange &ck_range, CassConsistency consistency,
2246 : const GenDb::FieldNamesToReadVec &read_vec,
2247 : GenDb::ColListVec *out) {
2248 0 : if (session_state_ != SessionState::CONNECTED) {
2249 0 : return false;
2250 : }
2251 : std::string query(
2252 : impl::PartitionKeyAndClusteringKeyRange2CassSelectFromTable(cfname,
2253 0 : rkeys, ck_range, read_vec));
2254 0 : assert(IsTableDynamic(cfname));
2255 0 : return impl::DynamicCfGetResultSync(cci_, session_.get(),
2256 0 : query.c_str(), read_vec, consistency, out);
2257 0 : }
2258 :
2259 :
2260 0 : bool CqlIfImpl::SelectFromTableClusteringKeyRangeSync(const std::string &cfname,
2261 : const GenDb::DbDataValueVec &rkey,
2262 : const GenDb::ColumnNameRange &ck_range, CassConsistency consistency,
2263 : GenDb::NewColVec *out) {
2264 0 : if (session_state_ != SessionState::CONNECTED) {
2265 0 : return false;
2266 : }
2267 : std::string query(
2268 : impl::PartitionKeyAndClusteringKeyRange2CassSelectFromTable(cfname,
2269 0 : rkey, ck_range));
2270 0 : assert(IsTableDynamic(cfname));
2271 : size_t rk_count;
2272 0 : assert(impl::GetCassTablePartitionKeyCount(cci_, session_.get(),
2273 : keyspace_, cfname, &rk_count));
2274 : size_t ck_count;
2275 0 : assert(impl::GetCassTableClusteringKeyCount(cci_, session_.get(),
2276 : keyspace_, cfname, &ck_count));
2277 0 : return impl::DynamicCfGetResultSync(cci_, session_.get(),
2278 0 : query.c_str(), rk_count, ck_count, consistency, out);
2279 0 : }
2280 :
2281 0 : void CqlIfImpl::SetRequestTimeout(uint32_t timeout_ms) {
2282 0 : CQLIF_DEBUG_TRACE("request timeout set to " << timeout_ms);
2283 0 : cci_->CassClusterSetRequestTimeout(cluster_.get(), timeout_ms);
2284 0 : }
2285 :
2286 0 : bool CqlIfImpl::ConnectSchemaSync() {
2287 : /* If Connect is called multiple times due to DB failure,
2288 : * then it is better to delete previous session and use
2289 : * a new one to avoid gradual leak.
2290 : */
2291 0 : schema_session_.reset();
2292 0 : impl::CassSessionPtr schema_session(cci_->CassSessionNew(), cci_);
2293 0 : schema_session_.swap(schema_session);
2294 :
2295 : // First set the cluster whitelist filtering to just one node
2296 0 : cci_->CassClusterSetWhitelistFiltering(cluster_.get(),
2297 : schema_contact_point_.c_str());
2298 :
2299 0 : impl::CassFuturePtr future(cci_->CassSessionConnect(schema_session_.get(),
2300 0 : cluster_.get()), cci_);
2301 0 : bool success(impl::SyncFutureWait(cci_, future.get()));
2302 0 : if (success) {
2303 0 : schema_session_state_ = SessionState::CONNECTED;
2304 0 : CQLIF_INFO_TRACE("ConnectSchemaSync Done");
2305 : } else {
2306 0 : CQLIF_ERR_TRACE("ConnectSchemaSync FAILED");
2307 : }
2308 :
2309 0 : cci_->CassClusterSetWhitelistFiltering(cluster_.get(), "");
2310 0 : return success;
2311 0 : }
2312 :
2313 25 : bool CqlIfImpl::ConnectSync() {
2314 : /* If Connect is called multiple times due to DB failure,
2315 : * then it is better to delete previous session and use
2316 : * a new one to avoid gradual leak.
2317 : */
2318 25 : session_.reset();
2319 25 : impl::CassSessionPtr session(cci_->CassSessionNew(), cci_);
2320 25 : session_.swap(session);
2321 :
2322 25 : impl::CassFuturePtr future(cci_->CassSessionConnect(session_.get(),
2323 25 : cluster_.get()), cci_);
2324 25 : bool success(impl::SyncFutureWait(cci_, future.get()));
2325 25 : if (success) {
2326 25 : session_state_ = SessionState::CONNECTED;
2327 25 : CQLIF_INFO_TRACE("ConnectSync Done");
2328 : } else {
2329 0 : CQLIF_ERR_TRACE("ConnectSync FAILED");
2330 : }
2331 25 : return success;
2332 25 : }
2333 :
2334 25 : bool CqlIfImpl::DisconnectSync() {
2335 : // Close all session and pending queries
2336 25 : impl::CassFuturePtr future(cci_->CassSessionClose(session_.get()), cci_);
2337 25 : bool success(impl::SyncFutureWait(cci_, future.get()));
2338 25 : if (success) {
2339 25 : session_state_ = SessionState::DISCONNECTED;
2340 25 : CQLIF_INFO_TRACE("DisconnectSync Done");
2341 : } else {
2342 0 : CQLIF_ERR_TRACE("DisconnectSync FAILED");
2343 : }
2344 25 : return success;
2345 25 : }
2346 :
2347 0 : bool CqlIfImpl::DisconnectSchemaSync() {
2348 : // Close the schema session
2349 0 : impl::CassFuturePtr future(cci_->CassSessionClose(schema_session_.get()),
2350 0 : cci_);
2351 0 : bool success(impl::SyncFutureWait(cci_, future.get()));
2352 0 : if (success) {
2353 0 : schema_session_state_ = SessionState::DISCONNECTED;
2354 0 : CQLIF_INFO_TRACE("DisconnectSchemaSync Done");
2355 : } else {
2356 0 : CQLIF_ERR_TRACE("DisconnectSchemaSync FAILED");
2357 : }
2358 0 : return success;
2359 0 : }
2360 :
2361 27 : bool CqlIfImpl::GetMetrics(Metrics *metrics) const {
2362 27 : if (session_state_ != SessionState::CONNECTED) {
2363 0 : return false;
2364 : }
2365 : CassMetrics cass_metrics;
2366 27 : cci_->CassSessionGetMetrics(session_.get(), &cass_metrics);
2367 : // Requests
2368 27 : metrics->requests.min = cass_metrics.requests.min;
2369 27 : metrics->requests.max = cass_metrics.requests.max;
2370 27 : metrics->requests.mean = cass_metrics.requests.mean;
2371 27 : metrics->requests.stddev = cass_metrics.requests.stddev;
2372 27 : metrics->requests.median = cass_metrics.requests.median;
2373 27 : metrics->requests.percentile_75th =
2374 27 : cass_metrics.requests.percentile_75th;
2375 27 : metrics->requests.percentile_95th =
2376 27 : cass_metrics.requests.percentile_95th;
2377 27 : metrics->requests.percentile_98th =
2378 27 : cass_metrics.requests.percentile_98th;
2379 27 : metrics->requests.percentile_99th =
2380 27 : cass_metrics.requests.percentile_99th;
2381 27 : metrics->requests.percentile_999th =
2382 27 : cass_metrics.requests.percentile_999th;
2383 27 : metrics->requests.mean_rate = cass_metrics.requests.mean_rate;
2384 27 : metrics->requests.one_minute_rate =
2385 27 : cass_metrics.requests.one_minute_rate;
2386 27 : metrics->requests.five_minute_rate =
2387 27 : cass_metrics.requests.five_minute_rate;
2388 27 : metrics->requests.fifteen_minute_rate =
2389 27 : cass_metrics.requests.fifteen_minute_rate;
2390 : // Stats
2391 27 : metrics->stats.total_connections =
2392 27 : cass_metrics.stats.total_connections;
2393 27 : metrics->stats.available_connections =
2394 27 : cass_metrics.stats.available_connections;
2395 27 : metrics->stats.exceeded_pending_requests_water_mark =
2396 27 : cass_metrics.stats.exceeded_pending_requests_water_mark;
2397 27 : metrics->stats.exceeded_write_bytes_water_mark =
2398 27 : cass_metrics.stats.exceeded_write_bytes_water_mark;
2399 : // Errors
2400 27 : metrics->errors.connection_timeouts =
2401 27 : cass_metrics.errors.connection_timeouts;
2402 27 : metrics->errors.pending_request_timeouts =
2403 27 : cass_metrics.errors.pending_request_timeouts;
2404 27 : metrics->errors.request_timeouts =
2405 27 : cass_metrics.errors.request_timeouts;
2406 27 : return true;
2407 : }
2408 :
2409 0 : bool CqlIfImpl::InsertIntoTableInternal(std::auto_ptr<GenDb::ColList> v_columns,
2410 : CassConsistency consistency, bool sync,
2411 : impl::CassAsyncQueryCallback cb) {
2412 0 : if (session_state_ != SessionState::CONNECTED) {
2413 0 : return false;
2414 : }
2415 0 : std::string query;
2416 0 : if (IsTableStatic(v_columns->cfname_) == 1) {
2417 0 : query = impl::StaticCf2CassInsertIntoTable(v_columns.get());
2418 0 : } else if (IsTableStatic(v_columns->cfname_) == 0){
2419 0 : query = impl::DynamicCf2CassInsertIntoTable(v_columns.get());
2420 : } else {
2421 0 : return false;
2422 : }
2423 0 : if (sync) {
2424 0 : return impl::ExecuteQuerySync(cci_, session_.get(), query.c_str(),
2425 0 : consistency);
2426 : } else {
2427 0 : impl::ExecuteQueryAsync(cci_, session_.get(), query.c_str(),
2428 : consistency, cb);
2429 0 : return true;
2430 : }
2431 0 : }
2432 :
2433 0 : bool CqlIfImpl::PrepareInsertIntoTableSync(const GenDb::NewCf &cf,
2434 : impl::CassPreparedPtr *prepared) {
2435 0 : if (schema_session_state_ != SessionState::CONNECTED) {
2436 0 : return false;
2437 : }
2438 0 : std::string query;
2439 0 : switch (cf.cftype_) {
2440 0 : case GenDb::NewCf::COLUMN_FAMILY_SQL:
2441 : {
2442 0 : query = impl::StaticCf2CassPrepareInsertIntoTable(cf);
2443 0 : break;
2444 : }
2445 0 : case GenDb::NewCf::COLUMN_FAMILY_NOSQL:
2446 : {
2447 0 : boost::system::error_code ec;
2448 0 : query = impl::DynamicCf2CassPrepareInsertIntoTable(cf, &ec);
2449 0 : if (ec.value() != boost::system::errc::success) {
2450 0 : return false;
2451 : }
2452 0 : break;
2453 : }
2454 0 : default:
2455 : {
2456 0 : return false;
2457 : }
2458 : }
2459 0 : return impl::PrepareSync(cci_, schema_session_.get(), query.c_str(),
2460 0 : prepared);
2461 0 : }
2462 :
2463 0 : bool CqlIfImpl::InsertIntoTablePrepareInternal(
2464 : std::auto_ptr<GenDb::ColList> v_columns,
2465 : CassConsistency consistency, bool sync,
2466 : impl::CassAsyncQueryCallback cb) {
2467 0 : if (session_state_ != SessionState::CONNECTED) {
2468 0 : return false;
2469 : }
2470 0 : impl::CassPreparedPtr prepared(NULL, cci_);
2471 0 : bool success(GetPrepareInsertIntoTable(v_columns->cfname_, &prepared));
2472 0 : if (!success) {
2473 0 : CQLIF_ERR_TRACE("CassPrepared statement NOT found: " <<
2474 : v_columns->cfname_);
2475 0 : return false;
2476 : }
2477 0 : impl::CassStatementPtr qstatement(cci_->CassPreparedBind(prepared.get()),
2478 0 : cci_);
2479 0 : if (IsTableStatic(v_columns->cfname_) == 1) {
2480 0 : success = impl::StaticCf2CassPrepareBind(cci_, qstatement.get(),
2481 0 : v_columns.get());
2482 0 : } else if (IsTableStatic(v_columns->cfname_) == 0){
2483 0 : success = impl::DynamicCf2CassPrepareBind(cci_, qstatement.get(),
2484 0 : v_columns.get());
2485 : } else {
2486 0 : return false;
2487 : }
2488 0 : if (!success) {
2489 0 : return false;
2490 : }
2491 0 : if (sync) {
2492 0 : return impl::ExecuteQueryStatementSync(cci_, session_.get(),
2493 0 : qstatement.get(), consistency);
2494 : } else {
2495 0 : std::string qid("Prepare: " + v_columns->cfname_);
2496 0 : impl::ExecuteQueryStatementAsync(cci_, session_.get(), qid.c_str(),
2497 : qstatement.get(), consistency, cb);
2498 0 : return true;
2499 0 : }
2500 0 : }
2501 :
2502 : const char * CqlIfImpl::kQCreateKeyspaceIfNotExists(
2503 : "CREATE KEYSPACE IF NOT EXISTS \"%s\" WITH "
2504 : "replication = { 'class' : 'SimpleStrategy', 'replication_factor' : %s }");
2505 : const char * CqlIfImpl::kQUseKeyspace("USE \"%s\"");
2506 : const char * CqlIfImpl::kTaskName("CqlIfImpl::Task");
2507 :
2508 : //
2509 : // CqlIf
2510 : //
2511 3703 : CqlIf::CqlIf(EventManager *evm,
2512 : const std::vector<std::string> &cassandra_ips,
2513 : int cassandra_port,
2514 : const std::string &cassandra_user,
2515 : const std::string &cassandra_password,
2516 : bool use_ssl,
2517 : const std::string &ca_certs_path,
2518 3703 : bool create_schema) :
2519 3703 : cci_(new interface::CassDatastaxLibrary),
2520 3703 : impl_(new CqlIfImpl(evm, cassandra_ips, cassandra_port,
2521 : cassandra_user, cassandra_password, use_ssl,
2522 3703 : ca_certs_path, cci_.get())),
2523 3703 : use_prepared_for_insert_(true),
2524 7406 : create_schema_(create_schema) {
2525 : // Setup library logging
2526 7406 : cci_->CassLogSetLevel(impl::Log4Level2CassLogLevel(
2527 7406 : log4cplus::Logger::getRoot().getLogLevel()));
2528 3703 : cci_->CassLogSetCallback(impl::CassLibraryLog, NULL);
2529 3703 : initialized_ = false;
2530 3755 : BOOST_FOREACH(const std::string &cassandra_ip, cassandra_ips) {
2531 26 : boost::system::error_code ec;
2532 : boost::asio::ip::address cassandra_addr(
2533 26 : AddressFromString(cassandra_ip, &ec));
2534 26 : GenDb::Endpoint endpoint(cassandra_addr, cassandra_port);
2535 26 : endpoints_.push_back(endpoint);
2536 : }
2537 3703 : }
2538 :
2539 26 : CqlIf::CqlIf() : impl_(NULL) {
2540 26 : }
2541 :
2542 3918 : CqlIf::~CqlIf() {
2543 3918 : }
2544 :
2545 : // Init/Uninit
2546 25 : bool CqlIf::Db_Init() {
2547 25 : if (create_schema_) {
2548 0 : impl_->SetRequestTimeout(GenDb::g_gendb_constants.SCHEMA_REQUEST_TIMEOUT);
2549 0 : bool success(impl_->ConnectSchemaSync());
2550 0 : if (!success) {
2551 0 : return success;
2552 : }
2553 : }
2554 25 : return impl_->ConnectSync();
2555 : }
2556 :
2557 25 : void CqlIf::Db_Uninit() {
2558 25 : if (create_schema_) {
2559 0 : impl_->DisconnectSchemaSync();
2560 : }
2561 25 : impl_->DisconnectSync();
2562 25 : }
2563 :
2564 50 : void CqlIf::Db_SetInitDone(bool init_done) {
2565 50 : initialized_ = init_done;
2566 : // No need for schema session if initialization is done
2567 50 : if (create_schema_) {
2568 0 : if (initialized_) {
2569 0 : impl_->SetRequestTimeout(GenDb::g_gendb_constants.DEFAULT_REQUEST_TIMEOUT);
2570 0 : impl_->DisconnectSchemaSync();
2571 : }
2572 : }
2573 50 : }
2574 :
2575 : // Tablespace
2576 0 : bool CqlIf::Db_AddSetTablespace(const std::string &tablespace,
2577 : const std::string &replication_factor) {
2578 0 : bool success(impl_->CreateKeyspaceIfNotExistsSync(tablespace,
2579 : replication_factor, CASS_CONSISTENCY_QUORUM));
2580 0 : if (!success) {
2581 0 : IncrementErrors(GenDb::IfErrors::ERR_WRITE_TABLESPACE);
2582 0 : return success;
2583 : }
2584 0 : success = impl_->UseKeyspaceSyncOnSchemaSession(tablespace,
2585 : CASS_CONSISTENCY_ONE);
2586 0 : if (!success) {
2587 0 : IncrementErrors(GenDb::IfErrors::ERR_READ_TABLESPACE);
2588 0 : return success;
2589 : }
2590 0 : return success;
2591 : }
2592 :
2593 25 : bool CqlIf::Db_SetTablespace(const std::string &tablespace) {
2594 25 : bool success(impl_->UseKeyspaceSync(tablespace, CASS_CONSISTENCY_ONE));
2595 25 : if (!success) {
2596 0 : IncrementErrors(GenDb::IfErrors::ERR_READ_TABLESPACE);
2597 0 : return success;
2598 : }
2599 25 : return success;
2600 : }
2601 :
2602 : // Column family
2603 0 : bool CqlIf::Db_AddColumnfamily(const GenDb::NewCf &cf,
2604 : const std::string &compaction_strategy) {
2605 : bool success(
2606 0 : impl_->CreateTableIfNotExistsSync(cf, compaction_strategy,
2607 : CASS_CONSISTENCY_QUORUM));
2608 0 : if (!success) {
2609 0 : IncrementTableWriteFailStats(cf.cfname_);
2610 0 : IncrementErrors(GenDb::IfErrors::ERR_WRITE_COLUMN_FAMILY);
2611 0 : return success;
2612 : }
2613 : // Locate (add if not exists) INSERT INTO prepare statement
2614 0 : success = impl_->LocatePrepareInsertIntoTable(cf);
2615 0 : if (!success) {
2616 0 : IncrementTableWriteFailStats(cf.cfname_);
2617 0 : IncrementErrors(GenDb::IfErrors::ERR_WRITE_COLUMN_FAMILY);
2618 0 : return success;
2619 : }
2620 0 : IncrementTableWriteStats(cf.cfname_);
2621 0 : return success;
2622 : }
2623 :
2624 125 : bool CqlIf::Db_UseColumnfamily(const GenDb::NewCf &cf) {
2625 : // Check existence of table
2626 125 : return Db_UseColumnfamily(cf.cfname_);
2627 : }
2628 :
2629 150 : bool CqlIf::Db_UseColumnfamily(const std::string &cfname) {
2630 : // Check existence of table
2631 150 : bool success(impl_->IsTablePresent(cfname));
2632 150 : if (!success) {
2633 25 : IncrementTableReadFailStats(cfname);
2634 25 : IncrementErrors(GenDb::IfErrors::ERR_READ_COLUMN_FAMILY);
2635 25 : return success;
2636 : }
2637 125 : IncrementTableReadStats(cfname);
2638 125 : return success;
2639 : }
2640 :
2641 : // Index
2642 0 : bool CqlIf::Db_CreateIndex(const std::string &cfname,
2643 : const std::string &column, const std::string &indexname,
2644 : const GenDb::ColIndexMode::type index_mode) {
2645 0 : bool success(impl_->CreateIndexIfNotExistsSync(cfname, column, indexname,
2646 : CASS_CONSISTENCY_QUORUM, index_mode));
2647 0 : if (!success) {
2648 0 : IncrementTableWriteFailStats(cfname);
2649 0 : IncrementErrors(GenDb::IfErrors::ERR_WRITE_COLUMN_FAMILY);
2650 0 : return success;
2651 : }
2652 0 : return success;
2653 : }
2654 :
2655 : // Column
2656 0 : void CqlIf::OnAsyncColumnAddCompletion(GenDb::DbOpResult::type drc,
2657 : std::auto_ptr<GenDb::ColList> row,
2658 : std::string cfname, GenDb::GenDbIf::DbAddColumnCb cb) {
2659 0 : if (drc == GenDb::DbOpResult::OK) {
2660 0 : IncrementTableWriteStats(cfname);
2661 0 : } else if (drc == GenDb::DbOpResult::BACK_PRESSURE) {
2662 0 : IncrementTableWriteBackPressureFailStats(cfname);
2663 0 : IncrementErrors(GenDb::IfErrors::ERR_WRITE_COLUMN);
2664 : } else {
2665 0 : IncrementTableWriteFailStats(cfname);
2666 0 : IncrementErrors(GenDb::IfErrors::ERR_WRITE_COLUMN);
2667 : }
2668 0 : if (!cb.empty()) {
2669 0 : cb(drc);
2670 : }
2671 0 : }
2672 :
2673 : struct AsyncRowGetCallbackContext {
2674 0 : AsyncRowGetCallbackContext(GenDb::GenDbIf::DbGetRowCb cb,
2675 0 : GenDb::DbOpResult::type drc, std::auto_ptr<GenDb::ColList> row) :
2676 0 : cb_(cb),
2677 0 : drc_(drc),
2678 0 : row_(row) {
2679 0 : }
2680 : GenDb::GenDbIf::DbGetRowCb cb_;
2681 : GenDb::DbOpResult::type drc_;
2682 : std::auto_ptr<GenDb::ColList> row_;
2683 : };
2684 :
2685 0 : static void AsyncRowGetCompletionCallback(
2686 : boost::shared_ptr<AsyncRowGetCallbackContext> cb_ctx) {
2687 0 : cb_ctx->cb_(cb_ctx->drc_, cb_ctx->row_);
2688 0 : }
2689 :
2690 108123 : void CqlIf::OnAsyncRowGetCompletion(GenDb::DbOpResult::type drc,
2691 : std::auto_ptr<GenDb::ColList> row, std::string cfname,
2692 : GenDb::GenDbIf::DbGetRowCb cb, bool use_worker, int task_id,
2693 : int task_instance) {
2694 108123 : if (drc == GenDb::DbOpResult::OK) {
2695 108139 : IncrementTableReadStats(cfname);
2696 0 : } else if (drc == GenDb::DbOpResult::BACK_PRESSURE) {
2697 0 : IncrementTableReadBackPressureFailStats(cfname);
2698 0 : IncrementErrors(GenDb::IfErrors::ERR_READ_COLUMN);
2699 : } else {
2700 0 : IncrementTableReadFailStats(cfname);
2701 0 : IncrementErrors(GenDb::IfErrors::ERR_READ_COLUMN);
2702 : }
2703 108192 : if (use_worker) {
2704 0 : if (!cb.empty()) {
2705 : boost::shared_ptr<AsyncRowGetCallbackContext> ctx(
2706 0 : new AsyncRowGetCallbackContext(cb, drc, row));
2707 : impl::WorkerTask *worker(new impl::WorkerTask(
2708 0 : boost::bind(&AsyncRowGetCompletionCallback, ctx),
2709 0 : task_id, task_instance));
2710 0 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
2711 0 : scheduler->Enqueue(worker);
2712 0 : }
2713 : } else {
2714 108192 : if (!cb.empty()) {
2715 108192 : cb(drc, row);
2716 : }
2717 : }
2718 108182 : }
2719 :
2720 108118 : void CqlIf::OnAsyncRowGetCompletion(GenDb::DbOpResult::type drc,
2721 : std::auto_ptr<GenDb::ColList> row, std::string cfname,
2722 : GenDb::GenDbIf::DbGetRowCb cb) {
2723 108118 : OnAsyncRowGetCompletion(drc, row, cfname, cb, false, -1, -2);
2724 108185 : }
2725 0 : bool CqlIf::Db_AddColumn(std::auto_ptr<GenDb::ColList> cl,
2726 : GenDb::DbConsistency::type dconsistency,
2727 : GenDb::GenDbIf::DbAddColumnCb cb) {
2728 0 : std::string cfname(cl->cfname_);
2729 0 : if (!initialized_) {
2730 0 : IncrementTableWriteFailStats(cfname);
2731 0 : IncrementErrors(GenDb::IfErrors::ERR_WRITE_COLUMN);
2732 0 : return false;
2733 : }
2734 0 : CassConsistency consistency(impl::Db2CassConsistency(dconsistency));
2735 : bool success;
2736 0 : if (use_prepared_for_insert_ &&
2737 0 : impl_->IsInsertIntoTablePrepareSupported(cfname)) {
2738 0 : success = impl_->InsertIntoTablePrepareAsync(cl, consistency,
2739 0 : boost::bind(&CqlIf::OnAsyncColumnAddCompletion, this, _1, _2, cfname,
2740 : cb));
2741 : } else {
2742 0 : success = impl_->InsertIntoTableAsync(cl, consistency,
2743 0 : boost::bind(&CqlIf::OnAsyncColumnAddCompletion, this, _1, _2, cfname,
2744 : cb));
2745 : }
2746 0 : if (!success) {
2747 0 : IncrementTableWriteFailStats(cfname);
2748 0 : IncrementErrors(GenDb::IfErrors::ERR_WRITE_COLUMN);
2749 0 : return success;
2750 : }
2751 0 : return success;
2752 0 : }
2753 :
2754 0 : bool CqlIf::Db_AddColumnSync(std::auto_ptr<GenDb::ColList> cl,
2755 : GenDb::DbConsistency::type dconsistency) {
2756 0 : std::string cfname(cl->cfname_);
2757 0 : CassConsistency consistency(impl::Db2CassConsistency(dconsistency));
2758 0 : bool success(impl_->InsertIntoTableSync(cl, consistency));
2759 0 : if (!success) {
2760 0 : IncrementTableWriteFailStats(cfname);
2761 0 : IncrementErrors(GenDb::IfErrors::ERR_WRITE_COLUMN);
2762 0 : return success;
2763 : }
2764 0 : IncrementTableWriteStats(cfname);
2765 0 : return success;
2766 0 : }
2767 :
2768 : // Read
2769 0 : bool CqlIf::Db_GetRowAsync(const std::string &cfname,
2770 : const GenDb::DbDataValueVec &rowkey, const GenDb::ColumnNameRange &crange,
2771 : GenDb::DbConsistency::type dconsistency, GenDb::GenDbIf::DbGetRowCb cb) {
2772 0 : CassConsistency consistency(impl::Db2CassConsistency(dconsistency));
2773 0 : bool success(impl_->SelectFromTableClusteringKeyRangeAsync(cfname, rowkey,
2774 0 : crange, consistency, boost::bind(&CqlIf::OnAsyncRowGetCompletion, this,
2775 : _1, _2, cfname, cb)));
2776 0 : if (!success) {
2777 0 : IncrementTableReadFailStats(cfname);
2778 0 : IncrementErrors(GenDb::IfErrors::ERR_READ_COLUMN_FAMILY);
2779 : }
2780 0 : return success;
2781 : }
2782 :
2783 0 : bool CqlIf::Db_GetRowAsync(const std::string &cfname,
2784 : const GenDb::DbDataValueVec &rowkey, const GenDb::ColumnNameRange &crange,
2785 : GenDb::DbConsistency::type dconsistency, int task_id, int task_instance,
2786 : GenDb::GenDbIf::DbGetRowCb cb) {
2787 0 : CassConsistency consistency(impl::Db2CassConsistency(dconsistency));
2788 0 : bool success(impl_->SelectFromTableClusteringKeyRangeAsync(cfname, rowkey,
2789 0 : crange, consistency, boost::bind(&CqlIf::OnAsyncRowGetCompletion, this,
2790 : _1, _2, cfname, cb, true, task_id, task_instance)));
2791 0 : if (!success) {
2792 0 : IncrementTableReadFailStats(cfname);
2793 0 : IncrementErrors(GenDb::IfErrors::ERR_READ_COLUMN_FAMILY);
2794 : }
2795 0 : return success;
2796 : }
2797 :
2798 0 : bool CqlIf::Db_GetRowAsync(const std::string &cfname,
2799 : const GenDb::DbDataValueVec &rowkey,
2800 : GenDb::DbConsistency::type dconsistency,
2801 : GenDb::GenDbIf::DbGetRowCb cb) {
2802 0 : CassConsistency consistency(impl::Db2CassConsistency(dconsistency));
2803 0 : bool success(impl_->SelectFromTableAsync(cfname, rowkey,
2804 0 : consistency, boost::bind(&CqlIf::OnAsyncRowGetCompletion, this, _1, _2,
2805 : cfname, cb)));
2806 0 : if (!success) {
2807 0 : IncrementTableReadFailStats(cfname);
2808 0 : IncrementErrors(GenDb::IfErrors::ERR_READ_COLUMN_FAMILY);
2809 : }
2810 0 : return success;
2811 : }
2812 :
2813 0 : bool CqlIf::Db_GetRowAsync(const std::string &cfname,
2814 : const GenDb::DbDataValueVec &rowkey,
2815 : GenDb::DbConsistency::type dconsistency, int task_id, int task_instance,
2816 : GenDb::GenDbIf::DbGetRowCb cb) {
2817 0 : CassConsistency consistency(impl::Db2CassConsistency(dconsistency));
2818 0 : bool success(impl_->SelectFromTableAsync(cfname, rowkey,
2819 0 : consistency, boost::bind(&CqlIf::OnAsyncRowGetCompletion, this, _1, _2,
2820 : cfname, cb, true, task_id, task_instance)));
2821 0 : if (!success) {
2822 0 : IncrementTableReadFailStats(cfname);
2823 0 : IncrementErrors(GenDb::IfErrors::ERR_READ_COLUMN_FAMILY);
2824 : }
2825 0 : return success;
2826 : }
2827 :
2828 107505 : bool CqlIf::Db_GetRowAsync(const std::string &cfname,
2829 : const GenDb::DbDataValueVec &rowkey, const GenDb::ColumnNameRange &crange,
2830 : const GenDb::WhereIndexInfoVec &where_vec,
2831 : GenDb::DbConsistency::type dconsistency, GenDb::GenDbIf::DbGetRowCb cb) {
2832 107505 : CassConsistency consistency(impl::Db2CassConsistency(dconsistency));
2833 324207 : bool success(impl_->SelectFromTableClusteringKeyRangeAndIndexValueAsync(cfname,
2834 215878 : rowkey, crange, where_vec, GenDb::FieldNamesToReadVec(), consistency,
2835 216073 : boost::bind(&CqlIf::OnAsyncRowGetCompletion, this, _1, _2, cfname, cb)));
2836 108066 : if (!success) {
2837 0 : IncrementTableReadFailStats(cfname);
2838 0 : IncrementErrors(GenDb::IfErrors::ERR_READ_COLUMN_FAMILY);
2839 : }
2840 108070 : return success;
2841 : }
2842 :
2843 25 : bool CqlIf::Db_GetRow(GenDb::ColList *out, const std::string &cfname,
2844 : const GenDb::DbDataValueVec &rowkey,
2845 : GenDb::DbConsistency::type dconsistency) {
2846 25 : CassConsistency consistency(impl::Db2CassConsistency(dconsistency));
2847 25 : bool success(impl_->SelectFromTableSync(cfname, rowkey,
2848 : consistency, &out->columns_));
2849 25 : if (!success) {
2850 0 : IncrementTableReadFailStats(cfname);
2851 0 : IncrementErrors(GenDb::IfErrors::ERR_READ_COLUMN);
2852 0 : return success;
2853 : }
2854 25 : IncrementTableReadStats(cfname);
2855 25 : return success;
2856 : }
2857 :
2858 0 : bool CqlIf::Db_GetRow(GenDb::ColList *out, const std::string &cfname,
2859 : const GenDb::DbDataValueVec &rowkey,
2860 : GenDb::DbConsistency::type dconsistency,
2861 : const GenDb::ColumnNameRange &crange,
2862 : const GenDb::FieldNamesToReadVec &read_vec) {
2863 0 : CassConsistency consistency(impl::Db2CassConsistency(dconsistency));
2864 0 : bool success(impl_->SelectFromTableClusteringKeyRangeFieldNamesSync(cfname,
2865 : rowkey, crange, consistency, read_vec, &out->columns_));
2866 0 : if (!success) {
2867 0 : IncrementTableReadFailStats(cfname);
2868 0 : IncrementErrors(GenDb::IfErrors::ERR_READ_COLUMN);
2869 0 : return success;
2870 : }
2871 0 : IncrementTableReadStats(cfname);
2872 0 : return success;
2873 : }
2874 :
2875 7 : bool CqlIf::Db_GetMultiRow(GenDb::ColListVec *out, const std::string &cfname,
2876 : const std::vector<GenDb::DbDataValueVec> &v_rowkey) {
2877 1007 : BOOST_FOREACH(const GenDb::DbDataValueVec &rkey, v_rowkey) {
2878 500 : std::auto_ptr<GenDb::ColList> v_columns(new GenDb::ColList);
2879 : // Partition Key
2880 500 : v_columns->rowkey_ = rkey;
2881 1000 : bool success(impl_->SelectFromTableSync(cfname, rkey,
2882 500 : CASS_CONSISTENCY_ONE, &v_columns->columns_));
2883 500 : if (!success) {
2884 0 : CQLIF_ERR_TRACE("SELECT FROM Table: " << cfname << " Partition Key: "
2885 : << GenDb::DbDataValueVecToString(rkey) << " FAILED");
2886 0 : IncrementTableReadFailStats(cfname);
2887 0 : IncrementErrors(GenDb::IfErrors::ERR_READ_COLUMN);
2888 0 : return false;
2889 : }
2890 500 : out->push_back(v_columns.release());
2891 500 : }
2892 7 : IncrementTableReadStats(cfname, v_rowkey.size());
2893 7 : return true;
2894 : }
2895 :
2896 0 : bool CqlIf::Db_GetMultiRow(GenDb::ColListVec *out, const std::string &cfname,
2897 : const std::vector<GenDb::DbDataValueVec> &v_rowkey,
2898 : const GenDb::ColumnNameRange &crange) {
2899 0 : BOOST_FOREACH(const GenDb::DbDataValueVec &rkey, v_rowkey) {
2900 0 : std::auto_ptr<GenDb::ColList> v_columns(new GenDb::ColList);
2901 : // Partition Key
2902 0 : v_columns->rowkey_ = rkey;
2903 0 : bool success(impl_->SelectFromTableClusteringKeyRangeSync(cfname,
2904 0 : rkey, crange, CASS_CONSISTENCY_ONE, &v_columns->columns_));
2905 0 : if (!success) {
2906 0 : CQLIF_ERR_TRACE("SELECT FROM Table: " << cfname << " Partition Key: "
2907 : << GenDb::DbDataValueVecToString(rkey) <<
2908 : " Clustering Key Range: " << crange.ToString() << " FAILED");
2909 0 : IncrementTableReadFailStats(cfname);
2910 0 : IncrementErrors(GenDb::IfErrors::ERR_READ_COLUMN);
2911 0 : return false;
2912 : }
2913 0 : out->push_back(v_columns.release());
2914 0 : }
2915 0 : IncrementTableReadStats(cfname, v_rowkey.size());
2916 0 : return true;
2917 : }
2918 :
2919 0 : bool CqlIf::Db_GetMultiRow(GenDb::ColListVec *out, const std::string &cfname,
2920 : const std::vector<GenDb::DbDataValueVec> &v_rowkey,
2921 : const GenDb::ColumnNameRange &crange,
2922 : const GenDb::FieldNamesToReadVec &read_vec,
2923 : GenDb::DbConsistency::type dconsistency) {
2924 0 : CassConsistency consistency(impl::Db2CassConsistency(dconsistency));
2925 0 : bool success(impl_->SelectFromTableClusteringKeyRangeFieldNamesSync(cfname,
2926 : v_rowkey, crange, consistency, read_vec, out));
2927 0 : if (!success) {
2928 0 : IncrementTableReadFailStats(cfname);
2929 0 : IncrementErrors(GenDb::IfErrors::ERR_READ_COLUMN);
2930 0 : return false;
2931 : }
2932 0 : IncrementTableReadStats(cfname, v_rowkey.size());
2933 0 : return true;
2934 : }
2935 :
2936 0 : bool CqlIf::Db_GetAllRows(GenDb::ColListVec *out, const std::string &cfname,
2937 : GenDb::DbConsistency::type dconsistency) {
2938 0 : CassConsistency consistency(impl::Db2CassConsistency(dconsistency));
2939 0 : bool success(impl_->SelectFromTableSync(cfname, consistency, out));
2940 0 : if (!success) {
2941 0 : IncrementTableReadFailStats(cfname);
2942 0 : IncrementErrors(GenDb::IfErrors::ERR_READ_COLUMN);
2943 0 : return success;
2944 : }
2945 0 : IncrementTableReadStats(cfname);
2946 0 : return success;
2947 : }
2948 :
2949 : // Queue
2950 0 : bool CqlIf::Db_GetQueueStats(uint64_t *queue_count,
2951 : uint64_t *enqueues) const {
2952 : //return impl_->Db_GetQueueStats(queue_count, enqueues);
2953 0 : return true;
2954 : }
2955 :
2956 0 : void CqlIf::Db_SetQueueWaterMark(bool high, size_t queue_count,
2957 : GenDb::GenDbIf::DbQueueWaterMarkCb cb) {
2958 : //impl_->Db_SetQueueWaterMark(high, queue_count, cb);
2959 0 : }
2960 :
2961 0 : void CqlIf::Db_ResetQueueWaterMarks() {
2962 : //impl_->Db_ResetQueueWaterMarks();
2963 0 : }
2964 :
2965 : // Stats
2966 23 : bool CqlIf::Db_GetStats(std::vector<GenDb::DbTableInfo> *vdbti,
2967 : GenDb::DbErrors *dbe) {
2968 23 : std::scoped_lock lock(stats_mutex_);
2969 23 : stats_.GetDiffs(vdbti, dbe);
2970 23 : return true;
2971 23 : }
2972 :
2973 4 : bool CqlIf::Db_GetCumulativeStats(std::vector<GenDb::DbTableInfo> *vdbti,
2974 : GenDb::DbErrors *dbe) const {
2975 4 : std::scoped_lock lock(stats_mutex_);
2976 4 : stats_.GetCumulative(vdbti, dbe);
2977 4 : return true;
2978 4 : }
2979 :
2980 4 : bool CqlIf::Db_GetCqlMetrics(Metrics *metrics) const {
2981 4 : return impl_->GetMetrics(metrics);
2982 : }
2983 :
2984 23 : bool CqlIf::Db_GetCqlStats(DbStats *db_stats) const {
2985 23 : Metrics metrics;
2986 23 : bool success(impl_->GetMetrics(&metrics));
2987 23 : if (!success) {
2988 0 : return success;
2989 : }
2990 23 : db_stats->requests_one_minute_rate = metrics.requests.one_minute_rate;
2991 23 : db_stats->stats = metrics.stats;
2992 23 : db_stats->errors = metrics.errors;
2993 23 : return success;
2994 23 : }
2995 :
2996 0 : void CqlIf::IncrementTableWriteStats(const std::string &table_name) {
2997 0 : std::scoped_lock lock(stats_mutex_);
2998 0 : stats_.IncrementTableWrite(table_name);
2999 0 : }
3000 :
3001 0 : void CqlIf::IncrementTableWriteStats(const std::string &table_name,
3002 : uint64_t num_writes) {
3003 0 : std::scoped_lock lock(stats_mutex_);
3004 0 : stats_.IncrementTableWrite(table_name, num_writes);
3005 0 : }
3006 :
3007 0 : void CqlIf::IncrementTableWriteFailStats(const std::string &table_name) {
3008 0 : std::scoped_lock lock(stats_mutex_);
3009 0 : stats_.IncrementTableWriteFail(table_name);
3010 0 : }
3011 :
3012 0 : void CqlIf::IncrementTableWriteFailStats(const std::string &table_name,
3013 : uint64_t num_writes) {
3014 0 : std::scoped_lock lock(stats_mutex_);
3015 0 : stats_.IncrementTableWriteFail(table_name, num_writes);
3016 0 : }
3017 :
3018 0 : void CqlIf::IncrementTableWriteBackPressureFailStats(
3019 : const std::string &table_name) {
3020 0 : std::scoped_lock lock(stats_mutex_);
3021 0 : stats_.IncrementTableWriteBackPressureFail(table_name);
3022 0 : }
3023 :
3024 0 : void CqlIf::IncrementTableReadBackPressureFailStats(
3025 : const std::string &table_name) {
3026 0 : std::scoped_lock lock(stats_mutex_);
3027 0 : stats_.IncrementTableReadBackPressureFail(table_name);
3028 0 : }
3029 :
3030 108280 : void CqlIf::IncrementTableReadStats(const std::string &table_name) {
3031 108280 : std::scoped_lock lock(stats_mutex_);
3032 108342 : stats_.IncrementTableRead(table_name);
3033 108342 : }
3034 :
3035 7 : void CqlIf::IncrementTableReadStats(const std::string &table_name,
3036 : uint64_t num_reads) {
3037 7 : std::scoped_lock lock(stats_mutex_);
3038 7 : stats_.IncrementTableRead(table_name, num_reads);
3039 7 : }
3040 :
3041 25 : void CqlIf::IncrementTableReadFailStats(const std::string &table_name) {
3042 25 : std::scoped_lock lock(stats_mutex_);
3043 25 : stats_.IncrementTableReadFail(table_name);
3044 25 : }
3045 :
3046 0 : void CqlIf::IncrementTableReadFailStats(const std::string &table_name,
3047 : uint64_t num_reads) {
3048 0 : std::scoped_lock lock(stats_mutex_);
3049 0 : stats_.IncrementTableReadFail(table_name, num_reads);
3050 0 : }
3051 :
3052 25 : void CqlIf::IncrementErrors(GenDb::IfErrors::Type err_type) {
3053 25 : std::scoped_lock lock(stats_mutex_);
3054 25 : stats_.IncrementErrors(err_type);
3055 25 : }
3056 :
3057 : // Connection
3058 25 : std::vector<GenDb::Endpoint> CqlIf::Db_GetEndpoints() const {
3059 25 : return endpoints_;
3060 : }
3061 :
3062 : namespace interface {
3063 :
3064 : //
3065 : // CassDatastaxLibrary
3066 : //
3067 3703 : CassDatastaxLibrary::CassDatastaxLibrary() {
3068 3703 : }
3069 :
3070 7406 : CassDatastaxLibrary::~CassDatastaxLibrary() {
3071 7406 : }
3072 :
3073 : // CassCluster
3074 3703 : CassCluster* CassDatastaxLibrary::CassClusterNew() {
3075 3703 : return cass_cluster_new();
3076 : }
3077 :
3078 3703 : void CassDatastaxLibrary::CassClusterFree(CassCluster* cluster) {
3079 3703 : cass_cluster_free(cluster);
3080 3703 : }
3081 :
3082 3703 : CassError CassDatastaxLibrary::CassClusterSetContactPoints(
3083 : CassCluster* cluster, const char* contact_points) {
3084 3703 : return cass_cluster_set_contact_points(cluster, contact_points);
3085 : }
3086 :
3087 3703 : CassError CassDatastaxLibrary::CassClusterSetPort(CassCluster* cluster,
3088 : int port) {
3089 3703 : return cass_cluster_set_port(cluster, port);
3090 : }
3091 :
3092 0 : void CassDatastaxLibrary::CassClusterSetSsl(CassCluster* cluster, CassSsl* ssl) {
3093 0 : cass_cluster_set_ssl(cluster, ssl);
3094 0 : }
3095 :
3096 0 : void CassDatastaxLibrary::CassClusterSetCredentials(CassCluster* cluster,
3097 : const char* username, const char* password) {
3098 0 : cass_cluster_set_credentials(cluster, username, password);
3099 0 : }
3100 :
3101 3703 : CassError CassDatastaxLibrary::CassClusterSetNumThreadsIo(CassCluster* cluster,
3102 : unsigned num_threads) {
3103 3703 : return cass_cluster_set_num_threads_io(cluster, num_threads);
3104 : }
3105 :
3106 3703 : CassError CassDatastaxLibrary::CassClusterSetPendingRequestsHighWaterMark(
3107 : CassCluster* cluster, unsigned num_requests) {
3108 3703 : return cass_cluster_set_pending_requests_high_water_mark(cluster,
3109 3703 : num_requests);
3110 : }
3111 :
3112 3703 : CassError CassDatastaxLibrary::CassClusterSetPendingRequestsLowWaterMark(
3113 : CassCluster* cluster, unsigned num_requests) {
3114 3703 : return cass_cluster_set_pending_requests_low_water_mark(cluster,
3115 3703 : num_requests);
3116 : }
3117 :
3118 3703 : CassError CassDatastaxLibrary::CassClusterSetWriteBytesHighWaterMark(
3119 : CassCluster* cluster, unsigned num_bytes) {
3120 3703 : return cass_cluster_set_write_bytes_high_water_mark(cluster, num_bytes);
3121 : }
3122 :
3123 3703 : CassError CassDatastaxLibrary::CassClusterSetWriteBytesLowWaterMark(
3124 : CassCluster* cluster, unsigned num_bytes) {
3125 3703 : return cass_cluster_set_write_bytes_low_water_mark(cluster, num_bytes);
3126 : }
3127 :
3128 0 : void CassDatastaxLibrary::CassClusterSetWhitelistFiltering(
3129 : CassCluster* cluster, const char* hosts) {
3130 0 : cass_cluster_set_whitelist_filtering(cluster, hosts);
3131 0 : }
3132 :
3133 : // CassSsl
3134 0 : CassSsl* CassDatastaxLibrary::CassSslNew() {
3135 0 : return cass_ssl_new();
3136 : }
3137 :
3138 0 : void CassDatastaxLibrary::CassSslFree(CassSsl* ssl) {
3139 0 : return cass_ssl_free(ssl);
3140 : }
3141 :
3142 0 : CassError CassDatastaxLibrary::CassSslAddTrustedCert(CassSsl* ssl,
3143 : const std::string &cert) {
3144 0 : return cass_ssl_add_trusted_cert_n(ssl, cert.c_str(), cert.length());
3145 : }
3146 :
3147 0 : void CassDatastaxLibrary::CassSslSetVerifyFlags(CassSsl* ssl, int flags) {
3148 0 : cass_ssl_set_verify_flags(ssl, flags);
3149 0 : }
3150 :
3151 : // CassSession
3152 7431 : CassSession* CassDatastaxLibrary::CassSessionNew() {
3153 7431 : return cass_session_new();
3154 : }
3155 :
3156 7431 : void CassDatastaxLibrary::CassSessionFree(CassSession* session) {
3157 7431 : cass_session_free(session);
3158 7431 : }
3159 :
3160 0 : void CassDatastaxLibrary::CassClusterSetRequestTimeout(CassCluster* cluster,
3161 : unsigned timeout_ms) {
3162 0 : return cass_cluster_set_request_timeout(cluster, timeout_ms);
3163 : }
3164 :
3165 25 : CassFuture* CassDatastaxLibrary::CassSessionConnect(CassSession* session,
3166 : const CassCluster* cluster) {
3167 25 : return cass_session_connect(session, cluster);
3168 : }
3169 :
3170 25 : CassFuture* CassDatastaxLibrary::CassSessionClose(CassSession* session) {
3171 25 : return cass_session_close(session);
3172 : }
3173 :
3174 108698 : CassFuture* CassDatastaxLibrary::CassSessionExecute(CassSession* session,
3175 : const CassStatement* statement) {
3176 108698 : return cass_session_execute(session, statement);
3177 : }
3178 :
3179 325903 : const CassSchemaMeta* CassDatastaxLibrary::CassSessionGetSchemaMeta(
3180 : const CassSession* session) {
3181 325903 : return cass_session_get_schema_meta(session);
3182 : }
3183 :
3184 0 : CassFuture* CassDatastaxLibrary::CassSessionPrepare(CassSession* session,
3185 : const char* query) {
3186 0 : return cass_session_prepare(session, query);
3187 : }
3188 :
3189 27 : void CassDatastaxLibrary::CassSessionGetMetrics(const CassSession* session,
3190 : CassMetrics* output) {
3191 27 : cass_session_get_metrics(session, output);
3192 27 : }
3193 :
3194 : // CassSchema
3195 326459 : void CassDatastaxLibrary::CassSchemaMetaFree(
3196 : const CassSchemaMeta* schema_meta) {
3197 326459 : cass_schema_meta_free(schema_meta);
3198 326645 : }
3199 :
3200 326245 : const CassKeyspaceMeta* CassDatastaxLibrary::CassSchemaMetaKeyspaceByName(
3201 : const CassSchemaMeta* schema_meta, const char* keyspace) {
3202 326245 : return cass_schema_meta_keyspace_by_name(schema_meta, keyspace);
3203 : }
3204 :
3205 326129 : const CassTableMeta* CassDatastaxLibrary::CassKeyspaceMetaTableByName(
3206 : const CassKeyspaceMeta* keyspace_meta, const char* table) {
3207 326129 : return cass_keyspace_meta_table_by_name(keyspace_meta, table);
3208 : }
3209 :
3210 108675 : size_t CassDatastaxLibrary::CassTableMetaPartitionKeyCount(
3211 : const CassTableMeta* table_meta) {
3212 108675 : return cass_table_meta_partition_key_count(table_meta);
3213 : }
3214 :
3215 217761 : size_t CassDatastaxLibrary::CassTableMetaClusteringKeyCount(
3216 : const CassTableMeta* table_meta) {
3217 217761 : return cass_table_meta_clustering_key_count(table_meta);
3218 : }
3219 :
3220 : // CassFuture
3221 108766 : void CassDatastaxLibrary::CassFutureFree(CassFuture* future) {
3222 108766 : cass_future_free(future);
3223 108786 : }
3224 :
3225 107962 : CassError CassDatastaxLibrary::CassFutureSetCallback(CassFuture* future,
3226 : CassFutureCallback callback, void* data) {
3227 107962 : return cass_future_set_callback(future, callback, data);
3228 : }
3229 :
3230 600 : void CassDatastaxLibrary::CassFutureWait(CassFuture* future) {
3231 600 : cass_future_wait(future);
3232 600 : }
3233 :
3234 108709 : const CassResult* CassDatastaxLibrary::CassFutureGetResult(
3235 : CassFuture* future) {
3236 108709 : return cass_future_get_result(future);
3237 : }
3238 :
3239 0 : void CassDatastaxLibrary::CassFutureErrorMessage(CassFuture* future,
3240 : const char** message, size_t* message_length) {
3241 0 : cass_future_error_message(future, message, message_length);
3242 0 : }
3243 :
3244 108777 : CassError CassDatastaxLibrary::CassFutureErrorCode(CassFuture* future) {
3245 108777 : return cass_future_error_code(future);
3246 : }
3247 :
3248 0 : const CassPrepared* CassDatastaxLibrary::CassFutureGetPrepared(
3249 : CassFuture* future) {
3250 0 : return cass_future_get_prepared(future);
3251 : }
3252 :
3253 : // CassResult
3254 108715 : void CassDatastaxLibrary::CassResultFree(const CassResult* result) {
3255 108715 : cass_result_free(result);
3256 108716 : }
3257 :
3258 112936 : size_t CassDatastaxLibrary::CassResultColumnCount(const CassResult* result) {
3259 112936 : return cass_result_column_count(result);
3260 : }
3261 :
3262 225 : CassError CassDatastaxLibrary::CassResultColumnName(const CassResult *result,
3263 : size_t index, const char** name, size_t* name_length) {
3264 225 : return cass_result_column_name(result, index, name, name_length);
3265 : }
3266 :
3267 : // CassIterator
3268 108708 : void CassDatastaxLibrary::CassIteratorFree(CassIterator* iterator) {
3269 108708 : cass_iterator_free(iterator);
3270 108660 : }
3271 :
3272 108662 : CassIterator* CassDatastaxLibrary::CassIteratorFromResult(
3273 : const CassResult* result) {
3274 108662 : return cass_iterator_from_result(result);
3275 : }
3276 :
3277 113445 : cass_bool_t CassDatastaxLibrary::CassIteratorNext(CassIterator* iterator) {
3278 113445 : return cass_iterator_next(iterator);
3279 : }
3280 :
3281 4759 : const CassRow* CassDatastaxLibrary::CassIteratorGetRow(
3282 : const CassIterator* iterator) {
3283 4759 : return cass_iterator_get_row(iterator);
3284 : }
3285 :
3286 : // CassStatement
3287 108339 : CassStatement* CassDatastaxLibrary::CassStatementNew(const char* query,
3288 : size_t parameter_count) {
3289 108339 : return cass_statement_new(query, parameter_count);
3290 : }
3291 :
3292 108682 : void CassDatastaxLibrary::CassStatementFree(CassStatement* statement) {
3293 108682 : cass_statement_free(statement);
3294 108735 : }
3295 :
3296 108671 : CassError CassDatastaxLibrary::CassStatementSetConsistency(
3297 : CassStatement* statement, CassConsistency consistency) {
3298 108671 : return cass_statement_set_consistency(statement, consistency);
3299 : }
3300 :
3301 0 : CassError CassDatastaxLibrary::CassStatementBindStringN(
3302 : CassStatement* statement,
3303 : size_t index, const char* value, size_t value_length) {
3304 0 : return cass_statement_bind_string_n(statement, index, value, value_length);
3305 : }
3306 :
3307 0 : CassError CassDatastaxLibrary::CassStatementBindInt32(CassStatement* statement,
3308 : size_t index, cass_int32_t value) {
3309 0 : return cass_statement_bind_int32(statement, index, value);
3310 : }
3311 :
3312 0 : CassError CassDatastaxLibrary::CassStatementBindInt64(CassStatement* statement,
3313 : size_t index, cass_int64_t value) {
3314 0 : return cass_statement_bind_int64(statement, index, value);
3315 : }
3316 :
3317 0 : CassError CassDatastaxLibrary::CassStatementBindUuid(CassStatement* statement,
3318 : size_t index, CassUuid value) {
3319 0 : return cass_statement_bind_uuid(statement, index, value);
3320 : }
3321 :
3322 0 : CassError CassDatastaxLibrary::CassStatementBindDouble(
3323 : CassStatement* statement, size_t index, cass_double_t value) {
3324 0 : return cass_statement_bind_double(statement, index, value);
3325 : }
3326 :
3327 0 : CassError CassDatastaxLibrary::CassStatementBindInet(CassStatement* statement,
3328 : size_t index, CassInet value) {
3329 0 : return cass_statement_bind_inet(statement, index, value);
3330 : }
3331 :
3332 0 : CassError CassDatastaxLibrary::CassStatementBindBytes(
3333 : CassStatement* statement,
3334 : size_t index, const cass_byte_t* value, size_t value_length) {
3335 0 : return cass_statement_bind_bytes(statement, index, value, value_length);
3336 : }
3337 :
3338 0 : CassError CassDatastaxLibrary::CassStatementBindStringByNameN(
3339 : CassStatement* statement,
3340 : const char* name, size_t name_length, const char* value,
3341 : size_t value_length) {
3342 0 : return cass_statement_bind_string_by_name_n(statement, name, name_length,
3343 0 : value, value_length);
3344 : }
3345 :
3346 0 : CassError CassDatastaxLibrary::CassStatementBindInt32ByName(
3347 : CassStatement* statement, const char* name, cass_int32_t value) {
3348 0 : return cass_statement_bind_int32_by_name(statement, name, value);
3349 : }
3350 :
3351 0 : CassError CassDatastaxLibrary::CassStatementBindInt64ByName(
3352 : CassStatement* statement, const char* name, cass_int64_t value) {
3353 0 : return cass_statement_bind_int64_by_name(statement, name, value);
3354 : }
3355 :
3356 0 : CassError CassDatastaxLibrary::CassStatementBindUuidByName(
3357 : CassStatement* statement, const char* name, CassUuid value) {
3358 0 : return cass_statement_bind_uuid_by_name(statement, name, value);
3359 : }
3360 :
3361 0 : CassError CassDatastaxLibrary::CassStatementBindDoubleByName(
3362 : CassStatement* statement, const char* name, cass_double_t value) {
3363 0 : return cass_statement_bind_double_by_name(statement, name, value);
3364 : }
3365 :
3366 0 : CassError CassDatastaxLibrary::CassStatementBindInetByName(
3367 : CassStatement* statement, const char* name, CassInet value) {
3368 0 : return cass_statement_bind_inet_by_name(statement, name, value);
3369 : }
3370 :
3371 0 : CassError CassDatastaxLibrary::CassStatementBindBytesByNameN(
3372 : CassStatement* statement,
3373 : const char* name, size_t name_length, const cass_byte_t* value,
3374 : size_t value_length) {
3375 0 : return cass_statement_bind_bytes_by_name_n(statement, name, name_length,
3376 0 : value, value_length);
3377 : }
3378 :
3379 : // CassPrepare
3380 0 : void CassDatastaxLibrary::CassPreparedFree(const CassPrepared* prepared) {
3381 0 : cass_prepared_free(prepared);
3382 0 : }
3383 :
3384 0 : CassStatement* CassDatastaxLibrary::CassPreparedBind(
3385 : const CassPrepared* prepared) {
3386 0 : return cass_prepared_bind(prepared);
3387 : }
3388 :
3389 : // CassValue
3390 67691 : CassValueType CassDatastaxLibrary::GetCassValueType(const CassValue* value) {
3391 67691 : return cass_value_type(value);
3392 : }
3393 :
3394 41514 : CassError CassDatastaxLibrary::CassValueGetString(const CassValue* value,
3395 : const char** output, size_t* output_size) {
3396 41514 : return cass_value_get_string(value, output, output_size);
3397 : }
3398 :
3399 0 : CassError CassDatastaxLibrary::CassValueGetInt8(const CassValue* value,
3400 : cass_int8_t* output) {
3401 0 : return cass_value_get_int8(value, output);
3402 : }
3403 :
3404 0 : CassError CassDatastaxLibrary::CassValueGetInt16(const CassValue* value,
3405 : cass_int16_t* output) {
3406 0 : return cass_value_get_int16(value, output);
3407 : }
3408 :
3409 17736 : CassError CassDatastaxLibrary::CassValueGetInt32(const CassValue* value,
3410 : cass_int32_t* output) {
3411 17736 : return cass_value_get_int32(value, output);
3412 : }
3413 :
3414 3464 : CassError CassDatastaxLibrary::CassValueGetInt64(const CassValue* value,
3415 : cass_int64_t* output) {
3416 3464 : return cass_value_get_int64(value, output);
3417 : }
3418 :
3419 4571 : CassError CassDatastaxLibrary::CassValueGetUuid(const CassValue* value,
3420 : CassUuid* output) {
3421 4571 : return cass_value_get_uuid(value, output);
3422 : }
3423 :
3424 0 : CassError CassDatastaxLibrary::CassValueGetDouble(const CassValue* value,
3425 : cass_double_t* output) {
3426 0 : return cass_value_get_double(value, output);
3427 : }
3428 :
3429 408 : CassError CassDatastaxLibrary::CassValueGetInet(const CassValue* value,
3430 : CassInet* output) {
3431 408 : return cass_value_get_inet(value, output);
3432 : }
3433 :
3434 0 : CassError CassDatastaxLibrary::CassValueGetBytes(const CassValue* value,
3435 : const cass_byte_t** output, size_t* output_size) {
3436 0 : return cass_value_get_bytes(value, output, output_size);
3437 : }
3438 :
3439 96340 : cass_bool_t CassDatastaxLibrary::CassValueIsNull(const CassValue* value) {
3440 96340 : return cass_value_is_null(value);
3441 : }
3442 :
3443 : // CassInet
3444 0 : CassInet CassDatastaxLibrary::CassInetInitV4(
3445 : const cass_uint8_t* address) {
3446 0 : return cass_inet_init_v4(address);
3447 : }
3448 :
3449 0 : CassInet CassDatastaxLibrary::CassInetInitV6(
3450 : const cass_uint8_t* address) {
3451 0 : return cass_inet_init_v6(address);
3452 : }
3453 :
3454 : // CassRow
3455 96340 : const CassValue* CassDatastaxLibrary::CassRowGetColumn(const CassRow* row,
3456 : size_t index) {
3457 96340 : return cass_row_get_column(row, index);
3458 : }
3459 :
3460 : // CassLog
3461 3703 : void CassDatastaxLibrary::CassLogSetLevel(CassLogLevel log_level) {
3462 3703 : cass_log_set_level(log_level);
3463 3703 : }
3464 :
3465 3703 : void CassDatastaxLibrary::CassLogSetCallback(CassLogCallback callback,
3466 : void* data) {
3467 3703 : cass_log_set_callback(callback, data);
3468 3703 : }
3469 :
3470 : } // namespace interface
3471 : } // namespace cql
3472 : } // namespace cass
|
