Skip to content

Commit a587c76

Browse files
ferenc-csakyferenc-csaky
committed
feat: Handle FlinkJsonType decoding in flexible-json format (#368)
1 parent 127bdd1 commit a587c76

5 files changed

Lines changed: 191 additions & 493 deletions

File tree

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

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,6 @@
3030
import org.apache.flink.formats.common.TimestampFormat;
3131
import org.apache.flink.formats.json.JsonFormatFactory;
3232
import org.apache.flink.formats.json.JsonFormatOptionsUtil;
33-
import org.apache.flink.formats.json.JsonRowDataDeserializationSchema;
3433
import org.apache.flink.table.connector.ChangelogMode;
3534
import org.apache.flink.table.connector.Projection;
3635
import org.apache.flink.table.connector.format.DecodingFormat;
@@ -74,8 +73,8 @@ public DeserializationSchema<RowData> createRuntimeDecoder(
7473
final RowType rowType = (RowType) producedDataType.getLogicalType();
7574
final TypeInformation<RowData> rowDataTypeInfo =
7675
context.createTypeInformation(producedDataType);
77-
JsonRowDataDeserializationSchema jsonRowDataDeserializationSchema =
78-
new JsonRowDataDeserializationSchema(
76+
var jsonRowDataDeserializationSchema =
77+
new SqrlJsonRowDataDeserializationSchema(
7978
rowType, rowDataTypeInfo, false, false, TimestampFormat.ISO_8601);
8079
return jsonRowDataDeserializationSchema;
8180
}

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

Lines changed: 91 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,91 @@
1+
/*
2+
* Copyright © 2026 DataSQRL (contact@datasqrl.com)
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package com.datasqrl.flinkrunner.format.json;
17+
18+
import java.io.IOException;
19+
import java.io.Serial;
20+
import lombok.NonNull;
21+
import org.apache.flink.api.common.typeinfo.TypeInformation;
22+
import org.apache.flink.formats.common.TimestampFormat;
23+
import org.apache.flink.formats.json.AbstractJsonDeserializationSchema;
24+
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
25+
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ArrayNode;
26+
import org.apache.flink.table.data.RowData;
27+
import org.apache.flink.table.types.logical.RowType;
28+
import org.apache.flink.util.Collector;
29+
30+
/** Deserializes JSON rows with support for SQRL-specific types. */
31+
public class SqrlJsonRowDataDeserializationSchema extends AbstractJsonDeserializationSchema {
32+
33+
@Serial private static final long serialVersionUID = 1L;
34+
35+
private final SqrlJsonToRowDataConverters.JsonToRowDataConverter runtimeConverter;
36+
37+
public SqrlJsonRowDataDeserializationSchema(
38+
@NonNull RowType rowType,
39+
TypeInformation<RowData> resultTypeInfo,
40+
boolean failOnMissingField,
41+
boolean ignoreParseErrors,
42+
TimestampFormat timestampFormat) {
43+
super(rowType, resultTypeInfo, failOnMissingField, ignoreParseErrors, timestampFormat);
44+
runtimeConverter =
45+
new SqrlJsonToRowDataConverters(failOnMissingField, ignoreParseErrors, timestampFormat)
46+
.createConverter(rowType);
47+
}
48+
49+
// Borrowed as is from org.apache.flink.formats.json.JsonRowDataDeserializationSchema
50+
@Override
51+
public void deserialize(byte[] message, Collector<RowData> out) throws IOException {
52+
if (message == null) {
53+
return;
54+
}
55+
try {
56+
var root = deserializeToJsonNode(message);
57+
if (root != null && root.isArray()) {
58+
var arrayNode = (ArrayNode) root;
59+
for (int i = 0; i < arrayNode.size(); i++) {
60+
try {
61+
var result = convertToRowData(arrayNode.get(i));
62+
if (result != null) {
63+
out.collect(result);
64+
}
65+
} catch (Throwable t) {
66+
if (!ignoreParseErrors) {
67+
throw t;
68+
}
69+
}
70+
}
71+
} else {
72+
var result = convertToRowData(root);
73+
if (result != null) {
74+
out.collect(result);
75+
}
76+
}
77+
} catch (Throwable t) {
78+
if (!ignoreParseErrors) {
79+
throw new IOException("Failed to deserialize JSON '%s'.".formatted(new String(message)), t);
80+
}
81+
}
82+
}
83+
84+
public JsonNode deserializeToJsonNode(byte[] message) throws IOException {
85+
return objectMapper.readTree(message);
86+
}
87+
88+
public RowData convertToRowData(JsonNode message) {
89+
return (RowData) runtimeConverter.convert(message);
90+
}
91+
}

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

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
/*
2+
* Copyright © 2026 DataSQRL (contact@datasqrl.com)
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package com.datasqrl.flinkrunner.format.json;
17+
18+
import com.datasqrl.flinkrunner.stdlib.json.FlinkJsonType;
19+
import org.apache.flink.formats.common.TimestampFormat;
20+
import org.apache.flink.formats.json.JsonToRowDataConverters;
21+
import org.apache.flink.table.data.RawValueData;
22+
import org.apache.flink.table.types.logical.LogicalType;
23+
import org.apache.flink.table.types.logical.LogicalTypeRoot;
24+
import org.apache.flink.table.types.logical.RawType;
25+
26+
/** Adds JSON deserialization support for the SQRL JSON RAW type. */
27+
public class SqrlJsonToRowDataConverters extends JsonToRowDataConverters {
28+
29+
public SqrlJsonToRowDataConverters(
30+
boolean failOnMissingField, boolean ignoreParseErrors, TimestampFormat timestampFormat) {
31+
super(failOnMissingField, ignoreParseErrors, timestampFormat);
32+
}
33+
34+
@Override
35+
public JsonToRowDataConverter createConverter(LogicalType type) {
36+
if (type.getTypeRoot() == LogicalTypeRoot.RAW) {
37+
var rawType = (RawType<?>) type;
38+
if (rawType.getOriginatingClass() == FlinkJsonType.class) {
39+
return jsonNode -> {
40+
if (jsonNode == null || jsonNode.isNull() || jsonNode.isMissingNode()) {
41+
return null;
42+
}
43+
return RawValueData.fromObject(new FlinkJsonType(jsonNode));
44+
};
45+
}
46+
}
47+
return super.createConverter(type);
48+
}
49+
}

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

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
/*
2+
* Copyright © 2026 DataSQRL (contact@datasqrl.com)
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package com.datasqrl.flinkrunner.format.json;
17+
18+
import static org.assertj.core.api.Assertions.assertThat;
19+
20+
import com.datasqrl.flinkrunner.stdlib.json.FlinkJsonType;
21+
import com.datasqrl.flinkrunner.stdlib.json.FlinkJsonTypeSerializer;
22+
import org.apache.flink.api.common.typeinfo.TypeInformation;
23+
import org.apache.flink.formats.common.TimestampFormat;
24+
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
25+
import org.apache.flink.table.data.RawValueData;
26+
import org.apache.flink.table.data.RowData;
27+
import org.apache.flink.table.types.logical.LogicalType;
28+
import org.apache.flink.table.types.logical.RawType;
29+
import org.apache.flink.table.types.logical.RowType;
30+
import org.junit.jupiter.api.Test;
31+
32+
class SqrlJsonRowDataDeserializationSchemaTest {
33+
34+
@Test
35+
void givenRawJsonField_whenDeserialize_thenWrapsPayloadInFlinkJsonType() throws Exception {
36+
var payloadType = new RawType<>(FlinkJsonType.class, new FlinkJsonTypeSerializer());
37+
var rowType = RowType.of(new LogicalType[] {payloadType}, new String[] {"payload"});
38+
var mapper = new ObjectMapper();
39+
var schema =
40+
new SqrlJsonRowDataDeserializationSchema(
41+
rowType, TypeInformation.of(RowData.class), false, false, TimestampFormat.ISO_8601);
42+
43+
var row = schema.convertToRowData(mapper.readTree("{\"payload\":{\"id\":42}}"));
44+
RawValueData<FlinkJsonType> payload = row.getRawValue(0);
45+
46+
assertThat(payload.toObject(payloadType.getTypeSerializer()).getJson())
47+
.isEqualTo(mapper.readTree("{\"id\":42}"));
48+
}
49+
}

0 commit comments

Comments
 (0)