[
https://issues.apache.org/jira/browse/FLINK-6073?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16627294#comment-16627294
]
ASF GitHub Bot commented on FLINK-6073:
---------------------------------------
fhueske closed pull request #3609: [FLINK-6073] [table] Support for SQL inner
queries for proctime
URL: https://github.com/apache/flink/pull/3609
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamJoin.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamJoin.scala
new file mode 100644
index 00000000000..c623c3ccd79
--- /dev/null
+++
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamJoin.scala
@@ -0,0 +1,241 @@
+/*
+ * 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.plan.nodes.datastream
+
+import org.apache.calcite.plan.{ RelOptCluster, RelTraitSet }
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.core.AggregateCall
+import org.apache.calcite.rel.{ RelNode, RelWriter, BiRel }
+import org.apache.flink.api.java.tuple.Tuple
+import org.apache.flink.streaming.api.datastream.{ AllWindowedStream,
DataStream, KeyedStream, WindowedStream }
+import org.apache.flink.streaming.api.windowing.assigners._
+import org.apache.flink.streaming.api.windowing.time.Time
+import org.apache.flink.streaming.api.windowing.windows.{ Window =>
DataStreamWindow }
+import org.apache.flink.table.api.StreamTableEnvironment
+import org.apache.flink.table.calcite.FlinkRelBuilder.NamedWindowProperty
+import org.apache.flink.table.calcite.FlinkTypeFactory
+import org.apache.flink.table.expressions._
+import org.apache.flink.table.plan.logical._
+import org.apache.flink.table.plan.nodes.CommonAggregate
+import org.apache.flink.table.plan.nodes.datastream.DataStreamAggregate._
+import org.apache.flink.table.runtime.aggregate.AggregateUtil._
+import org.apache.flink.table.runtime.aggregate._
+import org.apache.flink.table.typeutils.TypeCheckUtils.isTimeInterval
+import org.apache.flink.table.typeutils.{ RowIntervalTypeInfo,
TimeIntervalTypeInfo }
+import org.apache.flink.types.Row
+import org.apache.calcite.rel.logical.LogicalJoin
+import org.apache.calcite.rel.core.JoinRelType
+import org.apache.flink.table.api.TableException
+import org.apache.flink.api.java.functions.KeySelector
+import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.streaming.api.windowing.triggers.Trigger
+import org.apache.flink.streaming.api.windowing.windows.GlobalWindow
+import org.apache.flink.streaming.api.windowing.triggers.Trigger.TriggerContext
+import org.apache.flink.streaming.api.windowing.triggers.TriggerResult
+import org.apache.flink.streaming.api.datastream.CoGroupedStreams.TaggedUnion
+import org.apache.flink.streaming.api.windowing.evictors.Evictor
+import org.apache.flink.streaming.api.windowing.evictors.Evictor.EvictorContext
+import java.lang.Iterable
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord
+import org.apache.flink.streaming.runtime.operators.windowing.TimestampedValue
+import org.apache.flink.api.common.functions.RichFlatJoinFunction
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.api.common.state.ValueState
+import org.apache.flink.api.common.state.ValueStateDescriptor
+import org.apache.flink.util.Collector
+
+class DataStreamJoin(
+ calc: LogicalJoin,
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ inputLeft: RelNode,
+ inputRight: RelNode,
+ rowType: RelDataType,
+ description: String)
+ extends BiRel(cluster, traitSet, inputLeft, inputRight) with DataStreamRel
{
+
+ override def deriveRowType(): RelDataType = rowType
+
+ override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]):
RelNode = {
+ new DataStreamJoin(
+ calc,
+ cluster,
+ traitSet,
+ inputs.get(0),
+ inputs.get(1),
+ rowType,
+ description + calc.getId())
+ }
+
+ override def toString: String = {
+ s"Join(${
+ if (!calc.getCondition.isAlwaysTrue()) {
+ s"condition: (${calc.getCondition}), "
+ } else {
+ ""
+ }
+ }left: ($inputLeft), right($inputRight))"
+ }
+
+ override def explainTerms(pw: RelWriter): RelWriter = {
+ super.explainTerms(pw)
+ .itemIf("condition", calc.getCondition,
!calc.getCondition.isAlwaysTrue())
+ .item("join", calc)
+ .item("left", inputLeft)
+ .item("right", inputRight)
+ }
+
+ override def translateToPlan(tableEnv: StreamTableEnvironment):
DataStream[Row] = {
+
+ val inputDSLeft =
inputLeft.asInstanceOf[DataStreamRel].translateToPlan(tableEnv)
+ val inputDSRight =
inputRight.asInstanceOf[DataStreamRel].translateToPlan(tableEnv)
+
+ //define the setup for various types of joins to be supported
+ (calc.getCondition.isAlwaysTrue(), calc.getJoinType) match {
+ case (true, JoinRelType.LEFT) =>
+ createInnerQueryJoin(inputDSLeft, inputDSRight)
+ case (_, _) =>
+ throw new TableException("Table does not support this type of JOIN.")
+ }
+
+ null
+ }
+
+ def createInnerQueryJoin(
+ inputDSLeft: DataStream[Row], inputDSRight: DataStream[Row]):
DataStream[Row] = {
+
+ // get the output types
+ val rowTypeInfo =
FlinkTypeFactory.toInternalRowTypeInfo(getRowType).asInstanceOf[RowTypeInfo]
+
+ val result = inputDSLeft.join(inputDSRight)
+ .where(new EmptyKeySelector()).equalTo(new EmptyKeySelector())
+ .window(GlobalWindows.create())
+ .trigger(new ProcTimeLeftJoinTrigger())
+ .evictor(new FullEvictor())
+ .apply(new JoinProcTimeForInnerQuerry(rowTypeInfo))
+
+ null
+ }
+
+}
+
+class EmptyKeySelector extends KeySelector[Row, Integer] {
+ override def getKey(value: Row): Integer = {
+ 0
+ }
+}
+
+class ProcTimeLeftJoinTrigger extends Trigger[Object, GlobalWindow] with
Serializable {
+
+ /*
+ * Check if element comes from the left stream case in which we should fire
+ */
+ override def onElement(element: Object,
+ timestamp: Long,
+ window: GlobalWindow, ctx: TriggerContext): TriggerResult = {
+ element match {
+ case elementPair: TaggedUnion[Row, Row] => {
+ if (elementPair.isOne()) {
+ TriggerResult.FIRE_AND_PURGE
+ } else {
+ TriggerResult.CONTINUE
+ }
+ }
+ case _ => TriggerResult.CONTINUE
+ }
+ }
+
+ /*
+ * We operate on processing time so we move on each element
+ */
+ override def onProcessingTime(timestamp: Long, window:
+ GlobalWindow, ctx: TriggerContext): TriggerResult = {
+ TriggerResult.CONTINUE
+ }
+
+ /*
+ * We operate on processing time so we move on each element
+ */
+ override def onEventTime(timestamp: Long,
+ window: GlobalWindow,
+ ctx: TriggerContext): TriggerResult = {
+ TriggerResult.CONTINUE
+ }
+
+ override def clear(window: GlobalWindow, ctx: TriggerContext): Unit = {
+
+ }
+}
+
+class FullEvictor extends Evictor[Object, GlobalWindow] {
+ override def evictBefore(x1: Iterable[TimestampedValue[Object]],
+ size: Int, window: GlobalWindow, ctx: EvictorContext): Unit = {
+ }
+
+ override def evictAfter(x1: Iterable[TimestampedValue[Object]],
+ size: Int, window: GlobalWindow, ctx: EvictorContext): Unit = {
+ val iter = x1.iterator()
+ while (iter.hasNext()) {
+ iter.remove()
+ }
+ }
+}
+
+class JoinProcTimeForInnerQuerry(
+ private val rowTypeInfo: RowTypeInfo) extends RichFlatJoinFunction[Row,
Row, Row] {
+
+ private var lastValueRight: ValueState[Row] = _
+
+ override def open(configuration: Configuration): Unit = {
+ val stateDescriptor: ValueStateDescriptor[Row] =
+ new ValueStateDescriptor[Row]("overState", rowTypeInfo)
+ lastValueRight = getRuntimeContext.getState(stateDescriptor)
+ }
+ override def join(first: Row, second: Row, out: Collector[Row]): Unit = {
+
+ var secondarity = 0
+ var secondR = second
+ if (second != null) {
+ lastValueRight.update(second)
+ secondarity = second.getArity
+ } else {
+ secondR = lastValueRight.value()
+ if (secondR != null) {
+ secondarity = secondR.getArity
+ }
+ }
+
+ if (first != null) {
+ val outrez = new Row(first.getArity + secondarity)
+ var i = 0
+ while (i < first.getArity) {
+ outrez.setField(i, first.getField(i))
+ i += 1
+ }
+ i = 0
+ while (i < secondarity) {
+ outrez.setField(i, secondR.getField(i))
+ i += 1
+ }
+ out.collect(outrez)
+ }
+ }
+}
+
+
diff --git
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamNonWindowAggregate.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamNonWindowAggregate.scala
new file mode 100644
index 00000000000..f020438a259
--- /dev/null
+++
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamNonWindowAggregate.scala
@@ -0,0 +1,129 @@
+/*
+ * 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.plan.nodes.datastream
+
+import org.apache.calcite.plan.{ RelOptCluster, RelTraitSet }
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.core.AggregateCall
+import org.apache.calcite.rel.{ RelNode, RelWriter, SingleRel }
+import org.apache.flink.api.java.tuple.Tuple
+import org.apache.flink.streaming.api.datastream.{ AllWindowedStream,
DataStream, KeyedStream, WindowedStream }
+import org.apache.flink.streaming.api.windowing.assigners._
+import org.apache.flink.streaming.api.windowing.time.Time
+import org.apache.flink.streaming.api.windowing.windows.{ Window =>
DataStreamWindow }
+import org.apache.flink.table.api.StreamTableEnvironment
+import org.apache.flink.table.calcite.FlinkRelBuilder.NamedWindowProperty
+import org.apache.flink.table.calcite.FlinkTypeFactory
+import org.apache.flink.table.expressions._
+import org.apache.flink.table.plan.logical._
+import org.apache.flink.table.plan.nodes.CommonAggregate
+import org.apache.flink.table.plan.nodes.datastream.DataStreamAggregate._
+import org.apache.flink.table.runtime.aggregate.AggregateUtil._
+import org.apache.flink.table.runtime.aggregate._
+import org.apache.flink.table.typeutils.TypeCheckUtils.isTimeInterval
+import org.apache.flink.table.typeutils.{ RowIntervalTypeInfo,
TimeIntervalTypeInfo }
+import org.apache.flink.types.Row
+import org.apache.calcite.rel.logical.LogicalAggregate
+import org.apache.calcite.sql.SqlAggFunction
+import org.apache.flink.table.plan.nodes.datastream.DataStreamRel
+import org.apache.flink.table.api.TableException
+import org.apache.calcite.sql.fun.SqlSingleValueAggFunction
+import org.apache.flink.api.common.functions.RichMapFunction
+import org.apache.flink.api.common.functions.RichFlatMapFunction
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.util.Collector
+import org.apache.flink.api.common.state.ValueState
+import org.apache.flink.api.common.state.ValueStateDescriptor
+import org.apache.flink.api.java.typeutils.RowTypeInfo
+
+class DataStreamNonWindowAggregate(
+ calc: LogicalAggregate,
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ inputNode: RelNode,
+ rowType: RelDataType,
+ description: String)
+ extends SingleRel(cluster, traitSet, inputNode) with DataStreamRel {
+
+ override def deriveRowType(): RelDataType = rowType
+
+ override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]):
RelNode = {
+ new DataStreamNonWindowAggregate(
+ calc,
+ cluster,
+ traitSet,
+ inputs.get(0),
+ rowType,
+ description + calc.getId())
+ }
+
+ override def toString: String = {
+ s"Aggregate($calc)"
+ }
+
+ override def explainTerms(pw: RelWriter): RelWriter = {
+ super.explainTerms(pw)
+ .item("aggregate", calc)
+ .item("input", inputNode)
+ }
+
+ override def translateToPlan(tableEnv: StreamTableEnvironment):
DataStream[Row] = {
+
+ val inputDS =
getInput.asInstanceOf[DataStreamRel].translateToPlan(tableEnv)
+
+ //enable to extend for other types of aggregates that will not be
implemented in a window
+ calc.getAggCallList.get(0).getAggregation match {
+ case _: SqlSingleValueAggFunction =>
+ createLastSingleValueAggregate(inputDS)
+ case _ =>
+ throw new TableException("SQL/Table does not support such aggregates
type")
+ }
+ }
+
+ /**
+ * Create Aggregator for keeping the last value that was seen
+ */
+ def createLastSingleValueAggregate(
+ inputDS: DataStream[Row]): DataStream[Row] = {
+
+
+ // get the output types
+ val rowTypeInfo =
FlinkTypeFactory.toInternalRowTypeInfo(getRowType).asInstanceOf[RowTypeInfo]
+
+ inputDS.flatMap(new RichFlatMapFunction[Row, Row] {
+
+ @transient var state: ValueState[Row] = _
+
+ override def open(parameters: Configuration): Unit = {
+ super.open(parameters)
+ state = getRuntimeContext.getState(
+ new ValueStateDescriptor[Row]("state-last-element", rowTypeInfo))
+
+ }
+
+ override def flatMap(value: Row, out: Collector[Row]): Unit = {
+ state.update(value)
+ out.collect(value)
+ }
+ })
+ }
+
+}
+
+
diff --git
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/FlinkRuleSets.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/FlinkRuleSets.scala
index 952ee3495d4..7b2c6ca480a 100644
---
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/FlinkRuleSets.scala
+++
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/FlinkRuleSets.scala
@@ -183,7 +183,10 @@ object FlinkRuleSets {
// scan optimization
PushProjectIntoStreamTableSourceScanRule.INSTANCE,
- PushFilterIntoStreamTableSourceScanRule.INSTANCE
+ PushFilterIntoStreamTableSourceScanRule.INSTANCE,
+
+ DataStreamJoinRule.INSTANCE,
+ DataStreamNonWindowAggregateRule.INSTANCE
)
}
diff --git
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamJoinRule.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamJoinRule.scala
new file mode 100644
index 00000000000..f94ceca6801
--- /dev/null
+++
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamJoinRule.scala
@@ -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.plan.rules.datastream
+
+import org.apache.calcite.plan.{ Convention, RelOptRule, RelOptRuleCall,
RelTraitSet }
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.flink.table.api.TableException
+import org.apache.flink.table.expressions.Alias
+//import org.apache.flink.table.plan.logical.rel.
+import org.apache.calcite.rel.logical.LogicalJoin;
+import org.apache.flink.table.plan.nodes.datastream.DataStreamConvention
+import scala.collection.JavaConversions._
+import org.apache.calcite.plan.RelOptRule
+import org.apache.calcite.rel.logical.LogicalJoin
+import org.apache.calcite.plan.RelOptRuleCall
+import org.apache.calcite.plan.RelTraitSet
+import org.apache.flink.table.plan.nodes.datastream.DataStreamJoin
+import org.apache.calcite.rel.RelNode
+
+import org.apache.calcite.rel.convert.ConverterRule
+
+class DataStreamJoinRule
+ extends ConverterRule(
+ classOf[LogicalJoin],
+ Convention.NONE,
+ DataStreamConvention.INSTANCE,
+ "DataStreamJoinRule") {
+
+ override def matches(call: RelOptRuleCall): Boolean = {
+
+ super.matches(call)
+ }
+
+
+ override def convert(rel: RelNode): RelNode = {
+ val calc: LogicalJoin = rel.asInstanceOf[LogicalJoin]
+ val traitSet: RelTraitSet =
rel.getTraitSet.replace(DataStreamConvention.INSTANCE)
+
+ val left: RelNode = RelOptRule.convert(calc.getLeft(),
DataStreamConvention.INSTANCE)
+ val right: RelNode = RelOptRule.convert(calc.getRight(),
DataStreamConvention.INSTANCE)
+
+ new DataStreamJoin(
+ calc,
+ rel.getCluster,
+ traitSet,
+ left,
+ right,
+ rel.getRowType,
+ description + calc.getId())
+ }
+}
+
+object DataStreamJoinRule {
+ val INSTANCE: RelOptRule = new DataStreamJoinRule
+}
diff --git
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamNonWindowAggregateRule.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamNonWindowAggregateRule.scala
new file mode 100644
index 00000000000..276862b6f49
--- /dev/null
+++
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamNonWindowAggregateRule.scala
@@ -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.plan.rules.datastream
+
+import org.apache.calcite.plan.volcano.RelSubset
+import org.apache.calcite.plan.{ Convention, RelOptRule, RelOptRuleCall,
RelTraitSet }
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.calcite.rel.logical.{ LogicalFilter, LogicalCorrelate,
LogicalTableFunctionScan }
+import org.apache.calcite.rex.RexNode
+import org.apache.flink.table.plan.nodes.datastream.DataStreamConvention
+import org.apache.flink.table.plan.nodes.datastream.DataStreamCorrelate
+import org.apache.calcite.rel.logical.LogicalAggregate
+import
org.apache.flink.table.plan.nodes.datastream.DataStreamNonWindowAggregate
+
+/**
+ * Rule to convert a LogicalCorrelate into a DataStreamCorrelate.
+ */
+class DataStreamNonWindowAggregateRule
+ extends ConverterRule(
+ classOf[LogicalAggregate],
+ Convention.NONE,
+ DataStreamConvention.INSTANCE,
+ "DataStreamNonWindowAggregateRule") {
+
+ override def matches(call: RelOptRuleCall): Boolean = {
+ var iter = call.rels.iterator
+ while (iter.hasNext) {
+ var element = iter.next.asInstanceOf[LogicalAggregate]
+ //check for aggregates over windows defined via group clause
+ if (!element.groupSets.isEmpty()) {
+ return false
+ }
+ }
+ super.matches(call)
+ }
+
+ override def convert(rel: RelNode): RelNode = {
+ val calc: LogicalAggregate = rel.asInstanceOf[LogicalAggregate]
+ val traitSet: RelTraitSet =
rel.getTraitSet.replace(DataStreamConvention.INSTANCE)
+ val convInput: RelNode = RelOptRule.convert(calc.getInput(0),
DataStreamConvention.INSTANCE)
+
+ new DataStreamNonWindowAggregate(
+ calc,
+ rel.getCluster,
+ traitSet,
+ convInput,
+ rel.getRowType,
+ description + calc.getId())
+
+ }
+
+}
+
+object DataStreamNonWindowAggregateRule {
+ val INSTANCE: RelOptRule = new DataStreamNonWindowAggregateRule
+}
diff --git
a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/SqlITCase.scala
b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/SqlITCase.scala
index d5a140aca77..a8bc3600826 100644
---
a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/SqlITCase.scala
+++
b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/SqlITCase.scala
@@ -317,4 +317,47 @@ class SqlITCase extends StreamingWithStateTestBase {
result.addSink(new StreamITCase.StringSink)
env.execute()
}
+
+ @Test
+ def testInnerQuery(): Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val tEnv = TableEnvironment.getTableEnvironment(env)
+ StreamITCase.testResults = mutable.MutableList()
+
+ // for sum aggregation ensure that every time the order of each element is
consistent
+ env.setParallelism(1)
+
+ val t1 = env.fromCollection(data).toTable(tEnv).as('a, 'b, 'c)
+
+ tEnv.registerTable("s1", t1)
+
+ val t2 = StreamTestData.get5TupleDataStream(env)
+ .toTable(tEnv).as('a2, 'b2, 'c2, 'd2, 'e2)
+
+ tEnv.registerTable("s2", t2)
+
+ var sqlquery="SELECT a, (SELECT a2 FROM s2) FROM s1"
+
+ val result = tEnv.sql(sqlquery).toDataStream[Row]
+ result.addSink(new StreamITCase.StringSink)
+ env.execute()
+
+ val expected = mutable.MutableList(
+ "1,28",
+ "2,36",
+ "2,56",
+ "3,1",
+ "3,1",
+ "3,1",
+ "4,3",
+ "4,6",
+ "4,10",
+ "4,15",
+ "5,21",
+ "5,21",
+ "5,21",
+ "5,21",
+ "5,21")
+ assertEquals(expected.sorted, StreamITCase.testResults.sorted)
+ }
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
> Support for SQL inner queries for proctime
> ------------------------------------------
>
> Key: FLINK-6073
> URL: https://issues.apache.org/jira/browse/FLINK-6073
> Project: Flink
> Issue Type: New Feature
> Components: Table API & SQL
> Reporter: radu
> Assignee: radu
> Priority: Critical
> Labels: features
> Attachments: innerquery.png
>
>
> Time target: Proc Time
> **SQL targeted query examples:**
>
> Q1) `Select item, (select item2 from stream2 ) as itemExtern from stream1;`
> Comments: This is the main functionality targeted by this JIRA to enable to
> combine in the main query results from an inner query.
> Q2) `Select s1.item, (Select a2 from table as t2 where table.id = s1.id
> limit 1) from s1;`
> Comments:
> Another equivalent way to write the first example of inner query is with
> limit 1. This ensures the equivalency with the SingleElementAggregation used
> when translated the main target syntax for inner query. We must ensure that
> the 2 syntaxes are supported and implemented with the same functionality.
> There is the option also to select elements in the inner query from a table
> not just from a different stream. This should be a sub-JIRA issue implement
> this support.
> **Description:**
> Parsing the SQL inner query via calcite is translated to a join function
> (left join with always true condition) between the output of the query on the
> main stream and the output of a single output aggregation operation on the
> inner query. The translation logic is shown below
> ```
> LogicalJoin [condition=true;type=LEFT]
> LogicalSingleValue[type=aggregation]
> …logic of inner query (LogicalProject, LogicalScan…)
> …logical of main,external query (LogicalProject, LogicalScan…))
> ```
> `LogicalJoin[condition=true;type=LEFT] `– it can be considered as a special
> case operation rather than a proper join to be implemented between
> stream-to-stream. The implementation behavior should attach to the main
> stream output a value from a different query.
> `LogicalSingleValue[type=aggregation]` – it can be interpreted as the holder
> of the single value that results from the inner query. As this operator is
> the guarantee that the inner query will bring to the join no more than one
> value, there are several options on how to consider it’s functionality in the
> streaming context:
> 1. Throw an error if the inner query returns more than one result. This
> would be a typical behavior in the case of standard SQL over DB. However, it
> is very unlikely that a stream would only emit a single value. Therefore,
> such a behavior would be very limited for streams in the inner query.
> However, such a behavior might be more useful and common if the inner query
> is over a table.
> 1. We can interpret the usage of this parameter as the guarantee that at
> one moment only one value is selected. Therefore the behavior would rather be
> as a filter to select one value. This brings the option that the output of
> this operator evolves in time with the second stream that drives the inner
> query. The decision on when to evolve the stream should depend on what marks
> the evolution of the stream (processing time, watermarks/event time,
> ingestion time, window time partitions…).
> In this JIRA issue the evolution would be marked by the processing time. For
> this implementation the operator would work based on option 2. Hence at every
> moment the state of the operator that holds one value can evolve with the
> last elements. In this way the logic of the inner query is to select always
> the last element (fields, or other query related transformations based on the
> last value). This behavior is needed in many scenarios: (e.g., the typical
> problem of computing the total income, when incomes are in multiple
> currencies and the total needs to be computed in one currency by using always
> the last exchange rate).
> This behavior is motivated also by the functionality of the 3rd SQL query
> example – Q3 (using inner query as the input source for FROM ). In such
> scenarios, the selection in the main query would need to be done based on
> latest elements. Therefore with such a behavior the 2 types of queries (Q1
> and Q3) would provide the same, intuitive result.
> **Functionality example**
> Based on the logical translation plan, we exemplify next the behavior of the
> inner query applied on 2 streams that operate on processing time.
> SELECT amount, (SELECT exchange FROM inputstream1) AS field1 FROM inputstream2
> ||Time||Stream1||Stream2||Output||
> |T1| | 1.2| |
> |T2|User1,10| | (10,1.2)|
> |T3|User2,11| | (11,1.2)|
> |T4| | 1.3| |
> |T5|User3,9 | | (9,1.3)|
> |...|
> Note 1. For streams that would operate on event time, at moment T3 we would
> need to retract the previous outputs ((10, 1.2), (11,1.2) ) and reemit them
> as ((10,1.3), (11,1.3) ).
> Note 2. Rather than failing when a new value comes in the inner query we just
> update the state that holds the single value. If option 1 for the behavior of
> LogicalSingleValue is chosen, than an error should be triggered at moment T3.
> **Implementation option**
> Considering the notes and the option for the behavior the operator would be
> implemented by using the join function of flink with a custom always true
> join condition and an inner selection for the output based on the incoming
> direction (to mimic the left join). The single value selection can be
> implemented over a statefull flat map. In case the join is executed in
> parallel by multiple operators, than we either use a parallelism of 1 for the
> statefull flatmap (option 1) or we broadcast the outputs of the flatmap to
> all join instances to ensure consistency of the results (option 2).
> Considering that the flatMap functionality of selecting one value is light,
> option 1 is better. The design schema is shown below.
> !innerquery.png!
> **General logic of Join**
> ```
> leftDataStream.join(rightDataStream)
> .where(new ConstantConditionSelector())
> .equalTo(new ConstantConditionSelector())
> .window(window.create())
> .trigger(new LeftFireTrigger())
> .evictor(new Evictor())
> .apply(JoinFunction());
> ```
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
