/*
 * 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.plan.planner;

import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.path.AlignedPath;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.queryengine.common.FragmentInstanceId;
import org.apache.iotdb.db.queryengine.common.NodeRef;
import org.apache.iotdb.db.queryengine.execution.aggregation.Accumulator;
import org.apache.iotdb.db.queryengine.execution.aggregation.AccumulatorFactory;
import org.apache.iotdb.db.queryengine.execution.aggregation.Aggregator;
import org.apache.iotdb.db.queryengine.execution.aggregation.slidingwindow.SlidingWindowAggregatorFactory;
import org.apache.iotdb.db.queryengine.execution.aggregation.timerangeiterator.ITimeRangeIterator;
import org.apache.iotdb.db.queryengine.execution.driver.DataDriverContext;
import org.apache.iotdb.db.queryengine.execution.driver.SchemaDriverContext;
import org.apache.iotdb.db.queryengine.execution.exchange.MPPDataExchangeManager;
import org.apache.iotdb.db.queryengine.execution.exchange.MPPDataExchangeService;
import org.apache.iotdb.db.queryengine.execution.exchange.sink.DownStreamChannelIndex;
import org.apache.iotdb.db.queryengine.execution.exchange.sink.ISinkChannel;
import org.apache.iotdb.db.queryengine.execution.exchange.sink.ISinkHandle;
import org.apache.iotdb.db.queryengine.execution.exchange.sink.LocalSinkChannel;
import org.apache.iotdb.db.queryengine.execution.exchange.sink.ShuffleSinkHandle;
import org.apache.iotdb.db.queryengine.execution.exchange.source.ISourceHandle;
import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceManager;
import org.apache.iotdb.db.queryengine.execution.operator.AggregationUtil;
import org.apache.iotdb.db.queryengine.execution.operator.ExplainAnalyzeOperator;
import org.apache.iotdb.db.queryengine.execution.operator.Operator;
import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
import org.apache.iotdb.db.queryengine.execution.operator.process.AggregationMergeSortOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.AggregationOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.ColumnInjectOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.DeviceViewIntoOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.DeviceViewOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.FillOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.FilterAndProjectOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.IntoOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.LimitOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.LinearFillOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.MergeSortOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.OffsetOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.ProcessOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.ProjectOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.RawDataAggregationOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.SingleDeviceViewOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.SlidingWindowAggregationOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.SortOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.TagAggregationOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.TopKOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.TransformOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.IFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.IFillFilter;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.ILinearFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.constant.BinaryConstantFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.constant.BooleanConstantFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.constant.DoubleConstantFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.constant.FloatConstantFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.constant.IntConstantFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.constant.LongConstantFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.filter.FixedIntervalFillFilter;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.filter.MonthIntervalMSFillFilter;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.filter.MonthIntervalNSFillFilter;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.filter.MonthIntervalUSFillFilter;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.identity.IdentityFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.identity.IdentityLinearFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.linear.DoubleLinearFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.linear.FloatLinearFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.linear.IntLinearFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.linear.LongLinearFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.previous.BinaryPreviousFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.previous.BooleanPreviousFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.previous.DoublePreviousFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.previous.FloatPreviousFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.previous.IntPreviousFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.previous.LongPreviousFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.FullOuterTimeJoinOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.HorizontallyConcatOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.InnerTimeJoinOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.LeftOuterTimeJoinOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.AscTimeComparator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.ColumnMerger;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.DescTimeComparator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.MultiColumnMerger;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.NonOverlappedMultiColumnMerger;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.SingleColumnMerger;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.TimeComparator;
import org.apache.iotdb.db.queryengine.execution.operator.process.last.AlignedUpdateLastCacheOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.last.AlignedUpdateViewPathLastCacheOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.last.LastQueryCollectOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.last.LastQueryMergeOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.last.LastQueryOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.last.LastQuerySortOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.last.LastQueryTransformOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.last.LastQueryUtil;
import org.apache.iotdb.db.queryengine.execution.operator.process.last.UpdateLastCacheOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.last.UpdateViewPathLastCacheOperator;
import org.apache.iotdb.db.queryengine.execution.operator.schema.CountGroupByLevelMergeOperator;
import org.apache.iotdb.db.queryengine.execution.operator.schema.CountGroupByLevelScanOperator;
import org.apache.iotdb.db.queryengine.execution.operator.schema.CountMergeOperator;
import org.apache.iotdb.db.queryengine.execution.operator.schema.NodeManageMemoryMergeOperator;
import org.apache.iotdb.db.queryengine.execution.operator.schema.NodePathsConvertOperator;
import org.apache.iotdb.db.queryengine.execution.operator.schema.NodePathsCountOperator;
import org.apache.iotdb.db.queryengine.execution.operator.schema.SchemaCountOperator;
import org.apache.iotdb.db.queryengine.execution.operator.schema.SchemaFetchMergeOperator;
import org.apache.iotdb.db.queryengine.execution.operator.schema.SchemaFetchScanOperator;
import org.apache.iotdb.db.queryengine.execution.operator.schema.SchemaQueryMergeOperator;
import org.apache.iotdb.db.queryengine.execution.operator.schema.SchemaQueryOrderByHeatOperator;
import org.apache.iotdb.db.queryengine.execution.operator.schema.SchemaQueryScanOperator;
import org.apache.iotdb.db.queryengine.execution.operator.schema.source.SchemaSourceFactory;
import org.apache.iotdb.db.queryengine.execution.operator.sink.IdentitySinkOperator;
import org.apache.iotdb.db.queryengine.execution.operator.sink.ShuffleHelperOperator;
import org.apache.iotdb.db.queryengine.execution.operator.source.AlignedSeriesAggregationScanOperator;
import org.apache.iotdb.db.queryengine.execution.operator.source.AlignedSeriesScanOperator;
import org.apache.iotdb.db.queryengine.execution.operator.source.ExchangeOperator;
import org.apache.iotdb.db.queryengine.execution.operator.source.SeriesAggregationScanOperator;
import org.apache.iotdb.db.queryengine.execution.operator.source.SeriesScanOperator;
import org.apache.iotdb.db.queryengine.execution.operator.source.ShowQueriesOperator;
import org.apache.iotdb.db.queryengine.execution.operator.window.ConditionWindowParameter;
import org.apache.iotdb.db.queryengine.execution.operator.window.CountWindowParameter;
import org.apache.iotdb.db.queryengine.execution.operator.window.SessionWindowParameter;
import org.apache.iotdb.db.queryengine.execution.operator.window.TimeWindowParameter;
import org.apache.iotdb.db.queryengine.execution.operator.window.VariationWindowParameter;
import org.apache.iotdb.db.queryengine.execution.operator.window.WindowParameter;
import org.apache.iotdb.db.queryengine.execution.operator.window.WindowType;
import org.apache.iotdb.db.queryengine.plan.Coordinator;
import org.apache.iotdb.db.queryengine.plan.analyze.ExpressionTypeAnalyzer;
import org.apache.iotdb.db.queryengine.plan.analyze.PredicateUtils;
import org.apache.iotdb.db.queryengine.plan.analyze.TemplatedInfo;
import org.apache.iotdb.db.queryengine.plan.analyze.TypeProvider;
import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeSchemaCache;
import org.apache.iotdb.db.queryengine.plan.expression.Expression;
import org.apache.iotdb.db.queryengine.plan.expression.ExpressionFactory;
import org.apache.iotdb.db.queryengine.plan.expression.leaf.TimeSeriesOperand;
import org.apache.iotdb.db.queryengine.plan.expression.leaf.TimestampOperand;
import org.apache.iotdb.db.queryengine.plan.expression.multi.FunctionExpression;
import org.apache.iotdb.db.queryengine.plan.expression.visitor.ColumnTransformerVisitor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.ExplainAnalyzeNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.CountSchemaMergeNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.DevicesCountNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.DevicesSchemaScanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.LevelTimeSeriesCountNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.LogicalViewSchemaScanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.NodeManagementMemoryMergeNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.NodePathsConvertNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.NodePathsCountNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.NodePathsSchemaScanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.PathsUsingTemplateScanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.SchemaFetchMergeNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.SchemaFetchScanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.SchemaQueryMergeNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.SchemaQueryOrderByHeatNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.SchemaQueryScanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.TimeSeriesCountNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.TimeSeriesSchemaScanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.AggregationMergeSortNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.AggregationNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.ColumnInjectNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.DeviceViewIntoNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.DeviceViewNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.ExchangeNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.FillNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.FilterNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.GroupByLevelNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.GroupByTagNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.HorizontallyConcatNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.IntoNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.LimitNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.MergeSortNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.MultiChildProcessNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.OffsetNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.ProjectNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SingleDeviceViewNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SlidingWindowAggregationNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SortNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TopKNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TransformNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TwoChildProcessNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.join.FullOuterTimeJoinNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.join.InnerTimeJoinNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.join.LeftOuterTimeJoinNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.last.LastQueryCollectNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.last.LastQueryMergeNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.last.LastQueryNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.last.LastQueryTransformNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.sink.IdentitySinkNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.sink.ShuffleSinkNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.AlignedLastQueryScanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.AlignedSeriesAggregationScanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.AlignedSeriesScanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.LastQueryScanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.SeriesAggregationScanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.SeriesScanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.SeriesSourceNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.ShowQueriesNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.AggregationDescriptor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.CrossSeriesAggregationDescriptor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.DeviceViewIntoPathDescriptor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.FillDescriptor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.GroupByConditionParameter;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.GroupByCountParameter;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.GroupByParameter;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.GroupBySessionParameter;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.GroupByTimeParameter;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.GroupByVariationParameter;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.InputLocation;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.IntoPathDescriptor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.OutputColumn;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions;
import org.apache.iotdb.db.queryengine.plan.statement.component.FillPolicy;
import org.apache.iotdb.db.queryengine.plan.statement.component.OrderByKey;
import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
import org.apache.iotdb.db.queryengine.plan.statement.component.SortItem;
import org.apache.iotdb.db.queryengine.plan.statement.literal.Literal;
import org.apache.iotdb.db.queryengine.statistics.StatisticsManager;
import org.apache.iotdb.db.queryengine.transformation.dag.column.ColumnTransformer;
import org.apache.iotdb.db.queryengine.transformation.dag.column.leaf.LeafColumnTransformer;
import org.apache.iotdb.db.queryengine.transformation.dag.udf.UDTFContext;
import org.apache.iotdb.db.utils.columngenerator.ColumnGenerator;
import org.apache.iotdb.db.utils.columngenerator.ColumnGeneratorType;
import org.apache.iotdb.db.utils.columngenerator.SlidingTimeColumnGenerator;
import org.apache.iotdb.db.utils.columngenerator.parameter.ColumnGeneratorParameter;
import org.apache.iotdb.db.utils.columngenerator.parameter.SlidingTimeColumnGeneratorParameter;
import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.TimeValuePair;
import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import org.apache.iotdb.tsfile.read.filter.operator.TimeFilterOperators.TimeGt;
import org.apache.iotdb.tsfile.read.filter.operator.TimeFilterOperators.TimeGtEq;
import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.utils.Pair;
import org.apache.iotdb.tsfile.utils.TimeDuration;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.Validate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkState;
import static org.apache.iotdb.db.queryengine.common.DataNodeEndPoints.isSameNode;
import static org.apache.iotdb.db.queryengine.execution.operator.AggregationUtil.calculateMaxAggregationResultSize;
import static org.apache.iotdb.db.queryengine.execution.operator.AggregationUtil.calculateMaxAggregationResultSizeForLastQuery;
import static org.apache.iotdb.db.queryengine.execution.operator.AggregationUtil.getOutputColumnSizePerLine;
import static org.apache.iotdb.db.queryengine.execution.operator.AggregationUtil.initTimeRangeIterator;
import static org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.MergeSortComparator.getComparator;
import static org.apache.iotdb.db.queryengine.plan.analyze.PredicateUtils.convertPredicateToFilter;
import static org.apache.iotdb.db.queryengine.plan.expression.leaf.TimestampOperand.TIMESTAMP_EXPRESSION_STRING;
import static org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.AggregationDescriptor.getAggregationTypeByFuncName;
import static org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions.updateFilterUsingTTL;
import static org.apache.iotdb.db.utils.TimestampPrecisionUtils.TIMESTAMP_PRECISION;

/** This Visitor is responsible for transferring PlanNode Tree to Operator Tree. */
public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionPlanContext> {

  private static final Logger LOGGER = LoggerFactory.getLogger(OperatorTreeGenerator.class);

  private static final MPPDataExchangeManager MPP_DATA_EXCHANGE_MANAGER =
      MPPDataExchangeService.getInstance().getMPPDataExchangeManager();

  private static final DataNodeSchemaCache DATA_NODE_SCHEMA_CACHE =
      DataNodeSchemaCache.getInstance();

  private static final TimeComparator ASC_TIME_COMPARATOR = new AscTimeComparator();

  private static final TimeComparator DESC_TIME_COMPARATOR = new DescTimeComparator();

  private static final IdentityFill IDENTITY_FILL = new IdentityFill();

  private static final IdentityLinearFill IDENTITY_LINEAR_FILL = new IdentityLinearFill();

  private static final Comparator<Binary> ASC_BINARY_COMPARATOR = Comparator.naturalOrder();

  private static final Comparator<Binary> DESC_BINARY_COMPARATOR = Comparator.reverseOrder();

  private static final String UNKNOWN_DATATYPE = "Unknown data type: ";

  @Override
  public Operator visitPlan(PlanNode node, LocalExecutionPlanContext context) {
    throw new UnsupportedOperationException("should call the concrete visitXX() method");
  }

  @Override
  public Operator visitSeriesScan(SeriesScanNode node, LocalExecutionPlanContext context) {
    PartialPath seriesPath = node.getSeriesPath();

    SeriesScanOptions.Builder scanOptionsBuilder = getSeriesScanOptionsBuilder(node, context);
    scanOptionsBuilder.withAllSensors(
        context.getAllSensors(seriesPath.getDevice(), seriesPath.getMeasurement()));
    scanOptionsBuilder.withPushDownLimit(node.getPushDownLimit());
    scanOptionsBuilder.withPushDownOffset(node.getPushDownOffset());

    Expression pushDownPredicate = node.getPushDownPredicate();
    boolean predicateCanPushIntoScan = canPushIntoScan(pushDownPredicate);
    if (pushDownPredicate != null && predicateCanPushIntoScan) {
      scanOptionsBuilder.withPushDownFilter(
          convertPredicateToFilter(
              pushDownPredicate,
              Collections.singletonList(node.getSeriesPath().getMeasurement()),
              context.getTypeProvider().getTemplatedInfo() != null,
              context.getTypeProvider()));
    }

    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                SeriesScanOperator.class.getSimpleName());
    operatorContext.recordSpecifiedInfo("SeriesPath", seriesPath.getFullPath());
    SeriesScanOperator seriesScanOperator =
        new SeriesScanOperator(
            operatorContext,
            node.getPlanNodeId(),
            seriesPath,
            node.getScanOrder(),
            scanOptionsBuilder.build());

    ((DataDriverContext) context.getDriverContext()).addSourceOperator(seriesScanOperator);
    ((DataDriverContext) context.getDriverContext()).addPath(seriesPath);
    context.getDriverContext().setInputDriver(true);

    if (!predicateCanPushIntoScan) {
      checkState(!context.isBuildPlanUseTemplate(), "Push down predicate is not supported yet");
      return constructFilterOperator(
          pushDownPredicate,
          seriesScanOperator,
          Collections.singletonList(ExpressionFactory.timeSeries(node.getSeriesPath()))
              .toArray(new Expression[0]),
          Collections.singletonList(node.getSeriesPath().getSeriesType()),
          makeLayout(Collections.singletonList(node)),
          false,
          node.getPlanNodeId(),
          node.getScanOrder(),
          context);
    }
    return seriesScanOperator;
  }

  @Override
  public Operator visitAlignedSeriesScan(
      AlignedSeriesScanNode node, LocalExecutionPlanContext context) {
    AlignedPath seriesPath = node.getAlignedPath();

    SeriesScanOptions.Builder scanOptionsBuilder = getSeriesScanOptionsBuilder(node, context);
    scanOptionsBuilder.withPushDownLimit(node.getPushDownLimit());
    scanOptionsBuilder.withPushDownOffset(node.getPushDownOffset());
    scanOptionsBuilder.withAllSensors(
        new HashSet<>(
            context.isBuildPlanUseTemplate()
                ? context.getTemplatedInfo().getMeasurementList()
                : seriesPath.getMeasurementList()));

    Expression pushDownPredicate = node.getPushDownPredicate();
    boolean predicateCanPushIntoScan = canPushIntoScan(pushDownPredicate);
    if (pushDownPredicate != null && predicateCanPushIntoScan) {
      scanOptionsBuilder.withPushDownFilter(
          convertPredicateToFilter(
              pushDownPredicate,
              node.getAlignedPath().getMeasurementList(),
              context.getTypeProvider().getTemplatedInfo() != null,
              context.getTypeProvider()));
    }

    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                AlignedSeriesScanOperator.class.getSimpleName());

    int maxTsBlockLineNum = TSFileDescriptor.getInstance().getConfig().getMaxTsBlockLineNumber();
    if (context.getTypeProvider().getTemplatedInfo() != null) {
      maxTsBlockLineNum =
          (int)
              Math.min(
                  context.getTypeProvider().getTemplatedInfo().getLimitValue(), maxTsBlockLineNum);
    }

    AlignedSeriesScanOperator seriesScanOperator =
        new AlignedSeriesScanOperator(
            operatorContext,
            node.getPlanNodeId(),
            seriesPath,
            node.getScanOrder(),
            scanOptionsBuilder.build(),
            node.isQueryAllSensors(),
            context.getTypeProvider().getTemplatedInfo() != null
                ? context.getTypeProvider().getTemplatedInfo().getDataTypes()
                : null,
            maxTsBlockLineNum);

    ((DataDriverContext) context.getDriverContext()).addSourceOperator(seriesScanOperator);
    ((DataDriverContext) context.getDriverContext()).addPath(seriesPath);
    context.getDriverContext().setInputDriver(true);

    if (!predicateCanPushIntoScan) {
      if (context.isBuildPlanUseTemplate()) {
        TemplatedInfo templatedInfo = context.getTemplatedInfo();
        return constructFilterOperator(
            pushDownPredicate,
            seriesScanOperator,
            templatedInfo.getProjectExpressions(),
            templatedInfo.getDataTypes(),
            templatedInfo.getLayoutMap(),
            templatedInfo.isKeepNull(),
            node.getPlanNodeId(),
            templatedInfo.getScanOrder(),
            context);
      }

      AlignedPath alignedPath = node.getAlignedPath();
      List<Expression> expressions = new ArrayList<>();
      List<TSDataType> dataTypes = new ArrayList<>();
      for (int i = 0; i < alignedPath.getMeasurementList().size(); i++) {
        expressions.add(ExpressionFactory.timeSeries(alignedPath.getSubMeasurementPath(i)));
        dataTypes.add(alignedPath.getSubMeasurementDataType(i));
      }

      return constructFilterOperator(
          pushDownPredicate,
          seriesScanOperator,
          expressions.toArray(new Expression[0]),
          dataTypes,
          makeLayout(Collections.singletonList(node)),
          false,
          node.getPlanNodeId(),
          node.getScanOrder(),
          context);
    }
    return seriesScanOperator;
  }

  private boolean canPushIntoScan(Expression pushDownPredicate) {
    return pushDownPredicate == null || PredicateUtils.predicateCanPushIntoScan(pushDownPredicate);
  }

  @Override
  public Operator visitProject(ProjectNode node, LocalExecutionPlanContext context) {
    Operator child = node.getChild().accept(this, context);
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                ProjectOperator.class.getSimpleName());

    List<String> inputColumnNames;
    List<String> outputColumnNames = node.getOutputColumnNames();
    if (outputColumnNames == null) {
      outputColumnNames = context.getTypeProvider().getTemplatedInfo().getSelectMeasurements();
      // skip device column
      outputColumnNames = outputColumnNames.subList(1, outputColumnNames.size());
      inputColumnNames = context.getTypeProvider().getTemplatedInfo().getMeasurementList();
    } else {
      inputColumnNames = node.getChild().getOutputColumnNames();
    }

    if (inputColumnNames.equals(outputColumnNames)) {
      // no need to project
      return child;
    }

    List<Integer> remainingColumnIndexList = new ArrayList<>();
    for (String outputColumnName : outputColumnNames) {
      int index = inputColumnNames.indexOf(outputColumnName);
      if (index < 0) {
        throw new IllegalStateException(
            String.format("Cannot find column [%s] in child's output", outputColumnName));
      }
      remainingColumnIndexList.add(index);
    }
    return new ProjectOperator(operatorContext, child, remainingColumnIndexList);
  }

  @Override
  public Operator visitSeriesAggregationScan(
      SeriesAggregationScanNode node, LocalExecutionPlanContext context) {
    PartialPath seriesPath = node.getSeriesPath();
    boolean ascending = node.getScanOrder() == Ordering.ASC;
    List<AggregationDescriptor> aggregationDescriptors = node.getAggregationDescriptorList();
    List<Aggregator> aggregators = new ArrayList<>();
    aggregationDescriptors.forEach(
        o ->
            aggregators.add(
                new Aggregator(
                    AccumulatorFactory.createAccumulator(
                        o.getAggregationFuncName(),
                        o.getAggregationType(),
                        Collections.singletonList(node.getSeriesPath().getSeriesType()),
                        o.getInputExpressions(),
                        o.getInputAttributes(),
                        ascending,
                        o.getStep().isInputRaw()),
                    o.getStep())));

    GroupByTimeParameter groupByTimeParameter = node.getGroupByTimeParameter();
    ITimeRangeIterator timeRangeIterator =
        initTimeRangeIterator(groupByTimeParameter, ascending, true);
    long maxReturnSize =
        AggregationUtil.calculateMaxAggregationResultSize(
            node.getAggregationDescriptorList(), timeRangeIterator, context.getTypeProvider());

    SeriesScanOptions.Builder scanOptionsBuilder = getSeriesScanOptionsBuilder(node, context);
    scanOptionsBuilder.withAllSensors(
        context.getAllSensors(seriesPath.getDevice(), seriesPath.getMeasurement()));

    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                SeriesAggregationScanOperator.class.getSimpleName());
    SeriesAggregationScanOperator aggregateScanOperator =
        new SeriesAggregationScanOperator(
            node.getPlanNodeId(),
            seriesPath,
            node.getScanOrder(),
            node.isOutputEndTime(),
            scanOptionsBuilder.build(),
            operatorContext,
            aggregators,
            timeRangeIterator,
            node.getGroupByTimeParameter(),
            maxReturnSize);

    ((DataDriverContext) context.getDriverContext()).addSourceOperator(aggregateScanOperator);
    ((DataDriverContext) context.getDriverContext()).addPath(seriesPath);
    context.getDriverContext().setInputDriver(true);
    return aggregateScanOperator;
  }

  @Override
  public Operator visitAlignedSeriesAggregationScan(
      AlignedSeriesAggregationScanNode node, LocalExecutionPlanContext context) {
    AlignedPath seriesPath = node.getAlignedPath();
    boolean ascending = node.getScanOrder() == Ordering.ASC;
    List<Aggregator> aggregators = new ArrayList<>();
    for (AggregationDescriptor descriptor : node.getAggregationDescriptorList()) {
      checkArgument(
          descriptor.getInputExpressions().size() == 1,
          "descriptor's input expression size is not 1");
      Expression expression = descriptor.getInputExpressions().get(0);
      if (expression instanceof TimeSeriesOperand) {
        String inputSeries =
            ((TimeSeriesOperand) (descriptor.getInputExpressions().get(0)))
                .getPath()
                .getMeasurement();
        int seriesIndex = seriesPath.getMeasurementList().indexOf(inputSeries);
        TSDataType seriesDataType =
            seriesPath.getMeasurementSchema().getSubMeasurementsTSDataTypeList().get(seriesIndex);
        aggregators.add(
            new Aggregator(
                AccumulatorFactory.createAccumulator(
                    descriptor.getAggregationFuncName(),
                    descriptor.getAggregationType(),
                    Collections.singletonList(seriesDataType),
                    descriptor.getInputExpressions(),
                    descriptor.getInputAttributes(),
                    ascending,
                    descriptor.getStep().isInputRaw()),
                descriptor.getStep(),
                Collections.singletonList(
                    new InputLocation[] {new InputLocation(0, seriesIndex)})));
      } else if (expression instanceof TimestampOperand) {
        aggregators.add(
            new Aggregator(
                AccumulatorFactory.createAccumulator(
                    descriptor.getAggregationFuncName(),
                    descriptor.getAggregationType(),
                    Collections.singletonList(TSDataType.INT64),
                    descriptor.getInputExpressions(),
                    descriptor.getInputAttributes(),
                    ascending,
                    descriptor.getStep().isInputRaw()),
                descriptor.getStep(),
                Collections.singletonList(new InputLocation[] {new InputLocation(0, -1)})));
      } else {
        throw new IllegalArgumentException(
            "descriptor's input expression must be TimeSeriesOperand/TimestampOperand, current is "
                + expression);
      }
    }

    GroupByTimeParameter groupByTimeParameter = node.getGroupByTimeParameter();
    ITimeRangeIterator timeRangeIterator =
        initTimeRangeIterator(groupByTimeParameter, ascending, true);
    long maxReturnSize =
        AggregationUtil.calculateMaxAggregationResultSize(
            node.getAggregationDescriptorList(), timeRangeIterator, context.getTypeProvider());

    SeriesScanOptions.Builder scanOptionsBuilder = getSeriesScanOptionsBuilder(node, context);
    scanOptionsBuilder.withAllSensors(new HashSet<>(seriesPath.getMeasurementList()));

    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                AlignedSeriesAggregationScanOperator.class.getSimpleName());
    AlignedSeriesAggregationScanOperator seriesAggregationScanOperator =
        new AlignedSeriesAggregationScanOperator(
            node.getPlanNodeId(),
            seriesPath,
            node.getScanOrder(),
            node.isOutputEndTime(),
            scanOptionsBuilder.build(),
            operatorContext,
            aggregators,
            timeRangeIterator,
            groupByTimeParameter,
            maxReturnSize);

    ((DataDriverContext) context.getDriverContext())
        .addSourceOperator(seriesAggregationScanOperator);
    ((DataDriverContext) context.getDriverContext()).addPath(seriesPath);
    context.getDriverContext().setInputDriver(true);
    return seriesAggregationScanOperator;
  }

  private SeriesScanOptions.Builder getSeriesScanOptionsBuilder(
      SeriesSourceNode node, LocalExecutionPlanContext context) {
    SeriesScanOptions.Builder scanOptionsBuilder = new SeriesScanOptions.Builder();

    Filter globalTimeFilter = context.getGlobalTimeFilter();
    if (globalTimeFilter != null) {
      // time filter may be stateful, so we need to copy it
      scanOptionsBuilder.withGlobalTimeFilter(globalTimeFilter.copy());
    }

    return scanOptionsBuilder;
  }

  @Override
  public Operator visitSchemaQueryOrderByHeat(
      SchemaQueryOrderByHeatNode node, LocalExecutionPlanContext context) {
    List<Operator> children =
        node.getChildren().stream().map(n -> n.accept(this, context)).collect(Collectors.toList());

    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                SchemaQueryOrderByHeatOperator.class.getSimpleName());

    return new SchemaQueryOrderByHeatOperator(operatorContext, children);
  }

  @Override
  public Operator visitSchemaQueryScan(
      SchemaQueryScanNode node, LocalExecutionPlanContext context) {
    if (node instanceof TimeSeriesSchemaScanNode) {
      return visitTimeSeriesSchemaScan((TimeSeriesSchemaScanNode) node, context);
    } else if (node instanceof DevicesSchemaScanNode) {
      return visitDevicesSchemaScan((DevicesSchemaScanNode) node, context);
    } else if (node instanceof DevicesCountNode) {
      return visitDevicesCount((DevicesCountNode) node, context);
    } else if (node instanceof TimeSeriesCountNode) {
      return visitTimeSeriesCount((TimeSeriesCountNode) node, context);
    } else if (node instanceof LevelTimeSeriesCountNode) {
      return visitLevelTimeSeriesCount((LevelTimeSeriesCountNode) node, context);
    } else if (node instanceof NodePathsSchemaScanNode) {
      return visitNodePathsSchemaScan((NodePathsSchemaScanNode) node, context);
    } else if (node instanceof PathsUsingTemplateScanNode) {
      return visitPathsUsingTemplateScan((PathsUsingTemplateScanNode) node, context);
    } else if (node instanceof LogicalViewSchemaScanNode) {
      return visitLogicalViewSchemaScan((LogicalViewSchemaScanNode) node, context);
    }
    return visitPlan(node, context);
  }

  @Override
  public Operator visitTimeSeriesSchemaScan(
      TimeSeriesSchemaScanNode node, LocalExecutionPlanContext context) {
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                SchemaQueryScanOperator.class.getSimpleName());
    return new SchemaQueryScanOperator<>(
        node.getPlanNodeId(),
        operatorContext,
        SchemaSourceFactory.getTimeSeriesSchemaScanSource(
            node.getPath(),
            node.isPrefixPath(),
            node.getLimit(),
            node.getOffset(),
            node.getSchemaFilter(),
            node.getTemplateMap(),
            node.getScope()));
  }

  @Override
  public Operator visitDevicesSchemaScan(
      DevicesSchemaScanNode node, LocalExecutionPlanContext context) {
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                SchemaQueryScanOperator.class.getSimpleName());
    return new SchemaQueryScanOperator<>(
        node.getPlanNodeId(),
        operatorContext,
        SchemaSourceFactory.getDeviceSchemaSource(
            node.getPath(),
            node.isPrefixPath(),
            node.getLimit(),
            node.getOffset(),
            node.isHasSgCol(),
            node.getSchemaFilter(),
            node.getScope()));
  }

  @Override
  public Operator visitSchemaQueryMerge(
      SchemaQueryMergeNode node, LocalExecutionPlanContext context) {
    List<Operator> children = dealWithConsumeChildrenOneByOneNode(node, context);
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                SchemaQueryMergeOperator.class.getSimpleName());
    return new SchemaQueryMergeOperator(operatorContext, children);
  }

  @Override
  public Operator visitCountMerge(CountSchemaMergeNode node, LocalExecutionPlanContext context) {
    List<Operator> children = dealWithConsumeChildrenOneByOneNode(node, context);
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                CountMergeOperator.class.getSimpleName());
    if (node.getChildren().get(0) instanceof LevelTimeSeriesCountNode) {
      return new CountGroupByLevelMergeOperator(operatorContext, children);
    } else {
      return new CountMergeOperator(operatorContext, children);
    }
  }

  @Override
  public Operator visitDevicesCount(DevicesCountNode node, LocalExecutionPlanContext context) {
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                SchemaCountOperator.class.getSimpleName());
    return new SchemaCountOperator<>(
        node.getPlanNodeId(),
        operatorContext,
        SchemaSourceFactory.getDeviceSchemaSource(
            node.getPath(), node.isPrefixPath(), node.getScope()));
  }

  @Override
  public Operator visitTimeSeriesCount(
      TimeSeriesCountNode node, LocalExecutionPlanContext context) {
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                SchemaCountOperator.class.getSimpleName());
    return new SchemaCountOperator<>(
        node.getPlanNodeId(),
        operatorContext,
        SchemaSourceFactory.getTimeSeriesSchemaCountSource(
            node.getPath(),
            node.isPrefixPath(),
            node.getSchemaFilter(),
            node.getTemplateMap(),
            node.getScope()));
  }

  @Override
  public Operator visitLevelTimeSeriesCount(
      LevelTimeSeriesCountNode node, LocalExecutionPlanContext context) {
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                CountGroupByLevelScanOperator.class.getSimpleName());
    return new CountGroupByLevelScanOperator<>(
        node.getPlanNodeId(),
        operatorContext,
        node.getLevel(),
        SchemaSourceFactory.getTimeSeriesSchemaCountSource(
            node.getPath(),
            node.isPrefixPath(),
            node.getSchemaFilter(),
            node.getTemplateMap(),
            node.getScope()));
  }

  @Override
  public Operator visitNodePathsSchemaScan(
      NodePathsSchemaScanNode node, LocalExecutionPlanContext context) {
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                SchemaQueryScanOperator.class.getSimpleName());
    return new SchemaQueryScanOperator<>(
        node.getPlanNodeId(),
        operatorContext,
        SchemaSourceFactory.getNodeSchemaSource(
            node.getPrefixPath(), node.getLevel(), node.getScope()));
  }

  @Override
  public Operator visitNodeManagementMemoryMerge(
      NodeManagementMemoryMergeNode node, LocalExecutionPlanContext context) {
    Operator child = node.getChild().accept(this, context);
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                NodeManageMemoryMergeOperator.class.getSimpleName());
    return new NodeManageMemoryMergeOperator(operatorContext, node.getData(), child);
  }

  @Override
  public Operator visitNodePathConvert(
      NodePathsConvertNode node, LocalExecutionPlanContext context) {
    Operator child = node.getChild().accept(this, context);
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                NodePathsConvertOperator.class.getSimpleName());
    return new NodePathsConvertOperator(operatorContext, child);
  }

  @Override
  public Operator visitNodePathsCount(NodePathsCountNode node, LocalExecutionPlanContext context) {
    Operator child = node.getChild().accept(this, context);
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                NodePathsCountOperator.class.getSimpleName());
    return new NodePathsCountOperator(operatorContext, child);
  }

  @Override
  public Operator visitSingleDeviceView(
      SingleDeviceViewNode node, LocalExecutionPlanContext context) {
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                SingleDeviceViewOperator.class.getSimpleName());
    Operator child = node.getChild().accept(this, context);
    List<Integer> deviceColumnIndex = node.getDeviceToMeasurementIndexes();
    List<TSDataType> outputColumnTypes =
        node.isCacheOutputColumnNames()
            ? getOutputColumnTypes(node, context.getTypeProvider())
            : context.getCachedDataTypes();
    if (outputColumnTypes == null || outputColumnTypes.isEmpty()) {
      throw new IllegalStateException("OutputColumTypes should not be null/empty");
    }
    return new SingleDeviceViewOperator(
        operatorContext, node.getDevice(), child, deviceColumnIndex, outputColumnTypes);
  }

  @Override
  public Operator visitDeviceView(DeviceViewNode node, LocalExecutionPlanContext context) {
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                DeviceViewOperator.class.getSimpleName());
    List<Operator> children = dealWithConsumeChildrenOneByOneNode(node, context);
    List<List<Integer>> deviceColumnIndex =
        node.getDevices().stream()
            .map(deviceName -> node.getDeviceToMeasurementIndexesMap().get(deviceName))
            .collect(Collectors.toList());
    List<TSDataType> outputColumnTypes = getOutputColumnTypes(node, context.getTypeProvider());

    return new DeviceViewOperator(
        operatorContext, node.getDevices(), children, deviceColumnIndex, outputColumnTypes);
  }

  @Override
  public Operator visitMergeSort(MergeSortNode node, LocalExecutionPlanContext context) {
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                MergeSortOperator.class.getSimpleName());
    List<TSDataType> dataTypes = getOutputColumnTypes(node, context.getTypeProvider());
    context.setCachedDataTypes(dataTypes);
    List<Operator> children = dealWithConsumeAllChildrenPipelineBreaker(node, context);
    List<SortItem> sortItemList = node.getMergeOrderParameter().getSortItemList();

    List<Integer> sortItemIndexList = new ArrayList<>(sortItemList.size());
    List<TSDataType> sortItemDataTypeList = new ArrayList<>(sortItemList.size());
    genSortInformation(
        node.getOutputColumnNames(),
        dataTypes,
        sortItemList,
        sortItemIndexList,
        sortItemDataTypeList);
    return new MergeSortOperator(
        operatorContext,
        children,
        dataTypes,
        getComparator(sortItemList, sortItemIndexList, sortItemDataTypeList));
  }

  @Override
  public Operator visitAggregationMergeSort(
      AggregationMergeSortNode node, LocalExecutionPlanContext context) {
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                MergeSortOperator.class.getSimpleName());
    List<TSDataType> dataTypes = getOutputColumnTypes(node, context.getTypeProvider());
    List<Operator> children = dealWithConsumeAllChildrenPipelineBreaker(node, context);

    List<SortItem> sortItemList = node.getMergeOrderParameter().getSortItemList();
    if (!sortItemList.get(0).getSortKey().equalsIgnoreCase("Device")) {
      throw new IllegalArgumentException(
          "Only order by device align by device support AggregationMergeSortNode.");
    }

    boolean timeAscending = true;
    for (SortItem sortItem : sortItemList) {
      if (TIMESTAMP_EXPRESSION_STRING.equalsIgnoreCase(sortItem.getSortKey())
          && (sortItem.getOrdering() == Ordering.DESC)) {
        timeAscending = false;
        break;
      }
    }

    List<Accumulator> accumulators = new ArrayList<>();
    for (Expression expression : node.getSelectExpressions()) {
      if (expression instanceof FunctionExpression) {
        FunctionExpression functionExpression = (FunctionExpression) expression;
        String aggregationName = functionExpression.getFunctionName();
        Accumulator accumulator =
            AccumulatorFactory.createAccumulator(
                aggregationName,
                getAggregationTypeByFuncName(aggregationName),
                Collections.singletonList(
                    context.getTypeProvider().getType(functionExpression.getOutputSymbol())),
                functionExpression.getExpressions(),
                functionExpression.getFunctionAttributes(),
                timeAscending,
                false);
        accumulators.add(accumulator);
      }
    }

    List<Integer> sortItemIndexList = new ArrayList<>(sortItemList.size());
    List<TSDataType> sortItemDataTypeList = new ArrayList<>(sortItemList.size());
    genSortInformation(
        node.getOutputColumnNames(),
        dataTypes,
        sortItemList,
        sortItemIndexList,
        sortItemDataTypeList);
    return new AggregationMergeSortOperator(
        operatorContext,
        children,
        dataTypes,
        accumulators,
        node.isHasGroupBy(),
        getComparator(sortItemList, sortItemIndexList, sortItemDataTypeList));
  }

  @Override
  public Operator visitTopK(TopKNode node, LocalExecutionPlanContext context) {
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                TopKOperator.class.getSimpleName());
    List<TSDataType> dataTypes = getOutputColumnTypes(node, context.getTypeProvider());
    context.setCachedDataTypes(dataTypes);
    List<Operator> children = dealWithConsumeAllChildrenPipelineBreaker(node, context);
    List<SortItem> sortItemList = node.getMergeOrderParameter().getSortItemList();

    List<Integer> sortItemIndexList = new ArrayList<>(sortItemList.size());
    List<TSDataType> sortItemDataTypeList = new ArrayList<>(sortItemList.size());
    genSortInformation(
        node.getOutputColumnNames(),
        dataTypes,
        sortItemList,
        sortItemIndexList,
        sortItemDataTypeList);
    return new TopKOperator(
        operatorContext,
        children,
        dataTypes,
        getComparator(sortItemList, sortItemIndexList, sortItemDataTypeList),
        node.getTopValue(),
        !sortItemList.isEmpty()
            && sortItemList.get(0).getSortKey().equalsIgnoreCase(OrderByKey.TIME)
            && sortItemList.stream()
                .allMatch(
                    i ->
                        (i.getSortKey().equals(OrderByKey.TIME)
                            || i.getSortKey().equals(OrderByKey.DEVICE))));
  }

  private void genSortInformation(
      List<String> outputColumnNames,
      List<TSDataType> dataTypes,
      List<SortItem> sortItemList,
      List<Integer> sortItemIndexList,
      List<TSDataType> sortItemDataTypeList) {
    sortItemList.forEach(
        sortItem -> {
          if (sortItem.getSortKey().equals(OrderByKey.TIME)) {
            sortItemIndexList.add(-1);
            sortItemDataTypeList.add(TSDataType.INT64);
          } else {
            for (int i = 0; i < outputColumnNames.size(); i++) {
              if (sortItem.getSortKey().equalsIgnoreCase(outputColumnNames.get(i))) {
                sortItemIndexList.add(i);
                sortItemDataTypeList.add(dataTypes.get(i));
                break;
              }
              // there is no related column in outputColumnNames
              if (i == outputColumnNames.size() - 1) {
                sortItemIndexList.add(-2);
                sortItemDataTypeList.add(null);
              }
            }
          }
        });
  }

  @Override
  public Operator visitFill(FillNode node, LocalExecutionPlanContext context) {
    Operator child = node.getChild().accept(this, context);
    return getFillOperator(node, context, child);
  }

  private ProcessOperator getFillOperator(
      FillNode node, LocalExecutionPlanContext context, Operator child) {
    FillDescriptor descriptor = node.getFillDescriptor();
    List<TSDataType> inputDataTypes =
        getOutputColumnTypes(node.getChild(), context.getTypeProvider());
    int inputColumns = inputDataTypes.size();
    FillPolicy fillPolicy = descriptor.getFillPolicy();
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                FillOperator.class.getSimpleName());
    switch (fillPolicy) {
      case VALUE:
        Literal literal = descriptor.getFillValue();
        return new FillOperator(
            operatorContext, getConstantFill(inputColumns, inputDataTypes, literal), child);
      case PREVIOUS:
        return new FillOperator(
            operatorContext,
            getPreviousFill(
                inputColumns,
                inputDataTypes,
                descriptor.getTimeDurationThreshold(),
                context.getZoneId()),
            child);
      case LINEAR:
        return new LinearFillOperator(
            operatorContext, getLinearFill(inputColumns, inputDataTypes), child);
      default:
        throw new IllegalArgumentException("Unknown fill policy: " + fillPolicy);
    }
  }

  private IFill[] getConstantFill(
      int inputColumns, List<TSDataType> inputDataTypes, Literal literal) {
    IFill[] constantFill = new IFill[inputColumns];
    for (int i = 0; i < inputColumns; i++) {
      if (!literal.isDataTypeConsistency(inputDataTypes.get(i))) {
        constantFill[i] = IDENTITY_FILL;
        continue;
      }
      switch (inputDataTypes.get(i)) {
        case BOOLEAN:
          constantFill[i] = new BooleanConstantFill(literal.getBoolean());
          break;
        case TEXT:
          constantFill[i] = new BinaryConstantFill(literal.getBinary());
          break;
        case INT32:
          constantFill[i] = new IntConstantFill(literal.getInt());
          break;
        case INT64:
          constantFill[i] = new LongConstantFill(literal.getLong());
          break;
        case FLOAT:
          constantFill[i] = new FloatConstantFill(literal.getFloat());
          break;
        case DOUBLE:
          constantFill[i] = new DoubleConstantFill(literal.getDouble());
          break;
        default:
          throw new IllegalArgumentException(UNKNOWN_DATATYPE + inputDataTypes.get(i));
      }
    }
    return constantFill;
  }

  private IFill[] getPreviousFill(
      int inputColumns,
      List<TSDataType> inputDataTypes,
      TimeDuration timeDurationThreshold,
      ZoneId zoneId) {
    IFillFilter filter;
    if (timeDurationThreshold == null) {
      filter = IFillFilter.TRUE;
    } else if (!timeDurationThreshold.containsMonth()) {
      filter = new FixedIntervalFillFilter(timeDurationThreshold.nonMonthDuration);
    } else {
      switch (TIMESTAMP_PRECISION) {
        case "ms":
          filter =
              new MonthIntervalMSFillFilter(
                  timeDurationThreshold.monthDuration,
                  timeDurationThreshold.nonMonthDuration,
                  zoneId);
          break;
        case "us":
          filter =
              new MonthIntervalUSFillFilter(
                  timeDurationThreshold.monthDuration,
                  timeDurationThreshold.nonMonthDuration,
                  zoneId);
          break;
        case "ns":
          filter =
              new MonthIntervalNSFillFilter(
                  timeDurationThreshold.monthDuration,
                  timeDurationThreshold.nonMonthDuration,
                  zoneId);
          break;
        default:
          // this case will never reach
          throw new UnsupportedOperationException(
              "not supported time_precision: " + TIMESTAMP_PRECISION);
      }
    }

    IFill[] previousFill = new IFill[inputColumns];
    for (int i = 0; i < inputColumns; i++) {
      switch (inputDataTypes.get(i)) {
        case BOOLEAN:
          previousFill[i] = new BooleanPreviousFill(filter);
          break;
        case TEXT:
          previousFill[i] = new BinaryPreviousFill(filter);
          break;
        case INT32:
          previousFill[i] = new IntPreviousFill(filter);
          break;
        case INT64:
          previousFill[i] = new LongPreviousFill(filter);
          break;
        case FLOAT:
          previousFill[i] = new FloatPreviousFill(filter);
          break;
        case DOUBLE:
          previousFill[i] = new DoublePreviousFill(filter);
          break;
        default:
          throw new IllegalArgumentException(UNKNOWN_DATATYPE + inputDataTypes.get(i));
      }
    }
    return previousFill;
  }

  private ILinearFill[] getLinearFill(int inputColumns, List<TSDataType> inputDataTypes) {
    ILinearFill[] linearFill = new ILinearFill[inputColumns];
    for (int i = 0; i < inputColumns; i++) {
      switch (inputDataTypes.get(i)) {
        case INT32:
          linearFill[i] = new IntLinearFill();
          break;
        case INT64:
          linearFill[i] = new LongLinearFill();
          break;
        case FLOAT:
          linearFill[i] = new FloatLinearFill();
          break;
        case DOUBLE:
          linearFill[i] = new DoubleLinearFill();
          break;
        case BOOLEAN:
        case TEXT:
          linearFill[i] = IDENTITY_LINEAR_FILL;
          break;
        default:
          throw new IllegalArgumentException(UNKNOWN_DATATYPE + inputDataTypes.get(i));
      }
    }
    return linearFill;
  }

  @Override
  public Operator visitTransform(TransformNode node, LocalExecutionPlanContext context) {
    final OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                TransformOperator.class.getSimpleName());
    final Operator inputOperator = generateOnlyChildOperator(node, context);
    final List<TSDataType> inputDataTypes = getInputColumnTypes(node, context.getTypeProvider());
    final Map<String, List<InputLocation>> inputLocations = makeLayout(node);
    final Expression[] projectExpressions = node.getOutputExpressions();
    final Map<NodeRef<Expression>, TSDataType> expressionTypes = new HashMap<>();

    for (Expression projectExpression : projectExpressions) {
      ExpressionTypeAnalyzer.analyzeExpression(expressionTypes, projectExpression);
    }

    boolean hasNonMappableUDF = false;
    for (Expression expression : projectExpressions) {
      if (!expression.isMappable(expressionTypes)) {
        hasNonMappableUDF = true;
        break;
      }
    }

    // Use FilterAndProject Operator when project expressions are all mappable
    if (!hasNonMappableUDF) {
      // init project UDTFContext
      UDTFContext projectContext = new UDTFContext(context.getZoneId());
      projectContext.constructUdfExecutors(projectExpressions);

      List<ColumnTransformer> projectOutputTransformerList = new ArrayList<>();
      Map<Expression, ColumnTransformer> projectExpressionColumnTransformerMap = new HashMap<>();

      // records LeafColumnTransformer of project expressions
      List<LeafColumnTransformer> projectLeafColumnTransformerList = new ArrayList<>();

      ColumnTransformerVisitor visitor = new ColumnTransformerVisitor();
      ColumnTransformerVisitor.ColumnTransformerVisitorContext projectColumnTransformerContext =
          new ColumnTransformerVisitor.ColumnTransformerVisitorContext(
              projectContext,
              expressionTypes,
              projectLeafColumnTransformerList,
              inputLocations,
              projectExpressionColumnTransformerMap,
              ImmutableMap.of(),
              ImmutableList.of(),
              inputDataTypes,
              inputLocations.size() - 1,
              null);

      for (Expression expression : projectExpressions) {
        projectOutputTransformerList.add(
            visitor.process(expression, projectColumnTransformerContext));
      }

      return new FilterAndProjectOperator(
          operatorContext,
          inputOperator,
          inputDataTypes,
          ImmutableList.of(),
          null,
          ImmutableList.of(),
          projectLeafColumnTransformerList,
          projectOutputTransformerList,
          false,
          false);
    }

    try {
      return new TransformOperator(
          operatorContext,
          inputOperator,
          inputDataTypes,
          inputLocations,
          node.getOutputExpressions(),
          node.isKeepNull(),
          context.getZoneId(),
          expressionTypes,
          node.getScanOrder() == Ordering.ASC);
    } catch (QueryProcessException e) {
      throw new RuntimeException(e);
    }
  }

  @Override
  public Operator visitFilter(FilterNode node, LocalExecutionPlanContext context) {
    if (context.isBuildPlanUseTemplate()) {
      TemplatedInfo templatedInfo = context.getTemplatedInfo();
      return constructFilterOperator(
          node.getPredicate(),
          generateOnlyChildOperator(node, context),
          templatedInfo.getProjectExpressions(),
          templatedInfo.getDataTypes(),
          templatedInfo.getLayoutMap(),
          templatedInfo.isKeepNull(),
          node.getPlanNodeId(),
          templatedInfo.getScanOrder(),
          context);
    }

    return constructFilterOperator(
        node.getPredicate(),
        generateOnlyChildOperator(node, context),
        node.getOutputExpressions(),
        getInputColumnTypes(node, context.getTypeProvider()),
        makeLayout(node),
        node.isKeepNull(),
        node.getPlanNodeId(),
        node.getScanOrder(),
        context);
  }

  private Operator constructFilterOperator(
      Expression predicate,
      Operator inputOperator,
      Expression[] projectExpressions,
      List<TSDataType> inputDataTypes,
      Map<String, List<InputLocation>> inputLocations,
      boolean isKeepNull,
      PlanNodeId planNodeId,
      Ordering scanOrder,
      LocalExecutionPlanContext context) {
    final Map<NodeRef<Expression>, TSDataType> expressionTypes = new HashMap<>();
    if (context.isBuildPlanUseTemplate()) {
      ExpressionTypeAnalyzer.analyzeExpressionUsingTemplatedInfo(
          expressionTypes, predicate, context.getTypeProvider().getTemplatedInfo());
    } else {
      ExpressionTypeAnalyzer.analyzeExpression(expressionTypes, predicate);
    }

    // check whether predicate contains Non-Mappable UDF
    if (!predicate.isMappable(expressionTypes)) {
      throw new UnsupportedOperationException("Filter can not contain Non-Mappable UDF");
    }

    final List<TSDataType> filterOutputDataTypes = new ArrayList<>(inputDataTypes);

    for (Expression projectExpression : projectExpressions) {
      if (context.isBuildPlanUseTemplate()) {
        ExpressionTypeAnalyzer.analyzeExpressionUsingTemplatedInfo(
            expressionTypes, projectExpression, context.getTypeProvider().getTemplatedInfo());
      } else {
        ExpressionTypeAnalyzer.analyzeExpression(expressionTypes, projectExpression);
      }
    }

    boolean hasNonMappableUdf = false;
    for (Expression expression : projectExpressions) {
      if (!expression.isMappable(expressionTypes)) {
        hasNonMappableUdf = true;
        break;
      }
    }

    // init UDTFContext
    UDTFContext filterContext = new UDTFContext(context.getZoneId());
    filterContext.constructUdfExecutors(new Expression[] {predicate});

    // records LeafColumnTransformer of filter
    List<LeafColumnTransformer> filterLeafColumnTransformerList = new ArrayList<>();

    // records common ColumnTransformer between filter and project expressions
    List<ColumnTransformer> commonTransformerList = new ArrayList<>();

    // records LeafColumnTransformer of project expressions
    List<LeafColumnTransformer> projectLeafColumnTransformerList = new ArrayList<>();

    // records subexpression -> ColumnTransformer for filter
    Map<Expression, ColumnTransformer> filterExpressionColumnTransformerMap = new HashMap<>();

    ColumnTransformerVisitor visitor = new ColumnTransformerVisitor();

    ColumnTransformerVisitor.ColumnTransformerVisitorContext filterColumnTransformerContext =
        new ColumnTransformerVisitor.ColumnTransformerVisitorContext(
            filterContext,
            expressionTypes,
            filterLeafColumnTransformerList,
            inputLocations,
            filterExpressionColumnTransformerMap,
            ImmutableMap.of(),
            ImmutableList.of(),
            ImmutableList.of(),
            0,
            null);

    ColumnTransformer filterOutputTransformer =
        visitor.process(predicate, filterColumnTransformerContext);

    List<ColumnTransformer> projectOutputTransformerList = new ArrayList<>();

    Map<Expression, ColumnTransformer> projectExpressionColumnTransformerMap = new HashMap<>();

    // init project transformer when project expressions are all mappable
    if (!hasNonMappableUdf) {
      // init project UDTFContext
      UDTFContext projectContext = new UDTFContext(context.getZoneId());
      projectContext.constructUdfExecutors(projectExpressions);

      ColumnTransformerVisitor.ColumnTransformerVisitorContext projectColumnTransformerContext =
          new ColumnTransformerVisitor.ColumnTransformerVisitorContext(
              projectContext,
              expressionTypes,
              projectLeafColumnTransformerList,
              inputLocations,
              projectExpressionColumnTransformerMap,
              filterExpressionColumnTransformerMap,
              commonTransformerList,
              filterOutputDataTypes,
              inputLocations.size() - 1,
              null);

      for (Expression expression : projectExpressions) {
        projectOutputTransformerList.add(
            visitor.process(expression, projectColumnTransformerContext));
      }
    }

    final OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                planNodeId,
                FilterAndProjectOperator.class.getSimpleName());
    Operator filter =
        new FilterAndProjectOperator(
            operatorContext,
            inputOperator,
            filterOutputDataTypes,
            filterLeafColumnTransformerList,
            filterOutputTransformer,
            commonTransformerList,
            projectLeafColumnTransformerList,
            projectOutputTransformerList,
            hasNonMappableUdf,
            true);

    // Project expressions don't contain Non-Mappable UDF, TransformOperator is not needed
    if (!hasNonMappableUdf) {
      return filter;
    }

    // has Non-Mappable UDF, we wrap a TransformOperator for further calculation
    try {
      final OperatorContext transformContext =
          context
              .getDriverContext()
              .addOperatorContext(
                  context.getNextOperatorId(), planNodeId, TransformOperator.class.getSimpleName());
      return new TransformOperator(
          transformContext,
          filter,
          inputDataTypes,
          inputLocations,
          projectExpressions,
          isKeepNull,
          context.getZoneId(),
          expressionTypes,
          scanOrder == Ordering.ASC);
    } catch (QueryProcessException e) {
      throw new RuntimeException(e);
    }
  }

  @Override
  public Operator visitGroupByLevel(GroupByLevelNode node, LocalExecutionPlanContext context) {
    checkArgument(
        !node.getGroupByLevelDescriptors().isEmpty(),
        "GroupByLevel descriptorList cannot be empty");
    List<Operator> children = dealWithConsumeAllChildrenPipelineBreaker(node, context);
    boolean ascending = node.getScanOrder() == Ordering.ASC;
    List<Aggregator> aggregators = new ArrayList<>();
    Map<String, List<InputLocation>> layout = makeLayout(node);
    List<CrossSeriesAggregationDescriptor> aggregationDescriptors =
        node.getGroupByLevelDescriptors();
    for (CrossSeriesAggregationDescriptor descriptor : aggregationDescriptors) {
      List<InputLocation[]> inputLocationList = calcInputLocationList(descriptor, layout);
      // Use the first set of InputExpression
      List<TSDataType> inputDataTypes =
          IntStream.range(0, descriptor.getExpressionNumOfOneInput())
              .mapToObj(
                  x ->
                      context
                          .getTypeProvider()
                          .getType(descriptor.getInputExpressions().get(x).getExpressionString()))
              .collect(Collectors.toList());
      aggregators.add(
          new Aggregator(
              AccumulatorFactory.createAccumulator(
                  descriptor.getAggregationFuncName(),
                  descriptor.getAggregationType(),
                  inputDataTypes,
                  descriptor.getInputExpressions(),
                  descriptor.getInputAttributes(),
                  ascending,
                  descriptor.getStep().isInputRaw()),
              descriptor.getStep(),
              inputLocationList));
    }
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                AggregationOperator.class.getSimpleName());

    GroupByTimeParameter groupByTimeParameter = node.getGroupByTimeParameter();
    ITimeRangeIterator timeRangeIterator =
        initTimeRangeIterator(groupByTimeParameter, ascending, false);
    long maxReturnSize =
        calculateMaxAggregationResultSize(
            aggregationDescriptors, timeRangeIterator, context.getTypeProvider());

    return new AggregationOperator(
        operatorContext, aggregators, timeRangeIterator, children, false, maxReturnSize);
  }

  @Override
  public Operator visitGroupByTag(GroupByTagNode node, LocalExecutionPlanContext context) {
    checkArgument(!node.getTagKeys().isEmpty(), "GroupByTag tag keys cannot be empty");
    checkArgument(
        node.getTagValuesToAggregationDescriptors().size() >= 1,
        "GroupByTag aggregation descriptors cannot be empty");

    List<Operator> children = dealWithConsumeAllChildrenPipelineBreaker(node, context);

    boolean ascending = node.getScanOrder() == Ordering.ASC;
    Map<String, List<InputLocation>> layout = makeLayout(node);
    List<List<String>> groups = new ArrayList<>();
    List<List<Aggregator>> groupedAggregators = new ArrayList<>();
    int aggregatorCount = 0;
    for (Map.Entry<List<String>, List<CrossSeriesAggregationDescriptor>> entry :
        node.getTagValuesToAggregationDescriptors().entrySet()) {
      groups.add(entry.getKey());
      List<Aggregator> aggregators = new ArrayList<>();
      for (CrossSeriesAggregationDescriptor aggregationDescriptor : entry.getValue()) {
        if (aggregationDescriptor == null) {
          aggregators.add(null);
          continue;
        }
        List<InputLocation[]> inputLocations = calcInputLocationList(aggregationDescriptor, layout);
        List<TSDataType> inputDataTypes =
            aggregationDescriptor.getInputExpressions().stream()
                .map(x -> context.getTypeProvider().getType(x.getExpressionString()))
                .collect(Collectors.toList());
        aggregators.add(
            new Aggregator(
                AccumulatorFactory.createAccumulator(
                    aggregationDescriptor.getAggregationFuncName(),
                    aggregationDescriptor.getAggregationType(),
                    inputDataTypes,
                    aggregationDescriptor.getInputExpressions(),
                    aggregationDescriptor.getInputAttributes(),
                    ascending,
                    aggregationDescriptor.getStep().isInputRaw()),
                aggregationDescriptor.getStep(),
                inputLocations));
      }
      groupedAggregators.add(aggregators);
      aggregatorCount += aggregators.size();
    }
    GroupByTimeParameter groupByTimeParameter = node.getGroupByTimeParameter();
    ITimeRangeIterator timeRangeIterator =
        initTimeRangeIterator(groupByTimeParameter, ascending, false);
    List<AggregationDescriptor> aggregationDescriptors =
        node.getTagValuesToAggregationDescriptors().values().stream()
            .flatMap(Collection::stream)
            .filter(Objects::nonNull)
            .collect(Collectors.toList());

    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                TagAggregationOperator.class.getSimpleName());
    long maxReturnSize =
        calculateMaxAggregationResultSize(
            aggregationDescriptors, timeRangeIterator, context.getTypeProvider());

    return new TagAggregationOperator(
        operatorContext, groups, groupedAggregators, children, maxReturnSize);
  }

  @Override
  public Operator visitSlidingWindowAggregation(
      SlidingWindowAggregationNode node, LocalExecutionPlanContext context) {
    checkArgument(
        !node.getAggregationDescriptorList().isEmpty(),
        "Aggregation descriptorList cannot be empty");
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                SlidingWindowAggregationOperator.class.getSimpleName());
    Operator child = node.getChild().accept(this, context);
    boolean ascending = node.getScanOrder() == Ordering.ASC;
    List<Aggregator> aggregators = new ArrayList<>();
    Map<String, List<InputLocation>> layout = makeLayout(node);
    List<AggregationDescriptor> aggregationDescriptors = node.getAggregationDescriptorList();
    for (AggregationDescriptor descriptor : aggregationDescriptors) {
      List<InputLocation[]> inputLocationList = calcInputLocationList(descriptor, layout);
      aggregators.add(
          SlidingWindowAggregatorFactory.createSlidingWindowAggregator(
              descriptor.getAggregationFuncName(),
              descriptor.getAggregationType(),
              descriptor.getInputExpressions().stream()
                  .map(x -> context.getTypeProvider().getType(x.getExpressionString()))
                  .collect(Collectors.toList()),
              descriptor.getInputExpressions(),
              descriptor.getInputAttributes(),
              ascending,
              inputLocationList,
              descriptor.getStep()));
    }

    GroupByTimeParameter groupByTimeParameter = node.getGroupByTimeParameter();
    ITimeRangeIterator timeRangeIterator =
        initTimeRangeIterator(groupByTimeParameter, ascending, false);
    long maxReturnSize =
        calculateMaxAggregationResultSize(
            aggregationDescriptors, timeRangeIterator, context.getTypeProvider());

    return new SlidingWindowAggregationOperator(
        operatorContext,
        aggregators,
        timeRangeIterator,
        child,
        ascending,
        node.isOutputEndTime(),
        groupByTimeParameter,
        maxReturnSize);
  }

  @Override
  public Operator visitLimit(LimitNode node, LocalExecutionPlanContext context) {
    Operator child = node.getChild().accept(this, context);
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                LimitOperator.class.getSimpleName());

    return new LimitOperator(operatorContext, node.getLimit(), child);
  }

  @Override
  public Operator visitOffset(OffsetNode node, LocalExecutionPlanContext context) {
    Operator child = node.getChild().accept(this, context);
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                OffsetOperator.class.getSimpleName());

    return new OffsetOperator(operatorContext, node.getOffset(), child);
  }

  @Override
  public Operator visitAggregation(AggregationNode node, LocalExecutionPlanContext context) {
    checkArgument(
        !node.getAggregationDescriptorList().isEmpty(),
        "Aggregation descriptorList cannot be empty");
    List<Operator> children = dealWithConsumeAllChildrenPipelineBreaker(node, context);
    boolean ascending = node.getScanOrder() == Ordering.ASC;
    List<Aggregator> aggregators = new ArrayList<>();
    Map<String, List<InputLocation>> layout = makeLayout(node);
    List<AggregationDescriptor> aggregationDescriptors = node.getAggregationDescriptorList();
    for (AggregationDescriptor descriptor : node.getAggregationDescriptorList()) {
      List<InputLocation[]> inputLocationList = calcInputLocationList(descriptor, layout);
      aggregators.add(
          new Aggregator(
              AccumulatorFactory.createAccumulator(
                  descriptor.getAggregationFuncName(),
                  descriptor.getAggregationType(),
                  descriptor.getInputExpressions().stream()
                      .map(x -> context.getTypeProvider().getType(x.getExpressionString()))
                      .collect(Collectors.toList()),
                  descriptor.getInputExpressions(),
                  descriptor.getInputAttributes(),
                  ascending,
                  descriptor.getStep().isInputRaw()),
              descriptor.getStep(),
              inputLocationList));
    }
    boolean inputRaw = node.getAggregationDescriptorList().get(0).getStep().isInputRaw();
    GroupByTimeParameter groupByTimeParameter = node.getGroupByTimeParameter();
    GroupByParameter groupByParameter = node.getGroupByParameter();

    if (inputRaw) {
      checkArgument(children.size() == 1, "rawDataAggregateOperator can only accept one input");
      OperatorContext operatorContext =
          context
              .getDriverContext()
              .addOperatorContext(
                  context.getNextOperatorId(),
                  node.getPlanNodeId(),
                  RawDataAggregationOperator.class.getSimpleName());

      ITimeRangeIterator timeRangeIterator =
          initTimeRangeIterator(groupByTimeParameter, ascending, true);
      long maxReturnSize =
          calculateMaxAggregationResultSize(
              aggregationDescriptors, timeRangeIterator, context.getTypeProvider());

      // groupByParameter and groupByTimeParameter
      if (groupByParameter != null) {
        WindowType windowType = groupByParameter.getWindowType();

        WindowParameter windowParameter;
        switch (windowType) {
          case VARIATION_WINDOW:
            Expression groupByVariationExpression = node.getGroupByExpression();
            if (groupByVariationExpression == null) {
              throw new IllegalArgumentException("groupByVariationExpression can't be null");
            }
            String controlColumn = groupByVariationExpression.getExpressionString();
            TSDataType controlColumnType = context.getTypeProvider().getType(controlColumn);
            windowParameter =
                new VariationWindowParameter(
                    controlColumnType,
                    layout.get(controlColumn).get(0).getValueColumnIndex(),
                    node.isOutputEndTime(),
                    ((GroupByVariationParameter) groupByParameter).isIgnoringNull(),
                    ((GroupByVariationParameter) groupByParameter).getDelta());
            break;
          case CONDITION_WINDOW:
            Expression groupByConditionExpression = node.getGroupByExpression();
            if (groupByConditionExpression == null) {
              throw new IllegalArgumentException("groupByConditionExpression can't be null");
            }
            windowParameter =
                new ConditionWindowParameter(
                    node.isOutputEndTime(),
                    ((GroupByConditionParameter) groupByParameter).isIgnoringNull(),
                    layout
                        .get(groupByConditionExpression.getExpressionString())
                        .get(0)
                        .getValueColumnIndex(),
                    ((GroupByConditionParameter) groupByParameter).getKeepExpression());
            break;
          case SESSION_WINDOW:
            windowParameter =
                new SessionWindowParameter(
                    ((GroupBySessionParameter) groupByParameter).getTimeInterval(),
                    node.isOutputEndTime());
            break;
          case COUNT_WINDOW:
            Expression groupByCountExpression = node.getGroupByExpression();
            if (groupByCountExpression == null) {
              throw new IllegalArgumentException("groupByCountExpression can't be null");
            }
            windowParameter =
                new CountWindowParameter(
                    ((GroupByCountParameter) groupByParameter).getCountNumber(),
                    layout
                        .get(groupByCountExpression.getExpressionString())
                        .get(0)
                        .getValueColumnIndex(),
                    node.isOutputEndTime(),
                    ((GroupByCountParameter) groupByParameter).isIgnoreNull());
            break;
          default:
            throw new IllegalArgumentException("Unsupported window type");
        }
        return new RawDataAggregationOperator(
            operatorContext,
            aggregators,
            timeRangeIterator,
            children.get(0),
            ascending,
            maxReturnSize,
            windowParameter);
      }

      WindowParameter windowParameter = new TimeWindowParameter(node.isOutputEndTime());
      return new RawDataAggregationOperator(
          operatorContext,
          aggregators,
          timeRangeIterator,
          children.get(0),
          ascending,
          maxReturnSize,
          windowParameter);
    } else {
      OperatorContext operatorContext =
          context
              .getDriverContext()
              .addOperatorContext(
                  context.getNextOperatorId(),
                  node.getPlanNodeId(),
                  AggregationOperator.class.getSimpleName());

      ITimeRangeIterator timeRangeIterator =
          initTimeRangeIterator(groupByTimeParameter, ascending, true);
      long maxReturnSize =
          calculateMaxAggregationResultSize(
              aggregationDescriptors, timeRangeIterator, context.getTypeProvider());

      return new AggregationOperator(
          operatorContext,
          aggregators,
          timeRangeIterator,
          children,
          node.isOutputEndTime(),
          maxReturnSize);
    }
  }

  private List<InputLocation[]> calcInputLocationList(
      AggregationDescriptor descriptor, Map<String, List<InputLocation>> layout) {
    List<List<String>> inputColumnNames = descriptor.getInputColumnNamesList();
    List<InputLocation[]> inputLocationList = new ArrayList<>();

    for (List<String> inputColumnNamesOfOneInput : inputColumnNames) {
      // it may include double parts
      List<List<InputLocation>> inputLocationParts = new ArrayList<>();
      inputColumnNamesOfOneInput.forEach(o -> inputLocationParts.add(layout.get(o)));
      for (int i = 0; i < inputLocationParts.get(0).size(); i++) {
        if (inputColumnNamesOfOneInput.size() == 1) {
          inputLocationList.add(new InputLocation[] {inputLocationParts.get(0).get(i)});
        } else {
          inputLocationList.add(
              new InputLocation[] {
                inputLocationParts.get(0).get(i), inputLocationParts.get(1).get(i)
              });
        }
      }
    }
    return inputLocationList;
  }

  @Override
  public Operator visitSort(SortNode node, LocalExecutionPlanContext context) {
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                SortOperator.class.getSimpleName());

    List<TSDataType> dataTypes = getOutputColumnTypes(node, context.getTypeProvider());

    List<SortItem> sortItemList = node.getOrderByParameter().getSortItemList();

    List<Integer> sortItemIndexList = new ArrayList<>(sortItemList.size());
    List<TSDataType> sortItemDataTypeList = new ArrayList<>(sortItemList.size());
    genSortInformation(
        node.getOutputColumnNames(),
        dataTypes,
        sortItemList,
        sortItemIndexList,
        sortItemDataTypeList);

    String filePrefix =
        IoTDBDescriptor.getInstance().getConfig().getSortTmpDir()
            + File.separator
            + operatorContext.getDriverContext().getFragmentInstanceContext().getId().getFullId()
            + File.separator
            + operatorContext.getDriverContext().getPipelineId()
            + File.separator;

    context.getDriverContext().setHaveTmpFile(true);
    context.getDriverContext().getFragmentInstanceContext().setMayHaveTmpFile(true);

    Operator child = node.getChild().accept(this, context);

    return new SortOperator(
        operatorContext,
        child,
        dataTypes,
        filePrefix,
        getComparator(sortItemList, sortItemIndexList, sortItemDataTypeList));
  }

  @Override
  public Operator visitInto(IntoNode node, LocalExecutionPlanContext context) {
    Operator child = node.getChild().accept(this, context);
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                IntoOperator.class.getSimpleName());

    IntoPathDescriptor intoPathDescriptor = node.getIntoPathDescriptor();

    Map<String, InputLocation> sourceColumnToInputLocationMap =
        constructSourceColumnToInputLocationMap(node);
    Map<PartialPath, Map<String, InputLocation>> targetPathToSourceInputLocationMap =
        new HashMap<>();
    processTargetPathToSourceMap(
        intoPathDescriptor.getTargetPathToSourceMap(),
        targetPathToSourceInputLocationMap,
        sourceColumnToInputLocationMap);

    Map<PartialPath, Map<String, TSDataType>> targetPathToDataTypeMap =
        intoPathDescriptor.getTargetPathToDataTypeMap();
    long statementSizePerLine = calculateStatementSizePerLine(targetPathToDataTypeMap);

    List<Pair<String, PartialPath>> sourceTargetPathPairList =
        intoPathDescriptor.getSourceTargetPathPairList();
    List<String> sourceColumnToViewList = intoPathDescriptor.getSourceColumnToViewList();
    List<Pair<String, PartialPath>> sourceTargetPathPairWithViewList =
        new ArrayList<>(sourceTargetPathPairList);
    for (int i = 0; i < sourceColumnToViewList.size(); i++) {
      String viewPath = sourceColumnToViewList.get(i);
      if (StringUtils.isNotEmpty(viewPath)) {
        sourceTargetPathPairWithViewList.get(i).setLeft(viewPath);
      }
    }

    return new IntoOperator(
        operatorContext,
        child,
        getInputColumnTypes(node, context.getTypeProvider()),
        targetPathToSourceInputLocationMap,
        targetPathToDataTypeMap,
        intoPathDescriptor.getTargetDeviceToAlignedMap(),
        sourceTargetPathPairWithViewList,
        FragmentInstanceManager.getInstance().getIntoOperationExecutor(),
        statementSizePerLine);
  }

  @Override
  public Operator visitDeviceViewInto(DeviceViewIntoNode node, LocalExecutionPlanContext context) {
    Operator child = node.getChild().accept(this, context);
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                DeviceViewIntoOperator.class.getSimpleName());

    DeviceViewIntoPathDescriptor deviceViewIntoPathDescriptor =
        node.getDeviceViewIntoPathDescriptor();
    Map<String, InputLocation> sourceColumnToInputLocationMap =
        constructSourceColumnToInputLocationMap(node);

    Map<String, Map<PartialPath, Map<String, InputLocation>>>
        deviceToTargetPathSourceInputLocationMap = new HashMap<>();
    Map<String, Map<PartialPath, Map<String, TSDataType>>> deviceToTargetPathDataTypeMap =
        deviceViewIntoPathDescriptor.getSourceDeviceToTargetPathDataTypeMap();
    Map<String, Map<PartialPath, Map<String, String>>> sourceDeviceToTargetPathMap =
        deviceViewIntoPathDescriptor.getSourceDeviceToTargetPathMap();
    long statementSizePerLine = 0L;
    for (Map.Entry<String, Map<PartialPath, Map<String, String>>> deviceEntry :
        sourceDeviceToTargetPathMap.entrySet()) {
      String sourceDevice = deviceEntry.getKey();
      Map<PartialPath, Map<String, InputLocation>> targetPathToSourceInputLocationMap =
          new HashMap<>();
      processTargetPathToSourceMap(
          deviceEntry.getValue(),
          targetPathToSourceInputLocationMap,
          sourceColumnToInputLocationMap);
      deviceToTargetPathSourceInputLocationMap.put(
          sourceDevice, targetPathToSourceInputLocationMap);
      statementSizePerLine +=
          calculateStatementSizePerLine(deviceToTargetPathDataTypeMap.get(sourceDevice));
    }

    return new DeviceViewIntoOperator(
        operatorContext,
        child,
        getInputColumnTypes(node, context.getTypeProvider()),
        deviceToTargetPathSourceInputLocationMap,
        deviceToTargetPathDataTypeMap,
        deviceViewIntoPathDescriptor.getTargetDeviceToAlignedMap(),
        deviceViewIntoPathDescriptor.getDeviceToSourceTargetPathPairListMap(),
        sourceColumnToInputLocationMap,
        FragmentInstanceManager.getInstance().getIntoOperationExecutor(),
        statementSizePerLine);
  }

  private Map<String, InputLocation> constructSourceColumnToInputLocationMap(PlanNode node) {
    Map<String, InputLocation> sourceColumnToInputLocationMap = new HashMap<>();
    Map<String, List<InputLocation>> layout = makeLayout(node);
    for (Map.Entry<String, List<InputLocation>> layoutEntry : layout.entrySet()) {
      sourceColumnToInputLocationMap.put(layoutEntry.getKey(), layoutEntry.getValue().get(0));
    }
    return sourceColumnToInputLocationMap;
  }

  private void processTargetPathToSourceMap(
      Map<PartialPath, Map<String, String>> targetPathToSourceMap,
      Map<PartialPath, Map<String, InputLocation>> targetPathToSourceInputLocationMap,
      Map<String, InputLocation> sourceColumnToInputLocationMap) {
    for (Map.Entry<PartialPath, Map<String, String>> entry : targetPathToSourceMap.entrySet()) {
      PartialPath targetDevice = entry.getKey();
      Map<String, InputLocation> measurementToInputLocationMap = new HashMap<>();
      for (Map.Entry<String, String> measurementEntry : entry.getValue().entrySet()) {
        String targetMeasurement = measurementEntry.getKey();
        String sourceColumn = measurementEntry.getValue();
        measurementToInputLocationMap.put(
            targetMeasurement, sourceColumnToInputLocationMap.get(sourceColumn));
      }
      targetPathToSourceInputLocationMap.put(targetDevice, measurementToInputLocationMap);
    }
  }

  private long calculateStatementSizePerLine(
      Map<PartialPath, Map<String, TSDataType>> targetPathToDataTypeMap) {
    long maxStatementSize = Long.BYTES;
    List<TSDataType> dataTypes =
        targetPathToDataTypeMap.values().stream()
            .flatMap(stringTSDataTypeMap -> stringTSDataTypeMap.values().stream())
            .collect(Collectors.toList());
    for (TSDataType dataType : dataTypes) {
      maxStatementSize += getValueSizePerLine(dataType);
    }
    return maxStatementSize;
  }

  private static long getValueSizePerLine(TSDataType tsDataType) {
    switch (tsDataType) {
      case INT32:
        return Integer.BYTES;
      case INT64:
        return Long.BYTES;
      case FLOAT:
        return Float.BYTES;
      case DOUBLE:
        return Double.BYTES;
      case BOOLEAN:
        return Byte.BYTES;
      case TEXT:
        return StatisticsManager.getInstance().getMaxBinarySizeInBytes();
      default:
        throw new UnsupportedOperationException("Unknown data type " + tsDataType);
    }
  }

  @Override
  public Operator visitFullOuterTimeJoin(
      FullOuterTimeJoinNode node, LocalExecutionPlanContext context) {
    List<Operator> children = dealWithConsumeAllChildrenPipelineBreaker(node, context);
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                FullOuterTimeJoinOperator.class.getSimpleName());
    TimeComparator timeComparator =
        node.getMergeOrder() == Ordering.ASC ? ASC_TIME_COMPARATOR : DESC_TIME_COMPARATOR;
    List<OutputColumn> outputColumns = generateOutputColumnsFromChildren(node);
    List<ColumnMerger> mergers = createColumnMergers(outputColumns, timeComparator);
    List<TSDataType> outputColumnTypes =
        context.getTypeProvider().getTemplatedInfo() != null
            ? getOutputColumnTypesOfTimeJoinNode(node)
            : getOutputColumnTypes(node, context.getTypeProvider());

    return new FullOuterTimeJoinOperator(
        operatorContext,
        children,
        node.getMergeOrder(),
        outputColumnTypes,
        mergers,
        timeComparator);
  }

  @Override
  public Operator visitInnerTimeJoin(InnerTimeJoinNode node, LocalExecutionPlanContext context) {
    node.getTimePartitions().ifPresent(context::setTimePartitions);

    List<Operator> children = dealWithConsumeAllChildrenPipelineBreaker(node, context);
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                InnerTimeJoinOperator.class.getSimpleName());
    TimeComparator timeComparator =
        node.getMergeOrder() == Ordering.ASC ? ASC_TIME_COMPARATOR : DESC_TIME_COMPARATOR;
    List<TSDataType> outputColumnTypes =
        context.getTypeProvider().getTemplatedInfo() != null
            ? getOutputColumnTypesOfTimeJoinNode(node)
            : getOutputColumnTypes(node, context.getTypeProvider());

    return new InnerTimeJoinOperator(
        operatorContext, children, outputColumnTypes, timeComparator, getOutputColumnMap(node));
  }

  private Map<InputLocation, Integer> getOutputColumnMap(InnerTimeJoinNode innerTimeJoinNode) {
    Map<InputLocation, Integer> result = new HashMap<>();
    if (innerTimeJoinNode.outputColumnNamesIsNull()) {
      int outputIndex = 0;
      for (int i = 0, size = innerTimeJoinNode.getChildren().size(); i < size; i++) {
        PlanNode child = innerTimeJoinNode.getChildren().get(i);
        List<String> childOutputColumns = child.getOutputColumnNames();
        for (int j = 0, childSize = childOutputColumns.size(); j < childSize; j++) {
          result.put(new InputLocation(i, j), outputIndex++);
        }
      }
    } else {
      List<String> outputColumns = innerTimeJoinNode.getOutputColumnNames();
      Map<String, Integer> outputColumnIndexMap = new HashMap<>();
      for (int i = 0; i < outputColumns.size(); i++) {
        outputColumnIndexMap.put(outputColumns.get(i), i);
      }
      for (int i = 0, size = innerTimeJoinNode.getChildren().size(); i < size; i++) {
        PlanNode child = innerTimeJoinNode.getChildren().get(i);
        List<String> childOutputColumns = child.getOutputColumnNames();
        for (int j = 0, childSize = childOutputColumns.size(); j < childSize; j++) {
          result.put(new InputLocation(i, j), outputColumnIndexMap.get(childOutputColumns.get(j)));
        }
      }
    }
    return result;
  }

  @Override
  public Operator visitLeftOuterTimeJoin(
      LeftOuterTimeJoinNode node, LocalExecutionPlanContext context) {
    List<Operator> children = dealWithConsumeAllChildrenPipelineBreaker(node, context);
    checkState(children.size() == 2);
    Operator leftChild = children.get(0);
    Operator rightChild = children.get(1);

    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                LeftOuterTimeJoinOperator.class.getSimpleName());
    TimeComparator timeComparator =
        node.getMergeOrder() == Ordering.ASC ? ASC_TIME_COMPARATOR : DESC_TIME_COMPARATOR;
    List<TSDataType> outputColumnTypes =
        context.getTypeProvider().getTemplatedInfo() != null
            ? getOutputColumnTypesOfTimeJoinNode(node)
            : getOutputColumnTypes(node, context.getTypeProvider());

    return new LeftOuterTimeJoinOperator(
        operatorContext,
        leftChild,
        node.getLeftChild().getOutputColumnNames().size(),
        rightChild,
        outputColumnTypes,
        timeComparator);
  }

  @Override
  public Operator visitHorizontallyConcat(
      HorizontallyConcatNode node, LocalExecutionPlanContext context) {
    List<Operator> children = dealWithConsumeAllChildrenPipelineBreaker(node, context);
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                HorizontallyConcatOperator.class.getSimpleName());
    List<TSDataType> outputColumnTypes = getOutputColumnTypes(node, context.getTypeProvider());

    return new HorizontallyConcatOperator(operatorContext, children, outputColumnTypes);
  }

  @Override
  public Operator visitShowQueries(ShowQueriesNode node, LocalExecutionPlanContext context) {
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                ShowQueriesOperator.class.getSimpleName());

    return new ShowQueriesOperator(
        operatorContext, node.getPlanNodeId(), Coordinator.getInstance());
  }

  private List<OutputColumn> generateOutputColumnsFromChildren(MultiChildProcessNode node) {
    // TODO we should also sort the InputLocation for each column if they are not overlapped
    return makeLayout(node).entrySet().stream()
        .filter(entry -> !TIMESTAMP_EXPRESSION_STRING.equals(entry.getKey()))
        .map(entry -> new OutputColumn(entry.getValue(), entry.getValue().size() > 1))
        .collect(Collectors.toList());
  }

  private List<ColumnMerger> createColumnMergers(
      List<OutputColumn> outputColumns, TimeComparator timeComparator) {
    List<ColumnMerger> mergers = new ArrayList<>(outputColumns.size());
    for (OutputColumn outputColumn : outputColumns) {
      ColumnMerger merger;
      // only has one input column
      if (outputColumn.isSingleInputColumn()) {
        merger = new SingleColumnMerger(outputColumn.getSourceLocation(0), timeComparator);
      } else if (outputColumn.isOverlapped()) {
        // has more than one input columns but time of these input columns is overlapped
        merger = new MultiColumnMerger(outputColumn.getSourceLocations());
      } else {
        // has more than one input columns and time of these input columns is not overlapped
        merger =
            new NonOverlappedMultiColumnMerger(outputColumn.getSourceLocations(), timeComparator);
      }
      mergers.add(merger);
    }
    return mergers;
  }

  @Override
  public Operator visitExchange(ExchangeNode node, LocalExecutionPlanContext context) {
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                ExchangeOperator.class.getSimpleName());

    FragmentInstanceId localInstanceId = context.getInstanceContext().getId();
    FragmentInstanceId remoteInstanceId = node.getUpstreamInstanceId();

    TEndPoint upstreamEndPoint = node.getUpstreamEndpoint();
    boolean isSameNode = isSameNode(upstreamEndPoint);
    ISourceHandle sourceHandle =
        isSameNode
            ? MPP_DATA_EXCHANGE_MANAGER.createLocalSourceHandleForFragment(
                localInstanceId.toThrift(),
                node.getPlanNodeId().getId(),
                node.getUpstreamPlanNodeId().getId(),
                remoteInstanceId.toThrift(),
                node.getIndexOfUpstreamSinkHandle(),
                context.getInstanceContext()::failed)
            : MPP_DATA_EXCHANGE_MANAGER.createSourceHandle(
                localInstanceId.toThrift(),
                node.getPlanNodeId().getId(),
                node.getIndexOfUpstreamSinkHandle(),
                upstreamEndPoint,
                remoteInstanceId.toThrift(),
                context.getInstanceContext()::failed);
    if (!isSameNode) {
      context.addExchangeSumNum(1);
    }
    sourceHandle.setMaxBytesCanReserve(context.getMaxBytesOneHandleCanReserve());
    ExchangeOperator exchangeOperator =
        new ExchangeOperator(operatorContext, sourceHandle, node.getUpstreamPlanNodeId());
    context.addExchangeOperator(exchangeOperator);
    return exchangeOperator;
  }

  @Override
  public Operator visitIdentitySink(IdentitySinkNode node, LocalExecutionPlanContext context) {
    context.addExchangeSumNum(1);
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                IdentitySinkOperator.class.getSimpleName());

    checkArgument(
        MPP_DATA_EXCHANGE_MANAGER != null, "MPP_DATA_EXCHANGE_MANAGER should not be null");
    FragmentInstanceId localInstanceId = context.getInstanceContext().getId();
    DownStreamChannelIndex downStreamChannelIndex = new DownStreamChannelIndex(0);
    ISinkHandle sinkHandle =
        MPP_DATA_EXCHANGE_MANAGER.createShuffleSinkHandle(
            node.getDownStreamChannelLocationList(),
            downStreamChannelIndex,
            ShuffleSinkHandle.ShuffleStrategyEnum.PLAIN,
            localInstanceId.toThrift(),
            node.getPlanNodeId().getId(),
            context.getInstanceContext());
    sinkHandle.setMaxBytesCanReserve(context.getMaxBytesOneHandleCanReserve());
    context.getDriverContext().setSink(sinkHandle);

    List<Operator> children = dealWithConsumeChildrenOneByOneNode(node, context);
    return new IdentitySinkOperator(operatorContext, children, downStreamChannelIndex, sinkHandle);
  }

  @Override
  public Operator visitShuffleSink(ShuffleSinkNode node, LocalExecutionPlanContext context) {
    context.addExchangeSumNum(1);
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                ShuffleHelperOperator.class.getSimpleName());

    // TODO implement pipeline division for shuffle sink
    context.setDegreeOfParallelism(1);
    List<Operator> children = dealWithConsumeAllChildrenPipelineBreaker(node, context);

    checkArgument(
        MPP_DATA_EXCHANGE_MANAGER != null, "MPP_DATA_EXCHANGE_MANAGER should not be null");
    FragmentInstanceId localInstanceId = context.getInstanceContext().getId();
    DownStreamChannelIndex downStreamChannelIndex = new DownStreamChannelIndex(0);
    ISinkHandle sinkHandle =
        MPP_DATA_EXCHANGE_MANAGER.createShuffleSinkHandle(
            node.getDownStreamChannelLocationList(),
            downStreamChannelIndex,
            ShuffleSinkHandle.ShuffleStrategyEnum.SIMPLE_ROUND_ROBIN,
            localInstanceId.toThrift(),
            node.getPlanNodeId().getId(),
            context.getInstanceContext());
    sinkHandle.setMaxBytesCanReserve(context.getMaxBytesOneHandleCanReserve());
    context.getDriverContext().setSink(sinkHandle);

    return new ShuffleHelperOperator(operatorContext, children, downStreamChannelIndex, sinkHandle);
  }

  @Override
  public Operator visitSchemaFetchMerge(
      SchemaFetchMergeNode node, LocalExecutionPlanContext context) {
    List<Operator> children = dealWithConsumeChildrenOneByOneNode(node, context);
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                SchemaFetchMergeOperator.class.getSimpleName());
    return new SchemaFetchMergeOperator(operatorContext, children, node.getStorageGroupList());
  }

  @Override
  public Operator visitSchemaFetchScan(
      SchemaFetchScanNode node, LocalExecutionPlanContext context) {
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                SchemaFetchScanOperator.class.getSimpleName());
    return new SchemaFetchScanOperator(
        node.getPlanNodeId(),
        operatorContext,
        node.getPatternTree(),
        node.getTemplateMap(),
        ((SchemaDriverContext) (context.getDriverContext())).getSchemaRegion(),
        node.isWithTags(),
        node.isWithTemplate());
  }

  @Override
  public Operator visitLastQueryScan(LastQueryScanNode node, LocalExecutionPlanContext context) {
    PartialPath seriesPath = node.getSeriesPath().transformToPartialPath();
    TimeValuePair timeValuePair = null;
    context.dataNodeQueryContext.lock();
    try {
      if (!context.dataNodeQueryContext.unCached(seriesPath)) {
        timeValuePair = DATA_NODE_SCHEMA_CACHE.getLastCache(seriesPath);
        if (timeValuePair == null) {
          context.dataNodeQueryContext.addUnCachePath(seriesPath, node.getDataNodeSeriesScanNum());
        }
      }
    } finally {
      context.dataNodeQueryContext.unLock();
    }

    if (timeValuePair == null) { // last value is not cached
      return createUpdateLastCacheOperator(node, context, node.getSeriesPath());
    } else if (timeValuePair.getValue() == null) { // there is no data for this time series
      return null;
    } else if (!LastQueryUtil.satisfyFilter(
        updateFilterUsingTTL(context.getGlobalTimeFilter(), context.getDataRegionTTL()),
        timeValuePair)) { // cached last value is not satisfied

      if (!isFilterGtOrGe(context.getGlobalTimeFilter())) {
        // time filter is not > or >=, we still need to read from disk
        return createUpdateLastCacheOperator(node, context, node.getSeriesPath());
      } else { // otherwise, we just ignore it and return null
        return null;
      }
    } else { //  cached last value is satisfied, put it into LastCacheScanOperator
      context.addCachedLastValue(timeValuePair, node.outputPathSymbol());
      return null;
    }
  }

  private boolean isFilterGtOrGe(Filter filter) {
    return filter instanceof TimeGt || filter instanceof TimeGtEq;
  }

  private UpdateLastCacheOperator createUpdateLastCacheOperator(
      LastQueryScanNode node, LocalExecutionPlanContext context, MeasurementPath fullPath) {
    SeriesAggregationScanOperator lastQueryScan = createLastQueryScanOperator(node, context);
    if (node.getOutputViewPath() == null) {
      OperatorContext operatorContext =
          context
              .getDriverContext()
              .addOperatorContext(
                  context.getNextOperatorId(),
                  node.getPlanNodeId(),
                  UpdateLastCacheOperator.class.getSimpleName());
      return new UpdateLastCacheOperator(
          operatorContext,
          lastQueryScan,
          fullPath,
          node.getSeriesPath().getSeriesType(),
          DATA_NODE_SCHEMA_CACHE,
          context.isNeedUpdateLastCache(),
          context.isNeedUpdateNullEntry());
    } else {
      OperatorContext operatorContext =
          context
              .getDriverContext()
              .addOperatorContext(
                  context.getNextOperatorId(),
                  node.getPlanNodeId(),
                  UpdateViewPathLastCacheOperator.class.getSimpleName());
      return new UpdateViewPathLastCacheOperator(
          operatorContext,
          lastQueryScan,
          fullPath,
          node.getSeriesPath().getSeriesType(),
          DATA_NODE_SCHEMA_CACHE,
          context.isNeedUpdateLastCache(),
          context.isNeedUpdateNullEntry(),
          node.getOutputViewPath());
    }
  }

  private AlignedUpdateLastCacheOperator createAlignedUpdateLastCacheOperator(
      AlignedLastQueryScanNode node, AlignedPath unCachedPath, LocalExecutionPlanContext context) {
    AlignedSeriesAggregationScanOperator lastQueryScan =
        createLastQueryScanOperator(node, unCachedPath, context);

    if (node.getOutputViewPath() == null) {
      OperatorContext operatorContext =
          context
              .getDriverContext()
              .addOperatorContext(
                  context.getNextOperatorId(),
                  node.getPlanNodeId(),
                  AlignedUpdateLastCacheOperator.class.getSimpleName());
      return new AlignedUpdateLastCacheOperator(
          operatorContext,
          lastQueryScan,
          unCachedPath,
          DATA_NODE_SCHEMA_CACHE,
          context.isNeedUpdateLastCache(),
          context.isNeedUpdateNullEntry());
    } else {
      OperatorContext operatorContext =
          context
              .getDriverContext()
              .addOperatorContext(
                  context.getNextOperatorId(),
                  node.getPlanNodeId(),
                  AlignedUpdateViewPathLastCacheOperator.class.getSimpleName());
      return new AlignedUpdateViewPathLastCacheOperator(
          operatorContext,
          lastQueryScan,
          unCachedPath,
          DATA_NODE_SCHEMA_CACHE,
          context.isNeedUpdateLastCache(),
          context.isNeedUpdateNullEntry(),
          node.getOutputViewPath());
    }
  }

  private SeriesAggregationScanOperator createLastQueryScanOperator(
      LastQueryScanNode node, LocalExecutionPlanContext context) {
    MeasurementPath seriesPath = node.getSeriesPath();
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                SeriesAggregationScanOperator.class.getSimpleName());

    // last_time, last_value
    List<Aggregator> aggregators = LastQueryUtil.createAggregators(seriesPath.getSeriesType());
    ITimeRangeIterator timeRangeIterator = initTimeRangeIterator(null, false, false);
    long maxReturnSize = calculateMaxAggregationResultSizeForLastQuery(aggregators);

    SeriesScanOptions.Builder scanOptionsBuilder = new SeriesScanOptions.Builder();
    scanOptionsBuilder.withAllSensors(
        context.getAllSensors(seriesPath.getDevice(), seriesPath.getMeasurement()));
    scanOptionsBuilder.withGlobalTimeFilter(context.getGlobalTimeFilter());

    SeriesAggregationScanOperator seriesAggregationScanOperator =
        new SeriesAggregationScanOperator(
            node.getPlanNodeId(),
            seriesPath,
            Ordering.DESC,
            scanOptionsBuilder.build(),
            operatorContext,
            aggregators,
            timeRangeIterator,
            null,
            maxReturnSize);
    ((DataDriverContext) context.getDriverContext())
        .addSourceOperator(seriesAggregationScanOperator);
    ((DataDriverContext) context.getDriverContext()).addPath(seriesPath);
    return seriesAggregationScanOperator;
  }

  private AlignedSeriesAggregationScanOperator createLastQueryScanOperator(
      AlignedLastQueryScanNode node, AlignedPath unCachedPath, LocalExecutionPlanContext context) {
    // last_time, last_value
    List<Aggregator> aggregators = new ArrayList<>();
    for (int i = 0; i < unCachedPath.getMeasurementList().size(); i++) {
      aggregators.addAll(
          LastQueryUtil.createAggregators(unCachedPath.getSchemaList().get(i).getType(), i));
    }
    ITimeRangeIterator timeRangeIterator = initTimeRangeIterator(null, false, false);
    long maxReturnSize = calculateMaxAggregationResultSizeForLastQuery(aggregators);

    SeriesScanOptions.Builder scanOptionsBuilder = new SeriesScanOptions.Builder();
    scanOptionsBuilder.withAllSensors(new HashSet<>(unCachedPath.getMeasurementList()));
    scanOptionsBuilder.withGlobalTimeFilter(context.getGlobalTimeFilter());

    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                AlignedSeriesAggregationScanOperator.class.getSimpleName());
    AlignedSeriesAggregationScanOperator seriesAggregationScanOperator =
        new AlignedSeriesAggregationScanOperator(
            node.getPlanNodeId(),
            unCachedPath,
            Ordering.DESC,
            scanOptionsBuilder.build(),
            operatorContext,
            aggregators,
            timeRangeIterator,
            null,
            maxReturnSize);
    ((DataDriverContext) context.getDriverContext())
        .addSourceOperator(seriesAggregationScanOperator);
    ((DataDriverContext) context.getDriverContext()).addPath(unCachedPath);
    return seriesAggregationScanOperator;
  }

  @Override
  public Operator visitAlignedLastQueryScan(
      AlignedLastQueryScanNode node, LocalExecutionPlanContext context) {
    AlignedPath alignedPath = node.getSeriesPath();
    PartialPath devicePath = alignedPath.getDevicePath();
    // get series under aligned entity that has not been cached
    List<Integer> unCachedMeasurementIndexes = new ArrayList<>();
    List<String> measurementList = alignedPath.getMeasurementList();
    for (int i = 0; i < measurementList.size(); i++) {
      PartialPath measurementPath = devicePath.concatNode(measurementList.get(i));
      TimeValuePair timeValuePair = null;
      try {
        context.dataNodeQueryContext.lock();
        if (!context.dataNodeQueryContext.unCached(measurementPath)) {
          timeValuePair = DATA_NODE_SCHEMA_CACHE.getLastCache(measurementPath);
          if (timeValuePair == null) {
            context.dataNodeQueryContext.addUnCachePath(
                measurementPath, new AtomicInteger(node.getDataNodeSeriesScanNum().get()));
          }
        }
      } finally {
        context.dataNodeQueryContext.unLock();
      }

      if (timeValuePair == null) { // last value is not cached
        unCachedMeasurementIndexes.add(i);
      } else if (timeValuePair.getValue() == null) {
        // there is no data for this time series, just ignore
      } else if (!LastQueryUtil.satisfyFilter(
          updateFilterUsingTTL(context.getGlobalTimeFilter(), context.getDataRegionTTL()),
          timeValuePair)) { // cached last value is not satisfied

        if (!isFilterGtOrGe(context.getGlobalTimeFilter())) {
          // time filter is not > or >=, we still need to read from disk
          unCachedMeasurementIndexes.add(i);
        }
      } else { //  cached last value is satisfied, put it into LastCacheScanOperator
        if (node.getOutputViewPath() != null) {
          context.addCachedLastValue(timeValuePair, node.getOutputViewPath());
        } else {
          context.addCachedLastValue(timeValuePair, measurementPath.getFullPath());
        }
      }
    }
    if (unCachedMeasurementIndexes.isEmpty()) {
      return null;
    } else {
      AlignedPath unCachedPath = new AlignedPath(alignedPath.getDevicePath());
      for (int i : unCachedMeasurementIndexes) {
        unCachedPath.addMeasurement(measurementList.get(i), alignedPath.getSchemaList().get(i));
      }
      return createAlignedUpdateLastCacheOperator(node, unCachedPath, context);
    }
  }

  @Override
  public Operator visitLastQuery(LastQueryNode node, LocalExecutionPlanContext context) {
    Filter globalTimeFilter = context.getGlobalTimeFilter();
    context.setNeedUpdateLastCache(LastQueryUtil.needUpdateCache(globalTimeFilter));
    context.setNeedUpdateNullEntry(LastQueryUtil.needUpdateNullEntry(globalTimeFilter));

    List<Operator> operatorList =
        node.getChildren().stream()
            .map(child -> child.accept(this, context))
            .filter(Objects::nonNull)
            .collect(Collectors.toList());

    List<Pair<TimeValuePair, Binary>> cachedLastValueAndPathList =
        context.getCachedLastValueAndPathList();

    int initSize = cachedLastValueAndPathList != null ? cachedLastValueAndPathList.size() : 0;
    // no need to order by timeseries at first
    if (!node.needOrderByTimeseries()) {
      TsBlockBuilder builder = LastQueryUtil.createTsBlockBuilder(initSize);
      for (int i = 0; i < initSize; i++) {
        TimeValuePair timeValuePair = cachedLastValueAndPathList.get(i).left;
        LastQueryUtil.appendLastValue(
            builder,
            timeValuePair.getTimestamp(),
            cachedLastValueAndPathList.get(i).right,
            timeValuePair.getValue().getStringValue(),
            timeValuePair.getValue().getDataType().name());
      }
      OperatorContext operatorContext =
          context
              .getDriverContext()
              .addOperatorContext(
                  context.getNextOperatorId(),
                  node.getPlanNodeId(),
                  LastQueryOperator.class.getSimpleName());
      return new LastQueryOperator(operatorContext, operatorList, builder);
    } else {
      // order by timeseries
      Comparator<Binary> comparator =
          node.getTimeseriesOrdering() == Ordering.ASC
              ? ASC_BINARY_COMPARATOR
              : DESC_BINARY_COMPARATOR;
      // sort values from last cache
      if (initSize > 0) {
        cachedLastValueAndPathList.sort(Comparator.comparing(Pair::getRight, comparator));
      }

      TsBlockBuilder builder = LastQueryUtil.createTsBlockBuilder(initSize);
      for (int i = 0; i < initSize; i++) {
        TimeValuePair timeValuePair = cachedLastValueAndPathList.get(i).left;
        LastQueryUtil.appendLastValue(
            builder,
            timeValuePair.getTimestamp(),
            cachedLastValueAndPathList.get(i).right,
            timeValuePair.getValue().getStringValue(),
            timeValuePair.getValue().getDataType().name());
      }

      OperatorContext operatorContext =
          context
              .getDriverContext()
              .addOperatorContext(
                  context.getNextOperatorId(),
                  node.getPlanNodeId(),
                  LastQuerySortOperator.class.getSimpleName());
      return new LastQuerySortOperator(operatorContext, builder.build(), operatorList, comparator);
    }
  }

  @Override
  public Operator visitLastQueryMerge(LastQueryMergeNode node, LocalExecutionPlanContext context) {
    List<Operator> children =
        node.getChildren().stream()
            .map(child -> child.accept(this, context))
            .collect(Collectors.toList());
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                LastQueryMergeOperator.class.getSimpleName());

    Ordering timeseriesOrdering = node.getTimeseriesOrdering();
    Comparator<Binary> comparator =
        (timeseriesOrdering == null || timeseriesOrdering == Ordering.ASC)
            ? ASC_BINARY_COMPARATOR
            : DESC_BINARY_COMPARATOR;

    return new LastQueryMergeOperator(operatorContext, children, comparator);
  }

  @Override
  public Operator visitLastQueryCollect(
      LastQueryCollectNode node, LocalExecutionPlanContext context) {
    List<Operator> children = dealWithConsumeChildrenOneByOneNode(node, context);
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                LastQueryCollectOperator.class.getSimpleName());

    return new LastQueryCollectOperator(operatorContext, children);
  }

  @Override
  public Operator visitLastQueryTransform(
      LastQueryTransformNode node, LocalExecutionPlanContext context) {
    Operator operator = node.getChild().accept(this, context);
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                LastQueryCollectOperator.class.getSimpleName());

    return new LastQueryTransformOperator(
        node.getViewPath(), node.getDataType(), operatorContext, operator);
  }

  private Map<String, List<InputLocation>> makeLayout(PlanNode node) {
    return makeLayout(node.getChildren());
  }

  private Map<String, List<InputLocation>> makeLayout(List<PlanNode> children) {
    Map<String, List<InputLocation>> outputMappings = new LinkedHashMap<>();
    int tsBlockIndex = 0;
    for (PlanNode childNode : children) {
      outputMappings
          .computeIfAbsent(TIMESTAMP_EXPRESSION_STRING, key -> new ArrayList<>())
          .add(new InputLocation(tsBlockIndex, -1));
      int valueColumnIndex = 0;
      for (String columnName : childNode.getOutputColumnNames()) {
        outputMappings
            .computeIfAbsent(columnName, key -> new ArrayList<>())
            .add(new InputLocation(tsBlockIndex, valueColumnIndex));
        valueColumnIndex++;
      }
      tsBlockIndex++;
    }
    return outputMappings;
  }

  private List<TSDataType> getInputColumnTypes(PlanNode node, TypeProvider typeProvider) {
    if (typeProvider.getTemplatedInfo() == null) {
      return node.getChildren().stream()
          .map(PlanNode::getOutputColumnNames)
          .flatMap(List::stream)
          .map(typeProvider::getType)
          .collect(Collectors.toList());
    } else {
      return getInputColumnTypesUseTemplate(node);
    }
  }

  private List<TSDataType> getInputColumnTypesUseTemplate(PlanNode node) {
    // Only templated device + filter situation can invoke this method,
    // the children of FilterNode/TransformNode can be TimeJoinNode, ScanNode, any others?
    List<TSDataType> dataTypes = new ArrayList<>();
    for (PlanNode child : node.getChildren()) {
      if (child instanceof SeriesScanNode) {
        dataTypes.add(((SeriesScanNode) child).getSeriesPath().getSeriesType());
      } else if (child instanceof AlignedSeriesScanNode) {
        AlignedSeriesScanNode alignedSeriesScanNode = (AlignedSeriesScanNode) child;
        alignedSeriesScanNode
            .getAlignedPath()
            .getSchemaList()
            .forEach(c -> dataTypes.add(c.getType()));
      } else {
        dataTypes.addAll(getInputColumnTypesUseTemplate(child));
      }
    }
    return dataTypes;
  }

  private List<TSDataType> getOutputColumnTypes(PlanNode node, TypeProvider typeProvider) {
    return node.getOutputColumnNames().stream()
        .map(typeProvider::getType)
        .collect(Collectors.toList());
  }

  private List<TSDataType> getOutputColumnTypesOfTimeJoinNode(PlanNode node) {
    // Only templated device situation can invoke this method,
    // the children of TimeJoinNode can only be ScanNode or TimeJoinNode
    List<TSDataType> dataTypes = new ArrayList<>();
    for (PlanNode child : node.getChildren()) {
      if (child instanceof SeriesScanNode) {
        dataTypes.add(((SeriesScanNode) child).getSeriesPath().getSeriesType());
      } else if (child instanceof AlignedSeriesScanNode) {
        dataTypes.add(((AlignedSeriesScanNode) child).getAlignedPath().getSeriesType());
      } else if (child instanceof FullOuterTimeJoinNode
          || child instanceof InnerTimeJoinNode
          || child instanceof LeftOuterTimeJoinNode) {
        dataTypes.addAll(getOutputColumnTypesOfTimeJoinNode(child));
      } else {
        LOGGER.error(
            "Unexpected PlanNode in getOutputColumnTypesOfTimeJoinNode, type: {}",
            child.getOutputColumnNames());
      }
    }
    return dataTypes;
  }

  private Operator generateOnlyChildOperator(PlanNode node, LocalExecutionPlanContext context) {
    List<Operator> children =
        node.getChildren().stream()
            .map(child -> child.accept(this, context))
            .collect(Collectors.toList());
    Validate.isTrue(children.size() == 1);
    return children.get(0);
  }

  public Operator visitPathsUsingTemplateScan(
      PathsUsingTemplateScanNode node, LocalExecutionPlanContext context) {
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                SchemaQueryScanOperator.class.getSimpleName());
    return new SchemaQueryScanOperator<>(
        node.getPlanNodeId(),
        operatorContext,
        SchemaSourceFactory.getPathsUsingTemplateSource(
            node.getPathPatternList(), node.getTemplateId(), node.getScope()));
  }

  public Operator visitLogicalViewSchemaScan(
      LogicalViewSchemaScanNode node, LocalExecutionPlanContext context) {
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                SchemaQueryScanOperator.class.getSimpleName());
    return new SchemaQueryScanOperator<>(
        node.getPlanNodeId(),
        operatorContext,
        SchemaSourceFactory.getLogicalViewSchemaSource(
            node.getPath(),
            node.getLimit(),
            node.getOffset(),
            node.getSchemaFilter(),
            node.getScope()));
  }

  public List<Operator> dealWithConsumeAllChildrenPipelineBreaker(
      PlanNode node, LocalExecutionPlanContext context) {
    // children after pipelining
    List<Operator> parentPipelineChildren = new ArrayList<>();
    if (context.getDegreeOfParallelism() == 1 || node.getChildren().size() == 1) {
      // If dop = 1, we don't create extra pipeline
      for (PlanNode localChild : node.getChildren()) {
        Operator childOperation = localChild.accept(this, context);
        parentPipelineChildren.add(childOperation);
      }
    } else {
      int finalExchangeNum = context.getExchangeSumNum();
      // Keep it since we may change the structure of origin children nodes
      List<PlanNode> afterwardsNodes = new ArrayList<>();
      // 1. Calculate localChildren size
      int localChildrenSize = 0;
      int firstChildIndex = -1;
      for (int i = 0; i < node.getChildren().size(); i++) {
        if (!(node.getChildren().get(i) instanceof ExchangeNode)) {
          localChildrenSize++;
          firstChildIndex = firstChildIndex == -1 ? i : firstChildIndex;
          // deal with exchangeNode at head
        } else if (firstChildIndex == -1) {
          Operator childOperation = node.getChildren().get(i).accept(this, context);
          finalExchangeNum += 1;
          parentPipelineChildren.add(childOperation);
          afterwardsNodes.add(node.getChildren().get(i));
        }
      }
      if (firstChildIndex == -1) {
        context.setExchangeSumNum(finalExchangeNum);
        return parentPipelineChildren;
      }
      // If dop > localChildrenSize + 1, we can allocate extra dop to child node
      // Extra dop = dop - localChildrenSize, since dop = 1 means serial but not 0
      int dopForChild = Math.max(1, context.getDegreeOfParallelism() - localChildrenSize);
      // If dop > localChildrenSize, we create one new pipeline for each child
      if (context.getDegreeOfParallelism() > localChildrenSize) {
        for (int i = firstChildIndex; i < node.getChildren().size(); i++) {
          PlanNode childNode = node.getChildren().get(i);
          if (childNode instanceof ExchangeNode) {
            Operator childOperation = childNode.accept(this, context);
            finalExchangeNum += 1;
            parentPipelineChildren.add(childOperation);
          } else {
            LocalExecutionPlanContext subContext = context.createSubContext();
            subContext.setDegreeOfParallelism(dopForChild);

            int originPipeNum = context.getPipelineNumber();
            Operator sourceOperator = createNewPipelineForChildNode(context, subContext, childNode);
            parentPipelineChildren.add(sourceOperator);
            dopForChild =
                Math.max(1, dopForChild - (subContext.getPipelineNumber() - 1 - originPipeNum));
            finalExchangeNum += subContext.getExchangeSumNum() - context.getExchangeSumNum() + 1;
          }
        }
      } else {
        // If dop <= localChildrenSize, we have to divide every childNumInEachPipeline localChildren
        // to different pipeline
        int[] childNumInEachPipeline =
            getChildNumInEachPipeline(
                node.getChildren(), localChildrenSize, context.getDegreeOfParallelism());
        int childGroupNum = Math.min(context.getDegreeOfParallelism(), localChildrenSize);
        int startIndex;
        int endIndex = firstChildIndex;
        for (int i = 0; i < childGroupNum; i++) {
          startIndex = endIndex;
          endIndex += childNumInEachPipeline[i];
          // Only if dop >= size(children) + 1, split all children to new pipeline
          // Otherwise, the first group will belong to the parent pipeline
          if (i == 0) {
            for (int j = startIndex; j < endIndex; j++) {
              context.setDegreeOfParallelism(1);
              Operator childOperation = node.getChildren().get(j).accept(this, context);
              parentPipelineChildren.add(childOperation);
              afterwardsNodes.add(node.getChildren().get(j));
            }
            continue;
          }
          LocalExecutionPlanContext subContext = context.createSubContext();
          subContext.setDegreeOfParallelism(1);
          // Create partial parent operator for children
          PlanNode partialParentNode;
          if (endIndex - startIndex == 1) {
            partialParentNode = node.getChildren().get(startIndex);
          } else {
            // PartialParentNode is equals to parentNode except children
            partialParentNode = node.createSubNode(i, startIndex, endIndex);
          }

          Operator sourceOperator =
              createNewPipelineForChildNode(context, subContext, partialParentNode);
          parentPipelineChildren.add(sourceOperator);
          afterwardsNodes.add(partialParentNode);
          finalExchangeNum += subContext.getExchangeSumNum() - context.getExchangeSumNum() + 1;
        }

        if (node instanceof MultiChildProcessNode) {
          ((MultiChildProcessNode) node).setChildren(afterwardsNodes);
        } else if (node instanceof TwoChildProcessNode) {
          checkState(afterwardsNodes.size() == 2);
          ((TwoChildProcessNode) node).setLeftChild(afterwardsNodes.get(0));
          ((TwoChildProcessNode) node).setRightChild(afterwardsNodes.get(1));
        } else {
          throw new IllegalArgumentException("Unknown node type: " + node.getClass().getName());
        }
      }
      context.setExchangeSumNum(finalExchangeNum);
    }
    return parentPipelineChildren;
  }

  /**
   * Now, we allocate children to each pipeline as average as possible. For example, 5 children with
   * 3 dop, the children group will be [1, 2, 2]. After we can estimate the workload of each
   * operator, maybe we can allocate based on workload rather than child number.
   *
   * <p>If child is ExchangeNode, it won't affect the children number of current group.
   *
   * <p>This method can only be invoked when dop <= localChildrenSize.
   */
  public int[] getChildNumInEachPipeline(
      List<PlanNode> allChildren, int localChildrenSize, int dop) {
    int maxPipelineNum = Math.min(localChildrenSize, dop);
    int[] childNumInEachPipeline = new int[maxPipelineNum];
    int avgChildNum = Math.max(1, localChildrenSize / dop);
    // allocate remaining child to group from splitIndex
    int splitIndex = maxPipelineNum - localChildrenSize % dop;
    int childIndex = 0;
    // Skip ExchangeNode at head
    while (childIndex < allChildren.size() && allChildren.get(childIndex) instanceof ExchangeNode) {
      childIndex++;
    }
    int pipelineIndex = 0;
    while (pipelineIndex < maxPipelineNum) {
      int childNum = pipelineIndex < splitIndex ? avgChildNum : avgChildNum + 1;
      int originChildIndex = childIndex;
      while (childNum >= 0 && childIndex < allChildren.size()) {
        if (!(allChildren.get(childIndex) instanceof ExchangeNode)) {
          childNum--;
          // Try to keep the first of a pipeline is not a ExchangeNode
          if (childNum == -1) {
            childIndex--;
          }
        }
        childIndex++;
      }
      childNumInEachPipeline[pipelineIndex++] = childIndex - originChildIndex;
    }
    return childNumInEachPipeline;
  }

  private Operator createNewPipelineForChildNode(
      LocalExecutionPlanContext context, LocalExecutionPlanContext subContext, PlanNode childNode) {
    Operator childOperation = childNode.accept(this, subContext);
    ISinkChannel localSinkChannel =
        MPP_DATA_EXCHANGE_MANAGER.createLocalSinkChannelForPipeline(
            // Attention, there is no parent node, use first child node instead
            subContext.getDriverContext(), childNode.getPlanNodeId().getId());
    subContext.setISink(localSinkChannel);
    subContext.addPipelineDriverFactory(childOperation, subContext.getDriverContext(), 0);

    ExchangeOperator sourceOperator =
        new ExchangeOperator(
            context
                .getDriverContext()
                .addOperatorContext(
                    context.getNextOperatorId(), null, ExchangeOperator.class.getSimpleName()),
            MPP_DATA_EXCHANGE_MANAGER.createLocalSourceHandleForPipeline(
                ((LocalSinkChannel) localSinkChannel).getSharedTsBlockQueue(),
                context.getDriverContext()),
            childNode.getPlanNodeId(),
            childOperation.calculateMaxReturnSize());

    context.addExchangeOperator(sourceOperator);
    return sourceOperator;
  }

  public List<Operator> dealWithConsumeChildrenOneByOneNode(
      PlanNode node, LocalExecutionPlanContext context) {
    List<Operator> parentPipelineChildren = new ArrayList<>();
    int originExchangeNum = context.getExchangeSumNum();
    int finalExchangeNum = context.getExchangeSumNum();

    // 1. divide every child to pipeline using the max dop
    if (context.getDegreeOfParallelism() == 1 || node.getChildren().size() == 1) {
      // If dop = 1, we don't create extra pipeline
      for (PlanNode childSource : node.getChildren()) {
        Operator childOperation = childSource.accept(this, context);
        finalExchangeNum = Math.max(finalExchangeNum, context.getExchangeSumNum());
        context.setExchangeSumNum(originExchangeNum);
        parentPipelineChildren.add(childOperation);
      }
    } else {
      List<Integer> childPipelineNums = new ArrayList<>();
      List<Integer> childExchangeNums = new ArrayList<>();
      int sumOfChildPipelines = 0;
      int sumOfChildExchangeNums = 0;
      int dependencyChildNode = 0;
      int dependencyPipeId = 0;
      for (PlanNode childNode : node.getChildren()) {
        if (childNode instanceof ExchangeNode) {
          Operator childOperation = childNode.accept(this, context);
          finalExchangeNum = Math.max(finalExchangeNum, context.getExchangeSumNum());
          context.setExchangeSumNum(originExchangeNum);
          parentPipelineChildren.add(childOperation);
        } else {
          LocalExecutionPlanContext subContext = context.createSubContext();
          // Only context.getDegreeOfParallelism() - 1 can be allocated to child
          int dopForChild = context.getDegreeOfParallelism() - 1;
          subContext.setDegreeOfParallelism(dopForChild);
          int originPipeNum = context.getPipelineNumber();
          Operator childOperation = childNode.accept(this, subContext);
          ISinkChannel localSinkChannel =
              MPP_DATA_EXCHANGE_MANAGER.createLocalSinkChannelForPipeline(
                  // Attention, there is no parent node, use first child node instead
                  context.getDriverContext(), childNode.getPlanNodeId().getId());
          subContext.setISink(localSinkChannel);
          subContext.addPipelineDriverFactory(childOperation, subContext.getDriverContext(), 0);

          // OneByOneChild may be divided into more than dop pipelines, but the number of running
          // actually is dop
          int curChildPipelineNum =
              Math.min(dopForChild, subContext.getPipelineNumber() - originPipeNum);
          childPipelineNums.add(curChildPipelineNum);
          sumOfChildPipelines += curChildPipelineNum;
          // If sumOfChildPipelines > dopForChild, we have to wait until some pipelines finish
          if (sumOfChildPipelines > dopForChild) {
            // Update dependencyPipeId, after which finishes we can submit curChildPipeline
            while (sumOfChildPipelines > dopForChild) {
              sumOfChildPipelines -= childPipelineNums.get(dependencyChildNode);
              // The dependency pipeline must be a parent pipeline rather than a child pipeline
              dependencyPipeId = context.getPipelineNumber() - sumOfChildPipelines - 1;
              sumOfChildExchangeNums -= childExchangeNums.get(dependencyChildNode);
              dependencyChildNode++;
            }
          }
          // Add dependency for all pipelines under current node
          if (dependencyChildNode != 0) {
            for (int i = originPipeNum; i < subContext.getPipelineNumber(); i++) {
              context.getPipelineDriverFactories().get(i).setDependencyPipeline(dependencyPipeId);
            }
          }

          ExchangeOperator sourceOperator =
              new ExchangeOperator(
                  context
                      .getDriverContext()
                      .addOperatorContext(
                          context.getNextOperatorId(),
                          null,
                          ExchangeOperator.class.getSimpleName()),
                  MPP_DATA_EXCHANGE_MANAGER.createLocalSourceHandleForPipeline(
                      ((LocalSinkChannel) localSinkChannel).getSharedTsBlockQueue(),
                      context.getDriverContext()),
                  childNode.getPlanNodeId(),
                  childOperation.calculateMaxReturnSize());
          context.getCurrentPipelineDriverFactory().setDownstreamOperator(sourceOperator);
          parentPipelineChildren.add(sourceOperator);
          context.addExchangeOperator(sourceOperator);
          int childExchangeNum = subContext.getExchangeSumNum() - context.getExchangeSumNum() + 1;
          sumOfChildExchangeNums += childExchangeNum;
          childExchangeNums.add(childExchangeNum);
          finalExchangeNum =
              Math.max(finalExchangeNum, context.getExchangeSumNum() + sumOfChildExchangeNums);
        }
      }
    }
    context.setExchangeSumNum(finalExchangeNum);
    return parentPipelineChildren;
  }

  @Override
  public Operator visitColumnInject(ColumnInjectNode node, LocalExecutionPlanContext context) {
    final OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                ColumnInjectNode.class.getSimpleName());

    Operator childOperator = node.getChild().accept(this, context);
    ColumnGeneratorParameter parameter = node.getColumnGeneratorParameter();
    ColumnGenerator columnGenerator = genColumnGeneratorAccordingToParameter(parameter);
    long maxExtraColumnSize = 0;
    for (TSDataType dataType : node.getGeneratedColumnTypes()) {
      maxExtraColumnSize += getOutputColumnSizePerLine(dataType);
    }
    maxExtraColumnSize *= TSFileDescriptor.getInstance().getConfig().getMaxTsBlockLineNumber();

    return new ColumnInjectOperator(
        operatorContext, childOperator, columnGenerator, node.getTargetIndex(), maxExtraColumnSize);
  }

  private ColumnGenerator genColumnGeneratorAccordingToParameter(
      ColumnGeneratorParameter columnGeneratorParameter) {
    ColumnGeneratorType type = columnGeneratorParameter.getGeneratorType();
    if (type == ColumnGeneratorType.SLIDING_TIME) {
      SlidingTimeColumnGeneratorParameter slidingTimeColumnGeneratorParameter =
          (SlidingTimeColumnGeneratorParameter) columnGeneratorParameter;
      return new SlidingTimeColumnGenerator(
          initTimeRangeIterator(
              slidingTimeColumnGeneratorParameter.getGroupByTimeParameter(),
              slidingTimeColumnGeneratorParameter.isAscending(),
              false));
    } else {
      throw new UnsupportedOperationException("Unsupported column generator type: " + type);
    }
  }

  @Override
  public Operator visitExplainAnalyze(ExplainAnalyzeNode node, LocalExecutionPlanContext context) {
    Operator operator = node.getChild().accept(this, context);
    OperatorContext operatorContext =
        context
            .getDriverContext()
            .addOperatorContext(
                context.getNextOperatorId(),
                node.getPlanNodeId(),
                ExplainAnalyzeOperator.class.getSimpleName());
    return new ExplainAnalyzeOperator(
        operatorContext, operator, node.getQueryId(), node.isVerbose(), node.getTimeout());
  }
}