/*
 * Copyright (c) 2015 Pradeeban Kathiravelu and others. All rights reserved.
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
 * and is available at http://www.eclipse.org/legal/epl-v10.html
 */
package edu.emory.bmi.datacafe.mongo;

import edu.emory.bmi.datacafe.constants.MongoConstants;
import edu.emory.bmi.datacafe.core.AbstractExecutorEngine;
import edu.emory.bmi.datacafe.core.DataSource;
import edu.emory.bmi.datacafe.core.DataSourceBean;
import edu.emory.bmi.datacafe.core.DataSourceWrapper;
import edu.emory.bmi.datacafe.hdfs.HdfsConnector;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.jongo.MongoCursor;

import java.util.*;

/**
 * The core Data Cafe Executor Engine.
 */
public class ExecutorEngine extends AbstractExecutorEngine {
    private static Logger logger = LogManager.getLogger(ExecutorEngine.class.getName());
    private static Map<Class<? extends DataSourceBean>, MongoCursor> mongoCursorMap;

    private static Map<Class<? extends DataSourceBean>, QueryConstructor> queryConstructorMap;

    private static List[] dataSourceBeans;

    public ExecutorEngine(Map<Class<? extends DataSourceBean>, DataSource> dataSources) {
        super(dataSources);
        mongoCursorMap = new HashMap<>();
        initializeDatasourceNames();
    }

    private void initializeDatasourceNames() {
        MongoEngine dataSourceEngine = new MongoEngine();
        for (Class<? extends DataSourceBean> dataSourceBean : dataSourceMap.keySet()) {
            DataSource dataSource = dataSourceMap.get(dataSourceBean);
            dataSourceEngine.addDataSource(dataSource);
            datasourceNames.put(dataSourceBean, dataSource.getFullName());
        }
        if (logger.isDebugEnabled()) {
            logger.debug("Data source names are initialized");
        }
    }

    /**
     * Retrieve pointers to the interested data source IDs. Map: DataSourceBean -> idConstraints
     *
     * @param idConstraintsMap Queries for data source ids.
     */
    public void retrieveDataLakeIDs(Map<Class<? extends DataSourceBean>, String> idConstraintsMap) {
        // Get the IDs
        for (Class<? extends DataSourceBean> clazz : idConstraintsMap.keySet()) {
            MongoCursor tempCursors = MongoEngine.initializeEntry(
                    dataSourceMap.get(clazz).getDatabase(), dataSourceMap.get(clazz).getCollection(),
                    idConstraintsMap.get(clazz), clazz);
            mongoCursorMap.put(clazz, tempCursors);
        }
        dataSourceBeans = new List[mongoCursorMap.size()];
    }

    public void createList(Class clazz, DataSourceWrapper<? extends DataSourceBean> dataSourceWrapper, QueryConstructor queryConstructor) {
        List retrievedDataList = new ArrayList<>();


        for (Object obj_ : mongoCursorMap.get(clazz)) {
            DataSourceBean minimalObj = (DataSourceBean) obj_;
            DataSourceBean tempObject = (DataSourceBean) MongoEngine.findEntry(queryConstructor.getDatabase(),
                    queryConstructor.getCollection(), "{" + MongoConstants.ID_ATTRIBUTE + ":'" +
                            minimalObj.getKey() + "'}, " + queryConstructor.getConstraints(), clazz
            );
            retrievedDataList.add(tempObject);
        }
        int i = dataSourceWrapper.getAndIncrement();
        dataSourceBeans[i] = retrievedDataList;
    }


    /**
     * Construct the data lake
     */
    public void constructDataLake(String[][] interestedAttributes) {
        queryConstructorMap = new HashMap<>();

        String[] queries = MongoConnector.constructQueries(interestedAttributes);

        Collection<String> values = datasourceNames.values();
        String[] tempDataSourceNames = values.toArray(new String[values.size()]);

        HdfsConnector.writeDataSourcesToWarehouse(tempDataSourceNames, interestedAttributes, queries, dataSourceBeans);
    }
}

@@ -1100,126 +1100,66 @@
 tic 
-Map%3CClass%3C? extends DataSourceBean%3E, QueryConstructor%3E queryConstructorMap;%0A%0A    private static List%5B%5D dataSourceBeans
+List%5B%5D dataSourceBeans;%0A    private static int autoInc = 0
 ;%0A%0A 
@@ -2621,24 +2621,225 @@
 ()%5D;%0A    %7D%0A%0A
+    /**%0A     * Called once per each of the data sources%0A     * @param clazz the class param%0A     * @param dataSourceWrapper data source wrapper%0A     * @param queryConstructor query constructor%0A     */%0A
     public v
@@ -3019,17 +3019,16 @@
 t%3C%3E();%0A%0A
-%0A
         
@@ -3529,16 +3529,8 @@
     
- int i =
  dat
@@ -3540,60 +3540,23 @@
 urce
-Wrapper.getAndIncrement();%0A        dataSourceBeans%5Bi
+Beans%5BautoInc++
 %5D = 
@@ -3702,56 +3702,8 @@
 ) %7B%0A
-        queryConstructorMap = new HashMap%3C%3E();%0A%0A
     
@@ -3839,24 +3839,24 @@
 s.values();%0A
-
         Stri
@@ -3922,24 +3922,460 @@
 .size()%5D);%0A%0A
+        if (logger.isDebugEnabled()) %7B%0A            for (int i = 0; i %3C tempDataSourceNames.length; i++) %7B%0A                logger.debug(i + %22: %22 + tempDataSourceNames%5Bi%5D);%0A                logger.debug(queries%5Bi%5D);%0A                dataSourceBeans%5Bi%5D.get(i);%0A                for (int j = 0; j %3C interestedAttributes%5Bi%5D.length; j++) %7B%0A                    logger.debug(interestedAttributes%5Bi%5D%5Bj%5D);%0A                %7D%0A            %7D%0A        %7D%0A%0A
         Hdfs

Fix autoInc.