feat(iceberg): auto-configure IOConfig for Alibaba Cloud OSS (#6993)

## Changes Made
OSS is S3-compatible; it just needs `force_virtual_addressing` plus an
`oss`→`s3`protocol alias.
`_convert_iceberg_file_io_properties_to_io_config` now applies these
when the table `location` is `oss://`, routing it through the existing
S3 path — same approach as PyIceberg's `_initialize_oss_fs`. No new
backend. Works even with no table properties (env-var creds). Pass an
explicit `io_config` to opt out. Scoped to the Iceberg connector.

Author: plusplusjiajia

daft/dataframe/dataframe.py

@@ -1341,7 +1341,9 @@ def write_iceberg(
                     )
 
         io_config = (
-            _convert_iceberg_file_io_properties_to_io_config(table.io.properties) if io_config is None else io_config
+            _convert_iceberg_file_io_properties_to_io_config(table.io.properties, table.location())
+            if io_config is None
+            else io_config
         )
         io_config = get_context().daft_planning_config.default_io_config if io_config is None else io_config
 

daft/io/iceberg/_iceberg.py

@@ -1,12 +1,14 @@
 # ruff: noqa: I002
 # isort: dont-add-import: from __future__ import annotations
 
+import logging
 from typing import TYPE_CHECKING, Any, Union
 
 from daft import context, runners
 from daft.api_annotations import PublicAPI
 from daft.daft import IOConfig, ScanOperatorHandle, StorageConfig
 from daft.dataframe import DataFrame
+from daft.filesystem import get_protocol_from_path
 from daft.io._checkpoint import attach_checkpoint
 from daft.logical.builder import LogicalPlanBuilder
 
@@ -16,8 +18,18 @@
     from daft.checkpoint import CheckpointConfig
 
 
-def _convert_iceberg_file_io_properties_to_io_config(props: dict[str, Any]) -> IOConfig | None:
-    """Property keys defined here: https://github.com/apache/iceberg-python/blob/main/pyiceberg/io/__init__.py."""
+logger = logging.getLogger(__name__)
+
+
+def _convert_iceberg_file_io_properties_to_io_config(
+    props: dict[str, Any], location: str | None = None
+) -> IOConfig | None:
+    """Property keys defined here: https://github.com/apache/iceberg-python/blob/main/pyiceberg/io/__init__.py.
+
+    For an ``oss://`` ``location`` (Alibaba Cloud OSS, S3-compatible), the IOConfig gets
+    virtual-hosted addressing and an ``oss``->``s3`` alias so the S3 filesystem resolves
+    ``oss://`` paths -- applied even with no IO properties (e.g. env-var credentials).
+    """
     from daft.io import AzureConfig, GCSConfig, IOConfig, S3Config
 
     any_props_set = False
@@ -30,13 +42,16 @@ def get_first_property_value(*property_names: str) -> Any | None:
                 return property_value
         return None
 
+    is_oss = location is not None and get_protocol_from_path(location) == "oss"
+
     io_config = IOConfig(
         s3=S3Config(
             endpoint_url=get_first_property_value("s3.endpoint"),
             region_name=get_first_property_value("s3.region", "client.region"),
             key_id=get_first_property_value("s3.access-key-id", "client.access-key-id"),
             access_key=get_first_property_value("s3.secret-access-key", "client.secret-access-key"),
             session_token=get_first_property_value("s3.session-token", "client.session-token"),
+            force_virtual_addressing=True if is_oss else None,
         ),
         azure=AzureConfig(
             storage_account=get_first_property_value("adls.account-name", "adlfs.account-name"),
@@ -50,8 +65,12 @@ def get_first_property_value(*property_names: str) -> Any | None:
             project_id=get_first_property_value("gcs.project-id"),
             token=get_first_property_value("gcs.oauth2.token"),
         ),
+        protocol_aliases={"oss": "s3"} if is_oss else None,
     )
 
+    if is_oss:
+        logger.debug("oss:// table detected; applying S3-compatible settings to the IOConfig")
+        return io_config
     return io_config if any_props_set else None
 
 
@@ -107,7 +126,9 @@ def read_iceberg(
         table = StaticTable.from_metadata(metadata_location=table)
 
     io_config = (
-        _convert_iceberg_file_io_properties_to_io_config(table.io.properties) if io_config is None else io_config
+        _convert_iceberg_file_io_properties_to_io_config(table.io.properties, table.location())
+        if io_config is None
+        else io_config
     )
     io_config = context.get_context().daft_planning_config.default_io_config if io_config is None else io_config
 

tests/io/iceberg/test_iceberg_io_config.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+from daft.io.iceberg._iceberg import _convert_iceberg_file_io_properties_to_io_config
+
+
+def test_oss_location_applies_settings():
+    """An oss:// table location enables virtual-hosted addressing and the oss->s3 alias."""
+    props = {"s3.endpoint": "http://oss-cn-hangzhou.aliyuncs.com", "s3.access-key-id": "ak"}
+    result = _convert_iceberg_file_io_properties_to_io_config(props, "oss://my-bucket/warehouse/db/table")
+    assert result is not None
+    assert result.s3.force_virtual_addressing is True
+    assert result.protocol_aliases == {"oss": "s3"}
+    # Table properties are still applied.
+    assert result.s3.endpoint_url == "http://oss-cn-hangzhou.aliyuncs.com"
+    assert result.s3.key_id == "ak"
+
+
+def test_oss_location_no_props():
+    """An oss:// table with no IO properties still gets the OSS settings.
+
+    Covers credentials supplied via the environment rather than table properties.
+    """
+    result = _convert_iceberg_file_io_properties_to_io_config({}, "oss://my-bucket/warehouse/db/table")
+    assert result is not None
+    assert result.s3.force_virtual_addressing is True
+    assert result.protocol_aliases == {"oss": "s3"}
+
+
+def test_non_oss_location_with_props():
+    """A non-oss:// location with properties leaves the OSS settings off."""
+    props = {"s3.endpoint": "https://s3.us-west-2.amazonaws.com"}
+    result = _convert_iceberg_file_io_properties_to_io_config(props, "s3://my-bucket/warehouse/db/table")
+    assert result is not None
+    assert result.s3.force_virtual_addressing is False
+    assert result.protocol_aliases == {}
+    assert result.s3.endpoint_url == "https://s3.us-west-2.amazonaws.com"
+
+
+def test_non_oss_location_no_props_returns_none():
+    """A non-oss:// location with no properties yields no IOConfig."""
+    assert _convert_iceberg_file_io_properties_to_io_config({}, "s3://my-bucket/warehouse/db/table") is None
+
+
+def test_no_location_with_props():
+    """With no location (default), properties still convert and OSS settings stay off."""
+    props = {"s3.endpoint": "https://s3.us-west-2.amazonaws.com"}
+    result = _convert_iceberg_file_io_properties_to_io_config(props)
+    assert result is not None
+    assert result.s3.force_virtual_addressing is False
+    assert result.protocol_aliases == {}
+    assert result.s3.endpoint_url == "https://s3.us-west-2.amazonaws.com"
+
+
+def test_no_location_no_props_returns_none():
+    """With no location (default) and no properties, no IOConfig is produced."""
+    assert _convert_iceberg_file_io_properties_to_io_config({}) is None