refactor: unify ResultSet implementations on Arrow-backed path

Collapse the two ResultSet families (streaming Arrow + row-based
metadata) into a single Arrow-backed implementation so there is one
accessor pipeline, one set of type semantics, and one place to fix bugs.

Changes:

- ArrowStreamReaderCursor becomes source-agnostic: a BatchLoader drives
  a VectorSchemaRoot, whether sourced from an ArrowStreamReader or a
  pre-populated in-memory batch. The cursor also owns an AutoCloseable
  so it is responsible for releasing the allocator + reader on close —
  the old ArrowStreamReader.close() would only tear down vectors and
  leak the 100 MB RootAllocator.
- QueryResultArrowStream.toArrowStreamReader returns a Result holder
  that pairs the reader with the allocator and closes both in the
  right order so Arrow's accounting invariants hold.
- StreamingResultSet gains ofInMemory(root, owned, queryId, zone, cols)
  so metadata results funnel through the same result set. A columns
  override preserves the JDBC-spec typeName labels (e.g. TEXT) that
  would otherwise be lost when deriving from the Arrow schema.
- MetadataArrowBuilder materialises List<List<Object>> metadata rows
  into a populated VectorSchemaRoot using the existing HyperTypeToArrow
  mapping; MetadataResultSets is the factory callers use.
- QueryMetadataUtil and DataCloudDatabaseMetadata route getTables,
  getColumns, getSchemas, getTypeInfo and empty metadata results
  through the Arrow-backed StreamingResultSet.
- DataCloudMetadataResultSet, SimpleResultSet, and ColumnAccessor are
  removed now that no caller depends on them.
- StreamingResultSet.getObject(int, Class) gains an isInstance-based
  fallback so callers can retrieve String-typed VARCHAR columns without
  each accessor having to implement typed getObject.
- Tests moved to the unified path; integer-accessor-only assertions in
  DataCloudDatabaseMetadataTest updated to reflect stricter Arrow
  accessor semantics.

Author: mkaufmann

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

@@ -8,7 +8,6 @@
 
 import com.salesforce.datacloud.jdbc.core.accessor.QueryJDBCAccessor;
 import com.salesforce.datacloud.jdbc.core.accessor.QueryJDBCAccessorFactory;
-import java.io.IOException;
 import java.sql.SQLException;
 import java.time.ZoneId;
 import java.util.List;
@@ -21,31 +20,79 @@
 import org.apache.arrow.vector.VectorSchemaRoot;
 import org.apache.arrow.vector.ipc.ArrowStreamReader;
 
+/**
+ * Row cursor over a {@link VectorSchemaRoot} that drives the {@link StreamingResultSet}.
+ *
+ * <p>The cursor is source-agnostic: a {@link BatchLoader} loads the next batch into the vector
+ * schema root, whether that comes from an Arrow IPC stream or a pre-populated in-memory batch.
+ * This is the single codepath that the driver exposes to JDBC callers — streaming query results
+ * and materialised metadata results both funnel through here.
+ *
+ * <p>The cursor owns an optional {@link AutoCloseable} holding the resources that back the
+ * vector schema root (allocator, underlying reader, etc.). Closing the cursor closes that holder,
+ * guaranteeing root-allocator hygiene without requiring each call site to manage the allocator
+ * separately.
+ */
 @Slf4j
 class ArrowStreamReaderCursor implements AutoCloseable {
 
     private static final int INIT_ROW_NUMBER = -1;
 
-    private final ArrowStreamReader reader;
+    private final VectorSchemaRoot root;
+    private final BatchLoader batchLoader;
+    private final AutoCloseable ownedResources;
     private final ZoneId sessionZone;
 
     @lombok.Getter
     private int rowsSeen = 0;
 
     private final AtomicInteger currentIndex = new AtomicInteger(INIT_ROW_NUMBER);
 
-    ArrowStreamReaderCursor(ArrowStreamReader reader, ZoneId sessionZone) {
-        this.reader = reader;
-        this.sessionZone = sessionZone;
+    /**
+     * Loads the next batch of rows into the vector schema root.
+     *
+     * <p>Implementations should return {@code true} if the vector schema root now holds rows from
+     * a newly-loaded batch, and {@code false} if the source has no more data.
+     */
+    @FunctionalInterface
+    interface BatchLoader {
+        boolean loadNextBatch() throws Exception;
     }
 
+    /**
+     * Create a cursor that pulls batches from an {@link ArrowStreamReader}. The reader (and the
+     * allocator it was constructed with) are owned by the cursor — closing the cursor closes the
+     * supplied {@code ownedResources}.
+     */
     @SneakyThrows
-    private VectorSchemaRoot getSchemaRoot() {
-        return reader.getVectorSchemaRoot();
+    static ArrowStreamReaderCursor streaming(
+            ArrowStreamReader reader, AutoCloseable ownedResources, ZoneId sessionZone) {
+        val root = reader.getVectorSchemaRoot();
+        BatchLoader loader = reader::loadNextBatch;
+        return new ArrowStreamReaderCursor(root, loader, ownedResources, sessionZone);
+    }
+
+    /**
+     * Create a cursor over a single pre-populated {@link VectorSchemaRoot}. The root (and any
+     * backing allocator wrapped in {@code ownedResources}) are owned by the cursor — closing the
+     * cursor closes the supplied {@code ownedResources}.
+     */
+    static ArrowStreamReaderCursor inMemory(VectorSchemaRoot root, AutoCloseable ownedResources, ZoneId sessionZone) {
+        // The VSR is already populated, so there is nothing more to load — the cursor walks the
+        // row count until exhausted and then reports end-of-stream.
+        return new ArrowStreamReaderCursor(root, () -> false, ownedResources, sessionZone);
+    }
+
+    private ArrowStreamReaderCursor(
+            VectorSchemaRoot root, BatchLoader batchLoader, AutoCloseable ownedResources, ZoneId sessionZone) {
+        this.root = root;
+        this.batchLoader = batchLoader;
+        this.ownedResources = ownedResources;
+        this.sessionZone = sessionZone;
     }
 
     List<QueryJDBCAccessor> createAccessors() {
-        return getSchemaRoot().getFieldVectors().stream()
+        return root.getFieldVectors().stream()
                 .map(rethrowFunction(this::createAccessor))
                 .collect(Collectors.toList());
     }
@@ -59,13 +106,13 @@ private QueryJDBCAccessor createAccessor(FieldVector vector) throws SQLException
      *
      * <p>Empty IPC batches are valid; Hyper can send an empty first chunk. JDBC has no notion of
      * a "batch": {@link java.sql.ResultSet#next()} only knows "is there another row?". This
-     * cursor is the single place that translates batch-level signals from {@link
-     * ArrowStreamReader#loadNextBatch()} into row-level advances, so it must consume the empty
-     * batches itself rather than push that responsibility outward.
+     * cursor is the single place that translates batch-level signals from {@link BatchLoader}
+     * into row-level advances, so it must consume the empty batches itself rather than push
+     * that responsibility outward.
      */
-    private boolean loadNextNonEmptyBatch() throws IOException {
-        while (reader.loadNextBatch()) {
-            if (getSchemaRoot().getRowCount() > 0) {
+    private boolean loadNextNonEmptyBatch() throws Exception {
+        while (batchLoader.loadNextBatch()) {
+            if (root.getRowCount() > 0) {
                 currentIndex.set(0);
                 return true;
             }
@@ -76,7 +123,7 @@ private boolean loadNextNonEmptyBatch() throws IOException {
     @SneakyThrows
     public boolean next() {
         val current = currentIndex.incrementAndGet();
-        val total = getSchemaRoot().getRowCount();
+        val total = root.getRowCount();
 
         try {
             val next = current < total || loadNextNonEmptyBatch();
@@ -97,6 +144,8 @@ public boolean next() {
     @SneakyThrows
     @Override
     public void close() {
-        reader.close();
+        if (ownedResources != null) {
+            ownedResources.close();
+        }
     }
 }

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

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

@@ -12,7 +12,7 @@
 import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
-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.protocol.data.ColumnMetadata;
 import com.salesforce.datacloud.jdbc.protocol.data.HyperType;
@@ -75,7 +75,7 @@ public static ResultSet createTableResultSet(
     }
 
     static ResultSet getMetadataResultSet(List<ColumnMetadata> columns, List<Object> data) throws SQLException {
-        return DataCloudMetadataResultSet.of(new DataCloudResultSetMetaData(columns), data);
+        return MetadataResultSets.ofRawRows(columns, data);
     }
 
     private static List<Object> constructTableData(ResultSet resultSet) throws SQLException {