/*******************************************************************************
 * Copyright (c) 2013, Salesforce.com, Inc.
 * All rights reserved.
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 * 
 *     Redistributions of source code must retain the above copyright notice,
 *     this list of conditions and the following disclaimer.
 *     Redistributions in binary form must reproduce the above copyright notice,
 *     this list of conditions and the following disclaimer in the documentation
 *     and/or other materials provided with the distribution.
 *     Neither the name of Salesforce.com nor the names of its contributors may 
 *     be used to endorse or promote products derived from this software without 
 *     specific prior written permission.
 * 
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE 
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL 
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR 
 * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER 
 * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, 
 * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE 
 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 ******************************************************************************/
package com.salesforce.phoenix.expression;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.sql.SQLException;

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.io.WritableUtils;

import com.salesforce.phoenix.exception.SQLExceptionCode;
import com.salesforce.phoenix.exception.SQLExceptionInfo;
import com.salesforce.phoenix.expression.visitor.ExpressionVisitor;
import com.salesforce.phoenix.schema.ColumnModifier;
import com.salesforce.phoenix.schema.IllegalDataException;
import com.salesforce.phoenix.schema.PDataType;
import com.salesforce.phoenix.schema.TypeMismatchException;
import com.salesforce.phoenix.schema.tuple.Tuple;
import com.salesforce.phoenix.util.ByteUtil;
import com.salesforce.phoenix.util.StringUtil;



/**
 * 
 * Accessor for a literal value.
 *
 * @author jtaylor
 * @since 0.1
 */
public class LiteralExpression extends BaseTerminalExpression {
    public static final LiteralExpression NULL_EXPRESSION = new LiteralExpression(null, false);
    private static final LiteralExpression ND_NULL_EXPRESSION = new LiteralExpression(null, true);
    private static final LiteralExpression[] TYPED_NULL_EXPRESSIONS = new LiteralExpression[PDataType.values().length * 2];
    static {
        for (int i = 0; i < PDataType.values().length; i++) {
            TYPED_NULL_EXPRESSIONS[i] = new LiteralExpression(PDataType.values()[i], true);
        }
        for (int i = 0; i < PDataType.values().length; i++) {
            TYPED_NULL_EXPRESSIONS[i+PDataType.values().length] = new LiteralExpression(PDataType.values()[i], false);
        }
    }
    private static final LiteralExpression FALSE_EXPRESSION = new LiteralExpression(Boolean.FALSE, PDataType.BOOLEAN, PDataType.BOOLEAN.toBytes(Boolean.FALSE), true);
    private static final LiteralExpression TRUE_EXPRESSION = new LiteralExpression(Boolean.TRUE, PDataType.BOOLEAN, PDataType.BOOLEAN.toBytes(Boolean.TRUE), true);
    private static final LiteralExpression ND_FALSE_EXPRESSION = new LiteralExpression(Boolean.FALSE, PDataType.BOOLEAN, PDataType.BOOLEAN.toBytes(Boolean.FALSE), false);
    private static final LiteralExpression ND_TRUE_EXPRESSION = new LiteralExpression(Boolean.TRUE, PDataType.BOOLEAN, PDataType.BOOLEAN.toBytes(Boolean.TRUE), false);

    private Object value;
    private PDataType type;
    private boolean isDeterministic;
    private byte[] byteValue;
    private Integer byteSize;
    private Integer maxLength;
    private Integer scale;
    private ColumnModifier columnModifier;


    public static boolean isFalse(Expression child) {
        return child == FALSE_EXPRESSION || child == ND_FALSE_EXPRESSION;
    }
    
    public static boolean isTrue(Expression child) {
        return child == TRUE_EXPRESSION || child == ND_TRUE_EXPRESSION;
    }
    
    public static LiteralExpression newConstant(Object value) {
        return newConstant(value, true);
    }
    
    // TODO: cache?
    public static LiteralExpression newConstant(Object value, boolean isDeterministic) {
        if (Boolean.FALSE.equals(value)) {
            return isDeterministic ? FALSE_EXPRESSION : ND_FALSE_EXPRESSION;
        }
        if (Boolean.TRUE.equals(value)) {
            return isDeterministic ? TRUE_EXPRESSION : ND_TRUE_EXPRESSION;
        }
        if (value == null) {
            return isDeterministic ? NULL_EXPRESSION : ND_NULL_EXPRESSION;
        }
        PDataType type = PDataType.fromLiteral(value);
        byte[] b = type.toBytes(value);
        if (b.length == 0) {
            return TYPED_NULL_EXPRESSIONS[type.ordinal() + ( isDeterministic ? 0 : TYPED_NULL_EXPRESSIONS.length/2)];
        }
        if (type == PDataType.VARCHAR) {
            String s = (String) value;
            if (s.length() == b.length) { // single byte characters only
                type = PDataType.CHAR;
            }
        }
        return new LiteralExpression(value, type, b, isDeterministic);
    }

    public static LiteralExpression newConstant(Object value, PDataType type) throws SQLException {
        return newConstant(value, type, true);
    }
    
    public static LiteralExpression newConstant(Object value, PDataType type, boolean isDeterministic) throws SQLException {
        return newConstant(value, type, null, isDeterministic);
    }
    
    public static LiteralExpression newConstant(Object value, PDataType type, ColumnModifier columnModifier) throws SQLException {
        return newConstant(value, type, null, null, columnModifier, true);
    }
    
    public static LiteralExpression newConstant(Object value, PDataType type, ColumnModifier columnModifier, boolean isDeterministic) throws SQLException {
        return newConstant(value, type, null, null, columnModifier, isDeterministic);
    }
    
    public static LiteralExpression newConstant(Object value, PDataType type, Integer maxLength, Integer scale) throws SQLException {
        return newConstant(value, type, maxLength, scale, null, true);
    }
    
    public static LiteralExpression newConstant(Object value, PDataType type, Integer maxLength, Integer scale, boolean isDeterministic) throws SQLException { // remove?
        return newConstant(value, type, maxLength, scale, null, isDeterministic);
    }

    // TODO: cache?
    public static LiteralExpression newConstant(Object value, PDataType type, Integer maxLength, Integer scale, ColumnModifier columnModifier, boolean isDeterministic)
            throws SQLException {
        if (value == null) {
            if (type == null) {
                return NULL_EXPRESSION;
            }
            return TYPED_NULL_EXPRESSIONS[type.ordinal()];
        }
        if (Boolean.FALSE.equals(value)) {
            return isDeterministic ? FALSE_EXPRESSION : ND_FALSE_EXPRESSION;
        }
        if (Boolean.TRUE.equals(value)) {
            return isDeterministic ? TRUE_EXPRESSION : ND_TRUE_EXPRESSION;
        }
        PDataType actualType = PDataType.fromLiteral(value);
        // For array we should check individual element in it?
        // It would be costly though!!!!!
        if (!actualType.isCoercibleTo(type, value)) {
            throw TypeMismatchException.newException(type, actualType, value.toString());
        }
        value = type.toObject(value, actualType);
        try {
            byte[] b = type.toBytes(value, columnModifier);
            if (type == PDataType.VARCHAR || type == PDataType.CHAR) {
                if (type == PDataType.CHAR && maxLength != null  && b.length < maxLength) {
                    b = StringUtil.padChar(b, maxLength);
                } else if (value != null) {
                    maxLength = ((String)value).length();
                }
            }
            if (b.length == 0) {
                return TYPED_NULL_EXPRESSIONS[type.ordinal()];
            }
            return new LiteralExpression(value, type, b, maxLength, scale, columnModifier, isDeterministic);
        } catch (IllegalDataException e) {
            throw new SQLExceptionInfo.Builder(SQLExceptionCode.ILLEGAL_DATA).setRootCause(e).build().buildException();
        }
    }

    public LiteralExpression() {
    }

    private LiteralExpression(PDataType type, boolean isDeterministic) {
        this(null, type, ByteUtil.EMPTY_BYTE_ARRAY, isDeterministic);
    }

    private LiteralExpression(Object value, PDataType type, byte[] byteValue, boolean isDeterministic) {
        this(value, type, byteValue, type == null? null : type.getMaxLength(value), type == null? null : type.getScale(value), null, isDeterministic);
    }

    private LiteralExpression(Object value, PDataType type, byte[] byteValue,
            Integer maxLength, Integer scale, ColumnModifier columnModifier, boolean isDeterministic) {
        this.value = value;
        this.type = type;
        this.byteValue = byteValue;
        this.byteSize = byteValue.length;
        this.maxLength = maxLength;
        this.scale = scale;
        this.columnModifier = columnModifier;
        this.isDeterministic = isDeterministic;
    }

    @Override
    public boolean isDeterministic() {
        return isDeterministic;
    }
    
    @Override
    public String toString() {
        return type != null && type.isCoercibleTo(PDataType.VARCHAR) ? "'" + value + "'" : "" + value;
    }

    @Override
    public int hashCode() {
        final int prime = 31;
        int result = 1;
        result = prime * result + ((value == null) ? 0 : value.hashCode());
        return result;
    }

    @Override
    public boolean equals(Object obj) {
        if (this == obj) return true;
        if (obj == null) return false;
        if (getClass() != obj.getClass()) return false;
        LiteralExpression other = (LiteralExpression)obj;
        if (value == null) {
            if (other.value != null) return false;
        } else if (!value.equals(other.value)) return false;
        return true;
    }

    @Override
    public void readFields(DataInput input) throws IOException {
        int encodedByteLengthAndBool = WritableUtils.readVInt(input);
        this.isDeterministic = encodedByteLengthAndBool > 0;
        int byteLength = Math.abs(encodedByteLengthAndBool)-1;
        this.byteValue = new byte[byteLength];
        input.readFully(byteValue, 0, byteLength);
        columnModifier = ColumnModifier.fromSystemValue(WritableUtils.readVInt(input));
        int typeOrdinal = WritableUtils.readVInt(input);
        if (typeOrdinal < 0) {
            this.type = null;
        } else {
            this.type = PDataType.values()[typeOrdinal];
        }
        if (this.byteValue.length == 0) {
            this.value = null;
        } else {
            this.value = this.type.toObject(byteValue, 0, byteValue.length, this.type, columnModifier);
        }
        // Only to prevent continual reallocations of Integer
        this.byteSize = this.byteValue.length;
    }

    @Override
    public void write(DataOutput output) throws IOException {
        WritableUtils.writeVInt(output, (byteValue.length + 1) * (this.isDeterministic ? 1 : -1));
        output.write(byteValue);
        WritableUtils.writeVInt(output, ColumnModifier.toSystemValue(columnModifier));
        WritableUtils.writeVInt(output, type == null ? -1 : this.type.ordinal());
    }

    @Override
    public boolean evaluate(Tuple tuple, ImmutableBytesWritable ptr) {
        // Literal always evaluates, even when it returns null
        ptr.set(byteValue);
        return true;
    }

    @Override
    public PDataType getDataType() {
        return type;
    }

    @Override
    public Integer getByteSize() {
        return byteSize;
    }

    @Override
    public Integer getMaxLength() {
        return maxLength;
    }

    @Override
    public Integer getScale() {
        return scale;
    }
    
    @Override
    public ColumnModifier getColumnModifier() {
        return columnModifier;
    }

    @Override
    public boolean isNullable() {
        return value == null;
    }

    public Object getValue() {
        return value;
    }

    public byte[] getBytes() {
        return byteValue;
    }
    
    @Override
    public final <T> T accept(ExpressionVisitor<T> visitor) {
        return visitor.visit(this);
    }
    
    @Override
    public boolean isStateless() {
        return true;
    }
}
