[python] Add experimental asynchronous table prediction submission

docs/static/rest-catalog-open-api.yaml

@@ -28,6 +28,71 @@ servers:
   - url: http://localhost:8080
     description: Server URL in Development environment
 paths:
+  /v1/{prefix}/databases/{database}/tables/{table}/predict:
+    post:
+      tags: [table]
+      summary: Submit an asynchronous prediction (experimental extension)
+      operationId: predictTable
+      description: >-
+        A supporting service authorizes all referenced columns and the model,
+        retains the selected snapshot, freezes the model version, and durably
+        accepts the job before returning 202. This operation is not idempotent;
+        clients must not automatically retry an ambiguous submission failure.
+      parameters:
+        - {name: prefix, in: path, required: true, schema: {type: string}}
+        - {name: database, in: path, required: true, schema: {type: string}}
+        - {name: table, in: path, required: true, schema: {type: string}}
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              additionalProperties: false
+              required: [model, inputs, result_columns, prompt, response_format]
+              properties:
+                model: {type: string, minLength: 1}
+                prompt: {type: string, minLength: 1}
+                response_format: {type: string, minLength: 1}
+                where: {type: string, minLength: 1}
+                snapshot_id: {type: integer, format: int64, minimum: 1}
+                tag_name: {type: string, minLength: 1}
+                inputs:
+                  type: array
+                  minItems: 1
+                  items:
+                    type: object
+                    additionalProperties: false
+                    required: [column, modality]
+                    properties:
+                      column: {type: string, minLength: 1}
+                      modality: {type: string, enum: [text, image, audio, video]}
+                result_columns:
+                  type: array
+                  uniqueItems: true
+                  items: {type: string, minLength: 1}
+              description: >-
+                snapshot_id and tag_name are mutually exclusive. When neither
+                is provided, the service fixes the latest snapshot at acceptance.
+                Model identifiers and field names are resolved by the service;
+                endpoint, credentials and execution options are not accepted.
+      responses:
+        "202":
+          description: Durably accepted; the result table is readable by the caller
+          content:
+            application/json:
+              schema:
+                type: object
+                required: [job_id, result_table, snapshot_id, model_version]
+                properties:
+                  job_id: {type: string, minLength: 1}
+                  result_table: {type: string, minLength: 1}
+                  snapshot_id: {type: integer, format: int64, minimum: 1}
+                  model_version: {type: string, minLength: 1}
+        "400": {description: Invalid request}
+        "403": {description: Not authorized}
+        "404": {description: Table or model not found, or unsupported endpoint}
+        "501": {description: Prediction is not supported}
   /v1/config:
     get:
       tags:

paimon-python/pypaimon/api/auth/dlf_openapi_actions.py

@@ -38,6 +38,7 @@
     ("GET", "/v1/{prefix}/tables/id/{tableId}", "GetTableById"),
     ("POST", "/v1/{prefix}/tables/rename", "RenameTable"),
     ("POST", "/v1/{prefix}/databases/{database}/tables/{table}/auth", "AuthTableQuery"),
+    ("POST", "/v1/{prefix}/databases/{database}/tables/{table}/predict", "PredictTable"),
     ("POST", "/v1/{prefix}/databases/{database}/tables/{table}/commit", "CommitTable"),
     ("POST", "/v1/{prefix}/databases/{database}/tables/{table}/commit-lock", "AcquireCommitLock"),
     ("POST", "/v1/{prefix}/databases/{database}/tables/{table}/commit-lock/renew",

paimon-python/pypaimon/api/predict.py

@@ -0,0 +1,83 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Experimental, provider-neutral asynchronous prediction protocol."""
+
+from dataclasses import dataclass
+from typing import Dict, List, Optional
+
+from pypaimon.api.api_request import RESTRequest
+from pypaimon.api.api_response import RESTResponse
+from pypaimon.common.json_util import optional_json_field
+
+
+@dataclass
+class PredictRequest(RESTRequest):
+    model: str
+    inputs: List[Dict[str, str]]
+    result_columns: List[str]
+    prompt: str
+    response_format: str = "json_object"
+    where: Optional[str] = optional_json_field("where", "non_null")
+    snapshot_id: Optional[int] = optional_json_field("snapshot_id", "non_null")
+    tag_name: Optional[str] = optional_json_field("tag_name", "non_null")
+
+    def __post_init__(self):
+        for name in ("model", "prompt", "response_format"):
+            _nonblank(getattr(self, name), name)
+        if self.where is not None:
+            _nonblank(self.where, "where")
+        if self.tag_name is not None:
+            _nonblank(self.tag_name, "tag_name")
+        if self.snapshot_id is not None and (
+                type(self.snapshot_id) is not int or not 0 < self.snapshot_id < 2 ** 63):
+            raise ValueError("snapshot_id must be a positive integer")
+        if self.snapshot_id is not None and self.tag_name is not None:
+            raise ValueError("Specify snapshot_id or tag_name, not both")
+        if not isinstance(self.inputs, list) or not self.inputs:
+            raise ValueError("inputs must be a non-empty list")
+        for item in self.inputs:
+            if not isinstance(item, dict) or set(item) != {"column", "modality"}:
+                raise ValueError("Each input must contain only column and modality")
+            _nonblank(item["column"], "column")
+            if item["modality"] not in ("text", "image", "audio", "video"):
+                raise ValueError("Unsupported input modality")
+        if not isinstance(self.result_columns, list):
+            raise ValueError("result_columns must be a list")
+        for column in self.result_columns:
+            _nonblank(column, "result column")
+        if len(set(self.result_columns)) != len(self.result_columns):
+            raise ValueError("Duplicate result columns")
+
+
+@dataclass
+class PredictResponse(RESTResponse):
+    job_id: str
+    result_table: str
+    snapshot_id: int
+    model_version: str
+
+    def __post_init__(self):
+        for name in ("job_id", "result_table", "model_version"):
+            _nonblank(getattr(self, name), name)
+        if type(self.snapshot_id) is not int or not 0 < self.snapshot_id < 2 ** 63:
+            raise ValueError("Accepted prediction must identify a fixed snapshot")
+
+
+def _nonblank(value, name):
+    if not isinstance(value, str) or not value.strip():
+        raise ValueError("%s must be a non-empty string" % name)

paimon-python/pypaimon/api/resource_paths.py

@@ -76,6 +76,9 @@ def table_by_id(self, table_id: str) -> str:
         table_id = urllib.parse.quote_plus(table_id, safe="*~", errors="replace")
         return "{}/{}/{}/{}".format(self.base_path, self.TABLES, self.ID, table_id)
 
+    def predict(self, database_name: str, table_name: str) -> str:
+        return self.table(database_name, table_name) + "/predict"
+
     def table_details(self, database_name: str) -> str:
         database_name = urllib.parse.quote_plus(database_name, safe="*~", errors="replace")
         return "{}/{}/{}/{}".format(

paimon-python/pypaimon/api/rest_api.py

@@ -53,6 +53,7 @@
 from pypaimon.api.auth import AuthProviderFactory, RESTAuthFunction
 from pypaimon.api.client import HttpClient
 from pypaimon.api.resource_paths import ResourcePaths
+from pypaimon.api.predict import PredictRequest, PredictResponse
 from pypaimon.api.rest_exception import NoSuchResourceException
 from pypaimon.api.rest_util import RESTUtil
 from pypaimon.api.typedef import T
@@ -85,6 +86,12 @@ class RESTApi:
     TAG_NAME_PREFIX = "tagNamePrefix"
     TOKEN_EXPIRATION_SAFE_TIME_MILLIS = 3_600_000
 
+    def predict(self, identifier: Identifier, request: PredictRequest) -> PredictResponse:
+        database, table = self._validate_resource_path(identifier)
+        return self.client.post_with_response_type(
+            self.resource_paths.predict(database, table), request,
+            PredictResponse, self.rest_auth_function)
+
     # Function name validation pattern
     _FUNCTION_NAME_PATTERN = re.compile(r'^(?=.*[A-Za-z])[A-Za-z0-9._-]+$')
 

paimon-python/pypaimon/catalog/rest/rest_catalog.py

@@ -75,6 +75,18 @@
 
 
 class RESTCatalog(Catalog):
+    def predict(self, identifier, request):
+        """Submit a prediction to a supporting REST service; never execute locally."""
+        if not isinstance(identifier, Identifier):
+            identifier = Identifier.from_string(identifier)
+        check_not_branch(identifier, "predict")
+        check_not_system_table(identifier, "predict")
+        try:
+            return self.rest_api.predict(identifier, request)
+        except NotImplementedException as e:
+            raise NotImplementedError("REST Catalog does not support prediction") from e
+        # Preserve 404: it can also mean that the table or model does not exist.
+
     def __init__(self, context: CatalogContext, config_required: Optional[bool] = True):
         self.warehouse = context.options.get(CatalogOptions.WAREHOUSE)
         self.rest_api = RESTApi(context.options, config_required)

paimon-python/pypaimon/multimodal/table.py

@@ -108,6 +108,37 @@ def copy(self, options: dict) -> 'MultimodalTable':
         return MultimodalTable(
             self.catalog, raw_table.identifier.get_full_name(), raw_table)
 
+    def predict(self, *, model, inputs, result_columns, prompt,
+                where=None, response_format="json_object"):
+        """Submit an asynchronous prediction to a supporting REST Catalog.
+
+        No rows or BLOBs are read locally. The service must retain the selected
+        snapshot and freeze the model version before accepting the job. A
+        transport failure does not prove rejection; do not blindly resubmit.
+        """
+        from dataclasses import asdict
+        from pypaimon.api.predict import PredictRequest
+        from pypaimon.table.file_store_table import FileStoreTable
+
+        provider = getattr(self.catalog, "predict", None)
+        if provider is None:
+            raise NotImplementedError("Catalog does not support prediction")
+        if not isinstance(self.raw_table, FileStoreTable):
+            raise NotImplementedError("Prediction requires a Paimon file store table")
+        options = self.raw_table.options.options.to_map()
+        allowed = {"scan.snapshot-id", "scan.tag-name"}
+        unsupported = [key for key in options
+                       if (key.startswith("scan.") or key.startswith("incremental-"))
+                       and key not in allowed]
+        if unsupported:
+            raise ValueError("Prediction does not support scan options: %s" % unsupported)
+        request = PredictRequest(
+            model=model, inputs=inputs, result_columns=result_columns,
+            prompt=prompt, where=where, response_format=response_format,
+            snapshot_id=self.raw_table.options.scan_snapshot_id(),
+            tag_name=self.raw_table.options.scan_tag_name())
+        return asdict(provider(self.raw_table.identifier, request))
+
     def add(self, data):
         arrow_table = _to_arrow_table(data, _target_schema(self.raw_table))
         write_builder = self.raw_table.new_batch_write_builder()

paimon-python/pypaimon/tests/rest/predict_test.py

@@ -0,0 +1,132 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import json
+import threading
+import unittest
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from unittest.mock import Mock
+
+from pypaimon.api.predict import PredictRequest, PredictResponse
+from pypaimon.api.rest_api import RESTApi
+from pypaimon.api.rest_exception import NoSuchResourceException, ServiceUnavailableException
+from pypaimon.catalog.rest.rest_catalog import RESTCatalog
+from pypaimon.common.identifier import Identifier
+from pypaimon.common.json_util import JSON
+from pypaimon.common.options.core_options import CoreOptions
+from pypaimon.common.options.options import Options
+from pypaimon.multimodal.table import MultimodalTable
+from pypaimon.table.file_store_table import FileStoreTable
+
+
+def request(**kwargs):
+    values = dict(model="quality", inputs=[{"column": "content", "modality": "image"}],
+                  result_columns=["id"], prompt="Classify")
+    values.update(kwargs)
+    return PredictRequest(**values)
+
+
+class PredictTest(unittest.TestCase):
+    def test_request_round_trip(self):
+        value = request(snapshot_id=7, where="dt = 'today'")
+        body = JSON.to_json(value)
+        self.assertEqual(value, JSON.from_json(body, PredictRequest))
+        self.assertNotIn("tag_name", json.loads(body))
+        self.assertNotIn("identifier", json.loads(body))
+
+    def test_invalid_requests(self):
+        for changes in [dict(inputs=[]), dict(inputs=[{"column": "x", "modality": "other"}]),
+                        dict(inputs=[{"column": "x", "modality": "text", "endpoint": "url"}]),
+                        dict(result_columns=["id", "id"]), dict(result_columns="id"),
+                        dict(snapshot_id=True), dict(snapshot_id=0), dict(snapshot_id=2 ** 63),
+                        dict(model=" "),
+                        dict(tag_name="v1", snapshot_id=1), dict(where="")]:
+            with self.subTest(changes=changes), self.assertRaises(ValueError):
+                request(**changes)
+
+    def test_table_view_preserves_snapshot(self):
+        for options in ({}, {"scan.snapshot-id": "7"}, {"scan.tag-name": "release"}):
+            raw = Mock(spec=FileStoreTable)
+            raw.options = CoreOptions(Options(options))
+            raw.identifier = Identifier.create("db", "images")
+            catalog = Mock()
+            catalog.predict.return_value = PredictResponse("job1", "results.job1", 7, "v1")
+            table = MultimodalTable(catalog, "db.images", raw)
+            result = table.predict(model="quality", inputs=request().inputs,
+                                   result_columns=["id"], prompt="Classify")
+            sent = catalog.predict.call_args.args[1]
+            self.assertEqual(7 if "scan.snapshot-id" in options else None, sent.snapshot_id)
+            self.assertEqual(options.get("scan.tag-name"), sent.tag_name)
+            self.assertEqual("job1", result["job_id"])
+            raw.new_read_builder.assert_not_called()
+
+    def test_unsupported_view_does_not_submit(self):
+        raw = Mock(spec=FileStoreTable)
+        raw.options = CoreOptions(Options({"scan.timestamp-millis": "1234"}))
+        catalog = Mock()
+        table = MultimodalTable(catalog, "db.images", raw)
+        with self.assertRaises(ValueError):
+            table.predict(model="quality", inputs=request().inputs,
+                          result_columns=[], prompt="Classify")
+        catalog.predict.assert_not_called()
+        with self.assertRaises(NotImplementedError):
+            MultimodalTable(object(), "db.images", raw).predict(
+                model="quality", inputs=[], result_columns=[], prompt="Classify")
+
+    def test_http_submission_and_errors(self):
+        calls = []
+        state = {"status": 202, "body": JSON.to_json(
+            PredictResponse("job1", "results.job1", 7, "v1"))}
+
+        class Handler(BaseHTTPRequestHandler):
+            def log_message(self, *args):
+                pass
+
+            def do_POST(self):
+                calls.append((self.path, json.loads(self.rfile.read(
+                    int(self.headers["Content-Length"]))), self.headers.get("X-Test")))
+                body = state["body"].encode()
+                self.send_response(state["status"])
+                self.send_header("Content-Type", "application/json")
+                self.send_header("Content-Length", str(len(body)))
+                self.end_headers()
+                self.wfile.write(body)
+
+        server = HTTPServer(("127.0.0.1", 0), Handler)
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        try:
+            api = RESTApi({"uri": "http://127.0.0.1:%s" % server.server_port,
+                           "token.provider": "bear", "token": "test",
+                           "prefix": "catalog", "header.X-Test": "reuse"}, False)
+            catalog = object.__new__(RESTCatalog)
+            catalog.rest_api = api
+            result = catalog.predict(Identifier.create("a/b", "t x"), request())
+            self.assertEqual("job1", result.job_id)
+            self.assertEqual("/v1/catalog/databases/a%2Fb/tables/t+x/predict", calls[0][0])
+            self.assertEqual("reuse", calls[0][2])
+            for status, error in [(501, NotImplementedError), (404, NoSuchResourceException),
+                                  (503, ServiceUnavailableException)]:
+                state.update(status=status, body=json.dumps({"code": status, "message": "no",
+                                                             "resourceType": "TABLE"}))
+                with self.assertRaises(error):
+                    catalog.predict("db.images", request())
+            self.assertEqual(4, len(calls))
+        finally:
+            server.shutdown()
+            server.server_close()
+            thread.join()