Better DSN support for clickhouse

Database, Dialect and inserter now created base on the dsn driver type

Author: billettc

cmd/substreams-sink-sql/from_proto.go

@@ -14,6 +14,7 @@ import (
 	"github.com/streamingfast/substreams-sink-sql/db_proto"
 	"github.com/streamingfast/substreams-sink-sql/db_proto/proto"
 	protosql "github.com/streamingfast/substreams-sink-sql/db_proto/sql"
+	clickhouse "github.com/streamingfast/substreams-sink-sql/db_proto/sql/click_house"
 	"github.com/streamingfast/substreams-sink-sql/db_proto/sql/postgres"
 	schema2 "github.com/streamingfast/substreams-sink-sql/db_proto/sql/schema"
 	stats2 "github.com/streamingfast/substreams-sink-sql/db_proto/stats"
@@ -38,8 +39,8 @@ var fromProtoCmd = Command(fromProtoE,
 		//flags.Bool("no-transactions", false, "Do not use transactions when inserting data. This is useful to speed up the initial import of a large dataset.")
 		//flags.Bool("parallel", false, "Run the sinker in parallel mode. This is useful to speed up the initial import of a large dataset. This is will process blocks of a batch in parallel")
 		flags.Int("block-batch-size", 25, "number of blocks to process at a time")
-		//flags.String("clickhouse-sink-info-folder", "", "folder where to store the clickhouse sink info")
-		//flags.String("clickhouse-cursor-file-path", "cursor.txt", "file name where to store the clickhouse cursor")
+		flags.String("clickhouse-sink-info-folder", "", "folder where to store the clickhouse sink info")
+		flags.String("clickhouse-cursor-file-path", "cursor.txt", "file name where to store the clickhouse cursor")
 	}),
 )
 
@@ -52,7 +53,6 @@ var fromProtoCmd = Command(fromProtoE,
 //todo: post generate index
 //todo: external process
 //todo: handle network
-//todo: fix DSN for clickhouse
 
 func fromProtoE(cmd *cobra.Command, args []string) error {
 	//app := NewApplication(cmd.Context())
@@ -161,8 +161,6 @@ func fromProtoE(cmd *cobra.Command, args []string) error {
 		}
 	}
 
-	//todo: fix me
-	//schemaName := "test"
 	schemaName := dsn.Schema()
 
 	schema, err := schema2.NewSchema(schemaName, rootMessageDescriptor, useProtoOption, zlog)
@@ -185,38 +183,52 @@ func fromProtoE(cmd *cobra.Command, args []string) error {
 	}
 
 	connectionString := dsn.ConnString()
-	//todo: fix me
-	//connectionString = "http://localhost:8123?secure=false"
 
-	fmt.Println("connection string", connectionString)
+	zlog.Info("connecting to db", zap.String("dsn", connectionString))
 	sqlDB, err := sql.Open(dsn.Driver(), connectionString)
 	if err != nil {
 		return fmt.Errorf("open db connection: %w", err)
 	}
 
-	dialect, err := postgres.NewDialectPostgres(schema.Name, schema.TableRegistry, zlog)
-	//dialect, err := clickhouse.NewDialectClickHouse(schema.Name, schema.TableRegistry, zlog)
-	if err != nil {
-		return fmt.Errorf("creating dialect: %w", err)
-	}
-
+	var dialect protosql.Dialect
 	var database protosql.Database
-	//implDatabase, err := clickhouse.NewDatabase(
-	//	schemaName,
-	//	dialect,
-	//	sqlDB,
-	//	outputModuleName,
-	//	rootMessageDescriptor,
-	//	sflags.MustGetString(cmd, "clickhouse-sink-info-folder"),
-	//	sflags.MustGetString(cmd, "clickhouse-cursor-file-path"),
-	//	true,
-	//	zlog,
-	//)
-	implDatabase, err := postgres.NewDatabase(schemaName, dialect, sqlDB, outputModuleName, rootMessageDescriptor, useProtoOption, zlog)
-	if err != nil {
-		return fmt.Errorf("creating database: %w", err)
+
+	switch dsn.Driver() {
+	case "postgres":
+		d, err := postgres.NewDialectPostgres(schema.Name, schema.TableRegistry, zlog)
+		if err != nil {
+			return fmt.Errorf("creating postgres dialect: %w", err)
+		}
+		dialect = d
+		database, err = postgres.NewDatabase(schemaName, d, sqlDB, outputModuleName, rootMessageDescriptor, useProtoOption, zlog)
+		if err != nil {
+			return fmt.Errorf("creating postgres database: %w", err)
+		}
+
+	case "clickhouse":
+		d, err := clickhouse.NewDialectClickHouse(schema.Name, schema.TableRegistry, zlog)
+		if err != nil {
+			return fmt.Errorf("creating clickhouse dialect: %w", err)
+		}
+		dialect = d
+		database, err = clickhouse.NewDatabase(
+			schemaName,
+			d,
+			sqlDB,
+			outputModuleName,
+			rootMessageDescriptor,
+			sflags.MustGetString(cmd, "clickhouse-sink-info-folder"),
+			sflags.MustGetString(cmd, "clickhouse-cursor-file-path"),
+			true,
+			zlog,
+		)
+		if err != nil {
+			return fmt.Errorf("creating clickhouse database: %w", err)
+		}
+	default:
+		panic(fmt.Sprintf("unsupported driver: %s", dsn.Driver()))
+
 	}
-	database = implDatabase
 
 	sinkInfo, err := database.FetchSinkInfo(schema.Name)
 	if err != nil {
@@ -291,19 +303,24 @@ func fromProtoE(cmd *cobra.Command, args []string) error {
 		}
 	}
 
-	//inserter, err := clickhouse.NewAccumulatorInserter(implDatabase, zlog)
-
 	var inserter protosql.Inserter
-	if useConstraints {
-		inserter, err = postgres.NewRowInserter(implDatabase, zlog)
-		if err != nil {
-			return fmt.Errorf("creating row inserter: %w", err)
-		}
-	} else {
-		inserter, err = postgres.NewAccumulatorInserter(implDatabase, zlog)
-		if err != nil {
-			return fmt.Errorf("creating accumulator inserter: %w", err)
+	switch dsn.Driver() {
+	case "postgres":
+		if useConstraints {
+			inserter, err = postgres.NewRowInserter(database.(*postgres.Database), zlog)
+			if err != nil {
+				return fmt.Errorf("creating row inserter: %w", err)
+			}
+		} else {
+			inserter, err = postgres.NewAccumulatorInserter(database.(*postgres.Database), zlog)
+			if err != nil {
+				return fmt.Errorf("creating accumulator inserter: %w", err)
+			}
 		}
+	case "clickhouse":
+		inserter, err = clickhouse.NewAccumulatorInserter(database.(*clickhouse.Database), zlog)
+	default:
+		panic(fmt.Sprintf("unsupported driver: %s", dsn.Driver()))
 	}
 
 	database.SetInserter(inserter)

db_changes/db/dsn.go

@@ -105,6 +105,16 @@ func (c *DSN) Driver() string {
 
 func (c *DSN) ConnString() string {
 	if c.driver == "clickhouse" {
+		for _, option := range c.options {
+			if c.host == "localhost" {
+				c.host = "127.0.0.1"
+				scheme := "http"
+				if option == "secure=true" {
+					scheme = "https"
+				}
+				return strings.Replace(c.original, "clickhouse://", scheme+"://", 1)
+			}
+		}
 		return c.original
 	}
 	out := fmt.Sprintf("host=%s port=%d dbname=%s %s", c.host, c.port, c.database, strings.Join(c.options, " "))

db_proto/sql/click_house/accumulator_inserter.go

@@ -13,6 +13,7 @@ import (
 
 type accumulator struct {
 	ordinal   int
+	tableName string
 	query     string
 	rowValues [][]string
 }
@@ -34,8 +35,9 @@ func NewAccumulatorInserter(database *Database, logger *zap.Logger) (*Accumulato
 			return nil, fmt.Errorf("creating insert from descriptor for table %q: %w", table.Name, err)
 		}
 		accumulators[table.Name] = &accumulator{
-			ordinal: table.Ordinal,
-			query:   query,
+			tableName: table.Name,
+			ordinal:   table.Ordinal,
+			query:     query,
 		}
 	}
 	accumulators["_blocks_"] = &accumulator{
@@ -115,6 +117,7 @@ func (i *AccumulatorInserter) Flush(tx *sql.Tx) error {
 	})
 
 	for _, acc := range accumulators {
+		i.logger.Debug("flushing table", zap.String("table", acc.tableName), zap.Int("ordinal", acc.ordinal), zap.Int("row_count", len(acc.rowValues)))
 		if len(acc.rowValues) == 0 {
 			continue
 		}

db_proto/sql/click_house/dialect.go

@@ -23,7 +23,7 @@ const staticSqlCreateBlock = `
 		hash      text,
 		timestamp timestamp
 	)
-	ENGINE = MergeTree()
+	ENGINE = ReplacingMergeTree()
 	PRIMARY KEY (number)
 `