diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/LookupQueryOperator.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/LookupQueryOperator.java
index a70384749898f..34dcaa314117c 100644
--- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/LookupQueryOperator.java
+++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/LookupQueryOperator.java
@@ -55,6 +55,7 @@ public final class LookupQueryOperator implements Operator {
     private final IndexSearcher searcher;
     private final Warnings warnings;
     private final int maxPageSize;
+    private final boolean emptyResult;
 
     private Page currentInputPage;
     private int queryPosition = -1;
@@ -77,7 +78,8 @@ public LookupQueryOperator(
         IndexedByShardId<? extends ShardContext> shardContexts,
         int shardId,
         SearchExecutionContext searchExecutionContext,
-        Warnings warnings
+        Warnings warnings,
+        boolean emptyResult
     ) {
         this.blockFactory = blockFactory;
         this.maxPageSize = maxPageSize;
@@ -86,6 +88,7 @@ public LookupQueryOperator(
         this.shardContext = shardContexts.get(shardId);
         this.shardContext.incRef();
         this.searchExecutionContext = searchExecutionContext;
+        this.emptyResult = emptyResult;
         try {
             if (shardContext.searcher().getIndexReader() instanceof DirectoryReader directoryReader) {
                 // This optimization is currently disabled for ParallelCompositeReader
@@ -105,10 +108,14 @@ public void addInput(Page page) {
         if (currentInputPage != null) {
             throw new IllegalStateException("Operator already has input page, must consume it first");
         }
-        currentInputPage = page;
-        queryPosition = -1; // Reset query position for new page
         pagesReceived++;
         rowsReceived += page.getPositionCount();
+        if (emptyResult) {
+            page.releaseBlocks();
+            return;
+        }
+        currentInputPage = page;
+        queryPosition = -1; // Reset query position for new page
     }
 
     @Override
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/lookup/LookupQueryOperatorTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/lookup/LookupQueryOperatorTests.java
index 423ea452926f4..2a1585bb1eb07 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/lookup/LookupQueryOperatorTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/lookup/LookupQueryOperatorTests.java
@@ -124,20 +124,21 @@ public Operator get(DriverContext driverContext) {
                     new IndexedByShardIdFromSingleton<>(new LuceneSourceOperatorTests.MockShardContext(directoryData.reader)),
                     0,
                     directoryData.searchExecutionContext,
-                    warnings()
+                    warnings(),
+                    false
                 );
             }
 
             @Override
             public String describe() {
-                return "LookupQueryOperator[maxPageSize=256]";
+                return "LookupQueryOperator[maxPageSize=256, emptyResult=false]";
             }
         };
     }
 
     @Override
     protected Matcher<String> expectedDescriptionOfSimple() {
-        return equalTo("LookupQueryOperator[maxPageSize=256]");
+        return equalTo("LookupQueryOperator[maxPageSize=256, emptyResult=false]");
     }
 
     @Override
@@ -180,7 +181,8 @@ public void testNoMatchesScenario() throws Exception {
                     new IndexedByShardIdFromSingleton<>(new LuceneSourceOperatorTests.MockShardContext(noMatchDirectory.reader)),
                     0,
                     noMatchDirectory.searchExecutionContext,
-                    warnings()
+                    warnings(),
+                    false
                 )
             ) {
                 // Create input with non-matching terms
@@ -237,7 +239,8 @@ public void testGetOutputNeverNullWhileCanProduceMore() throws Exception {
                 new IndexedByShardIdFromSingleton<>(new LuceneSourceOperatorTests.MockShardContext(directoryData.reader)),
                 0,
                 directoryData.searchExecutionContext,
-                warnings()
+                warnings(),
+                false
             )
         ) {
             // Create input with many matching terms
@@ -283,7 +286,8 @@ public void testMixedMatchesAndNoMatches() throws Exception {
                 new IndexedByShardIdFromSingleton<>(new LuceneSourceOperatorTests.MockShardContext(directoryData.reader)),
                 0,
                 directoryData.searchExecutionContext,
-                warnings()
+                warnings(),
+                false
             )
         ) {
             // Mix of matching and non-matching terms
@@ -325,6 +329,48 @@ public void testMixedMatchesAndNoMatches() throws Exception {
         }
     }
 
+    /**
+     * Test that when emptyResult=true the operator discards all input pages without producing output.
+     */
+    public void testEmptyResultDiscardsInput() {
+        DriverContext driverContext = driverContext();
+        QueryList queryList = QueryList.rawTermQueryList(directoryData.field, AliasFilter.EMPTY, 0, ElementType.BYTES_REF);
+
+        try (
+            LookupQueryOperator operator = new LookupQueryOperator(
+                driverContext.blockFactory(),
+                LookupQueryOperator.DEFAULT_MAX_PAGE_SIZE,
+                queryList,
+                new IndexedByShardIdFromSingleton<>(new LuceneSourceOperatorTests.MockShardContext(directoryData.reader)),
+                0,
+                directoryData.searchExecutionContext,
+                warnings(),
+                true
+            )
+        ) {
+            assertTrue("Should need input initially", operator.needsInput());
+            assertFalse("Should not be finished before finish() is called", operator.isFinished());
+
+            // Feed multiple pages with terms that would normally match
+            for (int p = 0; p < 3; p++) {
+                try (BytesRefBlock.Builder builder = driverContext.blockFactory().newBytesRefBlockBuilder(10)) {
+                    for (int i = 0; i < 10; i++) {
+                        builder.appendBytesRef(new BytesRef("term-" + i));
+                    }
+                    operator.addInput(new Page(builder.build()));
+                }
+
+                assertNull("Should never produce output when emptyResult=true", operator.getOutput());
+                assertFalse("Should not be able to produce more data", operator.canProduceMoreDataWithoutExtraInput());
+                assertTrue("Should still need input (not finished)", operator.needsInput());
+            }
+
+            operator.finish();
+            assertTrue("Should be finished after finish()", operator.isFinished());
+            assertNull("Should return null after finish()", operator.getOutput());
+        }
+    }
+
     private static Warnings warnings() {
         return Warnings.createWarnings(DriverContext.WarningsMode.COLLECT, new TestWarningsSource("test"));
     }
diff --git a/x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/EsqlTestUtils.java b/x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/EsqlTestUtils.java
index 9c2d42302d052..59be16fe38dde 100644
--- a/x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/EsqlTestUtils.java
+++ b/x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/EsqlTestUtils.java
@@ -469,6 +469,7 @@ public enum Config {
 
         private final Map<Config, Set<String>> includes = new HashMap<>();
         private final Map<Config, Set<String>> excludes = new HashMap<>();
+        private final Map<String, String> constantValues = new HashMap<>();
 
         public TestConfigurableSearchStats include(Config key, String... fields) {
             // If this method is called with no fields, it is interpreted to mean include none, so we include a dummy field
@@ -513,6 +514,16 @@ public boolean hasExactSubfield(FieldName field) {
             return isConfigationSet(Config.EXACT_SUBFIELD, field.string());
         }
 
+        public TestConfigurableSearchStats withConstantValue(String field, String value) {
+            constantValues.put(field, value);
+            return this;
+        }
+
+        @Override
+        public String constantValue(FieldName name) {
+            return constantValues.get(name.string());
+        }
+
         @Override
         public String toString() {
             return "TestConfigurableSearchStats{" + "includes=" + includes + ", excludes=" + excludes + '}';
diff --git a/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/LookupFromIndexIT.java b/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/LookupFromIndexIT.java
index a124bce3b5c2d..cd6169e3c07bd 100644
--- a/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/LookupFromIndexIT.java
+++ b/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/LookupFromIndexIT.java
@@ -127,27 +127,19 @@ public void testJoinOnFourKeys() throws IOException {
                     new String[] { "one", "two", "three", "four" },
                     new Integer[] { 1, 2, 3, 4 }, }
             ),
-            buildGreaterThanFilter(1L)
+            1L
         );
     }
 
     public void testLongKey() throws IOException {
-        runLookup(
-            List.of(DataType.LONG),
-            new UsingSingleLookupTable(new Object[][] { new Long[] { 12L, 33L, 1L } }),
-            buildGreaterThanFilter(0L)
-        );
+        runLookup(List.of(DataType.LONG), new UsingSingleLookupTable(new Object[][] { new Long[] { 12L, 33L, 1L } }), 0L);
     }
 
     /**
      * LOOKUP multiple results match.
      */
     public void testLookupIndexMultiResults() throws IOException {
-        runLookup(
-            List.of(DataType.KEYWORD),
-            new UsingSingleLookupTable(new Object[][] { new String[] { "aa", "bb", "bb", "dd" } }),
-            buildGreaterThanFilter(-1L)
-        );
+        runLookup(List.of(DataType.KEYWORD), new UsingSingleLookupTable(new Object[][] { new String[] { "aa", "bb", "bb", "dd" } }), -1L);
     }
 
     public void testJoinOnTwoKeysMultiResults() throws IOException {
@@ -236,19 +228,28 @@ public void populate(int docCount, List<String> expected, Predicate<Integer> fil
         }
     }
 
-    private PhysicalPlan buildGreaterThanFilter(long value) {
-        FieldAttribute filterAttribute = new FieldAttribute(
+    private static PhysicalPlan buildGreaterThanFilter(long value, FieldAttribute filterAttribute) {
+        Expression greaterThan = new GreaterThan(Source.EMPTY, filterAttribute, new Literal(Source.EMPTY, value, DataType.LONG));
+        EsRelation esRelation = new EsRelation(
             Source.EMPTY,
-            "l",
-            new EsField("l", DataType.LONG, Collections.emptyMap(), true, EsField.TimeSeriesFieldType.NONE)
+            "test",
+            IndexMode.LOOKUP,
+            Map.of(),
+            Map.of(),
+            Map.of(),
+            List.of(filterAttribute)
         );
-        Expression greaterThan = new GreaterThan(Source.EMPTY, filterAttribute, new Literal(Source.EMPTY, value, DataType.LONG));
-        EsRelation esRelation = new EsRelation(Source.EMPTY, "test", IndexMode.LOOKUP, Map.of(), Map.of(), Map.of(), List.of());
         Filter filter = new Filter(Source.EMPTY, esRelation, greaterThan);
         return new FragmentExec(filter);
     }
 
-    private void runLookup(List<DataType> keyTypes, PopulateIndices populateIndices, PhysicalPlan pushedDownFilter) throws IOException {
+    private void runLookup(List<DataType> keyTypes, PopulateIndices populateIndices, Long filterValue) throws IOException {
+        FieldAttribute lAttribute = new FieldAttribute(
+            Source.EMPTY,
+            "l",
+            new EsField("l", DataType.LONG, Collections.emptyMap(), true, EsField.TimeSeriesFieldType.NONE)
+        );
+        PhysicalPlan pushedDownFilter = filterValue != null ? buildGreaterThanFilter(filterValue, lAttribute) : null;
         String[] fieldMappers = new String[keyTypes.size() * 2];
         for (int i = 0; i < keyTypes.size(); i++) {
             fieldMappers[2 * i] = "key" + i;
@@ -403,13 +404,7 @@ private void runLookup(List<DataType> keyTypes, PopulateIndices populateIndices,
                 ctx -> internalCluster().getInstance(TransportEsqlQueryAction.class, finalNodeWithShard).getLookupFromIndexService(),
                 "lookup",
                 "lookup",
-                List.of(
-                    new FieldAttribute(
-                        Source.EMPTY,
-                        "l",
-                        new EsField("l", DataType.LONG, Collections.emptyMap(), true, EsField.TimeSeriesFieldType.NONE)
-                    )
-                ),
+                List.of(lAttribute),
                 Source.EMPTY,
                 pushedDownFilter,
                 Predicates.combineAnd(joinOnConditions),
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/LookupExecutionPlanner.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/LookupExecutionPlanner.java
index efa3499760f26..6b6ec960f443b 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/LookupExecutionPlanner.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/LookupExecutionPlanner.java
@@ -18,6 +18,7 @@
 import org.elasticsearch.compute.lucene.ShardContext;
 import org.elasticsearch.compute.lucene.read.ValuesSourceReaderOperator;
 import org.elasticsearch.compute.operator.DriverContext;
+import org.elasticsearch.compute.operator.EvalOperator.EvalOperatorFactory;
 import org.elasticsearch.compute.operator.FilterOperator;
 import org.elasticsearch.compute.operator.Operator;
 import org.elasticsearch.compute.operator.Operator.OperatorFactory;
@@ -37,6 +38,7 @@
 import org.elasticsearch.index.query.SearchExecutionContext;
 import org.elasticsearch.search.internal.AliasFilter;
 import org.elasticsearch.xpack.esql.EsqlIllegalArgumentException;
+import org.elasticsearch.xpack.esql.core.expression.Alias;
 import org.elasticsearch.xpack.esql.core.expression.Attribute;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
 import org.elasticsearch.xpack.esql.core.expression.Expressions;
@@ -45,6 +47,7 @@
 import org.elasticsearch.xpack.esql.core.tree.Source;
 import org.elasticsearch.xpack.esql.core.type.DataType;
 import org.elasticsearch.xpack.esql.evaluator.EvalMapper;
+import org.elasticsearch.xpack.esql.plan.physical.EvalExec;
 import org.elasticsearch.xpack.esql.plan.physical.FieldExtractExec;
 import org.elasticsearch.xpack.esql.plan.physical.FilterExec;
 import org.elasticsearch.xpack.esql.plan.physical.OutputExec;
@@ -271,6 +274,8 @@ private PhysicalOperation planLookupNode(
             );
         } else if (node instanceof FieldExtractExec fieldExtractExec) {
             return planFieldExtractExec(plannerSettings, fieldExtractExec, source);
+        } else if (node instanceof EvalExec evalExec) {
+            return planEvalExec(evalExec, source, foldCtx);
         } else if (node instanceof FilterExec filterExec) {
             return planFilterExec(filterExec, source, foldCtx);
         } else if (node instanceof ProjectExec projectExec) {
@@ -304,7 +309,8 @@ private PhysicalOperation planParameterizedQueryExec(
             parameterizedQueryExec.joinOnConditions(),
             parameterizedQueryExec.query(),
             lookupSource,
-            queryListFromPlanFactory
+            queryListFromPlanFactory,
+            parameterizedQueryExec.emptyResult()
         );
 
         return PhysicalOperation.fromSource(sourceFactory, layout).with(enrichQueryFactory, layout);
@@ -408,6 +414,16 @@ public String describe() {
         }, layout);
     }
 
+    private PhysicalOperation planEvalExec(EvalExec evalExec, PhysicalOperation source, FoldContext foldCtx) {
+        for (Alias field : evalExec.fields()) {
+            var evaluatorSupplier = EvalMapper.toEvaluator(foldCtx, field.child(), source.layout());
+            Layout.Builder layout = source.layout().builder();
+            layout.append(field.toAttribute());
+            source = source.with(new EvalOperatorFactory(evaluatorSupplier), layout.build());
+        }
+        return source;
+    }
+
     private PhysicalOperation planFilterExec(FilterExec filterExec, PhysicalOperation source, FoldContext foldCtx) {
         return source.with(
             new FilterOperator.FilterOperatorFactory(EvalMapper.toEvaluator(foldCtx, filterExec.condition(), source.layout())),
@@ -441,7 +457,8 @@ private record LookupQueryOperatorFactory(
         @Nullable Expression joinOnConditions,
         @Nullable QueryBuilder query,
         Source planSource,
-        QueryListFromPlanFactory queryListFromPlanFactory
+        QueryListFromPlanFactory queryListFromPlanFactory,
+        boolean emptyResult
     ) implements OperatorFactory {
         @Override
         public Operator get(DriverContext driverContext) {
@@ -468,13 +485,14 @@ public Operator get(DriverContext driverContext) {
                 shardContexts,
                 shardId,
                 searchExecutionContext,
-                warnings
+                warnings,
+                emptyResult
             );
         }
 
         @Override
         public String describe() {
-            return "LookupQueryOperator[maxPageSize=" + maxPageSize + "]";
+            return "LookupQueryOperator[maxPageSize=" + maxPageSize + ", emptyResult=" + emptyResult + "]";
         }
     }
 
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/LookupFromIndexService.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/LookupFromIndexService.java
index d6f592a458d72..0c7876db6ac47 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/LookupFromIndexService.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/LookupFromIndexService.java
@@ -57,7 +57,9 @@
 import org.elasticsearch.xpack.esql.expression.predicate.Predicates;
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput;
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamOutput;
+import org.elasticsearch.xpack.esql.optimizer.LocalLogicalOptimizerContext;
 import org.elasticsearch.xpack.esql.optimizer.LocalPhysicalOptimizerContext;
+import org.elasticsearch.xpack.esql.optimizer.LookupLogicalOptimizer;
 import org.elasticsearch.xpack.esql.optimizer.LookupPhysicalPlanOptimizer;
 import org.elasticsearch.xpack.esql.plan.logical.EsRelation;
 import org.elasticsearch.xpack.esql.plan.logical.Filter;
@@ -849,9 +851,10 @@ public static LogicalPlan buildLocalLogicalPlan(
     /**
      * Builds the output attributes for a {@link ParameterizedQuery}, mirroring how {@link EsRelation}
      * exposes all index fields. This ensures the logical verifier can validate that all field references
-     * in the plan are satisfied. At the physical level, {@code ReplaceSourceAttributes}
+     * in the plan are satisfied. At the physical level,
+     * {@link org.elasticsearch.xpack.esql.optimizer.rules.physical.local.ReplaceSourceAttributes ReplaceSourceAttributes}
      * strips the output back down to just {@code [_doc, _positions]}, and {@code InsertFieldExtraction}
-     * adds the needed fields back — the same pattern used for {@code EsRelation / ReplaceSourceAttributes}.
+     * adds the needed fields back — the same pattern used for {@code EsRelation}.
      */
     private static List<Attribute> buildParameterizedQueryOutput(
         FieldAttribute docAttribute,
@@ -879,7 +882,7 @@ private static List<Attribute> buildParameterizedQueryOutput(
 
     /**
      * Builds the physical plan for the lookup node by running:
-     * LocalMapper.map -> LookupPhysicalPlanOptimizer.
+     * LookupLogicalOptimizer -> LocalMapper.map -> LookupPhysicalPlanOptimizer.
      * The caller is responsible for building the logical plan via {@link #buildLocalLogicalPlan}.
      */
     public static PhysicalPlan createLookupPhysicalPlan(
@@ -890,7 +893,9 @@ public static PhysicalPlan createLookupPhysicalPlan(
         SearchStats searchStats,
         EsqlFlags flags
     ) {
-        PhysicalPlan physicalPlan = LocalMapper.INSTANCE.map(logicalPlan);
+        LogicalPlan optimizedLogical = new LookupLogicalOptimizer(new LocalLogicalOptimizerContext(configuration, foldCtx, searchStats))
+            .localOptimize(logicalPlan);
+        PhysicalPlan physicalPlan = LocalMapper.INSTANCE.map(optimizedLogical);
         LocalPhysicalOptimizerContext context = new LocalPhysicalOptimizerContext(
             plannerSettings,
             flags,
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/fulltext/FullTextFunction.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/fulltext/FullTextFunction.java
index 7e36c84545833..9884f6540f9be 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/fulltext/FullTextFunction.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/fulltext/FullTextFunction.java
@@ -51,6 +51,7 @@
 import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan;
 import org.elasticsearch.xpack.esql.plan.logical.MvExpand;
 import org.elasticsearch.xpack.esql.plan.logical.OrderBy;
+import org.elasticsearch.xpack.esql.plan.logical.ParameterizedQuery;
 import org.elasticsearch.xpack.esql.plan.logical.Project;
 import org.elasticsearch.xpack.esql.plan.logical.UnionAll;
 import org.elasticsearch.xpack.esql.plan.logical.join.InlineJoin;
@@ -264,7 +265,10 @@ private static void checkFullTextQueryFunctionForCondition(
                         plan,
                         condition,
                         functionClass,
-                        lp -> (lp instanceof Filter || lp instanceof OrderBy || lp instanceof EsRelation),
+                        lp -> (lp instanceof Filter
+                            || lp instanceof OrderBy
+                            || lp instanceof EsRelation
+                            || lp instanceof ParameterizedQuery),
                         fullTextFunction -> "[" + fullTextFunction.functionName() + "] " + fullTextFunction.functionType(),
                         failures
                     );
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/LocalLogicalPlanOptimizer.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/LocalLogicalPlanOptimizer.java
index bf45cdb2bb640..19dc6c7437932 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/LocalLogicalPlanOptimizer.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/LocalLogicalPlanOptimizer.java
@@ -77,7 +77,7 @@ private static Batch<LogicalPlan> localCleanup() {
     }
 
     @SuppressWarnings("unchecked")
-    private static Batch<LogicalPlan> localBatch(Batch<LogicalPlan> batch, Rule<?, LogicalPlan>... additionalRules) {
+    static Batch<LogicalPlan> localBatch(Batch<LogicalPlan> batch, Rule<?, LogicalPlan>... additionalRules) {
         Rule<?, LogicalPlan>[] rules = batch.rules();
 
         List<Rule<?, LogicalPlan>> newRules = new ArrayList<>(rules.length);
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/LookupLogicalOptimizer.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/LookupLogicalOptimizer.java
new file mode 100644
index 0000000000000..6f171d3d1b43e
--- /dev/null
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/LookupLogicalOptimizer.java
@@ -0,0 +1,58 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.optimizer;
+
+import org.elasticsearch.xpack.esql.optimizer.rules.logical.PruneFilters;
+import org.elasticsearch.xpack.esql.optimizer.rules.logical.ReplaceStringCasingWithInsensitiveRegexMatch;
+import org.elasticsearch.xpack.esql.optimizer.rules.logical.local.InferIsNotNull;
+import org.elasticsearch.xpack.esql.optimizer.rules.logical.local.LookupPruneFilters;
+import org.elasticsearch.xpack.esql.optimizer.rules.logical.local.ReplaceFieldWithConstantOrNull;
+import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan;
+import org.elasticsearch.xpack.esql.rule.Rule;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.elasticsearch.xpack.esql.optimizer.LogicalPlanOptimizer.operators;
+
+/**
+ * Logical plan optimizer for the lookup node. Extends {@link LocalLogicalPlanOptimizer} with a
+ * reduced rule set appropriate for lookup plans (rooted at
+ * {@link org.elasticsearch.xpack.esql.plan.logical.ParameterizedQuery}, not
+ * {@link org.elasticsearch.xpack.esql.plan.logical.EsRelation}).
+ *
+ * <p>The lookup logical plan is narrow: {@code Project -> optional Filter -> ParameterizedQuery}.
+ * This optimizer runs {@link ReplaceFieldWithConstantOrNull} to replace missing/constant fields,
+ * then the standard operator-optimization rules to fold nulls, simplify booleans, and prune filters.</p>
+ */
+public class LookupLogicalOptimizer extends LocalLogicalPlanOptimizer {
+
+    private static final List<Batch<LogicalPlan>> RULES = List.of(
+        new Batch<>("Lookup local rewrite", Limiter.ONCE, new ReplaceFieldWithConstantOrNull(), new InferIsNotNull()),
+        lookupOperators()
+    );
+
+    public LookupLogicalOptimizer(LocalLogicalOptimizerContext context) {
+        super(context);
+    }
+
+    @Override
+    protected List<Batch<LogicalPlan>> batches() {
+        return RULES;
+    }
+
+    @SuppressWarnings("unchecked")
+    private static Batch<LogicalPlan> lookupOperators() {
+        Batch<LogicalPlan> batch = localBatch(operators(), new ReplaceStringCasingWithInsensitiveRegexMatch());
+        List<Rule<?, LogicalPlan>> rules = new ArrayList<>(batch.rules().length);
+        for (Rule<?, LogicalPlan> r : batch.rules()) {
+            rules.add(r instanceof PruneFilters ? new LookupPruneFilters() : r);
+        }
+        return batch.with(rules.toArray(Rule[]::new));
+    }
+}
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/PruneFilters.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/PruneFilters.java
index 88c3b46549d42..bc4010d3d05c5 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/PruneFilters.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/PruneFilters.java
@@ -20,7 +20,7 @@
 import static org.elasticsearch.xpack.esql.core.expression.Literal.FALSE;
 import static org.elasticsearch.xpack.esql.core.expression.Literal.TRUE;
 
-public final class PruneFilters extends OptimizerRules.OptimizerRule<Filter> {
+public class PruneFilters extends OptimizerRules.OptimizerRule<Filter> {
     @Override
     protected LogicalPlan rule(Filter filter) {
         Expression condition = filter.condition().transformUp(BinaryLogic.class, PruneFilters::foldBinaryLogic);
@@ -30,7 +30,7 @@ protected LogicalPlan rule(Filter filter) {
                 return filter.child();
             }
             if (FALSE.equals(condition) || Expressions.isGuaranteedNull(condition)) {
-                return PruneEmptyPlans.skipPlan(filter);
+                return handleAlwaysFalseFilter(filter);
             }
         }
 
@@ -40,6 +40,15 @@ protected LogicalPlan rule(Filter filter) {
         return filter;
     }
 
+    /**
+     * Handles a filter whose condition has been folded to {@code false} or {@code null}.
+     * By default, collapses the plan via {@link PruneEmptyPlans#skipPlan}; subclasses may
+     * override to preserve plan structure when collapsing is not appropriate (e.g. lookup plans).
+     */
+    protected LogicalPlan handleAlwaysFalseFilter(Filter filter) {
+        return PruneEmptyPlans.skipPlan(filter);
+    }
+
     private static Expression foldBinaryLogic(BinaryLogic binaryLogic) {
         if (binaryLogic instanceof Or or) {
             boolean nullLeft = Expressions.isGuaranteedNull(or.left());
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/local/LookupPruneFilters.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/local/LookupPruneFilters.java
new file mode 100644
index 0000000000000..d59b05ed4898a
--- /dev/null
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/local/LookupPruneFilters.java
@@ -0,0 +1,34 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.optimizer.rules.logical.local;
+
+import org.elasticsearch.xpack.esql.optimizer.rules.logical.PruneFilters;
+import org.elasticsearch.xpack.esql.plan.logical.Filter;
+import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan;
+import org.elasticsearch.xpack.esql.plan.logical.ParameterizedQuery;
+
+/**
+ * Lookup-specific variant of {@link PruneFilters}. When a filter condition evaluates to {@code false}
+ * and the filter's subtree contains a {@link ParameterizedQuery}, marks it as {@code emptyResult=true}
+ * instead of collapsing the entire plan to a {@code LocalRelation}. This preserves the plan structure
+ * so the {@code LookupExecutionPlanner} can still build the operator chain.
+ */
+public class LookupPruneFilters extends PruneFilters {
+
+    @Override
+    protected LogicalPlan handleAlwaysFalseFilter(Filter filter) {
+        if (filter.anyMatch(n -> n instanceof ParameterizedQuery)) {
+            return filter.child()
+                .transformUp(
+                    ParameterizedQuery.class,
+                    pq -> new ParameterizedQuery(pq.source(), pq.output(), pq.matchFields(), pq.joinOnConditions(), true)
+                );
+        }
+        return super.handleAlwaysFalseFilter(filter);
+    }
+}
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/local/ReplaceFieldWithConstantOrNull.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/local/ReplaceFieldWithConstantOrNull.java
index 465c429371a25..27fe30ae590ea 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/local/ReplaceFieldWithConstantOrNull.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/local/ReplaceFieldWithConstantOrNull.java
@@ -26,6 +26,7 @@
 import org.elasticsearch.xpack.esql.plan.logical.Filter;
 import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan;
 import org.elasticsearch.xpack.esql.plan.logical.OrderBy;
+import org.elasticsearch.xpack.esql.plan.logical.ParameterizedQuery;
 import org.elasticsearch.xpack.esql.plan.logical.Project;
 import org.elasticsearch.xpack.esql.plan.logical.RegexExtract;
 import org.elasticsearch.xpack.esql.plan.logical.TopN;
@@ -64,22 +65,15 @@ public LogicalPlan apply(LogicalPlan plan, LocalLogicalOptimizerContext localLog
             }
             // find constant values only in the main indices
             else if (esRelation.indexMode() == IndexMode.STANDARD) {
-                for (Attribute attribute : esRelation.output()) {
-                    if (attribute instanceof FieldAttribute fa) {
-                        // Do not use the attribute name, this can deviate from the field name for union types; use fieldName() instead.
-                        String val = localLogicalOptimizerContext.searchStats().constantValue(fa.fieldName());
-                        if (val != null) {
-                            attrToConstant.put(attribute, Literal.of(attribute, BytesRefs.toBytesRef(val)));
-                        }
-                    } else if (attribute instanceof MetadataAttribute ma && ma.name().startsWith(PROJECT_METADATA_PREFIX)) {
-                        String val = localLogicalOptimizerContext.searchStats().constantValue(new FieldAttribute.FieldName(ma.name()));
-                        if (val != null) {
-                            attrToConstant.put(attribute, Literal.of(attribute, BytesRefs.toBytesRef(val)));
-                        }
-                    }
-                }
+                collectConstants(esRelation.output(), localLogicalOptimizerContext, attrToConstant);
             }
         });
+        // ParameterizedQuery only appears in the lookup-node plan (via LookupLogicalOptimizer);
+        // this is a no-op when the rule runs inside LocalLogicalPlanOptimizer on a data-node plan.
+        plan.forEachUp(
+            ParameterizedQuery.class,
+            paramQuery -> collectConstants(paramQuery.output(), localLogicalOptimizerContext, attrToConstant)
+        );
         AttributeSet lookupFields = lookupFieldsBuilder.build();
         AttributeSet externalFields = externalFieldsBuilder.build();
 
@@ -101,14 +95,15 @@ private LogicalPlan replaceWithNullOrConstant(
         Predicate<FieldAttribute> shouldBeRetained,
         Map<Attribute, Expression> attrToConstant
     ) {
-        if (plan instanceof EsRelation relation) {
-            // For any missing field, place an Eval right after the EsRelation to assign null values to that attribute (using the same name
-            // id!), thus avoiding that InsertFieldExtrations inserts a field extraction later.
+        if (plan instanceof EsRelation || plan instanceof ParameterizedQuery) {
+            // For any missing field, place an Eval right after the EsRelation/ParameterizedQuery
+            // to assign null values to that attribute (using the same name id!),
+            // thus avoiding that InsertFieldExtraction inserts a field extraction later.
             // This means that an EsRelation[field1, field2, field3] where field1 and field 3 are missing will be replaced by
             // Project[field1, field2, field3] <- keeps the ordering intact
             // \_Eval[field1 = null, field3 = null]
             // \_EsRelation[field1, field2, field3]
-            List<Attribute> relationOutput = relation.output();
+            List<Attribute> relationOutput = plan.output();
             var aliasedNulls = RuleUtils.aliasedNulls(
                 relationOutput,
                 attr -> attr instanceof FieldAttribute f && shouldBeRetained.test(f) == false
@@ -120,7 +115,7 @@ private LogicalPlan replaceWithNullOrConstant(
                 return plan;
             }
 
-            Eval eval = new Eval(plan.source(), relation, nullLiterals);
+            Eval eval = new Eval(plan.source(), plan, nullLiterals);
             // This projection is redundant if there's another projection downstream (and no commands depend on the order until we hit it).
             return new Project(plan.source(), eval, newProjections);
         }
@@ -154,4 +149,25 @@ private LogicalPlan replaceWithNullOrConstant(
 
         return plan;
     }
+
+    private static void collectConstants(
+        List<Attribute> output,
+        LocalLogicalOptimizerContext context,
+        Map<Attribute, Expression> attrToConstant
+    ) {
+        for (Attribute attribute : output) {
+            if (attribute instanceof FieldAttribute fa) {
+                // Do not use the attribute name, this can deviate from the field name for union types; use fieldName() instead.
+                String val = context.searchStats().constantValue(fa.fieldName());
+                if (val != null) {
+                    attrToConstant.put(attribute, Literal.of(attribute, BytesRefs.toBytesRef(val)));
+                }
+            } else if (attribute instanceof MetadataAttribute ma && ma.name().startsWith(PROJECT_METADATA_PREFIX)) {
+                String val = context.searchStats().constantValue(new FieldAttribute.FieldName(ma.name()));
+                if (val != null) {
+                    attrToConstant.put(attribute, Literal.of(attribute, BytesRefs.toBytesRef(val)));
+                }
+            }
+        }
+    }
 }
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/physical/local/ReplaceSourceAttributes.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/physical/local/ReplaceSourceAttributes.java
index bf061b8832f93..27aaaa7b72898 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/physical/local/ReplaceSourceAttributes.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/physical/local/ReplaceSourceAttributes.java
@@ -85,12 +85,19 @@ private static PhysicalPlan replaceParameterizedQuery(ParameterizedQueryExec pla
                 strippedOutput.add(attr);
             }
         }
-        return new ParameterizedQueryExec(plan.source(), strippedOutput, plan.matchFields(), plan.joinOnConditions(), plan.query());
+        return new ParameterizedQueryExec(
+            plan.source(),
+            strippedOutput,
+            plan.matchFields(),
+            plan.joinOnConditions(),
+            plan.query(),
+            plan.emptyResult()
+        );
     }
 
     private static Attribute getDocAttribute(EsSourceExec plan) {
-        // The source (or doc) field is sometimes added to the relation output as a hack to enable late materialization in the reduce
-        // driver. In that case, we should take it instead of replacing it with a new one to ensure the same attribute is used throughout.
+        // Reuse the existing doc attribute from the relation output when present, rather than creating a new one,
+        // to ensure the same attribute instance is used throughout the plan (needed for late materialization in the reduce driver).
         var sourceAttributes = plan.output().stream().filter(EsQueryExec::isDocAttribute).toList();
         if (sourceAttributes.size() > 1) {
             throw new IllegalStateException("Expected at most one source attribute, found: " + sourceAttributes);
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/ParameterizedQuery.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/ParameterizedQuery.java
index 6958dca7c39ee..75e4e52326d4d 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/ParameterizedQuery.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/ParameterizedQuery.java
@@ -39,12 +39,29 @@ public class ParameterizedQuery extends LeafPlan {
     private final List<MatchConfig> matchFields;
     @Nullable
     private final Expression joinOnConditions;
+    /**
+     * Runtime-only flag set by the {@link org.elasticsearch.xpack.esql.optimizer.LookupLogicalOptimizer}
+     * when a filter folds to {@code false}/{@code null}. Not serialized — it is computed locally on the
+     * lookup node after deserialization.
+     */
+    private final boolean emptyResult;
 
     public ParameterizedQuery(Source source, List<Attribute> output, List<MatchConfig> matchFields, @Nullable Expression joinOnConditions) {
+        this(source, output, matchFields, joinOnConditions, false);
+    }
+
+    public ParameterizedQuery(
+        Source source,
+        List<Attribute> output,
+        List<MatchConfig> matchFields,
+        @Nullable Expression joinOnConditions,
+        boolean emptyResult
+    ) {
         super(source);
         this.output = output;
         this.matchFields = matchFields;
         this.joinOnConditions = joinOnConditions;
+        this.emptyResult = emptyResult;
     }
 
     private static ParameterizedQuery readFrom(StreamInput in) throws IOException {
@@ -69,6 +86,10 @@ public Expression joinOnConditions() {
         return joinOnConditions;
     }
 
+    public boolean emptyResult() {
+        return emptyResult;
+    }
+
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         Source.EMPTY.writeTo(out);
@@ -89,12 +110,12 @@ public boolean expressionsResolved() {
 
     @Override
     protected NodeInfo<ParameterizedQuery> info() {
-        return NodeInfo.create(this, ParameterizedQuery::new, output, matchFields, joinOnConditions);
+        return NodeInfo.create(this, ParameterizedQuery::new, output, matchFields, joinOnConditions, emptyResult);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(output, matchFields, joinOnConditions);
+        return Objects.hash(output, matchFields, joinOnConditions, emptyResult);
     }
 
     @Override
@@ -108,6 +129,7 @@ public boolean equals(Object obj) {
         ParameterizedQuery other = (ParameterizedQuery) obj;
         return Objects.equals(output, other.output)
             && Objects.equals(matchFields, other.matchFields)
-            && Objects.equals(joinOnConditions, other.joinOnConditions);
+            && Objects.equals(joinOnConditions, other.joinOnConditions)
+            && emptyResult == other.emptyResult;
     }
 }
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/physical/ParameterizedQueryExec.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/physical/ParameterizedQueryExec.java
index 1e7ecd46ee080..5d8afb1756491 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/physical/ParameterizedQueryExec.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/physical/ParameterizedQueryExec.java
@@ -34,6 +34,7 @@ public class ParameterizedQueryExec extends LeafExec {
     private final Expression joinOnConditions;
     @Nullable
     private final QueryBuilder query;
+    private final boolean emptyResult;
 
     public ParameterizedQueryExec(
         Source source,
@@ -41,12 +42,24 @@ public ParameterizedQueryExec(
         List<MatchConfig> matchFields,
         @Nullable Expression joinOnConditions,
         @Nullable QueryBuilder query
+    ) {
+        this(source, output, matchFields, joinOnConditions, query, false);
+    }
+
+    public ParameterizedQueryExec(
+        Source source,
+        List<Attribute> output,
+        List<MatchConfig> matchFields,
+        @Nullable Expression joinOnConditions,
+        @Nullable QueryBuilder query,
+        boolean emptyResult
     ) {
         super(source);
         this.output = output;
         this.matchFields = matchFields;
         this.joinOnConditions = joinOnConditions;
         this.query = query;
+        this.emptyResult = emptyResult;
     }
 
     @Override
@@ -68,10 +81,14 @@ public QueryBuilder query() {
         return query;
     }
 
+    public boolean emptyResult() {
+        return emptyResult;
+    }
+
     public ParameterizedQueryExec withQuery(QueryBuilder query) {
         return Objects.equals(this.query, query)
             ? this
-            : new ParameterizedQueryExec(source(), output, matchFields, joinOnConditions, query);
+            : new ParameterizedQueryExec(source(), output, matchFields, joinOnConditions, query, emptyResult);
     }
 
     @Override
@@ -86,7 +103,7 @@ public String getWriteableName() {
 
     @Override
     protected NodeInfo<ParameterizedQueryExec> info() {
-        return NodeInfo.create(this, ParameterizedQueryExec::new, output, matchFields, joinOnConditions, query);
+        return NodeInfo.create(this, ParameterizedQueryExec::new, output, matchFields, joinOnConditions, query, emptyResult);
     }
 
     @Override
@@ -97,11 +114,12 @@ public boolean equals(Object o) {
         return Objects.equals(output, that.output)
             && Objects.equals(matchFields, that.matchFields)
             && Objects.equals(joinOnConditions, that.joinOnConditions)
-            && Objects.equals(query, that.query);
+            && Objects.equals(query, that.query)
+            && emptyResult == that.emptyResult;
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(output, matchFields, joinOnConditions, query);
+        return Objects.hash(output, matchFields, joinOnConditions, query, emptyResult);
     }
 }
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/mapper/LocalMapper.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/mapper/LocalMapper.java
index 59ed80155ac60..9b7c6b6c9e05a 100644
--- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/mapper/LocalMapper.java
+++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/mapper/LocalMapper.java
@@ -77,7 +77,7 @@ private PhysicalPlan mapLeaf(LeafPlan leaf) {
         }
 
         if (leaf instanceof ParameterizedQuery pq) {
-            return new ParameterizedQueryExec(pq.source(), pq.output(), pq.matchFields(), pq.joinOnConditions(), null);
+            return new ParameterizedQueryExec(pq.source(), pq.output(), pq.matchFields(), pq.joinOnConditions(), null, pq.emptyResult());
         }
 
         if (leaf instanceof ExternalRelation external) {
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/enrich/LookupFromIndexOperatorTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/enrich/LookupFromIndexOperatorTests.java
index 3a5d2283fd153..d6189dd3a1ea2 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/enrich/LookupFromIndexOperatorTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/enrich/LookupFromIndexOperatorTests.java
@@ -64,6 +64,7 @@
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.esql.action.EsqlCapabilities;
+import org.elasticsearch.xpack.esql.core.expression.Attribute;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
 import org.elasticsearch.xpack.esql.core.expression.FieldAttribute;
 import org.elasticsearch.xpack.esql.core.expression.FoldContext;
@@ -90,7 +91,6 @@
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Locale;
@@ -282,7 +282,7 @@ protected Operator.OperatorFactory simple(SimpleOptions options) {
             matchFields.add(new MatchConfig(matchField, i, inputDataType));
         }
         Expression joinOnExpression = null;
-        FragmentExec rightPlanWithOptionalPreJoinFilter = buildLessThanFilter(LESS_THAN_VALUE);
+        FragmentExec rightPlanWithOptionalPreJoinFilter = buildLessThanFilter(LESS_THAN_VALUE, loadFields);
         if (operation != null) {
             List<Expression> conditions = new ArrayList<>();
             for (int i = 0; i < numberOfJoinColumns; i++) {
@@ -329,14 +329,15 @@ protected Operator.OperatorFactory simple(SimpleOptions options) {
         );
     }
 
-    static FragmentExec buildLessThanFilter(int value) {
-        FieldAttribute filterAttribute = new FieldAttribute(
-            Source.EMPTY,
-            "lint",
-            new EsField("lint", DataType.INTEGER, Collections.emptyMap(), true, EsField.TimeSeriesFieldType.NONE)
-        );
+    static FragmentExec buildLessThanFilter(int value, List<NamedExpression> loadFields) {
+        FieldAttribute filterAttribute = loadFields.stream()
+            .filter(f -> f.name().equals("lint"))
+            .map(FieldAttribute.class::cast)
+            .findFirst()
+            .orElseThrow();
         Expression lessThan = new LessThan(Source.EMPTY, filterAttribute, new Literal(Source.EMPTY, value, DataType.INTEGER));
-        EsRelation esRelation = new EsRelation(Source.EMPTY, "test", IndexMode.LOOKUP, Map.of(), Map.of(), Map.of(), List.of());
+        List<Attribute> attrs = loadFields.stream().map(f -> (Attribute) f).toList();
+        EsRelation esRelation = new EsRelation(Source.EMPTY, "test", IndexMode.LOOKUP, Map.of(), Map.of(), Map.of(), attrs);
         Filter filter = new Filter(Source.EMPTY, esRelation, lessThan);
         return new FragmentExec(filter);
     }
@@ -387,7 +388,7 @@ protected Matcher<String> expectedToStringOfSimple() {
                 .append("Filter\\[lint\\{f}#\\d+ < ")
                 .append(LESS_THAN_VALUE)
                 .append("\\[INTEGER]]\\n")
-                .append("\\\\_EsRelation\\[test]\\[LOOKUP]\\[\\]<>\\]\\]");
+                .append("\\\\_EsRelation\\[test]\\[LOOKUP]\\[lkwd\\{f}#\\d+, lint\\{f}#\\d+]<>\\]\\]");
             sb.append(")");
         }
 
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/enrich/StreamingLookupFromIndexOperatorTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/enrich/StreamingLookupFromIndexOperatorTests.java
index 56c0dcf1f1a78..374c2af4b5748 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/enrich/StreamingLookupFromIndexOperatorTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/enrich/StreamingLookupFromIndexOperatorTests.java
@@ -199,7 +199,7 @@ protected Operator.OperatorFactory simple(SimpleOptions options, int maxOutstand
             matchFields.add(new MatchConfig(matchField, i, inputDataType));
         }
         Expression joinOnExpression = null;
-        FragmentExec rightPlanWithOptionalPreJoinFilter = LookupFromIndexOperatorTests.buildLessThanFilter(LESS_THAN_VALUE);
+        FragmentExec rightPlanWithOptionalPreJoinFilter = LookupFromIndexOperatorTests.buildLessThanFilter(LESS_THAN_VALUE, loadFields);
         if (operation != null) {
             List<Expression> conditions = new ArrayList<>();
             for (int i = 0; i < numberOfJoinColumns; i++) {
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LookupLogicalOptimizerTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LookupLogicalOptimizerTests.java
new file mode 100644
index 0000000000000..0555252687819
--- /dev/null
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LookupLogicalOptimizerTests.java
@@ -0,0 +1,276 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.optimizer;
+
+import org.elasticsearch.index.IndexMode;
+import org.elasticsearch.index.mapper.MapperServiceTestCase;
+import org.elasticsearch.license.XPackLicenseState;
+import org.elasticsearch.xpack.esql.EsqlTestUtils;
+import org.elasticsearch.xpack.esql.analysis.Analyzer;
+import org.elasticsearch.xpack.esql.analysis.EnrichResolution;
+import org.elasticsearch.xpack.esql.analysis.Verifier;
+import org.elasticsearch.xpack.esql.core.expression.FieldAttribute;
+import org.elasticsearch.xpack.esql.core.expression.FoldContext;
+import org.elasticsearch.xpack.esql.core.expression.NamedExpression;
+import org.elasticsearch.xpack.esql.core.type.EsField;
+import org.elasticsearch.xpack.esql.enrich.LookupFromIndexService;
+import org.elasticsearch.xpack.esql.enrich.MatchConfig;
+import org.elasticsearch.xpack.esql.expression.function.EsqlFunctionRegistry;
+import org.elasticsearch.xpack.esql.index.EsIndex;
+import org.elasticsearch.xpack.esql.index.EsIndexGenerator;
+import org.elasticsearch.xpack.esql.plan.logical.Eval;
+import org.elasticsearch.xpack.esql.plan.logical.Filter;
+import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan;
+import org.elasticsearch.xpack.esql.plan.logical.ParameterizedQuery;
+import org.elasticsearch.xpack.esql.plan.logical.Project;
+import org.elasticsearch.xpack.esql.plan.physical.LookupJoinExec;
+import org.elasticsearch.xpack.esql.plan.physical.PhysicalPlan;
+import org.elasticsearch.xpack.esql.stats.SearchStats;
+import org.elasticsearch.xpack.esql.telemetry.Metrics;
+import org.junit.Before;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.elasticsearch.xpack.esql.EsqlTestUtils.TEST_CFG;
+import static org.elasticsearch.xpack.esql.EsqlTestUtils.TEST_SEARCH_STATS;
+import static org.elasticsearch.xpack.esql.EsqlTestUtils.as;
+import static org.elasticsearch.xpack.esql.EsqlTestUtils.emptyInferenceResolution;
+import static org.elasticsearch.xpack.esql.EsqlTestUtils.loadMapping;
+import static org.elasticsearch.xpack.esql.EsqlTestUtils.testAnalyzerContext;
+import static org.elasticsearch.xpack.esql.EsqlTestUtils.withDefaultLimitWarning;
+import static org.elasticsearch.xpack.esql.analysis.AnalyzerTestUtils.defaultLookupResolution;
+import static org.elasticsearch.xpack.esql.analysis.AnalyzerTestUtils.indexResolutions;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.is;
+
+/**
+ * Tests for {@link LookupLogicalOptimizer}, verifying that logical optimization rules are applied
+ * to the lookup node's logical plan before physical planning.
+ */
+public class LookupLogicalOptimizerTests extends MapperServiceTestCase {
+
+    private Analyzer analyzer;
+    private TestPlannerOptimizer plannerOptimizer;
+
+    @Before
+    public void init() {
+        Map<String, EsField> mapping = loadMapping("mapping-basic.json");
+        EsIndex test = EsIndexGenerator.esIndex("test", mapping, Map.of("test", IndexMode.STANDARD));
+
+        analyzer = new Analyzer(
+            testAnalyzerContext(
+                TEST_CFG,
+                new EsqlFunctionRegistry(),
+                indexResolutions(test),
+                defaultLookupResolution(),
+                new EnrichResolution(),
+                emptyInferenceResolution()
+            ),
+            new Verifier(new Metrics(new EsqlFunctionRegistry(), true, true), new XPackLicenseState(() -> 0L))
+        );
+        plannerOptimizer = new TestPlannerOptimizer(TEST_CFG, analyzer);
+    }
+
+    @Override
+    protected List<String> filteredWarnings() {
+        return withDefaultLimitWarning(super.filteredWarnings());
+    }
+
+    /**
+     * Simple lookup with no filters.
+     * Expects: Project -> ParameterizedQuery
+     */
+    public void testSimpleLookup() {
+        LogicalPlan plan = optimizeLookupLogicalPlan("FROM test | LOOKUP JOIN test_lookup ON emp_no", TEST_SEARCH_STATS);
+
+        Project project = as(plan, Project.class);
+        ParameterizedQuery pq = as(project.child(), ParameterizedQuery.class);
+        assertFalse("Expected emptyResult=false on ParameterizedQuery", pq.emptyResult());
+    }
+
+    /**
+     * Filter referencing an existing field should be preserved.
+     * Expects: Project -> Filter -> ParameterizedQuery
+     */
+    public void testFilterOnExistingField() {
+        LogicalPlan plan = optimizeLookupLogicalPlan("""
+            FROM test
+            | RENAME languages AS language_code
+            | LOOKUP JOIN languages_lookup ON language_code
+            | WHERE language_name == "English"
+            """, TEST_SEARCH_STATS);
+
+        Project project = as(plan, Project.class);
+        Filter filter = as(project.child(), Filter.class);
+        ParameterizedQuery pq = as(filter.child(), ParameterizedQuery.class);
+        assertFalse("Expected emptyResult=false on ParameterizedQuery", pq.emptyResult());
+    }
+
+    /**
+     * Filter referencing a missing field should be folded away (the condition becomes null/false).
+     * ReplaceFieldWithConstantOrNull replaces the missing field with null, then LookupPruneFilters
+     * marks the ParameterizedQuery as emptyResult instead of collapsing the plan to LocalRelation,
+     * preserving the plan structure for the LookupExecutionPlanner.
+     * Expects: Project -> Eval -> ParameterizedQuery(emptyResult=true)
+     */
+    public void testFilterOnMissingFieldFolded() {
+        EsqlTestUtils.TestConfigurableSearchStats stats = new EsqlTestUtils.TestConfigurableSearchStats().exclude(
+            EsqlTestUtils.TestConfigurableSearchStats.Config.EXISTS,
+            "language_name"
+        );
+
+        LogicalPlan plan = optimizeLookupLogicalPlan("""
+            FROM test
+            | RENAME languages AS language_code
+            | LOOKUP JOIN languages_lookup ON language_code
+            | WHERE language_name == "English"
+            """, stats);
+
+        Project project = as(plan, Project.class);
+        Eval eval = as(project.child(), Eval.class);
+        ParameterizedQuery pq = as(eval.child(), ParameterizedQuery.class);
+        assertTrue("Expected emptyResult=true on ParameterizedQuery", pq.emptyResult());
+    }
+
+    /**
+     * Filter that becomes always-true due to missing field stats should be pruned.
+     * "language_name IS NULL" with language_name missing → "null IS NULL" → true → filter removed.
+     * Expects: Project -> Eval -> ParameterizedQuery
+     * See {@link LookupPhysicalPlanOptimizerTests#testDropMissingFieldPrunesEval} for verification that the Eval is removed during
+     * physical optimization.
+     */
+    public void testFilterOnMissingFieldFoldedToTrue() {
+        EsqlTestUtils.TestConfigurableSearchStats stats = new EsqlTestUtils.TestConfigurableSearchStats().exclude(
+            EsqlTestUtils.TestConfigurableSearchStats.Config.EXISTS,
+            "language_name"
+        );
+
+        LogicalPlan plan = optimizeLookupLogicalPlan("""
+            FROM test
+            | RENAME languages AS language_code
+            | LOOKUP JOIN languages_lookup ON language_code
+            | WHERE language_name IS NULL
+            """, stats);
+
+        Project project = as(plan, Project.class);
+        Eval eval = as(project.child(), Eval.class);
+        ParameterizedQuery pq = as(eval.child(), ParameterizedQuery.class);
+        assertFalse("Expected emptyResult=false on ParameterizedQuery", pq.emptyResult());
+    }
+
+    /**
+     * Constant field matching the filter value: {@code language_name} is a constant {@code "English"},
+     * and the filter is {@code WHERE language_name == "English"}.  The constant replaces the field reference,
+     * the filter folds to {@code true} and is pruned.
+     * Expects: Project -> ParameterizedQuery (no Filter, no Eval since the field exists and is constant).
+     */
+    public void testConstantFieldMatchingFilter() {
+        EsqlTestUtils.TestConfigurableSearchStats stats = new EsqlTestUtils.TestConfigurableSearchStats().withConstantValue(
+            "language_name",
+            "English"
+        );
+
+        LogicalPlan plan = optimizeLookupLogicalPlan("""
+            FROM test
+            | RENAME languages AS language_code
+            | LOOKUP JOIN languages_lookup ON language_code
+            | WHERE language_name == "English"
+            """, stats);
+
+        Project project = as(plan, Project.class);
+        ParameterizedQuery pq = as(project.child(), ParameterizedQuery.class);
+        assertFalse("Expected emptyResult=false on ParameterizedQuery", pq.emptyResult());
+    }
+
+    /**
+     * Constant field NOT matching the filter value: {@code language_name} is a constant {@code "Spanish"},
+     * but the filter is {@code WHERE language_name == "English"}.  The constant replaces the field reference,
+     * the filter folds to {@code false}, and LookupPruneFilters marks the ParameterizedQuery as emptyResult.
+     * Expects: Project -> ParameterizedQuery(emptyResult=true)
+     */
+    public void testConstantFieldMismatchFoldsToEmpty() {
+        EsqlTestUtils.TestConfigurableSearchStats stats = new EsqlTestUtils.TestConfigurableSearchStats().withConstantValue(
+            "language_name",
+            "Spanish"
+        );
+
+        LogicalPlan plan = optimizeLookupLogicalPlan("""
+            FROM test
+            | RENAME languages AS language_code
+            | LOOKUP JOIN languages_lookup ON language_code
+            | WHERE language_name == "English"
+            """, stats);
+
+        Project project = as(plan, Project.class);
+        ParameterizedQuery pq = as(project.child(), ParameterizedQuery.class);
+        assertTrue("Expected emptyResult=true on ParameterizedQuery", pq.emptyResult());
+    }
+
+    private LogicalPlan optimizeLookupLogicalPlan(String esql, SearchStats searchStats) {
+        List<LogicalPlan> plans = optimizeAllLookupLogicalPlans(esql, searchStats);
+        assertThat("Expected exactly one LOOKUP JOIN", plans, hasSize(1));
+        return plans.getFirst();
+    }
+
+    /**
+     * Runs the full planning pipeline, finds LookupJoinExec nodes, then builds and logically optimizes
+     * each lookup plan. Returns the optimized logical plans in tree traversal order.
+     */
+    private List<LogicalPlan> optimizeAllLookupLogicalPlans(String esql, SearchStats searchStats) {
+        PhysicalPlan dataNodePlan = plannerOptimizer.plan(esql);
+
+        List<LookupJoinExec> joins = findAllLookupJoins(dataNodePlan);
+        assertThat("Expected at least one LookupJoinExec in the plan", joins.isEmpty(), is(false));
+
+        List<LogicalPlan> lookupPlans = new ArrayList<>(joins.size());
+        for (LookupJoinExec join : joins) {
+            lookupPlans.add(buildAndOptimizeLookupLogicalPlan(join, searchStats));
+        }
+        return lookupPlans;
+    }
+
+    private static LogicalPlan buildAndOptimizeLookupLogicalPlan(LookupJoinExec join, SearchStats searchStats) {
+        List<MatchConfig> matchFields = new ArrayList<>(join.leftFields().size());
+        for (int i = 0; i < join.leftFields().size(); i++) {
+            FieldAttribute right = (FieldAttribute) join.rightFields().get(i);
+            String fieldName = right.exactAttribute().fieldName().string();
+            if (join.isOnJoinExpression()) {
+                fieldName = join.leftFields().get(i).name();
+            }
+            matchFields.add(new MatchConfig(fieldName, i, join.leftFields().get(i).dataType()));
+        }
+
+        LogicalPlan logicalPlan = LookupFromIndexService.buildLocalLogicalPlan(
+            join.source(),
+            matchFields,
+            join.joinOnConditions(),
+            join.right(),
+            join.addedFields().stream().map(f -> (NamedExpression) f).toList()
+        );
+
+        var context = new LocalLogicalOptimizerContext(TEST_CFG, FoldContext.small(), searchStats);
+        return new LookupLogicalOptimizer(context).localOptimize(logicalPlan);
+    }
+
+    private static List<LookupJoinExec> findAllLookupJoins(PhysicalPlan plan) {
+        List<LookupJoinExec> joins = new ArrayList<>();
+        collectLookupJoins(plan, joins);
+        return joins;
+    }
+
+    private static void collectLookupJoins(PhysicalPlan plan, List<LookupJoinExec> joins) {
+        if (plan instanceof LookupJoinExec join) {
+            joins.add(join);
+        }
+        for (PhysicalPlan child : plan.children()) {
+            collectLookupJoins(child, joins);
+        }
+    }
+}
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LookupPhysicalPlanOptimizerTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LookupPhysicalPlanOptimizerTests.java
index 96dea551c9b8d..2a5cfe6a323fe 100644
--- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LookupPhysicalPlanOptimizerTests.java
+++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LookupPhysicalPlanOptimizerTests.java
@@ -12,6 +12,7 @@
 import org.elasticsearch.index.mapper.MapperServiceTestCase;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.license.XPackLicenseState;
+import org.elasticsearch.xpack.esql.EsqlTestUtils;
 import org.elasticsearch.xpack.esql.action.EsqlCapabilities;
 import org.elasticsearch.xpack.esql.analysis.Analyzer;
 import org.elasticsearch.xpack.esql.analysis.EnrichResolution;
@@ -48,6 +49,7 @@
 import org.elasticsearch.xpack.esql.planner.PlannerSettings;
 import org.elasticsearch.xpack.esql.plugin.EsqlFlags;
 import org.elasticsearch.xpack.esql.session.Configuration;
+import org.elasticsearch.xpack.esql.stats.SearchStats;
 import org.elasticsearch.xpack.esql.telemetry.Metrics;
 import org.junit.Before;
 
@@ -72,7 +74,7 @@
 
 /**
  * Tests for {@link LookupPhysicalPlanOptimizer}, verifying that the lookup-node planning pipeline
- * (buildLocalLogicalPlan -> LocalMapper -> LookupPhysicalPlanOptimizer) produces correct physical plans.
+ * (buildLocalLogicalPlan -> LookupLogicalOptimizer -> LocalMapper -> LookupPhysicalPlanOptimizer) produces correct physical plans.
  */
 public class LookupPhysicalPlanOptimizerTests extends MapperServiceTestCase {
 
@@ -116,6 +118,7 @@ public void testSimpleLookup() {
         ParameterizedQueryExec paramQuery = as(fieldExtract.child(), ParameterizedQueryExec.class);
 
         assertThat(paramQuery.query(), nullValue());
+        assertThat(paramQuery.emptyResult(), is(false));
         assertThat(fieldExtract.attributesToExtract().isEmpty(), is(false));
     }
 
@@ -139,6 +142,7 @@ public void testPushableRightOnlyFilter() {
         QueryBuilder query = paramQuery.query();
         assertNotNull("Expected filter to be pushed to ParameterizedQueryExec", query);
         assertThat(query.toString(), containsString("language_name"));
+        assertThat(paramQuery.emptyResult(), is(false));
     }
 
     /**
@@ -160,6 +164,7 @@ public void testNonPushableRightOnlyFilter() {
         ParameterizedQueryExec paramQuery = as(fieldExtract.child(), ParameterizedQueryExec.class);
 
         assertThat(paramQuery.query(), nullValue());
+        assertThat(paramQuery.emptyResult(), is(false));
         assertThat(filter.condition().toString(), containsString("LENGTH"));
     }
 
@@ -182,6 +187,7 @@ public void testMixedPushableAndNonPushableFilters() {
 
         assertNotNull("Expected pushable filter on ParameterizedQueryExec", paramQuery.query());
         assertThat(paramQuery.query().toString(), containsString("language_name"));
+        assertThat(paramQuery.emptyResult(), is(false));
         assertThat(filter.condition().toString(), containsString("LENGTH"));
     }
 
@@ -211,6 +217,7 @@ public void testOnExpressionFilterWithWhereClause() {
         // language_name == "English" (pushable ON right-only filter) is pushed to query
         assertNotNull("Expected pushable ON filter on ParameterizedQueryExec", paramQuery.query());
         assertThat(paramQuery.query().toString(), containsString("language_name"));
+        assertThat(paramQuery.emptyResult(), is(false));
 
         // joinOnConditions is the left-right join key comparison
         assertNotNull("Expected join on conditions", paramQuery.joinOnConditions());
@@ -262,6 +269,83 @@ public void testPushFilterThroughEvalExec() {
 
         assertNotNull("Expected filter pushed through EvalExec to ParameterizedQueryExec", paramQuery.query());
         assertThat(paramQuery.query().toString(), containsString("language_name"));
+        assertThat(paramQuery.emptyResult(), is(false));
+    }
+
+    /**
+     * Filter that becomes always-true due to missing field stats should be pruned during logical optimization.
+     * "language_name IS NULL" with language_name missing → "null IS NULL" → true → filter removed.
+     * The null Eval from ReplaceFieldWithConstantOrNull should be pruned by the physical optimizer since
+     * the field is not needed for extraction.
+     */
+    public void testFilterOnMissingFieldFoldedToTrue() {
+        EsqlTestUtils.TestConfigurableSearchStats stats = new EsqlTestUtils.TestConfigurableSearchStats().exclude(
+            EsqlTestUtils.TestConfigurableSearchStats.Config.EXISTS,
+            "language_name"
+        );
+
+        PhysicalPlan plan = optimizeLookupPlan("""
+            FROM test
+            | RENAME languages AS language_code
+            | LOOKUP JOIN languages_lookup ON language_code
+            | WHERE language_name IS NULL
+            """, stats);
+
+        ProjectExec project = as(plan, ProjectExec.class);
+        EvalExec eval = as(project.child(), EvalExec.class);
+        ParameterizedQueryExec paramQuery = as(eval.child(), ParameterizedQueryExec.class);
+        assertThat("Filter should have been pruned, no query on ParameterizedQueryExec", paramQuery.query(), nullValue());
+        assertThat(paramQuery.emptyResult(), is(false));
+    }
+
+    /**
+     * Filter on a missing field with equality (e.g. {@code language_name == "English"}) folds to {@code null == "English"} → null,
+     * which marks the {@link ParameterizedQueryExec} as {@code emptyResult=true} instead of collapsing the plan.
+     */
+    public void testFilterOnMissingFieldFoldedToEmpty() {
+        EsqlTestUtils.TestConfigurableSearchStats stats = new EsqlTestUtils.TestConfigurableSearchStats().exclude(
+            EsqlTestUtils.TestConfigurableSearchStats.Config.EXISTS,
+            "language_name"
+        );
+
+        PhysicalPlan plan = optimizeLookupPlan("""
+            FROM test
+            | RENAME languages AS language_code
+            | LOOKUP JOIN languages_lookup ON language_code
+            | WHERE language_name == "English"
+            """, stats);
+
+        ProjectExec project = as(plan, ProjectExec.class);
+        EvalExec eval = as(project.child(), EvalExec.class);
+        ParameterizedQueryExec paramQuery = as(eval.child(), ParameterizedQueryExec.class);
+        assertThat(paramQuery.emptyResult(), is(true));
+    }
+
+    /**
+     * When a missing field is dropped from the output, it never appears in the lookup plan's addedFields,
+     * so no null Eval is needed. Using expression-based join so that language_code remains as an
+     * extractable added field after dropping language_name.
+     * Expects: ProjectExec -> FieldExtractExec -> ParameterizedQueryExec
+     */
+    public void testDropMissingFieldPrunesEval() {
+        assumeTrue("Requires LOOKUP JOIN on expression", EsqlCapabilities.Cap.LOOKUP_JOIN_WITH_FULL_TEXT_FUNCTION.isEnabled());
+
+        EsqlTestUtils.TestConfigurableSearchStats stats = new EsqlTestUtils.TestConfigurableSearchStats().exclude(
+            EsqlTestUtils.TestConfigurableSearchStats.Config.EXISTS,
+            "language_name"
+        );
+
+        PhysicalPlan plan = optimizeLookupPlan("""
+            FROM test
+            | LOOKUP JOIN languages_lookup ON languages == language_code
+            | DROP language_name
+            """, stats, ESQL_LOOKUP_JOIN_FULL_TEXT_FUNCTION);
+
+        ProjectExec project = as(plan, ProjectExec.class);
+        FieldExtractExec fieldExtract = as(project.child(), FieldExtractExec.class);
+        ParameterizedQueryExec paramQuery = as(fieldExtract.child(), ParameterizedQueryExec.class);
+        assertThat(paramQuery.query(), nullValue());
+        assertThat(paramQuery.emptyResult(), is(false));
     }
 
     /**
@@ -274,7 +358,7 @@ public void testTwoLookupJoins() {
             | LOOKUP JOIN test_lookup ON emp_no
             | RENAME languages AS language_code
             | LOOKUP JOIN languages_lookup ON language_code
-            """, null);
+            """, null, TEST_SEARCH_STATS);
 
         assertThat(plans, hasSize(2));
 
@@ -284,6 +368,7 @@ public void testTwoLookupJoins() {
         FieldExtractExec extract0 = as(project0.child(), FieldExtractExec.class);
         ParameterizedQueryExec paramQuery0 = as(extract0.child(), ParameterizedQueryExec.class);
         assertThat(paramQuery0.query(), nullValue());
+        assertThat(paramQuery0.emptyResult(), is(false));
 
         // Inner join (test_lookup) is found second
         PhysicalPlan testPlan = plans.get(1);
@@ -291,16 +376,25 @@ public void testTwoLookupJoins() {
         FieldExtractExec extract1 = as(project1.child(), FieldExtractExec.class);
         ParameterizedQueryExec paramQuery1 = as(extract1.child(), ParameterizedQueryExec.class);
         assertThat(paramQuery1.query(), nullValue());
+        assertThat(paramQuery1.emptyResult(), is(false));
     }
 
     private PhysicalPlan optimizeLookupPlan(String esql) {
-        List<PhysicalPlan> plans = optimizeAllLookupPlans(esql, null);
+        return optimizeLookupPlan(esql, TEST_SEARCH_STATS);
+    }
+
+    private PhysicalPlan optimizeLookupPlan(String esql, SearchStats searchStats) {
+        List<PhysicalPlan> plans = optimizeAllLookupPlans(esql, null, searchStats);
         assertThat("Expected exactly one LOOKUP JOIN", plans, hasSize(1));
         return plans.getFirst();
     }
 
     private PhysicalPlan optimizeLookupPlan(String esql, TransportVersion minVersion) {
-        List<PhysicalPlan> plans = optimizeAllLookupPlans(esql, minVersion);
+        return optimizeLookupPlan(esql, TEST_SEARCH_STATS, minVersion);
+    }
+
+    private PhysicalPlan optimizeLookupPlan(String esql, SearchStats searchStats, TransportVersion minVersion) {
+        List<PhysicalPlan> plans = optimizeAllLookupPlans(esql, minVersion, searchStats);
         assertThat("Expected exactly one LOOKUP JOIN", plans, hasSize(1));
         return plans.getFirst();
     }
@@ -309,7 +403,7 @@ private PhysicalPlan optimizeLookupPlan(String esql, TransportVersion minVersion
      * Runs the full planning pipeline and returns a lookup-node physical plan for each LookupJoinExec
      * found in the data-node plan. The plans are returned in tree traversal order (outermost join first).
      */
-    private List<PhysicalPlan> optimizeAllLookupPlans(String esql, TransportVersion minVersion) {
+    private List<PhysicalPlan> optimizeAllLookupPlans(String esql, TransportVersion minVersion, SearchStats searchStats) {
         PhysicalPlan dataNodePlan;
         if (minVersion != null) {
             MutableAnalyzerContext mutableContext = (MutableAnalyzerContext) analyzer.context();
@@ -327,12 +421,12 @@ private List<PhysicalPlan> optimizeAllLookupPlans(String esql, TransportVersion
 
         List<PhysicalPlan> lookupPlans = new ArrayList<>(joins.size());
         for (LookupJoinExec join : joins) {
-            lookupPlans.add(buildLookupPlan(join));
+            lookupPlans.add(buildLookupPlan(join, searchStats));
         }
         return lookupPlans;
     }
 
-    private static PhysicalPlan buildLookupPlan(LookupJoinExec join) {
+    private static PhysicalPlan buildLookupPlan(LookupJoinExec join, SearchStats searchStats) {
         List<MatchConfig> matchFields = new ArrayList<>(join.leftFields().size());
         for (int i = 0; i < join.leftFields().size(); i++) {
             FieldAttribute right = (FieldAttribute) join.rightFields().get(i);
@@ -355,7 +449,7 @@ private static PhysicalPlan buildLookupPlan(LookupJoinExec join) {
             TEST_CFG,
             PlannerSettings.DEFAULTS,
             FoldContext.small(),
-            TEST_SEARCH_STATS,
+            searchStats,
             new EsqlFlags(true)
         );
     }
diff --git a/x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/190_lookup_join.yml b/x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/190_lookup_join.yml
index 052f854def2ba..92a579c9cb39e 100644
--- a/x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/190_lookup_join.yml
+++ b/x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/190_lookup_join.yml
@@ -95,6 +95,22 @@ setup:
                 type: keyword
               shade:
                 type: keyword
+  - do:
+      indices.create:
+        index: test-lookup-const
+        body:
+          settings:
+            index:
+              mode: lookup
+          mappings:
+            properties:
+              key:
+                type: long
+              color:
+                type: keyword
+              const_tag:
+                type: constant_keyword
+                value: "alpha"
   - do:
       indices.create:
         index: test2
@@ -167,6 +183,15 @@ setup:
           - { "key": 20, "color": "blue" , shade: "dark" }
           - { "index": { } }
           - { "key": 30, "color": "pink" , shade: "hot" }
+  - do:
+      bulk:
+        index: "test-lookup-const"
+        refresh: true
+        body:
+          - { "index": { } }
+          - { "key": 1, "color": "cyan" }
+          - { "index": { } }
+          - { "key": 2, "color": "yellow" }
   - do:
       bulk:
         index: "test2"
@@ -344,3 +369,48 @@ basic join on two fields:
     #for 20 and yellow, no rows match, but we keep the row as it is a lookup join
   - match: { values.2: [ 20, "yellow", null ] }
 
+---
+constant field filter matches:
+  - requires:
+      capabilities:
+        - method: POST
+          path: /_query
+          parameters: [ ]
+          capabilities: [ join_lookup_v12 ]
+      reason: "uses LOOKUP JOIN"
+  - do:
+      esql.query:
+        body:
+          query: 'FROM test | LOOKUP JOIN test-lookup-const ON key | WHERE const_tag == "alpha" | SORT key | LIMIT 10'
+
+  - match: { columns.0.name: "key" }
+  - match: { columns.0.type: "long" }
+  - match: { columns.1.name: "color" }
+  - match: { columns.1.type: "keyword" }
+  - match: { columns.2.name: "const_tag" }
+  - match: { columns.2.type: "keyword" }
+  - match: { values.0: [ 1, "cyan", "alpha" ] }
+  - match: { values.1: [ 2, "yellow", "alpha" ] }
+
+---
+constant field filter does not match:
+  - requires:
+      capabilities:
+        - method: POST
+          path: /_query
+          parameters: [ ]
+          capabilities: [ join_lookup_v12 ]
+      reason: "uses LOOKUP JOIN"
+  - do:
+      esql.query:
+        body:
+          query: 'FROM test | LOOKUP JOIN test-lookup-const ON key | WHERE const_tag == "beta" | SORT key | LIMIT 10'
+
+  - match: { columns.0.name: "key" }
+  - match: { columns.0.type: "long" }
+  - match: { columns.1.name: "color" }
+  - match: { columns.1.type: "keyword" }
+  - match: { columns.2.name: "const_tag" }
+  - match: { columns.2.type: "keyword" }
+  - length: { values: 0 }
+