MDL-14992 towards better db sessions
[moodle.git] / lib / dml / pgsql_native_moodle_database.php
CommitLineData
158622bd 1<?php //$Id$
2
3require_once($CFG->libdir.'/dml/moodle_database.php');
4require_once($CFG->libdir.'/dml/pgsql_native_moodle_recordset.php');
5
6/**
7 * Native pgsql class representing moodle database interface.
8 * @package dml
9 */
10class pgsql_native_moodle_database extends moodle_database {
11
db7aea38 12 protected $pgsql = null;
db7aea38 13 protected $bytea_oid = null;
158622bd 14
ba98912f 15 protected $last_debug;
16
158622bd 17 /**
18 * Detects if all needed PHP stuff installed.
19 * Note: can be used before connect()
20 * @return mixed true if ok, string if something
21 */
22 public function driver_installed() {
23 if (!extension_loaded('pgsql')) {
24 return get_string('pgsqlextensionisnotpresentinphp', 'install');
25 }
26 return true;
27 }
28
29 /**
30 * Returns database family type - describes SQL dialect
31 * Note: can be used before connect()
32 * @return string db family name (mysql, postgres, mssql, oracle, etc.)
33 */
34 public function get_dbfamily() {
35 return 'postgres';
36 }
37
38 /**
39 * Returns more specific database driver type
40 * Note: can be used before connect()
41 * @return string db type mysql, pgsql, postgres7
42 */
43 protected function get_dbtype() {
44 return 'pgsql';
45 }
46
47 /**
48 * Returns general database library name
49 * Note: can be used before connect()
50 * @return string db type adodb, pdo, native
51 */
52 protected function get_dblibrary() {
53 return 'native';
54 }
55
56 /**
57 * Returns localised database type name
58 * Note: can be used before connect()
59 * @return string
60 */
61 public function get_name() {
62 return get_string('nativepgsql', 'install'); // TODO: localise
63 }
64
65 /**
66 * Returns localised database description
67 * Note: can be used before connect()
68 * @return string
69 */
70 public function get_configuration_hints() {
71 return get_string('databasesettingssub_postgres7', 'install'); // TODO: improve
72 }
73
74 /**
75 * Connect to db
76 * Must be called before other methods.
77 * @param string $dbhost
78 * @param string $dbuser
79 * @param string $dbpass
80 * @param string $dbname
158622bd 81 * @param mixed $prefix string means moodle db prefix, false used for external databases where prefix not used
82 * @param array $dboptions driver specific options
ce152606 83 * @return bool true
84 * @throws dml_connection_exception if error
158622bd 85 */
beaa43db 86 public function connect($dbhost, $dbuser, $dbpass, $dbname, $prefix, array $dboptions=null) {
189f3ee9 87 if ($prefix == '' and !$this->external) {
88 //Enforce prefixes for everybody but mysql
89 throw new dml_exception('prefixcannotbeempty', $this->get_dbfamily());
90 }
91
ce152606 92 $driverstatus = $this->driver_installed();
93
94 if ($driverstatus !== true) {
95 throw new dml_exception('dbdriverproblem', $driverstatus);
96 }
97
beaa43db 98 $this->store_settings($dbhost, $dbuser, $dbpass, $dbname, $prefix, $dboptions);
158622bd 99
dd2ce459 100 $pass = addcslashes($this->dbpass, "'\\");
101
102 // Unix socket connections should have lower overhead
e6b854a0 103 if (!empty($this->dboptions['dbsocket']) and ($this->dbhost === 'localhost' or $this->dbhost === '127.0.0.1')) {
dd2ce459 104 $connection = "user='$this->dbuser' password='$pass' dbname='$this->dbname'";
105 } else {
e6b854a0 106 if (empty($this->dbname)) {
107 // probably old style socket connection - do not add port
108 $port = "";
109 } else if (empty($this->dboptions['dbport'])) {
110 $port = "port ='5432'";
d139b067 111 } else {
e6b854a0 112 $port = "port ='".$this->dboptions['dbport']."'";
d139b067 113 }
e6b854a0 114 $connection = "host='$this->dbhost' $port user='$this->dbuser' password='$pass' dbname='$this->dbname'";
dd2ce459 115 }
116
ce152606 117 ob_start();
dd2ce459 118 if (empty($this->dboptions['dbpersit'])) {
119 $this->pgsql = pg_connect($connection, PGSQL_CONNECT_FORCE_NEW);
120 } else {
121 $this->pgsql = pg_pconnect($connection, PGSQL_CONNECT_FORCE_NEW);
122 }
ce152606 123 $dberr = ob_get_contents();
124 ob_end_clean();
125
158622bd 126 $status = pg_connection_status($this->pgsql);
9a4f9e33 127
128 if ($status === false or $status === PGSQL_CONNECTION_BAD) {
158622bd 129 $this->pgsql = null;
ce152606 130 throw new dml_connection_exception($dberr);
158622bd 131 }
9a4f9e33 132
1500142b 133 $this->query_start("--pg_set_client_encoding()", null, SQL_QUERY_AUX);
158622bd 134 pg_set_client_encoding($this->pgsql, 'utf8');
1500142b 135 $this->query_end(true);
136
db7aea38 137 // find out the bytea oid
7063c802 138 $sql = "SELECT oid FROM pg_type WHERE typname = 'bytea'";
1500142b 139 $this->query_start($sql, null, SQL_QUERY_AUX);
db7aea38 140 $result = pg_query($this->pgsql, $sql);
1500142b 141 $this->query_end($result);
ce152606 142
db7aea38 143 $this->bytea_oid = pg_fetch_result($result, 0);
144 pg_free_result($result);
145 if ($this->bytea_oid === false) {
ce152606 146 $this->pgsql = null;
147 throw new dml_connection_exception('Can not read bytea type.');
db7aea38 148 }
1500142b 149
158622bd 150 return true;
151 }
152
153 /**
154 * Close database connection and release all resources
155 * and memory (especially circular memory references).
156 * Do NOT use connect() again, create a new instance if needed.
157 */
158 public function dispose() {
159 if ($this->pgsql) {
160 pg_close($this->pgsql);
161 $this->pgsql = null;
162 }
163 parent::dispose();
164 }
165
ba98912f 166
167 /**
168 * Called before each db query.
169 * @param string $sql
170 * @param array array of parameters
171 * @param int $type type of query
172 * @param mixed $extrainfo driver specific extra information
173 * @return void
174 */
175 protected function query_start($sql, array $params=null, $type, $extrainfo=null) {
176 parent::query_start($sql, $params, $type, $extrainfo);
177 // pgsql driver tents to send debug to output, we do not need that ;-)
178 $this->last_debug = error_reporting(0);
179 }
180
181 /**
182 * Called immediately after each db query.
183 * @param mixed db specific result
184 * @return void
185 */
186 protected function query_end($result) {
187 //reset original debug level
188 error_reporting($this->last_debug);
189 parent::query_end($result);
190 }
191
158622bd 192 /**
193 * Returns database server info array
194 * @return array
195 */
196 public function get_server_info() {
197 static $info;
198 if (!$info) {
1500142b 199 $this->query_start("--pg_version()", null, SQL_QUERY_AUX);
158622bd 200 $info = pg_version($this->pgsql);
1500142b 201 $this->query_end(true);
158622bd 202 }
203 return array('description'=>$info['server'], 'version'=>$info['server']);
204 }
205
206 protected function is_min_version($version) {
207 $server = $this->get_server_info();
208 $server = $server['version'];
209 return version_compare($server, $version, '>=');
210 }
211
212 /**
213 * Returns supported query parameter types
214 * @return bitmask
215 */
216 protected function allowed_param_types() {
217 return SQL_PARAMS_DOLLAR;
218 }
219
220 /**
221 * Returns last error reported by database engine.
222 */
223 public function get_last_error() {
224 return pg_last_error($this->pgsql);
225 }
226
227 /**
228 * Return tables in database WITHOUT current prefix
229 * @return array of table names in lowercase and without prefix
230 */
117679db 231 public function get_tables($usecache=true) {
232 if ($usecache and $this->tables !== null) {
233 return $this->tables;
234 }
235 $this->tables = array();
158622bd 236 $prefix = str_replace('_', '\\\\_', $this->prefix);
237 $sql = "SELECT tablename
238 FROM pg_catalog.pg_tables
239 WHERE tablename LIKE '$prefix%'";
1500142b 240 $this->query_start($sql, null, SQL_QUERY_AUX);
241 $result = pg_query($this->pgsql, $sql);
242 $this->query_end($result);
d139b067 243
1500142b 244 if ($result) {
158622bd 245 while ($row = pg_fetch_row($result)) {
246 $tablename = reset($row);
247 if (strpos($tablename, $this->prefix) !== 0) {
248 continue;
249 }
250 $tablename = substr($tablename, strlen($this->prefix));
117679db 251 $this->tables[$tablename] = $tablename;
158622bd 252 }
253 pg_free_result($result);
254 }
117679db 255 return $this->tables;
158622bd 256 }
257
258 /**
259 * Return table indexes - everything lowercased
260 * @return array of arrays
261 */
262 public function get_indexes($table) {
263 $indexes = array();
264 $tablename = $this->prefix.$table;
265
266 $sql = "SELECT *
267 FROM pg_catalog.pg_indexes
268 WHERE tablename = '$tablename'";
1500142b 269
270 $this->query_start($sql, null, SQL_QUERY_AUX);
271 $result = pg_query($this->pgsql, $sql);
272 $this->query_end($result);
273
274 if ($result) {
158622bd 275 while ($row = pg_fetch_assoc($result)) {
276 if (!preg_match('/CREATE (|UNIQUE )INDEX ([^\s]+) ON '.$tablename.' USING ([^\s]+) \(([^\)]+)\)/i', $row['indexdef'], $matches)) {
277 continue;
278 }
279 if ($matches[4] === 'id') {
280 continue;
281 }
80ffbad3 282 $columns = explode(',', $matches[4]);
283 $columns = array_map('trim', $columns);
158622bd 284 $indexes[$matches[2]] = array('unique'=>!empty($matches[1]),
80ffbad3 285 'columns'=>$columns);
158622bd 286 }
287 pg_free_result($result);
288 }
289 return $indexes;
290 }
291
292 /**
293 * Returns datailed information about columns in table. This information is cached internally.
294 * @param string $table name
295 * @param bool $usecache
296 * @return array array of database_column_info objects indexed with column names
297 */
298 public function get_columns($table, $usecache=true) {
299 if ($usecache and isset($this->columns[$table])) {
300 return $this->columns[$table];
301 }
302
303 $this->columns[$table] = array();
304
305 $tablename = $this->prefix.$table;
306
307 $sql = "SELECT a.attnum, a.attname AS field, t.typname AS type, a.attlen, a.atttypmod, a.attnotnull, a.atthasdef, d.adsrc
308 FROM pg_catalog.pg_class c
9a4f9e33 309 JOIN pg_catalog.pg_attribute a ON a.attrelid = c.oid
158622bd 310 JOIN pg_catalog.pg_type t ON t.oid = a.atttypid
311 LEFT JOIN pg_catalog.pg_attrdef d ON (d.adrelid = c.oid AND d.adnum = a.attnum)
312 WHERE relkind = 'r' AND c.relname = '$tablename' AND c.reltype > 0 AND a.attnum > 0
313 ORDER BY a.attnum";
314
1500142b 315 $this->query_start($sql, null, SQL_QUERY_AUX);
316 $result = pg_query($this->pgsql, $sql);
317 $this->query_end($result);
318
319 if (!$result) {
158622bd 320 return array();
321 }
322 while ($rawcolumn = pg_fetch_object($result)) {
323
324 $info = new object();
325 $info->name = $rawcolumn->field;
326 $matches = null;
327
328 if ($rawcolumn->type === 'varchar') {
329 //TODO add some basic enum support here
330 $info->type = 'varchar';
331 $info->meta_type = 'C';
332 $info->max_length = $rawcolumn->atttypmod - 4;
333 $info->scale = null;
298d9250 334 $info->not_null = ($rawcolumn->attnotnull === 't');
335 $info->has_default = ($rawcolumn->atthasdef === 't');
158622bd 336 if ($info->has_default) {
337 $parts = explode('::', $rawcolumn->adsrc);
338 if (count($parts) > 1) {
339 $info->default_value = reset($parts);
340 $info->default_value = trim($info->default_value, "'");
341 } else {
342 $info->default_value = $rawcolumn->adsrc;
343 }
344 } else {
345 $info->default_value = null;
346 }
347 $info->primary_key = false;
348 $info->binary = false;
349 $info->unsigned = null;
350 $info->auto_increment= false;
351 $info->unique = null;
352
353 } else if (preg_match('/int(\d)/i', $rawcolumn->type, $matches)) {
354 $info->type = 'int';
355 if (strpos($rawcolumn->adsrc, 'nextval') === 0) {
356 $info->primary_key = true;
357 $info->meta_type = 'R';
358 $info->unique = true;
359 $info->auto_increment= true;
360 $info->has_default = false;
361 } else {
362 $info->primary_key = false;
363 $info->meta_type = 'I';
364 $info->unique = null;
365 $info->auto_increment= false;
298d9250 366 $info->has_default = ($rawcolumn->atthasdef === 't');
158622bd 367 }
368 $info->max_length = $matches[1];
369 $info->scale = null;
298d9250 370 $info->not_null = ($rawcolumn->attnotnull === 't');
158622bd 371 if ($info->has_default) {
372 $info->default_value = $rawcolumn->adsrc;
373 } else {
374 $info->default_value = null;
375 }
376 $info->binary = false;
377 $info->unsigned = false;
378
379 } else if ($rawcolumn->type === 'numeric') {
380 $info->type = $rawcolumn->type;
381 $info->meta_type = 'N';
382 $info->primary_key = false;
383 $info->binary = false;
384 $info->unsigned = null;
385 $info->auto_increment= false;
386 $info->unique = null;
298d9250 387 $info->not_null = ($rawcolumn->attnotnull === 't');
388 $info->has_default = ($rawcolumn->atthasdef === 't');
158622bd 389 if ($info->has_default) {
390 $info->default_value = $rawcolumn->adsrc;
391 } else {
392 $info->default_value = null;
393 }
394 $info->max_length = $rawcolumn->atttypmod >> 16;
395 $info->scale = ($rawcolumn->atttypmod & 0xFFFF) - 4;
396
397 } else if (preg_match('/float(\d)/i', $rawcolumn->type, $matches)) {
398 $info->type = 'float';
399 $info->meta_type = 'N';
400 $info->primary_key = false;
401 $info->binary = false;
402 $info->unsigned = null;
403 $info->auto_increment= false;
404 $info->unique = null;
298d9250 405 $info->not_null = ($rawcolumn->attnotnull === 't');
406 $info->has_default = ($rawcolumn->atthasdef === 't');
158622bd 407 if ($info->has_default) {
408 $info->default_value = $rawcolumn->adsrc;
409 } else {
410 $info->default_value = null;
411 }
412 // just guess expected number of deciaml places :-(
413 if ($matches[1] == 8) {
414 // total 15 digits
415 $info->max_length = 8;
416 $info->scale = 7;
417 } else {
418 // total 6 digits
419 $info->max_length = 4;
420 $info->scale = 2;
421 }
422
423 } else if ($rawcolumn->type === 'text') {
424 $info->type = $rawcolumn->type;
425 $info->meta_type = 'X';
426 $info->max_length = -1;
427 $info->scale = null;
298d9250 428 $info->not_null = ($rawcolumn->attnotnull === 't');
429 $info->has_default = ($rawcolumn->atthasdef === 't');
158622bd 430 if ($info->has_default) {
431 $parts = explode('::', $rawcolumn->adsrc);
432 if (count($parts) > 1) {
433 $info->default_value = reset($parts);
434 $info->default_value = trim($info->default_value, "'");
435 } else {
436 $info->default_value = $rawcolumn->adsrc;
437 }
438 } else {
439 $info->default_value = null;
440 }
441 $info->primary_key = false;
442 $info->binary = false;
443 $info->unsigned = null;
444 $info->auto_increment= false;
445 $info->unique = null;
446
447 } else if ($rawcolumn->type === 'bytea') {
448 $info->type = $rawcolumn->type;
449 $info->meta_type = 'B';
450 $info->max_length = -1;
451 $info->scale = null;
298d9250 452 $info->not_null = ($rawcolumn->attnotnull === 't');
158622bd 453 $info->has_default = false;
454 $info->default_value = null;
455 $info->primary_key = false;
456 $info->binary = true;
457 $info->unsigned = null;
458 $info->auto_increment= false;
459 $info->unique = null;
460
461 }
462
463 $this->columns[$table][$info->name] = new database_column_info($info);
464 }
465
e4f9c142 466 pg_free_result($result);
467
158622bd 468 return $this->columns[$table];
469 }
470
158622bd 471 /**
472 * Is db in unicode mode?
473 * @return bool
474 */
475 public function setup_is_unicodedb() {
476 /// Get PostgreSQL server_encoding value
1500142b 477 $sql = "SHOW server_encoding";
478 $this->query_start($sql, null, SQL_QUERY_AUX);
479 $result = pg_query($this->pgsql, $sql);
480 $this->query_end($result);
481
482 if (!$result) {
158622bd 483 return false;
484 }
485 $rawcolumn = pg_fetch_object($result);
486 $encoding = $rawcolumn->server_encoding;
487 pg_free_result($result);
488
489 return (strtoupper($encoding) == 'UNICODE' || strtoupper($encoding) == 'UTF8');
490 }
491
158622bd 492 /**
493 * Do NOT use in code, to be used by database_manager only!
494 * @param string $sql query
22d77567 495 * @return bool true
496 * @throws dml_exception if error
158622bd 497 */
498 public function change_database_structure($sql) {
117679db 499 $this->reset_caches();
1500142b 500
501 $this->query_start($sql, null, SQL_QUERY_STRUCTURE);
502 $result = pg_query($this->pgsql, $sql);
503 $this->query_end($result);
504
e4f9c142 505 pg_free_result($result);
158622bd 506 return true;
507 }
508
509 /**
510 * Execute general sql query. Should be used only when no other method suitable.
511 * Do NOT use this to make changes in db structure, use database_manager::execute_sql() instead!
512 * @param string $sql query
513 * @param array $params query parameters
22d77567 514 * @return bool true
515 * @throws dml_exception if error
158622bd 516 */
517 public function execute($sql, array $params=null) {
518 list($sql, $params, $type) = $this->fix_sql_params($sql, $params);
519
520 if (strpos($sql, ';') !== false) {
22d77567 521 throw new coding_exception('moodle_database::execute() Multiple sql statements found or bound parameters not used properly in query!');
158622bd 522 }
523
1500142b 524 $this->query_start($sql, $params, SQL_QUERY_UPDATE);
158622bd 525 $result = pg_query_params($this->pgsql, $sql, $params);
1500142b 526 $this->query_end($result);
158622bd 527
e4f9c142 528 pg_free_result($result);
158622bd 529 return true;
530 }
531
532 /**
533 * Get a number of records as a moodle_recordset using a SQL statement.
534 *
535 * Since this method is a little less readable, use of it should be restricted to
536 * code where it's possible there might be large datasets being returned. For known
537 * small datasets use get_records_sql - it leads to simpler code.
538 *
539 * The return type is as for @see function get_recordset.
540 *
541 * @param string $sql the SQL select query to execute.
542 * @param array $params array of sql parameters
543 * @param int $limitfrom return a subset of records, starting at this point (optional, required if $limitnum is set).
544 * @param int $limitnum return a subset comprising this many records (optional, required if $limitfrom is set).
22d77567 545 * @return mixed an moodle_recordset object
546 * @throws dml_exception if error
158622bd 547 */
548 public function get_recordset_sql($sql, array $params=null, $limitfrom=0, $limitnum=0) {
088a4df8 549 $limitfrom = (int)$limitfrom;
550 $limitnum = (int)$limitnum;
3ff8bf26 551 $limitfrom = ($limitfrom < 0) ? 0 : $limitfrom;
552 $limitnum = ($limitnum < 0) ? 0 : $limitnum;
158622bd 553 if ($limitfrom or $limitnum) {
158622bd 554 if ($limitnum < 1) {
555 $limitnum = "18446744073709551615";
556 }
557 $sql .= " LIMIT $limitnum OFFSET $limitfrom";
558 }
559
560 list($sql, $params, $type) = $this->fix_sql_params($sql, $params);
561
1500142b 562 $this->query_start($sql, $params, SQL_QUERY_SELECT);
158622bd 563 $result = pg_query_params($this->pgsql, $sql, $params);
1500142b 564 $this->query_end($result);
158622bd 565
158622bd 566 return $this->create_recordset($result);
567 }
568
569 protected function create_recordset($result) {
db7aea38 570 return new pgsql_native_moodle_recordset($result, $this->bytea_oid);
158622bd 571 }
572
573 /**
574 * Get a number of records as an array of objects using a SQL statement.
575 *
576 * Return value as for @see function get_records.
577 *
578 * @param string $sql the SQL select query to execute. The first column of this SELECT statement
579 * must be a unique value (usually the 'id' field), as it will be used as the key of the
580 * returned array.
581 * @param array $params array of sql parameters
582 * @param int $limitfrom return a subset of records, starting at this point (optional, required if $limitnum is set).
583 * @param int $limitnum return a subset comprising this many records (optional, required if $limitfrom is set).
22d77567 584 * @return mixed an array of objects, or empty array if no records were found
585 * @throws dml_exception if error
158622bd 586 */
587 public function get_records_sql($sql, array $params=null, $limitfrom=0, $limitnum=0) {
088a4df8 588 $limitfrom = (int)$limitfrom;
589 $limitnum = (int)$limitnum;
3ff8bf26 590 $limitfrom = ($limitfrom < 0) ? 0 : $limitfrom;
591 $limitnum = ($limitnum < 0) ? 0 : $limitnum;
158622bd 592 if ($limitfrom or $limitnum) {
158622bd 593 if ($limitnum < 1) {
594 $limitnum = "18446744073709551615";
595 }
596 $sql .= " LIMIT $limitnum OFFSET $limitfrom";
597 }
598
599 list($sql, $params, $type) = $this->fix_sql_params($sql, $params);
1500142b 600 $this->query_start($sql, $params, SQL_QUERY_SELECT);
158622bd 601 $result = pg_query_params($this->pgsql, $sql, $params);
1500142b 602 $this->query_end($result);
158622bd 603
db7aea38 604 // find out if there are any blobs
605 $numrows = pg_num_fields($result);
606 $blobs = array();
607 for($i=0; $i<$numrows; $i++) {
608 $type_oid = pg_field_type_oid($result, $i);
609 if ($type_oid == $this->bytea_oid) {
610 $blobs[] = pg_field_name($result, $i);
611 }
612 }
158622bd 613
614 $rows = pg_fetch_all($result);
615 pg_free_result($result);
616
617 $return = array();
618 if ($rows) {
619 foreach ($rows as $row) {
620 $id = reset($row);
db7aea38 621 if ($blobs) {
622 foreach ($blobs as $blob) {
623 $row[$blob] = pg_unescape_bytea($row[$blob]);
624 }
625 }
758ba89a 626 if (isset($return[$id])) {
627 $colname = key($row);
628 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);
629 }
158622bd 630 $return[$id] = (object)$row;
631 }
632 }
db7aea38 633
158622bd 634 return $return;
635 }
636
637 /**
638 * Selects records and return values (first field) as an array using a SQL statement.
639 *
640 * @param string $sql The SQL query
641 * @param array $params array of sql parameters
22d77567 642 * @return mixed array of values
643 * @throws dml_exception if error
158622bd 644 */
645 public function get_fieldset_sql($sql, array $params=null) {
646 list($sql, $params, $type) = $this->fix_sql_params($sql, $params);
647
1500142b 648 $this->query_start($sql, $params, SQL_QUERY_SELECT);
158622bd 649 $result = pg_query_params($this->pgsql, $sql, $params);
1500142b 650 $this->query_end($result);
158622bd 651
158622bd 652 $return = pg_fetch_all_columns($result, 0);
653 pg_free_result($result);
9a4f9e33 654
158622bd 655 return $return;
656 }
657
658 /**
659 * Insert new record into database, as fast as possible, no safety checks, lobs not supported.
660 * @param string $table name
661 * @param mixed $params data record as object or array
662 * @param bool $returnit return it of inserted record
663 * @param bool $bulk true means repeated inserts expected
664 * @param bool $customsequence true if 'id' included in $params, disables $returnid
22d77567 665 * @return true or new id
666 * @throws dml_exception if error
158622bd 667 */
668 public function insert_record_raw($table, $params, $returnid=true, $bulk=false, $customsequence=false) {
669 if (!is_array($params)) {
670 $params = (array)$params;
671 }
672
673 $returning = "";
674
675 if ($customsequence) {
676 if (!isset($params['id'])) {
22d77567 677 throw new coding_exception('moodle_database::insert_record_raw() id field must be specified if custom sequences used.');
158622bd 678 }
679 $returnid = false;
680 } else {
681 if ($returnid) {
682 if ($this->is_min_version('8.2.0')) {
683 $returning = "RETURNING id";
684 unset($params['id']);
685 } else {
686 //ugly workaround for pg < 8.2
d55132a7 687 $seqsql = "SELECT NEXTVAL('{$this->prefix}{$table}_id_seq') AS id";
1500142b 688 $this->query_start($seqsql, NULL, SQL_QUERY_AUX);
158622bd 689 $result = pg_query($this->pgsql, $seqsql);
1500142b 690 $this->query_end($result);
158622bd 691 if ($result === false) {
692 throw new dml_exception('missingidsequence', "{$this->prefix}{$table}"); // TODO: add localised string
693 }
694 $row = pg_fetch_assoc($result);
695 $params['id'] = reset($row);
696 pg_free_result($result);
697 }
698 } else {
699 unset($params['id']);
700 }
701 }
702
703 if (empty($params)) {
22d77567 704 throw new coding_exception('moodle_database::insert_record_raw() no fields found.');
158622bd 705 }
706
707 $fields = implode(',', array_keys($params));
708 $values = array();
709 $count = count($params);
710 for ($i=1; $i<=$count; $i++) {
711 $values[] = "\$".$i;
9a4f9e33 712 }
158622bd 713 $values = implode(',', $values);
714
715 $sql = "INSERT INTO {$this->prefix}$table ($fields) VALUES($values) $returning";
1500142b 716 $this->query_start($sql, $params, SQL_QUERY_INSERT);
158622bd 717 $result = pg_query_params($this->pgsql, $sql, $params);
1500142b 718 $this->query_end($result);
158622bd 719
158622bd 720 if ($returning !== "") {
721 $row = pg_fetch_assoc($result);
722 $params['id'] = reset($row);
158622bd 723 }
e4f9c142 724 pg_free_result($result);
158622bd 725
726 if (!$returnid) {
727 return true;
728 }
729
730 return (int)$params['id'];
731 }
732
733 /**
734 * Insert a record into a table and return the "id" field if required.
735 *
736 * Some conversions and safety checks are carried out. Lobs are supported.
737 * If the return ID isn't required, then this just reports success as true/false.
738 * $data is an object containing needed data
739 * @param string $table The database table to be inserted into
740 * @param object $data A data object with values for one or more fields in the record
741 * @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.
22d77567 742 * @return true or new id
743 * @throws dml_exception if error
158622bd 744 */
745 public function insert_record($table, $dataobject, $returnid=true, $bulk=false) {
746 if (!is_object($dataobject)) {
747 $dataobject = (object)$dataobject;
748 }
749
750 $columns = $this->get_columns($table);
751
752 unset($dataobject->id);
753 $cleaned = array();
754 $blobs = array();
755
756 foreach ($dataobject as $field=>$value) {
757 if (!isset($columns[$field])) {
758 continue;
759 }
760 $column = $columns[$field];
761 if ($column->meta_type == 'B') {
762 if (is_null($value)) {
763 $cleaned[$field] = null;
764 } else {
765 $blobs[$field] = $value;
766 $cleaned[$field] = '@#BLOB#@';
767 }
768 continue;
769
770 } else if (is_bool($value)) {
771 $value = (int)$value; // prevent false '' problems
772
773 } else if ($value === '') {
774 if ($column->meta_type == 'I' or $column->meta_type == 'F' or $column->meta_type == 'N') {
775 $value = 0; // prevent '' problems in numeric fields
776 }
777 }
778
779 $cleaned[$field] = $value;
780 }
781
158622bd 782 if (empty($blobs)) {
783 return $this->insert_record_raw($table, $cleaned, $returnid, $bulk);
784 }
785
22d77567 786 $id = $this->insert_record_raw($table, $cleaned, true, $bulk);
158622bd 787
788 foreach ($blobs as $key=>$value) {
158622bd 789 $value = pg_escape_bytea($this->pgsql, $value);
790 $sql = "UPDATE {$this->prefix}$table SET $key = '$value'::bytea WHERE id = $id";
1500142b 791 $this->query_start($sql, NULL, SQL_QUERY_UPDATE);
e4f9c142 792 $result = pg_query($this->pgsql, $sql);
1500142b 793 $this->query_end($result);
e4f9c142 794 if ($result !== false) {
795 pg_free_result($result);
796 }
158622bd 797 }
798
799 return ($returnid ? $id : true);
800
801 }
802
803 /**
804 * Import a record into a table, id field is required.
805 * Safety checks are NOT carried out. Lobs are supported.
806 *
807 * @param string $table name of database table to be inserted into
808 * @param object $dataobject A data object with values for one or more fields in the record
22d77567 809 * @return bool true
810 * @throws dml_exception if error
158622bd 811 */
812 public function import_record($table, $dataobject) {
813 $dataobject = (object)$dataobject;
814
158622bd 815 $columns = $this->get_columns($table);
816 $cleaned = array();
817
818 foreach ($dataobject as $field=>$value) {
819 if (!isset($columns[$field])) {
820 continue;
821 }
822 $cleaned[$field] = $value;
823 }
824
825 return $this->insert_record_raw($table, $cleaned, false, true, true);
826 }
827
828 /**
829 * Update record in database, as fast as possible, no safety checks, lobs not supported.
830 * @param string $table name
831 * @param mixed $params data record as object or array
832 * @param bool true means repeated updates expected
22d77567 833 * @return bool true
834 * @throws dml_exception if error
158622bd 835 */
836 public function update_record_raw($table, $params, $bulk=false) {
837 if (!is_array($params)) {
838 $params = (array)$params;
839 }
840 if (!isset($params['id'])) {
22d77567 841 throw new coding_exception('moodle_database::update_record_raw() id field must be specified.');
158622bd 842 }
843 $id = $params['id'];
844 unset($params['id']);
845
846 if (empty($params)) {
22d77567 847 throw new coding_exception('moodle_database::update_record_raw() no fields found.');
158622bd 848 }
849
850 $i = 1;
851
852 $sets = array();
853 foreach ($params as $field=>$value) {
854 $sets[] = "$field = \$".$i++;
855 }
856
857 $params[] = $id; // last ? in WHERE condition
858
859 $sets = implode(',', $sets);
860 $sql = "UPDATE {$this->prefix}$table SET $sets WHERE id=\$".$i;
861
1500142b 862 $this->query_start($sql, $params, SQL_QUERY_UPDATE);
158622bd 863 $result = pg_query_params($this->pgsql, $sql, $params);
1500142b 864 $this->query_end($result);
158622bd 865
e4f9c142 866 pg_free_result($result);
158622bd 867 return true;
868 }
869
870 /**
871 * Update a record in a table
872 *
873 * $dataobject is an object containing needed data
874 * Relies on $dataobject having a variable "id" to
875 * specify the record to update
876 *
877 * @param string $table The database table to be checked against.
878 * @param object $dataobject An object with contents equal to fieldname=>fieldvalue. Must have an entry for 'id' to map to the table specified.
879 * @param bool true means repeated updates expected
22d77567 880 * @return bool true
881 * @throws dml_exception if error
158622bd 882 */
883 public function update_record($table, $dataobject, $bulk=false) {
884 if (!is_object($dataobject)) {
885 $dataobject = (object)$dataobject;
886 }
887
158622bd 888 $columns = $this->get_columns($table);
889 $cleaned = array();
d246cdd2 890 $blobs = array();
158622bd 891
892 foreach ($dataobject as $field=>$value) {
893 if (!isset($columns[$field])) {
894 continue;
895 }
d246cdd2 896 $column = $columns[$field];
897 if ($column->meta_type == 'B') {
898 if (is_null($value)) {
899 $cleaned[$field] = null;
900 } else {
901 $blobs[$field] = $value;
902 $cleaned[$field] = '@#BLOB#@';
903 }
904 continue;
905
906 } else if (is_bool($value)) {
907 $value = (int)$value; // prevent false '' problems
908
909 } else if ($value === '') {
910 if ($column->meta_type == 'I' or $column->meta_type == 'F' or $column->meta_type == 'N') {
911 $value = 0; // prevent '' problems in numeric fields
912 }
158622bd 913 }
d246cdd2 914
158622bd 915 $cleaned[$field] = $value;
916 }
917
22d77567 918 $this->update_record_raw($table, $cleaned, $bulk);
d246cdd2 919
920 if (empty($blobs)) {
921 return true;
922 }
923
22d77567 924 $id = (int)$dataobject->id;
925
d246cdd2 926 foreach ($blobs as $key=>$value) {
d246cdd2 927 $value = pg_escape_bytea($this->pgsql, $value);
928 $sql = "UPDATE {$this->prefix}$table SET $key = '$value'::bytea WHERE id = $id";
1500142b 929 $this->query_start($sql, NULL, SQL_QUERY_UPDATE);
d246cdd2 930 $result = pg_query($this->pgsql, $sql);
1500142b 931 $this->query_end($result);
22d77567 932
d246cdd2 933 pg_free_result($result);
934 }
935
936 return true;
158622bd 937 }
938
939 /**
940 * Set a single field in every table record which match a particular WHERE clause.
941 *
942 * @param string $table The database table to be checked against.
943 * @param string $newfield the field to set.
944 * @param string $newvalue the value to set the field to.
945 * @param string $select A fragment of SQL to be used in a where clause in the SQL call.
946 * @param array $params array of sql parameters
22d77567 947 * @return bool true
948 * @throws dml_exception if error
158622bd 949 */
950 public function set_field_select($table, $newfield, $newvalue, $select, array $params=null) {
951 if ($select) {
952 $select = "WHERE $select";
953 }
954 if (is_null($params)) {
955 $params = array();
956 }
957 list($select, $params, $type) = $this->fix_sql_params($select, $params);
958 $i = count($params)+1;
959
960 if (is_bool($newvalue)) {
961 $newvalue = (int)$newvalue; // prevent "false" problems
962 }
963 if (is_null($newvalue)) {
964 $newfield = "$newfield = NULL";
965 } else {
966 $newfield = "$newfield = \$".$i;
967 $params[] = $newvalue;
968 }
969 $sql = "UPDATE {$this->prefix}$table SET $newfield $select";
970
1500142b 971 $this->query_start($sql, $params, SQL_QUERY_UPDATE);
158622bd 972 $result = pg_query_params($this->pgsql, $sql, $params);
1500142b 973 $this->query_end($result);
158622bd 974
e4f9c142 975 pg_free_result($result);
158622bd 976
977 return true;
978 }
979
980 /**
981 * Delete one or more records from a table which match a particular WHERE clause.
982 *
983 * @param string $table The database table to be checked against.
984 * @param string $select A fragment of SQL to be used in a where clause in the SQL call (used to define the selection criteria).
985 * @param array $params array of sql parameters
22d77567 986 * @return bool true
987 * @throws dml_exception if error
158622bd 988 */
989 public function delete_records_select($table, $select, array $params=null) {
990 if ($select) {
991 $select = "WHERE $select";
992 }
993 $sql = "DELETE FROM {$this->prefix}$table $select";
994
995 list($sql, $params, $type) = $this->fix_sql_params($sql, $params);
996
1500142b 997 $this->query_start($sql, $params, SQL_QUERY_UPDATE);
158622bd 998 $result = pg_query_params($this->pgsql, $sql, $params);
1500142b 999 $this->query_end($result);
158622bd 1000
e4f9c142 1001 pg_free_result($result);
158622bd 1002
1003 return true;
1004 }
1005
1006 public function sql_ilike() {
1007 return 'ILIKE';
1008 }
1009
1010 public function sql_bitxor($int1, $int2) {
1011 return '(' . $this->sql_bitor($int1, $int2) . ' - ' . $this->sql_bitand($int1, $int2) . ')';
1012 }
1013
1014 public function sql_cast_char2int($fieldname, $text=false) {
1015 return ' CAST(' . $fieldname . ' AS INT) ';
1016 }
1017
1018 public function sql_cast_char2real($fieldname, $text=false) {
1019 return " $fieldname::real ";
1020 }
1021
1022 public function sql_concat() {
1023 $arr = func_get_args();
1024 $s = implode(' || ', $arr);
1025 if ($s === '') {
1026 return " '' ";
1027 }
1028 return " $s ";
1029 }
1030
1031 public function sql_concat_join($separator="' '", $elements=array()) {
1032 for ($n=count($elements)-1; $n > 0 ; $n--) {
1033 array_splice($elements, $n, 0, $separator);
1034 }
1035 $s = implode(' || ', $elements);
1036 if ($s === '') {
1037 return " '' ";
1038 }
1039 return " $s ";
1040 }
1041
158622bd 1042 public function sql_regex_supported() {
1043 return true;
1044 }
1045
1046 public function sql_regex($positivematch=true) {
1047 return $positivematch ? '~*' : '!~*';
1048 }
1049
fb76304b 1050/// transactions
1051 /**
1052 * on DBs that support it, switch to transaction mode and begin a transaction
1053 * you'll need to ensure you call commit_sql() or your changes *will* be lost.
1054 *
1055 * this is _very_ useful for massive updates
1056 */
1057 public function begin_sql() {
1500142b 1058 $sql = "BEGIN ISOLATION LEVEL READ COMMITTED";
1059 $this->query_start($sql, NULL, SQL_QUERY_AUX);
1060 $result = pg_query($this->pgsql, $sql);
1061 $this->query_end($result);
1062
fb76304b 1063 pg_free_result($result);
1064 return true;
1065 }
1066
1067 /**
1068 * on DBs that support it, commit the transaction
1069 */
1070 public function commit_sql() {
1500142b 1071 $sql = "COMMIT";
1072 $this->query_start($sql, NULL, SQL_QUERY_AUX);
1073 $result = pg_query($this->pgsql, $sql);
1074 $this->query_end($result);
1075
fb76304b 1076 pg_free_result($result);
1077 return true;
1078 }
1079
1080 /**
1081 * on DBs that support it, rollback the transaction
1082 */
1083 public function rollback_sql() {
1500142b 1084 $sql = "ROLLBACK";
1085 $this->query_start($sql, NULL, SQL_QUERY_AUX);
1086 $result = pg_query($this->pgsql, $sql);
1087 $this->query_end($result);
1088
fb76304b 1089 pg_free_result($result);
1090 return true;
1091 }
158622bd 1092}