/*
 * 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.flink.test.operators;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.MapPartitionOperator;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.utils.DataSetUtils;
import org.apache.flink.test.operators.util.CollectionDataSets;
import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.apache.flink.util.Collector;

import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;

import java.util.List;
import java.util.Random;

import static org.junit.Assert.assertEquals;

/** Integration tests for {@link DataSetUtils#sample}. */
@SuppressWarnings("serial")
@RunWith(Parameterized.class)
public class SampleITCase extends MultipleProgramsTestBase {

    private static final Random RNG = new Random();

    public SampleITCase(TestExecutionMode mode) {
        super(mode);
    }

    @Before
    public void initiate() {
        ExecutionEnvironment.getExecutionEnvironment().setParallelism(5);
    }

    @Test
    public void testSamplerWithFractionWithoutReplacement() throws Exception {
        verifySamplerWithFractionWithoutReplacement(0d);
        verifySamplerWithFractionWithoutReplacement(0.2d);
        verifySamplerWithFractionWithoutReplacement(1.0d);
    }

    @Test
    public void testSamplerWithFractionWithReplacement() throws Exception {
        verifySamplerWithFractionWithReplacement(0d);
        verifySamplerWithFractionWithReplacement(0.2d);
        verifySamplerWithFractionWithReplacement(1.0d);
        verifySamplerWithFractionWithReplacement(2.0d);
    }

    @Test
    public void testSamplerWithSizeWithoutReplacement() throws Exception {
        verifySamplerWithFixedSizeWithoutReplacement(0);
        verifySamplerWithFixedSizeWithoutReplacement(2);
        verifySamplerWithFixedSizeWithoutReplacement(21);
    }

    @Test
    public void testSamplerWithSizeWithReplacement() throws Exception {
        verifySamplerWithFixedSizeWithReplacement(0);
        verifySamplerWithFixedSizeWithReplacement(2);
        verifySamplerWithFixedSizeWithReplacement(21);
    }

    private void verifySamplerWithFractionWithoutReplacement(double fraction) throws Exception {
        verifySamplerWithFractionWithoutReplacement(fraction, RNG.nextLong());
    }

    private void verifySamplerWithFractionWithoutReplacement(double fraction, long seed)
            throws Exception {
        verifySamplerWithFraction(false, fraction, seed);
    }

    private void verifySamplerWithFractionWithReplacement(double fraction) throws Exception {
        verifySamplerWithFractionWithReplacement(fraction, RNG.nextLong());
    }

    private void verifySamplerWithFractionWithReplacement(double fraction, long seed)
            throws Exception {
        verifySamplerWithFraction(true, fraction, seed);
    }

    private void verifySamplerWithFraction(boolean withReplacement, double fraction, long seed)
            throws Exception {
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        FlatMapOperator<Tuple3<Integer, Long, String>, String> ds = getSourceDataSet(env);
        MapPartitionOperator<String, String> sampled =
                DataSetUtils.sample(ds, withReplacement, fraction, seed);
        List<String> result = sampled.collect();
        containsResultAsText(result, getSourceStrings());
    }

    private void verifySamplerWithFixedSizeWithoutReplacement(int numSamples) throws Exception {
        verifySamplerWithFixedSizeWithoutReplacement(numSamples, RNG.nextLong());
    }

    private void verifySamplerWithFixedSizeWithoutReplacement(int numSamples, long seed)
            throws Exception {
        verifySamplerWithFixedSize(false, numSamples, seed);
    }

    private void verifySamplerWithFixedSizeWithReplacement(int numSamples) throws Exception {
        verifySamplerWithFixedSizeWithReplacement(numSamples, RNG.nextLong());
    }

    private void verifySamplerWithFixedSizeWithReplacement(int numSamples, long seed)
            throws Exception {
        verifySamplerWithFixedSize(true, numSamples, seed);
    }

    private void verifySamplerWithFixedSize(boolean withReplacement, int numSamples, long seed)
            throws Exception {
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        FlatMapOperator<Tuple3<Integer, Long, String>, String> ds = getSourceDataSet(env);
        DataSet<String> sampled =
                DataSetUtils.sampleWithSize(ds, withReplacement, numSamples, seed);
        List<String> result = sampled.collect();
        assertEquals(numSamples, result.size());
        containsResultAsText(result, getSourceStrings());
    }

    private FlatMapOperator<Tuple3<Integer, Long, String>, String> getSourceDataSet(
            ExecutionEnvironment env) {
        return CollectionDataSets.get3TupleDataSet(env)
                .flatMap(
                        new FlatMapFunction<Tuple3<Integer, Long, String>, String>() {
                            @Override
                            public void flatMap(
                                    Tuple3<Integer, Long, String> value, Collector<String> out)
                                    throws Exception {
                                out.collect(value.f2);
                            }
                        });
    }

    private String getSourceStrings() {
        return "Hi\n"
                + "Hello\n"
                + "Hello world\n"
                + "Hello world, how are you?\n"
                + "I am fine.\n"
                + "Luke Skywalker\n"
                + "Comment#1\n"
                + "Comment#2\n"
                + "Comment#3\n"
                + "Comment#4\n"
                + "Comment#5\n"
                + "Comment#6\n"
                + "Comment#7\n"
                + "Comment#8\n"
                + "Comment#9\n"
                + "Comment#10\n"
                + "Comment#11\n"
                + "Comment#12\n"
                + "Comment#13\n"
                + "Comment#14\n"
                + "Comment#15\n";
    }
}