/*
 * Licensed 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 io.prestosql.spi.block;

import io.airlift.slice.DynamicSliceOutput;
import io.prestosql.spi.type.Type;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;

import java.util.Arrays;
import java.util.Random;
import java.util.stream.IntStream;

import static io.prestosql.spi.block.BlockTestUtils.assertBlockEquals;

public abstract class BaseBlockEncodingTest<T>
{
    private static final int[] RANDOM_BLOCK_SIZES = {2, 4, 8, 9, 16, 17, 32, 33, 64, 65, 1000, 1000000};

    private final BlockEncodingSerde blockEncodingSerde = new TestingBlockEncodingSerde();

    protected abstract Type getType();

    protected abstract void write(BlockBuilder blockBuilder, T value);

    protected abstract T randomValue(Random random);

    @Test
    public void testEmpty()
    {
        roundTrip();
    }

    @Test
    public void testSingleNull()
    {
        roundTrip((T) null);
    }

    @Test
    public void testSingleValue()
    {
        roundTrip(randomValue(getRandom()));
    }

    @Test
    public void testNullAtTheBeginningAndEnd()
    {
        Random random = getRandom();
        roundTrip(null, null, randomValue(random), null, randomValue(random), null, null);
    }

    @Test(dataProvider = "testRandomDataDataProvider")
    public void testRandomData(int size, BlockFill fill)
    {
        Random random = getRandom();
        Object[] values = new Object[size];
        for (int i = 0; i < size; i++) {
            switch (fill) {
                case ONLY_NULLS:
                    values[i] = null;
                    break;
                case ONLY_VALUES:
                    values[i] = randomValue(random);
                    break;
                case MIXED:
                    if (random.nextBoolean()) {
                        values[i] = null;
                    }
                    else {
                        values[i] = randomValue(random);
                    }
                    break;
            }
        }

        roundTrip(values);
    }

    @DataProvider
    public static Object[][] testRandomDataDataProvider()
    {
        return Arrays.stream(BlockFill.values())
                .flatMap(fill -> IntStream.of(RANDOM_BLOCK_SIZES).mapToObj(size -> new Object[] {size, fill}))
                .toArray(Object[][]::new);
    }

    protected final void roundTrip(Object... values)
    {
        BlockBuilder expectedBlockBuilder = createBlockBuilder(values.length);

        for (Object value : values) {
            if (value == null) {
                expectedBlockBuilder.appendNull();
            }
            else {
                //noinspection unchecked
                write(expectedBlockBuilder, (T) value);
            }
        }

        Block expectedBlock = expectedBlockBuilder.build();
        DynamicSliceOutput sliceOutput = new DynamicSliceOutput(1024);
        blockEncodingSerde.writeBlock(sliceOutput, expectedBlock);
        Block actualBlock = blockEncodingSerde.readBlock(sliceOutput.slice().getInput());
        assertBlockEquals(getType(), actualBlock, expectedBlock);
    }

    private BlockBuilder createBlockBuilder(int length)
    {
        return getType().createBlockBuilder(null, length);
    }

    private static Random getRandom()
    {
        return new Random(32167);
    }

    private enum BlockFill
    {
        ONLY_NULLS,
        ONLY_VALUES,
        MIXED,
        /**/
    }
}

@@ -832,16 +832,48 @@
 tStream;
+%0Aimport java.util.stream.Stream;
 %0A%0Aimport
@@ -1821,153 +1821,829 @@
 Test
-(dataProvider = %22testRandomDataDataProvider%22)%0A    public void testRandomData(int size, BlockFill fill)%0A    %7B%0A        Random random = getRandom();
+%0A    public void testBlocksOf8()%0A    %7B%0A        Random random = getRandom();%0A%0A        Object%5B%5D values = Stream.of(%0A                BlockFill.MIXED,%0A                BlockFill.ONLY_NULLS,%0A                BlockFill.MIXED,%0A                BlockFill.ONLY_VALUES,%0A                BlockFill.MIXED,%0A                BlockFill.ONLY_NULLS,%0A                BlockFill.ONLY_VALUES,%0A                BlockFill.MIXED)%0A                .map(fill -%3E getObjects(8, fill, random))%0A                .flatMap(Arrays::stream)%0A                .toArray();%0A%0A        roundTrip(values);%0A    %7D%0A%0A    @Test(dataProvider = %22testRandomDataDataProvider%22)%0A    public void testRandomData(int size, BlockFill fill)%0A    %7B%0A        roundTrip(getObjects(size, fill, getRandom()));%0A    %7D%0A%0A    private Object%5B%5D getObjects(int size, BlockFill fill, Random random)%0A    %7B
 %0A   
@@ -3253,33 +3253,32 @@
     %7D%0A        %7D%0A
-%0A
         roundTri
@@ -3270,32 +3270,28 @@
        r
-oundTrip(
+eturn 
 values
-)
 ;%0A    %7D%0A

Add test of 8-value blocks to block encoding test