lib/schema.php

   1 <?php
   2 /**
   3  * StatusNet, the distributed open-source microblogging tool
   4  *
   5  * Database schema utilities
   6  *
   7  * PHP version 5
   8  *
   9  * LICENCE: This program is free software: you can redistribute it and/or modify
  10  * it under the terms of the GNU Affero General Public License as published by
  11  * the Free Software Foundation, either version 3 of the License, or
  12  * (at your option) any later version.
  13  *
  14  * This program is distributed in the hope that it will be useful,
  15  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  17  * GNU Affero General Public License for more details.
  18  *
  19  * You should have received a copy of the GNU Affero General Public License
  20  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  21  *
  22  * @category  Database
  23  * @package   StatusNet
  24  * @author    Evan Prodromou <evan@status.net>
  25  * @copyright 2009 StatusNet, Inc.
  26  * @license   http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
  27  * @link      http://status.net/
  28  */
  29
  30 if (!defined('STATUSNET')) {
  31     exit(1);
  32 }
  33
  34 /**
  35  * Class representing the database schema
  36  *
  37  * A class representing the database schema. Can be used to
  38  * manipulate the schema -- especially for plugins and upgrade
  39  * utilities.
  40  *
  41  * @category Database
  42  * @package  StatusNet
  43  * @author   Evan Prodromou <evan@status.net>
  44  * @author   Brion Vibber <brion@status.net>
  45  * @license  http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
  46  * @link     http://status.net/
  47  */
  48
  49 class Schema
  50 {
  51     static $_static = null;
  52     protected $conn = null;
  53
  54     /**
  55      * Constructor. Only run once for singleton object.
  56      */
  57
  58     protected function __construct($conn = null)
  59     {
  60         if (is_null($conn)) {
  61             // XXX: there should be an easier way to do this.
  62             $user = new User();
  63             $conn = $user->getDatabaseConnection();
  64             $user->free();
  65             unset($user);
  66         }
  67
  68         $this->conn = $conn;
  69     }
  70
  71     /**
  72      * Main public entry point. Use this to get
  73      * the schema object.
  74      *
  75      * @return Schema the Schema object for the connection
  76      */
  77
  78     static function get($conn = null)
  79     {
  80         if (is_null($conn)) {
  81             $key = 'default';
  82         } else {
  83             $key = md5(serialize($conn->dsn));
  84         }
  85
  86         $type = common_config('db', 'type');
  87         if (empty(self::$_static[$key])) {
  88             $schemaClass = ucfirst($type).'Schema';
  89             self::$_static[$key] = new $schemaClass($conn);
  90         }
  91         return self::$_static[$key];
  92     }
  93
  94     /**
  95      * Gets a ColumnDef object for a single column.
  96      *
  97      * Throws an exception if the table is not found.
  98      *
  99      * @param string $table  name of the table
 100      * @param string $column name of the column
 101      *
 102      * @return ColumnDef definition of the column or null
 103      *                   if not found.
 104      */
 105
 106     public function getColumnDef($table, $column)
 107     {
 108         $td = $this->getTableDef($table);
 109
 110         foreach ($td->columns as $cd) {
 111             if ($cd->name == $column) {
 112                 return $cd;
 113             }
 114         }
 115
 116         return null;
 117     }
 118
 119     /**
 120      * Creates a table with the given names and columns.
 121      *
 122      * @param string $tableName    Name of the table
 123      * @param array  $def          Table definition array listing fields and indexes.
 124      *
 125      * @return boolean success flag
 126      */
 127
 128     public function createTable($tableName, $def)
 129     {
 130         $statements = $this->buildCreateTable($tableName, $def);
 131         return $this->runSqlSet($statements);
 132     }
 133
 134     /**
 135      * Build a set of SQL statements to create a table with the given
 136      * name and columns.
 137      *
 138      * @param string $name    Name of the table
 139      * @param array  $def     Table definition array
 140      *
 141      * @return boolean success flag
 142      */
 143     public function buildCreateTable($name, $def)
 144     {
 145         $def = $this->validateDef($name, $def);
 146         $def = $this->filterDef($def);
 147         $sql = array();
 148
 149         foreach ($def['fields'] as $col => $colDef) {
 150             $this->appendColumnDef($sql, $col, $colDef);
 151         }
 152
 153         // Primary, unique, and foreign keys are constraints, so go within
 154         // the CREATE TABLE statement normally.
 155         if (!empty($def['primary key'])) {
 156             $this->appendPrimaryKeyDef($sql, $def['primary key']);
 157         }
 158
 159         if (!empty($def['unique keys'])) {
 160             foreach ($def['unique keys'] as $col => $colDef) {
 161                 $this->appendUniqueKeyDef($sql, $col, $colDef);
 162             }
 163         }
 164
 165         if (!empty($def['foreign keys'])) {
 166             foreach ($def['foreign keys'] as $keyName => $keyDef) {
 167                 $this->appendForeignKeyDef($sql, $keyName, $keyDef);
 168             }
 169         }
 170
 171         // Wrap the CREATE TABLE around the main body chunks...
 172         $statements = array();
 173         $statements[] = $this->startCreateTable($name, $def) . "\n" .
 174                         implode($sql, ",\n") . "\n" .
 175                         $this->endCreateTable($name, $def);
 176
 177         // Multi-value indexes are advisory and for best portability
 178         // should be created as separate statements.
 179         if (!empty($def['indexes'])) {
 180             foreach ($def['indexes'] as $col => $colDef) {
 181                 $this->appendCreateIndex($statements, $name, $col, $colDef);
 182             }
 183         }
 184         if (!empty($def['fulltext indexes'])) {
 185             foreach ($def['fulltext indexes'] as $col => $colDef) {
 186                 $this->appendCreateFulltextIndex($statements, $name, $col, $colDef);
 187             }
 188         }
 189
 190         return $statements;
 191     }
 192
 193     /**
 194      * Set up a 'create table' SQL statement.
 195      *
 196      * @param string $name table name
 197      * @param array $def table definition
 198      * @param $string
 199      */
 200     function startCreateTable($name, array $def)
 201     {
 202         return 'CREATE TABLE ' . $this->quoteIdentifier($name)  . ' (';
 203     }
 204
 205     /**
 206      * Close out a 'create table' SQL statement.
 207      *
 208      * @param string $name table name
 209      * @param array $def table definition
 210      * @return string
 211      */
 212     function endCreateTable($name, array $def)
 213     {
 214         return ')';
 215     }
 216
 217     /**
 218      * Append an SQL fragment with a column definition in a CREATE TABLE statement.
 219      *
 220      * @param array $sql
 221      * @param string $name
 222      * @param array $def
 223      */
 224     function appendColumnDef(array &$sql, $name, array $def)
 225     {
 226         $sql[] = "$name " . $this->columnSql($def);
 227     }
 228
 229     /**
 230      * Append an SQL fragment with a constraint definition for a primary
 231      * key in a CREATE TABLE statement.
 232      *
 233      * @param array $sql
 234      * @param array $def
 235      */
 236     function appendPrimaryKeyDef(array &$sql, array $def)
 237     {
 238         $sql[] = "PRIMARY KEY " . $this->buildIndexList($def);
 239     }
 240
 241     /**
 242      * Append an SQL fragment with a constraint definition for a unique
 243      * key in a CREATE TABLE statement.
 244      *
 245      * @param array $sql
 246      * @param string $name
 247      * @param array $def
 248      */
 249     function appendUniqueKeyDef(array &$sql, $name, array $def)
 250     {
 251         $sql[] = "CONSTRAINT $name UNIQUE " . $this->buildIndexList($def);
 252     }
 253
 254     /**
 255      * Append an SQL fragment with a constraint definition for a foreign
 256      * key in a CREATE TABLE statement.
 257      *
 258      * @param array $sql
 259      * @param string $name
 260      * @param array $def
 261      */
 262     function appendForeignKeyDef(array &$sql, $name, array $def)
 263     {
 264         if (count($def) != 2) {
 265             throw new Exception("Invalid foreign key def for $name: " . var_export($def, true));
 266         }
 267         list($refTable, $map) = $def;
 268         $srcCols = array_keys($map);
 269         $refCols = array_values($map);
 270         $sql[] = "CONSTRAINT $name FOREIGN KEY " .
 271                  $this->buildIndexList($srcCols) .
 272                  " REFERENCES " .
 273                  $this->quoteIdentifier($refTable) .
 274                  " " .
 275                  $this->buildIndexList($refCols);
 276     }
 277
 278     /**
 279      * Append an SQL statement with an index definition for an advisory
 280      * index over one or more columns on a table.
 281      *
 282      * @param array $statements
 283      * @param string $table
 284      * @param string $name
 285      * @param array $def
 286      */
 287     function appendCreateIndex(array &$statements, $table, $name, array $def)
 288     {
 289         $statements[] = "CREATE INDEX $name ON $table " . $this->buildIndexList($def);
 290     }
 291
 292     /**
 293      * Append an SQL statement with an index definition for a full-text search
 294      * index over one or more columns on a table.
 295      *
 296      * @param array $statements
 297      * @param string $table
 298      * @param string $name
 299      * @param array $def
 300      */
 301     function appendCreateFulltextIndex(array &$statements, $table, $name, array $def)
 302     {
 303         throw new Exception("Fulltext index not supported in this database");
 304     }
 305
 306     /**
 307      * Append an SQL statement to drop an index from a table.
 308      *
 309      * @param array $statements
 310      * @param string $table
 311      * @param string $name
 312      * @param array $def
 313      */
 314     function appendDropIndex(array &$statements, $table, $name)
 315     {
 316         $statements[] = "DROP INDEX $name ON " . $this->quoteIdentifier($table);
 317     }
 318
 319     function buildIndexList(array $def)
 320     {
 321         // @fixme
 322         return '(' . implode(',', array_map(array($this, 'buildIndexItem'), $def)) . ')';
 323     }
 324
 325     function buildIndexItem($def)
 326     {
 327         if (is_array($def)) {
 328             list($name, $size) = $def;
 329             return $this->quoteIdentifier($name) . '(' . intval($size) . ')';
 330         }
 331         return $this->quoteIdentifier($def);
 332     }
 333
 334     /**
 335      * Drops a table from the schema
 336      *
 337      * Throws an exception if the table is not found.
 338      *
 339      * @param string $name Name of the table to drop
 340      *
 341      * @return boolean success flag
 342      */
 343
 344     public function dropTable($name)
 345     {
 346         $res = $this->conn->query("DROP TABLE $name");
 347
 348         if (PEAR::isError($res)) {
 349             throw new Exception($res->getMessage());
 350         }
 351
 352         return true;
 353     }
 354
 355     /**
 356      * Adds an index to a table.
 357      *
 358      * If no name is provided, a name will be made up based
 359      * on the table name and column names.
 360      *
 361      * Throws an exception on database error, esp. if the table
 362      * does not exist.
 363      *
 364      * @param string $table       Name of the table
 365      * @param array  $columnNames Name of columns to index
 366      * @param string $name        (Optional) name of the index
 367      *
 368      * @return boolean success flag
 369      */
 370
 371     public function createIndex($table, $columnNames, $name=null)
 372     {
 373         if (!is_array($columnNames)) {
 374             $columnNames = array($columnNames);
 375         }
 376
 377         if (empty($name)) {
 378             $name = "{$table}_".implode("_", $columnNames)."_idx";
 379         }
 380
 381         $res = $this->conn->query("ALTER TABLE $table ".
 382                                    "ADD INDEX $name (".
 383                                    implode(",", $columnNames).")");
 384
 385         if (PEAR::isError($res)) {
 386             throw new Exception($res->getMessage());
 387         }
 388
 389         return true;
 390     }
 391
 392     /**
 393      * Drops a named index from a table.
 394      *
 395      * @param string $table name of the table the index is on.
 396      * @param string $name  name of the index
 397      *
 398      * @return boolean success flag
 399      */
 400
 401     public function dropIndex($table, $name)
 402     {
 403         $res = $this->conn->query("ALTER TABLE $table DROP INDEX $name");
 404
 405         if (PEAR::isError($res)) {
 406             throw new Exception($res->getMessage());
 407         }
 408
 409         return true;
 410     }
 411
 412     /**
 413      * Adds a column to a table
 414      *
 415      * @param string    $table     name of the table
 416      * @param ColumnDef $columndef Definition of the new
 417      *                             column.
 418      *
 419      * @return boolean success flag
 420      */
 421
 422     public function addColumn($table, $columndef)
 423     {
 424         $sql = "ALTER TABLE $table ADD COLUMN " . $this->_columnSql($columndef);
 425
 426         $res = $this->conn->query($sql);
 427
 428         if (PEAR::isError($res)) {
 429             throw new Exception($res->getMessage());
 430         }
 431
 432         return true;
 433     }
 434
 435     /**
 436      * Modifies a column in the schema.
 437      *
 438      * The name must match an existing column and table.
 439      *
 440      * @param string    $table     name of the table
 441      * @param ColumnDef $columndef new definition of the column.
 442      *
 443      * @return boolean success flag
 444      */
 445
 446     public function modifyColumn($table, $columndef)
 447     {
 448         $sql = "ALTER TABLE $table MODIFY COLUMN " .
 449           $this->_columnSql($columndef);
 450
 451         $res = $this->conn->query($sql);
 452
 453         if (PEAR::isError($res)) {
 454             throw new Exception($res->getMessage());
 455         }
 456
 457         return true;
 458     }
 459
 460     /**
 461      * Drops a column from a table
 462      *
 463      * The name must match an existing column.
 464      *
 465      * @param string $table      name of the table
 466      * @param string $columnName name of the column to drop
 467      *
 468      * @return boolean success flag
 469      */
 470
 471     public function dropColumn($table, $columnName)
 472     {
 473         $sql = "ALTER TABLE $table DROP COLUMN $columnName";
 474
 475         $res = $this->conn->query($sql);
 476
 477         if (PEAR::isError($res)) {
 478             throw new Exception($res->getMessage());
 479         }
 480
 481         return true;
 482     }
 483
 484     /**
 485      * Ensures that a table exists with the given
 486      * name and the given column definitions.
 487      *
 488      * If the table does not yet exist, it will
 489      * create the table. If it does exist, it will
 490      * alter the table to match the column definitions.
 491      *
 492      * @param string $tableName name of the table
 493      * @param array  $def       Table definition array
 494      *
 495      * @return boolean success flag
 496      */
 497
 498     public function ensureTable($tableName, $def)
 499     {
 500         $statements = $this->buildEnsureTable($tableName, $def);
 501         return $this->runSqlSet($statements);
 502     }
 503
 504     /**
 505      * Run a given set of SQL commands on the connection in sequence.
 506      * Empty input is ok.
 507      *
 508      * @fixme if multiple statements, wrap in a transaction?
 509      * @param array $statements
 510      * @return boolean success flag
 511      */
 512     function runSqlSet(array $statements)
 513     {
 514         $ok = true;
 515         foreach ($statements as $sql) {
 516             if (defined('DEBUG_INSTALLER')) {
 517                 echo "<tt>" . htmlspecialchars($sql) . "</tt><br/>\n";
 518             }
 519             $res = $this->conn->query($sql);
 520
 521             if (PEAR::isError($res)) {
 522                 throw new Exception($res->getMessage());
 523             }
 524         }
 525         return $ok;
 526     }
 527
 528     /**
 529      * Check a table's status, and if needed build a set
 530      * of SQL statements which change it to be consistent
 531      * with the given table definition.
 532      *
 533      * If the table does not yet exist, statements will
 534      * be returned to create the table. If it does exist,
 535      * statements will be returned to alter the table to
 536      * match the column definitions.
 537      *
 538      * @param string $tableName name of the table
 539      * @param array  $columns   array of ColumnDef
 540      *                          objects for the table
 541      *
 542      * @return array of SQL statements
 543      */
 544
 545     function buildEnsureTable($tableName, array $def)
 546     {
 547         try {
 548             $old = $this->getTableDef($tableName);
 549         } catch (SchemaTableMissingException $e) {
 550             return $this->buildCreateTable($tableName, $def);
 551         }
 552
 553         // Filter the DB-independent table definition to match the current
 554         // database engine's features and limitations.
 555         $def = $this->validateDef($tableName, $def);
 556         $def = $this->filterDef($def);
 557
 558         $statements = array();
 559         $fields = $this->diffArrays($old, $def, 'fields', array($this, 'columnsEqual'));
 560         $uniques = $this->diffArrays($old, $def, 'unique keys');
 561         $indexes = $this->diffArrays($old, $def, 'indexes');
 562         $foreign = $this->diffArrays($old, $def, 'foreign keys');
 563
 564         // Drop any obsolete or modified indexes ahead...
 565         foreach ($indexes['del'] + $indexes['mod'] as $indexName) {
 566             $this->appendDropIndex($statements, $tableName, $indexName);
 567         }
 568
 569         // For efficiency, we want this all in one
 570         // query, instead of using our methods.
 571
 572         $phrase = array();
 573
 574         foreach ($foreign['del'] + $foreign['mod'] as $keyName) {
 575             $this->appendAlterDropForeign($phrase, $keyName);
 576         }
 577
 578         foreach ($uniques['del'] + $uniques['mod'] as $keyName) {
 579             $this->appendAlterDropUnique($phrase, $keyName);
 580         }
 581
 582         foreach ($fields['add'] as $columnName) {
 583             $this->appendAlterAddColumn($phrase, $columnName,
 584                     $def['fields'][$columnName]);
 585         }
 586
 587         foreach ($fields['mod'] as $columnName) {
 588             $this->appendAlterModifyColumn($phrase, $columnName,
 589                     $old['fields'][$columnName],
 590                     $def['fields'][$columnName]);
 591         }
 592
 593         foreach ($fields['del'] as $columnName) {
 594             $this->appendAlterDropColumn($phrase, $columnName);
 595         }
 596
 597         foreach ($uniques['mod'] + $uniques['add'] as $keyName) {
 598             $this->appendAlterAddUnique($phrase, $keyName, $def['unique keys'][$keyName]);
 599         }
 600
 601         foreach ($foreign['mod'] + $foreign['add'] as $keyName) {
 602             $this->appendAlterAddForeign($phrase, $keyName, $def['foreign keys'][$keyName]);
 603         }
 604
 605         $this->appendAlterExtras($phrase, $tableName, $def);
 606
 607         if (count($phrase) > 0) {
 608             $sql = 'ALTER TABLE ' . $tableName . ' ' . implode(",\n", $phrase);
 609             $statements[] = $sql;
 610         }
 611
 612         // Now create any indexes...
 613         foreach ($indexes['mod'] + $indexes['add'] as $indexName) {
 614             $this->appendCreateIndex($statements, $tableName, $indexName, $def['indexes'][$indexName]);
 615         }
 616
 617         return $statements;
 618     }
 619
 620     function diffArrays($oldDef, $newDef, $section, $compareCallback=null)
 621     {
 622         $old = isset($oldDef[$section]) ? $oldDef[$section] : array();
 623         $new = isset($newDef[$section]) ? $newDef[$section] : array();
 624
 625         $oldKeys = array_keys($old);
 626         $newKeys = array_keys($new);
 627
 628         $toadd  = array_diff($newKeys, $oldKeys);
 629         $todrop = array_diff($oldKeys, $newKeys);
 630         $same   = array_intersect($newKeys, $oldKeys);
 631         $tomod  = array();
 632         $tokeep = array();
 633
 634         // Find which fields have actually changed definition
 635         // in a way that we need to tweak them for this DB type.
 636         foreach ($same as $name) {
 637             if ($compareCallback) {
 638                 $same = call_user_func($compareCallback, $old[$name], $new[$name]);
 639             } else {
 640                 $same = ($old[$name] == $new[$name]);
 641             }
 642             if ($same) {
 643                 $tokeep[] = $name;
 644                 continue;
 645             }
 646             $tomod[] = $name;
 647         }
 648         return array('add' => $toadd,
 649                      'del' => $todrop,
 650                      'mod' => $tomod,
 651                      'keep' => $tokeep,
 652                      'count' => count($toadd) + count($todrop) + count($tomod));
 653     }
 654
 655     /**
 656      * Append phrase(s) to an array of partial ALTER TABLE chunks in order
 657      * to add the given column definition to the table.
 658      *
 659      * @param array $phrase
 660      * @param string $columnName
 661      * @param array $cd
 662      */
 663     function appendAlterAddColumn(array &$phrase, $columnName, array $cd)
 664     {
 665         $phrase[] = 'ADD COLUMN ' .
 666                     $this->quoteIdentifier($columnName) .
 667                     ' ' .
 668                     $this->columnSql($cd);
 669     }
 670
 671     /**
 672      * Append phrase(s) to an array of partial ALTER TABLE chunks in order
 673      * to alter the given column from its old state to a new one.
 674      *
 675      * @param array $phrase
 676      * @param string $columnName
 677      * @param array $old previous column definition as found in DB
 678      * @param array $cd current column definition
 679      */
 680     function appendAlterModifyColumn(array &$phrase, $columnName, array $old, array $cd)
 681     {
 682         $phrase[] = 'MODIFY COLUMN ' .
 683                     $this->quoteIdentifier($columnName) .
 684                     ' ' .
 685                     $this->columnSql($cd);
 686     }
 687
 688     /**
 689      * Append phrase(s) to an array of partial ALTER TABLE chunks in order
 690      * to drop the given column definition from the table.
 691      *
 692      * @param array $phrase
 693      * @param string $columnName
 694      */
 695     function appendAlterDropColumn(array &$phrase, $columnName)
 696     {
 697         $phrase[] = 'DROP COLUMN ' . $this->quoteIdentifier($columnName);
 698     }
 699
 700     function appendAlterAddUnique(array &$phrase, $keyName, array $def)
 701     {
 702         $sql = array();
 703         $sql[] = 'ADD';
 704         $this->appendUniqueKeyDef($sql, $keyName, $def);
 705         $phrase[] = implode(' ', $sql);
 706     }
 707
 708     function appendAlterAddForeign(array &$phrase, $keyName, array $def)
 709     {
 710         $sql = array();
 711         $sql[] = 'ADD';
 712         $this->appendForeignKeyDef($sql, $keyName, $def);
 713         $phrase[] = implode(' ', $sql);
 714     }
 715
 716     function appendAlterDropUnique(array &$phrase, $keyName)
 717     {
 718         $phrase[] = 'DROP CONSTRAINT ' . $keyName;
 719     }
 720
 721     function appendAlterDropForeign(array &$phrase, $keyName)
 722     {
 723         $phrase[] = 'DROP FOREIGN KEY ' . $keyName;
 724     }
 725
 726     function appendAlterExtras(array &$phrase, $tableName, array $def)
 727     {
 728         // no-op
 729     }
 730
 731     /**
 732      * Quote a db/table/column identifier if necessary.
 733      *
 734      * @param string $name
 735      * @return string
 736      */
 737     function quoteIdentifier($name)
 738     {
 739         return $name;
 740     }
 741
 742     function quoteDefaultValue($cd)
 743     {
 744         if ($cd['type'] == 'datetime' && $cd['default'] == 'CURRENT_TIMESTAMP') {
 745             return $cd['default'];
 746         } else {
 747             return $this->quoteValue($cd['default']);
 748         }
 749     }
 750
 751     function quoteValue($val)
 752     {
 753         return $this->conn->quoteSmart($val);
 754     }
 755
 756     /**
 757      * Check if two column definitions are equivalent.
 758      * The default implementation checks _everything_ but in many cases
 759      * you may be able to discard a bunch of equivalencies.
 760      *
 761      * @param array $a
 762      * @param array $b
 763      * @return boolean
 764      */
 765     function columnsEqual(array $a, array $b)
 766     {
 767         return !array_diff_assoc($a, $b) && !array_diff_assoc($b, $a);
 768     }
 769
 770     /**
 771      * Returns the array of names from an array of
 772      * ColumnDef objects.
 773      *
 774      * @param array $cds array of ColumnDef objects
 775      *
 776      * @return array strings for name values
 777      */
 778
 779     protected function _names($cds)
 780     {
 781         $names = array();
 782
 783         foreach ($cds as $cd) {
 784             $names[] = $cd->name;
 785         }
 786
 787         return $names;
 788     }
 789
 790     /**
 791      * Get a ColumnDef from an array matching
 792      * name.
 793      *
 794      * @param array  $cds  Array of ColumnDef objects
 795      * @param string $name Name of the column
 796      *
 797      * @return ColumnDef matching item or null if no match.
 798      */
 799
 800     protected function _byName($cds, $name)
 801     {
 802         foreach ($cds as $cd) {
 803             if ($cd->name == $name) {
 804                 return $cd;
 805             }
 806         }
 807
 808         return null;
 809     }
 810
 811     /**
 812      * Return the proper SQL for creating or
 813      * altering a column.
 814      *
 815      * Appropriate for use in CREATE TABLE or
 816      * ALTER TABLE statements.
 817      *
 818      * @param ColumnDef $cd column to create
 819      *
 820      * @return string correct SQL for that column
 821      */
 822
 823     function columnSql(array $cd)
 824     {
 825         $line = array();
 826         $line[] = $this->typeAndSize($cd);
 827
 828         if (isset($cd['default'])) {
 829             $line[] = 'default';
 830             $line[] = $this->quoteDefaultValue($cd);
 831         } else if (!empty($cd['not null'])) {
 832             // Can't have both not null AND default!
 833             $line[] = 'not null';
 834         }
 835
 836         return implode(' ', $line);
 837     }
 838
 839     /**
 840      *
 841      * @param string $column canonical type name in defs
 842      * @return string native DB type name
 843      */
 844     function mapType($column)
 845     {
 846         return $column;
 847     }
 848
 849     function typeAndSize($column)
 850     {
 851         //$type = $this->mapType($column);
 852         $type = $column['type'];
 853         if (isset($column['size'])) {
 854             $type = $column['size'] . $type;
 855         }
 856         $lengths = array();
 857
 858         if (isset($column['precision'])) {
 859             $lengths[] = $column['precision'];
 860             if (isset($column['scale'])) {
 861                 $lengths[] = $column['scale'];
 862             }
 863         } else if (isset($column['length'])) {
 864             $lengths[] = $column['length'];
 865         }
 866
 867         if ($lengths) {
 868             return $type . '(' . implode(',', $lengths) . ')';
 869         } else {
 870             return $type;
 871         }
 872     }
 873
 874     /**
 875      * Convert an old-style set of ColumnDef objects into the current
 876      * Drupal-style schema definition array, for backwards compatibility
 877      * with plugins written for 0.9.x.
 878      *
 879      * @param string $tableName
 880      * @param array $defs: array of ColumnDef objects
 881      * @return array
 882      */
 883     protected function oldToNew($tableName, array $defs)
 884     {
 885         $table = array();
 886         $prefixes = array(
 887             'tiny',
 888             'small',
 889             'medium',
 890             'big',
 891         );
 892         foreach ($defs as $cd) {
 893             $column = array();
 894             $column['type'] = $cd->type;
 895             foreach ($prefixes as $prefix) {
 896                 if (substr($cd->type, 0, strlen($prefix)) == $prefix) {
 897                     $column['type'] = substr($cd->type, strlen($prefix));
 898                     $column['size'] = $prefix;
 899                     break;
 900                 }
 901             }
 902
 903             if ($cd->size) {
 904                 if ($cd->type == 'varchar' || $cd->type == 'char') {
 905                     $column['length'] = $cd->size;
 906                 }
 907             }
 908             if (!$cd->nullable) {
 909                 $column['not null'] = true;
 910             }
 911             if ($cd->auto_increment) {
 912                 $column['type'] = 'serial';
 913             }
 914             if ($cd->default) {
 915                 $column['default'] = $cd->default;
 916             }
 917             $table['fields'][$cd->name] = $column;
 918
 919             if ($cd->key == 'PRI') {
 920                 // If multiple columns are defined as primary key,
 921                 // we'll pile them on in sequence.
 922                 if (!isset($table['primary key'])) {
 923                     $table['primary key'] = array();
 924                 }
 925                 $table['primary key'][] = $cd->name;
 926             } else if ($cd->key == 'MUL') {
 927                 // Individual multiple-value indexes are only per-column
 928                 // using the old ColumnDef syntax.
 929                 $idx = "{$tableName}_{$cd->name}_idx";
 930                 $table['indexes'][$idx] = array($cd->name);
 931             } else if ($cd->key == 'UNI') {
 932                 // Individual unique-value indexes are only per-column
 933                 // using the old ColumnDef syntax.
 934                 $idx = "{$tableName}_{$cd->name}_idx";
 935                 $table['unique keys'][$idx] = array($cd->name);
 936             }
 937         }
 938
 939         return $table;
 940     }
 941
 942     /**
 943      * Filter the given table definition array to match features available
 944      * in this database.
 945      *
 946      * This lets us strip out unsupported things like comments, foreign keys,
 947      * or type variants that we wouldn't get back from getTableDef().
 948      *
 949      * @param array $tableDef
 950      */
 951     function filterDef(array $tableDef)
 952     {
 953         return $tableDef;
 954     }
 955
 956     /**
 957      * Validate a table definition array, checking for basic structure.
 958      *
 959      * If necessary, converts from an old-style array of ColumnDef objects.
 960      *
 961      * @param string $tableName
 962      * @param array $def: table definition array
 963      * @return array validated table definition array
 964      *
 965      * @throws Exception on wildly invalid input
 966      */
 967     function validateDef($tableName, array $def)
 968     {
 969         if (isset($def[0]) && $def[0] instanceof ColumnDef) {
 970             $def = $this->oldToNew($tableName, $def);
 971         }
 972
 973         // A few quick checks :D
 974         if (!isset($def['fields'])) {
 975             throw new Exception("Invalid table definition for $tableName: no fields.");
 976         }
 977
 978         return $def;
 979     }
 980
 981     function isNumericType($type)
 982     {
 983         $type = strtolower($type);
 984         $known = array('int', 'serial', 'numeric');
 985         return in_array($type, $known);
 986     }
 987
 988     /**
 989      * Pull info from the query into a fun-fun array of dooooom
 990      *
 991      * @param string $sql
 992      * @return array of arrays
 993      */
 994     protected function fetchQueryData($sql)
 995     {
 996         $res = $this->conn->query($sql);
 997         if (PEAR::isError($res)) {
 998             throw new Exception($res->getMessage());
 999         }
1000
1001         $out = array();
1002         $row = array();
1003         while ($res->fetchInto($row, DB_FETCHMODE_ASSOC)) {
1004             $out[] = $row;
1005         }
1006         $res->free();
1007
1008         return $out;
1009     }
1010
1011 }
1012
1013 class SchemaTableMissingException extends Exception
1014 {
1015     // no-op
1016 }
1017