xiangfu0 commented on code in PR #19311:
URL: https://github.com/apache/pinot/pull/19311#discussion_r3866523273
##########
pinot-common/src/main/proto/plan.proto:
##########
@@ -44,6 +44,7 @@ message PlanNode {
// serialized by an older-version broker can still be deserialized. No
current broker sets this field.
EnrichedJoinNode enrichedJoinNode = 17 [deprecated = true];
UnnestNode unnestNode = 18;
+ MatchNode matchNode = 19;
Review Comment:
This tag is carried from broker to server in `Worker.StagePlan`. With a new
broker and old server, field 19 is unknown, the old protobuf reports
`NODE_NOT_SET`, and its `PlanNodeDeserializer` throws. Current serde tests use
only current-version classes. Please add mixed-version/golden-byte coverage and
either gate MATCH with an actionable “upgrade servers” error or document and
test that MATCH is unavailable until all servers are upgraded.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MatchOperator.java:
##########
@@ -0,0 +1,356 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.query.runtime.operator;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.datatable.StatMap;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
+import org.apache.pinot.core.data.table.Key;
+import org.apache.pinot.query.planner.plannode.MatchNode;
+import org.apache.pinot.query.planner.plannode.PatternSymbol;
+import org.apache.pinot.query.runtime.blocks.MseBlock;
+import org.apache.pinot.query.runtime.blocks.RowHeapDataBlock;
+import org.apache.pinot.query.runtime.operator.match.MatchExpression;
+import org.apache.pinot.query.runtime.operator.match.MatchLimits;
+import org.apache.pinot.query.runtime.operator.match.MatchTape;
+import org.apache.pinot.query.runtime.operator.match.PartitionMatcher;
+import org.apache.pinot.query.runtime.operator.match.PatternNfa;
+import org.apache.pinot.query.runtime.operator.match.PatternToNfaCompiler;
+import org.apache.pinot.query.runtime.operator.utils.AggregationUtils;
+import org.apache.pinot.query.runtime.operator.utils.TypeUtils;
+import org.apache.pinot.query.runtime.plan.OpChainExecutionContext;
+import org.apache.pinot.spi.exception.QueryErrorCode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Evaluates SQL:2016 `MATCH_RECOGNIZE` (row pattern recognition) with `ONE
ROW PER MATCH`.
+///
+/// ## What it does per partition
+///
+/// For every `PARTITION BY` partition, in `ORDER BY` order, it walks a scan
position from the first row to
+/// the last. At each position it asks [PartitionMatcher] for the preferred
match starting exactly there. On a
+/// match it emits one row - the partition key columns followed by the
`MEASURES` - and then moves the scan
+/// position according to the `AFTER MATCH SKIP` mode. On no match it moves
one row forward.
+///
+/// ## What it expects from the plan
+///
+/// Like [WindowAggregateOperator], this operator does not sort.
`PinotMatchExchangeNodeInsertRule` puts a
+/// sort exchange underneath that hash distributes on the partition keys and
sorts the receiver side on
+/// `(partitionKeys..., orderKeys...)`, so rows arrive grouped by partition
and ordered within a partition. The
+/// operator therefore buffers one partition at a time and releases it at each
boundary, and never reads
+/// [MatchNode#getCollations()]: the ordering has already been established
below it.
+///
+/// The grouping half of that assumption is verified rather than trusted: if a
partition key reappears after its
+/// partition was closed, the operator fails instead of silently splitting one
partition into two and reporting matches
+/// that do not exist. The ordering half is not re-checked per row, because an
exchange that grouped correctly but
+/// sorted incorrectly is not a failure mode the exchange can produce - losing
the sort loses the grouping too, which
+/// the reappearance check already catches.
+///
+/// ## Guardrails throw, they never truncate
+///
+/// [MatchLimits#MAX_ROWS_IN_MATCH] bounds the rows buffered for a partition
and
+/// [MatchLimits#MAX_STEPS_PER_MATCH_ATTEMPT] bounds the backtracking of one
match attempt. Both raise an error,
+/// because a truncated pattern result is a wrong result that nothing in the
response would flag.
+///
+/// ## Not supported yet
+///
+/// `ALL ROWS PER MATCH` is rejected here as well as during planning: this
operator emits exactly one row per
+/// match, so accepting it would silently return the wrong shape of result.
+public class MatchOperator extends MultiStageOperator {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(MatchOperator.class);
+ private static final String EXPLAIN_NAME = "MATCH_RECOGNIZE";
+
+ private final MultiStageOperator _input;
+ private final DataSchema _resultSchema;
+ private final ColumnDataType[] _resultStoredTypes;
+ private final int[] _partitionKeys;
+ private final List<PatternSymbol> _patternSymbols;
+ private final MatchExpression[] _measures;
+ private final PartitionMatcher _matcher;
+ private final MatchNode.AfterMatchSkipMode _skipMode;
+ private final int _skipToSymbolOrdinal;
+ private final int _maxRowsInMatch;
+ private final StatMap<StatKey> _statMap = new StatMap<>(StatKey.class);
+
+ private final Set<Key> _closedPartitionKeys = new HashSet<>();
+ private List<Object[]> _partitionRows = new ArrayList<>();
+ private List<Object[]> _outputRows = new ArrayList<>();
+ @Nullable
+ private Key _currentPartitionKey;
+ @Nullable
+ private MseBlock.Eos _inputEos;
+ private int _numRows;
+
+ public MatchOperator(OpChainExecutionContext context, MultiStageOperator
input, DataSchema inputSchema,
+ MatchNode node) {
+ super(context);
+ if (node.getRowsPerMatchMode() !=
MatchNode.RowsPerMatchMode.ONE_ROW_PER_MATCH) {
+ throw QueryErrorCode.QUERY_EXECUTION.asException(
+ "ALL ROWS PER MATCH is not supported yet in MATCH_RECOGNIZE. Use ONE
ROW PER MATCH (the default) and "
+ + "expose the per-match values you need through the MEASURES
clause.");
+ }
+ _input = input;
+ _resultSchema = node.getDataSchema();
+ _resultStoredTypes = _resultSchema.getStoredColumnDataTypes();
+ List<Integer> partitionKeys = node.getPartitionKeys();
+ _partitionKeys = new int[partitionKeys.size()];
+ for (int i = 0; i < _partitionKeys.length; i++) {
+ _partitionKeys[i] = partitionKeys.get(i);
+ }
+ _patternSymbols = node.getPatternSymbols();
+ List<MatchNode.Measure> measures = node.getMeasures();
+ _measures = new MatchExpression[measures.size()];
+ for (int i = 0; i < _measures.length; i++) {
+ _measures[i] = MatchExpression.compile(measures.get(i).getExpression(),
inputSchema);
+ }
+ if (_resultSchema.size() != _partitionKeys.length + _measures.length) {
+ throw QueryErrorCode.QUERY_EXECUTION.asException(
+ "MATCH_RECOGNIZE output schema " + _resultSchema + " does not match
" + _partitionKeys.length
+ + " partition key(s) plus " + _measures.length + " measure(s)");
+ }
+ _skipMode = node.getAfterMatchSkipMode();
+ _skipToSymbolOrdinal = node.getAfterMatchSkipToSymbolOrdinal();
+
+ PatternNfa nfa = PatternToNfaCompiler.compile(node.getPattern());
+ _maxRowsInMatch =
MatchLimits.getMaxRowsInMatch(context.getOpChainMetadata(), node.getNodeHint());
+ long maxStepsPerMatchAttempt =
+ MatchLimits.getMaxStepsPerMatchAttempt(context.getOpChainMetadata(),
node.getNodeHint());
+ _matcher = new PartitionMatcher(nfa, _patternSymbols, inputSchema,
maxStepsPerMatchAttempt);
+ }
+
+ @Override
+ protected Logger logger() {
+ return LOGGER;
+ }
+
+ @Override
+ public List<MultiStageOperator> getChildOperators() {
+ return List.of(_input);
+ }
+
+ @Override
+ public Type getOperatorType() {
+ return Type.MATCH;
+ }
+
+ @Override
+ public String toExplainString() {
+ return EXPLAIN_NAME;
+ }
+
+ @Override
+ public void registerExecution(long time, int numRows, long memoryUsedBytes,
long gcTimeMs) {
+ _statMap.merge(StatKey.EXECUTION_TIME_MS, time);
+ _statMap.merge(StatKey.EMITTED_ROWS, numRows);
+ _statMap.merge(StatKey.ALLOCATED_MEMORY_BYTES, memoryUsedBytes);
+ _statMap.merge(StatKey.GC_TIME_MS, gcTimeMs);
+ }
+
+ @Override
+ public StatMap<StatKey> copyStatMaps() {
+ return new StatMap<>(_statMap);
+ }
+
+ @Override
+ protected MseBlock getNextBlock() {
+ while (_outputRows.isEmpty()) {
+ if (_inputEos != null) {
+ return _inputEos;
+ }
+ MseBlock block = _input.nextBlock();
+ if (block.isData()) {
+ consumeBlock((MseBlock.Data) block);
+ } else {
+ _inputEos = (MseBlock.Eos) block;
+ if (_inputEos.isError()) {
+ return _inputEos;
+ }
+ closeCurrentPartition();
+ }
+ }
+ List<Object[]> rows = _outputRows;
+ _outputRows = new ArrayList<>();
+ return new RowHeapDataBlock(rows, _resultSchema);
+ }
+
+ /// Buffers the rows of one input block, matching and releasing a partition
as soon as its last row went by.
+ private void consumeBlock(MseBlock.Data block) {
+ for (Object[] row : block.asRowHeap().getRows()) {
+ Key partitionKey = AggregationUtils.extractRowKey(row, _partitionKeys);
Review Comment:
This top-level row loop calls `extractRowKey()`, which allocates both an
`Object[]` and a `Key` for every input row, including every subsequent row in
the same partition. A one-million-row partition therefore creates roughly two
million short-lived objects before matching. Please compare the selected fields
with the current key and materialize a `Key` only when opening a partition.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/match/MatchTerm.java:
##########
@@ -0,0 +1,285 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.query.runtime.operator.match;
+
+import it.unimi.dsi.fastutil.ints.IntArrayList;
+import java.math.BigDecimal;
+import java.math.MathContext;
+import java.util.Arrays;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
+import org.apache.pinot.query.runtime.operator.operands.TransformOperand;
+import org.apache.pinot.query.runtime.operator.utils.TypeUtils;
+import org.apache.pinot.spi.exception.QueryErrorCode;
+
+
+/// A leaf of a MEASURES or DEFINE expression whose value depends on the match
rather than on a single row: a row
+/// pattern navigation, `CLASSIFIER()`, `MATCH_NUMBER()`, or a single variable
aggregate.
+///
+/// [MatchExpression] replaces every such leaf with a slot in a synthetic row,
so that everything above the
+/// leaves - comparisons, boolean connectives, arithmetic, scalar functions -
is evaluated by Pinot's ordinary
+/// [TransformOperand] machinery instead of a second expression interpreter.
+///
+/// Implementations are stateless with respect to the match: all match state
is passed in through the
+/// [MatchTape], so one instance is reused across every match of every
partition.
+public interface MatchTerm {
+
+ /// The type this term reports to the enclosing expression. Values returned
by [#evaluate] are in the
+ /// corresponding [stored][ColumnDataType#getStoredType()] representation,
exactly like the values of a real
+ /// input row.
+ ColumnDataType getResultType();
+
+ @Nullable
+ Object evaluate(MatchTape tape);
+
+ /// A row pattern navigation: an optional logical step (`FIRST` / `LAST`)
that designates a row of the
+ /// match, followed by an optional physical step (`PREV` / `NEXT`) that
moves a fixed number of rows
+ /// relative to it, and finally a column read.
+ ///
+ /// Both steps are needed because SQL:2016 nests them, e.g.
`PREV(LAST(A.price), 2)` designates the last row
+ /// mapped to `A` and then moves two rows back. The logical step is bounded
by the match; the physical step is
+ /// bounded only by the partition, so it may legally read a row outside the
match. Either step falling off its bound
+ /// yields `null`, as the standard requires.
+ final class Navigation implements MatchTerm {
+ private final int _symbolOrdinal;
+ private final boolean _fromEnd;
+ private final int _logicalOffset;
+ private final int _physicalDelta;
+ private final int _columnIndex;
+ private final ColumnDataType _resultType;
+ private final ColumnDataType _storedType;
+
+ /// @param symbolOrdinal pattern variable to navigate, or
+ ///
[org.apache.pinot.query.planner.logical.RexExpression.PatternFieldRef#UNIVERSAL_SYMBOL_ORDINAL]
+ /// for an unqualified column reference, which navigates the rows
of the whole match
+ /// @param fromEnd `true` for `LAST`, `false` for `FIRST`
+ /// @param logicalOffset how many rows back from the end (or forward from
the start) of the designated variable
+ /// @param physicalDelta rows to move in the partition afterwards;
negative for `PREV`, positive for
+ /// `NEXT`, zero when there is no physical step
+ /// @param columnIndex column to read, as an index into the input row of
the MATCH_RECOGNIZE node
+ public Navigation(int symbolOrdinal, boolean fromEnd, int logicalOffset,
int physicalDelta, int columnIndex,
+ ColumnDataType resultType) {
+ _symbolOrdinal = symbolOrdinal;
+ _fromEnd = fromEnd;
+ _logicalOffset = logicalOffset;
+ _physicalDelta = physicalDelta;
+ _columnIndex = columnIndex;
+ _resultType = resultType;
+ _storedType = resultType.getStoredType();
+ }
+
+ @Override
+ public ColumnDataType getResultType() {
+ return _resultType;
+ }
+
+ @Nullable
+ @Override
+ public Object evaluate(MatchTape tape) {
+ int rowIndex = _fromEnd ? tape.lastRow(_symbolOrdinal, _logicalOffset)
+ : tape.firstRow(_symbolOrdinal, _logicalOffset);
+ if (rowIndex == MatchTape.NO_ROW) {
+ return null;
+ }
+ rowIndex += _physicalDelta;
+ List<Object[]> rows = tape.getPartitionRows();
+ if (rowIndex < 0 || rowIndex >= rows.size()) {
+ return null;
+ }
+ Object value = rows.get(rowIndex)[_columnIndex];
+ // The declared type of the navigation may be wider than the column's
(e.g. BIG_DECIMAL for a DOUBLE column),
+ // and the enclosing operand compares against that declared type.
+ return value != null ? TypeUtils.convert(value, _storedType) : null;
+ }
+ }
+
+ /// `CLASSIFIER()`: the name of the pattern variable the designated row is
mapped to. With ONE ROW PER MATCH
+ /// the designated row is the last row of the match, which is also the
current row while a DEFINE predicate is being
+ /// evaluated.
+ final class Classifier implements MatchTerm {
+ public static final Classifier INSTANCE = new Classifier();
+
+ private Classifier() {
+ }
+
+ @Override
+ public ColumnDataType getResultType() {
+ return ColumnDataType.STRING;
+ }
+
+ @Nullable
+ @Override
+ public Object evaluate(MatchTape tape) {
+ return tape.classifierAt(tape.getEndPos() - 1);
+ }
+ }
+
+ /// `MATCH_NUMBER()`: the sequential number of the match within its
partition, starting at 1.
+ final class MatchNumber implements MatchTerm {
+ public static final MatchNumber INSTANCE = new MatchNumber();
+
+ private MatchNumber() {
+ }
+
+ @Override
+ public ColumnDataType getResultType() {
+ return ColumnDataType.LONG;
+ }
+
+ @Override
+ public Object evaluate(MatchTape tape) {
+ return tape.getMatchNumber();
+ }
+ }
+
+ /// A single variable aggregate in MEASURES, e.g. `SUM(A.price)` or
`COUNT(*)`: the aggregate of an
+ /// expression evaluated over every row of the match that is mapped to one
pattern variable.
+ ///
+ /// The argument is evaluated by an ordinary [TransformOperand] against the
raw input row, so any scalar
+ /// expression works, e.g. `SUM(A.price * A.quantity)`. Nulls are skipped,
as SQL requires; an aggregate over
+ /// zero rows is `null` except for `COUNT`, which is `0`.
+ final class Aggregate implements MatchTerm {
+ private final Kind _kind;
+ private final int _symbolOrdinal;
+ @Nullable
+ private final TransformOperand _argument;
+ private final ColumnDataType _resultType;
+ private final ColumnDataType _storedType;
+
+ /// @param argument the aggregated expression evaluated against an input
row, or `null` for `COUNT(*)`,
+ /// which counts rows rather than values
+ public Aggregate(Kind kind, int symbolOrdinal, @Nullable TransformOperand
argument, ColumnDataType resultType) {
+ _kind = kind;
+ _symbolOrdinal = symbolOrdinal;
+ _argument = argument;
+ _resultType = resultType;
+ _storedType = resultType.getStoredType();
+ }
+
+ @Override
+ public ColumnDataType getResultType() {
+ return _resultType;
+ }
+
+ @Nullable
+ @Override
+ public Object evaluate(MatchTape tape) {
+ IntArrayList rows = tape.rowsOf(_symbolOrdinal);
Review Comment:
`COUNT(*)` uses the universal symbol, so `rowsOf()` allocates and fills an
`IntArrayList` containing every matched row before this branch reads only its
size. That is O(match length) allocation per result and becomes quadratic with
overlapping matches. Please return `(long) tape.getLength()` before calling
`rowsOf()`.
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/MatchRecognizeIntegrationTest.java:
##########
@@ -0,0 +1,622 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.integration.tests.custom;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.File;
+import java.util.List;
+import org.apache.avro.file.DataFileWriter;
+import org.apache.avro.generic.GenericData;
+import org.apache.pinot.integration.tests.QueryAssert;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.builder.TableNameBuilder;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+
+
+/// End-to-end coverage of SQL:2016 `MATCH_RECOGNIZE` against a real Pinot
cluster (ZooKeeper + controller +
+/// broker + two servers), as opposed to the mock server enclosures used by
+/// `pinot-query-runtime/src/test/resources/queries/MatchRecognize.json`.
+///
+/// The fixture is a small stock ticker with five symbols of different
lengths, written round-robin into
+/// [#getNumAvroFiles()] Avro files, so every partition is spread over several
segments and both servers. That
+/// makes every assertion here also an assertion that the sort exchange
inserted by
+/// `PinotMatchExchangeNodeInsertRule` really does deliver each partition
contiguously and in ORDER BY order.
+///
+/// Every expected result set below is hand-computed from [#PRICES] rather
than captured from a run.
+@Test(suiteName = "CustomClusterIntegrationTest")
+public class MatchRecognizeIntegrationTest extends
CustomDataQueryClusterIntegrationTest {
+
+ private static final String DEFAULT_TABLE_NAME =
"MatchRecognizeIntegrationTest";
+ private static final String SYMBOL_COLUMN = "symbolCol";
+ private static final String SEQ_COLUMN = "seqCol";
+ private static final String PRICE_COLUMN = "priceCol";
+
+ /// Symbols in ascending order, which is also the order every query below
sorts its output by.
+ private static final String[] SYMBOLS = {"AAPL", "AMZN", "GOOG", "MSFT",
"NFLX"};
+
+ /// Prices per symbol, indexed like [#SYMBOLS]. `seqCol` is the 1-based
index within the symbol.
+ /// - AAPL: two disjoint V shapes, the second one overlapping the first
under SKIP TO NEXT ROW.
+ /// - AMZN: one long descent, so a single match covers most of the partition.
+ /// - GOOG: flat, so no strict rise or fall matches anywhere - a partition
that contributes nothing.
+ /// - MSFT: the minimal V shape.
+ /// - NFLX: strictly increasing, which is what separates greedy from
reluctant quantifiers.
+ private static final int[][] PRICES = {
+ {10, 8, 5, 9, 12, 7, 11},
+ {20, 15, 10, 5, 25},
+ {4, 4, 4},
+ {5, 3, 8},
+ {1, 2, 3, 4, 5, 6, 7, 8}
+ };
+
+ private static final String V_SHAPE_DEFINE =
+ " DEFINE DOWN AS DOWN.priceCol < PREV(DOWN.priceCol), UP AS UP.priceCol
> PREV(UP.priceCol)";
+
+ /// The canonical vendor-documentation V-shape query: a start row, a
strictly falling run, then a strictly rising run.
+ @Test(dataProvider = "useV2QueryEngine")
+ public void testCanonicalVShape(boolean useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+ String query = "SELECT * FROM " + getTableName() + " MATCH_RECOGNIZE ("
+ + " PARTITION BY symbolCol"
+ + " ORDER BY seqCol"
+ + " MEASURES MATCH_NUMBER() AS mno, STRT.seqCol AS start_seq,
LAST(DOWN.seqCol) AS bottom_seq,"
+ + " LAST(UP.seqCol) AS end_seq, LAST(DOWN.priceCol) AS bottom_price"
+ + " ONE ROW PER MATCH"
+ + " AFTER MATCH SKIP PAST LAST ROW"
+ + " PATTERN (STRT DOWN+ UP+)"
+ + V_SHAPE_DEFINE
+ + ") AS mr ORDER BY symbolCol, start_seq";
+ // STRT consumes one row, so the second AAPL V (seq 5..7) is unreachable:
the first match ends at seq 5 and
+ // SKIP PAST LAST ROW resumes at seq 6, leaving no row for STRT before the
fall at seq 6.
+ assertMatchRows(query, new Object[][]{
+ {"AAPL", 1, 1, 3, 5, 5},
+ {"AMZN", 1, 1, 4, 5, 5},
+ {"MSFT", 1, 1, 2, 3, 3}
+ });
+ }
+
+ /// THE critical default. SQL:2016, Trino, Snowflake and Oracle all default
an omitted AFTER MATCH clause to
+ /// SKIP PAST LAST ROW, while Calcite's `SqlToRelConverter` silently
substitutes SKIP TO NEXT ROW. The two
+ /// differ in whether matches may overlap, so a regression here changes
results without changing anything visible.
+ @Test(dataProvider = "useV2QueryEngine")
+ public void testOmittedAfterMatchIsSkipPastLastRow(boolean
useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+ Object[][] nonOverlapping = new Object[][]{
+ {"AAPL", 2, 8, 12},
+ {"AAPL", 6, 7, 11},
+ {"AMZN", 2, 15, 25},
+ {"MSFT", 2, 3, 8}
+ };
+ assertMatchRows(vShapeQuery("AFTER MATCH SKIP PAST LAST ROW"),
nonOverlapping);
+ assertMatchRows(vShapeQuery(""), nonOverlapping);
+
+ // Same rows plus the overlapping ones. An engine that kept Calcite's
default would return exactly this for the
+ // two queries above.
+ assertMatchRows(vShapeQuery("AFTER MATCH SKIP TO NEXT ROW"), new
Object[][]{
+ {"AAPL", 2, 8, 12},
+ {"AAPL", 3, 5, 12},
+ {"AAPL", 6, 7, 11},
+ {"AMZN", 2, 15, 25},
+ {"AMZN", 3, 10, 25},
+ {"AMZN", 4, 5, 25},
+ {"MSFT", 2, 3, 8}
+ });
+ }
+
+ /// All four skip modes over the same pattern, each producing a different
row set. `S{2}` pushes the first row
+ /// mapped to `U` two rows past the start of the match, which is what makes
SKIP TO FIRST U differ from
+ /// SKIP TO NEXT ROW.
+ @Test(dataProvider = "useV2QueryEngine")
+ public void testSkipToFirstAndLastOfPatternVariable(boolean
useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+ // NFLX (8 strictly increasing rows) admits a match at every start row
1..5; each mode consumes them differently.
+ assertMatchRows(skipTargetQuery("AFTER MATCH SKIP PAST LAST ROW"), new
Object[][]{
+ {"AAPL", 2, 5},
+ {"NFLX", 1, 4},
+ {"NFLX", 5, 8}
+ });
+ assertMatchRows(skipTargetQuery("AFTER MATCH SKIP TO NEXT ROW"), new
Object[][]{
+ {"AAPL", 2, 5},
+ {"NFLX", 1, 4},
+ {"NFLX", 2, 5},
+ {"NFLX", 3, 6},
+ {"NFLX", 4, 7},
+ {"NFLX", 5, 8}
+ });
+ // FIRST(U) is the third row of the match, so matching resumes two rows in.
+ assertMatchRows(skipTargetQuery("AFTER MATCH SKIP TO FIRST U"), new
Object[][]{
+ {"AAPL", 2, 5},
+ {"NFLX", 1, 4},
+ {"NFLX", 3, 6},
+ {"NFLX", 5, 8}
+ });
+ // LAST(U) is the fourth and last row of the match, so matching resumes
three rows in.
+ assertMatchRows(skipTargetQuery("AFTER MATCH SKIP TO LAST U"), new
Object[][]{
+ {"AAPL", 2, 5},
+ {"NFLX", 1, 4},
+ {"NFLX", 4, 7}
+ });
+ }
+
+ /// A greedy quantifier takes the longest run it can, a reluctant one the
shortest. On NFLX the difference is one
+ /// seven-row match versus four two-row matches.
+ @Test(dataProvider = "useV2QueryEngine")
+ public void testGreedyVersusReluctantQuantifier(boolean
useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+ assertMatchRows(quantifierQuery("+"), new Object[][]{
+ {"AAPL", 3, 2, 5},
+ {"AAPL", 6, 1, 7},
+ {"AMZN", 4, 1, 5},
+ {"MSFT", 2, 1, 3},
+ {"NFLX", 1, 7, 8}
+ });
+ assertMatchRows(quantifierQuery("+?"), new Object[][]{
+ {"AAPL", 3, 1, 4},
+ {"AAPL", 6, 1, 7},
+ {"AMZN", 4, 1, 5},
+ {"MSFT", 2, 1, 3},
+ {"NFLX", 1, 1, 2},
+ {"NFLX", 3, 1, 4},
+ {"NFLX", 5, 1, 6},
+ {"NFLX", 7, 1, 8}
+ });
+ }
+
+ /// Alternation prefers the leftmost branch that lets the whole pattern
complete, which CLASSIFIER() reports.
+ @Test(dataProvider = "useV2QueryEngine")
+ public void testAlternation(boolean useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+ String query = "SELECT * FROM " + getTableName() + " MATCH_RECOGNIZE ("
+ + " PARTITION BY symbolCol ORDER BY seqCol"
+ + " MEASURES FIRST(S.seqCol) AS start_seq, CLASSIFIER() AS cls"
+ + " ONE ROW PER MATCH AFTER MATCH SKIP PAST LAST ROW"
+ + " PATTERN (S (UP | DN))"
+ + " DEFINE UP AS UP.priceCol > PREV(UP.priceCol), DN AS DN.priceCol <
PREV(DN.priceCol)"
+ + ") AS mr ORDER BY symbolCol, start_seq";
+ // GOOG is flat, so neither branch can ever match and the partition
contributes nothing.
+ assertMatchRows(query, new Object[][]{
+ {"AAPL", 1, "DN"},
+ {"AAPL", 3, "UP"},
+ {"AAPL", 5, "DN"},
+ {"AMZN", 1, "DN"},
+ {"AMZN", 3, "DN"},
+ {"MSFT", 1, "DN"},
+ {"NFLX", 1, "UP"},
+ {"NFLX", 3, "UP"},
+ {"NFLX", 5, "UP"},
+ {"NFLX", 7, "UP"}
+ });
+ }
+
+ /// A bounded quantifier honours both bounds: AAPL can only supply the
minimum of two rows, NFLX could supply seven
+ /// but is capped at the maximum of three.
+ @Test(dataProvider = "useV2QueryEngine")
+ public void testBoundedQuantifier(boolean useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+ String query = "SELECT * FROM " + getTableName() + " MATCH_RECOGNIZE ("
+ + " PARTITION BY symbolCol ORDER BY seqCol"
+ + " MEASURES FIRST(S.seqCol) AS start_seq, COUNT(UP.priceCol) AS
up_count, LAST(UP.seqCol) AS end_seq"
+ + " ONE ROW PER MATCH AFTER MATCH SKIP PAST LAST ROW"
+ + " PATTERN (S UP{2,3})"
+ + " DEFINE UP AS UP.priceCol > PREV(UP.priceCol)"
+ + ") AS mr ORDER BY symbolCol, start_seq";
+ assertMatchRows(query, new Object[][]{
+ {"AAPL", 3, 2, 5},
+ {"NFLX", 1, 3, 4},
+ {"NFLX", 5, 3, 8}
+ });
+ }
+
+ /// MATCH_NUMBER() restarts at 1 in every partition, CLASSIFIER() reports
the label of the final row of the match,
+ /// and each single-variable aggregate sees only the rows bound to its own
pattern variable.
+ ///
+ /// COUNT is pinned here because `MatchTerm.Aggregate` used to route
`COUNT(<expr>)` through the
+ /// accumulator, which has no COUNT branch. AVG is pinned because an
integral result type would truncate 6.5 to 6.
+ @Test(dataProvider = "useV2QueryEngine")
+ public void testMatchNumberClassifierAndAggregates(boolean
useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+ String query = "SELECT * FROM " + getTableName() + " MATCH_RECOGNIZE ("
+ + " PARTITION BY symbolCol ORDER BY seqCol"
+ + " MEASURES MATCH_NUMBER() AS mno, CLASSIFIER() AS cls,
COUNT(DOWN.priceCol) AS down_count,"
+ + " COUNT(*) AS all_count, SUM(DOWN.priceCol) AS down_sum,
MIN(DOWN.priceCol) AS down_min,"
+ + " MAX(DOWN.priceCol) AS down_max, AVG(DOWN.priceCol) AS down_avg"
+ + " ONE ROW PER MATCH"
+ + " PATTERN (DOWN+ UP+)"
+ + V_SHAPE_DEFINE
+ + ") AS mr ORDER BY symbolCol, mno";
+ assertMatchRows(query, new Object[][]{
+ // AAPL match 1 maps prices 8 and 5 to DOWN, prices 9 and 12 to UP.
+ {"AAPL", 1, "UP", 2, 4, 13, 5, 8, 6.5d},
+ {"AAPL", 2, "UP", 1, 2, 7, 7, 7, 7.0d},
+ {"AMZN", 1, "UP", 3, 4, 30, 5, 15, 10.0d},
+ {"MSFT", 1, "UP", 1, 2, 3, 3, 3, 3.0d}
+ });
+ }
+
+ /// PREV and NEXT are bounded by the partition, not by the input: at a
partition boundary they must yield NULL rather
+ /// than the neighbouring partition's row.
+ ///
+ /// The navigations live in DEFINE rather than in MEASURES because Calcite's
`SqlValidatorImpl
+ /// .PatternValidator` unconditionally rejects PREV/NEXT inside a MEASURES
item (see
+ /// [#testDeferredConstructsAreRejected]), so DEFINE is the only place a
query can reach them.
+ ///
+ /// `PREV(...)` being NULL makes the whole predicate NULL, which SQL:2016
treats as "not matched". So the
+ /// proof that PREV stops at the partition start is that **no** row with
`seqCol = 1` appears in the first
+ /// result, and the proof that NEXT stops at the partition end is that no
partition's last row appears in the second
+ /// one - even though every one of those rows would satisfy the predicate
against a neighbouring partition's price.
+ @Test(dataProvider = "useV2QueryEngine")
+ public void testPrevAndNextAreBoundedByPartition(boolean
useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+ // Rows whose predecessor within the partition is cheaper. seqCol = 1 is
absent for every symbol.
+ assertMatchRows(neighbourQuery("PREV(A.priceCol) < A.priceCol"), new
Object[][]{
+ {"AAPL", 4, 9}, {"AAPL", 5, 12}, {"AAPL", 7, 11},
+ {"AMZN", 5, 25},
+ {"MSFT", 3, 8},
+ {"NFLX", 2, 2}, {"NFLX", 3, 3}, {"NFLX", 4, 4}, {"NFLX", 5, 5},
{"NFLX", 6, 6}, {"NFLX", 7, 7},
+ {"NFLX", 8, 8}
+ });
+ // Rows whose successor within the partition is dearer. The last row of
AAPL (7), AMZN (5), GOOG (3), MSFT (3)
+ // and NFLX (8) is absent from every one of them.
+ assertMatchRows(neighbourQuery("NEXT(A.priceCol) > A.priceCol"), new
Object[][]{
+ {"AAPL", 3, 5}, {"AAPL", 4, 9}, {"AAPL", 6, 7},
+ {"AMZN", 4, 5},
+ {"MSFT", 2, 3},
+ {"NFLX", 1, 1}, {"NFLX", 2, 2}, {"NFLX", 3, 3}, {"NFLX", 4, 4},
{"NFLX", 5, 5}, {"NFLX", 6, 6},
+ {"NFLX", 7, 7}
+ });
+
+ // And directly: the NULL is observable, and it happens exactly once per
partition on each side.
+ assertMatchRows(neighbourQuery("PREV(A.priceCol) IS NULL"), new Object[][]{
+ {"AAPL", 1, 10}, {"AMZN", 1, 20}, {"GOOG", 1, 4}, {"MSFT", 1, 5},
{"NFLX", 1, 1}
+ });
+ assertMatchRows(neighbourQuery("NEXT(A.priceCol) IS NULL"), new Object[][]{
+ {"AAPL", 7, 11}, {"AMZN", 5, 25}, {"GOOG", 3, 4}, {"MSFT", 3, 8},
{"NFLX", 8, 8}
+ });
+ }
+
+ /// PARTITION BY genuinely isolates matches. The `^` and `$` anchors are
relative to the partition, so
+ /// each must match exactly once per symbol; if partitioning leaked they
would match once for the whole table.
+ @Test(dataProvider = "useV2QueryEngine")
+ public void testPartitionIsolationAcrossSegments(boolean
useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+ // The fixture really is spread over several segments, so the exchange
below the match operator is exercised.
+ int numSegments = getSharedHelixResourceManager()
+
.getSegmentsFor(TableNameBuilder.OFFLINE.tableNameWithType(getTableName()),
false).size();
+ assertTrue(numSegments > 1, "Expected the fixture to span more than one
segment, but found " + numSegments);
+
+ assertMatchRows(anchoredQuery("^ A"), new Object[][]{
+ {"AAPL", 1, 10},
+ {"AMZN", 1, 20},
+ {"GOOG", 1, 4},
+ {"MSFT", 1, 5},
+ {"NFLX", 1, 1}
+ });
+ assertMatchRows(anchoredQuery("A $"), new Object[][]{
+ {"AAPL", 7, 11},
+ {"AMZN", 5, 25},
+ {"GOOG", 3, 4},
+ {"MSFT", 3, 8},
+ {"NFLX", 8, 8}
+ });
+
+ // A pattern that pairs up adjacent rows. Partition sizes are 7, 5, 3, 3
and 8, so the odd row at the end of AAPL,
+ // GOOG, MSFT and AMZN is dropped rather than paired with the next
partition's first row: 11 matches, not the 13
+ // that a single 26-row stream would produce.
+ String pairs = "SELECT * FROM " + getTableName() + " MATCH_RECOGNIZE ("
+ + " PARTITION BY symbolCol ORDER BY seqCol"
+ + " MEASURES FIRST(A.seqCol) AS start_seq, LAST(B.seqCol) AS end_seq"
+ + " ONE ROW PER MATCH AFTER MATCH SKIP PAST LAST ROW"
+ + " PATTERN (A B)"
+ + " DEFINE A AS A.priceCol > 0, B AS B.priceCol > 0"
+ + ") AS mr ORDER BY symbolCol, start_seq";
+ assertMatchRows(pairs, new Object[][]{
+ {"AAPL", 1, 2}, {"AAPL", 3, 4}, {"AAPL", 5, 6},
+ {"AMZN", 1, 2}, {"AMZN", 3, 4},
+ {"GOOG", 1, 2},
+ {"MSFT", 1, 2},
+ {"NFLX", 1, 2}, {"NFLX", 3, 4}, {"NFLX", 5, 6}, {"NFLX", 7, 8}
+ });
+ }
+
+ /// A multi-column PARTITION BY whose source order is not ascending
input-column order.
+ ///
+ /// The Pinot schema is a `TreeMap`, so the input columns are `priceCol(0),
seqCol(1), symbolCol(2)`
+ /// and `PARTITION BY symbolCol, priceCol` is therefore in **descending**
index order. Calcite's
+ /// `Match#getPartitionKeys()` is an `ImmutableBitSet`, which loses that
order, while the output row type
+ /// keeps it - so an engine that read the partition keys off the bit set
would write `priceCol`'s Integer into
+ /// the STRING slot and fail the whole query with a ClassCastException.
+ ///
+ /// GOOG is the load-bearing row: its three rows all cost 4, so they form
one three-row partition and
+ /// `COUNT(*)` reports 3. Every other (symbol, price) pair is unique and
yields a one-row partition.
+ @Test(dataProvider = "useV2QueryEngine")
+ public void testPartitionByKeepsItsSourceOrder(boolean
useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+ String query = "SELECT * FROM " + getTableName() + " MATCH_RECOGNIZE ("
+ + " PARTITION BY symbolCol, priceCol"
+ + " ORDER BY seqCol"
+ + " MEASURES COUNT(*) AS cnt"
+ + " ONE ROW PER MATCH AFTER MATCH SKIP PAST LAST ROW"
+ + " PATTERN (A+)"
+ + " DEFINE A AS A.priceCol > 0"
+ + ") AS mr ORDER BY symbolCol, priceCol";
+ assertMatchRows(query, new Object[][]{
+ {"AAPL", 5, 1}, {"AAPL", 7, 1}, {"AAPL", 8, 1}, {"AAPL", 9, 1},
{"AAPL", 10, 1}, {"AAPL", 11, 1},
+ {"AAPL", 12, 1},
+ {"AMZN", 5, 1}, {"AMZN", 10, 1}, {"AMZN", 15, 1}, {"AMZN", 20, 1},
{"AMZN", 25, 1},
+ {"GOOG", 4, 3},
+ {"MSFT", 3, 1}, {"MSFT", 5, 1}, {"MSFT", 8, 1},
+ {"NFLX", 1, 1}, {"NFLX", 2, 1}, {"NFLX", 3, 1}, {"NFLX", 4, 1},
{"NFLX", 5, 1}, {"NFLX", 6, 1},
+ {"NFLX", 7, 1}, {"NFLX", 8, 1}
+ });
+ }
+
+ /// Every deferred construct is rejected during planning with a message that
names it, rather than producing a wrong
+ /// result or an internal Calcite failure.
+ @Test(dataProvider = "useV2QueryEngine")
+ public void testDeferredConstructsAreRejected(boolean
useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+ assertPlanningError("SELECT * FROM " + getTableName() + " MATCH_RECOGNIZE
("
+ + " PARTITION BY symbolCol ORDER BY seqCol"
+ + " MEASURES CLASSIFIER() AS cls"
+ + " ALL ROWS PER MATCH"
+ + " PATTERN (DOWN+ UP+)"
+ + V_SHAPE_DEFINE
+ + ") AS mr", "ALL ROWS PER MATCH is not supported yet");
+
+ assertPlanningError("SELECT * FROM " + getTableName() + " MATCH_RECOGNIZE
("
+ + " PARTITION BY symbolCol ORDER BY seqCol"
+ + " MEASURES LAST(DOWN.priceCol) AS p"
+ + " ONE ROW PER MATCH"
+ + " PATTERN (DOWN+ UP+)"
+ + " SUBSET BOTH = (DOWN, UP)"
+ + V_SHAPE_DEFINE
+ + ") AS mr", "SUBSET is not supported yet");
+
+ assertPlanningError("SELECT * FROM " + getTableName() + " MATCH_RECOGNIZE
("
+ + " PARTITION BY symbolCol"
+ + " MEASURES LAST(DOWN.priceCol) AS p"
+ + " ONE ROW PER MATCH"
+ + " PATTERN (DOWN+ UP+)"
+ + V_SHAPE_DEFINE
+ + ") AS mr", "MATCH_RECOGNIZE requires an ORDER BY clause");
+
+ assertPlanningError("SELECT * FROM " + getTableName() + " MATCH_RECOGNIZE
("
+ + " PARTITION BY symbolCol ORDER BY seqCol"
+ + " MEASURES LAST(DOWN.priceCol) AS p"
+ + " ONE ROW PER MATCH"
+ + " PATTERN (DOWN+ UP+)"
+ + " DEFINE DOWN AS COUNT(DOWN.priceCol) < 3, UP AS UP.priceCol >
PREV(UP.priceCol)"
+ + ") AS mr", "is not supported yet in the MATCH_RECOGNIZE DEFINE
clause");
+
+ // PERMUTE is a non-reserved keyword, so the single argument form parses
as a concatenation of an undefined
+ // pattern variable named PERMUTE and a group. MatchRecognizeValidator
turns that silently wrong plan into an
+ // explicit error.
+ assertPlanningError("SELECT * FROM " + getTableName() + " MATCH_RECOGNIZE
("
+ + " PARTITION BY symbolCol ORDER BY seqCol"
+ + " MEASURES LAST(DOWN.priceCol) AS p"
+ + " ONE ROW PER MATCH"
+ + " PATTERN (PERMUTE(DOWN))"
+ + " DEFINE DOWN AS DOWN.priceCol < PREV(DOWN.priceCol)"
+ + ") AS mr", "PERMUTE is not supported yet");
+
+ // Calcite's SqlValidatorImpl.PatternValidator rejects PREV/NEXT anywhere
inside a MEASURES item, however they are
+ // nested. SQL:2016, Oracle and Trino all allow them there, and
MatchTerm.Navigation implements them, but no query
+ // can reach that path: physical navigation is only usable from DEFINE.
Pinned here so the day the restriction is
+ // lifted or wrapped in a Pinot specific message, this test says so.
+ assertPlanningError("SELECT * FROM " + getTableName() + " MATCH_RECOGNIZE
("
+ + " PARTITION BY symbolCol ORDER BY seqCol"
+ + " MEASURES PREV(A.priceCol) AS prev_price"
+ + " ONE ROW PER MATCH"
+ + " PATTERN (A)"
+ + " DEFINE A AS A.priceCol > 0"
+ + ") AS mr", "Cannot use PREV/NEXT in MEASURE");
+
+ // MEASURES is optional in SQL:2016, but Calcite then makes the
MATCH_RECOGNIZE row type the whole *input* row
+ // type instead of the ONE ROW PER MATCH shape, so the query used to plan
and then die on the server.
+ assertPlanningError("SELECT * FROM " + getTableName() + " MATCH_RECOGNIZE
("
+ + " PARTITION BY symbolCol ORDER BY seqCol"
+ + " ONE ROW PER MATCH"
+ + " PATTERN (DOWN+ UP+)"
+ + V_SHAPE_DEFINE
+ + ") AS mr", "MATCH_RECOGNIZE requires a MEASURES clause");
+
+ // Calcite adds a measure to the row type only when its alias is still
free, so a measure aliased to a PARTITION
+ // BY column silently disappears from the output.
+ assertPlanningError("SELECT * FROM " + getTableName() + " MATCH_RECOGNIZE
("
+ + " PARTITION BY symbolCol ORDER BY seqCol"
+ + " MEASURES LAST(DOWN.priceCol) AS symbolCol"
+ + " ONE ROW PER MATCH"
+ + " PATTERN (DOWN+ UP+)"
+ + V_SHAPE_DEFINE
+ + ") AS mr", "collides with a PARTITION BY column or an earlier
measure alias");
+
+ // Calcite qualifies an unqualified column reference with the row source
alias, so a pattern variable of the same
+ // name steals every unqualified reference from the SQL:2016 universal row
pattern variable - silently changing
+ // both measure values and which rows match.
+ assertPlanningError("SELECT * FROM " + getTableName() + " AS DOWN
MATCH_RECOGNIZE ("
+ + " PARTITION BY symbolCol ORDER BY seqCol"
+ + " MEASURES LAST(priceCol) AS p"
+ + " ONE ROW PER MATCH"
+ + " PATTERN (DOWN+ UP+)"
+ + V_SHAPE_DEFINE
+ + ")", "collides with the row source alias");
+ }
+
+ private String neighbourQuery(String definition) {
+ return "SELECT * FROM " + getTableName() + " MATCH_RECOGNIZE ("
+ + " PARTITION BY symbolCol ORDER BY seqCol"
+ + " MEASURES LAST(A.seqCol) AS seq, LAST(A.priceCol) AS cur_price"
+ + " ONE ROW PER MATCH AFTER MATCH SKIP PAST LAST ROW"
+ + " PATTERN (A)"
+ + " DEFINE A AS " + definition
+ + ") AS mr ORDER BY symbolCol, seq";
+ }
+
+ private String anchoredQuery(String pattern) {
+ return "SELECT * FROM " + getTableName() + " MATCH_RECOGNIZE ("
+ + " PARTITION BY symbolCol ORDER BY seqCol"
+ + " MEASURES LAST(A.seqCol) AS seq, LAST(A.priceCol) AS cur_price"
+ + " ONE ROW PER MATCH AFTER MATCH SKIP PAST LAST ROW"
+ + " PATTERN (" + pattern + ")"
+ + " DEFINE A AS A.priceCol > 0"
+ + ") AS mr ORDER BY symbolCol";
+ }
+
+ private String vShapeQuery(String afterMatch) {
+ return "SELECT * FROM " + getTableName() + " MATCH_RECOGNIZE ("
+ + " PARTITION BY symbolCol ORDER BY seqCol"
+ + " MEASURES FIRST(DOWN.seqCol) AS start_seq, FIRST(DOWN.priceCol) AS
start_price,"
+ + " LAST(UP.priceCol) AS end_price"
+ + " ONE ROW PER MATCH " + afterMatch
+ + " PATTERN (DOWN+ UP+)"
+ + V_SHAPE_DEFINE
+ + ") AS mr ORDER BY symbolCol, start_seq";
+ }
+
+ private String skipTargetQuery(String afterMatch) {
+ return "SELECT * FROM " + getTableName() + " MATCH_RECOGNIZE ("
+ + " PARTITION BY symbolCol ORDER BY seqCol"
+ + " MEASURES FIRST(S.seqCol) AS start_seq, LAST(U.seqCol) AS end_seq"
+ + " ONE ROW PER MATCH " + afterMatch
+ + " PATTERN (S{2} U{2})"
+ + " DEFINE U AS U.priceCol > PREV(U.priceCol)"
+ + ") AS mr ORDER BY symbolCol, start_seq";
+ }
+
+ private String quantifierQuery(String quantifier) {
+ return "SELECT * FROM " + getTableName() + " MATCH_RECOGNIZE ("
+ + " PARTITION BY symbolCol ORDER BY seqCol"
+ + " MEASURES FIRST(S.seqCol) AS start_seq, COUNT(U.priceCol) AS
u_count, LAST(U.seqCol) AS end_seq"
+ + " ONE ROW PER MATCH AFTER MATCH SKIP PAST LAST ROW"
+ + " PATTERN (S U" + quantifier + ")"
+ + " DEFINE U AS U.priceCol > PREV(U.priceCol)"
+ + ") AS mr ORDER BY symbolCol, start_seq";
+ }
+
+ /// Asserts the full result set of `query`, cell by cell. `null` expects a
SQL NULL, a [String]
+ /// expects a string, a [Double] expects an approximate double and any other
[Number] an exact integer.
+ private void assertMatchRows(String query, Object[][] expected)
+ throws Exception {
+ JsonNode response = postQuery(query);
+ QueryAssert.assertThat(response).hasNoExceptions();
+ JsonNode rows = response.get("resultTable").get("rows");
+ assertNotNull(rows, "No rows in response for query: " + query);
+ String context = "\nQuery: " + query + "\nRows: " + rows;
+ assertEquals(rows.size(), expected.length, "Unexpected number of matches."
+ context);
+ for (int i = 0; i < expected.length; i++) {
+ JsonNode row = rows.get(i);
+ Object[] expectedRow = expected[i];
+ assertEquals(row.size(), expectedRow.length, "Unexpected number of
columns in row " + i + "." + context);
+ for (int j = 0; j < expectedRow.length; j++) {
+ Object expectedValue = expectedRow[j];
+ JsonNode actual = row.get(j);
+ String where = "Mismatch at row " + i + ", column " + j + "." +
context;
+ if (expectedValue == null) {
+ assertTrue(actual.isNull(), where + "\nExpected NULL but got: " +
actual);
+ } else if (expectedValue instanceof String) {
+ assertEquals(actual.asText(), expectedValue, where);
+ } else if (expectedValue instanceof Double) {
+ assertEquals(actual.asDouble(), (double) (Double) expectedValue,
1e-9, where);
+ } else {
+ assertEquals(actual.asLong(), ((Number) expectedValue).longValue(),
where);
+ }
+ }
+ }
+ }
+
+ private void assertPlanningError(String query, String expectedMessage)
+ throws Exception {
+
QueryAssert.assertThat(postQuery(query)).firstException().containsMessage(expectedMessage);
+ }
+
+ @Override
+ public String getTableName() {
+ return DEFAULT_TABLE_NAME;
+ }
+
+ @Override
+ public Schema createSchema() {
+ return new Schema.SchemaBuilder().setSchemaName(getTableName())
+ .addSingleValueDimension(SYMBOL_COLUMN, FieldSpec.DataType.STRING)
+ .addSingleValueDimension(SEQ_COLUMN, FieldSpec.DataType.INT)
+ .addSingleValueDimension(PRICE_COLUMN, FieldSpec.DataType.INT)
+ .build();
+ }
+
+ @Override
+ public int getNumAvroFiles() {
+ return 3;
+ }
+
+ @Override
+ protected long getCountStarResult() {
+ int total = 0;
+ for (int[] prices : PRICES) {
+ total += prices.length;
+ }
+ return total;
+ }
+
+ @Override
+ public List<File> createAvroFiles()
+ throws Exception {
+ org.apache.avro.Schema avroSchema =
org.apache.avro.Schema.createRecord("myRecord", null, null, false);
+ avroSchema.setFields(List.of(
+ new org.apache.avro.Schema.Field(SYMBOL_COLUMN,
+ org.apache.avro.Schema.create(org.apache.avro.Schema.Type.STRING),
null, null),
+ new org.apache.avro.Schema.Field(SEQ_COLUMN,
+ org.apache.avro.Schema.create(org.apache.avro.Schema.Type.INT),
null, null),
+ new org.apache.avro.Schema.Field(PRICE_COLUMN,
Review Comment:
All fixture fields are non-null Avro primitives, so the PREV/NEXT `IS NULL`
cases only test navigation outside a partition, not source NULLs. Source NULL
behavior differs materially: with null handling enabled, DEFINE sees UNKNOWN
and aggregates skip the value; when disabled, the stored INT default is
evaluated and counted. Please ingest a nullable `priceCol` with column-based
null handling and run a focused DEFINE plus `COUNT(expr)`/all-null aggregate
query under both `SET enableNullHandling=true` and `false`.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/match/MatchTerm.java:
##########
@@ -0,0 +1,285 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.query.runtime.operator.match;
+
+import it.unimi.dsi.fastutil.ints.IntArrayList;
+import java.math.BigDecimal;
+import java.math.MathContext;
+import java.util.Arrays;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
+import org.apache.pinot.query.runtime.operator.operands.TransformOperand;
+import org.apache.pinot.query.runtime.operator.utils.TypeUtils;
+import org.apache.pinot.spi.exception.QueryErrorCode;
+
+
+/// A leaf of a MEASURES or DEFINE expression whose value depends on the match
rather than on a single row: a row
+/// pattern navigation, `CLASSIFIER()`, `MATCH_NUMBER()`, or a single variable
aggregate.
+///
+/// [MatchExpression] replaces every such leaf with a slot in a synthetic row,
so that everything above the
+/// leaves - comparisons, boolean connectives, arithmetic, scalar functions -
is evaluated by Pinot's ordinary
+/// [TransformOperand] machinery instead of a second expression interpreter.
+///
+/// Implementations are stateless with respect to the match: all match state
is passed in through the
+/// [MatchTape], so one instance is reused across every match of every
partition.
+public interface MatchTerm {
+
+ /// The type this term reports to the enclosing expression. Values returned
by [#evaluate] are in the
+ /// corresponding [stored][ColumnDataType#getStoredType()] representation,
exactly like the values of a real
+ /// input row.
+ ColumnDataType getResultType();
+
+ @Nullable
+ Object evaluate(MatchTape tape);
+
+ /// A row pattern navigation: an optional logical step (`FIRST` / `LAST`)
that designates a row of the
+ /// match, followed by an optional physical step (`PREV` / `NEXT`) that
moves a fixed number of rows
+ /// relative to it, and finally a column read.
+ ///
+ /// Both steps are needed because SQL:2016 nests them, e.g.
`PREV(LAST(A.price), 2)` designates the last row
+ /// mapped to `A` and then moves two rows back. The logical step is bounded
by the match; the physical step is
+ /// bounded only by the partition, so it may legally read a row outside the
match. Either step falling off its bound
+ /// yields `null`, as the standard requires.
+ final class Navigation implements MatchTerm {
+ private final int _symbolOrdinal;
+ private final boolean _fromEnd;
+ private final int _logicalOffset;
+ private final int _physicalDelta;
+ private final int _columnIndex;
+ private final ColumnDataType _resultType;
+ private final ColumnDataType _storedType;
+
+ /// @param symbolOrdinal pattern variable to navigate, or
+ ///
[org.apache.pinot.query.planner.logical.RexExpression.PatternFieldRef#UNIVERSAL_SYMBOL_ORDINAL]
+ /// for an unqualified column reference, which navigates the rows
of the whole match
+ /// @param fromEnd `true` for `LAST`, `false` for `FIRST`
+ /// @param logicalOffset how many rows back from the end (or forward from
the start) of the designated variable
+ /// @param physicalDelta rows to move in the partition afterwards;
negative for `PREV`, positive for
+ /// `NEXT`, zero when there is no physical step
+ /// @param columnIndex column to read, as an index into the input row of
the MATCH_RECOGNIZE node
+ public Navigation(int symbolOrdinal, boolean fromEnd, int logicalOffset,
int physicalDelta, int columnIndex,
+ ColumnDataType resultType) {
+ _symbolOrdinal = symbolOrdinal;
+ _fromEnd = fromEnd;
+ _logicalOffset = logicalOffset;
+ _physicalDelta = physicalDelta;
+ _columnIndex = columnIndex;
+ _resultType = resultType;
+ _storedType = resultType.getStoredType();
+ }
+
+ @Override
+ public ColumnDataType getResultType() {
+ return _resultType;
+ }
+
+ @Nullable
+ @Override
+ public Object evaluate(MatchTape tape) {
+ int rowIndex = _fromEnd ? tape.lastRow(_symbolOrdinal, _logicalOffset)
+ : tape.firstRow(_symbolOrdinal, _logicalOffset);
+ if (rowIndex == MatchTape.NO_ROW) {
+ return null;
+ }
+ rowIndex += _physicalDelta;
+ List<Object[]> rows = tape.getPartitionRows();
+ if (rowIndex < 0 || rowIndex >= rows.size()) {
+ return null;
+ }
+ Object value = rows.get(rowIndex)[_columnIndex];
+ // The declared type of the navigation may be wider than the column's
(e.g. BIG_DECIMAL for a DOUBLE column),
+ // and the enclosing operand compares against that declared type.
+ return value != null ? TypeUtils.convert(value, _storedType) : null;
+ }
+ }
+
+ /// `CLASSIFIER()`: the name of the pattern variable the designated row is
mapped to. With ONE ROW PER MATCH
+ /// the designated row is the last row of the match, which is also the
current row while a DEFINE predicate is being
+ /// evaluated.
+ final class Classifier implements MatchTerm {
+ public static final Classifier INSTANCE = new Classifier();
+
+ private Classifier() {
+ }
+
+ @Override
+ public ColumnDataType getResultType() {
+ return ColumnDataType.STRING;
+ }
+
+ @Nullable
+ @Override
+ public Object evaluate(MatchTape tape) {
+ return tape.classifierAt(tape.getEndPos() - 1);
+ }
+ }
+
+ /// `MATCH_NUMBER()`: the sequential number of the match within its
partition, starting at 1.
+ final class MatchNumber implements MatchTerm {
+ public static final MatchNumber INSTANCE = new MatchNumber();
+
+ private MatchNumber() {
+ }
+
+ @Override
+ public ColumnDataType getResultType() {
+ return ColumnDataType.LONG;
+ }
+
+ @Override
+ public Object evaluate(MatchTape tape) {
+ return tape.getMatchNumber();
+ }
+ }
+
+ /// A single variable aggregate in MEASURES, e.g. `SUM(A.price)` or
`COUNT(*)`: the aggregate of an
+ /// expression evaluated over every row of the match that is mapped to one
pattern variable.
+ ///
+ /// The argument is evaluated by an ordinary [TransformOperand] against the
raw input row, so any scalar
+ /// expression works, e.g. `SUM(A.price * A.quantity)`. Nulls are skipped,
as SQL requires; an aggregate over
+ /// zero rows is `null` except for `COUNT`, which is `0`.
+ final class Aggregate implements MatchTerm {
+ private final Kind _kind;
+ private final int _symbolOrdinal;
+ @Nullable
+ private final TransformOperand _argument;
+ private final ColumnDataType _resultType;
+ private final ColumnDataType _storedType;
+
+ /// @param argument the aggregated expression evaluated against an input
row, or `null` for `COUNT(*)`,
+ /// which counts rows rather than values
+ public Aggregate(Kind kind, int symbolOrdinal, @Nullable TransformOperand
argument, ColumnDataType resultType) {
+ _kind = kind;
+ _symbolOrdinal = symbolOrdinal;
+ _argument = argument;
+ _resultType = resultType;
+ _storedType = resultType.getStoredType();
+ }
+
+ @Override
+ public ColumnDataType getResultType() {
+ return _resultType;
+ }
+
+ @Nullable
+ @Override
+ public Object evaluate(MatchTape tape) {
+ IntArrayList rows = tape.rowsOf(_symbolOrdinal);
+ List<Object[]> partitionRows = tape.getPartitionRows();
+ if (_kind == Kind.COUNT && _argument == null) {
+ return (long) rows.size();
+ }
+ long count = 0;
+ Object accumulator = null;
+ for (int i = 0; i < rows.size(); i++) {
+ Object value = _argument.apply(partitionRows.get(rows.getInt(i)));
+ if (value == null) {
+ continue;
+ }
+ count++;
+ // COUNT only needs the tally; accumulate() has no COUNT case and
would throw.
+ if (_kind != Kind.COUNT) {
+ accumulator = accumulate(accumulator, value);
+ }
+ }
+ if (_kind == Kind.COUNT) {
+ return count;
+ }
+ if (count == 0) {
+ return null;
+ }
+ Object result = _kind == Kind.AVG ? divide(accumulator, count) :
accumulator;
+ return TypeUtils.convert(result, _storedType);
+ }
+
+ private Object accumulate(@Nullable Object accumulator, Object value) {
+ switch (_kind) {
+ case MIN:
+ return accumulator == null || compare(value, accumulator) < 0 ?
value : accumulator;
+ case MAX:
+ return accumulator == null || compare(value, accumulator) > 0 ?
value : accumulator;
+ case SUM:
+ case AVG:
+ return add(accumulator, value);
+ default:
+ throw new IllegalStateException("Unexpected MATCH_RECOGNIZE
aggregate: " + _kind);
+ }
+ }
+
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private static int compare(Object left, Object right) {
+ return ((Comparable) left).compareTo(right);
+ }
+
+ /// Accumulates without losing precision: exactly for `BIG_DECIMAL` and
for the integral types, and in
+ /// `double` otherwise, which is what the declared result type of the
aggregate already implies.
+ private Object add(@Nullable Object accumulator, Object value) {
+ switch (_storedType) {
Review Comment:
This switch implements distinct LONG, DOUBLE, and BIG_DECIMAL arithmetic,
but the tests only reach LONG SUM and DOUBLE AVG; the BIG_DECIMAL accumulation
and DECIMAL128 division path is untested. Please add focused SUM/AVG coverage
using a high-precision BIG_DECIMAL value that would expose accidental double
conversion.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/match/MatchLimits.java:
##########
@@ -0,0 +1,125 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.query.runtime.operator.match;
+
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.pinot.query.planner.plannode.PlanNode;
+
+
+/// Resolution of the two MATCH_RECOGNIZE resource limits.
+///
+/// ## Both limits throw, they never truncate
+///
+/// Pattern recognition has no meaningful partial answer: dropping a match, or
cutting a match short, returns rows that
+/// look plausible but are wrong, and nothing in the response says so. Unlike
the window operator, which offers a
+/// `BREAK` overflow mode, both limits here can only raise an error.
+///
+/// ## Resolution order
+///
+/// Highest precedence first, mirroring `maxRowsInWindow`:
+/// 1. the per node hint, read from the `matchOptions` hint of the plan node;
+/// 2. the query option, e.g. `SET maxRowsInMatch = 50000`;
+/// 3. the server cluster config, e.g. `pinot.query.match.max.rows`, which
`QueryRunner` folds into the
+/// op chain metadata under the query option key when the query did not set
one;
+/// 4. the default declared here.
+///
+/// The hint tier exists so a planner rule can pin a limit on an individual
MATCH_RECOGNIZE node; no SQL syntax
+/// attaches such a hint yet, so in practice the query option is the highest
tier a user can reach today.
+public final class MatchLimits {
+ private MatchLimits() {
+ }
+
+ /// Hint namespace on the MATCH_RECOGNIZE plan node.
+ public static final String MATCH_HINT_OPTIONS = "matchOptions";
+ /// Hint key for [#MAX_ROWS_IN_MATCH].
+ public static final String MAX_ROWS_IN_MATCH_HINT = "max_rows_in_match";
+ /// Hint key for [#MAX_STEPS_PER_MATCH_ATTEMPT].
+ public static final String MAX_STEPS_PER_MATCH_ATTEMPT_HINT =
"max_steps_per_match_attempt";
+
+ /// Query option capping the number of rows buffered for a single
MATCH_RECOGNIZE partition. A match cannot span
+ /// partitions, so this also caps the number of rows in one match.
+ public static final String MAX_ROWS_IN_MATCH = "maxRowsInMatch";
Review Comment:
Please declare both MATCH query-option keys in
`CommonConstants.Broker.Request.QueryOptionKey`. `QueryOptionsUtils` derives
its SQL validation allow-list and case-insensitive canonicalization map from
that class. As written, the documented `SET maxRowsInMatch=...` is
warned/rejected in WARN/REJECT mode, and a case variant remains noncanonical,
allowing `QueryRunner` to inject the server default under the exact key and
override the query value. Please add SQL validation, case-normalization, and
query-over-server precedence tests as well.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/match/MatchExpression.java:
##########
@@ -0,0 +1,379 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.query.runtime.operator.match;
+
+import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
+import org.apache.pinot.query.planner.logical.RexExpression;
+import org.apache.pinot.query.runtime.operator.operands.TransformOperand;
+import
org.apache.pinot.query.runtime.operator.operands.TransformOperandFactory;
+import org.apache.pinot.segment.spi.AggregationFunctionType;
+import org.apache.pinot.spi.exception.QueryErrorCode;
+
+
+/// A compiled MEASURES item or DEFINE predicate, evaluated against the state
of one match.
+///
+/// ## How it reuses Pinot's row expression evaluation
+///
+/// A MATCH_RECOGNIZE expression such as `PREV(A.price, 1) > B.price * 2` is
only *partly* match specific:
+/// the navigations `PREV(A.price, 1)` and `B.price` depend on the match, but
the `>` and the
+/// `*` are ordinary scalar operators. Compilation therefore splits the
expression in two:
+/// 1. every maximal match specific sub-expression - a navigation,
`CLASSIFIER()`, `MATCH_NUMBER()` or
+/// a single variable aggregate - becomes a [MatchTerm] bound to a slot of
a synthetic row;
+/// 2. what is left is a plain [RexExpression] over those slots, compiled by
+/// [TransformOperandFactory] exactly like any other Pinot row expression.
+///
+/// The example compiles to slots `[PREV(A.price,1), B.price]` plus the
operand `$0 > $1 * 2`. Comparisons,
+/// boolean connectives, arithmetic, casts, null semantics and every scalar
UDF therefore behave exactly as they do
+/// elsewhere in the multi-stage engine, and this class does not reimplement
any of them.
+///
+/// `RUNNING` and `FINAL` are stripped during compilation: with ONE ROW PER
MATCH the measures are
+/// computed once the match is complete, and running semantics evaluated at
the last row of a match coincide with final
+/// semantics. Both modifiers therefore have no effect and are unwrapped
rather than rejected.
+///
+/// Not thread safe: the slot array is reused between evaluations, so one
instance belongs to one operator.
+public class MatchExpression {
+ /// Navigation and match state functions, as Calcite names them in the
converted expression.
+ private static final String PREV = "PREV";
+ private static final String NEXT = "NEXT";
+ private static final String FIRST = "FIRST";
+ private static final String LAST = "LAST";
+ private static final String CLASSIFIER = "CLASSIFIER";
+ private static final String MATCH_NUMBER = "MATCH_NUMBER";
+ private static final String RUNNING = "RUNNING";
+ private static final String FINAL = "FINAL";
+ private static final Set<String> NAVIGATION_FUNCTIONS = Set.of(PREV, NEXT,
FIRST, LAST);
+ private static final Set<String> MATCH_FUNCTIONS =
+ Set.of(PREV, NEXT, FIRST, LAST, CLASSIFIER, MATCH_NUMBER, RUNNING,
FINAL);
+
+ private final TransformOperand _operand;
+ private final List<MatchTerm> _terms;
+ private final Object[] _slots;
+
+ private MatchExpression(TransformOperand operand, List<MatchTerm> terms) {
+ _operand = operand;
+ _terms = terms;
+ _slots = new Object[terms.size()];
+ }
+
+ /// Compiles `expression`, which addresses columns of `inputSchema` through
pattern field references.
+ ///
+ /// @throws org.apache.pinot.spi.exception.QueryException if the expression
uses a construct that is not supported
+ /// yet. Nothing is dropped or approximated, because either would
return wrong rows instead of an error.
+ public static MatchExpression compile(RexExpression expression, DataSchema
inputSchema) {
+ List<MatchTerm> terms = new ArrayList<>();
+ RexExpression slotExpression = rewrite(expression, inputSchema, terms);
+ int numTerms = terms.size();
+ String[] slotNames = new String[numTerms];
+ ColumnDataType[] slotTypes = new ColumnDataType[numTerms];
+ for (int i = 0; i < numTerms; i++) {
+ slotNames[i] = "$" + i;
+ slotTypes[i] = terms.get(i).getResultType();
+ }
+ TransformOperand operand =
+ TransformOperandFactory.getTransformOperand(slotExpression, new
DataSchema(slotNames, slotTypes));
+ return new MatchExpression(operand, terms);
+ }
+
+ /// Evaluates this expression against the current state of `tape`.
+ @Nullable
+ public Object evaluate(MatchTape tape) {
+ for (int i = 0; i < _terms.size(); i++) {
+ _slots[i] = _terms.get(i).evaluate(tape);
+ }
+ return _operand.apply(_slots);
+ }
+
+ /// Evaluates this expression as a DEFINE predicate. SQL three valued logic
collapses to false here: a row is mapped
+ /// to a pattern variable only if its condition is definitely true.
+ public boolean test(MatchTape tape) {
+ Object value = evaluate(tape);
+ if (value == null) {
+ return false;
+ }
+ if (value instanceof Boolean) {
+ return (Boolean) value;
+ }
+ return ((Number) value).intValue() != 0;
+ }
+
+ /// Replaces every match specific sub-expression of `expression` with an
input reference to a freshly appended
+ /// slot of `terms`, and returns the resulting expression over the synthetic
slot row.
+ private static RexExpression rewrite(RexExpression expression, DataSchema
inputSchema, List<MatchTerm> terms) {
+ if (expression instanceof RexExpression.PatternFieldRef) {
+ // A bare column reference such as `A.price` is `LAST(A.price, 0)` per
SQL:2016.
+ RexExpression.PatternFieldRef ref = (RexExpression.PatternFieldRef)
expression;
+ return addTerm(terms, new MatchTerm.Navigation(ref.getSymbolOrdinal(),
true, 0, 0, ref.getIndex(),
+ columnType(inputSchema, ref)));
+ }
+ if (expression instanceof RexExpression.InputRef) {
+ // Column references inside MEASURES / DEFINE always arrive as pattern
field references. A plain input
+ // reference would index the synthetic slot row instead of the input
row, so refuse it rather than read the
+ // wrong column.
+ throw QueryErrorCode.QUERY_EXECUTION.asException(
+ "Unsupported column reference in a MATCH_RECOGNIZE MEASURES or
DEFINE expression: " + expression
+ + ". Qualify the column with a pattern variable, e.g.
'A.column'.");
+ }
+ if (!(expression instanceof RexExpression.FunctionCall)) {
+ return expression;
+ }
+
+ RexExpression.FunctionCall call = (RexExpression.FunctionCall) expression;
+ String functionName = call.getFunctionName();
+ if (RUNNING.equals(functionName) || FINAL.equals(functionName)) {
+ return rewrite(singleOperand(call), inputSchema, terms);
+ }
+ if (CLASSIFIER.equals(functionName)) {
+ if (!call.getFunctionOperands().isEmpty()) {
+ throw QueryErrorCode.QUERY_EXECUTION.asException(
+ "CLASSIFIER with a pattern variable argument is not supported yet
in MATCH_RECOGNIZE. Use the no "
+ + "argument form CLASSIFIER().");
+ }
+ return addTerm(terms, MatchTerm.Classifier.INSTANCE);
+ }
+ if (MATCH_NUMBER.equals(functionName)) {
+ return addTerm(terms, MatchTerm.MatchNumber.INSTANCE);
+ }
+ if (NAVIGATION_FUNCTIONS.contains(functionName)) {
+ // Calcite distributes navigation over an expression, so `LAST(A.price *
2)` arrives as
+ // `LAST(A.price, 0) * LAST(2, 0)`. Navigating a constant is the
constant itself.
+ if (!containsPatternFieldRef(call)) {
+ return rewrite(singleOperand(call), inputSchema, terms);
+ }
+ return addTerm(terms, navigation(call));
+ }
+ if (AggregationFunctionType.isAggregationFunction(functionName)) {
+ return addTerm(terms, aggregate(call, inputSchema));
+ }
+
+ List<RexExpression> operands = call.getFunctionOperands();
+ List<RexExpression> rewritten = new ArrayList<>(operands.size());
+ for (RexExpression operand : operands) {
+ rewritten.add(rewrite(operand, inputSchema, terms));
+ }
+ return new RexExpression.FunctionCall(call.getDataType(), functionName,
rewritten, call.isDistinct(),
+ call.isIgnoreNulls());
+ }
+
+ private static RexExpression addTerm(List<MatchTerm> terms, MatchTerm term) {
+ terms.add(term);
+ return new RexExpression.InputRef(terms.size() - 1);
+ }
+
+ /// Flattens a nest of navigation calls such as `PREV(LAST(A.price, 1), 2)`
into a single
+ /// [MatchTerm.Navigation]: at most one logical step (`FIRST` / `LAST`)
selects a row of the match,
+ /// and the `PREV` / `NEXT` offsets around it add up into one physical delta.
+ private static MatchTerm.Navigation navigation(RexExpression.FunctionCall
call) {
+ boolean fromEnd = true;
+ int logicalOffset = 0;
+ boolean logicalSeen = false;
+ int physicalDelta = 0;
+ RexExpression current = call;
+ while (current instanceof RexExpression.FunctionCall) {
+ RexExpression.FunctionCall currentCall = (RexExpression.FunctionCall)
current;
+ String functionName = currentCall.getFunctionName();
+ if (RUNNING.equals(functionName) || FINAL.equals(functionName)) {
+ current = singleOperand(currentCall);
+ continue;
+ }
+ if (!NAVIGATION_FUNCTIONS.contains(functionName)) {
+ throw QueryErrorCode.QUERY_EXECUTION.asException(
+ "Unsupported expression inside a MATCH_RECOGNIZE row pattern
navigation: '" + currentCall
+ + "'. Only a column reference qualified by a pattern variable,
optionally wrapped in "
+ + "FIRST / LAST / PREV / NEXT, is supported.");
+ }
+ int offset = navigationOffset(currentCall);
+ if (PREV.equals(functionName)) {
+ physicalDelta -= offset;
+ } else if (NEXT.equals(functionName)) {
+ physicalDelta += offset;
+ } else {
+ if (logicalSeen) {
+ throw QueryErrorCode.QUERY_EXECUTION.asException(
+ "Nested FIRST / LAST is not supported in MATCH_RECOGNIZE: '" +
call + "'.");
+ }
+ logicalSeen = true;
+ fromEnd = LAST.equals(functionName);
+ logicalOffset = offset;
+ }
+ current = currentCall.getFunctionOperands().get(0);
+ }
+ if (!(current instanceof RexExpression.PatternFieldRef)) {
+ throw QueryErrorCode.QUERY_EXECUTION.asException(
+ "Unsupported operand of a MATCH_RECOGNIZE row pattern navigation: '"
+ call
+ + "'. Expecting a column reference qualified by a pattern
variable.");
+ }
+ RexExpression.PatternFieldRef ref = (RexExpression.PatternFieldRef)
current;
+ return new MatchTerm.Navigation(ref.getSymbolOrdinal(), fromEnd,
logicalOffset, physicalDelta, ref.getIndex(),
+ call.getDataType());
+ }
+
+ /// The offset operand of a navigation call. `PREV` and `NEXT` default to
one row, `FIRST` and
+ /// `LAST` to the first / last row itself.
+ private static int navigationOffset(RexExpression.FunctionCall call) {
+ List<RexExpression> operands = call.getFunctionOperands();
+ if (operands.size() < 2) {
+ String functionName = call.getFunctionName();
+ return PREV.equals(functionName) || NEXT.equals(functionName) ? 1 : 0;
+ }
+ RexExpression offset = operands.get(1);
+ if (!(offset instanceof RexExpression.Literal)) {
+ throw QueryErrorCode.QUERY_EXECUTION.asException(
+ "The offset of a MATCH_RECOGNIZE row pattern navigation must be a
constant, got: '" + offset + "'.");
+ }
+ Object value = ((RexExpression.Literal) offset).getValue();
+ if (!(value instanceof Number)) {
+ throw QueryErrorCode.QUERY_EXECUTION.asException(
+ "The offset of a MATCH_RECOGNIZE row pattern navigation must be an
integer, got: '" + value + "'.");
+ }
+ int intValue = ((Number) value).intValue();
Review Comment:
Could we validate this as an exact, non-negative `int` before converting it?
The planner currently accepts offsets such as `LAST(A.col3, 1.5)` and
`LAST(A.col3, 4294967296)`, but `Number.intValue()` converts them to `1` and
`0`, respectively. The server can therefore navigate to a different row instead
of rejecting the unsupported offset. Please use exact integrality/range checks
and add coverage for fractional and overflowing values.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/match/PartitionMatcher.java:
##########
@@ -0,0 +1,271 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.query.runtime.operator.match;
+
+import java.util.Arrays;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.query.planner.plannode.PatternSymbol;
+import org.apache.pinot.query.runtime.operator.match.PatternNfa.Transition;
+import org.apache.pinot.spi.exception.QueryErrorCode;
+
+
+/// Runs a [PatternNfa] over the rows of one partition and reports the
SQL:2016 preferred match starting at a
+/// given row.
+///
+/// ## First match found is the preferred match
+///
+/// The search is a depth first exploration that always takes the first
not-yet-tried transition of the current state.
+/// [PatternToNfaCompiler] orders the transitions of every state by SQL:2016
preference, so this enumeration is
+/// exactly the preferment order and [#match] can return as soon as it reaches
the accepting state. No candidate
+/// match is ever scored or compared against another.
+///
+/// ## Backtracking is O(1) per step
+///
+/// The search stack is held in parallel primitive arrays rather than objects,
and each frame records the single undo
+/// it needs: whether it pushed a row onto the [MatchTape], and which counter
register it changed together with
+/// that register's previous value. Popping a frame therefore costs a couple
of array writes; nothing is copied or
+/// rebuilt, and the classifier tape stays append only.
+///
+/// ## Running semantics come for free
+///
+/// A candidate row is pushed onto the tape *before* its DEFINE predicate is
evaluated, so the predicate sees the
+/// candidate as the current row and every navigation resolves against the
rows matched so far. That is precisely the
+/// SQL:2016 running semantics of DEFINE. If the predicate fails, the row is
popped again.
+///
+/// Not thread safe: one instance owns one tape and one search stack.
+public class PartitionMatcher {
+ /// Returned by [#match] when no match starts at the requested row.
+ public static final int NO_MATCH = -1;
+
+ private static final int NO_COUNTER = -1;
+ private static final int NO_POS = -1;
+ private static final int INITIAL_STACK_CAPACITY = 64;
+
+ private final PatternNfa _nfa;
+ /// DEFINE predicate per symbol ordinal; `null` means the variable matches
every row, per SQL:2016.
+ private final MatchExpression[] _definitions;
+ private final MatchTape _tape;
+ private final long _maxStepsPerMatchAttempt;
+ private final int[] _counters;
+ private final int[] _loopStartPos;
+
+ private int[] _frameState = new int[INITIAL_STACK_CAPACITY];
+ private int[] _framePos = new int[INITIAL_STACK_CAPACITY];
+ private int[] _frameNextTransition = new int[INITIAL_STACK_CAPACITY];
+ private int[] _frameUndoCounter = new int[INITIAL_STACK_CAPACITY];
+ private int[] _frameUndoCount = new int[INITIAL_STACK_CAPACITY];
+ private int[] _frameUndoLoopStart = new int[INITIAL_STACK_CAPACITY];
+ private boolean[] _frameUndoTapePush = new boolean[INITIAL_STACK_CAPACITY];
+ private int _stackSize;
+
+ public PartitionMatcher(PatternNfa nfa, List<PatternSymbol> patternSymbols,
DataSchema inputSchema,
+ long maxStepsPerMatchAttempt) {
+ _nfa = nfa;
+ _tape = new MatchTape(patternSymbols);
+ _maxStepsPerMatchAttempt = maxStepsPerMatchAttempt;
+ _definitions = new MatchExpression[patternSymbols.size()];
+ for (int i = 0; i < _definitions.length; i++) {
+ // A pattern variable that appears in PATTERN without a DEFINE entry
matches every row, per SQL:2016.
+ _definitions[i] = patternSymbols.get(i).getDefinition() == null ? null
+ : MatchExpression.compile(patternSymbols.get(i).getDefinition(),
inputSchema);
+ }
+ _counters = new int[nfa.getNumCounters()];
+ _loopStartPos = new int[nfa.getNumCounters()];
+ }
+
+ /// The classifier tape. After a successful [#match] it describes that match
and stays valid until the next
+ /// call, so MEASURES are evaluated through it with final semantics.
+ public MatchTape getTape() {
+ return _tape;
+ }
+
+ /// Finds the preferred match that starts exactly at `startPos`.
+ ///
+ /// @param matchNumber the value `MATCH_NUMBER()` reports; SQL:2016 assigns
it before the match is known to
+ /// succeed, so it is passed in rather than derived
+ /// @return the partition index one past the last row of the match, which
equals `startPos` for an empty match,
+ /// or [#NO_MATCH] if no match starts here
+ /// @throws org.apache.pinot.spi.exception.QueryException if the attempt
exceeds the configured step budget. It
+ /// throws rather than giving up, because giving up would silently
drop matches.
+ public int match(List<Object[]> partitionRows, int startPos, long
matchNumber) {
+ _tape.reset(partitionRows, startPos, matchNumber);
+ Arrays.fill(_counters, 0);
+ Arrays.fill(_loopStartPos, NO_POS);
+ _stackSize = 0;
+ pushFrame(_nfa.getStartState(), startPos, NO_COUNTER, 0, 0, false);
+
+ int partitionSize = partitionRows.size();
+ int acceptState = _nfa.getAcceptState();
+ long steps = 0;
+ while (_stackSize > 0) {
+ int top = _stackSize - 1;
+ if (_frameState[top] == acceptState) {
+ return _framePos[top];
+ }
+ List<Transition> transitions =
_nfa.getState(_frameState[top]).getTransitions();
+ boolean advanced = false;
+ while (_frameNextTransition[top] < transitions.size()) {
+ Transition transition = transitions.get(_frameNextTransition[top]++);
+ if (++steps > _maxStepsPerMatchAttempt) {
+ // The step count includes the transitions that make linear
progress, so a long match over a large
+ // partition can hit this without any backtracking at all. Reporting
the partition size and the number of
+ // rows consumed so far is what makes the two distinguishable, and
the tuning remedy comes first because a
+ // linear blowup has no ambiguity to remove.
+ throw QueryErrorCode.SERVER_RESOURCE_LIMIT_EXCEEDED.asException(
+ "MATCH_RECOGNIZE exceeded the maximum of " +
_maxStepsPerMatchAttempt
+ + " pattern matching steps for the match attempt starting at
row " + startPos + " of a "
+ + partitionSize + "-row partition (" + (_framePos[top] -
startPos) + " rows consumed so far). "
+ + "Raise the '" + MatchLimits.MAX_STEPS_PER_MATCH_ATTEMPT +
"' query option, or if the row count "
+ + "consumed is far below the step count, make the PATTERN
less ambiguous and tighten the DEFINE "
+ + "predicates.");
+ }
+ if (tryApply(transition, top, partitionSize)) {
+ advanced = true;
+ break;
+ }
+ }
+ if (!advanced) {
+ popFrame();
+ }
+ }
+ return NO_MATCH;
+ }
+
+ /// Applies `transition` to the frame at `parent`, pushing the resulting
frame if its guard holds.
+ ///
+ /// @return whether the transition was taken
+ private boolean tryApply(Transition transition, int parent, int
partitionSize) {
+ int pos = _framePos[parent];
+ int target = transition.getTarget();
+ switch (transition.getKind()) {
+ case MATCH: {
+ if (pos >= partitionSize) {
+ return false;
+ }
+ int symbolOrdinal = transition.getOperand();
+ // Push before evaluating so the predicate sees the candidate row as
the current row.
+ _tape.push(symbolOrdinal);
+ MatchExpression definition = _definitions[symbolOrdinal];
+ if (definition != null && !definition.test(_tape)) {
+ _tape.pop();
+ return false;
+ }
+ pushFrame(target, pos + 1, NO_COUNTER, 0, 0, true);
+ return true;
+ }
+ case EPSILON:
+ pushFrame(target, pos, NO_COUNTER, 0, 0, false);
+ return true;
+ case START_LOOP: {
+ int counterId = transition.getOperand();
+ pushFrame(target, pos, counterId, _counters[counterId],
_loopStartPos[counterId], false);
+ _counters[counterId] = 0;
+ _loopStartPos[counterId] = NO_POS;
+ return true;
+ }
+ case REPEAT: {
+ int counterId = transition.getOperand();
+ int maxRepeat = transition.getBound();
+ if (maxRepeat != PatternNfa.UNBOUNDED && _counters[counterId] >=
maxRepeat) {
+ return false;
+ }
+ // Empty cycle guard: the previous iteration of this quantifier
consumed no row, so another one never will.
+ if (_loopStartPos[counterId] == pos) {
+ return false;
+ }
+ pushFrame(target, pos, counterId, _counters[counterId],
_loopStartPos[counterId], false);
+ _counters[counterId]++;
+ _loopStartPos[counterId] = pos;
+ return true;
+ }
+ case EXIT_LOOP: {
+ int counterId = transition.getOperand();
+ // An empty iteration may be repeated vacuously any number of times,
so it satisfies any minimum.
+ if (_counters[counterId] < transition.getBound() &&
_loopStartPos[counterId] != pos) {
+ return false;
+ }
+ pushFrame(target, pos, NO_COUNTER, 0, 0, false);
+ return true;
+ }
+ case ANCHOR_START:
+ if (pos != 0) {
+ return false;
+ }
+ pushFrame(target, pos, NO_COUNTER, 0, 0, false);
+ return true;
+ case ANCHOR_END:
+ if (pos != partitionSize) {
+ return false;
+ }
+ pushFrame(target, pos, NO_COUNTER, 0, 0, false);
+ return true;
+ default:
+ throw QueryErrorCode.QUERY_EXECUTION.asException(
+ "Unsupported MATCH_RECOGNIZE pattern transition: " +
transition.getKind());
+ }
+ }
+
+ /// Pushes a frame together with the undo of the transition that created it:
at most one tape row and at most one
+ /// counter register change.
+ private void pushFrame(int state, int pos, int undoCounter, int undoCount,
int undoLoopStart,
+ boolean undoTapePush) {
+ if (_stackSize == _frameState.length) {
+ growStack();
+ }
+ _frameState[_stackSize] = state;
+ _framePos[_stackSize] = pos;
+ _frameNextTransition[_stackSize] = 0;
+ _frameUndoCounter[_stackSize] = undoCounter;
+ _frameUndoCount[_stackSize] = undoCount;
+ _frameUndoLoopStart[_stackSize] = undoLoopStart;
+ _frameUndoTapePush[_stackSize] = undoTapePush;
+ _stackSize++;
+ }
+
+ private void popFrame() {
+ int top = --_stackSize;
+ if (_frameUndoTapePush[top]) {
+ _tape.pop();
+ }
+ int undoCounter = _frameUndoCounter[top];
+ if (undoCounter != NO_COUNTER) {
+ _counters[undoCounter] = _frameUndoCount[top];
+ _loopStartPos[undoCounter] = _frameUndoLoopStart[top];
+ }
+ }
+
+ private void growStack() {
+ int capacity = _frameState.length * 2;
+ _frameState = Arrays.copyOf(_frameState, capacity);
Review Comment:
The stack retains one frame per transition, not just backtracking choices,
and these six `int[]` arrays plus a `boolean[]` retain their high-water
capacity. An allowed one-million-row `PATTERN (A+)` match reaches about three
million frames, rounds to capacity 4,194,304, and retains about 100 MiB for
these arrays alone. Please store only choice/undo points or otherwise compact
deterministic paths, and enforce a frame or memory limit.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/match/PatternNfa.java:
##########
@@ -0,0 +1,186 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.query.runtime.operator.match;
+
+import java.util.ArrayList;
+import java.util.List;
+
+
+/// A non-deterministic finite automaton compiled from a MATCH_RECOGNIZE
`PATTERN` clause by
+/// [PatternToNfaCompiler], and executed by [PartitionMatcher].
+///
+/// ## Prioritized transitions
+///
+/// The transitions of a state are stored in **preference order**, highest
preference first. A depth first search
+/// that always tries the transitions of a state in list order therefore
enumerates candidate matches in the SQL:2016
+/// "preferment order", and the first complete match it reaches is the
preferred one. See
+/// [PatternToNfaCompiler] for how each pattern construct maps onto that
ordering.
+///
+/// ## Counter registers
+///
+/// Bounded quantifiers such as `A{2,5}` are **not** unrolled into repeated
states. Every quantifier node
+/// allocates one counter register; the loop is a single cycle in the
automaton whose entry and exit edges are guarded
+/// by that counter. The number of states is therefore linear in the size of
the pattern text and independent of the
+/// repetition bounds, so `A{1,10000}` costs the same to compile as `A+`.
+///
+/// Instances are immutable and safe to share between the threads that execute
different partitions.
+public class PatternNfa {
+ /// Sentinel for [Transition#getBound()] of an unbounded
[TransitionKind#REPEAT].
+ public static final int UNBOUNDED = -1;
+
+ private final List<State> _states;
+ private final int _startState;
+ private final int _acceptState;
+ private final int _numCounters;
+
+ PatternNfa(List<State> states, int startState, int acceptState, int
numCounters) {
+ _states = List.copyOf(states);
+ _startState = startState;
+ _acceptState = acceptState;
+ _numCounters = numCounters;
+ }
+
+ public List<State> getStates() {
+ return _states;
+ }
+
+ public State getState(int stateId) {
+ return _states.get(stateId);
+ }
+
+ public int getNumStates() {
+ return _states.size();
+ }
+
+ public int getStartState() {
+ return _startState;
+ }
+
+ /// The single accepting state. Reaching it means the whole pattern has been
matched.
+ public int getAcceptState() {
+ return _acceptState;
+ }
+
+ /// Number of counter registers the automaton uses, i.e. the number of
quantifiers in the pattern.
+ public int getNumCounters() {
+ return _numCounters;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder builder = new
StringBuilder("PatternNfa{start=").append(_startState).append(", accept=")
+ .append(_acceptState).append(",
counters=").append(_numCounters).append('}');
+ for (int stateId = 0; stateId < _states.size(); stateId++) {
+ builder.append('\n').append(stateId).append(':');
+ for (Transition transition : _states.get(stateId).getTransitions()) {
+ builder.append(' ').append(transition);
+ }
+ }
+ return builder.toString();
+ }
+
+ /// What a transition does when it is taken.
+ public enum TransitionKind {
+ /// Consumes the current row if the DEFINE predicate of
[Transition#getOperand()] holds for it.
+ MATCH,
+ /// Consumes nothing and has no side effect.
+ EPSILON,
+ /// Enters a quantifier: resets the counter register
[Transition#getOperand()].
+ START_LOOP,
+ /// Runs one more iteration of a quantifier. Allowed while the counter is
below
+ /// [Transition#getBound()] repetitions and the previous iteration
consumed at least one row; increments the
+ /// counter.
+ REPEAT,
+ /// Leaves a quantifier. Allowed once the counter reached
[Transition#getBound()] repetitions, or once an
+ /// iteration turned out to be empty.
+ EXIT_LOOP,
+ /// The `^` anchor: allowed only at the first row of the partition.
+ ANCHOR_START,
+ /// The `$` anchor: allowed only past the last row of the partition.
+ ANCHOR_END
+ }
+
+ /// One outgoing edge of a [State]. Its position in [State#getTransitions()]
is its preference.
+ public static final class Transition {
+ private final TransitionKind _kind;
+ private final int _target;
+ private final int _operand;
+ private final int _bound;
+
+ Transition(TransitionKind kind, int target, int operand, int bound) {
+ _kind = kind;
+ _target = target;
+ _operand = operand;
+ _bound = bound;
+ }
+
+ public TransitionKind getKind() {
+ return _kind;
+ }
+
+ /// The state this transition leads to.
+ public int getTarget() {
+ return _target;
+ }
+
+ /// Pattern symbol ordinal for [TransitionKind#MATCH], counter register id
for the loop kinds, unused
+ /// otherwise.
+ public int getOperand() {
+ return _operand;
+ }
+
+ /// Maximum repetition count for [TransitionKind#REPEAT] ([#UNBOUNDED] if
there is no upper bound), and
+ /// the minimum repetition count for [TransitionKind#EXIT_LOOP]. Unused
otherwise.
+ public int getBound() {
+ return _bound;
+ }
+
+ @Override
+ public String toString() {
+ switch (_kind) {
+ case MATCH:
+ return "MATCH(s" + _operand + ")->" + _target;
+ case EPSILON:
+ return "EPS->" + _target;
+ case START_LOOP:
+ return "START(c" + _operand + ")->" + _target;
+ case REPEAT:
+ return "REPEAT(c" + _operand + ",max=" + (_bound == UNBOUNDED ? "*"
: _bound) + ")->" + _target;
+ case EXIT_LOOP:
+ return "EXIT(c" + _operand + ",min=" + _bound + ")->" + _target;
+ default:
+ return _kind + "->" + _target;
+ }
+ }
+ }
+
+ /// A state of the automaton. Mutable only while [PatternToNfaCompiler]
builds it.
+ public static final class State {
+ private final List<Transition> _transitions = new ArrayList<>(2);
+
+ /// The outgoing transitions in preference order, highest preference first.
+ public List<Transition> getTransitions() {
+ return _transitions;
Review Comment:
This exposes the mutable backing list even though `PatternNfa` is documented
as immutable and safe to share. A caller can obtain it through
`getState(...).getTransitions()` and reorder or clear transitions;
`PartitionMatcher` reads the same list, whose order determines match
preference. Please freeze each state’s transition list before exposing the
compiled NFA.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MatchOperator.java:
##########
@@ -0,0 +1,356 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.query.runtime.operator;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.datatable.StatMap;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
+import org.apache.pinot.core.data.table.Key;
+import org.apache.pinot.query.planner.plannode.MatchNode;
+import org.apache.pinot.query.planner.plannode.PatternSymbol;
+import org.apache.pinot.query.runtime.blocks.MseBlock;
+import org.apache.pinot.query.runtime.blocks.RowHeapDataBlock;
+import org.apache.pinot.query.runtime.operator.match.MatchExpression;
+import org.apache.pinot.query.runtime.operator.match.MatchLimits;
+import org.apache.pinot.query.runtime.operator.match.MatchTape;
+import org.apache.pinot.query.runtime.operator.match.PartitionMatcher;
+import org.apache.pinot.query.runtime.operator.match.PatternNfa;
+import org.apache.pinot.query.runtime.operator.match.PatternToNfaCompiler;
+import org.apache.pinot.query.runtime.operator.utils.AggregationUtils;
+import org.apache.pinot.query.runtime.operator.utils.TypeUtils;
+import org.apache.pinot.query.runtime.plan.OpChainExecutionContext;
+import org.apache.pinot.spi.exception.QueryErrorCode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Evaluates SQL:2016 `MATCH_RECOGNIZE` (row pattern recognition) with `ONE
ROW PER MATCH`.
+///
+/// ## What it does per partition
+///
+/// For every `PARTITION BY` partition, in `ORDER BY` order, it walks a scan
position from the first row to
+/// the last. At each position it asks [PartitionMatcher] for the preferred
match starting exactly there. On a
+/// match it emits one row - the partition key columns followed by the
`MEASURES` - and then moves the scan
+/// position according to the `AFTER MATCH SKIP` mode. On no match it moves
one row forward.
+///
+/// ## What it expects from the plan
+///
+/// Like [WindowAggregateOperator], this operator does not sort.
`PinotMatchExchangeNodeInsertRule` puts a
+/// sort exchange underneath that hash distributes on the partition keys and
sorts the receiver side on
+/// `(partitionKeys..., orderKeys...)`, so rows arrive grouped by partition
and ordered within a partition. The
+/// operator therefore buffers one partition at a time and releases it at each
boundary, and never reads
+/// [MatchNode#getCollations()]: the ordering has already been established
below it.
+///
+/// The grouping half of that assumption is verified rather than trusted: if a
partition key reappears after its
+/// partition was closed, the operator fails instead of silently splitting one
partition into two and reporting matches
+/// that do not exist. The ordering half is not re-checked per row, because an
exchange that grouped correctly but
+/// sorted incorrectly is not a failure mode the exchange can produce - losing
the sort loses the grouping too, which
+/// the reappearance check already catches.
+///
+/// ## Guardrails throw, they never truncate
+///
+/// [MatchLimits#MAX_ROWS_IN_MATCH] bounds the rows buffered for a partition
and
+/// [MatchLimits#MAX_STEPS_PER_MATCH_ATTEMPT] bounds the backtracking of one
match attempt. Both raise an error,
+/// because a truncated pattern result is a wrong result that nothing in the
response would flag.
+///
+/// ## Not supported yet
+///
+/// `ALL ROWS PER MATCH` is rejected here as well as during planning: this
operator emits exactly one row per
+/// match, so accepting it would silently return the wrong shape of result.
+public class MatchOperator extends MultiStageOperator {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(MatchOperator.class);
+ private static final String EXPLAIN_NAME = "MATCH_RECOGNIZE";
+
+ private final MultiStageOperator _input;
+ private final DataSchema _resultSchema;
+ private final ColumnDataType[] _resultStoredTypes;
+ private final int[] _partitionKeys;
+ private final List<PatternSymbol> _patternSymbols;
+ private final MatchExpression[] _measures;
+ private final PartitionMatcher _matcher;
+ private final MatchNode.AfterMatchSkipMode _skipMode;
+ private final int _skipToSymbolOrdinal;
+ private final int _maxRowsInMatch;
+ private final StatMap<StatKey> _statMap = new StatMap<>(StatKey.class);
+
+ private final Set<Key> _closedPartitionKeys = new HashSet<>();
+ private List<Object[]> _partitionRows = new ArrayList<>();
+ private List<Object[]> _outputRows = new ArrayList<>();
+ @Nullable
+ private Key _currentPartitionKey;
+ @Nullable
+ private MseBlock.Eos _inputEos;
+ private int _numRows;
+
+ public MatchOperator(OpChainExecutionContext context, MultiStageOperator
input, DataSchema inputSchema,
+ MatchNode node) {
+ super(context);
+ if (node.getRowsPerMatchMode() !=
MatchNode.RowsPerMatchMode.ONE_ROW_PER_MATCH) {
+ throw QueryErrorCode.QUERY_EXECUTION.asException(
+ "ALL ROWS PER MATCH is not supported yet in MATCH_RECOGNIZE. Use ONE
ROW PER MATCH (the default) and "
+ + "expose the per-match values you need through the MEASURES
clause.");
+ }
+ _input = input;
+ _resultSchema = node.getDataSchema();
+ _resultStoredTypes = _resultSchema.getStoredColumnDataTypes();
+ List<Integer> partitionKeys = node.getPartitionKeys();
+ _partitionKeys = new int[partitionKeys.size()];
+ for (int i = 0; i < _partitionKeys.length; i++) {
+ _partitionKeys[i] = partitionKeys.get(i);
+ }
+ _patternSymbols = node.getPatternSymbols();
+ List<MatchNode.Measure> measures = node.getMeasures();
+ _measures = new MatchExpression[measures.size()];
+ for (int i = 0; i < _measures.length; i++) {
+ _measures[i] = MatchExpression.compile(measures.get(i).getExpression(),
inputSchema);
+ }
+ if (_resultSchema.size() != _partitionKeys.length + _measures.length) {
+ throw QueryErrorCode.QUERY_EXECUTION.asException(
+ "MATCH_RECOGNIZE output schema " + _resultSchema + " does not match
" + _partitionKeys.length
+ + " partition key(s) plus " + _measures.length + " measure(s)");
+ }
+ _skipMode = node.getAfterMatchSkipMode();
+ _skipToSymbolOrdinal = node.getAfterMatchSkipToSymbolOrdinal();
+
+ PatternNfa nfa = PatternToNfaCompiler.compile(node.getPattern());
+ _maxRowsInMatch =
MatchLimits.getMaxRowsInMatch(context.getOpChainMetadata(), node.getNodeHint());
+ long maxStepsPerMatchAttempt =
+ MatchLimits.getMaxStepsPerMatchAttempt(context.getOpChainMetadata(),
node.getNodeHint());
+ _matcher = new PartitionMatcher(nfa, _patternSymbols, inputSchema,
maxStepsPerMatchAttempt);
+ }
+
+ @Override
+ protected Logger logger() {
+ return LOGGER;
+ }
+
+ @Override
+ public List<MultiStageOperator> getChildOperators() {
+ return List.of(_input);
+ }
+
+ @Override
+ public Type getOperatorType() {
+ return Type.MATCH;
+ }
+
+ @Override
+ public String toExplainString() {
+ return EXPLAIN_NAME;
+ }
+
+ @Override
+ public void registerExecution(long time, int numRows, long memoryUsedBytes,
long gcTimeMs) {
+ _statMap.merge(StatKey.EXECUTION_TIME_MS, time);
+ _statMap.merge(StatKey.EMITTED_ROWS, numRows);
+ _statMap.merge(StatKey.ALLOCATED_MEMORY_BYTES, memoryUsedBytes);
+ _statMap.merge(StatKey.GC_TIME_MS, gcTimeMs);
+ }
+
+ @Override
+ public StatMap<StatKey> copyStatMaps() {
+ return new StatMap<>(_statMap);
+ }
+
+ @Override
+ protected MseBlock getNextBlock() {
+ while (_outputRows.isEmpty()) {
+ if (_inputEos != null) {
+ return _inputEos;
+ }
+ MseBlock block = _input.nextBlock();
+ if (block.isData()) {
+ consumeBlock((MseBlock.Data) block);
+ } else {
+ _inputEos = (MseBlock.Eos) block;
+ if (_inputEos.isError()) {
+ return _inputEos;
+ }
+ closeCurrentPartition();
+ }
+ }
+ List<Object[]> rows = _outputRows;
+ _outputRows = new ArrayList<>();
+ return new RowHeapDataBlock(rows, _resultSchema);
+ }
+
+ /// Buffers the rows of one input block, matching and releasing a partition
as soon as its last row went by.
+ private void consumeBlock(MseBlock.Data block) {
+ for (Object[] row : block.asRowHeap().getRows()) {
+ Key partitionKey = AggregationUtils.extractRowKey(row, _partitionKeys);
+ if (_currentPartitionKey == null) {
+ openPartition(partitionKey);
+ } else if (!_currentPartitionKey.equals(partitionKey)) {
+ closeCurrentPartition();
+ openPartition(partitionKey);
+ }
+ if (_partitionRows.size() >= _maxRowsInMatch) {
+ throw QueryErrorCode.SERVER_RESOURCE_LIMIT_EXCEEDED.asException(
+ "MATCH_RECOGNIZE partition exceeds the maximum of " +
_maxRowsInMatch
+ + " rows. Partition on a column with more distinct values, or
raise the '"
+ + MatchLimits.MAX_ROWS_IN_MATCH + "' query option.");
+ }
+ _partitionRows.add(row);
+ _numRows++;
+ checkTerminationAndSampleUsagePeriodically(_numRows, EXPLAIN_NAME);
+ }
+ }
+
+ private void openPartition(Key partitionKey) {
+ if (!_closedPartitionKeys.add(partitionKey)) {
+ // The sort exchange below this operator guarantees that the rows of a
partition arrive contiguously. If they
+ // did not, matching each fragment on its own would report matches that
do not exist in the real partition.
+ throw QueryErrorCode.QUERY_EXECUTION.asException(
+ "MATCH_RECOGNIZE input is not grouped by partition key: partition "
+ partitionKey
+ + " reappeared after it was closed.");
+ }
+ _currentPartitionKey = partitionKey;
+ }
+
+ private void closeCurrentPartition() {
+ if (!_partitionRows.isEmpty()) {
+ matchPartition();
+ _partitionRows = new ArrayList<>();
+ }
+ _currentPartitionKey = null;
+ }
+
+ /// Scans the buffered partition, emitting one row per match and advancing
the scan position per the
+ /// `AFTER MATCH SKIP` mode.
+ private void matchPartition() {
+ List<Object[]> rows = _partitionRows;
+ int numPartitionRows = rows.size();
+ MatchTape tape = _matcher.getTape();
+ long matchNumber = 0;
+ int scanStart = 0;
+ while (scanStart < numPartitionRows) {
+ checkTerminationAndSampleUsage();
+ int endPos = _matcher.match(rows, scanStart, matchNumber + 1);
+ if (endPos == PartitionMatcher.NO_MATCH) {
+ scanStart++;
+ continue;
+ }
+ matchNumber++;
+ _outputRows.add(buildOutputRow(rows, tape));
Review Comment:
`matchPartition()` accumulates every match before `getNextBlock()` can
return. A legal one-million-row partition with `PATTERN (A)` creates one
million output arrays in one `RowHeapDataBlock` while the input partition
remains resident. Please preserve the scan position as operator state and emit
bounded batches so output memory does not scale with the entire partition.
##########
pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java:
##########
@@ -813,6 +813,11 @@ public static class QueryOptionKey {
// un-upgraded server cannot honor, so only enable it once all servers
support it.
// NOTE: This is a no-op under usePhysicalOptimizer (the v2 path does
not go through RelToPlanNodeConverter).
public static final String UNNEST_COLUMN_PRUNING =
"unnestColumnPruning";
+ /// A MATCH_RECOGNIZE partition is evaluated as a whole on a single
worker, so a query without a PARTITION BY
+ /// clause hashes on zero keys and routes the entire table to one
worker. The multi-stage planner rejects that
+ /// plan by default; set this to true to accept single-worker
execution (fine for small tables, or when the
+ /// WHERE clause already narrows the input down to one logical
partition).
+ public static final String ALLOW_MATCH_RECOGNIZE_WITHOUT_PARTITION_BY
= "allowMatchRecognizeWithoutPartitionBy";
Review Comment:
This user-visible opt-in is covered only by the exchange-rule test and an
EXPLAIN plan; every real-cluster MATCH query still uses `PARTITION BY`. Please
add one cluster E2E query with this option enabled and no `PARTITION BY`, using
rows spread across the existing segments/servers and asserting one global
result. That exercises option propagation, the zero-key exchange, dispatch, and
the runtime’s empty-partition-key path.
--
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]
