[Streams][StreamLang] Add support for arrays with single values to conditionToPainless (elastic#248102)

Resolves elastic/streams-program#618

## Summary

This expands `conditionToPainless` to handle fields containing
single-element arrays. When a field value is a List with exactly one
element, it gets unwrapped so comparisons work the same as with scalar
values.

## Details

The generated Painless now includes variable declarations with
unwrapping logic:

```
def val_foo = $('foo', null);> if (val_foo instanceof List && val_foo.size() == 1) { val_foo = val_foo[0]; }>
```

To implement this efficiently, we:

1. Extract all unique field names from the condition tree before
generating the statement
2. Generate a variable declaration block with the List unwrapping logic
for each field
3. Use variable references throughout the condition instead of repeated
inline $() accessors

This approach avoids duplicating the unwrapping logic for each field
reference in the generated script.

## Question for reviewers

- Should this be backported?

---------

Co-authored-by: kibanamachine <42973632+kibanamachine@users.noreply.github.com>

Author: CoenWarmer

x-pack/platform/packages/shared/kbn-streamlang-tests/test/scout/api/tests/cross_compatibility/filter_conditions.spec.ts

@@ -969,4 +969,57 @@ apiTest.describe('Cross-compatibility - Filter Conditions', { tag: ['@ess', '@sv
       })
     );
   });
+
+  apiTest('should handle single-element arrays in filter conditions', async ({ testBed, esql }) => {
+    const streamlangDSL: StreamlangDSL = {
+      steps: [
+        {
+          action: 'set',
+          to: 'attributes.matched',
+          value: 'yes',
+          where: {
+            field: 'attributes.tag',
+            eq: 'important',
+          },
+        } as SetProcessor,
+      ],
+    };
+
+    const { processors } = transpileIngestPipeline(streamlangDSL);
+    const { query } = transpileEsql(streamlangDSL);
+
+    const mappingDoc = { attributes: { tag: 'null', matched: 'null' } };
+    const docs = [
+      { attributes: { tag: ['important'] } }, // single-element array - should match
+      { attributes: { tag: 'important' } }, // regular string - should match
+      { attributes: { tag: ['other'] } }, // single-element array - should NOT match
+      { attributes: { tag: 'other' } }, // regular string - should NOT match
+    ];
+
+    await testBed.ingest('ingest-single-array', docs, processors);
+    const ingestResult = await testBed.getDocsOrdered('ingest-single-array');
+
+    await testBed.ingest('esql-single-array', [mappingDoc, ...docs]);
+    const esqlResult = await esql.queryOnIndex('esql-single-array', query);
+
+    // Single-element array ['important'] should match like 'important'
+    expect(ingestResult[0].attributes).toStrictEqual(expect.objectContaining({ matched: 'yes' }));
+    expect(ingestResult[1].attributes).toStrictEqual(expect.objectContaining({ matched: 'yes' }));
+    expect(ingestResult[2].attributes).not.toHaveProperty('matched');
+    expect(ingestResult[3].attributes).not.toHaveProperty('matched');
+
+    // ES|QL should produce the same results
+    expect(esqlResult.documentsOrdered[1]).toStrictEqual(
+      expect.objectContaining({ 'attributes.matched': 'yes' })
+    );
+    expect(esqlResult.documentsOrdered[2]).toStrictEqual(
+      expect.objectContaining({ 'attributes.matched': 'yes' })
+    );
+    expect(esqlResult.documentsOrdered[3]).toStrictEqual(
+      expect.objectContaining({ 'attributes.matched': null })
+    );
+    expect(esqlResult.documentsOrdered[4]).toStrictEqual(
+      expect.objectContaining({ 'attributes.matched': null })
+    );
+  });
 });

x-pack/platform/packages/shared/kbn-streamlang/src/conditions/condition_to_painless.test.ts

@@ -91,6 +91,31 @@ const operatorConditionAndResults = [
 ];
 
 describe('conditionToPainless', () => {
+  describe('single-element array unwrapping', () => {
+    test('should unwrap single-element arrays before comparison', () => {
+      const condition = { field: 'foo', eq: 'bar' };
+      const result = conditionToPainless(condition);
+
+      // Should contain List instanceof check and size check
+      expect(result).toContain('instanceof List');
+      expect(result).toContain('.size() == 1');
+    });
+
+    test('should handle multiple fields with array unwrapping', () => {
+      const condition = {
+        and: [
+          { field: 'foo', eq: 'bar' },
+          { field: 'baz', eq: 'qux' },
+        ],
+      };
+      const result = conditionToPainless(condition);
+
+      // Should contain List checks for both fields
+      expect(result).toContain('instanceof List');
+      expect(result).toContain('.size() == 1');
+    });
+  });
+
   describe('conditionToStatement', () => {
     describe('operators', () => {
       operatorConditionAndResults.forEach((setup) => {
@@ -300,7 +325,11 @@ describe('conditionToPainless', () => {
     expect(conditionToPainless(condition)).toMatchInlineSnapshot(`
       "
         try {
-        if ($('log', null) !== null) {
+        
+        def val_log = $('log', null); if (val_log instanceof List && val_log.size() == 1) { val_log = val_log[0]; }
+        
+        
+        if (val_log !== null) {
           return true;
         }
         return false;
@@ -326,7 +355,12 @@ describe('conditionToPainless', () => {
     expect(conditionToPainless(condition)).toMatchInlineSnapshot(`
       "
         try {
-        if (($('log.logger.name', null) !== null && (($('log.logger.name', null) instanceof Number && $('log.logger.name', null).toString() == \\"nginx_proxy\\") || $('log.logger.name', null) == \\"nginx_proxy\\")) && (($('log.level', null) !== null && (($('log.level', null) instanceof Number && $('log.level', null).toString() == \\"error\\") || $('log.level', null) == \\"error\\")) || ($('log.level', null) !== null && (($('log.level', null) instanceof Number && $('log.level', null).toString() == \\"ERROR\\") || $('log.level', null) == \\"ERROR\\")))) {
+        
+        def val_log_logger_name = $('log.logger.name', null); if (val_log_logger_name instanceof List && val_log_logger_name.size() == 1) { val_log_logger_name = val_log_logger_name[0]; }
+        def val_log_level = $('log.level', null); if (val_log_level instanceof List && val_log_level.size() == 1) { val_log_level = val_log_level[0]; }
+        
+        
+        if ((val_log_logger_name !== null && ((val_log_logger_name instanceof Number && val_log_logger_name.toString() == \\"nginx_proxy\\") || val_log_logger_name == \\"nginx_proxy\\")) && ((val_log_level !== null && ((val_log_level instanceof Number && val_log_level.toString() == \\"error\\") || val_log_level == \\"error\\")) || (val_log_level !== null && ((val_log_level instanceof Number && val_log_level.toString() == \\"ERROR\\") || val_log_level == \\"ERROR\\")))) {
           return true;
         }
         return false;

x-pack/platform/packages/shared/kbn-streamlang/src/conditions/condition_to_painless.ts

@@ -19,12 +19,48 @@ import { painlessFieldAccessor } from '../../types/utils';
 import { encodeValue } from '../../types/utils';
 import { evaluateDateMath } from './painless_date_math_helpers';
 
-// Utility: get the field name from a filter condition
-function safePainlessField(conditionOrField: FilterCondition | string) {
-  if (typeof conditionOrField === 'string') {
-    return painlessFieldAccessor(conditionOrField);
+// Type for mapping field names to variable names
+type FieldVarMap = Map<string, string>;
+
+// Extract all unique field names from a condition recursively
+function extractFieldNames(condition: Condition, fields: Set<string> = new Set()): Set<string> {
+  if ('field' in condition && typeof condition.field === 'string') {
+    fields.add(condition.field);
+  }
+  if ('and' in condition && Array.isArray(condition.and)) {
+    condition.and.forEach((c) => extractFieldNames(c, fields));
+  }
+  if ('or' in condition && Array.isArray(condition.or)) {
+    condition.or.forEach((c) => extractFieldNames(c, fields));
   }
-  return painlessFieldAccessor(conditionOrField.field);
+  if ('not' in condition && condition.not) {
+    extractFieldNames(condition.not, fields);
+  }
+  return fields;
+}
+
+// Convert a field name to a valid Painless variable name
+function fieldToVarName(field: string): string {
+  // Replace special characters with underscores and prefix with 'val_'
+  return 'val_' + field.replace(/[^a-zA-Z0-9]/g, '_');
+}
+
+// Generate variable declaration with single-element List unwrapping
+function generateFieldDeclaration(field: string, varName: string): string {
+  return `def ${varName} = $('${field}', null); if (${varName} instanceof List && ${varName}.size() == 1) { ${varName} = ${varName}[0]; }`;
+}
+
+// Utility: get the field accessor - uses varMap if provided, otherwise inline accessor
+function safePainlessField(conditionOrField: FilterCondition | string, varMap?: FieldVarMap) {
+  const fieldName =
+    typeof conditionOrField === 'string' ? conditionOrField : conditionOrField.field;
+
+  // If we have a varMap and it contains this field, use the variable name
+  if (varMap && varMap.has(fieldName)) {
+    return varMap.get(fieldName)!;
+  }
+
+  return painlessFieldAccessor(fieldName);
 }
 
 function generateRangeComparisonClauses(
@@ -64,8 +100,11 @@ function generateRangeComparisonClauses(
 }
 
 // Convert a shorthand binary filter condition to painless
-function shorthandBinaryToPainless(condition: ShorthandBinaryFilterCondition) {
-  const safeFieldAccessor = safePainlessField(condition);
+function shorthandBinaryToPainless(
+  condition: ShorthandBinaryFilterCondition,
+  varMap?: FieldVarMap
+) {
+  const safeFieldAccessor = safePainlessField(condition, varMap);
   // Find which operator is present
   const op = BINARY_OPERATORS.find((k) => condition[k] !== undefined);
 
@@ -161,12 +200,12 @@ function shorthandBinaryToPainless(condition: ShorthandBinaryFilterCondition) {
 }
 
 // Convert a shorthand unary filter condition to painless
-function shorthandUnaryToPainless(condition: ShorthandUnaryFilterCondition) {
+function shorthandUnaryToPainless(condition: ShorthandUnaryFilterCondition, varMap?: FieldVarMap) {
   if ('exists' in condition) {
     if (typeof condition.exists === 'boolean') {
       return condition.exists
-        ? `${safePainlessField(condition)} !== null`
-        : `${safePainlessField(condition)} == null`;
+        ? `${safePainlessField(condition, varMap)} !== null`
+        : `${safePainlessField(condition, varMap)} == null`;
     } else {
       throw new Error('Invalid value for exists operator, expected boolean');
     }
@@ -176,27 +215,36 @@ function shorthandUnaryToPainless(condition: ShorthandUnaryFilterCondition) {
 }
 
 // Main recursive conversion to painless
-export function conditionToStatement(condition: Condition, nested = false): string {
+export function conditionToStatement(
+  condition: Condition,
+  nested = false,
+  varMap?: FieldVarMap
+): string {
   if ('field' in condition && typeof condition.field === 'string') {
     // Shorthand unary
     if ('exists' in condition) {
-      return shorthandUnaryToPainless(condition as ShorthandUnaryFilterCondition);
+      return shorthandUnaryToPainless(condition as ShorthandUnaryFilterCondition, varMap);
     }
     // Shorthand binary
-    return `(${safePainlessField(condition)} !== null && ${shorthandBinaryToPainless(
-      condition as ShorthandBinaryFilterCondition
+    return `(${safePainlessField(condition, varMap)} !== null && ${shorthandBinaryToPainless(
+      condition as ShorthandBinaryFilterCondition,
+      varMap
     )})`;
   }
   if ('and' in condition && Array.isArray(condition.and)) {
-    const and = condition.and.map((filter) => conditionToStatement(filter, true)).join(' && ');
+    const and = condition.and
+      .map((filter) => conditionToStatement(filter, true, varMap))
+      .join(' && ');
     return nested ? `(${and})` : and;
   }
   if ('or' in condition && Array.isArray(condition.or)) {
-    const or = condition.or.map((filter) => conditionToStatement(filter, true)).join(' || ');
+    const or = condition.or
+      .map((filter) => conditionToStatement(filter, true, varMap))
+      .join(' || ');
     return nested ? `(${or})` : or;
   }
   if ('not' in condition && condition.not) {
-    return `!(${conditionToStatement(condition.not, true)})`;
+    return `!(${conditionToStatement(condition.not, true, varMap)})`;
   }
   // Always/never conditions (if you have them)
   if ('always' in condition) {
@@ -217,9 +265,27 @@ export function conditionToPainless(condition: Condition): string {
     return `return true`;
   }
 
+  // Extract all field names and create variable mappings
+  const fields = extractFieldNames(condition);
+  const varMap: FieldVarMap = new Map();
+  const declarations: string[] = [];
+
+  for (const field of fields) {
+    const varName = fieldToVarName(field);
+    varMap.set(field, varName);
+    declarations.push(generateFieldDeclaration(field, varName));
+  }
+
+  // declarationsBlock will look like this:
+  // def val_field1 = $('field1', null); if (val_field1 instanceof List && val_field1.size() == 1) { val_field1 = val_field1[0]; }
+  const declarationsBlock = declarations.length > 0 ? declarations.join('\n  ') + '\n  ' : '';
+
   return `
   try {
-  if (${conditionToStatement(condition)}) {
+  
+  ${declarationsBlock}
+  
+  if (${conditionToStatement(condition, false, varMap)}) {
     return true;
   }
   return false;