[Summary view] [Print] [Text view]

   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/>.
  16  
  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   */
  24  
  25  defined('MOODLE_INTERNAL') || die();
  26  
  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');
  30  
  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 {
  39  
  40      /** @var resource $pgsql database resource */
  41      protected $pgsql     = null;
  42      protected $bytea_oid = null;
  43  
  44      protected $last_error_reporting; // To handle pgsql driver default verbosity
  45  
  46      /** @var bool savepoint hack for MDL-35506 - workaround for automatic transaction rollback on error */
  47      protected $savepointpresent = false;
  48  
  49      /**
  50       * Detects if all needed PHP stuff installed.
  51       * Note: can be used before connect()
  52       * @return mixed true if ok, string if something
  53       */
  54      public function driver_installed() {
  55          if (!extension_loaded('pgsql')) {
  56              return get_string('pgsqlextensionisnotpresentinphp', 'install');
  57          }
  58          return true;
  59      }
  60  
  61      /**
  62       * Returns database family type - describes SQL dialect
  63       * Note: can be used before connect()
  64       * @return string db family name (mysql, postgres, mssql, oracle, etc.)
  65       */
  66      public function get_dbfamily() {
  67          return 'postgres';
  68      }
  69  
  70      /**
  71       * Returns more specific database driver type
  72       * Note: can be used before connect()
  73       * @return string db type mysqli, pgsql, oci, mssql, sqlsrv
  74       */
  75      protected function get_dbtype() {
  76          return 'pgsql';
  77      }
  78  
  79      /**
  80       * Returns general database library name
  81       * Note: can be used before connect()
  82       * @return string db type pdo, native
  83       */
  84      protected function get_dblibrary() {
  85          return 'native';
  86      }
  87  
  88      /**
  89       * Returns localised database type name
  90       * Note: can be used before connect()
  91       * @return string
  92       */
  93      public function get_name() {
  94          return get_string('nativepgsql', 'install');
  95      }
  96  
  97      /**
  98       * Returns localised database configuration help.
  99       * Note: can be used before connect()
 100       * @return string
 101       */
 102      public function get_configuration_help() {
 103          return get_string('nativepgsqlhelp', 'install');
 104      }
 105  
 106      /**
 107       * Connect to db
 108       * Must be called before other methods.
 109       * @param string $dbhost The database host.
 110       * @param string $dbuser The database username.
 111       * @param string $dbpass The database username's password.
 112       * @param string $dbname The name of the database being connected to.
 113       * @param mixed $prefix string means moodle db prefix, false used for external databases where prefix not used
 114       * @param array $dboptions driver specific options
 115       * @return bool true
 116       * @throws dml_connection_exception if error
 117       */
 118      public function connect($dbhost, $dbuser, $dbpass, $dbname, $prefix, array $dboptions=null) {
 119          if ($prefix == '' and !$this->external) {
 120              //Enforce prefixes for everybody but mysql
 121              throw new dml_exception('prefixcannotbeempty', $this->get_dbfamily());
 122          }
 123  
 124          $driverstatus = $this->driver_installed();
 125  
 126          if ($driverstatus !== true) {
 127              throw new dml_exception('dbdriverproblem', $driverstatus);
 128          }
 129  
 130          $this->store_settings($dbhost, $dbuser, $dbpass, $dbname, $prefix, $dboptions);
 131  
 132          $pass = addcslashes($this->dbpass, "'\\");
 133  
 134          // Unix socket connections should have lower overhead
 135          if (!empty($this->dboptions['dbsocket']) and ($this->dbhost === 'localhost' or $this->dbhost === '127.0.0.1')) {
 136              $connection = "user='$this->dbuser' password='$pass' dbname='$this->dbname'";
 137              if (strpos($this->dboptions['dbsocket'], '/') !== false) {
 138                  $connection = $connection." host='".$this->dboptions['dbsocket']."'";
 139                  if (!empty($this->dboptions['dbport'])) {
 140                      // Somehow non-standard port is important for sockets - see MDL-44862.
 141                      $connection = $connection." port ='".$this->dboptions['dbport']."'";
 142                  }
 143              }
 144          } else {
 145              $this->dboptions['dbsocket'] = '';
 146              if (empty($this->dbname)) {
 147                  // probably old style socket connection - do not add port
 148                  $port = "";
 149              } else if (empty($this->dboptions['dbport'])) {
 150                  $port = "port ='5432'";
 151              } else {
 152                  $port = "port ='".$this->dboptions['dbport']."'";
 153              }
 154              $connection = "host='$this->dbhost' $port user='$this->dbuser' password='$pass' dbname='$this->dbname'";
 155          }
 156  
 157          ob_start();
 158          if (empty($this->dboptions['dbpersist'])) {
 159              $this->pgsql = pg_connect($connection, PGSQL_CONNECT_FORCE_NEW);
 160          } else {
 161              $this->pgsql = pg_pconnect($connection, PGSQL_CONNECT_FORCE_NEW);
 162          }
 163          $dberr = ob_get_contents();
 164          ob_end_clean();
 165  
 166          $status = pg_connection_status($this->pgsql);
 167  
 168          if ($status === false or $status === PGSQL_CONNECTION_BAD) {
 169              $this->pgsql = null;
 170              throw new dml_connection_exception($dberr);
 171          }
 172  
 173          $this->query_start("--pg_set_client_encoding()", null, SQL_QUERY_AUX);
 174          pg_set_client_encoding($this->pgsql, 'utf8');
 175          $this->query_end(true);
 176  
 177          $sql = '';
 178          // Only for 9.0 and upwards, set bytea encoding to old format.
 179          if ($this->is_min_version('9.0')) {
 180              $sql = "SET bytea_output = 'escape'; ";
 181          }
 182  
 183          // Select schema if specified, otherwise the first one wins.
 184          if (!empty($this->dboptions['dbschema'])) {
 185              $sql .= "SET search_path = '".$this->dboptions['dbschema']."'; ";
 186          }
 187  
 188          // Find out the bytea oid.
 189          $sql .= "SELECT oid FROM pg_type WHERE typname = 'bytea'";
 190          $this->query_start($sql, null, SQL_QUERY_AUX);
 191          $result = pg_query($this->pgsql, $sql);
 192          $this->query_end($result);
 193  
 194          $this->bytea_oid = pg_fetch_result($result, 0, 0);
 195          pg_free_result($result);
 196          if ($this->bytea_oid === false) {
 197              $this->pgsql = null;
 198              throw new dml_connection_exception('Can not read bytea type.');
 199          }
 200  
 201          // Connection stabilised and configured, going to instantiate the temptables controller
 202          $this->temptables = new pgsql_native_moodle_temptables($this);
 203  
 204          return true;
 205      }
 206  
 207      /**
 208       * Close database connection and release all resources
 209       * and memory (especially circular memory references).
 210       * Do NOT use connect() again, create a new instance if needed.
 211       */
 212      public function dispose() {
 213          parent::dispose(); // Call parent dispose to write/close session and other common stuff before closing connection
 214          if ($this->pgsql) {
 215              pg_close($this->pgsql);
 216              $this->pgsql = null;
 217          }
 218      }
 219  
 220  
 221      /**
 222       * Called before each db query.
 223       * @param string $sql
 224       * @param array array of parameters
 225       * @param int $type type of query
 226       * @param mixed $extrainfo driver specific extra information
 227       * @return void
 228       */
 229      protected function query_start($sql, array $params=null, $type, $extrainfo=null) {
 230          parent::query_start($sql, $params, $type, $extrainfo);
 231          // pgsql driver tents to send debug to output, we do not need that ;-)
 232          $this->last_error_reporting = error_reporting(0);
 233      }
 234  
 235      /**
 236       * Called immediately after each db query.
 237       * @param mixed db specific result
 238       * @return void
 239       */
 240      protected function query_end($result) {
 241          // reset original debug level
 242          error_reporting($this->last_error_reporting);
 243          try {
 244              parent::query_end($result);
 245              if ($this->savepointpresent and $this->last_type != SQL_QUERY_AUX and $this->last_type != SQL_QUERY_SELECT) {
 246                  $res = @pg_query($this->pgsql, "RELEASE SAVEPOINT moodle_pg_savepoint; SAVEPOINT moodle_pg_savepoint");
 247                  if ($res) {
 248                      pg_free_result($res);
 249                  }
 250              }
 251          } catch (Exception $e) {
 252              if ($this->savepointpresent) {
 253                  $res = @pg_query($this->pgsql, "ROLLBACK TO SAVEPOINT moodle_pg_savepoint; SAVEPOINT moodle_pg_savepoint");
 254                  if ($res) {
 255                      pg_free_result($res);
 256                  }
 257              }
 258              throw $e;
 259          }
 260      }
 261  
 262      /**
 263       * Returns database server info array
 264       * @return array Array containing 'description' and 'version' info
 265       */
 266      public function get_server_info() {
 267          static $info;
 268          if (!$info) {
 269              $this->query_start("--pg_version()", null, SQL_QUERY_AUX);
 270              $info = pg_version($this->pgsql);
 271              $this->query_end(true);
 272          }
 273          return array('description'=>$info['server'], 'version'=>$info['server']);
 274      }
 275  
 276      /**
 277       * Returns if the RDBMS server fulfills the required version
 278       *
 279       * @param string $version version to check against
 280       * @return bool returns if the version is fulfilled (true) or no (false)
 281       */
 282      private function is_min_version($version) {
 283          $server = $this->get_server_info();
 284          $server = $server['version'];
 285          return version_compare($server, $version, '>=');
 286      }
 287  
 288      /**
 289       * Returns supported query parameter types
 290       * @return int bitmask of accepted SQL_PARAMS_*
 291       */
 292      protected function allowed_param_types() {
 293          return SQL_PARAMS_DOLLAR;
 294      }
 295  
 296      /**
 297       * Returns last error reported by database engine.
 298       * @return string error message
 299       */
 300      public function get_last_error() {
 301          return pg_last_error($this->pgsql);
 302      }
 303  
 304      /**
 305       * Return tables in database WITHOUT current prefix.
 306       * @param bool $usecache if true, returns list of cached tables.
 307       * @return array of table names in lowercase and without prefix
 308       */
 309      public function get_tables($usecache=true) {
 310          if ($usecache and $this->tables !== null) {
 311              return $this->tables;
 312          }
 313          $this->tables = array();
 314          $prefix = str_replace('_', '|_', $this->prefix);
 315          $sql = "SELECT c.relname
 316                    FROM pg_catalog.pg_class c
 317                    JOIN pg_catalog.pg_namespace as ns ON ns.oid = c.relnamespace
 318                   WHERE c.relname LIKE '$prefix%' ESCAPE '|'
 319                         AND c.relkind = 'r'
 320                         AND (ns.nspname = current_schema() OR ns.oid = pg_my_temp_schema())";
 321          $this->query_start($sql, null, SQL_QUERY_AUX);
 322          $result = pg_query($this->pgsql, $sql);
 323          $this->query_end($result);
 324  
 325          if ($result) {
 326              while ($row = pg_fetch_row($result)) {
 327                  $tablename = reset($row);
 328                  if ($this->prefix !== false && $this->prefix !== '') {
 329                      if (strpos($tablename, $this->prefix) !== 0) {
 330                          continue;
 331                      }
 332                      $tablename = substr($tablename, strlen($this->prefix));
 333                  }
 334                  $this->tables[$tablename] = $tablename;
 335              }
 336              pg_free_result($result);
 337          }
 338          return $this->tables;
 339      }
 340  
 341      /**
 342       * Return table indexes - everything lowercased.
 343       * @param string $table The table we want to get indexes from.
 344       * @return array of arrays
 345       */
 346      public function get_indexes($table) {
 347          $indexes = array();
 348          $tablename = $this->prefix.$table;
 349  
 350          $sql = "SELECT i.*
 351                    FROM pg_catalog.pg_indexes i
 352                    JOIN pg_catalog.pg_namespace as ns ON ns.nspname = i.schemaname
 353                   WHERE i.tablename = '$tablename'
 354                         AND (i.schemaname = current_schema() OR ns.oid = pg_my_temp_schema())";
 355  
 356          $this->query_start($sql, null, SQL_QUERY_AUX);
 357          $result = pg_query($this->pgsql, $sql);
 358          $this->query_end($result);
 359  
 360          if ($result) {
 361              while ($row = pg_fetch_assoc($result)) {
 362                  if (!preg_match('/CREATE (|UNIQUE )INDEX ([^\s]+) ON '.$tablename.' USING ([^\s]+) \(([^\)]+)\)/i', $row['indexdef'], $matches)) {
 363                      continue;
 364                  }
 365                  if ($matches[4] === 'id') {
 366                      continue;
 367                  }
 368                  $columns = explode(',', $matches[4]);
 369                  foreach ($columns as $k=>$column) {
 370                      $column = trim($column);
 371                      if ($pos = strpos($column, ' ')) {
 372                          // index type is separated by space
 373                          $column = substr($column, 0, $pos);
 374                      }
 375                      $columns[$k] = $this->trim_quotes($column);
 376                  }
 377                  $indexes[$row['indexname']] = array('unique'=>!empty($matches[1]),
 378                                                'columns'=>$columns);
 379              }
 380              pg_free_result($result);
 381          }
 382          return $indexes;
 383      }
 384  
 385      /**
 386       * Returns detailed information about columns in table. This information is cached internally.
 387       * @param string $table name
 388       * @param bool $usecache
 389       * @return database_column_info[] array of database_column_info objects indexed with column names
 390       */
 391      public function get_columns($table, $usecache=true) {
 392          if ($usecache) {
 393              $properties = array('dbfamily' => $this->get_dbfamily(), 'settings' => $this->get_settings_hash());
 394              $cache = cache::make('core', 'databasemeta', $properties);
 395              if ($data = $cache->get($table)) {
 396                  return $data;
 397              }
 398          }
 399  
 400          $structure = array();
 401  
 402          $tablename = $this->prefix.$table;
 403  
 404          $sql = "SELECT a.attnum, a.attname AS field, t.typname AS type, a.attlen, a.atttypmod, a.attnotnull, a.atthasdef, d.adsrc
 405                    FROM pg_catalog.pg_class c
 406                    JOIN pg_catalog.pg_namespace as ns ON ns.oid = c.relnamespace
 407                    JOIN pg_catalog.pg_attribute a ON a.attrelid = c.oid
 408                    JOIN pg_catalog.pg_type t ON t.oid = a.atttypid
 409               LEFT JOIN pg_catalog.pg_attrdef d ON (d.adrelid = c.oid AND d.adnum = a.attnum)
 410                   WHERE relkind = 'r' AND c.relname = '$tablename' AND c.reltype > 0 AND a.attnum > 0
 411                         AND (ns.nspname = current_schema() OR ns.oid = pg_my_temp_schema())
 412                ORDER BY a.attnum";
 413  
 414          $this->query_start($sql, null, SQL_QUERY_AUX);
 415          $result = pg_query($this->pgsql, $sql);
 416          $this->query_end($result);
 417  
 418          if (!$result) {
 419              return array();
 420          }
 421          while ($rawcolumn = pg_fetch_object($result)) {
 422  
 423              $info = new stdClass();
 424              $info->name = $rawcolumn->field;
 425              $matches = null;
 426  
 427              if ($rawcolumn->type === 'varchar') {
 428                  $info->type          = 'varchar';
 429                  $info->meta_type     = 'C';
 430                  $info->max_length    = $rawcolumn->atttypmod - 4;
 431                  $info->scale         = null;
 432                  $info->not_null      = ($rawcolumn->attnotnull === 't');
 433                  $info->has_default   = ($rawcolumn->atthasdef === 't');
 434                  if ($info->has_default) {
 435                      $parts = explode('::', $rawcolumn->adsrc);
 436                      if (count($parts) > 1) {
 437                          $info->default_value = reset($parts);
 438                          $info->default_value = trim($info->default_value, "'");
 439                      } else {
 440                          $info->default_value = $rawcolumn->adsrc;
 441                      }
 442                  } else {
 443                      $info->default_value = null;
 444                  }
 445                  $info->primary_key   = false;
 446                  $info->binary        = false;
 447                  $info->unsigned      = null;
 448                  $info->auto_increment= false;
 449                  $info->unique        = null;
 450  
 451              } else if (preg_match('/int(\d)/i', $rawcolumn->type, $matches)) {
 452                  $info->type = 'int';
 453                  if (strpos($rawcolumn->adsrc, 'nextval') === 0) {
 454                      $info->primary_key   = true;
 455                      $info->meta_type     = 'R';
 456                      $info->unique        = true;
 457                      $info->auto_increment= true;
 458                      $info->has_default   = false;
 459                  } else {
 460                      $info->primary_key   = false;
 461                      $info->meta_type     = 'I';
 462                      $info->unique        = null;
 463                      $info->auto_increment= false;
 464                      $info->has_default   = ($rawcolumn->atthasdef === 't');
 465                  }
 466                  // Return number of decimals, not bytes here.
 467                  if ($matches[1] >= 8) {
 468                      $info->max_length = 18;
 469                  } else if ($matches[1] >= 4) {
 470                      $info->max_length = 9;
 471                  } else if ($matches[1] >= 2) {
 472                      $info->max_length = 4;
 473                  } else if ($matches[1] >= 1) {
 474                      $info->max_length = 2;
 475                  } else {
 476                      $info->max_length = 0;
 477                  }
 478                  $info->scale         = null;
 479                  $info->not_null      = ($rawcolumn->attnotnull === 't');
 480                  if ($info->has_default) {
 481                      $info->default_value = trim($rawcolumn->adsrc, '()');
 482                  } else {
 483                      $info->default_value = null;
 484                  }
 485                  $info->binary        = false;
 486                  $info->unsigned      = false;
 487  
 488              } else if ($rawcolumn->type === 'numeric') {
 489                  $info->type = $rawcolumn->type;
 490                  $info->meta_type     = 'N';
 491                  $info->primary_key   = false;
 492                  $info->binary        = false;
 493                  $info->unsigned      = null;
 494                  $info->auto_increment= false;
 495                  $info->unique        = null;
 496                  $info->not_null      = ($rawcolumn->attnotnull === 't');
 497                  $info->has_default   = ($rawcolumn->atthasdef === 't');
 498                  if ($info->has_default) {
 499                      $info->default_value = trim($rawcolumn->adsrc, '()');
 500                  } else {
 501                      $info->default_value = null;
 502                  }
 503                  $info->max_length    = $rawcolumn->atttypmod >> 16;
 504                  $info->scale         = ($rawcolumn->atttypmod & 0xFFFF) - 4;
 505  
 506              } else if (preg_match('/float(\d)/i', $rawcolumn->type, $matches)) {
 507                  $info->type = 'float';
 508                  $info->meta_type     = 'N';
 509                  $info->primary_key   = false;
 510                  $info->binary        = false;
 511                  $info->unsigned      = null;
 512                  $info->auto_increment= false;
 513                  $info->unique        = null;
 514                  $info->not_null      = ($rawcolumn->attnotnull === 't');
 515                  $info->has_default   = ($rawcolumn->atthasdef === 't');
 516                  if ($info->has_default) {
 517                      $info->default_value = trim($rawcolumn->adsrc, '()');
 518                  } else {
 519                      $info->default_value = null;
 520                  }
 521                  // just guess expected number of deciaml places :-(
 522                  if ($matches[1] == 8) {
 523                      // total 15 digits
 524                      $info->max_length = 8;
 525                      $info->scale      = 7;
 526                  } else {
 527                      // total 6 digits
 528                      $info->max_length = 4;
 529                      $info->scale      = 2;
 530                  }
 531  
 532              } else if ($rawcolumn->type === 'text') {
 533                  $info->type          = $rawcolumn->type;
 534                  $info->meta_type     = 'X';
 535                  $info->max_length    = -1;
 536                  $info->scale         = null;
 537                  $info->not_null      = ($rawcolumn->attnotnull === 't');
 538                  $info->has_default   = ($rawcolumn->atthasdef === 't');
 539                  if ($info->has_default) {
 540                      $parts = explode('::', $rawcolumn->adsrc);
 541                      if (count($parts) > 1) {
 542                          $info->default_value = reset($parts);
 543                          $info->default_value = trim($info->default_value, "'");
 544                      } else {
 545                          $info->default_value = $rawcolumn->adsrc;
 546                      }
 547                  } else {
 548                      $info->default_value = null;
 549                  }
 550                  $info->primary_key   = false;
 551                  $info->binary        = false;
 552                  $info->unsigned      = null;
 553                  $info->auto_increment= false;
 554                  $info->unique        = null;
 555  
 556              } else if ($rawcolumn->type === 'bytea') {
 557                  $info->type          = $rawcolumn->type;
 558                  $info->meta_type     = 'B';
 559                  $info->max_length    = -1;
 560                  $info->scale         = null;
 561                  $info->not_null      = ($rawcolumn->attnotnull === 't');
 562                  $info->has_default   = false;
 563                  $info->default_value = null;
 564                  $info->primary_key   = false;
 565                  $info->binary        = true;
 566                  $info->unsigned      = null;
 567                  $info->auto_increment= false;
 568                  $info->unique        = null;
 569  
 570              }
 571  
 572              $structure[$info->name] = new database_column_info($info);
 573          }
 574  
 575          pg_free_result($result);
 576  
 577          if ($usecache) {
 578              $cache->set($table, $structure);
 579          }
 580  
 581          return $structure;
 582      }
 583  
 584      /**
 585       * Normalise values based in RDBMS dependencies (booleans, LOBs...)
 586       *
 587       * @param database_column_info $column column metadata corresponding with the value we are going to normalise
 588       * @param mixed $value value we are going to normalise
 589       * @return mixed the normalised value
 590       */
 591      protected function normalise_value($column, $value) {
 592          $this->detect_objects($value);
 593  
 594          if (is_bool($value)) { // Always, convert boolean to int
 595              $value = (int)$value;
 596  
 597          } else if ($column->meta_type === 'B') { // BLOB detected, we return 'blob' array instead of raw value to allow
 598              if (!is_null($value)) {             // binding/executing code later to know about its nature
 599                  $value = array('blob' => $value);
 600              }
 601  
 602          } else if ($value === '') {
 603              if ($column->meta_type === 'I' or $column->meta_type === 'F' or $column->meta_type === 'N') {
 604                  $value = 0; // prevent '' problems in numeric fields
 605              }
 606          }
 607          return $value;
 608      }
 609  
 610      /**
 611       * Is db in unicode mode?
 612       * @return bool
 613       */
 614      public function setup_is_unicodedb() {
 615          // Get PostgreSQL server_encoding value
 616          $sql = "SHOW server_encoding";
 617          $this->query_start($sql, null, SQL_QUERY_AUX);
 618          $result = pg_query($this->pgsql, $sql);
 619          $this->query_end($result);
 620  
 621          if (!$result) {
 622              return false;
 623          }
 624          $rawcolumn = pg_fetch_object($result);
 625          $encoding = $rawcolumn->server_encoding;
 626          pg_free_result($result);
 627  
 628          return (strtoupper($encoding) == 'UNICODE' || strtoupper($encoding) == 'UTF8');
 629      }
 630  
 631      /**
 632       * Do NOT use in code, to be used by database_manager only!
 633       * @param string|array $sql query
 634       * @return bool true
 635       * @throws ddl_change_structure_exception A DDL specific exception is thrown for any errors.
 636       */
 637      public function change_database_structure($sql) {
 638          $this->get_manager(); // Includes DDL exceptions classes ;-)
 639          if (is_array($sql)) {
 640              $sql = implode("\n;\n", $sql);
 641          }
 642          if (!$this->is_transaction_started()) {
 643              // It is better to do all or nothing, this helps with recovery...
 644              $sql = "BEGIN ISOLATION LEVEL SERIALIZABLE;\n$sql\n; COMMIT";
 645          }
 646  
 647          try {
 648              $this->query_start($sql, null, SQL_QUERY_STRUCTURE);
 649              $result = pg_query($this->pgsql, $sql);
 650              $this->query_end($result);
 651              pg_free_result($result);
 652          } catch (ddl_change_structure_exception $e) {
 653              if (!$this->is_transaction_started()) {
 654                  $result = @pg_query($this->pgsql, "ROLLBACK");
 655                  @pg_free_result($result);
 656              }
 657              $this->reset_caches();
 658              throw $e;
 659          }
 660  
 661          $this->reset_caches();
 662          return true;
 663      }
 664  
 665      /**
 666       * Execute general sql query. Should be used only when no other method suitable.
 667       * Do NOT use this to make changes in db structure, use database_manager methods instead!
 668       * @param string $sql query
 669       * @param array $params query parameters
 670       * @return bool true
 671       * @throws dml_exception A DML specific exception is thrown for any errors.
 672       */
 673      public function execute($sql, array $params=null) {
 674          list($sql, $params, $type) = $this->fix_sql_params($sql, $params);
 675  
 676          if (strpos($sql, ';') !== false) {
 677              throw new coding_exception('moodle_database::execute() Multiple sql statements found or bound parameters not used properly in query!');
 678          }
 679  
 680          $this->query_start($sql, $params, SQL_QUERY_UPDATE);
 681          $result = pg_query_params($this->pgsql, $sql, $params);
 682          $this->query_end($result);
 683  
 684          pg_free_result($result);
 685          return true;
 686      }
 687  
 688      /**
 689       * Get a number of records as a moodle_recordset using a SQL statement.
 690       *
 691       * Since this method is a little less readable, use of it should be restricted to
 692       * code where it's possible there might be large datasets being returned.  For known
 693       * small datasets use get_records_sql - it leads to simpler code.
 694       *
 695       * The return type is like:
 696       * @see function get_recordset.
 697       *
 698       * @param string $sql the SQL select query to execute.
 699       * @param array $params array of sql parameters
 700       * @param int $limitfrom return a subset of records, starting at this point (optional, required if $limitnum is set).
 701       * @param int $limitnum return a subset comprising this many records (optional, required if $limitfrom is set).
 702       * @return moodle_recordset instance
 703       * @throws dml_exception A DML specific exception is thrown for any errors.
 704       */
 705      public function get_recordset_sql($sql, array $params=null, $limitfrom=0, $limitnum=0) {
 706  
 707          list($limitfrom, $limitnum) = $this->normalise_limit_from_num($limitfrom, $limitnum);
 708  
 709          if ($limitfrom or $limitnum) {
 710              if ($limitnum < 1) {
 711                  $limitnum = "ALL";
 712              } else if (PHP_INT_MAX - $limitnum < $limitfrom) {
 713                  // this is a workaround for weird max int problem
 714                  $limitnum = "ALL";
 715              }
 716              $sql .= " LIMIT $limitnum OFFSET $limitfrom";
 717          }
 718  
 719          list($sql, $params, $type) = $this->fix_sql_params($sql, $params);
 720  
 721          $this->query_start($sql, $params, SQL_QUERY_SELECT);
 722          $result = pg_query_params($this->pgsql, $sql, $params);
 723          $this->query_end($result);
 724  
 725          return $this->create_recordset($result);
 726      }
 727  
 728      protected function create_recordset($result) {
 729          return new pgsql_native_moodle_recordset($result, $this->bytea_oid);
 730      }
 731  
 732      /**
 733       * Get a number of records as an array of objects using a SQL statement.
 734       *
 735       * Return value is like:
 736       * @see function get_records.
 737       *
 738       * @param string $sql the SQL select query to execute. The first column of this SELECT statement
 739       *   must be a unique value (usually the 'id' field), as it will be used as the key of the
 740       *   returned array.
 741       * @param array $params array of sql parameters
 742       * @param int $limitfrom return a subset of records, starting at this point (optional, required if $limitnum is set).
 743       * @param int $limitnum return a subset comprising this many records (optional, required if $limitfrom is set).
 744       * @return array of objects, or empty array if no records were found
 745       * @throws dml_exception A DML specific exception is thrown for any errors.
 746       */
 747      public function get_records_sql($sql, array $params=null, $limitfrom=0, $limitnum=0) {
 748  
 749          list($limitfrom, $limitnum) = $this->normalise_limit_from_num($limitfrom, $limitnum);
 750  
 751          if ($limitfrom or $limitnum) {
 752              if ($limitnum < 1) {
 753                  $limitnum = "ALL";
 754              } else if (PHP_INT_MAX - $limitnum < $limitfrom) {
 755                  // this is a workaround for weird max int problem
 756                  $limitnum = "ALL";
 757              }
 758              $sql .= " LIMIT $limitnum OFFSET $limitfrom";
 759          }
 760  
 761          list($sql, $params, $type) = $this->fix_sql_params($sql, $params);
 762          $this->query_start($sql, $params, SQL_QUERY_SELECT);
 763          $result = pg_query_params($this->pgsql, $sql, $params);
 764          $this->query_end($result);
 765  
 766          // find out if there are any blobs
 767          $numrows = pg_num_fields($result);
 768          $blobs = array();
 769          for($i=0; $i<$numrows; $i++) {
 770              $type_oid = pg_field_type_oid($result, $i);
 771              if ($type_oid == $this->bytea_oid) {
 772                  $blobs[] = pg_field_name($result, $i);
 773              }
 774          }
 775  
 776          $rows = pg_fetch_all($result);
 777          pg_free_result($result);
 778  
 779          $return = array();
 780          if ($rows) {
 781              foreach ($rows as $row) {
 782                  $id = reset($row);
 783                  if ($blobs) {
 784                      foreach ($blobs as $blob) {
 785                          // 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
 786                          $row[$blob] = $row[$blob] !== null ? pg_unescape_bytea($row[$blob]) : null;
 787                      }
 788                  }
 789                  if (isset($return[$id])) {
 790                      $colname = key($row);
 791                      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);
 792                  }
 793                  $return[$id] = (object)$row;
 794              }
 795          }
 796  
 797          return $return;
 798      }
 799  
 800      /**
 801       * Selects records and return values (first field) as an array using a SQL statement.
 802       *
 803       * @param string $sql The SQL query
 804       * @param array $params array of sql parameters
 805       * @return array of values
 806       * @throws dml_exception A DML specific exception is thrown for any errors.
 807       */
 808      public function get_fieldset_sql($sql, array $params=null) {
 809          list($sql, $params, $type) = $this->fix_sql_params($sql, $params);
 810  
 811          $this->query_start($sql, $params, SQL_QUERY_SELECT);
 812          $result = pg_query_params($this->pgsql, $sql, $params);
 813          $this->query_end($result);
 814  
 815          $return = pg_fetch_all_columns($result, 0);
 816          pg_free_result($result);
 817  
 818          return $return;
 819      }
 820  
 821      /**
 822       * Insert new record into database, as fast as possible, no safety checks, lobs not supported.
 823       * @param string $table name
 824       * @param mixed $params data record as object or array
 825       * @param bool $returnit return it of inserted record
 826       * @param bool $bulk true means repeated inserts expected
 827       * @param bool $customsequence true if 'id' included in $params, disables $returnid
 828       * @return bool|int true or new id
 829       * @throws dml_exception A DML specific exception is thrown for any errors.
 830       */
 831      public function insert_record_raw($table, $params, $returnid=true, $bulk=false, $customsequence=false) {
 832          if (!is_array($params)) {
 833              $params = (array)$params;
 834          }
 835  
 836          $returning = "";
 837  
 838          if ($customsequence) {
 839              if (!isset($params['id'])) {
 840                  throw new coding_exception('moodle_database::insert_record_raw() id field must be specified if custom sequences used.');
 841              }
 842              $returnid = false;
 843          } else {
 844              if ($returnid) {
 845                  $returning = "RETURNING id";
 846                  unset($params['id']);
 847              } else {
 848                  unset($params['id']);
 849              }
 850          }
 851  
 852          if (empty($params)) {
 853              throw new coding_exception('moodle_database::insert_record_raw() no fields found.');
 854          }
 855  
 856          $fields = implode(',', array_keys($params));
 857          $values = array();
 858          $i = 1;
 859          foreach ($params as $value) {
 860              $this->detect_objects($value);
 861              $values[] = "\$".$i++;
 862          }
 863          $values = implode(',', $values);
 864  
 865          $sql = "INSERT INTO {$this->prefix}$table ($fields) VALUES($values) $returning";
 866          $this->query_start($sql, $params, SQL_QUERY_INSERT);
 867          $result = pg_query_params($this->pgsql, $sql, $params);
 868          $this->query_end($result);
 869  
 870          if ($returning !== "") {
 871              $row = pg_fetch_assoc($result);
 872              $params['id'] = reset($row);
 873          }
 874          pg_free_result($result);
 875  
 876          if (!$returnid) {
 877              return true;
 878          }
 879  
 880          return (int)$params['id'];
 881      }
 882  
 883      /**
 884       * Insert a record into a table and return the "id" field if required.
 885       *
 886       * Some conversions and safety checks are carried out. Lobs are supported.
 887       * If the return ID isn't required, then this just reports success as true/false.
 888       * $data is an object containing needed data
 889       * @param string $table The database table to be inserted into
 890       * @param object $data A data object with values for one or more fields in the record
 891       * @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.
 892       * @return bool|int true or new id
 893       * @throws dml_exception A DML specific exception is thrown for any errors.
 894       */
 895      public function insert_record($table, $dataobject, $returnid=true, $bulk=false) {
 896          $dataobject = (array)$dataobject;
 897  
 898          $columns = $this->get_columns($table);
 899          if (empty($columns)) {
 900              throw new dml_exception('ddltablenotexist', $table);
 901          }
 902  
 903          $cleaned = array();
 904          $blobs   = array();
 905  
 906          foreach ($dataobject as $field=>$value) {
 907              if ($field === 'id') {
 908                  continue;
 909              }
 910              if (!isset($columns[$field])) {
 911                  continue;
 912              }
 913              $column = $columns[$field];
 914              $normalised_value = $this->normalise_value($column, $value);
 915              if (is_array($normalised_value) && array_key_exists('blob', $normalised_value)) {
 916                  $cleaned[$field] = '@#BLOB#@';
 917                  $blobs[$field] = $normalised_value['blob'];
 918              } else {
 919                  $cleaned[$field] = $normalised_value;
 920              }
 921          }
 922  
 923          if (empty($blobs)) {
 924              return $this->insert_record_raw($table, $cleaned, $returnid, $bulk);
 925          }
 926  
 927          $id = $this->insert_record_raw($table, $cleaned, true, $bulk);
 928  
 929          foreach ($blobs as $key=>$value) {
 930              $value = pg_escape_bytea($this->pgsql, $value);
 931              $sql = "UPDATE {$this->prefix}$table SET $key = '$value'::bytea WHERE id = $id";
 932              $this->query_start($sql, NULL, SQL_QUERY_UPDATE);
 933              $result = pg_query($this->pgsql, $sql);
 934              $this->query_end($result);
 935              if ($result !== false) {
 936                  pg_free_result($result);
 937              }
 938          }
 939  
 940          return ($returnid ? $id : true);
 941  
 942      }
 943  
 944      /**
 945       * Insert multiple records into database as fast as possible.
 946       *
 947       * Order of inserts is maintained, but the operation is not atomic,
 948       * use transactions if necessary.
 949       *
 950       * This method is intended for inserting of large number of small objects,
 951       * do not use for huge objects with text or binary fields.
 952       *
 953       * @since Moodle 2.7
 954       *
 955       * @param string $table  The database table to be inserted into
 956       * @param array|Traversable $dataobjects list of objects to be inserted, must be compatible with foreach
 957       * @return void does not return new record ids
 958       *
 959       * @throws coding_exception if data objects have different structure
 960       * @throws dml_exception A DML specific exception is thrown for any errors.
 961       */
 962      public function insert_records($table, $dataobjects) {
 963          if (!is_array($dataobjects) and !($dataobjects instanceof Traversable)) {
 964              throw new coding_exception('insert_records() passed non-traversable object');
 965          }
 966  
 967          // PostgreSQL does not seem to have problems with huge queries.
 968          $chunksize = 500;
 969          if (!empty($this->dboptions['bulkinsertsize'])) {
 970              $chunksize = (int)$this->dboptions['bulkinsertsize'];
 971          }
 972  
 973          $columns = $this->get_columns($table, true);
 974  
 975          // Make sure there are no nasty blobs!
 976          foreach ($columns as $column) {
 977              if ($column->binary) {
 978                  parent::insert_records($table, $dataobjects);
 979                  return;
 980              }
 981          }
 982  
 983          $fields = null;
 984          $count = 0;
 985          $chunk = array();
 986          foreach ($dataobjects as $dataobject) {
 987              if (!is_array($dataobject) and !is_object($dataobject)) {
 988                  throw new coding_exception('insert_records() passed invalid record object');
 989              }
 990              $dataobject = (array)$dataobject;
 991              if ($fields === null) {
 992                  $fields = array_keys($dataobject);
 993                  $columns = array_intersect_key($columns, $dataobject);
 994                  unset($columns['id']);
 995              } else if ($fields !== array_keys($dataobject)) {
 996                  throw new coding_exception('All dataobjects in insert_records() must have the same structure!');
 997              }
 998  
 999              $count++;
1000              $chunk[] = $dataobject;
1001  
1002              if ($count === $chunksize) {
1003                  $this->insert_chunk($table, $chunk, $columns);
1004                  $chunk = array();
1005                  $count = 0;
1006              }
1007          }
1008  
1009          if ($count) {
1010              $this->insert_chunk($table, $chunk, $columns);
1011          }
1012      }
1013  
1014      /**
1015       * Insert records in chunks, no binary support, strict param types...
1016       *
1017       * Note: can be used only from insert_records().
1018       *
1019       * @param string $table
1020       * @param array $chunk
1021       * @param database_column_info[] $columns
1022       */
1023      protected function insert_chunk($table, array $chunk, array $columns) {
1024          $i = 1;
1025          $params = array();
1026          $values = array();
1027          foreach ($chunk as $dataobject) {
1028              $vals = array();
1029              foreach ($columns as $field => $column) {
1030                  $params[] = $this->normalise_value($column, $dataobject[$field]);
1031                  $vals[] = "\$".$i++;
1032              }
1033              $values[] = '('.implode(',', $vals).')';
1034          }
1035  
1036          $fieldssql = '('.implode(',', array_keys($columns)).')';
1037          $valuessql = implode(',', $values);
1038  
1039          $sql = "INSERT INTO {$this->prefix}$table $fieldssql VALUES $valuessql";
1040          $this->query_start($sql, $params, SQL_QUERY_INSERT);
1041          $result = pg_query_params($this->pgsql, $sql, $params);
1042          $this->query_end($result);
1043          pg_free_result($result);
1044      }
1045  
1046      /**
1047       * Import a record into a table, id field is required.
1048       * Safety checks are NOT carried out. Lobs are supported.
1049       *
1050       * @param string $table name of database table to be inserted into
1051       * @param object $dataobject A data object with values for one or more fields in the record
1052       * @return bool true
1053       * @throws dml_exception A DML specific exception is thrown for any errors.
1054       */
1055      public function import_record($table, $dataobject) {
1056          $dataobject = (array)$dataobject;
1057  
1058          $columns = $this->get_columns($table);
1059          $cleaned = array();
1060          $blobs   = array();
1061  
1062          foreach ($dataobject as $field=>$value) {
1063              $this->detect_objects($value);
1064              if (!isset($columns[$field])) {
1065                  continue;
1066              }
1067              if ($columns[$field]->meta_type === 'B') {
1068                  if (!is_null($value)) {
1069                      $cleaned[$field] = '@#BLOB#@';
1070                      $blobs[$field] = $value;
1071                      continue;
1072                  }
1073              }
1074  
1075              $cleaned[$field] = $value;
1076          }
1077  
1078          $this->insert_record_raw($table, $cleaned, false, true, true);
1079          $id = $dataobject['id'];
1080  
1081          foreach ($blobs as $key=>$value) {
1082              $value = pg_escape_bytea($this->pgsql, $value);
1083              $sql = "UPDATE {$this->prefix}$table SET $key = '$value'::bytea WHERE id = $id";
1084              $this->query_start($sql, NULL, SQL_QUERY_UPDATE);
1085              $result = pg_query($this->pgsql, $sql);
1086              $this->query_end($result);
1087              if ($result !== false) {
1088                  pg_free_result($result);
1089              }
1090          }
1091  
1092          return true;
1093      }
1094  
1095      /**
1096       * Update record in database, as fast as possible, no safety checks, lobs not supported.
1097       * @param string $table name
1098       * @param mixed $params data record as object or array
1099       * @param bool true means repeated updates expected
1100       * @return bool true
1101       * @throws dml_exception A DML specific exception is thrown for any errors.
1102       */
1103      public function update_record_raw($table, $params, $bulk=false) {
1104          $params = (array)$params;
1105  
1106          if (!isset($params['id'])) {
1107              throw new coding_exception('moodle_database::update_record_raw() id field must be specified.');
1108          }
1109          $id = $params['id'];
1110          unset($params['id']);
1111  
1112          if (empty($params)) {
1113              throw new coding_exception('moodle_database::update_record_raw() no fields found.');
1114          }
1115  
1116          $i = 1;
1117  
1118          $sets = array();
1119          foreach ($params as $field=>$value) {
1120              $this->detect_objects($value);
1121              $sets[] = "$field = \$".$i++;
1122          }
1123  
1124          $params[] = $id; // last ? in WHERE condition
1125  
1126          $sets = implode(',', $sets);
1127          $sql = "UPDATE {$this->prefix}$table SET $sets WHERE id=\$".$i;
1128  
1129          $this->query_start($sql, $params, SQL_QUERY_UPDATE);
1130          $result = pg_query_params($this->pgsql, $sql, $params);
1131          $this->query_end($result);
1132  
1133          pg_free_result($result);
1134          return true;
1135      }
1136  
1137      /**
1138       * Update a record in a table
1139       *
1140       * $dataobject is an object containing needed data
1141       * Relies on $dataobject having a variable "id" to
1142       * specify the record to update
1143       *
1144       * @param string $table The database table to be checked against.
1145       * @param object $dataobject An object with contents equal to fieldname=>fieldvalue. Must have an entry for 'id' to map to the table specified.
1146       * @param bool true means repeated updates expected
1147       * @return bool true
1148       * @throws dml_exception A DML specific exception is thrown for any errors.
1149       */
1150      public function update_record($table, $dataobject, $bulk=false) {
1151          $dataobject = (array)$dataobject;
1152  
1153          $columns = $this->get_columns($table);
1154          $cleaned = array();
1155          $blobs   = array();
1156  
1157          foreach ($dataobject as $field=>$value) {
1158              if (!isset($columns[$field])) {
1159                  continue;
1160              }
1161              $column = $columns[$field];
1162              $normalised_value = $this->normalise_value($column, $value);
1163              if (is_array($normalised_value) && array_key_exists('blob', $normalised_value)) {
1164                  $cleaned[$field] = '@#BLOB#@';
1165                  $blobs[$field] = $normalised_value['blob'];
1166              } else {
1167                  $cleaned[$field] = $normalised_value;
1168              }
1169          }
1170  
1171          $this->update_record_raw($table, $cleaned, $bulk);
1172  
1173          if (empty($blobs)) {
1174              return true;
1175          }
1176  
1177          $id = (int)$dataobject['id'];
1178  
1179          foreach ($blobs as $key=>$value) {
1180              $value = pg_escape_bytea($this->pgsql, $value);
1181              $sql = "UPDATE {$this->prefix}$table SET $key = '$value'::bytea WHERE id = $id";
1182              $this->query_start($sql, NULL, SQL_QUERY_UPDATE);
1183              $result = pg_query($this->pgsql, $sql);
1184              $this->query_end($result);
1185  
1186              pg_free_result($result);
1187          }
1188  
1189          return true;
1190      }
1191  
1192      /**
1193       * Set a single field in every table record which match a particular WHERE clause.
1194       *
1195       * @param string $table The database table to be checked against.
1196       * @param string $newfield the field to set.
1197       * @param string $newvalue the value to set the field to.
1198       * @param string $select A fragment of SQL to be used in a where clause in the SQL call.
1199       * @param array $params array of sql parameters
1200       * @return bool true
1201       * @throws dml_exception A DML specific exception is thrown for any errors.
1202       */
1203      public function set_field_select($table, $newfield, $newvalue, $select, array $params=null) {
1204  
1205          if ($select) {
1206              $select = "WHERE $select";
1207          }
1208          if (is_null($params)) {
1209              $params = array();
1210          }
1211          list($select, $params, $type) = $this->fix_sql_params($select, $params);
1212          $i = count($params)+1;
1213  
1214          // Get column metadata
1215          $columns = $this->get_columns($table);
1216          $column = $columns[$newfield];
1217  
1218          $normalised_value = $this->normalise_value($column, $newvalue);
1219          if (is_array($normalised_value) && array_key_exists('blob', $normalised_value)) {
1220              // Update BYTEA and return
1221              $normalised_value = pg_escape_bytea($this->pgsql, $normalised_value['blob']);
1222              $sql = "UPDATE {$this->prefix}$table SET $newfield = '$normalised_value'::bytea $select";
1223              $this->query_start($sql, NULL, SQL_QUERY_UPDATE);
1224              $result = pg_query_params($this->pgsql, $sql, $params);
1225              $this->query_end($result);
1226              pg_free_result($result);
1227              return true;
1228          }
1229  
1230          if (is_null($normalised_value)) {
1231              $newfield = "$newfield = NULL";
1232          } else {
1233              $newfield = "$newfield = \$".$i;
1234              $params[] = $normalised_value;
1235          }
1236          $sql = "UPDATE {$this->prefix}$table SET $newfield $select";
1237  
1238          $this->query_start($sql, $params, SQL_QUERY_UPDATE);
1239          $result = pg_query_params($this->pgsql, $sql, $params);
1240          $this->query_end($result);
1241  
1242          pg_free_result($result);
1243  
1244          return true;
1245      }
1246  
1247      /**
1248       * Delete one or more records from a table which match a particular WHERE clause.
1249       *
1250       * @param string $table The database table to be checked against.
1251       * @param string $select A fragment of SQL to be used in a where clause in the SQL call (used to define the selection criteria).
1252       * @param array $params array of sql parameters
1253       * @return bool true
1254       * @throws dml_exception A DML specific exception is thrown for any errors.
1255       */
1256      public function delete_records_select($table, $select, array $params=null) {
1257          if ($select) {
1258              $select = "WHERE $select";
1259          }
1260          $sql = "DELETE FROM {$this->prefix}$table $select";
1261  
1262          list($sql, $params, $type) = $this->fix_sql_params($sql, $params);
1263  
1264          $this->query_start($sql, $params, SQL_QUERY_UPDATE);
1265          $result = pg_query_params($this->pgsql, $sql, $params);
1266          $this->query_end($result);
1267  
1268          pg_free_result($result);
1269  
1270          return true;
1271      }
1272  
1273      /**
1274       * Returns 'LIKE' part of a query.
1275       *
1276       * @param string $fieldname usually name of the table column
1277       * @param string $param usually bound query parameter (?, :named)
1278       * @param bool $casesensitive use case sensitive search
1279       * @param bool $accensensitive use accent sensitive search (not all databases support accent insensitive)
1280       * @param bool $notlike true means "NOT LIKE"
1281       * @param string $escapechar escape char for '%' and '_'
1282       * @return string SQL code fragment
1283       */
1284      public function sql_like($fieldname, $param, $casesensitive = true, $accentsensitive = true, $notlike = false, $escapechar = '\\') {
1285          if (strpos($param, '%') !== false) {
1286              debugging('Potential SQL injection detected, sql_like() expects bound parameters (? or :named)');
1287          }
1288          if ($escapechar === '\\') {
1289              // Prevents problems with C-style escapes of enclosing '\',
1290              // E'... bellow prevents compatibility warnings.
1291              $escapechar = '\\\\';
1292          }
1293  
1294          // postgresql does not support accent insensitive text comparisons, sorry
1295          if ($casesensitive) {
1296              $LIKE = $notlike ? 'NOT LIKE' : 'LIKE';
1297          } else {
1298              $LIKE = $notlike ? 'NOT ILIKE' : 'ILIKE';
1299          }
1300          return "$fieldname $LIKE $param ESCAPE E'$escapechar'";
1301      }
1302  
1303      public function sql_bitxor($int1, $int2) {
1304          return '((' . $int1 . ') # (' . $int2 . '))';
1305      }
1306  
1307      public function sql_cast_char2int($fieldname, $text=false) {
1308          return ' CAST(' . $fieldname . ' AS INT) ';
1309      }
1310  
1311      public function sql_cast_char2real($fieldname, $text=false) {
1312          return " $fieldname::real ";
1313      }
1314  
1315      public function sql_concat() {
1316          $arr = func_get_args();
1317          $s = implode(' || ', $arr);
1318          if ($s === '') {
1319              return " '' ";
1320          }
1321          // Add always empty string element so integer-exclusive concats
1322          // will work without needing to cast each element explicitly
1323          return " '' || $s ";
1324      }
1325  
1326      public function sql_concat_join($separator="' '", $elements=array()) {
1327          for ($n=count($elements)-1; $n > 0 ; $n--) {
1328              array_splice($elements, $n, 0, $separator);
1329          }
1330          $s = implode(' || ', $elements);
1331          if ($s === '') {
1332              return " '' ";
1333          }
1334          return " $s ";
1335      }
1336  
1337      public function sql_regex_supported() {
1338          return true;
1339      }
1340  
1341      public function sql_regex($positivematch=true) {
1342          return $positivematch ? '~*' : '!~*';
1343      }
1344  
1345      /**
1346       * Does this driver support tool_replace?
1347       *
1348       * @since Moodle 2.6.1
1349       * @return bool
1350       */
1351      public function replace_all_text_supported() {
1352          return true;
1353      }
1354  
1355      public function session_lock_supported() {
1356          return true;
1357      }
1358  
1359      /**
1360       * Obtain session lock
1361       * @param int $rowid id of the row with session record
1362       * @param int $timeout max allowed time to wait for the lock in seconds
1363       * @return bool success
1364       */
1365      public function get_session_lock($rowid, $timeout) {
1366          // NOTE: there is a potential locking problem for database running
1367          //       multiple instances of moodle, we could try to use pg_advisory_lock(int, int),
1368          //       luckily there is not a big chance that they would collide
1369          if (!$this->session_lock_supported()) {
1370              return;
1371          }
1372  
1373          parent::get_session_lock($rowid, $timeout);
1374  
1375          $timeoutmilli = $timeout * 1000;
1376  
1377          $sql = "SET statement_timeout TO $timeoutmilli";
1378          $this->query_start($sql, null, SQL_QUERY_AUX);
1379          $result = pg_query($this->pgsql, $sql);
1380          $this->query_end($result);
1381  
1382          if ($result) {
1383              pg_free_result($result);
1384          }
1385  
1386          $sql = "SELECT pg_advisory_lock($rowid)";
1387          $this->query_start($sql, null, SQL_QUERY_AUX);
1388          $start = time();
1389          $result = pg_query($this->pgsql, $sql);
1390          $end = time();
1391          try {
1392              $this->query_end($result);
1393          } catch (dml_exception $ex) {
1394              if ($end - $start >= $timeout) {
1395                  throw new dml_sessionwait_exception();
1396              } else {
1397                  throw $ex;
1398              }
1399          }
1400  
1401          if ($result) {
1402              pg_free_result($result);
1403          }
1404  
1405          $sql = "SET statement_timeout TO DEFAULT";
1406          $this->query_start($sql, null, SQL_QUERY_AUX);
1407          $result = pg_query($this->pgsql, $sql);
1408          $this->query_end($result);
1409  
1410          if ($result) {
1411              pg_free_result($result);
1412          }
1413      }
1414  
1415      public function release_session_lock($rowid) {
1416          if (!$this->session_lock_supported()) {
1417              return;
1418          }
1419          if (!$this->used_for_db_sessions) {
1420              return;
1421          }
1422  
1423          parent::release_session_lock($rowid);
1424  
1425          $sql = "SELECT pg_advisory_unlock($rowid)";
1426          $this->query_start($sql, null, SQL_QUERY_AUX);
1427          $result = pg_query($this->pgsql, $sql);
1428          $this->query_end($result);
1429  
1430          if ($result) {
1431              pg_free_result($result);
1432          }
1433      }
1434  
1435      /**
1436       * Driver specific start of real database transaction,
1437       * this can not be used directly in code.
1438       * @return void
1439       */
1440      protected function begin_transaction() {
1441          $this->savepointpresent = true;
1442          $sql = "BEGIN ISOLATION LEVEL READ COMMITTED; SAVEPOINT moodle_pg_savepoint";
1443          $this->query_start($sql, NULL, SQL_QUERY_AUX);
1444          $result = pg_query($this->pgsql, $sql);
1445          $this->query_end($result);
1446  
1447          pg_free_result($result);
1448      }
1449  
1450      /**
1451       * Driver specific commit of real database transaction,
1452       * this can not be used directly in code.
1453       * @return void
1454       */
1455      protected function commit_transaction() {
1456          $this->savepointpresent = false;
1457          $sql = "RELEASE SAVEPOINT moodle_pg_savepoint; COMMIT";
1458          $this->query_start($sql, NULL, SQL_QUERY_AUX);
1459          $result = pg_query($this->pgsql, $sql);
1460          $this->query_end($result);
1461  
1462          pg_free_result($result);
1463      }
1464  
1465      /**
1466       * Driver specific abort of real database transaction,
1467       * this can not be used directly in code.
1468       * @return void
1469       */
1470      protected function rollback_transaction() {
1471          $this->savepointpresent = false;
1472          $sql = "RELEASE SAVEPOINT moodle_pg_savepoint; ROLLBACK";
1473          $this->query_start($sql, NULL, SQL_QUERY_AUX);
1474          $result = pg_query($this->pgsql, $sql);
1475          $this->query_end($result);
1476  
1477          pg_free_result($result);
1478      }
1479  
1480      /**
1481       * Helper function trimming (whitespace + quotes) any string
1482       * needed because PG uses to enclose with double quotes some
1483       * fields in indexes definition and others
1484       *
1485       * @param string $str string to apply whitespace + quotes trim
1486       * @return string trimmed string
1487       */
1488      private function trim_quotes($str) {
1489          return trim(trim($str), "'\"");
1490      }
1491  }