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