Optimizing speed

def query_iterator(file_paths: list[str]) -> Iterator[str]:
    for path in file_paths:
        with open(path, "r") as file:
            for line in file:
                yield line


def batch_iterator(file_paths: list[str], batch_size: int) -> Iterator[list[str]]:
    next_batch: list[str] = list()

    for query in query_iterator(file_paths):
        next_batch.append(query)

        if len(next_batch) >= batch_size:
            yield next_batch
            next_batch: list[str] = list()

    if len(next_batch) > 0:
        yield next_batch

class QueryIterator implements AsyncIterable<string> {
    private filepaths: Array<string>;
    constructor(filepaths: Array<string>) {
        this.filepaths = filepaths;
    }

    async *[Symbol.asyncIterator](): AsyncIterableIterator<string> {

        for (let filepath of this.filepaths) {
            let file = readline.createInterface(fs.createReadStream(filepath));

            for await (let query of file) {
                yield query;
            }
        }
    }
}


class BatchIterator implements AsyncIterable<Array<string>> {
    private filepaths: Array<string>;
    private batchSize: number;

    constructor(filepaths: Array<string>, batchSize: number) {
        this.filepaths = filepaths;
        this.batchSize = batchSize;
    }

    async *[Symbol.asyncIterator](): AsyncIterator<Array<string>> {
        let queryIterator = new QueryIterator(this.filepaths);
        let nextBatch: Array<string> = [];

        for await (let query of queryIterator) {
            nextBatch.push(query);

            if (nextBatch.length >= this.batchSize) {
                yield nextBatch;
                nextBatch = [];
            }
        }

        if (nextBatch.length > 0) { yield nextBatch }
    }
}

public class QueryIterator implements Iterator<String> {
    public final List<String> filepaths;
    private Iterator<String> lineIterator;
    private boolean finished;

    public QueryIterator(ArrayList<String> filepaths) throws FileNotFoundException {
        this.filepaths = filepaths;
        finished = false;
        prepNextIterator();
    }

    private static InputStream inputStream(String filepath) throws FileNotFoundException {
        try {
            return new BufferedInputStream(new FileInputStream(filepath));
        } catch (IOException e) {
            throw new FileNotFoundException();
        }
    }

    private void prepNextIterator() throws FileNotFoundException {
        if (this.filepaths.isEmpty()) {
            finished = true;
        } else {
            InputStream inputStream = inputStream(filepaths.remove(0));
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream, UTF_8));
            lineIterator = bufferedReader.lines().iterator();
            if (!lineIterator.hasNext()) prepNextIterator();
        }
    }

    @Override
    public boolean hasNext() {
        return !finished;
    }

    @Override
    public String next() {
        String nextItem = lineIterator.next();

        if (!lineIterator.hasNext()) {
            try {
                this.prepNextIterator();
            } catch (FileNotFoundException e) {
                throw new RuntimeException(e);
            }
        }

        return nextItem;
    }
}


public class BatchIterator implements Iterator<List<String>> {
    public int batchSize;
    private final Iterator<String> queries;

    public BatchIterator(ArrayList<String> filepaths, int batchSize) throws FileNotFoundException {
        this.batchSize = batchSize;
        this.queries = new QueryIterator(filepaths);
    }

    @Override
    public boolean hasNext() { return queries.hasNext(); }

    @Override
    public List<String> next() {
        if (!this.hasNext()) throw new NoSuchElementException();
        List<String> batch = new ArrayList<>();

        while (queries.hasNext() && batch.size() < batchSize) {
            batch.add(queries.next());
        }

        return batch;
    }
}

def load_batch(session: TypeDBSession, batch: list[str]) -> None:
    with session.transaction(TransactionType.WRITE) as transaction:
        for query in batch:
            transaction.query.insert(query)

        transaction.commit()


def load_data(addresses: str | list[str], username: str, password: str) -> None:
    database = "bookstore"
    data_files = ["contributors.tql", "publishers.tql", "books.tql"]
    batch_size = 100
    credential = TypeDBCredential(username, password, tls_enabled=True)

    with TypeDB.cloud_driver(addresses, credential) as driver:
        with driver.session(database, SessionType.DATA) as session:
            for batch in batch_iterator(data_files, batch_size):
                load_batch(session, batch)

async function loadBatch(batch: Array<string>, session: TypeDBSession): Promise<void> {
    let transaction: TypeDBTransaction;

    try {
        transaction = await session.transaction(TransactionType.WRITE);

        for (let query of batch) {
            transaction.query.insert(query)
        }

        await transaction.commit();
    } finally {
        if (transaction.isOpen()) {
            await transaction.close();
        }
    }
}


async function loadData(addresses: Array<string>, username: string, password: string): Promise<void> {
    const database: string = "bookstore";
    const dataFiles: Array<string> = ["contributors.tql", "publishers.tql", "books.tql"];
    const batchSize: number = 100;

    let credential: TypeDBCredential = new TypeDBCredential(username, password);
    let driver: TypeDBDriver;

    try {
        driver = await TypeDB.cloudDriver(addresses, credential);
        let session: TypeDBSession;

        try {
            session = await driver.session(database, SessionType.DATA);
            let batchIterator = new BatchIterator(dataFiles, batchSize);

            for await (let batch of batchIterator) {
                await loadBatch(batch, session);
            }
        } finally { await session?.close() }
    } finally { await driver?.close() }
}

public static void loadBatch(TypeDBSession session, List<String> batch) {
    try (TypeDBTransaction transaction = session.transaction(TypeDBTransaction.Type.WRITE)) {
        for (String query: batch) {
            transaction.query().insert(query);
        }
        transaction.commit();
    }
}


public static void loadData(Set<String> addresses, String username, String password) throws FileNotFoundException {
    String database = "bookstore";
    ArrayList<String> dataFiles = new ArrayList<>(List.of("contributors.tql", "publishers.tql", "books.tql"));
    int batchSize = 100;
    TypeDBCredential credential = new TypeDBCredential(username, password, true);

    try (TypeDBDriver driver = TypeDB.cloudDriver(addresses, credential)) {
        try (TypeDBSession session = driver.session(database, TypeDBSession.Type.DATA)) {
            BatchIterator batchIterator = new BatchIterator(dataFiles, batchSize);

            while (batchIterator.hasNext()) {
                List<String> batch = batchIterator.next();
                loadBatch(session, batch);
            }
        }
    }
}

def batch_loader(
    queue: Queue,
    addresses: str | list[str],
    username: str,
    password: str,
    database: str
) -> None:
    credential = TypeDBCredential(username, password, tls_enabled=True)

    with TypeDB.cloud_driver(addresses, credential) as driver:
        with driver.session(database, SessionType.DATA) as session:
            while True:
                batch: list[str] | None = queue.get()

                if batch is None:
                    break
                else:
                    with session.transaction(TransactionType.WRITE) as transaction:
                        for query in batch:
                            transaction.query.insert(query)

                        transaction.commit()


def load_data_async(addresses: str | list[str], username: str, password: str) -> None:
    database = "bookstore"
    data_files = ["contributors.tql", "publishers.tql", "books.tql"]
    batch_size = 100

    for data_file in data_files:
        with Manager() as manager:
            pool_size = os.cpu_count()
            pool = Pool(pool_size)
            queue = manager.Queue(4 * pool_size)

            kwargs = {
                "queue": queue,
                "addresses": addresses,
                "username": username,
                "password": password,
                "database": database,
            }

            for _ in range(pool_size):
                pool.apply_async(batch_loader, kwds=kwargs)

            for batch in batch_iterator([data_file], batch_size):
                queue.put(batch)

            for _ in range(pool_size):
                queue.put(None)  # Sentinel value

            pool.close()
            pool.join()

class PromisePool<T> {
    public size: number;
    public promises: Array<Promise<{index: number; result: T}>>;
    constructor(size: number) {
        this.size = size;
        this.promises = [];
    }

    async put(value: Promise<T>): Promise<void | T> {
        if (this.promises.length < this.size) {
            let index = this.promises.length;
            this.promises.push(value.then(result => ({index: index, result})));
            return;
        } else {
            let {index, result} = await Promise.race(this.promises)
            this.promises[index] = value.then(result => ({index: index, result}));
            return result;
        }
    }

    async join(): Promise<Array<T>> {
        let results: Array<T> = [];
        let outputs = await Promise.all(this.promises)

        for (let output of outputs) {
            results.push(output.result)
        }

        return results;
    }
}


async function loadDataAsync(addresses: Array<string>, username: string, password: string): Promise<void> {
    const database: string = "bookstore";
    const dataFiles: Array<string> = ["contributors.tql", "publishers.tql", "books.tql"];
    const batchSize: number = 100;

    let credential: TypeDBCredential = new TypeDBCredential(username, password);
    let driver: TypeDBDriver;

    try {
        driver = await TypeDB.cloudDriver(addresses, credential);
        let session: TypeDBSession;

        try {
            session = await driver.session(database, SessionType.DATA);

            for (let dataFile of dataFiles) {
                let poolSize = os.cpus().length;
                let promisePool = new PromisePool(poolSize);
                let batchIterator = new BatchIterator([dataFile], batchSize);

                for await (let batch of batchIterator) {
                    await promisePool.put(loadBatch(batch, session));
                }

                await promisePool.join();
            }
        } finally { await session?.close() }
    } finally { await driver?.close() }
}

public static CompletableFuture<Void> scheduleBatchLoader(
        ExecutorService executor,
        LinkedBlockingQueue<Either<List<String>, NullType>> queue,
        TypeDBSession session,
        AtomicBoolean hasError
) {
    return CompletableFuture.runAsync(() -> {
        Either<List<String>, NullType> queries;
        try {
            while ((queries = queue.take()).isFirst() && !hasError.get()) {
                try (TypeDBTransaction transaction = session.transaction(TypeDBTransaction.Type.WRITE)) {
                    for (String query: queries.first()) {
                        transaction.query().insert(query);
                    }
                    transaction.commit();
                }
            }
        } catch (Throwable e) {
            hasError.set(true);
            throw new RuntimeException(e);
        }
    }, executor);
}


public static void loadDataAsync(Set<String> addresses, String username, String password) throws InterruptedException, FileNotFoundException {
    String database = "bookstore";
    ArrayList<String> dataFiles = new ArrayList<>(List.of("contributors.tql", "publishers.tql", "books.tql"));
    int batchSize = 100;
    TypeDBCredential credential = new TypeDBCredential(username, password, true);

    try (TypeDBDriver driver = TypeDB.cloudDriver(addresses, credential)) {
        try (TypeDBSession session = driver.session(database, TypeDBSession.Type.DATA)) {
            for (String dataFile: dataFiles) {
                int poolSize = Runtime.getRuntime().availableProcessors();
                ExecutorService executor = Executors.newFixedThreadPool(poolSize, new NamedThreadFactory(database));
                LinkedBlockingQueue<Either<List<String>, NullType>> queue = new LinkedBlockingQueue<>(4 * poolSize);
                List<CompletableFuture<Void>> batchLoadersFutures = new ArrayList<>(poolSize);
                AtomicBoolean hasError = new AtomicBoolean(false);

                for (int i = 0; i < poolSize; i++) {
                    batchLoadersFutures.add(scheduleBatchLoader(executor, queue, session, hasError));
                }

                BatchIterator batchIterator = new BatchIterator(new ArrayList<>(List.of(dataFile)), batchSize);

                while (batchIterator.hasNext() && !hasError.get()) {
                    List<String> batch = batchIterator.next();
                    queue.put(Either.first(batch));
                }

                for (int i = 0; i < poolSize; i++) {
                    queue.put(Either.second(null));
                }

                CompletableFuture.allOf(batchLoadersFutures.toArray(new CompletableFuture[0])).join();
                executor.shutdown();
            }
        }
    }
}