feat: Add support for multiplexed sessions

[rahul2393]

Thank you for opening a Pull Request! Before submitting your PR, there are a few things you can do to make sure it goes smoothly:

• Make sure to open an issue as a bug/issue before writing your code! That way we can discuss the change, evaluate designs, and agree on the general idea
• Ensure the tests and linter pass
• Code coverage does not decrease (if any source code was changed)
• Appropriate docs were updated (if necessary)

Fixes #<issue_number_goes_here> 🦕

[rahul2393]

graph TD
    A[Client Request] --> B{Multiplexed Sessions Enabled?}
    
    B -->|No| C[Regular Session Behavior]
    B -->|Yes| D[Multiplexed Session Behavior]
    
    %% Regular Session Path
    C --> C1[Read-Only Operations]
    C --> C2[Partitioned Operations]
    C --> C3[Read-Write Operations]
    
    C1 --> C1A[Get Session from Pool]
    C1A --> C1B[Create/Use Regular Session]
    C1B --> C1C[Execute Read Operation]
    C1C --> C1D[Return Session to Pool]
    
    C2 --> C2A[BatchSnapshot Creation]
    C2A --> C2B[Get Session from Pool<br/>TransactionType.PARTITIONED]
    C2B --> C2C[Create Regular Session]
    C2C --> C2D[Generate Partitions]
    C2D --> C2E[Each Partition Uses<br/>SAME Session]
    C2E --> C2F[Process All Partitions<br/>Sequentially]
    
    C3 --> C3A[Get Session from Pool]
    C3A --> C3B[Create/Use Regular Session]
    C3B --> C3C[Execute Transaction]
    C3C --> C3D[Return Session to Pool]
    
    %% Multiplexed Session Path
    D --> D1[Read-Only Operations]
    D --> D2[Partitioned Operations]
    D --> D3[Read-Write Operations]
    
    D1 --> D1A[Get Multiplexed Session]
    D1A --> D1B{Multiplexed Session Exists?}
    D1B -->|No| D1C[Create Multiplexed Session]
    D1B -->|Yes| D1D[Reuse Existing Session]
    D1C --> D1E[Start Maintenance Thread]
    D1E --> D1F[Execute Read Operation]
    D1D --> D1F
    D1F --> D1G[Session Remains Active<br/>No Return Needed]
    
    D2 --> D2A[BatchSnapshot Creation]
    D2A --> D2B[Get Multiplexed Session<br/>TransactionType.PARTITIONED]
    D2B --> D2C{Multiplexed Session Exists?}
    D2C -->|No| D2D[Create Multiplexed Session]
    D2C -->|Yes| D2E[Reuse Existing Session]
    D2D --> D2F[Generate Partitions]
    D2E --> D2F
    D2F --> D2G[Each Partition Uses<br/>SAME Multiplexed Session]
    D2G --> D2H[Process All Partitions<br/>Concurrently Possible]
    
    D3 --> D3A[Fallback to Regular Session]
    D3A --> D3B[Get Session from Pool]
    D3B --> D3C[Create/Use Regular Session]
    D3C --> D3D[Execute Transaction]
    D3D --> D3E[Return Session to Pool]
    
    %% Environment Variables
    E[Environment Variables] --> E1[GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS=true]
    E --> E2[GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS_PARTITIONED_OPS=true]
    E1 --> B
    E2 --> D2
    
    %% Key Differences
    F[Key Differences] --> F1[Session Reuse]
    F --> F2[Concurrency]
    F --> F3[Resource Usage]
    
    F1 --> F1A[Regular: New session per operation]
    F1 --> F1B[Multiplexed: Same session shared]
    
    F2 --> F2A[Regular: Sequential partition processing]
    F2 --> F2B[Multiplexed: Concurrent partition processing]
    
    F3 --> F3A[Regular: Multiple sessions = more resources]
    F3 --> F3B[Multiplexed: Single session = fewer resources]
    
    %% Styling
    classDef regularPath fill:#ffcccc,stroke:#ff0000,stroke-width:2px
    classDef multiplexedPath fill:#ccffcc,stroke:#00ff00,stroke-width:2px
    classDef envVar fill:#ffffcc,stroke:#ffaa00,stroke-width:2px
    classDef keyDiff fill:#ccccff,stroke:#0000ff,stroke-width:2px
    
    class C,C1,C2,C3,C1A,C1B,C1C,C1D,C2A,C2B,C2C,C2D,C2E,C2F,C3A,C3B,C3C,C3D regularPath
    class D,D1,D2,D3,D1A,D1B,D1C,D1D,D1E,D1F,D1G,D2A,D2B,D2C,D2D,D2E,D2F,D2G,D2H multiplexedPath
    class E,E1,E2 envVar
    class F,F1,F2,F3,F1A,F1B,F2A,F2B,F3A,F3B keyDiff

Loading

[currantw]

Tests seem to be failing here because the snapshot's session has does not exist. This means that Session.create() has not been called. Not sure exactly why that wouldn't be the case, but tough to find out without being able to run locally.

[rahul2393]

Issue was some tests were calling session.delete which prematurely deletes the multiplex session and hence "Session Not Found" error, Using Emulator calling session.delete() does not actually delete it from Emulator so issue does not occur using Emulator