(flink) branch master updated: [FLINK-37752][json] Make configuration 'json.decode.json-parser.enabled' take effect

[hangxiang]

This is an automated email from the ASF dual-hosted git repository.

hangxiang 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 8064123b00f [FLINK-37752][json] Make configuration 
'json.decode.json-parser.enabled' take effect
8064123b00f is described below

commit 8064123b00f93a4c0b99f820dafd04d47c06c7b0
Author: Hongjia Liang <lianghongjia....@outlook.com>
AuthorDate: Wed Apr 30 16:56:47 2025 +0800

    [FLINK-37752][json] Make configuration 'json.decode.json-parser.enabled' 
take effect
---
 .../flink/formats/json/JsonFormatFactory.java      |  2 ++
 .../flink/formats/json/JsonFormatFactoryTest.java  | 39 ++++++++++++++++++++++
 2 files changed, 41 insertions(+)

diff --git 
a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonFormatFactory.java
 
b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonFormatFactory.java
index 6ecc9a0f086..9fb1787dcd5 100644
--- 
a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonFormatFactory.java
+++ 
b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonFormatFactory.java
@@ -191,6 +191,7 @@ public class JsonFormatFactory implements 
DeserializationFormatFactory, Serializ
         options.add(MAP_NULL_KEY_LITERAL);
         options.add(ENCODE_DECIMAL_AS_PLAIN_NUMBER);
         options.add(ENCODE_IGNORE_NULL_FIELDS);
+        options.add(DECODE_JSON_PARSER_ENABLED);
         return options;
     }
 
@@ -202,6 +203,7 @@ public class JsonFormatFactory implements 
DeserializationFormatFactory, Serializ
         options.add(MAP_NULL_KEY_LITERAL);
         options.add(ENCODE_DECIMAL_AS_PLAIN_NUMBER);
         options.add(ENCODE_IGNORE_NULL_FIELDS);
+        options.add(DECODE_JSON_PARSER_ENABLED);
         return options;
     }
 }
diff --git 
a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonFormatFactoryTest.java
 
b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonFormatFactoryTest.java
index 4430203b28e..f1183436099 100644
--- 
a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonFormatFactoryTest.java
+++ 
b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonFormatFactoryTest.java
@@ -125,6 +125,13 @@ class JsonFormatFactoryTest {
         testSchemaDeserializationSchema(tableOptions);
     }
 
+    @Test
+    void testDecodeJsonParseEnabled() {
+        testJsonParserConfiguration(true, 
JsonParserRowDataDeserializationSchema.class);
+
+        testJsonParserConfiguration(false, 
JsonRowDataDeserializationSchema.class);
+    }
+
     // ------------------------------------------------------------------------
     //  Utilities
     // ------------------------------------------------------------------------
@@ -229,6 +236,38 @@ class JsonFormatFactoryTest {
         options.put("json.map-null-key.literal", "null");
         options.put("json.encode.decimal-as-plain-number", "true");
         options.put("json.encode.ignore-null-fields", "true");
+        options.put("json.decode.json-parser.enabled", "true");
         return options;
     }
+
+    private void testJsonParserConfiguration(boolean enabled, Class<?> 
expectedClass) {
+        Map<String, String> options =
+                getModifyOptions(
+                        opt -> opt.put("json.decode.json-parser.enabled", 
String.valueOf(enabled)));
+
+        DeserializationSchema<RowData> actualDeser =
+                createTableSource(options)
+                        .valueFormat
+                        .createRuntimeDecoder(
+                                ScanRuntimeProviderContext.INSTANCE,
+                                SCHEMA.toPhysicalRowDataType());
+
+        DeserializationSchema<RowData> expectedDeser =
+                enabled
+                        ? new JsonParserRowDataDeserializationSchema(
+                                PHYSICAL_TYPE,
+                                InternalTypeInfo.of(PHYSICAL_TYPE),
+                                false,
+                                true,
+                                TimestampFormat.ISO_8601)
+                        : new JsonRowDataDeserializationSchema(
+                                PHYSICAL_TYPE,
+                                InternalTypeInfo.of(PHYSICAL_TYPE),
+                                false,
+                                true,
+                                TimestampFormat.ISO_8601);
+
+        assertThat(actualDeser).isInstanceOf(expectedClass);
+        assertThat(actualDeser).isEqualTo(expectedDeser);
+    }
 }