Fix insert-select + insert_deduplication_token bug by setting streams to 1

src/Interpreters/InterpreterInsertQuery.cpp

@@ -513,6 +513,12 @@ BlockIO InterpreterInsertQuery::execute()
                 const bool resize_to_max_insert_threads = !table->isView() && views.empty();
                 pre_streams_size = resize_to_max_insert_threads ? settings.max_insert_threads
                                                                 : std::min<size_t>(settings.max_insert_threads, pipeline.getNumStreams());
+
+                /// Deduplication when passing insert_deduplication_token breaks if using more than one thread
+                const String & deduplication_token = settings.insert_deduplication_token;
+                if (!deduplication_token.empty())
+                    pre_streams_size = 1;
+
                 if (table->supportsParallelInsert())
                     sink_streams_size = pre_streams_size;
             }

tests/queries/0_stateless/03001_insert_threads_deduplication.reference

@@ -0,0 +1,2 @@
+3
+6

tests/queries/0_stateless/03001_insert_threads_deduplication.sql

@@ -0,0 +1,42 @@
+DROP TABLE IF EXISTS landing SYNC;
+DROP TABLE IF EXISTS ds SYNC;
+
+CREATE TABLE landing
+(
+    timestamp DateTime64(3),
+    status String,
+    id String
+)
+ENGINE = MergeTree()
+ORDER BY timestamp;
+
+SYSTEM STOP MERGES landing; -- Stopping merges to force 3 parts
+
+INSERT INTO landing (status, id, timestamp) SELECT * FROM generateRandom() LIMIT 1;
+INSERT INTO landing (status, id, timestamp) SELECT * FROM generateRandom() LIMIT 1;
+INSERT INTO landing (status, id, timestamp) SELECT * FROM generateRandom() LIMIT 1;
+
+CREATE TABLE ds
+(
+    timestamp DateTime64(3),
+    status String,
+    id String
+)
+ENGINE = MergeTree()
+ORDER BY timestamp
+SETTINGS non_replicated_deduplication_window=1000;
+
+INSERT INTO ds SELECT * FROM landing
+SETTINGS insert_deduplicate=1, insert_deduplication_token='token1',
+         max_insert_threads=5;
+
+SELECT count() FROM ds;
+
+INSERT INTO ds SELECT * FROM landing
+SETTINGS insert_deduplicate=1, insert_deduplication_token='token2',
+         max_insert_threads=1;
+
+SELECT count() FROM ds;
+
+DROP TABLE IF EXISTS landing SYNC;
+DROP TABLE IF EXISTS ds SYNC;