Widen numeric mapping conflicts to one deterministic type

docs/user/ppl/limitations/limitations.md

@@ -3,6 +3,7 @@
 ## Inconsistent Field Types across indices  
 
 * If the same field has different types across indices (e.g., `field` is a `string` in one index and an `integer` in another), PPL selects a field type from one of the indices—this selection is non-deterministic. Fields with other types are ignored during query execution.  
+* One exception is a field mapped as `byte`, `short`, `integer`, `long`, `half_float`, `float`, `scaled_float` or `double` in every index. PPL merges it to one type, whatever the order of the indices. A mix of `byte`, `short`, `integer` and `long` merges to the widest of them. A mix of `half_float` and `float` merges to `float`. Any other mix merges to `double`, which may be a type that no index holds.  
 * For `object` fields, [PPL merges subfields from different indices to tolerate schema variations](https://github.com/opensearch-project/sql/issues/3625).  
 
 ## Unsupported OpenSearch Field Types  

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

@@ -0,0 +1,172 @@
+setup:
+  - do:
+      query.settings:
+        body:
+          transient:
+            plugins.calcite.enabled : true
+  - do:
+      indices.create:
+        index: integers_byte_5901
+        body:
+          mappings:
+            properties:
+              f:
+                type: byte
+  - do:
+      indices.create:
+        index: integers_short_5901
+        body:
+          mappings:
+            properties:
+              f:
+                type: short
+  - do:
+      indices.create:
+        index: integers_integer_5901
+        body:
+          mappings:
+            properties:
+              f:
+                type: integer
+  - do:
+      indices.create:
+        index: integers_long_5901
+        body:
+          mappings:
+            properties:
+              f:
+                type: long
+  - do:
+      indices.create:
+        index: long_double_long_5901
+        body:
+          mappings:
+            properties:
+              f:
+                type: long
+  - do:
+      indices.create:
+        index: long_double_double_5901
+        body:
+          mappings:
+            properties:
+              f:
+                type: double
+  - do:
+      indices.create:
+        index: long_half_long_5901
+        body:
+          mappings:
+            properties:
+              f:
+                type: long
+  - do:
+      indices.create:
+        index: long_half_half_float_5901
+        body:
+          mappings:
+            properties:
+              f:
+                type: half_float
+  - do:
+      bulk:
+        refresh: true
+        body:
+          - '{"index": {"_index": "integers_byte_5901", "_id": "1"}}'
+          - '{"f": 1}'
+          - '{"index": {"_index": "integers_short_5901", "_id": "1"}}'
+          - '{"f": 2}'
+          - '{"index": {"_index": "integers_integer_5901", "_id": "1"}}'
+          - '{"f": 3}'
+          - '{"index": {"_index": "integers_long_5901", "_id": "1"}}'
+          - '{"f": 5000000000}'
+          - '{"index": {"_index": "long_double_long_5901", "_id": "1"}}'
+          - '{"f": 5}'
+          - '{"index": {"_index": "long_double_double_5901", "_id": "1"}}'
+          - '{"f": 1.5}'
+          - '{"index": {"_index": "long_half_long_5901", "_id": "1"}}'
+          - '{"f": 16777217}'
+          - '{"index": {"_index": "long_half_half_float_5901", "_id": "1"}}'
+          - '{"f": 1.5}'
+
+---
+teardown:
+  - do:
+      query.settings:
+        body:
+          transient:
+            plugins.calcite.enabled : null
+  - do:
+      indices.delete:
+        index: integers_byte_5901
+        ignore_unavailable: true
+  - do:
+      indices.delete:
+        index: integers_short_5901
+        ignore_unavailable: true
+  - do:
+      indices.delete:
+        index: integers_integer_5901
+        ignore_unavailable: true
+  - do:
+      indices.delete:
+        index: integers_long_5901
+        ignore_unavailable: true
+  - do:
+      indices.delete:
+        index: long_double_long_5901
+        ignore_unavailable: true
+  - do:
+      indices.delete:
+        index: long_double_double_5901
+        ignore_unavailable: true
+  - do:
+      indices.delete:
+        index: long_half_long_5901
+        ignore_unavailable: true
+  - do:
+      indices.delete:
+        index: long_half_half_float_5901
+        ignore_unavailable: true
+
+---
+"Byte, short, integer and long across indices merge to bigint and keep a long beyond the integer range":
+  - skip:
+      features:
+        - headers
+  - do:
+      headers:
+        Content-Type: 'application/json'
+      ppl:
+        body:
+          query: 'source=integers_*_5901 | fields f | sort f'
+  - match: {"schema": [{"name": "f", "type": "bigint"}]}
+  - match: {"datarows": [[1], [2], [3], [5000000000]]}
+
+---
+"Long and double across indices merge to double and keep the fraction":
+  - skip:
+      features:
+        - headers
+  - do:
+      headers:
+        Content-Type: 'application/json'
+      ppl:
+        body:
+          query: 'source=long_double_*_5901 | fields f | sort f'
+  - match: {"schema": [{"name": "f", "type": "double"}]}
+  - match: {"datarows": [[1.5], [5.0]]}
+
+---
+"Long and half_float across indices merge to double and keep an integer above 2^24":
+  - skip:
+      features:
+        - headers
+  - do:
+      headers:
+        Content-Type: 'application/json'
+      ppl:
+        body:
+          query: 'source=long_half_*_5901 | fields f | sort f'
+  - match: {"schema": [{"name": "f", "type": "double"}]}
+  - match: {"datarows": [[1.5], [16777217.0]]}

opensearch/src/main/java/org/opensearch/sql/opensearch/util/MergeRules/MergeRuleHelper.java

@@ -12,7 +12,10 @@
 public class MergeRuleHelper {
   private static final List<MergeRule> RULES =
       List.of(
-          new DeepMergeRule(), new TextKeywordConflictRule(), new LatestRule() // must come last
+          new DeepMergeRule(),
+          new TextKeywordConflictRule(),
+          new NumericWideningRule(),
+          new LatestRule() // must come last
           );
 
   public static MergeRule selectRule(OpenSearchDataType source, OpenSearchDataType target) {

opensearch/src/main/java/org/opensearch/sql/opensearch/util/MergeRules/NumericWideningRule.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.opensearch.util.MergeRules;
+
+import java.util.List;
+import java.util.Map;
+import org.opensearch.sql.opensearch.data.type.OpenSearchDataType;
+import org.opensearch.sql.opensearch.data.type.OpenSearchDataType.MappingType;
+
+/**
+ * Merge rule for a field mapped as two different numeric types across indices. Two integer types
+ * merge to the wider one, half_float and float merge to float, and any other pair merges to double.
+ * When every index maps the field to a numeric type, the merged type does not depend on the order
+ * the indices merge in. Unlike WideningTypeRule in core, which picks float, an integer paired with
+ * half_float or float widens to double. Only a long beyond 2^53 next to a floating point type loses
+ * precision.
+ *
+ * <p>See GitHub issue #5901.
+ */
+public class NumericWideningRule implements MergeRule {
+
+  /** Integer types from narrowest to widest. */
+  private static final List<MappingType> INTEGER_TYPES_BY_WIDTH =
+      List.of(MappingType.Byte, MappingType.Short, MappingType.Integer, MappingType.Long);
+
+  /** Mapping types whose ExprCoreType is FLOAT. */
+  private static final List<MappingType> FLOAT_TYPES =
+      List.of(MappingType.HalfFloat, MappingType.Float);
+
+  /** Mapping types whose ExprCoreType is DOUBLE. */
+  private static final List<MappingType> DOUBLE_TYPES =
+      List.of(MappingType.ScaledFloat, MappingType.Double);
+
+  @Override
+  public boolean isMatch(OpenSearchDataType source, OpenSearchDataType target) {
+    if (source == null || target == null) {
+      return false;
+    }
+    MappingType sourceMapping = source.getMappingType();
+    MappingType targetMapping = target.getMappingType();
+    // A type built from an ExprCoreType, such as an alias, has no mapping type.
+    if (sourceMapping == null || targetMapping == null) {
+      return false;
+    }
+    return isNumeric(sourceMapping) && isNumeric(targetMapping) && sourceMapping != targetMapping;
+  }
+
+  @Override
+  public void mergeInto(
+      String key, OpenSearchDataType source, Map<String, OpenSearchDataType> target) {
+    MappingType widened = widenedType(source.getMappingType(), target.get(key).getMappingType());
+    target.put(key, OpenSearchDataType.of(widened));
+  }
+
+  private static MappingType widenedType(MappingType a, MappingType b) {
+    if (INTEGER_TYPES_BY_WIDTH.contains(a) && INTEGER_TYPES_BY_WIDTH.contains(b)) {
+      return INTEGER_TYPES_BY_WIDTH.indexOf(a) > INTEGER_TYPES_BY_WIDTH.indexOf(b) ? a : b;
+    }
+    if (FLOAT_TYPES.contains(a) && FLOAT_TYPES.contains(b)) {
+      return MappingType.Float;
+    }
+    // Every other pair merges to double. A 32-bit float holds integers exactly only up to 2^24,
+    // so an integer next to a half_float or float lands here too.
+    return MappingType.Double;
+  }
+
+  private static boolean isNumeric(MappingType mappingType) {
+    return INTEGER_TYPES_BY_WIDTH.contains(mappingType)
+        || FLOAT_TYPES.contains(mappingType)
+        || DOUBLE_TYPES.contains(mappingType);
+  }
+}