/*
    * $Source: /usr/cvsroot/melati/poem/src/main/java/org/melati/poem/JdbcTable.java,v $
    * $Revision: 1.14 $
    *
    * Copyright (C) 2008 Tim Pizey
    * 
    * Part of Melati (http://melati.org), a framework for the rapid
    * development of clean, maintainable web applications.
    *
   * Melati is free software; Permission is granted to copy, distribute
   * and/or modify this software under the terms either:
   *
   * a) the GNU General Public License as published by the Free Software
   *    Foundation; either version 2 of the License, or (at your option)
   *    any later version,
   *
   *    or
   *
   * b) any version of the Melati Software License, as published
   *    at http://melati.org
   *
   * You should have received a copy of the GNU General Public License and
   * the Melati Software License along with this program;
   * if not, write to the Free Software Foundation, Inc.,
   * 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA to obtain the
   * GNU General Public License and visit http://melati.org to obtain the
   * Melati Software License.
   *
   * Feel free to contact the Developers of Melati (http://melati.org),
   * if you would like to work out a different arrangement than the options
   * outlined here.  It is our intention to allow Melati to be used by as
   * wide an audience as possible.
   *
   * This program is distributed in the hope that it will be useful,
   * but WITHOUT ANY WARRANTY; without even the implied warranty of
   * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   * GNU General Public License for more details.
   *
   * Contact details for copyright holder:
   *
   *     Tim Pizey <timp At paneris.org>
   *     http://paneris.org/~timp
   */
  
  package org.melati.poem;
  
  import java.io.PrintStream;
  import java.sql.Connection;
  import java.sql.PreparedStatement;
  import java.sql.ResultSet;
  import java.sql.SQLException;
  import java.sql.Statement;
  import java.util.Enumeration;
  import java.util.Hashtable;
  import java.util.Vector;
  
  import org.melati.poem.dbms.Dbms;
  import org.melati.poem.transaction.Transactioned;
  import org.melati.poem.transaction.TransactionedSerial;
  import org.melati.poem.util.ArrayEnumeration;
  import org.melati.poem.util.ArrayUtils;
  import org.melati.poem.util.Cache;
  import org.melati.poem.util.CachedIndexFactory;
  import org.melati.poem.util.EnumUtils;
  import org.melati.poem.util.MappedEnumeration;
  import org.melati.poem.util.Procedure;
  import org.melati.poem.util.FilteredEnumeration;
  import org.melati.poem.util.FlattenedEnumeration;
  import org.melati.poem.util.Order;
  import org.melati.poem.util.SortUtils;
  import org.melati.poem.util.StringUtils;
  
  /**
   * A Table.
   * @since 14 April 2008 
   */
  public class JdbcTable implements Selectable, Table {
  
    /** Default limit for row cache. */
    private static final int CACHE_LIMIT_DEFAULT = 100;
    private static final int DISPLAY_ORDER_DEFAULT = 100;
  
    private JdbcTable _this = this;
  
    private Database database;
    private String name;
    private String quotedName;
    private DefinitionSource definitionSource;
  
    private TableInfo info = null;
  
    private TableListener[] listeners = {};
  
    private Column[] columns = {};
    private Hashtable columnsByName = new Hashtable();
  
    private Column troidColumn = null;
    private Column deletedColumn = null;
    private Column canReadColumn = null;
   private Column canSelectColumn = null;
   private Column canWriteColumn = null;
   private Column canDeleteColumn = null;
   private Column displayColumn = null;
   private Column searchColumn = null;
 
   private String defaultOrderByClause = null;
 
   private Column[][] displayColumns = new Column[DisplayLevel.count()][];
   private Column[] searchColumns = null;
 
   private TransactionedSerial serial;
 
   private CachedSelection allTroids = null;
   private Hashtable cachedSelections = new Hashtable();
   private Hashtable cachedCounts = new Hashtable();
   private Hashtable cachedExists = new Hashtable();
 
   private int mostRecentTroid = -1;
   private int extrasIndex = 0;
 
   
   /**
    * Constructor.
    */
   public JdbcTable(Database database, String name,
                DefinitionSource definitionSource) {
     this.database = database;
     this.name = name;
     // Don't do this here as the database does not know about the dbms yet
     // this.quotedName = database.quotedName(name);
     // this is actually set the first time it is accessed in quotedName()
     this.definitionSource = definitionSource;
     serial = new TransactionedSerial(database);
   }
 
   /**
    * Do stuff immediately after table initialisation.
    * <p>
    * This base method clears the column info caches and adds a listener
    * to the column info table to maintain the caches.
    * <p>
    * It may be overridden to perform other actions. For example to
    * ensure required rows exist in tables that define numeric ID's for
    * codes.
    *
    * @see #notifyColumnInfo(ColumnInfo)
    * @see #clearColumnInfoCaches()
    */
   public void postInitialise() {
     clearColumnInfoCaches();
     database.getColumnInfoTable().addListener(
         new TableListener() {
           public void notifyTouched(PoemTransaction transaction, Table table,
                                     Persistent persistent) {
             _this.notifyColumnInfo((ColumnInfo)persistent);
           }
 
           public void notifyUncached(Table table) {
             _this.clearColumnInfoCaches();
           }
         });
   }
 
   // 
   // ===========
   //  Accessors
   // ===========
   // 
 
   /**
    * The database to which the table is attached.
    * @return the db
    */
   public final Database getDatabase() {
     return database;
   }
 
   /** 
    * The table's programmatic name.  Identical with its name in the DSD (if the
    * table was defined there) and in its <TT>tableinfo</TT> entry.
    * This will normally be the same as the name in the RDBMS itself, however that name 
    * may be translated to avoid DBMS specific name clashes. 
    *
    * @return the table name, case as defined in the DSD
    * @see org.melati.poem.dbms.Dbms#melatiName(String)
    */
   public final String getName() {
     return name;
   }
 
  /**
   * @return table name quoted using the DBMS' specific quoting rules.
   */
   public final String quotedName() {
     if (quotedName == null) quotedName = database.quotedName(name);
     return quotedName;
   }
 
  /**
   * The human-readable name of the table.  POEM itself doesn't use this, but
   * it's available to applications and Melati's generic admin system as a
   * default label for the table and caption for its records.
    * @return The human-readable name of the table
    */
   public final String getDisplayName() {
     return info.getDisplayname();
   }
 
  /**
   * A brief description of the table's function.  POEM itself doesn't use
   * this, but it's available to applications and Melati's generic admin system
   * as a default label for the table and caption for its records.
   * @return the brief description
   */
   public final String getDescription() {
     return info.getDescription();
   }
 
   /**
    * The category of this table.  POEM itself doesn't use
    * this, but it's available to applications and Melati's generic admin system
    * as a default label for the table and caption for its records.
    * 
    * @return the category
    */
   public final TableCategory getCategory() {
      return info.getCategory();
   }
 
  /**
   * @return the {@link TableInfo} for this table
   */
   public final TableInfo getInfo() {
      return info;
   }
 
  /**
   * The troid (<TT>id</TT>) of the table's entry in the <TT>tableinfo</TT>
   * table.  It will always have one (except during initialisation, which the
   * application programmer will never see).
   * 
   * @return id in TableInfo metadata table
   */
   public final Integer tableInfoID() {
     return info == null ? null : info.troid();
   }
 
   /**
    * The table's column with a given name.  If the table is defined in the DSD
    * under the name <TT><I>foo</I></TT>, there will be an
    * application-specialised <TT>Table</TT> subclass, called
    * <TT><I>Foo</I>Table</TT> (and available as <TT>get<I>Foo</I>Table</TT>
    * from the application-specialised <TT>Database</TT> subclass) which has
    * extra named methods for accessing the table's predefined <TT>Column</TT>s.
    *
    * @param nameP name of column to get
    * @return column of that name
    * @throws NoSuchColumnPoemException if there is no column with that name
    */
   public final Column getColumn(String nameP) throws NoSuchColumnPoemException {
     Column column = _getColumn(nameP); 
     if (column == null)
       throw new NoSuchColumnPoemException(this, nameP);
     else
       return column;
   }
   protected final Column _getColumn(String nameP) {
     Column column = (Column)columnsByName.get(nameP.toLowerCase());    
     return column;
   }
   
   /**
    * All the table's columns.
    *
    * @return an <TT>Enumeration</TT> of <TT>Column</TT>s
    * @see Column
    */
   public final Enumeration columns() {
     return new ArrayEnumeration(columns);
   }
 
  /**
   * @return the number of columns in this table.
   */
   public final int getColumnsCount() {
     return columns.length;
   }
 
   /**
    * @param columnInfoID
    * @return the Column with a TROID equal to columnInfoID
    */
   public Column columnWithColumnInfoID(int columnInfoID) {
     for (Enumeration c = columns(); c.hasMoreElements();) {
       Column column = (Column)c.nextElement();
       Integer id = column.columnInfoID();
       if (id != null && id.intValue() == columnInfoID)
         return column;
     }
     return null; // Happens when columns exist but are not defined in DSD
   }
 
   /**
    * The table's troid column.  Every table in a POEM database must have a
    * troid (table row ID, or table-unique non-nullable integer primary key),
    * often but not necessarily called <TT>id</TT>, so that it can be
    * conveniently `named'.
    *
    * @return the id column
    * @see #getObject(java.lang.Integer)
    */
   public final Column troidColumn() {
     return troidColumn;
   }
 
   /**
    * @return The table's deleted-flag column, if any.
    */
   public final Column deletedColumn() {
     return deletedColumn;
   }
 
   /**
    * The table's primary display column, the Troid column if not set.  
    * This is the column used to represent records from the table 
    * concisely in reports or whatever.  It is determined 
    * at initialisation time by examining the <TT>Column</TT>s
    * <TT>getPrimaryDisplay()</TT> flags.
    *
    * @return the table's display column, or <TT>null</TT> if it hasn't got one
    *
    * see Column#setColumnInfo
    * @see ReferencePoemType#_stringOfCooked
    * @see DisplayLevel#primary
    */
   public final Column displayColumn() {
     return displayColumn == null ? troidColumn : displayColumn;
   }
 
   /**
    * @param column the display column to set
    */
   public final void setDisplayColumn(Column column) {
     displayColumn = column;
   }
 
  /**
   * In a similar manner to the primary display column, each table can have 
   * one primary criterion column.
   * <p>
   * The Primary Criterion is the main grouping field of the table, 
   * ie the most important non-unique type field.
   * <p>
   * For example the Primary Criterion for a User table might be Nationality.
   *
   * @return the search column, if any
   * @see Searchability
   */
   public final Column primaryCriterionColumn() {
     return searchColumn;
   }
 
   /**
    * @param column the search column to set
    */
   public void setSearchColumn(Column column) {
     searchColumn = column;
   }
 
   /**
    * If the troidColumn has yet to be set then returns an empty string.
    *  
    * @return comma separated list of the columns to order by
    */
   public String defaultOrderByClause() {
     String clause = defaultOrderByClause;
 
     if (clause == null) {
       clause = EnumUtils.concatenated(
           ", ",
           new MappedEnumeration(new ArrayEnumeration(SortUtils.sorted(
               new Order() {
                 public boolean lessOrEqual(Object a, Object b) {
                   return
                       ((Column)a).getDisplayOrderPriority().intValue() <=
                       ((Column)b).getDisplayOrderPriority().intValue();
                 }
               },
               new FilteredEnumeration(columns()) {
                 public boolean isIncluded(Object column) {
                   return ((Column)column).getDisplayOrderPriority() != null;
                 }
               }))) {
             public Object mapped(Object column) {
               String sort = ((Column)column).fullQuotedName();
               if (((Column)column).getSortDescending()) sort += " desc";
               return sort;
             }
           });
 
       if (clause.equals("") && displayColumn() != null)
         clause = displayColumn().fullQuotedName();
 
       defaultOrderByClause = clause;
     }
 
     return clause;
   }
 
   /**
    * Clear caches.
    */
   public void clearColumnInfoCaches() {
     defaultOrderByClause = null;
     for (int i = 0; i < displayColumns.length; ++i)
       displayColumns[i] = null;
   }
 
   /**
    * Clears columnInfo caches, normally a no-op.
    * 
    * @param infoP the possibly null ColumnInfo meta-data persistent
    */
   public void notifyColumnInfo(ColumnInfo infoP) {
     // FIXME info == null means deleted: effect is too broad really
     if (infoP == null || infoP.getTableinfo_unsafe().equals(tableInfoID()))
       clearColumnInfoCaches();
   }
 
   /**
    * Get an Array of columns meeting the criteria of whereClause.
    * 
    * It is the programmer's responsibility to ensure that the where clause 
    * is suitable for the target DBMS.
    * 
    * @param whereClause an SQL snippet
    * @return an array of Columns
    */
   private Column[] columnsWhere(String whereClause) {
     // get the col IDs from the committed session
     Enumeration colIDs =
         getDatabase().getColumnInfoTable().troidSelection(
             database.quotedName("tableinfo") + " = " + tableInfoID() + 
               " AND (" + whereClause + ")",
             null, false, PoemThread.inSession() ? PoemThread.transaction() : null);
 
     Vector them = new Vector();
     while (colIDs.hasMoreElements()) {
       Column column =
           columnWithColumnInfoID(((Integer)colIDs.nextElement()).intValue());
       // null shouldn't happen but let's not gratuitously fail if it does
       if (column != null)
         them.addElement(column);
     }
 
     Column[] columnsLocal = new Column[them.size()];
     them.copyInto(columnsLocal);
     return columnsLocal;
   }
 
   /**
    * Return columns at a display level in display order.
    *
    * @param level the {@link DisplayLevel} to select
    * @return an Enumeration of columns at the given level
    */ 
   public final Enumeration displayColumns(DisplayLevel level) {
     Column[] columnsLocal = displayColumns[level.getIndex().intValue()];
 
     if (columnsLocal == null) {
       columnsLocal =
         columnsWhere(database.quotedName("displaylevel") + " <= " + 
                                                          level.getIndex());
       displayColumns[level.getIndex().intValue()] = columnsLocal;
     }
     return new ArrayEnumeration(columnsLocal);
   }
 
   /**
    * @param level the {@link DisplayLevel} to select
    * @return the number of columns at a display level.
    */ 
   public final int displayColumnsCount(DisplayLevel level) {
     int l = level.getIndex().intValue();
     if (displayColumns[l] == null)
       // FIXME Race 
       displayColumns(level);
 
     return displayColumns[l].length;
   }
 
   /**
    * The table's columns for detailed display in display order.
    *
    * @return an <TT>Enumeration</TT> of <TT>Column</TT>s
    * @see Column
    * @see #displayColumns(DisplayLevel)
    * @see DisplayLevel#detail
    */
   public final Enumeration getDetailDisplayColumns() {
     return displayColumns(DisplayLevel.detail);
   }
 
   /**
    * @return the number of columns at display level <tt>Detail</tt>
    */ 
   public final int getDetailDisplayColumnsCount() {
     return displayColumnsCount(DisplayLevel.detail);
   }
 
   /**
    * The table's columns designated for display in a record, in display order.
    *
    * @return an <TT>Enumeration</TT> of <TT>Column</TT>s
    * @see Column
    * @see #displayColumns(DisplayLevel)
    * @see DisplayLevel#record
    */
   public final Enumeration getRecordDisplayColumns() {
     return displayColumns(DisplayLevel.record);
   }
 
   /**
    * @return the number of columns at display level <tt>Record</tt>
    */ 
   public final int getRecordDisplayColumnsCount() {
     return displayColumnsCount(DisplayLevel.record);
   }
 
   /**
    * The table's columns designated for display in a record summary, in display
    * order.
    *
    * @return an <TT>Enumeration</TT> of <TT>Column</TT>s
    * @see Column
    * @see #displayColumns(DisplayLevel)
    * @see DisplayLevel#summary
    */
   public final Enumeration getSummaryDisplayColumns() {
     return displayColumns(DisplayLevel.summary);
   }
   
   /**
    * @return the number of columns at display level <tt>Summary</tt>
    */ 
   public final int getSummaryDisplayColumnsCount() {
     return displayColumnsCount(DisplayLevel.summary);
   }
 
   /**
    * The table's columns designated for use as search criteria, in display
    * order.
    *
    * @return an <TT>Enumeration</TT> of <TT>Column</TT>s
    * @see Column
    */
   public final Enumeration getSearchCriterionColumns() {
     Column[] columnsLocal = searchColumns;
 
     if (columnsLocal == null) {
       columnsLocal = 
          columnsWhere(database.quotedName("searchability") + " <= " +
                                           Searchability.yes.getIndex());
       searchColumns = columnsLocal;
     }
     return new ArrayEnumeration(searchColumns);
   }
 
   /**
    * @return the number of columns which are searchable
    */ 
   public final int getSearchCriterionColumnsCount() {
     if (searchColumns == null)
       // FIXME Race 
       getSearchCriterionColumns();
       
     return searchColumns.length;
   }
 
   private Dbms dbms() {
     return getDatabase().getDbms();
   }
 
   // 
   // =========================
   //  Low-level DB operations
   // =========================
   // 
 
   // 
   // -----------
   //  Structure
   // -----------
   // 
 
   /**
    * Use this for DDL statements, ie those which alter the structure of the db.
    * Postgresql in particular does not like DDL statements being executed within a transaction.
    * 
    * @param sql the SQL DDL statement to execute
    * @throws StructuralModificationFailedPoemException
    */
   public void dbModifyStructure(String sql)
       throws StructuralModificationFailedPoemException {
     
     // We have to do this to avoid blocking
     if (PoemThread.inSession())
       PoemThread.commit();
 
     try {
       if (database.logSQL()) database.log("about to execute:" + sql);
 
       Statement updateStatement = database.getCommittedConnection().createStatement();
       updateStatement.executeUpdate(sql);
       updateStatement.close();
       database.getCommittedConnection().commit();
       if (database.logCommits()) database.log(new CommitLogEvent(null));
       if (database.logSQL()) database.log(new StructuralModificationLogEvent(sql));
       database.incrementQueryCount(sql);
     }
     catch (SQLException e) {
       throw new StructuralModificationFailedPoemException(sql, e);
     }
   }
 
   private void dbCreateTable() {
     String createTableSql = dbms().createTableSql(this);
     dbModifyStructure(createTableSql);
     String tableSetup = database.getDbms().tableInitialisationSql(this); 
     if (tableSetup != null) { 
       dbModifyStructure(tableSetup);
     }
   }
   
     
 
   /**
    * @return A type string eg "TEXT"
    * @see {@link org.melati.poem.dbms.Hsqldb}
    */
   public String getDbmsTableType() {
     return null;
   }
 
   /**
    * Constraints are not used in POEM, but you might want to use them if 
    * exporting the db or using schema visualisation tools.
    */
   public void dbAddConstraints() {
     StringBuffer sqb = new StringBuffer();
     for (int c = 0; c < columns.length; ++c) {
       if (columns[c].getSQLType() instanceof TroidPoemType){
         sqb.append("ALTER TABLE " + quotedName());
         sqb.append(dbms().getPrimaryKeyDefinition(
             columns[c].getName()));
         try {
           dbModifyStructure(sqb.toString());
         } catch (StructuralModificationFailedPoemException e) {
           // It is more expensive to only add constaints 
           // if they are missing than to ignore exceptions.  
           e = null;
         }
       }
     }
     for (int c = 0; c < columns.length; ++c) {
       if (columns[c].getSQLType() instanceof ReferencePoemType){
         IntegrityFix fix = columns[c].getIntegrityFix();
         sqb = new StringBuffer();
         sqb.append("ALTER TABLE " + quotedName());
         sqb.append(dbms().getForeignKeyDefinition(
                       getName(),
                       columns[c].getName(),
                       ((ReferencePoemType)columns[c].getSQLType()).
                           targetTable().getName(),
                       ((ReferencePoemType)columns[c].getSQLType()).
                           targetTable().troidColumn().getName(),
                        fix.getName()));
         try {
           dbModifyStructure(sqb.toString());
         } catch (StructuralModificationFailedPoemException e) {
           // It is more expensive to only add constaints 
           // if they are missing than to ignore exceptions.  
           e = null;          
         }
       }
     }
 
 
   }
 
   private void dbAddColumn(Column column) {
     if (column.getType().getNullable()) {
       dbModifyStructure(
           "ALTER TABLE " + quotedName() +
           " ADD " + column.quotedName() +
           " " + column.getSQLType().sqlDefinition(dbms()));
     } else {
       dbModifyStructure(
           "ALTER TABLE " + quotedName() +
           " ADD " + column.quotedName() +
           " " + column.getSQLType().sqlTypeDefinition(dbms()));
       dbModifyStructure(
           "UPDATE " + quotedName() +
           " SET " + column.quotedName() +
           " = " + dbms().getQuotedValue(column.getSQLType(), 
                       column.getSQLType().sqlDefaultValue(dbms())));
       dbModifyStructure(
           dbms().alterColumnNotNullableSQL(name, column));      
     }
   }
 
   
   private void dbCreateIndex(Column column) {
     if (column.getIndexed()) {
       if (!dbms().canBeIndexed(column)) {
         database.log(new UnindexableLogEvent(column));
       } else {
         dbModifyStructure(
             "CREATE " + (column.getUnique() ? "UNIQUE " : "") + "INDEX " +
             indexName(column) +
             " ON " + quotedName() + " " +
             "(" + column.quotedName() + 
              dbms().getIndexLength(column) + ")");
       }
     }
   }
 
   private String indexName(Column column) { 
     return database.quotedName(
             dbms().unreservedName(name) + "_" + 
             dbms().unreservedName(column.getName()) + "_index");
   }
   // 
   // -------------------------------
   //  Standard `PreparedStatement's
   // -------------------------------
   // 
 
   /**
    * 
    * @param connection the connection the PreparedStatement is tied to
    * @return a PreparedStatment to perform a simple INSERT
    */
   private PreparedStatement simpleInsert(Connection connection) {
     StringBuffer sql = new StringBuffer();
     sql.append("INSERT INTO " + quotedName() + " (");
     for (int c = 0; c < columns.length; ++c) {
       if (c > 0) sql.append(", ");
       sql.append(columns[c].quotedName());
     }
     sql.append(") VALUES (");
     for (int c = 0; c < columns.length; ++c) {
       if (c > 0) sql.append(", ");
       sql.append("?");
     }
 
     sql.append(")");
 
     try {
       return connection.prepareStatement(sql.toString());
     }
     catch (SQLException e) {
       throw new SimplePrepareFailedPoemException(sql.toString(), e);
     }
   }
 
   private PreparedStatement simpleGet(Connection connection) {
     StringBuffer sql = new StringBuffer();
     sql.append("SELECT ");
     for (int c = 0; c < columns.length; ++c) {
       if (c > 0) sql.append(", ");
       sql.append(columns[c].quotedName());
     }
     sql.append(" FROM " + quotedName() +
                " WHERE " + troidColumn.quotedName() + " = ?");
 
     try {
       return connection.prepareStatement(sql.toString());
     }
     catch (SQLException e) {
       throw new SimplePrepareFailedPoemException(sql.toString(), e);
     }
   }
 
   private PreparedStatement simpleModify(Connection connection) {
     // FIXME synchronize this too
     StringBuffer sql = new StringBuffer();
     sql.append("UPDATE " + quotedName() + " SET ");
     for (int c = 0; c < columns.length; ++c) {
       if (c > 0) sql.append(", ");
       sql.append(columns[c].quotedName());
       sql.append(" = ?");
     }
     sql.append(" WHERE " + troidColumn.quotedName() + " = ?");
 
     try {
       return connection.prepareStatement(sql.toString());
     }
     catch (SQLException e) {
       throw new SimplePrepareFailedPoemException(sql.toString(), e);
     }
   }
 
   // 
   // -----------------------------
   //  Transaction-specific things
   // -----------------------------
   // 
 
   private class TransactionStuff {
     PreparedStatement insert, modify, get;
     TransactionStuff(Connection connection) {
       insert = _this.simpleInsert(connection);
       modify = _this.simpleModify(connection);
       get = _this.simpleGet(connection);
     }
   }
 
   private CachedIndexFactory transactionStuffs = new CachedIndexFactory() {
     public Object reallyGet(int index) {
       // "Table.this" is attempt to work around Dietmar's problem with JDK1.3.1
       return new TransactionStuff(
           JdbcTable.this.database.poemTransaction(index).getConnection());
     }
   };
 
   private TransactionStuff committedTransactionStuff = null;
 
   /**
    * When deleting a table and used in tests.
    */
   public void invalidateTransactionStuffs() { 
     transactionStuffs.invalidate();
   }
   /**
    * Called when working outside a Transaction.
    * @return the TransactionStuff for the committed transaction
    * @see org.melati.poem.PoemDatabase#inCommittedTransaction(AccessToken, PoemTask)
    */
   private synchronized TransactionStuff getCommittedTransactionStuff() {
     if (committedTransactionStuff == null)
       committedTransactionStuff =
           new TransactionStuff(database.getCommittedConnection());
     return committedTransactionStuff;
   }
 
   // 
   // --------------------
   //  Loading and saving
   // --------------------
   // 
 
   private void load(PreparedStatement select, Persistent p) {
     JdbcPersistent persistent = (JdbcPersistent)p;
     try {
       synchronized (select) {
         select.setInt(1, persistent.troid().intValue());
         ResultSet rs = select.executeQuery();
         if (database.logSQL())
           database.log(new SQLLogEvent(select.toString()));
         database.incrementQueryCount(select.toString());
         try {
           if (!rs.next())
             persistent.setStatusNonexistent();
           else {
             persistent.setStatusExistent();
             for (int c = 0; c < columns.length; ++c)
               columns[c].load_unsafe(rs, c + 1, persistent);
           }
           persistent.setDirty(false);
           persistent.markValid();
           if (rs.next())
             throw new DuplicateTroidPoemException(this, persistent.troid());
         }
         finally {
           try { rs.close(); } catch (Exception e) {
             System.err.println("Cannot close resultset after exception.");  
           }
         }
       }
     }
     catch (SQLException e) {
       throw new SimpleRetrievalFailedPoemException(e, select.toString());
     }
     catch (ValidationPoemException e) {
       throw new UnexpectedValidationPoemException(e);
     }
   }
 
   /**
    * @param transaction possibly null if working with the committed transaction
    * @param persistent the Persistent to load
    */
   public void load(PoemTransaction transaction, Persistent persistent) {
     load(transaction == null ?
             getCommittedTransactionStuff().get :
             ((TransactionStuff)transactionStuffs.get(transaction.index)).get,
          persistent);
   }
 
   private void modify(PoemTransaction transaction, Persistent persistent) {
     PreparedStatement modify =
         ((TransactionStuff)transactionStuffs.get(transaction.index)).modify;
     synchronized (modify) {
       for (int c = 0; c < columns.length; ++c)
         columns[c].save_unsafe(persistent, modify, c + 1);
 
       try {
         modify.setInt(columns.length + 1, persistent.troid().intValue());
       }
       catch (SQLException e) {
         throw new SQLSeriousPoemException(e);
       }
 
       try {
         modify.executeUpdate();
       }
       catch (SQLException e) {
         throw dbms().exceptionForUpdate(this, modify, false, e);
       }
       database.incrementQueryCount(modify.toString());
 
       if (database.logSQL())
         database.log(new SQLLogEvent(modify.toString()));
     }
     persistent.postModify();
   }
 
   private void insert(PoemTransaction transaction, Persistent persistent) {
     
     PreparedStatement insert =
         ((TransactionStuff)transactionStuffs.get(transaction.index)).insert;
     synchronized (insert) {
       for (int c = 0; c < columns.length; ++c)
         columns[c].save_unsafe(persistent, insert, c + 1);
       try {
         insert.executeUpdate();
       }
       catch (SQLException e) {
         throw dbms().exceptionForUpdate(this, insert, true, e);
       }
       database.incrementQueryCount(insert.toString());
       if (database.logSQL())
         database.log(new SQLLogEvent(insert.toString()));
     }
     persistent.postInsert();
   }
 
   /**
    * The Transaction cannot be null, as this is trapped in 
    * #deleteLock(SessionToken).
    * @param troid id of row to delete
    * @param transaction a non-null transaction
    */
   public void delete(Integer troid, PoemTransaction transaction) {
     String sql =
         "DELETE FROM " + quotedName() +
         " WHERE " + troidColumn.quotedName() + " = " +
         troid.toString();
     try {
       transaction.writeDown();
       Connection connection = transaction.getConnection();
 
       Statement deleteStatement = connection.createStatement();
       int deleted = deleteStatement.executeUpdate(sql);
       if (deleted != 1) { 
         throw new RowDisappearedPoemException(this,troid);
       }
       deleteStatement.close();
       database.incrementQueryCount(sql);
       if (database.logSQL())
         database.log(new SQLLogEvent(sql));
 
       cache.delete(troid);
     }
     catch (SQLException e) {
       throw new ExecutingSQLPoemException(sql, e);
     }
   }
 
   /**
    * @param transaction our PoemTransaction 
    * @param p the Persistent to write
    */
   public void writeDown(PoemTransaction transaction, Persistent p) {
     JdbcPersistent persistent = (JdbcPersistent)p;
     // NOTE No race, provided that the one-thread-per-transaction parity is
     // maintained
 
     if (persistent.isDirty()) {
       troidColumn.setRaw_unsafe(persistent, persistent.troid());
 
       if (persistent.statusExistent()) {
         modify(transaction, persistent);
       } else if (persistent.statusNonexistent()) {
         insert(transaction, persistent);
         persistent.setStatusExistent();
       }
 
       persistent.setDirty(false);
       persistent.postWrite();
     }
   }
 
   // 
   // ============
   //  Operations
   // ============
   // 
 
   // 
   // ----------
   //  Cacheing
   // ----------
   // 
 
   private Cache cache = new Cache(CACHE_LIMIT_DEFAULT);
 
   private static final Procedure invalidator =
       new Procedure() {
         public void apply(Object arg) {
           ((Transactioned)arg).invalidate();
         }
       };
 
   /**
    * Invalidate table cache.
    * 
    * NOTE Invalidated cache elements are reloaded when next read
    */
   public void uncache() {
     cache.iterate(invalidator);
     serial.invalidate();
     TableListener[] listenersLocal = this.listeners;
     for (int l = 0; l < listenersLocal.length; ++l)
       listenersLocal[l].notifyUncached(this);
   }
 
   /**
    * @param maxSize new maximum size
    */
   public void trimCache(int maxSize) {
     cache.trim(maxSize);
   }
 
   /**
    * @return the Cache Info object
    */ 
   public Cache.Info getCacheInfo() {
     return cache.getInfo();
   }
 
   /**
    * Add a {@link TableListener} to this Table.
    */ 
   public void addListener(TableListener listener) {
     listeners = (TableListener[])ArrayUtils.added(listeners, listener);
   }
 
   /**
    * Notify the table that one if its records is about to be changed in a
    * transaction.  You can (with care) use this to support cacheing of
    * frequently-used facts about the table's records.  
    *
    * @param transaction the transaction in which the change will be made
    * @param persistent  the record to be changed
    */
   public void notifyTouched(PoemTransaction transaction, Persistent persistent) {
     serial.increment(transaction);
 
     TableListener[] listenersLocal = this.listeners;
     for (int l = 0; l < listenersLocal.length; ++l)
       listenersLocal[l].notifyTouched(transaction, this, persistent);
   }
 
   /**
    * @return the Transaction serial 
    */ 
   public long serial(PoemTransaction transaction) {
     return serial.current(transaction);
   }
 
   /**
    * Lock this record.
    */ 
   public void readLock() {
     serial(PoemThread.transaction());
   }
 
   // 
   // ----------
   //  Fetching
   // ----------
   // 
 
   /**
    * The object from the table with a given troid.
    *
    * @param troid       Every record (object) in a POEM database must have a
    *                    troid (table row ID, or table-unique non-nullable
    *                    integer primary key), often but not necessarily called
    *                    <TT>id</TT>, so that it can be conveniently `named' for
    *                    retrieval by this method.
    *
    * @return A <TT>Persistent</TT> of the record with the given troid;
    *         or, if the table was defined in the DSD under the name
    *         <TT><I>foo</I></TT>, an application-specialised subclass
    *         <TT><I>Foo</I></TT> of <TT>Persistent</TT>.  In that case, there
    *         will also be an application-specialised <TT>Table</TT> subclass,
    *         called <TT><I>Foo</I>Table</TT> (and available as
    *         <TT>get<I>Foo</I>Table</TT> from the application-specialised
    *         <TT>Database</TT> subclass), which has a matching method
    *         <TT>get<I>Foo</I>Object</TT> for obtaining the specialised object
    *         under its own type.  Note that no access checks are done at this
    *         stage: you may not be able to do anything with the object handle
    *         returned from this method without provoking a
    *         <TT>PoemAccessException</TT>.
    *
    * @exception NoSuchRowPoemException
    *                if there is no row in the table with the given troid
    *
    * @see Persistent#getTroid()
    */
   public Persistent getObject(Integer troid) throws NoSuchRowPoemException {
     JdbcPersistent persistent = (JdbcPersistent)cache.get(troid);
 
     if (persistent == null) {
       persistent = (JdbcPersistent)newPersistent();
       claim(persistent, troid);
       load(PoemThread.transaction(), persistent);
       if (persistent.statusExistent())
         synchronized (cache) {
           JdbcPersistent tryAgain = (JdbcPersistent)cache.get(troid);
           if (tryAgain == null)
             cache.put(troid, persistent);
           else
             persistent = tryAgain;
         }
     }
 
     if (!persistent.statusExistent())
       throw new NoSuchRowPoemException(this, troid);
 
     persistent.existenceLock(PoemThread.sessionToken());
 
     return persistent;
   }
 
   /**
    * The object from the table with a given troid.  See previous.
    *
    * @param troid the table row id
    * @return the Persistent
    * @throws NoSuchRowPoemException if not found
    * @see #getObject(java.lang.Integer)
    */
   public Persistent getObject(int troid) throws NoSuchRowPoemException {
     return getObject(new Integer(troid));
   }
 
   // 
   // -----------
   //  Searching
   // -----------
   // 
 
   /**
    * The from clause has been added as an argument because it is
    * inextricably linked to the where clause, but the default is 
    * {@link #quotedName()}.
    *
    * It is the programmer's responsibility to ensure that the where clause 
    * is suitable for the target DBMS.
    * 
    * @param fromClause Comma separated list of table names or null for default.
    * @param whereClause SQL fragment
    * @param orderByClause Comma separated list
    * @param includeDeleted Flag as to whether to include soft deleted records
    * @param excludeUnselectable Whether to append unselectable exclusion SQL 
    * @todo Should work within some kind of limit
    * @return an SQL SELECT statement put together from the arguments and
    * default order by clause.
    */
   public String selectionSQL(String fromClause, String whereClause, 
                              String orderByClause, boolean includeDeleted, 
                              boolean excludeUnselectable) {
     return selectOrCountSQL(troidColumn().fullQuotedName(),
                             fromClause, whereClause, orderByClause,
                             includeDeleted, excludeUnselectable);
   }
 
   /**
    * It is the programmer's responsibility to ensure that the where clause 
    * is suitable for the target DBMS.
    * 
    * @param fromClause SQL fragment
    * @param whereClause SQL fragment
    * @param orderByClause comma separated list
    * @param includeDeleted flag as to whether to include soft deleted records
    * @param excludeUnselectable whether to append unselectable exclusion SQL 
    * @param transaction null now defaults to 
    *                    {@link PoemThread#transaction()} but
    *                    we do not rely on this much yet.
    * @return a ResultSet                     
    * @throws SQLPoemException if necessary
    */
   private ResultSet selectionResultSet(String fromClause, String whereClause,
                                        String orderByClause, 
                                        boolean includeDeleted, 
                                        boolean excludeUnselectable,
                                        PoemTransaction transaction)
       throws SQLPoemException {
 
     String sql = selectionSQL(fromClause, whereClause, orderByClause,
                               includeDeleted, excludeUnselectable);
 
 
     try {
       Connection connection;
       if (transaction == null) {
         connection = getDatabase().getCommittedConnection();
       } else {
         transaction.writeDown();
         connection = transaction.getConnection();
       }
 
       Statement selectionStatement = connection.createStatement();
       ResultSet rs = selectionStatement.executeQuery(sql);
       database.incrementQueryCount(sql);
 
       SessionToken token = PoemThread._sessionToken();
       if (token != null) {
         token.toTidy().add(rs);
         token.toTidy().add(selectionStatement);
       }
       if (database.logSQL())
         database.log(new SQLLogEvent(sql));
       return rs;
     }
     catch (SQLException e) {
       throw new ExecutingSQLPoemException(sql, e);
     }
   }
 
   /**
    * It is the programmer's responsibility to ensure that the where clause 
    * is suitable for the target DBMS.
    * 
    * @return an {@link Enumeration} of Troids satisfying the criteria.
    */ 
   public Enumeration troidSelection(String whereClause, String orderByClause,
                                     boolean includeDeleted, 
                                     PoemTransaction transaction) {
     return troidsFrom(selectionResultSet(null, whereClause, orderByClause,
                                          includeDeleted, true,
                                          transaction));
   }
 
   /**
    *
    * @see #troidSelection(String, String, boolean, PoemTransaction)
    * @param criteria Represents selection criteria possibly on joined tables
    * @param transaction A transaction or null for 
    *                    {@link PoemThread#transaction()}
    * @return a selection of troids given arguments specifying a query
    */
   public Enumeration troidSelection(Persistent criteria, String orderByClause,
                                     boolean includeDeleted, 
                                     boolean excludeUnselectable,
                                     PoemTransaction transaction) {
     return troidsFrom(selectionResultSet(((JdbcPersistent)criteria).fromClause(), 
                                          whereClause(