[CALCITE-3073] Support read from special timestamp in KafkaAdapter

[wangzzu]

There are some changes in this PR：

1. support KafkaConsumer to read from special timestamp in KafkaAdapter;
2. there is some error in kafka_adapter.md (the row.converter in the json example should in operand range)

[wangzzu]

@XuMingmin Can you help me review this pr?

[zhztheplayer]

Is it able to have test cases for the changes? Also, please correct the case of word "support" to "Support" in commit message, as well as JIRA/PR title.

[wangzzu]

@zhztheplayer I have add the test case, can you help me review it again?

[mingmxu]

@wangzzu thanks to prepare it quickly, the implementation is a bit different from my thought.

FYI my idea, KAFKA_TIMESTAMP is a reserved keyword in KafkaAdapter, given a query like SELECT * FROM KAFKA_TABLE WHERE KAFKA_TIMESTAMP > CURRENT_TIMESTAMP-1, KafkaAdapter would take the filter clause KAFKA_TIMESTAMP > CURRENT_TIMESTAMP-1 directly.

In this way, users do not change table definition again and again, the only question is, is it good to have the reserved keyword, and how. (for HOW, we may have the column by default in KafkaRowConverter)

@danny0405 @asereda-gs any comments?

[asereda-gs]

@XuMingmin I will check this PR over the week-end

[wangzzu]

@XuMingmin I understand what you're saying, but I'm not sure is a good idea. Just add a special reserved keyword(KAFKA_TIMESTAMP) in KafkaAdapter maybe too weird, but it will make it easy to use.

[zhztheplayer]

@XuMingmin - Thanks for the comments. Just out of curiosity (If I get your point correctly), how does the columnKAFKA_TIMESTAMP distinguish with the existing MSG_TIMESTAMP? Does that mean we should emit 2 different timestamp columns?

[zhztheplayer]

Let's continue the discussion in JIRA - https://issues.apache.org/jira/browse/CALCITE-3073 (since it becomes a design discussion).