You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by va...@apache.org on 2016/03/18 14:48:00 UTC
[06/50] [abbrv] flink git commit: [FLINK-3225] Implemented
optimization of Table API queries via Calcite
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/operators/DataSetTable.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/operators/DataSetTable.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/operators/DataSetTable.scala
deleted file mode 100644
index 65b97fb..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/operators/DataSetTable.scala
+++ /dev/null
@@ -1,66 +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.api.table.plan.operators
-
-import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeFactory}
-import org.apache.calcite.schema.impl.AbstractTable
-import org.apache.calcite.sql.`type`.SqlTypeName
-import org.apache.flink.api.common.typeutils.CompositeType
-import org.apache.flink.api.java.DataSet
-import org.apache.flink.api.table.plan.TypeConverter
-
-class DataSetTable[T](
- val dataSet: DataSet[T],
- val fieldNames: Array[String]) extends AbstractTable {
-
- // check uniquenss of field names
- if (fieldNames.length != fieldNames.toSet.size) {
- throw new scala.IllegalArgumentException(
- "Table field names must be unique.")
- }
-
- val dataSetType: CompositeType[T] =
- dataSet.getType match {
- case cType: CompositeType[T] =>
- cType
- case _ =>
- throw new scala.IllegalArgumentException(
- "DataSet must have a composite type.")
- }
-
- val fieldTypes: Array[SqlTypeName] =
- if (fieldNames.length == dataSetType.getArity) {
- (0 until dataSetType.getArity)
- .map(i => dataSetType.getTypeAt(i))
- .map(TypeConverter.typeInfoToSqlType)
- .toArray
- }
- else {
- throw new IllegalArgumentException(
- "Arity of DataSet type not equal to number of field names.")
- }
-
- override def getRowType(typeFactory: RelDataTypeFactory): RelDataType = {
- val builder = typeFactory.builder
- fieldNames.zip(fieldTypes)
- .foreach( f => builder.add(f._1, f._2).nullable(true) )
- builder.build
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/FlinkRuleSets.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/FlinkRuleSets.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/FlinkRuleSets.scala
new file mode 100644
index 0000000..97e8b32
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/FlinkRuleSets.scala
@@ -0,0 +1,120 @@
+/*
+ * 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.api.table.plan.rules
+
+import org.apache.calcite.rel.rules._
+import org.apache.calcite.tools.{RuleSets, RuleSet}
+import org.apache.flink.api.table.plan.rules.logical._
+import org.apache.flink.api.table.plan.rules.dataset._
+
+object FlinkRuleSets {
+
+ /**
+ * RuleSet to optimize plans for batch / DataSet exeuction
+ */
+ val DATASET_OPT_RULES: RuleSet = RuleSets.ofList(
+
+ // filter rules
+ FilterJoinRule.FILTER_ON_JOIN,
+ FilterJoinRule.JOIN,
+ FilterMergeRule.INSTANCE,
+ FilterAggregateTransposeRule.INSTANCE,
+
+ // push and merge projection rules
+ AggregateProjectMergeRule.INSTANCE,
+ ProjectMergeRule.INSTANCE,
+ ProjectFilterTransposeRule.INSTANCE,
+ FilterProjectTransposeRule.INSTANCE,
+ AggregateProjectPullUpConstantsRule.INSTANCE,
+ JoinPushExpressionsRule.INSTANCE,
+ ProjectJoinTransposeRule.INSTANCE,
+ ProjectRemoveRule.INSTANCE,
+ SortProjectTransposeRule.INSTANCE,
+ ProjectSortTransposeRule.INSTANCE,
+
+ // merge and push unions rules
+ // TODO: Add a rule to enforce binary unions
+ UnionEliminatorRule.INSTANCE,
+ JoinUnionTransposeRule.LEFT_UNION,
+ JoinUnionTransposeRule.RIGHT_UNION,
+ // non-all Union to all-union + distinct
+ UnionToDistinctRule.INSTANCE,
+
+ // aggregation rules
+ AggregateRemoveRule.INSTANCE,
+ AggregateJoinTransposeRule.EXTENDED,
+ AggregateUnionAggregateRule.INSTANCE,
+ AggregateReduceFunctionsRule.INSTANCE,
+ AggregateExpandDistinctAggregatesRule.INSTANCE,
+
+ // remove unnecessary sort rule
+ SortRemoveRule.INSTANCE,
+
+ // join reordering rules
+ JoinPushThroughJoinRule.LEFT,
+ JoinPushThroughJoinRule.RIGHT,
+ JoinAssociateRule.INSTANCE,
+ JoinCommuteRule.INSTANCE,
+ JoinCommuteRule.SWAP_OUTER,
+
+ // simplify expressions rules
+ ReduceExpressionsRule.CALC_INSTANCE,
+ ReduceExpressionsRule.FILTER_INSTANCE,
+ ReduceExpressionsRule.JOIN_INSTANCE,
+ ReduceExpressionsRule.PROJECT_INSTANCE,
+
+ // prune empty results rules
+ PruneEmptyRules.AGGREGATE_INSTANCE,
+ PruneEmptyRules.FILTER_INSTANCE,
+ PruneEmptyRules.JOIN_LEFT_INSTANCE,
+ PruneEmptyRules.JOIN_RIGHT_INSTANCE,
+ PruneEmptyRules.PROJECT_INSTANCE,
+ PruneEmptyRules.SORT_INSTANCE,
+ PruneEmptyRules.UNION_INSTANCE,
+
+ // calc rules
+ FilterCalcMergeRule.INSTANCE,
+ ProjectCalcMergeRule.INSTANCE,
+ FilterToCalcRule.INSTANCE,
+ ProjectToCalcRule.INSTANCE,
+ CalcMergeRule.INSTANCE,
+
+ // translate to logical Flink nodes
+ FlinkAggregateRule.INSTANCE,
+ FlinkCalcRule.INSTANCE,
+ FlinkFilterRule.INSTANCE,
+ FlinkJoinRule.INSTANCE,
+ FlinkProjectRule.INSTANCE,
+ FlinkScanRule.INSTANCE,
+ FlinkUnionRule.INSTANCE
+ )
+
+ val DATASET_TRANS_RULES: RuleSet = RuleSets.ofList(
+
+ // translate to DataSet nodes
+ DataSetAggregateRule.INSTANCE,
+ DataSetCalcRule.INSTANCE,
+ DataSetFilterRule.INSTANCE,
+ DataSetJoinRule.INSTANCE,
+ DataSetProjectRule.INSTANCE,
+ DataSetScanRule.INSTANCE,
+ DataSetUnionRule.INSTANCE
+ )
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetAggregateRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetAggregateRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetAggregateRule.scala
new file mode 100644
index 0000000..1d17d63
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetAggregateRule.scala
@@ -0,0 +1,53 @@
+/*
+ * 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.api.table.plan.rules.dataset
+
+import org.apache.calcite.plan.{RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetReduce}
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkAggregate, FlinkConvention}
+
+class DataSetAggregateRule
+ extends ConverterRule(
+ classOf[FlinkAggregate],
+ FlinkConvention.INSTANCE,
+ DataSetConvention.INSTANCE,
+ "DataSetAggregateRule")
+{
+
+ def convert(rel: RelNode): RelNode = {
+ val agg: FlinkAggregate = rel.asInstanceOf[FlinkAggregate]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+ val convInput: RelNode = RelOptRule.convert(agg.getInput, DataSetConvention.INSTANCE)
+
+ new DataSetReduce(
+ rel.getCluster,
+ traitSet,
+ convInput,
+ rel.getRowType,
+ agg.toString,
+ Array[Int](),
+ null)
+ }
+}
+
+object DataSetAggregateRule {
+ val INSTANCE: RelOptRule = new DataSetAggregateRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetCalcRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetCalcRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetCalcRule.scala
new file mode 100644
index 0000000..85e090c
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetCalcRule.scala
@@ -0,0 +1,52 @@
+/*
+ * 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.api.table.plan.rules.dataset
+
+import org.apache.calcite.plan.{RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetFlatMap}
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkCalc, FlinkConvention}
+
+class DataSetCalcRule
+ extends ConverterRule(
+ classOf[FlinkCalc],
+ FlinkConvention.INSTANCE,
+ DataSetConvention.INSTANCE,
+ "DataSetCalcRule")
+{
+
+ def convert(rel: RelNode): RelNode = {
+ val calc: FlinkCalc = rel.asInstanceOf[FlinkCalc]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+ val convInput: RelNode = RelOptRule.convert(calc.getInput, DataSetConvention.INSTANCE)
+
+ new DataSetFlatMap(
+ rel.getCluster,
+ traitSet,
+ convInput,
+ rel.getRowType,
+ calc.toString,
+ null)
+ }
+}
+
+object DataSetCalcRule {
+ val INSTANCE: RelOptRule = new DataSetCalcRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetFilterRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetFilterRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetFilterRule.scala
new file mode 100644
index 0000000..383c965
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetFilterRule.scala
@@ -0,0 +1,52 @@
+/*
+ * 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.api.table.plan.rules.dataset
+
+import org.apache.calcite.plan.{RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetFlatMap}
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkFilter, FlinkConvention}
+
+class DataSetFilterRule
+ extends ConverterRule(
+ classOf[FlinkFilter],
+ FlinkConvention.INSTANCE,
+ DataSetConvention.INSTANCE,
+ "DataSetFilterRule")
+{
+
+ def convert(rel: RelNode): RelNode = {
+ val filter: FlinkFilter = rel.asInstanceOf[FlinkFilter]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+ val convInput: RelNode = RelOptRule.convert(filter.getInput, DataSetConvention.INSTANCE)
+
+ new DataSetFlatMap(
+ rel.getCluster,
+ traitSet,
+ convInput,
+ rel.getRowType,
+ filter.toString,
+ null)
+ }
+}
+
+object DataSetFilterRule {
+ val INSTANCE: RelOptRule = new DataSetFilterRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetJoinRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetJoinRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetJoinRule.scala
new file mode 100644
index 0000000..3d2117d
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetJoinRule.scala
@@ -0,0 +1,59 @@
+/*
+ * 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.api.table.plan.rules.dataset
+
+import org.apache.calcite.plan.{RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.flink.api.java.operators.join.JoinType
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetJoin}
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkJoin, FlinkConvention}
+
+class DataSetJoinRule
+ extends ConverterRule(
+ classOf[FlinkJoin],
+ FlinkConvention.INSTANCE,
+ DataSetConvention.INSTANCE,
+ "DataSetJoinRule")
+{
+
+ def convert(rel: RelNode): RelNode = {
+ val join: FlinkJoin = rel.asInstanceOf[FlinkJoin]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+ val convLeft: RelNode = RelOptRule.convert(join.getInput(0), DataSetConvention.INSTANCE)
+ val convRight: RelNode = RelOptRule.convert(join.getInput(1), DataSetConvention.INSTANCE)
+
+ new DataSetJoin(
+ rel.getCluster,
+ traitSet,
+ convLeft,
+ convRight,
+ rel.getRowType,
+ join.toString,
+ Array[Int](),
+ Array[Int](),
+ JoinType.INNER,
+ null,
+ null)
+ }
+}
+
+object DataSetJoinRule {
+ val INSTANCE: RelOptRule = new DataSetJoinRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetProjectRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetProjectRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetProjectRule.scala
new file mode 100644
index 0000000..7796d66
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetProjectRule.scala
@@ -0,0 +1,52 @@
+/*
+ * 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.api.table.plan.rules.dataset
+
+import org.apache.calcite.plan.{RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetMap}
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkProject, FlinkConvention}
+
+class DataSetProjectRule
+ extends ConverterRule(
+ classOf[FlinkProject],
+ FlinkConvention.INSTANCE,
+ DataSetConvention.INSTANCE,
+ "DataSetProjectRule")
+{
+
+ def convert(rel: RelNode): RelNode = {
+ val proj: FlinkProject = rel.asInstanceOf[FlinkProject]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+ val convInput: RelNode = RelOptRule.convert(proj.getInput, DataSetConvention.INSTANCE)
+
+ new DataSetMap(
+ rel.getCluster,
+ traitSet,
+ convInput,
+ rel.getRowType,
+ proj.toString,
+ null)
+ }
+}
+
+object DataSetProjectRule {
+ val INSTANCE: RelOptRule = new DataSetProjectRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetScanRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetScanRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetScanRule.scala
new file mode 100644
index 0000000..937f3e2
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetScanRule.scala
@@ -0,0 +1,53 @@
+/*
+ * 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.api.table.plan.rules.dataset
+
+import org.apache.calcite.plan.{RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetSource}
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkScan, FlinkConvention}
+import org.apache.flink.api.table.plan.schema.DataSetTable
+
+class DataSetScanRule
+ extends ConverterRule(
+ classOf[FlinkScan],
+ FlinkConvention.INSTANCE,
+ DataSetConvention.INSTANCE,
+ "DataSetScanRule")
+{
+ def convert(rel: RelNode): RelNode = {
+ val scan: FlinkScan = rel.asInstanceOf[FlinkScan]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+ val dataSet: DataSet[_] = scan.getTable().unwrap(classOf[DataSetTable[_]]).dataSet
+
+ new DataSetSource(
+ rel.getCluster,
+ traitSet,
+ scan.getTable,
+ rel.getRowType,
+ dataSet
+ )
+ }
+}
+
+object DataSetScanRule {
+ val INSTANCE: RelOptRule = new DataSetScanRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetUnionRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetUnionRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetUnionRule.scala
new file mode 100644
index 0000000..a390374
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetUnionRule.scala
@@ -0,0 +1,53 @@
+/*
+ * 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.api.table.plan.rules.dataset
+
+import org.apache.calcite.plan.{RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetUnion}
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkUnion, FlinkConvention}
+
+class DataSetUnionRule
+ extends ConverterRule(
+ classOf[FlinkUnion],
+ FlinkConvention.INSTANCE,
+ DataSetConvention.INSTANCE,
+ "DataSetUnionRule")
+{
+
+ def convert(rel: RelNode): RelNode = {
+ val union: FlinkUnion = rel.asInstanceOf[FlinkUnion]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+ val convLeft: RelNode = RelOptRule.convert(union.getInput(0), DataSetConvention.INSTANCE)
+ val convRight: RelNode = RelOptRule.convert(union.getInput(1), DataSetConvention.INSTANCE)
+
+ new DataSetUnion(
+ rel.getCluster,
+ traitSet,
+ convLeft,
+ convRight,
+ rel.getRowType,
+ union.toString)
+ }
+}
+
+object DataSetUnionRule {
+ val INSTANCE: RelOptRule = new DataSetUnionRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkAggregateRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkAggregateRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkAggregateRule.scala
new file mode 100644
index 0000000..a5bfeb6
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkAggregateRule.scala
@@ -0,0 +1,53 @@
+/*
+ * 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.api.table.plan.rules.logical
+
+import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.calcite.rel.logical.LogicalAggregate
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkAggregate, FlinkConvention}
+
+class FlinkAggregateRule
+ extends ConverterRule(
+ classOf[LogicalAggregate],
+ Convention.NONE,
+ FlinkConvention.INSTANCE,
+ "FlinkAggregateRule")
+ {
+
+ def convert(rel: RelNode): RelNode = {
+ val agg: LogicalAggregate = rel.asInstanceOf[LogicalAggregate]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(FlinkConvention.INSTANCE)
+ val convInput: RelNode = RelOptRule.convert(agg.getInput, FlinkConvention.INSTANCE)
+
+ new FlinkAggregate(
+ rel.getCluster,
+ traitSet,
+ convInput,
+ agg.indicator,
+ agg.getGroupSet,
+ agg.getGroupSets,
+ agg.getAggCallList)
+ }
+ }
+
+object FlinkAggregateRule {
+ val INSTANCE: RelOptRule = new FlinkAggregateRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkCalcRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkCalcRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkCalcRule.scala
new file mode 100644
index 0000000..f40b04d
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkCalcRule.scala
@@ -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.api.table.plan.rules.logical
+
+import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.calcite.rel.logical.LogicalCalc
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkCalc, FlinkConvention}
+
+class FlinkCalcRule
+ extends ConverterRule(
+ classOf[LogicalCalc],
+ Convention.NONE,
+ FlinkConvention.INSTANCE,
+ "FlinkCalcRule")
+ {
+
+ def convert(rel: RelNode): RelNode = {
+ val calc: LogicalCalc = rel.asInstanceOf[LogicalCalc]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(FlinkConvention.INSTANCE)
+ val convInput: RelNode = RelOptRule.convert(calc.getInput, FlinkConvention.INSTANCE)
+
+ new FlinkCalc(
+ rel.getCluster,
+ traitSet,
+ convInput,
+ calc.getProgram)
+ }
+ }
+
+object FlinkCalcRule {
+ val INSTANCE: RelOptRule = new FlinkCalcRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkFilterRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkFilterRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkFilterRule.scala
new file mode 100644
index 0000000..25df8e8
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkFilterRule.scala
@@ -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.api.table.plan.rules.logical
+
+import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.calcite.rel.logical.LogicalFilter
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkFilter, FlinkConvention}
+
+class FlinkFilterRule
+ extends ConverterRule(
+ classOf[LogicalFilter],
+ Convention.NONE,
+ FlinkConvention.INSTANCE,
+ "FlinkFilterRule")
+ {
+
+ def convert(rel: RelNode): RelNode = {
+ val filter: LogicalFilter = rel.asInstanceOf[LogicalFilter]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(FlinkConvention.INSTANCE)
+ val convInput: RelNode = RelOptRule.convert(filter.getInput, FlinkConvention.INSTANCE)
+
+ new FlinkFilter(
+ rel.getCluster,
+ traitSet,
+ convInput,
+ filter.getCondition)
+ }
+ }
+
+object FlinkFilterRule {
+ val INSTANCE: RelOptRule = new FlinkFilterRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinRule.scala
new file mode 100644
index 0000000..3826c9a
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinRule.scala
@@ -0,0 +1,54 @@
+/*
+ * 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.api.table.plan.rules.logical
+
+import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.calcite.rel.logical.LogicalJoin
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkJoin, FlinkConvention}
+
+class FlinkJoinRule
+ extends ConverterRule(
+ classOf[LogicalJoin],
+ Convention.NONE,
+ FlinkConvention.INSTANCE,
+ "FlinkJoinRule")
+ {
+
+ def convert(rel: RelNode): RelNode = {
+ val join: LogicalJoin = rel.asInstanceOf[LogicalJoin]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(FlinkConvention.INSTANCE)
+ val convLeft: RelNode = RelOptRule.convert(join.getInput(0), FlinkConvention.INSTANCE)
+ val convRight: RelNode = RelOptRule.convert(join.getInput(1), FlinkConvention.INSTANCE)
+
+ new FlinkJoin(
+ rel.getCluster,
+ traitSet,
+ convLeft,
+ convRight,
+ join.getCondition,
+ join.getJoinType,
+ join.getVariablesStopped)
+ }
+ }
+
+object FlinkJoinRule {
+ val INSTANCE: RelOptRule = new FlinkJoinRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkProjectRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkProjectRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkProjectRule.scala
new file mode 100644
index 0000000..d0e1410
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkProjectRule.scala
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.table.plan.rules.logical
+
+import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.calcite.rel.logical.LogicalProject
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkProject, FlinkConvention}
+
+class FlinkProjectRule
+ extends ConverterRule(
+ classOf[LogicalProject],
+ Convention.NONE,
+ FlinkConvention.INSTANCE,
+ "FlinkProjectRule")
+ {
+
+ def convert(rel: RelNode): RelNode = {
+ val proj: LogicalProject = rel.asInstanceOf[LogicalProject]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(FlinkConvention.INSTANCE)
+ val convInput: RelNode = RelOptRule.convert(proj.getInput, FlinkConvention.INSTANCE)
+
+ new FlinkProject(
+ rel.getCluster,
+ traitSet,
+ convInput,
+ proj.getProjects,
+ proj.getRowType)
+ }
+ }
+
+object FlinkProjectRule {
+ val INSTANCE: RelOptRule = new FlinkProjectRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkScanRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkScanRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkScanRule.scala
new file mode 100644
index 0000000..d789770
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkScanRule.scala
@@ -0,0 +1,53 @@
+/*
+ * 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.api.table.plan.rules.logical
+
+import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.calcite.rel.core.TableScan
+import org.apache.calcite.rel.logical.LogicalTableScan
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetSource}
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkScan, FlinkConvention}
+import org.apache.flink.api.table.plan.schema.DataSetTable
+
+class FlinkScanRule
+ extends ConverterRule(
+ classOf[LogicalTableScan],
+ Convention.NONE,
+ FlinkConvention.INSTANCE,
+ "FlinkScanRule")
+ {
+ def convert(rel: RelNode): RelNode = {
+ val scan: TableScan = rel.asInstanceOf[TableScan]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(FlinkConvention.INSTANCE)
+ val dataSet: DataSet[_] = scan.getTable().unwrap(classOf[DataSetTable[_]]).dataSet
+
+ new FlinkScan(
+ rel.getCluster,
+ traitSet,
+ scan.getTable
+ )
+ }
+ }
+
+object FlinkScanRule {
+ val INSTANCE: RelOptRule = new FlinkScanRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkUnionRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkUnionRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkUnionRule.scala
new file mode 100644
index 0000000..d9869f8
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkUnionRule.scala
@@ -0,0 +1,54 @@
+/*
+ * 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.api.table.plan.rules.logical
+
+import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.calcite.rel.logical.LogicalUnion
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkUnion, FlinkConvention}
+
+import scala.collection.JavaConversions._
+
+class FlinkUnionRule
+ extends ConverterRule(
+ classOf[LogicalUnion],
+ Convention.NONE,
+ FlinkConvention.INSTANCE,
+ "FlinkUnionRule")
+ {
+
+ def convert(rel: RelNode): RelNode = {
+ val union: LogicalUnion = rel.asInstanceOf[LogicalUnion]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(FlinkConvention.INSTANCE)
+ val convInputs = union.getInputs.toList.map(
+ RelOptRule.convert(_, FlinkConvention.INSTANCE)
+ )
+
+ new FlinkUnion(
+ rel.getCluster,
+ traitSet,
+ convInputs,
+ union.all)
+ }
+ }
+
+object FlinkUnionRule {
+ val INSTANCE: RelOptRule = new FlinkUnionRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/schema/DataSetTable.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/schema/DataSetTable.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/schema/DataSetTable.scala
new file mode 100644
index 0000000..e6aecab
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/schema/DataSetTable.scala
@@ -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.api.table.plan.schema
+
+import java.lang.Double
+import java.util
+import java.util.Collections
+
+import org.apache.calcite.rel.{RelCollation, RelDistribution}
+import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeFactory}
+import org.apache.calcite.schema.Statistic
+import org.apache.calcite.schema.impl.AbstractTable
+import org.apache.calcite.sql.`type`.SqlTypeName
+import org.apache.calcite.util.ImmutableBitSet
+import org.apache.flink.api.common.typeutils.CompositeType
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.plan.TypeConverter
+
+class DataSetTable[T](
+ val dataSet: DataSet[T],
+ val fieldNames: Array[String])
+ extends AbstractTable {
+
+ // check uniquenss of field names
+ if (fieldNames.length != fieldNames.toSet.size) {
+ throw new scala.IllegalArgumentException(
+ "Table field names must be unique.")
+ }
+
+ val dataSetType: CompositeType[T] =
+ dataSet.getType match {
+ case cType: CompositeType[T] =>
+ cType
+ case _ =>
+ throw new scala.IllegalArgumentException(
+ "DataSet must have a composite type.")
+ }
+
+ val fieldTypes: Array[SqlTypeName] =
+ if (fieldNames.length == dataSetType.getArity) {
+ (0 until dataSetType.getArity)
+ .map(i => dataSetType.getTypeAt(i))
+ .map(TypeConverter.typeInfoToSqlType)
+ .toArray
+ }
+ else {
+ throw new IllegalArgumentException(
+ "Arity of DataSet type not equal to number of field names.")
+ }
+
+ override def getRowType(typeFactory: RelDataTypeFactory): RelDataType = {
+ val builder = typeFactory.builder
+ fieldNames.zip(fieldTypes)
+ .foreach( f => builder.add(f._1, f._2).nullable(true) )
+ builder.build
+ }
+
+// override def getStatistic: Statistic = {
+// new DefaultDataSetStatistic
+// }
+
+}
+
+class DefaultDataSetStatistic extends Statistic {
+
+ override def getRowCount: Double = 1000d
+
+ override def getCollations: util.List[RelCollation] = Collections.emptyList()
+
+ override def isKey(columns: ImmutableBitSet): Boolean = false
+
+ override def getDistribution: RelDistribution = null
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/table.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/table.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/table.scala
index 271aa99..138cd70 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/table.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/table.scala
@@ -270,7 +270,6 @@ class Table(
relBuilder.join(JoinRelType.INNER, relBuilder.literal(true))
val join = relBuilder.build()
- val rowT = join.getRowType()
new Table(join, relBuilder)
}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AggregationsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AggregationsITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AggregationsITCase.java
index 447acad..a3d5edc 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AggregationsITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AggregationsITCase.java
@@ -35,7 +35,9 @@ package org.apache.flink.api.java.table.test;
* limitations under the License.
*/
+import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.table.ExpressionException;
+import org.apache.flink.api.table.Row;
import org.apache.flink.api.table.Table;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.table.TableEnvironment;
@@ -48,6 +50,9 @@ import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import scala.NotImplementedError;
+
+import java.util.List;
@RunWith(Parameterized.class)
public class AggregationsITCase extends MultipleProgramsTestBase {
@@ -57,7 +62,7 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
super(mode);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testAggregationTypes() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -66,10 +71,10 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
Table result = table.select("f0.sum, f0.min, f0.max, f0.count, f0.avg");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "231,1,21,21,11";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "231,1,21,21,11";
+ compareResultAsText(results, expected);
}
@Test(expected = IllegalArgumentException.class)
@@ -83,13 +88,13 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
Table result =
table.select("foo.avg");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testWorkingAggregationDataTypes() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -105,13 +110,13 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
Table result =
table.select("f0.avg, f1.avg, f2.avg, f3.avg, f4.avg, f5.avg, f6.count");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "1,1,1,1,1.5,1.5,2";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "1,1,1,1,1.5,1.5,2";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testAggregationWithArithmetic() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -128,13 +133,13 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
table.select("(f0 + 2).avg + 2, f1.count + 5");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "5.5,7";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "5.5,7";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testAggregationWithTwoCount() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -150,11 +155,10 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
Table result =
table.select("f0.count, f1.count");
-
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "2,2";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "2,2";
+ compareResultAsText(results, expected);
}
// Calcite does not eagerly check type compatibility
@@ -172,11 +176,10 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
Table result =
table.select("f1.sum");
-
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
@Test(expected = ExpressionException.class)
@@ -192,11 +195,10 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
Table result =
table.select("f0.sum.sum");
-
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AsITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AsITCase.java
index f706b48..f257c32 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AsITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AsITCase.java
@@ -18,7 +18,6 @@
package org.apache.flink.api.java.table.test;
-import org.apache.flink.api.table.ExpressionException;
import org.apache.flink.api.table.Table;
import org.apache.flink.api.table.Row;
import org.apache.flink.api.java.DataSet;
@@ -29,6 +28,7 @@ import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import scala.NotImplementedError;
import java.util.List;
@@ -40,7 +40,7 @@ public class AsITCase extends MultipleProgramsTestBase {
super(mode);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testAs() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -48,15 +48,15 @@ public class AsITCase extends MultipleProgramsTestBase {
Table table =
tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b, c");
-// DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "1,1,Hi\n" + "2,2,Hello\n" + "3,2,Hello world\n" + "4,3,Hello world, " +
-// "how are you?\n" + "5,3,I am fine.\n" + "6,3,Luke Skywalker\n" + "7,4," +
-// "Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5," +
-// "Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5," +
-// "Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19," +
-// "6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "1,1,Hi\n" + "2,2,Hello\n" + "3,2,Hello world\n" + "4,3,Hello world, " +
+ "how are you?\n" + "5,3,I am fine.\n" + "6,3,Luke Skywalker\n" + "7,4," +
+ "Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5," +
+ "Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5," +
+ "Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19," +
+ "6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n";
+ compareResultAsText(results, expected);
}
@Test(expected = IllegalArgumentException.class)
@@ -67,10 +67,10 @@ public class AsITCase extends MultipleProgramsTestBase {
Table table =
tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b");
-// DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
@Test(expected = IllegalArgumentException.class)
@@ -81,10 +81,10 @@ public class AsITCase extends MultipleProgramsTestBase {
Table table =
tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b, c, d");
-// DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
@Test(expected = IllegalArgumentException.class)
@@ -95,10 +95,10 @@ public class AsITCase extends MultipleProgramsTestBase {
Table table =
tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b, b");
-// DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
@Test(expected = IllegalArgumentException.class)
@@ -109,10 +109,10 @@ public class AsITCase extends MultipleProgramsTestBase {
Table table =
tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a + 1, b, c");
-// DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
@Test(expected = IllegalArgumentException.class)
@@ -124,10 +124,10 @@ public class AsITCase extends MultipleProgramsTestBase {
tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a as foo, b," +
" c");
-// DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/CastingITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/CastingITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/CastingITCase.java
index 5e99fd9..cde78ce 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/CastingITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/CastingITCase.java
@@ -32,6 +32,7 @@ import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import scala.NotImplementedError;
import java.util.List;
@@ -43,7 +44,7 @@ public class CastingITCase extends MultipleProgramsTestBase {
super(mode);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testNumericAutocastInArithmetic() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -57,13 +58,13 @@ public class CastingITCase extends MultipleProgramsTestBase {
Table result = table.select("f0 + 1, f1 +" +
" 1, f2 + 1L, f3 + 1.0f, f4 + 1.0d, f5 + 1");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "2,2,2,2.0,2.0,2.0";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "2,2,2,2.0,2.0,2.0";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testNumericAutocastInComparison() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -79,13 +80,13 @@ public class CastingITCase extends MultipleProgramsTestBase {
Table result = table
.filter("a > 1 && b > 1 && c > 1L && d > 1.0f && e > 1.0d && f > 1");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "2,2,2,2,2.0,2.0,Hello";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "2,2,2,2,2.0,2.0,Hello";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testCastFromString() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -99,13 +100,13 @@ public class CastingITCase extends MultipleProgramsTestBase {
Table result = table.select(
"f0.cast(BYTE), f0.cast(SHORT), f0.cast(INT), f0.cast(LONG), f2.cast(DOUBLE), f2.cast(FLOAT), f1.cast(BOOL)");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "1,1,1,1,2.0,2.0,true\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "1,1,1,1,2.0,2.0,true\n";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testCastDateFromString() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -120,14 +121,14 @@ public class CastingITCase extends MultipleProgramsTestBase {
.select("f0.cast(DATE) AS f0, f1.cast(DATE) AS f1, f2.cast(DATE) AS f2, f3.cast(DATE) AS f3")
.select("f0.cast(STRING), f1.cast(STRING), f2.cast(STRING), f3.cast(STRING)");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "2011-05-03 00:00:00.000,1970-01-01 15:51:36.000,2011-05-03 15:51:36.000," +
-// "1970-01-17 17:47:53.775\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "2011-05-03 00:00:00.000,1970-01-01 15:51:36.000,2011-05-03 15:51:36.000," +
+ "1970-01-17 17:47:53.775\n";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testCastDateToStringAndLong() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -142,10 +143,10 @@ public class CastingITCase extends MultipleProgramsTestBase {
.select("f0.cast(DATE) AS f0, f1.cast(DATE) AS f1")
.select("f0.cast(STRING), f0.cast(LONG), f1.cast(STRING), f1.cast(LONG)");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "2011-05-03 15:51:36.000,1304437896000,2011-05-03 15:51:36.000,1304437896000\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "2011-05-03 15:51:36.000,1304437896000,2011-05-03 15:51:36.000,1304437896000\n";
+ compareResultAsText(results, expected);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/ExpressionsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/ExpressionsITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/ExpressionsITCase.java
index 3bbc120..2a17087 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/ExpressionsITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/ExpressionsITCase.java
@@ -18,7 +18,6 @@
package org.apache.flink.api.java.table.test;
-import org.apache.flink.api.table.ExpressionException;
import org.apache.flink.api.table.Table;
import org.apache.flink.api.table.Row;
import org.apache.flink.api.java.DataSet;
@@ -31,6 +30,7 @@ import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import scala.NotImplementedError;
import java.util.List;
@@ -42,7 +42,7 @@ public class ExpressionsITCase extends MultipleProgramsTestBase {
super(mode);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testArithmetic() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -56,13 +56,13 @@ public class ExpressionsITCase extends MultipleProgramsTestBase {
Table result = table.select(
"a - 5, a + 5, a / 2, a * 2, a % 2, -a");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "0,10,2,10,1,-5";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "0,10,2,10,1,-5";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testLogic() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -76,13 +76,13 @@ public class ExpressionsITCase extends MultipleProgramsTestBase {
Table result = table.select(
"b && true, b && false, b || false, !b");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "true,false,true,false";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "true,false,true,false";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testComparisons() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -96,10 +96,10 @@ public class ExpressionsITCase extends MultipleProgramsTestBase {
Table result = table.select(
"a > c, a >= b, a < c, a.isNull, a.isNotNull");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "true,true,false,false,true";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "true,true,false,false,true";
+ compareResultAsText(results, expected);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
index 68925fb..a3ab10f 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
@@ -29,6 +29,7 @@ import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import scala.NotImplementedError;
import java.util.List;
@@ -40,7 +41,7 @@ public class FilterITCase extends MultipleProgramsTestBase {
super(mode);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testAllRejectingFilter() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -53,13 +54,13 @@ public class FilterITCase extends MultipleProgramsTestBase {
Table result = table
.filter("false");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "\n";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testAllPassingFilter() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -72,18 +73,18 @@ public class FilterITCase extends MultipleProgramsTestBase {
Table result = table
.filter("true");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "1,1,Hi\n" + "2,2,Hello\n" + "3,2,Hello world\n" + "4,3,Hello world, " +
-// "how are you?\n" + "5,3,I am fine.\n" + "6,3,Luke Skywalker\n" + "7,4," +
-// "Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5," +
-// "Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5," +
-// "Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19," +
-// "6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "1,1,Hi\n" + "2,2,Hello\n" + "3,2,Hello world\n" + "4,3,Hello world, " +
+ "how are you?\n" + "5,3,I am fine.\n" + "6,3,Luke Skywalker\n" + "7,4," +
+ "Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5," +
+ "Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5," +
+ "Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19," +
+ "6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testFilterOnIntegerTupleField() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -96,15 +97,15 @@ public class FilterITCase extends MultipleProgramsTestBase {
Table result = table
.filter(" a % 2 = 0 ");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "2,2,Hello\n" + "4,3,Hello world, how are you?\n" + "6,3,Luke Skywalker\n" + "8,4," +
-// "Comment#2\n" + "10,4,Comment#4\n" + "12,5,Comment#6\n" + "14,5,Comment#8\n" + "16,6," +
-// "Comment#10\n" + "18,6,Comment#12\n" + "20,6,Comment#14\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "2,2,Hello\n" + "4,3,Hello world, how are you?\n" + "6,3,Luke Skywalker\n" + "8,4," +
+ "Comment#2\n" + "10,4,Comment#4\n" + "12,5,Comment#6\n" + "14,5,Comment#8\n" + "16,6," +
+ "Comment#10\n" + "18,6,Comment#12\n" + "20,6,Comment#14\n";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testNotEquals() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -117,15 +118,15 @@ public class FilterITCase extends MultipleProgramsTestBase {
Table result = table
.filter("!( a % 2 <> 0 ) ");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "2,2,Hello\n" + "4,3,Hello world, how are you?\n" + "6,3,Luke Skywalker\n" + "8,4," +
-// "Comment#2\n" + "10,4,Comment#4\n" + "12,5,Comment#6\n" + "14,5,Comment#8\n" + "16,6," +
-// "Comment#10\n" + "18,6,Comment#12\n" + "20,6,Comment#14\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "2,2,Hello\n" + "4,3,Hello world, how are you?\n" + "6,3,Luke Skywalker\n" + "8,4," +
+ "Comment#2\n" + "10,4,Comment#4\n" + "12,5,Comment#6\n" + "14,5,Comment#8\n" + "16,6," +
+ "Comment#10\n" + "18,6,Comment#12\n" + "20,6,Comment#14\n";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testIntegerBiggerThan128() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -136,10 +137,10 @@ public class FilterITCase extends MultipleProgramsTestBase {
Table result = table.filter("a = 300 ");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "300,1,Hello\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "300,1,Hello\n";
+ compareResultAsText(results, expected);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/GroupedAggregationsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/GroupedAggregationsITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/GroupedAggregationsITCase.java
index 2add694..d5dc56a 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/GroupedAggregationsITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/GroupedAggregationsITCase.java
@@ -18,7 +18,6 @@
package org.apache.flink.api.java.table.test;
-import org.apache.flink.api.table.ExpressionException;
import org.apache.flink.api.table.Table;
import org.apache.flink.api.table.Row;
import org.apache.flink.api.java.DataSet;
@@ -30,6 +29,7 @@ import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import scala.NotImplementedError;
import java.util.List;
@@ -54,13 +54,13 @@ public class GroupedAggregationsITCase extends MultipleProgramsTestBase {
Table result = table
.groupBy("foo").select("a.avg");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testGroupedAggregate() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -73,13 +73,13 @@ public class GroupedAggregationsITCase extends MultipleProgramsTestBase {
Table result = table
.groupBy("b").select("b, a.sum");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "1,1\n" + "2,5\n" + "3,15\n" + "4,34\n" + "5,65\n" + "6,111\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "1,1\n" + "2,5\n" + "3,15\n" + "4,34\n" + "5,65\n" + "6,111\n";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testGroupingKeyForwardIfNotUsed() throws Exception {
// the grouping key needs to be forwarded to the intermediate DataSet, even
@@ -96,13 +96,13 @@ public class GroupedAggregationsITCase extends MultipleProgramsTestBase {
Table result = table
.groupBy("b").select("a.sum");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "1\n" + "5\n" + "15\n" + "34\n" + "65\n" + "111\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "1\n" + "5\n" + "15\n" + "34\n" + "65\n" + "111\n";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testGroupNoAggregation() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
@@ -116,10 +116,10 @@ public class GroupedAggregationsITCase extends MultipleProgramsTestBase {
Table result = table
.groupBy("b").select("a.sum as d, b").groupBy("b, d").select("b");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// String expected = "1\n" + "2\n" + "3\n" + "4\n" + "5\n" + "6\n";
-// List<Row> results = ds.collect();
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ String expected = "1\n" + "2\n" + "3\n" + "4\n" + "5\n" + "6\n";
+ List<Row> results = ds.collect();
+ compareResultAsText(results, expected);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/JoinITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/JoinITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/JoinITCase.java
index 2b44a87..2e1f4a7 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/JoinITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/JoinITCase.java
@@ -18,6 +18,7 @@
package org.apache.flink.api.java.table.test;
+import org.apache.flink.api.table.Row;
import org.apache.flink.api.table.Table;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
@@ -30,6 +31,9 @@ import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import scala.NotImplementedError;
+
+import java.util.List;
@RunWith(Parameterized.class)
@@ -40,7 +44,7 @@ public class JoinITCase extends MultipleProgramsTestBase {
super(mode);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testJoin() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -53,13 +57,13 @@ public class JoinITCase extends MultipleProgramsTestBase {
Table result = in1.join(in2).where("b === e").select("c, g");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt\n";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testJoinWithFilter() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -72,13 +76,13 @@ public class JoinITCase extends MultipleProgramsTestBase {
Table result = in1.join(in2).where("b === e && b < 2").select("c, g");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "Hi,Hallo\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "Hi,Hallo\n";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testJoinWithMultipleKeys() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -91,11 +95,11 @@ public class JoinITCase extends MultipleProgramsTestBase {
Table result = in1.join(in2).where("a === d && b === h").select("c, g");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt wie gehts?\n" +
-// "Hello world,ABC\n" + "I am fine.,HIJ\n" + "I am fine.,IJK\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt wie gehts?\n" +
+ "Hello world,ABC\n" + "I am fine.,HIJ\n" + "I am fine.,IJK\n";
+ compareResultAsText(results, expected);
}
@Test(expected = IllegalArgumentException.class)
@@ -111,10 +115,10 @@ public class JoinITCase extends MultipleProgramsTestBase {
Table result = in1.join(in2).where("foo === e").select("c, g");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
// Calcite does not eagerly check the compatibility of compared types
@@ -133,10 +137,10 @@ public class JoinITCase extends MultipleProgramsTestBase {
Table result = in1
.join(in2).where("a === g").select("c, g");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
@Test(expected = IllegalArgumentException.class)
@@ -153,13 +157,13 @@ public class JoinITCase extends MultipleProgramsTestBase {
Table result = in1
.join(in2).where("a === d").select("c, g");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testJoinWithAggregation() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -173,10 +177,10 @@ public class JoinITCase extends MultipleProgramsTestBase {
Table result = in1
.join(in2).where("a === d").select("g.count");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "6";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "6";
+ compareResultAsText(results, expected);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/dee1876b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/PojoGroupingITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/PojoGroupingITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/PojoGroupingITCase.java
index f19b8c1..5ef0235 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/PojoGroupingITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/PojoGroupingITCase.java
@@ -30,6 +30,7 @@ import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import scala.NotImplementedError;
@RunWith(Parameterized.class)
public class PojoGroupingITCase extends MultipleProgramsTestBase {
@@ -38,7 +39,7 @@ public class PojoGroupingITCase extends MultipleProgramsTestBase {
super(mode);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testPojoGrouping() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
@@ -54,14 +55,14 @@ public class PojoGroupingITCase extends MultipleProgramsTestBase {
.select("groupMe, value, name")
.where("groupMe != 'B'");
-// DataSet<MyPojo> myPojos = tableEnv.toDataSet(table, MyPojo.class);
-//
-// DataSet<MyPojo> result = myPojos.groupBy("groupMe")
-// .sortGroup("value", Order.DESCENDING)
-// .first(1);
-//
-// List<MyPojo> resultList = result.collect();
-// compareResultAsText(resultList, "A,24.0,Y");
+ DataSet<MyPojo> myPojos = tableEnv.toDataSet(table, MyPojo.class);
+
+ DataSet<MyPojo> result = myPojos.groupBy("groupMe")
+ .sortGroup("value", Order.DESCENDING)
+ .first(1);
+
+ List<MyPojo> resultList = result.collect();
+ compareResultAsText(resultList, "A,24.0,Y");
}
public static class MyPojo implements Serializable {