Thread: How batch processing works

Hello,

Saw multiple threads around the same , so I want some clarification. As we know row by row is slow by slow processing , so in heavy write systems(say the client app is in Java) , people asked to do DMLS in batches rather in a row by row fashion to minimize the chatting or context switches between database and client which is resource intensive. What I understand is that , a true batch processing means the client has to collect all the input bind values and  prepare the insert statement and submit to the database at oneshot and then commit.

What it means actually and if we divide the option as below, which method truly does batch processing or there exists some other method for doing the batch processing considering postgres as backend database?

I understand, the first method below is truly a row by row processing in which context switches happen between client and database with each row, whereas the second method is just batching the commits but not a true batch processing as it will do the same amount of context switching between the database and client. But regarding the third and fourth method, will both execute in a similar fashion in the database with the same number of context switches? If any other better method exists to do these inserts in batches? Appreciate your guidance.


CREATE TABLE parent_table (
    id SERIAL PRIMARY KEY,
    name TEXT
);

CREATE TABLE child_table (
    id SERIAL PRIMARY KEY,
    parent_id INT REFERENCES parent_table(id),
    value TEXT
);


Method-1

insert into parent_table values(1,'a');
commit;
insert into parent_table values(2,'a');
commit;
insert into child_table values(1,1,'a');
Commit;
insert into child_table values(1,2,'a');
commit;

VS

Method-2

insert into parent_table values(1,'a');
insert into parent_table values(2,'a');
insert into child_table values(1,1,'a');
insert into child_table values(1,2,'a');  
Commit;

Method-4

INSERT INTO parent_table VALUES  (1, 'a'), (2, 'a');
INSERT INTO child_table VALUES   (1,1, 'a'), (1,2, 'a');
commit;

[snip] 

Method-4

INSERT INTO parent_table VALUES  (1, 'a'), (2, 'a');
INSERT INTO child_table VALUES   (1,1, 'a'), (1,2, 'a');
commit;

If I knew that I had to load a structured input data file (even if it had parent and child records), this is how I'd do it (but probably first try and see if "in-memory COPY INTO" is such a thing).

On Thu, Sep 19, 2024 at 11:31 AM Ron Johnson <ronljohnsonjr@gmail.com> wrote:

[snip] 

Method-4

INSERT INTO parent_table VALUES  (1, 'a'), (2, 'a');
INSERT INTO child_table VALUES   (1,1, 'a'), (1,2, 'a');
commit;

If I knew that I had to load a structured input data file (even if it had parent and child records), this is how I'd do it (but probably first try and see if "in-memory COPY INTO" is such a thing).

I was trying to reproduce this behaviour using row by row commit vs just batch commit vs true batch insert as you mentioned, i am not able to see any difference between "batch commit" and "true batch insert" response. Am I missing anything?

DO $$
DECLARE
    num_inserts INTEGER := 100000;
    batch_size INTEGER := 50;
    start_time TIMESTAMP;
    end_time TIMESTAMP;
    elapsed_time INTERVAL;
    i INTEGER;
BEGIN
    -- Method 1: Individual Inserts with Commit after every Row
    start_time := clock_timestamp();

    FOR i IN 1..num_inserts LOOP
        INSERT INTO parent_table VALUES (i, 'a');
        COMMIT;  
    END LOOP;

    end_time := clock_timestamp();
    elapsed_time := end_time - start_time;
    INSERT INTO debug_log (method1, start_time, end_time, elapsed_time)
    VALUES ('Method 1: Individual Inserts with Commit after every Row', start_time, end_time, elapsed_time);

    -- Method 2: Individual Inserts with Commit after 100 Rows
    start_time := clock_timestamp();

    FOR i IN 1..num_inserts LOOP
        INSERT INTO parent_table2 VALUES (i, 'a');
        -- Commit after every 100 rows
        IF i % batch_size = 0 THEN
            COMMIT;  
        END IF;
    END LOOP;

    -- Final commit if not already committed
   commit;

    end_time := clock_timestamp();
    elapsed_time := end_time - start_time;
    INSERT INTO debug_log (method1, start_time, end_time, elapsed_time)
    VALUES ('Method 2: Individual Inserts with Commit after 100 Rows', start_time, end_time, elapsed_time);

    -- Method 3: Batch Inserts with Commit after all
    start_time := clock_timestamp();

    FOR i IN 1..(num_inserts / batch_size) LOOP
        INSERT INTO parent_table3 VALUES
            (1 + (i - 1) * batch_size, 'a'),

            (49 + (i - 1) * batch_size, 'a'),
            (50 + (i - 1) * batch_size, 'a'));
COMMIT;
    END LOOP;

    COMMIT;  -- Final commit for all
    end_time := clock_timestamp();
    elapsed_time := end_time - start_time;
    INSERT INTO debug_log (method1, start_time, end_time, elapsed_time)
    VALUES ('Method 3: Batch Inserts with Commit after All', start_time, end_time, elapsed_time);

END $$;

On Thu, Sep 19, 2024 at 5:24 AM Lok P <loknath.73@gmail.com> wrote:

[snip] 

DO $$
DECLARE
    num_inserts INTEGER := 100000;
    batch_size INTEGER := 50;
    start_time TIMESTAMP;
    end_time TIMESTAMP;
    elapsed_time INTERVAL;
    i INTEGER;
BEGIN
    -- Method 1: Individual Inserts with Commit after every Row
    start_time := clock_timestamp();

    FOR i IN 1..num_inserts LOOP
        INSERT INTO parent_table VALUES (i, 'a');
        COMMIT;  
    END LOOP;

    end_time := clock_timestamp();
    elapsed_time := end_time - start_time;
    INSERT INTO debug_log (method1, start_time, end_time, elapsed_time)
    VALUES ('Method 1: Individual Inserts with Commit after every Row', start_time, end_time, elapsed_time);

    -- Method 2: Individual Inserts with Commit after 100 Rows
    start_time := clock_timestamp();

    FOR i IN 1..num_inserts LOOP
        INSERT INTO parent_table2 VALUES (i, 'a');
        -- Commit after every 100 rows
        IF i % batch_size = 0 THEN
            COMMIT;  
        END IF;
    END LOOP;

    -- Final commit if not already committed
   commit;

    end_time := clock_timestamp();
    elapsed_time := end_time - start_time;
    INSERT INTO debug_log (method1, start_time, end_time, elapsed_time)
    VALUES ('Method 2: Individual Inserts with Commit after 100 Rows', start_time, end_time, elapsed_time);

    -- Method 3: Batch Inserts with Commit after all
    start_time := clock_timestamp();

    FOR i IN 1..(num_inserts / batch_size) LOOP
        INSERT INTO parent_table3 VALUES
            (1 + (i - 1) * batch_size, 'a'),

[snip] 

            (49 + (i - 1) * batch_size, 'a'),
            (50 + (i - 1) * batch_size, 'a'));
COMMIT;
    END LOOP;

    COMMIT;  -- Final commit for all
    end_time := clock_timestamp();
    elapsed_time := end_time - start_time;
    INSERT INTO debug_log (method1, start_time, end_time, elapsed_time)
    VALUES ('Method 3: Batch Inserts with Commit after All', start_time, end_time, elapsed_time);

END $$;

Reproduce what behavior?

Anyway, plpgsql functions (including anonymous DO statements) are -- to Postgresql -- single statements.  Thus, they'll be faster than individual calls..

An untrusted language like plpython3u might speed things up even more, if you have to read a heterogeneous external file and insert all the records into the db.

Hi,

> On 19 Sep 2024, at 07:30, Lok P <loknath.73@gmail.com> wrote:
>
[snip]
>
> Method-4
>
> INSERT INTO parent_table VALUES  (1, 'a'), (2, 'a');
> INSERT INTO child_table VALUES   (1,1, 'a'), (1,2, 'a');
> commit;

I’ve done some batch processing of JSON messages from Kafka in Java.
By far the most performant way was to:

1. Use prepared statements
2. Parse JSON messages in Postgres
3. Process messages in batches

All three can be achieved by using arrays to pass batches:

WITH parsed AS (
  SELECT msg::json FROM unnest(?)
),
parents AS (
  INSERT INTO parent SELECT … FROM parsed RETURNING ...
)
INSERT INTO child SELECT … FROM parsed…

Not the single parameter that you can bind to String[]

Hope that helps.

On 2024-09-21 16:44:08 +0530, Lok P wrote:
> But wondering why we don't see any difference in performance between method-2
> and method-3 above.

The code runs completely inside the database. So there isn't much
difference between a single statement which inserts 50 rows and 50
statements which insert 1 row each. The work to be done is (almost) the
same.

This changes once you consider an application which runs outside of the
database (maybe even on a different host). Such an application has to
wait for the result of each statement before it can send the next one.
Now it makes a difference whether you are waiting 50 times for a
statement which does very little or just once for a statement which does
more work.

> So does it mean that,I am testing this in a wrong way or

That depends on what you want to test. If you are interested in the
behaviour of stored procedures, the test is correct. If you want to know
about the performance of a database client (whether its written in Java,
Python, Go or whatever), this is the wrong test. You have to write the
test in your target language and run it on the client system to get
realistic results (for example, the round-trip times will be a lot
shorter if the client and database are on the same computer than when
one is in Europe and the other in America).

For example, here are the three methods as Python scripts:

---------------------------------------------------------------------------------------------------
#!/usr/bin/python3

import time
import psycopg2

num_inserts = 10_000

db = psycopg2.connect()
csr = db.cursor()

csr.execute("drop table if exists parent_table")
csr.execute("create table parent_table (id int primary key, t text)")

start_time = time.monotonic()
for i in range(1, num_inserts+1):
    csr.execute("insert into parent_table values(%s, %s)", (i, 'a'))
    db.commit()
end_time = time.monotonic()
elapsed_time = end_time - start_time
print(f"Method 1: Individual Inserts with Commit after every Row: {elapsed_time:.3} seconds")

# vim: tw=99
---------------------------------------------------------------------------------------------------
#!/usr/bin/python3

import time
import psycopg2

num_inserts = 10_000
batch_size = 50

db = psycopg2.connect()
csr = db.cursor()

csr.execute("drop table if exists parent_table")
csr.execute("create table parent_table (id int primary key, t text)")
db.commit()

start_time = time.monotonic()
for i in range(1, num_inserts+1):
    csr.execute("insert into parent_table values(%s, %s)", (i, 'a'))
    if i % batch_size == 0:
        db.commit()
db.commit()
end_time = time.monotonic()
elapsed_time = end_time - start_time
print(f"Method 2: Individual Inserts with Commit after {batch_size}  Rows: {elapsed_time:.3} seconds")

# vim: tw=99
---------------------------------------------------------------------------------------------------
#!/usr/bin/python3

import itertools
import time
import psycopg2

num_inserts = 10_000
batch_size = 50

db = psycopg2.connect()
csr = db.cursor()

csr.execute("drop table if exists parent_table")
csr.execute("create table parent_table (id int primary key, t text)")
db.commit()

start_time = time.monotonic()
batch = []
for i in range(1, num_inserts+1):
    batch.append((i, 'a'))
    if i % batch_size == 0:
        q = "insert into parent_table values" + ",".join(["(%s, %s)"] * len(batch))
        params = list(itertools.chain.from_iterable(batch))
        csr.execute(q, params)
        db.commit()
        batch = []
if batch:
    q = "insert into parent_table values" + ",".join(["(%s, %s)"] * len(batch))
    csr.execute(q, list(itertools.chain(batch)))
    db.commit()
    batch = []

end_time = time.monotonic()
elapsed_time = end_time - start_time
print(f"Method 3: Batch Inserts ({batch_size})  with Commit after each batch: {elapsed_time:.3} seconds")

# vim: tw=99
---------------------------------------------------------------------------------------------------

On my laptop, method2 is about twice as fast as method3. But if I
connect to a database on the other side of the city, method2 is now more
than 16 times faster than method3 . Simply because the delay in
communication is now large compared to the time it takes to insert those
rows.

On 9/21/24 07:36, Peter J. Holzer wrote:
> On 2024-09-21 16:44:08 +0530, Lok P wrote:

> ---------------------------------------------------------------------------------------------------
> #!/usr/bin/python3
>
> import time
> import psycopg2
>
> num_inserts = 10_000
> batch_size = 50
>
> db = psycopg2.connect()
> csr = db.cursor()
>
> csr.execute("drop table if exists parent_table")
> csr.execute("create table parent_table (id int primary key, t text)")
> db.commit()
>
> start_time = time.monotonic()
> for i in range(1, num_inserts+1):
>      csr.execute("insert into parent_table values(%s, %s)", (i, 'a'))
>      if i % batch_size == 0:
>          db.commit()
> db.commit()
> end_time = time.monotonic()
> elapsed_time = end_time - start_time
> print(f"Method 2: Individual Inserts with Commit after {batch_size}  Rows: {elapsed_time:.3} seconds")
>
> # vim: tw=99
> ---------------------------------------------------------------------------------------------------

FYI, this is less of problem with psycopg(3) and pipeline mode:

import time
import psycopg

num_inserts = 10_000
batch_size = 50

db = psycopg.connect("dbname=test user=postgres host=104.237.158.68")
csr = db.cursor()

csr.execute("drop table if exists parent_table")
csr.execute("create table parent_table (id int primary key, t text)")
db.commit()

start_time = time.monotonic()
with db.pipeline():
     for i in range(1, num_inserts+1):
         csr.execute("insert into parent_table values(%s, %s)", (i, 'a'))
         if i % batch_size == 0:
             db.commit()
db.commit()
end_time = time.monotonic()
elapsed_time = end_time - start_time
print(f"Method 2: Individual Inserts(psycopg3 pipeline mode) with Commit
after {batch_size}  Rows: {elapsed_time:.3} seconds")


For remote to a database in another state that took the  time from:

Method 2: Individual Inserts with Commit after 50  Rows: 2.42e+02 seconds

to:

Method 2: Individual Inserts(psycopg3 pipeline mode) with Commit after
50  Rows: 9.83 seconds

> #!/usr/bin/python3
>
> import itertools
> import time
> import psycopg2
>
> num_inserts = 10_000
> batch_size = 50
>
> db = psycopg2.connect()
> csr = db.cursor()
>
> csr.execute("drop table if exists parent_table")
> csr.execute("create table parent_table (id int primary key, t text)")
> db.commit()
>
> start_time = time.monotonic()
> batch = []
> for i in range(1, num_inserts+1):
>      batch.append((i, 'a'))
>      if i % batch_size == 0:
>          q = "insert into parent_table values" + ",".join(["(%s, %s)"] * len(batch))
>          params = list(itertools.chain.from_iterable(batch))
>          csr.execute(q, params)
>          db.commit()
>          batch = []
> if batch:
>      q = "insert into parent_table values" + ",".join(["(%s, %s)"] * len(batch))
>      csr.execute(q, list(itertools.chain(batch)))
>      db.commit()
>      batch = []
>
> end_time = time.monotonic()
> elapsed_time = end_time - start_time
> print(f"Method 3: Batch Inserts ({batch_size})  with Commit after each batch: {elapsed_time:.3} seconds")
>
> # vim: tw=99
> ---------------------------------------------------------------------------------------------------

The above can also be handled with execute_batch() and execute_values()
from:

https://www.psycopg.org/docs/extras.html#fast-execution-helpers

>
> On my laptop, method2 is about twice as fast as method3. But if I
> connect to a database on the other side of the city, method2 is now more
> than 16 times faster than method3 . Simply because the delay in
> communication is now large compared to the time it takes to insert those
> rows.
>