Merge remote-tracking branch 'origin/main' into issues/399

Signed-off-by: Yuanchun Shen <[email protected]>

Author: yuancu

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

@@ -123,10 +123,14 @@ private Timechart timechart(UnresolvedExpression newAggregateFunction) {
     return this.toBuilder().aggregateFunction(newAggregateFunction).build();
   }
 
-  /** TODO: extend to support additional per_* functions */
   @RequiredArgsConstructor
   static class PerFunction {
-    private static final Map<String, Integer> UNIT_SECONDS = Map.of("per_second", 1);
+    private static final Map<String, Integer> UNIT_SECONDS =
+        Map.of(
+            "per_second", 1,
+            "per_minute", 60,
+            "per_hour", 3600,
+            "per_day", 86400);
     private final String aggName;
     private final UnresolvedExpression aggArg;
     private final int seconds;

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

@@ -284,10 +284,13 @@ public RelNode visitRex(Rex node, CalcitePlanContext context) {
           "Rex pattern must contain at least one named capture group");
     }
 
+    // TODO: Once JDK 20+ is supported, consider using Pattern.namedGroups() API for more efficient
+    // named group handling instead of manual parsing in RegexCommonUtils
+
     List<RexNode> newFields = new ArrayList<>();
     List<String> newFieldNames = new ArrayList<>();
 
-    for (int i = 0; i < namedGroups.size(); i++) {
+    for (String groupName : namedGroups) {
       RexNode extractCall;
       if (node.getMaxMatch().isPresent() && node.getMaxMatch().get() > 1) {
         extractCall =
@@ -296,7 +299,7 @@ public RelNode visitRex(Rex node, CalcitePlanContext context) {
                 BuiltinFunctionName.REX_EXTRACT_MULTI,
                 fieldRex,
                 context.rexBuilder.makeLiteral(patternStr),
-                context.relBuilder.literal(i + 1),
+                context.rexBuilder.makeLiteral(groupName),
                 context.relBuilder.literal(node.getMaxMatch().get()));
       } else {
         extractCall =
@@ -305,10 +308,10 @@ public RelNode visitRex(Rex node, CalcitePlanContext context) {
                 BuiltinFunctionName.REX_EXTRACT,
                 fieldRex,
                 context.rexBuilder.makeLiteral(patternStr),
-                context.relBuilder.literal(i + 1));
+                context.rexBuilder.makeLiteral(groupName));
       }
       newFields.add(extractCall);
-      newFieldNames.add(namedGroups.get(i));
+      newFieldNames.add(groupName);
     }
 
     if (node.getOffsetField().isPresent()) {
@@ -852,7 +855,7 @@ private void projectPlusOverriding(
     List<String> originalFieldNames = context.relBuilder.peek().getRowType().getFieldNames();
     List<RexNode> toOverrideList =
         originalFieldNames.stream()
-            .filter(newNames::contains)
+            .filter(originalName -> shouldOverrideField(originalName, newNames))
             .map(a -> (RexNode) context.relBuilder.field(a))
             .toList();
     // 1. add the new fields, For example "age0, country0"
@@ -872,6 +875,17 @@ private void projectPlusOverriding(
     context.relBuilder.rename(expectedRenameFields);
   }
 
+  private boolean shouldOverrideField(String originalName, List<String> newNames) {
+    return newNames.stream()
+        .anyMatch(
+            newName ->
+                // Match exact field names (e.g., "age" == "age") for flat fields
+                newName.equals(originalName)
+                    // OR match nested paths (e.g., "resource.attributes..." starts with
+                    // "resource.")
+                    || newName.startsWith(originalName + "."));
+  }
+
   private List<List<RexInputRef>> extractInputRefList(List<RelBuilder.AggCall> aggCalls) {
     return aggCalls.stream()
         .map(RelBuilder.AggCall::over)
@@ -1803,18 +1817,16 @@ public RelNode visitMultisearch(Multisearch node, CalcitePlanContext context) {
   }
 
   /**
-   * Finds the timestamp field for multisearch ordering.
+   * Finds the @timestamp field for multisearch ordering. Only @timestamp field is used for
+   * timestamp interleaving. Other timestamp-like fields are ignored.
    *
-   * @param rowType The row type to search for timestamp fields
-   * @return The name of the timestamp field, or null if not found
+   * @param rowType The row type to search for @timestamp field
+   * @return "@timestamp" if the field exists, or null if not found
    */
   private String findTimestampField(RelDataType rowType) {
-    String[] candidates = {"@timestamp", "_time", "timestamp", "time"};
-    for (String fieldName : candidates) {
-      RelDataTypeField field = rowType.getField(fieldName, false, false);
-      if (field != null) {
-        return fieldName;
-      }
+    RelDataTypeField field = rowType.getField("@timestamp", false, false);
+    if (field != null) {
+      return "@timestamp";
     }
     return null;
   }

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

@@ -7,29 +7,27 @@
 
 import java.util.ArrayList;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import java.util.stream.Collectors;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeField;
 import org.apache.calcite.rex.RexNode;
-import org.apache.calcite.sql.validate.SqlValidatorUtil;
 
 /**
- * Utility class for unifying schemas across multiple RelNodes with type conflict resolution. Uses
- * the same strategy as append command - renames conflicting fields to avoid type conflicts.
+ * Utility class for unifying schemas across multiple RelNodes. Throws an exception when type
+ * conflicts are detected.
  */
 public class SchemaUnifier {
 
   /**
-   * Builds a unified schema for multiple nodes with type conflict resolution.
+   * Builds a unified schema for multiple nodes. Throws an exception if type conflicts are detected.
    *
    * @param nodes List of RelNodes to unify schemas for
    * @param context Calcite plan context
    * @return List of projected RelNodes with unified schema
+   * @throws IllegalArgumentException if type conflicts are detected
    */
   public static List<RelNode> buildUnifiedSchemaWithConflictResolution(
       List<RelNode> nodes, CalcitePlanContext context) {
@@ -41,7 +39,7 @@ public static List<RelNode> buildUnifiedSchemaWithConflictResolution(
       return nodes;
     }
 
-    // Step 1: Build the unified schema by processing all nodes
+    // Step 1: Build the unified schema by processing all nodes (throws on conflict)
     List<SchemaField> unifiedSchema = buildUnifiedSchema(nodes);
 
     // Step 2: Create projections for each node to align with unified schema
@@ -55,47 +53,37 @@ public static List<RelNode> buildUnifiedSchemaWithConflictResolution(
       projectedNodes.add(projectedNode);
     }
 
-    // Step 3: Unify names to handle type conflicts (this creates age0, age1, etc.)
-    List<String> uniqueNames =
-        SqlValidatorUtil.uniquify(fieldNames, SqlValidatorUtil.EXPR_SUGGESTER, true);
-
-    // Step 4: Re-project with unique names if needed
-    if (!uniqueNames.equals(fieldNames)) {
-      List<RelNode> renamedNodes = new ArrayList<>();
-      for (RelNode node : projectedNodes) {
-        RelNode renamedNode =
-            context.relBuilder.push(node).project(context.relBuilder.fields(), uniqueNames).build();
-        renamedNodes.add(renamedNode);
-      }
-      return renamedNodes;
-    }
-
     return projectedNodes;
   }
 
   /**
-   * Builds a unified schema by merging fields from all nodes. Fields with the same name but
-   * different types are added as separate entries (which will be renamed during uniquification).
+   * Builds a unified schema by merging fields from all nodes. Throws an exception if fields with
+   * the same name have different types.
    *
    * @param nodes List of RelNodes to merge schemas from
-   * @return List of SchemaField representing the unified schema (may contain duplicate names)
+   * @return List of SchemaField representing the unified schema
+   * @throws IllegalArgumentException if type conflicts are detected
    */
   private static List<SchemaField> buildUnifiedSchema(List<RelNode> nodes) {
     List<SchemaField> schema = new ArrayList<>();
-    Map<String, Set<RelDataType>> seenFields = new HashMap<>();
+    Map<String, RelDataType> seenFields = new HashMap<>();
 
     for (RelNode node : nodes) {
       for (RelDataTypeField field : node.getRowType().getFieldList()) {
         String fieldName = field.getName();
         RelDataType fieldType = field.getType();
 
-        // Track which (name, type) combinations we've seen
-        Set<RelDataType> typesForName = seenFields.computeIfAbsent(fieldName, k -> new HashSet<>());
-
-        if (!typesForName.contains(fieldType)) {
-          // New field or same name with different type - add to schema
+        RelDataType existingType = seenFields.get(fieldName);
+        if (existingType == null) {
+          // New field - add to schema
           schema.add(new SchemaField(fieldName, fieldType));
-          typesForName.add(fieldType);
+          seenFields.put(fieldName, fieldType);
+        } else if (!areTypesCompatible(existingType, fieldType)) {
+          // Same field name but different type - throw exception
+          throw new IllegalArgumentException(
+              String.format(
+                  "Unable to process column '%s' due to incompatible types: '%s' and '%s'",
+                  fieldName, existingType.getSqlTypeName(), fieldType.getSqlTypeName()));
         }
         // If we've seen this exact (name, type) combination, skip it
       }
@@ -104,6 +92,10 @@ private static List<SchemaField> buildUnifiedSchema(List<RelNode> nodes) {
     return schema;
   }
 
+  private static boolean areTypesCompatible(RelDataType type1, RelDataType type2) {
+    return type1.getSqlTypeName() != null && type1.getSqlTypeName().equals(type2.getSqlTypeName());
+  }
+
   /**
    * Builds a projection for a node to align with the unified schema. For each field in the unified
    * schema: - If the node has a matching field with the same type, use it - Otherwise, project NULL
@@ -125,8 +117,8 @@ private static List<RexNode> buildProjectionForNode(
       RelDataType expectedType = schemaField.getType();
       RelDataTypeField nodeField = nodeFieldMap.get(fieldName);
 
-      if (nodeField != null && nodeField.getType().equals(expectedType)) {
-        // Field exists with matching type - use it
+      if (nodeField != null && areTypesCompatible(nodeField.getType(), expectedType)) {
+        // Field exists with compatible type - use it
         projection.add(context.rexBuilder.makeInputRef(node, nodeField.getIndex()));
       } else {
         // Field missing or type mismatch - project NULL

core/src/main/java/org/opensearch/sql/calcite/utils/PlanUtils.java

@@ -14,17 +14,22 @@
 import com.google.common.collect.ImmutableList;
 import java.lang.reflect.Method;
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Objects;
+import java.util.Set;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 import org.apache.calcite.plan.RelOptTable;
 import org.apache.calcite.rel.RelHomogeneousShuttle;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.RelShuttle;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.core.Sort;
 import org.apache.calcite.rel.core.TableScan;
 import org.apache.calcite.rel.logical.LogicalProject;
+import org.apache.calcite.rel.logical.LogicalSort;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rex.RexCall;
 import org.apache.calcite.rex.RexCorrelVariable;
@@ -38,6 +43,7 @@
 import org.apache.calcite.sql.fun.SqlStdOperatorTable;
 import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.util.Pair;
 import org.apache.calcite.util.Util;
 import org.opensearch.sql.ast.AbstractNodeVisitor;
 import org.opensearch.sql.ast.Node;
@@ -474,13 +480,51 @@ public Void visitInputRef(RexInputRef inputRef) {
     return selectedColumns;
   }
 
+  // `RelDecorrelator` may generate a Project with duplicated fields, e.g. Project($0,$0).
+  // There will be problem if pushing down the pattern like `Aggregate(AGG($0),{1})-Project($0,$0)`,
+  // as it will lead to field-name conflict.
+  // We should wait and rely on `AggregateProjectMergeRule` to mitigate it by having this constraint
+  // Nevertheless, that rule cannot handle all cases if there is RexCall in the Project,
+  // e.g. Project($0, $0, +($0,1)). We cannot push down the Aggregate for this corner case.
+  // TODO: Simplify the Project where there is RexCall by adding a new rule.
+  static boolean distinctProjectList(LogicalProject project) {
+    // Change to Set<Pair<RexNode, String>> to resolve
+    // https://github.com/opensearch-project/sql/issues/4347
+    Set<Pair<RexNode, String>> rexSet = new HashSet<>();
+    return project.getNamedProjects().stream().allMatch(rexSet::add);
+  }
+
+  static boolean containsRexOver(LogicalProject project) {
+    return project.getProjects().stream().anyMatch(RexOver::containsOver);
+  }
+
+  /**
+   * The LogicalSort is a LIMIT that should be pushed down when its fetch field is not null and its
+   * collation is empty. For example: <code>sort name | head 5</code> should not be pushed down
+   * because it has a field collation.
+   *
+   * @param sort The LogicalSort to check.
+   * @return True if the LogicalSort is a LIMIT, false otherwise.
+   */
+  static boolean isLogicalSortLimit(LogicalSort sort) {
+    return sort.fetch != null;
+  }
+
+  static boolean projectContainsExpr(Project project) {
+    return project.getProjects().stream().anyMatch(p -> p instanceof RexCall);
+  }
+
+  static boolean sortByFieldsOnly(Sort sort) {
+    return !sort.getCollation().getFieldCollations().isEmpty() && sort.fetch == null;
+  }
+
   /**
    * Get a string representation of the argument types expressed in ExprType for error messages.
    *
    * @param argTypes the list of argument types as {@link RelDataType}
    * @return a string in the format [type1,type2,...] representing the argument types
    */
-  public static String getActualSignature(List<RelDataType> argTypes) {
+  static String getActualSignature(List<RelDataType> argTypes) {
     return "["
         + argTypes.stream()
             .map(OpenSearchTypeFactory::convertRelDataTypeToExprType)

core/src/main/java/org/opensearch/sql/data/utils/ComparableLinkedHashMap.java

@@ -42,15 +42,29 @@ private int compareRecursive(
     if (!thisHasNext) return -1;
     if (!otherHasNext) return 1;
 
-    V thisValue = thisIterator.next().getValue();
-    V otherValue = otherIterator.next().getValue();
-    int comparison = compareValues(thisValue, otherValue);
+    Map.Entry<K, V> thisEntry = thisIterator.next();
+    Map.Entry<K, V> otherEntry = otherIterator.next();
+    K thisKey = thisEntry.getKey();
+    K otherKey = otherEntry.getKey();
+    V thisValue = thisEntry.getValue();
+    V otherValue = otherEntry.getValue();
+    int comparison = compareKV(thisKey, otherKey, thisValue, otherValue);
     if (comparison != 0) return comparison;
     return compareRecursive(thisIterator, otherIterator);
   }
 
   @SuppressWarnings("unchecked")
-  private int compareValues(V value1, V value2) {
+  private int compareKV(K key1, K key2, V value1, V value2) {
+    int keyCompare;
+    if (key1 instanceof Comparable) {
+      keyCompare = ((Comparable<K>) key1).compareTo(key2);
+    } else {
+      keyCompare = key1.toString().compareTo(key2.toString());
+    }
+    if (keyCompare != 0) {
+      return keyCompare;
+    }
+
     if (value1 == null && value2 == null) return 0;
     if (value1 == null) return -1;
     if (value2 == null) return 1;