rosemarYuan commented on code in PR #821: URL: https://github.com/apache/flink-agents/pull/821#discussion_r3541015349
########## runtime/src/main/java/org/apache/flink/agents/runtime/condition/CelConditionEvaluator.java: ########## @@ -0,0 +1,242 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.agents.runtime.condition; + +import com.fasterxml.jackson.databind.ObjectMapper; +import dev.cel.runtime.CelEvaluationException; +import dev.cel.runtime.CelRuntime; +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.api.configuration.CelEvaluationFailurePolicy; +import org.apache.flink.agents.plan.condition.ParsedCondition.CelExpression; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.math.BigDecimal; +import java.math.BigInteger; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** Evaluates CEL condition expressions against event data. */ +public class CelConditionEvaluator { + + private static final Logger LOG = LoggerFactory.getLogger(CelConditionEvaluator.class); + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + /** Frozen after {@link #initPrograms}; cleared by {@link #close}. */ + @Nullable private Map<String, CelRuntime.Program> programCache; + + private final CelEvaluationFailurePolicy failurePolicy; + + public CelConditionEvaluator() { + this(CelEvaluationFailurePolicy.WARN_AND_SKIP); + } + + public CelConditionEvaluator(CelEvaluationFailurePolicy failurePolicy) { + this.failurePolicy = failurePolicy; + } + + /** Pre-compiles {@code expressions} and freezes the cache. Nulls are skipped. */ + public void initPrograms(Collection<CelExpression> expressions) { + Map<String, CelRuntime.Program> programs = new HashMap<>(); + for (CelExpression expression : expressions) { + if (expression == null) { + continue; + } + String source = expression.source(); + programs.computeIfAbsent(source, CelExpressionFacade::toProgram); + } + this.programCache = Collections.unmodifiableMap(programs); + } + + public void close() { + programCache = null; + } + + /** Evaluates {@code expression} (which must have been pre-compiled). Null returns true. */ + public boolean evaluate(@Nullable CelExpression expression, Map<String, Object> activation) { + if (expression == null) { + return true; + } + String source = expression.source(); + try { + CelRuntime.Program program = programCache.get(source); + if (program == null) { + throw new IllegalStateException( + "CEL condition was not pre-compiled via initPrograms(): \"" + + source + + "\""); + } + return evaluateProgram(source, program, activation); + } catch (CelEvaluationException e) { + if (failurePolicy == CelEvaluationFailurePolicy.FAIL) { + throw new IllegalStateException( + "CEL condition evaluation failed for '" + source + "'", e); + } + LOG.warn("CEL condition evaluation failed for '{}', skipping action", source, e); + return false; + } + } + + private boolean evaluateProgram( + String condition, CelRuntime.Program program, Map<String, Object> activation) + throws CelEvaluationException { + Object result = program.eval(activation); + if (result instanceof Boolean) { + return (Boolean) result; + } + String msg = + String.format( + "CEL condition '%s' returned non-boolean type %s, treating as false", + condition, result == null ? "null" : result.getClass().getName()); + if (failurePolicy == CelEvaluationFailurePolicy.FAIL) { + throw new IllegalStateException(msg); + } + LOG.warn(msg); + return false; + } + + /** + * Builds the CEL activation. Contract (mirror of Python {@code cel_facade}): + * + * <ul> + * <li>{@code type} and {@code EventType} are framework-owned and always win. + * <li>{@code attributes} holds the single-level merge of user data: {@code output.*} subkeys, + * then root attribute fields, then {@code input.*} subkeys ({@code output > root > input} + * on collision, via {@link Map#putIfAbsent}). Only one level is flattened — nested fields + * stay nested ({@code mylist.name}, not {@code name}). + * <li>Every merged attribute is also promoted to the activation top level, so conditions can + * use bare identifiers ({@code score > 0.8}) without any AST rewriting. Framework keys + * are never shadowed. + * <li>{@code id} is the user-supplied {@code id} attribute when present, otherwise falls back + * to the event UUID. + * </ul> + * + * <p>JSON-shaped strings auto-parse first; narrow numerics widen to long/double. + */ + @SuppressWarnings("unchecked") + public Map<String, Object> createActivation(Event event) { + Map<String, Object> activation = new HashMap<>(); + activation.put("type", event.getType()); + activation.put("EventType", CelExpressionFacade.EVENT_TYPE_CONSTANTS); + + Object normalizedAttrs = normalizeValue(event.getAttributes(), 0); Review Comment: Good point. I agree that normalizing only the attributes referenced by the condition would be a useful optimization, but I don't think this is the right time to introduce it yet, mainly because the current Python CEL engine has limited support for doing this lazily. Today `buildTriggerVariables` normalizes the full `attributes` tree for each event, regardless of which variables the condition actually reads. On the Java side, this could be optimized with CEL Java's `Program.eval(CelVariableResolver)`: we could use a memoized resolver and normalize only the referenced root/subtree on demand, while still preserving the current `output > root > input` precedence. The Python side is the limiting factor. `celpy`, the more mature community-driven implementation we use today, loads values before evaluation: it eagerly scans the input values and builds the `NameContainer` / CEL value tree up front, so a lazy mapping would still be enumerated during loading. There is a possible future path through `cel-expr-python`, the new official Google-backed Python implementation released this year. Its underlying `cel-cpp` runtime has a value-provider mechanism that could support lazy attribute binding. But the current Python binding still loads eagerly, requires Python >= 3.11 while we still support 3.10, and is pre-1.0 (`0.1.3`) without API/stability guarantees. I’ve also opened an upstream issue to ask whether Python 3.10 support could be added. So for this PR, I’d keep the behavior simple and consistent: activation is eagerly built, and the cross-language contract remains behavioral rather than mechanism-level. We can revisit lazy normalization once the Python engine story is more stable. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
