Hello, I am a Flink beginner and I'm trying to set up a Kinesis -> Hudi streaming pipeline. I'm trying to understand the cause of the following error message:

pyflink.util.exceptions.TableException: org.apache.flink.table.api.TableException: Unsupported Python SqlFunction CAST.
	at org.apache.flink.table.planner.plan.nodes.exec.utils.CommonPythonUtil.createPythonFunctionInfo(CommonPythonUtil.java:146)
	at org.apache.flink.table.planner.plan.nodes.exec.utils.CommonPythonUtil.createPythonFunctionInfo(CommonPythonUtil.java:429)

Basically, when I add watermarking to the Kinesis source, the error appears.

def get_replication_event_schema_for_kinesis_source() -> Schema:
    return (
        Schema.new_builder()
        .column("op", DataTypes.STRING())
        .column("table", DataTypes.STRING())
        .column("data", DataTypes.STRING())
        .column("event_time", DataTypes.TIMESTAMP(3))
        .column_by_metadata(
            column_name="kinesis_ts",
            data_type=DataTypes.TIMESTAMP(3),
            metadata_key="timestamp",
        )
        .column_by_metadata(
            column_name="kinesis_shard",
            data_type=DataTypes.STRING(),
            metadata_key="shard-id",
        )
        .column_by_metadata(
            column_name="kinesis_sequence_number",
            data_type=DataTypes.STRING(),
            metadata_key="sequence-number",
        )
        # >>>>>>>>>> this line <<<<<<<<<<<<<<<<,
        #.watermark("event_time", "event_time - interval '5' second")
        .build()
    )

Also, when I don't stream to Hudi, but to console, the error does not appear. Flink version is 1.15.4 (latest supported by KDA). I wonder if somebody has advice on how I should approach debugging this error or point out if I've overlooked something important.

The error messages seems to indicate the data type of one or more fields is not specified correctly. I had a similar issue while I was working with the data stream API. I specified epoch milliseconds (eg 1698263980000) as

Types.BIG_INT()

and it failed to sink to Kafka while it was okay to print. It was only after I changed it into

Types.LONG()

that the error was gone.

Hi Jaehyeon, thanks for the tip! I'll double check the column types.