cassandra-trunk/tools/fqltool/test/unit/org/apache/cassandra/fqltool/FQLReplayTest.java
/*
 * 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.fqltool;

import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
import java.util.Random;
import java.util.stream.Collectors;

import com.google.common.collect.Lists;
import org.junit.Test;

import com.datastax.driver.core.CodecRegistry;
import com.datastax.driver.core.SimpleStatement;
import com.datastax.driver.core.Statement;
import net.openhft.chronicle.core.io.IORuntimeException;
import net.openhft.chronicle.queue.ChronicleQueue;
import net.openhft.chronicle.queue.impl.single.SingleChronicleQueueBuilder;
import net.openhft.chronicle.queue.ExcerptAppender;
import net.openhft.chronicle.queue.ExcerptTailer;
import net.openhft.chronicle.wire.WireOut;
import org.apache.cassandra.fql.FullQueryLogger;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.cql3.statements.BatchStatement;
import org.apache.cassandra.fqltool.commands.Compare;
import org.apache.cassandra.fqltool.commands.Replay;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.tools.Util;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.MergeIterator;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.binlog.BinLog;

import static org.apache.cassandra.fqltool.QueryReplayer.ParsedTargetHost.fromString;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;

public class FQLReplayTest
{
    public FQLReplayTest()
    {
        Util.initDatabaseDescriptor();
    }

    @Test
    public void testOrderedReplay() throws IOException
    {
        File f = generateQueries(100, true);
        int queryCount = 0;
        try (ChronicleQueue queue = SingleChronicleQueueBuilder.single(f).build();
             FQLQueryIterator iter = new FQLQueryIterator(queue.createTailer(), 101))
        {
            long last = -1;
            while (iter.hasNext())
            {
                FQLQuery q = iter.next();
                assertTrue(q.queryStartTime >= last);
                last = q.queryStartTime;
                queryCount++;
            }
        }
        assertEquals(100, queryCount);
    }

    @Test
    public void testQueryIterator() throws IOException
    {
        File f = generateQueries(100, false);
        int queryCount = 0;
        try (ChronicleQueue queue = SingleChronicleQueueBuilder.single(f).build();
             FQLQueryIterator iter = new FQLQueryIterator(queue.createTailer(), 1))
        {
            long last = -1;
            while (iter.hasNext())
            {
                FQLQuery q = iter.next();
                assertTrue(q.queryStartTime >= last);
                last = q.queryStartTime;
                queryCount++;
            }
        }
        assertEquals(100, queryCount);
    }

    @Test
    public void testMergingIterator() throws IOException
    {
        File f = generateQueries(100, false);
        File f2 = generateQueries(100, false);
        int queryCount = 0;
        try (ChronicleQueue queue = SingleChronicleQueueBuilder.single(f).build();
             ChronicleQueue queue2 = SingleChronicleQueueBuilder.single(f2).build();
             FQLQueryIterator iter = new FQLQueryIterator(queue.createTailer(), 101);
             FQLQueryIterator iter2 = new FQLQueryIterator(queue2.createTailer(), 101);
             MergeIterator<FQLQuery, List<FQLQuery>> merger = MergeIterator.get(Lists.newArrayList(iter, iter2), FQLQuery::compareTo, new Replay.Reducer()))
        {
            long last = -1;

            while (merger.hasNext())
            {
                List<FQLQuery> qs = merger.next();
                assertEquals(2, qs.size());
                assertEquals(0, qs.get(0).compareTo(qs.get(1)));
                assertTrue(qs.get(0).queryStartTime >= last);
                last = qs.get(0).queryStartTime;
                queryCount++;
            }
        }
        assertEquals(100, queryCount);
    }

    @Test
    public void testFQLQueryReader() throws IOException
    {
        FQLQueryReader reader = new FQLQueryReader();

        try (ChronicleQueue queue = SingleChronicleQueueBuilder.single(generateQueries(1000, true)).build())
        {
            ExcerptTailer tailer = queue.createTailer();
            int queryCount = 0;
            while (tailer.readDocument(reader))
            {
                assertNotNull(reader.getQuery());
                if (reader.getQuery() instanceof FQLQuery.Single)
                {
                    assertTrue(reader.getQuery().keyspace() == null || reader.getQuery().keyspace().equals("querykeyspace"));
                }
                else
                {
                    assertEquals("someks", reader.getQuery().keyspace());
                }
                queryCount++;
            }
            assertEquals(1000, queryCount);
        }
    }

    @Test
    public void testStoringResults() throws Throwable
    {
        File tmpDir = Files.createTempDirectory("results").toFile();
        File queryDir = Files.createTempDirectory("queries").toFile();

        ResultHandler.ComparableResultSet res = createResultSet(10, 10, true);
        ResultStore rs = new ResultStore(Collections.singletonList(tmpDir), queryDir);
        FQLQuery query = new FQLQuery.Single("abc", QueryOptions.DEFAULT.getProtocolVersion().asInt(), QueryOptions.DEFAULT, 12345, 11111, 22, "select * from abc", Collections.emptyList());
        try
        {
            rs.storeColumnDefinitions(query, Collections.singletonList(res.getColumnDefinitions()));
            Iterator<ResultHandler.ComparableRow> it = res.iterator();
            while (it.hasNext())
            {
                List<ResultHandler.ComparableRow> row = Collections.singletonList(it.next());
                rs.storeRows(row);
            }
            // this marks the end of the result set:
            rs.storeRows(Collections.singletonList(null));
        }
        finally
        {
            rs.close();
        }

        compareResults(Collections.singletonList(Pair.create(query, res)),
                       readResultFile(tmpDir, queryDir));

    }

    private static void compareResults(List<Pair<FQLQuery, ResultHandler.ComparableResultSet>> expected,
                                List<Pair<FQLQuery, ResultHandler.ComparableResultSet>> other)
    {
        ResultComparator comparator = new ResultComparator();
        assertEquals(expected.size(), other.size());
        for (int i = 0; i < expected.size(); i++)
        {
            assertEquals(expected.get(i).left, other.get(i).left);
            ResultHandler.ComparableResultSet expectedResultSet = expected.get(i).right;
            ResultHandler.ComparableResultSet otherResultSet = other.get(i).right;
            List<String> hosts = Lists.newArrayList("a", "b");
            comparator.compareColumnDefinitions(hosts,
                                                expected.get(i).left,
                                                Lists.newArrayList(expectedResultSet.getColumnDefinitions(),
                                                                   otherResultSet.getColumnDefinitions()));
            Iterator<ResultHandler.ComparableRow> expectedRowIter = expectedResultSet.iterator();
            Iterator<ResultHandler.ComparableRow> otherRowIter = otherResultSet.iterator();


            while(expectedRowIter.hasNext() && otherRowIter.hasNext())
            {
                ResultHandler.ComparableRow expectedRow = expectedRowIter.next();
                ResultHandler.ComparableRow otherRow = otherRowIter.next();
                assertTrue(comparator.compareRows(hosts, expected.get(i).left, Lists.newArrayList(expectedRow, otherRow)));
            }
            assertFalse(expectedRowIter.hasNext());
            assertFalse(otherRowIter.hasNext());
        }
    }

    @Test
    public void testCompareColumnDefinitions()
    {
        ResultHandler.ComparableResultSet res = createResultSet(10, 10, false);
        ResultComparator rc = new ResultComparator();

        List<ResultHandler.ComparableColumnDefinitions> colDefs = new ArrayList<>(100);
        List<String> targetHosts = new ArrayList<>(100);
        for (int i = 0; i < 100; i++)
        {
            targetHosts.add("host"+i);
            colDefs.add(res.getColumnDefinitions());
        }
        assertTrue(rc.compareColumnDefinitions(targetHosts, null, colDefs));
        colDefs.set(50, createResultSet(9, 9, false).getColumnDefinitions());
        assertFalse(rc.compareColumnDefinitions(targetHosts, null, colDefs));
    }

    @Test
    public void testCompareEqualRows()
    {
        ResultComparator rc = new ResultComparator();

        ResultHandler.ComparableResultSet res = createResultSet(10, 10, false);
        ResultHandler.ComparableResultSet res2 = createResultSet(10, 10, false);
        List<ResultHandler.ComparableResultSet> toCompare = Lists.newArrayList(res, res2);
        List<Iterator<ResultHandler.ComparableRow>> iters = toCompare.stream().map(Iterable::iterator).collect(Collectors.toList());

        while (true)
        {
            List<ResultHandler.ComparableRow> rows = ResultHandler.rows(iters);
            assertTrue(rc.compareRows(Lists.newArrayList("eq1", "eq2"), null, rows));
            if (rows.stream().allMatch(Objects::isNull))
                break;
        }
    }

    @Test
    public void testCompareRowsDifferentCount()
    {
        ResultComparator rc = new ResultComparator();
        ResultHandler.ComparableResultSet res = createResultSet(10, 10, false);
        ResultHandler.ComparableResultSet res2 = createResultSet(10, 10, false);
        List<ResultHandler.ComparableResultSet> toCompare = Lists.newArrayList(res, res2, createResultSet(10, 11, false));
        List<Iterator<ResultHandler.ComparableRow>> iters = toCompare.stream().map(Iterable::iterator).collect(Collectors.toList());
        boolean foundMismatch = false;
        while (true)
        {
            List<ResultHandler.ComparableRow> rows = ResultHandler.rows(iters);
            if (rows.stream().allMatch(Objects::isNull))
                break;
            if (!rc.compareRows(Lists.newArrayList("eq1", "eq2", "diff"), null, rows))
            {
                foundMismatch = true;
            }
        }
        assertTrue(foundMismatch);
    }

    @Test
    public void testCompareRowsDifferentContent()
    {
        ResultComparator rc = new ResultComparator();
        ResultHandler.ComparableResultSet res = createResultSet(10, 10, false);
        ResultHandler.ComparableResultSet res2 = createResultSet(10, 10, false);
        List<ResultHandler.ComparableResultSet> toCompare = Lists.newArrayList(res, res2, createResultSet(10, 10, true));
        List<Iterator<ResultHandler.ComparableRow>> iters = toCompare.stream().map(Iterable::iterator).collect(Collectors.toList());
        while (true)
        {
            List<ResultHandler.ComparableRow> rows = ResultHandler.rows(iters);
            if (rows.stream().allMatch(Objects::isNull))
                break;
            assertFalse(rows.toString(), rc.compareRows(Lists.newArrayList("eq1", "eq2", "diff"), null, rows));
        }
    }

    @Test
    public void testCompareRowsDifferentColumnCount()
    {
        ResultComparator rc = new ResultComparator();
        ResultHandler.ComparableResultSet res = createResultSet(10, 10, false);
        ResultHandler.ComparableResultSet res2 = createResultSet(10, 10, false);
        List<ResultHandler.ComparableResultSet> toCompare = Lists.newArrayList(res, res2, createResultSet(11, 10, false));
        List<Iterator<ResultHandler.ComparableRow>> iters = toCompare.stream().map(Iterable::iterator).collect(Collectors.toList());
        while (true)
        {
            List<ResultHandler.ComparableRow> rows = ResultHandler.rows(iters);
            if (rows.stream().allMatch(Objects::isNull))
                break;
            assertFalse(rows.toString(), rc.compareRows(Lists.newArrayList("eq1", "eq2", "diff"), null, rows));
        }
    }

    @Test
    public void testResultHandler() throws IOException
    {
        List<String> targetHosts = Lists.newArrayList("hosta", "hostb", "hostc");
        File tmpDir = Files.createTempDirectory("testresulthandler").toFile();
        File queryDir = Files.createTempDirectory("queries").toFile();
        List<File> resultPaths = new ArrayList<>();
        targetHosts.forEach(host -> { File f = new File(tmpDir, host); f.mkdir(); resultPaths.add(f);});

        ResultHandler.ComparableResultSet res = createResultSet(10, 10, false);
        ResultHandler.ComparableResultSet res2 = createResultSet(10, 10, false);
        ResultHandler.ComparableResultSet res3 = createResultSet(10, 10, false);
        List<ResultHandler.ComparableResultSet> toCompare = Lists.newArrayList(res, res2, res3);
        FQLQuery query = new FQLQuery.Single("abcabc", QueryOptions.DEFAULT.getProtocolVersion().asInt(), QueryOptions.DEFAULT, 1111, 2222, 3333, "select * from xyz", Collections.emptyList());
        try (ResultHandler rh = new ResultHandler(targetHosts, resultPaths, queryDir))
        {
            rh.handleResults(query, toCompare);
        }
        List<Pair<FQLQuery, ResultHandler.ComparableResultSet>> results1 = readResultFile(resultPaths.get(0), queryDir);
        List<Pair<FQLQuery, ResultHandler.ComparableResultSet>> results2 = readResultFile(resultPaths.get(1), queryDir);
        List<Pair<FQLQuery, ResultHandler.ComparableResultSet>> results3 = readResultFile(resultPaths.get(2), queryDir);
        compareResults(results1, results2);
        compareResults(results1, results3);
        compareResults(results3, Collections.singletonList(Pair.create(query, res)));
    }

    @Test
    public void testResultHandlerWithDifference() throws IOException
    {
        List<String> targetHosts = Lists.newArrayList("hosta", "hostb", "hostc");
        File tmpDir = Files.createTempDirectory("testresulthandler").toFile();
        File queryDir = Files.createTempDirectory("queries").toFile();
        List<File> resultPaths = new ArrayList<>();
        targetHosts.forEach(host -> { File f = new File(tmpDir, host); f.mkdir(); resultPaths.add(f);});

        ResultHandler.ComparableResultSet res = createResultSet(10, 10, false);
        ResultHandler.ComparableResultSet res2 = createResultSet(10, 5, false);
        ResultHandler.ComparableResultSet res3 = createResultSet(10, 10, false);
        List<ResultHandler.ComparableResultSet> toCompare = Lists.newArrayList(res, res2, res3);
        FQLQuery query = new FQLQuery.Single("aaa", QueryOptions.DEFAULT.getProtocolVersion().asInt(), QueryOptions.DEFAULT, 123123, 11111, 22222, "select * from abcabc", Collections.emptyList());
        try (ResultHandler rh = new ResultHandler(targetHosts, resultPaths, queryDir))
        {
            rh.handleResults(query, toCompare);
        }
        List<Pair<FQLQuery, ResultHandler.ComparableResultSet>> results1 = readResultFile(resultPaths.get(0), queryDir);
        List<Pair<FQLQuery, ResultHandler.ComparableResultSet>> results2 = readResultFile(resultPaths.get(1), queryDir);
        List<Pair<FQLQuery, ResultHandler.ComparableResultSet>> results3 = readResultFile(resultPaths.get(2), queryDir);
        compareResults(results1, results3);
        compareResults(results2, Collections.singletonList(Pair.create(query, res2)));
    }

    @Test
    public void testResultHandlerMultipleResultSets() throws IOException
    {
        List<String> targetHosts = Lists.newArrayList("hosta", "hostb", "hostc");
        File tmpDir = Files.createTempDirectory("testresulthandler").toFile();
        File queryDir = Files.createTempDirectory("queries").toFile();
        List<File> resultPaths = new ArrayList<>();
        targetHosts.forEach(host -> { File f = new File(tmpDir, host); f.mkdir(); resultPaths.add(f);});
        List<Pair<FQLQuery, List<ResultHandler.ComparableResultSet>>> resultSets = new ArrayList<>();
        Random random = new Random();
        for (int i = 0; i < 10; i++)
        {
            List<ResultHandler.ComparableResultSet> results = new ArrayList<>();
            List<ByteBuffer> values = Collections.singletonList(ByteBufferUtil.bytes(i * 50));
            for (int jj = 0; jj < targetHosts.size(); jj++)
            {
                results.add(createResultSet(5, 1 + random.nextInt(10), true));
            }
            FQLQuery q = i % 2 == 0
                         ? new FQLQuery.Single("abc"+i,
                                             3,
                                             QueryOptions.forInternalCalls(values),
                                             i * 1000,
                                             12345,
                                             54321,
                                             "select * from xyz where id = "+i,
                                             values)
                         : new FQLQuery.Batch("abc"+i,
                                              3,
                                              QueryOptions.forInternalCalls(values),
                                              i * 1000,
                                              i * 54321,
                                              i * 12345,
                                              com.datastax.driver.core.BatchStatement.Type.UNLOGGED,
                                              Lists.newArrayList("select * from aaaa"),
                                              Collections.singletonList(values));

            resultSets.add(Pair.create(q, results));
        }
        try (ResultHandler rh = new ResultHandler(targetHosts, resultPaths, queryDir))
        {
            for (int i = 0; i < resultSets.size(); i++)
                rh.handleResults(resultSets.get(i).left, resultSets.get(i).right);
        }

        for (int i = 0; i < targetHosts.size(); i++)
            compareWithFile(resultPaths, queryDir, resultSets, i);
    }

    @Test
    public void testResultHandlerFailedQuery() throws IOException
    {
        List<String> targetHosts = Lists.newArrayList("hosta", "hostb", "hostc", "hostd");
        File tmpDir = Files.createTempDirectory("testresulthandler").toFile();
        File queryDir = Files.createTempDirectory("queries").toFile();
        List<File> resultPaths = new ArrayList<>();
        targetHosts.forEach(host -> { File f = new File(tmpDir, host); f.mkdir(); resultPaths.add(f);});

        List<Pair<FQLQuery, List<ResultHandler.ComparableResultSet>>> resultSets = new ArrayList<>();
        Random random = new Random();
        for (int i = 0; i < 10; i++)
        {
            List<ResultHandler.ComparableResultSet> results = new ArrayList<>();
            List<ByteBuffer> values = Collections.singletonList(ByteBufferUtil.bytes(i * 50));
            for (int jj = 0; jj < targetHosts.size(); jj++)
            {
                results.add(createResultSet(5, 1 + random.nextInt(10), true));
            }
            results.set(0, StoredResultSet.failed("testing abc"));
            results.set(3, StoredResultSet.failed("testing abc"));
            FQLQuery q = new FQLQuery.Single("abc"+i,
                                             3,
                                             QueryOptions.forInternalCalls(values),
                                             i * 1000,
                                             i * 12345,
                                             i * 54321,
                                             "select * from xyz where id = "+i,
                                             values);
            resultSets.add(Pair.create(q, results));
        }
        try (ResultHandler rh = new ResultHandler(targetHosts, resultPaths, queryDir))
        {
            for (int i = 0; i < resultSets.size(); i++)
                rh.handleResults(resultSets.get(i).left, resultSets.get(i).right);
        }
        for (int i = 0; i < targetHosts.size(); i++)
            compareWithFile(resultPaths, queryDir, resultSets, i);
    }

    @Test
    public void testCompare()
    {
        FQLQuery q1 = new FQLQuery.Single("abc", 0, QueryOptions.DEFAULT, 123, 111, 222, "aaaa", Collections.emptyList());
        FQLQuery q2 = new FQLQuery.Single("abc", 0, QueryOptions.DEFAULT, 123, 111, 222,"aaaa", Collections.emptyList());

        assertEquals(0, q1.compareTo(q2));
        assertEquals(0, q2.compareTo(q1));

        FQLQuery q3 = new FQLQuery.Batch("abc", 0, QueryOptions.DEFAULT, 123, 111, 222, com.datastax.driver.core.BatchStatement.Type.UNLOGGED, Collections.emptyList(), Collections.emptyList());
        // single queries before batch queries
        assertTrue(q1.compareTo(q3) < 0);
        assertTrue(q3.compareTo(q1) > 0);

        // check that smaller query time
        FQLQuery q4 = new FQLQuery.Single("abc", 0, QueryOptions.DEFAULT, 124, 111, 222, "aaaa", Collections.emptyList());
        assertTrue(q1.compareTo(q4) < 0);
        assertTrue(q4.compareTo(q1) > 0);

        FQLQuery q5 = new FQLQuery.Batch("abc", 0, QueryOptions.DEFAULT, 124, 111, 222, com.datastax.driver.core.BatchStatement.Type.UNLOGGED, Collections.emptyList(), Collections.emptyList());
        assertTrue(q1.compareTo(q5) < 0);
        assertTrue(q5.compareTo(q1) > 0);

        FQLQuery q6 = new FQLQuery.Single("abc", 0, QueryOptions.DEFAULT, 123, 111, 222, "aaaa", Collections.singletonList(ByteBufferUtil.bytes(10)));
        FQLQuery q7 = new FQLQuery.Single("abc", 0, QueryOptions.DEFAULT, 123, 111, 222, "aaaa", Collections.emptyList());
        assertTrue(q6.compareTo(q7) > 0);
        assertTrue(q7.compareTo(q6) < 0);

        FQLQuery q8 = new FQLQuery.Single("abc", 0, QueryOptions.DEFAULT, 123, 111, 222, "aaaa", Collections.singletonList(ByteBufferUtil.bytes("a")));
        FQLQuery q9 = new FQLQuery.Single("abc", 0, QueryOptions.DEFAULT, 123, 111, 222, "aaaa", Collections.singletonList(ByteBufferUtil.bytes("b")));
        assertTrue(q8.compareTo(q9) < 0);
        assertTrue(q9.compareTo(q8) > 0);
    }

    @Test
    public void testFQLQuerySingleToStatement()
    {
        List<ByteBuffer> values = new ArrayList<>();
        for (int i = 0; i < 10; i++)
            values.add(ByteBufferUtil.bytes(i));
        FQLQuery.Single single = new FQLQuery.Single("xyz",
                                                     QueryOptions.DEFAULT.getProtocolVersion().asInt(),
                                                     QueryOptions.forInternalCalls(values),
                                                     1234,
                                                     12345,
                                                     54321,
                                                     "select * from aaa",
                                                     values);
        Statement stmt = single.toStatement();
        assertEquals(stmt.getDefaultTimestamp(), 12345);
        assertTrue(stmt instanceof SimpleStatement);
        SimpleStatement simpleStmt = (SimpleStatement)stmt;
        assertEquals("select * from aaa",simpleStmt.getQueryString(CodecRegistry.DEFAULT_INSTANCE));
        assertArrayEquals(values.toArray(), simpleStmt.getValues(com.datastax.driver.core.ProtocolVersion.fromInt(QueryOptions.DEFAULT.getProtocolVersion().asInt()), CodecRegistry.DEFAULT_INSTANCE));
    }


    @Test
    public void testFQLQueryBatchToStatement()
    {
        List<List<ByteBuffer>> values = new ArrayList<>();
        List<String> queries = new ArrayList<>();
        for (int bqCount = 0; bqCount < 10; bqCount++)
        {
            queries.add("select * from asdf where x = ? and y = " + bqCount);
            List<ByteBuffer> queryValues = new ArrayList<>();
            for (int i = 0; i < 10; i++)
                queryValues.add(ByteBufferUtil.bytes(i + ":" + bqCount));
            values.add(queryValues);
        }

        FQLQuery.Batch batch = new FQLQuery.Batch("xyz",
                                                   QueryOptions.DEFAULT.getProtocolVersion().asInt(),
                                                   QueryOptions.DEFAULT,
                                                   1234,
                                                   12345,
                                                   54321,
                                                   com.datastax.driver.core.BatchStatement.Type.UNLOGGED,
                                                   queries,
                                                   values);
        Statement stmt = batch.toStatement();
        assertEquals(stmt.getDefaultTimestamp(), 12345);
        assertTrue(stmt instanceof com.datastax.driver.core.BatchStatement);
        com.datastax.driver.core.BatchStatement batchStmt = (com.datastax.driver.core.BatchStatement)stmt;
        List<Statement> statements = Lists.newArrayList(batchStmt.getStatements());
        List<Statement> fromFQLQueries = batch.queries.stream().map(FQLQuery.Single::toStatement).collect(Collectors.toList());
        assertEquals(statements.size(), fromFQLQueries.size());
        assertEquals(12345, batchStmt.getDefaultTimestamp());
        for (int i = 0; i < statements.size(); i++)
            compareStatements(statements.get(i), fromFQLQueries.get(i));
    }

    @Test
    public void testParser() {
        QueryReplayer.ParsedTargetHost pth;
        pth = fromString("127.0.0.1");
        assertEquals("127.0.0.1", pth.host);
        assertEquals(9042, pth.port );
        assertNull(pth.user);
        assertNull(pth.password);

        pth = fromString("127.0.0.1:3333");
        assertEquals("127.0.0.1", pth.host);
        assertEquals(3333, pth.port );
        assertNull(pth.user);
        assertNull(pth.password);

        pth = fromString("aaa:bbb@127.0.0.1:3333");
        assertEquals("127.0.0.1", pth.host);
        assertEquals(3333, pth.port );
        assertEquals("aaa", pth.user);
        assertEquals("bbb", pth.password);

        pth = fromString("aaa:bbb@127.0.0.1");
        assertEquals("127.0.0.1", pth.host);
        assertEquals(9042, pth.port );
        assertEquals("aaa", pth.user);
        assertEquals("bbb", pth.password);
    }

    @Test(expected = RuntimeException.class)
    public void testNoPass()
    {
        fromString("blabla@abc.com:1234");
    }

    @Test(expected = RuntimeException.class)
    public void testBadPort()
    {
        fromString("aaa:bbb@abc.com:xyz");
    }

    @Test (expected = IORuntimeException.class)
    public void testFutureVersion() throws Exception
    {
        FQLQueryReader reader = new FQLQueryReader();
        File dir = Files.createTempDirectory("chronicle").toFile();
        try (ChronicleQueue queue = SingleChronicleQueueBuilder.single(dir).build())
        {
            ExcerptAppender appender = queue.acquireAppender();
            appender.writeDocument(new BinLog.ReleaseableWriteMarshallable() {
                protected long version()
                {
                    return 999;
                }

                protected String type()
                {
                    return FullQueryLogger.SINGLE_QUERY;
                }

                public void writeMarshallablePayload(WireOut wire)
                {
                    wire.write("future-field").text("future_value");
                }

                public void release()
                {

                }
            });

            ExcerptTailer tailer = queue.createTailer();
            tailer.readDocument(reader);
        }
        catch (Exception e)
        {
            assertTrue(e.getMessage().contains("Unsupported record version"));
            throw e;
        }

    }

    @Test (expected = IORuntimeException.class)
    public void testUnknownRecord() throws Exception
    {
        FQLQueryReader reader = new FQLQueryReader();
        File dir = Files.createTempDirectory("chronicle").toFile();
        try (ChronicleQueue queue = SingleChronicleQueueBuilder.single(dir).build())
        {
            ExcerptAppender appender = queue.acquireAppender();
            appender.writeDocument(new BinLog.ReleaseableWriteMarshallable() {
                protected long version()
                {
                    return FullQueryLogger.CURRENT_VERSION;
                }

                protected String type()
                {
                    return "unknown-type";
                }

                public void writeMarshallablePayload(WireOut wire)
                {
                    wire.write("unknown-field").text("unknown_value");
                }

                public void release()
                {

                }
            });

            ExcerptTailer tailer = queue.createTailer();
            tailer.readDocument(reader);
        }
        catch (Exception e)
        {
            assertTrue(e.getMessage().contains("Unsupported record type field"));
            throw e;
        }

    }

    private void compareStatements(Statement statement1, Statement statement2)
    {
        assertTrue(statement1 instanceof SimpleStatement && statement2 instanceof SimpleStatement);
        SimpleStatement simpleStmt1 = (SimpleStatement)statement1;
        SimpleStatement simpleStmt2 = (SimpleStatement)statement2;
        assertEquals(simpleStmt1.getQueryString(CodecRegistry.DEFAULT_INSTANCE), simpleStmt2.getQueryString(CodecRegistry.DEFAULT_INSTANCE));
        assertArrayEquals(simpleStmt1.getValues(com.datastax.driver.core.ProtocolVersion.fromInt(QueryOptions.DEFAULT.getProtocolVersion().asInt()), CodecRegistry.DEFAULT_INSTANCE),
                          simpleStmt2.getValues(com.datastax.driver.core.ProtocolVersion.fromInt(QueryOptions.DEFAULT.getProtocolVersion().asInt()), CodecRegistry.DEFAULT_INSTANCE));

    }

    private File generateQueries(int count, boolean random) throws IOException
    {
        Random r = new Random();
        File dir = Files.createTempDirectory("chronicle").toFile();
        try (ChronicleQueue readQueue = SingleChronicleQueueBuilder.single(dir).build())
        {
            ExcerptAppender appender = readQueue.acquireAppender();

            for (int i = 0; i < count; i++)
            {
                long timestamp = random ? Math.abs(r.nextLong() % 10000) : i;
                if (random ? r.nextBoolean() : i % 2 == 0)
                {
                    String query = "abcdefghijklm " + i;
                    QueryState qs = r.nextBoolean() ? queryState() : queryState("querykeyspace");
                    FullQueryLogger.Query  q = new FullQueryLogger.Query(query, QueryOptions.DEFAULT, qs, timestamp);
                    appender.writeDocument(q);
                    q.release();
                }
                else
                {
                    int batchSize = random ? r.nextInt(99) + 1 : i + 1;
                    List<String> queries = new ArrayList<>(batchSize);
                    List<List<ByteBuffer>> values = new ArrayList<>(batchSize);
                    for (int jj = 0; jj < (random ? r.nextInt(batchSize) : 10); jj++)
                    {
                        queries.add("aaaaaa batch "+i+":"+jj);
                        values.add(Collections.emptyList());
                    }
                    FullQueryLogger.Batch batch = new FullQueryLogger.Batch(BatchStatement.Type.UNLOGGED,
                                                                            queries,
                                                                            values,
                                                                            QueryOptions.DEFAULT,
                                                                            queryState("someks"),
                                                                            timestamp);
                    appender.writeDocument(batch);
                    batch.release();
                }
            }
        }
        return dir;
    }

    private QueryState queryState()
    {
        return QueryState.forInternalCalls();
    }

    private QueryState queryState(String keyspace)
    {
        ClientState clientState = ClientState.forInternalCalls(keyspace);
        return new QueryState(clientState);
    }

    static ResultHandler.ComparableResultSet createResultSet(int columnCount, int rowCount, boolean random)
    {
        List<Pair<String, String>> columnDefs = new ArrayList<>(columnCount);
        Random r = new Random();
        for (int i = 0; i < columnCount; i++)
        {
            columnDefs.add(Pair.create("a" + i, "int"));
        }
        ResultHandler.ComparableColumnDefinitions colDefs = new StoredResultSet.StoredComparableColumnDefinitions(columnDefs, false, null);
        List<ResultHandler.ComparableRow> rows = new ArrayList<>();
        for (int i = 0; i < rowCount; i++)
        {
            List<ByteBuffer> row = new ArrayList<>(columnCount);
            for (int jj = 0; jj < columnCount; jj++)
                row.add(ByteBufferUtil.bytes(i + " col " + jj + (random ? r.nextInt() : "")));

            rows.add(new StoredResultSet.StoredComparableRow(row, colDefs));
        }
        return new StoredResultSet(colDefs, true, false, null, rows::iterator);
    }

    private static void compareWithFile(List<File> dirs, File resultDir, List<Pair<FQLQuery, List<ResultHandler.ComparableResultSet>>> resultSets, int idx)
    {
        List<Pair<FQLQuery, ResultHandler.ComparableResultSet>> results1 = readResultFile(dirs.get(idx), resultDir);
        for (int i = 0; i < resultSets.size(); i++)
        {
            FQLQuery query = resultSets.get(i).left;
            List<Pair<FQLQuery, ResultHandler.ComparableResultSet>> toCompare = Collections.singletonList(Pair.create(query, resultSets.get(i).right.get(idx)));
            compareResults(Collections.singletonList(results1.get(i)), toCompare);
        }
    }

    private static List<Pair<FQLQuery, ResultHandler.ComparableResultSet>> readResultFile(File dir, File queryDir)
    {
        List<Pair<FQLQuery, ResultHandler.ComparableResultSet>> resultSets = new ArrayList<>();
        try (ChronicleQueue q = SingleChronicleQueueBuilder.single(dir).build();
             ChronicleQueue queryQ = SingleChronicleQueueBuilder.single(queryDir).build())
        {
            ExcerptTailer queryTailer = queryQ.createTailer();
            FQLQueryReader queryReader = new FQLQueryReader();
            Compare.StoredResultSetIterator resultSetIterator = new Compare.StoredResultSetIterator(q.createTailer());
            // we need to materialize the rows in-memory to compare them easier in these tests
            while (resultSetIterator.hasNext())
            {
                ResultHandler.ComparableResultSet resultSetFromDisk = resultSetIterator.next();
                Iterator<ResultHandler.ComparableRow> rowIterFromDisk = resultSetFromDisk.iterator();
                queryTailer.readDocument(queryReader);

                FQLQuery query = queryReader.getQuery();
                List<ResultHandler.ComparableRow> rows = new ArrayList<>();
                while (rowIterFromDisk.hasNext())
                {
                    rows.add(rowIterFromDisk.next());
                }
                resultSets.add(Pair.create(query, new StoredResultSet(resultSetFromDisk.getColumnDefinitions(),
                                                                      resultSetIterator.hasNext(),
                                                                      resultSetFromDisk.wasFailed(),
                                                                      resultSetFromDisk.getFailureException(),
                                                                      rows::iterator)));
            }
        }
        return resultSets;
    }
}
