sequenceDiagram
    actor UserQuery
    participant PrestoCoordinator
    participant DeltaSplitManager
    participant DeltaExpressionUtils
    participant HadoopFileSystem

    UserQuery->>PrestoCoordinator: submit SQL query on Delta table
    PrestoCoordinator->>DeltaSplitManager: getNextBatch(partitionHandle, table, splitCount)
    DeltaSplitManager->>DeltaSplitManager: read addFileStatus.getPath()
    DeltaSplitManager->>DeltaSplitManager: filePath = URI.create(path).toString()
    DeltaSplitManager->>HadoopFileSystem: open(filePath s3a_uri)
    HadoopFileSystem-->>DeltaSplitManager: input stream to Parquet file

    PrestoCoordinator->>DeltaExpressionUtils: evaluatePartitionPredicate(row, partitionPredicate)
    DeltaExpressionUtils->>DeltaExpressionUtils: addFileStatus = InternalScanFileUtils.getAddFileStatus(row)
    DeltaExpressionUtils->>DeltaExpressionUtils: filePath = URI.create(addFileStatus.getPath()).toString()
    DeltaExpressionUtils-->>PrestoCoordinator: domain for partition pruning using full s3a_uri

    PrestoCoordinator-->>UserQuery: return query results without FileNotFoundException

classDiagram
    class DeltaSplitManager {
        +CompletableFuture~ConnectorSplitBatch~ getNextBatch(ConnectorPartitionHandle partitionHandle, ConnectorTableLayoutHandle layoutHandle, List~ConnectorSplit~ splits, int maxSize)
        -ConnectorId connectorId
        -DeltaTable deltaTable
        -DeltaMetadata deltaMetadata
    }

    class DeltaExpressionUtils {
        <<utility>>
        -static boolean evaluatePartitionPredicate(DeltaColumnHandle partitionColumn, TupleDomain~DeltaColumnHandle~ partitionPredicate, TypeManager typeManager, Object row)
    }

    class InternalScanFileUtils {
        <<utility>>
        +Map~String,String~ getPartitionValues(Object row)
        +AddFileStatus getAddFileStatus(Object row)
    }

    class AddFileStatus {
        +String getPath()
        +long getSize()
    }

    class Domain {
        +static Domain getDomain(DeltaColumnHandle column, String partitionValue, TypeManager typeManager, String filePath)
    }

    class DeltaColumnHandle {
        +String getName()
    }

    class TypeManager
    class ConnectorSplitBatch
    class ConnectorSplit
    class ConnectorPartitionHandle
    class ConnectorTableLayoutHandle
    class ConnectorId
    class DeltaTable {
        +String getSchemaName()
        +String getTableName()
    }
    class DeltaMetadata
    class String

    DeltaSplitManager --> DeltaTable : uses
    DeltaSplitManager --> DeltaMetadata : uses
    DeltaSplitManager --> AddFileStatus : uses getPath and getSize
    DeltaSplitManager --> ConnectorSplitBatch : returns
    DeltaSplitManager --> ConnectorSplit : creates
    DeltaSplitManager --> ConnectorId : uses

    DeltaExpressionUtils --> InternalScanFileUtils : uses
    DeltaExpressionUtils --> AddFileStatus : uses getPath
    DeltaExpressionUtils --> DeltaColumnHandle : uses
    DeltaExpressionUtils --> Domain : computes
    DeltaExpressionUtils --> TypeManager : uses

    InternalScanFileUtils --> AddFileStatus : returns

    AddFileStatus --> String : path preserved as full s3a_uri
    Domain --> String : filePath parameter is full s3a_uri
    DeltaSplitManager --> String : filePath parameter is full s3a_uri for splits
    DeltaExpressionUtils --> String : filePath parameter is full s3a_uri for domains

flowchart LR
    A["Delta addFileStatus.getPath() returns s3a://bucket/path/file.parquet"] --> B{Old vs new handling}

    B -->|Old behavior| C["URI.create(path).getPath()"]
    C --> D["Produces /path/file.parquet (scheme and bucket stripped)"]
    D --> E["Hadoop LocalFileSystem attempts to open /path/file.parquet"]
    E --> F["FileNotFoundException: File does not exist"]

    B -->|New behavior| G["URI.create(path).toString()"]
    G --> H["Preserves s3a://bucket/path/file.parquet"]
    H --> I["S3 filesystem opens s3a://bucket/path/file.parquet"]
    I --> J["Presto reads Parquet file successfully"]