|
| 1 | +import { typeName } from './schema.js' |
1 | 2 | import { applyTransform } from './write/transform.js' |
| 3 | +import { compare, deserializeValue } from './write/serde.js' |
2 | 4 |
|
3 | 5 | /** |
4 | 6 | * Partition-level scan pruning. Given a hyparquet query filter (keyed by |
@@ -26,34 +28,65 @@ export function partitionMightMatch(filter, dataEntry, schema, metadata) { |
26 | 28 | const spec = metadata['partition-specs'].find(s => s['spec-id'] === dataEntry.partition_spec_id) |
27 | 29 | // No spec or unpartitioned: nothing to prune on. |
28 | 30 | if (!spec || spec.fields.length === 0) return true |
29 | | - return nodeMightMatch(filter, { spec, schema, partition: dataEntry.data_file.partition }) |
| 31 | + /** @type {PruneContext} */ |
| 32 | + const ctx = { spec, schema, partition: dataEntry.data_file.partition } |
| 33 | + return nodeMightMatch(filter, (column, condition) => columnMightMatch(column, condition, ctx)) |
| 34 | +} |
| 35 | + |
| 36 | +/** |
| 37 | + * File-level scan pruning via per-column manifest bounds. Given a hyparquet |
| 38 | + * query filter (keyed by iceberg field name) and a data manifest entry, decide |
| 39 | + * whether the entry's `lower_bounds` / `upper_bounds` could contain a row |
| 40 | + * matching the filter. |
| 41 | + * |
| 42 | + * Like `partitionMightMatch` this is an inclusive projection: a file is skipped |
| 43 | + * only when its bounds prove that no row can match, and any uncertainty keeps |
| 44 | + * the file. Bounds for string/binary/fixed/uuid are stored truncated (Iceberg |
| 45 | + * `truncate(16)` metrics) so those types are never range-pruned. Pruning never |
| 46 | + * changes query results, only which files are read. |
| 47 | + * |
| 48 | + * @param {ParquetQueryFilter} filter - Filter keyed by iceberg field name. |
| 49 | + * @param {ManifestEntry} dataEntry |
| 50 | + * @param {Schema} schema - Current schema (filter column names map to its fields). |
| 51 | + * @returns {boolean} true if the file must be read, false if it can be skipped. |
| 52 | + */ |
| 53 | +export function fileMightMatch(filter, dataEntry, schema) { |
| 54 | + const { lower_bounds, upper_bounds } = dataEntry.data_file |
| 55 | + // No bounds at all: nothing to prune on. |
| 56 | + if (lower_bounds === undefined && upper_bounds === undefined) return true |
| 57 | + return nodeMightMatch(filter, (column, condition) => |
| 58 | + boundsMightMatch(column, condition, dataEntry.data_file, schema)) |
30 | 59 | } |
31 | 60 |
|
32 | 61 | /** |
33 | 62 | * @typedef {{ spec: PartitionSpec, schema: Schema, partition: DataFile['partition'] }} PruneContext |
34 | 63 | */ |
35 | 64 |
|
36 | 65 | /** |
| 66 | + * Generic AND/OR/$nor walk over a hyparquet filter. Top-level keys are |
| 67 | + * AND-combined: the file is ruled out if any branch is ruled out. The leaf |
| 68 | + * evaluator decides a single `{column: condition}` predicate and returns false |
| 69 | + * only when it proves the file cannot match. |
| 70 | + * |
37 | 71 | * @param {ParquetQueryFilter} node |
38 | | - * @param {PruneContext} ctx |
| 72 | + * @param {(column: string, condition: any) => boolean} leafFn |
39 | 73 | * @returns {boolean} |
40 | 74 | */ |
41 | | -function nodeMightMatch(node, ctx) { |
| 75 | +function nodeMightMatch(node, leafFn) { |
42 | 76 | if (!node || typeof node !== 'object') return true |
43 | 77 | const anyNode = /** @type {Record<string, any>} */ (node) |
44 | | - // Top-level keys are AND-combined: the file is ruled out if any is ruled out. |
45 | 78 | for (const [key, val] of Object.entries(anyNode)) { |
46 | 79 | if (key === '$and') { |
47 | 80 | const subs = /** @type {ParquetQueryFilter[]} */ (val) |
48 | | - if (!subs.every(sub => nodeMightMatch(sub, ctx))) return false |
| 81 | + if (!subs.every(sub => nodeMightMatch(sub, leafFn))) return false |
49 | 82 | } else if (key === '$or') { |
50 | 83 | const subs = /** @type {ParquetQueryFilter[]} */ (val) |
51 | | - if (!subs.some(sub => nodeMightMatch(sub, ctx))) return false |
| 84 | + if (!subs.some(sub => nodeMightMatch(sub, leafFn))) return false |
52 | 85 | } else if (key === '$nor') { |
53 | 86 | // NOT(a OR b): can't safely prune, keep. |
54 | 87 | continue |
55 | 88 | } else { |
56 | | - if (!columnMightMatch(key, val, ctx)) return false |
| 89 | + if (!leafFn(key, val)) return false |
57 | 90 | } |
58 | 91 | } |
59 | 92 | return true |
@@ -331,3 +364,175 @@ function numericOf(x) { |
331 | 364 | function sign(n) { |
332 | 365 | return n < 0 ? -1 : n > 0 ? 1 : 0 |
333 | 366 | } |
| 367 | + |
| 368 | +/** |
| 369 | + * Whether the file could match `condition` on `column` given the column's |
| 370 | + * decoded [lower, upper] bounds in the manifest entry. Returns true (keep) on |
| 371 | + * any uncertainty, including columns without bounds and non-orderable types. |
| 372 | + * |
| 373 | + * @param {string} column - iceberg field name |
| 374 | + * @param {any} condition - operator object like {$eq: x} or a bare value (eq) |
| 375 | + * @param {DataFile} dataFile |
| 376 | + * @param {Schema} schema |
| 377 | + * @returns {boolean} |
| 378 | + */ |
| 379 | +function boundsMightMatch(column, condition, dataFile, schema) { |
| 380 | + const field = schema.fields.find(f => f.name === column) |
| 381 | + if (!field) return true |
| 382 | + // Bounds and metric ordering are only defined for orderable scalar types. |
| 383 | + // String/binary/fixed/uuid bounds are truncated prefixes, so we never use |
| 384 | + // them for pruning (equality on a truncated prefix is undecidable). |
| 385 | + if (!isOrderableForBounds(field.type)) return true |
| 386 | + |
| 387 | + const lowerBytes = boundForField(dataFile.lower_bounds, field.id) |
| 388 | + const upperBytes = boundForField(dataFile.upper_bounds, field.id) |
| 389 | + if (lowerBytes === undefined && upperBytes === undefined) return true |
| 390 | + const lo = lowerBytes !== undefined ? deserializeValue(lowerBytes, field.type) : undefined |
| 391 | + const hi = upperBytes !== undefined ? deserializeValue(upperBytes, field.type) : undefined |
| 392 | + if (lo === undefined && hi === undefined) return true |
| 393 | + |
| 394 | + for (const { op, value } of normalizeCondition(condition)) { |
| 395 | + if (!boundsOpMightMatch(op, value, lo, hi, field.type)) return false |
| 396 | + } |
| 397 | + return true |
| 398 | +} |
| 399 | + |
| 400 | +/** |
| 401 | + * Look up a column's bound bytes by field id. Read-decoded Iceberg maps arrive |
| 402 | + * as an array of `{key, value}` records (Avro int-keyed maps); hand-built |
| 403 | + * entries may instead be a plain `Record<fieldId, bytes>`. Handle both. |
| 404 | + * |
| 405 | + * @param {any} map - lower_bounds or upper_bounds from a manifest data_file |
| 406 | + * @param {number} fieldId |
| 407 | + * @returns {Uint8Array | undefined} |
| 408 | + */ |
| 409 | +function boundForField(map, fieldId) { |
| 410 | + if (map === undefined || map === null) return undefined |
| 411 | + if (Array.isArray(map)) { |
| 412 | + const entry = map.find(e => e && Number(e.key) === fieldId) |
| 413 | + return entry ? entry.value : undefined |
| 414 | + } |
| 415 | + const v = map[fieldId] |
| 416 | + return v instanceof Uint8Array ? v : undefined |
| 417 | +} |
| 418 | + |
| 419 | +/** |
| 420 | + * Whether a type has totally-ordered, untruncated single-value bounds suitable |
| 421 | + * for range/equality pruning. String/binary/fixed/uuid are excluded because |
| 422 | + * their bounds are stored truncated. |
| 423 | + * |
| 424 | + * @param {IcebergType} type |
| 425 | + * @returns {boolean} |
| 426 | + */ |
| 427 | +function isOrderableForBounds(type) { |
| 428 | + const name = typeName(type) |
| 429 | + if (name.startsWith('decimal(')) return true |
| 430 | + switch (name) { |
| 431 | + case 'boolean': |
| 432 | + case 'int': |
| 433 | + case 'long': |
| 434 | + case 'float': |
| 435 | + case 'double': |
| 436 | + case 'date': |
| 437 | + case 'time': |
| 438 | + case 'timestamp': |
| 439 | + case 'timestamptz': |
| 440 | + case 'timestamp_ns': |
| 441 | + case 'timestamptz_ns': |
| 442 | + return true |
| 443 | + default: |
| 444 | + return false |
| 445 | + } |
| 446 | +} |
| 447 | + |
| 448 | +/** |
| 449 | + * Whether a value range [lo, hi] (either side may be open/undefined) could |
| 450 | + * satisfy `op value`. Mirrors hyparquet's `canSkipRowGroup` operator semantics |
| 451 | + * but at file granularity. Returns true (keep) on any uncertainty; returns |
| 452 | + * false only when the predicate is provably unsatisfiable for the whole range. |
| 453 | + * |
| 454 | + * @param {string} op - mongo-style operator |
| 455 | + * @param {any} value - the predicate literal (array for $in/$nin) |
| 456 | + * @param {any} lo - decoded lower bound, or undefined (open below) |
| 457 | + * @param {any} hi - decoded upper bound, or undefined (open above) |
| 458 | + * @param {IcebergType} type |
| 459 | + * @returns {boolean} |
| 460 | + */ |
| 461 | +function boundsOpMightMatch(op, value, lo, hi, type) { |
| 462 | + switch (op) { |
| 463 | + case '$lt': { |
| 464 | + // need some x < value; smallest is lo. Skip if lo >= value. |
| 465 | + if (lo === undefined) return true |
| 466 | + const c = safeCompare(lo, value, type) |
| 467 | + return c === undefined ? true : c < 0 |
| 468 | + } |
| 469 | + case '$lte': { |
| 470 | + if (lo === undefined) return true |
| 471 | + const c = safeCompare(lo, value, type) |
| 472 | + return c === undefined ? true : c <= 0 |
| 473 | + } |
| 474 | + case '$gt': { |
| 475 | + // need some x > value; largest is hi. Skip if hi <= value. |
| 476 | + if (hi === undefined) return true |
| 477 | + const c = safeCompare(hi, value, type) |
| 478 | + return c === undefined ? true : c > 0 |
| 479 | + } |
| 480 | + case '$gte': { |
| 481 | + if (hi === undefined) return true |
| 482 | + const c = safeCompare(hi, value, type) |
| 483 | + return c === undefined ? true : c >= 0 |
| 484 | + } |
| 485 | + case '$eq': |
| 486 | + return eqInRange(value, lo, hi, type) |
| 487 | + case '$in': |
| 488 | + if (!Array.isArray(value)) return true |
| 489 | + // Keep if any listed value could fall in [lo, hi]. |
| 490 | + return value.some(x => eqInRange(x, lo, hi, type)) |
| 491 | + // $ne / $nin can only prune a single-valued file fully covered by the |
| 492 | + // excluded value(s); too rare to bother — keep. |
| 493 | + default: |
| 494 | + return true |
| 495 | + } |
| 496 | +} |
| 497 | + |
| 498 | +/** |
| 499 | + * Whether `value` could lie within [lo, hi] (open sides allowed). Keep on any |
| 500 | + * undecidable comparison. |
| 501 | + * |
| 502 | + * @param {any} value |
| 503 | + * @param {any} lo |
| 504 | + * @param {any} hi |
| 505 | + * @param {IcebergType} type |
| 506 | + * @returns {boolean} |
| 507 | + */ |
| 508 | +function eqInRange(value, lo, hi, type) { |
| 509 | + if (lo !== undefined) { |
| 510 | + const c = safeCompare(value, lo, type) |
| 511 | + if (c !== undefined && c < 0) return false |
| 512 | + } |
| 513 | + if (hi !== undefined) { |
| 514 | + const c = safeCompare(value, hi, type) |
| 515 | + if (c !== undefined && c > 0) return false |
| 516 | + } |
| 517 | + return true |
| 518 | +} |
| 519 | + |
| 520 | +/** |
| 521 | + * Type-aware comparison that returns undefined (rather than throwing or |
| 522 | + * returning NaN) when the two values cannot be meaningfully ordered, so the |
| 523 | + * caller keeps the file. |
| 524 | + * |
| 525 | + * @param {any} a |
| 526 | + * @param {any} b |
| 527 | + * @param {IcebergType} type |
| 528 | + * @returns {number | undefined} |
| 529 | + */ |
| 530 | +function safeCompare(a, b, type) { |
| 531 | + if (a === null || a === undefined || b === null || b === undefined) return undefined |
| 532 | + try { |
| 533 | + const c = compare(a, b, type) |
| 534 | + return Number.isNaN(c) ? undefined : c |
| 535 | + } catch { |
| 536 | + return undefined |
| 537 | + } |
| 538 | +} |
0 commit comments