Spark: Fix type mismatch in SPJ with bucket partition key on string column

[ammarchalifah]

Problem

When a table is partitioned by bucket(N, string_column), the bucket transform produces an Integer partition value. During Storage Partitioned Joins (SPJ), Spark reads partition values through StructInternalRow, which calls struct.get(ordinal, CharSequence.class) in getUTF8StringInternal(). This assumes the value is always a CharSequence, causing a ClassCastException:

IllegalArgumentException: Wrong class, expected java.lang.CharSequence, but was java.lang.Integer, for object: 1

This affects any SPJ query (e.g. MERGE INTO or JOIN) on tables partitioned
with bucket(N, string_column).

Fix

Changed getUTF8StringInternal() to use struct.get(ordinal, Object.class) instead of struct.get(ordinal, CharSequence.class), then call value.toString(). This follows the same pattern already used by getBinaryInternal() in the same class, which uses Object.class to handle multiple possible runtime types.

The fix is applied to all Spark versions: 3.4, 3.5, 4.0, and 4.1.

Testing

• Added testJoinsWithBucketingOnStringColumn using the existing checkJoin helper to cover bucket-only partitioning on string columns.
• Added testJoinsWithIdentityAndBucketOnStringColumn as a targeted regression test for the exact scenario from the issue: identity + bucket partitioning on a string column with an SPJ join.

Both tests are added consistently across all 4 Spark versions.

Notes

AI tools were used to assist with drafting this change. I have reviewed and
validated the logic, tests, and code style end-to-end.

Closes#15349

[ammarchalifah]

This PR is a re-implementation of this closed PR: #15555

I was the reporter that filed the bug report, and really needed this bug to be fixed.

[ammarchalifah]

@huaxingao could you review this PR when you have time?

[RussellSpitzer]

Generally we only do a single Spark version (the latest) for changes followed by a backport PR to avoid duplicates during review

[RussellSpitzer]

My gut reading here is that we shouldn't be changing StructInternalRow, calling "getString" on a non string object should not default to calling "toString" on that object. This would just silently widen types in a way that we don't do anywhere else.

It feels like we are fixing a bad caller by changing the behavior rather than fixing the caller.

[RussellSpitzer]

I couldn't get a repo on 4.0 or 4.1, I think this is already fixed in the newer Spark versions (calling the correct function) That leaves 3.5 with our custom code. Let me take a look there

[ammarchalifah]

Generally we only do a single Spark version (the latest) for changes followed by a backport PR to avoid duplicates during review

Got it, I will change this PR to reflect this convention.

[ammarchalifah]

I couldn't get a repo on 4.0 or 4.1, I think this is already fixed in the newer Spark versions (calling the correct function) That leaves 3.5 with our custom code. Let me take a look there

Recently I upgraded my EMR to EMR 8.0.0 (running Spark 4.0.2 in the background), and using Iceberg 11, but the problem persisted.

[RussellSpitzer]

I am running the tests you added to the PR, so far I have not reproduced the issue on any Spark build.

Took your PR, removed the changes to Internal Row. Ran for example

./gradlew :iceberg-spark:iceberg-spark-4.1_2.13:test --tests "org.apache.iceberg.spark.sql.TestStoragePartitionedJoins.testJoinsWithBucketingOnStringColumn" --tests "org.apache.iceberg.spark.sql.TestStoragePartitionedJoins.testJoinsWithIdentityAndBucketOnStringColumn" -DsparkVersions=4.1 

Passed

Spark version	testJoinsWithBucketingOnStringColumn	testJoinsWithIdentityAndBucketOnStringColumn	Total
3.4	2/2 pass	2/2 pass	4/4
3.5	2/2 pass	2/2 pass	4/4
4.0	2/2 pass	2/2 pass	4/4
4.1	2/2 pass	2/2 pass	4/4

[ammarchalifah]

I am running the tests you added to the PR, so far I have not reproduced the issue on any Spark build.

Took your PR, removed the changes to Internal Row. Ran for example

./gradlew :iceberg-spark:iceberg-spark-4.1_2.13:test --tests "org.apache.iceberg.spark.sql.TestStoragePartitionedJoins.testJoinsWithBucketingOnStringColumn" --tests "org.apache.iceberg.spark.sql.TestStoragePartitionedJoins.testJoinsWithIdentityAndBucketOnStringColumn" -DsparkVersions=4.1 

Passed

Spark version testJoinsWithBucketingOnStringColumn testJoinsWithIdentityAndBucketOnStringColumn Total
3.4 2/2 pass 2/2 pass 4/4
3.5 2/2 pass 2/2 pass 4/4
4.0 2/2 pass 2/2 pass 4/4
4.1 2/2 pass 2/2 pass 4/4

Hmm, this is confusing. In my production case, my actual table has more partition keys, and I'm doing a join on a subset of key with spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled. Let me add additional test case that mirror this.

[ammarchalifah]

Added a few test cases and executed it without my change, still passing on all test cases. I'm struggling to reproduce the issue, could it possibly caused by the fact I'm using AWS Glue as my catalog? 🤔

[RussellSpitzer]

That could be it, it's possible the catalog is replying with the incorrect types? or it could be something EMR specific?

[ammarchalifah]

Oh, there is one more hypothesis that I'm checking right now. My workload is using MERGE INTO, let me add the test case and see if I can reproduce it.

[ammarchalifah]

Example of my spec from Iceberg metadata

{"spec-id":5,"fields":[{"name":"collected_date","transform":"identity","source-id":42,"field-id":1000},{"name":"user_id_bucket_32","transform":"bucket[32]","source-id":1,"field-id":1001},{"name":"category_id_bucket_32","transform":"bucket[32]","source-id":3,"field-id":1004}]}]

Error message (AWS Glue catalog, EMR 8.0.0, Spark 4.0.2)

2026-05-19 07:57:00,385 - table-raw-processor - INFO - Executing SPJ MERGE INTO: MERGE INTO glue_catalog.`db`.`table` AS t
USING glue_catalog.`temp-db`.`tmp_table` AS s
ON t.user_id = s.user_id AND t.category_id = s.category_id
WHEN MATCHED AND (t.collected_at IS NULL OR t.collected_at <= s.collected_at)
THEN UPDATE SET *
WHEN NOT MATCHED
THEN INSERT *
2026-05-19 07:57:41,230 -table-raw-processor - ERROR - Error: Wrong class, expected java.lang.CharSequence, but was java.lang.Integer, for object: 2
Traceback (most recent call last):
...
File "/mnt1/yarn/usercache/hadoop/appcache/application_1779177216710_0001/container_1779177216710_0001_01_000001/pyspark.zip/pyspark/sql/session.py", line 1810, in sql
File "/mnt1/yarn/usercache/hadoop/appcache/application_1779177216710_0001/container_1779177216710_0001_01_000001/py4j-0.10.9.9-src.zip/py4j/java_gateway.py", line 1362, in __call__
File "/mnt1/yarn/usercache/hadoop/appcache/application_1779177216710_0001/container_1779177216710_0001_01_000001/pyspark.zip/pyspark/errors/exceptions/captured.py", line 288, in deco
pyspark.errors.exceptions.captured.IllegalArgumentException: Wrong class, expected java.lang.CharSequence, but was java.lang.Integer, for object: 2

[ammarchalifah]

Managed to reproduce the issue with a lightweight Spark job triggered on EMR 8.0.0.

I put the full repro script + full stdout in this issue: #15349

[ammarchalifah]

Changed the test case to test SPJ for SELECT and MERGE INTO for join keys subset of partition keys

[ammarchalifah]

Reproduced a test suite to mimic the behaviour with vanilla Spark & Hadoop catalog. The error message is slightly different, but the gist of the issue is the same

java.lang.IllegalStateException: Unknown type for int field. Type name: java.lang.String
at org.apache.iceberg.spark.source.StructInternalRow.getInt(StructInternalRow.java:138)

Happens both in SELECT & MERGE INTO where join keys are subset of partition keys. Partition keys are identity(a), bucket(b), bucket(c), while join keys are b & c.

When partition columns are reorganized e.g. bucket(b), bucket(c), identity(a), the test suite passed, suggesting the issue lies in ordinal matching between Iceberg's partition ordering vs Spark's expectation. @RussellSpitzer seems like you're on point that this is not an Iceberg bug, but Spark-side bug.

[ammarchalifah]

Seems like the issue is already fixed on Spark 4.2, but not backported to earlier versions. Related PR: apache/spark#54330

[RussellSpitzer]

Thanks for looking into that, If that's the case I would recommend we don't do this patch for Iceberg but push for backporting the fix on Spark

[ammarchalifah]

Agree on the not fixing this on Iceberg part.
Unfortunately the Spark committer stated the changes will not be backported to earlier versions, and he clarified that the PR might not help with this particular bug.

Hi @ammarchalifah, no, those can't be backported to earlier versions. I doubt that the refactor or the other improvement would fix your issue, but there are many other SPJ fixes in the upcoming 4.2 so you should probably give it a try.

However, the bug still exists, as demonstrated by the newly added test in this PR + the EMR repro script in the issue.

[github-actions]

This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.

[github-actions]

This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.