Db.php

Go to the documentation of this file.

<?php
 
namespace Rsi\Fred;
 
/**
 *  Database access layer (PDO based).
 */
class Db extends Component{
 
  const ACTION_INSERT = 'insert'; //!< Insert a new record.
  const ACTION_CREATE = 'create'; //!< Create a new record (if none exists, otherwise ignore).
  const ACTION_REPLACE = 'replace'; //!< Replace an existing record (overwrite).
  const ACTION_UPDATE = 'update'; //!< Update an existing record.
  const ACTION_UPSERT = 'upsert'; //!< Insert a new record, or update the existing record if it already exists.
  const ACTION_DELETE = 'delete'; //!< Delete an existing record.
 
  const EVENT_OPEN = 'db:open';
  const EVENT_LOG_CHANGE = 'db:logChange';
  const EVENT_LOG_ADD_ID = 'db:logAddId';
 
  public $multiOperators = ['|' => 'or','&' => 'and']; //!< Operators (key) and glue (value) for multi column where conditions.
  public $statTime = 3600; //!< Query time above which to increment a stat counter for the specific query (id = SQL).
  public $statPrefix = 'db';
  public $logTimes = [ //!< Query time above which to add a note to the log (key = prio, value = edge; higher times first).
    Log::CRITICAL => 10.0,
    Log::ERROR => 5.0,
    Log::WARNING => 2.5
  ];
  public $logTimeSignificant = 3; //!< Number of significant figures (rounding).
  public $logTables = []; //!< Tables (key) and key columns (value) to log changes for.
  public $defTables = null; //!< Table definition to use (table name).
  public $migrateClassName = __CLASS__ . '\\Migrate'; //!< Class name for the migration tool.
 
  public $queryCount = 0; //!< Number of queries executed.
  public $queryTime = 0; //!< Total time spent on queries (seconds).
  public $allowInsert = true; //!< Allow inserts (for secondary processes like logs).
 
  protected $_defaultOptions = [\PDO::MYSQL_ATTR_FOUND_ROWS => true]; //default connection options (functional, but driver specific).
  protected $_defaultAttributes = [\PDO::MYSQL_ATTR_USE_BUFFERED_QUERY => true]; //default attributes (functional, but driver specific).
  protected $_eachOptions = [\PDO::ATTR_PERSISTENT => false]; //!< Do not (re-) use an existing connection (might be busy).
  protected $_eachAttributes = [\PDO::MYSQL_ATTR_USE_BUFFERED_QUERY => false]; //!< Do not buffer; retrieve one by one.
  protected $_connection = []; //!< Connection parameters (keys 'dsn', 'username', 'password', 'options').
  protected $_attributes = []; //!< Connection specific attributes (key = attribute, value = value).
 
  protected $_version = null;
  protected $_pdo = null;
  protected $_migrate = null;
 
  protected $_startTime = null;
  protected $_duploCount = [];
  protected $_logAddId = false;
  protected $_scout = [];
  /**
   * Create a PDO instance.
   * @param $options array  Connection options.
   * @param $attributes  Attributes (applied after connection).
   * @return PDO
   */
  public function createPdo($options = null,$attributes = null){
    $connection = new \Rsi\Wrapper\Record($this->_connection);
    $pdo = new \PDO($connection->dsn,$connection->username,$connection->password,array_replace(
      $this->_defaultOptions,
      $connection->options ? $connection->options->data : [],
      $options ?: []
    ));
    foreach(array_replace(
      $this->_defaultAttributes,
      $this->_attributes,
      $attributes ?: [],
      [\PDO::ATTR_DEFAULT_FETCH_MODE => \PDO::FETCH_ASSOC,\PDO::ATTR_ERRMODE => \PDO::ERRMODE_EXCEPTION]
    ) as $attribute => $value) $pdo->setAttribute($attribute,$value);
    $this->component('event')->trigger(self::EVENT_OPEN,$this,$pdo);
    return $pdo;
  }
  /**
   * Convert a Unix timestamp to database date format.
   * @param int $time  Timestamp (empty = now).
   * @return string  Date as a string.
   */
  public function date($time = null){
    return date('Y-m-d',$time ?: time());
  }
  /**
   * Convert a Unix timestamp to database date+time format.
   * @param int $time  Timestamp (empty = now).
   * @return string  Date+time as a string.
   */
  public function dateTime($time = null){
    return date('Y-m-d H:i:s',$time ?: time());
  }
  /**
   * Escape a value used for a "like" comparision.
   * @param string $value  String to escape.
   * @param string $escape  Escape character.
   * @return  Escaped string.
   */
  public function escapeLike($value,$escape = '\\'){
    foreach([$escape,'%','_'] as $char) $value = str_replace($char,$escape . $char,$value);
    return $value;
  }
  /**
   * Check if an SQL statement is a select statement.
   * @param string $sql  SQL statement.
   * @return bool  True if it is a select statement.
   */
  public function isSelection($sql){
    return !strcasecmp(substr(trim($sql),0,7),'select ');
  }
  /**
   * Primary key column.
   * @param string $table  For this table.
   * @return string  Column name.
   */
  public function keyColumn($table){
    return 'id';
  }
  /**
   * Begin a transaction.
   */
  public function begin(){
    $this->pdo->beginTransaction();
  }
  /**
   * Roll a transaction back.
   */
  public function rollBack(){
    $this->pdo->rollBack();
  }
  /**
   * Commit a transaction.
   */
  public function commit(){
    $this->pdo->commit();
  }
  /**
   * Wrap a callback function in a transaction.
   * @param callable $callback  Callback function (first and only parameter is this instance).
   * @param bool $throw  If true, an exception is re-thrown after the rollback.
   * @return bool  True on success, false on error (and $throw set to false).
   */
  public function transaction($callback,$throw = true){
    $this->begin();
    try{
      call_user_func($callback,$this);
      $this->commit();
    }
    catch(\Exception $e){
      $this->rollBack();
      if($throw) throw $e;
      return false;
    }
    return true;
  }
  /**
   * Get the last auto-incremented ID.
   * @return string
   */
  public function lastInsertId(){
    return $this->pdo->lastInsertId();
  }
 
  protected function startTimer(){
    $this->_startTime = microtime(true);
  }
 
  protected function checkTimer($sql,$args){
    $this->queryCount++;
    $this->queryTime += ($time = microtime(true) - $this->_startTime);
    $this->allowInsert = $this->isSelection($sql);
    if(($time >= $this->statTime) && ($stats = $this->component('stats')))
      $stats->inc($this->statPrefix . ':' . trim(preg_replace('/\\s+/',' ',$sql)),$time);
    if($log = $this->component('log')){
      foreach($this->logTimes as $prio => $edge) if($time >= $edge){
        $log->add($prio,'Slow query',compact('sql','args') + [
          'time' => round($time,max(0,$this->logTimeSignificant - ceil(log($time,10)))),
          'trace' => debug_backtrace(DEBUG_BACKTRACE_IGNORE_ARGS)
        ]);
        break;
      }
      if($this->_fred->debug){
        if(!array_key_exists($hash = md5($sql . serialize($args)),$this->_duploCount)) $this->_duploCount[$hash] = 0;
        else $log->info('Duplicate query' . ($this->_duploCount[$hash]++ ? ' (' . $this->_duploCount[$hash] . ')' : ''),__FILE__,__LINE__,compact('sql','args'));
      }
    }
    $this->allowInsert = true;
  }
 
  protected function prepareArgs(&$sql,&$args){
    if(!$args) $args = [];
    else foreach($args as $key => $value)
      if(!preg_match($pattern = "/:$key\\b/",$sql)) unset($args[$key]); //not used in SQL statement
      elseif(is_array($value)){
        if(!$value) throw new \Exception("Empty array for '$key'");
        unset($args[$key]);
        $keys = [];
        foreach($value as $sub) $args[$keys[] = $key . 's' . count($keys)] = $sub;
        $sql = preg_replace($pattern,':' . implode(',:',$keys),$sql);
      }
  }
  /**
   * Create a PDO statement.
   * @param string $sql  SQL statement to execute.
   * @param array $args  Variables to bind to the statement.
   * @param PDO $pdo  Specific PDO connection to use.
   * @return PDOStatement (false on failure).
   */
  public function statement($sql,$args = null,$pdo = null){
    if(!$pdo) $pdo = $this->pdo;
    if($statement = $pdo->prepare($sql)){
      if($args) foreach($args as $key => $value){
        $statement->bindValue(':' . $key,$value);
        unset($value);
      }
      $statement->execute();
    }
    return $statement;
  }
  /**
   * Execute an SQL statement.
   * @param string $sql  SQL statement to execute.
   * @param array $args  Variables to bind to the statement.
   * @param PDO $pdo  Specific PDO connection to use.
   * @return int  The number of affected rows.
   */
  public function execute($sql,$args = null,$pdo = null){
    if($log = $this->component('log')) $log->debug(__CLASS__ . "::execute('$sql',args)",__FILE__,__LINE__,compact('sql','args'));
    if(!$pdo) $pdo = $this->pdo;
    $result = null;
    $this->prepareArgs($sql,$args);
    $this->startTimer();
    if(!$args) $result = $pdo->exec($sql);
    elseif($statement = $this->statement($sql,$args,$pdo)){
      $result = $statement->rowCount();
      $statement->closeCursor();
      $statement = null;
    }
    $this->checkTimer($sql,$args);
    return $result;
  }
  /**
   * Execute an SQL statement.
   * @param string $sql  SQL statement to execute.
   * @param array $args  Variables to bind to the statement.
   * @param PDO $pdo  Specific PDO connection to use.
   * @return PDOStatement (false on failure).
   */
  public function query($sql,$args = null,$pdo = null){
    if($log = $this->component('log')) $log->debug(__CLASS__ . "::query('$sql',args)",__FILE__,__LINE__,compact('sql','args'));
    if(!$pdo) $pdo = $this->pdo;
    $this->prepareArgs($sql,$args);
    $this->startTimer();
    $result = $args ? $this->statement($sql,$args,$pdo) : $pdo->query($sql);
    $this->checkTimer($sql,$args);
    return $result;
  }
  /**
   * Fetch a row from an SQL statement.
   * @param PDOStatement $statement
   * @return array
   */
  public function fetch($statement){
    $row = $statement->fetch();
    return $this->defTables ? $this->component('def')->convertRecord($row,$this->defTables) : $row;
  }
  /**
   * Return al rows from an SQL statement.
   * @param string $sql  SQL statement to execute.
   * @param array $args  Variables to bind to the statement.
   * @return array  All rows (false on failure).
   */
  public function all($sql,$args = null){
    if($statement = $this->query($sql,$args)){
      if(!$this->defTables) return $statement->fetchAll();
      $rows = [];
      while($row = $this->fetch($statement)) $rows[] = $row;
      $statement->closeCursor();
      $statement = null;
      return $rows;
    }
    return false;
  }
  /**
   * Return a single row from an SQL statement.
   * Returns a single value if the resulting row has only one column.
   * @param string $sql  SQL statement to execute.
   * @param array $args  Variables to bind to the statement.
   * @param bool $auto  Set to false to always return a row.
   * @return mixed  Row (multi column) or value (single column).
   */
  public function single($sql,$args = null,$auto = true){
    if($statement = $this->query($sql,$args)){
      $row = $this->fetch($statement);
      $statement->closeCursor();
      $statement = null;
      if($row) return $auto && (count($row) == 1) ? array_pop($row) : $row;
    }
    return false;
  }
  /**
   * Returns an array from an SQL statement.
   * If the query returns only one column, the result is an array of those values. With two columns, the first one becomes the
   * key, and the second one the value of an assoc.array. With three or more columns the first column will be the key, and the
   * others the value.
   * @param string $sql  SQL statement to execute.
   * @param array $args  Variables to bind to the statement.
   * @return array
   */
  public function record($sql,$args = null){
    $result = [];
    if($statement = $this->query($sql,$args)){
      while($row = $this->fetch($statement)){
        $key = array_shift($row);
        switch(count($row)){
          case 0: $result[] = $key; break;
          case 1: $result[$key] = array_pop($row); break;
          default: $result[$key] = $row;
        }
      }
      $statement->closeCursor();
      $statement = null;
    }
    return $result;
  }
  /**
   * Run a callback function for every row in an SQL resultset.
   * A separate conection is used without buffering. This makes it possible to process huge datasets one record at a time,
   * without running out of memory.
   * @param callable $callback  Callback function (parameters row and index). If the function returns explicitly false the
   *   execution of further rows is halted.
   * @param string $sql  SQL statement to execute.
   * @param array $args  Variables to bind to the statement.
   * @return int  Number of records processed (false on error).
   */
  public function each($callback,$sql,$args = null){
    $result = false;
    if($statement = $this->query($sql,$args,$pdo = $this->createPdo($this->_eachOptions,$this->_eachAttributes))){
      $result = 0;
      while($row = $this->fetch($statement)) if(call_user_func($callback,$row,$result++) === false) break;
      $statement->closeCursor();
      $statement = null;
    }
    $pdo = null;
    return $result;
  }
  /**
   * Run a callback function for every row in a batched SQL resultset.
   * The rows are fetched in batches. This makes it possible to process huge datasets one record at a time, without using a
   *   separate connection, and without running out of memory. Note that this may not work perfectly when the dataset changes
   *   frequently (items added or deleted in the beginning shift the contents of later batches).
   * @param callable $callback  Callback function (parameters row and index). If the function returns explicitly false the
   *   execution of further rows is halted.
   * @param string $sql  SQL statement to execute.
   * @param array $args  Variables to bind to the statement.
   * @param int $limit  Batch size.
   * @return int  Number of records processed.
   */
  public function batch($callback,$sql,$args = null,$limit = 1000){
    if(!stripos($sql,'order by')) throw new \Exception('No ordering');
    $result = $offset = 0;
    while(($result == $offset) && ($rows = $this->all($this->limit($sql,$limit,$offset),$args))){
      foreach($rows as $row) if(call_user_func($callback,$row,$result++) === false) break 2;
      $offset += $limit;
    }
    return $result;
  }
  /**
   * Expand an array with scalar values and arrays to multiple arrays.
   * @param array $columns  Columns (key = column name, value = array with values or single scalar value).
   * @return array  Array of column arrays (key = column name, value = value).
   */
  public function multiColumns($columns){
    $keys = $count = false;
    foreach($columns as $column => $value) if(is_array($value)){
      if($keys === false) $count = count($keys = array_keys($value));
      elseif(($count != count($value)) || array_diff($keys,array_keys($value)))
        throw new \Exception("Different keys for column '$column'");
    }
    $result = [];
    foreach((array)$keys as $key){
      $result[$key] = [];
      foreach($columns as $column => $value) $result[$key][$column] = is_array($value) ? $value[$key] : $value;
    }
    return $result;
  }
  /**
   * Prepare a where statement.
   * @param string|array $where  If this a string, it is used directly as the where clause. If it is an assoc.array, it is
   *   translated to a where statement. Key => value pairs are translated as follow:
   *   - 'key' => 'value' : "key = 'value'" (or "key is null" when the value is null, or "key in (...)" when the value is an
   *     array).
   *   - 'key<>' => 'value' : "key <> 'value'" (or "key is not null" when the value is null, or "key not in (...)" when the
   *     value is an array).
   *   - 'key~' => 'value' : "key like 'value'"
   *   - 'key^' => [key1 => value1,key2 => value2] : "(key1 = value1 or key2 = value2)" (same rules as above apply for the 'or'
   *     part).
   * @param string|array $args  If the $where is a string, and this is also, this value is added to the where. Otherwise the
   *   arguments resulting from the translation will be added to this array.
   * @param string $glue  Glue to combine seperate arguments.
   */
  public function prepareWhere(&$where,&$args = null,$glue = 'and'){
    if($extra = $args && is_string($args) ? "\n" . $args : null) $args = null;
    if(is_array($where)){
      if(!is_array($args)) $args = [];
      $def = $this->defTables ? $this->component('def') : null;
      foreach($where as $column => &$value){
        if($raw = $column[0] == '!') $column = substr($column,1);
        $negation = '';
        if($operator = preg_match('/\\W+$/',$column,$match) ? $match[0] : null){
          $column = substr($column,0,-strlen($operator));
          if($operator[0] == '!'){
            $negation = 'not ';
            $operator = substr($operator,1);
          }
          switch($short = $operator){
            case '~' : $operator = 'like'; break;
          }
        }
        else $short = $operator = '=';
 
        if($value === null) switch($operator){
          case '<>': $negation = 'not ';
          case '=': $value = "`$column` is {$negation}null"; break;
          default: $value = '0=1'; //undefined
        }
        elseif(is_array($value)){
          if(array_key_exists($operator,$this->multiOperators)){
            $multi = count($value) > 1;
            $this->prepareWhere($value,$args,$this->multiOperators[$operator]);
            if($multi) $value = "($value)";
          }
          elseif($value){
            if($def) foreach($value as &$sub) $sub = $def->formatColumn($column,$sub,$this->defTables);
            unset($sub);
            if(!$negation && (count($value) == 1)){
              $args[$key = 'a' . count($args)] = array_pop($value);
              $value = "`$column` $operator :$key";
            }
            else{
              switch($short){
                case '<>':
                  $negation = 'not ';
                case '=':
                  $args[$key = 'a' . count($args)] = $value;
                  $value = "`$column` {$negation}in (:$key)";
                  break;
                case '~':
                  $likes = [];
                  foreach($value as $sub){
                    $args[$key = 'l' . count($args)] = $sub;
                    $likes[] = "`$column` like :$key";
                  }
                  $value = $negation . '(' . implode(' or ',$likes) . ')';
                  break;
                case '..':
                  $key = 'b' . count($args);
                  $args += ['l' . $key => array_shift($value),'u' . $key => array_shift($value)];
                  if($value) throw new \InvalidArgumentException("Too many arguments for 'between' for column '$column'");
                  $value = "$negation`$column` between :l$key and :u$key";
                  break;
                default:
                  throw new \DomainException("Invalid array operator '$operator' for column '$column'");
              }
            }
          }
          else $value = (int)($operator == '<>') . '=1';
        }
        else{
          if($null = substr($operator,0,1) == '#') $operator = substr($operator,1);
          if($raw) $value = "$negation`$column` $operator $value";
          else{
            $args[$key = 'v' . count($args)] = $def ? $def->formatColumn($column,$value,$this->defTables) : $value;
            $value = "$negation`$column` $operator :$key";
          }
          if($null) $value = "(`$column` is null or $value)";
        }
      }
      unset($value);
      $where = implode("\n  $glue ",$where);
    }
    elseif(!$args) $args = [];
    $where = ($where ?: '1=1') . $extra;
  }
  /**
   * Limit an SQL statement.
   * @param string $sql  Original SQL statement.
   * @param int $limit  Limit (number of rows).
   * @param int $offset  Offset.
   * @return string  SQL statement with limit.
   */
  public function limit($sql,$limit,$offset = null){
    return $sql . ' limit ' . max(0,(int)$offset) . ',' . max(0,(int)$limit);
  }
  /**
   * Log a change.
   * @param string $table  Table name.
   * @param string $action  Change type (see ACTION_* constants).
   * @param array $key  Key of the changed record (key = column name, value = value).
   * @param array $old  Original values for changed columns (key = column name, value = value).
   * @param array $new  New values for changed columns (key = column name, value = value).
   */
  protected function logChange($table,$action,$key,$old,$new){
    $this->_fred->event->trigger(self::EVENT_LOG_CHANGE,$this,$table,$action,$key,$old,$new);
  }
  /**
   * Add an auto-increment ID to the latest change log.
   * @param string $id  The last auto-incremented ID (the key is stored in $this->_logAddId).
   */
  protected function logAddId($id){
    $this->_fred->event->trigger(self::EVENT_LOG_ADD_ID,$this,$this->_logAddId,$id);
  }
  /**
   * Log the changes for an action.
   * @param string $table  Table name.
   * @param string $action  Change type (see ACTION_* constants).
   * @param array $columns  New values (key = column name, value = value).
   * @param string|array $where  Where.
   * @param string|array $args  Arguments (assoc.array) or extra statement for the where.
   */
  protected function logChanges($table,$action,$columns,$where,$args = null){
    $this->_logAddId = false;
    if($key_columns = $this->logTables[$table] ?? null) try{
      $key = is_array($where) ? \Rsi\Record::select($where,$key_columns) : [];
      if($keys = count($key) == count($key_columns) ? [$key] : $this->select($table,$key_columns,$where,$args)){
        foreach($keys as $key) if(
          ($current = $this->select($table,$columns === false ? '*' : array_keys($columns),$key,null,1)) &&
          ($changed = array_diff_assoc($current,$columns === false ? $key : $columns))
        ) $this->logChange($table,$action,$key,$changed,$columns === false ? false : array_intersect_key($columns,$changed));
      }
      elseif($action != self::ACTION_DELETE){
        $this->logChange($table,$action,false,false,$columns);
        if(count($key_columns) == 1) $this->_logAddId = array_pop($key_columns);
      }
    }
    catch(\Exception $e){
      if($this->_fred->debug) throw $e;
      $this->component('log')->critical("Could not log '$action' on '$table': " . $e->getMessage(),__FILE__,__LINE__,compact('columns','where','args'));
    }
  }
  /**
   * Pre-select records (will be used by select()).
   * @param string $table  Table name.
   * @param array $columns  Columns to select from the table. Leaving this empty will clear the cache for this table.
   * @param string|array $where  Where.
   * @param string|array $args  Arguments (assoc.array) or extra statement for the where.
   * @return array  Selected rows (null on clear).
   */
  public function scout($table,$columns = null,$where = null,$args = null){
    $result = null;
    if($columns){
      if(!array_key_exists($table,$this->_scout)) $this->_scout[$table] = [];
      if(!array_key_exists($key = serialize($columns),$this->_scout[$table])) $this->_scout[$table][$key] = [];
      foreach(($result = $this->select($table,$columns,$where,$args)) as $record) $this->_scout[$table][$key][serialize($record)] = $record;
    }
    else unset($this->_scout[$table]);
    return $result;
  }
 
  protected function updateSql($columns,&$args){
    if(!is_array($columns)) return $columns;
    $def = $this->defTables ? $this->component('def') : null;
    $sql = [];
    foreach($columns as $column => $value){
      if($raw = substr($column,0,1) == '!') $column = substr($column,1);
      elseif($def) $value = $def->formatColumn($column,$value,$this->defTables);
      $args[$key = 'u' . count($args)] = $value;
      $sql[] = "`$column` = " . ($raw ? $value : ':' . $key);
    }
    return implode(',',$sql);
  }
  /**
   * Insert a record.
   * @param string $table  Table name.
   * @param array $columns  Columns (key = column name, value = value or values - for multiple records).
   * @param string $action  Insert type to perform (see ACTION_* constants).
   * @param string|array $update  Update for an upsert action.
   * @return int  The number of affected rows (0 = failure, 1 = success).
   */
  public function insert($table,$columns,$action = self::ACTION_INSERT,$update = null){
    $this->scout($table);
    $def = $this->defTables ? $this->component('def') : null;
    $result = 0;
    foreach($this->multiColumns($columns) as $columns){
      $values = [];
      foreach($columns as $column => &$value){
        if($raw = substr($column,0,1) == '!') $column = substr($column,1);
        elseif($def) $value = $def->formatColumn($column,$value,$this->defTables);
        $values[$column] = $raw ? $value : ':' . $column;
      }
      unset($value);
      $this->logChanges($table,$action,$update ?: $columns,$args = $columns);
      $sql = " into `$table` (`" . implode('`,`',array_keys($values)) . "`)\nvalues (" . implode(',',$values) . ')';
      switch($action){
        case self::ACTION_UPSERT: $sql .= "\non duplicate key update " . $this->updateSql($update,$args);
        case self::ACTION_INSERT: $sql = 'insert' . $sql; break;
        case self::ACTION_CREATE: $sql = 'insert ignore' . $sql; break;
        case self::ACTION_REPLACE: $sql = 'replace' . $sql; break;
        default: throw new \Exception("Unknown insert action '$action'");
      }
      if($this->execute($sql,$args)){
        $result++;
        if($this->_logAddId && ($id = $this->lastInsertId())) try{
          $this->logAddId($id);
        }
        catch(\Exception $e){
          if($this->_fred->debug) throw $e;
          $this->component('log')->critical($e);
        }
      }
    }
    return $result;
  }
  /**
   * Create a record (if it does not exists).
   * @param string $table  Table name.
   * @param array $columns  Columns (key = column name, value = value or values - for multiple records).
   * @return int  The number of affected rows (0 = failure, 1 = success).
   */
  public function create($table,$columns){
    return $this->insert($table,$columns,self::ACTION_CREATE);
  }
  /**
   * Replace a record (insert if not exists).
   * @param string $table  Table name.
   * @param array $columns  Columns (key = column name, value = value or values - for multiple records).
   * @return int  The number of affected rows (0 = failure, 1 = success).
   */
  public function replace($table,$columns){
    return $this->insert($table,$columns,self::ACTION_REPLACE);
  }
  /**
   * Insert a record, or update it if it already exists.
   * @param string $table  Table name.
   * @param array $columns  Columns (key = column name, value = value or values - for multiple records).
   * @param array $update  Columns to update if the record already exists (key = column name, value = new value).
   */
  public function upsert($table,$columns,$update){
    return $this->insert($table,$columns,self::ACTION_UPSERT,$update);
  }
  /**
   * Select records.
   * @param string $table  Table name.
   * @param array $columns  Columns to select from the table. Prefix the first columns with a '+' to make this the key.
   * @param string|array $where  Where.
   * @param string|array $args  Arguments (assoc.array) or extra statement for the where.
   * @param int|bool $limit  Limit the number of rows (0 = no limit).
   * @param int $offset  Offset.
   * @return array  Single value when limit is true, single row when limit is 1, otherwise an array of records.
   * @see prepareWhere()
   */
  public function select($table,$columns = '*',$where = null,$args = null,$limit = null,$offset = null){
    if(
      ($limit == 1) && !$offset &&
      array_key_exists($table,$this->_scout) && array_key_exists($key = serialize($columns),$this->_scout[$table]) &&
      is_array($where) && !array_filter($where,function($value){
        return is_array($value);
      })
    ) foreach($this->_scout[$table][$key] as $record){
      foreach($where as $key => $value) if(($record[$key] ?? false) !== $value) continue 2;
      return $record;
    }
    $columns = preg_replace('/^(`)?\\+/','$1',is_array($columns) ? '`' . implode('`,`',$columns) . '`' : $columns,1,$record);
    $this->prepareWhere($where,$args);
    $sql = "select $columns\nfrom `$table`\nwhere $where";
    if($limit) $sql = $this->limit($sql,$limit,$offset);
    if($record) return $this->record($sql,$args);
    $rows = $this->all($sql,$args);
    return $limit === true
      ? (($row = array_shift($rows)) ? array_shift($row) : false) //single column
      : ($rows ? ($limit == 1 ? array_shift($rows) : $rows) : false);
  }
  /**
   * Update one or more record(s).
   * @param string $table  Table name.
   * @param array $columns  Columns to update (key = column name, value = new value).
   * @param string|array $where  Where (empty = all).
   * @param string|array $args  Arguments (assoc.array) or extra statement for the where.
   * @param bool $insert_if_not_exists  If true, a new record will be inserted when there is not yet an existing record.
   * @return int  The number of affected rows (0 = failure, 1 or more = success).
   */
  public function update($table,$columns,$where,$args = null,$insert_if_not_exists = false){
    $this->scout($table);
    $this->logChanges($table,self::ACTION_UPDATE,$columns,$where,$args);
    $this->prepareWhere($where,$args);
    $sql = "update `$table` set " . $this->updateSql($columns,$args) . "\nwhere $where";
    $result = $this->execute($sql,$args);
    if(!$result && $insert_if_not_exists) $result = $this->insert($table,$columns);
    return $result;
  }
  /**
   * Increment one or more columns, or insert a new record if none exists.
   * @param string $table  Table name.
   * @param string|array $columns  Column to increment (string), or columns to update (array; key = column name,
   *   value = increment).
   * @param array $where  Key to increment.
   * @return int  The number of affected rows (0 = failure, 1 or more = success).
   */
  public function inc($table,$columns,$where){
    if(!is_array($columns)) $columns = [$columns => 1];
    $update = [];
    foreach($columns as $column => $value) $update['!' . $column] = "`$column` + $value";
    return $this->upsert($table,$columns + $where,$update);
  }
  /**
   * Delete one or more record(s).
   * @param string $table  Table name.
   * @param string|array $where  Where (empty = all).
   * @param string|array $args  Arguments (assoc.array) or extra statement for the where.
   * @return int  The number of affected rows (0 = failure, 1 or more = success).
   */
  public function delete($table,$where,$args = null){
    $this->scout($table);
    $this->logChanges($table,self::ACTION_DELETE,false,$where,$args);
    $this->prepareWhere($where,$args);
    $sql = "delete from `$table` where $where";
    return $this->execute($sql,$args);
  }
  /**
   * Check if a record exists that meets the requirement.
   * @param string $table  Table name.
   * @param string|array $where  Where.
   * @param string|array $args  Arguments (assoc.array) or extra statement for the where.
   * @return bool  True if a record exists.
   */
  public function exists($table,$where = null,$args = null){
    $this->prepareWhere($where,$args);
    $sql = $this->limit("select 1 from `$table` where $where",1);
    return (bool)$this->fetch($this->query($sql,$args));
  }
  /**
   * Run an aggregation query.
   * @param string $table  Table name.
   * @param string $function Aggregation function (e.g. "sum").
   * @param string|array $where  Where.
   * @param string|array $args  Arguments (assoc.array) or extra statement for the where.
   * @return mixed
   */
  protected function aggregate($table,$function,$where = null,$args = null){
    $this->prepareWhere($where,$args);
    $sql = "select $function from `$table` where $where";
    $row = $this->fetch($this->query($sql,$args));
    return array_pop($row);
  }
  /**
   * The number of records that meet the requirements.
   * @param string $table  Table name.
   * @param string|array $where  Where.
   * @param string|array $args  Arguments (assoc.array) or extra statement for the where.
   * @return int  Number of rows.
   */
  public function count($table,$where = null,$args = null){
    return $this->aggregate($table,'count(*)',$where,$args);
  }
  /**
   * Smallest value from records that meet the requirements.
   * @param string $table  Table name.
   * @param string $column  Column to get value from.
   * @param string|array $where  Where.
   * @param string|array $args  Arguments (assoc.array) or extra statement for the where.
   * @return mixed  Smallest value.
   */
  public function min($table,$column,$where = null,$args = null){
    return $this->aggregate($table,"min(`$column`)",$where,$args);
  }
  /**
   * Biggest value from records that meet the requirements.
   * @param string $table  Table name.
   * @param string $column  Column to get value from.
   * @param string|array $where  Where.
   * @param string|array $args  Arguments (assoc.array) or extra statement for the where.
   * @return mixed  Biggest value.
   */
  public function max($table,$column,$where = null,$args = null){
    return $this->aggregate($table,"max(`$column`)",$where,$args);
  }
  /**
   * Average value from records that meet the requirements.
   * @param string $table  Table name.
   * @param string $column  Column to get value from.
   * @param string|array $where  Where.
   * @param string|array $args  Arguments (assoc.array) or extra statement for the where.
   * @return mixed  Average value.
   */
  public function average($table,$column,$where = null,$args = null){
    return $this->aggregate($table,"avg(`$column`)",$where,$args);
  }
  /**
   * Total value from records that meet the requirements.
   * @param string $table  Table name.
   * @param string $column  Column to get value from.
   * @param string|array $where  Where.
   * @param string|array $args  Arguments (assoc.array) or extra statement for the where.
   * @return mixed  Total value.
   */
  public function sum($table,$column,$where = null,$args = null){
    return $this->aggregate($table,"sum(`$column`)",$where,$args);
  }
  /**
   * Name of the current database.
   * @return string
   */
  public function database(){
    return $this->single('select database()');
  }
  /**
   * All tables in current database.
   * @return array
   */
  public function tables(){
    return $this->record('show tables');
  }
  /**
   * Column properties for a table.
   * @param string $table  Name of the table.
   * @param string $database  Name of the database (current database when empty).
   * @return array  Key = column name, value = assoc.array with properties.
   */
  public function columns($table,$database = null){
    return $this->record('
      select
        COLUMN_NAME,
        DATA_TYPE as `type`,
        CHARACTER_MAXIMUM_LENGTH as `length`,
        NUMERIC_PRECISION as `precision`,
        NUMERIC_SCALE as `scale`,
        if(right(COLUMN_TYPE,8) = "unsigned",1,0) as `unsigned`,
        COLLATION_NAME as `collation`,
        COLUMN_DEFAULT as `default`,
        if(IS_NULLABLE = "NO",1,0) as `required`,
        if(COLUMN_KEY = "PRI",1,0) as `primary`,
        (
          select concat(REFERENCED_TABLE_NAME,".",REFERENCED_COLUMN_NAME)
          from INFORMATION_SCHEMA.KEY_COLUMN_USAGE
          where TABLE_SCHEMA = col.TABLE_SCHEMA
            and TABLE_NAME = col.TABLE_NAME
            and COLUMN_NAME = col.COLUMN_NAME
            and REFERENCED_COLUMN_NAME is not null
        ) as `ref`
      from INFORMATION_SCHEMA.COLUMNS as col
      where TABLE_SCHEMA = :database
        and TABLE_NAME = :table
      order by ORDINAL_POSITION',
      [
        'database' => $database ?: $this->database(),
        'table' => $table
      ]
    );
  }
  /**
   * Optimize table(s).
   * @param string $table  Table to optimize (empty for all tables).
   */
  public function optimize($table = null){
    if($table){
      $this->component('log')->debug(__CLASS__ . "::optimize('$table')",__FILE__,__LINE__);
      $this->record("optimize table `$table`");
    }
    else foreach($this->tables() as $table) $this->optimize($table);
  }
  /**
   * Explain plan.
   * @param string $sql  SQL statement to explain.
   * @param array $args  Variables to bind to the statement.
   * @return array  Steps in the plan.
   */
  public function explain($sql,$args = null){
    return $this->all('explain ' . $sql,$args);
  }
  /**
   * Running processes.
   * @return array  Key = process ID, value = array with info.
   */
  public function processes(){
    $sql = '
      select
        ID,USER as `user`,HOST as `host`,DB as `database`,
        if(STATE = "",COMMAND,STATE) as `status`,TIME' . (stripos($this->version,'MariaDB') ? '_MS / 1000' : '') . ' as `time`,
        INFO as `sql`
      from INFORMATION_SCHEMA.PROCESSLIST';
    return $this->record($sql . ' where INFO not like :sql', ['sql' => trim($sql) . '%']);
  }
  /**
   * Kill a process
   * @param int $id  Process ID.
   * @return bool  True on success.
   */
  public function kill($id){
    try{
      $this->execute('kill :id',['id' => $id]);
      return true;
    }
    catch(\Exception $e){
      $this->component('log')->info($e);
    }
    return false;
  }
 
  protected function getMigrate(){
    if(!$this->_migrate){
      $class_name = $this->migrateClassName;
      $this->_migrate = new $class_name($this->_fred,$this->config('migrate',[]) + ['db' => $this]);
    }
    return $this->_migrate;
  }
 
  protected function getPdo(){
    if(!$this->_pdo) $this->_pdo = $this->createPdo();
    return $this->_pdo;
  }
 
  protected function getVersion(){
    if($this->_version === null)
      $this->_version = \Rsi\Record::value($this->record('show variables where variable_name="version"'),false);
    return $this->_version;
  }
 
  public function __call($func_name,$params){
    $result = false;
    if(substr($func_name,0,1) == '_'){
      $func_name = substr($func_name,1);
      $table = array_shift($params);
      if(in_array($func_name,['insert','replace','select','update','delete','exists','count'])) array_unshift($params,$table);
      $prev_tables = $this->defTables;
      $this->defTables = (array)$table;
      try{
        $result = call_user_func_array([$this,$func_name],$params);
      }
      finally{
        $this->defTables = $prev_tables;
      }
    }
    else{
      if(is_numeric($where = array_shift($params)) || (is_string($where) && !preg_match('/[\\s=<>]/',$where)))
        $where = [$this->keyColumn($func_name) => $where];
      $result = $this->select($func_name,'*',$where,array_shift($params),1);
    }
    return $result;
  }
 
  public function __invoke($sql,$args = null){
    return $this->execute($sql,$args);
  }
 
  public function __sleep(){
    return [];
  }
 
}