Skip to content

[FLINK-40376][formats] Support VARIANT type in flink-json converters #28959

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ferenc-csaky wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[FLINK-40376][formats] Support VARIANT type in flink-json converters #28959

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions ...flink-json/src/main/java/org/apache/flink/formats/json/JsonParserToRowDataConverters.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,8 @@
import org.apache.flink.table.types.logical.MultisetType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.utils.LogicalTypeUtils;
import org.apache.flink.types.variant.BinaryVariant;
import org.apache.flink.types.variant.BinaryVariantInternalBuilder;

import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParser;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonToken;
Expand Down Expand Up @@ -167,6 +169,8 @@ private JsonParserToRowDataConverter createNotNullConverter(LogicalType type) {
case BINARY:
case VARBINARY:
return JsonParser::getBinaryValue;
case VARIANT:
return this::convertToVariant;
case DECIMAL:
return createDecimalConverter((DecimalType) type);
case ARRAY:
Expand Down Expand Up @@ -323,6 +327,10 @@ private StringData convertToString(JsonParser jp) throws IOException {
}
}

private BinaryVariant convertToVariant(JsonParser jp) throws IOException {
return BinaryVariantInternalBuilder.parseJson(jp.readValueAsTree().toString(), false);
}

private JsonParserToRowDataConverter createDecimalConverter(DecimalType decimalType) {
final int precision = decimalType.getPrecision();
final int scale = decimalType.getScale();
Expand Down
12 changes: 12 additions & 0 deletions ...rmats/flink-json/src/main/java/org/apache/flink/formats/json/JsonToRowDataConverters.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,8 @@
import org.apache.flink.table.types.logical.MultisetType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.utils.LogicalTypeUtils;
import org.apache.flink.types.variant.BinaryVariant;
import org.apache.flink.types.variant.BinaryVariantInternalBuilder;

import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ArrayNode;
Expand Down Expand Up @@ -137,6 +139,8 @@ private JsonToRowDataConverter createNotNullConverter(LogicalType type) {
case BINARY:
case VARBINARY:
return this::convertToBytes;
case VARIANT:
return this::convertToVariant;
case DECIMAL:
return createDecimalConverter((DecimalType) type);
case ARRAY:
Expand Down Expand Up @@ -270,6 +274,14 @@ private StringData convertToString(JsonNode jsonNode) {
}
}

private BinaryVariant convertToVariant(JsonNode jsonNode) {
try {
return BinaryVariantInternalBuilder.parseJson(jsonNode.toString(), false);
} catch (IOException e) {
throw new JsonParseException("Unable to deserialize VARIANT value.", e);
}
}

private byte[] convertToBytes(JsonNode jsonNode) {
try {
return jsonNode.binaryValue();
Expand Down
12 changes: 12 additions & 0 deletions ...rmats/flink-json/src/main/java/org/apache/flink/formats/json/RowDataToJsonConverters.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,12 +33,14 @@
import org.apache.flink.table.types.logical.MapType;
import org.apache.flink.table.types.logical.MultisetType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.types.variant.Variant;

import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ArrayNode;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;

import java.io.IOException;
import java.io.Serializable;
import java.math.BigDecimal;
import java.time.LocalDate;
Expand Down Expand Up @@ -149,6 +151,8 @@ private RowDataToJsonConverter createNotNullConverter(LogicalType type) {
new IntType());
case ROW:
return createRowConverter((RowType) type);
case VARIANT:
return this::convertVariant;
case RAW:
default:
throw new UnsupportedOperationException("Not support to parse type: " + type);
Expand All @@ -166,6 +170,14 @@ private RowDataToJsonConverter createDecimalConverter() {
};
}

private JsonNode convertVariant(ObjectMapper mapper, JsonNode reuse, Object value) {
try {
return mapper.readTree(((Variant) value).toJson());
} catch (IOException e) {
throw new JsonParseException("Unable to serialize VARIANT value.", e);
}
}

private RowDataToJsonConverter createDateConverter() {
return (mapper, reuse, value) -> {
int days = (int) value;
Expand Down
109 changes: 109 additions & 0 deletions ...ts/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowDataSerDeSchemaTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
import org.apache.flink.testutils.logging.LoggerAuditingExtension;
import org.apache.flink.types.Row;
import org.apache.flink.types.variant.BinaryVariantBuilder;
import org.apache.flink.util.Collector;
import org.apache.flink.util.jackson.JacksonMapperFactory;

Expand Down Expand Up @@ -87,6 +88,7 @@
import static org.apache.flink.table.api.DataTypes.TIMESTAMP;
import static org.apache.flink.table.api.DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE;
import static org.apache.flink.table.api.DataTypes.TINYINT;
import static org.apache.flink.table.api.DataTypes.VARIANT;
import static org.apache.flink.table.types.utils.TypeConversions.fromLogicalToDataType;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
Expand Down Expand Up @@ -239,6 +241,113 @@ void testSerDe() throws Exception {
assertThat(serializedJson).containsExactly(actualBytes);
}

@TestTemplate
void testVariantScalarsAndNestedTypes() throws Exception {
String json =
"{\"string\":\"Flink\",\"boolean\":true,\"integer\":42,\"decimal\":3.14,"
+ "\"array\":[1,{\"nested\":\"value\"}],\"nestedArray\":[\"array\"],"
+ "\"nestedMap\":{\"key\":\"map\"},\"nestedRow\":{\"field\":\"row\"}}";
DataType dataType =
ROW(
FIELD("string", VARIANT()),
FIELD("boolean", VARIANT()),
FIELD("integer", VARIANT()),
FIELD("decimal", VARIANT()),
FIELD("array", VARIANT()),
FIELD("nestedArray", ARRAY(VARIANT())),
FIELD("nestedMap", MAP(STRING(), VARIANT())),
FIELD("nestedRow", ROW(FIELD("field", VARIANT()))));
RowType rowType = (RowType) dataType.getLogicalType();

DeserializationSchema<RowData> deserializationSchema =
createDeserializationSchema(
isJsonParser, rowType, false, false, TimestampFormat.ISO_8601);
open(deserializationSchema);

RowData rowData = deserializationSchema.deserialize(json.getBytes());
assertThat(rowData.getVariant(0).toJson()).isEqualTo("\"Flink\"");
assertThat(rowData.getVariant(1).toJson()).isEqualTo("true");
assertThat(rowData.getVariant(2).toJson()).isEqualTo("42");
assertThat(rowData.getVariant(3).toJson()).isEqualTo("3.14");
assertThat(rowData.getVariant(4).toJson()).isEqualTo("[1,{\"nested\":\"value\"}]");
assertThat(rowData.getArray(5).getVariant(0).toJson()).isEqualTo("\"array\"");
assertThat(rowData.getMap(6).valueArray().getVariant(0).toJson()).isEqualTo("\"map\"");
assertThat(rowData.getRow(7, 1).getVariant(0).toJson()).isEqualTo("\"row\"");

JsonRowDataSerializationSchema serializationSchema =
new JsonRowDataSerializationSchema(
rowType,
TimestampFormat.ISO_8601,
JsonFormatOptions.MapNullKeyMode.LITERAL,
"null",
true,
false);
open(serializationSchema);

assertThat(OBJECT_MAPPER.readTree(serializationSchema.serialize(rowData)))
.isEqualTo(OBJECT_MAPPER.readTree(json));
}

@TestTemplate
void testVariantDuplicateKeysUseLastValue() throws Exception {
byte[] json = "{\"variant\":{\"key\":1,\"key\":2}}".getBytes();
RowType rowType = (RowType) ROW(FIELD("variant", VARIANT())).getLogicalType();

DeserializationSchema<RowData> deserializationSchema =
createDeserializationSchema(
isJsonParser, rowType, false, false, TimestampFormat.ISO_8601);
open(deserializationSchema);
assertThat(deserializationSchema.deserialize(json).getVariant(0).toJson())
.isEqualTo("{\"key\":2}");
}

@TestTemplate
void testVariantJsonNullIsDeserializedAsSqlNull() throws Exception {
byte[] json = "{\"variant\":null}".getBytes();
RowType rowType = (RowType) ROW(FIELD("variant", VARIANT())).getLogicalType();

DeserializationSchema<RowData> deserializationSchema =
createDeserializationSchema(
isJsonParser, rowType, false, false, TimestampFormat.ISO_8601);
open(deserializationSchema);
assertThat(deserializationSchema.deserialize(json).isNullAt(0)).isTrue();

JsonRowDataSerializationSchema serializationSchema =
new JsonRowDataSerializationSchema(
rowType,
TimestampFormat.ISO_8601,
JsonFormatOptions.MapNullKeyMode.LITERAL,
"null",
true,
false);
open(serializationSchema);
assertThat(
serializationSchema.serialize(
GenericRowData.of(new BinaryVariantBuilder().ofNull())))
.containsExactly(json);
}

@Test
void testVariantSerializationRejectsNonFiniteNumbers() {
RowType rowType = (RowType) ROW(FIELD("variant", VARIANT())).getLogicalType();
JsonRowDataSerializationSchema serializationSchema =
new JsonRowDataSerializationSchema(
rowType,
TimestampFormat.ISO_8601,
JsonFormatOptions.MapNullKeyMode.LITERAL,
"null",
true,
false);
open(serializationSchema);

assertThatThrownBy(
() ->
serializationSchema.serialize(
GenericRowData.of(
new BinaryVariantBuilder().of(Double.NaN))))
.hasMessage("Non-finite value NaN cannot be serialized to JSON.");
}

@Test
public void testEmptyJsonArrayDeserialization() throws Exception {
DataType dataType = ROW(FIELD("f1", INT()), FIELD("f2", BOOLEAN()), FIELD("f3", STRING()));
Expand Down