[WIP] Rust Acceleration Kernel POC#208
Conversation
Strategic doc (RUST_REWRITE_DESIGN.md) and Phase 1 detailed design (docs/native/phase-1-design.md) for progressively accelerating Pinot's query execution with Rust kernels exposed via JNI. Key settled decisions captured in the per-doc decision logs: - Phase 1 target is aggregation + group-by, not filter scan - Integration point is AggregationFunctionFactory (covers SSE V1, MSE leaf, MSE intermediate, star-tree, realtime, and MV refresh in one hook) - POC scope is SUM(LONG) end-to-end via NativeSumAggregationFunction - HLL deferred to Phase 1.E to keep clearspring parity off the critical path - Classic JNI for now; FFM/Panama planned when Java 22 becomes the floor Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Scaffolds the Phase 0 / 1.A POC plumbing for native acceleration:
- Cargo workspace under pinot-native/native/ with two crates:
- kernels/: pure-Rust SIMD kernels (sum_i64_to_f64, 4-way unrolled),
no JNI dep, testable standalone
- ffi/: cdylib that re-exports kernels via JNI, with panic-catching
wrappers and GetPrimitiveArrayCritical zero-copy pinning
- Maven module pinot-native/ targeting Java 11 bytecode, with
exec-maven-plugin invoking cargo during process-resources and test
phases. OS+arch profiles set ${native.lib.filename} so surefire can
pass -Dpinot.native.lib.path=... to the JVM.
- PinotNativeAgg (Java) declares the static native entry points. Class
initializer runs NativeLibLoader, which resolves the library via:
1. -Dpinot.native.lib.path system property (dev)
2. classpath resource /native/<os>-<arch>/lib<name>.<ext> (packaged)
3. System.loadLibrary fallback to java.library.path
Load failure sets isAvailable() = false and callers fall back to Java.
- Five TestNG smoke tests pass on darwin-aarch64: probe magic number,
empty input, small-range sum, length-arg respected, 1M-element random
sum matching Java reference within float tolerance.
Required explicit jsr305 + jspecify deps since the root pom's
package-info-maven-plugin generates @ParametersAreNonnullByDefault and
@NullMarked annotations on every package, and other modules pick those
up transitively via Guava — which this module deliberately doesn't pull
in.
Verified: ./mvnw -pl pinot-native test passes end-to-end in 4.3 s.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Adds the single integration point: AggregationFunctionFactory checks
NativeAggregationRouter.shouldAccelerate() at the top of
getAggregationFunction() and, on eligibility, constructs the native
subclass instead of the standard Java AggregationFunction. Because all
aggregation contexts in Pinot (SSE V1, MSE leaf, MSE intermediate,
star-tree, realtime, MV refresh) obtain functions from this factory,
this single fork accelerates all of them.
Eligibility rules (short-circuiting):
1. -Dpinot.native.aggregation.enabled=true
2. PinotNativeAgg.isAvailable()
3. nullHandlingEnabled == false (no native null path yet)
4. function name in {SUM, SUM0} (POC scope; expands in Phase 1.B)
5. single-argument IDENTIFIER expression (no transforms)
NativeSumAggregationFunction extends SumAggregationFunction and
overrides aggregate() to call PinotNativeAgg.sumLong for LONG-typed
single-value columns; falls through to super for all other type /
encoding combinations and for kernel failures (NaN sentinel). Group-by
remains on the Java path for now (Phase 1.D will add).
NativeSumAggregationFunctionTest exercises:
- factory returns NativeSumAggregationFunction with flag on
- factory returns plain SumAggregationFunction with flag off or when
null handling is enabled
- aggregate() on LONG matches Java reference on 100k random values
- aggregate() falls through to super on INT (out of POC scope)
WIP — test currently fails checkstyle on cosmetic LeftCurly violations
in the StubBlockValSet inner class. Functional path (router + native
function) is implemented; only the test formatting needs cleanup.
See docs/native/phase-1-design.md sections 2, 3, 8 for the engine
landscape and the rationale for routing at the factory layer rather
than the originally-proposed plan-maker fork.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Reformats the StubBlockValSet inner class so each method body sits on its own line (Pinot's LeftCurly rule rejects single-line method bodies) and drops two unused imports. Also picks up a trivial whitespace cleanup in PinotNativeAgg. With this commit, ./mvnw -pl pinot-core -am -Dtest=NativeSumAggregationFunctionTest -Dsurefire.failIfNoSpecifiedTests=false test passes all five cases: factoryReturnsNativeImplWhenFlagOnAndEligible factoryReturnsJavaImplWhenFlagOff factoryReturnsJavaImplWhenNullHandlingEnabled aggregateLongMatchesJavaReference (100k random longs) aggregateFallsThroughForIntColumn Phase 1.A POC is now demonstrated end-to-end: a SUM(LONG) query routes through AggregationFunctionFactory → NativeAggregationRouter → NativeSumAggregationFunction → PinotNativeAgg.sumLong → Rust kernel, with identical results to the Java path. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
The Phase 1.A POC kernel was scalar Rust with manual 4-way unrolling,
relying on LLVM auto-vectorization. On Apple Silicon the compiler did
emit NEON instructions for the load and convert, but the reduction was
scalarized (extracting one f64 lane at a time and adding into a scalar
accumulator), and on AVX2 x86 the same source would not vectorize the
i64->f64 conversion at all. That fell short of the "state of the art
kernels" Phase 1 spec.
Rewrites sum_i64_to_f64 with four explicit backends behind runtime ISA
dispatch:
AVX-512DQ (x86_64): _mm512_cvtepi64_pd + 4 × 512-bit accumulators
(8 lanes each = 32-wide ILP)
AVX2 (x86_64): scalar vcvtsi2sd + _mm256_set_pd packing into
4 × 256-bit accumulators (16-wide ILP). AVX2 has
no vcvtqq2pd; the conversion is the bottleneck.
Mysticial bit-trick is a future optimization.
NEON (aarch64): vld1q_s64 + vcvtq_f64_s64 + vaddq_f64 with 4 ×
128-bit accumulators (8-wide ILP). Native i64->f64
vector convert exists on ARM.
scalar (any): existing 4-way unrolled fallback, also serves as
the reference for property-based equivalence tests.
All non-scalar paths are #[target_feature(...)] unsafe fns called only
after is_x86_feature_detected! / is_aarch64_feature_detected! returns
true. Detection results are cached by std::arch after the first call,
so the per-call dispatch cost is one atomic load.
Generated assembly verified: the hot loop is now full-vector through
the reduction (fadd.2d across all four NEON accumulators, then faddp.2d
for horizontal collapse), where the previous version reduced into a
scalar register.
Java integration test (NativeSumAggregationFunctionTest, 5 cases
including 100k-element correctness vs Java reference) still passes —
the JNI signature is unchanged and result semantics are preserved
within the documented float tolerance.
Updates docs/native/phase-1-design.md §6.1 and adds a decision log
entry noting this work moves from Phase 1.B back to 1.A.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Apache RAT plugin requires every file in the repo to carry the standard ASF license header (with limited exclusions for binary artifacts, lock files, etc.). The newly-added markdown docs, Cargo.toml manifests, and the pinot-native README were unflagged in CI until ./mvnw install ran the verify phase. Adds the standard ASF header in the comment style each file format supports: - RUST_REWRITE_DESIGN.md (HTML-comment header, matches CONTRIBUTING.md) - docs/native/phase-1-design.md (HTML-comment header) - pinot-native/README.md (HTML-comment header) - pinot-native/native/Cargo.toml (#-comment header, supported by TOML) - pinot-native/native/ffi/Cargo.toml (#-comment header) - pinot-native/native/kernels/Cargo.toml (#-comment header) Verified with ./mvnw -pl pinot-perf -am install -DskipTests=true running through RAT cleanly. No functional changes. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Compares NativeSumAggregationFunction (JNI -> Rust SIMD kernel) against
SumAggregationFunction on a synthetic LONG column, parameterized over
block length so we can characterise where JNI per-call overhead
amortises against the SIMD speedup of the kernel.
Parameters:
_engine in {java, native}
_length in {100, 1000, 10000, 100000}
(Pinot's typical block size is 10_000)
Construction is direct (not through AggregationFunctionFactory) so the
benchmark measures kernel + JNI cost, not routing logic. The native
library is loaded via the dev-build location ../pinot-native/native/
target/release/libpinot_native.{dylib,so,dll} — main() resolves the
path and forwards -Dpinot.native.lib.path=... to forked JVMs.
Build prerequisite: ./mvnw -pl pinot-native package (or any -am build
through pinot-native) to produce the dylib.
Status: benchmark compiles cleanly (passes checkstyle, license, RAT).
Run path via 'mvn exec:java' fails because Maven's exec plugin doesn't
build a JMH-friendly classpath — JMH forks the JVM and can't find
ForkedMain in the resulting class loader. Will resolve in a follow-up
by running via 'java -cp <build-classpath>' directly or by adding the
jmh-maven-plugin to pinot-perf. No numbers yet.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Expose sum_i64_to_f64_scalar via a dedicated JNI entry point (sumLongScalar) and add a benchmark-only NativeScalarSumAggregationFunction in pinot-perf so BenchmarkNativeSumLongAggregation can compare three implementations: Java SUM, Rust scalar + JNI, and Rust + NEON SIMD + JNI. Lets the harness attribute the speedup between language code-gen and explicit SIMD. The FFI helper sum_long_with is generic over the kernel so the production sumLong path remains a direct call (compile-time monomorphization, no fn-pointer indirection introduced). @fork bumped to 3 so the attribution numbers carry cross-fork confidence. Added sumLongScalarMatchesSumLong correctness test. At 100K rows on Apple Silicon (NEON): 6.48x total speedup decomposes as ~3.15x from Rust code-gen and ~2.05x from SIMD on top, with ~85 ns JNI fixed cost. Three-way table and methodology recorded in section 11.A of docs/native/phase-1-design.md. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
§1 (Scope) and §15 (Decision Log) updated to reflect the
2026-05-30 scope-locking decisions:
* Phase 1 agg scope = {SUM, MIN, MAX, COUNT, DISTINCT_COUNT (exact),
DISTINCT_COUNT_HLL} × {INT, LONG, FLOAT, DOUBLE} where applicable.
All other DISTINCT_COUNT variants (RAW_HLL, THETA_SKETCH,
TUPLE_SKETCH, etc.) and AVG/PERCENTILE/STDDEV are explicitly
Phase 2+.
* SwissTable (Task #11) is the shared substrate for both
DISTINCT_COUNT exact and vectorized GROUP BY. A set is a map with
() values, and the batch-probe path that unlocks GROUP BY perf
is identical to what DISTINCT_COUNT exact wants for SV blocks.
Designing for both callers from day one prevents a costly
retrofit.
No code changes.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Phase 1.B.1 stage 1: widen the SUM kernel family from LONG-only to all four numeric primitive types. Rust side --------- Split sum.rs into per-type submodules under sum/. Each kernel (int, float, double, long) ships the same four variants: * public dispatcher -- runtime ISA detection (NEON/AVX2/AVX-512) * scalar fallback -- 4-way unrolled, pub for benchmark attribution * SIMD intrinsics -- explicit NEON, AVX2, and AVX-512F lanes Lane widths per loop: * INT (i32 -> f64): 16 (NEON) / 16 (AVX2) / 32 (AVX-512F) * FLOAT (f32 -> f64): 16 / 16 / 32 * DOUBLE (f64 -> f64): 8 / 16 / 32 (no conversion, pure reduce) * LONG (i64 -> f64): 8 / 16 / 32 (AVX-512DQ required for cvt) 25 Rust unit tests pass (dispatch-vs-scalar, NEON-vs-scalar, tail handling, extreme values, etc.). FFI side -------- Refactor ffi/src/lib.rs around a single define_sum_jni! macro that handles panic::catch_unwind, GetPrimitiveArrayCritical pinning, and length clamping uniformly. Five entry points exported: sumLong, sumLongScalar, sumInt, sumFloat, sumDouble. Kernel is referenced by path so LLVM emits a direct call (no fn-pointer indirection). Java side --------- PinotNativeAgg gains three native declarations (sumInt, sumFloat, sumDouble). PinotNativeAggTest gains 9 new tests (3 per new type: EmptyIsZero, RespectsLengthArgument, LargeRandomMatchesJava). All 17 tests pass. The forced-scalar sumLongScalar entry exists only for the benchmark attribution in design doc S11.A; production routing does not use it. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Phase 1.B.1 stage 2: wire the new JNI entry points (sumInt, sumFloat, sumDouble) into NativeSumAggregationFunction so the native engine handles all four numeric primitive types end to end, not just LONG. NativeAggregationRouter needs no change -- it filters on function name and arg shape but never inspects column type. Type dispatch lives in NativeSumAggregationFunction.aggregate where the BlockValSet is available; the method now switches on getStoredType() and dispatches to the matching PinotNativeAgg kernel. Unsupported types (BIG_DECIMAL, etc.) fall through to the Java parent class as before. Tests ----- NativeSumAggregationFunctionTest grows three new aggregate*MatchesJavaReference cases (INT, FLOAT, DOUBLE), mirroring the existing LONG test. StubBlockValSet is refactored around forInt/forLong/forFloat/forDouble factory methods and now holds optional arrays for all four primitive types. The previous aggregateFallsThroughForIntColumn test is repurposed: INT is no longer a fallthrough, it is the native path. NOTE: not verified locally on this Mac. li-pinot/master pins grpc.version=1.68.3 whose osx-aarch_64 artifact on Maven Central is mislabeled (actually x86_64), and Rosetta is not installed here. apache/master uses grpc 1.81.0 (universal binary, arm64-capable). Will validate on LinkedIn Linux CI. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Two pre-existing cherry-pick drift issues only surfaced when we tried
the first cross-module pinot-core build on li-pinot base for the
per-type JMH harness:
1. pinot-native/pom.xml's <parent><version> was hardcoded
1.6.0-SNAPSHOT (apache/master's version when the module was
authored). After cherry-pick onto li-pinot/master (version
1.4.0-SNAPSHOT) it was never updated, so pinot-native installed
as 1.6.0-SNAPSHOT while pinot-core resolved its dependency via
${project.version} = 1.4.0-SNAPSHOT, and never found the artifact.
Earlier Phase 1.B work stayed inside the pinot-native module, so
this was latent.
2. NativeSumAggregationFunctionTest's StubBlockValSet override of
BigDecimal[][] getBigDecimalValuesMV() does not exist on
li-pinot's older BlockValSet interface. Apache added it
post-divergence. Remove the override (and method body) for li-pinot
compatibility; will need to be added back if/when this branch is
ever ported to apache/master.
No functional change. Just makes pinot-core compile on li-pinot base.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Extends the §11.A three-way attribution pattern (Java vs native-scalar vs native) from LONG-only to all four numeric primitive types so the per-type JMH matrix in the next commit can isolate the SIMD contribution per type. * FFI: three more define_sum_jni! macro invocations for sumIntScalar, sumFloatScalar, sumDoubleScalar -- each routes to the matching sum_<type>_to_f64_scalar kernel (the 4-way unrolled scalar fallback that already existed for benchmark visibility). * PinotNativeAgg: three native method declarations alongside the existing sumInt/sumFloat/sumDouble. * PinotNativeAggTest: three parity tests asserting that the scalar entry's result matches the dispatched entry's result within floating-point tolerance (catches accidental divergence between the two kernels that would invalidate any attribution numbers). * NativeScalarSumAggregationFunction (pinot-perf): widened to switch on getStoredType() and dispatch to the right *Scalar entry, mirroring the production NativeSumAggregationFunction shape. Production routing (NativeSumAggregationFunction in pinot-core) does NOT call the *Scalar entries; they remain pinot-perf-only by design, so the benchmark-only surface does not leak into production module APIs. All 20 PinotNativeAggTest tests pass (17 prior + 3 new parity tests). Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Benchmark: rename BenchmarkNativeSumLongAggregation to
BenchmarkNativeSumAggregation (file rename was already in a previous
commit; this commit fills in the new contents). Add _type Param over
{INT, LONG, FLOAT, DOUBLE} crossed with the existing _engine and
_length axes, yielding 48 cells. Inline a self-contained
PrimitiveBlockValSet stub so the benchmark does not need INT/FLOAT
helpers added to pinot-core's test-only SyntheticBlockValSets.
Results (Apple Silicon NEON, @fork(3) × 20 iter, written to design
doc §11.B):
type Java -> Rust+NEON Rust-lang SIMD ns/elem (native, 100K)
INT 6.47x 3.15x 2.05x 0.0843
LONG 6.44x 3.16x 2.04x 0.0842
FLOAT 6.11x 3.28x 1.86x 0.0844
DOUBLE 5.73x 3.44x 1.66x 0.0852
At production block size (10K):
INT 5.92x, LONG 6.09x, FLOAT 5.58x, DOUBLE 5.87x.
All four types clear the §11.1 NonGroupedAggBench ≥4× single-thread
target. Phase 1.B.1 exit criterion met; cleared for Phase 1.B.2
(MIN/MAX × 4 types).
Design doc updates:
* §11.B new section with full per-type tables, cross-type comparison,
JNI-fixed-cost breakdown, interpretation, and decision signal.
* §15 four new decision log entries for 2026-05-30: 1.B.1 exit,
scalar JNI design, pom.xml port fix, BlockValSet drift handling.
Key observation: at large N the per-element throughput converges to
~0.084 ns/elem for all four types -- INT (4 bytes) and LONG (8 bytes)
hit the same throughput, so the kernel is compute-bound on cvt+add
at M-series clocks, NOT memory-bandwidth-bound. x86 hosts may show
different per-type ordering once we run the cloud harness.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Phase 1.B.2 Rust + JNI surface. 8 new SIMD kernels (MIN/MAX x 4 types), each with the same NEON/AVX2/AVX-512F/scalar dispatch shape as SUM. Notable kernel-side details --------------------------- * i64 MIN/MAX lacks a native vector instruction on NEON and AVX2 (no vminq_s64, no _mm256_min_epi64). Synthesized from vcgtq_s64 + vbslq_s64 (NEON) and _mm256_cmpgt_epi64 + _mm256_blendv_epi8 (AVX2). AVX-512F has _mm512_min_epi64 natively. * FP MIN/MAX must propagate NaN to match Java's Math.min/max. NEON's vminq_*/vmaxq_* are NaN-propagating per IEEE 754-2019 (FMIN/FMAX) -- matches Java for free. AVX2/AVX-512 _mm*_min_* are asymmetric: NaN in operand `a` does NOT propagate. To match Java semantics we run a per-chunk _mm*_cmp_*(load, load, _CMP_UNORD_Q) to detect NaN, OR the masks into a sticky "saw NaN" accumulator, and return NaN at the end if it was ever set. ~5-10% overhead, much cheaper than per-iter NaN masking. * Integer MIN/MAX compute in native i32/i64 precision; the f64 conversion at return matches Pinot's per-block min.doubleValue() semantics. Lossy for |min/max| > 2^53 -- same lossy step Java takes. * Empty input returns f64::INFINITY (MIN) / NEG_INFINITY (MAX) to match Pinot's Min/MaxAggregationFunction.DEFAULT_VALUE. FFI macro --------- Renamed define_sum_jni! to define_reduce_jni! and parameterized on the empty-input value. SUM call sites pass 0.0, MIN passes f64::INFINITY, MAX passes f64::NEG_INFINITY. No behavior change to SUM. JNI surface ----------- 8 new entry points: minInt/minLong/minFloat/minDouble + maxInt/maxLong/maxFloat/maxDouble. nm-verified, all symbols exported. Matching native method declarations added to PinotNativeAgg.java. Tests ----- * 71 new Rust unit tests (96 total): per-kernel empty-input returns +/-INF, dispatch-vs-scalar parity on random data, NEON-vs-scalar (when host has NEON), NaN propagation at start / middle / end / chunk-boundary / tail positions, extreme value handling (i32::MIN/MAX, i64::MIN/MAX, +/-INF). * 20 new PinotNativeAggTest tests (40 total): empty-input default values, large-random parity vs Java Math.min/max, NaN propagation for FP. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Phase 1.B.2 Java-side wiring. NativeMinAggregationFunction / NativeMaxAggregationFunction ----------------------------------------------------------- Extend Min/MaxAggregationFunction. Override aggregate() to switch on getStoredType() and dispatch to the matching PinotNativeAgg kernel (minInt / minLong / minFloat / minDouble / maxInt / maxLong / maxFloat / maxDouble). Result is merged into the holder via Math.min/Math.max for ±0 ordering parity. Other types (BIG_DECIMAL) fall through to the Java parent. NaN behavior: a NaN return from the kernel can mean either "JNI sentinel for kernel failure" OR "the block legitimately contained NaN." We can't distinguish at the boundary, so we conservatively fall through to the Java parent on NaN. Java will compute the same NaN result for the legitimate case using its own NaN-propagating Math.min/max path. Net externally-visible behavior is identical. NativeCountAggregationFunction ------------------------------ No JNI. Under the router gate (null handling disabled, simple column or *), COUNT(*) and COUNT(col) both reduce to "holder.setValue(holder.getDoubleResult() + length)" in pure Java. Crossing JNI for a value that IS the block length would add the ~85 ns FFI fixed cost from §11.A with zero kernel benefit. The class exists so NativeAggregationRouter can be uniform across SUM/MIN/MAX/COUNT (every routed function gets a Native* impl), even when the implementation is trivial. NativeAggregationRouter ----------------------- isInScopeFunction widened to accept MIN/MAX/COUNT in addition to SUM/SUM0. createNative dispatches to the matching Native*AggregationFunction. Tests ----- * Extracted NativeSumAggregationFunctionTest's inline StubBlockValSet into a package-private NativePrimitiveBlockValSet shared across all four Native*AggregationFunctionTest classes. Removes ~150 lines of duplication. * NativeMinMaxAggregationFunctionTest: 13 tests. Factory routing for MIN and MAX, NaN fallthrough disqualification, 8 per-type parity-against-Java tests (4 types x 2 ops), 2 NaN propagation tests. * NativeCountAggregationFunctionTest: 6 tests. Factory routing for COUNT(*), COUNT(col), null-handling fallback, length-only semantics (COUNT(*) with empty blockValSetMap adds length to holder), multi-block accumulation, zero-length no-op. * All 26 pinot-core integration tests pass. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Updates: * §6.2 (MIN/MAX) — replaced the stub with the actual landed design: i32 native vector min on every ISA, i64 synthesized compare-and-select on NEON/AVX2 (no native instruction) and native on AVX-512F, FP NaN-propagation tactics per ISA (NEON propagates for free, x86 needs sticky _CMP_UNORD_Q tracking). * §6.3 (COUNT) — documented the no-JNI fast path design and why crossing JNI for length-as-value would be a regression. * §15 decision log — Phase 1.B.2 landed 2026-05-31 with kernel details, NaN tactics, FFI macro generalization, test counts (96 Rust + 40 PinotNativeAgg + 26 pinot-core integration), and rationale for skipping a per-type MIN/MAX JMH this round. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
|
Update ... Have implemented 3 kernels and have taken observations with Rust Scalar + JNI and Rust + SIMD (Neon on ARM for now) + JNI. Some interesting observations x86 (Intel or AMD) testing is pending. Will be sharing the performance numbers shortly Moving on to implementing vectorized group by with cache friendly hash table format (Swiss) that is type aware and can decide to leverage vector instructions for probing as and when possible + specialized optimizations for single and multi key group by. |
| //! `Math.max(double, double)`. See [`super::super::min::double`] for the | ||
| //! mirror MIN kernel. | ||
|
|
||
| #[inline] |
There was a problem hiding this comment.
Neat! Just as a datapoint you could consider how much the manual simd gains over an auto-vectorizing friendly version: https://godbolt.org/z/aYMbnffxG
There was a problem hiding this comment.
Right. I am measuring this specifically in the beginning itself.
If it's marginal on top of what Rust is already giving, then probably not worth the effort / complexity and we instead just focus on getting to Rust (while ensuring we are using Rust ergonomics with high craft).
There was a problem hiding this comment.
Great, sounds good to me.
One thing that might be a bit less obvious is that since these are compiler directives #[cfg(target_arch = "aarch64")] will only do anything if the targeted cpu supports the options :)
Practically this means:
https://nnethercote.github.io/perf-book/build-configuration.html#cpu-specific-instructions
Wouldn't really recommend target-cpu=native for anything but exploring though (it means target exactly my cpu)
| //! Matches `SumAggregationFunction.aggregateSV(LONG)` Java semantics. See the | ||
| //! parent module docs (`pinot_native_kernels::sum`) for tolerance rules and | ||
| //! a note on lane-reduction reorder vs Java's left-to-right accumulation. | ||
|
|
||
| /// Sums a slice of `i64` as `f64`, dispatching to the fastest available | ||
| /// implementation for the current host CPU. | ||
| #[inline] | ||
| pub fn sum_i64_to_f64(values: &[i64]) -> f64 { | ||
| #[cfg(target_arch = "x86_64")] | ||
| { | ||
| if std::is_x86_feature_detected!("avx512dq") { | ||
| // SAFETY: feature was just detected at runtime. | ||
| return unsafe { sum_i64_to_f64_avx512(values) }; | ||
| } | ||
| if std::is_x86_feature_detected!("avx2") { | ||
| // SAFETY: feature was just detected at runtime. | ||
| return unsafe { sum_i64_to_f64_avx2(values) }; | ||
| } | ||
| } | ||
|
|
||
| #[cfg(target_arch = "aarch64")] | ||
| { | ||
| if std::arch::is_aarch64_feature_detected!("neon") { | ||
| // SAFETY: feature was just detected at runtime. | ||
| return unsafe { sum_i64_to_f64_neon(values) }; | ||
| } | ||
| } | ||
|
|
||
| sum_i64_to_f64_scalar(values) | ||
| } |
There was a problem hiding this comment.
Okay really interesting.
Focusing just on neon -- the simd version is about twice as fast as scalar, due to accumulation + fp addition...
Benchmarking shows this is basically exactly the same as the simd version:
fn simple_sum(values: &[i64]) -> f64 {
let mut acc = [0.0_f64; 8];
let chunks = values.chunks_exact(8);
let rem: f64 = chunks.remainder().iter().map(|&v| v as f64).sum();
for c in chunks {
for j in 0..8 {
acc[j] += c[j] as f64;
}
}
(acc[0]+acc[1]) + (acc[2]+acc[3]) + (acc[4]+acc[5]) + (acc[6]+acc[7])
}AVX512: https://godbolt.org/z/aWxW3asG4
I tried to get godbolt to be happy with the neon targets but it's being a bit annoying...
Do we have formal rules about i64 overflowing sums? (I assume yes, which is why we are using f64 in the first place?), otherwise we would of course do:
pub fn sum_i64_to_f64(values: &[i64]) -> f64 {
values.iter().copied().sum::<i64>() as f64
}There was a problem hiding this comment.
(totally understand and am on board w/ goals of getting something that we can see real results from -- just want to try and limit the amount of hand crafted simd / unsafe code that we need to maintain)
Auto-vectorization is also fragile but only in the sense that you may see performance differences
| /// Sums a slice of `i32` as `f64`, dispatching to the fastest available | ||
| /// implementation for the current host CPU. | ||
| #[inline] |
There was a problem hiding this comment.
I might humbly suggest:
pub fn sum_i32_to_f64(values: &[i32]) -> f64 {
values.iter().copied().map(|v| v as i64).sum::<i64>() as f64
}Widening to i64 basically makes the overflow practically impossible.
This should also be pretty substantially faster than even the manually written simd version:
https://godbolt.org/z/5TzEPW5zP
|
Initial Performance Numbers on Neon. x86 numbers are pending
Read further to understand the details SUM Kernel speedup is uniform across types (5.6–6.1×)
MIN/MAX swings widely (1.0× to 12×)
Native Function
Current Java Baseline
Currently the memory management between JVM -> JNI -> Rust is not yet optimized. Thought of an approach and noted it in the design doc but yet to work on it. Prioritizing getting end to end working and testing a real harness on x86 machine. Used SIMD as well and measured it's benefit separately
Further verified through disassembly
JNI numbers - Want to measure this more but here is an initial glimpse
Native Rust Scalar vs Java scalar gap comes from explicit 4-way unrolling with four independent accumulators (good instruction level parallelism), no virtual-method. Java's hot SUM loop is JIT-friendly but accumulates into a single double, limiting ILP. |
dinoocch
left a comment
There was a problem hiding this comment.
floats and doubles would guess are the places the simd wins due to NaN? Would have to spend more time thinking about the autovectorizing / how to structure it to make it obvious :(
| #[inline] | ||
| pub fn max_i32_to_f64(values: &[i32]) -> f64 { | ||
| if values.is_empty() { | ||
| return f64::NEG_INFINITY; | ||
| } | ||
| #[cfg(target_arch = "x86_64")] | ||
| { | ||
| if std::is_x86_feature_detected!("avx512f") { | ||
| return unsafe { max_i32_to_f64_avx512(values) }; | ||
| } | ||
| if std::is_x86_feature_detected!("avx2") { | ||
| return unsafe { max_i32_to_f64_avx2(values) }; | ||
| } | ||
| } | ||
|
|
||
| #[cfg(target_arch = "aarch64")] | ||
| { | ||
| if std::arch::is_aarch64_feature_detected!("neon") { | ||
| return unsafe { max_i32_to_f64_neon(values) }; | ||
| } | ||
| } | ||
|
|
||
| max_i32_to_f64_scalar(values) | ||
| } |
There was a problem hiding this comment.
Similarly, try:
#[inline(never)]
pub fn max_i32_to_f64(values: &[i32]) -> f64 {
if values.is_empty() { return f64::NEG_INFINITY; }
values.iter().copied().fold(i32::MIN, i32::max) as f64
}| #[inline] | ||
| pub fn max_i64_to_f64(values: &[i64]) -> f64 { | ||
| if values.is_empty() { | ||
| return f64::NEG_INFINITY; | ||
| } | ||
| #[cfg(target_arch = "x86_64")] | ||
| { | ||
| if std::is_x86_feature_detected!("avx512f") { | ||
| return unsafe { max_i64_to_f64_avx512(values) }; | ||
| } | ||
| if std::is_x86_feature_detected!("avx2") { | ||
| return unsafe { max_i64_to_f64_avx2(values) }; | ||
| } | ||
| } | ||
|
|
||
| #[cfg(target_arch = "aarch64")] | ||
| { | ||
| if std::arch::is_aarch64_feature_detected!("neon") { | ||
| return unsafe { max_i64_to_f64_neon(values) }; | ||
| } | ||
| } | ||
|
|
||
| max_i64_to_f64_scalar(values) | ||
| } |
There was a problem hiding this comment.
https://godbolt.org/z/Mfz1Ksx5f
pub fn max_i64_to_f64(values: &[i64]) -> f64 {
if values.is_empty() { return f64::NEG_INFINITY; }
values.iter().copied().fold(i64::MIN, i64::max) as f64
}| #[inline] | ||
| pub fn min_i32_to_f64(values: &[i32]) -> f64 { | ||
| if values.is_empty() { | ||
| return f64::INFINITY; | ||
| } | ||
| #[cfg(target_arch = "x86_64")] | ||
| { | ||
| if std::is_x86_feature_detected!("avx512f") { | ||
| return unsafe { min_i32_to_f64_avx512(values) }; | ||
| } | ||
| if std::is_x86_feature_detected!("avx2") { | ||
| return unsafe { min_i32_to_f64_avx2(values) }; | ||
| } | ||
| } | ||
|
|
||
| #[cfg(target_arch = "aarch64")] | ||
| { | ||
| if std::arch::is_aarch64_feature_detected!("neon") { | ||
| return unsafe { min_i32_to_f64_neon(values) }; | ||
| } | ||
| } | ||
|
|
||
| min_i32_to_f64_scalar(values) | ||
| } |
There was a problem hiding this comment.
#[inline(never)]
pub fn min_i32_to_f64(values: &[i32]) -> f64 {
if values.is_empty() { return f64::INFINITY; }
values.iter().copied().fold(i32::MAX, i32::min) as f64
}| pub fn min_i64_to_f64(values: &[i64]) -> f64 { | ||
| if values.is_empty() { | ||
| return f64::INFINITY; | ||
| } | ||
| #[cfg(target_arch = "x86_64")] | ||
| { | ||
| if std::is_x86_feature_detected!("avx512f") { | ||
| return unsafe { min_i64_to_f64_avx512(values) }; | ||
| } | ||
| if std::is_x86_feature_detected!("avx2") { | ||
| return unsafe { min_i64_to_f64_avx2(values) }; | ||
| } | ||
| } | ||
|
|
||
| #[cfg(target_arch = "aarch64")] | ||
| { | ||
| if std::arch::is_aarch64_feature_detected!("neon") { | ||
| return unsafe { min_i64_to_f64_neon(values) }; | ||
| } | ||
| } | ||
|
|
||
| min_i64_to_f64_scalar(values) | ||
| } |
There was a problem hiding this comment.
https://godbolt.org/z/7KaqTjfMz
pub fn min_i64_to_f64(values: &[i64]) -> f64 {
if values.is_empty() { return f64::INFINITY; }
values.iter().copied().fold(i64::MAX, i64::min) as f64
}There was a problem hiding this comment.
@dinoocch - please excuse my Rust. I am also using this to get better at Rust so both hand-writing some code + AI based and eyeballing everything. This is why taking a bit longer. So idiomatically, it's not great yet. But will get there.
In any case, my goal is to not check-in this PR since reviewing 10K+ lines of Rust code in a single shot will be crazy. But please give feedback whatever you can. Once we get this to a point where data is convincing (may be another week or so), let's decide a path on how to break this down for shipping in a way that makes sense.
There was a problem hiding this comment.
Yes, definitely understand. Just want to demonstrate that auto-vectorization in rust is very good / powerful so long as the compiler can make assumptions about the code :)
I have found that occasionally it seems like you need to "convince" the compiler that it's worth it by chunking...usually o3 will get the picture with some iterations.
Godbolt, especially with llvm-mca helps a ton
There was a problem hiding this comment.
Makes sense. I will leverage the snippets you posted in the benchmark.
|
For GROUP BY, I am leveraging hashBrown (Rust port of Google's Swiss HashTable) This is same as DataFusion. Adding minor customizations. Did some research / comparison with Clickhouse and DuckDB. They are using highly specialized stuff and multiple type specific flavors. Not taking that direction yet until get data points on end to end real GROUP BY workload. |
How does the memory management work with this + jni? Does rust own/allocate the memory and exposes functions to java? What is controlling the lifetime? I'm totally ignorant for how most jni works (ffm makes a bit more sense with the memory arenas?)? |
|
Learning Notes for posterity - Java implementation's double whammy for MIN/MAX for INT, FLOAT and DOUBLE Point 1 - Widening conversion to f64
The Native Code skips this entirely. The kernel takes the raw int[] (or float[]) over JNI, stays in source type the whole loop, and only widens the single final result to f64 at the return boundary. We pay 1 conversion total, not N. Point 2 - Math.min(double, double) blocks JIT auto- vectorization Java's
These semantics don't match the raw hardware FP min instructions:
So when the JIT sees this loop, it can't make a deterministic decision on what to do. It cannot legally lower it to a SIMD min instruction in one shot — the hardware min has different NaN behavior than Java spec. To vectorize correctly, the JIT would have to emit:
That's 3 SIMD ops per element instead of 1. The JIT decides it's not worth the complexity and falls back to a scalar Math.min call per element. Net result - Java's MIN() inner loop runs at ~1 element per cycle (scalar), not 2-4 per cycle. Native Kernel handles this carefully:
|
Phase 1.D-core foundation (Tasks #38–#43, #56–#58, #47): New pinot-native-groupby Rust crate housing the per-segment GROUP BY data structures. Path C architecture — prototype BOTH our custom SwissTable AND a hashbrown wrapper, decide at end-to-end measurement. Custom SwissTable (table.rs, ctrl.rs): - Open addressing, 16-slot groups, triangular probing within group - 1-byte ctrl array, 2-state machine (EMPTY/FULL, no DELETED — Pinot GROUP BY never removes, frees up h2 bit) - SIMD ctrl-byte probe: NEON (vceqq_u8 + movemask), SSE2 (_mm_cmpeq_epi8 + pmovmskb), AVX-512BW (_mm_cmpeq_epi8_mask returns __mmask16 directly — not in hashbrown today) - Runtime ISA dispatch matching the existing kernels module - Pre-sized layout via with_capacity(N) — 3.5× win at 1M cardinality vs hashbrown's incremental realloc strategy (measured in microbench) - Batch-probe API (probe_or_insert_batch) with reused hash scratch Type-specialized wyhash (hash.rs): - Vendored wyhash final3 family inline (~190 LOC, no crate dep) - HashKey trait monomorphizes hash into the probe loop - Primitive impls: i32/i64/u32/u64 use 3-multiply finalizer - f32/f64 impls canonicalize NaN + ±0 for GROUP BY parity - &[u8] impl for variable-length keys (Phase 1.D-strings prep) HashbrownTable wrapper (hashbrown_table.rs): - Wraps hashbrown::hash_table::HashTable<(u32, u64)> following DataFusion's pattern verbatim (slot stores (group_id, hash), key in separate Vec<K> indexed by group_id) - Same public API as Table<K> for backend interchangeability - Microbench confirms wrapper adds essentially zero overhead vs raw hashbrown (0.93×–1.12× across cardinality regimes) GroupByBackend trait (backend.rs): - Common API surface; both Table and HashbrownTable implement it - Segment driver (Task #47) is generic over the backend so the Task #59 JMH harness picks via type parameter — no runtime branch Segment driver (segment_driver.rs, Task #47 Phase 1.D-core-D): - GroupBySumLongByDictInt<B: GroupByBackend<i32>>: per-segment SUM(longCol) GROUP BY dictEncodedIntCol accumulator - process_block(dict_ids, values) does batch-probe → walks the batch to allocate keys for new group_ids → fused agg-state update loop (the inlined Vec<i64> indexed by group_id that hashbrown's entry() API cannot expose) - Driver-owned Vec<i32> keys (DataFusion pattern) avoids needing a backend iterator and matches Phase 1.D-core-I segment-end materialization shape JNI surface (ffi/src/lib.rs): - Stateful handle-based API: createSwiss / createHashbrown(capacityHint) → jlong processBlock(handle, dictIds[], values[], n) numGroups(handle) → int extractKeys(handle, out[]) / extractSums(handle, out[]) backendTag(handle) → int (for Task #59 JMH _backend axis label) destroy(handle) - BoxedDriver enum-tagged wrapper dispatches per-backend - processBlock uses two sequential critical-pin scopes (Vec<i32> + Vec<i64> scratch) — works around jni-rs requiring &mut env for each critical pin, with zero per-call allocation in steady state - 8 MIN/MAX scalar JNI entries (minIntScalar..maxDoubleScalar) bundled here for the Phase 1.B.2 follow-up JMH harness landing in the next commit Java integration (PinotNativeGroupBy.java): - Thin wrapper over the JNI surface, library loading piggybacks on PinotNativeAgg's static init - PinotNativeAgg.java extended with 8 MIN/MAX scalar native decls matching the new FFI entries Differential test (PinotNativeGroupByTest.java, 21 tests, all pass): - Lifecycle (create/destroy/zero-handle safety) - Single-block correctness (empty / single / mixed dict_ids) - Multi-block accumulation (3 blocks introducing groups incrementally) - 100K-row randomized differential vs LinkedHashMap reference (both single-block + 10-block-split variants) - Cross-backend parity (50K rows × 1K cardinality, byte-identical keys and sums) - Edge cases (all-distinct dense keys, i64 wrapping) - @dataProvider runs every test under both swisstable + hashbrown Microbench (tests/microbench.rs): - 4 backends × 6 cardinality regimes + h2-collision stress = 28 cells - Runs in 6.76s total (NOT a JMH-style 50-min disaster) - Marked #[ignore] so default cargo test skips it - Results documented in design doc §18.2 + §18.2.1 Design doc (docs/native/phase-1-design.md): - §7 expanded to full type × encoding × multiplicity × column-count scope tree (§7.0.1–§7.6) - §11.B.2 per-type MIN/MAX three-way attribution matrix - §15 decision log: 7 new entries covering 1.B.3/1.B.4 deferral, MIN/MAX scalar variants kept post-measurement, Phase 1.D scope broadening with combine path explicit, type-specialized wyhash lock, from-scratch SwissTable lock, Path C decision - §17 appendix: Phase 1.D scoping discussion (verbatim Q&A from 2026-06-01, including the fabricated-statistic correction) - §18 appendix: hash-table research and Path C decision — verbatim cross-system analysis with file paths + code excerpts from DataFusion, Polars, DuckDB, ClickHouse local clones Test summary at this commit: - 96 Rust unit tests pass (kernels) + 74 new Rust unit tests pass (groupby crate) + 21 Java integration tests pass - The previous PinotNativeAggTest count (40) is preserved by PinotNativeAgg.java's MIN/MAX scalar decl additions; runtime surfaces all 8 new symbols via nm Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Uses the minIntScalar..maxDoubleScalar JNI entries added in the previous
commit to give MIN/MAX the same three-way (Java / Rust scalar /
Rust+SIMD) attribution shape we have for SUM.
- BenchmarkNativeMinMaxAggregation: _engine × _op × _type × _length
matrix (3 engines × 2 ops × 4 types × 4 lengths = 96 cells).
JMH @State, @fork(3), 10 warmup + 20 measurement iterations
× 200 ms — same harness shape as BenchmarkNativeSumAggregation.
- NativeScalarMin/MaxAggregationFunction: benchmark-only wrappers
routing to the *Scalar JNI entries, mirroring
NativeScalarSumAggregationFunction's structure. Live in pinot-perf,
not pinot-core, so they don't widen the production API surface.
Headline measured (NEON Apple Silicon, production block size 10K):
Op INT LONG FLOAT DOUBLE
SUM 5.92× 6.09× 5.58× 5.87× (uniform ~6×, all clear ≥4×)
MIN 11.83× 1.00× 11.53× 6.52×
MAX 11.89× 0.99× 11.72× 6.67×
The MIN/MAX matrix splits into three regimes (full analysis preserved
in design doc §11.B.2):
- INT/FLOAT win ~12×: Pinot's Java path widens to f64 via
getDoubleValuesSV() AND runs scalar Math.min/max(double, double)
that JIT can't autovectorize (NaN semantics don't lower to a single
vector min on x86). Native skips both costs.
- DOUBLE wins ~6.5×: same story without the conversion. SIMD lane
count drops from 4 (f32) to 2 (f64) — half the throughput, half
the speedup.
- LONG is parity (~1×): Math.min(long, long) is autovectorizable by
JIT (associative + branchless CMOV + no NaN equivalent). NEON has
no vminq_s64 — synthesized via vcgtq_s64 + vbslq_s64. Both effects
compound. AVX-512F should restore the LONG win on x86 (pending
cloud-harness verification).
Run time: 54 min on the local NEON harness (publication-grade JMH
defaults). Flagged in §11.B.2 as harness over-provisioning to address
in future iterations.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
…admap §19/§20 Snapshot taken before laptop re-image. Mix of in-progress code and design-doc updates; will be cleaned up post-resume. Pushing to preserve state. Rust core (complete, all 116 tests pass via cargo test): * agg.rs — AggKind + AggState + Java-NaN-semantics FP min/max helpers (13 tests) * driver_multi_agg.rs — GroupByDriverDictInt<B> with two-phase block protocol (process_block_keys then apply_long/int/double/float/count); 29 tests covering all 11 AggKind variants, FP NaN propagation, multi-block accumulation, both Path C backends. FFI (cargo build clean, not yet validated via mvnw): * lib.rs — migrated BoxedDriver from GroupBySumLongByDictInt to GroupByDriverDictInt; legacy entry points (createSwiss/processBlock/extractSums) now thin shims internally using a single pre-declared SumLong agg. Added new entries: createSwiss/HashbrownMultiAgg, numAggs, aggKindAt, processBlockKeys, applyAggLong/Int/Double/Float/Count, extractAggLong/Int/Double/Float. Java wrapper: * PinotNativeGroupBy.java — added NativeAggKind enum (must match Rust AggKind u8 encoding) and native declarations for every new JNI entry point. Design doc: * §19.0 Resume-from-here pointer with exact in-progress state. * §20 Autovec verification appendix + corrected 3-way attribution. Key finding: our "scalar Rust" SUM kernel is SLP-vectorized (NOT scalar) by both LLVM and HotSpot. End-to-end speedup numbers unaffected; only the per-column decomposition wording needs updating. FP MIN/MAX is now the cleanest 3-way attribution. * §15 decision log entries for 2026-06-04 (roadmap restructure) and 2026-06-08 (attribution correction). Verification artifacts (not part of production build): * pinot-native/verification/asm_probe — Rust crate for LLVM asm inspection. * pinot-native/verification/jit_probe — Java probe for HotSpot timing inference. * Re-run on x86 at plan step (10) to confirm same qualitative pattern. Pending inside plan step (1a) to resume from: 1. Build native lib via mvnw + run existing PinotNativeGroupByTest (21 tests). 2. Add Java multi-agg differential tests. 3. Mark Task #60 complete. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
@dinoocch - it's likely that latest push is very unclean. I was forced to push a lot from local on Monday (while I was iterating on it) as LinkedIn asked to re-image the laptop on Monday. So I backed up everything in rush But yea feel free to leave comments... just want you to be aware ^^ |
Pinot's MemberName rule requires non-static fields match ^_[a-z][a-zA-Z0-9]*$. The NativeAggKind enum (added in the plan step 1a multi-agg FFI migration) used a bare `ordinalByte` field, failing the pinot-native checkstyle gate before surefire could run. Renamed to `_ordinalByte`; all 5 references are confined to the enum. No behavior change. Unblocks the §19.0 Pending #1 gate (PinotNativeGroupByTest, 21 tests, both Path C backends) which now passes. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Pinot's SumAggregationFunction.aggregateGroupBySV reads every numeric column via getDoubleValuesSV() and accumulates in a double holder — there is no long-precise group-by SUM. The 1a multi-agg driver's SumLong accumulates in i64 with wrapping, which would diverge from Pinot for any sum > 2^53. Add typed-input / f64-accumulator agg kinds so the segment executor (step 1b) can route SUM with exact Java parity AND without forcing a per-block double[] widening/allocation on the Java side (the same approach as the non-grouped sum_i64_to_f64 kernel): AggKind::SumIntToDouble = 11 (i32 input -> f64 accumulator) AggKind::SumLongToDouble = 12 (i64 input -> f64 accumulator) AggKind::SumFloatToDouble = 13 (f32 input -> f64 accumulator) SumDouble (1) already covers DOUBLE columns. SumLong (i64-wrapping) is retained for the legacy single-SUM shim and a possible future long-precise/combine mode; it is not used by the Pinot-parity segment executor. MIN/MAX stay typed (lossless when viewed as double); COUNT stays i64. - agg.rs: new AggKind + AggState variants, predicates, identity, double accessor - driver_multi_agg.rs: apply_int/long/float arms widen to f64; +4 tests - PinotNativeGroupBy.java: matching NativeAggKind ordinals 11-13 - design doc: §15 decision-log entry (2026-06-19) + §19.0 resume pointer 120 Rust tests pass; legacy 21-test Java gate still green (both backends). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Record that segment-native GROUP BY alone is not enough — the cross-segment server combine (step 10b, Task #54) must be native too, or Amdahl caps the win and a Java combine forces a per-segment JNI round-trip. Decisions (§15 entry 2026-06-19, §19.0, step 10b): - JVM/native boundary at the server->broker DataTable handoff, not per-segment. - Segment->combine seam = B1: opaque native handle, zero-copy, in-process. - Handoff carries raw key values (Task #52 materialization), not dict_ids. - The Java-holder drain in NativeGroupByExecutor is a temporary bridge to the existing Java combine; the segment executor's real product is the native raw-key partial output. Native combine is co-designed with 1b, not deferred. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Route eligible segment-level GROUP BY through the Rust multi-agg driver (PinotNativeGroupBy) instead of DefaultGroupByExecutor. Per block the executor probes the dict-id key column once (processBlockKeys) then applies each aggregation's value column (applyAgg*); on getResult() the native per-group results are drained into standard double GroupByResultHolders and the keys are exposed via a NativeGroupKeyGenerator, so the existing Java combine path is unchanged (the temporary "bridge" until native combine, Task #54, lands). New pinot-core classes: - NativeGroupByExecutor — implements GroupByExecutor; owns the native handle, maps each agg to a native AggKind (SUM->Sum*ToDouble, MIN/MAX->typed, COUNT), drains results, frees the handle eagerly + via a Cleaner safety net. - NativeGroupKeyGenerator — decodes native group_id -> key via the column Dictionary (getInternal), mirroring the single-column dict path. Type-agnostic decode means dict-encoded keys of any fixed-width type work (covers 1c). - NativeGroupByRouter — eligibility gate behind flag pinot.native.groupby.enabled (single dict-encoded SV fixed-width key; SUM/MIN/MAX/COUNT over SV fixed-width columns; null handling off; no in-segment trim). - GroupByOperator hook routes to the native executor when eligible. Verification: NativeGroupByQueriesTest runs SUM/MIN/MAX over INT/LONG/FLOAT/ DOUBLE + COUNT GROUP BY a dict INT key over two segments, native-on vs Java-off, and asserts identical broker responses. Passes. (Also fixes a pre-existing checkstyle blank-line nit in NativeSumAggregationFunctionTest.) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Captures everything from today's session: step 1a f64-SUM extension, step 1b segment GROUP BY wiring + verification, the JMH backend sweep (low-cardinality win decaying to a ~2x high-cardinality regression), JNI-boundary dismissal (~42us/9.5ms), the scatter-vs-reduction / dict-direct / Amdahl / memory-bound analysis, SIMD-probe clarification, hashbrown/std/custom-SwissTable + DataFusion wrapper notes, the Pattern A->B redesign, the prioritized SOTA levers, and the original-vs-revised 15-step table pulling native server combine forward. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
- BenchmarkNativeGroupBy (pinot-perf): segment-level GROUP BY, java vs native-swiss vs native-hashbrown, across a cardinality sweep. Single segment, getBrokerResponse path. Produced the §22.2 numbers. - NativeGroupByExecutor: add pinot.native.groupby.backend = swiss|hashbrown toggle so the JMH can sweep both Path C backends through the real operator. - verification/: add missing Apache license headers (asm_probe, jit_probe, README) so the repo passes `mvn install` (apache-rat + license-maven-plugin). - BenchmarkAdaptiveServerSelection: drop 3 unused imports (pre-existing checkstyle violations blocking pinot-perf compile). Note: pinot-perf has two pre-existing files stale vs current li-pinot APIs (BenchmarkWorkloadBudgetManager, BenchmarkDimensionTableOverhead) that must be fixed or moved aside to build the benchmark jar. See design §22.11. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Replace the condensed version with the actual chat responses as-is, in order: - Tokio Q (no - async I/O vs CPU-bound merge; who-owns-threads framing, initial JVM-owns lean) - threading-redesign correction (Rust owns parallelism; morsel-driven, work-stealing, radix-partitioned two-phase aggregation; one mechanism gives dynamic DOP + lock-free parallel merge + cache-resident partitions; Rayon-first) - raw-value-keyed combine table (generic over real key types, always hashes, cache hash beside key, pulls Task #45/#53 forward; Java combine also hashes raw values -> combine is a more favorable battleground than segment) Preserved verbatim per request so no nuance is lost to next session. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Instrument NativeGroupByExecutor with -Dpinot.native.groupby.profile timers that split segment-operator time into keyProbe / aggApply / drain and log a breakdown on materialize(). Zero overhead when off (gated static final). Plus NativeGroupByProfileTest, a non-CI harness that runs the native query at contrasting cardinalities (1K / 100K / 1M) and prints the full-query time so the broker-reduce residual can be backed out. Finding (refutes the yesterday "drain is the culprit" hypothesis): the per-group drain is only 1-5% of segment time (<2ms even at 632K groups). The dominant high-cardinality cost is the result/reduce/serialize RESIDUAL outside the executor (~87% at 632K groups), which scales with group count, not rows. See the next design-doc update. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The SOTA combine engine (design §22.8.1 / §23), built correctness-first then
parallelized. Pure Rust, no JNI yet.
- agg.rs: AggState::merge_slot (cross-segment merge of partial accumulators:
SUM/COUNT add, MIN/MAX min/max, Java-NaN FP) + AggState::append (concat
disjoint radix partitions). All associative+commutative -> safe to merge
partitions in any order on any worker.
- combine.rs: CombineDriver<K, B> — single-threaded merge core, raw-value keyed
(dict ids are segment-local; combine hashes raw values), generic over key
type K (i32/i64/f32/f64 via HashKey) and Path C backend B. merge_partials /
merge_one / merge_driver / extract. SegmentPartial<K> input type.
- combine_parallel.rs: two-phase radix-partitioned, work-stealing parallel merge
via Rayon. Phase 1 partitions (segment,group) indices by key-hash top bits;
Phase 2 merges each partition in its own driver (disjoint by hash -> lock-free,
no atomics), work-stealing across partitions; concat the disjoint partitions.
default_radix_bits over-subscribes the pool (~4 partitions/worker) for skew
balancing + cache-resident partitions.
11 new tests (131 total): single-threaded + parallel both match a HashMap
reference and each other, across radix_bits {0,1,4,8,10} and both backends.
rayon added to the groupby crate (CPU-bound merge -> rayon, not tokio). FFI builds.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The SOTA structure for dict-encoded GROUP BY keys (§22.9 lever 2 / §23): a dict-encoded column gives dense dict_ids in [0, cardinality), so grouping needs NO hash table — map dict_id -> group_id by direct array index, dropping the hash + SIMD control-byte probe entirely (what Java's ArrayBasedHolder / ClickHouse FixedHashMap / DuckDB do). This is why a SwissTable showed little segment win on dict keys (we were hashing ids that can just be indexed). - dict_direct.rs: DictDirectTable implements GroupByBackend<i32> via a dict_id -> dense group_id slot array (grows on demand). Drop-in third backend, so the multi-agg driver + combine compose over it UNCHANGED — only key->group_id changes from hash+probe to one array load. Dense group_ids keep accumulators compact (cache-friendly when a filter touches a dict subset). Type-uniform: INT/LONG/FLOAT/DOUBLE/STRING dict keys are all i32 dict_ids here; type only affects the final decode. - agg.rs: extracted the per-kind apply loops into AggState::apply_*_batch (long/int/double/float/count), now SHARED by the hash-keyed driver and (next) the dict-direct path — no duplication. - driver_multi_agg.rs: apply_* delegate to the AggState batch methods. 135 tests (4 new): driver over DictDirectTable produces results identical to the hash backend (same dense group_ids, same accumulation), without hashing. FFI builds. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Unblocks GROUP BY a, b, c over dict-encoded columns (the prod-test shape). - driver_multi_agg.rs: generalize GroupByDriverDictInt<B: GroupByBackend<i32>> to GroupByDriver<K, B: GroupByBackend<K>> (keys: Vec<K>). Keep GroupByDriverDictInt<B> = GroupByDriver<i32, B> as an alias so the FFI single/ multi-agg surface is untouched. Now i64 packed composite keys (and future CanonicalFP / row-encoded keys) reuse the exact same probe+apply machinery. - multi_key.rs: PackedKeyEncoder packs several dict columns' dict_ids into one i64 when the bit widths sum to <= 64 (ClickHouse keys64 / DuckDB perfect-hash packing). bits_for(cardinality), pack/unpack (exact inverses for decode at the materialization boundary), pack_block (column-major), fits_i32 (small packed range -> can ride the i32 dict-direct path). Returns None when > 64 bits -> caller row-encodes (foundation #4). 141 tests (6 new): pack/unpack roundtrip, width edge cases, >64-bit rejection, and a 2-column dict GROUP BY via packed i64 through the generalized driver matching a tuple-keyed HashMap reference. FFI builds. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
f32/f64 can't be hash keys directly (not Eq: NaN != NaN; raw bits differ for "equal" values). CanonicalF32/F64 are Copy newtypes over the canonicalized bits with total Eq + Hash, so FP columns can be raw GROUP BY keys (combine now, raw-FP segment driver later). Canonicalization: every NaN -> one canonical NaN (all NaN rows group together, matching Java doubleToLongBits + boxed-Double GROUP BY); -0.0/+0.0 -> +0.0 (group together). The exact +/-0.0 grouping is flagged to confirm against Pinot in the step-5 differential test. With this, the combine engine supports all fixed-width key types: INT (i32), LONG (i64), FLOAT (CanonicalF32), DOUBLE (CanonicalF64), single + multi-key (packed i64). STRING (arena + row-encoding) is the remaining combine key type. 145 tests (4 new): NaN/zero canonicalization + a DOUBLE GROUP BY through the combine driver (all NaNs one group, +/-0.0 together). FFI builds. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ild log Records the 2026-06-20 foundation work: the segment-dict-id-uniform / combine-typed key-encoding split, dict-direct + multi-column-packing concepts, the foundation build order, the day's build log (combine parallel engine, attribution profiling that refuted the drain hypothesis, dict-direct backend, multi-key generic driver + packed-key encoder, CanonicalF32/F64), threading status, segment-impact note, and open parity items for step 5. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The SOTA string-key structure (DuckDB string heap / ClickHouse Arena+StringRef / DataFusion GroupValuesBytes): zero per-key allocation. - string_table.rs: StringTable interns each distinct key once into ONE arena (a single growable Vec<u8>), with parallel (offset,len) refs indexed by dense group_id and a hashbrown raw HashTable<(group_id, hash)> keyed by wyhash_bytes, confirming collisions by arena-byte compare. No Vec<u8>/String per key. - StringCombineDriver: cross-segment combine keyed by interned byte strings — mirrors CombineDriver but for non-Copy variable-length keys. Dense group_ids keep the AggState accumulators compact, same as fixed-width. This completes the combine key-type foundation: INT (i32), LONG (i64), FLOAT (CanonicalF32), DOUBLE (CanonicalF64), multi-key (packed i64), and STRING (arena). Row-encoding wide composites reuses StringTable (arbitrary byte keys); the row-encoder is the remaining small piece. Small-string inline is a noted future optimization. 148 tests (3 new): interning (each key once, arena byte-exact), 5K-key stress, and a cross-segment STRING combine matching a HashMap<Vec<u8>> reference. FFI builds. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
… (LONG keys)
The first JNI surface for the native server combine — native combine now runs
end-to-end from Java.
- combine.rs: CombineSession<K,B> — the JNI-facing orchestration (begin_partial /
set_agg_* / commit_partial per segment, then finish(radix_bits) parallel merge,
then result_keys/result_agg). Pure Rust, testable without a JVM.
- agg.rs: AggState::from_{long,int,double,float}_vec — rebuild typed partials from
a segment's typed extraction (inverse of the as_*_slice extract families).
- ffi/lib.rs: PinotNativeGroupByCombine JNI surface for raw i64 keys, monomorphized
per Path C backend (swiss/hashbrown): createLong / beginPartial / setAgg{Long,
Int,Double,Float} / commitPartial / finish / numGroups / extractKeys /
extractAgg{Long,Int,Double,Float} / destroy. Critical-pinned array IO helpers.
- PinotNativeGroupByCombine.java: Java bridge declaring the native surface.
- PinotNativeGroupByCombineTest: feeds 50 synthetic segment partials over JNI,
runs the parallel merge, and verifies SUM/MIN/COUNT GROUP BY against a Java
reference — both backends. Passes.
INT/FLOAT(canonical)/DOUBLE(canonical)/STRING combine keys mirror this surface and
land next; then the Pinot GroupByCombineOperator integration + materialization +
the bounded/cancellable pool. 150 Rust tests, 63 pinot-native Java tests green.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Generalizes the combine JNI surface from the LONG-only first slice to all
grouping-key types, verified end-to-end through JNI.
- ffi: BoxedCombine enum over {LongSwiss, LongHashbrown, DoubleSwiss,
DoubleHashbrown, Strings} with a delegate_all! macro for shared methods;
createCombine(aggKinds, keyType, useHashbrown) + per-type begin/extractKeys.
- groupby: StringCombineSession (parallel radix string combine via StringTable)
+ buffer/offset accessors; lib exports.
- Java bridge PinotNativeGroupByCombine: KEY_TYPE_LONG/DOUBLE/STRING, unified
createCombine + beginPartialLong/Double/String + extractKeysLong/Double/String.
- Test parameterized over {long,double}x{swiss,hashbrown} + string: 40 synthetic
segment partials -> JNI -> parallel merge -> matches Java reference.
Tests run: 5, Failures: 0 (PinotNativeGroupByCombineTest); 96 Rust crate tests green.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…achinery - §24: 2026-06-20 state (all-types FFI verified; multi-type operator + differential test written) + the ClassCastException finding that the IndexedTable re-drain reintroduces the ~87% residual; result-path decision (bypass IndexedTable vs native-serialize) + ORDER BY / feed-seam analysis. - §25: lossless verbatim technical log of the 2026-06-19/20 session (§25.0 Pinot GROUP BY internals findings + 38 Q&A exchanges across 8 themes), preserving the diagrams that §22/§23/§24 had only summarized. - §26: GROUP BY data-limiting machinery (numGroupsLimit 100k / warn 150k / minSegmentGroupTrimSize -1 / minServerGroupTrimSize 5000 / groupTrimThreshold 1M), trim-size/threshold formulas, where each limit bites, accuracy/determinism subtleties, and locked scope (§26.7): ORDER BY on group/agg cols is the main path via native exact top-K; no-ORDER-BY parity; HAVING/post-agg/in-segment-trim deferred. Parity first, accuracy second. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…(verbatim) Verbatim log of the result-path/boundary exchanges: the honest accounting that the WIP IndexedTable drain was never measured (operator never ran green), the two avoidable native<->Java boundaries (feed=B1, result=Option 2), the Pattern A/B refresher, the correction that B1 is decided-but-unbuilt (feed side still marshals through Java holders), and the locked sequencing: boundary 2 first via top-K -> Option-1 parity gate -> native-serialize (byte-exact DataTableImplV4 + round-trip test) -> then boundary 1 (B1). Plus the TOP-K time estimate. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…tep 1)
Adds the result-selection stage the combine needs before serialization: reduce
the fully-merged group set to resultSize (design doc §26/§27). Pure Rust,
unit-tested in isolation; FFI/operator wiring follows.
- topk: OrderRef{Key|Agg(i)} + OrderTerm{ref,asc} + OrderKey trait (i64,
CanonicalF32/F64 ordered by numeric value via total_cmp). compute_selection()
-> permutation: no-ORDER-BY caps to first resultSize (Java's selection there is
itself arbitrary); ORDER BY does an exact top-resultSize (select_nth + sort),
deterministic tie-break on key then index -> at least as accurate as Java's
approximate trim, and stable.
- agg: AggState::cmp_slots (natural total order per accumulator type) +
AggState::gather (reorder groups by a permutation).
- CombineSession::select / StringCombineSession::select apply the permutation in
place (string keys order lexicographically; buffer+offsets rebuilt).
Scope per §26.7: ORDER BY on group-key and/or aggregation columns, any
combination, asc/desc; post-agg exprs + HAVING out of scope (router falls back).
159 groupby-crate tests pass (+9: 6 topk, 2 CombineSession.select, 1 string);
FFI builds; no new clippy warnings.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
|
Few next steps
cc @dinoocch |
dinoocch
left a comment
There was a problem hiding this comment.
flushing some inital thoughts on ffi/lib.rs, but only half done so far
| // specific language governing permissions and limitations | ||
| // under the License. | ||
|
|
||
| //! JNI bindings for Pinot's native aggregation engine. |
There was a problem hiding this comment.
Since we are working on bazelifying pinot stuff I wonder if we can pull directly up to jdk 25 for panama api stabilization. I wouldn't block on this though...
Definitely I think we should consider upgrading to jni 0.22 --
https://github.com/jni-rs/jni-rs/blob/master/crates/jni/docs/0.22-MIGRATION.md
Reading through it seems like they made a lot of substantive improvements...
| crate-type = ["cdylib"] | ||
|
|
||
| [dependencies] | ||
| jni = { version = "0.21", default-features = false } |
| macro_rules! define_reduce_jni { | ||
| ($name:ident, $jarray:ty, $elem:ty, $kernel:path, $empty_value:expr) => { | ||
| #[no_mangle] | ||
| pub extern "system" fn $name( | ||
| mut env: JNIEnv, | ||
| _class: JClass, | ||
| values: $jarray, | ||
| length: jint, | ||
| ) -> jdouble { | ||
| let result = panic::catch_unwind(AssertUnwindSafe(|| -> jdouble { | ||
| if length <= 0 { | ||
| return $empty_value; | ||
| } | ||
| // SAFETY: We hold the critical pin for the duration of the kernel call; | ||
| // no JNI calls are made in between. The slice we cast is valid for the | ||
| // lifetime of `auto`. | ||
| let auto = match unsafe { | ||
| env.get_array_elements_critical(&values, ReleaseMode::NoCopyBack) | ||
| } { | ||
| Ok(a) => a, | ||
| Err(_) => return f64::NAN, | ||
| }; | ||
| let len_usize = length as usize; | ||
| let array_len = auto.len(); | ||
| let effective = if len_usize > array_len { | ||
| array_len | ||
| } else { | ||
| len_usize | ||
| }; | ||
| // SAFETY: auto.as_ptr() points to a contiguous region of `array_len` | ||
| // elements of the underlying primitive type; `effective <= array_len`. | ||
| let slice: &[$elem] = unsafe { | ||
| std::slice::from_raw_parts(auto.as_ptr() as *const $elem, effective) | ||
| }; | ||
| $kernel(slice) | ||
| })); | ||
| match result { | ||
| Ok(v) => v, | ||
| Err(_) => f64::NAN, | ||
| } | ||
| } | ||
| }; | ||
| } | ||
|
|
||
| // SUM(LONG) — production path (runtime-dispatched SIMD). | ||
| define_reduce_jni!( | ||
| Java_org_apache_pinot_nativeengine_agg_PinotNativeAgg_sumLong, | ||
| JLongArray, | ||
| i64, | ||
| sum::long::sum_i64_to_f64, | ||
| 0.0 | ||
| ); |
There was a problem hiding this comment.
We can simplify this macro by separating the jni logic from the naming...
This assumes you move to jni 0.22 which adds JPrimitiveArray::get_elements_critical +
#[inline(always)]
fn reduce_kernel<'local, T>(
unowned: &mut EnvUnowned<'local>,
array: &JPrimitiveArray<'local, T>,
length: jint,
empty: f64,
kernel: impl Fn(&[T]) -> f64,
) -> jdouble
where
T: TypeArray + Copy,
{
if length <= 0 {
return empty;
}
let outcome = unowned.with_env(|env| -> Result<f64, jni::errors::Error> {
// SAFETY: ...
let auto = unsafe { array.get_elements_critical(env, ReleaseMode::NoCopyBack) }?;
let effective = (length as usize).min(auto.len());
// SAFETY: ...
let slice = unsafe { std::slice::from_raw_parts(auto.as_ptr() as *const T, effective) };
Ok(kernel(slice))
});
match outcome.into_outcome() {
Outcome::Ok(v) => v,
Outcome::Err(_) | Outcome::Panic(_) => f64::NAN,
}
}
macro_rules! define_reduce_jni {
($name:ident, $elem:ty, $kernel:path, $empty:expr) => {
#[no_mangle]
pub extern "system" fn $name<'local>(
mut env: JNIEnv<'local>,
_class: JClass<'local>,
values: JPrimitiveArray<'local, $elem>,
length: jint,
) -> jdouble {
reduce_kernel(&mut env, &values, length, $empty, $kernel)
}
};
}
define_reduce_jni!(
Java_org_apache_pinot_nativeengine_agg_PinotNativeAgg_sumLong,
i64,
sum::long::sum_i64_to_f64,
0.0
);The above also means you don't have to make sure that you match the java type and array types
| #[repr(u8)] | ||
| enum BackendTag { | ||
| Swiss = 0, | ||
| Hashbrown = 1, | ||
| } | ||
|
|
||
| /// Boxed driver — either backend, identified by the tag. Java holds a | ||
| /// `jlong` pointer to this struct. | ||
| /// | ||
| /// One scratch buffer per primitive value type. We need them because | ||
| /// `jni-rs`'s `get_array_elements_critical` takes `&mut env`, so we can't | ||
| /// hold two critical pins simultaneously — each input array is copied out | ||
| /// under its own pin scope, then the scratch slices are passed to the | ||
| /// Rust driver. Scratch buffers are reused across all calls on the same | ||
| /// handle, so steady-state allocation is zero. | ||
| struct BoxedDriver { | ||
| tag: BackendTag, | ||
| swiss: Option<GroupByDriverDictInt<Table<i32>>>, | ||
| hashbrown: Option<GroupByDriverDictInt<HashbrownTable<i32>>>, | ||
| dict_id_scratch: Vec<i32>, | ||
| long_scratch: Vec<i64>, | ||
| int_scratch: Vec<i32>, | ||
| double_scratch: Vec<f64>, | ||
| float_scratch: Vec<f32>, | ||
| } |
There was a problem hiding this comment.
I know we are getting rid of the handwritten hash table but, more ergonomically I think you can just do:
enum DriverBackend {
Swiss(GroupByDriverDictInt<Table<i32>>),
Hashbrown(GroupByDriverDictInt<HashbrownTable<i32>>),
}
struct BoxedDriver {
backend: DriverBackend,
dict_id_scratch: Vec<i32>,
long_scratch: Vec<i64>,
int_scratch: Vec<i32>,
double_scratch: Vec<f64>,
float_scratch: Vec<f32>,
}Also as a small nit usually you wouldn't box the value for users / you would let the caller box it themselves.
It's also probably a minor performance note but this seems like it will have a decent amount of indirection -- each Vec is a pointer to some data (plus a length and capacity)
Also we could use a tiny macro to get rid of some of the match arms:
macro_rules! dispatch {
(mut $self:expr, |$d:ident| $body:expr) => {
match &mut $self.backend {
DriverBackend::Swiss($d) => $body,
DriverBackend::Hashbrown($d) => $body,
}
};
($self:expr, |$d:ident| $body:expr) => {
match &$self.backend {
DriverBackend::Swiss($d) => $body,
DriverBackend::Hashbrown($d) => $body,
}
};
}but this is being simplified so it doesn't matter :) just notes for learning
| let copy_n = pin.len().min(n); | ||
| unsafe { | ||
| std::ptr::copy_nonoverlapping( | ||
| pin.as_ptr() as *const i32, |
There was a problem hiding this comment.
(nit) I'm kinda sure you don't need these casts?
| driver.dict_id_scratch.clear(); | ||
| driver.dict_id_scratch.reserve(n); | ||
| // SAFETY: reserve guarantees capacity >= n; we initialize all n | ||
| // elements via copy_nonoverlapping below before reading them. | ||
| unsafe { driver.dict_id_scratch.set_len(n) }; | ||
| driver.long_scratch.clear(); | ||
| driver.long_scratch.reserve(n); | ||
| unsafe { driver.long_scratch.set_len(n) }; |
There was a problem hiding this comment.
ok need to come back here for usre.
| let driver_inner = unsafe { driver_mut(handle) }; | ||
| let dict_slice = driver_inner.dict_id_scratch.as_slice(); | ||
| let value_slice = driver_inner.long_scratch.as_slice(); | ||
| let (dict_ptr, dict_len) = (dict_slice.as_ptr(), dict_slice.len()); | ||
| let (val_ptr, val_len) = (value_slice.as_ptr(), value_slice.len()); | ||
| let dict_s: &[i32] = unsafe { std::slice::from_raw_parts(dict_ptr, dict_len) }; | ||
| let val_s: &[i64] = unsafe { std::slice::from_raw_parts(val_ptr, val_len) }; |
There was a problem hiding this comment.
uh same here, need to come back
| unsafe fn read_long_vec(env: &mut JNIEnv, arr: &JLongArray) -> Vec<i64> { | ||
| match env.get_array_elements_critical(arr, ReleaseMode::NoCopyBack) { | ||
| Ok(pin) => { | ||
| let n = pin.len(); | ||
| let mut v = vec![0i64; n]; | ||
| std::ptr::copy_nonoverlapping(pin.as_ptr() as *const i64, v.as_mut_ptr(), n); | ||
| v | ||
| } | ||
| Err(_) => Vec::new(), | ||
| } | ||
| } |
There was a problem hiding this comment.
Maybe we should use JPrimitiveArray::get_region?
pub fn get_region(
&self,
env: &Env<'_>,
start: jsize,
buf: &mut [T],
) -> Result<()>This is kinda the signature you want all these to have, right? Plus it's safe
| return; | ||
| } | ||
| let v = unsafe { read_long_vec(&mut env, &keys) }; | ||
| unsafe { combine_mut(handle) }.begin_partial_long(v); |
There was a problem hiding this comment.
Hmm since we only need to hold one jni pin, would it make more sense to use get_elements_critical and operate on the slice without coping?
Edit: looks like the call tree below this wants an owned copy of the data which will live across jni calls (?) but I still need to study a bit better
| let keys = unsafe { combine_ref(handle) }.result_keys_long(); | ||
| if let Ok(pin) = unsafe { env.get_array_elements_critical(&out, ReleaseMode::CopyBack) } { | ||
| let n = pin.len().min(keys.len()); | ||
| unsafe { std::ptr::copy_nonoverlapping(keys.as_ptr(), pin.as_ptr() as *mut i64, n) }; |
There was a problem hiding this comment.
We probably can avoid the pin here by using JPrimitiveArray::set_region
It probably has the same performance characteristics I would guess (and still copies memory) but should be cleaner at least?
| #[inline] | ||
| pub fn new(v: f64) -> Self { | ||
| let canon = if v.is_nan() { | ||
| f64::NAN |
There was a problem hiding this comment.
Food for pondering -- it might be worth picking a specific quiet nan?
| //! `f32` / `f64` cannot be hash-table keys directly: they implement `PartialEq` | ||
| //! but not `Eq` (NaN ≠ NaN), and raw bit patterns for "equal" values can | ||
| //! differ (every NaN payload, and ±0.0). [`CanonicalF64`] / [`CanonicalF32`] | ||
| //! are `Copy` newtypes over the **canonicalized bit pattern**, giving total | ||
| //! `Eq` + `Hash` so FP columns can be raw GROUP BY keys (at combine, and at the | ||
| //! future raw-FP segment driver, steps 1d/1e). |
There was a problem hiding this comment.
Consider https://crates.io/crates/ordered-float (we could benchmark to see the comparison)
I'm guessing it should be equivalent in performance (this has higher construction cost, that has higher Hash and Eq)
Overall I'm not too worried about perf with this since this would only be hit for non-dictionary floating point group-by right?
| where | ||
| K: HashKey + Eq + Copy, | ||
| B: GroupByBackend<K>, |
There was a problem hiding this comment.
(nit) Usually you would not include trait bounds on the struct and keep them on only the impl
| staging_keys: Option<Vec<K>>, | ||
| staging_aggs: Vec<Option<AggState>>, |
There was a problem hiding this comment.
Todo, come back...
Things I'm thinking of:
- Is it worthwhile to have something like:
struct Partial {
keys: Vec<K>,
aggs: Vec<Option<AggState>>,
}- I wonder if the Options are useful vs a zero value? (probably they are)
| partials: Vec<SegmentPartial<K>>, | ||
| staging_keys: Option<Vec<K>>, | ||
| staging_aggs: Vec<Option<AggState>>, | ||
| result: Option<(Vec<K>, Vec<AggState>)>, |
There was a problem hiding this comment.
Instead of a tuple does it make sense to just use SegmentPartial
|
|
||
| impl<K, B> CombineSession<K, B> | ||
| where | ||
| K: HashKey + Eq + Copy + Send + Sync, |
There was a problem hiding this comment.
Do you need all of these bounds? Especially Copy is very restrictive?
| for slot in self.staging_aggs.iter_mut() { | ||
| *slot = None; | ||
| } |
There was a problem hiding this comment.
(nit) you can do:
self.staging_aggs.fill_with(|| None);|
|
||
| // --- Phase 2: merge each partition independently (work-stealing). --- | ||
| let drivers: Vec<CombineDriver<K, B>> = buckets | ||
| .into_par_iter() |
There was a problem hiding this comment.
Rayon is probably good for initial perf validations, it might be good to think about an executor abstraction or possibly consider async code (lots of async runtimes in rust can be very good task schedulers)
Probably we also need to at least limit the resource usage?
| for v in out.iter_mut() { | ||
| *v = 0; | ||
| } |
There was a problem hiding this comment.
Tiny nit but I think you could instead do something like:
let first = columns[0];
let first_shift = self.shifts[0];
for i in 0..n {
out[i] = (first[i] as u64 as i64) << first_shift;
}
for c in 1..columns.len() {
let shift = self.shifts[c];
let col = columns[c];
for i in 0..n {
out[i] |= (col[i] as u64 as i64) << shift;
}
}(to skip the memory fill to 0)
| widths: Vec<u32>, | ||
| shifts: Vec<u32>, |
There was a problem hiding this comment.
I wonder if it's better to do something like:
struct PackedField {
width: u32,
shift: u32,
// maybe mask? this will remove the if from unpack...
}
struct PackedKeyEncoder {
fields: Box<[PackedField]>,
}| pub fn pack(&self, dict_ids: &[i32]) -> i64 { | ||
| debug_assert_eq!(dict_ids.len(), self.widths.len()); | ||
| let mut packed: u64 = 0; | ||
| for i in 0..self.widths.len() { |
There was a problem hiding this comment.
I wonder if it's worthwhile to unroll common column lengths to avoid the multiple stores/loads and loop?
ie we might know that the majority of group-by's are less than 5.
| debug_assert_eq!(dict_ids.len(), self.widths.len()); | ||
| let mut packed: u64 = 0; | ||
| for i in 0..self.widths.len() { | ||
| packed |= (dict_ids[i] as u64) << self.shifts[i]; |
There was a problem hiding this comment.
Consider:
for (&id, &shift) in dict_ids.iter().zip(self.shifts.iter()) {
packed |= (id as u64) << shift;
}It's often a good idea to help convince the compiler that bounds checks are not needed
| /// Reused scratch for batch-probe output. Avoids per-block allocation. | ||
| group_id_scratch: Vec<u32>, |
There was a problem hiding this comment.
I suggest we remove this for now unless it's obviously better for performance
There was a problem hiding this comment.
Also one contemplative thought -- if we pick a "standard" batch size of rows (or make it a compile time constant) a lot of these vecs can become fixed size arrays?
| // Phase 1: batch probe-or-insert. Backend allocates new group_ids | ||
| // for previously-unseen keys; existing keys return their original | ||
| // group_id. | ||
| self.group_id_scratch.clear(); |
There was a problem hiding this comment.
I wonder a bit if this scratch buffer actually helps us? Might be interesting to compare with a more naive implementation like:
pub fn process_block(&mut self, dict_ids: &[i32], values: &[i64]) {
assert_eq!(dict_ids.len(), values.len());
for (&dict_id, &value) in dict_ids.iter().zip(values) {
let prev_len = self.table.len();
let gid = self.table.probe_or_insert(dict_id) as usize;
if gid == prev_len {
self.keys.push(dict_id);
self.sums.push(0);
}
self.sums[gid] = self.sums[gid].wrapping_add(value);
}
}| //! Variable-length (STRING / BYTES) GROUP BY keys — arena-backed (Task #53; | ||
| //! design doc §23 foundation step 4b). |
There was a problem hiding this comment.
I wonder a bit about the cost of shuffling string / byte objects across the JNI boundary :/ I guess this would involve at minimum memory copying overhead. Just food for thought
| pub struct StringTable { | ||
| /// `(group_id, cached_hash)` entries; probe by hash, confirm by arena bytes. | ||
| map: HashTable<(u32, u64)>, | ||
| /// All interned key bytes, concatenated. | ||
| arena: Vec<u8>, | ||
| /// `offsets[group_id] = (arena_start, len)`. | ||
| offsets: Vec<(u32, u32)>, | ||
| } |
There was a problem hiding this comment.
It's worth reviewing string interning implementations in the wild ~
https://crates.io/crates/lasso is especially interesting
| //! A future optimization (not yet done): inline small strings (≤ ~15 bytes) | ||
| //! directly in the slot to skip the arena indirection on the hot path | ||
| //! (DuckDB / Umbra "German strings"). |
There was a problem hiding this comment.
Fun! Leaving some notes here for inspiration:
First, for storing small strings it might be worth checking out compact_str. It stores up to 24 byte strings on the stack and has some neat properties. And also some more specialized things like byteview
German strings go a bit further for long strings by encoding the prefix, check out: https://ltungv.com/note/an-optimization-thats-impossible-in-rust/
Summary
Initially I was thinking of speeding up filters but filtering is generally not where most amount of time is spent especially when we have right indexes. Our indexes have proven to be efficient.
Full scans on range queries have proven to be expensive but that's also because we have barely scratched surface with SOTA (state of the art) range index in OLAP.
So the scope I am POCing is an end to end kernel written in Rust (glued via JNI) for GROUP BY and Aggregations with full benefits of vectorization + SIMD + cache aware hash tables formats.
Reasoning -- To reduce cost / QPS (read path specifically), we need to target bottlenecks on the query path -- whether it's within the operator itself, memory management, multi-threading or IO. GROUP BY Aggregations (commonly used by LinkedIn users of Pinot) is a frequent query that typically dominates workload cost. Use of Native opens a lot of opportunities for OLAP (e.g HW acceleration given repeatable nature of operations) in addition to the usual benefits (e.g memory safety) of a native language like Rust. Filtering is also a great candidate for acceleration to target but it's not inefficient for vast majority of workloads. Other candidates are memory mapped IO, column/segment readers, explicit core management etc.
Testing
Initial numbers observed after writing few kernels is upto 6x speed up (Rust + SIMD + JNI) and upto 2.5x speed up (Rust + JNI)
Design
This draft tracks ongoing progress. Strategic and detailed designs live in the diff. I will be converting this into a design doc on google doc to also facilitate reviews. Initially it's in MD file as I was trying to get POC started.
Status