diff --git a/runtime/drivers/clickhouse/clickhouse.go b/runtime/drivers/clickhouse/clickhouse.go index b52d938defa..80783016f24 100644 --- a/runtime/drivers/clickhouse/clickhouse.go +++ b/runtime/drivers/clickhouse/clickhouse.go @@ -16,7 +16,6 @@ import ( "github.com/mitchellh/mapstructure" "github.com/rilldata/rill/runtime/drivers" "github.com/rilldata/rill/runtime/pkg/activity" - "github.com/rilldata/rill/runtime/pkg/ctxsync" "github.com/rilldata/rill/runtime/pkg/priorityqueue" "github.com/rilldata/rill/runtime/storage" "go.opentelemetry.io/otel/attribute" @@ -384,7 +383,6 @@ func (d driver) Open(ctx context.Context, connectorName, instanceID string, conf cancel: cancel, metaSem: semaphore.NewWeighted(1), olapSem: priorityqueue.NewSemaphore(olapSemSize), - tempTableMu: ctxsync.NewRWMutex(), opts: opts, embed: embed, } @@ -439,10 +437,6 @@ type Connection struct { metaSem *semaphore.Weighted olapSem *priorityqueue.Semaphore - // tempTableMu serializes the creation of temporary tables (read lock) with the sweep of stale ones (write lock). - // Temporary tables have deterministic names, so without it an execution could recreate a stale table after the sweep listed it and before it dropped it. - tempTableMu ctxsync.RWMutex - // options used to open clickhouse connections opts *clickhouse.Options // embed is embedded clickhouse server for local run diff --git a/runtime/drivers/clickhouse/crud.go b/runtime/drivers/clickhouse/crud.go index d632078f632..40648417dc2 100644 --- a/runtime/drivers/clickhouse/crud.go +++ b/runtime/drivers/clickhouse/crud.go @@ -2,7 +2,6 @@ package clickhouse import ( "context" - "crypto/md5" "errors" "fmt" "strings" @@ -120,8 +119,8 @@ func (c *Connection) insertTableAsSelect(ctx context.Context, name, sql string, if err != nil { return nil, err } - // create temp table with the same schema using a deterministic name - tempName := fmt.Sprintf("%s%s_%x", tempTablePrefix, name, md5.Sum([]byte(sql))) + // create temp table with the same schema + tempName := tempTableName(name) // clean up the temp table defer func() { // cleanup using a different ctx to prevent cleanups being impacted by the main ctx cancellation @@ -137,30 +136,28 @@ func (c *Connection) insertTableAsSelect(ctx context.Context, name, sql string, } }() // create temp table - // Hold the read lock while creating it, so that a sweep of stale temp tables that listed an older table of the same name cannot drop it. - err = c.tempTableMu.RLock(ctx) - if err != nil { - return nil, err - } if engine == "Distributed" { // create a local table first err = c.Exec(ctx, &drivers.Statement{ Query: fmt.Sprintf("CREATE OR REPLACE TABLE %s %s AS %s", safeSQLName(localTableName(tempName)), onClusterClause, safeSQLName(localTableName(name))), Priority: 1, }) - if err == nil { - // then create the distributed table - err = c.createDistributedTable(ctx, tempName, outputProps) + if err != nil { + return nil, err + } + // then create the distributed table + err = c.createDistributedTable(ctx, tempName, outputProps) + if err != nil { + return nil, err } } else { err = c.Exec(ctx, &drivers.Statement{ Query: fmt.Sprintf("CREATE OR REPLACE TABLE %s %s AS %s", safeSQLName(tempName), onClusterClause, safeSQLName(name)), Priority: 1, }) - } - c.tempTableMu.RUnlock() - if err != nil { - return nil, err + if err != nil { + return nil, err + } } // insert into temp table @@ -240,6 +237,12 @@ func (c *Connection) insertTableAsSelect(ctx context.Context, name, sql string, // tempTablePrefix is the prefix of the temporary tables created while writing a model's table, e.g. for column inference and partition inserts. const tempTablePrefix = "__rill_temp_" +// tempTableName returns a unique name for a temporary table used while writing table `name`. +// Unique names prevent concurrent executions (e.g. partitions of the same model) and the stale table sweep from operating on the same table. +func tempTableName(name string) string { + return fmt.Sprintf("%s%s_%d_%s", tempTablePrefix, name, time.Now().UnixMilli(), uuid.New().String()[0:8]) +} + // tempTableTTL is the age after which a temporary table is assumed to be left behind by an execution that never dropped it. // The default model timeout is 3 hours, so this leaves a wide margin for executions that are still running. const tempTableTTL = 24 * time.Hour @@ -249,12 +252,6 @@ const tempTableTTL = 24 * time.Hour // and it may fail, e.g. by timing out while a cancelled insert still holds a lock on the table. // Failures to drop individual tables are logged, so that the remaining tables are still dropped. func (c *Connection) dropStaleTempTables(ctx context.Context, olderThan time.Duration) error { - // Hold the lock until the drops are done: a temporary table created before the listing is not stale, and one created after it waits. - if err := c.tempTableMu.Lock(ctx); err != nil { - return err - } - defer c.tempTableMu.Unlock() - names, err := c.listStaleTempTables(ctx, olderThan) if err != nil { return err @@ -532,7 +529,7 @@ func (c *Connection) createTable(ctx context.Context, name, sql string, outputPr return fmt.Errorf("clickhouse: no columns specified for table %q", name) } // infer columns - v := safeSQLName(fmt.Sprintf("%s%s_%x", tempTablePrefix, name, md5.Sum([]byte(sql)))) + v := safeSQLName(tempTableName(name)) defer func() { // cleanup using a different ctx to prevent cleanups being impacted by the main ctx cancellation // this is a best effort cleanup and query can still timeout and we don't want to wait forever due to blocked calls @@ -542,13 +539,7 @@ func (c *Connection) createTable(ctx context.Context, name, sql string, outputPr defer cancel() _ = c.Exec(ctx, &drivers.Statement{Query: fmt.Sprintf("DROP VIEW IF EXISTS %s %s", v, onClusterClause)}) }() - // Hold the read lock while creating the view, so that a sweep of stale temp tables that listed an older view of the same name cannot drop it. - err := c.tempTableMu.RLock(ctx) - if err != nil { - return err - } - err = c.Exec(ctx, &drivers.Statement{Query: fmt.Sprintf("CREATE OR REPLACE VIEW %s %s AS %s", v, onClusterClause, sql)}) - c.tempTableMu.RUnlock() + err := c.Exec(ctx, &drivers.Statement{Query: fmt.Sprintf("CREATE OR REPLACE VIEW %s %s AS %s", v, onClusterClause, sql)}) if err != nil { return err } diff --git a/runtime/drivers/clickhouse/model_executor_self.go b/runtime/drivers/clickhouse/model_executor_self.go index 51340214f49..cbafe8961f4 100644 --- a/runtime/drivers/clickhouse/model_executor_self.go +++ b/runtime/drivers/clickhouse/model_executor_self.go @@ -4,9 +4,11 @@ import ( "context" "fmt" "strings" + "time" "github.com/mitchellh/mapstructure" "github.com/rilldata/rill/runtime/drivers" + "github.com/rilldata/rill/runtime/pkg/graceful" "github.com/rilldata/rill/runtime/pkg/mapstructureutil" ) @@ -104,6 +106,8 @@ func (e *selfToSelfExecutor) Execute(ctx context.Context, opts *drivers.ModelExe var err error metrics, err = e.c.createTableAsSelect(ctx, stagingTableName, inputProps.SQL, outputProps, inputProps.PreExec, afterCreate) if err != nil { + ctx, cancel := graceful.WithMinimumDuration(ctx, 10*time.Second) + defer cancel() _ = e.c.dropTable(ctx, stagingTableName) return nil, fmt.Errorf("failed to create model: %w", err) }