Add multithreaded operators #78

ritwizsinha · 2025-01-12T07:14:27Z

Fixes #72

dentiny · 2025-01-12T10:45:18Z

nit: Is formatting correct? Maybe we could run make format before commit (or add it into precommit hook).

dpxcc · 2025-01-14T20:57:55Z

test/sql/heavy_inserts_and_deletes.sql

@@ -0,0 +1,18 @@
+-- Create a temporary table for testing
+CREATE TEMPORARY TABLE test_table (


you are testing heap table, but not columnstore table here

no need to be a temp table

primary key and auto increment column impacts parallelism

dpxcc · 2025-01-14T21:00:34Z

test/sql/heavy_inserts_and_deletes.sql

+
+select * from test_table;
+-- Drop the temporary table
+DROP TABLE test_table;


nit: new line after each file

dpxcc · 2025-01-14T21:01:13Z

src/columnstore/execution/columnstore_delete.cpp

        }
        return SinkResultType::NEED_MORE_INPUT;
    }

+    SinkCombineResultType Combine(ExecutionContext &context, OperatorSinkCombineInput &input) const override {
+        auto &gstate = input.global_state.Cast<ColumnstoreDeleteGlobalState>();
+        auto &lstate_delete = input.local_state.Cast<ColumnstoreDeleteLocalState>();


nit: just lstate

dpxcc · 2025-01-14T21:03:41Z

src/columnstore/execution/columnstore_delete.cpp

+    SinkCombineResultType Combine(ExecutionContext &context, OperatorSinkCombineInput &input) const override {
+        auto &gstate = input.global_state.Cast<ColumnstoreDeleteGlobalState>();
+        auto &lstate_delete = input.local_state.Cast<ColumnstoreDeleteLocalState>();
+        gstate.row_ids.insert(lstate_delete.local_row_ids.begin(), lstate_delete.local_row_ids.end());


need lock on gstate to ensure thread-safe

Yeah, ofc, I misinterpreted Combine to be thread safe from the documentation, but multiple combines can run concurrently, added a lock

dpxcc · 2025-01-14T21:10:28Z

src/columnstore/execution/columnstore_delete.cpp

@@ -21,6 +21,11 @@ class ColumnstoreDeleteGlobalState : public GlobalSinkState {
    ColumnDataCollection return_collection;
 };

+class ColumnstoreDeleteLocalState : public LocalSinkState {
+public:
+    unordered_set<row_t> local_row_ids;


nit: i would just name it row_ids since the meaning is clear from the context, e.g. lstate.row_ids vs gstate.row_ids

dpxcc · 2025-01-14T21:12:00Z

src/columnstore/execution/columnstore_delete.cpp

@@ -101,5 +121,4 @@ unique_ptr<PhysicalOperator> Columnstore::PlanDelete(ClientContext &context, Log
    del->children.push_back(std::move(plan));
    return std::move(del);
 }
-


nit: add back the new line

dpxcc · 2025-01-14T21:15:53Z

src/columnstore/execution/columnstore_insert.cpp

    bool IsSink() const override {
        return true;
    }
+
+    bool ParallelSink() const override {
+        return true;


It seems that DuckDB doesn't always parallelize its PhysicalInsert (See DuckCatalog::PlanInsert)

Checked if the plan supports parallelism and if number of threads > 1, as done in PhysicalInsert

DuckDB also doesn't parallelize PhysicalInsert when there's RETURNING

dpxcc · 2025-01-14T21:19:43Z

src/columnstore/execution/columnstore_insert.cpp

-        : executor(context, bound_defaults), insert_count(0), return_collection(context, types) {
+        : executor(context, bound_defaults), insert_count(0), return_collection(context, types) {}
+
+    ExpressionExecutor executor;


this is not thread-safe to put in global state

Yeah, replicated what PhysicalInsert was doing with having this in the local state

dpxcc · 2025-01-14T21:23:15Z

src/columnstore/execution/columnstore_insert.cpp

            }
        }
        if (return_chunk) {
-            gstate.return_collection.Append(gstate.chunk);
+            lstate.return_collection.Append(lstate.chunk);


DuckDB directly writes to gstate.return_collecion. It appears that Append is thread-safe

I am not sure if Append is thread safe or not, atleast the documentation doesn't mention that

You are right, ColumnDataCollection::Append() is not thread-safe
I mis-read PhysicalInsert::Sink() that gstate.return_collection.Append() is only used under !parallel branch

dpxcc · 2025-01-14T21:24:02Z

You also need to parallelize ColumnstoreUpdate

ritwizsinha · 2025-01-21T19:18:21Z

A bit confused why make installcheck fails for updates, for some reason in this test case

CREATE TABLE t (a int, b text) USING columnstore;
INSERT INTO t VALUES (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd'), (5, 'e');
INSERT INTO t VALUES (2, 'f'), (3, 'g'), (4, 'h');
UPDATE t SET b = a + 1 WHERE a > 3;

Here while inserting, two parquet files get generated. While updating those files if we run a simple update on all, only a single file is updated, somehow the chunks passed to the sink operator running in threads differs from what it is being passed in a single threaded execution. In multi threaded execution both threads are getting jumbled data. Checked in gdb and the chunks passed to sink operators seem overwritten in some way.

dpxcc · 2025-01-21T20:04:46Z

I skimmed through ColumnstoreUpdate but didn’t notice anything obviously wrong. You’ll likely need to debug further in GDB

On a separate note, the current ColumnstoreUpdate::Sink() implementation is mostly serial. To improve parallelism, you’ll need to make the lock more fine-grained

Also, can we close #98? I assume it’s related to this PR

YuweiXiao · 2025-02-13T06:10:47Z

src/columnstore/execution/columnstore_insert.cpp

+            D_ASSERT(!return_chunk);
+            lock_guard<mutex> lock(gstate.insert_lock);
+            table.Insert(context.client, lstate.insert_chunk);
+            gstate.insert_count += lstate.insert_chunk.size();


iiuc, the parquet write is still serial with this PR?

my expectation of parallelism would be writing N data files concurrently (e.g., each local state produces its own data files). To avoid small-file problem, it could start with a single files (as in the current implementation) and dynamically open additional files when data keep flowing in.

btw, how duckdb determine the level of parallelism? simply equals to max_threads?

yep, the current ColumnstoreUpdate::Sink() implementation is still mostly serial

there is a GUC mooncake.maximum_threads to control the total number of threads

YuweiXiao · 2025-02-13T06:15:05Z

src/columnstore/execution/columnstore_insert.cpp

+                                               op.column_index_map, std::move(op.bound_defaults), op.return_chunk,
+                                               parallel_streaming_insert && num_threads > 1);
+    std::cout << "For parallelism number of threads: " << num_threads << "and streaming insert state"
+              << parallel_streaming_insert << std::endl;


logging with elog(DEBUG1, xxx)

should just get rid of temp logging

even if we want to log, pd_log is preferred over elog - we don't want to mix pg headers into columnstore code
https://github.com/duckdb/pg_duckdb/blob/main/include/pgduckdb/logger.hpp#L59

YuweiXiao · 2025-02-13T06:45:28Z

A bit confused why make installcheck fails for updates, for some reason in this test case
CREATE TABLE t (a int, b text) USING columnstore;
INSERT INTO t VALUES (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd'), (5, 'e');
INSERT INTO t VALUES (2, 'f'), (3, 'g'), (4, 'h');
UPDATE t SET b = a + 1 WHERE a > 3;
Here while inserting, two parquet files get generated. While updating those files if we run a simple update on all, only a single file is updated, somehow the chunks passed to the sink operator running in threads differs from what it is being passed in a single threaded execution. In multi threaded execution both threads are getting jumbled data. Checked in gdb and the chunks passed to sink operators seem overwritten in some way.

Not sure if this is related. DELETE/UPDATE comes after a SCAN, right? so is the mapping structure used for row_ids (and maintained in SCAN) thread-safe?

ritwizsinha added 2 commits January 12, 2025 12:43

Add multithreaded delete

7dc5205

Make insert operator multithreaded

5a378ce

Run make format

b9d0157

dpxcc reviewed Jan 14, 2025

View reviewed changes

dpxcc force-pushed the main branch from 64a3e22 to 6a655ea Compare January 15, 2025 22:11

ritwizsinha added 3 commits January 16, 2025 22:48

Address review comments

5de2e29

FIx regression diffs

abca396

Parallelise updates

515f990

ritwizsinha marked this pull request as draft January 16, 2025 20:56

Fix multithreading

c51c2f8

dpxcc mentioned this pull request Jan 20, 2025

Postgres crashing on inserting data with generate_series #98

Closed

Fix tests and delete operator

1efff92

ritwizsinha marked this pull request as ready for review January 20, 2025 10:44

ritwizsinha added 3 commits January 20, 2025 16:16

Add newline

e4f7244

Simplify delete

04ec88b

Fix insert

f5d0d49

YuweiXiao reviewed Feb 13, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add multithreaded operators #78

Add multithreaded operators #78

ritwizsinha commented Jan 12, 2025 •

edited

Loading

dentiny commented Jan 12, 2025

dpxcc Jan 14, 2025

dpxcc Jan 14, 2025

dpxcc Jan 14, 2025

ritwizsinha Jan 16, 2025

dpxcc Jan 14, 2025

ritwizsinha Jan 16, 2025

dpxcc Jan 14, 2025

ritwizsinha Jan 16, 2025

dpxcc Jan 14, 2025

ritwizsinha Jan 16, 2025

dpxcc Jan 14, 2025

ritwizsinha Jan 16, 2025 •

edited

Loading

dpxcc Jan 16, 2025

dpxcc Jan 14, 2025

ritwizsinha Jan 16, 2025

dpxcc Jan 14, 2025

ritwizsinha Jan 16, 2025

dpxcc Jan 16, 2025

dpxcc commented Jan 14, 2025

ritwizsinha commented Jan 21, 2025

dpxcc commented Jan 21, 2025

YuweiXiao Feb 13, 2025 •

edited

Loading

dpxcc Feb 13, 2025

YuweiXiao Feb 13, 2025

dpxcc Feb 13, 2025

YuweiXiao commented Feb 13, 2025 •

edited

Loading

		@@ -0,0 +1,18 @@
		-- Create a temporary table for testing
		CREATE TEMPORARY TABLE test_table (

Add multithreaded operators #78

Are you sure you want to change the base?

Add multithreaded operators #78

Conversation

ritwizsinha commented Jan 12, 2025 • edited Loading

dentiny commented Jan 12, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ritwizsinha Jan 16, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dpxcc commented Jan 14, 2025

ritwizsinha commented Jan 21, 2025

dpxcc commented Jan 21, 2025

YuweiXiao Feb 13, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

YuweiXiao commented Feb 13, 2025 • edited Loading

ritwizsinha commented Jan 12, 2025 •

edited

Loading

ritwizsinha Jan 16, 2025 •

edited

Loading

YuweiXiao Feb 13, 2025 •

edited

Loading

YuweiXiao commented Feb 13, 2025 •

edited

Loading