[SUPPORT] Errors ingesting enum's when using SQL Transformer

Created off the back of https://github.com/apache/hudi/issues/7867 to focus more on the SQL Transformer

**Describe the problem you faced**

We are using the DeltaStreamer on EMR 6.8.0, sourcing data from Confluent Kafka Avro topics and using our Confluent Schema Registry to deserialize the messages, which we write to the Glue Data Catalog and query with Athena.

For the majority of topics this works well, however, we noticed deserialisation errors when topics have Avro enum types in the schema.

Errors come in two forms, based on whether we use the default KafkaAvroDeserializer, or the KafkaAvroSchemaDeserializer **with an SQL Transformer**

With the KafkaAvroDeserializer
```
Scala.MatchError
Caused by: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 1.0 failed 4 times, most recent failure: Lost task 0.3 in stage 1.0 (TID 4) (ip-10-154-13-123.eu-west-1.compute.internal executor 1): scala.MatchError: processing (of class org.apache.avro.generic.GenericData$EnumSymbol)
at org.apache.hudi.org.apache.spark.sql.avro.AvroDeserializer.$anonfun$newWriter$13(AvroDeserializer.scala:178)
at org.apache.hudi.org.apache.spark.sql.avro.AvroDeserializer.$anonfun$newWriter$13$adapted(AvroDeserializer.scala:177)
at org.apache.hudi.org.apache.spark.sql.avro.AvroDeserializer.$anonfun$getRecordWriter$2(AvroDeserializer.scala:379)
at org.apache.hudi.org.apache.spark.sql.avro.AvroDeserializer.$anonfun$getRecordWriter$2$adapted(AvroDeserializer.scala:375)
at org.apache.hudi.org.apache.spark.sql.avro.AvroDeserializer.$anonfun$getRecordWriter$3(AvroDeserializer.scala:389)
at org.apache.hudi.org.apache.spark.sql.avro.AvroDeserializer.$anonfun$getRecordWriter$3$adapted(AvroDeserializer.scala:385)
at org.apache.hudi.org.apache.spark.sql.avro.AvroDeserializer.$anonfun$converter$4(AvroDeserializer.scala:87)
at org.apache.hudi.org.apache.spark.sql.avro.AvroDeserializer.deserialize(AvroDeserializer.scala:105)
at org.apache.hudi.org.apache.spark.sql.avro.HoodieSpark3_3AvroDeserializer.deserialize(HoodieSpark3_3AvroDeserializer.scala:30)
```

When using the KafkaAvroSchemaDeserializer, we get the following:
```
org.apache.avro.AvroTypeException
Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition {topic}-0 at offset 7202. If needed, please seek past the record to continue consumption.
Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id 4144
Caused by: org.apache.avro.AvroTypeException: Found {avro_record_namespace}.{enum_name}, expecting string
```

for an enum field resembling the following (note that sometimes there are default: added to the field, but never a default: at the symbol level...

```
{
"name": "status",
"type": {
"type": "enum",
"name": "status_options",
"symbols": [
"processing",
"completed",
"error"
]
}
},
```

With the KafkaAvroSchemaDeserializer --> org.apache.avro.AvroTypeException: Found {avro_record_namespace}.status_options, expecting string.
Without the KafkaAvroSchemaDeserializer --> scala.MatchError: {one_of_the_enum_symbols} (of class org.apache.avro.generic.GenericData$EnumSymbol

**To Reproduce**

With a duplicate environment to the one mentioned at the beginning, our Spark command is:

```
"spark-submit",
"--class", "org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer",
"--conf", "spark.scheduler.mode=FAIR",
"--conf", "spark.serializer=org.apache.spark.serializer.KryoSerializer",
"--conf", "spark.sql.catalogImplementation=hive",
"--conf", "spark.sql.hive.convertMetastoreParquet=false",
"--conf", "spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension",
"--conf", "spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog",
"--conf", "spark.hadoop.hive.metastore.client.factory.class=com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory",
"--conf", "spark.streaming.kafka.allowNonConsecutiveOffsets=true",
# IMPORTANT: hudi-utilities-bundle must be declared immediately before any Hudi spark commands
"/usr/lib/hudi/hudi-utilities-bundle.jar",
"--source-class", "org.apache.hudi.utilities.sources.{{ source_type }}",
"--source-ordering-field", "{{ timestamp_field }}",
"--table-type", "COPY_ON_WRITE",
"--op", "UPSERT",
"--enable-sync",
"--continuous",
# Hudi write config

"--target-base-path", f"s3://{bucket}/raw/{{ table }}",
"--target-table", "{{ table }}",
"--hoodie-conf", "hoodie.database.name={{ database }}_raw",
"--hoodie-conf", "hoodie.table.name={{ table }}",
"--hoodie-conf", "hoodie.datasource.write.recordkey.field={{ primary_key }}",
"--hoodie-conf", "hoodie.datasource.write.precombine.field={{ timestamp_field }}",
"--hoodie-conf", "hoodie.datasource.write.keygenerator.class=org.apache.hudi.keygen.TimestampBasedKeyGenerator",
"--hoodie-conf", "hoodie.datasource.write.partitionpath.field={{ timestamp_field }}",
"--hoodie-conf", "hoodie.deltastreamer.keygen.timebased.timestamp.type=EPOCHMILLISECONDS",
"--hoodie-conf", "hoodie.deltastreamer.keygen.timebased.output.dateformat=yyyy/MM/dd",
# Filter invalid records
"--transformer-class", "org.apache.hudi.utilities.transform.SqlQueryBasedTransformer",
"--hoodie-conf", "hoodie.deltastreamer.transformer.sql=SELECT * FROM WHERE {{ primary_key }} is not null AND {{ timestamp_field }} is not null",
# AWS Glue Data Catalog config
"--hoodie-conf", "hoodie.datasource.hive_sync.enable=true",
"--hoodie-conf", "hoodie.datasource.hive_sync.database={{ database }}_raw",
"--hoodie-conf", "hoodie.datasource.hive_sync.table={{ table }}",
"--hoodie-conf", "hoodie.datasource.hive_sync.partition_fields=_event_date",
"--hoodie-conf", f"hoodie.deltastreamer.source.kafka.topic={self.kafka_topic}",
"--hoodie-conf", "auto.offset.reset=earliest",
"--hoodie-conf", "sasl.mechanism=PLAIN",
"--hoodie-conf", "security.protocol=SASL_SSL",
"--hoodie-conf", f"bootstrap.servers={self.kafka_bootstrap_servers}",
"--hoodie-conf", f'sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required
username="{self.kafka_sasl_username}" password="{self.kafka_sasl_password}";',
"--schemaprovider-class", "org.apache.hudi.utilities.schema.SchemaRegistryProvider",
"--hoodie-conf", f"schema.registry.url=https://{self.schema_registry_url}",
"--hoodie-conf", "basic.auth.credentials.source=USER_INFO",
"--hoodie-conf", f"schema.registry.basic.auth.user.info={self.schema_registry_auth}",
"--hoodie-conf", f"hoodie.deltastreamer.schemaprovider.registry.url=https://{self.schema_registry_auth}@{self.schema_registry_url}//subjects/{self.kafka_topic}-value/versions/latest",
```

We are using this transformer to attempt to drop null records from being ingested, to avoid the below error we may sometimes face when there is empty data in our selected hoodie.datasource.write.partitionpath.field.

```
Caused by: org.apache.hudi.exception.HoodieException: The value of {timestamp_field} can not be null
at org.apache.hudi.avro.HoodieAvroUtils.getNestedFieldVal(HoodieAvroUtils.java:534)
at org.apache.hudi.utilities.deltastreamer.DeltaSync.lambda$fetchFromSource$b741bfe4$1(DeltaSync.java:490)
at org.apache.spark.api.java.JavaPairRDD$.$anonfun$toScalaFunction$1(JavaPairRDD.scala:1070)
```

Based on the fact that the SQL transformer appears to be causing the errors, there may be a better way of doing this?

**Expected behavior**

I expected the enum data types to be able to be ingested, even when including an SQL transformer to only select records that fit the query description, [based on this documentation](https://spark.apache.org/docs/latest/sql-data-sources-avro.html#supported-types-for-avro---spark-sql-conversion)

**Answers Required**

If the above is an incorrect assumption, the cause is known and this is expected behaviour, then my question becomes "How can I filter out potentially null records that will break the primary key / partition path field handling i'm using for my TimestampBasedKeyGenerator UPSERT/INSERT behaviour"

**Environment Description**

Hudi version : Deltastreamer on EMR 6.8.0 running Hudi 0.11.1-amzn-0
Spark version : 3.3.0
Hive version : 3.1.3
Hadoop version : Amazon 3.2.1
Storage (HDFS/S3/GCS..) : S3
Running on Docker? (yes/no) : No
EMR 6.9.0 that utilizes Hudi 0.12.1 wasn't used due to getting constant org.apache.hudi.exception.HoodieException: Could not sync using the meta sync class org.apache.hudi.hive.HiveSyncTool

**Additional context**

- I have visited the Hudi office hours before writing this issue, where it was suggested that I investigate and write up my findings here if the root cause is still unknown.
- If there's any detail missing from these tests please let me know and I can recreate and provide more clarity!
- The same configurations (With the default Deserializer, and without the SQL transformer) has been used with "INSERT" mode instead of "UPSERT" mode, with the same outcome

**Stacktraces**

[scala_match_error_stacktrace.txt](https://github.com/apache/hudi/files/10663859/scala_match_afmas_st_clean.txt)
Using KafkaAvroSchemaDeserializer - org.apache.avro.AvroTypeException
[avro_type_error_stacktrace.txt](https://github.com/apache/hudi/files/10663860/avro_type_afmas_st_clean.txt)

**Potentially useful information about the topic:**

-These 3 traces are all processing the same topic with the different configs discussed, one that throws the The value of {timestamp_field} can not be null error when the transformer is removed.
-Before each test or investigation, if s3 data or tables have been populated, I cancelled running EMR jobs, dropped any created tables and deleted underlying S3 data.
- This schema has 5 versions and although the schema registry only enforces BACKWARDS compatibility, I believe the current evolutions to adhere also to BACKWARDS_TRANSITIVE.
- The Enum failing in both the Scala Match Error and the org.apache.avro.AvroTypeException message “status” has been unchanged the entire time, and is the first enum field in the schema
- The topic appears to start at offset 7202, which is the offset declared in the org.apache.avro.AvroTypeException stack trace

Start with the provided AvroDeserializer.scala stack trace and the SqlQueryBasedTransformer configuration, then trace the DeltaSync path through the TimestampBasedKeyGenerator handling. Reproduce the enum failures with the supplied Spark, Kafka, Schema Registry, and SQL transformer settings. Done means identifying whether the failure is in Avro deserialization, transformation, or null-key handling and documenting or testing the supported behavior.

Written by the indexing model from the issue text.