3 * This is the Postgres database abstraction layer.
5 * This program is free software; you can redistribute it and/or modify
6 * it under the terms of the GNU General Public License as published by
7 * the Free Software Foundation; either version 2 of the License, or
8 * (at your option) any later version.
10 * This program is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 * GNU General Public License for more details.
15 * You should have received a copy of the GNU General Public License along
16 * with this program; if not, write to the Free Software Foundation, Inc.,
17 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
18 * http://www.gnu.org/copyleft/gpl.html
23 use Wikimedia\WaitConditionLoop
;
28 class DatabasePostgres
extends Database
{
33 protected $mLastResult = null;
34 /** @var int The number of rows affected as an integer */
35 protected $mAffectedRows = null;
38 private $mInsertId = null;
39 /** @var float|string */
40 private $numericVersion = null;
41 /** @var string Connect string to open a PostgreSQL connection */
42 private $connectString;
46 public function __construct( array $params ) {
47 $this->port
= isset( $params['port'] ) ?
$params['port'] : false;
48 parent
::__construct( $params );
51 public function getType() {
55 public function implicitGroupby() {
59 public function implicitOrderby() {
63 public function hasConstraint( $name ) {
64 $conn = $this->getBindingHandle();
66 $sql = "SELECT 1 FROM pg_catalog.pg_constraint c, pg_catalog.pg_namespace n " .
67 "WHERE c.connamespace = n.oid AND conname = '" .
68 pg_escape_string( $conn, $name ) . "' AND n.nspname = '" .
69 pg_escape_string( $conn, $this->getCoreSchema() ) . "'";
70 $res = $this->doQuery( $sql );
72 return $this->numRows( $res );
75 public function open( $server, $user, $password, $dbName ) {
76 # Test for Postgres support, to avoid suppressed fatal error
77 if ( !function_exists( 'pg_connect' ) ) {
78 throw new DBConnectionError(
80 "Postgres functions missing, have you compiled PHP with the --with-pgsql\n" .
81 "option? (Note: if you recently installed PHP, you may need to restart your\n" .
82 "webserver and database)\n"
86 $this->mServer
= $server;
88 $this->mPassword
= $password;
89 $this->mDBname
= $dbName;
94 'password' => $password
96 if ( $server != false && $server != '' ) {
97 $connectVars['host'] = $server;
99 if ( (int)$this->port
> 0 ) {
100 $connectVars['port'] = (int)$this->port
;
102 if ( $this->mFlags
& self
::DBO_SSL
) {
103 $connectVars['sslmode'] = 1;
106 $this->connectString
= $this->makeConnectionString( $connectVars );
108 $this->installErrorHandler();
111 // Use new connections to let LoadBalancer/LBFactory handle reuse
112 $this->mConn
= pg_connect( $this->connectString
, PGSQL_CONNECT_FORCE_NEW
);
113 } catch ( Exception
$ex ) {
114 $this->restoreErrorHandler();
118 $phpError = $this->restoreErrorHandler();
120 if ( !$this->mConn
) {
121 $this->queryLogger
->debug(
122 "DB connection error\n" .
123 "Server: $server, Database: $dbName, User: $user, Password: " .
124 substr( $password, 0, 3 ) . "...\n"
126 $this->queryLogger
->debug( $this->lastError() . "\n" );
127 throw new DBConnectionError( $this, str_replace( "\n", ' ', $phpError ) );
130 $this->mOpened
= true;
132 # If called from the command-line (e.g. importDump), only show errors
133 if ( $this->cliMode
) {
134 $this->doQuery( "SET client_min_messages = 'ERROR'" );
137 $this->query( "SET client_encoding='UTF8'", __METHOD__
);
138 $this->query( "SET datestyle = 'ISO, YMD'", __METHOD__
);
139 $this->query( "SET timezone = 'GMT'", __METHOD__
);
140 $this->query( "SET standard_conforming_strings = on", __METHOD__
);
141 if ( $this->getServerVersion() >= 9.0 ) {
142 $this->query( "SET bytea_output = 'escape'", __METHOD__
); // PHP bug 53127
145 $this->determineCoreSchema( $this->mSchema
);
146 // The schema to be used is now in the search path; no need for explicit qualification
153 * Postgres doesn't support selectDB in the same way MySQL does. So if the
154 * DB name doesn't match the open connection, open a new one
158 public function selectDB( $db ) {
159 if ( $this->mDBname
!== $db ) {
160 return (bool)$this->open( $this->mServer
, $this->mUser
, $this->mPassword
, $db );
167 * @param string[] $vars
170 private function makeConnectionString( $vars ) {
172 foreach ( $vars as $name => $value ) {
173 $s .= "$name='" . str_replace( "'", "\\'", $value ) . "' ";
179 protected function closeConnection() {
180 return $this->mConn ?
pg_close( $this->mConn
) : true;
183 public function doQuery( $sql ) {
184 $conn = $this->getBindingHandle();
186 $sql = mb_convert_encoding( $sql, 'UTF-8' );
187 // Clear previously left over PQresult
188 while ( $res = pg_get_result( $conn ) ) {
189 pg_free_result( $res );
191 if ( pg_send_query( $conn, $sql ) === false ) {
192 throw new DBUnexpectedError( $this, "Unable to post new query to PostgreSQL\n" );
194 $this->mLastResult
= pg_get_result( $conn );
195 $this->mAffectedRows
= null;
196 if ( pg_result_error( $this->mLastResult
) ) {
200 return $this->mLastResult
;
203 protected function dumpError() {
207 PGSQL_DIAG_MESSAGE_PRIMARY
,
208 PGSQL_DIAG_MESSAGE_DETAIL
,
209 PGSQL_DIAG_MESSAGE_HINT
,
210 PGSQL_DIAG_STATEMENT_POSITION
,
211 PGSQL_DIAG_INTERNAL_POSITION
,
212 PGSQL_DIAG_INTERNAL_QUERY
,
214 PGSQL_DIAG_SOURCE_FILE
,
215 PGSQL_DIAG_SOURCE_LINE
,
216 PGSQL_DIAG_SOURCE_FUNCTION
218 foreach ( $diags as $d ) {
219 $this->queryLogger
->debug( sprintf( "PgSQL ERROR(%d): %s\n",
220 $d, pg_result_error_field( $this->mLastResult
, $d ) ) );
224 public function reportQueryError( $error, $errno, $sql, $fname, $tempIgnore = false ) {
226 /* Check for constraint violation */
227 if ( $errno === '23505' ) {
228 parent
::reportQueryError( $error, $errno, $sql, $fname, $tempIgnore );
233 /* Transaction stays in the ERROR state until rolled back */
234 if ( $this->mTrxLevel
) {
235 $ignore = $this->ignoreErrors( true );
236 $this->rollback( __METHOD__
);
237 $this->ignoreErrors( $ignore );
239 parent
::reportQueryError( $error, $errno, $sql, $fname, false );
242 public function freeResult( $res ) {
243 if ( $res instanceof ResultWrapper
) {
246 MediaWiki\
suppressWarnings();
247 $ok = pg_free_result( $res );
248 MediaWiki\restoreWarnings
();
250 throw new DBUnexpectedError( $this, "Unable to free Postgres result\n" );
254 public function fetchObject( $res ) {
255 if ( $res instanceof ResultWrapper
) {
258 MediaWiki\
suppressWarnings();
259 $row = pg_fetch_object( $res );
260 MediaWiki\restoreWarnings
();
261 # @todo FIXME: HACK HACK HACK HACK debug
263 # @todo hashar: not sure if the following test really trigger if the object
265 $conn = $this->getBindingHandle();
266 if ( pg_last_error( $conn ) ) {
267 throw new DBUnexpectedError(
269 'SQL error: ' . htmlspecialchars( pg_last_error( $conn ) )
276 public function fetchRow( $res ) {
277 if ( $res instanceof ResultWrapper
) {
280 MediaWiki\
suppressWarnings();
281 $row = pg_fetch_array( $res );
282 MediaWiki\restoreWarnings
();
284 $conn = $this->getBindingHandle();
285 if ( pg_last_error( $conn ) ) {
286 throw new DBUnexpectedError(
288 'SQL error: ' . htmlspecialchars( pg_last_error( $conn ) )
295 public function numRows( $res ) {
296 if ( $res instanceof ResultWrapper
) {
299 MediaWiki\
suppressWarnings();
300 $n = pg_num_rows( $res );
301 MediaWiki\restoreWarnings
();
303 $conn = $this->getBindingHandle();
304 if ( pg_last_error( $conn ) ) {
305 throw new DBUnexpectedError(
307 'SQL error: ' . htmlspecialchars( pg_last_error( $conn ) )
314 public function numFields( $res ) {
315 if ( $res instanceof ResultWrapper
) {
319 return pg_num_fields( $res );
322 public function fieldName( $res, $n ) {
323 if ( $res instanceof ResultWrapper
) {
327 return pg_field_name( $res, $n );
331 * Return the result of the last call to nextSequenceValue();
332 * This must be called after nextSequenceValue().
336 public function insertId() {
337 return $this->mInsertId
;
340 public function dataSeek( $res, $row ) {
341 if ( $res instanceof ResultWrapper
) {
345 return pg_result_seek( $res, $row );
348 public function lastError() {
349 if ( $this->mConn
) {
350 if ( $this->mLastResult
) {
351 return pg_result_error( $this->mLastResult
);
353 return pg_last_error();
357 return $this->getLastPHPError() ?
: 'No database connection';
360 public function lastErrno() {
361 if ( $this->mLastResult
) {
362 return pg_result_error_field( $this->mLastResult
, PGSQL_DIAG_SQLSTATE
);
368 public function affectedRows() {
369 if ( !is_null( $this->mAffectedRows
) ) {
370 // Forced result for simulated queries
371 return $this->mAffectedRows
;
373 if ( empty( $this->mLastResult
) ) {
377 return pg_affected_rows( $this->mLastResult
);
381 * Estimate rows in dataset
382 * Returns estimated count, based on EXPLAIN output
383 * This is not necessarily an accurate estimate, so use sparingly
384 * Returns -1 if count cannot be found
385 * Takes same arguments as Database::select()
387 * @param string $table
388 * @param string $vars
389 * @param string $conds
390 * @param string $fname
391 * @param array $options
394 public function estimateRowCount( $table, $vars = '*', $conds = '',
395 $fname = __METHOD__
, $options = []
397 $options['EXPLAIN'] = true;
398 $res = $this->select( $table, $vars, $conds, $fname, $options );
401 $row = $this->fetchRow( $res );
403 if ( preg_match( '/rows=(\d+)/', $row[0], $count ) ) {
404 $rows = (int)$count[1];
411 public function indexInfo( $table, $index, $fname = __METHOD__
) {
412 $sql = "SELECT indexname FROM pg_indexes WHERE tablename='$table'";
413 $res = $this->query( $sql, $fname );
417 foreach ( $res as $row ) {
418 if ( $row->indexname
== $this->indexName( $index ) ) {
426 public function indexAttributes( $index, $schema = false ) {
427 if ( $schema === false ) {
428 $schema = $this->getCoreSchema();
431 * A subquery would be not needed if we didn't care about the order
432 * of attributes, but we do
434 $sql = <<<__INDEXATTR__
438 i.indoption[s.g] as option,
441 (SELECT generate_series(array_lower(isub.indkey,1), array_upper(isub.indkey,1)) AS g
445 ON cis.oid=isub.indexrelid
447 ON cis.relnamespace = ns.oid
448 WHERE cis.relname='$index' AND ns.nspname='$schema') AS s,
454 ON ci.oid=i.indexrelid
456 ON ct.oid = i.indrelid
458 ON ci.relnamespace = n.oid
460 ci.relname='$index' AND n.nspname='$schema'
461 AND attrelid = ct.oid
462 AND i.indkey[s.g] = attnum
463 AND i.indclass[s.g] = opcls.oid
464 AND pg_am.oid = opcls.opcmethod
466 $res = $this->query( $sql, __METHOD__ );
469 foreach ( $res as $row ) {
483 public function indexUnique( $table, $index, $fname = __METHOD__ ) {
484 $sql = "SELECT indexname FROM pg_indexes WHERE tablename='{$table}'" .
485 " AND indexdef LIKE 'CREATE UNIQUE%(" .
486 $this->strencode( $this->indexName( $index ) ) .
488 $res = $this->query( $sql, $fname );
493 return $res->numRows() > 0;
496 public function selectSQLText(
497 $table, $vars, $conds = '', $fname = __METHOD__, $options = [], $join_conds = []
499 // Change the FOR UPDATE option as necessary based on the join conditions. Then pass
500 // to the parent function to get the actual SQL text.
501 // In Postgres when using FOR UPDATE, only the main table and tables that are inner joined
502 // can be locked. That means tables in an outer join cannot be FOR UPDATE locked. Trying to
503 // do so causes a DB error. This wrapper checks which tables can be locked and adjusts it
505 // MySQL uses "ORDER BY NULL" as an optimization hint, but that is illegal in PostgreSQL.
506 if ( is_array( $options ) ) {
507 $forUpdateKey = array_search( 'FOR UPDATE', $options, true );
508 if ( $forUpdateKey !== false && $join_conds ) {
509 unset( $options[$forUpdateKey] );
511 foreach ( $join_conds as $table_cond => $join_cond ) {
512 if ( 0 === preg_match( '/^(?:LEFT|RIGHT|FULL)(?: OUTER)? JOIN$/i', $join_cond[0] ) ) {
513 $options['FOR UPDATE'][] = $table_cond;
518 if ( isset( $options['ORDER BY'] ) && $options['ORDER BY'] == 'NULL' ) {
519 unset( $options['ORDER BY'] );
523 return parent::selectSQLText( $table, $vars, $conds, $fname, $options, $join_conds );
527 * INSERT wrapper, inserts an array into a table
529 * $args may be a single associative array, or an array of these with numeric keys,
530 * for multi-row insert (Postgres version 8.2 and above only).
532 * @param string $table Name of the table to insert to.
533 * @param array $args Items to insert into the table.
534 * @param string $fname Name of the function, for profiling
535 * @param array|string $options String or array. Valid options: IGNORE
536 * @return bool Success of insert operation. IGNORE always returns true.
538 public function insert( $table, $args, $fname = __METHOD__, $options = [] ) {
539 if ( !count( $args ) ) {
543 $table = $this->tableName( $table );
544 if ( !isset( $this->numericVersion ) ) {
545 $this->getServerVersion();
548 if ( !is_array( $options ) ) {
549 $options = [ $options ];
552 if ( isset( $args[0] ) && is_array( $args[0] ) ) {
554 $keys = array_keys( $args[0] );
557 $keys = array_keys( $args );
560 // If IGNORE is set, we use savepoints to emulate mysql's behavior
561 $savepoint = $olde = null;
562 $numrowsinserted = 0;
563 if ( in_array( 'IGNORE', $options ) ) {
564 $savepoint = new SavepointPostgres( $this, 'mw', $this->queryLogger );
565 $olde = error_reporting( 0 );
566 // For future use, we may want to track the number of actual inserts
567 // Right now, insert (all writes) simply return true/false
570 $sql = "INSERT INTO $table (" . implode( ',', $keys ) . ') VALUES ';
573 if ( $this->numericVersion >= 8.2 && !$savepoint ) {
575 foreach ( $args as $row ) {
581 $sql .= '(' . $this->makeList( $row ) . ')';
583 $res = (bool)$this->query( $sql, $fname, $savepoint );
587 foreach ( $args as $row ) {
589 $tempsql .= '(' . $this->makeList( $row ) . ')';
592 $savepoint->savepoint();
595 $tempres = (bool)$this->query( $tempsql, $fname, $savepoint );
598 $bar = pg_result_error( $this->mLastResult );
599 if ( $bar != false ) {
600 $savepoint->rollback();
602 $savepoint->release();
607 // If any of them fail, we fail overall for this function call
608 // Note that this will be ignored if IGNORE is set
615 // Not multi, just a lone insert
617 $savepoint->savepoint();
620 $sql .= '(' . $this->makeList( $args ) . ')';
621 $res = (bool)$this->query( $sql, $fname, $savepoint );
623 $bar = pg_result_error( $this->mLastResult );
624 if ( $bar != false ) {
625 $savepoint->rollback();
627 $savepoint->release();
633 error_reporting( $olde );
634 $savepoint->commit();
636 // Set the affected row count for the whole operation
637 $this->mAffectedRows = $numrowsinserted;
639 // IGNORE always returns true
647 * INSERT SELECT wrapper
648 * $varMap must be an associative array of the form [ 'dest1' => 'source1', ... ]
649 * Source items may be literals rather then field names, but strings should
650 * be quoted with Database::addQuotes()
651 * $conds may be "*" to copy the whole table
652 * srcTable may be an array of tables.
653 * @todo FIXME: Implement this a little better (seperate select/insert)?
655 * @param string $destTable
656 * @param array|string $srcTable
657 * @param array $varMap
658 * @param array $conds
659 * @param string $fname
660 * @param array $insertOptions
661 * @param array $selectOptions
664 public function nativeInsertSelect(
665 $destTable, $srcTable, $varMap, $conds, $fname = __METHOD__,
666 $insertOptions = [], $selectOptions = []
668 $destTable = $this->tableName( $destTable );
670 if ( !is_array( $insertOptions ) ) {
671 $insertOptions = [ $insertOptions ];
675 * If IGNORE is set, we use savepoints to emulate mysql's behavior
676 * Ignore LOW PRIORITY option, since it is MySQL-specific
678 $savepoint = $olde = null;
679 $numrowsinserted = 0;
680 if ( in_array( 'IGNORE', $insertOptions ) ) {
681 $savepoint = new SavepointPostgres( $this, 'mw', $this->queryLogger );
682 $olde = error_reporting( 0 );
683 $savepoint->savepoint();
686 if ( !is_array( $selectOptions ) ) {
687 $selectOptions = [ $selectOptions ];
689 list( $startOpts, $useIndex, $tailOpts, $ignoreIndex ) =
690 $this->makeSelectOptions( $selectOptions );
691 if ( is_array( $srcTable ) ) {
692 $srcTable = implode( ',', array_map( [ &$this, 'tableName' ], $srcTable ) );
694 $srcTable = $this->tableName( $srcTable );
697 $sql = "INSERT INTO $destTable (" . implode( ',', array_keys( $varMap ) ) . ')' .
698 " SELECT $startOpts " . implode( ',', $varMap ) .
699 " FROM $srcTable $useIndex $ignoreIndex ";
701 if ( $conds != '*' ) {
702 $sql .= ' WHERE ' . $this->makeList( $conds, LIST_AND );
705 $sql .= " $tailOpts";
707 $res = (bool)$this->query( $sql, $fname, $savepoint );
709 $bar = pg_result_error( $this->mLastResult );
710 if ( $bar != false ) {
711 $savepoint->rollback();
713 $savepoint->release();
716 error_reporting( $olde );
717 $savepoint->commit();
719 // Set the affected row count for the whole operation
720 $this->mAffectedRows = $numrowsinserted;
722 // IGNORE always returns true
729 public function tableName( $name, $format = 'quoted' ) {
730 // Replace reserved words with better ones
731 $name = $this->remappedTableName( $name );
733 return parent::tableName( $name, $format );
737 * @param string $name
738 * @return string Value of $name or remapped name if $name is a reserved keyword
739 * @TODO: dependency inject these...
741 public function remappedTableName( $name ) {
742 if ( $name === 'user' ) {
744 } elseif ( $name === 'text' ) {
745 return 'pagecontent';
752 * @param string $name
753 * @param string $format
754 * @return string Qualified and encoded (if requested) table name
756 public function realTableName( $name, $format = 'quoted' ) {
757 return parent::tableName( $name, $format );
760 public function nextSequenceValue( $seqName ) {
761 $safeseq = str_replace( "'", "''", $seqName );
762 $res = $this->query( "SELECT nextval('$safeseq')" );
763 $row = $this->fetchRow( $res );
764 $this->mInsertId = $row[0];
766 return $this->mInsertId;
770 * Return the current value of a sequence. Assumes it has been nextval'ed in this session.
772 * @param string $seqName
775 public function currentSequenceValue( $seqName ) {
776 $safeseq = str_replace( "'", "''", $seqName );
777 $res = $this->query( "SELECT currval('$safeseq')" );
778 $row = $this->fetchRow( $res );
784 public function textFieldSize( $table, $field ) {
785 $table = $this->tableName( $table );
786 $sql = "SELECT t.typname as ftype,a.atttypmod as size
787 FROM pg_class c, pg_attribute a, pg_type t
788 WHERE relname='$table' AND a.attrelid=c.oid AND
789 a.atttypid=t.oid and a.attname='$field'";
790 $res = $this->query( $sql );
791 $row = $this->fetchObject( $res );
792 if ( $row->ftype == 'varchar' ) {
793 $size = $row->size - 4;
801 public function limitResult( $sql, $limit, $offset = false ) {
802 return "$sql LIMIT $limit " . ( is_numeric( $offset ) ? " OFFSET {$offset} " : '' );
805 public function wasDeadlock() {
806 return $this->lastErrno() == '40P01';
809 public function duplicateTableStructure(
810 $oldName, $newName, $temporary = false, $fname = __METHOD__
812 $newName = $this->addIdentifierQuotes( $newName );
813 $oldName = $this->addIdentifierQuotes( $oldName );
815 return $this->query( 'CREATE ' . ( $temporary ? 'TEMPORARY ' : '' ) . " TABLE $newName " .
816 "(LIKE $oldName INCLUDING DEFAULTS)", $fname );
819 public function listTables( $prefix = null, $fname = __METHOD__ ) {
820 $eschema = $this->addQuotes( $this->getCoreSchema() );
821 $result = $this->query(
822 "SELECT tablename FROM pg_tables WHERE schemaname = $eschema", $fname );
825 foreach ( $result as $table ) {
826 $vars = get_object_vars( $table );
827 $table = array_pop( $vars );
828 if ( !$prefix || strpos( $table, $prefix ) === 0 ) {
829 $endArray[] = $table;
836 public function timestamp( $ts = 0 ) {
837 $ct = new ConvertibleTimestamp( $ts );
839 return $ct->getTimestamp( TS_POSTGRES );
843 * Posted by cc[plus]php[at]c2se[dot]com on 25-Mar-2009 09:12
844 * to http://www.php.net/manual/en/ref.pgsql.php
846 * Parsing a postgres array can be a tricky problem, he's my
847 * take on this, it handles multi-dimensional arrays plus
848 * escaping using a nasty regexp to determine the limits of each
851 * This should really be handled by PHP PostgreSQL module
854 * @param string $text Postgreql array returned in a text form like {a,b}
855 * @param string $output
856 * @param int|bool $limit
860 private function pg_array_parse( $text, &$output, $limit = false, $offset = 1 ) {
861 if ( false === $limit ) {
862 $limit = strlen( $text ) - 1;
865 if ( '{}' == $text ) {
869 if ( '{' != $text[$offset] ) {
870 preg_match( "/(\\{?\"([^\"\\\\]|\\\\.)*\"|[^,{}]+)+([,}]+)/",
871 $text, $match, 0, $offset );
872 $offset += strlen( $match[0] );
873 $output[] = ( '"' != $match[1][0]
875 : stripcslashes( substr( $match[1], 1, -1 ) ) );
876 if ( '},' == $match[3] ) {
880 $offset = $this->pg_array_parse( $text, $output, $limit, $offset + 1 );
882 } while ( $limit > $offset );
887 public function aggregateValue( $valuedata, $valuename = 'value' ) {
891 public function getSoftwareLink() {
892 return '[{{int:version-db-postgres-url}} PostgreSQL]';
896 * Return current schema (executes SELECT current_schema())
900 * @return string Default schema for the current session
902 public function getCurrentSchema() {
903 $res = $this->query( "SELECT current_schema()", __METHOD__ );
904 $row = $this->fetchRow( $res );
910 * Return list of schemas which are accessible without schema name
911 * This is list does not contain magic keywords like "$user"
914 * @see getSearchPath()
915 * @see setSearchPath()
917 * @return array List of actual schemas for the current sesson
919 public function getSchemas() {
920 $res = $this->query( "SELECT current_schemas(false)", __METHOD__ );
921 $row = $this->fetchRow( $res );
924 /* PHP pgsql support does not support array type, "{a,b}" string is returned */
926 return $this->pg_array_parse( $row[0], $schemas );
930 * Return search patch for schemas
931 * This is different from getSchemas() since it contain magic keywords
936 * @return array How to search for table names schemas for the current user
938 public function getSearchPath() {
939 $res = $this->query( "SHOW search_path", __METHOD__ );
940 $row = $this->fetchRow( $res );
942 /* PostgreSQL returns SHOW values as strings */
944 return explode( ",", $row[0] );
948 * Update search_path, values should already be sanitized
949 * Values may contain magic keywords like "$user"
952 * @param array $search_path List of schemas to be searched by default
954 private function setSearchPath( $search_path ) {
955 $this->query( "SET search_path = " . implode( ", ", $search_path ) );
959 * Determine default schema for the current application
960 * Adjust this session schema search path if desired schema exists
961 * and is not alread there.
963 * We need to have name of the core schema stored to be able
964 * to query database metadata.
966 * This will be also called by the installer after the schema is created
970 * @param string $desiredSchema
972 public function determineCoreSchema( $desiredSchema ) {
973 $this->begin( __METHOD__, self::TRANSACTION_INTERNAL );
974 if ( $this->schemaExists( $desiredSchema ) ) {
975 if ( in_array( $desiredSchema, $this->getSchemas() ) ) {
976 $this->mCoreSchema = $desiredSchema;
977 $this->queryLogger->debug(
978 "Schema \"" . $desiredSchema . "\" already in the search path\n" );
981 * Prepend our schema (e.g. 'mediawiki') in front
985 $search_path = $this->getSearchPath();
986 array_unshift( $search_path,
987 $this->addIdentifierQuotes( $desiredSchema ) );
988 $this->setSearchPath( $search_path );
989 $this->mCoreSchema = $desiredSchema;
990 $this->queryLogger->debug(
991 "Schema \"" . $desiredSchema . "\" added to the search path\n" );
994 $this->mCoreSchema = $this->getCurrentSchema();
995 $this->queryLogger->debug(
996 "Schema \"" . $desiredSchema . "\" not found, using current \"" .
997 $this->mCoreSchema . "\"\n" );
999 /* Commit SET otherwise it will be rollbacked on error or IGNORE SELECT */
1000 $this->commit( __METHOD__, self::FLUSHING_INTERNAL );
1004 * Return schema name for core application tables
1007 * @return string Core schema name
1009 public function getCoreSchema() {
1010 return $this->mCoreSchema;
1013 public function getServerVersion() {
1014 if ( !isset( $this->numericVersion ) ) {
1015 $conn = $this->getBindingHandle();
1016 $versionInfo = pg_version( $conn );
1017 if ( version_compare( $versionInfo['client'], '7.4.0', 'lt' ) ) {
1018 // Old client, abort install
1019 $this->numericVersion = '7.3 or earlier';
1020 } elseif ( isset( $versionInfo['server'] ) ) {
1022 $this->numericVersion = $versionInfo['server'];
1024 // Bug 16937: broken pgsql extension from PHP<5.3
1025 $this->numericVersion = pg_parameter_status( $conn, 'server_version' );
1029 return $this->numericVersion;
1033 * Query whether a given relation exists (in the given schema, or the
1034 * default mw one if not given)
1035 * @param string $table
1036 * @param array|string $types
1037 * @param bool|string $schema
1040 private function relationExists( $table, $types, $schema = false ) {
1041 if ( !is_array( $types ) ) {
1042 $types = [ $types ];
1044 if ( $schema === false ) {
1045 $schema = $this->getCoreSchema();
1047 $etable = $this->addQuotes( $table );
1048 $eschema = $this->addQuotes( $schema );
1049 $sql = "SELECT 1 FROM pg_catalog.pg_class c, pg_catalog.pg_namespace n "
1050 . "WHERE c.relnamespace = n.oid AND c.relname = $etable AND n.nspname = $eschema "
1051 . "AND c.relkind IN ('" . implode( "','", $types ) . "')";
1052 $res = $this->query( $sql );
1053 $count = $res ? $res->numRows() : 0;
1055 return (bool)$count;
1059 * For backward compatibility, this function checks both tables and views.
1060 * @param string $table
1061 * @param string $fname
1062 * @param bool|string $schema
1065 public function tableExists( $table, $fname = __METHOD__, $schema = false ) {
1066 return $this->relationExists( $table, [ 'r', 'v' ], $schema );
1069 public function sequenceExists( $sequence, $schema = false ) {
1070 return $this->relationExists( $sequence, 'S', $schema );
1073 public function triggerExists( $table, $trigger ) {
1075 SELECT 1 FROM pg_class, pg_namespace, pg_trigger
1076 WHERE relnamespace=pg_namespace.oid AND relkind='r'
1077 AND tgrelid=pg_class.oid
1078 AND nspname=%s AND relname=%s AND tgname=%s
1080 $res = $this->query(
1083 $this->addQuotes( $this->getCoreSchema() ),
1084 $this->addQuotes( $table ),
1085 $this->addQuotes( $trigger )
1091 $rows = $res->numRows();
1096 public function ruleExists( $table, $rule ) {
1097 $exists = $this->selectField( 'pg_rules', 'rulename',
1099 'rulename' => $rule,
1100 'tablename' => $table,
1101 'schemaname' => $this->getCoreSchema()
1105 return $exists === $rule;
1108 public function constraintExists( $table, $constraint ) {
1109 $sql = sprintf( "SELECT 1 FROM information_schema.table_constraints " .
1110 "WHERE constraint_schema = %s AND table_name = %s AND constraint_name = %s",
1111 $this->addQuotes( $this->getCoreSchema() ),
1112 $this->addQuotes( $table ),
1113 $this->addQuotes( $constraint )
1115 $res = $this->query( $sql );
1119 $rows = $res->numRows();
1125 * Query whether a given schema exists. Returns true if it does, false if it doesn't.
1126 * @param string $schema
1129 public function schemaExists( $schema ) {
1130 if ( !strlen( $schema ) ) {
1131 return false; // short-circuit
1134 $exists = $this->selectField(
1135 '"pg_catalog"."pg_namespace"', 1, [ 'nspname' => $schema ], __METHOD__ );
1137 return (bool)$exists;
1141 * Returns true if a given role (i.e. user) exists, false otherwise.
1142 * @param string $roleName
1145 public function roleExists( $roleName ) {
1146 $exists = $this->selectField( '"pg_catalog"."pg_roles"', 1,
1147 [ 'rolname' => $roleName ], __METHOD__ );
1149 return (bool)$exists;
1153 * @var string $table
1154 * @var string $field
1155 * @return PostgresField|null
1157 public function fieldInfo( $table, $field ) {
1158 return PostgresField::fromText( $this, $table, $field );
1162 * pg_field_type() wrapper
1163 * @param ResultWrapper|resource $res ResultWrapper or PostgreSQL query result resource
1164 * @param int $index Field number, starting from 0
1167 public function fieldType( $res, $index ) {
1168 if ( $res instanceof ResultWrapper ) {
1169 $res = $res->result;
1172 return pg_field_type( $res, $index );
1175 public function encodeBlob( $b ) {
1176 return new PostgresBlob( pg_escape_bytea( $b ) );
1179 public function decodeBlob( $b ) {
1180 if ( $b instanceof PostgresBlob ) {
1182 } elseif ( $b instanceof Blob ) {
1186 return pg_unescape_bytea( $b );
1189 public function strencode( $s ) {
1190 // Should not be called by us
1191 return pg_escape_string( $this->getBindingHandle(), $s );
1194 public function addQuotes( $s ) {
1195 $conn = $this->getBindingHandle();
1197 if ( is_null( $s ) ) {
1199 } elseif ( is_bool( $s ) ) {
1200 return intval( $s );
1201 } elseif ( $s instanceof Blob ) {
1202 if ( $s instanceof PostgresBlob ) {
1205 $s = pg_escape_bytea( $conn, $s->fetch() );
1210 return "'" . pg_escape_string( $conn, $s ) . "'";
1214 * Postgres specific version of replaceVars.
1215 * Calls the parent version in Database.php
1217 * @param string $ins SQL string, read from a stream (usually tables.sql)
1218 * @return string SQL string
1220 protected function replaceVars( $ins ) {
1221 $ins = parent::replaceVars( $ins );
1223 if ( $this->numericVersion >= 8.3 ) {
1224 // Thanks for not providing backwards-compatibility, 8.3
1225 $ins = preg_replace( "/to_tsvector\s*\(\s*'default'\s*,/", 'to_tsvector(', $ins );
1228 if ( $this->numericVersion <= 8.1 ) { // Our minimum version
1229 $ins = str_replace( 'USING gin', 'USING gist', $ins );
1235 public function makeSelectOptions( $options ) {
1236 $preLimitTail = $postLimitTail = '';
1237 $startOpts = $useIndex = $ignoreIndex = '';
1240 foreach ( $options as $key => $option ) {
1241 if ( is_numeric( $key ) ) {
1242 $noKeyOptions[$option] = true;
1246 $preLimitTail .= $this->makeGroupByWithHaving( $options );
1248 $preLimitTail .= $this->makeOrderBy( $options );
1250 // if ( isset( $options['LIMIT'] ) ) {
1251 // $tailOpts .= $this->limitResult( '', $options['LIMIT'],
1252 // isset( $options['OFFSET'] ) ? $options['OFFSET']
1256 if ( isset( $options['FOR UPDATE'] ) ) {
1257 $postLimitTail .= ' FOR UPDATE OF ' .
1258 implode( ', ', array_map( [ &$this, 'tableName' ], $options['FOR UPDATE'] ) );
1259 } elseif ( isset( $noKeyOptions['FOR UPDATE'] ) ) {
1260 $postLimitTail .= ' FOR UPDATE';
1263 if ( isset( $noKeyOptions['DISTINCT'] ) || isset( $noKeyOptions['DISTINCTROW'] ) ) {
1264 $startOpts .= 'DISTINCT';
1267 return [ $startOpts, $useIndex, $preLimitTail, $postLimitTail, $ignoreIndex ];
1270 public function getDBname() {
1271 return $this->mDBname;
1274 public function getServer() {
1275 return $this->mServer;
1278 public function buildConcat( $stringList ) {
1279 return implode( ' || ', $stringList );
1282 public function buildGroupConcatField(
1283 $delimiter, $table, $field, $conds = '', $options = [], $join_conds = []
1285 $fld = "array_to_string(array_agg($field)," . $this->addQuotes( $delimiter ) . ')';
1287 return '(' . $this->selectSQLText( $table, $fld, $conds, null, [], $join_conds ) . ')';
1290 public function buildStringCast( $field ) {
1291 return $field . '::text';
1294 public function streamStatementEnd( &$sql, &$newLine ) {
1295 # Allow dollar quoting for function declarations
1296 if ( substr( $newLine, 0, 4 ) == '$mw$' ) {
1297 if ( $this->delimiter ) {
1298 $this->delimiter = false;
1300 $this->delimiter = ';';
1304 return parent::streamStatementEnd( $sql, $newLine );
1307 public function lockIsFree( $lockName, $method ) {
1308 // http://www.postgresql.org/docs/8.2/static/functions-admin.html#FUNCTIONS-ADVISORY-LOCKS
1309 $key = $this->addQuotes( $this->bigintFromLockName( $lockName ) );
1310 $result = $this->query( "SELECT (CASE(pg_try_advisory_lock($key))
1311 WHEN 'f' THEN 'f' ELSE pg_advisory_unlock($key) END) AS lockstatus", $method );
1312 $row = $this->fetchObject( $result );
1314 return ( $row->lockstatus === 't' );
1317 public function lock( $lockName, $method, $timeout = 5 ) {
1318 // http://www.postgresql.org/docs/8.2/static/functions-admin.html#FUNCTIONS-ADVISORY-LOCKS
1319 $key = $this->addQuotes( $this->bigintFromLockName( $lockName ) );
1320 $loop = new WaitConditionLoop(
1321 function () use ( $lockName, $key, $timeout, $method ) {
1322 $res = $this->query( "SELECT pg_try_advisory_lock($key) AS lockstatus", $method );
1323 $row = $this->fetchObject( $res );
1324 if ( $row->lockstatus === 't' ) {
1325 parent::lock( $lockName, $method, $timeout ); // record
1329 return WaitConditionLoop::CONDITION_CONTINUE;
1334 return ( $loop->invoke() === $loop::CONDITION_REACHED );
1337 public function unlock( $lockName, $method ) {
1338 // http://www.postgresql.org/docs/8.2/static/functions-admin.html#FUNCTIONS-ADVISORY-LOCKS
1339 $key = $this->addQuotes( $this->bigintFromLockName( $lockName ) );
1340 $result = $this->query( "SELECT pg_advisory_unlock($key) as lockstatus", $method );
1341 $row = $this->fetchObject( $result );
1343 if ( $row->lockstatus === 't' ) {
1344 parent::unlock( $lockName, $method ); // record
1348 $this->queryLogger->debug( __METHOD__ . " failed to release lock\n" );
1354 * @param string $lockName
1355 * @return string Integer
1357 private function bigintFromLockName( $lockName ) {
1358 return Wikimedia\base_convert( substr( sha1( $lockName ), 0, 15 ), 16, 10 );