pnowojski commented on a change in pull request #6787: [FLINK-8577][table]
Implement proctime DataStream to Table upsert conversion
URL: https://github.com/apache/flink/pull/6787#discussion_r236601884
##########
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
##########
@@ -555,18 +555,105 @@ abstract class StreamTableEnvironment(
s"But is: ${execEnv.getStreamTimeCharacteristic}")
}
+ // Can not apply key on append stream
+ if (extractUniqueKeys(fields).nonEmpty) {
+ throw new TableException(
+ s"Can not apply key on append stream, use fromUpsertStream instead.")
+ }
+
// adjust field indexes and field names
val indexesWithIndicatorFields = adjustFieldIndexes(fieldIndexes, rowtime,
proctime)
val namesWithIndicatorFields = adjustFieldNames(fieldNames, rowtime,
proctime)
- val dataStreamTable = new DataStreamTable[T](
+ val dataStreamTable = new AppendStreamTable[T](
dataStream,
indexesWithIndicatorFields,
namesWithIndicatorFields
)
registerTableInternal(name, dataStreamTable)
}
+ /**
+ * Registers an upsert [[DataStream]] as a table under a given name in the
[[TableEnvironment]]'s
+ * catalog.
+ *
+ * @param name The name under which the table is registered in the catalog.
+ * @param dataStream The [[DataStream]] to register as table in the catalog.
+ * @tparam T the type of the [[DataStream]].
+ */
+ protected def registerUpsertStreamInternal[T](name: String, dataStream:
DataStream[T]): Unit = {
Review comment:
Ok, I get it - upsert streams without a primary key are single row tables.
But what do you mean by:
> The primary key would be necessary.
?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
