diff --git a/src/query.zig b/src/query.zig
index 8c14c1f..0d2c03d 100644
--- a/src/query.zig
+++ b/src/query.zig
@@ -1112,3 +1112,166 @@ pub const Table = struct {
         return results.toOwnedSlice();
     }
 
+    /// Streams primary keys straight from the index cursor instead of first
+    /// materializing the whole candidate list (see `idx.EqCursor`/
+    /// `idx.RangeCursor`): once `page.limit` rows are collected the
+    /// underlying `Db` scan is simply dropped without decoding or even
+    /// reading whatever index entries would have come after it, which
+    /// matters when the indexed value has far more matches than the page
+    /// being asked for.
+    fn selectByIndexPage(self: *Table, p: Predicate, cursorIn: IndexCursor, page: Page) ![]Row {
+        var cursor = cursorIn;
+        defer cursor.deinit();
+
+        var results = std.ArrayList(Row).init(self.allocator);
+        errdefer {
+            for (results.items) |*r| r.deinit();
+            results.deinit();
+        }
+        var skipped: usize = 0;
+        while (try cursor.next(self.allocator)) |pk| {
+            defer pk.free(self.allocator);
+            var r = try self.get(pk) orelse continue;
+            if (!p.evaluate(&r)) {
+                r.deinit();
+                continue;
+            }
+            if (skipped < page.offset) {
+                skipped += 1;
+                r.deinit();
+                continue;
+            }
+            try results.append(r);
+            if (page.limit != 0 and results.items.len >= page.limit) break;
+        }
+        return results.toOwnedSlice();
+    }
+
+    fn findIndexableLeaf(self: *const Table, p: Predicate) ?Predicate {
+        switch (p) {
+            .eq, .lt, .lte, .gt, .gte => |c| {
+                if (self.indexes.has(c.field)) return p;
+                return null;
+            },
+            .and_ => |preds| {
+                for (preds) |sub| {
+                    if (self.findIndexableLeaf(sub)) |found| return found;
+                }
+                return null;
+            },
+            else => return null,
+        }
+    }
+
+    /// Streaming counterpart of the old `pksForLeaf` (which returned a fully
+    /// materialized `[]Value`): wraps whichever `idx` cursor constructor
+    /// matches the leaf's comparison so `selectByIndexPage` can decode and
+    /// discard candidates one at a time.
+    const IndexCursor = union(enum) {
+        eq: idx.EqCursor,
+        range: idx.RangeCursor,
+        compositeEq: idx.CompositeEqCursor,
+        compositeRange: idx.CompositeRangeCursor,
+
+        fn next(self: *IndexCursor, allocator: Allocator) !?Value {
+            return switch (self.*) {
+                .eq => |*c| c.next(allocator),
+                .range => |*c| c.next(allocator),
+                .compositeEq => |*c| c.next(allocator),
+                .compositeRange => |*c| c.next(allocator),
+            };
+        }
+
+        fn deinit(self: *IndexCursor) void {
+            switch (self.*) {
+                .eq => |*c| c.deinit(),
+                .range => |*c| c.deinit(),
+                .compositeEq => |*c| c.deinit(),
+                .compositeRange => |*c| c.deinit(),
+            }
+        }
+    };
+
+    fn cursorForLeaf(self: *Table, leaf: Predicate) !IndexCursor {
+        return switch (leaf) {
+            .eq => |c| .{ .eq = try idx.scanEqCursor(self.db, self.allocator, self.name, c.field, c.value) },
+            .lt => |c| .{ .range = try idx.scanRangeCursor(self.db, self.allocator, self.name, c.field, null, .{ .value = c.value, .inclusive = false }) },
+            .lte => |c| .{ .range = try idx.scanRangeCursor(self.db, self.allocator, self.name, c.field, null, .{ .value = c.value, .inclusive = true }) },
+            .gt => |c| .{ .range = try idx.scanRangeCursor(self.db, self.allocator, self.name, c.field, .{ .value = c.value, .inclusive = false }, null) },
+            .gte => |c| .{ .range = try idx.scanRangeCursor(self.db, self.allocator, self.name, c.field, .{ .value = c.value, .inclusive = true }, null) },
+            else => unreachable,
+        };
+    }
+
+    const CompositeMatch = struct {
+        specIdx: usize,
+        values: [maxCompositeFields]Value,
+        matchedLen: usize,
+        // A range condition on the column immediately after the matched
+        // eq prefix (spec.fields[matchedLen]), if the predicate has one -
+        // same leftmost-prefix rule a real multi-column B-tree range scan
+        // follows: only the column right after the pinned prefix can be
+        // range-bound by the index, anything further is unconstrained.
+        rangeLow: ?idx.Bound = null,
+        rangeHigh: ?idx.Bound = null,
+    };
+
+    /// Looks for a composite index whose leading ("leftmost") columns are
+    /// all pinned down by a top-level `eq` conjunct somewhere in `p` (an
+    /// `and_`, in any order - unlike a real SQL planner this does not
+    /// recurse into nested and_/or_, matching `findIndexableLeaf`'s scope).
+    /// The match may cover a strict prefix of the index's columns, not
+    /// necessarily all of them: a query naming only the first K of N
+    /// indexed fields still gets a narrowed lookup, same leftmost-prefix
+    /// rule a multi-column B-tree index gives you. When the column right
+    /// after the matched prefix also has a top-level lt/lte/gt/gte conjunct
+    /// on it, that becomes a range bound narrowing the scan further instead
+    /// of leaving it unconstrained. `selectByIndexPage` re-evaluates the
+    /// full predicate against every candidate row regardless, so an
+    /// unmatched suffix of columns (or any other conjunct entirely) never
+    /// affects correctness, only how much the index narrows the scan.
+    fn findCompositeMatch(self: *const Table, p: Predicate) ?CompositeMatch {
+        const preds = switch (p) {
+            .and_ => |ps| ps,
+            else => return null,
+        };
+        for (self.composite.specs.items, 0..) |spec, si| {
+            var values: [maxCompositeFields]Value = undefined;
+            var matched: usize = 0;
+            for (spec.fields) |f| {
+                const found = for (preds) |leaf| {
+                    if (leaf == .eq and std.mem.eql(u8, leaf.eq.field, f)) break leaf.eq.value;
+                } else null;
+                const v = found orelse break;
+                values[matched] = v;
+                matched += 1;
+            }
+            if (matched == 0) continue;
+
+            var rangeLow: ?idx.Bound = null;
+            var rangeHigh: ?idx.Bound = null;
+            if (matched < spec.fields.len) {
+                const nextField = spec.fields[matched];
+                for (preds) |leaf| {
+                    switch (leaf) {
+                        .gt => |c| if (std.mem.eql(u8, c.field, nextField)) {
+                            rangeLow = .{ .value = c.value, .inclusive = false };
+                        },
+                        .gte => |c| if (std.mem.eql(u8, c.field, nextField)) {
+                            rangeLow = .{ .value = c.value, .inclusive = true };
+                        },
+                        .lt => |c| if (std.mem.eql(u8, c.field, nextField)) {
+                            rangeHigh = .{ .value = c.value, .inclusive = false };
+                        },
+                        .lte => |c| if (std.mem.eql(u8, c.field, nextField)) {
+                            rangeHigh = .{ .value = c.value, .inclusive = true };
+                        },
+                        else => {},
+                    }
+                }
+            }
+            return .{ .specIdx = si, .values = values, .matchedLen = matched, .rangeLow = rangeLow, .rangeHigh = rangeHigh };
+        }
+        return null;
+    }
+