Skip to content
Open
Show file tree
Hide file tree
Changes from 6 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 24 additions & 7 deletions iceberg/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -11,18 +11,35 @@ and the directory that contains the corresponding support code.

| Iceberg Version | Spark Version | Directory |
|-----------------|----------------------------|-------------------|
| 1.6.x | Spark 3.5.0-3.5.3 | `iceberg-1-6-x` |
| 1.9.x | Spark 3.5.4-3.5.9 | `iceberg-1-9-x` |
| 1.10.x | Spark 3.5.4-3.5.9, 4.0.x | `iceberg-1-10-x` |
| 1.6.x | Spark 3.5.1-3.5.3 | `iceberg-1-6-x` |
| 1.9.x | Spark 3.5.5-3.5.9 | `iceberg-1-9-x` |
| 1.10.x | Spark 3.5.6-3.5.9, 4.0.x | `iceberg-1-10-x` |
| 1.11.x | Spark 4.0.2+, 4.1.x | `iceberg-1-11-x` |

Iceberg GPU acceleration is currently supported on Spark 3.5.x, 4.0.x, and 4.1.x.
The authoritative integration-test compatibility list, including upstream-compatible
combinations that are not currently packaged, is maintained in
[`iceberg-versions.json`](iceberg-versions.json).

Each matrix entry describes one Apache Iceberg runtime version tested by cudf-spark. For that
Iceberg release, `upstream_minimums` is copied from the Spark versions in Apache Iceberg's
`gradle/libs.versions.toml`. Each key is a Spark major/minor family, and its value is the patch
release that Iceberg builds and tests against; cudf-spark treats that patch as the minimum
upstream-compatible version.

The `spark_versions` list is computed from the `spark*.version` properties in the root `pom.xml`.
For each family in `upstream_minimums`, it contains every cudf-spark shim whose patch version is
greater than or equal to the upstream minimum. A shim is marked as supported when cudf-spark
packages the corresponding Iceberg integration module. Upstream-compatible shims that are not
packaged remain in the list with `supported` set to `false` and an explanation in `reason`.

For Spark 3.5.4+, both `iceberg-1-9-x` and `iceberg-1-10-x` modules are compiled into the
build. The correct version-specific implementation is selected at runtime by probing the
`iceberg-spark-runtime` jar on the classpath. Version-specific code lives in distinct
sub-packages (`iceberg19x`, `iceberg110x`, `iceberg111x`) to avoid class conflicts, and the
common `ShimUtils` dispatcher delegates to the appropriate implementation.
build. The integration-test support baseline follows the Spark patch versions used to build
the corresponding Apache Iceberg release. The correct version-specific implementation is
selected at runtime by probing the `iceberg-spark-runtime` jar on the classpath.
Version-specific code lives in distinct sub-packages (`iceberg19x`, `iceberg110x`,
`iceberg111x`) to avoid class conflicts, and the common `ShimUtils` dispatcher delegates to
the appropriate implementation.

For Spark 4.0.0-4.0.1, only `iceberg-1-10-x` is compiled during the build. For Spark
4.0.2+, both `iceberg-1-10-x` and `iceberg-1-11-x` are compiled, and the correct
Expand Down
206 changes: 206 additions & 0 deletions iceberg/iceberg-versions.json

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Supported Spark × Iceberg combinations can and should be derived from Maven release profiles and module POMs. This JSON is not the source of truth has already diverged:

  • Spark 3.5.0 => no Iceberg version.
  • Spark 3.5.4 => despite packaging 1.9.x and 1.10.x selects none for testging
  • Spark 3.5.5 => omits packaged 1.10.x support.

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's a combination of upstream oss with our pom, and they are not diverged.

For example: Spark 3.5.0 => no Iceberg version, there is no 3.5.0 listed the json.

Original file line number Diff line number Diff line change
@@ -0,0 +1,206 @@
{
"iceberg_versions": [
{
"version": "1.6.1",
"upstream_minimums": {
"3.3": "3.3.4",
"3.4": "3.4.3",
"3.5": "3.5.1"
},
"spark_versions": [
{
"version": "3.3.4",
"supported": false,
"reason": "Iceberg GPU acceleration is not currently packaged for Spark 3.3.x"
},
{
"version": "3.4.3",
"supported": false,
"reason": "Iceberg GPU acceleration is not currently packaged for Spark 3.4.x"
},
{
"version": "3.4.4",
"supported": false,
"reason": "Iceberg GPU acceleration is not currently packaged for Spark 3.4.x"
},
{
"version": "3.5.1",
"supported": true
},
{
"version": "3.5.2",
"supported": true
},
{
"version": "3.5.3",
"supported": true
},
{
"version": "3.5.4",
"supported": false,
"reason": "The Iceberg 1.6.x integration module is not packaged for Spark 3.5.4 and later"
},
{
"version": "3.5.5",
"supported": false,
"reason": "The Iceberg 1.6.x integration module is not packaged for Spark 3.5.4 and later"
},
{
"version": "3.5.6",
"supported": false,
"reason": "The Iceberg 1.6.x integration module is not packaged for Spark 3.5.4 and later"
},
{
"version": "3.5.7",
"supported": false,
"reason": "The Iceberg 1.6.x integration module is not packaged for Spark 3.5.4 and later"
},
{
"version": "3.5.8",
"supported": false,
"reason": "The Iceberg 1.6.x integration module is not packaged for Spark 3.5.4 and later"
},
{
"version": "3.5.9",
"supported": false,
"reason": "The Iceberg 1.6.x integration module is not packaged for Spark 3.5.4 and later"
}
]
},
{
"version": "1.9.2",
"upstream_minimums": {
"3.4": "3.4.4",
"3.5": "3.5.5"
},
"spark_versions": [
{
"version": "3.4.4",
"supported": false,
"reason": "Iceberg GPU acceleration is not currently packaged for Spark 3.4.x"
},
{
"version": "3.5.5",
"supported": true
},
{
"version": "3.5.6",
"supported": true
},
{
"version": "3.5.7",
"supported": true
},
{
"version": "3.5.8",
"supported": true
},
{
"version": "3.5.9",
"supported": true
}
]
},
{
"version": "1.10.1",
"upstream_minimums": {
"3.4": "3.4.4",
"3.5": "3.5.6",
"4.0": "4.0.0"
},
"spark_versions": [
{
"version": "3.4.4",
"supported": false,
"reason": "Iceberg GPU acceleration is not currently packaged for Spark 3.4.x"
},
{
"version": "3.5.6",
"supported": true
},
{
"version": "3.5.7",
"supported": true
},
{
"version": "3.5.8",
"supported": true
},
{
"version": "3.5.9",
"supported": true
},
{
"version": "4.0.0",
"supported": true
},
{
"version": "4.0.1",
"supported": true
},
{
"version": "4.0.2",
"supported": true
},
{
"version": "4.0.3",
"supported": true
},
{
"version": "4.0.4",
"supported": true
}
]
},
{
"version": "1.11.0",
"upstream_minimums": {
"3.4": "3.4.4",
"3.5": "3.5.8",
"4.0": "4.0.2",
"4.1": "4.1.1"
},
"spark_versions": [
{
"version": "3.4.4",
"supported": false,
"reason": "Iceberg GPU acceleration is not currently packaged for Spark 3.4.x"
},
{
"version": "3.5.8",
"supported": false,
"reason": "The Iceberg 1.11.x integration module is not currently packaged for Spark 3.5.x"
},
{
"version": "3.5.9",
"supported": false,
"reason": "The Iceberg 1.11.x integration module is not currently packaged for Spark 3.5.x"
},
{
"version": "4.0.2",
"supported": true
},
{
"version": "4.0.3",
"supported": true
},
{
"version": "4.0.4",
"supported": true
},
{
"version": "4.1.1",
"supported": true
},
{
"version": "4.1.2",
"supported": true
},
{
"version": "4.1.3",
"supported": true
}
]
}
]
}
3 changes: 3 additions & 0 deletions integration_tests/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -539,6 +539,9 @@ properly without it. These tests assume Iceberg is not configured and are disabl
If Spark has been configured to support Iceberg then these tests can be enabled by adding the
`--iceberg` option to the command.

Set `ICEBERG_TEST_FAST_RUN=1` to skip redundant, high-cost cases while retaining tests that
specifically require a local Hadoop catalog. The flag defaults to `0` when it is not set.

When testing Iceberg package-private access paths, load the local Iceberg runtime jar with
`ICEBERG_EXTRA_CLASSPATH` instead of `PYSP_TEST_spark_jars` or
`PYSP_TEST_spark_jars_packages`. The test driver will place the RAPIDS, test, and Iceberg
Expand Down
4 changes: 4 additions & 0 deletions integration_tests/src/main/python/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -119,6 +119,10 @@ def is_iceberg_remote_catalog():
v = os.environ.get('ICEBERG_TEST_REMOTE_CATALOG')
return v == "1"

def is_iceberg_test_fast_run():
v = os.environ.get('ICEBERG_TEST_FAST_RUN')
return v == "1"

def is_iceberg_rest_catalog():
v = os.environ.get('ICEBERG_TEST_CATALOG_TYPE')
return v == "rest"
Expand Down
38 changes: 25 additions & 13 deletions integration_tests/src/main/python/iceberg/iceberg_append_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

from asserts import assert_equal_with_local_sort, assert_gpu_fallback_collect, \
assert_gpu_fallback_write_sql
from conftest import is_iceberg_remote_catalog
from conftest import is_iceberg_remote_catalog, is_iceberg_test_fast_run
from data_gen import gen_df, copy_and_update
from iceberg import create_iceberg_table, \
iceberg_base_table_cols, iceberg_gens_list, get_full_table_name, \
Expand Down Expand Up @@ -97,7 +97,8 @@ def insert_data(spark, table_name):
@iceberg
@ignore_order(local=True)
@allow_non_gpu('AppendDataExec')
@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time")
@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(),
reason="Skip for remote catalog or fast Iceberg run to reduce test time")
@pytest.mark.parametrize("partition_table", [True, False], ids=lambda x: f"partition_table={x}")
def test_insert_into_unpartitioned_table_values(spark_tmp_table_factory,
partition_table):
Expand Down Expand Up @@ -137,7 +138,8 @@ def insert_data(spark, table_name: str):
@iceberg
@ignore_order(local=True)
@allow_non_gpu('LocalTableScanExec', 'ShuffleExchangeExec')
@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time")
@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(),
reason="Skip for remote catalog or fast Iceberg run to reduce test time")
@pytest.mark.parametrize("partition_table", [True, False], ids=lambda x: f"partition_table={x}")
def test_insert_into_table_values_aqe(spark_tmp_table_factory, partition_table):
"""Regression test for GPU V2 writes with AQE and a CPU VALUES input plan."""
Expand Down Expand Up @@ -175,7 +177,8 @@ def insert_data(spark, table_name: str):

@iceberg
@ignore_order(local=True)
@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time")
@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(),
reason="Skip for remote catalog or fast Iceberg run to reduce test time")
def test_insert_into_unpartitioned_table_all_cols(spark_tmp_table_factory):
table_prop = {"format-version": "2"}
cols = [f"_c{idx}" for idx, _ in enumerate(iceberg_full_gens_list)]
Expand Down Expand Up @@ -241,7 +244,8 @@ def test_insert_into_partitioned_table(spark_tmp_table_factory, partition_col_sq
@iceberg
@datagen_overrides(seed=0, reason='https://github.com/NVIDIA/spark-rapids-jni/issues/4016')
@ignore_order(local=True)
@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time")
@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(),
reason="Skip for remote catalog or fast Iceberg run to reduce test time")
@pytest.mark.parametrize("partition_col_sql", full_coverage_partition_transforms)
def test_insert_into_partitioned_table_full_coverage(spark_tmp_table_factory, partition_col_sql):
"""Partition-transform coverage anchor: this is the single test that exercises
Expand All @@ -254,7 +258,8 @@ def test_insert_into_partitioned_table_full_coverage(spark_tmp_table_factory, pa

@iceberg
@ignore_order(local=True)
@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time")
@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(),
reason="Skip for remote catalog or fast Iceberg run to reduce test time")
def test_insert_into_partitioned_table_all_cols(spark_tmp_table_factory):
table_prop = {"format-version": "2"}
cols = [f"_c{idx}" for idx, _ in enumerate(iceberg_full_gens_list)]
Expand Down Expand Up @@ -299,7 +304,8 @@ def insert_data(spark, table_name: str):
@iceberg
@ignore_order(local=True)
@allow_non_gpu('AppendDataExec', 'ShuffleExchangeExec', 'ProjectExec')
@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time")
@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(),
reason="Skip for remote catalog or fast Iceberg run to reduce test time")
@pytest.mark.parametrize("file_format", ["orc", "avro"], ids=lambda x: f"file_format={x}")
def test_insert_into_table_unsupported_file_format_fallback(
spark_tmp_table_factory, file_format):
Expand All @@ -322,7 +328,8 @@ def insert_data(spark, table_name: str):
@iceberg
@ignore_order(local=True)
@allow_non_gpu('AppendDataExec', 'ShuffleExchangeExec', 'ProjectExec')
@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time")
@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(),
reason="Skip for remote catalog or fast Iceberg run to reduce test time")
@pytest.mark.parametrize("partition_col_sql", [
pytest.param("bucket(4, contact.email)", id="bucket_nested_struct_field"),
pytest.param("truncate(3, contact.email)", id="truncate_nested_struct_field"),
Expand Down Expand Up @@ -358,7 +365,8 @@ def insert_data(spark):
@iceberg
@ignore_order(local=True)
@allow_non_gpu('AppendDataExec', 'ShuffleExchangeExec', 'ProjectExec')
@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time")
@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(),
reason="Skip for remote catalog or fast Iceberg run to reduce test time")
@pytest.mark.parametrize("conf_key", ["spark.rapids.sql.format.iceberg.enabled",
"spark.rapids.sql.format.iceberg.write.enabled"],
ids=lambda x: f"{x}=False")
Expand Down Expand Up @@ -428,7 +436,8 @@ def insert_data(spark, table_name: str):

@iceberg
@ignore_order(local=True)
@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time")
@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(),
reason="Skip for remote catalog or fast Iceberg run to reduce test time")
def test_insert_after_drop_partition_field(spark_tmp_table_factory):
"""Test INSERT on table after dropping a partition field (void transform).

Expand Down Expand Up @@ -485,7 +494,8 @@ def insert_data(spark, table_name):
@iceberg
@datagen_overrides(seed=0, reason='https://github.com/NVIDIA/spark-rapids-jni/issues/4016')
@ignore_order(local=True)
@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time")
@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(),
reason="Skip for remote catalog or fast Iceberg run to reduce test time")
def test_insert_into_partitioned_table_fanout_enabled(spark_tmp_table_factory):
# Use bucket(2, ...) to keep partition count low and avoid OOM from Iceberg's FanoutDataWriter.
_do_test_insert_into_partitioned_table(
Expand All @@ -500,7 +510,8 @@ def test_insert_into_partitioned_table_fanout_enabled(spark_tmp_table_factory):
# GPU would silently use Spark's session codec instead of Iceberg's zstd default.
@iceberg
@ignore_order(local=True)
@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time")
@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(),
reason="Skip for remote catalog or fast Iceberg run to reduce test time")
# Restricted to codecs whose footer metadata is reliable on small inputs. cuDF skips
# compression for tiny row groups (rapidsai/cudf#14017), so codecs like snappy can leave
# `UNCOMPRESSED` in the footer of small per-task files and make the assertion flaky;
Expand Down Expand Up @@ -549,7 +560,8 @@ def create_table(spark):
@iceberg
@ignore_order(local=True)
@allow_non_gpu('AppendDataExec', 'ShuffleExchangeExec', 'ProjectExec')
@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time")
@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(),
reason="Skip for remote catalog or fast Iceberg run to reduce test time")
@pytest.mark.parametrize("codec", ["gzip", "lz4"])
def test_insert_into_table_falls_back_on_unsupported_codec(spark_tmp_table_factory, codec):
table_name = get_full_table_name(spark_tmp_table_factory)
Expand Down
Loading
Loading