You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by li...@apache.org on 2020/11/04 12:47:26 UTC
[flink] branch master updated: [FLINK-19790][json] Clear reused
ObjectNode's content for map converter in RowDataToJsonConverters
This is an automated email from the ASF dual-hosted git repository.
libenchao 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 b7487bd [FLINK-19790][json] Clear reused ObjectNode's content for map converter in RowDataToJsonConverters
b7487bd is described below
commit b7487bd85bedea8bd50e706c253a61ebbcf8a8bc
Author: Benchao Li <li...@gmail.com>
AuthorDate: Fri Oct 23 23:03:44 2020 +0800
[FLINK-19790][json] Clear reused ObjectNode's content for map converter in RowDataToJsonConverters
This closes #13777
---
.../formats/json/RowDataToJsonConverters.java | 1 +
.../formats/json/JsonRowDataSerDeSchemaTest.java | 23 +++++++++++++++++++++-
2 files changed, 23 insertions(+), 1 deletion(-)
diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/RowDataToJsonConverters.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/RowDataToJsonConverters.java
index e7530f1..132efc5 100644
--- a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/RowDataToJsonConverters.java
+++ b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/RowDataToJsonConverters.java
@@ -242,6 +242,7 @@ public class RowDataToJsonConverters implements Serializable {
node = mapper.createObjectNode();
} else {
node = (ObjectNode) reuse;
+ node.removeAll();
}
MapData map = (MapData) object;
diff --git a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowDataSerDeSchemaTest.java b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowDataSerDeSchemaTest.java
index e2a4762..bba9cc1 100644
--- a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowDataSerDeSchemaTest.java
+++ b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowDataSerDeSchemaTest.java
@@ -246,7 +246,12 @@ public class JsonRowDataSerDeSchemaTest {
RowType rowType = (RowType) ROW(
FIELD("f1", INT()),
FIELD("f2", BOOLEAN()),
- FIELD("f3", STRING())
+ FIELD("f3", STRING()),
+ FIELD("f4", MAP(STRING(), STRING())),
+ FIELD("f5", ARRAY(STRING())),
+ FIELD("f6", ROW(
+ FIELD("f1", STRING()),
+ FIELD("f2", INT())))
).getLogicalType();
JsonRowDataDeserializationSchema deserializationSchema = new JsonRowDataDeserializationSchema(
@@ -261,6 +266,14 @@ public class JsonRowDataSerDeSchemaTest {
root.put("f1", 1);
root.put("f2", true);
root.put("f3", "str");
+ ObjectNode map = root.putObject("f4");
+ map.put("hello1", "flink");
+ ArrayNode array = root.putArray("f5");
+ array.add("element1");
+ array.add("element2");
+ ObjectNode row = root.putObject("f6");
+ row.put("f1", "this is row1");
+ row.put("f2", 12);
byte[] serializedJson = objectMapper.writeValueAsBytes(root);
RowData rowData = deserializationSchema.deserialize(serializedJson);
byte[] actual = serializationSchema.serialize(rowData);
@@ -273,6 +286,14 @@ public class JsonRowDataSerDeSchemaTest {
root.put("f1", 10);
root.put("f2", false);
root.put("f3", "newStr");
+ ObjectNode map = root.putObject("f4");
+ map.put("hello2", "json");
+ ArrayNode array = root.putArray("f5");
+ array.add("element3");
+ array.add("element4");
+ ObjectNode row = root.putObject("f6");
+ row.put("f1", "this is row2");
+ row.putNull("f2");
byte[] serializedJson = objectMapper.writeValueAsBytes(root);
RowData rowData = deserializationSchema.deserialize(serializedJson);
byte[] actual = serializationSchema.serialize(rowData);