Support special chars in S3URI

* Remove code that interpreted URI query and fragments parts according
  to RFC 3986. In practice, S3 locations do not encode special chars
  and, therefore, do not really have query and fragment parts.

* Add TestS3FileIOMinio for a small subset of tests using Minio as
  a realistic S3 protocol implementation for validating the handling
  of special chars.

Fixes apache#10279

aws/src/main/java/org/apache/iceberg/aws/s3/S3URI.java

@@ -34,8 +34,6 @@
 class S3URI {
   private static final String SCHEME_DELIM = "://";
   private static final String PATH_DELIM = "/";
-  private static final String QUERY_DELIM = "\\?";
-  private static final String FRAGMENT_DELIM = "#";
 
   private final String location;
   private final String scheme;
@@ -80,11 +78,12 @@ class S3URI {
             ? authoritySplit[0]
             : bucketToAccessPointMapping.getOrDefault(authoritySplit[0], authoritySplit[0]);
 
-    // Strip query and fragment if they exist
-    String path = authoritySplit.length > 1 ? authoritySplit[1] : "";
-    path = path.split(QUERY_DELIM, -1)[0];
-    path = path.split(FRAGMENT_DELIM, -1)[0];
-    this.key = path;
+    // Note: AWS UI (as an example) embeds special chars like `?` and `#` in `s3://` URIs without
+    // encoding. This technically does not match the URI spec per RFC 3986, but it is worth
+    // supporting to ensure interoperability for S3FileIO. Therefore, the whole sub-string after the
+    // "authority" part is treated as the "key" without interpreting "query" and "fragment" parts
+    // defined by the RFC.
+    this.key = authoritySplit.length > 1 ? authoritySplit[1] : "";
   }
 
   /** Returns S3 bucket name. */

aws/src/test/java/org/apache/iceberg/aws/s3/TestS3FileIOMinio.java

@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.aws.s3;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.io.InputStream;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.SeekableInputStream;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.*;
+import software.amazon.awssdk.utils.IoUtils;
+
+/** Specialized test cases that validate {@link S3FileIO} against the Minio S3 implementation. */
+public class TestS3FileIOMinio {
+  private static final Region REGION = Region.US_WEST_2;
+  private static final String BUCKET = "iceberg-s3-signer-test";
+  static final AwsCredentialsProvider CREDENTIALS_PROVIDER =
+      StaticCredentialsProvider.create(
+          AwsBasicCredentials.create("accessKeyId", "secretAccessKey"));
+  private static final MinioContainer MINIO_CONTAINER =
+      new MinioContainer(CREDENTIALS_PROVIDER.resolveCredentials());
+
+  private static S3Client s3;
+  private S3FileIO s3FileIO;
+
+  @BeforeAll
+  static void start() {
+    MINIO_CONTAINER.start();
+
+    s3 =
+        S3Client.builder()
+            .region(REGION)
+            .credentialsProvider(CREDENTIALS_PROVIDER)
+            .applyMutation(
+                s3ClientBuilder ->
+                    s3ClientBuilder.httpClientBuilder(
+                        software.amazon.awssdk.http.apache.ApacheHttpClient.builder()))
+            .endpointOverride(MINIO_CONTAINER.getURI())
+            .forcePathStyle(true) // OSX won't resolve subdomains
+            .build();
+
+    s3.createBucket(CreateBucketRequest.builder().bucket(BUCKET).build());
+  }
+
+  @AfterAll
+  static void stop() {
+    MINIO_CONTAINER.close();
+  }
+
+  @BeforeEach
+  public void setupFileIO() {
+    s3FileIO = new S3FileIO(() -> s3);
+  }
+
+  @ParameterizedTest
+  @ValueSource(
+      strings = {
+        "test",
+        "te_st",
+        "te st",
+        "te~!@$%^&*()-+st",
+        "te#st",
+        "te?st",
+      })
+  void testReadFileWithSpecialChars(String dir) throws IOException {
+    String key = dir + "/test_file";
+    // Make sure the characters in the key are supported by the S3 client and server.
+    s3.putObject(
+        PutObjectRequest.builder().bucket(BUCKET).key(key).build(), RequestBody.fromString("test"));
+    InputStream in = s3.getObject(GetObjectRequest.builder().bucket(BUCKET).key(key).build());
+    assertThat(IoUtils.toUtf8String(in)).isEqualTo("test");
+
+    // Note: Iceberg produces similar S3 location strings when a table is partitioned
+    // by a column with special characters in its name. Also, similar URIs are produced
+    // by the AWS S3 UI.
+    InputFile inputFile = s3FileIO.newInputFile("s3://" + BUCKET + "/" + key);
+    SeekableInputStream in2 = inputFile.newStream();
+    assertThat(IoUtils.toUtf8String(in2)).isEqualTo("test");
+  }
+}

aws/src/test/java/org/apache/iceberg/aws/s3/TestS3URI.java

@@ -66,12 +66,13 @@ public void testOnlyBucketNameLocation() {
   }
 
   @Test
-  public void testQueryAndFragment() {
+  public void testSpecialChars() {
     String p1 = "s3://bucket/path/to/file?query=foo#bar";
     S3URI uri1 = new S3URI(p1);
 
     Assertions.assertThat(uri1.bucket()).isEqualTo("bucket");
-    Assertions.assertThat(uri1.key()).isEqualTo("path/to/file");
+    // query and fragment parts are not interpreted
+    Assertions.assertThat(uri1.key()).isEqualTo("path/to/file?query=foo#bar");
     Assertions.assertThat(uri1.toString()).isEqualTo(p1);
   }
 
@@ -91,7 +92,7 @@ public void testS3URIWithBucketToAccessPointMapping() {
     S3URI uri1 = new S3URI(p1, bucketToAccessPointMapping);
 
     Assertions.assertThat(uri1.bucket()).isEqualTo("access-point");
-    Assertions.assertThat(uri1.key()).isEqualTo("path/to/file");
+    Assertions.assertThat(uri1.key()).isEqualTo("path/to/file?query=foo#bar");
     Assertions.assertThat(uri1.toString()).isEqualTo(p1);
   }
 }