You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@calcite.apache.org by jh...@apache.org on 2019/09/17 21:20:55 UTC
[calcite] 02/02: [CALCITE-3302] Implement CLASSIFIER and LAST
functions for MATCH_RECOGNIZE
This is an automated email from the ASF dual-hosted git repository.
jhyde pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/calcite.git
commit f4dde669631f7fa4f64b42eb1481ba9a8f27bd39
Author: Julian Feinauer <j....@pragmaticminds.de>
AuthorDate: Wed Aug 28 10:34:53 2019 +0200
[CALCITE-3302] Implement CLASSIFIER and LAST functions for MATCH_RECOGNIZE
Close apache/calcite#1449
---
.../adapter/enumerable/EnumerableMatch.java | 144 ++++++++++++++++++---
.../adapter/enumerable/MatchImplementor.java | 46 +++++++
.../calcite/adapter/enumerable/MatchUtils.java | 51 ++++++++
.../calcite/adapter/enumerable/RexImpTable.java | 100 ++++++++++++--
.../org/apache/calcite/runtime/Enumerables.java | 2 +-
.../java/org/apache/calcite/runtime/Matcher.java | 12 +-
.../org/apache/calcite/sql/SqlMatchFunction.java | 33 +++++
.../calcite/sql/fun/SqlStdOperatorTable.java | 13 +-
.../org/apache/calcite/util/BuiltInMethod.java | 3 +
.../org/apache/calcite/plan/RelWriterTest.java | 10 +-
.../org/apache/calcite/runtime/AutomatonTest.java | 36 ++++--
core/src/test/resources/sql/match.iq | 50 +++++++
.../elasticsearch/ElasticsearchSchemaFactory.java | 1 -
13 files changed, 439 insertions(+), 62 deletions(-)
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableMatch.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableMatch.java
index 2cd14ac..e56e0bd 100644
--- a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableMatch.java
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableMatch.java
@@ -41,6 +41,7 @@ import org.apache.calcite.rex.RexProgramBuilder;
import org.apache.calcite.rex.RexSubQuery;
import org.apache.calcite.rex.RexVisitorImpl;
import org.apache.calcite.runtime.Enumerables;
+import org.apache.calcite.sql.SqlMatchFunction;
import org.apache.calcite.util.BuiltInMethod;
import org.apache.calcite.util.ImmutableBitSet;
import org.apache.calcite.util.Pair;
@@ -126,7 +127,9 @@ public class EnumerableMatch extends Match implements EnumerableRel {
final RelDataTypeFactory.Builder typeBuilder =
implementor.getTypeFactory().builder();
- measures.forEach((name, value) -> typeBuilder.add(name, value.getType()));
+
+ measures.forEach((name, value) ->
+ typeBuilder.add(name, value.getType()).nullable(true));
final PhysType emitType =
PhysTypeImpl.of(implementor.getTypeFactory(), typeBuilder.build(),
@@ -163,26 +166,31 @@ public class EnumerableMatch extends Match implements EnumerableRel {
Expressions.parameter(int.class, "match");
final ParameterExpression consumer_ =
Expressions.parameter(Consumer.class, "consumer");
+ final ParameterExpression i_ = Expressions.parameter(int.class, "i");
final ParameterExpression row_ =
Expressions.parameter(inputPhysType.getJavaRowType(), "row");
+
final BlockBuilder builder2 = new BlockBuilder();
+ // Add loop variable initialization
+ builder2.add(
+ Expressions.declare(0, row_,
+ Types.castIfNecessary(inputPhysType.getJavaRowType(),
+ Expressions.call(rows_, BuiltInMethod.LIST_GET.method, i_))));
+
RexBuilder rexBuilder = new RexBuilder(implementor.getTypeFactory());
- RexProgramBuilder rexProgramBuilder = new RexProgramBuilder(inputPhysType.getRowType(),
- rexBuilder);
+ RexProgramBuilder rexProgramBuilder =
+ new RexProgramBuilder(inputPhysType.getRowType(), rexBuilder);
for (Map.Entry<String, RexNode> entry : measures.entrySet()) {
rexProgramBuilder.addProject(entry.getValue(), entry.getKey());
}
- final List<Expression> arguments =
- RexToLixTranslator.translateProjects(rexProgramBuilder.getProgram(),
- (JavaTypeFactory) getCluster().getTypeFactory(),
- implementor.getConformance(), builder2, physType,
- implementor.getRootExpression(),
- new RexToLixTranslator.InputGetterImpl(
- Collections.singletonList(
- Pair.of(row_, inputPhysType))),
- implementor.allCorrelateVariables);
+
+ final RexToLixTranslator translator = RexToLixTranslator.forAggregation(
+ (JavaTypeFactory) getCluster().getTypeFactory(),
+ builder2,
+ new PassedRowsInputGetter(row_, rows_, inputPhysType),
+ implementor.getConformance());
final ParameterExpression result_ =
Expressions.parameter(physType.getJavaRowType());
@@ -190,19 +198,30 @@ public class EnumerableMatch extends Match implements EnumerableRel {
builder2.add(
Expressions.declare(Modifier.FINAL, result_,
Expressions.new_(physType.getJavaRowType())));
- for (int i = 0; i < arguments.size(); i++) {
- builder2.add(
- Expressions.statement(
- Expressions.assign(physType.fieldReference(result_, i),
- arguments.get(i))));
- }
+ Ord.forEach(measures.values(), (measure, i) ->
+ builder2.add(
+ Expressions.statement(
+ Expressions.assign(physType.fieldReference(result_, i),
+ implementMeasure(translator, rows_, symbols_, i_, row_,
+ measure)))));
builder2.add(
Expressions.statement(
Expressions.call(consumer_, BuiltInMethod.CONSUMER_ACCEPT.method,
result_)));
final BlockBuilder builder = new BlockBuilder();
- builder.add(Expressions.forEach(row_, rows_, builder2.toBlock()));
+
+ // Loop Length
+
+ // we have to use an explicit for (int i = ...) loop, as we need to know later
+ // which of the matched rows are already passed (in MatchUtils), so foreach cannot be used
+ builder.add(
+ Expressions.for_(
+ Expressions.declare(0, i_, Expressions.constant(0)),
+ Expressions.lessThan(i_,
+ Expressions.call(rows_, BuiltInMethod.COLLECTION_SIZE.method)),
+ Expressions.preIncrementAssign(i_),
+ builder2.toBlock()));
return Expressions.new_(
Types.of(Enumerables.Emitter.class), NO_EXPRS,
@@ -214,6 +233,48 @@ public class EnumerableMatch extends Match implements EnumerableRel {
builder.toBlock())));
}
+ private Expression implementMeasure(RexToLixTranslator translator,
+ ParameterExpression rows_, ParameterExpression symbols_,
+ ParameterExpression i_, ParameterExpression row_, RexNode value) {
+ final SqlMatchFunction matchFunction;
+ final MatchImplementor matchImplementor;
+ switch (value.getKind()) {
+ case LAST:
+ case PREV:
+ case CLASSIFIER:
+ matchFunction = (SqlMatchFunction) ((RexCall) value).getOperator();
+ matchImplementor = RexImpTable.INSTANCE.get(matchFunction);
+
+ // Work with the implementor
+ return matchImplementor.implement(translator, (RexCall) value,
+ row_, rows_, symbols_, i_);
+
+ case RUNNING:
+ case FINAL:
+ // See [CALCITE-3341], this should be changed a bit, to implement
+ // FINAL behavior
+ final List<RexNode> operands = ((RexCall) value).getOperands();
+ assert operands.size() == 1;
+
+ switch (operands.get(0).getKind()) {
+ case LAST:
+ case PREV:
+ case CLASSIFIER:
+ final RexCall call = (RexCall) operands.get(0);
+ matchFunction = (SqlMatchFunction) call.getOperator();
+ matchImplementor = RexImpTable.INSTANCE.get(matchFunction);
+ // Work with the implementor
+ ((PassedRowsInputGetter) translator.inputGetter).setIndex(null);
+ return matchImplementor.implement(translator, call, row_, rows_,
+ symbols_, i_);
+ }
+ return translator.translate(operands.get(0));
+
+ default:
+ return translator.translate(value);
+ }
+ }
+
private Expression implementMatcher(EnumerableRelImplementor implementor,
PhysType physType, BlockBuilder builder, ParameterExpression row_) {
final Expression patternBuilder_ = builder.append("patternBuilder",
@@ -289,7 +350,7 @@ public class EnumerableMatch extends Match implements EnumerableRel {
// return this.test(row, (List) rows);
// }
final ParameterExpression row0_ =
- Expressions.parameter(Object.class, "row");
+ Expressions.parameter(Object.class, "row");
final ParameterExpression rowsO_ =
Expressions.parameter(Object.class, "rows");
BlockBuilder bridgeBody = new BlockBuilder();
@@ -300,7 +361,7 @@ public class EnumerableMatch extends Match implements EnumerableRel {
BuiltInMethod.PREDICATE_TEST.method,
Expressions.convert_(row0_,
Types.of(MemoryFactory.Memory.class,
- physType.getJavaRowType())))));
+ physType.getJavaRowType())))));
memberDeclarations.add(
EnumUtils.overridingMethodDecl(
BuiltInMethod.PREDICATE_TEST.method,
@@ -382,6 +443,47 @@ public class EnumerableMatch extends Match implements EnumerableRel {
}
/**
+ * A special Getter that is able to return a field from a list of objects.
+ */
+ static class PassedRowsInputGetter implements RexToLixTranslator.InputGetter {
+ private Expression index;
+ private final ParameterExpression row;
+ private final ParameterExpression passedRows;
+ private final Function<Expression, RexToLixTranslator.InputGetter> generator;
+ private final PhysType physType;
+
+ PassedRowsInputGetter(ParameterExpression row, ParameterExpression passedRows,
+ PhysType physType) {
+ this.row = row;
+ this.passedRows = passedRows;
+ generator = e -> new RexToLixTranslator.InputGetterImpl(
+ Collections.singletonList(
+ Pair.of(e, physType)));
+ this.physType = physType;
+ }
+
+ void setIndex(Expression index) {
+ this.index = index;
+ }
+
+ @Override public Expression field(BlockBuilder list, int index,
+ Type storageType) {
+ if (this.index == null) {
+ return generator.apply(this.row).field(list, index, storageType);
+ }
+
+ return Expressions.condition(
+ Expressions.greaterThanOrEqual(this.index, Expressions.constant(0)),
+ generator.apply(
+ Types.castIfNecessary(physType.getJavaRowType(),
+ Expressions.call(this.passedRows,
+ BuiltInMethod.LIST_GET.method, this.index)))
+ .field(list, index, storageType),
+ Expressions.constant(null));
+ }
+ }
+
+ /**
* A special Getter that "interchanges" the PREV and the field call.
*/
static class PrevInputGetter implements RexToLixTranslator.InputGetter {
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/MatchImplementor.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/MatchImplementor.java
new file mode 100644
index 0000000..8f632c5
--- /dev/null
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/MatchImplementor.java
@@ -0,0 +1,46 @@
+/*
+ * 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.calcite.adapter.enumerable;
+
+import org.apache.calcite.linq4j.tree.Expression;
+import org.apache.calcite.linq4j.tree.ParameterExpression;
+import org.apache.calcite.rex.RexCall;
+
+/** Implementor of Functions used in MATCH_RECOGNIZE Context. */
+public interface MatchImplementor {
+
+ /**
+ * Implements a call.
+ *
+ * @param translator Translator for the call
+ * @param call Call that should be implemented
+ * @param row Current Row
+ * @param rows All Rows that are traversed so far
+ * @param symbols All Symbols of the rows that were traversed so far
+ * @return Translated call
+ */
+ Expression implement(
+ RexToLixTranslator translator,
+ RexCall call,
+ ParameterExpression row,
+ ParameterExpression rows,
+ ParameterExpression symbols,
+ ParameterExpression currentIndex);
+
+}
+
+// End MatchImplementor.java
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/MatchUtils.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/MatchUtils.java
new file mode 100644
index 0000000..a2a4451
--- /dev/null
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/MatchUtils.java
@@ -0,0 +1,51 @@
+/*
+ * 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.calcite.adapter.enumerable;
+
+import java.util.List;
+
+/** Class with static Helpers for MATCH_RECOGNIZE. */
+public class MatchUtils {
+
+ // Should not be instantiated
+ private MatchUtils() {
+ throw new IllegalStateException();
+ }
+
+ /**
+ * Returns the row with the highest index whose corresponding symbol matches, null otherwese.
+ * @param symbol Target Symbol
+ * @param rows List of passed rows
+ * @param symbols Corresponding symbols to rows
+ * @return index or -1
+ */
+ public static <E> int lastWithSymbol(String symbol, List<E> rows, List<String> symbols,
+ int startIndex) {
+ for (int i = startIndex; i >= 0; i--) {
+ if (symbol.equals(symbols.get(i))) {
+ return i;
+ }
+ }
+ return -1;
+ }
+
+ public static void print(int s) {
+ System.out.println(s);
+ }
+}
+
+// End MatchUtils.java
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/RexImpTable.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/RexImpTable.java
index 181a9f3..1737967 100644
--- a/core/src/main/java/org/apache/calcite/adapter/enumerable/RexImpTable.java
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/RexImpTable.java
@@ -20,6 +20,7 @@ import org.apache.calcite.avatica.util.DateTimeUtils;
import org.apache.calcite.avatica.util.TimeUnit;
import org.apache.calcite.avatica.util.TimeUnitRange;
import org.apache.calcite.linq4j.Ord;
+import org.apache.calcite.linq4j.tree.BinaryExpression;
import org.apache.calcite.linq4j.tree.BlockBuilder;
import org.apache.calcite.linq4j.tree.BlockStatement;
import org.apache.calcite.linq4j.tree.ConstantExpression;
@@ -39,6 +40,7 @@ import org.apache.calcite.rel.type.RelDataTypeFactoryImpl;
import org.apache.calcite.rex.RexCall;
import org.apache.calcite.rex.RexLiteral;
import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexPatternFieldRef;
import org.apache.calcite.runtime.SqlFunctions;
import org.apache.calcite.schema.ImplementableAggFunction;
import org.apache.calcite.schema.ImplementableFunction;
@@ -46,6 +48,7 @@ import org.apache.calcite.schema.impl.AggregateFunctionImpl;
import org.apache.calcite.sql.SqlAggFunction;
import org.apache.calcite.sql.SqlBinaryOperator;
import org.apache.calcite.sql.SqlJsonConstructorNullClause;
+import org.apache.calcite.sql.SqlMatchFunction;
import org.apache.calcite.sql.SqlOperator;
import org.apache.calcite.sql.fun.SqlJsonArrayAggAggFunction;
import org.apache.calcite.sql.fun.SqlJsonObjectAggAggFunction;
@@ -132,6 +135,7 @@ import static org.apache.calcite.sql.fun.SqlStdOperatorTable.CAST;
import static org.apache.calcite.sql.fun.SqlStdOperatorTable.CEIL;
import static org.apache.calcite.sql.fun.SqlStdOperatorTable.CHARACTER_LENGTH;
import static org.apache.calcite.sql.fun.SqlStdOperatorTable.CHAR_LENGTH;
+import static org.apache.calcite.sql.fun.SqlStdOperatorTable.CLASSIFIER;
import static org.apache.calcite.sql.fun.SqlStdOperatorTable.COALESCE;
import static org.apache.calcite.sql.fun.SqlStdOperatorTable.COLLECT;
import static org.apache.calcite.sql.fun.SqlStdOperatorTable.CONCAT;
@@ -156,7 +160,6 @@ import static org.apache.calcite.sql.fun.SqlStdOperatorTable.ELEMENT;
import static org.apache.calcite.sql.fun.SqlStdOperatorTable.EQUALS;
import static org.apache.calcite.sql.fun.SqlStdOperatorTable.EXP;
import static org.apache.calcite.sql.fun.SqlStdOperatorTable.EXTRACT;
-import static org.apache.calcite.sql.fun.SqlStdOperatorTable.FINAL;
import static org.apache.calcite.sql.fun.SqlStdOperatorTable.FIRST_VALUE;
import static org.apache.calcite.sql.fun.SqlStdOperatorTable.FLOOR;
import static org.apache.calcite.sql.fun.SqlStdOperatorTable.FUSION;
@@ -194,6 +197,7 @@ import static org.apache.calcite.sql.fun.SqlStdOperatorTable.JSON_QUERY;
import static org.apache.calcite.sql.fun.SqlStdOperatorTable.JSON_VALUE_ANY;
import static org.apache.calcite.sql.fun.SqlStdOperatorTable.JSON_VALUE_EXPRESSION;
import static org.apache.calcite.sql.fun.SqlStdOperatorTable.LAG;
+import static org.apache.calcite.sql.fun.SqlStdOperatorTable.LAST;
import static org.apache.calcite.sql.fun.SqlStdOperatorTable.LAST_DAY;
import static org.apache.calcite.sql.fun.SqlStdOperatorTable.LAST_VALUE;
import static org.apache.calcite.sql.fun.SqlStdOperatorTable.LEAD;
@@ -287,6 +291,8 @@ public class RexImpTable {
new HashMap<>();
private final Map<SqlAggFunction, Supplier<? extends WinAggImplementor>> winAggMap =
new HashMap<>();
+ private final Map<SqlMatchFunction, Supplier<? extends MatchImplementor>> matchMap =
+ new HashMap<>();
RexImpTable() {
defineMethod(ROW, BuiltInMethod.ARRAY.method, NullPolicy.ANY);
@@ -622,16 +628,9 @@ public class RexImpTable {
winAggMap.put(REGR_COUNT, constructorSupplier(CountWinImplementor.class));
// Functions for MATCH_RECOGNIZE
- defineMethod(FINAL, "abs", NullPolicy.ANY);
-
- map.put(PREV, (translator, call, nullAs) -> {
- final RexNode node = call.getOperands().get(0);
- final RexNode offset = call.getOperands().get(1);
- final Expression offs = Expressions.multiply(translator.translate(offset),
- Expressions.constant(-1));
- ((EnumerableMatch.PrevInputGetter) translator.inputGetter).setOffset(offs);
- return translator.translate(node, nullAs);
- });
+ matchMap.put(CLASSIFIER, ClassifierImplementor::new);
+ matchMap.put(LAST, LastImplementor::new);
+ map.put(PREV, new PrevImplementor());
}
private <T> Supplier<T> constructorSupplier(Class<T> klass) {
@@ -916,6 +915,16 @@ public class RexImpTable {
return aggSupplier.get();
}
+ public MatchImplementor get(final SqlMatchFunction function) {
+ final Supplier<? extends MatchImplementor> supplier =
+ matchMap.get(function);
+ if (supplier != null) {
+ return supplier.get();
+ } else {
+ throw new IllegalStateException("Supplier should not be null");
+ }
+ }
+
static Expression maybeNegate(boolean negate, Expression expression) {
if (!negate) {
return expression;
@@ -2925,6 +2934,75 @@ public class RexImpTable {
}
}
}
+
+ /** Implements CLASSIFIER match-recognize function. */
+ private static class ClassifierImplementor implements MatchImplementor {
+ @Override public Expression implement(RexToLixTranslator translator, RexCall call,
+ ParameterExpression row, ParameterExpression rows,
+ ParameterExpression symbols, ParameterExpression i) {
+ return Types.castIfNecessary(String.class,
+ Expressions.call(symbols, BuiltInMethod.LIST_GET.method, i));
+ }
+ }
+
+ /** Implements the LAST match-recognize function. */
+ private static class LastImplementor implements MatchImplementor {
+ @Override public Expression implement(RexToLixTranslator translator, RexCall call,
+ ParameterExpression row, ParameterExpression rows,
+ ParameterExpression symbols, ParameterExpression i) {
+ final RexNode node = call.getOperands().get(0);
+
+ final String alpha = ((RexPatternFieldRef) call.getOperands().get(0)).getAlpha();
+
+ final BinaryExpression lastIndex = Expressions.subtract(
+ Expressions.call(rows, BuiltInMethod.COLLECTION_SIZE.method),
+ Expressions.constant(1));
+
+ // Just take the last one, if exists
+ if ("*".equals(alpha)) {
+ ((EnumerableMatch.PassedRowsInputGetter) translator.inputGetter).setIndex(i);
+ // Important, unbox the node / expression to avoid NullAs.NOT_POSSIBLE
+ final RexPatternFieldRef ref = (RexPatternFieldRef) node;
+ final RexPatternFieldRef newRef =
+ new RexPatternFieldRef(ref.getAlpha(),
+ ref.getIndex(),
+ translator.typeFactory.createTypeWithNullability(ref.getType(),
+ true));
+ final Expression expression = translator.translate(newRef, NullAs.NULL);
+ ((EnumerableMatch.PassedRowsInputGetter) translator.inputGetter).setIndex(null);
+ return expression;
+ } else {
+ // Alpha != "*" so we have to search for a specific one to find and use that, if found
+ ((EnumerableMatch.PassedRowsInputGetter) translator.inputGetter).setIndex(
+ Expressions.call(BuiltInMethod.MATCH_UTILS_LAST_WITH_SYMBOL.method,
+ Expressions.constant(alpha), rows, symbols, i));
+
+ // Important, unbox the node / expression to avoid NullAs.NOT_POSSIBLE
+ final RexPatternFieldRef ref = (RexPatternFieldRef) node;
+ final RexPatternFieldRef newRef =
+ new RexPatternFieldRef(ref.getAlpha(),
+ ref.getIndex(),
+ translator.typeFactory.createTypeWithNullability(ref.getType(),
+ true));
+ final Expression expression = translator.translate(newRef, NullAs.NULL);
+ ((EnumerableMatch.PassedRowsInputGetter) translator.inputGetter).setIndex(null);
+ return expression;
+ }
+ }
+ }
+
+ /** Implements PREV match-recognize function. */
+ private static class PrevImplementor implements CallImplementor {
+ @Override public Expression implement(RexToLixTranslator translator, RexCall call,
+ NullAs nullAs) {
+ final RexNode node = call.getOperands().get(0);
+ final RexNode offset = call.getOperands().get(1);
+ final Expression offs = Expressions.multiply(translator.translate(offset),
+ Expressions.constant(-1));
+ ((EnumerableMatch.PrevInputGetter) translator.inputGetter).setOffset(offs);
+ return translator.translate(node, nullAs);
+ }
+ }
}
// End RexImpTable.java
diff --git a/core/src/main/java/org/apache/calcite/runtime/Enumerables.java b/core/src/main/java/org/apache/calcite/runtime/Enumerables.java
index b3054e5..374e82f 100644
--- a/core/src/main/java/org/apache/calcite/runtime/Enumerables.java
+++ b/core/src/main/java/org/apache/calcite/runtime/Enumerables.java
@@ -123,7 +123,7 @@ public class Enumerables {
matcher.matchOne(partitionState.getRows(), partitionState,
// TODO 26.12.18 jf: add row states (whatever this is?)
- list -> emitter.emit(list, null, null,
+ matches -> emitter.emit(matches.rows, null, matches.symbols,
matchCounter.getAndIncrement(), emitRows::add));
/*
recentRows.add(e);
diff --git a/core/src/main/java/org/apache/calcite/runtime/Matcher.java b/core/src/main/java/org/apache/calcite/runtime/Matcher.java
index a6a9b4c..d5a4c6b 100644
--- a/core/src/main/java/org/apache/calcite/runtime/Matcher.java
+++ b/core/src/main/java/org/apache/calcite/runtime/Matcher.java
@@ -72,14 +72,14 @@ public class Matcher<E> {
return new Builder<>(automaton);
}
- public List<List<E>> match(E... rows) {
+ public List<PartialMatch<E>> match(E... rows) {
return match(Arrays.asList(rows));
}
- public List<List<E>> match(Iterable<E> rows) {
- final ImmutableList.Builder<List<E>> resultMatchBuilder =
+ public List<PartialMatch<E>> match(Iterable<E> rows) {
+ final ImmutableList.Builder<PartialMatch<E>> resultMatchBuilder =
ImmutableList.builder();
- final Consumer<List<E>> resultMatchConsumer = resultMatchBuilder::add;
+ final Consumer<PartialMatch<E>> resultMatchConsumer = resultMatchBuilder::add;
final PartitionState<E> partitionState = createPartitionState(0, 0);
for (E row : rows) {
partitionState.getMemoryFactory().add(row);
@@ -98,10 +98,10 @@ public class Matcher<E> {
* This method ignores the symbols that caused a transition.
*/
protected void matchOne(MemoryFactory.Memory<E> rows,
- PartitionState<E> partitionState, Consumer<List<E>> resultMatches) {
+ PartitionState<E> partitionState, Consumer<PartialMatch<E>> resultMatches) {
List<PartialMatch<E>> matches = matchOneWithSymbols(rows, partitionState);
for (PartialMatch<E> pm : matches) {
- resultMatches.accept(pm.rows);
+ resultMatches.accept(pm);
}
}
diff --git a/core/src/main/java/org/apache/calcite/sql/SqlMatchFunction.java b/core/src/main/java/org/apache/calcite/sql/SqlMatchFunction.java
new file mode 100644
index 0000000..52c3bfc
--- /dev/null
+++ b/core/src/main/java/org/apache/calcite/sql/SqlMatchFunction.java
@@ -0,0 +1,33 @@
+/*
+ * 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.calcite.sql;
+
+import org.apache.calcite.sql.type.SqlOperandTypeChecker;
+import org.apache.calcite.sql.type.SqlOperandTypeInference;
+import org.apache.calcite.sql.type.SqlReturnTypeInference;
+
+/** Base class for all functions used in MATCH_RECOGNIZE. */
+public class SqlMatchFunction extends SqlFunction {
+
+ public SqlMatchFunction(String name, SqlKind kind, SqlReturnTypeInference returnTypeInference,
+ SqlOperandTypeInference operandTypeInference,
+ SqlOperandTypeChecker operandTypeChecker, SqlFunctionCategory category) {
+ super(name, kind, returnTypeInference, operandTypeInference, operandTypeChecker, category);
+ }
+}
+
+// End SqlMatchFunction.java
diff --git a/core/src/main/java/org/apache/calcite/sql/fun/SqlStdOperatorTable.java b/core/src/main/java/org/apache/calcite/sql/fun/SqlStdOperatorTable.java
index d803beb..377f4bb 100644
--- a/core/src/main/java/org/apache/calcite/sql/fun/SqlStdOperatorTable.java
+++ b/core/src/main/java/org/apache/calcite/sql/fun/SqlStdOperatorTable.java
@@ -31,6 +31,7 @@ import org.apache.calcite.sql.SqlJsonConstructorNullClause;
import org.apache.calcite.sql.SqlKind;
import org.apache.calcite.sql.SqlLateralOperator;
import org.apache.calcite.sql.SqlLiteral;
+import org.apache.calcite.sql.SqlMatchFunction;
import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.sql.SqlNullTreatmentOperator;
import org.apache.calcite.sql.SqlNumericLiteral;
@@ -1735,13 +1736,13 @@ public class SqlStdOperatorTable extends ReflectiveSqlOperatorTable {
null, OperandTypes.ANY_NUMERIC, SqlFunctionCategory.MATCH_RECOGNIZE);
/** {@code LAST} function to be used within {@code MATCH_RECOGNIZE}. */
- public static final SqlFunction LAST =
- new SqlFunction("LAST", SqlKind.LAST, ReturnTypes.ARG0_NULLABLE,
+ public static final SqlMatchFunction LAST =
+ new SqlMatchFunction("LAST", SqlKind.LAST, ReturnTypes.ARG0_NULLABLE,
null, OperandTypes.ANY_NUMERIC, SqlFunctionCategory.MATCH_RECOGNIZE);
/** {@code PREV} function to be used within {@code MATCH_RECOGNIZE}. */
- public static final SqlFunction PREV =
- new SqlFunction("PREV", SqlKind.PREV, ReturnTypes.ARG0_NULLABLE,
+ public static final SqlMatchFunction PREV =
+ new SqlMatchFunction("PREV", SqlKind.PREV, ReturnTypes.ARG0_NULLABLE,
null, OperandTypes.ANY_NUMERIC, SqlFunctionCategory.MATCH_RECOGNIZE);
/** {@code NEXT} function to be used within {@code MATCH_RECOGNIZE}. */
@@ -1750,8 +1751,8 @@ public class SqlStdOperatorTable extends ReflectiveSqlOperatorTable {
OperandTypes.ANY_NUMERIC, SqlFunctionCategory.MATCH_RECOGNIZE);
/** {@code CLASSIFIER} function to be used within {@code MATCH_RECOGNIZE}. */
- public static final SqlFunction CLASSIFIER =
- new SqlFunction("CLASSIFIER", SqlKind.CLASSIFIER, ReturnTypes.VARCHAR_2000,
+ public static final SqlMatchFunction CLASSIFIER =
+ new SqlMatchFunction("CLASSIFIER", SqlKind.CLASSIFIER, ReturnTypes.VARCHAR_2000,
null, OperandTypes.NILADIC, SqlFunctionCategory.MATCH_RECOGNIZE);
/** {@code MATCH_NUMBER} function to be used within {@code MATCH_RECOGNIZE}. */
diff --git a/core/src/main/java/org/apache/calcite/util/BuiltInMethod.java b/core/src/main/java/org/apache/calcite/util/BuiltInMethod.java
index 0507204..a8959c1 100644
--- a/core/src/main/java/org/apache/calcite/util/BuiltInMethod.java
+++ b/core/src/main/java/org/apache/calcite/util/BuiltInMethod.java
@@ -21,6 +21,7 @@ import org.apache.calcite.adapter.enumerable.AggregateLambdaFactory;
import org.apache.calcite.adapter.enumerable.BasicAggregateLambdaFactory;
import org.apache.calcite.adapter.enumerable.BasicLazyAccumulator;
import org.apache.calcite.adapter.enumerable.LazyAggregateLambdaFactory;
+import org.apache.calcite.adapter.enumerable.MatchUtils;
import org.apache.calcite.adapter.enumerable.SourceSorter;
import org.apache.calcite.adapter.java.ReflectiveSchema;
import org.apache.calcite.adapter.jdbc.JdbcSchema;
@@ -180,6 +181,8 @@ public enum BuiltInMethod {
MATCHER_BUILDER_ADD(Matcher.Builder.class, "add", String.class,
Predicate.class),
MATCHER_BUILDER_BUILD(Matcher.Builder.class, "build"),
+ MATCH_UTILS_LAST_WITH_SYMBOL(MatchUtils.class, "lastWithSymbol", String.class,
+ List.class, List.class, int.class),
EMITTER_EMIT(Enumerables.Emitter.class, "emit", List.class, List.class,
List.class, int.class, Consumer.class),
MERGE_JOIN(EnumerableDefaults.class, "mergeJoin", Enumerable.class,
diff --git a/core/src/test/java/org/apache/calcite/plan/RelWriterTest.java b/core/src/test/java/org/apache/calcite/plan/RelWriterTest.java
index 701a136..d563b8f 100644
--- a/core/src/test/java/org/apache/calcite/plan/RelWriterTest.java
+++ b/core/src/test/java/org/apache/calcite/plan/RelWriterTest.java
@@ -631,12 +631,9 @@ public class RelWriterTest {
final RelDataTypeField field = scan.getRowType().getField("SAL", false, false);
programBuilder.addIdentity();
programBuilder.addCondition(
- rexBuilder.makeCall(
- SqlStdOperatorTable.GREATER_THAN,
+ rexBuilder.makeCall(SqlStdOperatorTable.GREATER_THAN,
new RexInputRef(field.getIndex(), field.getType()),
- builder.literal(10)
- )
- );
+ builder.literal(10)));
final LogicalCalc calc = LogicalCalc.create(scan, programBuilder.getProgram());
String relJson = RelOptUtil.dumpPlan("", calc,
SqlExplainFormat.JSON, SqlExplainLevel.EXPPLAN_ATTRIBUTES);
@@ -682,8 +679,7 @@ public class RelWriterTest {
+ " LogicalFilter(condition=[=($0, $cor0.DEPTNO)])\n"
+ " LogicalTableScan(table=[[scott, DEPT]])\n";
- assertThat(s, isLinux(expected)
- );
+ assertThat(s, isLinux(expected));
}
/** Returns the schema of a {@link org.apache.calcite.rel.core.TableScan}
diff --git a/core/src/test/java/org/apache/calcite/runtime/AutomatonTest.java b/core/src/test/java/org/apache/calcite/runtime/AutomatonTest.java
index e7ec2db..38be52f 100644
--- a/core/src/test/java/org/apache/calcite/runtime/AutomatonTest.java
+++ b/core/src/test/java/org/apache/calcite/runtime/AutomatonTest.java
@@ -17,18 +17,35 @@
package org.apache.calcite.runtime;
import org.apache.calcite.linq4j.MemoryFactory;
+import org.apache.calcite.test.Matchers;
import com.google.common.collect.ImmutableList;
+import org.hamcrest.Factory;
+import org.hamcrest.core.Is;
import org.junit.Test;
import java.util.AbstractList;
+import java.util.List;
+import java.util.stream.Collectors;
import static org.hamcrest.core.Is.is;
import static org.junit.Assert.assertThat;
/** Unit tests for {@link Automaton}. */
public class AutomatonTest {
+
+ /** Creates a Matcher that matches a list of
+ * {@link org.apache.calcite.runtime.Matcher.PartialMatch} if they
+ * a formatted to a given string. */
+ @Factory
+ private static <E> org.hamcrest.Matcher<List<Matcher.PartialMatch<E>>>
+ isMatchList(final String value) {
+ return Matchers.compose(Is.is(value),
+ match -> match.stream().map(pm -> pm.rows).collect(Collectors.toList())
+ .toString());
+ }
+
@Test public void testSimple() {
// pattern(a)
final Pattern p = Pattern.builder().symbol("a").build();
@@ -40,7 +57,8 @@ public class AutomatonTest {
.add("a", s -> s.get().contains("a"))
.build();
final String expected = "[[a], [a]]";
- assertThat(matcher.match(rows).toString(), is(expected));
+
+ assertThat(matcher.match(rows), isMatchList(expected));
}
@Test public void testSequence() {
@@ -56,7 +74,7 @@ public class AutomatonTest {
.add("b", s -> s.get().contains("b"))
.build();
final String expected = "[[a, ab], [ab, b]]";
- assertThat(matcher.match(rows).toString(), is(expected));
+ assertThat(matcher.match(rows), isMatchList(expected));
}
@Test public void testStar() {
@@ -74,7 +92,7 @@ public class AutomatonTest {
.build();
final String expected = "[[b], [ab], [ab], [ab, a, ab], [a, ab], [b], [ab, b], [ab, a, ab, b], "
+ "[a, ab, b], [b]]";
- assertThat(matcher.match(rows).toString(), is(expected));
+ assertThat(matcher.match(rows), isMatchList(expected));
}
@Test public void testPlus() {
@@ -91,7 +109,7 @@ public class AutomatonTest {
.add("b", s -> s.get().contains("b"))
.build();
final String expected = "[[ab, a, ab], [a, ab], [ab, b], [ab, a, ab, b], [a, ab, b]]";
- assertThat(matcher.match(rows).toString(), is(expected));
+ assertThat(matcher.match(rows), isMatchList(expected));
}
@Test public void testOr() {
@@ -109,7 +127,7 @@ public class AutomatonTest {
.add("b", s -> s.get().contains("b"))
.build();
final String expected = "[[a], [b], [ab], [ab], [a], [ab], [ab], [b], [b]]";
- assertThat(matcher.match(rows).toString(), is(expected));
+ assertThat(matcher.match(rows), isMatchList(expected));
}
@Test public void testOptional() {
@@ -129,7 +147,7 @@ public class AutomatonTest {
.add("c", s -> s.get() == 'c')
.build();
final String expected = "[[a, c], [a, b, c]]";
- assertThat(matcher.match(chars(rows)).toString(), is(expected));
+ assertThat(matcher.match(chars(rows)), isMatchList(expected));
}
@Test public void testRepeat() {
@@ -164,7 +182,7 @@ public class AutomatonTest {
.add("b", s -> s.get() == 'b')
.add("c", s -> s.get() == 'c')
.build();
- assertThat(matcher.match(chars(rows)).toString(), is(expected));
+ assertThat(matcher.match(chars(rows)), isMatchList(expected));
}
@Test public void testRepeatComposite() {
@@ -184,8 +202,8 @@ public class AutomatonTest {
.add("b", s -> s.get() == 'b')
.add("c", s -> s.get() == 'c')
.build();
- assertThat(matcher.match(chars(rows)).toString(),
- is("[[a, b, a, c], [a, b, a, c], [a, b, a, b, a, c]]"));
+ assertThat(matcher.match(chars(rows)),
+ isMatchList("[[a, b, a, c], [a, b, a, c], [a, b, a, b, a, c]]"));
}
@Test public void testResultWithLabels() {
diff --git a/core/src/test/resources/sql/match.iq b/core/src/test/resources/sql/match.iq
index 1aa8ef9..3df3753 100755
--- a/core/src/test/resources/sql/match.iq
+++ b/core/src/test/resources/sql/match.iq
@@ -110,6 +110,56 @@ C EMPID
!ok
+# Test Classifier
+select *
+from "hr"."emps" match_recognize (
+ order by "empid" desc
+ measures "commission" as c,
+ "empid" as empid,
+ CLASSIFIER() as cl
+ pattern (s up)
+ define up as up."commission" < prev(up."commission"));
+
+C EMPID CL
+---- ----- --
+1000 100 S
+ 500 200 UP
+
+!ok
+
+# Test Simple LAST
+select *
+from "hr"."emps" match_recognize (
+ order by "empid" desc
+ measures "commission" as c,
+ LAST("empid") as empid
+ pattern (s up)
+ define up as up."commission" < prev(up."commission"));
+
+C EMPID
+---- -----
+1000 100
+ 500 200
+
+!ok
+
+# Test LAST with Classifier
+select *
+from "hr"."emps" match_recognize (
+ order by "empid" desc
+ measures "commission" as c,
+ CLASSIFIER() as cl,
+ LAST(S."empid") as empid
+ pattern (s up)
+ define up as up."commission" < prev(up."commission"));
+
+C CL EMPID
+---- -- -----
+1000 S 100
+ 500 UP 100
+
+!ok
+
!if (false) {
# Match recognize
SELECT *
diff --git a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchSchemaFactory.java b/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchSchemaFactory.java
index a5a01ea..9b6997e 100644
--- a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchSchemaFactory.java
+++ b/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchSchemaFactory.java
@@ -37,7 +37,6 @@ import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;
-
/**
* Factory that creates an {@link ElasticsearchSchema}.
*