Rong's analysis is correct. The UNION with duplicate elimination is translated into a UNION ALL and a subsequent grouping operator on all attributes without an aggregation function.
Flink assumes that all grouping operators can produce retractions (updates) and window-grouped aggregates cannot handle retraction (yet).
However, there are few things to add here:
1) UNION ALL does not produce retractions, i.e., we never have to retract a row that we previously forwarded (unless the operator receives a retraction, however this is a different case as the operator only forwards but not produces a retraction).
2) We can implement a special operator UNION ALL operator for tables with timestamp attributes that is more efficient and able to automatically clean its state based on the progress of watermarks.
I've added JIRAs for both issues I've described above  .