php-db-drivers/DatabasePdoPgsql.php

948 lines
30 KiB
PHP

<?php
/**
* PDO/PostgreSQL wrapper with (mostly) DatabaseMySQL interface :)
* Select builder is inspired by MediaWiki's one.
* Version: 2020-02-18
* (c) Vitaliy Filippov, 2015-2020
*/
if (!defined('MS_HASH'))
{
define('MS_HASH', 0);
define('MS_LIST', 1);
define('MS_ROW', 2);
define('MS_ONE', 2);
define('MS_COL', 4);
define('MS_VALUE', 6);
define('MS_RESULT', 8);
}
if (!class_exists('DatabaseException'))
{
class DatabaseException extends Exception {}
}
class DatabasePgsqlException extends DatabaseException
{
function isDuplicateKey()
{
return $this->getCode() == 23505;
}
}
if (!interface_exists('Database'))
{
interface Database {}
}
class DatabasePdoPgsql implements Database
{
var $host, $port, $socket, $username, $password, $dbname, $sslmode, $sslcert, $sslkey;
var $tableNames = array();
var $init = array();
var $queryLogFile;
var $reconnect = true;
var $autoBegin;
var $beginHook;
var $ondestroy = 'commit';
var $pgVersion;
var $queryCount = 0;
var $link;
var $transactions = array();
protected $calcFoundRows, $foundRows;
/**
* Creates a PostgreSQL connection object.
*
* @param array $options Possible options:
* host Host name or IP address to connect to [localhost]
* socket Directory of UNIX socket [/var/run/postgresql]
* port TCP port to connect to [5432]
* sslcert client SSL certificate to use
* sslkey client SSL key to use
* dbname DB name to use
* username Username
* password Password
* tableNames Table name mappings (virtual => real)
* queryLogFile Path to query log file or 'error_log' to log via error_log()
* reconnect Whether to reconnect on idle timeout [true]
* autoBegin Whether to automatically begin a transaction of first query [false]
* beginHook Callback to call when starting a transaction
* ondestroy commit/rollback/none during destruction [commit]
* init Initialisation queries (array)
*/
function __construct($options)
{
$defOpts = array(
'host' => 'localhost',
'port' => 5432,
'socket' => '/var/run/postgresql',
'dbname' => '',
'username' => '',
'password' => '',
'sslcert' => '',
'sslkey' => '',
'reconnect' => true,
'tableNames' => array(),
'queryLogFile' => '',
'autoBegin' => true,
'beginHook' => NULL,
'ondestroy' => 'commit',
'init' => array(),
);
$options += $defOpts;
if ($options['socket'] && !$options['host'])
{
$options['host'] = 'localhost';
}
foreach ($defOpts as $k => $v)
{
$this->$k = $options[$k];
}
}
function __destruct()
{
$o = $this->ondestroy;
if (($o === 'commit' || $o === 'rollback') && $this->transactions)
{
$this->transactions = array(false);
$this->$o();
}
}
function connect()
{
$str = "pgsql:port=".intval($this->port);
$str .= ";host='".($this->host == 'localhost' && $this->socket !== NULL ? $this->socket : $this->host)."';dbname='".$this->dbname."'";
if ($this->sslcert && $this->sslkey)
{
$str .= ";sslmode=require;sslcert='".$this->sslcert."';sslkey='".$this->sslkey."'";
}
try
{
$this->pgVersion = NULL;
$this->link = new PDO($str, $this->username, $this->password, array(
PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC,
PDO::ATTR_EMULATE_PREPARES => true,
));
foreach ($this->init as $q)
$this->link->query($q);
}
catch (PDOException $e)
{
throw new DatabaseException($e->getMessage(), $e->getCode(), $e);
}
$this->transactions = array();
}
function getPgVersion()
{
if (!$this->pgVersion)
{
// Версия в формате целого числа, типа 90600
$this->pgVersion = $this->select('show server_version_num', MS_VALUE);
}
return $this->pgVersion;
}
function getDBName()
{
return $this->dbname;
}
function quoteId($name)
{
return '"'.str_replace('"', '""', str_replace("\0", "", $name)).'"';
}
function quoteInto($str, $params)
{
$i = 0;
$r = '';
while (($p = strpos($str, '?')) !== false)
{
$r .= substr($str, 0, $p) . $this->quote($params[$i++]);
$str = substr($str, $p+1);
}
return $r.$str;
}
function quote($value)
{
if ($value === NULL)
return "NULL";
elseif ($value instanceof DatabasePdoPgsql_Fragment)
return $value->bind ? $this->quoteInto($value->text, $value->bind) : $value->text;
if (!$this->link)
$this->connect();
return $this->link->quote($value);
}
function query($sql)
{
if (!$this->link)
$this->connect();
if ($this->autoBegin && !$this->transactions)
$this->begin();
$this->queryCount++;
retry: try
{
if ($this->queryLogFile)
$start = microtime(true);
$r = $this->link->query($sql);
if ($this->queryLogFile)
{
$end = microtime(true);
if (strlen($sql) <= 100)
$sql = preg_replace("/\n\s*/", ' ', $sql);
$t = substr($start-intval($start), 1);
if ($this->queryLogFile === 'error_log')
error_log(sprintf("%.03f s ", $end-$start).$sql);
else
file_put_contents($this->queryLogFile, date("[Y-m-d H:i:s$t] ").sprintf("%.03f s ", $end-$start).$sql."\n", FILE_APPEND);
}
}
catch (Exception $e)
{
if ($this->reconnect && substr($e->getMessage(), 0, strlen('SQLSTATE[HY000]: General error: 7')) == 'SQLSTATE[HY000]: General error: 7')
{
$this->connect();
goto retry;
}
if ($this->queryLogFile)
{
if ($this->queryLogFile === 'error_log')
error_log("Error: $e\n---- Query: $sql");
else
file_put_contents($this->queryLogFile, date("[Y-m-d H:i:s] ")."Error: $e\n---- Query: $sql\n", FILE_APPEND);
}
throw $e;
}
return $r;
}
/**
* Starts a transaction, supports nested calls and savepoints.
* @param boolean $savepoint Creates savepoints only if this parameter is true.
*/
function begin($savepoint = false)
{
$this->transactions[] = $savepoint;
$n = count($this->transactions);
if ($n == 1)
{
$r = $this->query("BEGIN");
if ($this->beginHook)
{
$cb = $this->beginHook;
$cb($this);
}
return $r;
}
elseif ($savepoint)
{
return $this->query("SAVEPOINT sp$n");
}
return true;
}
/**
* Commits transaction or releases last savepoint.
* If there is no last savepoint, just returns true.
*/
function commit()
{
$r = true;
$savepoint = array_pop($this->transactions);
if (count($this->transactions) == 0)
$r = $this->query("COMMIT");
elseif ($savepoint)
$r = $this->query("RELEASE SAVEPOINT sp".count($this->transactions));
return $r;
}
/**
* Commits transaction
*/
function commitAll()
{
$r = true;
if ($this->transactions)
{
$r = $this->query("COMMIT");
$this->transactions = [];
}
return $r;
}
/**
* Rollbacks transaction
*/
function rollbackAll()
{
$r = true;
if ($this->transactions)
{
$r = $this->query("ROLLBACK");
$this->transactions = [];
}
return $r;
}
/**
* Rollbacks transaction or last savepoint.
* If there is no savepoint, returns false.
*/
function rollback()
{
$r = true;
$savepoint = array_pop($this->transactions);
if (count($this->transactions) == 1)
$r = $this->query("ROLLBACK");
elseif ($savepoint)
$r = $this->query("ROLLBACK TO SAVEPOINT sp".count($this->transactions));
return $r;
}
/**
* Builds WHERE-part of an SQL query.
* $where can also be a string - then it's passed as-is.
*
* @param array $where Query conditions:
* array(
* 'conditional expression',
* 'field_name' => 'value',
* 'field_name' => array('one', 'of', 'values'),
* 'field_name < ?' => 'value',
* 'field_name < DATE_SUB(?, ?)' => array('arg1', 'arg2'),
* 'field1,field2' => array(array(1, 2), array(3, 4)),
* )
*/
function where_builder($where)
{
if (!is_array($where))
return $where;
$wh = array();
foreach ($where as $k => $v)
{
if (ctype_digit("$k"))
{
if (is_array($v))
{
$str = array_shift($v);
$wh[] = $this->quoteInto($str, $v);
}
else
{
$wh[] = $v;
}
}
elseif (($p = strrpos($k, '?')) !== false)
{
$wh[] = $this->quoteInto($k, (array)$v);
}
elseif (is_array($v))
{
if (!$v)
{
// FIXME: It seems we should return empty result in that case
throw new DatabaseException("Error: empty array for '$k IN (...)', don't know what to do");
}
else
{
if (is_array(reset($v)))
foreach ($v as &$l)
$l = "(" . implode(",", array_map(array($this, 'quote'), $l)) . ")";
else
$v = array_map(array($this, 'quote'), $v);
$wh[] = "$k IN (" . implode(",", $v) . ")";
}
}
elseif ($v !== NULL)
$wh[] = "$k=".$this->quote($v);
else
$wh[] = "$k IS NULL";
}
if ($where)
$where = '(' . join(") AND (", $wh) . ')';
else
$where = '';
return $where;
}
/**
* Builds SQL query text.
*
* @param mixed $tables see $this->tablesBuilder()
* @param mixed $fields Field definitions - either a string or an array.
* Strings are passed to resulting query text as-is.
* Arrays have the following format:
* array('field1', 'alias2' => 'expression2', ...)
* @param mixed $where see $this->whereBuilder()
* @param array $options query options - array of:
* 'CALC_FOUND_ROWS' (emulated with window function)
* 'FOR UPDATE' or 'FOR SHARE' = 'LOCK IN SHARE MODE'
* 'GROUP BY' => array($groupby_field1 => 'ASC', $groupby_field2 => 'DESC')
* 'ORDER BY' => array($orderby_field1 => 'ASC', $orderby_field2 => 'DESC')
* 'LIMIT' => array($offset, $limit) or array($limit) or just $limit
* 'OFFSET' => $offset, for the case when 'LIMIT' is just $limit
* 'DISTINCT' => true
* 'DISTINCT ON' => $fields
* 'HAVING' => $having (like WHERE)
*/
function select_builder($tables, $fields, $where, $options = NULL)
{
if (!$options)
$options = array();
else
{
foreach ($options as $k => $v)
if (ctype_digit("$k"))
$options[$v] = true;
}
if (is_array($fields))
{
foreach ($fields as $k => &$v)
{
if ($v instanceof DatabasePdoPgsql_Fragment)
$v = $v->bind ? $this->quoteInto($v->text, $v->bind) : $v->text;
if (!ctype_digit("$k"))
$v = "$v AS ".$this->quoteId($k);
}
unset($v);
$fields = join(",\n ", str_replace("\n", "\n ", $fields));
}
$more = NULL;
$tables = $this->tables_builder($tables, $more);
if ($more)
$where = array_merge($where, (array)$more);
$where = $this->where_builder($where);
$this->calcFoundRows = isset($options['CALC_FOUND_ROWS']) || isset($options['SQL_CALC_FOUND_ROWS']);
if ($this->calcFoundRows)
$fields .= ",\n COUNT(*) OVER () \"*\"";
$sql = "SELECT";
if (isset($options['DISTINCT ON']))
$sql .= " DISTINCT ON (".implode(', ', (array)$options['DISTINCT ON']).")";
elseif (isset($options['DISTINCT']))
$sql .= " DISTINCT";
$sql .= " $fields\nFROM $tables";
if ($where)
$sql .= "\nWHERE $where";
if (!empty($options['GROUP BY']) && $options['GROUP BY'] !== '0')
$sql .= "\nGROUP BY ".$this->order_option($options['GROUP BY']);
if (!empty($options['HAVING']))
$sql .= "\nHAVING ".$this->where_builder($options['HAVING']);
if (!empty($options['ORDER BY']) && $options['ORDER BY'] !== '0')
$sql .= "\nORDER BY ".$this->order_option($options['ORDER BY']);
$sql .= $this->limit_option($options);
if (isset($options['FOR UPDATE']))
$sql .= ' FOR UPDATE';
elseif (isset($options['LOCK IN SHARE MODE']) || isset($options['FOR SHARE']))
$sql .= ' FOR SHARE';
return $sql;
}
/**
* Handles ORDER BY or GROUP BY options
*/
protected function order_option($g)
{
if (is_array($g))
{
$g1 = array();
foreach ($g as $k => $v)
{
if (ctype_digit("$k"))
$g1[] = $v;
else
$g1[] = "$k $v";
}
$g = join(',', $g1);
}
return $g;
}
/**
* Handles a single LIMIT or LIMIT and OFFSET options.
*/
protected function limit_option($options)
{
$g = '';
if (!empty($options['LIMIT']))
{
$g = $options['LIMIT'];
if (is_array($g) && count($g) != 2)
$g = $g[0];
if (is_array($g))
return " LIMIT ".$g[1].($g[0] > 0 ? " OFFSET ".$g[0] : '');
$g = " LIMIT $g";
}
if (!empty($options['OFFSET']))
{
$g .= " OFFSET ".$options['OFFSET'];
}
return $g;
}
/**
* Builds FROM-part of an SQL query.
*
* $tables = array(
* 'table',
* 'alias' => 'table',
* 'alias' => array('INNER', 'table_name', $where_for_on_clause),
* 'alias(ignored)' => array('INNER', $nested_tables, $on_for_join_group),
* 'alias' => array('INNER', $this->values_table($rows), $where_for_on_clause),
* )
* or just a string "table1 INNER JOIN \"table2\" ON ..."
* escaped names ("table2") will be transformed using $this->tableNames
*/
function tables_builder($tables, &$where = NULL)
{
if (is_array($tables))
{
$t = '';
foreach ($tables as $k => $v)
{
if (!is_array($v))
$v = [ 'INNER', $v, NULL ];
$join = strtolower(substr($v[0], 0, 4));
if ($join == 'righ')
$join = 'RIGHT';
elseif ($join == 'left')
$join = 'LEFT';
elseif ($join == 'full')
$join = 'FULL';
else /* if (!$join || $join == 'inne' || $join == 'join') */
$join = 'INNER';
if ($v[1] instanceof DatabasePdoPgsql_Values)
{
$v[1]->alias = $k;
$table = ''.$v[1];
}
elseif (is_array($v[1])) // nested join (left join (A join B on ...) on ...)
{
$more = NULL;
$table = $this->tables_builder($v[1], $more);
if ($more)
$v[2] = array_merge((array)$v[2], (array)$more);
if (count($v[1]) > 1)
$table = "($table)";
}
else
{
$table = (isset($this->tableNames[$v[1]]) ? $this->quoteId($this->tableNames[$v[1]]) : $v[1]);
if (!ctype_digit("$k"))
$table .= ' ' . $k;
}
if ($t)
$t .= "\n$join JOIN $table ON ".($this->where_builder($v[2]) ?: '1=1');
else
{
$t = $table;
$where = $v[2]; // extract ON to WHERE if only a single join is specified
}
}
$tables = $t;
}
else
$tables = preg_replace_callback('/((?:,|JOIN)\s*")([^"]+)/s', array($this, 'tables_builder_pregcb1'), $tables);
return $tables;
}
function tables_builder_pregcb1($m)
{
if (isset($this->tableNames[$m[2]]))
return $m[1].$this->tableNames[$m[2]];
return $m[1].$m[2];
}
/**
* Run a SELECT query and return results.
*
* Usage: either
* $this->select($tables, $fields, $where, $options, $format)
* using $this->select_builder() or
* $this->select($sql_text, $format)
* using query text.
*
* @param int $format Return format, bitmask of MS_XX constants:
* MS_RESULT = return PDO result object to manually fetch from
* MS_LIST = return rows as indexed arrays
* MS_HASH = return rows as associative arrays (default)
* MS_ROW = only return the first row
* MS_COL = only return the first column
* MS_VALUE = only return the first cell (just 1 value)
*/
function select($tables, $fields = '*', $where = 1, $options = NULL, $format = 0)
{
if (is_int($fields))
{
$this->calcFoundRows = false;
$sql = $tables;
$format = $fields;
}
else
$sql = $this->select_builder($tables, $fields, $where, $options);
$res = $this->query($sql);
if ($format & MS_RESULT)
return $res;
if (!$res)
return $format == MS_VALUE ? NULL : [];
if ($this->calcFoundRows)
{
// Extract and remove last column from query result
$fs = $format & MS_LIST ? PDO::FETCH_NUM : PDO::FETCH_ASSOC;
$rows = $format & MS_ROW ? [ $res->fetch($fs) ] : $res->fetchAll($fs);
$this->foundRows = reset($rows);
$this->foundRows = $this->foundRows ? end($this->foundRows) : 0;
if ($format & MS_COL)
foreach ($rows as &$row)
$row = reset($row);
else
foreach ($rows as &$row)
array_pop($row);
if ($format & MS_ROW)
$rows = $rows[0];
return $rows;
}
$fs = $format & MS_COL ? PDO::FETCH_COLUMN : ($format & MS_LIST ? PDO::FETCH_NUM : PDO::FETCH_ASSOC);
return $format & MS_ROW ? $res->fetch($fs) : $res->fetchAll($fs);
}
function found_rows()
{
return $this->foundRows;
}
/**
* Delete a set of rows.
* @param mixed $tables see $this->tables_builder() (deletes rows only from the first table)
* @param mixed $where see $this->where_builder()
* @param array $options Options for query:
* 'LIMIT' => array($limit, $offset) or array($limit) or just $limit
* 'OFFSET' => $offset, for the case when 'LIMIT' is just $limit
*/
function delete($tables, $where, $options = NULL)
{
$sql = $this->delete_builder($tables, $where, $options);
return $this->query($sql);
}
/**
* Same as delete(), but only returns SQL text
*/
function delete_builder($tables, $where, $options = NULL)
{
list($what, $using) = $this->split_using($tables, $where);
$more = NULL;
$sql = "DELETE FROM $what".($using ? " USING ".$this->tables_builder($using, $more) : '');
if ($more)
$where = array_merge($where, (array)$more);
$where = $this->where_builder($where) ?: '1=1';
$sql .= " WHERE $where";
$sql .= $this->limit_option($options);
return $sql;
}
public function values($rows, $forInsert)
{
$key = reset($rows);
if (!is_array($key))
{
$key = $rows;
$rows = [ $rows ];
}
else
{
foreach ($rows as $r)
$key += $r;
}
$key = array_keys($key);
foreach ($rows as &$r)
{
$rs = array();
foreach ($key as &$k)
{
$rs[] = isset($r[$k]) ? $this->quote($r[$k]) : ($forInsert ? 'DEFAULT' : 'NULL');
}
$r = implode(",", $rs);
}
foreach ($key as &$k)
{
if (strpos($k, '"') === false)
{
$k = $this->quoteId($k);
}
}
return array($key, $rows);
}
function values_table($rows, $alias = '')
{
list($keys, $values) = $this->values($rows, false);
return new DatabasePdoPgsql_Values($keys, $values, $alias);
}
function fragment($text, array $bind = [])
{
return new DatabasePdoPgsql_Fragment($text, $bind);
}
/**
* Builds an INSERT query.
*
* @param string $table Table name to insert rows to.
* @param array $rows Array of table rows to be inserted.
* @param string $action NULL, "UPDATE" or "IGNORE", for PgSQL 9.5 "on conflict do ..."
* @param array|string $uniqueColumns unique columns for conflict
* (defaults to {$table}_pkey for "insert or update" queries)
* @param array|NULL $updateCols Columns to update in case of a conflict
* may be array('field1', 'field2', ...) or array('field1' => 'expression', ...)
* NEW.any_field will be replaced with EXCLUDED.any_field in these expressions
*/
function insert_builder($table, $rows, $action = NULL, $uniqueColumns = NULL, $updateCols = NULL)
{
$alias = is_array($table) ? array_keys($table)[0] : NULL;
$table = is_array($table) ? $table[$alias] : $table;
if (isset($this->tableNames[$table]))
{
$table = $this->quoteId($this->tableNames[$table]);
}
list($keys, $values) = $this->values($rows, true);
$sql = "INSERT INTO $table".($alias ? " AS $alias" : "").
" (".implode(',', $keys).") VALUES (".implode('),(', $values).")";
if ($action)
{
$sql .= " ON CONFLICT";
if ($uniqueColumns)
$sql .= " (".implode(", ", (array)$uniqueColumns).")";
elseif ($action == "UPDATE")
$sql .= " ON CONSTRAINT ${table}_pkey";
if ($action == "UPDATE")
{
if ($updateCols)
$key = (array)$updateCols;
else
{
$key = reset($rows);
$key = is_array($key) ? array_keys($key) : array_keys($rows);
}
foreach ($key as $i => &$k)
{
if (is_int($i))
$k = "$k = EXCLUDED.$k";
else
$k = "$i = ".str_replace('NEW.', 'EXCLUDED.', $k);
}
$sql .= " DO UPDATE SET ".implode(", ", $key);
}
else
$sql .= " DO NOTHING";
}
return $sql;
}
/**
* Insert a single row into $table and return inserted ID.
*
* @param string $table Table name to insert row to.
* @param array $row Row to be inserted.
* @return int $insert_id Autoincrement ID of inserted row (if appropriate).
*/
function insert_row($table, $row)
{
$sql = $this->insert_builder($table, array($row)) . ' RETURNING LASTVAL()';
return $this->select($sql, MS_VALUE);
}
/**
* Insert multiple rows and return them all with their new values (including IDs)
*
* @param string $table Table name to insert row to.
* @param array $rows Rows to be inserted.
* @return array $rows Inserted rows with all fields (array of assoc arrays).
*/
function insert_returning($table, $rows)
{
if (!$rows)
return array();
$sql = $this->insert_builder($table, $rows) . ' RETURNING *';
return $this->select($sql, MS_HASH);
}
function insert_id()
{
return $this->link->lastInsertId();
}
/**
* Разбивает набор таблиц на основную обновляемую + набор дополнительных
*
* Идея в том, чтобы обрабатывать хотя бы 2 простые ситуации:
* UPDATE table1 INNER JOIN table2 ...
* UPDATE table1 LEFT JOIN table2 ...
*/
protected function split_using($tables, &$where)
{
$tables = (array)$tables;
$first = NULL;
$isNextInner = true;
$i = 0;
foreach ($tables as $k => $t)
{
if ($i == 0)
{
if (is_array($t) && ($t[1] instanceof DatabasePdoPgsql_Values || is_array($t[1])))
{
throw new DatabaseException("Can only update/delete from real tables, not sub-select, sub-join or VALUES");
}
$first = $k;
}
elseif ($i == 1)
{
$isNextInner = !is_array($t) || strtolower($t[0]) == 'inner';
}
else
break;
$i++;
}
$more = NULL;
if ($isNextInner)
{
$what = $this->tables_builder([ $first => $tables[$first] ], $more);
unset($tables[$first]);
}
else
{
$what = $this->tables_builder([ "_$first" => $tables[$first] ], $more);
$where[] = "_$first.ctid=".(ctype_digit("$first") ? $tables[$first] : $first).".ctid";
}
if ($more)
$where = array_merge($where, (array)$more);
return array($what, $tables);
}
/**
* Update row(s) in $table.
* $this->update($table, $set, $where);
*
* @param string $table Table(s) to update, see $this->tables_builder()
* @param array $set Assoc array with values for update query.
* @param array $where Conditions for update query, see $this->where_builder().
*/
function update($table, $set, $where = NULL, $options = NULL)
{
$sql = $this->update_builder($table, $set, $where, $options);
if (!$sql)
return false;
$r = $this->query($sql);
if ($r)
return $r->rowCount();
return false;
}
function update_builder($table, $set, $where = NULL, $options = NULL)
{
if (!$set)
return false;
$sql = array();
foreach ((array)$set as $k => $v)
{
if (!ctype_digit("$k"))
$sql[] = $k.'='.$this->quote($v);
else
$sql[] = $v;
}
list($what, $using) = $this->split_using($table, $where);
$more = NULL;
$sql = 'UPDATE ' . $what . ' SET ' . implode(', ', $sql) .
($using ? ' FROM '.$this->tables_builder($using, $more) : '');
if ($more)
$where = array_merge($where, $more);
$where = $this->where_builder($where) ?: '1=1';
$sql .= ' WHERE ' . $where;
return $sql;
}
/**
* Insert or update rows
*/
function insert($table, $rows, $onConflict = NULL, $uniqueKey = NULL, $updateCols = NULL)
{
if (!$rows)
return false;
$sql = $this->insert_builder($table, $rows, $onConflict, $uniqueKey, $updateCols);
$r = $this->query($sql);
if ($r)
return $r->rowCount();
return false;
}
function insert_ignore($table, $rows, $uniqueKey = NULL)
{
return $this->insert($table, $rows, 'IGNORE', $uniqueKey);
}
function upsert($table, $rows, $uniqueKey = NULL, $updateCols = NULL)
{
return $this->insert($table, $rows, 'UPDATE', $uniqueKey, $updateCols);
}
function replace($table, $rows, $uniqueKey = NULL)
{
$first = reset($rows);
if (!is_array($first))
{
$first = $rows;
$rows = [ $first ];
}
$key = $uniqueKey ? (is_array($uniqueKey) ? $uniqueKey : array_map('trim', explode(',', $uniqueKey))) : array_keys($first);
$where = [];
foreach ($rows as $row)
{
$r = [];
foreach ($key as $k)
$r[] = $row[$k];
$where[] = $r;
}
$n = $this->delete($table, array("(".implode(", ", $key).")" => $where));
$n += $this->insert($table, $rows);
return $n;
}
}
class DatabasePdoPgsql_Values
{
public $keys, $values, $alias;
function __construct($keys, $values, $alias)
{
$this->keys = $keys;
$this->values = $values;
$this->alias = $alias;
}
function __toString()
{
return "(VALUES (".implode("),(", $this->values).")) AS ".$this->alias." (".implode(',', $this->keys).")";
}
}
class DatabasePdoPgsql_Fragment
{
public $text, $bind;
function __construct($text, $bind)
{
$this->text = $text;
$this->bind = $bind;
}
}