fix(lakehouse): prevent ClassCastException on mixed numeric types in comparisons

Normalize Number types to double before comparison in TopKMerger.compareValues()
and AggregationReducer min/max. Integer.compareTo(Long) throws ClassCastException
because compareTo expects the same type. DataFusion can return different numeric
widths across workers for the same column.

Author: vamsimanohar

sandbox/plugins/lakehouse-iceberg/src/main/java/org/opensearch/lakehouse/distributed/AggregationReducer.java

@@ -84,13 +84,12 @@ static Object sumColumn(List<WorkerQueryResponse> responses, int colIdx) {
      * @param colIdx    the column index to minimize
      * @return the minimum value, or null if all values are null
      */
-    @SuppressWarnings("unchecked")
     static Object minColumn(List<WorkerQueryResponse> responses, int colIdx) {
-        Comparable<Object> min = null;
+        Object min = null;
         for (WorkerQueryResponse r : responses) {
             if (hasValue(r, colIdx)) {
-                Comparable<Object> val = (Comparable<Object>) r.getColumnData()[colIdx][0];
-                if (min == null || val.compareTo((Object) min) < 0) {
+                Object val = r.getColumnData()[colIdx][0];
+                if (min == null || compareForAgg(val, min) < 0) {
                     min = val;
                 }
             }
@@ -106,20 +105,35 @@ static Object minColumn(List<WorkerQueryResponse> responses, int colIdx) {
      * @param colIdx    the column index to maximize
      * @return the maximum value, or null if all values are null
      */
-    @SuppressWarnings("unchecked")
     static Object maxColumn(List<WorkerQueryResponse> responses, int colIdx) {
-        Comparable<Object> max = null;
+        Object max = null;
         for (WorkerQueryResponse r : responses) {
             if (hasValue(r, colIdx)) {
-                Comparable<Object> val = (Comparable<Object>) r.getColumnData()[colIdx][0];
-                if (max == null || val.compareTo((Object) max) > 0) {
+                Object val = r.getColumnData()[colIdx][0];
+                if (max == null || compareForAgg(val, max) > 0) {
                     max = val;
                 }
             }
         }
         return max;
     }
 
+    /**
+     * Compares two values for aggregation, handling mixed numeric types safely.
+     * Normalizes Number types to double to avoid ClassCastException when comparing
+     * different widths (e.g., Integer vs Long from different workers).
+     */
+    @SuppressWarnings("unchecked")
+    private static int compareForAgg(Object v1, Object v2) {
+        if (v1 instanceof Number && v2 instanceof Number) {
+            return Double.compare(((Number) v1).doubleValue(), ((Number) v2).doubleValue());
+        }
+        if (v1 instanceof Comparable && v2 instanceof Comparable) {
+            return ((Comparable<Object>) v1).compareTo(v2);
+        }
+        return v1.toString().compareTo(v2.toString());
+    }
+
     /**
      * Finds the first non-null value in a column across all worker responses.
      */

sandbox/plugins/lakehouse-iceberg/src/main/java/org/opensearch/lakehouse/distributed/TopKMerger.java

@@ -99,13 +99,19 @@ static Comparator<Object[]> buildComparator(int[] sortColumns, boolean[] sortAsc
      * @param v2 second value
      * @return negative if v1 &lt; v2, positive if v1 &gt; v2, zero if equal
      */
-    @SuppressWarnings("unchecked")
     static int compareValues(Object v1, Object v2) {
         if (v1 == null && v2 == null) return 0;
         if (v1 == null) return 1;
         if (v2 == null) return -1;
+        // Normalize numeric types to avoid ClassCastException when comparing
+        // different widths (e.g., Integer vs Long from different workers)
+        if (v1 instanceof Number && v2 instanceof Number) {
+            return Double.compare(((Number) v1).doubleValue(), ((Number) v2).doubleValue());
+        }
         if (v1 instanceof Comparable && v2 instanceof Comparable) {
-            return ((Comparable<Object>) v1).compareTo(v2);
+            @SuppressWarnings("unchecked")
+            int cmp = ((Comparable<Object>) v1).compareTo(v2);
+            return cmp;
         }
         return v1.toString().compareTo(v2.toString());
     }

sandbox/plugins/lakehouse-iceberg/src/test/java/org/opensearch/lakehouse/distributed/AggregationReducerTests.java

@@ -220,6 +220,21 @@ public void testMaxColumnSkipsEmptyResponses() {
         assertEquals(42, result);
     }
 
+    public void testMinColumnMixedNumericTypes() {
+        // Integer vs Long — previously threw ClassCastException
+        WorkerQueryResponse r1 = makeResponse(new Object[][]{{Integer.valueOf(10)}}, 1);
+        WorkerQueryResponse r2 = makeResponse(new Object[][]{{Long.valueOf(5)}}, 1);
+        Object result = AggregationReducer.minColumn(List.of(r1, r2), 0);
+        assertEquals(Long.valueOf(5), result);
+    }
+
+    public void testMaxColumnMixedNumericTypes() {
+        WorkerQueryResponse r1 = makeResponse(new Object[][]{{Integer.valueOf(10)}}, 1);
+        WorkerQueryResponse r2 = makeResponse(new Object[][]{{Long.valueOf(5)}}, 1);
+        Object result = AggregationReducer.maxColumn(List.of(r1, r2), 0);
+        assertEquals(Integer.valueOf(10), result);
+    }
+
     // --- Helper ---
 
     private WorkerQueryResponse makeResponse(Object[][] columnData, int rowCount) {

sandbox/plugins/lakehouse-iceberg/src/test/java/org/opensearch/lakehouse/distributed/TopKMergerTests.java

@@ -149,6 +149,17 @@ public String toString() {
         assertTrue(TopKMerger.compareValues(a, b) < 0);
     }
 
+    public void testCompareValuesMixedNumericTypes() {
+        // Integer vs Long — previously threw ClassCastException
+        assertTrue(TopKMerger.compareValues(Integer.valueOf(5), Long.valueOf(10)) < 0);
+        assertTrue(TopKMerger.compareValues(Long.valueOf(10), Integer.valueOf(5)) > 0);
+        assertEquals(0, TopKMerger.compareValues(Integer.valueOf(5), Long.valueOf(5)));
+        // Integer vs Double
+        assertTrue(TopKMerger.compareValues(Integer.valueOf(3), Double.valueOf(3.5)) < 0);
+        // Float vs Long
+        assertTrue(TopKMerger.compareValues(Float.valueOf(2.5f), Long.valueOf(3)) < 0);
+    }
+
     // --- buildComparator tests ---
 
     public void testBuildComparatorAscending() {