[
https://issues.apache.org/jira/browse/FLINK-3192?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15081348#comment-15081348
]
ASF GitHub Bot commented on FLINK-3192:
---------------------------------------
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/1477#discussion_r48751597
--- Diff:
flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/Table.scala
---
@@ -1,271 +1,291 @@
-/*
- * 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
-
-import org.apache.flink.api.table.expressions.analysis.{GroupByAnalyzer,
PredicateAnalyzer, SelectionAnalyzer}
-import org.apache.flink.api.table.expressions.{Expression,
ResolvedFieldReference, UnresolvedFieldReference}
-import org.apache.flink.api.table.parser.ExpressionParser
-import org.apache.flink.api.table.plan._
-
-/**
- * The abstraction for writing Table API programs. Similar to how the
batch and streaming APIs
- * have [[org.apache.flink.api.scala.DataSet]] and
- * [[org.apache.flink.streaming.api.scala.DataStream]].
- *
- * Use the methods of [[Table]] to transform data. Use
- * [[org.apache.flink.api.java.table.TableEnvironment]] to convert a
[[Table]] back to a DataSet
- * or DataStream.
- *
- * When using Scala a [[Table]] can also be converted using implicit
conversions.
- *
- * Example:
- *
- * {{{
- * val table = set.toTable('a, 'b)
- * ...
- * val table2 = ...
- * val set = table2.toDataSet[MyType]
- * }}}
- *
- * Operations such as [[join]], [[select]], [[where]] and [[groupBy]]
either take arguments
- * in a Scala DSL or as an expression String. Please refer to the
documentation for the expression
- * syntax.
- */
-case class Table(private[flink] val operation: PlanNode) {
-
- /**
- * Performs a selection operation. Similar to an SQL SELECT statement.
The field expressions
- * can contain complex expressions and aggregations.
- *
- * Example:
- *
- * {{{
- * in.select('key, 'value.avg + " The average" as 'average,
'other.substring(0, 10))
- * }}}
- */
- def select(fields: Expression*): Table = {
- val analyzer = new SelectionAnalyzer(operation.outputFields)
- val analyzedFields = fields.map(analyzer.analyze)
- val fieldNames = analyzedFields map(_.name)
- if (fieldNames.toSet.size != fieldNames.size) {
- throw new ExpressionException(s"Resulting fields names are not
unique in expression" +
- s""" "${fields.mkString(", ")}".""")
- }
- this.copy(operation = Select(operation, analyzedFields))
- }
-
- /**
- * Performs a selection operation. Similar to an SQL SELECT statement.
The field expressions
- * can contain complex expressions and aggregations.
- *
- * Example:
- *
- * {{{
- * in.select("key, value.avg + " The average" as average,
other.substring(0, 10)")
- * }}}
- */
- def select(fields: String): Table = {
- val fieldExprs = ExpressionParser.parseExpressionList(fields)
- select(fieldExprs: _*)
- }
-
- /**
- * Renames the fields of the expression result. Use this to disambiguate
fields before
- * joining to operations.
- *
- * Example:
- *
- * {{{
- * in.as('a, 'b)
- * }}}
- */
- def as(fields: Expression*): Table = {
- fields forall {
- f => f.isInstanceOf[UnresolvedFieldReference]
- } match {
- case true =>
- case false => throw new ExpressionException("Only field expression
allowed in as().")
- }
- this.copy(operation = As(operation, fields.toArray map { _.name }))
- }
-
- /**
- * Renames the fields of the expression result. Use this to disambiguate
fields before
- * joining to operations.
- *
- * Example:
- *
- * {{{
- * in.as("a, b")
- * }}}
- */
- def as(fields: String): Table = {
- val fieldExprs = ExpressionParser.parseExpressionList(fields)
- as(fieldExprs: _*)
- }
-
- /**
- * Filters out elements that don't pass the filter predicate. Similar to
a SQL WHERE
- * clause.
- *
- * Example:
- *
- * {{{
- * in.filter('name === "Fred")
- * }}}
- */
- def filter(predicate: Expression): Table = {
- val analyzer = new PredicateAnalyzer(operation.outputFields)
- val analyzedPredicate = analyzer.analyze(predicate)
- this.copy(operation = Filter(operation, analyzedPredicate))
- }
-
- /**
- * Filters out elements that don't pass the filter predicate. Similar to
a SQL WHERE
- * clause.
- *
- * Example:
- *
- * {{{
- * in.filter("name = 'Fred'")
- * }}}
- */
- def filter(predicate: String): Table = {
- val predicateExpr = ExpressionParser.parseExpression(predicate)
- filter(predicateExpr)
- }
-
- /**
- * Filters out elements that don't pass the filter predicate. Similar to
a SQL WHERE
- * clause.
- *
- * Example:
- *
- * {{{
- * in.where('name === "Fred")
- * }}}
- */
- def where(predicate: Expression): Table = {
- filter(predicate)
- }
-
- /**
- * Filters out elements that don't pass the filter predicate. Similar to
a SQL WHERE
- * clause.
- *
- * Example:
- *
- * {{{
- * in.where("name = 'Fred'")
- * }}}
- */
- def where(predicate: String): Table = {
- filter(predicate)
- }
-
- /**
- * Groups the elements on some grouping keys. Use this before a
selection with aggregations
- * to perform the aggregation on a per-group basis. Similar to a SQL
GROUP BY statement.
- *
- * Example:
- *
- * {{{
- * in.groupBy('key).select('key, 'value.avg)
- * }}}
- */
- def groupBy(fields: Expression*): Table = {
- val analyzer = new GroupByAnalyzer(operation.outputFields)
- val analyzedFields = fields.map(analyzer.analyze)
-
- val illegalKeys = analyzedFields filter {
- case fe: ResolvedFieldReference => false // OK
- case e => true
- }
-
- if (illegalKeys.nonEmpty) {
- throw new ExpressionException("Illegal key expressions: " +
illegalKeys.mkString(", "))
- }
-
- this.copy(operation = GroupBy(operation, analyzedFields))
- }
-
- /**
- * Groups the elements on some grouping keys. Use this before a
selection with aggregations
- * to perform the aggregation on a per-group basis. Similar to a SQL
GROUP BY statement.
- *
- * Example:
- *
- * {{{
- * in.groupBy("key").select("key, value.avg")
- * }}}
- */
- def groupBy(fields: String): Table = {
- val fieldsExpr = ExpressionParser.parseExpressionList(fields)
- groupBy(fieldsExpr: _*)
- }
-
- /**
- * Joins two [[Table]]s. Similar to an SQL join. The fields of the two
joined
- * operations must not overlap, use [[as]] to rename fields if
necessary. You can use
- * where and select clauses after a join to further specify the
behaviour of the join.
- *
- * Example:
- *
- * {{{
- * left.join(right).where('a === 'b && 'c > 3).select('a, 'b, 'd)
- * }}}
- */
- def join(right: Table): Table = {
- val leftInputNames = operation.outputFields.map(_._1).toSet
- val rightInputNames = right.operation.outputFields.map(_._1).toSet
- if (leftInputNames.intersect(rightInputNames).nonEmpty) {
- throw new ExpressionException(
- "Overlapping fields names on join input, result would be
ambiguous: " +
- operation.outputFields.mkString(", ") +
- " and " +
- right.operation.outputFields.mkString(", ") )
- }
- this.copy(operation = Join(operation, right.operation))
- }
-
- /**
- * Union two[[Table]]s. Similar to an SQL UNION ALL. The fields of the
two union operations
- * must fully overlap.
- *
- * Example:
- *
- * {{{
- * left.unionAll(right)
- * }}}
- */
- def unionAll(right: Table): Table = {
- val leftInputFields = operation.outputFields
- val rightInputFields = right.operation.outputFields
- if (!leftInputFields.equals(rightInputFields)) {
- throw new ExpressionException(
- "The fields names of join inputs should be fully overlapped, left
inputs fields:" +
- operation.outputFields.mkString(", ") +
- " and right inputs fields" +
- right.operation.outputFields.mkString(", ")
- )
- }
- this.copy(operation = UnionAll(operation, right.operation))
- }
-
- override def toString: String = s"Expression($operation)"
-}
+/*
+ * 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
+
+import org.apache.flink.api.java.io.DiscardingOutputFormat
+import org.apache.flink.api.table.expressions.analysis.{GroupByAnalyzer,
PredicateAnalyzer, SelectionAnalyzer}
+import org.apache.flink.api.table.expressions.{Expression,
ResolvedFieldReference, UnresolvedFieldReference}
+import org.apache.flink.api.table.parser.ExpressionParser
+import org.apache.flink.api.table.plan._
+import org.apache.flink.api.scala._
+import org.apache.flink.api.scala.table._
+
+/**
+ * The abstraction for writing Table API programs. Similar to how the
batch and streaming APIs
+ * have [[org.apache.flink.api.scala.DataSet]] and
+ * [[org.apache.flink.streaming.api.scala.DataStream]].
+ *
+ * Use the methods of [[Table]] to transform data. Use
+ * [[org.apache.flink.api.java.table.TableEnvironment]] to convert a
[[Table]] back to a DataSet
+ * or DataStream.
+ *
+ * When using Scala a [[Table]] can also be converted using implicit
conversions.
+ *
+ * Example:
+ *
+ * {{{
+ * val table = set.toTable('a, 'b)
+ * ...
+ * val table2 = ...
+ * val set = table2.toDataSet[MyType]
+ * }}}
+ *
+ * Operations such as [[join]], [[select]], [[where]] and [[groupBy]]
either take arguments
+ * in a Scala DSL or as an expression String. Please refer to the
documentation for the expression
+ * syntax.
+ */
+case class Table(private[flink] val operation: PlanNode) {
+
+ /**
+ * Performs a selection operation. Similar to an SQL SELECT statement.
The field expressions
+ * can contain complex expressions and aggregations.
+ *
+ * Example:
+ *
+ * {{{
+ * in.select('key, 'value.avg + " The average" as 'average,
'other.substring(0, 10))
+ * }}}
+ */
+ def select(fields: Expression*): Table = {
+ val analyzer = new SelectionAnalyzer(operation.outputFields)
+ val analyzedFields = fields.map(analyzer.analyze)
+ val fieldNames = analyzedFields map(_.name)
+ if (fieldNames.toSet.size != fieldNames.size) {
+ throw new ExpressionException(s"Resulting fields names are not
unique in expression" +
+ s""" "${fields.mkString(", ")}".""")
+ }
+ this.copy(operation = Select(operation, analyzedFields))
+ }
+
+ /**
+ * Performs a selection operation. Similar to an SQL SELECT statement.
The field expressions
+ * can contain complex expressions and aggregations.
+ *
+ * Example:
+ *
+ * {{{
+ * in.select("key, value.avg + " The average" as average,
other.substring(0, 10)")
+ * }}}
+ */
+ def select(fields: String): Table = {
+ val fieldExprs = ExpressionParser.parseExpressionList(fields)
+ select(fieldExprs: _*)
+ }
+
+ /**
+ * Renames the fields of the expression result. Use this to disambiguate
fields before
+ * joining to operations.
+ *
+ * Example:
+ *
+ * {{{
+ * in.as('a, 'b)
+ * }}}
+ */
+ def as(fields: Expression*): Table = {
+ fields forall {
+ f => f.isInstanceOf[UnresolvedFieldReference]
+ } match {
+ case true =>
+ case false => throw new ExpressionException("Only field expression
allowed in as().")
+ }
+ this.copy(operation = As(operation, fields.toArray map { _.name }))
+ }
+
+ /**
+ * Renames the fields of the expression result. Use this to disambiguate
fields before
+ * joining to operations.
+ *
+ * Example:
+ *
+ * {{{
+ * in.as("a, b")
+ * }}}
+ */
+ def as(fields: String): Table = {
+ val fieldExprs = ExpressionParser.parseExpressionList(fields)
+ as(fieldExprs: _*)
+ }
+
+ /**
+ * Filters out elements that don't pass the filter predicate. Similar to
a SQL WHERE
+ * clause.
+ *
+ * Example:
+ *
+ * {{{
+ * in.filter('name === "Fred")
+ * }}}
+ */
+ def filter(predicate: Expression): Table = {
+ val analyzer = new PredicateAnalyzer(operation.outputFields)
+ val analyzedPredicate = analyzer.analyze(predicate)
+ this.copy(operation = Filter(operation, analyzedPredicate))
+ }
+
+ /**
+ * Filters out elements that don't pass the filter predicate. Similar to
a SQL WHERE
+ * clause.
+ *
+ * Example:
+ *
+ * {{{
+ * in.filter("name = 'Fred'")
+ * }}}
+ */
+ def filter(predicate: String): Table = {
+ val predicateExpr = ExpressionParser.parseExpression(predicate)
+ filter(predicateExpr)
+ }
+
+ /**
+ * Filters out elements that don't pass the filter predicate. Similar to
a SQL WHERE
+ * clause.
+ *
+ * Example:
+ *
+ * {{{
+ * in.where('name === "Fred")
+ * }}}
+ */
+ def where(predicate: Expression): Table = {
+ filter(predicate)
+ }
+
+ /**
+ * Filters out elements that don't pass the filter predicate. Similar to
a SQL WHERE
+ * clause.
+ *
+ * Example:
+ *
+ * {{{
+ * in.where("name = 'Fred'")
+ * }}}
+ */
+ def where(predicate: String): Table = {
+ filter(predicate)
+ }
+
+ /**
+ * Groups the elements on some grouping keys. Use this before a
selection with aggregations
+ * to perform the aggregation on a per-group basis. Similar to a SQL
GROUP BY statement.
+ *
+ * Example:
+ *
+ * {{{
+ * in.groupBy('key).select('key, 'value.avg)
+ * }}}
+ */
+ def groupBy(fields: Expression*): Table = {
+ val analyzer = new GroupByAnalyzer(operation.outputFields)
+ val analyzedFields = fields.map(analyzer.analyze)
+
+ val illegalKeys = analyzedFields filter {
+ case fe: ResolvedFieldReference => false // OK
+ case e => true
+ }
+
+ if (illegalKeys.nonEmpty) {
+ throw new ExpressionException("Illegal key expressions: " +
illegalKeys.mkString(", "))
+ }
+
+ this.copy(operation = GroupBy(operation, analyzedFields))
+ }
+
+ /**
+ * Groups the elements on some grouping keys. Use this before a
selection with aggregations
+ * to perform the aggregation on a per-group basis. Similar to a SQL
GROUP BY statement.
+ *
+ * Example:
+ *
+ * {{{
+ * in.groupBy("key").select("key, value.avg")
+ * }}}
+ */
+ def groupBy(fields: String): Table = {
+ val fieldsExpr = ExpressionParser.parseExpressionList(fields)
+ groupBy(fieldsExpr: _*)
+ }
+
+ /**
+ * Joins two [[Table]]s. Similar to an SQL join. The fields of the two
joined
+ * operations must not overlap, use [[as]] to rename fields if
necessary. You can use
+ * where and select clauses after a join to further specify the
behaviour of the join.
+ *
+ * Example:
+ *
+ * {{{
+ * left.join(right).where('a === 'b && 'c > 3).select('a, 'b, 'd)
+ * }}}
+ */
+ def join(right: Table): Table = {
+ val leftInputNames = operation.outputFields.map(_._1).toSet
+ val rightInputNames = right.operation.outputFields.map(_._1).toSet
+ if (leftInputNames.intersect(rightInputNames).nonEmpty) {
+ throw new ExpressionException(
+ "Overlapping fields names on join input, result would be
ambiguous: " +
+ operation.outputFields.mkString(", ") +
+ " and " +
+ right.operation.outputFields.mkString(", ") )
+ }
+ this.copy(operation = Join(operation, right.operation))
+ }
+
+ /**
+ * Union two[[Table]]s. Similar to an SQL UNION ALL. The fields of the
two union operations
+ * must fully overlap.
+ *
+ * Example:
+ *
+ * {{{
+ * left.unionAll(right)
+ * }}}
+ */
+ def unionAll(right: Table): Table = {
+ val leftInputFields = operation.outputFields
+ val rightInputFields = right.operation.outputFields
+ if (!leftInputFields.equals(rightInputFields)) {
+ throw new ExpressionException(
+ "The fields names of join inputs should be fully overlapped, left
inputs fields:" +
+ operation.outputFields.mkString(", ") +
+ " and right inputs fields" +
+ right.operation.outputFields.mkString(", ")
+ )
+ }
+ this.copy(operation = UnionAll(operation, right.operation))
+ }
+
+ /**
+ * Get the process of the sql parsing, print AST and physical execution
plan.The AST
+ * show the structure of the supplied statement. The execution plan
shows how the table
+ * referenced by the statement will be scanned.
+ */
+ def explain(extended: Boolean): String = {
--- End diff --
It looks like you add only this method.
Can you check why the whole file is marked as changed?
> Add explain support to print ast and sql physical execution plan.
> ------------------------------------------------------------------
>
> Key: FLINK-3192
> URL: https://issues.apache.org/jira/browse/FLINK-3192
> Project: Flink
> Issue Type: New Feature
> Components: Table API
> Reporter: GaoLun
> Assignee: GaoLun
> Priority: Minor
> Labels: features
>
> Table API doesn't support sql-explanation now. Add the explain support to
> print ast (abstract syntax tree) and the physical execution plan of sql.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
