Fix #375 (#376)

Added `SQLDBConfig::getStatementAndConnection` method and deprecated
`SQLDBConfig::getStatement`, switching `SQLDatasource` to use the
latter. This deals with a leak where `getStatement` wasn't closing DB
connections

Data/src/main/java/org/tribuo/data/sql/SQLDBConfig.java

@@ -22,6 +22,7 @@
 import com.oracle.labs.mlrg.olcut.provenance.ConfiguredObjectProvenance;
 import com.oracle.labs.mlrg.olcut.provenance.Provenancable;
 import com.oracle.labs.mlrg.olcut.provenance.impl.ConfiguredObjectProvenanceImpl;
+import com.oracle.labs.mlrg.olcut.util.Pair;
 
 import java.sql.Connection;
 import java.sql.DriverManager;
@@ -167,16 +168,34 @@ public Connection getConnection() throws SQLException {
      * Constructs a statement based on the object fields. Uses fetchSize to determine fetch size and sets defaults
      * for querying data.
      *
+     * @deprecated because this leaks connections, use {{@link #getStatementAndConnection()}} instead.
+     *
      * @return A statement object for querying the database.
      * @throws SQLException If the connection failed.
      */
+    @Deprecated
     public Statement getStatement() throws SQLException {
         Statement stmt = getConnection().createStatement();
         stmt.setFetchSize(fetchSize);
         stmt.setFetchDirection(ResultSet.FETCH_FORWARD);
         return stmt;
     }
 
+    /**
+     * Creates a DB connection and constructs a statement based on the object fields. Uses fetchSize to determine fetch
+     * size and sets default for querying data.
+     *
+     * @return A pair of the statement object for querying the database, and the connection that it belongs to.
+     * @throws SQLException If the connection failed.
+     */
+    public Pair<Statement, Connection> getStatementAndConnection() throws SQLException {
+        Connection conn = getConnection();
+        Statement stmt = conn.createStatement();
+        stmt.setFetchSize(fetchSize);
+        stmt.setFetchDirection(ResultSet.FETCH_FORWARD);
+        return new Pair<>(stmt, conn);
+    }
+
     @Override
     public String toString() {
         if (connectionString != null) {

Data/src/main/java/org/tribuo/data/sql/SQLDataSource.java

@@ -23,13 +23,15 @@
 import com.oracle.labs.mlrg.olcut.provenance.impl.SkeletalConfiguredObjectProvenance;
 import com.oracle.labs.mlrg.olcut.provenance.primitives.DateTimeProvenance;
 import com.oracle.labs.mlrg.olcut.provenance.primitives.StringProvenance;
+import com.oracle.labs.mlrg.olcut.util.Pair;
 import org.tribuo.Output;
 import org.tribuo.OutputFactory;
 import org.tribuo.data.columnar.ColumnarDataSource;
 import org.tribuo.data.columnar.ColumnarIterator;
 import org.tribuo.data.columnar.RowProcessor;
 import org.tribuo.provenance.ConfiguredDataSourceProvenance;
 
+import java.sql.Connection;
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.time.OffsetDateTime;
@@ -64,7 +66,7 @@ public class SQLDataSource<T extends Output<T>> extends ColumnarDataSource<T> im
     @Config(mandatory = true,description="SQL query to run.")
     private String sqlString;
 
-    private final Set<Statement> statements = new HashSet<>();
+    private final Set<Pair<Statement, Connection>> statementsAndConnections = new HashSet<>();
 
     /**
      * For OLCUT.
@@ -94,24 +96,30 @@ public String toString() {
     @Override
     public ColumnarIterator rowIterator() {
         try {
-            Statement stmt = sqlConfig.getStatement();
-            statements.add(stmt);
-            return new ResultSetIterator(stmt.executeQuery(sqlString), stmt.getFetchSize());
+            Pair<Statement, Connection> stmtAndConn = sqlConfig.getStatementAndConnection();
+            statementsAndConnections.add(stmtAndConn);
+            return new ResultSetIterator(stmtAndConn.getA().executeQuery(sqlString), stmtAndConn.getA().getFetchSize());
         } catch (SQLException e) {
             throw new IllegalArgumentException("Error Processing SQL", e);
         }
     }
 
     @Override
     public void close() {
-        for (Statement statement: statements) {
+        for (Pair<Statement, Connection> statementsAndConnection : statementsAndConnections) {
             try {
-                statement.close();
+                statementsAndConnection.getA().close();
             } catch (SQLException e) {
                 logger.log(Level.WARNING, "Error closing statement", e);
             }
+
+            try {
+                statementsAndConnection.getB().close();
+            } catch (SQLException e) {
+                logger.log(Level.WARNING, "Error closing connection", e);
+            }
         }
-        statements.clear();
+        statementsAndConnections.clear();
     }
 
     @Override