This is an automated email from the ASF dual-hosted git repository. mchades pushed a commit to branch codex/strict-function-data-type-input in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit 803dc36a5f2440f384234936297a2b59d190aa95 Author: mchades <[email protected]> AuthorDate: Fri Sep 4 18:39:31 2026 +0800 [#12927] fix(function): Reject unresolved data types in new definitions --- .../dto/requests/FunctionDataTypeValidator.java | 124 ++++++++++++++++ .../dto/requests/FunctionRegisterRequest.java | 11 +- .../dto/requests/FunctionUpdateRequest.java | 7 +- .../dto/requests/FunctionUpdatesRequest.java | 15 +- .../gravitino/dto/function/TestFunctionDTO.java | 27 ++++ .../dto/requests/TestFunctionRegisterRequest.java | 147 ++++++++++++++++++ .../dto/requests/TestFunctionUpdateRequest.java | 78 ++++++++++ .../dto/requests/TestFunctionUpdatesRequest.java | 47 ++++++ .../catalog/TestManagedFunctionOperations.java | 50 +++++++ .../storage/relational/po/TestFunctionPO.java | 59 ++++++++ docs/open-api/datatype.yaml | 165 +++++++++++++++++++++ docs/open-api/functions.yaml | 72 ++++++++- .../server/web/rest/TestFunctionOperations.java | 110 ++++++++++++++ 13 files changed, 903 insertions(+), 9 deletions(-) diff --git a/common/src/main/java/org/apache/gravitino/dto/requests/FunctionDataTypeValidator.java b/common/src/main/java/org/apache/gravitino/dto/requests/FunctionDataTypeValidator.java new file mode 100644 index 0000000000..bef5a6fcf9 --- /dev/null +++ b/common/src/main/java/org/apache/gravitino/dto/requests/FunctionDataTypeValidator.java @@ -0,0 +1,124 @@ +/* + * 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. + */ +package org.apache.gravitino.dto.requests; + +import com.google.common.base.Preconditions; +import org.apache.commons.lang3.StringUtils; +import org.apache.gravitino.dto.function.FunctionColumnDTO; +import org.apache.gravitino.dto.function.FunctionDefinitionDTO; +import org.apache.gravitino.dto.function.FunctionParamDTO; +import org.apache.gravitino.rel.types.Type; +import org.apache.gravitino.rel.types.Types; + +final class FunctionDataTypeValidator { + + private FunctionDataTypeValidator() {} + + static void validateDefinition(FunctionDefinitionDTO definition, String fieldPath) { + Preconditions.checkArgument( + definition != null, "\"%s\" field is required and cannot be null", fieldPath); + + FunctionParamDTO[] parameters = definition.getParameters(); + if (parameters != null) { + for (int i = 0; i < parameters.length; i++) { + String parameterPath = fieldPath + ".parameters[" + i + "]"; + Preconditions.checkArgument( + parameters[i] != null, "\"%s\" field is required and cannot be null", parameterPath); + validateDataType(parameters[i].getDataType(), parameterPath + ".dataType"); + } + } + + if (definition.getReturnType() != null) { + validateDataType(definition.getReturnType(), fieldPath + ".returnType"); + } + + FunctionColumnDTO[] returnColumns = definition.getReturnColumns(); + if (returnColumns != null) { + for (int i = 0; i < returnColumns.length; i++) { + String returnColumnPath = fieldPath + ".returnColumns[" + i + "]"; + Preconditions.checkArgument( + returnColumns[i] != null, + "\"%s\" field is required and cannot be null", + returnColumnPath); + validateDataType(returnColumns[i].getDataType(), returnColumnPath + ".dataType"); + } + } + } + + private static void validateDataType(Type dataType, String fieldPath) { + Preconditions.checkArgument( + dataType != null, "\"%s\" field is required and cannot be null", fieldPath); + Preconditions.checkArgument( + !(dataType instanceof Types.UnparsedType), + "\"%s\" must be a Gravitino-recognized data type or an explicit ExternalType; " + + "UnparsedType is not allowed for new function definitions", + fieldPath); + + if (dataType instanceof Types.ExternalType) { + Types.ExternalType externalType = (Types.ExternalType) dataType; + Preconditions.checkArgument( + StringUtils.isNotBlank(externalType.catalogString()), + "\"%s.catalogString\" field is required and cannot be empty", + fieldPath); + return; + } + + if (dataType instanceof Types.StructType) { + Types.StructType.Field[] fields = ((Types.StructType) dataType).fields(); + Preconditions.checkArgument( + fields != null, "\"%s.fields\" field is required and cannot be null", fieldPath); + for (int i = 0; i < fields.length; i++) { + Preconditions.checkArgument( + fields[i] != null, + "\"%s.fields[%s]\" field is required and cannot be null", + fieldPath, + i); + validateDataType(fields[i].type(), fieldPath + ".fields[" + i + "].type"); + } + return; + } + + if (dataType instanceof Types.ListType) { + validateDataType(((Types.ListType) dataType).elementType(), fieldPath + ".elementType"); + return; + } + + if (dataType instanceof Types.MapType) { + Types.MapType mapType = (Types.MapType) dataType; + validateDataType(mapType.keyType(), fieldPath + ".keyType"); + validateDataType(mapType.valueType(), fieldPath + ".valueType"); + return; + } + + if (dataType instanceof Types.UnionType) { + Type[] types = ((Types.UnionType) dataType).types(); + Preconditions.checkArgument( + types != null, "\"%s.types\" field is required and cannot be null", fieldPath); + for (int i = 0; i < types.length; i++) { + validateDataType(types[i], fieldPath + ".types[" + i + "]"); + } + return; + } + + Preconditions.checkArgument( + dataType instanceof Type.PrimitiveType || dataType instanceof Types.NullType, + "\"%s\" must be a Gravitino-recognized data type or an explicit ExternalType", + fieldPath); + } +} diff --git a/common/src/main/java/org/apache/gravitino/dto/requests/FunctionRegisterRequest.java b/common/src/main/java/org/apache/gravitino/dto/requests/FunctionRegisterRequest.java index 24ea83b667..201fac2b86 100644 --- a/common/src/main/java/org/apache/gravitino/dto/requests/FunctionRegisterRequest.java +++ b/common/src/main/java/org/apache/gravitino/dto/requests/FunctionRegisterRequest.java @@ -72,15 +72,20 @@ public class FunctionRegisterRequest implements RESTRequest { "\"definitions\" field is required and cannot be empty"); // Validate each definition has appropriate return type/columns based on function type - for (FunctionDefinitionDTO definition : definitions) { + for (int i = 0; i < definitions.length; i++) { + FunctionDefinitionDTO definition = definitions[i]; + String definitionPath = "definitions[" + i + "]"; + FunctionDataTypeValidator.validateDefinition(definition, definitionPath); if (functionType == FunctionType.TABLE) { Preconditions.checkArgument( definition.getReturnColumns() != null && definition.getReturnColumns().length > 0, - "\"returnColumns\" is required in each definition for TABLE function type"); + "\"%s.returnColumns\" is required for TABLE function type", + definitionPath); } else if (functionType == FunctionType.SCALAR || functionType == FunctionType.AGGREGATE) { Preconditions.checkArgument( definition.getReturnType() != null, - "\"returnType\" is required in each definition for SCALAR or AGGREGATE function type"); + "\"%s.returnType\" is required for SCALAR or AGGREGATE function type", + definitionPath); } else { throw new IllegalArgumentException("Unsupported function type: " + functionType); } diff --git a/common/src/main/java/org/apache/gravitino/dto/requests/FunctionUpdateRequest.java b/common/src/main/java/org/apache/gravitino/dto/requests/FunctionUpdateRequest.java index f0cbdef3a4..c1d3618831 100644 --- a/common/src/main/java/org/apache/gravitino/dto/requests/FunctionUpdateRequest.java +++ b/common/src/main/java/org/apache/gravitino/dto/requests/FunctionUpdateRequest.java @@ -104,8 +104,11 @@ public interface FunctionUpdateRequest extends RESTRequest { @Override public void validate() throws IllegalArgumentException { - Preconditions.checkArgument( - definition != null, "\"definition\" field is required and cannot be null"); + validateDefinition("definition"); + } + + void validateDefinition(String fieldPath) { + FunctionDataTypeValidator.validateDefinition(definition, fieldPath); } } diff --git a/common/src/main/java/org/apache/gravitino/dto/requests/FunctionUpdatesRequest.java b/common/src/main/java/org/apache/gravitino/dto/requests/FunctionUpdatesRequest.java index 2f5468234f..7ae181c44b 100644 --- a/common/src/main/java/org/apache/gravitino/dto/requests/FunctionUpdatesRequest.java +++ b/common/src/main/java/org/apache/gravitino/dto/requests/FunctionUpdatesRequest.java @@ -25,7 +25,6 @@ import lombok.EqualsAndHashCode; import lombok.Getter; import lombok.NoArgsConstructor; import lombok.ToString; -import org.apache.gravitino.rest.RESTMessage; import org.apache.gravitino.rest.RESTRequest; /** Request to represent updates to a function. */ @@ -44,6 +43,18 @@ public class FunctionUpdatesRequest implements RESTRequest { if (updates == null) { throw new IllegalArgumentException("Updates list cannot be null"); } - updates.forEach(RESTMessage::validate); + for (int i = 0; i < updates.size(); i++) { + FunctionUpdateRequest update = updates.get(i); + if (update == null) { + throw new IllegalArgumentException("\"updates[" + i + "]\" field cannot be null"); + } + + if (update instanceof FunctionUpdateRequest.AddDefinitionRequest) { + ((FunctionUpdateRequest.AddDefinitionRequest) update) + .validateDefinition("updates[" + i + "].definition"); + } else { + update.validate(); + } + } } } diff --git a/common/src/test/java/org/apache/gravitino/dto/function/TestFunctionDTO.java b/common/src/test/java/org/apache/gravitino/dto/function/TestFunctionDTO.java index 5a414c602e..7f696d3ff0 100644 --- a/common/src/test/java/org/apache/gravitino/dto/function/TestFunctionDTO.java +++ b/common/src/test/java/org/apache/gravitino/dto/function/TestFunctionDTO.java @@ -24,6 +24,7 @@ import java.time.Instant; import java.util.Map; import org.apache.gravitino.dto.AuditDTO; import org.apache.gravitino.dto.rel.expressions.LiteralDTO; +import org.apache.gravitino.function.FunctionDefinition; import org.apache.gravitino.function.FunctionImpl; import org.apache.gravitino.function.FunctionType; import org.apache.gravitino.json.JsonUtils; @@ -290,4 +291,30 @@ public class TestFunctionDTO { Assertions.assertEquals(2, deserialized.returnColumns().length); Assertions.assertEquals("id", deserialized.returnColumns()[0].name()); } + + @Test + public void testLegacyUnparsedDataTypesRemainReadable() throws JsonProcessingException { + FunctionDTO function = + JsonUtils.objectMapper() + .readValue( + """ + { + "name": "legacy_function", + "functionType": "SCALAR", + "definitions": [{ + "parameters": [{"name": "input", "dataType": "legacy_parameter"}], + "returnType": "legacy_return", + "returnColumns": [{"name": "output", "dataType": "legacy_column"}] + }] + } + """, + FunctionDTO.class); + + FunctionDefinition definition = function.definitions()[0]; + Assertions.assertEquals( + Types.UnparsedType.of("legacy_parameter"), definition.parameters()[0].dataType()); + Assertions.assertEquals(Types.UnparsedType.of("legacy_return"), definition.returnType()); + Assertions.assertEquals( + Types.UnparsedType.of("legacy_column"), definition.returnColumns()[0].dataType()); + } } diff --git a/common/src/test/java/org/apache/gravitino/dto/requests/TestFunctionRegisterRequest.java b/common/src/test/java/org/apache/gravitino/dto/requests/TestFunctionRegisterRequest.java index ff206e12fa..7d62b6606b 100644 --- a/common/src/test/java/org/apache/gravitino/dto/requests/TestFunctionRegisterRequest.java +++ b/common/src/test/java/org/apache/gravitino/dto/requests/TestFunctionRegisterRequest.java @@ -227,4 +227,151 @@ public class TestFunctionRegisterRequest { // This should succeed - different definitions can have different return types Assertions.assertDoesNotThrow(request::validate); } + + @Test + public void testValidateNativeAndExternalDataTypes() { + FunctionParamDTO nativeParam = + FunctionParamDTO.builder() + .withName("native_param") + .withDataType(Types.IntegerType.get()) + .build(); + FunctionParamDTO externalParam = + FunctionParamDTO.builder() + .withName("external_param") + .withDataType(Types.ExternalType.of("engine_specific_param")) + .build(); + FunctionDefinitionDTO scalarDefinition = + FunctionDefinitionDTO.builder() + .withParameters(new FunctionParamDTO[] {nativeParam, externalParam}) + .withReturnType(Types.ExternalType.of("engine_specific_result")) + .build(); + FunctionRegisterRequest scalarRequest = + FunctionRegisterRequest.builder() + .withName("external_func") + .withFunctionType(FunctionType.SCALAR) + .withDefinitions(new FunctionDefinitionDTO[] {scalarDefinition}) + .build(); + + Assertions.assertDoesNotThrow(scalarRequest::validate); + + FunctionColumnDTO externalColumn = + FunctionColumnDTO.builder() + .withName("external_column") + .withDataType(Types.ExternalType.of("engine_specific_column")) + .build(); + FunctionDefinitionDTO tableDefinition = + FunctionDefinitionDTO.builder() + .withParameters(new FunctionParamDTO[0]) + .withReturnColumns(new FunctionColumnDTO[] {externalColumn}) + .build(); + FunctionRegisterRequest tableRequest = + FunctionRegisterRequest.builder() + .withName("external_table_func") + .withFunctionType(FunctionType.TABLE) + .withDefinitions(new FunctionDefinitionDTO[] {tableDefinition}) + .build(); + + Assertions.assertDoesNotThrow(tableRequest::validate); + } + + @Test + public void testRejectBlankExternalDataTypeWithFieldPath() { + FunctionDefinitionDTO definition = + FunctionDefinitionDTO.builder() + .withParameters(new FunctionParamDTO[0]) + .withReturnType(Types.ExternalType.of(" ")) + .build(); + FunctionRegisterRequest request = + FunctionRegisterRequest.builder() + .withName("invalid_external_func") + .withFunctionType(FunctionType.SCALAR) + .withDefinitions(new FunctionDefinitionDTO[] {definition}) + .build(); + + IllegalArgumentException exception = + Assertions.assertThrows(IllegalArgumentException.class, request::validate); + Assertions.assertTrue( + exception.getMessage().contains("definitions[0].returnType.catalogString")); + } + + @Test + public void testRejectUnknownParameterTypeWithFieldPath() throws JsonProcessingException { + assertInvalidDataType( + """ + { + "name": "test_func", + "functionType": "SCALAR", + "definitions": [{ + "parameters": [ + {"name": "known", "dataType": "integer"}, + {"name": "unknown", "dataType": "future_type"} + ], + "returnType": "integer" + }] + } + """, + "definitions[0].parameters[1].dataType"); + } + + @Test + public void testRejectPrimitiveTypeEncodedAsObjectWithFieldPath() throws JsonProcessingException { + assertInvalidDataType( + """ + { + "name": "test_func", + "functionType": "SCALAR", + "definitions": [{ + "parameters": [], + "returnType": {"type": "integer"} + }] + } + """, + "definitions[0].returnType"); + } + + @Test + public void testRejectExplicitUnparsedReturnColumnWithFieldPath() throws JsonProcessingException { + assertInvalidDataType( + """ + { + "name": "test_func", + "functionType": "TABLE", + "definitions": [{ + "parameters": [], + "returnColumns": [{ + "name": "result", + "dataType": {"type": "unparsed", "unparsedType": "legacy_type"} + }] + }] + } + """, + "definitions[0].returnColumns[0].dataType"); + } + + @Test + public void testRejectNestedUnparsedTypeWithFieldPath() throws JsonProcessingException { + assertInvalidDataType( + """ + { + "name": "test_func", + "functionType": "SCALAR", + "definitions": [{ + "parameters": [], + "returnType": {"type": "list", "elementType": "future_type"} + }] + } + """, + "definitions[0].returnType.elementType"); + } + + private static void assertInvalidDataType(String json, String expectedFieldPath) + throws JsonProcessingException { + FunctionRegisterRequest request = + JsonUtils.objectMapper().readValue(json, FunctionRegisterRequest.class); + + IllegalArgumentException exception = + Assertions.assertThrows(IllegalArgumentException.class, request::validate); + Assertions.assertTrue(exception.getMessage().contains('"' + expectedFieldPath + '"')); + Assertions.assertTrue(exception.getMessage().contains("UnparsedType is not allowed")); + } } diff --git a/common/src/test/java/org/apache/gravitino/dto/requests/TestFunctionUpdateRequest.java b/common/src/test/java/org/apache/gravitino/dto/requests/TestFunctionUpdateRequest.java index cfd49a0e02..8ef75e5c06 100644 --- a/common/src/test/java/org/apache/gravitino/dto/requests/TestFunctionUpdateRequest.java +++ b/common/src/test/java/org/apache/gravitino/dto/requests/TestFunctionUpdateRequest.java @@ -103,6 +103,84 @@ public class TestFunctionUpdateRequest { Assertions.assertThrows(IllegalArgumentException.class, invalidRequest::validate); } + @Test + public void testAddDefinitionRequestDataTypeValidation() { + FunctionParamDTO externalParam = + FunctionParamDTO.builder() + .withName("external") + .withDataType(Types.ExternalType.of("engine_specific_type")) + .build(); + FunctionDefinitionDTO externalDefinition = + FunctionDefinitionDTO.builder() + .withParameters(new FunctionParamDTO[] {externalParam}) + .withReturnType(Types.ExternalType.of("engine_specific_result")) + .build(); + + FunctionUpdateRequest.AddDefinitionRequest validRequest = + new FunctionUpdateRequest.AddDefinitionRequest(externalDefinition); + Assertions.assertDoesNotThrow(validRequest::validate); + + FunctionParamDTO unparsedParam = + FunctionParamDTO.builder() + .withName("legacy") + .withDataType(Types.UnparsedType.of("legacy_type")) + .build(); + FunctionDefinitionDTO unparsedDefinition = + FunctionDefinitionDTO.builder() + .withParameters(new FunctionParamDTO[] {unparsedParam}) + .withReturnType(Types.IntegerType.get()) + .build(); + + FunctionUpdateRequest.AddDefinitionRequest invalidRequest = + new FunctionUpdateRequest.AddDefinitionRequest(unparsedDefinition); + IllegalArgumentException exception = + Assertions.assertThrows(IllegalArgumentException.class, invalidRequest::validate); + Assertions.assertTrue(exception.getMessage().contains("definition.parameters[0].dataType")); + } + + @Test + public void testSelectorRequestsAllowLegacyUnparsedDataType() { + FunctionParamDTO legacyParam = + FunctionParamDTO.builder() + .withName("legacy") + .withDataType(Types.UnparsedType.of("legacy_type")) + .build(); + FunctionParamDTO[] legacyParams = new FunctionParamDTO[] {legacyParam}; + SQLImplDTO impl = new SQLImplDTO(FunctionImpl.RuntimeType.SPARK.name(), null, null, "SELECT 1"); + + FunctionUpdateRequest.RemoveDefinitionRequest removeDefinition = + new FunctionUpdateRequest.RemoveDefinitionRequest(legacyParams); + FunctionUpdateRequest.AddImplRequest addImpl = + new FunctionUpdateRequest.AddImplRequest(legacyParams, impl); + FunctionUpdateRequest.UpdateImplRequest updateImpl = + new FunctionUpdateRequest.UpdateImplRequest( + legacyParams, FunctionImpl.RuntimeType.SPARK.name(), impl); + FunctionUpdateRequest.RemoveImplRequest removeImpl = + new FunctionUpdateRequest.RemoveImplRequest( + legacyParams, FunctionImpl.RuntimeType.SPARK.name()); + + Assertions.assertDoesNotThrow(removeDefinition::validate); + Assertions.assertDoesNotThrow(addImpl::validate); + Assertions.assertDoesNotThrow(updateImpl::validate); + Assertions.assertDoesNotThrow(removeImpl::validate); + + FunctionChange.RemoveDefinition removeDefinitionChange = + (FunctionChange.RemoveDefinition) removeDefinition.functionChange(); + FunctionChange.AddImpl addImplChange = (FunctionChange.AddImpl) addImpl.functionChange(); + FunctionChange.UpdateImpl updateImplChange = + (FunctionChange.UpdateImpl) updateImpl.functionChange(); + FunctionChange.RemoveImpl removeImplChange = + (FunctionChange.RemoveImpl) removeImpl.functionChange(); + Assertions.assertEquals( + Types.UnparsedType.of("legacy_type"), removeDefinitionChange.parameters()[0].dataType()); + Assertions.assertEquals( + Types.UnparsedType.of("legacy_type"), addImplChange.parameters()[0].dataType()); + Assertions.assertEquals( + Types.UnparsedType.of("legacy_type"), updateImplChange.parameters()[0].dataType()); + Assertions.assertEquals( + Types.UnparsedType.of("legacy_type"), removeImplChange.parameters()[0].dataType()); + } + @Test public void testRemoveDefinitionRequestSerDe() throws JsonProcessingException { FunctionParamDTO param = diff --git a/common/src/test/java/org/apache/gravitino/dto/requests/TestFunctionUpdatesRequest.java b/common/src/test/java/org/apache/gravitino/dto/requests/TestFunctionUpdatesRequest.java index 9da5b2cedb..9e2b8c2443 100644 --- a/common/src/test/java/org/apache/gravitino/dto/requests/TestFunctionUpdatesRequest.java +++ b/common/src/test/java/org/apache/gravitino/dto/requests/TestFunctionUpdatesRequest.java @@ -24,6 +24,7 @@ import java.util.Collections; import org.apache.gravitino.dto.function.FunctionDefinitionDTO; import org.apache.gravitino.dto.function.FunctionParamDTO; import org.apache.gravitino.json.JsonUtils; +import org.apache.gravitino.rel.types.Types; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -79,4 +80,50 @@ public class TestFunctionUpdatesRequest { Assertions.assertThrows(IllegalArgumentException.class, request::validate); } + + @Test + public void testAddDefinitionValidationIncludesUpdatePath() throws JsonProcessingException { + FunctionUpdatesRequest request = + JsonUtils.objectMapper() + .readValue( + """ + { + "updates": [{ + "@type": "addDefinition", + "definition": { + "parameters": [{"name": "value", "dataType": "future_type"}], + "returnType": "integer" + } + }] + } + """, + FunctionUpdatesRequest.class); + + IllegalArgumentException exception = + Assertions.assertThrows(IllegalArgumentException.class, request::validate); + Assertions.assertTrue( + exception.getMessage().contains("updates[0].definition.parameters[0].dataType")); + } + + @Test + public void testSelectorValidationAllowsLegacyUnparsedDataType() throws JsonProcessingException { + FunctionUpdatesRequest request = + JsonUtils.objectMapper() + .readValue( + """ + { + "updates": [{ + "@type": "removeDefinition", + "parameters": [{"name": "value", "dataType": "future_type"}] + }] + } + """, + FunctionUpdatesRequest.class); + + Assertions.assertDoesNotThrow(request::validate); + FunctionUpdateRequest.RemoveDefinitionRequest update = + (FunctionUpdateRequest.RemoveDefinitionRequest) request.getUpdates().get(0); + Assertions.assertEquals( + Types.UnparsedType.of("future_type"), update.getParameters()[0].getDataType()); + } } diff --git a/core/src/test/java/org/apache/gravitino/catalog/TestManagedFunctionOperations.java b/core/src/test/java/org/apache/gravitino/catalog/TestManagedFunctionOperations.java index 60d5513f99..8d8f06cfdc 100644 --- a/core/src/test/java/org/apache/gravitino/catalog/TestManagedFunctionOperations.java +++ b/core/src/test/java/org/apache/gravitino/catalog/TestManagedFunctionOperations.java @@ -445,6 +445,56 @@ public class TestManagedFunctionOperations { Assertions.assertEquals(1, updatedFunc.definitions().length); } + @Test + public void testLegacyUnparsedDefinitionCanBeManagedBySelectors() { + NameIdentifier funcIdent = getFunctionIdent("legacy_unparsed_func"); + FunctionParam[] legacyParams = + new FunctionParam[] {FunctionParams.of("value", Types.UnparsedType.of("legacy_type"))}; + FunctionParam[] nativeParams = + new FunctionParam[] {FunctionParams.of("value", Types.IntegerType.get())}; + FunctionImpl sparkImpl = + FunctionImpls.ofJava(FunctionImpl.RuntimeType.SPARK, "com.example.LegacySparkUDF"); + FunctionDefinition[] legacyDefinitions = + new FunctionDefinition[] { + createDefinitionWithImpls( + legacyParams, Types.UnparsedType.of("legacy_return"), new FunctionImpl[] {sparkImpl}), + createSimpleDefinition(nativeParams, Types.StringType.get()) + }; + + // Seed metadata that could have been persisted before strict REST write validation. + functionOperations.registerFunction( + funcIdent, "Legacy function", FunctionType.SCALAR, true, legacyDefinitions); + + FunctionImpl trinoImpl = + FunctionImpls.ofJava(FunctionImpl.RuntimeType.TRINO, "com.example.LegacyTrinoUDF"); + Function afterAdd = + functionOperations.alterFunction( + funcIdent, FunctionChange.addImpl(legacyParams, trinoImpl)); + Assertions.assertEquals(2, afterAdd.definitions()[0].impls().length); + + FunctionImpl updatedTrinoImpl = + FunctionImpls.ofJava(FunctionImpl.RuntimeType.TRINO, "com.example.UpdatedTrinoUDF"); + Function afterUpdate = + functionOperations.alterFunction( + funcIdent, + FunctionChange.updateImpl( + legacyParams, FunctionImpl.RuntimeType.TRINO, updatedTrinoImpl)); + FunctionImpl updatedImpl = afterUpdate.definitions()[0].impls()[1]; + Assertions.assertInstanceOf(JavaImpl.class, updatedImpl); + Assertions.assertEquals("com.example.UpdatedTrinoUDF", ((JavaImpl) updatedImpl).className()); + + Function afterRemoveImpl = + functionOperations.alterFunction( + funcIdent, FunctionChange.removeImpl(legacyParams, FunctionImpl.RuntimeType.TRINO)); + Assertions.assertEquals(1, afterRemoveImpl.definitions()[0].impls().length); + + Function afterRemoveDefinition = + functionOperations.alterFunction(funcIdent, FunctionChange.removeDefinition(legacyParams)); + Assertions.assertEquals(1, afterRemoveDefinition.definitions().length); + Assertions.assertEquals( + Types.IntegerType.get(), afterRemoveDefinition.definitions()[0].parameters()[0].dataType()); + } + @Test public void testAlterFunctionRemoveOnlyDefinition() { NameIdentifier funcIdent = getFunctionIdent("func_remove_only"); diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/po/TestFunctionPO.java b/core/src/test/java/org/apache/gravitino/storage/relational/po/TestFunctionPO.java index efbd6bd005..437dc9e53d 100644 --- a/core/src/test/java/org/apache/gravitino/storage/relational/po/TestFunctionPO.java +++ b/core/src/test/java/org/apache/gravitino/storage/relational/po/TestFunctionPO.java @@ -18,6 +18,14 @@ */ package org.apache.gravitino.storage.relational.po; +import com.fasterxml.jackson.core.JsonProcessingException; +import java.time.Instant; +import org.apache.gravitino.Namespace; +import org.apache.gravitino.function.FunctionDefinition; +import org.apache.gravitino.json.JsonUtils; +import org.apache.gravitino.meta.AuditInfo; +import org.apache.gravitino.meta.FunctionEntity; +import org.apache.gravitino.rel.types.Types; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -90,6 +98,57 @@ public class TestFunctionPO { Assertions.assertEquals(1L, functionMaxVersionPO.version()); } + @Test + public void testLoadLegacyUnparsedFunctionDefinition() throws JsonProcessingException { + AuditInfo auditInfo = + AuditInfo.builder() + .withCreator("legacy-user") + .withCreateTime(Instant.parse("2025-01-01T00:00:00Z")) + .build(); + String auditJson = JsonUtils.anyFieldMapper().writeValueAsString(auditInfo); + String definitions = + """ + [{ + "parameters": [{"name": "input", "dataType": "legacy_parameter"}], + "returnType": "legacy_return", + "impls": [{"language": "SQL", "runtime": "SPARK", "sql": "SELECT input"}] + }] + """; + FunctionVersionPO versionPO = + FunctionVersionPO.builder() + .withFunctionId(1L) + .withMetalakeId(1L) + .withCatalogId(2L) + .withSchemaId(3L) + .withFunctionVersion(1) + .withDefinitions(definitions) + .withAuditInfo(auditJson) + .withDeletedAt(0L) + .build(); + FunctionPO functionPO = + FunctionPO.builder() + .withFunctionId(1L) + .withFunctionName("legacy_function") + .withMetalakeId(1L) + .withCatalogId(2L) + .withSchemaId(3L) + .withFunctionType("SCALAR") + .withDeterministic(1) + .withFunctionLatestVersion(1) + .withFunctionCurrentVersion(1) + .withAuditInfo(auditJson) + .withDeletedAt(0L) + .withFunctionVersionPO(versionPO) + .build(); + + FunctionEntity function = + FunctionPO.fromFunctionPO(functionPO, Namespace.of("metalake", "catalog", "schema")); + FunctionDefinition definition = function.definitions()[0]; + Assertions.assertEquals( + Types.UnparsedType.of("legacy_parameter"), definition.parameters()[0].dataType()); + Assertions.assertEquals(Types.UnparsedType.of("legacy_return"), definition.returnType()); + } + @Test public void testEqualsAndHashCode() { FunctionPO functionPO1 = diff --git a/docs/open-api/datatype.yaml b/docs/open-api/datatype.yaml index 213f439138..a4c3003534 100644 --- a/docs/open-api/datatype.yaml +++ b/docs/open-api/datatype.yaml @@ -20,6 +20,11 @@ components: schemas: DataType: + description: >- + The general, read-compatible data type representation. UnparsedType + preserves types that the current server cannot resolve when reading + existing metadata. Endpoints that require resolved new metadata use + WritableDataType instead. oneOf: - $ref: "#/components/schemas/PrimitiveType" - $ref: "#/components/schemas/StructType" @@ -29,6 +34,166 @@ components: - $ref: "#/components/schemas/UnparsedType" - $ref: "#/components/schemas/ExternalType" + WritableDataType: + description: >- + A data type accepted when writing resolved metadata. Use a native + Gravitino type, including recursively nested native types, or an + explicit ExternalType for a catalog- or engine-specific type. + UnparsedType is reserved for read compatibility and is not accepted. + This schema does not guarantee that a particular catalog supports the + represented type. + oneOf: + - $ref: "#/components/schemas/WritablePrimitiveType" + - $ref: "#/components/schemas/WritableStructType" + - $ref: "#/components/schemas/WritableListType" + - $ref: "#/components/schemas/WritableMapType" + - $ref: "#/components/schemas/WritableUnionType" + - $ref: "#/components/schemas/WritableExternalType" + + WritableExternalType: + type: object + description: >- + The explicit catalog- or engine-specific type form accepted when + writing resolved metadata. + required: + - type + - catalogString + properties: + type: + type: string + enum: + - "external" + catalogString: + type: string + minLength: 1 + pattern: '.*\S.*' + description: The non-blank string representation of this type in the catalog + + WritablePrimitiveType: + description: >- + The canonical string form of a Gravitino-recognized primitive type. + Parameters are validated by the server. + oneOf: + - type: string + enum: + - boolean + - byte + - byte unsigned + - short + - short unsigned + - integer + - integer unsigned + - long + - long unsigned + - float + - double + - date + - time + - timestamp + - timestamp_tz + - string + - binary + - uuid + - interval_day + - interval_year + - variant + - "null" + - geometry + - geography + - type: string + pattern: '^(decimal\(\s*[0-9]+\s*,\s*[0-9]+\s*\)|time\([0-9]+\)|timestamp\([0-9]+\)|timestamp_tz\([0-9]+\)|char\(\s*[0-9]+\s*\)|varchar\(\s*[0-9]+\s*\)|fixed\(\s*[0-9]+\s*\)|geometry\(\s*.+?\s*\)|geography\(\s*.+?\s*,\s*.+?\s*\))$' + example: "integer" + + WritableUnionType: + type: object + required: + - type + - types + properties: + type: + type: string + enum: + - "union" + types: + type: array + items: + $ref: "#/components/schemas/WritableDataType" + + WritableMapType: + type: object + required: + - type + - keyType + - valueType + properties: + type: + type: string + enum: + - "map" + keyType: + $ref: "#/components/schemas/WritableDataType" + valueType: + $ref: "#/components/schemas/WritableDataType" + valueContainsNull: + type: boolean + description: Whether the value of the map contains null values + nullable: true + default: true + + WritableListType: + type: object + required: + - type + - elementType + properties: + type: + type: string + enum: + - "list" + containsNull: + type: boolean + description: Whether the list contains null values + nullable: true + default: true + elementType: + $ref: "#/components/schemas/WritableDataType" + + WritableStructType: + type: object + required: + - type + - fields + properties: + type: + type: string + enum: + - "struct" + fields: + type: array + items: + $ref: "#/components/schemas/WritableStructField" + + WritableStructField: + type: object + required: + - name + - type + properties: + name: + type: string + description: The name of the struct field + type: + $ref: "#/components/schemas/WritableDataType" + nullable: + type: boolean + description: Whether the struct field is nullable + nullable: true + default: true + comment: + type: string + description: The comment of the struct field + nullable: true + PrimitiveType: type: string description: >- diff --git a/docs/open-api/functions.yaml b/docs/open-api/functions.yaml index 2201129bf3..5fde235573 100644 --- a/docs/open-api/functions.yaml +++ b/docs/open-api/functions.yaml @@ -65,6 +65,8 @@ paths: responses: "200": $ref: "#/components/responses/FunctionResponse" + "400": + $ref: "./openapi.yaml#/components/responses/BadRequestErrorResponse" "409": description: Conflict - The target function already exists content: @@ -208,6 +210,9 @@ components: FunctionDefinition: type: object + description: >- + A function definition returned by Gravitino. Its data types can include + UnparsedType when reading legacy or otherwise unresolved metadata. properties: parameters: type: array @@ -266,6 +271,69 @@ components: description: The comment of the return column nullable: true + WritableFunctionDefinition: + type: object + description: >- + A new function definition. Every parameter, return type, and return + column type must be a native Gravitino type or an explicit ExternalType. + properties: + parameters: + type: array + description: The parameters of the function definition + items: + $ref: "#/components/schemas/WritableFunctionParam" + returnType: + $ref: "./datatype.yaml#/components/schemas/WritableDataType" + description: The return type of the function (for SCALAR and AGGREGATE functions) + returnColumns: + type: array + description: The return columns of the function (for TABLE functions) + nullable: true + items: + $ref: "#/components/schemas/WritableFunctionColumn" + impls: + type: array + description: The implementations of the function definition + items: + $ref: "#/components/schemas/FunctionImpl" + + WritableFunctionParam: + type: object + required: + - name + - dataType + properties: + name: + type: string + description: The name of the parameter + dataType: + $ref: "./datatype.yaml#/components/schemas/WritableDataType" + description: The data type of the parameter + comment: + type: string + description: The comment of the parameter + nullable: true + defaultValue: + $ref: "./expression.yaml#/components/schemas/FunctionArg" + description: The default value expression of the parameter + + WritableFunctionColumn: + type: object + required: + - name + - dataType + properties: + name: + type: string + description: The name of the return column + dataType: + $ref: "./datatype.yaml#/components/schemas/WritableDataType" + description: The data type of the return column + comment: + type: string + description: The comment of the return column + nullable: true + FunctionImpl: type: object description: A function implementation, discriminated by the language field @@ -427,7 +495,7 @@ components: description: The function definitions. Must contain at least one definition. minItems: 1 items: - $ref: "#/components/schemas/FunctionDefinition" + $ref: "#/components/schemas/WritableFunctionDefinition" FunctionUpdatesRequest: type: object @@ -487,7 +555,7 @@ components: enum: - "addDefinition" definition: - $ref: "#/components/schemas/FunctionDefinition" + $ref: "#/components/schemas/WritableFunctionDefinition" example: { "@type": "addDefinition", "definition": { diff --git a/server/src/test/java/org/apache/gravitino/server/web/rest/TestFunctionOperations.java b/server/src/test/java/org/apache/gravitino/server/web/rest/TestFunctionOperations.java index 820874e935..17957719c4 100644 --- a/server/src/test/java/org/apache/gravitino/server/web/rest/TestFunctionOperations.java +++ b/server/src/test/java/org/apache/gravitino/server/web/rest/TestFunctionOperations.java @@ -266,6 +266,36 @@ public class TestFunctionOperations extends BaseOperationsTest { Assertions.assertEquals(RuntimeException.class.getSimpleName(), errorResp1.getType()); } + @Test + public void testGetLegacyFunctionWithUnparsedDataType() { + Function legacyFunction = mockFunction("legacy_func", "legacy comment", FunctionType.SCALAR); + FunctionParam[] params = + new FunctionParam[] {FunctionParams.of("input", Types.UnparsedType.of("legacy_input"))}; + FunctionImpl[] impls = + new FunctionImpl[] {FunctionImpls.ofSql(FunctionImpl.RuntimeType.SPARK, "SELECT input")}; + when(legacyFunction.definitions()) + .thenReturn( + new FunctionDefinition[] { + FunctionDefinitions.of(params, Types.UnparsedType.of("legacy_return"), impls) + }); + NameIdentifier funcId = NameIdentifierUtil.ofFunction(metalake, catalog, schema, "legacy_func"); + when(functionDispatcher.getFunction(funcId)).thenReturn(legacyFunction); + + Response response = + target(functionPath()) + .path("legacy_func") + .request(MediaType.APPLICATION_JSON_TYPE) + .accept("application/vnd.gravitino.v1+json") + .get(); + + Assertions.assertEquals(Response.Status.OK.getStatusCode(), response.getStatus()); + FunctionDefinition definition = + response.readEntity(FunctionResponse.class).getFunction().definitions()[0]; + Assertions.assertEquals( + Types.UnparsedType.of("legacy_input"), definition.parameters()[0].dataType()); + Assertions.assertEquals(Types.UnparsedType.of("legacy_return"), definition.returnType()); + } + @Test public void testRegisterFunctionWithNullRequest() { Response resp = @@ -277,6 +307,32 @@ public class TestFunctionOperations extends BaseOperationsTest { assertNullRequestBodyRejected(resp); } + @Test + public void testRegisterFunctionRejectsUnparsedDataType() { + String request = + """ + { + "name": "invalid_func", + "functionType": "SCALAR", + "definitions": [{ + "parameters": [{"name": "input", "dataType": "future_type"}], + "returnType": "integer" + }] + } + """; + + Response response = + target(functionPath()) + .request(MediaType.APPLICATION_JSON_TYPE) + .accept("application/vnd.gravitino.v1+json") + .post(Entity.entity(request, MediaType.APPLICATION_JSON_TYPE)); + + Assertions.assertEquals(Response.Status.BAD_REQUEST.getStatusCode(), response.getStatus()); + ErrorResponse error = response.readEntity(ErrorResponse.class); + Assertions.assertEquals(ErrorConstants.ILLEGAL_ARGUMENTS_CODE, error.getCode()); + Assertions.assertTrue(error.getMessage().contains("definitions[0].parameters[0].dataType")); + } + @Test public void testRegisterScalarFunction() { NameIdentifier funcId = NameIdentifierUtil.ofFunction(metalake, catalog, schema, "func1"); @@ -535,6 +591,34 @@ public class TestFunctionOperations extends BaseOperationsTest { Assertions.assertEquals(Response.Status.OK.getStatusCode(), resp.getStatus()); } + @Test + public void testAlterFunctionAddDefinitionRejectsUnparsedDataType() { + String request = + """ + { + "updates": [{ + "@type": "addDefinition", + "definition": { + "parameters": [], + "returnType": {"type": "unparsed", "unparsedType": "legacy_return"} + } + }] + } + """; + + Response response = + target(functionPath()) + .path("func1") + .request(MediaType.APPLICATION_JSON_TYPE) + .accept("application/vnd.gravitino.v1+json") + .put(Entity.entity(request, MediaType.APPLICATION_JSON_TYPE)); + + Assertions.assertEquals(Response.Status.BAD_REQUEST.getStatusCode(), response.getStatus()); + ErrorResponse error = response.readEntity(ErrorResponse.class); + Assertions.assertEquals(ErrorConstants.ILLEGAL_ARGUMENTS_CODE, error.getCode()); + Assertions.assertTrue(error.getMessage().contains("updates[0].definition.returnType")); + } + @Test public void testAlterFunctionRemoveDefinition() { NameIdentifier funcId = NameIdentifierUtil.ofFunction(metalake, catalog, schema, "func1"); @@ -562,6 +646,32 @@ public class TestFunctionOperations extends BaseOperationsTest { Assertions.assertEquals(Response.Status.OK.getStatusCode(), resp.getStatus()); } + @Test + public void testAlterFunctionSelectorAllowsLegacyUnparsedDataType() { + NameIdentifier funcId = NameIdentifierUtil.ofFunction(metalake, catalog, schema, "func1"); + Function mockFunction = mockFunction("func1", "comment", FunctionType.SCALAR); + when(functionDispatcher.alterFunction(eq(funcId), any(FunctionChange[].class))) + .thenReturn(mockFunction); + String request = + """ + { + "updates": [{ + "@type": "removeDefinition", + "parameters": [{"name": "input", "dataType": "legacy_type"}] + }] + } + """; + + Response response = + target(functionPath()) + .path("func1") + .request(MediaType.APPLICATION_JSON_TYPE) + .accept("application/vnd.gravitino.v1+json") + .put(Entity.entity(request, MediaType.APPLICATION_JSON_TYPE)); + + Assertions.assertEquals(Response.Status.OK.getStatusCode(), response.getStatus()); + } + @Test public void testAlterFunctionAddImpl() { NameIdentifier funcId = NameIdentifierUtil.ofFunction(metalake, catalog, schema, "func1");
