Datalight,With regards to 1) I see that you are using the Medallion Architecture. Have you considered using AutoLoader for the detection and ingestion of new files in ADLS Gen2
@harraz- Replace.select("*", input_file_name().alias("source_file"), current_timestamp().alias("processing_time"))with.selectExpr("*","_metadata.file_path as source_file","current_timestamp() as processing_time")
