ChinmaySKulkarni commented on a change in pull request #746: PHOENIX-5802:
Connection leaks in UPSERT SELECT/DELETE paths due to
MutatingParallelIteratorFactory iterator not being closed
URL: https://github.com/apache/phoenix/pull/746#discussion_r399483172
##########
File path:
phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertSelectIT.java
##########
@@ -101,145 +122,163 @@ private void testUpsertSelect(boolean createIndex,
boolean saltTable) throws Exc
Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
props.setProperty(QueryServices.ENABLE_SERVER_SIDE_UPSERT_MUTATIONS,
allowServerSideMutations);
- String aTable = initATableValues(tenantId, saltTable ? null : splits,
null, null, getUrl(), saltTable ? "salt_buckets = 2" : null);
+ String aTable = initATableValues(tenantId, saltTable ? null : splits,
null,
+ null, getUrl(), saltTable ? "salt_buckets = 2" : null);
String customEntityTable = generateUniqueName();
- Connection conn = DriverManager.getConnection(getUrl(), props);
- String ddl = "create table " + customEntityTable +
- " (organization_id char(15) not null, \n" +
- " key_prefix char(3) not null,\n" +
- " custom_entity_data_id char(12) not null,\n" +
- " created_by varchar,\n" +
- " created_date date,\n" +
- " currency_iso_code char(3),\n" +
- " deleted char(1),\n" +
- " division decimal(31,10),\n" +
- " last_activity date,\n" +
- " last_update date,\n" +
- " last_update_by varchar,\n" +
- " name varchar(240),\n" +
- " owner varchar,\n" +
- " record_type_id char(15),\n" +
- " setup_owner varchar,\n" +
- " system_modstamp date,\n" +
- " b.val0 varchar,\n" +
- " b.val1 varchar,\n" +
- " b.val2 varchar,\n" +
- " b.val3 varchar,\n" +
- " b.val4 varchar,\n" +
- " b.val5 varchar,\n" +
- " b.val6 varchar,\n" +
- " b.val7 varchar,\n" +
- " b.val8 varchar,\n" +
- " b.val9 varchar\n" +
- " CONSTRAINT pk PRIMARY KEY (organization_id, key_prefix,
custom_entity_data_id)) " + (saltTable ? "salt_buckets = 2" : "");
- conn.createStatement().execute(ddl);
- conn.close();
-
+ try (Connection conn = DriverManager.getConnection(getUrl(), props);
+ Statement stmt = conn.createStatement()) {
+ String ddl = "create table " + customEntityTable +
+ " (organization_id char(15) not null, \n" +
+ " key_prefix char(3) not null,\n" +
+ " custom_entity_data_id char(12) not null,\n" +
+ " created_by varchar,\n" +
+ " created_date date,\n" +
+ " currency_iso_code char(3),\n" +
+ " deleted char(1),\n" +
+ " division decimal(31,10),\n" +
+ " last_activity date,\n" +
+ " last_update date,\n" +
+ " last_update_by varchar,\n" +
+ " name varchar(240),\n" +
+ " owner varchar,\n" +
+ " record_type_id char(15),\n" +
+ " setup_owner varchar,\n" +
+ " system_modstamp date,\n" +
+ " b.val0 varchar,\n" +
+ " b.val1 varchar,\n" +
+ " b.val2 varchar,\n" +
+ " b.val3 varchar,\n" +
+ " b.val4 varchar,\n" +
+ " b.val5 varchar,\n" +
+ " b.val6 varchar,\n" +
+ " b.val7 varchar,\n" +
+ " b.val8 varchar,\n" +
+ " b.val9 varchar\n" +
+ " CONSTRAINT pk PRIMARY KEY " +
+ "(organization_id, key_prefix, custom_entity_data_id)) " +
+ (saltTable ? "salt_buckets = 2" : "");
+ stmt.execute(ddl);
+ }
+
String indexName = generateUniqueName();
if (createIndex) {
- conn = DriverManager.getConnection(getUrl(), props);
- conn.createStatement().execute("CREATE INDEX IF NOT EXISTS " +
indexName + " ON " + aTable + "(a_string)" );
- conn.close();
- }
- PreparedStatement upsertStmt;
- props.setProperty(UPSERT_BATCH_SIZE_ATTRIB, Integer.toString(3)); //
Trigger multiple batches
- conn = DriverManager.getConnection(getUrl(), props);
- conn.setAutoCommit(true);
- String upsert = "UPSERT INTO " + customEntityTable +
"(custom_entity_data_id, key_prefix, organization_id, created_by) " +
- "SELECT substr(entity_id, 4), substr(entity_id, 1, 3),
organization_id, a_string FROM " + aTable + " WHERE ?=a_string";
- if (createIndex) { // Confirm index is used
- upsertStmt = conn.prepareStatement("EXPLAIN " + upsert);
- upsertStmt.setString(1, tenantId);
- ResultSet ers = upsertStmt.executeQuery();
- assertTrue(ers.next());
- String explainPlan = QueryUtil.getExplainPlan(ers);
- assertTrue(explainPlan.contains(" SCAN OVER " + indexName));
+ try (Connection conn = DriverManager.getConnection(getUrl(),
props);
+ Statement stmt = conn.createStatement()) {
+ stmt.execute("CREATE INDEX IF NOT EXISTS " + indexName +
+ " ON " + aTable + "(a_string)" );
+ }
}
-
- upsertStmt = conn.prepareStatement(upsert);
- upsertStmt.setString(1, A_VALUE);
- int rowsInserted = upsertStmt.executeUpdate();
- assertEquals(4, rowsInserted);
- conn.commit();
- conn.close();
-
- String query = "SELECT key_prefix, substr(custom_entity_data_id, 1,
1), created_by FROM " + customEntityTable + " WHERE organization_id = ? ";
- conn = DriverManager.getConnection(getUrl(), props);
- PreparedStatement statement = conn.prepareStatement(query);
- statement.setString(1, tenantId);
- ResultSet rs = statement.executeQuery();
-
- assertTrue (rs.next());
- assertEquals("00A", rs.getString(1));
- assertEquals("1", rs.getString(2));
- assertEquals(A_VALUE, rs.getString(3));
-
- assertTrue (rs.next());
- assertEquals("00A", rs.getString(1));
- assertEquals("2", rs.getString(2));
- assertEquals(A_VALUE, rs.getString(3));
-
- assertTrue (rs.next());
- assertEquals("00A", rs.getString(1));
- assertEquals("3", rs.getString(2));
- assertEquals(A_VALUE, rs.getString(3));
-
- assertTrue (rs.next());
- assertEquals("00A", rs.getString(1));
- assertEquals("4", rs.getString(2));
- assertEquals(A_VALUE, rs.getString(3));
+ // Trigger multiple batches
+ props.setProperty(UPSERT_BATCH_SIZE_ATTRIB, Integer.toString(3));
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ conn.setAutoCommit(true);
+ String upsert = "UPSERT INTO " + customEntityTable +
+ "(custom_entity_data_id, key_prefix, organization_id,
created_by) " +
+ "SELECT substr(entity_id, 4), substr(entity_id, 1, 3),
organization_id, " +
+ "a_string FROM " + aTable + " WHERE ?=a_string";
+ if (createIndex) { // Confirm index is used
+ try (PreparedStatement upsertStmt =
+ conn.prepareStatement("EXPLAIN " + upsert)) {
+ upsertStmt.setString(1, tenantId);
+ ResultSet ers = upsertStmt.executeQuery();
+ assertTrue(ers.next());
+ String explainPlan = QueryUtil.getExplainPlan(ers);
+ assertTrue(explainPlan.contains(" SCAN OVER " +
indexName));
+ }
+ }
+
+ try (PreparedStatement upsertStmt = conn.prepareStatement(upsert))
{
+ upsertStmt.setString(1, A_VALUE);
+ int rowsInserted = upsertStmt.executeUpdate();
+ assertEquals(4, rowsInserted);
+ }
+ conn.commit();
+ }
+
+ String query = "SELECT key_prefix, substr(custom_entity_data_id, 1,
1), created_by FROM " +
+ customEntityTable + " WHERE organization_id = ? ";
+ try (Connection conn = DriverManager.getConnection(getUrl(), props);
+ PreparedStatement statement = conn.prepareStatement(query)) {
+ statement.setString(1, tenantId);
+ ResultSet rs = statement.executeQuery();
+
+ assertTrue (rs.next());
+ assertEquals("00A", rs.getString(1));
+ assertEquals("1", rs.getString(2));
+ assertEquals(A_VALUE, rs.getString(3));
+
+ assertTrue (rs.next());
+ assertEquals("00A", rs.getString(1));
+ assertEquals("2", rs.getString(2));
+ assertEquals(A_VALUE, rs.getString(3));
+
+ assertTrue (rs.next());
+ assertEquals("00A", rs.getString(1));
+ assertEquals("3", rs.getString(2));
+ assertEquals(A_VALUE, rs.getString(3));
+
+ assertTrue (rs.next());
+ assertEquals("00A", rs.getString(1));
+ assertEquals("4", rs.getString(2));
+ assertEquals(A_VALUE, rs.getString(3));
- assertFalse(rs.next());
- conn.close();
+ assertFalse(rs.next());
Review comment:
looks like this was a tab before and now a space (which is consistent) so
valid diff imo
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
