class.ilDBPdo.php

Go to the documentation of this file.

<?php
/* Copyright (c) 1998-2009 ILIAS open source, Extended GPL, see docs/LICENSE */

require_once("./Services/Database/classes/PDO/class.ilPDOStatement.php");
require_once("./Services/Database/classes/QueryUtils/class.ilMySQLQueryUtils.php");
require_once('./Services/Database/classes/PDO/Manager/class.ilDBPdoManager.php');
require_once('./Services/Database/classes/PDO/Reverse/class.ilDBPdoReverse.php');
require_once('./Services/Database/interfaces/interface.ilDBInterface.php');
require_once('./Services/Database/classes/class.ilDBConstants.php');
require_once('./Services/Database/interfaces/interface.ilDBLegacyInterface.php');

abstract class ilDBPdo implements ilDBInterface, ilDBPdoInterface
{
    const FEATURE_TRANSACTIONS = 'transactions';
    const FEATURE_FULLTEXT = 'fulltext';
    const FEATURE_SLAVE = 'slave';
    protected $host = '';
    protected $dbname = '';
    protected $charset = 'utf8';
    protected $username = '';
    protected $password = '';
    protected $port = 3306;
    protected $pdo;
    protected $manager;
    protected $reverse;
    protected $limit = null;
    protected $offset = null;
    protected $storage_engine = 'MyISAM';
    protected $dsn = '';
    protected $attributes = array(
        //              PDO::ATTR_EMULATE_PREPARES => true,
        PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
    );
    protected $db_type = '';
    protected $error_code = 0;
    protected $field_definition;


    public function connect($return_false_for_error = false)
    {
        $this->generateDSN();
        try {
            $options = $this->getAttributes();
            $this->pdo = new PDO($this->getDSN(), $this->getUsername(), $this->getPassword(), $options);
            $this->initHelpers();
            $this->initSQLMode();
        } catch (Exception $e) {
            $this->error_code = $e->getCode();
            if ($return_false_for_error) {
                return false;
            }
            throw $e;
        }

        return ($this->pdo->errorCode() == PDO::ERR_NONE);
    }


    abstract public function initHelpers();


    protected function initSQLMode()
    {
    }


    protected function getAttributes()
    {
        $options = $this->attributes;
        foreach ($this->getAdditionalAttributes() as $k => $v) {
            $options[$k] = $v;
        }

        return $options;
    }


    protected function getAdditionalAttributes()
    {
        return array();
    }


    public function getFieldDefinition()
    {
        return $this->field_definition;
    }


    public function setFieldDefinition($field_definition)
    {
        $this->field_definition = $field_definition;
    }


    public function createDatabase($a_name, $a_charset = "utf8", $a_collation = "")
    {
        $this->setDbname(null);
        $this->generateDSN();
        $this->connect(true);
        try {
            return $this->query($this->manager->getQueryUtils()->createDatabase($a_name, $a_charset, $a_collation));
        } catch (PDOException $e) {
            return false;
        }
    }


    public function getLastErrorCode()
    {
        if ($this->pdo instanceof PDO) {
            return $this->pdo->errorCode();
        }

        return $this->error_code;
    }


    public function initFromIniFile($tmpClientIniFile = null)
    {
        global $ilClientIniFile;
        if ($tmpClientIniFile instanceof ilIniFile) {
            $clientIniFile = $tmpClientIniFile;
        } else {
            $clientIniFile = $ilClientIniFile;
        }

        $this->setUsername($clientIniFile->readVariable("db", "user"));
        $this->setHost($clientIniFile->readVariable("db", "host"));
        $this->setPort((int) $clientIniFile->readVariable("db", "port"));
        $this->setPassword($clientIniFile->readVariable("db", "pass"));
        $this->setDbname($clientIniFile->readVariable("db", "name"));
        $this->setDBType($clientIniFile->readVariable("db", "type"));

        $this->generateDSN();
    }


    public function generateDSN()
    {
        $port = $this->getPort() ? ";port=" . $this->getPort() : "";
        $dbname = $this->getDbname() ? ';dbname=' . $this->getDbname() : '';
        $host = $this->getHost();
        $charset = ';charset=' . $this->getCharset();
        $this->dsn = 'mysql:host=' . $host . $port . $dbname . $charset;
    }


    public function quoteIdentifier($identifier, $check_option = false)
    {
        return '`' . $identifier . '`';
    }


    public function nextId($table_name)
    {
        $sequence_table_name = $table_name . '_seq';

        $last_insert_id = $this->pdo->lastInsertId($table_name);
        if ($last_insert_id) {
            //                  return $last_insert_id;
        }

        if ($this->tableExists($sequence_table_name)) {
            $stmt = $this->pdo->prepare("SELECT sequence FROM $sequence_table_name");
            $stmt->execute();
            $rows = $stmt->fetch(PDO::FETCH_ASSOC);
            $stmt->closeCursor();
            $next_id = $rows['sequence'] + 1;
            $stmt = $this->pdo->prepare("DELETE FROM $sequence_table_name");
            $stmt->execute(array("next_id" => $next_id));
            $stmt = $this->pdo->prepare("INSERT INTO $sequence_table_name (sequence) VALUES (:next_id)");
            $stmt->execute(array("next_id" => $next_id));

            return $next_id;
        }

        return 1;
    }


    public function createTable($table_name, $fields, $drop_table = false, $ignore_erros = false)
    {
        // check table name
        if (!$this->checkTableName($table_name) && !$ignore_erros) {
            throw new ilDatabaseException("ilDB Error: createTable(" . $table_name . ")");
        }

        // check definition array
        if (!$this->checkTableColumns($fields) && !$ignore_erros) {
            throw new ilDatabaseException("ilDB Error: createTable(" . $table_name . ")");
        }

        if ($drop_table) {
            $this->dropTable($table_name, false);
        }

        return $this->manager->createTable($table_name, $fields, array());
    }


    protected function checkTableColumns($a_cols)
    {
        foreach ($a_cols as $col => $def) {
            if (!$this->checkColumn($col, $def)) {
                return false;
            }
        }

        return true;
    }


    protected function checkColumn($a_col, $a_def)
    {
        if (!$this->checkColumnName($a_col)) {
            return false;
        }

        if (!$this->checkColumnDefinition($a_def)) {
            return false;
        }

        return true;
    }


    protected function checkColumnDefinition($a_def, $a_modify_mode = false)
    {
        return $this->field_definition->checkColumnDefinition($a_def);
    }


    public function checkColumnName($a_name)
    {
        return $this->field_definition->checkColumnName($a_name);
    }


    public function addPrimaryKey($table_name, $primary_keys)
    {
        assert(is_array($primary_keys));

        $fields = array();
        foreach ($primary_keys as $f) {
            $fields[$f] = array();
        }
        $definition = array(
            'primary' => true,
            'fields' => $fields,
        );
        $this->manager->createConstraint($table_name, $this->constraintName($table_name, $this->getPrimaryKeyIdentifier()), $definition);

        return true;
    }


    public function dropIndexByFields($table_name, $fields)
    {
        foreach ($this->manager->listTableIndexes($table_name) as $idx_name) {
            $def = $this->reverse->getTableIndexDefinition($table_name, $idx_name);
            $idx_fields = array_keys((array) $def['fields']);

            if ($idx_fields === $fields) {
                return $this->dropIndex($table_name, $idx_name);
            }
        }

        return false;
    }


    public function getPrimaryKeyIdentifier()
    {
        return "PRIMARY";
    }


    public function createSequence($table_name, $start = 1)
    {
        $this->manager->createSequence($table_name, $start);
    }


    public function tableExists($table_name)
    {
        $result = $this->pdo->prepare("SHOW TABLES LIKE :table_name");
        $result->execute(array('table_name' => $table_name));
        $return = $result->rowCount();
        $result->closeCursor();

        return $return > 0;
    }


    public function tableColumnExists($table_name, $column_name)
    {
        $fields = $this->loadModule(ilDBConstants::MODULE_MANAGER)->listTableFields($table_name);

        $in_array = in_array($column_name, $fields);

        return $in_array;
    }


    public function addTableColumn($table_name, $column_name, $attributes)
    {
        if (!$this->checkColumnName($column_name)) {
            throw new ilDatabaseException("ilDB Error: addTableColumn(" . $table_name . ", " . $column_name . ")");
        }
        if (!$this->checkColumnDefinition($attributes)) {
            throw new ilDatabaseException("ilDB Error: addTableColumn(" . $table_name . ", " . $column_name . ")");
        }

        $changes = array(
            "add" => array(
                $column_name => $attributes,
            ),
        );

        return $this->manager->alterTable($table_name, $changes, false);
    }


    public function dropTable($table_name, $error_if_not_existing = true)
    {
        $ilDBPdoManager = $this->loadModule(ilDBConstants::MODULE_MANAGER);
        $tables = $ilDBPdoManager->listTables();
        $table_exists = in_array($table_name, $tables);
        if (!$table_exists && $error_if_not_existing) {
            throw new ilDatabaseException("Table {$table_name} does not exist");
        }

        // drop sequence
        $sequences = $ilDBPdoManager->listSequences();
        if (in_array($table_name, $sequences)) {
            $ilDBPdoManager->dropSequence($table_name);
        }

        // drop table
        if ($table_exists) {
            $ilDBPdoManager->dropTable($table_name);
        }

        return true;
    }


    public function query($query)
    {
        global $DIC;
        $ilBench = $DIC['ilBench'];

        $query = $this->appendLimit($query);

        try {
            if ($ilBench instanceof ilBenchmark) {
                $ilBench->startDbBench($query);
            }
            $res = $this->pdo->query($query);
            if ($ilBench instanceof ilBenchmark) {
                $ilBench->stopDbBench();
            }
        } catch (PDOException $e) {
            throw new ilDatabaseException($e->getMessage() . ' QUERY: ' . $query);
        }

        $err = $this->pdo->errorCode();
        if ($err != PDO::ERR_NONE) {
            $info = $this->pdo->errorInfo();
            $info_message = $info[2];
            throw new ilDatabaseException($info_message . ' QUERY: ' . $query);
        }

        return new ilPDOStatement($res);
    }


    public function fetchAll($query_result, $fetch_mode = ilDBConstants::FETCHMODE_ASSOC)
    {
        $return = array();
        while ($data = $query_result->fetch($fetch_mode)) {
            $return[] = $data;
        }

        return $return;
    }


    public function dropSequence($table_name)
    {
        $this->manager->dropSequence($table_name);
    }


    public function dropTableColumn($table_name, $column_name)
    {
        $changes = array(
            "remove" => array(
                $column_name => array(),
            ),
        );

        return $this->manager->alterTable($table_name, $changes, false);
    }


    public function renameTableColumn($table_name, $column_old_name, $column_new_name)
    {
        // check table name
        if (!$this->checkColumnName($column_new_name)) {
            throw new ilDatabaseException("ilDB Error: renameTableColumn(" . $table_name . "," . $column_old_name . "," . $column_new_name . ")");
        }

        $def = $this->reverse->getTableFieldDefinition($table_name, $column_old_name);

        $analyzer = new ilDBAnalyzer($this);
        $best_alt = $analyzer->getBestDefinitionAlternative($def);
        $def = $def[$best_alt];
        unset($def["nativetype"]);
        unset($def["mdb2type"]);

        $f["definition"] = $def;
        $f["name"] = $column_new_name;

        $changes = array(
            "rename" => array(
                $column_old_name => $f,
            ),
        );

        return $this->manager->alterTable($table_name, $changes, false);
    }


    public function insert($table_name, $values)
    {
        $real = array();
        $fields = array();
        foreach ($values as $key => $val) {
            $real[] = $this->quote($val[1], $val[0]);
            $fields[] = $this->quoteIdentifier($key);
        }
        $values = implode(",", $real);
        $fields = implode(",", $fields);
        $query = "INSERT INTO " . $table_name . " (" . $fields . ") VALUES (" . $values . ")";

        $query = $this->sanitizeMB4StringIfNotSupported($query);

        return $this->pdo->exec($query);
    }


    public function fetchObject($query_result)
    {
        $res = $query_result->fetchObject();
        if ($res == null) {
            $query_result->closeCursor();

            return null;
        }

        return $res;
    }


    public function update($table_name, $columns, $where)
    {
        $fields = array();
        $field_values = array();
        $placeholders = array();
        $placeholders_full = array();
        $types = array();
        $values = array();
        $lobs = false;
        $lob = array();
        foreach ($columns as $k => $col) {
            $field_value = $col[1];
            $fields[] = $k;
            $placeholders[] = "%s";
            $placeholders_full[] = ":$k";
            $types[] = $col[0];

            if ($col[0] == "blob" || $col[0] == "clob" || $col[0] == 'text') {
                $field_value = $this->sanitizeMB4StringIfNotSupported($field_value);
            }

            // integer auto-typecast (this casts bool values to integer)
            if ($col[0] == 'integer' && !is_null($field_value)) {
                $field_value = (int) $field_value;
            }

            $values[] = $field_value;
            $field_values[$k] = $field_value;
            if ($col[0] == "blob" || $col[0] == "clob") {
                $lobs = true;
                $lob[$k] = $k;
            }
        }

        if ($lobs) {
            $q = "UPDATE " . $table_name . " SET ";
            $lim = "";
            foreach ($fields as $k => $field) {
                $q .= $lim . $field . " = " . $placeholders_full[$k];
                $lim = ", ";
            }
            $q .= " WHERE ";
            $lim = "";
            foreach ($where as $k => $col) {
                $q .= $lim . $k . " = " . $this->quote($col[1], $col[0]);
                $lim = " AND ";
            }

            $r = $this->prepareManip($q, $types);
            $this->execute($r, $field_values);
            $this->free($r);
        } else {
            foreach ($where as $k => $col) {
                $types[] = $col[0];
                $values[] = $col[1];
                $field_values[$k] = $col;
            }
            $q = "UPDATE " . $table_name . " SET ";
            $lim = "";
            foreach ($fields as $k => $field) {
                $q .= $lim . $this->quoteIdentifier($field) . " = " . $placeholders[$k];
                $lim = ", ";
            }
            $q .= " WHERE ";
            $lim = "";
            foreach ($where as $k => $col) {
                $q .= $lim . $k . " = %s";
                $lim = " AND ";
            }

            $r = $this->manipulateF($q, $types, $values);
        }

        return $r;
    }



    public function manipulate($query)
    {
        global $DIC;
        $ilBench = $DIC['ilBench'];
        try {
            $query = $this->sanitizeMB4StringIfNotSupported($query);
            if ($ilBench instanceof ilBenchmark) {
                $ilBench->startDbBench($query);
            }
            $r = $this->pdo->exec($query);
            if ($ilBench instanceof ilBenchmark) {
                $ilBench->stopDbBench();
            }
        } catch (PDOException $e) {
            throw new ilDatabaseException($e->getMessage() . ' QUERY: ' . $query);
        }

        return $r;
    }


    public function fetchAssoc($query_result)
    {
        $res = $query_result->fetch(PDO::FETCH_ASSOC);
        if ($res == null) {
            $query_result->closeCursor();

            return null;
        }

        return $res;
    }


    public function numRows($query_result)
    {
        return $query_result->rowCount();
    }


    public function quote($value, $type = null)
    {
        if ($value === null) {
            return 'NULL';
        }

        $pdo_type = PDO::PARAM_STR;
        switch ($type) {
            case ilDBConstants::T_TIMESTAMP:
            case ilDBConstants::T_DATETIME:
            case ilDBConstants::T_DATE:
                if ($value === '') {
                    return 'NULL';
                }
                break;
            case ilDBConstants::T_INTEGER:
                $value = (int) $value;

                return $value;
                break;
            case ilDBConstants::T_FLOAT:
                $pdo_type = PDO::PARAM_INT;
                break;
            case ilDBConstants::T_TEXT:
            default:
                $pdo_type = PDO::PARAM_STR;
                break;
        }

        return $this->pdo->quote($value, $pdo_type);
    }


    public function indexExistsByFields($table_name, $fields)
    {
        foreach ($this->manager->listTableIndexes($table_name) as $idx_name) {
            $def = $this->reverse->getTableIndexDefinition($table_name, $idx_name);
            $idx_fields = array_keys((array) $def['fields']);

            if ($idx_fields === $fields) {
                return true;
            }
        }

        return false;
    }


    public function addIndex($table_name, $fields, $index_name = '', $fulltext = false)
    {
        assert(is_array($fields));
        $this->field_definition->checkIndexName($index_name);

        $definition_fields = array();
        foreach ($fields as $f) {
            $definition_fields[$f] = array();
        }
        $definition = array(
            'fields' => $definition_fields,
        );

        if (!$fulltext) {
            $this->manager->createIndex($table_name, $this->constraintName($table_name, $index_name), $definition);
        } else {
            if ($this->supportsFulltext()) {
                $this->addFulltextIndex($table_name, $fields, $index_name); // TODO
            }
        }

        return true;
    }


    public function addFulltextIndex($a_table, $a_fields, $a_name = "in")
    {
        $i_name = $this->constraintName($a_table, $a_name) . "_idx";
        $f_str = implode($a_fields, ",");
        $q = "ALTER TABLE $a_table ADD FULLTEXT $i_name ($f_str)";
        $this->query($q);
    }


    public function dropFulltextIndex($a_table, $a_name)
    {
        $i_name = $this->constraintName($a_table, $a_name) . "_idx";
        $this->query("ALTER TABLE $a_table DROP FULLTEXT $i_name");
    }


    public function isFulltextIndex($a_table, $a_name)
    {
        $set = $this->query("SHOW INDEX FROM " . $a_table);
        while ($rec = $this->fetchAssoc($set)) {
            if ($rec["Key_name"] == $a_name && $rec["Index_type"] == "FULLTEXT") {
                return true;
            }
        }

        return false;
    }


    public function getIndexName($index_name_base)
    {
        return sprintf(ilDBPdoFieldDefinition::INDEX_FORMAT, preg_replace('/[^a-z0-9_\$]/i', '_', $index_name_base));
    }


    public function getSequenceName($table_name)
    {
        return sprintf(ilDBPdoFieldDefinition::SEQUENCE_FORMAT, preg_replace('/[^a-z0-9_\$.]/i', '_', $table_name));
    }


    public function constraintName($a_table, $a_constraint)
    {
        return $a_constraint;
    }


    public function getDSN()
    {
        return $this->dsn;
    }


    public function getDBType()
    {
        return $this->db_type;
    }


    public function setDBType($type)
    {
        $this->db_type = $type;
    }


    public static function getReservedWords()
    {
        global $ilDB;

        return $ilDB->getFieldDefinition()->getReservedMysql();
    }


    public function lockTables($tables)
    {
        assert(is_array($tables));

        $lock = $this->manager->getQueryUtils()->lock($tables);
        global $ilLog;
        if ($ilLog instanceof ilLog) {
            $ilLog->write('ilDB::lockTables(): ' . $lock);
        }

        $this->pdo->exec($lock);
    }


    public function unlockTables()
    {
        $this->pdo->exec($this->manager->getQueryUtils()->unlock());
    }


    public function in($field, $values, $negate = false, $type = "")
    {
        return $this->manager->getQueryUtils()->in($field, $values, $negate, $type);
    }


    public function queryF($query, $types, $values)
    {
        if (!is_array($types) || !is_array($values) || count($types) != count($values)) {
            throw new ilDatabaseException("ilDB::queryF: Types and values must be arrays of same size. ($query)");
        }
        $quoted_values = array();
        foreach ($types as $k => $t) {
            $quoted_values[] = $this->quote($values[$k], $t);
        }
        $query = vsprintf($query, $quoted_values);

        return $this->query($query);
    }


    public function manipulateF($query, $types, $values)
    {
        if (!is_array($types) || !is_array($values) || count($types) != count($values)) {
            throw new ilDatabaseException("ilDB::manipulateF: types and values must be arrays of same size. ($query)");
        }
        $quoted_values = array();
        foreach ($types as $k => $t) {
            $quoted_values[] = $this->quote($values[$k], $t);
        }
        $query = vsprintf($query, $quoted_values);

        return $this->manipulate($query);
    }


    public function useSlave($bool)
    {
        return false;
    }


    public function setLimit($limit, $offset = 0)
    {
        $this->limit = $limit;
        $this->offset = $offset;
    }


    public function like($column, $type, $value = "?", $case_insensitive = true)
    {
        return $this->manager->getQueryUtils()->like($column, $type, $value, $case_insensitive);
    }


    public function now()
    {
        return $this->manager->getQueryUtils()->now();
    }


    public function replace($table, $primaryKeys, $otherColumns)
    {
        $a_columns = array_merge($primaryKeys, $otherColumns);
        $fields = array();
        $field_values = array();
        $placeholders = array();
        $types = array();
        $values = array();

        foreach ($a_columns as $k => $col) {
            $fields[] = $k;
            $placeholders[] = "%s";
            $placeholders2[] = ":$k";
            $types[] = $col[0];

            // integer auto-typecast (this casts bool values to integer)
            if ($col[0] == 'integer' && !is_null($col[1])) {
                $col[1] = (int) $col[1];
            }

            $values[] = $col[1];
            $field_values[$k] = $col[1];
        }

        $q = "REPLACE INTO " . $table . " (" . implode($fields, ",") . ") VALUES (" . implode($placeholders, ",") . ")";

        $r = $this->manipulateF($q, $types, $values);

        return $r;
    }


    public function equals($columns, $value, $type, $emptyOrNull = false)
    {
        if (!$emptyOrNull || $value != "") {
            return $columns . " = " . $this->quote($value, $type);
        } else {
            return "(" . $columns . " = '' OR $columns IS NULL)";
        }
    }


    public function getHost()
    {
        return $this->host;
    }


    public function setHost($host)
    {
        $this->host = $host;
    }


    public function getDbname()
    {
        return $this->dbname;
    }


    public function setDbname($dbname)
    {
        $this->dbname = $dbname;
    }


    public function getCharset()
    {
        return $this->charset;
    }


    public function setCharset($charset)
    {
        $this->charset = $charset;
    }


    public function getUsername()
    {
        return $this->username;
    }


    public function setUsername($username)
    {
        $this->username = $username;
    }


    public function getPassword()
    {
        return $this->password;
    }


    public function setPassword($password)
    {
        $this->password = $password;
    }


    public function getPort()
    {
        return $this->port;
    }


    public function setPort($port)
    {
        $this->port = $port;
    }


    public function setDBUser($user)
    {
        $this->setUsername($user);
    }


    public function setDBPort($port)
    {
        $this->setPort($port);
    }


    public function setDBPassword($password)
    {
        $this->setPassword($password);
    }


    public function setDBHost($host)
    {
        $this->setHost($host);
    }


    public function upper($a_exp)
    {
        return " UPPER(" . $a_exp . ") ";
    }


    public function lower($a_exp)
    {
        return " LOWER(" . $a_exp . ") ";
    }


    public function substr($a_exp, $a_pos = 1, $a_len = -1)
    {
        $lenstr = "";
        if ($a_len > -1) {
            $lenstr = ", " . $a_len;
        }

        return " SUBSTR(" . $a_exp . ", " . $a_pos . $lenstr . ") ";
    }


    public function prepareManip($query, $types = null)
    {
        return new ilPDOStatement($this->pdo->prepare($query));
    }


    public function prepare($query, $types = null, $result_types = null)
    {
        return new ilPDOStatement($this->pdo->prepare($query));
    }


    public function enableResultBuffering($a_status)
    {
        $this->pdo->setAttribute(PDO::MYSQL_ATTR_USE_BUFFERED_QUERY, $a_status);
    }


    public function execute($stmt, $data = array())
    {
        $result = $stmt->execute($data);
        if ($result === false) {
            throw new ilDatabaseException(implode(', ', $stmt->errorInfo()), $stmt->errorCode());
        }
        return $stmt;
    }


    public function supportsSlave()
    {
        return false;
    }


    public function supportsFulltext()
    {
        return false;
    }


    public function supportsTransactions()
    {
        return false;
    }


    public function supports($feature)
    {
        switch ($feature) {
            case self::FEATURE_TRANSACTIONS:
                return $this->supportsTransactions();
            case self::FEATURE_FULLTEXT:
                return $this->supportsFulltext();
            case self::FEATURE_SLAVE:
                return $this->supportsSlave();
            default:
                return false;
        }
    }


    public function listTables()
    {
        return $this->manager->listTables();
    }


    public function loadModule($module)
    {
        switch ($module) {
            case ilDBConstants::MODULE_MANAGER:
                return $this->manager;
            case ilDBConstants::MODULE_REVERSE:
                return $this->reverse;
        }
    }


    public function getAllowedAttributes()
    {
        return $this->field_definition->getAllowedAttributes();
    }


    public function sequenceExists($sequence)
    {
        return in_array($sequence, $this->listSequences());
    }


    public function listSequences()
    {
        return $this->manager->listSequences();
    }


    public function concat(array $values, $allow_null = true)
    {
        return $this->manager->getQueryUtils()->concat($values, $allow_null);
    }


    protected function appendLimit($query)
    {
        if ($this->limit !== null && $this->offset !== null) {
            $query .= ' LIMIT ' . (int) $this->offset . ', ' . (int) $this->limit;
            $this->limit = null;
            $this->offset = null;

            return $query;
        }

        return $query;
    }


    public function locate($a_needle, $a_string, $a_start_pos = 1)
    {
        return $this->manager->getQueryUtils()->locate($a_needle, $a_string, $a_start_pos);
    }


    public function modifyTableColumn($table, $a_column, $a_attributes)
    {
        $def = $this->reverse->getTableFieldDefinition($table, $a_column);

        $analyzer = new ilDBAnalyzer($this);
        $best_alt = $analyzer->getBestDefinitionAlternative($def);
        $def = $def[$best_alt];
        unset($def["nativetype"]);
        unset($def["mdb2type"]);

        // check attributes
        $ilDBPdoFieldDefinition = $this->field_definition;

        $type = ($a_attributes["type"] != "") ? $a_attributes["type"] : $def["type"];
        foreach ($def as $k => $v) {
            if ($k != "type" && !$ilDBPdoFieldDefinition->isAllowedAttribute($k, $type)) {
                unset($def[$k]);
            }
        }
        $check_array = $def;
        foreach ($a_attributes as $k => $v) {
            $check_array[$k] = $v;
        }
        if (!$this->checkColumnDefinition($check_array, true)) {
            throw new ilDatabaseException("ilDB Error: modifyTableColumn(" . $table . ", " . $a_column . ")");
        }

        foreach ($a_attributes as $a => $v) {
            $def[$a] = $v;
        }

        $a_attributes["definition"] = $def;

        $changes = array(
            "change" => array(
                $a_column => $a_attributes,
            ),
        );

        return $this->manager->alterTable($table, $changes, false);
    }


    public function free($a_st)
    {
        return $a_st->closeCursor();
    }


    public function renameTable($a_name, $a_new_name)
    {
        // check table name
        try {
            $this->checkTableName($a_new_name);
        } catch (ilDatabaseException $e) {
            throw new ilDatabaseException("ilDB Error: renameTable(" . $a_name . "," . $a_new_name . ")<br />" . $e->getMessage());
        }

        $this->manager->alterTable($a_name, array("name" => $a_new_name), false);

        // The abstraction_progress is no longer used in ILIAS, see http://www.ilias.de/mantis/view.php?id=19513
        //              $query = "UPDATE abstraction_progress " . "SET table_name = " . $this->quote($a_new_name, 'text') . " " . "WHERE table_name = "
        //                       . $this->quote($a_name, 'text');
        //              $this->pdo->query($query);

        return true;
    }


    public function checkTableName($a_name)
    {
        return $this->field_definition->checkTableName($a_name);
    }


    public static function isReservedWord($a_word)
    {
        require_once('./Services/Database/classes/PDO/FieldDefinition/class.ilDBPdoMySQLFieldDefinition.php');
        global $DIC;
        $ilDBPdoMySQLFieldDefinition = new ilDBPdoMySQLFieldDefinition($DIC['ilDB']);

        return $ilDBPdoMySQLFieldDefinition->isReserved($a_word);
    }


    public function beginTransaction()
    {
        if (!$this->supports(self::FEATURE_TRANSACTIONS)) {
            throw new ilDatabaseException("ilDB::beginTransaction: Transactions are not supported.");
        }

        return $this->pdo->beginTransaction();
    }


    public function commit()
    {
        if (!$this->supports(self::FEATURE_TRANSACTIONS)) {
            throw new ilDatabaseException("ilDB::beginTransaction: Transactions are not supported.");
        }

        return $this->pdo->commit();
    }


    public function rollback()
    {
        if (!$this->supports(self::FEATURE_TRANSACTIONS)) {
            throw new ilDatabaseException("ilDB::beginTransaction: Transactions are not supported.");
        }

        return $this->pdo->rollBack();
    }


    public function dropIndex($a_table, $a_name = "i1")
    {
        return $this->manager->dropIndex($a_table, $a_name);
    }


    public function setStorageEngine($storage_engine)
    {
        $this->storage_engine = $storage_engine;
    }


    public function getStorageEngine()
    {
        return $this->storage_engine;
    }


    public function queryCol($query, $type = PDO::FETCH_ASSOC, $colnum = 0)
    {
        switch ($type) {
            case ilDBConstants::FETCHMODE_ASSOC:
                $type = PDO::FETCH_ASSOC;
                break;
            case ilDBConstants::FETCHMODE_OBJECT:
                $type = PDO::FETCH_OBJ;
                break;
            default:
                $type = PDO::FETCH_ASSOC;
                break;
        }

        return $this->pdo->query($query, PDO::FETCH_ASSOC)->fetchAll(PDO::FETCH_COLUMN, $colnum);
    }


    public function queryRow($query, $types = null, $fetchmode = ilDBConstants::FETCHMODE_DEFAULT)
    {
        switch ($fetchmode) {
            case ilDBConstants::FETCHMODE_ASSOC:
                $type = PDO::FETCH_ASSOC;
                break;
            case ilDBConstants::FETCHMODE_OBJECT:
                $type = PDO::FETCH_OBJ;
                break;
            default:
                $type = PDO::FETCH_ASSOC;
                break;
        }

        return $this->pdo->query($query, $type)->fetch();
    }


    public function getServerVersion($native = false)
    {
        return $this->pdo->query('SELECT VERSION()')->fetchColumn();
    }


    public function escape($value, $escape_wildcards = false)
    {
        return $value;
    }


    public function escapePattern($text)
    {
        return $text;
    }


    public function migrateAllTablesToEngine($engine = ilDBConstants::MYSQL_ENGINE_INNODB)
    {
        return array();
    }


    public function migrateAllTablesToCollation($collation = ilDBConstants::MYSQL_COLLATION_UTF8MB4)
    {
        return array();
    }


    public function supportsCollationMigration()
    {
        return false;
    }


    public function supportsEngineMigration()
    {
        return false;
    }


    public function checkIndexName($name)
    {
        return $this->getFieldDefinition()->checkIndexName($name);
    }


    public function addUniqueConstraint($table, $fields, $name = "con")
    {
        assert(is_array($fields));
        $manager = $this->manager;

        // check index name
        if (!$this->checkIndexName($name)) {
            throw new ilDatabaseException("ilDB Error: addUniqueConstraint(" . $table . "," . $name . ")");
        }

        $fields_corrected = array();
        foreach ($fields as $f) {
            $fields_corrected[$f] = array();
        }
        $definition = array(
            'unique' => true,
            'fields' => $fields_corrected,
        );

        return $manager->createConstraint($table, $this->constraintName($table, $name), $definition);
    }


    public function dropUniqueConstraint($a_table, $a_name = "con")
    {
        return $this->manager->dropConstraint($a_table, $this->constraintName($a_table, $a_name), false);
    }


    public function dropUniqueConstraintByFields($a_table, $a_fields)
    {
        $analyzer = new ilDBAnalyzer();
        $cons = $analyzer->getConstraintsInformation($a_table);
        foreach ($cons as $c) {
            if ($c["type"] == "unique" && count($a_fields) == count($c["fields"])) {
                $all_in = true;
                foreach ($a_fields as $f) {
                    if (!isset($c["fields"][$f])) {
                        $all_in = false;
                    }
                }
                if ($all_in) {
                    return $this->dropUniqueConstraint($a_table, $c['name']);
                }
            }
        }

        return false;
    }


    public function getLastInsertId()
    {
        return $this->pdo->lastInsertId();
    }


    public function buildAtomQuery()
    {
        require_once('./Services/Database/classes/Atom/class.ilAtomQueryLock.php');

        return new ilAtomQueryLock($this);
    }


    public function uniqueConstraintExists($table, array $fields)
    {
        require_once('./Services/Database/classes/class.ilDBAnalyzer.php');
        $analyzer = new ilDBAnalyzer();
        $cons = $analyzer->getConstraintsInformation($table);
        foreach ($cons as $c) {
            if ($c["type"] == "unique" && count($fields) == count($c["fields"])) {
                $all_in = true;
                foreach ($fields as $f) {
                    if (!isset($c["fields"][$f])) {
                        $all_in = false;
                    }
                }
                if ($all_in) {
                    return true;
                }
            }
        }

        return false;
    }


    public function dropPrimaryKey($table_name)
    {
        return $this->manager->dropConstraint($table_name, "PRIMARY", true);
    }


    public function executeMultiple($stmt, $a_data)
    {
        for ($i = 0, $j = count($a_data); $i < $j; $i++) {
            $stmt->execute($a_data[$i]);
        }
    }


    public function fromUnixtime($a_expr, $a_to_text = true)
    {
        return "FROM_UNIXTIME(" . $a_expr . ")";
    }


    public function unixTimestamp()
    {
        return "UNIX_TIMESTAMP()";
    }


    public function autoExecute($tablename, $fields, $mode = ilDBConstants::MDB2_AUTOQUERY_INSERT, $where = false)
    {
        $fields_values = (array) $fields;
        if ($mode == ilDBConstants::MDB2_AUTOQUERY_INSERT) {
            if (!empty($fields_values)) {
                $keys = $fields_values;
            } else {
                $keys = array();
            }
        } else {
            $keys = array_keys($fields_values);
        }
        $params = array_values($fields_values);
        if (empty($params)) {
            $query = $this->buildManipSQL($tablename, $keys, $mode, $where);
            $result = $this->pdo->query($query);
        } else {
            $stmt = $this->autoPrepare($tablename, $keys, $mode, $where, $types, $result_types);
            $this->execute($stmt);
            $this->free($stmt);
            $result = $stmt;
        }

        return $result;
    }


    protected function autoPrepare($table, $table_fields, $mode = ilDBConstants::MDB2_AUTOQUERY_INSERT, $where = false, $types = null, $result_types = ilDBConstants::MDB2_PREPARE_MANIP)
    {
        $query = $this->buildManipSQL($table, $table_fields, $mode, $where);

        return $this->prepare($query, $types, $result_types);
    }


    protected function buildManipSQL($table, $table_fields, $mode, $where = false)
    {
        if ($this->options['quote_identifier']) {
            $table = $this->quoteIdentifier($table);
        }

        if (!empty($table_fields) && $this->options['quote_identifier']) {
            foreach ($table_fields as $key => $field) {
                $table_fields[$key] = $this->quoteIdentifier($field);
            }
        }

        if ($where !== false && !is_null($where)) {
            if (is_array($where)) {
                $where = implode(' AND ', $where);
            }
            $where = ' WHERE ' . $where;
        }

        switch ($mode) {
            case ilDBConstants::MDB2_AUTOQUERY_INSERT:
                if (empty($table_fields)) {
                    throw new ilDatabaseException('Insert requires table fields');
                }
                $cols = implode(', ', $table_fields);
                $values = '?' . str_repeat(', ?', (count($table_fields) - 1));

                return 'INSERT INTO ' . $table . ' (' . $cols . ') VALUES (' . $values . ')';
                break;
            case ilDBConstants::MDB2_AUTOQUERY_UPDATE:
                if (empty($table_fields)) {
                    throw new ilDatabaseException('Update requires table fields');
                }
                $set = implode(' = ?, ', $table_fields) . ' = ?';
                $sql = 'UPDATE ' . $table . ' SET ' . $set . $where;

                return $sql;
                break;
            case ilDBConstants::MDB2_AUTOQUERY_DELETE:
                $sql = 'DELETE FROM ' . $table . $where;

                return $sql;
                break;
            case ilDBConstants::MDB2_AUTOQUERY_SELECT:
                $cols = !empty($table_fields) ? implode(', ', $table_fields) : '*';
                $sql = 'SELECT ' . $cols . ' FROM ' . $table . $where;

                return $sql;
                break;
        }

        throw new ilDatabaseException('Syntax error');
    }


    public function getDBVersion()
    {
        $d = $this->fetchObject($this->query("SELECT VERSION() AS version"));

        return ($d->version ? $d->version : 'Unknown');
    }


    public function sanitizeMB4StringIfNotSupported($query)
    {
        if (!$this->doesCollationSupportMB4Strings()) {
            $query_replaced = preg_replace(
                '/[\x{10000}-\x{10FFFF}]/u',
                ilDBConstants::MB4_REPLACEMENT,
                $query
            );
            if (!empty($query_replaced)) {
                return $query_replaced;
            }
        }

        return $query;
    }

    public function doesCollationSupportMB4Strings()
    {
        return false;
    }


    public function groupConcat($a_field_name, $a_seperator = ",", $a_order = null)
    {
        return $this->manager->getQueryUtils()->groupConcat($a_field_name, $a_seperator, $a_order);
    }

    public function cast($a_field_name, $a_dest_type)
    {
        return $this->manager->getQueryUtils()->cast($a_field_name, $a_dest_type);
    }
}