fix: circumvent KAFKA-10179 by forcing changelog topics for tables

[agavra]

Description

Partial fix for #5673. This PR introduces an extra map phase for any new kafka streams application that reads from a table, thereby preventing the streams optimization that uses the source topic as a changelog topic and (on recovery) loads data directly from the topic into rocksDB. For more discussion and context, see #5673 and https://issues.apache.org/jira/browse/KAFKA-10179

This is only a partial fix because it means any new queries running this code will not hit the problem, it doesn't fix existing queries. To do that, we will follow up this commit with one that preemptively registers the schema of the original topic into the subject of the changelog topic.

Review Guide

NOTE: This PR is split into two commits because it requires rewriting almost all historical plans that used tables. Also, if someone can take a look at joins.json and let me know if this fixes a bug or if it introduces a bug - I feel like the behavior after this PR is the right one.

Testing done

Unit testing, QTT and manual test; see tables.json

Reviewer checklist

• Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).
• Ensure relevant issues are linked (description should include text like "Fixes #")

[mjsax]

From what I can tell, LGTM.

[rodesai]

LGTM

[agavra]

Merging assuming @rodesai meant to press the green tick