改成能执行的--resource_functionon1;CREATETABLEsourceidTINYINTprincipal_partSTRINGprincipal_part_typeTINYINTresourcesSTRINGresources_typeTINYINTspace_codeSTRINGnoteSTRINGstatusTINYINTgmt_createTIMESTAMPgmt_m

--@resource_function{"on1"}

CREATE TABLE source ( id TINYINT, principal_part STRING, principal_part_type TINYINT, resources STRING, resources_type TINYINT, space_code STRING, note STRING, status TINYINT, gmt_create TIMESTAMP, gmt_modified TIMESTAMP(3), PROCTIME AS PROCTIME(), WATERMARK FOR gmt_modified AS gmt_modified - INTERVAL '1' SECOND ) WITH ( 'properties.bootstrap.servers' = 'ip:9092', 'format' = 'debezium-json', 'connector' = 'kafka', 'topic' = 'simba_test', 'debezium-json.timestamp-format.standard' = 'SQL', 'scan.startup.mode' = 'latest-offset' );

CREATE TABLE dim_sql ( sex_id INT, sex_name STRING, PRIMARY KEY (sex_id) NOT ENFORCED ) WITH ( 'url' = 'jdbc:mysql://ip:3306/simba_test?nullCatalogMeansCurrent=true', 'lookup.cache.ttl' = '1000', 'lookup.cache.type' = 'all', 'table-name' = 'flink_test_sex', 'password' = 'password', 'connector' = 'mysql', 'lookup.cache.max-rows' = '10000', 'username' = 'username' );

CREATE TABLE result_sql ( id INT, principal_part STRING, principal_part_type TINYINT, resources STRING, resources_type TINYINT, space_code STRING, note BIGINT, status TINYINT, gmt_create TIMESTAMP, gmt_modified TIMESTAMP, PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'sink.buffer-flush.max-rows' = '100', 'url' = 'jdbc:mysql://ip:3306/simba_test?nullCatalogMeansCurrent=true', 'sink.buffer-flush.interval' = '1000', 'table-name' = 'tbl_test_01_kafka_mysql', 'password' = 'password', 'connector' = 'mysql', 'username' = 'username' );

INSERT INTO result_sql SELECT a.id, b.sex_name AS principal_part, a.principal_part_type, a.resources, a.resources_type, a.space_code, on1(CAST(a.note AS BIGINT)) AS note, a.status, a.gmt_create, a.gmt_modified FROM source a LEFT JOIN dim_sql FOR SYSTEM_TIME AS OF a.PROCTIME AS b ON a.principal_part_type = b.sex_id