refactor!: unify ResultSet implementations on Arrow-backed path

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

@@ -17,25 +17,37 @@
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import lombok.val;
+import org.apache.arrow.memory.BufferAllocator;
 import org.apache.arrow.vector.FieldVector;
 import org.apache.arrow.vector.VectorSchemaRoot;
 import org.apache.arrow.vector.ipc.ArrowStreamReader;
 
+/**
+ * Row cursor over an {@link ArrowStreamReader} that drives the {@link DataCloudResultSet}.
+ *
+ * <p>The cursor owns the supplied {@link BufferAllocator} alongside the reader: closing the
+ * cursor closes the reader (which releases ArrowBuf accounting) and then the allocator (which
+ * returns its budget). This is the single place that guarantees root-allocator hygiene for the
+ * driver; callers of {@link DataCloudResultSet#of} hand ownership over and do not close the
+ * allocator themselves.
+ */
 @Slf4j
 class ArrowStreamReaderCursor implements AutoCloseable {
 
     private static final int INIT_ROW_NUMBER = -1;
 
     private final ArrowStreamReader reader;
+    private final BufferAllocator allocator;
     private final ZoneId sessionZone;
 
     @lombok.Getter
     private int rowsSeen = 0;
 
     private final AtomicInteger currentIndex = new AtomicInteger(INIT_ROW_NUMBER);
 
-    ArrowStreamReaderCursor(ArrowStreamReader reader, ZoneId sessionZone) {
+    ArrowStreamReaderCursor(ArrowStreamReader reader, BufferAllocator allocator, ZoneId sessionZone) {
         this.reader = reader;
+        this.allocator = allocator;
         this.sessionZone = sessionZone;
     }
 
@@ -91,6 +103,13 @@ public boolean next() {
     @SneakyThrows
     @Override
     public void close() {
-        reader.close();
+        // try-with-resources closes in reverse declaration order: reader first (releases the
+        // buffers accounted against the allocator so its closing budget check passes), then
+        // allocator. If both throw, Java attaches the second as suppressed onto the first
+        // instead of dropping the reader exception via the standard try/finally semantics.
+        try (BufferAllocator a = allocator;
+                ArrowStreamReader r = reader) {
+            // resource cleanup happens at exit
+        }
     }
 }

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

@@ -48,6 +48,7 @@
 import java.sql.Statement;
 import java.sql.Struct;
 import java.time.Duration;
+import java.time.ZoneId;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
@@ -220,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.of(arrowStream, queryId);
+            return DataCloudResultSet.of(arrowStream, queryId, ZoneId.systemDefault());
         } catch (StatusRuntimeException ex) {
             throw QueryExceptionHandler.createException(
                     connectionProperties.isIncludeCustomerDetailInReason(), null, queryId, ex);
@@ -263,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.of(arrowStream, queryId);
+            return DataCloudResultSet.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/DataCloudDatabaseMetadata.java

@@ -13,7 +13,7 @@
 
 import com.google.common.collect.ImmutableList;
 import com.salesforce.datacloud.jdbc.config.DriverVersion;
-import com.salesforce.datacloud.jdbc.core.metadata.DataCloudResultSetMetaData;
+import com.salesforce.datacloud.jdbc.core.metadata.MetadataResultSets;
 import com.salesforce.datacloud.jdbc.core.types.HyperTypes;
 import com.salesforce.datacloud.jdbc.util.JdbcURL;
 import com.salesforce.datacloud.jdbc.util.ThrowingJdbcSupplier;
@@ -706,39 +706,39 @@ public ResultSet getColumns(String catalog, String schemaPattern, String tableNa
     @Override
     public ResultSet getColumnPrivileges(String catalog, String schema, String table, String columnNamePattern)
             throws SQLException {
-        return DataCloudMetadataResultSet.empty();
+        return MetadataResultSets.emptyNoColumns();
     }
 
     @Override
     public ResultSet getTablePrivileges(String catalog, String schemaPattern, String tableNamePattern)
             throws SQLException {
-        return DataCloudMetadataResultSet.empty();
+        return MetadataResultSets.emptyNoColumns();
     }
 
     @Override
     public ResultSet getBestRowIdentifier(String catalog, String schema, String table, int scope, boolean nullable)
             throws SQLException {
-        return DataCloudMetadataResultSet.empty();
+        return MetadataResultSets.emptyNoColumns();
     }
 
     @Override
     public ResultSet getVersionColumns(String catalog, String schema, String table) throws SQLException {
-        return DataCloudMetadataResultSet.empty();
+        return MetadataResultSets.emptyNoColumns();
     }
 
     @Override
     public ResultSet getPrimaryKeys(String catalog, String schema, String table) throws SQLException {
-        return DataCloudMetadataResultSet.empty();
+        return MetadataResultSets.emptyNoColumns();
     }
 
     @Override
     public ResultSet getImportedKeys(String catalog, String schema, String table) throws SQLException {
-        return DataCloudMetadataResultSet.empty();
+        return MetadataResultSets.emptyNoColumns();
     }
 
     @Override
     public ResultSet getExportedKeys(String catalog, String schema, String table) throws SQLException {
-        return DataCloudMetadataResultSet.empty();
+        return MetadataResultSets.emptyNoColumns();
     }
 
     @Override
@@ -750,19 +750,18 @@ public ResultSet getCrossReference(
             String foreignSchema,
             String foreignTable)
             throws SQLException {
-        return DataCloudMetadataResultSet.empty();
+        return MetadataResultSets.emptyNoColumns();
     }
 
     @Override
     public ResultSet getTypeInfo() throws SQLException {
-        return DataCloudMetadataResultSet.of(
-                new DataCloudResultSetMetaData(MetadataSchemas.TYPE_INFO), HyperTypes.typeInfoRows());
+        return MetadataResultSets.ofRawRows(MetadataSchemas.TYPE_INFO, HyperTypes.typeInfoRows());
     }
 
     @Override
     public ResultSet getIndexInfo(String catalog, String schema, String table, boolean unique, boolean approximate)
             throws SQLException {
-        return DataCloudMetadataResultSet.empty();
+        return MetadataResultSets.emptyNoColumns();
     }
 
     @Override