[BUG] serialize_to_bytes

[mbroecheler]

Describe the bug
A clear and concise description of what the bug is.

When using the function, I get the following at runtime:

java.lang.NullPointerException: Cannot invoke "org.apache.flink.table.annotation.DataTypeHint.rawSerializer()" because "hint" is null
        at com.datasqrl.flinkrunner.stdlib.commons.serialize_to_bytes.eval(serialize_to_bytes.java:33)
        at StreamExecCalc$18.processElement_split179(Unknown Source)
        at StreamExecCalc$18.processElement(Unknown Source)
        at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:75)
        at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50)
        at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
        at org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:52)
        at org.apache.flink.table.runtime.operators.aggregate.GroupAggFunction.processElement(GroupAggFunction.java:194)
        at org.apache.flink.table.runtime.operators.aggregate.GroupAggFunction.processElement(GroupAggFunction.java:43)
        at org.apache.flink.streaming.api.operators.KeyedProcessOperator.processElement(KeyedProcessOperator.java:83)
        at org.apache.flink.streaming.runtime.io.RecordProcessorUtils.lambda$getRecordProcessor$0(RecordProcessorUtils.java:60)
        at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:238)
        at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:157)
        at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:114)
        at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
        at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:580)
        at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231)
        at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:917)
        at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:859)
        at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:958)
        at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:937)
        at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:751)
        at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566)
        at java.base/java.lang.Thread.run(Thread.java:840)

To Reproduce
Steps to reproduce the behavior:

Expected behavior
A clear and concise description of what you expected to happen.

Environment

• Java version:
• Flink version:
• OS:

Additional context
Add any other context about the problem here.

[ferenc-csaky]

How do we wanna improve this? I wonder about what's more important performance or usability?

Cause now this udf will only serialize objects that explicitly define a rawSerializer via the @DataTypeHint annotation. It would be quite easy to fallback to a default serialized via TypeExtractor in Flink, but for complex types (List, Map, JsonNode that will mean Kryo, and worse performance.