You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2016/07/19 12:17:38 UTC
flink git commit: [FLINK-2985] [table] Allow different field names
for unionAll() in Table API
Repository: flink
Updated Branches:
refs/heads/master de8406aab -> 5758c9199
[FLINK-2985] [table] Allow different field names for unionAll() in Table API
This closes #2078.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/5758c919
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/5758c919
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/5758c919
Branch: refs/heads/master
Commit: 5758c91999efcf45457e4c25d4a75b13ce13e486
Parents: de8406a
Author: gallenvara <ga...@126.com>
Authored: Mon Jun 6 16:30:41 2016 +0800
Committer: twalthr <tw...@apache.org>
Committed: Tue Jul 19 14:16:39 2016 +0200
----------------------------------------------------------------------
.../flink/api/table/codegen/CodeGenerator.scala | 2 +-
.../api/table/plan/logical/operators.scala | 3 ++-
.../plan/nodes/dataset/DataSetAggregate.scala | 2 +-
.../table/plan/schema/TableSourceTable.scala | 2 +-
.../flink/api/table/typeutils/RowTypeInfo.scala | 21 +++-----------------
.../api/table/typeutils/TypeConverter.scala | 6 +++---
.../scala/batch/sql/SetOperatorsITCase.scala | 8 ++++----
.../scala/batch/table/SetOperatorsITCase.scala | 8 ++++----
.../api/scala/stream/table/UnionITCase.scala | 2 +-
.../connectors/kafka/KafkaTableSource.java | 2 +-
.../JsonRowDeserializationSchema.java | 2 +-
11 files changed, 22 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/5758c919/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala
index 135b6f6..c1508d8 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/CodeGenerator.scala
@@ -522,7 +522,7 @@ class CodeGenerator(
(input2.getOrElse(throw new CodeGenException("Invalid input access.")), input2Term)
}
- val index = if (input._1 == input1) {
+ val index = if (input._2 == input1Term) {
inputRef.getIndex
} else {
inputRef.getIndex - input1.getArity
http://git-wip-us.apache.org/repos/asf/flink/blob/5758c919/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/logical/operators.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/logical/operators.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/logical/operators.scala
index f3f9412..4df8a5e 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/logical/operators.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/logical/operators.scala
@@ -283,7 +283,8 @@ case class Union(left: LogicalNode, right: LogicalNode, all: Boolean) extends Bi
s" ${left.output.size} and ${right.output.size}")
}
val sameSchema = left.output.zip(right.output).forall { case (l, r) =>
- l.resultType == r.resultType && l.name == r.name }
+ l.resultType == r.resultType
+ }
if (!sameSchema) {
failValidation(s"Union two tables of different schema:" +
s" [${left.output.map(a => (a.name, a.resultType)).mkString(", ")}] and" +
http://git-wip-us.apache.org/repos/asf/flink/blob/5758c919/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetAggregate.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetAggregate.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetAggregate.scala
index 0fbac5e..8aa18ca 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetAggregate.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetAggregate.scala
@@ -110,7 +110,7 @@ class DataSetAggregate(
.name(prepareOpName)
val groupReduceFunction = aggregateResult._2
- val rowTypeInfo = new RowTypeInfo(fieldTypes, rowType.getFieldNames.asScala)
+ val rowTypeInfo = new RowTypeInfo(fieldTypes)
val result = {
if (groupingKeys.length > 0) {
http://git-wip-us.apache.org/repos/asf/flink/blob/5758c919/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/schema/TableSourceTable.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/schema/TableSourceTable.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/schema/TableSourceTable.scala
index 03646f9..a11e8c1 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/schema/TableSourceTable.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/schema/TableSourceTable.scala
@@ -25,6 +25,6 @@ import org.apache.flink.api.table.typeutils.RowTypeInfo
/** Table which defines an external table via a [[TableSource]] */
class TableSourceTable(val tableSource: TableSource[_])
extends FlinkTable[Row](
- typeInfo = new RowTypeInfo(tableSource.getFieldTypes, tableSource.getFieldsNames),
+ typeInfo = new RowTypeInfo(tableSource.getFieldTypes),
fieldIndexes = 0.until(tableSource.getNumberOfFields).toArray,
fieldNames = tableSource.getFieldsNames)
http://git-wip-us.apache.org/repos/asf/flink/blob/5758c919/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/RowTypeInfo.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/RowTypeInfo.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/RowTypeInfo.scala
index d606a76..489edca 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/RowTypeInfo.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/RowTypeInfo.scala
@@ -25,34 +25,19 @@ import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo
import scala.collection.mutable.ArrayBuffer
import org.apache.flink.api.common.typeutils.TypeSerializer
-import org.apache.flink.api.table.{Row, TableException}
+import org.apache.flink.api.table.Row
/**
* TypeInformation for [[Row]].
*/
-class RowTypeInfo(fieldTypes: Seq[TypeInformation[_]], fieldNames: Seq[String])
+class RowTypeInfo(fieldTypes: Seq[TypeInformation[_]])
extends CaseClassTypeInfo[Row](
classOf[Row],
Array(),
fieldTypes,
- fieldNames)
+ for (i <- fieldTypes.indices) yield "f" + i)
{
- if (fieldTypes.length != fieldNames.length) {
- throw new TableException("Number of field types and names is different.")
- }
- if (fieldNames.length != fieldNames.toSet.size) {
- throw new TableException("Field names are not unique.")
- }
-
- def this(fieldTypes: Seq[TypeInformation[_]]) = {
- this(fieldTypes, for (i <- fieldTypes.indices) yield "f" + i)
- }
-
- def this(fieldTypes: Array[TypeInformation[_]], fieldNames: Array[String]) = {
- this(fieldTypes.toSeq, fieldNames.toSeq)
- }
-
def this(fieldTypes: Array[TypeInformation[_]]) = {
this(fieldTypes.toSeq)
}
http://git-wip-us.apache.org/repos/asf/flink/blob/5758c919/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeConverter.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeConverter.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeConverter.scala
index ea1f7ce..a162d9f 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeConverter.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeConverter.scala
@@ -32,7 +32,7 @@ import scala.collection.JavaConversions._
object TypeConverter {
- val DEFAULT_ROW_TYPE = new RowTypeInfo(Seq(), Seq()).asInstanceOf[TypeInformation[Any]]
+ val DEFAULT_ROW_TYPE = new RowTypeInfo(Seq()).asInstanceOf[TypeInformation[Any]]
/**
* Determines the return type of Flink operators based on the logical fields, the expected
@@ -115,7 +115,7 @@ object TypeConverter {
// Row is expected, create the arity for it
case Some(typeInfo) if typeInfo.getTypeClass == classOf[Row] =>
- new RowTypeInfo(logicalFieldTypes, logicalFieldNames)
+ new RowTypeInfo(logicalFieldTypes)
// no physical type
// determine type based on logical fields and configuration parameters
@@ -123,7 +123,7 @@ object TypeConverter {
// no need for efficient types -> use Row
// we cannot use efficient types if row arity > tuple arity or nullable
if (!useEfficientTypes || logicalFieldTypes.length > Tuple.MAX_ARITY || nullable) {
- new RowTypeInfo(logicalFieldTypes, logicalFieldNames)
+ new RowTypeInfo(logicalFieldTypes)
}
// use efficient type tuple or atomic type
else {
http://git-wip-us.apache.org/repos/asf/flink/blob/5758c919/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/sql/SetOperatorsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/sql/SetOperatorsITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/sql/SetOperatorsITCase.scala
index b25f84a..c4d78d3 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/sql/SetOperatorsITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/sql/SetOperatorsITCase.scala
@@ -46,12 +46,12 @@ class SetOperatorsITCase(
val env = ExecutionEnvironment.getExecutionEnvironment
val tEnv = TableEnvironment.getTableEnvironment(env, config)
- val sqlQuery = "SELECT c FROM t1 UNION ALL (SELECT c FROM t2)"
+ val sqlQuery = "SELECT c FROM t1 UNION ALL (SELECT f FROM t2)"
val ds1 = CollectionDataSets.getSmall3TupleDataSet(env)
val ds2 = CollectionDataSets.getSmall3TupleDataSet(env)
tEnv.registerDataSet("t1", ds1, 'a, 'b, 'c)
- tEnv.registerDataSet("t2", ds2, 'a, 'b, 'c)
+ tEnv.registerDataSet("t2", ds2, 'd, 'e, 'f)
val result = tEnv.sql(sqlQuery)
@@ -66,12 +66,12 @@ class SetOperatorsITCase(
val env = ExecutionEnvironment.getExecutionEnvironment
val tEnv = TableEnvironment.getTableEnvironment(env, config)
- val sqlQuery = "SELECT c FROM t1 UNION (SELECT c FROM t2)"
+ val sqlQuery = "SELECT c FROM t1 UNION (SELECT f FROM t2)"
val ds1 = CollectionDataSets.getSmall3TupleDataSet(env)
val ds2 = CollectionDataSets.getSmall3TupleDataSet(env)
tEnv.registerDataSet("t1", ds1, 'a, 'b, 'c)
- tEnv.registerDataSet("t2", ds2, 'a, 'b, 'c)
+ tEnv.registerDataSet("t2", ds2, 'd, 'e, 'f)
val result = tEnv.sql(sqlQuery)
http://git-wip-us.apache.org/repos/asf/flink/blob/5758c919/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/SetOperatorsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/SetOperatorsITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/SetOperatorsITCase.scala
index 83579e0..283ba10 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/SetOperatorsITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/SetOperatorsITCase.scala
@@ -46,7 +46,7 @@ class SetOperatorsITCase(
val tEnv = TableEnvironment.getTableEnvironment(env, config)
val ds1 = CollectionDataSets.getSmall3TupleDataSet(env).toTable(tEnv, 'a, 'b, 'c)
- val ds2 = CollectionDataSets.getSmall3TupleDataSet(env).toTable(tEnv, 'a, 'b, 'c)
+ val ds2 = CollectionDataSets.getSmall3TupleDataSet(env).toTable(tEnv, 'd, 'e, 'f)
val unionDs = ds1.unionAll(ds2).select('c)
@@ -61,7 +61,7 @@ class SetOperatorsITCase(
val tEnv = TableEnvironment.getTableEnvironment(env, config)
val ds1 = CollectionDataSets.getSmall3TupleDataSet(env).toTable(tEnv, 'a, 'b, 'c)
- val ds2 = CollectionDataSets.getSmall3TupleDataSet(env).toTable(tEnv, 'a, 'b, 'c)
+ val ds2 = CollectionDataSets.getSmall3TupleDataSet(env).toTable(tEnv, 'd, 'e, 'f)
val unionDs = ds1.union(ds2).select('c)
@@ -105,14 +105,14 @@ class SetOperatorsITCase(
}
@Test(expected = classOf[ValidationException])
- def testUnionDifferentFieldNames(): Unit = {
+ def testUnionDifferentColumnSize(): Unit = {
val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
val tEnv = TableEnvironment.getTableEnvironment(env, config)
val ds1 = CollectionDataSets.getSmall3TupleDataSet(env).toTable(tEnv, 'a, 'b, 'c)
val ds2 = CollectionDataSets.get5TupleDataSet(env).toTable(tEnv, 'a, 'b, 'd, 'c, 'e)
- // must fail. Union inputs have different field names.
+ // must fail. Union inputs have different column size.
ds1.unionAll(ds2)
}
http://git-wip-us.apache.org/repos/asf/flink/blob/5758c919/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/stream/table/UnionITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/stream/table/UnionITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/stream/table/UnionITCase.scala
index feda75f..131974e 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/stream/table/UnionITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/stream/table/UnionITCase.scala
@@ -38,7 +38,7 @@ class UnionITCase extends StreamingMultipleProgramsTestBase {
StreamITCase.testResults = mutable.MutableList()
val ds1 = StreamTestData.getSmall3TupleDataStream(env).toTable(tEnv, 'a, 'b, 'c)
- val ds2 = StreamTestData.getSmall3TupleDataStream(env).toTable(tEnv, 'a, 'b, 'c)
+ val ds2 = StreamTestData.getSmall3TupleDataStream(env).toTable(tEnv, 'd, 'e, 'f)
val unionDs = ds1.unionAll(ds2).select('c)
http://git-wip-us.apache.org/repos/asf/flink/blob/5758c919/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSource.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSource.java b/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSource.java
index e43760b..c6904fe 100644
--- a/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSource.java
+++ b/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSource.java
@@ -123,7 +123,7 @@ abstract class KafkaTableSource implements StreamTableSource<Row> {
@Override
public TypeInformation<Row> getReturnType() {
- return new RowTypeInfo(fieldTypes, fieldNames);
+ return new RowTypeInfo(fieldTypes);
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/5758c919/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/JsonRowDeserializationSchema.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/JsonRowDeserializationSchema.java b/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/JsonRowDeserializationSchema.java
index 1bbef4d..970c73e 100644
--- a/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/JsonRowDeserializationSchema.java
+++ b/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/JsonRowDeserializationSchema.java
@@ -118,7 +118,7 @@ public class JsonRowDeserializationSchema implements DeserializationSchema<Row>
@Override
public TypeInformation<Row> getProducedType() {
- return new RowTypeInfo(fieldTypes, fieldNames);
+ return new RowTypeInfo(fieldTypes);
}
/**