This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/main by this push:
new a50a8a180 [api][java][python] Render a flat RowTypeInfo output schema
as JSON Schema in the ReAct Agent prompt (#1270)
a50a8a180 is described below
commit a50a8a180d5ee0f6fa4f6259b098161c458b257c
Author: Sangkyoon Nam <[email protected]>
AuthorDate: Mon Oct 19 19:55:18 2026 +0900
[api][java][python] Render a flat RowTypeInfo output schema as JSON Schema
in the ReAct Agent prompt (#1270)
Generated-by: Claude Code 2.1.296 (Claude Opus 5.5)
---
.../apache/flink/agents/api/agents/ReActAgent.java | 6 +-
.../flink/agents/api/agents/RowTypeJsonSchema.java | 102 ++++++++++++++
.../flink/agents/api/agents/ReActAgentTest.java | 24 +++-
.../agents/api/agents/RowTypeJsonSchemaTest.java | 146 ++++++++++++++++++++
.../development/calling-llms/structured-output.md | 2 +-
python/flink_agents/api/agents/react_agent.py | 10 +-
.../api/agents/tests/test_react_agent.py | 30 +++-
.../tests/test_render_row_type_json_schema.py | 151 +++++++++++++++++++++
python/flink_agents/api/agents/types.py | 53 ++++++++
9 files changed, 512 insertions(+), 12 deletions(-)
diff --git
a/api/src/main/java/org/apache/flink/agents/api/agents/ReActAgent.java
b/api/src/main/java/org/apache/flink/agents/api/agents/ReActAgent.java
index 372a80d7c..6a48bff79 100644
--- a/api/src/main/java/org/apache/flink/agents/api/agents/ReActAgent.java
+++ b/api/src/main/java/org/apache/flink/agents/api/agents/ReActAgent.java
@@ -126,8 +126,10 @@ public class ReActAgent extends Agent {
if (outputSchema != null) {
String jsonSchema;
if (outputSchema instanceof RowTypeInfo) {
- jsonSchema = outputSchema.toString();
- outputSchema = new OutputSchema((RowTypeInfo) outputSchema);
+ RowTypeInfo rowType = (RowTypeInfo) outputSchema;
+ String rowSchema = RowTypeJsonSchema.render(rowType);
+ jsonSchema = rowSchema != null ? rowSchema :
rowType.toString();
+ outputSchema = new OutputSchema(rowType);
} else if (outputSchema instanceof Class) {
Class<?> schemaClass = (Class<?>) outputSchema;
try {
diff --git
a/api/src/main/java/org/apache/flink/agents/api/agents/RowTypeJsonSchema.java
b/api/src/main/java/org/apache/flink/agents/api/agents/RowTypeJsonSchema.java
new file mode 100644
index 000000000..bbc267f23
--- /dev/null
+++
b/api/src/main/java/org/apache/flink/agents/api/agents/RowTypeJsonSchema.java
@@ -0,0 +1,102 @@
+/*
+ * 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.flink.agents.api.agents;
+
+import com.fasterxml.jackson.core.JsonFactory;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.json.JsonWriteFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+
+import javax.annotation.Nullable;
+
+/**
+ * Renders a flat {@link RowTypeInfo} as a JSON Schema object, so a row output
schema reaches the
+ * model as JSON Schema rather than as the type's string form.
+ *
+ * <p>The Python {@code render_row_type_json_schema} produces the same text
for the same row, so
+ * both languages describe a row to the model identically.
+ */
+@Internal
+public final class RowTypeJsonSchema {
+
+ /** Writes lowercase hex digits in escapes, as Python's {@code json.dumps}
does. */
+ private static final ObjectMapper MAPPER =
+ new ObjectMapper(
+
JsonFactory.builder().disable(JsonWriteFeature.WRITE_HEX_UPPER_CASE).build());
+
+ private RowTypeJsonSchema() {}
+
+ /**
+ * Returns the JSON Schema of {@code rowType}: every field required, with
no additional
+ * properties.
+ *
+ * @return the schema as compact JSON, or {@code null} when a field has no
supported mapping in
+ * this renderer, such as a char, date or instant field, which the
caller then describes
+ * another way.
+ */
+ @Nullable
+ public static String render(RowTypeInfo rowType) {
+ ObjectNode schema = MAPPER.createObjectNode();
+ schema.put("type", "object");
+ ObjectNode properties = schema.putObject("properties");
+ ArrayNode required = MAPPER.createArrayNode();
+ String[] names = rowType.getFieldNames();
+ for (int i = 0; i < names.length; i++) {
+ String jsonType = jsonType(rowType.getTypeAt(i));
+ if (jsonType == null) {
+ return null;
+ }
+ properties.putObject(names[i]).put("type", jsonType);
+ required.add(names[i]);
+ }
+ schema.set("required", required);
+ schema.put("additionalProperties", false);
+ try {
+ return MAPPER.writeValueAsString(schema);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Nullable
+ private static String jsonType(TypeInformation<?> type) {
+ if (BasicTypeInfo.STRING_TYPE_INFO.equals(type)) {
+ return "string";
+ } else if (BasicTypeInfo.BOOLEAN_TYPE_INFO.equals(type)) {
+ return "boolean";
+ } else if (BasicTypeInfo.BYTE_TYPE_INFO.equals(type)
+ || BasicTypeInfo.SHORT_TYPE_INFO.equals(type)
+ || BasicTypeInfo.INT_TYPE_INFO.equals(type)
+ || BasicTypeInfo.LONG_TYPE_INFO.equals(type)
+ || BasicTypeInfo.BIG_INT_TYPE_INFO.equals(type)) {
+ return "integer";
+ } else if (BasicTypeInfo.FLOAT_TYPE_INFO.equals(type)
+ || BasicTypeInfo.DOUBLE_TYPE_INFO.equals(type)
+ || BasicTypeInfo.BIG_DEC_TYPE_INFO.equals(type)) {
+ return "number";
+ }
+ return null;
+ }
+}
diff --git
a/api/src/test/java/org/apache/flink/agents/api/agents/ReActAgentTest.java
b/api/src/test/java/org/apache/flink/agents/api/agents/ReActAgentTest.java
index b45613984..66dad94fb 100644
--- a/api/src/test/java/org/apache/flink/agents/api/agents/ReActAgentTest.java
+++ b/api/src/test/java/org/apache/flink/agents/api/agents/ReActAgentTest.java
@@ -128,10 +128,30 @@ public class ReActAgentTest {
}
@Test
- @DisplayName("An agent built on a RowTypeInfo schema registers the native
prompt with its text")
+ @DisplayName(
+ "An agent built on a RowTypeInfo schema registers the native
prompt with its JSON Schema")
public void testRowTypeSchemaRegistersNativeSchemaPrompt() {
assertThat(nativeSchemaPromptOf(agentWithSchema(rowTypeInfo())))
- .isEqualTo(NATIVE_SCHEMA_INSTRUCTION + rowTypeInfo());
+ .isEqualTo(NATIVE_SCHEMA_INSTRUCTION +
RowTypeJsonSchema.render(rowTypeInfo()));
+ }
+
+ @Test
+ @DisplayName("An agent built on a RowTypeInfo schema asks for its JSON
Schema in the prompt")
+ public void testRowTypeSchemaPromptCarriesJsonSchema() {
+ assertThat(schemaPromptOf(agentWithSchema(rowTypeInfo())))
+
.contains("{\"type\":\"object\",\"properties\":{\"count\":{\"type\":\"integer\"}}");
+ }
+
+ @Test
+ @DisplayName("A RowTypeInfo schema with a field no JSON Schema type fits
keeps its string form")
+ public void testRowTypeSchemaWithoutJsonTypeKeepsItsText() {
+ RowTypeInfo row =
+ new RowTypeInfo(
+ new TypeInformation[] {BasicTypeInfo.CHAR_TYPE_INFO},
+ new String[] {"initial"});
+ ReActAgent agent = agentWithSchema(row);
+
assertThat(nativeSchemaPromptOf(agent)).isEqualTo(NATIVE_SCHEMA_INSTRUCTION +
row);
+ assertThat(schemaPromptOf(agent)).contains(row.toString());
}
@Test
diff --git
a/api/src/test/java/org/apache/flink/agents/api/agents/RowTypeJsonSchemaTest.java
b/api/src/test/java/org/apache/flink/agents/api/agents/RowTypeJsonSchemaTest.java
new file mode 100644
index 000000000..da387df18
--- /dev/null
+++
b/api/src/test/java/org/apache/flink/agents/api/agents/RowTypeJsonSchemaTest.java
@@ -0,0 +1,146 @@
+/*
+ * 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.flink.agents.api.agents;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link RowTypeJsonSchema}. */
+class RowTypeJsonSchemaTest {
+
+ /** The same row's schema is asserted in the Python
test_render_row_type_json_schema.py. */
+ static final String MIXED_ROW_SCHEMA =
+
"{\"type\":\"object\",\"properties\":{\"name\":{\"type\":\"string\"},"
+ +
"\"count\":{\"type\":\"integer\"},\"score\":{\"type\":\"number\"},"
+ +
"\"ok\":{\"type\":\"boolean\"}},\"required\":[\"name\",\"count\",\"score\","
+ + "\"ok\"],\"additionalProperties\":false}";
+
+ /** The same row's schema is asserted in the Python
test_render_row_type_json_schema.py. */
+ static final String UNNAMED_ROW_SCHEMA =
+
"{\"type\":\"object\",\"properties\":{\"f0\":{\"type\":\"string\"},"
+ +
"\"f1\":{\"type\":\"integer\"}},\"required\":[\"f0\",\"f1\"],"
+ + "\"additionalProperties\":false}";
+
+ /** The same row's schema is asserted in the Python
test_render_row_type_json_schema.py. */
+ static final String ESCAPED_ROW_SCHEMA =
+
"{\"type\":\"object\",\"properties\":{\"a\\\"b\":{\"type\":\"string\"},"
+ +
"\"c\\\\d\":{\"type\":\"string\"},\"e\\u000bf\":{\"type\":\"string\"},"
+ +
"\"g\\u001fh\":{\"type\":\"string\"},\"i\\nj\":{\"type\":\"string\"},"
+ +
"\"k/l\":{\"type\":\"string\"},\"\uC810\uC218\":{\"type\":\"string\"},"
+ +
"\"m\u2028n\":{\"type\":\"string\"},\"\uD83D\uDE00\":{\"type\":\"string\"}},"
+ +
"\"required\":[\"a\\\"b\",\"c\\\\d\",\"e\\u000bf\",\"g\\u001fh\","
+ +
"\"i\\nj\",\"k/l\",\"\uC810\uC218\",\"m\u2028n\",\"\uD83D\uDE00\"],"
+ + "\"additionalProperties\":false}";
+
+ @Test
+ @DisplayName("A flat row renders as an object schema with every field
required, in field order")
+ void rendersFlatRowInFieldOrder() {
+ RowTypeInfo row =
+ new RowTypeInfo(
+ new TypeInformation[] {
+ BasicTypeInfo.STRING_TYPE_INFO,
+ BasicTypeInfo.INT_TYPE_INFO,
+ BasicTypeInfo.DOUBLE_TYPE_INFO,
+ BasicTypeInfo.BOOLEAN_TYPE_INFO
+ },
+ new String[] {"name", "count", "score", "ok"});
+
+ assertThat(RowTypeJsonSchema.render(row)).isEqualTo(MIXED_ROW_SCHEMA);
+ }
+
+ @Test
+ @DisplayName("Integral and decimal basic types map to integer and number")
+ void mapsNumericTypes() {
+ RowTypeInfo row =
+ new RowTypeInfo(
+ new TypeInformation[] {
+ BasicTypeInfo.BYTE_TYPE_INFO,
+ BasicTypeInfo.SHORT_TYPE_INFO,
+ BasicTypeInfo.LONG_TYPE_INFO,
+ BasicTypeInfo.BIG_INT_TYPE_INFO,
+ BasicTypeInfo.FLOAT_TYPE_INFO,
+ BasicTypeInfo.BIG_DEC_TYPE_INFO
+ },
+ new String[] {"b", "s", "l", "bi", "f", "bd"});
+
+ assertThat(RowTypeJsonSchema.render(row))
+ .isEqualTo(
+
"{\"type\":\"object\",\"properties\":{\"b\":{\"type\":\"integer\"},"
+ +
"\"s\":{\"type\":\"integer\"},\"l\":{\"type\":\"integer\"},"
+ +
"\"bi\":{\"type\":\"integer\"},\"f\":{\"type\":\"number\"},"
+ +
"\"bd\":{\"type\":\"number\"}},\"required\":[\"b\",\"s\",\"l\","
+ +
"\"bi\",\"f\",\"bd\"],\"additionalProperties\":false}");
+ }
+
+ @Test
+ @DisplayName("A field with no supported mapping leaves the row unrendered")
+ void returnsNullForFieldsWithoutJsonType() {
+ for (TypeInformation<?> type :
+ new TypeInformation<?>[] {
+ BasicTypeInfo.CHAR_TYPE_INFO,
+ BasicTypeInfo.DATE_TYPE_INFO,
+ BasicTypeInfo.INSTANT_TYPE_INFO,
+ BasicTypeInfo.VOID_TYPE_INFO
+ }) {
+ RowTypeInfo row =
+ new RowTypeInfo(
+ new TypeInformation[]
{BasicTypeInfo.STRING_TYPE_INFO, type},
+ new String[] {"name", "other"});
+
+
assertThat(RowTypeJsonSchema.render(row)).as(type.toString()).isNull();
+ }
+ }
+
+ @Test
+ @DisplayName("Escaped field names render the same text as in Python")
+ void escapesFieldNamesLikePython() {
+ String[] names = {
+ "a\"b",
+ "c\\d",
+ "e\u000Bf",
+ "g\u001Fh",
+ "i\nj",
+ "k/l",
+ "\uC810\uC218",
+ "m\u2028n",
+ "\uD83D\uDE00"
+ };
+ TypeInformation<?>[] types = new TypeInformation<?>[names.length];
+ Arrays.fill(types, BasicTypeInfo.STRING_TYPE_INFO);
+
+ assertThat(RowTypeJsonSchema.render(new RowTypeInfo(types, names)))
+ .isEqualTo(ESCAPED_ROW_SCHEMA);
+ }
+
+ @Test
+ @DisplayName("A row built without names renders the default f0, f1, ...
names")
+ void rendersDefaultFieldNames() {
+ RowTypeInfo row =
+ new RowTypeInfo(BasicTypeInfo.STRING_TYPE_INFO,
BasicTypeInfo.INT_TYPE_INFO);
+
+
assertThat(RowTypeJsonSchema.render(row)).isEqualTo(UNNAMED_ROW_SCHEMA);
+ }
+}
diff --git a/docs/content/docs/development/calling-llms/structured-output.md
b/docs/content/docs/development/calling-llms/structured-output.md
index abb3a06f2..e725823ec 100644
--- a/docs/content/docs/development/calling-llms/structured-output.md
+++ b/docs/content/docs/development/calling-llms/structured-output.md
@@ -42,7 +42,7 @@ A schema is one of:
| Typed class | Pydantic `BaseModel` subclass | POJO `Class` |
| Flink row type | `RowTypeInfo` | `RowTypeInfo` |
-Only a typed class can be applied natively. A `RowTypeInfo` always uses the
prompt path.
+Only a typed class can be applied natively. A `RowTypeInfo` always uses the
prompt path. For a flat `RowTypeInfo` whose fields are all string, boolean or
numeric types, the ReAct Agent describes the schema to the model as JSON Schema
with every field required and `additionalProperties: false`. Other row types
keep their existing behavior.
Pass the schema to a [ReAct Agent]({{< ref "docs/development/react_agent" >}})
with `output_schema`, or put it on a `ChatRequestEvent` when you [send chat
requests yourself]({{< ref "docs/development/calling-llms/chat" >}}).
diff --git a/python/flink_agents/api/agents/react_agent.py
b/python/flink_agents/api/agents/react_agent.py
index c7052a949..8c43483e5 100644
--- a/python/flink_agents/api/agents/react_agent.py
+++ b/python/flink_agents/api/agents/react_agent.py
@@ -27,7 +27,11 @@ from pyflink.common import Row
from pyflink.common.typeinfo import RowTypeInfo
from flink_agents.api.agents.agent import Agent
-from flink_agents.api.agents.types import OutputSchema, render_output_schema
+from flink_agents.api.agents.types import (
+ OutputSchema,
+ render_output_schema,
+ render_row_type_json_schema,
+)
from flink_agents.api.chat_message import (
ChatMessage,
MessageRole,
@@ -236,7 +240,9 @@ class ReActAgent(Agent):
output_schema, lambda model: model.model_json_schema()
)
elif isinstance(output_schema, RowTypeInfo):
- json_schema = str(output_schema)
+ json_schema = render_row_type_json_schema(output_schema) or
str(
+ output_schema
+ )
else:
err_msg = f"Output schema {output_schema.__class__} is not
supported."
raise TypeError(err_msg)
diff --git a/python/flink_agents/api/agents/tests/test_react_agent.py
b/python/flink_agents/api/agents/tests/test_react_agent.py
index 0fc71a444..0556214b5 100644
--- a/python/flink_agents/api/agents/tests/test_react_agent.py
+++ b/python/flink_agents/api/agents/tests/test_react_agent.py
@@ -28,6 +28,7 @@ from flink_agents.api.agents.react_agent import (
_DEFAULT_SCHEMA_PROMPT,
ReActAgent,
)
+from flink_agents.api.agents.types import render_row_type_json_schema
from flink_agents.api.chat_message import MessageRole
from flink_agents.api.events.event import InputEvent
from flink_agents.api.prompts.prompt import Prompt
@@ -234,10 +235,29 @@ def
test_renderable_output_schema_keeps_the_schema_prompt() -> None:
)
-def test_row_type_info_output_schema_keeps_the_prompt_fallback() -> None:
- """A RowTypeInfo has no JSON Schema render and keeps its own prompt
text."""
+def test_row_type_info_output_schema_renders_json_schema_in_the_prompt() ->
None:
+ """A flat RowTypeInfo reaches the prompt as its JSON Schema."""
row_type = Types.ROW_NAMED(["name"], [Types.STRING()])
- assert _schema_prompt(_agent(row_type)) == _expected_prompt(row_type)
+ assert _schema_prompt(_agent(row_type)) == _expected_prompt(
+ '{"type":"object","properties":{"name":{"type":"string"}},'
+ '"required":["name"],"additionalProperties":false}'
+ )
+
+
+def test_row_type_info_without_json_type_keeps_the_prompt_fallback() -> None:
+ """A row with a field no JSON Schema type fits keeps its own prompt
text."""
+ row_type = Types.ROW_NAMED(["initial"], [Types.CHAR()])
+ agent = _agent(row_type)
+ assert _schema_prompt(agent) == _expected_prompt(row_type)
+ assert _native_schema_prompt(agent) ==
f"{_NATIVE_SCHEMA_INSTRUCTION}{row_type}"
+
+
+def test_row_type_info_with_duplicate_names_keeps_the_prompt_fallback() ->
None:
+ """A row with duplicate field names keeps its own prompt text."""
+ row_type = Types.ROW_NAMED(["x", "x"], [Types.STRING(), Types.INT()])
+ agent = _agent(row_type)
+ assert _schema_prompt(agent) == _expected_prompt(row_type)
+ assert _native_schema_prompt(agent) ==
f"{_NATIVE_SCHEMA_INSTRUCTION}{row_type}"
def test_renderable_output_schema_registers_the_native_schema_prompt() -> None:
@@ -257,11 +277,11 @@ def
test_native_schema_prompt_shows_field_descriptions_and_required_fields() ->
def test_row_type_info_output_schema_registers_the_native_schema_prompt() ->
None:
- """A RowTypeInfo reaches the native prompt as its own text."""
+ """A flat RowTypeInfo reaches the native prompt as its JSON Schema."""
row_type = Types.ROW_NAMED(["name"], [Types.STRING()])
assert (
_native_schema_prompt(_agent(row_type))
- == f"{_NATIVE_SCHEMA_INSTRUCTION}{row_type}"
+ ==
f"{_NATIVE_SCHEMA_INSTRUCTION}{render_row_type_json_schema(row_type)}"
)
diff --git
a/python/flink_agents/api/agents/tests/test_render_row_type_json_schema.py
b/python/flink_agents/api/agents/tests/test_render_row_type_json_schema.py
new file mode 100644
index 000000000..62efb5453
--- /dev/null
+++ b/python/flink_agents/api/agents/tests/test_render_row_type_json_schema.py
@@ -0,0 +1,151 @@
+################################################################################
+# 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 pytest
+from pyflink.common.typeinfo import RowTypeInfo, Types
+
+from flink_agents.api.agents.types import render_row_type_json_schema
+
+# The same row's schema is asserted in the Java RowTypeJsonSchemaTest.
+_MIXED_ROW_SCHEMA = (
+ '{"type":"object","properties":{"name":{"type":"string"},'
+ '"count":{"type":"integer"},"score":{"type":"number"},'
+ '"ok":{"type":"boolean"}},"required":["name","count","score",'
+ '"ok"],"additionalProperties":false}'
+)
+
+# The same row's schema is asserted in the Java RowTypeJsonSchemaTest.
+_UNNAMED_ROW_SCHEMA = (
+ '{"type":"object","properties":{"f0":{"type":"string"},'
+ '"f1":{"type":"integer"}},"required":["f0","f1"],'
+ '"additionalProperties":false}'
+)
+
+# The same row's schema is asserted in the Java RowTypeJsonSchemaTest.
+_ESCAPED_ROW_SCHEMA = (
+ '{"type":"object","properties":{"a\\"b":{"type":"string"},'
+ '"c\\\\d":{"type":"string"},"e\\u000bf":{"type":"string"},'
+ '"g\\u001fh":{"type":"string"},"i\\nj":{"type":"string"},'
+ '"k/l":{"type":"string"},"\uc810\uc218":{"type":"string"},'
+ '"m\u2028n":{"type":"string"},"\U0001f600":{"type":"string"}},'
+ '"required":["a\\"b","c\\\\d","e\\u000bf","g\\u001fh",'
+ '"i\\nj","k/l","\uc810\uc218","m\u2028n","\U0001f600"],'
+ '"additionalProperties":false}'
+)
+
+
+def test_renders_flat_row_in_field_order() -> None:
+ """A flat row renders as an object schema with every field required."""
+ row = RowTypeInfo(
+ [Types.STRING(), Types.INT(), Types.DOUBLE(), Types.BOOLEAN()],
+ ["name", "count", "score", "ok"],
+ )
+
+ assert render_row_type_json_schema(row) == _MIXED_ROW_SCHEMA
+
+
+def test_maps_numeric_types() -> None:
+ """Integral and decimal basic types map to integer and number."""
+ row = RowTypeInfo(
+ [
+ Types.BYTE(),
+ Types.SHORT(),
+ Types.LONG(),
+ Types.BIG_INT(),
+ Types.FLOAT(),
+ Types.BIG_DEC(),
+ ],
+ ["b", "s", "l", "bi", "f", "bd"],
+ )
+
+ assert render_row_type_json_schema(row) == (
+ '{"type":"object","properties":{"b":{"type":"integer"},'
+ '"s":{"type":"integer"},"l":{"type":"integer"},'
+ '"bi":{"type":"integer"},"f":{"type":"number"},'
+ '"bd":{"type":"number"}},"required":["b","s","l",'
+ '"bi","f","bd"],"additionalProperties":false}'
+ )
+
+
[email protected](
+ "field_type", [Types.CHAR(), Types.INSTANT(), Types.SQL_DATE()]
+)
+def test_returns_none_for_fields_without_json_type(field_type: object) -> None:
+ """A field with no supported mapping leaves the row unrendered."""
+ row = RowTypeInfo([Types.STRING(), field_type], ["name", "other"])
+
+ assert render_row_type_json_schema(row) is None
+
+
+def test_escapes_field_names_like_java() -> None:
+ """Escaped field names render the same text as in Java."""
+ names = [
+ 'a"b',
+ "c\\d",
+ "e\x0bf",
+ "g\x1fh",
+ "i\nj",
+ "k/l",
+ "\uc810\uc218",
+ "m\u2028n",
+ "\U0001f600",
+ ]
+ row = RowTypeInfo([Types.STRING()] * len(names), names)
+
+ assert render_row_type_json_schema(row) == _ESCAPED_ROW_SCHEMA
+
+
+def _without_jvm(row: RowTypeInfo, monkeypatch: pytest.MonkeyPatch) ->
RowTypeInfo:
+ def fail() -> None:
+ msg = "the renderer must not start the JVM"
+ raise AssertionError(msg)
+
+ monkeypatch.setattr(row, "get_java_type_info", fail)
+ return row
+
+
+def test_unnamed_row_uses_default_names_without_the_jvm(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ """A row built without names renders f0, f1, ... as Java does."""
+ row = _without_jvm(RowTypeInfo([Types.STRING(), Types.INT()]), monkeypatch)
+
+ assert render_row_type_json_schema(row) == _UNNAMED_ROW_SCHEMA
+
+
+def test_unnamed_unmapped_row_returns_none_without_the_jvm(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ """An unnamed row with an unmapped field is left unrendered."""
+ row = _without_jvm(RowTypeInfo([Types.CHAR()]), monkeypatch)
+
+ assert render_row_type_json_schema(row) is None
+
+
[email protected](
+ ("names", "types"),
+ [
+ ([], [Types.STRING()]),
+ (["a"], [Types.STRING(), Types.INT()]),
+ (["x", "x"], [Types.STRING(), Types.INT()]),
+ ],
+ ids=["empty-names", "fewer-names", "duplicate-names"],
+)
+def test_returns_none_for_names_java_rejects(names: list[str], types: list) ->
None:
+ """Names Java would reject leave the row unrendered."""
+ assert render_row_type_json_schema(RowTypeInfo(types, names)) is None
diff --git a/python/flink_agents/api/agents/types.py
b/python/flink_agents/api/agents/types.py
index 381f0da1e..cc1528356 100644
--- a/python/flink_agents/api/agents/types.py
+++ b/python/flink_agents/api/agents/types.py
@@ -16,6 +16,7 @@
# limitations under the License.
#################################################################################
import importlib
+import json
from typing import Any, Callable
from pydantic import BaseModel, ConfigDict, model_serializer, model_validator
@@ -179,3 +180,55 @@ def _admits_undeclared_keys(schema: Any) -> bool:
return _admits_undeclared_keys(schema.get("items")) or
_admits_undeclared_keys(
schema.get("not")
)
+
+
+_ROW_JSON_SCHEMA_TYPES = {
+ BasicType.STRING: "string",
+ BasicType.BOOLEAN: "boolean",
+ BasicType.BYTE: "integer",
+ BasicType.SHORT: "integer",
+ BasicType.INT: "integer",
+ BasicType.LONG: "integer",
+ BasicType.BIG_INT: "integer",
+ BasicType.FLOAT: "number",
+ BasicType.DOUBLE: "number",
+ BasicType.BIG_DEC: "number",
+}
+
+
+def render_row_type_json_schema(row_type: RowTypeInfo) -> str | None:
+ """Render a flat ``RowTypeInfo`` as a JSON Schema object.
+
+ Every field is required and no additional properties are allowed. The Java
+ ``RowTypeJsonSchema`` produces the same text for the same row, so both
+ languages describe a row to the model identically.
+
+ Returns ``None`` when a field has no supported mapping in this renderer,
+ such as a char or instant field, or when the names are duplicated or don't
+ match the types; the caller then describes the row another way.
+ """
+ json_types = [
+ _ROW_JSON_SCHEMA_TYPES.get(getattr(field_type, "_basic_type", None))
+ for field_type in row_type.get_field_types()
+ ]
+ if None in json_types:
+ return None
+ # get_field_names() starts the JVM for a row built without names; Java
names
+ # those fields f0, f1, ... too.
+ names = row_type._field_names
+ if names is None:
+ names = [f"f{i}" for i in range(len(json_types))]
+ # Java rejects these rows; Python keeps describing them another way.
+ if len(names) != len(json_types) or len(set(names)) != len(names):
+ return None
+ properties = {
+ name: {"type": json_type}
+ for name, json_type in zip(names, json_types, strict=True)
+ }
+ schema = {
+ "type": "object",
+ "properties": properties,
+ "required": list(properties),
+ "additionalProperties": False,
+ }
+ return json.dumps(schema, separators=(",", ":"), ensure_ascii=False)
