Reconsider how we create KTables in KSQL

[rodesai]

Currently, KSQL creates a KTable for a Table source by:

1. Build a KStream over the changelog topic
2. Add some stateless transform processors for adding the rowtime and rowkey fields
3. Group the KStream and call GroupedKStream.reduce

This has a few problems:

• We're creating an extra changelog topic when we really don't need to. The state we need should be in the source topic.
• Because we are using an aggregator instead of building a KTable directly, we'll miss out on the timestamp semantics Kafka Streams is adding for dealing with out-of-order table updates.
• Reduce is dropping tombstones (we should fix this in ksql asap - tracked here: )

Longer term, it would be nice to add apis to Kafka Streams that KSQL can leverage to solve the above. Some ideas:

• Add a toTable API to KStream that returns a KTable for a changelog representation of a KStream
• add a transformValues API to KTable. We need transformValues to set the ROWTIME column since the message timestamp is only available from the processor context. Then we can just build the table directly and then call KTable.mapValues/KTable.transformValues

[guozhangwang]

Another idea as I was proposing for this issue, is to piggy back on KIP-159 to add an overloaded function of

StreamsBuilder.table(topic, context, mapper)

That it returns a new value from the mapper in which it can access the record context that returns the timestamp. It is similar to the transformValues idea above.