From 545cb661958803c4a218a8893d9bc02492dd3ae4 Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Thu, 24 Sep 2026 23:11:45 -0700 Subject: [PATCH 1/2] fix(io): pass S3 CA cert to PyArrow Signed-off-by: 1fanwang <1fannnw@gmail.com> --- pyiceberg/io/__init__.py | 1 + pyiceberg/io/pyarrow.py | 17 +++++++++++++++++ tests/io/test_pyarrow.py | 20 +++++++++++++++++++- 3 files changed, 37 insertions(+), 1 deletion(-) diff --git a/pyiceberg/io/__init__.py b/pyiceberg/io/__init__.py index c44e105e62..9cae219f7f 100644 --- a/pyiceberg/io/__init__.py +++ b/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" diff --git a/pyiceberg/io/pyarrow.py b/pyiceberg/io/pyarrow.py index 1e59107da3..a0f1896ba5 100644 --- a/pyiceberg/io/pyarrow.py +++ b/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,18 @@ 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): + from packaging import version + + min_pyarrow_version_supporting_tls_ca_file_path = "21.0.0" + if version.parse(pyarrow.__version__) < version.parse(min_pyarrow_version_supporting_tls_ca_file_path): + raise ImportError( + f"pyarrow version >= {min_pyarrow_version_supporting_tls_ca_file_path} required for " + f"S3FileSystem tls_ca_file_path support, but found version {pyarrow.__version__}." + ) + client_kwargs["tls_ca_file_path"] = tls_ca_file_path + def _initialize_oss_fs(self) -> FileSystem: from pyarrow.fs import S3FileSystem @@ -483,6 +496,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 +552,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: diff --git a/tests/io/test_pyarrow.py b/tests/io/test_pyarrow.py index 892d8e54eb..103e200e47 100644 --- a/tests/io/test_pyarrow.py +++ b/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.pyarrow.__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", From 45a932244f447281b79d162d4d8735fb29ecab96 Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Thu, 24 Sep 2026 23:26:02 -0700 Subject: [PATCH 2/2] fix(io): use PyArrow version alias Signed-off-by: 1fanwang <1fannnw@gmail.com> --- pyiceberg/io/pyarrow.py | 10 ++++------ tests/io/test_pyarrow.py | 2 +- 2 files changed, 5 insertions(+), 7 deletions(-) diff --git a/pyiceberg/io/pyarrow.py b/pyiceberg/io/pyarrow.py index a0f1896ba5..e7cc5c5926 100644 --- a/pyiceberg/io/pyarrow.py +++ b/pyiceberg/io/pyarrow.py @@ -456,13 +456,11 @@ def _initialize_fs(self, scheme: str, netloc: str | None = None) -> FileSystem: 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): - from packaging import version - - min_pyarrow_version_supporting_tls_ca_file_path = "21.0.0" - if version.parse(pyarrow.__version__) < version.parse(min_pyarrow_version_supporting_tls_ca_file_path): + 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( - f"pyarrow version >= {min_pyarrow_version_supporting_tls_ca_file_path} required for " - f"S3FileSystem tls_ca_file_path support, but found version {pyarrow.__version__}." + "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 diff --git a/tests/io/test_pyarrow.py b/tests/io/test_pyarrow.py index 103e200e47..9a303ea084 100644 --- a/tests/io/test_pyarrow.py +++ b/tests/io/test_pyarrow.py @@ -410,7 +410,7 @@ def test_pyarrow_s3_ssl_ca_cert_requires_supported_pyarrow_version() -> None: } with ( - patch("pyiceberg.io.pyarrow.pyarrow.__version__", "20.0.0"), + patch("pyiceberg.io.pyarrow.pa.__version__", "20.0.0"), patch("pyarrow.fs.S3FileSystem"), patch("pyarrow.fs.resolve_s3_region") as mock_s3_region_resolver, ):