/*
 * 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.
 */
package org.apache.openjpa.jdbc.sql;

import java.io.InputStream;
import java.io.StringReader;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.security.AccessController;
import java.sql.Blob;
import java.sql.Clob;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Timestamp;
import java.sql.Types;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.OffsetDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Locale;
import java.util.Map;

import org.apache.openjpa.jdbc.identifier.DBIdentifier;
import org.apache.openjpa.jdbc.kernel.JDBCFetchConfiguration;
import org.apache.openjpa.jdbc.kernel.JDBCStore;
import org.apache.openjpa.jdbc.kernel.exps.FilterValue;
import org.apache.openjpa.jdbc.meta.JavaSQLTypes;
import org.apache.openjpa.jdbc.schema.Column;
import org.apache.openjpa.jdbc.schema.ForeignKey;
import org.apache.openjpa.jdbc.schema.ForeignKey.FKMapKey;
import org.apache.openjpa.jdbc.schema.Index;
import org.apache.openjpa.jdbc.schema.PrimaryKey;
import org.apache.openjpa.jdbc.schema.Table;
import org.apache.openjpa.jdbc.schema.Unique;
import org.apache.openjpa.lib.jdbc.DelegatingDatabaseMetaData;
import org.apache.openjpa.lib.jdbc.DelegatingPreparedStatement;
import org.apache.openjpa.lib.util.J2DoPrivHelper;
import org.apache.openjpa.lib.util.Localizer;
import org.apache.openjpa.meta.JavaTypes;
import org.apache.openjpa.util.StoreException;
import org.apache.openjpa.util.UserException;


/**
 * Dictionary for Oracle.
 */
public class OracleDictionary
    extends DBDictionary {

    public static final String SELECT_HINT = "openjpa.hint.OracleSelectHint";
    public static final String VENDOR_ORACLE = "oracle";

    private static final int BEHAVE_OTHER = 0;
    private static final int BEHAVE_ORACLE = 1;
    private static final int BEHAVE_DATADIRECT31 = 2;

    private static Blob EMPTY_BLOB = null;
    private static Clob EMPTY_CLOB = null;

    private static final Localizer _loc = Localizer.forPackage
        (OracleDictionary.class);
    /**
     * If true, then simulate auto-assigned values in Oracle by
     * using a trigger that inserts a sequence value into the
     * primary key value when a row is inserted.
     */
    @Deprecated
    public boolean useTriggersForAutoAssign = false;

    /**
     * The global sequence name to use for autoassign simulation.
     */
    @Deprecated
    public String autoAssignSequenceName = null;

    /**
     * Whether this JDBC driver has native java.time support for
     * LocalDate, LocalDateTime, LocalTime, OffsetTime and OffsetDateTime
     *
     *
     */
    public boolean nativeJavaTimeSupport = true;

    /**
     * Flag to use OpenJPA 0.3 style naming for auto assign sequence name and
     * trigger name for backwards compatibility.
     */
    public boolean openjpa3GeneratedKeyNames = false;

    /**
     * If true, then OpenJPA will attempt to use the special
     * OraclePreparedStatement.setFormOfUse method to
     * configure statements that it detects are operating on unicode fields.
     */
    public boolean useSetFormOfUseForUnicode = true;

    /**
     * This variable was used prior to 2.1.x to indicate that OpenJPA should attempt to use
     * a Reader-based JDBC 4.0 method to set Clob or XML data.  It allowed XMLType and
     * Clob values larger than 4000 bytes to be used.  For 2.1.x+, code was added to allow
     * said functionality by default (see OPENJPA-1691).  For forward compatibility, this
     * variable should not be removed.
     */
    @Deprecated
    public boolean supportsSetClob = false;

    /**
     * If a user sets the previous variable (supportsSetClob) to true, we should log a
     * warning indicating that the variable no longer has an effect due to the code changes
     * of OPENJPA-1691.  We only want to log the warning once per instance, thus this
     * variable will be used to indicate if the warning should be printed or not.
     */
    @Deprecated
    private boolean logSupportsSetClobWarning = true;

    /**
     * Type constructor for XML column, used in INSERT and UPDATE statements.
     */
    public String xmlTypeMarker = "XMLType(?)";

    // some oracle drivers have problems with select for update; warn the
    // first time locking is attempted
    private boolean _checkedUpdateBug = false;

    private boolean _warnedCharColumn = false;
    private boolean _warnedNcharColumn = false;
    private int _driverBehavior = -1;
    // cache lob methods
    private Method _putBytes = null;

    private Method _putString = null;
    private Method _putChars = null;
    // cache some native Oracle classes and methods
    private Class oraclePreparedStatementClass = null;

    private Field oraclePreparedStatementFormNvarcharField = null;
    private Method oracleClob_empty_lob_Method = null;
    private Method oracleBlob_empty_lob_Method = null;
    private Method oracleClob_isEmptyLob_Method = null;


    // batch limit
    private int defaultBatchLimit = 100;

    public OracleDictionary() {
        platform = "Oracle";
        validationSQL = "SELECT SYSDATE FROM DUAL";
        nextSequenceQuery = "SELECT {0}.NEXTVAL FROM DUAL";
        stringLengthFunction = "LENGTH({0})";
        joinSyntax = SYNTAX_DATABASE;
        maxTableNameLength = 30;
        maxColumnNameLength = 30;
        maxIndexNameLength = 30;
        maxConstraintNameLength = 30;
        maxEmbeddedBlobSize = 4000;
        maxEmbeddedClobSize = 4000;
        inClauseLimit = 1000;

        if (supportsAutoAssign && useTriggersForAutoAssign) {
            log.warn("Both 'supportsAutoAssign' and 'useTriggersForAutoAssign' were specified, such configuration will never work,"
                    + " please use 'supportsAutoAssign'");
        }
        // support auto increment columns jakarta.persistence.GenerationType#IDENTITY
        supportsAutoAssign = true;
        autoAssignClause = "GENERATED ALWAYS AS IDENTITY";

        supportsDeferredConstraints = true;
        supportsLockingWithDistinctClause = false;
        supportsSelectStartIndex = true;
        supportsSelectEndIndex = true;

        systemSchemaSet.addAll(Arrays.asList(new String[]{
            "CTXSYS", "MDSYS", "SYS", "SYSTEM", "WKSYS", "WMSYS", "XDB",
        }));

        supportsXMLColumn = true;
        xmlTypeName = "XMLType";
        bigintTypeName = "NUMBER{0}";
        bitTypeName = "NUMBER{0}";
        decimalTypeName = "NUMBER{0}";
        doubleTypeName = "NUMBER{0}";
        integerTypeName = "NUMBER{0}";
        numericTypeName = "NUMBER{0}";
        smallintTypeName = "NUMBER{0}";
        tinyintTypeName = "NUMBER{0}";
        longVarcharTypeName = "LONG";
        binaryTypeName = "BLOB";
        varbinaryTypeName = "BLOB";
        longVarbinaryTypeName = "BLOB";
        timeTypeName = "DATE";
        timeWithZoneTypeName = "DATE";

        // default TIMESTAMP is TIMESTAMP(6) which means MICROs
        datePrecision = MICRO;

        varcharTypeName = "VARCHAR2{0}";
        fixedSizeTypeNameSet.addAll(Arrays.asList(new String[]{
            "LONG RAW", "RAW", "LONG", "REF",
        }));
        reservedWordSet.addAll(Arrays.asList(new String[]{
            "ACCESS", "AUDIT", "CLUSTER", "COMMENT", "COMPRESS", "EXCLUSIVE",
            "FILE", "IDENTIFIED", "INCREMENT", "INDEX", "INITIAL", "LOCK",
            "LONG", "MAXEXTENTS", "MINUS", "MODE", "NOAUDIT", "NOCOMPRESS",
            "NOWAIT", "OFFLINE", "ONLINE", "PCTFREE", "ROW",
        }));

        // reservedWordSet subset that CANNOT be used as valid column names
        // (i.e., without surrounding them with double-quotes)
        // generated at 2021-05-02T16:36:42.608 via org.apache.openjpa.reservedwords.ReservedWordsIT
        invalidColumnWordSet.addAll(Arrays.asList(new String[] {
            "ACCESS", "ADD", "ALL", "ALTER", "AND", "ANY", "AS", "ASC", "AUDIT", "BETWEEN", "BY", "CHAR", "CHECK", "CLUSTER",
            "COLUMN", "COMMENT", "COMPRESS", "CONNECT", "CREATE", "CURRENT", "DATE", "DECIMAL", "DEFAULT", "DELETE", "DESC",
            "DISTINCT", "DROP", "ELSE", "END-EXEC", "EXCLUSIVE", "EXISTS", "FILE", "FLOAT", "FOR", "FROM", "GRANT", "GROUP",
            "HAVING", "IDENTIFIED", "IMMEDIATE", "IN", "INCREMENT", "INDEX", "INITIAL", "INSERT", "INTEGER", "INTERSECT", "INTO",
            "IS", "LEVEL", "LIKE", "LOCK", "LONG", "MAXEXTENTS", "MINUS", "MLSLABEL", "MODE", "MODIFY", "NOAUDIT", "NOCOMPRESS",
            "NOT", "NOWAIT", "NULL", "NUMBER", "OF", "OFFLINE", "ON", "ONLINE", "OPTION", "OR", "ORDER", "PCTFREE", "PRIOR",
            "PUBLIC", "RAW", "RENAME", "RESOURCE", "REVOKE", "ROW", "ROWID", "ROWNUM", "ROWS", "SELECT", "SESSION", "SET",
            "SHARE", "SIZE", "SMALLINT", "START", "SUCCESSFUL", "SYNONYM", "SYSDATE", "TABLE", "THEN", "TO", "TRIGGER", "UID",
            "UNION", "UNIQUE", "UPDATE", "USER", "VALIDATE", "VALUES", "VARCHAR", "VARCHAR2", "VIEW", "WHENEVER", "WHERE",
            "WITH",
            // end generated.
            // The following keywords used to be defined as reserved words in the past, but now seem to work
            // we still add them for compat reasons
            "PRIVILEGES"
        }));

        substringFunctionName = "SUBSTR";
        super.setBatchLimit(defaultBatchLimit);
        selectWordSet.add("WITH");
        reportsSuccessNoInfoOnBatchUpdates = true;

        try {
            oraclePreparedStatementClass = Class.forName("oracle.jdbc.OraclePreparedStatement");
            try {
                oraclePreparedStatementFormNvarcharField = oraclePreparedStatementClass.getField("FORM_NCHAR");
                oraclePreparedStatementFormNvarcharField.setAccessible(true);
            }
            catch (NoSuchFieldException e) {
                log.warn("OraclePreparedStatement without FORM_NCHAR field found");
            }
        }
        catch (ClassNotFoundException e) {
            // all fine
        }

        oracleClob_empty_lob_Method = getMethodByReflection("oracle.sql.CLOB", "getEmptyCLOB");
        oracleBlob_empty_lob_Method = getMethodByReflection("oracle.sql.BLOB", "getEmptyBLOB");
        oracleClob_isEmptyLob_Method = getMethodByReflection("oracle.sql.CLOB", "isEmptyLob");

        indexPhysicalForeignKeys = true; // Oracle does not automatically create an index for a foreign key so we will
    }

    private Method getMethodByReflection(String className, String methodName, Class<?>... paramTypes) {
        try {
            return Class.forName(className,true,
                    AccessController.doPrivileged(J2DoPrivHelper
                            .getContextClassLoaderAction())).
                    getMethod(methodName, paramTypes);
        }
        catch (Exception e) {
            // all fine
        }

        return null;
    }

    @Override
    public void endConfiguration() {
        super.endConfiguration();
        if (useTriggersForAutoAssign)
            supportsAutoAssign = false;
    }

    @Override
    public void connectedConfiguration(Connection conn)
        throws SQLException {
        super.connectedConfiguration(conn);
        if (driverVendor == null) {
            DatabaseMetaData meta = conn.getMetaData();
            String url = (meta.getURL() == null) ? "" : meta.getURL();
            String driverName = meta.getDriverName();
            String metadataClassName;
            if (meta instanceof DelegatingDatabaseMetaData)
                metadataClassName = ((DelegatingDatabaseMetaData) meta).
                    getInnermostDelegate().getClass().getName();
            else
                metadataClassName = meta.getClass().getName();

            // check both the driver class name and the URL for known patterns
            if (metadataClassName.startsWith("oracle.")
                || url.indexOf("jdbc:oracle:") != -1
                || "Oracle JDBC driver".equals(driverName)) {
                int jdbcMajor = meta.getDriverMajorVersion();
                int jdbcMinor = meta.getDriverMinorVersion();
                driverVendor = VENDOR_ORACLE + jdbcMajor + jdbcMinor;

                int jdbcVersion = jdbcMajor * 1000 + jdbcMinor;
                if( jdbcVersion >= 11002) {
                    maxEmbeddedBlobSize = -1;
                    maxEmbeddedClobSize = -1;
                }
                if (jdbcMajor < 18) {
                    // no native java.time support for old JDBC drivers.
                    nativeJavaTimeSupport = false;
                }
                String productVersion = meta.getDatabaseProductVersion()
                    .split("Release ",0)[1].split("\\.",0)[0];
                int release = Integer.parseInt(productVersion);

                // warn sql92
                if (release <= 8) {
                    if (joinSyntax == SYNTAX_SQL92 && log.isWarnEnabled())
                        log.warn(_loc.get("oracle-syntax"));
                    joinSyntax = SYNTAX_DATABASE;
                    dateTypeName = "DATE"; // added oracle 9
                    timestampTypeName = "DATE"; // added oracle 9
                    supportsXMLColumn = false;
                }
                // select of an xml column requires ".getStringVal()" (for values <= 4000 bytes only)
                // or ".getClobVal()" suffix. eg. t0.xmlcol.getClobVal()
                getStringVal = ".getClobVal()";
            } else if (metadataClassName.startsWith("com.ddtek.")
                || url.indexOf("jdbc:datadirect:oracle:") != -1
                || "Oracle".equals(driverName)) {
                driverVendor = VENDOR_DATADIRECT + meta.getDriverMajorVersion()
                    + meta.getDriverMinorVersion();
            } else
                driverVendor = VENDOR_OTHER;
        }
        cacheDriverBehavior(driverVendor);
        guessJDBCVersion(conn);
    }

    /**
     * Cache constant for drivers with behaviors we have to deal with.
     */
    private void cacheDriverBehavior(String driverVendor) {
        if (_driverBehavior != -1)
            return;

        driverVendor = driverVendor.toLowerCase(Locale.ENGLISH);
        if (driverVendor.startsWith(VENDOR_ORACLE))
            _driverBehavior = BEHAVE_ORACLE;
        else if (driverVendor.equals(VENDOR_DATADIRECT + "30")
            || driverVendor.equals(VENDOR_DATADIRECT + "31"))
            _driverBehavior = BEHAVE_DATADIRECT31;
        else
            _driverBehavior = BEHAVE_OTHER;
    }

    /**
     * Ensure that the driver vendor has been set, and if not, set it now.
     */
    public void ensureDriverVendor() {
        if (driverVendor != null) {
            cacheDriverBehavior(driverVendor);
            return;
        }

        if (log.isInfoEnabled())
            log.info(_loc.get("oracle-connecting-for-driver"));
        Connection conn = null;
        try {
            conn = conf.getDataSource2(null).getConnection();
            connectedConfiguration(conn);
        } catch (SQLException se) {
            throw SQLExceptions.getStore(se, this);
        } finally {
            if (conn != null)
                try {
                    conn.close();
                } catch (SQLException se) {
                }
        }
    }

    @Override
    public boolean supportsLocking(Select sel) {
        if (!super.supportsLocking(sel))
            return false;
        return !requiresSubselectForRange(sel.getStartIndex(),
            sel.getEndIndex(), sel.isDistinct(), sel.getOrdering());
    }

    /**
     * Return the declaration SQL for the given column. This method is used
     * for each column from within {@link #getCreateTableSQL} and
     * {@link #getAddColumnSQL}.
     *
     * Oracle needs a bit special handling for auto assign columns.
     * For those ("GENERATED ALWAYS AS IDENTITY") we must not generate NOT NULL
     * as this would create invalid statements for Oracle.
     */
    @Override
    protected String getDeclareColumnSQL(Column col, boolean alter) {
        StringBuilder buf = new StringBuilder();
        String columnName = checkNameLength(toDBName(col.getIdentifier()), maxColumnNameLength,
                "long-column-name");
        buf.append(columnName).append(" ");
        buf.append(getTypeName(col));

        // can't add constraints to a column we're adding after table
        // creation, cause some data might already be inserted
        if (!alter
            && !col.isAutoAssigned()) { // this is actually the only 'special' case for oracle
            if (col.getDefaultString() != null && !col.isAutoAssigned())
                buf.append(" DEFAULT ").append(col.getDefaultString());
            if (col.isNotNull() || (!supportsNullUniqueColumn && col.hasConstraint(Unique.class)))
                buf.append(" NOT NULL");
        }
        if (col.isAutoAssigned()) {
            if (!supportsAutoAssign)
                log.warn(_loc.get("invalid-autoassign", platform, col));
            else if (autoAssignClause != null)
                buf.append(" ").append(autoAssignClause);
        }
        return buf.toString();
    }

    @Override
    protected SQLBuffer getSelects(Select sel, boolean distinctIdentifiers,
        boolean forUpdate) {
        // if range doesn't require a subselect can use super
        if (!requiresSubselectForRange(sel.getStartIndex(),
            sel.getEndIndex(), sel.isDistinct(), sel.getOrdering()))
            return super.getSelects(sel, distinctIdentifiers, forUpdate);

        // if there are no joins involved or we're using a from select so
        // that all cols already have unique aliases, can use super
        if (sel.getFromSelect() != null || sel.getTableAliases().size() < 2)
            return super.getSelects(sel, distinctIdentifiers, forUpdate);

        // since none of the conditions above were met, we're dealing with
        // a select that uses joins and requires subselects to select the
        // proper range; alias all column values so that they are unique within
        // the subselect
        SQLBuffer selectSQL = new SQLBuffer(this);
        List aliases;
        if (distinctIdentifiers)
            aliases = sel.getIdentifierAliases();
        else
            aliases = sel.getSelectAliases();

        Object alias;
        int i = 0;
        for (Iterator itr = aliases.iterator(); itr.hasNext(); i++) {
            alias = itr.next();
            String asString = null;
            if (alias instanceof SQLBuffer) {
                asString = ((SQLBuffer) alias).getSQL();
                selectSQL.appendParamOnly((SQLBuffer) alias);
            } else {
                asString = alias.toString();
            }
            selectSQL.append(asString);
            if (asString.indexOf(" AS ") == -1)
                selectSQL.append(" AS c").append(String.valueOf(i));
            if (itr.hasNext())
                selectSQL.append(", ");
        }
        return selectSQL;
    }

    @Override
    public boolean canOuterJoin(int syntax, ForeignKey fk) {
        if (!super.canOuterJoin(syntax, fk))
            return false;
        if (fk != null && syntax == SYNTAX_DATABASE) {
            if (fk.getConstants().length > 0)
                return false;
            if (fk.getPrimaryKeyConstants().length > 0)
                return false;
        }
        return true;
    }

    @Override
    public SQLBuffer toNativeJoin(Join join) {
        if (join.getType() != Join.TYPE_OUTER)
            return toTraditionalJoin(join);

        ForeignKey fk = join.getForeignKey();
        if (fk == null)
            return null;

        boolean inverse = join.isForeignKeyInversed();
        Column[] from = (inverse) ? fk.getPrimaryKeyColumns()
            : fk.getColumns();
        Column[] to = (inverse) ? fk.getColumns()
            : fk.getPrimaryKeyColumns();

        // do column joins
        SQLBuffer buf = new SQLBuffer(this);
        int count = 0;
        for (int i = 0; i < from.length; i++, count++) {
            if (count > 0)
                buf.append(" AND ");
            buf.append(join.getAlias1()).append(".").append(from[i]);
            buf.append(" = ");
            buf.append(join.getAlias2()).append(".").append(to[i]);
            buf.append("(+)");
        }

        // check constant joins
        if (fk.getConstantColumns().length > 0)
            throw new StoreException(_loc.get("oracle-constant",
                join.getTable1(), join.getTable2())).setFatal(true);

        if (fk.getConstantPrimaryKeyColumns().length > 0)
            throw new StoreException(_loc.get("oracle-constant",
                join.getTable1(), join.getTable2())).setFatal(true);
        return buf;
    }

    @Override
    protected SQLBuffer toSelect(SQLBuffer select, JDBCFetchConfiguration fetch,
        SQLBuffer tables, SQLBuffer where, SQLBuffer group,
        SQLBuffer having, SQLBuffer order,
        boolean distinct, boolean forUpdate, long start, long end,
        boolean subselect, Select sel) {
        return toSelect(select, fetch, tables, where, group, having, order,
            distinct, forUpdate, start, end, sel);
    }

    @Override
    protected SQLBuffer toSelect(SQLBuffer select, JDBCFetchConfiguration fetch,
        SQLBuffer tables, SQLBuffer where, SQLBuffer group,
        SQLBuffer having, SQLBuffer order,
        boolean distinct, boolean forUpdate, long start, long end,
        Select sel) {
        if (!_checkedUpdateBug) {
            ensureDriverVendor();
            if (forUpdate && _driverBehavior == BEHAVE_DATADIRECT31)
                log.warn(_loc.get("dd-lock-bug"));
            _checkedUpdateBug = true;
        }

        // if no range, use standard select
        if (!isUsingRange(start, end)) {
            return super.toSelect(select, fetch, tables, where, group, having,
                order, distinct, forUpdate, 0, Long.MAX_VALUE, sel);
        }

        // if no skip, ordering, or distinct can use rownum directly
        SQLBuffer buf = new SQLBuffer(this);
        if (!requiresSubselectForRange(start, end, distinct, order)) {
            if (where != null && !where.isEmpty())
                buf.append(where).append(" AND ");
            buf.append("ROWNUM <= ").appendValue(end);
            return super.toSelect(select, fetch, tables, buf, group, having,
                order, distinct, forUpdate, 0, Long.MAX_VALUE, sel);
        }

        // if there is ordering, skip, or distinct we have to use subselects
        SQLBuffer newsel = super.toSelect(select, fetch, tables, where,
            group, having, order, distinct, forUpdate, 0, Long.MAX_VALUE,
            sel);

        // if no skip, can use single nested subselect
        if (!isUsingOffset(start)) {
            buf.append(getSelectOperation(fetch) + " * FROM (");
            buf.append(newsel);
            buf.append(") WHERE ROWNUM <= ").appendValue(end);
            return buf;
        }

        // with a skip, we have to use a double-nested subselect to put
        // where conditions on the rownum
        buf.append(getSelectOperation(fetch))
           .append(" * FROM (SELECT r.*, ROWNUM RNUM FROM (");
        buf.append(newsel);
        buf.append(") r");
        if (isUsingLimit(end))
            buf.append(" WHERE ROWNUM <= ").appendValue(end);
        buf.append(") WHERE RNUM > ").appendValue(start);
        return buf;
    }

    /**
     * Return true if the select with the given parameters needs a
     * subselect to apply a range.
     */
    private boolean requiresSubselectForRange(long start, long end,
        boolean distinct, SQLBuffer order) {
        if (!isUsingRange(start, end)) {
            return false;
        }
        return isUsingOffset(start) || distinct || isUsingOrderBy(order);
    }

    /**
     * Check to see if we have set the {@link #SELECT_HINT} in the
     * fetch configuration, and if so, append the Oracle hint after the
     * "SELECT" part of the query.
     */
    @Override
    public String getSelectOperation(JDBCFetchConfiguration fetch) {
        Object hint = fetch == null ? null : fetch.getHint(SELECT_HINT);
        String select = "SELECT";
        if (hint != null)
            select += " " + hint;
        return select;
    }

    @Override
    public void setString(PreparedStatement stmnt, int idx, String val,
        Column col)
        throws SQLException {
        // oracle NCHAR/NVARCHAR/NCLOB unicode columns require some
        // special handling to configure them correctly; see:
        // http://www.oracle.com/technology/sample_code/tech/java/
        // sqlj_jdbc/files/9i_jdbc/NCHARsupport4UnicodeSample/Readme.html
        String typeName = (col == null) ? null : col.getTypeIdentifier().getName();
        if (useSetFormOfUseForUnicode && typeName != null &&
            (typeName.toLowerCase(Locale.ENGLISH).startsWith("nvarchar") ||
                typeName.toLowerCase(Locale.ENGLISH).startsWith("nchar") ||
                typeName.toLowerCase(Locale.ENGLISH).startsWith("nclob"))) {
            Statement inner = stmnt;
            if (inner instanceof DelegatingPreparedStatement)
                inner = ((DelegatingPreparedStatement) inner).
                    getInnermostDelegate();
            if (isOraclePreparedStatement(inner)) {
                try {
                    inner.getClass().getMethod("setFormOfUse",
                        new Class[]{ int.class, short.class }).
                        invoke(inner,
                            new Object[]{
                                    idx,
                                oraclePreparedStatementFormNvarcharField.get(null)
                            });
                } catch (Exception e) {
                    log.warn(e);
                }
            } else if (!_warnedNcharColumn && log.isWarnEnabled()) {
                _warnedNcharColumn = true;
                log.warn(_loc.get("unconfigured-nchar-cols"));
            }
        }

        // call setFixedCHAR for fixed width character columns to get padding
        // semantics
        if (col != null && col.getType() == Types.CHAR
            && val != null && val.length() != col.getSize()) {
            Statement inner = stmnt;
            if (inner instanceof DelegatingPreparedStatement)
                inner = ((DelegatingPreparedStatement) inner).
                    getInnermostDelegate();
            if (isOraclePreparedStatement(inner)) {
                try {
                    Method setFixedCharMethod = inner.getClass().getMethod("setFixedCHAR",
                                                         new Class[]{int.class, String.class});
                    if (!setFixedCharMethod.isAccessible()) {
                        setFixedCharMethod.setAccessible(true);
                    }

                    setFixedCharMethod.invoke(inner, new Object[]{idx, val });
                    return;
                } catch (Exception e) {
                    log.warn(e);
                }
            }

            if (!_warnedCharColumn && log.isWarnEnabled()) {
                _warnedCharColumn = true;
                log.warn(_loc.get("unpadded-char-cols"));
            }
        }
        super.setString(stmnt, idx, val, col);
    }

    @Override
    public void setBinaryStream(PreparedStatement stmnt, int idx,
        InputStream val, int length, Column col)
        throws SQLException {
        if (length == 0)
            stmnt.setBlob(idx, getEmptyBlob());
        else {
            super.setBinaryStream(stmnt, idx, val, length, col);
        }
    }

    @Override
    public void setClobString(PreparedStatement stmnt, int idx, String val,
        Column col)
        throws SQLException {

        //We need a place to detect if the user is setting the 'supportsSetClob' property.
        //While in previous releases this property had meaning, it is no longer useful
        //given the code added via OPENJPA-1691.  As such, we need to warn user's the
        //property no longer has meaning.  While it would be nice to have a better way
        //to detect if the supportsSetClob property has been set, the best we can do
        //is detect the variable in this code path as this is the path a user's code
        //would go down if they are still executing code which actually made use of
        //the support provided via setting supportsSetClob.
        if (supportsSetClob && logSupportsSetClobWarning){
            log.warn(_loc.get("oracle-set-clob-warning"));
            logSupportsSetClobWarning=false;
        }

        if (col.isXML()) {
            if (isJDBC4) {
                // This JDBC 4 method handles values longer than 4000 bytes.
                stmnt.setClob(idx, new StringReader(val), val.length());
            } else {
                // This method is limited to 4000 bytes.
                setCharacterStream(stmnt, idx, new StringReader(val), val.length(), col);
            }
            return;
        }
        if (!useSetStringForClobs && val.length() == 0)
            stmnt.setClob(idx, getEmptyClob());
        else {
            super.setClobString(stmnt, idx, val, col);
        }
    }

    @Override
    public void setNull(PreparedStatement stmnt, int idx, int colType,
        Column col)
        throws SQLException {
        if ((colType == Types.CLOB || colType == Types.BLOB) && col.isNotNull())
            throw new UserException(_loc.get("null-blob-in-not-nullable", toDBName(col
                .getFullDBIdentifier())));
        if (colType == Types.BLOB && _driverBehavior == BEHAVE_ORACLE)
            stmnt.setBlob(idx, getEmptyBlob());
        else if (colType == Types.CLOB && _driverBehavior == BEHAVE_ORACLE
            && !col.isXML())
            stmnt.setClob(idx, getEmptyClob());
        else if ((colType == Types.STRUCT || colType == Types.OTHER)
            && col != null && !DBIdentifier.isNull(col.getTypeIdentifier()))
            stmnt.setNull(idx, Types.STRUCT, col.getTypeIdentifier().getName());
            // some versions of the Oracle JDBC driver will fail if calling
            // setNull with DATE; see bug #1171
        else if (colType == Types.DATE)
            super.setNull(stmnt, idx, Types.TIMESTAMP, col);
        // the Oracle driver does not support Types.OTHER with setNull
        else if (colType == Types.OTHER || col.isXML())
            super.setNull(stmnt, idx, Types.NULL, col);
        else
            super.setNull(stmnt, idx, colType, col);
    }

    @Override
    public void setLocalDate(PreparedStatement stmnt, int idx, LocalDate val, Column col) throws SQLException {
        stmnt.setObject(idx, val);
    }

    @Override
    public LocalDate getLocalDate(ResultSet rs, int column) throws SQLException {
        return rs.getObject(column, LocalDate.class);
    }

    @Override
    public void setLocalTime(PreparedStatement stmnt, int idx, LocalTime val, Column col) throws SQLException {
        stmnt.setObject(idx, val);
    }

    @Override
    public LocalTime getLocalTime(ResultSet rs, int column) throws SQLException {
        return rs.getObject(column, LocalTime.class);
    }

    @Override
    public void setLocalDateTime(PreparedStatement stmnt, int idx, LocalDateTime val, Column col) throws SQLException {
        stmnt.setObject(idx, val);
    }

    @Override
    public LocalDateTime getLocalDateTime(ResultSet rs, int column) throws SQLException {
        return rs.getObject(column, LocalDateTime.class);
    }

    @Override
    public void setOffsetDateTime(PreparedStatement stmnt, int idx, OffsetDateTime val, Column col) throws SQLException {
        stmnt.setObject(idx, val);
    }

