/*
   Copyright (c) 2012, Oracle and/or its affiliates. All rights reserved.

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   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.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
 */

package com.mysql.clusterj.tie;

import java.math.BigDecimal;
import java.math.BigInteger;

import java.nio.ByteBuffer;
import java.nio.ByteOrder;

import java.util.ArrayList;
import java.util.List;

import com.mysql.clusterj.ClusterJFatalInternalException;
import com.mysql.clusterj.ClusterJFatalUserException;
import com.mysql.clusterj.ClusterJUserException;
import com.mysql.clusterj.ColumnType;

import com.mysql.clusterj.core.store.Column;
import com.mysql.clusterj.core.store.Index;
import com.mysql.clusterj.core.store.Table;

import com.mysql.clusterj.core.util.I18NHelper;
import com.mysql.clusterj.core.util.Logger;
import com.mysql.clusterj.core.util.LoggerFactoryService;

import com.mysql.clusterj.tie.DbImpl.BufferManager;

import com.mysql.ndbjtie.ndbapi.NdbRecord;
import com.mysql.ndbjtie.ndbapi.NdbRecordConst;
import com.mysql.ndbjtie.ndbapi.NdbDictionary.ColumnConst;
import com.mysql.ndbjtie.ndbapi.NdbDictionary.Dictionary;
import com.mysql.ndbjtie.ndbapi.NdbDictionary.IndexConst;
import com.mysql.ndbjtie.ndbapi.NdbDictionary.RecordSpecification;
import com.mysql.ndbjtie.ndbapi.NdbDictionary.RecordSpecificationArray;
import com.mysql.ndbjtie.ndbapi.NdbDictionary.TableConst;

/**
 * Wrapper around an NdbRecord. Operations may use one or two instances.
 * <ul><li>The table implementation can be used for create, read, update, or delete
 * using an NdbRecord that defines every column in the table.
 * </li><li>The index implementation for unique indexes can be used with a unique lookup operation.
 * </li><li>The index implementation for ordered (non-unique) indexes can be used with an index scan operation.
 * </li></ul>
 * After construction, the instance is
 * read-only and can be shared among all threads that use the same cluster connection; and the size of the
 * buffer required for operations is available. 
 * Methods on the instance generally require a buffer to be passed, which is modified by the method.
 * The NdbRecord instance is released when the cluster
 * connection is closed or when schema change invalidates it. Column values can be set using a provided
 * buffer and buffer manager.
 */
public class NdbRecordImpl {

    /** My message translator */
    static final I18NHelper local = I18NHelper
            .getInstance(NdbRecordImpl.class);

    /** My logger */
    static final Logger logger = LoggerFactoryService.getFactory()
            .getInstance(NdbRecordImpl.class);

    /** The size of the NdbRecord struct */
    protected final static int SIZEOF_RECORD_SPECIFICATION = ClusterConnectionServiceImpl.SIZEOF_RECORD_SPECIFICATION;

    /** The NdbRecord for this operation, created at construction */
    private NdbRecord ndbRecord = null;

    /** The store columns for this operation */
    protected Column[] storeColumns = null;

    /** The RecordSpecificationArray used to define the columns in the NdbRecord */
    private RecordSpecificationArray recordSpecificationArray;

    /** The NdbTable */
    TableConst tableConst = null;

    /** The NdbIndex, which will be null for complete-table instances */
    IndexConst indexConst = null;

    /** The size of the buffer for this NdbRecord */
    protected int bufferSize;

    /** The maximum column id for this NdbRecord */
    protected int maximumColumnId;

    /** The offsets into the buffer for each column */
    protected int[] offsets;

    /** The lengths of the column data */
    protected int[] lengths;

    /** Values for setting column mask and null bit mask: 0x01, 0x02, 0x04, 0x08, 0x10, 0x20, 0x40, 0x80 */
    protected final static byte[] BIT_IN_BYTE_MASK = new byte[] {1, 2, 4, 8, 16, 32, 64, -128};

    /** Values for resetting column mask and null bit mask: 0xfe, 0xfd, 0xfb, 0xf7, 0xef, 0xdf, 0xbf, 0x7f */
    protected static final byte[] RESET_BIT_IN_BYTE_MASK = new byte[] {-2, -3, -5, -9, -17, -33, -65, 127};

    /** The null indicator for the field bit in the byte */
    protected int nullbitBitInByte[] = null;

    /** The null indicator for the field byte offset*/
    protected int nullbitByteOffset[] = null;

    /** The size of the null indicator byte array */
    protected int nullIndicatorSize;

    /** The maximum length of any column in this operation */
    protected int maximumColumnLength;

    /** The dictionary used to create (and release) the NdbRecord */
    private Dictionary ndbDictionary;

    /** Number of columns for this NdbRecord */
    private int numberOfTableColumns;

    /** These fields are only used during construction of the RecordSpecificationArray */
    int offset = 0;
    int nullablePosition = 0;
    byte[] defaultValues;

    private int[] recordSpecificationIndexes = null;

    /** Constructor for table operations.
     * 
     * @param storeTable the store table
     * @param ndbDictionary the ndb dictionary
     */
    protected NdbRecordImpl(Table storeTable, Dictionary ndbDictionary) {
        this.ndbDictionary = ndbDictionary;
        this.tableConst = getNdbTable(storeTable.getName());
        this.numberOfTableColumns = tableConst.getNoOfColumns();
        this.recordSpecificationArray = RecordSpecificationArray.create(numberOfTableColumns);
        recordSpecificationIndexes = new int[numberOfTableColumns];
        this.offsets = new int[numberOfTableColumns];
        this.lengths = new int[numberOfTableColumns];
        this.nullbitBitInByte = new int[numberOfTableColumns];
        this.nullbitByteOffset = new int[numberOfTableColumns];
        this.storeColumns = new Column[numberOfTableColumns];
        this.ndbRecord = createNdbRecord(storeTable, ndbDictionary);
        if (logger.isDetailEnabled()) logger.detail(storeTable.getName() + " " + dumpDefinition());
        initializeDefaultBuffer();
    }

    /** Constructor for index operations. The NdbRecord has columns just for
     * the columns in the index. 
     * 
     * @param storeIndex the store index
     * @param storeTable the store table
     * @param ndbDictionary the ndb dictionary
     */
    protected NdbRecordImpl(Index storeIndex, Table storeTable, Dictionary ndbDictionary) {
        this.ndbDictionary = ndbDictionary;
        this.tableConst = getNdbTable(storeTable.getName());
        this.indexConst = getNdbIndex(storeIndex.getInternalName(), tableConst.getName());
        this.numberOfTableColumns = tableConst.getNoOfColumns();
        int numberOfIndexColumns = this.indexConst.getNoOfColumns();
        this.recordSpecificationArray = RecordSpecificationArray.create(numberOfIndexColumns);
        recordSpecificationIndexes = new int[numberOfTableColumns];
        this.offsets = new int[numberOfTableColumns];
        this.lengths = new int[numberOfTableColumns];
        this.nullbitBitInByte = new int[numberOfTableColumns];
        this.nullbitByteOffset = new int[numberOfTableColumns];
        this.storeColumns = new Column[numberOfTableColumns];
        this.ndbRecord = createNdbRecord(storeIndex, storeTable, ndbDictionary);
        if (logger.isDetailEnabled()) logger.detail(storeIndex.getInternalName() + " " + dumpDefinition());
        initializeDefaultBuffer();
    }

    /** Initialize the byte buffer containing default values for all columns.
     * Non-null columns are initialized to zero. Nullable columns are initialized to null.
     * When a new byte buffer is required, the byte buffer is used for initialization.
     */
    private void initializeDefaultBuffer() {
        // create the default value for the buffer: null values or zeros for all columns
        defaultValues = new byte[bufferSize];
        ByteBuffer zeros = ByteBuffer.allocateDirect(bufferSize);
        zeros.order(ByteOrder.nativeOrder());
        // just to be sure, initialize with zeros
        zeros.put(defaultValues);
        // not all columns are set at this point, so only check for those that are set
        for (Column storeColumn: storeColumns) {
            // nullable columns get the null bit set
            if (storeColumn != null && storeColumn.getNullable()) {
                setNull(zeros, storeColumn);
            }
        }
        zeros.position(0);
        zeros.limit(bufferSize);
        zeros.get(defaultValues);
        // default values is now immutable and can be used thread-safe
    }

    /** Initialize a new direct buffer with default values for all columns.
     * The buffer returned is positioned at 0 with the limit set to the buffer size.
     * @return a new byte buffer for use with this NdbRecord.
     */
    protected ByteBuffer newBuffer() {
        ByteBuffer result = ByteBuffer.allocateDirect(bufferSize);
        initializeBuffer(result);
        return result;
    }

    /** Initialize an already-allocated buffer with default values for all columns.
     * 
     * @param buffer
     */
    protected void initializeBuffer(ByteBuffer buffer) {
        buffer.order(ByteOrder.nativeOrder());
        buffer.limit(bufferSize);
        buffer.position(0);
        buffer.put(defaultValues);
        buffer.position(0);
    }

    public int setNull(ByteBuffer buffer, Column storeColumn) {
        int columnId = storeColumn.getColumnId();
        if (!storeColumn.getNullable()) {
            return columnId;
        }
        int index = nullbitByteOffset[columnId];
        byte mask = BIT_IN_BYTE_MASK[nullbitBitInByte[columnId]];
        byte nullbyte = buffer.get(index);
        buffer.put(index, (byte)(nullbyte|mask));
        return columnId;
    }

    public int resetNull(ByteBuffer buffer, Column storeColumn) {
        int columnId = storeColumn.getColumnId();
        if (!storeColumn.getNullable()) {
            return columnId;
        }
        int index = nullbitByteOffset[columnId];
        byte mask = RESET_BIT_IN_BYTE_MASK[nullbitBitInByte[columnId]];
        byte nullbyte = buffer.get(index);
        buffer.put(index, (byte)(nullbyte & mask));
        return columnId;
    }

    public int setBigInteger(ByteBuffer buffer, Column storeColumn, BigInteger value) {
        resetNull(buffer, storeColumn);
        int columnId = storeColumn.getColumnId();
        int newPosition = offsets[columnId];
        buffer.position(newPosition);
        // TODO provide the buffer to Utility.convertValue to avoid copying
        ByteBuffer bigIntegerBuffer = Utility.convertValue(storeColumn, value);
        buffer.put(bigIntegerBuffer);
        buffer.limit(bufferSize);
        buffer.position(0);
        return columnId;
    }

    public int setByte(ByteBuffer buffer, Column storeColumn, byte value) {
        resetNull(buffer, storeColumn);
        int columnId = storeColumn.getColumnId();
        if (storeColumn.getType() == ColumnType.Bit) {
            // the byte is stored as a BIT array of four bytes
            buffer.putInt(offsets[columnId], value & 0xff);
        } else {
            buffer.put(offsets[columnId], value);
        }
        buffer.limit(bufferSize);
        buffer.position(0);
        return columnId;
    }

    public int setBytes(ByteBuffer buffer, Column storeColumn, byte[] value) {
        resetNull(buffer, storeColumn);
        int columnId = storeColumn.getColumnId();
        int offset = offsets[columnId];
        int length = storeColumn.getLength();
        int prefixLength = storeColumn.getPrefixLength();
        if (length < value.length) {
            throw new ClusterJUserException(local.message("ERR_Data_Too_Long",
                    storeColumn.getName(), length, value.length));
        }
        buffer.limit(offset + prefixLength + length);
        buffer.position(offset);
        Utility.convertValue(buffer, storeColumn, value);
        buffer.limit(bufferSize);
        buffer.position(0);
        return columnId;
    }

    public int setDecimal(ByteBuffer buffer, Column storeColumn, BigDecimal value) {
        resetNull(buffer, storeColumn);
        int columnId = storeColumn.getColumnId();
        int newPosition = offsets[columnId];
        buffer.position(newPosition);
        // TODO provide the buffer to Utility.convertValue to avoid copying
        ByteBuffer decimalBuffer = Utility.convertValue(storeColumn, value);
        buffer.put(decimalBuffer);
        return columnId;
    }

    public int setDouble(ByteBuffer buffer, Column storeColumn, double value) {
        resetNull(buffer, storeColumn);
        int columnId = storeColumn.getColumnId();
        buffer.putDouble(offsets[columnId], value);
        return columnId;
    }

    public int setFloat(ByteBuffer buffer, Column storeColumn, float value) {
        resetNull(buffer, storeColumn);
        int columnId = storeColumn.getColumnId();
        buffer.putFloat(offsets[columnId], value);
        return columnId;
    }

    public int setInt(ByteBuffer buffer, Column storeColumn, int value) {
        resetNull(buffer, storeColumn);
        int columnId = storeColumn.getColumnId();
        int storageValue = Utility.convertIntValueForStorage(storeColumn, value);
        buffer.putInt(offsets[columnId], storageValue);
        return columnId;
    }

    public int setLong(ByteBuffer buffer, Column storeColumn, long value) {
        resetNull(buffer, storeColumn);
        int columnId = storeColumn.getColumnId();
        long storeValue = Utility.convertLongValueForStorage(storeColumn, value);
        buffer.putLong(offsets[columnId], storeValue);
        return columnId;
    }

    public int setShort(ByteBuffer buffer, Column storeColumn, short value) {
        resetNull(buffer, storeColumn);
        int columnId = storeColumn.getColumnId();
        if (storeColumn.getLength() == 4) {
            // the short is stored as a BIT array of four bytes
            buffer.putInt(offsets[columnId], value & 0xffff);
        } else {
            buffer.putShort(offsets[columnId], (short)value);
        }
        return columnId;
    }

    public int setString(ByteBuffer buffer, BufferManager bufferManager, Column storeColumn, String value) {
        resetNull(buffer, storeColumn);
        int columnId = storeColumn.getColumnId();
        int offset = offsets[columnId];
        int prefixLength = storeColumn.getPrefixLength();
        int length = storeColumn.getLength() + prefixLength;
        buffer.limit(offset + length);
        buffer.position(offset);
        // TODO provide the buffer to Utility.encode to avoid copying
        // for now, use the encode method to encode the value then copy it
        ByteBuffer converted = Utility.encode(value, storeColumn, bufferManager);
        if (length < converted.remaining()) {
            throw new ClusterJUserException(local.message("ERR_Data_Too_Long",
                    storeColumn.getName(), length - prefixLength, converted.remaining() - prefixLength));
        }
        buffer.put(converted);
        buffer.limit(bufferSize);
        buffer.position(0);
        return columnId;
    }

    public boolean getBoolean(ByteBuffer buffer, int columnId) {
        int value = buffer.getInt(offsets[columnId]);
        return Utility.getBoolean(storeColumns[columnId], value);
    }

    public byte getByte(ByteBuffer buffer, int columnId) {
        Column storeColumn = storeColumns[columnId];
        if (storeColumn.getType() == ColumnType.Bit) {
            // the byte was stored in a BIT column as four bytes
            return (byte)(buffer.getInt(offsets[columnId]));
        } else {
            // the byte was stored as a byte
            return buffer.get(offsets[columnId]);
        }
    }

    public byte[] getBytes(ByteBuffer byteBuffer, int columnId) {
        return getBytes(byteBuffer, storeColumns[columnId]);
    }

    public byte[] getBytes(ByteBuffer byteBuffer, Column storeColumn) {
        int columnId = storeColumn.getColumnId();
        if (isNull(byteBuffer, columnId)) {
            return null;
        }
        int prefixLength = storeColumn.getPrefixLength();
        int actualLength = lengths[columnId];
        int offset = offsets[columnId];
        switch (prefixLength) {
            case 0:
                break;
            case 1:
                actualLength = (byteBuffer.get(offset) + 256) % 256;
                offset += 1;
                break;
            case 2:
                actualLength = (byteBuffer.get(offset) + 256) % 256;
                int length2 = (byteBuffer.get(offset + 1) + 256) % 256;
                actualLength += 256 * length2;
                offset += 2;
                break;
            default:
                throw new ClusterJFatalInternalException(
                        local.message("ERR_Invalid_Prefix_Length", prefixLength));
        }
        byteBuffer.position(offset);
        byte[] result = new byte[actualLength];
        byteBuffer.get(result);
        byteBuffer.limit(bufferSize);
        byteBuffer.position(0);
        return result;
     }

    public double getDouble(ByteBuffer buffer, int columnId) {
        double result = buffer.getDouble(offsets[columnId]);
        return result;
    }

    public float getFloat(ByteBuffer buffer, int columnId) {
        float result = buffer.getFloat(offsets[columnId]);
        return result;
    }

    public int getInt(ByteBuffer buffer, int columnId) {
        int value = buffer.getInt(offsets[columnId]);
        return Utility.getInt(storeColumns[columnId], value);
    }

    public long getLong(ByteBuffer buffer, int columnId) {
        long value = buffer.getLong(offsets[columnId]);
        return Utility.getLong(storeColumns[columnId], value);
    }

    public short getShort(ByteBuffer buffer, int columnId) {
        Column storeColumn = storeColumns[columnId];
        if (storeColumn.getLength() == 4) {
            // the short was stored in a BIT column as four bytes
            return (short)buffer.getInt(offsets[columnId]);
        } else {
            // the short was stored as a short
            return buffer.getShort(offsets[columnId]);
        }
    }

    public String getString(ByteBuffer byteBuffer, int columnId, BufferManager bufferManager) {
      if (isNull(byteBuffer, columnId)) {
          return null;
      }
      Column storeColumn = storeColumns[columnId];
      int prefixLength = storeColumn.getPrefixLength();
      int actualLength;
      int offset = offsets[columnId];
      byteBuffer.limit(byteBuffer.capacity());
      switch (prefixLength) {
          case 0:
              actualLength = lengths[columnId];
              break;
          case 1:
              actualLength = (byteBuffer.get(offset) + 256) % 256;
              offset += 1;
              break;
          case 2:
              actualLength = (byteBuffer.get(offset) + 256) % 256;
              int length2 = (byteBuffer.get(offset + 1) + 256) % 256;
              actualLength += 256 * length2;
              offset += 2;
              break;
          default:
              throw new ClusterJFatalInternalException(
                      local.message("ERR_Invalid_Prefix_Length", prefixLength));
      }
      byteBuffer.position(offset);
      byteBuffer.limit(offset + actualLength);

      String result = Utility.decode(byteBuffer, storeColumn.getCharsetNumber(), bufferManager);
      byteBuffer.limit(bufferSize);
      byteBuffer.position(0);
      return result;
    }

    public BigInteger getBigInteger(ByteBuffer byteBuffer, int columnId) {
        Column storeColumn = storeColumns[columnId];
        int index = storeColumn.getColumnId();
        int offset = offsets[index];
        int precision = storeColumn.getPrecision();
        int scale = storeColumn.getScale();
        int length = Utility.getDecimalColumnSpace(precision, scale);
        byteBuffer.position(offset);
        BigInteger result = Utility.getBigInteger(byteBuffer, length, precision, scale);
        byteBuffer.limit(bufferSize);
        byteBuffer.position(0);
        return result;
    }

    public BigInteger getBigInteger(ByteBuffer byteBuffer, Column storeColumn) {
        int index = storeColumn.getColumnId();
        int offset = offsets[index];
        int precision = storeColumn.getPrecision();
        int scale = storeColumn.getScale();
        int length = Utility.getDecimalColumnSpace(precision, scale);
        byteBuffer.position(offset);
        BigInteger result = Utility.getBigInteger(byteBuffer, length, precision, scale);
        byteBuffer.limit(bufferSize);
        byteBuffer.position(0);
        return result;
    }

    public BigDecimal getDecimal(ByteBuffer byteBuffer, int columnId) {
        Column storeColumn = storeColumns[columnId];
        int index = storeColumn.getColumnId();
        int offset = offsets[index];
        int precision = storeColumn.getPrecision();
        int scale = storeColumn.getScale();
        int length = Utility.getDecimalColumnSpace(precision, scale);
        byteBuffer.position(offset);
        BigDecimal result = Utility.getDecimal(byteBuffer, length, precision, scale);
        byteBuffer.limit(bufferSize);
        byteBuffer.position(0);
        return result;
      }

    public BigDecimal getDecimal(ByteBuffer byteBuffer, Column storeColumn) {
      int index = storeColumn.getColumnId();
      int offset = offsets[index];
      int precision = storeColumn.getPrecision();
      int scale = storeColumn.getScale();
      int length = Utility.getDecimalColumnSpace(precision, scale);
      byteBuffer.position(offset);
      BigDecimal result = Utility.getDecimal(byteBuffer, length, precision, scale);
      byteBuffer.limit(bufferSize);
      byteBuffer.position(0);
      return result;
    }

    public Boolean getObjectBoolean(ByteBuffer byteBuffer, int columnId) {
        if (isNull(byteBuffer, columnId)) {
            return null;
        }
        return Boolean.valueOf(getBoolean(byteBuffer, columnId));        
    }

    public Boolean getObjectBoolean(ByteBuffer byteBuffer, Column storeColumn) {
        return getObjectBoolean(byteBuffer, storeColumn.getColumnId());
    }

    public Byte getObjectByte(ByteBuffer byteBuffer, int columnId) {
        if (isNull(byteBuffer, columnId)) {
            return null;
        }
        return getByte(byteBuffer, columnId);        
    }

    public Byte getObjectByte(ByteBuffer byteBuffer, Column storeColumn) {
        return getObjectByte(byteBuffer, storeColumn.getColumnId());
    }

    public Float getObjectFloat(ByteBuffer byteBuffer, int columnId) {
        if (isNull(byteBuffer, columnId)) {
            return null;
        }
        return getFloat(byteBuffer, columnId);        
    }

    public Float getObjectFloat(ByteBuffer byteBuffer, Column storeColumn) {
        return getObjectFloat(byteBuffer, storeColumn.getColumnId());
    }

    public Double getObjectDouble(ByteBuffer byteBuffer, int columnId) {
        if (isNull(byteBuffer, columnId)) {
            return null;
        }
        return getDouble(byteBuffer, columnId);        
    }

    public Double getObjectDouble(ByteBuffer byteBuffer, Column storeColumn) {
        return getObjectDouble(byteBuffer, storeColumn.getColumnId());
    }

    public Integer getObjectInteger(ByteBuffer byteBuffer, int columnId) {
        if (isNull(byteBuffer, columnId)) {
            return null;
        }
        return getInt(byteBuffer, columnId);        
    }

    public Integer getObjectInteger(ByteBuffer byteBuffer, Column storeColumn) {
        return getObjectInteger(byteBuffer, storeColumn.getColumnId());
    }

    public Long getObjectLong(ByteBuffer byteBuffer, int columnId) {
        if (isNull(byteBuffer, columnId)) {
            return null;
        }
        return getLong(byteBuffer, columnId);        
    }

    public Long getObjectLong(ByteBuffer byteBuffer, Column storeColumn) {
        return getObjectLong(byteBuffer, storeColumn.getColumnId());
    }

    public Short getObjectShort(ByteBuffer byteBuffer, int columnId) {
        if (isNull(byteBuffer, columnId)) {
            return null;
        }
        return getShort(byteBuffer, columnId);        
    }

    public Short getObjectShort(ByteBuffer byteBuffer, Column storeColumn) {
        return getObjectShort(byteBuffer, storeColumn.getColumnId());
    }

    public boolean isNull(ByteBuffer buffer, int columnId) {
        if (!storeColumns[columnId].getNullable()) {
            return false;
        }
        byte nullbyte = buffer.get(nullbitByteOffset[columnId]);
        boolean result = isSet(nullbyte, nullbitBitInByte[columnId]);
        return result;
    }

    public boolean isPresent(byte[] mask, int columnId) {
        byte present = mask[columnId/8];
        return isSet(present, columnId % 8);
    }

    public void markPresent(byte[] mask, int columnId) {
        int offset = columnId/8;
        int bitMask = BIT_IN_BYTE_MASK[columnId % 8];
        mask[offset] |= (byte)bitMask;
    }

    protected boolean isSet(byte test, int bitInByte) {
        int mask = BIT_IN_BYTE_MASK[bitInByte];
        boolean result = (test & mask) != 0;
        return result;
    }

    protected static void handleError(Object object, Dictionary ndbDictionary) {
        if (object != null) {
            return;
        } else {
            Utility.throwError(null, ndbDictionary.getNdbError());
        }
    }

    protected NdbRecord createNdbRecord(Index storeIndex, Table storeTable, Dictionary ndbDictionary) {
        String[] columnNames = storeIndex.getColumnNames();
        // analyze columns; sort into alignment buckets, allocate space in the buffer
        // and build the record specification array
        analyzeColumns(storeTable, columnNames);
        // create the NdbRecord
        NdbRecord result = ndbDictionary.createRecord(indexConst, tableConst, recordSpecificationArray,
                columnNames.length, SIZEOF_RECORD_SPECIFICATION, 0);
        // delete the RecordSpecificationArray since it is no longer needed
        RecordSpecificationArray.delete(recordSpecificationArray);
        handleError(result, ndbDictionary);
        return result;
    }

    protected NdbRecord createNdbRecord(Table storeTable, Dictionary ndbDictionary) {
        String[] columnNames = storeTable.getColumnNames();
        // analyze columns; sort into alignment buckets, allocate space in the buffer,
        // and build the record specification array
        analyzeColumns(storeTable, columnNames);
        // create the NdbRecord
        NdbRecord result = ndbDictionary.createRecord(tableConst, recordSpecificationArray,
                columnNames.length, SIZEOF_RECORD_SPECIFICATION, 0);
        // delete the RecordSpecificationArray since it is no longer needed
        RecordSpecificationArray.delete(recordSpecificationArray);
        handleError(result, ndbDictionary);
        return result;
    }

    private void analyzeColumns(Table storeTable, String[] columnNames) {
        List<Column> align8 = new ArrayList<Column>();
        List<Column> align4 = new ArrayList<Column>();
        List<Column> align2 = new ArrayList<Column>();
        List<Column> align1 = new ArrayList<Column>();
        List<Column> nullables = new ArrayList<Column>();
        int i = 0;
        for (String columnName: columnNames) {
            Column storeColumn = storeTable.getColumn(columnName);
            int columnId = storeColumn.getColumnId();
            recordSpecificationIndexes[columnId] = i;
            if (logger.isDetailEnabled()) logger.detail("storeColumn: " + storeColumn.getName() + " id: " + storeColumn.getColumnId() + " index: " + i);
            lengths[i] = storeColumn.getLength();
            storeColumns[i++] = storeColumn;
            // for each column, put into alignment bucket
            switch (storeColumn.getType()) {
                case Bigint:
                case Bigunsigned:
                case Bit:
                case Blob:
                case Date:
                case Datetime:
                case Double:
                case Text:
                case Time:
                case Timestamp:
                    align8.add(storeColumn);
                    break;
                case Binary:
                case Char:
                case Decimal:
                case Decimalunsigned:
                case Longvarbinary:
                case Longvarchar:
                case Olddecimal:
                case Olddecimalunsigned:
                case Tinyint:
                case Tinyunsigned:
                case Varbinary:
                case Varchar:
                    align1.add(storeColumn);
                    break;
                case Float:
                case Int:
                case Mediumint:
                case Mediumunsigned:
                case Unsigned:
                    align4.add(storeColumn);
                    break;
                case Smallint:
                case Smallunsigned:
                case Year:
                    align2.add(storeColumn);
                    break;
                case Undefined:
                    throw new ClusterJFatalUserException(local.message("ERR_Unknown_Column_Type",
                            storeTable.getName(), columnName, storeColumn.getType()));
                default:
                    throw new ClusterJFatalInternalException(local.message("ERR_Unknown_Column_Type",
                            storeTable.getName(), columnName, storeColumn.getType()));
            }
            if (storeColumn.getNullable()) {
                nullables.add(storeColumn);
            }
        }
        // for each column, allocate space in the buffer, starting with align8 and ending with align1
        // null indicators are allocated first, with one bit per nullable column
        // nullable columns take one bit each
        offset = nullables.size() + 7 / 8;
        // align the first column following the nullable column indicators to 8
        offset = (7 + offset) / 8 * 8;
        nullIndicatorSize = offset;
        for (Column storeColumn: align8) {
            analyzeColumn(8, storeColumn);
        }
        for (Column storeColumn: align4) {
            analyzeColumn(4, storeColumn);
        }
        for (Column storeColumn: align2) {
            analyzeColumn(2, storeColumn);
        }
        for (Column storeColumn: align1) {
            analyzeColumn(1, storeColumn);
        }
        bufferSize = offset;

        if (logger.isDebugEnabled()) logger.debug(dumpDefinition());
    }

    /** Create a record specification for a column. Keep track of the offset into the buffer
     * and the null indicator position for each column.
     * 
     * @param alignment the alignment for this column in the buffer
     * @param storeColumn the column
     */
    private void analyzeColumn(int alignment, Column storeColumn) {
        int columnId = storeColumn.getColumnId();
        int recordSpecificationIndex = recordSpecificationIndexes[columnId];
        RecordSpecification recordSpecification = recordSpecificationArray.at(recordSpecificationIndex);
        ColumnConst columnConst = tableConst.getColumn(columnId);
        recordSpecification.column(columnConst);
        recordSpecification.offset(offset);
        offsets[columnId] = offset;
        int columnSpace = storeColumn.getColumnSpace();
        offset += ((columnSpace==0)?alignment:columnSpace);
        if (storeColumn.getNullable()) {
            int nullbitByteOffsetValue = nullablePosition/8;
            int nullbitBitInByteValue = nullablePosition - nullablePosition / 8 * 8;
            nullbitBitInByte[columnId] = nullbitBitInByteValue;
            nullbitByteOffset[columnId] = nullbitByteOffsetValue;
            recordSpecification.nullbit_byte_offset(nullbitByteOffsetValue);
            recordSpecification.nullbit_bit_in_byte(nullbitBitInByteValue);
            ++nullablePosition;
        } else {
            recordSpecification.nullbit_byte_offset(0);
            recordSpecification.nullbit_bit_in_byte(0);
        }
    }

    private String dumpDefinition() {
        StringBuilder builder = new StringBuilder(tableConst.getName());
        builder.append(" numberOfColumns: ");
        builder.append(numberOfTableColumns);
        builder.append('\n');
        for (int columnId = 0; columnId < numberOfTableColumns; ++columnId) {
            Column storeColumn = storeColumns[columnId];
            if (storeColumn != null) {
                builder.append(" column: ");
                builder.append(storeColumn.getName());
                builder.append(" offset: ");
                builder.append(offsets[columnId]);
                builder.append(" length: ");
                builder.append(lengths[columnId]);
                builder.append(" nullbitBitInByte: ");
                builder.append(nullbitBitInByte[columnId]);
                builder.append(" nullbitByteOffset: ");
                builder.append(nullbitByteOffset[columnId]);
                builder.append('\n');
            }
        }
        return builder.toString();
    }

    public String dumpValues(ByteBuffer data, byte[] mask) {
        StringBuilder builder = new StringBuilder(tableConst.getName());
        builder.append(" numberOfColumns: ");
        builder.append(numberOfTableColumns);
        builder.append('\n');
        for (int columnId = 0; columnId < numberOfTableColumns; ++columnId) {
            Column storeColumn = storeColumns[columnId];
            if (storeColumn != null) {
                builder.append(" column: ");
                builder.append(storeColumn.getName());
                builder.append(" offset: ");
                builder.append(offsets[columnId]);
                builder.append(" length: ");
                builder.append(lengths[columnId]);
                builder.append(" nullbitBitInByte: ");
                int nullBitInByte = nullbitBitInByte[columnId];
                builder.append(nullBitInByte);
                builder.append(" nullbitByteOffset: ");
                int nullByteOffset = nullbitByteOffset[columnId];
                builder.append(nullByteOffset);
                builder.append(" data: ");
                int size = storeColumn.getColumnSpace() != 0 ? storeColumn.getColumnSpace():storeColumn.getSize();
                int offset = offsets[columnId];
                data.limit(bufferSize);
                data.position(0);
                for (int index = offset; index < offset + size; ++index) {
                    builder.append(String.format("%2x ", data.get(index)));
                }
                builder.append(" null: ");
                builder.append(isNull(data, columnId));
                if (mask != null) {
                    builder.append(" present: ");
                    builder.append(isPresent(mask, columnId));
                }
                builder.append('\n');
            }
        }
        data.position(0);
        return builder.toString();
    }

    TableConst getNdbTable(String tableName) {
        TableConst ndbTable = ndbDictionary.getTable(tableName);
        if (ndbTable == null) {
            // try the lower case table name
            ndbTable = ndbDictionary.getTable(tableName.toLowerCase());
        }
        if (ndbTable == null) {
            Utility.throwError(ndbTable, ndbDictionary.getNdbError(), tableName);
        }
        return ndbTable;
    }

    TableConst getNdbTable() {
        return tableConst;
    }

    IndexConst getNdbIndex(String indexName, String tableName) {
        IndexConst ndbIndex = ndbDictionary.getIndex(indexName, tableName);
        if (ndbIndex == null) {
            Utility.throwError(ndbIndex, ndbDictionary.getNdbError(),  tableName+ "+" + indexName);
        }
        return ndbIndex;
    }

    IndexConst getNdbIndex() {
        return indexConst;
    }

    public int getBufferSize() {
        return bufferSize;
    }

    public NdbRecordConst getNdbRecord() {
        return ndbRecord;
    }

    public int getNumberOfColumns() {
        return numberOfTableColumns;
    }

    protected void releaseNdbRecord() {
        if (ndbRecord != null) {
            if (logger.isDebugEnabled())logger.debug("Releasing NdbRecord for " + tableConst.getName());
            ndbDictionary.releaseRecord(ndbRecord);
            ndbRecord = null;
        }
    }

    public int getNullIndicatorSize() {
        return nullIndicatorSize;
    }

    public boolean isLob(int columnId) {
        return storeColumns[columnId].isLob();
    }

}
