diff --git a/benchmarks/tables.ts b/benchmarks/tables.ts
index bade37b..66b24be 100644
--- a/benchmarks/tables.ts
+++ b/benchmarks/tables.ts
@@ -1,4 +1,7 @@
-import { QWP_COLUMN_TYPE, QwpTableBuffer } from "../packages/client-core/src/_qwp/_core";
+import {
+  QWP_COLUMN_TYPE,
+  QwpTableBuffer,
+} from "../packages/client-core/src/_qwp/_core";
 import type { BenchmarkRow } from "./workloads";
 
 export function buildBenchmarkTable(
diff --git a/packages/client-core/src/_qwp/_core/result-batch.ts b/packages/client-core/src/_qwp/_core/result-batch.ts
index c386962..fc5cced 100644
--- a/packages/client-core/src/_qwp/_core/result-batch.ts
+++ b/packages/client-core/src/_qwp/_core/result-batch.ts
@@ -13,7 +13,7 @@ import {
 } from "./constants";
 import { QwpResultBatchMessage } from "./egress";
 import { QwpProtocolError } from "./errors";
-import { readQwpVarint } from "./varint";
+import { readQwpVarintSmall } from "./varint-number";
 import { decompressQwpZstdFrame } from "./zstd";
 
 // One axis bound for both directions: a decoder that accepted an axis the
@@ -98,8 +98,19 @@ export class QwpResultBatch {
   }
 
   *rows(): IterableIterator<readonly QwpResultValue[]> {
+    // Look each column's values up once per batch, not once per cell.
+    const columns = this.columns;
+    const width = columns.length;
+    const values = new Array<readonly QwpResultValue[]>(width);
+    for (let column = 0; column < width; column++) {
+      values[column] = columns[column].values;
+    }
     for (let row = 0; row < this.rowCount; row++) {
-      yield this.columns.map((column) => column.values[row]);
+      const out = new Array<QwpResultValue>(width);
+      for (let column = 0; column < width; column++) {
+        out[column] = values[column][row];
+      }
+      yield out;
     }
   }
 }
@@ -908,12 +919,20 @@ export class QwpResultBatchView {
       this.columnViews.map((column) => ({
         name: column.name,
         type: column.type,
-        values: Array.from({ length: this._rowCount }, (_, row) => {
-          const value = column.get(row);
-          // Binary values are zero-copy slices in the view API. materialize()
-          // promises independently owned data, so detach those slices here.
-          return value instanceof Uint8Array ? value.slice() : value;
-        }),
+        values:
+          column.nonNullCount === this._rowCount &&
+          INTEGER_COLUMN_TYPES.has(column.type)
+            ? fillIntegers(this._rowCount, (row) => column.get(row) as number)
+            : column.nonNullCount === this._rowCount &&
+                FLOAT_COLUMN_TYPES.has(column.type)
+              ? fillFloats(this._rowCount, (row) => column.get(row) as number)
+              : fillArray(this._rowCount, (row) => {
+                  const value = column.get(row);
+                  // Binary values are zero-copy slices in the view API.
+                  // materialize() promises independently owned data, so
+                  // detach those slices here.
+                  return value instanceof Uint8Array ? value.slice() : value;
+                }),
         ...(column.scale === undefined ? {} : { scale: column.scale }),
         ...(column.precisionBits === undefined
           ? {}
@@ -1062,7 +1081,8 @@ function signedLittleEndianValue(
 }
 
 interface NullLayout {
-  nulls: boolean[];
+  /** Per-row null flags, or null when the column carries no nulls. */
+  nulls: boolean[] | null;
   nonNullCount: number;
 }
 
@@ -1100,7 +1120,13 @@ function readCount(
   maximum: number,
   label: string,
 ): number {
-  const value = readQwpVarint(reader);
+  const value = readQwpVarintSmall(reader);
+  if (typeof value === "number") {
+    if (value > maximum) {
+      throw new QwpProtocolError(`${label} out of range: ${value}`);
+    }
+    return value;
+  }
   if (value > BigInt(maximum)) {
     throw new QwpProtocolError(`${label} out of range: ${value}`);
   }
@@ -1112,8 +1138,8 @@ function readNullLayout(reader: QwpByteReader, rowCount: number): NullLayout {
   if (flag !== 0 && flag !== 1) {
     throw new QwpProtocolError(`invalid column null flag: ${flag}`);
   }
+  if (flag === 0) return { nulls: null, nonNullCount: rowCount };
   const nulls = new Array<boolean>(rowCount).fill(false);
-  if (flag === 0) return { nulls, nonNullCount: rowCount };
 
   const bitmap = reader.readBytes(
     Math.ceil(rowCount / 8),
@@ -1129,14 +1155,18 @@ function readNullLayout(reader: QwpByteReader, rowCount: number): NullLayout {
   return { nulls, nonNullCount };
 }
 
-function expandNulls<T extends QwpResultValue>(
-  dense: readonly T[],
+function expandNulls(
+  dense: QwpResultValue[],
   layout: NullLayout,
 ): QwpResultValue[] {
-  const values = new Array<QwpResultValue>(layout.nulls.length);
+  const nulls = layout.nulls;
+  // Every dense producer returns a fresh array of exactly nonNullCount
+  // elements, so a null-free column can hand it over without a copy.
+  if (nulls === null) return dense;
+  const values = new Array<QwpResultValue>(nulls.length);
   let denseIndex = 0;
-  for (let row = 0; row < layout.nulls.length; row++) {
-    values[row] = layout.nulls[row] ? null : dense[denseIndex++];
+  for (let row = 0; row < nulls.length; row++) {
+    values[row] = nulls[row] ? null : dense[denseIndex++];
   }
   return values;
 }
@@ -1230,21 +1260,165 @@ function decodeGorillaValues(reader: QwpByteReader, count: number): bigint[] {
   return values;
 }
 
+// Array.from({ length }, mapper) takes the generic array-like path; a
+// preallocated array with an indexed loop avoids that overhead while still
+// reading values in order.
+function fillArray<T>(count: number, mapper: (index: number) => T): T[] {
+  const values = new Array<T>(count);
+  for (let index = 0; index < count; index++) values[index] = mapper(index);
+  return values;
+}
+
+// The helpers below duplicate fillArray on purpose. V8 tracks an elements kind
+// per allocation site: fillArray's `new Array` is shared with schema, bigint,
+// and object callers, so V8 pretransitions it to generic elements, and every
+// double stored there becomes a boxed HeapNumber (~24 instead of 8 bytes per
+// element, and slower consumer loops). Arrays handed to users that hold only
+// numbers -- DOUBLE_ARRAY elements, null-free numeric columns from
+// materialize(), and the numeric cases of readFixedWidthValues() -- get their
+// own sites so they keep unboxed SMI/double elements. Do not fold these back
+// into fillArray, and never store null or a non-number through them.
+function readFloat64Values(reader: QwpByteReader, count: number): number[] {
+  const values = new Array<number>(count);
+  for (let index = 0; index < count; index++) {
+    values[index] = reader.readFloat64("double array element");
+  }
+  return values;
+}
+
+const INTEGER_COLUMN_TYPES: ReadonlySet<number> = new Set([
+  QWP_COLUMN_TYPE.BYTE,
+  QWP_COLUMN_TYPE.SHORT,
+  QWP_COLUMN_TYPE.INT,
+  QWP_COLUMN_TYPE.IPV4,
+]);
+const FLOAT_COLUMN_TYPES: ReadonlySet<number> = new Set([
+  QWP_COLUMN_TYPE.FLOAT,
+  QWP_COLUMN_TYPE.DOUBLE,
+]);
+
+function fillIntegers(
+  count: number,
+  mapper: (index: number) => number,
+): number[] {
+  const values = new Array<number>(count);
+  for (let index = 0; index < count; index++) values[index] = mapper(index);
+  return values;
+}
+
+function fillFloats(
+  count: number,
+  mapper: (index: number) => number,
+): number[] {
+  const values = new Array<number>(count);
+  for (let index = 0; index < count; index++) values[index] = mapper(index);
+  return values;
+}
+
+// Fixed-width columns: one bounds check for the whole column, then direct
+// little-endian DataView reads, instead of a closure call and a bounds-checked
+// reader method per value. Truncation throws the same "truncated QWP payload
+// while reading <label>" error, detected before the first value rather than
+// at the first missing one. Each case allocates its own array, so V8 keeps a
+// separate elements kind per value type (see the note above readFloat64Values):
+// SMI for the integer kinds, unboxed doubles for FLOAT/DOUBLE.
+function fixedWidthView(
+  reader: QwpByteReader,
+  byteLength: number,
+  label: string,
+): DataView {
+  const bytes = reader.readBytes(byteLength, label);
+  return new DataView(bytes.buffer, bytes.byteOffset, byteLength);
+}
+
+function readInt64Values(
+  reader: QwpByteReader,
+  count: number,
+  label: string,
+): bigint[] {
+  const view = fixedWidthView(reader, count * 8, label);
+  const values = new Array<bigint>(count);
+  for (let index = 0; index < count; index++) {
+    values[index] = view.getBigInt64(index * 8, true);
+  }
+  return values;
+}
+
+function readFixedWidthValues(
+  reader: QwpByteReader,
+  type: QwpColumnType,
+  count: number,
+): QwpResultValue[] {
+  switch (type) {
+    case QWP_COLUMN_TYPE.BYTE: {
+      const view = fixedWidthView(reader, count, "byte value");
+      const values = new Array<number>(count);
+      for (let index = 0; index < count; index++) {
+        values[index] = view.getInt8(index);
+      }
+      return values;
+    }
+    case QWP_COLUMN_TYPE.SHORT: {
+      const view = fixedWidthView(reader, count * 2, "short value");
+      const values = new Array<number>(count);
+      for (let index = 0; index < count; index++) {
+        values[index] = view.getInt16(index * 2, true);
+      }
+      return values;
+    }
+    case QWP_COLUMN_TYPE.CHAR: {
+      const view = fixedWidthView(reader, count * 2, "char value");
+      const values = new Array<string>(count);
+      for (let index = 0; index < count; index++) {
+        values[index] = String.fromCharCode(view.getUint16(index * 2, true));
+      }
+      return values;
+    }
+    case QWP_COLUMN_TYPE.INT:
+    case QWP_COLUMN_TYPE.IPV4: {
+      const view = fixedWidthView(reader, count * 4, "int value");
+      const values = new Array<number>(count);
+      for (let index = 0; index < count; index++) {
+        values[index] = view.getInt32(index * 4, true);
+      }
+      return values;
+    }
+    case QWP_COLUMN_TYPE.FLOAT: {
+      const view = fixedWidthView(reader, count * 4, "float value");
+      const values = new Array<number>(count);
+      for (let index = 0; index < count; index++) {
+        values[index] = view.getFloat32(index * 4, true);
+      }
+      return values;
+    }
+    case QWP_COLUMN_TYPE.DOUBLE: {
+      const view = fixedWidthView(reader, count * 8, "double value");
+      const values = new Array<number>(count);
+      for (let index = 0; index < count; index++) {
+        values[index] = view.getFloat64(index * 8, true);
+      }
+      return values;
+    }
+    case QWP_COLUMN_TYPE.LONG:
+      return readInt64Values(reader, count, "long value");
+    default:
+      throw new QwpProtocolError(
+        `not a fixed-width QWP result column type: ${String(type)}`,
+      );
+  }
+}
+
 function readTimestampValues(
   reader: QwpByteReader,
   count: number,
   gorilla: boolean,
 ): bigint[] {
   if (!gorilla) {
-    return Array.from({ length: count }, () =>
-      reader.readBigInt64("timestamp value"),
-    );
+    return readInt64Values(reader, count, "timestamp value");
   }
   const encoding = reader.readUint8("timestamp encoding");
   if (encoding === 0) {
-    return Array.from({ length: count }, () =>
-      reader.readBigInt64("timestamp value"),
-    );
+    return readInt64Values(reader, count, "timestamp value");
   }
   if (encoding !== 1) {
     throw new QwpProtocolError(`unknown timestamp encoding: ${encoding}`);
@@ -1285,14 +1459,12 @@ function readArrayValue(
   if (type === QWP_COLUMN_TYPE.DOUBLE_ARRAY) {
     return {
       dimensions: shape,
-      values: Array.from({ length: elementCount }, () =>
-        reader.readFloat64("double array element"),
-      ),
+      values: readFloat64Values(reader, elementCount),
     };
   }
   return {
     dimensions: shape,
-    values: Array.from({ length: elementCount }, () =>
+    values: fillArray(elementCount, () =>
       reader.readBigInt64("long array element"),
     ),
   };
@@ -1464,7 +1636,7 @@ export class QwpResultBatchDecoder {
         QWP_MAX_COLUMNS_PER_TABLE,
         "result column count",
       );
-      this.schema = Array.from({ length: columnCount }, () => {
+      this.schema = fillArray(columnCount, () => {
         const nameLength = readCount(
           reader,
           QWP_MAX_IDENTIFIER_BYTES,
@@ -1825,47 +1997,21 @@ export class QwpResultBatchDecoder {
     switch (schema.type) {
       case QWP_COLUMN_TYPE.BOOLEAN: {
         const bytes = reader.readBytes(Math.ceil(count / 8), "boolean values");
-        dense = Array.from(
-          { length: count },
-          (_, index) => (bytes[index >>> 3] & (1 << (index & 7))) !== 0,
+        dense = fillArray(
+          count,
+          (index) => (bytes[index >>> 3] & (1 << (index & 7))) !== 0,
         );
         break;
       }
       case QWP_COLUMN_TYPE.BYTE:
-        dense = Array.from({ length: count }, () =>
-          reader.readInt8("byte value"),
-        );
-        break;
       case QWP_COLUMN_TYPE.SHORT:
-        dense = Array.from({ length: count }, () =>
-          reader.readInt16("short value"),
-        );
-        break;
       case QWP_COLUMN_TYPE.CHAR:
-        dense = Array.from({ length: count }, () =>
-          String.fromCharCode(reader.readUint16("char value")),
-        );
-        break;
       case QWP_COLUMN_TYPE.INT:
       case QWP_COLUMN_TYPE.IPV4:
-        dense = Array.from({ length: count }, () =>
-          reader.readInt32("int value"),
-        );
-        break;
       case QWP_COLUMN_TYPE.FLOAT:
-        dense = Array.from({ length: count }, () =>
-          reader.readFloat32("float value"),
-        );
-        break;
       case QWP_COLUMN_TYPE.DOUBLE:
-        dense = Array.from({ length: count }, () =>
-          reader.readFloat64("double value"),
-        );
-        break;
       case QWP_COLUMN_TYPE.LONG:
-        dense = Array.from({ length: count }, () =>
-          reader.readBigInt64("long value"),
-        );
+        dense = readFixedWidthValues(reader, schema.type, count);
         break;
       case QWP_COLUMN_TYPE.DATE:
       case QWP_COLUMN_TYPE.TIMESTAMP:
@@ -1886,13 +2032,13 @@ export class QwpResultBatchDecoder {
         dense = this.readSymbols(reader, count, rowCount, deltaMode);
         break;
       case QWP_COLUMN_TYPE.UUID:
-        dense = Array.from({ length: count }, () => ({
+        dense = fillArray(count, () => ({
           low: reader.readBigUint64("UUID low bits"),
           high: reader.readBigUint64("UUID high bits"),
         }));
         break;
       case QWP_COLUMN_TYPE.LONG256:
-        dense = Array.from({ length: count }, () => ({
+        dense = fillArray(count, () => ({
           words: [
             reader.readBigInt64("LONG256 word 0"),
             reader.readBigInt64("LONG256 word 1"),
@@ -1911,7 +2057,7 @@ export class QwpResultBatchDecoder {
             : schema.type === QWP_COLUMN_TYPE.DECIMAL128
               ? 16
               : 32;
-        dense = Array.from({ length: count }, () => ({
+        dense = fillArray(count, () => ({
           unscaled: readSignedLittleEndian(reader, bytes, "decimal value"),
           scale: scale!,
         }));
@@ -1925,7 +2071,7 @@ export class QwpResultBatchDecoder {
           );
         }
         const byteCount = Math.ceil(precisionBits / 8);
-        dense = Array.from({ length: count }, () => {
+        dense = fillArray(count, () => {
           const bytes = reader.readBytes(byteCount, "geohash value");
           return {
             bits: geohashLittleEndianValue(bytes, 0, precisionBits!),
@@ -1936,9 +2082,7 @@ export class QwpResultBatchDecoder {
       }
       case QWP_COLUMN_TYPE.DOUBLE_ARRAY:
       case QWP_COLUMN_TYPE.LONG_ARRAY:
-        dense = Array.from({ length: count }, () =>
-          readArrayValue(reader, schema.type),
-        );
+        dense = fillArray(count, () => readArrayValue(reader, schema.type));
         break;
       default:
         throw new QwpProtocolError(
@@ -2012,7 +2156,7 @@ export class QwpResultBatchDecoder {
       }
       dictionary = local;
     }
-    return Array.from({ length: count }, () => {
+    return fillArray(count, () => {
       const id = readCount(reader, dictionary.length, "symbol ID");
       if (id >= dictionary.length) {
         throw new QwpProtocolError(`symbol ID out of range: ${id}`);
diff --git a/packages/client-core/src/_qwp/_core/varint-number.ts b/packages/client-core/src/_qwp/_core/varint-number.ts
new file mode 100644
index 0000000..42e5132
--- /dev/null
+++ b/packages/client-core/src/_qwp/_core/varint-number.ts
@@ -0,0 +1,39 @@
+import { QwpByteReader } from "./bytes";
+import { QwpProtocolError } from "./errors";
+
+/**
+ * Reads an unsigned LEB128 uint64 with the exact validation and reader
+ * advancement of readQwpVarint(), but accumulates encodings of up to 7 bytes
+ * (49 bits, always exact) in a plain number and returns it without touching
+ * BigInt. Counts, lengths, and symbol IDs take this path on every cell, so
+ * building each one as a BigInt only to convert it back dominated decoding.
+ * 8..10-byte encodings, including zero-padded small values, fall back to
+ * BigInt and return a bigint.
+ *
+ * Deliberately not re-exported from the `_core` barrel, so it stays out of
+ * both public packages.
+ *
+ * @internal
+ */
+export function readQwpVarintSmall(reader: QwpByteReader): number | bigint {
+  let byte = reader.readUint8("varint");
+  if ((byte & 0x80) === 0) return byte;
+  let value = byte & 0x7f;
+  let scale = 0x80;
+  for (let index = 1; index < 7; index++) {
+    byte = reader.readUint8("varint");
+    value += (byte & 0x7f) * scale;
+    if ((byte & 0x80) === 0) return value;
+    scale *= 0x80;
+  }
+  let big = BigInt(value);
+  for (let index = 7; index < 10; index++) {
+    byte = reader.readUint8("varint");
+    if (index === 9 && (byte & 0xfe) !== 0) {
+      throw new QwpProtocolError("QWP varint exceeds uint64 range");
+    }
+    big |= BigInt(byte & 0x7f) << BigInt(index * 7);
+    if ((byte & 0x80) === 0) return big;
+  }
+  throw new QwpProtocolError("QWP varint exceeds 10 bytes");
+}
diff --git a/packages/client-core/src/_qwp/_core/varint.ts b/packages/client-core/src/_qwp/_core/varint.ts
index bb73d3f..8bc4d76 100644
--- a/packages/client-core/src/_qwp/_core/varint.ts
+++ b/packages/client-core/src/_qwp/_core/varint.ts
@@ -1,5 +1,6 @@
 import { QwpByteReader, QwpByteWriter } from "./bytes";
 import { QwpProtocolError } from "./errors";
+import { readQwpVarintSmall } from "./varint-number";
 
 const MAX_UINT64 = 0xffffffffffffffffn;
 
@@ -60,7 +61,8 @@ export function readQwpVarintNumber(
   reader: QwpByteReader,
   label = "varint",
 ): number {
-  const value = readQwpVarint(reader);
+  const value = readQwpVarintSmall(reader);
+  if (typeof value === "number") return value;
   if (value > BigInt(Number.MAX_SAFE_INTEGER)) {
     throw new QwpProtocolError(
       `${label} exceeds JavaScript's safe integer range`,
diff --git a/test/qwp/core.test.ts b/test/qwp/core.test.ts
index fcd01cd..a096474 100644
--- a/test/qwp/core.test.ts
+++ b/test/qwp/core.test.ts
@@ -43,6 +43,7 @@ import {
   qwpGorillaSize,
   qwpVarintSize,
   readQwpVarint,
+  readQwpVarintNumber,
   writeQwpFrameHeader,
   writeQwpVarint,
 } from "../../packages/client-core/src/qwp";
@@ -52,6 +53,7 @@ import {
   utf8Length,
 } from "../../packages/client-core/src/_qwp/_core/bytes";
 import { measureQwpIngressFrame } from "../../packages/client-core/src/_qwp/_core/ingress";
+import { readQwpVarintSmall } from "../../packages/client-core/src/_qwp/_core/varint-number";
 
 function dataView(bytes: Uint8Array): DataView {
   return new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength);
@@ -100,6 +102,105 @@ describe("QWP browser-safe byte core", () => {
     ).toThrow(/uint64/i);
   });
 
+  it("reads number-valued varints exactly like readQwpVarint", () => {
+    // The decoder's counts and symbol IDs use readQwpVarintSmall, which skips
+    // BigInt for encodings up to 7 bytes. It must accept, reject, and advance
+    // exactly as readQwpVarint does; only the result type may differ.
+    const outcome = (
+      bytes: Uint8Array,
+      read: (reader: QwpByteReader) => number | bigint,
+    ) => {
+      const reader = new QwpByteReader(bytes);
+      try {
+        return { value: BigInt(read(reader)), position: reader.position };
+      } catch (error) {
+        return { error: String(error), position: reader.position };
+      }
+    };
+    const check = (bytes: Uint8Array) => {
+      expect(outcome(bytes, readQwpVarintSmall)).toEqual(
+        outcome(bytes, readQwpVarint),
+      );
+    };
+
+    for (const value of [
+      0n,
+      127n,
+      128n,
+      2n ** 49n - 1n, // largest 7-byte value, still a number
+      2n ** 49n, // first 8-byte value, bigint fallback
+      2n ** 53n,
+      2n ** 64n - 1n,
+    ]) {
+      const encoded = encodeQwpVarint(value);
+      check(encoded);
+      const expectedType = encoded.length <= 7 ? "number" : "bigint";
+      expect(typeof readQwpVarintSmall(new QwpByteReader(encoded))).toBe(
+        expectedType,
+      );
+    }
+    // Zero-padded (non-canonical) encodings of 1, crossing the 7-byte limit.
+    for (let length = 2; length <= 11; length++) {
+      const bytes = new Uint8Array(length).fill(0x80);
+      bytes[0] = 0x81;
+      bytes[length - 1] = 0;
+      check(bytes);
+    }
+    check(Uint8Array.from([0x80, 0x80])); // truncated
+    check(new Uint8Array(11).fill(0xff)); // over 10 bytes
+    // A 10th byte may only carry bit 63: 0x00 and 0x01 are accepted, while
+    // 0x02..0x7f overflow uint64 and must fail with the uint64 error rather
+    // than a caller's later range check.
+    for (const last of [0x00, 0x01, 0x02, 0x03, 0x40, 0x7f]) {
+      const bytes = new Uint8Array(10).fill(0x80);
+      bytes[9] = last;
+      check(bytes);
+    }
+
+    let seed = 0x5eed;
+    const random = () => {
+      seed = (Math.imul(seed, 1103515245) + 12345) >>> 0;
+      return seed >>> 24;
+    };
+    for (let run = 0; run < 5000; run++) {
+      const bytes = new Uint8Array(1 + (random() % 12));
+      for (let index = 0; index < bytes.length; index++) {
+        bytes[index] = random();
+      }
+      check(bytes);
+    }
+  });
+
+  it("reads 8-10 byte varints as numbers up to the safe integer limit", () => {
+    // readQwpVarintNumber returns 7-byte encodings straight from
+    // readQwpVarintSmall; longer ones arrive as a bigint and must still come
+    // back as a number, or fail the safe-integer check.
+    const read = (bytes: Uint8Array) => {
+      const reader = new QwpByteReader(bytes);
+      const value = readQwpVarintNumber(reader, "test count");
+      return { value, position: reader.position };
+    };
+    for (let length = 8; length <= 10; length++) {
+      const bytes = new Uint8Array(length).fill(0x80);
+      bytes[0] = 0x81;
+      bytes[length - 1] = 0;
+      expect(read(bytes)).toEqual({ value: 1, position: length });
+    }
+    const maxSafe = encodeQwpVarint(Number.MAX_SAFE_INTEGER);
+    expect(maxSafe.length).toBe(8);
+    expect(read(maxSafe)).toEqual({
+      value: Number.MAX_SAFE_INTEGER,
+      position: 8,
+    });
+    for (const value of [2n ** 53n, 2n ** 64n - 1n]) {
+      expect(() => read(encodeQwpVarint(value))).toThrow(
+        new QwpProtocolError(
+          "test count exceeds JavaScript's safe integer range",
+        ),
+      );
+    }
+  });
+
   it("measures UTF-8 byte length identically to encoding it", () => {
     // utf8Length feeds frame sizing, so it must count exactly what encodeUtf8()
     // writes -- including the 3-byte replacement for an unpaired surrogate --
diff --git a/test/qwp/egress.test.ts b/test/qwp/egress.test.ts
index eee7baa..c65dc0f 100644
--- a/test/qwp/egress.test.ts
+++ b/test/qwp/egress.test.ts
@@ -20,12 +20,14 @@ import {
   QwpBinaryConnection,
   QwpByteReader,
   QwpByteWriter,
+  QwpColumnType,
   QwpConnectionCloseInfo,
   QwpEgressQueryAbandonedError,
   QwpEgressQueryCancelTimeoutError,
   QwpEgressQueryError,
   QwpEgressQueryTimeoutError,
   QwpEgressSession,
+  QwpProtocolError,
   QwpResultBatchDecoder,
   QwpTableBuffer,
   QwpResultBatchView,
@@ -512,6 +514,242 @@ describe("QWP result batch decoder", () => {
     ]);
   });
 
+  // RESULT_BATCH message type, request ID, and batch sequence.
+  function resultBatchPrefix(): QwpByteWriter {
+    const payload = new QwpByteWriter();
+    payload.writeUint8(QWP_EGRESS_MESSAGE.RESULT_BATCH).writeBigUint64(0n);
+    writeQwpVarint(payload, 0); // batch sequence
+    return payload;
+  }
+
+  function resultBatchHeader(
+    rows: number,
+    schema: readonly (readonly [string, QwpColumnType])[],
+  ): QwpByteWriter {
+    const payload = resultBatchPrefix();
+    writeQwpVarint(payload, 0); // table name
+    writeQwpVarint(payload, rows);
+    writeQwpVarint(payload, schema.length);
+    for (const [name, type] of schema) {
+      writeString(payload, name);
+      payload.writeUint8(type);
+    }
+    return payload;
+  }
+
+  it("decodes values in wire order across nulls and arrays", () => {
+    const payload = resultBatchHeader(4, [
+      ["bool", QWP_COLUMN_TYPE.BOOLEAN],
+      ["ts", QWP_COLUMN_TYPE.TIMESTAMP],
+      ["array", QWP_COLUMN_TYPE.LONG_ARRAY],
+      ["sym", QWP_COLUMN_TYPE.SYMBOL],
+    ]);
+
+    payload.writeUint8(0).writeUint8(0b1010); // bit-packed booleans
+    payload.writeUint8(1).writeUint8(0b0010); // timestamp row 1 is null
+    payload.writeBigInt64(100n).writeBigInt64(200n).writeBigInt64(300n);
+    payload.writeUint8(0); // no array nulls
+    for (const values of [[1n, 2n], [], [3n], [4n, 5n, 6n]]) {
+      payload.writeUint8(1).writeInt32(values.length);
+      for (const value of values) payload.writeBigInt64(value);
+    }
+    payload.writeUint8(0); // no SYMBOL nulls; local dictionary follows
+    writeQwpVarint(payload, 2);
+    writeString(payload, "alpha");
+    writeString(payload, "beta");
+    for (const id of [1, 0, 1, 0]) writeQwpVarint(payload, id);
+
+    const message = decodeQwpEgressMessage(
+      encodeQwpFrame(payload.toUint8Array(), 0, 1),
+    );
+    if (message.kind !== "result-batch") throw new Error("unexpected message");
+    const batch = new QwpResultBatchDecoder().decode(message);
+    expect(batch.columns.map((column) => column.values)).toEqual([
+      [false, true, false, true],
+      [100n, null, 200n, 300n],
+      [
+        { dimensions: [2], values: [1n, 2n] },
+        { dimensions: [0], values: [] },
+        { dimensions: [1], values: [3n] },
+        { dimensions: [3], values: [4n, 5n, 6n] },
+      ],
+      ["beta", "alpha", "beta", "alpha"],
+    ]);
+  });
+
+  // Fixed-width columns decode straight from a DataView in decode(), with a
+  // separate array per value type; null-free numeric columns also take
+  // dedicated fill paths in materialize(). Distinct per-row values catch any
+  // reordering, and the nullable DOUBLE covers the null-expansion path.
+  const fixedWidthColumns = [
+    ["byte", QWP_COLUMN_TYPE.BYTE, "byte value", [-1, 2, 3]],
+    ["short", QWP_COLUMN_TYPE.SHORT, "short value", [-300, 400, 500]],
+    ["char", QWP_COLUMN_TYPE.CHAR, "char value", ["x", "é", "Z"]],
+    ["int", QWP_COLUMN_TYPE.INT, "int value", [7, -8, 9]],
+    [
+      "ipv4",
+      QWP_COLUMN_TYPE.IPV4,
+      "int value",
+      [0x0a000001, 0x0a000002, 0x0a000003],
+    ],
+    ["float", QWP_COLUMN_TYPE.FLOAT, "float value", [1.5, -2.25, 3.75]],
+    ["double", QWP_COLUMN_TYPE.DOUBLE, "double value", [0.1, -0.2, 0.3]],
+    ["long", QWP_COLUMN_TYPE.LONG, "long value", [-(2n ** 63n), 0n, 2n ** 62n]],
+    ["ts", QWP_COLUMN_TYPE.TIMESTAMP, "timestamp value", [100n, 300n, 200n]],
+  ] as const;
+
+  function writeFixedWidthValue(
+    payload: QwpByteWriter,
+    type: QwpColumnType,
+    value: number | bigint | string,
+  ): void {
+    switch (type) {
+      case QWP_COLUMN_TYPE.BYTE:
+        payload.writeInt8(value as number);
+        break;
+      case QWP_COLUMN_TYPE.SHORT:
+        payload.writeInt16(value as number);
+        break;
+      case QWP_COLUMN_TYPE.CHAR:
+        payload.writeUint16((value as string).charCodeAt(0));
+        break;
+      case QWP_COLUMN_TYPE.FLOAT:
+        payload.writeFloat32(value as number);
+        break;
+      case QWP_COLUMN_TYPE.DOUBLE:
+        payload.writeFloat64(value as number);
+        break;
+      case QWP_COLUMN_TYPE.LONG:
+      case QWP_COLUMN_TYPE.TIMESTAMP:
+        payload.writeBigInt64(value as bigint);
+        break;
+      default:
+        payload.writeInt32(value as number);
+    }
+  }
+
+  it("keeps fixed-width columns in row order through decode, rows, and materialize", () => {
+    const payload = resultBatchHeader(3, [
+      ...fixedWidthColumns.map(([name, type]) => [name, type] as const),
+      ["nullable", QWP_COLUMN_TYPE.DOUBLE] as const,
+    ]);
+    for (const [, type, , values] of fixedWidthColumns) {
+      payload.writeUint8(0); // no nulls
+      for (const value of values) writeFixedWidthValue(payload, type, value);
+    }
+    payload.writeUint8(1).writeUint8(0b010); // nullable DOUBLE, row 1 null
+    payload.writeFloat64(4.5).writeFloat64(-5.5);
+
+    const expected = [
+      ...fixedWidthColumns.map(([, , , values]) => [...values]),
+      [4.5, null, -5.5],
+    ];
+    const expectedRows = [0, 1, 2].map((row) =>
+      expected.map((values) => values[row]),
+    );
+    const frame = encodeQwpFrame(payload.toUint8Array(), 0, 1);
+    const decoded = decodeQwpEgressMessage(frame);
+    if (decoded.kind !== "result-batch") throw new Error("unexpected message");
+    const batch = new QwpResultBatchDecoder().decode(decoded);
+    expect(batch.columns.map((column) => column.values)).toEqual(expected);
+    expect([...batch.rows()]).toEqual(expectedRows);
+
+    const viewed = decodeQwpEgressMessage(frame);
+    if (viewed.kind !== "result-batch") throw new Error("unexpected message");
+    const view = new QwpResultBatchDecoder().decodeView(viewed);
+    const retained = view.materialize();
+    view.release();
+    expect(retained.columns.map((column) => column.values)).toEqual(expected);
+    expect([...retained.rows()]).toEqual(expectedRows);
+  });
+
+  it("names the column type when a fixed-width column is truncated", () => {
+    for (const [name, type, label, values] of fixedWidthColumns) {
+      const payload = resultBatchHeader(3, [[name, type]]);
+      payload.writeUint8(0); // no nulls, but only two of three values follow
+      for (const value of values.slice(0, 2)) {
+        writeFixedWidthValue(payload, type, value);
+      }
+      const message = decodeQwpEgressMessage(
+        encodeQwpFrame(payload.toUint8Array(), 0, 1),
+      );
+      if (message.kind !== "result-batch") {
+        throw new Error("unexpected message");
+      }
+      const decode = () => new QwpResultBatchDecoder().decode(message);
+      expect(decode).toThrow(QwpProtocolError);
+      expect(decode).toThrow(`truncated QWP payload while reading ${label}`);
+    }
+  });
+
+  it("keeps raw-fallback timestamps in row order in Gorilla-flagged batches", () => {
+    // With the Gorilla flag set, a timestamp column still falls back to raw
+    // int64 values (encoding 0) when a delta-of-delta leaves int32 range, as
+    // these irregular nanosecond gaps do. The trailing INT column fails if the
+    // raw values consume the wrong number of bytes.
+    const timestamps = [
+      1_700_000_000_000_000_000n,
+      1_700_000_000_000_000_001n,
+      1_700_000_005_000_000_000n,
+      1_700_000_005_000_000_002n,
+    ];
+    const schema = [
+      ["date", QWP_COLUMN_TYPE.DATE],
+      ["ts", QWP_COLUMN_TYPE.TIMESTAMP],
+      ["nanos", QWP_COLUMN_TYPE.TIMESTAMP_NANOS],
+      ["nullable", QWP_COLUMN_TYPE.TIMESTAMP_NANOS],
+      ["int", QWP_COLUMN_TYPE.INT],
+    ] as const;
+    const payload = resultBatchHeader(timestamps.length, schema);
+    for (let column = 0; column < 3; column++) {
+      payload.writeUint8(0).writeUint8(0); // no nulls, raw encoding
+      for (const value of timestamps) payload.writeBigInt64(value);
+    }
+    // Rows 1 and 3 are null, so only two raw values follow.
+    payload.writeUint8(1).writeUint8(0b1010).writeUint8(0);
+    payload.writeBigInt64(timestamps[3]).writeBigInt64(timestamps[0]);
+    payload.writeUint8(0);
+    for (const value of [11, 22, 33, 44]) payload.writeInt32(value);
+
+    const expected = [
+      timestamps,
+      timestamps,
+      timestamps,
+      [timestamps[3], null, timestamps[0], null],
+      [11, 22, 33, 44],
+    ];
+    const expectedRows = timestamps.map((_, row) =>
+      expected.map((values) => values[row]),
+    );
+    const frame = encodeQwpFrame(payload.toUint8Array(), QWP_FLAG_GORILLA, 1);
+    const decoded = decodeQwpEgressMessage(frame);
+    if (decoded.kind !== "result-batch") throw new Error("unexpected message");
+    const batch = new QwpResultBatchDecoder().decode(decoded);
+    expect(batch.columns.map((column) => column.values)).toEqual(expected);
+    expect([...batch.rows()]).toEqual(expectedRows);
+
+    const viewed = decodeQwpEgressMessage(frame);
+    if (viewed.kind !== "result-batch") throw new Error("unexpected message");
+    const view = new QwpResultBatchDecoder().decodeView(viewed);
+    const retained = view.materialize();
+    view.release();
+    expect(retained.columns.map((column) => column.values)).toEqual(expected);
+
+    // Three rows declared, two raw values present.
+    const truncated = resultBatchHeader(3, [["ts", QWP_COLUMN_TYPE.TIMESTAMP]]);
+    truncated.writeUint8(0).writeUint8(0);
+    truncated.writeBigInt64(timestamps[0]).writeBigInt64(timestamps[1]);
+    const message = decodeQwpEgressMessage(
+      encodeQwpFrame(truncated.toUint8Array(), QWP_FLAG_GORILLA, 1),
+    );
+    if (message.kind !== "result-batch") throw new Error("unexpected message");
+    const decode = () => new QwpResultBatchDecoder().decode(message);
+    expect(decode).toThrow(QwpProtocolError);
+    expect(decode).toThrow(
+      "truncated QWP payload while reading timestamp value",
+    );
+  });
+
   it("decodes identifiers at the defensive egress byte bound", () => {
     // Query results may expose existing Java metadata created through another
     // protocol. Keep accepting up to 127 UTF-16 code units on egress, while
@@ -915,6 +1153,67 @@ describe("QWP result batch decoder", () => {
     expect(process.memoryUsage().heapUsed - before).toBeLessThan(50e6);
   });
 
+  it("range-checks counts sent as zero-padded 8-10 byte varints", () => {
+    // readCount() reads encodings of up to 7 bytes as a number and longer ones
+    // as a bigint. A non-minimal encoding sends a small count down the bigint
+    // branch, which must enforce the same limits as the number branch.
+    const writePaddedVarint = (
+      writer: QwpByteWriter,
+      value: number,
+      length: number,
+    ) => {
+      let remaining = value;
+      for (let index = 0; index < length; index++) {
+        const low = remaining % 0x80;
+        remaining = Math.floor(remaining / 0x80);
+        writer.writeUint8(index < length - 1 ? low | 0x80 : low);
+      }
+    };
+    const cases: [string, (payload: QwpByteWriter, length: number) => void][] =
+      [
+        [
+          "table name length out of range: 1000000",
+          (payload, length) => writePaddedVarint(payload, 1_000_000, length),
+        ],
+        [
+          // One above the decoder's 1,048,576-row batch limit.
+          "result row count out of range: 1048577",
+          (payload, length) => {
+            writeQwpVarint(payload, 0); // table name
+            writePaddedVarint(payload, 1_048_577, length);
+          },
+        ],
+        [
+          `result column count out of range: ${QWP_MAX_COLUMNS_PER_TABLE + 1}`,
+          (payload, length) => {
+            writeQwpVarint(payload, 0); // table name
+            writeQwpVarint(payload, 1); // rows
+            writePaddedVarint(payload, QWP_MAX_COLUMNS_PER_TABLE + 1, length);
+          },
+        ],
+      ];
+
+    for (const length of [8, 9, 10]) {
+      for (const [error, writeCount] of cases) {
+        const payload = resultBatchPrefix();
+        writeCount(payload, length);
+        const message = decodeQwpEgressMessage(
+          encodeQwpFrame(payload.toUint8Array(), 0, 1),
+        );
+        if (message.kind !== "result-batch") {
+          throw new Error("unexpected message");
+        }
+        for (const decode of [
+          () => new QwpResultBatchDecoder().decode(message),
+          () => new QwpResultBatchDecoder().decodeView(message),
+        ]) {
+          expect(decode).toThrow(QwpProtocolError);
+          expect(decode).toThrow(error);
+        }
+      }
+    }
+  });
+
   it("still decodes a delta dictionary that fits its frame", () => {
     // A real delta carries actual symbols, so its entry count never exceeds the
     // bytes that transmitted it: the bound only rejects counts Zstd manufactured.
diff --git a/test/qwp/reconnect.test.ts b/test/qwp/reconnect.test.ts
index 178dd05..51f3c13 100644
--- a/test/qwp/reconnect.test.ts
+++ b/test/qwp/reconnect.test.ts
@@ -8753,23 +8753,27 @@ describe("QWP Node file replay store", () => {
     try {
       // Five segments become fully acknowledged in one maintenance batch.
       await store.acknowledgeThrough(4n);
-      await vi.waitFor(() => expect(unlinked).toHaveLength(1));
+      // The lapse after the first trim ends this batch, and arming the retry
+      // timer is the last thing it does. Wait for that rather than for the
+      // queue to shrink: pendingTrimSegments reads 4 while the batch is still
+      // re-proving ownership, and un-pinning `lost` then lets that check
+      // succeed, so the batch would keep trimming alongside the direct call
+      // below and both would unlink the same segment.
       await vi.waitFor(() =>
-        expect(internals.pendingTrimSegments).toHaveLength(4),
+        expect(internals.maintenanceRetryTimer).toBeDefined(),
       );
+      expect(unlinked).toHaveLength(1);
+      expect(internals.pendingTrimSegments).toHaveLength(4);
 
       // Re-proving the same token resumes the preserved queue. Calling the
       // batch directly avoids making this regression test wait for its 1s timer.
       delete (lock as unknown as Record<string, unknown>).lost;
       // Production only ever reaches runMaintenanceBatch() through the store's
-      // serializing queue. Calling it directly races the retry timer that the
-      // ownership lapse just armed: both batches read pendingTrimSegments[0],
-      // both unlink it, and both shift. Disarm it so this stays a test of the
-      // preserved queue rather than an intermittent double-unlink.
-      if (internals.maintenanceRetryTimer) {
-        clearTimeout(internals.maintenanceRetryTimer);
-        internals.maintenanceRetryTimer = undefined;
-      }
+      // serializing queue, so disarm the timer: a timer-driven batch would run
+      // alongside the direct call, both would read pendingTrimSegments[0], and
+      // both would unlink it.
+      clearTimeout(internals.maintenanceRetryTimer);
+      internals.maintenanceRetryTimer = undefined;
       await internals.runMaintenanceBatch();
       expect(unlinked).toHaveLength(5);
       expect(internals.pendingTrimSegments).toHaveLength(0);