[Streams][Streamlang] Align ES|QL condition transpiler with Painless on null propagation

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

@@ -142,7 +142,7 @@ apiTest.describe(
     });
 
     apiTest(
-      'documents with missing condition field: ingest else-branch fires, esql excludes from both branches',
+      'documents with missing condition field: both transpilers fire the else-branch',
       async ({ testBed, esql }) => {
         const streamlangDSL: StreamlangDSL = {
           steps: [
@@ -198,12 +198,9 @@ apiTest.describe(
           expect.objectContaining({ 'attributes.outcome': 'else_branch' })
         );
 
-        // Missing-field doc: KNOWN DIVERGENCE between transpilers
-        // Painless: null check makes inner condition false, NOT(false) = true → else fires
         expect(asDoc(asDoc(ingestResult[2])?.attributes)?.outcome).toBe('else_branch');
-        // ES|QL: NULL propagation means NOT(NULL) = NULL → neither branch fires, outcome stays null
         expect(esqlResult.documentsOrdered[3]).toStrictEqual(
-          expect.objectContaining({ 'attributes.outcome': null })
+          expect.objectContaining({ 'attributes.outcome': 'else_branch' })
         );
       }
     );

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

@@ -121,6 +121,59 @@ apiTest.describe(
       );
     });
 
+    apiTest(
+      'neq fires on documents where the condition field is missing (parity with not(eq))',
+      async ({ testBed, esql }) => {
+        // A missing field "is not equal to" any concrete value, so `neq: "deleted"` must
+        // fire the `set` for documents without `attributes.status`. Both transpilers must
+        // agree, mirroring `not(eq)` and Query DSL semantics.
+        const streamlangDSL: StreamlangDSL = {
+          steps: [
+            {
+              action: 'set',
+              to: 'attributes.not_deleted',
+              value: 'kept',
+              where: {
+                field: 'attributes.status',
+                neq: 'deleted',
+              },
+            } as SetProcessor,
+          ],
+        };
+
+        const { processors } = await transpileIngestPipeline(streamlangDSL);
+        const { query } = await transpileEsql(streamlangDSL);
+
+        const mappingDoc = { attributes: { status: 'null', not_deleted: 'null' } };
+        const docs = [
+          { attributes: { status: 'deleted' } },
+          // Document with missing condition field — exercises the neq-on-missing case.
+          { attributes: { other: 'value' } },
+        ];
+
+        await testBed.ingest('ingest-neq-missing', docs, processors);
+        const ingestResult = await testBed.getDocsOrdered('ingest-neq-missing');
+
+        await testBed.ingest('esql-neq-missing', [mappingDoc, ...docs]);
+        const esqlResult = await esql.queryOnIndex('esql-neq-missing', query);
+
+        // Deleted doc: set does NOT fire (positive match against "deleted").
+        expect(asDoc(asDoc(ingestResult[0])?.attributes)?.not_deleted).toBeUndefined();
+        expect(esqlResult.documentsOrdered[1]).toStrictEqual(
+          expect.objectContaining({
+            'attributes.status': 'deleted',
+            'attributes.not_deleted': null,
+          })
+        );
+
+        // Missing-status doc: set DOES fire on both transpilers.
+        expect(asDoc(asDoc(ingestResult[1])?.attributes)?.not_deleted).toBe('kept');
+        expect(esqlResult.documentsOrdered[2]).toStrictEqual(
+          expect.objectContaining({ 'attributes.not_deleted': 'kept' })
+        );
+      }
+    );
+
     apiTest('should handle gt (greater than) filter condition', async ({ testBed, esql }) => {
       const streamlangDSL: StreamlangDSL = {
         steps: [

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

@@ -15,9 +15,11 @@ const operatorConditionAndResults = [
       "($('log.logger', null) !== null && (($('log.logger', null) instanceof Number && $('log.logger', null).toString() == \"nginx_proxy\") || $('log.logger', null) == \"nginx_proxy\"))",
   },
   {
+    // `neq` uses a disjunctive null-guard so a missing field counts as a match,
+    // mirroring ES|QL's `COALESCE(field != value, TRUE)` and Query DSL semantics.
     condition: { field: 'log.logger', neq: 'nginx_proxy' },
     result:
-      "($('log.logger', null) !== null && (($('log.logger', null) instanceof Number && $('log.logger', null).toString() != \"nginx_proxy\") || $('log.logger', null) != \"nginx_proxy\"))",
+      "($('log.logger', null) === null || (($('log.logger', null) instanceof Number && $('log.logger', null).toString() != \"nginx_proxy\") || $('log.logger', null) != \"nginx_proxy\"))",
   },
   {
     condition: { field: 'http.response.status_code', lt: 500 },
@@ -205,9 +207,8 @@ describe('conditionToPainless', () => {
           neq: false,
         };
 
-        // This test expects the field to be compared against boolean literal `false`
         expect(conditionToStatement(condition)).toEqual(
-          "($('is_active', null) !== null && (($('is_active', null) instanceof Number && $('is_active', null).toString() != \"false\") || $('is_active', null) != false))"
+          "($('is_active', null) === null || (($('is_active', null) instanceof Number && $('is_active', null).toString() != \"false\") || $('is_active', null) != false))"
         );
       });
 
@@ -229,11 +230,15 @@ describe('conditionToPainless', () => {
           neq: true,
         };
 
-        // This test expects the field to be compared against boolean literal `true`
         expect(conditionToStatement(condition)).toEqual(
-          "($('should_skip', null) !== null && (($('should_skip', null) instanceof Number && $('should_skip', null).toString() != \"true\") || $('should_skip', null) != true))"
+          "($('should_skip', null) === null || (($('should_skip', null) instanceof Number && $('should_skip', null).toString() != \"true\") || $('should_skip', null) != true))"
         );
       });
+
+      test('neq uses `=== null ||` so missing fields satisfy "not equals"', () => {
+        const condition = { field: 'status', neq: 'inactive' };
+        expect(conditionToStatement(condition)).toMatch(/^\(\$\('status', null\) === null \|\|/);
+      });
     });
 
     describe('and', () => {

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

@@ -234,8 +234,13 @@ export function conditionToStatement(
     if ('exists' in condition) {
       return shorthandUnaryToPainless(condition as ShorthandUnaryFilterCondition, varMap);
     }
-    // Shorthand binary
-    return `(${safePainlessField(condition, varMap)} !== null && ${shorthandBinaryToPainless(
+    // Shorthand binary. `neq` treats a missing field as a match (mirroring ES|QL's
+    // `COALESCE(field != value, TRUE)` and `not(eq)` semantics), so the null-guard is
+    // disjunctive (`=== null ||`) rather than conjunctive (`!== null &&`). Every other
+    // shorthand binary leaf treats a missing field as a non-match.
+    const isNeq = (condition as ShorthandBinaryFilterCondition).neq !== undefined;
+    const nullGuard = isNeq ? '=== null ||' : '!== null &&';
+    return `(${safePainlessField(condition, varMap)} ${nullGuard} ${shorthandBinaryToPainless(
       condition as ShorthandBinaryFilterCondition,
       varMap
     )})`;