package org.maachang.mimdb.core ;

import java.lang.ref.SoftReference;
import java.util.concurrent.atomic.AtomicReference;

import org.maachang.mimdb.MimdbException;
import org.maachang.mimdb.core.util.jsnappy.JSnappy;
import org.maachang.mimdb.core.util.jsnappy.JSnappyBuffer;


/**
 * 圧縮(JSnappy)テーブル行データ.
 * 
 * @version 2013/10/14
 * @author masahito suzuki
 * @since MasterInMemDB 1.00
 */
@SuppressWarnings("unchecked")
public class MimdbSnappyRow extends MimdbRow implements Comparable<MimdbSnappyRow> {
    
    /** NULLを示す情報. **/
    private static final byte NULL_CODE = (byte)0x80 ;
    
    /** 圧縮率. **/
    private static final int JSNAPPY_SHIFT = 3 ;
    
    /** 解凍キャッシュ情報. **/
    protected AtomicReference<SoftReference<Object[]>> decompressCache = null ;
    
    /** 圧縮バイナリ. **/
    protected byte[] compressInfo = null ;
    
    /** プライマリキー. **/
    protected Object primaryKey = null ;
    
    /** 圧縮フラグ. **/
    protected boolean compressFlag = true ;
    
    /** 親子オブジェクトで利用. **/
    
    /** 親オブジェクト(子オブジェクトのみ). **/
    protected MimdbSnappyRow parentRow = null ;
    
    /** 親管理数(親オブジェクトのみ1以上、子は0、それ以外は1). **/
    protected int parentLength = 1 ;
    
    /** 子No(子要素のみ0以上). **/
    protected int childNo = 0 ;
    
    
    /**
     * コンストラクタ.
     */
    private MimdbSnappyRow(){}
    
    /**
     * コンストラクタ.
     * ※1行オブジェクトとして生成.
     * @param table 対象テーブルを取得します.
     * @param no 行番号を設定します.
     * @param val １行データを設定します.
     * @param compressFlag [true]の場合は圧縮処理します.
     * @param binLen バイナリ長を設定します.
     * @exception Exception 例外.
     */
    protected MimdbSnappyRow( MimdbTable table,int no,Object[] val,boolean compressFlag,int binLen )
        throws Exception {
        init( table,no,val,compressFlag,binLen ) ;
    }
    
    /**
     * コンストラクタ.
     * ※親オブジェクトとして作成.
     * @param table 対象テーブルを取得します.
     * @param no 行番号を設定します.
     * @param list 対象の１行データ群を設定します.
     * @param binLen 親オブジェクトが保持する全体のデータ長を設定します.
     * @exception Exception 例外.
     */
    protected MimdbSnappyRow( MimdbTable table,int no,Object[] list,int binLen )
        throws Exception {
        init( table,no,list,binLen ) ;
    }
    
    /**
     * コンストラクタ.
     * ※子オブジェクトとして作成.
     * @param parent 親オブジェクトを設定します.
     * @param no 行番号を設定します.
     * @param val 対象の１行データ群を設定します.
     * @exception Exception 例外.
     */
    protected MimdbSnappyRow( MimdbSnappyRow parent,MimdbTable table,int no,Object[] val )
        throws Exception {
        init( parent,table,no,val ) ;
    }
    
    /** init. **/
    /** 1行データとして作成. **/
    protected final void init( MimdbTable table,int no,Object[] val,boolean compressFlag,int binLen )
        throws Exception {
        this.dbId = table.dbId ;
        this.baseTable = table ;
        this.lineNo = no ;
        this.primaryKey = ( table.primaryIndexKeyNo == -1 ) ?
            null : val[ table.primaryIndexKeyNo ] ;
        this.compressInfo = compressByOneLine( table,binLen,compressFlag,val ) ;
        this.compressFlag = compressFlag ;
        this.decompressCache = new AtomicReference<SoftReference<Object[]>>(
            new SoftReference<Object[]>( null ) ) ;
        
        this.parentRow = null ;     // 親オブジェクトなし.
        this.parentLength = 1 ;     // 親子なし(1).
        this.childNo = 0 ;          // 子要素ではない.
    }
    
    /** init. **/
    /** 親オブジェクトとして作成. **/
    protected final void init( MimdbTable table,int no,Object[] list,int binLen )
        throws Exception {
        if( list.length == 1 ) {
            init( table,no,(Object[])(list[ 0 ]),true,binLen ) ;
        }
        else {
            this.dbId = table.dbId ;
            this.baseTable = table ;
            this.lineNo = no ;
            
            this.primaryKey = ( table.primaryIndexKeyNo == -1 ) ?
                null : ((Object[])list[ 0 ])[ table.primaryIndexKeyNo ] ;
            
            int len = list.length ;
            byte[] bin = new byte[ binLen ] ;
            int off = 0 ;
            for( int i = 0 ; i < len ; i ++ ) {
                off = lineByBinary( table,(Object[])list[ i ],bin,off ) ;
            }
            
            JSnappyBuffer buf = JSnappy.compress( bin,0,bin.length,null,JSNAPPY_SHIFT ) ;
            this.compressInfo = buf.toByteArray() ;
            this.compressFlag = true ;
            this.decompressCache = new AtomicReference<SoftReference<Object[]>>(
                new SoftReference<Object[]>( null ) ) ;
            
            this.parentRow = null ;     // 親オブジェクトなし.
            this.parentLength = len ;   // 親オブジェクトが格納する情報長(1以上).
            this.childNo = 0 ;          // 子要素ではない.
        }
    }
    
    /** init. **/
    /** 子オブジェクトとして作成. **/
    protected final void init( MimdbSnappyRow parent,MimdbTable table,int no,Object[] val )
        throws Exception {
        this.dbId = table.dbId ;
        this.baseTable = table ;
        this.lineNo = no ;
        this.primaryKey = ( table.primaryIndexKeyNo == -1 ) ?
            null : val[ table.primaryIndexKeyNo ] ;
        this.compressFlag = true ;
        
        this.parentRow = parent ;           // 親オブジェクトをセット.
        this.parentLength = 0 ;             // 子オブジェクトでゼロをセット.
        this.childNo = no - parent.lineNo ; // 子要素の番号をセット.
        
        // 親情報で持つので必要なし.
        this.decompressCache = null ;
        this.compressInfo = null ;
    }
    
    
    /**
     * カラム項番で情報を取得.
     * @param n 対象の項番を設定します.
     * @return Object 情報が返却されます.
     */
    public Object getValue( int n ) {
        return _get()[ n ] ;
    }
    
    /**
     * カラム名で情報を取得.
     * @param name 対象の名前を設定します.
     * @return Object 情報が返却されます.
     */
    public Object getValue( String name ) {
        Integer n = baseTable.columnsMap.get(
            ((String)name).trim().toLowerCase() ) ;
        if( n == null ) {
            return null ;
        }
        return _get()[ n ] ;
    }
    
    /**
     * カラムオブジェクト一覧を取得.
     * @return Object[] カラムオブジェクト一覧が返却されます.
     */
    public Object[] getValues() {
        return _get() ;
    }
    
    /**
     * プライマリーキーオブジェクトを取得.
     * @return Object プライマリーキーオブジェクトが返却されます.
     */
    public Object getPrimaryKey() {
        return primaryKey ;
    }
    
    /**
     * 圧縮長を取得.
     * @return int 圧縮されたバイナリ長が返却されます.
     */
    public int compressLength() {
        return compressInfo.length ;
    }
    
    /**
     * 比較処理.
     * @param n 比較条件を設定します.
     * @return int 比較結果が返却されます.
     *            マイナスの場合、第一引数の方が大きい.
     *            プラスの場合、第一引数の方が小さい.
     *            0の場合、第一引数と同じ.
     */
    public int compareTo( MimdbSnappyRow n ) {
        if( primaryKey == null ) {
            return 0 ;
        }
        // 主キーでは、NULL条件は有り得ないので、この条件でOK.
        return ((Comparable)primaryKey).compareTo( n.primaryKey ) ;
    }
    
    
    /** １行情報取得. **/
    protected final Object[] _get() {
        // １行情報での管理の場合.
        if( parentLength == 1 ) {
            Object[] ret ;
            // キャッシュ内に情報が存在しない場合.
            if( ( ret = decompressCache.get().get() ) == null ) {
                // キャッシュ情報の作成.
                SoftReference<Object[]> cache = new SoftReference<Object[]>(
                    decompressByOneLine( baseTable,compressFlag,compressInfo ) ) ;
                while( !decompressCache.compareAndSet( decompressCache.get(),cache ) ) ;
                ret = cache.get() ;
            }
            return ret ;
        }
        // 複数管理の親オブジェクトの場合.
        else if( parentLength != 0 ) {
            return (Object[])_getParent()[ 0 ] ;
        }
        // 子オブジェクトの場合.
        else {
            return (Object[])parentRow._getParent()[ childNo ] ;
        }
    }
    
    /** 親オブジェクトの情報取得. **/
    protected final Object[] _getParent() {
        if( parentLength != 0 ) {
            Object[] ret ;
            // キャッシュ内に情報が存在しない場合.
            if( ( ret = decompressCache.get().get() ) == null ) {
                // キャッシュ情報の作成.
                SoftReference<Object[]> cache = new SoftReference<Object[]>(
                    decompressParent( baseTable,compressInfo,parentLength ) ) ;
                while( !decompressCache.compareAndSet( decompressCache.get(),cache ) ) ;
                ret = cache.get() ;
            }
            return ret ;
        }
        return null ;
    }
    
    /** １行データ長を取得. **/
    protected static final int byteLength( int[] columnTypes,Object[] values ) {
        int ret = 0 ;
        int len = columnTypes.length ;
        for( int i = 0 ; i < len ; i ++ ) {
            switch( columnTypes[ i ] ) {
                case MimdbIndex.COLUMN_BOOL :
                    ret += 1 ;
                    break ;
                case MimdbIndex.COLUMN_INT :
                    ret += 5 ;
                    break ;
                case MimdbIndex.COLUMN_LONG :
                case MimdbIndex.COLUMN_FLOAT :
                case MimdbIndex.COLUMN_DATE :
                case MimdbIndex.COLUMN_TIME :
                case MimdbIndex.COLUMN_TIMESTAMP :
                    ret += 9 ;
                    break ;
                case MimdbIndex.COLUMN_STRING :
                    if( values[ i ] == null ) {
                        ret += 1 ;
                    }
                    else {
                        ret += (( (String)values[ i ] ).length() * 2 ) + 4 ;
                    }
                    break ;
            }
        }
        return ret ;
    }
    
    /** １行データ圧縮. **/
    protected static final byte[] compressByOneLine( MimdbTable table,int binLen,boolean compressFlag,Object[] values )
        throws Exception {
        byte[] src = new byte[ binLen ] ;
        // １行オブジェクトをバイナリに変換.
        lineByBinary( table,values,src,0 ) ;
        // 圧縮処理を行う場合.
        if( compressFlag ) {
            // バイナリ化した情報を圧縮.
            JSnappyBuffer buf = JSnappy.compress( src,0,src.length,null,JSNAPPY_SHIFT ) ;
            return buf.toByteArray() ;
        }
        // 圧縮処理を行わない場合.
        else {
            return src ;
        }
    }
    
    /** １行データ解凍. **/
    protected static final Object[] decompressByOneLine( MimdbTable table,boolean compressFlag,byte[] binary ) {
        
        // 圧縮処理を行う場合.
        if( compressFlag ) {
            // 解凍処理.
            try {
                binary = JSnappy.decompress( binary ).getData() ;
            } catch( Exception e ) {
                // エラーの場合はランタイムエラーを返却.
                throw new MimdbException( e ) ;
            }
        }
        int lineLen = table.columnTypes.length ;
        Object[] ret = new Object[ lineLen ] ;
        // バイナリデータを１行オブジェクトに変換.
        binaryByLine( ret,table,binary,0,lineLen ) ;
        return ret ;
    }
    
    /** 親オブジェクトでの解凍処理. **/
    protected static final Object[] decompressParent( MimdbTable table,byte[] binary,int parentLength ) {
        // 解凍処理.
        try {
            binary = JSnappy.decompress( binary ).getData() ;
        } catch( Exception e ) {
            // エラーの場合はランタイムエラーを返却.
            throw new MimdbException( e ) ;
        }
        int lineLen = table.columnTypes.length ;
        Object[] n ;
        Object[] ret = new Object[ parentLength ] ;
        int off = 0 ;
        for( int i = 0 ; i < parentLength ; i ++ ) {
            n = new Object[ lineLen ] ;
            off = binaryByLine( n,table,binary,off,lineLen ) ;
            ret[ i ] = n ;
        }
        return ret ;
    }
    
    /** 1行オブジェクトデータをバイナリ変換. **/
    protected static final int lineByBinary( MimdbTable table,Object[] values,byte[] out,int offset ) {
        int t,j,p,n ;
        long tt ;
        char c ;
        String s ;
        
        // １行データバイナリ化.
        int off = offset ;
        int len = values.length ;
        for( int i = 0 ; i < len ; i ++ ) {
            switch( ( n = table.columnTypes[ i ] ) ) {
            case MimdbIndex.COLUMN_BOOL :
                if( values[ i ] == null ) {
                    out[ off ++ ] = NULL_CODE ; // null.
                }
                else {
                    out[ off ++ ] = ( (Boolean)values[ i ] == true ) ? (byte)1 : (byte)0 ;
                }
                break ;
            case MimdbIndex.COLUMN_INT :
                if( values[ i ] == null ) {
                    out[ off ] = NULL_CODE ; // null.
                    off += 5 ;
                }
                else {
                    t = (Integer)values[ i ] ;
                    out[ off ++ ] = 0 ;// notNull.
                    out[ off ++ ] = (byte)((t&0xff000000)>>24) ;
                    out[ off ++ ] = (byte)((t&0xff0000)>>16) ;
                    out[ off ++ ] = (byte)((t&0xff00)>>8) ;
                    out[ off ++ ] = (byte)( t&0xff) ;
                }
                break ;
            case MimdbIndex.COLUMN_STRING :
                s = (String)values[ i ] ;
                if( s == null ) {
                    out[ off ++ ] = NULL_CODE ;// null.
                }
                else {
                    t = s.length() ;
                    p = 0 ;
                    // 長さをセット.
                    out[ off + p ] = (byte)(  (t&0xff000000)>>24) ;
                    out[ off + p+1 ] = (byte)((t&0xff0000)>>16) ;
                    out[ off + p+2 ] = (byte)((t&0xff00)>>8) ;
                    out[ off + p+3 ] = (byte)( t&0xff) ;
                    p += 4 ;
                    for( j = 0 ; j < t ; j ++ ) {
                        c = s.charAt( j ) ;
                        out[ off + p ] = (byte)( (c&0xff00)>>8) ;
                        out[ off + p+1 ] = (byte)(c&0xff) ;
                        p += 2 ;
                    }
                    // 全体の長さをセット.
                    off += ( t * 2 ) + 4 ;
                }
                break ;
            case MimdbIndex.COLUMN_LONG :
            case MimdbIndex.COLUMN_FLOAT :
            case MimdbIndex.COLUMN_DATE :
            case MimdbIndex.COLUMN_TIME :
            case MimdbIndex.COLUMN_TIMESTAMP :
                if( values[ i ] == null ) {
                    out[ off ] = NULL_CODE ; // null.
                    off += 9 ;
                }
                else {
                    if( MimdbIndex.COLUMN_LONG == n ) {
                        tt = (Long)values[ i ] ;
                    }
                    else if( MimdbIndex.COLUMN_FLOAT == n ) {
                        tt = Double.doubleToRawLongBits( (Double)values[ i ] ) ;
                    }
                    else {
                        tt = ( (java.util.Date)values[ i ] ).getTime() ;
                    }
                    out[ off ++ ] = 0 ;// notNull.
                    out[ off ++ ] = (byte)((tt&0xff00000000000000L)>>56L) ;
                    out[ off ++ ] = (byte)((tt&0xff000000000000L)>>48L) ;
                    out[ off ++ ] = (byte)((tt&0xff0000000000L)>>40L) ;
                    out[ off ++ ] = (byte)((tt&0xff00000000L)>>32L) ;
                    out[ off ++ ] = (byte)((tt&0xff000000L)>>24L) ;
                    out[ off ++ ] = (byte)((tt&0xff0000L)>>16L) ;
                    out[ off ++ ] = (byte)((tt&0xff00L)>>8L) ;
                    out[ off ++ ] = (byte)( tt&0xffL) ;
                }
                break ;
            }
        }
        return off ;
    }
    
    /** バイナリを1行オブジェクトデータに変換. **/
    protected static final int binaryByLine( Object[] out,MimdbTable table,byte[] binary,int offset,int lineLen ) {
        int t,j,p,n ;
        char[] cc ;
        long tt ;
        
        int off = offset ;
        for( int i = 0 ; i < lineLen ; i ++ ) {
            switch( (n = table.columnTypes[ i ]) ) {
            case MimdbIndex.COLUMN_BOOL :
                if( binary[ off ] == NULL_CODE ) { // null.
                    off ++ ;
                }
                else {
                    out[ i ] = new Boolean( binary[ off++ ] == 1 ) ;
                }
                break ;
            case MimdbIndex.COLUMN_INT :
                if( binary[ off++ ] == NULL_CODE ) { // null.
                    off += 4 ;
                }
                else {
                    out[ i ] = new Integer(
                        ( ((binary[ off ++ ]&0xff)<<24)|
                        ((binary[ off ++ ]&0xff)<<16)|
                        ((binary[ off ++ ]&0xff)<<8)|
                        ((binary[ off ++ ]&0xff)) ) ) ;
                }
                break ;
            case MimdbIndex.COLUMN_STRING :
                if( binary[ off ] == NULL_CODE ) { // null.
                    off += 1 ;
                }
                else {
                    p = 0 ;
                    t = ( ((binary[ off + p ]&0xff)<<24)|
                        ((binary[ off + p+1 ]&0xff)<<16)|
                        ((binary[ off + p+2 ]&0xff)<<8)|
                        ((binary[ off + p+3 ]&0xff)) ) ;
                    p += 4 ;
                    cc = new char[ t ] ;
                    for( j = 0 ; j < t ; j ++ ) {
                        cc[ j ] = (char)( ((binary[ off + p ]&0xff)<<8)|
                            (binary[ off + p+1 ]&0xff) ) ;
                        p += 2 ;
                    }
                    out[ i ] = new String( cc ) ;
                    off += ( t * 2 ) + 4 ;
                }
                break ;
            case MimdbIndex.COLUMN_LONG :
            case MimdbIndex.COLUMN_FLOAT :
            case MimdbIndex.COLUMN_DATE :
            case MimdbIndex.COLUMN_TIME :
            case MimdbIndex.COLUMN_TIMESTAMP :
                if( binary[ off++ ] == NULL_CODE ) { // null.
                    off += 8 ;
                }
                else {
                    tt = ( ((binary[ off ++ ]&0xffL)<<56L)|
                        ((binary[ off ++ ]&0xffL)<<48L)|
                        ((binary[ off ++ ]&0xffL)<<40L)|
                        ((binary[ off ++ ]&0xffL)<<32L)|
                        ((binary[ off ++ ]&0xffL)<<24L)|
                        ((binary[ off ++ ]&0xffL)<<16L)|
                        ((binary[ off ++ ]&0xffL)<<8L)|
                        ((binary[ off ++ ]&0xffL)) ) ;
                    if( MimdbIndex.COLUMN_LONG == n ) {
                        out[ i ] = new Long( tt ) ;
                    }
                    else if( MimdbIndex.COLUMN_FLOAT == n ) {
                        out[ i ] = Double.longBitsToDouble( tt ) ;
                    }
                    else if( MimdbIndex.COLUMN_DATE == n ) {
                        out[ i ] = new java.sql.Date( tt ) ;
                    }
                    else if( MimdbIndex.COLUMN_TIME == n ) {
                        out[ i ] = new java.sql.Time( tt ) ;
                    }
                    else {
                        out[ i ] = new java.sql.Timestamp( tt ) ;
                    }
                }
                break ;
            }
        }
        return off ;
    }
}
