[FLINK]Support function `Reinterpret`

[KevinyhZou]

Description

While run nexmark q0 query with kafka table source, a runtime exceptions throws

java.lang.RuntimeException: Function not supported: Reinterpret

Gluten version

None

[KevinyhZou]

The example sql

create table s(
  a int, 
  b string,
  c string,
  d as case when a = 1 then cast(c as timestamp(3)) else cast(b as timestamp(3)) end,
  WATERMARK FOR d AS d - INTERVAL '4' SECOND
) with (
  'connector' = 'kafka',
  'topic' = '****',
  'properties.bootstrap.servers' = '****',
  'properties.group.id' = 'nexmark',
  'format' = 'json'
);

explain select e d from s where a = 1;

The query plan

Calc(select=[Reinterpret(CASE(=(a, 1), CAST(c AS TIMESTAMP(3)), CAST(b AS TIMESTAMP(3)))) AS d], where=[=(a, 1)])
+- TableSourceScan(table=[[default_catalog, default_database, s, watermark=[-(CASE(=(a, 1), CAST(c AS TIMESTAMP(3)), CAST(b AS TIMESTAMP(3))), 4000:INTERVAL SECOND)], watermarkEmitStrategy=[on-periodic]]], fields=[a, b, c])