@Test public void batchInsertWithTransaction() throws Exception { assertMemoryLeak(() -> { final CountDownLatch haltLatch = new CountDownLatch(1); final AtomicBoolean running = new AtomicBoolean(true); try { startBasicServer(NetworkFacadeImpl.INSTANCE, new DefaultPGWireConfiguration(), haltLatch, running ); Properties properties = new Properties(); properties.setProperty("user", "admin"); properties.setProperty("password", "quest"); properties.setProperty("sslmode", "disable"); final Connection connection = DriverManager.getConnection("jdbc:postgresql: try (Statement statement = connection.createStatement()) { statement.executeUpdate("create table test (id long,val int)"); statement.executeUpdate("create table test2(id long,val int)"); } connection.setAutoCommit(false); try (PreparedStatement batchInsert = connection.prepareStatement("insert into test(id,val) values(?,?)")) { batchInsert.setLong(1, 0L); batchInsert.setInt(2, 1); batchInsert.addBatch(); batchInsert.setLong(1, 1L); batchInsert.setInt(2, 2); batchInsert.addBatch(); batchInsert.setLong(1, 2L); batchInsert.setInt(2, 3); batchInsert.addBatch(); batchInsert.clearParameters(); batchInsert.executeLargeBatch(); } try (PreparedStatement batchInsert = connection.prepareStatement("insert into test2(id,val) values(?,?)")) { batchInsert.setLong(1, 0L); batchInsert.setInt(2, 1); batchInsert.addBatch(); batchInsert.setLong(1, 1L); batchInsert.setInt(2, 2); batchInsert.addBatch(); batchInsert.setLong(1, 2L); batchInsert.setInt(2, 3); batchInsert.addBatch(); batchInsert.clearParameters(); batchInsert.executeLargeBatch(); } connection.commit(); connection.setAutoCommit(true); StringSink sink = new StringSink(); String expected = "id[BIGINT],val[INTEGER]\n" + "0,1\n" + "1,2\n" + "2,3\n"; Statement statement = connection.createStatement(); ResultSet rs = statement.executeQuery("select * from test"); assertResultSet(expected, sink, rs); sink.clear(); Statement statement2 = connection.createStatement(); ResultSet rs2 = statement2.executeQuery("select * from test2"); assertResultSet(expected, sink, rs2); connection.setAutoCommit(true); try (PreparedStatement batchInsert = connection.prepareStatement("insert into test(id,val) values(?,?)")) { batchInsert.setLong(1, 3L); batchInsert.setInt(2, 4); batchInsert.addBatch(); batchInsert.setLong(1, 4L); batchInsert.setInt(2, 5); batchInsert.addBatch(); batchInsert.setLong(1, 5L); batchInsert.setInt(2, 6); batchInsert.addBatch(); batchInsert.clearParameters(); batchInsert.executeLargeBatch(); } sink.clear(); expected = "id[BIGINT],val[INTEGER]\n" + "0,1\n" + "1,2\n" + "2,3\n" + "3,4\n" + "4,5\n" + "5,6\n"; Statement statement3 = connection.createStatement(); ResultSet rs3 = statement3.executeQuery("select * from test"); assertResultSet(expected, sink, rs3); try (Statement statement4 = connection.createStatement()) { statement4.executeUpdate("create table anothertab(id long, val int, k timestamp) timestamp(k) "); } connection.setAutoCommit(false); try (PreparedStatement batchInsert = connection.prepareStatement("insert into anothertab(id, val, k) values(?,?,?)")) { batchInsert.setLong(1, 3L); batchInsert.setInt(2, 4); batchInsert.setLong(3, 1_000L); batchInsert.addBatch(); batchInsert.setLong(1, 4L); batchInsert.setInt(2, 5); batchInsert.setLong(3, 0L); batchInsert.addBatch(); batchInsert.setLong(1, 5L); batchInsert.setInt(2, 6); batchInsert.setLong(3, 2_000L); batchInsert.addBatch(); batchInsert.clearParameters(); batchInsert.executeLargeBatch(); Assert.fail(); } catch (Exception e) { LOG.error().$(e).$(); } connection.commit(); connection.rollback(); connection.setAutoCommit(true); sink.clear(); expected = "id[BIGINT],val[INTEGER],k[TIMESTAMP]\n"; Statement statement4 = connection.createStatement(); ResultSet rs4 = statement4.executeQuery("select * from anothertab"); assertResultSet(expected, sink, rs4); connection.close(); } finally { running.set(false); haltLatch.await(); } }); }
@Override public void close() { Misc.free(compiler); }
PGJobContext implements Closeable { @Override public void close() { Misc.free(compiler); } }
PGJobContext implements Closeable { @Override public void close() { Misc.free(compiler); } PGJobContext(PGWireConfiguration configuration, CairoEngine engine, MessageBus messageBus, FunctionFactoryCache functionFactoryCache); }
PGJobContext implements Closeable { @Override public void close() { Misc.free(compiler); } PGJobContext(PGWireConfiguration configuration, CairoEngine engine, MessageBus messageBus, FunctionFactoryCache functionFactoryCache); @Override void close(); void handleClientOperation(PGConnectionContext context); }
PGJobContext implements Closeable { @Override public void close() { Misc.free(compiler); } PGJobContext(PGWireConfiguration configuration, CairoEngine engine, MessageBus messageBus, FunctionFactoryCache functionFactoryCache); @Override void close(); void handleClientOperation(PGConnectionContext context); static final int PG_VARCHAR; static final int PG_TIMESTAMP; static final int PG_TIMESTAMPZ; static final int PG_FLOAT8; static final int PG_FLOAT4; static final int PG_INT4; static final int PG_INT2; static final int PG_INT8; static final int PG_NUMERIC; static final int PG_BOOL; static final int PG_CHAR; static final int PG_DATE; static final int PG_BYTEA; static final int PG_UNSPECIFIED; }