>From Shahrzad Shirazi <[email protected]>: Shahrzad Shirazi has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21387?usp=email )
Change subject: [ASTERIXDB-3786][COMP] Fix parenthesized single-item list in IN clause ...................................................................... [ASTERIXDB-3786][COMP] Fix parenthesized single-item list in IN clause - user model changes: no - storage format changes: no - interface changes: no Ext-ref: MB-71612 Change-Id: I7d5e69bfecf51dee3b7101894a2ff658cbbd81f8 --- A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/subquery/in_as_or/in_as_or.11.query.sqlpp A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/subquery/in_as_or/in_as_or.12.query.sqlpp A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/subquery/in_as_or/in_as_or.13.query.sqlpp M asterixdb/asterix-app/src/test/resources/runtimets/results/column/filter/sql-compat/sql-compat.013.plan A asterixdb/asterix-app/src/test/resources/runtimets/results/subquery/in_as_or/in_as_or.11.adm A asterixdb/asterix-app/src/test/resources/runtimets/results/subquery/in_as_or/in_as_or.12.adm A asterixdb/asterix-app/src/test/resources/runtimets/results/subquery/in_as_or/in_as_or.13.adm M asterixdb/asterix-lang-sqlpp/src/main/javacc/SQLPP.jj M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/HashDistinctCountAggregateFunction.java 9 files changed, 401 insertions(+), 65 deletions(-) git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/87/21387/1 diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/subquery/in_as_or/in_as_or.11.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/subquery/in_as_or/in_as_or.11.query.sqlpp new file mode 100644 index 0000000..d62d279 --- /dev/null +++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/subquery/in_as_or/in_as_or.11.query.sqlpp @@ -0,0 +1,28 @@ +/* + * 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. + */ + +// A single-value parenthesized list "(v)" must behave like the multi-value list "(v1, v2)" +// and like the array "[v]". Previously "in ('...')" failed with a type mismatch. + +use tpch; + +select value c.c_custkey +from Customer c +where c.c_name in ('Customer#000000001') +order by c.c_custkey diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/subquery/in_as_or/in_as_or.12.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/subquery/in_as_or/in_as_or.12.query.sqlpp new file mode 100644 index 0000000..d2e708d --- /dev/null +++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/subquery/in_as_or/in_as_or.12.query.sqlpp @@ -0,0 +1,28 @@ +/* + * 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. + */ + +// Single-value parenthesized list on the NOT IN path must also work (same root cause). + +use tpch; + +select value c.c_custkey +from Customer c +where c.c_name not in ('Customer#000000001') +order by c.c_custkey +limit 2 diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/subquery/in_as_or/in_as_or.13.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/subquery/in_as_or/in_as_or.13.query.sqlpp new file mode 100644 index 0000000..bfbbe6b --- /dev/null +++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/subquery/in_as_or/in_as_or.13.query.sqlpp @@ -0,0 +1,27 @@ +/* + * 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. + */ + +// Single-element bracket list: must give the same result as the single parenthesized value in part .11. + +use tpch; + +select value c.c_custkey +from Customer c +where c.c_name in ['Customer#000000001'] +order by c.c_custkey diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/column/filter/sql-compat/sql-compat.013.plan b/asterixdb/asterix-app/src/test/resources/runtimets/results/column/filter/sql-compat/sql-compat.013.plan index f27743f..4d7b3a8 100644 --- a/asterixdb/asterix-app/src/test/resources/runtimets/results/column/filter/sql-compat/sql-compat.013.plan +++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/column/filter/sql-compat/sql-compat.013.plan @@ -1,57 +1,42 @@ -distribute result [$$197] [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] +distribute result [$$232] [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] -- DISTRIBUTE_RESULT |PARTITIONED| exchange [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - assign [$$197] <- [{"str2": $$str2, "sum:num3:ok": $$201}] project: [$$197] [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] + assign [$$232] <- [{"str2": $$str2, "sum:num3:ok": $$235}] project: [$$232] [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] -- ASSIGN |PARTITIONED| exchange [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] -- SORT_MERGE_EXCHANGE [$$str2(ASC) ] |PARTITIONED| - group by ([$$str2 := $$233]) decor ([]) { - aggregate [$$201] <- [agg-global-sql-sum($$232)] [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] + group by ([$$str2 := $$266]) decor ([]) { + aggregate [$$235] <- [agg-global-sql-sum($$265)] [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] -- AGGREGATE |LOCAL| nested tuple source [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] -- NESTED_TUPLE_SOURCE |LOCAL| } [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] - -- SORT_GROUP_BY[$$233] |PARTITIONED| + -- SORT_GROUP_BY[$$266] |PARTITIONED| exchange [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] - -- HASH_PARTITION_EXCHANGE [$$233] |PARTITIONED| - group by ([$$233 := $$231]) decor ([]) { - aggregate [$$232] <- [agg-local-sql-sum($$195)] [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] + -- HASH_PARTITION_EXCHANGE [$$266] |PARTITIONED| + group by ([$$266 := $$267]) decor ([]) { + aggregate [$$265] <- [agg-local-sql-sum($$230)] [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] -- AGGREGATE |LOCAL| nested tuple source [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] -- NESTED_TUPLE_SOURCE |LOCAL| } [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] - -- SORT_GROUP_BY[$$231] |PARTITIONED| + -- SORT_GROUP_BY[$$267] |PARTITIONED| exchange [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - assign [$$195] <- [double-default-null($$206)] project: [$$195, $$231] [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] + assign [$$230] <- [double-default-null($#4.getField("num3"))] project: [$$230, $$267] [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] -- ASSIGN |PARTITIONED| - select (or($$184, is-unknown($$231))) project: [$$206, $$231] [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] + select (or(eq($$267, "sixteen"), is-unknown($$267))) [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] -- STREAM_SELECT |PARTITIONED| - project ([$$206, $$231, $$184]) [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] - -- STREAM_PROJECT |PARTITIONED| - subplan { - aggregate [$$184] <- [non-empty-stream()] [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] - -- AGGREGATE |LOCAL| - select (eq(string-default-null($$210), "sixteen")) [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] - -- STREAM_SELECT |LOCAL| - project ([$$210]) [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] - -- STREAM_PROJECT |LOCAL| - nested tuple source [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] - -- NESTED_TUPLE_SOURCE |LOCAL| - } [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] - -- SUBPLAN |PARTITIONED| - assign [$$231] <- [string-default-null($$210)] [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] - -- ASSIGN |PARTITIONED| - assign [$$210, $$206] <- [$#4.getField("str2"), $#4.getField("num3")] project: [$$210, $$206] [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] - -- ASSIGN |PARTITIONED| - project ([$#4]) [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] - -- STREAM_PROJECT |PARTITIONED| - exchange [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] - -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - data-scan []<-[$$200, $#4] <- test.ColumnDataset project ({str2:any,num3:any}) [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] - -- DATASOURCE_SCAN |PARTITIONED| - exchange [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] - -- ONE_TO_ONE_EXCHANGE |PARTITIONED| - empty-tuple-source [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] - -- EMPTY_TUPLE_SOURCE |PARTITIONED| + assign [$$267] <- [string-default-null($#4.getField("str2"))] [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] + -- ASSIGN |PARTITIONED| + project ([$#4]) [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] + -- STREAM_PROJECT |PARTITIONED| + exchange [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] + -- ONE_TO_ONE_EXCHANGE |PARTITIONED| + data-scan []<-[$$234, $#4] <- test.ColumnDataset project ({str2:any,num3:any}) [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] + -- DATASOURCE_SCAN |PARTITIONED| + exchange [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] + -- ONE_TO_ONE_EXCHANGE |PARTITIONED| + empty-tuple-source [cardinality: 0.0, doc-size: 0.0, op-cost: 0.0, total-cost: 0.0] + -- EMPTY_TUPLE_SOURCE |PARTITIONED| \ No newline at end of file diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/subquery/in_as_or/in_as_or.11.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/subquery/in_as_or/in_as_or.11.adm new file mode 100644 index 0000000..56a6051 --- /dev/null +++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/subquery/in_as_or/in_as_or.11.adm @@ -0,0 +1 @@ +1 \ No newline at end of file diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/subquery/in_as_or/in_as_or.12.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/subquery/in_as_or/in_as_or.12.adm new file mode 100644 index 0000000..1234e84 --- /dev/null +++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/subquery/in_as_or/in_as_or.12.adm @@ -0,0 +1,2 @@ +2 +3 \ No newline at end of file diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/subquery/in_as_or/in_as_or.13.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/subquery/in_as_or/in_as_or.13.adm new file mode 100644 index 0000000..56a6051 --- /dev/null +++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/subquery/in_as_or/in_as_or.13.adm @@ -0,0 +1 @@ +1 \ No newline at end of file diff --git a/asterixdb/asterix-lang-sqlpp/src/main/javacc/SQLPP.jj b/asterixdb/asterix-lang-sqlpp/src/main/javacc/SQLPP.jj index 968cd0f..2b1291e 100644 --- a/asterixdb/asterix-lang-sqlpp/src/main/javacc/SQLPP.jj +++ b/asterixdb/asterix-lang-sqlpp/src/main/javacc/SQLPP.jj @@ -322,6 +322,10 @@ private INamespaceResolver namespaceResolver; + // Last LiteralExpr returned by ParenthesizedExpression() for a single parenthesized literal like "('a')"; + // RelExpr() uses it to promote "x IN ('a')" to "x IN ['a']". + private Expression lastParenthesizedLiteral; + private static class IndexParams { public IndexType type; public int gramLength; @@ -4645,6 +4649,7 @@ Expression RelExpr() throws ParseException: { boolean not = false; + boolean isInOp = false; OperatorExpr op = null; Token opToken = null; Expression operand = null; @@ -4655,7 +4660,7 @@ operand = BetweenExpr() ( - LOOKAHEAD(3)( <LT> | <GT> | <LE> | <GE> | <EQ> | <NE> | <LG> |<SIMILAR> | (<NOT> { not = true; })? <IN> | + LOOKAHEAD(3)( <LT> | <GT> | <LE> | <GE> | <EQ> | <NE> | <LG> |<SIMILAR> | (<NOT> { not = true; })? <IN> { isInOp = true; } | <IS> (<NOT> { not = true; })? <DISTINCT> { opToken = token; } <FROM> ) { if (opToken == null) { @@ -4691,10 +4696,21 @@ } catch (CompilationException e){ throw new SqlppParseException(getSourceLocation(token), e.getMessage()); } + // Reset so the check below reflects only the right-hand side. + lastParenthesizedLiteral = null; } operand = BetweenExpr() { + // For IN/NOT IN only: promote a single parenthesized literal "('a')" to the list "['a']". + // Everything else (bare "'a'", fields, "(a + b)", subqueries, "[...]") is left unchanged. + if (isInOp && operand == lastParenthesizedLiteral) { + List<Expression> inItemList = new ArrayList<Expression>(1); + inItemList.add(operand); + ListConstructor inListExpr = new ListConstructor(ListConstructor.Type.ORDERED_LIST_CONSTRUCTOR, inItemList); + inListExpr.setSourceLocation(operand.getSourceLocation()); + operand = inListExpr; + } op.addOperand(operand); } )? @@ -5675,6 +5691,11 @@ ListConstructor listExpr = new ListConstructor(ListConstructor.Type.ORDERED_LIST_CONSTRUCTOR, exprList); return addSourceLocation(listExpr, startToken); } else { + // startToken is set only on the "( Expression )" branch (not Subquery()); record a single + // parenthesized literal for RelExpr()'s IN handling. + if (startToken != null && expr.getKind() == Expression.Kind.LITERAL_EXPRESSION) { + lastParenthesizedLiteral = expr; + } return expr; } } diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/HashDistinctCountAggregateFunction.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/HashDistinctCountAggregateFunction.java index bdcb2e1..9c2c354 100644 --- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/HashDistinctCountAggregateFunction.java +++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/HashDistinctCountAggregateFunction.java @@ -153,6 +153,20 @@ private byte[] prevValue; // last distinct value seen during the merge (for adjacent comparison) private int prevValueLen; + /** + * Creates the aggregator for one task. Runs once per task (not per group), so it sets up everything + * reused across all groups: the value evaluator, and the type-specialized hash function, comparator, + * normalized-key factory, serde and record descriptor — all built from the argument's actual type + * ({@link BuiltinType#ANY} when the type is unknown, which is the common schemaless case). It also + * derives the spill threshold {@code memoryCapBytes} from the sort-memory frame budget + * ({@code numFrames}); a value of {@code 0} disables spilling (unbounded), preserving legacy behavior. + * + * @param args evaluator factories; {@code args[0]} produces the value to be counted + * @param context evaluator context, used for the task context (frames, I/O) and the scalar evaluator + * @param sourceLoc source location for error reporting + * @param itemType the argument's actual type, or {@code null}/{@code ANY} when unknown + * @param numFrames the sort-memory budget in frames; the in-memory table spills once it reaches this + */ public HashDistinctCountAggregateFunction(IScalarEvaluatorFactory[] args, IEvaluatorContext context, SourceLocation sourceLoc, IAType itemType, int numFrames) throws HyracksDataException { super(sourceLoc); @@ -180,10 +194,21 @@ } /** - * Allocates the hash table at its initial size and clears the byte/slot counters. Shared by the - * constructor (first-time setup) and releasing a grown table on - * spill. It does not touch {@code currentStamp}: the constructor sets that to 0 itself, and on a - * spill the live stamp must be preserved while the freshly zeroed {@code slotStamp} reads as empty. + * (Re)allocates the hash table at {@link #INITIAL_CAPACITY} and resets {@code mask}, {@code growAt}, + * the four parallel slot arrays, the value arena, and the byte/slot counters. It is shared by two + * callers with different intents: + * <ul> + * <li>the <b>constructor</b>, for first-time setup; and</li> + * <li>{@link #startSpilling()}, to <b>release</b> a table that one large group grew, so the freed + * memory isn't held while the sorter runs.</li> + * </ul> + * + * <p>It deliberately does <em>not</em> touch {@code currentStamp}. The constructor sets that to 0 + * itself afterwards; on a spill the live generation must be preserved because the group is still in + * progress. Either way the freshly allocated {@code slotStamp} is all zeros, which reads as empty + * against any live stamp ≥ 1 — so no explicit clearing is ever needed. Note this allocates new + * arrays (it does not reuse the existing ones), which is what actually returns the grown memory to the + * GC on the spill path. */ private void allocateInitialTable() { capacity = INITIAL_CAPACITY; @@ -198,6 +223,31 @@ liveSlots = 0; } + /** + * Begins a new group. The same aggregator instance is reused for <em>every</em> group of a GROUP BY / + * GROUPING SETS query, so this method is on a hot path and is deliberately O(1). + * + * <h4>How the reset works</h4> + * Rather than zeroing the slot arrays or the arena (which would be O(capacity) per group and dominate + * cost when there are many small groups), it advances the generation counter {@code currentStamp} by + * one. A slot is considered occupied only when {@code slotStamp[i] == currentStamp}; bumping the stamp + * makes every slot left over from the previous group fail that test, so the whole table instantly + * reads as empty without a single write to the slot arrays. The stale bytes are simply overwritten as + * the new group fills the table. + * + * <h4>What is and isn't reset</h4> + * The grown {@code capacity} and the allocated arena are intentionally <em>retained</em>: a later group + * reuses the space instead of regrowing from {@link #INITIAL_CAPACITY}, which matters when one group + * already proved the table needs to be large. The per-group counters ({@code liveSlots}, + * {@code arenaEnd}, {@code distinctCount}) are zeroed, and {@code spilling} is cleared so the group + * starts on the fast in-memory path. + * + * <h4>Generation wrap-around</h4> + * The only non-constant branch is the wrap after 2<sup>32</sup> groups: when {@code currentStamp} + * overflows back to 0 it would collide with the default stamp of never-used slots, so the stamps are + * cleared once and the generation restarts at 1. This costs O(capacity) but happens at most once every + * ~4 billion groups, so its amortized cost is negligible. + */ @Override public void init() throws HyracksDataException { // O(1) reset: a fresh generation makes every slot read as empty without touching memory. @@ -212,6 +262,31 @@ spilling = false; } + /** + * Consumes one input tuple of the current group. Called once per row, so it is the hottest method in + * the class. + * + * <h4>Steps</h4> + * <ol> + * <li>Evaluate {@code args[0]} against the tuple into {@code inputVal} (the value's tagged bytes).</li> + * <li>Read the leading type-tag byte and skip {@code NULL}, {@code MISSING} and {@code SYSTEM_NULL} — + * SQL {@code COUNT(DISTINCT)} does not count these. {@code SYSTEM_NULL} is the internal + * placeholder produced by some plans and must be ignored just like a user {@code NULL}.</li> + * <li>Route by mode. If the group has already overflowed ({@code spilling}), forward the value + * straight to the sorted-run generator and return; no in-memory dedup is attempted, because the + * table is full and the final dedup is done by the merge in {@link #finish}. Otherwise insert + * into the in-memory set; only a genuinely new value increments {@code distinctCount}.</li> + * </ol> + * + * <h4>The spill trigger</h4> + * After inserting a new value, the estimated table footprint — {@code arenaEnd} (value bytes so far) + * plus {@code capacity * BYTES_PER_SLOT} (the four parallel slot arrays) — is compared against + * {@code memoryCapBytes}. Crossing it calls {@link #startSpilling()} for this group. The comparison is + * widened to {@code long} to avoid overflow on large tables, and is only evaluated on the insert path, + * so duplicates (which dominate low-cardinality groups) add no per-row branch cost beyond the lookup. + * + * @param tuple the current input tuple; {@code args[0]} is evaluated against it + */ @Override public void step(IFrameTupleReference tuple) throws HyracksDataException { eval.evaluate(tuple, inputVal); @@ -239,6 +314,45 @@ } } + /** + * Inserts a value into the in-memory hash set unless an equal value is already present. This is the + * core of the fast path and is written to keep the common case (a miss, or a hash-rejected collision) + * as cheap as possible. + * + * <h4>Probe</h4> + * The set is an open-addressing table with linear probing. The starting slot is {@code hash & mask}; + * because {@code capacity} is always a power of two, this is equivalent to {@code hash % capacity} but + * branch-free, and it yields a valid index even for a negative hash (the mask's top bit is 0). The + * loop walks forward while {@code slotStamp[slot] == currentStamp} (i.e. over slots occupied in this + * generation), wrapping with {@code (slot + 1) & mask}. It stops at the first slot whose stamp differs + * — an empty slot — which is guaranteed to exist because the load factor is kept below 1. + * + * <h4>Two-level equality check</h4> + * For each occupied slot it first compares the cached 32-bit {@code slotHash}. That single int compare + * rejects essentially all non-matching collisions without touching the value bytes. Only when the + * cached hashes are equal does it invoke the (potentially expensive, type-dispatching) comparator over + * the stored bytes in the arena versus the incoming bytes. A comparator result of 0 means the value is + * already present, so it returns {@code false} and stores nothing. + * + * <h4>Insert</h4> + * On reaching an empty slot the value is new: {@link #ensureArenaCapacity(int)} guarantees room, the + * bytes are copied into {@code valueArena} at {@code arenaEnd}, and the slot records the value's arena + * offset, length, cached hash, and the current stamp (which makes it read as occupied). {@code arenaEnd} + * and {@code liveSlots} advance, and if the table has crossed {@code growAt} it is doubled by + * {@link #rehash()}. + * + * <h4>Performance and correctness notes</h4> + * The slot arrays, mask, stamp and arena are hoisted into locals for the probe loop to avoid repeated + * field loads. The copy on the insert path deliberately uses the {@code valueArena} <em>field</em>, not + * the hoisted {@code arena} local, because {@link #ensureArenaCapacity(int)} may have just replaced the + * arena with a larger array; the local is only safe for the read-only probe that ran before any growth. + * This method performs no allocation per value other than the amortized arena/array growth, and never + * boxes — values live as raw bytes in the arena. + * + * @param item pointer to the (already null/missing-filtered) value's tagged bytes + * @return {@code true} if the value was new and was stored; {@code false} if an equal value already + * existed in this group + */ private boolean insertIfAbsent(IPointable item) throws HyracksDataException { final byte[] itemBytes = item.getByteArray(); final int itemStart = item.getStartOffset(); @@ -279,9 +393,24 @@ } /** - * Doubles the slot table and reinserts the live entries. The arena is untouched — only the slot - * index is rebuilt. New stamp arrays default to 0, which reads as empty because the live stamp is - * always >= 1, so no clearing is needed. + * Grows the table by doubling its capacity and re-filing the live entries, keeping probe chains short + * as a group accumulates distinct values. Triggered from {@link #insertIfAbsent} once {@code liveSlots} + * exceeds {@code growAt} ({@code capacity * LOAD_FACTOR}). + * + * <h4>What moves and what doesn't</h4> + * Only the <em>index</em> is rebuilt: new slot arrays of double the size are allocated and each live + * entry (one whose stamp matches {@code currentStamp}) is re-probed into the new arrays using its + * <em>cached</em> hash — so values are never re-hashed and, crucially, the {@code valueArena} is never + * touched or copied. The stored bytes stay where they are; only the small fixed-size slot records move. + * That makes a rehash cost O(capacity) ints, independent of the (potentially large) total value bytes. + * + * <h4>Why no clearing is needed</h4> + * The new {@code slotStamp} array is all zeros by default, which reads as empty against the live stamp + * (always ≥ 1), so the new table starts clean with no explicit initialization. Capacity stays a + * power of two, so {@code mask} and {@code growAt} are recomputed accordingly. + * + * <p>Doubling keeps the amortized insert cost O(1): the total re-probing work across all growths up to + * size N is proportional to N. */ private void rehash() { int newCapacity = capacity << 1; @@ -313,6 +442,13 @@ slotStamp = newStamps; } + /** + * Ensures the value arena has room for {@code needed} more bytes, growing it (at least doubling, or + * enough to fit the new value) if the current value would overflow it. Slot offsets stay valid because + * the bytes are copied into the new, larger array. + * + * @param needed number of additional bytes about to be appended to the arena + */ private void ensureArenaCapacity(int needed) { if (arenaEnd + needed > valueArena.length) { valueArena = Arrays.copyOf(valueArena, Math.max(valueArena.length << 1, arenaEnd + needed)); @@ -320,9 +456,13 @@ } /** - * Transition the current group from in-memory hashing to sort-spilling. The sort structures are - * created once and reused across groups; the live values already in the hash table are pushed into - * the run generator first so the merge in finish() sees every value for this group. + * Transitions the current group from in-memory hashing to sort-spilling, a one-way switch for the + * remainder of this group. It lazily builds the sort structures (see {@link #ensureSpillStructures}), + * then pushes every value currently live in the hash table into the run generator first — so the merge + * in {@link #finish} sees the whole group, not just the values that arrive after the switch. Finally it + * calls {@link #allocateInitialTable} to release the large table the group grew (its contents are now + * safely in the generator and the hash is never read again this group), keeping peak memory near one + * budget rather than carrying a dead table alongside the active sorter. */ private void startSpilling() throws HyracksDataException { ensureSpillStructures(); @@ -342,13 +482,13 @@ } /** - * Releases the large slot arrays and value arena that a big group grew, returning the table to its - * initial size. Called only when a group switches to spilling — a rare, already-expensive event — so - * the common in-memory groups keep their grown capacity for fast reuse. The freshly allocated stamp - * array is all zeros, which reads as empty because the live stamp ({@code currentStamp}) is always - * >= 1, so the next group starts clean without any explicit clearing. + * Lazily builds the sort-spill machinery the first time any group overflows, and reuses it for all + * later spilling groups (guarded by the {@code runsGenerator != null} check). This keeps the common + * case — groups that never spill — from paying for the frames, appender, tuple builder, accessor, + * dedicated merge comparator, normalized-key computer, and the {@link ExternalSortRunGenerator}. The + * run generator is bounded to {@code numFrames} frames and sorts by the single value field using the + * type's comparator and normalized-key factory. */ - private void ensureSpillStructures() throws HyracksDataException { if (runsGenerator != null) { return; @@ -365,10 +505,25 @@ EnumFreeSlotPolicy.LAST_FIT, numFrames); } + /** + * Convenience overload that feeds a pointable's value into the sorted-run generator. + * + * @param value pointer to the value's tagged bytes + */ private void appendToRuns(IPointable value) throws HyracksDataException { appendBytesToRuns(value.getByteArray(), value.getStartOffset(), value.getLength()); } + /** + * Appends one value as a single-field tuple to the run generator. The generator buffers it in a frame, + * and when a frame fills it sorts and spills a run to disk — so this call is bounded by the frame + * budget regardless of how many values are appended. No deduplication happens here; duplicates are + * collapsed later by the sorted merge in {@link #finish}. + * + * @param bytes backing array holding the value's tagged bytes + * @param start start offset of the value within {@code bytes} + * @param len length of the value in bytes + */ private void appendBytesToRuns(byte[] bytes, int start, int len) throws HyracksDataException { tupleBuilder.reset(); tupleBuilder.addField(bytes, start, len); @@ -376,6 +531,13 @@ tupleBuilder.getByteArray(), 0, tupleBuilder.getSize()); } + /** + * Produces the group's result. If the group never spilled, the answer is simply the in-memory + * {@code distinctCount}; if it spilled, the answer comes from {@link #finishSpilled} (the bounded + * merge over the sorted runs). The count is serialized as an {@code AInt64} into the result pointable. + * + * @param resultPointable output pointable set to the serialized distinct count + */ @Override public void finish(IPointable resultPointable) throws HyracksDataException { long count = spilling ? finishSpilled() : distinctCount; @@ -390,9 +552,41 @@ } /** - * Flush the remaining buffered values, finalize the runs, then merge them and count distinct - * adjacent values. Equal values are adjacent in the merged sorted stream, so duplicates split across - * runs collapse and the count is exact. + * Computes the exact distinct count for a group that overflowed onto disk. Invoked from + * {@link #finish} only when {@code spilling} is set. + * + * <h4>Algorithm</h4> + * Distinct counting reduces to "sort the values, then count the points where the value changes." This + * method: + * <ol> + * <li>flushes the last partially filled input frame into the run generator and {@code close()}s it, + * which sorts/flushes any in-memory remainder into runs;</li> + * <li>gathers the sorted runs via {@link #materializeRuns()} (which also handles the case where the + * data never actually spilled to disk);</li> + * <li>merges them with {@link ExternalSortRunMerger} into one globally sorted stream and feeds that + * stream to a {@link DistinctCountingFrameWriter}, which counts each value that differs from its + * predecessor.</li> + * </ol> + * + * <h4>Why the count is exact</h4> + * In a single globally sorted stream all copies of a value are adjacent — including copies that were + * written into <em>different</em> runs (e.g. a value left in the hash at spill time and the same value + * appended afterwards). Counting only the transitions therefore collapses every duplicate exactly once, + * with no double counting and nothing missed. + * + * <h4>Why the memory stays bounded</h4> + * {@code ExternalSortRunMerger} performs a <em>bounded, multi-pass</em> merge: it merges at most + * {@code numFrames - 1} runs at a time, writing intermediate runs and repeating until a single stream + * remains. Peak memory is therefore ~{@code numFrames} frames regardless of how many runs a huge group + * produced. (A naive single-pass k-way merge would need one input frame per run, which is unbounded in + * the number of runs — that is the bug this design avoids.) The merger and its temporary run files are + * delete-on-close, so disk is reclaimed as soon as the count completes. + * + * <h4>Reuse</h4> + * The merger is built lazily on the first spilling group and {@code reset(...)} for subsequent ones, so + * a query with several large groups doesn't reconstruct it each time. + * + * @return the exact number of distinct non-null values in the group */ private long finishSpilled() throws HyracksDataException { if (appender.getTupleCount() > 0) { @@ -428,13 +622,25 @@ } /** - * Receives the merger's globally sorted output frame by frame and counts distinct values. Because the - * stream is sorted, equal values are adjacent, so counting each value that differs from the previous - * one yields the exact distinct count. {@code prevValue} is copied (not referenced) because the merger - * reuses its output buffer between frames. + * The terminal {@link IFrameWriter} of the spill path: it receives the merger's globally sorted output + * one frame at a time and tallies distinct values, instead of writing them anywhere. Because the stream + * is sorted, all equal values are adjacent, so counting each value that differs from its immediate + * predecessor yields the exact distinct count in a single linear pass over the merged output. + * + * <p>It is an inner (non-static) class so it can reuse the enclosing instance's {@code comparator}, + * {@code mergeAccessor}, and {@code prevValue} scratch state. {@code open()} resets the running + * {@code count} and the {@code hasPrev} flag; {@code fail()}/{@code close()} are no-ops because there is + * no downstream writer or external resource to release here. + * + * <p><b>Why {@code prevValue} is copied, not referenced:</b> the comparison spans frame boundaries (a + * value's duplicate may be the first tuple of the <em>next</em> frame), but the merger reuses its output + * buffer between {@code nextFrame} calls. Keeping a pointer into the buffer would compare against + * overwritten bytes, so {@link #rememberPrev} copies the last value into a private array. */ private final class DistinctCountingFrameWriter implements IFrameWriter { + /** Running number of distinct values seen so far in the merged stream. */ private long count; + /** False until the first value is seen, so the first value always counts. */ private boolean hasPrev; @Override @@ -443,6 +649,12 @@ hasPrev = false; } + /** + * Counts the distinct values in one frame of the merged (globally sorted) stream. For each tuple it + * locates the value's bytes via {@code mergeAccessor} and, using the same comparator as the + * in-memory probe, increments {@code count} only when the value differs from the previous one, + * then remembers it for the next comparison. + */ @Override public void nextFrame(ByteBuffer buffer) throws HyracksDataException { mergeAccessor.reset(buffer); @@ -469,8 +681,21 @@ } /** - * Returns the sorted runs to merge. When the whole group fit in sort memory (no spill), the sorted - * in-memory data is flushed to a single managed run file so the merge path is uniform. + * Returns the list of sorted runs for {@link #finishSpilled} to merge, normalizing the two cases the + * run generator can leave behind after {@code close()}: + * <ul> + * <li><b>It spilled.</b> {@code getRuns()} already holds the sorted run files; return them as-is.</li> + * <li><b>It didn't spill</b> (the group fit within the sort-memory budget). {@code getRuns()} is empty + * and the sorted data is still sitting in the in-memory sorter. This method flushes that sorter to + * a single managed run file and returns it, so the caller always sees a uniform "merge these runs" + * interface and needs no separate in-memory code path.</li> + * </ul> + * + * <p>The flushed in-memory run is created as a <em>delete-on-close</em> reader, matching how the spilled + * runs are produced, so the temporary file is removed as soon as the merge finishes reading it rather + * than lingering until job cleanup. + * + * @return the sorted runs to merge; empty only if the group had no countable values */ private List<GeneratedRunFileReader> materializeRuns() throws HyracksDataException { List<GeneratedRunFileReader> runs = new ArrayList<>(runsGenerator.getRuns()); @@ -496,6 +721,16 @@ return runs; } + /** + * Saves the last distinct value seen during the merge so the next value can be compared against it. + * The bytes are <em>copied</em> into {@code prevValue} (growing it if needed) rather than referenced, + * because the merger reuses its output frame between {@code nextFrame} calls — a duplicate that lands + * at the start of the next frame would otherwise be compared against overwritten memory. + * + * @param buf backing array of the current merged frame + * @param start start offset of the value to remember + * @param len length of the value in bytes + */ private void rememberPrev(byte[] buf, int start, int len) { if (prevValue == null || prevValue.length < len) { prevValue = new byte[len]; @@ -504,9 +739,17 @@ prevValueLen = len; } + /** + * Emits a partial result. {@code HASH_DISTINCT_COUNT} is single-step (not combinable): it is registered + * with no local/global decomposition and always runs after the data is partitioned by group key (or + * gathered to one place), so it sees each group complete. There is therefore nothing to combine + * downstream, and the partial result is identical to the final result — hence this delegates to + * {@link #finish}. (A true partial would be invalid here, since distinct counts cannot be summed.) + * + * @param resultPointable output pointable set to the (final) distinct count + */ @Override public void finishPartial(IPointable resultPointable) throws HyracksDataException { - // HASH_DISTINCT_COUNT is single-step (not combinable): the partial result is the final result. finish(resultPointable); } } -- To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21387?usp=email To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings?usp=email Gerrit-MessageType: newchange Gerrit-Project: asterixdb Gerrit-Branch: master Gerrit-Change-Id: I7d5e69bfecf51dee3b7101894a2ff658cbbd81f8 Gerrit-Change-Number: 21387 Gerrit-PatchSet: 1 Gerrit-Owner: Shahrzad Shirazi <[email protected]>
