Merge pull request #2164 from taozhi8833998/feat-tumble-named-params-flink

feat: support named params for tumble function in flinksql

Author: taozhi8833998

pegjs/flinksql.pegjs

@@ -1890,6 +1890,8 @@ table_join
     };
   }
 
+tumble_args
+  = n:('DATA'i __ IMPLIES_ARROW) __ KW_TABLE __ d:table_name
 //NOTE that, the table assigned to `var` shouldn't write in `table_join`
 table_base
   = KW_DUAL {
@@ -1919,16 +1921,36 @@ table_base
         as: alias
       };
     }
-  / KW_TABLE __ LPAREN __ KW_TUMBLE __ LPAREN __ KW_TABLE __ d:table_name __ COMMA __ 'DESCRIPTOR'i __ LPAREN __ t:column_ref __ RPAREN __ COMMA __ s:interval_expr __ RPAREN __ RPAREN __ alias:alias_clause? {
-    return {
+  / KW_TABLE __ LPAREN __ KW_TUMBLE __ LPAREN __ dn:('DATA'i __ IMPLIES_ARROW)? __ KW_TABLE __ d:table_name __ COMMA __ tn:('TIMECOL'i __ IMPLIES_ARROW)? __ 'DESCRIPTOR'i __ LPAREN __ t:column_ref __ RPAREN __ COMMA __ sn:('SIZE'i __ IMPLIES_ARROW)? __ s:interval_expr o:(__ COMMA __ ('OFFSET'i __ IMPLIES_ARROW)? __ interval_expr)? __ RPAREN __ RPAREN __ alias:alias_clause? {
+    const result = {
       expr: {
         type: 'tumble',
-        data: d,
-        timecol: t,
-        size: s
+        data: {
+          name: dn && dn[0],
+          symbol: dn && dn[2],
+          expr: d
+        },
+        timecol: {
+          name: tn && tn[0],
+          symbol: tn && tn[2],
+          expr: t,
+        },
+        size: {
+          name: sn && sn[0],
+          symbol: sn && sn[2],
+          expr: s,
+        },
       },
       as: alias
     }
+    if (o) {
+      result.expr.offset = {
+        name: o[3] && o[3][0],
+        symbol: o[3] && o[3][2],
+        expr: o[5],
+      }
+    }
+    return result
   }
 
 join_op
@@ -3562,6 +3584,7 @@ SINGLE_ARROW = '->'
 DOUBLE_ARROW = '->>'
 WELL_ARROW = '#>'
 DOUBLE_WELL_ARROW = '#>>'
+IMPLIES_ARROW = '=>'
 
 OPERATOR_CONCATENATION = '||'
 OPERATOR_AND = '&&'

src/tables.js

@@ -64,12 +64,20 @@ function tableHintToSQL(tableHintExpr) {
   return result.filter(hasVal).join(' ')
 }
 
+function tableTumbleArgsToSQL(param, expr) {
+  const { name, symbol } = param
+  return [toUpper(name), symbol, expr].filter(hasVal).join(' ')
+}
 function tableTumbleToSQL(tumble) {
   if (!tumble) return ''
-  const { data: tableInfo, timecol, size } = tumble
-  const fullTableName = [identifierToSql(tableInfo.db), identifierToSql(tableInfo.table)].filter(hasVal).join('.')
-  const result = ['TABLE(TUMBLE(TABLE', fullTableName, `DESCRIPTOR(${columnRefToSQL(timecol)})`, `${intervalToSQL(size)}))`]
-  return result.filter(hasVal).join(' ')
+  const { data: tableInfo, timecol, offset, size } = tumble
+  const fullTableName = [identifierToSql(tableInfo.expr.db), identifierToSql(tableInfo.expr.schema), identifierToSql(tableInfo.expr.table)].filter(hasVal).join('.')
+  const timeColSQL = `DESCRIPTOR(${columnRefToSQL(timecol.expr)})`
+  const result = [`TABLE(TUMBLE(TABLE ${tableTumbleArgsToSQL(tableInfo, fullTableName)}`, tableTumbleArgsToSQL(timecol, timeColSQL)]
+  const sizeSQL = tableTumbleArgsToSQL(size, intervalToSQL(size.expr))
+  if (offset && offset.expr) result.push(sizeSQL, `${tableTumbleArgsToSQL(offset, intervalToSQL(offset.expr))}))`)
+  else result.push(`${sizeSQL}))`)
+  return result.filter(hasVal).join(', ')
 }
 
 function temporalTableOptionToSQL(stmt) {

test/flink.spec.js

@@ -376,7 +376,38 @@ describe('Flink', () => {
             INTERVAL '60' SECONDS
           )
         )`,
-        "SELECT `window_start`, `window_end`, `http_status`, COUNT(*) AS `count_http_status` FROM TABLE(TUMBLE(TABLE `parsed_logs` DESCRIPTOR(`_operationTs`) INTERVAL '60' SECONDS))"
+        "SELECT `window_start`, `window_end`, `http_status`, COUNT(*) AS `count_http_status` FROM TABLE(TUMBLE(TABLE `parsed_logs`, DESCRIPTOR(`_operationTs`), INTERVAL '60' SECONDS))"
+      ]
+    },
+    {
+      title: 'tumble table with offset',
+      sql: [
+        `SELECT
+          window_start,
+          window_end,
+          http_status,
+          count(*) as count_http_status
+        FROM
+        TABLE (
+          TUMBLE (
+            TABLE parsed_logs,
+            DESCRIPTOR (_operationTs),
+            INTERVAL '60' SECONDS,
+            INTERVAL '10' MINUTES
+          )
+        )`,
+        "SELECT `window_start`, `window_end`, `http_status`, COUNT(*) AS `count_http_status` FROM TABLE(TUMBLE(TABLE `parsed_logs`, DESCRIPTOR(`_operationTs`), INTERVAL '60' SECONDS, INTERVAL '10' MINUTES))"
+      ]
+    },
+    {
+      title: 'tumble table with named params',
+      sql: [
+        `SELECT * FROM TABLE(
+        TUMBLE(
+          DATA => TABLE Bid,
+          TIMECOL => DESCRIPTOR(bidtime),
+          SIZE => INTERVAL '10' MINUTES))`,
+        "SELECT * FROM TABLE(TUMBLE(TABLE DATA => `Bid`, TIMECOL => DESCRIPTOR(`bidtime`), SIZE => INTERVAL '10' MINUTES))"
       ]
     },
     {