Merge pull request #1882 from taozhi8833998/feat-json-object-flink

feat: support json object in flinksql

Author: taozhi8833998

pegjs/flinksql.pegjs

@@ -2914,6 +2914,32 @@ position_func_clause
     };
   }
 
+json_object_func_arg
+  = key:literal_string __ 'VALUE'i __ value:or_and_expr __ on:(KW_ON __ 'NULL'i __ ('NULL'i / 'ABSENT'i))? {
+    return {
+      type: 'json_object_arg',
+      expr: {
+        key,
+        value,
+        on: on && { type: 'origin', value: on[4] }
+      }
+    }
+  }
+
+json_object_func_args
+  = head:json_object_func_arg tail:(__ COMMA __ json_object_func_arg)* {
+    return { type: 'expr_list', value: createList(head, tail) }
+  }
+
+json_object_func_clause
+  = 'json_object'i __ LPAREN __ args:json_object_func_args __ RPAREN {
+    return {
+        type: 'function',
+        name: { name: [{ type: 'origin', value: 'json_object' }]},
+        args,
+    };
+  }
+
 trim_position
   = 'BOTH'i / 'LEADING'i / 'TRAILING'i
 
@@ -2994,6 +3020,7 @@ substring_func_clause
 
 func_call
   = position_func_clause
+  / json_object_func_clause
   / trim_func_clause
   / substring_func_clause
   / overlay_func_clause

src/expr.js

@@ -5,7 +5,7 @@ import { assignToSQL } from './assign'
 import { binaryToSQL } from './binary'
 import { caseToSQL } from './case'
 import { columnDefinitionToSQL, columnRefToSQL, fullTextSearchToSQL } from './column'
-import { anyValueFuncToSQL, castToSQL, extractFunToSQL, flattenFunToSQL, funcToSQL, lambdaToSQL, tablefuncFunToSQL } from './func'
+import { anyValueFuncToSQL, castToSQL, extractFunToSQL, flattenFunToSQL, funcToSQL, jsonObjectArgToSQL, lambdaToSQL, tablefuncFunToSQL } from './func'
 import { intervalToSQL } from './interval'
 import { jsonExprToSQL, jsonVisitorExprToSQL } from './json'
 import { selectToSQL } from './select'
@@ -36,6 +36,7 @@ const exprToSQLConvertFn = {
   insert            : unionToSQL,
   interval          : intervalToSQL,
   json              : jsonExprToSQL,
+  json_object_arg   : jsonObjectArgToSQL,
   json_visitor      : jsonVisitorExprToSQL,
   show              : showToSQL,
   struct            : arrayStructExprToSQL,

src/func.js

@@ -59,6 +59,14 @@ function flattenArgToSQL(arg) {
   return result.filter(hasVal).join(' ')
 }
 
+function jsonObjectArgToSQL(argExpr) {
+  const { expr } = argExpr
+  const { key, value, on } = expr
+  const result = [exprToSQL(key), 'VALUE', exprToSQL(value)]
+  if (on) result.push('ON', 'NULL', exprToSQL(on))
+  return result.filter(hasVal).join(' ')
+}
+
 function flattenFunToSQL(stmt) {
   const { args, type } = stmt
   const keys = ['input', 'path', 'outer', 'recursive', 'mode']
@@ -103,6 +111,7 @@ export {
   extractFunToSQL,
   flattenFunToSQL,
   funcToSQL,
+  jsonObjectArgToSQL,
   lambdaToSQL,
   tablefuncFunToSQL,
 }

test/flink.spec.js

@@ -394,6 +394,24 @@ describe('Flink', () => {
         "INSERT INTO `sink_soc_10086_node_11` (pk, f1) SELECT CAST(MAP['uri', uri] AS VARCHAR), `uri` FROM `view_soc_10086_node_8`"
       ]
     },
+    {
+      title: 'json_object',
+      sql: [
+        `select name, eventTime, eventDetail
+        from (
+            select
+            concat('AK中文信息') as name,
+            cast(event_time as varchar) as eventTime,
+            json_object(
+                'risk-tag' value risk_tag,
+                'abc' VALUE (10 * 2),
+                'user-agent' value JSON_OBJECT('city' VALUE 'New York', 'postalCode' VALUE '10001')
+            ) as eventDetail
+        from check_risk
+        );`,
+        "SELECT `name`, `eventTime`, `eventDetail` FROM (SELECT concat('AK中文信息') AS `name`, CAST(`event_time` AS VARCHAR) AS `eventTime`, JSON_OBJECT('risk-tag' VALUE `risk_tag`, 'abc' VALUE (10 * 2), 'user-agent' VALUE JSON_OBJECT('city' VALUE 'New York', 'postalCode' VALUE '10001')) AS `eventDetail` FROM `check_risk`)"
+      ]
+    },
   ];
 
   SQL_LIST.forEach(sqlInfo => {