Grouping key field type can only be overwritten when the ExprCoreType are different (#4850)

Signed-off-by: Lantao Jin <ltjin@amazon.com>
(cherry picked from commit 1c27a6d)
Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

Author: github-actions[bot]

integ-test/src/yamlRestTest/resources/rest-api-spec/test/issues/4845.yml

@@ -0,0 +1,49 @@
+setup:
+  - do:
+      query.settings:
+        body:
+          transient:
+            plugins.calcite.enabled: true
+  - do:
+      indices.create:
+        index: test
+        body:
+          mappings:
+            properties:
+              "EventDate":
+                type: date
+                format: yyyy-MM-dd HH:mm:ss||strict_date_optional_time||epoch_millis
+  - do:
+      bulk:
+        index: test
+        refresh: true
+        body:
+          - '{"index": {"_id": "1"}}'
+          - '{"EventDate": "2013-07-15 10:47:34"}'
+
+---
+teardown:
+  - do:
+      query.settings:
+        body:
+          transient:
+            plugins.calcite.enabled : false
+
+---
+"handle custom format field with pushdown":
+  - skip:
+      features:
+        - headers
+        - allowed_warnings
+  - do:
+      allowed_warnings:
+        - 'Loading the fielddata on the _id field is deprecated and will be removed in future versions. If you require sorting or aggregating on this field you should also include the id in the body of your documents, and map this field as a keyword field that has [doc_values] enabled'
+      headers:
+        Content-Type: 'application/json'
+      ppl:
+        body:
+          query: source=test | stats count() by EventDate
+
+  - match: { total: 1 }
+  - match: {"schema": [{"name": "count()", "type": "bigint"},{"name": "EventDate", "type": "timestamp"}]}
+  - match: {"datarows": [[1, "2013-07-15 10:47:34"]]}

opensearch/src/main/java/org/opensearch/sql/opensearch/data/value/OpenSearchExprValueFactory.java

@@ -81,13 +81,20 @@ public class OpenSearchExprValueFactory {
   private final boolean fieldTypeTolerance;
 
   /**
-   * Extend existing mapping by new data without overwrite. Called from aggregation only {@see
-   * AggregationQueryBuilder#buildTypeMapping}.
+   * Extend existing mapping by new data. Overwrite only when the ExprCoreType of them are
+   * different. Called from aggregation only {@see AggregationQueryBuilder#buildTypeMapping}.
    *
    * @param typeMapping A data type mapping produced by aggregation.
    */
   public void extendTypeMapping(Map<String, OpenSearchDataType> typeMapping) {
-    this.typeMapping.putAll(typeMapping);
+    typeMapping.forEach(
+        (groupKey, extendedType) -> {
+          OpenSearchDataType existedType = this.typeMapping.get(groupKey);
+          if (existedType == null
+              || !existedType.getExprCoreType().equals(extendedType.getExprCoreType())) {
+            this.typeMapping.put(groupKey, extendedType);
+          }
+        });
   }
 
   @Getter @Setter private OpenSearchAggregationResponseParser parser;