title: Structured Streaming Aggregations return wrong values
body: <p>I have written a Structured Streaming aggregation that takes events from a Kafka Source, performs a simple count and writes them back to a Cassandra Database. The code looks like this:</p>
  
  <pre class="lang-scala prettyprint-override"><code>val data = stream
    .groupBy(functions.to_date($"timestamp").as("date"), $"type".as("type"))
    .agg(functions.count("*").as("value"))
  
  val query: StreamingQuery = data
    .writeStream
    .queryName("group-by-type")
    .format("org.apache.spark.sql.streaming.cassandra.CassandraSinkProvider")
    .outputMode(OutputMode.Complete())
    .option("checkpointLocation", config.getString("checkpointLocation") + "/" + "group-by-type")
    .option("keyspace", "analytics")
    .option("table", "aggregations")
    .option("partitionKeyColumns", "project,type")
    .option("clusteringKeyColumns", "date")
    .start()
  </code></pre>
  
  <p>The problem is that the count is just over every single batch. So I will see counts dropping in Cassandra. The counts should never drop over a day, how can I achieve that?</p>
  
  <p>Edit:
  I have tried using window aggregations too, same thing</p>