xiangfu0 commented on code in PR #19311:
URL: https://github.com/apache/pinot/pull/19311#discussion_r3838038639
##########
pinot-query-runtime/src/test/resources/queries/MatchRecognize.json:
##########
@@ -0,0 +1,93 @@
+{
+ "match_recognize_v_shape": {
+ "tables": {
+ "ticker": {
+ "schema": [
+ {"name": "symbol", "type": "STRING"},
+ {"name": "seq", "type": "LONG"},
+ {"name": "price", "type": "INT"}
+ ],
+ "inputs": [
+ ["A", 1, 10],
+ ["A", 2, 8],
+ ["A", 3, 6],
+ ["A", 4, 9],
+ ["A", 5, 12],
+ ["A", 6, 7],
+ ["A", 7, 11],
+ ["B", 1, 5],
+ ["B", 2, 3],
+ ["B", 3, 8]
+ ]
+ }
+ },
+ "queries": [
+ {
+ "description": "V-shape detection with an explicit AFTER MATCH SKIP
PAST LAST ROW: matches never overlap. H2 has no MATCH_RECOGNIZE, so the
expected rows are hand-computed.",
+ "sql": "SELECT * FROM {ticker} MATCH_RECOGNIZE (PARTITION BY symbol
ORDER BY seq MEASURES FIRST(DOWN.price) AS start_price, LAST(UP.price) AS
end_price ONE ROW PER MATCH AFTER MATCH SKIP PAST LAST ROW PATTERN (DOWN+ UP+)
DEFINE DOWN AS DOWN.price < PREV(DOWN.price), UP AS UP.price > PREV(UP.price))
AS mr",
+ "outputs": [
+ ["A", 8, 12],
+ ["A", 7, 11],
+ ["B", 3, 8]
+ ],
+ "ignoreV2Optimizer": true,
Review Comment:
These `ignoreV2Optimizer`/`ignoreLiteMode` flags make
`ResourceBasedQueriesTest` `SkipException` out of the v2 and lite paths
entirely, so the behavior of MATCH_RECOGNIZE under `usePhysicalOptimizer=true`
is never asserted.
Right now that path isn't a clean rejection: `RelToPRelConverter` has no
`LogicalMatch` branch, so the query falls through to `throw new
IllegalStateException("Unexpected relNode type: ...LogicalMatch")` — an obscure
internal error rather than the actionable "not supported" message the PR is
careful to produce everywhere else.
Suggest (a) adding a targeted, validator-style rejection for MATCH_RECOGNIZE
under v2/lite, and (b) a test that *positively asserts* that rejection instead
of skipping, so a future default-flip of the physical optimizer can't silently
regress it.
##########
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<>();
Review Comment:
`_closedPartitionKeys` grows unbounded in the number of *distinct
partitions* for the lifetime of the operator. `MatchLimits` caps
rows-per-partition (`maxRowsInMatch`) and steps-per-attempt, but nothing bounds
the partition count.
A `PARTITION BY <high-cardinality col>` that lands many millions of tiny
(even single-row) partitions on one worker retains one `Key` per closed
partition, which can OOM even though every individual partition is well within
`maxRowsInMatch`. Bounded-by-cardinality rather than a classic leak, since it's
needed for the reappearance-detection guard — but worth a comment noting the
growth, or a cap.
(Not a correctness bug; the guard itself is a sound backstop for the
exchange-clustering assumption.)
--
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]
