Add CEL support for Data Quality rules (#313)

* Add CEL support for Data Quality rules

* Minor cleanup

Author: rayokota

package-lock.json

@@ -33,6 +33,7 @@
     "@aws-sdk/credential-providers": "^3.637.0",
     "@azure/identity": "^4.4.1",
     "@azure/keyvault-keys": "^4.8.0",
+    "@bufbuild/cel": "^0.1.0",
     "@bufbuild/protobuf": "^2.0.0",
     "@criteria/json-schema": "^0.10.0",
     "@criteria/json-schema-validation": "^0.10.0",

schemaregistry/package.json

@@ -0,0 +1,87 @@
+import {RuleRegistry} from "../../serde/rule-registry";
+import {RuleContext, RuleExecutor} from "../../serde/serde";
+import {ClientConfig} from "../../rest-service";
+import stringify from "json-stringify-deterministic";
+import {LRUCache} from "lru-cache";
+import {createEnv} from "@bufbuild/cel";
+import {createRegistry} from "@bufbuild/protobuf";
+
+export class CelExecutor implements RuleExecutor {
+  config: Map<string, string> | null = null
+  env = createEnv("", createRegistry());
+  cache: LRUCache<string, any> = new LRUCache({max: 1000})
+
+  static register(): CelExecutor {
+    const executor = new CelExecutor()
+    RuleRegistry.registerRuleExecutor(executor)
+    return executor
+  }
+
+  configure(clientConfig: ClientConfig, config: Map<string, string>) {
+    this.config = config
+  }
+
+  type(): string {
+    return "CEL"
+  }
+
+  async transform(ctx: RuleContext, msg: any): Promise<any> {
+    const args = {
+      message: msg
+    }
+    return await this.execute(ctx, msg, args)
+  }
+
+  async execute(ctx: RuleContext, msg: any, args: { [key: string]: any }): Promise<any> {
+    let expr = ctx.rule.expr
+    if (expr == null) {
+      return msg
+    }
+    const index = expr.indexOf(';')
+    if (index >= 0) {
+      const guard = expr.substring(0, index)
+      if (guard.trim().length != 0) {
+        const guardResult = await this.executeRule(ctx, guard, msg, args)
+        if (guardResult === false) {
+          // skip the expr
+          if (ctx.rule.kind === 'CONDITION') {
+            return true
+          }
+          return msg
+        }
+      }
+      expr = expr.substring(index + 1)
+    }
+    return await this.executeRule(ctx, expr, msg, args)
+  }
+
+  async executeRule(ctx: RuleContext, expr: string, obj: any, args: { [key: string]: any }): Promise<any> {
+    const schema = ctx.target.schema
+    const scriptType = ctx.target.schemaType
+    const rule: RuleWithArgs = {
+      rule: expr,
+      scriptType: scriptType,
+      schema: schema
+    }
+    const ruleJson = stringify(rule)
+    let program = this.cache.get(ruleJson)
+    if (program == null) {
+      const parsedExpr = this.env.parse(expr)
+      program = this.env.plan(parsedExpr)
+      this.cache.set(ruleJson, program)
+    }
+    for (const [key, value] of Object.entries(args)) {
+      this.env.set(key, value)
+    }
+    return this.env.eval(program)
+  }
+
+  async close(): Promise<void> {
+  }
+}
+
+interface RuleWithArgs {
+  rule?: string
+  scriptType?: string
+  schema?: string
+}

schemaregistry/rules/cel/cel-executor.ts

@@ -0,0 +1,60 @@
+import {RuleRegistry} from "../../serde/rule-registry";
+import {
+  FieldContext,
+  FieldRuleExecutor,
+  FieldTransform,
+  RuleContext,
+} from "../../serde/serde";
+import {ClientConfig} from "../../rest-service";
+import {CelExecutor} from "./cel-executor";
+
+export class CelFieldExecutor extends FieldRuleExecutor {
+  executor: CelExecutor = new CelExecutor()
+
+  static register(): CelFieldExecutor {
+    const executor = new CelFieldExecutor()
+    RuleRegistry.registerRuleExecutor(executor)
+    return executor
+  }
+
+  configure(clientConfig: ClientConfig, config: Map<string, string>) {
+    this.config = config
+  }
+
+  type(): string {
+    return "CEL_FIELD"
+  }
+
+  override newTransform(ctx: RuleContext): FieldTransform {
+    return new CelFieldExecutorTransform(this.executor)
+  }
+
+  async close(): Promise<void> {
+  }
+}
+
+export class CelFieldExecutorTransform implements FieldTransform {
+  private executor: CelExecutor
+
+  constructor(executor: CelExecutor) {
+    this.executor = executor
+  }
+
+  async transform(ctx: RuleContext, fieldCtx: FieldContext, fieldValue: any): Promise<any> {
+    if (fieldValue == null) {
+      return null
+    }
+    if (!fieldCtx.isPrimitive()) {
+      return fieldValue
+    }
+    const args = {
+      value: fieldValue,
+      fullName: fieldCtx.fullName,
+      name: fieldCtx.name,
+      typeName: fieldCtx.typeName(),
+      tags: Array.from(fieldCtx.tags),
+      message: fieldCtx.containingMessage
+    }
+    return await this.executor.execute(ctx, fieldValue, args)
+  }
+}