如何在Flink SQL中反序列化Avro枚举类型？

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-avro</artifactId>
    <version>${flink.version}</version>
</dependency>

CREATE TABLE avro_table (
    id INT,
    name STRING,
    status ENUM('ACTIVE', 'INACTIVE')
) WITH (
    'connector' = 'kafka',
    'format' = 'avro',
    'avro-schema' = '
        {
            "type": "record",
            "name": "MyRecord",
            "fields": [
                {"name": "id", "type": "int"},
                {"name": "name", "type": "string"},
                {"name": "status", "type": {
                    "type": "enum",
                    "name": "Status",
                    "symbols": ["ACTIVE", "INACTIVE"]
                }}
            ]
        }
    '
);

SELECT * FROM avro_table WHERE status = 'ACTIVE';

import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.formats.avro.AvroDeserializationSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;

public class AvroEnumDeserializationExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 创建AvroDeserializationSchema
        AvroDeserializationSchema<MyRecord> avroSchema = AvroDeserializationSchema.forSpecific(MyRecord.class);

        // 创建DataStream并指定AvroDeserializationSchema
        DataStream<MyRecord> dataStream = env
                .addSource(new FlinkKafkaConsumer<>("topic", avroSchema, properties))
                .name("Avro Source");

        // 将DataStream转换为Table
        Table table = tEnv.fromDataStream(dataStream);

        // 执行查询操作
        TableResult result = tEnv.executeSql("SELECT * FROM avro_table WHERE status = 'ACTIVE'");

        // 打印查询结果
        result.print();
    }

    // Avro记录类型
    public static class MyRecord {
        public int id;
        public String name;
        public Status status;
    }

    // Avro枚举类型
    public enum Status {
        ACTIVE,
        INACTIVE
    }
}