/*
 * 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.giraph.io.accumulo;

import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.mapreduce.AccumuloInputFormat;
import org.apache.accumulo.core.client.mapreduce.AccumuloOutputFormat;
import org.apache.accumulo.core.client.mock.MockInstance;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.util.ByteBufferUtil;
import org.apache.accumulo.core.util.Pair;
import org.apache.giraph.BspCase;
import org.apache.giraph.graph.BasicComputation;
import org.apache.giraph.conf.GiraphConfiguration;
import org.apache.giraph.io.accumulo.edgemarker.AccumuloEdgeInputFormat;
import org.apache.giraph.io.accumulo.edgemarker.AccumuloEdgeOutputFormat;
import org.apache.giraph.job.GiraphJob;
import org.apache.giraph.graph.Vertex;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.apache.log4j.Logger;
import org.junit.Test;

import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashSet;
import java.util.Map;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

/*
    Test class for Accumulo vertex input/output formats.
 */
public class TestAccumuloVertexFormat extends BspCase{

    private final String TABLE_NAME = "simple_graph";
    private final String INSTANCE_NAME = "instance";
    private final Text FAMILY = new Text("cf");
    private final Text CHILDREN = new Text("children");
    private final String USER = "root";
    private final byte[] PASSWORD = new byte[] {};
    private final Text OUTPUT_FIELD = new Text("parent");


    private final Logger log = Logger.getLogger(TestAccumuloVertexFormat.class);

    /**
     * Create the test case
     */
    public TestAccumuloVertexFormat() {
        super(TestAccumuloVertexFormat.class.getName());
    }

    /*
     Write a simple parent-child directed graph to Accumulo.
     Run a job which reads the values
     into subclasses that extend AccumuloVertex I/O formats.
     Check the output after the job.
     */
    @Test
    public void testAccumuloInputOutput() throws Exception {
        if (System.getProperty("prop.mapred.job.tracker") != null) {
            if(log.isInfoEnabled())
                log.info("testAccumuloInputOutput: " +
                        "Ignore this test if not local mode.");
            return;
        }

        File jarTest = new File(System.getProperty("prop.jarLocation"));
        if(!jarTest.exists()) {
            fail("Could not find Giraph jar at " +
                    "location specified by 'prop.jarLocation'. " +
                    "Make sure you built the main Giraph artifact?.");
        }

        //Write out vertices and edges out to a mock instance.
        MockInstance mockInstance = new MockInstance(INSTANCE_NAME);
        Connector c = mockInstance.getConnector("root", new byte[] {});
        c.tableOperations().create(TABLE_NAME);
        BatchWriter bw = c.createBatchWriter(TABLE_NAME, 10000L, 1000L, 4);

        Mutation m1 = new Mutation(new Text("0001"));
        m1.put(FAMILY, CHILDREN, new Value("0002".getBytes()));
        bw.addMutation(m1);

        Mutation m2 = new Mutation(new Text("0002"));
        m2.put(FAMILY, CHILDREN, new Value("0003".getBytes()));
        bw.addMutation(m2);
        if(log.isInfoEnabled())
            log.info("Writing mutations to Accumulo table");
        bw.close();

        Configuration conf = new Configuration();
        conf.set(AccumuloVertexOutputFormat.OUTPUT_TABLE, TABLE_NAME);

        /*
        Very important to initialize the formats before
        sending configuration to the GiraphJob. Otherwise
        the internally constructed Job in GiraphJob will
        not have the proper context initialization.
         */
        AccumuloInputFormat.setInputInfo(conf, USER, "".getBytes(),
                TABLE_NAME, new Authorizations());
        AccumuloInputFormat.setMockInstance(conf, INSTANCE_NAME);

        AccumuloOutputFormat.setOutputInfo(conf, USER, PASSWORD, true, null);
        AccumuloOutputFormat.setMockInstance(conf, INSTANCE_NAME);

        GiraphJob job = new GiraphJob(conf, getCallingMethodName());
        setupConfiguration(job);
        GiraphConfiguration giraphConf = job.getConfiguration();
        giraphConf.setComputationClass(EdgeNotification.class);
        giraphConf.setVertexInputFormatClass(AccumuloEdgeInputFormat.class);
        giraphConf.setVertexOutputFormatClass(AccumuloEdgeOutputFormat.class);

        HashSet<Pair<Text, Text>> columnsToFetch = new HashSet<Pair<Text,Text>>();
        columnsToFetch.add(new Pair<Text, Text>(FAMILY, CHILDREN));
        AccumuloInputFormat.fetchColumns(job.getConfiguration(), columnsToFetch);

        if(log.isInfoEnabled())
            log.info("Running edge notification job using Accumulo input");
        assertTrue(job.run(true));
        Scanner scanner = c.createScanner(TABLE_NAME, new Authorizations());
        scanner.setRange(new Range("0002", "0002"));
        scanner.fetchColumn(FAMILY, OUTPUT_FIELD);
        boolean foundColumn = false;

        if(log.isInfoEnabled())
            log.info("Verify job output persisted correctly.");
        //make sure we found the qualifier.
        assertTrue(scanner.iterator().hasNext());


        //now we check to make sure the expected value from the job persisted correctly.
        for(Map.Entry<Key,Value> entry : scanner) {
            Text row = entry.getKey().getRow();
            assertEquals("0002", row.toString());
            Value value = entry.getValue();
            assertEquals("0001", ByteBufferUtil.toString(
                    ByteBuffer.wrap(value.get())));
            foundColumn = true;
        }
    }
    /*
    Test compute method that sends each edge a notification of its parents.
    The test set only has a 1-1 parent-to-child ratio for this unit test.
     */
    public static class EdgeNotification
            extends BasicComputation<Text, Text, Text, Text> {
      @Override
      public void compute(Vertex<Text, Text, Text> vertex,
          Iterable<Text> messages) throws IOException {
          for (Text message : messages) {
            vertex.getValue().set(message);
          }
          if(getSuperstep() == 0) {
            sendMessageToAllEdges(vertex, vertex.getId());
          }
        vertex.voteToHalt();
      }
    }
}