feat(scan): implement limit properly — truncate last chunk to exact row count

ScanIterator now truncates the final chunk to exactly limit-rowsReturned
rows instead of returning the full chunk and stopping at the next iteration.
Covers: all primitive types, BoolArray, NullArray, VarBinArray (dict and
non-dict), MaskedArray. Unsupported types throw VortexException.

VarBinArray.truncate(long): new public method handles both dict mode (slice
codes segment) and non-dict mode (slice bytes + offsets).
MaskedArray.validity(): new accessor returns the validity bitmap or null.
ScanOptions.limit(long): new factory method for preview/head scans.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: dfa1

core/src/main/java/io/github/dfa1/vortex/core/array/MaskedArray.java

@@ -51,6 +51,11 @@ public Array child(int i) {
 		};
 	}
 
+	/// Returns the validity bitmap, or {@code null} if all positions are valid.
+	public BoolArray validity() {
+		return validity;
+	}
+
 	public boolean isValid(long i) {
 		return validity == null || validity.getBoolean(i);
 	}

core/src/main/java/io/github/dfa1/vortex/core/array/VarBinArray.java

@@ -188,4 +188,24 @@ private long dictReadOff(long i) {
 		}
 		return dictValOffsets.getAtIndex(PTypeIO.LE_LONG, i);
 	}
+
+	/// Returns a new VarBinArray containing only the first {@code rows} elements.
+	public VarBinArray truncate(long rows) {
+		if (rows >= length) {
+			return this;
+		}
+		if (dictCodesSegs != null) {
+			int codeBytes = dictCodesPType.byteSize();
+			return VarBinArray.ofDict(dtype, rows, bytes, dictValOffsets, dictValOffPType,
+					dictCodesSegs.asSlice(0, rows * codeBytes), dictCodesPType, ArrayStats.empty());
+		}
+		long byteEnd = readOffset(rows);
+		int offBytes = (offsetsPtype == PType.I32 || offsetsPtype == PType.U32) ? Integer.BYTES : Long.BYTES;
+		MemorySegment newOffsetsSeg = offsetsSeg.asSlice(0, (rows + 1) * offBytes);
+		DType offDtype = new DType.Primitive(offsetsPtype, false);
+		Array newOffsetsArr = (offBytes == Integer.BYTES)
+				? new IntArray(offDtype, rows + 1, newOffsetsSeg, ArrayStats.empty())
+				: new LongArray(offDtype, rows + 1, newOffsetsSeg, ArrayStats.empty());
+		return new VarBinArray(dtype, rows, bytes.asSlice(0, byteEnd > 0 ? byteEnd : 0), newOffsetsArr, offsetsPtype, ArrayStats.empty());
+	}
 }

reader/src/main/java/io/github/dfa1/vortex/scan/ScanIterator.java

@@ -9,7 +9,11 @@
 import io.github.dfa1.vortex.core.Layout;
 import io.github.dfa1.vortex.core.PType;
 import io.github.dfa1.vortex.core.SegmentSpec;
+import io.github.dfa1.vortex.core.array.BoolArray;
+import io.github.dfa1.vortex.core.array.MaskedArray;
 import io.github.dfa1.vortex.core.array.ByteArray;
+import io.github.dfa1.vortex.core.array.EmptyArray;
+import io.github.dfa1.vortex.core.array.NullArray;
 import io.github.dfa1.vortex.core.array.DoubleArray;
 import io.github.dfa1.vortex.core.array.FloatArray;
 import io.github.dfa1.vortex.core.array.IntArray;
@@ -132,8 +136,15 @@ public boolean hasNext() {
 				chunkArena.close();
 			}
 			chunkArena = Arena.ofConfined();
-			current = new ScanResult(chunk.rowCount(), buildColumnMap(chunk));
-			rowsReturned += chunk.rowCount();
+
+			long remaining = options.limit() - rowsReturned;
+			long chunkRows = Math.min(chunk.rowCount(), remaining);
+			Map<String, Array> columns = buildColumnMap(chunk);
+			if (chunkRows < chunk.rowCount()) {
+				columns = truncateColumns(columns, chunkRows);
+			}
+			current = new ScanResult(chunkRows, columns);
+			rowsReturned += chunkRows;
 			return true;
 		}
 		// Do not close here — the last chunk's arena stays open until close() is called.
@@ -397,6 +408,41 @@ private static long readUnsigned(MemorySegment seg, long idx, PType ptype) {
 		};
 	}
 
+	// ── Limit truncation ─────────────────────────────────────────────────────
+
+	private static Map<String, Array> truncateColumns(Map<String, Array> columns, long rows) {
+		var result = new LinkedHashMap<String, Array>(columns.size());
+		for (var entry : columns.entrySet()) {
+			result.put(entry.getKey(), truncateArray(entry.getValue(), rows));
+		}
+		return Map.copyOf(result);
+	}
+
+	private static Array truncateArray(Array arr, long rows) {
+		if (arr.length() <= rows) {
+			return arr;
+		}
+		return switch (arr) {
+			case LongArray   a -> new LongArray(a.dtype(), rows, a.buffer(0).asSlice(0, rows * Long.BYTES), ArrayStats.empty());
+			case IntArray    a -> new IntArray(a.dtype(), rows, a.buffer(0).asSlice(0, rows * Integer.BYTES), ArrayStats.empty());
+			case DoubleArray a -> new DoubleArray(a.dtype(), rows, a.buffer(0).asSlice(0, rows * Double.BYTES), ArrayStats.empty());
+			case FloatArray  a -> new FloatArray(a.dtype(), rows, a.buffer(0).asSlice(0, rows * Float.BYTES), ArrayStats.empty());
+			case ShortArray  a -> new ShortArray(a.dtype(), rows, a.buffer(0).asSlice(0, rows * Short.BYTES), ArrayStats.empty());
+			case ByteArray   a -> new ByteArray(a.dtype(), rows, a.buffer(0).asSlice(0, rows), ArrayStats.empty());
+			case BoolArray   a -> new BoolArray(a.dtype(), rows, a.buffer(0).asSlice(0, (rows + 7) / 8), ArrayStats.empty());
+			case NullArray   a -> new NullArray(a.dtype(), rows);
+			case VarBinArray a -> a.truncate(rows);
+			case MaskedArray a -> {
+				Array truncChild = truncateArray((Array) a.child(0), rows);
+				BoolArray v = a.validity();
+				BoolArray truncValidity = (v != null) ? (BoolArray) truncateArray(v, rows) : null;
+				yield new MaskedArray(truncChild, truncValidity);
+			}
+			case EmptyArray  a -> a;
+			default -> throw new VortexException("limit: truncation not supported for " + arr.getClass().getSimpleName());
+		};
+	}
+
 	private boolean canPruneChunk(ChunkSpec chunk, RowFilter filter) {
 		return switch (filter) {
 			case RowFilter.And(var filters) -> {

reader/src/main/java/io/github/dfa1/vortex/scan/ScanOptions.java

@@ -21,6 +21,10 @@ public static ScanOptions columns(String... names) {
 		return new ScanOptions(List.of(names), null, NO_LIMIT);
 	}
 
+	public static ScanOptions limit(long limit) {
+		return new ScanOptions(List.of(), null, limit);
+	}
+
 	public boolean hasProjection() {
 		return !columns.isEmpty();
 	}

reader/src/test/java/io/github/dfa1/vortex/io/VortexReaderTest.java

@@ -265,6 +265,73 @@ void scan_chunkedFixture_producesExactlyOneLayoutChunk() throws URISyntaxExcepti
 		assertThat(chunkCount).isEqualTo(1);
 	}
 
+	@Test
+	void scan_withLimit_returnsExactlyNRows() throws URISyntaxException, IOException {
+		// Given — primitives.vortex has 3 rows; limit=2 forces truncation of the single chunk
+		Path path = fixtureFile("primitives.vortex");
+		long limit = 2;
+
+		// When
+		long totalRows = 0;
+		try (var sut = VortexReader.open(path);
+		     var iter = sut.scan(ScanOptions.limit(limit))) {
+			while (iter.hasNext()) {
+				ScanResult chunk = iter.next();
+				totalRows += chunk.rowCount();
+				for (Array col : chunk.columns().values()) {
+					assertThat(col.length()).isLessThanOrEqualTo(limit);
+				}
+			}
+		}
+
+		// Then
+		assertThat(totalRows).isEqualTo(limit);
+	}
+
+	@Test
+	void scan_withLimitExceedingTotal_returnsAllRows() throws URISyntaxException, IOException {
+		// Given
+		Path path = fixtureFile("primitives.vortex");
+		long totalWithoutLimit;
+		try (var sut = VortexReader.open(path);
+		     var iter = sut.scan(ScanOptions.all())) {
+			totalWithoutLimit = 0;
+			while (iter.hasNext()) {
+				totalWithoutLimit += iter.next().rowCount();
+			}
+		}
+
+		// When
+		long totalWithLimit = 0;
+		try (var sut = VortexReader.open(path);
+		     var iter = sut.scan(ScanOptions.limit(Long.MAX_VALUE))) {
+			while (iter.hasNext()) {
+				totalWithLimit += iter.next().rowCount();
+			}
+		}
+
+		// Then
+		assertThat(totalWithLimit).isEqualTo(totalWithoutLimit);
+	}
+
+	@Test
+	void scan_withLimitZero_returnsNoRows() throws URISyntaxException, IOException {
+		// Given
+		Path path = fixtureFile("primitives.vortex");
+
+		// When
+		long totalRows = 0;
+		try (var sut = VortexReader.open(path);
+		     var iter = sut.scan(ScanOptions.limit(0))) {
+			while (iter.hasNext()) {
+				totalRows += iter.next().rowCount();
+			}
+		}
+
+		// Then
+		assertThat(totalRows).isZero();
+	}
+
 	// --- helpers ---
 
 	private Path fixtureFile(String name) throws URISyntaxException {