/**
 * 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.storm.trident;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import org.apache.storm.generated.Grouping;
import org.apache.storm.generated.NullStruct;
import org.apache.storm.generated.SharedMemory;
import org.apache.storm.grouping.CustomStreamGrouping;
import org.apache.storm.topology.ResourceDeclarer;
import org.apache.storm.topology.base.BaseWindowedBolt;
import org.apache.storm.trident.fluent.ChainedAggregatorDeclarer;
import org.apache.storm.trident.fluent.GlobalAggregationScheme;
import org.apache.storm.trident.fluent.GroupedStream;
import org.apache.storm.trident.fluent.IAggregatableStream;
import org.apache.storm.trident.operation.Aggregator;
import org.apache.storm.trident.operation.Assembly;
import org.apache.storm.trident.operation.CombinerAggregator;
import org.apache.storm.trident.operation.Consumer;
import org.apache.storm.trident.operation.Filter;
import org.apache.storm.trident.operation.FlatMapFunction;
import org.apache.storm.trident.operation.Function;
import org.apache.storm.trident.operation.MapFunction;
import org.apache.storm.trident.operation.ReducerAggregator;
import org.apache.storm.trident.operation.builtin.ComparisonAggregator;
import org.apache.storm.trident.operation.builtin.Max;
import org.apache.storm.trident.operation.builtin.MaxWithComparator;
import org.apache.storm.trident.operation.builtin.Min;
import org.apache.storm.trident.operation.builtin.MinWithComparator;
import org.apache.storm.trident.operation.impl.CombinerAggStateUpdater;
import org.apache.storm.trident.operation.impl.ConsumerExecutor;
import org.apache.storm.trident.operation.impl.FilterExecutor;
import org.apache.storm.trident.operation.impl.FlatMapFunctionExecutor;
import org.apache.storm.trident.operation.impl.GlobalBatchToPartition;
import org.apache.storm.trident.operation.impl.IndexHashBatchToPartition;
import org.apache.storm.trident.operation.impl.MapFunctionExecutor;
import org.apache.storm.trident.operation.impl.ReducerAggStateUpdater;
import org.apache.storm.trident.operation.impl.SingleEmitAggregator.BatchToPartition;
import org.apache.storm.trident.operation.impl.TrueFilter;
import org.apache.storm.trident.partition.GlobalGrouping;
import org.apache.storm.trident.partition.IdentityGrouping;
import org.apache.storm.trident.partition.IndexHashGrouping;
import org.apache.storm.trident.planner.Node;
import org.apache.storm.trident.planner.NodeStateInfo;
import org.apache.storm.trident.planner.PartitionNode;
import org.apache.storm.trident.planner.ProcessorNode;
import org.apache.storm.trident.planner.processor.AggregateProcessor;
import org.apache.storm.trident.planner.processor.EachProcessor;
import org.apache.storm.trident.planner.processor.MapProcessor;
import org.apache.storm.trident.planner.processor.PartitionPersistProcessor;
import org.apache.storm.trident.planner.processor.ProjectedProcessor;
import org.apache.storm.trident.planner.processor.StateQueryProcessor;
import org.apache.storm.trident.state.QueryFunction;
import org.apache.storm.trident.state.StateFactory;
import org.apache.storm.trident.state.StateSpec;
import org.apache.storm.trident.state.StateUpdater;
import org.apache.storm.trident.tuple.TridentTuple;
import org.apache.storm.trident.util.TridentUtils;
import org.apache.storm.trident.windowing.InMemoryWindowsStoreFactory;
import org.apache.storm.trident.windowing.WindowTridentProcessor;
import org.apache.storm.trident.windowing.WindowsStateFactory;
import org.apache.storm.trident.windowing.WindowsStateUpdater;
import org.apache.storm.trident.windowing.WindowsStoreFactory;
import org.apache.storm.trident.windowing.config.SlidingCountWindow;
import org.apache.storm.trident.windowing.config.SlidingDurationWindow;
import org.apache.storm.trident.windowing.config.TumblingCountWindow;
import org.apache.storm.trident.windowing.config.TumblingDurationWindow;
import org.apache.storm.trident.windowing.config.WindowConfig;
import org.apache.storm.tuple.Fields;
import org.apache.storm.utils.Utils;

/**
 * A Stream represents the core data model in Trident, and can be thought of as a "stream" of tuples that are processed as a series of small
 * batches. A stream is partitioned accross the nodes in the cluster, and operations are applied to a stream in parallel accross each
 * partition.
 *
 * <p>There are five types of operations that can be performed on streams in Trident
 *
 * <p>1. **Partiton-Local Operations** - Operations that are applied locally to each partition and do not involve network transfer 2.
 * **Repartitioning Operations** - Operations that change how tuples are partitioned across tasks(thus causing network transfer), but do not
 * change the content of the stream. 3. **Aggregation Operations** - Operations that *may* repartition a stream (thus causing network
 * transfer) 4. **Grouping Operations** - Operations that may repartition a stream on specific fields and group together tuples whose fields
 * values are equal. 5. **Merge and Join Operations** - Operations that combine different streams together.
 */
// TODO: need to be able to replace existing fields with the function fields (like Cascading Fields.REPLACE)
public class Stream implements IAggregatableStream, ResourceDeclarer<Stream> {
    final Node node;
    final String name;
    private final TridentTopology topology;

    protected Stream(TridentTopology topology, String name, Node node) {
        this.topology = topology;
        this.node = node;
        this.name = name;
    }

    /**
     * Applies a label to the stream. Naming a stream will append the label to the name of the bolt(s) created by Trident and will be
     * visible in the Storm UI.
     *
     * @param name - The label to apply to the stream
     */
    public Stream name(String name) {
        return new Stream(topology, name, node);
    }

    /**
     * Applies a parallelism hint to a stream.
     */
    public Stream parallelismHint(int hint) {
        node.parallelismHint = hint;
        return this;
    }

    /**
     * Sets the CPU Load resource for the current operation.
     */
    @Override
    public Stream setCPULoad(Number load) {
        node.setCPULoad(load);
        return this;
    }

    /**
     * Sets the Memory Load resources for the current operation. offHeap becomes default.
     */
    @Override
    public Stream setMemoryLoad(Number onHeap) {
        node.setMemoryLoad(onHeap);
        return this;
    }

    /**
     * Sets the Memory Load resources for the current operation.
     */
    @Override
    public Stream setMemoryLoad(Number onHeap, Number offHeap) {
        node.setMemoryLoad(onHeap, offHeap);
        return this;
    }

    @Override
    public Stream addSharedMemory(SharedMemory request) {
        node.addSharedMemory(request);
        return this;
    }

    /**
     * Filters out fields from a stream, resulting in a Stream containing only the fields specified by `keepFields`.
     *
     * <p>For example, if you had a Stream `mystream` containing the fields `["a", "b", "c","d"]`, calling"
     *
     * <p>```java mystream.project(new Fields("b", "d")) ```
     *
     * <p>would produce a stream containing only the fields `["b", "d"]`.
     *
     * @param keepFields The fields in the Stream to keep
     */
    public Stream project(Fields keepFields) {
        projectionValidation(keepFields);
        return topology.addSourcedNode(this,
                new ProcessorNode(topology.getUniqueStreamId(),
                        name,
                        keepFields,
                        new Fields(),
                        new ProjectedProcessor(keepFields)));
    }

    /**
     * ## Grouping Operation.
     */
    public GroupedStream groupBy(Fields fields) {
        projectionValidation(fields);
        return new GroupedStream(this, fields);
    }

    /**
     * ## Repartitioning Operation.
     */
    public Stream partitionBy(Fields fields) {
        projectionValidation(fields);
        return partition(Grouping.fields(fields.toList()));
    }

    /**
     * ## Repartitioning Operation.
     */
    public Stream partition(CustomStreamGrouping partitioner) {
        return partition(Grouping.custom_serialized(Utils.javaSerialize(partitioner)));
    }

    /**
     * ## Repartitioning Operation.
     *
     * <p>This method takes in a custom partitioning function that implements {@link org.apache.storm.grouping.CustomStreamGrouping}
     */
    public Stream partition(Grouping grouping) {
        if (node instanceof PartitionNode) {
            return each(new Fields(), new TrueFilter()).partition(grouping);
        } else {
            return topology.addSourcedNode(this, new PartitionNode(node.streamId, name, getOutputFields(), grouping));
        }
    }

    /**
     * ## Repartitioning Operation.
     *
     * <p>Use random round robin algorithm to evenly redistribute tuples across all target partitions.
     */
    public Stream shuffle() {
        return partition(Grouping.shuffle(new NullStruct()));
    }

    /**
     * ## Repartitioning Operation.
     *
     * <p>Use random round robin algorithm to evenly redistribute tuples across all target partitions, with a preference for local tasks.
     */
    public Stream localOrShuffle() {
        return partition(Grouping.local_or_shuffle(new NullStruct()));
    }


    /**
     * ## Repartitioning Operation.
     *
     * <p>All tuples are sent to the same partition. The same partition is chosen for all batches in the stream.
     */
    public Stream global() {
        // use this instead of storm's built in one so that we can specify a singleemitbatchtopartition
        // without knowledge of storm's internals
        return partition(new GlobalGrouping());
    }

    /**
     * ## Repartitioning Operation.
     *
     * <p>All tuples in the batch are sent to the same partition. Different batches in the stream may go to different partitions.
     */
    public Stream batchGlobal() {
        // the first field is the batch id
        return partition(new IndexHashGrouping(0));
    }

    /**
     * ## Repartitioning Operation.
     *
     * <p>Every tuple is replicated to all target partitions. This can useful during DRPC – for example, if you need to do a stateQuery on
     * every partition of data.
     */
    public Stream broadcast() {
        return partition(Grouping.all(new NullStruct()));
    }

    /**
     * ## Repartitioning Operation.
     */
    public Stream identityPartition() {
        return partition(new IdentityGrouping());
    }

    /**
     * Applies an `Assembly` to this `Stream`.
     *
     * @see org.apache.storm.trident.operation.Assembly
     */
    public Stream applyAssembly(Assembly assembly) {
        return assembly.apply(this);
    }

    @Override
    public Stream each(Fields inputFields, Function function, Fields functionFields) {
        projectionValidation(inputFields);
        return topology.addSourcedNode(this,
                new ProcessorNode(topology.getUniqueStreamId(),
                        name,
                        TridentUtils.fieldsConcat(getOutputFields(), functionFields),
                        functionFields,
                        new EachProcessor(inputFields, function)));
    }

    public Stream each(Function function, Fields functionFields) {
        return each(null, function, functionFields);
    }

    public Stream each(Fields inputFields, Filter filter) {
        return each(inputFields, new FilterExecutor(filter), new Fields());
    }

    //creates brand new tuples with brand new fields
    @Override
    public Stream partitionAggregate(Fields inputFields, Aggregator agg, Fields functionFields) {
        projectionValidation(inputFields);
        return topology.addSourcedNode(this,
                new ProcessorNode(topology.getUniqueStreamId(),
                        name,
                        functionFields,
                        functionFields,
                        new AggregateProcessor(inputFields, agg)));
    }

    public Stream partitionAggregate(Aggregator agg, Fields functionFields) {
        return partitionAggregate(null, agg, functionFields);
    }

    public Stream partitionAggregate(CombinerAggregator agg, Fields functionFields) {
        return partitionAggregate(null, agg, functionFields);
    }

    public Stream partitionAggregate(Fields inputFields, CombinerAggregator agg, Fields functionFields) {
        projectionValidation(inputFields);
        return chainedAgg()
                .partitionAggregate(inputFields, agg, functionFields)
                .chainEnd();
    }

    public Stream partitionAggregate(ReducerAggregator agg, Fields functionFields) {
        return partitionAggregate(null, agg, functionFields);
    }

    public Stream partitionAggregate(Fields inputFields, ReducerAggregator agg, Fields functionFields) {
        projectionValidation(inputFields);
        return chainedAgg()
                .partitionAggregate(inputFields, agg, functionFields)
                .chainEnd();
    }

    public Stream stateQuery(TridentState state, Fields inputFields, QueryFunction function, Fields functionFields) {
        projectionValidation(inputFields);
        String stateId = state.node.stateInfo.id;
        Node n = new ProcessorNode(topology.getUniqueStreamId(),
                name,
                TridentUtils.fieldsConcat(getOutputFields(), functionFields),
                functionFields,
                new StateQueryProcessor(stateId, inputFields, function));
        topology.colocate.get(stateId).add(n);
        return topology.addSourcedNode(this, n);
    }

    public Stream stateQuery(TridentState state, QueryFunction function, Fields functionFields) {
        return stateQuery(state, null, function, functionFields);
    }

    public TridentState partitionPersist(StateFactory stateFactory, Fields inputFields, StateUpdater updater, Fields functionFields) {
        return partitionPersist(new StateSpec(stateFactory), inputFields, updater, functionFields);
    }

    public TridentState partitionPersist(StateSpec stateSpec, Fields inputFields, StateUpdater updater, Fields functionFields) {
        projectionValidation(inputFields);
        String id = topology.getUniqueStateId();
        ProcessorNode n = new ProcessorNode(topology.getUniqueStreamId(),
                name,
                functionFields,
                functionFields,
                new PartitionPersistProcessor(id, inputFields, updater));
        n.committer = true;
        n.stateInfo = new NodeStateInfo(id, stateSpec);
        return topology.addSourcedStateNode(this, n);
    }

    public TridentState partitionPersist(StateFactory stateFactory, Fields inputFields, StateUpdater updater) {
        return partitionPersist(stateFactory, inputFields, updater, new Fields());
    }

    public TridentState partitionPersist(StateSpec stateSpec, Fields inputFields, StateUpdater updater) {
        return partitionPersist(stateSpec, inputFields, updater, new Fields());
    }

    public TridentState partitionPersist(StateFactory stateFactory, StateUpdater updater, Fields functionFields) {
        return partitionPersist(new StateSpec(stateFactory), updater, functionFields);
    }

    public TridentState partitionPersist(StateSpec stateSpec, StateUpdater updater, Fields functionFields) {
        return partitionPersist(stateSpec, null, updater, functionFields);
    }

    public TridentState partitionPersist(StateFactory stateFactory, StateUpdater updater) {
        return partitionPersist(stateFactory, updater, new Fields());
    }

    public TridentState partitionPersist(StateSpec stateSpec, StateUpdater updater) {
        return partitionPersist(stateSpec, updater, new Fields());
    }

    /**
     * Returns a stream consisting of the elements of this stream that match the given filter.
     *
     * @param filter the filter to apply to each trident tuple to determine if it should be included.
     * @return the new stream
     */
    public Stream filter(Filter filter) {
        return each(getOutputFields(), filter);
    }

    /**
     * Returns a stream consisting of the elements of this stream that match the given filter.
     *
     * @param inputFields the fields of the input trident tuple to be selected.
     * @param filter      the filter to apply to each trident tuple to determine if it should be included.
     * @return the new stream
     */
    public Stream filter(Fields inputFields, Filter filter) {
        return each(inputFields, filter);
    }

    /**
     * Returns a stream consisting of the result of applying the given mapping function to the values of this stream.
     *
     * @param function a mapping function to be applied to each value in this stream.
     * @return the new stream
     */
    public Stream map(MapFunction function) {
        projectionValidation(getOutputFields());
        return topology.addSourcedNode(this,
                new ProcessorNode(topology.getUniqueStreamId(),
                        name,
                        getOutputFields(),
                        getOutputFields(),
                        new MapProcessor(getOutputFields(), new MapFunctionExecutor(function))));
    }

    /**
     * Returns a stream consisting of the result of applying the given mapping function to the values of this stream. This method replaces
     * old output fields with new output fields, achieving T -> V conversion.
     *
     * @param function     a mapping function to be applied to each value in this stream.
     * @param outputFields new output fields
     * @return the new stream
     */
    public Stream map(MapFunction function, Fields outputFields) {
        projectionValidation(getOutputFields());
        return topology.addSourcedNode(this,
                new ProcessorNode(topology.getUniqueStreamId(),
                        name,
                        outputFields,
                        outputFields,
                        new MapProcessor(getOutputFields(), new MapFunctionExecutor(function))));
    }

    /**
     * Returns a stream consisting of the results of replacing each value of this stream with the contents produced by applying the provided
     * mapping function to each value. This has the effect of applying a one-to-many transformation to the values of the stream, and then
     * flattening the resulting elements into a new stream.
     *
     * @param function a mapping function to be applied to each value in this stream which produces new values.
     * @return the new stream
     */
    public Stream flatMap(FlatMapFunction function) {
        projectionValidation(getOutputFields());
        return topology.addSourcedNode(this,
                new ProcessorNode(topology.getUniqueStreamId(),
                        name,
                        getOutputFields(),
                        getOutputFields(),
                        new MapProcessor(getOutputFields(), new FlatMapFunctionExecutor(function))));
    }

    /**
     * Returns a stream consisting of the results of replacing each value of this stream with the contents produced by applying the provided
     * mapping function to each value. This has the effect of applying a one-to-many transformation to the values of the stream, and then
     * flattening the resulting elements into a new stream. This method replaces old output fields with new output fields, achieving T -> V
     * conversion.
     *
     * @param function     a mapping function to be applied to each value in this stream which produces new values.
     * @param outputFields new output fields
     * @return the new stream
     */
    public Stream flatMap(FlatMapFunction function, Fields outputFields) {
        projectionValidation(getOutputFields());
        return topology.addSourcedNode(this,
                new ProcessorNode(topology.getUniqueStreamId(),
                        name,
                        outputFields,
                        outputFields,
                        new MapProcessor(getOutputFields(), new FlatMapFunctionExecutor(function))));
    }

    /**
     * Returns a stream consisting of the trident tuples of this stream, additionally performing the provided action on each trident tuple
     * as they are consumed from the resulting stream. This is mostly useful for debugging to see the tuples as they flow past a certain
     * point in a pipeline.
     *
     * @param action the action to perform on the trident tuple as they are consumed from the stream
     * @return the new stream
     */
    public Stream peek(Consumer action) {
        projectionValidation(getOutputFields());
        return topology.addSourcedNode(this,
                new ProcessorNode(topology.getUniqueStreamId(),
                        name,
                        getOutputFields(),
                        getOutputFields(),
                        new MapProcessor(getOutputFields(), new ConsumerExecutor(action))));
    }

    public ChainedAggregatorDeclarer chainedAgg() {
        return new ChainedAggregatorDeclarer(this, new BatchGlobalAggScheme());
    }

    /**
     * This aggregator operation computes the minimum of tuples by the given {@code inputFieldName} and it is assumed that its value is an
     * instance of {@code Comparable}. If the value of tuple with field {@code inputFieldName} is not an instance of {@code Comparable} then
     * it throws {@code ClassCastException}
     *
     * @param inputFieldName input field name
     * @return the new stream with this operation.
     */
    public Stream minBy(String inputFieldName) {
        Aggregator<ComparisonAggregator.State> min = new Min(inputFieldName);
        return comparableAggregateStream(inputFieldName, min);
    }

    /**
     * This aggregator operation computes the minimum of tuples by the given {@code inputFieldName} in a stream by using the given {@code
     * comparator}. If the value of tuple with field {@code inputFieldName} is not an instance of {@code T} then it throws {@code
     * ClassCastException}
     *
     * @param inputFieldName input field name
     * @param comparator     comparator used in for finding minimum of two tuple values of {@code inputFieldName}.
     * @return the new stream with this operation.
     */
    public <T> Stream minBy(String inputFieldName, Comparator<T> comparator) {
        Aggregator<ComparisonAggregator.State> min = new MinWithComparator<>(inputFieldName, comparator);
        return comparableAggregateStream(inputFieldName, min);
    }

    /**
     * This aggregator operation computes the minimum of tuples in a stream by using the given {@code comparator} with {@code
     * TridentTuple}s.
     *
     * @param comparator comparator used in for finding minimum of two tuple values.
     * @return the new stream with this operation.
     */
    public Stream min(Comparator<TridentTuple> comparator) {
        Aggregator<ComparisonAggregator.State> min = new MinWithComparator<>(comparator);
        return comparableAggregateStream(null, min);
    }

    /**
     * This aggregator operation computes the maximum of tuples by the given {@code inputFieldName} and it is assumed that its value is an
     * instance of {@code Comparable}. If the value of tuple with field {@code inputFieldName} is not an instance of {@code Comparable} then
     * it throws {@code ClassCastException}
     *
     * @param inputFieldName input field name
     * @return the new stream with this operation.
     */
    public Stream maxBy(String inputFieldName) {
        Aggregator<ComparisonAggregator.State> max = new Max(inputFieldName);
        return comparableAggregateStream(inputFieldName, max);
    }

    /**
     * This aggregator operation computes the maximum of tuples by the given {@code inputFieldName} in a stream by using the given {@code
     * comparator}. If the value of tuple with field {@code inputFieldName} is not an instance of {@code T} then it throws {@code
     * ClassCastException}
     *
     * @param inputFieldName input field name
     * @param comparator     comparator used in for finding maximum of two tuple values of {@code inputFieldName}.
     * @return the new stream with this operation.
     */
    public <T> Stream maxBy(String inputFieldName, Comparator<T> comparator) {
        Aggregator<ComparisonAggregator.State> max = new MaxWithComparator<>(inputFieldName, comparator);
        return comparableAggregateStream(inputFieldName, max);
    }

    /**
     * This aggregator operation computes the maximum of tuples in a stream by using the given {@code comparator} with {@code
     * TridentTuple}s.
     *
     * @param comparator comparator used in for finding maximum of two tuple values.
     * @return the new stream with this operation.
     */
    public Stream max(Comparator<TridentTuple> comparator) {
        Aggregator<ComparisonAggregator.State> max = new MaxWithComparator<>(comparator);
        return comparableAggregateStream(null, max);
    }

    private <T> Stream comparableAggregateStream(String inputFieldName, Aggregator<T> aggregator) {
        if (inputFieldName != null) {
            projectionValidation(new Fields(inputFieldName));
        }
        return partitionAggregate(getOutputFields(), aggregator, getOutputFields());
    }

    public Stream aggregate(Aggregator agg, Fields functionFields) {
        return aggregate(null, agg, functionFields);
    }

    public Stream aggregate(Fields inputFields, Aggregator agg, Fields functionFields) {
        projectionValidation(inputFields);
        return chainedAgg()
            .aggregate(inputFields, agg, functionFields)
            .chainEnd();
    }

    public Stream aggregate(CombinerAggregator agg, Fields functionFields) {
        return aggregate(null, agg, functionFields);
    }

    public Stream aggregate(Fields inputFields, CombinerAggregator agg, Fields functionFields) {
        projectionValidation(inputFields);
        return chainedAgg()
            .aggregate(inputFields, agg, functionFields)
            .chainEnd();
    }

    public Stream aggregate(ReducerAggregator agg, Fields functionFields) {
        return aggregate(null, agg, functionFields);
    }

    public Stream aggregate(Fields inputFields, ReducerAggregator agg, Fields functionFields) {
        projectionValidation(inputFields);
        return chainedAgg()
            .aggregate(inputFields, agg, functionFields)
            .chainEnd();
    }

    /**
     * Returns a stream of tuples which are aggregated results of a tumbling window with every {@code windowCount} of tuples.
     *
     * @param windowCount        represents number of tuples in the window
     * @param windowStoreFactory intermediary tuple store for storing windowing tuples
     * @param inputFields        projected fields for aggregator
     * @param aggregator         aggregator to run on the window of tuples to compute the result and emit to the stream.
     * @param functionFields     fields of values to emit with aggregation.
     * @return the new stream with this operation.
     */
    public Stream tumblingWindow(int windowCount, WindowsStoreFactory windowStoreFactory,
                                 Fields inputFields, Aggregator aggregator, Fields functionFields) {
        return window(TumblingCountWindow.of(windowCount), windowStoreFactory, inputFields, aggregator, functionFields);
    }

    /**
     * Returns a stream of tuples which are aggregated results of a window that tumbles at duration of {@code windowDuration}.
     *
     * @param windowDuration     represents tumbling window duration configuration
     * @param windowStoreFactory intermediary tuple store for storing windowing tuples
     * @param inputFields        projected fields for aggregator
     * @param aggregator         aggregator to run on the window of tuples to compute the result and emit to the stream.
     * @param functionFields     fields of values to emit with aggregation.
     * @return the new stream with this operation.
     */
    public Stream tumblingWindow(BaseWindowedBolt.Duration windowDuration, WindowsStoreFactory windowStoreFactory,
            Fields inputFields, Aggregator aggregator, Fields functionFields) {
        return window(TumblingDurationWindow.of(windowDuration), windowStoreFactory, inputFields, aggregator, functionFields);
    }

    /**
     * Returns a stream of tuples which are aggregated results of a sliding window with every {@code windowCount} of tuples and slides the
     * window after {@code slideCount}.
     *
     * @param windowCount        represents tuples count of a window
     * @param slideCount         the number of tuples after which the window slides
     * @param windowStoreFactory intermediary tuple store for storing windowing tuples
     * @param inputFields        projected fields for aggregator
     * @param aggregator         aggregator to run on the window of tuples to compute the result and emit to the stream.
     * @param functionFields     fields of values to emit with aggregation.
     * @return the new stream with this operation.
     */
    public Stream slidingWindow(int windowCount, int slideCount, WindowsStoreFactory windowStoreFactory,
                                Fields inputFields, Aggregator aggregator, Fields functionFields) {
        return window(SlidingCountWindow.of(windowCount, slideCount), windowStoreFactory, inputFields, aggregator, functionFields);
    }

    /**
     * Returns a stream of tuples which are aggregated results of a window which slides at duration of {@code slidingInterval} and completes
     * a window at {@code windowDuration}.
     *
     * @param windowDuration     represents window duration configuration
     * @param slidingInterval    the time duration after which the window slides
     * @param windowStoreFactory intermediary tuple store for storing windowing tuples
     * @param inputFields        projected fields for aggregator
     * @param aggregator         aggregator to run on the window of tuples to compute the result and emit to the stream.
     * @param functionFields     fields of values to emit with aggregation.
     * @return the new stream with this operation.
     */
    public Stream slidingWindow(BaseWindowedBolt.Duration windowDuration, BaseWindowedBolt.Duration slidingInterval,
            WindowsStoreFactory windowStoreFactory, Fields inputFields, Aggregator aggregator, Fields functionFields) {
        return window(SlidingDurationWindow.of(windowDuration, slidingInterval), windowStoreFactory, inputFields, aggregator,
                functionFields);
    }

    /**
     * Returns a stream of aggregated results based on the given window configuration which uses inmemory windowing tuple store.
     *
     * @param windowConfig   window configuration like window length and slide length.
     * @param inputFields    input fields
     * @param aggregator     aggregator to run on the window of tuples to compute the result and emit to the stream.
     * @param functionFields fields of values to emit with aggregation.
     * @return the new stream with this operation.
     */
    public Stream window(WindowConfig windowConfig, Fields inputFields, Aggregator aggregator, Fields functionFields) {
        // this store is used only for storing triggered aggregated results but not tuples as storeTuplesInStore is set
        // as false int he below call.
        InMemoryWindowsStoreFactory inMemoryWindowsStoreFactory = new InMemoryWindowsStoreFactory();
        return window(windowConfig, inMemoryWindowsStoreFactory, inputFields, aggregator, functionFields, false);
    }

    /**
     * Returns stream of aggregated results based on the given window configuration.
     *
     * @param windowConfig       window configuration like window length and slide length.
     * @param windowStoreFactory intermediary tuple store for storing tuples for windowing
     * @param inputFields        input fields
     * @param aggregator         aggregator to run on the window of tuples to compute the result and emit to the stream.
     * @param functionFields     fields of values to emit with aggregation.
     * @return the new stream with this operation.
     */
    public Stream window(WindowConfig windowConfig, WindowsStoreFactory windowStoreFactory, Fields inputFields,
                         Aggregator aggregator, Fields functionFields) {
        return window(windowConfig, windowStoreFactory, inputFields, aggregator, functionFields, true);
    }

    private Stream window(WindowConfig windowConfig, WindowsStoreFactory windowStoreFactory, Fields inputFields, Aggregator aggregator,
                          Fields functionFields, boolean storeTuplesInStore) {
        projectionValidation(inputFields);
        windowConfig.validate();

        Fields fields = addTriggerField(functionFields);

        // when storeTuplesInStore is false then the given windowStoreFactory is only used to store triggers and
        // that store is passed to WindowStateUpdater to remove them after committing the batch.
        Stream stream = topology.addSourcedNode(this,
                new ProcessorNode(topology.getUniqueStreamId(),
                        name,
                        fields,
                        fields,
                        new WindowTridentProcessor(windowConfig,
                                topology.getUniqueWindowId(),
                                windowStoreFactory,
                                inputFields, aggregator,
                                storeTuplesInStore)));

        Stream effectiveStream = stream.project(functionFields);

        // create StateUpdater with the given windowStoreFactory to remove triggered aggregation results form store
        // when they are successfully processed.
        StateFactory stateFactory = new WindowsStateFactory();
        StateUpdater stateUpdater = new WindowsStateUpdater(windowStoreFactory);
        stream.partitionPersist(stateFactory, new Fields(WindowTridentProcessor.TRIGGER_FIELD_NAME), stateUpdater, new Fields());

        return effectiveStream;
    }

    private Fields addTriggerField(Fields functionFields) {
        List<String> fieldsList = new ArrayList<>();
        fieldsList.add(WindowTridentProcessor.TRIGGER_FIELD_NAME);
        for (String field : functionFields) {
            fieldsList.add(field);
        }
        return new Fields(fieldsList);
    }

    public TridentState persistentAggregate(StateFactory stateFactory, CombinerAggregator agg, Fields functionFields) {
        return persistentAggregate(new StateSpec(stateFactory), agg, functionFields);
    }

    public TridentState persistentAggregate(StateSpec spec, CombinerAggregator agg, Fields functionFields) {
        return persistentAggregate(spec, null, agg, functionFields);
    }

    public TridentState persistentAggregate(StateFactory stateFactory, Fields inputFields, CombinerAggregator agg, Fields functionFields) {
        return persistentAggregate(new StateSpec(stateFactory), inputFields, agg, functionFields);
    }

    public TridentState persistentAggregate(StateSpec spec, Fields inputFields, CombinerAggregator agg, Fields functionFields) {
        projectionValidation(inputFields);
        // replaces normal aggregation here with a global grouping because it needs to be consistent across batches 
        return new ChainedAggregatorDeclarer(this, new GlobalAggScheme())
            .aggregate(inputFields, agg, functionFields)
            .chainEnd()
            .partitionPersist(spec, functionFields, new CombinerAggStateUpdater(agg), functionFields);
    }

    public TridentState persistentAggregate(StateFactory stateFactory, ReducerAggregator agg, Fields functionFields) {
        return persistentAggregate(new StateSpec(stateFactory), agg, functionFields);
    }

    public TridentState persistentAggregate(StateSpec spec, ReducerAggregator agg, Fields functionFields) {
        return persistentAggregate(spec, null, agg, functionFields);
    }

    public TridentState persistentAggregate(StateFactory stateFactory, Fields inputFields, ReducerAggregator agg, Fields functionFields) {
        return persistentAggregate(new StateSpec(stateFactory), inputFields, agg, functionFields);
    }

    public TridentState persistentAggregate(StateSpec spec, Fields inputFields, ReducerAggregator agg, Fields functionFields) {
        projectionValidation(inputFields);
        return global().partitionPersist(spec, inputFields, new ReducerAggStateUpdater(agg), functionFields);
    }

    @Override
    public Stream toStream() {
        return this;
    }

    @Override
    public Fields getOutputFields() {
        return node.allOutputFields;
    }

    private void projectionValidation(Fields projFields) {
        if (projFields == null) {
            return;
        }

        Fields allFields = this.getOutputFields();
        for (String field : projFields) {
            if (!allFields.contains(field)) {
                throw new IllegalArgumentException(
                    "Trying to select non-existent field: '" + field + "' from stream containing fields fields: <" + allFields + ">");
            }
        }
    }

    static class BatchGlobalAggScheme implements GlobalAggregationScheme<Stream> {

        @Override
        public IAggregatableStream aggPartition(Stream s) {
            return s.batchGlobal();
        }

        @Override
        public BatchToPartition singleEmitPartitioner() {
            return new IndexHashBatchToPartition();
        }

    }

    static class GlobalAggScheme implements GlobalAggregationScheme<Stream> {

        @Override
        public IAggregatableStream aggPartition(Stream s) {
            return s.global();
        }

        @Override
        public BatchToPartition singleEmitPartitioner() {
            return new GlobalBatchToPartition();
        }

    }
}
