forked from mysql/mysql-server
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathha_ndb_ddl_fk.cc
2730 lines (2391 loc) · 76.9 KB
/
ha_ndb_ddl_fk.cc
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
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
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
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
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
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
436
437
438
439
440
441
442
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
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
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
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
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
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
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
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
/*
Copyright (c) 2011, 2015, Oracle and/or its affiliates. All rights reserved.
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; version 2 of the License.
This program 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 General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
*/
#include "ha_ndbcluster_glue.h"
#include "ha_ndbcluster.h"
#include "ndb_table_guard.h"
#include "mysql/service_thd_alloc.h"
#define ERR_RETURN(err) \
{ \
const NdbError& tmp= err; \
DBUG_RETURN(ndb_to_mysql_error(&tmp)); \
}
// Typedefs for long names
typedef NdbDictionary::Dictionary NDBDICT;
typedef NdbDictionary::Table NDBTAB;
typedef NdbDictionary::Column NDBCOL;
typedef NdbDictionary::Index NDBINDEX;
typedef NdbDictionary::ForeignKey NDBFK;
/*
Foreign key data where this table is child or parent or both.
Like indexes, these are cached under each handler instance.
Unlike indexes, no references to global dictionary are kept.
*/
struct Ndb_fk_item : Sql_alloc
{
FOREIGN_KEY_INFO f_key_info;
int update_action; // NDBFK::FkAction
int delete_action;
bool is_child;
bool is_parent;
};
struct Ndb_fk_data : Sql_alloc
{
List<Ndb_fk_item> list;
uint cnt_child;
uint cnt_parent;
};
// Forward decl
static
const char *
fk_split_name(char dst[], const char * src, bool index= false);
/*
Create all the fks for a table.
The actual foreign keys are not passed in handler interface
so gets them from thd->lex :-(
*/
static
const NDBINDEX*
find_matching_index(NDBDICT* dict,
const NDBTAB * tab,
const NDBCOL * columns[],
/* OUT */ bool & matches_primary_key)
{
/**
* First check if it matches primary key
*/
{
matches_primary_key= FALSE;
uint cnt_pk= 0, cnt_col= 0;
for (unsigned i = 0; columns[i] != 0; i++)
{
cnt_col++;
if (columns[i]->getPrimaryKey())
cnt_pk++;
}
// check if all columns was part of full primary key
if (cnt_col == (uint)tab->getNoOfPrimaryKeys() &&
cnt_col == cnt_pk)
{
matches_primary_key= TRUE;
return 0;
}
}
/**
* check indexes...
* first choice is unique index
* second choice is ordered index...with as many columns as possible
*/
const int noinvalidate= 0;
uint best_matching_columns= 0;
const NDBINDEX* best_matching_index= 0;
NDBDICT::List index_list;
dict->listIndexes(index_list, *tab);
for (unsigned i = 0; i < index_list.count; i++)
{
const char * index_name= index_list.elements[i].name;
const NDBINDEX* index= dict->getIndexGlobal(index_name, *tab);
if (index->getType() == NDBINDEX::UniqueHashIndex)
{
uint cnt= 0;
for (unsigned j = 0; columns[j] != 0; j++)
{
/*
* Search for matching columns in any order
* since order does not matter for unique index
*/
bool found= FALSE;
for (unsigned c = 0; c < index->getNoOfColumns(); c++)
{
if (!strcmp(columns[j]->getName(), index->getColumn(c)->getName()))
{
found= TRUE;
break;
}
}
if (found)
cnt++;
else
break;
}
if (cnt == index->getNoOfColumns())
{
/**
* Full match...return this index, no need to look further
*/
if (best_matching_index)
{
// release ref to previous best candidate
dict->removeIndexGlobal(* best_matching_index, noinvalidate);
}
return index; // NOTE: also returns reference
}
/**
* Not full match...i.e not usable
*/
dict->removeIndexGlobal(* index, noinvalidate);
continue;
}
else if (index->getType() == NDBINDEX::OrderedIndex)
{
uint cnt= 0;
for (; columns[cnt] != 0; cnt++)
{
const NDBCOL * ndbcol= index->getColumn(cnt);
if (ndbcol == 0)
break;
if (strcmp(columns[cnt]->getName(), ndbcol->getName()) != 0)
break;
}
if (cnt > best_matching_columns)
{
/**
* better match...
*/
if (best_matching_index)
{
dict->removeIndexGlobal(* best_matching_index, noinvalidate);
}
best_matching_index= index;
best_matching_columns= cnt;
}
else
{
dict->removeIndexGlobal(* index, noinvalidate);
}
}
else
{
// what ?? unknown index type
assert(false);
dict->removeIndexGlobal(* index, noinvalidate);
continue;
}
}
return best_matching_index; // NOTE: also returns reference
}
static
void
setDbName(Ndb* ndb, const char * name)
{
if (name && strlen(name) != 0)
{
ndb->setDatabaseName(name);
}
}
struct Ndb_db_guard
{
Ndb_db_guard(Ndb* ndb) {
this->ndb = ndb;
strcpy(save_db, ndb->getDatabaseName());
}
void restore() {
ndb->setDatabaseName(save_db);
}
~Ndb_db_guard() {
ndb->setDatabaseName(save_db);
}
private:
Ndb* ndb;
char save_db[FN_REFLEN + 1];
};
/**
* ndbapi want's c-strings (null terminated)
* mysql frequently uses LEX-string...(ptr + len)
*
* also...they have changed between 5.1 and 5.5...
* add a small compability-kit
*/
static inline
const char *
lex2str(const LEX_STRING& str, char buf[], size_t len)
{
my_snprintf(buf, len, "%.*s", (int)str.length, str.str);
return buf;
}
static inline
const char *
lex2str(const char * str, char buf[], size_t len)
{
return str;
}
static inline
bool
isnull(const LEX_STRING& str)
{
return str.str == 0 || str.length == 0;
}
static inline
bool
isnull(const char * str)
{
return str == 0;
}
// copied from unused table_case_convert() in mysqld.h
static void
ndb_fk_casedn(char *name)
{
DBUG_ASSERT(name != 0);
uint length = (uint)strlen(name);
DBUG_ASSERT(files_charset_info != 0 &&
files_charset_info->casedn_multiply == 1);
files_charset_info->cset->casedn(files_charset_info,
name, length, name, length);
}
static int
ndb_fk_casecmp(const char* name1, const char* name2)
{
if (!lower_case_table_names)
{
return strcmp(name1, name2);
}
char tmp1[FN_LEN + 1];
char tmp2[FN_LEN + 1];
strcpy(tmp1, name1);
strcpy(tmp2, name2);
ndb_fk_casedn(tmp1);
ndb_fk_casedn(tmp2);
return strcmp(tmp1, tmp2);
}
extern bool ndb_show_foreign_key_mock_tables(THD* thd);
class Fk_util
{
THD* m_thd;
void
info(const char* fmt, ...) const
{
va_list args;
char msg[MYSQL_ERRMSG_SIZE];
va_start(args,fmt);
my_vsnprintf(msg, sizeof(msg), fmt, args);
va_end(args);
// Push as warning if user has turned on ndb_show_foreign_key_mock_tables
if (ndb_show_foreign_key_mock_tables(m_thd))
{
push_warning(m_thd, Sql_condition::SL_WARNING, ER_YES, msg);
}
// Print info to log
sql_print_information("NDB FK: %s", msg);
}
void
warn(const char* fmt, ...) const
{
va_list args;
char msg[MYSQL_ERRMSG_SIZE];
va_start(args,fmt);
my_vsnprintf(msg, sizeof(msg), fmt, args);
va_end(args);
push_warning(m_thd, Sql_condition::SL_WARNING, ER_CANNOT_ADD_FOREIGN, msg);
// Print warning to log
sql_print_warning("NDB FK: %s", msg);
}
void
error(const NdbDictionary::Dictionary* dict, const char* fmt, ...) const
{
va_list args;
char msg[MYSQL_ERRMSG_SIZE];
va_start(args,fmt);
my_vsnprintf(msg, sizeof(msg), fmt, args);
va_end(args);
push_warning(m_thd, Sql_condition::SL_WARNING,
ER_CANNOT_ADD_FOREIGN, msg);
char ndb_msg[MYSQL_ERRMSG_SIZE] = {0};
if (dict)
{
// Extract message from Ndb
const NdbError& error = dict->getNdbError();
my_snprintf(ndb_msg, sizeof(ndb_msg),
"%d '%s'", error.code, error.message);
push_warning_printf(m_thd, Sql_condition::SL_WARNING,
ER_CANNOT_ADD_FOREIGN, "Ndb error: %s", ndb_msg);
}
// Print error to log
sql_print_error("NDB FK: %s, Ndb error: %s", msg, ndb_msg);
}
void
remove_index_global(NdbDictionary::Dictionary* dict, const NdbDictionary::Index* index) const
{
if (!index)
return;
dict->removeIndexGlobal(*index, 0);
}
bool
copy_fk_to_new_parent(NdbDictionary::Dictionary* dict, NdbDictionary::ForeignKey& fk,
const char* new_parent_name, const char* column_names[]) const
{
DBUG_ENTER("copy_fk_to_new_parent");
DBUG_PRINT("info", ("new_parent_name: %s", new_parent_name));
// Load up the new parent table
Ndb_table_guard new_parent_tab(dict, new_parent_name);
if (!new_parent_tab.get_table())
{
error(dict, "Failed to load potentially new parent '%s'", new_parent_name);
DBUG_RETURN(false);
}
// Build new parent column list from parent column names
const NdbDictionary::Column* columns[NDB_MAX_ATTRIBUTES_IN_INDEX + 1];
{
unsigned num_columns = 0;
for (unsigned i = 0; column_names[i] != 0; i++)
{
DBUG_PRINT("info", ("column: %s", column_names[i]));
const NdbDictionary::Column* col =
new_parent_tab.get_table()->getColumn(column_names[i]);
if (!col)
{
// Parent table didn't have any column with the given name, can happen
warn("Could not resolve '%s' as fk parent for '%s' since it didn't have "
"all the referenced columns", new_parent_name, fk.getChildTable());
DBUG_RETURN(false);
}
columns[num_columns++]= col;
}
columns[num_columns]= 0;
}
NdbDictionary::ForeignKey new_fk(fk);
// Create name for the new fk by splitting the fk's name and replacing
// the <parent id> part in format "<parent_id>/<child_id>/<name>"
{
char name[FN_REFLEN+1];
unsigned parent_id, child_id;
if (sscanf(fk.getName(), "%u/%u/%s",
&parent_id, &child_id, name) != 3)
{
warn("Skip, failed to parse name of fk: %s", fk.getName());
DBUG_RETURN(false);
}
char fk_name[FN_REFLEN+1];
my_snprintf(fk_name, sizeof(fk_name), "%s",
name);
DBUG_PRINT("info", ("Setting new fk name: %s", fk_name));
new_fk.setName(fk_name);
}
// Find matching index
bool parent_primary_key= FALSE;
const NdbDictionary::Index* parent_index= find_matching_index(dict,
new_parent_tab.get_table(),
columns,
parent_primary_key);
DBUG_PRINT("info", ("parent_primary_key: %d", parent_primary_key));
// Check if either pk or index matched
if (!parent_primary_key && parent_index == 0)
{
warn("Could not resolve '%s' as fk parent for '%s' since no matching index "
"could be found", new_parent_name, fk.getChildTable());
DBUG_RETURN(false);
}
if (parent_index != 0)
{
DBUG_PRINT("info", ("Setting parent with index %s", parent_index->getName()));
new_fk.setParent(*new_parent_tab.get_table(), parent_index, columns);
}
else
{
DBUG_PRINT("info", ("Setting parent without index"));
new_fk.setParent(*new_parent_tab.get_table(), 0, columns);
}
// Old fk is dropped by cascading when the mock table is dropped
// Create new fk referencing the new table
DBUG_PRINT("info", ("Create new fk: %s", new_fk.getName()));
int flags = 0;
if (thd_test_options(m_thd, OPTION_NO_FOREIGN_KEY_CHECKS))
{
flags |= NdbDictionary::Dictionary::CreateFK_NoVerify;
}
NdbDictionary::ObjectId objid;
if (dict->createForeignKey(new_fk, &objid, flags) != 0)
{
error(dict, "Failed to create foreign key '%s'", new_fk.getName());
remove_index_global(dict, parent_index);
DBUG_RETURN(false);
}
remove_index_global(dict, parent_index);
DBUG_RETURN(true);
}
void
resolve_mock(NdbDictionary::Dictionary* dict,
const char* new_parent_name, const char* mock_name) const
{
DBUG_ENTER("resolve_mock");
DBUG_PRINT("enter", ("mock_name '%s'", mock_name));
DBUG_ASSERT(is_mock_name(mock_name));
// Load up the mock table
Ndb_table_guard mock_tab(dict, mock_name);
if (!mock_tab.get_table())
{
error(dict, "Failed to load the listed mock table '%s'", mock_name);
DBUG_ASSERT(false);
DBUG_VOID_RETURN;
}
// List dependent objects of mock table
NdbDictionary::Dictionary::List list;
if (dict->listDependentObjects(list, *mock_tab.get_table()) != 0)
{
error(dict, "Failed to list dependent objects for mock table '%s'", mock_name);
DBUG_VOID_RETURN;
}
for (unsigned i = 0; i < list.count; i++)
{
const NdbDictionary::Dictionary::List::Element& element = list.elements[i];
if (element.type != NdbDictionary::Object::ForeignKey)
continue;
DBUG_PRINT("info", ("fk: %s", element.name));
NdbDictionary::ForeignKey fk;
if (dict->getForeignKey(fk, element.name) != 0)
{
error(dict, "Could not find the listed fk '%s'", element.name);
continue;
}
// Build column name list for parent
const char* col_names[NDB_MAX_ATTRIBUTES_IN_INDEX + 1];
{
unsigned num_columns = 0;
for (unsigned j = 0; j < fk.getParentColumnCount(); j++)
{
const NdbDictionary::Column* col =
mock_tab.get_table()->getColumn(fk.getParentColumnNo(j));
if (!col)
{
error(NULL, "Could not find column '%s' in mock table '%s'",
fk.getParentColumnNo(j), mock_name);
continue;
}
col_names[num_columns++]= col->getName();
}
col_names[num_columns]= 0;
if (num_columns != fk.getParentColumnCount())
{
error(NULL, "Could not find all columns referenced by fk in mock table '%s'",
mock_name);
continue;
}
}
if (!copy_fk_to_new_parent(dict, fk, new_parent_name, col_names))
continue;
// New fk has been created between child and new parent, drop the mock
// table and it's related fk
const int drop_flags= NDBDICT::DropTableCascadeConstraints;
if (dict->dropTableGlobal(*mock_tab.get_table(), drop_flags) != 0)
{
error(dict, "Failed to drop mock table '%s'", mock_name);
continue;
}
info("Dropped mock table '%s' - resolved by '%s'", mock_name, new_parent_name);
}
DBUG_VOID_RETURN;
}
bool
create_mock_tables_and_drop(Ndb* ndb, NdbDictionary::Dictionary* dict,
const NdbDictionary::Table* table)
{
DBUG_ENTER("create_mock_tables_and_drop");
DBUG_PRINT("enter", ("table: %s", table->getName()));
/*
List all foreign keys referencing the table to be dropped
and recreate those to point at a new mock
*/
NdbDictionary::Dictionary::List list;
if (dict->listDependentObjects(list, *table) != 0)
{
error(dict, "Failed to list dependent objects for table '%s'", table->getName());
DBUG_RETURN(false);
}
uint fk_index = 0;
for (unsigned i = 0; i < list.count; i++)
{
const NdbDictionary::Dictionary::List::Element& element = list.elements[i];
if (element.type != NdbDictionary::Object::ForeignKey)
continue;
DBUG_PRINT("fk", ("name: %s, type: %d", element.name, element.type));
NdbDictionary::ForeignKey fk;
if (dict->getForeignKey(fk, element.name) != 0)
{
// Could not find the listed fk
DBUG_ASSERT(false);
continue;
}
// Parent of the found fk should be the table to be dropped
DBUG_PRINT("info", ("fk.parent: %s", fk.getParentTable()));
char parent_db_and_name[FN_LEN + 1];
const char * parent_name = fk_split_name(parent_db_and_name, fk.getParentTable());
if (strcmp(parent_db_and_name, ndb->getDatabaseName()) != 0 ||
strcmp(parent_name, table->getName()) != 0)
{
DBUG_PRINT("info", ("fk is not parent, skip"));
continue;
}
DBUG_PRINT("info", ("fk.child: %s", fk.getChildTable()));
char child_db_and_name[FN_LEN + 1];
const char * child_name = fk_split_name(child_db_and_name, fk.getChildTable());
// Open child table
Ndb_db_guard db_guard(ndb);
setDbName(ndb, child_db_and_name);
Ndb_table_guard child_tab(dict, child_name);
if (child_tab.get_table() == 0)
{
error(dict, "Failed to open child table '%s'", child_name);
DBUG_RETURN(false);
}
/* Format mock table name */
char mock_name[FN_REFLEN];
if (!format_name(mock_name, sizeof(mock_name),
child_tab.get_table()->getObjectId(),
fk_index, parent_name))
{
error(NULL, "Failed to create mock parent table, too long mock name");
DBUG_RETURN(false);
}
// Build both column name and column type list from parent(which will be dropped)
const char* col_names[NDB_MAX_ATTRIBUTES_IN_INDEX + 1];
const NdbDictionary::Column* col_types[NDB_MAX_ATTRIBUTES_IN_INDEX + 1];
{
unsigned num_columns = 0;
for (unsigned j = 0; j < fk.getParentColumnCount(); j++)
{
const NdbDictionary::Column* col =
table->getColumn(fk.getParentColumnNo(j));
DBUG_PRINT("col", ("[%u] %s", i, col->getName()));
if (!col)
{
error(NULL, "Could not find column '%s' in parent table '%s'",
fk.getParentColumnNo(j), table->getName());
continue;
}
col_names[num_columns] = col->getName();
col_types[num_columns] = col;
num_columns++;
}
col_names[num_columns]= 0;
col_types[num_columns] = 0;
if (num_columns != fk.getParentColumnCount())
{
error(NULL, "Could not find all columns referenced by fk in parent table '%s'",
table->getName());
continue;
}
}
db_guard.restore(); // restore db
// Create new mock
if (!create(dict, mock_name, child_name,
col_names, col_types))
{
error(dict, "Failed to create mock parent table '%s", mock_name);
DBUG_ASSERT(false);
DBUG_RETURN(false);
}
// Recreate fks to point at new mock
if (!copy_fk_to_new_parent(dict, fk, mock_name, col_names))
{
DBUG_RETURN(false);
}
fk_index++;
}
// Drop the requested table and all foreign keys refering to it
// i.e the old fks
const int drop_flags= NDBDICT::DropTableCascadeConstraints;
if (dict->dropTableGlobal(*table, drop_flags) != 0)
{
error(dict, "Failed to drop the requested table");
DBUG_RETURN(false);
}
DBUG_RETURN(true);
}
public:
Fk_util(THD* thd) : m_thd(thd) {}
static
bool split_mock_name(const char* name,
unsigned* child_id_ptr = NULL,
unsigned* child_index_ptr = NULL,
const char** parent_name = NULL)
{
const struct {
const char* str;
size_t len;
} prefix = { STRING_WITH_LEN("NDB$FKM_") };
if (strncmp(name, prefix.str, prefix.len) != 0)
return false;
char* end;
const char* ptr= name + prefix.len + 1;
// Parse child id
long child_id = strtol(ptr, &end, 10);
if (ptr == end || child_id < 0 || *end == 0 || *end != '_')
return false;
ptr = end+1;
// Parse child index
long child_index = strtol(ptr, &end, 10);
if (ptr == end || child_id < 0 || *end == 0 || *end != '_')
return false;
ptr = end+1;
// Assign and return OK
if (child_id_ptr)
*child_id_ptr = child_id;
if (child_index_ptr)
*child_index_ptr = child_index;
if (parent_name)
*parent_name = ptr;
return true;
}
static
bool is_mock_name(const char* name)
{
return split_mock_name(name);
}
static
const char* format_name(char buf[], size_t buf_size, int child_id,
uint fk_index, const char* parent_name)
{
DBUG_ENTER("format_name");
DBUG_PRINT("enter", ("child_id: %d, fk_index: %u, parent_name: %s",
child_id, fk_index, parent_name));
const size_t len = my_snprintf(buf, buf_size, "NDB$FKM_%d_%u_%s",
child_id, fk_index, parent_name);
DBUG_PRINT("info", ("len: %lu, buf_size: %lu", len, buf_size));
if (len >= buf_size - 1)
{
DBUG_PRINT("info", ("Size of buffer too small"));
DBUG_RETURN(NULL);
}
DBUG_PRINT("exit", ("buf: '%s', len: %lu", buf, len));
DBUG_RETURN(buf);
}
// Adaptor function for calling create() with List<key_part_spec>
bool create(NDBDICT *dict, const char* mock_name, const char* child_name,
List<Key_part_spec> key_part_list, const NDBCOL * col_types[])
{
// Convert List<Key_part_spec> into null terminated const char* array
const char* col_names[NDB_MAX_ATTRIBUTES_IN_INDEX + 1];
{
unsigned i = 0;
Key_part_spec* key = 0;
List_iterator<Key_part_spec> it1(key_part_list);
while ((key= it1++))
{
char col_name_buf[FN_REFLEN];
const char* col_name = lex2str(key->field_name, col_name_buf, sizeof(col_name_buf));
col_names[i++] = strdup(col_name);
}
col_names[i] = 0;
}
const bool ret = create(dict, mock_name, child_name, col_names, col_types);
// Free the strings in col_names array
for (unsigned i = 0; col_names[i] != 0; i++)
{
const char* col_name = col_names[i];
free(const_cast<char*>(col_name));
}
return ret;
}
bool create(NDBDICT *dict, const char* mock_name, const char* child_name,
const char* col_names[], const NDBCOL * col_types[])
{
NDBTAB mock_tab;
DBUG_ENTER("mock_table::create");
DBUG_PRINT("enter", ("mock_name: %s", mock_name));
DBUG_ASSERT(is_mock_name(mock_name));
if (mock_tab.setName(mock_name))
{
DBUG_RETURN(false);
}
mock_tab.setLogging(FALSE);
unsigned i = 0;
while (col_names[i])
{
NDBCOL mock_col;
const char* col_name = col_names[i];
DBUG_PRINT("info", ("name: %s", col_name));
if (mock_col.setName(col_name))
{
DBUG_ASSERT(false);
DBUG_RETURN(false);
}
const NDBCOL * col= col_types[i];
if (!col)
{
// Internal error, the two lists should be same size
DBUG_ASSERT(col);
DBUG_RETURN(false);
}
// Use column spec as requested(normally built from child table)
mock_col.setType(col->getType());
mock_col.setPrecision(col->getPrecision());
mock_col.setScale(col->getScale());
mock_col.setLength(col->getLength());
mock_col.setCharset(col->getCharset());
// Make column part of primary key and thus not nullable
mock_col.setPrimaryKey(true);
mock_col.setNullable(false);
if (mock_tab.addColumn(mock_col))
{
DBUG_RETURN(false);
}
i++;
}
// Create the table in NDB
if (dict->createTable(mock_tab) != 0)
{
// Error is available to caller in dict*
DBUG_RETURN(false);
}
info("Created mock table '%s' referenced by '%s'", mock_name, child_name);
DBUG_RETURN(true);
}
bool
build_mock_list(NdbDictionary::Dictionary* dict,
const NdbDictionary::Table* table, List<char> &mock_list)
{
DBUG_ENTER("build_mock_list");
NdbDictionary::Dictionary::List list;
if (dict->listDependentObjects(list, *table) != 0)
{
error(dict, "Failed to list dependent objects for table '%s'", table->getName());
DBUG_RETURN(false);
}
for (unsigned i = 0; i < list.count; i++)
{
const NdbDictionary::Dictionary::List::Element& element = list.elements[i];
if (element.type != NdbDictionary::Object::ForeignKey)
continue;
NdbDictionary::ForeignKey fk;
if (dict->getForeignKey(fk, element.name) != 0)
{
// Could not find the listed fk
DBUG_ASSERT(false);
continue;
}
char parent_db_and_name[FN_LEN + 1];
const char * name = fk_split_name(parent_db_and_name,fk.getParentTable());
if (!Fk_util::is_mock_name(name))
continue;
mock_list.push_back(thd_strdup(m_thd, fk.getParentTable()));
}
DBUG_RETURN(true);
}
void
drop_mock_list(Ndb* ndb, NdbDictionary::Dictionary* dict, List<char> &drop_list)
{
const char* full_name;
List_iterator_fast<char> it(drop_list);
while ((full_name=it++))
{
DBUG_PRINT("info", ("drop table: '%s'", full_name));
char db_name[FN_LEN + 1];
const char * table_name = fk_split_name(db_name, full_name);
Ndb_db_guard db_guard(ndb);
setDbName(ndb, db_name);
Ndb_table_guard mocktab_g(dict, table_name);
if (!mocktab_g.get_table())
{
// Could not open the mock table
DBUG_PRINT("error", ("Could not open the listed mock table, ignore it"));
DBUG_ASSERT(false);
continue;
}
if (dict->dropTableGlobal(*mocktab_g.get_table()) != 0)
{
DBUG_PRINT("error", ("Failed to drop the mock table '%s'",
mocktab_g.get_table()->getName()));
DBUG_ASSERT(false);
continue;
}
info("Dropped mock table '%s' - referencing table dropped", table_name);
}
}
bool
drop(Ndb* ndb, NdbDictionary::Dictionary* dict,
const NdbDictionary::Table* table)
{
DBUG_ENTER("drop");
// Start schema transaction to make this operation atomic
if (dict->beginSchemaTrans() != 0)
{
error(dict, "Failed to start schema transaction");
DBUG_RETURN(false);
}
bool result = true;
if (!create_mock_tables_and_drop(ndb, dict, table))
{
// Operation failed, set flag to abort when ending trans
result = false;
}
// End schema transaction
const Uint32 end_trans_flag = result ? 0 : NdbDictionary::Dictionary::SchemaTransAbort;
if (dict->endSchemaTrans(end_trans_flag) != 0)
{
error(dict, "Failed to end schema transaction");
result = false;
}
DBUG_RETURN(result);
}
bool count_fks(NdbDictionary::Dictionary* dict,
const NdbDictionary::Table* table, uint& count) const
{
DBUG_ENTER("count_fks");
NdbDictionary::Dictionary::List list;
if (dict->listDependentObjects(list, *table) != 0)
{
error(dict, "Failed to list dependent objects for table '%s'", table->getName());
DBUG_RETURN(false);
}
for (unsigned i = 0; i < list.count; i++)
{
if (list.elements[i].type == NdbDictionary::Object::ForeignKey)
count++;
}
DBUG_PRINT("exit", ("count: %u", count));
DBUG_RETURN(true);
}
bool drop_fk(Ndb* ndb, NdbDictionary::Dictionary* dict, const char* fk_name)
{
DBUG_ENTER("drop_fk");
NdbDictionary::ForeignKey fk;
if (dict->getForeignKey(fk, fk_name) != 0)
{
error(dict, "Could not find fk '%s'", fk_name);
DBUG_ASSERT(false);
DBUG_RETURN(false);
}
char parent_db_and_name[FN_LEN + 1];
const char * parent_name = fk_split_name(parent_db_and_name,fk.getParentTable());
if (Fk_util::is_mock_name(parent_name))
{
// Fk is referencing a mock table, drop the table
// and the constraint at the same time
Ndb_db_guard db_guard(ndb);
setDbName(ndb, parent_db_and_name);
Ndb_table_guard mocktab_g(dict, parent_name);
if (mocktab_g.get_table())