0c9b18e21f8ddb8a8f71f221b5eeb2242412986c
[moodle.git] / lib / dml / pgsql_native_moodle_database.php
1 <?php
2 // This file is part of Moodle - http://moodle.org/
3 //
4 // Moodle is free software: you can redistribute it and/or modify
5 // it under the terms of the GNU General Public License as published by
6 // the Free Software Foundation, either version 3 of the License, or
7 // (at your option) any later version.
8 //
9 // Moodle is distributed in the hope that it will be useful,
10 // but WITHOUT ANY WARRANTY; without even the implied warranty of
11 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12 // GNU General Public License for more details.
13 //
14 // You should have received a copy of the GNU General Public License
15 // along with Moodle.  If not, see <http://www.gnu.org/licenses/>.
17 /**
18  * Native pgsql class representing moodle database interface.
19  *
20  * @package    core_dml
21  * @copyright  2008 Petr Skoda (http://skodak.org)
22  * @license    http://www.gnu.org/copyleft/gpl.html GNU GPL v3 or later
23  */
25 defined('MOODLE_INTERNAL') || die();
27 require_once(__DIR__.'/moodle_database.php');
28 require_once(__DIR__.'/pgsql_native_moodle_recordset.php');
29 require_once(__DIR__.'/pgsql_native_moodle_temptables.php');
31 /**
32  * Native pgsql class representing moodle database interface.
33  *
34  * @package    core_dml
35  * @copyright  2008 Petr Skoda (http://skodak.org)
36  * @license    http://www.gnu.org/copyleft/gpl.html GNU GPL v3 or later
37  */
38 class pgsql_native_moodle_database extends moodle_database {
40     protected $pgsql     = null;
41     protected $bytea_oid = null;
43     protected $last_error_reporting; // To handle pgsql driver default verbosity
45     /**
46      * Detects if all needed PHP stuff installed.
47      * Note: can be used before connect()
48      * @return mixed true if ok, string if something
49      */
50     public function driver_installed() {
51         if (!extension_loaded('pgsql')) {
52             return get_string('pgsqlextensionisnotpresentinphp', 'install');
53         }
54         return true;
55     }
57     /**
58      * Returns database family type - describes SQL dialect
59      * Note: can be used before connect()
60      * @return string db family name (mysql, postgres, mssql, oracle, etc.)
61      */
62     public function get_dbfamily() {
63         return 'postgres';
64     }
66     /**
67      * Returns more specific database driver type
68      * Note: can be used before connect()
69      * @return string db type mysqli, pgsql, oci, mssql, sqlsrv
70      */
71     protected function get_dbtype() {
72         return 'pgsql';
73     }
75     /**
76      * Returns general database library name
77      * Note: can be used before connect()
78      * @return string db type pdo, native
79      */
80     protected function get_dblibrary() {
81         return 'native';
82     }
84     /**
85      * Returns localised database type name
86      * Note: can be used before connect()
87      * @return string
88      */
89     public function get_name() {
90         return get_string('nativepgsql', 'install');
91     }
93     /**
94      * Returns localised database configuration help.
95      * Note: can be used before connect()
96      * @return string
97      */
98     public function get_configuration_help() {
99         return get_string('nativepgsqlhelp', 'install');
100     }
102     /**
103      * Returns localised database description
104      * Note: can be used before connect()
105      * @return string
106      */
107     public function get_configuration_hints() {
108         return get_string('databasesettingssub_postgres7', 'install');
109     }
111     /**
112      * Connect to db
113      * Must be called before other methods.
114      * @param string $dbhost The database host.
115      * @param string $dbuser The database username.
116      * @param string $dbpass The database username's password.
117      * @param string $dbname The name of the database being connected to.
118      * @param mixed $prefix string means moodle db prefix, false used for external databases where prefix not used
119      * @param array $dboptions driver specific options
120      * @return bool true
121      * @throws dml_connection_exception if error
122      */
123     public function connect($dbhost, $dbuser, $dbpass, $dbname, $prefix, array $dboptions=null) {
124         if ($prefix == '' and !$this->external) {
125             //Enforce prefixes for everybody but mysql
126             throw new dml_exception('prefixcannotbeempty', $this->get_dbfamily());
127         }
129         $driverstatus = $this->driver_installed();
131         if ($driverstatus !== true) {
132             throw new dml_exception('dbdriverproblem', $driverstatus);
133         }
135         $this->store_settings($dbhost, $dbuser, $dbpass, $dbname, $prefix, $dboptions);
137         $pass = addcslashes($this->dbpass, "'\\");
139         // Unix socket connections should have lower overhead
140         if (!empty($this->dboptions['dbsocket']) and ($this->dbhost === 'localhost' or $this->dbhost === '127.0.0.1')) {
141             $connection = "user='$this->dbuser' password='$pass' dbname='$this->dbname'";
142             if (strpos($this->dboptions['dbsocket'], '/') !== false) {
143                 $connection = $connection." host='".$this->dboptions['dbsocket']."'";
144             }
145         } else {
146             $this->dboptions['dbsocket'] = '';
147             if (empty($this->dbname)) {
148                 // probably old style socket connection - do not add port
149                 $port = "";
150             } else if (empty($this->dboptions['dbport'])) {
151                 $port = "port ='5432'";
152             } else {
153                 $port = "port ='".$this->dboptions['dbport']."'";
154             }
155             $connection = "host='$this->dbhost' $port user='$this->dbuser' password='$pass' dbname='$this->dbname'";
156         }
158         ob_start();
159         if (empty($this->dboptions['dbpersist'])) {
160             $this->pgsql = pg_connect($connection, PGSQL_CONNECT_FORCE_NEW);
161         } else {
162             $this->pgsql = pg_pconnect($connection, PGSQL_CONNECT_FORCE_NEW);
163         }
164         $dberr = ob_get_contents();
165         ob_end_clean();
167         $status = pg_connection_status($this->pgsql);
169         if ($status === false or $status === PGSQL_CONNECTION_BAD) {
170             $this->pgsql = null;
171             throw new dml_connection_exception($dberr);
172         }
174         $this->query_start("--pg_set_client_encoding()", null, SQL_QUERY_AUX);
175         pg_set_client_encoding($this->pgsql, 'utf8');
176         $this->query_end(true);
178         $sql = '';
179         // Only for 9.0 and upwards, set bytea encoding to old format.
180         if ($this->is_min_version('9.0')) {
181             $sql = "SET bytea_output = 'escape'; ";
182         }
184         // Find out the bytea oid.
185         $sql .= "SELECT oid FROM pg_type WHERE typname = 'bytea'";
186         $this->query_start($sql, null, SQL_QUERY_AUX);
187         $result = pg_query($this->pgsql, $sql);
188         $this->query_end($result);
190         $this->bytea_oid = pg_fetch_result($result, 0, 0);
191         pg_free_result($result);
192         if ($this->bytea_oid === false) {
193             $this->pgsql = null;
194             throw new dml_connection_exception('Can not read bytea type.');
195         }
197         // Connection stabilised and configured, going to instantiate the temptables controller
198         $this->temptables = new pgsql_native_moodle_temptables($this);
200         return true;
201     }
203     /**
204      * Close database connection and release all resources
205      * and memory (especially circular memory references).
206      * Do NOT use connect() again, create a new instance if needed.
207      */
208     public function dispose() {
209         parent::dispose(); // Call parent dispose to write/close session and other common stuff before closing connection
210         if ($this->pgsql) {
211             pg_close($this->pgsql);
212             $this->pgsql = null;
213         }
214     }
217     /**
218      * Called before each db query.
219      * @param string $sql
220      * @param array array of parameters
221      * @param int $type type of query
222      * @param mixed $extrainfo driver specific extra information
223      * @return void
224      */
225     protected function query_start($sql, array $params=null, $type, $extrainfo=null) {
226         parent::query_start($sql, $params, $type, $extrainfo);
227         // pgsql driver tents to send debug to output, we do not need that ;-)
228         $this->last_error_reporting = error_reporting(0);
229     }
231     /**
232      * Called immediately after each db query.
233      * @param mixed db specific result
234      * @return void
235      */
236     protected function query_end($result) {
237         // reset original debug level
238         error_reporting($this->last_error_reporting);
239         parent::query_end($result);
240     }
242     /**
243      * Returns database server info array
244      * @return array Array containing 'description' and 'version' info
245      */
246     public function get_server_info() {
247         static $info;
248         if (!$info) {
249             $this->query_start("--pg_version()", null, SQL_QUERY_AUX);
250             $info = pg_version($this->pgsql);
251             $this->query_end(true);
252         }
253         return array('description'=>$info['server'], 'version'=>$info['server']);
254     }
256     /**
257      * Returns if the RDBMS server fulfills the required version
258      *
259      * @param string $version version to check against
260      * @return bool returns if the version is fulfilled (true) or no (false)
261      */
262     private function is_min_version($version) {
263         $server = $this->get_server_info();
264         $server = $server['version'];
265         return version_compare($server, $version, '>=');
266     }
268     /**
269      * Returns supported query parameter types
270      * @return int bitmask of accepted SQL_PARAMS_*
271      */
272     protected function allowed_param_types() {
273         return SQL_PARAMS_DOLLAR;
274     }
276     /**
277      * Returns last error reported by database engine.
278      * @return string error message
279      */
280     public function get_last_error() {
281         return pg_last_error($this->pgsql);
282     }
284     /**
285      * Return tables in database WITHOUT current prefix.
286      * @param bool $usecache if true, returns list of cached tables.
287      * @return array of table names in lowercase and without prefix
288      */
289     public function get_tables($usecache=true) {
290         if ($usecache and $this->tables !== null) {
291             return $this->tables;
292         }
293         $this->tables = array();
294         $prefix = str_replace('_', '|_', $this->prefix);
295         if ($this->is_min_version('9.1')) {
296             // Use ANSI standard information_schema in recent versions where it is fast enough.
297             $sql = "SELECT table_name
298                       FROM information_schema.tables
299                      WHERE table_name LIKE '$prefix%' ESCAPE '|'
300                        AND table_type IN ('BASE TABLE', 'LOCAL TEMPORARY')";
301         } else {
302             // information_schema is horribly slow in <= 9.0, so use pg internals.
303             // Note the pg_is_other_temp_schema. We only want temp objects from our own session.
304             $sql = "SELECT c.relname
305                       FROM pg_class c
306                      WHERE c.relname LIKE '$prefix%' ESCAPE '|'
307                        AND c.relkind = 'r'
308                        AND NOT pg_is_other_temp_schema(c.relnamespace)";
309         }
310         $this->query_start($sql, null, SQL_QUERY_AUX);
311         $result = pg_query($this->pgsql, $sql);
312         $this->query_end($result);
314         if ($result) {
315             while ($row = pg_fetch_row($result)) {
316                 $tablename = reset($row);
317                 if ($this->prefix !== '') {
318                     if (strpos($tablename, $this->prefix) !== 0) {
319                         continue;
320                     }
321                     $tablename = substr($tablename, strlen($this->prefix));
322                 }
323                 $this->tables[$tablename] = $tablename;
324             }
325             pg_free_result($result);
326         }
327         return $this->tables;
328     }
330     /**
331      * Return table indexes - everything lowercased.
332      * @param string $table The table we want to get indexes from.
333      * @return array of arrays
334      */
335     public function get_indexes($table) {
336         $indexes = array();
337         $tablename = $this->prefix.$table;
339         $sql = "SELECT *
340                   FROM pg_catalog.pg_indexes
341                  WHERE tablename = '$tablename'";
343         $this->query_start($sql, null, SQL_QUERY_AUX);
344         $result = pg_query($this->pgsql, $sql);
345         $this->query_end($result);
347         if ($result) {
348             while ($row = pg_fetch_assoc($result)) {
349                 if (!preg_match('/CREATE (|UNIQUE )INDEX ([^\s]+) ON '.$tablename.' USING ([^\s]+) \(([^\)]+)\)/i', $row['indexdef'], $matches)) {
350                     continue;
351                 }
352                 if ($matches[4] === 'id') {
353                     continue;
354                 }
355                 $columns = explode(',', $matches[4]);
356                 $columns = array_map(array($this, 'trim_quotes'), $columns);
357                 $indexes[$row['indexname']] = array('unique'=>!empty($matches[1]),
358                                               'columns'=>$columns);
359             }
360             pg_free_result($result);
361         }
362         return $indexes;
363     }
365     /**
366      * Returns detailed information about columns in table. This information is cached internally.
367      * @param string $table name
368      * @param bool $usecache
369      * @return array array of database_column_info objects indexed with column names
370      */
371     public function get_columns($table, $usecache=true) {
372         if ($usecache and isset($this->columns[$table])) {
373             return $this->columns[$table];
374         }
376         $this->columns[$table] = array();
378         $tablename = $this->prefix.$table;
380         $sql = "SELECT a.attnum, a.attname AS field, t.typname AS type, a.attlen, a.atttypmod, a.attnotnull, a.atthasdef, d.adsrc
381                   FROM pg_catalog.pg_class c
382                   JOIN pg_catalog.pg_attribute a ON a.attrelid = c.oid
383                   JOIN pg_catalog.pg_type t ON t.oid = a.atttypid
384              LEFT JOIN pg_catalog.pg_attrdef d ON (d.adrelid = c.oid AND d.adnum = a.attnum)
385                  WHERE relkind = 'r' AND c.relname = '$tablename' AND c.reltype > 0 AND a.attnum > 0
386               ORDER BY a.attnum";
388         $this->query_start($sql, null, SQL_QUERY_AUX);
389         $result = pg_query($this->pgsql, $sql);
390         $this->query_end($result);
392         if (!$result) {
393             return array();
394         }
395         while ($rawcolumn = pg_fetch_object($result)) {
397             $info = new stdClass();
398             $info->name = $rawcolumn->field;
399             $matches = null;
401             if ($rawcolumn->type === 'varchar') {
402                 $info->type          = 'varchar';
403                 $info->meta_type     = 'C';
404                 $info->max_length    = $rawcolumn->atttypmod - 4;
405                 $info->scale         = null;
406                 $info->not_null      = ($rawcolumn->attnotnull === 't');
407                 $info->has_default   = ($rawcolumn->atthasdef === 't');
408                 if ($info->has_default) {
409                     $parts = explode('::', $rawcolumn->adsrc);
410                     if (count($parts) > 1) {
411                         $info->default_value = reset($parts);
412                         $info->default_value = trim($info->default_value, "'");
413                     } else {
414                         $info->default_value = $rawcolumn->adsrc;
415                     }
416                 } else {
417                     $info->default_value = null;
418                 }
419                 $info->primary_key   = false;
420                 $info->binary        = false;
421                 $info->unsigned      = null;
422                 $info->auto_increment= false;
423                 $info->unique        = null;
425             } else if (preg_match('/int(\d)/i', $rawcolumn->type, $matches)) {
426                 $info->type = 'int';
427                 if (strpos($rawcolumn->adsrc, 'nextval') === 0) {
428                     $info->primary_key   = true;
429                     $info->meta_type     = 'R';
430                     $info->unique        = true;
431                     $info->auto_increment= true;
432                     $info->has_default   = false;
433                 } else {
434                     $info->primary_key   = false;
435                     $info->meta_type     = 'I';
436                     $info->unique        = null;
437                     $info->auto_increment= false;
438                     $info->has_default   = ($rawcolumn->atthasdef === 't');
439                 }
440                 $info->max_length    = $matches[1];
441                 $info->scale         = null;
442                 $info->not_null      = ($rawcolumn->attnotnull === 't');
443                 if ($info->has_default) {
444                     $info->default_value = trim($rawcolumn->adsrc, '()');
445                 } else {
446                     $info->default_value = null;
447                 }
448                 $info->binary        = false;
449                 $info->unsigned      = false;
451             } else if ($rawcolumn->type === 'numeric') {
452                 $info->type = $rawcolumn->type;
453                 $info->meta_type     = 'N';
454                 $info->primary_key   = false;
455                 $info->binary        = false;
456                 $info->unsigned      = null;
457                 $info->auto_increment= false;
458                 $info->unique        = null;
459                 $info->not_null      = ($rawcolumn->attnotnull === 't');
460                 $info->has_default   = ($rawcolumn->atthasdef === 't');
461                 if ($info->has_default) {
462                     $info->default_value = trim($rawcolumn->adsrc, '()');
463                 } else {
464                     $info->default_value = null;
465                 }
466                 $info->max_length    = $rawcolumn->atttypmod >> 16;
467                 $info->scale         = ($rawcolumn->atttypmod & 0xFFFF) - 4;
469             } else if (preg_match('/float(\d)/i', $rawcolumn->type, $matches)) {
470                 $info->type = 'float';
471                 $info->meta_type     = 'N';
472                 $info->primary_key   = false;
473                 $info->binary        = false;
474                 $info->unsigned      = null;
475                 $info->auto_increment= false;
476                 $info->unique        = null;
477                 $info->not_null      = ($rawcolumn->attnotnull === 't');
478                 $info->has_default   = ($rawcolumn->atthasdef === 't');
479                 if ($info->has_default) {
480                     $info->default_value = trim($rawcolumn->adsrc, '()');
481                 } else {
482                     $info->default_value = null;
483                 }
484                 // just guess expected number of deciaml places :-(
485                 if ($matches[1] == 8) {
486                     // total 15 digits
487                     $info->max_length = 8;
488                     $info->scale      = 7;
489                 } else {
490                     // total 6 digits
491                     $info->max_length = 4;
492                     $info->scale      = 2;
493                 }
495             } else if ($rawcolumn->type === 'text') {
496                 $info->type          = $rawcolumn->type;
497                 $info->meta_type     = 'X';
498                 $info->max_length    = -1;
499                 $info->scale         = null;
500                 $info->not_null      = ($rawcolumn->attnotnull === 't');
501                 $info->has_default   = ($rawcolumn->atthasdef === 't');
502                 if ($info->has_default) {
503                     $parts = explode('::', $rawcolumn->adsrc);
504                     if (count($parts) > 1) {
505                         $info->default_value = reset($parts);
506                         $info->default_value = trim($info->default_value, "'");
507                     } else {
508                         $info->default_value = $rawcolumn->adsrc;
509                     }
510                 } else {
511                     $info->default_value = null;
512                 }
513                 $info->primary_key   = false;
514                 $info->binary        = false;
515                 $info->unsigned      = null;
516                 $info->auto_increment= false;
517                 $info->unique        = null;
519             } else if ($rawcolumn->type === 'bytea') {
520                 $info->type          = $rawcolumn->type;
521                 $info->meta_type     = 'B';
522                 $info->max_length    = -1;
523                 $info->scale         = null;
524                 $info->not_null      = ($rawcolumn->attnotnull === 't');
525                 $info->has_default   = false;
526                 $info->default_value = null;
527                 $info->primary_key   = false;
528                 $info->binary        = true;
529                 $info->unsigned      = null;
530                 $info->auto_increment= false;
531                 $info->unique        = null;
533             }
535             $this->columns[$table][$info->name] = new database_column_info($info);
536         }
538         pg_free_result($result);
540         return $this->columns[$table];
541     }
543     /**
544      * Normalise values based in RDBMS dependencies (booleans, LOBs...)
545      *
546      * @param database_column_info $column column metadata corresponding with the value we are going to normalise
547      * @param mixed $value value we are going to normalise
548      * @return mixed the normalised value
549      */
550     protected function normalise_value($column, $value) {
551         $this->detect_objects($value);
553         if (is_bool($value)) { // Always, convert boolean to int
554             $value = (int)$value;
556         } else if ($column->meta_type === 'B') { // BLOB detected, we return 'blob' array instead of raw value to allow
557             if (!is_null($value)) {             // binding/executing code later to know about its nature
558                 $value = array('blob' => $value);
559             }
561         } else if ($value === '') {
562             if ($column->meta_type === 'I' or $column->meta_type === 'F' or $column->meta_type === 'N') {
563                 $value = 0; // prevent '' problems in numeric fields
564             }
565         }
566         return $value;
567     }
569     /**
570      * Is db in unicode mode?
571      * @return bool
572      */
573     public function setup_is_unicodedb() {
574         // Get PostgreSQL server_encoding value
575         $sql = "SHOW server_encoding";
576         $this->query_start($sql, null, SQL_QUERY_AUX);
577         $result = pg_query($this->pgsql, $sql);
578         $this->query_end($result);
580         if (!$result) {
581             return false;
582         }
583         $rawcolumn = pg_fetch_object($result);
584         $encoding = $rawcolumn->server_encoding;
585         pg_free_result($result);
587         return (strtoupper($encoding) == 'UNICODE' || strtoupper($encoding) == 'UTF8');
588     }
590     /**
591      * Do NOT use in code, to be used by database_manager only!
592      * @param string $sql query
593      * @return bool true
594      * @throws dml_exception A DML specific exception is thrown for any errors.
595      */
596     public function change_database_structure($sql) {
597         $this->reset_caches();
599         $this->query_start($sql, null, SQL_QUERY_STRUCTURE);
600         $result = pg_query($this->pgsql, $sql);
601         $this->query_end($result);
603         pg_free_result($result);
604         return true;
605     }
607     /**
608      * Execute general sql query. Should be used only when no other method suitable.
609      * Do NOT use this to make changes in db structure, use database_manager methods instead!
610      * @param string $sql query
611      * @param array $params query parameters
612      * @return bool true
613      * @throws dml_exception A DML specific exception is thrown for any errors.
614      */
615     public function execute($sql, array $params=null) {
616         list($sql, $params, $type) = $this->fix_sql_params($sql, $params);
618         if (strpos($sql, ';') !== false) {
619             throw new coding_exception('moodle_database::execute() Multiple sql statements found or bound parameters not used properly in query!');
620         }
622         $this->query_start($sql, $params, SQL_QUERY_UPDATE);
623         $result = pg_query_params($this->pgsql, $sql, $params);
624         $this->query_end($result);
626         pg_free_result($result);
627         return true;
628     }
630     /**
631      * Get a number of records as a moodle_recordset using a SQL statement.
632      *
633      * Since this method is a little less readable, use of it should be restricted to
634      * code where it's possible there might be large datasets being returned.  For known
635      * small datasets use get_records_sql - it leads to simpler code.
636      *
637      * The return type is like:
638      * @see function get_recordset.
639      *
640      * @param string $sql the SQL select query to execute.
641      * @param array $params array of sql parameters
642      * @param int $limitfrom return a subset of records, starting at this point (optional, required if $limitnum is set).
643      * @param int $limitnum return a subset comprising this many records (optional, required if $limitfrom is set).
644      * @return moodle_recordset instance
645      * @throws dml_exception A DML specific exception is thrown for any errors.
646      */
647     public function get_recordset_sql($sql, array $params=null, $limitfrom=0, $limitnum=0) {
648         $limitfrom = (int)$limitfrom;
649         $limitnum  = (int)$limitnum;
650         $limitfrom = ($limitfrom < 0) ? 0 : $limitfrom;
651         $limitnum  = ($limitnum < 0)  ? 0 : $limitnum;
652         if ($limitfrom or $limitnum) {
653             if ($limitnum < 1) {
654                 $limitnum = "ALL";
655             } else if (PHP_INT_MAX - $limitnum < $limitfrom) {
656                 // this is a workaround for weird max int problem
657                 $limitnum = "ALL";
658             }
659             $sql .= " LIMIT $limitnum OFFSET $limitfrom";
660         }
662         list($sql, $params, $type) = $this->fix_sql_params($sql, $params);
664         $this->query_start($sql, $params, SQL_QUERY_SELECT);
665         $result = pg_query_params($this->pgsql, $sql, $params);
666         $this->query_end($result);
668         return $this->create_recordset($result);
669     }
671     protected function create_recordset($result) {
672         return new pgsql_native_moodle_recordset($result, $this->bytea_oid);
673     }
675     /**
676      * Get a number of records as an array of objects using a SQL statement.
677      *
678      * Return value is like:
679      * @see function get_records.
680      *
681      * @param string $sql the SQL select query to execute. The first column of this SELECT statement
682      *   must be a unique value (usually the 'id' field), as it will be used as the key of the
683      *   returned array.
684      * @param array $params array of sql parameters
685      * @param int $limitfrom return a subset of records, starting at this point (optional, required if $limitnum is set).
686      * @param int $limitnum return a subset comprising this many records (optional, required if $limitfrom is set).
687      * @return array of objects, or empty array if no records were found
688      * @throws dml_exception A DML specific exception is thrown for any errors.
689      */
690     public function get_records_sql($sql, array $params=null, $limitfrom=0, $limitnum=0) {
691         $limitfrom = (int)$limitfrom;
692         $limitnum  = (int)$limitnum;
693         $limitfrom = ($limitfrom < 0) ? 0 : $limitfrom;
694         $limitnum  = ($limitnum < 0)  ? 0 : $limitnum;
695         if ($limitfrom or $limitnum) {
696             if ($limitnum < 1) {
697                 $limitnum = "ALL";
698             } else if (PHP_INT_MAX - $limitnum < $limitfrom) {
699                 // this is a workaround for weird max int problem
700                 $limitnum = "ALL";
701             }
702             $sql .= " LIMIT $limitnum OFFSET $limitfrom";
703         }
705         list($sql, $params, $type) = $this->fix_sql_params($sql, $params);
706         $this->query_start($sql, $params, SQL_QUERY_SELECT);
707         $result = pg_query_params($this->pgsql, $sql, $params);
708         $this->query_end($result);
710         // find out if there are any blobs
711         $numrows = pg_num_fields($result);
712         $blobs = array();
713         for($i=0; $i<$numrows; $i++) {
714             $type_oid = pg_field_type_oid($result, $i);
715             if ($type_oid == $this->bytea_oid) {
716                 $blobs[] = pg_field_name($result, $i);
717             }
718         }
720         $rows = pg_fetch_all($result);
721         pg_free_result($result);
723         $return = array();
724         if ($rows) {
725             foreach ($rows as $row) {
726                 $id = reset($row);
727                 if ($blobs) {
728                     foreach ($blobs as $blob) {
729                         // note: in PostgreSQL 9.0 the returned blobs are hexencoded by default - see http://www.postgresql.org/docs/9.0/static/runtime-config-client.html#GUC-BYTEA-OUTPUT
730                         $row[$blob] = $row[$blob] !== null ? pg_unescape_bytea($row[$blob]) : null;
731                     }
732                 }
733                 if (isset($return[$id])) {
734                     $colname = key($row);
735                     debugging("Did you remember to make the first column something unique in your call to get_records? Duplicate value '$id' found in column '$colname'.", DEBUG_DEVELOPER);
736                 }
737                 $return[$id] = (object)$row;
738             }
739         }
741         return $return;
742     }
744     /**
745      * Selects records and return values (first field) as an array using a SQL statement.
746      *
747      * @param string $sql The SQL query
748      * @param array $params array of sql parameters
749      * @return array of values
750      * @throws dml_exception A DML specific exception is thrown for any errors.
751      */
752     public function get_fieldset_sql($sql, array $params=null) {
753         list($sql, $params, $type) = $this->fix_sql_params($sql, $params);
755         $this->query_start($sql, $params, SQL_QUERY_SELECT);
756         $result = pg_query_params($this->pgsql, $sql, $params);
757         $this->query_end($result);
759         $return = pg_fetch_all_columns($result, 0);
760         pg_free_result($result);
762         return $return;
763     }
765     /**
766      * Insert new record into database, as fast as possible, no safety checks, lobs not supported.
767      * @param string $table name
768      * @param mixed $params data record as object or array
769      * @param bool $returnit return it of inserted record
770      * @param bool $bulk true means repeated inserts expected
771      * @param bool $customsequence true if 'id' included in $params, disables $returnid
772      * @return bool|int true or new id
773      * @throws dml_exception A DML specific exception is thrown for any errors.
774      */
775     public function insert_record_raw($table, $params, $returnid=true, $bulk=false, $customsequence=false) {
776         if (!is_array($params)) {
777             $params = (array)$params;
778         }
780         $returning = "";
782         if ($customsequence) {
783             if (!isset($params['id'])) {
784                 throw new coding_exception('moodle_database::insert_record_raw() id field must be specified if custom sequences used.');
785             }
786             $returnid = false;
787         } else {
788             if ($returnid) {
789                 $returning = "RETURNING id";
790                 unset($params['id']);
791             } else {
792                 unset($params['id']);
793             }
794         }
796         if (empty($params)) {
797             throw new coding_exception('moodle_database::insert_record_raw() no fields found.');
798         }
800         $fields = implode(',', array_keys($params));
801         $values = array();
802         $i = 1;
803         foreach ($params as $value) {
804             $this->detect_objects($value);
805             $values[] = "\$".$i++;
806         }
807         $values = implode(',', $values);
809         $sql = "INSERT INTO {$this->prefix}$table ($fields) VALUES($values) $returning";
810         $this->query_start($sql, $params, SQL_QUERY_INSERT);
811         $result = pg_query_params($this->pgsql, $sql, $params);
812         $this->query_end($result);
814         if ($returning !== "") {
815             $row = pg_fetch_assoc($result);
816             $params['id'] = reset($row);
817         }
818         pg_free_result($result);
820         if (!$returnid) {
821             return true;
822         }
824         return (int)$params['id'];
825     }
827     /**
828      * Insert a record into a table and return the "id" field if required.
829      *
830      * Some conversions and safety checks are carried out. Lobs are supported.
831      * If the return ID isn't required, then this just reports success as true/false.
832      * $data is an object containing needed data
833      * @param string $table The database table to be inserted into
834      * @param object $data A data object with values for one or more fields in the record
835      * @param bool $returnid Should the id of the newly created record entry be returned? If this option is not requested then true/false is returned.
836      * @return bool|int true or new id
837      * @throws dml_exception A DML specific exception is thrown for any errors.
838      */
839     public function insert_record($table, $dataobject, $returnid=true, $bulk=false) {
840         $dataobject = (array)$dataobject;
842         $columns = $this->get_columns($table);
843         $cleaned = array();
844         $blobs   = array();
846         foreach ($dataobject as $field=>$value) {
847             if ($field === 'id') {
848                 continue;
849             }
850             if (!isset($columns[$field])) {
851                 continue;
852             }
853             $column = $columns[$field];
854             $normalised_value = $this->normalise_value($column, $value);
855             if (is_array($normalised_value) && array_key_exists('blob', $normalised_value)) {
856                 $cleaned[$field] = '@#BLOB#@';
857                 $blobs[$field] = $normalised_value['blob'];
858             } else {
859                 $cleaned[$field] = $normalised_value;
860             }
861         }
863         if (empty($blobs)) {
864             return $this->insert_record_raw($table, $cleaned, $returnid, $bulk);
865         }
867         $id = $this->insert_record_raw($table, $cleaned, true, $bulk);
869         foreach ($blobs as $key=>$value) {
870             $value = pg_escape_bytea($this->pgsql, $value);
871             $sql = "UPDATE {$this->prefix}$table SET $key = '$value'::bytea WHERE id = $id";
872             $this->query_start($sql, NULL, SQL_QUERY_UPDATE);
873             $result = pg_query($this->pgsql, $sql);
874             $this->query_end($result);
875             if ($result !== false) {
876                 pg_free_result($result);
877             }
878         }
880         return ($returnid ? $id : true);
882     }
884     /**
885      * Import a record into a table, id field is required.
886      * Safety checks are NOT carried out. Lobs are supported.
887      *
888      * @param string $table name of database table to be inserted into
889      * @param object $dataobject A data object with values for one or more fields in the record
890      * @return bool true
891      * @throws dml_exception A DML specific exception is thrown for any errors.
892      */
893     public function import_record($table, $dataobject) {
894         $dataobject = (array)$dataobject;
896         $columns = $this->get_columns($table);
897         $cleaned = array();
898         $blobs   = array();
900         foreach ($dataobject as $field=>$value) {
901             $this->detect_objects($value);
902             if (!isset($columns[$field])) {
903                 continue;
904             }
905             if ($columns[$field]->meta_type === 'B') {
906                 if (!is_null($value)) {
907                     $cleaned[$field] = '@#BLOB#@';
908                     $blobs[$field] = $value;
909                     continue;
910                 }
911             }
913             $cleaned[$field] = $value;
914         }
916         $this->insert_record_raw($table, $cleaned, false, true, true);
917         $id = $dataobject['id'];
919         foreach ($blobs as $key=>$value) {
920             $value = pg_escape_bytea($this->pgsql, $value);
921             $sql = "UPDATE {$this->prefix}$table SET $key = '$value'::bytea WHERE id = $id";
922             $this->query_start($sql, NULL, SQL_QUERY_UPDATE);
923             $result = pg_query($this->pgsql, $sql);
924             $this->query_end($result);
925             if ($result !== false) {
926                 pg_free_result($result);
927             }
928         }
930         return true;
931     }
933     /**
934      * Update record in database, as fast as possible, no safety checks, lobs not supported.
935      * @param string $table name
936      * @param mixed $params data record as object or array
937      * @param bool true means repeated updates expected
938      * @return bool true
939      * @throws dml_exception A DML specific exception is thrown for any errors.
940      */
941     public function update_record_raw($table, $params, $bulk=false) {
942         $params = (array)$params;
944         if (!isset($params['id'])) {
945             throw new coding_exception('moodle_database::update_record_raw() id field must be specified.');
946         }
947         $id = $params['id'];
948         unset($params['id']);
950         if (empty($params)) {
951             throw new coding_exception('moodle_database::update_record_raw() no fields found.');
952         }
954         $i = 1;
956         $sets = array();
957         foreach ($params as $field=>$value) {
958             $this->detect_objects($value);
959             $sets[] = "$field = \$".$i++;
960         }
962         $params[] = $id; // last ? in WHERE condition
964         $sets = implode(',', $sets);
965         $sql = "UPDATE {$this->prefix}$table SET $sets WHERE id=\$".$i;
967         $this->query_start($sql, $params, SQL_QUERY_UPDATE);
968         $result = pg_query_params($this->pgsql, $sql, $params);
969         $this->query_end($result);
971         pg_free_result($result);
972         return true;
973     }
975     /**
976      * Update a record in a table
977      *
978      * $dataobject is an object containing needed data
979      * Relies on $dataobject having a variable "id" to
980      * specify the record to update
981      *
982      * @param string $table The database table to be checked against.
983      * @param object $dataobject An object with contents equal to fieldname=>fieldvalue. Must have an entry for 'id' to map to the table specified.
984      * @param bool true means repeated updates expected
985      * @return bool true
986      * @throws dml_exception A DML specific exception is thrown for any errors.
987      */
988     public function update_record($table, $dataobject, $bulk=false) {
989         $dataobject = (array)$dataobject;
991         $columns = $this->get_columns($table);
992         $cleaned = array();
993         $blobs   = array();
995         foreach ($dataobject as $field=>$value) {
996             if (!isset($columns[$field])) {
997                 continue;
998             }
999             $column = $columns[$field];
1000             $normalised_value = $this->normalise_value($column, $value);
1001             if (is_array($normalised_value) && array_key_exists('blob', $normalised_value)) {
1002                 $cleaned[$field] = '@#BLOB#@';
1003                 $blobs[$field] = $normalised_value['blob'];
1004             } else {
1005                 $cleaned[$field] = $normalised_value;
1006             }
1007         }
1009         $this->update_record_raw($table, $cleaned, $bulk);
1011         if (empty($blobs)) {
1012             return true;
1013         }
1015         $id = (int)$dataobject['id'];
1017         foreach ($blobs as $key=>$value) {
1018             $value = pg_escape_bytea($this->pgsql, $value);
1019             $sql = "UPDATE {$this->prefix}$table SET $key = '$value'::bytea WHERE id = $id";
1020             $this->query_start($sql, NULL, SQL_QUERY_UPDATE);
1021             $result = pg_query($this->pgsql, $sql);
1022             $this->query_end($result);
1024             pg_free_result($result);
1025         }
1027         return true;
1028     }
1030     /**
1031      * Set a single field in every table record which match a particular WHERE clause.
1032      *
1033      * @param string $table The database table to be checked against.
1034      * @param string $newfield the field to set.
1035      * @param string $newvalue the value to set the field to.
1036      * @param string $select A fragment of SQL to be used in a where clause in the SQL call.
1037      * @param array $params array of sql parameters
1038      * @return bool true
1039      * @throws dml_exception A DML specific exception is thrown for any errors.
1040      */
1041     public function set_field_select($table, $newfield, $newvalue, $select, array $params=null) {
1043         if ($select) {
1044             $select = "WHERE $select";
1045         }
1046         if (is_null($params)) {
1047             $params = array();
1048         }
1049         list($select, $params, $type) = $this->fix_sql_params($select, $params);
1050         $i = count($params)+1;
1052         // Get column metadata
1053         $columns = $this->get_columns($table);
1054         $column = $columns[$newfield];
1056         $normalised_value = $this->normalise_value($column, $newvalue);
1057         if (is_array($normalised_value) && array_key_exists('blob', $normalised_value)) {
1058             // Update BYTEA and return
1059             $normalised_value = pg_escape_bytea($this->pgsql, $normalised_value['blob']);
1060             $sql = "UPDATE {$this->prefix}$table SET $newfield = '$normalised_value'::bytea $select";
1061             $this->query_start($sql, NULL, SQL_QUERY_UPDATE);
1062             $result = pg_query_params($this->pgsql, $sql, $params);
1063             $this->query_end($result);
1064             pg_free_result($result);
1065             return true;
1066         }
1068         if (is_null($normalised_value)) {
1069             $newfield = "$newfield = NULL";
1070         } else {
1071             $newfield = "$newfield = \$".$i;
1072             $params[] = $normalised_value;
1073         }
1074         $sql = "UPDATE {$this->prefix}$table SET $newfield $select";
1076         $this->query_start($sql, $params, SQL_QUERY_UPDATE);
1077         $result = pg_query_params($this->pgsql, $sql, $params);
1078         $this->query_end($result);
1080         pg_free_result($result);
1082         return true;
1083     }
1085     /**
1086      * Delete one or more records from a table which match a particular WHERE clause.
1087      *
1088      * @param string $table The database table to be checked against.
1089      * @param string $select A fragment of SQL to be used in a where clause in the SQL call (used to define the selection criteria).
1090      * @param array $params array of sql parameters
1091      * @return bool true
1092      * @throws dml_exception A DML specific exception is thrown for any errors.
1093      */
1094     public function delete_records_select($table, $select, array $params=null) {
1095         if ($select) {
1096             $select = "WHERE $select";
1097         }
1098         $sql = "DELETE FROM {$this->prefix}$table $select";
1100         list($sql, $params, $type) = $this->fix_sql_params($sql, $params);
1102         $this->query_start($sql, $params, SQL_QUERY_UPDATE);
1103         $result = pg_query_params($this->pgsql, $sql, $params);
1104         $this->query_end($result);
1106         pg_free_result($result);
1108         return true;
1109     }
1111     /**
1112      * Returns 'LIKE' part of a query.
1113      *
1114      * @param string $fieldname usually name of the table column
1115      * @param string $param usually bound query parameter (?, :named)
1116      * @param bool $casesensitive use case sensitive search
1117      * @param bool $accensensitive use accent sensitive search (not all databases support accent insensitive)
1118      * @param bool $notlike true means "NOT LIKE"
1119      * @param string $escapechar escape char for '%' and '_'
1120      * @return string SQL code fragment
1121      */
1122     public function sql_like($fieldname, $param, $casesensitive = true, $accentsensitive = true, $notlike = false, $escapechar = '\\') {
1123         if (strpos($param, '%') !== false) {
1124             debugging('Potential SQL injection detected, sql_like() expects bound parameters (? or :named)');
1125         }
1126         if ($escapechar === '\\') {
1127             // Prevents problems with C-style escapes of enclosing '\',
1128             // E'... bellow prevents compatibility warnings.
1129             $escapechar = '\\\\';
1130         }
1132         // postgresql does not support accent insensitive text comparisons, sorry
1133         if ($casesensitive) {
1134             $LIKE = $notlike ? 'NOT LIKE' : 'LIKE';
1135         } else {
1136             $LIKE = $notlike ? 'NOT ILIKE' : 'ILIKE';
1137         }
1138         return "$fieldname $LIKE $param ESCAPE E'$escapechar'";
1139     }
1141     public function sql_bitxor($int1, $int2) {
1142         return '((' . $int1 . ') # (' . $int2 . '))';
1143     }
1145     public function sql_cast_char2int($fieldname, $text=false) {
1146         return ' CAST(' . $fieldname . ' AS INT) ';
1147     }
1149     public function sql_cast_char2real($fieldname, $text=false) {
1150         return " $fieldname::real ";
1151     }
1153     public function sql_concat() {
1154         $arr = func_get_args();
1155         $s = implode(' || ', $arr);
1156         if ($s === '') {
1157             return " '' ";
1158         }
1159         // Add always empty string element so integer-exclusive concats
1160         // will work without needing to cast each element explicitly
1161         return " '' || $s ";
1162     }
1164     public function sql_concat_join($separator="' '", $elements=array()) {
1165         for ($n=count($elements)-1; $n > 0 ; $n--) {
1166             array_splice($elements, $n, 0, $separator);
1167         }
1168         $s = implode(' || ', $elements);
1169         if ($s === '') {
1170             return " '' ";
1171         }
1172         return " $s ";
1173     }
1175     public function sql_regex_supported() {
1176         return true;
1177     }
1179     public function sql_regex($positivematch=true) {
1180         return $positivematch ? '~*' : '!~*';
1181     }
1183     public function session_lock_supported() {
1184         return true;
1185     }
1187     /**
1188      * Obtain session lock
1189      * @param int $rowid id of the row with session record
1190      * @param int $timeout max allowed time to wait for the lock in seconds
1191      * @return bool success
1192      */
1193     public function get_session_lock($rowid, $timeout) {
1194         // NOTE: there is a potential locking problem for database running
1195         //       multiple instances of moodle, we could try to use pg_advisory_lock(int, int),
1196         //       luckily there is not a big chance that they would collide
1197         if (!$this->session_lock_supported()) {
1198             return;
1199         }
1201         parent::get_session_lock($rowid, $timeout);
1203         $timeoutmilli = $timeout * 1000;
1205         $sql = "SET statement_timeout TO $timeoutmilli";
1206         $this->query_start($sql, null, SQL_QUERY_AUX);
1207         $result = pg_query($this->pgsql, $sql);
1208         $this->query_end($result);
1210         if ($result) {
1211             pg_free_result($result);
1212         }
1214         $sql = "SELECT pg_advisory_lock($rowid)";
1215         $this->query_start($sql, null, SQL_QUERY_AUX);
1216         $start = time();
1217         $result = pg_query($this->pgsql, $sql);
1218         $end = time();
1219         try {
1220             $this->query_end($result);
1221         } catch (dml_exception $ex) {
1222             if ($end - $start >= $timeout) {
1223                 throw new dml_sessionwait_exception();
1224             } else {
1225                 throw $ex;
1226             }
1227         }
1229         if ($result) {
1230             pg_free_result($result);
1231         }
1233         $sql = "SET statement_timeout TO DEFAULT";
1234         $this->query_start($sql, null, SQL_QUERY_AUX);
1235         $result = pg_query($this->pgsql, $sql);
1236         $this->query_end($result);
1238         if ($result) {
1239             pg_free_result($result);
1240         }
1241     }
1243     public function release_session_lock($rowid) {
1244         if (!$this->session_lock_supported()) {
1245             return;
1246         }
1247         parent::release_session_lock($rowid);
1249         $sql = "SELECT pg_advisory_unlock($rowid)";
1250         $this->query_start($sql, null, SQL_QUERY_AUX);
1251         $result = pg_query($this->pgsql, $sql);
1252         $this->query_end($result);
1254         if ($result) {
1255             pg_free_result($result);
1256         }
1257     }
1259     /**
1260      * Driver specific start of real database transaction,
1261      * this can not be used directly in code.
1262      * @return void
1263      */
1264     protected function begin_transaction() {
1265         $sql = "BEGIN ISOLATION LEVEL READ COMMITTED";
1266         $this->query_start($sql, NULL, SQL_QUERY_AUX);
1267         $result = pg_query($this->pgsql, $sql);
1268         $this->query_end($result);
1270         pg_free_result($result);
1271     }
1273     /**
1274      * Driver specific commit of real database transaction,
1275      * this can not be used directly in code.
1276      * @return void
1277      */
1278     protected function commit_transaction() {
1279         $sql = "COMMIT";
1280         $this->query_start($sql, NULL, SQL_QUERY_AUX);
1281         $result = pg_query($this->pgsql, $sql);
1282         $this->query_end($result);
1284         pg_free_result($result);
1285     }
1287     /**
1288      * Driver specific abort of real database transaction,
1289      * this can not be used directly in code.
1290      * @return void
1291      */
1292     protected function rollback_transaction() {
1293         $sql = "ROLLBACK";
1294         $this->query_start($sql, NULL, SQL_QUERY_AUX);
1295         $result = pg_query($this->pgsql, $sql);
1296         $this->query_end($result);
1298         pg_free_result($result);
1299     }
1301     /**
1302      * Helper function trimming (whitespace + quotes) any string
1303      * needed because PG uses to enclose with double quotes some
1304      * fields in indexes definition and others
1305      *
1306      * @param string $str string to apply whitespace + quotes trim
1307      * @return string trimmed string
1308      */
1309     private function trim_quotes($str) {
1310         return trim(trim($str), "'\"");
1311     }