Add polymorphic window functions

pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/window/aggregate/BoolAndValueAggregator.java → pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/window/aggregate/BoolAndWindowValueAggregator.java

@@ -25,7 +25,7 @@
 /**
  * Window value aggregator for BOOL_AND window function.
  */
-public class BoolAndValueAggregator implements WindowValueAggregator<Object> {
+public class BoolAndWindowValueAggregator implements WindowValueAggregator<Object> {
   private int _numFalse = 0;
   private int _numTrue = 0;
   private int _numNull = 0;

pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/window/aggregate/BoolOrValueAggregator.java → pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/window/aggregate/BoolOrWindowValueAggregator.java

@@ -25,7 +25,7 @@
 /**
  * Window value aggregator for BOOL_OR window function.
  */
-public class BoolOrValueAggregator implements WindowValueAggregator<Object> {
+public class BoolOrWindowValueAggregator implements WindowValueAggregator<Object> {
   private int _numFalse = 0;
   private int _numTrue = 0;
   private int _numNull = 0;

pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/window/aggregate/MaxComparableWindowValueAggregator.java

@@ -0,0 +1,95 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.query.runtime.operator.window.aggregate;
+
+import java.util.ArrayDeque;
+import javax.annotation.Nullable;
+
+
+/**
+ * Window value aggregator for MAX window function that preserves the input type by using {@link Comparable} for
+ * comparisons. Used for types like BIG_DECIMAL that don't have a dedicated primitive-typed aggregator.
+ */
+public class MaxComparableWindowValueAggregator implements WindowValueAggregator<Object> {
+
+  private final boolean _supportRemoval;
+  private final ArrayDeque<Object> _deque = new ArrayDeque<>();
+  private Object _maxValue = null;
+
+  /**
+   * @param supportRemoval whether this window value aggregator should support removal of values. Some cases require
+   *                       only addition of values in which case this value aggregator will have O(1) space complexity;
+   *                       if {@code supportRemoval} is true, this value aggregator will have O(K) space complexity
+   *                       (where K is the max size of the window).
+   */
+  public MaxComparableWindowValueAggregator(boolean supportRemoval) {
+    _supportRemoval = supportRemoval;
+  }
+
+  @Override
+  public void addValue(@Nullable Object value) {
+    if (value != null) {
+      if (_supportRemoval) {
+        // Remove previously added elements if they're < than the current element since they're no longer useful
+        while (!_deque.isEmpty() && compare(_deque.peekLast(), value) < 0) {
+          _deque.pollLast();
+        }
+        _deque.addLast(value);
+      } else {
+        if (_maxValue == null || compare(value, _maxValue) > 0) {
+          _maxValue = value;
+        }
+      }
+    }
+  }
+
+  @Override
+  public void removeValue(@Nullable Object value) {
+    if (!_supportRemoval) {
+      throw new UnsupportedOperationException();
+    }
+
+    if (value != null) {
+      if (!_deque.isEmpty() && compare(_deque.peekFirst(), value) == 0) {
+        _deque.pollFirst();
+      }
+    }
+  }
+
+  @Nullable
+  @Override
+  public Object getCurrentAggregatedValue() {
+    if (_supportRemoval) {
+      return _deque.peekFirst();
+    } else {
+      return _maxValue;
+    }
+  }
+
+  @Override
+  public void clear() {
+    _deque.clear();
+    _maxValue = null;
+  }
+
+  @SuppressWarnings({"unchecked", "rawtypes"})
+  private static int compare(Object a, Object b) {
+    return ((Comparable) a).compareTo(b);
+  }
+}

pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/window/aggregate/MaxWindowValueAggregator.java → pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/window/aggregate/MaxDoubleWindowValueAggregator.java

@@ -25,7 +25,7 @@
 /**
  * Window value aggregator for MAX window function.
  */
-public class MaxWindowValueAggregator implements WindowValueAggregator<Object> {
+public class MaxDoubleWindowValueAggregator implements WindowValueAggregator<Object> {
 
   private final boolean _supportRemoval;
   private final DoubleArrayFIFOQueue _deque = new DoubleArrayFIFOQueue();
@@ -37,7 +37,7 @@ public class MaxWindowValueAggregator implements WindowValueAggregator<Object> {
    *                       if {@code supportRemoval} is true, this value aggregator will have O(K) space complexity
    *                       (where K is the max size of the window).
    */
-  public MaxWindowValueAggregator(boolean supportRemoval) {
+  public MaxDoubleWindowValueAggregator(boolean supportRemoval) {
     _supportRemoval = supportRemoval;
   }
 
@@ -73,6 +73,7 @@ public void removeValue(@Nullable Object value) {
     }
   }
 
+  @Nullable
   @Override
   public Object getCurrentAggregatedValue() {
     if (_supportRemoval) {

pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/window/aggregate/MaxIntWindowValueAggregator.java

@@ -0,0 +1,96 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.query.runtime.operator.window.aggregate;
+
+import it.unimi.dsi.fastutil.ints.IntArrayFIFOQueue;
+import javax.annotation.Nullable;
+
+
+/**
+ * Window value aggregator for MAX window function over INT types.
+ * Uses primitive {@code int} operations and {@link IntArrayFIFOQueue} to avoid boxing overhead and improve cache
+ * locality compared to the {@link Comparable}-based aggregator.
+ */
+public class MaxIntWindowValueAggregator implements WindowValueAggregator<Object> {
+
+  private final boolean _supportRemoval;
+  private final IntArrayFIFOQueue _deque = new IntArrayFIFOQueue();
+  private Integer _maxValue = null;
+
+  /**
+   * @param supportRemoval whether this window value aggregator should support removal of values. Some cases require
+   *                       only addition of values in which case this value aggregator will have O(1) space complexity;
+   *                       if {@code supportRemoval} is true, this value aggregator will have O(K) space complexity
+   *                       (where K is the max size of the window).
+   */
+  public MaxIntWindowValueAggregator(boolean supportRemoval) {
+    _supportRemoval = supportRemoval;
+  }
+
+  @Override
+  public void addValue(@Nullable Object value) {
+    if (value != null) {
+      int intValue = ((Number) value).intValue();
+      if (_supportRemoval) {
+        // Remove previously added elements if they're < than the current element since they're no longer useful
+        while (!_deque.isEmpty() && _deque.lastInt() < intValue) {
+          _deque.dequeueLastInt();
+        }
+        _deque.enqueue(intValue);
+      } else {
+        if (_maxValue == null || intValue > _maxValue) {
+          _maxValue = intValue;
+        }
+      }
+    }
+  }
+
+  @Override
+  public void removeValue(@Nullable Object value) {
+    if (!_supportRemoval) {
+      throw new UnsupportedOperationException();
+    }
+
+    if (value != null) {
+      int intValue = ((Number) value).intValue();
+      if (!_deque.isEmpty() && _deque.firstInt() == intValue) {
+        _deque.dequeueInt();
+      }
+    }
+  }
+
+  @Nullable
+  @Override
+  public Object getCurrentAggregatedValue() {
+    if (_supportRemoval) {
+      if (_deque.isEmpty()) {
+        return null;
+      }
+      return _deque.firstInt();
+    } else {
+      return _maxValue;
+    }
+  }
+
+  @Override
+  public void clear() {
+    _deque.clear();
+    _maxValue = null;
+  }
+}

pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/window/aggregate/MaxLongWindowValueAggregator.java

@@ -0,0 +1,97 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.query.runtime.operator.window.aggregate;
+
+import it.unimi.dsi.fastutil.longs.LongArrayFIFOQueue;
+import javax.annotation.Nullable;
+
+
+/**
+ * Window value aggregator for MAX window function over LONG types.
+ * Uses primitive {@code long} operations and {@link LongArrayFIFOQueue} to avoid boxing overhead and improve cache
+ * locality compared to the {@link Comparable}-based aggregator. Also avoids precision loss that occurs when converting
+ * large long values to double.
+ */
+public class MaxLongWindowValueAggregator implements WindowValueAggregator<Object> {
+
+  private final boolean _supportRemoval;
+  private final LongArrayFIFOQueue _deque = new LongArrayFIFOQueue();
+  private Long _maxValue = null;
+
+  /**
+   * @param supportRemoval whether this window value aggregator should support removal of values. Some cases require
+   *                       only addition of values in which case this value aggregator will have O(1) space complexity;
+   *                       if {@code supportRemoval} is true, this value aggregator will have O(K) space complexity
+   *                       (where K is the max size of the window).
+   */
+  public MaxLongWindowValueAggregator(boolean supportRemoval) {
+    _supportRemoval = supportRemoval;
+  }
+
+  @Override
+  public void addValue(@Nullable Object value) {
+    if (value != null) {
+      long longValue = ((Number) value).longValue();
+      if (_supportRemoval) {
+        // Remove previously added elements if they're < than the current element since they're no longer useful
+        while (!_deque.isEmpty() && _deque.lastLong() < longValue) {
+          _deque.dequeueLastLong();
+        }
+        _deque.enqueue(longValue);
+      } else {
+        if (_maxValue == null || longValue > _maxValue) {
+          _maxValue = longValue;
+        }
+      }
+    }
+  }
+
+  @Override
+  public void removeValue(@Nullable Object value) {
+    if (!_supportRemoval) {
+      throw new UnsupportedOperationException();
+    }
+
+    if (value != null) {
+      long longValue = ((Number) value).longValue();
+      if (!_deque.isEmpty() && _deque.firstLong() == longValue) {
+        _deque.dequeueLong();
+      }
+    }
+  }
+
+  @Nullable
+  @Override
+  public Object getCurrentAggregatedValue() {
+    if (_supportRemoval) {
+      if (_deque.isEmpty()) {
+        return null;
+      }
+      return _deque.firstLong();
+    } else {
+      return _maxValue;
+    }
+  }
+
+  @Override
+  public void clear() {
+    _deque.clear();
+    _maxValue = null;
+  }
+}