← Back to branch summary

~ubuntu-branches/ubuntu/precise/mysql-5.1/precise

« back to all changes in this revision

Viewing changes to storage/ndb/tools/restore/consumer_restore.cpp

  • Committer: Bazaar Package Importer
  • Author(s): Norbert Tretkowski
  • Date: 2010-03-17 14:56:02 UTC
  • Revision ID: james.westby@ubuntu.com-20100317145602-x7e30l1b2sb5s6w6
Tags: upstream-5.1.45
Import upstream version 5.1.45

Show diffs side-by-side

added added

removed removed

Lines of Context:
storage/ndb/tools/restore/consumer_restore.cpp
 
1
/* Copyright (C) 2003 MySQL AB
 
2
 
 
3
   This program is free software; you can redistribute it and/or modify
 
4
   it under the terms of the GNU General Public License as published by
 
5
   the Free Software Foundation; version 2 of the License.
 
6
 
 
7
   This program is distributed in the hope that it will be useful,
 
8
   but WITHOUT ANY WARRANTY; without even the implied warranty of
 
9
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
10
   GNU General Public License for more details.
 
11
 
 
12
   You should have received a copy of the GNU General Public License
 
13
   along with this program; if not, write to the Free Software
 
14
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
 
15
 
 
16
#include <NDBT_ReturnCodes.h>
 
17
#include "consumer_restore.hpp"
 
18
#include <my_sys.h>
 
19
#include <NdbSleep.h>
 
20
 
 
21
extern my_bool opt_core;
 
22
 
 
23
extern FilteredNdbOut err;
 
24
extern FilteredNdbOut info;
 
25
extern FilteredNdbOut debug;
 
26
 
 
27
static void callback(int, NdbTransaction*, void*);
 
28
static Uint32 get_part_id(const NdbDictionary::Table *table,
 
29
                          Uint32 hash_value);
 
30
 
 
31
extern const char * g_connect_string;
 
32
extern BaseString g_options;
 
33
 
 
34
bool
 
35
BackupRestore::init()
 
36
{
 
37
  release();
 
38
 
 
39
  if (!m_restore && !m_restore_meta && !m_restore_epoch)
 
40
    return true;
 
41
 
 
42
  m_cluster_connection = new Ndb_cluster_connection(g_connect_string);
 
43
  m_cluster_connection->set_name(g_options.c_str());
 
44
  if(m_cluster_connection->connect(12, 5, 1) != 0)
 
45
  {
 
46
    return false;
 
47
  }
 
48
 
 
49
  m_ndb = new Ndb(m_cluster_connection);
 
50
 
 
51
  if (m_ndb == NULL)
 
52
    return false;
 
53
  
 
54
  m_ndb->init(1024);
 
55
  if (m_ndb->waitUntilReady(30) != 0)
 
56
  {
 
57
    err << "Failed to connect to ndb!!" << endl;
 
58
    return false;
 
59
  }
 
60
  info << "Connected to ndb!!" << endl;
 
61
 
 
62
  m_callback = new restore_callback_t[m_parallelism];
 
63
 
 
64
  if (m_callback == 0)
 
65
  {
 
66
    err << "Failed to allocate callback structs" << endl;
 
67
    return false;
 
68
  }
 
69
 
 
70
  m_free_callback= m_callback;
 
71
  for (Uint32 i= 0; i < m_parallelism; i++) {
 
72
    m_callback[i].restore= this;
 
73
    m_callback[i].connection= 0;
 
74
    if (i > 0)
 
75
      m_callback[i-1].next= &(m_callback[i]);
 
76
  }
 
77
  m_callback[m_parallelism-1].next = 0;
 
78
 
 
79
  return true;
 
80
}
 
81
 
 
82
void BackupRestore::release()
 
83
{
 
84
  if (m_ndb)
 
85
  {
 
86
    delete m_ndb;
 
87
    m_ndb= 0;
 
88
  }
 
89
 
 
90
  if (m_callback)
 
91
  {
 
92
    delete [] m_callback;
 
93
    m_callback= 0;
 
94
  }
 
95
 
 
96
  if (m_cluster_connection)
 
97
  {
 
98
    delete m_cluster_connection;
 
99
    m_cluster_connection= 0;
 
100
  }
 
101
}
 
102
 
 
103
BackupRestore::~BackupRestore()
 
104
{
 
105
  release();
 
106
}
 
107
 
 
108
static
 
109
int 
 
110
match_blob(const char * name){
 
111
  int cnt, id1, id2;
 
112
  char buf[256];
 
113
  if((cnt = sscanf(name, "%[^/]/%[^/]/NDB$BLOB_%d_%d", buf, buf, &id1, &id2)) == 4){
 
114
    return id1;
 
115
  }
 
116
  
 
117
  return -1;
 
118
}
 
119
 
 
120
const NdbDictionary::Table*
 
121
BackupRestore::get_table(const NdbDictionary::Table* tab){
 
122
  if(m_cache.m_old_table == tab)
 
123
    return m_cache.m_new_table;
 
124
  m_cache.m_old_table = tab;
 
125
 
 
126
  int cnt, id1, id2;
 
127
  char db[256], schema[256];
 
128
  if((cnt = sscanf(tab->getName(), "%[^/]/%[^/]/NDB$BLOB_%d_%d", 
 
129
                   db, schema, &id1, &id2)) == 4){
 
130
    m_ndb->setDatabaseName(db);
 
131
    m_ndb->setSchemaName(schema);
 
132
    
 
133
    BaseString::snprintf(db, sizeof(db), "NDB$BLOB_%d_%d", 
 
134
                         m_new_tables[id1]->getTableId(), id2);
 
135
    
 
136
    m_cache.m_new_table = m_ndb->getDictionary()->getTable(db);
 
137
    
 
138
  } else {
 
139
    m_cache.m_new_table = m_new_tables[tab->getTableId()];
 
140
  }
 
141
  assert(m_cache.m_new_table);
 
142
  return m_cache.m_new_table;
 
143
}
 
144
 
 
145
bool
 
146
BackupRestore::finalize_table(const TableS & table){
 
147
  bool ret= true;
 
148
  if (!m_restore && !m_restore_meta)
 
149
    return ret;
 
150
  if (!table.have_auto_inc())
 
151
    return ret;
 
152
 
 
153
  Uint64 max_val= table.get_max_auto_val();
 
154
  do
 
155
  {
 
156
    Uint64 auto_val = ~(Uint64)0;
 
157
    int r= m_ndb->readAutoIncrementValue(get_table(table.m_dictTable), auto_val);
 
158
    if (r == -1 && m_ndb->getNdbError().status == NdbError::TemporaryError)
 
159
    {
 
160
      NdbSleep_MilliSleep(50);
 
161
      continue; // retry
 
162
    }
 
163
    else if (r == -1 && m_ndb->getNdbError().code != 626)
 
164
    {
 
165
      ret= false;
 
166
    }
 
167
    else if ((r == -1 && m_ndb->getNdbError().code == 626) ||
 
168
             max_val+1 > auto_val || auto_val == ~(Uint64)0)
 
169
    {
 
170
      r= m_ndb->setAutoIncrementValue(get_table(table.m_dictTable),
 
171
                                      max_val+1, false);
 
172
      if (r == -1 &&
 
173
            m_ndb->getNdbError().status == NdbError::TemporaryError)
 
174
      {
 
175
        NdbSleep_MilliSleep(50);
 
176
        continue; // retry
 
177
      }
 
178
      ret = (r == 0);
 
179
    }
 
180
    return (ret);
 
181
  } while (1);
 
182
}
 
183
 
 
184
 
 
185
#ifdef NOT_USED
 
186
static bool default_nodegroups(NdbDictionary::Table *table)
 
187
{
 
188
  Uint16 *node_groups = (Uint16*)table->getFragmentData();
 
189
  Uint32 no_parts = table->getFragmentDataLen() >> 1;
 
190
  Uint32 i;
 
191
 
 
192
  if (node_groups[0] != 0)
 
193
    return false; 
 
194
  for (i = 1; i < no_parts; i++) 
 
195
  {
 
196
    if (node_groups[i] != UNDEF_NODEGROUP)
 
197
      return false;
 
198
  }
 
199
  return true;
 
200
}
 
201
#endif
 
202
 
 
203
 
 
204
static Uint32 get_no_fragments(Uint64 max_rows, Uint32 no_nodes)
 
205
{
 
206
  Uint32 i = 0;
 
207
  Uint32 acc_row_size = 27;
 
208
  Uint32 acc_fragment_size = 512*1024*1024;
 
209
  Uint32 no_parts= (max_rows*acc_row_size)/acc_fragment_size + 1;
 
210
  Uint32 reported_parts = no_nodes; 
 
211
  while (reported_parts < no_parts && ++i < 4 &&
 
212
         (reported_parts + no_parts) < MAX_NDB_PARTITIONS)
 
213
    reported_parts+= no_nodes;
 
214
  if (reported_parts < no_parts)
 
215
  {
 
216
    err << "Table will be restored but will not be able to handle the maximum";
 
217
    err << " amount of rows as requested" << endl;
 
218
  }
 
219
  return reported_parts;
 
220
}
 
221
 
 
222
 
 
223
static void set_default_nodegroups(NdbDictionary::Table *table)
 
224
{
 
225
  Uint32 no_parts = table->getFragmentCount();
 
226
  Uint16 node_group[MAX_NDB_PARTITIONS];
 
227
  Uint32 i;
 
228
 
 
229
  node_group[0] = 0;
 
230
  for (i = 1; i < no_parts; i++)
 
231
  {
 
232
    node_group[i] = UNDEF_NODEGROUP;
 
233
  }
 
234
  table->setFragmentData((const void*)node_group, 2 * no_parts);
 
235
}
 
236
 
 
237
Uint32 BackupRestore::map_ng(Uint32 ng)
 
238
{
 
239
  NODE_GROUP_MAP *ng_map = m_nodegroup_map;
 
240
 
 
241
  if (ng == UNDEF_NODEGROUP ||
 
242
      ng_map[ng].map_array[0] == UNDEF_NODEGROUP)
 
243
  {
 
244
    return ng;
 
245
  }
 
246
  else
 
247
  {
 
248
    Uint32 new_ng;
 
249
    Uint32 curr_inx = ng_map[ng].curr_index;
 
250
    Uint32 new_curr_inx = curr_inx + 1;
 
251
 
 
252
    assert(ng < MAX_NDB_PARTITIONS);
 
253
    assert(curr_inx < MAX_MAPS_PER_NODE_GROUP);
 
254
    assert(new_curr_inx < MAX_MAPS_PER_NODE_GROUP);
 
255
 
 
256
    if (new_curr_inx >= MAX_MAPS_PER_NODE_GROUP)
 
257
      new_curr_inx = 0;
 
258
    else if (ng_map[ng].map_array[new_curr_inx] == UNDEF_NODEGROUP)
 
259
      new_curr_inx = 0;
 
260
    new_ng = ng_map[ng].map_array[curr_inx];
 
261
    ng_map[ng].curr_index = new_curr_inx;
 
262
    return new_ng;
 
263
  }
 
264
}
 
265
 
 
266
 
 
267
bool BackupRestore::map_nodegroups(Uint16 *ng_array, Uint32 no_parts)
 
268
{
 
269
  Uint32 i;
 
270
  bool mapped = FALSE;
 
271
  DBUG_ENTER("map_nodegroups");
 
272
 
 
273
  assert(no_parts < MAX_NDB_PARTITIONS);
 
274
  for (i = 0; i < no_parts; i++)
 
275
  {
 
276
    Uint32 ng;
 
277
    ng = map_ng((Uint32)ng_array[i]);
 
278
    if (ng != ng_array[i])
 
279
      mapped = TRUE;
 
280
    ng_array[i] = ng;
 
281
  }
 
282
  DBUG_RETURN(mapped);
 
283
}
 
284
 
 
285
 
 
286
static void copy_byte(const char **data, char **new_data, uint *len)
 
287
{
 
288
  **new_data = **data;
 
289
  (*data)++;
 
290
  (*new_data)++;
 
291
  (*len)++;
 
292
}
 
293
 
 
294
 
 
295
bool BackupRestore::search_replace(char *search_str, char **new_data,
 
296
                                   const char **data, const char *end_data,
 
297
                                   uint *new_data_len)
 
298
{
 
299
  uint search_str_len = strlen(search_str);
 
300
  uint inx = 0;
 
301
  bool in_delimiters = FALSE;
 
302
  bool escape_char = FALSE;
 
303
  char start_delimiter = 0;
 
304
  DBUG_ENTER("search_replace");
 
305
 
 
306
  do
 
307
  {
 
308
    char c = **data;
 
309
    copy_byte(data, new_data, new_data_len);
 
310
    if (escape_char)
 
311
    {
 
312
      escape_char = FALSE;
 
313
    }
 
314
    else if (in_delimiters)
 
315
    {
 
316
      if (c == start_delimiter)
 
317
        in_delimiters = FALSE;
 
318
    }
 
319
    else if (c == '\'' || c == '\"')
 
320
    {
 
321
      in_delimiters = TRUE;
 
322
      start_delimiter = c;
 
323
    }
 
324
    else if (c == '\\')
 
325
    {
 
326
      escape_char = TRUE;
 
327
    }
 
328
    else if (c == search_str[inx])
 
329
    {
 
330
      inx++;
 
331
      if (inx == search_str_len)
 
332
      {
 
333
        bool found = FALSE;
 
334
        uint number = 0;
 
335
        while (*data != end_data)
 
336
        {
 
337
          if (isdigit(**data))
 
338
          {
 
339
            found = TRUE;
 
340
            number = (10 * number) + (**data);
 
341
            if (number > MAX_NDB_NODES)
 
342
              break;
 
343
          }
 
344
          else if (found)
 
345
          {
 
346
            /*
 
347
               After long and tedious preparations we have actually found
 
348
               a node group identifier to convert. We'll use the mapping
 
349
               table created for node groups and then insert the new number
 
350
               instead of the old number.
 
351
            */
 
352
            uint temp = map_ng(number);
 
353
            int no_digits = 0;
 
354
            char digits[10];
 
355
            while (temp != 0)
 
356
            {
 
357
              digits[no_digits] = temp % 10;
 
358
              no_digits++;
 
359
              temp/=10;
 
360
            }
 
361
            for (no_digits--; no_digits >= 0; no_digits--)
 
362
            {
 
363
              **new_data = digits[no_digits];
 
364
              *new_data_len+=1;
 
365
            }
 
366
            DBUG_RETURN(FALSE); 
 
367
          }
 
368
          else
 
369
            break;
 
370
          (*data)++;
 
371
        }
 
372
        DBUG_RETURN(TRUE);
 
373
      }
 
374
    }
 
375
    else
 
376
      inx = 0;
 
377
  } while (*data < end_data);
 
378
  DBUG_RETURN(FALSE);
 
379
}
 
380
 
 
381
bool BackupRestore::map_in_frm(char *new_data, const char *data,
 
382
                                       uint data_len, uint *new_data_len)
 
383
{
 
384
  const char *end_data= data + data_len;
 
385
  const char *end_part_data;
 
386
  const char *part_data;
 
387
  char *extra_ptr;
 
388
  uint start_key_definition_len = uint2korr(data + 6);
 
389
  uint key_definition_len = uint4korr(data + 47);
 
390
  uint part_info_len;
 
391
  DBUG_ENTER("map_in_frm");
 
392
 
 
393
  if (data_len < 4096) goto error;
 
394
  extra_ptr = (char*)data + start_key_definition_len + key_definition_len;
 
395
  if ((int)data_len < ((extra_ptr - data) + 2)) goto error;
 
396
  extra_ptr = extra_ptr + 2 + uint2korr(extra_ptr);
 
397
  if ((int)data_len < ((extra_ptr - data) + 2)) goto error;
 
398
  extra_ptr = extra_ptr + 2 + uint2korr(extra_ptr);
 
399
  if ((int)data_len < ((extra_ptr - data) + 4)) goto error;
 
400
  part_info_len = uint4korr(extra_ptr);
 
401
  part_data = extra_ptr + 4;
 
402
  if ((int)data_len < ((part_data + part_info_len) - data)) goto error;
 
403
 
 
404
  do
 
405
  {
 
406
    copy_byte(&data, &new_data, new_data_len);
 
407
  } while (data < part_data);
 
408
  end_part_data = part_data + part_info_len;
 
409
  do
 
410
  {
 
411
    if (search_replace((char*)" NODEGROUP = ", &new_data, &data,
 
412
                       end_part_data, new_data_len))
 
413
      goto error;
 
414
  } while (data != end_part_data);
 
415
  do
 
416
  {
 
417
    copy_byte(&data, &new_data, new_data_len);
 
418
  } while (data < end_data);
 
419
  DBUG_RETURN(FALSE);
 
420
error:
 
421
  DBUG_RETURN(TRUE);
 
422
}
 
423
 
 
424
 
 
425
bool BackupRestore::translate_frm(NdbDictionary::Table *table)
 
426
{
 
427
  uchar *pack_data, *data, *new_pack_data;
 
428
  char *new_data;
 
429
  uint new_data_len;
 
430
  size_t data_len, new_pack_len;
 
431
  uint no_parts, extra_growth;
 
432
  DBUG_ENTER("translate_frm");
 
433
 
 
434
  pack_data = (uchar*) table->getFrmData();
 
435
  no_parts = table->getFragmentCount();
 
436
  /*
 
437
    Add max 4 characters per partition to handle worst case
 
438
    of mapping from single digit to 5-digit number.
 
439
    Fairly future-proof, ok up to 99999 node groups.
 
440
  */
 
441
  extra_growth = no_parts * 4;
 
442
  if (unpackfrm(&data, &data_len, pack_data))
 
443
  {
 
444
    DBUG_RETURN(TRUE);
 
445
  }
 
446
  if ((new_data = (char*) my_malloc(data_len + extra_growth, MYF(0))))
 
447
  {
 
448
    DBUG_RETURN(TRUE);
 
449
  }
 
450
  if (map_in_frm(new_data, (const char*)data, data_len, &new_data_len))
 
451
  {
 
452
    my_free(new_data, MYF(0));
 
453
    DBUG_RETURN(TRUE);
 
454
  }
 
455
  if (packfrm((uchar*) new_data, new_data_len,
 
456
              &new_pack_data, &new_pack_len))
 
457
  {
 
458
    my_free(new_data, MYF(0));
 
459
    DBUG_RETURN(TRUE);
 
460
  }
 
461
  table->setFrm(new_pack_data, new_pack_len);
 
462
  DBUG_RETURN(FALSE);
 
463
}
 
464
 
 
465
#include <signaldata/DictTabInfo.hpp>
 
466
 
 
467
bool
 
468
BackupRestore::object(Uint32 type, const void * ptr)
 
469
{
 
470
  if (!m_restore_meta)
 
471
    return true;
 
472
  
 
473
  NdbDictionary::Dictionary* dict = m_ndb->getDictionary();
 
474
  switch(type){
 
475
  case DictTabInfo::Tablespace:
 
476
  {
 
477
    NdbDictionary::Tablespace old(*(NdbDictionary::Tablespace*)ptr);
 
478
 
 
479
    Uint32 id = old.getObjectId();
 
480
 
 
481
    if (!m_no_restore_disk)
 
482
    {
 
483
      NdbDictionary::LogfileGroup * lg = m_logfilegroups[old.getDefaultLogfileGroupId()];
 
484
      old.setDefaultLogfileGroup(* lg);
 
485
      info << "Creating tablespace: " << old.getName() << "..." << flush;
 
486
      int ret = dict->createTablespace(old);
 
487
      if (ret)
 
488
      {
 
489
        NdbError errobj= dict->getNdbError();
 
490
        info << "FAILED" << endl;
 
491
        err << "Create tablespace failed: " << old.getName() << ": " << errobj << endl;
 
492
        return false;
 
493
      }
 
494
      info << "done" << endl;
 
495
    }
 
496
    
 
497
    NdbDictionary::Tablespace curr = dict->getTablespace(old.getName());
 
498
    NdbError errobj = dict->getNdbError();
 
499
    if ((int) errobj.classification == (int) ndberror_cl_none)
 
500
    {
 
501
      NdbDictionary::Tablespace* currptr = new NdbDictionary::Tablespace(curr);
 
502
      NdbDictionary::Tablespace * null = 0;
 
503
      m_tablespaces.set(currptr, id, null);
 
504
      debug << "Retreived tablespace: " << currptr->getName() 
 
505
            << " oldid: " << id << " newid: " << currptr->getObjectId() 
 
506
            << " " << (void*)currptr << endl;
 
507
      return true;
 
508
    }
 
509
    
 
510
    err << "Failed to retrieve tablespace \"" << old.getName() << "\": "
 
511
        << errobj << endl;
 
512
    
 
513
    return false;
 
514
    break;
 
515
  }
 
516
  case DictTabInfo::LogfileGroup:
 
517
  {
 
518
    NdbDictionary::LogfileGroup old(*(NdbDictionary::LogfileGroup*)ptr);
 
519
    
 
520
    Uint32 id = old.getObjectId();
 
521
    
 
522
    if (!m_no_restore_disk)
 
523
    {
 
524
      info << "Creating logfile group: " << old.getName() << "..." << flush;
 
525
      int ret = dict->createLogfileGroup(old);
 
526
      if (ret)
 
527
      {
 
528
        NdbError errobj= dict->getNdbError();
 
529
        info << "FAILED" << endl;
 
530
        err << "Create logfile group failed: " << old.getName() << ": " << errobj << endl;
 
531
        return false;
 
532
      }
 
533
      info << "done" << endl;
 
534
    }
 
535
    
 
536
    NdbDictionary::LogfileGroup curr = dict->getLogfileGroup(old.getName());
 
537
    NdbError errobj = dict->getNdbError();
 
538
    if ((int) errobj.classification == (int) ndberror_cl_none)
 
539
    {
 
540
      NdbDictionary::LogfileGroup* currptr = 
 
541
        new NdbDictionary::LogfileGroup(curr);
 
542
      NdbDictionary::LogfileGroup * null = 0;
 
543
      m_logfilegroups.set(currptr, id, null);
 
544
      debug << "Retreived logfile group: " << currptr->getName() 
 
545
            << " oldid: " << id << " newid: " << currptr->getObjectId() 
 
546
            << " " << (void*)currptr << endl;
 
547
      return true;
 
548
    }
 
549
    
 
550
    err << "Failed to retrieve logfile group \"" << old.getName() << "\": "
 
551
        << errobj << endl;
 
552
    
 
553
    return false;
 
554
    break;
 
555
  }
 
556
  case DictTabInfo::Datafile:
 
557
  {
 
558
    if (!m_no_restore_disk)
 
559
    {
 
560
      NdbDictionary::Datafile old(*(NdbDictionary::Datafile*)ptr);
 
561
      NdbDictionary::ObjectId objid;
 
562
      old.getTablespaceId(&objid);
 
563
      NdbDictionary::Tablespace * ts = m_tablespaces[objid.getObjectId()];
 
564
      debug << "Connecting datafile " << old.getPath() 
 
565
            << " to tablespace: oldid: " << objid.getObjectId()
 
566
            << " newid: " << ts->getObjectId() << endl;
 
567
      old.setTablespace(* ts);
 
568
      info << "Creating datafile \"" << old.getPath() << "\"..." << flush;
 
569
      if (dict->createDatafile(old))
 
570
      {
 
571
        NdbError errobj= dict->getNdbError();
 
572
        info << "FAILED" << endl;
 
573
        err << "Create datafile failed: " << old.getPath() << ": " << errobj << endl;
 
574
        return false;
 
575
      }
 
576
      info << "done" << endl;
 
577
    }
 
578
    return true;
 
579
    break;
 
580
  }
 
581
  case DictTabInfo::Undofile:
 
582
  {
 
583
    if (!m_no_restore_disk)
 
584
    {
 
585
      NdbDictionary::Undofile old(*(NdbDictionary::Undofile*)ptr);
 
586
      NdbDictionary::ObjectId objid;
 
587
      old.getLogfileGroupId(&objid);
 
588
      NdbDictionary::LogfileGroup * lg = m_logfilegroups[objid.getObjectId()];
 
589
      debug << "Connecting undofile " << old.getPath() 
 
590
            << " to logfile group: oldid: " << objid.getObjectId()
 
591
            << " newid: " << lg->getObjectId() 
 
592
            << " " << (void*)lg << endl;
 
593
      old.setLogfileGroup(* lg);
 
594
      info << "Creating undofile \"" << old.getPath() << "\"..." << flush;
 
595
      if (dict->createUndofile(old))
 
596
      {
 
597
        NdbError errobj= dict->getNdbError();
 
598
        info << "FAILED" << endl;
 
599
        err << "Create undofile failed: " << old.getPath() << ": " << errobj << endl;
 
600
        return false;
 
601
      }
 
602
      info << "done" << endl;
 
603
    }
 
604
    return true;
 
605
    break;
 
606
  }
 
607
  }
 
608
  return true;
 
609
}
 
610
 
 
611
bool
 
612
BackupRestore::has_temp_error(){
 
613
  return m_temp_error;
 
614
}
 
615
 
 
616
bool
 
617
BackupRestore::update_apply_status(const RestoreMetaData &metaData)
 
618
{
 
619
  if (!m_restore_epoch)
 
620
    return true;
 
621
 
 
622
  bool result= false;
 
623
  unsigned apply_table_format= 0;
 
624
 
 
625
  m_ndb->setDatabaseName(NDB_REP_DB);
 
626
  m_ndb->setSchemaName("def");
 
627
 
 
628
  NdbDictionary::Dictionary *dict= m_ndb->getDictionary();
 
629
  const NdbDictionary::Table *ndbtab= dict->getTable(Ndb_apply_table);
 
630
  if (!ndbtab)
 
631
  {
 
632
    err << Ndb_apply_table << ": "
 
633
        << dict->getNdbError() << endl;
 
634
    return false;
 
635
  }
 
636
  if
 
637
    (ndbtab->getColumn(0)->getType() == NdbDictionary::Column::Unsigned &&
 
638
     ndbtab->getColumn(1)->getType() == NdbDictionary::Column::Bigunsigned)
 
639
  {
 
640
    if (ndbtab->getNoOfColumns() == 2)
 
641
    {
 
642
      apply_table_format= 1;
 
643
    }
 
644
    else if
 
645
      (ndbtab->getColumn(2)->getType() == NdbDictionary::Column::Varchar &&
 
646
       ndbtab->getColumn(3)->getType() == NdbDictionary::Column::Bigunsigned &&
 
647
       ndbtab->getColumn(4)->getType() == NdbDictionary::Column::Bigunsigned)
 
648
    {
 
649
      apply_table_format= 2;
 
650
    }
 
651
  }
 
652
  if (apply_table_format == 0)
 
653
  {
 
654
    err << Ndb_apply_table << " has wrong format\n";
 
655
    return false;
 
656
  }
 
657
 
 
658
  Uint32 server_id= 0;
 
659
  Uint64 epoch= metaData.getStopGCP();
 
660
  Uint64 zero= 0;
 
661
  char empty_string[1];
 
662
  empty_string[0]= 0;
 
663
  NdbTransaction * trans= m_ndb->startTransaction();
 
664
  if (!trans)
 
665
  {
 
666
    err << Ndb_apply_table << ": "
 
667
        << m_ndb->getNdbError() << endl;
 
668
    return false;
 
669
  }
 
670
  NdbOperation * op= trans->getNdbOperation(ndbtab);
 
671
  if (!op)
 
672
  {
 
673
    err << Ndb_apply_table << ": "
 
674
        << trans->getNdbError() << endl;
 
675
    goto err;
 
676
  }
 
677
  if (op->writeTuple() ||
 
678
      op->equal(0u, (const char *)&server_id, sizeof(server_id)) ||
 
679
      op->setValue(1u, (const char *)&epoch, sizeof(epoch)))
 
680
  {
 
681
    err << Ndb_apply_table << ": "
 
682
        << op->getNdbError() << endl;
 
683
    goto err;
 
684
  }
 
685
  if ((apply_table_format == 2) &&
 
686
      (op->setValue(2u, (const char *)&empty_string, 1) ||
 
687
       op->setValue(3u, (const char *)&zero, sizeof(zero)) ||
 
688
       op->setValue(4u, (const char *)&zero, sizeof(zero))))
 
689
  {
 
690
    err << Ndb_apply_table << ": "
 
691
        << op->getNdbError() << endl;
 
692
    goto err;
 
693
  }
 
694
  if (trans->execute(NdbTransaction::Commit))
 
695
  {
 
696
    err << Ndb_apply_table << ": "
 
697
        << trans->getNdbError() << endl;
 
698
    goto err;
 
699
  }
 
700
  result= true;
 
701
err:
 
702
  m_ndb->closeTransaction(trans);
 
703
  return result;
 
704
}
 
705
 
 
706
bool
 
707
BackupRestore::table_equal(const TableS &tableS)
 
708
{
 
709
  if (!m_restore)
 
710
    return true;
 
711
 
 
712
  const char *tablename = tableS.getTableName();
 
713
 
 
714
  if(tableS.m_dictTable == NULL){
 
715
    ndbout<<"Table %s has no m_dictTable " << tablename << endl;
 
716
    return false;
 
717
  }
 
718
  /**
 
719
   * Ignore blob tables
 
720
   */
 
721
  if(match_blob(tablename) >= 0)
 
722
    return true;
 
723
 
 
724
  const NdbTableImpl & tmptab = NdbTableImpl::getImpl(* tableS.m_dictTable);
 
725
  if ((int) tmptab.m_indexType != (int) NdbDictionary::Index::Undefined){
 
726
    return true;
 
727
  }
 
728
 
 
729
  BaseString tmp(tablename);
 
730
  Vector<BaseString> split;
 
731
  if(tmp.split(split, "/") != 3){
 
732
    err << "Invalid table name format " << tablename << endl;
 
733
    return false;
 
734
  }
 
735
 
 
736
  m_ndb->setDatabaseName(split[0].c_str());
 
737
  m_ndb->setSchemaName(split[1].c_str());
 
738
 
 
739
  NdbDictionary::Dictionary* dict = m_ndb->getDictionary();  
 
740
  const NdbDictionary::Table* tab = dict->getTable(split[2].c_str());
 
741
  if(tab == 0){
 
742
    err << "Unable to find table: " << split[2].c_str() << endl;
 
743
    return false;
 
744
  }
 
745
 
 
746
  if(tab->getNoOfColumns() != tableS.m_dictTable->getNoOfColumns())
 
747
  {
 
748
    ndbout_c("m_columns.size %d != %d",tab->getNoOfColumns(),
 
749
                       tableS.m_dictTable->getNoOfColumns());
 
750
    return false;
 
751
  }
 
752
 
 
753
 for(int i = 0; i<tab->getNoOfColumns(); i++)
 
754
  {
 
755
    if(!tab->getColumn(i)->equal(*(tableS.m_dictTable->getColumn(i))))
 
756
    {
 
757
      ndbout_c("m_columns %s != %s",tab->getColumn(i)->getName(),
 
758
                tableS.m_dictTable->getColumn(i)->getName());
 
759
      return false;
 
760
    }
 
761
  }
 
762
 
 
763
  return true;
 
764
}
 
765
 
 
766
bool
 
767
BackupRestore::createSystable(const TableS & tables){
 
768
  if (!m_restore && !m_restore_meta && !m_restore_epoch)
 
769
    return true;
 
770
  const char *tablename = tables.getTableName();
 
771
 
 
772
  if( strcmp(tablename, NDB_REP_DB "/def/" NDB_APPLY_TABLE) != 0 &&
 
773
      strcmp(tablename, NDB_REP_DB "/def/" NDB_SCHEMA_TABLE) != 0 )
 
774
  {
 
775
    return true;
 
776
  }
 
777
 
 
778
  BaseString tmp(tablename);
 
779
  Vector<BaseString> split;
 
780
  if(tmp.split(split, "/") != 3){
 
781
    err << "Invalid table name format " << tablename << endl;
 
782
    return false;
 
783
  }
 
784
 
 
785
  m_ndb->setDatabaseName(split[0].c_str());
 
786
  m_ndb->setSchemaName(split[1].c_str());
 
787
 
 
788
  NdbDictionary::Dictionary* dict = m_ndb->getDictionary();
 
789
  if( dict->getTable(split[2].c_str()) != NULL ){
 
790
    return true;
 
791
  }
 
792
  return table(tables);
 
793
}
 
794
 
 
795
bool
 
796
BackupRestore::table(const TableS & table){
 
797
  if (!m_restore && !m_restore_meta)
 
798
    return true;
 
799
 
 
800
  const char * name = table.getTableName();
 
801
 
 
802
  /**
 
803
   * Ignore blob tables
 
804
   */
 
805
  if(match_blob(name) >= 0)
 
806
    return true;
 
807
  
 
808
  const NdbTableImpl & tmptab = NdbTableImpl::getImpl(* table.m_dictTable);
 
809
  if ((int) tmptab.m_indexType != (int) NdbDictionary::Index::Undefined){
 
810
    m_indexes.push_back(table.m_dictTable);
 
811
    return true;
 
812
  }
 
813
  
 
814
  BaseString tmp(name);
 
815
  Vector<BaseString> split;
 
816
  if(tmp.split(split, "/") != 3){
 
817
    err << "Invalid table name format `" << name << "`" << endl;
 
818
    return false;
 
819
  }
 
820
 
 
821
  m_ndb->setDatabaseName(split[0].c_str());
 
822
  m_ndb->setSchemaName(split[1].c_str());
 
823
  
 
824
  NdbDictionary::Dictionary* dict = m_ndb->getDictionary();
 
825
  if(m_restore_meta)
 
826
  {
 
827
    NdbDictionary::Table copy(*table.m_dictTable);
 
828
 
 
829
    copy.setName(split[2].c_str());
 
830
    Uint32 id;
 
831
    if (copy.getTablespace(&id))
 
832
    {
 
833
      debug << "Connecting " << name << " to tablespace oldid: " << id << flush;
 
834
      NdbDictionary::Tablespace* ts = m_tablespaces[id];
 
835
      debug << " newid: " << ts->getObjectId() << endl;
 
836
      copy.setTablespace(* ts);
 
837
    }
 
838
    
 
839
    if (copy.getDefaultNoPartitionsFlag())
 
840
    {
 
841
      /*
 
842
        Table was defined with default number of partitions. We can restore
 
843
        it with whatever is the default in this cluster.
 
844
        We use the max_rows parameter in calculating the default number.
 
845
      */
 
846
      Uint32 no_nodes = m_cluster_connection->no_db_nodes();
 
847
      copy.setFragmentCount(get_no_fragments(copy.getMaxRows(),
 
848
                            no_nodes));
 
849
      set_default_nodegroups(&copy);
 
850
    }
 
851
    else
 
852
    {
 
853
      /*
 
854
        Table was defined with specific number of partitions. It should be
 
855
        restored with the same number of partitions. It will either be
 
856
        restored in the same node groups as when backup was taken or by
 
857
        using a node group map supplied to the ndb_restore program.
 
858
      */
 
859
      Uint16 *ng_array = (Uint16*)copy.getFragmentData();
 
860
      Uint16 no_parts = copy.getFragmentCount();
 
861
      if (map_nodegroups(ng_array, no_parts))
 
862
      {
 
863
        if (translate_frm(&copy))
 
864
        {
 
865
          err << "Create table " << table.getTableName() << " failed: ";
 
866
          err << "Translate frm error" << endl;
 
867
          return false;
 
868
        }
 
869
      }
 
870
      copy.setFragmentData((const void *)ng_array, no_parts << 1);
 
871
    }
 
872
 
 
873
    /**
 
874
     * Force of varpart was introduced in 5.1.18, telco 6.1.7 and 6.2.1
 
875
     * Since default from mysqld is to add force of varpart (disable with
 
876
     * ROW_FORMAT=FIXED) we force varpart onto tables when they are restored
 
877
     * from backups taken with older versions. This will be wrong if
 
878
     * ROW_FORMAT=FIXED was used on original table, however the likelyhood of
 
879
     * this is low, since ROW_FORMAT= was a NOOP in older versions.
 
880
     */
 
881
 
 
882
    if (table.getBackupVersion() < MAKE_VERSION(5,1,18))
 
883
      copy.setForceVarPart(true);
 
884
    else if (getMajor(table.getBackupVersion()) == 6 &&
 
885
             (table.getBackupVersion() < MAKE_VERSION(6,1,7) ||
 
886
              table.getBackupVersion() == MAKE_VERSION(6,2,0)))
 
887
      copy.setForceVarPart(true);
 
888
 
 
889
    /*
 
890
      update min and max rows to reflect the table, this to
 
891
      ensure that memory is allocated properly in the ndb kernel
 
892
    */
 
893
    copy.setMinRows(table.getNoOfRecords());
 
894
    if (table.getNoOfRecords() > copy.getMaxRows())
 
895
    {
 
896
      copy.setMaxRows(table.getNoOfRecords());
 
897
    }
 
898
    
 
899
    NdbTableImpl &tableImpl = NdbTableImpl::getImpl(copy);
 
900
    if (table.getBackupVersion() < MAKE_VERSION(5,1,0) && !m_no_upgrade){
 
901
      for(int i= 0; i < copy.getNoOfColumns(); i++)
 
902
      {
 
903
        NdbDictionary::Column::Type t = copy.getColumn(i)->getType();
 
904
 
 
905
        if (t == NdbDictionary::Column::Varchar ||
 
906
          t == NdbDictionary::Column::Varbinary)
 
907
          tableImpl.getColumn(i)->setArrayType(NdbDictionary::Column::ArrayTypeShortVar);
 
908
        if (t == NdbDictionary::Column::Longvarchar ||
 
909
          t == NdbDictionary::Column::Longvarbinary)
 
910
          tableImpl.getColumn(i)->setArrayType(NdbDictionary::Column::ArrayTypeMediumVar);
 
911
      }
 
912
    }
 
913
 
 
914
    if (dict->createTable(copy) == -1) 
 
915
    {
 
916
      err << "Create table `" << table.getTableName() << "` failed: "
 
917
          << dict->getNdbError() << endl;
 
918
      if (dict->getNdbError().code == 771)
 
919
      {
 
920
        /*
 
921
          The user on the cluster where the backup was created had specified
 
922
          specific node groups for partitions. Some of these node groups
 
923
          didn't exist on this cluster. We will warn the user of this and
 
924
          inform him of his option.
 
925
        */
 
926
        err << "The node groups defined in the table didn't exist in this";
 
927
        err << " cluster." << endl << "There is an option to use the";
 
928
        err << " the parameter ndb-nodegroup-map to define a mapping from";
 
929
        err << endl << "the old nodegroups to new nodegroups" << endl; 
 
930
      }
 
931
      return false;
 
932
    }
 
933
    info << "Successfully restored table `"
 
934
         << table.getTableName() << "`" << endl;
 
935
  }  
 
936
  
 
937
  const NdbDictionary::Table* tab = dict->getTable(split[2].c_str());
 
938
  if(tab == 0){
 
939
    err << "Unable to find table: `" << split[2].c_str() << "`" << endl;
 
940
    return false;
 
941
  }
 
942
  if(m_restore_meta)
 
943
  {
 
944
    if (tab->getFrmData())
 
945
    {
 
946
      // a MySQL Server table is restored, thus an event should be created
 
947
      BaseString event_name("REPL$");
 
948
      event_name.append(split[0].c_str());
 
949
      event_name.append("/");
 
950
      event_name.append(split[2].c_str());
 
951
 
 
952
      NdbDictionary::Event my_event(event_name.c_str());
 
953
      my_event.setTable(*tab);
 
954
      my_event.addTableEvent(NdbDictionary::Event::TE_ALL);
 
955
 
 
956
      // add all columns to the event
 
957
      bool has_blobs = false;
 
958
      for(int a= 0; a < tab->getNoOfColumns(); a++)
 
959
      {
 
960
        my_event.addEventColumn(a);
 
961
        NdbDictionary::Column::Type t = tab->getColumn(a)->getType();
 
962
        if (t == NdbDictionary::Column::Blob ||
 
963
            t == NdbDictionary::Column::Text)
 
964
          has_blobs = true;
 
965
      }
 
966
      if (has_blobs)
 
967
        my_event.mergeEvents(true);
 
968
 
 
969
      while ( dict->createEvent(my_event) ) // Add event to database
 
970
      {
 
971
        if (dict->getNdbError().classification == NdbError::SchemaObjectExists)
 
972
        {
 
973
          info << "Event for table " << table.getTableName()
 
974
               << " already exists, removing.\n";
 
975
          if (!dict->dropEvent(my_event.getName()))
 
976
            continue;
 
977
        }
 
978
        err << "Create table event for " << table.getTableName() << " failed: "
 
979
            << dict->getNdbError() << endl;
 
980
        dict->dropTable(split[2].c_str());
 
981
        return false;
 
982
      }
 
983
      info << "Successfully restored table event " << event_name << endl ;
 
984
    }
 
985
  }
 
986
  const NdbDictionary::Table* null = 0;
 
987
  m_new_tables.fill(table.m_dictTable->getTableId(), null);
 
988
  m_new_tables[table.m_dictTable->getTableId()] = tab;
 
989
  return true;
 
990
}
 
991
 
 
992
bool
 
993
BackupRestore::endOfTables(){
 
994
  if(!m_restore_meta)
 
995
    return true;
 
996
 
 
997
  NdbDictionary::Dictionary* dict = m_ndb->getDictionary();
 
998
  for(size_t i = 0; i<m_indexes.size(); i++){
 
999
    NdbTableImpl & indtab = NdbTableImpl::getImpl(* m_indexes[i]);
 
1000
 
 
1001
    Vector<BaseString> split;
 
1002
    {
 
1003
      BaseString tmp(indtab.m_primaryTable.c_str());
 
1004
      if (tmp.split(split, "/") != 3)
 
1005
      {
 
1006
        err << "Invalid table name format `" << indtab.m_primaryTable.c_str()
 
1007
            << "`" << endl;
 
1008
        return false;
 
1009
      }
 
1010
    }
 
1011
    
 
1012
    m_ndb->setDatabaseName(split[0].c_str());
 
1013
    m_ndb->setSchemaName(split[1].c_str());
 
1014
    
 
1015
    const NdbDictionary::Table * prim = dict->getTable(split[2].c_str());
 
1016
    if(prim == 0){
 
1017
      err << "Unable to find base table `" << split[2].c_str() 
 
1018
          << "` for index `"
 
1019
          << indtab.getName() << "`" << endl;
 
1020
      return false;
 
1021
    }
 
1022
    NdbTableImpl& base = NdbTableImpl::getImpl(*prim);
 
1023
    NdbIndexImpl* idx;
 
1024
    Vector<BaseString> split_idx;
 
1025
    {
 
1026
      BaseString tmp(indtab.getName());
 
1027
      if (tmp.split(split_idx, "/") != 4)
 
1028
      {
 
1029
        err << "Invalid index name format `" << indtab.getName() << "`" << endl;
 
1030
        return false;
 
1031
      }
 
1032
    }
 
1033
    if(NdbDictInterface::create_index_obj_from_table(&idx, &indtab, &base))
 
1034
    {
 
1035
      err << "Failed to create index `" << split_idx[3]
 
1036
          << "` on " << split[2].c_str() << endl;
 
1037
        return false;
 
1038
    }
 
1039
    idx->setName(split_idx[3].c_str());
 
1040
    if(dict->createIndex(* idx) != 0)
 
1041
    {
 
1042
      delete idx;
 
1043
      err << "Failed to create index `" << split_idx[3].c_str()
 
1044
          << "` on `" << split[2].c_str() << "`" << endl
 
1045
          << dict->getNdbError() << endl;
 
1046
 
 
1047
      return false;
 
1048
    }
 
1049
    delete idx;
 
1050
    info << "Successfully created index `" << split_idx[3].c_str()
 
1051
         << "` on `" << split[2].c_str() << "`" << endl;
 
1052
  }
 
1053
  return true;
 
1054
}
 
1055
 
 
1056
void BackupRestore::tuple(const TupleS & tup, Uint32 fragmentId)
 
1057
{
 
1058
  if (!m_restore) 
 
1059
    return;
 
1060
 
 
1061
  while (m_free_callback == 0)
 
1062
  {
 
1063
    assert(m_transactions == m_parallelism);
 
1064
    // send-poll all transactions
 
1065
    // close transaction is done in callback
 
1066
    m_ndb->sendPollNdb(3000, 1);
 
1067
  }
 
1068
  
 
1069
  restore_callback_t * cb = m_free_callback;
 
1070
  
 
1071
  if (cb == 0)
 
1072
    assert(false);
 
1073
  
 
1074
  m_free_callback = cb->next;
 
1075
  cb->retries = 0;
 
1076
  cb->fragId = fragmentId;
 
1077
  cb->tup = tup; // must do copy!
 
1078
  tuple_a(cb);
 
1079
 
 
1080
}
 
1081
 
 
1082
void BackupRestore::tuple_a(restore_callback_t *cb)
 
1083
{
 
1084
  Uint32 partition_id = cb->fragId;
 
1085
  while (cb->retries < 10) 
 
1086
  {
 
1087
    /**
 
1088
     * start transactions
 
1089
     */
 
1090
    cb->connection = m_ndb->startTransaction();
 
1091
    if (cb->connection == NULL) 
 
1092
    {
 
1093
      if (errorHandler(cb)) 
 
1094
      {
 
1095
        m_ndb->sendPollNdb(3000, 1);
 
1096
        continue;
 
1097
      }
 
1098
      err << "Cannot start transaction" << endl;
 
1099
      exitHandler();
 
1100
    } // if
 
1101
    
 
1102
    const TupleS &tup = cb->tup;
 
1103
    const NdbDictionary::Table * table = get_table(tup.getTable()->m_dictTable);
 
1104
 
 
1105
    NdbOperation * op = cb->connection->getNdbOperation(table);
 
1106
    
 
1107
    if (op == NULL) 
 
1108
    {
 
1109
      if (errorHandler(cb)) 
 
1110
        continue;
 
1111
      err << "Cannot get operation: " << cb->connection->getNdbError() << endl;
 
1112
      exitHandler();
 
1113
    } // if
 
1114
    
 
1115
    if (op->writeTuple() == -1) 
 
1116
    {
 
1117
      if (errorHandler(cb))
 
1118
        continue;
 
1119
      err << "Error defining op: " << cb->connection->getNdbError() << endl;
 
1120
      exitHandler();
 
1121
    } // if
 
1122
 
 
1123
    if (table->getFragmentType() == NdbDictionary::Object::UserDefined)
 
1124
    {
 
1125
      if (table->getDefaultNoPartitionsFlag())
 
1126
      {
 
1127
        /*
 
1128
          This can only happen for HASH partitioning with
 
1129
          user defined hash function where user hasn't
 
1130
          specified the number of partitions and we
 
1131
          have to calculate it. We use the hash value
 
1132
          stored in the record to calculate the partition
 
1133
          to use.
 
1134
        */
 
1135
        int i = tup.getNoOfAttributes() - 1;
 
1136
        const AttributeData  *attr_data = tup.getData(i);
 
1137
        Uint32 hash_value =  *attr_data->u_int32_value;
 
1138
        op->setPartitionId(get_part_id(table, hash_value));
 
1139
      }
 
1140
      else
 
1141
      {
 
1142
        /*
 
1143
          Either RANGE or LIST (with or without subparts)
 
1144
          OR HASH partitioning with user defined hash
 
1145
          function but with fixed set of partitions.
 
1146
        */
 
1147
        op->setPartitionId(partition_id);
 
1148
      }
 
1149
    }
 
1150
    int ret = 0;
 
1151
    for (int j = 0; j < 2; j++)
 
1152
    {
 
1153
      for (int i = 0; i < tup.getNoOfAttributes(); i++) 
 
1154
      {
 
1155
        const AttributeDesc * attr_desc = tup.getDesc(i);
 
1156
        const AttributeData * attr_data = tup.getData(i);
 
1157
        int size = attr_desc->size;
 
1158
        int arraySize = attr_desc->arraySize;
 
1159
        char * dataPtr = attr_data->string_value;
 
1160
        Uint32 length = 0;
 
1161
       
 
1162
        if (!attr_data->null)
 
1163
        {
 
1164
          const unsigned char * src = (const unsigned char *)dataPtr;
 
1165
          switch(attr_desc->m_column->getType()){
 
1166
          case NdbDictionary::Column::Varchar:
 
1167
          case NdbDictionary::Column::Varbinary:
 
1168
            length = src[0] + 1;
 
1169
            break;
 
1170
          case NdbDictionary::Column::Longvarchar:
 
1171
          case NdbDictionary::Column::Longvarbinary:
 
1172
            length = src[0] + (src[1] << 8) + 2;
 
1173
            break;
 
1174
          default:
 
1175
            length = attr_data->size;
 
1176
            break;
 
1177
          }
 
1178
        }
 
1179
        if (j == 0 && tup.getTable()->have_auto_inc(i))
 
1180
          tup.getTable()->update_max_auto_val(dataPtr,size*arraySize);
 
1181
        
 
1182
        if (attr_desc->m_column->getPrimaryKey())
 
1183
        {
 
1184
          if (j == 1) continue;
 
1185
          ret = op->equal(i, dataPtr, length);
 
1186
        }
 
1187
        else
 
1188
        {
 
1189
          if (j == 0) continue;
 
1190
          if (attr_data->null) 
 
1191
            ret = op->setValue(i, NULL, 0);
 
1192
          else
 
1193
            ret = op->setValue(i, dataPtr, length);
 
1194
        }
 
1195
        if (ret < 0) {
 
1196
          ndbout_c("Column: %d type %d %d %d %d",i,
 
1197
                   attr_desc->m_column->getType(),
 
1198
                   size, arraySize, length);
 
1199
          break;
 
1200
        }
 
1201
      }
 
1202
      if (ret < 0)
 
1203
        break;
 
1204
    }
 
1205
    if (ret < 0)
 
1206
    {
 
1207
      if (errorHandler(cb)) 
 
1208
        continue;
 
1209
      err << "Error defining op: " << cb->connection->getNdbError() << endl;
 
1210
      exitHandler();
 
1211
    }
 
1212
 
 
1213
    // Prepare transaction (the transaction is NOT yet sent to NDB)
 
1214
    cb->connection->executeAsynchPrepare(NdbTransaction::Commit,
 
1215
                                         &callback, cb);
 
1216
    m_transactions++;
 
1217
    return;
 
1218
  }
 
1219
  err << "Retried transaction " << cb->retries << " times.\nLast error"
 
1220
      << m_ndb->getNdbError(cb->error_code) << endl
 
1221
      << "...Unable to recover from errors. Exiting..." << endl;
 
1222
  exitHandler();
 
1223
}
 
1224
 
 
1225
void BackupRestore::cback(int result, restore_callback_t *cb)
 
1226
{
 
1227
  m_transactions--;
 
1228
 
 
1229
  if (result < 0)
 
1230
  {
 
1231
    /**
 
1232
     * Error. temporary or permanent?
 
1233
     */
 
1234
    if (errorHandler(cb))
 
1235
      tuple_a(cb); // retry
 
1236
    else
 
1237
    {
 
1238
      err << "Restore: Failed to restore data due to a unrecoverable error. Exiting..." << endl;
 
1239
      exitHandler();
 
1240
    }
 
1241
  }
 
1242
  else
 
1243
  {
 
1244
    /**
 
1245
     * OK! close transaction
 
1246
     */
 
1247
    m_ndb->closeTransaction(cb->connection);
 
1248
    cb->connection= 0;
 
1249
    cb->next= m_free_callback;
 
1250
    m_free_callback= cb;
 
1251
    m_dataCount++;
 
1252
  }
 
1253
}
 
1254
 
 
1255
/**
 
1256
 * returns true if is recoverable,
 
1257
 * Error handling based on hugo
 
1258
 *  false if it is an  error that generates an abort.
 
1259
 */
 
1260
bool BackupRestore::errorHandler(restore_callback_t *cb) 
 
1261
{
 
1262
  NdbError error;
 
1263
  if(cb->connection)
 
1264
  {
 
1265
    error= cb->connection->getNdbError();
 
1266
    m_ndb->closeTransaction(cb->connection);
 
1267
    cb->connection= 0;
 
1268
  }
 
1269
  else
 
1270
  {
 
1271
    error= m_ndb->getNdbError();
 
1272
  } 
 
1273
 
 
1274
  Uint32 sleepTime = 100 + cb->retries * 300;
 
1275
  
 
1276
  cb->retries++;
 
1277
  cb->error_code = error.code;
 
1278
 
 
1279
  switch(error.status)
 
1280
  {
 
1281
  case NdbError::Success:
 
1282
    err << "Success error: " << error << endl;
 
1283
    return false;
 
1284
    // ERROR!
 
1285
    
 
1286
  case NdbError::TemporaryError:
 
1287
    err << "Temporary error: " << error << endl;
 
1288
    m_temp_error = true;
 
1289
    NdbSleep_MilliSleep(sleepTime);
 
1290
    return true;
 
1291
    // RETRY
 
1292
    
 
1293
  case NdbError::UnknownResult:
 
1294
    err << "Unknown: " << error << endl;
 
1295
    return false;
 
1296
    // ERROR!
 
1297
    
 
1298
  default:
 
1299
  case NdbError::PermanentError:
 
1300
    //ERROR
 
1301
    err << "Permanent: " << error << endl;
 
1302
    return false;
 
1303
  }
 
1304
  err << "No error status" << endl;
 
1305
  return false;
 
1306
}
 
1307
 
 
1308
void BackupRestore::exitHandler() 
 
1309
{
 
1310
  release();
 
1311
  NDBT_ProgramExit(NDBT_FAILED);
 
1312
  if (opt_core)
 
1313
    abort();
 
1314
  else
 
1315
    exit(NDBT_FAILED);
 
1316
}
 
1317
 
 
1318
 
 
1319
void
 
1320
BackupRestore::tuple_free()
 
1321
{
 
1322
  if (!m_restore)
 
1323
    return;
 
1324
 
 
1325
  // Poll all transactions
 
1326
  while (m_transactions)
 
1327
  {
 
1328
    m_ndb->sendPollNdb(3000);
 
1329
  }
 
1330
}
 
1331
 
 
1332
void
 
1333
BackupRestore::endOfTuples()
 
1334
{
 
1335
  tuple_free();
 
1336
}
 
1337
 
 
1338
#ifdef NOT_USED
 
1339
static bool use_part_id(const NdbDictionary::Table *table)
 
1340
{
 
1341
  if (table->getDefaultNoPartitionsFlag() &&
 
1342
      (table->getFragmentType() == NdbDictionary::Object::UserDefined))
 
1343
    return false;
 
1344
  else
 
1345
    return true;
 
1346
}
 
1347
#endif
 
1348
 
 
1349
static Uint32 get_part_id(const NdbDictionary::Table *table,
 
1350
                          Uint32 hash_value)
 
1351
{
 
1352
  Uint32 no_frags = table->getFragmentCount();
 
1353
  
 
1354
  if (table->getLinearFlag())
 
1355
  {
 
1356
    Uint32 part_id;
 
1357
    Uint32 mask = 1;
 
1358
    while (no_frags > mask) mask <<= 1;
 
1359
    mask--;
 
1360
    part_id = hash_value & mask;
 
1361
    if (part_id >= no_frags)
 
1362
      part_id = hash_value & (mask >> 1);
 
1363
    return part_id;
 
1364
  }
 
1365
  else
 
1366
    return (hash_value % no_frags);
 
1367
}
 
1368
 
 
1369
void
 
1370
BackupRestore::logEntry(const LogEntry & tup)
 
1371
{
 
1372
  if (!m_restore)
 
1373
    return;
 
1374
 
 
1375
  NdbTransaction * trans = m_ndb->startTransaction();
 
1376
  if (trans == NULL) 
 
1377
  {
 
1378
    // TODO: handle the error
 
1379
    err << "Cannot start transaction" << endl;
 
1380
    exitHandler();
 
1381
  } // if
 
1382
  
 
1383
  const NdbDictionary::Table * table = get_table(tup.m_table->m_dictTable);
 
1384
  NdbOperation * op = trans->getNdbOperation(table);
 
1385
  if (op == NULL) 
 
1386
  {
 
1387
    err << "Cannot get operation: " << trans->getNdbError() << endl;
 
1388
    exitHandler();
 
1389
  } // if
 
1390
  
 
1391
  int check = 0;
 
1392
  switch(tup.m_type)
 
1393
  {
 
1394
  case LogEntry::LE_INSERT:
 
1395
    check = op->insertTuple();
 
1396
    break;
 
1397
  case LogEntry::LE_UPDATE:
 
1398
    check = op->updateTuple();
 
1399
    break;
 
1400
  case LogEntry::LE_DELETE:
 
1401
    check = op->deleteTuple();
 
1402
    break;
 
1403
  default:
 
1404
    err << "Log entry has wrong operation type."
 
1405
           << " Exiting...";
 
1406
    exitHandler();
 
1407
  }
 
1408
 
 
1409
  if (check != 0) 
 
1410
  {
 
1411
    err << "Error defining op: " << trans->getNdbError() << endl;
 
1412
    exitHandler();
 
1413
  } // if
 
1414
 
 
1415
  if (table->getFragmentType() == NdbDictionary::Object::UserDefined)
 
1416
  {
 
1417
    if (table->getDefaultNoPartitionsFlag())
 
1418
    {
 
1419
      const AttributeS * attr = tup[tup.size()-1];
 
1420
      Uint32 hash_value = *(Uint32*)attr->Data.string_value;
 
1421
      op->setPartitionId(get_part_id(table, hash_value));
 
1422
    }
 
1423
    else
 
1424
      op->setPartitionId(tup.m_frag_id);
 
1425
  }
 
1426
 
 
1427
  Bitmask<4096> keys;
 
1428
  for (Uint32 i= 0; i < tup.size(); i++) 
 
1429
  {
 
1430
    const AttributeS * attr = tup[i];
 
1431
    int size = attr->Desc->size;
 
1432
    int arraySize = attr->Desc->arraySize;
 
1433
    const char * dataPtr = attr->Data.string_value;
 
1434
    
 
1435
    if (tup.m_table->have_auto_inc(attr->Desc->attrId))
 
1436
      tup.m_table->update_max_auto_val(dataPtr,size*arraySize);
 
1437
 
 
1438
    const Uint32 length = (size / 8) * arraySize;
 
1439
    if (attr->Desc->m_column->getPrimaryKey())
 
1440
    {
 
1441
      if(!keys.get(attr->Desc->attrId))
 
1442
      {
 
1443
        keys.set(attr->Desc->attrId);
 
1444
        check= op->equal(attr->Desc->attrId, dataPtr, length);
 
1445
      }
 
1446
    }
 
1447
    else
 
1448
      check= op->setValue(attr->Desc->attrId, dataPtr, length);
 
1449
    
 
1450
    if (check != 0) 
 
1451
    {
 
1452
      err << "Error defining op: " << trans->getNdbError() << endl;
 
1453
      exitHandler();
 
1454
    } // if
 
1455
  }
 
1456
  
 
1457
  const int ret = trans->execute(NdbTransaction::Commit);
 
1458
  if (ret != 0)
 
1459
  {
 
1460
    // Both insert update and delete can fail during log running
 
1461
    // and it's ok
 
1462
    // TODO: check that the error is either tuple exists or tuple does not exist?
 
1463
    bool ok= false;
 
1464
    NdbError errobj= trans->getNdbError();
 
1465
    switch(tup.m_type)
 
1466
    {
 
1467
    case LogEntry::LE_INSERT:
 
1468
      if(errobj.status == NdbError::PermanentError &&
 
1469
         errobj.classification == NdbError::ConstraintViolation)
 
1470
        ok= true;
 
1471
      break;
 
1472
    case LogEntry::LE_UPDATE:
 
1473
    case LogEntry::LE_DELETE:
 
1474
      if(errobj.status == NdbError::PermanentError &&
 
1475
         errobj.classification == NdbError::NoDataFound)
 
1476
        ok= true;
 
1477
      break;
 
1478
    }
 
1479
    if (!ok)
 
1480
    {
 
1481
      err << "execute failed: " << errobj << endl;
 
1482
      exitHandler();
 
1483
    }
 
1484
  }
 
1485
  
 
1486
  m_ndb->closeTransaction(trans);
 
1487
  m_logCount++;
 
1488
}
 
1489
 
 
1490
void
 
1491
BackupRestore::endOfLogEntrys()
 
1492
{
 
1493
  if (!m_restore)
 
1494
    return;
 
1495
 
 
1496
  info << "Restored " << m_dataCount << " tuples and "
 
1497
       << m_logCount << " log entries" << endl;
 
1498
}
 
1499
 
 
1500
/*
 
1501
 *   callback : This is called when the transaction is polled
 
1502
 *              
 
1503
 *   (This function must have three arguments: 
 
1504
 *   - The result of the transaction, 
 
1505
 *   - The NdbTransaction object, and 
 
1506
 *   - A pointer to an arbitrary object.)
 
1507
 */
 
1508
 
 
1509
static void
 
1510
callback(int result, NdbTransaction* trans, void* aObject)
 
1511
{
 
1512
  restore_callback_t *cb = (restore_callback_t *)aObject;
 
1513
  (cb->restore)->cback(result, cb);
 
1514
}
 
1515
 
 
1516
#if 0 // old tuple impl
 
1517
void
 
1518
BackupRestore::tuple(const TupleS & tup)
 
1519
{
 
1520
  if (!m_restore)
 
1521
    return;
 
1522
  while (1) 
 
1523
  {
 
1524
    NdbTransaction * trans = m_ndb->startTransaction();
 
1525
    if (trans == NULL) 
 
1526
    {
 
1527
      // TODO: handle the error
 
1528
      ndbout << "Cannot start transaction" << endl;
 
1529
      exitHandler();
 
1530
    } // if
 
1531
    
 
1532
    const TableS * table = tup.getTable();
 
1533
    NdbOperation * op = trans->getNdbOperation(table->getTableName());
 
1534
    if (op == NULL) 
 
1535
    {
 
1536
      ndbout << "Cannot get operation: ";
 
1537
      ndbout << trans->getNdbError() << endl;
 
1538
      exitHandler();
 
1539
    } // if
 
1540
    
 
1541
    // TODO: check return value and handle error
 
1542
    if (op->writeTuple() == -1) 
 
1543
    {
 
1544
      ndbout << "writeTuple call failed: ";
 
1545
      ndbout << trans->getNdbError() << endl;
 
1546
      exitHandler();
 
1547
    } // if
 
1548
    
 
1549
    for (int i = 0; i < tup.getNoOfAttributes(); i++) 
 
1550
    {
 
1551
      const AttributeS * attr = tup[i];
 
1552
      int size = attr->Desc->size;
 
1553
      int arraySize = attr->Desc->arraySize;
 
1554
      const char * dataPtr = attr->Data.string_value;
 
1555
      
 
1556
      const Uint32 length = (size * arraySize) / 8;
 
1557
      if (attr->Desc->m_column->getPrimaryKey()) 
 
1558
        op->equal(i, dataPtr, length);
 
1559
    }
 
1560
    
 
1561
    for (int i = 0; i < tup.getNoOfAttributes(); i++) 
 
1562
    {
 
1563
      const AttributeS * attr = tup[i];
 
1564
      int size = attr->Desc->size;
 
1565
      int arraySize = attr->Desc->arraySize;
 
1566
      const char * dataPtr = attr->Data.string_value;
 
1567
      
 
1568
      const Uint32 length = (size * arraySize) / 8;
 
1569
      if (!attr->Desc->m_column->getPrimaryKey())
 
1570
        if (attr->Data.null)
 
1571
          op->setValue(i, NULL, 0);
 
1572
        else
 
1573
          op->setValue(i, dataPtr, length);
 
1574
    }
 
1575
    int ret = trans->execute(NdbTransaction::Commit);
 
1576
    if (ret != 0)
 
1577
    {
 
1578
      ndbout << "execute failed: ";
 
1579
      ndbout << trans->getNdbError() << endl;
 
1580
      exitHandler();
 
1581
    }
 
1582
    m_ndb->closeTransaction(trans);
 
1583
    if (ret == 0)
 
1584
      break;
 
1585
  }
 
1586
  m_dataCount++;
 
1587
}
 
1588
#endif
 
1589
 
 
1590
template class Vector<NdbDictionary::Table*>;
 
1591
template class Vector<const NdbDictionary::Table*>;
 
1592
template class Vector<NdbDictionary::Tablespace*>;
 
1593
template class Vector<NdbDictionary::LogfileGroup*>;