From 4bd0d671d4804d505f7c6ffc1fd5eefe7539b845 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Frederik=20Sch=C3=B6ll?= <fschoell@posteo.de>
Date: Thu, 4 Apr 2024 17:13:05 +0200
Subject: [PATCH] allow multiple inserts for the same pk on Clickhouse

---
 db/dialect.go            | 1 +
 db/dialect_clickhouse.go | 3 +++
 db/dialect_postgres.go   | 2 ++
 db/ops.go                | 2 +-
 4 files changed, 7 insertions(+), 1 deletion(-)

diff --git a/db/dialect.go b/db/dialect.go
index 4e3ea5d..6a89e89 100644
--- a/db/dialect.go
+++ b/db/dialect.go
@@ -26,6 +26,7 @@ type dialect interface {
 	Flush(tx Tx, ctx context.Context, l *Loader, outputModuleHash string, lastFinalBlock uint64) (int, error)
 	Revert(tx Tx, ctx context.Context, l *Loader, lastValidFinalBlock uint64) error
 	OnlyInserts() bool
+	AllowPkDuplicates() bool
 	CreateUser(tx Tx, ctx context.Context, l *Loader, username string, password string, database string, readOnly bool) error
 }
 
diff --git a/db/dialect_clickhouse.go b/db/dialect_clickhouse.go
index 6387c1a..ea19ff5 100644
--- a/db/dialect_clickhouse.go
+++ b/db/dialect_clickhouse.go
@@ -131,6 +131,8 @@ func (d clickhouseDialect) OnlyInserts() bool {
 	return true
 }
 
+func (d clickhouseDialect) AllowPkDuplicates() bool { return true }
+
 func (d clickhouseDialect) CreateUser(tx Tx, ctx context.Context, l *Loader, username string, password string, _database string, readOnly bool) error {
 	user, pass := EscapeIdentifier(username), escapeStringValue(password)
 
@@ -237,6 +239,7 @@ func convertToType(value string, valueType reflect.Type) (any, error) {
 			if err != nil {
 				return "", fmt.Errorf("could not convert %s to time: %w", value, err)
 			}
+
 			return v.Unix(), nil
 		}
 		return "", fmt.Errorf("unsupported struct type %s", valueType)
diff --git a/db/dialect_postgres.go b/db/dialect_postgres.go
index 43152df..cd24696 100644
--- a/db/dialect_postgres.go
+++ b/db/dialect_postgres.go
@@ -246,6 +246,8 @@ func (d postgresDialect) OnlyInserts() bool {
 	return false
 }
 
+func (d postgresDialect) AllowPkDuplicates() bool { return false }
+
 func (d postgresDialect) CreateUser(tx Tx, ctx context.Context, l *Loader, username string, password string, database string, readOnly bool) error {
 	user, pass, db := EscapeIdentifier(username), password, EscapeIdentifier(database)
 	var q string
diff --git a/db/ops.go b/db/ops.go
index 2922ef0..b2cdb12 100644
--- a/db/ops.go
+++ b/db/ops.go
@@ -32,7 +32,7 @@ func (l *Loader) Insert(tableName string, primaryKey map[string]string, data map
 		l.entries.Set(tableName, entry)
 	}
 
-	if _, found := entry.Get(uniqueID); found {
+	if _, found := entry.Get(uniqueID); found && !l.getDialect().AllowPkDuplicates() {
 		return fmt.Errorf("attempting to insert in table %q a primary key %q, that is already scheduled for insertion, insert should only be called once for a given primary key", tableName, primaryKey)
 	}