Initial CDC support for DHC

[jcferretti]

[cpwright]

I think you've basically taken away a stream table as a valid output. You can:
(1) ignore the keys and get a lastBy
(2) just make it append only, so you have a huge table
(3) what about if I know what I'm doing and want the raw stream table instead? Like maybe I don't actually care about these individual things, but I want to aggregate them with counts and sums. Or in an advanced use case maybe I can instead do something like:
purchases = streamTable.where(op = add).view("Dollars").sumBy()
returns = streamTable.where(op = del).view("Dollars").sumBy()
netRevenue = purchases.naturalJoin(returns, "", "ReturnDollars=Dollars").update("NetRevenue=Dollars-ReturnDollars")

[jcferretti]

Ref #1819 (review)
(I can't answer directly under that comment, sadly, very poor github UI design).
Makes sense and it becomes obvious when looking at the implementation code; just returning the streaming table before wrapping it in a call to streamingToAppendTable is obviously more general and the user themselves can do the wrapping if they so desire.

I've reworked some other parts of the change too; there was a bug in that the after__X columns are not present on a delete, so it is important to take the primary key columns always from the Key component (those are present in a delete).

Something else I learned from debezium documentation, when an update in the underlying database table modifies a primary key, debezium generates two events, one to delete the old row and one to add the new row. Is very important for our algorithm that they do this, otherwise the trick for

.where("op = `d`")

wouldn't work; I thought about this case, went into panic, went to read the debezium docs, and was relieved.