refactor: centralize type mapping to fix metadata-path divergence (#174)

## Why

Column type information currently flows through three independent paths
— Arrow schema, `pg_catalog.format_type()`, and Spark — each with its
own per-site mapping logic. That scatter has caused repeated,
silently-diverging bugs: `pg_catalog` rows returning raw PostgreSQL OIDs
because `QueryMetadataUtil.dbTypeToSql` keyed on `int2`/`int4`/`int8`
aliases that Hyper never emits; `ResultSetMetaData` disagreeing with
`DatabaseMetaData.getColumns()` on type names and codes for the same
column; DECIMAL vs NUMERIC inconsistency between the two paths;
`getTypeInfo()` returning an empty result; engine-level concerns
(parameter binding, vector population) reaching into JDBC exception
types.

## Observable behavior changes

**New capabilities / bug fixes:**
- `DatabaseMetaData.getTypeInfo()` returns a real 18-column result
enumerated from `HyperTypeKind` (was empty before).
- Prepared-statement parameter Arrow round-trip: CHAR(1), CHAR(n), and
VARCHAR(n) now round-trip through Arrow losslessly via `hyper:type` /
`hyper:max_string_length` field metadata, matching the convention
Hyper's server already uses on the inbound side.
- `DatabaseMetaData.getColumns().COLUMN_SIZE` and `DECIMAL_DIGITS` are
now derived from the parsed type's declared precision/scale (e.g.
`numeric(10,5)` reports `COLUMN_SIZE=10, DECIMAL_DIGITS=5`; `char(1)`
reports `COLUMN_SIZE=1`) instead of being hardcoded to `255` / `2`.
- `DatabaseMetaData.getColumns().DATA_TYPE` is always a `java.sql.Types`
value (was a raw PG OID for unmapped types); unsupported types surface
as `Types.OTHER` with the raw name preserved via `HyperTypeKind.UNKNOWN`
rather than throwing.
- `DECIMAL_DIGITS` stays `0` for `FLOAT4` / `FLOAT8` (binary floats have
no meaningful decimal scale) instead of being hardcoded to `2`.

**Syntactic / naming changes:**
- `DatabaseMetaData.getColumns().TYPE_NAME` is consistently
JDBC-uppercase (`"BIGINT"`, `"INTEGER"`, `"DECIMAL"`, …) instead of a
mix of Hyper-lowercase and uppercase. `INTERVAL` and `JSON` show the
Hyper-uppercase form; unmodeled system-catalog types (e.g. `aclitem`)
show the raw Hyper name.
- Numeric/decimal unified to `DECIMAL` (both Arrow Decimal and Hyper
`numeric` map to `HyperTypeKind.DECIMAL` / `JDBCType.DECIMAL`).

## What's introduced

Introduce `HyperType`, a Hyper-native internal type model, and route
every Arrow, `pg_catalog`, prepared-statement, and Spark conversion
through it. The data layer lives under a new
`com.salesforce.datacloud.jdbc.protocol.data` package that has no JDBC
dependency, so engine-level callers can use it without a JDBC import;
the JDBC-bound helpers stay in
`com.salesforce.datacloud.jdbc.core.types.HyperTypes`. Translation to
`java.sql.JDBCType` / `Types` only happens at the JDBC API boundary.

**New in `protocol.data` (JDBC-free):**
- `HyperTypeKind` + `HyperType` value class with a visitor and native
`toHyperTypeName()`. `HyperTypeKind.UNKNOWN` preserves the raw Hyper
type name for system-catalog types the driver doesn't model (e.g.
`aclitem`).
- `PgCatalogTypeParser`, `ArrowToHyperTypeMapper`, `HyperTypeToArrow`
(format-boundary converters).
- `ColumnMetadata` with optional `typeName` override.
- `ParameterBinding` + `ParameterAccumulator` (replaces
`ParameterManager`); throws `IllegalArgumentException` so it is usable
without JDBC.
- `ArrowUtils`, `VectorPopulator` (moved from `.util`).

**`HyperTypes` stays in `core.types`** as the JDBC-bound boundary:
`toJdbcType*`, `fromJdbcTypeCode`, `toJavaClass`,
`getPrecision`/`getScale`/`getDisplaySize`,
`isSigned`/`isCaseSensitive`/`isNullable`, `needsDecimalDigits`,
`needsCharOctetLength`, `typeInfoRows`, plus the `isIntegerLike` /
`isStringLike` helpers that wrap exhaustive enum switches so callers can
do a family check without replicating the switch.

## Structural details

**Deleted:** `ColumnType`, `ArrowToColumnTypeMapper`, the
`QueryMetadataUtil.dbTypeToSql` + `isDecimalType`/`isCharType` helpers,
the `ArrowUtils.SQL_TYPE_TO_FIELD_TYPE` map.

**Migrated callers:** `ColumnMetadata` holds `HyperType`;
`DataCloudResultSetMetaData`, `StreamingResultSet`, `DataCloudArray`,
`DataCloudConnection.getSchemaForQueryId`, and
`ArrowUtils.toColumnMetaData` go through `ArrowToHyperTypeMapper` +
`HyperTypes`; `QueryMetadataUtil.constructColumnData` parses via
`PgCatalogTypeParser` and falls back to `HyperType.unknown(raw)` for
unmodeled system-catalog types; `SimpleResultSet` switches on
`HyperTypeKind` via the new helpers (and picks up TINYINT numeric
coercions and the `getFloat` wasNull/range-check fix that were planned
as Phase 0 work); `DataCloudPreparedStatement` wraps
`IllegalArgumentException` from `ParameterAccumulator` into
`SQLException`; `VectorPopulator` is keyed by `HyperTypeKind`; Spark
`TypeMapping` handles TINYINT, NUMERIC, CHAR, NULL, FLOAT, NCHAR,
NVARCHAR, LONGNVARCHAR.

**Follow-ups noted in code / comments (not in this PR):**
- `SimpleResultSet` is documented for removal once
`DataCloudMetadataResultSet` migrates to `StreamingResultSet` — one
result-set implementation instead of two.
- `QueryResultArrowStream.asArrowStreamReader`'s `RootAllocator`
lifetime needs a dedicated fix (it flows into `ArrowStreamReader`, which
doesn't close it).
- `TIME WITH TIME ZONE` Arrow round-trip is pinned as intended asymmetry
— offset honored on input but not preserved on output, matching Hyper's
read-side behavior. Proper round-trip would need either a synthetic
Arrow schema or Hyper-protocol coordination.

Author: mkaufmann

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

@@ -4,7 +4,7 @@
  */
 package com.salesforce.datacloud.jdbc.core;
 
-import com.salesforce.datacloud.jdbc.core.metadata.ColumnMetadata;
+import com.salesforce.datacloud.jdbc.protocol.data.ColumnMetadata;
 import java.sql.SQLException;
 import java.util.HashMap;
 import java.util.List;

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

@@ -7,10 +7,9 @@
 import static com.salesforce.datacloud.jdbc.config.DriverVersion.formatDriverInfo;
 import static com.salesforce.datacloud.jdbc.exception.QueryExceptionHandler.createException;
 import static com.salesforce.datacloud.jdbc.logging.ElapsedLogger.logTimedValue;
-import static com.salesforce.datacloud.jdbc.util.ArrowUtils.toColumnMetaData;
+import static com.salesforce.datacloud.jdbc.protocol.data.ArrowUtils.toColumnMetaData;
 
 import com.google.protobuf.Empty;
-import com.salesforce.datacloud.jdbc.core.metadata.ColumnMetadata;
 import com.salesforce.datacloud.jdbc.core.metadata.DataCloudResultSetMetaData;
 import com.salesforce.datacloud.jdbc.core.partial.DataCloudQueryPolling;
 import com.salesforce.datacloud.jdbc.exception.DataCloudJDBCException;
@@ -22,6 +21,7 @@
 import com.salesforce.datacloud.jdbc.protocol.RowRangeIterator;
 import com.salesforce.datacloud.jdbc.protocol.async.core.AsyncStreamObserverIterator;
 import com.salesforce.datacloud.jdbc.protocol.async.core.SyncIteratorAdapter;
+import com.salesforce.datacloud.jdbc.protocol.data.ColumnMetadata;
 import com.salesforce.datacloud.jdbc.protocol.grpc.QueryAccessGrpcClient;
 import com.salesforce.datacloud.jdbc.util.Deadline;
 import com.salesforce.datacloud.jdbc.util.JdbcURL;
@@ -182,7 +182,7 @@ public PreparedStatement prepareStatement(String sql) {
     }
 
     private DataCloudPreparedStatement getQueryPreparedStatement(String sql) {
-        return new DataCloudPreparedStatement(this, sql, new DefaultParameterManager());
+        return new DataCloudPreparedStatement(this, sql);
     }
 
     /**
@@ -281,6 +281,12 @@ public ResultSetMetaData getSchemaForQueryId(String queryId) throws SQLException
             return new DataCloudResultSetMetaData(columns);
         } catch (StatusRuntimeException ex) {
             throw createException(connectionProperties.isIncludeCustomerDetailInReason(), null, queryId, ex);
+        } catch (IllegalArgumentException ex) {
+            // Thrown by ArrowToHyperTypeMapper for Arrow types the driver does not model.
+            throw new SQLException(
+                    "Unsupported column type in query schema for queryId=" + queryId + ": " + ex.getMessage(),
+                    "0A000",
+                    ex);
         }
     }
 

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

@@ -13,6 +13,8 @@
 
 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.types.HyperTypes;
 import com.salesforce.datacloud.jdbc.util.JdbcURL;
 import com.salesforce.datacloud.jdbc.util.ThrowingJdbcSupplier;
 import java.sql.Connection;
@@ -753,7 +755,8 @@ public ResultSet getCrossReference(
 
     @Override
     public ResultSet getTypeInfo() throws SQLException {
-        return DataCloudMetadataResultSet.empty();
+        return DataCloudMetadataResultSet.of(
+                new DataCloudResultSetMetaData(MetadataSchemas.TYPE_INFO), HyperTypes.typeInfoRows());
     }
 
     @Override

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

@@ -4,10 +4,10 @@
  */
 package com.salesforce.datacloud.jdbc.core;
 
-import com.salesforce.datacloud.jdbc.core.metadata.ColumnMetadata;
 import com.salesforce.datacloud.jdbc.core.metadata.DataCloudResultSetMetaData;
 import com.salesforce.datacloud.jdbc.core.resultset.ColumnAccessor;
 import com.salesforce.datacloud.jdbc.core.resultset.SimpleResultSet;
+import com.salesforce.datacloud.jdbc.protocol.data.ColumnMetadata;
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.Collections;

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

@@ -4,13 +4,15 @@
  */
 package com.salesforce.datacloud.jdbc.core;
 
-import static com.salesforce.datacloud.jdbc.util.ArrowUtils.toArrowByteArray;
+import static com.salesforce.datacloud.jdbc.protocol.data.ArrowUtils.toArrowByteArray;
 import static com.salesforce.datacloud.jdbc.util.DateTimeUtils.getUTCDateFromDateAndCalendar;
 import static com.salesforce.datacloud.jdbc.util.DateTimeUtils.getUTCTimeFromTimeAndCalendar;
 
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
 import com.google.protobuf.ByteString;
+import com.salesforce.datacloud.jdbc.protocol.data.HyperType;
+import com.salesforce.datacloud.jdbc.protocol.data.ParameterAccumulator;
 import com.salesforce.datacloud.jdbc.util.QueryTimeout;
 import com.salesforce.datacloud.jdbc.util.SqlErrorCodes;
 import java.io.IOException;
@@ -51,25 +53,28 @@
 @Slf4j
 public class DataCloudPreparedStatement extends DataCloudStatement implements PreparedStatement {
     private String sql;
-    private final ParameterManager parameterManager;
+    // Package-private for tests that need to inspect bound parameters.
+    final ParameterAccumulator parameters = new ParameterAccumulator();
     private final Calendar calendar = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
     // True if we are currently fetching metadata from the server, this influences the query param generation
     // to not return any data.
     private boolean fetchingMetadata = false;
 
-    DataCloudPreparedStatement(DataCloudConnection connection, ParameterManager parameterManager) {
+    DataCloudPreparedStatement(DataCloudConnection connection) {
         super(connection);
-        this.parameterManager = parameterManager;
     }
 
-    DataCloudPreparedStatement(DataCloudConnection connection, String sql, ParameterManager parameterManager) {
+    DataCloudPreparedStatement(DataCloudConnection connection, String sql) {
         super(connection);
         this.sql = sql;
-        this.parameterManager = parameterManager;
     }
 
-    private <T> void setParameter(int parameterIndex, int sqlType, T value) throws SQLException {
-        parameterManager.setParameter(parameterIndex, sqlType, value);
+    private <T> void setParameter(int parameterIndex, HyperType type, T value) throws SQLException {
+        try {
+            parameters.setParameter(parameterIndex, type, value);
+        } catch (IllegalArgumentException ex) {
+            throw new SQLException(ex.getMessage(), ex);
+        }
     }
 
     @Override
@@ -91,9 +96,13 @@ protected QueryParam.Builder getQueryParamBuilder(
 
         final byte[] encodedRow;
         try {
-            encodedRow = toArrowByteArray(parameterManager.getParameters(), calendar);
+            encodedRow = toArrowByteArray(parameters.getParameters(), calendar);
         } catch (IOException e) {
             throw new SQLException("Failed to encode parameters on prepared statement", e);
+        } catch (IllegalArgumentException e) {
+            // Thrown when a parameter is bound with a HyperType we cannot currently encode as
+            // Arrow (e.g. INTERVAL, JSON). Surface as a JDBC spec-compliant SQLException.
+            throw new SQLException("Failed to encode parameters on prepared statement: " + e.getMessage(), "HY000", e);
         }
 
         if (fetchingMetadata) {
@@ -131,52 +140,67 @@ public int executeUpdate() throws SQLException {
 
     @Override
     public void setNull(int parameterIndex, int sqlType) throws SQLException {
-        setParameter(parameterIndex, sqlType, null);
+        setParameter(parameterIndex, hyperTypeForJdbcCode(sqlType), null);
+    }
+
+    private static HyperType hyperTypeForJdbcCode(int sqlType) throws SQLException {
+        try {
+            return com.salesforce.datacloud.jdbc.core.types.HyperTypes.fromJdbcTypeCode(sqlType, true);
+        } catch (IllegalArgumentException ex) {
+            throw new SQLException("Unsupported JDBC type code: " + sqlType, "HYC00", ex);
+        }
     }
 
     @Override
     public void setBoolean(int parameterIndex, boolean x) throws SQLException {
-        setParameter(parameterIndex, Types.BOOLEAN, x);
+        setParameter(parameterIndex, HyperType.bool(false), x);
     }
 
     @Override
     public void setByte(int parameterIndex, byte x) throws SQLException {
-        setParameter(parameterIndex, Types.TINYINT, x);
+        setParameter(parameterIndex, HyperType.int8(false), x);
     }
 
     @Override
     public void setShort(int parameterIndex, short x) throws SQLException {
-        setParameter(parameterIndex, Types.SMALLINT, x);
+        setParameter(parameterIndex, HyperType.int16(false), x);
     }
 
     @Override
     public void setInt(int parameterIndex, int x) throws SQLException {
-        setParameter(parameterIndex, Types.INTEGER, x);
+        setParameter(parameterIndex, HyperType.int32(false), x);
     }
 
     @Override
     public void setLong(int parameterIndex, long x) throws SQLException {
-        setParameter(parameterIndex, Types.BIGINT, x);
+        setParameter(parameterIndex, HyperType.int64(false), x);
     }
 
     @Override
     public void setFloat(int parameterIndex, float x) throws SQLException {
-        setParameter(parameterIndex, Types.FLOAT, x);
+        setParameter(parameterIndex, HyperType.float8(false), x);
     }
 
     @Override
     public void setDouble(int parameterIndex, double x) throws SQLException {
-        setParameter(parameterIndex, Types.DOUBLE, x);
+        setParameter(parameterIndex, HyperType.float8(false), x);
     }
 
     @Override
     public void setBigDecimal(int parameterIndex, BigDecimal x) throws SQLException {
-        setParameter(parameterIndex, Types.DECIMAL, x);
+        final HyperType type;
+        if (x == null) {
+            // Precision/scale are unknown; let Arrow encoding pick a suitable default.
+            type = HyperType.decimal(0, 0, true);
+        } else {
+            type = HyperType.decimal(x.precision(), x.scale(), true);
+        }
+        setParameter(parameterIndex, type, x);
     }
 
     @Override
     public void setString(int parameterIndex, String x) throws SQLException {
-        setParameter(parameterIndex, Types.VARCHAR, x);
+        setParameter(parameterIndex, HyperType.varcharUnlimited(true), x);
     }
 
     @Override
@@ -186,17 +210,17 @@ public void setBytes(int parameterIndex, byte[] x) throws SQLException {
 
     @Override
     public void setDate(int parameterIndex, Date x) throws SQLException {
-        setParameter(parameterIndex, Types.DATE, x);
+        setParameter(parameterIndex, HyperType.date(true), x);
     }
 
     @Override
     public void setTime(int parameterIndex, Time x) throws SQLException {
-        setParameter(parameterIndex, Types.TIME, x);
+        setParameter(parameterIndex, HyperType.time(true), x);
     }
 
     @Override
     public void setTimestamp(int parameterIndex, Timestamp x) throws SQLException {
-        setParameter(parameterIndex, Types.TIMESTAMP, toWallClockAsUtc(x, null));
+        setParameter(parameterIndex, HyperType.timestamp(true), toWallClockAsUtc(x, null));
     }
 
     @Override
@@ -216,7 +240,7 @@ public void setBinaryStream(int parameterIndex, InputStream x, int length) throw
 
     @Override
     public void clearParameters() {
-        parameterManager.clearParameters();
+        parameters.clearParameters();
     }
 
     @Override
@@ -229,17 +253,17 @@ public void setObject(int parameterIndex, Object x, int targetSqlType) throws SQ
         // or store LocalDateTime digits directly.
         if (targetSqlType == Types.TIMESTAMP) {
             if (x instanceof Timestamp) {
-                setParameter(parameterIndex, Types.TIMESTAMP, toWallClockAsUtc((Timestamp) x, null));
+                setParameter(parameterIndex, HyperType.timestamp(true), toWallClockAsUtc((Timestamp) x, null));
                 return;
             }
             if (x instanceof LocalDateTime) {
                 LocalDateTime ldt = (LocalDateTime) x;
                 // Encode LDT digits as if they were in UTC — no JVM-TZ shift applied.
-                setParameter(parameterIndex, Types.TIMESTAMP, Timestamp.from(ldt.toInstant(ZoneOffset.UTC)));
+                setParameter(parameterIndex, HyperType.timestamp(true), Timestamp.from(ldt.toInstant(ZoneOffset.UTC)));
                 return;
             }
         }
-        setParameter(parameterIndex, targetSqlType, x);
+        setParameter(parameterIndex, hyperTypeForJdbcCode(targetSqlType), x);
     }
 
     @Override
@@ -307,18 +331,18 @@ public ResultSetMetaData getMetaData() throws SQLException {
     @Override
     public void setDate(int parameterIndex, Date x, Calendar cal) throws SQLException {
         val utcDate = getUTCDateFromDateAndCalendar(x, cal);
-        setParameter(parameterIndex, Types.DATE, utcDate);
+        setParameter(parameterIndex, HyperType.date(true), utcDate);
     }
 
     @Override
     public void setTime(int parameterIndex, Time x, Calendar cal) throws SQLException {
         val utcTime = getUTCTimeFromTimeAndCalendar(x, cal);
-        setParameter(parameterIndex, Types.TIME, utcTime);
+        setParameter(parameterIndex, HyperType.time(true), utcTime);
     }
 
     @Override
     public void setTimestamp(int parameterIndex, Timestamp x, Calendar cal) throws SQLException {
-        setParameter(parameterIndex, Types.TIMESTAMP, toWallClockAsUtc(x, cal));
+        setParameter(parameterIndex, HyperType.timestamp(true), toWallClockAsUtc(x, cal));
     }
 
     /**