Skip to content

[spark] Support merge-on-read for postpone bucket tables#8802

Open
Zouxxyy wants to merge 3 commits into
apache:masterfrom
Zouxxyy:xinyu/postpone-query
Open

[spark] Support merge-on-read for postpone bucket tables#8802
Zouxxyy wants to merge 3 commits into
apache:masterfrom
Zouxxyy:xinyu/postpone-query

Conversation

@Zouxxyy

@Zouxxyy Zouxxyy commented Jul 22, 2026

Copy link
Copy Markdown
Contributor

Purpose

Postpone bucket tables buffer batch-written records in the postpone bucket, so these records are not visible to ordinary queries until postpone compaction finishes.

This change introduces opt-in merge-on-read support through spark.paimon.postpone.merge-on-read=true.

  • Add an engine-neutral Core plan and read API for planning real-bucket splits and physical postpone-file tasks from one snapshot.
  • Preserve postpone-file replay order with deterministic sequence ranges, route postpone records to their target buckets, and reuse Paimon's existing merge engine for each bucket.
  • Add a Spark DSv2 scan and physical execution path which clusters real split metadata and postpone records by partition and target bucket before delegating the merge to Core.
  • Fall back to the ordinary Paimon scan when no postpone files are present, keeping the existing pushdowns and execution path.
  • Share bucket assignment and postpone-file ordering between merge-on-read and the Spark postpone compact procedure.

The option is disabled by default, so existing delayed-visibility behavior is unchanged.

Tests

  • Core targeted tests: PostponeUtilsTest and TableScanTest#testPostponeMergeReadBuilderAndPushDown — 11 tests passed.
  • Spark PostponeBucketTableTest — 14 tests passed.
  • Spark PrimaryKeyVectorSearchTest — 12 tests passed.
  • Compiled the Spark 3.2, 3.5, and 4.1 modules.
  • Ran Spotless, regenerated the Core configuration documentation, and verified git diff --check.

@Zouxxyy
Zouxxyy marked this pull request as draft July 22, 2026 11:22
@Zouxxyy
Zouxxyy marked this pull request as ready for review July 23, 2026 02:12
@JingsongLi

Copy link
Copy Markdown
Contributor

PR generates a global sequence in PostponeUtils based on creationTime → fileName → line order within the file. However, the actual behavior of Postpone should be to guarantee the file order for a given (table, partition, writeId); when different writers enter the same bucket writer in parallel, the final order depends on the arrival order of the multi-input channels in Flink.

@JingsongLi

Copy link
Copy Markdown
Contributor

create() first checks a snapshot to determine whether to follow a special execution path, while plan() re-parses the latest snapshot; they should use the same snapshot.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants