Forem: Francesco Tisiot

Introducing Developer Tier for Aiven for PostgreSQL® services

Francesco Tisiot — Wed, 12 Nov 2025 14:10:00 +0000

Starting at $8 USD, the new Developer tier includes everything from the Free tier, with extra disk space, preserved uptime for idle services, and Basic support to keep you building without interruption.

Aiven is introducing a new pricing plan for Aiven for PostgreSQL® services. Starting at $8 USD per month, the Developer tier offers more storage, so you can scale up your free PostgreSQL service in a cost-effective way. Unlike the Free tier, services on the Developer tier are not automatically powered off if inactive. The Developer tier also automatically includes Basic support services.

More information on the Developer tier is available in the Aiven docs.

What do I get with the Developer tier?

Compared to the Free tier, you still get a single node, with 1 CPU per virtual machine, monitoring and backups. But beyond that:

Increased storage from 1GB to a maximum of 8GB.
Basic tier support.
Inactive services don’t automatically power off, meaning that your database will always be up and running, even if there’s no traffic to it. See the documentation for full technical details.

If your database needs are more specific, you can easily switch and scale to the Professional plans, where you get additional benefits including the ability to choose a specific cloud provider and region for your service, the option for more nodes, even more disk space, longer backup, integrations with other services, forking and much more. Find out more in the Aiven for PostgreSQL documentation.

Why three tiers?

We originally started out with our selection of paid service plans, which you can now find neatly grouped under the Professional tier.

We introduced the Free tier as a way of paying back to our community, providing a way to start and host your non critical PostgreSQL infrastructure on Aiven, and creating a great way to learn about using a database in the cloud.

Now we’ve introduced the Developer tier as a cost-effective choice between those two. The Developer tier inherits some traits from the free tier and enhances it with some features you might find useful for your more advanced needs at a lower price point compared to our Professional plans.

You can choose the balance that suits your project:

Free to explore and learn the platform and PostgreSQL in the cloud at no cost.
Developer as a cost-effective option for test and personal projects.
Professional for highly available business-critical workloads. And as your project grows, there’s always an upgrade path available.

Questions you might ask

Do I get to choose where the service runs?

As with the Free tier, you can choose the geographical area (Asia Pacific, Australia, Europe or North America), but not the cloud provider or region. If that’s a requirement, the selection is available in the Professional level plans.

Can I go (back) to the Free tier?

If your service data will fit in the smaller instance size (1GB disk instead of 8GB), then it’s always possible to downgrade a service to the Free tier. See the documentation for how to Change a service plan.

Can I upgrade to the Professional tier?

Yes, it’s always possible to upgrade to a higher tier. See the documentation for how to Change a service plan.

Should I use a Developer tier service in production?

As it says in the description in the Aiven Console, the Developer tier is intended to be “A cost-effective option for test and personal projects”.

The Developer tier provides a working PostgreSQL environment with basic daily backup and Basic level support, but without advanced networking, point in time recovery, forking, integrations, or cloud provider and region selection. If any of the above features are interesting for you, then please look in the Professional plans.

Why is this just for PostgreSQL?

We started with adding this new tier for Aiven for PostgreSQL because we’re confident that our users will find it useful. We do have ideas for other services - keep tuned to see what we announce in the future.

Try it out!

If you’re creating a new Aiven for PostgreSQL service and know that a Free service isn’t enough, but you don’t need all the resources of the Professional “Hobbyist” plan, then start a Developer service instead.

If you’ve got an existing Free tier Aiven for PostgreSQL service that could use just that bit more disk space, or that you only use intermittently, or if you want some support, then consider upgrading to the Developer tier.

And of course, if you’ve got an existing Hobbyist service that feels over-provisioned, you’ve now got the option of downgrading to Developer (although please do read the documentation to check that everything you need is still provided).

Find out more in the Aiven Console, or check out the documentation.

Free PostgreSQL® in the Cloud - Aug 25

Francesco Tisiot — Thu, 14 Aug 2025 08:20:11 +0000

A short summary of Free PostgreSQL® services in the cloud as of August 14th 2025

Provider	Free Tier Highlights	Best For
Aiven	1 CPU, 1 GB RAM, 1GB Storage	Production-like prototyping
Neon	10 projects, 0.5 GB storage, serverless, branching	Modern development workflows, serverless apps
Fly.io	3 GB storage, 160 GB outbound	Lightweight global apps
Render.com	0.1 CPU, 256 RAM, 100 connections, daily backups, 30 day limit	Small apps needing more capacity
Amazon RDS	20 GB + backups, 750 hrs/month	Temporary usage via AWS Free Tier
Supabase	500 MB storage	Small prototypes

Helping PostgreSQL® professionals with AI-assisted performance recommendations

Francesco Tisiot — Tue, 28 May 2024 15:00:00 +0000

Since the beginning of my journey into the data world I've been keen on making professionals better at their data job. In the previous years that took the shape of creating materials in several different forms that could help people understand, use, and avoid mistakes on their data tool of choice. But now there's much more into it: a trusted AI solution to help data professional in their day to day optimization job.

From content to tooling

The big advantage of the content creation approach is the 1-N effect: the material, once created and updated, can serve a multitude of people interested in the same technology or facing the same problem. You write an article once, and it gets found and adopted by a vast amount of professionals.

The limit of content tho, it's that it is an extra resource, that people need to find and read elsewhere. While this is useful, it forces a context switch, moving people away from the problem they are facing. Here is where tooling helps, providing assistance in the same IDE that professionals are using for their day to day work.

Tooling for database professionals

I have the luxury of working for Aiven which provides professionals an integrated platform for all their data needs. In the last three years I witnessed the growth of the platform and its evolution with the clear objective to make it better usable at scale. Tooling like integrations, Terraform providers and the Console facilitate the work that platform administrators have to perform on daily basis.

But what about the day to day work of developers and CloudOps teams? This was facilitated when dealing with administrative tasks like backups, creation of read only replicas or upgrades, but the day to day work of optimizing the workloads was still completely on their hands.

Using trusted AI to optimize database workloads

This, however, is now changing. With the recent launch of Aiven AI Database Optimizer we are able to help both developer and CloudOps in their day to day optimization work!

Aiven AI Database Optimizer provides, directly in the Console, insights on database workloads alongside a one-click Optimize button that suggests index and SQL rewrites. It's a non intrusive solution who gathers informations from the slow query log and database metadata, and leverages an in-built AI engine, to provide accurate suggestions to improve performance.

The solution, based on the EverSQL by Aiven technology has been already adopted by 120.000 professionals optimizing over 2 million queries. It's not a wrapper around a public Generative AI provider, is a dedicated solution that keeps data privacy and security as a priority.

You can experience it for Free in the Early Availability phase, just navigate to the Aiven Console and create an Aiven for PostgreSQL® service! Once you have some workload on the service, the AI Insights page will show you the queries and provide index and SQL rewrite suggestion to take your database performance to the next level!

More information on the dedicated Aiven AI Database Optimizer page

Introducing Aiven's AI Database Optimizer: The First Built-In SQL Optimizer for Enhanced Performance

Francesco Tisiot — Tue, 28 May 2024 12:00:00 +0000

An efficient data infrastructure is a vital component in building & operating scalable and performant applications that are widely adopted, satisfy customers, and ultimately, drive business growth. Unfortunately, the speed of new feature delivery coupled with a lack of database optimization knowledge is exposing organizations to high risk performance issues. The new Aiven AI Database Optimizer helps organizations address performance both in the development and production phase, making it simple to quickly deploy, fully optimized, scalable, and cost efficient applications.

Fully integrated with Aiven for PostgreSQL®, Aiven AI Database Optimizer offers AI-driven performance insights, index, and SQL rewrite suggestions to maximize database performance, minimize costs, and make the best out of your cloud investment.

How does AI Database Optimizer work?

Aiven AI Database Optimizer is a non-intrusive solution powered by EverSQL by Aiven that gathers information about database workloads, metadata and supporting data structures, such as indexes. Information about the number of query executions and average query times are continually processed by a mix of heuristic and AI models to determine which SQL statements can be further optimized. AI Database Optimizer then delivers accurate, secure optimization suggestions that you can trust, and that can be adopted to speed up query performance.

Recommendations from Aiven’s AI Database Optimizer are already trusted by over 120,000 users in organizations ranging from startups to the largest global enterprises, who have optimized more than 2 million queries to date.

How does AI Database Optimizer help organizations?

During development, AI Database Optimizer enables early performance testing, allowing easier redesign or refactoring of queries before they impact production. This enables customers to foster a culture of considering performance from the get-go, ensuring it is a priority throughout development rather than an afterthought.

AI Database Optimizer also helps businesses gain an optimal user experience:

With fast query response times that ensure a smooth and responsive user experience, especially in data-intensive applications.
By identifying and fixing performance bottlenecks organizations can reduce costs, avoid outages and deliver continuous service availability.

A fleet of Aiven for PostgreSQL® databases is powering La Redoute’s marketplace functionality, driving 30% of their business.
Diogo Passadouro - OPS-DBA Team Lead stated "Aiven AI Database Optimizer has revolutionized the way we analyze database performance, providing a simple, clear and highly effective approach and has proven instrumental in enhancing the performance of our databases."

Conrad is an advanced B2B sourcing platform selling 9 million products from 6,000 brands, powered by Aiven. Janek Wonner - Head of SRE & Cloud Technology stated "Aiven for PostgreSQL is underpinning our fundamental company functionalities, we are looking forward to adopt Aiven AI database optimizer to empower our developers to create scalable code and empower our development teams with better performance insights and improvement suggestions to reduce the time to fix performance issues"

More information is available in the Aiven AI Database Optimizer page. You can experience it for yourself in any Aiven for PostgreSQL service for free during the Early Availability phase. Simply navigate to the “AI Insights” tab.

Try it now or Contact us today to check it out!

How to use pgbench to test PostgreSQL® performance

Francesco Tisiot — Tue, 12 Mar 2024 15:00:00 +0000

Testing a database performance is a must in every company. Despite everyone's needs beings slightly different, a good starting point for PostgreSQL® database is using pgbench: a tool shipped with the PostgreSQL installation that allows you to stress test a local or remote database.
This blog post showcases how to install (on a Mac) and use pgbench to create load on a remote PostgreSQL database on Aiven.

If you need a FREE PostgreSQL database?
🦀 Check Aiven's FREE plans! 🦀
⚡️ Need to optimize your SQL query with AI? ⚡️
🐧 Check Aiven AI database optimizer! Powered by EverSQL 🐧

Install pgbench locally

In a Mac, pgbench comes with the default PostgreSQL installation via brew. Therefore to have pgbench all you need is to:

brew install postgresql

Create a test PostgreSQL environment

While you could create a test PostgreSQL environment locally (the 1brc challenge blog post contains all the details), this time we'll create an Aiven for PostgreSQL service in minutes:

Navigate to Aiven Console
Create an account
Create an Aiven for PostgreSQL service on your favorite cloud provider and region
Select the startup-4 plan, it will be sufficient for the test.

Use pgbench to load test the a PostgreSQL database

The Aiven for PostgreSQL database comes with a defaultdb database we can use for our testing.

Step 1: Initialize the database

All we need is to grab the connection details from the Aiven Console and we are ready to initialize the database with:

pgbench -h <HOST>       \
    -p <PORT>           \
    -U <USER>           \
    -i                  \
    -s <SCALEFACTOR>    \
    <DBNAME>

Where:

<HOST> is the database hostname
<PORT> is the database port
<USER> is the database user
<DBNAME> is the database name
SCALEFACTOR is the test scale factor, 100 could be a good place to start. The default is 1 which will create a 16MB database, the 100 scale will create a 1.6GB database.

We'll be prompted to write the password, that we can find in the Aiven Console.

Note: You can automate the flow without the need to include the password by storing it in the PGPASSWORD environment variable with
export PGPASSWORD=<PASSWORD> and then rerun the pgbench command above.

After this pgbench will execute some work including the following tables:

       Name       | nr_rows
------------------+---------
 pgbench_accounts | 10000000
 pgbench_branches |      100
 pgbench_history  |        0 
 pgbench_tellers  |     1000

Step 2: execute a first test to set the baseline

After creating the dataset, it's time to define the performance baseline (a.k.a how much does my database perform under current conditions). Let's create one with:

pgbench  -h <HOST>                  \
    -p <PORT>                       \
    -U <USER>                       \
    -c <NUMBER_OF_CLIENTS>          \
    -j <NUMBER_OF_PGBENCH_THREADS>  \
    -t <NUMBER_OF_TRANSACTIONS>     \
    <DBNAME>

Where:

<NUMBER_OF_CLIENTS> is the number of clients to connect to PostgreSQL with, a good starting point is 20
<NUMBER_OF_PGBENCH_THREADS> is the number pf pgbench threads to run concurrently
<NUMBER_OF_TRANSACTIONS> is the overall number of transactions to execute

The following code will execute 1000 transactions using 5 parallel threads and 5 clients

pgbench  -h <HOST>      \
    -p <PORT>           \
    -U <USER>           \
    -c 5                \
    -j 5                \
    -t 1000             \
    <DBNAME>

The result is the following:

pgbench (16.1, server 16.2)
starting vacuum...end.
transaction type: <builtin: TPC-B (sort of)>
scaling factor: 100
query mode: simple
number of clients: 5
number of threads: 5
maximum number of tries: 1
number of transactions per client: 1000
number of transactions actually processed: 5000/5000
number of failed transactions: 0 (0.000%)
latency average = 180.179 ms
initial connection time = 164.343 ms
tps = 27.750256 (without initial connection time)

Showcasing that we are generating almost 28 transactions per second (tps in the above one) from my laptop sitting in Verona to a database sitting in Turin!

Step 3: change parameters and validate the new performance

What could we change? We could increase the percentage of total RAM dedicated to shared buffers to 60% by going in the Aiven Console selecting the PostgreSQL service, selecting the Service settings menu and clicking on the Configure button next to the Advanced configuration section.

Now, if we try with the same pgbench command:

pgbench  -h <HOST>      \
    -p <PORT>           \
    -U <USER>           \
    -c 5                \
    -j 5                \
    -t 1000             \
    <DBNAME>

We get updated results with 36 tps.

Step 4: define you custom workload in a SQL file

You can customize the type of queries pgbench will execute based on your workloads. All you need to do is to define a SQL file and call it with pgbench -f flag. Let's create a file named test.sql with:

\set r1 (random(1, 2000))
select * from pgbench_accounts where bid = :r1;

In the above we are setting a r1 variable with a random number between 1 and 2000 and then use it to query the pgbench_accounts table for rows with bid equal to r1.
We can now execute the custom sql file with:

pgbench  -h <HOST>      \
    -p <PORT>           \
    -U <USER>           \
    -c 5                \
    -j 5                \
    -t 1000             \
    -f test.sql         \
    <DBNAME>

In our example the above produces the following output:

scaling factor: 1
query mode: simple
number of clients: 5
number of threads: 5
maximum number of tries: 1
number of transactions per client: 100
number of transactions actually processed: 500/500
number of failed transactions: 0 (0.000%)
latency average = 10514.270 ms
initial connection time = 245.617 ms
tps = 0.475544 (without initial connection time)

A miserable 0.47 transactions per second. Can we do better?

Step 5: Optimize your database

Once you took time to write a custom workload to test with pgbench and got a baseline, it's time to optimize your database. You could do it based on your knowledge and available data but, if you are using Aiven, you can receive automated insights about your PostgreSQL database workload and index and (in the future) SQL optimization suggestions. Check out the video below!

For more information, check Aiven and EverSQL

In our case, Aiven will suggest to add an index on the bid column.

CREATE INDEX pgbench_accounts_idx_bid ON "pgbench_accounts" ("bid");

After creating it, and retrying the pgbench command above, the custom workload speed raised from 0.45 to 35 transaction per second.

Conclusion

pgbench is a great tool to perform performance checks on your PostgreSQL database and the custom script option provides a way to customize it to match your workloads. Used with Aiven and its AI insights is a great way to test and optimize your database!

From dbf to PostgreSQL with Python

Francesco Tisiot — Mon, 04 Mar 2024 15:00:00 +0000

Some time ago I found an interesting database file suffix I never faced before: the .dbf and saw aroung that it was first introduced in 1983 with dBASE II. This article showcases how we can automatically generate the PostgreSQL table and fill it with data using Python and dbfread.

If you need a FREE PostgreSQL database?
🦀 Check Aiven's FREE plans! 🦀
⚡️ Need to optimize your SQL query with AI? ⚡️
🐧 Check Aiven AI database optimizer! Powered by EverSQL 🐧

Download a sample dbf file

We can get a sample .dbf file from Infused with the following in our terminal

wget https://github.com/infused/dbf/raw/master/spec/fixtures/cp1251.dbf

This will store a file named cp1251.dbf in the current folder.

Use dbfread to move the data into PostgreSQL

We need to install simpledf with

pip install dbfread

Then we can write a Python script (named convert_bdf_to_sql.py) that opens the sample.dbf file and creates the PostgreSQL DDL and loads the data into a CSV file we can use to populate the database

from dbfread import DBF
import dataset

db = dataset.connect('postgresql://[USER]:[PWD]@[HOST]:[PORT]/[DBNAME]?sslmode=require')
table = db['people']

for record in DBF('cp1251.dbf', lowernames=True):
    table.insert(record)

In the above script we:

connect to a PostgreSQL instance using
- [USER]: the username
- [PWD]: the password
- [HOST]: the hostname
- [PORT]: the port
- [DBNAME]: the database name
define a table named people that will be created and populated
insert into the people table all the records in cp1251.dbf

We can then execute it with:

python convert_bdf_to_sql.py

If we now connect to our PostgreSQL database:

psql postgres://[USER]:[PWD]@[HOST]:[PORT]/[DBNAME]?sslmode=require

we can check the people table being populated with:

select * from people;

We can see the data in the table!

 id | rn |                  name
----+----+----------------------------------------
  1 |  1 | амбулаторно-поликлиническое
  2 |  2 | больничное
  3 |  3 | НИИ
  4 |  4 | образовательное медицинское учреждение
(4 rows)

List of PostgreSQL® AI Projects and Resources

Francesco Tisiot — Thu, 29 Feb 2024 10:00:00 +0000

Everyone is now talking about AI, and modern databases like PostgreSQL® are increasingly being adopted in companies' AI journey as sources of data or key pieces of the AI infrastructure. Moreover there's a new set projects that are solving PostgreSQL problems with AI.

The List of PostgreSQL® AI projects and resources is open-source project to collect PostgreSQL® extensions, applications and resources (video or blogs) talking about how our loved database can fit in the AI journey.

Did you write an article, gave a talk, are you working on an extension regarding PostgreSQL and AI? PRs are welcome!

11 Lessons to learn when using NULLs in PostgreSQL®

Francesco Tisiot — Wed, 28 Feb 2024 10:00:00 +0000

A boolean value should only contain two values, True or False, but is it correct? Usually people assume so, but sometimes miss the fact that there could be the absence of the value all-together. In databases this is absence is usually stored as NULL and this blog showcases how to find them, use them properly and 11 lessons to learn to be a NULL Pro!

Keep in mind, it's not only booleans that can contain NULL values, it's all the columns where you don't define a NOT NULL constraint!

If you need a FREE PostgreSQL database?
🦀 Check Aiven's FREE plans! 🦀
⚡️ Need to optimize your SQL query with AI? ⚡️
🐧 Check Aiven AI database optimizer! Powered by EverSQL 🐧

It all starts with some columns and rows

Let's start from the basics: you have a PostgreSQL® database and a table, called users like:

CREATE TABLE users (
    id SERIAL,
    username TEXT PRIMARY KEY,
    name TEXT,
    surname TEXT,
    age INT
);

Hey, do you want to test the above code but not having a PostgreSQL database handy? Past your code in PostgreSQL Playground and quickly check the results!

Let's insert some data:

INSERT INTO users (username, name, surname, age) VALUES
    ('jdoe', 'Jon', 'Doe', 25),
    ('lspencer','Liz', 'Spencer', 35),
    ('hlondon','Hanna', 'London', 45);

Querying the data showcases the table with all the columns filled.

 id | username | name  | surname | age 
----+----------+-------+---------+-----
  1 | jdoe     | Jon   | Doe     |  25
  2 | lspencer | Liz   | Spencer |  35
  3 | hlondon  | Hanna | London  |  45
(3 rows)

Insert NULLs

Now, let's check if we can insert some NULLs, let's try by inserting them in the name, surname and age columns:

INSERT INTO users (username, name, surname, age) VALUES ('test',NULL, NULL, NULL);

This works since we don't have any constraint

 id | username | name  | surname | age 
----+----------+-------+---------+-----
  1 | jdoe     | Jon   | Doe     |  25
  2 | lspencer | Liz   | Spencer |  35
  3 | hlondon  | Hanna | London  |  45
  4 | test     |       |         |    
(4 rows)

We can even simplify the insert with the same effect

INSERT INTO users (username) VALUES ('test1');

Result:

 id | username | name  | surname | age 
----+----------+-------+---------+-----
  1 | jdoe     | Jon   | Doe     |  25
  2 | lspencer | Liz   | Spencer |  35
  3 | hlondon  | Hanna | London  |  45
  4 | test     |       |         |    
  5 | test1    |       |         |    
(5 rows)

Inserting NULLs

The first step to avoid NULLs from appearing in a table is to forbid inserting them. The section below showcases a few situations on how to avoid inserting NULLs in new and existing columns.

Inserting NULLs in the primary key

Can I insert a NULL in the table primary key? In theory this could be allowed since there's no explicit NOT NULL constraint on the column. Let's try:

INSERT INTO users (username) VALUES (NULL);

However this fails with:

ERROR:  null value in column "username" of relation "users" violates not-null constraint
DETAIL:  Failing row contains (6, null, null, null, null).

💡 Lesson 1 💡: Primary Keys are by default NOT NULL

Create a new column with NOT NULL constraint

Let's add a new column to our table:

ALTER TABLE users ADD COLUMN points INT NOT NULL;

This will fail, since the previously inserted data don't have any associated value for points

ERROR:  column "points" of relation "users" contains null values

Let's add a default value with:

ALTER TABLE users ADD COLUMN points INT DEFAULT 0 NOT NULL;

The table was altered, and now contains 0 points for all the existing rows.

 id | username | name  | surname | age | points 
----+----------+-------+---------+-----+--------
  1 | jdoe     | Jon   | Doe     |  25 |      0
  2 | lspencer | Liz   | Spencer |  35 |      0
  3 | hlondon  | Hanna | London  |  45 |      0
  4 | test     |       |         |     |      0
  5 | test1    |       |         |     |      0
(5 rows)

If we try to insert a new row without specifying the points column:

INSERT INTO users (username) VALUES ('usrwithnullpoints');

The row is inserted with the default 0 points

 id |     username      | name  | surname | age | points 
----+-------------------+-------+---------+-----+--------
  1 | jdoe              | Jon   | Doe     |  25 |      0
  2 | lspencer          | Liz   | Spencer |  35 |      0
  3 | hlondon           | Hanna | London  |  45 |      0
  4 | test              |       |         |     |      0
  5 | test1             |       |         |     |      0
  7 | usrwithnullpoints |       |         |     |      0
(6 rows)

What if we try to push a NULL into points?

INSERT INTO users (username, points) VALUES ('forcenullpoints', NULL);

We get the ERROR: null value in column "points" of relation "users" violates not-null constraint stating that our insert is violating the NOT NULL constraint.

💡 Lesson 2 💡: Setting a NOT NULL constraint doesn't allow inserting NULLs

💡 Lesson 3 💡: If we want to add a NOT NULL column to an existing table with data, we need to specify the default value for existing rows.

Add a NOT NULL constraint to an existing column

If we try to change an existing column (e.g. name) to add the NOT NULL constraint:

ALTER TABLE users ALTER COLUMN name SET NOT NULL;

We get the similar error ERROR: column "name" of relation "users" contains null values, therefore we need also need to amend the current NULL values (and possibly set a default)

ALTER TABLE users 
    ALTER COLUMN name TYPE TEXT USING (COALESCE(name, 'Hugo')),
    ALTER COLUMN name SET DEFAULT 'Hugo',
    ALTER COLUMN name SET NOT NULL
    ;

In the above we:

Altered the type, setting it to TEXT again, to apply the function COALESCE(name, 'Hugo') which is replacing NULLs in the name field with Hugo
Set the default value for the name column to Hugo <- This is not strictly necessary
Applied the NOT NULL constraint since we don't have any NULL values anymore

Note: the above method locks the entire table until done. A faster approach could be to split the COALESCE(name, 'Hugo') into a dedicated update and then only perform the NOT NULL and DEFAULT settings on the ALTER TABLE statement.

The results are in line with the expected, no NULLs in the name column, being replaced by Hugo

 id |     username      | name  | surname | age | points 
----+-------------------+-------+---------+-----+--------
  1 | jdoe              | Jon   | Doe     |  25 |      0
  2 | lspencer          | Liz   | Spencer |  35 |      0
  3 | hlondon           | Hanna | London  |  45 |      0
  4 | test              | Hugo  |         |     |      0
  5 | test1             | Hugo  |         |     |      0
  7 | usrwithnullpoints | Hugo  |         |     |      0
(6 rows)

💡 Lesson 4 💡: If we need to add a NOT NULL column to an existing column in a table, we need to amend the existing NULL values and optionally set the default value for incoming rows.

What if I want to remove the NOT NULL constraint? We can do it with:

ALTER TABLE users 
    ALTER COLUMN name DROP NOT NULL
    ;

Is DEFAULT enough?

What if we leave the NOT NULL constraint aside, and just set the DEFAULT? Let's try with the date_first_login new column

ALTER TABLE users ADD COLUMN date_first_login DATE DEFAULT CURRENT_DATE;

Checking what's in the table, we see all the date_first_login filled

 id |     username      | name  | surname | age | points | date_first_login 
----+-------------------+-------+---------+-----+--------+------------------
  1 | jdoe              | Jon   | Doe     |  25 |      0 | 2024-02-27
  2 | lspencer          | Liz   | Spencer |  35 |      0 | 2024-02-27
  3 | hlondon           | Hanna | London  |  45 |      0 | 2024-02-27
  4 | test              | Hugo  |         |     |      0 | 2024-02-27
  5 | test1             | Hugo  |         |     |      0 | 2024-02-27
  7 | usrwithnullpoints | Hugo  |         |     |      0 | 2024-02-27
(6 rows)

If we try to insert a new row, without specifying the date_first_login, it's correctly assigned to today

INSERT INTO users (username) VALUES ('user_with_no_date_first_login');

The above is correctly inserted with a not null date_first_login

 id |           username            | name  | surname | age | points | date_first_login 
----+-------------------------------+-------+---------+-----+--------+------------------
  1 | jdoe                          | Jon   | Doe     |  25 |      0 | 2024-02-27
  2 | lspencer                      | Liz   | Spencer |  35 |      0 | 2024-02-27
  3 | hlondon                       | Hanna | London  |  45 |      0 | 2024-02-27
  4 | test                          | Hugo  |         |     |      0 | 2024-02-27
  5 | test1                         | Hugo  |         |     |      0 | 2024-02-27
  7 | usrwithnullpoints             | Hugo  |         |     |      0 | 2024-02-27
  9 | user_with_no_date_first_login | Hugo  |         |     |      0 | 2024-02-27
(7 rows)

So... is the NOT NULL constraint really needed? The reply is YES since I could insert NULLs by specifying it in the INSERT statement

INSERT INTO users (username, date_first_login) VALUES ('user_with_null_date_first_login', NULL);

Which will be stored correctly with a NULL value for date_first_login

 id |            username             | name  | surname | age | points | date_first_login 
----+---------------------------------+-------+---------+-----+--------+------------------
  1 | jdoe                            | Jon   | Doe     |  25 |      0 | 2024-02-27
  2 | lspencer                        | Liz   | Spencer |  35 |      0 | 2024-02-27
  3 | hlondon                         | Hanna | London  |  45 |      0 | 2024-02-27
  4 | test                            | Hugo  |         |     |      0 | 2024-02-27
  5 | test1                           | Hugo  |         |     |      0 | 2024-02-27
  7 | usrwithnullpoints               | Hugo  |         |     |      0 | 2024-02-27
  9 | user_with_no_date_first_login   | Hugo  |         |     |      0 | 2024-02-27
 10 | user_with_null_date_first_login | Hugo  |         |     |      0 | 
(8 rows)

💡 Lesson 5 💡: The DEFAULT is going to prevent NULLs from being inserted only if the NULL is NOT specified in the target column. To enforce the value we need an explicit NOT NULL constraint

Querying NULLs

In the above section we tried to avoid storing NULLs in our systems, what if we need to deal with existing NULLs in our column? Here we can use a few SQL functions to detect them and change their value.

Detecting NULLs with the `IS NULL` condition

What if we want to detect the NULL values? We can use the IS NULL condition, for example:

SELECT * from users where surname IS NULL;

Which will retrieve the list of NULLs surnames

 id |            username             | name | surname | age | points | date_first_login 
----+---------------------------------+------+---------+-----+--------+------------------
  4 | test                            | Hugo |         |     |      0 | 2024-02-27
  5 | test1                           | Hugo |         |     |      0 | 2024-02-27
  7 | usrwithnullpoints               | Hugo |         |     |      0 | 2024-02-27
  9 | user_with_no_date_first_login   | Hugo |         |     |      0 | 2024-02-27
 10 | user_with_null_date_first_login | Hugo |         |     |      0 | 
(5 rows)

The opposite list could be retrieved by applying the IS NOT NULL condition:

SELECT * from users where surname IS NOT NULL;

Resulting in

 id | username | name  | surname | age | points | date_first_login 
----+----------+-------+---------+-----+--------+------------------
  1 | jdoe     | Jon   | Doe     |  25 |      0 | 2024-02-27
  2 | lspencer | Liz   | Spencer |  35 |      0 | 2024-02-27
  3 | hlondon  | Hanna | London  |  45 |      0 | 2024-02-27
(3 rows)

What if I try to use the equal comparison (=)

SELECT * from users where surname = NULL;

The above retrieves NO rows since the NULL value means there is NO value and is NOT equal to any other value (NULL=NULL is False)

 id | username | name | surname | age | points | date_first_login 
----+----------+------+---------+-----+--------+------------------
(0 rows)

If we apply the different operator (<>) with NULL we also get back NO rows, since the NULL value means there is NO value, and therefore is not equal or different to other values.

SELECT * from users where surname <> NULL;

💡 Lesson 6 💡: To find NULLs we should always use the IS NULL; the IS NOT NULL to fetch the not null rows.

Associating a default value to NULLs with COALESCE

In the above we saw how to find NULLs, what if we need to include them in a SELECT statement with other values? We can either add an OR condition like

SELECT * from users 
WHERE surname = 'London' 
OR surname IS NULL;

Or we can apply a transformation to change the NULLs into a value we can compare

SELECT * from users 
WHERE COALESCE(surname,'Verona') in ('London', 'Verona');

Both will have the same result

 id |            username             | name  | surname | age | points | date_first_login 
----+---------------------------------+-------+---------+-----+--------+------------------
  3 | hlondon                         | Hanna | London  |  45 |      0 | 2024-02-27
  4 | test                            | Hugo  |         |     |      0 | 2024-02-27
  5 | test1                           | Hugo  |         |     |      0 | 2024-02-27
  7 | usrwithnullpoints               | Hugo  |         |     |      0 | 2024-02-27
  9 | user_with_no_date_first_login   | Hugo  |         |     |      0 | 2024-02-27
 10 | user_with_null_date_first_login | Hugo  |         |     |      0 | 
(6 rows)

However, if we apply the second option, we must be sure that Verona (the label we associate to NULL values) is a value not present in the column or a value we want to include in our selection.

💡 Lesson 7 💡: When using COALESCE check out that the value you want to use as substitute of NULL is not already present in the table and outside of your business logic otherwise you might include in the filter more values than expected.

Performing Math on NULL columns

A similar scenario is due when we need to perform a calculation over a NULLable column. Let's take the example of the age column: what if we want to calculate the average age?

Let's see the data:

 id |            username             | name  | surname | age | points | date_first_login 
----+---------------------------------+-------+---------+-----+--------+------------------
  1 | jdoe                            | Jon   | Doe     |  25 |      0 | 2024-02-27
  2 | lspencer                        | Liz   | Spencer |  35 |      0 | 2024-02-27
  3 | hlondon                         | Hanna | London  |  45 |      0 | 2024-02-27
  4 | test                            | Hugo  |         |     |      0 | 2024-02-27
  5 | test1                           | Hugo  |         |     |      0 | 2024-02-27
  7 | usrwithnullpoints               | Hugo  |         |     |      0 | 2024-02-27
  9 | user_with_no_date_first_login   | Hugo  |         |     |      0 | 2024-02-27
 10 | user_with_null_date_first_login | Hugo  |         |     |      0 | 
(8 rows)

We can calculate the average age with:

select avg(age) 
from users;

Resulting in 35 which is the average between the 3 NOT NULL ages! Similarly if we count the date_first_logins with:

select count(date_first_login) 
from users;

The result is 7, one row for every user having a not null date_first_login

💡 Lesson 8 💡: When performing Mathematical operation with nullable columns, only NOT NULL values will be elaborated!

What if we try to calculate the average points? First let's update the table to give jdoe 25 points

UPDATE users set points=25 where username='jdoe';

Now let's calcuate the average points

select avg(points) from users;

The result is 3.125 which is the mathematical average between the 25 points of jdoe and the 0 of the rest due to us previously setting 0 as default value. Is this correct? It depends! If we wanted to calculate the average number of points of users actually playing some games, this could be a misleading calculation.

💡 Lesson 9 💡: When performing Mathematical operation with nullable columns, the default value of a column will impact the results.

Joining on NULL columns

Building on what we explored before about the NULL not being equal or different to any other value, we need to pay closer attention when we deal with nullable columns in join condition.

Let's create another table called sessions and insert some data. For a weird reason, the design of this table will force us to join with users on the surname which is a NULLABLE field.

CREATE TABLE sessions (id serial, surname text, session_time timestamp);
INSERT INTO sessions (surname, session_time) values
    ('Doe', CURRENT_TIMESTAMP - INTERVAL '1 day'),
    ('Doe', CURRENT_TIMESTAMP - INTERVAL '1 day'*2),
    ('Spencer', CURRENT_TIMESTAMP - INTERVAL '1 day'*2),
    ('Spencer', CURRENT_TIMESTAMP),
    (NULL, CURRENT_TIMESTAMP),
    (NULL, CURRENT_TIMESTAMP - INTERVAL '1 day'*2),
    (NULL, CURRENT_TIMESTAMP - INTERVAL '1 day'*2),
    (NULL, CURRENT_TIMESTAMP - INTERVAL '1 day'*1),
    (NULL, CURRENT_TIMESTAMP);

Now, if we want to count users having a session on each day we could think of simply writing the above query:

SELECT EXTRACT(DAY FROM session_time),
    count(distinct username) 
FROM sessions join users 
    on sessions.surname=users.surname
group by EXTRACT(DAY FROM session_time);

However the results shows only the 4 sessions below

 extract | count 
---------+-------
      25 |     2
      26 |     1
      27 |     1
(3 rows)

Why? It's again because of NULL values in the join condition. This problem of poor data quality on both ends of the join is stopping us from reporting any useful number. How could we fix it? Well, the fix should be done uphill by setting the sessions table to use the username instead of the surname and defining a proper FOREIGN KEY to the users table.

💡 Lesson 10 💡: Joining tables on nullable columns without caring about the NULL management will impact the results by reducing the number of rows included in the join.

Still, if we want a simplistic solution, we could do the apply the COALESCE to both sides of the join:

SELECT EXTRACT(DAY FROM session_time),
    count(distinct username) 
FROM sessions join users 
    on coalesce(sessions.surname, 'NoSurname')=coalesce(users.surname, 'NoSurname')
group by EXTRACT(DAY FROM session_time);

The result shows an increased count of sessions

 extract | count 
---------+-------
      25 |     7
      26 |     6
      27 |     6
(3 rows)

However this is still not the correct answer, since we are seeing 19 session being aggregated while only 9 sessions were inserted in the original sessions table. Why is that? It's because we now casted all the NULLs as NoSurname on both sides of the join. Since we had 5 sessions without surname and 5 users without surname we are creating a cartesian join of all the empty surnames with the empty sessions.

We had the following users without surname

test
test1
usrwithnullpoints
user_with_no_date_first_login
user_with_null_date_first_login

And the following sessions without surname

  6 | 2024-02-27 14:10:08.419021
  7 | 2024-02-25 14:10:08.419021
  8 | 2024-02-25 14:10:08.419021
  9 | 2024-02-26 14:10:08.419021
 10 | 2024-02-27 14:10:08.419021

Each of the users was associated with each sessions, creating a overflow of sessions.

A more correct solution, attribuiting all the NULL session to a same phantom user NoUser could be:

SELECT EXTRACT(DAY FROM session_time),
    count(distinct coalesce(username,'NoUser'))
FROM sessions left outer join users 
    on sessions.surname=users.surname
group by EXTRACT(DAY FROM session_time);

However this is an effort to fix data quality issues at query time, which is never a good idea.

💡 Lesson 11 💡: Fixing data quality issues related to nullable columns at query time is a risky approach. Solving them with proper table design and constraints allows you to keep quality data in the tables.

Conclusion

Dealing with NULLs is something to care upfront, during table design, in order to avoid data quality issues downstream!
This is it for the 11 lessons learnt using NULLs, do you have others to share? Feel free to reach me on X/Twitter at @ftisiot or on Linkedin

How to CONCAT in MySQL

Francesco Tisiot — Tue, 13 Feb 2024 13:00:00 +0000

One of the most common tasks with strings is concatenation! This blog post showcases several techniques to perform it with MySQL.

If you need a FREE MySQL database? 🦀 Check Aiven's FREE plans! 🦀
If you need to optimize your SQL query? 🐧 Check EverSQL! 🐧

String concatenation in MySQL by placing strings next to each other

Let's take a basic example: a table containing firstname and surname fields we want to concatenate:

CREATE TABLE CLIENTS (
    id INT PRIMARY KEY, 
    firstname VARCHAR(100), 
    surname VARCHAR(100),
    region VARCHAR(100),
    creation_date date,
    preferred_item VARCHAR(100)
    );
INSERT INTO CLIENTS VALUES 
    (1, 'John', 'Doe', 'EMEA', '2024-01-01', 'football'),
    (2, 'Laura', 'Smith', 'EMEA', '2024-02-01', 'tennis ball'),
    (3, 'Eva', 'Book', 'APAC', '2024-01-03', null),
    (4, 'Andy', 'Novel', 'APAC', '2024-02-05', 'teddy bear'),
    (5, 'Eva', 'Leeds', 'APAC', '2024-02-01', 'football');

We can concatenate two quoted strings by placing one string next to the other like:

select 'MyFirstname' 'MySurname';

The result is

+----------------------+
| MyFirstname          |
+----------------------+
| MyFirstnameMySurname |
+----------------------+
1 row in set (0.03 sec)

We can see that:

the result is the correct concatenation
the column name is the first quoted string

However, if we apply the same principle to standard columns instead of quoted strings with:

select firstname surname from CLIENTS;

We get a different result:

+---------+
| surname |
+---------+
| John    |
| Laura   |
| Eva     |
| Andy    |
| Eva     |
+---------+
5 rows in set (0.02 sec)

The result showcases only the firstname in the results, with the surname being included as the result column heading.

TLDR: concatenating strings in MySQL by placing strings next to each other is possible only if we are dealing with quoted strings.

Concatenating a non string

Is it possible with the same method to concatenate a non string?

select 'MyFirstname' 1;

Nope, we get ERROR 1064 (42000): You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near '1' at line 1

Concatenating a NULL

What about concatenating a NULL value?

select 'MyFirstname' NULL;

Same error ERROR 1064 (42000): You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near 'NULL' at line 1

String concatenation in MySQL with the `||` operator

We can also concatenate firstname and surname using the || operator with:

select firstname || surname from CLIENTS;

The result will be:

+-----------------------+
| firstname || surname  |
+-----------------------+
| JohnDoe               |
| LauraSmith            |
| EvaBook               |
| AndyNovel             |
| EvaLeeds              |
+-----------------------+
5 rows in set (0.02 sec)

Please check that there's no space between the firstname and surname, but we can add it with:

select firstname || ' ' || surname from CLIENTS;

Which shows the expected result

+------------------------------+
| firstname || ' ' || surname  |
+------------------------------+
| John Doe                     |
| Laura Smith                  |
| Eva Book                     |
| Andy Novel                   |
| Eva Leeds                    |
+------------------------------+
5 rows in set (0.02 sec)

Concatenating a non string

What about concatenating strings with other fields that are not a string? If, for example, we try to concat the firstname and id (an integer) with the following:

select firstname || id from CLIENTS;

We get the expected result

+------------------+
| firstname || id  |
+------------------+
| John1            |
| Laura2           |
| Eva3             |
| Andy4            |
| Eva5             |
+------------------+

The same if we try to concatenate the creation_date date field

select firstname || creation_date from CLIENTS;

Result:

+-----------------------------+
| firstname || creation_date  |
+-----------------------------+
| John2024-01-01              |
| Laura2024-02-01             |
| Eva2024-01-03               |
| Andy2024-02-05              |
| Eva2024-02-01               |
+-----------------------------+
5 rows in set (0.02 sec)

However we might want to add a function to the date to get alternative formats from the default, for example the DATE_FORMAT function below converts the date in a weird format separating years, months and days with ..

select firstname || DATE_FORMAT(creation_date, '..%Y..%M..%D') from CLIENTS;

Results

+---------------------------------------------------------+
| firstname || DATE_FORMAT(creation_date, '..%Y..%M..%D') |
+---------------------------------------------------------+
| John..2024..January..1st                                |
| Laura..2024..February..1st                              |
| Eva..2024..January..3rd                                 |
| Andy..2024..February..5th                               |
| Eva..2024..February..1st                                |
+---------------------------------------------------------+
5 rows in set (0.03 sec)

Concatenating a NULL value

What does it happen if we try to concatenate a NULL value? If you check the dataset created, the row with id=3 for Eva Book has a preferred_item equal to NULL, let's check the output of concatenating the firstname, surname and preferred_item:

select firstname || surname || preferred_item  from CLIENTS;

Result

+------------------------------------------+
| firstname || surname || preferred_item   |
+------------------------------------------+
| JohnDoefootball                          |
| LauraSmithtennis ball                    |
| NULL                                     |
| AndyNovelteddy bear                      |
| EvaLeedsfootball                         |
+------------------------------------------+
5 rows in set (0.03 sec)

The result shows that the entire concatenation string is NULL!

TLDR: a concatenation will return NULL if any of the strings is NULL

String concatenation in MySQL with the `CONCAT` function

Another option is to use MySQL Concat function. Let's try with the basic firstname and surname example:

select concat(firstname, surname) from CLIENTS;

The results are in line with the previous || example, not having any additional separator added into the resulting string.

+----------------------------+
| concat(firstname, surname) |
+----------------------------+
| JohnDoe                    |
| LauraSmith                 |
| EvaBook                    |
| AndyNovel                  |
| EvaLeeds                   |
+----------------------------+
5 rows in set (0.03 sec)

How can we add a space in between? We have two options:

Add a space to the CONCAT function

select concat(firstname, ' ' ,surname) from CLIENTS;

Use the CONCAT_WS function (concat with separator) passing the space as separator

select concat_ws(' ', firstname, surname) from CLIENTS;

Concatenating a non string with the CONCAT function

What happens if we include non strings in the CONCAT function?

select concat(firstname, id) from CLIENTS;

As before we get the expected result

+-----------------------+
| concat(firstname, id) |
+-----------------------+
| John1                 |
| Laura2                |
| Eva3                  |
| Andy4                 |
| Eva5                  |
+-----------------------+
5 rows in set (0.03 sec)

Concatenating a NULL value with the CONCAT function

What about NULL values?

select concat(firstname, surname, preferred_item)  from CLIENTS;

As before, if any of the source strings is NULL, the result of the concatenation is NULL

+--------------------------------------------+
| concat(firstname, surname, preferred_item) |
+--------------------------------------------+
| JohnDoefootball                            |
| LauraSmithtennis ball                      |
| NULL                                       |
| AndyNovelteddy bear                        |
| EvaLeedsfootball                           |
+--------------------------------------------+
5 rows in set (0.03 sec)

String aggregation via concatenation in MySQL with the `GROUP_CONCAT` function

If we need to concatenate all the strings belonging to the same group, we can use MySQL GROUP_CONCAT function.

If, in our example, we want to concatenate the firstname of all the people grouped by their region we can write the following:

select region,
    group_concat(firstname)
from
    CLIENTS
group by region

The result is the following, with , as the default separator:

+--------+-------------------------+
| region | group_concat(firstname) |
+--------+-------------------------+
| APAC   | Eva,Andy,Eva            |
| EMEA   | John,Laura              |
+--------+-------------------------+
2 rows in set (0.03 sec)

We can also order how the strings are ordered in the list with:

select region,
    group_concat(firstname ORDER BY firstname)
from
    CLIENTS
group by region;

Result

+--------+--------------------------------------------+
| region | group_concat(firstname ORDER BY firstname) |
+--------+--------------------------------------------+
| APAC   | Andy,Eva,Eva                               |
| EMEA   | John,Laura                                 |
+--------+--------------------------------------------+
2 rows in set (0.03 sec)

And change the separator with

select region,
    group_concat(firstname ORDER BY firstname SEPARATOR '$$') res
from
    CLIENTS
group by region;

Result

+--------+----------------+
| region | res            |
+--------+----------------+
| APAC   | Andy$$Eva$$Eva |
| EMEA   | John$$Laura    |
+--------+----------------+
2 rows in set (0.02 sec)

We can also only list the distinct values, removing the duplication of Eva with:

select region,
    group_concat(DISTINCT firstname ORDER BY firstname SEPARATOR '$$') res
from
    CLIENTS
group by region;

Result

+--------+-------------+
| region | res         |
+--------+-------------+
| APAC   | Andy$$Eva   |
| EMEA   | John$$Laura |
+--------+-------------+
2 rows in set (0.03 sec)

Aggregating and concatenating a non string with the GROUP_CONCAT function

Let's try with a non string, like the id field:

select region,
    group_concat(id) res
from
    CLIENTS
group by region;

Works!

+--------+-------+
| region | res   |
+--------+-------+
| APAC   | 3,4,5 |
| EMEA   | 1,2   |
+--------+-------+
2 rows in set (0.02 sec)

The same if we try with creation_date

+--------+----------------------------------+
| region | res                              |
+--------+----------------------------------+
| APAC   | 2024-01-03,2024-02-05,2024-02-01 |
| EMEA   | 2024-01-01,2024-02-01            |
+--------+----------------------------------+
2 rows in set (0.02 sec)

Aggregating and concatenating a NULL value with the GROUP_CONCAT function

What about our NULL value? Let's check it out!

select region,
    group_concat(preferred_item) res
from
    CLIENTS
group by region;

In this case the behaviour is different:

+--------+----------------------+
| region | res                  |
+--------+----------------------+
| APAC   | teddy bear,football  |
| EMEA   | football,tennis ball |
+--------+----------------------+
2 rows in set (0.02 sec)

Even if the row with id=3 part of the APAC group had preferred_item equal to NULL, the resulting aggregate is not NULL! This is a common behaviour for aggregate functions to ignore the NULL values.

Conclusion

Here's an handy overview of the concatenation possibilities in MySQL:

Method	code	works with quoted strings	works with standard columns	works with non strings	Works with `NULL`	aggregates results
Placing strings next to each other	'a' 'b'	✅	❌	❌	❌	❌
\| operator	'a' \|\| 'b'	✅	✅	✅	✅	❌
CONCAT function	concat('a','b')	✅	✅	✅	✅	❌
GROUP_CONCAT function	group_concat('a','b')	✅	✅	✅	✅	✅

How to load JSON data in PostgreSQL with the COPY command

Francesco Tisiot — Fri, 02 Feb 2024 13:00:00 +0000

You have a JSON dataset that you want to upload to a PostgreSQL table containing properly formatted rows and columns... How do you do it?

All the main sources like my own blog and others tell you to load the JSON in a dedicated temporary table containing a unique JSON column, then parse it and load into the target table. However there could be another way, avoiding the temp table!

In this blog we'll see how to upload the JSON directly using PostgreSQL COPY command and using an utility called jq!

If you need a FREE PostgreSQL database? 🦀 Check Aiven's FREE plans! 🦀
⚡️ Need to optimize your SQL query with AI? ⚡️
🐧 Check Aiven AI database optimizer! Powered by EverSQL 🐧

PostgreSQL COPY command

First of all, let's check PostgreSQL COPY command. It's a command that allows you to copy into a PostgreSQL table data from a file, there are two versions:

COPY if the file is sitting in the PG server already
\COPY if the file is sitting in the client machine connected to the server via psql

In both cases, the standard COPY command has the following minimal set of parameters:

\copy <TARGET TABLE>(<OPTIONAL LIST OF COLUMNS>) FROM <SOURCE FILE> <FORMAT>

Where:

<TARGET TABLE> is the name of the target table
(<OPTIONAL LIST OF COLUMNS>) defines the columns in the table to load
<SOURCE FILE> is pointing to the source file to load
<FORMAT> defines the format of the data

The full list of parameters is available in the PostgreSQL COPY documentation

PostgreSQL COPY command - the out of the box formats

Let's focus on the formats available in the PostgreSQL COPY documentation are listed:

TEXT: can be used to load a full text
CSV: to load comma (or otherwise separated) values
BINARY: to load binary files

Unfortunately there doesn't seem to be an out of the box way to load JSON files!

The `PROGRAM` option to the rescue!

A, maybe not known, option of the COPY command is to point to a program to execute instead of the file. This option can be called with the following command:

\copy <TARGET TABLE>(<OPTIONAL LIST OF COLUMNS>) FROM PROGRAM "<SET OF INSTRUCTIONS>"

Compared to the previous \copy call, this time we are adding the PROGRAM with a set of instructions delimited by quotes or double quotes that will be executed on the client machine before loading the data.

If you are using the COPY command on the server, you'll probably need a superuser. This is the error message shown in Aiven: ERROR: must be superuser or have privileges of the pg_read_server_files role to COPY from a file

So, what we can do, is to reshape the data before loading it.

jq - the indispensable JSON parsing tool

I've been using jq quite a while in a lot of blog posts, it's a very handy tool to parse, reshape, select JSON documents. For the purpose of this blog, we'll use to reshape the JSON input into a CSV format, digestible from the PostgreSQL COPY command.

You need to have jq installed on the workstation from where the COPY command is executed!

Let's create a basic JSON file with named test.json with the following content:

{
    "id":1,
    "mystring":"ciao"
}
{
    "id":2,
    "mystring":"sole"
}
{
    "id":3,
    "mystring":"mare"
}

With jq we can read and reshape the above JSON to a CSV format with:

more test.json | jq -r ". | [.id, .mystring] | @csv"

In the above command:

more test.json reads the file
jq -r prints the raw output
the first . selects all the elements at the root level
| [.id, .mystring] retrieves the id and mystring keys from each element
|@csv sets the output format as CSV

The output is:

1,"ciao"
2,"sole"
3,"mare"

To check the complete set of options available with jq please view the manual

Stitching all together

So, how can we load a target table with just 1 COPY command? Let's first create the target table with:

CREATE TABLE MYTARGETTABLE (id serial, myid int, mystring text);

We now want to load the myid and mystring columns of the MYTARGETTABLE table with the following COPY command reading from the test.json and applying the transformation with jq.

\copy MYTARGETTABLE(myid, mystring) 
FROM PROGRAM 'more test.json | jq -r ". | [.id, .mystring] | @csv"'  
CSV

The output is the data properly loaded in the MYTARGETTABLE table

 id | myid | mystring
----+------+----------
  1 |    1 | ciao
  2 |    2 | sole
  3 |    3 | mare
(3 rows)

Solving problems at the source sometimes is useful to avoid extra hops! Stitching together COPY (with PROGRAM) and jq provides us the flexibility to load JSON files without intermediary tables.

Load StackOverflow's StackExchange data in PostgreSQL®

Francesco Tisiot — Wed, 31 Jan 2024 10:00:00 +0000

I recently found about the StackOverflow dataset in Kaggle, which points to the StackExchange link to download the entire data. In this blog I'll show a (maybe not 100% polished) way to upload the data to several tables in a PostgreSQL database!

The data is provided by the Stack Exchange Network

If you need a FREE PostgreSQL database? 🦀 Check Aiven's FREE plans! 🦀
If you need to optimize your SQL query? 🐧 Check EverSQL! 🐧

Create a PostgreSQL database

You can create one with Aiven or just have one local, the below code works in both examples.

Create database tables

Every downloaded file, referring to a specific site, contains the following files:

Posts
Users
Votes
Comments
PostHistory
PostLinks

The detailed schema information is available in this meta post.

We'll load the data in a two step approach:

First load the XML files in a temporary table
Then load the data from the temporary table into the proper tables with optimised column types

To load the data we need the following data structures:

a table data_load containing a unique data TEXT column we'll use to load the XML data row by row
a set of tables matching the file structures in Posts, Users, Votes, Comments, PostHistory, PostLinks

We can create the data_load table with:

CREATE TABLE data_load (
    data text
);

The data_load table will be used to load the XML files on the database and then parse them accordingly.

The following tables are needed to properly store the data in the database.

CREATE TABLE users(
    id int PRIMARY KEY,
    reputation int,
    CreationDate text,
    DisplayName text,
    LastAccessDate timestamp,
    Location text,
    AboutMe text,
    views int,
    UpVotes int,
    DownVotes int,
    AccountId int
);

CREATE TABLE posts (
    id int PRIMARY KEY,
    PostTypeId int,
    CreationDate timestamp,
    score int,
    viewcount int,
    body text,
    OwnerUserId int,
    LastActivityDate text,
    Title text,
    Tags text,
    AnswerCount int,
    CommentCount int,
    ContentLicense text
);

CREATE TABLE badges (
    id int PRIMARY KEY,
    userId int,
    Name text,
    dt timestamp,
    class int,
    tagbased boolean
);

CREATE TABLE comments (
    id int,
    postId int,
    score int,
    text text,
    creationdate timestamp,
    userid int,
    contentlicense text
);

CREATE TABLE posthistory (
    id int,
    PostHistoryTypeId int,
    postId int,
    RevisionGUID text,
    CreationDate timestamp,
    userID int,
    text text,
    ContentLicense text
);

CREATE TABLE postlinks (
    id int,
    creationdate timestamp,
    postId int,
    relatedPostId int,
    LinkTypeId int
);

CREATE TABLE tags (
    id int,
    tagname text,
    count int,
    ExcerptPostId int,
    WikiPostId int,
    IsRequired boolean,
    IsModeratorOnly boolean
);

CREATE TABLE votes(
    id int,
    postid int,
    votetypeid int,
    creationdate timestamp
);

Load the data

Let's download the ai.stackexchange.com the section from StackExchange dedicated to AI. Within the downloaded folder we can find 8 files (1-1 with our tables). Let's try to load the Users.xml first.

As mentioned before, we'll perform a two step loading approach:

Load the XML into a table, with each XML row on a different database row
Parse the XML to populate the proper tables and columns

The two step approach is needed since the original dataset threats all columns, including ids, as strings. We could either define all the Ids as strings or, do a bit more work to load the data into the proper column definitions.

Step 1: Load the User XML into the `data_load` table

To load the entire XML of the Users.xml file into a temporary data_load table we can connect to the database using psql from the same folder where the Users.xml is located and execute:

\copy data_load from program 'tr -d "\t" < Users.xml | sed -e ''s/\\/\\\\/g''' HEADER

The \copy command allows us to load the dataset into the data_load table. However we need to perform a couple of tricks in order to load it properly:

tr -d "\t" < Users.xml removes the tabs from the file.
sed -e ''s/\\/\\\\/g'' properly escapes the \ in the strings.
HEADER avoids to load the initial <?xml version="1.0" encoding="utf-8"?> row, unnecessary for our parsing

Step 2: Load the data into the `Users` table

The above command loads the data into the data_load table with one row in the table for each row in the original file. This is not optimal but avoids having to deal with very large files included in just a single blob.

We can make use the fact that each user is contained in a <raw> tag to leverage PostgreSQL ability to parse XML fields with the following:

INSERT INTO USERS (
    Id, CreationDate, DisplayName, LastAccessDate, Location, 
    AboutMe, Views, UpVotes, DownVotes,AccountId, Reputation)
SELECT
     (xpath('//@Id', x))[1]::text::int AS id
    ,(xpath('//@CreationDate', x))[1]::text::timestamp AS CreationDate
    ,(xpath('//@DisplayName', x))[1]::text AS DisplayName
    ,(xpath('//@LastAccessDate', x))[1]::text::timestamp AS LastAccessDate
    ,(xpath('//@Location', x))[1]::text AS Location
    ,(xpath('//@AboutMe', x))[1]::text AS AboutMe
    ,(xpath('//@Views', x))[1]::text::int AS Views
    ,(xpath('//@UpVotes', x))[1]::text::int AS UpVotes
    ,(xpath('//@DownVotes', x))[1]::text::int AS DownVotes
    ,(xpath('//@AccountId', x))[1]::text::int AS AccountId
    ,(xpath('//@Reputation', x))[1]::text::int AS Reputation
FROM data_load, unnest(xpath('//row', data::xml)) x
WHERE regexp_like(data,'[ ]+\<row');

The above uses:

the xpath to extract the relevant fields from each user
the ::text::int to cast the extracted field to the proper column type
the filter regexp_like(data,'[ ]+\<row') to remove the other unnecessary rows, including for example, <users> or </users>

Load the other tables

Similar to the example with Users.xml above, we can load the other tables with the following steps (please not we are truncating the data_load table before loading the next file):

-- Loading posts
truncate data_load;

\copy data_load from program 'tr -d "\t" < Posts.xml | sed -e ''s/\\/\\\\/g''' HEADER

INSERT INTO POSTS (
    Id, PostTypeId, CreationDate, score, viewcount, 
    body, OwnerUserId, LastActivityDate, Title, Tags, AnswerCount,
    CommentCount, ContentLicense)
SELECT
     (xpath('//@Id', x))[1]::text::int
    ,(xpath('//@PostTypeId', x))[1]::text::int
    ,(xpath('//@CreationDate', x))[1]::text::timestamp
    ,(xpath('//@Score', x))[1]::text::int
    ,(xpath('//@ViewCount', x))[1]::text::int
    ,(xpath('//@Body', x))[1]::text 
    ,(xpath('//@OwnerUserId', x))[1]::text::int 
    ,(xpath('//@LastActivityDate', x))[1]::text::timestamp 
    ,(xpath('//@Title', x))[1]::text 
    ,(xpath('//@Tags', x))[1]::text 
    ,(xpath('//@AnswerCount', x))[1]::text::int 
    ,(xpath('//@CommentCount', x))[1]::text::int 
    ,(xpath('//@ContentLicense', x))[1]::text 
FROM data_load, unnest(xpath('//row', data::xml)) x
WHERE regexp_like(data,'[ ]+\<row');

-- Loading badges

truncate data_load;

\copy data_load from program 'tr -d "\t" < Badges.xml | sed -e ''s/\\/\\\\/g''' HEADER

INSERT INTO BADGES (
    Id, userId, Name, dt, class, 
    tagbased)
SELECT
     (xpath('//@Id', x))[1]::text::int
    ,(xpath('//@UserId', x))[1]::text::int
    ,(xpath('//@Name', x))[1]::text
    ,(xpath('//@Date', x))[1]::text::timestamp
    ,(xpath('//@Class', x))[1]::text::int
    ,(xpath('//@TagBased', x))[1]::text::boolean
FROM data_load, unnest(xpath('//row', data::xml)) x
WHERE regexp_like(data,'[ ]+\<row');

-- Loading comments

truncate data_load;

\copy data_load from program 'tr -d "\t" < Comments.xml | sed -e ''s/\\/\\\\/g''' HEADER

INSERT INTO COMMENTS (
    Id, postId, score, text, creationdate, 
    userid, contentlicense)
SELECT
     (xpath('//@Id', x))[1]::text::int
    ,(xpath('//@PostId', x))[1]::text::int
    ,(xpath('//@Score', x))[1]::text::int
    ,(xpath('//@Text', x))[1]::text
    ,(xpath('//@CreationDate', x))[1]::text::timestamp
    ,(xpath('//@UserId', x))[1]::text::int
    ,(xpath('//@ContentLicense', x))[1]::text
FROM data_load, unnest(xpath('//row', data::xml)) x
WHERE regexp_like(data,'[ ]+\<row');

-- Loading posthistory

truncate data_load;

\copy data_load from program 'tr -d "\t" < PostHistory.xml | sed -e ''s/\\/\\\\/g''' HEADER

INSERT INTO posthistory (
    Id, PostHistoryTypeId, postId, RevisionGUID, CreationDate, 
    userID, text, ContentLicense)
SELECT
     (xpath('//@Id', x))[1]::text::int
    ,(xpath('//@PostHistoryTypeId', x))[1]::text::int
    ,(xpath('//@PostId', x))[1]::text::int
    ,(xpath('//@RevisionGUID', x))[1]::text
    ,(xpath('//@CreationDate', x))[1]::text::timestamp
    ,(xpath('//@UserId', x))[1]::text::int
    ,(xpath('//@Text', x))[1]::text
    ,(xpath('//@ContentLicense', x))[1]::text
FROM data_load, unnest(xpath('//row', data::xml)) x
WHERE regexp_like(data,'[ ]+\<row');

-- Loading postlinks

truncate data_load;

\copy data_load from program 'tr -d "\t" < PostLinks.xml | sed -e ''s/\\/\\\\/g''' HEADER

INSERT INTO postlinks (
    Id, creationdate, postId, relatedPostId, LinkTypeId)
SELECT
     (xpath('//@Id', x))[1]::text::int
    ,(xpath('//@CreationDate', x))[1]::text::timestamp
    ,(xpath('//@PostId', x))[1]::text::int
    ,(xpath('//@RelatedPostId', x))[1]::text::int
    ,(xpath('//@LinkTypeId', x))[1]::text::int
FROM data_load, unnest(xpath('//row', data::xml)) x
WHERE regexp_like(data,'[ ]+\<row');


-- Loading tags

truncate data_load;

\copy data_load from program 'tr -d "\t" < Tags.xml | sed -e ''s/\\/\\\\/g''' HEADER

INSERT INTO tags (
    Id, tagname, count, ExcerptPostId, WikiPostId, IsModeratorOnly, IsRequired)
SELECT
     (xpath('//@Id', x))[1]::text::int
    ,(xpath('//@TagName', x))[1]::text
    ,(xpath('//@Count', x))[1]::text::int
    ,(xpath('//@ExcerptPostId', x))[1]::text::int
    ,(xpath('//@WikiPostId', x))[1]::text::int
    ,(xpath('//@IsModeratorOnly', x))[1]::text::boolean
    ,(xpath('//@IsRequired', x))[1]::text::boolean
FROM data_load, unnest(xpath('//row', data::xml)) x
WHERE regexp_like(data,'[ ]+\<row');

-- Loading votes

truncate data_load;

\copy data_load from program 'tr -d "\t" < Votes.xml | sed -e ''s/\\/\\\\/g''' HEADER

INSERT INTO votes (
    Id, postid, votetypeid, creationdate)
SELECT
     (xpath('//@Id', x))[1]::text::int
    ,(xpath('//@PostId', x))[1]::text::int
    ,(xpath('//@VoteTypeId', x))[1]::text::int
    ,(xpath('//@CreationDate', x))[1]::text::timestamp
FROM data_load, unnest(xpath('//row', data::xml)) x
WHERE regexp_like(data,'[ ]+\<row');

Query Time

Once the data is loaded, we can start querying the dataset. For example, finding the top 2 post having a comment with highest score using the following SQL:

SELECT POSTS.Id, 
    POSTS.Title, 
    COMMENTS.Text, 
    COMMENTS.SCORE
FROM
POSTS JOIN COMMENTS ON POSTS.ID = COMMENTS.POSTID
ORDER BY SCORE DESC
LIMIT 2;

Enjoy the dataset and find your own questions and answers!

The following links provide the entire set of DDLs and Loading SQL!

Managing data drift with Apache Kafka® Connect and a schema registry

Francesco Tisiot — Mon, 29 Jan 2024 14:00:00 +0000

Data flows across many technologies, teams, and people in today's businesses. Businesses are always growing and changing, so the way we collect and share data changes all the time too. We need to know not only who owns certain data but also what to do if that data changes. This problem is often referred to as "data drift."

Consider the scenario where a piece of data is modified at its source — what implications does this have for other systems reliant on it? How do we communicate necessary changes to stakeholders? Conversely, how do we prevent changes that could disrupt the system?

Having a robust plan for managing data drift is imperative. Businesses require data systems that function seamlessly and remain consistent, even amidst changes at the data source. Additionally, mechanisms are needed to assess and decide on changes, ensuring smooth operations for everyone involved with the data.

This tutorial will show you how tools like Apache Kafka®, Apache Kafka Connect, and the built in schema registry functionality provided by Karapace, can help businesses keep an eye on data drift. It will also explain how to either deny or allow changes based on what a business needs.

Why Apache Kafka and why a schema registry?

Apache Kafka is widely adopted as a backend data hub, empowering companies to move their data supported by a reliable, fast and scalable technology. Apache Kafka provides the benefit of decoupling data producers and consumers, by allowing the producers to reliably send the data without having to worry about consumers being ready to read, or being fast enough to keep up with throughput.

By default, Apache Kafka doesn't impose or verify the structure of data. Messages are pushed and retrieved in any format agreed upon by the producer and the consumer. However, in complex systems, where the same information needs to be reused across multiple consumers from different parts of the company, a simple external agreement is often insufficient. Apache Kafka must not only ensure that consumers can retrieve the data but also make sense of it, even if the structure of the messages changes slightly over time.

This is where the schema registry functionality enabled by Karapace comes into play: a way to decouple the structure of the message from its content and a method to verify that updates in the data structure won't break downstream consumers of the information. With Karapace, we can define the structure of each topic, along with the compatibility level that determines which data structure changes are allowed or rejected.

In the following sections, we will explore how the schema registry can be used in conjunction with Apache Kafka® Connect, both as a source and a sink, to check data structure changes and propagate them if they meet compatibility requirements..

The overall architecture

To simulate a typical company data flow, we will employ PostgreSQL® as our source, serving as our transactional database. Extracting data from it will involve using Apache Kafka, Apache Kafka Connect, and the Debezium source connector, enabling a real-time change data capture process. Once the data resides in Apache Kafka, we will leverage the integrated integration with Karapace to store the data schema and assess changes for compatibility. Finally, the results of our data changes will manifest in a MySQL database and an Amazon S3 bucket, mirroring two use cases: departmental analytics and long-term data storage.

We'll use Aiven for Apache Kafka®, Aiven for PostgreSQL®, Aiven for MySQL and a Debezium Kafka Connector to demonstrate this. Sign up for an Aiven account to follow along.

We can create the whole flow using Aiven's command line interface. You'll also need to install psql. Run the following commands:

avn service create demo-drift-postgresql -t pg --cloud aws-eu-west-1 -p free-1-5gb
avn service create demo-drift-mysqldb -t mysql --cloud aws-eu-west-1 -p free-1-5gb
avn service create demo-drift-kafka         \
    -t kafka                                \
    --cloud aws-eu-west-1                   \
    -p business-4                           \
    -c kafka.auto_create_topics_enable=true \
    -c kafka_connect=true                   \
    -c kafka_rest=true                      \
    -c schema_registry=true

The above three commands will start:

An Aiven for PostgreSQL database named demo-drift-postgresql in the aws-eu-west-1 cloud region using Aiven's free tier
An Aiven for MySQL database named demo-drift-mysql in the aws-eu-west-1 cloud region using Aiven's free tier
An Aiven for Apache Kafka® service named demo-drift-kafka in the aws-eu-west-1 cloud region using Aiven's business-4 plan and enabling:
- The automatic creation of topics
- Apache Kafka Connect, running on the same nodes as Apache Kafka
- Kafka REST APIs
- Kafka Schema Registry functionality powered by Karapace

We can wait for the above services to be created with:

avn service wait demo-drift-postgresql
avn service wait demo-drift-kafka
avn service wait demo-drift-mysqldb

Create the source dataset in PostgreSQL

The first step of the data journey will be in the PostgreSQL database, acting as a company transactional backend. In this section we'll connect to the database and include some data. To connect, we can use the prebuilt Aiven CLI command (that requires psql to be installed locally):

avn service cli demo-drift-postgresql

After connecting, we can create a basic USERS table and include some data:

CREATE TABLE USERS (ID SERIAL PRIMARY KEY, USERNAME VARCHAR, HERO BOOLEAN);

INSERT INTO USERS (USERNAME, HERO)
  VALUES ('Spiderman', TRUE), ('Flash', TRUE), ('Joker', FALSE), ('Batman', TRUE);

Change Data Capture from PostgreSQL to Apache Kafka

After mimicking the OLTP (Online Transaction Processing) system, we can now create the change data capture pipeline allowing us to track the USERS table in Apache Kafka. We'll set up the CDC flow using a Debezium connector and the following configuration file, which we'll name cdc-deb.json. Be sure to replace values like <DATABASE_HOST> in the below example.

{   
    "name": "pg-source-users",
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.server.name": "sourcepg",
    "database.hostname": "<DATABASE_HOST>",
    "database.port": "<DATABASE_PORT>",
    "database.user": "avnadmin",
    "database.password": "<POSTGRESQL_PASSWORD>",
    "database.dbname": "defaultdb",
    "plugin.name": "pgoutput",
    "slot.name": "myslot1",
    "publication.name": "mypub1",
    "publication.autocreate.mode": "filtered",
    "database.sslmode": "require",
    "table.include.list": "public.users",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "https://<KAFKA_HOST>:<SCHEMA_REGISTRY_PORT>",
    "key.converter.basic.auth.credentials.source": "USER_INFO",
    "key.converter.schema.registry.basic.auth.user.info": "avnadmin:<SCHEMA_REGSITRY_PASSWORD>",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "https://<KAFKA_HOST>:<SCHEMA_REGISTRY_PORT>",
    "value.converter.basic.auth.credentials.source": "USER_INFO",
    "value.converter.schema.registry.basic.auth.user.info": "avnadmin:<SCHEMA_REGISTRY_PASSWORD>"
}

In the above connector we are defining:

The Debezium PostgreSQL connector in the connector.class parameter
The PostgreSQL connection settings in the set of database.* parameters, We can get the list of needed parameters with the following call:

  avn service get demo-drift-postgresql --format '{service_uri_params}'

The PostgreSQL replication plugin name, slot name, publication name and mode. We can either create the slot and publication in PostgreSQL beforehand or have the connector create them for us.
The list of tables to include in the replication (public.users)
The usage of Avro and Apache Kafka schema registry functionality for both message keys and values. We can fetch the needed connection parameters (<KAFKA_HOST>, <SCHEMA_REGISTRY_PORT>, <SCHEMA_REGISTRY_PASSWORD>)

  avn service get demo-drift-kafka --json | jq -r '.connection_info.schema_registry_uri'

The above command will report the Kafka schema registry URI in the form:

  https://avnadmin:<SCHEMA_REGISTRY_PASSWORD>@<KAFKA_HOST>:<SCHEMA_REGISTRY_PORT>

Once we've replaced the placeholder values in the file, we can create the connector with the following call where cdc-deb.json is the file containing the connector settings:

avn service connector create demo-drift-kafka @cdc-deb.json

Check the data in Kafka

Once the connector is working, we can use kcat to check the data in Apache Kafka.

To get the kcat command for connecting to our Kafka service and also download the necessary SSL certificates, run:

avn service connection-info kcat demo-drift-kafka -W

Next, we can get the avnadmin password with:

avn service user-list --format '{password}' --project devrel-francesco demo-drift-kafka

Finally, we can take that kcat command and use it to check the data.
We need to add some parameters to explain what we want to read:

-C to tell it to act as a Consumer,
-t sourcepg.public.users to tell it which topic to read from,
-s avro to tell it to use Avro, and
-r https://avnadmin:<SCHEMA_REGISTRY_PWD>@<KAFKA_HOST>:<SCHEMA_REGISTRY_PORT> to tell it where the schema registry is

Putting all of that together, the command you run should look like this:

kcat -b <KAFKA_HOST>:<KAFKA_PORT> \
    -X security.protocol=SSL \
    -X ssl.ca.location=ca.pem \
    -X ssl.key.location=service.key \
    -X ssl.certificate.location=service.cert \
    -s avro \
    -r https://avnadmin:<SCHEMA_REGISTRY_PWD>@<KAFKA_HOST>:<SCHEMA_REGISTRY_PORT> \
    -C -t sourcepg.public.users

We should see the same four rows we inserted previously appearing in the standard Debezium format:

{"id": 1}{"before": null, "after": {"Value": {"id": 1, "username": {"string": "Spiderman"}, "hero": {"boolean": true}}}, "source": {"version": "1.9.7.aiven", "connector": "postgresql", "name": "sourcepg", "ts_ms": 1692017248192, "snapshot": {"string": "true"}, "db": "defaultdb", "sequence": {"string": "[null,\"235090528\"]"}, "schema": "public", "table": "users", "txId": {"long": 1036}, "lsn": {"long": 235090528}, "xmin": null}, "op": "r", "ts_ms": {"long": 1692017248487}, "transaction": null}
{"id": 2}{"before": null, "after": {"Value": {"id": 2, "username": {"string": "Flash"}, "hero": {"boolean": true}}}, "source": {"version": "1.9.7.aiven", "connector": "postgresql", "name": "sourcepg", "ts_ms": 1692017248192, "snapshot": {"string": "true"}, "db": "defaultdb", "sequence": {"string": "[null,\"235090528\"]"}, "schema": "public", "table": "users", "txId": {"long": 1036}, "lsn": {"long": 235090528}, "xmin": null}, "op": "r", "ts_ms": {"long": 1692017248493}, "transaction": null}
{"id": 3}{"before": null, "after": {"Value": {"id": 3, "username": {"string": "Joker"}, "hero": {"boolean": false}}}, "source": {"version": "1.9.7.aiven", "connector": "postgresql", "name": "sourcepg", "ts_ms": 1692017248192, "snapshot": {"string": "true"}, "db": "defaultdb", "sequence": {"string": "[null,\"235090528\"]"}, "schema": "public", "table": "users", "txId": {"long": 1036}, "lsn": {"long": 235090528}, "xmin": null}, "op": "r", "ts_ms": {"long": 1692017248494}, "transaction": null}
{"id": 4}{"before": null, "after": {"Value": {"id": 4, "username": {"string": "Batman"}, "hero": {"boolean": true}}}, "source": {"version": "1.9.7.aiven", "connector": "postgresql", "name": "sourcepg", "ts_ms": 1692017248192, "snapshot": {"string": "last"}, "db": "defaultdb", "sequence": {"string": "[null,\"235090528\"]"}, "schema": "public", "table": "users", "txId": {"long": 1036}, "lsn": {"long": 235090528}, "xmin": null}, "op": "r", "ts_ms": {"long": 1692017248494}, "transaction": null}

Check the data definition in Karapace

Having created the connector using Avro and the Karapace schema registry, we can examine the schema definition for the topic. By default, when utilizing Kafka Connect with a schema registry, two schemas are generated with names -value and -key to store the schema definition for the value and key, respectively.

We can get the list of schemas defined in Karapace with:

curl https://avnadmin:<SCHEMA_REGISTRY_PASSWORD>@<KAFKA_HOST>:<KAFKA_PORT>/subjects

Which returns output similar to:

["sourcepg.public.users-key","sourcepg.public.users-value"]

The above is the names of the two schemas for the Debezium topic. Each name is the concatenation of the database.server.name parameter (sourcepg), the schema and table name (public.users) and either the key or value suffix.

We can check which versions we have for the sourcepg.public.users-key topic with:

curl -X GET https://avnadmin:<SCHEMA_REGISTRY_PASSWORD>@<KAFKA_HOST>:<KAFKA_PORT>/subjects/sourcepg.public.users-key/versions

The output should show version 1 being available.

To check the definition of the schema sourcepg.public.users-key version 1 we can use the following command:

curl -X GET https://avnadmin:<SCHEMA_REGISTRY_PASSWORD>@<KAFKA_HOST>:<KAFKA_PORT>/subjects/sourcepg.public.users-key/versions/1

The output shows all the fields included in the key, including the id and name we defined in the original PostgreSQL table.

{
  "id": 1,
  "schema": "{\"connect.name\":\"sourcepg.public.users.Key\",\"fields\":[{\"default\":0,\"name\":\"id\",\"type\":{\"connect.default\":0,\"type\":\"int\"}}],\"name\":\"Key\",\"namespace\":\"sourcepg.public.users\",\"type\":\"record\"}",
  "subject": "sourcepg.public.users-key",
  "version": 1
}

Sink the data to MySQL

Now that we have the data in Apache Kafka, let's set up a consumer for the data to demonstrate how the solution manages drift. The initial consumer will be a MySQL database. We can establish the flow using a dedicated JDBC sink connector and the following code stored in mysql_jdbc_sink.json.

{
    "name": "cdc-sink-mysql",
    "connector.class": "io.aiven.connect.jdbc.JdbcSinkConnector",
    "topics": "sourcepg.public.users",
    "transforms": "extract",
    "connection.url": "jdbc:mysql://<MYSQL_HOST>:<MYSQL_PORT>/<MYSQL_DB_NAME>?ssl-mode=REQUIRED",
    "connection.user": "avnadmin",
    "connection.password": "<MYSQL_PASSWORD>",
    "table.name.format": "users_mysql",
    "insert.mode": "upsert",
    "pk.mode": "record_key",
    "pk.fields": "id",
    "auto.create": "true",
    "auto.evolve": "true",
    "transforms": "extract",
    "transforms.extract.type": "io.debezium.transforms.ExtractNewRecordState",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "https://<KAFKA_HOST>:<SCHEMA_REGISTRY_PORT>",
    "key.converter.basic.auth.credentials.source": "USER_INFO",
    "key.converter.schema.registry.basic.auth.user.info": "avnadmin:<SCHEMA_REGISTRY_PASSWORD>",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "https://<KAFKA_HOST>:<SCHEMA_REGISTRY_PORT>",
    "value.converter.basic.auth.credentials.source": "USER_INFO",
    "value.converter.schema.registry.basic.auth.user.info": "avnadmin:<SCHEMA_REGISTRY_PASSWORD>"
}

In the above connector we are defining:

The JDBC sink connector in the connector.class parameter
The MySQL connection settings in the connection.url parameter, We can get the parameters to compose the URL and the credentials with the following call

  avn service get demo-drift-mysqldb --format '{service_uri_params}'

The target table name will be users_mysql with upsert mode (see insert.mode), inserting or updating existing rows based on the id field (see pk.mode and pk.fields parameters)
The table will be created automatically if it does not exist ("auto.create": "true") and evolve following the changes in the Apache Kafka topic ("auto.evolve": "true"). This will be key to propagating the drift to downstream technologies (MySQL in this case).
A transformation called extract to retrieve and propagate the status of the row after the change from the Debezium format
The usage of Avro and Apache Kafka schema registry functionality for both message keys and values. We can fetch the needed connection parameters (<KAFKA_HOST>, <SCHEMA_REGISTRY_PORT>, <SCHEMA_REGISTRY_PASSWORD>)

  avn service get demo-drift-kafka --json | jq -r '.connection_info.schema_registry_uri'

The above command will provide the Kafka schema registry uri in the form:

  https://avnadmin:<SCHEMA_REGISTRY_PASSWORD>@<KAFKA_HOST>:<SCHEMA_REGISTRY_PORT>

Having replaced the placeholder values, we can create the connector with the following call, where cdc-deb.json is the file containing the connector settings:

avn service connector create demo-drift-kafka @mysql_jdbc_sink.json

We can verify the status of the connector with:

avn service connector status demo-drift-kafka cdc-sink-mysql

The above command should show the connector in RUNNING state

Check the data in MySQL

Once the above connector is running, we can head to MySQL to check the data. To get the connection parameters, we can retype the following command:

avn service get demo-drift-mysqldb --format '{service_uri_params}'

And then connect with the following command, replacing the placeholders. Note the absence of spaces between the -p parameter and the password.

mysql -u avnadmin   \
    -P <MYSQL_PORT>        \
    -h <MYSQL_HOST> \
    -D defaultdb    \
    -p<MYSQL_PASSWORD>

We can then check the data with:

select * from users_mysql;

The table is users_mysql following the table.name.format in the connector. The data should be in line with what we have in PostgreSQL.

+----+-----------+------+
| id | username  | hero |
+----+-----------+------+
|  1 | Spiderman |    1 |
|  2 | Flash     |    1 |
|  3 | Joker     |    0 |
|  4 | Batman    |    1 |
+----+-----------+------+

If we check the table structure with describe users_mysql, we can see that the hero column has been mapped to a TINYINT in MySQL.

+----------+--------------+------+-----+---------+-------+
| Field    | Type         | Null | Key | Default | Extra |
+----------+--------------+------+-----+---------+-------+
| id       | int          | NO   | PRI | 0       |       |
| username | varchar(256) | YES  |     | NULL    |       |
| hero     | tinyint      | YES  |     | NULL    |       |
| points   | int          | YES  |     | NULL    |       |
+----------+--------------+------+-----+---------+-------+

Let's talk Drift

So far we've built a fairly traditional data pipeline. Now, let's include some changes to the original data structure in PostgreSQL to mimic drift.

Adding a column

In the terminal connected to the PostgreSQL database, execute the following command to add a POINTS integer column:

ALTER TABLE USERS ADD COLUMN POINTS INT;

Nothing happens immediately in the target MySQL table after the DDL execution in PostgreSQL. The structure and the data of USERS_MYSQL is still the same.

Now change the data in PostgreSQL, using the following update statement:

UPDATE USERS SET POINTS = CASE WHEN USERNAME = 'Batman' then 5 else 10 end;

In MySQL, execute:

SELECT * FROM users_mysql;

We can see the effect on the MySQL table points in near real time:

+----+-----------+------+--------+
| id | username  | hero | points |
+----+-----------+------+--------+
|  1 | Spiderman |    1 |     10 |
|  2 | Flash     |    1 |     10 |
|  3 | Joker     |    0 |     10 |
|  4 | Batman    |    1 |      5 |
+----+-----------+------+--------+

As mentioned in the sink connector definition, "auto.create": "true" allows the automatic creation of the table if it doesn't exist, and "auto.evolve": "true" allows the evolution of the table in cases when new data columns are included.

Removing a column

What about removing columns? Let's test it! Let's drop the same points column we just added from the PostgreSQL terminal with:

ALTER TABLE USERS DROP COLUMN POINTS;

If we execute our previous query in MySQL again:

SELECT * FROM users_mysql;

We see that column is not dropped in MySQL, the structure of the users_mysql is the same and the points column is still filled.

+----+-----------+------+--------+
| id | username  | hero | points |
+----+-----------+------+--------+
|  1 | Spiderman |    1 |     10 |
|  2 | Flash     |    1 |     10 |
|  3 | Joker     |    0 |     10 |
|  4 | Batman    |    1 |      5 |
+----+-----------+------+--------+

This makes sense because downstream applications might be using the points column. An unexpected and unhandled drop of a column could have disastrous effects on the downstream data pipelines. However the risk actually is dealing with updated information, as the points column has been dropped from PostgreSQL and therefore cannot be updated.

Changing the column type

What about changing the column type? A change in the column type could be needed in cases, like this example, where we want to migrate from a BOOLEAN to a VARCHAR for the HERO column. Let's execute the following in PostgreSQL:

ALTER TABLE USERS ALTER COLUMN HERO TYPE VARCHAR;

As before nothing happens on the DDL statement, but, when we try to add some data using the new VARCHAR column type:

INSERT INTO USERS (USERNAME, HERO) VALUES ('Panda', 'middle');

The insert goes well PostgreSQL as expected, but the Debezium source connector crashes with the following error:

ERROR "Caused by: org.apache.kafka.common.config.ConfigException: Failed to access Avro data from topic sourcepg.public.users : Incompatible schema, compatibility_mode=BACKWARD reader union lacking writer type: RECORD; error code: 409"
Backwards compatibility, old schema type is boolean (with null), new schema type is string... incompatible

This is because the schema is stored in Karapace with the BACKWARDS compatibility setting. The BACKWARDS compatibility ensures that consumers using an older schema definition are able to consume events produced with the current schema. The change, from BOOLEAN to VARCHAR could stop old consumers from being able to parse the information correctly, so it's not allowed and the connector fails.

Changing the compatibility level

For the sake of this example, let's remove the BACKWARDS compatibility setting and allowing all changes in the source system to propagate. We'll set compatibility to NONE allowing all the changes to propagate to the Apache Kafka topic.

First, we check the default compatibility level for the Apache Kafka service with:

avn service schema configuration demo-drift-kafka

This shows BACKWARD being the default. The same default setting is applied to the sourcepg.public.users-value topic, that we can check with:

avn service schema subject-configuration demo-drift-kafka \
    --subject sourcepg.public.users-value

To change the compatibility level to NONE for both key and value, run the following commands:

avn service schema subject-configuration-update demo-drift-kafka \
    --subject sourcepg.public.users-value                        \
    --compatibility NONE
avn service schema subject-configuration-update demo-drift-kafka \
    --subject sourcepg.public.users-key                          \
    --compatibility NONE

Now, if we restart the Debezium Source connector task 0 with:

avn service connector restart-task demo-drift-kafka pg-source-users 0

We see that the source connector restarts correctly. Using avn service connector status demo-drift-kafka pg-source-users shows the connector in the RUNNING state:

    {
      "status": {
          "state": "RUNNING",
          "tasks": [
              {
                  "id": 0,
                  "state": "RUNNING",
                  "trace": ""
              }
          ]
      }
  }

The JDBC sink connector to MySQL fails. Running avn service connector status demo-drift-kafka cdc-sink-mysql returns an error:

Caused by: org.apache.kafka.connect.errors.ConnectException: java.sql.SQLException: java.sql.BatchUpdateException: Incorrect integer value: 'middle' for column 'hero' at row 1
java.sql.SQLException: Incorrect integer value: 'middle' for column 'hero' at row 1

The error indicates that the connector attempted to insert the new value (middle) into an integer column. This implies that the auto-evolution process did not alter the structure of the pre-existing column.

To confirm this, we can execute describe users_mysql on the MySQL database and validate that the hero column remains a tinyint.

In the JDBC sink connector documentation, the auto.evolution section says:

The connector does not delete columns.

The connector does not alter column types.

The connector does not add primary keys constraints.

We already talked about automatic column deletion being a dangerous action. The same is true for the automatic change of column types, since downstream applications could rely on functions that work specifically on particular column types. Therefore, modifying a column type should be handled as a breaking change, correctly making the sink connector fail.

What about non relational targets? The AWS S3 example

The scenario described above is one of the more strict scenarios possible, in terms of data evolution. Both the source and the target are relational databases with strict column type definition. In this second example we'll sink the data to an S3 bucket where the data structure is not defined upfront.

We can create a sink connector to S3 with the following JSON configuration file stored in a file named s3_sink.json

{
    "name": "s3sink",
    "connector.class": "io.aiven.kafka.connect.s3.AivenKafkaConnectS3SinkConnector",
    "aws.access.key.id": "<AWS_SECRET_ID>",
    "aws.secret.access.key": "<AWS_SECRET_ACCESS>",
    "aws.s3.bucket.name": ">AWS_BUCKET_NAME>",
    "aws.s3.region": "<AWS_REGION>",
    "topics": "sourcepg.public.users",
    "format.output.type": "json",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "https://<KAFKA_HOST>:<SCHEMA_REGISTRY_PORT>",
    "key.converter.basic.auth.credentials.source": "USER_INFO",
    "key.converter.schema.registry.basic.auth.user.info": "avnadmin:<SCHEMA_REGISTRY_PASSWORD>",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "https://<KAFKA_HOST>:<SCHEMA_REGISTRY_PORT>",
    "value.converter.basic.auth.credentials.source": "USER_INFO",
    "value.converter.schema.registry.basic.auth.user.info": "avnadmin:<SCHEMA_REGISTRY_PASSWORD>",
    "transforms": "extract",
    "transforms.extract.type": "io.debezium.transforms.ExtractNewRecordState"
}

Where:

The set of aws.* parameters refers to the S3-related secrets, as detailed in the connector prerequisites documentation.
The topics parameter defines the source of information (specifically, the sourcepg.public.users topic).
The format.output.type parameter specifies how the data will be stored (in this case, as json).
The key.converter and value.converter parameters enable the connector to retrieve schema information from Karapace.
The transforms section allows the extraction of the value after the change from the Debezium format.

We can start the above connector with:

avn service connector create demo-drift-kafka @s3_sink.json

If we check the data in S3, we should see a document in the bucket containing all the changes implemented in PostgreSQL.

[
{"value":{"id":1,"username":"Spiderman","hero":true}},
{"value":{"id":2,"username":"Flash","hero":true}},
{"value":{"id":3,"username":"Joker","hero":false}},
{"value":{"id":4,"username":"Batman","hero":true}},
{"value":{"id":1,"username":"Spiderman","hero":true,"points":10}},
{"value":{"id":2,"username":"Flash","hero":true,"points":10}},
{"value":{"id":3,"username":"Joker","hero":false,"points":10}},
{"value":{"id":4,"username":"Batman","hero":true,"points":5}},
{"value":{"id":5,"username":"Panda","hero":"middle"}}
]

The output from the CDC -> Kafka -> S3 flow encompasses all events. Due to the Debezium compatibility mode being set to NONE, every change successfully stored in Kafka. Moreover, as S3 does not enforce a specific structure on the data, all changes, whether they involve new or deleted columns, have been written to the target bucket in JSON format.

Terminate the services

If you followed this tutorial and want to remove the services used for testing, you can run the commands below:

avn service terminate demo-drift-postgresql --force
avn service terminate demo-drift-kafka --force
avn service terminate demo-drift-mysqldb --force

Summary

Data and Schema drift must be managed in scenarios where multiple consumers want to access the changes happening in a source system. Apache Kafka, and the Karapace schema registry, provide a method to propagate compatible changes and forbid breaking ones by stopping the pipeline. Pay special attention to column drops, since they are not propagated automatically to target systems (specifically if the target is another relational database) and could cause problems with updated data on the deleted columns.

To summarize how changes are propagated:

Action	Status	Description
Add column	✅	Propagates downstream if `auto.evolve` is set to `true`.
Remove column	⚠️	Does not propagate downstream in case of sink to relational database. Possible use of stale data for the dropped column.
Change datatype	⚠️	Depends on the change, compatibility settings and target technology. Not propagated in case of JDBC sink.

A summary of Schema registry compatibility:

BACKWARDS allows you to stop the pipeline before ingesting data in Kafka, since breaking changes will not be included in the topic
NONE allows you to continue ingesting, but might break downstream data pipelines if the downstream tech is relational or has precise column definition and evolution is not straightforward

Forem: Francesco Tisiot

Introducing Developer Tier for Aiven for PostgreSQL® services

What do I get with the Developer tier?

Why three tiers?

Questions you might ask

Try it out!

Free PostgreSQL® in the Cloud - Aug 25

Helping PostgreSQL® professionals with AI-assisted performance recommendations

From content to tooling

Tooling for database professionals

Using trusted AI to optimize database workloads

Introducing Aiven's AI Database Optimizer: The First Built-In SQL Optimizer for Enhanced Performance

How does AI Database Optimizer work?

How does AI Database Optimizer help organizations?

How to use pgbench to test PostgreSQL® performance

Install pgbench locally

Create a test PostgreSQL environment

Use pgbench to load test the a PostgreSQL database

Step 1: Initialize the database

Step 2: execute a first test to set the baseline

Step 3: change parameters and validate the new performance

Step 4: define you custom workload in a SQL file

Step 5: Optimize your database

Conclusion

From dbf to PostgreSQL with Python

Download a sample dbf file

Use dbfread to move the data into PostgreSQL

List of PostgreSQL® AI Projects and Resources

11 Lessons to learn when using NULLs in PostgreSQL®

It all starts with some columns and rows

Insert NULLs

Inserting NULLs

Inserting NULLs in the primary key

Create a new column with NOT NULL constraint

Add a NOT NULL constraint to an existing column

Is DEFAULT enough?

Querying NULLs

Detecting NULLs with the IS NULL condition

Associating a default value to NULLs with COALESCE

Performing Math on NULL columns

Joining on NULL columns

Conclusion

How to CONCAT in MySQL

String concatenation in MySQL by placing strings next to each other

Concatenating a non string

Concatenating a NULL

String concatenation in MySQL with the || operator

Concatenating a non string

Concatenating a NULL value

String concatenation in MySQL with the CONCAT function

Concatenating a non string with the CONCAT function

Concatenating a NULL value with the CONCAT function

String aggregation via concatenation in MySQL with the GROUP_CONCAT function

Aggregating and concatenating a non string with the GROUP_CONCAT function

Aggregating and concatenating a NULL value with the GROUP_CONCAT function

Conclusion

How to load JSON data in PostgreSQL with the COPY command

PostgreSQL COPY command

PostgreSQL COPY command - the out of the box formats

The PROGRAM option to the rescue!

jq - the indispensable JSON parsing tool

Stitching all together

Load StackOverflow's StackExchange data in PostgreSQL®

Create a PostgreSQL database

Create database tables

Load the data

Step 1: Load the User XML into the data_load table

Step 2: Load the data into the Users table

Load the other tables

Query Time

Managing data drift with Apache Kafka® Connect and a schema registry

Why Apache Kafka and why a schema registry?

The overall architecture

Create the source dataset in PostgreSQL

Change Data Capture from PostgreSQL to Apache Kafka

Check the data in Kafka

Check the data definition in Karapace

Sink the data to MySQL

Check the data in MySQL

Let's talk Drift

Detecting NULLs with the `IS NULL` condition

String concatenation in MySQL with the `||` operator

String concatenation in MySQL with the `CONCAT` function

String aggregation via concatenation in MySQL with the `GROUP_CONCAT` function

The `PROGRAM` option to the rescue!

Step 1: Load the User XML into the `data_load` table

Step 2: Load the data into the `Users` table