fix(io): add connection pool semaphore to Azure Blob Storage backend

[singularityDLW]

Changes Made

Add a connection pool semaphore to AzureBlobSource, matching the existing pattern in S3/GCS/TOS backends. Azure was the only major backend missing this, causing
unbounded concurrent connections (400-800+) when reading multiple large parquet files in parallel.

• Add max_connections_per_io_thread (default 8) to AzureConfig
• Add connection_pool_sema to AzureBlobSource with permit lifecycle tied to GetResult::Stream
• Extract get_size_internal() to avoid deadlock for GetRange::Suffix (Azure SDK doesn't support native suffix ranges)
• Update Python bindings, type stubs, SQL config

Related Issues

Closes #6279

[greptile-apps]

Greptile Summary

This PR adds connection pool management to the Azure Blob Storage backend, limiting concurrent connections to prevent resource exhaustion when reading multiple large files in parallel.

Key changes:

• Added connection_pool_sema semaphore to AzureBlobSource initialized with max_connections_per_io_thread (default 8) × IO pool threads
• Permits are acquired using acquire_owned() in get() and held for the lifetime of the stream via GetResult::Stream
• Introduced get_size_internal() helper method to avoid deadlock when handling GetRange::Suffix (which requires fetching file size while already holding a permit)
• Added max_connections parameter to Python bindings (AzureConfig) with proper type stubs and SQL config support
• Clamped max_connections_per_io_thread to minimum 1 to prevent deadlock when users pass 0

The implementation follows the established pattern from S3/GCS/TOS backends and includes comprehensive test coverage. This resolves the unbounded connection issue described in #6279.

Confidence Score: 5/5

• This PR is safe to merge with minimal risk
• The implementation correctly follows established patterns from other storage backends (S3/GCS/TOS) for connection pool management. The semaphore lifecycle is properly managed (permits acquired with acquire_owned and released when streams drop), the get_size_internal pattern correctly avoids potential deadlock, and all configuration layers are updated consistently. Comprehensive test coverage validates the new parameter including edge cases like max_connections=0.
• No files require special attention

Important Files Changed

Filename	Overview
src/daft-io/src/azure_blob.rs	Added connection pool semaphore with proper lifecycle management and get_size_internal to avoid deadlock
src/common/io-config/src/azure.rs	Added max_connections_per_io_thread field (default 8) with display formatting
src/common/io-config/src/python.rs	Added max_connections parameter to Python bindings with proper mapping to internal field

Sequence Diagram

sequenceDiagram
    participant Client
    participant AzureBlobSource
    participant Semaphore as Connection Pool<br/>Semaphore
    participant Azure as Azure Blob<br/>Storage

    Note over AzureBlobSource,Semaphore: Initialization (max 8 × num_threads)
    AzureBlobSource->>Semaphore: Create with capacity

    Note over Client,Azure: get() operation
    Client->>AzureBlobSource: get(uri, range)
    AzureBlobSource->>Semaphore: acquire_owned()
    Semaphore-->>AzureBlobSource: permit
    
    alt GetRange::Suffix
        AzureBlobSource->>AzureBlobSource: get_size_internal()<br/>(no semaphore)
        AzureBlobSource->>Azure: get_properties()
        Azure-->>AzureBlobSource: file size
        AzureBlobSource->>AzureBlobSource: calculate range
    end
    
    AzureBlobSource->>Azure: download with range
    Azure-->>AzureBlobSource: byte stream
    AzureBlobSource-->>Client: GetResult::Stream<br/>(with permit)
    
    Note over Client,Semaphore: Permit released when<br/>stream is dropped

    Note over Client,Azure: get_size() operation
    Client->>AzureBlobSource: get_size(uri)
    AzureBlobSource->>Semaphore: acquire()
    Semaphore-->>AzureBlobSource: permit guard
    AzureBlobSource->>AzureBlobSource: get_size_internal()
    AzureBlobSource->>Azure: get_properties()
    Azure-->>AzureBlobSource: file size
    AzureBlobSource-->>Client: size
    Note over AzureBlobSource,Semaphore: Permit released<br/>when guard drops

Loading

_{Last reviewed commit: 20583c3}

[greptile-apps]

_{6 files reviewed, 1 comment}

_{Edit Code Review Agent Settings | Greptile}

[universalmind303]

LGTM, thanks @singularityDLW!

[gsmit]

Great! Thanks a lot @singularityDLW