/**
 * Copyright © 2012-2023 by Data Virtuality GmbH
 */
package org.teiid.translator.jdbc.databricks;

import org.teiid.core.types.DataTypeManager;
import org.teiid.language.Call;
import org.teiid.language.Command;
import org.teiid.language.Function;
import org.teiid.metadata.RuntimeMetadata;
import org.teiid.translator.*;
import org.teiid.translator.jdbc.*;

import java.sql.Connection;
import java.sql.SQLException;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * @author Adil Hamiani
 * @date 2023.09.24
 */
@Translator(name = "databricks", description = "Databricks translator")
public class DataBricksExecutionFactory extends JDBCExecutionFactory {

    private ConvertModifier convert = new ConvertModifier();

    public DataBricksExecutionFactory() {
        setSupportsSelectDistinct(true);
        setSupportsInnerJoins(true);
        setSupportsOuterJoins(true);
        setSupportsRowLimit(true);
        setSupportsOrderBy(true);
        setSupportsOrderByAlias(true);
        setSupportsOrderByString(true);
        setSupportsSelectDistinct(true);
        setSupportsFullOuterJoins(true);
        setUseBindVariables(false);
        setTransactionSupport(TransactionSupport.NONE);
    }


    @Override
    public boolean useAnsiJoin() {
        return true;
    }


    @Override
    public ProcedureExecution createProcedureExecution(Call command, ExecutionContext executionContext, RuntimeMetadata metadata, Connection conn)
            throws TranslatorException {
        throw new TranslatorException(notSupported(command));
    }

    private String notSupported(Command command) {
        return MessageFormat.format("This {0} command is not supported on Amazon Athena.", command);
    }

    @Override
    public boolean isSourceRequiredForMetadata() {
        return true;
    }

    @Override
    public boolean supportsConvert(int fromType, int toType) {
        if (!super.supportsConvert(fromType, toType)) {
            return false;
        }
        if (convert.hasTypeMapping(toType)) {
            return true;
        }
        return false;
    }


    @Override
    public void start() throws TranslatorException {
        super.start();

        convert.addTypeMapping("tinyint", FunctionModifier.SHORT); 
        convert.addTypeMapping("integer", FunctionModifier.LONG);
        convert.addTypeMapping("smallint", FunctionModifier.INTEGER);
        convert.addTypeMapping("boolean", FunctionModifier.BOOLEAN); 
        convert.addTypeMapping("bigint", FunctionModifier.BIGINTEGER, FunctionModifier.LONG); 
        convert.addTypeMapping("double", FunctionModifier.DOUBLE); 
        convert.addTypeMapping("varchar", FunctionModifier.STRING); 
        convert.addTypeMapping("date", FunctionModifier.DATE);
        convert.addTypeMapping("decimal", FunctionModifier.BIGDECIMAL);
        convert.addTypeMapping("timestamp", FunctionModifier.TIMESTAMP);
        convert.addTypeMapping("real", FunctionModifier.FLOAT);
        convert.addTypeMapping("varbinary", FunctionModifier.BLOB);

        convert.addConvert(FunctionModifier.FLOAT, FunctionModifier.DOUBLE, new FunctionModifier() {
            @Override
            public List<?> translate(Function function) {
                return Arrays.asList("cast("+function.getParameters().get(0) +" as double)");  //$NON-NLS-2$
            }
        });

        convert.addConvert(FunctionModifier.DOUBLE, FunctionModifier.FLOAT, new FunctionModifier() {
            @Override
            public List<?> translate(Function function) {
                return Arrays.asList("cast("+function.getParameters().get(0) +" as float)");  //$NON-NLS-2$
            }
        });

        registerFunctionModifier(SourceSystemFunctions.CONVERT, convert);
    }

    @Override
    public List<String> getSupportedFunctions() {
        List<String> supportedFunctions = new ArrayList<String>();
        supportedFunctions.addAll(super.getDefaultSupportedFunctions());

        supportedFunctions.add(SourceSystemFunctions.ABS);
        supportedFunctions.add(SourceSystemFunctions.ACOS);
        supportedFunctions.add(SourceSystemFunctions.ASIN);
        supportedFunctions.add(SourceSystemFunctions.ATAN);
        supportedFunctions.add(SourceSystemFunctions.ATAN2);
        supportedFunctions.add(SourceSystemFunctions.CEILING);
        supportedFunctions.add(SourceSystemFunctions.CHAR);
        supportedFunctions.add(SourceSystemFunctions.COALESCE);
        supportedFunctions.add(SourceSystemFunctions.CONCAT);
        supportedFunctions.add(SourceSystemFunctions.COS);
        supportedFunctions.add(SourceSystemFunctions.CURDATE);
        supportedFunctions.add(SourceSystemFunctions.DAYOFMONTH);
        supportedFunctions.add(SourceSystemFunctions.DAYOFWEEK);
        supportedFunctions.add(SourceSystemFunctions.DAYOFYEAR);
        supportedFunctions.add(SourceSystemFunctions.EXP);
        supportedFunctions.add(SourceSystemFunctions.FLOOR);

        supportedFunctions.add(SourceSystemFunctions.HOUR);
        supportedFunctions.add(SourceSystemFunctions.IFNULL);
        supportedFunctions.add(SourceSystemFunctions.LCASE);

        supportedFunctions.add(SourceSystemFunctions.LENGTH);
        supportedFunctions.add(SourceSystemFunctions.LTRIM);
        supportedFunctions.add(SourceSystemFunctions.LOG);
        supportedFunctions.add(SourceSystemFunctions.LOG10);
        supportedFunctions.add(SourceSystemFunctions.MINUTE);
        supportedFunctions.add(SourceSystemFunctions.MOD);
        supportedFunctions.add(SourceSystemFunctions.MONTH);
        supportedFunctions.add(SourceSystemFunctions.NOW);
        supportedFunctions.add(SourceSystemFunctions.PI);
        supportedFunctions.add(SourceSystemFunctions.POWER);
        supportedFunctions.add(SourceSystemFunctions.QUARTER);
        supportedFunctions.add(SourceSystemFunctions.RAND);
        supportedFunctions.add(SourceSystemFunctions.REPLACE);
        supportedFunctions.add(SourceSystemFunctions.ROUND);
        supportedFunctions.add(SourceSystemFunctions.RTRIM);
        supportedFunctions.add(SourceSystemFunctions.SECOND);
        supportedFunctions.add(SourceSystemFunctions.SQRT);
        supportedFunctions.add(SourceSystemFunctions.SIN);
        supportedFunctions.add(SourceSystemFunctions.SUBSTRING);
        supportedFunctions.add(SourceSystemFunctions.TAN);
        supportedFunctions.add(SourceSystemFunctions.TRANSLATE);
        supportedFunctions.add(SourceSystemFunctions.TRIM);
        supportedFunctions.add(SourceSystemFunctions.UCASE);
        supportedFunctions.add(SourceSystemFunctions.YEAR);
        return supportedFunctions;
    }

    public JDBCMetadataProcessor.ITypeConverter getTypeConverter() {
        return new JDBCMetadataProcessor.ITypeConverter() {

            @Override
            public String getRuntimeType(int type, String typeName, int precision, int scale) {
                return null;
            }

            @Override
            protected String usualNumericCheck(String typeName, int precision, int scale) {
                if ("DECIMAL".equalsIgnoreCase(typeName)) {
                    return DataTypeManager.DefaultDataTypes.BIG_DECIMAL;
                }
                return super.usualNumericCheck(typeName, precision, scale);
            }

        };
    }


    @Override
    public boolean supportsInlineViews() {
        return true;
    }

    @Override
    public boolean supportsExistsCriteria() {
        return true;
    }

    public boolean supportsOnlyLiteralComparison() {
        return true;
    }

    @Override
    public boolean supportsOrderByNullOrdering() {
        return true;
    }

    @Override
    public boolean supportsAggregatesEnhancedNumeric() {
        return true;
    }

    @Override
    public boolean supportsIntersect() {
        return true;
    }

    @Override
    public boolean supportsExcept() {
        return true;
    }

    @Override
    public boolean supportsRowLimit() {
        return true;
    }

    @Override
    public boolean supportsRowOffset() {
        return false;
    }

    @Override
    public boolean supportsFunctionsInGroupBy() {
        return true;
    }

    @Override
    public boolean supportsBatchedUpdates() {
        return false;
    }

    @Override
    public boolean supportsCommonTableExpressions() {
        return true;
    }

    @Override
    public boolean supportsElementaryOlapOperations() {
        return true;
    }

    @Override
    public boolean supportsArrayType() {
        return true;
    }

    @Override
    public boolean supportsCorrelatedSubqueries() {
        return false;
    }


    @Override
    public UpdateExecution createUpdateExecution(Command command, ExecutionContext executionContext, RuntimeMetadata metadata, Connection conn)
            throws TranslatorException {
        obtainedConnection(conn);
        return new DatabricksJDBCUpdateExecution(command, conn, executionContext, this);
    }

    static class DatabricksJDBCUpdateExecution extends JDBCUpdateExecution {

        public DatabricksJDBCUpdateExecution(Command command, Connection connection, ExecutionContext context, JDBCExecutionFactory env) {
            super(command, connection, context, env);
        }

        @Override
        public void execute() throws TranslatorException {
            boolean commitType;
            try {
                commitType = connection.getAutoCommit();
            } catch (SQLException e) {
                throw new TranslatorException(e, TranslatorJdbcPlugin.Util.gs(TranslatorJdbcPlugin.Event.DVCORE94092));
            }
            if (commitType) {
                try {
                    connection.setAutoCommit(false);
                } catch (SQLException e) {
                    throw new TranslatorException(e, TranslatorJdbcPlugin.Util.gs(TranslatorJdbcPlugin.Event.DVCORE94093));
                }
            }
            try {
                super.execute();
            } catch (TranslatorException e) {
                if (commitType)
                    try {
                        connection.rollback();
                    } catch (SQLException e1) {
                        e1.printStackTrace();
                    }
                throw e;
            } finally {
                if (commitType) {
                    try {
                        connection.setAutoCommit(true);
                    } catch (SQLException e) {
                        throw new TranslatorException(e, TranslatorJdbcPlugin.Util.gs(TranslatorJdbcPlugin.Event.DVCORE94094));
                    }
                }
            }
        }
    }


}
