feat(core): spark-parity writer: all write operations, table types, query types, markers, archival, locks, MDT indexes, parity harness - #666
Draft
vinothchandar wants to merge 24 commits into
Conversation
Introduce pyIceberg-style Table::create/append over Arrow batches with storage mutations, commit metadata serialization, and timeline reload so append-only COW Parquet tables can be written and read back locally.
Bootstrap MDT on create when enabled, encode files partition records, and update the files listing on append via a minimal HFile writer and log-block writer so v8 tables can serve listings from metadata.
Add SimpleIndex scaffolding and Table upsert/delete/overwrite paths that rewrite unpartitioned COW tables with commit-time merge semantics, including partial column updates and replacecommit overwrites.
Route upsert merges through RecordMerger so ordering fields keep the newer event-time record, with commit-seqno as tie-breaker, and reject custom payload or merger configs.
Support MERGE_ON_READ append via deltacommit base files, upsert via parquet log appends, and deletes via log blocks, with snapshot and read-optimized coverage on written tables.
Wire SimpleIndex into MOR upsert/delete tagging, clean up orphans on commit failure, and add regression coverage for empty inputs, schema mismatch, duplicate keys, multi-append upsert, and missing deletes.
COW upsert/delete with metadata enabled primed the MDT cache during listing, then updated MDT on disk without reloading it, so replacecommits hid old bases while MDT still listed them and snapshot reads returned empty. Also add COW/MOR lifecycle tests that assert MDT consistency and on-disk base vs log file counts.
Prefer fresh metadata-table opens and a timeline reload at write start so MDT/data reads stay relative to that snapshot. Commit locking and conflict checks remain for the concurrency-control project.
Encode commit and replace-commit timeline bytes from the Java .avsc files so Spark SpecificRecord readers see a compatible writer schema.
Switch metadata-table encoding to the vendored Java schema and add record-index entry encode/decode for RLI HFile payloads.
Generate Spark-style auto record keys when recordkey fields are unset, treat precombine as optional (COMMIT_TIME_ORDERING), and default Parquet compression to zstd.
Introduce RecordIndex backed by the metadata table, fall back to SimpleIndex when RLI is off, and extend write metadata helpers for RLI upserts and deletes.
Enable MDT+RLI by default with Java-shaped hoodie.properties, write meta fields, and append into hive-style partitions with file-size splits and RLI maintenance.
Add Iceberg-style update(filter, SET batch) for any-column predicates. Route delete(=|/IN) on record key or _hoodie_record_key through delete_keys/RLI, and keep a scan path for non-key filters.
…cycles Add cases for non-key delete, key-filter RLI routing, Iceberg-style update, and partitioned COW/MOR write layouts.
Clear Avro key fields and embed avro.java.string props so Java GenericDatumReader yields String keys; write valid MDT deltacommit Avro plus SCHEMA headers; fix log footer size to match Java (header includes trailing long; footer = header + MAGIC).
Spark FileSystemBackedTableMetadata only treats dirs as partitions when .hoodie_partition_metadata is present; without it MDT-off reads return count 0 despite valid data files.
Match Java ActiveTimeline: write .{action}.requested and inflight before
completed commit/deltacommit/replacecommit so Spark sees a fenced timeline.
Generate `{uuid}-0` data file ids, pack UUID bits into RLI (encoding 0),
and bootstrap/update record_index across the Java default of 10 file groups.
Honor page/block/dictionary/compression options with Java defaults for sizes and dictionary; keep zstd as the unset compression default.
Rewrite only indexed file groups on COW upsert/delete, assign a unique UUID base per partition write, and leave unrelated file groups untouched.
Write MDT/RLI before completing the data timeline so a mid-commit crash cannot leave a completed data instant without matching metadata.
Expose append_only, full-table overwrite as INSERT_OVERWRITE_TABLE, and dynamic_partition_overwrite for partition-scoped replaces, with COW tests.
…s, parity harness Bring the native write path to storage parity with the Apache Hudi Spark writer (Hudi 1.1.x, table version 9 default; tv8 retained) and prove it with a Spark-in-the-loop parity harness. Writer: - Two-phase locking per write (LockProvider + InProcessLockProvider): request instant under lock, complete + bookkeeping under lock. - Skew-adjusted monotonic instant times, formatted in the table's declared timeline timezone (LOCAL default, matching Spark) so interleaved writers produce ordered timelines. - Timeline-server-based marker files and eager rollback of failed writes (rollback plan/metadata, orphan MDT cleanup). - LSM timeline archival (manifest_N + _version_ + level-0 parquet) with reader-side archival-floor and fencing unions. - Small-file packing (UpsertPartitioner semantics) and group-aware COW rewrites; parallel write tasks via hoodie.write.task.parallelism. - MDT files/column_stats/partition_stats/record_index maintained per commit; MetadataPayloadMerger mirrors HoodieMetadataPayload combine semantics and is resolved from the MDT merge-strategy id. - Deletes that empty a file group write an empty base file version; commit metadata stores the data schema without meta fields; batches differing only in nullability are aligned. - MDT records decode with the writer schema of their source container, so mixed-writer tables (rs + Spark) read correctly. - Incremental queries follow Hudi 1.x completion-time semantics with exact instant-set filtering, shared by table.read and manual reader paths. Parity harness (make parity, HUDI_SPARK_PARITY=1, scripts/parity/): six scenarios against a local Spark 3.5 + Hudi bundle — rs<->Spark reads and writes, Spark compaction/clustering/clean on rs tables followed by rs writes, interleaved writers, a 36-commit mixed workload with archival verified via snapshot/time-travel/incremental in both engines, and a CDC-enabled table. Diff size justified: this lands the writer feature set plus its test matrix (six new integration suites and the parity harness).
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
Bring the native write path to storage parity with the Apache Hudi Spark
writer (Hudi 1.1.x, table version 9 default; tv8 retained) and prove it with
a Spark-in-the-loop parity harness.
How are the changes test-covered