silverstripe-postgresql/code/PostgreSQLDatabase.php

<?php

namespace SilverStripe\PostgreSQL;

use SilverStripe\Core\Config\Configurable;
use SilverStripe\Core\Injector\Injector;
use SilverStripe\ORM\DB;
use SilverStripe\ORM\DataObject;
use SilverStripe\ORM\ArrayList;
use SilverStripe\ORM\Connect\Database;
use SilverStripe\ORM\PaginatedList;
use ErrorException;
use Exception;

/**
 * PostgreSQL connector class.
 *
 * @package sapphire
 * @subpackage model
 */
class PostgreSQLDatabase extends Database
{
    use Configurable;

    /**
     * Database schema manager object
     *
     * @var PostgreSQLSchemaManager
     */
    protected $schemaManager;

    /**
     * The currently selected database schema name.
     *
     * @var string
     */
    protected $schema;

    /**
     * @var bool
     */
    protected $transactionNesting = 0;

    /**
     * Toggle if transactions are supported. Defaults to true.
     *
     * @var bool
     */
    protected $supportsTransactions = true;

    /**
     * Determines whether to check a database exists on the host by
     * querying the 'postgres' database and running createDatabase.
     *
     * Some locked down systems prevent access to the 'postgres' table in
     * which case you need to set this to false.
     *
     * If allow_query_master_postgres is false, and model_schema_as_database is also false,
     * then attempts to create or check databases beyond the initial connection will
     * result in a runtime error.
     *
     * @config
     * @var bool
     */
    private static $allow_query_master_postgres = true;

    /**
     * For instances where multiple databases are used beyond the initial connection
     * you may set this option to true to force database switches to switch schemas
     * instead of using databases. This may be useful if the database user does not
     * have cross-database permissions, and in cases where multiple databases are used
     * (such as in running test cases).
     *
     * If this is true then the database will only be set during the initial connection,
     * and attempts to change to this database will use the 'public' schema instead
     *
     * If this is false then errors may be generated during some cross database operations.
     */
    private static $model_schema_as_database = true;

    /**
     * Override the language that tsearch uses.  By default it is 'english, but
     * could be any of the supported languages that can be found in the
     * pg_catalog.pg_ts_config table.
     */
    private static $search_language = 'english';

    /*
     * Describe how T-search will work.
     * You can use either GiST or GIN, and '@@' (gist) or '@@@' (gin)
     * Combinations of these two will also work, so you'll need to pick
     * one which works best for you
     */
    private static $default_fts_cluster_method = 'GIN';

    /*
     * Describe how T-search will work.
     * You can use either GiST or GIN, and '@@' (gist) or '@@@' (gin)
     * Combinations of these two will also work, so you'll need to pick
     * one which works best for you
     */
    private static $default_fts_search_method = '@@@';

    const MASTER_DATABASE = 'postgres';

    const MASTER_SCHEMA = 'public';

    /**
     * Full text cluster method. (e.g. GIN or GiST)
     *
     * @return string
     */
    public static function default_fts_cluster_method()
    {
        return static::config()->default_fts_cluster_method;
    }

    /**
     * Full text search method.
     *
     * @return string
     */
    public static function default_fts_search_method()
    {
        return static::config()->default_fts_search_method;
    }

    /**
     * Determines whether to check a database exists on the host by
     * querying the 'postgres' database and running createDatabase.
     *
     * Some locked down systems prevent access to the 'postgres' table in
     * which case you need to set this to false.
     *
     * If allow_query_master_postgres is false, and model_schema_as_database is also false,
     * then attempts to create or check databases beyond the initial connection will
     * result in a runtime error.
     *
     * @return bool
     */
    public static function allow_query_master_postgres()
    {
        return static::config()->allow_query_master_postgres;
    }

    /**
     * For instances where multiple databases are used beyond the initial connection
     * you may set this option to true to force database switches to switch schemas
     * instead of using databases. This may be useful if the database user does not
     * have cross-database permissions, and in cases where multiple databases are used
     * (such as in running test cases).
     *
     * If this is true then the database will only be set during the initial connection,
     * and attempts to change to this database will use the 'public' schema instead
     *
     * @return bool
     */
    public static function model_schema_as_database()
    {
        return static::config()->model_schema_as_database;
    }

    /**
     * Override the language that tsearch uses.  By default it is 'english, but
     * could be any of the supported languages that can be found in the
     * pg_catalog.pg_ts_config table.
     *
     * @return string
     */
    public static function search_language()
    {
        return static::config()->search_language;
    }

    /**
     * The database name specified at initial connection
     *
     * @var string
     */
    protected $databaseOriginal = '';

    /**
     * The schema name specified at initial construction. When model_schema_as_database
     * is set to true selecting the $databaseOriginal database will instead reset
     * the schema to this
     *
     * @var string
     */
    protected $schemaOriginal = '';

    /**
     * Connection parameters specified at inital connection
     *
     * @var array
     */
    protected $parameters = array();

    public function connect($parameters)
    {
        // Check database name
        if (empty($parameters['database'])) {
            // Check if we can use the master database
            if (!self::allow_query_master_postgres()) {
                throw new ErrorException('PostegreSQLDatabase::connect called without a database name specified');
            }
            // Fallback to master database connection if permission allows
            $parameters['database'] = self::MASTER_DATABASE;
        }
        $this->databaseOriginal = $parameters['database'];

        // check schema name
        if (empty($parameters['schema'])) {
            $parameters['schema'] = self::MASTER_SCHEMA;
        }
        $this->schemaOriginal = $parameters['schema'];

        // Ensure that driver is available (required by PDO)
        if (empty($parameters['driver'])) {
            $parameters['driver'] = $this->getDatabaseServer();
        }

        // Ensure port number is set (required by postgres)
        if (empty($parameters['port'])) {
            $parameters['port'] = 5432;
        }

        $this->parameters = $parameters;

        // If allowed, check that the database exists. Otherwise naively assume
        // that the original database exists
        if (self::allow_query_master_postgres()) {
            // Use master connection to setup initial schema
            $this->connectMaster();
            if (!$this->schemaManager->postgresDatabaseExists($this->databaseOriginal)) {
                $this->schemaManager->createPostgresDatabase($this->databaseOriginal);
            }
        }

        // Connect to the actual database we're requesting
        $this->connectDefault();

        // Set up the schema if required
        $this->setSchema($this->schemaOriginal, true);

        // Set the timezone if required.
        if (isset($parameters['timezone'])) {
            $this->selectTimezone($parameters['timezone']);
        }
    }

    protected function connectMaster()
    {
        $parameters = $this->parameters;
        $parameters['database'] = self::MASTER_DATABASE;
        $this->connector->connect($parameters, true);
    }

    protected function connectDefault()
    {
        $parameters = $this->parameters;
        $parameters['database'] = $this->databaseOriginal;
        $this->connector->connect($parameters, true);
    }

    /**
     * Sets the system timezone for the database connection
     *
     * @param string $timezone
     */
    public function selectTimezone($timezone)
    {
        if (empty($timezone)) {
            return;
        }
        $this->query("SET SESSION TIME ZONE '$timezone';");
    }

    public function supportsCollations()
    {
        return true;
    }

    public function supportsTimezoneOverride()
    {
        return true;
    }

    public function getDatabaseServer()
    {
        return "pgsql";
    }

    /**
     * Returns the name of the current schema in use
     *
     * @return string Name of current schema
     */
    public function currentSchema()
    {
        return $this->schema;
    }

    /**
     * Utility method to manually set the schema to an alternative
     * Check existance & sets search path to the supplied schema name
     *
     * @param string $schema Name of the schema
     * @param boolean $create Flag indicating whether the schema should be created
     * if it doesn't exist. If $create is false and the schema doesn't exist
     * then an error will be raised
     * @param int|boolean $errorLevel The level of error reporting to enable for
     * the query, or false if no error should be raised
     * @return boolean Flag indicating success
     */
    public function setSchema($schema, $create = false, $errorLevel = E_USER_ERROR)
    {
        if (!$this->schemaManager->schemaExists($schema)) {
            // Check DB creation permisson
            if (!$create) {
                if ($errorLevel !== false) {
                    user_error("Schema $schema does not exist", $errorLevel);
                }
                $this->schema = null;
                return false;
            }
            $this->schemaManager->createSchema($schema);
        }
        $this->setSchemaSearchPath($schema);
        $this->schema = $schema;
        return true;
    }

    /**
     * Override the schema search path. Search using the arguments supplied.
     * NOTE: The search path is normally set through setSchema() and only
     * one schema is selected. The facility to add more than one schema to
     * the search path is provided as an advanced PostgreSQL feature for raw
     * SQL queries. Sapphire cannot search for datamodel tables in alternate
     * schemas, so be wary of using alternate schemas within the ORM environment.
     *
     * @param string ...$arg Schema name to use. Add additional schema names as extra arguments.
     */
    public function setSchemaSearchPath($arg = null)
    {
        if (!$arg) {
            user_error('At least one Schema must be supplied to set a search path.', E_USER_ERROR);
        }
        $schemas = array_values(func_get_args());
        $this->query("SET search_path TO \"" . implode("\",\"", $schemas) . "\"");
    }

    /**
     * The core search engine configuration.
     * @todo Properly extract the search functions out of the core.
     *
     * @param array $classesToSearch
     * @param string $keywords Keywords as a space separated string
     * @param int $start
     * @param int $pageLength
     * @param string $sortBy
     * @param string $extraFilter
     * @param bool $booleanSearch
     * @param string $alternativeFileFilter
     * @param bool $invertedMatch
     * @return PaginatedList List of result pages
     * @throws Exception
     */
    public function searchEngine($classesToSearch, $keywords, $start, $pageLength, $sortBy = "ts_rank DESC", $extraFilter = "", $booleanSearch = false, $alternativeFileFilter = "", $invertedMatch = false)
    {
        $start = (int)$start;
        $pageLength = (int)$pageLength;

        //Fix the keywords to be ts_query compatitble:
        //Spaces must have pipes
        //@TODO: properly handle boolean operators here.
        $keywords= trim($keywords);
        $keywords= str_replace(' ', ' | ', $keywords);
        $keywords= str_replace('"', "'", $keywords);


        $keywords = $this->quoteString(trim($keywords));

        // Get tables
        $tablesToSearch = [];
        foreach ($classesToSearch as $class) {
            $tablesToSearch[$class] = DataObject::getSchema()->baseDataTable($class);
        }

        //We can get a list of all the tsvector columns though this query:
        //We know what tables to search in based on the $classesToSearch variable:
        $classesPlaceholders = DB::placeholders($classesToSearch);
        $searchableColumns = $this->preparedQuery(
            "
            SELECT table_name, column_name, data_type
            FROM information_schema.columns
            WHERE data_type='tsvector' AND table_name in ($classesPlaceholders);",
            array_values($tablesToSearch)
        );
        if (!$searchableColumns->numRecords()) {
            throw new Exception('there are no full text columns to search');
        }

        $tables = array();
        $tableParameters = array();

        // Make column selection lists
        $pageClass = 'SilverStripe\\CMS\\Model\\SiteTree';
        $fileClass = 'SilverStripe\\Assets\\File';
        $select = array(
            $pageClass => array(
                '"ClassName"',
                '"' . $tablesToSearch[$pageClass] . '"."ID"',
                '"ParentID"',
                '"Title"',
                '"URLSegment"',
                '"Content"',
                '"LastEdited"',
                '"Created"',
                'NULL AS "Name"',
                '"CanViewType"'
            ),
            $fileClass => array(
                '"ClassName"',
                '"' . $tablesToSearch[$fileClass] . '"."ID"',
                '0 AS "ParentID"',
                '"Title"',
                'NULL AS "URLSegment"',
                'NULL AS "Content"',
                '"LastEdited"',
                '"Created"',
                '"Name"',
                'NULL AS "CanViewType"'
            )
        );

        foreach ($searchableColumns as $searchableColumn) {
            $conditions = array();
            $tableName = $searchableColumn['table_name'];
            $columnName = $searchableColumn['column_name'];
            $className = DataObject::getSchema()->tableClass($tableName);
            if (DataObject::getSchema()->fieldSpec($className, 'ShowInSearch')) {
                $conditions[] = array('"ShowInSearch"' => 1);
            }

            $method = self::default_fts_search_method();
            $conditions[] = "\"{$tableName}\".\"{$columnName}\" $method q ";
            $query = DataObject::get($className, $conditions)->dataQuery()->query();

            // Could parameterise this, but convention is only to to so for where conditions
            $query->addFrom(array(
                'q' => ", to_tsquery('" . self::search_language() . "', $keywords)"
            ));
            $query->setSelect(array());

            foreach ($select[$className] as $clause) {
                if (preg_match('/^(.*) +AS +"?([^"]*)"?/i', $clause, $matches)) {
                    $query->selectField($matches[1], $matches[2]);
                } else {
                    $query->selectField($clause);
                }
            }

            $query->selectField("ts_rank(\"{$tableName}\".\"{$columnName}\", q)", 'Relevance');
            $query->setOrderBy(array());

            //Add this query to the collection
            $tables[] = $query->sql($parameters);
            $tableParameters = array_merge($tableParameters, $parameters);
        }

        $limit = $pageLength;
        $offset = $start;

        if ($keywords) {
            $orderBy = " ORDER BY $sortBy";
        } else {
            $orderBy='';
        }

        $fullQuery = "SELECT *, count(*) OVER() as _fullcount FROM (" . implode(" UNION ", $tables) . ") AS q1 $orderBy LIMIT $limit OFFSET $offset";

        // Get records
        $records = $this->preparedQuery($fullQuery, $tableParameters);
        $totalCount = 0;
        $objects = [];
        foreach ($records as $record) {
            $objects[] = Injector::inst()->createWithArgs($record['ClassName'], [$record]);
            $totalCount = $record['_fullcount'];
        }

        if ($objects) {
            $results = new ArrayList($objects);
        } else {
            $results = new ArrayList();
        }
        $list = new PaginatedList($results);
        $list->setLimitItems(false);
        $list->setPageStart($start);
        $list->setPageLength($pageLength);
        $list->setTotalItems($totalCount);
        return $list;
    }

    public function supportsTransactions()
    {
        return $this->supportsTransactions;
    }

    /*
     * This is a quick lookup to discover if the database supports particular extensions
     */
    public function supportsExtensions($extensions = array('partitions', 'tablespaces', 'clustering'))
    {
        if (isset($extensions['partitions'])) {
            return true;
        } elseif (isset($extensions['tablespaces'])) {
            return true;
        } elseif (isset($extensions['clustering'])) {
            return true;
        } else {
            return false;
        }
    }

    public function transactionStart($transaction_mode = false, $session_characteristics = false)
    {
        if ($this->transactionNesting > 0) {
            $this->transactionSavepoint('NESTEDTRANSACTION' . $this->transactionNesting);
        } else {
            $this->query('BEGIN;');

            if ($transaction_mode) {
                $this->query("SET TRANSACTION {$transaction_mode};");
            }

            if ($session_characteristics) {
                $this->query("SET SESSION CHARACTERISTICS AS TRANSACTION {$session_characteristics};");
            }
        }
        ++$this->transactionNesting;
    }

    public function transactionSavepoint($savepoint)
    {
        $this->query("SAVEPOINT {$savepoint};");
    }

    public function transactionRollback($savepoint = false)
    {
        // Named savepoint
        if ($savepoint) {
            $this->query('ROLLBACK TO ' . $savepoint);
            return true;
        }

        // Abort if unable to unnest, otherwise jump up a level
        if (!$this->transactionNesting) {
            return false;
        }
        --$this->transactionNesting;

        // Rollback nested
        if ($this->transactionNesting > 0) {
            return $this->transactionRollback('NESTEDTRANSACTION' . $this->transactionNesting);
        }

        // Rollback top level
        $this->query('ROLLBACK');
        return true;
    }

    public function transactionDepth()
    {
        return $this->transactionNesting;
    }

    public function transactionEnd($chain = false)
    {
        --$this->transactionNesting;
        if ($this->transactionNesting <= 0) {
            $this->transactionNesting = 0;
            $this->query('COMMIT;');
        }
    }

    public function comparisonClause($field, $value, $exact = false, $negate = false, $caseSensitive = null, $parameterised = false)
    {
        if ($exact && $caseSensitive === null) {
            $comp = ($negate) ? '!=' : '=';
        } else {
            $comp = ($caseSensitive === true) ? 'LIKE' : 'ILIKE';
            if ($negate) {
                $comp = 'NOT ' . $comp;
            }
            $field.='::text';
        }

        if ($parameterised) {
            return sprintf("%s %s ?", $field, $comp);
        } else {
            return sprintf("%s %s '%s'", $field, $comp, $value);
        }
    }

    /**
     * Function to return an SQL datetime expression that can be used with Postgres
     * used for querying a datetime in a certain format
     * @param string $date to be formated, can be either 'now', literal datetime like '1973-10-14 10:30:00' or field name, e.g. '"SiteTree"."Created"'
     * @param string $format to be used, supported specifiers:
     * %Y = Year (four digits)
     * %m = Month (01..12)
     * %d = Day (01..31)
     * %H = Hour (00..23)
     * %i = Minutes (00..59)
     * %s = Seconds (00..59)
     * %U = unix timestamp, can only be used on it's own
     * @return string SQL datetime expression to query for a formatted datetime
     */
    public function formattedDatetimeClause($date, $format)
    {
        preg_match_all('/%(.)/', $format, $matches);
        foreach ($matches[1] as $match) {
            if (array_search($match, array('Y','m','d','H','i','s','U')) === false) {
                user_error('formattedDatetimeClause(): unsupported format character %' . $match, E_USER_WARNING);
            }
        }

        $translate = array(
            '/%Y/' => 'YYYY',
            '/%m/' => 'MM',
            '/%d/' => 'DD',
            '/%H/' => 'HH24',
            '/%i/' => 'MI',
            '/%s/' => 'SS',
        );
        $format = preg_replace(array_keys($translate), array_values($translate), $format);

        if (preg_match('/^now$/i', $date)) {
            $date = "NOW()";
        } elseif (preg_match('/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$/i', $date)) {
            $date = "TIMESTAMP '$date'";
        }

        if ($format == '%U') {
            return "FLOOR(EXTRACT(epoch FROM $date))";
        }

        return "to_char($date, TEXT '$format')";
    }

    /**
     * Function to return an SQL datetime expression that can be used with Postgres
     * used for querying a datetime addition
     * @param string $date, can be either 'now', literal datetime like '1973-10-14 10:30:00' or field name, e.g. '"SiteTree"."Created"'
     * @param string $interval to be added, use the format [sign][integer] [qualifier], e.g. -1 Day, +15 minutes, +1 YEAR
     * supported qualifiers:
     * - years
     * - months
     * - days
     * - hours
     * - minutes
     * - seconds
     * This includes the singular forms as well
     * @return string SQL datetime expression to query for a datetime (YYYY-MM-DD hh:mm:ss) which is the result of the addition
     */
    public function datetimeIntervalClause($date, $interval)
    {
        if (preg_match('/^now$/i', $date)) {
            $date = "NOW()";
        } elseif (preg_match('/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$/i', $date)) {
            $date = "TIMESTAMP '$date'";
        }

        // ... when being too precise becomes a pain. we need to cut of the fractions.
        // TIMESTAMP(0) doesn't work because it rounds instead flooring
        return "CAST(SUBSTRING(CAST($date + INTERVAL '$interval' AS VARCHAR) FROM 1 FOR 19) AS TIMESTAMP)";
    }

    /**
     * Function to return an SQL datetime expression that can be used with Postgres
     * used for querying a datetime substraction
     * @param string $date1, can be either 'now', literal datetime like '1973-10-14 10:30:00' or field name, e.g. '"SiteTree"."Created"'
     * @param string $date2 to be substracted of $date1, can be either 'now', literal datetime like '1973-10-14 10:30:00' or field name, e.g. '"SiteTree"."Created"'
     * @return string SQL datetime expression to query for the interval between $date1 and $date2 in seconds which is the result of the substraction
     */
    public function datetimeDifferenceClause($date1, $date2)
    {
        if (preg_match('/^now$/i', $date1)) {
            $date1 = "NOW()";
        } elseif (preg_match('/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$/i', $date1)) {
            $date1 = "TIMESTAMP '$date1'";
        }

        if (preg_match('/^now$/i', $date2)) {
            $date2 = "NOW()";
        } elseif (preg_match('/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$/i', $date2)) {
            $date2 = "TIMESTAMP '$date2'";
        }

        return "(FLOOR(EXTRACT(epoch FROM $date1)) - FLOOR(EXTRACT(epoch from $date2)))";
    }

    public function now()
    {
        return 'NOW()';
    }

    public function random()
    {
        return 'RANDOM()';
    }

    /**
     * Determines the name of the current database to be reported externally
     * by substituting the schema name for the database name.
     * Should only be used when model_schema_as_database is true
     *
     * @param string $schema Name of the schema
     * @return string Name of the database to report
     */
    public function schemaToDatabaseName($schema)
    {
        switch ($schema) {
            case $this->schemaOriginal:
                return $this->databaseOriginal;
            default:
                return $schema;
        }
    }

    /**
     * Translates a requested database name to a schema name to substitute internally.
     * Should only be used when model_schema_as_database is true
     *
     * @param string $database Name of the database
     * @return string Name of the schema to use for this database internally
     */
    public function databaseToSchemaName($database)
    {
        switch ($database) {
            case $this->databaseOriginal:
                return $this->schemaOriginal;
            default:
                return $database;
        }
    }

    public function dropSelectedDatabase()
    {
        if (self::model_schema_as_database()) {
            // Check current schema is valid
            $oldSchema = $this->schema;
            if (empty($oldSchema)) {
                return;
            } // Nothing selected to drop

            // Select another schema
            if ($oldSchema !== $this->schemaOriginal) {
                $this->setSchema($this->schemaOriginal);
            } elseif ($oldSchema !== self::MASTER_SCHEMA) {
                $this->setSchema(self::MASTER_SCHEMA);
            } else {
                $this->schema = null;
            }

            // Remove this schema
            $this->schemaManager->dropSchema($oldSchema);
        } else {
            parent::dropSelectedDatabase();
        }
    }

    public function getSelectedDatabase()
    {
        if (self::model_schema_as_database()) {
            return $this->schemaToDatabaseName($this->schema);
        }
        return parent::getSelectedDatabase();
    }

    public function selectDatabase($name, $create = false, $errorLevel = E_USER_ERROR)
    {
        // Substitute schema here as appropriate
        if (self::model_schema_as_database()) {
            // Selecting the database itself should be treated as selecting the public schema
            $schemaName = $this->databaseToSchemaName($name);
            return $this->setSchema($schemaName, $create, $errorLevel);
        }

        // Database selection requires that a new connection is established.
        // This is not ideal postgres practise
        if (!$this->schemaManager->databaseExists($name)) {
            // Check DB creation permisson
            if (!$create) {
                if ($errorLevel !== false) {
                    user_error("Attempted to connect to non-existing database \"$name\"", $errorLevel);
                }
                // Unselect database
                $this->connector->unloadDatabase();
                return false;
            }
            $this->schemaManager->createDatabase($name);
        }

        // New connection made here, treating the new database name as the new original
        $this->databaseOriginal = $name;
        $this->connectDefault();
        return true;
    }

    /**
     * Delete all entries from the table instead of truncating it.
     *
     * This gives a massive speed improvement compared to using TRUNCATE, with
     * the caveat that primary keys are not reset etc.
     *
     * @see DatabaseAdmin::clearAllData()
     *
     * @param string $table
     */
    public function clearTable($table)
    {
        $this->query('DELETE FROM "'.$table.'";');
    }
}