← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to drizzled/schema.cc

  • Committer: Daniel Nichter
  • Date: 2011-10-23 16:01:37 UTC
  • mto: This revision was merged to the branch mainline in revision 2448.
  • Revision ID: daniel@percona.com-20111023160137-7ac3blgz8z4tf8za
Add Administration Getting Started and Logging.  Capitalize SQL clause keywords.

Show diffs side-by-side

added added

removed removed

Lines of Context:
drizzled/schema.cc
11
11
 
12
12
   You should have received a copy of the GNU General Public License
13
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 */
 
14
   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA */
15
15
 
16
16
 
17
17
/* create and drop of databases */
18
 
#include <drizzled/global.h>
19
 
#include CSTDINT_H
20
 
#include CINTTYPES_H
 
18
#include <config.h>
 
19
 
 
20
#include <fcntl.h>
 
21
#include <sys/stat.h>
 
22
#include <sys/types.h>
 
23
 
 
24
#include <set>
21
25
#include <string>
22
26
#include <fstream>
23
 
#include <drizzled/serialize/serialize.h>
24
 
using namespace std;
25
 
#include <drizzled/server_includes.h>
26
 
#include <mysys/mysys_err.h>
27
 
#include <mysys/my_dir.h>
28
 
#include "log.h"
 
27
 
29
28
#include <drizzled/error.h>
30
29
#include <drizzled/gettext.h>
31
 
#include <mysys/hash.h>
 
30
#include <drizzled/internal/m_string.h>
32
31
#include <drizzled/session.h>
33
 
#include <drizzled/db.h>
 
32
#include <drizzled/schema.h>
34
33
#include <drizzled/sql_base.h>
35
 
 
36
 
#define MAX_DROP_TABLE_Q_LEN      1024
37
 
 
38
 
const char *del_exts[]= {".frm", ".BAK", ".TMD",".opt", NULL};
39
 
static TYPELIB deletable_extentions=
40
 
{array_elements(del_exts)-1,"del_exts", del_exts, NULL};
41
 
 
42
 
static long mysql_rm_known_files(Session *session, MY_DIR *dirp,
43
 
                                 const char *db, const char *path,
44
 
                                 uint32_t level,
45
 
                                 TableList **dropped_tables);
46
 
 
47
 
static bool rm_dir_w_symlink(const char *org_path, bool send_error);
48
 
static void mysql_change_db_impl(Session *session,
49
 
                                 LEX_STRING *new_db_name,
50
 
                                 const CHARSET_INFO * const new_db_charset);
51
 
 
52
 
 
53
 
/* Database lock hash */
54
 
HASH lock_db_cache;
55
 
pthread_mutex_t LOCK_lock_db;
56
 
int creating_database= 0;  // how many database locks are made
57
 
 
58
 
 
59
 
/* Structure for database lock */
60
 
typedef struct my_dblock_st
61
 
{
62
 
  char *name;        /* Database name        */
63
 
  uint32_t name_length;  /* Database length name */
64
 
} my_dblock_t;
65
 
 
66
 
 
67
 
/*
68
 
  lock_db key.
69
 
*/
70
 
 
71
 
extern "C" unsigned char* lock_db_get_key(my_dblock_t *, size_t *, bool not_used);
72
 
 
73
 
unsigned char* lock_db_get_key(my_dblock_t *ptr, size_t *length,
74
 
                       bool not_used __attribute__((unused)))
75
 
{
76
 
  *length= ptr->name_length;
77
 
  return (unsigned char*) ptr->name;
78
 
}
79
 
 
80
 
 
81
 
/*
82
 
  Free lock_db hash element.
83
 
*/
84
 
 
85
 
extern "C" void lock_db_free_element(void *ptr);
86
 
 
87
 
void lock_db_free_element(void *ptr)
88
 
{
89
 
  free(ptr);
90
 
}
91
 
 
92
 
 
93
 
/*
94
 
  Delete a database lock entry from hash.
95
 
*/
96
 
 
97
 
void lock_db_delete(const char *name, uint32_t length)
98
 
{
99
 
  my_dblock_t *opt;
100
 
  safe_mutex_assert_owner(&LOCK_lock_db);
101
 
  if ((opt= (my_dblock_t *)hash_search(&lock_db_cache,
102
 
                                       (const unsigned char*) name, length)))
103
 
    hash_delete(&lock_db_cache, (unsigned char*) opt);
104
 
}
105
 
 
106
 
 
107
 
/* Database options hash */
108
 
static HASH dboptions;
109
 
static bool dboptions_init= 0;
110
 
static rw_lock_t LOCK_dboptions;
111
 
 
112
 
/* Structure for database options */
113
 
typedef struct my_dbopt_st
114
 
{
115
 
  char *name;                   /* Database name                  */
116
 
  uint32_t name_length;         /* Database length name           */
117
 
  const CHARSET_INFO *charset;  /* Database default character set */
118
 
} my_dbopt_t;
119
 
 
120
 
 
121
 
/*
122
 
  Function we use in the creation of our hash to get key.
123
 
*/
124
 
 
125
 
extern "C" unsigned char* dboptions_get_key(my_dbopt_t *opt, size_t *length,
126
 
                                    bool not_used);
127
 
 
128
 
unsigned char* dboptions_get_key(my_dbopt_t *opt, size_t *length,
129
 
                         bool not_used __attribute__((unused)))
130
 
{
131
 
  *length= opt->name_length;
132
 
  return (unsigned char*) opt->name;
133
 
}
134
 
 
135
 
 
136
 
/*
137
 
  Helper function to write a query to binlog used by mysql_rm_db()
138
 
*/
139
 
 
140
 
static inline void write_to_binlog(Session *session, char *query, uint32_t q_len,
141
 
                                   char *db, uint32_t db_len)
142
 
{
143
 
  Query_log_event qinfo(session, query, q_len, 0, 0);
144
 
  qinfo.error_code= 0;
145
 
  qinfo.db= db;
146
 
  qinfo.db_len= db_len;
147
 
  drizzle_bin_log.write(&qinfo);
148
 
}  
149
 
 
150
 
 
151
 
/*
152
 
  Function to free dboptions hash element
153
 
*/
154
 
 
155
 
extern "C" void free_dbopt(void *dbopt);
156
 
 
157
 
void free_dbopt(void *dbopt)
158
 
{
159
 
  free((unsigned char*) dbopt);
160
 
}
161
 
 
162
 
 
163
 
/* 
164
 
  Initialize database option hash and locked database hash.
165
 
 
166
 
  SYNOPSIS
167
 
    my_database_names()
168
 
 
169
 
  NOTES
170
 
    Must be called before any other database function is called.
171
 
 
172
 
  RETURN
173
 
    0   ok
174
 
    1   Fatal error
175
 
*/
176
 
 
177
 
bool my_database_names_init(void)
178
 
{
179
 
  bool error= false;
180
 
  (void) my_rwlock_init(&LOCK_dboptions, NULL);
181
 
  if (!dboptions_init)
182
 
  {
183
 
    dboptions_init= 1;
184
 
    error= hash_init(&dboptions, lower_case_table_names ? 
185
 
                     &my_charset_bin : system_charset_info,
186
 
                     32, 0, 0, (hash_get_key) dboptions_get_key,
187
 
                     free_dbopt,0) ||
188
 
           hash_init(&lock_db_cache, lower_case_table_names ? 
189
 
                     &my_charset_bin : system_charset_info,
190
 
                     32, 0, 0, (hash_get_key) lock_db_get_key,
191
 
                     lock_db_free_element,0);
192
 
 
193
 
  }
194
 
  return error;
195
 
}
196
 
 
197
 
 
198
 
 
199
 
/* 
200
 
  Free database option hash and locked databases hash.
201
 
*/
202
 
 
203
 
void my_database_names_free(void)
204
 
{
205
 
  if (dboptions_init)
206
 
  {
207
 
    dboptions_init= 0;
208
 
    hash_free(&dboptions);
209
 
    (void) rwlock_destroy(&LOCK_dboptions);
210
 
    hash_free(&lock_db_cache);
211
 
  }
212
 
}
213
 
 
214
 
 
215
 
/*
216
 
  Cleanup cached options
217
 
*/
218
 
 
219
 
void my_dbopt_cleanup(void)
220
 
{
221
 
  rw_wrlock(&LOCK_dboptions);
222
 
  hash_free(&dboptions);
223
 
  hash_init(&dboptions, lower_case_table_names ? 
224
 
            &my_charset_bin : system_charset_info,
225
 
            32, 0, 0, (hash_get_key) dboptions_get_key,
226
 
            free_dbopt,0);
227
 
  rw_unlock(&LOCK_dboptions);
228
 
}
229
 
 
230
 
 
231
 
/*
232
 
  Find database options in the hash.
233
 
  
234
 
  DESCRIPTION
235
 
    Search a database options in the hash, usings its path.
236
 
    Fills "create" on success.
237
 
  
238
 
  RETURN VALUES
239
 
    0 on success.
240
 
    1 on error.
241
 
*/
242
 
 
243
 
static bool get_dbopt(const char *dbname, HA_CREATE_INFO *create)
244
 
{
245
 
  my_dbopt_t *opt;
246
 
  uint32_t length;
247
 
  bool error= true;
248
 
  
249
 
  length= (uint) strlen(dbname);
250
 
  
251
 
  rw_rdlock(&LOCK_dboptions);
252
 
  if ((opt= (my_dbopt_t*) hash_search(&dboptions, (unsigned char*) dbname, length)))
253
 
  {
254
 
    create->default_table_charset= opt->charset;
255
 
    error= true;
256
 
  }
257
 
  rw_unlock(&LOCK_dboptions);
258
 
  return error;
259
 
}
260
 
 
261
 
 
262
 
/*
263
 
  Writes database options into the hash.
264
 
  
265
 
  DESCRIPTION
266
 
    Inserts database options into the hash, or updates
267
 
    options if they are already in the hash.
268
 
  
269
 
  RETURN VALUES
270
 
    0 on success.
271
 
    1 on error.
272
 
*/
273
 
 
274
 
static bool put_dbopt(const char *dbname, HA_CREATE_INFO *create)
275
 
{
276
 
  my_dbopt_t *opt;
277
 
  uint32_t length;
278
 
  bool error= false;
279
 
 
280
 
  length= (uint) strlen(dbname);
281
 
  
282
 
  rw_wrlock(&LOCK_dboptions);
283
 
  if (!(opt= (my_dbopt_t*) hash_search(&dboptions, (unsigned char*) dbname, length)))
284
 
  { 
285
 
    /* Options are not in the hash, insert them */
286
 
    char *tmp_name;
287
 
    if (!my_multi_malloc(MYF(MY_WME | MY_ZEROFILL),
288
 
                         &opt, (uint) sizeof(*opt), &tmp_name, (uint) length+1,
289
 
                         NULL))
290
 
    {
291
 
      error= true;
292
 
      goto end;
293
 
    }
294
 
    
295
 
    opt->name= tmp_name;
296
 
    my_stpcpy(opt->name, dbname);
297
 
    opt->name_length= length;
298
 
    
299
 
    if ((error= my_hash_insert(&dboptions, (unsigned char*) opt)))
300
 
    {
301
 
      free(opt);
302
 
      goto end;
303
 
    }
304
 
  }
305
 
 
306
 
  /* Update / write options in hash */
307
 
  opt->charset= create->default_table_charset;
308
 
 
309
 
end:
310
 
  rw_unlock(&LOCK_dboptions);  
311
 
  return(error);
312
 
}
313
 
 
314
 
 
315
 
/*
316
 
  Deletes database options from the hash.
317
 
*/
318
 
 
319
 
void del_dbopt(const char *path)
320
 
{
321
 
  my_dbopt_t *opt;
322
 
  rw_wrlock(&LOCK_dboptions);
323
 
  if ((opt= (my_dbopt_t *)hash_search(&dboptions, (const unsigned char*) path,
324
 
                                      strlen(path))))
325
 
    hash_delete(&dboptions, (unsigned char*) opt);
326
 
  rw_unlock(&LOCK_dboptions);
327
 
}
328
 
 
329
 
 
330
 
/*
331
 
  Create database options file:
332
 
 
333
 
  DESCRIPTION
334
 
    Currently database default charset is only stored there.
335
 
 
336
 
  RETURN VALUES
337
 
  0     ok
338
 
  1     Could not create file or write to it.  Error sent through my_error()
339
 
*/
340
 
 
341
 
static bool write_db_opt(Session *session, const char *path, const char *name, HA_CREATE_INFO *create)
342
 
{
343
 
  bool error= true;
344
 
  drizzle::Schema db;
345
 
 
346
 
  assert(name);
347
 
 
348
 
  if (!create->default_table_charset)
349
 
    create->default_table_charset= session->variables.collation_server;
350
 
 
351
 
  if (put_dbopt(path, create))
352
 
    return 1;
353
 
 
354
 
  db.set_name(name);
355
 
  db.set_characterset(create->default_table_charset->csname);
356
 
  db.set_collation(create->default_table_charset->name);
357
 
 
358
 
  fstream output(path, ios::out | ios::trunc | ios::binary);
359
 
  if (!db.SerializeToOstream(&output)) 
360
 
    error= false;
361
 
 
362
 
  return error;
363
 
}
364
 
 
365
 
 
366
 
/*
367
 
  Load database options file
368
 
 
369
 
  load_db_opt()
370
 
  path          Path for option file
371
 
  create        Where to store the read options
372
 
 
373
 
  DESCRIPTION
374
 
 
375
 
  RETURN VALUES
376
 
  0     File found
377
 
  1     No database file or could not open it
378
 
 
379
 
*/
380
 
 
381
 
bool load_db_opt(Session *session, const char *path, HA_CREATE_INFO *create)
382
 
{
383
 
  bool error=1;
384
 
  drizzle::Schema db;
385
 
  string buffer;
386
 
 
387
 
  memset(create, 0, sizeof(*create));
388
 
  create->default_table_charset= session->variables.collation_server;
389
 
 
390
 
  /* Check if options for this database are already in the hash */
391
 
  if (!get_dbopt(path, create))
392
 
    return(0);
393
 
 
394
 
  fstream input(path, ios::in | ios::binary);
395
 
  if (!input)
396
 
    goto err1;
397
 
  else if (!db.ParseFromIstream(&input))
398
 
    goto err1;
399
 
 
400
 
  buffer= db.characterset();
401
 
  if (!(create->default_table_charset= get_charset_by_csname(buffer.c_str(), MY_CS_PRIMARY, MYF(0))))
402
 
  {
403
 
    sql_print_error(_("Error while loading database options: '%s':"),path);
404
 
    sql_print_error(ER(ER_UNKNOWN_COLLATION), buffer.c_str());
405
 
    create->default_table_charset= default_charset_info;
406
 
  }
407
 
 
408
 
  buffer= db.collation();
409
 
  if (!(create->default_table_charset= get_charset_by_name(buffer.c_str(), MYF(0))))
410
 
  {
411
 
    sql_print_error(_("Error while loading database options: '%s':"),path);
412
 
    sql_print_error(ER(ER_UNKNOWN_COLLATION), buffer.c_str());
413
 
    create->default_table_charset= default_charset_info;
414
 
  }
415
 
 
416
 
  /*
417
 
    Put the loaded value into the hash.
418
 
    Note that another thread could've added the same
419
 
    entry to the hash after we called get_dbopt(),
420
 
    but it's not an error, as put_dbopt() takes this
421
 
    possibility into account.
422
 
  */
423
 
  error= put_dbopt(path, create);
424
 
 
425
 
err1:
426
 
  return(error);
427
 
}
428
 
 
429
 
 
430
 
/*
431
 
  Retrieve database options by name. Load database options file or fetch from
432
 
  cache.
433
 
 
434
 
  SYNOPSIS
435
 
    load_db_opt_by_name()
436
 
    db_name         Database name
437
 
    db_create_info  Where to store the database options
438
 
 
439
 
  DESCRIPTION
440
 
    load_db_opt_by_name() is a shortcut for load_db_opt().
441
 
 
442
 
  NOTE
443
 
    Although load_db_opt_by_name() (and load_db_opt()) returns status of
444
 
    the operation, it is useless usually and should be ignored. The problem
445
 
    is that there are 1) system databases ("mysql") and 2) virtual
446
 
    databases ("information_schema"), which do not contain options file.
447
 
    So, load_db_opt[_by_name]() returns false for these databases, but this
448
 
    is not an error.
449
 
 
450
 
    load_db_opt[_by_name]() clears db_create_info structure in any case, so
451
 
    even on failure it contains valid data. So, common use case is just
452
 
    call load_db_opt[_by_name]() without checking return value and use
453
 
    db_create_info right after that.
454
 
 
455
 
  RETURN VALUES (read NOTE!)
456
 
    false   Success
457
 
    true    Failed to retrieve options
458
 
*/
459
 
 
460
 
bool load_db_opt_by_name(Session *session, const char *db_name,
461
 
                         HA_CREATE_INFO *db_create_info)
462
 
{
463
 
  char db_opt_path[FN_REFLEN];
464
 
 
465
 
  /*
466
 
    Pass an empty file name, and the database options file name as extension
467
 
    to avoid table name to file name encoding.
468
 
  */
469
 
  (void) build_table_filename(db_opt_path, sizeof(db_opt_path),
470
 
                              db_name, "", MY_DB_OPT_FILE, 0);
471
 
 
472
 
  return load_db_opt(session, db_opt_path, db_create_info);
473
 
}
474
 
 
475
 
 
476
 
/**
477
 
  Return default database collation.
478
 
 
479
 
  @param session     Thread context.
480
 
  @param db_name Database name.
481
 
 
482
 
  @return CHARSET_INFO object. The operation always return valid character
483
 
    set, even if the database does not exist.
484
 
*/
485
 
 
486
 
const CHARSET_INFO *get_default_db_collation(Session *session, const char *db_name)
487
 
{
488
 
  HA_CREATE_INFO db_info;
489
 
 
490
 
  if (session->db != NULL && strcmp(db_name, session->db) == 0)
491
 
    return session->db_charset;
492
 
 
493
 
  load_db_opt_by_name(session, db_name, &db_info);
494
 
 
495
 
  /*
496
 
    NOTE: even if load_db_opt_by_name() fails,
497
 
    db_info.default_table_charset contains valid character set
498
 
    (collation_server). We should not fail if load_db_opt_by_name() fails,
499
 
    because it is valid case. If a database has been created just by
500
 
    "mkdir", it does not contain db.opt file, but it is valid database.
501
 
  */
502
 
 
503
 
  return db_info.default_table_charset;
504
 
}
505
 
 
 
34
#include <drizzled/lock.h>
 
35
#include <drizzled/errmsg_print.h>
 
36
#include <drizzled/transaction_services.h>
 
37
#include <drizzled/message/schema.pb.h>
 
38
#include <drizzled/sql_table.h>
 
39
#include <drizzled/plugin/storage_engine.h>
 
40
#include <drizzled/plugin/authorization.h>
 
41
#include <drizzled/pthread_globals.h>
 
42
#include <drizzled/charset.h>
 
43
#include <drizzled/internal/my_sys.h>
 
44
#include <drizzled/catalog/instance.h>
 
45
#include <boost/thread/mutex.hpp>
 
46
 
 
47
using namespace std;
 
48
 
 
49
namespace drizzled {
 
50
namespace schema {
506
51
 
507
52
/*
508
53
  Create a database
509
54
 
510
55
  SYNOPSIS
511
 
  mysql_create_db()
 
56
  create_db()
512
57
  session               Thread handler
513
58
  db            Name of database to create
514
59
                Function assumes that this is already validated.
515
60
  create_info   Database create options (like character set)
516
 
  silent        Used by replication when internally creating a database.
517
 
                In this case the entry should not be logged.
518
61
 
519
62
  SIDE-EFFECTS
520
63
   1. Report back to client that command succeeded (my_ok)
521
64
   2. Report errors to client
522
65
   3. Log event to binary log
523
 
   (The 'silent' flags turns off 1 and 3.)
524
66
 
525
67
  RETURN VALUES
526
68
  false ok
528
70
 
529
71
*/
530
72
 
531
 
int mysql_create_db(Session *session, char *db, HA_CREATE_INFO *create_info, bool silent)
 
73
bool create(Session &session, const message::Schema &schema_message, const bool is_if_not_exists)
532
74
{
533
 
  char   path[FN_REFLEN+16];
534
 
  char   tmp_query[FN_REFLEN+16];
535
 
  long result= 1;
536
 
  int error= 0;
537
 
  struct stat stat_info;
538
 
  uint32_t create_options= create_info ? create_info->options : 0;
539
 
  uint32_t path_len;
540
 
 
541
 
  /* do not create 'information_schema' db */
542
 
  if (!my_strcasecmp(system_charset_info, db, INFORMATION_SCHEMA_NAME.c_str()))
543
 
  {
544
 
    my_error(ER_DB_CREATE_EXISTS, MYF(0), db);
545
 
    return(-1);
546
 
  }
 
75
  bool error= false;
547
76
 
548
77
  /*
549
78
    Do not create database if another thread is holding read lock.
550
 
    Wait for global read lock before acquiring LOCK_drizzle_create_db.
 
79
    Wait for global read lock before acquiring session->catalog()->schemaLock().
551
80
    After wait_if_global_read_lock() we have protection against another
552
 
    global read lock. If we would acquire LOCK_drizzle_create_db first,
 
81
    global read lock. If we would acquire session->catalog()->schemaLock() first,
553
82
    another thread could step in and get the global read lock before we
554
83
    reach wait_if_global_read_lock(). If this thread tries the same as we
555
 
    (admin a db), it would then go and wait on LOCK_drizzle_create_db...
 
84
    (admin a db), it would then go and wait on session->catalog()->schemaLock()...
556
85
    Furthermore wait_if_global_read_lock() checks if the current thread
557
86
    has the global read lock and refuses the operation with
558
87
    ER_CANT_UPDATE_WITH_READLOCK if applicable.
559
88
  */
560
 
  if (wait_if_global_read_lock(session, 0, 1))
561
 
  {
562
 
    error= -1;
563
 
    goto exit2;
564
 
  }
565
 
 
566
 
  pthread_mutex_lock(&LOCK_drizzle_create_db);
567
 
 
568
 
  /* Check directory */
569
 
  path_len= build_table_filename(path, sizeof(path), db, "", "", 0);
570
 
  path[path_len-1]= 0;                    // Remove last '/' from path
571
 
 
572
 
  if (!stat(path,&stat_info))
573
 
  {
574
 
    if (!(create_options & HA_LEX_CREATE_IF_NOT_EXISTS))
575
 
    {
576
 
      my_error(ER_DB_CREATE_EXISTS, MYF(0), db);
577
 
      error= -1;
578
 
      goto exit;
579
 
    }
580
 
    push_warning_printf(session, DRIZZLE_ERROR::WARN_LEVEL_NOTE,
581
 
                        ER_DB_CREATE_EXISTS, ER(ER_DB_CREATE_EXISTS), db);
582
 
    if (!silent)
583
 
      my_ok(session);
584
 
    error= 0;
585
 
    goto exit;
586
 
  }
587
 
  else
588
 
  {
589
 
    if (errno != ENOENT)
590
 
    {
591
 
      my_error(EE_STAT, MYF(0), path, errno);
592
 
      goto exit;
593
 
    }
594
 
    if (my_mkdir(path,0777,MYF(0)) < 0)
595
 
    {
596
 
      my_error(ER_CANT_CREATE_DB, MYF(0), db, my_errno);
597
 
      error= -1;
598
 
      goto exit;
599
 
    }
600
 
  }
601
 
 
602
 
  path[path_len-1]= FN_LIBCHAR;
603
 
  strmake(path+path_len, MY_DB_OPT_FILE, sizeof(path)-path_len-1);
604
 
  if (write_db_opt(session, path, db, create_info))
605
 
  {
606
 
    /*
607
 
      Could not create options file.
608
 
      Restore things to beginning.
609
 
    */
610
 
    path[path_len]= 0;
611
 
    if (rmdir(path) >= 0)
612
 
    {
613
 
      error= -1;
614
 
      goto exit;
615
 
    }
616
 
    /*
617
 
      We come here when we managed to create the database, but not the option
618
 
      file.  In this case it's best to just continue as if nothing has
619
 
      happened.  (This is a very unlikely senario)
620
 
    */
621
 
  }
622
 
  
623
 
  if (!silent)
624
 
  {
625
 
    char *query;
626
 
    uint32_t query_length;
627
 
 
628
 
    if (!session->query)                                // Only in replication
629
 
    {
630
 
      query=         tmp_query;
631
 
      query_length= (uint) (strxmov(tmp_query,"create database `",
632
 
                                    db, "`", NULL) - tmp_query);
633
 
    }
634
 
    else
635
 
    {
636
 
      query=        session->query;
637
 
      query_length= session->query_length;
638
 
    }
639
 
 
640
 
    if (drizzle_bin_log.is_open())
641
 
    {
642
 
      Query_log_event qinfo(session, query, query_length, 0, 
643
 
                            /* suppress_use */ true);
644
 
 
645
 
      /*
646
 
        Write should use the database being created as the "current
647
 
        database" and not the threads current database, which is the
648
 
        default. If we do not change the "current database" to the
649
 
        database being created, the CREATE statement will not be
650
 
        replicated when using --binlog-do-db to select databases to be
651
 
        replicated. 
652
 
 
653
 
        An example (--binlog-do-db=sisyfos):
654
 
       
655
 
          CREATE DATABASE bob;        # Not replicated
656
 
          USE bob;                    # 'bob' is the current database
657
 
          CREATE DATABASE sisyfos;    # Not replicated since 'bob' is
658
 
                                      # current database.
659
 
          USE sisyfos;                # Will give error on slave since
660
 
                                      # database does not exist.
661
 
      */
662
 
      qinfo.db     = db;
663
 
      qinfo.db_len = strlen(db);
664
 
 
665
 
      /* These DDL methods and logging protected with LOCK_drizzle_create_db */
666
 
      drizzle_bin_log.write(&qinfo);
667
 
    }
668
 
    my_ok(session, result);
669
 
  }
670
 
 
671
 
exit:
672
 
  pthread_mutex_unlock(&LOCK_drizzle_create_db);
673
 
  start_waiting_global_read_lock(session);
674
 
exit2:
675
 
  return(error);
 
89
  if (session.wait_if_global_read_lock(false, true))
 
90
  {
 
91
    return false;
 
92
  }
 
93
 
 
94
  assert(schema_message.has_name());
 
95
  assert(schema_message.has_collation());
 
96
 
 
97
  // @todo push this lock down into the engine
 
98
  {
 
99
    boost::mutex::scoped_lock scopedLock(session.catalog().schemaLock());
 
100
 
 
101
    // Check to see if it exists already.  
 
102
    identifier::Schema schema_identifier(schema_message.name());
 
103
    if (plugin::StorageEngine::doesSchemaExist(schema_identifier))
 
104
    {
 
105
      if (not is_if_not_exists)
 
106
      {
 
107
        my_error(ER_DB_CREATE_EXISTS, schema_identifier);
 
108
        error= true;
 
109
      }
 
110
      else
 
111
      {
 
112
        push_warning_printf(&session, DRIZZLE_ERROR::WARN_LEVEL_NOTE,
 
113
                            ER_DB_CREATE_EXISTS, ER(ER_DB_CREATE_EXISTS),
 
114
                            schema_message.name().c_str());
 
115
        session.my_ok();
 
116
      }
 
117
    }
 
118
    else if (not plugin::StorageEngine::createSchema(schema_message)) // Try to create it 
 
119
    {
 
120
      my_error(ER_CANT_CREATE_DB, MYF(0), schema_message.name().c_str(), errno);
 
121
      error= true;
 
122
    }
 
123
    else // Created !
 
124
    {
 
125
      TransactionServices::createSchema(session, schema_message);
 
126
      session.my_ok(1);
 
127
    }
 
128
  }
 
129
  session.startWaitingGlobalReadLock();
 
130
 
 
131
  return error;
676
132
}
677
133
 
678
134
 
679
135
/* db-name is already validated when we come here */
680
136
 
681
 
bool mysql_alter_db(Session *session, const char *db, HA_CREATE_INFO *create_info)
 
137
bool alter(Session &session,
 
138
           const message::Schema &schema_message,
 
139
           const message::Schema &original_schema)
682
140
{
683
 
  char path[FN_REFLEN+16];
684
 
  long result=1;
685
 
  int error= false;
686
 
 
687
141
  /*
688
142
    Do not alter database if another thread is holding read lock.
689
 
    Wait for global read lock before acquiring LOCK_drizzle_create_db.
 
143
    Wait for global read lock before acquiring session->catalog()->schemaLock().
690
144
    After wait_if_global_read_lock() we have protection against another
691
 
    global read lock. If we would acquire LOCK_drizzle_create_db first,
 
145
    global read lock. If we would acquire session->catalog()->schemaLock() first,
692
146
    another thread could step in and get the global read lock before we
693
147
    reach wait_if_global_read_lock(). If this thread tries the same as we
694
 
    (admin a db), it would then go and wait on LOCK_drizzle_create_db...
 
148
    (admin a db), it would then go and wait on session->catalog()->schemaLock()...
695
149
    Furthermore wait_if_global_read_lock() checks if the current thread
696
150
    has the global read lock and refuses the operation with
697
151
    ER_CANT_UPDATE_WITH_READLOCK if applicable.
698
152
  */
699
 
  if ((error=wait_if_global_read_lock(session,0,1)))
700
 
    goto exit2;
701
 
 
702
 
  pthread_mutex_lock(&LOCK_drizzle_create_db);
703
 
 
704
 
  /* 
705
 
     Recreate db options file: /dbpath/.db.opt
706
 
     We pass MY_DB_OPT_FILE as "extension" to avoid
707
 
     "table name to file name" encoding.
708
 
  */
709
 
  build_table_filename(path, sizeof(path), db, "", MY_DB_OPT_FILE, 0);
710
 
  if ((error=write_db_opt(session, path, db, create_info)))
711
 
    goto exit;
712
 
 
713
 
  /* Change options if current database is being altered. */
714
 
 
715
 
  if (session->db && !strcmp(session->db,db))
716
 
  {
717
 
    session->db_charset= create_info->default_table_charset ?
718
 
                     create_info->default_table_charset :
719
 
                     session->variables.collation_server;
720
 
    session->variables.collation_database= session->db_charset;
721
 
  }
722
 
 
723
 
  if (drizzle_bin_log.is_open())
724
 
  {
725
 
    Query_log_event qinfo(session, session->query, session->query_length, 0,
726
 
                          /* suppress_use */ true);
727
 
 
728
 
    /*
729
 
      Write should use the database being created as the "current
730
 
      database" and not the threads current database, which is the
731
 
      default.
732
 
    */
733
 
    qinfo.db     = db;
734
 
    qinfo.db_len = strlen(db);
735
 
 
736
 
    session->clear_error();
737
 
    /* These DDL methods and logging protected with LOCK_drizzle_create_db */
738
 
    drizzle_bin_log.write(&qinfo);
739
 
  }
740
 
  my_ok(session, result);
741
 
 
742
 
exit:
743
 
  pthread_mutex_unlock(&LOCK_drizzle_create_db);
744
 
  start_waiting_global_read_lock(session);
745
 
exit2:
746
 
  return(error);
 
153
  if ((session.wait_if_global_read_lock(false, true)))
 
154
    return false;
 
155
 
 
156
  bool success;
 
157
  {
 
158
    boost::mutex::scoped_lock scopedLock(session.catalog().schemaLock());
 
159
 
 
160
    identifier::Schema schema_idenifier(schema_message.name());
 
161
    if (not plugin::StorageEngine::doesSchemaExist(schema_idenifier))
 
162
    {
 
163
      my_error(ER_SCHEMA_DOES_NOT_EXIST, schema_idenifier);
 
164
      return false;
 
165
    }
 
166
 
 
167
    /* Change options if current database is being altered. */
 
168
    success= plugin::StorageEngine::alterSchema(schema_message);
 
169
 
 
170
    if (success)
 
171
    {
 
172
      TransactionServices::alterSchema(session, original_schema, schema_message);
 
173
      session.my_ok(1);
 
174
    }
 
175
    else
 
176
    {
 
177
      my_error(ER_ALTER_SCHEMA, schema_idenifier);
 
178
    }
 
179
  }
 
180
  session.startWaitingGlobalReadLock();
 
181
 
 
182
  return success;
747
183
}
748
184
 
749
185
 
751
187
  Drop all tables in a database and the database itself
752
188
 
753
189
  SYNOPSIS
754
 
    mysql_rm_db()
 
190
    rm_db()
755
191
    session                     Thread handle
756
192
    db                  Database name in the case given by user
757
193
                        It's already validated and set to lower case
764
200
    ERROR Error
765
201
*/
766
202
 
767
 
bool mysql_rm_db(Session *session,char *db,bool if_exists, bool silent)
 
203
bool drop(Session &session, const identifier::Schema &schema_identifier, bool if_exists)
768
204
{
769
 
  long deleted=0;
770
 
  int error= false;
771
 
  char  path[FN_REFLEN+16];
772
 
  MY_DIR *dirp;
773
 
  uint32_t length;
774
 
  TableList* dropped_tables= 0;
775
 
 
776
 
  if (db && (strcmp(db, "information_schema") == 0))
777
 
  {
778
 
    my_error(ER_DBACCESS_DENIED_ERROR, MYF(0), "", "", INFORMATION_SCHEMA_NAME.c_str());
779
 
    return(true);
780
 
  }
781
 
 
782
205
  /*
783
206
    Do not drop database if another thread is holding read lock.
784
 
    Wait for global read lock before acquiring LOCK_drizzle_create_db.
 
207
    Wait for global read lock before acquiring session->catalog()->schemaLock().
785
208
    After wait_if_global_read_lock() we have protection against another
786
 
    global read lock. If we would acquire LOCK_drizzle_create_db first,
 
209
    global read lock. If we would acquire session->catalog()->schemaLock() first,
787
210
    another thread could step in and get the global read lock before we
788
211
    reach wait_if_global_read_lock(). If this thread tries the same as we
789
 
    (admin a db), it would then go and wait on LOCK_drizzle_create_db...
 
212
    (admin a db), it would then go and wait on session->catalog()->schemaLock()...
790
213
    Furthermore wait_if_global_read_lock() checks if the current thread
791
214
    has the global read lock and refuses the operation with
792
215
    ER_CANT_UPDATE_WITH_READLOCK if applicable.
793
216
  */
794
 
  if (wait_if_global_read_lock(session, 0, 1))
795
 
  {
796
 
    error= -1;
797
 
    goto exit2;
798
 
  }
799
 
 
800
 
  pthread_mutex_lock(&LOCK_drizzle_create_db);
801
 
 
802
 
  length= build_table_filename(path, sizeof(path), db, "", "", 0);
803
 
  my_stpcpy(path+length, MY_DB_OPT_FILE);               // Append db option file name
804
 
  del_dbopt(path);                              // Remove dboption hash entry
805
 
  path[length]= '\0';                           // Remove file name
806
 
 
807
 
  /* See if the directory exists */
808
 
  if (!(dirp= my_dir(path,MYF(MY_DONT_SORT))))
809
 
  {
810
 
    if (!if_exists)
811
 
    {
812
 
      error= -1;
813
 
      my_error(ER_DB_DROP_EXISTS, MYF(0), db);
814
 
      goto exit;
815
 
    }
816
 
    else
817
 
      push_warning_printf(session, DRIZZLE_ERROR::WARN_LEVEL_NOTE,
818
 
                          ER_DB_DROP_EXISTS, ER(ER_DB_DROP_EXISTS), db);
819
 
  }
820
 
  else
821
 
  {
822
 
    pthread_mutex_lock(&LOCK_open);
823
 
    remove_db_from_cache(db);
824
 
    pthread_mutex_unlock(&LOCK_open);
825
 
 
826
 
    
827
 
    error= -1;
828
 
    if ((deleted= mysql_rm_known_files(session, dirp, db, path, 0,
829
 
                                       &dropped_tables)) >= 0)
830
 
    {
831
 
      ha_drop_database(path);
832
 
      error = 0;
833
 
    }
834
 
  }
835
 
  if (!silent && deleted>=0)
836
 
  {
837
 
    const char *query;
838
 
    uint32_t query_length;
839
 
    if (!session->query)
840
 
    {
841
 
      /* The client used the old obsolete mysql_drop_db() call */
842
 
      query= path;
843
 
      query_length= (uint) (strxmov(path, "drop database `", db, "`",
844
 
                                     NULL) - path);
845
 
    }
846
 
    else
847
 
    {
848
 
      query =session->query;
849
 
      query_length= session->query_length;
850
 
    }
851
 
    if (drizzle_bin_log.is_open())
852
 
    {
853
 
      Query_log_event qinfo(session, query, query_length, 0, 
854
 
                            /* suppress_use */ true);
855
 
      /*
856
 
        Write should use the database being created as the "current
857
 
        database" and not the threads current database, which is the
858
 
        default.
859
 
      */
860
 
      qinfo.db     = db;
861
 
      qinfo.db_len = strlen(db);
862
 
 
863
 
      session->clear_error();
864
 
      /* These DDL methods and logging protected with LOCK_drizzle_create_db */
865
 
      drizzle_bin_log.write(&qinfo);
866
 
    }
867
 
    session->clear_error();
868
 
    session->server_status|= SERVER_STATUS_DB_DROPPED;
869
 
    my_ok(session, (uint32_t) deleted);
870
 
    session->server_status&= ~SERVER_STATUS_DB_DROPPED;
871
 
  }
872
 
  else if (drizzle_bin_log.is_open())
873
 
  {
874
 
    char *query, *query_pos, *query_end, *query_data_start;
875
 
    TableList *tbl;
876
 
    uint32_t db_len;
877
 
 
878
 
    if (!(query= (char*) session->alloc(MAX_DROP_TABLE_Q_LEN)))
879
 
      goto exit; /* not much else we can do */
880
 
    query_pos= query_data_start= my_stpcpy(query,"drop table ");
881
 
    query_end= query + MAX_DROP_TABLE_Q_LEN;
882
 
    db_len= strlen(db);
883
 
 
884
 
    for (tbl= dropped_tables; tbl; tbl= tbl->next_local)
885
 
    {
886
 
      uint32_t tbl_name_len;
887
 
 
888
 
      /* 3 for the quotes and the comma*/
889
 
      tbl_name_len= strlen(tbl->table_name) + 3;
890
 
      if (query_pos + tbl_name_len + 1 >= query_end)
891
 
      {
892
 
        /* These DDL methods and logging protected with LOCK_drizzle_create_db */
893
 
        write_to_binlog(session, query, query_pos -1 - query, db, db_len);
894
 
        query_pos= query_data_start;
895
 
      }
896
 
 
897
 
      *query_pos++ = '`';
898
 
      query_pos= my_stpcpy(query_pos,tbl->table_name);
899
 
      *query_pos++ = '`';
900
 
      *query_pos++ = ',';
901
 
    }
902
 
 
903
 
    if (query_pos != query_data_start)
904
 
    {
905
 
      /* These DDL methods and logging protected with LOCK_drizzle_create_db */
906
 
      write_to_binlog(session, query, query_pos -1 - query, db, db_len);
907
 
    }
908
 
  }
909
 
 
910
 
exit:
 
217
  if (session.wait_if_global_read_lock(false, true))
 
218
  {
 
219
    return true;
 
220
  }
 
221
 
 
222
  bool error= false;
 
223
  {
 
224
    boost::mutex::scoped_lock scopedLock(session.catalog().schemaLock());
 
225
    if (message::schema::shared_ptr message= plugin::StorageEngine::getSchemaDefinition(schema_identifier))
 
226
    {
 
227
      error= plugin::StorageEngine::dropSchema(session, schema_identifier, *message);
 
228
    }
 
229
    else if (if_exists)
 
230
    {
 
231
      push_warning_printf(&session, DRIZZLE_ERROR::WARN_LEVEL_NOTE, ER_DB_DROP_EXISTS, ER(ER_DB_DROP_EXISTS),
 
232
                          schema_identifier.getSQLPath().c_str());
 
233
    }
 
234
    else
 
235
    {
 
236
      error= true;
 
237
      my_error(ER_DB_DROP_EXISTS, schema_identifier);
 
238
    }
 
239
  };
 
240
 
911
241
  /*
912
242
    If this database was the client's selected database, we silently
913
243
    change the client's selected database to nothing (to have an empty
914
244
    SELECT DATABASE() in the future). For this we free() session->db and set
915
245
    it to 0.
916
246
  */
917
 
  if (session->db && !strcmp(session->db, db))
918
 
    mysql_change_db_impl(session, NULL, session->variables.collation_server);
919
 
  pthread_mutex_unlock(&LOCK_drizzle_create_db);
920
 
  start_waiting_global_read_lock(session);
921
 
exit2:
922
 
  return(error);
923
 
}
924
 
 
925
 
/*
926
 
  Removes files with known extensions plus.
927
 
  session MUST be set when calling this function!
928
 
*/
929
 
 
930
 
static long mysql_rm_known_files(Session *session, MY_DIR *dirp, const char *db,
931
 
                                 const char *org_path, uint32_t level,
932
 
                                 TableList **dropped_tables)
933
 
{
934
 
  long deleted=0;
935
 
  uint32_t found_other_files=0;
936
 
  char filePath[FN_REFLEN];
937
 
  TableList *tot_list=0, **tot_list_next;
938
 
 
939
 
  tot_list_next= &tot_list;
940
 
 
941
 
  for (uint32_t idx=0 ;
942
 
       idx < (uint) dirp->number_off_files && !session->killed ;
943
 
       idx++)
944
 
  {
945
 
    FILEINFO *file=dirp->dir_entry+idx;
946
 
    char *extension;
947
 
 
948
 
    /* skiping . and .. */
949
 
    if (file->name[0] == '.' && (!file->name[1] ||
950
 
       (file->name[1] == '.' &&  !file->name[2])))
951
 
      continue;
952
 
 
953
 
    if (!(extension= strrchr(file->name, '.')))
954
 
      extension= strchr(file->name, '\0');
955
 
    if (find_type(extension, &deletable_extentions,1+2) <= 0)
956
 
    {
957
 
      /*
958
 
        ass ass ass.
959
 
 
960
 
        strange checking for magic extensions that are then deleted if
961
 
        not reg_ext (i.e. .frm).
962
 
 
963
 
        and (previously) we'd err out on drop database if files not matching
964
 
        engine ha_known_exts() or deletable_extensions were present.
965
 
 
966
 
        presumably this was to avoid deleting other user data... except if that
967
 
        data happened to be in files ending in .BAK, .opt or .TMD. *fun*
968
 
       */
969
 
      find_type(extension, ha_known_exts(),1+2);
970
 
      continue;
971
 
    }
972
 
    /* just for safety we use files_charset_info */
973
 
    if (db && !my_strcasecmp(files_charset_info,
974
 
                             extension, reg_ext))
975
 
    {
976
 
      /* Drop the table nicely */
977
 
      *extension= 0;                    // Remove extension
978
 
      TableList *table_list=(TableList*)
979
 
                              session->calloc(sizeof(*table_list) + 
980
 
                                          strlen(db) + 1 +
981
 
                                          MYSQL50_TABLE_NAME_PREFIX_LENGTH + 
982
 
                                          strlen(file->name) + 1);
983
 
 
984
 
      if (!table_list)
985
 
        goto err;
986
 
      table_list->db= (char*) (table_list+1);
987
 
      table_list->table_name= my_stpcpy(table_list->db, db) + 1;
988
 
      filename_to_tablename(file->name, table_list->table_name,
989
 
                            MYSQL50_TABLE_NAME_PREFIX_LENGTH +
990
 
                            strlen(file->name) + 1);
991
 
      table_list->alias= table_list->table_name;        // If lower_case_table_names=2
992
 
      table_list->internal_tmp_table= is_prefix(file->name, TMP_FILE_PREFIX);
993
 
      /* Link into list */
994
 
      (*tot_list_next)= table_list;
995
 
      tot_list_next= &table_list->next_local;
996
 
      deleted++;
997
 
    }
998
 
    else
999
 
    {
1000
 
      strxmov(filePath, org_path, "/", file->name, NULL);
1001
 
      if (my_delete_with_symlink(filePath,MYF(MY_WME)))
1002
 
      {
1003
 
        goto err;
1004
 
      }
1005
 
    }
1006
 
  }
1007
 
  if (session->killed ||
1008
 
      (tot_list && mysql_rm_table_part2(session, tot_list, 1, 0, 1)))
1009
 
    goto err;
1010
 
 
1011
 
  my_dirend(dirp);  
1012
 
  
1013
 
  if (dropped_tables)
1014
 
    *dropped_tables= tot_list;
1015
 
  
1016
 
  /*
1017
 
    If the directory is a symbolic link, remove the link first, then
1018
 
    remove the directory the symbolic link pointed at
1019
 
  */
1020
 
  if (found_other_files)
1021
 
  {
1022
 
    my_error(ER_DB_DROP_RMDIR, MYF(0), org_path, EEXIST);
1023
 
    return(-1);
1024
 
  }
1025
 
  else
1026
 
  {
1027
 
    /* Don't give errors if we can't delete 'RAID' directory */
1028
 
    if (rm_dir_w_symlink(org_path, level == 0))
1029
 
      return(-1);
1030
 
  }
1031
 
 
1032
 
  return(deleted);
1033
 
 
1034
 
err:
1035
 
  my_dirend(dirp);
1036
 
  return(-1);
1037
 
}
1038
 
 
1039
 
 
1040
 
/*
1041
 
  Remove directory with symlink
1042
 
 
1043
 
  SYNOPSIS
1044
 
    rm_dir_w_symlink()
1045
 
    org_path    path of derictory
1046
 
    send_error  send errors
1047
 
  RETURN
1048
 
    0 OK
1049
 
    1 ERROR
1050
 
*/
1051
 
 
1052
 
static bool rm_dir_w_symlink(const char *org_path, bool send_error)
1053
 
{
1054
 
  char tmp_path[FN_REFLEN], *pos;
1055
 
  char *path= tmp_path;
1056
 
  unpack_filename(tmp_path, org_path);
1057
 
#ifdef HAVE_READLINK
1058
 
  int error;
1059
 
  char tmp2_path[FN_REFLEN];
1060
 
 
1061
 
  /* Remove end FN_LIBCHAR as this causes problem on Linux in readlink */
1062
 
  pos= strchr(path, '\0');
1063
 
  if (pos > path && pos[-1] == FN_LIBCHAR)
1064
 
    *--pos=0;
1065
 
 
1066
 
  if ((error= my_readlink(tmp2_path, path, MYF(MY_WME))) < 0)
1067
 
    return(1);
1068
 
  if (!error)
1069
 
  {
1070
 
    if (my_delete(path, MYF(send_error ? MY_WME : 0)))
1071
 
    {
1072
 
      return(send_error);
1073
 
    }
1074
 
    /* Delete directory symbolic link pointed at */
1075
 
    path= tmp2_path;
1076
 
  }
1077
 
#endif
1078
 
  /* Remove last FN_LIBCHAR to not cause a problem on OS/2 */
1079
 
  pos= strchr(path, '\0');
1080
 
 
1081
 
  if (pos > path && pos[-1] == FN_LIBCHAR)
1082
 
    *--pos=0;
1083
 
  if (rmdir(path) < 0 && send_error)
1084
 
  {
1085
 
    my_error(ER_DB_DROP_RMDIR, MYF(0), path, errno);
1086
 
    return(1);
1087
 
  }
1088
 
  return(0);
1089
 
}
1090
 
 
1091
 
 
1092
 
/**
1093
 
  @brief Internal implementation: switch current database to a valid one.
1094
 
 
1095
 
  @param session            Thread context.
1096
 
  @param new_db_name    Name of the database to switch to. The function will
1097
 
                        take ownership of the name (the caller must not free
1098
 
                        the allocated memory). If the name is NULL, we're
1099
 
                        going to switch to NULL db.
1100
 
  @param new_db_charset Character set of the new database.
1101
 
*/
1102
 
 
1103
 
static void mysql_change_db_impl(Session *session,
1104
 
                                 LEX_STRING *new_db_name,
1105
 
                                 const CHARSET_INFO * const new_db_charset)
1106
 
{
1107
 
  /* 1. Change current database in Session. */
1108
 
 
1109
 
  if (new_db_name == NULL)
1110
 
  {
1111
 
    /*
1112
 
      Session::set_db() does all the job -- it frees previous database name and
1113
 
      sets the new one.
1114
 
    */
1115
 
 
1116
 
    session->set_db(NULL, 0);
1117
 
  }
1118
 
  else if (my_strcasecmp(system_charset_info, new_db_name->str,
1119
 
                         INFORMATION_SCHEMA_NAME.c_str()) == 0)
1120
 
  {
1121
 
    /*
1122
 
      Here we must use Session::set_db(), because we want to copy
1123
 
      INFORMATION_SCHEMA_NAME constant.
1124
 
    */
1125
 
 
1126
 
    session->set_db(INFORMATION_SCHEMA_NAME.c_str(),
1127
 
                    INFORMATION_SCHEMA_NAME.length());
1128
 
  }
1129
 
  else
1130
 
  {
1131
 
    /*
1132
 
      Here we already have a copy of database name to be used in Session. So,
1133
 
      we just call Session::reset_db(). Since Session::reset_db() does not releases
1134
 
      the previous database name, we should do it explicitly.
1135
 
    */
1136
 
 
1137
 
    if (session->db)
1138
 
      free(session->db);
1139
 
 
1140
 
    session->reset_db(new_db_name->str, new_db_name->length);
1141
 
  }
1142
 
 
1143
 
  /* 3. Update db-charset environment variables. */
1144
 
 
1145
 
  session->db_charset= new_db_charset;
1146
 
  session->variables.collation_database= new_db_charset;
1147
 
}
1148
 
 
1149
 
 
1150
 
 
1151
 
/**
1152
 
  Backup the current database name before switch.
1153
 
 
1154
 
  @param[in]      session             thread handle
1155
 
  @param[in, out] saved_db_name   IN: "str" points to a buffer where to store
1156
 
                                  the old database name, "length" contains the
1157
 
                                  buffer size
1158
 
                                  OUT: if the current (default) database is
1159
 
                                  not NULL, its name is copied to the
1160
 
                                  buffer pointed at by "str"
1161
 
                                  and "length" is updated accordingly.
1162
 
                                  Otherwise "str" is set to NULL and
1163
 
                                  "length" is set to 0.
1164
 
*/
1165
 
 
1166
 
static void backup_current_db_name(Session *session,
1167
 
                                   LEX_STRING *saved_db_name)
1168
 
{
1169
 
  if (!session->db)
1170
 
  {
1171
 
    /* No current (default) database selected. */
1172
 
 
1173
 
    saved_db_name->str= NULL;
1174
 
    saved_db_name->length= 0;
1175
 
  }
1176
 
  else
1177
 
  {
1178
 
    strmake(saved_db_name->str, session->db, saved_db_name->length - 1);
1179
 
    saved_db_name->length= session->db_length;
1180
 
  }
1181
 
}
1182
 
 
1183
 
 
1184
 
/**
1185
 
  Return true if db1_name is equal to db2_name, false otherwise.
1186
 
 
1187
 
  The function allows to compare database names according to the MySQL
1188
 
  rules. The database names db1 and db2 are equal if:
1189
 
     - db1 is NULL and db2 is NULL;
1190
 
     or
1191
 
     - db1 is not-NULL, db2 is not-NULL, db1 is equal (ignoring case) to
1192
 
       db2 in system character set (UTF8).
1193
 
*/
1194
 
 
1195
 
static inline bool
1196
 
cmp_db_names(const char *db1_name,
1197
 
             const char *db2_name)
1198
 
{
1199
 
  return
1200
 
         /* db1 is NULL and db2 is NULL */
1201
 
         (!db1_name && !db2_name) ||
1202
 
 
1203
 
         /* db1 is not-NULL, db2 is not-NULL, db1 == db2. */
1204
 
         (db1_name && db2_name && my_strcasecmp(system_charset_info, db1_name, db2_name) == 0);
1205
 
}
1206
 
 
 
247
  if (not error and schema_identifier.compare(*session.schema()))
 
248
    session.set_schema("");
 
249
 
 
250
  session.startWaitingGlobalReadLock();
 
251
 
 
252
  return error;
 
253
}
1207
254
 
1208
255
/**
1209
256
  @brief Change the current database and its attributes unconditionally.
1267
314
    @retval true  Error
1268
315
*/
1269
316
 
1270
 
bool mysql_change_db(Session *session, const LEX_STRING *new_db_name, bool force_switch)
 
317
bool change(Session &session, const identifier::Schema &schema_identifier)
1271
318
{
1272
 
  LEX_STRING new_db_file_name;
1273
 
  const CHARSET_INFO *db_default_cl;
1274
 
 
1275
 
  if (new_db_name == NULL ||
1276
 
      new_db_name->length == 0)
1277
 
  {
1278
 
    if (force_switch)
1279
 
    {
1280
 
      /*
1281
 
        This can happen only if we're switching the current database back
1282
 
        after loading stored program. The thing is that loading of stored
1283
 
        program can happen when there is no current database.
1284
 
 
1285
 
        TODO: actually, new_db_name and new_db_name->str seem to be always
1286
 
        non-NULL. In case of stored program, new_db_name->str == "" and
1287
 
        new_db_name->length == 0.
1288
 
      */
1289
 
 
1290
 
      mysql_change_db_impl(session, NULL, session->variables.collation_server);
1291
 
 
1292
 
      return(false);
1293
 
    }
1294
 
    else
1295
 
    {
1296
 
      my_message(ER_NO_DB_ERROR, ER(ER_NO_DB_ERROR), MYF(0));
1297
 
 
1298
 
      return(true);
1299
 
    }
1300
 
  }
1301
 
 
1302
 
  if (my_strcasecmp(system_charset_info, new_db_name->str,
1303
 
                    INFORMATION_SCHEMA_NAME.c_str()) == 0)
1304
 
  {
1305
 
    /* Switch the current database to INFORMATION_SCHEMA. */
1306
 
    /* const_cast<> is safe here: mysql_change_db_impl does a copy */
1307
 
    LEX_STRING is_name= { const_cast<char *>(INFORMATION_SCHEMA_NAME.c_str()),
1308
 
                          INFORMATION_SCHEMA_NAME.length() };
1309
 
    mysql_change_db_impl(session, &is_name, system_charset_info);
1310
 
 
1311
 
    return(false);
1312
 
  }
1313
 
 
1314
 
  /*
1315
 
    Now we need to make a copy because check_db_name requires a
1316
 
    non-constant argument. Actually, it takes database file name.
1317
 
 
1318
 
    TODO: fix check_db_name().
1319
 
  */
1320
 
 
1321
 
  new_db_file_name.str= my_strndup(new_db_name->str, new_db_name->length,
1322
 
                                   MYF(MY_WME));
1323
 
  new_db_file_name.length= new_db_name->length;
1324
 
 
1325
 
  if (new_db_file_name.str == NULL)
1326
 
    return(true);                             /* the error is set */
1327
 
 
1328
 
  /*
1329
 
    NOTE: if check_db_name() fails, we should throw an error in any case,
1330
 
    even if we are called from sp_head::execute().
1331
 
 
1332
 
    It's next to impossible however to get this error when we are called
1333
 
    from sp_head::execute(). But let's switch the current database to NULL
1334
 
    in this case to be sure.
1335
 
  */
1336
 
 
1337
 
  if (check_db_name(&new_db_file_name))
1338
 
  {
1339
 
    my_error(ER_WRONG_DB_NAME, MYF(0), new_db_file_name.str);
1340
 
    free(new_db_file_name.str);
1341
 
 
1342
 
    if (force_switch)
1343
 
      mysql_change_db_impl(session, NULL, session->variables.collation_server);
1344
 
 
1345
 
    return(true);
1346
 
  }
1347
 
 
1348
 
  if (check_db_dir_existence(new_db_file_name.str))
1349
 
  {
1350
 
    if (force_switch)
1351
 
    {
1352
 
      /* Throw a warning and free new_db_file_name. */
1353
 
 
1354
 
      push_warning_printf(session, DRIZZLE_ERROR::WARN_LEVEL_NOTE,
1355
 
                          ER_BAD_DB_ERROR, ER(ER_BAD_DB_ERROR),
1356
 
                          new_db_file_name.str);
1357
 
 
1358
 
      free(new_db_file_name.str);
1359
 
 
1360
 
      /* Change db to NULL. */
1361
 
 
1362
 
      mysql_change_db_impl(session, NULL, session->variables.collation_server);
1363
 
 
1364
 
      /* The operation succeed. */
1365
 
 
1366
 
      return(false);
1367
 
    }
1368
 
    else
1369
 
    {
1370
 
      /* Report an error and free new_db_file_name. */
1371
 
 
1372
 
      my_error(ER_BAD_DB_ERROR, MYF(0), new_db_file_name.str);
1373
 
      free(new_db_file_name.str);
1374
 
 
1375
 
      /* The operation failed. */
1376
 
 
1377
 
      return(true);
1378
 
    }
1379
 
  }
1380
 
 
1381
 
  /*
1382
 
    NOTE: in mysql_change_db_impl() new_db_file_name is assigned to Session
1383
 
    attributes and will be freed in Session::~Session().
1384
 
  */
1385
 
 
1386
 
  db_default_cl= get_default_db_collation(session, new_db_file_name.str);
1387
 
 
1388
 
  mysql_change_db_impl(session, &new_db_file_name, db_default_cl);
1389
 
 
1390
 
  return(false);
 
319
 
 
320
  if (not plugin::Authorization::isAuthorized(*session.user(), schema_identifier))
 
321
  {
 
322
    /* Error message is set in isAuthorized */
 
323
    return true;
 
324
  }
 
325
 
 
326
  if (not check(session, schema_identifier))
 
327
  {
 
328
    my_error(ER_WRONG_DB_NAME, schema_identifier);
 
329
 
 
330
    return true;
 
331
  }
 
332
 
 
333
  if (not plugin::StorageEngine::doesSchemaExist(schema_identifier))
 
334
  {
 
335
    my_error(ER_BAD_DB_ERROR, schema_identifier);
 
336
 
 
337
    /* The operation failed. */
 
338
 
 
339
    return true;
 
340
  }
 
341
 
 
342
  session.set_schema(schema_identifier.getSchemaName());
 
343
 
 
344
  return false;
1391
345
}
1392
346
 
1393
 
 
1394
347
/**
1395
 
  Change the current database and its attributes if needed.
1396
 
 
1397
 
  @param          session             thread handle
1398
 
  @param          new_db_name     database name
1399
 
  @param[in, out] saved_db_name   IN: "str" points to a buffer where to store
1400
 
                                  the old database name, "length" contains the
1401
 
                                  buffer size
1402
 
                                  OUT: if the current (default) database is
1403
 
                                  not NULL, its name is copied to the
1404
 
                                  buffer pointed at by "str"
1405
 
                                  and "length" is updated accordingly.
1406
 
                                  Otherwise "str" is set to NULL and
1407
 
                                  "length" is set to 0.
1408
 
  @param          force_switch    @see mysql_change_db()
1409
 
  @param[out]     cur_db_changed  out-flag to indicate whether the current
1410
 
                                  database has been changed (valid only if
1411
 
                                  the function suceeded)
1412
 
*/
1413
 
 
1414
 
bool mysql_opt_change_db(Session *session,
1415
 
                         const LEX_STRING *new_db_name,
1416
 
                         LEX_STRING *saved_db_name,
1417
 
                         bool force_switch,
1418
 
                         bool *cur_db_changed)
 
348
  @brief Internal implementation: switch current database to a valid one.
 
349
 
 
350
  @param session            Thread context.
 
351
  @param new_db_name    Name of the database to switch to. The function will
 
352
                        take ownership of the name (the caller must not free
 
353
                        the allocated memory). If the name is NULL, we're
 
354
                        going to switch to NULL db.
 
355
  @param new_db_charset Character set of the new database.
 
356
*/
 
357
 
 
358
 
 
359
/*
 
360
  Check if database name is valid
 
361
 
 
362
  SYNPOSIS
 
363
    check()
 
364
    org_name            Name of database and length
 
365
 
 
366
  RETURN
 
367
    false error
 
368
    true ok
 
369
*/
 
370
 
 
371
bool check(Session &session, const identifier::Schema &schema_identifier)
1419
372
{
1420
 
  *cur_db_changed= !cmp_db_names(session->db, new_db_name->str);
1421
 
 
1422
 
  if (!*cur_db_changed)
 
373
  if (not plugin::Authorization::isAuthorized(*session.user(), schema_identifier))
1423
374
    return false;
1424
 
 
1425
 
  backup_current_db_name(session, saved_db_name);
1426
 
 
1427
 
  return mysql_change_db(session, new_db_name, force_switch);
1428
 
}
1429
 
 
1430
 
 
1431
 
/*
1432
 
  Check if there is directory for the database name.
1433
 
 
1434
 
  SYNOPSIS
1435
 
    check_db_dir_existence()
1436
 
    db_name   database name
1437
 
 
1438
 
  RETURN VALUES
1439
 
    false   There is directory for the specified database name.
1440
 
    true    The directory does not exist.
1441
 
*/
1442
 
 
1443
 
bool check_db_dir_existence(const char *db_name)
1444
 
{
1445
 
  char db_dir_path[FN_REFLEN];
1446
 
  uint32_t db_dir_path_len;
1447
 
 
1448
 
  db_dir_path_len= build_table_filename(db_dir_path, sizeof(db_dir_path),
1449
 
                                        db_name, "", "", 0);
1450
 
 
1451
 
  if (db_dir_path_len && db_dir_path[db_dir_path_len - 1] == FN_LIBCHAR)
1452
 
    db_dir_path[db_dir_path_len - 1]= 0;
1453
 
 
1454
 
  /* Check access. */
1455
 
 
1456
 
  return my_access(db_dir_path, F_OK);
1457
 
}
 
375
  return schema_identifier.isValid();
 
376
}
 
377
 
 
378
} /* namespace schema */
 
379
 
 
380
} /* namespace drizzled */