From 73c7b0e918c3e156358e12b80963c9be68d647a4 Mon Sep 17 00:00:00 2001 From: Phillip Cunliffe Date: Mon, 22 Jun 2026 11:07:36 -0700 Subject: [PATCH] Fix: pushed-down WHERE + LIMIT/OFFSET silently drops matching rows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 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. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/sql/icebergDataSource.js | 39 ++++++++++++-------- test/sql/icebergDataSource.test.js | 58 ++++++++++++++++++++++++++---- 2 files changed, 76 insertions(+), 21 deletions(-) diff --git a/src/sql/icebergDataSource.js b/src/sql/icebergDataSource.js index 0152e3d..3901007 100644 --- a/src/sql/icebergDataSource.js +++ b/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 diff --git a/test/sql/icebergDataSource.test.js b/test/sql/icebergDataSource.test.js index 0b371a3..8516716 100644 --- a/test/sql/icebergDataSource.test.js +++ b/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} */ (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 () => {