/*
 * 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.phoenix.end2end;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.StringReader;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.Collections;
import java.util.Properties;

import com.google.common.collect.ImmutableList;
import org.apache.commons.csv.CSVParser;
import org.apache.commons.csv.CSVRecord;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixTestDriver;
import org.apache.phoenix.schema.IllegalDataException;
import org.apache.phoenix.schema.types.PInteger;
import org.apache.phoenix.schema.types.PArrayDataType;
import org.apache.phoenix.util.CSVCommonsLoader;
import org.apache.phoenix.util.DateUtil;
import org.apache.phoenix.util.PhoenixRuntime;
import org.junit.Test;

public class CSVCommonsLoaderIT extends ParallelStatsDisabledIT {

    private static final String DATATYPE_TABLE = "DATATYPE";
    private static final String DATATYPES_CSV_VALUES = "CKEY, CVARCHAR, CCHAR, CINTEGER, CDECIMAL, CUNSIGNED_INT, CBOOLEAN, CBIGINT, CUNSIGNED_LONG, CTIME, CDATE\n"
            + "KEY1,A,A,2147483647,1.1,0,TRUE,9223372036854775807,0,1990-12-31 10:59:59,1999-12-31 23:59:59\n"
            + "KEY2,B,B,-2147483648,-1.1,2147483647,FALSE,-9223372036854775808,9223372036854775807,2000-01-01 00:00:01,2012-02-29 23:59:59\n"
            + "KEY3,,,,,,,,,,\n";
    private static final String STOCK_CSV_VALUES = "AAPL,APPLE Inc.\n"
            + "CRM,SALESFORCE\n" + "GOOG,Google\n"
            + "HOG,Harlet-Davidson Inc.\n" + "HPQ,Hewlett Packard\n"
            + "INTC,Intel\n" + "MSFT,Microsoft\n" + "WAG,Walgreens\n"
            + "WMT,Walmart\n";
    private static final String[] STOCK_COLUMNS_WITH_BOGUS = new String[] {
            "SYMBOL", "BOGUS" };
    private static final String[] STOCK_COLUMNS = new String[] { "SYMBOL",
            "COMPANY" };
    private static final String STOCK_CSV_VALUES_WITH_HEADER = STOCK_COLUMNS[0]
            + "," + STOCK_COLUMNS[1] + "\n" + STOCK_CSV_VALUES;
    private static final String STOCK_CSV_VALUES_WITH_DELIMITER = "APPL"
            + '\u0001' + '\u0002' + "APPLE\n" + " Inc" + '\u0002' + "\n"
            + "MSFT" + '\u0001' + "Microsoft\n";

    private static final String STOCK_TDV_VALUES = "AAPL\tAPPLE Inc\n"
            + "CRM\tSALESFORCE\n" + "GOOG\tGoogle\n"
            + "HOG\tHarlet-Davidson Inc.\n" + "HPQ\tHewlett Packard\n"
            + "INTC\tIntel\n" + "MSFT\tMicrosoft\n" + "WAG\tWalgreens\n"
            + "WMT\tWalmart\n";
    private static final String STOCK_TDV_VALUES_WITH_HEADER = STOCK_COLUMNS[0]
            + "\t" + STOCK_COLUMNS[1] + "\n" + STOCK_TDV_VALUES;


    private static final String ENCAPSULATED_CHARS_TABLE = "ENCAPSULATEDCHAR";
    private static final String[] ENCAPSULATED_CHARS_COLUMNS = new String[] {
            "MYKEY", "MYVALUE" };
    private static final String CSV_VALUES_ENCAPSULATED_CONTROL_CHARS = "ALL THREEF,\"This has a all three , , \"\" \r\n in it. \"\n"
            + "COMMA,\"This has a comma , in it. \"\n"
            + "CRLF,\"This has a crlf \r\n in it. \"\n"
            + "QUOTE,\"This has a quote \"\" in it. \"\n";
    private static final String CSV_VALUES_ENCAPSULATED_CONTROL_CHARS_WITH_HEADER = ENCAPSULATED_CHARS_COLUMNS[0]
            + ","
            + ENCAPSULATED_CHARS_COLUMNS[1]
            + "\n"
            + CSV_VALUES_ENCAPSULATED_CONTROL_CHARS;
    private static final String CSV_VALUES_BAD_ENCAPSULATED_CONTROL_CHARS = "ALL THREEF,\"This has a all three , , \"\" \r\n in it. \"\n"
            + "COMMA,\"This has a comma , in it. \"\n"
            + "CRLF,\"This has a crlf \r\n in it. \"\n"
            + "BADENCAPSULATEDQUOTE,\"\"This has a bad quote in it. \"\n";
    private static final String CSV_VALUES_BAD_ENCAPSULATED_CONTROL_CHARS_WITH_HEADER = ENCAPSULATED_CHARS_COLUMNS[0]
            + ","
            + ENCAPSULATED_CHARS_COLUMNS[1]
            + "\n"
            + CSV_VALUES_BAD_ENCAPSULATED_CONTROL_CHARS;

    @Test
    public void testCSVCommonsUpsert() throws Exception {
        CSVParser parser = null;
        PhoenixConnection conn = null;
        try {
            String stockTableName = generateUniqueName();
            // Create table
            String statements = "CREATE TABLE IF NOT EXISTS " + stockTableName
                    + "(SYMBOL VARCHAR NOT NULL PRIMARY KEY, COMPANY VARCHAR);";
            conn = DriverManager.getConnection(getUrl()).unwrap(
                    PhoenixConnection.class);
            PhoenixRuntime.executeStatements(conn,
                    new StringReader(statements), null);

            // Upsert CSV file
            CSVCommonsLoader csvUtil = new CSVCommonsLoader(conn, stockTableName,
                    Collections.<String> emptyList(), true);
            csvUtil.upsert(new StringReader(STOCK_CSV_VALUES_WITH_HEADER));

            // Compare Phoenix ResultSet with CSV file content
            PreparedStatement statement = conn
                    .prepareStatement("SELECT SYMBOL, COMPANY FROM "
                            + stockTableName);
            ResultSet phoenixResultSet = statement.executeQuery();
            parser = new CSVParser(new StringReader(
                    STOCK_CSV_VALUES_WITH_HEADER), csvUtil.getFormat());
            for (CSVRecord record : parser) {
                assertTrue(phoenixResultSet.next());
                int i = 0;
                for (String value : record) {
                    assertEquals(value, phoenixResultSet.getString(i + 1));
                    i++;
                }
            }
            assertFalse(phoenixResultSet.next());
        } finally {
            if (parser != null)
                parser.close();
            if (conn != null)
                conn.close();
        }
    }

    @Test
    public void testCSVCommonsUpsert_MultiTenant() throws Exception {
        CSVParser parser = null;
        PhoenixConnection globalConn = null;
        PhoenixConnection tenantConn = null;
        try {
            String stockTableMultiName = generateUniqueName();

            // Create table using the global connection
            String statements = "CREATE TABLE IF NOT EXISTS " + stockTableMultiName
                    + "(TENANT_ID VARCHAR NOT NULL, SYMBOL VARCHAR NOT NULL, COMPANY VARCHAR," +
                    " CONSTRAINT PK PRIMARY KEY(TENANT_ID,SYMBOL)) MULTI_TENANT = true;";
            globalConn = DriverManager.getConnection(getUrl()).unwrap(
                    PhoenixConnection.class);
            PhoenixRuntime.executeStatements(globalConn,
                    new StringReader(statements), null);
            globalConn.close();

            tenantConn = new PhoenixTestDriver().connect(getUrl() + ";TenantId=acme", new Properties()).unwrap(
                    PhoenixConnection.class);

            // Upsert CSV file
            CSVCommonsLoader csvUtil = new CSVCommonsLoader(tenantConn, stockTableMultiName,
                    Collections.<String> emptyList(), true);
            csvUtil.upsert(new StringReader(STOCK_CSV_VALUES_WITH_HEADER));

            // Compare Phoenix ResultSet with CSV file content
            PreparedStatement statement = tenantConn
                    .prepareStatement("SELECT SYMBOL, COMPANY FROM "
                            + stockTableMultiName);
            ResultSet phoenixResultSet = statement.executeQuery();
            parser = new CSVParser(new StringReader(
                    STOCK_CSV_VALUES_WITH_HEADER), csvUtil.getFormat());
            for (CSVRecord record : parser) {
                assertTrue(phoenixResultSet.next());
                int i = 0;
                for (String value : record) {
                    assertEquals(value, phoenixResultSet.getString(i + 1));
                    i++;
                }
            }
            assertFalse(phoenixResultSet.next());
        } finally {
            if (parser != null)
                parser.close();
            if (tenantConn != null)
                tenantConn.close();
        }
    }

    @Test
    public void testTDVCommonsUpsert() throws Exception {
        CSVParser parser = null;
        PhoenixConnection conn = null;
        try {
            String stockTableName = generateUniqueName();

            // Create table
            String statements = "CREATE TABLE IF NOT EXISTS " + stockTableName
                    + "(SYMBOL VARCHAR NOT NULL PRIMARY KEY, COMPANY VARCHAR);";
            conn = DriverManager.getConnection(getUrl()).unwrap(
                    PhoenixConnection.class);
            PhoenixRuntime.executeStatements(conn,
                    new StringReader(statements), null);

            // Upsert TDV file
            CSVCommonsLoader csvUtil = new CSVCommonsLoader(conn, stockTableName,Collections.<String> emptyList()
                    , true, '\t', '"', null, CSVCommonsLoader.DEFAULT_ARRAY_ELEMENT_SEPARATOR);
            csvUtil.upsert(new StringReader(STOCK_TDV_VALUES_WITH_HEADER));

            // Compare Phoenix ResultSet with CSV file content
            PreparedStatement statement = conn
                    .prepareStatement("SELECT SYMBOL, COMPANY FROM "
                            + stockTableName);
            ResultSet phoenixResultSet = statement.executeQuery();
            parser = new CSVParser(new StringReader(
                    STOCK_TDV_VALUES_WITH_HEADER), csvUtil.getFormat());
            for (CSVRecord record : parser) {
                assertTrue(phoenixResultSet.next());
                int i = 0;
                for (String value : record) {
                    assertEquals(value, phoenixResultSet.getString(i + 1));
                    i++;
                }
            }
            assertFalse(phoenixResultSet.next());
        } finally {
            if (parser != null)
                parser.close();
            if (conn != null)
                conn.close();
        }
    }

    @Test
    public void testCSVUpsertWithCustomDelimiters() throws Exception {
        CSVParser parser = null;
        PhoenixConnection conn = null;
        try {
            String stockTableName = generateUniqueName();

            // Create table
            String statements = "CREATE TABLE IF NOT EXISTS " + stockTableName
                    + "(SYMBOL VARCHAR NOT NULL PRIMARY KEY, COMPANY VARCHAR);";
            conn = DriverManager.getConnection(getUrl()).unwrap(
                    PhoenixConnection.class);
            PhoenixRuntime.executeStatements(conn,
                    new StringReader(statements), null);

            // Upsert CSV file
            CSVCommonsLoader csvUtil = new CSVCommonsLoader(conn, stockTableName,
                    Arrays.<String> asList(STOCK_COLUMNS), true,
                    '1', '2', '3', CSVCommonsLoader.DEFAULT_ARRAY_ELEMENT_SEPARATOR);
            csvUtil.upsert(new StringReader(STOCK_CSV_VALUES_WITH_DELIMITER));

            // Compare Phoenix ResultSet with CSV file content
            PreparedStatement statement = conn
                    .prepareStatement("SELECT SYMBOL, COMPANY FROM "
                            + stockTableName);
            ResultSet phoenixResultSet = statement.executeQuery();
            parser = new CSVParser(new StringReader(
                    STOCK_CSV_VALUES_WITH_DELIMITER), csvUtil.getFormat());
            for (CSVRecord record : parser) {
                assertTrue(phoenixResultSet.next());
                int i = 0;
                for (String value : record) {
                    assertEquals(value, phoenixResultSet.getString(i + 1));
                    i++;
                }
            }
            assertFalse(phoenixResultSet.next());
        } finally {
            if (parser != null)
                parser.close();
            if (conn != null)
                conn.close();
        }
    }

    @Test
    public void testCSVUpsertWithColumns() throws Exception {
        CSVParser parser = null;
        PhoenixConnection conn = null;
        try {
            String stockTableName = generateUniqueName();

            // Create table
            String statements = "CREATE TABLE IF NOT EXISTS " + stockTableName
                    + "(SYMBOL VARCHAR NOT NULL PRIMARY KEY, COMPANY VARCHAR);";
            conn = DriverManager.getConnection(getUrl())
                    .unwrap(PhoenixConnection.class);
            PhoenixRuntime.executeStatements(conn,
                    new StringReader(statements), null);

            // Upsert CSV file
            CSVCommonsLoader csvUtil = new CSVCommonsLoader(conn, stockTableName,
                    Arrays.<String> asList(STOCK_COLUMNS), true);
            // no header
            csvUtil.upsert(new StringReader(STOCK_CSV_VALUES));

            // Compare Phoenix ResultSet with CSV file content
            PreparedStatement statement = conn
                    .prepareStatement("SELECT SYMBOL, COMPANY FROM "
                            + stockTableName);
            ResultSet phoenixResultSet = statement.executeQuery();
            parser = new CSVParser(new StringReader(
                    STOCK_CSV_VALUES), csvUtil.getFormat());
            for (CSVRecord record : parser) {
                assertTrue(phoenixResultSet.next());
                int i = 0;
                for (String value : record) {
                    assertEquals(value, phoenixResultSet.getString(i + 1));
                    i++;
                }
            }

            assertFalse(phoenixResultSet.next());
        } finally {
            if (parser != null)
                parser.close();
            if (conn != null)
                conn.close();
        }
    }

    @Test
    public void testCSVUpsertWithNoColumns() throws Exception {
        CSVParser parser = null;
        PhoenixConnection conn = null;
        try {
            String stockTableName = generateUniqueName();

            // Create table
            String statements = "CREATE TABLE IF NOT EXISTS " + stockTableName
                    + "(SYMBOL VARCHAR NOT NULL PRIMARY KEY, COMPANY VARCHAR);";
            conn = DriverManager.getConnection(getUrl())
                    .unwrap(PhoenixConnection.class);
            PhoenixRuntime.executeStatements(conn,
                    new StringReader(statements), null);

            // Upsert CSV file
            CSVCommonsLoader csvUtil = new CSVCommonsLoader(conn, stockTableName,
                    null, true);
            csvUtil.upsert(new StringReader(STOCK_CSV_VALUES));

            // Compare Phoenix ResultSet with CSV file content
            PreparedStatement statement = conn
                    .prepareStatement("SELECT SYMBOL, COMPANY FROM "
                            + stockTableName);
            ResultSet phoenixResultSet = statement.executeQuery();
            parser = new CSVParser(new StringReader(
                    STOCK_CSV_VALUES), csvUtil.getFormat());
            for (CSVRecord record : parser) {
                assertTrue(phoenixResultSet.next());
                int i = 0;
                for (String value : record) {
                    assertEquals(value, phoenixResultSet.getString(i + 1));
                    i++;
                }
            }

            assertFalse(phoenixResultSet.next());
        } finally {
            if (parser != null)
                parser.close();
            if (conn != null)
                conn.close();
        }
    }

    @Test
    public void testCSVUpsertWithBogusColumn() throws Exception {
        CSVParser parser = null;
        PhoenixConnection conn = null;
        try {
            String stockTableName = generateUniqueName();

            // Create table
            String statements = "CREATE TABLE IF NOT EXISTS " + stockTableName
                    + "(SYMBOL VARCHAR NOT NULL PRIMARY KEY, COMPANY VARCHAR);";
            conn = DriverManager.getConnection(getUrl())
                    .unwrap(PhoenixConnection.class);
            PhoenixRuntime.executeStatements(conn,
                    new StringReader(statements), null);

            // Upsert CSV file, not strict
            CSVCommonsLoader csvUtil = new CSVCommonsLoader(conn, stockTableName,
                    Arrays.asList(STOCK_COLUMNS_WITH_BOGUS), false);
            csvUtil.upsert(new StringReader(STOCK_CSV_VALUES));

            // Compare Phoenix ResultSet with CSV file content
            PreparedStatement statement = conn
                    .prepareStatement("SELECT SYMBOL, COMPANY FROM "
                            + stockTableName);
            ResultSet phoenixResultSet = statement.executeQuery();
            parser = new CSVParser(new StringReader(STOCK_CSV_VALUES),
                    csvUtil.getFormat());
            for (CSVRecord record : parser) {
                assertTrue(phoenixResultSet.next());
                assertEquals(record.get(0), phoenixResultSet.getString(1));
                assertNull(phoenixResultSet.getString(2));
            }

            assertFalse(phoenixResultSet.next());
        } finally {
            if (parser != null)
                parser.close();
            if (conn != null)
                conn.close();
        }
    }

    // Ensure that strict mode also causes the import to stop if a data type on a single
    // row is not correct
    @Test
    public void testCSVUpsertWithInvalidNumericalData_StrictMode() throws Exception {
        CSVParser parser = null;
        PhoenixConnection conn = null;
        try {
            String stockTableName = generateUniqueName();

            // Create table
            String statements = "CREATE TABLE IF NOT EXISTS " + stockTableName
                    + "(SYMBOL VARCHAR NOT NULL PRIMARY KEY, COMPANY_ID BIGINT);";
            conn = DriverManager.getConnection(getUrl())
                    .unwrap(PhoenixConnection.class);
            PhoenixRuntime.executeStatements(conn,
                    new StringReader(statements), null);

            // Upsert CSV file in strict mode
            CSVCommonsLoader csvUtil = new CSVCommonsLoader(conn, stockTableName,
                    Arrays.asList("SYMBOL", "COMPANY_ID"), true);
            try {
                csvUtil.upsert(new StringReader(STOCK_CSV_VALUES));
                fail("Running an upsert with data that can't be upserted in strict mode "
                        + "should throw an exception");
            } catch (IllegalDataException e) {
                // Expected
            }

        } finally {
            if (parser != null)
                parser.close();
            if (conn != null)
                conn.close();
        }
    }

    @Test
    public void testCSVUpsertWithAllColumn() throws Exception {
        CSVParser parser = null;
        PhoenixConnection conn = null;
        try {
            String stockTableName = generateUniqueName();

            // Create table
            String statements = "CREATE TABLE IF NOT EXISTS " + stockTableName
                    + "(SYMBOL VARCHAR NOT NULL PRIMARY KEY, COMPANY VARCHAR);";
            conn = DriverManager.getConnection(getUrl())
                    .unwrap(PhoenixConnection.class);
            PhoenixRuntime.executeStatements(conn,
                    new StringReader(statements), null);

            // Upsert CSV file
            CSVCommonsLoader csvUtil = new CSVCommonsLoader(conn, stockTableName,
                    Arrays.asList("FOO", "BAR"), false);

            try {
                csvUtil.upsert(new StringReader(STOCK_CSV_VALUES));
                fail();
            } catch (SQLException e) {
                assertTrue(
                        e.getMessage(),
                        e.getMessage()
                                .contains(
                                        "ERROR 504 (42703): Undefined column. columnName=" + stockTableName + ".[FOO, BAR]"));
            }
        } finally {
            if (parser != null)
                parser.close();
            if (conn != null)
                conn.close();
        }
    }

    @Test
    public void testCSVUpsertWithBogusColumnStrict() throws Exception {
        CSVParser parser = null;
        PhoenixConnection conn = null;
        try {
            String stockTableName = generateUniqueName();

            // Create table
            String statements = "CREATE TABLE IF NOT EXISTS " + stockTableName
                    + "(SYMBOL VARCHAR NOT NULL PRIMARY KEY, COMPANY VARCHAR);";
            conn = DriverManager.getConnection(getUrl())
                    .unwrap(PhoenixConnection.class);
            PhoenixRuntime.executeStatements(conn,
                    new StringReader(statements), null);

            // Upsert CSV file
            CSVCommonsLoader csvUtil = new CSVCommonsLoader(conn, stockTableName,
                    Arrays.asList(STOCK_COLUMNS_WITH_BOGUS), true);
            try {
                csvUtil.upsert(new StringReader(STOCK_CSV_VALUES));
                fail();
            } catch (SQLException e) {
                assertTrue(
                        e.getMessage(),
                        e.getMessage()
                                .contains(
                                        "ERROR 504 (42703): Undefined column. columnName=" + stockTableName + ".BOGUS"));
            }
        } finally {
            if (parser != null)
                parser.close();
            if (conn != null)
                conn.close();
        }
    }

    @Test
    public void testAllDatatypes() throws Exception {
        CSVParser parser = null;
        PhoenixConnection conn = null;
        try {
            // Create table
            String statements = "CREATE TABLE IF NOT EXISTS "
                    + DATATYPE_TABLE
                    + " (CKEY VARCHAR NOT NULL PRIMARY KEY,"
                    + "  CVARCHAR VARCHAR, CCHAR CHAR(10), CINTEGER INTEGER, CDECIMAL DECIMAL(31,10), CUNSIGNED_INT UNSIGNED_INT, CBOOLEAN BOOLEAN, CBIGINT BIGINT, CUNSIGNED_LONG UNSIGNED_LONG, CTIME TIME, CDATE DATE);";
            conn = DriverManager.getConnection(getUrl())
                    .unwrap(PhoenixConnection.class);
            PhoenixRuntime.executeStatements(conn,
                    new StringReader(statements), null);

            // Upsert CSV file
            CSVCommonsLoader csvUtil = new CSVCommonsLoader(conn,
                    DATATYPE_TABLE, Collections.<String> emptyList(), true);
            csvUtil.upsert(new StringReader(DATATYPES_CSV_VALUES));

            // Compare Phoenix ResultSet with CSV file content
            PreparedStatement statement = conn
                    .prepareStatement("SELECT CKEY, CVARCHAR, CCHAR, CINTEGER, CDECIMAL, CUNSIGNED_INT, CBOOLEAN, CBIGINT, CUNSIGNED_LONG, CTIME, CDATE FROM "
                            + DATATYPE_TABLE);
            ResultSet phoenixResultSet = statement.executeQuery();
            parser = new CSVParser(new StringReader(DATATYPES_CSV_VALUES),
                    csvUtil.getFormat());

            for (CSVRecord record : parser) {
                assertTrue(phoenixResultSet.next());
                int i = 0;
                int size = record.size();
                for (String value : record) {
                    assertEquals(value, phoenixResultSet.getObject(i + 1)
                            .toString().toUpperCase());
                    if (i < size - 2)
                        break;
                    i++;
                }
                // special case for matching date, time values
                String timeFieldValue = record.get(9);
                assertEquals(timeFieldValue.isEmpty() ? null : DateUtil.parseTime(record.get(9)),
                        phoenixResultSet.getTime("CTIME"));

                String dateField = record.get(10);
                assertEquals(dateField.isEmpty() ? null : DateUtil.parseDate(record.get(10)),
                        phoenixResultSet.getDate("CDATE"));
            }

            assertFalse(phoenixResultSet.next());
        } finally {
            if (parser != null)
                parser.close();
            if (conn != null)
                conn.close();
        }
    }

    @Test
    public void testCSVCommonsUpsertEncapsulatedControlChars() throws Exception {
        CSVParser parser = null;
        PhoenixConnection conn = null;
        try {
            // Create table
            String statements = "CREATE TABLE IF NOT EXISTS "
                    + ENCAPSULATED_CHARS_TABLE
                    + "(MYKEY VARCHAR NOT NULL PRIMARY KEY, MYVALUE VARCHAR);";
            conn = DriverManager.getConnection(getUrl())
                    .unwrap(PhoenixConnection.class);
            PhoenixRuntime.executeStatements(conn,
                    new StringReader(statements), null);

            // Upsert CSV file
            CSVCommonsLoader csvUtil = new CSVCommonsLoader(conn,
                    ENCAPSULATED_CHARS_TABLE, Collections.<String> emptyList(),
                    true);
            csvUtil.upsert(new StringReader(
                    CSV_VALUES_ENCAPSULATED_CONTROL_CHARS_WITH_HEADER));

            // Compare Phoenix ResultSet with CSV file content
            PreparedStatement statement = conn
                    .prepareStatement("SELECT MYKEY, MYVALUE FROM "
                            + ENCAPSULATED_CHARS_TABLE);
            ResultSet phoenixResultSet = statement.executeQuery();
            parser = new CSVParser(new StringReader(
                    CSV_VALUES_ENCAPSULATED_CONTROL_CHARS_WITH_HEADER),
                    csvUtil.getFormat());
            for (CSVRecord record : parser) {
                assertTrue(phoenixResultSet.next());
                int i = 0;
                for (String value : record) {
                    assertEquals(value, phoenixResultSet.getString(i + 1));
                    i++;
                }
            }

            assertFalse(phoenixResultSet.next());
        } finally {
            if (parser != null)
                parser.close();
            if (conn != null)
                conn.close();
        }
    }

    @Test
    public void testCSVCommonsUpsertBadEncapsulatedControlChars()
            throws Exception {
        CSVParser parser = null;
        PhoenixConnection conn = null;
        try {
            // Create table
            String statements = "CREATE TABLE IF NOT EXISTS "
                    + ENCAPSULATED_CHARS_TABLE
                    + "(MYKEY VARCHAR NOT NULL PRIMARY KEY, MYVALUE VARCHAR);";
            conn = DriverManager.getConnection(getUrl())
                    .unwrap(PhoenixConnection.class);
            PhoenixRuntime.executeStatements(conn,
                    new StringReader(statements), null);

            // Upsert CSV file
            CSVCommonsLoader csvUtil = new CSVCommonsLoader(conn,
                    ENCAPSULATED_CHARS_TABLE, Collections.<String> emptyList(),
                    true);
            try {
                csvUtil.upsert(new StringReader(
                        CSV_VALUES_BAD_ENCAPSULATED_CONTROL_CHARS_WITH_HEADER));
                fail();
            } catch (RuntimeException e) {
                assertTrue(
                        e.getMessage(),
                        e.getMessage()
                                .contains(
                                        "invalid char between encapsulated token and delimiter"));
            }
        } finally {
            if (parser != null)
                parser.close();
            if (conn != null)
                conn.close();
        }
    }

    @Test
    public void testCSVCommonsUpsert_WithArray() throws Exception {
        CSVParser parser = null;
        PhoenixConnection conn = null;
        try {

            // Create table
            String statements = "CREATE TABLE IF NOT EXISTS ARRAY_TABLE "
                    + "(ID BIGINT NOT NULL PRIMARY KEY, VALARRAY INTEGER ARRAY);";
            conn = DriverManager.getConnection(getUrl()).unwrap(
                    PhoenixConnection.class);
            PhoenixRuntime.executeStatements(conn,
                    new StringReader(statements), null);

            // Upsert CSV file
            CSVCommonsLoader csvUtil = new CSVCommonsLoader(conn, "ARRAY_TABLE",
                    ImmutableList.<String>of(), true, ',', '"', null, "!");
            csvUtil.upsert(
                    new StringReader("ID,VALARRAY\n"
                            + "1,2!3!4\n"));

            // Compare Phoenix ResultSet with CSV file content
            PreparedStatement statement = conn
                    .prepareStatement("SELECT ID, VALARRAY FROM ARRAY_TABLE");
            ResultSet phoenixResultSet = statement.executeQuery();
            assertTrue(phoenixResultSet.next());
            assertEquals(1L, phoenixResultSet.getLong(1));
            assertEquals(
                    PArrayDataType.instantiatePhoenixArray(PInteger.INSTANCE, new Integer[]{2, 3, 4}),
                    phoenixResultSet.getArray(2));
            assertFalse(phoenixResultSet.next());
        } finally {
            if (parser != null)
                parser.close();
            if (conn != null)
                conn.close();
        }
    }

    @Test
    public void testCSVCommonsUpsert_WithTimestamp() throws Exception {
        CSVParser parser = null;
        PhoenixConnection conn = null;
        try {

            // Create table
            String statements = "CREATE TABLE IF NOT EXISTS TS_TABLE "
                    + "(ID BIGINT NOT NULL PRIMARY KEY, TS TIMESTAMP);";
            conn = DriverManager.getConnection(getUrl()).unwrap(
                    PhoenixConnection.class);
            PhoenixRuntime.executeStatements(conn,
                    new StringReader(statements), null);

            // Upsert CSV file
            CSVCommonsLoader csvUtil = new CSVCommonsLoader(conn, "TS_TABLE",
                    ImmutableList.<String>of(), true, ',', '"', null, "!");
            csvUtil.upsert(
                    new StringReader("ID,TS\n"
                            + "1,1970-01-01 00:00:10\n"
                            + "2,1970-01-01 00:00:10.123\n"));

            // Compare Phoenix ResultSet with CSV file content
            PreparedStatement statement = conn
                    .prepareStatement("SELECT ID, TS FROM TS_TABLE ORDER BY ID");
            ResultSet phoenixResultSet = statement.executeQuery();
            assertTrue(phoenixResultSet.next());
            assertEquals(1L, phoenixResultSet.getLong(1));
            assertEquals(10000L, phoenixResultSet.getTimestamp(2).getTime());
            assertTrue(phoenixResultSet.next());
            assertEquals(2L, phoenixResultSet.getLong(1));
            assertEquals(10123L, phoenixResultSet.getTimestamp(2).getTime());
            assertFalse(phoenixResultSet.next());
        } finally {
            if (parser != null)
                parser.close();
            if (conn != null)
                conn.close();
        }
    }

    @Test
    public void testCSVCommonsUpsert_NonExistentTable() throws Exception {
        PhoenixConnection conn = null;
        try {
            conn = DriverManager.getConnection(getUrl()).unwrap(
                    PhoenixConnection.class);
            CSVCommonsLoader csvUtil = new CSVCommonsLoader(conn, "NONEXISTENTTABLE",
                    null, true, ',', '"', null, "!");
            csvUtil.upsert(
                    new StringReader("ID,VALARRAY\n"
                            + "1,2!3!4\n"));
            fail("Trying to load a non-existent table should fail");
        } catch (IllegalArgumentException e) {
            assertEquals("Table NONEXISTENTTABLE not found", e.getMessage());
        } finally {
            if (conn != null) {
                conn.close();
            }
        }

    }
}