~ubuntu-branches/ubuntu/saucy/horde3/saucy

<?php
/**
 * The DataTree_sql:: class provides an SQL implementation of the Horde
 * DataTree system.
 *
 * Required parameter for the constructor:
 *   'phptype'  -- The database type (ie. 'pgsql', 'mysql, etc.).
 *   'hostspec' -- The hostname of the database server.
 *   'protocol' -- The communication protocol ('tcp', 'unix', etc.).
 *   'username' -- The username with which to connect to the database.
 *   'password' -- The password associated with 'username'.
 *   'database' -- The name of the database.
 *   'charset'  -- The charset used by the database.
 *
 * Optional values:
 *   'table'    -- The name of the data table in 'database'. Defaults to
 *                 'horde_datatree'.
 *
 * The table structure for the DataTree system is in
 * horde/scripts/db/datatree.sql.
 *
 * $Horde: framework/DataTree/DataTree/sql.php,v 1.156.2.6 2005/03/23 10:39:06 jan Exp $
 *
 * Copyright 1999-2005 Stephane Huther <shuther@bigfoot.com>
 * Copyright 2001-2005 Chuck Hagenbuch <chuck@horde.org>
 * Copyright 2005 Jan Schneider <jan@horde.org>
 *
 * See the enclosed file COPYING for license information (LGPL). If
 * you did not receive this file, see
 * http://www.fsf.org/copyleft/lgpl.html.
 *
 * @author  Chuck Hagenbuch <chuck@horde.org>
 * @author  Jan Schneider <jan@horde.org>
 * @author  Stephane Huther <shuther@bigfoot.com>
 * @since   Horde 2.1
 * @package Horde_DataTree
 */
class DataTree_sql extends DataTree {

    /**
     * Handle for the current database connection.
     *
     * @var resource $_db
     */
    var $_db;

    /**
     * The number of copies of the horde_datatree_attributes table
     * that we need to join on in the current query.
     *
     * @var integer $_tableCount
     */
    var $_tableCount = 1;

    /**
     * Constructs a new SQL DataTree object.
     *
     * @param array $params  A hash containing connection parameters.
     */
    function DataTree_sql($params)
    {
        parent::DataTree($params);
        $this->_connect();
    }

    /**
     * Returns a list of all groups (root nodes) of the data tree.
     *
     * @return array  The the group IDs
     */
    function getGroups()
    {
        $query = sprintf('SELECT DISTINCT group_uid FROM %s',
                         $this->_params['table']);

        Horde::logMessage('SQL Query by DataTree_sql::getGroups(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);

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

    /**
     * Load (a subset of) the datatree into the $_data array.
     *
     * @access private
     *
     * @param string  $root      Which portion of the tree to load.
     *                           Defaults to all of it.
     * @param boolean $loadTree  Load a tree starting at $root, or just the
     *                           requested level and direct parents?
     *                           Defaults to single level.
     * @param boolean $reload    Re-load already loaded values?
     *
     * @return mixed  True on success or a PEAR_Error on failure.
     */
    function _load($root = DATATREE_ROOT, $loadTree = false, $reload = false)
    {
        /* Do NOT use DataTree::exists() here; that would cause an
         * infinite loop. */
        if (!$reload &&
            (in_array($root, $this->_nameMap) ||
             (count($this->_data) && $root == DATATREE_ROOT))) {
            return true;
        }

        $query = $this->_buildLoadQuery($root, $loadTree, false);
        if (is_a($query, 'PEAR_Error')) {
            return $query;
        }
        if (empty($query)) {
            return true;
        }

        Horde::logMessage('SQL Query by DataTree_sql::_load(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
        $data = $this->_db->getAll($query);
        if (is_a($data, 'PEAR_Error')) {
            return $data;
        }
        return $this->set($data, $this->_params['charset']);
    }

    /**
     * Counts (a subset of) the datatree which would be loaded into
     * the $_data array if _load() is called with the same value of
     * $root.
     *
     * @access private
     *
     * @param string  $root      Which portion of the tree to load.
     *                           Defaults to all of it.
     *
     * @return integer  Number of objects
     */
    function _count($root = DATATREE_ROOT)
    {
        $query = $this->_buildLoadQuery($root, true, true);
        if (is_a($query, 'PEAR_Error')) {
            return $query;
        }
        if (empty($query)) {
            return 0;
        }
        Horde::logMessage('SQL Query by DataTree_sql::_count(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
        return $this->_db->getOne($query);
    }

    /**
     * Load (a subset of) the datatree into the $_data array.
     *
     * @access private
     *
     * @param string  $root      Which portion of the tree to load.
     *                           Defaults to all of it.
     * @param boolean $loadTree  Load a tree starting at $root, or just the
     *                           requested level and direct parents?
     *                           Defaults to single level.
     * @param integer $count     Only count objects?
     *
     * @return mixed  True on success or a PEAR_Error on failure.
     */
    function _buildLoadQuery($root = DATATREE_ROOT, $loadTree = false, $count = false)
    {
        if (!empty($root) && $root != DATATREE_ROOT) {
            if (strpos($root, ':') !== false) {
                $parts = explode(':', $root);
                $root = array_pop($parts);
            }
            $root = (string)$root;

            $query = sprintf('SELECT datatree_id, datatree_parents FROM %s' .
                             ' WHERE datatree_name = %s AND group_uid = %s ORDER BY datatree_id',
                             $this->_params['table'],
                             $this->_db->quote($root),
                             $this->_db->quote($this->_params['group']));

            Horde::logMessage('SQL Query by DataTree_sql::_buildLoadQuery(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
            $root = $this->_db->getAssoc($query);
            if (is_a($root, 'PEAR_Error')) {
                return $root;
            }
            if (count($root) == 0) {
                return '';
            }

            $where = '';
            $first_time = true;
            foreach ($root as $object_id => $object_parents) {
                $pstring = $object_parents . ':' . $object_id . '%';
                $pquery = '';
                if (!empty($object_parents)) {
                    $ids = substr($object_parents, 1);
                    $pquery = ' OR datatree_id IN (' . str_replace(':', ', ', $ids) . ')';
                }
                if ($loadTree) {
                    $pquery .= ' OR datatree_parents = ' . $this->_db->quote(substr($pstring, 0, -2));
                }

                if (!$first_time) {
                    $where .= ' OR ';
                }
                $where .= sprintf('datatree_parents LIKE %s OR datatree_id = %s%s',
                                  $this->_db->quote($pstring),
                                  $object_id,
                                  $pquery);

                $first_time = false;
            }

            $query = sprintf('SELECT %s FROM %s WHERE (%s) AND group_uid = %s',
                             $count ? 'COUNT(*)' : 'datatree_id, datatree_name, datatree_parents, datatree_order',
                             $this->_params['table'],
                             $where,
                             $this->_db->quote($this->_params['group']));
        } else {
            $query = sprintf('SELECT %s FROM %s WHERE group_uid = %s',
                             $count ? 'COUNT(*)' : 'datatree_id, datatree_name, datatree_parents, datatree_order',
                             $this->_params['table'],
                             $this->_db->quote($this->_params['group']));
        }

        return $query;
    }

    /**
     * Load a set of objects identified by their unique IDs, and their
     * parents, into the $_data array.
     *
     * @access private
     *
     * @param mixed $cids  The unique ID of the object to load, or an array of
     *                     object ids.
     *
     * @return mixed  True on success or a PEAR_Error on failure.
     */
    function _loadById($cids)
    {
        /* Make sure we have an array. */
        if (!is_array($cids)) {
            $cids = array($cids);
        }

        /* Bail out now if there's nothing to load. */
        if (!count($cids)) {
            return true;
        }

        /* Don't load any that are already loaded. Also, make sure
         * that everything in the $ids array that we are building is
         * an integer. */
        $ids = array();
        foreach ($cids as $cid) {
            /* Do NOT use DataTree::exists() here; that would cause an
             * infinite loop. */
            if (!isset($this->_data[$cid])) {
                $ids[] = (int)$cid;
            }
        }

        /* If there are none left to load, return. */
        if (!count($ids)) {
            return true;
        }

        $in = array_search(DATATREE_ROOT, $ids) === false ? sprintf('datatree_id IN (%s) AND ', implode(', ', $ids)) : '';
        $query = sprintf('SELECT datatree_id, datatree_parents FROM %s' .
                         ' WHERE %sgroup_uid = %s' .
                         ' ORDER BY datatree_id',
                         $this->_params['table'],
                         $in,
                         $this->_db->quote($this->_params['group']));
        Horde::logMessage('SQL Query by DataTree_sql::_loadById(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
        $parents = $this->_db->getAssoc($query);
        if (is_a($parents, 'PEAR_Error')) {
            return $parents;
        }

        $ids = array();
        foreach ($parents as $cid => $parent) {
            $ids[(int)$cid] = (int)$cid;

            $pids = explode(':', substr($parent, 1));
            foreach ($pids as $pid) {
                $pid = (int)$pid;
                if (!isset($this->_data[$pid])) {
                    $ids[$pid] = $pid;
                }
            }
        }

        /* If $ids is empty, we have nothing to load. */
        if (!count($ids)) {
            return true;
        }

        $query = sprintf('SELECT datatree_id, datatree_name, datatree_parents, datatree_order FROM %s' .
                         ' WHERE datatree_id IN (%s)'.
                         ' AND group_uid = %s ORDER BY datatree_id',
                         $this->_params['table'],
                         implode(', ', $ids),
                         $this->_db->quote($this->_params['group']));

        Horde::logMessage('SQL Query by DataTree_sql::_loadById(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
        $data = $this->_db->getAll($query);
        if (is_a($data, 'PEAR_Error')) {
            return $data;
        }

        return $this->set($data, $this->_params['charset']);
    }

    /**
     * Adds an object.
     *
     * @param mixed $object     The object to add (string or DataTreeObject).
     * @param bool $id_as_name  Whether the object ID is to be used as object
     *                          name.  Used in situations where there is no
     *                          available unique input for object name.
     *                          Defaults to false.
     */
    function add($object, $id_as_name = false)
    {
        $attributes = false;
        if (is_a($object, 'DataTreeObject')) {
            $fullname = $object->getName();
            $order = $object->order;

            /* We handle data differently if we can map it to the
             * horde_datatree_attributes table. */
            if (method_exists($object, '_toAttributes')) {
                $data = '';
                $ser = null;

                /* Set a flag for later so that we know to insert the
                 * attribute rows. */
                $attributes = true;
            } else {
                require_once 'Horde/Serialize.php';
                $ser = SERIALIZE_UTF7_BASIC;
                $data = Horde_Serialize::serialize($object->getData(), $ser, NLS::getCharset());
            }
        } else {
            $fullname = $object;
            $order = null;
            $data = '';
            $ser = null;
        }

        /* Get the next unique ID. */
        $id = $this->_db->nextId($this->_params['table']);
        if (is_a($id, 'PEAR_Error')) {
            Horde::logMessage($id, __FILE__, __LINE__, PEAR_LOG_ERR);
            return $id;
        }

        if (strpos($fullname, ':') !== false) {
            $parts = explode(':', $fullname);
            $parents = '';
            $pstring = '';
            if ($id_as_name) {
                /* Requested use of ID as name, so discard current name. */
                array_pop($parts);
                /* Set name to ID. */
                $name = $id;
                /* Modify fullname to reflect new name. */
                $fullname = implode(':', $parts) . ':' . $id;
                if (is_a($object, 'DataTreeObject')) {
                    $object->setName($fullname);
                } else {
                    $object = $fullname;
                }
            } else {
                $name = array_pop($parts);
            }
            foreach ($parts as $par) {
                $pstring .= (empty($pstring) ? '' : ':') . $par;
                $pid = $this->getId($pstring);
                if (is_a($pid, 'PEAR_Error')) {
                    /* Auto-create parents. */
                    $pid = $this->add($pstring);
                    if (is_a($pid, 'PEAR_Error')) {
                        return $pid;
                    }
                }
                $parents .= ':' . $pid;
            }
        } else {
            if ($id_as_name) {
                /* Requested use of ID as name, set fullname and name to ID. */
                $fullname = $id;
                $name = $id;
                if (is_a($object, 'DataTreeObject')) {
                    $object->setName($fullname);
                } else {
                    $object = $fullname;
                }
            } else {
                $name = $fullname;
            }
            $parents = '';
            $pid = DATATREE_ROOT;
        }

        if (parent::exists($fullname)) {
            return PEAR::raiseError(_("Already exists"));
        }

        $query = sprintf('INSERT INTO %s (datatree_id, group_uid, datatree_name, datatree_order, datatree_data, user_uid, datatree_serialized, datatree_parents)' .
                         ' VALUES (%s, %s, %s, %s, %s, %s, %s, %s)',
                         $this->_params['table'],
                         (int)$id,
                         $this->_db->quote($this->_params['group']),
                         $this->_db->quote(String::convertCharset($name, NLS::getCharset(), $this->_params['charset'])),
                         is_null($order) ? 'NULL' : (int)$order,
                         $this->_db->quote($data),
                         $this->_db->quote((string)Auth::getAuth()),
                         (int)$ser,
                         $this->_db->quote($parents));

        Horde::logMessage('SQL Query by DataTree_sql::add(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
        $result = $this->_db->query($query);
        if (is_a($result, 'PEAR_Error')) {
            Horde::logMessage($result, __FILE__, __LINE__, PEAR_LOG_ERR);
            return $result;
        }

        $reorder = $this->reorder($parents, $order, $id);
        if (is_a($reorder, 'PEAR_Error')) {
            Horde::logMessage($reorder, __FILE__, __LINE__, PEAR_LOG_ERR);
            return $reorder;
        }

        $result = parent::_add($fullname, $id, $pid, $order);
        if (is_a($result, 'PEAR_Error')) {
            return $result;
        }

        /* If we succesfully inserted the object and it supports
         * being mapped to the attributes table, do that now: */
        if (!empty($attributes)) {
            $result = $this->updateData($object);
            if (is_a($result, 'PEAR_Error')) {
                return $result;
            }
        }

        return $id;
    }

    /**
     * Changes the order of the children of an object.
     *
     * @param string $parent  The full id path of the parent object.
     * @param mixed $order    If an array it specifies the new positions for
     *                        all child objects.
     *                        If an integer and $cid is specified, the position
     *                        where the child specified by $cid is inserted. If
     *                        $cid is not specified, the position gets deleted,
     *                        causing the following positions to shift up.
     * @param integer $cid    See $order.
     */
    function reorder($parent, $order = null, $cid = null)
    {
        if (!$parent || is_a($parent, 'PEAR_Error')) {
            // Abort immediately if the parent string is empty; we
            // cannot safely reorder all top-level elements.
            return;
        }

        $pquery = '';
        if (!is_array($order) && !is_null($order)) {
            /* Single update (add/del). */
            if (is_null($cid)) {
                /* No object id given so shuffle down. */
                $direction = '-';
            } else {
                /* We have an object id so shuffle up. */
                $direction = '+';

                /* Leaving the newly inserted object alone. */
                $pquery = sprintf(' AND datatree_id != %s', (int)$cid);
            }
            $query = sprintf('UPDATE %s SET datatree_order = datatree_order %s 1 WHERE group_uid = %s AND datatree_parents = %s AND datatree_order >= %s',
                             $this->_params['table'],
                             $direction,
                             $this->_db->quote($this->_params['group']),
                             $this->_db->quote($parent),
                             is_null($order) ? 'NULL' : (int)$order) . $pquery;

            Horde::logMessage('SQL Query by DataTree_sql::reorder(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
            $result = $this->_db->query($query);

        } elseif (is_array($order)) {
            /* Multi update. */
            $query = sprintf('SELECT COUNT(datatree_id) FROM %s WHERE group_uid = %s AND datatree_parents = %s GROUP BY datatree_parents',
                             $this->_params['table'],
                             $this->_db->quote($this->_params['group']),
                             $this->_db->quote($parent));

            Horde::logMessage('SQL Query by DataTree_sql::reorder(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);

            $result = $this->_db->getOne($query);
            if (is_a($result, 'PEAR_Error')) {
                return $result;
            } elseif (count($order) != $result) {
                return PEAR::raiseError(_("Cannot reorder, number of entries supplied for reorder does not match number stored."));
            }

            $o_key = 0;
            foreach ($order as $o_cid) {
                $query = sprintf('UPDATE %s SET datatree_order = %s WHERE datatree_id = %s',
                                 $this->_params['table'],
                                 (int)$o_key,
                                 is_null($o_cid) ? 'NULL' : (int)$o_cid);

                Horde::logMessage('SQL Query by DataTree_sql::reorder(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
                $result = $this->_db->query($query);
                if (is_a($result, 'PEAR_Error')) {
                    return $result;
                }

                $o_key++;
            }

            $pid = $this->getId($parent);

            /* Re-order our cache. */
            return $this->_reorder($pid, $order);
        }
    }

    /**
     * Explicitly set the order for a datatree object.
     *
     * @param integer $id     The datatree object id to change.
     * @param integer $order  The new order.
     */
    function setOrder($id, $order)
    {
        $query = sprintf('UPDATE %s SET datatree_order = %s WHERE datatree_id = %s',
                         $this->_params['table'],
                         is_null($order) ? 'NULL' : (int)$order,
                         (int)$id);

        Horde::logMessage('SQL Query by DataTree_sql::setOrder(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
        return $this->_db->query($query);
    }

    /**
     * Removes an object.
     *
     * @param mixed   $object  The object to remove.
     * @param boolean $force   Force removal of every child object?
     */
    function remove($object, $force = false)
    {
        $id = $this->getId($object);
        $order = $this->getOrder($object);

        $query = sprintf('SELECT datatree_id FROM %s ' .
                         ' WHERE group_uid = %s AND datatree_parents LIKE %s' .
                         ' ORDER BY datatree_id',
                         $this->_params['table'],
                         $this->_db->quote($this->_params['group']),
                         $this->_db->quote('%:' . (int)$id . ''));

        Horde::logMessage('SQL Query by DataTree_sql::remove(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
        $children = $this->_db->getAll($query, DB_FETCHMODE_ASSOC);

        if (count($children)) {
            if ($force) {
                foreach ($children as $child) {
                    $cat = $this->getName($child['datatree_id']);
                    $result = $this->remove($cat, true);
                    if (is_a($result, 'PEAR_Error')) {
                        return $result;
                    }
                }
            } else {
                return PEAR::raiseError(sprintf(_("Cannot remove, %d children exist."), count($children)));
            }
        }

        /* Remove attributes for this object. */
        $query = sprintf('DELETE FROM %s WHERE datatree_id = %s',
                         $this->_params['table_attributes'],
                         (int)$id);

        Horde::logMessage('SQL Query by DataTree_sql::remove(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
        $result = $this->_db->query($query);
        if (is_a($result, 'PEAR_Error')) {
            return $result;
        }

        $query = sprintf('DELETE FROM %s WHERE datatree_id = %s',
                         $this->_params['table'],
                         (int)$id);

        Horde::logMessage('SQL Query by DataTree_sql::remove(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
        $result = $this->_db->query($query);
        if (is_a($result, 'PEAR_Error')) {
            return $result;
        }

        $parents = $this->getParentIdString($object);
        if (is_a($parents, 'PEAR_Error')) {
            return $parents;
        }

        $reorder = $this->reorder($parents, $order);
        if (is_a($reorder, 'PEAR_Error')) {
            return $reorder;
        }

        return is_a(parent::remove($object), 'PEAR_Error') ? $id : true;
    }

    /**
     * Remove one or more objects by id. This function does *not* do
     * the validation, reordering, etc. that remove() does. If you
     * need to check for children, re-do ordering, etc., then you must
     * remove() objects one-by-one. This is for code that knows it's
     * dealing with single (non-parented) objects and needs to delete
     * a batch of them quickly.
     *
     * @param array $ids  The objects to remove.
     */
    function removeByIds($ids)
    {
        /* Sanitize input. */
        if (!is_array($ids)) {
            $ids = array((int)$ids);
        } else {
            foreach ($ids as $key => $id) {
                $ids[$key] = (int)$id;
            }
        }

        /* Remove attributes for $ids. */
        $query = sprintf('DELETE FROM %s WHERE datatree_id IN (%s)',
                         $this->_params['table_attributes'],
                         implode(',', $ids));

        Horde::logMessage('SQL Query by DataTree_sql::removeByIds(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
        $result = $this->_db->query($query);
        if (is_a($result, 'PEAR_Error')) {
            return $result;
        }

        $query = sprintf('DELETE FROM %s WHERE datatree_id IN %s',
                         $this->_params['table'],
                         implode(',', $ids));

        Horde::logMessage('SQL Query by DataTree_sql::removeByIds(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
        return $this->_db->query($query);
    }

    /**
     * Remove one or more objects by name. This function does *not* do
     * the validation, reordering, etc. that remove() does. If you
     * need to check for children, re-do ordering, etc., then you must
     * remove() objects one-by-one. This is for code that knows it's
     * dealing with single (non-parented) objects and needs to delete
     * a batch of them quickly.
     *
     * @param array $names  The objects to remove.
     */
    function removeByNames($names)
    {
        /* Sanitize input. */
        if (!is_array($names)) {
            $names = array($this->_db->quote($names));
        } else {
            foreach ($names as $key => $name) {
                $names[$key] = $this->_db->quote($name);
            }
        }

        /* Remove attributes for $names. */
        $query = sprintf('DELETE FROM %s WHERE datatree_name IN (%s)',
                         $this->_params['table_attributes'],
                         implode(',', $names));

        Horde::logMessage('SQL Query by DataTree_sql::removeByNames(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
        $result = $this->_db->query($query);
        if (is_a($result, 'PEAR_Error')) {
            return $result;
        }

        $query = sprintf('DELETE FROM %s WHERE datatree_name IN %s',
                         $this->_params['table'],
                         implode(',', $names));

        Horde::logMessage('SQL Query by DataTree_sql::removeByNames(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
        return $this->_db->query($query);
    }

    /**
     * Move an object to a new parent.
     *
     * @param mixed  $object     The object to move.
     * @param string $newparent  The new parent object. Defaults to the root.
     */
    function move($object, $newparent = null)
    {
        $old_parent_path = $this->getParentIdString($object);
        $result = parent::move($object, $newparent);
        if (is_a($result, 'PEAR_Error')) {
            return $result;
        }
        $id = $this->getId($object);
        $new_parent_path = $this->getParentIdString($object);

        /* Fetch the object being moved and all of its children, since
         * we also need to update their parent paths to avoid creating
         * orphans. */
        $query = sprintf('SELECT datatree_id, datatree_parents FROM %s' .
                         ' WHERE datatree_parents = %s OR datatree_parents LIKE %s OR datatree_id = %s',
                         $this->_params['table'],
                         $this->_db->quote($old_parent_path . ':' . $id),
                         $this->_db->quote($old_parent_path . ':' . $id . ':%'),
                         (int)$id);

        Horde::logMessage('SQL Query by DataTree_sql::move(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
        $rowset = $this->_db->query($query);
        if (is_a($rowset, 'PEAR_Error')) {
            return $rowset;
        }

        /* Update each object, replacing the old parent path with the
         * new one. */
        while ($row = $rowset->fetchRow(DB_FETCHMODE_ASSOC)) {
            if (is_a($row, 'PEAR_Error')) {
                return $row;
            }

            $oquery = '';
            if ($row['datatree_id'] == $id) {
                $oquery = ', datatree_order = 0 ';
            }

            /* Do str_replace() only if this is not a first level
             * object. */
            if (!empty($row['datatree_parents'])) {
                $ppath = str_replace($old_parent_path, $new_parent_path, $row['datatree_parents']);
            } else {
                $ppath = $new_parent_path;
            }
            $query = sprintf('UPDATE %s SET datatree_parents = %s' . $oquery . ' WHERE datatree_id = %s',
                             $this->_params['table'],
                             $this->_db->quote($ppath),
                             (int)$row['datatree_id']);

            Horde::logMessage('SQL Query by DataTree_sql::move(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
            $result = $this->_db->query($query);
            if (is_a($result, 'PEAR_Error')) {
                return $result;
            }
        }

        $order = $this->getOrder($object);

        /* Shuffle down the old order positions. */
        $reorder = $this->reorder($old_parent_path, $order);

        /* Shuffle up the new order positions. */
        $reorder = $this->reorder($new_parent_path, 0, $id);

        return true;
    }

    /**
     * Change an object's name.
     *
     * @param mixed  $old_object       The old object.
     * @param string $new_object_name  The new object name.
     */
    function rename($old_object, $new_object_name)
    {
        /* Do the cache renaming first */
        $result = parent::rename($old_object, $new_object_name);
        if (is_a($result, 'PEAR_Error')) {
            return $result;
        }

        /* Get the object id and set up the sql query. */
        $id = $this->getId($old_object);
        $query = sprintf('UPDATE %s SET datatree_name = %s' .
                         ' WHERE datatree_id = %s',
                         $this->_params['table'],
                         $this->_db->quote(String::convertCharset($new_object_name, NLS::getCharset(), $this->_params['charset'])),
                         (int)$id);

        Horde::logMessage('SQL Query by DataTree_sql::rename(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
        $result = $this->_db->query($query);

        return is_a($result, 'PEAR_Error') ? $result : true;
    }

    /**
     * Retrieve data for an object from the datatree_data field.
     *
     * @param integer $cid  The object id to fetch, or an array of object ids.
     */
    function getData($cid)
    {
        require_once 'Horde/Serialize.php';

        if (is_array($cid)) {
            if (!count($cid)) {
                return array();
            }

            $query = sprintf('SELECT datatree_id, datatree_data, datatree_serialized FROM %s WHERE datatree_id IN (%s)',
                             $this->_params['table'],
                             implode(', ', $cid));

            Horde::logMessage('SQL Query by DataTree_sql::getData(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
            $result = $this->_db->getAssoc($query);
            if (is_a($result, 'PEAR_Error')) {
                Horde::logMessage($result, __FILE__, __LINE__, PEAR_LOG_ERR);
                return $result;
            }

            $data = array();
            foreach ($result as $id => $row) {
                $data[$id] = Horde_Serialize::unserialize($row[0], $row[1], NLS::getCharset());
                /* Convert old data to the new format. */
                if ($row[1] == SERIALIZE_BASIC) {
                    $data[$id] = String::convertCharset($data[$id], NLS::getCharset(true));
                }

                $data[$id] = (is_null($data[$id]) || !is_array($data[$id])) ? array() : $data[$id];
            }

            return $data;
        } else {
            $query = sprintf('SELECT datatree_data, datatree_serialized FROM %s WHERE datatree_id = %s',
                             $this->_params['table'],
                             (int)$cid);

            Horde::logMessage('SQL Query by DataTree_sql::getData(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
            $row = $this->_db->getRow($query, DB_FETCHMODE_ASSOC);

            $data = Horde_Serialize::unserialize($row['datatree_data'], $row['datatree_serialized'], NLS::getCharset());
            /* Convert old data to the new format. */
            if ($row['datatree_serialized'] == SERIALIZE_BASIC) {
                $data = String::convertCharset($data, NLS::getCharset(true));
            }
            return (is_null($data) || !is_array($data)) ? array() : $data;
        }
    }

    /**
     * Retrieve data for an object from the horde_datatree_attributes
     * table.
     *
     * @param integer | array $cid  The object id to fetch,
     *                              or an array of object ids.
     *
     * @param array $keys  The attributes keys to fetch.
     *
     * @return array  A hash of attributes, or a multi-level hash
     *                of object ids => their attributes.
     */
    function getAttributes($cid, $keys = false)
    {
        if ($keys) {
            $filter = sprintf(' AND attribute_key IN (\'%s\')', implode("', '", $keys));
        } else {
            $filter = '';
        }

        if (is_array($cid)) {
            $query = sprintf('SELECT datatree_id, attribute_name AS name, attribute_key AS "key", attribute_value AS value FROM %s WHERE datatree_id IN (%s)%s',
                             $this->_params['table_attributes'],
                             implode(', ', $cid),
                             $filter);

            Horde::logMessage('SQL Query by DataTree_sql::getAttributes(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
            $rows = $this->_db->getAll($query, DB_FETCHMODE_ASSOC);
            if (is_a($rows, 'PEAR_Error')) {
                return $rows;
            }

            $data = array();
            foreach ($rows as $row) {
                if (empty($data[$row['datatree_id']])) {
                    $data[$row['datatree_id']] = array();
                }
                $data[$row['datatree_id']][] = array('name' => $row['name'],
                                                     'key' => $row['key'],
                                                     'value' => String::convertCharset($row['value'], $this->_params['charset'], NLS::getCharset()));
            }
            return $data;
        } else {
            $query = sprintf('SELECT attribute_name AS name, attribute_key AS "key", attribute_value AS value FROM %s WHERE datatree_id = %s%s',
                             $this->_params['table_attributes'],
                             (int)$cid,
                             $filter);

            Horde::logMessage('SQL Query by DataTree_sql::getAttributes(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
            $rows = $this->_db->getAll($query, DB_FETCHMODE_ASSOC);
            for ($i = 0; $i < count($rows); $i++) {
                $rows[$i]['value'] = String::convertCharset($rows[$i]['value'], $this->_params['charset'], NLS::getCharset());
            }
            return $rows;
        }
    }

    /**
     * Returns the number of objects matching a set of attribute
     * criteria.
     *
     * @see buildAttributeQuery()
     *
     * @param array   $criteria   The array of criteria.
     * @param string  $parent     The parent node to start searching from.
     * @param boolean $allLevels  Return all levels, or just the direct
     *                            children of $parent? Defaults to all levels.
     * @param string  $restrict   Only return attributes with the same
     *                            attribute_name or attribute_id.
     */
    function countByAttributes($criteria, $parent = DATATREE_ROOT, $allLevels = true, $restrict = 'name')
    {
        if (!count($criteria)) {
            return 0;
        }

        list($query, $values) = $this->buildAttributeQuery($criteria, $parent, $allLevels, $restrict, true);

        Horde::logMessage('SQL Query by DataTree_sql::countByAttributes(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
        $stmt = $this->_db->prepare($query);
        if (is_a($stmt, 'PEAR_Error')) {
            Horde::logMessage($stmt, __FILE__, __LINE__, PEAR_LOG_ERR);
            return $stmt;
        }
        $result = $this->_db->execute($stmt, $values);
        if (is_a($result, 'PEAR_Error')) {
            Horde::logMessage($result, __FILE__, __LINE__, PEAR_LOG_ERR);
            return $result;
        }
        $row = &$result->fetchRow();
        if (is_a($row, 'PEAR_Error')) {
            Horde::logMessage($row, __FILE__, __LINE__, PEAR_LOG_ERR);
            return $row;
        }

        return $row[0];
    }

    /**
     * Returns a set of object ids based on a set of attribute criteria.
     *
     * @see buildAttributeQuery()
     *
     * @param array   $criteria   The array of criteria.
     * @param string  $parent     The parent node to start searching from.
     * @param boolean $allLevels  Return all levels, or just the direct
     *                            children of $parent? Defaults to all levels.
     * @param string  $restrict   Only return attributes with the same
     *                            attribute_name or attribute_id.
     * @param integer $from       The object to start to fetching
     * @param integer $count      The number of objects to fetch
     */
    function getByAttributes($criteria, $parent = DATATREE_ROOT, $allLevels = true, $restrict = 'name', $from = 0, $count = 0)
    {
        if (!count($criteria)) {
            return PEAR::raiseError('no criteria');
        }

        list($query, $values) = $this->buildAttributeQuery($criteria, $parent, $allLevels, $restrict);

        if ($count) {
            $query = $this->_db->modifyLimitQuery($query, $from, $count);
            if (is_a($query, 'PEAR_Error')) {
                return $query;
            }
        }

        Horde::logMessage('SQL Query by DataTree_sql::getByAttributes(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
        $stmt = $this->_db->prepare($query);
        if (is_a($stmt, 'PEAR_Error')) {
            Horde::logMessage($stmt, __FILE__, __LINE__, PEAR_LOG_ERR);
            return $stmt;
        }
        $result = $this->_db->execute($stmt, $values);
        if (is_a($result, 'PEAR_Error')) {
            Horde::logMessage($result, __FILE__, __LINE__, PEAR_LOG_ERR);
            return $result;
        }
        $rows = array();
        while ($row = &$result->fetchRow()) {
            $rows[$row[0]] = $row[1];
        }

        return $rows;
    }

    /**
     * Builds an attribute query.
     *
     * @param array   $criteria   The array of criteria. Example:
     *                            $criteria['OR'] = array(
     *                                array('AND' => array(
     *                                    array('field' => 'name',
     *                                          'op'    => '=',
     *                                          'test'  => 'foo'),
     *                                    array('field' => 'key',
     *                                          'op'    => '=',
     *                                          'test'  => 'abc'))),
     *                                array('AND' => array(
     *                                    array('field' => 'name',
     *                                          'op'    => '=',
     *                                          'test'  => 'bar'),
     *                                    array('field' => 'key',
     *                                          'op'    => '=',
     *                                          'test'  => 'xyz'))));
     *                            This would fetch all object ids where
     *                            attribute name is "foo" AND key is "abc", OR
     *                            "bar" AND "xyz".
     * @param string  $parent     The parent node to start searching from.
     * @param boolean $allLevels  Return all levels, or just the direct
     *                            children of $parent? Defaults to all levels.
     * @param string  $restrict   Only return attributes with the same
     *                            attribute_name or attribute_id.
     * @param boolean $count      Just count the number of elements that'd match.
     *
     * @return array  An SQL query and a list of values suitable for binding
     *                as an array.
     */
    function buildAttributeQuery($criteria, $parent = DATATREE_ROOT, $allLevels = true, $restrict = 'name', $count = false)
    {
        if (!count($criteria)) {
            return 0;
        }

        /* Build the query. */
        $this->_tableCount = 1;
        $query = '';
        $values = array();
        foreach ($criteria as $key => $vals) {
            if ($key == 'OR' || $key == 'AND') {
                if (!empty($query)) {
                    $query .= ' ' . $key . ' ';
                }
                $binds = $this->_buildAttributeQuery($key, $vals);
                $query .= '(' . $binds[0] . ')';
                $values += $binds[1];
            }
        }

        // Add filtering by parent, and for one or all levels.
        $levelQuery = '';
        $levelValues = array();
        if ($parent != DATATREE_ROOT) {
            $parts = explode(':', $parent);
            $parents = '';
            $pstring = '';
            foreach ($parts as $part) {
                $pstring .= (empty($pstring) ? '' : ':') . $part;
                $pid = $this->getId($pstring);
                if (is_a($pid, 'PEAR_Error')) {
                    return $pid;
                }
                $parents .= ':' . $pid;
            }

            if ($allLevels) {
                $levelQuery = 'AND (datatree_parents = ? OR datatree_parents LIKE ?)';
                $levelValues = array($parents, $parents . ':%');
            } else {
                $levelQuery = 'AND datatree_parents = ?';
                $levelValues = array($parents);
            }
        } elseif (!$allLevels) {
            $levelQuery = "AND datatree_parents = ''";
        }

        // Build the FROM/JOIN clauses.
        $joins = array();
        $pairs = array();
        for ($i = 1; $i <= $this->_tableCount; $i++) {
            $joins[] = 'LEFT JOIN ' . $this->_params['table_attributes'] . ' a' . $i . ' ON a' . $i . '.datatree_id = c.datatree_id';

            if ($i != 1) {
                if ($restrict == 'name') {
                    $pairs[] = 'AND a1.attribute_name = a' . $i . '.attribute_name';
                } elseif ($restrict == 'id') {
                    $pairs[] = 'AND a1.datatree_id = a' . $i . '.datatree_id';
                }
            }
        }
        $joins = implode(' ', $joins);
        $pairs = implode(' ', $pairs);

        return array(sprintf('SELECT %s FROM %s c %s WHERE c.group_uid = ? AND %s %s %s %s',
                             $count ? 'COUNT(DISTINCT c.datatree_id)' : 'c.datatree_id, c.datatree_name',
                             $this->_params['table'],
                             $joins,
                             $query,
                             $levelQuery,
                             $pairs,
                             $count ? '' : 'GROUP BY c.datatree_id, c.datatree_name, c.datatree_order ORDER BY c.datatree_order, c.datatree_name, c.datatree_id'),
                     array_merge(array($this->_params['group']), $values, $levelValues));
    }

    /**
     * Build a piece of an attribute query.
     *
     * @param string $glue     The glue to join the criteria (OR/AND).
     * @param array $criteria  The array of criteria.
     * @param boolean $join    Should we join on a clean
     *                         horde_datatree_attributes table? Defaults to
     *                         false.
     *
     * @return array  An SQL fragment and a list of values suitable for binding
     *                as an array.
     */
    function _buildAttributeQuery($glue, $criteria, $join = false)
    {
        require_once 'Horde/SQL.php';

        // Initialize the clause that we're building.
        $clause = '';
        $values = array();

        // Get the table alias to use for this set of criteria.
        $alias = $this->_getAlias($join);

        foreach ($criteria as $key => $vals) {
            if (!empty($vals['OR']) || !empty($vals['AND'])) {
                if (!empty($clause)) {
                    $clause .= ' ' . $glue . ' ';
                }
                $binds = $this->_buildAttributeQuery($glue, $vals);
                $clause .= '(' . $binds[0] . ')';
                $values = array_merge($values, $binds[1]);
            } elseif (!empty($vals['JOIN'])) {
                if (!empty($clause)) {
                    $clause .= ' ' . $glue . ' ';
                }
                $binds = $this->_buildAttributeQuery($glue, $vals['JOIN'], true);
                $clause .= $binds[0];
                $values = array_merge($values, $binds[1]);
            } else {
                if (isset($vals['field'])) {
                    if (!empty($clause)) {
                        $clause .= ' ' . $glue . ' ';
                    }
                    $binds = Horde_SQL::buildClause($this->_db, $alias . '.attribute_' . $vals['field'], $vals['op'], $vals['test'], true);
                    $clause .= $binds[0];
                    $values = array_merge($values, $binds[1]);
                } else {
                    if (!empty($clause)) {
                        $clause .= ' ' . $glue . ' ';
                    }
                    $binds = $this->_buildAttributeQuery($key, $vals);
                    $clause .= $binds[0];
                    $values = array_merge($values, $binds[1]);
                }
            }
        }

        return array($clause, $values);
    }

    /**
     * Get an alias to horde_datatree_attributes, incrementing it if
     * necessary.
     *
     * @param boolean $increment  Increment the alias count? Defaults to no.
     */
    function _getAlias($increment = false)
    {
        static $seen = array();

        if ($increment && !empty($seen[$this->_tableCount])) {
            $this->_tableCount++;
        }

        $seen[$this->_tableCount] = true;
        return 'a' . $this->_tableCount;
    }

    /**
     * Update the data in an object. Does not change the object's
     * parent or name, just serialized data or attributes.
     *
     * @param DataTree $object  A DataTree object.
     */
    function updateData($object)
    {
        if (!is_a($object, 'DataTreeObject')) {
            /* Nothing to do for non objects. */
            return true;
        }

        /* Get the object id. */
        $id = $this->getId($object->getName());
        if (is_a($id, 'PEAR_Error')) {
            return $id;
        }

        /* See if we can break the object out to datatree_attributes table. */
        if (method_exists($object, '_toAttributes')) {
            /* If we can, clear out the datatree_data field to make
             * sure it doesn't get picked up by
             * getData(). Intentionally don't check for errors here in
             * case datatree_data goes away in the future. */
            $query = sprintf('UPDATE %s SET datatree_data = NULL WHERE datatree_id = %s',
                             $this->_params['table'],
                             (int)$id);

            Horde::logMessage('SQL Query by DataTree_sql::updateData(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
            $this->_db->query($query);

            /* Start a transaction. */
            $this->_db->autoCommit(false);

            /* Delete old attributes. */
            $query = sprintf('DELETE FROM %s WHERE datatree_id = %s',
                             $this->_params['table_attributes'],
                             (int)$id);

            Horde::logMessage('SQL Query by DataTree_sql::updateData(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
            $result = $this->_db->query($query);
            if (is_a($result, 'PEAR_Error')) {
                $this->_db->rollback();
                $this->_db->autoCommit(true);
                return $result;
            }

            /* Get the new attribute set, and insert each into the DB. If
             * anything fails in here, rollback the transaction, return the
             * relevant error, and bail out. */
            $attributes = $object->_toAttributes();
            foreach ($attributes as $attr) {
                $query = sprintf('INSERT INTO %s (datatree_id, attribute_name, attribute_key, attribute_value) VALUES (%s, %s, %s, ?)',
                                 $this->_params['table_attributes'],
                                 (int)$id,
                                 $this->_db->quote($attr['name']),
                                 $this->_db->quote($attr['key']));

                Horde::logMessage('SQL Query by DataTree_sql::updateData(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);

                $statement = $this->_db->prepare($query);
                $result = $this->_db->execute($statement, array(String::convertCharset($attr['value'], NLS::getCharset(), $this->_params['charset'])));
                if (is_a($result, 'PEAR_Error')) {
                    $this->_db->rollback();
                    $this->_db->autoCommit(true);
                    return $result;
                }
            }

            /* Commit the transaction, and turn autocommit back on. */
            $result = $this->_db->commit();
            $this->_db->autoCommit(true);

            return is_a($result, 'PEAR_Error') ? $result : true;
        } else {
            /* Write to the datatree_data field. */
            require_once 'Horde/Serialize.php';
            $ser = SERIALIZE_UTF7_BASIC;
            $data = Horde_Serialize::serialize($object->getData(), $ser, NLS::getCharset());

            $query = sprintf('UPDATE %s SET datatree_data = %s, datatree_serialized = %s' .
                             ' WHERE datatree_id = %s',
                             $this->_params['table'],
                             $this->_db->quote($data),
                             (int)$ser,
                             (int)$id);

            Horde::logMessage('SQL Query by DataTree_sql::updateData(): ' . $query, __FILE__, __LINE__, PEAR_LOG_DEBUG);
            $result = $this->_db->query($query);

            return is_a($result, 'PEAR_Error') ? $result : true;
        }
    }

    /**
     * Attempts to open a connection to the SQL server.
     *
     * @return boolean  True.
     */
    function _connect()
    {
        Horde::assertDriverConfig($this->_params, 'storage',
            array('phptype', 'hostspec', 'username', 'database', 'charset'),
            'DataTree SQL');

        if (!isset($this->_params['password'])) {
            $this->_params['password'] = '';
        }

        if (!isset($this->_params['table'])) {
            $this->_params['table'] = 'horde_datatree';
        }

        if (!isset($this->_params['table_attributes'])) {
            $this->_params['table_attributes'] = 'horde_datatree_attributes';
        }

        /* Connect to the SQL server using the supplied parameters. */
        require_once 'DB.php';
        $this->_db = &DB::connect($this->_params,
                                  array('persistent' => !empty($this->_params['persistent'])));
        if (is_a($this->_db, 'PEAR_Error')) {
            Horde::fatal($this->_db, __FILE__, __LINE__);
        }

        $this->_db->setOption('portability', DB_PORTABILITY_LOWERCASE | DB_PORTABILITY_ERRORS);

        return true;
    }

}