/*
 * Copyright 2004-2008 H2 Group. Licensed under the H2 License, Version 1.0
 * (http://h2database.com/html/license.html).
 * Initial Developer: H2 Group
 */
package org.h2.table;

import java.sql.SQLException;
import java.util.Comparator;
import java.util.HashSet;
import org.h2.api.DatabaseEventListener;
import org.h2.constant.ErrorCode;
import org.h2.constant.SysProperties;
import org.h2.constraint.Constraint;
import org.h2.constraint.ConstraintReferential;
import org.h2.engine.Constants;
import org.h2.engine.DbObject;
import org.h2.engine.Session;
import org.h2.index.BtreeIndex;
import org.h2.index.Cursor;
import org.h2.index.HashIndex;
import org.h2.index.Index;
import org.h2.index.IndexType;
import org.h2.index.LinearHashIndex;
import org.h2.index.MultiVersionIndex;
import org.h2.index.ScanIndex;
import org.h2.index.TreeIndex;
import org.h2.message.Message;
import org.h2.message.Trace;
import org.h2.result.Row;
import org.h2.schema.Schema;
import org.h2.store.DataPage;
import org.h2.store.Record;
import org.h2.store.RecordReader;
import org.h2.util.MathUtils;
import org.h2.util.ObjectArray;
import org.h2.util.StringUtils;
import org.h2.value.DataType;
import org.h2.value.Value;

/**
 * Most tables are an instance of this class. For this table, the data is stored
 * in the database. The actual data is not kept here, instead it is kept in the
 * indexes. There is at least one index, the scan index.
 */
public class TableData extends Table implements RecordReader {
    private final boolean clustered;
    private ScanIndex scanIndex;
    private long rowCount;
    private Session lockExclusive;
    private HashSet lockShared = new HashSet();
    private Trace traceLock;
    private boolean globalTemporary;
    private final ObjectArray indexes = new ObjectArray();
    private long lastModificationId;
    private boolean containsLargeObject;

    public TableData(Schema schema, String tableName, int id, ObjectArray columns,
            boolean persistent, boolean clustered) throws SQLException {
        super(schema, id, tableName, persistent);
        Column[] cols = new Column[columns.size()];
        columns.toArray(cols);
        setColumns(cols);
        this.clustered = clustered;
        if (!clustered) {
            scanIndex = new ScanIndex(this, id, IndexColumn.wrap(cols), IndexType.createScan(persistent));
            indexes.add(scanIndex);
        }
        for (int i = 0; i < cols.length; i++) {
            if (DataType.isLargeObject(cols[i].getType())) {
                containsLargeObject = true;
                memoryPerRow = Row.MEMORY_CALCULATE;
            }
        }
        traceLock = database.getTrace(Trace.LOCK);
    }

    public void close(Session session) throws SQLException {
        for (int i = 0; i < indexes.size(); i++) {
            Index index = (Index) indexes.get(i);
            index.close(session);
        }
    }

    public Row getRow(Session session, int key) throws SQLException {
        return scanIndex.getRow(session, key);
    }

    public void addRow(Session session, Row row) throws SQLException {
        int i = 0;
        lastModificationId = database.getNextModificationDataId();
        if (database.isMultiVersion()) {
            row.setSessionId(session.getId());
        }
        try {
            for (; i < indexes.size(); i++) {
                Index index = (Index) indexes.get(i);
                index.add(session, row);
                checkRowCount(session, index, 1);
            }
            rowCount++;
        } catch (Throwable e) {
            try {
                while (--i >= 0) {
                    Index index = (Index) indexes.get(i);
                    index.remove(session, row);
                    checkRowCount(session, index, 0);
                }
            } catch (SQLException e2) {
                // this could happen, for example on failure in the storage
                // but if that is not the case it means there is something wrong
                // with the database
                // TODO log this problem
                throw e2;
            }
            throw Message.convert(e);
        }
    }

    private void checkRowCount(Session session, Index index, int offset) {
        if (SysProperties.CHECK && !database.isMultiVersion()) {
            long rc = index.getRowCount(session);
            if (rc != rowCount + offset) {
                throw Message.getInternalError("rowCount expected " + (rowCount + offset) + " got " + rc + " " + getName() + "." + index.getName());
            }
        }
    }

    public Index getScanIndex(Session session) {
        return (Index) indexes.get(0);
    }

    public Index getUniqueIndex() {
        for (int i = 0; i < indexes.size(); i++) {
            Index idx = (Index) indexes.get(i);
            if (idx.getIndexType().isUnique()) {
                return idx;
            }
        }
        return null;
    }

    public ObjectArray getIndexes() {
        return indexes;
    }

    public Index addIndex(Session session, String indexName, int indexId, IndexColumn[] cols, IndexType indexType,
            int headPos, String indexComment) throws SQLException {
        if (indexType.isPrimaryKey()) {
            for (int i = 0; i < cols.length; i++) {
                Column column = cols[i].column;
                if (column.getNullable()) {
                    throw Message.getSQLException(ErrorCode.COLUMN_MUST_NOT_BE_NULLABLE_1, column.getName());
                }
                column.setPrimaryKey(true);
            }
        }
        Index index;
        if (isPersistent() && indexType.isPersistent()) {
            if (indexType.isHash()) {
                index = new LinearHashIndex(session, this, indexId, indexName, cols, indexType);
            } else {
                index = new BtreeIndex(session, this, indexId, indexName, cols, indexType, headPos);
            }
        } else {
            if (indexType.isHash()) {
                index = new HashIndex(this, indexId, indexName, cols, indexType);
            } else {
                index = new TreeIndex(this, indexId, indexName, cols, indexType);
            }
        }
        if (database.isMultiVersion()) {
            index = new MultiVersionIndex(index, this);
        }
        if (index.needRebuild() && rowCount > 0) {
            try {
                Index scan = getScanIndex(session);
                long remaining = scan.getRowCount(session);
                long total = remaining;
                Cursor cursor = scan.find(session, null, null);
                long i = 0;
                int bufferSize = Constants.DEFAULT_MAX_MEMORY_ROWS;
                ObjectArray buffer = new ObjectArray(bufferSize);
                while (cursor.next()) {
                    database.setProgress(DatabaseEventListener.STATE_CREATE_INDEX, getName(), MathUtils
                            .convertLongToInt(i++), MathUtils.convertLongToInt(total));
                    Row row = cursor.get();
                    // index.add(session, row);
                    buffer.add(row);
                    if (buffer.size() >= bufferSize) {
                        addRowsToIndex(session, buffer, index);
                    }
                    remaining--;
                }
                addRowsToIndex(session, buffer, index);
                if (SysProperties.CHECK && remaining != 0) {
                    throw Message.getInternalError("rowcount remaining=" + remaining + " " + getName());
                }
            } catch (SQLException e) {
                getSchema().freeUniqueName(indexName);
                try {
                    index.remove(session);
                } catch (SQLException e2) {
                    // this could happen, for example on failure in the storage
                    // but if that is not the case it means 
                    // there is something wrong with the database
                    // TODO log this problem
                    throw e2;
                }
                throw e;
            }
        }
        boolean temporary = getTemporary();
        index.setTemporary(temporary);
        if (index.getCreateSQL() != null) {
            index.setComment(indexComment);
            database.addSchemaObject(session, index);
            // Need to update, because maybe the index is rebuilt at startup,
            // and so the head pos may have changed, which needs to be stored now.
            // addSchemaObject doesn't update the sys table at startup
            if (index.getIndexType().isPersistent() && !database.getReadOnly()
                    && !database.getLog().containsInDoubtTransactions()) {
                // can not save anything in the log file if it contains in-doubt transactions
                database.update(session, index);
            }
        }
        indexes.add(index);
        setModified();
        return index;
    }

    public boolean canGetRowCount() {
        return true;
    }

    private void addRowsToIndex(Session session, ObjectArray list, Index index) throws SQLException {
        final Index idx = index;
        try {
            list.sort(new Comparator() {
                public int compare(Object o1, Object o2) {
                    Row r1 = (Row) o1;
                    Row r2 = (Row) o2;
                    try {
                        return idx.compareRows(r1, r2);
                    } catch (SQLException e) {
                        throw Message.convertToInternal(e);
                    }
                }
            });
        } catch (Exception e) {
            throw Message.convert(e);
        }
        for (int i = 0; i < list.size(); i++) {
            Row row = (Row) list.get(i);
            index.add(session, row);
        }
        list.clear();
    }

    public boolean canDrop() {
        return true;
    }

    public long getRowCount(Session session) {
        if (database.isMultiVersion()) {
            return getScanIndex(session).getRowCount(session);
        }
        return rowCount;
    }

    public void removeRow(Session session, Row row) throws SQLException {
        lastModificationId = database.getNextModificationDataId();
        if (database.isMultiVersion()) {
            if (row.getDeleted()) {
                throw Message.getSQLException(ErrorCode.CONCURRENT_UPDATE_1, getName());
            }
            int old = row.getSessionId();
            int newId = session.getId();