/*
 *                 Sun Public License Notice
 * 
 * The contents of this file are subject to the Sun Public License
 * Version 1.0 (the "License"). You may not use this file except in
 * compliance with the License. A copy of the License is available at
 * http://www.sun.com/
 * 
 * The Original Code is NetBeans. The Initial Developer of the Original
 * Code is Sun Microsystems, Inc. Portions Copyright 1997-2001 Sun
 * Microsystems, Inc. All Rights Reserved.
 */
package org.netbeans.mdr.persistence.jdbcimpl;

import org.netbeans.mdr.persistence.*;
import org.netbeans.mdr.util.*;

import java.sql.*;
import java.util.*;
import java.io.*;

/**
 * JdbcStorage implements the MDR Storage interface using JDBC.
 *
 * @author John V. Sichi
 * @version $Id: JdbcStorage.java,v 1.6 2004/03/05 00:33:57 jsichi Exp $
 */
class JdbcStorage implements Storage
{
    public static final String MOFID_SEQ_TABLE_NAME = "MOFID_SEQ";
    
    public static final String MOFID_SEQ_COL_NAME = "MOFID_SEQ_NEXT";
    
    public static final String KEY_COL_PREFIX = "IDX_KEY";
    
    public static final String SINGLE_VAL_COL_PREFIX = "IDX_SVAL";
    
    public static final String MULTI_VAL_COL_PREFIX = "IDX_MVAL";
    
    public static final String ORDINAL_COL_NAME = "IDX_ORD";
    
    public static final String SURROGATE_COL_NAME = "IDX_SUR";

    public static final String PRIMARY_INDEX_NAME = "PRIMARY_INDEX";
    
    private final Connection jdbcConnection;

    private final DatabaseMetaData dbMetaData;

    private final String idQuote;

    private final String schemaName;

    private final String userName;

    private final String storageId;

    private final boolean realSchema;

    private final Map entryTypeToDataTypeMap;

    private Statement jdbcStmt;
    
    private ResultSet jdbcResultSet;

    private long nextSerialNumber;
    
    private Map nameToIndexMap;

    private LazyPreparedStatement sqlUpdateSerialNumber;

    // REVIEW: if this gets too bloated for a big model, may need to turn this
    // into a cache instead of a map
    private Map sqlToPreparedStatementMap;

    private JdbcPrimaryIndex primaryIndex;

    JdbcStorage(
        Properties properties,
        String storageId)
        throws StorageException
    {
        this.storageId = storageId;
        
        schemaName = properties.getProperty(
            JdbcStorageFactory.STORAGE_SCHEMA_NAME);
        userName = properties.getProperty(
            JdbcStorageFactory.STORAGE_USER_NAME);

        entryTypeToDataTypeMap = new HashMap();
        createTypeMap(properties);
        
        String url = properties.getProperty(
            JdbcStorageFactory.STORAGE_URL);
        String password = properties.getProperty(
            JdbcStorageFactory.STORAGE_PASSWORD);
        boolean success = false;
        
        try {
            // REVIEW:  maybe connect/disconnect should correspond to
            // open/close instead of constructor/shutdown?
            jdbcConnection =
                DriverManager.getConnection(url,userName,password);
            jdbcConnection.setAutoCommit(false);
            jdbcStmt = jdbcConnection.createStatement();
            dbMetaData = jdbcConnection.getMetaData();
            realSchema = dbMetaData.supportsSchemasInTableDefinitions();
            idQuote = dbMetaData.getIdentifierQuoteString();
            success = true;
        } catch (SQLException ex) {
            throw new JdbcStorageException(ex);
        } finally {
            if (!success) {
                rollbackConnection();
                closeConnection();
            }
        }
    }

    private void createTypeMap(Properties properties)
    {
        entryTypeToDataTypeMap.put(
            EntryType.MOFID,
            properties.getProperty(
                JdbcStorageFactory.STORAGE_DATATYPE_MOFID,
                "BIGINT"));
        
        entryTypeToDataTypeMap.put(
            EntryType.STREAMABLE,
            properties.getProperty(
                JdbcStorageFactory.STORAGE_DATATYPE_STREAMABLE,
                "LONGVARBINARY"));
        
        entryTypeToDataTypeMap.put(
            EntryType.STRING,
            properties.getProperty(
                JdbcStorageFactory.STORAGE_DATATYPE_STRING,
                "VARCHAR(2000)"));
        
        entryTypeToDataTypeMap.put(
            EntryType.INT,            
            properties.getProperty(
                JdbcStorageFactory.STORAGE_DATATYPE_INT,
                "BIGINT"));
    }

    // implement Storage
    public String getName()
    {
        return schemaName + ".jdbc";
    }
    
    // implement Storage
    public String getStorageId ()
    {
        return storageId;
    }

    private void writeSerialNumber(long serialNumber)
        throws StorageException
    {
        executeUpdate(
            sqlUpdateSerialNumber,
            new Object[]{new Long(serialNumber)});
    }
    
    // implement Storage
    public synchronized long getSerialNumber()
    {
        return nextSerialNumber++;
    }
    
    // implement Storage
    public MOFID readMOFID (java.io.InputStream inputStream)
        throws StorageException
    {
        // NOTE:  ripped from memoryimpl
        try {
            String storageId = IOUtils.readString(inputStream);
            if (storageId == null) {
                storageId = this.storageId;
            }
            long serial = IOUtils.readLong(inputStream);
            return new MOFID(serial, storageId);
        } catch (java.io.IOException ioException) {
            throw new StorageIOException(ioException);
        }
    }

    // implement Storage
    public void writeMOFID (java.io.OutputStream outputStream, MOFID mofid)
        throws StorageException
    {
        // NOTE:  ripped from memoryimpl
        try {
            if (storageId.equals(mofid.getStorageID())) {
                IOUtils.writeString(outputStream, null);
            } else {
                IOUtils.writeString(outputStream, mofid.getStorageID());
            }
            IOUtils.writeLong(outputStream, mofid.getSerialNumber());
        } catch (IOException ioException) {
            throw new StorageIOException(ioException);
        }
    }

    DatabaseMetaData getDatabaseMetaData()
    {
        return dbMetaData;
    }

    private String getQualifiedTableName(String tableName)
    {
        if (realSchema) {
            return idQuote + schemaName + idQuote + "." + idQuote
                + tableName + idQuote;
        } else {
            return idQuote + schemaName + "_" + tableName + idQuote;
        }
    }
    
    private String getQualifiedSchemaName()
    {
        return idQuote + schemaName + idQuote;
    }

    private void rollbackConnection()
    {
        closeResultSet();
        if (jdbcConnection != null) {
            try {
                jdbcConnection.rollback();
            } catch (SQLException ex) {
                // TODO: trace
            }
        }
    }

    private long readSerialNumber()
    {
        try {
            jdbcResultSet = jdbcStmt.executeQuery(
                "select * from "+getQualifiedTableName(MOFID_SEQ_TABLE_NAME));
            jdbcResultSet.next();
            long x = jdbcResultSet.getLong(1);
            return x;
        } catch (SQLException ex) {
            return -1;
        }
    }
    
    // implement Storage
    public synchronized boolean exists() throws StorageException
    {
        long x = readSerialNumber();
        return x != -1;
    }
    
    // implement Storage
    public synchronized boolean delete() throws StorageException
    {
        rollbackConnection();
        try {
            if (realSchema) {
                jdbcResultSet = dbMetaData.getSchemas();
                boolean found = false;
                while (jdbcResultSet.next()) {
                    String name = jdbcResultSet.getString(1);
                    if (name.equals(schemaName)) {
                        found = true;
                        break;
                    }
                }
                closeResultSet();
                if (!found) {
                    // schema doesn't even exist
                    return true;
                }
                jdbcStmt.execute(
                    "drop schema " + getQualifiedSchemaName() + " cascade");
            } else {
                jdbcResultSet = dbMetaData.getTables(
                    null,null,schemaName + "%",null);
                List tables = new ArrayList();
                while (jdbcResultSet.next()) {
                    tables.add(jdbcResultSet.getString("TABLE_NAME"));
                }
                closeResultSet();
                Iterator iter = tables.iterator();
                while (iter.hasNext()) {
                    String tableName = (String) iter.next();
                    jdbcStmt.execute(
                        "drop table " + idQuote + tableName + idQuote);
                }
            }
            jdbcConnection.commit();
            return true;
        } catch (SQLException ex) {
            rollbackConnection();
            return false;
        }
    }

    private boolean isBlank(String s)
    {
        return (s == null) || (s.length() == 0);
    }
    
    // implement Storage
    public synchronized void create(boolean replace, ObjectResolver resolver)
        throws StorageException
    {
        try {
            if (replace) {
                delete();
            }
            rollbackConnection();
            if (realSchema) {
                String sql = "create schema " + getQualifiedSchemaName();
                if (!isBlank(userName)) {
                    sql = sql + " authorization " + userName;
                }
                jdbcStmt.execute(sql);
            }
            String intType = getDataType(EntryType.INT);
            jdbcStmt.execute(
                "create table " + getQualifiedTableName(MOFID_SEQ_TABLE_NAME)
                + "(" + MOFID_SEQ_COL_NAME + " " + intType
                + " not null primary key)");
            jdbcStmt.executeUpdate(
                "insert into " + getQualifiedTableName(MOFID_SEQ_TABLE_NAME)
                + " values(1)");
            nextSerialNumber = 1;
            openImpl();
            createSinglevaluedIndex(
                PRIMARY_INDEX_NAME,
                EntryType.MOFID,
                EntryType.STREAMABLE);
            loadPrimaryIndex();
            jdbcConnection.commit();
        } catch (SQLException ex) {
            rollbackConnection();
            throw new JdbcStorageException(ex);
        }
    }
    
    // implement Storage
    public synchronized void open(
        boolean createOnNoExist, ObjectResolver resolver)
        throws StorageException
    {
        nextSerialNumber = readSerialNumber();
        if (nextSerialNumber == -1) {
            if (createOnNoExist) {
                create(false,resolver);
                return;
            } else {
                throw new StorageBadRequestException(
                    "Storage " + getName() + " does not exist.");
            }
        }
        try {
            openImpl();
            loadPrimaryIndex();
        } catch (SQLException ex) {
            throw new JdbcStorageException(ex);
        }
    }

    private void openImpl() throws SQLException
    {
        nameToIndexMap = new HashMap();
        sqlToPreparedStatementMap = new HashMap();
        sqlUpdateSerialNumber = new LazyPreparedStatement(
            "update " + getQualifiedTableName(MOFID_SEQ_TABLE_NAME)
            + " set " + MOFID_SEQ_COL_NAME + " = ?");
    }

    private void loadPrimaryIndex()
        throws StorageException
    {
        primaryIndex = (JdbcPrimaryIndex) getIndex(PRIMARY_INDEX_NAME);
    }
    
    // implement Storage
    public synchronized void close() throws StorageException
    {
        nameToIndexMap = null;
        sqlUpdateSerialNumber = null;
        if (sqlToPreparedStatementMap != null) {
            Iterator iter = sqlToPreparedStatementMap.values().iterator();
            while (iter.hasNext()) {
                PreparedStatement ps = (PreparedStatement) iter.next();
                closePreparedStatement(ps);
            }
            sqlToPreparedStatementMap = null;
        }
        rollbackConnection();
    }

    private void closePreparedStatement(PreparedStatement ps)
    {
        if (ps == null) {
            return;
        }
        try {
            ps.close();
        } catch (SQLException ex) {
            // TODO:  trace
        }
    }

    private void closeStatement()
    {
        if (jdbcStmt == null) {
            return;
        }
        try {
            jdbcStmt.close();
        } catch (SQLException ex) {
            // TODO:  trace
        } finally {
            jdbcStmt = null;
        }
    }

    private void closeResultSet()
    {
        if (jdbcResultSet == null) {
            return;
        }
        try {
            jdbcResultSet.close();
        } catch (SQLException ex) {
            // TODO:  trace
        } finally {
            jdbcResultSet = null;
        }
    }

    private void closeConnection()
    {
        if (jdbcConnection == null) {
            return;
        }
        try {
            jdbcConnection.close();
        } catch (SQLException ex) {
            // TODO:  trace
        }
    }

    // implement Storage
    public synchronized SinglevaluedIndex createSinglevaluedIndex(
        String name, EntryType keyType,
        EntryType valueType) throws StorageException
    {
        createIndex(name,keyType,valueType,true,true,false);
        return getSinglevaluedIndex(name);
    }

    // implement Storage
    public synchronized MultivaluedOrderedIndex createMultivaluedOrderedIndex(
        String name, EntryType keyType, EntryType valueType, boolean unique)
        throws StorageException
    {
        // NOTE:  ignore unique because it appears to lie
        createIndex(name,keyType,valueType,false,false,true);
        return getMultivaluedOrderedIndex(name);
    }

    // implement Storage
    public synchronized MultivaluedIndex createMultivaluedIndex(
        String name, EntryType keyType, EntryType valueType, boolean unique)
        throws StorageException
    {
        createIndex(name,keyType,valueType,false,unique,false);
        return getMultivaluedIndex(name);
    }

    private String getDataType(EntryType entryType)
    {
        return (String) entryTypeToDataTypeMap.get(entryType);
    }

    private EntryType getEntryType(ResultSetMetaData md,int i)
        throws SQLException
    {
        String colName = md.getColumnName(i).toUpperCase();
        int lastUnderscore = colName.lastIndexOf('_');
        String entryTypeName =
            colName.substring(lastUnderscore + 1);
        return EntryType.decodeEntryType(entryTypeName);
    }

    private String stripMofId(String indexName)
    {
        int i = indexName.indexOf(storageId);
        if (i == -1) {
            return indexName;
        }
        int j = i + storageId.length();
        if (indexName.charAt(j) != ':') {
            return indexName;
        }
        int n = indexName.length();
        for (++j; j < n; ++j) {
            if (indexName.charAt(j) != '0') {
                break;
            }
        }
        return indexName.substring(0,i) + indexName.substring(j);
    }
    
    private String getTableNameForIndex(String indexName)
    {
        // Assume we're getting something of the form
        //   :: for indexName.
        // Replace this with
        //   __
        indexName = stripMofId(indexName);
        indexName = stripMofId(indexName);
        // MySQL doesn't like ':' even in quoted identifiers, so
        // replace it with an innocuous underscore.
        indexName = indexName.replace(':','_');
        return getQualifiedTableName(indexName);
    }
    
    private void createIndex(
        String name, EntryType keyType, EntryType valueType,
        boolean singleValued,boolean uniqueValued,boolean ordered)
        throws StorageException
    {
        /*
        System.out.println(
            "creating index " + name
            + ", singleValued = " + singleValued
            + ", uniqueValued = " + uniqueValued
            + ", ordered = " + ordered);
        */
        try {
            StringBuffer sb = new StringBuffer();
            sb.append("create table ");
            sb.append(getTableNameForIndex(name));
            sb.append("(");

            String keyColName = KEY_COL_PREFIX + "_" + keyType;
            sb.append(keyColName);
            sb.append(" ");
            sb.append(getDataType(keyType));
            sb.append(" not null, ");

            String valColName;
            if (singleValued) {
                valColName = SINGLE_VAL_COL_PREFIX;
            } else {
                valColName = MULTI_VAL_COL_PREFIX;
            }
            valColName = valColName + "_" + valueType;
            sb.append(valColName);
            sb.append(" ");
            sb.append(getDataType(valueType));
            sb.append(" not null,");

            String intType = getDataType(EntryType.INT);
            
            if (ordered) {
                sb.append(ORDINAL_COL_NAME);
                sb.append(" ");
                sb.append(intType);
                sb.append(" not null,");
            }

            if (!uniqueValued) {
                sb.append(SURROGATE_COL_NAME);
                sb.append(" ");
                sb.append(intType);
                sb.append(" not null,");
            }
            
            sb.append(" primary key(");
            sb.append(keyColName);
            if (singleValued) {
                // nothing more needed
            } else if (uniqueValued) {
                sb.append(",");
                sb.append(valColName);
            } else {
                if (ordered) {
                    // NOTE: could use just (KEY,ORDINAL) as primary key.
                    // However, we have to be able to modify ordinals, and even
                    // PostgreSQL doesn't get the deferred constraint
                    // enforcement right in that case).  So, throw in both the
                    // ORDINAL and the SURROGATE so that ORDER BY can use the
                    // index.
                    sb.append(",");
                    sb.append(ORDINAL_COL_NAME);
                }
                sb.append(",");
                sb.append(SURROGATE_COL_NAME);
            }
            sb.append(")");
            
            sb.append(")");
            jdbcStmt.execute(sb.toString());
        } catch (SQLException ex) {
            throw new JdbcStorageException(ex);
        }
    }

    // implement Storage
    public synchronized SinglevaluedIndex getPrimaryIndex()
        throws StorageException
    {
        return getSinglevaluedIndex(PRIMARY_INDEX_NAME);
    }

    private Index loadIndex(String name)
        throws StorageException
    {
        PreparedStatement ps = null;
        try {
            ps = jdbcConnection.prepareStatement(
                "select * from "+getTableNameForIndex(name));
            ResultSetMetaData md = null;
            try {
                md = ps.getMetaData();
            } catch (SQLException ex) {
                // Some drivers don't support metadata pre-execution.
                // Fall through to recovery below.
            }
            if (md == null) {
                jdbcResultSet = ps.executeQuery();
                md = jdbcResultSet.getMetaData();
            }
            EntryType keyType = getEntryType(md,1);
            EntryType valueType = getEntryType(md,2);
            boolean singleValued = false;
            boolean ordered = false;
            boolean needSurrogate = false;
            String keyColName = md.getColumnName(1).toUpperCase();
            String valColName = md.getColumnName(2).toUpperCase();
            if (valColName.startsWith(SINGLE_VAL_COL_PREFIX)) {
                singleValued = true;
            }
            for (int i = 3; i <= md.getColumnCount(); ++i) {
                String colName = md.getColumnName(i).toUpperCase();
                if (colName.equals(ORDINAL_COL_NAME)) {
                    ordered = true;
                } else if (colName.equals(SURROGATE_COL_NAME)) {
                    needSurrogate = true;
                } else {
                    // TODO:  assert
                }
            }
            JdbcIndex index;
            if (singleValued) {
                if (name.equals(PRIMARY_INDEX_NAME)) {
                    index = new JdbcPrimaryIndex();
                } else {
                    index = new JdbcSinglevaluedIndex();
                }
            } else {
                if (ordered) {
                    index = new JdbcMultivaluedOrderedIndex();
                } else {
                    index = new JdbcMultivaluedIndex();
                }
            }
            index.init(
                this,
                getTableNameForIndex(name),
                name,keyColName,valColName,keyType,valueType,
                needSurrogate);
            nameToIndexMap.put(name,index);
            return index;
        } catch (SQLException ex) {
            throw new JdbcStorageException(ex);
        } finally {
            closeResultSet();
            closePreparedStatement(ps);
        }
    }

    // implement Storage
    public synchronized Index getIndex(String name) throws StorageException
    {
        synchronized(nameToIndexMap) {
            Index index = (Index) nameToIndexMap.get(name);
            if (index == null) {
                index = loadIndex(name);
            }
            return index;
        }
    }
    
    // implement Storage
    public synchronized SinglevaluedIndex getSinglevaluedIndex(String name)
        throws StorageException
    {
        return (SinglevaluedIndex) getIndex(name);
    }
    
    // implement Storage
    public synchronized MultivaluedIndex getMultivaluedIndex(String name)
        throws StorageException
    {
        return (MultivaluedIndex) getIndex(name);
    }
    
    // implement Storage
    public synchronized MultivaluedOrderedIndex getMultivaluedOrderedIndex(
        String name) throws StorageException
    {
        return (MultivaluedOrderedIndex) getIndex(name);
    }
    
    // implement Storage
    public synchronized void dropIndex(String name) throws StorageException
    {
        try {
            jdbcStmt.execute("drop table "+getTableNameForIndex(name));
        } catch (SQLException ex) {
            throw new JdbcStorageException(ex);
        }
        nameToIndexMap.remove(name);
    }

    // implement Storage
    public void objectStateWillChange(Object key) throws StorageException
    {
        // ignore
    }
    
    // implement Storage
    public synchronized void objectStateChanged(Object key)
        throws StorageException
    {
        primaryIndex.objectStateChanged(key);
    }
    
    // implement Storage
    public synchronized void commitChanges() throws StorageException
    {
        try {
            writeSerialNumber(nextSerialNumber);
            primaryIndex.flushChanges();
            jdbcConnection.commit();
        } catch (SQLException ex) {
            throw new JdbcStorageException(ex);
        }
    }

    // implement Storage
    public synchronized void rollBackChanges () throws StorageException
    {
        try {
            jdbcConnection.rollback();
        } catch (SQLException ex) {
            throw new JdbcStorageException(ex);
        }
        // NOTE:  this is taken from BtreeStorage.  It seems rather brutal,
        // but it means that we don't have to worry about state inconsistencies
        // with nameToIndexMap, primary index cache, etc.
        close();
        open(false,null);
    }
    
    // implement Storage
    public synchronized void shutDown() throws StorageException
    {
        commitChanges();
        closeStatement();
        closeConnection();
    }

    private PreparedStatement prepareStatement(
        LazyPreparedStatement lps)
        throws StorageException
    {
        if (lps.ps != null) {
            // lps is already prepared
            return lps.ps;
        }
        PreparedStatement ps = (PreparedStatement)
            sqlToPreparedStatementMap.get(lps.sql);
        if (ps != null) {
            lps.ps = ps;
            return ps;
        }
        try {
            ps = jdbcConnection.prepareStatement(lps.sql);
            sqlToPreparedStatementMap.put(lps.sql,ps);
            lps.ps = ps;
            return ps;
        } catch (SQLException ex) {
            throw new JdbcStorageException(ex);
        }
    }

    private void bindArgs(PreparedStatement ps,Object [] args)
        throws SQLException, StorageException
    {
        if (args == null) {
            return;
        }
        for (int i = 0; i < args.length; ++i) {
            Object arg = args[i];
            int iParam = i + 1;
            if (arg instanceof MOFID) {
                MOFID mofid = (MOFID) arg;
                if (!mofid.getStorageID().equals(storageId)) {
                    throw new IllegalArgumentException("Foreign MOFID");
                }
                ps.setLong(
                    iParam,
                    mofid.getSerialNumber());
            } else if (arg instanceof Streamable) {
                ps.setBytes(
                    iParam,
                    writeByteArray((Streamable) arg));
            } else {
                ps.setObject(
                    iParam,
                    arg);
            }
        }
    }

    private Object getResultObj(EntryType entryType) 
        throws StorageException, SQLException
    {
        if (entryType == EntryType.MOFID) {
            return new MOFID(
                jdbcResultSet.getLong(1),
                storageId);
        } else if (entryType == EntryType.STRING) {
            return jdbcResultSet.getString(1);
        } else if (entryType == EntryType.INT) {
            return new Long(jdbcResultSet.getLong(1));
        } else {
            byte [] bytes = jdbcResultSet.getBytes(1);
            // make a copy in case JDBC driver reuses buffer
            byte [] copy = new byte[bytes.length];
            System.arraycopy(bytes,0,copy,0,bytes.length);
            return copy;
        }
    }

    synchronized ListIterator getResultSetIterator(
        LazyPreparedStatement lps,Object [] args,EntryType entryType)
        throws StorageException
    {
        try {
            PreparedStatement ps = prepareStatement(lps);
            bindArgs(ps,args);
            jdbcResultSet = ps.executeQuery();
            // TODO:  assert exactly one column
            List list = new ArrayList();
            try {
                while (jdbcResultSet.next()) {
                    Object obj = getResultObj(entryType);
                    list.add(obj);
                }
            } finally {
                closeResultSet();
            }
            if (entryType == EntryType.STREAMABLE) {
                // Postprocess the returned byte arrays, converting them into
                // real objects.  Note that we do this outside of the
                // main fetch loop to avoid nasty reentrancy issues.
                ListIterator listIter = list.listIterator();
                while (listIter.hasNext()) {
                    byte [] bytes = (byte []) listIter.next();
                    listIter.set(readByteArray(bytes));
                }
            }
            return list.listIterator();
        } catch (SQLException ex) {
            throw new JdbcStorageException(ex);
        }
    }

    synchronized int getResultSetCount(
        LazyPreparedStatement lps,Object [] args)
        throws StorageException
    {
        try {
            PreparedStatement ps = prepareStatement(lps);
            bindArgs(ps,args);
            jdbcResultSet = ps.executeQuery();
            try {
                int n = 0;
                while (jdbcResultSet.next()) {
                    ++n;
                }
                return n;
            } finally {
                closeResultSet();
            }
        } catch (SQLException ex) {
            throw new JdbcStorageException(ex);
        }
    }

    synchronized int getSingletonInt(
        LazyPreparedStatement lps,Object [] args)
        throws StorageException
    {
        try {
            PreparedStatement ps = prepareStatement(lps);
            bindArgs(ps,args);
            jdbcResultSet = ps.executeQuery();
            try {
                // TODO:  assert exactly one column
                if (!jdbcResultSet.next()) {
                    return -1;
                }
                return jdbcResultSet.getInt(1);
            } finally {
                closeResultSet();
            }
        } catch (SQLException ex) {
            throw new JdbcStorageException(ex);
        }
    }

    private byte [] writeByteArray(Streamable data)
        throws StorageException
    {
        try {
            ByteArrayOutputStream byteStream = new ByteArrayOutputStream();
            DataOutputStream dataStream = new DataOutputStream(byteStream);
            dataStream.writeUTF(data.getClass().getName());
            data.write(dataStream);
            dataStream.flush();
            return byteStream.toByteArray();
        } catch (IOException ex) {
            throw new StorageIOException(ex);
        }
    }

    private Streamable readByteArray(byte [] bytes)
        throws StorageException
    {
        ByteArrayInputStream byteStream = new ByteArrayInputStream(bytes);
        DataInputStream dataStream = new DataInputStream(byteStream);

        // TODO:  rip classCode stuff from BtreeDatabase
        String className;
        Streamable data;
        
        try {
            className = dataStream.readUTF();
        } catch (IOException ex) {
            throw new StorageIOException(ex);
        }
        try {
            Class cls = Class.forName(className);
            data = (Streamable)cls.newInstance();
        } catch (Exception ex) {
            throw new StoragePersistentDataException(ex.getMessage());
        }
        if (data instanceof StorageClient) {
            ((StorageClient)data).setStorage(this);
        }
        data.read(dataStream);
        return data;
    }

    synchronized int executeUpdate(LazyPreparedStatement lps,Object [] args)
        throws StorageException
    {
        try {
            PreparedStatement ps = prepareStatement(lps);
            bindArgs(ps,args);
            return ps.executeUpdate();
        } catch (SQLException ex) {
            throw new JdbcStorageException(ex);
        }
    }
}

// End JdbcStorage.java