/*
 * 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.phoenix.index;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.DoNotRetryIOException;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.phoenix.coprocessor.GlobalIndexRegionScanner;
import org.apache.phoenix.coprocessor.IndexRebuildRegionScanner;
import org.apache.phoenix.coprocessor.IndexToolVerificationResult;
import org.apache.phoenix.hbase.index.IndexRegionObserver;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.mapreduce.index.IndexVerificationOutputRepository;
import org.apache.phoenix.query.BaseConnectionlessQueryTest;
import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTableKey;
import org.apache.phoenix.util.*;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.mockito.Matchers;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.MockitoAnnotations;

import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

import static org.apache.phoenix.hbase.index.IndexRegionObserver.UNVERIFIED_BYTES;
import static org.apache.phoenix.hbase.index.IndexRegionObserver.VERIFIED_BYTES;
import static org.apache.phoenix.query.QueryConstants.EMPTY_COLUMN_BYTES;
import static org.junit.Assert.*;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.when;

public class VerifySingleIndexRowTest extends BaseConnectionlessQueryTest {

    private static final int INDEX_TABLE_EXPIRY_SEC = 1;
    private static final String UNEXPECTED_COLUMN = "0:UNEXPECTED_COLUMN";
    public static final String FIRST_ID = "FIRST_ID";
    public static final String SECOND_ID = "SECOND_ID";
    public static final String FIRST_VALUE = "FIRST_VALUE";
    public static final String SECOND_VALUE = "SECOND_VALUE";
    public static final String
            CREATE_TABLE_DDL = "CREATE TABLE IF NOT EXISTS %s (FIRST_ID BIGINT NOT NULL, "
            + "SECOND_ID BIGINT NOT NULL, FIRST_VALUE VARCHAR(20), "
            + "SECOND_VALUE INTEGER "
            + "CONSTRAINT PK PRIMARY KEY(FIRST_ID, SECOND_ID)) COLUMN_ENCODED_BYTES=0";

    public static final String
            CREATE_INDEX_DDL = "CREATE INDEX %s ON %s (SECOND_VALUE) INCLUDE (FIRST_VALUE)";
    public static final String COMPLETE_ROW_UPSERT = "UPSERT INTO %s VALUES (?,?,?,?)";
    public static final String PARTIAL_ROW_UPSERT = "UPSERT INTO %s (%s, %s, %s) VALUES (?,?,?)";
    public static final String DELETE_ROW_DML = "DELETE FROM %s WHERE %s = ?  AND %s = ?";
    public static final String INCLUDED_COLUMN = "0:FIRST_VALUE";

    @Rule
    public ExpectedException exceptionRule = ExpectedException.none();

    private enum TestType {
        //set of mutations matching expected mutations
        VALID_EXACT_MATCH,
        //mix of delete and put mutations
        VALID_MIX_MUTATIONS,
        //only incoming unverified mutations
        VALID_NEW_UNVERIFIED_MUTATIONS,
        //extra mutations mimicking incoming mutations
        VALID_MORE_MUTATIONS,
        // mimicking the case where the data cells expired but index has them still
        VALID_EXTRA_CELL,
        EXPIRED,
        INVALID_EXTRA_CELL,
        INVALID_EMPTY_CELL,
        INVALID_CELL_VALUE,
        INVALID_COLUMN
    }

    public static class UnitTestClock extends EnvironmentEdge {
        long initialTime;
        long delta;

        public UnitTestClock(long delta) {
            initialTime = System.currentTimeMillis() + delta;
            this.delta = delta;
        }

        @Override
        public long currentTime() {
            return System.currentTimeMillis() + delta;
        }
    }

    @Mock
    Result indexRow;
    @Mock
    IndexRebuildRegionScanner rebuildScanner;
    List<Mutation> actualMutationList;
    String schema, table, dataTableFullName, index, indexTableFullName;
    PTable pIndexTable, pDataTable;
    Put put = null;
    Delete delete = null;
    PhoenixConnection pconn;
    IndexToolVerificationResult.PhaseResult actualPR;
    public Map<byte[], List<Mutation>> indexKeyToMutationMap = null;
    Set<byte[]> mostRecentIndexRowKeys;
    private IndexMaintainer indexMaintainer;

    @Before
    public void setup() throws SQLException, IOException {
        MockitoAnnotations.initMocks(this);
        createDBObject();
        createMutationsWithUpserts();
        initializeRebuildScannerAttributes();
        initializeGlobalMockitoSetup();
    }

    public void createDBObject() throws SQLException {
        try(Connection conn = DriverManager.getConnection(getUrl(), new Properties())) {
            schema = generateUniqueName();
            table = generateUniqueName();
            index = generateUniqueName();
            dataTableFullName = SchemaUtil.getQualifiedTableName(schema, table);
            indexTableFullName = SchemaUtil.getQualifiedTableName(schema, index);

            conn.createStatement().execute(String.format(CREATE_TABLE_DDL, dataTableFullName));
            conn.createStatement().execute(String.format(CREATE_INDEX_DDL, index, dataTableFullName));
            conn.commit();

            pconn = conn.unwrap(PhoenixConnection.class);
            pIndexTable = pconn.getTable(new PTableKey(pconn.getTenantId(), indexTableFullName));
            pDataTable = pconn.getTable(new PTableKey(pconn.getTenantId(), dataTableFullName));
        }
    }

    private void createMutationsWithUpserts() throws SQLException, IOException {
        deleteRow(2, 3);
        upsertPartialRow(2, 3, "abc");
        upsertCompleteRow(2, 3, "hik", 8);
        upsertPartialRow(2, 3, 10);
        upsertPartialRow(2,3,4);
        deleteRow(2, 3);
        upsertPartialRow(2,3, "def");
        upsertCompleteRow(2, 3, null, 20);
        upsertPartialRow(2,3, "wert");
    }

    private void deleteRow(int key1, int key2) throws SQLException, IOException {
        try(Connection conn = DriverManager.getConnection(getUrl(), new Properties())){
            PreparedStatement ps =
                    conn.prepareStatement(
                            String.format(DELETE_ROW_DML, dataTableFullName, FIRST_ID, SECOND_ID));
            ps.setInt(1, key1);
            ps.setInt(2, key2);
            ps.execute();
            convertUpsertToMutations(conn);
        }
    }

    private void upsertPartialRow(int key1, int key2, String val1)
            throws SQLException, IOException {

        try(Connection conn = DriverManager.getConnection(getUrl(), new Properties())){
            PreparedStatement ps =
                    conn.prepareStatement(
                            String.format(PARTIAL_ROW_UPSERT, dataTableFullName, FIRST_ID, SECOND_ID,
                                    FIRST_VALUE));
            ps.setInt(1, key1);
            ps.setInt(2, key2);
            ps.setString(3, val1);
            ps.execute();
            convertUpsertToMutations(conn);
        }
    }

    private void upsertPartialRow(int key1, int key2, int value1)
            throws SQLException, IOException {

        try(Connection conn = DriverManager.getConnection(getUrl(), new Properties())){
            PreparedStatement
                    ps =
                    conn.prepareStatement(
                            String.format(PARTIAL_ROW_UPSERT, dataTableFullName, FIRST_ID, SECOND_ID,
                                    SECOND_VALUE));
            ps.setInt(1, key1);
            ps.setInt(2, key2);
            ps.setInt(3, value1);
            ps.execute();
            convertUpsertToMutations(conn);
        }
    }

    private void upsertCompleteRow(int key1, int key2, String val1
            , int val2) throws SQLException, IOException {
        try(Connection conn = DriverManager.getConnection(getUrl(), new Properties())) {
            PreparedStatement
                    ps = conn.prepareStatement(String.format(COMPLETE_ROW_UPSERT, dataTableFullName));
            ps.setInt(1, key1);
            ps.setInt(2, key2);
            ps.setString(3, val1);
            ps.setInt(4, val2);
            ps.execute();
            convertUpsertToMutations(conn);
        }
    }

    private void convertUpsertToMutations(Connection conn) throws SQLException, IOException {
        Iterator<Pair<byte[],List<KeyValue>>>
                dataTableNameAndMutationKeyValuesIter = PhoenixRuntime.getUncommittedDataIterator(conn);
        Pair<byte[], List<KeyValue>> elem = dataTableNameAndMutationKeyValuesIter.next();
        byte[] key = elem.getSecond().get(0).getRow();
        long mutationTS = EnvironmentEdgeManager.currentTimeMillis();

        for (KeyValue kv : elem.getSecond()) {
            Cell cell =
                    CellUtil.createCell(kv.getRow(), kv.getFamily(), kv.getQualifier(),
                            mutationTS, kv.getType(), kv.getValue());
            if (KeyValue.Type.codeToType(cell.getTypeByte()) == KeyValue.Type.Put) {
                if (put == null ) {
                    put = new Put(key);
                }
                put.add(cell);
            } else {
                if (delete == null) {
                    delete = new Delete(key);
                }
                delete.addDeleteMarker(cell);
            }
        }
    }

    private void initializeRebuildScannerAttributes() {
        when(rebuildScanner.setIndexTableTTL(Matchers.anyInt())).thenCallRealMethod();
        when(rebuildScanner.setIndexMaintainer(Matchers.<IndexMaintainer>any())).thenCallRealMethod();
        when(rebuildScanner.setMaxLookBackInMills(Matchers.anyLong())).thenCallRealMethod();
        rebuildScanner.setIndexTableTTL(HConstants.FOREVER);
        indexMaintainer = pIndexTable.getIndexMaintainer(pDataTable, pconn);
        rebuildScanner.setIndexMaintainer(indexMaintainer);
        // set the maxLookBack to infinite to avoid the compaction
        rebuildScanner.setMaxLookBackInMills(Long.MAX_VALUE);
    }

    private void initializeGlobalMockitoSetup() throws IOException {
        //setup
        when(GlobalIndexRegionScanner.getIndexRowKey(indexMaintainer, put)).thenCallRealMethod();
        when(rebuildScanner.prepareIndexMutations(put, delete, indexKeyToMutationMap, mostRecentIndexRowKeys)).thenCallRealMethod();
        when(rebuildScanner.verifySingleIndexRow(Matchers.<Result>any(), Matchers.<Map>any(), Matchers.<Set>any(),
                Matchers.<IndexToolVerificationResult.PhaseResult>any(), Matchers.anyBoolean())).thenCallRealMethod();
        doNothing().when(rebuildScanner)
                .logToIndexToolOutputTable(Matchers.<byte[]>any(),Matchers.<byte[]>any(),
                        Mockito.anyLong(),Mockito.anyLong(), Mockito.anyString(),
                        Matchers.<byte[]>any(), Matchers.<byte[]>any(), Matchers.anyBoolean(),
                    Mockito.<IndexVerificationOutputRepository.IndexVerificationErrorType>any());
        doNothing().when(rebuildScanner)
                .logToIndexToolOutputTable(Matchers.<byte[]>any(),Matchers.<byte[]>any(),
                        Mockito.anyLong(),Mockito.anyLong(), Mockito.anyString(),
                    Matchers.anyBoolean(),
                    Mockito.<IndexVerificationOutputRepository.IndexVerificationErrorType>any());

        //populate the local map to use to create actual mutations
        indexKeyToMutationMap = Maps.newTreeMap(Bytes.BYTES_COMPARATOR);
        rebuildScanner.prepareIndexMutations(put, delete, indexKeyToMutationMap, mostRecentIndexRowKeys);
    }

    private byte[] getValidRowKey() {
        return indexKeyToMutationMap.entrySet().iterator().next().getKey();
    }

    @Test
    public void testVerifySingleIndexRow_validIndexRowCount_nonZero() throws IOException {
        IndexToolVerificationResult.PhaseResult expectedPR = getValidPhaseResult();
        for (Map.Entry<byte[], List<Mutation>>
                entry : indexKeyToMutationMap.entrySet()) {
            initializeLocalMockitoSetup(entry, TestType.VALID_EXACT_MATCH);
            //test code
            rebuildScanner.verifySingleIndexRow(indexRow, indexKeyToMutationMap, mostRecentIndexRowKeys, actualPR, true);

            assertTrue(actualPR.equals(expectedPR));
        }
    }

    @Test
    public void testVerifySingleIndexRow_validIndexRowCount_moreActual() throws IOException {
        IndexToolVerificationResult.PhaseResult expectedPR = getValidPhaseResult();
        for (Map.Entry<byte[], List<Mutation>>
                entry : indexKeyToMutationMap.entrySet()) {
            initializeLocalMockitoSetup(entry, TestType.VALID_MORE_MUTATIONS);
            //test code
            rebuildScanner.verifySingleIndexRow(indexRow, indexKeyToMutationMap, mostRecentIndexRowKeys, actualPR, true);

            assertTrue(actualPR.equals(expectedPR));
        }
    }

    @Test
    public void testVerifySingleIndexRow_allMix() throws IOException {
        IndexToolVerificationResult.PhaseResult expectedPR = getValidPhaseResult();
        for (Map.Entry<byte[], List<Mutation>>
                entry : indexKeyToMutationMap.entrySet()) {
            initializeLocalMockitoSetup(entry, TestType.VALID_MIX_MUTATIONS);
            //test code
            rebuildScanner.verifySingleIndexRow(indexRow, indexKeyToMutationMap, mostRecentIndexRowKeys, actualPR, true);

            assertTrue(actualPR.equals(expectedPR));
        }
    }

    @Test
    public void testVerifySingleIndexRow_allUnverified() throws IOException {
        IndexToolVerificationResult.PhaseResult expectedPR = getValidPhaseResult();
        for (Map.Entry<byte[], List<Mutation>>
                entry : indexKeyToMutationMap.entrySet()) {
            initializeLocalMockitoSetup(entry, TestType.VALID_NEW_UNVERIFIED_MUTATIONS);
            //test code
            rebuildScanner.verifySingleIndexRow(indexRow, indexKeyToMutationMap, mostRecentIndexRowKeys, actualPR, true);

            assertTrue(actualPR.equals(expectedPR));
        }
    }

    @Test
    public void testVerifySingleIndexRow_expiredIndexRowCount_nonZero() throws IOException {
        IndexToolVerificationResult.PhaseResult
                expectedPR = new IndexToolVerificationResult.PhaseResult(0, 1, 0, 0, 0, 0,0,0);
        try {
            for (Map.Entry<byte[], List<Mutation>>
                    entry : indexKeyToMutationMap.entrySet()) {
                initializeLocalMockitoSetup(entry, TestType.EXPIRED);
                expireThisRow();
                //test code
                rebuildScanner.verifySingleIndexRow(indexRow, indexKeyToMutationMap, mostRecentIndexRowKeys, actualPR, true);

                assertTrue(actualPR.equals(expectedPR));
            }
        } finally {
            EnvironmentEdgeManager.reset();
        }
    }

    @Test
    public void testVerifySingleIndexRow_invalidIndexRowCount_cellValue() throws IOException {
        IndexToolVerificationResult.PhaseResult expectedPR = getInvalidPhaseResult();
        expectedPR.setIndexHasExtraCellsCount(1);
        for (Map.Entry<byte[], List<Mutation>>
                entry : indexKeyToMutationMap.entrySet()) {
            initializeLocalMockitoSetup(entry, TestType.INVALID_CELL_VALUE);
            //test code
            rebuildScanner.verifySingleIndexRow(indexRow, indexKeyToMutationMap, mostRecentIndexRowKeys, actualPR, true);

            assertTrue(actualPR.equals(expectedPR));
        }
    }

    @Test
    public void testVerifySingleIndexRow_invalidIndexRowCount_emptyCell() throws IOException {
        IndexToolVerificationResult.PhaseResult expectedPR = getInvalidPhaseResult();
        for (Map.Entry<byte[], List<Mutation>>
                entry : indexKeyToMutationMap.entrySet()) {
            initializeLocalMockitoSetup(entry, TestType.INVALID_EMPTY_CELL);
            //test code
            rebuildScanner.verifySingleIndexRow(indexRow, indexKeyToMutationMap, mostRecentIndexRowKeys, actualPR, true);

            assertTrue(actualPR.equals(expectedPR));
        }
    }

    @Test
    public void testVerifySingleIndexRow_invalidIndexRowCount_diffColumn() throws IOException {
        IndexToolVerificationResult.PhaseResult expectedPR = getInvalidPhaseResult();
        expectedPR.setIndexHasExtraCellsCount(1);
        for (Map.Entry<byte[], List<Mutation>>
                entry : indexKeyToMutationMap.entrySet()) {
            initializeLocalMockitoSetup(entry, TestType.INVALID_COLUMN);
            //test code
            rebuildScanner.verifySingleIndexRow(indexRow, indexKeyToMutationMap, mostRecentIndexRowKeys, actualPR, true);

            assertTrue(actualPR.equals(expectedPR));
        }
    }

    @Test
    public void testVerifySingleIndexRow_invalidIndexRowCount_extraCell() throws IOException {
        IndexToolVerificationResult.PhaseResult expectedPR = getInvalidPhaseResult();
        for (Map.Entry<byte[], List<Mutation>>
                entry : indexKeyToMutationMap.entrySet()) {
            initializeLocalMockitoSetup(entry, TestType.INVALID_EXTRA_CELL);
            //test code
            rebuildScanner.verifySingleIndexRow(indexRow, indexKeyToMutationMap, mostRecentIndexRowKeys, actualPR, true);

            assertTrue(actualPR.equals(expectedPR));
        }
    }

    @Test
    public void testVerifySingleIndexRow_expectedMutations_null() throws IOException {
        when(indexRow.getRow()).thenReturn(Bytes.toBytes(1));
        exceptionRule.expect(DoNotRetryIOException.class);
        exceptionRule.expectMessage(IndexRebuildRegionScanner.NO_EXPECTED_MUTATION);
        rebuildScanner.verifySingleIndexRow(indexRow, indexKeyToMutationMap, mostRecentIndexRowKeys, actualPR, true);
    }

    @Test
    public void testVerifySingleIndexRow_validIndexRowCount_extraCell() throws IOException {
        for (Map.Entry<byte[], List<Mutation>>
                entry : indexKeyToMutationMap.entrySet()) {
            initializeLocalMockitoSetup(entry, TestType.VALID_EXTRA_CELL);
            //test code
            rebuildScanner.verifySingleIndexRow(indexRow, indexKeyToMutationMap, mostRecentIndexRowKeys, actualPR, true);

            assertEquals(1, actualPR.getIndexHasExtraCellsCount());
        }
    }

    // Test the major compaction on index table only.
    // There is at least one expected mutation within maxLookBack that has its matching one in the actual list.
    // However there are some expected mutations outside of maxLookBack, which matching ones in actual list may be compacted away.
    // We will report such row as a valid row.
    @Test
    public void testVerifySingleIndexRow_compactionOnIndexTable_atLeastOneExpectedMutationWithinMaxLookBack() throws Exception {
        String dataRowKey = "k1";
        byte[] indexRowKey1Bytes = generateIndexRowKey(dataRowKey, "val1");
        ManualEnvironmentEdge injectEdge = new ManualEnvironmentEdge();
        injectEdge.setValue(1);
        EnvironmentEdgeManager.injectEdge(injectEdge);

        List<Mutation> expectedMutations = new ArrayList<>();
        List<Mutation> actualMutations = new ArrayList<>();
        // change the maxLookBack from infinite to some interval, which allows to simulate the mutation beyond the maxLookBack window.
        long maxLookbackInMills = 10 * 1000;
        rebuildScanner.setMaxLookBackInMills(maxLookbackInMills);

        Put put = new Put(indexRowKey1Bytes);
        Cell cell = CellUtil.createCell(indexRowKey1Bytes,
                QueryConstants.DEFAULT_COLUMN_FAMILY_BYTES,
                QueryConstants.EMPTY_COLUMN_BYTES,
                EnvironmentEdgeManager.currentTimeMillis(),
                KeyValue.Type.Put.getCode(),
                IndexRegionObserver.VERIFIED_BYTES);
        put.add(cell);
        // This mutation is beyond maxLookBack, so add it to expectedMutations only.
        expectedMutations.add(put);

        // advance the time of maxLookBack, so last mutation will be outside of maxLookBack,
        // next mutation will be within maxLookBack
        injectEdge.incrementValue(maxLookbackInMills);
        put =  new Put(indexRowKey1Bytes);
        cell = CellUtil.createCell(indexRowKey1Bytes,
                QueryConstants.DEFAULT_COLUMN_FAMILY_BYTES,
                QueryConstants.EMPTY_COLUMN_BYTES,
                EnvironmentEdgeManager.currentTimeMillis(),
                KeyValue.Type.Put.getCode(),
                IndexRegionObserver.VERIFIED_BYTES);
        put.add(cell);
        // This mutation is in both expectedMutations and actualMutations, as it is within the maxLookBack, so it will not get chance to be compacted away
        expectedMutations.add(put);
        actualMutations.add(put);
        Result actualMutationsScanResult = Result.create(Arrays.asList(cell));

        Map<byte[], List<Mutation>> indexKeyToMutationMap = Maps.newTreeMap((Bytes.BYTES_COMPARATOR));
        indexKeyToMutationMap.put(indexRowKey1Bytes, expectedMutations);
        when(rebuildScanner.prepareActualIndexMutations(any(Result.class))).thenReturn(actualMutations);
        when(indexRow.getRow()).thenReturn(indexRowKey1Bytes);
        injectEdge.incrementValue(1);
        IndexToolVerificationResult.PhaseResult actualPR = new IndexToolVerificationResult.PhaseResult();
        // Report this validation as a success
        assertTrue(rebuildScanner.verifySingleIndexRow(indexRow, indexKeyToMutationMap, mostRecentIndexRowKeys, actualPR, true));
        // validIndexRowCount = 1
        IndexToolVerificationResult.PhaseResult expectedPR = new IndexToolVerificationResult.PhaseResult(1, 0, 0, 0, 0, 0, 0, 0);
        assertTrue(actualPR.equals(expectedPR));
    }

    // Test the major compaction on index table only.
    // All expected mutations are beyond the maxLookBack, and there are no matching ones in the actual list because of major compaction.
    // We will report such row as an invalid beyond maxLookBack row.
    @Test
    public void testVerifySingleIndexRow_compactionOnIndexTable_noExpectedMutationWithinMaxLookBack() throws Exception {
        String dataRowKey = "k1";
        byte[] indexRowKey1Bytes = generateIndexRowKey(dataRowKey, "val1");
        List<Mutation> expectedMutations = new ArrayList<>();
        List<Mutation> actualMutations = new ArrayList<>();
        // change the maxLookBack from infinite to some interval, which allows to simulate the mutation beyond the maxLookBack window.
        long maxLookbackInMills = 10 * 1000;
        rebuildScanner.setMaxLookBackInMills(maxLookbackInMills);

        ManualEnvironmentEdge injectEdge = new ManualEnvironmentEdge();
        injectEdge.setValue(1);
        EnvironmentEdgeManager.injectEdge(injectEdge);

        Put put = new Put(indexRowKey1Bytes);
        Cell cell = CellUtil.createCell(indexRowKey1Bytes,
                QueryConstants.DEFAULT_COLUMN_FAMILY_BYTES,
                QueryConstants.EMPTY_COLUMN_BYTES,
                EnvironmentEdgeManager.currentTimeMillis(),
                KeyValue.Type.Put.getCode(),
                VERIFIED_BYTES);
        put.add(cell);
        // This mutation is beyond maxLookBack, so add it to expectedMutations only.
        expectedMutations.add(put);

        injectEdge.incrementValue(maxLookbackInMills);
        put =  new Put(indexRowKey1Bytes);
        cell = CellUtil.createCell(indexRowKey1Bytes,
                QueryConstants.DEFAULT_COLUMN_FAMILY_BYTES,
                QueryConstants.EMPTY_COLUMN_BYTES,
                EnvironmentEdgeManager.currentTimeMillis(),
                KeyValue.Type.Put.getCode(),
                UNVERIFIED_BYTES);
        put.add(cell);
        // This mutation is actualMutations only, as it is an unverified put
        actualMutations.add(put);
        Result actualMutationsScanResult = Result.create(Arrays.asList(cell));

        Map<byte[], List<Mutation>> indexKeyToMutationMap = Maps.newTreeMap((Bytes.BYTES_COMPARATOR));
        indexKeyToMutationMap.put(indexRowKey1Bytes, expectedMutations);
        when(rebuildScanner.prepareActualIndexMutations(any(Result.class))).thenReturn(actualMutations);
        when(indexRow.getRow()).thenReturn(indexRowKey1Bytes);

        injectEdge.incrementValue(1);
        IndexToolVerificationResult.PhaseResult actualPR = new IndexToolVerificationResult.PhaseResult();
        // Report this validation as a failure
        assertFalse(rebuildScanner.verifySingleIndexRow(indexRow, indexKeyToMutationMap, mostRecentIndexRowKeys, actualPR, true));
        // beyondMaxLookBackInvalidIndexRowCount = 1
        IndexToolVerificationResult.PhaseResult expectedPR = new IndexToolVerificationResult.PhaseResult(0, 0, 0, 0, 0, 1, 0, 0);
        assertTrue(actualPR.equals(expectedPR));
    }

    private static byte[] generateIndexRowKey(String dataRowKey, String dataVal){
        List<Byte> idxKey = new ArrayList<>();
        if (dataVal != null && !dataVal.isEmpty())
            idxKey.addAll(com.google.common.primitives.Bytes.asList(Bytes.toBytes(dataVal)));
        idxKey.add(QueryConstants.SEPARATOR_BYTE);
        idxKey.addAll(com.google.common.primitives.Bytes.asList(Bytes.toBytes(dataRowKey)));
        return com.google.common.primitives.Bytes.toArray(idxKey);
    }

    private IndexToolVerificationResult.PhaseResult getValidPhaseResult() {
        return new IndexToolVerificationResult.PhaseResult(1, 0, 0, 0, 0, 0, 0, 0);
    }

    private IndexToolVerificationResult.PhaseResult getInvalidPhaseResult() {
        return new IndexToolVerificationResult.PhaseResult(0, 0, 0, 1, 0, 0, 0, 0);
    }

    private void initializeLocalMockitoSetup(Map.Entry<byte[], List<Mutation>> entry,
            TestType testType)
            throws IOException {
        actualPR = new IndexToolVerificationResult.PhaseResult();
        byte[] indexKey = entry.getKey();
        when(indexRow.getRow()).thenReturn(indexKey);
        actualMutationList = buildActualIndexMutationsList(testType);
        when(rebuildScanner.prepareActualIndexMutations(indexRow)).thenReturn(actualMutationList);
    }

    private List<Mutation> buildActualIndexMutationsList(TestType testType) {
        List<Mutation> actualMutations = new ArrayList<>();
        actualMutations.addAll(indexKeyToMutationMap.get(indexRow.getRow()));
        if(testType.equals(TestType.EXPIRED)) {
            return actualMutations;
        }
        if(testType.toString().startsWith("VALID")) {
            return getValidActualMutations(testType, actualMutations);
        }
        if(testType.toString().startsWith("INVALID")) {
            return getInvalidActualMutations(testType, actualMutations);
        }
        return null;
    }

    private List <Mutation> getValidActualMutations(TestType testType,
            List<Mutation> actualMutations) {
        List <Mutation> newActualMutations = new ArrayList<>();
        if(testType.equals(TestType.VALID_EXACT_MATCH)) {
            return actualMutations;
        }
        if (testType.equals(TestType.VALID_MIX_MUTATIONS)) {
            newActualMutations.add(getUnverifiedPutMutation(actualMutations.get(0), null));
            newActualMutations.add(getDeleteMutation(actualMutations.get(0), new Long(1)));
            newActualMutations.add(getUnverifiedPutMutation(actualMutations.get(0), null));
        }
        if (testType.equals(TestType.VALID_NEW_UNVERIFIED_MUTATIONS)) {
            newActualMutations.add(getUnverifiedPutMutation(actualMutations.get(0), null));
            newActualMutations.add(getUnverifiedPutMutation(actualMutations.get(0), null));
            newActualMutations.add(getUnverifiedPutMutation(actualMutations.get(0), null));
            newActualMutations.add(getUnverifiedPutMutation(actualMutations.get(0), new Long(1)));
        }
        newActualMutations.addAll(actualMutations);
        if(testType.equals(TestType.VALID_MORE_MUTATIONS)) {
            newActualMutations.add(getUnverifiedPutMutation(actualMutations.get(0), null));
            newActualMutations.add(getDeleteMutation(actualMutations.get(0), null));
            newActualMutations.add(getDeleteMutation(actualMutations.get(0), new Long(1)));
            newActualMutations.add(getUnverifiedPutMutation(actualMutations.get(0), new Long(1)));
        }
        if(testType.equals(TestType.VALID_EXTRA_CELL)) {
            for (Mutation m : newActualMutations) {
                if (m instanceof Put) {
                    List<Cell> origList = m.getFamilyCellMap().firstEntry().getValue();
                    Cell newCell =
                            CellUtil.createCell(m.getRow(), CellUtil.cloneFamily(origList.get(0)),
                                    Bytes.toBytes("EXTRACOL"), m.getTimeStamp(), KeyValue.Type.Put.getCode(),
                                    Bytes.toBytes("asdfg"));
                    byte[] fam = CellUtil.cloneFamily(origList.get(0));
                    m.getFamilyCellMap().get(fam).add(newCell);
                    break;
                }
            }
        }
        return newActualMutations;
    }

    private List <Mutation> getInvalidActualMutations(TestType testType,
            List<Mutation> actualMutations) {
        List <Mutation> newActualMutations = new ArrayList<>();
        newActualMutations.addAll(actualMutations);
        for (Mutation m : actualMutations) {
            newActualMutations.remove(m);
            NavigableMap<byte[], List<Cell>> familyCellMap = m.getFamilyCellMap();
            List<Cell> cellList = familyCellMap.firstEntry().getValue();
            List<Cell> newCellList = new ArrayList<>();
            byte[] fam = CellUtil.cloneFamily(cellList.get(0));
            for (Cell c : cellList) {
                infiltrateCell(c, newCellList, testType);
            }
            familyCellMap.put(fam, newCellList);
            m.setFamilyCellMap(familyCellMap);
            newActualMutations.add(m);
        }
        return newActualMutations;
    }

    private void infiltrateCell(Cell c, List<Cell> newCellList, TestType e) {
        Cell newCell;
        Cell emptyCell;
        switch(e) {
        case INVALID_COLUMN:
            newCell =
                    CellUtil.createCell(CellUtil.cloneRow(c), CellUtil.cloneFamily(c),
                            Bytes.toBytes(UNEXPECTED_COLUMN),
                            c.getTimestamp(),
                            KeyValue.Type.Put.getCode(), Bytes.toBytes("zxcv"));
            newCellList.add(newCell);
            newCellList.add(c);
            break;
        case INVALID_CELL_VALUE:
            if (CellUtil.matchingQualifier(c, EMPTY_COLUMN_BYTES)) {
                newCell = getCellWithPut(c);
                emptyCell = getVerifiedEmptyCell(c);
                newCellList.add(newCell);
                newCellList.add(emptyCell);
            } else {
                newCellList.add(c);
            }
            break;
        case INVALID_EMPTY_CELL:
            if (CellUtil.matchingQualifier(c, EMPTY_COLUMN_BYTES)) {
                newCell =
                        CellUtil.createCell(CellUtil.cloneRow(c), CellUtil.cloneFamily(c),
                                CellUtil.cloneQualifier(c), c.getTimestamp(),
                                KeyValue.Type.Delete.getCode(), VERIFIED_BYTES);
                newCellList.add(newCell);
            } else {
                newCellList.add(c);
            }
            break;
        case INVALID_EXTRA_CELL:
            newCell = getCellWithPut(c);
            emptyCell = getVerifiedEmptyCell(c);
            newCellList.add(newCell);
            newCellList.add(emptyCell);
            newCellList.add(c);
        }
    }

    private Cell getVerifiedEmptyCell(Cell c) {
        return CellUtil.createCell(CellUtil.cloneRow(c), CellUtil.cloneFamily(c),
                indexMaintainer.getEmptyKeyValueQualifier(),
                c.getTimestamp(),
                KeyValue.Type.Put.getCode(), VERIFIED_BYTES);
    }

    private Cell getCellWithPut(Cell c) {
        return CellUtil.createCell(CellUtil.cloneRow(c),
                CellUtil.cloneFamily(c), Bytes.toBytes(INCLUDED_COLUMN),
                c.getTimestamp(), KeyValue.Type.Put.getCode(),
                Bytes.toBytes("zxcv"));
    }

    private void expireThisRow() {
        rebuildScanner.setIndexTableTTL(INDEX_TABLE_EXPIRY_SEC);
        UnitTestClock expiryClock = new UnitTestClock(5000);
        EnvironmentEdgeManager.injectEdge(expiryClock);
    }

    private Mutation getDeleteMutation(Mutation orig, Long ts) {
        Mutation m = new Delete(orig.getRow());
        List<Cell> origList = orig.getFamilyCellMap().firstEntry().getValue();
        ts = ts == null ? EnvironmentEdgeManager.currentTimeMillis() : ts;
        Cell c = getNewPutCell(orig, origList, ts, KeyValue.Type.DeleteFamilyVersion);
        Cell empty = getEmptyCell(orig, origList, ts, KeyValue.Type.Put, true);
        byte[] fam = CellUtil.cloneFamily(origList.get(0));
        List<Cell> famCells = Lists.newArrayList();
        m.getFamilyCellMap().put(fam, famCells);
        famCells.add(c);
        famCells.add(empty);
        return m;
    }

    private Mutation getUnverifiedPutMutation(Mutation orig, Long ts) {
        Mutation m = new Put(orig.getRow());
        if (orig.getAttributesMap() != null) {
            for (Map.Entry<String,byte[]> entry : orig.getAttributesMap().entrySet()) {
                m.setAttribute(entry.getKey(), entry.getValue());
            }
        }
        List<Cell> origList = orig.getFamilyCellMap().firstEntry().getValue();
        ts = ts == null ? EnvironmentEdgeManager.currentTimeMillis() : ts;
        Cell c = getNewPutCell(orig, origList, ts, KeyValue.Type.Put);
        Cell empty = getEmptyCell(orig, origList, ts, KeyValue.Type.Put, false);
        byte[] fam = CellUtil.cloneFamily(origList.get(0));
        List<Cell> famCells = Lists.newArrayList();
        m.getFamilyCellMap().put(fam, famCells);
        famCells.add(c);
        famCells.add(empty);
        return m;
    }

    private Cell getEmptyCell(Mutation orig, List<Cell> origList, Long ts, KeyValue.Type type,
            boolean verified) {
        return CellUtil.createCell(orig.getRow(), CellUtil.cloneFamily(origList.get(0)),
                indexMaintainer.getEmptyKeyValueQualifier(),
                ts, type.getCode(), verified ? VERIFIED_BYTES : UNVERIFIED_BYTES);
    }

    private Cell getNewPutCell(Mutation orig, List<Cell> origList, Long ts, KeyValue.Type type) {
        return CellUtil.createCell(orig.getRow(),
                CellUtil.cloneFamily(origList.get(0)), Bytes.toBytes(INCLUDED_COLUMN),
                ts, type.getCode(), Bytes.toBytes("asdfg"));
    }
}