feat: add rowMode option to keep columns with duplicate names

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Unreleased
 
+- Add the `rowMode` option to `executeStatement()`: `rowMode: 'array'` returns each row as an array of values in column order, so columns with the same name (e.g. `SELECT carat AS a, color AS a`) are no longer lost; the default `'object'` rows are unchanged, and the driver now logs a warning when a result has duplicate column names in object mode. (#118)
 - CloudFetch retry-exhaustion errors now identify the download and the last failure: the `RetryError` message reads `CloudFetch result download failed: <original message> (last error: <cause>)`. `<cause>` is `HTTP <status>: <statusText>` (or `HTTP <status>` when the status text is empty) for HTTP failures, and `network error <code>`, `network error (<type>)` (e.g. `aborted`, when there is no `code`) or a bare `network error` for network failures (the raw network-error message, which can contain the presigned URL, is not copied). If no cause can be determined, the ` (last error: …)` suffix is omitted. The `RetryError` instance, `errorCode` and `payload` (including the node-fetch `Request` / `Response`) are unchanged. (#471)
 - Report untyped NULL columns as STRING in kernel result metadata, matching Thrift. (#470)
 - Reject non-scalar query parameter values at bind time with a `ParameterError` on both the Thrift and kernel backends, instead of stringifying them (which could leave a Thrift operation stuck before FINISHED). Newly rejected on the Thrift backend: arrays, plain objects, boxed primitives (e.g. `new String('x')`, `new Number(1)`), functions, symbols, and a `DBSQLParameter` that wraps another `DBSQLParameter`. Primitives, `Date`, `Int64`, and `DBSQLParameter` are still accepted.

lib/DBSQLSession.ts

@@ -29,6 +29,8 @@ import ISessionBackend from './contracts/ISessionBackend';
 import IOperationBackend from './contracts/IOperationBackend';
 import { numberToInt64 as numberToInt64Impl } from './thrift-backend/ThriftSessionBackend';
 import { safeEmit } from './telemetry/telemetryUtils';
+import { ResultMetadata } from './contracts/ResultMetadata';
+import { getSchemaColumns } from './result/utils';
 
 // Explicitly promisify a callback-style `pipeline` because `node:stream/promises` is not available in Node 14
 const pipeline = util.promisify(stream.pipeline);
@@ -91,6 +93,9 @@ export default class DBSQLSession implements IDBSQLSession {
    * const operation = await session.executeStatement(query);
    */
   public async executeStatement(statement: string, options: ExecuteStatementOptions = {}): Promise<IOperation> {
+    if (options.rowMode !== undefined && options.rowMode !== 'object' && options.rowMode !== 'array') {
+      throw new HiveDriverError(`Unsupported rowMode: ${options.rowMode}. Expected 'object' or 'array'.`);
+    }
     const opBackend = await this.runBackend(() => this.backend.executeStatement(statement, options));
     const operation = this.wrapOperation(opBackend);
 
@@ -107,13 +112,17 @@ export default class DBSQLSession implements IDBSQLSession {
         const allowedLocalPath = Array.isArray(options.stagingAllowedLocalPath)
           ? options.stagingAllowedLocalPath
           : [options.stagingAllowedLocalPath];
-        return this.handleStagingOperation(operation, allowedLocalPath);
+        return this.handleStagingOperation(operation, allowedLocalPath, metadata);
       }
     }
     return operation;
   }
 
-  private async handleStagingOperation(operation: IOperation, allowedLocalPath: Array<string>): Promise<IOperation> {
+  private async handleStagingOperation(
+    operation: IOperation,
+    allowedLocalPath: Array<string>,
+    metadata: ResultMetadata,
+  ): Promise<IOperation> {
     type StagingResponse = {
       presignedUrl: string;
       localFile?: string;
@@ -124,7 +133,15 @@ export default class DBSQLSession implements IDBSQLSession {
     if (rows.length !== 1) {
       throw new StagingError('Staging operation: expected only one row in result');
     }
-    const row = rows[0] as StagingResponse;
+    let row = rows[0] as StagingResponse;
+    // `rowMode: 'array'` returns positional rows; key them by column name.
+    if (Array.isArray(row)) {
+      const values: Array<unknown> = row;
+      row = getSchemaColumns(metadata.schema).reduce(
+        (result, column, i) => ({ ...result, [column.columnName]: values[i] }),
+        {} as StagingResponse,
+      );
+    }
 
     if (row.localFile !== undefined) {
       let allowOperation = false;

lib/contracts/IDBSQLSession.ts

@@ -4,6 +4,15 @@ import Status from '../dto/Status';
 import InfoValue from '../dto/InfoValue';
 import { DBSQLParameter, DBSQLParameterValue } from '../DBSQLParameter';
 
+/**
+ * Shape of the rows returned by a statement's fetch methods:
+ * - `'object'` (default): each row is an object keyed by column name. Columns
+ *   that share a name (e.g. `SELECT a.id, b.id`) overwrite each other.
+ * - `'array'`: each row is an array of values in result-schema order, so every
+ *   column is kept. Use `getResultMetadata()` / `getSchema()` for column names.
+ */
+export type RowMode = 'object' | 'array';
+
 export type ExecuteStatementOptions = {
   /**
    * The number of seconds after which the query will time out on the server.
@@ -50,6 +59,11 @@ export type ExecuteStatementOptions = {
    * `query_tags` key). Ignored by the Thrift backend.
    */
   statementConf?: Record<string, string>;
+  /**
+   * Row shape returned by this statement's `fetchChunk` / `fetchAll` / iterators.
+   * Defaults to `'object'`; see {@link RowMode}.
+   */
+  rowMode?: RowMode;
 };
 
 export type TypeInfoRequest = {

lib/index.ts

@@ -22,6 +22,7 @@ import { LogLevel } from './contracts/IDBSQLLogger';
 
 // Re-export types for TypeScript users
 export type { default as ITokenProvider } from './connection/auth/tokenProvider/ITokenProvider';
+export type { RowMode } from './contracts/IDBSQLSession';
 
 export { LOG_FILE_UNAVAILABLE_WARNING_CODE } from './DBSQLLogger';
 export { CircuitBreakerOpenError, CIRCUIT_BREAKER_OPEN_CODE } from './telemetry/CircuitBreaker';

lib/kernel/KernelOperationBackend.ts

@@ -43,6 +43,7 @@ import { OperationStatus, OperationState } from '../contracts/OperationStatus';
 import { ResultMetadata, ResultFormat } from '../contracts/ResultMetadata';
 import IClientContext from '../contracts/IClientContext';
 import { LogLevel } from '../contracts/IDBSQLLogger';
+import { RowMode } from '../contracts/IDBSQLSession';
 import Status from '../dto/Status';
 import HiveDriverError from '../errors/HiveDriverError';
 import OperationStateError, { OperationStateErrorCode } from '../errors/OperationStateError';
@@ -167,6 +168,8 @@ export interface KernelOperationBackendOptions {
    * handle exposes one, else a fresh UUIDv4.
    */
   id?: string;
+  /** Row shape returned by `fetchChunk`; defaults to `'object'`. */
+  rowMode?: RowMode;
 }
 
 export default class KernelOperationBackend implements IOperationBackend {
@@ -202,6 +205,8 @@ export default class KernelOperationBackend implements IOperationBackend {
 
   private readonly lifecycle: KernelOperationLifecycleState = createLifecycleState();
 
+  private readonly rowMode?: RowMode;
+
   private resultSlicer?: ResultSlicer<any>;
 
   private resultsProvider?: KernelResultsProvider;
@@ -215,7 +220,14 @@ export default class KernelOperationBackend implements IOperationBackend {
   // already-terminal statement. Drives both fetch and result-metadata.
   private fetchHandlePromise?: Promise<KernelFetchHandle>;
 
-  constructor({ asyncStatement, statement, cancellableExecution, context, id }: KernelOperationBackendOptions) {
+  constructor({
+    asyncStatement,
+    statement,
+    cancellableExecution,
+    context,
+    id,
+    rowMode,
+  }: KernelOperationBackendOptions) {
     // Exactly one of the three handle kinds must be supplied.
     const providedCount =
       (asyncStatement !== undefined ? 1 : 0) +
@@ -244,6 +256,7 @@ export default class KernelOperationBackend implements IOperationBackend {
         }
       : ((asyncStatement ?? statement) as KernelStatementHandle);
     this.context = context;
+    this.rowMode = rowMode;
     this._id =
       id ?? asyncStatement?.statementId ?? statement?.statementId ?? cancellableExecution?.statementId ?? uuidv4();
   }
@@ -789,6 +802,7 @@ export default class KernelOperationBackend implements IOperationBackend {
     const converter = new ArrowResultConverter(this.context, this.resultsProvider, metadata, {
       preserveBigNumericPrecision: this.context.getConfig().preserveBigNumericPrecision ?? false,
       disableRowMaterialization: this.context.getConfig().disableRowMaterialization ?? false,
+      rowMode: this.rowMode,
     });
     this.resultSlicer = new ResultSlicer(this.context, converter);
     return this.resultSlicer;

lib/kernel/KernelSessionBackend.ts

@@ -257,9 +257,9 @@ export default class KernelSessionBackend implements ISessionBackend {
       // Feature-detect the arm via a type guard: only the Running `AsyncStatement`
       // exposes `awaitResult`; the terminal `Statement` (Completed arm) does not.
       if (isKernelAsyncStatement(direct)) {
-        return new KernelOperationBackend({ asyncStatement: direct, context: this.context });
+        return new KernelOperationBackend({ asyncStatement: direct, context: this.context, rowMode: options.rowMode });
       }
-      return new KernelOperationBackend({ statement: direct, context: this.context });
+      return new KernelOperationBackend({ statement: direct, context: this.context, rowMode: options.rowMode });
     }
 
     // Async path: submit (`wait_timeout=0s`) returns a pending handle the
@@ -279,6 +279,7 @@ export default class KernelSessionBackend implements ISessionBackend {
     return new KernelOperationBackend({
       asyncStatement: asyncStatement!,
       context: this.context,
+      rowMode: options.rowMode,
     });
   }
 

lib/result/ArrowResultConverter.ts

@@ -17,9 +17,10 @@ import {
 } from 'apache-arrow';
 import { TTableSchema, TColumnDesc, TTypeId } from '../../thrift/TCLIService_types';
 import IClientContext from '../contracts/IClientContext';
+import { RowMode } from '../contracts/IDBSQLSession';
 import HiveDriverError from '../errors/HiveDriverError';
 import IResultsProvider, { ResultsProviderFetchNextOptions } from './IResultsProvider';
-import { ArrowBatch, getSchemaColumns, convertThriftValue } from './utils';
+import { ArrowBatch, getSchemaColumns, convertThriftValue, warnOnDuplicateColumnNames } from './utils';
 
 const { isArrowBigNumSymbol, bigNumToBigInt } = arrowUtils;
 
@@ -208,6 +209,8 @@ export default class ArrowResultConverter implements IResultsProvider<Array<any>
   // (only the row count is meaningful). See `ConnectionOptions.disableRowMaterialization`.
   private readonly disableRowMaterialization: boolean;
 
+  private readonly rowMode: RowMode;
+
   private recordBatchReader?: IterableIterator<RecordBatch<TypeMap>>;
 
   // Remaining rows in current Arrow batch (not the record batch!)
@@ -232,13 +235,18 @@ export default class ArrowResultConverter implements IResultsProvider<Array<any>
     {
       preserveBigNumericPrecision = false,
       disableRowMaterialization = false,
-    }: { preserveBigNumericPrecision?: boolean; disableRowMaterialization?: boolean } = {},
+      rowMode = 'object',
+    }: { preserveBigNumericPrecision?: boolean; disableRowMaterialization?: boolean; rowMode?: RowMode } = {},
   ) {
     this.context = context;
     this.source = source;
     this.schema = getSchemaColumns(schema);
     this.preserveBigNumericPrecision = preserveBigNumericPrecision;
     this.disableRowMaterialization = disableRowMaterialization;
+    this.rowMode = rowMode;
+    if (rowMode === 'object' && !disableRowMaterialization) {
+      warnOnDuplicateColumnNames(context, this.schema);
+    }
   }
 
   public async hasMore() {
@@ -268,15 +276,21 @@ export default class ArrowResultConverter implements IResultsProvider<Array<any>
 
     if (this.prefetchedRecordBatch) {
       // Consume a record batch fetched during previous call to `fetchNext`
-      const table = new Table(this.prefetchedRecordBatch);
+      const batch = this.prefetchedRecordBatch;
       this.prefetchedRecordBatch = undefined;
       // Get table rows, but not more than remaining count. When materialization
       // is disabled, skip `toArray()`/`getRows()` and return `null` placeholders
       // (filled, not sparse, so `fetchAll`'s `.flat()` keeps the count).
-      const batchRows = Math.min(table.numRows, this.remainingRows);
-      const result = this.disableRowMaterialization
-        ? new Array(batchRows).fill(null)
-        : this.getRows(table.schema, table.toArray().slice(0, this.remainingRows));
+      const batchRows = Math.min(batch.numRows, this.remainingRows);
+      let result: Array<any>;
+      if (this.disableRowMaterialization) {
+        result = new Array(batchRows).fill(null);
+      } else if (this.rowMode === 'array') {
+        result = this.getRowArrays(batch, batchRows);
+      } else {
+        const table = new Table(batch);
+        result = this.getRows(table.schema, table.toArray().slice(0, this.remainingRows));
+      }
 
       // Reduce remaining rows count by a count of rows we just processed.
       // If the remaining count reached zero - we're done with current arrow
@@ -345,6 +359,29 @@ export default class ArrowResultConverter implements IResultsProvider<Array<any>
     });
   }
 
+  // Reads cells by column index, so same-named columns are all kept. Fields come from the
+  // batch data type: `RecordBatch.schema` merges same-named fields into one type.
+  private getRowArrays(batch: RecordBatch<TypeMap>, rowCount: number): Array<Array<any>> {
+    const fields: Array<ArrowSchemaField> = batch.data.type.children;
+    const vectors = this.schema.map((column, index) => batch.getChildAt(index));
+    const rows: Array<Array<any>> = [];
+    for (let i = 0; i < rowCount; i += 1) {
+      const row: Array<any> = [];
+      for (let j = 0; j < this.schema.length; j += 1) {
+        const field = fields[j];
+        const value = this.convertArrowTypes(
+          vectors[j]?.get(i) ?? null,
+          field?.type,
+          field?.type.children || [],
+          field,
+        );
+        row.push(this.convertThriftColumnValue(this.schema[j], value));
+      }
+      rows.push(row);
+    }
+    return rows;
+  }
+
   private convertArrowTypes(
     value: any,
     valueType: DataType | undefined,
@@ -489,4 +526,20 @@ export default class ArrowResultConverter implements IResultsProvider<Array<any>
 
     return result;
   }
+
+  // Per-cell equivalent of `convertThriftTypes`, used by array rows. Kept separate so
+  // the default object path stays as is.
+  private convertThriftColumnValue(column: TColumnDesc, value: any): any {
+    if (value === null) {
+      return null;
+    }
+    const typeDescriptor = column.typeDesc.types[0]?.primitiveEntry;
+    if (
+      this.preserveBigNumericPrecision &&
+      (typeDescriptor?.type === TTypeId.DECIMAL_TYPE || typeDescriptor?.type === TTypeId.BIGINT_TYPE)
+    ) {
+      return value;
+    }
+    return convertThriftValue(typeDescriptor, value);
+  }
 }

lib/result/JsonResultHandler.ts

@@ -1,7 +1,8 @@
 import { TGetResultSetMetadataResp, TRowSet, TColumn, TColumnDesc } from '../../thrift/TCLIService_types';
 import IClientContext from '../contracts/IClientContext';
+import { RowMode } from '../contracts/IDBSQLSession';
 import IResultsProvider, { ResultsProviderFetchNextOptions } from './IResultsProvider';
-import { getSchemaColumns, convertThriftValue, getColumnValue } from './utils';
+import { getSchemaColumns, convertThriftValue, getColumnValue, warnOnDuplicateColumnNames } from './utils';
 
 export default class JsonResultHandler implements IResultsProvider<Array<any>> {
   private readonly context: IClientContext;
@@ -10,14 +11,21 @@ export default class JsonResultHandler implements IResultsProvider<Array<any>> {
 
   private readonly schema: Array<TColumnDesc>;
 
+  private readonly rowMode: RowMode;
+
   constructor(
     context: IClientContext,
     source: IResultsProvider<TRowSet | undefined>,
     { schema }: TGetResultSetMetadataResp,
+    { rowMode = 'object' }: { rowMode?: RowMode } = {},
   ) {
     this.context = context;
     this.source = source;
     this.schema = getSchemaColumns(schema);
+    this.rowMode = rowMode;
+    if (rowMode === 'object') {
+      warnOnDuplicateColumnNames(context, this.schema);
+    }
   }
 
   public async hasMore() {
@@ -35,7 +43,20 @@ export default class JsonResultHandler implements IResultsProvider<Array<any>> {
     }
 
     const columns = data.columns || [];
-    return this.getRows(columns, this.schema);
+    return this.rowMode === 'array' ? this.getRowArrays(columns, this.schema) : this.getRows(columns, this.schema);
+  }
+
+  private getRowArrays(columns: Array<TColumn>, descriptors: Array<TColumnDesc>): Array<Array<any>> {
+    const rows: Array<Array<any>> = [];
+    descriptors.forEach((descriptor, columnIndex) => {
+      this.getSchemaValues(descriptor, columns[descriptor.position - 1]).forEach((value, i) => {
+        if (!rows[i]) {
+          rows[i] = [];
+        }
+        rows[i][columnIndex] = value;
+      });
+    });
+    return rows;
   }
 
   private getRows(columns: Array<TColumn>, descriptors: Array<TColumnDesc>): Array<any> {

lib/result/utils.ts

@@ -18,6 +18,8 @@ import {
 } from 'apache-arrow';
 import { TTableSchema, TColumnDesc, TPrimitiveTypeEntry, TTypeId, TColumn } from '../../thrift/TCLIService_types';
 import HiveDriverError from '../errors/HiveDriverError';
+import IClientContext from '../contracts/IClientContext';
+import { LogLevel } from '../contracts/IDBSQLLogger';
 
 export interface ArrowBatch {
   batches: Array<Buffer>;
@@ -32,6 +34,28 @@ export function getSchemaColumns(schema?: TTableSchema): Array<TColumnDesc> {
   return [...schema.columns].sort((c1, c2) => c1.position - c2.position);
 }
 
+// Object rows are keyed by column name, so same-named columns overwrite each other.
+export function warnOnDuplicateColumnNames(context: IClientContext, columns: Array<TColumnDesc>): void {
+  const seen = new Set<string>();
+  const duplicates = new Set<string>();
+  for (const { columnName } of columns) {
+    if (seen.has(columnName)) {
+      duplicates.add(columnName);
+    }
+    seen.add(columnName);
+  }
+  if (duplicates.size > 0) {
+    const names = [...duplicates].map((name) => JSON.stringify(name)).join(', ');
+    context
+      .getLogger()
+      .log(
+        LogLevel.warn,
+        `Result has duplicate column names (${names}); object rows keep only the last column with each name. ` +
+          "Alias the columns or pass `rowMode: 'array'` to executeStatement to keep every column.",
+      );
+  }
+}
+
 function isString(value: unknown): value is string {
   return typeof value === 'string' || value instanceof String;
 }