[FLINK-35406] Use inner serializer when casting RAW type to BINARY or…

[docete]

… STRING in cast rules

What is the purpose of the change

This pull request fix the wrong behaviour in casting RAW to BINARY or STRING.
The generated code in RawToStringCastRule and RawToBinaryCastRule use
BinaryRawValueData::toBytes and BinaryRawValueData::toObject to convert
RawValueData(to java object or byte array), which should use inner serializer
instead of RawValueDataSerializer.

Brief change log

• 049fbd2 fix the wrong behaviour and add tests.

Verifying this change

This change is convered by new test cases in CastFunctionMiscITCase and CastFunctionMiscLegacyITCase

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): (yes / no)
• The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
• The serializers: (yes / no / don't know)
• The runtime per-record code paths (performance sensitive): (yes / no / don't know)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
• The S3 file system connector: (yes / no / don't know)

Documentation

• Does this pull request introduce a new feature? (yes / no)
• If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

[flinkbot]

CI report:

• 049fbd2 UNKNOWN
• df83a22 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[twalthr]

That is a nasty bug. Thanks for fixing it @docete. I just had a tiny code improvement suggestion. Otherwise LGTM.

[twalthr]

One final question before merging.

[twalthr]

Why do we actually need the call to getInnerSerializer() during runtime shouldn't we simply use this serializer as the typeSerializer$292 for code generation? Or does the implementation not allow this?

[docete]

Why do we actually need the call to getInnerSerializer() during runtime shouldn't we simply use this serializer as the typeSerializer$292 for code generation? Or does the implementation not allow this?

I think the RAW type should bind with RawValueDataSerializer, and use getInnerSerializer() in se/de phase. Follow the same pattern also make the Generated code more clear. See: AbstractBinaryWriter::writeRawValue

    public void writeRawValue(
            int pos, RawValueData<?> input, RawValueDataSerializer<?> serializer) {
        TypeSerializer innerSerializer = serializer.getInnerSerializer();
        // RawValueData only has one implementation which is BinaryRawValueData
        BinaryRawValueData rawValue = (BinaryRawValueData) input;
        rawValue.ensureMaterialized(innerSerializer);
        writeSegmentsToVarLenPart(
                pos, rawValue.getSegments(), rawValue.getOffset(), rawValue.getSizeInBytes());
    }

[docete]

BTW, modify the code generator make things more complicated