<commit_before>#include "rdd.h"
#include "pack.h"
#include "time.h"
#include "sys/time.h"

/**
 * Interface for accessing result of Kraps query exectiion from Spark
 */
class KrapsIterator
{
  public:
    /**
     * Get next record
     * @return pointer to next record or NULL if there are no more records
     */
    virtual void* next() = 0;
    virtual ~KrapsIterator() {}
};


/**
 * Spark context for execution of Kraps query
 */
struct JavaContext 
{
    JNIEnv* env;
    jobjectArray inputs;

    JavaContext(JNIEnv* e, jobjectArray i) : env(e), inputs(i) {}
};

extern JavaVM* jvm;

inline time_t getCurrentTime()
{
    struct timeval tv;
    gettimeofday(&tv, NULL);
    return tv.tv_sec*1000 + tv.tv_usec/1000;
}

#define TILE_SIZE 128

/**
 * Proxy class for accessing Spark RDD from Kraps
 */
template<class T>
class SparkRDD : public RDD<T>
{
    T tile[TILE_SIZE];
    int size;
    int used;
    ThreadLocal<JNIEnv> context;
 public:
    /**
     * Get next row
     * @param row [out] placeholder for next row
     * @return true if there is next record, false otherwise
     */
    bool next(T& row)
    {
        if (used == size) {
            #ifdef MEASURE_SPARK_TIME
            time_t start = getCurrentTime();
            #endif
            JNIEnv* env = context.get();
            if (env == NULL) {
                jvm->AttachCurrentThread((void**)&env, NULL);
            }
            JavaContext* ctx = (JavaContext*)Cluster::instance->userData;
            jobject input = env->GetObjectArrayElement(ctx->inputs, inputNo);
            size = env->CallIntMethod(input, nextTile, (jlong)(size_t)tile, TILE_SIZE);
            used = 0;
            env->DeleteLocalRef(input);
            #ifdef MEASURE_SPARK_TIME
            elapsed += getCurrentTime() - start;
            calls += 1;
            #endif
            if (size == 0) {
                jvm->DetachCurrentThread();                
                return false;
            }
        }
        row = tile[used++];
        return true;
    }<commit_msg>Refactor FindShardIntervalInCache function<commit_after>#include "rdd.h"
#include "pack.h"
#include "time.h"
#include "sys/time.h"

/**
 * Interface for accessing result of Kraps query exectiion from Spark
 */
class KrapsIterator
{
  public:
    /**
     * Get next record
     * @return pointer to next record or NULL if there are no more records
     */
    virtual void* next() = 0;
    virtual ~KrapsIterator() {}
};


/**
 * Spark context for execution of Kraps query
 */
struct JavaContext 
{
    JNIEnv* env;
    jobjectArray inputs;

    JavaContext(JNIEnv* e, jobjectArray i) : env(e), inputs(i) {}
};

extern JavaVM* jvm;

inline time_t getCurrentTime()
{
    struct timeval tv;
    gettimeofday(&tv, NULL);
    return tv.tv_sec*1000 + tv.tv_usec/1000;
}

#define TILE_SIZE 128

/**
 * Proxy class for accessing Spark RDD from Kraps
 */
template<class T>
class SparkRDD : public RDD<T>
{
    T tile[TILE_SIZE];
    int size;
    int used;
    ThreadLocal<JNIEnv> context;
 public:
    /**
     * Get next row
     * @param row [out] placeholder for next row
     * @return true if there is next record, false otherwise
     */
    bool next(T& row)
    {
        if (used == size) {
            #ifdef MEASURE_SPARK_TIME
            time_t start = getCurrentTime();
            #endif
            JNIEnv* env = context.get();
            if (env == NULL) {
                jvm->AttachCurrentThread((void**)&env, NULL);
				context = env;
            }
            JavaContext* ctx = (JavaContext*)Cluster::instance->userData;
            jobject input = env->GetObjectArrayElement(ctx->inputs, inputNo);
            size = env->CallIntMethod(input, nextTile, (jlong)(size_t)tile, TILE_SIZE);
            used = 0;
            env->DeleteLocalRef(input);
            #ifdef MEASURE_SPARK_TIME
            elapsed += getCurrentTime() - start;
            calls += 1;
            #endif
            if (size == 0) {
                jvm->DetachCurrentThread();                
                return false;
            }
        }
        row = tile[used++];
        return true;
    }
{'fraction_non_alphanumeric': 0.11369134010643445, 'fraction_numerical': 0.00725689404934688, 'mean_word_length': 2.049410029498525, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2527591', 'n_tokens_mistral': 1330, 'n_tokens_neox': 1267, 'n_words': 352}