diff --git a/runtime/drivers/clickhouse/model_manager.go b/runtime/drivers/clickhouse/model_manager.go index 0a1e167cf2e..efe0e34168c 100644 --- a/runtime/drivers/clickhouse/model_manager.go +++ b/runtime/drivers/clickhouse/model_manager.go @@ -180,6 +180,14 @@ func (c *Connection) validateAndApplyDefaults(opts *drivers.ModelExecuteOptions, ip.SQL = ip.SQL + " SETTINGS " + op.QuerySettings } + if ip != nil { + ip.PostExec = strings.TrimSpace(ip.PostExec) + ip.PreExec = strings.TrimSpace(ip.PreExec) + } + if op != nil { + op.PostExec = strings.TrimSpace(op.PostExec) + op.PreExec = strings.TrimSpace(op.PreExec) + } return nil } diff --git a/runtime/drivers/duckdb/model_executor_localfile_self.go b/runtime/drivers/duckdb/model_executor_localfile_self.go index af285fec8c7..f4ab52209bd 100644 --- a/runtime/drivers/duckdb/model_executor_localfile_self.go +++ b/runtime/drivers/duckdb/model_executor_localfile_self.go @@ -69,7 +69,7 @@ func (e *localFileToSelfExecutor) Execute(ctx context.Context, opts *drivers.Mod } warnings = append(warnings, fmt.Sprintf("Undefined fields %q in output properties. Will be ignored.", strings.Join(unused, ", "))) } - if err := outputProps.validateAndApplyDefaults(opts, &ModelInputProperties{}, outputProps); err != nil { + if err := outputProps.validateAndApplyDefaults(opts, &ModelInputProperties{}); err != nil { return nil, fmt.Errorf("invalid output properties: %w", err) } diff --git a/runtime/drivers/duckdb/model_executor_self.go b/runtime/drivers/duckdb/model_executor_self.go index a3158c2d70a..6aae90720d1 100644 --- a/runtime/drivers/duckdb/model_executor_self.go +++ b/runtime/drivers/duckdb/model_executor_self.go @@ -54,7 +54,7 @@ func (e *selfToSelfExecutor) Execute(ctx context.Context, opts *drivers.ModelExe } warnings = append(warnings, fmt.Sprintf("Undefined fields %q in input properties. Will be ignored.", strings.Join(unused, ", "))) } - if err := inputProps.Validate(); err != nil { + if err := inputProps.ValidateAndApplyDefaults(); err != nil { return nil, fmt.Errorf("invalid input properties: %w", err) } @@ -69,7 +69,7 @@ func (e *selfToSelfExecutor) Execute(ctx context.Context, opts *drivers.ModelExe } warnings = append(warnings, fmt.Sprintf("Undefined fields %q in output properties. Will be ignored.", strings.Join(unused, ", "))) } - if err := outputProps.validateAndApplyDefaults(opts, inputProps, outputProps); err != nil { + if err := outputProps.validateAndApplyDefaults(opts, inputProps); err != nil { return nil, fmt.Errorf("invalid output properties: %w", err) } diff --git a/runtime/drivers/duckdb/model_executor_self_file.go b/runtime/drivers/duckdb/model_executor_self_file.go index 7a655710f6b..100cd96bcb2 100644 --- a/runtime/drivers/duckdb/model_executor_self_file.go +++ b/runtime/drivers/duckdb/model_executor_self_file.go @@ -46,7 +46,7 @@ func (e *selfToFileExecutor) Execute(ctx context.Context, opts *drivers.ModelExe } warnings = append(warnings, fmt.Sprintf("Undefined fields %q in input properties. Will be ignored.", strings.Join(unused, ", "))) } - if err := inputProps.Validate(); err != nil { + if err := inputProps.ValidateAndApplyDefaults(); err != nil { return nil, fmt.Errorf("invalid input properties: %w", err) } diff --git a/runtime/drivers/duckdb/model_executor_self_objectstore.go b/runtime/drivers/duckdb/model_executor_self_objectstore.go index 4b25c1ea6c0..e005920375c 100644 --- a/runtime/drivers/duckdb/model_executor_self_objectstore.go +++ b/runtime/drivers/duckdb/model_executor_self_objectstore.go @@ -43,7 +43,7 @@ func (e *selfToObjectStoreExecutor) Execute(ctx context.Context, opts *drivers.M } warnings = append(warnings, fmt.Sprintf("Undefined fields %q in input properties. Will be ignored.", strings.Join(unused, ", "))) } - if err := inputProps.Validate(); err != nil { + if err := inputProps.ValidateAndApplyDefaults(); err != nil { return nil, fmt.Errorf("invalid input properties: %w", err) } diff --git a/runtime/drivers/duckdb/model_manager.go b/runtime/drivers/duckdb/model_manager.go index 6133c1e671b..93abf794f82 100644 --- a/runtime/drivers/duckdb/model_manager.go +++ b/runtime/drivers/duckdb/model_manager.go @@ -26,7 +26,10 @@ type ModelInputProperties struct { InternalDropSecretSQL string `mapstructure:"internal_drop_secret_sql"` } -func (p *ModelInputProperties) Validate() error { +func (p *ModelInputProperties) ValidateAndApplyDefaults() error { + p.SQL = strings.TrimSpace(p.SQL) + p.PreExec = strings.TrimSpace(p.PreExec) + p.PostExec = strings.TrimSpace(p.PostExec) if p.SQL == "" { return fmt.Errorf("missing property 'sql'") } @@ -47,7 +50,7 @@ type ModelOutputProperties struct { CreateSecretsFromConnectors []string `mapstructure:"create_secrets_from_connectors"` } -func (p *ModelOutputProperties) validateAndApplyDefaults(opts *drivers.ModelExecuteOptions, ip *ModelInputProperties, op *ModelOutputProperties) error { +func (p *ModelOutputProperties) validateAndApplyDefaults(opts *drivers.ModelExecuteOptions, ip *ModelInputProperties) error { if opts.Incremental || opts.PartitionRun { if p.Materialize != nil && !*p.Materialize { return fmt.Errorf("incremental or partitioned models must be materialized") @@ -78,21 +81,23 @@ func (p *ModelOutputProperties) validateAndApplyDefaults(opts *drivers.ModelExec // We want to use partition_overwrite as the default incremental strategy for models with partitions. // This requires us to inject the partition key into the SQL query, so this only works for SQL models. - if op.IncrementalStrategy == drivers.IncrementalStrategyUnspecified { - if len(op.UniqueKey) > 0 { - op.IncrementalStrategy = drivers.IncrementalStrategyMerge + if p.IncrementalStrategy == drivers.IncrementalStrategyUnspecified { + if len(p.UniqueKey) > 0 { + p.IncrementalStrategy = drivers.IncrementalStrategyMerge } else if opts.PartitionRun && ip != nil && ip.SQL != "" { ip.SQL = fmt.Sprintf("SELECT %s AS __rill_partition, * FROM (%s\n)", safeSQLString(opts.PartitionKey), ip.SQL) - op.IncrementalStrategy = drivers.IncrementalStrategyPartitionOverwrite - op.PartitionBy = "__rill_partition" + p.IncrementalStrategy = drivers.IncrementalStrategyPartitionOverwrite + p.PartitionBy = "__rill_partition" } } // If we failed to apply a better incremental strategy, fall back to append. - if op.IncrementalStrategy == drivers.IncrementalStrategyUnspecified { - op.IncrementalStrategy = drivers.IncrementalStrategyAppend + if p.IncrementalStrategy == drivers.IncrementalStrategyUnspecified { + p.IncrementalStrategy = drivers.IncrementalStrategyAppend } + p.PreExec = strings.TrimSpace(p.PreExec) + p.PostExec = strings.TrimSpace(p.PostExec) return nil } diff --git a/runtime/reconcilers/model_test.go b/runtime/reconcilers/model_test.go index bb3b0631273..3c768261381 100644 --- a/runtime/reconcilers/model_test.go +++ b/runtime/reconcilers/model_test.go @@ -151,7 +151,7 @@ sql: SELECT 1 AS num } func TestPartitionedIncrementalPostExecSeesIncrementalFlag(t *testing.T) { - rt, instanceID := testruntime.NewInstance(t) + rt, instanceID := testruntime.NewInstanceWithOptions(t, testruntime.InstanceOptions{StageChanges: true}) testruntime.PutFiles(t, rt, instanceID, map[string]string{ "rill.yaml": ``,