apache / apache/paimon

[Feature] Support per-partition bucket counts in the Spark writer (fixed-bucket tables)

Open
#8,349 0 comments 0 reactions 0 assignees View on GitHub
enhancement
Dominant language
Java
Stars
3.4k
Forks
1.4k
Avg merge
1d 11h
Merged PRs (30d)
396

Description

### Search before asking

- [x] I searched in the [issues](https://github.com/apache/paimon/issues) and found nothing similar.

### Motivation

## Summary

Context: follow-up to PR #7865, review comment https://github.com/apache/paimon/pull/7865#discussion_r3458901133.

Per-partition bucket counts are currently honored by the **Flink** writer (via `table.createRowKeyExtractor()` backed by `PartitionBucketMapping`), but the **Spark** fixed-bucket write path still derives the bucket from the single table-level bucket count. After changing the table-level `bucket` while existing partitions keep a different bucket count (e.g. following a per-partition rescale), Spark can route rows to buckets that do not belong to the partition, breaking the per-partition layout.

## Background / Root cause
On the Spark side, `BUCKET_COL` for fixed-bucket (`HASH_FIXED`) tables is computed in `PaimonSparkWriter.write` using a single table-level literal `coreOptions.bucket()`:

- **Fast path** (`paimonExtensionEnabled` + `BucketFunction.supportsTable`): computes `BUCKET_COL` via the `fixed_bucket` UDF (`BucketExpression.FIXED_BUCKET`) with `numBuckets = coreOptions.bucket()` — a literal, so it cannot vary per partition.
- **Fallback path** (`CommonBucketProcessor`): uses `table.createRowKeyExtractor()`, which *is* per-partition aware via `PartitionBucketMapping`, but reloads the mapping per task.
- `PaimonWriteRequirement` (V2 distribution) also clusters by `Expressions.bucket(numBuckets, keys)` using the table-level count.

The two Spark write paths are mutually exclusive per write:
- **V2 path:** `PaimonV2Write` → `PaimonWriteRequirement` (no `PaimonSparkWriter`).
- **V1 / command path** (MERGE/UPDATE/DELETE/`WriteIntoPaimonTable`): `PaimonSparkWriter`.

## Failure scenario (reproducible)
1. Create a partitioned fixed-bucket table with `bucket = 2`.
2. Rescale one partition (e.g. `p1`) to 4 buckets, leaving `p2` at 2 (`CALL sys.rescale(...)`).
3. `ALTER TABLE ... SET TBLPROPERTIES ('bucket' = '4')`.
4. Write new keys into `p2` via Spark (a `MERGE INTO` reaches the `PaimonSparkWriter` fast path).
5. Spark computes the bucket with count = 4 and attempts to write `p2` into bucket 2/3, which don't exist in that 2-bucket partition.

Today this is caught by the core write-side safety net added in commit `c5172e855` ("Reject bucket writes outside partition layout"), which throws:

> `Trying to write bucket 3 to partition {pt=p2}, but the partition only has 2 buckets ...`

So data isn't silently corrupted, but the write simply fails — i.e. Spark cannot write to tables with per-partition bucket layouts.

## Goal
Make the Spark fixed-bucket write path resolve the bucket count **per partition**, matching Flink, so writes succeed and rows land in the correct bucket for their partition.

### Solution

_No response_

### Anything else?

_No response_

### Are you willing to submit a PR?

- [ ] I'm willing to submit a PR!

Contributor guide

No contributing guide indexed for this repository

Research direction

Start by tracing the V2 path through PaimonV2Write and PaimonWriteRequirement, then the V1 path through PaimonSparkWriter and CommonBucketProcessor. Compare their fixed-bucket handling with table.createRowKeyExtractor() and PartitionBucketMapping. Done means Spark resolves bucket counts per partition in both paths and the reproducible p1/p2 rescale scenario writes successfully.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, spark
Domain
data-engineering, distributed-systems
Issue type
Feature
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
55/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.