/**
 * Copyright 2015 Palantir Technologies
 *
 * Licensed under the BSD-3 License (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://opensource.org/licenses/BSD-3-Clause
 *
 * 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 com.palantir.atlasdb.keyvalue.dbkvs.impl;

import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Ordering;
import com.palantir.atlasdb.keyvalue.api.Cell;
import com.palantir.atlasdb.keyvalue.api.KeyAlreadyExistsException;
import com.palantir.atlasdb.keyvalue.api.TableReference;
import com.palantir.atlasdb.keyvalue.api.Value;
import com.palantir.atlasdb.keyvalue.dbkvs.DdlConfig;
import com.palantir.exception.PalantirSqlException;
import com.palantir.nexus.db.sql.ExceptionCheck;

public abstract class AbstractDbWriteTable implements DbWriteTable {
    protected final DdlConfig config;
    protected final ConnectionSupplier conns;
    protected final TableReference tableRef;
    private final PrefixedTableNames prefixedTableNames;

    public AbstractDbWriteTable(
            DdlConfig config,
            ConnectionSupplier conns,
            TableReference tableRef,
            PrefixedTableNames prefixedTableNames) {
        this.config = config;
        this.conns = conns;
        this.tableRef = tableRef;
        this.prefixedTableNames = prefixedTableNames;
    }

    @Override
    public void put(Collection<Map.Entry<Cell, byte[]>> data, long ts) {
        List<Object[]> args = Lists.newArrayListWithCapacity(data.size());
        for (Entry<Cell, byte[]> entry : data) {
            Cell cell = entry.getKey();
            byte[] val = entry.getValue();
            args.add(new Object[] { cell.getRowName(), cell.getColumnName(), ts, val });
        }
        put(args);
    }

    @Override
    public void put(Collection<Map.Entry<Cell, Value>> data) {
        List<Object[]> args = Lists.newArrayListWithCapacity(data.size());
        for (Entry<Cell, Value> entry : data) {
            Cell cell = entry.getKey();
            Value val = entry.getValue();
            args.add(new Object[] { cell.getRowName(), cell.getColumnName(), val.getTimestamp(), val.getContents() });
        }
        put(args);
    }

    private void put(List<Object[]> args) {
        try {
            String prefixedTableName = prefixedTableNames.get(tableRef);
            conns.get().insertManyUnregisteredQuery("/* INSERT_ONE (" + prefixedTableName + ") */"
                    + " INSERT INTO " + prefixedTableName + " (row_name, col_name, ts, val) "
                    + " VALUES (?, ?, ?, ?) ",
                    args);
        } catch (PalantirSqlException e) {
            if (ExceptionCheck.isUniqueConstraintViolation(e)) {
                throw new KeyAlreadyExistsException("primary key violation", e);
            }
            throw e;
        }
    }

    @Override
    public void putSentinels(Iterable<Cell> cells) {
        byte[] value = new byte[0];
        long ts = Value.INVALID_VALUE_TIMESTAMP;
        for (List<Cell> batch : Iterables.partition(Ordering.natural().immutableSortedCopy(cells), 1000)) {
            List<Object[]> args = Lists.newArrayListWithCapacity(batch.size());
            for (Cell cell : batch) {
                args.add(new Object[] {cell.getRowName(), cell.getColumnName(), ts, value,
                        cell.getRowName(), cell.getColumnName(), ts});
            }
            while (true) {
                try {
                    String prefixedTableName = prefixedTableNames.get(tableRef);
                    conns.get().insertManyUnregisteredQuery("/* INSERT_WHERE_NOT_EXISTS (" + prefixedTableName + ") */"
                            + " INSERT INTO " + prefixedTableName + " (row_name, col_name, ts, val) "
                            + " SELECT ?, ?, ?, ? FROM DUAL"
                            + " WHERE NOT EXISTS (SELECT * FROM " + prefixedTableName + " WHERE"
                            + " row_name = ? AND"
                            + " col_name = ? AND"
                            + " ts = ?)",
                            args);
                    break;
                } catch (PalantirSqlException e) {
                    // we can't do atomic put if not exists, so retry if we get constraint violations
                    // TODO: Actually you can. Evaluate use of MERGE or UPSERT here.
                    if (!ExceptionCheck.isUniqueConstraintViolation(e)) {
                        throw e;
                    }
                }
            }
        }
    }

    @Override
    public void update(Cell cell, long ts, byte[] oldValue, byte[] newValue) {
        new UpdateExecutor(conns, tableRef, prefixedTableNames).update(cell, ts, oldValue, newValue);
    }

    @Override
    public void delete(List<Entry<Cell, Long>> entries) {
        List<Object[]> args = Lists.newArrayListWithCapacity(entries.size());
        for (Map.Entry<Cell, Long> entry : entries) {
            Cell cell = entry.getKey();
            args.add(new Object[] {cell.getRowName(), cell.getColumnName(), entry.getValue()});
        }

        String prefixedTableName = prefixedTableNames.get(tableRef);
        conns.get().updateManyUnregisteredQuery(" /* DELETE_ONE (" + prefixedTableName + ") */ "
                + " DELETE /*+ INDEX(m pk_" + prefixedTableName + ") */ "
                + " FROM " + prefixedTableName + " m "
                + " WHERE m.row_name = ? "
                + "  AND m.col_name = ? "
                + "  AND m.ts = ?",
                args);
    }
}

@@ -1137,16 +1137,142 @@
 Config;%0A
+import com.palantir.atlasdb.keyvalue.dbkvs.OracleDdlConfig;%0Aimport com.palantir.atlasdb.keyvalue.dbkvs.OracleTableNameMapper;%0A
 import c
@@ -5906,15 +5906,40 @@
 X(m 
-pk_
 %22 + 
+getPrimaryKeyConstraintName(
 pref
@@ -5943,32 +5943,33 @@
 refixedTableName
+)
  + %22) */ %22%0A     
@@ -6162,16 +6162,534 @@
  args);%0A
-
     %7D%0A
+%0A    private String getPrimaryKeyConstraintName(String tableName) %7B%0A        String pkConstraintName = %22pk_%22 + tableName;%0A        if (config.type().equals(OracleDdlConfig.TYPE)) %7B%0A            return truncateToMaxOracleLength(pkConstraintName);%0A        %7D%0A        return pkConstraintName;%0A    %7D%0A%0A    private String truncateToMaxOracleLength(String constraintName) %7B%0A        return constraintName%0A                .substring(0, Math.max(OracleTableNameMapper.ORACLE_MAX_TABLE_NAME_LENGTH, constraintName.length()));%0A    %7D%0A%0A
 %7D%0A

Fix AbstractDbWriteTable