1 /*
2 Copyright (c) 2003, 2021, Oracle and/or its affiliates.
3
4 This program is free software; you can redistribute it and/or modify
5 it under the terms of the GNU General Public License, version 2.0,
6 as published by the Free Software Foundation.
7
8 This program is also distributed with certain software (including
9 but not limited to OpenSSL) that is licensed under separate terms,
10 as designated in a particular file or component or in included license
11 documentation. The authors of MySQL hereby grant you an additional
12 permission to link the program and your derivative works with the
13 separately licensed software that they have included with MySQL.
14
15 This program is distributed in the hope that it will be useful,
16 but WITHOUT ANY WARRANTY; without even the implied warranty of
17 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
18 GNU General Public License, version 2.0, for more details.
19
20 You should have received a copy of the GNU General Public License
21 along with this program; if not, write to the Free Software
22 Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
23 */
24
25 #include <HugoOperations.hpp>
26
27 #undef NDB_ERR
28 #define NDB_ERR(error) \
29 { \
30 const NdbError &_error= (error); \
31 if (!m_quiet) NDB_ERR_OUT(g_err, _error); \
32 }
33
startTransaction(Ndb * pNdb,const NdbDictionary::Table * table,const char * keyData,Uint32 keyLen)34 int HugoOperations::startTransaction(Ndb* pNdb,
35 const NdbDictionary::Table *table,
36 const char *keyData, Uint32 keyLen){
37
38 if (pTrans != NULL){
39 ndbout << "HugoOperations::startTransaction, pTrans != NULL" << endl;
40 return NDBT_FAILED;
41 }
42 pTrans = pNdb->startTransaction(table, keyData, keyLen);
43 if (pTrans == NULL) {
44 const NdbError err = pNdb->getNdbError();
45 NDB_ERR(err);
46 setNdbError(err);
47 return NDBT_FAILED;
48 }
49 return NDBT_OK;
50 }
51
startTransaction(Ndb * pNdb,Uint32 node_id,Uint32 instance_id)52 int HugoOperations::startTransaction(Ndb* pNdb,
53 Uint32 node_id,
54 Uint32 instance_id)
55 {
56 if (pTrans != NULL)
57 {
58 ndbout << "HugoOperations::startTransaction, pTrans != NULL" << endl;
59 return NDBT_FAILED;
60 }
61 pTrans = pNdb->startTransaction(node_id, instance_id);
62 if (pTrans == NULL) {
63 const NdbError err = pNdb->getNdbError();
64 NDB_ERR(err);
65 setNdbError(err);
66 return NDBT_FAILED;
67 }
68 return NDBT_OK;
69 }
70
setTransaction(NdbTransaction * new_trans,bool not_null_ok)71 int HugoOperations::setTransaction(NdbTransaction* new_trans, bool not_null_ok){
72
73 if (pTrans != NULL && !not_null_ok){
74 ndbout << "HugoOperations::setTransaction, pTrans != NULL" << endl;
75 return NDBT_FAILED;
76 }
77 pTrans = new_trans;
78 if (pTrans == NULL) {
79 ndbout << "HugoOperations::setTransaction, pTrans == NULL" << endl;
80 return NDBT_FAILED;
81 }
82 return NDBT_OK;
83 }
84
85 void
setTransactionId(Uint64 id)86 HugoOperations::setTransactionId(Uint64 id){
87 if (pTrans != NULL){
88 pTrans->setTransactionId(id);
89 }
90 }
91
closeTransaction(Ndb * pNdb)92 int HugoOperations::closeTransaction(Ndb* pNdb){
93
94 UtilTransactions::closeTransaction(pNdb);
95
96 m_result_sets.clear();
97 m_executed_result_sets.clear();
98
99 pTrans = NULL;
100 return NDBT_OK;
101 }
102
getTransaction()103 NdbConnection* HugoOperations::getTransaction(){
104 return pTrans;
105 }
106
pkReadRecord(Ndb * pNdb,int recordNo,int numRecords,NdbOperation::LockMode lm,NdbOperation::LockMode * lmused)107 int HugoOperations::pkReadRecord(Ndb* pNdb,
108 int recordNo,
109 int numRecords,
110 NdbOperation::LockMode lm,
111 NdbOperation::LockMode *lmused){
112 int a;
113 allocRows(numRecords);
114 indexScans.clear();
115 int check;
116
117 NdbOperation* pOp = 0;
118 pIndexScanOp = 0;
119
120 for(int r=0; r < numRecords; r++){
121
122 if(pOp == 0)
123 {
124 pOp = getOperation(pTrans, NdbOperation::ReadRequest);
125 }
126 if (pOp == NULL) {
127 NDB_ERR(pTrans->getNdbError());
128 setNdbError(pTrans->getNdbError());
129 return NDBT_FAILED;
130 }
131
132 rand_lock_mode:
133 switch(lm){
134 case NdbOperation::LM_Read:
135 case NdbOperation::LM_Exclusive:
136 case NdbOperation::LM_CommittedRead:
137 case NdbOperation::LM_SimpleRead:
138 if (lmused)
139 * lmused = lm;
140 if(idx && idx->getType() == NdbDictionary::Index::OrderedIndex)
141 {
142 if (pIndexScanOp == 0)
143 {
144 pIndexScanOp = ((NdbIndexScanOperation*)pOp);
145 bool mrrScan= (numRecords > 1);
146 Uint32 flags= mrrScan? NdbScanOperation::SF_MultiRange : 0;
147 check = pIndexScanOp->readTuples(lm, flags);
148 /* Record NdbIndexScanOperation ptr for later... */
149 indexScans.push_back(pIndexScanOp);
150 }
151 }
152 else
153 check = pOp->readTuple(lm);
154 break;
155 default:
156 lm = (NdbOperation::LockMode)((rand() >> 16) & 3);
157 goto rand_lock_mode;
158 }
159
160 if( check == -1 ) {
161 NDB_ERR(pTrans->getNdbError());
162 setNdbError(pTrans->getNdbError());
163 return NDBT_FAILED;
164 }
165
166 // Define primary keys
167 if (equalForRow(pOp, r+recordNo) != 0)
168 {
169 g_err << __LINE__ << " equal for row failed" << endl;
170 return NDBT_FAILED;
171 }
172
173 Uint32 partId;
174 /* Do we need to set the partitionId for this operation? */
175 if (getPartIdForRow(pOp, r+recordNo, partId))
176 {
177 g_info << "Setting operation partition Id" << endl;
178 pOp->setPartitionId(partId);
179 }
180
181 if(pIndexScanOp)
182 pIndexScanOp->end_of_bound(r);
183
184 if(r == 0 || pIndexScanOp == 0)
185 {
186 // Define attributes to read
187 for(a = 0; a<tab.getNoOfColumns(); a++){
188 if((rows[r]->attributeStore(a) =
189 pOp->getValue(tab.getColumn(a))) == 0) {
190 NDB_ERR(pTrans->getNdbError());
191 setNdbError(pTrans->getNdbError());
192 return NDBT_FAILED;
193 }
194 }
195 }
196 /* Note pIndexScanOp will point to the 'last' index scan op
197 * we used. The full list is in the indexScans vector
198 */
199 pOp = pIndexScanOp;
200 }
201 return NDBT_OK;
202 }
203
pkReadRandRecord(Ndb * pNdb,int records,int numRecords,NdbOperation::LockMode lm,NdbOperation::LockMode * lmused)204 int HugoOperations::pkReadRandRecord(Ndb* pNdb,
205 int records,
206 int numRecords,
207 NdbOperation::LockMode lm,
208 NdbOperation::LockMode *lmused){
209 int a;
210 allocRows(numRecords);
211 indexScans.clear();
212 int check;
213
214 NdbOperation* pOp = 0;
215 pIndexScanOp = 0;
216
217 for(int r=0; r < numRecords; r++){
218
219 if(pOp == 0)
220 {
221 pOp = getOperation(pTrans, NdbOperation::ReadRequest);
222 }
223 if (pOp == NULL) {
224 NDB_ERR(pTrans->getNdbError());
225 setNdbError(pTrans->getNdbError());
226 return NDBT_FAILED;
227 }
228
229 rand_lock_mode:
230 switch(lm){
231 case NdbOperation::LM_Read:
232 case NdbOperation::LM_Exclusive:
233 case NdbOperation::LM_CommittedRead:
234 case NdbOperation::LM_SimpleRead:
235 if (lmused)
236 * lmused = lm;
237 if(idx && idx->getType() == NdbDictionary::Index::OrderedIndex &&
238 pIndexScanOp == 0)
239 {
240 pIndexScanOp = ((NdbIndexScanOperation*)pOp);
241 check = pIndexScanOp->readTuples(lm);
242 /* Record NdbIndexScanOperation ptr for later... */
243 indexScans.push_back(pIndexScanOp);
244 }
245 else
246 check = pOp->readTuple(lm);
247 break;
248 default:
249 lm = (NdbOperation::LockMode)((rand() >> 16) & 3);
250 goto rand_lock_mode;
251 }
252
253 if( check == -1 ) {
254 NDB_ERR(pTrans->getNdbError());
255 setNdbError(pTrans->getNdbError());
256 return NDBT_FAILED;
257 }
258
259 int rowid= rand() % records;
260
261 // Define primary keys
262 if (equalForRow(pOp, rowid) != 0)
263 {
264 g_err << __LINE__ << " equal for row failed" << endl;
265 return NDBT_FAILED;
266 }
267
268 Uint32 partId;
269 /* Do we need to set the partitionId for this operation? */
270 if (getPartIdForRow(pOp, rowid, partId))
271 {
272 g_info << "Setting operation partition Id" << endl;
273 pOp->setPartitionId(partId);
274 }
275
276 if(pIndexScanOp)
277 pIndexScanOp->end_of_bound(r);
278
279 if(r == 0 || pIndexScanOp == 0)
280 {
281 // Define attributes to read
282 for(a = 0; a<tab.getNoOfColumns(); a++){
283 if((rows[r]->attributeStore(a) =
284 pOp->getValue(tab.getColumn(a))) == 0) {
285 NDB_ERR(pTrans->getNdbError());
286 setNdbError(pTrans->getNdbError());
287 return NDBT_FAILED;
288 }
289 }
290 }
291 /* Note pIndexScanOp will point to the 'last' index scan op
292 * we used. The full list is in the indexScans vector
293 */
294 pOp = pIndexScanOp;
295 }
296 return NDBT_OK;
297 }
298
pkReadRecordLockHandle(Ndb * pNdb,Vector<const NdbLockHandle * > & lockHandles,int recordNo,int numRecords,NdbOperation::LockMode lm,NdbOperation::LockMode * lmused)299 int HugoOperations::pkReadRecordLockHandle(Ndb* pNdb,
300 Vector<const NdbLockHandle*>& lockHandles,
301 int recordNo,
302 int numRecords,
303 NdbOperation::LockMode lm,
304 NdbOperation::LockMode *lmused){
305 if (idx)
306 {
307 g_err << "ERROR : Cannot call pkReadRecordLockHandle on an index"
308 << endl;
309 return NDBT_FAILED;
310 }
311
312 /* If something other than LM_Read or LM_Exclusive is
313 * passed in then we'll choose, and PkReadRecord
314 * will update lm_used
315 */
316 while (lm != NdbOperation::LM_Read &&
317 lm != NdbOperation::LM_Exclusive)
318 {
319 lm = (NdbOperation::LockMode)((rand() >> 16) & 1);
320 }
321
322 const NdbOperation* prevOp = pTrans->getLastDefinedOperation();
323
324 int readRc = pkReadRecord(pNdb,
325 recordNo,
326 numRecords,
327 lm,
328 lmused);
329
330 if (readRc == NDBT_OK)
331 {
332 /* Now traverse operations added, requesting
333 * LockHandles
334 */
335 const NdbOperation* definedOp = (prevOp)? prevOp->next() :
336 pTrans->getFirstDefinedOperation();
337
338 while (definedOp)
339 {
340 /* Look away now */
341 const NdbLockHandle* lh =
342 (const_cast<NdbOperation*>(definedOp))->getLockHandle();
343
344 if (lh == NULL)
345 {
346 NDB_ERR(definedOp->getNdbError());
347 setNdbError(definedOp->getNdbError());
348 return NDBT_FAILED;
349 }
350
351 lockHandles.push_back(lh);
352 definedOp = definedOp->next();
353 }
354 }
355
356 return readRc;
357 }
358
pkUnlockRecord(Ndb * pNdb,Vector<const NdbLockHandle * > & lockHandles,int offset,int numRecords,NdbOperation::AbortOption ao)359 int HugoOperations::pkUnlockRecord(Ndb* pNdb,
360 Vector<const NdbLockHandle*>& lockHandles,
361 int offset,
362 int numRecords,
363 NdbOperation::AbortOption ao)
364 {
365 if (numRecords == ~(0))
366 {
367 numRecords = lockHandles.size() - offset;
368 }
369
370 if (lockHandles.size() < (unsigned)(offset + numRecords))
371 {
372 g_err << "ERROR : LockHandles size is "
373 << lockHandles.size()
374 << " offset ("
375 << offset
376 << ") and/or numRecords ("
377 << numRecords
378 << ") too large." << endl;
379 return NDBT_FAILED;
380 }
381
382 for (int i = 0; i < numRecords; i++)
383 {
384 const NdbLockHandle* lh = lockHandles[offset + i];
385 if (lh != NULL)
386 {
387 const NdbOperation* unlockOp = pTrans->unlock(lh, ao);
388
389 if (unlockOp == NULL)
390 {
391 NDB_ERR(pTrans->getNdbError());
392 setNdbError(pTrans->getNdbError());
393 return NDBT_FAILED;
394 }
395 }
396 else
397 {
398 g_err << "ERROR : LockHandle number "
399 << i+offset << " is NULL. "
400 << " offset is " << offset << endl;
401 return NDBT_FAILED;
402 }
403 }
404
405 return NDBT_OK;
406 }
407
pkUpdateRecord(Ndb * pNdb,int recordNo,int numRecords,int updatesValue)408 int HugoOperations::pkUpdateRecord(Ndb* pNdb,
409 int recordNo,
410 int numRecords,
411 int updatesValue){
412 allocRows(numRecords);
413 int check;
414 for(int r=0; r < numRecords; r++){
415 NdbOperation* pOp = getOperation(pTrans, NdbOperation::UpdateRequest);
416 if (pOp == NULL) {
417 NDB_ERR(pTrans->getNdbError());
418 setNdbError(pTrans->getNdbError());
419 return NDBT_FAILED;
420 }
421
422 check = pOp->updateTuple();
423 if( check == -1 ) {
424 NDB_ERR(pTrans->getNdbError());
425 setNdbError(pTrans->getNdbError());
426 return NDBT_FAILED;
427 }
428
429 if(setValues(pOp, r+recordNo, updatesValue) != NDBT_OK)
430 {
431 g_err << __LINE__ << " setValues failed" << endl;
432 return NDBT_FAILED;
433 }
434
435 Uint32 partId;
436 if(getPartIdForRow(pOp, r+recordNo, partId))
437 pOp->setPartitionId(partId);
438
439 pOp->setAnyValue(getAnyValueForRowUpd(r+recordNo, updatesValue));
440
441 }
442 return NDBT_OK;
443 }
444
445 int
setValues(NdbOperation * pOp,int rowId,int updateId)446 HugoOperations::setValues(NdbOperation* pOp, int rowId, int updateId)
447 {
448 // Define primary keys
449 if (equalForRow(pOp, rowId) != 0)
450 {
451 g_err << __LINE__ << " equal for row failed" << endl;
452 return NDBT_FAILED;
453 }
454
455 if (setNonPkValues(pOp, rowId, updateId) != 0)
456 {
457 g_err << __LINE__ << " setNonPkValues failed" << endl;
458 return NDBT_FAILED;
459 }
460
461 return NDBT_OK;
462 }
463
464 int
setNonPkValues(NdbOperation * pOp,int rowId,int updateId)465 HugoOperations::setNonPkValues(NdbOperation* pOp, int rowId, int updateId)
466 {
467 for(int a = 0; a<tab.getNoOfColumns(); a++)
468 {
469 if (tab.getColumn(a)->getPrimaryKey() == false)
470 {
471 if(setValueForAttr(pOp, a, rowId, updateId ) != 0)
472 {
473 NDB_ERR(pTrans->getNdbError());
474 setNdbError(pTrans->getNdbError());
475 return NDBT_FAILED;
476 }
477 }
478 }
479 return NDBT_OK;
480 }
481
pkInsertRecord(Ndb * pNdb,int recordNo,int numRecords,int updatesValue)482 int HugoOperations::pkInsertRecord(Ndb* pNdb,
483 int recordNo,
484 int numRecords,
485 int updatesValue){
486
487 int check;
488 for(int r=0; r < numRecords; r++){
489 NdbOperation* pOp = getOperation(pTrans, NdbOperation::InsertRequest);
490 if (pOp == NULL) {
491 NDB_ERR(pTrans->getNdbError());
492 setNdbError(pTrans->getNdbError());
493 return NDBT_FAILED;
494 }
495
496 check = pOp->insertTuple();
497 if( check == -1 ) {
498 NDB_ERR(pTrans->getNdbError());
499 setNdbError(pTrans->getNdbError());
500 return NDBT_FAILED;
501 }
502
503 if(setValues(pOp, r+recordNo, updatesValue) != NDBT_OK)
504 {
505 m_error.code = pTrans->getNdbError().code;
506 g_err << __LINE__ << " setValues failed" << endl;
507 return NDBT_FAILED;
508 }
509
510 Uint32 partId;
511 if(getPartIdForRow(pOp, r+recordNo, partId))
512 pOp->setPartitionId(partId);
513
514 }
515 return NDBT_OK;
516 }
517
pkWriteRecord(Ndb * pNdb,int recordNo,int numRecords,int updatesValue)518 int HugoOperations::pkWriteRecord(Ndb* pNdb,
519 int recordNo,
520 int numRecords,
521 int updatesValue){
522 int a, check;
523 for(int r=0; r < numRecords; r++){
524 NdbOperation* pOp = pTrans->getNdbOperation(tab.getName());
525 if (pOp == NULL) {
526 NDB_ERR(pTrans->getNdbError());
527 setNdbError(pTrans->getNdbError());
528 return NDBT_FAILED;
529 }
530
531 check = pOp->writeTuple();
532 if( check == -1 ) {
533 NDB_ERR(pTrans->getNdbError());
534 setNdbError(pTrans->getNdbError());
535 return NDBT_FAILED;
536 }
537
538 // Define primary keys
539 if (equalForRow(pOp, r+recordNo) != 0)
540 {
541 g_err << __LINE__ << " equal for row failed" << endl;
542 return NDBT_FAILED;
543 }
544
545 Uint32 partId;
546 if(getPartIdForRow(pOp, r+recordNo, partId))
547 pOp->setPartitionId(partId);
548
549
550 // Define attributes to update
551 for(a = 0; a<tab.getNoOfColumns(); a++){
552 if (tab.getColumn(a)->getPrimaryKey() == false){
553 if(setValueForAttr(pOp, a, recordNo+r, updatesValue ) != 0){
554 NDB_ERR(pTrans->getNdbError());
555 setNdbError(pTrans->getNdbError());
556 return NDBT_FAILED;
557 }
558 }
559 }
560 }
561 return NDBT_OK;
562 }
563
pkWritePartialRecord(Ndb * pNdb,int recordNo,int numRecords)564 int HugoOperations::pkWritePartialRecord(Ndb* pNdb,
565 int recordNo,
566 int numRecords){
567
568 int check;
569 for(int r=0; r < numRecords; r++){
570 NdbOperation* pOp = pTrans->getNdbOperation(tab.getName());
571 if (pOp == NULL) {
572 NDB_ERR(pTrans->getNdbError());
573 setNdbError(pTrans->getNdbError());
574 return NDBT_FAILED;
575 }
576
577 check = pOp->writeTuple();
578 if( check == -1 ) {
579 NDB_ERR(pTrans->getNdbError());
580 setNdbError(pTrans->getNdbError());
581 return NDBT_FAILED;
582 }
583
584 // Define primary keys
585 if (equalForRow(pOp, r+recordNo) != 0)
586 {
587 g_err << __LINE__ << " equal for row failed" << endl;
588 return NDBT_FAILED;
589 }
590
591 Uint32 partId;
592 if(getPartIdForRow(pOp, r+recordNo, partId))
593 pOp->setPartitionId(partId);
594
595 }
596 return NDBT_OK;
597 }
598
pkDeleteRecord(Ndb * pNdb,int recordNo,int numRecords)599 int HugoOperations::pkDeleteRecord(Ndb* pNdb,
600 int recordNo,
601 int numRecords){
602
603 int check;
604 for(int r=0; r < numRecords; r++){
605 NdbOperation* pOp = getOperation(pTrans, NdbOperation::DeleteRequest);
606 if (pOp == NULL) {
607 NDB_ERR(pTrans->getNdbError());
608 setNdbError(pTrans->getNdbError());
609 return NDBT_FAILED;
610 }
611
612 check = pOp->deleteTuple();
613 if( check == -1 ) {
614 NDB_ERR(pTrans->getNdbError());
615 setNdbError(pTrans->getNdbError());
616 return NDBT_FAILED;
617 }
618
619 // Define primary keys
620 if (equalForRow(pOp, r+recordNo) != 0)
621 {
622 g_err << __LINE__ << " equal for row failed" << endl;
623 return NDBT_FAILED;
624 }
625
626 Uint32 partId;
627 if(getPartIdForRow(pOp, r+recordNo, partId))
628 pOp->setPartitionId(partId);
629 }
630 return NDBT_OK;
631 }
632
pkRefreshRecord(Ndb * pNdb,int recordNo,int numRecords,int anyValueInfo)633 int HugoOperations::pkRefreshRecord(Ndb* pNdb,
634 int recordNo,
635 int numRecords,
636 int anyValueInfo){
637
638 char buffer[NDB_MAX_TUPLE_SIZE];
639 const NdbDictionary::Table * pTab =
640 pNdb->getDictionary()->getTable(tab.getName());
641
642 if (pTab == 0)
643 {
644 g_err << __LINE__ << " pTab == 0" << endl;
645 return NDBT_FAILED;
646 }
647
648 const NdbRecord * record = pTab->getDefaultRecord();
649 NdbOperation::OperationOptions opts;
650 opts.optionsPresent = NdbOperation::OperationOptions::OO_ANYVALUE;
651 for(int r=0; r < numRecords; r++)
652 {
653 bzero(buffer, sizeof(buffer));
654 if (calc.equalForRow((Uint8*)buffer, record, r + recordNo))
655 {
656 g_err << __LINE__ << " equal for row failed" << endl;
657 return NDBT_FAILED;
658 }
659
660 opts.anyValue = anyValueInfo?
661 (anyValueInfo << 16) | (r+recordNo) :
662 0;
663
664 const NdbOperation* pOp = pTrans->refreshTuple(record, buffer,
665 &opts, sizeof(opts));
666 if (pOp == NULL)
667 {
668 NDB_ERR(pTrans->getNdbError());
669 setNdbError(pTrans->getNdbError());
670 return NDBT_FAILED;
671 }
672 }
673 return NDBT_OK;
674 }
675
execute_Commit(Ndb * pNdb,AbortOption eao)676 int HugoOperations::execute_Commit(Ndb* pNdb,
677 AbortOption eao){
678
679 int check = 0;
680 check = pTrans->execute(Commit, eao);
681
682 const NdbError err = pTrans->getNdbError();
683 if( check == -1 || err.code) {
684 NDB_ERR(err);
685 setNdbError(err);
686 NdbOperation* pOp = pTrans->getNdbErrorOperation();
687 if (pOp != NULL){
688 const NdbError err2 = pOp->getNdbError();
689 NDB_ERR(err2);
690 setNdbError(err2);
691 }
692 if (err.code == 0)
693 {
694 g_err << __LINE__ << " execute_Commit failed with errcode = 0" << endl;
695 return NDBT_FAILED;
696 }
697 return err.code;
698 }
699
700 for(unsigned int i = 0; i<m_result_sets.size(); i++){
701 m_executed_result_sets.push_back(m_result_sets[i]);
702
703 int rows = m_result_sets[i].records;
704 NdbScanOperation* rs = m_result_sets[i].m_result_set;
705 int res = rs->nextResult();
706 switch(res){
707 case 1:
708 return 626;
709 case -1:
710 const NdbError err = pTrans->getNdbError();
711 NDB_ERR(err);
712 setNdbError(err);
713 return (err.code > 0 ? err.code : NDBT_FAILED);
714 }
715
716 // A row found
717
718 switch(rows){
719 case 0:
720 return 4000;
721 default:
722 m_result_sets[i].records--;
723 break;
724 }
725 }
726
727 m_result_sets.clear();
728
729 return NDBT_OK;
730 }
731
execute_NoCommit(Ndb * pNdb,AbortOption eao)732 int HugoOperations::execute_NoCommit(Ndb* pNdb, AbortOption eao){
733
734 int check;
735 check = pTrans->execute(NoCommit, eao);
736
737 const NdbError err = pTrans->getNdbError();
738 if( check == -1 || err.code) {
739 NDB_ERR(err);
740 setNdbError(err);
741 const NdbOperation* pOp = pTrans->getNdbErrorOperation();
742 while (pOp != NULL)
743 {
744 const NdbError err2 = pOp->getNdbError();
745 if (err2.code)
746 {
747 NDB_ERR(err2);
748 setNdbError(err2);
749 }
750 pOp = pTrans->getNextCompletedOperation(pOp);
751 }
752 if (err.code == 0)
753 {
754 g_err << __LINE__ << " equal for row failed" << endl;
755 return NDBT_FAILED;
756 }
757 return err.code;
758 }
759
760 for(unsigned int i = 0; i<m_result_sets.size(); i++){
761 m_executed_result_sets.push_back(m_result_sets[i]);
762
763 int rows = m_result_sets[i].records;
764 NdbScanOperation* rs = m_result_sets[i].m_result_set;
765 int res = rs->nextResult();
766 switch(res){
767 case 1:
768 return 626;
769 case -1:
770 const NdbError err = pTrans->getNdbError();
771 NDB_ERR(err);
772 setNdbError(err);
773 return (err.code > 0 ? err.code : NDBT_FAILED);
774 }
775
776 // A row found
777
778 switch(rows){
779 case 0:
780 return 4000;
781 default:
782 case 1:
783 break;
784 }
785 }
786
787 m_result_sets.clear();
788
789 return NDBT_OK;
790 }
791
execute_Rollback(Ndb * pNdb)792 int HugoOperations::execute_Rollback(Ndb* pNdb){
793 int check;
794 check = pTrans->execute(Rollback);
795 if( check == -1 ) {
796 const NdbError err = pTrans->getNdbError();
797 NDB_ERR(err);
798 setNdbError(err);
799 return NDBT_FAILED;
800 }
801 return NDBT_OK;
802 }
803
804 void
HugoOperations_async_callback(int res,NdbTransaction * pCon,void * ho)805 HugoOperations_async_callback(int res, NdbTransaction* pCon, void* ho)
806 {
807 ((HugoOperations*)ho)->callback(res, pCon);
808 }
809
810 void
callback(int res,NdbTransaction * pCon)811 HugoOperations::callback(int res, NdbTransaction* pCon)
812 {
813 require(pCon == pTrans);
814 m_async_reply= 1;
815 if(res)
816 {
817 m_async_return = pCon->getNdbError().code;
818 }
819 else
820 {
821 m_async_return = 0;
822 }
823 }
824
825 int
execute_async(Ndb * pNdb,NdbTransaction::ExecType et,NdbOperation::AbortOption eao)826 HugoOperations::execute_async(Ndb* pNdb, NdbTransaction::ExecType et,
827 NdbOperation::AbortOption eao){
828
829 m_async_reply= 0;
830 pTrans->executeAsynchPrepare(et,
831 HugoOperations_async_callback,
832 this,
833 eao);
834
835 pNdb->sendPreparedTransactions();
836
837 return NDBT_OK;
838 }
839
840 int
execute_async_prepare(Ndb * pNdb,NdbTransaction::ExecType et,NdbOperation::AbortOption eao)841 HugoOperations::execute_async_prepare(Ndb* pNdb, NdbTransaction::ExecType et,
842 NdbOperation::AbortOption eao){
843
844 m_async_reply= 0;
845 pTrans->executeAsynchPrepare(et,
846 HugoOperations_async_callback,
847 this,
848 eao);
849
850 return NDBT_OK;
851 }
852
853 int
wait_async(Ndb * pNdb,int timeout)854 HugoOperations::wait_async(Ndb* pNdb, int timeout)
855 {
856 volatile int * wait = &m_async_reply;
857 while (!* wait)
858 {
859 pNdb->sendPollNdb(1000);
860
861 if(* wait)
862 {
863 if(m_async_return)
864 ndbout << "ERROR: " << pNdb->getNdbError(m_async_return) << endl;
865 return m_async_return;
866 }
867 }
868 ndbout_c("wait returned nothing...");
869 return -1;
870 }
871
HugoOperations(const NdbDictionary::Table & _tab,const NdbDictionary::Index * idx)872 HugoOperations::HugoOperations(const NdbDictionary::Table& _tab,
873 const NdbDictionary::Index* idx):
874 UtilTransactions(_tab, idx),
875 pIndexScanOp(NULL),
876 calc(_tab),
877 m_quiet(false),
878 avCallback(NULL)
879 {
880 }
881
~HugoOperations()882 HugoOperations::~HugoOperations(){
883 deallocRows();
884 if (pTrans != NULL)
885 {
886 pTrans->close();
887 pTrans = NULL;
888 }
889 }
890
891 int
equalForRow(NdbOperation * pOp,int row)892 HugoOperations::equalForRow(NdbOperation* pOp, int row)
893 {
894 for(int a = 0; a<tab.getNoOfColumns(); a++)
895 {
896 if (tab.getColumn(a)->getPrimaryKey() == true)
897 {
898 if(equalForAttr(pOp, a, row) != 0)
899 {
900 NDB_ERR(pOp->getNdbError());
901 setNdbError(pOp->getNdbError());
902 return NDBT_FAILED;
903 }
904 }
905 }
906 return NDBT_OK;
907 }
908
getPartIdForRow(const NdbOperation * pOp,int rowid,Uint32 & partId)909 bool HugoOperations::getPartIdForRow(const NdbOperation* pOp,
910 int rowid,
911 Uint32& partId)
912 {
913 if (tab.getFragmentType() == NdbDictionary::Object::UserDefined)
914 {
915 /* Primary keys and Ordered indexes are partitioned according
916 * to the row number
917 * PartitionId must be set for PK access. Ordered indexes
918 * can scan all partitions.
919 */
920 if (pOp->getType() == NdbOperation::PrimaryKeyAccess)
921 {
922 /* Need to set the partitionId for this op
923 * For Hugo, we use 'HASH' partitioning, which is probably
924 * better called 'MODULO' partitioning with
925 * FragId == rowNum % NumPartitions
926 * This gives a good balance with the normal Hugo data, but different
927 * row to partition assignments than normal key partitioning.
928 */
929 const Uint32 numFrags= tab.getFragmentCount();
930 partId= rowid % numFrags;
931 g_info << "Returning partition Id of " << partId << endl;
932 return true;
933 }
934 }
935 partId= ~0;
936 return false;
937 }
938
equalForAttr(NdbOperation * pOp,int attrId,int rowId)939 int HugoOperations::equalForAttr(NdbOperation* pOp,
940 int attrId,
941 int rowId){
942 const NdbDictionary::Column* attr = tab.getColumn(attrId);
943 if (attr->getPrimaryKey() == false){
944 g_err << "Can't call equalForAttr on non PK attribute" << endl;
945 return NDBT_FAILED;
946 }
947
948 int len = attr->getSizeInBytes();
949 char buf[NDB_MAX_TUPLE_SIZE];
950 memset(buf, 0, sizeof(buf));
951 Uint32 real_len;
952 const char * value = calc.calcValue(rowId, attrId, 0, buf, len, &real_len);
953 return pOp->equal( attr->getName(), value, real_len);
954 }
955
setValueForAttr(NdbOperation * pOp,int attrId,int rowId,int updateId)956 int HugoOperations::setValueForAttr(NdbOperation* pOp,
957 int attrId,
958 int rowId,
959 int updateId){
960 const NdbDictionary::Column* attr = tab.getColumn(attrId);
961
962 if (! (attr->getType() == NdbDictionary::Column::Blob))
963 {
964 int len = attr->getSizeInBytes();
965 char buf[NDB_MAX_TUPLE_SIZE];
966 memset(buf, 0, sizeof(buf));
967 Uint32 real_len;
968 const char * value = calc.calcValue(rowId, attrId,
969 updateId, buf, len, &real_len);
970 return pOp->setValue( attr->getName(), value, real_len);
971 }
972 else
973 {
974 char buf[32000];
975 int len = (int)sizeof(buf);
976 Uint32 real_len;
977 const char * value = calc.calcValue(rowId, attrId,
978 updateId, buf, len, &real_len);
979 NdbBlob * b = pOp->getBlobHandle(attrId);
980 if (b == 0)
981 return -1;
982
983 if (real_len == 0)
984 return b->setNull();
985 else
986 return b->setValue(value, real_len);
987 }
988 }
989
990 int
verifyUpdatesValue(int updatesValue,int _numRows)991 HugoOperations::verifyUpdatesValue(int updatesValue, int _numRows){
992 _numRows = (_numRows == 0 ? rows.size() : _numRows);
993
994 int result = NDBT_OK;
995
996 for(int i = 0; i<_numRows; i++){
997 if(calc.verifyRowValues(rows[i]) != NDBT_OK){
998 g_err << "Inconsistent row"
999 << endl << "\t" << rows[i]->c_str().c_str() << endl;
1000 result = NDBT_FAILED;
1001 continue;
1002 }
1003
1004 if(calc.getUpdatesValue(rows[i]) != updatesValue){
1005 result = NDBT_FAILED;
1006 g_err << "Invalid updates value for row " << i << endl
1007 << " updatesValue: " << updatesValue << endl
1008 << " calc.getUpdatesValue: " << calc.getUpdatesValue(rows[i]) << endl
1009 << rows[i]->c_str().c_str() << endl;
1010 continue;
1011 }
1012 }
1013
1014 if(_numRows == 0){
1015 g_err << "No rows -> Invalid updates value" << endl;
1016 return NDBT_FAILED;
1017 }
1018
1019 return result;
1020 }
1021
allocRows(int _numRows)1022 void HugoOperations::allocRows(int _numRows){
1023 if(_numRows <= 0){
1024 g_info << "Illegal value for num rows : " << _numRows << endl;
1025 abort();
1026 }
1027
1028 for(int b=rows.size(); b<_numRows; b++){
1029 rows.push_back(new NDBT_ResultRow(tab));
1030 }
1031 }
1032
deallocRows()1033 void HugoOperations::deallocRows(){
1034 while(rows.size() > 0){
1035 delete rows.back();
1036 rows.erase(rows.size() - 1);
1037 }
1038 }
1039
saveCopyOfRecord(int numRecords)1040 int HugoOperations::saveCopyOfRecord(int numRecords ){
1041
1042 if (numRecords > (int)rows.size())
1043 {
1044 g_err << __LINE__ << " number of rows wrong" << endl;
1045 return NDBT_FAILED;
1046 }
1047
1048 for (int i = 0; i < numRecords; i++){
1049 savedRecords.push_back(rows[i]->c_str());
1050 }
1051 return NDBT_OK;
1052 }
1053
getRecordStr(int recordNum)1054 BaseString HugoOperations::getRecordStr(int recordNum){
1055 if (recordNum > (int)rows.size())
1056 return NULL;
1057 return rows[recordNum]->c_str();
1058 }
1059
getRecordGci(int recordNum)1060 int HugoOperations::getRecordGci(int recordNum){
1061 return pTrans->getGCI();
1062 }
1063
1064
compareRecordToCopy(int numRecords)1065 int HugoOperations::compareRecordToCopy(int numRecords ){
1066 if (numRecords > (int)rows.size())
1067 {
1068 g_err << __LINE__ << " number of rows wrong" << endl;
1069 return NDBT_FAILED;
1070 }
1071 if ((unsigned)numRecords > savedRecords.size())
1072 {
1073 g_err << __LINE__ << " number of rows wrong" << endl;
1074 return NDBT_FAILED;
1075 }
1076
1077 int result = NDBT_OK;
1078 for (int i = 0; i < numRecords; i++){
1079 BaseString str = rows[i]->c_str();
1080 ndbout << "row["<<i<<"]: " << str << endl;
1081 ndbout << "sav["<<i<<"]: " << savedRecords[i] << endl;
1082 if (savedRecords[i] == str){
1083 ;
1084 } else {
1085 g_err << __LINE__ << " row " << i << " wrong" << endl;
1086 result = NDBT_FAILED;
1087 }
1088 }
1089 return result;
1090 }
1091
1092 void
refresh()1093 HugoOperations::refresh() {
1094 NdbTransaction * t = getTransaction();
1095 if(t)
1096 t->refresh();
1097 }
1098
indexReadRecords(Ndb *,const char * idxName,int recordNo,bool exclusive,int numRecords)1099 int HugoOperations::indexReadRecords(Ndb*, const char * idxName, int recordNo,
1100 bool exclusive,
1101 int numRecords){
1102
1103 int a;
1104 allocRows(numRecords);
1105 int check;
1106 for(int r=0; r < numRecords; r++){
1107 NdbOperation* pOp = pTrans->getNdbIndexOperation(idxName, tab.getName());
1108 if (pOp == NULL) {
1109 NDB_ERR(pTrans->getNdbError());
1110 setNdbError(pTrans->getNdbError());
1111 return NDBT_FAILED;
1112 }
1113
1114 if (exclusive == true)
1115 check = pOp->readTupleExclusive();
1116 else
1117 check = pOp->readTuple();
1118 if( check == -1 ) {
1119 NDB_ERR(pTrans->getNdbError());
1120 setNdbError(pTrans->getNdbError());
1121 return NDBT_FAILED;
1122 }
1123
1124 // Define primary keys
1125 if (equalForRow(pOp, r+recordNo) != 0)
1126 {
1127 g_err << __LINE__ << " equal for row failed" << endl;
1128 return NDBT_FAILED;
1129 }
1130
1131 // Define attributes to read
1132 for(a = 0; a<tab.getNoOfColumns(); a++){
1133 if((rows[r]->attributeStore(a) =
1134 pOp->getValue(tab.getColumn(a))) == 0) {
1135 NDB_ERR(pTrans->getNdbError());
1136 setNdbError(pTrans->getNdbError());
1137 return NDBT_FAILED;
1138 }
1139 }
1140 }
1141 return NDBT_OK;
1142 }
1143
1144 int
indexUpdateRecord(Ndb *,const char * idxName,int recordNo,int numRecords,int updatesValue)1145 HugoOperations::indexUpdateRecord(Ndb*,
1146 const char * idxName,
1147 int recordNo,
1148 int numRecords,
1149 int updatesValue){
1150 int a;
1151 allocRows(numRecords);
1152 int check;
1153 for(int r=0; r < numRecords; r++){
1154 NdbOperation* pOp = pTrans->getNdbIndexOperation(idxName, tab.getName());
1155 if (pOp == NULL) {
1156 NDB_ERR(pTrans->getNdbError());
1157 setNdbError(pTrans->getNdbError());
1158 return NDBT_FAILED;
1159 }
1160
1161 check = pOp->updateTuple();
1162 if( check == -1 ) {
1163 NDB_ERR(pTrans->getNdbError());
1164 setNdbError(pTrans->getNdbError());
1165 return NDBT_FAILED;
1166 }
1167
1168 // Define primary keys
1169 if (equalForRow(pOp, r+recordNo) != 0)
1170 {
1171 g_err << __LINE__ << " equal for row failed" << endl;
1172 return NDBT_FAILED;
1173 }
1174
1175 // Define attributes to update
1176 for(a = 0; a<tab.getNoOfColumns(); a++){
1177 if (tab.getColumn(a)->getPrimaryKey() == false){
1178 if(setValueForAttr(pOp, a, recordNo+r, updatesValue ) != 0){
1179 NDB_ERR(pTrans->getNdbError());
1180 setNdbError(pTrans->getNdbError());
1181 return NDBT_FAILED;
1182 }
1183 }
1184 }
1185 }
1186 return NDBT_OK;
1187 }
1188
1189 int
scanReadRecords(Ndb * pNdb,NdbScanOperation::LockMode lm,int records)1190 HugoOperations::scanReadRecords(Ndb* pNdb, NdbScanOperation::LockMode lm,
1191 int records){
1192
1193 allocRows(records);
1194 NdbScanOperation * pOp = pTrans->getNdbScanOperation(tab.getName());
1195
1196 if(!pOp)
1197 return -1;
1198
1199 if(pOp->readTuples(lm, 0, 1)){
1200 return -1;
1201 }
1202
1203 for(int a = 0; a<tab.getNoOfColumns(); a++){
1204 if((rows[0]->attributeStore(a) =
1205 pOp->getValue(tab.getColumn(a))) == 0) {
1206 NDB_ERR(pTrans->getNdbError());
1207 setNdbError(pTrans->getNdbError());
1208 return NDBT_FAILED;
1209 }
1210 }
1211
1212 RsPair p = {pOp, records};
1213 m_result_sets.push_back(p);
1214
1215 return 0;
1216 }
1217
1218 int
releaseLockHandles(Ndb * pNdb,Vector<const NdbLockHandle * > & lockHandles,int offset,int numRecords)1219 HugoOperations::releaseLockHandles(Ndb* pNdb,
1220 Vector<const NdbLockHandle*>& lockHandles,
1221 int offset,
1222 int numRecords)
1223 {
1224 int totalSize = lockHandles.size();
1225 if (numRecords == ~(0))
1226 {
1227 numRecords = totalSize - offset;
1228 }
1229
1230 if (totalSize < (offset + numRecords))
1231 {
1232 g_err << "ERROR : LockHandles size is "
1233 << lockHandles.size()
1234 << " offset ("
1235 << offset
1236 << ") and/or numRecords ("
1237 << numRecords
1238 << ") too large." << endl;
1239 return NDBT_FAILED;
1240 }
1241
1242 for (int i = 0; i < numRecords; i++)
1243 {
1244 const NdbLockHandle* lh = lockHandles[offset + i];
1245 if (lh != NULL)
1246 {
1247 if (pTrans->releaseLockHandle(lh) != 0)
1248 {
1249 NDB_ERR(pTrans->getNdbError());
1250 setNdbError(pTrans->getNdbError());
1251 return NDBT_FAILED;
1252 }
1253 const NdbLockHandle* nullPtr = NULL;
1254 (void)nullPtr; // ??
1255 //lockHandles.set(nullPtr, offset + i, nullPtr);
1256 }
1257 else
1258 {
1259 g_err << "ERROR : LockHandle number "
1260 << i+offset << " is NULL. "
1261 << " offset is " << offset << endl;
1262 return NDBT_FAILED;
1263 }
1264 }
1265
1266 return NDBT_OK;
1267
1268 }
1269
1270 static void
update(const NdbError & _err)1271 update(const NdbError & _err)
1272 {
1273 NdbError & error = (NdbError &) _err;
1274 ndberror_struct ndberror = (ndberror_struct)error;
1275 ndberror_update(&ndberror);
1276 error = NdbError(ndberror);
1277 }
1278
1279 const NdbError &
getNdbError() const1280 HugoOperations::getNdbError() const
1281 {
1282 update(m_error);
1283 return m_error;
1284 }
1285
1286 void
setNdbError(const NdbError & error)1287 HugoOperations::setNdbError(const NdbError& error)
1288 {
1289 m_error.code = error.code ? error.code : 1;
1290 }
1291
1292 void
setAnyValueCallback(AnyValueCallback avc)1293 HugoOperations::setAnyValueCallback(AnyValueCallback avc)
1294 {
1295 avCallback = avc;
1296 }
1297
1298 Uint32
getAnyValueForRowUpd(int row,int update)1299 HugoOperations::getAnyValueForRowUpd(int row, int update)
1300 {
1301 if (avCallback == NULL)
1302 return 0;
1303
1304 return (avCallback)(pTrans->getNdb(), pTrans,
1305 row, update);
1306 }
1307
1308 template class Vector<HugoOperations::RsPair>;
1309 template class Vector<const NdbLockHandle*>;
1310