diff --git a/CHANGELOG.md b/CHANGELOG.md index 1f9f0045..908674a7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,6 +33,7 @@ Group lines under `Added` / `Changed` / `Fixed` / `Removed`. Append a PR link ### Changed +- Driver insert-method defaults now fill only load requests that leave their method unset, preserving methods selected by workloads. ([#152](https://github.com/stroppy-io/stroppy/pull/152)) - SIGINT and SIGTERM now cancel the running workload and trigger graceful teardown; a second signal forces immediate exit. Exit status is 130 (SIGINT) or 143 (SIGTERM) after a graceful cancellation, 2 after a forced exit, and 1 for other errors. ([#148](https://github.com/stroppy-io/stroppy/pull/148)) - Built-in workloads expose their tuning options as typed parameters while preserving the existing environment-variable names. ([#128](https://github.com/stroppy-io/stroppy/pull/128)) - TPC-DS typed loads format common cell types directly into reusable buffers. ([#126](https://github.com/stroppy-io/stroppy/pull/126)) diff --git a/cmd/stroppy/commands/help/topic_drivers.go b/cmd/stroppy/commands/help/topic_drivers.go index 91773a7b..7f8c9055 100644 --- a/cmd/stroppy/commands/help/topic_drivers.go +++ b/cmd/stroppy/commands/help/topic_drivers.go @@ -95,6 +95,7 @@ DRIVER OPTIONS (-D / --driver-opt) authUser string Username for static credentials auth authPassword string Password for static credentials auth tlsInsecureSkipVerify bool Skip TLS cert verification (testing only) + defaultInsertMethod string Fallback for load requests without a method TLS is enabled automatically when the URL uses a secure scheme (e.g. grpcs:// for YDB). The options above are only needed when the server uses @@ -103,9 +104,8 @@ DRIVER OPTIONS (-D / --driver-opt) pool.* options are sugar — they map to the driver-specific pool config (pgx pool or sql pool) based on driverType. They are ignored for noop/csv. - "defaultInsertMethod" remains accepted in config files for compatibility, - but does not control workload execution. Insert methods are owned by each - workload's load request. + defaultInsertMethod is used only when a workload leaves its load request's + method unset. A workload-selected method always takes precedence. HOW IT WORKS diff --git a/cmd/stroppy/commands/run/run.go b/cmd/stroppy/commands/run/run.go index 518c7374..0aa5111e 100644 --- a/cmd/stroppy/commands/run/run.go +++ b/cmd/stroppy/commands/run/run.go @@ -18,6 +18,7 @@ import ( "github.com/stroppy-io/stroppy/pkg/bench" "github.com/stroppy-io/stroppy/pkg/common/logger" "github.com/stroppy-io/stroppy/pkg/config" + "github.com/stroppy-io/stroppy/pkg/driver" ) const ( @@ -507,8 +508,9 @@ func runGoWorkload( // No -d given: default to the local postgres preset (mirrors TS // declareDriverSetup defaults). drivers[0] = &config.DriverConfig{ //nolint:gosec // G101: URL field name, not an embedded credential - DriverType: config.DriverTypePostgres, - URL: "postgres://postgres:postgres@localhost:5432", + DriverType: config.DriverTypePostgres, + URL: "postgres://postgres:postgres@localhost:5432", + DefaultInsertMethod: "native", } } @@ -540,6 +542,8 @@ func buildDriverConfig(idx int, cfg *runner.DriverCLIConfig) (*config.DriverConf driverType := cfg.DriverType url := cfg.URL + defaultInsertMethod := cfg.DefaultInsertMethod + hasDefaultInsertMethod := cfg.HasDefaultInsertMethod() if overrides != nil { if overrides.DriverType != nil { @@ -549,6 +553,11 @@ func buildDriverConfig(idx int, cfg *runner.DriverCLIConfig) (*config.DriverConf if overrides.URL != nil { url = overrides.GetURL() } + + if overrides.DefaultInsertMethod != nil { + defaultInsertMethod = overrides.GetDefaultInsertMethod() + hasDefaultInsertMethod = true + } } dc := &config.DriverConfig{URL: url} @@ -562,7 +571,15 @@ func buildDriverConfig(idx int, cfg *runner.DriverCLIConfig) (*config.DriverConf dc.DriverType = t } - // defaultInsertMethod is owned by each Go workload's InsertRequest. + if hasDefaultInsertMethod { + method, err := driver.ResolveInsertMethod(dc.DriverType, defaultInsertMethod) + if err != nil { + return nil, invalidConfig(fmt.Errorf("driver %d: %w", idx, err)) + } + + dc.DefaultInsertMethod = method.String() + } + if err := applyDriverExtras(idx, dc, cfg.Extra); err != nil { return nil, invalidConfig(err) } diff --git a/cmd/stroppy/commands/run/run_test.go b/cmd/stroppy/commands/run/run_test.go index 201559f4..ff7df219 100644 --- a/cmd/stroppy/commands/run/run_test.go +++ b/cmd/stroppy/commands/run/run_test.go @@ -1171,6 +1171,82 @@ func TestApplyDriverPresetAliasesReachRuntimeConfig(t *testing.T) { } } +func TestApplyDriverOptInsertMethodAliasesRespectPrecedenceAndIndex(t *testing.T) { + postgres := "postgres" + plainBulk := "plain_bulk" + + for _, key := range []string{ + "defaultInsertMethod", + "default_insert_method", + "insertMethod", + "insert_method", + } { + t.Run(key, func(t *testing.T) { + configs, err := runner.DriverCLIConfigsFromFile(map[uint32]*config.DriverRunConfig{ + 0: {DriverType: &postgres, DefaultInsertMethod: &plainBulk}, + 1: {DriverType: &postgres, DefaultInsertMethod: &plainBulk}, + }) + if err != nil { + t.Fatalf("DriverCLIConfigsFromFile() error = %v", err) + } + + for _, idx := range []int{0, 1} { + if err := applyDriverPreset(configs, idx, "pg"); err != nil { + t.Fatalf("applyDriverPreset(%d) error = %v", idx, err) + } + } + + if err := applyDriverOpt(configs, 1, key, "columnar"); err != nil { + t.Fatalf("applyDriverOpt(%q) error = %v", key, err) + } + + first, err := buildDriverConfig(0, configs[0]) + if err != nil { + t.Fatalf("buildDriverConfig(0) error = %v", err) + } + + second, err := buildDriverConfig(1, configs[1]) + if err != nil { + t.Fatalf("buildDriverConfig(1) error = %v", err) + } + + if first.DefaultInsertMethod != "native" || second.DefaultInsertMethod != "columnar" { + t.Fatalf( + "defaults = (%q, %q), want preset then indexed CLI override", + first.DefaultInsertMethod, + second.DefaultInsertMethod, + ) + } + }) + } +} + +func TestApplyDriverOptInsertMethodAliasConflictIsOrderIndependent(t *testing.T) { + for _, alias := range []string{"default_insert_method", "insertMethod", "insert_method"} { + t.Run(alias, func(t *testing.T) { + first := runner.DriverCLIConfigs{} + if err := applyDriverOpt(first, 0, "defaultInsertMethod", "native"); err != nil { + t.Fatal(err) + } + + firstErr := applyDriverOpt(first, 0, alias, "columnar") + if firstErr == nil { + t.Fatal("second insert method override succeeded") + } + + second := runner.DriverCLIConfigs{} + if err := applyDriverOpt(second, 0, alias, "columnar"); err != nil { + t.Fatal(err) + } + + secondErr := applyDriverOpt(second, 0, "defaultInsertMethod", "native") + if secondErr == nil || secondErr.Error() != firstErr.Error() { + t.Fatalf("reverse collision error = %v, want %v", secondErr, firstErr) + } + }) + } +} + func TestDriverExtrasRejectAliasCollisionsAndWrongCase(t *testing.T) { tests := []struct { name string @@ -1400,6 +1476,165 @@ func TestRemovedIsolationRejectedAtDriverCLISurfaces(t *testing.T) { } } +func TestBuildDriverConfigEmptyDefaultInsertMethod(t *testing.T) { + postgres := "postgres" + empty := "" + + build := func(t *testing.T, cfg *runner.DriverCLIConfig, want string) { + t.Helper() + + got, err := buildDriverConfig(0, cfg) + if err != nil { + t.Fatalf("buildDriverConfig() error = %v", err) + } + + if got.DefaultInsertMethod != want { + t.Fatalf("DefaultInsertMethod = %q, want %q", got.DefaultInsertMethod, want) + } + } + + for _, key := range []string{ + "defaultInsertMethod", + "default_insert_method", + "insertMethod", + "insert_method", + } { + t.Run("driver option/"+key, func(t *testing.T) { + cfg := &runner.DriverCLIConfig{DriverType: postgres} + if err := cfg.ApplyOverride(key, empty); err != nil { + t.Fatalf("ApplyOverride(%q) error = %v", key, err) + } + + build(t, cfg, "plain_query") + }) + } + + t.Run("raw driver JSON", func(t *testing.T) { + cfg, err := runner.NewDriverCLIConfigFromJSON(`{"driverType":"postgres","defaultInsertMethod":""}`) + if err != nil { + t.Fatalf("NewDriverCLIConfigFromJSON() error = %v", err) + } + + build(t, &cfg, "plain_query") + }) + + t.Run("config file", func(t *testing.T) { + configs, err := runner.DriverCLIConfigsFromFile(map[uint32]*config.DriverRunConfig{ + 0: {DriverType: &postgres, DefaultInsertMethod: &empty}, + }) + if err != nil { + t.Fatalf("DriverCLIConfigsFromFile() error = %v", err) + } + + build(t, configs[0], "plain_query") + }) + + t.Run("absent", func(t *testing.T) { + build(t, &runner.DriverCLIConfig{DriverType: postgres}, "") + + configs, err := runner.DriverCLIConfigsFromFile(map[uint32]*config.DriverRunConfig{ + 0: {DriverType: &postgres}, + }) + if err != nil { + t.Fatalf("DriverCLIConfigsFromFile() error = %v", err) + } + + build(t, configs[0], "") + }) +} + +func TestBuildDriverConfigDefaultInsertMethod(t *testing.T) { + for _, test := range []struct { + name string + cfg *runner.DriverCLIConfig + want string + wantErr bool + }{ + { + name: "winning supported default", + cfg: &runner.DriverCLIConfig{ + DriverType: "postgres", DefaultInsertMethod: "columnar", + }, + want: "columnar", + }, + { + name: "winning unsupported default", + cfg: &runner.DriverCLIConfig{ + DriverType: "mysql", DefaultInsertMethod: "columnar", + }, + wantErr: true, + }, + } { + t.Run(test.name, func(t *testing.T) { + got, err := buildDriverConfig(0, test.cfg) + if test.wantErr { + if err == nil { + t.Fatal("buildDriverConfig() succeeded") + } + + return + } + + if err != nil { + t.Fatalf("buildDriverConfig() error = %v", err) + } + + if got.DefaultInsertMethod != test.want { + t.Fatalf("DefaultInsertMethod = %q, want %q", got.DefaultInsertMethod, test.want) + } + }) + } +} + +func TestDriverDefaultInputPrecedenceAndIndices(t *testing.T) { + mysql := "mysql" + unsupported := "columnar" + + configs, err := runner.DriverCLIConfigsFromFile(map[uint32]*config.DriverRunConfig{ + 0: {DriverType: &mysql, DefaultInsertMethod: &unsupported}, + }) + if err != nil { + t.Fatalf("DriverCLIConfigsFromFile() error = %v", err) + } + + if err := applyDriverPreset(configs, 0, "pg"); err != nil { + t.Fatalf("applyDriverPreset() error = %v", err) + } + + if err := applyDriverPreset(configs, 1, "mysql"); err != nil { + t.Fatalf("applyDriverPreset() error = %v", err) + } + + if err := applyDriverOpt(configs, 1, "defaultInsertMethod", "native"); err != nil { + t.Fatalf("applyDriverOpt() error = %v", err) + } + + first, err := buildDriverConfig(0, configs[0]) + if err != nil { + t.Fatalf("buildDriverConfig(0) error = %v", err) + } + + second, err := buildDriverConfig(1, configs[1]) + if err != nil { + t.Fatalf("buildDriverConfig(1) error = %v", err) + } + + if first.DefaultInsertMethod != "native" || second.DefaultInsertMethod != "native" { + t.Fatalf("defaults = (%q, %q), want native per driver", first.DefaultInsertMethod, second.DefaultInsertMethod) + } + + if first.DriverType != config.DriverTypePostgres || second.DriverType != config.DriverTypeMySQL { + t.Fatalf("driver types = (%s, %s), want postgres, mysql", first.DriverType, second.DriverType) + } +} + +func TestGlobalInsertMethodFlagIsRejected(t *testing.T) { + err := Cmd.RunE(Cmd, []string{"simple", "--insert-method", "native"}) + if err == nil || !contains(err.Error(), `unknown CLI parameter "insert-method"`) { + t.Fatalf("RunE() error = %v, want unknown CLI parameter", err) + } +} + func marshalDriverConfig(t *testing.T, cfg *runner.DriverCLIConfig) map[string]any { t.Helper() diff --git a/internal/runner/driver_preset.go b/internal/runner/driver_preset.go index 1fe29091..48f7872e 100644 --- a/internal/runner/driver_preset.go +++ b/internal/runner/driver_preset.go @@ -8,9 +8,11 @@ import ( "maps" "net/url" "path/filepath" + "regexp" "strings" "github.com/stroppy-io/stroppy/pkg/config" + "github.com/stroppy-io/stroppy/pkg/driver" ) // Driver preset literals reused across the postgres-family presets and @@ -21,10 +23,14 @@ const ( ) var ( - errUnknownDriver = errors.New("unknown driver") - errInvalidDriverOverride = errors.New("invalid driver override") - errDriverOverrideConflict = errors.New("driver override conflicts with existing non-object value") - errNilDriverConfig = errors.New("nil driver config") + errUnknownDriver = errors.New("unknown driver") + errInvalidDriverOverride = errors.New("invalid driver override") + errDriverOverrideConflict = errors.New("driver override conflicts with existing non-object value") + errInsertMethodAliasConflict = errors.New("insert method aliases conflict") + errRawInsertMethodNotString = errors.New("raw insert method must be a string") + errNilDriverConfig = errors.New("nil driver config") + + legacyRawInsertMethodKey = regexp.MustCompile(`([,{]\s*)"insertMethod"(\s*:)`) ) // DriverPreset contains default configuration for a known database driver. @@ -95,6 +101,10 @@ func LookupDriverPreset(name string) (DriverPreset, error) { return DriverPreset{}, fmt.Errorf("%w %q (available: %s)", errUnknownDriver, name, strings.Join(known, ", ")) } + if err := validateDefaultInsertMethod(preset.DefaultInsertMethod); err != nil { + return DriverPreset{}, err + } + return preset, nil } @@ -106,6 +116,8 @@ type DriverCLIConfig struct { URL string `json:"url,omitempty"` DefaultInsertMethod string `json:"defaultInsertMethod,omitempty"` + defaultInsertMethodSet bool + // Extra fields from config-file drivers that don't map to known fields. Extra map[string]any `json:"-"` @@ -127,7 +139,7 @@ func (d DriverCLIConfig) MarshalJSON() ([]byte, error) { merged["url"] = d.URL } - if d.DefaultInsertMethod != "" { + if d.HasDefaultInsertMethod() { merged["defaultInsertMethod"] = d.DefaultInsertMethod } @@ -136,6 +148,11 @@ func (d DriverCLIConfig) MarshalJSON() ([]byte, error) { return json.Marshal(merged) } +// HasDefaultInsertMethod reports whether a default was explicitly provided. +func (d *DriverCLIConfig) HasDefaultInsertMethod() bool { + return d.defaultInsertMethodSet || d.DefaultInsertMethod != "" +} + // DriverOverride is one -D key=value occurrence. type DriverOverride struct { Key string @@ -153,13 +170,24 @@ func (d *DriverCLIConfig) ApplyOverride(key, value string) error { return err } + key = canonicalDriverOverrideKey(key) + switch key { case "driverType", "driver_type": d.DriverType = value case "url": d.URL = value - case "defaultInsertMethod", "default_insert_method": + case "defaultInsertMethod": + if _, err := driver.ParseInsertMethod(value); err != nil { + return fmt.Errorf("%w: %w", errInvalidDriverOverride, err) + } + + if d.hasInsertMethodOverride() { + return errInsertMethodAliasConflict + } + d.DefaultInsertMethod = value + d.defaultInsertMethodSet = true default: if err := d.setExtraPath(path, driverOverrideValue(value)); err != nil { return err @@ -171,6 +199,32 @@ func (d *DriverCLIConfig) ApplyOverride(key, value string) error { return nil } +func canonicalDriverOverrideKey(key string) string { + switch key { + case "default_insert_method", "insertMethod", "insert_method": + return "defaultInsertMethod" + default: + return key + } +} + +func (d *DriverCLIConfig) hasInsertMethodOverride() bool { + for _, override := range d.Overrides { + switch override.Key { + case "insertMethod", "insert_method", "defaultInsertMethod", "default_insert_method": + return true + } + } + + return false +} + +func validateDefaultInsertMethod(method string) error { + _, err := driver.ParseInsertMethod(method) + + return err +} + func validateOverridePath(path []string) error { for _, part := range path { if part == "" { @@ -239,7 +293,8 @@ func driverOverrideValue(value string) any { func isDriverCLIField(key string) bool { switch key { - case "driverType", "driver_type", "url", "defaultInsertMethod", "default_insert_method": + case "driverType", "driver_type", "url", + "insertMethod", "insert_method", "defaultInsertMethod", "default_insert_method": return true default: return false @@ -405,33 +460,79 @@ func resolveDriverConfigPaths(fileConfig *config.DriverRunConfig) { // NewDriverCLIConfigFromPreset creates a DriverCLIConfig from a preset. func NewDriverCLIConfigFromPreset(p DriverPreset) DriverCLIConfig { return DriverCLIConfig{ - DriverType: p.DriverType, - URL: p.URL, - DefaultInsertMethod: p.DefaultInsertMethod, + DriverType: p.DriverType, + URL: p.URL, + DefaultInsertMethod: p.DefaultInsertMethod, + defaultInsertMethodSet: p.DefaultInsertMethod != "", } } // NewDriverCLIConfigFromJSON strictly validates a raw -d JSON object before // separating its base fields from the nested driver extras. func NewDriverCLIConfigFromJSON(raw string) (DriverCLIConfig, error) { + normalized, err := normalizeRawDriverJSON(raw) + if err != nil { + return DriverCLIConfig{}, fmt.Errorf("invalid driver JSON: %w", err) + } + fileConfig := &config.DriverRunConfig{} - if err := config.Unmarshal([]byte(raw), fileConfig); err != nil { + if err := config.Unmarshal([]byte(normalized), fileConfig); err != nil { return DriverCLIConfig{}, fmt.Errorf("invalid driver JSON: %w", err) } return driverCLIConfigFromFile(fileConfig) } +func normalizeRawDriverJSON(raw string) (string, error) { + if !json.Valid([]byte(raw)) || !strings.HasPrefix(strings.TrimSpace(raw), "{") { + return raw, nil + } + + var fields map[string]json.RawMessage + if err := json.Unmarshal([]byte(raw), &fields); err != nil { + return raw, fmt.Errorf("%w: %w", errInvalidDriverOverride, err) + } + + for _, key := range []string{"insertMethod", "defaultInsertMethod", "default_insert_method"} { + value, ok := fields[key] + if !ok { + continue + } + + if len(value) == 0 || value[0] != '"' { + return "", errRawInsertMethodNotString + } + + var method string + if err := json.Unmarshal(value, &method); err != nil { + return "", errRawInsertMethodNotString + } + + if err := validateDefaultInsertMethod(method); err != nil { + return "", err + } + } + + return legacyRawInsertMethodKey.ReplaceAllString(raw, `${1}"defaultInsertMethod"${2}`), nil +} + func driverCLIConfigFromFile(fileConfig *config.DriverRunConfig) (DriverCLIConfig, error) { if fileConfig == nil { return DriverCLIConfig{}, errNilDriverConfig } + if fileConfig.DefaultInsertMethod != nil { + if err := validateDefaultInsertMethod(fileConfig.GetDefaultInsertMethod()); err != nil { + return DriverCLIConfig{}, err + } + } + extraConfig := *fileConfig cfg := DriverCLIConfig{ - DriverType: fileConfig.GetDriverType(), - URL: fileConfig.GetURL(), - DefaultInsertMethod: fileConfig.GetDefaultInsertMethod(), + DriverType: fileConfig.GetDriverType(), + URL: fileConfig.GetURL(), + DefaultInsertMethod: fileConfig.GetDefaultInsertMethod(), + defaultInsertMethodSet: fileConfig.DefaultInsertMethod != nil, } extraConfig.DriverType = nil diff --git a/internal/runner/driver_preset_test.go b/internal/runner/driver_preset_test.go index f5eb9ba0..35bcc7c2 100644 --- a/internal/runner/driver_preset_test.go +++ b/internal/runner/driver_preset_test.go @@ -7,6 +7,8 @@ import ( "github.com/stretchr/testify/require" "github.com/stroppy-io/stroppy/internal/runner" + "github.com/stroppy-io/stroppy/pkg/config" + "github.com/stroppy-io/stroppy/pkg/driver" ) func TestNewDriverCLIConfigFromJSONStrictCompatibility(t *testing.T) { @@ -104,6 +106,41 @@ func TestDriverCLIConfigDecodeOverridesPreservesLexemes(t *testing.T) { } } +func TestDriverCLIConfigCanonicalizesInsertMethodOverrides(t *testing.T) { + for _, key := range []string{ + "defaultInsertMethod", + "default_insert_method", + "insertMethod", + "insert_method", + } { + t.Run(key, func(t *testing.T) { + cfg := &runner.DriverCLIConfig{} + require.NoError(t, cfg.ApplyOverride(key, "columnar")) + require.Equal(t, []runner.DriverOverride{{Key: "defaultInsertMethod", Value: "columnar"}}, cfg.Overrides) + + overrides, err := cfg.DecodeOverrides() + require.NoError(t, err) + require.Equal(t, "columnar", overrides.GetDefaultInsertMethod()) + }) + } +} + +func TestDriverCLIConfigInsertMethodAliasConflictIsOrderIndependent(t *testing.T) { + for _, alias := range []string{"default_insert_method", "insertMethod", "insert_method"} { + t.Run(alias, func(t *testing.T) { + first := &runner.DriverCLIConfig{} + require.NoError(t, first.ApplyOverride("defaultInsertMethod", "native")) + firstErr := first.ApplyOverride(alias, "columnar") + require.Error(t, firstErr) + + second := &runner.DriverCLIConfig{} + require.NoError(t, second.ApplyOverride(alias, "columnar")) + secondErr := second.ApplyOverride("defaultInsertMethod", "native") + require.EqualError(t, secondErr, firstErr.Error()) + }) + } +} + func TestDriverCLIConfigDecodeOverridesRejectsRemovedIsolation(t *testing.T) { for _, key := range []string{"defaultTxIsolation", "default_tx_isolation"} { t.Run(key, func(t *testing.T) { @@ -132,3 +169,48 @@ func TestNewDriverCLIConfigAliasCollisionOrderIsDeterministic(t *testing.T) { require.Error(t, first) require.EqualError(t, second, first.Error()) } + +func TestDefaultInsertMethodInputValidation(t *testing.T) { + preset, err := runner.LookupDriverPreset("pg") + require.NoError(t, err) + require.Equal(t, "native", runner.NewDriverCLIConfigFromPreset(preset).DefaultInsertMethod) + + for _, doc := range []string{ + `{"driverType":"mysql","defaultInsertMethod":"columnar"}`, + `{"driverType":"postgres","insertMethod":"native"}`, + } { + cfg, err := runner.NewDriverCLIConfigFromJSON(doc) + require.NoError(t, err) + require.NotEmpty(t, cfg.DefaultInsertMethod) + } + + for _, doc := range []string{ + `{"defaultInsertMethod":true}`, + `{"defaultInsertMethod":null}`, + `{"insertMethod":1}`, + `{"defaultInsertMethod":"bogus"}`, + } { + _, err := runner.NewDriverCLIConfigFromJSON(doc) + require.Error(t, err) + } + + _, firstErr := runner.NewDriverCLIConfigFromJSON(`{"insertMethod":"native","defaultInsertMethod":"plain_bulk"}`) + require.Error(t, firstErr) + + _, secondErr := runner.NewDriverCLIConfigFromJSON(`{"defaultInsertMethod":"plain_bulk","insertMethod":"native"}`) + require.EqualError(t, secondErr, firstErr.Error()) + + invalid := "bogus" + _, err = runner.DriverCLIConfigsFromFile(map[uint32]*config.DriverRunConfig{ + 0: {DefaultInsertMethod: &invalid}, + }) + require.ErrorIs(t, err, driver.ErrUnknownInsertMethod) + + cli := &runner.DriverCLIConfig{} + require.NoError(t, cli.ApplyOverride("insertMethod", "native")) + require.ErrorIs(t, cli.ApplyOverride("defaultInsertMethod", "bogus"), driver.ErrUnknownInsertMethod) + + cli = &runner.DriverCLIConfig{} + require.NoError(t, cli.ApplyOverride("defaultInsertMethod", "native")) + require.Error(t, cli.ApplyOverride("insertMethod", "plain_bulk")) +} diff --git a/pkg/bench/insert_method_test.go b/pkg/bench/insert_method_test.go new file mode 100644 index 00000000..e28d4e92 --- /dev/null +++ b/pkg/bench/insert_method_test.go @@ -0,0 +1,175 @@ +package bench + +import ( + "context" + "errors" + "testing" + + "go.uber.org/zap" + + "github.com/stroppy-io/stroppy/pkg/config" + "github.com/stroppy-io/stroppy/pkg/driver" + "github.com/stroppy-io/stroppy/pkg/driver/insertprogress" + "github.com/stroppy-io/stroppy/pkg/driver/stats" + "github.com/stroppy-io/stroppy/pkg/gen" +) + +func validInsertSource() *gen.IndexedSource { + b := gen.NewSchemaBuilder() + id := b.Int64("id") + + return gen.NewIndexedSource(b.Build(), gen.Root{}, "test/insert-method@1", 1, 1, + func(r gen.Row, entity uint64) error { + r.SetInt64(id, int64(entity)) + + return nil + }) +} + +type recordingDriver struct { + calls int + method driver.InsertMethod + tracker *insertprogress.Tracker +} + +func (d *recordingDriver) Insert(ctx context.Context, req *driver.InsertRequest) (*stats.Query, error) { + d.calls++ + d.method = req.Method + d.tracker = insertprogress.FromContext(ctx) + + return &stats.Query{Rows: 1}, nil +} + +func (d *recordingDriver) RunQuery(context.Context, string, map[string]any) (*driver.QueryResult, error) { + return nil, errors.New("unexpected RunQuery") +} + +func (d *recordingDriver) Begin(context.Context, config.TxIsolationLevel) (driver.Tx, error) { + return nil, errors.New("unexpected Begin") +} + +func (d *recordingDriver) ClassifyError(error) driver.ErrorFacts { return driver.ErrorFacts{} } + +func (d *recordingDriver) Teardown(context.Context) error { return nil } + +func TestInsertUsesDriverFallbackWithoutMutatingRequest(t *testing.T) { + installRuntimeTestRoot(t) + + drv := &recordingDriver{} + b := &Bench{ + root: root, + vu: &VU{root: root, vuid: 1, ctx: context.Background()}, + lg: zap.NewNop(), + drv: drv, + cfg: &config.DriverConfig{ + DriverType: config.DriverTypePostgres, + DefaultInsertMethod: "plain_query", + }, + } + req := &driver.InsertRequest{Table: "t", Workers: 1, Source: validInsertSource()} + + if _, err := b.Insert(context.Background(), req); err != nil { + t.Fatalf("Insert() error = %v", err) + } + + if drv.method != driver.InsertPlainQuery { + t.Fatalf("driver method = %v, want plain_query", drv.method) + } + + if req.Method != 0 { + t.Fatalf("request method = %v, want zero", req.Method) + } + + if drv.tracker == nil { + t.Fatal("driver did not receive progress tracker") + } + + if snapshot := drv.tracker.Finish(nil); snapshot.Method != "plain_query" { + t.Fatalf("progress method = %q, want plain_query", snapshot.Method) + } +} + +func TestInsertWorkloadMethodOverridesDriverFallback(t *testing.T) { + installRuntimeTestRoot(t) + + drv := &recordingDriver{} + b := &Bench{ + root: root, + vu: &VU{root: root, vuid: 1, ctx: context.Background()}, + lg: zap.NewNop(), + drv: drv, + cfg: &config.DriverConfig{ + DriverType: config.DriverTypePostgres, + DefaultInsertMethod: "native", + }, + } + req := &driver.InsertRequest{ + Table: "t", Method: driver.InsertPlainQuery, Workers: 1, Source: validInsertSource(), + } + + if _, err := b.Insert(context.Background(), req); err != nil { + t.Fatalf("Insert() error = %v", err) + } + + if drv.method != driver.InsertPlainQuery { + t.Fatalf("driver method = %v, want plain_query", drv.method) + } + + if req.Method != driver.InsertPlainQuery { + t.Fatalf("request method = %v, want plain_query", req.Method) + } +} + +func TestInsertRejectsMissingOrUnsupportedEffectiveMethod(t *testing.T) { + tests := []struct { + name string + cfg *config.DriverConfig + req *driver.InsertRequest + err error + }{ + { + name: "no default", + cfg: &config.DriverConfig{DriverType: config.DriverTypePostgres}, + req: &driver.InsertRequest{Table: "t", Workers: 1, Source: validInsertSource()}, + err: driver.ErrInsertMethodUnsupported, + }, + { + name: "unsupported workload override", + cfg: &config.DriverConfig{DriverType: config.DriverTypeMySQL, DefaultInsertMethod: "plain_bulk"}, + req: &driver.InsertRequest{ + Table: "t", Method: driver.InsertColumnar, Workers: 1, Source: validInsertSource(), + }, + err: driver.ErrInsertMethodUnsupported, + }, + { + name: "shape before method resolution", + cfg: &config.DriverConfig{DriverType: config.DriverTypeCSV}, + req: &driver.InsertRequest{Table: "t", Workers: 1}, + err: driver.ErrNilInsertSource, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + installRuntimeTestRoot(t) + + drv := &recordingDriver{} + b := &Bench{ + root: root, + vu: &VU{root: root, vuid: 1, ctx: context.Background()}, + lg: zap.NewNop(), + drv: drv, + cfg: tc.cfg, + } + + _, err := b.Insert(context.Background(), tc.req) + if !errors.Is(err, tc.err) { + t.Fatalf("Insert() error = %v, want %v", err, tc.err) + } + + if drv.calls != 0 { + t.Fatalf("driver calls = %d, want 0", drv.calls) + } + }) + } +} diff --git a/pkg/bench/query.go b/pkg/bench/query.go index b6efe1e9..093e2073 100644 --- a/pkg/bench/query.go +++ b/pkg/bench/query.go @@ -144,7 +144,22 @@ func (b *Bench) Insert(ctx context.Context, req *driver.InsertRequest) (*stats.Q return nil, fmt.Errorf("insert: %w", err) } - tracker := b.newBatchInsertTracker(req) + effectiveReq := *req + if effectiveReq.Method == 0 && b.cfg.GetDefaultInsertMethod() != "" { + method, err := driver.ResolveInsertMethod(b.cfg.DriverType, b.cfg.GetDefaultInsertMethod()) + if err != nil { + return nil, fmt.Errorf("insert: %w", err) + } + + effectiveReq.Method = method + } + + if !driver.SupportsInsertMethod(b.cfg.DriverType, effectiveReq.Method) { + return nil, fmt.Errorf("insert: %w %q (%s driver)", + driver.ErrInsertMethodUnsupported, effectiveReq.Method.String(), b.cfg.DriverType) + } + + tracker := b.newBatchInsertTracker(&effectiveReq) runCtx := ctx if tracker.Enabled() { @@ -152,7 +167,7 @@ func (b *Bench) Insert(ctx context.Context, req *driver.InsertRequest) (*stats.Q tracker.Start(runCtx) } - result, err := b.drv.Insert(runCtx, req) + result, err := b.drv.Insert(runCtx, &effectiveReq) if tracker.Enabled() { tracker.Finish(err) } @@ -162,13 +177,13 @@ func (b *Bench) Insert(ctx context.Context, req *driver.InsertRequest) (*stats.Q elapsed = result.Elapsed } - b.root.txMetrics.recordInsertResult(b.vu, req.Table, elapsed, err) + b.root.txMetrics.recordInsertResult(b.vu, effectiveReq.Table, elapsed, err) if err != nil { - return nil, fmt.Errorf("insert %q: %w", req.Table, err) + return nil, fmt.Errorf("insert %q: %w", effectiveReq.Table, err) } - b.root.txMetrics.recordInsert(b.vu, req.Table, result.Rows) + b.root.txMetrics.recordInsert(b.vu, effectiveReq.Table, result.Rows) return result, nil } diff --git a/pkg/config/config.go b/pkg/config/config.go index 9e41ceb6..6c879903 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -475,8 +475,11 @@ func (c *PoolConfig) GetConnMaxIdleTime() string { // DriverConfig is the runtime driver configuration consumed by the drivers and // bench engine. It is assembled internally, never JSON-decoded directly. type DriverConfig struct { - URL string `json:"url,omitempty"` - DriverType DriverType `json:"driverType,omitempty"` + URL string `json:"url,omitempty"` + DriverType DriverType `json:"driverType,omitempty"` + // DefaultInsertMethod is the driver fallback when a workload leaves + // InsertRequest.Method unset. + DefaultInsertMethod string `json:"defaultInsertMethod,omitempty"` BulkSize *int32 `json:"bulkSize,omitempty"` ErrorMode ErrorMode `json:"errorMode,omitempty"` Postgres *PostgresConfig `json:"postgres,omitempty"` @@ -489,6 +492,14 @@ type DriverConfig struct { InsertProgress *InsertProgressConfig `json:"insertProgress,omitempty"` } +func (c *DriverConfig) GetDefaultInsertMethod() string { + if c != nil { + return c.DefaultInsertMethod + } + + return "" +} + func (c *DriverConfig) GetBulkSize() int32 { if c != nil && c.BulkSize != nil { return *c.BulkSize diff --git a/pkg/driver/insert_methods.go b/pkg/driver/insert_methods.go index f036a880..e1635de2 100644 --- a/pkg/driver/insert_methods.go +++ b/pkg/driver/insert_methods.go @@ -1,6 +1,8 @@ package driver import ( + "errors" + "fmt" "slices" "github.com/stroppy-io/stroppy/pkg/config" @@ -76,3 +78,28 @@ func InsertCapabilities() []InsertCapability { return capabilities } + +// ErrInsertMethodUnsupported is returned when a resolved insert method is not +// served by the selected driver. +var ErrInsertMethodUnsupported = errors.New("insert method not supported by driver") + +// SupportsInsertMethod reports whether driverType's implementation serves method. +func SupportsInsertMethod(driverType config.DriverType, method InsertMethod) bool { + return slices.Contains(insertMethodsByDriver[driverType], method) +} + +// ResolveInsertMethod parses an authoring string and rejects methods the +// selected driver does not serve, so an invalid or unsupported value fails +// before a run starts loading. +func ResolveInsertMethod(driverType config.DriverType, s string) (InsertMethod, error) { + method, err := ParseInsertMethod(s) + if err != nil { + return 0, err + } + + if !SupportsInsertMethod(driverType, method) { + return 0, fmt.Errorf("%w %q (%s driver)", ErrInsertMethodUnsupported, s, driverType) + } + + return method, nil +} diff --git a/pkg/driver/insert_methods_test.go b/pkg/driver/insert_methods_test.go index 3da981b9..004cea3b 100644 --- a/pkg/driver/insert_methods_test.go +++ b/pkg/driver/insert_methods_test.go @@ -1,7 +1,9 @@ package driver import ( + "errors" "slices" + "strings" "testing" "github.com/stroppy-io/stroppy/pkg/config" @@ -67,3 +69,81 @@ func TestInsertCapabilitiesDeterministic(t *testing.T) { t.Errorf("capabilities not ordered by driver enum value: %v", types) } } + +func TestResolveInsertMethod(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + driverType config.DriverType + method string + want InsertMethod + wantErr error + wantErrMsg string + }{ + {name: "native postgres", driverType: config.DriverTypePostgres, method: "native", want: InsertNative}, + {name: "columnar postgres", driverType: config.DriverTypePostgres, method: "columnar", want: InsertColumnar}, + {name: "empty selects plain_query", driverType: config.DriverTypePostgres, method: "", want: InsertPlainQuery}, + {name: "invalid value", driverType: config.DriverTypePostgres, method: "bogus", wantErr: ErrUnknownInsertMethod}, + { + name: "columnar unsupported on mysql", + driverType: config.DriverTypeMySQL, + method: "columnar", + wantErrMsg: "not supported", + }, + { + name: "native unsupported on csv", + driverType: config.DriverTypeCSV, + method: "plain_bulk", + wantErrMsg: "not supported", + }, + {name: "native supported on csv", driverType: config.DriverTypeCSV, method: "native", want: InsertNative}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + got, err := ResolveInsertMethod(tc.driverType, tc.method) + if tc.wantErr != nil { + if !errors.Is(err, tc.wantErr) { + t.Fatalf("ResolveInsertMethod(%q) error = %v, want %v", tc.method, err, tc.wantErr) + } + + return + } + + if tc.wantErrMsg != "" { + if err == nil || !strings.Contains(err.Error(), tc.wantErrMsg) { + t.Fatalf("ResolveInsertMethod(%q) error = %v, want containing %q", tc.method, err, tc.wantErrMsg) + } + + return + } + + if err != nil { + t.Fatalf("ResolveInsertMethod(%q): %v", tc.method, err) + } + + if got != tc.want { + t.Fatalf("ResolveInsertMethod(%q) = %v, want %v", tc.method, got, tc.want) + } + }) + } +} + +func TestSupportsInsertMethod(t *testing.T) { + t.Parallel() + + if !SupportsInsertMethod(config.DriverTypePostgres, InsertColumnar) { + t.Fatal("postgres should support columnar") + } + + if SupportsInsertMethod(config.DriverTypeMySQL, InsertColumnar) { + t.Fatal("mysql should not support columnar") + } + + if SupportsInsertMethod(config.DriverTypeCSV, InsertPlainQuery) { + t.Fatal("csv should only support native") + } +}