[FEATURE]WIP: support Flink 1.11 and using flink-sql DDL to operate with pulsar about pulsar-flink HOT 3 CLOSED

set global.disable.operator.chain = true;

create table test_flink_sql(
	`rip` VARCHAR,
	`rtime`	VARCHAR,
	`uid`	bigint,
  	`rchannel`	VARCHAR,
  	`be_time`	bigint,
  	`be_time`	VARCHAR,
	`activity_id` VARCHAR,
	`country_code`	VARCHAR,
	`os`	VARCHAR,
	`recv_time`	bigint,
	`remark`	VARCHAR,
	`client_ip`	VARCHAR,
	`day` as TO_DATE(rtime),
	`hour` as date_format(rtime, 'HH')
) with (
	'connector.type' = 'pulsar',
	'connector.version' = '1',
	'connector.topic' = 'persistent://test/test-gray/test_flink_sql',
  	'connector.service-url' = 'pulsar://xxx',
  	'connector.admin-url' = 'http://xxx',
	'connector.startup-mode' = 'external-subscription',
	'connector.sub-name' = 'test_flink_sql_v1',
	'connector.properties.0.key' = 'pulsar.reader.readerName',
	'connector.properties.0.value' = 'test_flink_sql_v1',
	'connector.properties.1.key' = 'pulsar.reader.subscriptionRolePrefix',
	'connector.properties.1.value' = 'test_flink_sql_v1',
	'connector.properties.2.key' = 'pulsar.reader.receiverQueueSize',
	'connector.properties.2.value' = '1000',
	'connector.properties.3.key' = 'partitiondiscoveryintervalmillis',
	'connector.properties.3.value' = '5000',
	'format.type' = 'json',
	'format.derive-schema' = 'true',
	'format.ignore-parse-errors' = 'true',
    'update-mode' = 'append'
);

insert into hive.test.test_flink_sql
select
rip, rtime, 
if (uid is null, 0, uid) as uid,
if (activity_id is null, '', activity_id) as activity_id,
if (country_code is null, '', country_code) as country_code,
if (os is null, '', os) as os,
if (recv_time is null, 0, recv_time) as recv_time,
if (remark is null, '', remark) as remark,
if (client_ip is null, '', client_ip) as client_ip,
cast(`day` as string) as `day`, 
cast(`hour` as string) as `hour`
from test_flink_sql;