peter-toth commented on code in PR #57742:
URL: https://github.com/apache/spark/pull/57742#discussion_r3755075346
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -4156,6 +4156,46 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val ADAPTIVE_PARTIAL_AGGREGATION_ENABLED =
+
buildConf("spark.sql.execution.aggregate.adaptivePartialAggregation.enabled")
+ .doc("When true, hash aggregation adaptively bypasses the pre-shuffle
partial aggregation " +
+ "at runtime when it observes that the partial aggregation is not
reducing the number of " +
+ "rows enough to be worthwhile. Once bypassed, the remaining input rows
are passed " +
+ "through as single-row partial aggregation buffers for the final
aggregation to merge, " +
+ "which avoids the cost of maintaining and spilling a large aggregation
map with little " +
+ "reduction benefit. This applies only to hash aggregation with
grouping keys.")
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .booleanConf
+ .createWithDefault(true)
Review Comment:
**Finding 16.** Not a blocker, and I want to be straight about how strong
this is. Low-cardinality, no spill, whole-stage codegen, off -> on, best / avg
/ stdev from the files committed here:
| results file | off | on |
| --- | --- | --- |
| `-jdk25-results.txt` | 269 / 275 / 4 | 284 / 294 / 6 |
| `-results.txt` (JDK 17) | 270 / 311 / 37 | 293 / 318 / 22 |
| `-jdk21-results.txt` | 281 / 292 / 7 | 291 / 323 / 52 |
The JDK 17 and 21 runs are too noisy to carry it on their own — the JDK 17
off arm has a stdev of 37 against an avg delta of 7. The JDK 25 run is the
clean one: stdevs of 4 and 6, best +15, avg +19, so 3-5 stdev. All three agree
in sign on best time, and two of the three files report it as `1.0X -> 0.9X`.
What makes me think it isn't just noise is that a mechanism predicts it. Two
costs are paid whether or not anything bypasses: `needStopCheck` becomes
`true`, putting `if (shouldStop()) return;` back into every upstream produce
loop for a plan that never appends mid-build, and `doConsume` gains the per-row
guards, the counter increment and the check-point compare. About 1.4 ns/row
over the 16.8M-row input is the right order for a couple of extra branches per
row.
Two measurements would settle it: re-run the low-cardinality cases with the
iteration count raised so the stdevs come down, and run one arm with
`needStopCheck` hard-coded to `false` to say how much of it is the stop check.
If it's mostly the stop check that's worth knowing, though I don't think it's a
one-liner to avoid — `shouldStopCheckCode` is emitted by the child from
`parent.needStopCheck` at codegen time, so it can't easily be gated on a flag
that only flips at runtime.
Either way, the benchmark comment on this scenario says the two runs "must
match (no regression)", which these numbers don't support.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -535,19 +642,27 @@ case class HashAggregateExec(
// `addNewFunction` spills this helper into a nested class (as can happen
// once the outer class passes the code-size threshold), the bare field
// reference fails with `IllegalAccessError`.
- val doAggFuncName = ctx.addNewFunction(doAgg,
- s"""
- |private void $doAgg(int partitionIndex) throws java.io.IOException {
- | ${child.asInstanceOf[CodegenSupport].produce(ctx, this)}
- | $finishHashMap
- |}
- """.stripMargin)
- // generate code for output
+ // Generate code for output. This must happen before the `doAgg` helper
below, because with
+ // adaptive partial aggregation enabled, `doConsumeWithKeys` (invoked from
the child's produce
+ // inside `doAgg`) emits pass-through rows by calling this output function
directly.
val keyTerm = ctx.freshName("aggKey")
val bufferTerm = ctx.freshName("aggBuffer")
- val outputFunc = generateResultFunction(ctx)
-
+ outputFunc = generateResultFunction(ctx)
Review Comment:
**Finding 17.** This moved `generateResultFunction` ahead of the `doAgg`
registration for every keyed hash aggregate, not only the adaptive ones. Only
the adaptive path needs it, so `doConsumeWithKeys` can call `outputFunc` while
the child's produce is generated inside `doAgg`.
The reorder is not consequence-free when the feature is off:
`generateResultFunction` calls `consume(...)`, so the parent's consume code,
its fresh names and its mutable-state slots are now allocated before the
child's produce rather than after. `ctx.addNewFunction` decides
inline-versus-nested-class from the size accumulated so far, so which helpers
end up in a nested class can change for a large stage — and the comment just
above at `:639-644` exists because a nested-class placement already caused an
`IllegalAccessError` on this method. Nothing is known to break, but there is no
reason for this to reach plans the feature never touches.
```scala
if (adaptivePartialAggEnabled) {
outputFunc = generateResultFunction(ctx)
}
// ... adaptiveOutputMapFuncName, then doAggFuncName ...
if (!adaptivePartialAggEnabled) {
outputFunc = generateResultFunction(ctx)
}
```
The second assignment only has to land before `adaptiveFinalOutput` is built
at `:812`. `keyTerm`/`bufferTerm` are read only by the output loops, so they
can stay where they are.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/AdaptivePartialAggregationSuite.scala:
##########
@@ -0,0 +1,1136 @@
+/*
+ * 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.spark.sql.execution.aggregate
+
+import org.apache.spark.sql.{DataFrame, QueryTest, Row}
+import org.apache.spark.sql.catalyst.expressions.aggregate.Partial
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.functions._
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * Tests for runtime adaptive partial aggregation
+ * (see [[SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_ENABLED]]). When a partial
aggregate is not reducing
+ * rows, the operator stops aggregating and streams the remaining rows through
as single-row partial
+ * buffers for the Final aggregate to merge. It must never change results.
+ *
+ * The suite has two halves:
+ * 1. Correctness: output is identical to the reference (feature-off) run
across the full matrix
+ * of codegen on/off, two-level map on/off, and spill/no-spill, over a
range of aggregate
+ * shapes, key types, and `Expand`-bearing plans (ROLLUP / CUBE /
GROUPING SETS /
+ * multi-distinct).
+ * 2. Triggering: the `numBypassingRows` metric proves the bypass actually
fires when (and only
+ * when) it should -- high-cardinality input bypasses, low-cardinality
input keeps aggregating,
+ * the feature switch and eligibility rules are honored, and both check
points work.
+ */
+class AdaptivePartialAggregationSuite extends QueryTest with SharedSparkSession
+ with AdaptiveSparkPlanHelper {
+
+ import testImplicits._
+
+ // A `testFallbackStartsAt` setting ("fastMapCounter, regularMapCounter")
that makes the regular
+ // map fall back (spill) periodically, exercising the spill-check decision
path in both the
+ // codegen and interpreted aggregation paths. Kept moderate so
low-cardinality inputs (which are
+ // never bypassed and therefore really spill) do not open an unbounded
number of spill readers.
+ private val forceSpillFallback = "4, 16"
+
+ // The upstream `CombineAdjacentAggregation` and `ReplaceHashWithSortAgg`
rules would change the
+ // plan of these small single-partition queries away from a Partial+Final
`HashAggregateExec`:
+ // the former merges the two adjacent phases (no shuffle in between) into a
single `Complete`
+ // aggregate, and the latter converts a hash aggregate to a sort aggregate
when the input is
+ // already sorted by the grouping key (a `Range` over an ascending `id`
key). The adaptive
+ // feature lives in the partial hash aggregation, so both rules are disabled
to keep that
+ // structure in the tests.
+ private val fixedPlanConfs = Seq(
+ SQLConf.COMBINE_ADJACENT_AGGREGATION_ENABLED.key -> "false",
+ SQLConf.REPLACE_HASH_WITH_SORT_AGG_ENABLED.key -> "false")
+
+ /**
+ * Runs `build` with adaptive partial aggregation disabled (the reference)
and then across the
+ * full configuration matrix with it enabled, asserting every enabled run
matches the reference.
+ *
+ * `build` takes the number of input partitions, which the matrix varies
along with everything
+ * else, because the plan shape decides which parts of the feature run at
all. When the two
+ * aggregates end up in one whole-stage -- no `Exchange` between them -- the
partial aggregate's
+ * output feeds the Final's `doConsume` directly and never reaches
+ * `BufferedRowIterator.currentRows`, so `shouldStop()` stays false for the
whole build and
+ * neither `needStopCheck` nor the resumed-build path is exercised.
Splitting them puts the
+ * streamed rows through the output buffer and runs both.
+ *
+ * More than one input partition is necessary but not sufficient for that
split: a `Range` keyed
+ * directly on `id` already reports an output partitioning that satisfies
the Final aggregate's
+ * `ClusteredDistribution`, so `EnsureRequirements` inserts no `Exchange`
however many partitions
+ * it has. Tests that want the split shape group on a derived key (a cast,
say) so the input
+ * partitioning no longer satisfies the requirement.
+ *
+ * `expectBypass` ties the correctness guarantee to the triggering
guarantee: beyond matching the
+ * reference, every cell must either actually stream rows through (when
true) or keep
+ * aggregating (when false). Without it a test could silently stop
exercising pass-through if the
+ * input stopped being bypassable, and only this assertion makes that fail
loudly.
+ */
+ private def checkAdaptiveMatchesReference(
+ build: Int => DataFrame,
+ expectBypass: Boolean = true): Unit = {
+ for {
+ inputPartitions <- Seq(1, 2)
+ wholeStage <- Seq(true, false)
+ twoLevelMap <- Seq(true, false)
+ forceSpill <- Seq(true, false)
+ } {
+ // The reference is built with the same partitioning, so only the
feature differs.
+ val reference = withSQLConf(
+ (SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key -> "false") +:
fixedPlanConfs: _*) {
+ build(inputPartitions).collect().toSeq
+ }
+ val spillConf = if (forceSpill) {
+ Seq("spark.sql.TungstenAggregate.testFallbackStartsAt" ->
forceSpillFallback)
+ } else {
+ Nil
+ }
+ withSQLConf(
+ (Seq(
+ SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key -> "true",
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> wholeStage.toString,
+ SQLConf.ENABLE_TWOLEVEL_AGG_MAP.key -> twoLevelMap.toString,
+ // Small `minRows` so the periodic check runs on modest inputs.
+ SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_MIN_ROWS.key -> "8") ++
+ spillConf ++ fixedPlanConfs): _*) {
+ val msg = s"inputPartitions=$inputPartitions wholeStage=$wholeStage " +
+ s"twoLevelMap=$twoLevelMap forceSpill=$forceSpill"
+ withClue(msg) {
+ // Collect once so the metrics are populated, then check whether the
bypass fired for
+ // this cell. The metric lives on the `Partial`-mode
`HashAggregateExec`, so that is the
+ // operator the assertion reads.
+ val df = build(inputPartitions)
+ df.collect()
+ val skipped = collect(df.queryExecution.executedPlan) {
+ case agg: HashAggregateExec if
agg.aggregateExpressions.forall(_.mode == Partial) =>
+ agg.metrics.get("numBypassingRows").map(_.value).getOrElse(0L)
+ }.sum
+ if (expectBypass) {
+ assert(skipped > 0,
+ s"expected rows to bypass partial aggregation, got $skipped
bypassed rows")
+ } else {
+ assert(skipped == 0,
+ s"expected no rows to bypass partial aggregation, got $skipped
bypassed rows")
+ }
+ checkAnswer(df, reference)
+ }
+ }
+ }
+ }
+
+ /**
+ * The observable per-run counters we assert on, all read from the partial
`HashAggregateExec` in
+ * a single execution so the metrics are not double-counted:
+ * - `skipped`: our self-reported `numBypassingRows` metric.
+ * - `partialOutputRows`: the partial aggregate's own `numOutputRows`. An
independent,
+ * pre-existing counter driven by the normal output path, so it is the
ground truth for
+ * whether rows were streamed through -- it equals the distinct key
count when aggregation is
+ * effective and climbs toward the input row count once the operator
bypasses.
+ * - `spillBytes`: the partial aggregate's `spillSize`. Reliable only when
no fallback is
+ * forced: on the interpreted path this is derived from the
task-cumulative memory-spill
+ * counter, so a forced fallback (or downstream shuffle-write spill) can
inflate it.
+ * Asserted only by the periodic check test, which forces no fallback;
use
+ * `tasksFallBacked` otherwise.
+ * - `tasksFallBacked`: the partial aggregate's `numTasksFallBacked`,
incremented only when the
+ * regular map actually falls back into sort-based aggregation. When the
spill check bypasses
+ * at the spill boundary the sorter is never created, so this stays 0 --
direct, per-operator
+ * evidence the bypass replaced the sort fallback.
+ */
+ private case class AggCounters(
+ skipped: Long,
+ partialOutputRows: Long,
+ spillBytes: Long,
+ tasksFallBacked: Long)
+
+ // Verifies `df` (an already-collected bypassing run) produces the same
results as the feature-off
+ // reference. `build` is re-run for the reference so it gets a genuinely
non-adaptive plan rather
+ // than reusing the bypassing run's cached one.
+ private def checkAgainstReference(df: DataFrame, build: () => DataFrame):
Unit = {
+ val reference = withSQLConf(
+ SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key -> "false") {
+ build().collect().toSeq
+ }
+ checkAnswer(df, reference)
+ }
+
+ private def runAndReadCounters(build: () => DataFrame): AggCounters = {
+ // The triggering tests assert on metrics, so also verify the bypassing
run produces the same
+ // results as the feature-off reference.
+ val df = build()
+ df.collect()
+ val partialAggs = collect(df.queryExecution.executedPlan) {
+ case agg: HashAggregateExec if agg.aggregateExpressions.forall(_.mode ==
Partial) => agg
+ }
+ // A partial aggregate is always present for the grouped queries these
tests use.
+ assert(partialAggs.nonEmpty, "expected a partial HashAggregateExec in the
plan")
+ val counters = AggCounters(
+ // The metric is only registered on aggregates the feature applies to;
an aggregate without
+ // it bypassed nothing.
+ skipped =
partialAggs.map(_.metrics.get("numBypassingRows").map(_.value).getOrElse(0L)).sum,
+ partialOutputRows =
partialAggs.map(_.metrics("numOutputRows").value).sum,
+ spillBytes = partialAggs.map(_.metrics("spillSize").value).sum,
+ tasksFallBacked =
partialAggs.map(_.metrics("numTasksFallBacked").value).sum)
+ checkAgainstReference(df, build)
+ counters
+ }
+
+ private def numBypassingRows(build: () => DataFrame): Long =
runAndReadCounters(build).skipped
+
+ // Returns the bypassed-row count per Partial-mode `HashAggregateExec`,
keyed by the number of
+ // grouping keys, and verifies the run matches the feature-off reference. A
`count(DISTINCT ...)`
+ // group-by has two such Partial phases -- the de-duplication partial
(grouping on key + distinct
+ // columns) and the distinct partial (grouping on the keys only) -- so their
bypasses can be told
+ // apart by the grouping key count.
+ private def bypassRowsByGroupingKeyCount(build: () => DataFrame): Map[Int,
Long] = {
+ val df = build()
+ df.collect()
+ val byKeyCount = collect(df.queryExecution.executedPlan) {
+ case agg: HashAggregateExec if agg.aggregateExpressions.forall(_.mode ==
Partial) =>
+ agg.groupingExpressions.length ->
+ agg.metrics.get("numBypassingRows").map(_.value).getOrElse(0L)
+ }.groupBy(_._1).map { case (n, pairs) => n -> pairs.map(_._2).sum }
+ checkAgainstReference(df, build)
+ byKeyCount
+ }
+
+ /**
+ * Runs `body` once per (wholeStage, twoLevelMap) combination with the
feature enabled and a small
+ * `minRows`, threading a descriptive clue for failure messages.
+ *
+ * The fast (first-level) map is append-only and never spills, so only the
regular (second-level)
+ * map can reach a spill boundary. With the default fast-map capacity (2^16)
a small
+ * high-cardinality input would be fully absorbed by the fast map and never
reach the regular map,
+ * so nothing could ever bypass. To make the triggering tests meaningful
when the two-level map is
+ * on, we shrink the fast map via the first field of `testFallbackStartsAt`
so rows fall through
+ * to the regular map. `regularFallback` optionally sets the second field to
also force the
+ * regular map to spill (for the spill check); when 0 the regular map does
not spill.
+ */
+ private def forEachCodegenAndMap(
+ minRows: Long = 8,
+ regularFallback: Int = 0,
+ minCompaction: Double = -1.0)(
+ body: String => Unit): Unit = {
+ for {
+ wholeStage <- Seq(true, false)
+ twoLevelMap <- Seq(true, false)
+ } {
+ // Shrink the fast map to 4 keys when it is on so rows reach the regular
map. The second field
+ // controls regular-map spilling; 0 means "never" (a large sentinel).
+ val fallbackConf = if (twoLevelMap || regularFallback > 0) {
+ val fastCap = if (twoLevelMap) 4 else 1
+ val regular = if (regularFallback > 0) regularFallback else
Int.MaxValue
+ Seq("spark.sql.TungstenAggregate.testFallbackStartsAt" -> s"$fastCap,
$regular")
+ } else {
+ Nil
+ }
+ // A negative value means "leave the threshold at its default".
+ val thresholdConf = if (minCompaction >= 0.0) {
+ Seq(SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_MIN_COMPACTION.key ->
minCompaction.toString)
+ } else {
+ Nil
+ }
+ withSQLConf(
+ (Seq(
+ SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key -> "true",
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> wholeStage.toString,
+ SQLConf.ENABLE_TWOLEVEL_AGG_MAP.key -> twoLevelMap.toString,
+ SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_MIN_ROWS.key ->
minRows.toString) ++
+ fallbackConf ++ thresholdConf ++ fixedPlanConfs): _*) {
+ body(s"wholeStage=$wholeStage twoLevelMap=$twoLevelMap")
+ }
+ }
+ }
+
+ /////////////////////////////////////////////////////////////////////////////
+ // Part 1: Correctness -- results identical to the feature-off reference.
+ /////////////////////////////////////////////////////////////////////////////
+
+ test("results unchanged for high-cardinality input that bypasses partial
aggregation") {
+ // Every grouping key is distinct, so partial aggregation reduces nothing
and should be
+ // bypassed by the periodic check.
+ checkAdaptiveMatchesReference { parts =>
+ spark.range(0, 200, 1, parts)
+ .select($"id".cast("string") as "k", ($"id" * 2) as "v")
+ .groupBy($"k")
+ .agg(sum($"v") as "s", count(lit(1)) as "c", max($"v") as "m")
+ }
+ }
+
+ test("results unchanged for low-cardinality input that keeps partial
aggregation") {
+ // Few distinct keys, high reduction: partial aggregation is effective and
should be kept, so
+ // the bypass metric must stay zero in every cell.
+ checkAdaptiveMatchesReference(
+ expectBypass = false,
+ build = { parts =>
+ spark.range(0, 600, 1, parts)
+ .select(($"id" % 5).cast("string") as "k", $"id" as "v")
+ .groupBy($"k")
+ .agg(sum($"v") as "s", count(lit(1)) as "c", min($"v") as "mn",
max($"v") as "mx")
+ })
+ }
+
+ test("results unchanged for medium-cardinality input near the reduction
threshold") {
+ // Roughly half the rows are distinct keys; exercises the boundary of the
ratio checks. The
+ // overall compaction ratio (~2.0) is above the threshold, but the *first*
periodic check still
+ // sees the leading distinct keys and fires, so the bypass must be
observable too.
+ checkAdaptiveMatchesReference { parts =>
+ spark.range(0, 1000, 1, parts)
+ .select(($"id" % 500).cast("string") as "k", $"id" as "v")
+ .groupBy($"k")
+ .agg(sum($"v") as "s", count(lit(1)) as "c")
+ }
+ }
+
+ test("results unchanged with multiple grouping keys and string keys") {
+ checkAdaptiveMatchesReference { parts =>
+ spark.range(0, 500, 1, parts)
+ .select(
+ concat(lit("g"), ($"id" % 300).cast("string")) as "k1",
+ ($"id" % 7) as "k2",
+ $"id" as "v")
+ .groupBy($"k1", $"k2")
+ .agg(sum($"v") as "s", count(lit(1)) as "c")
+ }
+ }
+
+ test("results unchanged with nullable grouping keys") {
+ // Nulls are sparse enough (1 in 40) that the keys stay close to unique
and the input really
+ // does bypass; a denser null key would lift the compaction ratio above
the threshold and the
+ // test would never engage the feature.
+ checkAdaptiveMatchesReference { parts =>
+ spark.range(0, 400, 1, parts)
+ .select(
+ when($"id" % 40 === 0, lit(null)).otherwise($"id").cast("string") as
"k",
+ $"id" as "v")
+ .groupBy($"k")
+ .agg(sum($"v") as "s", count(lit(1)) as "c")
+ }
+ }
+
+ test("results unchanged with average (multi-slot buffer) aggregate") {
+ // avg has a two-slot partial buffer (sum, count); pass-through buffers
must carry all slots.
+ checkAdaptiveMatchesReference { parts =>
+ spark.range(0, 300, 1, parts)
+ .select($"id".cast("string") as "k", ($"id" + 1) as "v")
+ .groupBy($"k")
+ .agg(avg($"v") as "a", sum($"v") as "s")
+ }
+ }
+
+ test("results unchanged with a mix of many aggregate functions and buffer
types") {
+ // Exercises a wide pass-through buffer spanning several aggregate buffer
layouts at once:
+ // sum (decimal), avg (double), count, min/max, first/last, and stddev
(imperative buffer).
Review Comment:
**Finding 18.** `stddev` is not an imperative-buffer aggregate —
`CentralMomentAgg` is a `DeclarativeAggregate`, so this test covers a wide
declarative buffer and nothing else. That is worth a bit more than a comment
fix, because no test in the suite reaches a real `ImperativeAggregate` through
pass-through, and that is the one buffer kind where `nextPassThroughOutput`'s
`copyFrom(initialAggregationBuffer)` has to restore state written by
`initialize(buffer)` rather than by a projection.
`approx_count_distinct` is the easy one: `HyperLogLogPlusPlus` is an
`ImperativeAggregate` with a mutable `LongType` buffer, so it plans as a
`HashAggregateExec` whose `supportCodegen` is false and only ever runs on
`TungstenAggregationIterator`. Better as its own test than folded in here — one
imperative function makes `supportCodegen` false for the whole operator, which
would quietly turn this test's codegen cells into interpreted ones.
```scala
test("results unchanged with an imperative aggregate buffer") {
// `HyperLogLogPlusPlus` is an `ImperativeAggregate`, so
`supportCodegen` is false and this
// only runs on `TungstenAggregationIterator` -- the path where a
pass-through buffer is reset
// with `copyFrom(initialAggregationBuffer)` and has to carry state
`initialize` wrote.
checkAdaptiveMatchesReference { parts =>
spark.range(0, 400, 1, parts)
.select($"id".cast("string") as "k", ($"id" % 97) as "v")
.groupBy($"k")
.agg(approx_count_distinct($"v") as "acd", sum($"v") as "s")
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
