<commit_before>/*
 * Copyright 2004-present Facebook. All Rights Reserved.
 */
package com.facebook.presto.operator;

import com.facebook.presto.block.Block;
import com.facebook.presto.block.BlockCursor;
import com.facebook.presto.block.BlockIterable;
import com.facebook.presto.block.BlockIterables;
import com.facebook.presto.tuple.TupleInfo;
import com.google.common.base.Optional;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterables;
import io.airlift.units.DataSize;

import java.util.Iterator;
import java.util.List;

import static com.google.common.base.Preconditions.checkState;

public class AlignmentOperator implements Operator
{
    private final BlockIterable[] channels;
    private final List<TupleInfo> tupleInfos;
    private Optional<DataSize> expectedDataSize;
    private Optional<Integer> expectedPositionCount;

    public AlignmentOperator(BlockIterable... channels)
    {
        this.channels = channels;
        ImmutableList.Builder<TupleInfo> tupleInfos = ImmutableList.builder();
        for (BlockIterable channel : channels) {
            tupleInfos.add(channel.getTupleInfo());
        }
        this.tupleInfos = tupleInfos.build();
        expectedDataSize = BlockIterables.getDataSize(channels);
        expectedPositionCount = BlockIterables.getPositionCount(channels);
    }

    public AlignmentOperator(Iterable<BlockIterable> channels)
    {
        this.channels = Iterables.toArray(channels, BlockIterable.class);
        ImmutableList.Builder<TupleInfo> tupleInfos = ImmutableList.builder();
        for (BlockIterable channel : channels) {
            tupleInfos.add(channel.getTupleInfo());
        }
        this.tupleInfos = tupleInfos.build();
        expectedDataSize = BlockIterables.getDataSize(channels);
        expectedPositionCount = BlockIterables.getPositionCount(channels);
    }

    @Override
    public int getChannelCount()
    {
        return channels.length;
    }

    @Override
    public List<TupleInfo> getTupleInfos()
    {
        return tupleInfos;
    }

    public Optional<DataSize> getExpectedDataSize()
    {
        return expectedDataSize;
    }

    public Optional<Integer> getExpectedPositionCount()
    {
        return expectedPositionCount;
    }

    @Override
    public PageIterator iterator(OperatorStats operatorStats)
    {
        Iterator<? extends Block>[] iterators = new Iterator[channels.length];
        for (int i = 0; i < iterators.length; i++) {
            iterators[i] = channels[i].iterator();
        }

        // TODO: remove this hack after empty blocks are supported
        if (!iterators[0].hasNext()) {
            for (Iterator<? extends Block> iterator : iterators) {
                checkState(!iterator.hasNext(), "iterators are not aligned");
            }
            return PageIterators.emptyIterator(tupleInfos);
        }

        if (expectedDataSize.isPresent()) {
            operatorStats.addCompletedDataSize(expectedDataSize.get().toBytes());
        }
        operatorStats.addCompletedPositions(expectedPositionCount.or(0));
        return new AlignmentIterator(tupleInfos, iterators, operatorStats);
    }

    public static class AlignmentIterator
            extends AbstractPageIterator
    {
        private final Iterator<? extends Block>[] iterators;
        private final OperatorStats operatorStats;
        private final BlockCursor[] cursors;

        public AlignmentIterator(List<TupleInfo> tupleInfos, Iterator<? extends Block>[] iterators, OperatorStats operatorStats)
        {
            super(tupleInfos);

            this.iterators = iterators;
            this.operatorStats = operatorStats;

            cursors = new BlockCursor[iterators.length];
            for (int i = 0; i < iterators.length; i++) {
                cursors[i] = iterators[i].next().cursor();
            }
        }

        protected Page computeNext()<commit_msg>Only declare completed rows and bytes once<commit_after>/*
 * Copyright 2004-present Facebook. All Rights Reserved.
 */
package com.facebook.presto.operator;

import com.facebook.presto.block.Block;
import com.facebook.presto.block.BlockCursor;
import com.facebook.presto.block.BlockIterable;
import com.facebook.presto.block.BlockIterables;
import com.facebook.presto.tuple.TupleInfo;
import com.google.common.base.Optional;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterables;
import io.airlift.units.DataSize;

import java.util.Iterator;
import java.util.List;

import static com.google.common.base.Preconditions.checkState;

public class AlignmentOperator implements Operator
{
    private final BlockIterable[] channels;
    private final List<TupleInfo> tupleInfos;
    private Optional<DataSize> expectedDataSize;
    private Optional<Integer> expectedPositionCount;

    public AlignmentOperator(BlockIterable... channels)
    {
        this.channels = channels;
        ImmutableList.Builder<TupleInfo> tupleInfos = ImmutableList.builder();
        for (BlockIterable channel : channels) {
            tupleInfos.add(channel.getTupleInfo());
        }
        this.tupleInfos = tupleInfos.build();
        expectedDataSize = BlockIterables.getDataSize(channels);
        expectedPositionCount = BlockIterables.getPositionCount(channels);
    }

    public AlignmentOperator(Iterable<BlockIterable> channels)
    {
        this.channels = Iterables.toArray(channels, BlockIterable.class);
        ImmutableList.Builder<TupleInfo> tupleInfos = ImmutableList.builder();
        for (BlockIterable channel : channels) {
            tupleInfos.add(channel.getTupleInfo());
        }
        this.tupleInfos = tupleInfos.build();
        expectedDataSize = BlockIterables.getDataSize(channels);
        expectedPositionCount = BlockIterables.getPositionCount(channels);
    }

    @Override
    public int getChannelCount()
    {
        return channels.length;
    }

    @Override
    public List<TupleInfo> getTupleInfos()
    {
        return tupleInfos;
    }

    public Optional<DataSize> getExpectedDataSize()
    {
        return expectedDataSize;
    }

    public Optional<Integer> getExpectedPositionCount()
    {
        return expectedPositionCount;
    }

    @Override
    public PageIterator iterator(OperatorStats operatorStats)
    {
        Iterator<? extends Block>[] iterators = new Iterator[channels.length];
        for (int i = 0; i < iterators.length; i++) {
            iterators[i] = channels[i].iterator();
        }

        // TODO: remove this hack after empty blocks are supported
        if (!iterators[0].hasNext()) {
            for (Iterator<? extends Block> iterator : iterators) {
                checkState(!iterator.hasNext(), "iterators are not aligned");
            }
            return PageIterators.emptyIterator(tupleInfos);
        }

        return new AlignmentIterator(tupleInfos, iterators, operatorStats);
    }

    public static class AlignmentIterator
            extends AbstractPageIterator
    {
        private final Iterator<? extends Block>[] iterators;
        private final OperatorStats operatorStats;
        private final BlockCursor[] cursors;

        public AlignmentIterator(List<TupleInfo> tupleInfos, Iterator<? extends Block>[] iterators, OperatorStats operatorStats)
        {
            super(tupleInfos);

            this.iterators = iterators;
            this.operatorStats = operatorStats;

            cursors = new BlockCursor[iterators.length];
            for (int i = 0; i < iterators.length; i++) {
                cursors[i] = iterators[i].next().cursor();
            }
        }

        protected Page computeNext()
{'fraction_non_alphanumeric': 0.09350175804141164, 'fraction_numerical': 0.0019533793462690454, 'mean_word_length': 2.9648941662364483, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6126681', 'n_tokens_mistral': 2043, 'n_tokens_neox': 1996, 'n_words': 433}