This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 3da36f2f99 [python] Support create_partitions in RESTCatalog (#9198)
3da36f2f99 is described below
commit 3da36f2f99b9416a83c281c1c9bbfdb620af6a65
Author: Jiajia Li <[email protected]>
AuthorDate: Thu Aug 13 21:58:24 2026 +0800
[python] Support create_partitions in RESTCatalog (#9198)
---
paimon-python/pypaimon/api/api_request.py | 13 ++
paimon-python/pypaimon/api/api_response.py | 9 ++
paimon-python/pypaimon/api/rest_api.py | 24 +++-
paimon-python/pypaimon/catalog/catalog.py | 11 ++
.../pypaimon/catalog/catalog_exception.py | 5 +
.../pypaimon/catalog/rest/rest_catalog.py | 24 +++-
.../pypaimon/tests/api/test_partition_dto_serde.py | 103 ++++++++++++++++
.../api/test_rest_catalog_create_partitions.py | 101 ++++++++++++++++
.../tests/rest/rest_create_partitions_test.py | 131 +++++++++++++++++++++
paimon-python/pypaimon/tests/rest/rest_server.py | 49 ++++++--
10 files changed, 459 insertions(+), 11 deletions(-)
diff --git a/paimon-python/pypaimon/api/api_request.py
b/paimon-python/pypaimon/api/api_request.py
index e983618387..15c9372ac0 100644
--- a/paimon-python/pypaimon/api/api_request.py
+++ b/paimon-python/pypaimon/api/api_request.py
@@ -177,6 +177,19 @@ class CreateTagRequest(RESTRequest):
time_retained: Optional[str] = json_field(FIELD_TIME_RETAINED,
default=None)
+@dataclass
+class CreatePartitionsRequest(RESTRequest):
+ FIELD_PARTITION_SPECS = "partitionSpecs"
+ FIELD_IGNORE_IF_EXISTS = "ignoreIfExists"
+
+ partition_specs: List[Dict[str, str]] = json_field(FIELD_PARTITION_SPECS)
+ ignore_if_exists: Optional[bool] = json_field(FIELD_IGNORE_IF_EXISTS,
default=True)
+
+ def __post_init__(self):
+ if self.ignore_if_exists is None:
+ self.ignore_if_exists = True
+
+
# Branch CRUD wire DTOs. Mirrors Java requests in
# paimon-api/.../rest/requests/.
@dataclass
diff --git a/paimon-python/pypaimon/api/api_response.py
b/paimon-python/pypaimon/api/api_response.py
index a2168e301d..1014895152 100644
--- a/paimon-python/pypaimon/api/api_response.py
+++ b/paimon-python/pypaimon/api/api_response.py
@@ -167,6 +167,15 @@ class ListPartitionsResponse(PagedResponse['Partition']):
return self.next_page_token
+@dataclass
+class CreatePartitionsResponse(RESTResponse):
+ FIELD_CREATED = "created"
+ FIELD_EXISTED = "existed"
+
+ created: Optional[List[Dict[str, str]]] = json_field(FIELD_CREATED,
default=None)
+ existed: Optional[List[Dict[str, str]]] = json_field(FIELD_EXISTED,
default=None)
+
+
@dataclass
class ListTablesResponse(PagedResponse[str]):
FIELD_TABLES = "tables"
diff --git a/paimon-python/pypaimon/api/rest_api.py
b/paimon-python/pypaimon/api/rest_api.py
index 96a0321794..990536cb05 100755
--- a/paimon-python/pypaimon/api/rest_api.py
+++ b/paimon-python/pypaimon/api/rest_api.py
@@ -23,8 +23,9 @@ import re
from pypaimon.api.api_request import (AlterDatabaseRequest,
AlterFunctionRequest,
AlterTableRequest, CommitTableRequest,
CreateBranchRequest,
CreateDatabaseRequest,
- CreateFunctionRequest,
CreateTableRequest,
- CreateTagRequest, ForwardBranchRequest,
+ CreateFunctionRequest,
CreatePartitionsRequest,
+ CreateTableRequest, CreateTagRequest,
+ ForwardBranchRequest,
RenameBranchRequest, RenameTableRequest,
RollbackTableRequest)
from pypaimon.api.api_response import (CommitTableResponse, ConfigResponse,
@@ -36,6 +37,7 @@ from pypaimon.api.api_response import (CommitTableResponse,
ConfigResponse,
ListFunctionDetailsResponse,
ListFunctionsGloballyResponse,
ListFunctionsResponse,
+ CreatePartitionsResponse,
ListPartitionsResponse,
ListTablesResponse, ListTagsResponse,
PagedList,
@@ -451,6 +453,24 @@ class RESTApi:
partitions = response.data() or []
return PagedList(partitions, response.get_next_page_token())
+ def create_partitions(
+ self,
+ identifier: Identifier,
+ partitions: List[Dict[str, str]],
+ ignore_if_exists: bool = True,
+ ) -> CreatePartitionsResponse:
+ database_name, table_name = self.__validate_identifier(identifier)
+ request = CreatePartitionsRequest(
+ partition_specs=partitions,
+ ignore_if_exists=ignore_if_exists,
+ )
+ return self.client.post_with_response_type(
+ self.resource_paths.partitions(database_name, table_name),
+ request,
+ CreatePartitionsResponse,
+ self.rest_auth_function,
+ )
+
# Tag CRUD wrappers — mirror Java RESTApi tag methods.
def create_tag(
self,
diff --git a/paimon-python/pypaimon/catalog/catalog.py
b/paimon-python/pypaimon/catalog/catalog.py
index d3c9c6e17f..ea00ad6dd6 100644
--- a/paimon-python/pypaimon/catalog/catalog.py
+++ b/paimon-python/pypaimon/catalog/catalog.py
@@ -193,6 +193,17 @@ class Catalog(ABC):
"rollback_to is not supported by this catalog."
)
+ def create_partitions(
+ self,
+ identifier: Union[str, Identifier],
+ partitions: List[Dict[str, str]],
+ ignore_if_exists: bool = True,
+ ) -> None:
+ raise NotImplementedError(
+ "create_partitions is not supported by this catalog. "
+ "Use REST catalog for partition creation."
+ )
+
def drop_partitions(
self,
identifier: Union[str, Identifier],
diff --git a/paimon-python/pypaimon/catalog/catalog_exception.py
b/paimon-python/pypaimon/catalog/catalog_exception.py
index d81c90f63b..ffcecff21f 100644
--- a/paimon-python/pypaimon/catalog/catalog_exception.py
+++ b/paimon-python/pypaimon/catalog/catalog_exception.py
@@ -190,3 +190,8 @@ class TagAlreadyExistException(CatalogException):
class IllegalArgumentError(CatalogException):
"""Illegal argument exception"""
pass
+
+
+class IllegalStateError(CatalogException):
+ """Illegal state exception"""
+ pass
diff --git a/paimon-python/pypaimon/catalog/rest/rest_catalog.py
b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
index 35db60061d..fd325711de 100644
--- a/paimon-python/pypaimon/catalog/rest/rest_catalog.py
+++ b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
@@ -19,7 +19,7 @@ import logging
from typing import Any, Callable, Dict, List, Optional, Union
from pypaimon.api.api_response import GetTableResponse, PagedList,
ErrorResponse
from pypaimon.api.rest_api import RESTApi
-from pypaimon.catalog.catalog_exception import IllegalArgumentError
+from pypaimon.catalog.catalog_exception import IllegalArgumentError,
IllegalStateError
from pypaimon.api.rest_exception import (NoSuchResourceException,
AlreadyExistsException,
ForbiddenException,
BadRequestException,
ServiceFailureException,
NotImplementedException)
@@ -292,6 +292,28 @@ class RESTCatalog(Catalog):
except ForbiddenException as e:
raise TableNoPermissionException(identifier) from e
+ def create_partitions(
+ self,
+ identifier: Union[str, Identifier],
+ partitions: List[Dict[str, str]],
+ ignore_if_exists: bool = True,
+ ) -> None:
+ """Register partitions. A catalog operation: it does not touch table
data."""
+ if not isinstance(identifier, Identifier):
+ identifier = Identifier.from_string(identifier)
+ try:
+ self.rest_api.create_partitions(identifier, partitions,
ignore_if_exists)
+ except NoSuchResourceException as e:
+ raise TableNotExistException(identifier) from e
+ except ForbiddenException as e:
+ raise TableNoPermissionException(identifier) from e
+ except AlreadyExistsException as e:
+ raise IllegalStateError(
+ "Some partitions of table {} already exist: {}".format(
+ identifier.get_full_name(), e)) from e
+ except BadRequestException as e:
+ raise IllegalArgumentError(str(e)) from e
+
def drop_partitions(
self,
identifier: Union[str, Identifier],
diff --git a/paimon-python/pypaimon/tests/api/test_partition_dto_serde.py
b/paimon-python/pypaimon/tests/api/test_partition_dto_serde.py
new file mode 100644
index 0000000000..7b5ec38f40
--- /dev/null
+++ b/paimon-python/pypaimon/tests/api/test_partition_dto_serde.py
@@ -0,0 +1,103 @@
+# 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 unittest
+
+from pypaimon.api.api_request import CreatePartitionsRequest
+from pypaimon.api.api_response import CreatePartitionsResponse
+from pypaimon.api.resource_paths import ResourcePaths
+from pypaimon.common.json_util import JSON
+
+
+class CreatePartitionsRequestSerdeTest(unittest.TestCase):
+
+ def test_to_json_uses_java_field_names(self):
+ request = CreatePartitionsRequest(
+ partition_specs=[{"dt": "20260807", "hour": "01"}],
+ )
+ parsed = json.loads(JSON.to_json(request))
+ self.assertEqual(parsed["partitionSpecs"], [{"dt": "20260807", "hour":
"01"}])
+ self.assertEqual(set(parsed.keys()), {"partitionSpecs",
"ignoreIfExists"})
+ self.assertNotIn("partition_specs", parsed)
+ self.assertNotIn("ignore_if_exists", parsed)
+
+ def test_ignore_if_exists_defaults_to_true(self):
+ request = CreatePartitionsRequest(partition_specs=[{"dt": "20260807"}])
+ self.assertIs(json.loads(JSON.to_json(request))["ignoreIfExists"],
True)
+
+ def test_ignore_if_exists_false_is_serialized(self):
+ request = CreatePartitionsRequest(
+ partition_specs=[{"dt": "20260807"}],
+ ignore_if_exists=False,
+ )
+ self.assertIs(json.loads(JSON.to_json(request))["ignoreIfExists"],
False)
+
+ def test_explicit_null_ignore_if_exists_reads_as_true(self):
+ request = JSON.from_json(
+ json.dumps({"partitionSpecs": [{"dt": "20260807"}],
"ignoreIfExists": None}),
+ CreatePartitionsRequest,
+ )
+ self.assertIs(request.ignore_if_exists, True)
+
+ def test_multiple_specs_keep_order_and_all_keys(self):
+ specs = [
+ {"dt": "20260807", "hour": "01"},
+ {"dt": "20260807", "hour": "02"},
+ {"dt": "20260808", "hour": "00"},
+ ]
+ parsed =
json.loads(JSON.to_json(CreatePartitionsRequest(partition_specs=specs)))
+ self.assertEqual(parsed["partitionSpecs"], specs)
+
+
+class CreatePartitionsResponseSerdeTest(unittest.TestCase):
+
+ def test_from_json_splits_created_and_existed(self):
+ response = JSON.from_json(
+ json.dumps({
+ "created": [{"dt": "20260807"}],
+ "existed": [{"dt": "20260806"}],
+ }),
+ CreatePartitionsResponse,
+ )
+ self.assertEqual(response.created, [{"dt": "20260807"}])
+ self.assertEqual(response.existed, [{"dt": "20260806"}])
+
+ def test_from_json_tolerates_missing_fields(self):
+ response = JSON.from_json(json.dumps({"created": []}),
CreatePartitionsResponse)
+ self.assertEqual(response.created, [])
+ self.assertIsNone(response.existed)
+
+
+class ResourcePathsPartitionsTest(unittest.TestCase):
+
+ def test_partitions_collection_url(self):
+ paths = ResourcePaths(prefix="mock")
+ self.assertEqual(
+ paths.partitions("db", "tbl"),
+ "/v1/mock/databases/db/tables/tbl/partitions",
+ )
+
+ def test_partitions_url_url_encodes_names(self):
+ paths = ResourcePaths(prefix="mock")
+ self.assertEqual(
+ paths.partitions("my db", "my tbl"),
+ "/v1/mock/databases/my%20db/tables/my%20tbl/partitions",
+ )
+
+if __name__ == "__main__":
+ unittest.main()
diff --git
a/paimon-python/pypaimon/tests/api/test_rest_catalog_create_partitions.py
b/paimon-python/pypaimon/tests/api/test_rest_catalog_create_partitions.py
new file mode 100644
index 0000000000..e7e2170ff7
--- /dev/null
+++ b/paimon-python/pypaimon/tests/api/test_rest_catalog_create_partitions.py
@@ -0,0 +1,101 @@
+# 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 unittest
+
+from pypaimon.api.rest_exception import (AlreadyExistsException,
BadRequestException,
+ ForbiddenException,
NoSuchResourceException)
+from pypaimon.catalog.catalog_exception import (IllegalArgumentError,
IllegalStateError,
+ TableNoPermissionException,
+ TableNotExistException)
+from pypaimon.catalog.rest.rest_catalog import RESTCatalog
+from pypaimon.common.identifier import Identifier
+
+
+class _RecordingApi:
+ """Stub RESTApi: records create_partitions calls, optionally raises."""
+
+ def __init__(self, raises=None):
+ self.calls = []
+ self._raises = raises
+
+ def create_partitions(self, identifier, partitions, ignore_if_exists):
+ self.calls.append((identifier, partitions, ignore_if_exists))
+ if self._raises is not None:
+ raise self._raises
+ return None
+
+
+def _catalog_with(api):
+ catalog = object.__new__(RESTCatalog)
+ catalog.rest_api = api
+ return catalog
+
+
+class CreatePartitionsBehaviourTest(unittest.TestCase):
+
+ def test_specs_and_flag_reach_the_api_unchanged(self):
+ api = _RecordingApi()
+ specs = [{"dt": "20260807", "hour": "01"}]
+ _catalog_with(api).create_partitions("db.tbl", specs)
+
+ self.assertEqual(len(api.calls), 1)
+ identifier, sent, ignore_if_exists = api.calls[0]
+ self.assertEqual(identifier.get_full_name(), "db.tbl")
+ self.assertEqual(sent, specs)
+ self.assertTrue(ignore_if_exists)
+
+ def test_ignore_if_exists_false_is_passed_through(self):
+ api = _RecordingApi()
+ _catalog_with(api).create_partitions(
+ Identifier.from_string("db.tbl"), [{"dt": "20260807"}],
ignore_if_exists=False)
+ self.assertFalse(api.calls[0][2])
+
+ def test_empty_partitions_are_still_sent(self):
+ api = _RecordingApi()
+ self.assertIsNone(_catalog_with(api).create_partitions("db.tbl", []))
+ self.assertEqual(api.calls[0][1], [])
+
+ def test_string_identifier_is_parsed(self):
+ api = _RecordingApi()
+ _catalog_with(api).create_partitions("db.tbl", [{"dt": "1"}])
+ self.assertIsInstance(api.calls[0][0], Identifier)
+
+
+class CreatePartitionsErrorMappingTest(unittest.TestCase):
+ """Mirrors the four handlers in Java ``RESTCatalog.createPartitions``."""
+
+ def _raise_and_assert(self, rest_error, expected):
+ catalog = _catalog_with(_RecordingApi(raises=rest_error))
+ with self.assertRaises(expected):
+ catalog.create_partitions("db.tbl", [{"dt": "20260807"}])
+
+ def test_no_such_resource_becomes_table_not_exist(self):
+ self._raise_and_assert(
+ NoSuchResourceException(None, None, "no table"),
TableNotExistException)
+
+ def test_forbidden_becomes_table_no_permission(self):
+ self._raise_and_assert(ForbiddenException("denied"),
TableNoPermissionException)
+
+ def test_already_exists_becomes_illegal_state(self):
+ self._raise_and_assert(AlreadyExistsException(None, None, "exists"),
IllegalStateError)
+
+ def test_bad_request_becomes_illegal_argument(self):
+ self._raise_and_assert(BadRequestException("bad"),
IllegalArgumentError)
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/paimon-python/pypaimon/tests/rest/rest_create_partitions_test.py
b/paimon-python/pypaimon/tests/rest/rest_create_partitions_test.py
new file mode 100644
index 0000000000..3c36f93b7e
--- /dev/null
+++ b/paimon-python/pypaimon/tests/rest/rest_create_partitions_test.py
@@ -0,0 +1,131 @@
+# 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 pyarrow as pa
+
+from pypaimon import Schema
+from pypaimon.catalog.catalog_exception import IllegalStateError,
TableNotExistException
+from pypaimon.common.identifier import Identifier
+from pypaimon.table.format.format_table import FormatTable
+from pypaimon.tests.rest.rest_base_test import RESTBaseTest
+
+
+class RESTCreatePartitionsTest(RESTBaseTest):
+
+ def _create_table(self, table_name, partition_keys=None):
+ schema = Schema.from_pyarrow_schema(
+ pa.schema([("id", pa.int32()), ("dt", pa.string()), ("hh",
pa.string())]),
+ partition_keys=partition_keys if partition_keys is not None else
["dt", "hh"],
+ options={"type": "format-table", "file.format": "parquet",
+ "metastore.partitioned-table": "true"},
+ )
+ self.rest_catalog.drop_table(table_name, True)
+ self.rest_catalog.create_table(table_name, schema, False)
+ return table_name
+
+ def _listed_specs(self, table_name):
+ return [p.spec for p in
self.rest_catalog.list_partitions_paged(table_name).elements]
+
+ def test_created_partitions_come_back_from_list(self):
+ table_name = self._create_table("default.create_partitions_round_trip")
+ specs = [
+ {"dt": "20260807", "hh": "01"},
+ {"dt": "20260807", "hh": "02"},
+ ]
+ self.rest_catalog.create_partitions(table_name, specs)
+
+ listed = self._listed_specs(table_name)
+ for spec in specs:
+ self.assertIn(spec, listed)
+ fresh = self.rest_catalog.list_partitions_paged(table_name).elements[0]
+ self.assertEqual(fresh.record_count, 0)
+ self.assertEqual(fresh.last_file_creation_time, 0)
+
+ def test_creating_twice_is_a_no_op_by_default(self):
+ table_name = self._create_table("default.create_partitions_idempotent")
+ specs = [{"dt": "20260807", "hh": "01"}]
+
+ self.rest_catalog.create_partitions(table_name, specs)
+ self.rest_catalog.create_partitions(table_name, specs)
+
+ listed = self._listed_specs(table_name)
+ self.assertEqual(listed.count({"dt": "20260807", "hh": "01"}), 1)
+
+ def test_creating_twice_raises_when_not_ignoring(self):
+ table_name = self._create_table("default.create_partitions_strict")
+ specs = [{"dt": "20260807", "hh": "01"}]
+ self.rest_catalog.create_partitions(table_name, specs)
+
+ with self.assertRaises(IllegalStateError):
+ self.rest_catalog.create_partitions(table_name, specs,
ignore_if_exists=False)
+
+ def test_rejected_request_creates_nothing(self):
+ table_name = self._create_table("default.create_partitions_atomic")
+ self.rest_catalog.create_partitions(table_name, [{"dt": "20260807",
"hh": "01"}])
+ before = self._listed_specs(table_name)
+
+ with self.assertRaises(IllegalStateError):
+ self.rest_catalog.create_partitions(
+ table_name,
+ [{"dt": "20260808", "hh": "00"}, {"dt": "20260807", "hh":
"01"}],
+ ignore_if_exists=False,
+ )
+
+ self.assertEqual(self._listed_specs(table_name), before)
+
+ def test_empty_list_leaves_the_table_untouched(self):
+ table_name = self._create_table("default.create_partitions_empty")
+ before = self._listed_specs(table_name)
+ self.rest_catalog.create_partitions(table_name, [])
+ self.assertEqual(self._listed_specs(table_name), before)
+
+ def test_unknown_table_raises_table_not_exist(self):
+ with self.assertRaises(TableNotExistException):
+ self.rest_catalog.create_partitions(
+ "default.create_partitions_no_such_table", [{"dt":
"20260807"}])
+
+ def test_unknown_table_raises_on_an_empty_list_too(self):
+ with self.assertRaises(TableNotExistException):
+ self.rest_catalog.create_partitions(
+ "default.create_partitions_no_such_table", [])
+
+ def test_registers_partitions_of_a_format_table(self):
+ table_name =
self._create_table("default.create_partitions_format_table")
+ self.assertIsInstance(self.rest_catalog.get_table(table_name),
FormatTable)
+
+ self.rest_catalog.create_partitions(table_name, [{"dt": "20260807",
"hh": "01"}])
+
+ self.assertIn({"dt": "20260807", "hh": "01"},
self._listed_specs(table_name))
+
+ def test_response_separates_created_from_existed(self):
+ table_name = self._create_table("default.create_partitions_response")
+ old = {"dt": "20260807", "hh": "01"}
+ new = {"dt": "20260807", "hh": "02"}
+ self.rest_catalog.create_partitions(table_name, [old])
+
+ response = self.rest_catalog.rest_api.create_partitions(
+ Identifier.from_string(table_name), [old, new], True)
+
+ self.assertIsNotNone(response)
+ self.assertEqual(response.created, [new])
+ self.assertEqual(response.existed, [old])
+
+ def test_works_for_a_single_key_partitioned_table(self):
+ table_name = self._create_table(
+ "default.create_partitions_single_key", partition_keys=["dt"])
+ self.rest_catalog.create_partitions(table_name, [{"dt": "20260807"}])
+ self.assertIn({"dt": "20260807"}, self._listed_specs(table_name))
diff --git a/paimon-python/pypaimon/tests/rest/rest_server.py
b/paimon-python/pypaimon/tests/rest/rest_server.py
index 69d4361bb5..92d089b302 100755
--- a/paimon-python/pypaimon/tests/rest/rest_server.py
+++ b/paimon-python/pypaimon/tests/rest/rest_server.py
@@ -31,10 +31,12 @@ if TYPE_CHECKING:
from pypaimon.api.api_request import (AlterDatabaseRequest, AlterTableRequest,
CreateBranchRequest,
CreateDatabaseRequest,
+ CreatePartitionsRequest,
CreateTableRequest, CreateTagRequest,
RenameBranchRequest,
RenameTableRequest)
-from pypaimon.api.api_response import (ConfigResponse, GetDatabaseResponse,
+from pypaimon.api.api_response import (ConfigResponse,
CreatePartitionsResponse,
+ GetDatabaseResponse,
GetFunctionResponse,
GetTableResponse, GetTagResponse,
ListBranchesResponse,
@@ -432,7 +434,7 @@ class RESTCatalogServer:
if resource_type == ResourcePaths.TABLES:
return self._handle_table_resource(method, path_parts,
identifier, data, parameters)
elif resource_type == ResourcePaths.PARTITIONS:
- return self._table_partitions_handle(method,
identifier, parameters)
+ return self._table_partitions_handle(method, data,
identifier, parameters)
elif resource_type == ResourcePaths.FUNCTIONS:
return self._function_handle(method, data, identifier)
@@ -548,7 +550,7 @@ class RESTCatalogServer:
elif operation == "snapshot":
return self._table_snapshot_handle(method, lookup_identifier)
elif operation == ResourcePaths.PARTITIONS:
- return self._table_partitions_handle(method,
lookup_identifier, parameters)
+ return self._table_partitions_handle(method, data,
lookup_identifier, parameters)
elif operation == ResourcePaths.TAGS:
return self._tags_handle(method, data, lookup_identifier,
parameters)
elif operation == ResourcePaths.BRANCHES:
@@ -754,17 +756,48 @@ class RESTCatalogServer:
return self._mock_response(response, 200)
def _table_partitions_handle(
- self, method: str, identifier: Identifier, parameters: Dict[str,
str]) -> Tuple[str, int]:
- """Handle table partitions listing"""
- if method != "GET":
- return self._mock_response(ErrorResponse(None, None, "Method Not
Allowed", 405), 405)
-
+ self, method: str, data: str, identifier: Identifier,
+ parameters: Dict[str, str]) -> Tuple[str, int]:
+ """Handle the table-scoped partitions collection (POST create / GET
list-paged)."""
if identifier.get_full_name() not in self.table_metadata_store:
raise TableNotExistException(identifier)
+ if method == "POST":
+ request = JSON.from_json(data, CreatePartitionsRequest)
+ store =
self.table_partitions_store.setdefault(identifier.get_full_name(), [])
+ existing = {self._partition_spec_key(p.spec) for p in store}
+ seen, created, existed = set(existing), [], []
+ for spec in request.partition_specs or []:
+ key = self._partition_spec_key(spec)
+ if key in seen:
+ existed.append(spec)
+ continue
+ seen.add(key)
+ created.append(spec)
+ if existed and not request.ignore_if_exists:
+ return self._mock_response(
+ ErrorResponse(None, None,
+ "Partition already exists:
{}".format(existed[0]), 409),
+ 409)
+ store.extend(
+ Partition(spec=dict(spec), record_count=0,
file_size_in_bytes=0,
+ file_count=0, last_file_creation_time=0,
total_buckets=-1,
+ done=False)
+ for spec in created)
+ return self._mock_response(
+ CreatePartitionsResponse(created=created, existed=existed),
200)
+
+ if method != "GET":
+ return self._mock_response(ErrorResponse(None, None, "Method Not
Allowed", 405), 405)
+
partitions = self._list_partitions(identifier, parameters)
return self._generate_final_list_partitions_response(parameters,
partitions)
+ @staticmethod
+ def _partition_spec_key(spec: Dict[str, str]) -> str:
+ """Order-independent identity for a spec."""
+ return str(sorted((spec or {}).items()))
+
# ======================= Tag Handlers ====================================
def _tags_handle(self, method: str, data: str, identifier: Identifier,
