Fix: pushed-down WHERE + LIMIT/OFFSET silently drops matching rows (#27)

icebergDataSource.scan() pushed LIMIT/OFFSET down by physical row position
(seeking past `offset` rows and bounding the per-file read at
`fileRowStart + remaining`) whenever the WHERE was *resolved* — which
includes a WHERE fully pushed into the parquet read. But a pushed-down
WHERE is matched per row, so the first N physical rows of a file may
contain fewer than N (or zero) matches. Bounding by position then reads
only the leading rows and silently drops every match that sorts later in
the file.

Example: `SELECT ... WHERE node_type = 'File' LIMIT 5` over a file whose
leading 1000 rows are all `Session` reads physical rows [0,5), matches
nothing, and returns 0 rows — while `COUNT(*)` (no LIMIT) returns the
true count. Any `WHERE <pushable predicate> LIMIT n` can under-return.

Gate position-based pushdown on `!where` instead of `whereResolved`, so a
pushed-down filter takes the same path deletes already use: emit up to
`offset + limit` matched rows and let the engine apply the final slice.
Early termination via the per-row `remaining` break is preserved.

Adds a regression test (a match that sorts after the LIMIT window must
still be returned) and corrects two tests that asserted the buggy
`appliedLimitOffset === true` contract for a pushed WHERE.

Author: philcunliffe

src/sql/icebergDataSource.js

@@ -31,10 +31,13 @@ import { whereToParquetFilter } from './whereFilter.js'
  *   leave WHERE for the engine to apply.
  * - When WHERE is resolved at scan time (either absent or fully pushed) we
  *   cap the scan at `offset + limit` rows so the source terminates early.
- *   OFFSET is also pushed into the parquet seek when there are no deletes;
- *   with deletes the engine still applies OFFSET itself (record_count is
- *   pre-delete, so seeking by record_count would miscount visible rows in
- *   any file with applicable deletes).
+ *   OFFSET is also pushed into the parquet seek, and the per-file read bounded
+ *   by row position, only when there is no WHERE at all: a pushed-down WHERE is
+ *   matched per row, so physical row positions no longer line up with result
+ *   positions and a position-based bound would drop matching rows that sort
+ *   later in the file. Deletes disable position pushdown for the same reason
+ *   (record_count is pre-delete). In those cases the engine applies the final
+ *   LIMIT/OFFSET slice over the (at most offset+limit) rows the source emits.
  *
  * @param {object} options
  * @param {string} options.tableUrl - Base URL or path of the table.
@@ -104,18 +107,26 @@ export async function icebergDataSource({ tableUrl, metadataFileName, metadata,
         : dataEntries
       const pruned = scanEntries.length < dataEntries.length
       // Treat a fully-pushed-down WHERE the same as "no WHERE" for the
-      // purpose of LIMIT/OFFSET pushdown.
+      // purpose of capping how many rows the source emits (LIMIT).
       const whereResolved = !where || appliedWhere
-      // OFFSET pushdown (seeking past rows in the parquet file) is only safe
-      // when the WHERE is fully resolved at scan time AND the table has no
-      // deletes: record_count is pre-delete, so seeking by it would skip the
-      // wrong visible rows. It also assumes the cumulative record_count tracks
-      // row positions, so disable it once pruning has removed any file.
-      const canPushOffset = whereResolved && !hasDeletes && !pruned
+      // Position-based pushdown — seeking past `offset` physical rows and the
+      // `fileRowEnd` LIMIT bound below — translates a row *count* into a
+      // physical row *position*, which is only correct when every physical row
+      // is also a result row. That holds only when there is NO WHERE at all.
+      // A pushed-down WHERE (appliedWhere) is matched per-row inside the
+      // parquet read, so the first N physical rows may contain fewer than N
+      // (or zero) matches; bounding by position would silently drop matching
+      // rows that sort later in the file (e.g. WHERE node_type='File' LIMIT 5
+      // when the leading rows are all 'Session'). It is likewise unsafe with
+      // deletes (record_count is pre-delete) or once pruning has dropped a
+      // file (cumulative record_count no longer tracks row positions). In all
+      // those cases we keep emitting up to `offset + limit` matched rows and
+      // let the engine apply the final LIMIT/OFFSET slice.
+      const canPushOffset = !where && !hasDeletes && !pruned
       const skip = canPushOffset ? offset ?? 0 : 0
-      // LIMIT pushdown (early termination) is safe whenever WHERE is
-      // resolved: with deletes we yield offset+limit rows and let the engine
-      // apply the slice, which still saves reading later files.
+      // LIMIT (early termination) is safe whenever WHERE is resolved: we yield
+      // at most offset+limit rows and, when offset isn't pushed, let the engine
+      // apply the slice. This still saves reading later files/row groups.
       let take = Infinity
       if (whereResolved && limit !== undefined) {
         take = canPushOffset ? limit : (offset ?? 0) + limit

test/sql/icebergDataSource.test.js

@@ -102,7 +102,9 @@ describe.concurrent('icebergDataSource', () => {
     })
     const { rows, appliedWhere, appliedLimitOffset } = source.scan({ where })
     expect(appliedWhere).toBe(true)
-    expect(appliedLimitOffset).toBe(true)
+    // A pushed-down WHERE disables position-based LIMIT/OFFSET pushdown, so the
+    // engine owns the final slice even though none was requested here.
+    expect(appliedLimitOffset).toBe(false)
 
     const collected = []
     for await (const row of rows()) collected.push(row.resolved)
@@ -113,7 +115,11 @@ describe.concurrent('icebergDataSource', () => {
     }
   })
 
-  it('pushes WHERE down and combines with LIMIT/OFFSET when no deletes', async () => {
+  it('pushes WHERE down but lets the engine apply LIMIT/OFFSET', async () => {
+    // A pushed-down WHERE is matched per row, so OFFSET cannot be pushed by
+    // physical position (the first N physical rows are not the first N
+    // matches). appliedLimitOffset must be false; the source emits up to
+    // offset+limit matched rows and the engine slices the final window.
     const source = await icebergDataSource({
       tableUrl,
       resolver,
@@ -125,12 +131,50 @@ describe.concurrent('icebergDataSource', () => {
       left: { type: 'identifier', name: 'Popularity Rank' },
       right: { type: 'literal', value: 10n },
     })
-    const { rows, appliedWhere, appliedLimitOffset } = source.scan({ where, limit: 2, offset: 1 })
-    expect(appliedWhere).toBe(true)
-    expect(appliedLimitOffset).toBe(true)
+    // Full matched set in physical order, for the slice oracle.
+    const matched = []
+    for await (const row of source.scan({ where }).rows()) matched.push(row.resolved)
+
+    const offset = 1
+    const limit = 2
+    const plan = source.scan({ where, limit, offset })
+    expect(plan.appliedWhere).toBe(true)
+    expect(plan.appliedLimitOffset).toBe(false)
     const collected = []
-    for await (const row of rows()) collected.push(row.resolved)
-    expect(collected).toHaveLength(2)
+    for await (const row of plan.rows()) collected.push(row.resolved)
+    // Source emits the first offset+limit matches; engine slices [offset, offset+limit).
+    expect(collected).toEqual(matched.slice(0, offset + limit))
+    expect(collected.slice(offset, offset + limit)).toEqual(matched.slice(offset, offset + limit))
+  })
+
+  it('pushed WHERE + LIMIT returns matches that sort after the LIMIT window', async () => {
+    // Regression: with a pushed-down filter, bounding the per-file read at
+    // `offset + limit` physical rows silently dropped any match positioned
+    // after that window. Pick the LAST physical row and query for it with
+    // LIMIT 1: a position bound would read only row 0 and return nothing.
+    const source = await icebergDataSource({
+      tableUrl,
+      resolver,
+      metadataFileName: 'v2.metadata.json',
+    })
+    const all = []
+    for await (const row of source.scan({}).rows()) all.push(row.resolved)
+    expect(all.length).toBeGreaterThan(1)
+    const target = /** @type {Record<string, any>} */ (all[all.length - 1])
+    const targetRank = /** @type {bigint} */ (target['Popularity Rank'])
+
+    // Equality on a unique column → pushable, matches exactly the last row.
+    const where = /** @type {ExprNode} */ ({
+      type: 'binary',
+      op: '=',
+      left: { type: 'identifier', name: 'Popularity Rank' },
+      right: { type: 'literal', value: targetRank },
+    })
+    const plan = source.scan({ where, limit: 1 })
+    expect(plan.appliedWhere).toBe(true)
+    const collected = []
+    for await (const row of plan.rows()) collected.push(row.resolved)
+    expect(collected).toEqual([target])
   })
 
   it('pushed WHERE still respects row-level deletes', async () => {