title: cleanSource option does not delete any files
body: <p>I have a Structured Streaming Job with Trigger.Once() enabled which I run each 20 minutes. After each running, I wat remove my processed parquet files from S3, so I enabled the <em>cleanSource</em> delete option, but it does not work and I don't know why !</p>
  <p>Before showing my code, I have to comment about him. I'm running multiple structured streaming queries in paralell, I have 5 buckets and I submit this in parallel. The job works perfectly, but does not delete any processed files.</p>
  <pre><code>var table = ['table1','table2','table3','table4','table5']
  tables.par.map(table =&gt; {
          new ReplicationTables().run(table)
        })
  
  object ReplicationTables {
  
  def run(table): Unit = {
   val dataFrame = spark.readStream
        .option(&quot;mergeSchema&quot;, &quot;true&quot;)
        .schema(dfSchema)
        .option(&quot;cleanSource&quot;,&quot;delete&quot;)  
        .parquet(s&quot;s3a://my-bucket/${table}/*&quot;)
  
  // I do some transformation and after I write my new dataframe called df to S3 in Delta format
  
  df.writeStream
        .format(&quot;delta&quot;)
        .outputMode(&quot;append&quot;)
        .queryName(s&quot;Delta/${table.schema}/${table.name}&quot;)
        .trigger(Trigger.Once())
       .option(&quot;checkpointLocation&quot;, s&quot;s3a://my-bucket/checkpoints/${table.schema}/${table.name}&quot;)
        .start(s&quot;s3a://my-bucket/Delta_Tables/${table}/&quot;)
        .awaitTermination()
  
  
  
  }
  
  }
  </code></pre>
  <p>PS: Even with INFO log level I does not have any logs about the <em>cleanSource</em></p>
  <p>PS 2: Follow the docs of Structured Streaming about <em>cleanSource</em> <a href="https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#input-sources" rel="nofollow noreferrer">https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#input-sources</a></p>