/*
 *
 * 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 org.junit.Test;

import com.datastax.driver.core.Session;
import com.datastax.driver.core.SimpleStatement;
import com.datastax.driver.core.Statement;

import static org.junit.Assert.assertEquals;

public class IndexQueryPagingTest extends CQLTester
{
    /*
     * Some simple tests to verify the behaviour of paging during
     * 2i queries. We only use a single index type (CompositesIndexOnRegular)
     * as the code we want to exercise here is in their abstract
     * base class.
     */

    @Test
    public void pagingOnRegularColumn() throws Throwable
    {
        createTable("CREATE TABLE %s (" +
                    " k1 int," +
                    " v1 int," +
                    "PRIMARY KEY (k1))");
        createIndex("CREATE INDEX ON %s(v1)");

        int rowCount = 3;
        for (int i=0; i<rowCount; i++)
            execute("INSERT INTO %s (k1, v1) VALUES (?, ?)", i, 0);

        executePagingQuery("SELECT * FROM %s WHERE v1=0", rowCount);
    }

    @Test
    public void pagingOnRegularColumnWithPartitionRestriction() throws Throwable
    {
        createTable("CREATE TABLE %s (" +
                    " k1 int," +
                    " c1 int," +
                    " v1 int," +
                    "PRIMARY KEY (k1, c1))");
        createIndex("CREATE INDEX ON %s(v1)");

        int partitions = 3;
        int rowCount = 3;
        for (int i=0; i<partitions; i++)
            for (int j=0; j<rowCount; j++)
                execute("INSERT INTO %s (k1, c1, v1) VALUES (?, ?, ?)", i, j, 0);

        executePagingQuery("SELECT * FROM %s WHERE k1=0 AND v1=0", rowCount);
    }

    @Test
    public void pagingOnRegularColumnWithClusteringRestrictions() throws Throwable
    {
        createTable("CREATE TABLE %s (" +
                    " k1 int," +
                    " c1 int," +
                    " v1 int," +
                    "PRIMARY KEY (k1, c1))");
        createIndex("CREATE INDEX ON %s(v1)");

        int partitions = 3;
        int rowCount = 3;
        for (int i=0; i<partitions; i++)
            for (int j=0; j<rowCount; j++)
                execute("INSERT INTO %s (k1, c1, v1) VALUES (?, ?, ?)", i, j, 0);

        executePagingQuery("SELECT * FROM %s WHERE k1=0 AND c1>=0 AND c1<=3 AND v1=0", rowCount);
    }

    @Test
    public void testPagingOnPartitionsWithoutRows() throws Throwable
    {
        requireNetwork();

        createTable("CREATE TABLE %s (pk int, ck int, s int static, v int, PRIMARY KEY (pk, ck))");
        createIndex("CREATE INDEX on %s(s)");

        execute("INSERT INTO %s (pk, s) VALUES (201, 200);");
        execute("INSERT INTO %s (pk, s) VALUES (202, 200);");
        execute("INSERT INTO %s (pk, s) VALUES (203, 200);");
        execute("INSERT INTO %s (pk, s) VALUES (100, 100);");

        for (int pageSize = 1; pageSize < 10; pageSize++)
        {
            assertRowsNet(executeNetWithPaging("select * from %s where s = 200 and pk = 201;", pageSize),
                          row(201, null, 200, null));

            assertRowsNet(executeNetWithPaging("select * from %s where s = 200;", pageSize),
                          row(201, null, 200, null),
                          row(203, null, 200, null),
                          row(202, null, 200, null));

            assertRowsNet(executeNetWithPaging("select * from %s where s = 100;", pageSize),
                          row(100, null, 100, null));
        }
    }

    @Test
    public void testPagingOnPartitionsWithoutClusteringColumns() throws Throwable
    {
        createTable("CREATE TABLE %s (pk int PRIMARY KEY, v int)");
        createIndex("CREATE INDEX on %s(v)");

        execute("INSERT INTO %s (pk, v) VALUES (201, 200);");
        execute("INSERT INTO %s (pk, v) VALUES (202, 200);");
        execute("INSERT INTO %s (pk, v) VALUES (203, 200);");
        execute("INSERT INTO %s (pk, v) VALUES (100, 100);");

        for (int pageSize = 1; pageSize < 10; pageSize++)
        {
            assertRowsNet(executeNetWithPaging("select * from %s where v = 200 and pk = 201;", pageSize),
                          row(201, 200));

            assertRowsNet(executeNetWithPaging("select * from %s where v = 200;", pageSize),
                          row(201, 200),
                          row(203, 200),
                          row(202, 200));

            assertRowsNet(executeNetWithPaging("select * from %s where v = 100;", pageSize),
                          row(100, 100));
        }
    }

    private void executePagingQuery(String cql, int rowCount)
    {
        // Execute an index query which should return all rows,
        // setting the fetch size < than the row count. Assert
        // that all rows are returned, so we know that paging
        // of the results was involved.
        Session session = sessionNet();
        Statement stmt = new SimpleStatement(String.format(cql, KEYSPACE + '.' + currentTable()));
        stmt.setFetchSize(rowCount - 1);
        assertEquals(rowCount, session.execute(stmt).all().size());
    }
}