[WIP] Rust Native POC

pinot-core/pom.xml

@@ -106,6 +106,11 @@
       <groupId>org.apache.pinot</groupId>
       <artifactId>pinot-timeseries-spi</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.apache.pinot</groupId>
+      <artifactId>pinot-native</artifactId>
+      <version>${project.version}</version>
+    </dependency>
 
     <dependency>
       <groupId>org.apache.pinot</groupId>

pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionFactory.java

@@ -71,6 +71,12 @@ public static AggregationFunction getAggregationFunction(FunctionContext functio
       List<ExpressionContext> arguments = function.getArguments();
       int numArguments = arguments.size();
       ExpressionContext firstArgument = arguments.get(0);
+      // Single integration point for the native (Rust+JNI) acceleration path. See
+      // NativeAggregationRouter for eligibility rules. When disabled or ineligible,
+      // construction falls through to the standard Java factory branches below.
+      if (NativeAggregationRouter.shouldAccelerate(upperCaseFunctionName, arguments, nullHandlingEnabled)) {
+        return NativeAggregationRouter.createNative(upperCaseFunctionName, arguments, nullHandlingEnabled);
+      }
       if (upperCaseFunctionName.startsWith("PERCENTILE")) {
         String remainingFunctionName = upperCaseFunctionName.substring(10);
         if (remainingFunctionName.equals("SMARTTDIGEST")) {

pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/NativeAggregationRouter.java

@@ -0,0 +1,114 @@
+/**
+ * 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.core.query.aggregation.function;
+
+import java.util.List;
+import org.apache.pinot.common.request.context.ExpressionContext;
+import org.apache.pinot.nativeengine.agg.PinotNativeAgg;
+
+
+/**
+ * Routes eligible aggregation function constructions to the native (Rust+JNI) acceleration
+ * path. This is the single integration point between Pinot's aggregation engine and
+ * {@code pinot-native}.
+ *
+ * <p>Eligibility checks (short-circuiting):
+ * <ol>
+ *   <li>Feature flag {@code pinot.native.aggregation.enabled} is set to {@code true}</li>
+ *   <li>The native library loaded successfully ({@link PinotNativeAgg#isAvailable()})</li>
+ *   <li>Null handling is disabled (no native null path yet)</li>
+ *   <li>Function type is in the Phase 1 POC scope: {@code SUM} / {@code SUM0}</li>
+ *   <li>The aggregated expression is a simple column identifier (no transforms)</li>
+ * </ol>
+ *
+ * <p>When any check fails the caller must construct the original Java AggregationFunction
+ * unchanged. The router never throws.
+ *
+ * <p>Routing here covers all aggregation contexts in Pinot — SSE V1, MSE leaf (which
+ * delegates to V1), MSE intermediate, star-tree, realtime consuming segments, and
+ * Materialized View refresh — because every one of them obtains AggregationFunction
+ * instances from {@link AggregationFunctionFactory}. See {@code docs/native/phase-1-design.md}
+ * §2 for the engine landscape.
+ */
+public final class NativeAggregationRouter {
+
+  /** System property gating the native engine. Default {@code false}. */
+  public static final String ENABLED_PROPERTY = "pinot.native.aggregation.enabled";
+
+  private NativeAggregationRouter() {
+  }
+
+  /**
+   * @return {@code true} if a native AggregationFunction should be constructed for the
+   *         given function name + arguments. Caller must use {@link #createNative} to do so.
+   */
+  public static boolean shouldAccelerate(
+      String upperCaseFunctionName, List<ExpressionContext> arguments,
+      boolean nullHandlingEnabled) {
+    if (!enabled()) {
+      return false;
+    }
+    if (!PinotNativeAgg.isAvailable()) {
+      return false;
+    }
+    if (nullHandlingEnabled) {
+      return false;
+    }
+    if (!isInScopeFunction(upperCaseFunctionName)) {
+      return false;
+    }
+    return isSimpleColumnArg(arguments);
+  }
+
+  /**
+   * Constructs a native AggregationFunction for a name + arguments combination previously
+   * accepted by {@link #shouldAccelerate}. Caller is responsible for the eligibility check.
+   *
+   * @throws IllegalStateException if the function name is not in scope (programming error)
+   */
+  @SuppressWarnings("rawtypes")
+  public static AggregationFunction createNative(
+      String upperCaseFunctionName, List<ExpressionContext> arguments,
+      boolean nullHandlingEnabled) {
+    switch (upperCaseFunctionName) {
+      case "SUM":
+      case "SUM0":
+        return new NativeSumAggregationFunction(arguments, nullHandlingEnabled);
+      default:
+        throw new IllegalStateException(
+            "Native AggregationFunction requested for unsupported function: "
+                + upperCaseFunctionName);
+    }
+  }
+
+  static boolean enabled() {
+    return Boolean.getBoolean(ENABLED_PROPERTY);
+  }
+
+  private static boolean isInScopeFunction(String upperCaseFunctionName) {
+    return "SUM".equals(upperCaseFunctionName) || "SUM0".equals(upperCaseFunctionName);
+  }
+
+  private static boolean isSimpleColumnArg(List<ExpressionContext> arguments) {
+    if (arguments.size() != 1) {
+      return false;
+    }
+    return arguments.get(0).getType() == ExpressionContext.Type.IDENTIFIER;
+  }
+}

pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/NativeSumAggregationFunction.java

@@ -0,0 +1,67 @@
+/**
+ * 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.core.query.aggregation.function;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.pinot.common.request.context.ExpressionContext;
+import org.apache.pinot.core.common.BlockValSet;
+import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
+import org.apache.pinot.nativeengine.agg.PinotNativeAgg;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+
+
+/**
+ * SUM aggregation accelerated by the native (Rust+JNI) engine. POC scope: handles
+ * {@code LONG} single-value columns via {@link PinotNativeAgg#sumLong(long[], int)} and
+ * defers to the Java parent class for all other type / encoding combinations.
+ *
+ * <p>Construction is gated by {@link NativeAggregationRouter#shouldAccelerate}; this class
+ * is never instantiated directly by user code.
+ *
+ * <p>This class extends {@link SumAggregationFunction} so it inherits identical intermediate
+ * and final result types, merge semantics, and group-by hooks. Mixed-version clusters
+ * (native server + Java server) produce byte-for-byte identical intermediate results.
+ */
+@SuppressWarnings({"rawtypes", "unchecked"})
+public class NativeSumAggregationFunction extends SumAggregationFunction {
+
+  public NativeSumAggregationFunction(List<ExpressionContext> arguments,
+      boolean nullHandlingEnabled) {
+    super(arguments, nullHandlingEnabled);
+  }
+
+  @Override
+  public void aggregate(int length, AggregationResultHolder aggregationResultHolder,
+      Map<ExpressionContext, BlockValSet> blockValSetMap) {
+    BlockValSet blockValSet = blockValSetMap.get(_expression);
+    if (blockValSet.isSingleValue() && blockValSet.getValueType().getStoredType() == DataType.LONG) {
+      long[] values = blockValSet.getLongValuesSV();
+      double nativeSum = PinotNativeAgg.sumLong(values, length);
+      // NaN is the native sentinel for "kernel error" — fall through to Java in that case
+      // rather than propagate the sentinel into the result holder.
+      if (!Double.isNaN(nativeSum)) {
+        double prev = aggregationResultHolder.getDoubleResult();
+        aggregationResultHolder.setValue(prev + nativeSum);
+        return;
+      }
+    }
+    super.aggregate(length, aggregationResultHolder, blockValSetMap);
+  }
+}