feat: Handle FlinkJsonType decoding in flexible-json format

formats/flexible-json-format/src/main/java/com/datasqrl/flinkrunner/format/json/FlexibleJsonFormat.java

@@ -30,7 +30,6 @@
 import org.apache.flink.formats.common.TimestampFormat;
 import org.apache.flink.formats.json.JsonFormatFactory;
 import org.apache.flink.formats.json.JsonFormatOptionsUtil;
-import org.apache.flink.formats.json.JsonRowDataDeserializationSchema;
 import org.apache.flink.table.connector.ChangelogMode;
 import org.apache.flink.table.connector.Projection;
 import org.apache.flink.table.connector.format.DecodingFormat;
@@ -74,8 +73,8 @@ public DeserializationSchema<RowData> createRuntimeDecoder(
         final RowType rowType = (RowType) producedDataType.getLogicalType();
         final TypeInformation<RowData> rowDataTypeInfo =
             context.createTypeInformation(producedDataType);
-        JsonRowDataDeserializationSchema jsonRowDataDeserializationSchema =
-            new JsonRowDataDeserializationSchema(
+        var jsonRowDataDeserializationSchema =
+            new SqrlJsonRowDataDeserializationSchema(
                 rowType, rowDataTypeInfo, false, false, TimestampFormat.ISO_8601);
         return jsonRowDataDeserializationSchema;
       }

formats/flexible-json-format/src/main/java/com/datasqrl/flinkrunner/format/json/SqrlJsonRowDataDeserializationSchema.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright © 2026 DataSQRL (contact@datasqrl.com)
+ *
+ * Licensed 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 com.datasqrl.flinkrunner.format.json;
+
+import java.io.IOException;
+import java.io.Serial;
+import lombok.NonNull;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.formats.common.TimestampFormat;
+import org.apache.flink.formats.json.AbstractJsonDeserializationSchema;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ArrayNode;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.Collector;
+
+/** Deserializes JSON rows with support for SQRL-specific types. */
+public class SqrlJsonRowDataDeserializationSchema extends AbstractJsonDeserializationSchema {
+
+  @Serial private static final long serialVersionUID = 1L;
+
+  private final SqrlJsonToRowDataConverters.JsonToRowDataConverter runtimeConverter;
+
+  public SqrlJsonRowDataDeserializationSchema(
+      @NonNull RowType rowType,
+      TypeInformation<RowData> resultTypeInfo,
+      boolean failOnMissingField,
+      boolean ignoreParseErrors,
+      TimestampFormat timestampFormat) {
+    super(rowType, resultTypeInfo, failOnMissingField, ignoreParseErrors, timestampFormat);
+    runtimeConverter =
+        new SqrlJsonToRowDataConverters(failOnMissingField, ignoreParseErrors, timestampFormat)
+            .createConverter(rowType);
+  }
+
+  // Borrowed as is from org.apache.flink.formats.json.JsonRowDataDeserializationSchema
+  @Override
+  public void deserialize(byte[] message, Collector<RowData> out) throws IOException {
+    if (message == null) {
+      return;
+    }
+    try {
+      var root = deserializeToJsonNode(message);
+      if (root != null && root.isArray()) {
+        var arrayNode = (ArrayNode) root;
+        for (int i = 0; i < arrayNode.size(); i++) {
+          try {
+            var result = convertToRowData(arrayNode.get(i));
+            if (result != null) {
+              out.collect(result);
+            }
+          } catch (Throwable t) {
+            if (!ignoreParseErrors) {
+              throw t;
+            }
+            logParseErrorIfDebugEnabled(message, t);
+          }
+        }
+      } else {
+        var result = convertToRowData(root);
+        if (result != null) {
+          out.collect(result);
+        }
+      }
+    } catch (Throwable t) {
+      if (!ignoreParseErrors) {
+        throw new IOException("Failed to deserialize JSON '%s'.".formatted(new String(message)), t);
+      }
+      logParseErrorIfDebugEnabled(message, t);
+    }
+  }
+
+  public JsonNode deserializeToJsonNode(byte[] message) throws IOException {
+    return objectMapper.readTree(message);
+  }
+
+  public RowData convertToRowData(JsonNode message) {
+    return (RowData) runtimeConverter.convert(message);
+  }
+}

formats/flexible-json-format/src/main/java/com/datasqrl/flinkrunner/format/json/SqrlJsonToRowDataConverters.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright © 2026 DataSQRL (contact@datasqrl.com)
+ *
+ * Licensed 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 com.datasqrl.flinkrunner.format.json;
+
+import com.datasqrl.flinkrunner.stdlib.json.FlinkJsonType;
+import org.apache.flink.formats.common.TimestampFormat;
+import org.apache.flink.formats.json.JsonToRowDataConverters;
+import org.apache.flink.table.data.RawValueData;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.RawType;
+
+/** Adds JSON deserialization support for the SQRL JSON RAW type. */
+public class SqrlJsonToRowDataConverters extends JsonToRowDataConverters {
+
+  public SqrlJsonToRowDataConverters(
+      boolean failOnMissingField, boolean ignoreParseErrors, TimestampFormat timestampFormat) {
+    super(failOnMissingField, ignoreParseErrors, timestampFormat);
+  }
+
+  @Override
+  public JsonToRowDataConverter createConverter(LogicalType type) {
+    if (type.getTypeRoot() == LogicalTypeRoot.RAW) {
+      var rawType = (RawType<?>) type;
+      if (rawType.getOriginatingClass() == FlinkJsonType.class) {
+        return jsonNode -> {
+          if (jsonNode == null || jsonNode.isNull() || jsonNode.isMissingNode()) {
+            return null;
+          }
+          return RawValueData.fromObject(new FlinkJsonType(jsonNode));
+        };
+      }
+    }
+    return super.createConverter(type);
+  }
+}

formats/flexible-json-format/src/test/java/com/datasqrl/flinkrunner/format/json/SqrlJsonRowDataDeserializationSchemaTest.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright © 2026 DataSQRL (contact@datasqrl.com)
+ *
+ * Licensed 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 com.datasqrl.flinkrunner.format.json;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.datasqrl.flinkrunner.stdlib.json.FlinkJsonType;
+import com.datasqrl.flinkrunner.stdlib.json.FlinkJsonTypeSerializer;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.formats.common.TimestampFormat;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.table.data.RawValueData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RawType;
+import org.apache.flink.table.types.logical.RowType;
+import org.junit.jupiter.api.Test;
+
+class SqrlJsonRowDataDeserializationSchemaTest {
+
+  @Test
+  void givenRawJsonField_whenDeserialize_thenWrapsPayloadInFlinkJsonType() throws Exception {
+    var payloadType = new RawType<>(FlinkJsonType.class, new FlinkJsonTypeSerializer());
+    var rowType = RowType.of(new LogicalType[] {payloadType}, new String[] {"payload"});
+    var mapper = new ObjectMapper();
+    var schema =
+        new SqrlJsonRowDataDeserializationSchema(
+            rowType, TypeInformation.of(RowData.class), false, false, TimestampFormat.ISO_8601);
+
+    var row = schema.convertToRowData(mapper.readTree("{\"payload\":{\"id\":42}}"));
+    RawValueData<FlinkJsonType> payload = row.getRawValue(0);
+
+    assertThat(payload.toObject(payloadType.getTypeSerializer()).getJson())
+        .isEqualTo(mapper.readTree("{\"id\":42}"));
+  }
+}