jobdatastore.cpp 30.2 KB
Newer Older
1
2
3
4
5
6
7
8
9
//================================================================================
// Name        : jobdatastore.cpp
// Author      : Axel Auweter, Micha Mueller
// Copyright   : Leibniz Supercomputing Centre
// Description : C++ API implementation for inserting and querying DCDB job data.
//================================================================================

//================================================================================
// This file is part of DCDB (DataCenter DataBase)
10
// Copyright (C) 2011-2018 Leibniz Supercomputing Centre
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
//
// This library is free software; you can redistribute it and/or
// modify it under the terms of the GNU Lesser General Public
// License as published by the Free Software Foundation; either
// version 2.1 of the License, or (at your option) any later version.
//
// This library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
// Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public
// License along with this library; if not, write to the Free Software
// Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
//================================================================================

/**
28
29
30
31
 * @file
 * @brief This file contains actual implementations of the jobdatastore
 * interface. The interface itself forwards all functions to the internal
 * JobDataStoreImpl. The real logic is implemented in the JobDataStoreImpl.
32
33
 */

34
#include <cinttypes>
35
36
37
38
39
40
41
42
43
44
45
46
#include <list>
#include <string>

#include "cassandra.h"

#include "dcdb/jobdatastore.h"
#include "jobdatastore_internal.h"
#include "dcdb/connection.h"
#include "dcdbglobals.h"

using namespace DCDB;

47
48
49
50
51
52
53
54
55
56
/**
 * @details
 * Since we want high-performance inserts, we prepare the insert CQL query in
 * advance and only bind it on the actual insert.
 */
void JobDataStoreImpl::prepareInsert(uint64_t ttl) {
  CassError rc = CASS_OK;
  CassFuture* future = NULL;
  const char* query;

57
  /* Free the old prepared if necessary. */
58
59
60
61
62
63
  if (preparedInsert) {
    cass_prepared_free(preparedInsert);
  }

  char *queryBuf = NULL;
  if (ttl == 0) {
64
65
    query = "INSERT INTO " JD_KEYSPACE_NAME "." CF_JOBDATA
        " (jid, uid, start_ts, end_ts, nodes) VALUES (?, ?, ?, ?, ?);";
66
67
68
  }
  else {
    queryBuf = (char*)malloc(256);
69
70
71
    snprintf(queryBuf, 256, "INSERT INTO " JD_KEYSPACE_NAME "." CF_JOBDATA
        " (jid, uid, start_ts, end_ts, nodes) VALUES (?, ?, ?, ?, ?) "
        "USING TTL %" PRIu64 " ;", ttl);
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
    query = queryBuf;
  }

  future = cass_session_prepare(session, query);
  cass_future_wait(future);

  rc = cass_future_error_code(future);
  if (rc != CASS_OK) {
    connection->printError(future);
  } else {
    preparedInsert = cass_future_get_prepared(future);
  }

  cass_future_free(future);
  if (queryBuf) {
    free(queryBuf);
  }
}

91
92
/**
 * @details
93
 * Extract all data from the JobData object and push it into the data store.
94
 */
95
JDError JobDataStoreImpl::insertJob(JobData& jdata) {
96
97
  /* Check if the input for the primary key is valid and reasonable */
  if (jdata.startTime.getRaw() == 0) {
98
    return JD_BADPARAMS;
99
100
101
102
  }

  JDError error = JD_UNKNOWNERROR;

103
104
105
106
107
108
109
110
111
112
113
114
115
116
  /* Insert into Cassandra */
  CassError rc = CASS_OK;
  CassStatement* statement = NULL;
  CassFuture *future = NULL;

  statement = cass_prepared_bind(preparedInsert);

  cass_statement_bind_int64_by_name(statement, "jid", jdata.jobId);
  cass_statement_bind_int64_by_name(statement, "uid", jdata.userId);
  cass_statement_bind_int64_by_name(statement, "start_ts",
                                    jdata.startTime.getRaw());
  cass_statement_bind_int64_by_name(statement, "end_ts",
                                    jdata.endTime.getRaw());

117
  /* Copy the string node list to a varchar set */
118
119
120
121
122
123
124
125
  CassCollection* set = cass_collection_new(CASS_COLLECTION_TYPE_SET,
                                            jdata.nodes.size());
  for (auto& s : jdata.nodes) {
    cass_collection_append_string(set, s.c_str());
  }

  cass_statement_bind_collection_by_name(statement, "nodes", set);

126
  /* All parameters bound. Now execute the statement asynchronously */
127
  future = cass_session_execute(session, statement);
128
129
130
131
132

  /* Clean up in the meantime */
  cass_collection_free(set);

  /* Wait for the statement to finish */
133
134
135
136
  cass_future_wait(future);

  rc = cass_future_error_code(future);
  if (rc != CASS_OK) {
137
138
139
140
    connection->printError(future);
    error = JD_UNKNOWNERROR;
  } else {
    error = JD_OK;
141
142
143
144
  }

  cass_future_free(future);
  cass_statement_free(statement);
145
  return error;
146
147
148
149
}

/**
 * @details
150
151
152
153
154
155
156
 * Update the job with matching JobId and StartTs in the data store with the
 * values provided by the given JobData object. If no such job exists in the
 * data store yet, it is inserted.
 * The JobData object is expected to be complete. Partial
 * updates of only selected fields are not supported. Instead, one has to
 * retrieve the other JobData information via getJobById() or
 * getJobByPrimaryKey() first and complete its JobData object for the update.
157
 */
158
159
160
161
162
163
JDError JobDataStoreImpl::updateJob(JobData& jdata) {
  /* Check if the input for the primary key is valid and reasonable */
  if (jdata.startTime.getRaw() == 0) {
    return JD_BADPARAMS;
  }

164
165
  JDError error = JD_UNKNOWNERROR;

166
  /* Update entry in Cassandra (actually upserts) */
167
  CassError rc = CASS_OK;
168
169
170
171
  CassStatement* statement = nullptr;
  CassFuture* future = nullptr;
  const char* query = "UPDATE " JD_KEYSPACE_NAME "." CF_JOBDATA
      " SET uid = ?, end_ts = ?, nodes = ? WHERE jid = ? AND start_ts = ? ;";
172

173
  statement = cass_statement_new(query, 5);
174

175
176
177
178
  cass_statement_bind_int64(statement, 3, jdata.jobId);
  cass_statement_bind_int64(statement, 0, jdata.userId);
  cass_statement_bind_int64(statement, 4, jdata.startTime.getRaw());
  cass_statement_bind_int64(statement, 1, jdata.endTime.getRaw());
179

180
181
182
183
184
185
  /* Copy the string node list to a varchar set */
  CassCollection* set = cass_collection_new(CASS_COLLECTION_TYPE_SET,
                                            jdata.nodes.size());
  for (auto& s : jdata.nodes) {
    cass_collection_append_string(set, s.c_str());
  }
186

187
  cass_statement_bind_collection(statement, 2, set);
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207

  /* All parameters bound. Now execute the statement asynchronously */
  future = cass_session_execute(session, statement);

  /* Clean up in the meantime */
  cass_collection_free(set);

  /* Wait for the statement to finish */
  cass_future_wait(future);

  rc = cass_future_error_code(future);
  if (rc != CASS_OK) {
    connection->printError(future);
    error = JD_UNKNOWNERROR;
  } else {
    error = JD_OK;
  }

  cass_future_free(future);
  cass_statement_free(statement);
208

209
  return error;
210
211
212
213
}

/**
 * @details
214
215
216
 * Update the job with matching JobId and StartTs in the data store with the
 * provided end time. If no such job exists in the data store yet, it is
 * inserted.
217
 */
218
219
220
221
JDError JobDataStoreImpl::updateEndtime(JobId jobId, TimeStamp startTs,
                                        TimeStamp endTime) {
  /* Check if the input for the primary key is valid and reasonable */
  if (startTs.getRaw() == 0) {
222
    return JD_BADPARAMS;
223
224
225
226
  }

  JDError error = JD_UNKNOWNERROR;

227
  /* Update entry in Cassandra (actually upserts) */
228
229
230
231
  CassError rc = CASS_OK;
  CassStatement* statement = nullptr;
  CassFuture* future = nullptr;
  const char* query = "UPDATE " JD_KEYSPACE_NAME "." CF_JOBDATA
232
      " SET end_ts = ? WHERE jid = ?, start_ts = ? ;";
233

234
  statement = cass_statement_new(query, 3);
235

236
237
238
  cass_statement_bind_int64(statement, 1, jobId);
  cass_statement_bind_int64(statement, 2, startTs.getRaw());
  cass_statement_bind_int64(statement, 0, endTime.getRaw());
239

240
241
242
243
244
245
246
247
248
249
250
251
  /* All parameters bound. Now execute the statement asynchronously */
  future = cass_session_execute(session, statement);

  /* Wait for the statement to finish */
  cass_future_wait(future);

  rc = cass_future_error_code(future);
  if (rc != CASS_OK) {
    connection->printError(future);
    error = JD_UNKNOWNERROR;
  } else {
    error = JD_OK;
252
253
  }

254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
  cass_future_free(future);
  cass_statement_free(statement);

  return error;
}

/**
 * @details
 * Delete the entry with matching JobId and start TimeStamp from the data store.
 */
JDError JobDataStoreImpl::deleteJob(JobId jid, TimeStamp startTs) {
  JDError error = JD_UNKNOWNERROR;

  /* Remove entry from Cassandra */
  CassError rc = CASS_OK;
  CassStatement* statement = nullptr;
  CassFuture* future = nullptr;
  const char* query = "DELETE FROM " JD_KEYSPACE_NAME "." CF_JOBDATA
      " WHERE jid = ? AND start_ts = ?;";

  statement = cass_statement_new(query, 2);

  cass_statement_bind_int64(statement, 0, jid);
  cass_statement_bind_int64(statement, 1, startTs.getRaw());
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292

  /* All parameters bound. Now execute the statement asynchronously */
  future = cass_session_execute(session, statement);

  /* Wait for the statement to finish */
  cass_future_wait(future);

  rc = cass_future_error_code(future);
  if (rc != CASS_OK) {
    connection->printError(future);
    error = JD_UNKNOWNERROR;
  } else {
    error = JD_OK;
  }

293
  cass_future_free(future);
294
295
296
  cass_statement_free(statement);

  return error;
297
298
299
300
}

/**
 * @details
301
302
 * Find the entry in the data store with matching JobId and start_ts and store
 * the corresponding values in the JobData object.
303
 */
304
305
JDError JobDataStoreImpl::getJobByPrimaryKey(JobData& job, JobId jid,
                                             TimeStamp startTs) {
306
307
  JDError error = JD_UNKNOWNERROR;

308
  /* Select entry from Cassandra */
309
310
311
  CassError rc = CASS_OK;
  CassStatement* statement = nullptr;
  CassFuture* future = nullptr;
312
313
  const char* query = "SELECT * FROM " JD_KEYSPACE_NAME "." CF_JOBDATA
      " WHERE jid = ? AND start_ts = ?;";
314

315
  statement = cass_statement_new(query, 2);
316
317

  cass_statement_bind_int64(statement, 0, jid);
318
  cass_statement_bind_int64(statement, 1, startTs.getRaw());
319
320
321
322
323
324
325
326
327
328
329
330
331

  /* All parameters bound. Now execute the statement asynchronously */
  future = cass_session_execute(session, statement);

  /* Wait for the statement to finish */
  cass_future_wait(future);

  rc = cass_future_error_code(future);
  if (rc != CASS_OK) {
    connection->printError(future);
    error = JD_UNKNOWNERROR;
  } else {
    error = JD_OK;
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382

    const CassResult* cresult = cass_future_get_result(future);
    size_t rowCnt = cass_result_row_count(cresult);

    /* Check if the returned data is reasonable */
    if (rowCnt == 0) {
      error = JD_JOBKEYNOTFOUND;
    } else {
      /* Retrieve data from result */
      const CassRow* row = cass_result_first_row(cresult);

      cass_int64_t jobId, userId, startTs, endTs;

      /* jid and start_ts are always set. Other values should be checked */
      cass_value_get_int64(cass_row_get_column_by_name(row, "jid"), &jobId);
      cass_value_get_int64(cass_row_get_column_by_name(row, "start_ts"),
                                                       &startTs);
      if (cass_value_get_int64(cass_row_get_column_by_name(row, "uid"),
                               &userId) != CASS_OK) {
        userId = 0;
        error = JD_PARSINGERROR;
      }
      if (cass_value_get_int64(cass_row_get_column_by_name(row, "end_ts"),
                               &endTs) != CASS_OK) {
        endTs = 0;
        error = JD_PARSINGERROR;
      }

      /* Copy the data in the JobData object */
      job.jobId = (JobId) jobId;
      job.userId = (UserId) userId;
      job.startTime = (uint64_t) startTs;
      job.endTime = (uint64_t) endTs;

      /* Do not forget about the nodes... */
      const char* nodeStr;
      size_t nodeStr_len;

      const CassValue* set = cass_row_get_column_by_name(row, "nodes");
      CassIterator* setIt = cass_iterator_from_collection(set);

      while (cass_iterator_next(setIt)) {
        cass_value_get_string(cass_iterator_get_value(setIt),
                              &nodeStr, &nodeStr_len);
        job.nodes.emplace_back(nodeStr, nodeStr_len);
      }

      cass_iterator_free(setIt);
    }

    cass_result_free(cresult);
383
384
385
386
387
388
  }

  cass_future_free(future);
  cass_statement_free(statement);

  return error;
389
390
391
392
}

/**
 * @details
393
394
 * Find the entry in the data store with matching JobId and highest start_ts
 * value (= most recent job) and store the
395
 * corresponding values in the JobData object.
396
 */
397
JDError JobDataStoreImpl::getJobById(JobData& job, JobId jid) {
398
399
400
401
402
403
404
  JDError error = JD_UNKNOWNERROR;

  /* Select entry from Cassandra */
  CassError rc = CASS_OK;
  CassStatement* statement = nullptr;
  CassFuture* future = nullptr;
  const char* query = "SELECT * FROM " JD_KEYSPACE_NAME "." CF_JOBDATA
405
      " WHERE jid = ? ORDER BY start_ts DESC LIMIT 1;";
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435

  statement = cass_statement_new(query, 1);

  cass_statement_bind_int64(statement, 0, jid);

  /* All parameters bound. Now execute the statement asynchronously */
  future = cass_session_execute(session, statement);

  /* Wait for the statement to finish */
  cass_future_wait(future);

  rc = cass_future_error_code(future);
  if (rc != CASS_OK) {
    connection->printError(future);
    error = JD_UNKNOWNERROR;
  } else {
    error = JD_OK;

    const CassResult* cresult = cass_future_get_result(future);
    size_t rowCnt = cass_result_row_count(cresult);

    /* Check if the returned data is reasonable */
    if (rowCnt == 0) {
      error = JD_JOBIDNOTFOUND;
    } else {
      /* Retrieve data from result */
      const CassRow* row = cass_result_first_row(cresult);

      cass_int64_t jobId, userId, startTs, endTs;

436
      /* jid and start_ts are always set. Other values should be checked */
437
      cass_value_get_int64(cass_row_get_column_by_name(row, "jid"), &jobId);
438
439
440
441
442
      cass_value_get_int64(cass_row_get_column_by_name(row, "start_ts"),
                                                       &startTs);
      if (cass_value_get_int64(cass_row_get_column_by_name(row, "uid"),
                               &userId) != CASS_OK) {
        userId = 0;
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
        error = JD_PARSINGERROR;
      }
      if (cass_value_get_int64(cass_row_get_column_by_name(row, "end_ts"),
                               &endTs) != CASS_OK) {
        endTs = 0;
        error = JD_PARSINGERROR;
      }

      /* Copy the data in the JobData object */
      job.jobId = (JobId) jobId;
      job.userId = (UserId) userId;
      job.startTime = (uint64_t) startTs;
      job.endTime = (uint64_t) endTs;

      /* Do not forget about the nodes... */
      const char* nodeStr;
      size_t nodeStr_len;

      const CassValue* set = cass_row_get_column_by_name(row, "nodes");
      CassIterator* setIt = cass_iterator_from_collection(set);

      while (cass_iterator_next(setIt)) {
        cass_value_get_string(cass_iterator_get_value(setIt),
                              &nodeStr, &nodeStr_len);
        job.nodes.emplace_back(nodeStr, nodeStr_len);
      }

      cass_iterator_free(setIt);
    }

    cass_result_free(cresult);
  }

  cass_future_free(future);
  cass_statement_free(statement);

  return error;
480
481
482
483
}

/**
 * @details
484
485
 * Find all entries in the data store whose start_ts AND end_ts lay within
 * the specified interval. Store the found entries in the JobData list.
486
 */
487
488
489
JDError JobDataStoreImpl::getJobsInIntervalExcl(std::list<JobData>& jobs,
                                                TimeStamp intervalStart,
                                                TimeStamp intervalEnd) {
490
491
  /* Check if the input is valid and reasonable */
  if (intervalEnd.getRaw() == 0) {
492
    return JD_BADPARAMS;
493
494
  }
  if (intervalStart >= intervalEnd) {
495
    return JD_BADPARAMS;
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
  }

  JDError error = JD_UNKNOWNERROR;

  /* Select entries from Cassandra */
  CassError rc = CASS_OK;
  CassStatement* statement = nullptr;
  CassFuture* future = nullptr;
  const char* query = "SELECT * FROM " JD_KEYSPACE_NAME "." CF_JOBDATA
      " WHERE start_ts >= ? AND end_ts <= ? ;";

  statement = cass_statement_new(query, 2);

  cass_statement_bind_int64(statement, 0, intervalStart.getRaw());
  cass_statement_bind_int64(statement, 1, intervalEnd.getRaw());

  /* All parameters bound. Now execute the statement asynchronously */
  future = cass_session_execute(session, statement);

  /* Wait for the statement to finish */
  cass_future_wait(future);

  rc = cass_future_error_code(future);
  if (rc != CASS_OK) {
    connection->printError(future);
    error = JD_UNKNOWNERROR;
  } else {
    error = JD_OK;

    /* Retrieve data from result */
    const CassResult* cresult = cass_future_get_result(future);
    CassIterator* rowIt = cass_iterator_from_result(cresult);

    JobData job;

    while (cass_iterator_next(rowIt)) {
      const CassRow* row = cass_iterator_get_row(rowIt);

      cass_int64_t jobId, userId, startTs, endTs;

      /* jid and uid should always be set. Other values should be checked */
      cass_value_get_int64(cass_row_get_column_by_name(row, "jid"), &jobId);
      cass_value_get_int64(cass_row_get_column_by_name(row, "uid"), &userId);
      if (cass_value_get_int64(cass_row_get_column_by_name(row, "start_ts"),
                               &startTs) != CASS_OK) {
        startTs = 0;
        error = JD_PARSINGERROR;
      }
      if (cass_value_get_int64(cass_row_get_column_by_name(row, "end_ts"),
                               &endTs) != CASS_OK) {
        endTs = 0;
        error = JD_PARSINGERROR;
      }

      /* Copy the data into job object */
      job.jobId = (JobId) jobId;
      job.userId = (UserId) userId;
      job.startTime = (uint64_t) startTs;
      job.endTime = (uint64_t) endTs;

      /* Do not forget about the nodes... */
      const char* nodeStr;
      size_t nodeStr_len;

      const CassValue* set = cass_row_get_column_by_name(row, "nodes");
      CassIterator* setIt = cass_iterator_from_collection(set);

      while (cass_iterator_next(setIt)) {
        cass_value_get_string(cass_iterator_get_value(setIt),
                              &nodeStr, &nodeStr_len);
        job.nodes.emplace_back(nodeStr, nodeStr_len);
      }

      //TODO job.nodes list deep copied?
      jobs.push_back(job);
      job.nodes.clear();
      cass_iterator_free(setIt);
    }

    cass_iterator_free(rowIt);
    cass_result_free(cresult);
  }

  cass_future_free(future);
  cass_statement_free(statement);

  return error;
583
584
585
586
}

/**
 * @details
587
588
589
590
591
592
 * Find all entries in the data store whose start_ts OR end_ts lays within
 * the specified interval. Store the found entries in the JobData list.
 * Cassandra only supports AND conditions in its query language. Therefore
 * we cannot SELECT directly the required jobs. Instead we have to do two
 * selects and manually deduplicate the results.
 * TODO Doing two request successively may tangle up error codes.
593
 */
594
595
596
JDError JobDataStoreImpl::getJobsInIntervalIncl(std::list<JobData>& jobs,
                                                TimeStamp intervalStart,
                                                TimeStamp intervalEnd) {
597
598
  /* Check if the input is valid and reasonable */
  if (intervalEnd.getRaw() == 0) {
599
    return JD_BADPARAMS;
600
601
  }
  if (intervalStart >= intervalEnd) {
602
    return JD_BADPARAMS;
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
  }

  JDError error = JD_UNKNOWNERROR;

  /* +++ First SELECT +++ */
  /* Select entries from Cassandra where start_ts lays within the interval */
  CassError rc = CASS_OK;
  CassStatement* statement = nullptr;
  CassFuture* future = nullptr;
  const char* query = "SELECT * FROM " JD_KEYSPACE_NAME "." CF_JOBDATA
      " WHERE start_ts >= ? AND start_ts <= ? ;";

  statement = cass_statement_new(query, 2);

  cass_statement_bind_int64(statement, 0, intervalStart.getRaw());
  cass_statement_bind_int64(statement, 1, intervalEnd.getRaw());

  /* All parameters bound. Now execute the statement asynchronously */
  future = cass_session_execute(session, statement);

  /* Wait for the statement to finish */
  cass_future_wait(future);

  rc = cass_future_error_code(future);
  if (rc != CASS_OK) {
    connection->printError(future);
    error = JD_UNKNOWNERROR;
  } else {
    error = JD_OK;

    /* Retrieve data from result */
    const CassResult* cresult = cass_future_get_result(future);
    CassIterator* rowIt = cass_iterator_from_result(cresult);

    JobData job;

    while (cass_iterator_next(rowIt)) {
      const CassRow* row = cass_iterator_get_row(rowIt);

      cass_int64_t jobId, userId, startTs, endTs;

      /* jid and uid should always be set. Other values should be checked */
      cass_value_get_int64(cass_row_get_column_by_name(row, "jid"), &jobId);
      cass_value_get_int64(cass_row_get_column_by_name(row, "uid"), &userId);
      if (cass_value_get_int64(cass_row_get_column_by_name(row, "start_ts"),
                               &startTs) != CASS_OK) {
        startTs = 0;
        error = JD_PARSINGERROR;
      }
      if (cass_value_get_int64(cass_row_get_column_by_name(row, "end_ts"),
                               &endTs) != CASS_OK) {
        endTs = 0;
        error = JD_PARSINGERROR;
      }

      /* Copy the data into job object */
      job.jobId = (JobId) jobId;
      job.userId = (UserId) userId;
      job.startTime = (uint64_t) startTs;
      job.endTime = (uint64_t) endTs;

      /* Do not forget about the nodes... */
      const char* nodeStr;
      size_t nodeStr_len;

      const CassValue* set = cass_row_get_column_by_name(row, "nodes");
      CassIterator* setIt = cass_iterator_from_collection(set);

      while (cass_iterator_next(setIt)) {
        cass_value_get_string(cass_iterator_get_value(setIt),
                              &nodeStr, &nodeStr_len);
        job.nodes.emplace_back(nodeStr, nodeStr_len);
      }

      //TODO job.nodes list deep copied?
      jobs.push_back(job);
      job.nodes.clear();
      cass_iterator_free(setIt);
    }

    cass_iterator_free(rowIt);
    cass_result_free(cresult);
  }

  cass_future_free(future);
  cass_statement_free(statement);

  /* +++ Second SELECT +++ */
  /* Select entries from Cassandra where end_ts lays within the interval */
  query = "SELECT * FROM " JD_KEYSPACE_NAME "." CF_JOBDATA
      " WHERE end_ts >= ? AND end_ts <= ? ;";

  statement = cass_statement_new(query, 2);

  cass_statement_bind_int64(statement, 0, intervalStart.getRaw());
  cass_statement_bind_int64(statement, 1, intervalEnd.getRaw());

  /* All parameters bound. Now execute the statement asynchronously */
  future = cass_session_execute(session, statement);

  /* Wait for the statement to finish */
  cass_future_wait(future);

  rc = cass_future_error_code(future);
  if (rc != CASS_OK) {
    connection->printError(future);
    error = JD_UNKNOWNERROR;
  } else {
    /* Retrieve data from result */
    const CassResult* cresult = cass_future_get_result(future);
    CassIterator* rowIt = cass_iterator_from_result(cresult);

    JobData job;

    while (cass_iterator_next(rowIt)) {
      const CassRow* row = cass_iterator_get_row(rowIt);

      cass_int64_t jobId, userId, startTs, endTs;

      /* jid and uid should always be set. Other values should be checked */
      cass_value_get_int64(cass_row_get_column_by_name(row, "jid"), &jobId);
      cass_value_get_int64(cass_row_get_column_by_name(row, "uid"), &userId);
      if (cass_value_get_int64(cass_row_get_column_by_name(row, "start_ts"),
                               &startTs) != CASS_OK) {
        startTs = 0;
        error = JD_PARSINGERROR;
      }
      if (cass_value_get_int64(cass_row_get_column_by_name(row, "end_ts"),
                               &endTs) != CASS_OK) {
        endTs = 0;
        error = JD_PARSINGERROR;
      }

      /* Copy the data into job object */
      job.jobId = (JobId) jobId;

      /* Manual "deduplication" */
      bool alreadyPresent = false;
      /* TODO Possible optimization: iterate only jobs from first SELECT */
      for (const auto& j : jobs) {
        if (j.jobId == job.jobId) {
          alreadyPresent = true;
        }
      }

      if (!alreadyPresent) {
        job.userId = (UserId) userId;
        job.startTime = (uint64_t) startTs;
        job.endTime = (uint64_t) endTs;

        /* Do not forget about the nodes... */
        const char* nodeStr;
        size_t nodeStr_len;

        const CassValue* set = cass_row_get_column_by_name(row, "nodes");
        CassIterator* setIt = cass_iterator_from_collection(set);

        while (cass_iterator_next(setIt)) {
          cass_value_get_string(cass_iterator_get_value(setIt),
                                &nodeStr, &nodeStr_len);
          job.nodes.emplace_back(nodeStr, nodeStr_len);
        }

        //TODO job.nodes list deep copied?
        jobs.push_back(job);
        job.nodes.clear();
        cass_iterator_free(setIt);
      }
    }

    cass_iterator_free(rowIt);
    cass_result_free(cresult);
  }

  cass_future_free(future);
  cass_statement_free(statement);

  return error;
781
782
783
784
}

/**
 * @details
785
786
 * Find the entry in the data store with matching JobId and highest start_ts
 * value (= most recent job) and store the
787
 * corresponding nodes in the NodeList.
788
 */
789
790
JDError JobDataStoreImpl::getNodeList(NodeList& nodes, JobId jid,
                                      TimeStamp startTs) {
791
792
793
794
795
796
797
  JDError error = JD_UNKNOWNERROR;

  /* Select entry from Cassandra */
  CassError rc = CASS_OK;
  CassStatement* statement = nullptr;
  CassFuture* future = nullptr;
  const char* query = "SELECT nodes FROM " JD_KEYSPACE_NAME "." CF_JOBDATA
798
      " WHERE jid = ? ORDER BY start_ts LIMIT 1;";
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849

  statement = cass_statement_new(query, 1);

  cass_statement_bind_int64(statement, 0, jid);

  /* All parameters bound. Now execute the statement asynchronously */
  future = cass_session_execute(session, statement);

  /* Wait for the statement to finish */
  cass_future_wait(future);

  rc = cass_future_error_code(future);
  if (rc != CASS_OK) {
    connection->printError(future);
    error = JD_UNKNOWNERROR;
  } else {
    error = JD_OK;

    const CassResult* cresult = cass_future_get_result(future);
    size_t rowCnt = cass_result_row_count(cresult);

    /* Check if the returned data is reasonable */
    if (rowCnt == 0) {
      error = JD_JOBIDNOTFOUND;
    } else {
      /* Retrieve data from result */
      const CassRow* row = cass_result_first_row(cresult);

      /* Copy the nodes in the NodeList */
      const char* nodeStr;
      size_t nodeStr_len;

      const CassValue* set = cass_row_get_column_by_name(row, "nodes");
      CassIterator* setIt = cass_iterator_from_collection(set);

      while (cass_iterator_next(setIt)) {
        cass_value_get_string(cass_iterator_get_value(setIt),
                              &nodeStr, &nodeStr_len);
        nodes.emplace_back(nodeStr, nodeStr_len);
      }

      cass_iterator_free(setIt);
    }

    cass_result_free(cresult);
  }

  cass_future_free(future);
  cass_statement_free(statement);

  return error;
850
851
852
853
854
855
856
857
858
859
860
}

/**
 * @details
 * This constructor sets the internal connection variable to
 * the externally provided Connection object and also
 * retrieves the CassSession pointer of the connection.
 */
JobDataStoreImpl::JobDataStoreImpl(Connection* conn) {
  connection = conn;
  session = connection->getSessionHandle();
861
862
863

  preparedInsert = nullptr;
  prepareInsert(0);
864
865
866
867
868
}

/**
 * @details
 * The destructor just resets the internal pointers. Deletion of the pointers
869
 * (except preparedInsert) is not our responsibility.
870
871
872
873
 */
JobDataStoreImpl::~JobDataStoreImpl() {
  connection = nullptr;
  session = nullptr;
874
875
876
  if (preparedInsert) {
      cass_prepared_free(preparedInsert);
  }
877
878
879
880
881
882
883
884
885
}

/* ########################################################################## */

/**
 * @details
 * Instead of doing the actual work, this function simply forwards to the
 * corresponding function of the JobDataStoreImpl class.
 */
886
887
JDError JobDataStore::insertJob(JobData& jdata) {
  return impl->insertJob(jdata);
888
889
890
891
892
893
894
}

/**
 * @details
 * Instead of doing the actual work, this function simply forwards to the
 * corresponding function of the JobDataStoreImpl class.
 */
895
896
897
898
899
900
901
902
903
904
905
906
JDError JobDataStore::updateJob(JobData& jdata) {
  return impl->updateJob(jdata);
}

/**
 * @details
 * Instead of doing the actual work, this function simply forwards to the
 * corresponding function of the JobDataStoreImpl class.
 */
JDError JobDataStore::updateEndtime(JobId jobId, TimeStamp startTs,
                                    TimeStamp endTime) {
  return impl->updateEndtime(jobId, startTs, endTime);
907
908
909
910
911
912
913
}

/**
 * @details
 * Instead of doing the actual work, this function simply forwards to the
 * corresponding function of the JobDataStoreImpl class.
 */
914
915
JDError JobDataStore::deleteJob(JobId jid, TimeStamp startTs) {
  return impl->deleteJob(jid, startTs);
916
917
918
919
920
921
922
}

/**
 * @details
 * Instead of doing the actual work, this function simply forwards to the
 * corresponding function of the JobDataStoreImpl class.
 */
923
924
925
JDError JobDataStore::getJobByPrimaryKey(JobData& job, JobId jid,
                                         TimeStamp startTs) {
  return impl->getJobByPrimaryKey(job, jid, startTs);
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
}

/**
 * @details
 * Instead of doing the actual work, this function simply forwards to the
 * corresponding function of the JobDataStoreImpl class.
 */
JDError JobDataStore::getJobById(JobData& job, JobId jid) {
  return impl->getJobById(job, jid);
}

/**
 * @details
 * Instead of doing the actual work, this function simply forwards to the
 * corresponding function of the JobDataStoreImpl class.
 */
942
943
944
945
JDError JobDataStore::getJobsInIntervalExcl(std::list<JobData>& jobs,
                                            TimeStamp intervalStart,
                                            TimeStamp intervalEnd) {
  return impl->getJobsInIntervalExcl(jobs, intervalStart, intervalEnd);
946
947
948
949
950
951
952
}

/**
 * @details
 * Instead of doing the actual work, this function simply forwards to the
 * corresponding function of the JobDataStoreImpl class.
 */
953
954
955
956
JDError JobDataStore::getJobsInIntervalIncl(std::list<JobData>& jobs,
                                            TimeStamp intervalStart,
                                            TimeStamp intervalEnd) {
  return impl->getJobsInIntervalIncl(jobs, intervalStart, intervalEnd);
957
958
959
960
961
962
963
}

/**
 * @details
 * Instead of doing the actual work, this function simply forwards to the
 * corresponding function of the JobDataStoreImpl class.
 */
964
965
966
JDError JobDataStore::getNodeList(NodeList& nodes, JobId jid,
                                  TimeStamp startTs) {
  return impl->getNodeList(nodes, jid, startTs);
967
968
969
970
971
972
973
974
975
976
977
978
979
}

/**
 * @details
 * This constructor allocates the implementation class which
 * holds the actual implementation of the class functionality.
 */
JobDataStore::JobDataStore(Connection* conn) {
  impl = new JobDataStoreImpl(conn);
}

/**
 * @details
980
 * The JobDataStore destructor deallocates the
981
982
983
984
 * JobDataStoreImpl and CassandraBackend objects.
 */
JobDataStore::~JobDataStore() {
  /* Clean up... */
985
986
987
  if (impl) {
    delete impl;
  }
988
}