Moving Numeric Comparators to new framework. (#13192)

- Simplified the `GenericComparator` so a lot less code for specific type versions (building on the generic `StorageType` work).
- Fix bugs with equals and not-equals when the right-hand value and column is not of same type (both Generic and Bool).
- Extracted a `BinaryOperationNumeric` base class from `BinaryCoalescingOperationNumeric`.
*A fair amount of boilerplate is still needed for the comparators.*
- Added `NumericComparators` covering doing remaining typed comparisons.
- Simplified calling comparators from the `In_Memory_Implementation`.

ToDos:
- Review test cases for mixed matched types in comparators (did not fail any tests before).

Author: jdunkerley

distribution/lib/Standard/Table/0.0.0-dev/src/Internal/In_Memory_Column_Implementation.enso

@@ -82,8 +82,7 @@ type In_Memory_Column_Implementation
             if (a.is_a Float) || (b.is_a Float) then
                 problem_builder.reportFloatingPointEquality -1
             a == b
-        _call_comparator this_column other new_name (Comparators.eq (this_column:In_Memory_Column).java_column) <|
-            run_vectorized_binary_op_with_fallback_problem_handling this_column Java_Storage.Maps.EQ other fallback_fn=fallback new_name=new_name expected_result_type=Value_Type.Boolean skip_nulls=True
+        _call_comparator this_column other new_name (Comparators.eq (this_column:In_Memory_Column).java_column (_java_other other)) fallback
 
     equals_ignore_case this_column (other : Column | Any) locale:Locale=Locale.default -> Column =
         ## TODO currently this always runs the fallback which is slow due to the
@@ -100,28 +99,39 @@ type In_Memory_Column_Implementation
 
     not_equals this_column (other : Column | Any) -> Column =
         new_name = naming_helper.binary_operation_name "!=" this_column other
-        _call_comparator this_column other new_name (Comparators.notEq (this_column:In_Memory_Column).java_column) <|
-            (this_column == other).not . rename new_name
+        fallback problem_builder a b =
+            if (a.is_a Float) || (b.is_a Float) then
+                problem_builder.reportFloatingPointEquality -1
+            a != b
+        _call_comparator this_column other new_name (Comparators.notEq (this_column:In_Memory_Column).java_column (_java_other other)) fallback
 
     greater_than_eq this_column (other : Column | Any) -> Column = Value_Type.expect_comparable this_column other <| Incomparable_Values.handle_errors <|
         new_name = naming_helper.binary_operation_name ">=" this_column other
-        _call_comparator this_column other new_name (Comparators.greaterThanEq (this_column:In_Memory_Column).java_column) <|
-            run_vectorized_binary_op this_column Java_Storage.Maps.GTE new_name=new_name fallback_fn=(>=) other expected_result_type=Value_Type.Boolean
+        fallback problem_builder a b =
+            _ = problem_builder
+            a >= b
+        _call_comparator this_column other new_name (Comparators.greaterThanEq (this_column:In_Memory_Column).java_column (_java_other other)) fallback
 
     less_than_eq this_column (other : Column | Any) -> Column = Value_Type.expect_comparable this_column other <| Incomparable_Values.handle_errors <|
         new_name = naming_helper.binary_operation_name "<=" this_column other
-        _call_comparator this_column other new_name (Comparators.lessThanEq (this_column:In_Memory_Column).java_column) <|
-            run_vectorized_binary_op this_column Java_Storage.Maps.LTE new_name=new_name fallback_fn=(<=) other expected_result_type=Value_Type.Boolean
+        fallback problem_builder a b =
+            _ = problem_builder
+            a <= b
+        _call_comparator this_column other new_name (Comparators.lessThanEq (this_column:In_Memory_Column).java_column (_java_other other)) fallback
 
     greater_than this_column (other : Column | Any) -> Column = Value_Type.expect_comparable this_column other <| Incomparable_Values.handle_errors <|
         new_name = naming_helper.binary_operation_name ">" this_column other
-        _call_comparator this_column other new_name (Comparators.greaterThan (this_column:In_Memory_Column).java_column) <|
-            run_vectorized_binary_op this_column Java_Storage.Maps.GT new_name=new_name fallback_fn=(>) other expected_result_type=Value_Type.Boolean
+        fallback problem_builder a b =
+            _ = problem_builder
+            a > b
+        _call_comparator this_column other new_name (Comparators.greaterThan (this_column:In_Memory_Column).java_column (_java_other other)) fallback
 
     less_than this_column (other : Column | Any) -> Column = Value_Type.expect_comparable this_column other <| Incomparable_Values.handle_errors <|
         new_name = naming_helper.binary_operation_name "<" this_column other
-        _call_comparator this_column other new_name (Comparators.lessThan (this_column:In_Memory_Column).java_column) <|
-            run_vectorized_binary_op this_column Java_Storage.Maps.LT new_name=new_name fallback_fn=(<) other expected_result_type=Value_Type.Boolean
+        fallback problem_builder a b =
+            _ = problem_builder
+            a < b
+        _call_comparator this_column other new_name (Comparators.lessThan (this_column:In_Memory_Column).java_column (_java_other other)) fallback
 
     between this_column (lower : Column | Any) (upper : Column | Any) -> Column = Value_Type.expect_comparable this_column lower <| Value_Type.expect_comparable this_column upper <|
         new_name = naming_helper.concat <|
@@ -947,15 +957,21 @@ apply_unary_map column:Column new_name:Text function expected_result_type:Value_
         map_column = UnaryOperation.mapFunction (column:In_Memory_Column).java_column function nothing_unchanged storage_type new_name java_problem_aggregator
         In_Memory_Column.from_java_column map_column
 
-## PRIVATE
-   Temporary function to determine if to use the Comparators statics or embedded operations.
-private _call_comparator left:In_Memory_Column other new_name ~comparator ~enso_action =
-    if (Comparators.isSupported left.java_column) == False then enso_action else
-        _call_binary_operator left other new_name comparator
+private _java_other other =
+    if other.is_a Column then (other:In_Memory_Column).java_column else
+        enso_to_java other
+
+private _call_comparator left other new_name ~comparator fallback =
+    case (Comparators.isSupported (left:In_Memory_Column).java_column) of
+        True -> _call_binary_operator left other new_name comparator
+        False ->
+            Java_Problems.with_map_operation_problem_aggregator new_name Problem_Behavior.Report_Warning java_problem_aggregator->
+                fallback_fn = fallback java_problem_aggregator
+                run_binary_op left fallback_fn other new_name skip_nulls=True expected_result_type=Value_Type.Boolean
 
 private _call_binary_operator left:In_Memory_Column other new_name ~operation =
     Java_Problems.with_map_operation_problem_aggregator new_name Problem_Behavior.Report_Warning java_problem_aggregator->
-        java_column = operation.apply left.java_column (if other.is_a Column then (other:In_Memory_Column).java_column else other) new_name java_problem_aggregator
+        java_column = operation.apply left.java_column (_java_other other) new_name java_problem_aggregator
         In_Memory_Column.from_java_column java_column
 
 ## PRIVATE

std-bits/table/src/main/java/org/enso/table/data/column/operation/BinaryCoalescingOperationNumeric.java

@@ -6,16 +6,11 @@
 import org.enso.table.data.column.storage.ColumnDoubleStorage;
 import org.enso.table.data.column.storage.ColumnLongStorage;
 import org.enso.table.data.column.storage.ColumnStorage;
-import org.enso.table.data.column.storage.ColumnStorageFacade;
-import org.enso.table.data.column.storage.PreciseTypeOptions;
-import org.enso.table.data.column.storage.numeric.DoubleStorageFacade;
 import org.enso.table.data.column.storage.type.BigDecimalType;
 import org.enso.table.data.column.storage.type.BigIntegerType;
 import org.enso.table.data.column.storage.type.FloatType;
 import org.enso.table.data.column.storage.type.IntegerType;
-import org.enso.table.data.column.storage.type.NullType;
 import org.enso.table.data.column.storage.type.StorageType;
-import org.enso.table.data.table.Column;
 
 /**
  * A binary coalescing operation for numeric types. This class is used to perform operations on two
@@ -26,7 +21,7 @@
  *
  * @param <T> the type of the elements in the column
  */
-public abstract class BinaryCoalescingOperationNumeric<T> implements BinaryOperation<T> {
+public abstract class BinaryCoalescingOperationNumeric<T> extends BinaryOperationNumeric<T, T> {
   /**
    * An abstract class representing a numeric operation. This class defines the methods that must be
    * implemented by any numeric operation.
@@ -103,112 +98,24 @@ public static BinaryOperation<?> create(
     }
   }
 
-  private static StorageType<?> storageTypeForObject(Object right) {
-    if (right == null) {
-      return NullType.INSTANCE;
-    }
-
-    if (right instanceof Column rightColumn) {
-      return rightColumn.getStorage().getType();
-    }
-
-    return StorageType.forBoxedItem(right, PreciseTypeOptions.DEFAULT);
-  }
-
-  protected final StorageType<T> validType;
   protected final NumericOperation operation;
-  protected final StorageType<?>[] validInputs;
 
   protected BinaryCoalescingOperationNumeric(
-      StorageType<T> validType, NumericOperation operation, StorageType<?>... validInputs) {
-    this.validType = validType;
+      NumericColumnAdapter<T> adapter, StorageType<T> returnType, NumericOperation operation) {
+    super(adapter, false, returnType);
     this.operation = operation;
-    this.validInputs = validInputs;
   }
 
   @Override
-  public boolean canApplyMap(ColumnStorage<?> left, Object rightValue) {
-    var leftType = left.getType();
-    if (validType.isOfType(leftType)) {
-      return true;
-    }
-
-    for (var validInput : validInputs) {
-      if (validInput.isOfType(leftType)) {
-        return true;
-      }
-    }
-
-    return false;
-  }
-
-  @Override
-  public boolean canApplyZip(ColumnStorage<?> left, ColumnStorage<?> right) {
-    return canApplyMap(left, null) && canApplyMap(right, null);
+  protected ColumnStorage<T> applyNullMap(
+      ColumnStorage<?> left, MapOperationProblemAggregator problemAggregator) {
+    return adapter.asTypedStorage(left);
   }
 
-  @Override
-  public ColumnStorage<T> applyMap(
-      ColumnStorage<?> left, Object rightValue, MapOperationProblemAggregator problemAggregator) {
-    if (rightValue == null) {
-      return asTypedStorage(left);
-    }
-
-    T rightValueTyped = validType.valueAsType(rightValue);
-    if (rightValueTyped == null) {
-      throw new IllegalArgumentException(
-          "Unsupported right value type " + rightValue.getClass() + ".");
-    }
-
-    return innerApplyMap(asTypedStorage(left), rightValueTyped, problemAggregator);
-  }
-
-  @Override
-  public ColumnStorage<T> applyZip(
-      ColumnStorage<?> left,
-      ColumnStorage<?> right,
-      MapOperationProblemAggregator problemAggregator) {
-    if (NullType.INSTANCE.isOfType(right.getType())) {
-      return validType.asTypedStorage(left);
-    }
-
-    return innerApplyZip(asTypedStorage(left), asTypedStorage(right), problemAggregator);
-  }
-
-  protected abstract ColumnStorage<T> asTypedStorage(ColumnStorage<?> storage);
-
-  protected abstract ColumnStorage<T> innerApplyMap(
-      ColumnStorage<T> left, T right, MapOperationProblemAggregator problemAggregator);
-
-  protected abstract ColumnStorage<T> innerApplyZip(
-      ColumnStorage<T> left,
-      ColumnStorage<T> right,
-      MapOperationProblemAggregator problemAggregator);
-
   private static class BinaryCoalescingOperationDouble
       extends BinaryCoalescingOperationNumeric<Double> {
     public BinaryCoalescingOperationDouble(NumericOperation operation) {
-      super(
-          FloatType.FLOAT_64,
-          operation,
-          BigDecimalType.INSTANCE,
-          BigIntegerType.INSTANCE,
-          IntegerType.INT_64);
-    }
-
-    @Override
-    protected ColumnDoubleStorage asTypedStorage(ColumnStorage<?> storage) {
-      return switch (storage.getType()) {
-        case FloatType floatType -> floatType.asTypedStorage(storage);
-        case BigDecimalType bigDecimalType -> DoubleStorageFacade.forBigDecimal(
-            bigDecimalType.asTypedStorage(storage));
-        case BigIntegerType bigIntegerType -> DoubleStorageFacade.forBigInteger(
-            bigIntegerType.asTypedStorage(storage));
-        case IntegerType integerType -> DoubleStorageFacade.forLong(
-            integerType.asTypedStorage(storage));
-        default -> throw new IllegalArgumentException(
-            "Unsupported storage type: " + storage.getType());
-      };
+      super(DoubleColumnAdapter.INSTANCE, FloatType.FLOAT_64, operation);
     }
 
     @Override
@@ -245,107 +152,60 @@ protected ColumnStorage<Double> innerApplyZip(
             }
           });
     }
-  }
-
-  private abstract static class BinaryCoalescingOperationBigNumber<T>
-      extends BinaryCoalescingOperationNumeric<T> {
-    protected BinaryCoalescingOperationBigNumber(
-        StorageType<T> validType, NumericOperation operation, StorageType<?>... validInputs) {
-      super(validType, operation, validInputs);
-    }
-
-    @Override
-    protected ColumnStorage<T> innerApplyMap(
-        ColumnStorage<T> left, T right, MapOperationProblemAggregator problemAggregator) {
-      return StorageIterators.mapOverStorage(
-          left,
-          false,
-          validType.makeBuilder(left.getSize(), problemAggregator),
-          (index, value) -> value == null ? right : doSingle(value, right, index));
-    }
 
     @Override
-    protected ColumnStorage<T> innerApplyZip(
-        ColumnStorage<T> left,
-        ColumnStorage<T> right,
-        MapOperationProblemAggregator problemAggregator) {
-      return StorageIterators.zipOverStorages(
-          left,
-          right,
-          size -> validType.makeBuilder(size, problemAggregator),
-          false,
-          (index, x, y) -> {
-            if (x == null) {
-              return y;
-            } else {
-              return y == null ? x : doSingle(x, y, index);
-            }
-          });
+    protected Double doSingle(
+        Double left, Double right, long index, MapOperationProblemAggregator problemAggregator) {
+      if (left == null) {
+        return right;
+      } else if (right == null) {
+        return left;
+      } else {
+        return operation.doDouble(left, right, index);
+      }
     }
-
-    protected abstract T doSingle(T left, T right, long index);
   }
 
   private static class BinaryCoalescingOperationBigDecimal
-      extends BinaryCoalescingOperationBigNumber<BigDecimal> {
+      extends BinaryCoalescingOperationNumeric<BigDecimal> {
     public BinaryCoalescingOperationBigDecimal(NumericOperation operation) {
-      super(BigDecimalType.INSTANCE, operation, BigIntegerType.INSTANCE, IntegerType.INT_64);
-    }
-
-    @Override
-    protected ColumnStorage<BigDecimal> asTypedStorage(ColumnStorage<?> storage) {
-      return switch (storage.getType()) {
-        case BigDecimalType bigDecimalType -> bigDecimalType.asTypedStorage(storage);
-        case BigIntegerType bigIntegerType -> new ColumnStorageFacade<>(
-            bigIntegerType.asTypedStorage(storage), BigDecimal::new);
-        case IntegerType integerType -> new ColumnStorageFacade<>(
-            integerType.asTypedStorage(storage), BigDecimal::valueOf);
-        default -> throw new IllegalArgumentException(
-            "Unsupported storage type: " + storage.getType());
-      };
+      super(BigDecimalColumnAdapter.INSTANCE, BigDecimalType.INSTANCE, operation);
     }
 
     @Override
-    protected BigDecimal doSingle(BigDecimal left, BigDecimal right, long index) {
-      return operation.doBigDecimal(left, right, index);
+    protected BigDecimal doSingle(
+        BigDecimal left,
+        BigDecimal right,
+        long index,
+        MapOperationProblemAggregator problemAggregator) {
+      return left == null
+          ? right
+          : (right == null ? left : operation.doBigDecimal(left, right, index));
     }
   }
 
   private static class BinaryCoalescingOperationBigInteger
-      extends BinaryCoalescingOperationBigNumber<BigInteger> {
+      extends BinaryCoalescingOperationNumeric<BigInteger> {
     public BinaryCoalescingOperationBigInteger(NumericOperation operation) {
-      super(BigIntegerType.INSTANCE, operation, IntegerType.INT_64);
-    }
-
-    @Override
-    protected ColumnStorage<BigInteger> asTypedStorage(ColumnStorage<?> storage) {
-      return switch (storage.getType()) {
-        case BigIntegerType bigIntegerType -> bigIntegerType.asTypedStorage(storage);
-        case IntegerType integerType -> new ColumnStorageFacade<>(
-            integerType.asTypedStorage(storage), BigInteger::valueOf);
-        default -> throw new IllegalArgumentException(
-            "Unsupported storage type: " + storage.getType());
-      };
+      super(BigIntegerColumnAdapter.INSTANCE, BigIntegerType.INSTANCE, operation);
     }
 
     @Override
-    protected BigInteger doSingle(BigInteger left, BigInteger right, long index) {
-      return operation.doBigInteger(left, right, index);
+    protected BigInteger doSingle(
+        BigInteger left,
+        BigInteger right,
+        long index,
+        MapOperationProblemAggregator problemAggregator) {
+      return left == null
+          ? right
+          : (right == null ? left : operation.doBigInteger(left, right, index));
     }
   }
 
   private static class BinaryCoalescingOperationLong
       extends BinaryCoalescingOperationNumeric<Long> {
     public BinaryCoalescingOperationLong(NumericOperation operation) {
-      super(IntegerType.INT_64, operation);
-    }
-
-    @Override
-    protected ColumnLongStorage asTypedStorage(ColumnStorage<?> storage) {
-      if (storage.getType() instanceof IntegerType integerType) {
-        return integerType.asTypedStorage(storage);
-      }
-      throw new IllegalArgumentException("Unsupported storage type: " + storage.getType());
+      super(LongColumnAdapter.INSTANCE, IntegerType.INT_64, operation);
     }
 
     @Override
@@ -368,7 +228,7 @@ protected ColumnStorage<Long> innerApplyZip(
       return StorageIterators.zipOverLongStorages(
           (ColumnLongStorage) left,
           (ColumnLongStorage) right,
-          s -> validType.makeBuilder(s, problemAggregator),
+          s -> IntegerType.INT_64.makeBuilder(s, problemAggregator),
           true,
           (index, value1, isNothing1, value2, isNothing2) -> {
             if (isNothing1 && isNothing2) {
@@ -382,5 +242,17 @@ protected ColumnStorage<Long> innerApplyZip(
             }
           });
     }
+
+    @Override
+    protected Long doSingle(
+        Long left, Long right, long index, MapOperationProblemAggregator problemAggregator) {
+      if (left == null) {
+        return right;
+      } else if (right == null) {
+        return left;
+      } else {
+        return operation.doLong(left, right, index);
+      }
+    }
   }
 }