DEV Community: Maciej Kucharski

Streaming data from RDBMS with jooq

Maciej Kucharski — Sun, 23 Feb 2025 19:35:30 +0000

Introduction

Sometimes, it's necessary to process a large number of rows fetched from the database. When the row count reaches millions, things can get complicated, especially regarding memory usage. While implementing paging and processing rows page by page is a common approach, there are alternative solutions.

Relational databases provide components such as cursors, which can be used to process large datasets while fetching only a limited number of rows at a time. With the help of jooq framework,we can take it a step further—leveraging cursors under the hood while seamlessly integrating with Java's Stream API. Let’s see it in action!

Project setup

For the sake of a clean demonstration, I’ll keep things as simple as possible. The database will be set up using SQL scripts to avoid introducing any migration tools. It will contain 5 million rows, each with a few columns mimicking real-life data. I won’t use jOOQ’s type-safe query generation features (though they’re excellent—I highly recommend trying jOOQ) to keep the focus on one specific aspect. I’m using Spring Boot, but its usage is kept to a minimum. I couldn’t resist lombok, thou :)

Database

Database for this demo is initialized using a simple SQL script:

create table sample_data
(
    id             serial primary key,
    start_date     date      not null,
    type_id        integer   not null,
    description    text      not null,
    category_id    integer   not null,
    subcategory_id integer   not null,
    created_at     timestamp not null default current_timestamp
);

insert into sample_data
select id,
       date '2024-01-01' + (random() * 365)::int,
       round((random() * 15)::numeric, 2),
       md5(random()::text),
       round((random() * 30)::numeric, 2),
       round((random() * 200)::numeric, 2)
from generate_series(1, 5000000) as id;

A row from this database table is mapped to a Java record with the following structure:

@Builder
public record SampleDataEntity(int id, String description, LocalDate startDate, LocalDateTime createdAt, int typeId,
                               int categoryId, int subcategoryId) {
}

Standard way

Let’s start by evaluating the memory footprint in a scenario where all the data is loaded into memory. The repository implementation is straightforward:

@Repository
@FieldDefaults(makeFinal = true, level = AccessLevel.PRIVATE)
@RequiredArgsConstructor
public class SomeRepository {
    DSLContext dslContext;

    Collection<SampleDataEntity> fetchAll() {
        return dslContext.fetch("select * from sample_data")
                .stream()
                .map(SomeRepository::mapRow)
                .toList();
    }

    private static SampleDataEntity mapRow(org.jooq.Record row) {
        return SampleDataEntity.builder()
                .id(row.get("id", int.class))
                .description(row.get("description", String.class))
                .createdAt(row.get("created_at", LocalDateTime.class))
                .startDate(row.get("start_date", LocalDate.class))
                .typeId(row.get("type_id", int.class))
                .categoryId(row.get("category_id", int.class))
                .subcategoryId(row.get("subcategory_id", int.class))
                .build();
    }
}

For the purposes of this demo, let’s simply count the number of retrieved rows. The Thread.sleep() is included to give me time to open VisualVM and check the metrics:

@Component
@FieldDefaults(level = AccessLevel.PRIVATE, makeFinal = true)
@RequiredArgsConstructor
@Slf4j
public class SimpleRunner implements CommandLineRunner {

    SomeRepository someRepository;

    @Override
    public void run(String... args) throws Exception {
        log.info("Runner started with args={}", args);
        Thread.sleep(Duration.ofSeconds(10));
        var count = someRepository.fetchAll()
                    .size();
            log.info("Fetched {} rows", count);
    }
}

What are the results? The program runs smoothly but consumes around 5 GB of memory:

Cursor way

Let’s modify our repository slightly to use a variant of the jOOQ API that returns a stream:

@Repository
@FieldDefaults(makeFinal = true, level = AccessLevel.PRIVATE)
@RequiredArgsConstructor
public class SomeRepository {
    DSLContext dslContext;

    @SneakyThrows
    Stream<SampleDataEntity> streamAll() {
        return dslContext.fetchStream("select * from sample_data")
                .map(SomeRepository::mapRow);
    }

    private static SampleDataEntity mapRow(org.jooq.Record row) {
        return SampleDataEntity.builder()
                .id(row.get("id", int.class))
                .description(row.get("description", String.class))
                .createdAt(row.get("created_at", LocalDateTime.class))
                .startDate(row.get("start_date", LocalDate.class))
                .typeId(row.get("type_id", int.class))
                .categoryId(row.get("category_id", int.class))
                .subcategoryId(row.get("subcategory_id", int.class))
                .build();
    }
}

I’ll skip pasting the runner implementation since the only difference is in how the number of items is counted. So, how does the memory graph look this time? A significant improvement is immediately noticeable:

The numbers look impressive, but there’s some bad news—the entire result set is still loaded into memory. The improvement comes from streaming over the mapped results, making them garbage-collection eligible earlier. How do we know this? It will become clear once we try a different API that allows us to control the number of fetched rows:

    @SneakyThrows
    Stream<SampleDataEntity> streamAll() {
        dslContext.settings().setFetchSize(100);
        var cursor = dslContext.fetchLazy("select * from sample_data");
        return cursor.stream()
                .map(SomeRepository::mapRow);
    }

Running the program will produce the following log entry:
org.jooq.impl.AbstractResultQuery : Fetch Size : A fetch size of 100 was set on a auto-commit PostgreSQL connection, which is not recommended. See https://jdbc.postgresql.org/documentation/query/#getting-results-based-on-a-cursor

The documentation page linked above outlines several scenarios where the JDBC driver ignores fetchSize and loads the entire result set into memory. In this case, here’s what happens:

The Connection must not be in autocommit mode. The backend closes cursors at the end of transactions, so in autocommit mode the backend will have closed the cursor before anything can be fetched from it.

Cursor - the proper way

Now that we know what’s wrong, let’s fix it! First, we need to disable auto-commit. This can be done by setting a property on the datasource:

spring.datasource.hikari.auto-commit=false

Even though we’re only reading data and not writing, a transaction is now required. For the sake of this demo, let’s use a well-known annotation to handle it. Our modified runner code now looks like this:

@Component
@FieldDefaults(level = AccessLevel.PRIVATE, makeFinal = true)
@RequiredArgsConstructor
@Slf4j
public class StreamingRunner implements CommandLineRunner {

    SomeRepository repository;

    @Override
    @Transactional
    public void run(String... args) throws Exception {
        log.info("Streaming started with args={}", args);
        Thread.sleep(Duration.ofSeconds(10));

        var count = repository.streamAll()
                    .count();
        log.info("Found {} rows", count);

    }
}

What’s the result? I’d say the application now consumes a reasonable amount of memory relative to its complexity:

Conclusion

We’ve just learned another solution for situations where our application needs to process large amounts of data. I see this approach as more useful for batch processes that handle and aggregate vast datasets—for example, for reporting—rather than for serving data directly to the UI. With the significant reduction in memory consumption, it may even be possible to use the saved resources to spin up more instances, improving overall process throughput.

Final source code can be found on GitHub

Implementing passwordless login with Amazon Cognito

Maciej Kucharski — Thu, 16 Jan 2025 20:15:29 +0000

Intro

I'm going to skip the introduction and description of what Cognito is and why it's useful—I’ll assume you have a reason for reading this.

I’ll use Node.js with AWS SDK V3 for the code snippets, but the language doesn’t matter—you should easily find equivalents in other SDKs.

Please note that this is not a ready-to-use solution you can copy-paste into your app. Instead, think of it as a set of building blocks you can use to design an authentication flow that meets your specific requirements.

This post is based on the latest updates in Cognito (as of around November 2024), which no longer require implementing custom challenges. So, let’s get started!

Setup user pool

The first thing you need is a user pool. Setting up passwordless login is quite straightforward. In the pool's overview, you should see a tile similar to the one shown below:

Next, navigate to the configuration for choice-based sign-in:

Enable one-time password (OTP) by selecting the appropriate checkbox:

For basic usage (perhaps for evaluation purposes), simply select Send email with Cognito and click Save changes:

For more advanced use cases, setting up Amazon SES is required. However, this topic will not be covered in this post.

Setup sdk

We need the aws-sdk library to interact with AWS services:
npm install @aws-sdk/client-cognito-identity-provider --save

Once the SDK is downloaded, set up the client. The snippet below includes all the imports used throughout this post:

import {
    CognitoIdentityProviderClient,
    InitiateAuthCommand,
    RespondToAuthChallengeCommand,
    AdminGetUserCommand,
    AdminCreateUserCommand
} from "@aws-sdk/client-cognito-identity-provider";

const client = new CognitoIdentityProviderClient({region: "eu-north-1"});

Check existence of user

Cognito sends emails only to users that exist in the pool. Therefore, we must first check if this is the case. There is no need to verify the contents of the response, as the call will throw an exception for a non-existing user:

async function isExistingUser(username) {
    try {
        const getUserCommand = new AdminGetUserCommand({Username: username, UserPoolId: USER_POOL_ID})
        const getUserResponse = await client.send(getUserCommand);
        return true;
    } catch (err) {
        if (err.name === "UserNotFoundException") {
            return false;
        } else {
            console.error("Unexpected error when verifying existence of user", err)
            throw err;
        }
    }
}

Depending on your requirements, you may want to redirect the user to the dedicated sign-up process, which will result in account creation, and continue later. In my case, it was sufficient to simply create a new user:

const createUserCommand = new AdminCreateUserCommand({
        Username: username,
        UserPoolId: USER_POOL_ID,
        MessageAction: "SUPPRESS"
    })
    const userCreateResponse = await client.send(createUserCommand);

Pay attention to the value of the MessageAction property. By default, Cognito will automatically initiate the user verification flow, resulting in a welcome message being sent to the user. I didn't want that, so I decided to suppress it.

Trigger email

In this demo, the user will receive a verification code via email, which must be manually copied and entered into the application. If you would like to include a hyperlink that, for example, a mobile application can open and handle automatically, you will need to set up SES, which is beyond the scope of this post.

const initAuthCommand = new InitiateAuthCommand({
    AuthFlow: "USER_AUTH",
    ClientId: CLIENT_ID,
    AuthParameters: {"USERNAME": username, "PREFERRED_CHALLENGE": "EMAIL_OTP"},
})

const initResult = await client.send(initAuthCommand);

Obtain access token

To receive an access token, we need to exchange the one-time password. This is done by responding to an authentication challenge:

const challengeCommand = new RespondToAuthChallengeCommand({
    ChallengeName: "EMAIL_OTP",
    ClientId: CLIENT_ID,
    Session: initResult.Session,
    ChallengeResponses: {
        "EMAIL_OTP_CODE": code,
        "USERNAME": username
    }
})

const { AuthenticationResult } = await client.send(challengeCommand)

console.log(AuthenticationResult.AccessToken, AuthenticationResult.RefreshToken)

That's it! The AuthenticationResult object contains the AccessToken and RefreshToken, which should now be used just like in any other application that uses tokens for authentication.

Summary

As we all know, typing a password on a phone is something that nobody enjoys. I hope this post has provided you with insight into how a more convenient login flow can be implemented using Cognito for authentication management.