/*
 * 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.cassandra.cql3;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import com.google.common.base.Objects;

import org.junit.Test;

import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.utils.FBUtilities;

import static org.junit.Assert.fail;

/* ViewComplexTest class has been split into multiple ones because of timeout issues (CASSANDRA-16670, CASSANDRA-17167)
 * Any changes here check if they apply to the other classes:
 * - ViewComplexUpdatesTest
 * - ViewComplexDeletionsTest
 * - ViewComplexTTLTest
 * - ViewComplexTest
 * - ViewComplexLivenessTest
 * - ...
 * - ViewComplex*Test
 */
public class ViewComplexTest extends ViewComplexTester
{
    @Test
    public void testNonBaseColumnInViewPkWithFlush() throws Throwable
    {
        testNonBaseColumnInViewPk(true);
    }

    @Test
    public void testNonBaseColumnInViewPkWithoutFlush() throws Throwable
    {
        testNonBaseColumnInViewPk(false);
    }

    private void testNonBaseColumnInViewPk(boolean flush) throws Throwable
    {
        createTable("create table %s (p1 int, p2 int, v1 int, v2 int, primary key (p1,p2))");

        execute("USE " + keyspace());
        executeNet(version, "USE " + keyspace());
        Keyspace ks = Keyspace.open(keyspace());

        String mv = createView("create materialized view %s as select * from %%s " +
                               "where p1 is not null and p2 is not null primary key (p2, p1) " +
                               "with gc_grace_seconds=5");
        ColumnFamilyStore cfs = ks.getColumnFamilyStore(mv);
        cfs.disableAutoCompaction();

        updateView("UPDATE %s USING TIMESTAMP 1 set v1 =1 where p1 = 1 AND p2 = 1;");
        if (flush)
            FBUtilities.waitOnFutures(ks.flush());
        assertRowsIgnoringOrder(execute("SELECT p1, p2, v1, v2 from %s"), row(1, 1, 1, null));
        assertRowsIgnoringOrder(execute("SELECT p1, p2, v1, v2 from " + mv), row(1, 1, 1, null));

        updateView("UPDATE %s USING TIMESTAMP 2 set v1 = null, v2 = 1 where p1 = 1 AND p2 = 1;");
        if (flush)
            FBUtilities.waitOnFutures(ks.flush());
        assertRowsIgnoringOrder(execute("SELECT p1, p2, v1, v2 from %s"), row(1, 1, null, 1));
        assertRowsIgnoringOrder(execute("SELECT p1, p2, v1, v2 from " + mv), row(1, 1, null, 1));

        updateView("UPDATE %s USING TIMESTAMP 2 set v2 = null where p1 = 1 AND p2 = 1;");
        if (flush)
            FBUtilities.waitOnFutures(ks.flush());
        assertRowsIgnoringOrder(execute("SELECT p1, p2, v1, v2 from %s"));
        assertRowsIgnoringOrder(execute("SELECT p1, p2, v1, v2 from " + mv));

        updateView("INSERT INTO %s (p1,p2) VALUES(1,1) USING TIMESTAMP 3;");
        if (flush)
            FBUtilities.waitOnFutures(ks.flush());
        assertRowsIgnoringOrder(execute("SELECT p1, p2, v1, v2 from %s"), row(1, 1, null, null));
        assertRowsIgnoringOrder(execute("SELECT p1, p2, v1, v2 from " + mv), row(1, 1, null, null));

        updateView("DELETE FROM %s USING TIMESTAMP 4 WHERE p1 =1 AND p2 = 1;");
        if (flush)
            FBUtilities.waitOnFutures(ks.flush());
        assertRowsIgnoringOrder(execute("SELECT p1, p2, v1, v2 from %s"));
        assertRowsIgnoringOrder(execute("SELECT p1, p2, v1, v2 from " + mv));

        updateView("UPDATE %s USING TIMESTAMP 5 set v2 = 1 where p1 = 1 AND p2 = 1;");
        if (flush)
            FBUtilities.waitOnFutures(ks.flush());
        assertRowsIgnoringOrder(execute("SELECT p1, p2, v1, v2 from %s"), row(1, 1, null, 1));
        assertRowsIgnoringOrder(execute("SELECT p1, p2, v1, v2 from " + mv), row(1, 1, null, 1));
    }

    @Test
    public void testMVWithDifferentColumnsWithFlush() throws Throwable
    {
        testMVWithDifferentColumns(true);
    }

    @Test
    public void testMVWithDifferentColumnsWithoutFlush() throws Throwable
    {
        testMVWithDifferentColumns(false);
    }

    private void testMVWithDifferentColumns(boolean flush) throws Throwable
    {
        createTable("CREATE TABLE %s (a int, b int, c int, d int, e int, f int, PRIMARY KEY(a, b))");

        execute("USE " + keyspace());
        executeNet(version, "USE " + keyspace());
        List<String> viewNames = new ArrayList<>();
        List<String> mvStatements = Arrays.asList(
                                                  // all selected
                                                  "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE a IS NOT NULL AND b IS NOT NULL PRIMARY KEY (a,b)",
                                                  // unselected e,f
                                                  "CREATE MATERIALIZED VIEW %s AS SELECT a,b,c,d FROM %%s WHERE a IS NOT NULL AND b IS NOT NULL PRIMARY KEY (a,b)",
                                                  // no selected
                                                  "CREATE MATERIALIZED VIEW %s AS SELECT a,b FROM %%s WHERE a IS NOT NULL AND b IS NOT NULL PRIMARY KEY (a,b)",
                                                  // all selected, re-order keys
                                                  "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE a IS NOT NULL AND b IS NOT NULL PRIMARY KEY (b,a)",
                                                  // unselected e,f, re-order keys
                                                  "CREATE MATERIALIZED VIEW %s AS SELECT a,b,c,d FROM %%s WHERE a IS NOT NULL AND b IS NOT NULL PRIMARY KEY (b,a)",
                                                  // no selected, re-order keys
                                                  "CREATE MATERIALIZED VIEW %s AS SELECT a,b FROM %%s WHERE a IS NOT NULL AND b IS NOT NULL PRIMARY KEY (b,a)");

        Keyspace ks = Keyspace.open(keyspace());

        for (int i = 0; i < mvStatements.size(); i++)
        {
            String name = createView(mvStatements.get(i));
            viewNames.add(name);
            ks.getColumnFamilyStore(name).disableAutoCompaction();
        }

        // insert
        updateViewWithFlush("INSERT INTO %s (a,b,c,d,e,f) VALUES(1,1,1,1,1,1) using timestamp 1", flush);
        assertBaseViews(row(1, 1, 1, 1, 1, 1), viewNames);

        updateViewWithFlush("UPDATE %s using timestamp 2 SET c=0, d=0 WHERE a=1 AND b=1", flush);
        assertBaseViews(row(1, 1, 0, 0, 1, 1), viewNames);

        updateViewWithFlush("UPDATE %s using timestamp 2 SET e=0, f=0 WHERE a=1 AND b=1", flush);
        assertBaseViews(row(1, 1, 0, 0, 0, 0), viewNames);

        updateViewWithFlush("DELETE FROM %s using timestamp 2 WHERE a=1 AND b=1", flush);
        assertBaseViews(null, viewNames);

        // partial update unselected, selected
        updateViewWithFlush("UPDATE %s using timestamp 3 SET f=1 WHERE a=1 AND b=1", flush);
        assertBaseViews(row(1, 1, null, null, null, 1), viewNames);

        updateViewWithFlush("UPDATE %s using timestamp 4 SET e = 1, f=null WHERE a=1 AND b=1", flush);
        assertBaseViews(row(1, 1, null, null, 1, null), viewNames);

        updateViewWithFlush("UPDATE %s using timestamp 4 SET e = null WHERE a=1 AND b=1", flush);
        assertBaseViews(null, viewNames);

        updateViewWithFlush("UPDATE %s using timestamp 5 SET c = 1 WHERE a=1 AND b=1", flush);
        assertBaseViews(row(1, 1, 1, null, null, null), viewNames);

        updateViewWithFlush("UPDATE %s using timestamp 5 SET c = null WHERE a=1 AND b=1", flush);
        assertBaseViews(null, viewNames);

        updateViewWithFlush("UPDATE %s using timestamp 6 SET d = 1 WHERE a=1 AND b=1", flush);
        assertBaseViews(row(1, 1, null, 1, null, null), viewNames);

        updateViewWithFlush("UPDATE %s using timestamp 7 SET d = null WHERE a=1 AND b=1", flush);
        assertBaseViews(null, viewNames);

        updateViewWithFlush("UPDATE %s using timestamp 8 SET f = 1 WHERE a=1 AND b=1", flush);
        assertBaseViews(row(1, 1, null, null, null, 1), viewNames);

        updateViewWithFlush("UPDATE %s using timestamp 6 SET c = 1 WHERE a=1 AND b=1", flush);
        assertBaseViews(row(1, 1, 1, null, null, 1), viewNames);

        // view row still alive due to c=1@6
        updateViewWithFlush("UPDATE %s using timestamp 8 SET f = null WHERE a=1 AND b=1", flush);
        assertBaseViews(row(1, 1, 1, null, null, null), viewNames);

        updateViewWithFlush("UPDATE %s using timestamp 6 SET c = null WHERE a=1 AND b=1", flush);
        assertBaseViews(null, viewNames);
    }

    private void assertBaseViews(Object[] row, List<String> viewNames) throws Throwable
    {
        UntypedResultSet result = execute("SELECT * FROM %s");
        if (row == null)
            assertRowsIgnoringOrder(result);
        else
            assertRowsIgnoringOrder(result, row);
        for (int i = 0; i < viewNames.size(); i++)
            assertBaseView(result, execute(String.format("SELECT * FROM %s", viewNames.get(i))), viewNames.get(i));
    }

    private void assertBaseView(UntypedResultSet base, UntypedResultSet view, String mv)
    {
        List<ColumnSpecification> baseMeta = base.metadata();
        List<ColumnSpecification> viewMeta = view.metadata();

        Iterator<UntypedResultSet.Row> iter = base.iterator();
        Iterator<UntypedResultSet.Row> viewIter = view.iterator();

        List<UntypedResultSet.Row> baseData = com.google.common.collect.Lists.newArrayList(iter);
        List<UntypedResultSet.Row> viewData = com.google.common.collect.Lists.newArrayList(viewIter);

        if (baseData.size() != viewData.size())
            fail(String.format("Mismatch number of rows in view %s: <%s>, in base <%s>",
                               mv,
                               makeRowStrings(view),
                               makeRowStrings(base)));
        if (baseData.size() == 0)
            return;
        if (viewData.size() != 1)
            fail(String.format("Expect only one row in view %s, but got <%s>",
                               mv,
                               makeRowStrings(view)));

        UntypedResultSet.Row row = baseData.get(0);
        UntypedResultSet.Row viewRow = viewData.get(0);

        Map<String, ByteBuffer> baseValues = new HashMap<>();
        for (int j = 0; j < baseMeta.size(); j++)
        {
            ColumnSpecification column = baseMeta.get(j);
            ByteBuffer actualValue = row.getBytes(column.name.toString());
            baseValues.put(column.name.toString(), actualValue);
        }
        for (int j = 0; j < viewMeta.size(); j++)
        {
            ColumnSpecification column = viewMeta.get(j);
            String name = column.name.toString();
            ByteBuffer viewValue = viewRow.getBytes(name);
            if (!baseValues.containsKey(name))
            {
                fail(String.format("Extra column: %s with value %s in view", name, column.type.compose(viewValue)));
            }
            else if (!Objects.equal(baseValues.get(name), viewValue))
            {
                fail(String.format("Non equal column: %s, expected <%s> but got <%s>",
                                   name,
                                   column.type.compose(baseValues.get(name)),
                                   column.type.compose(viewValue)));
            }
        }
    }
}