Schema.php 22.8 KB
Newer Older
1 2
<?php

3
namespace Drupal\Core\Database\Driver\mysql;
4

5
use Drupal\Core\Database\SchemaException;
6 7 8
use Drupal\Core\Database\SchemaObjectExistsException;
use Drupal\Core\Database\SchemaObjectDoesNotExistException;
use Drupal\Core\Database\Schema as DatabaseSchema;
9
use Drupal\Component\Utility\Unicode;
10 11

/**
12
 * @addtogroup schemaapi
13 14 15
 * @{
 */

16 17 18
/**
 * MySQL implementation of \Drupal\Core\Database\Schema.
 */
19
class Schema extends DatabaseSchema {
20

21 22 23 24 25 26 27 28 29 30
  /**
   * Maximum length of a table comment in MySQL.
   */
  const COMMENT_MAX_TABLE = 60;

  /**
   * Maximum length of a column comment in MySQL.
   */
  const COMMENT_MAX_COLUMN = 255;

31 32 33 34
  /**
   * @var array
   *   List of MySQL string types.
   */
35
  protected $mysqlStringTypes = [
36 37 38 39 40 41
    'VARCHAR',
    'CHAR',
    'TINYTEXT',
    'MEDIUMTEXT',
    'LONGTEXT',
    'TEXT',
42
  ];
43

44
  /**
45
   * Get information about the table and database name from the prefix.
46 47 48 49
   *
   * @return
   *   A keyed array with information about the database, table name and prefix.
   */
50
  protected function getPrefixInfo($table = 'default', $add_prefix = TRUE) {
51
    $info = ['prefix' => $this->connection->tablePrefix($table)];
52 53 54 55 56 57
    if ($add_prefix) {
      $table = $info['prefix'] . $table;
    }
    if (($pos = strpos($table, '.')) !== FALSE) {
      $info['database'] = substr($table, 0, $pos);
      $info['table'] = substr($table, ++$pos);
58 59
    }
    else {
60
      $info['database'] = $this->connection->getConnectionOptions()['database'];
61
      $info['table'] = $table;
62 63 64 65
    }
    return $info;
  }

66
  /**
Dries's avatar
Dries committed
67
   * Build a condition to match a table name against a standard information_schema.
68 69 70
   *
   * MySQL uses databases like schemas rather than catalogs so when we build
   * a condition to query the information_schema.tables, we set the default
Dries's avatar
Dries committed
71
   * database as the schema unless specified otherwise, and exclude table_catalog
72 73
   * from the condition criteria.
   */
74 75
  protected function buildTableNameCondition($table_name, $operator = '=', $add_prefix = TRUE) {
    $table_info = $this->getPrefixInfo($table_name, $add_prefix);
76

77
    $condition = $this->connection->condition('AND');
78 79
    $condition->condition('table_schema', $table_info['database']);
    $condition->condition('table_name', $table_info['table'], $operator);
80 81
    return $condition;
  }
82 83 84 85 86 87 88 89

  /**
   * Generate SQL to create a new table from a Drupal schema definition.
   *
   * @param $name
   *   The name of the table to create.
   * @param $table
   *   A Schema API table definition array.
90
   *
91 92 93 94
   * @return
   *   An array of SQL statements to create the table.
   */
  protected function createTableSql($name, $table) {
95 96 97
    $info = $this->connection->getConnectionOptions();

    // Provide defaults if needed.
98
    $table += [
99
      'mysql_engine' => 'InnoDB',
100
      'mysql_character_set' => 'utf8mb4',
101
    ];
102

103
    $sql = "CREATE TABLE {" . $name . "} (\n";
104

105 106 107 108
    // Add the SQL statement for each field.
    foreach ($table['fields'] as $field_name => $field) {
      $sql .= $this->createFieldSql($field_name, $this->processField($field)) . ", \n";
    }
109

110
    // Process keys & indexes.
111 112 113
    if (!empty($table['primary key']) && is_array($table['primary key'])) {
      $this->ensureNotNullPrimaryKey($table['primary key'], $table['fields']);
    }
114 115 116 117
    $keys = $this->createKeysSql($table);
    if (count($keys)) {
      $sql .= implode(", \n", $keys) . ", \n";
    }
118

119 120
    // Remove the last comma and space.
    $sql = substr($sql, 0, -3) . "\n) ";
121

122
    $sql .= 'ENGINE = ' . $table['mysql_engine'] . ' DEFAULT CHARACTER SET ' . $table['mysql_character_set'];
123
    // By default, MySQL uses the default collation for new tables, which is
124 125 126
    // 'utf8mb4_general_ci' (MySQL 5) or 'utf8mb4_0900_ai_ci' (MySQL 8) for
    // utf8mb4. If an alternate collation has been set, it needs to be
    // explicitly specified.
127
    // @see \Drupal\Core\Database\Driver\mysql\Schema
128 129 130
    if (!empty($info['collation'])) {
      $sql .= ' COLLATE ' . $info['collation'];
    }
131

132 133 134 135 136
    // Add table comment.
    if (!empty($table['description'])) {
      $sql .= ' COMMENT ' . $this->prepareComment($table['description'], self::COMMENT_MAX_TABLE);
    }

137
    return [$sql];
138 139 140 141 142
  }

  /**
   * Create an SQL string for a field to be used in table creation or alteration.
   *
143
   * @param string $name
144
   *   Name of the field.
145
   * @param array $spec
146 147 148 149
   *   The field specification, as per the schema data structure format.
   */
  protected function createFieldSql($name, $spec) {
    $sql = "`" . $name . "` " . $spec['mysql_type'];
150

151
    if (in_array($spec['mysql_type'], $this->mysqlStringTypes)) {
152 153 154
      if (isset($spec['length'])) {
        $sql .= '(' . $spec['length'] . ')';
      }
155 156 157
      if (isset($spec['type']) && $spec['type'] == 'varchar_ascii') {
        $sql .= ' CHARACTER SET ascii';
      }
158 159 160
      if (!empty($spec['binary'])) {
        $sql .= ' BINARY';
      }
161
      // Note we check for the "type" key here. "mysql_type" is VARCHAR:
162 163
      elseif (isset($spec['type']) && $spec['type'] == 'varchar_ascii') {
        $sql .= ' COLLATE ascii_general_ci';
164
      }
165 166 167 168
    }
    elseif (isset($spec['precision']) && isset($spec['scale'])) {
      $sql .= '(' . $spec['precision'] . ', ' . $spec['scale'] . ')';
    }
169

170 171 172
    if (!empty($spec['unsigned'])) {
      $sql .= ' unsigned';
    }
173

174 175 176 177 178 179 180
    if (isset($spec['not null'])) {
      if ($spec['not null']) {
        $sql .= ' NOT NULL';
      }
      else {
        $sql .= ' NULL';
      }
181
    }
182

183 184 185
    if (!empty($spec['auto_increment'])) {
      $sql .= ' auto_increment';
    }
186

187
    // $spec['default'] can be NULL, so we explicitly check for the key here.
188
    if (array_key_exists('default', $spec)) {
189
      $sql .= ' DEFAULT ' . $this->escapeDefaultValue($spec['default']);
190
    }
191

192 193 194
    if (empty($spec['not null']) && !isset($spec['default'])) {
      $sql .= ' DEFAULT NULL';
    }
195

196 197 198 199 200
    // Add column comment.
    if (!empty($spec['description'])) {
      $sql .= ' COMMENT ' . $this->prepareComment($spec['description'], self::COMMENT_MAX_COLUMN);
    }

201 202
    return $sql;
  }
203

204 205 206 207 208 209 210
  /**
   * Set database-engine specific properties for a field.
   *
   * @param $field
   *   A field description array, as specified in the schema documentation.
   */
  protected function processField($field) {
211

212 213 214
    if (!isset($field['size'])) {
      $field['size'] = 'normal';
    }
215

216
    // Set the correct database-engine specific datatype.
217 218
    // In case one is already provided, force it to uppercase.
    if (isset($field['mysql_type'])) {
219
      $field['mysql_type'] = mb_strtoupper($field['mysql_type']);
220 221
    }
    else {
222
      $map = $this->getFieldTypeMap();
223 224
      $field['mysql_type'] = $map[$field['type'] . ':' . $field['size']];
    }
225

226
    if (isset($field['type']) && $field['type'] == 'serial') {
227 228
      $field['auto_increment'] = TRUE;
    }
229

230 231 232
    return $field;
  }

233 234 235
  /**
   * {@inheritdoc}
   */
236
  public function getFieldTypeMap() {
237
    // Put :normal last so it gets preserved by array_flip. This makes
238 239
    // it much easier for modules (such as schema.module) to map
    // database types back into schema types.
240
    // $map does not use drupal_static as its value never changes.
241
    static $map = [
242 243
      'varchar_ascii:normal' => 'VARCHAR',

244 245
      'varchar:normal'  => 'VARCHAR',
      'char:normal'     => 'CHAR',
246

247 248 249 250 251
      'text:tiny'       => 'TINYTEXT',
      'text:small'      => 'TINYTEXT',
      'text:medium'     => 'MEDIUMTEXT',
      'text:big'        => 'LONGTEXT',
      'text:normal'     => 'TEXT',
252

253 254 255 256 257
      'serial:tiny'     => 'TINYINT',
      'serial:small'    => 'SMALLINT',
      'serial:medium'   => 'MEDIUMINT',
      'serial:big'      => 'BIGINT',
      'serial:normal'   => 'INT',
258

259 260 261 262 263
      'int:tiny'        => 'TINYINT',
      'int:small'       => 'SMALLINT',
      'int:medium'      => 'MEDIUMINT',
      'int:big'         => 'BIGINT',
      'int:normal'      => 'INT',
264

265 266 267 268 269
      'float:tiny'      => 'FLOAT',
      'float:small'     => 'FLOAT',
      'float:medium'    => 'FLOAT',
      'float:big'       => 'DOUBLE',
      'float:normal'    => 'FLOAT',
270

271
      'numeric:normal'  => 'DECIMAL',
272

273 274
      'blob:big'        => 'LONGBLOB',
      'blob:normal'     => 'BLOB',
275
    ];
276 277
    return $map;
  }
278

279
  protected function createKeysSql($spec) {
280
    $keys = [];
281 282

    if (!empty($spec['primary key'])) {
283
      $keys[] = 'PRIMARY KEY (' . $this->createKeySql($spec['primary key']) . ')';
284 285 286
    }
    if (!empty($spec['unique keys'])) {
      foreach ($spec['unique keys'] as $key => $fields) {
287
        $keys[] = 'UNIQUE KEY `' . $key . '` (' . $this->createKeySql($fields) . ')';
288 289 290
      }
    }
    if (!empty($spec['indexes'])) {
291 292
      $indexes = $this->getNormalizedIndexes($spec);
      foreach ($indexes as $index => $fields) {
293
        $keys[] = 'INDEX `' . $index . '` (' . $this->createKeySql($fields) . ')';
294 295 296 297 298
      }
    }

    return $keys;
  }
299

300 301 302 303 304 305
  /**
   * Gets normalized indexes from a table specification.
   *
   * Shortens indexes to 191 characters if they apply to utf8mb4-encoded
   * fields, in order to comply with the InnoDB index limitation of 756 bytes.
   *
306
   * @param array $spec
307 308 309 310
   *   The table specification.
   *
   * @return array
   *   List of shortened indexes.
311 312 313
   *
   * @throws \Drupal\Core\Database\SchemaException
   *   Thrown if field specification is missing.
314
   */
315 316
  protected function getNormalizedIndexes(array $spec) {
    $indexes = isset($spec['indexes']) ? $spec['indexes'] : [];
317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332
    foreach ($indexes as $index_name => $index_fields) {
      foreach ($index_fields as $index_key => $index_field) {
        // Get the name of the field from the index specification.
        $field_name = is_array($index_field) ? $index_field[0] : $index_field;
        // Check whether the field is defined in the table specification.
        if (isset($spec['fields'][$field_name])) {
          // Get the MySQL type from the processed field.
          $mysql_field = $this->processField($spec['fields'][$field_name]);
          if (in_array($mysql_field['mysql_type'], $this->mysqlStringTypes)) {
            // Check whether we need to shorten the index.
            if ((!isset($mysql_field['type']) || $mysql_field['type'] != 'varchar_ascii') && (!isset($mysql_field['length']) || $mysql_field['length'] > 191)) {
              // Limit the index length to 191 characters.
              $this->shortenIndex($indexes[$index_name][$index_key]);
            }
          }
        }
333 334 335
        else {
          throw new SchemaException("MySQL needs the '$field_name' field specification in order to normalize the '$index_name' index");
        }
336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358
      }
    }
    return $indexes;
  }

  /**
   * Helper function for normalizeIndexes().
   *
   * Shortens an index to 191 characters.
   *
   * @param array $index
   *   The index array to be used in createKeySql.
   *
   * @see Drupal\Core\Database\Driver\mysql\Schema::createKeySql()
   * @see Drupal\Core\Database\Driver\mysql\Schema::normalizeIndexes()
   */
  protected function shortenIndex(&$index) {
    if (is_array($index)) {
      if ($index[1] > 191) {
        $index[1] = 191;
      }
    }
    else {
359
      $index = [$index, 191];
360 361 362
    }
  }

363
  protected function createKeySql($fields) {
364
    $return = [];
365 366
    foreach ($fields as $field) {
      if (is_array($field)) {
367
        $return[] = '`' . $field[0] . '`(' . $field[1] . ')';
368 369
      }
      else {
370
        $return[] = '`' . $field . '`';
371 372
      }
    }
373
    return implode(', ', $return);
374 375
  }

376 377 378
  /**
   * {@inheritdoc}
   */
379
  public function renameTable($table, $new_name) {
380
    if (!$this->tableExists($table)) {
381
      throw new SchemaObjectDoesNotExistException("Cannot rename '$table' to '$new_name': table '$table' doesn't exist.");
382 383
    }
    if ($this->tableExists($new_name)) {
384
      throw new SchemaObjectExistsException("Cannot rename '$table' to '$new_name': table '$new_name' already exists.");
385 386
    }

387
    $info = $this->getPrefixInfo($new_name);
388
    $this->connection->query('ALTER TABLE {' . $table . '} RENAME TO `' . $info['table'] . '`');
389
  }
390

391 392 393
  /**
   * {@inheritdoc}
   */
394
  public function dropTable($table) {
395 396 397 398
    if (!$this->tableExists($table)) {
      return FALSE;
    }

399
    $this->connection->query('DROP TABLE {' . $table . '}');
400
    return TRUE;
401 402
  }

403 404 405
  /**
   * {@inheritdoc}
   */
406
  public function addField($table, $field, $spec, $keys_new = []) {
407
    if (!$this->tableExists($table)) {
408
      throw new SchemaObjectDoesNotExistException("Cannot add field '$table.$field': table doesn't exist.");
409
    }
410
    if ($this->fieldExists($table, $field)) {
411
      throw new SchemaObjectExistsException("Cannot add field '$table.$field': field already exists.");
412 413
    }

414 415
    // Fields that are part of a PRIMARY KEY must be added as NOT NULL.
    $is_primary_key = isset($keys_new['primary key']) && in_array($field, $keys_new['primary key'], TRUE);
416 417 418
    if ($is_primary_key) {
      $this->ensureNotNullPrimaryKey($keys_new['primary key'], [$field => $spec]);
    }
419

420
    $fixnull = FALSE;
421
    if (!empty($spec['not null']) && !isset($spec['default']) && !$is_primary_key) {
422 423 424 425 426
      $fixnull = TRUE;
      $spec['not null'] = FALSE;
    }
    $query = 'ALTER TABLE {' . $table . '} ADD ';
    $query .= $this->createFieldSql($field, $this->processField($spec));
427
    if ($keys_sql = $this->createKeysSql($keys_new)) {
428 429 430 431 432 433 434
      // Make sure to drop the existing primary key before adding a new one.
      // This is only needed when adding a field because this method, unlike
      // changeField(), is supposed to handle primary keys automatically.
      if (isset($keys_new['primary key']) && $this->indexExists($table, 'PRIMARY')) {
        $query .= ', DROP PRIMARY KEY';
      }

435
      $query .= ', ADD ' . implode(', ADD ', $keys_sql);
436
    }
437
    $this->connection->query($query);
438
    if (isset($spec['initial_from_field'])) {
439 440 441 442 443 444 445 446
      if (isset($spec['initial'])) {
        $expression = 'COALESCE(' . $spec['initial_from_field'] . ', :default_initial_value)';
        $arguments = [':default_initial_value' => $spec['initial']];
      }
      else {
        $expression = $spec['initial_from_field'];
        $arguments = [];
      }
447
      $this->connection->update($table)
448
        ->expression($field, $expression, $arguments)
449 450
        ->execute();
    }
451 452 453 454 455
    elseif (isset($spec['initial'])) {
      $this->connection->update($table)
        ->fields([$field => $spec['initial']])
        ->execute();
    }
456 457
    if ($fixnull) {
      $spec['not null'] = TRUE;
458
      $this->changeField($table, $field, $field, $spec);
459 460 461
    }
  }

462 463 464
  /**
   * {@inheritdoc}
   */
465
  public function dropField($table, $field) {
466
    if (!$this->fieldExists($table, $field)) {
467 468 469
      return FALSE;
    }

470 471 472 473 474 475 476 477 478 479 480 481
    // When dropping a field that is part of a composite primary key MySQL
    // automatically removes the field from the primary key, which can leave the
    // table in an invalid state. MariaDB 10.2.8 requires explicitly dropping
    // the primary key first for this reason. We perform this deletion
    // explicitly which also makes the behavior on both MySQL and MariaDB
    // consistent with PostgreSQL.
    // @see https://mariadb.com/kb/en/library/alter-table
    $primary_key = $this->findPrimaryKeyColumns($table);
    if ((count($primary_key) > 1) && in_array($field, $primary_key, TRUE)) {
      $this->dropPrimaryKey($table);
    }

482
    $this->connection->query('ALTER TABLE {' . $table . '} DROP `' . $field . '`');
483
    return TRUE;
484 485
  }

486 487 488
  /**
   * {@inheritdoc}
   */
489
  public function indexExists($table, $name) {
490 491
    // Returns one row for each column in the index. Result is string or FALSE.
    // Details at http://dev.mysql.com/doc/refman/5.0/en/show-index.html
492
    $row = $this->connection->query('SHOW INDEX FROM {' . $table . '} WHERE key_name = ' . $this->connection->quote($name))->fetchAssoc();
493
    return isset($row['Key_name']);
494 495
  }

496 497 498
  /**
   * {@inheritdoc}
   */
499
  public function addPrimaryKey($table, $fields) {
500
    if (!$this->tableExists($table)) {
501
      throw new SchemaObjectDoesNotExistException("Cannot add primary key to table '$table': table doesn't exist.");
502 503
    }
    if ($this->indexExists($table, 'PRIMARY')) {
504
      throw new SchemaObjectExistsException("Cannot add primary key to table '$table': primary key already exists.");
505 506
    }

507
    $this->connection->query('ALTER TABLE {' . $table . '} ADD PRIMARY KEY (' . $this->createKeySql($fields) . ')');
508
  }
509

510 511 512
  /**
   * {@inheritdoc}
   */
513
  public function dropPrimaryKey($table) {
514 515 516 517
    if (!$this->indexExists($table, 'PRIMARY')) {
      return FALSE;
    }

518
    $this->connection->query('ALTER TABLE {' . $table . '} DROP PRIMARY KEY');
519
    return TRUE;
520 521
  }

522 523 524 525 526 527 528 529 530 531 532
  /**
   * {@inheritdoc}
   */
  protected function findPrimaryKeyColumns($table) {
    if (!$this->tableExists($table)) {
      return FALSE;
    }
    $result = $this->connection->query("SHOW KEYS FROM {" . $table . "} WHERE Key_name = 'PRIMARY'")->fetchAllAssoc('Column_name');
    return array_keys($result);
  }

533 534 535
  /**
   * {@inheritdoc}
   */
536
  public function addUniqueKey($table, $name, $fields) {
537
    if (!$this->tableExists($table)) {
538
      throw new SchemaObjectDoesNotExistException("Cannot add unique key '$name' to table '$table': table doesn't exist.");
539 540
    }
    if ($this->indexExists($table, $name)) {
541
      throw new SchemaObjectExistsException("Cannot add unique key '$name' to table '$table': unique key already exists.");
542 543
    }

544
    $this->connection->query('ALTER TABLE {' . $table . '} ADD UNIQUE KEY `' . $name . '` (' . $this->createKeySql($fields) . ')');
545 546
  }

547 548 549
  /**
   * {@inheritdoc}
   */
550
  public function dropUniqueKey($table, $name) {
551 552 553 554
    if (!$this->indexExists($table, $name)) {
      return FALSE;
    }

555
    $this->connection->query('ALTER TABLE {' . $table . '} DROP KEY `' . $name . '`');
556
    return TRUE;
557
  }
558

559 560 561 562
  /**
   * {@inheritdoc}
   */
  public function addIndex($table, $name, $fields, array $spec) {
563
    if (!$this->tableExists($table)) {
564
      throw new SchemaObjectDoesNotExistException("Cannot add index '$name' to table '$table': table doesn't exist.");
565 566
    }
    if ($this->indexExists($table, $name)) {
567
      throw new SchemaObjectExistsException("Cannot add index '$name' to table '$table': index already exists.");
568 569
    }

570 571 572 573
    $spec['indexes'][$name] = $fields;
    $indexes = $this->getNormalizedIndexes($spec);

    $this->connection->query('ALTER TABLE {' . $table . '} ADD INDEX `' . $name . '` (' . $this->createKeySql($indexes[$name]) . ')');
574 575
  }

576 577 578
  /**
   * {@inheritdoc}
   */
579
  public function dropIndex($table, $name) {
580 581 582 583
    if (!$this->indexExists($table, $name)) {
      return FALSE;
    }

584
    $this->connection->query('ALTER TABLE {' . $table . '} DROP INDEX `' . $name . '`');
585
    return TRUE;
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
  /**
   * {@inheritdoc}
   */
  protected function introspectIndexSchema($table) {
    if (!$this->tableExists($table)) {
      throw new SchemaObjectDoesNotExistException("The table $table doesn't exist.");
    }

    $index_schema = [
      'primary key' => [],
      'unique keys' => [],
      'indexes' => [],
    ];

    $result = $this->connection->query('SHOW INDEX FROM {' . $table . '}')->fetchAll();
    foreach ($result as $row) {
      if ($row->Key_name === 'PRIMARY') {
        $index_schema['primary key'][] = $row->Column_name;
      }
      elseif ($row->Non_unique == 0) {
        $index_schema['unique keys'][$row->Key_name][] = $row->Column_name;
      }
      else {
        $index_schema['indexes'][$row->Key_name][] = $row->Column_name;
      }
    }

    return $index_schema;
  }

618 619 620
  /**
   * {@inheritdoc}
   */
621
  public function changeField($table, $field, $field_new, $spec, $keys_new = []) {
622
    if (!$this->fieldExists($table, $field)) {
623
      throw new SchemaObjectDoesNotExistException("Cannot change the definition of field '$table.$field': field doesn't exist.");
624
    }
625
    if (($field != $field_new) && $this->fieldExists($table, $field_new)) {
626
      throw new SchemaObjectExistsException("Cannot rename field '$table.$field' to '$field_new': target field already exists.");
627
    }
628 629 630
    if (isset($keys_new['primary key']) && in_array($field_new, $keys_new['primary key'], TRUE)) {
      $this->ensureNotNullPrimaryKey($keys_new['primary key'], [$field_new => $spec]);
    }
631

632
    $sql = 'ALTER TABLE {' . $table . '} CHANGE `' . $field . '` ' . $this->createFieldSql($field_new, $this->processField($spec));
633 634
    if ($keys_sql = $this->createKeysSql($keys_new)) {
      $sql .= ', ADD ' . implode(', ADD ', $keys_sql);
635
    }
636
    $this->connection->query($sql);
637 638
  }

639 640 641
  /**
   * {@inheritdoc}
   */
642 643 644 645
  public function prepareComment($comment, $length = NULL) {
    // Truncate comment to maximum comment length.
    if (isset($length)) {
      // Add table prefixes before truncating.
646
      $comment = Unicode::truncate($this->connection->prefixTables($comment), $length, TRUE, TRUE);
647
    }
648
    // Remove semicolons to avoid triggering multi-statement check.
649
    $comment = strtr($comment, [';' => '.']);
650 651 652 653 654 655 656
    return $this->connection->quote($comment);
  }

  /**
   * Retrieve a table or column comment.
   */
  public function getComment($table, $column = NULL) {
657
    $condition = $this->buildTableNameCondition($table);
658 659
    if (isset($column)) {
      $condition->condition('column_name', $column);
660
      $condition->compile($this->connection, $this);
661
      // Don't use {} around information_schema.columns table.
662
      return $this->connection->query("SELECT column_comment AS column_comment FROM information_schema.columns WHERE " . (string) $condition, $condition->arguments())->fetchField();
663
    }
664
    $condition->compile($this->connection, $this);
665
    // Don't use {} around information_schema.tables table.
666
    $comment = $this->connection->query("SELECT table_comment AS table_comment FROM information_schema.tables WHERE " . (string) $condition, $condition->arguments())->fetchField();
667 668
    // Work-around for MySQL 5.0 bug http://bugs.mysql.com/bug.php?id=11379
    return preg_replace('/; InnoDB free:.*$/', '', $comment);
669 670
  }

671 672 673
  /**
   * {@inheritdoc}
   */
674 675 676 677 678 679
  public function tableExists($table) {
    // The information_schema table is very slow to query under MySQL 5.0.
    // Instead, we try to select from the table in question.  If it fails,
    // the most likely reason is that it does not exist. That is dramatically
    // faster than using information_schema.
    // @link http://bugs.mysql.com/bug.php?id=19588
680 681
    // @todo This override should be removed once we require a version of MySQL
    //   that has that bug fixed.
682 683 684 685
    try {
      $this->connection->queryRange("SELECT 1 FROM {" . $table . "}", 0, 1);
      return TRUE;
    }
686
    catch (\Exception $e) {
687 688 689 690
      return FALSE;
    }
  }

691 692 693
  /**
   * {@inheritdoc}
   */
694 695 696 697 698 699
  public function fieldExists($table, $column) {
    // The information_schema table is very slow to query under MySQL 5.0.
    // Instead, we try to select from the table and field in question. If it
    // fails, the most likely reason is that it does not exist. That is
    // dramatically faster than using information_schema.
    // @link http://bugs.mysql.com/bug.php?id=19588
700 701
    // @todo This override should be removed once we require a version of MySQL
    //   that has that bug fixed.
702 703 704 705
    try {
      $this->connection->queryRange("SELECT $column FROM {" . $table . "}", 0, 1);
      return TRUE;
    }
706
    catch (\Exception $e) {
707 708 709 710
      return FALSE;
    }
  }

711 712 713
}

/**
714
 * @} End of "addtogroup schemaapi".
715
 */