Finish window function query planning stage.

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/FilterAndProjectOperator.java

@@ -171,6 +171,9 @@ private TsBlock getFilterTsBlock(TsBlock input) {
  if (!hasNonMappableUDF) {
  // get result of calculated common sub expressions
  for (ColumnTransformer columnTransformer : commonTransformerList) {
+ // CASE WHEN clause would clear all its cache
+ // evaluate again to acquire cache
+ columnTransformer.tryEvaluate();
  resultColumns.add(columnTransformer.getColumn());
  }
  }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/window/TableWindowOperator.java

@@ -30,6 +30,7 @@
 import org.apache.iotdb.db.queryengine.plan.planner.memory.MemoryReservationManager;
 
 import com.google.common.collect.ImmutableList;
+import com.google.common.util.concurrent.ListenableFuture;
 import org.apache.tsfile.block.column.Column;
 import org.apache.tsfile.common.conf.TSFileDescriptor;
 import org.apache.tsfile.enums.TSDataType;
@@ -134,6 +135,11 @@ public OperatorContext getOperatorContext() {
  return operatorContext;
  }
 
+ @Override
+ public ListenableFuture<?> isBlocked() {
+ return inputOperator.isBlocked();
+ }
+
  @Override
  public TsBlock next() throws Exception {
  long startTime = System.nanoTime();

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/window/function/WindowFunctionFactory.java

@@ -0,0 +1,67 @@
+/*
+ * 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.
+ */
+
+package org.apache.iotdb.db.queryengine.execution.operator.process.window.function;
+
+import org.apache.iotdb.db.queryengine.execution.operator.process.window.function.rank.CumeDistFunction;
+import org.apache.iotdb.db.queryengine.execution.operator.process.window.function.rank.DenseRankFunction;
+import org.apache.iotdb.db.queryengine.execution.operator.process.window.function.rank.NTileFunction;
+import org.apache.iotdb.db.queryengine.execution.operator.process.window.function.rank.PercentRankFunction;
+import org.apache.iotdb.db.queryengine.execution.operator.process.window.function.rank.RankFunction;
+import org.apache.iotdb.db.queryengine.execution.operator.process.window.function.rank.RowNumberFunction;
+import org.apache.iotdb.db.queryengine.execution.operator.process.window.function.value.FirstValueFunction;
+import org.apache.iotdb.db.queryengine.execution.operator.process.window.function.value.LagFunction;
+import org.apache.iotdb.db.queryengine.execution.operator.process.window.function.value.LastValueFunction;
+import org.apache.iotdb.db.queryengine.execution.operator.process.window.function.value.LeadFunction;
+import org.apache.iotdb.db.queryengine.execution.operator.process.window.function.value.NthValueFunction;
+
+import java.util.List;
+
+public class WindowFunctionFactory {
+ public static WindowFunction createBuiltinWindowFunction(
+ String functionName, List<Integer> argumentChannels, boolean ignoreNulls) {
+ switch (functionName) {
+ case "nth_value":
+ return new NthValueFunction(argumentChannels, ignoreNulls);
+ case "first_value":
+ return new FirstValueFunction(argumentChannels.get(0), ignoreNulls);
+ case "last_value":
+ return new LastValueFunction(argumentChannels.get(0), ignoreNulls);
+ case "lead":
+ return new LeadFunction(argumentChannels, ignoreNulls);
+ case "lag":
+ return new LagFunction(argumentChannels, ignoreNulls);
+ case "rank":
+ return new RankFunction();
+ case "dense_rank":
+ return new DenseRankFunction();
+ case "row_number":
+ return new RowNumberFunction();
+ case "percent_rank":
+ return new PercentRankFunction();
+ case "cume_dist":
+ return new CumeDistFunction();
+ case "ntile":
+ return new NTileFunction(argumentChannels.get(0));
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported built-in window function name: " + functionName);
+ }
+ }
+}

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/window/function/rank/NTileFunction.java

@@ -24,10 +24,10 @@
 import org.apache.tsfile.block.column.ColumnBuilder;
 
 public class NTileFunction extends RankWindowFunction {
- private final int n;
+ private final int nChannel;
 
- public NTileFunction(int n) {
- this.n = n;
+ public NTileFunction(int nChannel) {
+ this.nChannel = nChannel;
  }
 
  @Override
@@ -37,7 +37,12 @@ public void transform(
  int index,
  boolean isNewPeerGroup,
  int peerGroupCount) {
- builder.writeLong(bucket(n, index, partition.getPositionCount()) + 1);
+ if (partition.isNull(nChannel, index)) {
+ builder.appendNull();
+ } else {
+ long n = partition.getLong(nChannel, index);
+ builder.writeLong(bucket(n, index, partition.getPositionCount()) + 1);
+ }
  }
 
  private long bucket(long buckets, int index, int count) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/window/function/value/LagFunction.java

@@ -22,23 +22,34 @@
 import org.apache.iotdb.db.queryengine.execution.operator.process.window.partition.Partition;
 
 import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.write.UnSupportedDataTypeException;
+
+import java.util.List;
 
 public class LagFunction extends ValueWindowFunction {
  private final int channel;
- private final Integer offset;
- private final Object defaultVal;
+ private final int offsetChannel;
+ private final int defaultValChannel;
  private final boolean ignoreNull;
 
- public LagFunction(int channel, Integer offset, Object defaultVal, boolean ignoreNull) {
- this.channel = channel;
- this.offset = offset == null ? 1 : offset;
- this.defaultVal = defaultVal;
+ public LagFunction(List<Integer> argumentChannels, boolean ignoreNull) {
+ this.channel = argumentChannels.get(0);
+ this.offsetChannel = argumentChannels.size() > 1 ? argumentChannels.get(1) : -1;
+ this.defaultValChannel = argumentChannels.size() > 2 ? argumentChannels.get(2) : -1;
  this.ignoreNull = ignoreNull;
  }
 
  @Override
  public void transform(
  Partition partition, ColumnBuilder builder, int index, int frameStart, int frameEnd) {
+ if (offsetChannel >= 0 && partition.isNull(offsetChannel, index)) {
+ builder.appendNull();
+ return;
+ }
+
+ int offset = offsetChannel >= 0 ? partition.getInt(offsetChannel, index) : 1;
+
  int pos;
  if (ignoreNull) {
  int nonNullCount = 0;
@@ -63,13 +74,45 @@ public void transform(
  } else {
  builder.appendNull();
  }
- } else if (defaultVal != null) {
-builder.writeObject(defaultVal);
+ } else if (defaultValChannel >= 0) {
+writeDefaultValue(partition, defaultValChannel, index, builder);
  } else {
  builder.appendNull();
  }
  }
 
+ private void writeDefaultValue(
+ Partition partition, int defaultValChannel, int index, ColumnBuilder builder) {
+ TSDataType dataType = builder.getDataType();
+ switch (dataType) {
+ case INT32:
+ case DATE:
+ builder.writeInt(partition.getInt(defaultValChannel, index));
+ return;
+ case INT64:
+ case TIMESTAMP:
+ builder.writeLong(partition.getLong(defaultValChannel, index));
+ return;
+ case FLOAT:
+ builder.writeFloat(partition.getFloat(defaultValChannel, index));
+ return;
+ case DOUBLE:
+ builder.writeDouble(partition.getDouble(defaultValChannel, index));
+ return;
+ case BOOLEAN:
+ builder.writeBoolean(partition.getBoolean(defaultValChannel, index));
+ return;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ builder.writeBinary(partition.getBinary(defaultValChannel, index));
+ return;
+ default:
+ throw new UnSupportedDataTypeException(
+ "Unsupported default value's data type in Lag: " + dataType);
+ }
+ }
+
  @Override
  public boolean needFrame() {
  return false;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/window/function/value/LeadFunction.java

@@ -22,24 +22,34 @@
 import org.apache.iotdb.db.queryengine.execution.operator.process.window.partition.Partition;
 
 import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.write.UnSupportedDataTypeException;
+
+import java.util.List;
 
 public class LeadFunction extends ValueWindowFunction {
  private final int channel;
- private final Integer offset;
- private final Integer defaultVal;
+ private final int offsetChannel;
+ private final int defaultValChannel;
  private final boolean ignoreNull;
 
- public LeadFunction(int channel, Integer offset, Integer defaultVal, boolean ignoreNull) {
- this.channel = channel;
- this.offset = offset == null ? 1 : offset;
- this.defaultVal = defaultVal;
+ public LeadFunction(List<Integer> argumentChannels, boolean ignoreNull) {
+ this.channel = argumentChannels.get(0);
+ this.offsetChannel = argumentChannels.size() > 1 ? argumentChannels.get(1) : -1;
+ this.defaultValChannel = argumentChannels.size() > 2 ? argumentChannels.get(2) : -1;
  this.ignoreNull = ignoreNull;
  }
 
  @Override
  public void transform(
  Partition partition, ColumnBuilder builder, int index, int frameStart, int frameEnd) {
+ if (offsetChannel >= 0 && partition.isNull(offsetChannel, index)) {
+ builder.appendNull();
+ return;
+ }
+
  int length = partition.getPositionCount();
+ int offset = offsetChannel >= 0 ? partition.getInt(offsetChannel, index) : 1;
 
  int pos;
  if (ignoreNull) {
@@ -65,13 +75,45 @@ public void transform(
  } else {
  builder.appendNull();
  }
- } else if (defaultVal != null) {
-builder.writeObject(defaultVal);
+ } else if (defaultValChannel >= 0) {
+writeDefaultValue(partition, defaultValChannel, index, builder);
  } else {
  builder.appendNull();
  }
  }
 
+ private void writeDefaultValue(
+ Partition partition, int defaultValChannel, int index, ColumnBuilder builder) {
+ TSDataType dataType = builder.getDataType();
+ switch (dataType) {
+ case INT32:
+ case DATE:
+ builder.writeInt(partition.getInt(defaultValChannel, index));
+ return;
+ case INT64:
+ case TIMESTAMP:
+ builder.writeLong(partition.getLong(defaultValChannel, index));
+ return;
+ case FLOAT:
+ builder.writeFloat(partition.getFloat(defaultValChannel, index));
+ return;
+ case DOUBLE:
+ builder.writeDouble(partition.getDouble(defaultValChannel, index));
+ return;
+ case BOOLEAN:
+ builder.writeBoolean(partition.getBoolean(defaultValChannel, index));
+ return;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ builder.writeBinary(partition.getBinary(defaultValChannel, index));
+ return;
+ default:
+ throw new UnSupportedDataTypeException(
+ "Unsupported default value's data type in Lag: " + dataType);
+ }
+ }
+
  @Override
  public boolean needFrame() {
  return false;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/window/function/value/NthValueFunction.java

@@ -23,33 +23,36 @@
 
 import org.apache.tsfile.block.column.ColumnBuilder;
 
+import java.util.List;
+
 public class NthValueFunction extends ValueWindowFunction {
- private final int n;
- private final int channel;
+ private final int valueChannel;
+ private final int nChannel;
  private final boolean ignoreNull;
 
- public NthValueFunction(int n, int channel, boolean ignoreNull) {
- this.n = n;
- this.channel = channel;
+ public NthValueFunction(List<Integer> argumentChannels, boolean ignoreNull) {
+ this.valueChannel = argumentChannels.get(0);
+ this.nChannel = argumentChannels.get(1);
  this.ignoreNull = ignoreNull;
  }
 
  @Override
  public void transform(
  Partition partition, ColumnBuilder builder, int index, int frameStart, int frameEnd) {
  // Empty frame
- if (frameStart < 0) {
+ if (frameStart < 0 || partition.isNull(nChannel, index)) {
  builder.appendNull();
  return;
  }
 
  int pos;
+ int n = partition.getInt(nChannel, index);
  if (ignoreNull) {
  // Handle nulls
  pos = frameStart;
  int nonNullCount = 0;
  while (pos <= frameEnd) {
- if (!partition.isNull(channel, pos)) {
+ if (!partition.isNull(valueChannel, pos)) {
  nonNullCount++;
  if (nonNullCount == n) {
  break;
@@ -59,7 +62,7 @@ public void transform(
  }
 
  if (pos <= frameEnd) {
- partition.writeTo(builder, channel, pos);
+ partition.writeTo(builder, valueChannel, pos);
  } else {
  builder.appendNull();
  }
@@ -69,8 +72,8 @@ public void transform(
  // n starts with 1
  pos = frameStart + n - 1;
  if (pos <= frameEnd) {
- if (!partition.isNull(channel, pos)) {
- partition.writeTo(builder, channel, pos);
+ if (!partition.isNull(valueChannel, pos)) {
+ partition.writeTo(builder, valueChannel, pos);
  } else {
  builder.appendNull();
  }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/window/partition/PartitionExecutor.java

@@ -111,7 +111,7 @@ public PartitionExecutor(
  frame = new RangeFrame(partition, frameInfo, sortedColumns, peerGroupComparator);
  break;
  case ROWS:
- frame = new RowsFrame(partition, frameInfo, partitionStart, partitionEnd);
+ frame = new RowsFrame(partition, frameInfo);
  break;
  case GROUPS:
  frame =