PgCopy { public void copy(Collection<T> items, Connection connection) { if (items == null || items.isEmpty()) { return; } try { var reader = buildCsv(items); Stopwatch stopwatch = Stopwatch.createStarted(); CopyManager copyManager = connection.unwrap(PGConnection.class).getCopyAPI(); long rowsCount = copyManager.copyIn(sql, reader, properties.getBufferSize()); insertDurationMetric.record(stopwatch.elapsed()); log.info("Copied {} rows to {} table in {}", rowsCount, tableName, stopwatch); } catch (Exception e) { throw new ParserException(String.format("Error copying %d items to table %s", items.size(), tableName), e); } } PgCopy(Class<T> entityClass, MeterRegistry meterRegistry, SqlProperties properties); void copy(Collection<T> items, Connection connection);  }
@Test void testCopy() throws SQLException { var cryptoTransfers = new HashSet<CryptoTransfer>(); cryptoTransfers.add(cryptoTransfer(1)); cryptoTransfers.add(cryptoTransfer(2)); cryptoTransfers.add(cryptoTransfer(3)); var tokenTransfers = new HashSet<TokenTransfer>(); tokenTransfers.add(tokenTransfer(1)); tokenTransfers.add(tokenTransfer(2)); tokenTransfers.add(tokenTransfer(3)); cryptoTransferPgCopy.copy(cryptoTransfers, dataSource.getConnection()); tokenTransferPgCopy.copy(tokenTransfers, dataSource.getConnection()); assertThat(cryptoTransferRepository.findAll()).containsExactlyInAnyOrderElementsOf(cryptoTransfers); assertThat(tokenTransferRepository.findAll()).containsExactlyInAnyOrderElementsOf(tokenTransfers); }
@Test void testCopyDuplicates() throws SQLException { var transactions = new HashSet<Transaction>(); transactions.add(transaction(1)); transactions.add(transaction(2)); transactions.add(transaction(2)); transactions.add(transaction(3)); transactionPgCopy.copy(transactions, dataSource.getConnection()); assertThat(transactionRepository.findAll()).hasSize(3).containsExactlyInAnyOrderElementsOf(transactions); }
@Test void throwsParserException() throws SQLException, IOException { CopyManager copyManager = mock(CopyManager.class); doThrow(SQLException.class).when(copyManager).copyIn(any(), (Reader) any(), anyInt()); PGConnection pgConnection = mock(PGConnection.class); doReturn(copyManager).when(pgConnection).getCopyAPI(); DataSource dataSource = mock(DataSource.class); Connection conn = mock(Connection.class); doReturn(conn).when(dataSource).getConnection(); doReturn(pgConnection).when(conn).unwrap(any()); var cryptoTransferPgCopy2 = new PgCopy<>(CryptoTransfer.class, meterRegistry, sqlProperties); var cryptoTransfers = new HashSet<CryptoTransfer>(); cryptoTransfers.add(cryptoTransfer(1)); assertThatThrownBy(() -> cryptoTransferPgCopy2.copy(cryptoTransfers, dataSource.getConnection())) .isInstanceOf(ParserException.class); }
@Test void testNullItems() throws SQLException { cryptoTransferPgCopy.copy(null, dataSource.getConnection()); assertThat(cryptoTransferRepository.count()).isEqualTo(0); }
@Test void testLargeConsensusSubmitMessage() throws SQLException { var topicMessages = new HashSet<TopicMessage>(); topicMessages.add(topicMessage(1, 6000)); topicMessages.add(topicMessage(2, 6000)); topicMessages.add(topicMessage(3, 6000)); topicMessages.add(topicMessage(4, 6000)); topicMessagePgCopy.copy(topicMessages, dataSource.getConnection()); assertThat(topicMessageRepository.findAll()).hasSize(4).containsExactlyInAnyOrderElementsOf(topicMessages); }