(wip | no review) InHouse Replication - #627
Draft
mkuchenbecker wants to merge 4 commits into
Draft
Conversation
Worktree-isolated PoC for replacing GaaS replication with a Spark Scala
function (eager full-history, retention-bounded, physical copy with
rename; snapshot-based recovery via source-snapshot-id summary stamp).
- replication-poc/{PLAN,CHECKLIST,NOTES}.md — scope, decisions, phases
P0-P2, the OH-acceptance spike (gate), iteration loop, oracle.
- oh-hadoop-spark harness: run_replicate.sh (spark-shell -i against the
local openhouse catalog), replicate.scala (spike + function outline),
seed_sources.scala (shape matrix); spark-master mounts the recipe dir
so the scala iterates live (edit on host -> docker exec re-run).
Background: ~/code/docs/gaas-analysis/{README,options}.md
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
OH accepts a snapshot committed through its catalog whose DataFiles point at files we placed (copied + relocated via DataFiles.builder.copy.withPath, source metrics carried), and source-snapshot-id stamps into the snapshot summary. Oracle dst==src. Files are ORC. Gate cleared; P0 next. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
Real replicateTable: full-history replication (floor full-materialization + append deltas), per-snapshot source-snapshot-id summary stamp, summary-based reconcile. Oracle (dest AS OF mapped_i == src AS OF i) passes at every snapshot vs real OpenHouse. Idempotent re-run = no-op; incremental replays only the new snapshot. Remaining: P1 resume, P2 (removed/MOR/compaction/schema). Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
Replay handles removed files (CoW delete/overwrite/compaction) via newOverwrite; all oracles pass, history preserved. Finding: OpenHouse default delete/overwrite is copy-on-write (no MOR delete files emitted by default DELETE/OVERWRITE). Refactor per review: distinct replay cases (Replay(add,remove) + dels-match for append-vs-overwrite, no nested if/else); functional idioms (Option.map/filter/foreach, Try.getOrElse, forall, zipWithIndex) replacing existence checks and mutable flags. Remaining: P1 resume; P2 MOR (force merge-on-read -> positional delete rewrite) + schema evolution. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
mkuchenbecker
commented
Jun 1, 2026
|
|
||
| def ensureDest(source: String, dest: String): Table = | ||
| Try(loadTbl(dest)).getOrElse { | ||
| spark.sql(s"CREATE TABLE $dest (${spark.table(source).schema.toDDL}) USING iceberg") |
Contributor
Author
There was a problem hiding this comment.
This isn't testing openhouse. This is testing iceberg. Are you actually following the openhouse guide for how to create openhouse tables?
mkuchenbecker
commented
Jun 1, 2026
|
|
||
| // committed history, oldest -> newest (walk parentId from current) | ||
| def chainOf(t: Table): List[Snapshot] = { | ||
| def walk(s: Snapshot, acc: List[Snapshot]): List[Snapshot] = |
Contributor
Author
There was a problem hiding this comment.
walking snapshots can get expensive. really expensive. lets add a limit, configured to default=3. If greater than limit, it makes the final as a single commit. That way you can say (1) to just get latest or 1000 if you want 1000, but "all" is not an option.
This is because it will be run on the driver is my expectation so our scalability is limited to the number os snapshots. This also prevents malformed tables with too many snopshots from exploding the tool.
If we could make this run on executors, that would be an option.
mkuchenbecker
commented
Jun 1, 2026
| if (s == null) acc else walk(Option(s.parentId()).map(t.snapshot(_)).orNull, s :: acc) | ||
| walk(t.currentSnapshot(), Nil) | ||
| } | ||
| def rowsAsOf(tbl: String, snapId: Long): List[String] = |
mkuchenbecker
commented
Jun 1, 2026
| spark.read.format("iceberg").option("snapshot-id", snapId).load(tbl).collect().map(_.toString).sorted.toList | ||
|
|
||
| // copy one source data file to a remapped path under dest's location; relocate the DataFile (metrics carried) | ||
| def relocate(dst: Table, df: DataFile, hconf: org.apache.hadoop.conf.Configuration): DataFile = { |
mkuchenbecker
commented
Jun 1, 2026
mkuchenbecker
left a comment
mkuchenbecker
left a comment
Contributor
Author
There was a problem hiding this comment.
any function only called 1-2 times does not need a helper. It decreases readability.
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.
(WIP — do not review)
InHouse Replication — a PoC for replacing GaaS (Gobblin-as-a-Service) carbon-copy replication with a
plain Spark Scala function that runs inside our own house: an eager, full-history (retention-bounded),
physical copy of an OpenHouse table to a renamed destination, committed through the normal OpenHouse
catalog path, with snapshot-based recovery.
This is exploratory work in an isolated worktree; it is not for review or merge yet.
What's here
replication-poc/{PLAN,CHECKLIST,NOTES}.md— decision trail, phased plan (P0–P2), iteration loop, oracle.infra/recipes/docker-compose/oh-hadoop-spark/:replicate.scala— the replication function + an in-script P0/P1/P2 scenario and a time-travel oracle.run_replicate.sh,seed_sources.scala— harness (spark-shell-iagainst the localopenhousecatalog).docker-compose.yml— mounts the recipe dir intospark-masterso the scala iterates live.Approach (decided; rationale in the docs)
recovery against corruption-between-runs.
metrics (no re-stat), stamp
source-snapshot-idin each dest snapshot summary (recovery index + reconcile).Status (validated against local Dockerized OpenHouse)
dest AS OF i == src AS OF iat every snapshot.equality deletes + schema/partition-spec evolution; P1 resume-after-partial-copy.
Background: GaaS analysis in
~/code/docs/gaas-analysis/(out-of-repo).🤖 Generated with Claude Code