/*
 * Copyright (c) 2012 - 2020 Splice Machine, Inc.
 *
 * This file is part of Splice Machine.
 * Splice Machine is free software: you can redistribute it and/or modify it under the terms of the
 * GNU Affero General Public License as published by the Free Software Foundation, either
 * version 3, or (at your option) any later version.
 * Splice Machine is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
 * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
 * See the GNU Affero General Public License for more details.
 * You should have received a copy of the GNU Affero General Public License along with Splice Machine.
 * If not, see <http://www.gnu.org/licenses/>.
 */

package com.splicemachine.derby.impl.load;

import com.splicemachine.derby.test.TPCHIT;
import com.splicemachine.derby.test.framework.*;
import com.splicemachine.homeless.TestUtils;
import com.splicemachine.test_tools.TableCreator;
import org.junit.*;
import org.junit.rules.RuleChain;
import org.junit.rules.TestRule;
import splice.com.google.common.collect.Lists;
import splice.com.google.common.primitives.Ints;

import java.io.File;
import java.sql.*;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;

/**
 * Correctness tests around the "UPSERT_FROM_FILE" or "UPPORT" functionality.
 *
 * @author Scott Fines
 *         Date: 10/20/14
 */
public class UpportIT extends SpliceUnitTest {
    public static final String CLASS_NAME = UpportIT.class.getSimpleName().toUpperCase();
    private static final SpliceSchemaWatcher schema =
            new SpliceSchemaWatcher(CLASS_NAME);
    private static final SpliceTableWatcher nullableBTable =
            new SpliceTableWatcher("empty_table",schema.schemaName,"(a int, b int, primary key(a))");
    private static final SpliceTableWatcher occupiedTable =
            new SpliceTableWatcher("occ_table",schema.schemaName,"(a int, b int not null, primary key(a))");
    private static final SpliceTableWatcher no_pk  =
            new SpliceTableWatcher("no_pk",schema.schemaName,"(a int, b int)");
    private static final SpliceTableWatcher tableWithAutoCol =
            new SpliceTableWatcher("table_with_auto_column",schema.schemaName,
                    "(a int, b int, c int generated by default as identity (start with 1, increment by 1), d int default 999, primary key(a))");
    private static final SpliceTableWatcher tableWithDefaultCol =
            new SpliceTableWatcher("table_with_default_column",schema.schemaName,
                    "(a int, b int, c int generated by default as identity (start with 1, increment by 1), d int default 999, primary key(a))");
    private static final SpliceTableWatcher tableForUpdateInDirectory =
            new SpliceTableWatcher("table_for_update_in_directory",schema.schemaName,
                    "(a int, b int, c int generated by default as identity (start with 1, increment by 1), d int default 999, primary key(a))");

    private static TestConnection conn;
    private static File BADDIR;
    private static File IMPORTDIR;

    protected static SpliceWatcher spliceClassWatcher = new SpliceWatcher(CLASS_NAME);
    protected static SpliceSchemaWatcher spliceSchemaWatcher = new SpliceSchemaWatcher(CLASS_NAME);

    @Rule
    public SpliceWatcher methodWatcher = new SpliceWatcher(UpportIT.class.getSimpleName().toUpperCase());

    @ClassRule
    public static TestRule chain = RuleChain.outerRule(schema)
            .around(spliceSchemaWatcher)
            .around(nullableBTable)
            .around(occupiedTable)
            .around(no_pk)
            .around(tableWithAutoCol)
            .around(tableWithDefaultCol)
            .around(tableForUpdateInDirectory);
    private static int size;
    private static SpliceUnitTest.TestFileGenerator fullTestFile;
    private static List<int[]> correctFullData;
    private static SpliceUnitTest.TestFileGenerator partialTestFile;
    private static List<int[]> correctPartialData;
    private static SpliceUnitTest.TestFileGenerator fullTestFileWithDuplicates;
    private static SpliceUnitTest.TestFileGenerator fullTestFileWithEmptyLine;


    @BeforeClass
    public static void setUpClass() throws Exception {

        BADDIR = createBadLogDirectory(schema.schemaName);
        assertNotNull(BADDIR);
        IMPORTDIR = createImportFileDirectory(schema.schemaName);
        assertNotNull(IMPORTDIR);

        size = 5;
        correctFullData = Lists.newArrayListWithExpectedSize(size);
        fullTestFile = generateFullRow(IMPORTDIR, "full", size, correctFullData, false);

        List<int[]> correctFullDataWithDuplicates = Lists.newArrayListWithExpectedSize(size); //thrown away
        fullTestFileWithDuplicates = generateFullRow(IMPORTDIR, "fullWithDuplicates", size, correctFullDataWithDuplicates, true);

        List<int[]> correctFullDataWithEmptyLine = Lists.newArrayListWithExpectedSize(size); //thrown away
        fullTestFileWithEmptyLine = generateFullRow(IMPORTDIR, "fullWithEmptyLine", size, correctFullDataWithEmptyLine, false, true);

        correctPartialData = Lists.newArrayListWithExpectedSize(size);
        partialTestFile = generatePartialRow(IMPORTDIR, "partial", size, correctPartialData);
        Connection conn = spliceClassWatcher.getOrCreateConnection();
        new TableCreator(conn)
                .withCreate("CREATE TABLE LINEITEM (\n" +
                        "  L_ORDERKEY      INTEGER NOT NULL,\n" +
                        "  L_PARTKEY       INTEGER NOT NULL,\n" +
                        "  L_SUPPKEY       INTEGER NOT NULL,\n" +
                        "  L_LINENUMBER    INTEGER NOT NULL,\n" +
                        "  L_QUANTITY      DECIMAL(15, 2),\n" +
                        "  L_EXTENDEDPRICE DECIMAL(15, 2),\n" +
                        "  L_DISCOUNT      DECIMAL(15, 2),\n" +
                        "  L_TAX           DECIMAL(15, 2),\n" +
                        "  L_RETURNFLAG    CHAR(1),\n" +
                        "  L_LINESTATUS    CHAR(1),\n" +
                        "  L_SHIPDATE      DATE,\n" +
                        "  L_COMMITDATE    DATE,\n" +
                        "  L_RECEIPTDATE   DATE,\n" +
                        "  L_SHIPINSTRUCT  CHAR(25),\n" +
                        "  L_SHIPMODE      CHAR(10),\n" +
                        "  L_COMMENT       VARCHAR(44),\n" +
                        " PRIMARY KEY (L_ORDERKEY, L_LINENUMBER)\n" +
                        ")")
                .withIndex("create index L_SHIPDATE_IDX on LINEITEM(\n" +
                        " L_SHIPDATE,\n" +
                        " L_PARTKEY,\n" +
                        " L_EXTENDEDPRICE,\n" +
                        " L_DISCOUNT\n" +
                        " )")
                .withIndex("create index L_PART_IDX on LINEITEM(\n" +
                        " L_PARTKEY,\n" +
                        " L_ORDERKEY,\n" +
                        " L_SUPPKEY,\n" +
                        " L_SHIPDATE,\n" +
                        " L_EXTENDEDPRICE,\n" +
                        " L_DISCOUNT,\n" +
                        " L_QUANTITY,\n" +
                        " L_SHIPMODE,\n" +
                        " L_SHIPINSTRUCT\n" +
                        " )")
                .create();
    }

    @Before
    public void before() throws Exception {
        conn = createConnection();
        conn.setAutoCommit(false);
    }

    @After
    public void tearDown() throws Exception {
        try {
            conn.rollback();
        } catch (SQLException e) {
            // ignore
        }
        try {
            conn.close();
        } catch (SQLException e) {
            // ignore
        }
    }

    private static final Comparator<int[]> intArrayComparator = new Comparator<int[]>() {
        @Override public int compare(int[] o1, int[] o2) { return Ints.compare(o1[0], o2[0]); }
    };

    @Test
    public void testUpsertFailsWithMissingNonNullColumn() throws Exception {
        CallableStatement statement =
                conn.prepareCall("call SYSCS_UTIL.UPSERT_DATA_FROM_FILE(?,?,?,?,null,null,null,null,null,0,?,null,null)");
        statement.setString(1,schema.schemaName);
        statement.setString(2,occupiedTable.tableName);
        statement.setString(3,"a");
        statement.setString(4,fullTestFile.getFilePath());
        statement.setString(5,BADDIR.getCanonicalPath());

        try{
            statement.execute();
            Assert.fail("Did not throw exception");
        }catch(SQLException se){
            SpliceUnitTest.assertBadFileContainsError(BADDIR, fullTestFile.getFileName(), "23502", null);
        }
    }

    @Test
    public void testUpsertFailsWithMissingPk() throws Exception {
        CallableStatement statement =
                conn.prepareCall("call SYSCS_UTIL.UPSERT_DATA_FROM_FILE(?,?,?,?,null,null,null,null,null,0,?,null,null)");
        statement.setString(1,schema.schemaName);
        statement.setString(2, nullableBTable.tableName);
        statement.setString(3,"b");
        statement.setString(4,fullTestFile.getFilePath());
        statement.setString(5,BADDIR.getCanonicalPath());

        try{
            statement.execute();
            Assert.fail("Did not throw exception");
        }catch(SQLException se){
            SpliceUnitTest.assertBadFileContainsError(BADDIR, fullTestFile.getFileName(), "23502", null);
        }
    }

    @Test
    public void testUpsertFailsWithNoPk() throws Exception {
        CallableStatement statement =
                conn.prepareCall("call SYSCS_UTIL.UPSERT_DATA_FROM_FILE(?,?,null,?,null,null,null,null,null,0,?,null,null)");
        statement.setString(1,schema.schemaName);
        statement.setString(2,no_pk.tableName);
        statement.setString(3,fullTestFile.getFilePath());
        statement.setString(4,BADDIR.getCanonicalPath());

        try{
            statement.execute();
            Assert.fail("Did not throw exception");
        }catch(SQLException se){
            Assert.assertEquals("Incorrect SQL State!", "SE018", se.getSQLState());
        }
    }

    @Test
    public void testUpsertIntoOccupiedTableWillUpdate() throws Exception {
        PreparedStatement ps = conn.prepareStatement("insert into " + occupiedTable + "(a,b) values (?,?)");
        List<int[]> newCorrect = Lists.newArrayList(correctFullData);
        for(int[] correctRow:newCorrect){
            //add a row that's different
            ps.setInt(1,correctRow[0]); ps.setInt(2,correctRow[1]/2); ps.executeUpdate();
        }
        //add one extra row so that we know that we don't overwrite the entire table
        ps.setInt(1,size+1); ps.setInt(2,size+1); ps.executeUpdate();
        newCorrect.add(new int[]{size + 1, size + 1});

        CallableStatement statement =
                conn.prepareCall("call SYSCS_UTIL.UPSERT_DATA_FROM_FILE(?,?,null,?,null,null,null,null,null,0,?,null,null)");
        statement.setString(1,schema.schemaName);
        statement.setString(2,occupiedTable.tableName);
        statement.setString(3,fullTestFile.getFilePath());
        statement.setString(4,BADDIR.getCanonicalPath());

        ResultSet resultSet = statement.executeQuery();
        //make sure that the bad records list is good
        validateImportResults(resultSet,size,0);

        //make sure that the data matches
        List<int[]> actualData = Lists.newArrayListWithExpectedSize(size);
        ResultSet rs = conn.query("select * from "+ occupiedTable);
        while(rs.next()){
            int[] data = new int[2];
            data[0] = rs.getInt(1);
            Assert.assertFalse("Column a was null!",rs.wasNull());
            data[1] = rs.getInt(2);
            Assert.assertFalse("Column b was null!",rs.wasNull());
            actualData.add(data);
        }

        Collections.sort(newCorrect, intArrayComparator);
        Collections.sort(actualData, intArrayComparator);
        assertCorrectResult(newCorrect, actualData);
    }

    @Test
    public void testUpsertIntoOccupiedPartiallyTableWillUpdate() throws Exception {
        PreparedStatement ps = conn.prepareStatement("insert into " + nullableBTable + "(a,b) values (?,?)");
        List<int[]> newCorrect = Lists.newArrayListWithExpectedSize(size);
        for(int[] correctRow:correctFullData){
            //add a row that's different
            ps.setInt(1,correctRow[0]); ps.setInt(2,correctRow[1]/2); ps.executeUpdate();
            newCorrect.add(new int[]{correctRow[0],correctRow[1]/2});
        }
        //add one extra row so that we know that we don't overwrite the entire table
        ps.setInt(1,size+1); ps.setInt(2,size+1); ps.executeUpdate();
        newCorrect.add(new int[]{size + 1, size + 1});

        CallableStatement statement =
                conn.prepareCall("call SYSCS_UTIL.UPSERT_DATA_FROM_FILE(?,?,?,?,null,null,null,null,null,0,?,null,null)");
        statement.setString(1,schema.schemaName);
        statement.setString(2,nullableBTable.tableName);
        statement.setString(3,"a");
        statement.setString(4,partialTestFile.getFilePath());
        statement.setString(5,BADDIR.getCanonicalPath());

        ResultSet resultSet = statement.executeQuery();
        //make sure that the bad records list is good
        validateImportResults(resultSet,size,0);

        //make sure that the data matches
        List<int[]> actualData = Lists.newArrayListWithExpectedSize(size);
        ResultSet rs = conn.query("select * from "+ nullableBTable);
        while(rs.next()){
            int[] data = new int[2];
            data[0] = rs.getInt(1);
            Assert.assertFalse("Column a was null!",rs.wasNull());
            data[1] = rs.getInt(2);
            Assert.assertFalse("Column b was null!",rs.wasNull());
            actualData.add(data);
        }

        Collections.sort(newCorrect, intArrayComparator);
        Collections.sort(actualData, intArrayComparator);
        assertCorrectResult(newCorrect, actualData);
    }

    @Test
    public void testUpsertWithPartialEmptyTableWillInsert() throws Exception {
        CallableStatement statement =
                conn.prepareCall("call SYSCS_UTIL.UPSERT_DATA_FROM_FILE(?,?,?,?,null,null,null,null,null,0,?,null,null)");
        statement.setString(1,schema.schemaName);
        statement.setString(2,nullableBTable.tableName);
        statement.setString(3,"a");
        statement.setString(4,partialTestFile.getFilePath());
        statement.setString(5,BADDIR.getCanonicalPath());

        ResultSet resultSet = statement.executeQuery();
        //make sure that the bad records list is good
        validateImportResults(resultSet,size,0);

        //make sure that the data matches
        List<int[]> actualData = Lists.newArrayListWithExpectedSize(size);
        ResultSet rs = conn.query("select * from "+ nullableBTable);
        while(rs.next()){
            int[] data = new int[2];
            data[0] = rs.getInt(1);
            Assert.assertFalse("Column a was null!",rs.wasNull());
            data[1] = rs.getInt(2);
            Assert.assertTrue("Column b was null!", rs.wasNull());
            actualData.add(data);
        }

        Collections.sort(correctPartialData, intArrayComparator);
        Collections.sort(actualData, intArrayComparator);
        assertCorrectResult(correctPartialData,actualData);
    }

    @Test
    public void testUpsertWithEmptyTableWillInsert() throws Exception {
        CallableStatement statement =
                conn.prepareCall("call SYSCS_UTIL.UPSERT_DATA_FROM_FILE(?,?,null,?,null,null,null,null,null,0,?,null,null)");
        statement.setString(1,schema.schemaName);
        statement.setString(2,nullableBTable.tableName);
        statement.setString(3,fullTestFile.getFilePath());
        statement.setString(4,BADDIR.getCanonicalPath());

        ResultSet resultSet = statement.executeQuery();
        //make sure that the bad records list is good
        validateImportResults(resultSet,size,0);

        //make sure that the data matches
        List<int[]> actualData = Lists.newArrayListWithExpectedSize(size);
        ResultSet rs = conn.query("select * from "+ nullableBTable);
        while(rs.next()){
            int[] data = new int[2];
            data[0] = rs.getInt(1);
            Assert.assertFalse("Column a was null!",rs.wasNull());
            data[1] = rs.getInt(2);
            Assert.assertFalse("Column b was null!",rs.wasNull());
            actualData.add(data);
        }

        Collections.sort(correctFullData, intArrayComparator);
        Collections.sort(actualData, intArrayComparator);
        assertCorrectResult(correctFullData,actualData);
    }

    @Test @Ignore("SPLICE-231")
    public void testUpsertWithEmptyTableWillInsertDuplicatesReportedAsBad() throws Exception {
        //  FIXME: JC - 'Additive WriteConflict'. Terminates connection, later tests fail, no good error message.
        CallableStatement statement =
                conn.prepareCall("call SYSCS_UTIL.UPSERT_DATA_FROM_FILE(?,?,null,?,null,null,null,null,null,0,?,null,null)");
        statement.setString(1,schema.schemaName);
        statement.setString(2,nullableBTable.tableName);
        statement.setString(3,fullTestFileWithDuplicates.getFilePath());
        statement.setString(4,BADDIR.getCanonicalPath());

        ResultSet resultSet = statement.executeQuery();
        //make sure that the bad records list is good
        validateImportResults(resultSet,size,1);

        //make sure that the data matches
        List<int[]> actualData = Lists.newArrayListWithExpectedSize(size);
        ResultSet rs = conn.query("select * from "+ nullableBTable);
        while(rs.next()){
            int[] data = new int[2];
            data[0] = rs.getInt(1);
            Assert.assertFalse("Column a was null!",rs.wasNull());
            data[1] = rs.getInt(2);
            Assert.assertFalse("Column b was null!",rs.wasNull());
            actualData.add(data);
        }

        Collections.sort(correctFullData, intArrayComparator);
        Collections.sort(actualData, intArrayComparator);
        assertCorrectResult(correctFullData,actualData);
    }

    @Test
    public void testMergeFailsWithMissingNonNullColumn() throws Exception {
        CallableStatement statement =
                conn.prepareCall("call SYSCS_UTIL.MERGE_DATA_FROM_FILE(?,?,?,?,null,null,null,null,null,0,?,null,null)");
        statement.setString(1,schema.schemaName);
        statement.setString(2,occupiedTable.tableName);
        statement.setString(3,"a");
        statement.setString(4,fullTestFile.getFilePath());
        statement.setString(5,BADDIR.getCanonicalPath());

        try{
            statement.execute();
            Assert.fail("Did not throw exception");
        }catch(SQLException se){
            SpliceUnitTest.assertBadFileContainsError(BADDIR, fullTestFile.getFileName(), "23502", null);
        }
    }

    @Test
    public void testMergeFailsWithMissingPk() throws Exception {
        CallableStatement statement =
                conn.prepareCall("call SYSCS_UTIL.MERGE_DATA_FROM_FILE(?,?,?,?,null,null,null,null,null,0,?,null,null)");
        statement.setString(1,schema.schemaName);
        statement.setString(2, nullableBTable.tableName);
        statement.setString(3,"b");
        statement.setString(4,fullTestFile.getFilePath());
        statement.setString(5,BADDIR.getCanonicalPath());

        try{
            statement.execute();
            Assert.fail("Did not throw exception");
        }catch(SQLException se){
            Assert.assertEquals("Incorrect SQL State!", "SE019", se.getSQLState());
        }
    }

    @Test
    public void testMergeFailsWithNoPk() throws Exception {
        CallableStatement statement =
                conn.prepareCall("call SYSCS_UTIL.MERGE_DATA_FROM_FILE(?,?,null,?,null,null,null,null,null,0,?,null,null)");
        statement.setString(1,schema.schemaName);
        statement.setString(2,no_pk.tableName);
        statement.setString(3,fullTestFile.getFilePath());
        statement.setString(4,BADDIR.getCanonicalPath());

        try{
            statement.execute();
            Assert.fail("Did not throw exception");
        }catch(SQLException se){
            Assert.assertEquals("Incorrect SQL State!", "SE018", se.getSQLState());
        }
    }

    @Test
    public void testMergeIntoOccupiedTableWillUpdate() throws Exception {
        PreparedStatement ps = conn.prepareStatement("insert into " + occupiedTable + "(a,b) values (?,?)");
        List<int[]> newCorrect = Lists.newArrayList(correctFullData);
        for(int[] correctRow:newCorrect){
            //add a row that's different
            ps.setInt(1,correctRow[0]); ps.setInt(2,correctRow[1]/2); ps.executeUpdate();
        }
        //add one extra row so that we know that we don't overwrite the entire table
        ps.setInt(1,size+1); ps.setInt(2,size+1); ps.executeUpdate();
        newCorrect.add(new int[]{size + 1, size + 1});

        CallableStatement statement =
                conn.prepareCall("call SYSCS_UTIL.MERGE_DATA_FROM_FILE(?,?,null,?,null,null,null,null,null,0,?,null,null)");
        statement.setString(1,schema.schemaName);
        statement.setString(2,occupiedTable.tableName);
        statement.setString(3,fullTestFile.getFilePath());
        statement.setString(4,BADDIR.getCanonicalPath());

        ResultSet resultSet = statement.executeQuery();
        //make sure that the bad records list is good
        //all 5 rows are updated, no extra row inserted
        validateMergeResults(resultSet,5,5,0, 0);

        //make sure that the data matches
        List<int[]> actualData = Lists.newArrayListWithExpectedSize(size);
        ResultSet rs = conn.query("select * from "+ occupiedTable);
        while(rs.next()){
            int[] data = new int[2];
            data[0] = rs.getInt(1);
            Assert.assertFalse("Column a was null!",rs.wasNull());
            data[1] = rs.getInt(2);
            Assert.assertFalse("Column b was null!",rs.wasNull());
            actualData.add(data);
        }

        Collections.sort(newCorrect, intArrayComparator);
        Collections.sort(actualData, intArrayComparator);
        assertCorrectResult(newCorrect, actualData);
    }

    @Test
    public void testMergeIntoOccupiedPartiallyTableWillUpdate() throws Exception {
        PreparedStatement ps = conn.prepareStatement("insert into " + nullableBTable + "(a,b) values (?,?)");
        List<int[]> newCorrect = Lists.newArrayListWithExpectedSize(size);
        for(int[] correctRow:correctFullData){
            //add a row that's different
            ps.setInt(1,correctRow[0]); ps.setInt(2,correctRow[1]/2); ps.executeUpdate();
            newCorrect.add(new int[]{correctRow[0],correctRow[1]/2});
        }
        //add one extra row so that we know that we don't overwrite the entire table
        ps.setInt(1,size+1); ps.setInt(2,size+1); ps.executeUpdate();
        newCorrect.add(new int[]{size + 1, size + 1});

        CallableStatement statement =
                conn.prepareCall("call SYSCS_UTIL.MERGE_DATA_FROM_FILE(?,?,?,?,null,null,null,null,null,0,?,null,null)");
        statement.setString(1,schema.schemaName);
        statement.setString(2,nullableBTable.tableName);
        statement.setString(3,"a");
        statement.setString(4,partialTestFile.getFilePath());
        statement.setString(5,BADDIR.getCanonicalPath());

        ResultSet resultSet = statement.executeQuery();
        //make sure that the bad records list is good
        //no row should be updated, as we only use the pk column for the merge_data_from_file call
        validateMergeResults(resultSet,0,0,0, 0);

        //make sure that the data matches
        List<int[]> actualData = Lists.newArrayListWithExpectedSize(size);
        ResultSet rs = conn.query("select * from "+ nullableBTable);
        while(rs.next()){
            int[] data = new int[2];
            data[0] = rs.getInt(1);
            Assert.assertFalse("Column a was null!",rs.wasNull());
            data[1] = rs.getInt(2);
            Assert.assertFalse("Column b was null!",rs.wasNull());
            actualData.add(data);
        }

        Collections.sort(newCorrect, intArrayComparator);
        Collections.sort(actualData, intArrayComparator);
        assertCorrectResult(newCorrect, actualData);
    }

    @Test
    public void testMergeWithPartialEmptyTableWillInsert() throws Exception {
        CallableStatement statement =
                conn.prepareCall("call SYSCS_UTIL.MERGE_DATA_FROM_FILE(?,?,?,?,null,null,null,null,null,0,?,null,null)");
        statement.setString(1,schema.schemaName);
        statement.setString(2,nullableBTable.tableName);
        statement.setString(3,"a");
        statement.setString(4,partialTestFile.getFilePath());
        statement.setString(5,BADDIR.getCanonicalPath());

        ResultSet resultSet = statement.executeQuery();
        //make sure that the bad records list is good
        validateMergeResults(resultSet,size, 0,size,0);

        //make sure that the data matches
        List<int[]> actualData = Lists.newArrayListWithExpectedSize(size);
        ResultSet rs = conn.query("select * from "+ nullableBTable);
        while(rs.next()){
            int[] data = new int[2];
            data[0] = rs.getInt(1);
            Assert.assertFalse("Column a was null!",rs.wasNull());
            data[1] = rs.getInt(2);
            Assert.assertTrue("Column b was null!", rs.wasNull());
            actualData.add(data);
        }

        Collections.sort(correctPartialData, intArrayComparator);
        Collections.sort(actualData, intArrayComparator);
        assertCorrectResult(correctPartialData,actualData);
    }

    @Test
    public void testMergeWithEmptyTableWillInsert() throws Exception {
        CallableStatement statement =
                conn.prepareCall("call SYSCS_UTIL.MERGE_DATA_FROM_FILE(?,?,null,?,null,null,null,null,null,0,?,null,null)");
        statement.setString(1,schema.schemaName);
        statement.setString(2,nullableBTable.tableName);
        statement.setString(3,fullTestFile.getFilePath());
        statement.setString(4,BADDIR.getCanonicalPath());

        ResultSet resultSet = statement.executeQuery();
        //make sure that the bad records list is good
        validateMergeResults(resultSet,size,0, size,0);

        //make sure that the data matches
        List<int[]> actualData = Lists.newArrayListWithExpectedSize(size);
        ResultSet rs = conn.query("select * from "+ nullableBTable);
        while(rs.next()){
            int[] data = new int[2];
            data[0] = rs.getInt(1);
            Assert.assertFalse("Column a was null!",rs.wasNull());
            data[1] = rs.getInt(2);
            Assert.assertFalse("Column b was null!",rs.wasNull());
            actualData.add(data);
        }

        Collections.sort(correctFullData, intArrayComparator);
        Collections.sort(actualData, intArrayComparator);
        assertCorrectResult(correctFullData,actualData);
    }

    @Test
    public void testMergeWithAutoGeneratedColumn() throws Exception {
        PreparedStatement ps = conn.prepareStatement("insert into " + tableWithAutoCol + "(a,b) values (?,?)");
        for(int i=1; i<= 3; i++){
            ps.setInt(1, i); ps.setInt(2, i*10); ps.executeUpdate();
        }

        //add one extra row that should not be overwritten
        ps.setInt(1,size+1); ps.setInt(2,(size+1)*10); ps.executeUpdate();


        String sqlText = "select a,b,c,d from "+ tableWithAutoCol.tableName + " order by 1";

        // this is the table content before the merge operation
        String expected = "A | B | C | D  |\n" +
                "-----------------\n" +
                " 1 |10 | 1 |999 |\n" +
                " 2 |20 | 2 |999 |\n" +
                " 3 |30 | 3 |999 |\n" +
                " 6 |60 | 4 |999 |";

        ResultSet rs = conn.query(sqlText);
        String resultString = TestUtils.FormattedResult.ResultFactory.toStringUnsorted(rs);
        assertEquals("\n" + sqlText + "\n" + "expected result: " + expected + "\n,actual result: " + resultString, expected, resultString);
        rs.close();

        CallableStatement statement =
                conn.prepareCall("call SYSCS_UTIL.MERGE_DATA_FROM_FILE(?,?,'a,b',?,null,null,null,null,null,0,?,null,null)");
        statement.setString(1,schema.schemaName);
        statement.setString(2,tableWithAutoCol.tableName);
        statement.setString(3,fullTestFile.getFilePath());
        statement.setString(4,BADDIR.getCanonicalPath());

        ResultSet resultSet = statement.executeQuery();
        //make sure that the bad records list is good
        //we should have two rows inserted (0,0) and (4,8)
        validateMergeResults(resultSet,5, 3,2,0);

        sqlText = "select a,b,c,d from "+ tableWithAutoCol.tableName + " order by 1";

        // we expect the 2nd to 4th rows to be updated, the 1st and 5th rows to be inserted, and the last row to be intact
        expected = "A | B | C | D  |\n" +
                "-----------------\n" +
                " 0 | 0 | 5 |999 |\n" +
                " 1 | 2 | 1 |999 |\n" +
                " 2 | 4 | 2 |999 |\n" +
                " 3 | 6 | 3 |999 |\n" +
                " 4 | 8 | 6 |999 |\n" +
                " 6 |60 | 4 |999 |";

        rs = conn.query(sqlText);
        resultString = TestUtils.FormattedResult.ResultFactory.toStringUnsorted(rs);
        assertEquals("\n" + sqlText + "\n" + "expected result: " + expected + "\n,actual result: " + resultString, expected, resultString);
        rs.close();
    }

    @Test
    public void testMergeWithColumnWithDefaultValue() throws Exception {
        PreparedStatement ps = conn.prepareStatement("insert into " + tableWithDefaultCol + "(a,b) values (?,?)");
        for(int i=1; i<= 3; i++){
            ps.setInt(1, i); ps.setInt(2, i*10); ps.executeUpdate();
        }

        //add one extra row that should not be overwritten
        ps.setInt(1,size+1); ps.setInt(2,(size+1)*10); ps.executeUpdate();


        String sqlText = "select a,b,c,d from "+ tableWithDefaultCol.tableName + " order by 1";

        // this is the table content before the merge operation
        String expected = "A | B | C | D  |\n" +
                "-----------------\n" +
                " 1 |10 | 1 |999 |\n" +
                " 2 |20 | 2 |999 |\n" +
                " 3 |30 | 3 |999 |\n" +
                " 6 |60 | 4 |999 |";

        ResultSet rs = conn.query(sqlText);
        String resultString = TestUtils.FormattedResult.ResultFactory.toStringUnsorted(rs);
        assertEquals("\n" + sqlText + "\n" + "expected result: " + expected + "\n,actual result: " + resultString, expected, resultString);
        rs.close();

        CallableStatement statement =
                conn.prepareCall("call SYSCS_UTIL.MERGE_DATA_FROM_FILE(?,?,'a,d',?,null,null,null,null,null,0,?,null,null)");
        statement.setString(1,schema.schemaName);
        statement.setString(2,tableWithDefaultCol.tableName);
        statement.setString(3,fullTestFile.getFilePath());
        statement.setString(4,BADDIR.getCanonicalPath());

        ResultSet resultSet = statement.executeQuery();
        //make sure that the bad records list is good
        //we should have two rows inserted (0,0) and (4,8)
        validateMergeResults(resultSet,5,3,2,0);

        sqlText = "select a,b,c,d from "+ tableWithDefaultCol.tableName + " order by 1";

        // we expect the 2nd to 4th rows to be updated, the 1st and 5th rows to be inserted, and the last row to be intact
        expected = "A |  B  | C | D  |\n" +
                "-------------------\n" +
                " 0 |NULL | 5 | 0  |\n" +
                " 1 | 10  | 1 | 2  |\n" +
                " 2 | 20  | 2 | 4  |\n" +
                " 3 | 30  | 3 | 6  |\n" +
                " 4 |NULL | 6 | 8  |\n" +
                " 6 | 60  | 4 |999 |";

        rs = conn.query(sqlText);
        resultString = TestUtils.FormattedResult.ResultFactory.toStringUnsorted(rs);
        assertEquals("\n" + sqlText + "\n" + "expected result: " + expected + "\n,actual result: " + resultString, expected, resultString);
        rs.close();
    }

    @Test
    public void testMergeIntoTableInDifferentSchema() throws Exception {
        PreparedStatement ps = conn.prepareStatement("insert into " + occupiedTable + "(a,b) values (?,?)");
        List<int[]> newCorrect = Lists.newArrayList(correctFullData);
        for(int[] correctRow:newCorrect){
            //add a row that's different
            ps.setInt(1,correctRow[0]); ps.setInt(2,correctRow[1]/2); ps.executeUpdate();
        }
        //add one extra row so that we know that we don't overwrite the entire table
        ps.setInt(1,size+1); ps.setInt(2,size+1); ps.executeUpdate();
        newCorrect.add(new int[]{size + 1, size + 1});

        // set to a schema different from the one where the table is located
        conn.setSchema("SPLICE");
        CallableStatement statement =
                conn.prepareCall("call SYSCS_UTIL.MERGE_DATA_FROM_FILE(?,?,null,?,null,null,null,null,null,0,?,null,null)");
        statement.setString(1,schema.schemaName);
        statement.setString(2,occupiedTable.tableName);
        statement.setString(3,fullTestFile.getFilePath());
        statement.setString(4,BADDIR.getCanonicalPath());

        ResultSet resultSet = statement.executeQuery();
        //make sure that the bad records list is good
        //all 5 rows are updated, no extra row inserted
        validateMergeResults(resultSet,5, 5,0, 0);

        // set back the schema
        conn.setSchema(schema.schemaName);
        //make sure that the data matches
        List<int[]> actualData = Lists.newArrayListWithExpectedSize(size);
        ResultSet rs = conn.query("select * from "+ occupiedTable);
        while(rs.next()){
            int[] data = new int[2];
            data[0] = rs.getInt(1);
            Assert.assertFalse("Column a was null!",rs.wasNull());
            data[1] = rs.getInt(2);
            Assert.assertFalse("Column b was null!",rs.wasNull());
            actualData.add(data);
        }

        Collections.sort(newCorrect, intArrayComparator);
        Collections.sort(actualData, intArrayComparator);
        assertCorrectResult(newCorrect, actualData);
    }

    @Test
    public void testMergeWithFilesInDirectory() throws Exception {
        PreparedStatement ps = conn.prepareStatement("insert into " + tableForUpdateInDirectory + "(a,b) values (?,?)");
        for(int i=1; i<= 3; i++){
            ps.setInt(1, i); ps.setInt(2, i*10); ps.executeUpdate();
        }

        //add one extra row that should not be overwritten
        ps.setInt(1,size+1); ps.setInt(2,(size+1)*10); ps.executeUpdate();


        String sqlText = "select a,b,c,d from "+ tableForUpdateInDirectory.tableName + " order by 1";

        // this is the table content before the merge operation
        String expected = "A | B | C | D  |\n" +
                "-----------------\n" +
                " 1 |10 | 1 |999 |\n" +
                " 2 |20 | 2 |999 |\n" +
                " 3 |30 | 3 |999 |\n" +
                " 6 |60 | 4 |999 |";

        ResultSet rs = conn.query(sqlText);
        String resultString = TestUtils.FormattedResult.ResultFactory.toStringUnsorted(rs);
        assertEquals("\n" + sqlText + "\n" + "expected result: " + expected + "\n,actual result: " + resultString, expected, resultString);
        rs.close();

        CallableStatement statement =
                conn.prepareCall("call SYSCS_UTIL.MERGE_DATA_FROM_FILE(?,?,'a,d',?,null,null,null,null,null,-1,?,null,null)");
        statement.setString(1,schema.schemaName);
        statement.setString(2,tableForUpdateInDirectory.tableName);
        statement.setString(3,IMPORTDIR.getCanonicalPath());
        statement.setString(4,BADDIR.getCanonicalPath());

        ResultSet resultSet = statement.executeQuery();

//        validateMergeResults(resultSet,9,2,5,3);

        sqlText = "select a,b,c,d from "+ tableForUpdateInDirectory.tableName + " order by 1";

        // This is the values among all imported files in the directory
        //        A          |2
        //        -----------------------
        //        0          |0
        //        1          |2
        //        2          |4
        //        3          |6
        //        4          |8
        //        0          |0
        //        1          |2
        //        2          |4
        //        3          |6
        //        4          |8
        //        4          |8
        //        0          |0
        //        1          |0
        //        2          |0
        //        3          |0
        //        4          |0
        // Since imported files have conflicted values, (1,2) vs (1,0), (2,4) vs (2,0) , (3,6) vs (3,0), (4,8) vs (4,0)
        // The sequence of updating data is not certain, the final result is not stable. It could lead data corruption
        // when imported files contain conflicted data.
        // This test case just used for verify MERGE_DATA_FROM_FILE support import from a directory

        expected = "A |  B  | C | D  |\n" +
                "-------------------\n" +
                " 0 |NULL | 5 | 0  |\n" +
                " 1 | 10  | 1 | 2  |\n" +
                " 2 | 20  | 2 | 4  |\n" +
                " 3 | 30  | 3 | 6  |\n" +
                " 4 |NULL | 6 | 8  |\n" +
                " 6 | 60  | 4 |999 |";

//        rs = conn.query(sqlText);
//        resultString = TestUtils.FormattedResult.ResultFactory.toStringUnsorted(rs);
//        assertEquals("\n" + sqlText + "\n" + "expected result: " + expected + "\n,actual result: " + resultString, expected, resultString);
//        rs.close();
    }

    @Test
    public void testMergeIntoTableWithErrorFormatData() throws Exception {
        PreparedStatement ps = conn.prepareStatement("insert into " + occupiedTable + "(a,b) values (?,?)");
        List<int[]> newCorrect = Lists.newArrayList(correctFullData);
        for(int[] correctRow:newCorrect){
            //add a row that's different
            ps.setInt(1,correctRow[0]); ps.setInt(2,correctRow[1]/2); ps.executeUpdate();
        }
        //add one extra row so that we know that we don't overwrite the entire table
        ps.setInt(1,size+1); ps.setInt(2,size+1); ps.executeUpdate();
        newCorrect.add(new int[]{size + 1, size + 1});

        // set to a schema different from the one where the table is located
        conn.setSchema("SPLICE");
        CallableStatement statement =
                conn.prepareCall("call SYSCS_UTIL.MERGE_DATA_FROM_FILE(?,?,null,?,null,null,null,null,null,-1,?,null,null)");
        statement.setString(1,schema.schemaName);
        statement.setString(2,occupiedTable.tableName);
        statement.setString(3,fullTestFileWithEmptyLine.getFilePath());
        statement.setString(4,BADDIR.getCanonicalPath());

        ResultSet resultSet = statement.executeQuery();
        //make sure that the bad records list is good
        //all 5 rows are updated, no extra row inserted
        validateMergeResults(resultSet,5,5,0,1);
    }


    @Test
    public void testUpsertIndexCorrectness() throws Exception {
        spliceClassWatcher.prepareStatement(format("call SYSCS_UTIL.IMPORT_DATA('%s','%s',null,'%s','|','\"',null,null,null,0,null,true,null)", CLASS_NAME, "LINEITEM", TPCHIT.getResource("lineitem.tbl"))).execute();
        assertEquals(9958L, (long)spliceClassWatcher.query("select count(*) from lineitem"));
        spliceClassWatcher.prepareStatement(format("call SYSCS_UTIL.UPSERT_DATA_FROM_FILE('%s','%s',null,'%s','|','\"',null,null,null,0,null,true,null)", CLASS_NAME, "LINEITEM", TPCHIT.getResource("lineitem.tbl"))).execute();
        assertEquals(9958L, (long)spliceClassWatcher.query("select count(*) from lineitem"));
        assertEquals(9958L, (long)spliceClassWatcher.query("select count(*) from lineitem --splice-properties index=L_SHIPDATE_IDX"));
        assertEquals(9958L, (long)spliceClassWatcher.query("select count(*) from lineitem --splice-properties index=L_PART_IDX"));
    }
    /*******************************************************************************************************************/
    /*helper methods*/

    private void assertCorrectResult(List<int[]> correct,List<int[]> actualData) {
        Assert.assertEquals("Incorrect result size!", correct.size(), actualData.size());
        for(int i=0;i<correct.size();i++){
            int[] corr = correct.get(i);
            int[] actual = actualData.get(i);
            Assert.assertEquals("Incorrect value for column A!",corr[0],actual[0]);
            Assert.assertEquals("Incorrect value for column B!",corr[1],actual[1]);
        }
    }

    private static TestConnection createConnection() throws Exception {
        Connection baseConn = SpliceNetConnection.getDefaultConnection();
        TestConnection conn = new TestConnection(baseConn);
        conn.setSchema(schema.schemaName);
        return conn;
    }
}
