[FLINK-36416][table][runtime] Enable splittable timers for temporal j…

…oin, temporal sort and windowed aggregations

This is a follow up for https://cwiki.apache.org/confluence/display/FLINK/FLIP-443%3A+Interruptible+timers+firing . Temporal join, temporal sort and both windowed and table windowed aggregations in Table API/SQL can have large amount of registered/fired time, while at the same time enabling splittable timers shouldn't cause any side effects for those operators.

flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/temporal/BaseTwoInputStreamOperatorWithStateRetention.java

@@ -80,6 +80,11 @@ protected BaseTwoInputStreamOperatorWithStateRetention(
         this.stateCleaningEnabled = minRetentionTime > 1;
     }
 
+    @Override
+    public boolean useSplittableTimers() {
+        return true;
+    }
+
     @Override
     public void open() throws Exception {
         initializeTimerService();

flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sort/BaseTemporalSortOperator.java

@@ -38,6 +38,11 @@ abstract class BaseTemporalSortOperator extends AbstractStreamOperator<RowData>
 
     BaseTemporalSortOperator() {}
 
+    @Override
+    public boolean useSplittableTimers() {
+        return true;
+    }
+
     @Override
     public void open() throws Exception {
         InternalTimerService<VoidNamespace> internalTimerService =

flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/window/groupwindow/operator/WindowOperator.java

@@ -208,6 +208,11 @@ public abstract class WindowOperator<K, W extends Window> extends AbstractStream
         this.recordCounter = RecordCounter.of(inputCountIndex);
     }
 
+    @Override
+    public boolean useSplittableTimers() {
+        return true;
+    }
+
     WindowOperator(
             GroupWindowAssigner<W> windowAssigner,
             Trigger<W> trigger,