diff --git a/.github/workflows/flow.yml b/.github/workflows/flow.yml index 24554347d..d874ed8f2 100644 --- a/.github/workflows/flow.yml +++ b/.github/workflows/flow.yml @@ -201,7 +201,137 @@ jobs: - name: Run ClickHouse run: | - ./clickhouse server & + cat > config1.xml < + + + + + + ::/0 + + default + default + 1 + 1 + + + none + var/lib/clickhouse + var/lib/clickhouse/tmp + var/lib/clickhouse/user_files + var/lib/clickhouse/format_schemas + 9000 + + + + + 1 + 1 + + + + localhost + 2181 + + + + /clickhouse/task_queue/ddl + + + + + + localhost + 9000 + + + + + localhost + 9001 + + + + + + EOF + cat > config2.xml < + + + + + + ::/0 + + default + default + 1 + 1 + + + none + var/lib/clickhouse + var/lib/clickhouse/tmp + var/lib/clickhouse/user_files + var/lib/clickhouse/format_schemas + 9001 + + + + + 2 + 1 + + + + localhost + 2181 + + + + /clickhouse/task_queue/ddl + + + + + + localhost + 9000 + + + + + localhost + 9001 + + + + + + EOF + cat > config-keeper.xml < + + 2181 + 1 + var/lib/clickhouse/coordination/log + var/lib/clickhouse/coordination/snapshots + + + 1 + localhost + 9234 + + + + + EOF + mkdir ch1 ch2 chkeep + (cd ch1 && ../clickhouse server -C ../config1.xml) & + (cd ch2 && ../clickhouse server -C ../config2.xml) & + (cd chkeep && ../clickhouse keeper -C ../config-keeper.xml) & - name: Install Temporal CLI uses: temporalio/setup-temporal@1059a504f87e7fa2f385e3fa40d1aa7e62f1c6ca # v0 diff --git a/flow/activities/flowable_core.go b/flow/activities/flowable_core.go index 1405f82ca..49180207a 100644 --- a/flow/activities/flowable_core.go +++ b/flow/activities/flowable_core.go @@ -225,7 +225,7 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon defer connectors.CloseConnector(ctx, dstConn) syncState.Store(shared.Ptr("updating schema")) - if err := dstConn.ReplayTableSchemaDeltas(ctx, config.Env, flowName, recordBatchSync.SchemaDeltas); err != nil { + if err := dstConn.ReplayTableSchemaDeltas(ctx, config.Env, flowName, options.TableMappings, recordBatchSync.SchemaDeltas); err != nil { return nil, fmt.Errorf("failed to sync schema: %w", err) } diff --git a/flow/connectors/bigquery/bigquery.go b/flow/connectors/bigquery/bigquery.go index 033eff097..3f27f1003 100644 --- a/flow/connectors/bigquery/bigquery.go +++ b/flow/connectors/bigquery/bigquery.go @@ -204,6 +204,7 @@ func (c *BigQueryConnector) ReplayTableSchemaDeltas( ctx context.Context, env map[string]string, flowJobName string, + _ []*protos.TableMapping, schemaDeltas []*protos.TableSchemaDelta, ) error { for _, schemaDelta := range schemaDeltas { diff --git a/flow/connectors/bigquery/qrep.go b/flow/connectors/bigquery/qrep.go index 25721c110..b2437e80b 100644 --- a/flow/connectors/bigquery/qrep.go +++ b/flow/connectors/bigquery/qrep.go @@ -87,7 +87,7 @@ func (c *BigQueryConnector) replayTableSchemaDeltasQRep( } if err := c.ReplayTableSchemaDeltas( - ctx, config.Env, config.FlowJobName, []*protos.TableSchemaDelta{tableSchemaDelta}, + ctx, config.Env, config.FlowJobName, nil, []*protos.TableSchemaDelta{tableSchemaDelta}, ); err != nil { return nil, fmt.Errorf("failed to add columns to destination table: %w", err) } diff --git a/flow/connectors/bigquery/qrep_avro_sync.go b/flow/connectors/bigquery/qrep_avro_sync.go index 73300f7da..4b92e87f9 100644 --- a/flow/connectors/bigquery/qrep_avro_sync.go +++ b/flow/connectors/bigquery/qrep_avro_sync.go @@ -98,7 +98,7 @@ func (s *QRepAvroSyncMethod) SyncRecords( slog.String(string(shared.FlowNameKey), req.FlowJobName), slog.String("dstTableName", rawTableName)) - if err := s.connector.ReplayTableSchemaDeltas(ctx, req.Env, req.FlowJobName, req.Records.SchemaDeltas); err != nil { + if err := s.connector.ReplayTableSchemaDeltas(ctx, req.Env, req.FlowJobName, req.TableMappings, req.Records.SchemaDeltas); err != nil { return nil, fmt.Errorf("failed to sync schema changes: %w", err) } diff --git a/flow/connectors/clickhouse/cdc.go b/flow/connectors/clickhouse/cdc.go index c067ffe14..1aa3fc57d 100644 --- a/flow/connectors/clickhouse/cdc.go +++ b/flow/connectors/clickhouse/cdc.go @@ -5,6 +5,7 @@ import ( "fmt" "log/slog" + chproto "github.com/ClickHouse/ch-go/proto" "github.com/ClickHouse/clickhouse-go/v2" _ "github.com/ClickHouse/clickhouse-go/v2/lib/driver" @@ -20,7 +21,18 @@ import ( const ( checkIfTableExistsSQL = `SELECT exists(SELECT 1 FROM system.tables WHERE database = %s AND name = %s) AS table_exists` - dropTableIfExistsSQL = "DROP TABLE IF EXISTS %s" + dropTableIfExistsSQL = "DROP TABLE IF EXISTS %s%s" + rawColumns = `( + _peerdb_uid UUID, + _peerdb_timestamp Int64, + _peerdb_destination_table_name String, + _peerdb_data String, + _peerdb_record_type Int, + _peerdb_match_data String, + _peerdb_batch_id Int64, + _peerdb_unchanged_toast_columns String + )` + zooPathPrefix = "/clickhouse/tables/{uuid}/{shard}/{database}/%s" ) // GetRawTableName returns the raw table name for the given table identifier. @@ -40,24 +52,42 @@ func (c *ClickHouseConnector) checkIfTableExists(ctx context.Context, databaseNa } func (c *ClickHouseConnector) CreateRawTable(ctx context.Context, req *protos.CreateRawTableInput) (*protos.CreateRawTableOutput, error) { + var rawDistributedName string rawTableName := c.GetRawTableName(req.FlowJobName) + engine := "MergeTree()" + if c.config.Replicated { + engine = fmt.Sprintf( + "ReplicatedMergeTree('%s%s','{replica}')", + zooPathPrefix, + peerdb_clickhouse.EscapeStr(rawTableName), + ) + } + onCluster := c.onCluster() + if onCluster != "" { + rawDistributedName = rawTableName + rawTableName += "_shard" + } - createRawTableSQL := `CREATE TABLE IF NOT EXISTS %s ( - _peerdb_uid UUID, - _peerdb_timestamp Int64, - _peerdb_destination_table_name String, - _peerdb_data String, - _peerdb_record_type Int, - _peerdb_match_data String, - _peerdb_batch_id Int64, - _peerdb_unchanged_toast_columns String - ) ENGINE = MergeTree() ORDER BY (_peerdb_batch_id, _peerdb_destination_table_name);` - - err := c.execWithLogging(ctx, - fmt.Sprintf(createRawTableSQL, rawTableName)) - if err != nil { + createRawTableSQL := `CREATE TABLE IF NOT EXISTS %s%s %s ENGINE = %s ORDER BY (_peerdb_batch_id, _peerdb_destination_table_name)` + if err := c.execWithLogging(ctx, + fmt.Sprintf(createRawTableSQL, peerdb_clickhouse.QuoteIdentifier(rawTableName), onCluster, rawColumns, engine), + ); err != nil { return nil, fmt.Errorf("unable to create raw table: %w", err) } + + if onCluster != "" { + createRawDistributedSQL := `CREATE TABLE IF NOT EXISTS %s%s %s ENGINE = Distributed(%s,%s,%s,cityHash64(_peerdb_uid))` + if err := c.execWithLogging(ctx, + fmt.Sprintf(createRawDistributedSQL, peerdb_clickhouse.QuoteIdentifier(rawDistributedName), onCluster, + rawColumns, + peerdb_clickhouse.QuoteIdentifier(c.config.Cluster), + peerdb_clickhouse.QuoteIdentifier(c.config.Database), + peerdb_clickhouse.QuoteIdentifier(rawTableName)), + ); err != nil { + return nil, fmt.Errorf("unable to create raw table: %w", err) + } + } + return &protos.CreateRawTableOutput{ TableIdentifier: rawTableName, }, nil @@ -101,7 +131,7 @@ func (c *ClickHouseConnector) syncRecordsViaAvro( } warnings := numericTruncator.Warnings() - if err := c.ReplayTableSchemaDeltas(ctx, req.Env, req.FlowJobName, req.Records.SchemaDeltas); err != nil { + if err := c.ReplayTableSchemaDeltas(ctx, req.Env, req.FlowJobName, req.TableMappings, req.Records.SchemaDeltas); err != nil { return nil, fmt.Errorf("failed to sync schema changes: %w", err) } @@ -133,17 +163,28 @@ func (c *ClickHouseConnector) ReplayTableSchemaDeltas( ctx context.Context, env map[string]string, flowJobName string, + tableMappings []*protos.TableMapping, schemaDeltas []*protos.TableSchemaDelta, ) error { if len(schemaDeltas) == 0 { return nil } + onCluster := c.onCluster() for _, schemaDelta := range schemaDeltas { if schemaDelta == nil || len(schemaDelta.AddedColumns) == 0 { continue } + var tm *protos.TableMapping + for _, tableMapping := range tableMappings { + if tableMapping.SourceTableIdentifier == schemaDelta.SrcTableName && + tableMapping.DestinationTableIdentifier == schemaDelta.DstTableName { + tm = tableMapping + break + } + } + for _, addedColumn := range schemaDelta.AddedColumns { qvKind := types.QValueKind(addedColumn.Type) clickHouseColType, err := qvalue.ToDWHColumnType( @@ -152,17 +193,30 @@ func (c *ClickHouseConnector) ReplayTableSchemaDeltas( if err != nil { return fmt.Errorf("failed to convert column type %s to ClickHouse type: %w", addedColumn.Type, err) } + + // Distributed table isn't created for null tables, no need to alter shard tables that don't exist + if c.config.Cluster != "" && (tm == nil || tm.Engine != protos.TableEngine_CH_ENGINE_NULL) { + if err := c.execWithLogging(ctx, + fmt.Sprintf("ALTER TABLE %s%s ADD COLUMN IF NOT EXISTS %s %s", + peerdb_clickhouse.QuoteIdentifier(schemaDelta.DstTableName+"_shard"), onCluster, + peerdb_clickhouse.QuoteIdentifier(addedColumn.Name), clickHouseColType), + ); err != nil { + return fmt.Errorf("failed to add column %s for table shards %s: %w", addedColumn.Name, schemaDelta.DstTableName, err) + } + } + if err := c.execWithLogging(ctx, - fmt.Sprintf("ALTER TABLE %s ADD COLUMN IF NOT EXISTS %s %s", - peerdb_clickhouse.QuoteIdentifier(schemaDelta.DstTableName), + fmt.Sprintf("ALTER TABLE %s%s ADD COLUMN IF NOT EXISTS %s %s", + peerdb_clickhouse.QuoteIdentifier(schemaDelta.DstTableName), onCluster, peerdb_clickhouse.QuoteIdentifier(addedColumn.Name), clickHouseColType), ); err != nil { return fmt.Errorf("failed to add column %s for table %s: %w", addedColumn.Name, schemaDelta.DstTableName, err) } c.logger.Info( - fmt.Sprintf("[schema delta replay] added column %s with data type %s", addedColumn.Name, clickHouseColType), - "destination table name", schemaDelta.DstTableName, - "source table name", schemaDelta.SrcTableName) + "[schema delta replay] added column", + slog.String("column", addedColumn.Name), slog.String("type", clickHouseColType), + slog.String("destination table name", schemaDelta.DstTableName), slog.String("source table name", schemaDelta.SrcTableName), + ) } } @@ -174,6 +228,7 @@ func (c *ClickHouseConnector) RenameTables( req *protos.RenameTablesInput, tableNameSchemaMapping map[string]*protos.TableSchema, ) (*protos.RenameTablesOutput, error) { + onCluster := c.onCluster() for _, renameRequest := range req.RenameTableOptions { if renameRequest.CurrentName == renameRequest.NewName { c.logger.Info("table rename is nop, probably Null table engine, skipping rename for it", @@ -202,17 +257,17 @@ func (c *ClickHouseConnector) RenameTables( c.logger.Info("attempting atomic exchange", slog.String("OldName", renameRequest.CurrentName), slog.String("NewName", renameRequest.NewName)) if err = c.execWithLogging(ctx, - fmt.Sprintf("EXCHANGE TABLES %s and %s", - peerdb_clickhouse.QuoteIdentifier(renameRequest.NewName), peerdb_clickhouse.QuoteIdentifier(renameRequest.CurrentName)), + fmt.Sprintf("EXCHANGE TABLES %s and %s%s", peerdb_clickhouse.QuoteIdentifier(renameRequest.NewName), + peerdb_clickhouse.QuoteIdentifier(renameRequest.CurrentName), onCluster), ); err == nil { if err := c.execWithLogging(ctx, - fmt.Sprintf(dropTableIfExistsSQL, peerdb_clickhouse.QuoteIdentifier(renameRequest.CurrentName)), + fmt.Sprintf(dropTableIfExistsSQL, peerdb_clickhouse.QuoteIdentifier(renameRequest.CurrentName), onCluster), ); err != nil { return nil, fmt.Errorf("unable to drop exchanged table %s: %w", renameRequest.CurrentName, err) } - } else if ex, ok := err.(*clickhouse.Exception); !ok || ex.Code != 48 { - // code 48 == not implemented -> move on to the fallback code, in all other error codes / types - // return, since we know/assume that the exchange would be the sensible action + } else if ex, ok := err.(*clickhouse.Exception); !ok || chproto.Error(ex.Code) != chproto.ErrNotImplemented { + // move on to the fallback code if unimplemented, in all other error codes / types return, + // since we know/assume exchange would be the sensible action return nil, fmt.Errorf("unable to exchange tables %s and %s: %w", renameRequest.NewName, renameRequest.CurrentName, err) } } @@ -221,14 +276,14 @@ func (c *ClickHouseConnector) RenameTables( // or err is set (in which case err comes from EXCHANGE TABLES) if !originalTableExists || err != nil { if err := c.execWithLogging(ctx, - fmt.Sprintf(dropTableIfExistsSQL, peerdb_clickhouse.QuoteIdentifier(renameRequest.NewName)), + fmt.Sprintf(dropTableIfExistsSQL, peerdb_clickhouse.QuoteIdentifier(renameRequest.NewName), onCluster), ); err != nil { return nil, fmt.Errorf("unable to drop table %s: %w", renameRequest.NewName, err) } - if err := c.execWithLogging(ctx, fmt.Sprintf("RENAME TABLE %s TO %s", + if err := c.execWithLogging(ctx, fmt.Sprintf("RENAME TABLE %s TO %s%s", peerdb_clickhouse.QuoteIdentifier(renameRequest.CurrentName), - peerdb_clickhouse.QuoteIdentifier(renameRequest.NewName), + peerdb_clickhouse.QuoteIdentifier(renameRequest.NewName), onCluster, )); err != nil { return nil, fmt.Errorf("unable to rename table %s to %s: %w", renameRequest.CurrentName, renameRequest.NewName, err) } @@ -246,10 +301,20 @@ func (c *ClickHouseConnector) RenameTables( func (c *ClickHouseConnector) SyncFlowCleanup(ctx context.Context, jobName string) error { // delete raw table if exists rawTableIdentifier := c.GetRawTableName(jobName) - if err := c.execWithLogging(ctx, fmt.Sprintf(dropTableIfExistsSQL, peerdb_clickhouse.QuoteIdentifier(rawTableIdentifier))); err != nil { + onCluster := c.onCluster() + if err := c.execWithLogging(ctx, + fmt.Sprintf(dropTableIfExistsSQL, peerdb_clickhouse.QuoteIdentifier(rawTableIdentifier), onCluster), + ); err != nil { return fmt.Errorf("[clickhouse] unable to drop raw table: %w", err) } - c.logger.Info("successfully dropped raw table " + rawTableIdentifier) + if onCluster != "" { + if err := c.execWithLogging(ctx, + fmt.Sprintf(dropTableIfExistsSQL, peerdb_clickhouse.QuoteIdentifier(rawTableIdentifier+"_shard"), onCluster), + ); err != nil { + return fmt.Errorf("[clickhouse] unable to drop raw table: %w", err) + } + } + c.logger.Info("successfully dropped raw table", slog.String("table", rawTableIdentifier)) return nil } @@ -258,11 +323,17 @@ func (c *ClickHouseConnector) RemoveTableEntriesFromRawTable( ctx context.Context, req *protos.RemoveTablesFromRawTableInput, ) error { + if c.config.Cluster != "" { + // this operation isn't crucial, okay to skip + c.logger.Info("skipping raw table cleanup of tables, DELETE not supported on Distributed table engine", + slog.Any("tables", req.DestinationTableNames)) + return nil + } + for _, tableName := range req.DestinationTableNames { - // Better to use lightweight deletes here as the main goal is to - // not have the rows in the table be visible by the NormalizeRecords' - // INSERT INTO SELECT queries - if err := c.execWithLogging(ctx, fmt.Sprintf("DELETE FROM `%s` WHERE _peerdb_destination_table_name = %s"+ + // Better to use lightweight deletes here as the main goal is to not have + // rows in the table be visible by the NormalizeRecords' INSERT INTO SELECT queries + if err := c.execWithLogging(ctx, fmt.Sprintf("DELETE FROM %s WHERE _peerdb_destination_table_name = %s"+ " AND _peerdb_batch_id > %d AND _peerdb_batch_id <= %d", c.GetRawTableName(req.FlowJobName), peerdb_clickhouse.QuoteLiteral(tableName), req.NormalizeBatchId, req.SyncBatchId), ); err != nil { diff --git a/flow/connectors/clickhouse/clickhouse.go b/flow/connectors/clickhouse/clickhouse.go index bf4346276..c0751bb09 100644 --- a/flow/connectors/clickhouse/clickhouse.go +++ b/flow/connectors/clickhouse/clickhouse.go @@ -23,7 +23,7 @@ import ( "github.com/PeerDB-io/peerdb/flow/generated/protos" "github.com/PeerDB-io/peerdb/flow/internal" "github.com/PeerDB-io/peerdb/flow/shared" - chvalidate "github.com/PeerDB-io/peerdb/flow/shared/clickhouse" + peerdb_clickhouse "github.com/PeerDB-io/peerdb/flow/shared/clickhouse" "github.com/PeerDB-io/peerdb/flow/shared/types" ) @@ -253,6 +253,9 @@ func Connect(ctx context.Context, env map[string]string, config *protos.Clickhou } else if maxInsertThreads != 0 { settings["max_insert_threads"] = maxInsertThreads } + if config.Cluster != "" { + settings["insert_distributed_sync"] = uint64(1) + } conn, err := clickhouse.Open(&clickhouse.Options{ Addr: []string{shared.JoinHostPort(config.Host, config.Port)}, @@ -288,19 +291,19 @@ func Connect(ctx context.Context, env map[string]string, config *protos.Clickhou } func (c *ClickHouseConnector) exec(ctx context.Context, query string) error { - return chvalidate.Exec(ctx, c.logger, c.database, query) + return peerdb_clickhouse.Exec(ctx, c.logger, c.database, query) } func (c *ClickHouseConnector) execWithConnection(ctx context.Context, conn clickhouse.Conn, query string) error { - return chvalidate.Exec(ctx, c.logger, conn, query) + return peerdb_clickhouse.Exec(ctx, c.logger, conn, query) } func (c *ClickHouseConnector) query(ctx context.Context, query string) (driver.Rows, error) { - return chvalidate.Query(ctx, c.logger, c.database, query) + return peerdb_clickhouse.Query(ctx, c.logger, c.database, query) } func (c *ClickHouseConnector) queryRow(ctx context.Context, query string) driver.Row { - return chvalidate.QueryRow(ctx, c.logger, c.database, query) + return peerdb_clickhouse.QueryRow(ctx, c.logger, c.database, query) } func (c *ClickHouseConnector) Close() error { @@ -323,7 +326,7 @@ func (c *ClickHouseConnector) execWithLogging(ctx context.Context, query string) } func (c *ClickHouseConnector) processTableComparison(dstTableName string, srcSchema *protos.TableSchema, - dstSchema []chvalidate.ClickHouseColumn, peerDBColumns []string, tableMapping *protos.TableMapping, + dstSchema []peerdb_clickhouse.ClickHouseColumn, peerDBColumns []string, tableMapping *protos.TableMapping, ) error { for _, srcField := range srcSchema.Columns { colName := srcField.Name @@ -483,3 +486,10 @@ func (c *ClickHouseConnector) GetTableSchema( return res, nil } + +func (c *ClickHouseConnector) onCluster() string { + if c.config.Cluster != "" { + return " ON CLUSTER " + peerdb_clickhouse.QuoteIdentifier(c.config.Cluster) + } + return "" +} diff --git a/flow/connectors/clickhouse/normalize.go b/flow/connectors/clickhouse/normalize.go index ab8ed4bea..8caa36650 100644 --- a/flow/connectors/clickhouse/normalize.go +++ b/flow/connectors/clickhouse/normalize.go @@ -58,7 +58,7 @@ func (c *ClickHouseConnector) SetupNormalizedTable( return true, nil } - normalizedTableCreateSQL, err := generateCreateTableSQLForNormalizedTable( + normalizedTableCreateSQL, err := c.generateCreateTableSQLForNormalizedTable( ctx, config, destinationTableIdentifier, @@ -69,129 +69,150 @@ func (c *ClickHouseConnector) SetupNormalizedTable( return false, fmt.Errorf("error while generating create table sql for destination ClickHouse table: %w", err) } - if err := c.execWithLogging(ctx, normalizedTableCreateSQL); err != nil { - return false, fmt.Errorf("[ch] error while creating destination ClickHouse table: %w", err) + for _, sql := range normalizedTableCreateSQL { + if err := c.execWithLogging(ctx, sql); err != nil { + return false, fmt.Errorf("[ch] error while creating destination ClickHouse table: %w", err) + } } return false, nil } -func getColName(overrides map[string]string, name string) string { - if newName, ok := overrides[name]; ok { - return newName - } - return name -} - -func generateCreateTableSQLForNormalizedTable( +func (c *ClickHouseConnector) generateCreateTableSQLForNormalizedTable( ctx context.Context, config *protos.SetupNormalizedTableBatchInput, tableIdentifier string, tableSchema *protos.TableSchema, chVersion *chproto.Version, -) (string, error) { +) ([]string, error) { + var engine string + tmEngine := protos.TableEngine_CH_ENGINE_REPLACING_MERGE_TREE + var tableMapping *protos.TableMapping for _, tm := range config.TableMappings { if tm.DestinationTableIdentifier == tableIdentifier { + tmEngine = tm.Engine tableMapping = tm break } } - var stmtBuilder strings.Builder - stmtBuilder.WriteString("CREATE ") - if config.IsResync { - stmtBuilder.WriteString("OR REPLACE ") + switch tmEngine { + case protos.TableEngine_CH_ENGINE_REPLACING_MERGE_TREE, protos.TableEngine_CH_ENGINE_REPLICATED_REPLACING_MERGE_TREE: + if c.config.Replicated { + engine = fmt.Sprintf( + "ReplicatedReplacingMergeTree(%s%s','{replica}',%s)", + zooPathPrefix, + peerdb_clickhouse.EscapeStr(tableIdentifier), + peerdb_clickhouse.QuoteIdentifier(versionColName), + ) + } else { + engine = fmt.Sprintf("ReplacingMergeTree(%s)", peerdb_clickhouse.QuoteIdentifier(versionColName)) + } + case protos.TableEngine_CH_ENGINE_MERGE_TREE, protos.TableEngine_CH_ENGINE_REPLICATED_MERGE_TREE: + if c.config.Replicated { + engine = fmt.Sprintf( + "ReplicatedMergeTree('%s%s','{replica}')", + zooPathPrefix, + peerdb_clickhouse.EscapeStr(tableIdentifier), + ) + } else { + engine = "MergeTree()" + } + case protos.TableEngine_CH_ENGINE_NULL: + engine = "Null" } - stmtBuilder.WriteString("TABLE ") - if !config.IsResync { - stmtBuilder.WriteString("IF NOT EXISTS ") + + sourceSchemaAsDestinationColumn, err := internal.PeerDBSourceSchemaAsDestinationColumn(ctx, config.Env) + if err != nil { + return nil, err + } + + var stmtBuilder strings.Builder + var stmtBuilderDistributed strings.Builder + var builders []*strings.Builder + if c.config.Cluster != "" && tmEngine != protos.TableEngine_CH_ENGINE_NULL { + builders = []*strings.Builder{&stmtBuilder, &stmtBuilderDistributed} + } else { + builders = []*strings.Builder{&stmtBuilder} } - fmt.Fprintf(&stmtBuilder, "%s (", peerdb_clickhouse.QuoteIdentifier(tableIdentifier)) colNameMap := make(map[string]string) - for _, column := range tableSchema.Columns { - colName := column.Name - dstColName := colName - colType := types.QValueKind(column.Type) - var columnNullableEnabled bool - var clickHouseType string - if tableMapping != nil { - for _, col := range tableMapping.Columns { - if col.SourceName == colName { - if col.DestinationName != "" { - dstColName = col.DestinationName - colNameMap[colName] = dstColName - } - if col.DestinationType != "" { - clickHouseType = col.DestinationType + for idx, builder := range builders { + builder.WriteString("CREATE ") + if config.IsResync { + builder.WriteString("OR REPLACE ") + } + builder.WriteString("TABLE ") + if !config.IsResync { + builder.WriteString("IF NOT EXISTS ") + } + if c.config.Cluster != "" && tmEngine != protos.TableEngine_CH_ENGINE_NULL && idx == 0 { + // distributed table gets destination name, avoid naming conflict + builder.WriteString(peerdb_clickhouse.QuoteIdentifier(tableIdentifier + "_shard")) + } else { + builder.WriteString(peerdb_clickhouse.QuoteIdentifier(tableIdentifier)) + } + if c.config.Cluster != "" { + fmt.Fprintf(builder, " ON CLUSTER %s", peerdb_clickhouse.QuoteIdentifier(c.config.Cluster)) + } + builder.WriteString(" (") + + for _, column := range tableSchema.Columns { + colName := column.Name + dstColName := colName + colType := types.QValueKind(column.Type) + var columnNullableEnabled bool + var clickHouseType string + if tableMapping != nil { + for _, col := range tableMapping.Columns { + if col.SourceName == colName { + if col.DestinationName != "" { + dstColName = col.DestinationName + colNameMap[colName] = dstColName + } + if col.DestinationType != "" { + clickHouseType = col.DestinationType + } + columnNullableEnabled = col.NullableEnabled + break } - columnNullableEnabled = col.NullableEnabled - break } } - } - if clickHouseType == "" { - var err error - clickHouseType, err = qvalue.ToDWHColumnType( - ctx, colType, config.Env, protos.DBType_CLICKHOUSE, chVersion, column, tableSchema.NullableEnabled || columnNullableEnabled, - ) - if err != nil { - return "", fmt.Errorf("error while converting column type to ClickHouse type: %w", err) + if clickHouseType == "" { + var err error + clickHouseType, err = qvalue.ToDWHColumnType( + ctx, colType, config.Env, protos.DBType_CLICKHOUSE, chVersion, column, tableSchema.NullableEnabled || columnNullableEnabled, + ) + if err != nil { + return nil, fmt.Errorf("error while converting column type to ClickHouse type: %w", err) + } + } else if (tableSchema.NullableEnabled || columnNullableEnabled) && column.Nullable && !colType.IsArray() { + clickHouseType = fmt.Sprintf("Nullable(%s)", clickHouseType) } - } else if (tableSchema.NullableEnabled || columnNullableEnabled) && column.Nullable && !colType.IsArray() { - clickHouseType = fmt.Sprintf("Nullable(%s)", clickHouseType) - } - fmt.Fprintf(&stmtBuilder, "%s %s, ", peerdb_clickhouse.QuoteIdentifier(dstColName), clickHouseType) - } - // TODO support soft delete - // synced at column will be added to all normalized tables - if config.SyncedAtColName != "" { - colName := strings.ToLower(config.SyncedAtColName) - fmt.Fprintf(&stmtBuilder, "%s DateTime64(9) DEFAULT now64(), ", peerdb_clickhouse.QuoteIdentifier(colName)) - } + fmt.Fprintf(builder, "%s %s, ", peerdb_clickhouse.QuoteIdentifier(dstColName), clickHouseType) + } + // TODO support hard delete + // synced at column will be added to all normalized tables + if config.SyncedAtColName != "" { + colName := strings.ToLower(config.SyncedAtColName) + fmt.Fprintf(builder, "%s DateTime64(9) DEFAULT now64(), ", peerdb_clickhouse.QuoteIdentifier(colName)) + } - // add _peerdb_source_schema_name column - sourceSchemaAsDestinationColumn, err := internal.PeerDBSourceSchemaAsDestinationColumn(ctx, config.Env) - if err != nil { - return "", err - } - if sourceSchemaAsDestinationColumn { - fmt.Fprintf(&stmtBuilder, "%s %s, ", peerdb_clickhouse.QuoteIdentifier(sourceSchemaColName), sourceSchemaColType) - } + // add _peerdb_source_schema_name column + if sourceSchemaAsDestinationColumn { + fmt.Fprintf(builder, "%s %s, ", peerdb_clickhouse.QuoteIdentifier(sourceSchemaColName), sourceSchemaColType) + } - var engine string - tmEngine := protos.TableEngine_CH_ENGINE_REPLACING_MERGE_TREE - if tableMapping != nil { - tmEngine = tableMapping.Engine - } - switch tmEngine { - case protos.TableEngine_CH_ENGINE_REPLACING_MERGE_TREE: - engine = fmt.Sprintf("ReplacingMergeTree(%s)", peerdb_clickhouse.QuoteIdentifier(versionColName)) - case protos.TableEngine_CH_ENGINE_MERGE_TREE: - engine = "MergeTree()" - case protos.TableEngine_CH_ENGINE_REPLICATED_REPLACING_MERGE_TREE: - engine = fmt.Sprintf( - "ReplicatedReplacingMergeTree('/clickhouse/tables/{shard}/{database}/%s','{replica}',%s)", - peerdb_clickhouse.EscapeStr(tableIdentifier), - peerdb_clickhouse.QuoteIdentifier(versionColName), - ) - case protos.TableEngine_CH_ENGINE_REPLICATED_MERGE_TREE: - engine = fmt.Sprintf( - "ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/%s','{replica}')", - peerdb_clickhouse.EscapeStr(tableIdentifier), - ) - case protos.TableEngine_CH_ENGINE_NULL: - engine = "Null" + // add sign and version columns + fmt.Fprintf(builder, "%s %s, %s %s)", + peerdb_clickhouse.QuoteIdentifier(signColName), signColType, peerdb_clickhouse.QuoteIdentifier(versionColName), versionColType) } - // add sign and version columns - fmt.Fprintf(&stmtBuilder, "%s %s, %s %s) ENGINE = %s", - peerdb_clickhouse.QuoteIdentifier(signColName), signColType, peerdb_clickhouse.QuoteIdentifier(versionColName), versionColType, engine) + fmt.Fprintf(&stmtBuilder, " ENGINE = %s", engine) orderByColumns := getOrderedOrderByColumns(tableMapping, tableSchema.PrimaryKeyColumns, colNameMap) - if sourceSchemaAsDestinationColumn { orderByColumns = append([]string{sourceSchemaColName}, orderByColumns...) } @@ -206,13 +227,34 @@ func generateCreateTableSQLForNormalizedTable( } if nullable, err := internal.PeerDBNullable(ctx, config.Env); err != nil { - return "", err + return nil, err } else if nullable { stmtBuilder.WriteString(" SETTINGS allow_nullable_key = 1") } + + if c.config.Cluster != "" { + fmt.Fprintf(&stmtBuilderDistributed, " ENGINE = Distributed(%s,%s,%s", + peerdb_clickhouse.QuoteIdentifier(c.config.Cluster), + peerdb_clickhouse.QuoteIdentifier(c.config.Database), + peerdb_clickhouse.QuoteIdentifier(tableIdentifier+"_shard"), + ) + if tableMapping.ShardingKey != "" { + stmtBuilderDistributed.WriteByte(',') + stmtBuilderDistributed.WriteString(tableMapping.ShardingKey) + if tableMapping.PolicyName != "" { + stmtBuilderDistributed.WriteByte(',') + stmtBuilderDistributed.WriteString(peerdb_clickhouse.QuoteLiteral(tableMapping.PolicyName)) + } + } + stmtBuilderDistributed.WriteByte(')') + } } - return stmtBuilder.String(), nil + result := make([]string, len(builders)) + for idx, builder := range builders { + result[idx] = builder.String() + } + return result, nil } // Returns a list of order by columns ordered by their ordering, and puts the pkeys at the end. @@ -411,6 +453,7 @@ func (c *ClickHouseConnector) NormalizeRecords( req.Env, rawTbl, c.chVersion, + c.config.Cluster != "", ) insertIntoSelectQuery, err := queryGenerator.BuildQuery(ctx) if err != nil { @@ -542,3 +585,10 @@ func (c *ClickHouseConnector) copyAvroStagesToDestination( } return nil } + +func getColName(overrides map[string]string, name string) string { + if newName, ok := overrides[name]; ok { + return newName + } + return name +} diff --git a/flow/connectors/clickhouse/normalize_query.go b/flow/connectors/clickhouse/normalize_query.go index ef81014ce..5f5dc1b2c 100644 --- a/flow/connectors/clickhouse/normalize_query.go +++ b/flow/connectors/clickhouse/normalize_query.go @@ -28,6 +28,7 @@ type NormalizeQueryGenerator struct { syncBatchID int64 enablePrimaryUpdate bool sourceSchemaAsDestinationColumn bool + cluster bool } // NewTableNormalizeQuery constructs a TableNormalizeQuery with required fields. @@ -44,6 +45,7 @@ func NewNormalizeQueryGenerator( env map[string]string, rawTableName string, chVersion *chproto.Version, + cluster bool, ) *NormalizeQueryGenerator { return &NormalizeQueryGenerator{ TableName: tableName, @@ -58,6 +60,7 @@ func NewNormalizeQueryGenerator( env: env, rawTableName: rawTableName, chVersion: chVersion, + cluster: cluster, } } @@ -295,6 +298,10 @@ func (t *NormalizeQueryGenerator) BuildQuery(ctx context.Context) (string, error } } + if t.cluster { + colSelector.WriteString(" SETTINGS parallel_distributed_insert_select=0") + } + insertIntoSelectQuery := fmt.Sprintf("INSERT INTO %s %s %s", peerdb_clickhouse.QuoteIdentifier(t.TableName), colSelector.String(), selectQuery.String()) diff --git a/flow/connectors/clickhouse/normalize_test.go b/flow/connectors/clickhouse/normalize_test.go index 862ffc1e3..48b5b21ee 100644 --- a/flow/connectors/clickhouse/normalize_test.go +++ b/flow/connectors/clickhouse/normalize_test.go @@ -190,6 +190,7 @@ func TestBuildQuery_Basic(t *testing.T) { env, rawTableName, nil, + false, ) query, err := g.BuildQuery(ctx) @@ -245,6 +246,7 @@ func TestBuildQuery_WithPrimaryUpdate(t *testing.T) { env, rawTableName, nil, + false, ) query, err := g.BuildQuery(ctx) @@ -297,11 +299,13 @@ func TestBuildQuery_WithSourceSchemaAsDestinationColumn(t *testing.T) { env, rawTableName, nil, + true, ) query, err := g.BuildQuery(ctx) require.NoError(t, err) require.Contains(t, query, " AS `_peerdb_source_schema`") + require.Contains(t, query, "parallel_distributed_insert_select=0") } func TestBuildQuery_WithNumParts(t *testing.T) { @@ -346,6 +350,7 @@ func TestBuildQuery_WithNumParts(t *testing.T) { env, rawTableName, nil, + false, ) query, err := g.BuildQuery(ctx) diff --git a/flow/connectors/core.go b/flow/connectors/core.go index 2eab92b03..deba52ef6 100644 --- a/flow/connectors/core.go +++ b/flow/connectors/core.go @@ -196,11 +196,10 @@ type CDCSyncConnectorCore interface { SyncFlowCleanup(ctx context.Context, jobName string) error // ReplayTableSchemaDelta changes a destination table to match the schema at source - // This could involve adding or dropping multiple columns. + // This could involve adding multiple columns. // Connectors which are non-normalizing should implement this as a nop. - ReplayTableSchemaDeltas( - ctx context.Context, env map[string]string, flowJobName string, schemaDeltas []*protos.TableSchemaDelta, - ) error + ReplayTableSchemaDeltas(ctx context.Context, env map[string]string, flowJobName string, + tableMappings []*protos.TableMapping, schemaDeltas []*protos.TableSchemaDelta) error } type CDCSyncConnector interface { diff --git a/flow/connectors/elasticsearch/elasticsearch.go b/flow/connectors/elasticsearch/elasticsearch.go index b951311b8..e124879a1 100644 --- a/flow/connectors/elasticsearch/elasticsearch.go +++ b/flow/connectors/elasticsearch/elasticsearch.go @@ -94,7 +94,7 @@ func (esc *ElasticsearchConnector) CreateRawTable(ctx context.Context, // we handle schema changes by not handling them since no mapping is being enforced right now func (esc *ElasticsearchConnector) ReplayTableSchemaDeltas(ctx context.Context, env map[string]string, - flowJobName string, schemaDeltas []*protos.TableSchemaDelta, + flowJobName string, _ []*protos.TableMapping, schemaDeltas []*protos.TableSchemaDelta, ) error { return nil } diff --git a/flow/connectors/eventhub/eventhub.go b/flow/connectors/eventhub/eventhub.go index ed15922fc..426338e57 100644 --- a/flow/connectors/eventhub/eventhub.go +++ b/flow/connectors/eventhub/eventhub.go @@ -371,8 +371,7 @@ func (c *EventHubConnector) CreateRawTable(ctx context.Context, req *protos.Crea } func (c *EventHubConnector) ReplayTableSchemaDeltas(_ context.Context, _ map[string]string, - flowJobName string, schemaDeltas []*protos.TableSchemaDelta, + flowJobName string, _ []*protos.TableMapping, schemaDeltas []*protos.TableSchemaDelta, ) error { - c.logger.Info("ReplayTableSchemaDeltas for event hub is a no-op") return nil } diff --git a/flow/connectors/kafka/kafka.go b/flow/connectors/kafka/kafka.go index fa74e02a2..b707c2bb0 100644 --- a/flow/connectors/kafka/kafka.go +++ b/flow/connectors/kafka/kafka.go @@ -142,7 +142,7 @@ func (c *KafkaConnector) CreateRawTable(ctx context.Context, req *protos.CreateR } func (c *KafkaConnector) ReplayTableSchemaDeltas(_ context.Context, _ map[string]string, - flowJobName string, schemaDeltas []*protos.TableSchemaDelta, + flowJobName string, _ []*protos.TableMapping, schemaDeltas []*protos.TableSchemaDelta, ) error { return nil } diff --git a/flow/connectors/postgres/postgres.go b/flow/connectors/postgres/postgres.go index d3b905162..0037f24fd 100644 --- a/flow/connectors/postgres/postgres.go +++ b/flow/connectors/postgres/postgres.go @@ -624,7 +624,7 @@ func syncRecordsCore[Items model.Items]( return nil, err } - if err := c.ReplayTableSchemaDeltas(ctx, req.Env, req.FlowJobName, req.Records.SchemaDeltas); err != nil { + if err := c.ReplayTableSchemaDeltas(ctx, req.Env, req.FlowJobName, req.TableMappings, req.Records.SchemaDeltas); err != nil { return nil, fmt.Errorf("failed to sync schema changes: %w", err) } @@ -1020,6 +1020,7 @@ func (c *PostgresConnector) ReplayTableSchemaDeltas( ctx context.Context, _ map[string]string, flowJobName string, + _ []*protos.TableMapping, schemaDeltas []*protos.TableSchemaDelta, ) error { if len(schemaDeltas) == 0 { diff --git a/flow/connectors/postgres/postgres_schema_delta_test.go b/flow/connectors/postgres/postgres_schema_delta_test.go index 1a5072e1b..4fc4333a1 100644 --- a/flow/connectors/postgres/postgres_schema_delta_test.go +++ b/flow/connectors/postgres/postgres_schema_delta_test.go @@ -37,8 +37,7 @@ func SetupSuite(t *testing.T) PostgresSchemaDeltaTestSuite { } }() schema := "pgdelta_" + strings.ToLower(shared.RandomString(8)) - _, err = setupTx.Exec(t.Context(), fmt.Sprintf("DROP SCHEMA IF EXISTS %s CASCADE", - schema)) + _, err = setupTx.Exec(t.Context(), fmt.Sprintf("DROP SCHEMA IF EXISTS %s CASCADE", schema)) require.NoError(t, err) _, err = setupTx.Exec(t.Context(), "CREATE SCHEMA "+schema) require.NoError(t, err) @@ -57,7 +56,7 @@ func (s PostgresSchemaDeltaTestSuite) TestSimpleAddColumn() { fmt.Sprintf("CREATE TABLE %s(id INT PRIMARY KEY)", tableName)) require.NoError(s.t, err) - err = s.connector.ReplayTableSchemaDeltas(s.t.Context(), nil, "schema_delta_flow", []*protos.TableSchemaDelta{{ + require.NoError(s.t, s.connector.ReplayTableSchemaDeltas(s.t.Context(), nil, "schema_delta_flow", nil, []*protos.TableSchemaDelta{{ SrcTableName: tableName, DstTableName: tableName, AddedColumns: []*protos.FieldDescription{ @@ -68,8 +67,7 @@ func (s PostgresSchemaDeltaTestSuite) TestSimpleAddColumn() { Nullable: true, }, }, - }}) - require.NoError(s.t, err) + }})) output, err := s.connector.GetTableSchema(s.t.Context(), nil, shared.InternalVersion_Latest, protos.TypeSystem_Q, []*protos.TableMapping{{SourceTableIdentifier: tableName}}) @@ -113,12 +111,11 @@ func (s PostgresSchemaDeltaTestSuite) TestAddAllColumnTypes() { } } - err = s.connector.ReplayTableSchemaDeltas(s.t.Context(), nil, "schema_delta_flow", []*protos.TableSchemaDelta{{ + require.NoError(s.t, s.connector.ReplayTableSchemaDeltas(s.t.Context(), nil, "schema_delta_flow", nil, []*protos.TableSchemaDelta{{ SrcTableName: tableName, DstTableName: tableName, AddedColumns: addedColumns, - }}) - require.NoError(s.t, err) + }})) output, err := s.connector.GetTableSchema(s.t.Context(), nil, shared.InternalVersion_Latest, protos.TypeSystem_Q, []*protos.TableMapping{{SourceTableIdentifier: tableName}}) @@ -145,12 +142,11 @@ func (s PostgresSchemaDeltaTestSuite) TestAddTrickyColumnNames() { } } - err = s.connector.ReplayTableSchemaDeltas(s.t.Context(), nil, "schema_delta_flow", []*protos.TableSchemaDelta{{ + require.NoError(s.t, s.connector.ReplayTableSchemaDeltas(s.t.Context(), nil, "schema_delta_flow", nil, []*protos.TableSchemaDelta{{ SrcTableName: tableName, DstTableName: tableName, AddedColumns: addedColumns, - }}) - require.NoError(s.t, err) + }})) output, err := s.connector.GetTableSchema(s.t.Context(), nil, shared.InternalVersion_Latest, protos.TypeSystem_Q, []*protos.TableMapping{{SourceTableIdentifier: tableName}}) @@ -177,12 +173,11 @@ func (s PostgresSchemaDeltaTestSuite) TestAddDropWhitespaceColumnNames() { } } - err = s.connector.ReplayTableSchemaDeltas(s.t.Context(), nil, "schema_delta_flow", []*protos.TableSchemaDelta{{ + require.NoError(s.t, s.connector.ReplayTableSchemaDeltas(s.t.Context(), nil, "schema_delta_flow", nil, []*protos.TableSchemaDelta{{ SrcTableName: tableName, DstTableName: tableName, AddedColumns: addedColumns, - }}) - require.NoError(s.t, err) + }})) output, err := s.connector.GetTableSchema(s.t.Context(), nil, shared.InternalVersion_Latest, protos.TypeSystem_Q, []*protos.TableMapping{{SourceTableIdentifier: tableName}}) @@ -203,11 +198,9 @@ func (s PostgresSchemaDeltaTestSuite) Teardown(ctx context.Context) { require.NoError(s.t, err) } }() - _, err = teardownTx.Exec(ctx, fmt.Sprintf("DROP SCHEMA IF EXISTS %s CASCADE", - s.schema)) - require.NoError(s.t, err) - err = teardownTx.Commit(ctx) + _, err = teardownTx.Exec(ctx, fmt.Sprintf("DROP SCHEMA IF EXISTS %s CASCADE", s.schema)) require.NoError(s.t, err) + require.NoError(s.t, teardownTx.Commit(ctx)) require.NoError(s.t, s.connector.ConnectionActive(ctx)) require.NoError(s.t, s.connector.Close()) diff --git a/flow/connectors/pubsub/pubsub.go b/flow/connectors/pubsub/pubsub.go index e259eff80..d84b2f5cc 100644 --- a/flow/connectors/pubsub/pubsub.go +++ b/flow/connectors/pubsub/pubsub.go @@ -68,7 +68,7 @@ func (c *PubSubConnector) CreateRawTable(ctx context.Context, req *protos.Create } func (c *PubSubConnector) ReplayTableSchemaDeltas(_ context.Context, _ map[string]string, - flowJobName string, schemaDeltas []*protos.TableSchemaDelta, + flowJobName string, _ []*protos.TableMapping, schemaDeltas []*protos.TableSchemaDelta, ) error { return nil } diff --git a/flow/connectors/s3/s3.go b/flow/connectors/s3/s3.go index d009b4984..858b9233b 100644 --- a/flow/connectors/s3/s3.go +++ b/flow/connectors/s3/s3.go @@ -116,8 +116,7 @@ func (c *S3Connector) SyncRecords(ctx context.Context, req *model.SyncRecordsReq } func (c *S3Connector) ReplayTableSchemaDeltas(_ context.Context, _ map[string]string, - flowJobName string, schemaDeltas []*protos.TableSchemaDelta, + flowJobName string, _ []*protos.TableMapping, schemaDeltas []*protos.TableSchemaDelta, ) error { - c.logger.Info("ReplayTableSchemaDeltas for S3 is a no-op") return nil } diff --git a/flow/connectors/snowflake/snowflake.go b/flow/connectors/snowflake/snowflake.go index 1aa3607ac..4c461be0d 100644 --- a/flow/connectors/snowflake/snowflake.go +++ b/flow/connectors/snowflake/snowflake.go @@ -336,6 +336,7 @@ func (c *SnowflakeConnector) ReplayTableSchemaDeltas( ctx context.Context, env map[string]string, flowJobName string, + _ []*protos.TableMapping, schemaDeltas []*protos.TableSchemaDelta, ) error { if len(schemaDeltas) == 0 { @@ -447,7 +448,7 @@ func (c *SnowflakeConnector) syncRecordsViaAvro( return nil, err } - if err := c.ReplayTableSchemaDeltas(ctx, req.Env, req.FlowJobName, req.Records.SchemaDeltas); err != nil { + if err := c.ReplayTableSchemaDeltas(ctx, req.Env, req.FlowJobName, req.TableMappings, req.Records.SchemaDeltas); err != nil { return nil, fmt.Errorf("failed to sync schema changes: %w", err) } diff --git a/flow/connectors/utils/avro_writer.go b/flow/connectors/utils/avro_writer.go index 9da161776..d8b61144a 100644 --- a/flow/connectors/utils/avro_writer.go +++ b/flow/connectors/utils/avro_writer.go @@ -138,6 +138,9 @@ func (p *peerDBOCFWriter) WriteRecordsToS3( uploader := manager.NewUploader(s3svc, func(u *manager.Uploader) { if partSize > 0 { u.PartSize = partSize + if partSize > 268435455 { // 256MiB + u.Concurrency = 1 + } } }) diff --git a/flow/e2e/api/api_test.go b/flow/e2e/api/api_test.go index 13abff5c6..7bd56d9e0 100644 --- a/flow/e2e/api/api_test.go +++ b/flow/e2e/api/api_test.go @@ -39,9 +39,9 @@ type Suite struct { protos.FlowServiceClient t *testing.T pg *e2e.PostgresSource - ch e2e_clickhouse.ClickHouseSuite source e2e.SuiteSource suffix string + ch e2e_clickhouse.ClickHouseSuite } func (s Suite) Teardown(ctx context.Context) { @@ -175,7 +175,7 @@ func testApi[TSource e2e.SuiteSource]( t: t, pg: pg, source: source, - ch: e2e_clickhouse.SetupSuite(t, func(*testing.T) (TSource, string, error) { + ch: e2e_clickhouse.SetupSuite(t, false, func(*testing.T) (TSource, string, error) { return source, suffix, nil })(t), suffix: suffix, diff --git a/flow/e2e/clickhouse/clickhouse.go b/flow/e2e/clickhouse/clickhouse.go index e8395e842..833d510cc 100644 --- a/flow/e2e/clickhouse/clickhouse.go +++ b/flow/e2e/clickhouse/clickhouse.go @@ -33,6 +33,7 @@ type ClickHouseSuite struct { s3Helper *e2e_s3.S3TestHelper connector *connclickhouse.ClickHouseConnector suffix string + cluster bool } func (s ClickHouseSuite) T() *testing.T { @@ -64,7 +65,28 @@ func (s ClickHouseSuite) Suffix() string { } func (s ClickHouseSuite) Peer() *protos.Peer { - return s.PeerForDatabase("e2e_test_" + s.suffix) + dbname := "e2e_test_" + s.suffix + if s.cluster { + ret := &protos.Peer{ + Name: e2e.AddSuffix(s, dbname), + Type: protos.DBType_CLICKHOUSE, + Config: &protos.Peer_ClickhouseConfig{ + ClickhouseConfig: &protos.ClickhouseConfig{ + Host: "localhost", + Port: 9001, + Database: dbname, + DisableTls: true, + S3: s.s3Helper.S3Config, + Cluster: "cicluster", + Replicated: false, + }, + }, + } + e2e.CreatePeer(s.t, ret) + return ret + } else { + return s.PeerForDatabase(dbname) + } } func (s ClickHouseSuite) PeerForDatabase(dbname string) *protos.Peer { @@ -117,12 +139,18 @@ func (s ClickHouseSuite) CreateRMTTable(tableName string, columns []TestClickHou columnStr := strings.Join(columnStrings, ", ") // Create the table with ReplacingMergeTree engine - createTableQuery := fmt.Sprintf("CREATE TABLE `%s` (%s) ENGINE = ReplacingMergeTree() ORDER BY `%s`", tableName, columnStr, orderingKey) + onCluster := "" + if s.cluster { + onCluster = " ON CLUSTER cicluster" + } + createTableQuery := fmt.Sprintf("CREATE TABLE `%s`%s (%s) ENGINE = ReplacingMergeTree() ORDER BY `%s`", + tableName, onCluster, columnStr, orderingKey) return ch.Exec(s.t.Context(), createTableQuery) } func (s ClickHouseSuite) GetRows(table string, cols string) (*model.QRecordBatch, error) { - ch, err := connclickhouse.Connect(s.t.Context(), nil, s.Peer().GetClickhouseConfig()) + peer := s.Peer() + ch, err := connclickhouse.Connect(s.t.Context(), nil, peer.GetClickhouseConfig()) if err != nil { return nil, err } @@ -339,6 +367,7 @@ func (s ClickHouseSuite) queryRawTable(conn clickhouse.Conn, table string, cols func SetupSuite[TSource e2e.SuiteSource]( t *testing.T, + cluster bool, setupSource func(*testing.T) (TSource, string, error), ) func(*testing.T) ClickHouseSuite { t.Helper() @@ -356,11 +385,16 @@ func SetupSuite[TSource e2e.SuiteSource]( source: e2e.SuiteSource(source), suffix: suffix, s3Helper: s3Helper, + cluster: cluster, } ch, err := connclickhouse.Connect(t.Context(), nil, s.PeerForDatabase("default").GetClickhouseConfig()) require.NoError(t, err, "failed to connect to clickhouse") - err = ch.Exec(t.Context(), "CREATE DATABASE e2e_test_"+suffix) + if cluster { + err = ch.Exec(t.Context(), "CREATE DATABASE e2e_test_"+suffix+" ON CLUSTER cicluster") + } else { + err = ch.Exec(t.Context(), "CREATE DATABASE e2e_test_"+suffix) + } require.NoError(t, err, "failed to create clickhouse database") connector, err := connclickhouse.NewClickHouseConnector(t.Context(), nil, s.Peer().GetClickhouseConfig()) diff --git a/flow/e2e/clickhouse/peer_flow_ch_test.go b/flow/e2e/clickhouse/peer_flow_ch_test.go index 08b8b2ba5..bfabd966a 100644 --- a/flow/e2e/clickhouse/peer_flow_ch_test.go +++ b/flow/e2e/clickhouse/peer_flow_ch_test.go @@ -32,7 +32,7 @@ import ( var testData embed.FS func TestPeerFlowE2ETestSuitePG_CH(t *testing.T) { - e2eshared.RunSuite(t, SetupSuite(t, func(t *testing.T) (*e2e.PostgresSource, string, error) { + e2eshared.RunSuite(t, SetupSuite(t, false, func(t *testing.T) (*e2e.PostgresSource, string, error) { t.Helper() suffix := "pgch_" + strings.ToLower(shared.RandomString(8)) source, err := e2e.SetupPostgres(t, suffix) @@ -41,7 +41,7 @@ func TestPeerFlowE2ETestSuitePG_CH(t *testing.T) { } func TestPeerFlowE2ETestSuiteMySQL_CH(t *testing.T) { - e2eshared.RunSuite(t, SetupSuite(t, func(t *testing.T) (*e2e.MySqlSource, string, error) { + e2eshared.RunSuite(t, SetupSuite(t, false, func(t *testing.T) (*e2e.MySqlSource, string, error) { t.Helper() suffix := "mych_" + strings.ToLower(shared.RandomString(8)) source, err := e2e.SetupMySQL(t, suffix) @@ -49,6 +49,24 @@ func TestPeerFlowE2ETestSuiteMySQL_CH(t *testing.T) { })) } +func TestPeerFlowE2ETestSuitePG_CH_Cluster(t *testing.T) { + e2eshared.RunSuite(t, SetupSuite(t, true, func(t *testing.T) (*e2e.PostgresSource, string, error) { + t.Helper() + suffix := "pgchcl_" + strings.ToLower(shared.RandomString(8)) + source, err := e2e.SetupPostgres(t, suffix) + return source, suffix, err + })) +} + +func TestPeerFlowE2ETestSuiteMySQL_CH_Cluster(t *testing.T) { + e2eshared.RunSuite(t, SetupSuite(t, true, func(t *testing.T) (*e2e.MySqlSource, string, error) { + t.Helper() + suffix := "mychcl_" + strings.ToLower(shared.RandomString(8)) + source, err := e2e.SetupMySQL(t, suffix) + return source, suffix, err + })) +} + func (s ClickHouseSuite) attachSchemaSuffix(tableName string) string { return fmt.Sprintf("e2e_test_%s.%s", s.suffix, tableName) } @@ -101,14 +119,16 @@ func (s ClickHouseSuite) Test_Addition_Removal() { if pgconn, ok := s.source.Connector().(*connpostgres.PostgresConnector); ok { conn := pgconn.Conn() _, err := conn.Exec(s.t.Context(), - `SELECT pg_terminate_backend(pid) FROM pg_stat_activity - WHERE query LIKE '%START_REPLICATION%' AND query LIKE '%clickhousetableremoval%' AND backend_type='walsender'`) + fmt.Sprintf(`SELECT pg_terminate_backend(pid) FROM pg_stat_activity + WHERE query LIKE '%%START_REPLICATION%%' AND query LIKE '%%%s%%' AND backend_type='walsender'`, + s.attachSuffix("clickhousetableremoval"))) require.NoError(s.t, err) e2e.EnvWaitFor(s.t, env, 3*time.Minute, "waiting for replication to stop", func() bool { rows, err := conn.Query(s.t.Context(), - `SELECT pid FROM pg_stat_activity - WHERE query LIKE '%START_REPLICATION%' AND query LIKE '%clickhousetableremoval%' AND backend_type='walsender'`) + fmt.Sprintf(`SELECT pid FROM pg_stat_activity + WHERE query LIKE '%%START_REPLICATION%%' AND query LIKE '%%%s%%' AND backend_type='walsender'`, + s.attachSuffix("clickhousetableremoval"))) require.NoError(s.t, err) defer rows.Close() return !rows.Next() @@ -117,12 +137,11 @@ func (s ClickHouseSuite) Test_Addition_Removal() { runID := e2e.EnvGetRunID(s.t, env) e2e.SignalWorkflow(s.t.Context(), env, model.CDCDynamicPropertiesSignal, &protos.CDCFlowConfigUpdate{ - AdditionalTables: []*protos.TableMapping{ - { - SourceTableIdentifier: addedSrcTableName, - DestinationTableIdentifier: addedDstTableName, - }, - }, + AdditionalTables: []*protos.TableMapping{{ + SourceTableIdentifier: addedSrcTableName, + DestinationTableIdentifier: addedDstTableName, + ShardingKey: "id", + }}, }) e2e.EnvWaitFor(s.t, env, 4*time.Minute, "adding table", func() bool { @@ -141,14 +160,16 @@ func (s ClickHouseSuite) Test_Addition_Removal() { if pgconn, ok := s.source.Connector().(*connpostgres.PostgresConnector); ok { conn := pgconn.Conn() _, err := conn.Exec(s.t.Context(), - `SELECT pg_terminate_backend(pid) FROM pg_stat_activity - WHERE query LIKE '%START_REPLICATION%' AND query LIKE '%clickhousetableremoval%' AND backend_type='walsender'`) + fmt.Sprintf(`SELECT pg_terminate_backend(pid) FROM pg_stat_activity + WHERE query LIKE '%%START_REPLICATION%%' AND query LIKE '%%%s%%' AND backend_type='walsender'`, + s.attachSuffix("clickhousetableremoval"))) require.NoError(s.t, err) e2e.EnvWaitFor(s.t, env, 3*time.Minute, "waiting for replication to stop", func() bool { rows, err := conn.Query(s.t.Context(), - `SELECT pid FROM pg_stat_activity - WHERE query LIKE '%START_REPLICATION%' AND query LIKE '%clickhousetableremoval%' AND backend_type='walsender'`) + fmt.Sprintf(`SELECT pid FROM pg_stat_activity + WHERE query LIKE '%%START_REPLICATION%%' AND query LIKE '%%%s%%' AND backend_type='walsender'`, + s.attachSuffix("clickhousetableremoval"))) require.NoError(s.t, err) defer rows.Close() return !rows.Next() @@ -156,12 +177,10 @@ func (s ClickHouseSuite) Test_Addition_Removal() { } e2e.SignalWorkflow(s.t.Context(), env, model.CDCDynamicPropertiesSignal, &protos.CDCFlowConfigUpdate{ - RemovedTables: []*protos.TableMapping{ - { - SourceTableIdentifier: srcTableName, - DestinationTableIdentifier: dstTableName, - }, - }, + RemovedTables: []*protos.TableMapping{{ + SourceTableIdentifier: srcTableName, + DestinationTableIdentifier: dstTableName, + }}, }) e2e.EnvWaitFor(s.t, env, 4*time.Minute, "removing table", func() bool { @@ -419,13 +438,13 @@ func (s ClickHouseSuite) Test_Replident_Full_Unchanged_TOAST_Updates() { require.NoError(s.t, err) contentStr := string(content) - _, err = s.Conn().Exec(s.t.Context(), fmt.Sprintf(` - INSERT INTO %s (c1,c2,t) VALUES ($1,$2,$3)`, srcFullName), 1, 2, contentStr) + _, err = s.Conn().Exec(s.t.Context(), fmt.Sprintf( + `INSERT INTO %s (c1,c2,t) VALUES ($1,$2,$3)`, srcFullName), 1, 2, contentStr) require.NoError(s.t, err) e2e.EnvWaitForEqualTablesWithNames(env, s, "waiting on initial insert", srcTableName, dstTableName, "id,c1,c2,t") - _, err = s.Conn().Exec(s.t.Context(), fmt.Sprintf(` - UPDATE %s SET c1=$1 WHERE id=$2`, srcFullName), 3, 1) + _, err = s.Conn().Exec(s.t.Context(), fmt.Sprintf( + `UPDATE %s SET c1=$1 WHERE id=$2`, srcFullName), 3, 1) require.NoError(s.t, err) e2e.EnvWaitForEqualTablesWithNames(env, s, "waiting on update", srcTableName, dstTableName, "id,c1,c2,t") @@ -434,6 +453,10 @@ func (s ClickHouseSuite) Test_Replident_Full_Unchanged_TOAST_Updates() { } func (s ClickHouseSuite) WeirdTable(tableName string) { + if s.cluster { + s.t.Skip("SetupCDCFlowStatusQuery stuck in snapshot somehow") + } + srcTableName := tableName srcFullName := s.attachSchemaSuffix(fmt.Sprintf("\"%s\"", tableName)) dstTableName := tableName @@ -455,13 +478,12 @@ func (s ClickHouseSuite) WeirdTable(tableName string) { connectionGen := e2e.FlowConnectionGenerationConfig{ FlowJobName: s.attachSuffix("clickhouse_test_weird_table_" + strings.ReplaceAll( strings.ToLower(tableName), "-", "_")), - TableMappings: []*protos.TableMapping{ - { - SourceTableIdentifier: s.attachSchemaSuffix(tableName), - DestinationTableIdentifier: dstTableName, - Exclude: []string{"excludedColumn"}, - }, - }, + TableMappings: []*protos.TableMapping{{ + SourceTableIdentifier: s.attachSchemaSuffix(tableName), + DestinationTableIdentifier: dstTableName, + Exclude: []string{"excludedColumn?"}, + ShardingKey: "id", + }}, Destination: s.Peer().Name, } flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s) @@ -490,7 +512,11 @@ func (s ClickHouseSuite) WeirdTable(tableName string) { // now test weird names with rename based resync ch, err := connclickhouse.Connect(s.t.Context(), nil, s.Peer().GetClickhouseConfig()) require.NoError(s.t, err) - require.NoError(s.t, ch.Exec(s.t.Context(), "DROP TABLE "+clickhouse.QuoteIdentifier(dstTableName))) + onCluster := "" + if s.cluster { + onCluster = " ON CLUSTER cicluster" + } + require.NoError(s.t, ch.Exec(s.t.Context(), "DROP TABLE "+clickhouse.QuoteIdentifier(dstTableName)+onCluster)) require.NoError(s.t, ch.Close()) flowConnConfig.Resync = true env = e2e.ExecutePeerflow(s.t.Context(), tc, peerflow.CDCFlowWorkflow, flowConnConfig, nil) @@ -508,7 +534,7 @@ func (s ClickHouseSuite) WeirdTable(tableName string) { // now test weird names with exchange based resync ch, err = connclickhouse.Connect(s.t.Context(), nil, s.Peer().GetClickhouseConfig()) require.NoError(s.t, err) - require.NoError(s.t, ch.Exec(s.t.Context(), "TRUNCATE TABLE "+clickhouse.QuoteIdentifier(dstTableName))) + require.NoError(s.t, ch.Exec(s.t.Context(), "TRUNCATE TABLE "+clickhouse.QuoteIdentifier(dstTableName)+onCluster)) require.NoError(s.t, ch.Close()) env = e2e.ExecutePeerflow(s.t.Context(), tc, peerflow.CDCFlowWorkflow, flowConnConfig, nil) e2e.SetupCDCFlowStatusQuery(s.t, env, flowConnConfig) @@ -1291,13 +1317,12 @@ func (s ClickHouseSuite) Test_Column_Exclusion() { config := &protos.FlowConnectionConfigs{ FlowJobName: s.attachSuffix(tableName), DestinationName: s.Peer().Name, - TableMappings: []*protos.TableMapping{ - { - SourceTableIdentifier: srcFullName, - DestinationTableIdentifier: dstTableName, - Exclude: []string{"c2"}, - }, - }, + TableMappings: []*protos.TableMapping{{ + SourceTableIdentifier: srcFullName, + DestinationTableIdentifier: dstTableName, + Exclude: []string{"c2"}, + ShardingKey: "id", + }}, SourceName: s.Source().GeneratePeer(s.t).Name, SyncedAtColName: "_PEERDB_SYNCED_AT", MaxBatchSize: 100, @@ -1507,6 +1532,7 @@ func (s ClickHouseSuite) Test_Unprivileged_Postgres_Columns() { SourceTableIdentifier: srcFullName, DestinationTableIdentifier: dstTableName, Exclude: []string{"se'cret"}, + ShardingKey: "id", }}, Destination: s.Peer().Name, } @@ -1612,7 +1638,11 @@ func (s ClickHouseSuite) Test_Normalize_Metadata_With_Retry() { ch, err := connclickhouse.Connect(s.t.Context(), nil, s.Peer().GetClickhouseConfig()) require.NoError(s.t, err) fakeDestination2 := "test_normalize_metadata_with_retry_dst_2_fake" - renameErr := ch.Exec(s.t.Context(), fmt.Sprintf(`RENAME TABLE %s TO %s`, dstTableName2, fakeDestination2)) + onCluster := "" + if s.cluster { + onCluster = " ON CLUSTER cicluster" + } + renameErr := ch.Exec(s.t.Context(), fmt.Sprintf(`RENAME TABLE %s TO %s%s`, dstTableName2, fakeDestination2, onCluster)) require.NoError(s.t, renameErr) require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`UPDATE %s SET "key"='update1'`, srcFullName2))) @@ -1652,7 +1682,7 @@ func (s ClickHouseSuite) Test_Normalize_Metadata_With_Retry() { }) // Rename the table back to simulate a successful push to ClickHouse - renameErr = ch.Exec(s.t.Context(), fmt.Sprintf(`RENAME TABLE %s TO %s`, fakeDestination2, dstTableName2)) + renameErr = ch.Exec(s.t.Context(), fmt.Sprintf(`RENAME TABLE %s TO %s%s`, fakeDestination2, dstTableName2, onCluster)) require.NoError(s.t, renameErr) require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`UPDATE %s SET "key"='update2'`, srcFullName2))) require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`UPDATE %s SET "key"='update2'`, srcFullName1))) @@ -1668,7 +1698,6 @@ func (s ClickHouseSuite) Test_Normalize_Metadata_With_Retry() { if len(rows.Records) == 0 { return false } - return rows.Records[0][0].Value().(int64) == 2 }) @@ -1698,7 +1727,7 @@ func (s ClickHouseSuite) Test_Normalize_Metadata_With_Retry() { s.t.Log("no records found in metadata_last_sync_state") return false } - s.t.Log("metadata_last_sync_state:", rows.Records[0][0].Value(), rows.Records[0][1].Value()) + s.t.Log("metadata_last_sync_state", rows.Records[0][0].Value(), rows.Records[0][1].Value()) return rows.Records[0][0].Value().(int64) == 2 && rows.Records[0][1].Value().(int64) == 2 }) @@ -1997,6 +2026,7 @@ func (s ClickHouseSuite) Test_NullEngine() { SourceTableIdentifier: srcFullName, DestinationTableIdentifier: dstTableName, Engine: protos.TableEngine_CH_ENGINE_NULL, + ShardingKey: "id", }}, Destination: s.Peer().Name, } @@ -2015,9 +2045,25 @@ func (s ClickHouseSuite) Test_NullEngine() { require.NoError(s.t, ch.Close()) require.NoError(s.t, s.source.Exec(s.t.Context(), - fmt.Sprintf(`insert into %s values (1, 'cdc', 'val')`, srcFullName))) + fmt.Sprintf(`insert into %s values (1,'cdc','val')`, srcFullName))) e2e.EnvWaitForEqualTablesWithNames(env, s, "null insert", srcTableName, "nulltarget", "id,\"key\"") + require.NoError(s.t, s.source.Exec(s.t.Context(), + fmt.Sprintf(`ALTER TABLE %s ADD COLUMN added INT`, srcFullName))) + require.NoError(s.t, s.source.Exec(s.t.Context(), + fmt.Sprintf(`insert into %s values (2,'no','add',0)`, srcFullName))) + e2e.EnvWaitForEqualTablesWithNames(env, s, "null insert after column added", srcTableName, "nulltarget", "id,\"key\"") + + var count uint64 + ch, err = connclickhouse.Connect(s.t.Context(), nil, chPeer) + require.NoError(s.t, err) + row := ch.QueryRow(s.t.Context(), + fmt.Sprintf("select count(*) from system.columns where database = '%s' and table = 'test_nullengine'", chPeer.Database)) + require.NoError(s.t, row.Err()) + require.NoError(s.t, row.Scan(&count)) + require.NoError(s.t, ch.Close()) + require.Equal(s.t, uint64(7), count) + env.Cancel(s.t.Context()) e2e.RequireEnvCanceled(s.t, env) env = e2e.ExecuteWorkflow(s.t.Context(), tc, shared.PeerFlowTaskQueue, peerflow.DropFlowWorkflow, &protos.DropFlowInput{ @@ -2039,15 +2085,14 @@ func (s ClickHouseSuite) Test_NullEngine() { e2e.SetupCDCFlowStatusQuery(s.t, env, flowConnConfig) e2e.EnvWaitForEqualTablesWithNames(env, s, "waiting on initial", srcTableName, "nulltarget", "id,\"key\"") - var count uint64 ch, err = connclickhouse.Connect(s.t.Context(), nil, chPeer) require.NoError(s.t, err) - row := ch.QueryRow(s.t.Context(), + row = ch.QueryRow(s.t.Context(), fmt.Sprintf("select count(*) from system.columns where database = '%s' and table = 'test_nullengine'", chPeer.Database)) require.NoError(s.t, row.Err()) require.NoError(s.t, row.Scan(&count)) require.NoError(s.t, ch.Close()) - require.Equal(s.t, uint64(5), count) + require.Equal(s.t, uint64(6), count) env.Cancel(s.t.Context()) e2e.RequireEnvCanceled(s.t, env) @@ -2084,6 +2129,7 @@ func (s ClickHouseSuite) Test_Partition_Key_Integer() { SourceTableIdentifier: srcFullName, DestinationTableIdentifier: dstTableName, PartitionKey: "id", + ShardingKey: "id", }}, Destination: s.Peer().Name, } @@ -2141,6 +2187,7 @@ func (s ClickHouseSuite) Test_Partition_Key_Timestamp() { SourceTableIdentifier: srcFullName, DestinationTableIdentifier: dstTableName, PartitionKey: "updated_at", + ShardingKey: "id", }}, Destination: s.Peer().Name, } diff --git a/flow/e2e/congen.go b/flow/e2e/congen.go index 8e332102e..cc579fbc7 100644 --- a/flow/e2e/congen.go +++ b/flow/e2e/congen.go @@ -31,6 +31,7 @@ func TableMappings(s GenericSuite, tables ...string) []*protos.TableMapping { tm = append(tm, &protos.TableMapping{ SourceTableIdentifier: AttachSchema(s, tables[i]), DestinationTableIdentifier: s.DestinationTable(tables[i+1]), + ShardingKey: "id", }) } return tm @@ -64,6 +65,7 @@ func (c *FlowConnectionGenerationConfig) GenerateFlowConnectionConfigs(s Suite) tblMappings = append(tblMappings, &protos.TableMapping{ SourceTableIdentifier: k, DestinationTableIdentifier: v, + ShardingKey: "id", }) } } diff --git a/flow/e2e/generic/generic_test.go b/flow/e2e/generic/generic_test.go index f88c9b1fd..6003d56da 100644 --- a/flow/e2e/generic/generic_test.go +++ b/flow/e2e/generic/generic_test.go @@ -35,7 +35,7 @@ func TestGenericBQ(t *testing.T) { } func TestGenericCH_PG(t *testing.T) { - e2eshared.RunSuite(t, SetupGenericSuite(e2e_clickhouse.SetupSuite(t, func(t *testing.T) (*e2e.PostgresSource, string, error) { + e2eshared.RunSuite(t, SetupGenericSuite(e2e_clickhouse.SetupSuite(t, false, func(t *testing.T) (*e2e.PostgresSource, string, error) { t.Helper() suffix := "pgchg_" + strings.ToLower(shared.RandomString(8)) source, err := e2e.SetupPostgres(t, suffix) @@ -44,7 +44,7 @@ func TestGenericCH_PG(t *testing.T) { } func TestGenericCH_MySQL(t *testing.T) { - e2eshared.RunSuite(t, SetupGenericSuite(e2e_clickhouse.SetupSuite(t, func(t *testing.T) (*e2e.MySqlSource, string, error) { + e2eshared.RunSuite(t, SetupGenericSuite(e2e_clickhouse.SetupSuite(t, false, func(t *testing.T) (*e2e.MySqlSource, string, error) { t.Helper() suffix := "mychg_" + strings.ToLower(shared.RandomString(8)) source, err := e2e.SetupMySQL(t, suffix) @@ -52,6 +52,24 @@ func TestGenericCH_MySQL(t *testing.T) { }))) } +func TestGenericChCluster_PG(t *testing.T) { + e2eshared.RunSuite(t, SetupGenericSuite(e2e_clickhouse.SetupSuite(t, true, func(t *testing.T) (*e2e.PostgresSource, string, error) { + t.Helper() + suffix := "pgchclg_" + strings.ToLower(shared.RandomString(8)) + source, err := e2e.SetupPostgres(t, suffix) + return source, suffix, err + }))) +} + +func TestGenericChCluster_MySQL(t *testing.T) { + e2eshared.RunSuite(t, SetupGenericSuite(e2e_clickhouse.SetupSuite(t, true, func(t *testing.T) (*e2e.MySqlSource, string, error) { + t.Helper() + suffix := "mychclg_" + strings.ToLower(shared.RandomString(8)) + source, err := e2e.SetupMySQL(t, suffix) + return source, suffix, err + }))) +} + type Generic struct { e2e.GenericSuite } diff --git a/flow/e2e/mongo/mongo_test.go b/flow/e2e/mongo/mongo_test.go index fa432f5b3..f0605ddc1 100644 --- a/flow/e2e/mongo/mongo_test.go +++ b/flow/e2e/mongo/mongo_test.go @@ -32,7 +32,7 @@ func TestMongoClickhouseSuite(t *testing.T) { func SetupMongoClickhouseSuite(t *testing.T) MongoClickhouseSuite { t.Helper() - return MongoClickhouseSuite{e2e_clickhouse.SetupSuite(t, func(t *testing.T) (*MongoSource, string, error) { + return MongoClickhouseSuite{e2e_clickhouse.SetupSuite(t, false, func(t *testing.T) (*MongoSource, string, error) { t.Helper() suffix := "mongoch_" + strings.ToLower(shared.RandomString(8)) source, err := SetupMongo(t, suffix) diff --git a/flow/e2e/snowflake/snowflake_schema_delta_test.go b/flow/e2e/snowflake/snowflake_schema_delta_test.go index ef8444a79..3a3099e26 100644 --- a/flow/e2e/snowflake/snowflake_schema_delta_test.go +++ b/flow/e2e/snowflake/snowflake_schema_delta_test.go @@ -53,7 +53,7 @@ func (s SnowflakeSchemaDeltaTestSuite) TestSimpleAddColumn() { err := s.sfTestHelper.RunCommand(s.t.Context(), fmt.Sprintf("CREATE TABLE %s(ID TEXT PRIMARY KEY)", tableName)) require.NoError(s.t, err) - err = s.connector.ReplayTableSchemaDeltas(s.t.Context(), nil, "schema_delta_flow", []*protos.TableSchemaDelta{{ + require.NoError(s.t, s.connector.ReplayTableSchemaDeltas(s.t.Context(), nil, "schema_delta_flow", nil, []*protos.TableSchemaDelta{{ SrcTableName: tableName, DstTableName: tableName, AddedColumns: []*protos.FieldDescription{ @@ -63,8 +63,7 @@ func (s SnowflakeSchemaDeltaTestSuite) TestSimpleAddColumn() { TypeModifier: -1, }, }, - }}) - require.NoError(s.t, err) + }})) output, err := s.connector.GetTableSchema(s.t.Context(), nil, 0, protos.TypeSystem_Q, []*protos.TableMapping{{SourceTableIdentifier: tableName}}) @@ -168,12 +167,11 @@ func (s SnowflakeSchemaDeltaTestSuite) TestAddAllColumnTypes() { } } - err = s.connector.ReplayTableSchemaDeltas(s.t.Context(), nil, "schema_delta_flow", []*protos.TableSchemaDelta{{ + require.NoError(s.t, s.connector.ReplayTableSchemaDeltas(s.t.Context(), nil, "schema_delta_flow", nil, []*protos.TableSchemaDelta{{ SrcTableName: tableName, DstTableName: tableName, AddedColumns: addedColumns, - }}) - require.NoError(s.t, err) + }})) output, err := s.connector.GetTableSchema(s.t.Context(), nil, 0, protos.TypeSystem_Q, []*protos.TableMapping{{SourceTableIdentifier: tableName}}) @@ -248,12 +246,11 @@ func (s SnowflakeSchemaDeltaTestSuite) TestAddTrickyColumnNames() { } } - err = s.connector.ReplayTableSchemaDeltas(s.t.Context(), nil, "schema_delta_flow", []*protos.TableSchemaDelta{{ + require.NoError(s.t, s.connector.ReplayTableSchemaDeltas(s.t.Context(), nil, "schema_delta_flow", nil, []*protos.TableSchemaDelta{{ SrcTableName: tableName, DstTableName: tableName, AddedColumns: addedColumns, - }}) - require.NoError(s.t, err) + }})) output, err := s.connector.GetTableSchema(s.t.Context(), nil, 0, protos.TypeSystem_Q, []*protos.TableMapping{{SourceTableIdentifier: tableName}}) @@ -304,12 +301,11 @@ func (s SnowflakeSchemaDeltaTestSuite) TestAddWhitespaceColumnNames() { } } - err = s.connector.ReplayTableSchemaDeltas(s.t.Context(), nil, "schema_delta_flow", []*protos.TableSchemaDelta{{ + require.NoError(s.t, s.connector.ReplayTableSchemaDeltas(s.t.Context(), nil, "schema_delta_flow", nil, []*protos.TableSchemaDelta{{ SrcTableName: tableName, DstTableName: tableName, AddedColumns: addedColumns, - }}) - require.NoError(s.t, err) + }})) output, err := s.connector.GetTableSchema(s.t.Context(), nil, 0, protos.TypeSystem_Q, []*protos.TableMapping{{SourceTableIdentifier: tableName}}) diff --git a/flow/e2e/test_utils.go b/flow/e2e/test_utils.go index c8ff225f3..75d0b2117 100644 --- a/flow/e2e/test_utils.go +++ b/flow/e2e/test_utils.go @@ -254,7 +254,7 @@ func SetupCDCFlowStatusQuery(t *testing.T, env WorkflowRun, config *protos.FlowC if err == nil { var status protos.FlowStatus if err := response.Get(&status); err != nil { - t.Fatal(err) + t.Fatal(err.Error()) } else if status == protos.FlowStatus_STATUS_RUNNING || status == protos.FlowStatus_STATUS_COMPLETED { return } else if counter > 30 { diff --git a/flow/internal/workflow.go b/flow/internal/workflow.go index f2fd31b7d..549aec373 100644 --- a/flow/internal/workflow.go +++ b/flow/internal/workflow.go @@ -75,8 +75,7 @@ func GetWorkflowStatus(ctx context.Context, pool shared.CatalogPool, func UpdateFlowStatusInCatalog(ctx context.Context, pool shared.CatalogPool, workflowID string, status protos.FlowStatus, ) (protos.FlowStatus, error) { - _, err := pool.Exec(ctx, "UPDATE flows SET status=$1,updated_at=now() WHERE workflow_id=$2", status, workflowID) - if err != nil { + if _, err := pool.Exec(ctx, "UPDATE flows SET status=$1,updated_at=now() WHERE workflow_id=$2", status, workflowID); err != nil { slog.Error("failed to update flow status", slog.Any("error", err), slog.String("flowID", workflowID)) return status, fmt.Errorf("failed to update flow status: %w", err) } diff --git a/nexus/analyzer/src/lib.rs b/nexus/analyzer/src/lib.rs index d62f68948..57e769c65 100644 --- a/nexus/analyzer/src/lib.rs +++ b/nexus/analyzer/src/lib.rs @@ -791,6 +791,14 @@ fn parse_db_options(db_type: DbType, with_options: &[SqlOption]) -> anyhow::Resu .get("tls_host") .map(|s| s.to_string()) .unwrap_or_default(), + cluster: opts + .get("cluster") + .map(|s| s.to_string()) + .unwrap_or_default(), + replicated: opts + .get("replicated") + .map(|s| s.parse::().unwrap_or_default()) + .unwrap_or_default(), s3: None, }; Config::ClickhouseConfig(clickhouse_config) diff --git a/nexus/flow-rs/src/grpc.rs b/nexus/flow-rs/src/grpc.rs index 92b5ccf47..4c59a39ee 100644 --- a/nexus/flow-rs/src/grpc.rs +++ b/nexus/flow-rs/src/grpc.rs @@ -105,8 +105,7 @@ impl FlowGrpcClient { destination_table_identifier: mapping.destination_table_identifier.clone(), partition_key: mapping.partition_key.clone().unwrap_or_default(), exclude: mapping.exclude.clone(), - columns: Default::default(), - engine: Default::default(), + ..Default::default() }) .collect::>(); diff --git a/protos/flow.proto b/protos/flow.proto index 2dfecb8f7..43a178f5c 100644 --- a/protos/flow.proto +++ b/protos/flow.proto @@ -30,6 +30,8 @@ message TableMapping { repeated string exclude = 4; repeated ColumnSetting columns = 5; TableEngine engine = 6; + string sharding_key = 7; + string policy_name = 8; } message SetupInput { diff --git a/protos/peers.proto b/protos/peers.proto index 26d8e1c5e..79d8e2c93 100644 --- a/protos/peers.proto +++ b/protos/peers.proto @@ -157,7 +157,7 @@ message ClickhouseConfig{ string user = 3; string password = 4 [(peerdb_redacted) = true]; string database = 5; - string s3_path = 6; // path to S3 bucket which will store avro files + string s3_path = 6; string access_key_id = 7 [(peerdb_redacted) = true]; string secret_access_key = 8 [(peerdb_redacted) = true]; string region = 9; @@ -168,6 +168,8 @@ message ClickhouseConfig{ optional string root_ca = 14 [(peerdb_redacted) = true]; string tls_host = 15; optional S3Config s3 = 16; + string cluster = 17; + bool replicated = 18; } message SqlServerConfig { diff --git a/ui/app/dto/MirrorsDTO.ts b/ui/app/dto/MirrorsDTO.ts index 53d910ef0..9cf33354e 100644 --- a/ui/app/dto/MirrorsDTO.ts +++ b/ui/app/dto/MirrorsDTO.ts @@ -1,8 +1,4 @@ -import { - ColumnSetting, - FlowConnectionConfigs, - TableEngine, -} from '@/grpc_generated/flow'; +import { FlowConnectionConfigs, TableMapping } from '@/grpc_generated/flow'; export enum MirrorType { CDC = 'CDC', @@ -15,16 +11,16 @@ export type CDCConfig = FlowConnectionConfigs & { envString: string; }; -export type TableMapRow = { +export type TableMapRow = Omit< + TableMapping, + 'exclude' | 'sourceTableIdentifier' | 'destinationTableIdentifier' +> & { schema: string; source: string; destination: string; - partitionKey: string; exclude: Set; selected: boolean; canMirror: boolean; tableSize: string; editingDisabled: boolean; - engine: TableEngine; - columns: ColumnSetting[]; }; diff --git a/ui/app/mirrors/create/cdc/schemabox.tsx b/ui/app/mirrors/create/cdc/schemabox.tsx index e5faa2313..02d7ae9f5 100644 --- a/ui/app/mirrors/create/cdc/schemabox.tsx +++ b/ui/app/mirrors/create/cdc/schemabox.tsx @@ -122,6 +122,20 @@ export default function SchemaBox({ setRows(newRows); }; + const updateShardingKey = (source: string, shardingKey: string) => { + const newRows = [...rows]; + const index = newRows.findIndex((row) => row.source === source); + newRows[index] = { ...newRows[index], shardingKey }; + setRows(newRows); + }; + + const updatePolicyName = (source: string, policyName: string) => { + const newRows = [...rows]; + const index = newRows.findIndex((row) => row.source === source); + newRows[index] = { ...newRows[index], policyName }; + setRows(newRows); + }; + const addTableColumns = useCallback( (table: string) => { const [schemaName, tableName] = table.split('.'); @@ -245,11 +259,6 @@ export default function SchemaBox({ const engineOptions = [ { value: 'CH_ENGINE_REPLACING_MERGE_TREE', label: 'ReplacingMergeTree' }, { value: 'CH_ENGINE_MERGE_TREE', label: 'MergeTree' }, - { - value: 'CH_ENGINE_REPLICATED_REPLACING_MERGE_TREE', - label: 'ReplicatedReplacingMergeTree', - }, - { value: 'CH_ENGINE_REPLICATED_MERGE_TREE', label: 'ReplicatedMergeTree' }, { value: 'CH_ENGINE_NULL', label: 'Null' }, ]; @@ -398,22 +407,60 @@ export default function SchemaBox({ {peerType?.toString() === DBType[DBType.CLICKHOUSE].toString() && ( -
- Engine: - - selectedOption && - updateEngine( - row.source, - tableEngineFromJSON(selectedOption.value) - ) - } - /> -
+ <> +
+ Engine: + + selectedOption && + updateEngine( + row.source, + tableEngineFromJSON(selectedOption.value) + ) + } + /> +
+
+ Sharding Key: + + ) => + updateShardingKey(row.source, e.target.value) + } + /> +
+
+ Policy Name: + + ) => + updatePolicyName(row.source, e.target.value) + } + /> +
+ )} diff --git a/ui/app/mirrors/create/handlers.ts b/ui/app/mirrors/create/handlers.ts index e7696d860..d6239788e 100644 --- a/ui/app/mirrors/create/handlers.ts +++ b/ui/app/mirrors/create/handlers.ts @@ -186,6 +186,8 @@ export function reformattedTableMapping( exclude: Array.from(row.exclude), columns: row.columns, engine: row.engine, + shardingKey: row.shardingKey, + policyName: row.policyName, })); } @@ -434,8 +436,6 @@ export async function fetchTables( const tableRes = tablesRes.tables; if (tableRes) { for (const tableObject of tableRes) { - // setting defaults: - // for bigquery, tables are not schema-qualified const dstName = getDefaultDestinationTable( peerType!, targetSchemaName, @@ -453,6 +453,8 @@ export async function fetchTables( editingDisabled: false, columns: [], engine: TableEngine.CH_ENGINE_REPLACING_MERGE_TREE, + shardingKey: '', + policyName: '', }); } } diff --git a/ui/app/peers/create/[peerType]/helpers/ch.ts b/ui/app/peers/create/[peerType]/helpers/ch.ts index 15138b361..63dfea9e4 100644 --- a/ui/app/peers/create/[peerType]/helpers/ch.ts +++ b/ui/app/peers/create/[peerType]/helpers/ch.ts @@ -37,6 +37,21 @@ export const clickhouseSetting: PeerSetting[] = [ setter((curr) => ({ ...curr, database: value as string })), tips: 'Specify which database to associate with this peer.', }, + { + label: 'Cluster', + stateHandler: (value, setter) => + setter((curr) => ({ ...curr, cluster: value as string })), + tips: 'Specify which cluster to associate with this peer. Not relevant on ClickHouse Cloud.', + optional: true, + }, + { + label: 'Replicated?', + stateHandler: (value, setter) => + setter((curr) => ({ ...curr, replicated: value as boolean })), + type: 'switch', + tips: 'Enable to use ReplicatedMergeTree & ReplicatedReplacingMergeTree. Not relevant on ClickHouse Cloud.', + optional: true, + }, { label: 'Disable TLS?', stateHandler: (value, setter) => @@ -232,4 +247,6 @@ export const blankClickHouseSetting: ClickhouseConfig = { disableTls: false, endpoint: undefined, tlsHost: '', + cluster: '', + replicated: false, };