package org.apache.rya.joinselect.mr;

/*
 * 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.
 */


import static org.apache.rya.joinselect.mr.utils.JoinSelectConstants.INPUTPATH;
import static org.apache.rya.joinselect.mr.utils.JoinSelectConstants.INSTANCE;
import static org.apache.rya.joinselect.mr.utils.JoinSelectConstants.OUTPUTPATH;
import static org.apache.rya.joinselect.mr.utils.JoinSelectConstants.PASSWORD;
import static org.apache.rya.joinselect.mr.utils.JoinSelectConstants.PROSPECTS_OUTPUTPATH;
import static org.apache.rya.joinselect.mr.utils.JoinSelectConstants.PROSPECTS_TABLE;
import static org.apache.rya.joinselect.mr.utils.JoinSelectConstants.SELECTIVITY_TABLE;
import static org.apache.rya.joinselect.mr.utils.JoinSelectConstants.SPO_OUTPUTPATH;
import static org.apache.rya.joinselect.mr.utils.JoinSelectConstants.SPO_TABLE;
import static org.apache.rya.joinselect.mr.utils.JoinSelectConstants.USERNAME;

import java.io.File;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

import org.apache.rya.accumulo.AccumuloRdfConfiguration;
import org.apache.rya.api.RdfCloudTripleStoreConstants.TABLE_LAYOUT;
import org.apache.rya.api.domain.RyaStatement;
import org.apache.rya.api.domain.RyaType;
import org.apache.rya.api.domain.RyaIRI;
import org.apache.rya.api.resolver.RyaTripleContext;
import org.apache.rya.api.resolver.triple.TripleRow;
import org.apache.rya.joinselect.mr.JoinSelectAggregate.JoinReducer;
import org.apache.rya.joinselect.mr.JoinSelectAggregate.JoinSelectAggregateMapper;
import org.apache.rya.joinselect.mr.JoinSelectAggregate.JoinSelectGroupComparator;
import org.apache.rya.joinselect.mr.JoinSelectAggregate.JoinSelectPartitioner;
import org.apache.rya.joinselect.mr.JoinSelectAggregate.JoinSelectSortComparator;
import org.apache.rya.joinselect.mr.JoinSelectProspectOutput.CardinalityMapper;
import org.apache.rya.joinselect.mr.JoinSelectSpoTableOutput.JoinSelectMapper;
import org.apache.rya.joinselect.mr.JoinSelectStatisticsSum.CardinalityIdentityCombiner;
import org.apache.rya.joinselect.mr.JoinSelectStatisticsSum.CardinalityIdentityMapper;
import org.apache.rya.joinselect.mr.JoinSelectStatisticsSum.CardinalityIdentityReducer;
import org.apache.rya.joinselect.mr.utils.CardList;
import org.apache.rya.joinselect.mr.utils.CompositeType;
import org.apache.rya.joinselect.mr.utils.JoinSelectStatsUtil;
import org.apache.rya.joinselect.mr.utils.TripleCard;
import org.apache.rya.joinselect.mr.utils.TripleEntry;

import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.TableExistsException;
import org.apache.accumulo.core.client.TableNotFoundException;
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.client.security.tokens.PasswordToken;
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.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

public class JoinSelectStatisticsTest {
    
    private static final String PREFIX = JoinSelectStatisticsTest.class.getSimpleName();
  
    private static final String DELIM = "\u0000";
    private static final String iri = "uri:";
    private List<String> cardList = Arrays.asList("subject", "predicate", "object");
    private List<String> aggCardList = Arrays.asList("subjectobject", "subjectpredicate", "subjectsubject", "predicateobject", "predicatepredicate", "predicatesubject");
    private static File SPOOUT;
    private static File PROSPECTSOUT;
    private static File tempDir;
    private Connector c;
    private RyaTripleContext ryaContext;
    private static final String INSTANCE_NAME = "mapreduce_instance";
    
    private static class JoinSelectTester1 extends Configured implements Tool {

       
        @Override
        public int run(String[] args) throws Exception {

            Configuration conf = getConf();
            
            String inTable = conf.get(SPO_TABLE);
            String outPath = conf.get(SPO_OUTPUTPATH);
           
            
            assert inTable != null && outPath != null;

            Job job = new Job(conf, this.getClass().getSimpleName() + "_" + System.currentTimeMillis());
            job.setJarByClass(this.getClass());
            conf.setBoolean(MRJobConfig.MAPREDUCE_JOB_USER_CLASSPATH_FIRST, true);
            
            initTabToSeqFileJob(job, inTable, outPath);
            job.setMapperClass(JoinSelectMapper.class);
            job.setNumReduceTasks(0);
            job.waitForCompletion(true);
           
            return job.isSuccessful() ? 0 : 1;
        }
    }

    private static class JoinSelectTester2 extends Configured implements Tool {

        
        @Override
        public int run(String[] args) throws Exception {

            Configuration conf = getConf();
            
            String inTable = conf.get(PROSPECTS_TABLE);
            System.out.println("Table is " + inTable);
            String outPath = conf.get(PROSPECTS_OUTPUTPATH);
           
            
            assert inTable != null && outPath != null;

            Job job = new Job(conf, this.getClass().getSimpleName() + "_" + System.currentTimeMillis());
            job.setJarByClass(this.getClass());
            conf.setBoolean(MRJobConfig.MAPREDUCE_JOB_USER_CLASSPATH_FIRST, true);
            
            initTabToSeqFileJob(job, inTable, outPath);
            job.setMapperClass(CardinalityMapper.class);
            job.setNumReduceTasks(0);
            job.waitForCompletion(true);
           
            return job.isSuccessful() ? 0 : 1;          
        }
    }
    
    
 private static class JoinSelectTester4 extends Configured implements Tool {

        
        @Override
        public int run(String[] args) throws Exception {

            Configuration conf = getConf();
            String outpath = conf.get(OUTPUTPATH);
    
            Job job = new Job(conf, this.getClass().getSimpleName() + "_" + System.currentTimeMillis());
            job.setJarByClass(this.getClass());
            conf.setBoolean(MRJobConfig.MAPREDUCE_JOB_USER_CLASSPATH_FIRST, true);
            
            MultipleInputs.addInputPath(job, new Path(PROSPECTSOUT.getAbsolutePath()), 
                    SequenceFileInputFormat.class, JoinSelectAggregateMapper.class);
            MultipleInputs.addInputPath(job,new Path(SPOOUT.getAbsolutePath()) , 
                    SequenceFileInputFormat.class, JoinSelectAggregateMapper.class);
            job.setMapOutputKeyClass(CompositeType.class);
            job.setMapOutputValueClass(TripleCard.class);

            tempDir = new File(File.createTempFile(outpath, "txt").getParentFile(), System.currentTimeMillis() + "");
            SequenceFileOutputFormat.setOutputPath(job, new Path(tempDir.getAbsolutePath()));
            job.setOutputFormatClass(SequenceFileOutputFormat.class);
            job.setOutputKeyClass(TripleEntry.class);
            job.setOutputValueClass(CardList.class);


            job.setSortComparatorClass(JoinSelectSortComparator.class);
            job.setGroupingComparatorClass(JoinSelectGroupComparator.class);
            job.setPartitionerClass(JoinSelectPartitioner.class);
            job.setReducerClass(JoinReducer.class);
            job.setNumReduceTasks(32);
            job.waitForCompletion(true);
            
            return job.isSuccessful() ? 0 : 1;          
        }
     }
    
    
    private static class JoinSelectTester3 extends Configured implements Tool {

        
        @Override
        public int run(String[] args) throws Exception {
            
            Configuration conf = getConfig();
            
            String outTable = conf.get(SELECTIVITY_TABLE);
            String inPath = conf.get(INPUTPATH);
            
             
            assert outTable != null && inPath != null;

            Job job = new Job(getConf(), this.getClass().getSimpleName() + "_" + System.currentTimeMillis());
            job.setJarByClass(this.getClass());
            initSumMRJob(job, inPath, outTable);

            job.setMapperClass(CardinalityIdentityMapper.class);
            job.setCombinerClass(CardinalityIdentityCombiner.class);
            job.setReducerClass(CardinalityIdentityReducer.class);
            job.setNumReduceTasks(32);
            job.waitForCompletion(true);
            
           return job.isSuccessful() ? 0 : 1;
           
        }
        
       
    }
    
   
    public class JoinSelectTestDriver extends Configured implements Tool {
        
        Configuration conf = getConfig();

        @Override
        public int run(String[] args) throws Exception {
            
            int res0 = ToolRunner.run(conf, new JoinSelectTester1(), args);
            int res1 = 1;
            int res2 = 1;
            int res3 = 1;
            
            
            if(res0 == 0) {
                res1 = ToolRunner.run(conf, new JoinSelectTester2(), args);
            }
            if(res1 == 0) {
                res2 = ToolRunner.run(conf, new JoinSelectTester4(), args);
            }
            if(res2 == 0) {
                res3 = ToolRunner.run(conf, new JoinSelectTester3(), args);
            }
            
            return res3;
        }

    }
    
    
    private static Configuration getConfig() {
        
        Configuration conf = new Configuration();
        conf.set("fs.default.name", "file:///");
        conf.set("mapreduce.framework.name", "local");
        conf.set("spo.table", "rya_spo");
        conf.set("prospects.table", "rya_prospects");
        conf.set("selectivity.table", "rya_selectivity");
        conf.set("auths", "");
        conf.set("instance",INSTANCE_NAME);
        conf.set("username","root");
        conf.set("password", "");
        conf.set("inputpath","temp");
        conf.set("outputpath","temp");
        conf.set("prospects.outputpath","prospects");
        conf.set("spo.outputpath", "spo");
        
        
        return conf;
        
    }
    
    
    public static void initTabToSeqFileJob(Job job, String intable, String outpath) throws AccumuloSecurityException, IOException {
        
        Configuration conf = job.getConfiguration();
       
        String username = conf.get(USERNAME);
        System.out.println("Username is " + username);
        String password = conf.get(PASSWORD);
        String instance = conf.get(INSTANCE);
        System.out.println("Instance is " + instance);
        
       
        AccumuloInputFormat.setMockInstance(job, instance);
        AccumuloInputFormat.setConnectorInfo(job, username, new PasswordToken(password));
        AccumuloInputFormat.setInputTableName(job, intable);
        
        job.setInputFormatClass(AccumuloInputFormat.class);
        job.setMapOutputKeyClass(CompositeType.class);
        job.setMapOutputValueClass(TripleCard.class);

        System.out.println("Outpath is " + outpath);
        
        // OUTPUT
        if(outpath.equals("spo")) {
            SPOOUT = new File(File.createTempFile(outpath, "txt").getParentFile(), System.currentTimeMillis() + "spo");
            SequenceFileOutputFormat.setOutputPath(job, new Path(SPOOUT.getAbsolutePath()));
        } else {
            PROSPECTSOUT = new File(File.createTempFile(outpath, "txt").getParentFile(), System.currentTimeMillis() + "prospects");
            SequenceFileOutputFormat.setOutputPath(job, new Path(PROSPECTSOUT.getAbsolutePath()));
        }
        job.setOutputFormatClass(SequenceFileOutputFormat.class);
        job.setOutputKeyClass(CompositeType.class);
        job.setOutputValueClass(TripleCard.class);
    
     }
    
    public static void initSumMRJob(Job job, String inputPath, String outtable) throws AccumuloSecurityException, IOException {
      
        Configuration conf = job.getConfiguration();
        
        String username = conf.get(USERNAME);
        String password = conf.get(PASSWORD);
        String instance = conf.get(INSTANCE);
       
        
        AccumuloOutputFormat.setConnectorInfo(job, username, new PasswordToken(password));
        AccumuloOutputFormat.setMockInstance(job, instance);
        AccumuloOutputFormat.setDefaultTableName(job, outtable);

        
        SequenceFileInputFormat.addInputPath(job, new Path(tempDir.getAbsolutePath()));
        job.setInputFormatClass(SequenceFileInputFormat.class);
        job.setMapOutputKeyClass(TripleEntry.class);
        job.setMapOutputValueClass(CardList.class);

       
        job.setOutputFormatClass(AccumuloOutputFormat.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Mutation.class);
         
      
    }
    
    
    @Before
    public void init() throws AccumuloException, AccumuloSecurityException, TableExistsException, TableNotFoundException {
      
        MockInstance mockInstance = new MockInstance(INSTANCE_NAME);
        c = mockInstance.getConnector("root", new PasswordToken(""));
        
        if (c.tableOperations().exists("rya_prospects")) {
            c.tableOperations().delete("rya_prospects");
        } 
        if (c.tableOperations().exists("rya_selectivity")) {
            c.tableOperations().delete("rya_selectivity");
        }
        if (c.tableOperations().exists("rya_spo")) {
            c.tableOperations().delete("rya_spo");
        } 
        
        
        c.tableOperations().create("rya_spo");
        c.tableOperations().create("rya_prospects");
        c.tableOperations().create("rya_selectivity");
        ryaContext = RyaTripleContext.getInstance(new AccumuloRdfConfiguration(getConfig()));
    }
    
    
    @Test
    public void testMap1() throws Exception {
        init();
        
        System.out.println("*****************************Test1**************************** ");
        
        BatchWriter bw_table1 = c.createBatchWriter("rya_spo", new BatchWriterConfig());
        for (int i = 1; i < 3; i++) {

            RyaStatement rs = new RyaStatement(new RyaIRI(iri + i), new RyaIRI(iri + 5), new RyaType(iri + (i + 2)));
            Map<TABLE_LAYOUT, TripleRow> tripleRowMap = ryaContext.serializeTriple(rs);
            TripleRow tripleRow = tripleRowMap.get(TABLE_LAYOUT.SPO);
            Mutation m = JoinSelectStatsUtil.createMutation(tripleRow);
            bw_table1.addMutation(m);
            

        }
        bw_table1.close();

        BatchWriter bw_table2 = c.createBatchWriter("rya_prospects", new BatchWriterConfig());
        for (int i = 1; i < 6; i++) {

            int j = 1;
            
            for (String s : cardList) {
                Mutation m = new Mutation(new Text(s + DELIM + iri + i + DELIM + i));
                m.put(new Text(), new Text(), new Value(new IntWritable(i + j).toString().getBytes()));
                bw_table2.addMutation(m);
                j++;
            }

        }
        bw_table2.close();
        
        
        Assert.assertEquals(0, ToolRunner.run(new JoinSelectTestDriver(), new String[]{""}));     
        Scanner scan = c.createScanner("rya_selectivity", new Authorizations());
        scan.setRange(new Range());

        for (Map.Entry<Key, Value> entry : scan) {
            System.out.println("Key row string is " + entry.getKey().getRow().toString());
            System.out.println("Join type is " + entry.getKey().getColumnFamily().toString());
            System.out.println("Value is " + entry.getKey().getColumnQualifier().toString());
        }

        Scanner scan1 = c.createScanner("rya_selectivity" , new Authorizations());
        scan1.setRange(Range.prefix("predicate" +DELIM + iri + 5));
        int i = 5;
        
        for (Map.Entry<Key, Value> entry : scan1) {
            
            int val1 = 5 + 2*i;
            int val2 = 5 + 2*(i-1);
            int val = Integer.parseInt(entry.getKey().getColumnQualifier().toString());
            
            if(i < 3) {
                Assert.assertTrue( val == val1);
            }
            if(i >= 3  && i < 6) {
                Assert.assertTrue(val == val2);
            }
            i--;
        }
        Assert.assertTrue(i == -1);
        
        
        Scanner scan2 = c.createScanner("rya_selectivity" , new Authorizations());
        scan2.setRange(Range.prefix("object" +DELIM + iri + 3));
        int j = 5;
        
        for (Map.Entry<Key, Value> entry : scan2) {
            
            int val1 = 5 + (j-2);
            int val2 = 2+j;
            int val = Integer.parseInt(entry.getKey().getColumnQualifier().toString());
            
            if(j < 3) {
                Assert.assertTrue( val == val2);
            }
            if(j >= 3  && j < 6) {
                Assert.assertTrue(val == val1);
            }
            j--;
        }
        Assert.assertTrue(j == -1);
        
        
        Scanner scan3 = c.createScanner("rya_selectivity", new Authorizations());
        scan3.setRange(Range.prefix("objectsubject" + DELIM + iri + 3 +DELIM +iri +1 ));
        int k = 8;

        for (Map.Entry<Key, Value> entry : scan3) {

            int val = Integer.parseInt(entry.getKey().getColumnQualifier().toString());

            Assert.assertTrue(val == k);
            k--;
        }
        Assert.assertTrue(k == 5);
        
        
    }


    @Test
    public void testMap2() throws Exception {
        
        System.out.println("*********************Test2******************* ");

        init();

        BatchWriter bw_table1 = c.createBatchWriter("rya_spo", new BatchWriterConfig());
        for (int i = 1; i < 4; i++) {

            RyaStatement rs = new RyaStatement(new RyaIRI(iri + 1), new RyaIRI(iri + 2), new RyaType(iri + i));
            Map<TABLE_LAYOUT, TripleRow> tripleRowMap = ryaContext.serializeTriple(rs);
            TripleRow tripleRow = tripleRowMap.get(TABLE_LAYOUT.SPO);
            Mutation m = JoinSelectStatsUtil.createMutation(tripleRow);
            bw_table1.addMutation(m);

        }
        bw_table1.close();

        BatchWriter bw_table2 = c.createBatchWriter("rya_prospects", new BatchWriterConfig());
        for (int i = 1; i < 4; i++) {

            for (String s : cardList) {
                Mutation m = new Mutation(new Text(s + DELIM + iri + i + DELIM + i));
                m.put(new Text(), new Text(), new Value(new IntWritable(i + 2).toString().getBytes()));
                bw_table2.addMutation(m);
            }

        }
        bw_table2.close();

        Assert.assertEquals(0, ToolRunner.run(new JoinSelectTestDriver(), new String[]{""}));
        Scanner scan1 = c.createScanner("rya_selectivity" , new Authorizations());
        scan1.setRange(Range.prefix("subject" +DELIM + iri + 1));
        int i = 0;
        
        for (Map.Entry<Key, Value> entry : scan1) {
            
            Assert.assertTrue(entry.getKey().getColumnQualifier().toString().equals("12")); 
            i++;
        }
        Assert.assertTrue(i == 6);
        
        Scanner scan2 = c.createScanner("rya_selectivity" , new Authorizations());
        scan2.setRange(Range.prefix("predicate" +DELIM + iri + 2));
        int j = 0;
        
        for (Map.Entry<Key, Value> entry : scan2) {
            
            if(j < 3) {
                Assert.assertTrue(entry.getKey().getColumnQualifier().toString().equals("12"));
            }
            if(j > 3  && j < 6) {
                Assert.assertTrue(entry.getKey().getColumnQualifier().toString().equals("9"));
            }
            j++;
        }
        Assert.assertTrue(j == 6);
        
        Scanner scan3 = c.createScanner("rya_selectivity" , new Authorizations());
        scan3.setRange(Range.prefix("predicateobject" +DELIM + iri + 2 +DELIM + iri + 2));
        int k = 0;
        
        for (Map.Entry<Key, Value> entry : scan3) {
            Assert.assertTrue(entry.getKey().getColumnQualifier().toString().equals("3")); 
            k++;
        }
        Assert.assertTrue(k == 3);
        

    }

    
    @Test
    public void testMap3() throws Exception {
        init();
        
        System.out.println("*************************Test3**************************** ");
        
        BatchWriter bw_table1 = c.createBatchWriter("rya_spo", new BatchWriterConfig());
        for (int i = 1; i < 3; i++) {
            for (int j = 1; j < 3; j++) {
                for (int k = 1; k < 3; k++) {

                    RyaStatement rs = new RyaStatement(new RyaIRI(iri + i), new RyaIRI(iri + (j)), new RyaType(iri + k));
                    Map<TABLE_LAYOUT, TripleRow> tripleRowMap = ryaContext.serializeTriple(rs);
                    TripleRow tripleRow = tripleRowMap.get(TABLE_LAYOUT.SPO);
                    Mutation m = JoinSelectStatsUtil.createMutation(tripleRow);
                    bw_table1.addMutation(m);

                }
            }

        }
        bw_table1.close();

        BatchWriter bw_table2 = c.createBatchWriter("rya_prospects", new BatchWriterConfig());
        for (int i = 1; i < 3; i++) {

            int k = 1;
            for (String s : cardList) {
                Mutation m = new Mutation(new Text(s + DELIM + iri + i + DELIM + i));
                m.put(new Text(), new Text(), new Value(new IntWritable(i + k).toString().getBytes()));
                bw_table2.addMutation(m);
                k++;
            }

            for (int j = 1; j < 3; j++) {
                k = 1;
                for (String s : aggCardList) {
                    Mutation m = new Mutation(new Text(s + DELIM + iri + i + DELIM + iri + j + DELIM + i));
                    m.put(new Text(), new Text(), new Value(new IntWritable(i + k +j).toString().getBytes()));
                    bw_table2.addMutation(m);
                    k++;
                }
            }

        }
        bw_table2.close();
        
        
        Assert.assertEquals(0, ToolRunner.run(new JoinSelectTestDriver(), new String[]{""}));
        Scanner scan = c.createScanner("rya_selectivity", new Authorizations());
        scan.setRange(new Range());

        for (Map.Entry<Key, Value> entry : scan) {
            System.out.println("Key row string is " + entry.getKey().getRow().toString());
            System.out.println("Join type is " + entry.getKey().getColumnFamily().toString());
            System.out.println("Value is " + entry.getKey().getColumnQualifier().toString());
        }

        
        Scanner scan1 = c.createScanner("rya_selectivity" , new Authorizations());
        scan1.setRange(Range.prefix("subject" +DELIM + iri + 1));
        int i = 0;
        
        for (Map.Entry<Key, Value> entry : scan1) {
            
            Key key = entry.getKey();
            String s = key.getColumnFamily().toString();
            int val = Integer.parseInt(key.getColumnQualifier().toString());
            
            if(s.equals("predicatepredicate")) {
                Assert.assertTrue(val == 14);
            } 
            if(s.equals("objectobject")) {
                Assert.assertTrue(val == 18);
            } 
            if(s.equals("predicateobjectpredicateobject")) {
                Assert.assertTrue(val == 28);
            } 
            if(s.equals("predicateobjectsubjectpredicate")) {
                Assert.assertTrue(val == 20);
            }
            if(s.equals("predicateobjectobjectsubject")) {
                Assert.assertTrue(val == 16);
            }
            
            i++;
        }
        Assert.assertTrue(i == 12);
        
        
    }


    @Test
    public void testMap4() throws Exception {
        init();
        
        System.out.println("*************************Test4**************************** ");
        System.out.println("*************************Test4**************************** ");
        
        BatchWriter bw_table1 = c.createBatchWriter("rya_spo", new BatchWriterConfig());
        for (int i = 1; i < 3; i++) {
            for (int j = 1; j < 3; j++) {
                for (int k = 1; k < 3; k++) {

                    if(j == 1 && k ==2) {
                        break;
                    }
                    
                    RyaStatement rs = new RyaStatement(new RyaIRI(iri + i), new RyaIRI(iri + (j)), new RyaType(iri + k));
                    Map<TABLE_LAYOUT, TripleRow> tripleRowMap = ryaContext.serializeTriple(rs);
                    TripleRow tripleRow = tripleRowMap.get(TABLE_LAYOUT.SPO);
                    Mutation m = JoinSelectStatsUtil.createMutation(tripleRow);
                    bw_table1.addMutation(m);

                }
            }

        }
        bw_table1.close();

        BatchWriter bw_table2 = c.createBatchWriter("rya_prospects", new BatchWriterConfig());
        for (int i = 1; i < 3; i++) {

            int k = 1;
            for (String s : cardList) {
                Mutation m = new Mutation(new Text(s + DELIM + iri + i + DELIM + i));
                m.put(new Text(), new Text(), new Value(new IntWritable(i + k).toString().getBytes()));
                bw_table2.addMutation(m);
                k++;
            }

            for (int j = 1; j < 3; j++) {
                k = 1;
                for (String s : aggCardList) {
                    Mutation m = new Mutation(new Text(s + DELIM + iri + i + DELIM + iri + j + DELIM + i));
                    m.put(new Text(), new Text(), new Value(new IntWritable(i + k + 2*j).toString().getBytes()));
                    bw_table2.addMutation(m);
                    k++;
                }
            }

        }
        bw_table2.close();
        
        
        Assert.assertEquals(0, ToolRunner.run(new JoinSelectTestDriver(), new String[]{""}));
        Scanner scan = c.createScanner("rya_selectivity", new Authorizations());
        scan.setRange(new Range());

        for (Map.Entry<Key, Value> entry : scan) {
            System.out.println("Key row string is " + entry.getKey().getRow().toString());
            System.out.println("Join type is " + entry.getKey().getColumnFamily().toString());
            System.out.println("Value is " + entry.getKey().getColumnQualifier().toString());
        }

        
        Scanner scan1 = c.createScanner("rya_selectivity" , new Authorizations());
        scan1.setRange(Range.prefix("subject" +DELIM + iri + 1));
        int i = 0;
        
        for (Map.Entry<Key, Value> entry : scan1) {
            
            Key key = entry.getKey();
            String s = key.getColumnFamily().toString();
            int val = Integer.parseInt(key.getColumnQualifier().toString());
            
            if(s.equals("predicatepredicate")) {
                Assert.assertTrue(val == 11);
            } 
            if(s.equals("objectobject")) {
                Assert.assertTrue(val == 13);
            } 
            if(s.equals("predicateobjectobjectpredicate")) {
                Assert.assertTrue(val == 26);
            } 
            if(s.equals("predicateobjectpredicateobject")) {
                Assert.assertTrue(val == 25);
            } 
            if(s.equals("predicateobjectsubjectpredicate")) {
                Assert.assertTrue(val == 19);
            }
            if(s.equals("predicateobjectpredicatesubject")) {
                Assert.assertTrue(val == 20);
            }
            
            i++;
        }
        Assert.assertTrue(i == 12);
        
        
        Scanner scan2 = c.createScanner("rya_selectivity" , new Authorizations());
        scan2.setRange(Range.prefix("predicate" +DELIM + iri + 1));
        int j = 0;
        
        for (Map.Entry<Key, Value> entry : scan2) {
            
            Key key = entry.getKey();
            String s = key.getColumnFamily().toString();
            int val = Integer.parseInt(key.getColumnQualifier().toString());
            
            if(s.equals("subjectsubject")) {
                Assert.assertTrue(val == 5);
            } 
            if(s.equals("objectobject")) {
                Assert.assertTrue(val == 8);
            } 
            if(s.equals("objectsubjectsubjectpredicate")) {
                Assert.assertTrue(val == 11);
            } 
            if(s.equals("objectsubjectpredicateobject")) {
                Assert.assertTrue(val == 15);
            } 
            if(s.equals("objectsubjectobjectsubject")) {
                Assert.assertTrue(val == 9);
            }
            if(s.equals("objectsubjectsubjectobject")) {
                Assert.assertTrue(val == 10);
            }
            
            j++;
        }
        Assert.assertTrue(j == 12);
        
        
    }

    
}