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

import java.util.Date;
import java.util.Iterator;

import com.datastax.driver.core.Row;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.SimpleStatement;
import com.google.common.collect.Iterators;

import org.junit.Assert;
import org.junit.Test;

import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.ICoordinator;
import org.apache.cassandra.serializers.SimpleDateSerializer;
import org.apache.cassandra.serializers.TimeSerializer;
import org.apache.cassandra.serializers.TimestampSerializer;

import static org.apache.cassandra.distributed.api.ConsistencyLevel.QUORUM;
import static org.apache.cassandra.distributed.api.Feature.NATIVE_PROTOCOL;
import static org.apache.cassandra.distributed.api.Feature.NETWORK;
import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows;
import static org.apache.cassandra.distributed.shared.AssertUtils.row;

public class GroupByTest extends TestBaseImpl
{
    @Test
    public void groupByWithDeletesAndSrpOnPartitions() throws Throwable
    {
        try (Cluster cluster = init(builder().withNodes(2).withConfig((cfg) -> cfg.set("user_defined_functions_enabled", "true")).start()))
        {
            cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (pk int, ck text, PRIMARY KEY (pk, ck))"));
            initFunctions(cluster);
            cluster.get(1).executeInternal(withKeyspace("INSERT INTO %s.tbl (pk, ck) VALUES (1, '1') USING TIMESTAMP 0"));
            cluster.get(1).executeInternal(withKeyspace("INSERT INTO %s.tbl (pk, ck) VALUES (2, '2') USING TIMESTAMP 0"));
            cluster.get(1).executeInternal(withKeyspace("DELETE FROM %s.tbl WHERE pk=0 AND ck='0'"));

            cluster.get(2).executeInternal(withKeyspace("INSERT INTO %s.tbl (pk, ck) VALUES (0, '0') USING TIMESTAMP 0"));
            cluster.get(2).executeInternal(withKeyspace("DELETE FROM %s.tbl WHERE pk=1 AND ck='1'"));
            cluster.get(2).executeInternal(withKeyspace("DELETE FROM %s.tbl WHERE pk=2 AND ck='2'"));

            for (String limitClause : new String[]{ "", "LIMIT 1", "LIMIT 10", "PER PARTITION LIMIT 1", "PER PARTITION LIMIT 10" })
            {
                String query = withKeyspace("SELECT concat(ck) FROM %s.tbl GROUP BY pk " + limitClause);
                for (int i = 1; i <= 4; i++)
                {
                    Iterator<Object[]> rows = cluster.coordinator(2).executeWithPaging(query, ConsistencyLevel.ALL, i);
                    assertRows(Iterators.toArray(rows, Object[].class));
                }
            }
        }
    }

    @Test
    public void groupByWithDeletesAndSrpOnRows() throws Throwable
    {
        try (Cluster cluster = init(builder().withNodes(2).withConfig((cfg) -> cfg.set("user_defined_functions_enabled", "true")).start()))
        {
            cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (pk int, ck text, PRIMARY KEY (pk, ck))"));
            initFunctions(cluster);
            cluster.get(1).executeInternal(withKeyspace("INSERT INTO %s.tbl (pk, ck) VALUES (0, '1') USING TIMESTAMP 0"));
            cluster.get(1).executeInternal(withKeyspace("INSERT INTO %s.tbl (pk, ck) VALUES (0, '2') USING TIMESTAMP 0"));
            cluster.get(1).executeInternal(withKeyspace("DELETE FROM %s.tbl WHERE pk=0 AND ck='0'"));

            cluster.get(2).executeInternal(withKeyspace("INSERT INTO %s.tbl (pk, ck) VALUES (0, '0') USING TIMESTAMP 0"));
            cluster.get(2).executeInternal(withKeyspace("DELETE FROM %s.tbl WHERE pk=0 AND ck='1'"));
            cluster.get(2).executeInternal(withKeyspace("DELETE FROM %s.tbl WHERE pk=0 AND ck='2'"));

            for (String limitClause : new String[]{ "", "LIMIT 1", "LIMIT 10", "PER PARTITION LIMIT 1", "PER PARTITION LIMIT 10" })
            {
                String query = withKeyspace("SELECT concat(ck) FROM %s.tbl GROUP BY pk " + limitClause);
                for (int i = 1; i <= 4; i++)
                {
                    Iterator<Object[]> rows = cluster.coordinator(2).executeWithPaging(query, ConsistencyLevel.ALL, i);
                    assertRows(Iterators.toArray(rows, Object[].class));
                }
            }
        }
    }

    @Test
    public void testGroupByWithAggregatesAndPaging() throws Throwable
    {
        try (Cluster cluster = init(builder().withNodes(2).withConfig((cfg) -> cfg.set("user_defined_functions_enabled", "true")).start()))
        {
            cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (pk int, ck int, v1 text, v2 text, v3 text, primary key (pk, ck))"));
            initFunctions(cluster);

            cluster.coordinator(1).execute(withKeyspace("insert into %s.tbl (pk, ck, v1, v2, v3) values (1,1,'1','1','1')"), ConsistencyLevel.ALL);
            cluster.coordinator(1).execute(withKeyspace("insert into %s.tbl (pk, ck, v1, v2, v3) values (1,2,'2','2','2')"), ConsistencyLevel.ALL);
            cluster.coordinator(1).execute(withKeyspace("insert into %s.tbl (pk, ck, v1, v2, v3) values (1,3,'3','3','3')"), ConsistencyLevel.ALL);

            for (int i = 1; i <= 4; i++)
            {
                assertRows(cluster.coordinator(1).executeWithPaging(withKeyspace("select concat(v1), concat(v2), concat(v3) from %s.tbl where pk = 1 group by pk"),
                                                                    ConsistencyLevel.ALL, i),
                           row("_ 1 2 3", "_ 1 2 3", "_ 1 2 3"));

                assertRows(cluster.coordinator(1).executeWithPaging(withKeyspace("select concat(v1), concat(v2), concat(v3) from %s.tbl where pk = 1 group by pk limit 1"),
                                                                    ConsistencyLevel.ALL, i),
                           row("_ 1 2 3", "_ 1 2 3", "_ 1 2 3"));

                assertRows(cluster.coordinator(1).executeWithPaging(withKeyspace("select * from %s.tbl where pk = 1 group by pk"),
                                                                    ConsistencyLevel.ALL, i),
                           row(1, 1, "1", "1", "1"));
            }
        }
    }

    @Test
    public void testGroupWithDeletesAndPaging() throws Throwable
    {
        try (Cluster cluster = init(builder().withNodes(2).withConfig(cfg -> cfg.with(Feature.GOSSIP, NETWORK, NATIVE_PROTOCOL)).start()))
        {
            cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (pk int, ck int, PRIMARY KEY (pk, ck))"));
            ICoordinator coordinator = cluster.coordinator(1);
            coordinator.execute(withKeyspace("INSERT INTO %s.tbl (pk, ck) VALUES (0, 0)"), ConsistencyLevel.ALL);
            coordinator.execute(withKeyspace("INSERT INTO %s.tbl (pk, ck) VALUES (1, 1)"), ConsistencyLevel.ALL);

            cluster.get(1).executeInternal(withKeyspace("DELETE FROM %s.tbl WHERE pk=0 AND ck=0"));
            cluster.get(2).executeInternal(withKeyspace("DELETE FROM %s.tbl WHERE pk=1 AND ck=1"));
            String query = withKeyspace("SELECT * FROM %s.tbl GROUP BY pk");
            Iterator<Object[]> rows = coordinator.executeWithPaging(query, ConsistencyLevel.ALL, 1);
            assertRows(Iterators.toArray(rows, Object[].class));

            try (com.datastax.driver.core.Cluster c = com.datastax.driver.core.Cluster.builder().addContactPoint("127.0.0.1").build();
                 Session session = c.connect())
            {
                SimpleStatement stmt = new SimpleStatement(withKeyspace("select * from %s.tbl where pk = 1 group by pk"));
                stmt.setFetchSize(1);
                Iterator<Row> rs = session.execute(stmt).iterator();
                Assert.assertFalse(rs.hasNext());
            }
        }
    }

    @Test
    public void testGroupByTimeRangesWithTimestampType() throws Throwable
    {
        try (Cluster cluster = init(builder().withNodes(3).start()))
        {
            cluster.schemaChange(withKeyspace("CREATE TABLE %s.testWithTimestamp (pk int, time timestamp, v int, primary key (pk, time))"));
            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithTimestamp (pk, time, v) VALUES (1, '2016-09-27 16:10:00 UTC', 1)"), ConsistencyLevel.QUORUM);
            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithTimestamp (pk, time, v) VALUES (1, '2016-09-27 16:12:00 UTC', 2)"), ConsistencyLevel.QUORUM);
            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithTimestamp (pk, time, v) VALUES (1, '2016-09-27 16:14:00 UTC', 3)"), ConsistencyLevel.QUORUM);
            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithTimestamp (pk, time, v) VALUES (1, '2016-09-27 16:15:00 UTC', 4)"), ConsistencyLevel.QUORUM);
            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithTimestamp (pk, time, v) VALUES (1, '2016-09-27 16:21:00 UTC', 5)"), ConsistencyLevel.QUORUM);
            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithTimestamp (pk, time, v) VALUES (1, '2016-09-27 16:22:00 UTC', 6)"), ConsistencyLevel.QUORUM);
            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithTimestamp (pk, time, v) VALUES (1, '2016-09-27 16:26:00 UTC', 7)"), ConsistencyLevel.QUORUM);
            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithTimestamp (pk, time, v) VALUES (1, '2016-09-27 16:26:20 UTC', 8)"), ConsistencyLevel.QUORUM);
            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithTimestamp (pk, time, v) VALUES (2, '2016-09-27 16:26:20 UTC', 10)"), ConsistencyLevel.QUORUM);
            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithTimestamp (pk, time, v) VALUES (2, '2016-09-27 16:30:00 UTC', 11)"), ConsistencyLevel.QUORUM);

            for (int pageSize : new int[] {2, 3, 4, 5, 7, 10})
            {
                for (String startingTime : new String[] {"", ", '2016-09-27 UTC'"} )
                {
                    String stmt = "SELECT pk, floor(time, 5m" + startingTime + "), min(v), max(v), count(v) FROM %s.testWithTimestamp GROUP BY pk, floor(time, 5m" + startingTime + ")";
                    Iterator<Object[]> pagingRows = cluster.coordinator(1).executeWithPaging(withKeyspace(stmt), QUORUM, pageSize);
                    assertRows(pagingRows, 
                               row(1, toTimestamp("2016-09-27 16:10:00 UTC"), 1, 3, 3L),
                               row(1, toTimestamp("2016-09-27 16:15:00 UTC"), 4, 4, 1L),
                               row(1, toTimestamp("2016-09-27 16:20:00 UTC"), 5, 6, 2L),
                               row(1, toTimestamp("2016-09-27 16:25:00 UTC"), 7, 8, 2L),
                               row(2, toTimestamp("2016-09-27 16:25:00 UTC"), 10, 10, 1L),
                               row(2, toTimestamp("2016-09-27 16:30:00 UTC"), 11, 11, 1L));

                    stmt = "SELECT pk, floor(time, 5m" + startingTime + "), min(v), max(v), count(v) FROM %s.testWithTimestamp GROUP BY pk, floor(time, 5m" + startingTime + ") LIMIT 2";
                    pagingRows = cluster.coordinator(1).executeWithPaging(withKeyspace(stmt), QUORUM, pageSize);
                    assertRows(pagingRows, 
                               row(1, toTimestamp("2016-09-27 16:10:00 UTC"), 1, 3, 3L),
                               row(1, toTimestamp("2016-09-27 16:15:00 UTC"), 4, 4, 1L));

                    stmt = "SELECT pk, floor(time, 5m" + startingTime + "), min(v), max(v), count(v) FROM %s.testWithTimestamp GROUP BY pk, floor(time, 5m" + startingTime + ") PER PARTITION LIMIT 1";
                    pagingRows = cluster.coordinator(1).executeWithPaging(withKeyspace(stmt), QUORUM, pageSize);
                    assertRows(pagingRows, 
                               row(1, toTimestamp("2016-09-27 16:10:00 UTC"), 1, 3, 3L),
                               row(2, toTimestamp("2016-09-27 16:25:00 UTC"), 10, 10, 1L));

                    stmt = "SELECT pk, floor(time, 5m" + startingTime + "), min(v), max(v), count(v) FROM %s.testWithTimestamp WHERE pk = 1 GROUP BY pk, floor(time, 5m" + startingTime + ") ORDER BY time DESC";
                    pagingRows = cluster.coordinator(1).executeWithPaging(withKeyspace(stmt), QUORUM, pageSize);
                    assertRows(pagingRows, 
                               row(1, toTimestamp("2016-09-27 16:25:00 UTC"), 7, 8, 2L),
                               row(1, toTimestamp("2016-09-27 16:20:00 UTC"), 5, 6, 2L),
                               row(1, toTimestamp("2016-09-27 16:15:00 UTC"), 4, 4, 1L),
                               row(1, toTimestamp("2016-09-27 16:10:00 UTC"), 1, 3, 3L));

                    stmt = "SELECT pk, floor(time, 5m" + startingTime + "), min(v), max(v), count(v) FROM %s.testWithTimestamp WHERE pk = 1 GROUP BY pk, floor(time, 5m" + startingTime + ") ORDER BY time DESC LIMIT 2";
                    pagingRows = cluster.coordinator(1).executeWithPaging(withKeyspace(stmt), QUORUM, pageSize);
                    assertRows(pagingRows, 
                               row(1, toTimestamp("2016-09-27 16:25:00 UTC"), 7, 8, 2L),
                               row(1, toTimestamp("2016-09-27 16:20:00 UTC"), 5, 6, 2L));
                }
            }
        }
    }

    @Test
    public void testGroupByTimeRangesWithDateType() throws Throwable
    {
        try (Cluster cluster = init(builder().withNodes(3).start()))
        {

            cluster.schemaChange(withKeyspace("CREATE TABLE %s.testWithDate (pk int, time date, v int, primary key (pk, time))"));

            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithDate (pk, time, v) VALUES (1, '2016-09-27', 1)"), ConsistencyLevel.QUORUM);
            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithDate (pk, time, v) VALUES (1, '2016-09-28', 2)"), ConsistencyLevel.QUORUM);
            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithDate (pk, time, v) VALUES (1, '2016-09-29', 3)"), ConsistencyLevel.QUORUM);
            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithDate (pk, time, v) VALUES (1, '2016-09-30', 4)"), ConsistencyLevel.QUORUM);
            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithDate (pk, time, v) VALUES (1, '2016-10-01', 5)"), ConsistencyLevel.QUORUM);
            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithDate (pk, time, v) VALUES (1, '2016-10-04', 6)"), ConsistencyLevel.QUORUM);
            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithDate (pk, time, v) VALUES (1, '2016-10-20', 7)"), ConsistencyLevel.QUORUM);
            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithDate (pk, time, v) VALUES (1, '2016-11-27', 8)"), ConsistencyLevel.QUORUM);
            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithDate (pk, time, v) VALUES (2, '2016-11-01', 10)"), ConsistencyLevel.QUORUM);
            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithDate (pk, time, v) VALUES (2, '2016-11-02', 11)"), ConsistencyLevel.QUORUM);

            for (int pageSize : new int[] {2, 3, 4, 5, 7, 10})
            {
                for (String startingTime : new String[] {"", ", '2016-06-01'"} )
                {

                    String stmt = "SELECT pk, floor(time, 1mo" + startingTime + "), min(v), max(v), count(v) FROM %s.testWithDate GROUP BY pk, floor(time, 1mo" + startingTime + ")";
                    Iterator<Object[]> pagingRows = cluster.coordinator(1).executeWithPaging(withKeyspace(stmt), QUORUM, pageSize);
                    assertRows(pagingRows, 
                               row(1, toLocalDate("2016-09-01"), 1, 4, 4L),
                               row(1, toLocalDate("2016-10-01"), 5, 7, 3L),
                               row(1, toLocalDate("2016-11-01"), 8, 8, 1L),
                               row(2, toLocalDate("2016-11-01"), 10, 11, 2L));

                    stmt = "SELECT pk, floor(time, 1mo" + startingTime + "), min(v), max(v), count(v) FROM %s.testWithDate GROUP BY pk, floor(time, 1mo" + startingTime + ") LIMIT 2";
                    pagingRows = cluster.coordinator(1).executeWithPaging(withKeyspace(stmt), QUORUM, pageSize);
                    assertRows(pagingRows, 
                               row(1, toLocalDate("2016-09-01"), 1, 4, 4L),
                               row(1, toLocalDate("2016-10-01"), 5, 7, 3L));

                    stmt = "SELECT pk, floor(time, 1mo" + startingTime + "), min(v), max(v), count(v) FROM %s.testWithDate GROUP BY pk, floor(time, 1mo" + startingTime + ") PER PARTITION LIMIT 1";
                    pagingRows = cluster.coordinator(1).executeWithPaging(withKeyspace(stmt), QUORUM, pageSize);
                    assertRows(pagingRows, 
                               row(1, toLocalDate("2016-09-01"), 1, 4, 4L),
                               row(2, toLocalDate("2016-11-01"), 10, 11, 2L));

                    stmt = "SELECT pk, floor(time, 1mo" + startingTime + "), min(v), max(v), count(v) FROM %s.testWithDate WHERE pk = 1 GROUP BY pk, floor(time, 1mo" + startingTime + ") ORDER BY time DESC";
                    pagingRows = cluster.coordinator(1).executeWithPaging(withKeyspace(stmt), QUORUM, pageSize);
                    assertRows(pagingRows, 
                               row(1, toLocalDate("2016-11-01"), 8, 8, 1L),
                               row(1, toLocalDate("2016-10-01"), 5, 7, 3L),
                               row(1, toLocalDate("2016-09-01"), 1, 4, 4L));

                    stmt = "SELECT pk, floor(time, 1mo" + startingTime + "), min(v), max(v), count(v) FROM %s.testWithDate WHERE pk = 1 GROUP BY pk, floor(time, 1mo" + startingTime + ") ORDER BY time DESC LIMIT 2";
                    pagingRows = cluster.coordinator(1).executeWithPaging(withKeyspace(stmt), QUORUM, pageSize);
                    assertRows(pagingRows, 
                               row(1, toLocalDate("2016-11-01"), 8, 8, 1L),
                               row(1, toLocalDate("2016-10-01"), 5, 7, 3L));
                }
            }
        }
    }

        @Test
        public void testGroupByTimeRangesWithTimeType() throws Throwable
        {
            try (Cluster cluster = init(builder().withNodes(3).start()))
            {

                cluster.schemaChange(withKeyspace("CREATE TABLE %s.testWithTime (pk int, date date, time time, v int, primary key (pk, date, time))"));

                cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithTime (pk, date, time, v) VALUES (1, '2016-09-27', '16:10:00', 1)"), ConsistencyLevel.QUORUM);
                cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithTime (pk, date, time, v) VALUES (1, '2016-09-27', '16:12:00', 2)"), ConsistencyLevel.QUORUM);
                cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithTime (pk, date, time, v) VALUES (1, '2016-09-27', '16:14:00', 3)"), ConsistencyLevel.QUORUM);
                cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithTime (pk, date, time, v) VALUES (1, '2016-09-27', '16:15:00', 4)"), ConsistencyLevel.QUORUM);
                cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithTime (pk, date, time, v) VALUES (1, '2016-09-27', '16:21:00', 5)"), ConsistencyLevel.QUORUM);
                cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithTime (pk, date, time, v) VALUES (1, '2016-09-27', '16:22:00', 6)"), ConsistencyLevel.QUORUM);
                cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithTime (pk, date, time, v) VALUES (1, '2016-09-27', '16:26:00', 7)"), ConsistencyLevel.QUORUM);
                cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithTime (pk, date, time, v) VALUES (1, '2016-09-27', '16:26:20', 8)"), ConsistencyLevel.QUORUM);
                cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithTime (pk, date, time, v) VALUES (1, '2016-09-28', '16:26:20', 9)"), ConsistencyLevel.QUORUM);
                cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.testWithTime (pk, date, time, v) VALUES (1, '2016-09-28', '16:26:30', 10)"), ConsistencyLevel.QUORUM);

                for (int pageSize : new int[] {2, 3, 4, 5, 7, 10})
                {

                    String stmt = "SELECT pk, date, floor(time, 5m), min(v), max(v), count(v) FROM %s.testWithTime GROUP BY pk, date, floor(time, 5m)";
                    Iterator<Object[]> pagingRows = cluster.coordinator(1).executeWithPaging(withKeyspace(stmt), QUORUM, pageSize);
                    assertRows(pagingRows, 
                               row(1, toLocalDate("2016-09-27"), toTime("16:10:00"), 1, 3, 3L),
                               row(1, toLocalDate("2016-09-27"), toTime("16:15:00"), 4, 4, 1L),
                               row(1, toLocalDate("2016-09-27"), toTime("16:20:00"), 5, 6, 2L),
                               row(1, toLocalDate("2016-09-27"), toTime("16:25:00"), 7, 8, 2L),
                               row(1, toLocalDate("2016-09-28"), toTime("16:25:00"), 9, 10, 2L));

                    stmt = "SELECT pk, date, floor(time, 5m), min(v), max(v), count(v) FROM %s.testWithTime GROUP BY pk, date, floor(time, 5m) LIMIT 2";
                    pagingRows = cluster.coordinator(1).executeWithPaging(withKeyspace(stmt), QUORUM, pageSize);
                    assertRows(pagingRows, 
                               row(1, toLocalDate("2016-09-27"), toTime("16:10:00"), 1, 3, 3L),
                               row(1, toLocalDate("2016-09-27"), toTime("16:15:00"), 4, 4, 1L));

                    stmt = "SELECT pk, date, floor(time, 5m), min(v), max(v), count(v) FROM %s.testWithTime WHERE pk = 1 GROUP BY pk, date, floor(time, 5m) ORDER BY date DESC, time DESC";
                    pagingRows = cluster.coordinator(1).executeWithPaging(withKeyspace(stmt), QUORUM, pageSize);
                    assertRows(pagingRows, 
                               row(1, toLocalDate("2016-09-28"), toTime("16:25:00"), 9, 10, 2L),
                               row(1, toLocalDate("2016-09-27"), toTime("16:25:00"), 7, 8, 2L),
                               row(1, toLocalDate("2016-09-27"), toTime("16:20:00"), 5, 6, 2L),
                               row(1, toLocalDate("2016-09-27"), toTime("16:15:00"), 4, 4, 1L),
                               row(1, toLocalDate("2016-09-27"), toTime("16:10:00"), 1, 3, 3L));

                    stmt = "SELECT pk, date, floor(time, 5m), min(v), max(v), count(v) FROM %s.testWithTime WHERE pk = 1 GROUP BY pk, date, floor(time, 5m) ORDER BY date DESC, time DESC LIMIT 2";
                    pagingRows = cluster.coordinator(1).executeWithPaging(withKeyspace(stmt), QUORUM, pageSize);
                    assertRows(pagingRows, 
                               row(1, toLocalDate("2016-09-28"), toTime("16:25:00"), 9, 10, 2L),
                               row(1, toLocalDate("2016-09-27"), toTime("16:25:00"), 7, 8, 2L));
                }
            }
        }

    private static void initFunctions(Cluster cluster)
    {
        cluster.schemaChange(withKeyspace("CREATE FUNCTION %s.concat_strings_fn(a text, b text) " +
                                          "RETURNS NULL ON NULL INPUT " +
                                          "RETURNS text " +
                                          "LANGUAGE java " +
                                          "AS 'return a + \" \" + b;'"));

        cluster.schemaChange(withKeyspace("CREATE AGGREGATE %s.concat(text)" +
                                          " SFUNC concat_strings_fn" +
                                          " STYPE text" +
                                          " INITCOND '_'"));
    }

    private static Date toTimestamp(String timestampAsString)
    {
        return new Date(TimestampSerializer.dateStringToTimestamp(timestampAsString));
    }

    private static int toLocalDate(String dateAsString)
    {
        return SimpleDateSerializer.dateStringToDays(dateAsString) ;
    }

    private static long toTime(String timeAsString)
    {
        return TimeSerializer.timeStringToLong(timeAsString) ;
    }

}