ulysses-you commented on code in PR #57742: URL: https://github.com/apache/spark/pull/57742#discussion_r3755858718
########## 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: Fixed. `approx_count_distinct` (`HyperLogLogPlusPlus`, an `ImperativeAggregate` with a mutable `LongType` buffer, `supportCodegen = false`) is now its own test, `results unchanged with an imperative-buffer aggregate`, so the `copyFrom(initialAggregationBuffer)` reset of state written by `initialize(buffer)` is exercised on the interpreted path, and it stays out of the codegen matrix so the mixed test keeps its codegen cells. The mix test's comment now says stddev is declarative. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala: ########## @@ -845,29 +1158,56 @@ case class HashAggregateExec( } } - val declareRowBuffer: String = if (isFastHashMapEnabled) { - val fastRowType = if (isVectorizedHashMapEnabled) { - classOf[MutableColumnarRow].getName + val declareRowBuffer: String = { + val declareBuffers = if (isFastHashMapEnabled) { + val fastRowType = if (isVectorizedHashMapEnabled) { + classOf[MutableColumnarRow].getName + } else { + "UnsafeRow" + } + s""" + |UnsafeRow $unsafeRowBuffer = null; + |$fastRowType $fastRowBuffer = null; + """.stripMargin + } else { + s"UnsafeRow $unsafeRowBuffer = null;" + } + val declareBypassed = if (adaptivePartialAggEnabled) { + s"boolean $adaptiveRowBypassedTerm = false;" } else { - "UnsafeRow" + "" } s""" - |UnsafeRow $unsafeRowBuffer = null; - |$fastRowType $fastRowBuffer = null; + |$declareBuffers + |$declareBypassed """.stripMargin - } else { - s"UnsafeRow $unsafeRowBuffer = null;" } // We try to do hash map based in-memory aggregation first. If there is not enough memory (the // hash map will return null for new key), we spill the hash map to disk to free memory, then // continue to do in-memory aggregation and spilling until all the rows had been processed. // Finally, sort the spilled aggregate buffers by key, and merge them together for same key. + // + // With adaptive partial aggregation, once pass-through is active `updateRowInHashMap` fills the + // single-row buffer built above; we then emit `key ++ buffer` straight to the parent so the row + // skips both the fast map and the regular map. + val emitPassThroughRow = if (adaptivePartialAggEnabled) { + val numBypassingRows = metricTerm(ctx, "numBypassingRows") + s""" + |if ($adaptiveRowBypassedTerm) { + | $numBypassingRows.add(1); + | $outputFunc(${unsafeRowKeyCode.value}, $unsafeRowBuffer); Review Comment: Fixed at this head, with your repro in as `fan-out below a split aggregate preserves the merge order`. A one-to-many child (`GenerateExec`/`ExpandExec`) has no `shouldStop()` in its fan-out loop, so with the drain deferred to the next re-entry the whole trigger batch was appended before the frozen maps drained: codegen gave `(400, 0)` against `(0, 400)` interpreted and `(0, 400)` feature-off. The drain now runs right after the trigger row whenever the child reports `needCopyResult`, which fan-out children always do; their output is a copy anyway, so the aliasing hazard that forced the deferred drain in the whole-stage root is gone, and the two paths and the reference all give `(0, 400)`. We kept trigger-first rather than draining before the first bypassed row, because map-first would need a per-row copy on the generated path or a park-the-trigger re-entry state machine, which cannot resume a fan-out loop that keeps no position; the residual case, the first bypassed row colliding w ith a key in the map, is documented in the order test. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala: ########## @@ -711,6 +995,54 @@ case class HashAggregateExec( } else { findOrInsertRegularHashMap } + + // Every row is either accepted by an aggregation map or streamed through -- the fast map + // serves a row without it ever reaching the regular map, so both buffers are consulted to + // tell the two apart. + // + // An accepted row counts toward the compaction ratio, so the numerator matches the + // operator-level denominator. Counting inside the regular-map branch alone would drop the + // rows the fast map absorbed from the ratio and bypass an aggregation that is in fact + // reducing. A row no map holds is streamed once pass-through is active: both probes are + // skipped (guarded above), so neither buffer is set, and `rowBypassed` marks exactly those + // rows. The row that fails to insert at the spill boundary lands here too, while the row + // that merely flipped pass-through at the check point is already aggregated in the map that + // took it and must not be re-emitted. + val countOrPassThroughRow = if (adaptivePartialAggEnabled) { + val heldByAMap = if (isFastHashMapEnabled) { + s"($fastRowBuffer != null || $unsafeRowBuffer != null)" + } else { + s"($unsafeRowBuffer != null)" + } + // The grouping key was already projected in `findOrInsertRegularHashMap` + // (`unsafeRowKeyCode.code`), so `unsafeRowKeyCode.value` holds this row's key. Only build + // the single-row partial buffer here. + s""" + |if ($heldByAMap) { + | if (!$adaptivePassThroughTerm) { + | $processedRowsTerm += 1; + | if ($processedRowsTerm == $adaptiveNextCheckRowTerm) { Review Comment: Thanks, and agreed it is not a blocker. The docs now state this directly: "once triggered, pass-through is not reversed for the rest of the task", and "a distinct-heavy prefix can trip the pass-through early and keep it tripped even where the rest of the input would aggregate well", alongside the `minCompaction`/`minRows` tuning notes. On the opt-out, `enabled` (default `false`) is the documented switch in the config table; I trimmed the explicit sentence to keep the paragraph tight, say the word if you want it back. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala: ########## @@ -141,6 +148,40 @@ case class HashAggregateExec( .map(_.asInstanceOf[DeclarativeAggregate]) private val bufferSchema = DataTypeUtils.fromAttributes(aggregateBufferAttributes) + /** + * Whether adaptive partial aggregation applies to this operator. When it does, the aggregation + * may bypass partial aggregation at runtime and pass the remaining input rows through as + * single-row partial buffers (see [[SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_ENABLED]]). It only + * applies to a pre-shuffle partial aggregation with grouping keys: + * - `Partial` mode only: the downstream `Final` aggregation merges the passed-through + * single-row buffers, so the output contract is unchanged. `Final`/`Complete` produce the + * result themselves and have no such downstream. `PartialMerge` does have one and could be + * supported by passing its incoming buffer through unchanged, but that is left for later. + * - grouping keys present: a global aggregation produces a single output row, so partial + * aggregation achieves the maximum reduction and must never be bypassed. + * - DISTINCT aggregate functions are allowed: the intermediate `PartialMerge` phase of the + * multi-phase distinct plan is not `Partial` mode (and requires a distribution), so it always + * aggregates and de-duplicates, and the passed-through rows from the distinct `Partial` phase + * therefore carry exactly one distinct value each. + */ + private val adaptivePartialAggEnabled: Boolean = { + conf.adaptivePartialAggregationEnabled && + groupingExpressions.nonEmpty && + // Only the pre-shuffle partial aggregation has a downstream `Final` to merge passed-through + // single-row buffers. `requiredChildDistributionExpressions` is `None` exactly for that + // pre-shuffle phase and `Some` for the `Final`/`Complete` phase. This check is what keeps a + // group-by-only aggregate (no aggregate functions, so an empty `aggregateExpressions`) from + // being admitted vacuously: `aggregateExpressions.forall(_.mode == Partial)` alone is true + // for the empty list, which would wrongly make the `Final` phase eligible as well. + requiredChildDistributionExpressions.isEmpty && Review Comment: You are right, and that sentence is gone. The doc now keeps the two mechanisms separate: the intermediate phase of a mixed DISTINCT plan is excluded by the mode check (its modes are `PartialMerge ++ Partial`), while the group-by-only intermediate phase of a `count(DISTINCT v)`-only plan is excluded by `requiredChildDistributionExpressions`. What makes the eligible distinct `Partial` phase safe is now stated as what you found: its output partitioning `(k, v)` does not satisfy the `Final`'s `ClusteredDistribution(k)`, so another `Exchange` lands between them and there is still a `Final` to merge the passed-through buffers. The PR description says the same. -- 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]
