Simplify CalciteExplainIT and add UT for AggregateAnalyzer

Signed-off-by: Kai Huang <ahkcs@amazon.com>

# Conflicts:
#	opensearch/src/test/java/org/opensearch/sql/opensearch/request/AggregateAnalyzerTest.java

Author: ahkcs

core/src/main/java/org/opensearch/sql/calcite/udf/udaf/FirstAggFunction.java

@@ -28,7 +28,7 @@ public FirstAccumulator add(FirstAccumulator acc, Object... values) {
     Object candidateValue = values[0];
     // Only accept the first non-null value encountered
     // Skip null values to find the first actual value
-    if (candidateValue != null && !acc.hasValue()) {
+    if (candidateValue != null) {
       acc.setValue(candidateValue);
     }
     return acc;
@@ -50,10 +50,6 @@ public synchronized void setValue(Object value) {
       }
     }
 
-    public boolean hasValue() {
-      return hasValue;
-    }
-
     @Override
     public Object value(Object... argList) {
       return first;

integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java

@@ -418,19 +418,6 @@ public void testExplainOnFirstLast() throws IOException {
                 TEST_INDEX_BANK)));
   }
 
-  // Only for Calcite
-  @Test
-  public void testExplainOnFirstLastDifferentFields() throws IOException {
-    String expected = loadExpectedPlan("explain_first_last_different_fields.json");
-    assertJsonEqualsIgnoreId(
-        expected,
-        explainQueryToString(
-            String.format(
-                "source=%s | stats first(account_number) as first_account,"
-                    + " last(balance) as last_balance, first(age) as first_age",
-                TEST_INDEX_BANK)));
-  }
-
   @Test
   public void testListAggregationExplain() throws IOException {
     String expected = loadExpectedPlan("explain_list_aggregation.json");

integ-test/src/test/resources/expectedOutput/calcite/explain_first_last_different_fields.json

@@ -42,6 +42,7 @@
 import org.junit.jupiter.api.Test;
 import org.opensearch.search.aggregations.AggregationBuilder;
 import org.opensearch.sql.data.type.ExprType;
+import org.opensearch.sql.expression.function.PPLBuiltinOperators;
 import org.opensearch.sql.opensearch.data.type.OpenSearchDataType;
 import org.opensearch.sql.opensearch.data.type.OpenSearchDataType.MappingType;
 import org.opensearch.sql.opensearch.request.AggregateAnalyzer.ExpressionNotAnalyzableException;
@@ -52,6 +53,7 @@
 import org.opensearch.sql.opensearch.response.agg.OpenSearchAggregationResponseParser;
 import org.opensearch.sql.opensearch.response.agg.SingleValueParser;
 import org.opensearch.sql.opensearch.response.agg.StatsParser;
+import org.opensearch.sql.opensearch.response.agg.TopHitsParser;
 
 class AggregateAnalyzerTest {
 
@@ -345,6 +347,137 @@ void analyze_groupBy_TextWithoutKeyword() {
     assertEquals("[field] must not be null", exception.getCause().getMessage());
   }
 
+  @Test
+  void analyze_firstAggregation() throws ExpressionNotAnalyzableException {
+    AggregateCall firstCall =
+        AggregateCall.create(
+            PPLBuiltinOperators.FIRST,
+            false,
+            false,
+            false,
+            ImmutableList.of(),
+            ImmutableList.of(0),
+            -1,
+            null,
+            RelCollations.EMPTY,
+            typeFactory.createSqlType(SqlTypeName.INTEGER),
+            "first_a");
+    List<String> outputFields = List.of("first_a");
+    Aggregate aggregate = createMockAggregate(List.of(firstCall), ImmutableBitSet.of());
+    Project project = null; // No project needed for simple aggregation
+    Pair<List<AggregationBuilder>, OpenSearchAggregationResponseParser> result =
+        AggregateAnalyzer.analyze(aggregate, project, rowType, fieldTypes, outputFields, null);
+
+    // Verify the aggregation builder JSON output
+    assertEquals(
+        "[{\"first_a\":{\"top_hits\":{\"from\":0,\"size\":1,\"version\":false,\"seq_no_primary_term\":false,\"explain\":false}}}]",
+        result.getLeft().toString());
+    assertInstanceOf(NoBucketAggregationParser.class, result.getRight());
+    MetricParserHelper metricsParser =
+        ((NoBucketAggregationParser) result.getRight()).getMetricsParser();
+    assertEquals(1, metricsParser.getMetricParserMap().size());
+    metricsParser
+        .getMetricParserMap()
+        .forEach(
+            (k, v) -> {
+              assertEquals("first_a", k);
+              assertInstanceOf(TopHitsParser.class, v);
+            });
+  }
+
+  @Test
+  void analyze_lastAggregation() throws ExpressionNotAnalyzableException {
+    AggregateCall lastCall =
+        AggregateCall.create(
+            PPLBuiltinOperators.LAST,
+            false,
+            false,
+            false,
+            ImmutableList.of(),
+            ImmutableList.of(1),
+            -1,
+            null,
+            RelCollations.EMPTY,
+            typeFactory.createSqlType(SqlTypeName.VARCHAR),
+            "last_b");
+    List<String> outputFields = List.of("last_b");
+    Aggregate aggregate = createMockAggregate(List.of(lastCall), ImmutableBitSet.of());
+    Project project = null; // No project needed for simple aggregation
+    Pair<List<AggregationBuilder>, OpenSearchAggregationResponseParser> result =
+        AggregateAnalyzer.analyze(aggregate, project, rowType, fieldTypes, outputFields, null);
+
+    // Verify the aggregation builder JSON output
+    assertEquals(
+        "[{\"last_b\":{\"top_hits\":{\"from\":0,\"size\":1,\"version\":false,\"seq_no_primary_term\":false,\"explain\":false,\"sort\":[{\"_doc\":{\"order\":\"desc\"}}]}}}]",
+        result.getLeft().toString());
+    assertInstanceOf(NoBucketAggregationParser.class, result.getRight());
+    MetricParserHelper metricsParser =
+        ((NoBucketAggregationParser) result.getRight()).getMetricsParser();
+    assertEquals(1, metricsParser.getMetricParserMap().size());
+    metricsParser
+        .getMetricParserMap()
+        .forEach(
+            (k, v) -> {
+              assertEquals("last_b", k);
+              assertInstanceOf(TopHitsParser.class, v);
+            });
+  }
+
+  @Test
+  void analyze_firstAndLastAggregations_withGroupBy() throws ExpressionNotAnalyzableException {
+    AggregateCall firstCall =
+        AggregateCall.create(
+            PPLBuiltinOperators.FIRST,
+            false,
+            false,
+            false,
+            ImmutableList.of(),
+            ImmutableList.of(0),
+            -1,
+            null,
+            RelCollations.EMPTY,
+            typeFactory.createSqlType(SqlTypeName.INTEGER),
+            "first_a");
+    AggregateCall lastCall =
+        AggregateCall.create(
+            PPLBuiltinOperators.LAST,
+            false,
+            false,
+            false,
+            ImmutableList.of(),
+            ImmutableList.of(0),
+            -1,
+            null,
+            RelCollations.EMPTY,
+            typeFactory.createSqlType(SqlTypeName.INTEGER),
+            "last_a");
+    List<String> outputFields = List.of("b", "first_a", "last_a");
+    Aggregate aggregate = createMockAggregate(List.of(firstCall, lastCall), ImmutableBitSet.of(1));
+    Project project =
+        createMockProject(List.of(0, 1)); // Need both fields for groupBy with aggregations
+    Pair<List<AggregationBuilder>, OpenSearchAggregationResponseParser> result =
+        AggregateAnalyzer.analyze(aggregate, project, rowType, fieldTypes, outputFields, null);
+
+    assertEquals(
+        "[{\"composite_buckets\":{\"composite\":{\"size\":1000,\"sources\":["
+            + "{\"b\":{\"terms\":{\"field\":\"b.keyword\",\"missing_bucket\":true,\"missing_order\":\"first\",\"order\":\"asc\"}}}]},"
+            + "\"aggregations\":{"
+            + "\"first_a\":{\"top_hits\":{\"from\":0,\"size\":1,\"version\":false,\"seq_no_primary_term\":false,\"explain\":false,\"_source\":{\"includes\":[\"a\"],\"excludes\":[]}}},"
+            + "\"last_a\":{\"top_hits\":{\"from\":0,\"size\":1,\"version\":false,\"seq_no_primary_term\":false,\"explain\":false,\"_source\":{\"includes\":[\"a\"],\"excludes\":[]},\"sort\":[{\"_doc\":{\"order\":\"desc\"}}]}}}}}]",
+        result.getLeft().toString());
+    assertInstanceOf(CompositeAggregationParser.class, result.getRight());
+    MetricParserHelper metricsParser =
+        ((CompositeAggregationParser) result.getRight()).getMetricsParser();
+    assertEquals(2, metricsParser.getMetricParserMap().size());
+    metricsParser
+        .getMetricParserMap()
+        .forEach(
+            (k, v) -> {
+              assertTrue(k.equals("first_a") || k.equals("last_a"));
+              assertInstanceOf(TopHitsParser.class, v);
+            });
+  }
+
   @Test
   void analyze_aggCall_simpleFilter() throws ExpressionNotAnalyzableException {
     buildAggregation("filter_cnt")