From 9bcff88a7f4b2a861840a8c557e2342aa634009c Mon Sep 17 00:00:00 2001 From: Scott Schenkein Date: Sat, 4 Jul 2026 20:23:51 -0400 Subject: [PATCH 1/2] readers: fix all findings from the table-format reader review Addresses the 26 findings in docs/TABLE_FORMAT_READER_REVIEW.md (included with a resolution-status table). Highlights: High severity: - H1/H2: parquet getTableInfo prunes first-level partition dirs with new three-valued PartitionPredicate::evaluate_partial (unknown column -> keep), and discovers all partition levels by walking one directory chain - H3: delta get_snapshot_info scans all checkpoint parts for the metaData row - H4: read + validate the protocol action; reject unsupported readerFeatures (e.g. v2Checkpoint), minReaderVersion > 3, and columnMapping mode 'id' - H5: stale _last_checkpoint guard via one offset-bounded LIST; re-probe from the newest checkpoint, error on a broken commit chain - H6: compact serialization keeps has_deletion_vector (delta) and content_type (iceberg) - H7: abfss:// container parsed from URL username; account derived from host - H8: S3/Azure builders start from from_env() so env/IMDS credential chains work; explicit config still overrides Medium severity: - M1: percent-decode Url::path() before building object keys (shared decoder in common.rs) - M2: malformed non-empty commit lines fail loudly with file + line number - M3: translate catalog-style credential keys to FileIO keys for executor-side iceberg manifest reads - M4: capture sequence_number on iceberg entries (TANT, Arrow FFI col 8, Java getter) - M5: error instead of silent current_schema() fallback on unresolvable snapshot schema_id - M6: embed resolved_snapshot_id in every serialized iceberg entry; unify missing-snapshot-id fallback to -1 - M7: unify runtime creation to new_current_thread() - M8: render date/timestamp/decimal partition literals human-readably using each manifest's partition result types Low severity: L1-L6, L8-L10 fixed; L7 (exception taxonomy) deferred and documented. Tests: 152/152 Rust unit tests (incl. ~20 new); 107/107 Java tests across the 7 Delta/Iceberg/Parquet reader test classes. Note: IcebergTableReader.readManifestFileArrowFfi now exports 8 columns (sequence_number appended); callers must allocate 8 FFI addresses. Co-Authored-By: Claude Fable 5 --- docs/TABLE_FORMAT_READER_REVIEW.md | 323 +++++++++++ native/src/common.rs | 245 ++++++++- native/src/delta_reader/distributed.rs | 517 ++++++++++++++++-- native/src/delta_reader/engine.rs | 34 +- native/src/delta_reader/jni.rs | 15 +- native/src/delta_reader/scan.rs | 9 + native/src/delta_reader/serialization.rs | 56 +- native/src/iceberg_reader/distributed.rs | 80 ++- native/src/iceberg_reader/jni.rs | 24 +- native/src/iceberg_reader/scan.rs | 169 +++++- native/src/iceberg_reader/serialization.rs | 73 ++- native/src/parquet_reader/distributed.rs | 109 ++-- native/src/parquet_reader/jni.rs | 8 +- native/src/parquet_reader/serialization.rs | 15 +- native/src/parquet_schema_reader.rs | 26 +- .../tantivy4java/delta/DeltaFileEntry.java | 8 +- .../tantivy4java/delta/DeltaTableReader.java | 10 +- .../iceberg/IcebergFileEntry.java | 31 +- .../iceberg/IcebergTableReader.java | 26 +- 19 files changed, 1518 insertions(+), 260 deletions(-) create mode 100644 docs/TABLE_FORMAT_READER_REVIEW.md diff --git a/docs/TABLE_FORMAT_READER_REVIEW.md b/docs/TABLE_FORMAT_READER_REVIEW.md new file mode 100644 index 00000000..be71f8ce --- /dev/null +++ b/docs/TABLE_FORMAT_READER_REVIEW.md @@ -0,0 +1,323 @@ +# Table-Format Reader Code Review + +**Scope:** `native/src/delta_reader/`, `native/src/iceberg_reader/`, `native/src/parquet_reader/`, +shared helpers in `native/src/common.rs`, and the corresponding Java entry points +(`io.indextables.tantivy4java.delta/iceberg/parquet`). + +**Date:** 2026-07-04 + +Overall the code is well-structured (clean driver/executor split, good unit-test coverage of the +serialization and predicate layers, careful Arrow FFI export with all-or-nothing writes). The +findings below are ordered by severity. Line numbers refer to the current `main` working tree. + +--- + +## Resolution status (2026-07-04) + +All findings were independently re-validated against the code (all confirmed) and addressed as +follows. Line numbers in the finding text below refer to the pre-fix tree. + +| Finding | Status | Resolution | +|---------|--------|------------| +| H1 | ✅ Fixed | `PartitionPredicate::evaluate_partial()` (Kleene three-valued logic); `getTableInfo` pruning keeps directories whose predicate is indeterminate at the first level; executors re-apply the full predicate | +| H2 | ✅ Fixed | `get_table_info_async` walks down one partition directory chain (one LIST per level) to discover all partition columns | +| H3 | ✅ Fixed | `get_snapshot_info` scans all checkpoint parts until metaData (and protocol) rows are found; missing column in one part is no longer an error | +| H4 | ✅ Fixed | `protocol` action is read from the checkpoint and validated (`validate_protocol`); unknown readerFeatures / minReaderVersion > 3 / columnMapping mode `id` are rejected with clear errors | +| H5 | ✅ Fixed | After the sequential probe, one offset-bounded LIST (`find_log_markers_after`) detects newer checkpoints; the reader re-probes from the newest checkpoint, and a broken commit chain errors loudly | +| H6 | ✅ Fixed | Compact mode now always includes `has_deletion_vector` (Delta) and `content_type` (Iceberg); only `partition_values` (and iceberg `sequence_number`) are skipped | +| H7 | ✅ Fixed | `create_object_store` parses the container from the URL username for `abfs`/`abfss` and derives the account from the host when not configured; `parquet_schema_reader` preserves userinfo in base URLs | +| H8 | ✅ Fixed | S3/Azure builders now start from `from_env()` (env vars, standard chain) with explicit config overriding | +| M1 | ✅ Fixed | Shared `percent_decode` in common.rs applied in `delta_log_prefix`, `url_to_object_path`, and `parse_file_url`. Delta `add.path` values remain verbatim (percent-encoded per Delta spec) — documented convention: callers joining `table_root + path` must decode | +| M2 | ✅ Fixed | Malformed non-empty commit lines now fail the call with file + line number | +| M3 | ✅ Fixed | `translate_to_file_io_props` maps catalog-style keys (aws_access_key_id, region_name, …) to FileIO keys; vended-credential catalogs documented as requiring the catalog path | +| M4 | ✅ Fixed | `sequence_number` captured on every entry (TANT full mode + Arrow FFI column 8 + Java getter) | +| M5 | ✅ Fixed | Unresolvable `schema_id` is now an error in both `read_schema_with_catalog` and `get_snapshot_info_with_catalog` | +| M6 | ✅ Fixed | `resolved_snapshot_id` embedded in every serialized entry (`IcebergFileEntry.getResolvedSnapshotId()`); missing per-entry snapshot fallback unified to `-1` sentinel in the manifest path | +| M7 | ✅ Partly fixed | Runtime creation unified to `new_current_thread()` everywhere (scan.rs was multi-threaded). Shared-runtime/client caching via `runtime_manager` left as a follow-up optimization | +| M8 | ✅ Fixed | `literal_to_string_typed` renders Date/Timestamp in ISO form and decimals with scale applied, using each manifest's partition result types | +| L1 | ✅ Fixed | Hidden-file (`.`/`_`) filter applied to root-level listing and schema selection | +| L2 | ✅ Fixed | Missing `size` now serializes as `-1` (documented on `DeltaFileEntry.getSize()`) | +| L3 | ✅ Fixed | Unused `LastCheckpointInfo.size` removed | +| L4 | ✅ Documented | Defensive mapping retained (only rewrites physical-name keys); collision hazard documented at the call site | +| L5 | ✅ Fixed | `write_field_header`/`write_string`/`extract_jlong_array` consolidated into common.rs | +| L6 | ✅ Fixed | `debug_assert` at the serialization layer; 2 GiB guard error now points at the distributed APIs | +| L7 | ⏸ Deferred | Exception taxonomy is a cross-layer API design change; all errors still surface as RuntimeException (messages made more distinguishable in M2/L9 fixes) | +| L8 | ✅ Fixed | `nativeGetCurrentSnapshotId` throws on config-extraction failure like its siblings | +| L9 | ✅ Fixed | Missing commit file now reports "version may exceed the table's latest version or the commit has been removed by log cleanup" | +| L10 | ✅ Fixed | Log-change entries serialize `table_version = -1` (documented sentinel) instead of a real-looking 0 | + +--- + +## High severity + +### H1. Parquet `getTableInfo` predicate pruning can silently drop *all* data for multi-level partitions + +`parquet_reader/jni.rs:62-70` prunes partition directories by evaluating the user predicate against +partition values parsed from the directory path. But `get_parquet_table_info` uses +`list_with_delimiter` (`parquet_reader/distributed.rs:100-140`), so each directory entry is only the +**first** partition level (e.g. `year=2024/`). Predicate semantics for a missing column are +"exclude" (`common.rs:342-379`: `Eq`/`Gt`/`In` on a missing column → `false`). + +Consequence: for a table partitioned by `year/month`, a filter like `month = '01'` — or +`AND(year=2024, month=01)` — evaluates `month` against a map that only contains `year`, returns +`false` for every directory, and prunes **everything**. The caller gets an empty table with no +error. The Java API (`ParquetTableReader.getTableInfo(url, config, filter)`) passes the same +`PartitionFilter` object to both levels, so this is the natural way users will hit it. + +Fix options: (a) evaluate only the sub-predicates whose column is present at this level (partial +evaluation with "unknown → keep"), or (b) recurse with `list_with_delimiter` down each partition +level before pruning, or (c) restrict driver-side pruning to predicates that reference only +first-level columns and document it. + +### H2. `ParquetTableInfo.partition_columns` only reports the first partition level + +Same root cause as H1: the "Walk the path to find all partition levels" loop +(`parquet_reader/distributed.rs:126-137`) walks a prefix that can never contain more than one +`key=value` segment, because `list_with_delimiter` returns single-level prefixes. A `year/month/day` +table reports `partition_columns = ["year"]`. Any downstream schema/partition handling built on this +field is wrong for multi-level tables. + +### H3. Delta distributed reads assume the `metaData` action lives in the first checkpoint part + +`get_snapshot_info` reads schema and partition columns only from `checkpoint_part_paths[0]` +(`delta_reader/distributed.rs:120-124`). The Delta protocol does not guarantee which part of a +multi-part checkpoint contains the `metaData` row — writers distribute actions across parts. When +`metaData` is in another part, this fails with "No metaData row found in checkpoint" (or, if a part +has no `metaData` **column** at all, "Checkpoint parquet has no 'metaData' column") for a perfectly +valid table. The fallback should scan subsequent parts until a `metaData` row is found. + +### H4. Delta distributed path performs no protocol / reader-feature check + +The hand-rolled distributed reader (`get_snapshot_info`, `read_checkpoint_part`, +`read_post_checkpoint_changes`) never reads the `protocol` action. It will happily process tables +whose reader features it does not implement: + +- **V2 checkpoints** (`v2Checkpoint` feature): checkpoint files are UUID-named and use + `sidecar` actions. `construct_checkpoint_paths` (`distributed.rs:355-367`) only builds classic + names, so this at least fails loudly — but with a confusing "Failed to HEAD checkpoint" error + rather than "unsupported table feature". +- **Deletion vectors:** files are surfaced with `has_deletion_vector`, but `num_records` still + reflects the pre-delete count and nothing forces the caller to notice (see H6). +- **Column mapping mode `id`** (vs `name`): `build_column_mapping` only handles + `delta.columnMapping.physicalName` metadata; `id` mode tables would pass through untranslated. + +A cheap `protocol` check in `get_snapshot_info` (the protocol row is in the checkpoint next to +`metaData`, and in commit 0) that rejects unknown `readerFeatures` would convert silent/confusing +failures into a clear error. Note the non-distributed `list_delta_files` path is safe here because +delta-kernel enforces the protocol. + +### H5. Stale `_last_checkpoint` + log cleanup can silently return an old snapshot + +`get_current_version` computes `current = checkpoint_version + count(sequential commit files after +it)` (`distributed.rs:264-265`), and `get_snapshot_info` builds its commit list the same way +(`list_commit_files_after`, `distributed.rs:378-409`). The Delta spec explicitly allows +`_last_checkpoint` to lag behind the newest checkpoint. If `_last_checkpoint` points at version *M* +while a newer checkpoint exists at *N* and log cleanup has deleted commit JSONs in `(M, N]`, the +sequential HEAD probe stops at the first missing file and the reader reports version *M* — an +outdated snapshot — with no error. Mitigation: after probing, verify the commit chain actually +reaches a version ≥ the newest checkpoint (e.g. one `list` of `_delta_log/` bounded by prefix, or a +probe for a newer `*.checkpoint.parquet`), or at minimum document the retention assumption. + +### H6. "Compact" serialization mode drops correctness-critical flags + +- Delta: compact mode omits `has_deletion_vector` (`delta_reader/serialization.rs:76-85`). A + compact-mode consumer scanning the listed parquet files directly will include logically deleted + rows with no signal that DVs exist. +- Iceberg: compact mode omits `content_type` (`iceberg_reader/serialization.rs:75-85`). Position- + and equality-delete files become indistinguishable from data files, so a consumer will read + delete files as data *and* miss the deletes they encode. + +Compact mode should either keep these one-byte/short fields (they are tiny compared to the path +string) or refuse to serialize entries where `has_deletion_vector == true` / `content_type != +"data"`. + +### H7. `abfss://` / `abfs://` URLs parse the container from the wrong URL component + +`create_object_store` uses `url.host_str()` as the Azure container name +(`delta_reader/engine.rs:77-98`). The standard ABFS URL shape is +`abfss://@.dfs.core.windows.net/path`, where the container is the **username** +component and `host_str()` is the account endpoint. The current code passes +`account.dfs.core.windows.net` as the container. `az://container/path` works; `abfss://` in its +canonical form cannot. Parse `url.username()` when non-empty for the `abfs`/`abfss` schemes (and +derive the account name from the host if not configured). + +### H8. No default cloud credential chain + +`AmazonS3Builder::new()` / `MicrosoftAzureBuilder::new()` are used with only the explicitly +configured keys (`engine.rs:48-98`). Neither reads environment variables, profiles, IMDS/IRSA, or +workload identity. Deployments that rely on IAM instance roles or `AWS_ACCESS_KEY_ID` env vars — +the common case for Spark executors — get anonymous clients and opaque 403s. Consider +`AmazonS3Builder::from_env()` (and the Azure equivalent) as the base builder, with the config map +overriding. Also note that when `aws_region` is absent no region is set at all, which fails for S3 +rather than falling back to the SDK default chain. + +--- + +## Medium severity + +### M1. Percent-encoded URL paths are passed to `object_store` as raw keys + +`delta_log_prefix` (`delta_reader/distributed.rs:927-936`) and `url_to_object_path` +(`parquet_reader/distributed.rs:390-408`) build object keys from `Url::path()`, which is +**percent-encoded** (`/tmp/my table` → `/tmp/my%20table` after `normalize_url` / +`Url::from_directory_path`). `ObjectPath::from` does not decode, so any table path containing a +space, `+`, `#`, unicode, etc. resolves to the wrong object key. All existing tests use +encoding-neutral paths, so the bug is latent. Decode the path (e.g. `percent_encoding:: +percent_decode_str(url.path())`) before constructing `ObjectPath`. + +Related inconsistency: Delta `add.path` values are percent-encoded per the Delta spec, and the +distributed reader returns them verbatim (no decode), while the parquet reader *does* percent-decode +partition values from paths (`parquet_reader/distributed.rs:317-356`). Whoever joins +`table_root + entry.path` on the Java side needs a single documented convention. + +### M2. Malformed Delta commit lines are silently skipped + +`read_post_checkpoint_changes_async` ignores any line that fails JSON parsing +(`distributed.rs:624-627`, `Err(_) => continue`). A truncated or corrupted commit file (partial +upload, torn write) silently produces an incomplete file list — exactly the failure mode where you +want a loud error, since this feeds incremental sync (`get_changes_between`). Recommend failing the +whole call on unparseable non-empty lines. + +### M3. Iceberg executor-side manifest reads bypass the catalog's credential model + +`read_iceberg_manifest` builds a `FileIO` directly from the raw config map +(`iceberg_reader/distributed.rs:129-155`). This only works when the map happens to contain +FileIO-style keys (`s3.access-key-id`, …). Two real setups break: + +- **Glue catalog** users configure `aws_access_key_id`/`region_name` (per `catalog.rs` docs); those + keys are meaningless to `FileIO`, so executor reads run unauthenticated. +- **REST catalogs with vended credentials** (including Unity Catalog): storage credentials are + vended per-table by the catalog at `load_table` time. Bypassing the catalog loses them entirely. + +At minimum, translate known catalog-style keys to FileIO keys and document that vended-credential +catalogs require the full (slow) catalog path; ideally offer a mode that obtains FileIO from a +one-time `load_table` and reuses it. + +### M4. Iceberg delete-file semantics are not actionable + +`list_files_with_catalog` / `read_manifest_with_file_io` return delete files interleaved with data +files, distinguished only by `content_type`. Sequence numbers — required to decide *which* data +files a position/equality delete applies to — are not captured (`iceberg_reader/scan.rs:243-252`, +`distributed.rs:250-277` ignore `entry.sequence_number()`). A consumer cannot correctly apply +deletes from this output, and `record_count` on data files overcounts. Either expose +`sequence_number` (data + delete files) or document that tables with delete files are unsupported +and consider failing when `content_type != "data"` is encountered. + +### M5. Silent schema fallback on Iceberg time travel + +When a snapshot's `schema_id` doesn't resolve, both `read_schema_with_catalog` +(`iceberg_reader/scan.rs:305-319`) and `get_snapshot_info_with_catalog` +(`distributed.rs:185-190`) silently fall back to `current_schema()`. For time-travel reads this can +return a schema that does not match the snapshot's data with no warning. An error (or at least a +logged warning) would be safer. + +### M6. Resolved snapshot ID is dropped from Iceberg `listFiles` results + +`serialize_iceberg_entries` takes `_actual_snapshot_id` and ignores it +(`iceberg_reader/serialization.rs:23-27`). The per-entry `snapshot_id` is the snapshot that *added* +each file, so a caller listing "latest" has no way to learn which snapshot was actually read — +unlike Delta, which embeds `table_version` in every entry. This matters for consistent +list-then-poll patterns (`getCurrentSnapshotId` comparisons against an unknown baseline). + +Related inconsistency: the fallback for a missing per-entry snapshot id is +`manifest_file.added_snapshot_id` in `scan.rs:250` but `0` in `distributed.rs:275`. + +### M7. One-shot Tokio runtimes and clients per JNI call + +Every call builds a fresh runtime, object store / catalog client, and connection pool +(`delta_reader/distributed.rs:85-87` et al.). Iceberg's `scan.rs` additionally uses a +**multi-threaded** `Runtime::new()` (`scan.rs:157`, `281`, `371`) where every other entry point uses +`new_current_thread()` — inconsistent and heavier (spawns worker threads per call). For +streaming-poll usage (`get_current_version` "on every poll cycle") this means re-doing TLS/auth +setup each tick. The repo already has `runtime_manager.rs`; routing these through a shared runtime +and caching `ObjectStore` instances keyed by (scheme, bucket, creds-hash) would cut latency and FD +churn. If per-call runtimes are kept, they at least guarantee no cross-call state, but the +current_thread/multi-thread mix should be unified. + +### M8. Iceberg partition values for non-string types use Debug/raw representations + +`literal_to_string` (`iceberg_reader/scan.rs:102-120`) renders `Date` as its underlying epoch-day +int, `Timestamp` as micros, and falls back to `format!("{:?}")` for decimals, fixed, binary, and +non-primitive literals (producing values like `Decimal(12345)`). Predicate evaluation is pure string +compare on these (`common.rs`), so a user filtering `date = '2024-01-01'` against an +identity-partitioned date column silently matches nothing (the stored value is `"19723"`). Delta, by +contrast, carries partition values as the human-readable strings from the log. At minimum this +asymmetry needs documenting; better, render date/timestamp literals in ISO form to match user +expectations and Delta behavior. + +--- + +## Low severity / polish + +- **L1. Hidden-file filtering is inconsistent in the parquet reader.** `list_partition_files` + skips `.`/`_`-prefixed files (`parquet_reader/distributed.rs:220-224`), but root-level collection + (`:143-154`) and `read_schema_from_first_file`'s root loop (`:255-271`) do not. A + `_delta_log`-adjacent temp file like `_tmp.parquet` at the root would be listed and could be + picked as the schema source. +- **L2. Missing-size sentinel conflation.** Checkpoint adds with a missing `size` become `0` + (`delta_reader/distributed.rs:565`) and missing `num_records` becomes `-1` in serialization; `0` + is a legal size. Consider `-1` for unknown size too. +- **L3. `LastCheckpointInfo.size` is parsed but never used** (`distributed.rs:61-66`). Either + validate the checkpoint row count against it or drop the field. +- **L4. Delta "defensive" column mapping can double-translate.** `list_delta_files` applies + `apply_column_mapping` on top of delta-kernel output "just in case" (`scan.rs:95-104`). If kernel + already returns logical names and some physical name collides with another column's logical name, + keys get remapped incorrectly. Prefer trusting the kernel (add a test pinning its behavior) over + double-mapping. +- **L5. `get_jstring`-style duplication.** `extract_jlong_array` is duplicated verbatim in + `delta_reader/jni.rs:458-469` and `iceberg_reader/jni.rs:435-446`; the TANT `write_field_header` / + `write_string` helpers are copy-pasted in all three `serialization.rs` files. Move to `common.rs`. + (The `std::mem::forget(safe_arr)` in `extract_jlong_array` is unnecessary in jni 0.21 — dropping a + borrowed `JLongArray` wrapper does not delete the reference — but it is harmless.) +- **L6. TANT offsets are `u32`.** Serialization silently assumes buffers < 4 GiB; the JNI layer's + 2 GiB `i32::MAX` guard in `buffer_to_jbytearray` (`common.rs:45-78`) happens to protect it, but a + debug assert at the serialization layer would make the invariant local. Related: the + non-distributed `nativeListFiles` materializes the entire table file list in one buffer, so very + large tables (the code's own test data mentions 61M add files) will hit this guard — the + distributed API is the answer, but the error message won't say so. +- **L7. All errors surface as `java.lang.RuntimeException`** (`common.rs:16-19`). Callers cannot + distinguish "table not found" from "credentials rejected" from "unsupported feature" without + string matching. A small exception taxonomy (or an error-code prefix convention) would help the + Spark-facing layer. +- **L8. `nativeGetCurrentSnapshotId` swallows config-extraction errors.** + `iceberg_reader/jni.rs:221-224` returns `-1` without throwing when `extract_hashmap` fails, + whereas every sibling entry point throws. `-1` is not otherwise a legal return here (errors throw), + so a config marshalling bug becomes indistinguishable from... nothing — the caller just sees -1. +- **L9. Delta `get_changes_between` trusts the caller's `to_version`.** If `to_version` exceeds the + actual latest version, the loop fails with "Failed to read commit N.json: not found" rather than a + clear "version out of range" (`distributed.rs:285-309`). Cheap to pre-validate via the same HEAD + probe used elsewhere. +- **L10. `serialize_log_changes` hard-codes `table_version = 0`** for added entries + (`delta_reader/serialization.rs:255-259`); the Java `DeltaFileEntry` presumably exposes this as a + real-looking version. A `-1` sentinel would be less misleading. + +--- + +## Positive observations + +- The Arrow FFI export paths validate all target addresses up front and build every + `FFI_ArrowArray`/`FFI_ArrowSchema` before writing any, so a mid-export failure cannot leave Java + holding half-initialized structs (`delta_reader/distributed.rs:1006-1048`). +- Delta log replay (`read_post_checkpoint_changes_async`) correctly implements last-action-wins + per path with oldest-first ordering, including re-add-after-remove, and is well tested. +- Iceberg partition-value extraction correctly uses **each manifest's own partition spec** rather + than the table default, which is the right call for spec-evolved tables, and there is a test + documenting the intent. +- `list_commit_files_after`'s sequential HEAD probing is a smart O(k) alternative to listing a + 200K-object `_delta_log/` (subject to H5 above). +- The predicate engine has clearly documented missing-column semantics, `total_cmp` NaN handling, + and thorough tests including numeric-vs-lexicographic edge cases. +- `percent_decode` in the parquet reader handles multi-byte UTF-8 sequences correctly (accumulating + bytes before UTF-8 validation), with tests. + +--- + +## Suggested priorities + +1. H1/H2 (parquet multi-level partition pruning) — silent, total data loss for a mainstream layout. +2. H6 (compact mode dropping DV/content-type flags) — silent wrong results on tables with deletes. +3. H7/H8 (abfss parsing, credential chain) — hard blockers for common deployments. +4. H3/H4/H5 (Delta distributed protocol gaps) — correctness under multi-part checkpoints, newer + Delta features, and log cleanup. +5. M1 (percent-encoded keys) — latent, will surface as unreproducible "file not found" on paths + with spaces. diff --git a/native/src/common.rs b/native/src/common.rs index ade73484..498d8c68 100644 --- a/native/src/common.rs +++ b/native/src/common.rs @@ -47,7 +47,10 @@ pub fn buffer_to_jbytearray(env: &mut JNIEnv, buffer: &[u8]) -> jbyteArray { to_java_exception( env, &anyhow::anyhow!( - "Buffer too large for Java byte array: {} bytes exceeds i32::MAX", + "Buffer too large for Java byte array: {} bytes exceeds i32::MAX (2 GiB). \ + For very large table file listings, use the distributed APIs \ + (getSnapshotInfo + readCheckpointPart / listPartitionFiles) instead of \ + materializing the whole listing in one call.", buffer.len() ), ); @@ -205,6 +208,91 @@ pub fn build_storage_config(env: &mut JNIEnv, config_map: &JObject) -> DeltaStor } } +// --------------------------------------------------------------------------- +// TANT batch protocol helpers (shared by the table-format serializers) +// --------------------------------------------------------------------------- + +/// Write a TANT field header: name length (u16) + name bytes + type code (u8) +/// + value count (u16). Matches BatchDocumentReader on the Java side. +pub fn write_field_header(buf: &mut Vec, name: &str, field_type: u8, value_count: u16) { + let name_bytes = name.as_bytes(); + buf.extend_from_slice(&(name_bytes.len() as u16).to_ne_bytes()); + buf.extend_from_slice(name_bytes); + buf.push(field_type); + buf.extend_from_slice(&value_count.to_ne_bytes()); +} + +/// Write a length-prefixed (u32) UTF-8 string into a TANT buffer. +pub fn write_string(buf: &mut Vec, s: &str) { + let bytes = s.as_bytes(); + // TANT offsets/lengths are u32; buffer_to_jbytearray rejects buffers over + // i32::MAX so this cannot truncate in practice — assert the invariant locally + debug_assert!(bytes.len() <= u32::MAX as usize); + buf.extend_from_slice(&(bytes.len() as u32).to_ne_bytes()); + buf.extend_from_slice(bytes); +} + +/// Extract a Java long[] into a Vec. +pub fn extract_jlong_array( + env: &mut JNIEnv, + arr: &jni::sys::jlongArray, +) -> Result, anyhow::Error> { + let safe_arr = unsafe { jni::objects::JLongArray::from_raw(*arr) }; + let len = env.get_array_length(&safe_arr) + .map_err(|e| anyhow::anyhow!("Failed to get array length: {}", e))? as usize; + let mut buf = vec![0i64; len]; + env.get_long_array_region(&safe_arr, 0, &mut buf) + .map_err(|e| anyhow::anyhow!("Failed to read long array: {}", e))?; + // Dropping the borrowed JLongArray wrapper does not delete the reference + // in jni 0.21, but forget() keeps the borrow semantics explicit. + std::mem::forget(safe_arr); + Ok(buf) +} + +// --------------------------------------------------------------------------- +// Percent-decoding +// --------------------------------------------------------------------------- + +/// Percent-decoding for URL paths and partition values, supporting multi-byte +/// UTF-8. Accumulates percent-encoded bytes and decodes them as a UTF-8 +/// sequence, correctly handling characters like `%C3%A9` → `é`. +/// +/// Needed because `Url::path()` returns the percent-encoded form and +/// `object_store::path::Path::from` does NOT decode — passing the encoded +/// form through would target the wrong object key for paths containing +/// spaces, `+`, unicode, etc. +pub fn percent_decode(s: &str) -> String { + let bytes = s.as_bytes(); + let mut result = Vec::with_capacity(bytes.len()); + let mut i = 0; + + while i < bytes.len() { + if bytes[i] == b'%' && i + 2 < bytes.len() { + let hi = bytes[i + 1]; + let lo = bytes[i + 2]; + if let (Some(h), Some(l)) = (hex_val(hi), hex_val(lo)) { + result.push(h << 4 | l); + i += 3; + continue; + } + } + result.push(bytes[i]); + i += 1; + } + + String::from_utf8(result).unwrap_or_else(|_| s.to_string()) +} + +/// Convert an ASCII hex digit to its numeric value. +fn hex_val(b: u8) -> Option { + match b { + b'0'..=b'9' => Some(b - b'0'), + b'a'..=b'f' => Some(b - b'a' + 10), + b'A'..=b'F' => Some(b - b'A' + 10), + _ => None, + } +} + // --------------------------------------------------------------------------- // Partition predicate filtering // --------------------------------------------------------------------------- @@ -400,6 +488,77 @@ impl PartitionPredicate { } } } + + /// Evaluate this predicate against a *partial* set of partition values, + /// e.g. only the first level of a multi-level Hive partition path. + /// + /// Unlike `evaluate`, a column that is absent from `partition_values` is + /// treated as *unknown* (`None`) rather than excluded, because the value + /// may be bound at a deeper partition level. `And`/`Or`/`Not` combine + /// unknowns with Kleene three-valued logic. Callers should keep entries + /// whose result is `None` (indeterminate at this level). + pub fn evaluate_partial( + &self, + partition_values: &HashMap, + ) -> Option { + match self { + PartitionPredicate::Eq { column, value } => partition_values + .get(column) + .map(|v| v == value), + PartitionPredicate::Neq { column, value } => partition_values + .get(column) + .map(|v| v != value), + PartitionPredicate::Gt { column, value, r#type } => partition_values + .get(column) + .map(|v| compare(v, value, r#type) == std::cmp::Ordering::Greater), + PartitionPredicate::Gte { column, value, r#type } => partition_values + .get(column) + .map(|v| compare(v, value, r#type) != std::cmp::Ordering::Less), + PartitionPredicate::Lt { column, value, r#type } => partition_values + .get(column) + .map(|v| compare(v, value, r#type) == std::cmp::Ordering::Less), + PartitionPredicate::Lte { column, value, r#type } => partition_values + .get(column) + .map(|v| compare(v, value, r#type) != std::cmp::Ordering::Greater), + PartitionPredicate::In { column, values } => partition_values + .get(column) + .map(|v| values.contains(v)), + // A column missing at this level may still appear deeper, so + // null-ness cannot be decided from a partial path. + PartitionPredicate::IsNull { column } | PartitionPredicate::IsNotNull { column } => { + if partition_values.contains_key(column) { + Some(matches!(self, PartitionPredicate::IsNotNull { .. })) + } else { + None + } + } + PartitionPredicate::And { filters } => { + let mut all_true = true; + for f in filters { + match f.evaluate_partial(partition_values) { + Some(false) => return Some(false), + Some(true) => {} + None => all_true = false, + } + } + if all_true { Some(true) } else { None } + } + PartitionPredicate::Or { filters } => { + let mut all_false = true; + for f in filters { + match f.evaluate_partial(partition_values) { + Some(true) => return Some(true), + Some(false) => {} + None => all_false = false, + } + } + if all_false { Some(false) } else { None } + } + PartitionPredicate::Not { filter } => { + filter.evaluate_partial(partition_values).map(|b| !b) + } + } + } } /// Parse an optional predicate JSON string into a PartitionPredicate. @@ -462,6 +621,90 @@ mod tests { .collect() } + // -- Partial (three-valued) evaluation tests -- + + fn parse(json: &str) -> PartitionPredicate { + serde_json::from_str(json).unwrap() + } + + #[test] + fn test_partial_eq_known_and_unknown() { + let pred = parse(r#"{"op": "eq", "column": "month", "value": "01"}"#); + // Column bound at this level → decidable + assert_eq!(pred.evaluate_partial(&pv(&[("month", "01")])), Some(true)); + assert_eq!(pred.evaluate_partial(&pv(&[("month", "02")])), Some(false)); + // Column not bound at this level (deeper partition) → unknown + assert_eq!(pred.evaluate_partial(&pv(&[("year", "2024")])), None); + } + + #[test] + fn test_partial_and_deeper_column_keeps_matching_prefix() { + // The H1 scenario: table partitioned by year/month, first-level + // listing only binds `year`. AND(year=2024, month=01) must not + // prune year=2024/ (month unknown) but must prune year=2023/. + let pred = parse( + r#"{"op": "and", "filters": [ + {"op": "eq", "column": "year", "value": "2024"}, + {"op": "eq", "column": "month", "value": "01"} + ]}"#, + ); + assert_eq!(pred.evaluate_partial(&pv(&[("year", "2024")])), None); + assert_eq!(pred.evaluate_partial(&pv(&[("year", "2023")])), Some(false)); + // Full values → decidable both ways + assert_eq!( + pred.evaluate_partial(&pv(&[("year", "2024"), ("month", "01")])), + Some(true) + ); + assert_eq!( + pred.evaluate_partial(&pv(&[("year", "2024"), ("month", "02")])), + Some(false) + ); + } + + #[test] + fn test_partial_or_kleene() { + let pred = parse( + r#"{"op": "or", "filters": [ + {"op": "eq", "column": "year", "value": "2024"}, + {"op": "eq", "column": "month", "value": "01"} + ]}"#, + ); + // year matches → true regardless of unknown month + assert_eq!(pred.evaluate_partial(&pv(&[("year", "2024")])), Some(true)); + // year doesn't match, month unknown → unknown + assert_eq!(pred.evaluate_partial(&pv(&[("year", "2023")])), None); + } + + #[test] + fn test_partial_not_propagates_unknown() { + let pred = parse( + r#"{"op": "not", "filter": {"op": "eq", "column": "month", "value": "01"}}"#, + ); + assert_eq!(pred.evaluate_partial(&pv(&[("year", "2024")])), None); + assert_eq!(pred.evaluate_partial(&pv(&[("month", "01")])), Some(false)); + assert_eq!(pred.evaluate_partial(&pv(&[("month", "02")])), Some(true)); + } + + #[test] + fn test_partial_null_checks_unknown_when_missing() { + let is_null = parse(r#"{"op": "is_null", "column": "month"}"#); + let is_not_null = parse(r#"{"op": "is_not_null", "column": "month"}"#); + // Missing at this level: may exist deeper → indeterminate + assert_eq!(is_null.evaluate_partial(&pv(&[("year", "2024")])), None); + assert_eq!(is_not_null.evaluate_partial(&pv(&[("year", "2024")])), None); + // Present → decidable + assert_eq!(is_null.evaluate_partial(&pv(&[("month", "01")])), Some(false)); + assert_eq!(is_not_null.evaluate_partial(&pv(&[("month", "01")])), Some(true)); + } + + #[test] + fn test_partial_numeric_compare() { + let pred = parse(r#"{"op": "gt", "column": "year", "value": "2022", "type": "long"}"#); + assert_eq!(pred.evaluate_partial(&pv(&[("year", "2024")])), Some(true)); + assert_eq!(pred.evaluate_partial(&pv(&[("year", "2020")])), Some(false)); + assert_eq!(pred.evaluate_partial(&pv(&[("month", "01")])), None); + } + // -- Deserialization tests -- #[test] diff --git a/native/src/delta_reader/distributed.rs b/native/src/delta_reader/distributed.rs index 76269243..203f5132 100644 --- a/native/src/delta_reader/distributed.rs +++ b/native/src/delta_reader/distributed.rs @@ -57,10 +57,11 @@ pub struct DeltaLogChanges { } /// Parsed _last_checkpoint JSON content. +/// (The spec's `size` field — checkpoint action count — is not parsed because +/// nothing consumes it.) #[derive(Debug, Clone)] struct LastCheckpointInfo { version: u64, - size: u64, parts: Option, num_of_add_files: Option, } @@ -104,12 +105,40 @@ pub fn get_snapshot_info( ); // Step 2: Construct checkpoint part paths - let checkpoint_part_paths = + let mut checkpoint_version = checkpoint_info.version; + let mut checkpoint_part_paths = construct_checkpoint_paths(checkpoint_info.version, checkpoint_info.parts); // Step 3: List post-checkpoint commit files - let commit_file_paths = - list_commit_files_after(&store, &log_prefix, checkpoint_info.version).await?; + let mut commit_file_paths = + list_commit_files_after(&store, &log_prefix, checkpoint_version).await?; + + // Step 3b: Guard against a stale _last_checkpoint. If a newer + // checkpoint exists and log cleanup removed the commit JSONs between + // the stale one and it, the sequential probe above stopped early and + // this snapshot would silently be outdated. + let probed_version = checkpoint_version + commit_file_paths.len() as u64; + let markers = find_log_markers_after(&store, &log_prefix, probed_version).await?; + if let Some((newer_cp_version, newer_cp_files)) = markers.newest_checkpoint { + debug_println!( + "🔧 DELTA_DIST: _last_checkpoint is stale (points at {}, newest checkpoint is {})", + checkpoint_version, newer_cp_version + ); + checkpoint_version = newer_cp_version; + checkpoint_part_paths = newer_cp_files; + commit_file_paths = + list_commit_files_after(&store, &log_prefix, checkpoint_version).await?; + } else if let Some(max_commit) = markers.max_commit_version { + // Commits exist beyond a gap with no covering checkpoint — the + // contiguous chain is broken (cleanup/corruption); fail loudly + // rather than silently returning an old snapshot. + anyhow::bail!( + "Delta log is not contiguous: commit version {} exists but versions after {} \ + are missing and no newer checkpoint covers the gap", + max_commit, + probed_version + ); + } debug_println!( "🔧 DELTA_DIST: {} checkpoint parts, {} post-checkpoint commits", @@ -117,11 +146,58 @@ pub fn get_snapshot_info( commit_file_paths.len() ); - // Step 4: Read schema from first checkpoint part - let first_part = &checkpoint_part_paths[0]; - let first_part_obj_path = make_log_path(&log_prefix, first_part); - let (schema_json, partition_columns) = - read_metadata_from_checkpoint(&store, &first_part_obj_path).await?; + // Step 4: Read metaData + protocol, scanning checkpoint parts in + // order — the Delta protocol does not guarantee which part of a + // multi-part checkpoint contains these action rows. + let mut metadata: Option = None; + let mut protocol: Option = None; + for part in &checkpoint_part_paths { + let part_obj_path = make_log_path(&log_prefix, part); + let (m, p) = read_metadata_from_checkpoint(&store, &part_obj_path).await?; + if metadata.is_none() { + metadata = m; + } + if protocol.is_none() { + protocol = p; + } + if metadata.is_some() && protocol.is_some() { + break; + } + } + let metadata = metadata.ok_or_else(|| { + anyhow::anyhow!( + "No metaData row found in any part of checkpoint version {} for table {}", + checkpoint_version, + url_str + ) + })?; + + // Step 4b: Reject tables whose reader requirements this reader does + // not implement (v2 checkpoints, unknown future features, id-mode + // column mapping) instead of failing confusingly or returning wrong + // results later. + if let Some(ref proto) = protocol { + validate_protocol(proto)?; + } else { + debug_println!( + "🔧 DELTA_DIST: WARNING: no protocol row found in checkpoint {} — skipping reader feature validation", + checkpoint_version + ); + } + let column_mapping_mode = metadata + .configuration + .get("delta.columnMapping.mode") + .map(|s| s.as_str()) + .unwrap_or("none"); + if column_mapping_mode == "id" { + anyhow::bail!( + "Unsupported Delta table: column mapping mode 'id' is not implemented \ + (only 'name' and 'none' are supported)" + ); + } + + let schema_json = metadata.schema_json; + let partition_columns = metadata.partition_columns; // Step 5: Build column mapping and translate partition column names let column_mapping = build_column_mapping(&schema_json); @@ -140,13 +216,21 @@ pub fn get_snapshot_info( partition_columns ); + // numOfAddFiles from _last_checkpoint only describes the checkpoint it + // points at — drop it if the stale-checkpoint guard switched to a newer one + let num_add_files = if checkpoint_version == checkpoint_info.version { + checkpoint_info.num_of_add_files + } else { + None + }; + Ok(DeltaSnapshotInfo { - version: checkpoint_info.version, + version: checkpoint_version, schema_json, partition_columns, checkpoint_part_paths, commit_file_paths, - num_add_files: checkpoint_info.num_of_add_files, + num_add_files, column_mapping, }) }) @@ -262,7 +346,29 @@ pub fn get_current_version( }; let commit_files = list_commit_files_after(&store, &log_prefix, base_version).await?; - let current_version = base_version + commit_files.len() as u64; + let mut current_version = base_version + commit_files.len() as u64; + + // Stale-_last_checkpoint guard: if a newer checkpoint exists past the + // contiguous commit chain (log cleanup removed the commits in + // between), continue probing from it instead of silently returning + // the old version. + let markers = find_log_markers_after(&store, &log_prefix, current_version).await?; + if let Some((newer_cp_version, _)) = markers.newest_checkpoint { + let newer_commits = + list_commit_files_after(&store, &log_prefix, newer_cp_version).await?; + debug_println!( + "🔧 DELTA_DIST: _last_checkpoint is stale (probe reached {}, newest checkpoint is {})", + current_version, newer_cp_version + ); + current_version = newer_cp_version + newer_commits.len() as u64; + } else if let Some(max_commit) = markers.max_commit_version { + anyhow::bail!( + "Delta log is not contiguous: commit version {} exists but versions after {} \ + are missing and no newer checkpoint covers the gap", + max_commit, + current_version + ); + } debug_println!( "🔧 DELTA_DIST: Current version={} (checkpoint={}, +{} commits)", @@ -336,13 +442,11 @@ fn parse_last_checkpoint(json_str: &str) -> Result { let version = v["version"] .as_u64() .ok_or_else(|| anyhow::anyhow!("_last_checkpoint missing 'version' field"))?; - let size = v["size"].as_u64().unwrap_or(0); let parts = v["parts"].as_u64(); let num_of_add_files = v["numOfAddFiles"].as_u64(); Ok(LastCheckpointInfo { version, - size, parts, num_of_add_files, }) @@ -408,6 +512,69 @@ async fn list_commit_files_after( Ok(commit_files) } +/// Newer log markers found past the sequentially-probed version (see +/// `find_log_markers_after`). +#[derive(Debug, Default)] +struct NewerLogMarkers { + /// Highest checkpoint version found, with its actual file names + newest_checkpoint: Option<(u64, Vec)>, + /// Highest commit JSON version found + max_commit_version: Option, +} + +/// Guard against a stale `_last_checkpoint`: one LIST bounded by an offset key +/// so only objects sorting after `{after_version:020}.json` are returned +/// (normally zero). The Delta spec allows `_last_checkpoint` to lag behind the +/// newest checkpoint; if log cleanup then removes the commit JSONs in between, +/// the sequential HEAD probe stops early and would silently report an old +/// version. This detects any newer checkpoint/commit markers. +async fn find_log_markers_after( + store: &Arc, + log_prefix: &str, + after_version: u64, +) -> Result { + use futures::TryStreamExt; + + let prefix = ObjectPath::from(log_prefix.to_string()); + let offset = make_log_path(log_prefix, &format!("{:020}.json", after_version)); + + let mut markers = NewerLogMarkers::default(); + let mut checkpoint_files: HashMap> = HashMap::new(); + + let mut stream = store.list_with_offset(Some(&prefix), &offset); + while let Some(meta) = stream.try_next().await.map_err(|e| { + anyhow::anyhow!("Failed to list {} for staleness check: {}", log_prefix, e) + })? { + let name = match meta.location.filename() { + Some(n) => n, + None => continue, + }; + if name.len() < 20 { + continue; + } + let version = match name[..20].parse::() { + Ok(v) => v, + Err(_) => continue, // _last_checkpoint, _sidecars/, etc. + }; + if version <= after_version { + continue; + } + if name.ends_with(".json") { + markers.max_commit_version = markers.max_commit_version.max(Some(version)); + } else if name.contains(".checkpoint.") && name.ends_with(".parquet") { + checkpoint_files.entry(version).or_default().push(name.to_string()); + } + } + + if let Some((&version, _)) = checkpoint_files.iter().max_by_key(|(v, _)| **v) { + let mut files = checkpoint_files.remove(&version).unwrap_or_default(); + files.sort(); + markers.newest_checkpoint = Some((version, files)); + } + + Ok(markers) +} + /// Find the current version of a table that has no _last_checkpoint. /// /// Checks if version 0 exists, then counts how many sequential versions follow. @@ -428,11 +595,80 @@ async fn find_latest_version_from_zero( } } -/// Read metaData from the first checkpoint part (column-projected, single row). +/// Table metadata extracted from a checkpoint's `metaData` action row. +#[derive(Debug, Clone)] +struct CheckpointMetadata { + schema_json: String, + partition_columns: Vec, + /// Table configuration (delta.columnMapping.mode etc.) + configuration: HashMap, +} + +/// Protocol information extracted from a checkpoint's `protocol` action row. +#[derive(Debug, Clone)] +struct ProtocolInfo { + min_reader_version: i64, + reader_features: Vec, +} + +/// Reader features this hand-rolled distributed reader actually implements. +/// +/// - `columnMapping`: name mode only (id mode is rejected via table +/// configuration in `get_snapshot_info`) +/// - `deletionVectors`: DVs are not applied, but `has_deletion_vector` is +/// surfaced on every entry (including compact mode) so consumers can act +/// - `timestampNtz` / `variantType-preview` style type features only affect +/// how consumers read the data files, not the log — accept the common ones +/// that require no log-format changes +/// - `v2Checkpoint` is intentionally absent: checkpoint files would be +/// UUID-named with sidecar actions, which construct_checkpoint_paths does +/// not understand +const SUPPORTED_READER_FEATURES: &[&str] = &[ + "columnMapping", + "deletionVectors", + "timestampNtz", + "vacuumProtocolCheck", +]; + +/// Validate a table's protocol against what this reader implements. +/// +/// Converts silent/confusing failures on unsupported tables (v2 checkpoints, +/// future reader features) into a clear error. The non-distributed +/// list_delta_files path is protected by delta-kernel's own protocol check. +fn validate_protocol(protocol: &ProtocolInfo) -> Result<()> { + if protocol.min_reader_version > 3 { + anyhow::bail!( + "Unsupported Delta table: minReaderVersion {} exceeds the maximum supported (3)", + protocol.min_reader_version + ); + } + let unsupported: Vec<&str> = protocol + .reader_features + .iter() + .map(|s| s.as_str()) + .filter(|f| !SUPPORTED_READER_FEATURES.contains(f)) + .collect(); + if !unsupported.is_empty() { + anyhow::bail!( + "Unsupported Delta reader feature(s) {:?}: this table requires reader capabilities \ + the distributed reader does not implement (supported: {:?})", + unsupported, + SUPPORTED_READER_FEATURES + ); + } + Ok(()) +} + +/// Read the `metaData` and `protocol` action rows from one checkpoint part +/// (column-projected). +/// +/// Returns `Ok((None, None))` when this part contains neither action — for +/// multi-part checkpoints the Delta protocol does not guarantee which part +/// holds them, so the caller must scan subsequent parts. async fn read_metadata_from_checkpoint( store: &Arc, checkpoint_path: &ObjectPath, -) -> Result<(String, Vec)> { +) -> Result<(Option, Option)> { use parquet::arrow::async_reader::ParquetObjectReader; use parquet::arrow::ParquetRecordBatchStreamBuilder; use arrow_array::cast::AsArray; @@ -447,49 +683,97 @@ async fn read_metadata_from_checkpoint( let builder = ParquetRecordBatchStreamBuilder::new(reader).await?; let arrow_schema = builder.schema().clone(); - // Find the metaData column index + // Locate metaData/protocol columns; either may be absent from a part's + // schema (not an error — the actions can live in another part) let metadata_idx = arrow_schema .fields() .iter() - .position(|f| f.name() == "metaData") - .ok_or_else(|| anyhow::anyhow!("Checkpoint parquet has no 'metaData' column"))?; + .position(|f| f.name() == "metaData"); + let protocol_idx = arrow_schema + .fields() + .iter() + .position(|f| f.name() == "protocol"); + + let projected: Vec = [metadata_idx, protocol_idx].iter().flatten().copied().collect(); + if projected.is_empty() { + return Ok((None, None)); + } - // Project only metaData column let mask = parquet::arrow::ProjectionMask::roots( builder.parquet_schema(), - std::iter::once(metadata_idx), + projected.iter().copied(), ); let mut stream = builder.with_projection(mask).build()?; + let mut found_meta: Option = None; + let mut found_protocol: Option = None; + while let Some(batch_result) = stream.next().await { let batch = batch_result?; - let metadata_col = batch.column(0); - // metaData is a struct column — find non-null rows - let struct_array = metadata_col.as_struct(); + // Projected batch columns keep schema order: metaData (if present) + // comes before protocol + let mut batch_col = 0; + let meta_col = metadata_idx.map(|_| { + let c = batch.column(batch_col).clone(); + batch_col += 1; + c + }); + let proto_col = protocol_idx.map(|_| batch.column(batch_col).clone()); - for row in 0..struct_array.len() { - if struct_array.is_null(row) { - continue; + if let (None, Some(col)) = (&found_meta, &meta_col) { + let struct_array = col.as_struct(); + for row in 0..struct_array.len() { + if struct_array.is_null(row) { + continue; + } + let schema_json = extract_struct_string_field(struct_array, "schemaString", row) + .unwrap_or_default(); + if schema_json.is_empty() { + continue; + } + let partition_columns = + extract_struct_string_list_field(struct_array, "partitionColumns", row) + .unwrap_or_default(); + let configuration = + extract_struct_string_map_field(struct_array, "configuration", row); + found_meta = Some(CheckpointMetadata { + schema_json, + partition_columns, + configuration, + }); + break; } + } - // Extract schemaString and partitionColumns from the struct - let schema_json = extract_struct_string_field(struct_array, "schemaString", row) - .unwrap_or_default(); - let partition_columns = extract_struct_string_list_field(struct_array, "partitionColumns", row) - .unwrap_or_default(); - - if !schema_json.is_empty() { - return Ok((schema_json, partition_columns)); + if let (None, Some(col)) = (&found_protocol, &proto_col) { + let struct_array = col.as_struct(); + for row in 0..struct_array.len() { + if struct_array.is_null(row) { + continue; + } + let min_reader_version = + extract_struct_i64_field(struct_array, "minReaderVersion", row) + .or_else(|| extract_struct_i32_field(struct_array, "minReaderVersion", row)) + .unwrap_or(1); + let reader_features = + extract_struct_string_list_field(struct_array, "readerFeatures", row) + .unwrap_or_default(); + found_protocol = Some(ProtocolInfo { + min_reader_version, + reader_features, + }); + break; } } + + if found_meta.is_some() && found_protocol.is_some() { + break; + } } - Err(anyhow::anyhow!( - "No metaData row found in checkpoint {}", - checkpoint_path - )) + Ok((found_meta, found_protocol)) } /// Read one checkpoint parquet part and extract add file entries. @@ -562,7 +846,8 @@ fn extract_add_files_from_batch( continue; } - let size = extract_struct_i64_field(struct_array, "size", row).unwrap_or(0); + // -1 = unknown (0 is a legal file size), matching num_records' sentinel + let size = extract_struct_i64_field(struct_array, "size", row).unwrap_or(-1); let modification_time = extract_struct_i64_field(struct_array, "modificationTime", row).unwrap_or(0); @@ -610,21 +895,39 @@ async fn read_post_checkpoint_changes_async( for commit_file in &sorted_paths { let path = make_log_path(log_prefix, commit_file); - let result = store.get(&path).await - .map_err(|e| anyhow::anyhow!("Failed to read commit {}: {}", commit_file, e))?; + let result = match store.get(&path).await { + Ok(r) => r, + Err(object_store::Error::NotFound { .. }) => { + return Err(anyhow::anyhow!( + "Commit file {} not found — the requested version may exceed the table's \ + latest version, or the commit has been removed by log cleanup", + commit_file + )); + } + Err(e) => { + return Err(anyhow::anyhow!("Failed to read commit {}: {}", commit_file, e)); + } + }; let bytes = result.bytes().await?; let text = std::str::from_utf8(&bytes)?; - for line in text.lines() { + for (line_no, line) in text.lines().enumerate() { let line = line.trim(); if line.is_empty() { continue; } - let v: serde_json::Value = match serde_json::from_str(line) { - Ok(v) => v, - Err(_) => continue, - }; + // A malformed non-empty line means a truncated/corrupted commit + // (partial upload, torn write). Skipping it would silently + // produce an incomplete file list, so fail the whole call. + let v: serde_json::Value = serde_json::from_str(line).map_err(|e| { + anyhow::anyhow!( + "Malformed JSON at line {} of commit {}: {}", + line_no + 1, + commit_file, + e + ) + })?; if let Some(add) = v.get("add") { let file_path = add["path"].as_str().unwrap_or_default().to_string(); @@ -634,7 +937,8 @@ async fn read_post_checkpoint_changes_async( let entry = DeltaFileEntry { path: file_path.clone(), - size: add["size"].as_i64().unwrap_or(0), + // -1 = unknown (0 is a legal file size) + size: add["size"].as_i64().unwrap_or(-1), modification_time: add["modificationTime"].as_i64().unwrap_or(0), num_records: add["stats"] .as_str() @@ -678,8 +982,6 @@ fn extract_struct_string_field( field_name: &str, row: usize, ) -> Option { - use arrow_array::cast::AsArray; - let col_idx = struct_array .fields() .iter() @@ -723,6 +1025,52 @@ fn extract_struct_i64_field( } } +/// Extract an i32 field from a StructArray at the given row (as i64). +fn extract_struct_i32_field( + struct_array: &arrow_array::StructArray, + field_name: &str, + row: usize, +) -> Option { + let col_idx = struct_array + .fields() + .iter() + .position(|f| f.name() == field_name)?; + let col = struct_array.column(col_idx); + + if col.is_null(row) { + return None; + } + + col.as_any() + .downcast_ref::() + .map(|arr| arr.value(row) as i64) +} + +/// Extract a string→string map field from a StructArray (e.g. metaData.configuration). +fn extract_struct_string_map_field( + struct_array: &arrow_array::StructArray, + field_name: &str, + row: usize, +) -> HashMap { + let col_idx = match struct_array + .fields() + .iter() + .position(|f| f.name() == field_name) + { + Some(i) => i, + None => return HashMap::new(), + }; + let col = struct_array.column(col_idx); + if col.is_null(row) { + return HashMap::new(); + } + if let Some(map_arr) = col.as_any().downcast_ref::() { + parse_map_array(map_arr, row) + } else { + HashMap::new() + } +} + /// Extract a string list field from a StructArray (for partitionColumns). fn extract_struct_string_list_field( struct_array: &arrow_array::StructArray, @@ -924,9 +1272,13 @@ pub fn parse_column_mapping_json(json: Option<&str>) -> HashMap // ─── Path helpers ─────────────────────────────────────────────────────────── /// Get the _delta_log prefix for an object store path. +/// +/// `Url::path()` is percent-encoded and `ObjectPath::from` does not decode, +/// so decode first — otherwise table paths containing spaces/unicode would +/// resolve to the wrong object keys. fn delta_log_prefix(url: &Url) -> String { - let path = url.path(); - let path = path.strip_prefix('/').unwrap_or(path); + let path = crate::common::percent_decode(url.path()); + let path = path.strip_prefix('/').unwrap_or(&path); let path = path.strip_suffix('/').unwrap_or(path); if path.is_empty() { "_delta_log".to_string() @@ -1061,7 +1413,6 @@ mod tests { let json = r#"{"version":94320,"size":61126995,"parts":1123,"numOfAddFiles":61126995}"#; let info = parse_last_checkpoint(json).unwrap(); assert_eq!(info.version, 94320); - assert_eq!(info.size, 61126995); assert_eq!(info.parts, Some(1123)); assert_eq!(info.num_of_add_files, Some(61126995)); } @@ -1071,11 +1422,57 @@ mod tests { let json = r#"{"version":100,"size":5000}"#; let info = parse_last_checkpoint(json).unwrap(); assert_eq!(info.version, 100); - assert_eq!(info.size, 5000); assert_eq!(info.parts, None); assert_eq!(info.num_of_add_files, None); } + #[test] + fn test_validate_protocol_basic_versions() { + // Reader versions 1-2 without features are fine + assert!(validate_protocol(&ProtocolInfo { + min_reader_version: 1, + reader_features: vec![], + }) + .is_ok()); + assert!(validate_protocol(&ProtocolInfo { + min_reader_version: 2, + reader_features: vec![], + }) + .is_ok()); + } + + #[test] + fn test_validate_protocol_supported_features() { + assert!(validate_protocol(&ProtocolInfo { + min_reader_version: 3, + reader_features: vec![ + "columnMapping".to_string(), + "deletionVectors".to_string(), + "timestampNtz".to_string(), + ], + }) + .is_ok()); + } + + #[test] + fn test_validate_protocol_rejects_v2_checkpoint() { + let err = validate_protocol(&ProtocolInfo { + min_reader_version: 3, + reader_features: vec!["v2Checkpoint".to_string()], + }) + .unwrap_err(); + assert!(err.to_string().contains("v2Checkpoint"), "{}", err); + } + + #[test] + fn test_validate_protocol_rejects_future_reader_version() { + assert!(validate_protocol(&ProtocolInfo { + min_reader_version: 4, + reader_features: vec![], + }) + .is_err()); + } + #[test] fn test_construct_checkpoint_paths_single() { let paths = construct_checkpoint_paths(100, None); @@ -1148,6 +1545,16 @@ mod tests { assert_eq!(delta_log_prefix(&url), "tmp/my_table/_delta_log"); } + #[test] + fn test_delta_log_prefix_percent_encoded_path() { + // Url::path() percent-encodes; the object key must be the decoded form + let url = Url::parse("file:///tmp/my table/").unwrap(); + assert_eq!(delta_log_prefix(&url), "tmp/my table/_delta_log"); + + let url = Url::parse("s3://bucket/path/caf%C3%A9/table/").unwrap(); + assert_eq!(delta_log_prefix(&url), "path/café/table/_delta_log"); + } + #[test] fn test_read_post_checkpoint_changes_empty() { // Empty commit list should return empty changes @@ -1226,7 +1633,7 @@ mod tests { adds: &[(&str, i64, i64, i64)], // (path, size, mod_time, num_records) ) { use arrow_array::{ - builder::{StringBuilder, Int64Builder, BooleanBuilder, MapBuilder}, + builder::{StringBuilder, Int64Builder, MapBuilder}, StructArray, RecordBatch, }; use arrow_schema::{DataType, Field, Fields, Schema}; diff --git a/native/src/delta_reader/engine.rs b/native/src/delta_reader/engine.rs index 1f1c7cf4..0607774b 100644 --- a/native/src/delta_reader/engine.rs +++ b/native/src/delta_reader/engine.rs @@ -46,7 +46,10 @@ pub fn create_object_store(url: &Url, config: &DeltaStorageConfig) -> Result = match scheme { "s3" | "s3a" => { - let mut builder = AmazonS3Builder::new() + // from_env() picks up AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY / + // AWS_SESSION_TOKEN / AWS_REGION etc., so deployments relying on + // env-var credentials work; explicit config below overrides. + let mut builder = AmazonS3Builder::from_env() .with_bucket_name( url .host_str() @@ -75,17 +78,36 @@ pub fn create_object_store(url: &Url, config: &DeltaStorageConfig) -> Result { - let mut builder = MicrosoftAzureBuilder::new() - .with_container_name( - url + // Canonical ABFS URLs are abfss://@.dfs.core.windows.net/path: + // the container is the USERNAME component and the host is the account + // endpoint. For az://container/path the host is the container. + let (container, account_from_url) = match scheme { + "abfs" | "abfss" if !url.username().is_empty() => { + let account = url .host_str() + .and_then(|h| h.split('.').next()) + .map(|s| s.to_string()); + (url.username().to_string(), account) + } + _ => ( + url.host_str() .ok_or_else(|| { anyhow::anyhow!("Azure URL missing container: {}", url) - })?, - ); + })? + .to_string(), + None, + ), + }; + + // from_env() picks up AZURE_STORAGE_ACCOUNT_NAME / _KEY etc.; + // explicit config below overrides. + let mut builder = MicrosoftAzureBuilder::from_env() + .with_container_name(&container); if let Some(ref account) = config.azure_account_name { builder = builder.with_account(account); + } else if let Some(account) = account_from_url { + builder = builder.with_account(account); } if let Some(ref key) = config.azure_access_key { builder = builder.with_access_key(key); diff --git a/native/src/delta_reader/jni.rs b/native/src/delta_reader/jni.rs index 50f71d83..2b0460a3 100644 --- a/native/src/delta_reader/jni.rs +++ b/native/src/delta_reader/jni.rs @@ -10,6 +10,7 @@ use jni::JNIEnv; use crate::common::{ to_java_exception, build_storage_config, buffer_to_jbytearray, extract_string_list, extract_optional_jstring, parse_optional_predicate, filter_by_predicate, + extract_jlong_array, }; use crate::debug_println; @@ -453,17 +454,3 @@ pub extern "system" fn Java_io_indextables_tantivy4java_delta_DeltaTableReader_n } } } - -/// Extract a Java long[] into a Vec. -fn extract_jlong_array(env: &mut JNIEnv, arr: &jlongArray) -> anyhow::Result> { - let safe_arr = unsafe { jni::objects::JLongArray::from_raw(*arr) }; - let len = env - .get_array_length(&safe_arr) - .map_err(|e| anyhow::anyhow!("Failed to get array length: {}", e))?; - let mut buf = vec![0i64; len as usize]; - env.get_long_array_region(&safe_arr, 0, &mut buf) - .map_err(|e| anyhow::anyhow!("Failed to read long array: {}", e))?; - // Prevent the safe wrapper from freeing the original Java array reference - std::mem::forget(safe_arr); - Ok(buf) -} diff --git a/native/src/delta_reader/scan.rs b/native/src/delta_reader/scan.rs index b54d8d58..04914fe7 100644 --- a/native/src/delta_reader/scan.rs +++ b/native/src/delta_reader/scan.rs @@ -94,6 +94,15 @@ pub fn list_delta_files( // Defensive column mapping: delta-kernel's ScanFile likely already uses logical // names, but apply mapping just in case to handle edge cases. + // + // NOTE: apply_column_mapping only rewrites keys that appear in the + // physical→logical map, so kernel-emitted logical names normally pass + // through untouched. The one theoretical hazard is a table where one + // column's LOGICAL name equals another column's PHYSICAL name — that key + // would be remapped incorrectly. Physical names in column-mapping tables + // are conventionally "col-", making such a collision practically + // impossible; if it is ever observed, remove this defensive pass and + // trust the kernel output instead. let column_mapping = super::distributed::build_column_mapping(&schema_json); if !column_mapping.is_empty() { debug_println!( diff --git a/native/src/delta_reader/serialization.rs b/native/src/delta_reader/serialization.rs index c3490f3d..0f2bd26b 100644 --- a/native/src/delta_reader/serialization.rs +++ b/native/src/delta_reader/serialization.rs @@ -5,6 +5,7 @@ // Rust→Java transfer via BatchDocumentReader.parseToMaps(). use super::scan::{DeltaFileEntry, DeltaSchemaField}; +use crate::common::{write_field_header, write_string}; /// Magic number for batch protocol validation ("TANT") const MAGIC_NUMBER: u32 = 0x54414E54; @@ -17,12 +18,14 @@ const FIELD_TYPE_JSON: u8 = 6; /// Serialize a list of DeltaFileEntry into the TANT byte buffer format. /// -/// Each entry becomes one "document" with either 7 fields (full) or 5 fields (compact): +/// Each entry becomes one "document" with either 7 fields (full) or 6 fields (compact): /// Full: path, size, modification_time, num_records, partition_values, has_deletion_vector, table_version -/// Compact: path, size, modification_time, num_records, table_version +/// Compact: path, size, modification_time, num_records, has_deletion_vector, table_version /// -/// Compact mode skips partition_values (JSON) and has_deletion_vector (BOOLEAN) -/// for callers that only need file identity and basic metadata. +/// Compact mode skips partition_values (JSON) for callers that only need file +/// identity and basic metadata. has_deletion_vector is correctness-critical +/// (scanning a DV'd file without it silently includes deleted rows) so it is +/// kept in both modes. pub fn serialize_delta_entries(entries: &[DeltaFileEntry], table_version: u64, compact: bool) -> Vec { let per_entry = if compact { 200 } else { 300 }; let estimated = 4 + entries.len() * per_entry + entries.len() * 4 + 12; @@ -35,7 +38,7 @@ pub fn serialize_delta_entries(entries: &[DeltaFileEntry], table_version: u64, c let mut offsets = Vec::with_capacity(entries.len()); for entry in entries { offsets.push(buf.len() as u32); - serialize_entry(&mut buf, entry, table_version, compact); + serialize_entry(&mut buf, entry, table_version as i64, compact); } // Offset table @@ -52,8 +55,10 @@ pub fn serialize_delta_entries(entries: &[DeltaFileEntry], table_version: u64, c buf } -fn serialize_entry(buf: &mut Vec, entry: &DeltaFileEntry, table_version: u64, compact: bool) { - let field_count: u16 = if compact { 5 } else { 7 }; +/// `table_version` uses -1 as the "unknown" sentinel (e.g. log-change entries +/// where the per-entry commit version is not tracked). +fn serialize_entry(buf: &mut Vec, entry: &DeltaFileEntry, table_version: i64, compact: bool) { + let field_count: u16 = if compact { 6 } else { 7 }; buf.extend_from_slice(&field_count.to_ne_bytes()); // 1. path (TEXT) @@ -78,15 +83,17 @@ fn serialize_entry(buf: &mut Vec, entry: &DeltaFileEntry, table_version: u64 write_field_header(buf, "partition_values", FIELD_TYPE_JSON, 1); let pv_json = serde_json::to_string(&entry.partition_values).unwrap_or_else(|_| "{}".to_string()); write_string(buf, &pv_json); - - // 6. has_deletion_vector (BOOLEAN) — skipped in compact mode - write_field_header(buf, "has_deletion_vector", FIELD_TYPE_BOOLEAN, 1); - buf.push(if entry.has_deletion_vector { 1 } else { 0 }); } + // has_deletion_vector (BOOLEAN) — always included: without it a consumer + // scanning the listed parquet files would silently read logically + // deleted rows. + write_field_header(buf, "has_deletion_vector", FIELD_TYPE_BOOLEAN, 1); + buf.push(if entry.has_deletion_vector { 1 } else { 0 }); + // table_version (INTEGER) — always included write_field_header(buf, "table_version", FIELD_TYPE_INTEGER, 1); - buf.extend_from_slice(&(table_version as i64).to_ne_bytes()); + buf.extend_from_slice(&table_version.to_ne_bytes()); } /// Serialize a list of DeltaSchemaField plus the raw schema JSON into the TANT byte buffer format. @@ -252,10 +259,12 @@ pub fn serialize_log_changes(changes: &super::distributed::DeltaLogChanges) -> V buf.extend_from_slice(&(changes.removed_paths.len() as i64).to_ne_bytes()); } - // Documents 1..N: added files (full format, version=0 since it's post-checkpoint) + // Documents 1..N: added files (full format). The per-entry commit version + // is not tracked during log replay, so use the -1 "unknown" sentinel + // rather than a real-looking version of 0. for entry in &changes.added_files { offsets.push(buf.len() as u32); - serialize_entry(&mut buf, entry, 0, false); + serialize_entry(&mut buf, entry, -1, false); } // Documents N+1..M: removed paths @@ -281,20 +290,6 @@ pub fn serialize_log_changes(changes: &super::distributed::DeltaLogChanges) -> V buf } -fn write_field_header(buf: &mut Vec, name: &str, field_type: u8, value_count: u16) { - let name_bytes = name.as_bytes(); - buf.extend_from_slice(&(name_bytes.len() as u16).to_ne_bytes()); - buf.extend_from_slice(name_bytes); - buf.push(field_type); - buf.extend_from_slice(&value_count.to_ne_bytes()); -} - -fn write_string(buf: &mut Vec, s: &str) { - let bytes = s.as_bytes(); - buf.extend_from_slice(&(bytes.len() as u32).to_ne_bytes()); - buf.extend_from_slice(bytes); -} - #[cfg(test)] mod tests { use super::*; @@ -438,10 +433,11 @@ mod tests { let compact_count = u32::from_ne_bytes([compact_buf[clen - 8], compact_buf[clen - 7], compact_buf[clen - 6], compact_buf[clen - 5]]); assert_eq!(compact_count, 1); - // Compact should NOT contain partition_values or has_deletion_vector field names + // Compact should NOT contain partition_values, but MUST keep the + // correctness-critical has_deletion_vector flag let compact_str = String::from_utf8_lossy(&compact_buf); assert!(!compact_str.contains("partition_values")); - assert!(!compact_str.contains("has_deletion_vector")); + assert!(compact_str.contains("has_deletion_vector")); // But compact SHOULD contain path, size, table_version assert!(compact_str.contains("path")); diff --git a/native/src/iceberg_reader/distributed.rs b/native/src/iceberg_reader/distributed.rs index e25ae882..ad538355 100644 --- a/native/src/iceberg_reader/distributed.rs +++ b/native/src/iceberg_reader/distributed.rs @@ -16,7 +16,7 @@ use iceberg::TableIdent; use crate::debug_println; use super::catalog::create_catalog; use super::scan::{ - parse_namespace, format_to_string, content_type_to_string, literal_to_string, + parse_namespace, format_to_string, content_type_to_string, literal_to_string_typed, IcebergFileEntry, }; @@ -142,11 +142,17 @@ pub fn read_iceberg_manifest( rt.block_on(async { // Build FileIO directly from config properties — no catalog or table load needed. - // The config map already contains the storage credential keys (s3.access-key-id, - // s3.secret-access-key, s3.region, adls.account-name, etc.) that FileIO needs. + // Catalog-style credential keys (aws_access_key_id, region_name, ...) are + // translated to the FileIO keys (s3.access-key-id, s3.region, ...) that this + // path understands; FileIO-style keys pass through unchanged. + // + // NOTE: catalogs that vend per-table storage credentials at load_table time + // (REST catalogs with credential vending, e.g. Unity Catalog) cannot be + // supported by this direct path — use the catalog-based listFiles() instead. + let props = translate_to_file_io_props(config); let file_io = FileIO::from_path(manifest_path) .map_err(|e| anyhow::anyhow!("Failed to create FileIO for {}: {}", manifest_path, e))? - .with_props(config.iter().map(|(k, v)| (k.as_str(), v.as_str()))) + .with_props(props.iter().map(|(k, v)| (k.as_str(), v.as_str()))) .build() .map_err(|e| anyhow::anyhow!("Failed to build FileIO: {}", e))?; @@ -154,6 +160,35 @@ pub fn read_iceberg_manifest( }) } +/// Translate catalog-style config keys to FileIO property keys so executor-side +/// manifest reads authenticate the same way the driver-side catalog path does. +/// +/// Keys already in FileIO form (s3.*, adls.*, gcs.*) are passed through +/// unchanged and take precedence over translated catalog-style keys. +fn translate_to_file_io_props(config: &HashMap) -> HashMap { + // (catalog-style key, FileIO key) + const KEY_MAP: &[(&str, &str)] = &[ + ("aws_access_key_id", "s3.access-key-id"), + ("aws_secret_access_key", "s3.secret-access-key"), + ("aws_session_token", "s3.session-token"), + ("region_name", "s3.region"), + ("s3_endpoint", "s3.endpoint"), + ]; + + let mut props: HashMap = HashMap::new(); + for (catalog_key, file_io_key) in KEY_MAP { + if let Some(v) = config.get(*catalog_key) { + props.insert(file_io_key.to_string(), v.clone()); + } + } + // Pass-through (and override) with any keys the caller already provided + // in FileIO form, plus everything else FileIO might understand. + for (k, v) in config { + props.insert(k.clone(), v.clone()); + } + props +} + // ─── Internal async functions ─────────────────────────────────────────────── /// Get snapshot info with a catalog reference (testable with MemoryCatalog). @@ -181,11 +216,17 @@ pub(crate) async fn get_snapshot_info_with_catalog( }; let actual_snap_id = snapshot.snapshot_id(); - // Get schema JSON + // Get schema JSON. A snapshot whose schema_id no longer resolves must NOT + // silently fall back to current_schema(): for time-travel reads that could + // return a schema that doesn't match the snapshot's data. let schema = match snapshot.schema_id() { - Some(schema_id) => metadata - .schema_by_id(schema_id) - .unwrap_or_else(|| metadata.current_schema()), + Some(schema_id) => metadata.schema_by_id(schema_id).ok_or_else(|| { + anyhow::anyhow!( + "Snapshot {} references schema_id {} which does not exist in table metadata", + actual_snap_id, + schema_id + ) + })?, None => metadata.current_schema(), }; let schema_json = serde_json::to_string(schema.as_ref()) @@ -245,6 +286,9 @@ async fn read_manifest_with_file_io( .map_err(|e| anyhow::anyhow!("Failed to parse manifest: {}", e))?; let manifest_partition_spec = manifest.metadata().partition_spec(); + let partition_type = manifest_partition_spec + .partition_type(manifest.metadata().schema()) + .ok(); let mut entries = Vec::new(); for entry in manifest.entries() { @@ -258,9 +302,13 @@ async fn read_manifest_with_file_io( let partition_fields = data_file.partition().fields(); for (idx, spec_field) in manifest_partition_spec.fields().iter().enumerate() { if let Some(Some(literal)) = partition_fields.get(idx) { + let field_type = partition_type + .as_ref() + .and_then(|st| st.fields().get(idx)) + .map(|f| f.field_type.as_ref()); partition_values.insert( spec_field.name.clone(), - literal_to_string(literal), + literal_to_string_typed(literal, field_type), ); } } @@ -272,7 +320,11 @@ async fn read_manifest_with_file_io( file_size_bytes: data_file.file_size_in_bytes() as i64, partition_values, content_type: content_type_to_string(data_file.content_type()), - snapshot_id: entry.snapshot_id().unwrap_or(0), + sequence_number: entry.sequence_number().unwrap_or(-1), + // The manifest-list context (added_snapshot_id) is not available + // when reading a single manifest file; -1 = unknown (matches the + // sentinel convention elsewhere, instead of a real-looking 0) + snapshot_id: entry.snapshot_id().unwrap_or(-1), }); } @@ -289,10 +341,10 @@ async fn read_manifest_with_file_io( /// Read an Iceberg manifest and export filtered entries via Arrow FFI. /// -/// Builds a flat RecordBatch with 7 columns: +/// Builds a flat RecordBatch with 8 columns: /// path (Utf8), file_format (Utf8), record_count (Int64), /// file_size_bytes (Int64), partition_values (Utf8/JSON), -/// content_type (Utf8), snapshot_id (Int64) +/// content_type (Utf8), snapshot_id (Int64), sequence_number (Int64) /// /// Returns the number of rows written. pub fn read_iceberg_manifest_arrow_ffi( @@ -306,7 +358,7 @@ pub fn read_iceberg_manifest_arrow_ffi( use arrow_array::{StringArray, Int64Array, Array}; use arrow_schema::{DataType, Field}; - const NUM_COLS: usize = 7; + const NUM_COLS: usize = 8; if array_addrs.len() < NUM_COLS || schema_addrs.len() < NUM_COLS { anyhow::bail!( @@ -337,6 +389,7 @@ pub fn read_iceberg_manifest_arrow_ffi( let pv_refs: Vec<&str> = pvs.iter().map(|s| s.as_str()).collect(); let content_types: Vec<&str> = entries.iter().map(|e| e.content_type.as_str()).collect(); let snap_ids: Vec = entries.iter().map(|e| e.snapshot_id).collect(); + let seq_nums: Vec = entries.iter().map(|e| e.sequence_number).collect(); let arrays: Vec<(Arc, Field)> = vec![ (Arc::new(StringArray::from(paths)), Field::new("path", DataType::Utf8, false)), @@ -346,6 +399,7 @@ pub fn read_iceberg_manifest_arrow_ffi( (Arc::new(StringArray::from(pv_refs)), Field::new("partition_values", DataType::Utf8, false)), (Arc::new(StringArray::from(content_types)), Field::new("content_type", DataType::Utf8, false)), (Arc::new(Int64Array::from(snap_ids)), Field::new("snapshot_id", DataType::Int64, false)), + (Arc::new(Int64Array::from(seq_nums)), Field::new("sequence_number", DataType::Int64, false)), ]; // 4. Validate ALL addresses upfront before writing anything. diff --git a/native/src/iceberg_reader/jni.rs b/native/src/iceberg_reader/jni.rs index b2d6c735..2d12a022 100644 --- a/native/src/iceberg_reader/jni.rs +++ b/native/src/iceberg_reader/jni.rs @@ -10,6 +10,7 @@ use jni::JNIEnv; use crate::common::{ to_java_exception, extract_hashmap, buffer_to_jbytearray, extract_optional_jstring, parse_optional_predicate, filter_by_predicate, + extract_jlong_array, }; use crate::debug_println; @@ -220,7 +221,10 @@ pub extern "system" fn Java_io_indextables_tantivy4java_iceberg_IcebergTableRead }; let config = match extract_hashmap(&mut env, &config_map) { Ok(m) => m, - Err(_) => return -1, + Err(e) => { + to_java_exception(&mut env, &anyhow::anyhow!("Failed to extract config map: {}", e)); + return -1; + } }; match get_current_iceberg_snapshot_id(&catalog_str, &config, &namespace_str, &table_str) { @@ -339,7 +343,9 @@ pub extern "system" fn Java_io_indextables_tantivy4java_iceberg_IcebergTableRead "🔧 ICEBERG_JNI: Read {} entries from manifest (after filtering)", entries.len() ); - let buffer = serialize_iceberg_entries(&entries, 0, compact != 0); + // -1 = resolved snapshot unknown: reading a single manifest has no + // snapshot-resolution context (contrast with nativeListFiles) + let buffer = serialize_iceberg_entries(&entries, -1, compact != 0); buffer_to_jbytearray(&mut env, &buffer) } Err(e) => { @@ -430,17 +436,3 @@ pub extern "system" fn Java_io_indextables_tantivy4java_iceberg_IcebergTableRead } } } - -/// Extract a Java long[] into a Vec. -fn extract_jlong_array(env: &mut JNIEnv, arr: &jlongArray) -> anyhow::Result> { - let safe_arr = unsafe { jni::objects::JLongArray::from_raw(*arr) }; - let len = env - .get_array_length(&safe_arr) - .map_err(|e| anyhow::anyhow!("Failed to get array length: {}", e))?; - let mut buf = vec![0i64; len as usize]; - env.get_long_array_region(&safe_arr, 0, &mut buf) - .map_err(|e| anyhow::anyhow!("Failed to read long array: {}", e))?; - // Prevent the safe wrapper from freeing the original Java array reference - std::mem::forget(safe_arr); - Ok(buf) -} diff --git a/native/src/iceberg_reader/scan.rs b/native/src/iceberg_reader/scan.rs index ccddd806..afe6d7ca 100644 --- a/native/src/iceberg_reader/scan.rs +++ b/native/src/iceberg_reader/scan.rs @@ -30,6 +30,11 @@ pub struct IcebergFileEntry { pub partition_values: HashMap, /// Content type: "data", "equality_deletes", "position_deletes" pub content_type: String, + /// Data sequence number (-1 if unknown). Required to decide which data + /// files a position/equality delete applies to: a delete file applies to + /// data files with a strictly lower (position) or lower-or-equal + /// (equality) sequence number. + pub sequence_number: i64, /// Snapshot ID that added this file pub snapshot_id: i64, } @@ -99,6 +104,11 @@ pub(crate) fn content_type_to_string(ct: DataContentType) -> String { } /// Convert an Iceberg Literal to a string for partition values. +/// +/// `PrimitiveLiteral` does not carry the logical type (a Date is physically +/// an `Int` of epoch-days), so this untyped fallback renders temporal and +/// decimal values as raw integers. Prefer `literal_to_string_typed` with the +/// partition field's result type whenever available. pub(crate) fn literal_to_string(lit: &iceberg::spec::Literal) -> String { match lit { iceberg::spec::Literal::Primitive(p) => { @@ -119,6 +129,74 @@ pub(crate) fn literal_to_string(lit: &iceberg::spec::Literal) -> String { } } +/// Convert an Iceberg Literal to a string using the partition field's result +/// type, rendering dates/timestamps in ISO form and decimals with their scale +/// applied — matching user expectations and Delta's human-readable partition +/// values (a `date = '2024-01-01'` predicate would otherwise never match the +/// raw epoch-day value "19723"). +pub(crate) fn literal_to_string_typed( + lit: &iceberg::spec::Literal, + result_type: Option<&iceberg::spec::Type>, +) -> String { + use iceberg::spec::{Literal, PrimitiveLiteral, PrimitiveType, Type}; + + if let (Literal::Primitive(p), Some(Type::Primitive(pt))) = (lit, result_type) { + match (pt, p) { + (PrimitiveType::Date, PrimitiveLiteral::Int(days)) => { + if let Some(dt) = chrono::DateTime::from_timestamp(*days as i64 * 86_400, 0) { + return dt.date_naive().to_string(); // YYYY-MM-DD + } + } + ( + PrimitiveType::Timestamp | PrimitiveType::Timestamptz, + PrimitiveLiteral::Long(micros), + ) => { + if let Some(dt) = chrono::DateTime::from_timestamp_micros(*micros) { + return dt.naive_utc().to_string(); // YYYY-MM-DD HH:MM:SS[.ffffff] + } + } + ( + PrimitiveType::TimestampNs | PrimitiveType::TimestamptzNs, + PrimitiveLiteral::Long(nanos), + ) => { + let dt = chrono::DateTime::from_timestamp_nanos(*nanos); + return dt.naive_utc().to_string(); + } + (PrimitiveType::Decimal { scale, .. }, PrimitiveLiteral::Int128(v)) => { + return format_decimal(*v, *scale); + } + _ => {} + } + } + literal_to_string(lit) +} + +/// Format an unscaled i128 decimal value with the given scale, +/// e.g. (12345, 2) → "123.45". +fn format_decimal(unscaled: i128, scale: u32) -> String { + let negative = unscaled < 0; + let digits = unscaled.unsigned_abs().to_string(); + let scale = scale as usize; + let mut s = String::new(); + if negative { + s.push('-'); + } + if scale == 0 { + s.push_str(&digits); + } else if digits.len() > scale { + s.push_str(&digits[..digits.len() - scale]); + s.push('.'); + s.push_str(&digits[digits.len() - scale..]); + } else { + s.push_str("0."); + for _ in 0..(scale - digits.len()) { + s.push('0'); + } + s.push_str(&digits); + } + s +} + /// Convert an Iceberg Type to a human-readable string. /// Primitive types return simple names; complex types return JSON. fn type_to_string(ty: &iceberg::spec::Type) -> String { @@ -154,7 +232,9 @@ pub fn list_iceberg_files( namespace, table_name, snapshot_id ); - let rt = tokio::runtime::Runtime::new() + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() .map_err(|e| anyhow::anyhow!("Failed to create Tokio runtime: {}", e))?; rt.block_on(async { @@ -218,6 +298,12 @@ pub(crate) async fn list_files_with_catalog( // critical for tables that have undergone partition spec evolution. let manifest_partition_spec = manifest.metadata().partition_spec(); + // Resolve the partition fields' result types (from the manifest's own + // schema) so temporal/decimal values render human-readably. + let partition_type = manifest_partition_spec + .partition_type(manifest.metadata().schema()) + .ok(); + for entry in manifest.entries() { // Skip deleted entries if entry.status() == ManifestStatus::Deleted { @@ -233,9 +319,13 @@ pub(crate) async fn list_files_with_catalog( let partition_fields = data_file.partition().fields(); for (idx, spec_field) in manifest_partition_spec.fields().iter().enumerate() { if let Some(Some(literal)) = partition_fields.get(idx) { + let field_type = partition_type + .as_ref() + .and_then(|st| st.fields().get(idx)) + .map(|f| f.field_type.as_ref()); partition_values.insert( spec_field.name.clone(), - literal_to_string(literal), + literal_to_string_typed(literal, field_type), ); } } @@ -247,6 +337,7 @@ pub(crate) async fn list_files_with_catalog( file_size_bytes: data_file.file_size_in_bytes() as i64, partition_values, content_type: content_type_to_string(data_file.content_type()), + sequence_number: entry.sequence_number().unwrap_or(-1), snapshot_id: entry.snapshot_id().unwrap_or(manifest_file.added_snapshot_id), }); } @@ -278,7 +369,9 @@ pub fn read_iceberg_schema( namespace, table_name, snapshot_id ); - let rt = tokio::runtime::Runtime::new() + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() .map_err(|e| anyhow::anyhow!("Failed to create Tokio runtime: {}", e))?; rt.block_on(async { @@ -301,13 +394,20 @@ pub(crate) async fn read_schema_with_catalog( let metadata = table.metadata(); - // Get the schema — use snapshot-specific schema if available, otherwise current + // Get the schema — use snapshot-specific schema if available, otherwise current. + // A snapshot whose schema_id no longer resolves must NOT silently fall back + // to current_schema(): for time-travel reads that could return a schema + // that does not match the snapshot's data. let schema = if let Some(snap_id) = snapshot_id { if let Some(snapshot) = metadata.snapshot_by_id(snap_id) { - // Try to get schema for this snapshot's schema_id if let Some(schema_id) = snapshot.schema_id() { - metadata.schema_by_id(schema_id) - .unwrap_or_else(|| metadata.current_schema()) + metadata.schema_by_id(schema_id).ok_or_else(|| { + anyhow::anyhow!( + "Snapshot {} references schema_id {} which does not exist in table metadata", + snap_id, + schema_id + ) + })? } else { metadata.current_schema() } @@ -368,7 +468,9 @@ pub fn list_iceberg_snapshots( namespace, table_name ); - let rt = tokio::runtime::Runtime::new() + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() .map_err(|e| anyhow::anyhow!("Failed to create Tokio runtime: {}", e))?; rt.block_on(async { @@ -455,6 +557,55 @@ mod tests { assert_eq!(content_type_to_string(DataContentType::PositionDeletes), "position_deletes"); } + #[test] + fn test_literal_to_string_typed_date_iso() { + use iceberg::spec::{Literal, PrimitiveLiteral, PrimitiveType, Type}; + // 2024-01-01 = 19723 days since epoch + let lit = Literal::Primitive(PrimitiveLiteral::Int(19723)); + let ty = Type::Primitive(PrimitiveType::Date); + assert_eq!(literal_to_string_typed(&lit, Some(&ty)), "2024-01-01"); + // Untyped fallback keeps the raw epoch-day rendering + assert_eq!(literal_to_string_typed(&lit, None), "19723"); + } + + #[test] + fn test_literal_to_string_typed_timestamp() { + use iceberg::spec::{Literal, PrimitiveLiteral, PrimitiveType, Type}; + // 2024-01-01T00:00:00 UTC in micros + let lit = Literal::Primitive(PrimitiveLiteral::Long(1_704_067_200_000_000)); + let ty = Type::Primitive(PrimitiveType::Timestamp); + assert_eq!( + literal_to_string_typed(&lit, Some(&ty)), + "2024-01-01 00:00:00" + ); + } + + #[test] + fn test_literal_to_string_typed_decimal_scale() { + use iceberg::spec::{Literal, PrimitiveLiteral, PrimitiveType, Type}; + let ty = Type::Primitive(PrimitiveType::Decimal { precision: 10, scale: 2 }); + let lit = Literal::Primitive(PrimitiveLiteral::Int128(12345)); + assert_eq!(literal_to_string_typed(&lit, Some(&ty)), "123.45"); + let neg = Literal::Primitive(PrimitiveLiteral::Int128(-7)); + assert_eq!(literal_to_string_typed(&neg, Some(&ty)), "-0.07"); + } + + #[test] + fn test_literal_to_string_typed_string_passthrough() { + use iceberg::spec::{Literal, PrimitiveLiteral, PrimitiveType, Type}; + let lit = Literal::Primitive(PrimitiveLiteral::String("us-east-1".to_string())); + let ty = Type::Primitive(PrimitiveType::String); + assert_eq!(literal_to_string_typed(&lit, Some(&ty)), "us-east-1"); + } + + #[test] + fn test_format_decimal() { + assert_eq!(format_decimal(0, 2), "0.00"); + assert_eq!(format_decimal(5, 0), "5"); + assert_eq!(format_decimal(-12345, 3), "-12.345"); + assert_eq!(format_decimal(100, 2), "1.00"); + } + #[test] fn test_iceberg_file_entry_construction() { let entry = IcebergFileEntry { @@ -464,10 +615,12 @@ mod tests { file_size_bytes: 50000, partition_values: HashMap::new(), content_type: "data".to_string(), + sequence_number: 7, snapshot_id: 12345, }; assert_eq!(entry.path, "s3://bucket/data/part-00000.parquet"); assert_eq!(entry.record_count, 1000); + assert_eq!(entry.sequence_number, 7); } #[test] diff --git a/native/src/iceberg_reader/serialization.rs b/native/src/iceberg_reader/serialization.rs index 0ba615d3..6a7411eb 100644 --- a/native/src/iceberg_reader/serialization.rs +++ b/native/src/iceberg_reader/serialization.rs @@ -5,6 +5,7 @@ // BatchDocumentReader.parseToMaps(). use super::scan::{IcebergFileEntry, IcebergSchemaField, IcebergSnapshot}; +use crate::common::{write_field_header, write_string}; /// Magic number for batch protocol validation ("TANT") const MAGIC_NUMBER: u32 = 0x54414E54; @@ -17,15 +18,24 @@ const FIELD_TYPE_JSON: u8 = 6; /// Serialize a list of IcebergFileEntry into the TANT byte buffer format. /// -/// Each entry becomes one document with either 7 fields (full) or 5 fields (compact): -/// Full: path, file_format, record_count, file_size_bytes, partition_values, content_type, snapshot_id -/// Compact: path, file_format, record_count, file_size_bytes, snapshot_id +/// Each entry becomes one document with either 9 fields (full) or 7 fields (compact): +/// Full: path, file_format, record_count, file_size_bytes, partition_values, +/// content_type, sequence_number, snapshot_id, resolved_snapshot_id +/// Compact: path, file_format, record_count, file_size_bytes, content_type, +/// snapshot_id, resolved_snapshot_id +/// +/// content_type is correctness-critical (position/equality-delete files must be +/// distinguishable from data files) so it is kept in both modes. +/// `actual_snapshot_id` is the snapshot that was actually read (as opposed to +/// the per-entry snapshot_id, which is the snapshot that *added* each file); +/// it is embedded as `resolved_snapshot_id` so "list latest" callers can learn +/// which snapshot they saw. Pass -1 when unknown. pub fn serialize_iceberg_entries( entries: &[IcebergFileEntry], - _actual_snapshot_id: i64, + actual_snapshot_id: i64, compact: bool, ) -> Vec { - let per_entry = if compact { 200 } else { 350 }; + let per_entry = if compact { 250 } else { 400 }; let estimated = 4 + entries.len() * per_entry + entries.len() * 4 + 12; let mut buf = Vec::with_capacity(estimated); @@ -35,7 +45,7 @@ pub fn serialize_iceberg_entries( let mut offsets = Vec::with_capacity(entries.len()); for entry in entries { offsets.push(buf.len() as u32); - serialize_file_entry(&mut buf, entry, compact); + serialize_file_entry(&mut buf, entry, actual_snapshot_id, compact); } // Offset table @@ -52,8 +62,13 @@ pub fn serialize_iceberg_entries( buf } -fn serialize_file_entry(buf: &mut Vec, entry: &IcebergFileEntry, compact: bool) { - let field_count: u16 = if compact { 5 } else { 7 }; +fn serialize_file_entry( + buf: &mut Vec, + entry: &IcebergFileEntry, + actual_snapshot_id: i64, + compact: bool, +) { + let field_count: u16 = if compact { 7 } else { 9 }; buf.extend_from_slice(&field_count.to_ne_bytes()); // 1. path (TEXT) @@ -73,20 +88,30 @@ fn serialize_file_entry(buf: &mut Vec, entry: &IcebergFileEntry, compact: bo buf.extend_from_slice(&entry.file_size_bytes.to_ne_bytes()); if !compact { - // 5. partition_values (JSON) + // partition_values (JSON) write_field_header(buf, "partition_values", FIELD_TYPE_JSON, 1); let pv_json = serde_json::to_string(&entry.partition_values) .unwrap_or_else(|_| "{}".to_string()); write_string(buf, &pv_json); - // 6. content_type (TEXT) - write_field_header(buf, "content_type", FIELD_TYPE_TEXT, 1); - write_string(buf, &entry.content_type); + // sequence_number (INTEGER, -1 if unknown) — needed to decide which + // data files a position/equality delete applies to + write_field_header(buf, "sequence_number", FIELD_TYPE_INTEGER, 1); + buf.extend_from_slice(&entry.sequence_number.to_ne_bytes()); } - // 7/5. snapshot_id (INTEGER) + // content_type (TEXT) — always included: without it delete files are + // indistinguishable from data files and would be read as data + write_field_header(buf, "content_type", FIELD_TYPE_TEXT, 1); + write_string(buf, &entry.content_type); + + // snapshot_id (INTEGER) — the snapshot that ADDED this file write_field_header(buf, "snapshot_id", FIELD_TYPE_INTEGER, 1); buf.extend_from_slice(&entry.snapshot_id.to_ne_bytes()); + + // resolved_snapshot_id (INTEGER) — the snapshot that was actually read + write_field_header(buf, "resolved_snapshot_id", FIELD_TYPE_INTEGER, 1); + buf.extend_from_slice(&actual_snapshot_id.to_ne_bytes()); } /// Serialize a list of IcebergSchemaField plus the raw schema JSON into TANT format. @@ -226,20 +251,6 @@ fn serialize_snapshot(buf: &mut Vec, snapshot: &IcebergSnapshot) { write_string(buf, &summary_json); } -fn write_field_header(buf: &mut Vec, name: &str, field_type: u8, value_count: u16) { - let name_bytes = name.as_bytes(); - buf.extend_from_slice(&(name_bytes.len() as u16).to_ne_bytes()); - buf.extend_from_slice(name_bytes); - buf.push(field_type); - buf.extend_from_slice(&value_count.to_ne_bytes()); -} - -fn write_string(buf: &mut Vec, s: &str) { - let bytes = s.as_bytes(); - buf.extend_from_slice(&(bytes.len() as u32).to_ne_bytes()); - buf.extend_from_slice(bytes); -} - /// Serialize an IcebergSnapshotInfo into the TANT byte buffer format. /// /// Document 0: header with snapshot_id, schema_json, partition_spec_json, manifest_count @@ -336,6 +347,7 @@ mod tests { file_size_bytes: 50000, partition_values: HashMap::new(), content_type: "data".to_string(), + sequence_number: 3, snapshot_id: 12345, }; let buf = serialize_iceberg_entries(&[entry], 12345, false); @@ -363,6 +375,7 @@ mod tests { file_size_bytes: 50000, partition_values: pv, content_type: "data".to_string(), + sequence_number: 3, snapshot_id: 12345, }; @@ -378,7 +391,11 @@ mod tests { let compact_str = String::from_utf8_lossy(&compact_buf); assert!(!compact_str.contains("partition_values")); - assert!(!compact_str.contains("content_type")); + assert!(!compact_str.contains("sequence_number")); + // content_type is correctness-critical (delete files vs data files) + // and must survive compact mode + assert!(compact_str.contains("content_type")); + assert!(compact_str.contains("resolved_snapshot_id")); assert!(compact_str.contains("path")); } diff --git a/native/src/parquet_reader/distributed.rs b/native/src/parquet_reader/distributed.rs index 520b4666..81a8adc0 100644 --- a/native/src/parquet_reader/distributed.rs +++ b/native/src/parquet_reader/distributed.rs @@ -12,6 +12,7 @@ use object_store::path::Path as ObjectPath; use object_store::ObjectStore; use url::Url; +use crate::common::percent_decode; use crate::debug_println; use crate::delta_reader::engine::{DeltaStorageConfig, create_object_store}; use crate::parquet_schema_reader::arrow_schema_to_json; @@ -122,9 +123,10 @@ async fn get_table_info_async( if last_segment.contains('=') { partition_directories.push(dir_name.to_string()); - // Extract partition column name from first directory + // Extract the first-level partition column name. Deeper levels + // are discovered below by walking down one directory chain, + // since list_with_delimiter only returns single-level prefixes. if !partition_columns_found { - // Walk the path to find all partition levels for segment in dir_name.split('/') { if let Some(eq_pos) = segment.find('=') { let key = &segment[..eq_pos]; @@ -139,11 +141,57 @@ async fn get_table_info_async( } } + // Discover deeper partition levels by walking down the first partition + // directory chain (one LIST per level). list_with_delimiter above only + // exposes the first level, so a year/month/day table would otherwise + // report partition_columns = ["year"]. + if let Some(first_dir) = partition_directories.first().cloned() { + let mut current = first_dir; + loop { + let sub = store + .list_with_delimiter(Some(&ObjectPath::from(current.as_str()))) + .await + .map_err(|e| { + anyhow::anyhow!( + "Failed to list partition directory '{}': {}", + current, + e + ) + })?; + + let mut next_dir = None; + for cp in &sub.common_prefixes { + let name = cp.as_ref(); + if let Some(last_segment) = name.rsplit('/').find(|s| !s.is_empty()) { + if let Some(eq_pos) = last_segment.find('=') { + let key = &last_segment[..eq_pos]; + if !partition_columns.contains(&key.to_string()) { + partition_columns.push(key.to_string()); + } + if next_dir.is_none() { + next_dir = Some(name.to_string()); + } + } + } + } + + match next_dir { + Some(d) => current = d, + None => break, + } + } + } + // Collect root .parquet files let mut root_parquet_files = Vec::new(); for obj in &list_result.objects { let path_str = obj.location.as_ref(); if path_str.ends_with(".parquet") || path_str.ends_with(".parq") { + // Skip hidden/metadata files (same rule as list_partition_files) + let filename = path_str.rsplit('/').next().unwrap_or(path_str); + if filename.starts_with('.') || filename.starts_with('_') { + continue; + } root_parquet_files.push(ParquetFileEntry { path: path_str.to_string(), size: obj.size as i64, @@ -255,6 +303,11 @@ async fn read_schema_from_first_file( for obj in &list_result.objects { let path_str = obj.location.as_ref(); if path_str.ends_with(".parquet") || path_str.ends_with(".parq") { + // Skip hidden/metadata files (same rule as list_partition_files) + let filename = path_str.rsplit('/').next().unwrap_or(path_str); + if filename.starts_with('.') || filename.starts_with('_') { + continue; + } let reader = ParquetObjectReader::new(Arc::clone(store), obj.location.clone()) .with_file_size(obj.size as u64); let builder = @@ -328,43 +381,6 @@ pub(crate) fn parse_partition_values_from_path(path: &str) -> HashMap String { - let bytes = s.as_bytes(); - let mut result = Vec::with_capacity(bytes.len()); - let mut i = 0; - - while i < bytes.len() { - if bytes[i] == b'%' && i + 2 < bytes.len() { - // Try to decode the two hex digits after % - let hi = bytes[i + 1]; - let lo = bytes[i + 2]; - if let (Some(h), Some(l)) = (hex_val(hi), hex_val(lo)) { - result.push(h << 4 | l); - i += 3; - continue; - } - } - result.push(bytes[i]); - i += 1; - } - - String::from_utf8(result).unwrap_or_else(|_| s.to_string()) -} - -/// Convert an ASCII hex digit to its numeric value. -fn hex_val(b: u8) -> Option { - match b { - b'0'..=b'9' => Some(b - b'0'), - b'a'..=b'f' => Some(b - b'a' + 10), - b'A'..=b'F' => Some(b - b'A' + 10), - _ => None, - } -} - /// Normalize a table URL: ensure trailing slash and parse. fn normalize_table_url(url_str: &str) -> Result { let mut s = url_str.to_string(); @@ -387,22 +403,23 @@ fn normalize_table_url(url_str: &str) -> Result { } /// Convert a URL to an ObjectPath for object_store operations. +/// +/// `Url::path()` is percent-encoded and `ObjectPath::from` does not decode, +/// so the path must be decoded first or keys containing spaces/unicode +/// would resolve to the wrong object. fn url_to_object_path(url: &Url) -> ObjectPath { + let decoded = percent_decode(url.path()); match url.scheme() { "s3" | "s3a" => { // For S3, the path starts after the bucket name - let path = url.path(); - let trimmed = path.trim_start_matches('/'); - ObjectPath::from(trimmed) + ObjectPath::from(decoded.trim_start_matches('/')) } "az" | "azure" | "abfs" | "abfss" => { - let path = url.path(); - let trimmed = path.trim_start_matches('/'); - ObjectPath::from(trimmed) + ObjectPath::from(decoded.trim_start_matches('/')) } _ => { // For file:// and others, use the full path - ObjectPath::from(url.path()) + ObjectPath::from(decoded.as_str()) } } } diff --git a/native/src/parquet_reader/jni.rs b/native/src/parquet_reader/jni.rs index b2234bd1..a06f87b9 100644 --- a/native/src/parquet_reader/jni.rs +++ b/native/src/parquet_reader/jni.rs @@ -58,11 +58,15 @@ pub extern "system" fn Java_io_indextables_tantivy4java_parquet_ParquetTableRead match get_parquet_table_info(&url_str, &config) { Ok(mut info) => { // Prune partition directories by parsing partition values from path - // and testing against predicate — biggest win for Hive tables + // and testing against predicate — biggest win for Hive tables. + // Directories are FIRST-LEVEL only (list_with_delimiter), so deeper + // partition columns are not bound yet: use partial evaluation and + // keep directories whose result is indeterminate at this level. + // Executors re-apply the full predicate once all levels are known. if let Some(ref pred) = predicate { info.partition_directories.retain(|dir| { let values = parse_partition_values_from_path(dir); - pred.evaluate(&values) + pred.evaluate_partial(&values).unwrap_or(true) }); info.root_parquet_files = filter_by_predicate( info.root_parquet_files, &predicate, |e| &e.partition_values, diff --git a/native/src/parquet_reader/serialization.rs b/native/src/parquet_reader/serialization.rs index 6e643b77..2e1b8577 100644 --- a/native/src/parquet_reader/serialization.rs +++ b/native/src/parquet_reader/serialization.rs @@ -4,6 +4,7 @@ // for efficient Rust→Java transfer via BatchDocumentReader.parseToMaps(). use super::distributed::{ParquetTableInfo, ParquetFileEntry}; +use crate::common::{write_field_header, write_string}; /// Magic number for batch protocol validation ("TANT") const MAGIC_NUMBER: u32 = 0x54414E54; @@ -143,20 +144,6 @@ fn serialize_file_entry(buf: &mut Vec, entry: &ParquetFileEntry) { // ─── TANT protocol helpers ────────────────────────────────────────────────── -fn write_field_header(buf: &mut Vec, name: &str, field_type: u8, value_count: u16) { - let name_bytes = name.as_bytes(); - buf.extend_from_slice(&(name_bytes.len() as u16).to_ne_bytes()); - buf.extend_from_slice(name_bytes); - buf.push(field_type); - buf.extend_from_slice(&value_count.to_ne_bytes()); -} - -fn write_string(buf: &mut Vec, s: &str) { - let bytes = s.as_bytes(); - buf.extend_from_slice(&(bytes.len() as u32).to_ne_bytes()); - buf.extend_from_slice(bytes); -} - #[cfg(test)] mod tests { use super::*; diff --git a/native/src/parquet_schema_reader.rs b/native/src/parquet_schema_reader.rs index 4e4f0088..df701750 100644 --- a/native/src/parquet_schema_reader.rs +++ b/native/src/parquet_schema_reader.rs @@ -313,8 +313,9 @@ fn parse_file_url(url_str: &str) -> Result<(Url, ObjectPath)> { if url_str.starts_with("s3://") || url_str.starts_with("s3a://") { let url = Url::parse(url_str) .map_err(|e| anyhow::anyhow!("Invalid S3 URL '{}': {}", url_str, e))?; - let path = url.path().trim_start_matches('/'); - let object_path = ObjectPath::from(path); + // Url::path() is percent-encoded; ObjectPath does not decode + let path = crate::common::percent_decode(url.path()); + let object_path = ObjectPath::from(path.trim_start_matches('/')); // Base URL is just scheme://bucket/ let base = Url::parse(&format!("{}://{}/", url.scheme(), url.host_str().unwrap_or(""))) .map_err(|e| anyhow::anyhow!("Failed to construct base URL: {}", e))?; @@ -326,10 +327,23 @@ fn parse_file_url(url_str: &str) -> Result<(Url, ObjectPath)> { { let url = Url::parse(url_str) .map_err(|e| anyhow::anyhow!("Invalid Azure URL '{}': {}", url_str, e))?; - let path = url.path().trim_start_matches('/'); - let object_path = ObjectPath::from(path); - let base = Url::parse(&format!("{}://{}/", url.scheme(), url.host_str().unwrap_or(""))) - .map_err(|e| anyhow::anyhow!("Failed to construct base URL: {}", e))?; + // Url::path() is percent-encoded; ObjectPath does not decode + let path = crate::common::percent_decode(url.path()); + let object_path = ObjectPath::from(path.trim_start_matches('/')); + // Preserve the userinfo component: for abfss://container@account.../ + // the container name lives in the username, not the host + let userinfo = if url.username().is_empty() { + String::new() + } else { + format!("{}@", url.username()) + }; + let base = Url::parse(&format!( + "{}://{}{}/", + url.scheme(), + userinfo, + url.host_str().unwrap_or("") + )) + .map_err(|e| anyhow::anyhow!("Failed to construct base URL: {}", e))?; Ok((base, object_path)) } else if url_str.starts_with("file://") { let url = Url::parse(url_str) diff --git a/src/main/java/io/indextables/tantivy4java/delta/DeltaFileEntry.java b/src/main/java/io/indextables/tantivy4java/delta/DeltaFileEntry.java index 6f0cf233..bfd4a76c 100644 --- a/src/main/java/io/indextables/tantivy4java/delta/DeltaFileEntry.java +++ b/src/main/java/io/indextables/tantivy4java/delta/DeltaFileEntry.java @@ -44,7 +44,7 @@ public DeltaFileEntry(String path, long size, long modificationTime, long numRec /** @return parquet file path relative to the table root */ public String getPath() { return path; } - /** @return file size in bytes */ + /** @return file size in bytes, or -1 if unknown */ public long getSize() { return size; } /** @return epoch milliseconds when the file was created */ @@ -62,7 +62,11 @@ public DeltaFileEntry(String path, long size, long modificationTime, long numRec /** @return true if this file has an associated deletion vector */ public boolean hasDeletionVector() { return hasDeletionVector; } - /** @return the Delta table snapshot version this file listing was read from */ + /** + * @return the Delta table snapshot version this file listing was read from, + * or -1 if unknown (entries from log-change reads, where the + * per-entry commit version is not tracked) + */ public long getTableVersion() { return tableVersion; } @Override diff --git a/src/main/java/io/indextables/tantivy4java/delta/DeltaTableReader.java b/src/main/java/io/indextables/tantivy4java/delta/DeltaTableReader.java index 7807834d..21d4aa17 100644 --- a/src/main/java/io/indextables/tantivy4java/delta/DeltaTableReader.java +++ b/src/main/java/io/indextables/tantivy4java/delta/DeltaTableReader.java @@ -39,7 +39,7 @@ * // Specific version * List files = DeltaTableReader.listFiles("s3://bucket/delta_table", config, 42); * - * // Compact mode — skip partition_values and has_deletion_vector for lightweight listing + * // Compact mode — skip partition_values for lightweight listing * List compact = DeltaTableReader.listFiles("s3://bucket/delta_table", config, true); * } */ @@ -97,7 +97,7 @@ public static List listFiles(String tableUrl, Map listFiles(String tableUrl, boolean compact) { * * @param tableUrl table location * @param config credential and storage configuration - * @param compact if true, skip partition_values and has_deletion_vector fields + * @param compact if true, skip the partition_values field (has_deletion_vector is always included) * @return list of active file entries * @throws RuntimeException if the table cannot be read */ @@ -124,7 +124,7 @@ public static List listFiles(String tableUrl, Map listFiles(String tableUrl, Map partitionValues; private final String contentType; + private final long sequenceNumber; private final long snapshotId; + private final long resolvedSnapshotId; public IcebergFileEntry(String path, String fileFormat, long recordCount, long fileSizeBytes, Map partitionValues, String contentType, long snapshotId) { + this(path, fileFormat, recordCount, fileSizeBytes, partitionValues, contentType, -1, snapshotId, -1); + } + + public IcebergFileEntry(String path, String fileFormat, long recordCount, long fileSizeBytes, + Map partitionValues, String contentType, + long sequenceNumber, long snapshotId, long resolvedSnapshotId) { this.path = path; this.fileFormat = fileFormat; this.recordCount = recordCount; @@ -40,7 +48,9 @@ public IcebergFileEntry(String path, String fileFormat, long recordCount, long f ? Collections.unmodifiableMap(partitionValues) : Collections.emptyMap(); this.contentType = contentType; + this.sequenceNumber = sequenceNumber; this.snapshotId = snapshotId; + this.resolvedSnapshotId = resolvedSnapshotId; } /** @@ -85,6 +95,14 @@ public String getContentType() { return contentType; } + /** + * @return data sequence number, or -1 if unknown. Needed to decide which + * data files a position/equality delete file applies to. + */ + public long getSequenceNumber() { + return sequenceNumber; + } + /** * @return snapshot ID that added this file */ @@ -92,6 +110,15 @@ public long getSnapshotId() { return snapshotId; } + /** + * @return the snapshot ID that was actually read to produce this listing + * (useful when listing "latest" to learn which snapshot was + * resolved), or -1 if unknown + */ + public long getResolvedSnapshotId() { + return resolvedSnapshotId; + } + @Override public String toString() { return String.format("IcebergFileEntry{path='%s', format='%s', records=%d, size=%d, snapshot=%d}", @@ -111,12 +138,14 @@ static IcebergFileEntry fromMap(Map map) { long recordCount = toLong(map.get("record_count")); long fileSizeBytes = toLong(map.get("file_size_bytes")); String contentType = (String) map.getOrDefault("content_type", "data"); + long sequenceNumber = toLong(map.get("sequence_number")); long snapshotId = toLong(map.get("snapshot_id")); + long resolvedSnapshotId = toLong(map.get("resolved_snapshot_id")); Map partitionValues = parsePartitionValues(map.get("partition_values")); return new IcebergFileEntry(path, fileFormat, recordCount, fileSizeBytes, - partitionValues, contentType, snapshotId); + partitionValues, contentType, sequenceNumber, snapshotId, resolvedSnapshotId); } static long toLong(Object value) { diff --git a/src/main/java/io/indextables/tantivy4java/iceberg/IcebergTableReader.java b/src/main/java/io/indextables/tantivy4java/iceberg/IcebergTableReader.java index 3c670635..ef94a890 100644 --- a/src/main/java/io/indextables/tantivy4java/iceberg/IcebergTableReader.java +++ b/src/main/java/io/indextables/tantivy4java/iceberg/IcebergTableReader.java @@ -140,7 +140,7 @@ public static List listFiles( * @param namespace Iceberg namespace * @param tableName table name * @param config catalog and storage configuration - * @param compact if true, skip partition_values and content_type + * @param compact if true, skip partition_values and sequence_number (content_type is always included) * @return list of active data file entries */ public static List listFiles( @@ -157,7 +157,7 @@ public static List listFiles( * @param tableName table name * @param config catalog and storage configuration * @param snapshotId snapshot ID (-1 for current) - * @param compact if true, skip partition_values and content_type + * @param compact if true, skip partition_values and sequence_number (content_type is always included) * @return list of active data file entries */ public static List listFiles( @@ -174,7 +174,7 @@ public static List listFiles( * @param tableName table name * @param config catalog and storage configuration * @param snapshotId snapshot ID (-1 for current) - * @param compact if true, skip partition_values and content_type + * @param compact if true, skip partition_values and sequence_number (content_type is always included) * @param filter partition filter (null for no filtering) * @return list of matching data file entries */ @@ -426,7 +426,7 @@ public static List readManifestFile( * @param tableName table name * @param config catalog and storage configuration * @param manifestPath full path to the manifest avro file - * @param compact if true, skip partition_values and content_type + * @param compact if true, skip partition_values and sequence_number (content_type is always included) * @return list of file entries from this manifest */ public static List readManifestFile( @@ -443,7 +443,7 @@ public static List readManifestFile( * @param tableName table name * @param config catalog and storage configuration * @param manifestPath full path to the manifest avro file - * @param compact if true, skip partition_values and content_type + * @param compact if true, skip partition_values and sequence_number (content_type is always included) * @param filter partition filter (null for no filtering) * @return list of matching file entries from this manifest */ @@ -482,10 +482,10 @@ public static List readManifestFile( /** * Read an Iceberg manifest file and export entries via Arrow FFI. * - *

Builds a flat RecordBatch with 7 columns: + *

Builds a flat RecordBatch with 8 columns: * path (Utf8), file_format (Utf8), record_count (Int64), * file_size_bytes (Int64), partition_values (Utf8/JSON), - * content_type (Utf8), snapshot_id (Int64). + * content_type (Utf8), snapshot_id (Int64), sequence_number (Int64). * *

The caller must pre-allocate FFI_ArrowArray and FFI_ArrowSchema structs * for each column and pass their memory addresses. @@ -496,8 +496,8 @@ public static List readManifestFile( * @param config catalog and storage configuration * @param manifestPath full path to the manifest avro file * @param filter partition filter (null for no filtering) - * @param arrayAddrs pre-allocated FFI_ArrowArray addresses (7 columns) - * @param schemaAddrs pre-allocated FFI_ArrowSchema addresses (7 columns) + * @param arrayAddrs pre-allocated FFI_ArrowArray addresses (8 columns) + * @param schemaAddrs pre-allocated FFI_ArrowSchema addresses (8 columns) * @return number of rows written, or -1 on error */ public static int readManifestFileArrowFfi( @@ -508,11 +508,11 @@ public static int readManifestFileArrowFfi( if (manifestPath == null || manifestPath.isEmpty()) { throw new IllegalArgumentException("manifestPath must not be null or empty"); } - if (arrayAddrs == null || arrayAddrs.length < 7) { - throw new IllegalArgumentException("arrayAddrs must have at least 7 elements"); + if (arrayAddrs == null || arrayAddrs.length < 8) { + throw new IllegalArgumentException("arrayAddrs must have at least 8 elements"); } - if (schemaAddrs == null || schemaAddrs.length < 7) { - throw new IllegalArgumentException("schemaAddrs must have at least 7 elements"); + if (schemaAddrs == null || schemaAddrs.length < 8) { + throw new IllegalArgumentException("schemaAddrs must have at least 8 elements"); } String predicateJson = filter != null ? filter.toJson() : null; From 4c8013ee398492cb1b864ba8c587fa9c8972a3ad Mon Sep 17 00:00:00 2001 From: Scott Schenkein Date: Sat, 4 Jul 2026 20:57:19 -0400 Subject: [PATCH 2/2] readers: fix bugs found in PR #182 review Addresses 6 findings from an independent code review of PR #182 (fix all findings from the table-format reader review): - delta_reader: read_metadata_from_checkpoint no longer assumes metaData precedes protocol in the projected Arrow batch; ProjectionMask::roots() returns columns in physical schema order, which the Delta spec does not guarantee, so column identity is now derived from each field's sorted projection index instead. - delta_reader: the stale-_last_checkpoint guard (H5) now detects a second, deeper log gap beyond the newly-adopted checkpoint instead of silently discarding the already-known max_commit_version and returning a still-stale snapshot. - delta_reader: a missing protocol row in a checkpoint is now a hard error (like metaData) instead of a debug-only skip, since H4's reader-feature validation must not be silently bypassed. Updated the checkpoint test helper to include a protocol column so the enforcement path is covered. - parquet_reader: multi-level partition-column discovery now walks every first-level partition directory instead of only the first, so sibling partitions with a deeper structure (schema drift, partial backfill) are no longer under-reported. - iceberg_reader: threaded an inherited_snapshot_id parameter through read_iceberg_manifest / read_manifest_with_file_io / the Arrow FFI path and IcebergTableReader's Java API (additive overloads) so getChangesSince passes ManifestFileInfo's known added_snapshot_id, resolving the same fallback as listFiles() instead of always defaulting to -1. - parquet_reader: deduplicated the 4x-copied hidden-file filename check into a single is_hidden_parquet_file() helper. Co-Authored-By: Claude Sonnet 5 --- native/src/delta_reader/distributed.rs | 132 +++++++++++++++--- native/src/iceberg_reader/distributed.rs | 25 +++- native/src/iceberg_reader/jni.rs | 21 ++- native/src/parquet_reader/distributed.rs | 33 +++-- .../iceberg/IcebergTableReader.java | 64 ++++++++- 5 files changed, 232 insertions(+), 43 deletions(-) diff --git a/native/src/delta_reader/distributed.rs b/native/src/delta_reader/distributed.rs index 203f5132..c5ce3de6 100644 --- a/native/src/delta_reader/distributed.rs +++ b/native/src/delta_reader/distributed.rs @@ -128,6 +128,24 @@ pub fn get_snapshot_info( checkpoint_part_paths = newer_cp_files; commit_file_paths = list_commit_files_after(&store, &log_prefix, checkpoint_version).await?; + + // The same staleness scan may also have observed a commit JSON + // version beyond what the sequential probe from the newly-adopted + // checkpoint reached — i.e. a second, deeper gap. Don't silently + // drop that information; fail loudly instead of returning a + // snapshot that is still stale. + let reached_version = checkpoint_version + commit_file_paths.len() as u64; + if let Some(max_commit) = markers.max_commit_version { + if max_commit > reached_version { + anyhow::bail!( + "Delta log is not contiguous: commit version {} exists but versions after {} \ + are missing, and the newest checkpoint found ({}) does not cover the gap", + max_commit, + reached_version, + newer_cp_version + ); + } + } } else if let Some(max_commit) = markers.max_commit_version { // Commits exist beyond a gap with no covering checkpoint — the // contiguous chain is broken (cleanup/corruption); fail loudly @@ -176,14 +194,23 @@ pub fn get_snapshot_info( // not implement (v2 checkpoints, unknown future features, id-mode // column mapping) instead of failing confusingly or returning wrong // results later. - if let Some(ref proto) = protocol { - validate_protocol(proto)?; - } else { - debug_println!( - "🔧 DELTA_DIST: WARNING: no protocol row found in checkpoint {} — skipping reader feature validation", - checkpoint_version - ); - } + // + // A checkpoint is a full state snapshot, so — like `metaData` — the + // `protocol` action must be present in every checkpoint (it is a + // Delta protocol requirement, not optional). Treat a missing row as + // a hard error rather than silently skipping reader-feature + // validation: doing so would defeat H4 for exactly the malformed or + // unsupported checkpoints it exists to catch, with no error and no + // visible warning in production (debug_println! is a no-op unless + // TANTIVY4JAVA_DEBUG=1 is set). + let protocol = protocol.ok_or_else(|| { + anyhow::anyhow!( + "No protocol row found in any part of checkpoint version {} for table {}", + checkpoint_version, + url_str + ) + })?; + validate_protocol(&protocol)?; let column_mapping_mode = metadata .configuration .get("delta.columnMapping.mode") @@ -361,6 +388,21 @@ pub fn get_current_version( current_version, newer_cp_version ); current_version = newer_cp_version + newer_commits.len() as u64; + + // Guard against a second, deeper gap: the same staleness scan may + // have observed a commit JSON version beyond what the sequential + // probe from the newly-adopted checkpoint reached. + if let Some(max_commit) = markers.max_commit_version { + if max_commit > current_version { + anyhow::bail!( + "Delta log is not contiguous: commit version {} exists but versions after {} \ + are missing, and the newest checkpoint found ({}) does not cover the gap", + max_commit, + current_version, + newer_cp_version + ); + } + } } else if let Some(max_commit) = markers.max_commit_version { anyhow::bail!( "Delta log is not contiguous: commit version {} exists but versions after {} \ @@ -694,10 +736,16 @@ async fn read_metadata_from_checkpoint( .iter() .position(|f| f.name() == "protocol"); - let projected: Vec = [metadata_idx, protocol_idx].iter().flatten().copied().collect(); + let mut projected: Vec = [metadata_idx, protocol_idx].iter().flatten().copied().collect(); if projected.is_empty() { return Ok((None, None)); } + // ProjectionMask::roots() returns columns in ascending physical schema + // order regardless of the order indices are passed in, so the projected + // batch's column order must be derived from `projected` sorted ascending + // (NOT assumed to be metaData-then-protocol — the Delta spec does not + // guarantee action-column ordering in a checkpoint part's schema). + projected.sort_unstable(); let mask = parquet::arrow::ProjectionMask::roots( builder.parquet_schema(), @@ -712,15 +760,18 @@ async fn read_metadata_from_checkpoint( while let Some(batch_result) = stream.next().await { let batch = batch_result?; - // Projected batch columns keep schema order: metaData (if present) - // comes before protocol - let mut batch_col = 0; - let meta_col = metadata_idx.map(|_| { - let c = batch.column(batch_col).clone(); - batch_col += 1; - c + let meta_col = metadata_idx.and_then(|idx| { + projected + .iter() + .position(|&p| p == idx) + .map(|pos| batch.column(pos).clone()) + }); + let proto_col = protocol_idx.and_then(|idx| { + projected + .iter() + .position(|&p| p == idx) + .map(|pos| batch.column(pos).clone()) }); - let proto_col = protocol_idx.map(|_| batch.column(batch_col).clone()); if let (None, Some(col)) = (&found_meta, &meta_col) { let struct_array = col.as_struct(); @@ -1688,6 +1739,45 @@ mod tests { Some(metadata_nulls), ); + // protocol struct fields — present at row 0 alongside metaData, like a + // real checkpoint (both are singleton state rows read via the same + // column-projected scan in `read_metadata_from_checkpoint`). + let protocol_fields = Fields::from(vec![ + Field::new("minReaderVersion", DataType::Int64, true), + Field::new( + "readerFeatures", + DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))), + true, + ), + ]); + + let mut min_reader_version_builder = Int64Builder::new(); + let mut reader_features_builder = arrow_array::builder::ListBuilder::new(StringBuilder::new()); + + // Row 0: protocol row (minReaderVersion=1, no reader features) + min_reader_version_builder.append_value(1); + reader_features_builder.append(true); + + for _ in 0..adds.len() { + min_reader_version_builder.append_null(); + reader_features_builder.append(false); + } + + let protocol_nulls = { + let mut bools = vec![false; num_rows]; + bools[0] = true; // only row 0 (protocol) is non-null + arrow::buffer::NullBuffer::from(bools.as_slice()) + }; + + let protocol_struct = StructArray::new( + protocol_fields, + vec![ + Arc::new(min_reader_version_builder.finish()), + Arc::new(reader_features_builder.finish()), + ], + Some(protocol_nulls), + ); + // add struct fields let add_fields = Fields::from(vec![ Field::new("path", DataType::Utf8, true), @@ -1758,13 +1848,18 @@ mod tests { Some(add_nulls), ); - // Create the RecordBatch with both columns + // Create the RecordBatch with all columns let schema = Arc::new(Schema::new(vec![ Field::new( "metaData", DataType::Struct(metadata_struct.fields().clone()), true, ), + Field::new( + "protocol", + DataType::Struct(protocol_struct.fields().clone()), + true, + ), Field::new( "add", DataType::Struct(add_struct.fields().clone()), @@ -1776,6 +1871,7 @@ mod tests { schema.clone(), vec![ Arc::new(metadata_struct), + Arc::new(protocol_struct), Arc::new(add_struct), ], ) diff --git a/native/src/iceberg_reader/distributed.rs b/native/src/iceberg_reader/distributed.rs index ad538355..4c7edd0e 100644 --- a/native/src/iceberg_reader/distributed.rs +++ b/native/src/iceberg_reader/distributed.rs @@ -129,6 +129,7 @@ pub fn get_current_iceberg_snapshot_id( pub fn read_iceberg_manifest( config: &HashMap, manifest_path: &str, + inherited_snapshot_id: Option, ) -> Result> { debug_println!( "🔧 ICEBERG_DIST: read_manifest path={}", @@ -156,7 +157,7 @@ pub fn read_iceberg_manifest( .build() .map_err(|e| anyhow::anyhow!("Failed to build FileIO: {}", e))?; - read_manifest_with_file_io(&file_io, manifest_path).await + read_manifest_with_file_io(&file_io, manifest_path, inherited_snapshot_id).await }) } @@ -273,9 +274,18 @@ pub(crate) async fn get_snapshot_info_with_catalog( } /// Read one manifest file using a FileIO instance. +/// +/// `inherited_snapshot_id`, when provided, is used as the Iceberg-spec +/// "inherited snapshot id" fallback for manifest entries whose own +/// `snapshot_id` is absent — mirroring `list_files_with_catalog`'s use of +/// `manifest_file.added_snapshot_id`. Callers with manifest-list context +/// (e.g. `getChangesSince`, which already has `ManifestFileInfo::added_snapshot_id` +/// for each manifest it reads) should pass it through so this path resolves +/// to the same value as the catalog-based listing path instead of `-1`. async fn read_manifest_with_file_io( file_io: &FileIO, manifest_path: &str, + inherited_snapshot_id: Option, ) -> Result> { let manifest_input = file_io.new_input(manifest_path) .map_err(|e| anyhow::anyhow!("Failed to open manifest {}: {}", manifest_path, e))?; @@ -321,10 +331,12 @@ async fn read_manifest_with_file_io( partition_values, content_type: content_type_to_string(data_file.content_type()), sequence_number: entry.sequence_number().unwrap_or(-1), - // The manifest-list context (added_snapshot_id) is not available - // when reading a single manifest file; -1 = unknown (matches the - // sentinel convention elsewhere, instead of a real-looking 0) - snapshot_id: entry.snapshot_id().unwrap_or(-1), + // Fall back to the caller-supplied inherited snapshot id (the + // Iceberg-spec "inherited" resolution used by manifest-list-aware + // callers) before defaulting to -1 = unknown. This keeps this + // path consistent with `list_files_with_catalog`, which resolves + // the same fallback via `manifest_file.added_snapshot_id`. + snapshot_id: entry.snapshot_id().or(inherited_snapshot_id).unwrap_or(-1), }); } @@ -353,6 +365,7 @@ pub fn read_iceberg_manifest_arrow_ffi( predicate: Option<&crate::common::PartitionPredicate>, array_addrs: &[i64], schema_addrs: &[i64], + inherited_snapshot_id: Option, ) -> Result { use arrow::ffi::{FFI_ArrowArray, FFI_ArrowSchema}; use arrow_array::{StringArray, Int64Array, Array}; @@ -368,7 +381,7 @@ pub fn read_iceberg_manifest_arrow_ffi( } // 1. Read manifest → Vec - let entries = read_iceberg_manifest(config, manifest_path)?; + let entries = read_iceberg_manifest(config, manifest_path, inherited_snapshot_id)?; // 2. Apply partition predicate filter let entries: Vec = match predicate { diff --git a/native/src/iceberg_reader/jni.rs b/native/src/iceberg_reader/jni.rs index 2d12a022..650dbd58 100644 --- a/native/src/iceberg_reader/jni.rs +++ b/native/src/iceberg_reader/jni.rs @@ -300,6 +300,7 @@ pub extern "system" fn Java_io_indextables_tantivy4java_iceberg_IcebergTableRead config_map: JObject, compact: jboolean, predicate_json: JString, + inherited_snapshot_id: jlong, ) -> jbyteArray { debug_println!("🔧 ICEBERG_JNI: nativeReadManifestFile called"); @@ -336,7 +337,17 @@ pub extern "system" fn Java_io_indextables_tantivy4java_iceberg_IcebergTableRead } }; - match read_iceberg_manifest(&config, &manifest_str) { + // Negative sentinel = no manifest-list context available; snapshot_id + // falls back to -1 (unknown) per entry. Callers that already know the + // manifest's added_snapshot_id (e.g. getChangesSince) pass it here so + // this path resolves the same fallback as nativeListFiles. + let inherited_snapshot_id_opt = if inherited_snapshot_id < 0 { + None + } else { + Some(inherited_snapshot_id) + }; + + match read_iceberg_manifest(&config, &manifest_str, inherited_snapshot_id_opt) { Ok(entries) => { let entries = filter_by_predicate(entries, &predicate, |e| &e.partition_values); debug_println!( @@ -369,6 +380,7 @@ pub extern "system" fn Java_io_indextables_tantivy4java_iceberg_IcebergTableRead predicate_json: JString, array_addrs: jlongArray, schema_addrs: jlongArray, + inherited_snapshot_id: jlong, ) -> jint { debug_println!("🔧 ICEBERG_JNI: nativeReadManifestFileArrowFfi called"); @@ -419,12 +431,19 @@ pub extern "system" fn Java_io_indextables_tantivy4java_iceberg_IcebergTableRead } }; + let inherited_snapshot_id_opt = if inherited_snapshot_id < 0 { + None + } else { + Some(inherited_snapshot_id) + }; + match read_iceberg_manifest_arrow_ffi( &config, &manifest_str, predicate.as_ref(), &arr_addrs, &sch_addrs, + inherited_snapshot_id_opt, ) { Ok(num_rows) => { debug_println!("🔧 ICEBERG_JNI: Arrow FFI exported {} rows", num_rows); diff --git a/native/src/parquet_reader/distributed.rs b/native/src/parquet_reader/distributed.rs index 81a8adc0..cd0c7f0e 100644 --- a/native/src/parquet_reader/distributed.rs +++ b/native/src/parquet_reader/distributed.rs @@ -35,6 +35,14 @@ pub struct ParquetTableInfo { pub is_partitioned: bool, } +/// Whether a parquet object's filename is a hidden/metadata file that should +/// be excluded from listings (e.g. `.DS_Store`, `_SUCCESS`, `_delta_log`). +/// `path_str` is the full object path; only the final path segment is checked. +fn is_hidden_parquet_file(path_str: &str) -> bool { + let filename = path_str.rsplit('/').next().unwrap_or(path_str); + filename.starts_with('.') || filename.starts_with('_') +} + /// A single parquet file entry with metadata. #[derive(Debug, Clone)] pub struct ParquetFileEntry { @@ -141,11 +149,14 @@ async fn get_table_info_async( } } - // Discover deeper partition levels by walking down the first partition - // directory chain (one LIST per level). list_with_delimiter above only - // exposes the first level, so a year/month/day table would otherwise - // report partition_columns = ["year"]. - if let Some(first_dir) = partition_directories.first().cloned() { + // Discover deeper partition levels by walking down every first-level + // partition directory (one LIST per level per directory). Walking only a + // single chain would silently under-report partition columns whenever + // sibling partitions have a different depth than the first one (schema + // drift, partial backfill, incremental repartitioning) — the same class + // of bug as reporting partition_columns = ["year"] for a year/month/day + // table when list_with_delimiter above only exposes the first level. + for first_dir in partition_directories.clone() { let mut current = first_dir; loop { let sub = store @@ -188,8 +199,7 @@ async fn get_table_info_async( let path_str = obj.location.as_ref(); if path_str.ends_with(".parquet") || path_str.ends_with(".parq") { // Skip hidden/metadata files (same rule as list_partition_files) - let filename = path_str.rsplit('/').next().unwrap_or(path_str); - if filename.starts_with('.') || filename.starts_with('_') { + if is_hidden_parquet_file(path_str) { continue; } root_parquet_files.push(ParquetFileEntry { @@ -266,8 +276,7 @@ async fn list_partition_files_async( } // Skip hidden files and metadata - let filename = path_str.rsplit('/').next().unwrap_or(path_str); - if filename.starts_with('.') || filename.starts_with('_') { + if is_hidden_parquet_file(path_str) { continue; } @@ -304,8 +313,7 @@ async fn read_schema_from_first_file( let path_str = obj.location.as_ref(); if path_str.ends_with(".parquet") || path_str.ends_with(".parq") { // Skip hidden/metadata files (same rule as list_partition_files) - let filename = path_str.rsplit('/').next().unwrap_or(path_str); - if filename.starts_with('.') || filename.starts_with('_') { + if is_hidden_parquet_file(path_str) { continue; } let reader = ParquetObjectReader::new(Arc::clone(store), obj.location.clone()) @@ -332,8 +340,7 @@ async fn read_schema_from_first_file( for obj in objects { let path_str = obj.location.as_ref(); if path_str.ends_with(".parquet") || path_str.ends_with(".parq") { - let filename = path_str.rsplit('/').next().unwrap_or(path_str); - if filename.starts_with('.') || filename.starts_with('_') { + if is_hidden_parquet_file(path_str) { continue; } diff --git a/src/main/java/io/indextables/tantivy4java/iceberg/IcebergTableReader.java b/src/main/java/io/indextables/tantivy4java/iceberg/IcebergTableReader.java index ef94a890..165cffdf 100644 --- a/src/main/java/io/indextables/tantivy4java/iceberg/IcebergTableReader.java +++ b/src/main/java/io/indextables/tantivy4java/iceberg/IcebergTableReader.java @@ -451,6 +451,35 @@ public static List readManifestFile( String catalogName, String namespace, String tableName, Map config, String manifestPath, boolean compact, PartitionFilter filter) { + return readManifestFile(catalogName, namespace, tableName, config, manifestPath, + compact, filter, -1L); + } + + /** + * Read one manifest file with partition predicate filtering and an inherited + * snapshot id fallback. + * + *

Per the Iceberg spec, a manifest entry's {@code snapshot_id} may be absent + * ("inherited"), in which case it resolves to the snapshot that added the + * containing manifest. Callers that already know this (e.g. {@link #getChangesSince} + * iterating {@link IcebergSnapshotInfo.ManifestFileInfo}) should pass + * {@code inheritedSnapshotId} so the resolved value matches {@link #listFiles} + * for the same underlying data; otherwise absent entries resolve to {@code -1}. + * + * @param catalogName catalog identifier + * @param namespace Iceberg namespace + * @param tableName table name + * @param config catalog and storage configuration + * @param manifestPath full path to the manifest avro file + * @param compact if true, skip partition_values and sequence_number (content_type is always included) + * @param filter partition filter (null for no filtering) + * @param inheritedSnapshotId fallback snapshot id for entries with no own snapshot_id (-1 if unknown) + * @return list of matching file entries from this manifest + */ + public static List readManifestFile( + String catalogName, String namespace, String tableName, + Map config, String manifestPath, boolean compact, + PartitionFilter filter, long inheritedSnapshotId) { validateParams(catalogName, namespace, tableName, config); if (manifestPath == null || manifestPath.isEmpty()) { throw new IllegalArgumentException("manifestPath must not be null or empty"); @@ -458,7 +487,8 @@ public static List readManifestFile( String predicateJson = filter != null ? filter.toJson() : null; byte[] bytes = nativeReadManifestFile(catalogName, namespace, tableName, manifestPath, - config != null ? config : Collections.emptyMap(), compact, predicateJson); + config != null ? config : Collections.emptyMap(), compact, predicateJson, + inheritedSnapshotId); if (bytes == null) { throw new RuntimeException("Native readManifestFile returned null (check preceding exception)"); @@ -504,6 +534,25 @@ public static int readManifestFileArrowFfi( String catalogName, String namespace, String tableName, Map config, String manifestPath, PartitionFilter filter, long[] arrayAddrs, long[] schemaAddrs) { + return readManifestFileArrowFfi(catalogName, namespace, tableName, config, manifestPath, + filter, arrayAddrs, schemaAddrs, -1L); + } + + /** + * Read an Iceberg manifest file and export entries via Arrow FFI, with an + * inherited snapshot id fallback. + * + *

See {@link #readManifestFile(String, String, String, Map, String, boolean, PartitionFilter, long)} + * for the semantics of {@code inheritedSnapshotId}. + * + * @param inheritedSnapshotId fallback snapshot id for entries with no own snapshot_id (-1 if unknown) + * @return number of rows written, or -1 on error + */ + public static int readManifestFileArrowFfi( + String catalogName, String namespace, String tableName, + Map config, String manifestPath, + PartitionFilter filter, long[] arrayAddrs, long[] schemaAddrs, + long inheritedSnapshotId) { validateParams(catalogName, namespace, tableName, config); if (manifestPath == null || manifestPath.isEmpty()) { throw new IllegalArgumentException("manifestPath must not be null or empty"); @@ -518,7 +567,7 @@ public static int readManifestFileArrowFfi( String predicateJson = filter != null ? filter.toJson() : null; return nativeReadManifestFileArrowFfi(catalogName, namespace, tableName, manifestPath, config != null ? config : Collections.emptyMap(), - predicateJson, arrayAddrs, schemaAddrs); + predicateJson, arrayAddrs, schemaAddrs, inheritedSnapshotId); } // ── Streaming / incremental snapshot methods ────────────────────────────── @@ -559,8 +608,12 @@ public static List getChangesSince( List result = new ArrayList<>(); for (IcebergSnapshotInfo.ManifestFileInfo mf : newManifests) { + // Pass the manifest's known added_snapshot_id so entries with no + // own snapshot_id resolve the same "inherited" value here as they + // would via listFiles(), instead of falling back to -1. List entries = readManifestFile( - catalogName, namespace, tableName, config, mf.getManifestPath()); + catalogName, namespace, tableName, config, mf.getManifestPath(), + false, null, mf.getAddedSnapshotId()); result.addAll(entries); } return result; @@ -588,12 +641,13 @@ private static native byte[] nativeGetSnapshotInfo( private static native byte[] nativeReadManifestFile( String catalogName, String namespace, String tableName, String manifestPath, Map config, boolean compact, - String predicateJson); + String predicateJson, long inheritedSnapshotId); private static native int nativeReadManifestFileArrowFfi( String catalogName, String namespace, String tableName, String manifestPath, Map config, - String predicateJson, long[] arrayAddrs, long[] schemaAddrs); + String predicateJson, long[] arrayAddrs, long[] schemaAddrs, + long inheritedSnapshotId); private static native long nativeGetCurrentSnapshotId( String catalogName, String namespace, String tableName,