[Data] Cherry-pick: Add enhanced support for Unity Catalog (#57954) (#58049)

## Description

This PR adds support for reading Unity Catalog Delta tables in Ray Data
with automatic credential vending. This enables secure, temporary access
to Delta Lake tables stored in Databricks Unity Catalog without
requiring users to manage cloud credentials manually.

### What's Added

- **`ray.data.read_unity_catalog()`** - Updated public API for reading
Unity Catalog Delta tables
- **`UnityCatalogConnector`** - Handles Unity Catalog REST API
integration and credential vending
- **Multi-cloud support** - Works with AWS S3, Azure Data Lake Storage,
and Google Cloud Storage
- **Automatic credential management** - Obtains temporary,
least-privilege credentials via Unity Catalog API
- **Delta Lake integration** - Properly configures PyArrow filesystem
for Delta tables with session tokens

### Key Features

✅ **Production-ready credential vending API** - Uses stable, public
Unity Catalog APIs
✅ **Secure by default** - Temporary credentials with automatic cleanup ✅
**Multi-cloud** - AWS (S3), Azure (Blob Storage), and GCP (Cloud
Storage)
✅ **Delta Lake optimized** - Handles session tokens and PyArrow
filesystem configuration
✅ **Comprehensive error handling** - Helpful messages for common issues
(deletion vectors, permissions, etc.)
✅ **Full logging support** - Debug and info logging throughout

### Usage Example

```python
import ray

# Read a Unity Catalog Delta table
ds = ray.data.read_unity_catalog(
    table="main.sales.transactions",
    url="https://dbc-XXXXXXX-XXXX.cloud.databricks.com",
    token="dapi...",
    region="us-west-2"  # Optional, for AWS
)

# Use standard Ray Data operations
ds = ds.filter(lambda row: row["amount"] > 100)
ds.show(5)
```

### Implementation Notes

This is a **simplified, focused implementation** that:
- Supports **Unity Catalog tables only** (no volumes - that's in private
preview)
- Assumes **Delta Lake format** (most common Unity Catalog use case)
- Uses **production-ready APIs** only (no private preview features)
- Provides ~600 lines of clean, reviewable code

The full implementation with volumes and multi-format support is
available in the `data_uc_volumes` branch and can be added in a future
PR once this foundation is reviewed.

### Testing

- ✅ All ruff lint checks pass
- ✅ Code formatted per Ray standards
- ✅ Tested with real Unity Catalog Delta tables on AWS S3
- ✅ Proper PyArrow filesystem configuration verified
- ✅ Credential vending flow validated

## Related issues

Related to Unity Catalog and Delta Lake support requests in Ray Data.

## Additional information

### Architecture

The implementation follows the **connector pattern** rather than a
`Datasource` subclass because Unity Catalog is a metadata/credential
layer, not a data format. The connector:

1. Fetches table metadata from Unity Catalog REST API
2. Obtains temporary credentials via credential vending API
3. Configures cloud-specific environment variables
4. Delegates to `ray.data.read_delta()` with proper filesystem
configuration

### Delta Lake Special Handling

Delta Lake on AWS requires explicit PyArrow S3FileSystem configuration
with session tokens (environment variables alone are insufficient). This
implementation correctly creates and passes the filesystem object to the
`deltalake` library.

### Cloud Provider Support

| Provider | Credential Type | Implementation |
|----------|----------------|----------------|
| AWS S3 | Temporary IAM credentials | PyArrow S3FileSystem with session
token |
| Azure Blob | SAS tokens | Environment variables
(AZURE_STORAGE_SAS_TOKEN) |
| GCP Cloud Storage | OAuth tokens / Service account | Environment
variables (GCP_OAUTH_TOKEN, GOOGLE_APPLICATION_CREDENTIALS) |

### Error Handling

Comprehensive error messages for common issues:
- **Deletion Vectors**: Guidance on upgrading deltalake library or
disabling the feature
- **Column Mapping**: Compatibility information and solutions
- **Permissions**: Clear list of required Unity Catalog permissions
- **Credential issues**: Detailed troubleshooting steps

### Future Enhancements

Potential follow-up PRs:
- Unity Catalog volumes support (when out of private preview)
- Multi-format support (Parquet, CSV, JSON, images, etc.)
- Custom datasource integration
- Advanced Delta Lake features (time travel, partition filters)

### Dependencies

- Requires `deltalake` package for Delta Lake support
- Uses standard Ray Data APIs (`read_delta`, `read_datasource`)
- Integrates with existing PyArrow filesystem infrastructure

### Documentation

- Full docstrings with examples
- Type hints throughout
- Inline comments with references to external documentation
- Comprehensive error messages with actionable guidance

---------

> Thank you for contributing to Ray! 🚀
> Please review the [Ray Contribution
Guide](https://docs.ray.io/en/master/ray-contribute/getting-involved.html)
before opening a pull request.

> ⚠️ Remove these instructions before submitting your PR.

> 💡 Tip: Mark as draft if you want early feedback, or ready for review
when it's complete.

## Description
> Briefly describe what this PR accomplishes and why it's needed.

## Related issues
> Link related issues: "Fixes #1234", "Closes #1234", or "Related to
#1234".

## Additional information
> Optional: Add implementation details, API changes, usage examples,
screenshots, etc.

Signed-off-by: soffer-anyscale <stephen.offer@anyscale.com>
Co-authored-by: soffer-anyscale <173827098+soffer-anyscale@users.noreply.github.com>

Author: gvspraveen

python/ray/data/_internal/datasource/uc_datasource.py

@@ -1,3 +1,4 @@
+import atexit
 import os
 import tempfile
 from typing import Any, Callable, Dict, Optional
@@ -42,6 +43,7 @@ def __init__(
         self.operation = operation
         self.ray_init_kwargs = ray_init_kwargs or {}
         self.reader_kwargs = reader_kwargs or {}
+        self._gcp_temp_file = None
 
     def _get_table_info(self) -> dict:
         url = f"{self.base_url}/api/2.1/unity-catalog/tables/{self.table_full_name}"
@@ -81,12 +83,17 @@ def _set_env(self):
             env_vars["AZURE_STORAGE_SAS_TOKEN"] = creds["azuresasuri"]
         elif "gcp_service_account" in creds:
             gcp_json = creds["gcp_service_account"]
-            with tempfile.NamedTemporaryFile(
-                prefix="gcp_sa_", suffix=".json", delete=True
-            ) as temp_file:
-                temp_file.write(gcp_json.encode())
-                temp_file.flush()
-                env_vars["GOOGLE_APPLICATION_CREDENTIALS"] = temp_file.name
+            temp_file = tempfile.NamedTemporaryFile(
+                mode="w",
+                prefix="gcp_sa_",
+                suffix=".json",
+                delete=False,
+            )
+            temp_file.write(gcp_json)
+            temp_file.close()
+            env_vars["GOOGLE_APPLICATION_CREDENTIALS"] = temp_file.name
+            self._gcp_temp_file = temp_file.name
+            atexit.register(self._cleanup_gcp_temp_file, temp_file.name)
         else:
             raise ValueError(
                 "No known credential type found in Databricks UC response."
@@ -96,6 +103,15 @@ def _set_env(self):
             os.environ[k] = v
         self._runtime_env = {"env_vars": env_vars}
 
+    @staticmethod
+    def _cleanup_gcp_temp_file(temp_file_path: str):
+        """Clean up temporary GCP service account file."""
+        if temp_file_path and os.path.exists(temp_file_path):
+            try:
+                os.unlink(temp_file_path)
+            except OSError:
+                pass
+
     def _infer_data_format(self) -> str:
         if self.data_format:
             return self.data_format
@@ -121,17 +137,59 @@ def _get_ray_reader(self, data_format: str) -> Callable[..., Any]:
                 return reader_func
         raise ValueError(f"Unsupported data format: {fmt}")
 
+    def _read_delta_with_credentials(self):
+        """Read Delta table with proper PyArrow filesystem for session tokens."""
+        import pyarrow.fs as pafs
+
+        creds = self._creds_response
+        reader_kwargs = self.reader_kwargs.copy()
+
+        # For AWS, create PyArrow S3FileSystem with session tokens
+        if "aws_temp_credentials" in creds:
+            if not self.region:
+                raise ValueError(
+                    "The 'region' parameter is required for AWS S3 access. "
+                    "Please specify the AWS region (e.g., region='us-west-2')."
+                )
+            aws = creds["aws_temp_credentials"]
+            filesystem = pafs.S3FileSystem(
+                access_key=aws["access_key_id"],
+                secret_key=aws["secret_access_key"],
+                session_token=aws["session_token"],
+                region=self.region,
+            )
+            reader_kwargs["filesystem"] = filesystem
+
+        # Call ray.data.read_delta with proper error handling
+        try:
+            return ray.data.read_delta(self._table_url, **reader_kwargs)
+        except Exception as e:
+            error_msg = str(e)
+            if (
+                "DeletionVectors" in error_msg
+                or "Unsupported reader features" in error_msg
+            ):
+                raise RuntimeError(
+                    f"Delta table uses Deletion Vectors, which requires deltalake>=0.10.0. "
+                    f"Error: {error_msg}\n"
+                    f"Solution: pip install --upgrade 'deltalake>=0.10.0'"
+                ) from e
+            raise
+
     def read(self):
         self._get_table_info()
         self._get_creds()
         self._set_env()
 
         data_format = self._infer_data_format()
-        reader = self._get_ray_reader(data_format)
 
         if not ray.is_initialized():
             ray.init(runtime_env=self._runtime_env, **self.ray_init_kwargs)
 
-        url = self._table_url
-        ds = reader(url, **self.reader_kwargs)
-        return ds
+        # Use special Delta reader for proper filesystem handling
+        if data_format == "delta":
+            return self._read_delta_with_credentials()
+
+        # Use standard reader for other formats
+        reader = self._get_ray_reader(data_format)
+        return reader(self._table_url, **self.reader_kwargs)