Skip to content

Commit a4dd4bf

Browse files
authored
[auto-merge] release/26.08 to main [skip ci] [bot] (NVIDIA#15398)
auto-merge triggered by github actions on `release/26.08` to create a PR keeping `main` up-to-date. If this PR is unable to be merged due to conflicts, it will remain open until manually fix.
2 parents f1566c2 + b701b32 commit a4dd4bf

5 files changed

Lines changed: 54 additions & 60 deletions

File tree

dist/unshimmed-common-from-single-shim.txt

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,6 @@ com/nvidia/spark/rapids/iceberg/spark/GpuSparkUtil$.class
3030
com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.class
3131
com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.class
3232
com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.class
33-
org/apache/iceberg/aws/s3/IcebergS3InputFileAccess.class
3433
org/apache/iceberg/data/GpuFileHelpers.class
3534
org/apache/iceberg/io/GpuClusteredWriterBridge.class
3635
org/apache/iceberg/io/GpuFanoutWriterBridge.class

iceberg/common/src/main/java/org/apache/iceberg/aws/s3/IcebergS3InputFile.java

Lines changed: 20 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -32,28 +32,30 @@
3232
import org.slf4j.LoggerFactory;
3333

3434
import java.io.IOException;
35-
import java.net.URI;
3635
import java.util.List;
3736
import java.util.OptionalLong;
3837

3938
/**
4039
* S3-backed {@link RapidsInputFile} that delegates byte-range reads to
4140
* {@link IcebergS3RangeCopier}. The supplied {@link FileIO} is only used for
4241
* its property map and any per-prefix storage-credential overlays.
43-
*
44-
* <p>The package-private S3 file access is isolated in {@link IcebergS3InputFileAccess}.
4542
*/
4643
public final class IcebergS3InputFile implements RapidsInputFile {
4744
private static final Logger LOG = LoggerFactory.getLogger(IcebergS3InputFile.class);
4845

4946
private final IcebergInputFile delegate;
50-
private final URI s3Uri;
47+
private final String s3Bucket;
48+
private final String s3Key;
5149
private final IcebergS3Client icebergS3Client;
5250

5351
private IcebergS3InputFile(
54-
IcebergInputFile delegate, URI s3Uri, IcebergS3Client icebergS3Client) {
52+
IcebergInputFile delegate,
53+
String s3Bucket,
54+
String s3Key,
55+
IcebergS3Client icebergS3Client) {
5556
this.delegate = delegate;
56-
this.s3Uri = s3Uri;
57+
this.s3Bucket = s3Bucket;
58+
this.s3Key = s3Key;
5759
this.icebergS3Client = icebergS3Client;
5860
}
5961

@@ -64,25 +66,27 @@ public static RapidsInputFile maybeCreate(InputFile inputFile, FileIO fileIO) {
6466
if (!RapidsInputFiles.isS3PerfEnabled()) {
6567
return delegate;
6668
}
67-
URI s3Uri = IcebergS3InputFileAccess.s3Uri(inputFile);
68-
if (s3Uri == null) {
69+
if (!(inputFile instanceof BaseS3File)) {
6970
return delegate;
7071
}
72+
S3URI s3Uri = ((BaseS3File) inputFile).uri();
73+
String s3Bucket = s3Uri.bucket();
74+
String s3Key = s3Uri.key();
7175
// Iceberg < 1.7 does not have SupportsStorageCredentials; ShimUtils returns
7276
// the per-prefix credential overlays (or an empty map on 1.6).
7377
IcebergS3Client icebergS3Client = IcebergS3RangeCopier.resolveClient(
74-
s3Uri.toString(),
78+
inputFile.location(),
7579
fileIO.properties(),
7680
ShimUtils.storageCredentialOverlays(fileIO));
7781
if (icebergS3Client == null) {
7882
if (TaskContext.get() != null) {
7983
GpuTaskMetrics$.MODULE$.get().recordPerfioS3IcebergFallback();
8084
}
81-
LOG.debug("IcebergS3RangeCopier path disabled for {}", s3Uri);
85+
LOG.debug("IcebergS3RangeCopier path disabled for {}", inputFile.location());
8286
return delegate;
8387
}
84-
LOG.debug("IcebergS3RangeCopier path active for {}", s3Uri);
85-
return new IcebergS3InputFile(delegate, s3Uri, icebergS3Client);
88+
LOG.debug("IcebergS3RangeCopier path active for {}", inputFile.location());
89+
return new IcebergS3InputFile(delegate, s3Bucket, s3Key, icebergS3Client);
8690
}
8791

8892
@Override
@@ -117,7 +121,8 @@ public InputFile getDelegate() {
117121
@Override
118122
public void readVectored(HostMemoryBuffer output, List<CopyRange> copyRanges)
119123
throws IOException {
120-
IcebergS3RangeCopier.copyToHMB(icebergS3Client, output, s3Uri, copyRanges);
124+
IcebergS3RangeCopier.copyToHMB(
125+
icebergS3Client, output, s3Bucket, s3Key, copyRanges);
121126
}
122127

123128
/**
@@ -133,6 +138,7 @@ public void readTail(long length, HostMemoryBuffer output) throws IOException {
133138
if (length < 0) {
134139
throw new IllegalArgumentException("length must be non-negative");
135140
}
136-
IcebergS3RangeCopier.copyTailToHMB(icebergS3Client, output, s3Uri, length, /*dstOffset*/ 0L);
141+
IcebergS3RangeCopier.copyTailToHMB(
142+
icebergS3Client, output, s3Bucket, s3Key, length, /*dstOffset*/ 0L);
137143
}
138144
}

iceberg/common/src/main/java/org/apache/iceberg/aws/s3/IcebergS3InputFileAccess.java

Lines changed: 0 additions & 41 deletions
This file was deleted.

integration_tests/src/main/python/iceberg/iceberg_test.py

Lines changed: 31 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@
1515
import pytest
1616

1717
from asserts import assert_equal_with_local_sort, assert_gpu_and_cpu_are_equal_collect, assert_gpu_and_cpu_row_counts_equal, assert_gpu_fallback_collect, assert_spark_exception
18-
from conftest import is_iceberg_remote_catalog
18+
from conftest import is_iceberg_remote_catalog, is_iceberg_rest_catalog
1919
from data_gen import *
2020
from iceberg import get_full_table_name, iceberg_unsupported_mark, _build_tblprops, \
2121
_BASE_TBLPROPS_SQL, create_iceberg_table
@@ -629,6 +629,36 @@ def setup_iceberg_table(spark):
629629
lambda spark: spark.sql("SELECT * FROM {}".format(table)),
630630
conf={'spark.rapids.sql.format.parquet.reader.type': reader_type})
631631

632+
@iceberg
633+
@ignore_order(local=True) # Iceberg plans with a thread pool and is not deterministic in file ordering
634+
@pytest.mark.parametrize('reader_type', rapids_reader_types)
635+
@pytest.mark.skipif(not is_iceberg_rest_catalog(),
636+
reason="S3 path handling is exercised only with the REST catalog")
637+
def test_iceberg_parquet_read_from_uri_invalid_s3_path(spark_tmp_table_factory, reader_type):
638+
table = get_full_table_name(spark_tmp_table_factory)
639+
tmp_view = spark_tmp_table_factory.get()
640+
641+
def setup_iceberg_table(spark):
642+
# A raw space is valid in an S3 object key but invalid in a URI. The RAPIDS reader must
643+
# retain Iceberg's original key for the S3 request instead of using a URI-encoded path.
644+
warehouse = spark.conf.get('spark.sql.catalog.spark_catalog.warehouse').rstrip('/')
645+
data_path = f'{warehouse}/{spark_tmp_table_factory.get()} uri invalid path/data'
646+
df = two_col_df(spark, long_gen, string_gen).sortWithinPartitions('b')
647+
df.createOrReplaceTempView(tmp_view)
648+
props = _build_tblprops({'write.data.path': data_path})
649+
props_sql = ", ".join(f"'{k}' = '{v}'" for k, v in props.items())
650+
spark.sql(f"CREATE TABLE {table} USING ICEBERG TBLPROPERTIES ({props_sql}) "
651+
f"AS SELECT * FROM {tmp_view}")
652+
653+
with_cpu_session(setup_iceberg_table)
654+
assert with_gpu_session(
655+
lambda spark:
656+
spark._jvm.com.nvidia.spark.rapids.fileio.RapidsInputFiles.isS3PerfEnabled()), \
657+
"PerfIO S3 must be enabled at Spark startup for REST catalog tests"
658+
assert_gpu_and_cpu_are_equal_collect(
659+
lambda spark: spark.sql(f"SELECT * FROM {table}"),
660+
conf={'spark.rapids.sql.format.parquet.reader.type': reader_type})
661+
632662
@iceberg
633663
@ignore_order(local=True) # Iceberg plans with a thread pool and is not deterministic in file ordering
634664
@pytest.mark.parametrize('reader_type', rapids_reader_types)

jenkins/spark-tests.sh

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -347,9 +347,8 @@ run_iceberg_tests() {
347347
echo "!!! Running iceberg tests with rest catalog"
348348
ICEBERG_REST_JARS="org.apache.iceberg:iceberg-spark-runtime-${ICEBERG_SPARK_VER}_${SCALA_BINARY_VER}:${ICEBERG_VERSION},\
349349
org.apache.iceberg:iceberg-aws-bundle:${ICEBERG_VERSION}"
350-
# filecache.enabled is a startup-only config, so it must be set here via
351-
# PYSP_TEST_ env var rather than as a session-level Spark config, because
352-
# FileCacheManager is initialized at executor startup time.
350+
# filecache.enabled and perfio.s3.enabled are startup-only configs, so they must
351+
# be set here via PYSP_TEST_ env vars rather than as session-level Spark configs.
353352
env \
354353
HOST_NAME=$PROJECT_REPO_HOST \
355354
EXPECTED_ICEBERG_VERSION=${ICEBERG_VERSION} \
@@ -358,6 +357,7 @@ org.apache.iceberg:iceberg-aws-bundle:${ICEBERG_VERSION}"
358357
PYSP_TEST_spark_driver_memory=1G \
359358
PYSP_TEST_spark_executor_memory=2G \
360359
PYSP_TEST_spark_rapids_filecache_enabled=true \
360+
PYSP_TEST_spark_rapids_perfio_s3_enabled=true \
361361
PYSP_TEST_spark_jars_packages="${ICEBERG_REST_JARS}" \
362362
PYSP_TEST_spark_jars_ivySettings="${WORKSPACE}/jenkins/ivysettings.xml" \
363363
PYSP_TEST_spark_sql_extensions="org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions" \

0 commit comments

Comments
 (0)