This is an automated email from the ASF dual-hosted git repository. twalthr pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push: new 7a27357 [FLINK-11908][table] Port window classes into flink-table-api-java 7a27357 is described below commit 7a273577021bf84927242c9372757215e1ff9fc3 Author: hequn8128 <chenghe...@gmail.com> AuthorDate: Sun Mar 17 12:57:01 2019 +0800 [FLINK-11908][table] Port window classes into flink-table-api-java This closes #7976. --- .../org/apache/flink/table/api/GroupWindow.java | 55 ++ .../main/java/org/apache/flink/table/api/Over.java | 105 +++ .../org/apache/flink/table/api/OverWindow.java | 74 ++ .../flink/table/api/OverWindowPartitioned.java | 69 ++ .../table/api/OverWindowPartitionedOrdered.java | 89 +++ .../api/OverWindowPartitionedOrderedPreceding.java | 88 +++ .../java/org/apache/flink/table/api/Session.java | 74 ++ .../org/apache/flink/table/api/SessionWithGap.java | 71 ++ .../flink/table/api/SessionWithGapOnTime.java | 62 ++ .../table/api/SessionWithGapOnTimeWithAlias.java | 40 ++ .../java/org/apache/flink/table/api/Slide.java | 84 +++ .../org/apache/flink/table/api/SlideWithSize.java | 71 ++ .../flink/table/api/SlideWithSizeAndSlide.java | 73 ++ .../table/api/SlideWithSizeAndSlideOnTime.java | 67 ++ .../api/SlideWithSizeAndSlideOnTimeWithAlias.java | 50 ++ .../java/org/apache/flink/table/api/Tumble.java | 72 ++ .../org/apache/flink/table/api/TumbleWithSize.java | 71 ++ .../flink/table/api/TumbleWithSizeOnTime.java | 62 ++ .../table/api/TumbleWithSizeOnTimeWithAlias.java | 40 ++ .../flink/table/expressions/ExpressionParser.java | 41 ++ .../table/expressions/PlannerExpressionParser.java | 68 ++ .../table/api/WindowCreationValidationTest.java | 60 ++ .../table/api/java/BatchTableEnvironment.scala | 6 +- .../table/api/java/StreamTableEnvironment.scala | 5 +- .../org/apache/flink/table/api/java/windows.scala | 64 -- .../flink/table/api/scala/expressionDsl.scala | 2 +- .../org/apache/flink/table/api/scala/windows.scala | 68 -- .../scala/org/apache/flink/table/api/table.scala | 64 +- .../scala/org/apache/flink/table/api/windows.scala | 741 --------------------- ...ser.scala => PlannerExpressionParserImpl.scala} | 25 +- .../GroupWindowStringExpressionTest.scala | 2 +- .../runtime/stream/TimeAttributesITCase.scala | 6 +- 32 files changed, 1530 insertions(+), 939 deletions(-) diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/GroupWindow.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/GroupWindow.java new file mode 100644 index 0000000..4601e75 --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/GroupWindow.java @@ -0,0 +1,55 @@ +/* + * 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.flink.table.api; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.expressions.Expression; + +/** + * A group window specification. + * + * <p>Group windows group rows based on time or row-count intervals and is therefore essentially a + * special type of groupBy. Just like groupBy, group windows allow to compute aggregates + * on groups of elements. + * + * <p>Infinite streaming tables can only be grouped into time or row intervals. Hence window + * grouping is required to apply aggregations on streaming tables. + * + * <p>For finite batch tables, group windows provide shortcuts for time-based groupBy. + */ +@PublicEvolving +public abstract class GroupWindow { + + /** Alias name for the group window. */ + private final Expression alias; + private final Expression timeField; + + GroupWindow(Expression alias, Expression timeField) { + this.alias = alias; + this.timeField = timeField; + } + + public Expression getAlias() { + return alias; + } + + public Expression getTimeField() { + return timeField; + } +} diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Over.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Over.java new file mode 100644 index 0000000..52acf85 --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Over.java @@ -0,0 +1,105 @@ +/* + * 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.flink.table.api; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.expressions.ExpressionParser; + +import java.util.Arrays; + +/** + * Helper class for creating an over window. Similar to SQL, over window aggregates compute an + * aggregate for each input row over a range of its neighboring rows. + * + * <p>Java Example: + * + * <pre> + * {@code + * Over.partitionBy("a").orderBy("rowtime").preceding("unbounded_range").as("w") + * } + * </pre> + * + * <p>Scala Example: + * + * <pre> + * {@code + * Over partitionBy 'a orderBy 'rowtime preceding UNBOUNDED_RANGE as 'w + * } + * </pre> + */ +@PublicEvolving +public final class Over { + + /** + * Partitions the elements on some partition keys. + * + * <p>Each partition is individually sorted and aggregate functions are applied to each + * partition separately. + * + * @param partitionBy list of field references + * @return an over window with defined partitioning + */ + public static OverWindowPartitioned partitionBy(String partitionBy) { + return new OverWindowPartitioned(ExpressionParser.parseExpressionList(partitionBy)); + } + + /** + * Partitions the elements on some partition keys. + * + * <p>Each partition is individually sorted and aggregate functions are applied to each + * partition separately. + * + * @param partitionBy list of field references + * @return an over window with defined partitioning + */ + public static OverWindowPartitioned partitionBy(Expression... partitionBy) { + return new OverWindowPartitioned(Arrays.asList(partitionBy)); + } + + /** + * Specifies the time attribute on which rows are ordered. + * + * <p>For streaming tables, reference a rowtime or proctime time attribute here + * to specify the time mode. + * + * <p>For batch tables, refer to a timestamp or long attribute. + * + * @param orderBy field reference + * @return an over window with defined order + */ + public static OverWindowPartitionedOrdered orderBy(String orderBy) { + return partitionBy().orderBy(orderBy); + } + + /** + * Specifies the time attribute on which rows are ordered. + * + * <p>For streaming tables, reference a rowtime or proctime time attribute here + * to specify the time mode. + * + * <p>For batch tables, refer to a timestamp or long attribute. + * + * @param orderBy field reference + * @return an over window with defined order + */ + public static OverWindowPartitionedOrdered orderBy(Expression orderBy) { + return partitionBy().orderBy(orderBy); + } +} diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindow.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindow.java new file mode 100644 index 0000000..852144d --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindow.java @@ -0,0 +1,74 @@ +/* + * 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.flink.table.api; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.expressions.Expression; + +import java.util.List; +import java.util.Optional; + +/** + * An over window specification. + * + * <p>Similar to SQL, over window aggregates compute an aggregate for each input row over a range + * of its neighboring rows. + */ +@PublicEvolving +public final class OverWindow { + + private final Expression alias; + private final List<Expression> partitioning; + private final Expression order; + private final Expression preceding; + private final Optional<Expression> following; + + OverWindow( + Expression alias, + List<Expression> partitionBy, + Expression orderBy, + Expression preceding, + Optional<Expression> following) { + this.alias = alias; + this.partitioning = partitionBy; + this.order = orderBy; + this.preceding = preceding; + this.following = following; + } + + public Expression getAlias() { + return alias; + } + + public List<Expression> getPartitioning() { + return partitioning; + } + + public Expression getOrder() { + return order; + } + + public Expression getPreceding() { + return preceding; + } + + public Optional<Expression> getFollowing() { + return following; + } +} diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindowPartitioned.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindowPartitioned.java new file mode 100644 index 0000000..21f6292 --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindowPartitioned.java @@ -0,0 +1,69 @@ +/* + * 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.flink.table.api; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.expressions.ExpressionParser; + +import java.util.List; + +/** + * Partially defined over window with partitioning. + */ +@PublicEvolving +public final class OverWindowPartitioned { + + /** Defines a partitioning of the input on one or more attributes. */ + private final List<Expression> partitionBy; + + OverWindowPartitioned(List<Expression> partitionBy) { + this.partitionBy = partitionBy; + } + + /** + * Specifies the time attribute on which rows are ordered. + * + * <p>For streaming tables, reference a rowtime or proctime time attribute here + * to specify the time mode. + * + * <p>For batch tables, refer to a timestamp or long attribute. + * + * @param orderBy field reference + * @return an over window with defined order + */ + public OverWindowPartitionedOrdered orderBy(String orderBy) { + return this.orderBy(ExpressionParser.parseExpression(orderBy)); + } + + /** + * Specifies the time attribute on which rows are ordered. + * + * <p>For streaming tables, reference a rowtime or proctime time attribute here + * to specify the time mode. + * + * <p>For batch tables, refer to a timestamp or long attribute. + * + * @param orderBy field reference + * @return an over window with defined order + */ + public OverWindowPartitionedOrdered orderBy(Expression orderBy) { + return new OverWindowPartitionedOrdered(partitionBy, orderBy); + } +} diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindowPartitionedOrdered.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindowPartitionedOrdered.java new file mode 100644 index 0000000..04b3e75 --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindowPartitionedOrdered.java @@ -0,0 +1,89 @@ +/* + * 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.flink.table.api; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.expressions.BuiltInFunctionDefinitions; +import org.apache.flink.table.expressions.CallExpression; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.expressions.ExpressionParser; + +import java.util.Collections; +import java.util.List; +import java.util.Optional; + +/** + * Partially defined over window with (optional) partitioning and order. + */ +@PublicEvolving +public final class OverWindowPartitionedOrdered { + + private final List<Expression> partitionBy; + private final Expression orderBy; + + OverWindowPartitionedOrdered(List<Expression> partitionBy, Expression orderBy) { + this.partitionBy = partitionBy; + this.orderBy = orderBy; + } + + /** + * Set the preceding offset (based on time or row-count intervals) for over window. + * + * @param preceding preceding offset relative to the current row. + * @return an over window with defined preceding + */ + public OverWindowPartitionedOrderedPreceding preceding(String preceding) { + return this.preceding(ExpressionParser.parseExpression(preceding)); + } + + /** + * Set the preceding offset (based on time or row-count intervals) for over window. + * + * @param preceding preceding offset relative to the current row. + * @return an over window with defined preceding + */ + public OverWindowPartitionedOrderedPreceding preceding(Expression preceding) { + return new OverWindowPartitionedOrderedPreceding(partitionBy, orderBy, preceding); + } + + /** + * Assigns an alias for this window that the following {@code select()} clause can refer to. + * + * @param alias alias for this over window + * @return the fully defined over window + */ + public OverWindow as(String alias) { + return as(ExpressionParser.parseExpression(alias)); + } + + /** + * Assigns an alias for this window that the following {@code select()} clause can refer to. + * + * @param alias alias for this over window + * @return the fully defined over window + */ + public OverWindow as(Expression alias) { + return new OverWindow( + alias, + partitionBy, + orderBy, + new CallExpression(BuiltInFunctionDefinitions.UNBOUNDED_RANGE, Collections.emptyList()), + Optional.empty()); + } +} diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindowPartitionedOrderedPreceding.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindowPartitionedOrderedPreceding.java new file mode 100644 index 0000000..f87a210 --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/OverWindowPartitionedOrderedPreceding.java @@ -0,0 +1,88 @@ +/* + * 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.flink.table.api; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.expressions.ExpressionParser; + +import java.util.List; +import java.util.Optional; + +/** + * Partially defined over window with (optional) partitioning, order, and preceding. + */ +@PublicEvolving +public final class OverWindowPartitionedOrderedPreceding { + + private final List<Expression> partitionBy; + private final Expression orderBy; + private final Expression preceding; + private Optional<Expression> optionalFollowing = Optional.empty(); + + OverWindowPartitionedOrderedPreceding( + List<Expression> partitionBy, + Expression orderBy, + Expression preceding) { + this.partitionBy = partitionBy; + this.orderBy = orderBy; + this.preceding = preceding; + } + + /** + * Assigns an alias for this window that the following {@code select()} clause can refer to. + * + * @param alias alias for this over window + * @return the fully defined over window + */ + public OverWindow as(String alias) { + return as(ExpressionParser.parseExpression(alias)); + } + + /** + * Assigns an alias for this window that the following {@code select()} clause can refer to. + * + * @param alias alias for this over window + * @return the fully defined over window + */ + public OverWindow as(Expression alias) { + return new OverWindow(alias, partitionBy, orderBy, preceding, optionalFollowing); + } + + /** + * Set the following offset (based on time or row-count intervals) for over window. + * + * @param following following offset that relative to the current row. + * @return an over window with defined following + */ + public OverWindowPartitionedOrderedPreceding following(String following) { + return this.following(ExpressionParser.parseExpression(following)); + } + + /** + * Set the following offset (based on time or row-count intervals) for over window. + * + * @param following following offset that relative to the current row. + * @return an over window with defined following + */ + public OverWindowPartitionedOrderedPreceding following(Expression following) { + optionalFollowing = Optional.of(following); + return this; + } +} diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Session.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Session.java new file mode 100644 index 0000000..2d27548 --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Session.java @@ -0,0 +1,74 @@ +/* + * 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.flink.table.api; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.expressions.ExpressionParser; + +/** + * Helper class for creating a session window. The boundary of session windows are defined by + * intervals of inactivity, i.e., a session window is closes if no event appears for a defined + * gap period. + * + * <p>Java Example: + * + * <pre> + * {@code + * Session.withGap("10.minutes").on("rowtime").as("w") + * } + * </pre> + * + * <p>Scala Example: + * + * <pre> + * {@code + * Session withGap 10.minutes on 'rowtime as 'w + * } + * </pre> + */ +@PublicEvolving +public final class Session { + + /** + * Creates a session window. The boundary of session windows are defined by + * intervals of inactivity, i.e., a session window is closes if no event appears for a defined + * gap period. + * + * @param gap specifies how long (as interval of milliseconds) to wait for new data before + * closing the session window. + * @return a partially defined session window + */ + public static SessionWithGap withGap(String gap) { + return withGap(ExpressionParser.parseExpression(gap)); + } + + /** + * Creates a session window. The boundary of session windows are defined by + * intervals of inactivity, i.e., a session window is closes if no event appears for a defined + * gap period. + * + * @param gap specifies how long (as interval of milliseconds) to wait for new data before + * closing the session window. + * @return a partially defined session window + */ + public static SessionWithGap withGap(Expression gap) { + return new SessionWithGap(gap); + } +} diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/SessionWithGap.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/SessionWithGap.java new file mode 100644 index 0000000..6e646f7 --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/SessionWithGap.java @@ -0,0 +1,71 @@ +/* + * 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.flink.table.api; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.expressions.ExpressionParser; + +/** + * Session window. + * + * <p>For streaming tables you can specify grouping by a event-time or processing-time attribute. + * + * <p>For batch tables you can specify grouping on a timestamp or long attribute. + */ +@PublicEvolving +public final class SessionWithGap { + + /** The time interval of inactivity before a window is closed. */ + private final Expression gap; + + SessionWithGap(Expression gap) { + this.gap = gap; + } + + /** + * Specifies the time attribute on which rows are grouped. + * + * <p>For streaming tables you can specify grouping by a event-time or processing-time + * attribute. + * + * <p>For batch tables you can specify grouping on a timestamp or long attribute. + * + * @param timeField time attribute for streaming and batch tables + * @return a tumbling window on event-time + */ + public SessionWithGapOnTime on(String timeField) { + return on(ExpressionParser.parseExpression(timeField)); + } + + /** + * Specifies the time attribute on which rows are grouped. + * + * <p>For streaming tables you can specify grouping by a event-time or processing-time + * attribute. + * + * <p>For batch tables you can specify grouping on a timestamp or long attribute. + * + * @param timeField time attribute for streaming and batch tables + * @return a tumbling window on event-time + */ + public SessionWithGapOnTime on(Expression timeField) { + return new SessionWithGapOnTime(timeField, gap); + } +} diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/SessionWithGapOnTime.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/SessionWithGapOnTime.java new file mode 100644 index 0000000..53c9c4f --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/SessionWithGapOnTime.java @@ -0,0 +1,62 @@ +/* + * 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.flink.table.api; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.expressions.ExpressionParser; + +/** + * Session window on time. + */ +@PublicEvolving +public final class SessionWithGapOnTime { + + private final Expression timeField; + private final Expression gap; + + SessionWithGapOnTime(Expression timeField, Expression gap) { + this.timeField = timeField; + this.gap = gap; + } + + /** + * Assigns an alias for this window that the following {@code groupBy()} and {@code select()} + * clause can refer to. {@code select()} statement can access window properties such as window + * start or end time. + * + * @param alias alias for this window + * @return this window + */ + public SessionWithGapOnTimeWithAlias as(String alias) { + return as(ExpressionParser.parseExpression(alias)); + } + + /** + * Assigns an alias for this window that the following {@code groupBy()} and {@code select()} + * clause can refer to. {@code select()} statement can access window properties such as window + * start or end time. + * + * @param alias alias for this window + * @return this window + */ + public SessionWithGapOnTimeWithAlias as(Expression alias) { + return new SessionWithGapOnTimeWithAlias(alias, timeField, gap); + } +} diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/SessionWithGapOnTimeWithAlias.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/SessionWithGapOnTimeWithAlias.java new file mode 100644 index 0000000..891ea7c --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/SessionWithGapOnTimeWithAlias.java @@ -0,0 +1,40 @@ +/* + * 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.flink.table.api; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.expressions.Expression; + +/** + * Session window on time with alias. Fully specifies a window. + */ +@PublicEvolving +public final class SessionWithGapOnTimeWithAlias extends GroupWindow { + + private final Expression gap; + + SessionWithGapOnTimeWithAlias(Expression alias, Expression timeField, Expression gap) { + super(alias, timeField); + this.gap = gap; + } + + public Expression getGap() { + return gap; + } +} diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Slide.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Slide.java new file mode 100644 index 0000000..85f79ce --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Slide.java @@ -0,0 +1,84 @@ +/* + * 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.flink.table.api; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.expressions.ExpressionParser; + +/** + * Helper class for creating a sliding window. Sliding windows have a fixed size and slide by + * a specified slide interval. If the slide interval is smaller than the window size, sliding + * windows are overlapping. Thus, an element can be assigned to multiple windows. + * + * <p>For example, a sliding window of size 15 minutes with 5 minutes sliding interval groups + * elements of 15 minutes and evaluates every five minutes. Each element is contained in three + * consecutive window evaluations. + * + * <p>Java Example: + * + * <pre> + * {@code + * Slide.over("10.minutes").every("5.minutes").on("rowtime").as("w") + * } + * </pre> + * + * <p>Scala Example: + * + * <pre> + * {@code + * Slide over 10.minutes every 5.minutes on 'rowtime as 'w + * } + * </pre> + */ +@PublicEvolving +public final class Slide { + + /** + * Creates a sliding window. Sliding windows have a fixed size and slide by + * a specified slide interval. If the slide interval is smaller than the window size, sliding + * windows are overlapping. Thus, an element can be assigned to multiple windows. + * + * <p>For example, a sliding window of size 15 minutes with 5 minutes sliding interval groups + * elements of 15 minutes and evaluates every five minutes. Each element is contained in three + * consecutive window evaluations. + * + * @param size the size of the window as time or row-count interval + * @return a partially specified sliding window + */ + public static SlideWithSize over(String size) { + return over(ExpressionParser.parseExpression(size)); + } + + /** + * Creates a sliding window. Sliding windows have a fixed size and slide by + * a specified slide interval. If the slide interval is smaller than the window size, sliding + * windows are overlapping. Thus, an element can be assigned to multiple windows. + * + * <p>For example, a sliding window of size 15 minutes with 5 minutes sliding interval groups + * elements of 15 minutes and evaluates every five minutes. Each element is contained in three + * consecutive + * + * @param size the size of the window as time or row-count interval + * @return a partially specified sliding window + */ + public static SlideWithSize over(Expression size) { + return new SlideWithSize(size); + } +} diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/SlideWithSize.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/SlideWithSize.java new file mode 100644 index 0000000..44470f9 --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/SlideWithSize.java @@ -0,0 +1,71 @@ +/* + * 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.flink.table.api; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.expressions.ExpressionParser; + +/** + * Partially specified sliding window. The size of the window either as time or row-count interval. + */ +@PublicEvolving +public final class SlideWithSize { + + /** The size of the window either as time or row-count interval. */ + private final Expression size; + + SlideWithSize(Expression size) { + this.size = size; + } + + /** + * Specifies the window's slide as time or row-count interval. + * + * <p>The slide determines the interval in which windows are started. Hence, sliding windows can + * overlap if the slide is smaller than the size of the window. + * + * <p>For example, you could have windows of size 15 minutes that slide by 3 minutes. With this + * 15 minutes worth of elements are grouped every 3 minutes and each row contributes to 5 + * windows. + * + * @param slide the slide of the window either as time or row-count interval. + * @return a sliding window + */ + public SlideWithSizeAndSlide every(String slide) { + return every(ExpressionParser.parseExpression(slide)); + } + + /** + * Specifies the window's slide as time or row-count interval. + * + * <p>The slide determines the interval in which windows are started. Hence, sliding windows can + * overlap if the slide is smaller than the size of the window. + * + * <p>For example, you could have windows of size 15 minutes that slide by 3 minutes. With this + * 15 minutes worth of elements are grouped every 3 minutes and each row contributes to 5 + * windows. + * + * @param slide the slide of the window either as time or row-count interval. + * @return a sliding window + */ + public SlideWithSizeAndSlide every(Expression slide) { + return new SlideWithSizeAndSlide(size, slide); + } +} diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/SlideWithSizeAndSlide.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/SlideWithSizeAndSlide.java new file mode 100644 index 0000000..4d509d1 --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/SlideWithSizeAndSlide.java @@ -0,0 +1,73 @@ +/* + * 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.flink.table.api; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.expressions.ExpressionParser; + +/** + * Sliding window. The size of the window either as time or row-count interval. + * + * <p>For streaming tables you can specify grouping by a event-time or processing-time attribute. + * + * <p>For batch tables you can specify grouping on a timestamp or long attribute. + */ +@PublicEvolving +public final class SlideWithSizeAndSlide { + + /** The size of the window either as time or row-count interval. */ + private final Expression size; + private final Expression slide; + + SlideWithSizeAndSlide(Expression size, Expression slide) { + this.size = size; + this.slide = slide; + } + + /** + * Specifies the time attribute on which rows are grouped. + * + * <p>For streaming tables you can specify grouping by a event-time or processing-time + * attribute. + * + * <p>For batch tables you can specify grouping on a timestamp or long attribute. + * + * @param timeField time attribute for streaming and batch tables + * @return a tumbling window on event-time + */ + public SlideWithSizeAndSlideOnTime on(String timeField) { + return on(ExpressionParser.parseExpression(timeField)); + } + + /** + * Specifies the time attribute on which rows are grouped. + * + * <p>For streaming tables you can specify grouping by a event-time or processing-time + * attribute. + * + * <p>For batch tables you can specify grouping on a timestamp or long attribute. + * + * @param timeField time attribute for streaming and batch tables + * @return a tumbling window on event-time + */ + public SlideWithSizeAndSlideOnTime on(Expression timeField) { + return new SlideWithSizeAndSlideOnTime(timeField, size, slide); + } +} diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/SlideWithSizeAndSlideOnTime.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/SlideWithSizeAndSlideOnTime.java new file mode 100644 index 0000000..604b9cd --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/SlideWithSizeAndSlideOnTime.java @@ -0,0 +1,67 @@ +/* + * 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.flink.table.api; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.expressions.ExpressionParser; + +/** + * Sliding window on time. + */ +@PublicEvolving +public final class SlideWithSizeAndSlideOnTime { + + private final Expression timeField; + private final Expression size; + private final Expression slide; + + SlideWithSizeAndSlideOnTime( + Expression timeField, + Expression size, + Expression slide) { + this.timeField = timeField; + this.size = size; + this.slide = slide; + } + + /** + * Assigns an alias for this window that the following {@code groupBy()} and {@code select()} + * clause can refer to. {@code select()} statement can access window properties such as window + * start or end time. + * + * @param alias alias for this window + * @return this window + */ + public SlideWithSizeAndSlideOnTimeWithAlias as(String alias) { + return as(ExpressionParser.parseExpression(alias)); + } + + /** + * Assigns an alias for this window that the following {@code groupBy()} and {@code select()} + * clause can refer to. {@code select()} statement can access window properties such as window + * start or end time. + * + * @param alias alias for this window + * @return this window + */ + public SlideWithSizeAndSlideOnTimeWithAlias as(Expression alias) { + return new SlideWithSizeAndSlideOnTimeWithAlias(alias, timeField, size, slide); + } +} diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/SlideWithSizeAndSlideOnTimeWithAlias.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/SlideWithSizeAndSlideOnTimeWithAlias.java new file mode 100644 index 0000000..50b3692 --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/SlideWithSizeAndSlideOnTimeWithAlias.java @@ -0,0 +1,50 @@ +/* + * 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.flink.table.api; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.expressions.Expression; + +/** + * Sliding window on time with alias. Fully specifies a window. + */ +@PublicEvolving +public final class SlideWithSizeAndSlideOnTimeWithAlias extends GroupWindow { + + private final Expression size; + private final Expression slide; + + SlideWithSizeAndSlideOnTimeWithAlias( + Expression alias, + Expression timeField, + Expression size, + Expression slide) { + super(alias, timeField); + this.size = size; + this.slide = slide; + } + + public Expression getSize() { + return size; + } + + public Expression getSlide() { + return slide; + } +} diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Tumble.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Tumble.java new file mode 100644 index 0000000..0b5d6b5 --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Tumble.java @@ -0,0 +1,72 @@ +/* + * 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.flink.table.api; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.expressions.ExpressionParser; + +/** + * Helper class for creating a tumbling window. Tumbling windows are consecutive, non-overlapping + * windows of a specified fixed length. For example, a tumbling window of 5 minutes size groups + * elements in 5 minutes intervals. + * + * <p>Java Example: + * + * <pre> + * {@code + * Tumble.over("10.minutes").on("rowtime").as("w") + * } + * </pre> + * + * <p>Scala Example: + * + * <pre> + * {@code + * Tumble over 5.minutes on 'rowtime as 'w + * } + * </pre> + */ +@PublicEvolving +public final class Tumble { + + /** + * Creates a tumbling window. Tumbling windows are fixed-size, consecutive, non-overlapping + * windows of a specified fixed length. For example, a tumbling window of 5 minutes size groups + * elements in 5 minutes intervals. + * + * @param size the size of the window as time or row-count interval. + * @return a partially defined tumbling window + */ + public static TumbleWithSize over(String size) { + return over(ExpressionParser.parseExpression(size)); + } + + /** + * Creates a tumbling window. Tumbling windows are fixed-size, consecutive, non-overlapping + * windows of a specified fixed length. For example, a tumbling window of 5 minutes size groups + * elements in 5 minutes intervals. + * + * @param size the size of the window as time or row-count interval. + * @return a partially defined tumbling window + */ + public static TumbleWithSize over(Expression size) { + return new TumbleWithSize(size); + } +} diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TumbleWithSize.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TumbleWithSize.java new file mode 100644 index 0000000..600e3a1 --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TumbleWithSize.java @@ -0,0 +1,71 @@ +/* + * 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.flink.table.api; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.expressions.ExpressionParser; + +/** + * Tumbling window. + * + * <p>For streaming tables you can specify grouping by a event-time or processing-time attribute. + * + * <p>For batch tables you can specify grouping on a timestamp or long attribute. + */ +@PublicEvolving +public final class TumbleWithSize { + + /** The size of the window either as time or row-count interval. */ + private Expression size; + + TumbleWithSize(Expression size) { + this.size = size; + } + + /** + * Specifies the time attribute on which rows are grouped. + * + * <p>For streaming tables you can specify grouping by a event-time or processing-time + * attribute. + * + * <p>For batch tables you can specify grouping on a timestamp or long attribute. + * + * @param timeField time attribute for streaming and batch tables + * @return a tumbling window on event-time + */ + public TumbleWithSizeOnTime on(Expression timeField) { + return new TumbleWithSizeOnTime(timeField, size); + } + + /** + * Specifies the time attribute on which rows are grouped. + * + * <p>For streaming tables you can specify grouping by a event-time or processing-time + * attribute. + * + * <p>For batch tables you can specify grouping on a timestamp or long attribute. + * + * @param timeField time attribute for streaming and batch tables + * @return a tumbling window on event-time + */ + public TumbleWithSizeOnTime on(String timeField) { + return on(ExpressionParser.parseExpression(timeField)); + } +} diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TumbleWithSizeOnTime.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TumbleWithSizeOnTime.java new file mode 100644 index 0000000..f635ba5 --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TumbleWithSizeOnTime.java @@ -0,0 +1,62 @@ +/* + * 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.flink.table.api; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.expressions.ExpressionParser; + +/** + * Tumbling window on time. + */ +@PublicEvolving +public final class TumbleWithSizeOnTime { + + private final Expression time; + private final Expression size; + + TumbleWithSizeOnTime(Expression time, Expression size) { + this.time = time; + this.size = size; + } + + /** + * Assigns an alias for this window that the following {@code groupBy()} and {@code select()} + * clause can refer to. {@code select()} statement can access window properties such as window + * start or end time. + * + * @param alias alias for this window + * @return this window + */ + public TumbleWithSizeOnTimeWithAlias as(Expression alias) { + return new TumbleWithSizeOnTimeWithAlias(alias, time, size); + } + + /** + * Assigns an alias for this window that the following {@code groupBy()} and {@code select()} + * clause can refer to. {@code select()} statement can access window properties such as window + * start or end time. + * + * @param alias alias for this window + * @return this window + */ + public TumbleWithSizeOnTimeWithAlias as(String alias) { + return as(ExpressionParser.parseExpression(alias)); + } +} diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TumbleWithSizeOnTimeWithAlias.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TumbleWithSizeOnTimeWithAlias.java new file mode 100644 index 0000000..5180d33 --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TumbleWithSizeOnTimeWithAlias.java @@ -0,0 +1,40 @@ +/* + * 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.flink.table.api; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.expressions.Expression; + +/** + * Tumbling window on time with alias. Fully specifies a window. + */ +@PublicEvolving +public final class TumbleWithSizeOnTimeWithAlias extends GroupWindow { + + private final Expression size; + + TumbleWithSizeOnTimeWithAlias(Expression alias, Expression timeField, Expression size) { + super(alias, timeField); + this.size = size; + } + + public Expression getSize() { + return size; + } +} diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/ExpressionParser.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/ExpressionParser.java new file mode 100644 index 0000000..d576cf5 --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/ExpressionParser.java @@ -0,0 +1,41 @@ +/* + * 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.flink.table.expressions; + +import org.apache.flink.annotation.Internal; + +import java.util.List; + +/** + * Parser for expressions inside a String. This parses exactly the same expressions that + * would be accepted by the Scala Expression DSL. + * + * <p>{@link ExpressionParser} use {@link PlannerExpressionParser} to parse expressions. + */ +@Internal +public final class ExpressionParser { + + public static Expression parseExpression(String exprString) { + return PlannerExpressionParser.create().parseExpression(exprString); + } + + public static List<Expression> parseExpressionList(String expression) { + return PlannerExpressionParser.create().parseExpressionList(expression); + } +} diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/PlannerExpressionParser.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/PlannerExpressionParser.java new file mode 100644 index 0000000..c347025 --- /dev/null +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/PlannerExpressionParser.java @@ -0,0 +1,68 @@ +/* + * 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.flink.table.expressions; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.api.TableException; + +import java.lang.reflect.Constructor; +import java.util.List; + +/** + * Parser for expressions inside a String. This parses exactly the same expressions that + * would be accepted by the Scala Expression DSL. + * + * <p>{@link PlannerExpressionParser} is used by {@link ExpressionParser} to parse expressions. + */ +@Internal +public interface PlannerExpressionParser { + + static PlannerExpressionParser create() { + return SingletonPlannerExpressionParser.getExpressionParser(); + } + + Expression parseExpression(String exprString); + + List<Expression> parseExpressionList(String expression); + + /** + * Util class to create {@link PlannerExpressionParser} instance. Use singleton pattern to avoid + * creating many {@link PlannerExpressionParser}. + */ + class SingletonPlannerExpressionParser { + + private static volatile PlannerExpressionParser expressionParser; + + private SingletonPlannerExpressionParser() {} + + public static PlannerExpressionParser getExpressionParser() { + + if (expressionParser == null) { + try { + Class<?> clazz = Class.forName("org.apache.flink.table.expressions.PlannerExpressionParserImpl"); + Constructor<?> con = clazz.getConstructor(); + expressionParser = (PlannerExpressionParser) con.newInstance(); + } catch (Throwable t) { + throw new TableException("Construction of PlannerExpressionParserImpl class failed.", t); + } + } + return expressionParser; + } + } +} diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/WindowCreationValidationTest.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/WindowCreationValidationTest.java new file mode 100644 index 0000000..e29d813 --- /dev/null +++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/WindowCreationValidationTest.java @@ -0,0 +1,60 @@ +/* + * 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.flink.table.api; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; + +/** + * Test failures for the creation of window. + */ +public class WindowCreationValidationTest { + + @Rule + public final ExpectedException exception = ExpectedException.none(); + + @Test + public void testTumbleOverForString() { + exception.expect(TableException.class); + exception.expectMessage("Construction of PlannerExpressionParserImpl class failed."); + Tumble.over("4.hours"); + } + + @Test + public void testSlideOverForString() { + exception.expect(TableException.class); + exception.expectMessage("Construction of PlannerExpressionParserImpl class failed."); + Slide.over("4.hours"); + } + + @Test + public void testSessionWithGapForString() { + exception.expect(TableException.class); + exception.expectMessage("Construction of PlannerExpressionParserImpl class failed."); + Session.withGap("4.hours"); + } + + @Test + public void testOverWithPartitionByForString() { + exception.expect(TableException.class); + exception.expectMessage("Construction of PlannerExpressionParserImpl class failed."); + Over.partitionBy("a"); + } +} diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/java/BatchTableEnvironment.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/java/BatchTableEnvironment.scala index e1d02b1..8c06a37 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/java/BatchTableEnvironment.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/java/BatchTableEnvironment.scala @@ -24,6 +24,8 @@ import org.apache.flink.table.api._ import org.apache.flink.table.expressions.ExpressionParser import org.apache.flink.table.functions.{AggregateFunction, TableFunction} +import _root_.scala.collection.JavaConverters._ + /** * The [[TableEnvironment]] for a Java batch [[ExecutionEnvironment]] that works * with [[DataSet]]s. @@ -80,7 +82,7 @@ class BatchTableEnvironment @Deprecated() ( */ def fromDataSet[T](dataSet: DataSet[T], fields: String): Table = { val exprs = ExpressionParser - .parseExpressionList(fields) + .parseExpressionList(fields).asScala .toArray val name = createUniqueTableName() @@ -124,7 +126,7 @@ class BatchTableEnvironment @Deprecated() ( */ def registerDataSet[T](name: String, dataSet: DataSet[T], fields: String): Unit = { val exprs = ExpressionParser - .parseExpressionList(fields) + .parseExpressionList(fields).asScala .toArray checkValidTableName(name) diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/java/StreamTableEnvironment.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/java/StreamTableEnvironment.scala index 59f5d2a..f14863f 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/java/StreamTableEnvironment.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/java/StreamTableEnvironment.scala @@ -26,6 +26,7 @@ import org.apache.flink.table.expressions.ExpressionParser import org.apache.flink.streaming.api.datastream.DataStream import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment import _root_.java.lang.{Boolean => JBool} +import _root_.scala.collection.JavaConverters._ /** * The [[TableEnvironment]] for a Java [[StreamExecutionEnvironment]] that works with @@ -84,7 +85,7 @@ class StreamTableEnvironment @Deprecated() ( */ def fromDataStream[T](dataStream: DataStream[T], fields: String): Table = { val exprs = ExpressionParser - .parseExpressionList(fields) + .parseExpressionList(fields).asScala .toArray val name = createUniqueTableName() @@ -129,7 +130,7 @@ class StreamTableEnvironment @Deprecated() ( */ def registerDataStream[T](name: String, dataStream: DataStream[T], fields: String): Unit = { val exprs = ExpressionParser - .parseExpressionList(fields) + .parseExpressionList(fields).asScala .toArray checkValidTableName(name) diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/java/windows.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/java/windows.scala deleted file mode 100644 index 24f39a1..0000000 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/java/windows.scala +++ /dev/null @@ -1,64 +0,0 @@ -/* - * 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.flink.table.api.java - -import org.apache.flink.table.api._ - -/** - * Helper class for creating a tumbling window. Tumbling windows are consecutive, non-overlapping - * windows of a specified fixed length. For example, a tumbling window of 5 minutes size groups - * elements in 5 minutes intervals. - * - * @deprecated This class will be replaced by [[org.apache.flink.table.api.Tumble]]. - */ -@Deprecated -object Tumble extends TumbleBase - -/** - * Helper class for creating a sliding window. Sliding windows have a fixed size and slide by - * a specified slide interval. If the slide interval is smaller than the window size, sliding - * windows are overlapping. Thus, an element can be assigned to multiple windows. - * - * For example, a sliding window of size 15 minutes with 5 minutes sliding interval groups elements - * of 15 minutes and evaluates every five minutes. Each element is contained in three consecutive - * window evaluations. - * - * @deprecated This class will be replaced by [[org.apache.flink.table.api.Slide]]. - */ -@Deprecated -object Slide extends SlideBase - -/** - * Helper class for creating a session window. The boundary of session windows are defined by - * intervals of inactivity, i.e., a session window is closes if no event appears for a defined - * gap period. - * - * @deprecated This class will be replaced by [[org.apache.flink.table.api.Session]]. - */ -@Deprecated -object Session extends SessionBase - -/** - * Helper class for creating an over window. Similar to SQL, over window aggregates compute an - * aggregate for each input row over a range of its neighboring rows. - * - * @deprecated This class will be replaced by [[org.apache.flink.table.api.Over]]. - */ -@Deprecated -object Over extends OverBase diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/scala/expressionDsl.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/scala/expressionDsl.scala index edeaf1d..c252fe5 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/scala/expressionDsl.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/scala/expressionDsl.scala @@ -22,7 +22,7 @@ import java.math.{BigDecimal => JBigDecimal} import java.sql.{Date, Time, Timestamp} import org.apache.flink.api.common.typeinfo.{SqlTimeTypeInfo, TypeInformation} -import org.apache.flink.table.api.{Table, ValidationException} +import org.apache.flink.table.api.{Over, Table, ValidationException} import org.apache.flink.table.expressions.ApiExpressionUtils._ import org.apache.flink.table.expressions.BuiltInFunctionDefinitions.{E => FDE, UUID => FDUUID, _} import org.apache.flink.table.expressions._ diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/scala/windows.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/scala/windows.scala deleted file mode 100644 index 4c2a5dd..0000000 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/scala/windows.scala +++ /dev/null @@ -1,68 +0,0 @@ -/* - * 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.flink.table.api.scala - -import org.apache.flink.table.api._ - -/** - * Helper object for creating a tumbling window. Tumbling windows are consecutive, non-overlapping - * windows of a specified fixed length. For example, a tumbling window of 5 minutes size groups - * elements in 5 minutes intervals. - * - * @deprecated This class will be replaced by [[org.apache.flink.table.api.Tumble]]. - */ -@deprecated( - "This class will be replaced by org.apache.flink.table.api.Tumble.", "1.8") -object Tumble extends TumbleBase - -/** - * Helper object for creating a sliding window. Sliding windows have a fixed size and slide by - * a specified slide interval. If the slide interval is smaller than the window size, sliding - * windows are overlapping. Thus, an element can be assigned to multiple windows. - * - * For example, a sliding window of size 15 minutes with 5 minutes sliding interval groups elements - * of 15 minutes and evaluates every five minutes. Each element is contained in three consecutive - * window evaluations. - * - * @deprecated This class will be replaced by [[org.apache.flink.table.api.Slide]]. - */ -@deprecated( - "This class will be replaced by org.apache.flink.table.api.Slide.", "1.8") -object Slide extends SlideBase - -/** - * Helper object for creating a session window. The boundary of session windows are defined by - * intervals of inactivity, i.e., a session window is closes if no event appears for a defined - * gap period. - * - * @deprecated This class will be replaced by [[org.apache.flink.table.api.Session]]. - */ -@deprecated( - "This class will be replaced by org.apache.flink.table.api.Session.", "1.8") -object Session extends SessionBase - -/** - * Helper class for creating an over window. Similar to SQL, over window aggregates compute an - * aggregate for each input row over a range of its neighboring rows. - * - * @deprecated This class will be replaced by [[org.apache.flink.table.api.Over]]. - */ -@deprecated( - "This class will be replaced by org.apache.flink.table.api.Over.", "1.8") -object Over extends OverBase diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/table.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/table.scala index 2c3738a..e66d14b 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/table.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/table.scala @@ -27,9 +27,11 @@ import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils import org.apache.flink.table.plan.ProjectionTranslator._ import org.apache.flink.table.plan.logical.{Minus, _} import org.apache.flink.table.sinks.TableSink +import org.apache.flink.table.util.JavaScalaConversionUtil import _root_.scala.annotation.varargs import _root_.scala.collection.JavaConverters._ +import _root_.scala.collection.JavaConversions._ /** * A Table is the core component of the Table API. @@ -1089,35 +1091,7 @@ class Table( * If the `groupBy(...)` only references a window alias, the streamed table will be processed * by a single task, i.e., with parallelism 1. * - * @param window window that specifies how elements are grouped. - * @return A windowed table. - * - * @deprecated Will be removed in a future release. Please use Table.window(window: GroupWindow) - * instead. - */ - @deprecated( - "This method will be removed. Please use Table.window(window: GroupWindow) instead.", - "1.8") - @Deprecated - def window(window: Window): WindowedTable = { - new WindowedTable(this, window) - } - - /** - * Groups the records of a table by assigning them to windows defined by a time or row interval. - * - * For streaming tables of infinite size, grouping into windows is required to define finite - * groups on which group-based aggregates can be computed. - * - * For batch tables of finite size, windowing essentially provides shortcuts for time-based - * groupBy. - * - * __Note__: Computing windowed aggregates on a streaming table is only a parallel operation - * if additional grouping attributes are added to the `groupBy(...)` clause. - * If the `groupBy(...)` only references a window alias, the streamed table will be processed - * by a single task, i.e., with parallelism 1. - * - * @param window group window that specifies how elements are grouped. + * @param window groupWindow that specifies how elements are grouped. * @return A group windowed table. */ def window(window: GroupWindow): GroupWindowedTable = { @@ -1231,16 +1205,11 @@ class GroupedTable( } /** - * A table that has been windowed for grouping [[Window]]s. - * - * @deprecated Will be replaced by [[GroupWindowedTable]]. + * A table that has been windowed for [[GroupWindow]]s. */ -@Deprecated -@deprecated( - "This class will be replaced by GroupWindowedTable.", "1.8") -class WindowedTable( +class GroupWindowedTable( private[flink] val table: Table, - private[flink] val window: Window) { + private[flink] val window: GroupWindow) { /** * Groups the elements by a mandatory window and one or more optional grouping attributes. @@ -1289,27 +1258,12 @@ class WindowedTable( } /** - * A table that has been windowed for [[GroupWindow]]s. - */ -class GroupWindowedTable( - override private[flink] val table: Table, - override private[flink] val window: GroupWindow) - extends WindowedTable(table, window) - -/** * A table that has been windowed and grouped for [[GroupWindow]]s. - * - * @deprecated The constructor contains [[Window]] parameter will be removed. Use constructor - * with [[GroupWindow]] instead. */ -class WindowGroupedTable @Deprecated() ( +class WindowGroupedTable( private[flink] val table: Table, private[flink] val groupKeys: Seq[Expression], - private[flink] val window: Window) { - - def this(table: Table, groupKeys: Seq[Expression], window: GroupWindow) { - this(table, groupKeys, window.asInstanceOf[Window]) - } + private[flink] val window: GroupWindow) { /** * Performs a selection operation on a window grouped table. Similar to an SQL SELECT statement. @@ -1466,7 +1420,7 @@ class OverWindowedTable( overWindow.getPartitioning.map(table.expressionBridge.bridge), table.expressionBridge.bridge(overWindow.getOrder), table.expressionBridge.bridge(overWindow.getPreceding), - overWindow.getFollowing.map(table.expressionBridge.bridge) + JavaScalaConversionUtil.toScala(overWindow.getFollowing).map(table.expressionBridge.bridge) ) } } diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/windows.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/windows.scala deleted file mode 100644 index 948a620..0000000 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/windows.scala +++ /dev/null @@ -1,741 +0,0 @@ -/* - * 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.flink.table.api - -import org.apache.flink.table.expressions._ - -/** - * An over window specification. - * - * Similar to SQL, over window aggregates compute an aggregate for each input row over a range - * of its neighboring rows. - */ -class OverWindow( - alias: Expression, - partitionBy: Seq[Expression], - orderBy: Expression, - preceding: Expression, - following: Option[Expression]) { - - def getAlias: Expression = alias - - def getPartitioning: Seq[Expression] = partitionBy - - def getOrder: Expression = orderBy - - def getPreceding: Expression = preceding - - def getFollowing: Option[Expression] = following -} - -// ------------------------------------------------------------------------------------------------ -// Over windows -// ------------------------------------------------------------------------------------------------ - -/** - * Partially defined over window with partitioning. - */ -class OverWindowPartitioned(partitionBy: Seq[Expression]) { - - /** - * Specifies the time attribute on which rows are ordered. - * - * For streaming tables, reference a rowtime or proctime time attribute here - * to specify the time mode. - * - * For batch tables, refer to a timestamp or long attribute. - * - * @param orderBy field reference - * @return an over window with defined order - */ - def orderBy(orderBy: String): OverWindowPartitionedOrdered = { - this.orderBy(ExpressionParser.parseExpression(orderBy)) - } - - /** - * Specifies the time attribute on which rows are ordered. - * - * For streaming tables, reference a rowtime or proctime time attribute here - * to specify the time mode. - * - * For batch tables, refer to a timestamp or long attribute. - * - * @param orderBy field reference - * @return an over window with defined order - */ - def orderBy(orderBy: Expression): OverWindowPartitionedOrdered = { - new OverWindowPartitionedOrdered(partitionBy, orderBy) - } -} - -/** - * Partially defined over window with (optional) partitioning and order. - */ -class OverWindowPartitionedOrdered(partitionBy: Seq[Expression], orderBy: Expression) { - - /** - * Set the preceding offset (based on time or row-count intervals) for over window. - * - * @param preceding preceding offset relative to the current row. - * @return an over window with defined preceding - */ - def preceding(preceding: String): OverWindowPartitionedOrderedPreceding = { - this.preceding(ExpressionParser.parseExpression(preceding)) - } - - /** - * Set the preceding offset (based on time or row-count intervals) for over window. - * - * @param preceding preceding offset relative to the current row. - * @return an over window with defined preceding - */ - def preceding(preceding: Expression): OverWindowPartitionedOrderedPreceding = { - new OverWindowPartitionedOrderedPreceding(partitionBy, orderBy, preceding) - } - - /** - * Assigns an alias for this window that the following `select()` clause can refer to. - * - * @param alias alias for this over window - * @return the fully defined over window - */ - def as(alias: String): OverWindow = as(ExpressionParser.parseExpression(alias)) - - /** - * Assigns an alias for this window that the following `select()` clause can refer to. - * - * @param alias alias for this over window - * @return the fully defined over window - */ - def as(alias: Expression): OverWindow = { - new OverWindow(alias, partitionBy, orderBy, UnboundedRange(), None) - } -} - -/** - * Partially defined over window with (optional) partitioning, order, and preceding. - */ -class OverWindowPartitionedOrderedPreceding( - private val partitionBy: Seq[Expression], - private val orderBy: Expression, - private val preceding: Expression) { - - private var optionalFollowing: Option[Expression] = None - - /** - * Assigns an alias for this window that the following `select()` clause can refer to. - * - * @param alias alias for this over window - * @return the fully defined over window - */ - def as(alias: String): OverWindow = as(ExpressionParser.parseExpression(alias)) - - /** - * Assigns an alias for this window that the following `select()` clause can refer to. - * - * @param alias alias for this over window - * @return the fully defined over window - */ - def as(alias: Expression): OverWindow = { - new OverWindow(alias, partitionBy, orderBy, preceding, optionalFollowing) - } - - /** - * Set the following offset (based on time or row-count intervals) for over window. - * - * @param following following offset that relative to the current row. - * @return an over window with defined following - */ - def following(following: String): OverWindowPartitionedOrderedPreceding = { - this.following(ExpressionParser.parseExpression(following)) - } - - /** - * Set the following offset (based on time or row-count intervals) for over window. - * - * @param following following offset that relative to the current row. - * @return an over window with defined following - */ - def following(following: Expression): OverWindowPartitionedOrderedPreceding = { - optionalFollowing = Some(following) - this - } -} - -// ------------------------------------------------------------------------------------------------ -// Group windows -// ------------------------------------------------------------------------------------------------ - -/** - * A group window specification. - * - * Group windows group rows based on time or row-count intervals and is therefore essentially a - * special type of groupBy. Just like groupBy, group windows allow to compute aggregates - * on groups of elements. - * - * Infinite streaming tables can only be grouped into time or row intervals. Hence window grouping - * is required to apply aggregations on streaming tables. - * - * For finite batch tables, group windows provide shortcuts for time-based groupBy. - * - * @deprecated Will be replaced by [[GroupWindow]] - */ -@Deprecated -@deprecated( - "This class will be replaced by GroupWindow.", "1.8") -abstract class Window(alias: Expression, timeField: Expression) { - - def getAlias: Expression = { - alias - } - - def getTimeField: Expression = { - timeField - } -} - -/** - * A group window specification. - * - * Group windows group rows based on time or row-count intervals and is therefore essentially a - * special type of groupBy. Just like groupBy, group windows allow to compute aggregates - * on groups of elements. - * - * Infinite streaming tables can only be grouped into time or row intervals. Hence window grouping - * is required to apply aggregations on streaming tables. - * - * For finite batch tables, group windows provide shortcuts for time-based groupBy. - */ -abstract class GroupWindow(alias: Expression, timeField: Expression) - extends Window(alias, timeField) - -// ------------------------------------------------------------------------------------------------ -// Tumbling windows -// ------------------------------------------------------------------------------------------------ - -/** - * Tumbling window. - * - * For streaming tables you can specify grouping by a event-time or processing-time attribute. - * - * For batch tables you can specify grouping on a timestamp or long attribute. - * - * @param size the size of the window either as time or row-count interval. - */ -class TumbleWithSize(size: Expression) { - - /** - * Tumbling window. - * - * For streaming tables you can specify grouping by a event-time or processing-time attribute. - * - * For batch tables you can specify grouping on a timestamp or long attribute. - * - * @param size the size of the window either as time or row-count interval. - */ - def this(size: String) = this(ExpressionParser.parseExpression(size)) - - /** - * Specifies the time attribute on which rows are grouped. - * - * For streaming tables you can specify grouping by a event-time or processing-time attribute. - * - * For batch tables you can specify grouping on a timestamp or long attribute. - * - * @param timeField time attribute for streaming and batch tables - * @return a tumbling window on event-time - */ - def on(timeField: Expression): TumbleWithSizeOnTime = - new TumbleWithSizeOnTime(timeField, size) - - /** - * Specifies the time attribute on which rows are grouped. - * - * For streaming tables you can specify grouping by a event-time or processing-time attribute. - * - * For batch tables you can specify grouping on a timestamp or long attribute. - * - * @param timeField time attribute for streaming and batch tables - * @return a tumbling window on event-time - */ - def on(timeField: String): TumbleWithSizeOnTime = - on(ExpressionParser.parseExpression(timeField)) -} - -/** - * Tumbling window on time. - */ -class TumbleWithSizeOnTime(time: Expression, size: Expression) { - - /** - * Assigns an alias for this window that the following `groupBy()` and `select()` clause can - * refer to. `select()` statement can access window properties such as window start or end time. - * - * @param alias alias for this window - * @return this window - */ - def as(alias: Expression): TumbleWithSizeOnTimeWithAlias = { - new TumbleWithSizeOnTimeWithAlias(alias, time, size) - } - - /** - * Assigns an alias for this window that the following `groupBy()` and `select()` clause can - * refer to. `select()` statement can access window properties such as window start or end time. - * - * @param alias alias for this window - * @return this window - */ - def as(alias: String): TumbleWithSizeOnTimeWithAlias = { - as(ExpressionParser.parseExpression(alias)) - } -} - -/** - * Tumbling window on time with alias. Fully specifies a window. - */ -class TumbleWithSizeOnTimeWithAlias( - alias: Expression, - timeField: Expression, - size: Expression) - extends GroupWindow( - alias, - timeField) { - - def getSize: Expression = { - size - } -} - -// ------------------------------------------------------------------------------------------------ -// Sliding windows -// ------------------------------------------------------------------------------------------------ - -/** - * Partially specified sliding window. - * - * @param size the size of the window either as time or row-count interval. - */ -class SlideWithSize(size: Expression) { - - /** - * Partially specified sliding window. - * - * @param size the size of the window either as time or row-count interval. - */ - def this(size: String) = this(ExpressionParser.parseExpression(size)) - - /** - * Specifies the window's slide as time or row-count interval. - * - * The slide determines the interval in which windows are started. Hence, sliding windows can - * overlap if the slide is smaller than the size of the window. - * - * For example, you could have windows of size 15 minutes that slide by 3 minutes. With this - * 15 minutes worth of elements are grouped every 3 minutes and each row contributes to 5 - * windows. - * - * @param slide the slide of the window either as time or row-count interval. - * @return a sliding window - */ - def every(slide: Expression): SlideWithSizeAndSlide = new SlideWithSizeAndSlide(size, slide) - - /** - * Specifies the window's slide as time or row-count interval. - * - * The slide determines the interval in which windows are started. Hence, sliding windows can - * overlap if the slide is smaller than the size of the window. - * - * For example, you could have windows of size 15 minutes that slide by 3 minutes. With this - * 15 minutes worth of elements are grouped every 3 minutes and each row contributes to 5 - * windows. - * - * @param slide the slide of the window either as time or row-count interval. - * @return a sliding window - */ - def every(slide: String): SlideWithSizeAndSlide = every(ExpressionParser.parseExpression(slide)) -} - -/** - * Sliding window. - * - * For streaming tables you can specify grouping by a event-time or processing-time attribute. - * - * For batch tables you can specify grouping on a timestamp or long attribute. - * - * @param size the size of the window either as time or row-count interval. - */ -class SlideWithSizeAndSlide(size: Expression, slide: Expression) { - - /** - * Specifies the time attribute on which rows are grouped. - * - * For streaming tables you can specify grouping by a event-time or processing-time attribute. - * - * For batch tables you can specify grouping on a timestamp or long attribute. - * - * @param timeField time attribute for streaming and batch tables - * @return a tumbling window on event-time - */ - def on(timeField: Expression): SlideWithSizeAndSlideOnTime = - new SlideWithSizeAndSlideOnTime(timeField, size, slide) - - /** - * Specifies the time attribute on which rows are grouped. - * - * For streaming tables you can specify grouping by a event-time or processing-time attribute. - * - * For batch tables you can specify grouping on a timestamp or long attribute. - * - * @param timeField time attribute for streaming and batch tables - * @return a tumbling window on event-time - */ - def on(timeField: String): SlideWithSizeAndSlideOnTime = - on(ExpressionParser.parseExpression(timeField)) -} - -/** - * Sliding window on time. - */ -class SlideWithSizeAndSlideOnTime(timeField: Expression, size: Expression, slide: Expression) { - - /** - * Assigns an alias for this window that the following `groupBy()` and `select()` clause can - * refer to. `select()` statement can access window properties such as window start or end time. - * - * @param alias alias for this window - * @return this window - */ - def as(alias: Expression): SlideWithSizeAndSlideOnTimeWithAlias = { - new SlideWithSizeAndSlideOnTimeWithAlias(alias, timeField, size, slide) - } - - /** - * Assigns an alias for this window that the following `groupBy()` and `select()` clause can - * refer to. `select()` statement can access window properties such as window start or end time. - * - * @param alias alias for this window - * @return this window - */ - def as(alias: String): SlideWithSizeAndSlideOnTimeWithAlias = { - as(ExpressionParser.parseExpression(alias)) - } -} - -/** - * Sliding window on time with alias. Fully specifies a window. - */ -class SlideWithSizeAndSlideOnTimeWithAlias( - alias: Expression, - timeField: Expression, - size: Expression, - slide: Expression) - extends GroupWindow( - alias, - timeField) { - - def getSize: Expression = { - size - } - - def getSlide: Expression = { - slide - } -} - -// ------------------------------------------------------------------------------------------------ -// Session windows -// ------------------------------------------------------------------------------------------------ - -/** - * Session window. - * - * For streaming tables you can specify grouping by a event-time or processing-time attribute. - * - * For batch tables you can specify grouping on a timestamp or long attribute. - * - * @param gap the time interval of inactivity before a window is closed. - */ -class SessionWithGap(gap: Expression) { - - /** - * Session window. - * - * For streaming tables you can specify grouping by a event-time or processing-time attribute. - * - * For batch tables you can specify grouping on a timestamp or long attribute. - * - * @param gap the time interval of inactivity before a window is closed. - */ - def this(gap: String) = this(ExpressionParser.parseExpression(gap)) - - /** - * Specifies the time attribute on which rows are grouped. - * - * For streaming tables you can specify grouping by a event-time or processing-time attribute. - * - * For batch tables you can specify grouping on a timestamp or long attribute. - * - * @param timeField time attribute for streaming and batch tables - * @return a tumbling window on event-time - */ - def on(timeField: Expression): SessionWithGapOnTime = - new SessionWithGapOnTime(timeField, gap) - - /** - * Specifies the time attribute on which rows are grouped. - * - * For streaming tables you can specify grouping by a event-time or processing-time attribute. - * - * For batch tables you can specify grouping on a timestamp or long attribute. - * - * @param timeField time attribute for streaming and batch tables - * @return a tumbling window on event-time - */ - def on(timeField: String): SessionWithGapOnTime = - on(ExpressionParser.parseExpression(timeField)) -} - -/** - * Session window on time. - */ -class SessionWithGapOnTime(timeField: Expression, gap: Expression) { - - /** - * Assigns an alias for this window that the following `groupBy()` and `select()` clause can - * refer to. `select()` statement can access window properties such as window start or end time. - * - * @param alias alias for this window - * @return this window - */ - def as(alias: Expression): SessionWithGapOnTimeWithAlias = { - new SessionWithGapOnTimeWithAlias(alias, timeField, gap) - } - - /** - * Assigns an alias for this window that the following `groupBy()` and `select()` clause can - * refer to. `select()` statement can access window properties such as window start or end time. - * - * @param alias alias for this window - * @return this window - */ - def as(alias: String): SessionWithGapOnTimeWithAlias = { - as(ExpressionParser.parseExpression(alias)) - } -} - -/** - * Session window on time with alias. Fully specifies a window. - */ -class SessionWithGapOnTimeWithAlias( - alias: Expression, - timeField: Expression, - gap: Expression) - extends GroupWindow( - alias, - timeField) { - - def getGap: Expression = { - gap - } -} - -/** - * Base class for Tumble Window Helper classes. This class contains help methods to create Tumble - * Windows. - */ -class TumbleBase { - - /** - * Creates a tumbling window. Tumbling windows are consecutive, non-overlapping - * windows of a specified fixed length. For example, a tumbling window of 5 minutes size groups - * elements in 5 minutes intervals. - * - * @param size the size of the window as time or row-count interval. - * @return a partially defined tumbling window - */ - def over(size: String): TumbleWithSize = new TumbleWithSize(size) - - /** - * Creates a tumbling window. Tumbling windows are fixed-size, consecutive, non-overlapping - * windows. For example, a tumbling window of 5 minutes size groups - * elements in 5 minutes intervals. - * - * @param size the size of the window as time or row-count interval. - * @return a partially defined tumbling window - */ - def over(size: Expression): TumbleWithSize = new TumbleWithSize(size) -} - -/** - * Base class for Slide Window Helper classes. This class contains help methods to create Slide - * Windows. - */ -class SlideBase { - - /** - * Creates a sliding window. Sliding windows have a fixed size and slide by - * a specified slide interval. If the slide interval is smaller than the window size, sliding - * windows are overlapping. Thus, an element can be assigned to multiple windows. - * - * For example, a sliding window of size 15 minutes with 5 minutes sliding interval groups - * elements of 15 minutes and evaluates every five minutes. Each element is contained in three - * consecutive window evaluations. - * - * @param size the size of the window as time or row-count interval - * @return a partially specified sliding window - */ - def over(size: String): SlideWithSize = new SlideWithSize(size) - - /** - * Creates a sliding window. Sliding windows have a fixed size and slide by - * a specified slide interval. If the slide interval is smaller than the window size, sliding - * windows are overlapping. Thus, an element can be assigned to multiple windows. - * - * For example, a sliding window of size 15 minutes with 5 minutes sliding interval groups - * elements of 15 minutes and evaluates every five minutes. Each element is contained in three - * consecutive - * - * @param size the size of the window as time or row-count interval - * @return a partially specified sliding window - */ - def over(size: Expression): SlideWithSize = new SlideWithSize(size) -} - -/** - * Base class for Session Window Helper classes. This class contains help methods to create Session - * Windows. - */ -class SessionBase { - - /** - * Creates a session window. The boundary of session windows are defined by - * intervals of inactivity, i.e., a session window is closes if no event appears for a defined - * gap period. - * - * @param gap specifies how long (as interval of milliseconds) to wait for new data before - * closing the session window. - * @return a partially defined session window - */ - def withGap(gap: String): SessionWithGap = new SessionWithGap(gap) - - /** - * Creates a session window. The boundary of session windows are defined by - * intervals of inactivity, i.e., a session window is closes if no event appears for a defined - * gap period. - * - * @param gap specifies how long (as interval of milliseconds) to wait for new data before - * closing the session window. - * @return a partially defined session window - */ - def withGap(gap: Expression): SessionWithGap = new SessionWithGap(gap) -} - -/** - * Base class for Over Window Helper classes. This class contains help methods to create Over - * Windows. - */ -class OverBase { - - /** - * Specifies the time attribute on which rows are ordered. - * - * For streaming tables, reference a rowtime or proctime time attribute here - * to specify the time mode. - * - * For batch tables, refer to a timestamp or long attribute. - * - * @param orderBy field reference - * @return an over window with defined order - */ - def orderBy(orderBy: String): OverWindowPartitionedOrdered = { - new OverWindowPartitionedOrdered(Seq(), ExpressionParser.parseExpression(orderBy)) - } - - /** - * Specifies the time attribute on which rows are ordered. - * - * For streaming tables, reference a rowtime or proctime time attribute here - * to specify the time mode. - * - * For batch tables, refer to a timestamp or long attribute. - * - * @param orderBy field reference - * @return an over window with defined order - */ - def orderBy(orderBy: Expression): OverWindowPartitionedOrdered = { - new OverWindowPartitionedOrdered(Seq(), orderBy) - } - - /** - * Partitions the elements on some partition keys. - * - * Each partition is individually sorted and aggregate functions are applied to each - * partition separately. - * - * @param partitionBy list of field references - * @return an over window with defined partitioning - */ - def partitionBy(partitionBy: String): OverWindowPartitioned = { - new OverWindowPartitioned(ExpressionParser.parseExpressionList(partitionBy)) - } - - /** - * Partitions the elements on some partition keys. - * - * Each partition is individually sorted and aggregate functions are applied to each - * partition separately. - * - * @param partitionBy list of field references - * @return an over window with defined partitioning - */ - def partitionBy(partitionBy: Expression*): OverWindowPartitioned = { - new OverWindowPartitioned(partitionBy) - } -} - -/** - * Helper class for creating a tumbling window. Tumbling windows are consecutive, non-overlapping - * windows of a specified fixed length. For example, a tumbling window of 5 minutes size groups - * elements in 5 minutes intervals. - */ -object Tumble extends TumbleBase - -/** - * Helper class for creating a sliding window. Sliding windows have a fixed size and slide by - * a specified slide interval. If the slide interval is smaller than the window size, sliding - * windows are overlapping. Thus, an element can be assigned to multiple windows. - * - * For example, a sliding window of size 15 minutes with 5 minutes sliding interval groups elements - * of 15 minutes and evaluates every five minutes. Each element is contained in three consecutive - * window evaluations. - */ -object Slide extends SlideBase - -/** - * Helper class for creating a session window. The boundary of session windows are defined by - * intervals of inactivity, i.e., a session window is closes if no event appears for a defined - * gap period. - */ -object Session extends SessionBase - -/** - * Helper class for creating an over window. Similar to SQL, over window aggregates compute an - * aggregate for each input row over a range of its neighboring rows. - */ -object Over extends OverBase diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/ExpressionParser.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/PlannerExpressionParserImpl.scala similarity index 96% rename from flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/ExpressionParser.scala rename to flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/PlannerExpressionParserImpl.scala index 72fba67..044e0d2 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/ExpressionParser.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/PlannerExpressionParserImpl.scala @@ -21,8 +21,24 @@ import org.apache.flink.api.common.typeinfo.{SqlTimeTypeInfo, TypeInformation} import org.apache.flink.table.api._ import org.apache.flink.table.expressions.ApiExpressionUtils._ +import _root_.java.util.{List => JList} import _root_.scala.language.implicitConversions import _root_.scala.util.parsing.combinator.{JavaTokenParsers, PackratParsers} +import _root_.scala.collection.JavaConversions._ + +/** + * The implementation of a [[PlannerExpressionParser]] which parsers expressions inside a String. + */ +class PlannerExpressionParserImpl extends PlannerExpressionParser { + + def parseExpression(exprString: String): Expression = { + PlannerExpressionParserImpl.parseExpression(exprString) + } + + override def parseExpressionList(expression: String): JList[Expression] = { + PlannerExpressionParserImpl.parseExpressionList(expression) + } +} /** * Parser for expressions inside a String. This parses exactly the same expressions that @@ -33,7 +49,10 @@ import _root_.scala.util.parsing.combinator.{JavaTokenParsers, PackratParsers} * available in the Scala Expression DSL. This parser must be kept in sync with the Scala DSL * lazy valined in the above files. */ -object ExpressionParser extends JavaTokenParsers with PackratParsers { +object PlannerExpressionParserImpl extends JavaTokenParsers + with PackratParsers + with PlannerExpressionParser { + case class Keyword(key: String) // Convert the keyword into an case insensitive Parser @@ -113,7 +132,7 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers { lazy val TRIM_MODE_TRAILING: Keyword = Keyword("TRAILING") lazy val TRIM_MODE_BOTH: Keyword = Keyword("BOTH") - def functionIdent: ExpressionParser.Parser[String] = super.ident + def functionIdent: PlannerExpressionParserImpl.Parser[String] = super.ident // symbols @@ -625,7 +644,7 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers { lazy val expressionList: Parser[List[Expression]] = rep1sep(expression, ",") - def parseExpressionList(expression: String): List[Expression] = { + def parseExpressionList(expression: String): JList[Expression] = { parseAll(expressionList, expression) match { case Success(lst, _) => lst diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/table/stringexpr/GroupWindowStringExpressionTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/table/stringexpr/GroupWindowStringExpressionTest.scala index 2e05139..b1e5860 100644 --- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/table/stringexpr/GroupWindowStringExpressionTest.scala +++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/table/stringexpr/GroupWindowStringExpressionTest.scala @@ -19,8 +19,8 @@ package org.apache.flink.table.api.stream.table.stringexpr import org.apache.flink.api.scala._ -import org.apache.flink.table.api.{Session, Slide, Tumble} import org.apache.flink.table.api.scala._ +import org.apache.flink.table.api.{Session, Slide, Tumble} import org.apache.flink.table.runtime.utils.JavaUserDefinedAggFunctions.WeightedAvg import org.apache.flink.table.functions.aggfunctions.CountAggFunction import org.apache.flink.table.utils.TableTestBase diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/runtime/stream/TimeAttributesITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/runtime/stream/TimeAttributesITCase.scala index a81b3ed..01a55f1 100644 --- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/runtime/stream/TimeAttributesITCase.scala +++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/runtime/stream/TimeAttributesITCase.scala @@ -43,6 +43,8 @@ import org.junit.Assert._ import org.junit.Test import scala.collection.mutable +import scala.collection.JavaConverters._ + /** * Tests for access and materialization of time attributes. @@ -560,11 +562,11 @@ class TimeAttributesITCase extends AbstractTestBase { // use aliases, swap all attributes, and skip b2 val table4 = stream.toTable( tEnv, - ExpressionParser.parseExpressionList("b.rowtime as b, c as c, a as a"): _*) + ExpressionParser.parseExpressionList("b.rowtime as b, c as c, a as a").asScala: _*) // no aliases, no swapping val table5 = stream.toTable( tEnv, - ExpressionParser.parseExpressionList("a, b.rowtime, c"): _*) + ExpressionParser.parseExpressionList("a, b.rowtime, c").asScala: _*) val t = table.select('b, 'c , 'a) .unionAll(table2.select('b, 'c, 'a))