- Patch #342503 by Josh Waihi, Damien Tournoud et al: schema function findTables fails on PostgreSQL.

4 files changed, 107 insertions, 38 deletions

diff --git a/includes/database/database.inc b/includes/database/database.inc
index 25f570e5d..669245629 100644
--- a/includes/database/database.inc
+++ b/includes/database/database.inc
@@ -380,14 +380,14 @@ abstract class DatabaseConnection extends PDO {
    * Queries sent to Drupal should wrap all table names in curly brackets. This
    * function searches for this syntax and adds Drupal's table prefix to all
    * tables, allowing Drupal to coexist with other systems in the same database
-   * if necessary.
+   * and/or schema if necessary.
    *
    * @param $sql
    *   A string containing a partial or entire SQL query.
    * @return
    *   The properly-prefixed string.
    */
-  protected function prefixTables($sql) {
+  public function prefixTables($sql) {
     global $db_prefix;
 
     if (is_array($db_prefix)) {
@@ -428,7 +428,7 @@ abstract class DatabaseConnection extends PDO {
    *   A PDO prepared statement ready for its execute() method.
    */
   protected function prepareQuery($query, $cache = TRUE) {
-    $query = self::prefixTables($query);
+    $query = $this->prefixTables($query);
     if (empty($this->preparedStatements[$query])) {
       // Call PDO::prepare.
       $this->preparedStatements[$query] = parent::prepare($query);
diff --git a/includes/database/mysql/schema.inc b/includes/database/mysql/schema.inc
index 3e19f05d0..b12aa525a 100644
--- a/includes/database/mysql/schema.inc
+++ b/includes/database/mysql/schema.inc
@@ -14,14 +14,29 @@
 
 class DatabaseSchema_mysql extends DatabaseSchema {
 
-  public function tableExists($table) {
-    return (bool) $this->connection->query("SHOW TABLES LIKE '{" . $table . "}'", array(), array())->fetchField();
-  }
+  /**
+   * Build a condition to match a table name against a standard information_schema. 
+   *
+   * MySQL uses databases like schemas rather than catalogs so when we build
+   * a condition to query the information_schema.tables, we set the default
+   * database as the schema unless specified otherwise, and exclude table_catalog 
+   * from the condition criteria.
+   */
+  protected function buildTableNameCondition($table_name, $operator = '=') {
+    $info = Database::getConnectionInfo();
 
-  public function columnExists($table, $column) {
-    return (bool) $this->connection->query("SHOW COLUMNS FROM {" . $this->connection->escapeTable($table) . "} LIKE '" . $this->connection->escapeTable($column) . "'", array(), array())->fetchField();
-  }
+    if (strpos($table_name, '.')) {
+      list($schema, $table_name) = explode('.', $table_name);
+    }
+    else {
+      $schema = $info['default']['database'];
+    }
 
+    $condition = db_and()
+      ->condition('table_schema', $schema)
+      ->condition('table_name', $table_name, $operator);
+    return $condition;
+  }
 
   /**
    * Generate SQL to create a new table from a Drupal schema definition.
diff --git a/includes/database/pgsql/schema.inc b/includes/database/pgsql/schema.inc
index cbbf36dc9..00f4b9fef 100644
--- a/includes/database/pgsql/schema.inc
+++ b/includes/database/pgsql/schema.inc
@@ -13,14 +13,6 @@
 
 class DatabaseSchema_pgsql extends DatabaseSchema {
 
-  public function tableExists($table) {
-    return (bool) db_result(db_query("SELECT COUNT(*) FROM pg_class WHERE relname = '{" . db_escape_table($table) . "}'"));
-  }
-
-  public function columnExists($table, $column) {
-    return (bool) db_result(db_query("SELECT COUNT(pg_attribute.attname) FROM pg_class, pg_attribute WHERE pg_attribute.attrelid = pg_class.oid AND pg_class.relname = '{" . db_escape_table($table) . "}' AND attname = '" . db_escape_table($column) . "'"));
-  }
-
   /**
    * Generate SQL to create a new table from a Drupal schema definition.
    *
diff --git a/includes/database/schema.inc b/includes/database/schema.inc
index d0c94ef21..0ad730335 100644
--- a/includes/database/schema.inc
+++ b/includes/database/schema.inc
@@ -125,14 +125,95 @@ abstract class DatabaseSchema {
   }
 
   /**
+   * Build a condition to match a table name against a standard information_schema.
+   *
+   * The information_schema is a SQL standard that provides information about the
+   * database server and the databases, schemas, tables, columns and users within
+   * it. This makes information_schema a useful tool to use across the drupal 
+   * database drivers and is used by a few different functions. The function below
+   * describes the conditions to be meet when querying information_schema.tables
+   * for drupal tables or information associated with drupal tables. Even though
+   * this is the standard method, not all databases follow standards and so this
+   * method should be overwritten by a database driver if the database provider
+   * uses alternate methods. Because information_schema.tables is used in a few
+   * different functions, a database driver will only need to override this function
+   * to make all the others work. For example see includes/databases/mysql/schema.inc.
+   *
+   * @param $table_name
+   *   The name of the table to explode.
+   * @param $operator
+   *   The operator to apply on the 'table' part of the condition.
+   * @return
+   *   A DatabaseCondition object.
+   */
+  protected function buildTableNameCondition($table_name, $operator = '=') {
+    $info = Database::getConnectionInfo();
+
+    // The table name may describe the schema eg. schema.table.
+    if (strpos($table_name, '.')) {
+      list($schema, $table_name) = explode('.', $table_name);
+    }
+    else {
+      $schema = 'public';
+    }
+
+    $condition = db_and()
+      ->condition('table_catalog', $info['default']['database'])
+      ->condition('table_schema', $schema)
+      ->condition('table_name', $table_name, $operator);
+    return $condition;
+  }
+
+  /**
    * Check if a table exists.
+   * 
+   * @param $table
+   *   The name of the table in drupal (no prefixing).
+   * @return
+   *   false is no table exists otherwise the actual table name.
    */
-  abstract public function tableExists($table);
+  public function tableExists($table) {
+    $condition = $this->buildTableNameCondition($this->connection->prefixTables('{' . $table . '}'));
+    $condition->compile($this->connection);
+    // Normally, we would heartily discourage the use of string
+    // concatination for conditionals like this however, we 
+    // couldn't use db_select() here because it would prefix
+    // information_schema.tables and the query would fail.
+    return db_query("SELECT table_name FROM information_schema.tables WHERE " . (string) $condition, $condition->arguments())->fetchField();
+  }
+
+  /**
+   * Find all tables that are like the specified base table name.
+   *
+   * @param $table_expression
+   *   An SQL expression, for example "simpletest%" (without the quotes).
+   *   BEWARE: this is not prefixed, the caller should take care of that.
+   * @return
+   *   Array, both the keys and the values are the matching tables.
+   */
+  public function findTables($table_expression) {
+    $condition = $this->buildTableNameCondition($table_expression, 'LIKE');
+    $condition->compile($this->connection);
+    // Normally, we would heartily discourage the use of string
+    // concatination for conditionals like this however, we 
+    // couldn't use db_select() here because it would prefix
+    // information_schema.tables and the query would fail.
+    return db_query("SELECT table_name FROM information_schema.tables WHERE " . (string) $condition, $condition->arguments())->fetchAllKeyed(0, 0);
+  }
 
   /**
    * Check if a column exists in the given table.
    */
-  abstract public function columnExists($table, $column);
+  public function columnExists($table, $column) {
+    $condition = $this->buildTableNameCondition($this->connection->prefixTables('{' . $table . '}'));
+    $condition->condition('column_name', $column);
+    $condition->compile($this->connection);
+    // Normally, we would heartily discourage the use of string
+    // concatination for conditionals like this however, we 
+    // couldn't use db_select() here because it would prefix
+    // information_schema.tables and the query would fail.
+    return db_query("SELECT column_name FROM information_schema.columns WHERE " . (string) $condition, $condition->arguments())->fetchAllKeyed(0, 0);
+  }
 
   /**
    * This maps a generic data type in combination with its data size
@@ -404,25 +485,6 @@ abstract class DatabaseSchema {
     }
     return $ret;
   }
-
-  /**
-   * Find all tables that are like the specified base table name.
-   *
-   * @param $table_expression
-   *   An SQL expression, for example "simpletest%" (without the quotes).
-   *   BEWARE: this is not prefixed, the caller should take care of that.
-   * @return
-   *   Array, both the keys and the values are the matching tables.
-   */
-  public function findTables($table_expression) {
-    global $db_prefix;
-    $info = Database::getConnectionInfo();
-    $result = db_query("SELECT table_name FROM information_schema.tables WHERE table_schema = :database AND table_name LIKE :table_name", array(
-      ':database' => $info['default']['database'],
-      ':table_name' => $table_expression,
-    ));
-    return $result->fetchAllKeyed(0, 0);
-  }
 }
 
 /**