package org.apache.hawq.pxf.plugins.jdbc;
/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

import org.apache.hawq.pxf.api.LogicalFilter;
import org.apache.hawq.pxf.plugins.jdbc.utils.DbProduct;
import org.apache.hawq.pxf.api.BasicFilter;
import org.apache.hawq.pxf.api.FilterParser;
import org.apache.hawq.pxf.api.io.DataType;
import org.apache.hawq.pxf.api.utilities.ColumnDescriptor;
import org.apache.hawq.pxf.api.utilities.InputData;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.util.ArrayList;
import java.util.List;
import java.text.ParseException;

/**
 * A WHERE queries builder
 *
 * Parses filter objects generated by {@link org.apache.hawq.pxf.plugins.jdbc.JdbcFilterBuilder} and builds WHERE statements
 * Only HDOP_AND is supported for multiple filters
 */
public class WhereSQLBuilder extends JdbcFilterBuilder {
    public WhereSQLBuilder(InputData input) {
        inputData = input;
    }

    /**
     * Insert WHERE constraints into a given query
     * Note that if filter is not supported, query is left unchanged
     *
     * @param dbName Database name (affects the behaviour for DATE constraints)
     * @param query SQL query to insert constraints to. The query may may contain other WHERE statements
     *
     * @throws ParseException if an error happens when parsing the constraints (provided to class constructor)
     */
    public void buildWhereSQL(String dbName, StringBuilder query) throws ParseException {
        if (!inputData.hasFilter()) {
            return;
        }

        try {
            StringBuilder prepared = new StringBuilder();
            if (!query.toString().contains("WHERE")) {
                prepared.append(" WHERE ");
            }
            else {
                prepared.append(" AND ");
            }

            // Get constraints and parse them
            String filterString = inputData.getFilterString();
            Object filterObj = getFilterObject(filterString);
            List<BasicFilter> filters = null;
            filters = convertBasicFilterList(filterObj, filters);

            String andDivisor = "";
            for (Object obj : filters) {
                prepared.append(andDivisor);
                andDivisor = " AND ";

                // Insert constraint column name
                BasicFilter filter = (BasicFilter) obj;
                ColumnDescriptor column = inputData.getColumn(filter.getColumn().index());
                prepared.append(column.columnName());

                // Insert constraint operator
                FilterParser.Operation op = filter.getOperation();
                switch (op) {
                    case HDOP_LT:
                        prepared.append(" < ");
                        break;
                    case HDOP_GT:
                        prepared.append(" > ");
                        break;
                    case HDOP_LE:
                        prepared.append(" <= ");
                        break;
                    case HDOP_GE:
                        prepared.append(" >= ");
                        break;
                    case HDOP_EQ:
                        prepared.append(" = ");
                        break;
                    case HDOP_LIKE:
                        prepared.append(" LIKE ");
                        break;
                    case HDOP_NE:
                        prepared.append(" <> ");
                        break;
                    case HDOP_IS_NULL:
                        prepared.append(" IS NULL");
                        continue;
                    case HDOP_IS_NOT_NULL:
                        prepared.append(" IS NOT NULL");
                        continue;
                    default:
                        throw new UnsupportedFilterException("Unsupported Filter operation: " + op);
                }

                // Insert constraint constant
                DbProduct dbProduct = DbProduct.getDbProduct(dbName);
                Object val = filter.getConstant().constant();
                switch (DataType.get(column.columnTypeCode())) {
                    case SMALLINT:
                    case INTEGER:
                    case BIGINT:
                    case FLOAT8:
                    case REAL:
                    case BOOLEAN:
                        prepared.append(val.toString());
                        break;
                    case TEXT:
                        prepared.append("'").append(val.toString()).append("'");
                        break;
                    case DATE:
                        // Date field has different format in different databases
                        prepared.append(dbProduct.wrapDate(val));
                        break;
                    case TIMESTAMP:
                        // Timestamp field has different format in different databases
                        prepared.append(dbProduct.wrapTimestamp(val));
                        break;
                    default:
                        throw new UnsupportedFilterException("Unsupported column type for filtering: " + column.columnTypeCode());
                }
            }

            // No exceptions were thrown, change the provided query
            query.append(prepared);
        }
        catch (UnsupportedFilterException e) {
            LOG.debug("WHERE clause is omitted: " + e.toString());
            // Silence the exception and do not insert constraints
        }
    }

    /**
     * Convert filter object into a list of {@link BasicFilter}
     *
     * @param filter Filter object
     * @param returnList A list of {@link BasicFilter} to append filters to. Must be null if the function is not called recursively
     */
    private static List<BasicFilter> convertBasicFilterList(Object filter, List<BasicFilter> returnList) throws UnsupportedFilterException {
        if (returnList == null) {
            returnList = new ArrayList<>();
        }

        if (filter instanceof BasicFilter) {
            returnList.add((BasicFilter) filter);
            return returnList;
        }

        LogicalFilter lfilter = (LogicalFilter) filter;
        if (lfilter.getOperator() != FilterParser.LogicalOperation.HDOP_AND) {
            throw new UnsupportedFilterException("Logical operation '" + lfilter.getOperator() + "' is not supported");
        }
        for (Object f : lfilter.getFilterList()) {
            returnList = convertBasicFilterList(f, returnList);
        }

        return returnList;
    }

    private static class UnsupportedFilterException extends Exception {
		UnsupportedFilterException(String message) {
            super(message);
        }
    }

    private static final Log LOG = LogFactory.getLog(WhereSQLBuilder.class);

    // {@link InputData} from PXF
    private InputData inputData;
}