[Feature] Support graphLookup with literal value as its start

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

@@ -45,8 +45,11 @@ public enum Direction {
   /** Target table for graph traversal lookup. */
   private final UnresolvedPlan fromTable;
 
-  /** Field in sourceTable to start with. */
-  private final Field startField;
+  /** Field in sourceTable to start with (piped mode). Null when using literal start values. */
+  private @Nullable final Field startField;
+
+  /** Literal start values for top-level graphlookup (mutually exclusive with startField). */
+  private @Nullable final List<Literal> startValues;
 
   /** Field in fromTable that represents the outgoing edge. */
   private final Field fromField;

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

@@ -2692,18 +2692,43 @@ public RelNode visitAddColTotals(AddColTotals node, CalcitePlanContext context)
 
   @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);
+
+    List<Object> startValuesForCalcite = null;
+    String startFieldName;
+    if (node.getStartValues() != null) {
+      // Literal start mode: create empty LogicalValues as dummy source (BiRel needs two inputs)
+      // And will ignore the previous pipe then.
+      RelDataType dummyType =
+          builder
+              .getTypeFactory()
+              .createStructType(
+                  List.of(builder.getTypeFactory().createSqlType(SqlTypeName.VARCHAR)),
+                  List.of("_dummy"));
+      builder.values(dummyType);
+      startFieldName = null;
+      startValuesForCalcite = new ArrayList<>();
+      for (var lit : node.getStartValues()) {
+        startValuesForCalcite.add(lit.getValue());
+      }
+    } else {
+      if (node.getChild().isEmpty()) {
+        throw new SemanticCheckException(
+            "Field reference start requires a piped source."
+                + " Use literal start values (e.g. start='value') for top-level graphLookup.");
+      }
+      // Piped mode: visit source child
+      visitChildren(node, context);
+      // TODO: Limit the number of source rows to 100 for now, make it configurable.
+      builder.limit(0, 100);
+      if (node.isBatchMode()) {
+        tryToRemoveMetaFields(context, true);
+      }
+      startFieldName = node.getStartField().getField().toString();
     }
     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();
@@ -2736,6 +2761,7 @@ public RelNode visitGraphLookup(GraphLookup node, CalcitePlanContext context) {
             sourceTable,
             lookupTable,
             startFieldName,
+            startValuesForCalcite,
             fromFieldName,
             toFieldName,
             outputFieldName,

core/src/main/java/org/opensearch/sql/calcite/plan/rel/GraphLookup.java

@@ -40,7 +40,8 @@
 public abstract class GraphLookup extends BiRel {
 
   // TODO: use RexInputRef instead of String for there fields
-  protected final String startField; // Field in source table (start entities)
+  @Nullable protected final String startField; // Field in source table (start entities)
+  @Nullable protected final List<Object> startValues; // Literal start values (top-level mode)
   protected final String fromField; // Field in lookup table (edge source)
   protected final String toField; // Field in lookup table (edge target)
   protected final String outputField; // Name of output array field
@@ -63,7 +64,8 @@ public abstract class GraphLookup extends BiRel {
    * @param traitSet Trait set
    * @param source Source table RelNode
    * @param lookup Lookup table RelNode
-   * @param startField Field name for start entities
+   * @param startField Field name for start entities (null in literal start mode)
+   * @param startValues Literal start values for top-level graphLookup (null in piped mode)
    * @param fromField Field name for outgoing edges
    * @param toField Field name for incoming edges
    * @param outputField Name of the output array field
@@ -81,7 +83,8 @@ protected GraphLookup(
       RelTraitSet traitSet,
       RelNode source,
       RelNode lookup,
-      String startField,
+      @Nullable String startField,
+      @Nullable List<Object> startValues,
       String fromField,
       String toField,
       String outputField,
@@ -94,6 +97,7 @@ protected GraphLookup(
       @Nullable RexNode filter) {
     super(cluster, traitSet, source, lookup);
     this.startField = startField;
+    this.startValues = startValues;
     this.fromField = fromField;
     this.toField = toField;
     this.outputField = outputField;
@@ -124,7 +128,19 @@ protected RelDataType deriveRowType() {
     if (outputRowType == null) {
       RelDataTypeFactory.Builder builder = getCluster().getTypeFactory().builder();
 
-      if (batchMode) {
+      if (startValues != null) {
+        // Literal start mode: Output = just [outputField: ARRAY<lookup_row>]
+        RelDataType lookupRowType = getLookup().getRowType();
+        if (this.depthField != null) {
+          final RelDataTypeFactory.Builder lookupBuilder = getCluster().getTypeFactory().builder();
+          lookupBuilder.addAll(lookupRowType.getFieldList());
+          RelDataType depthType = getCluster().getTypeFactory().createSqlType(SqlTypeName.INTEGER);
+          lookupBuilder.add(this.depthField, depthType);
+          lookupRowType = lookupBuilder.build();
+        }
+        RelDataType arrayType = getCluster().getTypeFactory().createArrayType(lookupRowType, -1);
+        builder.add(outputField, arrayType);
+      } else if (batchMode) {
         // Batch mode: Output = [Array<source>, Array<lookup>]
         // First field: aggregated source rows as array
         RelDataType sourceRowType = getSource().getRowType();
@@ -172,7 +188,7 @@ protected RelDataType deriveRowType() {
   @Override
   public double estimateRowCount(RelMetadataQuery mq) {
     // Batch mode aggregates all source rows into a single output row
-    return batchMode ? 1 : getSource().estimateRowCount(mq);
+    return (startValues != null || batchMode) ? 1 : getSource().estimateRowCount(mq);
   }
 
   @Override
@@ -184,6 +200,7 @@ public RelWriter explainTerms(RelWriter pw) {
         .item("depthField", depthField)
         .item("maxDepth", maxDepth)
         .item("bidirectional", bidirectional)
+        .itemIf("startValues", startValues, startValues != null)
         .itemIf("supportArray", supportArray, supportArray)
         .itemIf("batchMode", batchMode, batchMode)
         .itemIf("usePIT", usePIT, usePIT)

core/src/main/java/org/opensearch/sql/calcite/plan/rel/LogicalGraphLookup.java

@@ -21,31 +21,13 @@
 @Getter
 public class LogicalGraphLookup extends GraphLookup {
 
-  /**
-   * Creates a LogicalGraphLookup.
-   *
-   * @param cluster Cluster
-   * @param traitSet Trait set
-   * @param source Source table RelNode
-   * @param lookup Lookup table RelNode
-   * @param startField Field name for start entities
-   * @param fromField Field name for outgoing edges
-   * @param toField Field name for incoming edges
-   * @param outputField Name of the output array field
-   * @param depthField Name of the depth field
-   * @param maxDepth Maximum traversal depth (-1 for unlimited)
-   * @param bidirectional Whether to traverse edges in both directions
-   * @param supportArray Whether to support array-typed fields
-   * @param batchMode Whether to batch all source start values into a single unified BFS
-   * @param usePIT Whether to use PIT (Point In Time) search for complete results
-   * @param filter Optional filter condition for lookup table documents
-   */
   protected LogicalGraphLookup(
       RelOptCluster cluster,
       RelTraitSet traitSet,
       RelNode source,
       RelNode lookup,
-      String startField,
+      @Nullable String startField,
+      @Nullable List<Object> startValues,
       String fromField,
       String toField,
       String outputField,
@@ -62,6 +44,7 @@ protected LogicalGraphLookup(
         source,
         lookup,
         startField,
+        startValues,
         fromField,
         toField,
         outputField,
@@ -74,28 +57,11 @@ protected LogicalGraphLookup(
         filter);
   }
 
-  /**
-   * Creates a LogicalGraphLookup with Convention.NONE.
-   *
-   * @param source Source table RelNode
-   * @param lookup Lookup table RelNode
-   * @param startField Field name for start entities
-   * @param fromField Field name for outgoing edges
-   * @param toField Field name for incoming edges
-   * @param outputField Name of the output array field
-   * @param depthField Named of the output depth field
-   * @param maxDepth Maximum traversal depth (-1 for unlimited)
-   * @param bidirectional Whether to traverse edges in both directions
-   * @param supportArray Whether to support array-typed fields
-   * @param batchMode Whether to batch all source start values into a single unified BFS
-   * @param usePIT Whether to use PIT (Point In Time) search for complete results
-   * @param filter Optional filter condition for lookup table documents
-   * @return A new LogicalGraphLookup instance
-   */
   public static LogicalGraphLookup create(
       RelNode source,
       RelNode lookup,
-      String startField,
+      @Nullable String startField,
+      @Nullable List<Object> startValues,
       String fromField,
       String toField,
       String outputField,
@@ -114,6 +80,7 @@ public static LogicalGraphLookup create(
         source,
         lookup,
         startField,
+        startValues,
         fromField,
         toField,
         outputField,
@@ -134,6 +101,7 @@ public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
         inputs.get(0),
         inputs.get(1),
         startField,
+        startValues,
         fromField,
         toField,
         outputField,