Sink from-proto through a disk spool, and load without constraints

bin/test.sh

@@ -18,10 +18,15 @@ main() {
   set -e
 
   go test ./... "$@"
-  # commented while they don't work on github for now
+
+  # tests_e2e is its own Go module, so `go test ./...` above never reaches it. It stays
+  # out of this script until it passes on Linux: the node writes root-owned files into
+  # the bind-mounted t.TempDir, which fails every container test's cleanup on a CI
+  # runner. See the fix/tests-e2e-on-linux branch.
+  #
   # pushd tests_e2e &> /dev/null
   # go test ./... "$@"
-  # popd
+  # popd &> /dev/null
 }
 
 usage_error() {

cmd/substreams/init_test.go

@@ -3,9 +3,9 @@ package main
 import (
 	"testing"
 
+	pbconvo "github.com/streamingfast/substreams/pb/sf/codegen/conversation/v1"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	pbconvo "github.com/streamingfast/substreams/pb/sf/codegen/conversation/v1"
 )
 
 func TestProtocolOrderPreservation(t *testing.T) {

cmd/substreams/sink_clickhouse.go

@@ -17,19 +17,20 @@ var sinkClickhouseCmd = &cobra.Command{
 }
 
 var sinkClickhouseSetupCmd = &cobra.Command{
-	Use:   "setup <manifest>",
+	Use:   "setup <manifest> [<module>]",
 	Short: "Setup the required infrastructure to deploy a Substreams SQL deployable unit",
 	Long: cli.Dedent(`
 		Setup the database for the Substreams SQL sink, auto-detecting the mode from the
 		output module type, exactly like the run action:
 
-		- DatabaseChanges output: creates the system tables (cursors, history) and applies
-		  the 'schema.sql' bundled in the manifest sink config.
-		- Any other output type (from-proto): resolves the schema from the module's output
-		  proto and creates the database schema and tables, then exits. This step is
-		  idempotent and can be run again safely.
+		- Database Changes Mode ('DatabaseChanges' output): creates the system tables
+		  (cursors, history) and applies the 'schema.sql' bundled in the manifest sink
+		  config.
+		- Relational Mappings Mode (any other output type): resolves the schema from the
+		  module's output proto and creates the database schema and tables, then exits.
+		  This step is idempotent and can be run again safely.
 	`),
-	Args: cobra.ExactArgs(1),
+	Args: cobra.RangeArgs(1, 2),
 	RunE: newSinkSetupE(sinkClickhouseDriver),
 }
 
@@ -39,15 +40,18 @@ func init() {
 	addOperatorFlags(persistent)
 
 	addSinkRunFlags(sinkClickhouseCmd.Flags(), sinkClickhouseDriver)
+	setModeGroupedUsage(sinkClickhouseCmd)
 
 	setupFlags := sinkClickhouseSetupCmd.Flags()
 	addCursorTableFlags(setupFlags)
 	addClusterFlag(setupFlags)
 	addOnModuleHashMismatchFlag(setupFlags)
-	setupFlags.Bool("system-tables-only", false, "[DatabaseChanges mode] will only create/update the systems tables (cursors, substreams_history) and ignore the schema from the manifest")
-	setupFlags.Bool("ignore-duplicate-table-errors", false, "[DatabaseChanges mode][Dev] Use this if you want to ignore duplicate table errors, take caution that this means the 'schema.sql' file will not have run fully!")
+	setupFlags.Bool("system-tables-only", false, "will only create/update the systems tables (cursors, substreams_history) and ignore the schema from the manifest")
+	setupFlags.Bool("ignore-duplicate-table-errors", false, "[Dev] Use this if you want to ignore duplicate table errors, take caution that this means the 'schema.sql' file will not have run fully!")
 	addBytesEncodingFlag(setupFlags)
-	addFromProtoModeRunFlags(setupFlags, sinkClickhouseDriver)
+	addFromProtoSchemaFlags(setupFlags)
+	addConstraintTimingFlag(setupFlags)
+	addClickhouseStateFlags(setupFlags)
 
 	sinkClickhouseCmd.AddCommand(sinkClickhouseSetupCmd)
 	sinkClickhouseCmd.AddCommand(newSinkToolsCmd(sinkClickhouseDriver))

cmd/substreams/sink_postgres.go

@@ -17,39 +17,110 @@ var sinkPostgresCmd = &cobra.Command{
 }
 
 var sinkPostgresSetupCmd = &cobra.Command{
-	Use:   "setup <manifest>",
+	Use:   "setup <manifest> [<module>]",
 	Short: "Setup the required infrastructure to deploy a Substreams SQL deployable unit",
 	Long: cli.Dedent(`
 		Setup the database for the Substreams SQL sink, auto-detecting the mode from the
 		output module type, exactly like the run action:
 
-		- DatabaseChanges output: creates the system tables (cursors, history) and applies
-		  the 'schema.sql' bundled in the manifest sink config.
-		- Any other output type (from-proto): resolves the schema from the module's output
-		  proto and creates the database schema and tables, then exits. This step is
-		  idempotent and can be run again safely.
+		- Database Changes Mode ('DatabaseChanges' output): creates the system tables
+		  (cursors, history) and applies the 'schema.sql' bundled in the manifest sink
+		  config.
+		- Relational Mappings Mode (any other output type): resolves the schema from the
+		  module's output proto and creates the database schema and tables, then exits.
+		  This step is idempotent and can be run again safely.
 	`),
-	Args: cobra.ExactArgs(1),
+	Args: cobra.RangeArgs(1, 2),
 	RunE: newSinkSetupE(sinkPostgresDriver),
 }
 
+var sinkPostgresConstraintsCmd = &cobra.Command{
+	Use:   "constraints",
+	Short: "Create or drop the schema's constraints on an already loaded database",
+}
+
+var sinkPostgresConstraintsApplyCmd = &cobra.Command{
+	Use:   "apply <manifest> [<module>]",
+	Short: "Create the schema's constraints on an already loaded database",
+	Long: cli.Dedent(`
+		Create the primary keys, unique and foreign key constraints of a Relational
+		Mappings schema on a database the sink has already loaded, skipping the ones
+		already in place.
+
+		The sink loads without them on purpose: measured through binary COPY, loading with
+		foreign keys in place runs 27x slower than loading without, where building the very
+		same constraints afterwards costs 3.3x. Creating them is a stop-the-world
+		operation, though — every index is built and every foreign key validated, with the
+		tables locked while it runs — so on a large database this belongs in a maintenance
+		window, which is what --apply-constraints=manual leaves it to this command for.
+
+		The index on _block_number_ is not created here. The sink creates that one when it
+		starts, concurrently: this command is yours to schedule, and the reorg path cannot
+		wait for a maintenance window.
+
+		Running it again is safe: constraints already in place are left alone.
+
+		The module is inferred from the package when it is left out. A package with more
+		than one candidate has to be told which, or the schema this derives will not be the
+		one the run created.
+	`),
+	Args: cobra.RangeArgs(1, 2),
+	RunE: newSinkConstraintsE(sinkPostgresDriver, constraintsApply),
+}
+
+var sinkPostgresConstraintsDropCmd = &cobra.Command{
+	Use:   "drop <manifest> [<module>]",
+	Short: "Drop the schema's constraints",
+	Long: cli.Dedent(`
+		Drop the primary keys, unique and foreign key constraints of a Relational Mappings
+		schema, leaving anything the sink did not create alone — the index on _block_number_
+		included, that one being the sink's own and recreated when it next starts.
+
+		This is the escape hatch after --apply-constraints=always, and what makes a
+		backfill that has to be resumed fast again without setting the schema up afresh:
+		loading with foreign keys in place measured 27x slower than loading without them.
+
+		Running it again is safe: anything already absent is skipped.
+	`),
+	Args: cobra.RangeArgs(1, 2),
+	RunE: newSinkConstraintsE(sinkPostgresDriver, constraintsDrop),
+}
+
 func init() {
 	persistent := sinkPostgresCmd.PersistentFlags()
 	addDSNFlag(persistent)
 	addOperatorFlags(persistent)
 
 	addSinkRunFlags(sinkPostgresCmd.Flags(), sinkPostgresDriver)
+	setModeGroupedUsage(sinkPostgresCmd)
 
 	setupFlags := sinkPostgresSetupCmd.Flags()
 	addCursorTableFlags(setupFlags)
 	addOnModuleHashMismatchFlag(setupFlags)
-	setupFlags.Bool("postgraphile", false, "[DatabaseChanges mode] Will append the necessary 'comments' on cursors table to fully support postgraphile")
-	setupFlags.Bool("system-tables-only", false, "[DatabaseChanges mode] will only create/update the systems tables (cursors, substreams_history) and ignore the schema from the manifest")
-	setupFlags.Bool("ignore-duplicate-table-errors", false, "[DatabaseChanges mode][Dev] Use this if you want to ignore duplicate table errors, take caution that this means the 'schema.sql' file will not have run fully!")
+	setupFlags.Bool("postgraphile", false, "Will append the necessary 'comments' on cursors table to fully support postgraphile")
+	setupFlags.Bool("system-tables-only", false, "will only create/update the systems tables (cursors, substreams_history) and ignore the schema from the manifest")
+	setupFlags.Bool("ignore-duplicate-table-errors", false, "[Dev] Use this if you want to ignore duplicate table errors, take caution that this means the 'schema.sql' file will not have run fully!")
 	addBytesEncodingFlag(setupFlags)
-	addFromProtoModeRunFlags(setupFlags, sinkPostgresDriver)
+	addFromProtoSchemaFlags(setupFlags)
+	addConstraintTimingFlag(setupFlags)
+
+	applyFlags := sinkPostgresConstraintsApplyCmd.Flags()
+	addBytesEncodingFlag(applyFlags)
+	addFromProtoSchemaFlags(applyFlags)
+	addConstraintPassFlags(applyFlags)
+
+	// Drop always removes every constraint managed by the sink. The disable-* and
+	// --no-constraints flags describe what should be created, so exposing them here would
+	// suggest that drop honors a policy it cannot apply.
+	dropFlags := sinkPostgresConstraintsDropCmd.Flags()
+	addBytesEncodingFlag(dropFlags)
+	dropFlags.String("proto-file-override", "", "Override protobuf file to use instead of extracting from substreams package")
+	addConstraintPassFlags(dropFlags)
+	sinkPostgresConstraintsCmd.AddCommand(sinkPostgresConstraintsApplyCmd)
+	sinkPostgresConstraintsCmd.AddCommand(sinkPostgresConstraintsDropCmd)
 
 	sinkPostgresCmd.AddCommand(sinkPostgresSetupCmd)
+	sinkPostgresCmd.AddCommand(sinkPostgresConstraintsCmd)
 	sinkPostgresCmd.AddCommand(newSinkToolsCmd(sinkPostgresDriver))
 
 	SinkCmd.AddCommand(sinkPostgresCmd)

cmd/substreams/sink_postgres_generate_csv.go

@@ -104,7 +104,7 @@ func sinkPostgresGenerateCSVE(cmd *cobra.Command, args []string) error {
 		supportedOutputTypes,
 		manifestPath,
 		outputModule,
-		"sink_database_changes",
+		sinkUserAgent("sink_database_changes", sinkPostgresDriver),
 		zlog,
 		tracer,
 	)