flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fhue...@apache.org
Subject [42/50] [abbrv] flink git commit: [FLINK-3225] Implemented optimization of Table API queries via Calcite
Date Fri, 12 Feb 2016 11:30:07 GMT
[FLINK-3225] Implemented optimization of Table API queries via Calcite

- added logical Flink nodes and translation rules
- added stubs for DataSet translation rules
- ported DataSetNodes to Scala
- reactivated tests and added expected NotImplementedError


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/fe5e4065
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/fe5e4065
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/fe5e4065

Branch: refs/heads/tableOnCalcite
Commit: fe5e4065643dcb208f3990897f171780311502e9
Parents: 20235e0
Author: Fabian Hueske <fhueske@apache.org>
Authored: Tue Jan 26 13:22:38 2016 +0100
Committer: Fabian Hueske <fhueske@apache.org>
Committed: Fri Feb 12 11:34:09 2016 +0100

----------------------------------------------------------------------
 .../apache/flink/api/table/package-info.java    |  33 -----
 .../api/table/sql/calcite/DataSetRelNode.java   |  29 -----
 .../table/sql/calcite/node/DataSetExchange.java |  60 ----------
 .../table/sql/calcite/node/DataSetFlatMap.java  |  56 ---------
 .../api/table/sql/calcite/node/DataSetJoin.java |  80 -------------
 .../api/table/sql/calcite/node/DataSetMap.java  |  58 ---------
 .../table/sql/calcite/node/DataSetReduce.java   |  58 ---------
 .../sql/calcite/node/DataSetReduceGroup.java    |  62 ----------
 .../api/table/sql/calcite/node/DataSetSort.java |  59 ---------
 .../table/sql/calcite/node/DataSetSource.java   |  55 ---------
 .../table/sql/calcite/node/DataSetUnion.java    |  51 --------
 .../api/java/table/JavaBatchTranslator.scala    |  68 +++++++----
 .../api/table/plan/TranslationContext.scala     |  79 ++++++++++++
 .../plan/nodes/dataset/DataSetConvention.scala  |  42 +++++++
 .../plan/nodes/dataset/DataSetExchange.scala    |  63 ++++++++++
 .../plan/nodes/dataset/DataSetFlatMap.scala     |  62 ++++++++++
 .../plan/nodes/dataset/DataSetGroupReduce.scala |  63 ++++++++++
 .../table/plan/nodes/dataset/DataSetJoin.scala  |  73 +++++++++++
 .../table/plan/nodes/dataset/DataSetMap.scala   |  63 ++++++++++
 .../plan/nodes/dataset/DataSetReduce.scala      |  63 ++++++++++
 .../table/plan/nodes/dataset/DataSetRel.scala   |  33 +++++
 .../table/plan/nodes/dataset/DataSetSort.scala  |  62 ++++++++++
 .../plan/nodes/dataset/DataSetSource.scala      |  55 +++++++++
 .../table/plan/nodes/dataset/DataSetUnion.scala |  62 ++++++++++
 .../plan/nodes/logical/FlinkAggregate.scala     |  76 ++++++++++++
 .../table/plan/nodes/logical/FlinkCalc.scala    |  37 ++++++
 .../plan/nodes/logical/FlinkConvention.scala    |  42 +++++++
 .../table/plan/nodes/logical/FlinkFilter.scala  |  42 +++++++
 .../table/plan/nodes/logical/FlinkJoin.scala    |  46 +++++++
 .../table/plan/nodes/logical/FlinkProject.scala |  45 +++++++
 .../api/table/plan/nodes/logical/FlinkRel.scala |  25 ++++
 .../table/plan/nodes/logical/FlinkScan.scala    |  31 +++++
 .../table/plan/nodes/logical/FlinkUnion.scala   |  38 ++++++
 .../api/table/plan/operators/DataSetTable.scala |  66 ----------
 .../api/table/plan/rules/FlinkRuleSets.scala    | 120 +++++++++++++++++++
 .../rules/dataset/DataSetAggregateRule.scala    |  53 ++++++++
 .../plan/rules/dataset/DataSetCalcRule.scala    |  52 ++++++++
 .../plan/rules/dataset/DataSetFilterRule.scala  |  52 ++++++++
 .../plan/rules/dataset/DataSetJoinRule.scala    |  59 +++++++++
 .../plan/rules/dataset/DataSetProjectRule.scala |  52 ++++++++
 .../plan/rules/dataset/DataSetScanRule.scala    |  53 ++++++++
 .../plan/rules/dataset/DataSetUnionRule.scala   |  53 ++++++++
 .../plan/rules/logical/FlinkAggregateRule.scala |  53 ++++++++
 .../plan/rules/logical/FlinkCalcRule.scala      |  50 ++++++++
 .../plan/rules/logical/FlinkFilterRule.scala    |  50 ++++++++
 .../plan/rules/logical/FlinkJoinRule.scala      |  54 +++++++++
 .../plan/rules/logical/FlinkProjectRule.scala   |  51 ++++++++
 .../plan/rules/logical/FlinkScanRule.scala      |  53 ++++++++
 .../plan/rules/logical/FlinkUnionRule.scala     |  54 +++++++++
 .../api/table/plan/schema/DataSetTable.scala    |  89 ++++++++++++++
 .../org/apache/flink/api/table/table.scala      |   1 -
 .../api/java/table/test/AggregationsITCase.java |  72 +++++------
 .../flink/api/java/table/test/AsITCase.java     |  62 +++++-----
 .../api/java/table/test/CastingITCase.java      |  53 ++++----
 .../api/java/table/test/ExpressionsITCase.java  |  32 ++---
 .../flink/api/java/table/test/FilterITCase.java |  69 +++++------
 .../table/test/GroupedAggregationsITCase.java   |  40 +++----
 .../flink/api/java/table/test/JoinITCase.java   |  70 ++++++-----
 .../api/java/table/test/PojoGroupingITCase.java |  19 +--
 .../flink/api/java/table/test/SelectITCase.java |  68 +++++------
 .../table/test/StringExpressionsITCase.java     |  40 ++++---
 .../flink/api/java/table/test/UnionITCase.java  |  47 ++++----
 .../scala/table/test/AggregationsITCase.scala   |  59 +++++----
 .../flink/api/scala/table/test/AsITCase.scala   |  50 ++++----
 .../api/scala/table/test/CastingITCase.scala    |  44 +++----
 .../scala/table/test/ExpressionsITCase.scala    |  40 +++----
 .../table/test/GroupedAggregationsITCase.scala  |  31 ++---
 .../flink/api/scala/table/test/JoinITCase.scala |  54 ++++-----
 .../api/scala/table/test/SelectITCase.scala     |  72 +++++------
 .../table/test/StringExpressionsITCase.scala    |  28 ++---
 .../api/scala/table/test/UnionITCase.scala      |  36 +++---
 71 files changed, 2596 insertions(+), 1176 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/package-info.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/package-info.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/package-info.java
deleted file mode 100644
index d7fbc8e..0000000
--- a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/package-info.java
+++ /dev/null
@@ -1,33 +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.
- */
-
-/**
- * <strong>Table API</strong><br>
- *
- * This package contains the generic part of the Table API. It can be used with Flink Streaming
- * and Flink Batch. From Scala as well as from Java.
- *
- * When using the Table API, as user creates a [[org.apache.flink.api.table.Table]] from
- * a DataSet or DataStream. On this relational operations can be performed. A table can also
- * be converted back to a DataSet or DataStream.
- *
- * Packages [[org.apache.flink.api.scala.table]] and [[org.apache.flink.api.java.table]] contain
- * the language specific part of the API. Refer to these packages for documentation on how
- * the Table API can be used in Java and Scala.
- */
-package org.apache.flink.api.table;

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/DataSetRelNode.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/DataSetRelNode.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/DataSetRelNode.java
deleted file mode 100644
index df0ebc0..0000000
--- a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/DataSetRelNode.java
+++ /dev/null
@@ -1,29 +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.sql.calcite;
-
-import org.apache.calcite.rel.RelNode;
-import org.apache.flink.api.java.DataSet;
-
-public interface DataSetRelNode<T> extends RelNode {
-	
-	/**
-	 * Translate the FlinkRelNode into Flink operator.
-	 */
-	DataSet<T> translateToPlan();
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetExchange.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetExchange.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetExchange.java
deleted file mode 100644
index 1ddd884..0000000
--- a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetExchange.java
+++ /dev/null
@@ -1,60 +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.sql.calcite.node;
-
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.SingleRel;
-import org.apache.flink.api.common.operators.base.PartitionOperatorBase.PartitionMethod;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.table.sql.calcite.DataSetRelNode;
-
-/**
- * Flink RelNode which matches along with PartitionOperator.
- *
- * @param <T>
- */
-public class DataSetExchange<T> extends SingleRel implements DataSetRelNode<T> {
-	
-	public DataSetExchange(RelOptCluster cluster, RelTraitSet traits, RelNode input) {
-		super(cluster, traits, input);
-	}
-	
-	private TypeInformation<T> getType() {
-		return null;
-	}
-	
-	private String getName() {
-		return null;
-	}
-	
-	private PartitionMethod getPartitionMethod() {
-		return null;
-	}
-	
-	private int[] getPartitionKey() {
-		return null;
-	}
-	
-	@Override
-	public DataSet<T> translateToPlan() {
-		return null;
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetFlatMap.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetFlatMap.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetFlatMap.java
deleted file mode 100644
index cf597b2..0000000
--- a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetFlatMap.java
+++ /dev/null
@@ -1,56 +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.sql.calcite.node;
-
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.SingleRel;
-import org.apache.flink.api.common.functions.RichFlatMapFunction;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.table.sql.calcite.DataSetRelNode;
-
-/**
- * Flink RelNode which matches along with FlatMapOperator.
- *
- * @param <T>
- */
-public class DataSetFlatMap<T> extends SingleRel implements DataSetRelNode<T> {
-	
-	protected DataSetFlatMap(RelOptCluster cluster, RelTraitSet traits, RelNode input) {
-		super(cluster, traits, input);
-	}
-	
-	private TypeInformation<T> getType() {
-		return null;
-	}
-	
-	private String getName() {
-		return null;
-	}
-	
-	private RichFlatMapFunction<T, T> getFlatMapFunction() {
-		return null;
-	}
-	
-	@Override
-	public DataSet<T> translateToPlan() {
-		return null;
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetJoin.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetJoin.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetJoin.java
deleted file mode 100644
index 3d1146e..0000000
--- a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetJoin.java
+++ /dev/null
@@ -1,80 +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.sql.calcite.node;
-
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.BiRel;
-import org.apache.calcite.rel.RelNode;
-import org.apache.flink.api.common.functions.JoinFunction;
-import org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.operators.join.JoinType;
-import org.apache.flink.api.table.sql.calcite.DataSetRelNode;
-
-/**
- * Flink RelNode which matches along with JoinOperator and its related operations.
- */
-public class DataSetJoin<L, R, OUT> extends BiRel implements DataSetRelNode<OUT> {
-	
-	public DataSetJoin(RelOptCluster cluster, RelTraitSet traitSet, RelNode left, RelNode right) {
-		super(cluster, traitSet, left, right);
-	}
-	
-	private TypeInformation<L> getLeftInputType() {
-		return null;
-	}
-	
-	private TypeInformation<R> getRightInputType() {
-		return null;
-	}
-	
-	private TypeInformation<OUT> getType() {
-		return null;
-	}
-	
-	private String getName() {
-		return null;
-	}
-	
-	private JoinType getJoinType() {
-		return null;
-	}
-	
-	private JoinHint getJoinHint() {
-		return null;
-	}
-	
-	private int[] getLeftJoinKey() {
-		return null;
-	}
-	
-	private int[] getRightJoinKey() {
-		return null;
-	}
-	
-	private JoinFunction<L, R, OUT> getJoinFunction() {
-		return null;
-	}
-	
-	@Override
-	public DataSet<OUT> translateToPlan() {
-		return null;
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetMap.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetMap.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetMap.java
deleted file mode 100644
index 3098cb3..0000000
--- a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetMap.java
+++ /dev/null
@@ -1,58 +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.sql.calcite.node;
-
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.SingleRel;
-import org.apache.flink.api.common.functions.RichMapFunction;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.table.sql.calcite.DataSetRelNode;
-
-/**
- * Flink RelNode which matches along with MapOperator.
- */
-public class DataSetMap<IN, OUT> extends SingleRel implements DataSetRelNode<OUT> {
-	
-	protected DataSetMap(RelOptCluster cluster, RelTraitSet traits, RelNode input) {
-		super(cluster, traits, input);
-	}
-	
-	private TypeInformation<IN> getInputType() {
-		return null;
-	}
-	
-	private TypeInformation<OUT> getType() {
-		return null;
-	}
-	
-	private String getName() {
-		return null;
-	}
-	
-	private RichMapFunction<IN, OUT> getMapFunction() {
-		return null;
-	}
-	
-	@Override
-	public DataSet<OUT> translateToPlan() {
-		return null;
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetReduce.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetReduce.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetReduce.java
deleted file mode 100644
index 4aa2846..0000000
--- a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetReduce.java
+++ /dev/null
@@ -1,58 +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.sql.calcite.node;
-
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.SingleRel;
-import org.apache.flink.api.common.functions.RichReduceFunction;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.table.sql.calcite.DataSetRelNode;
-
-/**
- * Flink RelNode which matches along with ReduceOperator.
- */
-public class DataSetReduce<T> extends SingleRel implements DataSetRelNode<T> {
-	
-	public DataSetReduce(RelOptCluster cluster, RelTraitSet traits, RelNode input) {
-		super(cluster, traits, input);
-	}
-	
-	private TypeInformation<T> getType() {
-		return null;
-	}
-	
-	private String getName() {
-		return null;
-	}
-	
-	private RichReduceFunction<T> getReduceFunction() {
-		return null;
-	}
-	
-	private int[] getGroupingKeys() {
-		return null;
-	}
-	
-	@Override
-	public DataSet<T> translateToPlan() {
-		return null;
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetReduceGroup.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetReduceGroup.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetReduceGroup.java
deleted file mode 100644
index 5467c56..0000000
--- a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetReduceGroup.java
+++ /dev/null
@@ -1,62 +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.sql.calcite.node;
-
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.SingleRel;
-import org.apache.flink.api.common.functions.RichGroupReduceFunction;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.table.sql.calcite.DataSetRelNode;
-
-/**
- * Flink RelNode which matches along with ReduceGroupOperator.
- */
-public class DataSetReduceGroup<IN, OUT> extends SingleRel implements DataSetRelNode<OUT> {
-	
-	public DataSetReduceGroup(RelOptCluster cluster, RelTraitSet traits, RelNode input) {
-		super(cluster, traits, input);
-	}
-	
-	private TypeInformation<IN> getInputType() {
-		return null;
-	}
-	
-	private TypeInformation<OUT> getType() {
-		return null;
-	}
-	
-	private String getName() {
-		return null;
-	}
-	
-	private RichGroupReduceFunction<IN, OUT> getGroupReduceFunction() {
-		return null;
-	}
-	
-	private int[] getGroupingKeys() {
-		return null;
-	}
-	
-	@Override
-	public DataSet<OUT> translateToPlan() {
-		return null;
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetSort.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetSort.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetSort.java
deleted file mode 100644
index 7fa1c53..0000000
--- a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetSort.java
+++ /dev/null
@@ -1,59 +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.sql.calcite.node;
-
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.SingleRel;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.table.sql.calcite.DataSetRelNode;
-
-/**
- * Flink RelNode which matches along with SortPartitionOperator.
- *
- * @param <T>
- */
-public class DataSetSort<T> extends SingleRel implements DataSetRelNode<T> {
-	
-	public DataSetSort(RelOptCluster cluster, RelTraitSet traits, RelNode input) {
-		super(cluster, traits, input);
-	}
-	
-	private TypeInformation<T> getType() {
-		return null;
-	}
-	
-	private String getName() {
-		return null;
-	}
-	
-	private int[] getSortKey() {
-		return null;
-	}
-	
-	private boolean[] getSortKeyOrder() {
-		return null;
-	}
-	
-	@Override
-	public DataSet<T> translateToPlan() {
-		return null;
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetSource.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetSource.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetSource.java
deleted file mode 100644
index f24ee79..0000000
--- a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetSource.java
+++ /dev/null
@@ -1,55 +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.sql.calcite.node;
-
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelOptTable;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.core.TableScan;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.table.sql.calcite.DataSetRelNode;
-
-/**
- * Flink RelNode which matches along with DataSource.
- *
- * @param <T>
- */
-public class DataSetSource<T> extends TableScan implements DataSetRelNode<T> {
-	
-	public DataSetSource(RelOptCluster cluster, RelTraitSet traitSet, RelOptTable table) {
-		super(cluster, traitSet, table);
-	}
-	
-	private TypeInformation<T> getType() {
-		return null;
-	}
-	
-	private String getName() {
-		return null;
-	}
-	
-	private DataSet<T> getDatSource() {
-		return null;
-	}
-	
-	@Override
-	public DataSet<T> translateToPlan() {
-		return null;
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetUnion.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetUnion.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetUnion.java
deleted file mode 100644
index 8b435e9..0000000
--- a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetUnion.java
+++ /dev/null
@@ -1,51 +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.sql.calcite.node;
-
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.BiRel;
-import org.apache.calcite.rel.RelNode;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.table.sql.calcite.DataSetRelNode;
-
-/**
- * Flink RelNode which matches along with UnionOperator.
- *
- * @param <T>
- */
-public class DataSetUnion<T> extends BiRel implements DataSetRelNode<T> {
-	
-	public DataSetUnion(RelOptCluster cluster, RelTraitSet traitSet, RelNode left, RelNode right) {
-		super(cluster, traitSet, left, right);
-	}
-	
-	private TypeInformation<T> getType() {
-		return null;
-	}
-	
-	private String getName() {
-		return null;
-	}
-	
-	@Override
-	public DataSet<T> translateToPlan() {
-		return null;
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala
index f3f4e9d..66bfbe7 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala
@@ -18,14 +18,17 @@
 
 package org.apache.flink.api.java.table
 
-import org.apache.calcite.plan.RelOptUtil
+import org.apache.calcite.plan.{RelTraitSet, RelOptUtil}
 import org.apache.calcite.rel.RelNode
-import org.apache.calcite.tools.{RelBuilder, Frameworks}
+import org.apache.calcite.sql2rel.RelDecorrelator
+import org.apache.calcite.tools.Programs
 import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.java.{DataSet => JavaDataSet}
 import org.apache.flink.api.table.plan._
-import org.apache.flink.api.table.plan.operators.DataSetTable
 import org.apache.flink.api.table.Table
+import org.apache.flink.api.table.plan.nodes.dataset.DataSetRel
+import org.apache.flink.api.table.plan.rules.FlinkRuleSets
+import org.apache.flink.api.table.plan.schema.DataSetTable
 
 /**
  * [[PlanTranslator]] for creating [[Table]]s from Java [[org.apache.flink.api.java.DataSet]]s and
@@ -41,37 +44,58 @@ class JavaBatchTranslator extends PlanTranslator {
 
     // create table representation from DataSet
     val dataSetTable = new DataSetTable[A](
-    repr.asInstanceOf[JavaDataSet[A]],
-    fieldNames
+      repr.asInstanceOf[JavaDataSet[A]],
+      fieldNames
     )
 
-    // register table in Cascading schema
-    val schema = Frameworks.createRootSchema(true)
-    val tableName = repr.hashCode().toString
-    schema.add(tableName, dataSetTable)
-
-    // initialize RelBuilder
-    val frameworkConfig = Frameworks
-      .newConfigBuilder
-      .defaultSchema(schema)
-      .build
-    val relBuilder = RelBuilder.create(frameworkConfig)
+    val tabName = TranslationContext.addDataSet(dataSetTable)
+    val relBuilder = TranslationContext.getRelBuilder
 
     // create table scan operator
-    relBuilder.scan(tableName)
+    relBuilder.scan(tabName)
     new Table(relBuilder.build(), relBuilder)
   }
 
   override def translate[A](lPlan: RelNode)(implicit tpe: TypeInformation[A]): JavaDataSet[A] = {
 
+    // get the planner for the plan
+    val planner = lPlan.getCluster.getPlanner
+
+    // we do not have any special requirements for the output
+    val outputProps = RelTraitSet.createEmpty()
+
+    println("-----------")
+    println("Input Plan:")
+    println("-----------")
     println(RelOptUtil.toString(lPlan))
 
-    // TODO: optimize & translate:
-    // - optimize RelNode plan
-    // - translate to Flink RelNode plan
-    // - generate DataSet program
+    // decorrelate
+    val decorPlan = RelDecorrelator.decorrelateQuery(lPlan)
+
+    // optimize the logical Flink plan
+    val optProgram = Programs.ofRules(FlinkRuleSets.DATASET_OPT_RULES)
+    val optPlan = optProgram.run(planner, decorPlan, outputProps)
+
+    println("---------------")
+    println("Optimized Plan:")
+    println("---------------")
+    println(RelOptUtil.toString(optPlan))
+
+    // optimize the logical Flink plan
+    val dataSetProgram = Programs.ofRules(FlinkRuleSets.DATASET_TRANS_RULES)
+    val dataSetPlan = dataSetProgram.run(planner, optPlan, outputProps)
+
+    println("-------------")
+    println("DataSet Plan:")
+    println("-------------")
+    println(RelOptUtil.toString(dataSetPlan))
+
+    dataSetPlan match {
+      case node: DataSetRel =>
+        node.translateToPlan.asInstanceOf[JavaDataSet[A]]
+      case _ => ???
+    }
 
-    null
   }
 
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/TranslationContext.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/TranslationContext.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/TranslationContext.scala
new file mode 100644
index 0000000..b2b0c2b
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/TranslationContext.scala
@@ -0,0 +1,79 @@
+/*
+ * 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
+
+import java.util.concurrent.atomic.AtomicInteger
+
+import org.apache.calcite.plan.ConventionTraitDef
+import org.apache.calcite.schema.impl.AbstractTable
+import org.apache.calcite.schema.SchemaPlus
+import org.apache.calcite.tools.{Frameworks, RelBuilder}
+import org.apache.flink.api.table.plan.schema.DataSetTable
+
+object TranslationContext {
+
+  private var relBuilder: RelBuilder = null
+  private var tables: SchemaPlus = null
+  private var tabNames: Map[AbstractTable, String] = null
+  private val nameCntr: AtomicInteger = new AtomicInteger(0)
+
+  reset()
+
+  def reset(): Unit = {
+
+    // register table in Cascading schema
+    tables = Frameworks.createRootSchema(true)
+
+    // initialize RelBuilder
+    val frameworkConfig = Frameworks
+      .newConfigBuilder
+      .defaultSchema(tables)
+      .traitDefs(ConventionTraitDef.INSTANCE)
+      .build
+
+    tabNames = Map[AbstractTable, String]()
+
+    relBuilder = RelBuilder.create(frameworkConfig)
+
+  }
+
+  def addDataSet(newTable: DataSetTable[_]): String = {
+
+    // look up name
+    val tabName = tabNames.get(newTable)
+
+    tabName match {
+      case Some(name) =>
+        name
+      case None =>
+        val tabName = "DataSetTable_" + nameCntr.getAndIncrement()
+        tabNames += (newTable -> tabName)
+        tables.add(tabName, newTable)
+        tabName
+    }
+
+  }
+
+  def getRelBuilder: RelBuilder = {
+    relBuilder
+  }
+
+}
+
+

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetConvention.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetConvention.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetConvention.scala
new file mode 100644
index 0000000..cbacd16
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetConvention.scala
@@ -0,0 +1,42 @@
+/*
+ * 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.nodes.dataset
+
+import org.apache.calcite.plan._
+
+class DataSetConvention extends Convention {
+
+  override def toString: String = getName
+
+  def getInterface: Class[_] = classOf[DataSetRel]
+
+  def getName: String = "DATASET"
+
+  def getTraitDef: RelTraitDef[_ <: RelTrait] = ConventionTraitDef.INSTANCE
+
+  def satisfies(`trait`: RelTrait): Boolean = this eq `trait`
+
+  def register(planner: RelOptPlanner): Unit = { }
+
+}
+
+object DataSetConvention {
+
+  val INSTANCE = new DataSetConvention
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetExchange.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetExchange.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetExchange.scala
new file mode 100644
index 0000000..ec5805a
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetExchange.scala
@@ -0,0 +1,63 @@
+/*
+ * 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.nodes.dataset
+
+import org.apache.calcite.plan.{RelTraitSet, RelOptCluster}
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.{RelWriter, RelNode, SingleRel}
+import org.apache.flink.api.common.operators.base.PartitionOperatorBase.PartitionMethod
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.Row
+
+/**
+  * Flink RelNode which matches along with PartitionOperator.
+  */
+class DataSetExchange(
+    cluster: RelOptCluster,
+    traitSet: RelTraitSet,
+    input: RelNode,
+    rowType: RelDataType,
+    opName: String,
+    partitionKey: Array[Int],
+    partitionMethod: PartitionMethod)
+  extends SingleRel(cluster, traitSet, input)
+  with DataSetRel {
+
+  override def deriveRowType() = rowType
+
+  override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
+    new DataSetExchange(
+      cluster,
+      traitSet,
+      inputs.get(0),
+      rowType,
+      opName,
+      partitionKey,
+      partitionMethod
+    )
+  }
+
+  override def explainTerms(pw: RelWriter): RelWriter = {
+    super.explainTerms(pw).item("name", opName)
+  }
+
+  override def translateToPlan: DataSet[Any] = {
+    ???
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetFlatMap.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetFlatMap.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetFlatMap.scala
new file mode 100644
index 0000000..913cca0
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetFlatMap.scala
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.table.plan.nodes.dataset
+
+import org.apache.calcite.plan.{RelOptCost, RelOptPlanner, RelTraitSet, RelOptCluster}
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.{RelWriter, RelNode, SingleRel}
+import org.apache.flink.api.common.functions.FlatMapFunction
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.Row
+
+/**
+  * Flink RelNode which matches along with FlatMapOperator.
+  *
+  */
+class DataSetFlatMap(
+    cluster: RelOptCluster,
+    traitSet: RelTraitSet,
+    input: RelNode,
+    rowType: RelDataType,
+    opName: String,
+    func: FlatMapFunction[Row, Row])
+  extends SingleRel(cluster, traitSet, input)
+  with DataSetRel {
+
+  override def deriveRowType() = rowType
+
+  override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
+    new DataSetFlatMap(
+      cluster,
+      traitSet,
+      inputs.get(0),
+      rowType,
+      opName,
+      func
+    )
+  }
+
+  override def explainTerms(pw: RelWriter): RelWriter = {
+    super.explainTerms(pw).item("name", opName)
+  }
+
+  override def translateToPlan: DataSet[Any] = {
+    ???
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetGroupReduce.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetGroupReduce.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetGroupReduce.scala
new file mode 100644
index 0000000..11bb160
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetGroupReduce.scala
@@ -0,0 +1,63 @@
+/*
+ * 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.nodes.dataset
+
+import org.apache.calcite.plan.{RelTraitSet, RelOptCluster}
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.{RelWriter, RelNode, SingleRel}
+import org.apache.flink.api.common.functions.GroupReduceFunction
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.Row
+
+/**
+  * Flink RelNode which matches along with ReduceGroupOperator.
+  */
+class DataSetGroupReduce(
+    cluster: RelOptCluster,
+    traitSet: RelTraitSet,
+    input: RelNode,
+    rowType: RelDataType,
+    opName: String,
+    groupingKeys: Array[Int],
+    func: GroupReduceFunction[Row, Row])
+  extends SingleRel(cluster, traitSet, input)
+  with DataSetRel {
+
+  override def deriveRowType() = rowType
+
+  override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
+    new DataSetGroupReduce(
+      cluster,
+      traitSet,
+      inputs.get(0),
+      rowType,
+      opName,
+      groupingKeys,
+      func
+    )
+  }
+
+  override def explainTerms(pw: RelWriter): RelWriter = {
+    super.explainTerms(pw).item("name", opName)
+  }
+
+  override def translateToPlan: DataSet[Any] = {
+    ???
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetJoin.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetJoin.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetJoin.scala
new file mode 100644
index 0000000..c20cdc5
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetJoin.scala
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.table.plan.nodes.dataset
+
+import org.apache.calcite.plan.{RelTraitSet, RelOptCluster}
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.{RelWriter, BiRel, RelNode}
+import org.apache.flink.api.common.functions.JoinFunction
+import org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.java.operators.join.JoinType
+import org.apache.flink.api.table.Row
+
+/**
+  * Flink RelNode which matches along with JoinOperator and its related operations.
+  */
+class DataSetJoin(
+    cluster: RelOptCluster,
+    traitSet: RelTraitSet,
+    left: RelNode,
+    right: RelNode,
+    rowType: RelDataType,
+    opName: String,
+    joinKeysLeft: Array[Int],
+    joinKeysRight: Array[Int],
+    joinType: JoinType,
+    joinHint: JoinHint,
+    func: JoinFunction[Row, Row, Row])
+  extends BiRel(cluster, traitSet, left, right)
+  with DataSetRel {
+
+  override def deriveRowType() = rowType
+
+  override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
+    new DataSetJoin(
+      cluster,
+      traitSet,
+      inputs.get(0),
+      inputs.get(1),
+      rowType,
+      opName,
+      joinKeysLeft,
+      joinKeysRight,
+      joinType,
+      joinHint,
+      func
+    )
+  }
+
+  override def explainTerms(pw: RelWriter): RelWriter = {
+    super.explainTerms(pw).item("name", opName)
+  }
+
+  override def translateToPlan: DataSet[Any] = {
+    ???
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetMap.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetMap.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetMap.scala
new file mode 100644
index 0000000..be8bd9d
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetMap.scala
@@ -0,0 +1,63 @@
+/*
+ * 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.nodes.dataset
+
+import org.apache.calcite.plan.{RelTraitSet, RelOptCluster}
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.{RelWriter, RelNode, SingleRel}
+import org.apache.flink.api.common.functions.MapFunction
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.Row
+
+/**
+  * Flink RelNode which matches along with MapOperator.
+  */
+class DataSetMap(
+    cluster: RelOptCluster,
+    traitSet: RelTraitSet,
+    input: RelNode,
+    rowType: RelDataType,
+    opName: String,
+    func: MapFunction[Row, Row])
+  extends SingleRel(cluster, traitSet, input)
+  with DataSetRel {
+
+  override def deriveRowType() = rowType
+
+  override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
+    new DataSetMap(
+      cluster,
+      traitSet,
+      inputs.get(0),
+      rowType,
+      opName,
+      func
+    )
+  }
+
+  override def explainTerms(pw: RelWriter): RelWriter = {
+    super.explainTerms(pw).item("name", opName)
+  }
+
+  override def toString() = opName
+
+  override def translateToPlan: DataSet[Any] = {
+    ???
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetReduce.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetReduce.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetReduce.scala
new file mode 100644
index 0000000..567a91c
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetReduce.scala
@@ -0,0 +1,63 @@
+/*
+ * 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.nodes.dataset
+
+import org.apache.calcite.plan.{RelTraitSet, RelOptCluster}
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.{RelWriter, RelNode, SingleRel}
+import org.apache.flink.api.common.functions.ReduceFunction
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.Row
+
+/**
+  * Flink RelNode which matches along with ReduceOperator.
+  */
+class DataSetReduce(
+    cluster: RelOptCluster,
+    traits: RelTraitSet,
+    input: RelNode,
+    rowType: RelDataType,
+    opName: String,
+    groupingKeys: Array[Int],
+    func: ReduceFunction[Row])
+  extends SingleRel(cluster, traits, input)
+  with DataSetRel {
+
+  override def deriveRowType() = rowType
+
+  override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
+    new DataSetReduce(
+      cluster,
+      traitSet,
+      inputs.get(0),
+      rowType,
+      opName,
+      groupingKeys,
+      func
+    )
+  }
+
+  override def explainTerms(pw: RelWriter): RelWriter = {
+    super.explainTerms(pw).item("name", opName)
+  }
+
+  override def translateToPlan: DataSet[Any] = {
+    ???
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetRel.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetRel.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetRel.scala
new file mode 100644
index 0000000..20677b3
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetRel.scala
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.table.plan.nodes.dataset
+
+import org.apache.calcite.rel.RelNode
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.Row
+
+trait DataSetRel extends RelNode {
+
+  /**
+    * Translate the FlinkRelNode into Flink operator.
+    */
+  def translateToPlan: DataSet[Any]
+
+}
+

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSort.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSort.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSort.scala
new file mode 100644
index 0000000..df5301d
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSort.scala
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.table.plan.nodes.dataset
+
+import org.apache.calcite.plan.{RelTraitSet, RelOptCluster}
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.{RelWriter, RelNode, SingleRel}
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.Row
+
+/**
+  * Flink RelNode which matches along with SortPartitionOperator.
+  */
+class DataSetSort(
+    cluster: RelOptCluster,
+    traitSet: RelTraitSet,
+    input: RelNode,
+    rowType: RelDataType,
+    opName: String,
+    sortKey: Array[Int],
+    sortOrder: Array[Boolean])
+  extends SingleRel(cluster, traitSet, input)
+  with DataSetRel {
+
+  override def deriveRowType() = rowType
+
+  override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
+    new DataSetSort(
+      cluster,
+      traitSet,
+      inputs.get(0),
+      rowType,
+      opName,
+      sortKey,
+      sortOrder
+    )
+  }
+
+  override def explainTerms(pw: RelWriter): RelWriter = {
+    super.explainTerms(pw).item("name", opName)
+  }
+
+  override def translateToPlan: DataSet[Any] = {
+    ???
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSource.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSource.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSource.scala
new file mode 100644
index 0000000..effaf1a
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSource.scala
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.table.plan.nodes.dataset
+
+import org.apache.calcite.plan._
+import org.apache.calcite.rel.{RelWriter, RelNode}
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.core.TableScan
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.Row
+
+/**
+  * Flink RelNode which matches along with DataSource.
+  */
+class DataSetSource(
+    cluster: RelOptCluster,
+    traitSet: RelTraitSet,
+    table: RelOptTable,
+    rowType: RelDataType,
+    inputDataSet: DataSet[_])
+  extends TableScan(cluster, traitSet, table)
+  with DataSetRel {
+
+  override def deriveRowType() = rowType
+
+  override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
+    new DataSetSource(
+      cluster,
+      traitSet,
+      table,
+      rowType,
+      inputDataSet
+    )
+  }
+
+  override def translateToPlan: DataSet[Any] = {
+    ???
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetUnion.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetUnion.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetUnion.scala
new file mode 100644
index 0000000..a510fc9
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetUnion.scala
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.table.plan.nodes.dataset
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.{RelWriter, BiRel, RelNode}
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.Row
+
+/**
+* Flink RelNode which matches along with UnionOperator.
+*
+*/
+class DataSetUnion(
+    cluster: RelOptCluster,
+    traitSet: RelTraitSet,
+    left: RelNode,
+    right: RelNode,
+    rowType: RelDataType,
+    opName: String)
+  extends BiRel(cluster, traitSet, left, right)
+  with DataSetRel {
+
+  override def deriveRowType() = rowType
+
+  override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
+    new DataSetUnion(
+      cluster,
+      traitSet,
+      inputs.get(0),
+      inputs.get(1),
+      rowType,
+      opName
+    )
+  }
+
+  override def explainTerms(pw: RelWriter): RelWriter = {
+    super.explainTerms(pw).item("name", opName)
+  }
+
+  override def translateToPlan: DataSet[Any] = {
+    ???
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkAggregate.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkAggregate.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkAggregate.scala
new file mode 100644
index 0000000..f66cb71
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkAggregate.scala
@@ -0,0 +1,76 @@
+/*
+ * 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.nodes.logical
+
+import java.util
+
+import org.apache.calcite.plan.{RelOptCost, RelOptPlanner, RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.{AggregateCall, Aggregate}
+import org.apache.calcite.sql.fun.SqlAvgAggFunction
+import org.apache.calcite.util.ImmutableBitSet
+
+import scala.collection.JavaConversions._
+
+class FlinkAggregate(
+    cluster: RelOptCluster,
+    traitSet: RelTraitSet,
+    input: RelNode,
+    indicator: Boolean,
+    groupSet: ImmutableBitSet,
+    groupSets: java.util.List[ImmutableBitSet],
+    aggCalls: java.util.List[AggregateCall])
+  extends Aggregate(cluster, traitSet, input, indicator, groupSet, groupSets, aggCalls)
+  with FlinkRel {
+
+  override def copy(
+      traitSet: RelTraitSet,
+      input: RelNode,
+      indicator: Boolean,
+      groupSet: ImmutableBitSet,
+      groupSets: util.List[ImmutableBitSet],
+      aggCalls: util.List[AggregateCall]): Aggregate = {
+
+    new FlinkAggregate(
+      cluster,
+      traitSet,
+      input,
+      indicator,
+      groupSet,
+      groupSets,
+      aggCalls
+    )
+  }
+
+  override def computeSelfCost (planner: RelOptPlanner): RelOptCost = {
+
+    val origCosts = super.computeSelfCost(planner)
+    val deltaCost = planner.getCostFactory.makeHugeCost()
+
+    // only prefer aggregations with transformed Avg
+    aggCalls.toList.foldLeft[RelOptCost](origCosts){
+      (c: RelOptCost, a: AggregateCall) =>
+        if (a.getAggregation.isInstanceOf[SqlAvgAggFunction]) {
+          c.plus(deltaCost)
+        } else {
+          c
+        }
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkCalc.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkCalc.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkCalc.scala
new file mode 100644
index 0000000..bcfe8d7
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkCalc.scala
@@ -0,0 +1,37 @@
+/*
+ * 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.nodes.logical
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.Calc
+import org.apache.calcite.rex.RexProgram
+
+class FlinkCalc(
+    cluster: RelOptCluster,
+    traitSet: RelTraitSet,
+    input: RelNode,
+    program: RexProgram)
+  extends Calc(cluster, traitSet, input, program)
+  with FlinkRel {
+
+  override def copy(traitSet: RelTraitSet, child: RelNode, program: RexProgram): Calc = {
+    new FlinkCalc(cluster, traitSet, child, program)
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkConvention.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkConvention.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkConvention.scala
new file mode 100644
index 0000000..80137f2
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkConvention.scala
@@ -0,0 +1,42 @@
+/*
+ * 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.nodes.logical
+
+import org.apache.calcite.plan._
+
+class FlinkConvention extends Convention {
+
+  override def toString: String = getName
+
+  def getInterface: Class[_] = classOf[FlinkRel]
+
+  def getName: String = "FLINK"
+
+  def getTraitDef: RelTraitDef[_ <: RelTrait] = ConventionTraitDef.INSTANCE
+
+  def satisfies(`trait`: RelTrait): Boolean = this eq `trait`
+
+  def register(planner: RelOptPlanner): Unit = { }
+
+}
+
+object FlinkConvention {
+
+  val INSTANCE = new FlinkConvention
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkFilter.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkFilter.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkFilter.scala
new file mode 100644
index 0000000..9f0bf30
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkFilter.scala
@@ -0,0 +1,42 @@
+/*
+ * 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.nodes.logical
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.Filter
+import org.apache.calcite.rex.RexNode
+
+class FlinkFilter(
+    cluster: RelOptCluster,
+    traitSet: RelTraitSet,
+    input: RelNode,
+    condition: RexNode)
+  extends Filter(cluster, traitSet, input, condition)
+  with FlinkRel {
+
+  override def copy(traitSet: RelTraitSet, input: RelNode, condition: RexNode): Filter = {
+    new FlinkFilter(
+      cluster,
+      traitSet,
+      input,
+      condition
+    )
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkJoin.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkJoin.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkJoin.scala
new file mode 100644
index 0000000..8b04b50
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkJoin.scala
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.table.plan.nodes.logical
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.{JoinRelType, Join}
+import org.apache.calcite.rex.RexNode
+
+class FlinkJoin(
+    cluster: RelOptCluster,
+    traitSet: RelTraitSet,
+    left: RelNode,
+    right: RelNode,
+    condition: RexNode,
+    joinType: JoinRelType,
+    variablesStopped: java.util.Set[String])
+  extends Join(cluster, traitSet, left, right, condition, joinType, variablesStopped)
+  with FlinkRel {
+
+  override def copy(
+      traitSet: RelTraitSet,
+      condition: RexNode,
+      left: RelNode,
+      right: RelNode,
+      joinType: JoinRelType,
+      semiJoinDone: Boolean): Join = {
+    new FlinkJoin(cluster, traitSet, left, right, condition, joinType, getVariablesStopped)
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkProject.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkProject.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkProject.scala
new file mode 100644
index 0000000..1d93036
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkProject.scala
@@ -0,0 +1,45 @@
+/*
+ * 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.nodes.logical
+
+import java.util
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.core.Project
+import org.apache.calcite.rex.RexNode
+
+class FlinkProject(
+    cluster: RelOptCluster,
+    traitSet: RelTraitSet,
+    input: RelNode,
+    projects: java.util.List[RexNode],
+    rowType: RelDataType)
+  extends Project(cluster, traitSet, input, projects, rowType)
+  with FlinkRel {
+
+  override def copy(
+      traitSet: RelTraitSet,
+      input: RelNode,
+      projects: util.List[RexNode],
+      rowType: RelDataType): Project = {
+    new FlinkProject(cluster, traitSet, input, projects, rowType)
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkRel.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkRel.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkRel.scala
new file mode 100644
index 0000000..9ebd7e4
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkRel.scala
@@ -0,0 +1,25 @@
+/*
+ * 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.nodes.logical
+
+import org.apache.calcite.rel.RelNode
+
+trait FlinkRel extends RelNode {
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkScan.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkScan.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkScan.scala
new file mode 100644
index 0000000..6d53a75
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkScan.scala
@@ -0,0 +1,31 @@
+/*
+ * 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.nodes.logical
+
+import org.apache.calcite.plan.{RelOptTable, RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.core.TableScan
+
+class FlinkScan(
+    cluster: RelOptCluster,
+    traitSet: RelTraitSet,
+    table: RelOptTable)
+  extends TableScan(cluster, traitSet, table)
+  with FlinkRel {
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe5e4065/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkUnion.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkUnion.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkUnion.scala
new file mode 100644
index 0000000..fd791d3
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkUnion.scala
@@ -0,0 +1,38 @@
+/*
+ * 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.nodes.logical
+
+import java.util
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.{SetOp, Union}
+
+class FlinkUnion(
+    cluster: RelOptCluster,
+    traitSet: RelTraitSet,
+    inputs: java.util.List[RelNode],
+    all: Boolean)
+  extends Union(cluster, traitSet, inputs, all)
+  with FlinkRel {
+
+  override def copy(traitSet: RelTraitSet, inputs: util.List[RelNode], all: Boolean): SetOp = {
+    new FlinkUnion(cluster, traitSet, inputs, all)
+  }
+}


Mime
View raw message