CsvUpsertExecutor implements Closeable { public void execute(Iterable<CSVRecord> csvRecords) { for (CSVRecord csvRecord : csvRecords) { execute(csvRecord); } } CsvUpsertExecutor(Connection conn, List<ColumnInfo> columnInfoList, PreparedStatement preparedStatement,
            UpsertListener upsertListener, String arrayElementSeparator); static CsvUpsertExecutor create(PhoenixConnection conn, String tableName, List<ColumnInfo> columnInfoList,
            UpsertListener upsertListener, String arrayElementSeparator); void execute(Iterable<CSVRecord> csvRecords); @Override void close();  }
@Test public void testExecute() throws Exception { upsertExecutor.execute(createCsvRecord("123,NameValue,42,1:2:3")); verify(upsertListener).upsertDone(1L); verifyNoMoreInteractions(upsertListener); verify(preparedStatement).setObject(1, Long.valueOf(123L)); verify(preparedStatement).setObject(2, "NameValue"); verify(preparedStatement).setObject(3, Integer.valueOf(42)); verify(preparedStatement).setObject(4, PArrayDataType.instantiatePhoenixArray(PInteger.INSTANCE, new Object[]{1,2,3})); verify(preparedStatement).execute(); verifyNoMoreInteractions(preparedStatement); }
@Test public void testExecute_TooFewFields() throws Exception { CSVRecord csvRecordWithTooFewFields = createCsvRecord("123,NameValue"); upsertExecutor.execute(csvRecordWithTooFewFields); verify(upsertListener).errorOnRecord(eq(csvRecordWithTooFewFields), anyString()); verifyNoMoreInteractions(upsertListener); }
@Test public void testExecute_TooManyFields() throws Exception { CSVRecord csvRecordWithTooManyFields = createCsvRecord("123,NameValue,42,1:2:3,Garbage"); upsertExecutor.execute(csvRecordWithTooManyFields); verify(upsertListener).upsertDone(1L); verifyNoMoreInteractions(upsertListener); verify(preparedStatement).setObject(1, Long.valueOf(123L)); verify(preparedStatement).setObject(2, "NameValue"); verify(preparedStatement).setObject(3, Integer.valueOf(42)); verify(preparedStatement).setObject(4, PArrayDataType.instantiatePhoenixArray(PInteger.INSTANCE, new Object[]{1,2,3})); verify(preparedStatement).execute(); verifyNoMoreInteractions(preparedStatement); }
@Test public void testExecute_NullField() throws Exception { upsertExecutor.execute(createCsvRecord("123,NameValue,,1:2:3")); verify(upsertListener).upsertDone(1L); verifyNoMoreInteractions(upsertListener); verify(preparedStatement).setObject(1, Long.valueOf(123L)); verify(preparedStatement).setObject(2, "NameValue"); verify(preparedStatement).setNull(3, columnInfoList.get(2).getSqlType()); verify(preparedStatement).setObject(4, PArrayDataType.instantiatePhoenixArray(PInteger.INSTANCE, new Object[]{1,2,3})); verify(preparedStatement).execute(); verifyNoMoreInteractions(preparedStatement); }
@Test public void testExecute_InvalidType() throws Exception { CSVRecord csvRecordWithInvalidType = createCsvRecord("123,NameValue,ThisIsNotANumber,1:2:3"); upsertExecutor.execute(csvRecordWithInvalidType); verify(upsertListener).errorOnRecord(eq(csvRecordWithInvalidType), anyString()); verifyNoMoreInteractions(upsertListener); }