[Feature] Support bi-directional graph traversal command graphlookup

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

@@ -73,6 +73,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.GraphLookup;
 import org.opensearch.sql.ast.tree.Head;
 import org.opensearch.sql.ast.tree.Join;
 import org.opensearch.sql.ast.tree.Kmeans;
@@ -546,6 +547,11 @@ public LogicalPlan visitMvCombine(MvCombine node, AnalysisContext context) {
     throw getOnlyForCalciteException("mvcombine");
   }
 
+  @Override
+  public LogicalPlan visitGraphLookup(GraphLookup node, AnalysisContext context) {
+    throw getOnlyForCalciteException("graphlookup");
+  }
+
   /** 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

@@ -61,6 +61,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.GraphLookup;
 import org.opensearch.sql.ast.tree.Head;
 import org.opensearch.sql.ast.tree.Join;
 import org.opensearch.sql.ast.tree.Kmeans;
@@ -475,4 +476,8 @@ public T visitAddColTotals(AddColTotals node, C context) {
   public T visitMvCombine(MvCombine node, C context) {
     return visitChildren(node, context);
   }
+
+  public T visitGraphLookup(GraphLookup node, C context) {
+    return visitChildren(node, context);
+  }
 }

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

@@ -0,0 +1,104 @@
+/*
+ * 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 javax.annotation.Nullable;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.Setter;
+import lombok.ToString;
+import org.opensearch.sql.ast.AbstractNodeVisitor;
+import org.opensearch.sql.ast.expression.Field;
+import org.opensearch.sql.ast.expression.Literal;
+import org.opensearch.sql.ast.expression.UnresolvedExpression;
+
+/**
+ * AST node for graphLookup command. Performs BFS graph traversal on a lookup table.
+ *
+ * <p>Example: source=employees | graphLookup employees fromField=manager toField=name maxDepth=3
+ * depthField=level direction=uni as hierarchy
+ */
+@Getter
+@Setter
+@ToString
+@EqualsAndHashCode(callSuper = false)
+@RequiredArgsConstructor
+@AllArgsConstructor
+@Builder(toBuilder = true)
+public class GraphLookup extends UnresolvedPlan {
+  /** Direction mode for graph traversal. */
+  public enum Direction {
+    /** Unidirectional - traverse edges in one direction only. */
+    UNI,
+    /** Bidirectional - traverse edges in both directions. */
+    BI
+  }
+
+  /** Target table for graph traversal lookup. */
+  private final UnresolvedPlan fromTable;
+
+  /** Field in sourceTable to start with. */
+  private final Field startField;
+
+  /** Field in fromTable that represents the outgoing edge. */
+  private final Field fromField;
+
+  /** Field in input/fromTable to match against for traversal. */
+  private final Field toField;
+
+  /** Output field name for collected traversal results. */
+  private final Field as;
+
+  /** Maximum traversal depth. Zero means no limit. */
+  private final Literal maxDepth;
+
+  /** Optional field name to include recursion depth in output. */
+  private @Nullable final Field depthField;
+
+  /** Direction mode: UNI (default) or BIO for bidirectional. */
+  private final Direction direction;
+
+  /** Whether to support array-typed fields without early filter pushdown. */
+  private final boolean supportArray;
+
+  /** Whether to batch all source start values into a single unified BFS traversal. */
+  private final boolean batchMode;
+
+  /** Whether to use PIT (Point In Time) search for the lookup table to get complete results. */
+  private final boolean usePIT;
+
+  /**
+   * Optional filter condition to restrict which lookup table documents participate in traversal.
+   */
+  private @Nullable final UnresolvedExpression filter;
+
+  private UnresolvedPlan child;
+
+  public String getDepthFieldName() {
+    return depthField == null ? null : depthField.getField().toString();
+  }
+
+  @Override
+  public UnresolvedPlan 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> visitor, C context) {
+    return visitor.visitGraphLookup(this, context);
+  }
+}

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

@@ -118,6 +118,8 @@
 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.GraphLookup;
+import org.opensearch.sql.ast.tree.GraphLookup.Direction;
 import org.opensearch.sql.ast.tree.Head;
 import org.opensearch.sql.ast.tree.Join;
 import org.opensearch.sql.ast.tree.Kmeans;
@@ -151,6 +153,7 @@
 import org.opensearch.sql.ast.tree.Window;
 import org.opensearch.sql.calcite.plan.AliasFieldsWrappable;
 import org.opensearch.sql.calcite.plan.OpenSearchConstants;
+import org.opensearch.sql.calcite.plan.rel.LogicalGraphLookup;
 import org.opensearch.sql.calcite.plan.rel.LogicalSystemLimit;
 import org.opensearch.sql.calcite.plan.rel.LogicalSystemLimit.SystemLimitType;
 import org.opensearch.sql.calcite.utils.BinUtils;
@@ -2524,6 +2527,67 @@ public RelNode visitAddColTotals(AddColTotals node, CalcitePlanContext context)
         context, fieldsToAggregate, false, true, null, labelField, label);
   }
 
+  @Override
+  public RelNode visitGraphLookup(GraphLookup node, CalcitePlanContext context) {
+    // 1. Visit source (child) table
+    visitChildren(node, context);
+    RelBuilder builder = context.relBuilder;
+    // TODO: Limit the number of source rows to 100 for now, make it configurable.
+    builder.limit(0, 100);
+    if (node.isBatchMode()) {
+      tryToRemoveMetaFields(context, true);
+    }
+    RelNode sourceTable = builder.build();
+
+    // 2. Extract parameters
+    String startFieldName = node.getStartField().getField().toString();
+    String fromFieldName = node.getFromField().getField().toString();
+    String toFieldName = node.getToField().getField().toString();
+    String outputFieldName = node.getAs().getField().toString();
+    String depthFieldName = node.getDepthFieldName();
+    boolean bidirectional = node.getDirection() == Direction.BI;
+
+    RexLiteral maxDepthNode = (RexLiteral) rexVisitor.analyze(node.getMaxDepth(), context);
+    Integer maxDepthValue = maxDepthNode.getValueAs(Integer.class);
+    maxDepthValue = maxDepthValue == null ? 0 : maxDepthValue;
+    boolean supportArray = node.isSupportArray();
+    boolean batchMode = node.isBatchMode();
+    boolean usePIT = node.isUsePIT();
+
+    // 3. Visit and materialize lookup table
+    analyze(node.getFromTable(), context);
+    tryToRemoveMetaFields(context, true);
+
+    // 4. Convert filter expression to RexNode against lookup table schema
+    RexNode filterRex = null;
+    if (node.getFilter() != null) {
+      filterRex = rexVisitor.analyze(node.getFilter(), context);
+    }
+
+    RelNode lookupTable = builder.build();
+
+    // 5. Create LogicalGraphLookup RelNode
+    // The conversion rule will extract the OpenSearchIndex from the lookup table
+    RelNode graphLookup =
+        LogicalGraphLookup.create(
+            sourceTable,
+            lookupTable,
+            startFieldName,
+            fromFieldName,
+            toFieldName,
+            outputFieldName,
+            depthFieldName,
+            maxDepthValue,
+            bidirectional,
+            supportArray,
+            batchMode,
+            usePIT,
+            filterRex);
+
+    builder.push(graphLookup);
+    return builder.peek();
+  }
+
   /**
    * Cast integer sum to long, real/float to double to avoid ClassCastException
    *