[Feature] Push down Spark physical bucket predicates #10595

Search before asking

• I searched in the issues and found nothing similar.

Motivation

A customer's Spark batch workload uses predicates such as __paimon_bucket IN (0, 1) to process selected physical buckets without enumerating business keys. Typical use cases include validating transformation logic before a full batch run, investigating data issues, and performing exploratory analysis on a subset of a large table.

Currently, Spark can return correct results but still scans data files from unrelated buckets when physical bucket predicates are not pushed down. For large production tables, this causes unnecessary storage I/O and puts additional pressure on Spark executors, wasting compute resources and increasing query latency.

Pushing these predicates into the Paimon scan would allow Spark to skip irrelevant buckets early, significantly reducing scan overhead and resource consumption, especially when querying a small subset of buckets.

Core already supports this filtering through ReadBuilder.withBucket(int) and ReadBuilder.withBucketFilter(Filter<Integer>). This feature connects Spark's supported __paimon_bucket predicates to those existing APIs.

Spark reproduction and before/after comparison

With a disposable Paimon catalog/database and the Paimon Spark extensions enabled:

CREATE TABLE bucket_filter_repro (id INT, payload STRING)
USING paimon
TBLPROPERTIES (
  'bucket' = '4',
  'bucket-key' = 'id',
  'file.format' = 'parquet'
);

INSERT INTO bucket_filter_repro
SELECT CAST(id AS INT), concat('v', CAST(id AS STRING))
FROM range(400);

-- Baseline: all buckets
SELECT id, payload, __paimon_bucket
FROM bucket_filter_repro;

-- One physical bucket
SELECT id, payload, __paimon_bucket
FROM bucket_filter_repro WHERE __paimon_bucket = 1;

-- A subset of physical buckets, without enumerating business keys
SELECT id, payload, __paimon_bucket
FROM bucket_filter_repro WHERE __paimon_bucket IN (0, 1);

-- A range of physical bucket IDs
SELECT id, payload, __paimon_bucket
FROM bucket_filter_repro
WHERE __paimon_bucket >= 1 AND __paimon_bucket < 3;

-- No matching bucket
SELECT id, payload, __paimon_bucket
FROM bucket_filter_repro WHERE __paimon_bucket = 99;

The local fixture produced four data files, one per bucket. With this change, the queries return identical results while reading only files from matching buckets:

For example, IN (0, 1) reduces planned data files from 4 to 2 and scan output from 400 to 204 rows. A non-matching bucket now plans zero data files.

Measurement: local Spark 3.5.8, Scala 2.12 and Java 11.0.25, with AQE disabled. Both runs use the same table snapshot. “Before” is a controlled run with pushedBucketFilters cleared from a copied scan while retaining Spark's residual filter; “after” uses the normal scan with bucket pushdown enabled. This isolates the effect of pushdown rather than comparing separate version installations. Result sets were checked for equality.

Planned buckets and data files were inspected through scan input splits; scan output rows use BatchScan's numOutputRows, before Spark's residual filter. These are data-file counts, not manifest-file counts: membership and range predicates use existing Core bucket filtering and may still require reading manifests. The counts above are specific to this fixture.

Solution

For Spark batch scans that already support reading __paimon_bucket, push supported physical bucket predicates into the existing Core APIs: withBucket(int) for single-bucket equality and withBucketFilter(Filter<Integer>) for membership and range conditions.

Preserve Spark SQL semantics, existing scan filters, and residual predicates where needed.

Validation

Add Spark integration tests checking result correctness and planned buckets/data files for equality, membership, ranges, and non-matching IDs. Cover NULL handling, boolean composition, existing filters, append/primary-key tables, and historical bucket counts.

Are you willing to submit a PR?

• I'm willing to submit a PR!