Hi Yinhua,
This is actually a bug in Flink table, you can check this issue
https://issues.apache.org/jira/browse/FLINK-10290 <
https://issues.apache.org/jira/browse/FLINK-10290>.
I opened a PR for this issue a couple of days ago, but there is still some problem so it’s not ready to be merged. We have used that in our internal Flink version, and for now it works well. May be you can take a look at it.
Best,
wangsan