<!--

    Licensed to the Apache Software Foundation (ASF) under one
    or more contributor license agreements.  See the NOTICE file
    distributed with this work for additional information
    regarding copyright ownership.  The ASF licenses this file
    to you under the Apache License, Version 2.0 (the
    "License"); you may not use this file except in compliance
    with the License.  You may obtain a copy of the License at

        http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing,
    software distributed under the License is distributed on an
    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
    KIND, either express or implied.  See the License for the
    specific language governing permissions and limitations
    under the License.

-->

# 降采样补空值查询

GroupByFill 查询的主要逻辑在 `GroupByFillDataSet`

* `org.apache.iotdb.db.query.dataset.groupby.GroupByFillDataSet`

GroupByFill 是对原降采样结果进行填充，目前仅支持使用前值填充的方式。

* 在 Group By Fill 中，Group by 子句不支持滑动步长，否则抛异常
* Fill 子句中仅能使用 Previous 和 PREVIOUSUNTILLAST 这两种插值方式，Linear 不支持
* Previous 和 PREVIOUSUNTILLAST 对 fill 的时间不做限制
* 填充只针对 last_value 这一聚合函数，其他的函数不支持，如果其他函数的聚合值查询结果为 null，依旧为 null，不进行填充

## PREVIOUSUNTILLAST 与 PREVIOUS 填充的区别

Previous 填充方式的语意没有变，只要前面有值，就可以拿过来填充；
PREVIOUSUNTILLAST 考虑到在某些业务场景下，所填充的值的时间不能大于该时间序列 last 的时间戳（从业务角度考虑，取历史数据不能取未来历史数据）
看下面的例子，或许更容易理解

A 点时间戳为 1，B 为 5，C 为 20，D 为 30，N 为 8，M 为 38

原始数据为<img style="width:100%; max-width:800px; max-height:600px; margin-left:auto; margin-right:auto; display:block;" src="https://user-images.githubusercontent.com/16079446/78784824-9f41ae00-79d8-11ea-9920-0825e081cae0.png">

`select temperature FROM root.ln.wf01.wt01 where time >= 1 and time <= 38`

| Time   | root.ln.wf01.wt01.temperature  |
| ------ | ------------------------------ |
| 1      | 21                             |
| 3      | 23                             |
| 5      | 25                             |
| 20     | 26                             |
| 27     | 29                             |
| 28     | 30                             |
| 30     | 40                             |

当我们使用 Previous 插值方式时，即使 D 到 M 这一段是未来的数据，我们也会用 D 点的数据进行填充

`SELECT last_value(temperature) as last_temperature FROM root.ln.wf01.wt01 GROUP BY([8, 39), 5m) FILL (int32[previous])`

| Time   | last_temperature |
| ------ | ---------------- |
| 8      | 25               |
| 13     | 25               |
| 18     | 26               |
| 23     | 29               |
| 28     | 40               |
| 33     | 40               |
| 38     | 40               |

当我们使用 NONLASTPREVIOUS 插值方式时，因为 D 到 M 这一段是未来的数据，我们不会进行插值，还是返回 null

`SELECT last_value(temperature) as last_temperature FROM root.ln.wf01.wt01 GROUP BY([8, 39), 5m) FILL (int32[PREVIOUSUNTILLAST])`

| Time   | last_temperature |
| ------ | ---------------- |
| 8      | 25               |
| 13     | 25               |
| 18     | 26               |
| 23     | 29               |
| 28     | 40               |
| 33     | null             |
| 38     | null             |

## 核心查询逻辑

在`GroupByFillDataSet`中维护了两个主要变量

```
// the first value for each time series
private Object[] previousValue;
// last timestamp for each time series
private long[] lastTimeArray;
```
### `previousValue`

`previousValue`这个变量维护了当前时间窗口的前一个降采样值，在`GroupByFillDataSet`构造函数中调用了`initPreviousParis`方法对其进行初始化。

```
  private void initPreviousParis(QueryContext context, GroupByFillPlan groupByFillPlan)
          throws StorageEngineException, IOException, QueryProcessException {
    previousValue = new Object[paths.size()];
    for (int i = 0; i < paths.size(); i++) {
      Path path = paths.get(i);
      TSDataType dataType = dataTypes.get(i);
      IFill fill = new PreviousFill(dataType, groupByEngineDataSet.getStartTime(), -1L);
      fill.constructReaders(path, groupByFillPlan.getAllMeasurementsInDevice(path.getDevice()), context);

      TimeValuePair timeValuePair = fill.getFillResult();
      if (timeValuePair == null || timeValuePair.getValue() == null) {
        previousValue[i] = null;
      } else {
        previousValue[i] = timeValuePair.getValue().getValue();
      }
    }
  }
```

`initPreviousParis`方法主要为每个时间序列构造了一个单点补空值查询，`queryTime`设置为降采样时间窗口的起始值，`beforeRange`不作限制。

### `lastTimeArray`

`lastTimeArray`这个变量维护了每个时间序列的最近时间戳值，主要用于`PREVIOUSUNTILLAST`这一填充方式，在`GroupByFillDataSet`构造函数中调用了`initLastTimeArray`方法对其进行初始化。

```
  private void initLastTimeArray(QueryContext context)
      throws IOException, StorageEngineException, QueryProcessException {
    lastTimeArray = new long[paths.size()];
    Arrays.fill(lastTimeArray, Long.MAX_VALUE);
    for (int i = 0; i < paths.size(); i++) {
      TimeValuePair lastTimeValuePair =
          LastQueryExecutor.calculateLastPairForOneSeries(paths.get(i), dataTypes.get(i), context);
      if (lastTimeValuePair.getValue() != null) {
        lastTimeArray[i] = lastTimeValuePair.getTimestamp();
      }
    }
  }
```
`initPreviousParis`方法主要为每个时间序列构造了一个最近时间戳 Last 查询

### 填充过程

填充过程在`nextWithoutConstraint`方法中完成，主要逻辑如下：

```
protected RowRecord nextWithoutConstraint() throws IOException {

    // 首先通过 groupByEngineDataSet，获得原始的降采样查询结果行
    RowRecord rowRecord = groupByEngineDataSet.nextWithoutConstraint();

    // 接下来对每个时间序列判断需不需要填充
    for (int i = 0; i < paths.size(); i++) {
      Field field = rowRecord.getFields().get(i);
      // 当前值为 null，需要进行填充
      if (field.getDataType() == null) {
        // 当前一个值不为 null 并且 （填充方式不是 PREVIOUSUNTILLAST 或者 当前时间小于改时间序列的最近时间戳）
        if (previousValue[i] != null
            && (!((PreviousFill) fillTypes.get(dataTypes.get(i))).isUntilLast()
            || rowRecord.getTimestamp() <= lastTimeArray[i])) {
          rowRecord.getFields().set(i, Field.getField(previousValue[i], dataTypes.get(i)));
        }
      } else {
        // 当前值不为 null，不需要填充，用当前值更新 previousValue 数组
        previousValue[i] = field.getObjectValue(field.getDataType());
      }
    }
    return rowRecord;
  }
```