fix(io): pass S3 CA cert to PyArrow

pyiceberg/io/__init__.py

@@ -69,6 +69,7 @@ def _is_local_path(path: str) -> bool:
 S3_SESSION_TOKEN = "s3.session-token"
 S3_REGION = "s3.region"
 S3_RESOLVE_REGION = "s3.resolve-region"
+S3_SSL_CA_CERT = "s3.ssl.ca-cert"
 S3_PROXY_URI = "s3.proxy-uri"
 S3_CONNECT_TIMEOUT = "s3.connect-timeout"
 S3_REQUEST_TIMEOUT = "s3.request-timeout"

pyiceberg/io/pyarrow.py

@@ -116,6 +116,7 @@
     S3_ROLE_SESSION_NAME,
     S3_SECRET_ACCESS_KEY,
     S3_SESSION_TOKEN,
+    S3_SSL_CA_CERT,
     FileIO,
     InputFile,
     InputStream,
@@ -453,6 +454,16 @@ def _initialize_fs(self, scheme: str, netloc: str | None = None) -> FileSystem:
         else:
             raise ValueError(f"Unrecognized filesystem type in URI: {scheme}")
 
+    def _set_tls_ca_file_path(self, client_kwargs: dict[str, Any]) -> None:
+        if tls_ca_file_path := self.properties.get(S3_SSL_CA_CERT):
+            min_pyarrow_version_supporting_tls_ca_file_path = (21, 0)
+            if tuple(map(int, pa.__version__.split(".")[:2])) < min_pyarrow_version_supporting_tls_ca_file_path:
+                raise ImportError(
+                    "pyarrow version >= 21.0.0 required for S3FileSystem tls_ca_file_path support, "
+                    f"but found version {pa.__version__}."
+                )
+            client_kwargs["tls_ca_file_path"] = tls_ca_file_path
+
     def _initialize_oss_fs(self) -> FileSystem:
         from pyarrow.fs import S3FileSystem
 
@@ -483,6 +494,8 @@ def _initialize_oss_fs(self) -> FileSystem:
         if s3_anonymous := self.properties.get(S3_ANONYMOUS):
             client_kwargs["anonymous"] = strtobool(s3_anonymous)
 
+        self._set_tls_ca_file_path(client_kwargs)
+
         return S3FileSystem(**client_kwargs)
 
     def _initialize_s3_fs(self, netloc: str | None) -> FileSystem:
@@ -537,6 +550,8 @@ def _initialize_s3_fs(self, netloc: str | None) -> FileSystem:
         if s3_anonymous := self.properties.get(S3_ANONYMOUS):
             client_kwargs["anonymous"] = strtobool(s3_anonymous)
 
+        self._set_tls_ca_file_path(client_kwargs)
+
         return S3FileSystem(**client_kwargs)
 
     def _initialize_azure_fs(self) -> FileSystem:

tests/io/test_pyarrow.py

@@ -63,7 +63,7 @@
     Or,
 )
 from pyiceberg.expressions.literals import literal
-from pyiceberg.io import S3_RETRY_STRATEGY_IMPL, InputStream, OutputStream, load_file_io
+from pyiceberg.io import S3_RETRY_STRATEGY_IMPL, S3_SSL_CA_CERT, InputStream, OutputStream, load_file_io
 from pyiceberg.io.pyarrow import (
     ICEBERG_SCHEMA,
     PYARROW_PARQUET_FIELD_ID_KEY,
@@ -382,6 +382,7 @@ def test_pyarrow_s3_session_properties() -> None:
         "s3.secret-access-key": "password",
         "s3.region": "us-east-1",
         "s3.session-token": "s3.session-token",
+        S3_SSL_CA_CERT: "/path/to/ca.pem",
         **UNIFIED_AWS_SESSION_PROPERTIES,
     }
 
@@ -399,9 +400,26 @@ def test_pyarrow_s3_session_properties() -> None:
             secret_key="password",
             region="us-east-1",
             session_token="s3.session-token",
+            tls_ca_file_path="/path/to/ca.pem",
         )
 
 
+def test_pyarrow_s3_ssl_ca_cert_requires_supported_pyarrow_version() -> None:
+    session_properties: Properties = {
+        S3_SSL_CA_CERT: "/path/to/ca.pem",
+    }
+
+    with (
+        patch("pyiceberg.io.pyarrow.pa.__version__", "20.0.0"),
+        patch("pyarrow.fs.S3FileSystem"),
+        patch("pyarrow.fs.resolve_s3_region") as mock_s3_region_resolver,
+    ):
+        mock_s3_region_resolver.side_effect = OSError("S3 bucket is not found")
+
+        with pytest.raises(ImportError, match="pyarrow version >= 21.0.0 required"):
+            PyArrowFileIO(properties=session_properties).new_input(location=f"s3://warehouse/{uuid.uuid4()}")
+
+
 def test_pyarrow_s3_session_properties_with_anonymous() -> None:
     session_properties: Properties = {
         "s3.anonymous": "true",