Merge branch 'develop' into chris/from-proto-export

Author: chriswessels

.github/workflows/build.yml

@@ -0,0 +1,41 @@
+name: Build and Test
+
+on:
+  push:
+    branches: [ "develop", "main" ]
+  pull_request:
+    branches: [ "develop", "main" ]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    
+    steps:
+    - uses: actions/checkout@v4
+
+    - name: Set up Go
+      uses: actions/setup-go@v5
+      with:
+        go-version-file: 'go.mod'
+
+    - name: Cache Go modules
+      uses: actions/cache@v4
+      with:
+        path: |
+          ~/.cache/go-build
+          ~/go/pkg/mod
+        key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
+        restore-keys: |
+          ${{ runner.os }}-go-
+
+    - name: Download dependencies
+      run: go mod download
+
+    - name: Verify dependencies
+      run: go mod verify
+
+    - name: Build
+      run: go build -v ./...
+
+    - name: Run tests
+      run: go test -v ./...

.gitignore

@@ -10,4 +10,4 @@ build/
 /substreams-sink-sql
 /cursor.txt
 /replay.log
-/spl_all_schema_hash.txt
+/*_schema_hash.txt

CHANGELOG.md

@@ -5,14 +5,32 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## Unreleased
+## v4.6.6
+
+### `substreams-sink-sql run`
+
+* Fixed a potential ordering issue if some tables are seen before others but rows are still dependent of each other.
+
+## v4.6.5
+
+### `substreams-sink-sql run`
+
+* Fixed delete operation on composite keys.
+
+* Read tables information only for given schema to avoid walking non-related tables.
 
 ### Clickhouse & `substreams-sink-sql from-proto`
 
+* Improved signal handling by properly logging which signal was received.
+
 * If a `clickhouse_table_options.partition_fields` already contains some partition for `_block_timestamp_`, the default `(toYYYYMM(_block_timestamp_))` will not be added anymore.
 
 * The default partition on `_block_timestamp_` is now `(toYYYYMM(_block_timestamp_))` instead of just `_block_timestamp_`.
 
+### Both implementations
+
+* Automatically skip block flush if the average flush duration is above the time between blocks, preventing drift in these situations
+
 ## v4.6.4
 
 ### Clickhouse & `substreams-sink-sql from-proto`

README.md

@@ -57,7 +57,7 @@ The Substreams:SQL sink helps you quickly and easily sync Substreams modules to
    **ClickHouse**
 
    ```bash
-   export DSN="clickhouse://default:@localhost:9000/default"
+   export DSN="clickhouse://default:default@localhost:9000/default"
    substreams-sink-sql setup $DSN docs/tutorial/substreams.yaml
    ```
 

cmd/substreams-sink-sql/common_flags.go

@@ -2,17 +2,13 @@ package main
 
 import (
 	"context"
-	"fmt"
-	"strings"
 	"time"
 
-	"github.com/golang/protobuf/proto"
 	"github.com/spf13/pflag"
 	"github.com/streamingfast/bstream"
 	"github.com/streamingfast/cli"
 	"github.com/streamingfast/shutter"
 	sink "github.com/streamingfast/substreams-sink"
-	pbsql "github.com/streamingfast/substreams-sink-sql/pb/sf/substreams/sink/sql/services/v1"
 	pbsubstreams "github.com/streamingfast/substreams/pb/sf/substreams/v1"
 	"go.uber.org/zap"
 )
@@ -23,36 +19,6 @@ var (
 
 var supportedOutputTypes = "sf.substreams.sink.database.v1.DatabaseChanges,sf.substreams.database.v1.DatabaseChanges"
 
-var (
-	supportedDeployableUnits              []string
-	deprecated_supportedDeployableService = "sf.substreams.sink.sql.v1.Service"
-	supportedDeployableService            = "sf.substreams.sink.sql.service.v1.Service"
-)
-
-func init() {
-	supportedDeployableUnits = []string{
-		deprecated_supportedDeployableService,
-	}
-}
-
-func extractSinkService(pkg *pbsubstreams.Package) (*pbsql.Service, error) {
-	if pkg.SinkConfig == nil {
-		return nil, fmt.Errorf("no sink config found in spkg")
-	}
-
-	switch pkg.SinkConfig.TypeUrl {
-	case deprecated_supportedDeployableService, supportedDeployableService:
-		service := &pbsql.Service{}
-
-		if err := proto.Unmarshal(pkg.SinkConfig.Value, service); err != nil {
-			return nil, fmt.Errorf("failed to proto unmarshal: %w", err)
-		}
-		return service, nil
-	}
-
-	return nil, fmt.Errorf("invalid config type %q, supported configs are %q", pkg.SinkConfig.TypeUrl, strings.Join(supportedDeployableUnits, ", "))
-}
-
 // AddCommonSinkerFlags adds the flags common to all command that needs to create a sinker,
 // namely the `run` and `generate-csv` commands.
 func AddCommonSinkerFlags(flags *pflag.FlagSet) {

cmd/substreams-sink-sql/from_proto.go

@@ -6,9 +6,11 @@ import (
 	"github.com/jhump/protoreflect/desc"
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
+	"github.com/streamingfast/cli"
 	. "github.com/streamingfast/cli"
 	"github.com/streamingfast/cli/sflags"
 	sink "github.com/streamingfast/substreams-sink"
+	sinksql "github.com/streamingfast/substreams-sink-sql"
 	"github.com/streamingfast/substreams-sink-sql/db_changes/db"
 	"github.com/streamingfast/substreams-sink-sql/db_proto"
 	"github.com/streamingfast/substreams-sink-sql/db_proto/proto"
@@ -49,7 +51,7 @@ var fromProtoCmd = Command(fromProtoE,
 //todo: handle network
 
 func fromProtoE(cmd *cobra.Command, args []string) error {
-	//app := NewApplication(cmd.Context())
+	app := cli.NewApplication(cmd.Context())
 
 	dsnString := args[0]
 	manifestPath := args[1]
@@ -60,18 +62,11 @@ func fromProtoE(cmd *cobra.Command, args []string) error {
 	}
 
 	useConstraints := !sflags.MustGetBool(cmd, "no-constraints")
-	//useTransactions := !sflags.MustGetBool(cmd, "no-transactions")
 	blockBatchSize := sflags.MustGetInt(cmd, "block-batch-size")
 
 	useTransactions := true
 	parallel := false
 
-	//parallel := sflags.MustGetBool(cmd, "parallel")
-	//if parallel {
-	//	useConstraints = false
-	//	useTransactions = false
-	//}
-
 	endpoint := sflags.MustGetString(cmd, "substreams-endpoint")
 	if endpoint == "" {
 		network := sflags.MustGetString(cmd, "network")
@@ -121,7 +116,7 @@ func fromProtoE(cmd *cobra.Command, args []string) error {
 		return fmt.Errorf("could not find output type for module %s", outputModuleName)
 	}
 
-	service, err := extractSinkService(spkg)
+	service, err := sinksql.ExtractSinkService(spkg)
 	if err != nil {
 		service = &pbsql.Service{}
 	}
@@ -199,11 +194,11 @@ func fromProtoE(cmd *cobra.Command, args []string) error {
 		return fmt.Errorf("creating sinker: %w", err)
 	}
 
-	err = sinker.Run(cmd.Context())
-	if err != nil {
-		return fmt.Errorf("running sinker: %w", err)
+	app.SuperviseAndStartUsing(sinker, sinker.Run)
+
+	if err := app.WaitForTermination(zlog, 0, 0); err != nil {
+		cli.Quit("application terminated with error: %s", err)
 	}
 
-	sinker.LogStats()
 	return nil
 }

cmd/substreams-sink-sql/from_proto_apply_constraints.go

@@ -1,7 +1,6 @@
 package main
 
 import (
-	"errors"
 	"fmt"
 	"time"
 
@@ -10,7 +9,6 @@ import (
 	. "github.com/streamingfast/cli"
 	"github.com/streamingfast/cli/sflags"
 	sink "github.com/streamingfast/substreams-sink"
-	db "github.com/streamingfast/substreams-sink-sql/db_changes/db"
 	sinker2 "github.com/streamingfast/substreams-sink-sql/db_changes/sinker"
 	"github.com/streamingfast/substreams/manifest"
 )
@@ -100,42 +98,24 @@ func sinkRunE(cmd *cobra.Command, args []string) error {
 	flushRetryCount := sflags.MustGetInt(cmd, "flush-retry-count")
 	flushRetryDelay := sflags.MustGetDuration(cmd, "flush-retry-delay")
 
-	dsn, err := db.ParseDSN(dsnString)
-	if err != nil {
-		return fmt.Errorf("parsing dsn: %w", err)
-	}
-
 	cursorTableName := sflags.MustGetString(cmd, "cursors-table")
 	historyTableName := sflags.MustGetString(cmd, "history-table")
-
 	handleReorgs := sflags.MustGetInt(cmd, "undo-buffer-size") != 0
-	dbLoader, err := db.NewLoader(
-		dsn,
-		cursorTableName,
-		historyTableName,
-		sflags.MustGetString(cmd, "clickhouse-cluster"),
-		batchBlockFlushInterval, batchRowFlushInterval, liveBlockFlushInterval,
-		sflags.MustGetString(cmd, onModuleHashMistmatchFlag),
-		&handleReorgs,
-		zlog, tracer,
-	)
-
-	if err != nil {
-		return fmt.Errorf("creating loader: %w", err)
-	}
-
-	if err := dbLoader.LoadTables(dsn.Schema(), cursorTableName, historyTableName); err != nil {
-		var e *db.SystemTableError
-		if errors.As(err, &e) {
-			fmt.Printf("Error validating the system table: %s\n", e)
-			fmt.Println("Did you run setup ?")
-			return e
-		}
-
-		return fmt.Errorf("load psql table: %w", err)
-	}
 
-	postgresSinker, err := sinker2.New(sink, dbLoader, zlog, tracer, flushRetryCount, flushRetryDelay)
+	sinkerFactory := sinker2.SinkerFactory(sink, sinker2.SinkerFactoryOptions{
+		CursorTableName:         cursorTableName,
+		HistoryTableName:        historyTableName,
+		ClickhouseCluster:       sflags.MustGetString(cmd, "clickhouse-cluster"),
+		BatchBlockFlushInterval: batchBlockFlushInterval,
+		BatchRowFlushInterval:   batchRowFlushInterval,
+		LiveBlockFlushInterval:  liveBlockFlushInterval,
+		OnModuleHashMismatch:    sflags.MustGetString(cmd, onModuleHashMistmatchFlag),
+		HandleReorgs:            handleReorgs,
+		FlushRetryCount:         flushRetryCount,
+		FlushRetryDelay:         flushRetryDelay,
+	})
+
+	postgresSinker, err := sinkerFactory(app.Context(), dsnString, zlog, tracer)
 	if err != nil {
 		return fmt.Errorf("unable to setup postgres sinker: %w", err)
 	}