public class DebugFlagger {
    public enum DebuggingState {DEBUG_ON, DEBUG_OFF}
    public enum TestingState {TEST_ON, TEST_OFF}

    /*
    public static DebuggingState debugTimedExecution = DebuggingState.DEBUG_ON;

    public static DebuggingState debugMQL = DebuggingState.DEBUG_ON;

    public static DebuggingState debugSerializationFlag = DebuggingState
            .DEBUG_OFF;
    public static DebuggingState debugInsertionFlag = DebuggingState.DEBUG_OFF;

    public static DebuggingState debugHDFS = DebuggingState.DEBUG_OFF;

    public static DebuggingState debugRestart = DebuggingState.DEBUG_OFF;

    public static TestingState flushingTestFlag = TestingState.TEST_OFF;

    public static TestingState flushingMasterDataTestFlag = TestingState
            .TEST_OFF;
            */
} public class KiteInstance {
    private static KiteLaunchTool kite;
    private static FileSystem hdfs;

    private static String hdfsRootDirectory = "/";
    private static String hdfsHost = "hdfs://host:port";
    private static String hdfsUsername = "amr";
    private static String hdfsGroupname = null;

    private static ArrayList<MetadataEntry> metadata = new ArrayList<>();
    //private static ByteStream metadataSerializer = new ByteStream(7500);
    private static Hashtable<String, StreamDataset> streams = new Hashtable<>();

    private static BufferedWriter outLog, errorLog,showLog;
    private static Timer logTimer;
    private static String logsDirectory = "";

    public static Ignite getCluster() {
        return kite.getCluster();
    }

    public static void initSettings(KiteLaunchTool kite) {
        initSettings(kite,"kite.settings");
    }

    public static void initSettings(KiteLaunchTool kite, String
            settingsFilePath){
        readSettingsFile(settingsFilePath);
        setKite(kite);
        setHDFS();
        ConstantsAndDefaults.DISK_INDEXES_ROOT_DIRECTORY =
                                getKiteRootDirectory() +  "kite_disk_indexes/";

        //read existing streams and indexes
        loadMetadata();

        createLogs();
    }

    private static void createLogs() {
        String timestamp = getNowTimestamp();

        try {
            outLog = new BufferedWriter(new FileWriter
                    (logsDirectory+"kite-out-log-"+timestamp));
        } catch (IOException e) {
            String errMsg = "Unable to create log at " + logsDirectory +
                    "kite-out-log-"+timestamp;
            errMsg += "Error: "+e.getMessage();
            KiteInstance.logError(errMsg);
            System.err.println(errMsg);
        }

        try {
            errorLog = new BufferedWriter(new FileWriter
                    (logsDirectory + "kite-error-log-" + timestamp));
        } catch (IOException e) {
            String errMsg = "Unable to create log at " + logsDirectory +
                    "kite-error-log-"+timestamp;
            errMsg += "Error: "+e.getMessage();
            KiteInstance.logError(errMsg);
            System.err.println(errMsg);
        }

        try {
            showLog = new BufferedWriter(new FileWriter
                    (logsDirectory + "kite-show-log-" + timestamp));
        } catch (IOException e) {
            String errMsg = "Unable to create log at " + logsDirectory +
                    "kite-show-log-"+timestamp;
            errMsg += "Error: "+e.getMessage();
            KiteInstance.logError(errMsg);
            System.err.println(errMsg);
        }

        logTimer = new Timer(true);
        long timerPeriod = ConstantsAndDefaults.LOG_PERIOD_MILLISECONDS;
        logTimer.scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
                KiteInstance.flushLogs();
            }
        }, timerPeriod, timerPeriod);
    }

    private static void flushLogs() {
        try {
            if(outLog != null)
                outLog.flush();
            if(errorLog != null)
                errorLog.flush();
            if(showLog != null)
                showLog.flush();
        } catch (IOException e) {
        }
    }

    public static void logOut(String logStr) {
        if(outLog != null) {
            try {
                outLog.write(getNowTimestamp()+":"+logStr+System.lineSeparator());
            } catch (IOException e) {
            }
        }
    }

    public static void logError(String logStr) {
        if(errorLog != null) {
            try {
                errorLog.write(getNowTimestamp()+":"+logStr+System.lineSeparator());
            } catch (IOException e) {
            }
        }
    }

    public static void logShow(String logStr) {
        if(showLog != null) {
            try {
                showLog.write(getNowTimestamp()+":"+logStr+System.lineSeparator());
            } catch (IOException e) {
            }
        }
    }

    private static void readSettingsFile(String settingsFilePath) {
        try {
            Properties prop = new Properties();
            //load a properties configuration file
            prop.load(new FileInputStream(settingsFilePath));

            String tmpHdfsRootDirectory = prop.getProperty("hdfsRootDirectory");
            String tmpHdfsHost = prop.getProperty("hdfsHost");
            String tmpHdfsUsername = prop.getProperty("hdfsUsername");
            String tmpHdfsGroupname = prop.getProperty("hdfsGroupname");

            String tmpK = prop.getProperty("queryAnswerSize");
            String tmpTime = prop.getProperty("queryTimeMinutes");

            String tmpMemoryIndexCapacity = prop.getProperty
                    ("memoryIndexCapacity");
            String tmpMemoryIndexNumSegments = prop.getProperty
                    ("memoryIndexNumSegments");

            String tmpLogsDirectory = prop.getProperty("logsDirectory");

            if(tmpHdfsRootDirectory != null && tmpHdfsRootDirectory.compareTo
                    ("") != 0)
                hdfsRootDirectory = tmpHdfsRootDirectory;
            if(tmpHdfsHost != null && tmpHdfsHost.compareTo("") != 0)
                hdfsHost = tmpHdfsHost;
            if(tmpHdfsUsername != null && tmpHdfsUsername.compareTo("") != 0)
                hdfsUsername = tmpHdfsUsername;
            if(tmpHdfsGroupname != null && tmpHdfsGroupname.compareTo("") != 0)
                hdfsGroupname = tmpHdfsGroupname;

            if(tmpK != null && tmpK.compareTo("") != 0) {
                try {
                    ConstantsAndDefaults.QUERY_ANSWER_SIZE = Integer.
                            parseInt(tmpK);
                } catch (NumberFormatException e) {
                }
            }

            if(tmpTime != null && tmpTime.compareTo("") != 0) {
                try {
                    ConstantsAndDefaults.QUERY_TIME = Long.parseLong(tmpTime)
                            *60*1000;
                } catch (NumberFormatException e) {
                }
            }

            if(tmpMemoryIndexCapacity != null && tmpMemoryIndexCapacity
                    .compareTo("") != 0) {
                try {
                    ConstantsAndDefaults.MEMORY_INDEX_CAPACITY = Integer.
                            parseInt(tmpMemoryIndexCapacity);
                } catch (NumberFormatException e) {
                }
            }

            if(tmpMemoryIndexNumSegments != null && tmpMemoryIndexNumSegments
                    .compareTo("") != 0) {
                try {
                    ConstantsAndDefaults.MEMORY_INDEX_DEFAULT_SEGMENTS = Integer.
                            parseInt(tmpMemoryIndexNumSegments);
                } catch (NumberFormatException e) {
                }
            }

            if(tmpLogsDirectory != null && tmpLogsDirectory.compareTo("") !=
             0) {
                logsDirectory = tmpLogsDirectory;
                if(!logsDirectory.endsWith(File.separator))
                    logsDirectory += File.separator;
            }

        } catch (IOException e) {
        }
    }

    private static void loadMetadata() {
        byte [] metadataBytes = KiteUtils.readHDFSBlock(KiteInstance
                .getMetadataDirectory(), KiteInstance.getMetadataFileId());
        metadata.clear();
        if(metadataBytes != null) {
            ByteStream metadataSerializer = new ByteStream(metadataBytes);
            //load memory entries
            MetadataEntry entry = MetadataEntry.deserialize(metadataSerializer);
            while (entry != null) {
                if (entry.isStreamEntry() || entry.isIndexEntry()) {
                    metadata.add(entry);
                }
                entry = MetadataEntry.deserialize(metadataSerializer);
            }
        }
    }

    private static boolean restart(MetadataEntry metaData) {
        Pair<Pair<Boolean, String>, MQLResults> executionResults = null;
        if(metaData.isIndexEntry()) {
            IndexMetadataEntry indexMetadata = (IndexMetadataEntry)metaData;
            executionResults = MQL.executeIndexingRequest(indexMetadata, false);
            boolean successfulExecution = executionResults.getKey()
                    .getKey();
            String executionErrorMessage = executionResults.getKey()
                    .getValue();
            MQLResults results = executionResults.getValue();
            if(successfulExecution) {
                String msg = indexMetadata.getIndexName()+" " +
                        "index successfully restarted on "+indexMetadata
                        .getStreamName()+":"+indexMetadata.getAttributeName()
                        +"!";
                System.out.println(msg);
                KiteInstance.logOut(msg);
            }
            else {
                System.err.println(executionErrorMessage);
                KiteInstance.logError(executionErrorMessage);
            }
            return successfulExecution;
        } else if(metaData.isStreamEntry()) {
            StreamMetadataEntry streamMetadata = (StreamMetadataEntry)metaData;
            executionResults = MQL.executeStreamCreation(streamMetadata,false);
            boolean successfulExecution = executionResults.getKey()
                    .getKey();
            String executionErrorMessage = executionResults.getKey()
                    .getValue();
            MQLResults results = executionResults.getValue();
            if(successfulExecution) {
                String msg = "Stream "+streamMetadata.getStreamName()+" " +
                        "connection restarted successfully!";
                System.out.println(msg);
                KiteInstance.logOut(msg);
            }
            else {
                System.err.println(executionErrorMessage);
                KiteInstance.logError(executionErrorMessage);
            }
            return successfulExecution;
        }
        return false;
    }

    public static void setKite(KiteLaunchTool kite) {
        KiteInstance.kite = kite;
    }

    public static FileSystem hdfs() {
        return hdfs;
    }

    public static void setHDFS() {
        try {
            Configuration conf = new Configuration();
            conf.set("fs.default.name", hdfsHost);
            hdfs = FileSystem.get(conf);
            hdfs.setOwner(new Path(hdfsRootDirectory), hdfsUsername, hdfsGroupname);
        } catch (Exception e) {
            String errMsg = "Failed to access Hadoop Distributed File " +
                    "System (HDFS).";
            errMsg += System.lineSeparator();
            errMsg += "Error: "+e.getMessage();
            System.err.println(errMsg);
            KiteInstance.logError(errMsg);
            KiteInstance.closeGracefully(1);
        }
    }

    public static String getHdfsRootDirectory() {
        return hdfsRootDirectory;
    }

    public static String getKiteRootDirectory() {
        return KiteInstance.getHdfsRootDirectory()+"kite/";
    }

    public static boolean isExistingStream(String streamName) {
        StreamMetadataEntry entry = getStreamMetadataEntry(streamName);
        return entry != null;
    }

    public static boolean isExistingAttribute(String streamName,
                                              String attributeName) {
        StreamMetadataEntry entry = getStreamMetadataEntry(streamName);
        if(entry != null)
            return entry.attributeExists(attributeName);
        else
            return false;
    }

    public static boolean isSpatialAttribute(String streamName,
                                              String attributeName) {
        StreamMetadataEntry entry = getStreamMetadataEntry(streamName);
        if(entry != null)
            return entry.isSpatialAttributeExists(attributeName);
        else
            return false;
    }

    private static StreamMetadataEntry getStreamMetadataEntry(String streamName)
    {
        for(MetadataEntry entry: metadata) {
            if(entry.isStreamEntry()) {
                StreamMetadataEntry streamEntry = (StreamMetadataEntry) entry;
                if(streamEntry.isMatchingName(streamName)) {
                    return streamEntry;
                }
            }
        }
        return null;
    }

    public static void addStream(String streamName, StreamDataset stream) {
        streams.put(streamName, stream);
    }

    public static void removeStream(String streamName) {
        streams.remove(streamName);
    }

    public static StreamDataset getStream(String streamName) {
        return streams.get(streamName);
    }

    public static boolean addMetadata(MetadataEntry metadataEntry) {
        //add it to memory
        metadata.add(metadataEntry);
        boolean written = writeMetadataToDisk();
        //notify other nodes in the cluster with updated metadata
        if(!written)
            metadata.remove(metadata.size()-1);
        return written;
    }

    private static String getMetadataFileId() {
        return "metadata";
    }

    public static String getMetadataDirectory() {
        return getKiteRootDirectory()+"metadata/";
    }

    public static Attribute getAttribute(String streamName,
                                         String attributeName) {
        StreamMetadataEntry entry = getStreamMetadataEntry(streamName);
        if(entry != null)
            return entry.getAttribute(attributeName);
        else
            return null;
    }

    private static IndexMetadataEntry getIndexMetadataEntry(String streamName,
                                                             String indexName) {
        for(MetadataEntry entry: metadata) {
            if(entry.isIndexEntry()) {
                IndexMetadataEntry indexEntry = (IndexMetadataEntry) entry;
                if(indexEntry.isMatching(streamName, indexName)) {
                    return indexEntry;
                }
            }
        }
        return null;
    }

    public static boolean isExistingIndex(String streamName, String indexName) {
        return getIndexMetadataEntry(streamName, indexName) != null;
    }

    public static boolean isHomeNode(String streamName) {
        return streams.get(streamName)!=null;
    }

    public static boolean restartStream(String streamName) {
        StreamMetadataEntry streamMetadata = getStreamMetadataEntry(streamName);
        if(streamMetadata != null){
            return restart(streamMetadata);
        } else return false;
    }

    public static boolean restartStreamIndexes(String streamName) {
        ArrayList<IndexMetadataEntry> indexesMetadata = getIndexMetadataEntry
                (streamName);
        boolean successfulRestart = true;
        for(IndexMetadataEntry indexMetadata: indexesMetadata) {
            successfulRestart = restart(indexMetadata);
        }
        return successfulRestart;
    }

    private static ArrayList<IndexMetadataEntry> getIndexMetadataEntry(
                                                            String streamName) {
        ArrayList<IndexMetadataEntry> indexesMetadata = new ArrayList<>();
        for(MetadataEntry metadataEntry: metadata) {
            if(metadataEntry.isIndexEntry()) {
                IndexMetadataEntry indexMetadata = (IndexMetadataEntry)
                        metadataEntry;
                if(indexMetadata.isMatchingStream(streamName))
                    indexesMetadata.add(indexMetadata);
            }
        }
        return indexesMetadata;
    }

    public static boolean removeStreamMetadata(String streamName) {
        boolean result = false;
        ArrayList<MetadataEntry> tmpRemovedList = new ArrayList<>();
        //remove stream entry
        for(int i = metadata.size()-1; i >= 0; --i) {
            if(metadata.get(i).isStreamEntry()) {
                StreamMetadataEntry streamEntry = (StreamMetadataEntry)
                        metadata.get(i);
                if(streamEntry.isMatchingName(streamName)) {
                    tmpRemovedList.add(metadata.get(i));
                    metadata.remove(i);
                    result = true;
                }
            }
        }

        //remove stream indexes
        if(result) {
            for(int i = metadata.size()-1; i >= 0; --i) {
                if(metadata.get(i).isIndexEntry()) {
                    IndexMetadataEntry indexEntry = (IndexMetadataEntry)
                            metadata.get(i);
                    if(indexEntry.isMatchingStream(streamName)) {
                        tmpRemovedList.add(metadata.get(i));
                        metadata.remove(i);
                    }
                }
            }
            result = result && writeMetadataToDisk();
        }
        if(!result)
            metadata.addAll(tmpRemovedList);
        else
            tmpRemovedList.clear();
        return result;
    }

    private static boolean writeMetadataToDisk() {
        int estimateSize = metadata.size()*300;
        ByteStream metadataSerializer = new ByteStream(estimateSize);
        for(MetadataEntry entry: metadata)
            entry.serialize(metadataSerializer);

        try {
            KiteUtils.writeHDFSBlock(metadataSerializer, KiteInstance
                    .getMetadataDirectory(), KiteInstance.getMetadataFileId());
            if(getNodesCount() > 1)
                getCluster().compute(getCluster().cluster().forRemotes())
                        .broadcast(() -> KiteInstance.loadMetadata());
            return true;
        } catch (IOException e) {
            String errMsg = "Unable to write metadata to HDFS.";
            errMsg += System.lineSeparator();
            errMsg += "Error: "+e.getMessage();
            KiteInstance.logError(errMsg);
            System.err.println(errMsg);
            return false;
        }
    }

    public static boolean removeIndexMetadata(String streamName, String
            indexName) {
        MetadataEntry removedEntry = null;
        boolean result = false;
        for(int i = 0; i < metadata.size(); ++i) {
            if(metadata.get(i).isIndexEntry()) {
                IndexMetadataEntry indexEntry = (IndexMetadataEntry)
                        metadata.get(i);
                if(indexEntry.isMatching(streamName, indexName)) {
                    removedEntry = metadata.get(i);
                    metadata.remove(i);
                    result = writeMetadataToDisk();
                    if(!result)
                        metadata.add(removedEntry);
                    return result;
                }
            }
        }
        return false;
    }

    public static void descStream(String streamName) {
        StreamMetadataEntry streamMetadata = getStreamMetadataEntry(streamName);
        desc(streamMetadata);
        if(streamMetadata != null) {
            ArrayList<IndexMetadataEntry> indexesMetadata =
                                            getIndexMetadataEntry(streamName);
            for(IndexMetadataEntry entry: indexesMetadata)
                desc(entry);
        }
    }

    private static void desc(MetadataEntry entry) {
        if(entry != null) {
            String entryStr = entry.toString();
            System.out.println(entryStr);
            KiteInstance.logOut(entryStr);
        }
    }

    public static void descAllStreams() {
        for(MetadataEntry entry: metadata)
            desc(entry);
    }

    public static int getNodesCount() {
        long topVer = getCluster().cluster().topologyVersion();
        return kite.getCluster().cluster().topology(topVer).size();
    }

    public static String getNowTimestamp() {
        String timestamp = Instant.now().toString();
        timestamp = timestamp.replaceAll(":","-");
        timestamp = timestamp.replaceAll("/","-");
        return timestamp;
    }

    public static void pauseAllActiveStreams() {
        for(StreamDataset stream: streams.values()) {
            stream.stop();
        }
        //wait until on-going data insertion is done
        for(StreamDataset stream: streams.values()) {
            while (stream.inInsertion())
                KiteUtils.busyWait(50);
        }
    }

    public static void closeLogs() {
        try {
            if(outLog != null) {
                outLog.flush();
                outLog.close();
            }
            if(errorLog != null) {
                errorLog.flush();
                errorLog.close();
            }
            if(showLog != null) {
                showLog.flush();
                showLog.close();
            }
        } catch (IOException e) {
        }
    }

    public static void closeGracefully(int exitStatus) {
        KiteInstance.pauseAllActiveStreams();
        String msg = "All active streams are paused, Kite node exits!!";
        if(exitStatus == 0) {
            KiteInstance.logOut(msg);
            System.out.println(msg);
        } else {
            KiteInstance.logError(msg);
            System.err.println(msg);
        }
        KiteInstance.closeLogs();
        KiteInstance.printMessageOnAllClusterNodes("Node "+KiteInstance
                .getLocalNodeName()+" exits!");
        System.exit(exitStatus);
    }

    public static void printMessageOnAllClusterNodes(String msg) {
        getCluster().compute().broadcast(() -> System.out.println(msg));
    }

    public static String getLocalNodeName() {
        return KiteInstance.getCluster().cluster().localNode().hostNames().
                iterator().next();
    }
} public class Attribute {
    private String name;
    private String type;

    private static Hashtable<String, Class> dataTypes;
    static {
        dataTypes = new Hashtable<>();
        dataTypes.put("long", Long.class);
        dataTypes.put("string", String.class);
        dataTypes.put("geolocation", GeoLocation.class);
        dataTypes.put("timestamp", Date.class);

        dataTypes.put("list<long>", List.class);
        dataTypes.put("list<string>", List.class);
        dataTypes.put("list<leolocation>", List.class);
        dataTypes.put("list<timestamp>", List.class);
    }

    public Attribute(String name, String type) {
        this.name = name.toLowerCase();
        this.type = type.toLowerCase();
    }

    @Override public String toString(){return "<"+name+":"+type+">";}
    @Override public boolean equals(Object other) {
        Attribute otherAttribute = (Attribute)other;
        return this.getName().compareTo(otherAttribute.getName()) == 0 &&
               this.getType().compareTo(otherAttribute.getType()) == 0;
    }
    public String getName() {return name;}

    public String getType() {return type;}

    public boolean isSpatialAttribute(){
        return getType().compareTo("geolocation")==0 ||
                getType().compareTo("list<geolocation>")==0;
    }

    public static boolean isValidDataType(String attributeType) {
        return dataTypes.containsKey(attributeType);
    }

    @Override
    public int hashCode() {
        return name.hashCode();
    }

    public boolean isList(){return type.startsWith("list");}

    public Object parseValue(String valStr) {
        Object val = null;
        switch (type){
            case "string": case "list<string>":
                val = valStr.toLowerCase();
                break;
            case "long": case "list<long>":
            case "timestamp": case "list<timestamp>":
                try {
                    val = Long.parseLong(valStr);
                } catch (NumberFormatException e) {
                    val = null;
                }
                break;
            case "geolocation": case "list<geolocation>":
                GeoLocation loc = null;
                String [] parts = valStr.split(":");
                String locType = parts[0];

                try {
                    if (locType.compareTo("null") == 0)
                        loc = null;
                    else if (locType.compareTo("2") == 0) {// 2 coordinates(point)
                        loc = new PointLocation(Double.parseDouble(parts[1]),
                                Double.parseDouble(parts[2]));
                    } else if (locType.compareTo("4") == 0) {// 4 coordinates(rectangle)
                        loc = new Rectangle(Double.parseDouble(parts[1]),
                                Double.parseDouble(parts[2]), Double.parseDouble
                                (parts[3]), Double.parseDouble(parts[4]));
                    } else
                        loc = null;
                } catch (NumberFormatException e) {
                }
                val = loc;
                break;
            default:
                val = null;
        }
        return val;
    }

    public void serializeVal(Object val, ByteStream byteStream) {
        switch (type) {
            case "string":
                String valStr = (String)val;
                byteStream.write(valStr);
                break;
            case "long":
                Long valLng = (Long)val;
                byteStream.write(valLng);
                break;
            case "timestamp":
                Long valTs = (Long)val;
                byteStream.write(valTs);
                break;
            case "geolocation":
                GeoLocation valGeo = (GeoLocation)val;
                Serializer.serializeGeoLocation(valGeo, byteStream);
                break;
            case "list<string>":
                List<String> valStrLst = (List<String>)val;
                Serializer.serializeStringList(valStrLst, byteStream);
                break;
            case "list<long>":
                List<Long> valLngLst = (List<Long>)val;
                Serializer.serializeLongList(valLngLst, byteStream);
                break;
            case "list<timestamp>":
                valLngLst = (List<Long>)val;
                Serializer.serializeLongList(valLngLst, byteStream);
                break;
            case "list<geolocation>":
                List<GeoLocation> valGeoLst = (List<GeoLocation>)val;
                Serializer.serializeGeoLocationList(valGeoLst, byteStream);
                break;
        }
    }

    public Object deserializeVal(ByteStream byteStream) {
        Object val = null;
        switch (type) {
            case "string":
                String valStr = byteStream.readString();
                val = valStr;
                break;
            case "long":
                Long valLng = byteStream.readLong();
                val = valLng;
                break;
            case "timestamp":
                Long valTs = byteStream.readLong();
                val = valTs;
                break;
            case "geolocation":
                GeoLocation valGeo = GeoLocation.deserialize(byteStream);
                val = valGeo;
                break;
            case "list<string>":
                List<String> valStrLst = Serializer.deserializeStringList
                        (byteStream);
                val = valStrLst;
                break;
            case "list<long>":
                List<Long> valLngLst = Serializer.deserializeLongList
                        (byteStream);
                val = valLngLst;
                break;
            case "list<timestamp>":
                valLngLst = Serializer.deserializeLongList
                        (byteStream);
                val = valLngLst;
                break;
            case "list<geolocation>":
                List<GeoLocation> valGeoLst = Serializer
                        .deserializeGeoLocationList (byteStream);
                val = valGeoLst;
                break;
            default:
                val = null;
        }
        return val;
    }

    public List<String> getStringListValue(Object value) {
        ArrayList<String> strLstVal = new ArrayList<>();
        if(value == null)
            return strLstVal;
        switch (type) {
            case "string":
                String valStr = (String)value;
                strLstVal.add(valStr);
                break;
            case "long":
                Long valLng = (Long)value;
                strLstVal.add(valLng.toString());
                break;
            case "timestamp":
                Long valTs = (Long)value;
                strLstVal.add(valTs.toString());
                break;
            case "geolocation":
                GeoLocation valGeo = (GeoLocation)value;
                strLstVal.add(valGeo.toString());
                break;
            case "list<string>":
                List<String> strLst = (List<String>)value;
                strLstVal.addAll(strLst);
                break;
            case "list<long>":
                List<Long> valLngLst = (List<Long>)value;
                for(int i = 0; i < valLngLst.size(); ++i) {
                    strLstVal.add(valLngLst.get(i).toString());
                }
                break;
            case "list<timestamp>":
                valLngLst = (List<Long>)value;
                for(int i = 0; i < valLngLst.size(); ++i) {
                    strLstVal.add(valLngLst.get(i).toString());
                }
                break;
            case "list<geolocation>":
                List<GeoLocation> valGeoLst = (List<GeoLocation>)value;
                for(int i = 0; i < valGeoLst.size(); ++i) {
                    strLstVal.add(valGeoLst.get(i).toString());
                }
                break;
            default:
                strLstVal = null;
        }
        return strLstVal;
    }

    public List<GeoLocation> getGeoListValue(Object value) {
        List<GeoLocation> list = new ArrayList<>();
        if(isSpatialAttribute()) {
            switch (type) {
                case "geolocation":
                    GeoLocation valGeo = (GeoLocation)value;
                    list.add(valGeo);
                    break;
                case "list<geolocation>":
                    list = (List<GeoLocation>)value;
                    break;
                default:
                    list = null;
            }
        }
        return list;
    }

    public void serialize(ByteStream byteStream) {
        byteStream.write(name);
        byteStream.write(type);
    }

    public boolean equals(Object val1, Object val2) {
        boolean result;
        switch (type) {
            case "string":
                result = ((String)val1).equals((String)val2);
                break;
            case "long":
            case "timestamp":
                result = ((Long)val1) == ((Long)val2);
                break;
            case "geolocation":
                result = ((GeoLocation)val1).equals((GeoLocation)val2);
                break;
            case "list<string>":
                List<String> strLst1 = (List<String>)val1;
                List<String> strLst2 = (List<String>)val2;
                result = true;
                if(strLst1.size() == strLst2.size()){
                    for (int i = 0; i < strLst1.size(); ++i) {
                        result = strLst1.get(i).equals(strLst2.get(i));
                    }
                }
                else result = false;
                break;
            case "list<timestamp>":
            case "list<long>":
                List<Long> lngLst1 = (List<Long>)val1;
                List<Long> lngLst2 = (List<Long>)val2;
                result = true;
                if(lngLst1.size() == lngLst2.size()){
                    for (int i = 0; i < lngLst1.size(); ++i) {
                        result = lngLst1.get(i) == lngLst2.get(i);
                    }
                }
                else result = false;
                break;
            case "list<geolocation>":
                List<GeoLocation> geoLst1 = (List<GeoLocation>)val1;
                List<GeoLocation> geoLst2 = (List<GeoLocation>)val2;
                result = true;
                if(geoLst1.size() == geoLst2.size()){
                    for (int i = 0; i < geoLst1.size(); ++i) {
                        result = geoLst1.get(i).equals(geoLst2.get(i));
                    }
                }
                else result = false;
                break;
            default:
                result = false;
        }
        return result;
    }

    public boolean overlap(Object val1, Object val2) {
        boolean result;
        switch (type) {
            case "geolocation":
                result = ((GeoLocation)val1).overlap((GeoLocation)val2);
                break;
            case "list<geolocation>":
                List<GeoLocation> geoLst1 = (List<GeoLocation>)val1;
                List<GeoLocation> geoLst2 = (List<GeoLocation>)val2;
                result = true;
                if(geoLst1.size() == geoLst2.size()){
                    for (int i = 0; i < geoLst1.size(); ++i) {
                        result = geoLst1.get(i).overlap(geoLst2.get(i));
                    }
                }
                else result = false;
                break;
            default:
                result = false;
        }
        return result;
    }

    public static Attribute deserialize(ByteStream byteStream) {
        String name = byteStream.readString();
        String type = byteStream.readString();
        return new Attribute(name, type);
    }
} public class DirectoryEntry {
    private String filePath;
    private String keyStr = null;
    private Integer keySpatialPartitionId = null;
    private long offset;
    private int length;

    public DirectoryEntry(String filePath, String key, long offset, int length)
    {
        this.filePath = filePath;
        this.keyStr = key;
        this.offset = offset;
        this.length = length;
    }

    public long getOffset() {
        return offset;
    }

    public int getLength() {
        return length;
    }

    public DirectoryEntry(String filePath, Integer keySpatialPartitionId,
                          long offset, int length) {
        this.filePath = filePath;
        this.keySpatialPartitionId = keySpatialPartitionId;
        this.offset = offset;
        this.length = length;
    }

    public String getFilePath() { return filePath; }
    public String getKeyStr() {
        return keyStr;
    }
    public Integer getKeyGeo() {
        return keySpatialPartitionId;
    }

    public String toString() {
        return keyStr+","+offset+","+length+","+filePath;
    }
} public interface DiskIndexSegment<K> {
    ArrayList<Microblog> search(K key, int k, Query query);

    void destroy();

    boolean discardInMemoryDirectory();
} public class MemorySpatialIndexSegment extends SpatialIndex implements
        MemoryIndexSegment<GeoLocation, Long>
{
    private String uniqueName;
    private int ingestedDataSize = 0;
    private StreamDataset stream;
    private Attribute indexedAttribute;
    /*
     * A map: partition id -> partition data
     */
    protected IgniteCache<Integer,SpatialPartition> spatialPartitionsData;

    public MemorySpatialIndexSegment(String uniqueName, SpatialPartitioner
            partitioner, long batchUpdateTimeMilliseconds, StreamDataset
            stream, Attribute indexedAttribute)
    {
        this.setName(uniqueName);
        this.spatialPartitioner = partitioner;
        this.stream = stream;
        this.indexedAttribute = indexedAttribute;

        //Index creation

        //Setting index configuration
        CacheConfiguration<Integer,SpatialPartition> cnfig = new
                CacheConfiguration<> (this.getName());
        cnfig.setCacheMode(CacheMode.LOCAL);
        cnfig.setCacheStoreFactory(new IgniteReflectionFactory<BulkLoadSpatialCacheStore>
                (BulkLoadSpatialCacheStore.class));
        cnfig.setWriteThrough(true);

        Ignite cluster = KiteInstance.getCluster(); //getting underlying cluster
        spatialPartitionsData = cluster.getOrCreateCache(cnfig); //create
    }

    @Override
    public void setName(String uniqueName) { this.uniqueName = uniqueName; }
    @Override
    public String getName() { return uniqueName; }

    public int insert(Integer partitionId, Long value) {
        int inserted = 0;
        inserted += spatialPartitionsData.get(partitionId).insert(value);
        ingestedDataSize += inserted;
        return inserted;
    }

    @Override
    public int insert(GeoLocation key, Long value) {
        List<Integer> spatialPartitionIds =  spatialPartitioner.overlap(key);
        int inserted = 0;
        for(Integer partitionId : spatialPartitionIds) {
            inserted += insert(partitionId, value);
        }
        return inserted;
    }

    @Override
    public int insert(GeoLocation key, ArrayList<Long> values) {
        List<Integer> spatialPartitionIds =  spatialPartitioner.overlap(key);

        int inserted = 0;
        for(Integer partitionId : spatialPartitionIds) {
            inserted += insert(partitionId, values);
        }
        return inserted;
    }

    public int insert(Integer key, ArrayList<Long> values) {
        spatialPartitionsData.localLoadCache(null,spatialPartitionsData,key,values);
        ingestedDataSize += values.size();
        return values.size();
    }

    /**
     * Get all spatial object that overlap with given location
     * If location is a spatial range, this is a traditional range query.
     * If location is a point, this gives a range query for the point's
     * spatial partition.
     * @param location: search location
     * @return list of spatial objects represents the query answer
     */
    @Override
    public ArrayList<Long> search(GeoLocation location, Query query) {
        ArrayList<Long> answer = new ArrayList<>();

        if(query.isSpatialRange()) {
            Rectangle queryRange = location instanceof Rectangle? (Rectangle)
                    location : query.getParam_Rectangle("SpatialRange");

            List<Integer> spatialPartitionIds =  spatialPartitioner.overlap
                    (queryRange);

            for(Integer partitionId : spatialPartitionIds) {
                ArrayList<Long> mids = search(partitionId);
                if(spatialPartitioner.contained(partitionId, queryRange))
                    answer.addAll(mids);
                else {
                    //filter on spatial range
                    for(Long id: mids) {
                        Microblog record = stream.getRecord(id);
                        if(record.overlap(queryRange, indexedAttribute))
                            answer.add(id);
                    }
                }
            }
        }
        else  if(query.isSpatialKNN());
            //ToDo: implement KNN
        return answer;
    }

    public ArrayList<Long> search(Integer partitionId) {
        ArrayList<Long> answer = new ArrayList<>();
        SpatialPartition sPart = spatialPartitionsData.get(partitionId);
        if(sPart != null)
            answer.addAll(sPart.getData());
        return answer;
    }

    @Override
    public boolean clear() {
        spatialPartitionsData.clear();
        //GC.invoke();
        return true;
    }

    @Override
    public int getIngestedDataSize() {
        return ingestedDataSize;
    }

    public Iterator<Cache.Entry<Integer,SpatialPartition>>
            getEntriesIterator() {
        return spatialPartitionsData.iterator();
    }
} public abstract class SpatialIndex {
    protected SpatialPartitioner spatialPartitioner;
} public class SpatialPartition {
    //protected Rectangle partitionBoundaries;
    protected ArrayList<Long> data;

    public SpatialPartition(){data = new ArrayList<>();}
    public SpatialPartition(ArrayList<Long> data){this.data = data;}

    public boolean mergeData(SpatialPartition other) {
        this.data.addAll(other.getData());
        return true;
    }

    public boolean mergeData(ArrayList<Long> other) {
        this.data.addAll(other);
        return true;
    }

    public ArrayList<Long> getData() { return data; }

    public int insert(Long microblogId) {
        data.add(microblogId);
        return 1;
    }
} public interface SpatialPartitioner {

    //public boolean create(List<? extends GeoLocation> dataPoints,
    //                      Rectangle spaceMBR, int numPointsPerPartition);

    /**
     * Finds spatial partitions that overlap with a given geo-location
     * @param location: a given location (point, rectangle,...etc)
     * @return List of spatial partition ids that overlap with location
     */
    public List<Integer> overlap(GeoLocation location);
    public ByteStream serialize();

    boolean contained(Integer partitionId, Rectangle range);

    static SpatialPartitioner deserialize(byte [] partitionerBytes) {
        ByteStream bytes = new ByteStream(partitionerBytes);
        byte partitionerType = bytes.readByte();//0: grid partitioner
        if(partitionerType == 0) {
            Rectangle gridBoundaries = (Rectangle)GeoLocation.deserialize(bytes);
            int numRows = bytes.readInt();
            int numColumns = bytes.readInt();
            return new GridPartitioner(gridBoundaries, numRows, numColumns);
        }
        return null;
    }

    static boolean isValidType(String type) {
        ArrayList<String> spatialParitionsTypes = new ArrayList<>();
        spatialParitionsTypes.add("grid");

        return spatialParitionsTypes.contains(type);
    }
} public class Query {
    private QueryType queryType;
    private Condition condition = null;
    private Attribute primaryAttribute = null;
    private Object primaryAttributeValue = null;

    private Hashtable<String, Object> params;
    private TemporalPeriod time = ConstantsAndDefaults.QUERY_TIME();
    private int k = ConstantsAndDefaults.QUERY_ANSWER_SIZE;

    public Query(QueryType queryType) {
        this.queryType = queryType;
        params = new Hashtable<>();
    }

    @Override public String toString() {
        return "<"+queryType+","+condition+","+primaryAttribute+","+
                primaryAttributeValue+">";
    }
    public Query(QueryType type, Condition condition) {
        this.queryType = type;
        this.condition = condition;
    }

    public QueryType getQueryType() { return queryType; }

    public Rectangle getParam_Rectangle(String paramName) {
        Rectangle spatialRect = (Rectangle)params.get(paramName);
        return spatialRect;
    }

    public void putParam_Rectangle(String paramName, Rectangle rect) {
        params.put(paramName, rect);
    }

    public boolean isHash() {
        return queryType==QueryType.HASH_TEMPORAL;
    }

    public boolean isSpatial() {
        return queryType== QueryType.SPATIAL_RANGE_TEMPORAL || queryType== QueryType
        .SPATIAL_KNN_TEMPORAL;
    }

    public boolean isSpatialRange() {
        return queryType== QueryType.SPATIAL_RANGE_TEMPORAL;
    }

    public boolean isSpatialKNN() {
        return queryType== QueryType.SPATIAL_KNN_TEMPORAL;
    }

    public int getK() {
        return k;
    }

    public TemporalPeriod getTime() {
        return time;
    }

    public void setK(int k) {
        this.k = k;
    }

    public void setTime(TemporalPeriod time) {
        this.time = time;
    }

    public Condition getCondition() {
        return condition;
    }

    public void setQueryType(QueryType queryType) {
        this.queryType = queryType;
    }

    public void setPrimaryAttribute(Attribute primaryAttribute) {
        this.primaryAttribute = primaryAttribute;
    }

    public Attribute getPrimaryAttribute() {
        return primaryAttribute;
    }

    public boolean isSingleSearchAttribute() {
        return getCondition().getSearchAttributes().size()==1;
    }

    public void setPrimaryAttributeValue(Object primaryAttributeValue) {
        this.primaryAttributeValue = primaryAttributeValue;
    }

    public Object getPrimaryAttributeValue() {
        return primaryAttributeValue;
    }
} public class StreamDataset implements Runnable {

    private String uniqueName;
    private ArrayBigDataset streamData;
    private IdGenerator currId = null;

    private StreamingDataSource stream;
    private ArrayList<MemoryIndex> indexes;
    private ArrayList<Boolean> indexesActivators;
    private Preprocessor<String, Microblog> preprocessor;

    //recovery data structures
    private ByteStream recoveryData = new ByteStream(ConstantsAndDefaults
            .RECOVERY_BLK_SIZE_BYTES);
    private List<List<Microblog>> tempRecoveryDataList;
    boolean appendingToRecoveryData = false;
    private int recoveryBlkCounter = 0;
    private Object recoveryDataLock = new Object();
    private double microblogLengthEstimate = 200;//initiailly 200 bytes
    private int microblogLengthEstimateCount = 0;

    //Aux flags and counters
    private boolean starting = true;
    private boolean showEnable = false;
    private boolean stopped = false;
    private boolean resuming = false;
    int indexInInsertion = -1;

    long dataInserted = 0;
    int streamBatchesCount = 0;
    private boolean insertionInProgress = false;

    public Microblog getRecord(Long id) {
        Microblog record = streamData.getRecord(id);
        if(record != null)
            return record;
        else {
            //retrieve from disk
            record = getRecordFromDisk(id); //for now
            return record;
        }
    }

    private Microblog getLast() {
        return streamData.getLast();
    }

    private Microblog getRecordFromDisk(Long id) {
        //search the disk buffer before accessing the disk
        byte [] diskBufferData = null;
        synchronized (recoveryDataLock) {
            diskBufferData = recoveryData.getBytesClone();
        }
        if(diskBufferData.length > 0) {
            Pair<Pair<Boolean,Microblog>,Pair<Long,Long>> searchResults =
                    getLatestKFromBytes(id, diskBufferData);
            boolean searchSucceed = searchResults.getKey().getKey();
            if(searchSucceed)
                return searchResults.getKey().getValue();
        }

        //search disk
        Pair<String,String> dirFileidPair = getRecoveryFileIdPrefix();
        String parentDir = dirFileidPair.getKey();
        String fileIdPrefix = dirFileidPair.getValue();

        long baseId = 0;
        int low = 0;
        int high = recoveryBlkCounter-1;


        while (low <= high) {
            int estimateBlockId = low +(int)((id-baseId)*microblogLengthEstimate
                    /ConstantsAndDefaults.RECOVERY_BLK_SIZE_BYTES);

            if(estimateBlockId < low)
                estimateBlockId = low;
            if(estimateBlockId > high)
                estimateBlockId = high;

            String fileId = fileIdPrefix+KiteUtils.getRecoveryFileId
                    (estimateBlockId);
            Pair<Pair<Boolean,Microblog>,Pair<Long,Long>> searchResults =
                    searchHDFSBlock(id, parentDir, fileId);
            boolean searchSucceed = searchResults.getKey().getKey();
            if(searchSucceed)
                return searchResults.getKey().getValue();
            else {
                Long firstId = searchResults.getValue().getKey();
                Long lastId = searchResults.getValue().getValue();

                int length = (int)(lastId-firstId);
                double recordLengthEstimate = ConstantsAndDefaults
                        .RECOVERY_BLK_SIZE_BYTES/length;

                microblogLengthEstimate =
                        (microblogLengthEstimate*microblogLengthEstimateCount+
                                recordLengthEstimate)/
                                (microblogLengthEstimateCount+1);
                microblogLengthEstimateCount++;
                if(id < firstId)
                    high = estimateBlockId-1;
                else if(id > lastId) {
                    low = estimateBlockId + 1;
                    baseId = lastId+1;
                }
            }
        }
        return null;
    }

    private List<Microblog> getLatestKFromDisk(int k, TemporalPeriod time,
                                               Condition condition) {
        //search the disk for latest "k" microblogs within "time"
        List<Microblog> latestK = null;
        boolean continuingSearch = false;

        //search the disk buffer before accessing the disk
        byte [] diskBufferData = null;
        long blockTimeWidth = 5000;
        long baseTimestamp = new Date().getTime()-blockTimeWidth;
        int blockTimeWidthCount = 0;

        final long latestTimestamp = time.to().getTime();
        synchronized (recoveryDataLock) {
            diskBufferData = recoveryData.getBytesClone();
        }
        if(diskBufferData.length > 0) {
            Pair<Pair<Boolean,List<Microblog>>,Pair<Long,Long>> searchResults =
                    getLatestKFromBytes(k, time, diskBufferData);
            boolean searchSucceed = searchResults.getKey().getKey();
            if(searchSucceed) {
                if(condition == null)
                    latestK = searchResults.getKey().getValue();
                else {
                    latestK = new ArrayList<>();
                    List<Microblog> tmpList = searchResults.getKey().getValue();
                    for(Microblog record: tmpList) {
                        if(record.match(condition))
                            latestK.add(record);
                    }
                }
                if(latestK.size() < k)
                    continuingSearch = true;
                else
                    return latestK;
            } else {
                long firstTimestamp = searchResults.getValue().getKey();
                long lastTimestamp = searchResults.getValue().getValue();
                blockTimeWidth = lastTimestamp-firstTimestamp;
                baseTimestamp = firstTimestamp;
            }
        }

        //search disk
        Pair<String,String> dirFileidPair = getRecoveryFileIdPrefix();
        String parentDir = dirFileidPair.getKey();
        String fileIdPrefix = dirFileidPair.getValue();

        if(continuingSearch) {
            int blockId = recoveryBlkCounter-1;
            if(blockId >= 0) {
                int restCount = k - latestK.size();
                String fileId = fileIdPrefix + KiteUtils.getRecoveryFileId
                        (blockId);

                List<Microblog> restK = getLatestKFromHDFSBlock(restCount, null,
                        parentDir, fileId);
                if(condition == null)
                    latestK.addAll(restK);
                else {
                    for(Microblog microblog: restK)
                        if(microblog.match(condition))
                            latestK.add(microblog);
                }
            }
            return latestK;
        } else {
            //long baseId = 0;
            int low = 0;
            int high = recoveryBlkCounter-1;

            while (low <= high) {
                int estimateBlockId = high - (int)
                        (Math.ceil(Math.abs(latestTimestamp-baseTimestamp)/
                                (1.0*blockTimeWidth)));

                if(estimateBlockId < low)
                    estimateBlockId = low;
                if(estimateBlockId > high)
                    estimateBlockId = high;

                String fileId = fileIdPrefix+KiteUtils.getRecoveryFileId
                        (estimateBlockId);
                Pair<Pair<Boolean,List<Microblog>>,Pair<Long,Long>>
                        searchResults = searchHDFSBlock(k, time, parentDir,
                        fileId);
                boolean searchSucceed = searchResults.getKey().getKey();
                if(searchSucceed) {
                    if(condition == null) {
                        latestK = searchResults.getKey().getValue();
                    } else {
                        List<Microblog> tmpList = searchResults.getKey().
                                getValue();
                        latestK = new ArrayList<>();
                        for(Microblog microblog: tmpList)
                            if(microblog.match(condition))
                                latestK.add(microblog);
                    }
                    if(latestK.size() < k) {
                        int blockId = estimateBlockId-1;
                        if(blockId >= 0) {
                            int restCount = k - latestK.size();
                            fileId = fileIdPrefix + KiteUtils.getRecoveryFileId
                                    (blockId);

                            List<Microblog> restK = getLatestKFromHDFSBlock(
                                    restCount,null, parentDir, fileId);
                            if(condition == null)
                                latestK.addAll(restK);
                            else {
                                for(Microblog microblog: restK)
                                    if(microblog.match(condition))
                                        latestK.add(microblog);
                            }
                        }
                        return latestK;
                    }
                    else
                        return latestK;
                }
                else {
                    Long firstTimestamp = searchResults.getValue().getKey();
                    Long lastTimestamp = searchResults.getValue().getValue();
                    Long blockTimeWidthTmp = lastTimestamp-firstTimestamp;
                    blockTimeWidthCount++;
                    blockTimeWidth = blockTimeWidthCount==1?
                            blockTimeWidthTmp:blockTimeWidth+blockTimeWidthTmp;
                    blockTimeWidth /= blockTimeWidthCount;

                    if(latestTimestamp < firstTimestamp) {
                        high = estimateBlockId-1;
                        baseTimestamp = firstTimestamp;
                    }
                    else if(latestTimestamp > lastTimestamp) {
                        low = estimateBlockId + 1;
                    }
                }
            }
        }
        return null;
    }

    private Pair<Pair<Boolean,List<Microblog>>,Pair<Long,Long>> searchHDFSBlock(
            int k, TemporalPeriod time, String parentDir, String fileId) {
        byte [] fileBytes = KiteUtils.readHDFSBlock(parentDir, fileId);
        if (fileBytes != null)
            return getLatestKFromBytes(k, time, fileBytes);
        else
            return null;
    }

    private Pair<Pair<Boolean,Microblog>,Pair<Long,Long>> searchHDFSBlock(
            Long id, String parentDir, String fileId) {

        byte [] fileBytes = KiteUtils.readHDFSBlock(parentDir, fileId);
        if(fileBytes != null)
            return getLatestKFromBytes(id,fileBytes);
        else
            return null;
    }

    private Pair<Pair<Boolean,Microblog>,Pair<Long,Long>> getLatestKFromBytes(long id,
                                                                              byte [] bytes) {
        ByteStream byteStream = new ByteStream(bytes);
        Microblog microblog;
        Long firstId = null, lastId = null;
        while ((microblog = Microblog.deserialize(byteStream, stream
                .getScheme()))!= null) {
            if(microblog.getId() == id) {
                //successful search
                return new Pair<>(new Pair<>(true,microblog),null);
            }
            else {
                if(firstId == null)
                    firstId = microblog.getId();
                lastId = microblog.getId();
            }
        }
        return new Pair<>(new Pair<>(false,null),new Pair<>(firstId,lastId));
    }

    private List<Microblog> getLatestKFromHDFSBlock(int k, TemporalPeriod time,
            String parentDir, String fileId) {

        byte [] fileBytes = KiteUtils.readHDFSBlock(parentDir, fileId);
        if(fileBytes != null)
            return getLatestKFromBytes(k, time, fileBytes)
                    .getKey().getValue();
        else
            return new ArrayList<>();
    }

    private Pair<Pair<Boolean,List<Microblog>>,Pair<Long,Long>> getLatestKFromBytes
            (int k, TemporalPeriod time, byte [] bytes) {
        //get latest "k" microblogs, within "time" in "bytes"
        //Assumption 1: microblogs in bytes are ordered by time from
        // older to more recent
        ByteStream byteStream = new ByteStream(bytes);
        List<Microblog> microblogs = new ArrayList<>(k);
        Microblog tmpMicroblog;

        Long firstTimestamp = null, lastTimestamp = null;
        while ((tmpMicroblog = Microblog.deserialize(byteStream, stream
                .getScheme()))!= null) {
            if(time == null || time.overlap(tmpMicroblog.getTimestamp())) {
                //successful search
                microblogs.add(tmpMicroblog);
                if(microblogs.size() > k)
                    microblogs.remove(0); //Assumption 1 above
            }
            else {
                if (firstTimestamp == null)
                    firstTimestamp = tmpMicroblog.getTimestamp();
                lastTimestamp = tmpMicroblog.getTimestamp();
            }
        }
        return new Pair<>(new Pair<>(microblogs.size()>0,microblogs),new Pair<>
                (firstTimestamp,lastTimestamp));
    }

    public StreamDataset(String name, StreamingDataSource dataSource, int
            memoryCapacity) {
        this.uniqueName = name;
        this.stream = dataSource;
        indexes = new ArrayList<>();
        indexesActivators = new ArrayList<>();
        preprocessor = stream.getPreprocessor();

        if(memoryCapacity > 0)
            streamData = new ArrayBigDataset(memoryCapacity);
        else
            streamData = new ArrayBigDataset();

        List<List<Microblog>> tempLst = new ArrayList<>();
        tempRecoveryDataList = Collections.synchronizedList(tempLst);

        //if existing stream, load recoveryBlkCounter and latest id
        Pair<String,String> dirFileidPair = getRecoveryFileIdPrefix();
        Pair<Integer, Long> existingIds = KiteUtils.loadStreamIds
                (dirFileidPair, stream.getScheme());
        recoveryBlkCounter = existingIds.getKey();
        currId = new IdGenerator(existingIds.getValue()+1);
    }

    public StreamDataset(String name, StreamingDataSource dataSource) {
        this(name, dataSource, -1);
    }

    public void startStreaming() {
        Thread streamingThread = new Thread(this);
        streamingThread.start();
    }

    private void streaming() {

        List<String> lines;
        List<Microblog> microblogs;

        while (!stopped) {
            insertionInProgress = true;
            try {
                lines = stream.getData();

                if (showEnable) {
                    String msg = getName() + " got " + lines.size() + " records " +
                            "from " + stream.toString() + ".";
                    System.out.println(msg);
                    KiteInstance.logShow(msg);
                }

                microblogs = preprocessor.preprocess(lines, currId);
            } catch (Exception e) {
                stopped = true;
                String errMsg = "Stream "+getName()+" has accidentally " +
                        "paused. The other stream end may not be responding " +
                        "correctly.";
                System.err.println();
                System.err.println(errMsg);
                KiteInstance.logError(errMsg);
                break;
            }
            //add to recovery buffer
            tempRecoveryDataList.add(microblogs);
            if(!appendingToRecoveryData) {
                Thread recoveryThread = new Thread(this);
                recoveryThread.start();
            }

            //insert in master buffer
            streamData.add(microblogs);

            ++streamBatchesCount;

            if(showEnable) {
                String msg = microblogs.size()+" records inserted in the " +
                        "stream " + getName();
                System.out.println(msg);
                KiteInstance.logShow(msg);
            }

            if(indexes.size() > 0) {
                for(indexInInsertion = 0; indexInInsertion < indexes.size();
                    ++indexInInsertion) {
                    if(indexesActivators.get(indexInInsertion)) {
                        MemoryIndex index = indexes.get(indexInInsertion);
                        dataInserted += index.insert(microblogs);

                        if (showEnable) {
                            String msg = "Index: " + index.
                                    getName() + " total inserted data size: "
                                    + index.getSize() + ", batches: " +
                                    streamBatchesCount+ ", in-memory data " +
                                    "size: " + index.getInMemorySize();
                            System.out.println(msg);
                            KiteInstance.logShow(msg);
                        }
                    }
                }
            }
            streamData.checkFlushing();
            insertionInProgress = false;
        }
    }

    public boolean inInsertion() {return insertionInProgress;}

    public boolean createIndexHash(Attribute attribute, String indexName, int
            indexCapacity, int numIndexSegments, boolean loadDiskIndex){

        indexes.add(new MemoryHashIndex(getIndexUniqueName(indexName),this,
                attribute, indexCapacity, numIndexSegments, loadDiskIndex));
        indexesActivators.add(false);

        return true;
    }

    public boolean createIndexHash(Attribute attribute, String indexName,
                                   boolean loadDiskIndex){
        return createIndexHash(attribute, indexName, ConstantsAndDefaults
                .MEMORY_INDEX_CAPACITY, ConstantsAndDefaults
                .MEMORY_INDEX_DEFAULT_SEGMENTS, loadDiskIndex);
    }

    public boolean createIndexSpatial(Attribute attribute, String indexName,
                                      SpatialPartitioner spatialPartitioner,
                                      boolean loadDiskIndex) {
        return createIndexSpatial(attribute, indexName, spatialPartitioner,
                ConstantsAndDefaults.MEMORY_INDEX_CAPACITY, ConstantsAndDefaults
                .MEMORY_INDEX_DEFAULT_SEGMENTS, loadDiskIndex);

    }
    public boolean createIndexSpatial(Attribute attribute, String indexName,
                                      SpatialPartitioner spatialPartitioner,
                                      int indexCapacity, int numIndexSegments,
                                      boolean loadDiskIndex) {
        if(!attribute.isSpatialAttribute())
            return false;

        indexes.add(new MemorySpatialIndex(getIndexUniqueName(indexName),
                spatialPartitioner, this, attribute, indexCapacity,
                numIndexSegments, loadDiskIndex));
        indexesActivators.add(false);

        return true;
    }

    public int size() {
        return streamData.size();
    }

    private String getDirectoryId() {
        return KiteInstance.getKiteRootDirectory()+  "kite_streams/"+
                uniqueName + "/";
    }

    public void decrementIndexesCount(long id) {
        streamData.decrementIndexCount(id);
    }

    public void incrementIndexCount(long id) {
        streamData.incrementIndexCount(id);
    }

    @Override
    public void run() {
        if(starting || resuming) {
            starting = false;
            resuming = false;

            try {
                streaming();
            } catch (OutOfMemoryError e) {
                String errMsg = "All active streams will be paused. " +
                        "Accidental OutOfMemoryError while streaming "+
                        this.getName()+"!!";
                errMsg += System.lineSeparator();
                errMsg += "Error: "+e.getMessage();
                KiteInstance.logError(errMsg);
                System.err.println(errMsg);
                KiteInstance.pauseAllActiveStreams();
            } catch (Exception e) {
                String errMsg = "Accidental error while streaming "+
                        this.getName()+"!!";
                errMsg += System.lineSeparator();
                errMsg += "Error: "+e.getMessage();
                KiteInstance.logError(errMsg);
                System.err.println(errMsg);
            }
        } else {
            try {
                appendRecoveryData();
            } catch (IOException e) {
                String errMsg = "Unable to write recovery data of stream "+
                        this.getName()+"!!";
                errMsg += System.lineSeparator();
                errMsg += "Error: "+e.getMessage();
                KiteInstance.logError(errMsg);
                System.err.println(errMsg);
            }
        }
    }

    private void appendRecoveryData() throws IOException {
        if(!appendingToRecoveryData) {
            appendingToRecoveryData = true;

            while (tempRecoveryDataList.size() > 0) {
                List<Microblog> currRecoveryList = tempRecoveryDataList.get(0);
                tempRecoveryDataList.remove(0);
                for (Microblog m : currRecoveryList) {
                    synchronized (recoveryDataLock) {
                        m.serialize(recoveryData, stream.getScheme());
                    }
                    //System.out.println("appending recovery id="+m.getId());
                }

                if (recoveryData.getWrittenBytes() >= ConstantsAndDefaults.
                        RECOVERY_BLK_SIZE_BYTES) {
                    Pair<String,String> dirFileidPair =
                            getRecoveryFileIdPrefix();
                    String parentDirectory = dirFileidPair.getKey();
                    String filePrefix = dirFileidPair.getValue();
                    String fileId = filePrefix + getNextRecoveryFileId();

                    KiteUtils.writeHDFSBlock(recoveryData, parentDirectory, fileId);

                    synchronized (recoveryDataLock) {
                        recoveryData.clear();
                        recoveryData = new ByteStream(ConstantsAndDefaults.
                                RECOVERY_BLK_SIZE_BYTES);
                    }
                }
            }
            appendingToRecoveryData = false;
        }
    }

    private String getNextRecoveryFileId() {
        String id = KiteUtils.getRecoveryFileId(recoveryBlkCounter);
        recoveryBlkCounter++;
        return id;
    }

    private Pair<String,String> getRecoveryFileIdPrefix() {
        String fileId = uniqueName + "_data";
        String parentDirectory = getDirectoryId();

        Pair<String,String> fileDir = new Pair<String,String>
                (parentDirectory, fileId);
        return fileDir;
    }

    public Scheme getScheme() {return stream.getScheme();}

    public Attribute getAttribute(String attributeName) {
        return getScheme().getAttribute(attributeName);
    }

    public String getName() {
        return uniqueName;
    }

    public void show() {
        showEnable = true;
    }
    public void unshow() {
        showEnable = false;
    }

    public boolean isShown() { return showEnable; }

    public MQLResults search(Query query, ArrayList<String> attributeNames) {
        MQLResults results = null;
        QueryType queryType = query.getQueryType();
        switch (queryType) {
            case UNDECIDED:
                //decide on the query type and send it for evaluation
                Pair<QueryType, Pair<Attribute, Object>> decidedQueryType =
                        decideOnQueryType (query);
                query.setQueryType(decidedQueryType.getKey());
                query.setPrimaryAttribute(decidedQueryType.getValue().getKey());
                query.setPrimaryAttributeValue(decidedQueryType.getValue()
                        .getValue());
                results = this.search(query, attributeNames);
                break;
            case UNINDEXED_TEMPORAL:
            case PURE_TEMPORAL:
                Condition condition = null;
                if(queryType == QueryType.UNINDEXED_TEMPORAL)
                    condition = query.getCondition();
                int k = query.getK();
                TemporalPeriod time = query.getTime();
                ArrayList<Microblog> latestK = streamData.getLatestK(k, time,
                        condition);
                if(latestK.size() < k) {
                    //search disk
                    int restCount = k-latestK.size();
                    List<Microblog> rest = getLatestKFromDisk(restCount,
                            time, condition);
                    if(rest != null)
                        latestK.addAll(rest);
                }
                results = new MQLResults(latestK, attributeNames);
                break;
            case HASH_TEMPORAL:
                MemoryHashIndex hashIndex = (MemoryHashIndex) getIndex(query
                        .getPrimaryAttribute());


                if(hashIndex != null) {
                    List<String> keyLst = query.getPrimaryAttribute()
                          .getStringListValue(query.getPrimaryAttributeValue());
                    String key = keyLst != null? keyLst.get(0): null;

                    List<Long> mIds = hashIndex.search(key, query, query
                            .getTime());

                    List<Microblog> resultsMicroblogs = new ArrayList<>();
                    for(Long mId:mIds) {
                        Microblog microblog = getRecord(mId);
                        if(query.isSingleSearchAttribute()) {
                            resultsMicroblogs.add(microblog);
                            if(resultsMicroblogs.size() >= query.getK())
                                break;
                        }
                        else if(microblog.match(query.getCondition())) {
                            resultsMicroblogs.add(microblog);
                            if(resultsMicroblogs.size() >= query.getK())
                                break;
                        }
                    }
                    if(resultsMicroblogs.size() < query.getK()) {
                        //search corresponding disk index
                        int restCount = query.getK()-resultsMicroblogs.size();

                        List<Microblog> diskMicroblogs = hashIndex.searchDisk
                                (restCount, key, query, query.getTime());
                        if(diskMicroblogs != null) {
                            resultsMicroblogs.addAll(diskMicroblogs);
                        }
                    }
                    results = new MQLResults(resultsMicroblogs, attributeNames);
                }
                break;
            case SPATIAL_RANGE_TEMPORAL:
                MemorySpatialIndex spatialIndex = (MemorySpatialIndex)
                        getIndex(query.getPrimaryAttribute());
                if(spatialIndex != null) {
                    List<GeoLocation> keyLst = query.getPrimaryAttribute()
                            .getGeoListValue(query.getPrimaryAttributeValue());
                    GeoLocation key = keyLst != null? keyLst.get(0): null;

                    List<Long> mIds = spatialIndex.search(key, query,
                                                            query.getTime());
                    List<Microblog> resultsMicroblogs = new ArrayList<>();
                    for(Long mId:mIds) {
                        Microblog microblog = getRecord(mId);
                        if(query.isSingleSearchAttribute()) {
                            resultsMicroblogs.add(microblog);
                            if(resultsMicroblogs.size() >= query.getK())
                                break;
                        }
                        else if(microblog.match(query.getCondition())) {
                            resultsMicroblogs.add(microblog);
                            if(resultsMicroblogs.size() >= query.getK())
                                break;
                        }
                    }
                    if(resultsMicroblogs.size() < query.getK()) {
                        //search corresponding disk index
                        int restCount = query.getK()-resultsMicroblogs.size();
                        List<Microblog> diskMicroblogs = spatialIndex.searchDisk
                                (restCount, key, query, query.getTime());
                        resultsMicroblogs.addAll(diskMicroblogs);
                    }
                    results = new MQLResults(resultsMicroblogs, attributeNames);
                }
                break;
            case SPATIAL_KNN_TEMPORAL:
                //skip for now
                break;
            default:
                //unknown query type
                break;
        }
        return results;
    }

    private MemoryIndex getIndex(Attribute attribute) {
        for(MemoryIndex index: indexes)
            if(index.getIndexedAttribute().equals(attribute))
                return index;
        return null;
    }

    private Pair<QueryType, Pair<Attribute, Object>> decideOnQueryType
                                        (Query query) {
        boolean hasOR = query.getCondition().hasOR();
        List<Attribute> searchAttributes = query.getCondition()
                .getSearchAttributes();
        QueryType queryType = null;
        Attribute primaryAttribute = null;
        Object primaryAttributeVal = null;

        if(hasOR) {
            //do not use indexes for now
            //ToDo: handle OR queries with potential multiple index access
            queryType = QueryType.UNINDEXED_TEMPORAL;
        } else if(searchAttributes.size() > 0) {
            List<Attribute> indexedAttributes = new ArrayList<>();
            for(int i = 0; i < indexes.size(); ++i) {
                if(indexesActivators.get(i))
                    indexedAttributes.add(indexes.get(i).getIndexedAttribute());
            }

            //get the list of search attribute that are indexed
            List<Attribute> searchIndexedAttributes = new ArrayList<>();
            List<Attribute> list1, list2;
            if(searchAttributes.size() < indexedAttributes.size()) {
                list1 = searchAttributes;
                list2 = indexedAttributes;
            } else {
                list1 = indexedAttributes;
                list2 = searchAttributes;
            }
            for(Attribute attribute:list1)
                if(list2.contains(attribute))
                    searchIndexedAttributes.add(attribute);

            if(searchIndexedAttributes.size() == 0) {
                //the search attributes are not indexed
                queryType = QueryType.UNINDEXED_TEMPORAL;
            } else if(searchIndexedAttributes.size() == 1) {
                //single search attribute is indexed
                if(searchIndexedAttributes.get(0).isSpatialAttribute()) {
                    queryType = QueryType.SPATIAL_RANGE_TEMPORAL;
                } else {
                    queryType = QueryType.HASH_TEMPORAL;
                }
                primaryAttribute = searchIndexedAttributes.get(0);
            } else {//multiple search attributes are indexed
                for (Attribute attribute: searchIndexedAttributes) {
                    if(!attribute.isSpatialAttribute()) {
                        queryType = QueryType.HASH_TEMPORAL;
                        primaryAttribute = attribute;
                    }
                }
                if(primaryAttribute == null) {
                    queryType = QueryType.SPATIAL_RANGE_TEMPORAL;
                    primaryAttribute = searchIndexedAttributes.get(0);
                }
            }
        } else {
            queryType = QueryType.PURE_TEMPORAL;
        }

        if(primaryAttribute != null) {
            primaryAttributeVal = query.getCondition().getValue
                    (primaryAttribute);
        }

        return new Pair<>(queryType, new Pair<>(primaryAttribute,
                primaryAttributeVal));
    }

    public boolean activateIndex(String indexName) {
        String indexUniqueName = getIndexUniqueName(indexName);
        for(int i = indexes.size()-1; i >= 0; --i) {
            if(indexes.get(i).getName().compareTo(indexUniqueName)==0){
                indexesActivators.set(i,true);
                return true;
            }
        }
        return false;
    }

    public boolean deactivateIndex(String indexName) {
        String indexUniqueName = getIndexUniqueName(indexName);
        for(int i = indexes.size()-1; i >= 0; --i) {
            if(indexes.get(i).getName().compareTo(indexUniqueName)==0){
                indexesActivators.set(i,false);
                return true;
            }
        }
        return false;
    }

    private String getIndexUniqueName(String indexName) {
        return uniqueName+"_"+indexName;
    }

    public boolean destroyIndex(String indexName) {
        String indexUniqueName = getIndexUniqueName(indexName);
        for(int i = indexes.size()-1; i >= 0; --i) {
            if(indexes.get(i).getName().compareTo(indexUniqueName)==0){
                indexesActivators.set(i,false);
                while (indexInInsertion == i)
                    KiteUtils.busyWait(50);
                //ToDo: check for races among threads here (if any)
                indexes.get(i).destroy();
                indexes.remove(i);
                indexesActivators.remove(i);
                return true;
            }
        }
        return false;
    }

    public boolean destroyAllIndexes() {
        boolean result = true;
        for(int i = indexes.size()-1; i >= 0; --i) {
            indexesActivators.set(i,false);
            while (indexInInsertion == i)
                KiteUtils.busyWait(50);
            //ToDo: check for races among threads here (if any)
            indexes.get(i).destroy();
            indexes.remove(i);
            indexesActivators.remove(i);
            result = result && true;
        }
        return result;
    }

    public void stop() {
        stopped = true;
    }

    public void resume() {
        stopped = false;
        resuming = true;
        startStreaming();
    }
} public class ConstantsAndDefaults {
    public static long GC_PERIOD_MILLISECONDS = 15*60*1000;
    public static long LOG_PERIOD_MILLISECONDS = 2*60*1000;
    public static int FILE_BLK_SIZE_BYTES = 128*1024*1024*1;
    public static int BUFFER_SIZE_BYTES = 4096 * 1024;
    public static int RECOVERY_BLK_SIZE_BYTES = 64*1024*1024*1;

    //Memory data constants
    public final static int MAX_MEMORY_CHUNK_CAPACITY = 3000000;
    public final static double MEMORY_DATA_FLUSH_THRESHOLD = 0.2;

    //Memory indexes
    public static int MEMORY_INDEX_DEFAULT_SEGMENTS = 5;
    public static final long BATCH_UPDATE_TIME_MILLISECONDS = 3000;
    public static final int TEMPORAL_FLUSHING_PERIODIC_TIME_MINUTES = 360;
    public static final int TEMPORAL_FLUSHING_PERIODIC_DATA_SIZE = 1000000;

    public static int MEMORY_INDEX_CAPACITY =
            TEMPORAL_FLUSHING_PERIODIC_DATA_SIZE;

    //Disk indexes
    public static String DISK_INDEXES_ROOT_DIRECTORY;
    public static final int NUM_DISK_SEGMENTS_IN_MEMORY_DIRECTORY = 10;

    //Query default values
    public static int QUERY_ANSWER_SIZE = 20;
    public static long QUERY_TIME = 3*60*60*1000;//3 hours
    public static TemporalPeriod QUERY_TIME() {
        long to = new Date().getTime();
        long from = to - QUERY_TIME;
        return new TemporalPeriod(from, to);
    }
} public interface GeoLocation {
    int serialize(ByteStream byteStream);

    static GeoLocation deserialize(ByteStream byteStream) {
        byte locationType = byteStream.readByte();
        GeoLocation location;
        switch (locationType){
            case 0:
                location = null;
                break;
            case 1://PointLocation
                double latitude = byteStream.readDouble();
                double longitude = byteStream.readDouble();
                location = new PointLocation(latitude, longitude);
                break;
            case 2://Rectangle
                double north = byteStream.readDouble();
                double south = byteStream.readDouble();
                double east = byteStream.readDouble();
                double west = byteStream.readDouble();
                location = new Rectangle(north, south, east, west);
                break;
            default:
                location = null;
                break;
        }
        return location;
    }

    boolean overlap(GeoLocation rect);

    boolean isPoint();

    static int serializeNull(ByteStream byteStream) {
        byte type = 0;
        byteStream.write(type);
        return Byte.BYTES;
    }
} public class KiteUtils {
    public static double GreatCircleDistanceMiles(double lat1, double lat2,
                                                  double lng1, double lng2) {
        //approximate value
        return Math.sqrt(Math.pow(69.1*(lat1-lat2),2)+Math.pow(53*(lng1-lng2),
                2));
    }


    public static String int2str(int counter, int n) {
        String counterStr = Integer.toString(counter);
        while (counterStr.length() < n)
            counterStr = "0"+counterStr;
        return counterStr;
    }

    public static int binarySearch (Microblog[] microblogs, int low, int high,
            Timestamp timestamp) {
        //locate the latest microblog with time <= timestamp
        if(microblogs[high].getTimestamp() < timestamp.getTime())
            return high;
        if(microblogs[low].getTimestamp() > timestamp.getTime())
            return -1;

        double epsilon = 1.1;
        int l = low, r = high;
        while (l < r) {
            int m = (int)Math.floor(1.0*(l+r)/2.0);
            if(Math.abs(microblogs[m].getTimestamp()-timestamp.getTime()) <=
                    epsilon) {
                //approximate match, get the exact match
                if (microblogs[m].getTimestamp() <= timestamp.getTime()) {
                    while(microblogs[m].getTimestamp() <= timestamp.getTime())
                        ++m;
                    return m-1;
                }
                else {
                    while(microblogs[m].getTimestamp() > timestamp.getTime())
                        --m;
                    return m;
                }
            }
            else {
                if (microblogs[m].getTimestamp() < timestamp.getTime())
                    l = m+1;
                else if(microblogs[m].getTimestamp() > timestamp.getTime())
                    r = m-1;
            }
        }
        return -1;
    }

    public static int binarySearch_StartsWithin(List<TemporalPeriod>
                                                        timePeriodsList,
                                                Timestamp timestamp) {
        int l = 0, r = timePeriodsList.size();
        while (l < r) {
            int m = (int)Math.floor(1.0*(l+r)/2.0);
            if(timePeriodsList.get(m).overlap(timestamp))
                return m;
            else {
                if (timePeriodsList.get(m).endsBefore(timestamp))
                    l = m+1;
                else //(timePeriodsList.get(m).startsAfter(timestamp))
                            r = m-1;
            }
        }
        return -1;
    }

    public static int binarySearch_StartOverlap(List<TemporalPeriod>
                                                        timePeriodsList,
                                                TemporalPeriod period) {
        int ind = binarySearch_StartsWithin(timePeriodsList, period.from());
        if(ind < 0) {
            if(timePeriodsList != null && timePeriodsList.size() > 0) {
                if (period.overlap(timePeriodsList.get(0).from()))
                    return 0;
                else
                    return -1;
            } else
                return -1;
        } else
            return ind;
    }

    public static int binarySearch_StartOverlap2(List<TemporalPeriod>
                                                    timePeriodsList,
                                                TemporalPeriod period) {
        //check old periods that starts before the time horizon
        if(period.startsBefore(timePeriodsList.get(0)) &&
                !period.endsBefore(timePeriodsList.get(0)))
            return 0;

        //if not old, search within the time horizon
        int l = 0, r = timePeriodsList.size();
        while (l < r) {
            int m = (int)Math.floor(1.0*(l+r)/2.0);
            if(period.startsWithin(timePeriodsList.get(m)))
                return m;
            else {
                if (period.startsAfter(timePeriodsList.get(m)))
                    l = m+1;
                else if(period.startsBefore(timePeriodsList.get(m)))
                    r = m-1;
            }
        }
        return -1;
    }

    public static int binarySearch_EndOverlap(List<TemporalPeriod>
                                                        timePeriodsList,
                                                TemporalPeriod period) {
        int ind = binarySearch_StartsWithin(timePeriodsList, period.to());
        if(ind < 0){
            if(timePeriodsList != null && timePeriodsList.size() > 0) {
                if (period.overlap(timePeriodsList.get(timePeriodsList.size()
                        -1).to()))
                    return timePeriodsList.size()-1;
                else
                    return -1;
            } else
                return -1;
        } else
            return ind;
    }

    public static long getFileCount(String hdfsDirectory) {
        Path path = new Path(hdfsDirectory);
        FileSystem hdfs = KiteInstance.hdfs();
        long fileCount = 0;
        try {
            if(hdfs.exists(path)){
                ContentSummary cs = hdfs.getContentSummary(path);
                fileCount = cs.getFileCount();
            }
        } catch (IOException e) {
        }
        return fileCount;
    }
    public static byte[] readHDFSBlock(String parentDirectory, String fileId) {
        String hdfsPath = parentDirectory+fileId;

        Hashtable<String,ArrayList<Microblog>> blockEntries = new Hashtable<>();
        FileSystem hdfs = KiteInstance.hdfs();
        try {
            Path path = new Path(hdfsPath);

            long fileLenBytes = hdfs.getFileStatus(path).getLen();
            FSDataInputStream block = hdfs.open(path,
                    ConstantsAndDefaults.BUFFER_SIZE_BYTES);
            byte [] fileBytes = new byte[(int)fileLenBytes];
            block.readFully(0, fileBytes, 0, (int)fileLenBytes);
            return fileBytes;
        } catch (IOException e) {
            return null;
        }
    }

    public static String writeHDFSBlock(ByteStream serializedBytes, String
            parentDirectory, String fileId) throws
            IOException {
        byte [] bytes = serializedBytes.getBytes();
        int blockSize = Math.max(1024*1024, (int)(Math.ceil(serializedBytes
                .getWrittenBytes()/512.0)*512));

        FileSystem hdfs = KiteInstance.hdfs();

        Path fileP = new Path(parentDirectory, fileId);
        hdfs.mkdirs(fileP.getParent());

        FSDataOutputStream file = hdfs.create(fileP, true, ConstantsAndDefaults
                .BUFFER_SIZE_BYTES, (short)1, blockSize);
        file.write(bytes, 0, serializedBytes.getWrittenBytes());
        file.flush();
        file.close();

        return fileId;
    }

    public static ByteStream checkByteStreamCapacity(ByteStream bstream, int
            delta) {
        if(bstream.getWrittenBytes()+delta >= bstream.getCapacity()) {
            //if no space, enlarge the serializer
            int copyLength = bstream.getWrittenBytes();
            byte [] copyBytes = Arrays.copyOfRange(bstream.getBytes(), 0,
                    copyLength);
            int newLen = Math.max(bstream.getCapacity()+delta,
                                    bstream.getCapacity()*2);
            bstream = new ByteStream(newLen);
            bstream.write(copyBytes);
        }
        return bstream;
    }

    public static void busyWait(int durationMilliseconds) {
        long startTime = System.currentTimeMillis();
        int duration = 0;
        do {
            duration = (int) (System.currentTimeMillis()-startTime);
        }while (duration < durationMilliseconds);
    }

    public static Pair<Integer,Long> loadStreamIds(Pair<String, String>
                                                           dirFileidPair,
                                                   Scheme microblogScheme) {
        String parentDir = dirFileidPair.getKey();
        String fileIdPrefix = dirFileidPair.getValue();

        int recoveryBlockCount = (int) getFileCount(parentDir);
        long latestId = -1;
        if(recoveryBlockCount > 0) {
            String lastRecoveryFileId = fileIdPrefix+getRecoveryFileId
                    (recoveryBlockCount-1);
            byte [] lastBlockBytes = readHDFSBlock(parentDir,
                    lastRecoveryFileId);
            ByteStream byteStream = new ByteStream(lastBlockBytes);
            Microblog microblog, lastMicroblog = null;
            while ((microblog = Microblog.deserialize(byteStream,
                    microblogScheme))!= null) {
                lastMicroblog = microblog;
            }
            latestId = lastMicroblog.getId();
        }
        return new Pair<>(recoveryBlockCount, latestId);
    }

    public static String getRecoveryFileId(int cntr) {
        return KiteUtils.int2str(cntr, 6);
    }
} public class Rectangle implements GeoLocation {
    private double north, south, east, west;
    private double widthMiles, heightMiles;

    public Rectangle (double north, double south, double east, double west) {
        this.north = north;
        this.south = south;
        this.east = east;
        this.west = west;
        widthMiles = KiteUtils.GreatCircleDistanceMiles(north,north,west,east);
        heightMiles = KiteUtils.GreatCircleDistanceMiles(north,south,west,west);
    }

    public Rectangle(PointLocation center, double latWidth, double lngWidth) {
        this(center.getLatitude()+latWidth, center.getLatitude()-latWidth,
                center.getLatitude()+lngWidth, center.getLatitude()-lngWidth);
    }

    public double getWidth() {
        return widthMiles;
    }

    public double getHeight() {
        return heightMiles;
    }

    public boolean isInside(GeoLocation location) {
        if (location instanceof PointLocation)
            return isInside((PointLocation)location);
        else if (location instanceof Rectangle)
            return isInside((Rectangle)location);
        else throw new IllegalArgumentException("Niether point nor " +
                    "rectanlgular location type");
    }
    public boolean isInside(PointLocation point) {
        if(point.getLatitude() >= south && point.getLatitude() <= north
                && point.getLongitude() >= west && point.getLongitude() <= east)
            return true;
        else
            return false;
    }

    public boolean isInside(Rectangle other) {
        if(other.getSouth() >= south && other.getNorth() <= north
                && other.getWest() >= west && other.getEast() <= east)
            return true;
        else
            return false;
    }

    public double getNorth() {
        return north;
    }

    public double getSouth() {
        return south;
    }

    public double getEast() {
        return east;
    }

    public double getWest() {
        return west;
    }

    public boolean isValid() {
        return south <= north && west <= east;
    }

    public Rectangle merge(Rectangle other) {
        if(this.equals(other))
            return other;
        Rectangle mergedRect = new Rectangle(Math.max(north,other.getNorth()),
                Math.min(south,other.getSouth()), Math.max(east,other.getEast
                ()), Math.min(west, other.getWest()));
        return mergedRect;
    }

    @Override
    public int serialize(ByteStream byteStream) {
        int byteLen = 0;

        byte locationType = 2;
        byteLen += byteStream.write(locationType);
        byteLen += byteStream.write(north);
        byteLen += byteStream.write(south);
        byteLen += byteStream.write(east);
        byteLen += byteStream.write(west);

        return byteLen;
    }

    @Override
    public boolean overlap(GeoLocation loc) {
        if(loc.isPoint()) {
            PointLocation pt = (PointLocation)loc;
            return this.isInside(pt);
        } else {
            Rectangle rect = (Rectangle) loc;
            return isInside(rect.getNW()) || isInside(rect.getNE()) ||
                    isInside(rect.getSW()) || isInside(rect.getSE());
        }
    }

    @Override public boolean isPoint() {return false;}

    private PointLocation getNW() {
        return new PointLocation(this.getNorth(),this.getWest());
    }

    private PointLocation getNE() {
        return new PointLocation(this.getNorth(),this.getEast());
    }

    private PointLocation getSW() {
        return new PointLocation(this.getSouth(),this.getWest());
    }

    private PointLocation getSE() {
        return new PointLocation(this.getSouth(),this.getEast());
    }

    public String toString() { return "("+north+","+south+","+east+"," +
            ""+west+")"; }

    public static int numBytes() {
        return Byte.BYTES+4*Double.BYTES;
    }

    public boolean isBoundedCoordinates() {
        return -90 <= north && north <=90 && -90<= south && south <=90 &&
                -180<= east && east <=180 && -180<= west && west <=180;
    }

    @Override public boolean equals(Object other) {
        double epsilon = 0.000001;
        Rectangle pt = (Rectangle) other;
        return Math.abs(getNorth()-pt.getNorth()) < epsilon &&
                Math.abs(getSouth()-pt.getSouth()) < epsilon &&
                Math.abs(getEast()-pt.getEast()) < epsilon &&
                Math.abs(getWest()-pt.getWest()) < epsilon;
    }
} public class Microblog implements Serializable {
    //private long id;
    private Long timestamp;
    //private ArrayList<String> keywords;
    //private GeoLocation geolocation;
    //private String username;

    private HashMap<Attribute,Object> attributes;

    //system fields
    private long kiteId;
    //private static long currKiteId = 0;

    public Microblog(HashMap<Attribute,Object> attributes, long systemKey) {
        this.attributes = attributes;
        kiteId = systemKey;

        //assign timestamp
        timestamp = (Long)getAttributeValue("timestamp");
        if(timestamp == null)
            timestamp = new Date().getTime();
    }

    public Microblog(long id, long timestamp, ArrayList<String> keywords,
                     GeoLocation loc, String user, long systemKey) {
        //this.id = id;
        //this.timestamp = timestamp;
        //this.keywords = keywords;
        //this.geolocation = loc;
        //this.username = user;

        attributes = new HashMap<Attribute,Object>();

        kiteId = systemKey;
    }

    public long getId() { return kiteId; }


    public String toString()
    {
        String string = "";
        Set<Attribute> mAttributes = this.getAttributes();
        for(Attribute attribute: mAttributes) {
            String attributeName = attribute.getName();
            Object val = this.getAttributeValue(attributeName);
            List<String> strLst = attribute.getStringListValue(val);
            String valStr = "";
            for(String str: strLst)
                valStr += str+",";
            string += valStr;
        }
        return string;
    }

    public int serialize(ByteStream byteStream, Scheme scheme) {
        int capacity = 1400;
        ByteStream tempByteStream = new ByteStream(capacity);

        //serialize record attributes
        ArrayList<Attribute> schemeAttributes = scheme.getAttributes();
        for(int i = 0; i < schemeAttributes.size(); ++i) {
            Object val = this.attributes.get(schemeAttributes.get(i));
            //if(val == null)
            //    throw new IllegalArgumentException("Error in Microblog " +
            //            "record serialization. Mismatch scheme attributes.")
            schemeAttributes.get(i).serializeVal(val, tempByteStream);
        }

        //serialize system fields
        tempByteStream.write(kiteId);

        byte [] microblogBytes = tempByteStream.getBytes();
        int microblogBytesLen = tempByteStream.getWrittenBytes();

        int bytesLen = 0;
        bytesLen += byteStream.write(microblogBytesLen);
        bytesLen += byteStream.write(microblogBytes, microblogBytesLen);

        return bytesLen;
    }

    public static boolean skipMicroblog(ByteStream byteStream) {
        if(!byteStream.hasRemaining())
            return false;

        int microblogBytesLen = byteStream.readInt();
        return byteStream.skipBytes(microblogBytesLen);
    }

    public static Microblog deserialize(ByteStream byteStream, Scheme scheme) {
        if(!byteStream.hasRemaining())
            return null;

        int microblogBytesLen = byteStream.readInt();
        byte [] bytes = byteStream.readBytes(microblogBytesLen);


        ByteStream tmpByteStream = new ByteStream(bytes);
        HashMap<Attribute,Object> attributes = new HashMap<>();
        ArrayList<Attribute> schemeAttributes = scheme.getAttributes();
        for(int i = 0; i < schemeAttributes.size(); ++i) {
            Object val = schemeAttributes.get(i).deserializeVal(tmpByteStream);
            attributes.put(schemeAttributes.get(i), val);
        }

        long kiteId = tmpByteStream.readLong();

        Microblog microblog = new Microblog(attributes, kiteId);
        return microblog;
    }

    public long getTimestamp() { return timestamp; }

    public Object getAttributeValue(Attribute attribute) {
        return attributes.get(attribute);
    }

    public Object getAttributeValue(String attributeName) {
        for(Attribute attribute: attributes.keySet())
            if(attribute.getName().compareTo(attributeName) == 0)
                return attributes.get(attribute);
        return null;
    }

    public Attribute getAttribute(String attributeName) {
        for(Attribute attribute: attributes.keySet())
            if(attribute.getName().compareTo(attributeName) == 0)
                return attribute;
        return null;
    }

    public List<String> getKeysStr(Attribute indexedAttribute) {
        List<String> keys = indexedAttribute.getStringListValue(attributes
                .get(indexedAttribute));
        return keys;
    }

    public List<GeoLocation> getKeysGeo(Attribute indexedAttribute) {
        List<GeoLocation> keys = indexedAttribute.getGeoListValue(attributes
                .get(indexedAttribute));
        return keys;
    }

    public boolean overlap(Rectangle range, Attribute spatialAttribute) {
        GeoLocation loc = (GeoLocation)attributes.get(spatialAttribute);
        return loc.overlap(range);
    }

    public GeoLocation getLocation(Attribute spatialAttribute) {
        return (GeoLocation)attributes.get(spatialAttribute);
    }

    public Set<Attribute> getAttributes() {
        return attributes.keySet();
    }

    public boolean match(Condition condition) {
        return condition.evaluate(this.attributes);
    }
} public class ByteStream {

    private final int DELTA = 1024;
    private int capacity ;
    private ByteBuffer byteStream ;

    public ByteStream(int capacity) {
        this.capacity = 2*capacity+4*DELTA;
        byteStream = ByteBuffer.allocate(this.capacity);
    }

    public ByteStream(byte [] bytes) {
        byteStream = ByteBuffer.wrap(bytes);
    }

    public int write(byte[] bytes) {
        byteStream.put(bytes);
        return Byte.BYTES*bytes.length;
    }

    public int write(byte[] bytes, int length) {
        byteStream.put(bytes,0,length);
        return Byte.BYTES*length;
    }

    public int write(byte oneByte) {
        byteStream.put(oneByte);
        return Byte.BYTES;
    }

    public int write(int value) {
        byteStream.putInt(value);
        return Integer.BYTES;
    }

    public int write(long value) {
        byteStream.putLong(value);
        return Long.BYTES;
    }

    public int write(String str) {
        int bytesLen = 0;
        if(str == null) {
            bytesLen += this.write(0);
        }
        else {
            byte[] strBytes = str.getBytes();

            bytesLen += this.write(strBytes.length); //key length
            bytesLen += this.write(strBytes);
        }
        return bytesLen;
    }

    public byte[] getBytes() { return byteStream.array(); }

    public int getWrittenBytes() { return byteStream.position(); }

    public int getReadBytes() { return byteStream.position(); }

    public boolean hasRemaining(){return byteStream.hasRemaining();}

    public int write(double dblVal) {
        byteStream.putDouble(dblVal);
        return Double.BYTES;
    }

    public int readInt() {
        return byteStream.getInt();
    }

    public byte[] readBytes(int bytesLen) {
        byte [] bytes = new byte[bytesLen];
        byteStream.get(bytes,0,bytesLen);
        return bytes;
    }

    public boolean skipBytes(int bytesLen) {
        try {
            byteStream.position(byteStream.position()+bytesLen);
            return true;
        } catch (IllegalArgumentException e) {
            return false;
        }
    }

    public long readLong() {
        return byteStream.getLong();
    }

    public String readString() {
        int keyBytesLen = readInt();
        if(keyBytesLen == 0)
            return null;
        byte [] keyStrBytes = readBytes(keyBytesLen);
        return new String(keyStrBytes);
    }

    public byte readByte() {
        byte [] bytes = readBytes(1);
        return bytes[0];
    }

    public double readDouble() {
        return byteStream.getDouble();
    }

    public void clear() {
        byteStream.clear();
    }

    public byte[] getBytesClone() {
        int lenToCopy = getWrittenBytes();
        return Arrays.copyOfRange(byteStream.array(),0, lenToCopy);
    }

    public int getCapacity() {
        return capacity;
    }
} public class Serializer {

    public static Pair<Integer, Integer> serializeHashEntry(
            Cache.Entry<String, ArrayList<Long>> entry, StreamDataset stream,
            ByteStream recordBytes, int entryInd) {
        int bytesLen = 0;
        bytesLen += serializeKey(entry.getKey(), recordBytes);
        Pair<Integer,Integer> results = serializeMemoryValue(entry.getValue(),
                stream, recordBytes, entryInd);
        return new Pair<Integer,Integer>(results.getKey()+bytesLen,
                results.getValue());
    }

    private static int serializeKey(String key, ByteStream recordBytes) {
        int bytesLen = 0;
        byte type = 0;
        //key type, 0 is String, 1 is Spatial
        bytesLen += recordBytes.write(type);
        bytesLen += recordBytes.write(key);
        return bytesLen;
    }

    private static int serializeKeyGeo(Integer spatialPartitionId, ByteStream
            recordBytes) {
        int bytesLen = 0;
        byte type = 1;
        //key type, 0 is String, 1 is Spatial
        bytesLen += recordBytes.write(type);
        bytesLen += recordBytes.write(spatialPartitionId);
        return bytesLen;
    }

    private static Pair<Integer,Integer> serializeMemoryValue(ArrayList<Long>
                         idList, StreamDataset stream, ByteStream recordBytes,
                                                              int startInd) {
        int bytesLen = 0;
        for(int i = startInd; i < idList.size(); ++i) {
            Long mid = idList.get(i);
            Microblog microblog = stream.getRecord(mid);

            bytesLen += microblog.serialize(recordBytes, stream.getScheme());
            //null pointer
            // exception here, mostly during index flushing, looks like microblog is null for some reason (may be related to discard method)
            stream.decrementIndexesCount(microblog.getId());

            if(bytesLen >= ConstantsAndDefaults.FILE_BLK_SIZE_BYTES) {
                if((i+1) < idList.size())
                    return new Pair<Integer, Integer>(bytesLen, i + 1);
            }
        }
        return new Pair<Integer,Integer>(bytesLen,-1);
    }

    public static int serializeGeoLocation(GeoLocation geolocation,
                                           ByteStream byteStream) {
        if(geolocation == null)
            return GeoLocation.serializeNull(byteStream);
        else
            return geolocation.serialize(byteStream);
    }

    public static int serializeStringList(List<String> strings, ByteStream
            byteStream) {
        int bytesLen = 0;
        int listLen = strings == null? 0: strings.size();
        bytesLen += byteStream.write(listLen);
        for(int i = 0; i < listLen; ++i)
            bytesLen += byteStream.write(strings.get(i));
        return bytesLen;
    }

    public static int serializeLongList(List<Long> longs, ByteStream
            byteStream) {
        int bytesLen = 0;
        int listLen = longs == null? 0:longs.size();
        bytesLen += byteStream.write(listLen);
        for(int i = 0; i < listLen; ++i)
            bytesLen += byteStream.write(longs.get(i));
        return bytesLen;
    }

    public static int serializeGeoLocationList(List<GeoLocation> locs,
                                               ByteStream byteStream) {
        int bytesLen = 0;
        int listLen = locs==null? 0:locs.size();
        bytesLen += byteStream.write(listLen);
        for(int i = 0; i < listLen; ++i)
            bytesLen += locs.get(i).serialize(byteStream);
        return bytesLen;
    }

    public static ArrayList<Long> deserializeLongList(ByteStream byteStream) {
        int listSize = byteStream.readInt();
        ArrayList<Long> list = new ArrayList<>(listSize);

        for(int i = 0; i < listSize; ++i) {
            list.add(byteStream.readLong());
        }

        return list;
    }

    public static ArrayList<Integer> deserializeIntList(ByteStream byteStream) {
        int listSize = byteStream.readInt();
        ArrayList<Integer> list = new ArrayList<>(listSize);
        for(int i = 0; i < listSize; ++i)
            list.add(byteStream.readInt());
        return list;
    }

    public static ArrayList<String> deserializeStringList(ByteStream byteStream) {
        int listSize = byteStream.readInt();
        ArrayList<String> list = new ArrayList<>(listSize);

        for(int i = 0; i < listSize; ++i) {
            list.add(byteStream.readString());
        }

        return list;
    }

    public static ArrayList<GeoLocation> deserializeGeoLocationList(ByteStream
                                                                 byteStream) {
        int listSize = byteStream.readInt();
        ArrayList<GeoLocation> list = new ArrayList<>(listSize);

        for(int i = 0; i < listSize; ++i) {
            list.add(GeoLocation.deserialize(byteStream));
        }

        return list;
    }

    public static Pair<Integer, Integer> serializeSpatialEntry(
            Cache.Entry<Integer, SpatialPartition> entry, StreamDataset stream,
            ByteStream recordBytes, int entryInd) {
        int bytesLen = 0;
        bytesLen += serializeKeyGeo(entry.getKey(), recordBytes);
        Pair<Integer,Integer> results = serializeMemoryValue(
                entry.getValue().getData(), stream, recordBytes, entryInd);
        return new Pair<Integer,Integer>(results.getKey()+bytesLen,
                results.getValue());
    }

    public static Pair<Integer, ArrayList<Microblog>>
                            deserializeSpatialEntry(byte[] keyData,
                                                    Scheme scheme) {
        ByteStream byteStream = new ByteStream(keyData);
        Pair<Integer,Integer> result = deserializeKeyGeo(byteStream);
        Integer deserializedPartitionId = result.getKey();
        int deserializedLen = result.getValue();

        ArrayList<Microblog> microblogs = deserializeValue(byteStream, scheme);

        Pair<Integer, ArrayList<Microblog>> singleHashEntry = new Pair<>
        (deserializedPartitionId, microblogs);

        return singleHashEntry;
    }

    public static Hashtable<String,ArrayList<Microblog>> deserializeHashEntry(
            byte[] keyData, Scheme scheme) {

        ByteStream byteStream = new ByteStream(keyData);
        int offset = 0;
        Pair<String,Integer> result = deserializeKey(byteStream);
        String deserializedKey = result.getKey();
        int deserializedLen = result.getValue();
        offset = deserializedLen;

        ArrayList<Microblog> microblogs = deserializeValue(byteStream, scheme);

        Hashtable<String, ArrayList<Microblog>> singleHashEntry = new
                Hashtable<>();
        singleHashEntry.put(deserializedKey, microblogs);
        return singleHashEntry;
    }

    private static ArrayList<Microblog> deserializeValue(ByteStream
                                        byteStream, Scheme scheme)
    {
        ArrayList<Microblog> microblogs = new ArrayList<>();
        Microblog tmpMicroblog;
        while ((tmpMicroblog = Microblog.deserialize(byteStream,scheme))!=null)
            microblogs.add(tmpMicroblog);
        return microblogs;
    }

    private static Pair<String,Integer> deserializeKey(ByteStream byteStream) {
        int positionBefore = byteStream.getReadBytes();
        byte keyType = byteStream.readByte();
        String key = byteStream.readString();
        int positionAfter = byteStream.getReadBytes();
        int deserializedLen = positionAfter-positionBefore;

        return new Pair<>(key,deserializedLen);
    }

    private static Pair<Integer,Integer> deserializeKeyGeo(ByteStream byteStream)
    {
        int positionBefore = byteStream.getReadBytes();
        byte keyType = byteStream.readByte();
        Integer key = byteStream.readInt();
        int positionAfter = byteStream.getReadBytes();
        int deserializedLen = positionAfter-positionBefore;

        return new Pair<>(key,deserializedLen);
    }


    public static void serializeAttributeList(ArrayList<Attribute> attributes,
                                              ByteStream byteStream) {
        int listLen = attributes.size();
        byteStream.write(listLen);
        for(int i = 0; i < listLen; ++i)
            attributes.get(i).serialize(byteStream);
    }

    public static int serializeIntList(ArrayList<Integer> ints,
                                        ByteStream byteStream) {
        int bytesLen = 0;
        int listLen = ints == null? 0:ints.size();
        bytesLen += byteStream.write(listLen);
        for(int i = 0; i < listLen; ++i)
            bytesLen += byteStream.write(ints.get(i));
        return bytesLen;
    }

    public static ArrayList<Attribute> deserializeAttributeList(ByteStream
                                                                byteStream) {
        int listLen = byteStream.readInt();
        ArrayList<Attribute> list = new ArrayList<>(listLen);
        for(int i = 0; i < listLen; ++i)
            list.add(Attribute.deserialize(byteStream));
        return list;
    }
}
import edu.umn.cs.kite.common.DebugFlagger;
import edu.umn.cs.kite.common.KiteInstance;
import edu.umn.cs.kite.datamodel.Attribute;
import edu.umn.cs.kite.indexing.disk.DirectoryEntry;
import edu.umn.cs.kite.indexing.disk.DiskIndexSegment;
import edu.umn.cs.kite.indexing.memory.spatial.MemorySpatialIndexSegment;
import edu.umn.cs.kite.indexing.memory.spatial.SpatialIndex;
import edu.umn.cs.kite.indexing.memory.spatial.SpatialPartition;
import edu.umn.cs.kite.indexing.memory.spatial.SpatialPartitioner;
import edu.umn.cs.kite.querying.Query;
import edu.umn.cs.kite.streaming.StreamDataset;
import edu.umn.cs.kite.util.ConstantsAndDefaults;
import edu.umn.cs.kite.util.GeoLocation;
import edu.umn.cs.kite.util.KiteUtils;
import edu.umn.cs.kite.util.Rectangle;
import edu.umn.cs.kite.util.microblogs.Microblog;
import edu.umn.cs.kite.util.serialization.ByteStream;
import edu.umn.cs.kite.util.serialization.Serializer;
import javafx.util.Pair;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import javax.cache.Cache;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
                Pair<Integer,Integer> result = Serializer.serializeSpatialEntry
                        (entry, stream, indexSerializedBytes, entryInd);
                int bytesLength = result.getKey();
                int continuingEntryInd = result.getValue();
                serializedBytesCount += bytesLength;
                entryInd = continuingEntryInd;

                serializedKeys.add(new DirectoryEntry(null,entry.getKey(),
                        offset, bytesLength));
                offset = serializedBytesCount;

                //accumalted block size, flush it to HDFS
                if (serializedBytesCount >= singleFileSize) {
                    ArrayList<DirectoryEntry> blockDirectoryEntries =
                            writeBlockWithKeysGeo(serializedKeys,
                                    indexSerializedBytes);
                    directoryEntries.addAll(blockDirectoryEntries);
                    serializedKeys.clear();
                    serializedBytesCount = 0;
                    offset = 0;
                    indexSerializedBytes = new ByteStream(singleFileSize);
                }
            }
        }
        //write the final block
        ArrayList<DirectoryEntry> blockDirectoryEntries =
                writeBlockWithKeysGeo(serializedKeys, indexSerializedBytes);
        directoryEntries.addAll(blockDirectoryEntries);

        //write directory block
        writeDirectoryFile(directoryEntries);

        return true;
    }

    private String getSpatialPartitionerFileId() {
        return uniqueName+"_spatialPartitioner";
    }

    private String writeDirectoryFile(ArrayList<DirectoryEntry>
                                              directoryEntries)
            throws IOException {

        directoryFileFolder = getDirectoryId();
        directoryFileId =  getIndexDirectoryFileId();

        //serialize directory entries
        String tmp_string = directoryFileFolder + directoryFileId;
        double charByteRatio = tmp_string.getBytes().length*1.0/ tmp_string
                .length();

        int capacity = (int)(directoryEntries.size()*(uniqueName.length()+5
                +Integer.BYTES*2+14)*charByteRatio+200);
        ByteStream byteStream = new ByteStream(capacity);

        byteStream.write(directoryEntries.size());
        for(int i = 0; i < directoryEntries.size(); ++i) {
            byteStream.write(directoryEntries.get(i).getKeyGeo());
            byteStream.write(directoryEntries.get(i).getFilePath());
            byteStream.write(directoryEntries.get(i).getOffset());
            byteStream.write(directoryEntries.get(i).getLength());
        }

        KiteUtils.writeHDFSBlock(byteStream, directoryFileFolder,
                directoryFileId);
        return directoryFileFolder+directoryFileId;
    }

    private ArrayList<DirectoryEntry> loadDirectory(String dirFileParent,
                                                    String dirFileId) {
        byte[] directoryBytes = KiteUtils.readHDFSBlock(dirFileParent,
                dirFileId);

        ByteStream byteStream = new ByteStream(directoryBytes);

        int listLen = byteStream.readInt();
        ArrayList<DirectoryEntry> entries = new ArrayList<>(listLen);
        for(int i = 0; i < listLen; ++i) {
            Integer key = byteStream.readInt();
            String filePath = byteStream.readString();
            long offset = byteStream.readLong();
            int len = byteStream.readInt();

            DirectoryEntry entry=new DirectoryEntry(filePath, key, offset, len);
            entries.add(entry);
        }
        byteStream.clear();
        return entries;
    }

    private ArrayList<DirectoryEntry> writeBlockWithKeysGeo(
            ArrayList<DirectoryEntry> serializedKeys, ByteStream
            indexSerializedBytes)
            throws IOException {

        String parentDirectory = getDirectoryId();
        String fileId = getNextFileId();

        KiteUtils.writeHDFSBlock(indexSerializedBytes, parentDirectory, fileId);

        ArrayList<DirectoryEntry> directoryEntries = new ArrayList<>();
        for(DirectoryEntry entry: serializedKeys) {
            directoryEntries.add(new DirectoryEntry(fileId, entry.getKeyGeo(),
                    entry.getOffset(), entry.getLength()));
        }
        return directoryEntries;
    }

    private String getNextFileId() {
        String counterStr = KiteUtils.int2str(counter,4);
        ++counter;
        return uniqueName + "_" + counterStr;
    }

    private String getDirectoryId() {
        return ConstantsAndDefaults.DISK_INDEXES_ROOT_DIRECTORY +
                parentIndexUniqueName +"/" + uniqueName + "/";
    }

    @Override
    public ArrayList<Microblog> search(GeoLocation key, int k, Query query) {