From 2f01871a972a98372084d02ef15b5e20d37b674d Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Thu, 29 Oct 2026 10:22:48 -0700 Subject: [PATCH] [python] Add experimental asynchronous table prediction submission --- docs/static/rest-catalog-open-api.yaml | 65 +++++++++ .../pypaimon/api/auth/dlf_openapi_actions.py | 1 + paimon-python/pypaimon/api/predict.py | 83 +++++++++++ paimon-python/pypaimon/api/resource_paths.py | 3 + paimon-python/pypaimon/api/rest_api.py | 7 + .../pypaimon/catalog/rest/rest_catalog.py | 12 ++ paimon-python/pypaimon/multimodal/table.py | 31 ++++ .../pypaimon/tests/rest/predict_test.py | 132 ++++++++++++++++++ .../tests/rest/rest_api_coverage_test.py | 3 + 9 files changed, 337 insertions(+) create mode 100644 paimon-python/pypaimon/api/predict.py create mode 100644 paimon-python/pypaimon/tests/rest/predict_test.py diff --git a/docs/static/rest-catalog-open-api.yaml b/docs/static/rest-catalog-open-api.yaml index 6d16745d30b7..8816b80478d3 100644 --- a/docs/static/rest-catalog-open-api.yaml +++ b/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: diff --git a/paimon-python/pypaimon/api/auth/dlf_openapi_actions.py b/paimon-python/pypaimon/api/auth/dlf_openapi_actions.py index e6e1163600d2..a8d1800271ea 100644 --- a/paimon-python/pypaimon/api/auth/dlf_openapi_actions.py +++ b/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", diff --git a/paimon-python/pypaimon/api/predict.py b/paimon-python/pypaimon/api/predict.py new file mode 100644 index 000000000000..b7b17835cced --- /dev/null +++ b/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) diff --git a/paimon-python/pypaimon/api/resource_paths.py b/paimon-python/pypaimon/api/resource_paths.py index b47f74cc9986..f54365a89eab 100644 --- a/paimon-python/pypaimon/api/resource_paths.py +++ b/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( diff --git a/paimon-python/pypaimon/api/rest_api.py b/paimon-python/pypaimon/api/rest_api.py index bb2287529545..5ea6c265711b 100755 --- a/paimon-python/pypaimon/api/rest_api.py +++ b/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._-]+$') diff --git a/paimon-python/pypaimon/catalog/rest/rest_catalog.py b/paimon-python/pypaimon/catalog/rest/rest_catalog.py index 0974079c5364..8d14018e73f1 100644 --- a/paimon-python/pypaimon/catalog/rest/rest_catalog.py +++ b/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) diff --git a/paimon-python/pypaimon/multimodal/table.py b/paimon-python/pypaimon/multimodal/table.py index 859a9036dd80..b5e9f3920d35 100644 --- a/paimon-python/pypaimon/multimodal/table.py +++ b/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() diff --git a/paimon-python/pypaimon/tests/rest/predict_test.py b/paimon-python/pypaimon/tests/rest/predict_test.py new file mode 100644 index 000000000000..591968b484ce --- /dev/null +++ b/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() diff --git a/paimon-python/pypaimon/tests/rest/rest_api_coverage_test.py b/paimon-python/pypaimon/tests/rest/rest_api_coverage_test.py index b55e0da0dda3..f0be5634510b 100644 --- a/paimon-python/pypaimon/tests/rest/rest_api_coverage_test.py +++ b/paimon-python/pypaimon/tests/rest/rest_api_coverage_test.py @@ -26,6 +26,7 @@ import yaml from pypaimon.api.rest_api import RESTApi +from pypaimon.api.predict import PredictRequest from pypaimon.common.identifier import Identifier from pypaimon.function.function import FunctionImpl from pypaimon.function.function_change import FunctionChange @@ -53,6 +54,8 @@ # A value per parameter annotation, else per name; "SELECT" is valid both as a name and as an access. _ARGUMENTS = { + PredictRequest: PredictRequest("quality", [{"column": "content", "modality": "image"}], + ["id"], "Classify"), str: "SELECT", bool: False, Identifier: Identifier.create("db", "t"),