refactor: collapse StreamingResultSet.of factories into one

StreamingResultSet had two public factories — of(reader, allocator,
queryId[, zone]) (4 callers) and ofClosingOnFailure(Result, queryId,
zone) (5 callers). Every production caller wanted the close-on-failure
behavior; only tests and the metadata helper used the bare of(). Two
factories with overlapping responsibilities is one too many — a caller
hitting the bare of() and not knowing about ofClosingOnFailure would
silently leak the 100 MB RootAllocator on construction failure.

Collapse to one public factory:
- of(QueryResultArrowStream.Result, queryId, sessionZone) — the only
  factory callers see, always closes both reader and allocator on
  failure. Name is the unambiguous "of" because there is no other.
- create(reader, allocator, queryId, sessionZone) — private; just the
  construction body the factory wraps.

Production call sites (DataCloudConnection, DataCloudStatement) and
MetadataResultSets were already passing a (reader, allocator) pair,
so the call shape collapses to passing the Result holder. Tests that
were building the pair locally now wrap it in a Result the same way.

Author: mkaufmann

jdbc-core/src/main/java/com/salesforce/datacloud/jdbc/core/DataCloudConnection.java

@@ -221,7 +221,7 @@ public DataCloudResultSet getRowBasedResultSet(String queryId, long offset, long
                     QueryResultArrowStream.OUTPUT_FORMAT);
             val arrowStream = SQLExceptionQueryResultIterator.createSqlExceptionArrowStreamReader(
                     iterator, connectionProperties.isIncludeCustomerDetailInReason(), queryId, null);
-            return StreamingResultSet.ofClosingOnFailure(arrowStream, queryId, ZoneId.systemDefault());
+            return StreamingResultSet.of(arrowStream, queryId, ZoneId.systemDefault());
         } catch (StatusRuntimeException ex) {
             throw QueryExceptionHandler.createException(
                     connectionProperties.isIncludeCustomerDetailInReason(), null, queryId, ex);
@@ -264,7 +264,7 @@ public DataCloudResultSet getChunkBasedResultSet(String queryId, long chunkId, l
                     QueryResultArrowStream.OUTPUT_FORMAT);
             val arrowStream = SQLExceptionQueryResultIterator.createSqlExceptionArrowStreamReader(
                     iterator, connectionProperties.isIncludeCustomerDetailInReason(), queryId, null);
-            return StreamingResultSet.ofClosingOnFailure(arrowStream, queryId, ZoneId.systemDefault());
+            return StreamingResultSet.of(arrowStream, queryId, ZoneId.systemDefault());
         } catch (StatusRuntimeException ex) {
             throw QueryExceptionHandler.createException(
                     connectionProperties.isIncludeCustomerDetailInReason(), null, queryId, ex);

jdbc-core/src/main/java/com/salesforce/datacloud/jdbc/core/DataCloudStatement.java

@@ -196,8 +196,8 @@ public ResultSet executeQuery(String sql) throws SQLException {
             val iterator = executeAdaptiveQuery(sql);
             val arrowStream = SQLExceptionQueryResultIterator.createSqlExceptionArrowStreamReader(
                     iterator, includeCustomerDetail, iterator.getQueryStatus().getQueryId(), sql);
-            resultSet = StreamingResultSet.ofClosingOnFailure(
-                    arrowStream, iterator.getQueryStatus().getQueryId(), sessionZone);
+            resultSet =
+                    StreamingResultSet.of(arrowStream, iterator.getQueryStatus().getQueryId(), sessionZone);
             log.info(
                     "executeAdaptiveQuery completed. queryId={}, sessionZone={}",
                     queryHandle.getQueryStatus().getQueryId(),
@@ -436,7 +436,7 @@ public ResultSet getResultSet() throws SQLException {
                                     includeCustomerDetail,
                                     adaptiveIterator.getQueryStatus().getQueryId(),
                                     null);
-                            resultSet = StreamingResultSet.ofClosingOnFailure(
+                            resultSet = StreamingResultSet.of(
                                     arrowStream,
                                     adaptiveIterator.getQueryStatus().getQueryId(),
                                     sessionZone);

jdbc-core/src/main/java/com/salesforce/datacloud/jdbc/core/StreamingResultSet.java

@@ -74,25 +74,38 @@ private StreamingResultSet(
         this.closed = false;
     }
 
-    public static StreamingResultSet of(ArrowStreamReader reader, BufferAllocator allocator, String queryId)
-            throws SQLException {
-        return of(reader, allocator, queryId, ZoneId.systemDefault());
-    }
-
     /**
-     * Creates a StreamingResultSet from an {@link ArrowStreamReader} and its backing allocator.
+     * Creates a StreamingResultSet from a {@link QueryResultArrowStream.Result} (reader paired
+     * with its backing allocator).
      *
      * <p>Ownership of both the reader and the allocator transfers to the returned result set —
      * closing the result set closes the reader and then the allocator, in that order, so Arrow's
-     * buffer accounting clears before the allocator's budget check. Callers must not close
-     * either separately.
+     * buffer accounting clears before the allocator's budget check. If construction itself
+     * throws (for example a {@link SQLException} wrapping an unsupported Arrow type), this
+     * method closes both before re-throwing so the 100 MB {@link
+     * org.apache.arrow.memory.RootAllocator} does not leak. Callers must not close either
+     * separately on success.
      *
-     * @param reader The Arrow stream, owned by the result set.
-     * @param allocator The allocator backing the reader, owned by the result set.
-     * @param queryId The query identifier.
+     * @param arrowStream The Arrow stream + allocator pair, both owned by the result set.
+     * @param queryId The query identifier (may be {@code null} for synthesized result sets).
      * @param sessionZone The session timezone used for timestamp conversions.
      */
-    public static StreamingResultSet of(
+    public static StreamingResultSet of(QueryResultArrowStream.Result arrowStream, String queryId, ZoneId sessionZone)
+            throws SQLException {
+        try {
+            return create(arrowStream.getReader(), arrowStream.getAllocator(), queryId, sessionZone);
+        } catch (SQLException | RuntimeException ex) {
+            try {
+                arrowStream.getReader().close();
+            } catch (Exception suppressed) {
+                ex.addSuppressed(suppressed);
+            }
+            arrowStream.getAllocator().close();
+            throw ex;
+        }
+    }
+
+    private static StreamingResultSet create(
             ArrowStreamReader reader, BufferAllocator allocator, String queryId, ZoneId sessionZone)
             throws SQLException {
         try {
@@ -113,28 +126,6 @@ public static StreamingResultSet of(
         }
     }
 
-    /**
-     * Hand the reader + allocator pair from {@link QueryResultArrowStream.Result} to {@link
-     * #of(ArrowStreamReader, BufferAllocator, String, ZoneId)} and close both on construction
-     * failure. Without this, an {@code of} call that throws (for example {@code SQLException}
-     * wrapping an unsupported Arrow type) would leak the 100 MB
-     * {@link org.apache.arrow.memory.RootAllocator} held by {@code Result}.
-     */
-    public static StreamingResultSet ofClosingOnFailure(
-            QueryResultArrowStream.Result arrowStream, String queryId, ZoneId sessionZone) throws SQLException {
-        try {
-            return of(arrowStream.getReader(), arrowStream.getAllocator(), queryId, sessionZone);
-        } catch (SQLException | RuntimeException ex) {
-            try {
-                arrowStream.getReader().close();
-            } catch (Exception suppressed) {
-                ex.addSuppressed(suppressed);
-            }
-            arrowStream.getAllocator().close();
-            throw ex;
-        }
-    }
-
     // --- Core ResultSet navigation ---
 
     @Override

jdbc-core/src/main/java/com/salesforce/datacloud/jdbc/core/metadata/MetadataResultSets.java

@@ -5,6 +5,7 @@
 package com.salesforce.datacloud.jdbc.core.metadata;
 
 import com.salesforce.datacloud.jdbc.core.StreamingResultSet;
+import com.salesforce.datacloud.jdbc.protocol.QueryResultArrowStream;
 import com.salesforce.datacloud.jdbc.protocol.data.ColumnMetadata;
 import com.salesforce.datacloud.jdbc.protocol.data.HyperTypeToArrow;
 import com.salesforce.datacloud.jdbc.protocol.data.VectorPopulator;
@@ -62,13 +63,9 @@ public static StreamingResultSet of(List<ColumnMetadata> columns, List<List<Obje
         // Allocator is handed to StreamingResultSet along with the reader; the result set owns
         // its lifecycle and closes it when close() is called.
         RootAllocator allocator = new RootAllocator(Long.MAX_VALUE);
-        try {
-            ArrowStreamReader reader = new ArrowStreamReader(new ByteArrayInputStream(ipcBytes), allocator);
-            return StreamingResultSet.of(reader, allocator, /*queryId=*/ null, ZoneId.systemDefault());
-        } catch (SQLException | RuntimeException ex) {
-            allocator.close();
-            throw ex;
-        }
+        ArrowStreamReader reader = new ArrowStreamReader(new ByteArrayInputStream(ipcBytes), allocator);
+        return StreamingResultSet.of(
+                new QueryResultArrowStream.Result(reader, allocator), /*queryId=*/ null, ZoneId.systemDefault());
     }
 
     /**

jdbc-core/src/test/java/com/salesforce/datacloud/jdbc/core/StreamCloseTest.java

@@ -81,7 +81,7 @@ void closingResultSetClosesUnderlyingIterator() {
             // ByteStringReadableByteChannel(iterator, resource) → ArrowStreamReader
             val arrowStream = SQLExceptionQueryResultIterator.createSqlExceptionArrowStreamReader(
                     tracked, false, "test-query", null);
-            val resultSet = StreamingResultSet.of(arrowStream.getReader(), arrowStream.getAllocator(), "test-query");
+            val resultSet = StreamingResultSet.of(arrowStream, "test-query", java.time.ZoneId.systemDefault());
 
             // Read one row — stream is still open with remaining rows
             assertThat(resultSet.next()).isTrue();

jdbc-core/src/test/java/com/salesforce/datacloud/jdbc/core/StreamingResultSetMethodTest.java

@@ -84,7 +84,8 @@ private StreamingResultSet createSingleVarCharResultSet(boolean nullValue) {
 
         RootAllocator readerAllocator = new RootAllocator(Long.MAX_VALUE);
         ArrowStreamReader reader = new ArrowStreamReader(new ByteArrayInputStream(out.toByteArray()), readerAllocator);
-        return StreamingResultSet.of(reader, readerAllocator, QUERY_ID);
+        return StreamingResultSet.of(
+                new QueryResultArrowStream.Result(reader, readerAllocator), QUERY_ID, ZoneId.systemDefault());
     }
 
     // --- Unsupported methods ---
@@ -167,7 +168,7 @@ void ofClosingOnFailureClosesAllocatorWhenSchemaIsUnsupported() {
         val reader = spy(new ArrowStreamReader(new ByteArrayInputStream(out.toByteArray()), readerAllocator));
         val arrowStream = new QueryResultArrowStream.Result(reader, readerAllocator);
 
-        assertThatThrownBy(() -> StreamingResultSet.ofClosingOnFailure(arrowStream, QUERY_ID, ZoneId.systemDefault()))
+        assertThatThrownBy(() -> StreamingResultSet.of(arrowStream, QUERY_ID, ZoneId.systemDefault()))
                 .isInstanceOf(SQLException.class)
                 .hasMessageContaining("Unsupported column type");