[SUPPORT] SqlQueryBasedTransformer new field issue with PostgresDebeziumSource about hudi

[soumilshah1995]

slack chat Thread https://apache-hudi.slack.com/archives/C4D716NPQ/p1718691646054409

this issue arises when attempting to use transformer with
--source-class org.apache.hudi.utilities.sources.debezium.PostgresDebeziumSource \

--transformer-class org.apache.hudi.utilities.transform.SqlQueryBasedTransformer
--hoodie-conf 'hoodie.deltastreamer.transformer.sql=SELECT * FROM '

Throws an issue even when using Select *

[ad1happy2go]

@ashwinagalcha-ps The sql should have SRC also, right?
"--hoodie-conf", "hoodie.streamer.transformer.sql=SELECT *, extract(year from a.created_at) as year FROM a",

https://hudi.apache.org/docs/transforms/

[ashwinagalcha-ps]

@ad1happy2go I missed to add here. But we did test with in the query. Sorry for the confusion, i'll update the description.

[ashwinagalcha-ps]

@ad1happy2go It was already added here but since it was plain text and not code github skipped <> while saving.

[soumilshah1995]

Hey Aditya I personally verified SQL transformer throws error

Here is lab : https://github.com/soumilshah1995/universal-datalakehouse-postgres-ingestion-deltastreamer

I changed the streamer code

========================================================
spark-submit \
    --class org.apache.hudi.utilities.streamer.HoodieStreamer \
    --packages 'org.apache.hudi:hudi-spark3.4-bundle_2.12:0.14.0,org.apache.hadoop:hadoop-aws:3.3.2' \
    --properties-file spark-config.properties \
    --master 'local[*]' \
    --executor-memory 1g \
     /Users/soumilshah/IdeaProjects/SparkProject/DeltaStreamer/jar/hudi-utilities-slim-bundle_2.12-0.14.0.jar \
    --table-type COPY_ON_WRITE \
    --target-base-path 's3a://warehouse/database=default/table_name=salestest'  \
    --target-table salestest \
    --op UPSERT \
    --source-limit 4000000 \
    --source-class org.apache.hudi.utilities.sources.debezium.PostgresDebeziumSource \
    --min-sync-interval-seconds 10 \
    --continuous \
    --source-ordering-field _event_origin_ts_ms \
    --transformer-class org.apache.hudi.utilities.transform.SqlQueryBasedTransformer \
    --hoodie-conf bootstrap.servers=localhost:7092 \
    --hoodie-conf schema.registry.url=http://localhost:8081 \
    --hoodie-conf hoodie.deltastreamer.schemaprovider.registry.url=http://localhost:8081/subjects/hive.public.sales-value/versions/latest \
    --hoodie-conf hoodie.deltastreamer.source.kafka.value.deserializer.class=io.confluent.kafka.serializers.KafkaAvroDeserializer \
    --hoodie-conf hoodie.deltastreamer.source.kafka.topic=hive.public.sales \
    --hoodie-conf auto.offset.reset=earliest \
    --hoodie-conf 'hoodie.deltastreamer.transformer.sql=SELECT * FROM <SRC>' \
    --hoodie-conf hoodie.datasource.write.recordkey.field=salesid \
    --hoodie-conf 'hoodie.datasource.write.partitionpath.field=' \
    --hoodie-conf hoodie.datasource.write.precombine.field=_event_origin_ts_ms

if you see we just tried simple select *

    --hoodie-conf 'hoodie.deltastreamer.transformer.sql=SELECT * FROM <SRC>' \

[ad1happy2go]

Thanks @soumilshah1995 . Will checkout.

[soumilshah1995]

aye aye captain