

package io.crate.operation.projectors;

import com.google.common.base.Preconditions;
import io.crate.core.collections.Row;
import io.crate.operation.Input;
import io.crate.operation.collect.CollectExpression;

import java.util.List;

public class SimpleTopNProjector extends InputRowProjector {

    private int remainingOffset;
    private int toCollect;

    public SimpleTopNProjector(List<Input<?>> inputs,
                               Iterable<? extends CollectExpression<Row, ?>> collectExpressions,
                               int limit,
                               int offset) {
        super(inputs, collectExpressions);

        Preconditions.checkArgument(limit >= 0, "invalid limit");
        Preconditions.checkArgument(offset>=0, "invalid offset");
        this.remainingOffset = offset;
        this.toCollect = limit;
    }

    @Override
    public boolean setNextRow(Row row) {
        if (toCollect < 1){
            return false;
        }
        if (remainingOffset > 0) {
            remainingOffset--;
            return true;
        }
        boolean wantMore = super.setNextRow(row);
        if (!wantMore) {
            toCollect = -1;
            return false;
        } else {
            toCollect--;
            return toCollect > 0;
        }
    }
}
