[core] Introduce file resource management

[gavin9402]

Purpose

Introduce resource management capabilities to the REST Catalog, providing a unified way to manage file resources (FILE, JAR, PY, ARCHIVE) associated with databases. This lays the foundation for upcoming ML model and function features, where users will need to reference and manage external file resources such as model artifacts, UDF JARs, and Python scripts.

Changes

Resource Model (paimon-api)

• Resource interface and AbstractResource base class — define the resource abstraction with properties like name, type, description, URI, and custom properties
• FileResource, JarResource, PyResource, ArchiveResource — concrete resource types for FILE/JAR/PY/ARCHIVE
• ResourceType enum — four supported resource types
• ResourceChange — change operations for altering resources (setProperty, removeProperty, setDescription, setUri)
• ResourceDeserializer — Jackson deserializer for polymorphic resource deserialization

REST API (paimon-api)

• ResourcePaths — URL path builders for resource endpoints (/resources, /resource-details, /resources/{name})
• RESTApi — 8 new resource management API methods: listResources, listResourcesPaged, listResourceDetailsPaged, getResource, createResource, dropResource, alterResource, listResourcesPagedGlobally
• Request/Response classes: CreateResourceRequest, AlterResourceRequest, GetResourceResponse, ListResourcesResponse, ListResourceDetailsResponse, ListResourcesGloballyResponse

Catalog Interface (paimon-core)

• Catalog — 8 new interface methods for resource CRUD + ResourceAlreadyExistException and ResourceNotExistException inner exception classes
• AbstractCatalog — default UnsupportedOperationException implementations
• DelegateCatalog — delegation implementations
• RESTCatalog — full REST-backed implementations

Tests (paimon-core)

• RESTApiJsonTest — JSON serialization/deserialization tests for resource request/response classes
• RESTCatalogTest — integration tests for resource CRUD operations
• RESTCatalogServer — mock REST server with resource management route handlers
• MockRESTMessage — test helper methods for constructing resource test data

API Summary

Operation	Method	Endpoint
List resources	GET	/v1/{prefix}/databases/{db}/resources
List resource details	GET	/v1/{prefix}/databases/{db}/resource-details
Get resource	GET	/v1/{prefix}/databases/{db}/resources/{name}
Create resource	POST	/v1/{prefix}/databases/{db}/resources
Drop resource	DELETE	/v1/{prefix}/databases/{db}/resources/{name}
Alter resource	POST	/v1/{prefix}/databases/{db}/resources/{name}
List resources globally	GET	/v1/{prefix}/resources

Tests

mvn -pl paimon-core -am -Pfast-build -DfailIfNoTests=false -Dtest="RESTApiJsonTest,RESTCatalogTest" test

[TheR1sing3un]

It's a very surprising pr. Is there any relevant pip to provide more background information?

[gavin9402]

It's a very surprising pr. Is there any relevant pip to provide more background information?

Thank you for your suggestion. I will submit the PIP as soon as possible.

[JingsongLi]

First PR, I think you can focus on Resource introducing.

[gavin9402]

First PR, I think you can focus on Resource introducing.

Sure, let me revise it.

[JingsongLi]

How to use this URI? How to get rest token for this file?

[gavin9402]

The current design delegates URI handling to the file systems integrated with the engine. The metastore service is only responsible for permission management of the resource entity.

For example, in Daft, we assign the URI to the resources field of the PyFileResourceFunction instance.

    def _get_function(self, ident: Identifier) -> Function:
        ...
        paimon_daft_func: FunctionDefinition = self._inner.get_function(str(ident)).definitions()["daft"]
        ...

        # file_resources may be a list attribute or a callable method
        raw_resources = paimon_daft_func.file_resources
        resources = raw_resources() if callable(raw_resources) else raw_resources

        return PyFileResourceFunction(
            identifier=ident,
            module_name=paimon_daft_func.class_name,
            binding_name=paimon_daft_func.function_name,
            resources=[item.uri for item in resources],
        )

During execution, the engine resolves it by fetching the resources through the corresponding file system, and file permissions are also handled by the file system.

async def run_plan(
        self,
        plan: LocalPhysicalPlan,
        exec_cfg: PyDaftExecutionConfig,
        context: dict[str, str] | None,
        added_resources: dict[str, int] | None = None,
        **inputs: (
            Input | list[ray.ObjectRef]
        ),  # PyMicroPartitions are separated from Inputs because they are Ray ObjectRefs, which will be resolved by Ray.
    ) -> AsyncGenerator[MicroPartition | FlightPartitions | SwordfishTaskMetadata, None]:
        """Run a plan on swordfish and yield partitions."""
        if added_resources:
            file_resource_manager.resolve(added_resources)

More straightforwardly, we could also use Paimon FileIO for handling this. In fact, the engine’s behavior is similar to this.

[JingsongLi]

For our REST Catalog, the file system should be managed by Catalog for permissions, and here Resource feels that FileIO also needs to be exposed.

[gavin9402]

Your suggestion is absolutely right. I moved Resource to the paimon-core module and implemented the toBytes and newInputStream methods for it.

However, this introduces a small side effect: if Resource needs to be used in Function in the future, then Function would also need to be refactored into the paimon-core module.

[JingsongLi]

Why just using UriReaderFactory? This is not FileIO from Catalog.

[gavin9402]

Updated: using fileIOForData

[gavin9402]

Python implementation added

[JingsongLi]

This still breaks resource reads when the REST catalog has data-token.enabled=true. fileIOForData wraps the URI with RESTTokenFileIO, and RESTTokenFileIO.refreshToken() always calls loadTableToken(identifier). Here the identifier is the resource identifier, not a table identifier, so resource.toBytes() / newInputStream() will try to fetch /tables/<resource>/token and fail as soon as a token-enabled catalog is used. Please either use a non-table-token FileIO for resources or add a resource-token path instead of reusing the table-token flow.

[JingsongLi]

This has the same token-path problem as the Java implementation. When data-token.enabled is set, file_io_for_data returns a RESTTokenFileIO, whose refresh_token() calls load_table_token(table_identifier). The identifier passed here is a resource identifier, so reading the returned resource will request a table token for the resource name and fail in token-enabled REST catalogs. Please use a resource-aware token flow here, or avoid the table-token FileIO for resource URIs.

[JingsongLi]

We use loadTableToken to refresh FileIO, this is a resource instead of table. This is indeed a problem, I need to think about it.

[gavin9402]

We use loadTableToken to refresh FileIO, this is a resource instead of table. This is indeed a problem, I need to think about it.

Perhaps we should make the refreshToken method support identifiers of any type.