This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new f9220300856 [FLINK-41001][model] Accept compatible string inputs in
the OpenAI provider (#29486)
f9220300856 is described below
commit f922030085606c220430f97a0efb3254d274b49e
Author: Liu Liu <[email protected]>
AuthorDate: Thu Oct 15 22:23:42 2026 +0800
[FLINK-41001][model] Accept compatible string inputs in the OpenAI provider
(#29486)
Generated-by: Codex (GPT-6)
---
.../model/openai/AbstractOpenAIModelFunction.java | 15 ++++-
.../flink/model/openai/OpenAIChatModelTest.java | 67 ++++++++++++++++++++--
.../model/openai/OpenAIEmbeddingModelTest.java | 23 ++++++++
3 files changed, 98 insertions(+), 7 deletions(-)
diff --git
a/flink-models/flink-model-openai/src/main/java/org/apache/flink/model/openai/AbstractOpenAIModelFunction.java
b/flink-models/flink-model-openai/src/main/java/org/apache/flink/model/openai/AbstractOpenAIModelFunction.java
index 193da8a72e7..fa9cb1ef3a5 100644
---
a/flink-models/flink-model-openai/src/main/java/org/apache/flink/model/openai/AbstractOpenAIModelFunction.java
+++
b/flink-models/flink-model-openai/src/main/java/org/apache/flink/model/openai/AbstractOpenAIModelFunction.java
@@ -36,6 +36,7 @@ import org.apache.flink.table.functions.AsyncPredictFunction;
import org.apache.flink.table.functions.FunctionContext;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.Preconditions;
@@ -56,6 +57,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function;
+import java.util.function.Predicate;
import java.util.stream.Collectors;
import static org.apache.flink.configuration.description.TextElement.text;
@@ -95,7 +97,8 @@ public abstract class AbstractOpenAIModelFunction extends
AsyncPredictFunction {
validateSingleColumnSchema(
factoryContext.getCatalogModel().getResolvedInputSchema(),
new VarCharType(VarCharType.MAX_LENGTH),
- "input");
+ "input",
+ type -> type.is(LogicalTypeFamily.CHARACTER_STRING));
this.outputColumnNames =
factoryContext.getCatalogModel().getResolvedOutputSchema().getColumnNames();
@@ -142,6 +145,14 @@ public abstract class AbstractOpenAIModelFunction extends
AsyncPredictFunction {
protected void validateSingleColumnSchema(
ResolvedSchema schema, LogicalType expectedType, String
inputOrOutput) {
+ validateSingleColumnSchema(schema, expectedType, inputOrOutput,
expectedType::equals);
+ }
+
+ private void validateSingleColumnSchema(
+ ResolvedSchema schema,
+ LogicalType expectedType,
+ String inputOrOutput,
+ Predicate<LogicalType> supportedType) {
List<Column> columns = schema.getColumns();
List<String> physicalColumnNames =
columns.stream()
@@ -156,7 +167,7 @@ public abstract class AbstractOpenAIModelFunction extends
AsyncPredictFunction {
}
Column column = schema.getColumn(physicalColumnNames.get(0)).get();
- if (!expectedType.equals(column.getDataType().getLogicalType())) {
+ if (!supportedType.test(column.getDataType().getLogicalType())) {
throw new IllegalArgumentException(
String.format(
"%s column %s should be %s, but is a %s.",
diff --git
a/flink-models/flink-model-openai/src/test/java/org/apache/flink/model/openai/OpenAIChatModelTest.java
b/flink-models/flink-model-openai/src/test/java/org/apache/flink/model/openai/OpenAIChatModelTest.java
index 835ac2e7beb..03c0cb24380 100644
---
a/flink-models/flink-model-openai/src/test/java/org/apache/flink/model/openai/OpenAIChatModelTest.java
+++
b/flink-models/flink-model-openai/src/test/java/org/apache/flink/model/openai/OpenAIChatModelTest.java
@@ -19,13 +19,17 @@ package org.apache.flink.model.openai;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.ModelDescriptor;
import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.internal.TableEnvironmentImpl;
import org.apache.flink.table.catalog.CatalogManager;
import org.apache.flink.table.catalog.CatalogModel;
import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.types.ColumnList;
import org.apache.flink.types.Row;
import org.apache.flink.types.variant.Variant;
@@ -42,6 +46,8 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
import java.io.IOException;
import java.util.ArrayList;
@@ -49,6 +55,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -246,6 +253,46 @@ public class OpenAIChatModelTest {
assertThat(result).isEmpty();
}
+ @ParameterizedTest
+ @MethodSource("stringInputTypes")
+ public void testStringInputSchema(final DataType inputType) {
+ final Schema inputSchema = Schema.newBuilder().column("input",
inputType).build();
+ final Table input =
+ tEnv.fromValues(
+ DataTypes.ROW(DataTypes.FIELD("input", inputType)),
Row.of("hello"));
+ tEnv.createTemporaryView("TypedInput", input);
+ createModel(inputSchema, OUTPUT_SCHEMA);
+
+ assertThat(
+ tEnv.sqlQuery(
+ String.format(
+ "SELECT content FROM
ML_PREDICT(TABLE TypedInput, MODEL %s, DESCRIPTOR(`input`))",
+ MODEL_NAME))
+ .explain())
+ .isNotBlank();
+
+ final ModelDescriptor.Builder descriptor =
+ ModelDescriptor.forProvider("openai")
+ .inputSchema(inputSchema)
+ .outputSchema(OUTPUT_SCHEMA);
+ modelOptions.forEach(descriptor::option);
+ assertThat(
+ tEnv.fromModel(descriptor.build())
+ .predict(input, ColumnList.of("input"))
+ .explain())
+ .isNotBlank();
+ }
+
+ private static Stream<DataType> stringInputTypes() {
+ return Stream.of(
+ DataTypes.CHAR(14),
+ DataTypes.CHAR(14).notNull(),
+ DataTypes.VARCHAR(32),
+ DataTypes.VARCHAR(32).notNull(),
+ DataTypes.STRING(),
+ DataTypes.STRING().notNull());
+ }
+
@Test
public void testInvalidInputSchema() {
Schema inputSchemaWithInvalidColumnType =
@@ -263,10 +310,11 @@ public class OpenAIChatModelTest {
.hasMessageContainingAll("input", "DOUBLE", "STRING");
}
- @Test
- public void testInvalidOutputSchema() {
- Schema outputSchemaWithInvalidColumnType =
- Schema.newBuilder().column("output",
DataTypes.DOUBLE()).build();
+ @ParameterizedTest
+ @MethodSource("invalidOutputTypes")
+ public void testInvalidOutputSchema(final DataType outputType) {
+ final Schema outputSchemaWithInvalidColumnType =
+ Schema.newBuilder().column("output", outputType).build();
createModel(INPUT_SCHEMA, outputSchemaWithInvalidColumnType);
assertThatThrownBy(
@@ -277,7 +325,16 @@ public class OpenAIChatModelTest {
MODEL_NAME)))
.rootCause()
.isInstanceOf(IllegalArgumentException.class)
- .hasMessageContainingAll("output", "DOUBLE", "STRING");
+ .hasMessageContainingAll(
+ "output",
outputType.getLogicalType().asSummaryString(), "STRING");
+ }
+
+ private static Stream<DataType> invalidOutputTypes() {
+ return Stream.of(
+ DataTypes.DOUBLE(),
+ DataTypes.CHAR(14),
+ DataTypes.VARCHAR(32),
+ DataTypes.STRING().notNull());
}
private void createModel() {
diff --git
a/flink-models/flink-model-openai/src/test/java/org/apache/flink/model/openai/OpenAIEmbeddingModelTest.java
b/flink-models/flink-model-openai/src/test/java/org/apache/flink/model/openai/OpenAIEmbeddingModelTest.java
index 090e82e4922..ef420a8e873 100644
---
a/flink-models/flink-model-openai/src/test/java/org/apache/flink/model/openai/OpenAIEmbeddingModelTest.java
+++
b/flink-models/flink-model-openai/src/test/java/org/apache/flink/model/openai/OpenAIEmbeddingModelTest.java
@@ -19,13 +19,17 @@ package org.apache.flink.model.openai;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.ModelDescriptor;
import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.internal.TableEnvironmentImpl;
import org.apache.flink.table.catalog.CatalogManager;
import org.apache.flink.table.catalog.CatalogModel;
import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.types.ColumnList;
import org.apache.flink.types.Row;
import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
@@ -229,6 +233,25 @@ public class OpenAIEmbeddingModelTest {
assertThat(result).isEmpty();
}
+ @Test
+ public void testBoundedNonNullStringInputSchema() {
+ final DataType inputType = DataTypes.VARCHAR(32).notNull();
+ final Table input =
+ tEnv.fromValues(
+ DataTypes.ROW(DataTypes.FIELD("input", inputType)),
Row.of("hello"));
+ final ModelDescriptor.Builder descriptor =
+ ModelDescriptor.forProvider("openai")
+ .inputSchema(Schema.newBuilder().column("input",
inputType).build())
+ .outputSchema(OUTPUT_SCHEMA);
+ modelOptions.forEach(descriptor::option);
+
+ assertThat(
+ tEnv.fromModel(descriptor.build())
+ .predict(input, ColumnList.of("input"))
+ .explain())
+ .isNotBlank();
+ }
+
@Test
public void testInvalidInputSchema() {
CatalogManager catalogManager = ((TableEnvironmentImpl)
tEnv).getCatalogManager();
