feat: implement 2PC support in parquet writer by siiddhantt · Pull Request #959 · datazip-inc/olake

siiddhantt · 2026-05-20T12:25:07Z

Description

Fixes #856

Adds table-local 2PC metadata for the Parquet destination under:

<namespace>/<table>/_olake_2pc/

The goal is to make Parquet recover from destination-side committed progress instead of relying only on the external source state file. This is needed across full refresh, incremental, and CDC syncs:

full refresh needs destination-side tracking for chunk/thread IDs that were already committed, so retries do not rewrite completed chunks
incremental and CDC need committed cursor/LSN metadata to be recoverable if OLake fails before saving the external state file
no-op CDC/incremental syncs still need a destination-side progress record when the cursor/LSN advances without producing Parquet data files
interrupted Parquet writes should remain hidden under _olake_2pc instead of leaving final-path orphan files for later retries to reason about

This PR keeps the main change inside destination/parquet and uses hidden per-thread staging plus a completed marker:

_olake_2pc/<escaped-thread-id>.staging/_completed.json

Protocol:

data files are first written under _olake_2pc/<escaped-thread-id>.staging/
after staged files are durable, Close() writes _completed.json
staged Parquet files are then promoted to the final table paths
_completed.json is kept as the durable Parquet-local progress marker for that thread
Setup() calls load2PCState() to recover table-local metadata before a retry starts
if setup finds staging with _completed.json, it treats that thread as committed and promotes any remaining staged files
if setup finds staging without _completed.json, it treats that attempt as incomplete and removes the hidden staging prefix before retry

For full refresh, _completed.json can be {} because the committed thread/chunk ID is encoded in the staging folder name. For CDC, incremental, and no-op progress, _completed.json contains the optional types.MetadataState, tying cursor/LSN progress to the same completed marker.

Implementation details:

added parquetDataFile so close, upload, and staging promotion use one file identity instead of reconstructing paths in multiple places
added metadataState() to normalize destination final state into the same types.MetadataState shape used by destination recovery
added local and S3 helpers for completed marker reads/writes, staging discovery, staging promotion, and incomplete staging cleanup
removed the earlier Parquet-local state.json, commits/*.json, preparing/*.json, and prepare-marker file-list protocol
added metadata-only completed marker support so CDC/incremental no-op syncs can still persist destination-committed cursor/LSN state
added a narrow Postgres CDC recovery guard so a retry can accept Parquet’s destination-committed LSN when the external source state file is stale and the replication slot is already at that committed LSN

The completed marker is the single Parquet-local progress record. If the process fails before _completed.json is written, setup treats staging as incomplete and removes it. If the process fails after _completed.json is written but before promotion finishes, setup rolls that completed staging forward.

Type of change

Bug fix (non-breaking change which fixes an issue)
New feature (non-breaking change which adds functionality)
Breaking change (fix or feature that would cause existing functionality to not work as expected)
This change requires a documentation update

How Has This Been Tested?

Parquet 2PC unit tests: go test ./destination/parquet -count=1
Destination package tests: go test ./destination/... -count=1
Core state/parser tests: go test ./types ./pkg/parser -count=1
Postgres non-container unit tests: go test ./drivers/postgres/internal -run TestConfig -count=1
Binary compile: go build -o /private/tmp/olake-build-check .
Parquet lint: golangci-lint run ./destination/parquet
Manual Postgres-backed demo:
- verified interrupted full-refresh retry completes with no incomplete staging dirs and no staged Parquet files left behind
- verified full-refresh completed markers are written for committed chunks
- verified incremental retry recovers cursor state from Parquet completed markers
- verified CDC retry accepts destination-committed LSN when external source state is stale
- verified no-op/metadata-only completed marker behavior
- verified completed staging is promoted on recovery
- verified incomplete hidden staging is removed before retry

Screenshots or Recordings

WIP

Documentation

Documentation Link: [link to README, olake.io/docs, or olake-docs]
N/A (bug fix, refactor, or test changes only)

Related PR's (If Any):

Reference: feat: Implement 2-Phase Commit for Iceberg Sync #698

…nd marker

…into feat/parquet-2pc

siiddhantt · 2026-06-05T07:36:46Z

Updated the PR with the _completed.json-inside-staging approach discussed in #856.

Current Parquet 2PC shape:

<namespace>/<table>/_olake_2pc/<escaped-thread-id>.staging/_completed.json

Summary of the refactor:

kept _completed.json as the durable completed marker inside each staging folder
full refresh uses {} markers and derives committed chunk/thread IDs from the staging folder name
CDC/incremental/no-op progress stores optional types.MetadataState in _completed.json
setup promotes completed staging and removes incomplete staging before retry

Also reran the focused/unit/build/lint checks and the Postgres-backed manual demo for full refresh, incremental, and CDC recovery.

siiddhantt · 2026-06-08T07:27:16Z

@nayanj98 looping you in as @vaibhav-datazip suggested.

This PR is ready for review now and the description is updated with the final flow and test coverage.

nayanj98 · 2026-06-09T12:14:08Z

@siiddhantt Assigning @vaibhav-datazip to review your PR.

siiddhantt · 2026-06-10T14:10:54Z

@vaibhav-datazip I found the CI failure. It looks test-harness related rather than a 2PC flow failure.

The integration cleanup helper deletes only final .parquet files between Parquet subtests, but now _olake_2pc is part of the table-local destination state. That leaves stale _completed.json markers behind, so the next subtest can skip already-committed chunks while the actual data files were manually removed by the test helper.

I’ll push a small fix to clear _olake_2pc/ in the Parquet integration cleanup as well, matching how DropStreams() clears the full table prefix.

vaibhav-datazip · 2026-06-14T16:15:34Z

 }

-func (p *Parquet) closePqFiles(ctx context.Context, _ any, closeOnError bool) error {
+func (p *Parquet) dataFiles() []parquetDataFile {


can you rename this function to a more appropriate name, also add a comment for this function

Done, renamed this to stagedDataFiles() and added a short comment. The helper returns the current thread’s staged parquet file identities used during close/upload/promotion, so the old dataFiles() name was too generic.

vaibhav-datazip · 2026-06-14T16:26:04Z

-	if len(remainingStreams) > 0 {
+	// Destination 2PC metadata can be ahead of the external source state when
+	// the destination commit succeeded but saving the source state failed. If all
+	// streams are already committed and the slot is exactly at that committed LSN,
+	// there is no WAL left to replay; PostCDC will persist the recovered state.
+	skipValidation := recoveryLSN != nil && len(remainingStreams) == 0 && *recoveryLSN == slot.LSN
+	if !skipValidation {
+		// validate global state (might got invalid during full load)


what's the need of this change ?

This is for the stale external state recovery case.

Scenario:

Parquet commits destination progress at LSN B in _completed.json

the Postgres slot is already at B

OLake crashes before the external source state file is saved, so the next run still starts from old state A

Without this guard, Postgres fails on the A vs B LSN mismatch before destination metadata can repair the state. The skip is intentionally narrow: it only applies when destination metadata is ahead, all streams are already committed, and the replication slot is exactly at the destination-committed LSN. Other mismatches still fail as before.

vaibhav-datazip · 2026-06-14T17:03:24Z

+	return filepath.Join(p.local2PCPath(), p.stagingDirName(threadID))
+}
+
+func (p *Parquet) stagingDataDir(basePath string) string {


can you change the variable name of the argument, also can you add short comments for the functions in this file

Done, renamed the argument to avoid confusion with p.basePath, and added short comments around the 2PC helpers.

vaibhav-datazip · 2026-06-15T13:21:39Z

+}
+
+func (p *Parquet) listS3CompletedMarkers(ctx context.Context) ([]parquet2PCCompletedMarker, error) {
+	prefix := p.s3ObjectPath(filepath.Join(p.basePath, parquet2PCDir)) + "/"


why are we saving staging files under _olake_2pc folder and not directly in the root folder

Keeping staging under _olake_2pc is intentional. Staged files are not committed table data yet, so placing them under the table root can make raw Parquet readers see partial/uncommitted files. _olake_2pc keeps staging hidden but still table-local, and DropStreams() clears it naturally with the table prefix.

vaibhav-datazip · 2026-06-15T18:30:41Z

+type parquet2PCCompletedMarker struct {
+	ThreadID      string
+	MetadataState *types.MetadataState
+	modTime       time.Time


do we need modTime ?

Yes, we need an ordering value because CDC/incremental/no-op syncs can leave multiple completed markers over time, and setup needs to restore the latest metadata state. I renamed modTime to completedAt and added a short comment to make that purpose clearer.

vaibhav-datazip · 2026-06-16T06:30:34Z

+	err := p.s3Client.ListObjectsPagesWithContext(ctx, &s3.ListObjectsInput{
+		Bucket: aws.String(p.config.Bucket),
+		Prefix: aws.String(prefix),
+	}, func(page *s3.ListObjectsOutput, _ bool) bool {
+		keys := make([]string, 0, len(page.Contents))
+		for _, obj := range page.Contents {
+			if obj.Key != nil {
+				keys = append(keys, *obj.Key)
+			}
+		}
+		if len(keys) == 0 {
+			return true
+		}
+		concurrency := min(len(keys), 8)
+		pageErr = utils.Concurrent(ctx, keys, concurrency, func(_ context.Context, key string, _ int) error {
+			return p.deleteS3Object(ctx, key)
+		})
+		return pageErr == nil


Retries should be added in this as well as other s3 related functions as well like we have in clearS3files

Agreed, added retries for the new S3 2PC operations using the same RetryWithSkip + isRateLimitError pattern used by existing Parquet S3 cleanup. This now covers staging upload, marker read/write, listing, copy, and delete paths.

vaibhav-datazip · 2026-06-16T08:11:58Z

+		marker, err := p.readLocalCompletedMarker(entry.Name())
+		if os.IsNotExist(err) {
+			continue
+		}


why don't we create a map here only and not depend on list staging dirs function at all, we will mark the folders without completed marker as false otherwise true and data will be there

Agreed, refactored this into a single staging-entry map. Each staging dir now maps to either a completed marker or nil, so setup can promote committed staging and delete incomplete staging without separately listing completed markers and staging dirs.

…into feat/parquet-2pc

vaibhav-datazip · 2026-06-18T05:52:33Z

@siiddhantt once you are done with your changes after the review , please re-request the review from here so that I know the pr is ready again

feat: implement 2PC support in parquet writer with state management a…

a70a2da

…nd marker

siiddhantt requested a deployment to integration_tests May 20, 2026 12:25 — with GitHub Actions Waiting

Merge branch 'staging' into feat/parquet-2pc

730bb1b

siiddhantt requested a deployment to integration_tests May 20, 2026 12:27 — with GitHub Actions Waiting

siiddhantt added 2 commits May 20, 2026 19:27

refactor: add unit tests + lint fixes

5c42f45

Merge branch 'feat/parquet-2pc' of https://github.com/siiddhantt/olake …

8de4b98

…into feat/parquet-2pc

siiddhantt requested a deployment to integration_tests May 20, 2026 13:57 — with GitHub Actions Waiting

siiddhantt marked this pull request as ready for review May 20, 2026 14:25

siiddhantt marked this pull request as draft May 21, 2026 10:11

siiddhantt added 2 commits May 22, 2026 10:31

refactor(parquet): recover 2pc state from commit markers

6605399

fix(postgres): allow CDC recovery from committed destination LSN

27bd550

siiddhantt requested a deployment to integration_tests May 22, 2026 06:02 — with GitHub Actions Waiting

Merge branch 'staging' into feat/parquet-2pc

4170b6e

siiddhantt requested a deployment to integration_tests May 27, 2026 07:27 — with GitHub Actions Waiting

siiddhantt marked this pull request as ready for review May 27, 2026 09:51

refactor(parquet): update 2PC commit handling and staging management

8ee29ed

siiddhantt requested a deployment to integration_tests May 27, 2026 09:54 — with GitHub Actions Waiting

siiddhantt mentioned this pull request May 27, 2026

implement 2pc in parquet writer (difficulty: hard) #856

Open

Merge branch 'staging' into feat/parquet-2pc

41c2929

siiddhantt requested a deployment to integration_tests June 2, 2026 13:07 — with GitHub Actions Waiting

Merge branch 'staging' into feat/parquet-2pc

00be27a

siiddhantt requested a deployment to integration_tests June 4, 2026 23:28 — with GitHub Actions Waiting

siiddhantt added 2 commits June 5, 2026 12:47

refactor(parquet): store 2pc progress in completed staging markers

1eb835e

Merge branch 'feat/parquet-2pc' of https://github.com/siiddhantt/olake …

113d859

…into feat/parquet-2pc

siiddhantt requested a deployment to integration_tests June 5, 2026 07:18 — with GitHub Actions Waiting

Merge branch 'staging' into feat/parquet-2pc

b4511be

siiddhantt requested a deployment to integration_tests June 8, 2026 07:25 — with GitHub Actions Waiting

nayanj98 requested a review from vaibhav-datazip June 9, 2026 12:14

Merge branch 'staging' into feat/parquet-2pc

5d4bd83

vaibhav-datazip had a problem deploying to integration_tests June 9, 2026 12:22 — with GitHub Actions Failure

vaibhav-datazip temporarily deployed to integration_tests June 9, 2026 12:22 — with GitHub Actions Inactive

test: clean parquet 2pc metadata in integration reset

0924400

siiddhantt temporarily deployed to integration_tests June 10, 2026 14:18 — with GitHub Actions Inactive

siiddhantt temporarily deployed to integration_tests June 10, 2026 14:19 — with GitHub Actions Inactive

Merge branch 'staging' into feat/parquet-2pc

79d0393

siiddhantt requested a deployment to integration_tests June 16, 2026 08:22 — with GitHub Actions Waiting

vaibhav-datazip reviewed Jun 16, 2026

View reviewed changes

siiddhantt added 2 commits June 17, 2026 09:37

fix: address parquet 2pc review feedback

e380d0b

Merge branch 'feat/parquet-2pc' of https://github.com/siiddhantt/olake …

f4fc2b7

…into feat/parquet-2pc

siiddhantt requested a deployment to integration_tests June 17, 2026 04:08 — with GitHub Actions Waiting

Conversation

siiddhantt commented May 20, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Type of change

How Has This Been Tested?

Screenshots or Recordings

Documentation

Related PR's (If Any):

Uh oh!

siiddhantt commented Jun 5, 2026

Uh oh!

siiddhantt commented Jun 8, 2026

Uh oh!

nayanj98 commented Jun 9, 2026

Uh oh!

siiddhantt commented Jun 10, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

vaibhav-datazip commented Jun 18, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

siiddhantt commented May 20, 2026 •

edited

Loading