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_r236168761
##########
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:
aren't we missing here the user defined primary key? What's the value of
upsert stream without a defined primary key?
----------------------------------------------------------------
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:
us...@infra.apache.org
With regards,
Apache Git Services
