Quadcap Embeddable Database

com/quadcap/sql/Database.java

Go to the documentation of this file.

package com.quadcap.sql;

/* Copyright 1999 - 2003 Quadcap Software.  All rights reserved.
 *
 * This software is distributed under the Quadcap Free Software License.
 * This software may be used or modified for any purpose, personal or
 * commercial.  Open Source redistributions are permitted.  Commercial
 * redistribution of larger works derived from, or works which bundle
 * this software requires a "Commercial Redistribution License"; see
 * http://www.quadcap.com/purchase.
 *
 * Redistributions qualify as "Open Source" under  one of the following terms:
 *   
 *    Redistributions are made at no charge beyond the reasonable cost of
 *    materials and delivery.
 *
 *    Redistributions are accompanied by a copy of the Source Code or by an
 *    irrevocable offer to provide a copy of the Source Code for up to three
 *    years at the cost of materials and delivery.  Such redistributions
 *    must allow further use, modification, and redistribution of the Source
 *    Code under substantially the same terms as this license.
 *
 * Redistributions of source code must retain the copyright notices as they
 * appear in each source code file, these license terms, and the
 * disclaimer/limitation of liability set forth as paragraph 6 below.
 *
 * Redistributions in binary form must reproduce this Copyright Notice,
 * these license terms, and the disclaimer/limitation of liability set
 * forth as paragraph 6 below, in the documentation and/or other materials
 * provided with the distribution.
 *
 * The Software is provided on an "AS IS" basis.  No warranty is
 * provided that the Software is free of defects, or fit for a
 * particular purpose.  
 *
 * Limitation of Liability. Quadcap Software shall not be liable
 * for any damages suffered by the Licensee or any third party resulting
 * from use of the Software.
 */

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;

import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collection;
import java.util.Enumeration;
import java.util.Iterator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import java.util.zip.GZIPOutputStream;

import java.sql.SQLException;

import com.quadcap.sql.file.BlockFile;
import com.quadcap.sql.file.ByteUtil;
import com.quadcap.sql.file.Datafile;
import com.quadcap.sql.file.Log;
import com.quadcap.sql.file.SubPageManager;

import com.quadcap.sql.index.BCursor;
import com.quadcap.sql.index.Btree;

import com.quadcap.sql.lock.Transaction;

import com.quadcap.util.collections.LongIterator;
import com.quadcap.util.collections.LongMap;

import com.quadcap.util.ConfigString;
import com.quadcap.util.Debug;
import com.quadcap.util.Util;
 
/**
 * This class implements the QED SQL database outer API.
 *
 * @author Stan Bailes
 */
public class Database extends Datafile {
    QDriver                             driver;
    Btree                               tableIndex;
    // map index name to table name
    Btree                               indexIndex;
    Btree                               blobPermRefCounts;
    Map                                 tableMap        = new HashMap();
    MultiSet                            forwardDeps     = null;
    MultiSet                            reverseDeps     = null;
    long                                rootBlock       = 0;
    public DatabaseRoot                 nroot;

    /** The Morgue; BLOB/CLOBs with zero (persistent) refcounts:
        blobRef -> any */
    LongMap                     blobMorgue      = new LongMap(32);

    /** Map blob -> Vector<transid> */
    LongMap                     blobActivity    = new LongMap(128);
    
    
    int                                 connectionCount = 0;
    int                                 closeCount      = 0;
    long                                nextTransId     = 0;

    public Object                       driverLock;

    static final ConfigString           backupClass
    = ConfigString.find("qed.backupClass",
                        "com.quadcap.sql.tools.XmlDump");
    /**
     * Constructor
     */
    public Database() {}

    public void init(QDriver driver, String url, 
                     String fileName, Properties props)
        throws IOException, SQLException
    {
        this.driver = driver;
        init(url, fileName, props);
    }

    /**
     * Create a new root block
     */
    public void createRoot(String fileName, Properties props)
        throws IOException
    {
        this.rootBlock = getFile().getUserBlock(0);
        this.root = this.nroot = new DatabaseRoot(this, fileName, props);
        if (!readOnly && !inMemory) {
            new File(fileName, "logfile").delete();
            flushRoot();
        }
     }

    /**
     * Read the root block from an existing database
     */
    public void readRoot() throws IOException {
        this.rootBlock = getFile().getUserBlock(0);
        root = nroot = (DatabaseRoot)file.getObject(rootBlock);
        nroot.setDatabase(this);
    }

    /**
     * Write the database root
     */
    public void flushRoot() throws IOException {
        if (!inMemory /*?*/  && !readOnly) {
            getFile().updateObject(rootBlock, root);
        }
    }

    /**
     * Get ready to rumble!!!
     */
    public void bootFromRoot(boolean restart) throws IOException { 
        nextTransId = root.getNextTransId();
        nroot.setNextTransId((nextTransId | 0xff) + 1);

        blobPermRefCounts = new Btree(file,
                                      nroot.getBlobRefCountRoot(), !restart);
        tableIndex = new Btree(file, nroot.getRelationIndexNode(), !restart);
        indexIndex = new Btree(file, nroot.getIndexIndexNode(), !restart);
        forwardDeps =
            new MultiSet(new Btree(file, nroot.getForwardDepsNode(),
                                   !restart));
        reverseDeps =
            new MultiSet(new Btree(file, nroot.getReverseDepsNode(),
                                   !restart));
    }

    /**
     * Return the next available transaction ID.
     */
    public long getNextTransId() throws IOException {
        synchronized (fileLock) {
            long t = nextTransId++;
            if (!inMemory && (nextTransId & 0xff) == 0) {
                nroot.setNextTransId(nextTransId + 0x100);
                flushRoot();
            }
            return t;
        }
    }

    /**
     * Figure out if it's time to do a backup.
     */
    public void maybeBackup() throws IOException {
        if (inMemory) return;
        // are backups enabled?
        int days = getBackupDays();
        if (days == 0) return;

        // is today the day?
        Calendar now = Calendar.getInstance();
        int nowDay = now.get(Calendar.DAY_OF_WEEK);
        if (((days >> nowDay) & 1) == 0) return;
        
        // is now the time?
        int time = getBackupTime();
        int today = now.get(Calendar.DAY_OF_YEAR);
        int nowTime = 
            now.get(Calendar.HOUR_OF_DAY) * 60 +
            now.get(Calendar.MINUTE);
        if (nowTime < time || today == getBackupLastDay()) return;

        // cycle the old backup files
        int count = getBackupCount();
        for (int i = count-2; i >= 0; i--) {
            makeBackupFile(i).renameTo(makeBackupFile(i+1));
        }

        // write new backup in correct format
        OutputStream w = new FileOutputStream(makeBackupFile(0));
        if (getBackupFormat().equals("xml.gz")) {
            w = new GZIPOutputStream(w);
        }
        OutputStreamWriter ow = new OutputStreamWriter(w);
        BufferedWriter bw = new BufferedWriter(ow);
        
        // create the backup helper and do it.
        try {
            java.sql.Connection conn =
                driver.makeConnection(this, "backup", null);
            Class c = Class.forName(backupClass.toString());
            Backup b = (Backup)c.newInstance();
            b.backup(conn, bw);
            bw.close();
        } catch (IOException e) {
            throw e;
        } catch (Throwable t) {
            throw new IOException(t.toString());
        }
        setBackupLastDay(today);
    }

    File makeBackupFile(int n) throws IOException {
        File bdir = new File(getBackupDir());
        if (!bdir.exists()) {
            if (!bdir.mkdirs()) {
                throw new IOException("can't create backup directory: " +
                                      bdir.getAbsolutePath());
            }
        }
        String name = "backup-" + n + "." + getBackupFormat();
        return new File(bdir, name);
    }

    /**
     * TODO: XXX Unicode!!!
     * Add a new named index for the specified table.
     */
    public void addIndex(String indexName, String tableName) 
        throws IOException
    {
        synchronized (fileLock) {
            indexIndex.set(indexName.getBytes(), tableName.getBytes());
        }
    }


    /**
     * Return the table for the index of the specified name.
     */
    public String getTableForIndex(String indexName) throws IOException {
        synchronized (fileLock) {
            byte[] ret = indexIndex.get(indexName.getBytes());
            if (ret == null) return null;
            return new String(ret);
        }
    }

    /**
     * Remove the specified index
     */
    public void deleteIndex(String indexName) throws IOException {
        synchronized (fileLock) {
            indexIndex.delete(indexName.getBytes());
        }
    }

    /**
     * RENAME TABLE percolates into the index index...
     */
    void renameIndexedTable(String oldN, String newN) throws IOException {
        synchronized (fileLock) {
            BCursor bc = indexIndex.getCursor(false);
            if (bc != null) {
                try {
                    while (bc.next()) {
                        String name = new String(bc.getVal());
                        if (name.equals(oldN)) {
                            bc.prev();
                            bc.replace(newN.getBytes());
                            bc.next();
                        }
                    }
                } finally {
                    bc.release();
                }
            }
        }
    }

    /**
     * Return the table identity for the table
     *
     * @exception SQLException if the table can't be located
     */
    public long getTableIdentity(String name)
        throws IOException, SQLException
    {
        synchronized (fileLock) {
            if (inMemory) {
                Table t = (Table)getRelation(name);
                return t.tableIdentity;
            } else {
                BCursor tc = tableIndex.getCursor(true);
                try {
                    if (tc.seek(name.getBytes()) &&
                        tc.getValLen() >= 16) {
                        return ByteUtil.getLong(tc.getValBuf(), 8);
                    } else {
                        throw new SQLException("getTableIdentity(" +
                                               name + "): no table");
                    }
                } finally {
                    tc.release();
                }
            }
        }
    }

    /**
     * The "table identity" is a long value stored along with the table
     * stream pointer, so that it can easily be updated as long as the index
     * is in memory, without touching any data pages.
     */
    public void updateTableIdentity(String name, long num)
        throws IOException, SQLException
    {
        synchronized (fileLock) {
            if (inMemory) {
                Table t = (Table)getRelation(name);
                t.tableIdentity = num;
            } else {
                BCursor tc = tableIndex.getCursor(true);
                try {
                    if (tc.seek(name.getBytes())) {
                        int cnt = tc.getVal(temp);
                        ByteUtil.putLong(temp, 8, num);
                        tc.replace(temp, 0, cnt);
                        return;
                    } else {
                        throw new SQLException("getTableIdentity(" +
                                               name + "): no table");
                    }
                } finally {
                    tc.release();
                }
            }
        }
    }

    /**
     * Helper to execute a single SQL statement (in its own connection)
     * and return a result set.
     */
    public java.sql.ResultSet execute(String sql)
        throws SQLException, IOException
    {
        Connection c = new Connection(this, "__NO_AUTH__", null);
        Session session = c.createSession();
        return execute(session, sql);
    }

    /**
     * Helper to execute a single SQL statement for the given session.
     */
    public java.sql.ResultSet execute(Session session, String sql)
        throws IOException, SQLException
    {
        SQLParser p = new SQLParser(session, sql, true);
        Stmt s = null;
        try {
            s = p.statement();
        } catch (Throwable t) {
            Debug.print(t);
        }
        
        if (s == null) {
            throw new SQLException("Parse error", "42000");
        }
        session.doStatement(s);
        return session.getResultSet();
    }

    /**
     * Return the table/view with the specified name
     *
     * XXX Fix this "cache" implementation 
     */
    public Relation getRelation(String name)
        throws IOException
    {
        Relation t = null;
        synchronized (fileLock) {
            t = (Relation)tableMap.get(name);
            if (!inMemory && t == null) {
                BCursor tc = tableIndex.getCursor(true);
                try {
                    if (tc.seek(name.getBytes())) {
                        byte[] d = tc.getValBuf();
                        t = (Relation)getFile().getObject(ByteUtil.getLong(d, 0));
                        if (t != null) {
                            tableMap.put(name, t);
                        }
                    }
                } finally {
                    tc.release();
                }
            }
        }
        //#ifdef DEBUG
        if (Trace.bit(0)) {
            Debug.println("getRelation(" + name + ") = " +
                          ((t == null) ? "NOT FOUND" : "FOUND"));
        }
        //#endif
        return t;
    }

    /**
     * Add a new table/view
'     */
    public void addRelation(Relation table)    
        throws IOException, SQLException
    {
        synchronized (fileLock) {
            String name = table.getName();
            //#ifdef TRACE
            if (Trace.bit(0)) {
                Debug.println("addRelation(" + name + ")");
            }
            //#endif
            if (tableMap.get(name) == null) {
                tableMap.put(name, table);
                if (!inMemory) {
                    BCursor tc = tableIndex.getCursor(true);
                    try {
                        byte[] tkey = name.getBytes();
                        if (!tc.seek(name.getBytes())) {
                            long ref = getFile().putObject(table);
                            ByteUtil.putLong(temp, 0, ref);
                            ByteUtil.putLong(temp, 8, 1); // start counting at '1'
                            tc.insert(tkey, tkey.length, temp, 0, 16);
                        }
                    } finally {
                        tc.release();
                    }
                }
            } else {
                throw new SQLException("Table/view already exists: " + name,
                                       "42000");
            }
        }
    }   

    /**
     * Delete the specified table/view
     */
    public void removeRelation(String name)
        throws IOException, SQLException
    {
        synchronized (fileLock) {
            //#ifdef DEBUG
            if (Trace.bit(0)) {
                Debug.println("removeRelation(" + name + ")");
            }
            //#endif
            Relation t = getRelation(name);
            if (t == null) {
                throw new SQLException("No such table/view: " + name, "42000");
            }
            removeViewDependencies(name);
            tableMap.remove(name);
            if (!inMemory) {
                BCursor tc = tableIndex.getCursor(true);
                try {
                    if (tc.seek(name.getBytes())) {
                        getFile().removeObject(tc.getValAsLong());
                        tc.delete();
                    }
                } finally {
                    tc.release();
                }
            }
        }
        //#ifdef DEBUG
        if (getRelation(name) != null) {
            throw new SQLException("Relation not removed!!!");
        }
        //#endif
    }

    /**
     * Rename a relation.  This involves:
     *
     * 1. Changing the cached relation
     * 2. Changing the btree index to the relation
     * 3. Changing any indexIndex entries referring to the rel
     * 4. Changing any forwardDeps or backwardDeps which
     *    refer to the relation
     */
    public void renameRelation(Relation r, String newN) throws IOException {
        synchronized (fileLock) {
            String oldN = r.getName();
            r.setName(newN);
            tableMap.remove(oldN);
            tableMap.put(newN, r);
            if (!inMemory) {
                BCursor tc = tableIndex.getCursor(true);
                try {
                    if (tc.seek(oldN.getBytes())) {
                        byte[] d = tc.getValBuf();
                        long ref = ByteUtil.getLong(d, 0);
                        tc.delete();
                        byte[] nb = newN.getBytes();
                        if (!tc.seek(nb)) {
                            tc.insert(nb, nb.length, d, 0, 16);
                            getFile().updateObject(ref, r);
                        }
                    }
                } finally {
                    tc.release();
                }
            }
            if (true) {
                forwardDeps.rename(oldN, newN, reverseDeps);
                reverseDeps.rename(oldN, newN, forwardDeps);
            }
            if (true) {
                renameIndexedTable(oldN, newN);
            }
        }
    }
    
    /**
     * Store a modified version of the specified view/table to the
     * store
     */
    public void updateRelation(Relation table) throws IOException {
        synchronized (fileLock) {
            //#ifdef TRACE
            if (Trace.bit(0)) {
                Debug.println("updateRelation(" + table.getName() + ")"); 
                //Debug.println("updateRelation(" + table.getName() + ": " + table + ")"); // + Util.stackTrace());
            }
            //#endif
            tableMap.put(table.getName(), table);
            if (!inMemory) {
                BCursor tc = tableIndex.getCursor(true);
                try {
                    if (tc.seek(table.getName().getBytes())) {
                        getFile().updateObject(tc.getValAsLong(), table);
                    } else {
                        throw new IOException("Not found: " + table.getName());
                    }
                } finally {
                    tc.release();
                }
            }
        }
    }

    /**
     * Establish the dependency link between a view and a base table
     */
    public void addViewDependency(String base, String view) 
        throws IOException 
    {
        //#ifdef TRACE
        if (Trace.bit(0)) {
            Debug.println("addViewDependency(" + base + ", " + view + ")");
        }
        //#endif

        synchronized (fileLock) {
            forwardDeps.put(base, view);
            reverseDeps.put(view, base);
        }
    }

    /**
     * Make sure that the specified view dependency is valid
     */
    public void checkViewDependency(String base, String view)
        throws SQLException, IOException
    {
        Relation baseR = getRelation(base);
        if (baseR == null) {
            throw new SQLException("Base table not found: " + base, "42000");
        }
        if (baseR instanceof Table) {
            Table baseT = (Table)baseR;
            int modifiers = baseT.getModifiers();
            if ((modifiers & Table.TEMPORARY) != 0) {
                throw new SQLException("Can't create view of temporary " +
                                       "table " + base, "42000");
            }
        }
    }

    /**
     * Return an enumeration of all views for the specified base table
     */
    public Enumeration getViews(String base) throws IOException {
        synchronized (fileLock) {
            return forwardDeps.get(base);
        }
    }

    /**
     * Return an enumeration containing all base tables for the specified
     * view
     */
    public Enumeration getBases(String view) throws IOException {
        synchronized (fileLock) {
            return reverseDeps.get(view);
        }
    }

    public void removeViewDependencies(String view)
        throws SQLException, IOException
    {
        //#ifdef TRACE
        if (Trace.bit(0)) {
            Debug.println("removeViewDependencies(" + view + ")");
        }
        //#endif
        synchronized (fileLock) {
            Enumeration views = getViews(view);
            if (views.hasMoreElements()) {
                throw new SQLException("view has dependencies: " + view,
                                       "42000");
            }
            Enumeration bases = getBases(view);
            while (bases.hasMoreElements()) {
                forwardDeps.delete(bases.nextElement().toString(), view);
            }
        }
    }

    /**
     * This hacky crack of encapsulation is for sql.meta.MetaTables, to
     * support DatabaseMetaData.getTables().  All he really needs is
     * an Iterator which returns the table names....
     */
//     public Btree getRelationIndex() {
//      return tableIndex;
//     }

    public Iterator getRelationNameIterator() throws IOException {
        if (inMemory) {
            return tableMap.keySet().iterator();
        } else {
            final BCursor tc = tableIndex.getCursor(false);
            return new Iterator() {
                    boolean called = false;
                    boolean last = false;
                    boolean closed = false;
                    public boolean hasNext() {
                        try {
                            if (!closed && !called) {
                                last = tc.next();
                                called = true;
                            }
                        } catch (IOException ex) {
                            last = false;
                        }
                        if (!last && !closed) {
                            closed = true;
                            try {
                                tc.release();
                            } catch (Throwable t) {
                                Debug.print(t);
                            }
                        }                                    
                        return last;
                    }
                    public Object next() {
                        Object ret = null;
                        try {
                            if (!called && !closed) {
                                last = tc.next();
                            }
                            if (!last) {
                                if (!closed) {
                                    closed = true;
                                    try {
                                        tc.release();
                                    } catch (Throwable t) {
                                        Debug.print(t);
                                    }
                                }
                            } else {
                                ret = new String(tc.getKey());
                                called = false;
                            }
                        } catch (IOException ex) {
                            ret = "<Exception: " + ex + ">";
                        }
                        return ret;
                    }
                    public void remove() {
                        try {
                            tc.delete();
                        } catch (IOException e) {
                            Debug.print(e);
                            throw new RuntimeException("Error deleting record", e);
                        }
                    }
                };
        }
    }
    
//     public Collection getRelationNames() throws IOException {
//         ArrayList list = new ArrayList();
//         synchronized (fileLock) {
//             if (inMemory) {
//                 return tableMap.keySet();
//             }
//             BCursor tc = tableIndex.getCursor(false);
//             try {
//                 while (tc.next()) {
//                     list.add(new String(tc.getKey()));
//                 }
//             } finally {
//                 tc.release();
//             }
//         }
//         return list;
//     }

    /**
     * Add a new connection (just remember the count...)
     */
    public int addConnection() {
        synchronized (driverLock) {
            //#ifdef TRACE
            if (Trace.bit(8)) {
                Debug.println("addConnection, count now: " + (connectionCount+1));
            }
            //#endif
            return ++connectionCount;
        }
    }

    /**
     * Sleight of hand to get my driver interface
     */
    public QDriver getDriver() {
        return driver;
    }

    /**
     * Remove a connection.  If the connection count drops to zero, we
     * get crafty.
     */
    public void removeConnection() {
        synchronized (driverLock) {
            if (--connectionCount == 0 && closeCount++ < 1) {
                try {
                    getLog().checkpoint();
                } catch (Throwable e) {
                }
            }
            //#ifdef TRACE
            if (Trace.bit(8)) {
                Debug.println("removeConnection, count now: " + (connectionCount));
            }
            //#endif
            if (connectionCount == 0 && closeCount == 1) {
                driver.closeDatabase(origFileName);
            }
        }
    }

    /**
     * Return the number of active connections
     */
    public int getConnectionCount() {
        return connectionCount;
    }

    /**
     * Return the backup directory name
     */    
    public String getBackupDir() {
        return nroot.getBackupDir();
    }
    
    /**
     * Set the backup directory
     */    
    public void setBackupDir(String str) throws IOException {
        nroot.setBackupDir(str);
        flushRoot();
    }

    /**
     * Return the backup count
     */    
    public int getBackupCount() {
        return nroot.getBackupCount();
    }

    /**
     * Set the backup count
     */    
    public void setBackupCount(int count) throws IOException {
        nroot.setBackupCount(count);
        flushRoot();
    }

    /**
     * Return the backup days
     */    
    public int getBackupDays() {
        return nroot.getBackupDays();
    }

    /**
     * Set the backup days
     */    
    public void setBackupDays(int count) throws IOException {
        nroot.setBackupDays(count);
        flushRoot();
    }

    /**
     * Return the backup last day
     */    
    public int getBackupLastDay() {
        return nroot.getBackupLastDay();
    }

    /**
     * Set the backup last day
     */    
    public void setBackupLastDay(int day) throws IOException {
        nroot.setBackupLastDay(day);
        flushRoot();
    }

    /**
     * Return the backup time
     */    
    public int getBackupTime() {
        return nroot.getBackupTime();
    }

    /**
     * Set the backup time
     */    
    public void setBackupTime(int time) throws IOException {
        nroot.setBackupTime(time);
        flushRoot();
    }

    /**
     * Return the backup format
     */    
    public String getBackupFormat() {
        return nroot.getBackupFormat();
    }

    /**
     * Set the backup format
     */    
    public void setBackupFormat(String str) throws IOException {
        nroot.setBackupFormat(str);
        flushRoot();
    }

    /**
     * Check the specified authentication credentials;  throw an exception
     * if authorization shouldn't be granted.
     *
     * We really don't care about passwords.  This is an embedded database,
     * remember;  it's really just a fancy API on top of a simple file!
     */
    public void checkAuth(String user, String passwd) throws SQLException {
        // XXX impl
    }

    /**
     * Return the current persistent refcount for the specified blob.
     */
    public int getPermBlobRefCount(long blob) throws IOException {
        synchronized (fileLock) {
            ByteUtil.putLong(key, 0, blob);
            byte[] val = blobPermRefCounts.get(key);
            int refcount = val == null ? 0 : ByteUtil.getInt(val, 0);
            return refcount;
        }
    }


    //------------------------------------------------------------------
    // private functions
    //------------------------------------------------------------------

    private byte[] key = new byte[8];
    
    public int refCountBlob(long transId, long blob, int incr)
        throws IOException
    {
        synchronized (fileLock) {
            // find the current refcount
            ByteUtil.putLong(key, 0, blob);
            byte[] val = blobPermRefCounts.get(key);
            int refcount = val == null ? 0 : ByteUtil.getInt(val, 0);
            int preRefcount = refcount;

            // if refcount 0 -> 1, remove from morgue
            if (refcount == 0 && incr == 1) {
                val = new byte[4];
                delMorgue(blob);
            }
            refcount += incr;
            if (refcount < 0) {
                throw new RuntimeException("freePermBlobRef(" + blob +
                                           "), refcount negative");
            }

            if (refcount <= 0) {
                if (preRefcount > 0) {
                    addMorgue(blob, transId);
                }
                blobPermRefCounts.delete(key);
            } else {
                // otherwise, update perm refcount
                ByteUtil.putInt(val, 0, refcount);
                blobPermRefCounts.set(key, val);
            }

            return refcount;
        }
    }

    /**
     * Add a blob/clob to the morgue
     */
    public void addMorgue(long blob, long transId) {
        synchronized (fileLock) {       
            blobMorgue.put(blob, new Long(transId));
        }
    }

    /**
     * Remove a blob/clob from the morgue
     */
    public void delMorgue(long blob) {
        synchronized (fileLock) {
            blobMorgue.remove(blob);
        }
    }

    /**
     * Enumerate the morgue.
     */
    public LongIterator enumerateMorgue() {
        return blobMorgue.keys();
    }

    /**
     * Called during a checkpoint operation.
     *
     * @param activeTransactions is a map whose keys are the set of
     *   ids of active transactions.
     */
    public void checkpointHandler(LongMap activeTransactions)
        throws IOException
    {
        emptyMorgue(activeTransactions);
    }
    
    /**
     * Empty the BLOB morgue.  Remove any BLOBS in the morge that were
     * deleted by transactions which have committed.
     */
    public void emptyMorgue(LongMap activeTransactions)
        throws IOException
    {
        // ---- temporarily disassociate the log from the file so we don't
        // ---- log the file writes we're about to do.  This is OK because
        // ---- we're doing a checkpoint now.
        Log saveLog = getLog();
        getFile().setLog(null);
        try {
            synchronized (fileLock) {
                LongIterator iter = enumerateMorgue(); 
                while (iter.hasNext()) {
                    long blob = iter.nextLong();
                    long transId = ((Long)(blobMorgue.get(blob))).longValue();
                    if (activeTransactions.get(transId) == null) {
                        try {
                            getFile().freeStream(blob);
                        } catch (Throwable t) {
                            Debug.println("Error removing blob: " + blob);
                            Debug.print(t);
                        }
                        iter.remove();
                    }
                }
            }
        } finally {
            // ---- restore the {file <--> log} association
            getFile().setLog(saveLog);
        }
    }

    Session zSession = null;
    Connection zConn = null;

    final Session getSession(Transaction t) throws IOException, SQLException {
        if (zSession == null) {
            synchronized (this) {
                if (zSession == null) {
                    zConn = new Connection(this, "__SYSTEM", null);
                    zSession = zConn.createSession();
                }
            }
        }
        zConn.transId = t.getTransactionId();  // XXX what about sync?
        zConn.trans = t;
        return zSession;
    }

    //#ifdef DEBUG
    public void display(PrintWriter w) throws IOException {
        w.println("Root:");
        w.println("    blobs:        " + nroot.getBlobRefCountRoot());
        w.println("    relations:    " + nroot.getRelationIndexNode());
        w.println("    indexes:      " + nroot.getIndexIndexNode());
        w.println("    forward deps: " + nroot.getForwardDepsNode());
        w.println("    reverse deps: " + nroot.getReverseDepsNode());
        w.println("    next trans:   " + nroot.getNextTransId());
        if (true) {// iterate tables, build node map
        }
        if (true) { // display indexes, update node map
            w.println("indexIndex:");
            indexIndex.display(w);
            w.println("tableIndex:");
            tableIndex.display(w);
        }
        if (true) { // traverse freelists, update node map
            w.println("forwardDeps:");
            forwardDeps.display(w);
            w.println("reverseDeps:");
            reverseDeps.display(w);
        }
        if (true) { // search for blocks missing from node map
        }
        if (true) { // dump node map in block order
        }
        // traverse block file with node map used for annotations.
    }
    //#endif

    /**
      * Convenience method to write a new row to the database
      */
    public long putRow(Session session, BlockFile f,
                       Tuple t, Row row)
        throws SQLException, IOException
    {
        long rowId;
        if (inMemory) {
            rowId = f.putObject(row = new Row(row, t));
        } else {
            byte[] buf = LazyRow.writeRow(session, t, row);
            rowId = f.putBytes(buf);
        }
        //#ifdef TRACE
        if (Trace.bit(11)) {
            Debug.println((f == this.file ? "file" : "temp") + "." +
                          ((t == null) ? "temp" :
                           ("Table[" + t.getName() + "]")) +
                          ".putRow(" + row + "): " + rowId);
        }
        //#endif
        return rowId;
    }

    /**
      * Convenience method to update a table row
      */
    public final void putRow(Session session, Tuple t, 
                             long rowId, Row row)
        throws SQLException, IOException
    {
        //#ifdef TRACE
        if (Trace.bit(11)) {
            Debug.println("Table[" + t.getName() + "].putRow(" + row + ")");
        }
        //#endif
        if (inMemory) {
            getFile().updateObject(rowId, new Row(row, t));
        } else {
            byte[] ret = LazyRow.writeRow(session, t, row);
            getFile().updateBytes(rowId, ret);
        }
    }

    public void getRow(long rowId, LazyRow row, boolean isTemp) 
        throws IOException, SQLException
    {
        BlockFile f = isTemp ? getTempFile(false) : getFile();
        if (inMemory) {
            Row r = (Row)f.getObject(rowId);
            if (r == null) {
                throw new IOException("No row: " + rowId);
            }
            if (r.size() != row.size()) {
                //#ifdef DEBUG
                Debug.println((isTemp ? "temp" : "file") + ".getRow(" +
                          rowId + ") = " + row);
                Debug.println("Bad row size in getRow: " + r.size() +
                              " vs " + row.size());
                //#endif
                throw new IOException("Bad row size in getRow: " + r.size() +
                    " vs " + row.size());
            }
            for (int i = 1; i <= row.size(); i++) {
                row.set(i, r.getItem(i));
            }
        } else {
            byte[] rowBytes = f.getBytes(rowId);
            row.reset(rowBytes, this);
        }
        //#ifdef TRACE
        if (Trace.bit(11)) {
            Debug.println((isTemp ? "temp" : "file") + ".getRow(" +
                          rowId + ") = " + row);
        }
        //#endif
    }

    public void removeRow(long rowId) throws IOException {
        removeRow(getFile(), rowId);
    }

    public void removeRow(BlockFile file, long rowId) throws IOException {
        if (inMemory) {
            file.removeObject(rowId);
        } else {
            file.freeStream(rowId);
        }
    }

    public boolean inMemory() {
        return inMemory;
    }
}