Group By Fill query error in cluster mode

By replaying test cases in IoTDBGroupByFillIT.previousUntilLastTest1 in a 3-nodes with 2-replicas cluster, we can reappear the bug.

BUG1: select last_value(temperature) from root.ln.wf01.wt01 GROUP BY ([17, 48), 5ms) FILL(int32[previousUntilLast]);

Version: 0.12.2

Executing in a node having the timeseries data

Executing in a node without having the timeseries data

The last 2 rows, which filled by the previousUntilLast rule,are filled  with data. It is not as expected.

BUG2: select last_value(temperature) from root.ln.wf01.wt01 GROUP BY ([17, 48), 5ms) FILL(int32[previousUntilLast]) order by time desc;

Version: master & 0.12.2

call stack trace:

2021-11-15 14:14:54,172 [DataClientThread-3] ERROR o.a.t.ProcessFunction:47 - Internal error processing peekNextNotNullValue 
java.lang.NullPointerException: null
    at org.apache.iotdb.cluster.query.LocalQueryExecutor.peekNextNotNullValue(LocalQueryExecutor.java:836)
    at org.apache.iotdb.cluster.server.service.DataSyncService.peekNextNotNullValue(DataSyncService.java:427)
    at org.apache.iotdb.cluster.server.DataClusterServer.peekNextNotNullValue(DataClusterServer.java:969)
    at org.apache.iotdb.cluster.rpc.thrift.TSDataService$Processor$peekNextNotNullValue.getResult(TSDataService.java:2618)
    at org.apache.iotdb.cluster.rpc.thrift.TSDataService$Processor$peekNextNotNullValue.getResult(TSDataService.java:2598)
    at org.apache.thrift.ProcessFunction.process(ProcessFunction.java:38)
    at org.apache.thrift.TBaseProcessor.process(TBaseProcessor.java:38)
    at org.apache.thrift.server.TThreadPoolServer$WorkerProcess.run(TThreadPoolServer.java:248)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)

The main reason is:

1. GroupByFillDataSet uses LastQueryExecutor.calculateLastPairForSeriesLocally to init the lastTimeArray. But if the current node doesn't have the series data, the lastTimeArray will be left as Long.MAX_VALUE, which will fill any null fields until the last row. The solution is to pass a LastQueryExecutor as the GroupByFillDataSet field to call the calculateLastPairForSeries. In cluster mode, we use ClusterLastQueryExecutor.
2. The LocalQueryExecutor.peekNextNotNullValue returns null if there's no time-value pair satisfied. The result will be passed to the thrift response. However, thrift framework will treat null response as a failed one, an expection will occur. We should treat null result carefully.