// Path: src/main/java/be/bagofwords/db/CoreDataInterface.java
// public abstract class CoreDataInterface<T> extends BaseDataInterface<T> {
// 
//     protected final UpdateListenerCollection<T> updateListenerCollection = new UpdateListenerCollection<>();
// 
//     public CoreDataInterface(String name, Class<T> objectClass, Combinator<T> combinator, ObjectSerializer<T> objectSerializer, boolean isTemporary) {
//         super(name, objectClass, combinator, objectSerializer, isTemporary);
//     }
// 
//     @Override
//     public DataInterface<T> getCoreDataInterface() {
//         return this;
//     }
// 
//     @Override
//     public void registerUpdateListener(UpdateListener<T> updateListener) {
//         updateListenerCollection.registerUpdateListener(updateListener);
//     }
// 
//     @Override
//     public void flush() {
//         flushImpl();
//         updateListenerCollection.dataFlushed();
//     }
// 
//     protected abstract void flushImpl();
// 
// }
// 
// Path: src/main/java/be/bagofwords/db/combinator/Combinator.java
// public interface Combinator<T extends Object> extends Serializable {
// 
//     T combine(T first, T second);
// 
//     default void addRemoteClasses(RemoteObjectConfig objectConfig) {
//         //Don't add any classes by default
//     }
// 
//     default RemoteObjectConfig createExecConfig() {
//         RemoteObjectConfig result = RemoteObjectConfig.create(this);
//         result.add(getClass());
//         addRemoteClasses(result);
//         return result;
//     }
// }
// 
// Path: src/main/java/be/bagofwords/db/methods/ObjectSerializer.java
// public interface ObjectSerializer<T> extends Serializable {
// 
//     void writeValue(T obj, DataStream ds);
// 
//     T readValue(DataStream ds, int size);
// 
//     int getObjectSize();
// 
//     default RemoteObjectConfig createExecConfig() {
//         return RemoteObjectConfig.create(this).add(getClass());
//     }
// }


import be.bagofwords.db.CoreDataInterface;
import be.bagofwords.db.combinator.Combinator;
import be.bagofwords.db.combinator.LongCombinator;
import be.bagofwords.db.methods.ObjectSerializer;
import be.bagofwords.iterator.CloseableIterator;
import be.bagofwords.iterator.IterableUtils;
import be.bagofwords.util.DataLock;
import be.bagofwords.util.KeyValue;
import be.bagofwords.util.SerializationUtils;
import kyotocabinet.Cursor;
import kyotocabinet.DB;
import java.io.File;
import java.util.*;
                    cursor.disable();
                }
            }

        };
    }

    @Override
    public CloseableIterator<KeyValue<T>> iterator(CloseableIterator<Long> keyIterator) {
        return new CloseableIterator<KeyValue<T>>() {

            private Iterator<KeyValue<T>> valueBatch;

            {
                findNextBatch();
            }

            private void findNextBatch() {
                List<Long> keys = new ArrayList<>();
                while (keyIterator.hasNext() && keys.size() < BATCH_SIZE) {
                    keys.add(keyIterator.next());
                }
                if (!keys.isEmpty()) {
                    byte[][] keysAsBytes = new byte[keys.size()][];
                    for (int i = 0; i < keys.size(); i++) {
                        keysAsBytes[i] = SerializationUtils.longToBytes(keys.get(i));
                    }
                    byte[][] keysAndValues = db.get_bulk(keysAsBytes, false);
                    Map<Long, T> parsedValues = new HashMap<>();
                    for (int i = 0; i < keysAndValues.length; i += 2) {
// Path: src/main/java/be/bagofwords/db/CoreDataInterface.java
// public abstract class CoreDataInterface<T> extends BaseDataInterface<T> {
// 
//     protected final UpdateListenerCollection<T> updateListenerCollection = new UpdateListenerCollection<>();
// 
//     public CoreDataInterface(String name, Class<T> objectClass, Combinator<T> combinator, ObjectSerializer<T> objectSerializer, boolean isTemporary) {
//         super(name, objectClass, combinator, objectSerializer, isTemporary);
//     }
// 
//     @Override
//     public DataInterface<T> getCoreDataInterface() {
//         return this;
//     }
// 
//     @Override
//     public void registerUpdateListener(UpdateListener<T> updateListener) {
//         updateListenerCollection.registerUpdateListener(updateListener);
//     }
// 
//     @Override
//     public void flush() {
//         flushImpl();
//         updateListenerCollection.dataFlushed();
//     }
// 
//     protected abstract void flushImpl();
// 
// }
// 
// Path: src/main/java/be/bagofwords/db/combinator/Combinator.java
// public interface Combinator<T extends Object> extends Serializable {
// 
//     T combine(T first, T second);
// 
//     default void addRemoteClasses(RemoteObjectConfig objectConfig) {
//         //Don't add any classes by default
//     }
// 
//     default RemoteObjectConfig createExecConfig() {
//         RemoteObjectConfig result = RemoteObjectConfig.create(this);
//         result.add(getClass());
//         addRemoteClasses(result);
//         return result;
//     }
// }
// 
// Path: src/main/java/be/bagofwords/db/methods/ObjectSerializer.java
// public interface ObjectSerializer<T> extends Serializable {
// 
//     void writeValue(T obj, DataStream ds);
// 
//     T readValue(DataStream ds, int size);
// 
//     int getObjectSize();
// 
//     default RemoteObjectConfig createExecConfig() {
//         return RemoteObjectConfig.create(this).add(getClass());
//     }
// }


// Path: count-db-run/src/main/java/be/bagofwords/db/interfaces/kyoto/KyotoDataInterface.java
import be.bagofwords.db.CoreDataInterface;
import be.bagofwords.db.combinator.Combinator;
import be.bagofwords.db.combinator.LongCombinator;
import be.bagofwords.db.methods.ObjectSerializer;
import be.bagofwords.iterator.CloseableIterator;
import be.bagofwords.iterator.IterableUtils;
import be.bagofwords.util.DataLock;
import be.bagofwords.util.KeyValue;
import be.bagofwords.util.SerializationUtils;
import kyotocabinet.Cursor;
import kyotocabinet.DB;
import java.io.File;
import java.util.*;

                    cursor.disable();
                }
            }

        };
    }

    @Override
    public CloseableIterator<KeyValue<T>> iterator(CloseableIterator<Long> keyIterator) {
        return new CloseableIterator<KeyValue<T>>() {

            private Iterator<KeyValue<T>> valueBatch;

            {
                findNextBatch();
            }

            private void findNextBatch() {
                List<Long> keys = new ArrayList<>();
                while (keyIterator.hasNext() && keys.size() < BATCH_SIZE) {
                    keys.add(keyIterator.next());
                }
                if (!keys.isEmpty()) {
                    byte[][] keysAsBytes = new byte[keys.size()][];
                    for (int i = 0; i < keys.size(); i++) {
                        keysAsBytes[i] = SerializationUtils.longToBytes(keys.get(i));
                    }
                    byte[][] keysAndValues = db.get_bulk(keysAsBytes, false);
                    Map<Long, T> parsedValues = new HashMap<>();
                    for (int i = 0; i < keysAndValues.length; i += 2) {
                        long key = SerializationUtils.bytesToLong(keysAndValues[i]);