Add URI generator, resolver and corresponding document open & create functions

[bobbai00]

This PR refactor the storage key by representing it using the storage URI. And this PR also adds the openDocument and createDocument function using the given URI.

Major Changes

1. VFS URI resource type definition, resolve and decode functions

Two types of VFS resources are defined:

object VFSResourceType extends Enumeration {
  val RESULT: Value = Value("result")
  val MATERIALIZED_RESULT: Value = Value("materializedResult")
}

Two defs are added to the FileResolver

• resolve: create the URI pointing to the storage resource on the VFS

/**
    * Resolve a VFS resource to its URI. The URI can be used by the DocumentFactory to create resource or open resource
    *
    * @param resourceType   The type of the VFS resource.
    * @param workflowId     Workflow identifier.
    * @param executionId    Execution identifier.
    * @param operatorId     Operator identifier.
    * @param portIdentity   Optional port identifier. **Required** if `resourceType` is `RESULT` or `MATERIALIZED_RESULT`.
    * @return A VFS URI
    * @throws IllegalArgumentException if `resourceType` is `RESULT` but `portIdentity` is missing.
    */
  def resolve(
      resourceType: VFSResourceType.Value,
      workflowId: WorkflowIdentity,
      executionId: ExecutionIdentity,
      operatorId: OperatorIdentity,
      portIdentity: Option[PortIdentity] = None
  ): URI

• decodeVFSUri: decode a VFS URI to components

  /**
    * Parses a VFS URI and extracts its components
    *
    * @param uri The VFS URI to parse.
    * @return A `VFSUriComponents` object with the extracted data.
    * @throws IllegalArgumentException if the URI is malformed.
    */
  def decodeVFSUri(uri: URI): (
      WorkflowIdentity,
      ExecutionIdentity,
      OperatorIdentity,
      Option[PortIdentity],
      VFSResourceType.Value
  )

2. createDocument and openDocument functions to the DocumentFactory

createDocument and openDocument defs to create/open a storage resource pointed by the URI

• DocumentFactory.createDocument

  /**
    * Create a document for storage specified by the uri.
    * This document is suitable for storing structural data, i.e. the schema is required to create such document.
    * @param uri the location of the document
    * @param schema the schema of the data stored in the document
    * @return the created document
    */
  def createDocument(uri: URI, schema: Schema): VirtualDocument[_]

• DocumentFactory.openDocument

  /**
    * Open a document specified by the uri.
    * The document should be storing the structural data as the document and the schema will be returned
    * @param uri the uri of the document
    * @return the VirtualDocument, which is the handler of the data; the Schema, which is the schema of the data stored in the document
    */
  def openDocument(uri: URI): (VirtualDocument[_], Schema)

[shengquan-ni]

left minor comments.

[Yicong-Huang]

Left other comments offline.

[Yicong-Huang]

I suggest you use multiple explicit APIs instead:
createPortResultURI(workflowId, eId, opId, portId)
createOperatorLogURI(workflowId, eid, opId)

[Yicong-Huang]

don't we need to remove some URI?

[Yicong-Huang]

Physical plan should have context already, which contains the Eid?