Merge branch 'develop' into fix/clickhouse_array_type

CHANGELOG.md

@@ -8,8 +8,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## Unreleased
 
 * Added support for the Clickhouse `Date` type.
+
 * Fixed handling of the Clickhouse `Array` type.
 
+* Removed the check for duplicate primary keys on the Clickhouse dialect. This allows inserting multiple rows with the same primary key.
+
 ## v4.3.0
 
 * Added a check for non-existent columns in Clickhouse

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
 }
 

db/dialect_clickhouse.go

@@ -132,6 +132,10 @@ 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)
 

db/dialect_postgres.go

@@ -246,6 +246,10 @@ 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

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)
 	}