Support PPL foreach command

core/src/main/java/org/opensearch/sql/analysis/Analyzer.java

@@ -74,6 +74,7 @@
 import org.opensearch.sql.ast.tree.FillNull;
 import org.opensearch.sql.ast.tree.Filter;
 import org.opensearch.sql.ast.tree.Flatten;
+import org.opensearch.sql.ast.tree.Foreach;
 import org.opensearch.sql.ast.tree.GraphLookup;
 import org.opensearch.sql.ast.tree.Head;
 import org.opensearch.sql.ast.tree.Join;
@@ -571,6 +572,11 @@ public LogicalPlan visitGraphLookup(GraphLookup node, AnalysisContext context) {
     throw getOnlyForCalciteException("graphlookup");
   }
 
+  @Override
+  public LogicalPlan visitForeach(Foreach node, AnalysisContext context) {
+    throw getOnlyForCalciteException("foreach");
+  }
+
   /** Build {@link ParseExpression} to context and skip to child nodes. */
   @Override
   public LogicalPlan visitParse(Parse node, AnalysisContext context) {

core/src/main/java/org/opensearch/sql/ast/AbstractNodeVisitor.java

@@ -18,6 +18,7 @@
 import org.opensearch.sql.ast.expression.Compare;
 import org.opensearch.sql.ast.expression.EqualTo;
 import org.opensearch.sql.ast.expression.Field;
+import org.opensearch.sql.ast.expression.ForeachPlaceholder;
 import org.opensearch.sql.ast.expression.Function;
 import org.opensearch.sql.ast.expression.HighlightFunction;
 import org.opensearch.sql.ast.expression.In;
@@ -62,6 +63,7 @@
 import org.opensearch.sql.ast.tree.FillNull;
 import org.opensearch.sql.ast.tree.Filter;
 import org.opensearch.sql.ast.tree.Flatten;
+import org.opensearch.sql.ast.tree.Foreach;
 import org.opensearch.sql.ast.tree.GraphLookup;
 import org.opensearch.sql.ast.tree.Head;
 import org.opensearch.sql.ast.tree.Join;
@@ -173,6 +175,10 @@ public T visitProject(Project node, C context) {
     return visitChildren(node, context);
   }
 
+  public T visitForeach(Foreach node, C context) {
+    return visitChildren(node, context);
+  }
+
   public T visitAggregation(Aggregation node, C context) {
     return visitChildren(node, context);
   }
@@ -253,6 +259,10 @@ public T visitField(Field node, C context) {
     return visitChildren(node, context);
   }
 
+  public T visitForeachPlaceholder(ForeachPlaceholder node, C context) {
+    return visitChildren(node, context);
+  }
+
   public T visitQualifiedName(QualifiedName node, C context) {
     return visitChildren(node, context);
   }

core/src/main/java/org/opensearch/sql/ast/expression/ForeachPlaceholder.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.ast.expression;
+
+import com.google.common.collect.ImmutableList;
+import java.util.List;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.ToString;
+import org.opensearch.sql.ast.AbstractNodeVisitor;
+
+/** Placeholder used by the PPL foreach command, such as {@code <<FIELD>>}. */
+@Getter
+@ToString
+@EqualsAndHashCode(callSuper = false)
+@RequiredArgsConstructor
+public class ForeachPlaceholder extends UnresolvedExpression {
+  private final String name;
+
+  @Override
+  public List<UnresolvedExpression> getChild() {
+    return ImmutableList.of();
+  }
+
+  @Override
+  public <R, C> R accept(AbstractNodeVisitor<R, C> nodeVisitor, C context) {
+    return nodeVisitor.visitForeachPlaceholder(this, context);
+  }
+}

core/src/main/java/org/opensearch/sql/ast/tree/Foreach.java

@@ -0,0 +1,77 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.ast.tree;
+
+import com.google.common.collect.ImmutableList;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.ToString;
+import org.opensearch.sql.ast.AbstractNodeVisitor;
+import org.opensearch.sql.ast.expression.UnresolvedExpression;
+
+/** AST node representing the PPL foreach command. */
+@Getter
+@ToString
+@EqualsAndHashCode(callSuper = false)
+@RequiredArgsConstructor
+public class Foreach extends UnresolvedPlan {
+  private final Mode mode;
+  private final Map<String, String> options;
+  private final List<String> fieldPatterns;
+  private final UnresolvedExpression collectionExpression;
+  private final List<ForeachEvalClause> evalClauses;
+  private UnresolvedPlan child;
+
+  @Override
+  public Foreach attach(UnresolvedPlan child) {
+    this.child = child;
+    return this;
+  }
+
+  @Override
+  public List<UnresolvedPlan> getChild() {
+    return child == null ? ImmutableList.of() : ImmutableList.of(child);
+  }
+
+  @Override
+  public <T, C> T accept(AbstractNodeVisitor<T, C> nodeVisitor, C context) {
+    return nodeVisitor.visitForeach(this, context);
+  }
+
+  /** Iteration mode of the foreach command. */
+  public enum Mode {
+    MULTIFIELD,
+    MULTIVALUE,
+    JSON_ARRAY,
+    AUTO_COLLECTIONS;
+
+    public static Mode of(String name) {
+      try {
+        return valueOf(name.toUpperCase(Locale.ROOT));
+      } catch (IllegalArgumentException e) {
+        throw new IllegalArgumentException("foreach mode [" + name + "] is not supported");
+      }
+    }
+
+    @Override
+    public String toString() {
+      return name().toLowerCase(Locale.ROOT);
+    }
+  }
+
+  @Getter
+  @ToString
+  @EqualsAndHashCode
+  @RequiredArgsConstructor
+  public static class ForeachEvalClause {
+    private final String targetTemplate;
+    private final UnresolvedExpression expression;
+  }
+}

core/src/main/java/org/opensearch/sql/calcite/CalcitePlanContext.java

@@ -17,10 +17,12 @@
 import java.util.function.BiFunction;
 import lombok.Getter;
 import lombok.Setter;
+import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rex.RexCorrelVariable;
 import org.apache.calcite.rex.RexLambdaRef;
 import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.tools.FrameworkConfig;
+import org.checkerframework.checker.nullness.qual.Nullable;
 import org.opensearch.sql.ast.expression.AggregateFunction;
 import org.opensearch.sql.ast.expression.Function;
 import org.opensearch.sql.ast.expression.UnresolvedExpression;
@@ -66,6 +68,17 @@ public class CalcitePlanContext {
 
   @Getter public Map<String, RexLambdaRef> rexLambdaRefMap;
 
+  /**
+   * Foreach placeholder bindings active in this context, keyed by upper-cased placeholder name.
+   * Multifield mode activates bindings directly (placeholders resolve against the current row);
+   * collection modes stage bindings via {@link #stageForeachLambdaBindings} instead, so they only
+   * become active inside the lambda context cloned for the generated {@code reduce} call.
+   */
+  @Getter private Map<String, ForeachBinding> foreachBindings = new HashMap<>();
+
+  /** Bindings that become active in lambda contexts cloned from this one. */
+  private Map<String, ForeachBinding> stagedForeachLambdaBindings = new HashMap<>();
+
   /**
    * Maps AggregateFunction AST nodes to their output field index for HAVING/post-aggregate
    * resolution.
@@ -113,6 +126,9 @@ private CalcitePlanContext(CalcitePlanContext parent) {
     this.rexLambdaRefMap = new HashMap<>(); // New map for lambda variables
     this.capturedVariables = new ArrayList<>(); // New list for captured variables
     this.inLambdaContext = true; // Mark that we're inside a lambda
+    // Active bindings carry over; staged bindings become active inside the lambda.
+    this.foreachBindings = new HashMap<>(parent.foreachBindings);
+    this.foreachBindings.putAll(parent.stagedForeachLambdaBindings);
   }
 
   public RexNode resolveJoinCondition(
@@ -179,6 +195,37 @@ public static boolean isLegacyPreferred() {
     return legacyPreferredFlag.get();
   }
 
+  public void pushForeachBindings(Map<String, ForeachBinding> bindings) {
+    foreachBindings = new HashMap<>(bindings);
+  }
+
+  public void stageForeachLambdaBindings(Map<String, ForeachBinding> bindings) {
+    stagedForeachLambdaBindings = new HashMap<>(bindings);
+  }
+
+  public void clearForeachBindings() {
+    foreachBindings.clear();
+    stagedForeachLambdaBindings.clear();
+  }
+
+  /**
+   * A foreach placeholder binding. {@code FIELD} resolves to the named row field, {@code LITERAL}
+   * to a string literal, and {@code PAIR_SLOT} to slot {@code pairIndex} (typed {@code pairType})
+   * of the named lambda pair variable.
+   */
+  public record ForeachBinding(
+      String value, ForeachBindingType type, int pairIndex, @Nullable RelDataType pairType) {
+    public ForeachBinding(String value, ForeachBindingType type) {
+      this(value, type, -1, null);
+    }
+  }
+
+  public enum ForeachBindingType {
+    FIELD,
+    LITERAL,
+    PAIR_SLOT
+  }
+
   public void putRexLambdaRefMap(Map<String, RexLambdaRef> candidateMap) {
     this.rexLambdaRefMap.putAll(candidateMap);
   }

core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java

@@ -129,6 +129,7 @@
 import org.opensearch.sql.ast.tree.FillNull;
 import org.opensearch.sql.ast.tree.Filter;
 import org.opensearch.sql.ast.tree.Flatten;
+import org.opensearch.sql.ast.tree.Foreach;
 import org.opensearch.sql.ast.tree.GraphLookup;
 import org.opensearch.sql.ast.tree.GraphLookup.Direction;
 import org.opensearch.sql.ast.tree.Head;
@@ -210,12 +211,14 @@ public class CalciteRelNodeVisitor extends AbstractNodeVisitor<RelNode, CalciteP
   private final CalciteAggCallVisitor aggVisitor;
   private final DataSourceService dataSourceService;
   private final MapPathPreMaterializer mapPathMaterializer;
+  private final ForeachPlanner foreachPlanner;
 
   public CalciteRelNodeVisitor(DataSourceService dataSourceService) {
     this.rexVisitor = new CalciteRexNodeVisitor(this);
     this.aggVisitor = new CalciteAggCallVisitor(rexVisitor);
     this.dataSourceService = dataSourceService;
     this.mapPathMaterializer = new MapPathPreMaterializer(rexVisitor);
+    this.foreachPlanner = new ForeachPlanner(this, rexVisitor);
   }
 
   public RelNode analyze(UnresolvedPlan unresolved, CalcitePlanContext context) {
@@ -1239,6 +1242,12 @@ public RelNode visitEval(Eval node, CalcitePlanContext context) {
     return context.relBuilder.peek();
   }
 
+  @Override
+  public RelNode visitForeach(Foreach node, CalcitePlanContext context) {
+    visitChildren(node, context);
+    return foreachPlanner.plan(node, context);
+  }
+
   @Override
   public RelNode visitConvert(Convert node, CalcitePlanContext context) {
     visitChildren(node, context);
@@ -1370,7 +1379,7 @@ private RelNode buildConversionProjection(ConversionState state, CalcitePlanCont
     return context.relBuilder.peek();
   }
 
-  private void projectPlusOverriding(
+  void projectPlusOverriding(
       List<RexNode> newFields, List<String> newNames, CalcitePlanContext context) {
     Set<String> originalFieldNameSet =
         new HashSet<>(context.relBuilder.peek().getRowType().getFieldNames());

core/src/main/java/org/opensearch/sql/calcite/CalciteRexNodeVisitor.java

@@ -53,6 +53,7 @@
 import org.opensearch.sql.ast.expression.Cast;
 import org.opensearch.sql.ast.expression.Compare;
 import org.opensearch.sql.ast.expression.EqualTo;
+import org.opensearch.sql.ast.expression.ForeachPlaceholder;
 import org.opensearch.sql.ast.expression.Function;
 import org.opensearch.sql.ast.expression.HighlightFunction;
 import org.opensearch.sql.ast.expression.In;
@@ -79,6 +80,7 @@
 import org.opensearch.sql.ast.tree.Sort.SortOption;
 import org.opensearch.sql.ast.tree.Sort.SortOrder;
 import org.opensearch.sql.ast.tree.UnresolvedPlan;
+import org.opensearch.sql.calcite.CalcitePlanContext.ForeachBinding;
 import org.opensearch.sql.calcite.plan.rel.LogicalSystemLimit;
 import org.opensearch.sql.calcite.plan.rel.LogicalSystemLimit.SystemLimitType;
 import org.opensearch.sql.calcite.utils.OpenSearchTypeFactory;
@@ -92,6 +94,7 @@
 import org.opensearch.sql.exception.SemanticCheckException;
 import org.opensearch.sql.expression.function.BuiltinFunctionName;
 import org.opensearch.sql.expression.function.CoercionUtils;
+import org.opensearch.sql.expression.function.PPLBuiltinOperators;
 import org.opensearch.sql.expression.function.PPLFuncImpTable;
 
 @RequiredArgsConstructor
@@ -399,9 +402,51 @@ private boolean isBooleanLiteral(RexNode node) {
   /** Resolve qualified name. Note, the name should be case-sensitive. */
   @Override
   public RexNode visitQualifiedName(QualifiedName node, CalcitePlanContext context) {
+    ForeachBinding binding = foreachBinding(node.toString(), context);
+    if (binding != null) {
+      return foreachBindingToRexNode(node.toString(), binding, context);
+    }
     return QualifiedNameResolver.resolve(node, context);
   }
 
+  @Override
+  public RexNode visitForeachPlaceholder(ForeachPlaceholder node, CalcitePlanContext context) {
+    ForeachBinding binding = foreachBinding(node.getName(), context);
+    if (binding == null) {
+      throw new SemanticCheckException("Unresolved foreach placeholder <<" + node.getName() + ">>");
+    }
+    return foreachBindingToRexNode(node.getName(), binding, context);
+  }
+
+  private ForeachBinding foreachBinding(String name, CalcitePlanContext context) {
+    return context.getForeachBindings().get(name.toUpperCase(Locale.ROOT));
+  }
+
+  private RexNode foreachBindingToRexNode(
+      String name, ForeachBinding binding, CalcitePlanContext context) {
+    switch (binding.type()) {
+      case FIELD:
+        return context.relBuilder.field(binding.value());
+      case PAIR_SLOT:
+        RexLambdaRef pair = context.getRexLambdaRefMap().get(binding.value());
+        if (pair == null) {
+          throw new SemanticCheckException("Unresolved foreach lambda placeholder " + name);
+        }
+        // The slot's type is known at plan time but the extraction UDF infers ANY, so build the
+        // call with the type set explicitly. LambdaUtils' re-inference preserves it (a CAST would
+        // not survive the enumerable backend for complex types like arrays).
+        return context.rexBuilder.makeCall(
+            binding.pairType(),
+            PPLBuiltinOperators.FOREACH_PAIR_ITEM,
+            List.of(
+                pair,
+                context.rexBuilder.makeExactLiteral(BigDecimal.valueOf(binding.pairIndex()))));
+      case LITERAL:
+      default:
+        return context.rexBuilder.makeLiteral(binding.value());
+    }
+  }
+
   @Override
   public RexNode visitAlias(Alias node, CalcitePlanContext context) {
     RexNode expr = analyze(node.getDelegated(), context);