// The MIT License // // Copyright (c) 2020 Temporal Technologies Inc. All rights reserved. // // Copyright (c) 2020 Uber Technologies, Inc. // // Permission is hereby granted, free of charge, to any person obtaining a copy // of this software and associated documentation files (the "Software"), to deal // in the Software without restriction, including without limitation the rights // to use, copy, modify, merge, publish, distribute, sublicense, and/or sell // copies of the Software, and to permit persons to whom the Software is // furnished to do so, subject to the following conditions: // // The above copyright notice and this permission notice shall be included in // all copies or substantial portions of the Software. // // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, // FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE // AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN // THE SOFTWARE. package postgresql import ( "database/sql" "fmt" "strings" "go.temporal.io/server/common/persistence/sql/sqlplugin" ) const ( deleteMapQueryTemplate = `DELETE FROM %v WHERE shard_id = $1 AND namespace_id = $2 AND workflow_id = $3 AND run_id = $4` // %[2]v is the columns of the value struct (i.e. no primary key columns), comma separated // %[3]v should be %[2]v with colons prepended. // i.e. %[3]v = ",".join(":" + s for s in %[2]v) // %[5]v should be %[2]v with "excluded." prepended. // i.e. %[5]v = ",".join("excluded." + s for s in %[2]v) // So that this query can be used with BindNamed // %[4]v should be the name of the key associated with the map // e.g. for ActivityInfo it is "schedule_id" setKeyInMapQueryTemplate = `INSERT INTO %[1]v (shard_id, namespace_id, workflow_id, run_id, %[4]v, %[2]v) VALUES (:shard_id, :namespace_id, :workflow_id, :run_id, :%[4]v, %[3]v) ON CONFLICT (shard_id, namespace_id, workflow_id, run_id, %[4]v) DO UPDATE SET (shard_id, namespace_id, workflow_id, run_id, %[4]v, %[2]v) = (excluded.shard_id, excluded.namespace_id, excluded.workflow_id, excluded.run_id, excluded.%[4]v, %[5]v)` // %[2]v is the name of the key deleteKeyInMapQueryTemplate = `DELETE FROM %[1]v WHERE shard_id = $1 AND namespace_id = $2 AND workflow_id = $3 AND run_id = $4 AND %[2]v = $5` // %[1]v is the name of the table // %[2]v is the name of the key // %[3]v is the value columns, separated by commas getMapQueryTemplate = `SELECT %[2]v, %[3]v FROM %[1]v WHERE shard_id = $1 AND namespace_id = $2 AND workflow_id = $3 AND run_id = $4` ) const ( deleteAllSignalsRequestedSetQuery = `DELETE FROM signals_requested_sets WHERE shard_id = $1 AND namespace_id = $2 AND workflow_id = $3 AND run_id = $4 ` createSignalsRequestedSetQuery = `INSERT INTO signals_requested_sets (shard_id, namespace_id, workflow_id, run_id, signal_id) VALUES (:shard_id, :namespace_id, :workflow_id, :run_id, :signal_id) ON CONFLICT (shard_id, namespace_id, workflow_id, run_id, signal_id) DO NOTHING` deleteSignalsRequestedSetQuery = `DELETE FROM signals_requested_sets WHERE shard_id = $1 AND namespace_id = $2 AND workflow_id = $3 AND run_id = $4 AND signal_id = $5` getSignalsRequestedSetQuery = `SELECT signal_id FROM signals_requested_sets WHERE shard_id = $1 AND namespace_id = $2 AND workflow_id = $3 AND run_id = $4` ) func stringMap(a []string, f func(string) string) []string { b := make([]string, len(a)) for i, v := range a { b[i] = f(v) } return b } func makeDeleteMapQry(tableName string) string { return fmt.Sprintf(deleteMapQueryTemplate, tableName) } func makeSetKeyInMapQry(tableName string, nonPrimaryKeyColumns []string, mapKeyName string) string { return fmt.Sprintf(setKeyInMapQueryTemplate, tableName, strings.Join(nonPrimaryKeyColumns, ","), strings.Join(stringMap(nonPrimaryKeyColumns, func(x string) string { return ":" + x }), ","), mapKeyName, strings.Join(stringMap(nonPrimaryKeyColumns, func(x string) string { return "excluded." + x }), ",")) } func makeDeleteKeyInMapQry(tableName string, mapKeyName string) string { return fmt.Sprintf(deleteKeyInMapQueryTemplate, tableName, mapKeyName) } func makeGetMapQryTemplate(tableName string, nonPrimaryKeyColumns []string, mapKeyName string) string { return fmt.Sprintf(getMapQueryTemplate, tableName, mapKeyName, strings.Join(nonPrimaryKeyColumns, ",")) } var ( // Omit shard_id, run_id, namespace_id, workflow_id, schedule_id since they're in the primary key activityInfoColumns = []string{ "data", "data_encoding", } activityInfoTableName = "activity_info_maps" activityInfoKey = "schedule_id" deleteActivityInfoMapQry = makeDeleteMapQry(activityInfoTableName) setKeyInActivityInfoMapQry = makeSetKeyInMapQry(activityInfoTableName, activityInfoColumns, activityInfoKey) deleteKeyInActivityInfoMapQry = makeDeleteKeyInMapQry(activityInfoTableName, activityInfoKey) getActivityInfoMapQry = makeGetMapQryTemplate(activityInfoTableName, activityInfoColumns, activityInfoKey) ) // ReplaceIntoActivityInfoMaps replaces one or more rows in activity_info_maps table func (pdb *db) ReplaceIntoActivityInfoMaps(rows []sqlplugin.ActivityInfoMapsRow) (sql.Result, error) { return pdb.conn.NamedExec(setKeyInActivityInfoMapQry, rows) } // SelectFromActivityInfoMaps reads one or more rows from activity_info_maps table func (pdb *db) SelectFromActivityInfoMaps(filter *sqlplugin.ActivityInfoMapsFilter) ([]sqlplugin.ActivityInfoMapsRow, error) { var rows []sqlplugin.ActivityInfoMapsRow err := pdb.conn.Select(&rows, getActivityInfoMapQry, filter.ShardID, filter.NamespaceID, filter.WorkflowID, filter.RunID) for i := 0; i < len(rows); i++ { rows[i].ShardID = int64(filter.ShardID) rows[i].NamespaceID = filter.NamespaceID rows[i].WorkflowID = filter.WorkflowID rows[i].RunID = filter.RunID } return rows, err } // DeleteFromActivityInfoMaps deletes one or more rows from activity_info_maps table func (pdb *db) DeleteFromActivityInfoMaps(filter *sqlplugin.ActivityInfoMapsFilter) (sql.Result, error) { if filter.ScheduleID != nil { return pdb.conn.Exec(deleteKeyInActivityInfoMapQry, filter.ShardID, filter.NamespaceID, filter.WorkflowID, filter.RunID, *filter.ScheduleID) } return pdb.conn.Exec(deleteActivityInfoMapQry, filter.ShardID, filter.NamespaceID, filter.WorkflowID, filter.RunID) } var ( timerInfoColumns = []string{ "data", "data_encoding", } timerInfoTableName = "timer_info_maps" timerInfoKey = "timer_id" deleteTimerInfoMapSQLQuery = makeDeleteMapQry(timerInfoTableName) setKeyInTimerInfoMapSQLQuery = makeSetKeyInMapQry(timerInfoTableName, timerInfoColumns, timerInfoKey) deleteKeyInTimerInfoMapSQLQuery = makeDeleteKeyInMapQry(timerInfoTableName, timerInfoKey) getTimerInfoMapSQLQuery = makeGetMapQryTemplate(timerInfoTableName, timerInfoColumns, timerInfoKey) ) // ReplaceIntoTimerInfoMaps replaces one or more rows in timer_info_maps table func (pdb *db) ReplaceIntoTimerInfoMaps(rows []sqlplugin.TimerInfoMapsRow) (sql.Result, error) { return pdb.conn.NamedExec(setKeyInTimerInfoMapSQLQuery, rows) } // SelectFromTimerInfoMaps reads one or more rows from timer_info_maps table func (pdb *db) SelectFromTimerInfoMaps(filter *sqlplugin.TimerInfoMapsFilter) ([]sqlplugin.TimerInfoMapsRow, error) { var rows []sqlplugin.TimerInfoMapsRow err := pdb.conn.Select(&rows, getTimerInfoMapSQLQuery, filter.ShardID, filter.NamespaceID, filter.WorkflowID, filter.RunID) for i := 0; i < len(rows); i++ { rows[i].ShardID = int64(filter.ShardID) rows[i].NamespaceID = filter.NamespaceID rows[i].WorkflowID = filter.WorkflowID rows[i].RunID = filter.RunID } return rows, err } // DeleteFromTimerInfoMaps deletes one or more rows from timer_info_maps table func (pdb *db) DeleteFromTimerInfoMaps(filter *sqlplugin.TimerInfoMapsFilter) (sql.Result, error) { if filter.TimerID != nil { return pdb.conn.Exec(deleteKeyInTimerInfoMapSQLQuery, filter.ShardID, filter.NamespaceID, filter.WorkflowID, filter.RunID, *filter.TimerID) } return pdb.conn.Exec(deleteTimerInfoMapSQLQuery, filter.ShardID, filter.NamespaceID, filter.WorkflowID, filter.RunID) } var ( childExecutionInfoColumns = []string{ "data", "data_encoding", } childExecutionInfoTableName = "child_execution_info_maps" childExecutionInfoKey = "initiated_id" deleteChildExecutionInfoMapQry = makeDeleteMapQry(childExecutionInfoTableName) setKeyInChildExecutionInfoMapQry = makeSetKeyInMapQry(childExecutionInfoTableName, childExecutionInfoColumns, childExecutionInfoKey) deleteKeyInChildExecutionInfoMapQry = makeDeleteKeyInMapQry(childExecutionInfoTableName, childExecutionInfoKey) getChildExecutionInfoMapQry = makeGetMapQryTemplate(childExecutionInfoTableName, childExecutionInfoColumns, childExecutionInfoKey) ) // ReplaceIntoChildExecutionInfoMaps replaces one or more rows in child_execution_info_maps table func (pdb *db) ReplaceIntoChildExecutionInfoMaps(rows []sqlplugin.ChildExecutionInfoMapsRow) (sql.Result, error) { return pdb.conn.NamedExec(setKeyInChildExecutionInfoMapQry, rows) } // SelectFromChildExecutionInfoMaps reads one or more rows from child_execution_info_maps table func (pdb *db) SelectFromChildExecutionInfoMaps(filter *sqlplugin.ChildExecutionInfoMapsFilter) ([]sqlplugin.ChildExecutionInfoMapsRow, error) { var rows []sqlplugin.ChildExecutionInfoMapsRow err := pdb.conn.Select(&rows, getChildExecutionInfoMapQry, filter.ShardID, filter.NamespaceID, filter.WorkflowID, filter.RunID) for i := 0; i < len(rows); i++ { rows[i].ShardID = int64(filter.ShardID) rows[i].NamespaceID = filter.NamespaceID rows[i].WorkflowID = filter.WorkflowID rows[i].RunID = filter.RunID } return rows, err } // DeleteFromChildExecutionInfoMaps deletes one or more rows from child_execution_info_maps table func (pdb *db) DeleteFromChildExecutionInfoMaps(filter *sqlplugin.ChildExecutionInfoMapsFilter) (sql.Result, error) { if filter.InitiatedID != nil { return pdb.conn.Exec(deleteKeyInChildExecutionInfoMapQry, filter.ShardID, filter.NamespaceID, filter.WorkflowID, filter.RunID, *filter.InitiatedID) } return pdb.conn.Exec(deleteChildExecutionInfoMapQry, filter.ShardID, filter.NamespaceID, filter.WorkflowID, filter.RunID) } var ( requestCancelInfoColumns = []string{ "data", "data_encoding", } requestCancelInfoTableName = "request_cancel_info_maps" requestCancelInfoKey = "initiated_id" deleteRequestCancelInfoMapQry = makeDeleteMapQry(requestCancelInfoTableName) setKeyInRequestCancelInfoMapQry = makeSetKeyInMapQry(requestCancelInfoTableName, requestCancelInfoColumns, requestCancelInfoKey) deleteKeyInRequestCancelInfoMapQry = makeDeleteKeyInMapQry(requestCancelInfoTableName, requestCancelInfoKey) getRequestCancelInfoMapQry = makeGetMapQryTemplate(requestCancelInfoTableName, requestCancelInfoColumns, requestCancelInfoKey) ) // ReplaceIntoRequestCancelInfoMaps replaces one or more rows in request_cancel_info_maps table func (pdb *db) ReplaceIntoRequestCancelInfoMaps(rows []sqlplugin.RequestCancelInfoMapsRow) (sql.Result, error) { return pdb.conn.NamedExec(setKeyInRequestCancelInfoMapQry, rows) } // SelectFromRequestCancelInfoMaps reads one or more rows from request_cancel_info_maps table func (pdb *db) SelectFromRequestCancelInfoMaps(filter *sqlplugin.RequestCancelInfoMapsFilter) ([]sqlplugin.RequestCancelInfoMapsRow, error) { var rows []sqlplugin.RequestCancelInfoMapsRow err := pdb.conn.Select(&rows, getRequestCancelInfoMapQry, filter.ShardID, filter.NamespaceID, filter.WorkflowID, filter.RunID) for i := 0; i < len(rows); i++ { rows[i].ShardID = int64(filter.ShardID) rows[i].NamespaceID = filter.NamespaceID rows[i].WorkflowID = filter.WorkflowID rows[i].RunID = filter.RunID } return rows, err } // DeleteFromRequestCancelInfoMaps deletes one or more rows from request_cancel_info_maps table func (pdb *db) DeleteFromRequestCancelInfoMaps(filter *sqlplugin.RequestCancelInfoMapsFilter) (sql.Result, error) { if filter.InitiatedID != nil { return pdb.conn.Exec(deleteKeyInRequestCancelInfoMapQry, filter.ShardID, filter.NamespaceID, filter.WorkflowID, filter.RunID, *filter.InitiatedID) } return pdb.conn.Exec(deleteRequestCancelInfoMapQry, filter.ShardID, filter.NamespaceID, filter.WorkflowID, filter.RunID) } var ( signalInfoColumns = []string{ "data", "data_encoding", } signalInfoTableName = "signal_info_maps" signalInfoKey = "initiated_id" deleteSignalInfoMapQry = makeDeleteMapQry(signalInfoTableName) setKeyInSignalInfoMapQry = makeSetKeyInMapQry(signalInfoTableName, signalInfoColumns, signalInfoKey) deleteKeyInSignalInfoMapQry = makeDeleteKeyInMapQry(signalInfoTableName, signalInfoKey) getSignalInfoMapQry = makeGetMapQryTemplate(signalInfoTableName, signalInfoColumns, signalInfoKey) ) // ReplaceIntoSignalInfoMaps replaces one or more rows in signal_info_maps table func (pdb *db) ReplaceIntoSignalInfoMaps(rows []sqlplugin.SignalInfoMapsRow) (sql.Result, error) { return pdb.conn.NamedExec(setKeyInSignalInfoMapQry, rows) } // SelectFromSignalInfoMaps reads one or more rows from signal_info_maps table func (pdb *db) SelectFromSignalInfoMaps(filter *sqlplugin.SignalInfoMapsFilter) ([]sqlplugin.SignalInfoMapsRow, error) { var rows []sqlplugin.SignalInfoMapsRow err := pdb.conn.Select(&rows, getSignalInfoMapQry, filter.ShardID, filter.NamespaceID, filter.WorkflowID, filter.RunID) for i := 0; i < len(rows); i++ { rows[i].ShardID = int64(filter.ShardID) rows[i].NamespaceID = filter.NamespaceID rows[i].WorkflowID = filter.WorkflowID rows[i].RunID = filter.RunID } return rows, err } // DeleteFromSignalInfoMaps deletes one or more rows from signal_info_maps table func (pdb *db) DeleteFromSignalInfoMaps(filter *sqlplugin.SignalInfoMapsFilter) (sql.Result, error) { if filter.InitiatedID != nil { return pdb.conn.Exec(deleteKeyInSignalInfoMapQry, filter.ShardID, filter.NamespaceID, filter.WorkflowID, filter.RunID, *filter.InitiatedID) } return pdb.conn.Exec(deleteSignalInfoMapQry, filter.ShardID, filter.NamespaceID, filter.WorkflowID, filter.RunID) } // InsertIntoSignalsRequestedSets inserts one or more rows into signals_requested_sets table func (pdb *db) InsertIntoSignalsRequestedSets(rows []sqlplugin.SignalsRequestedSetsRow) (sql.Result, error) { return pdb.conn.NamedExec(createSignalsRequestedSetQuery, rows) } // SelectFromSignalsRequestedSets reads one or more rows from signals_requested_sets table func (pdb *db) SelectFromSignalsRequestedSets(filter *sqlplugin.SignalsRequestedSetsFilter) ([]sqlplugin.SignalsRequestedSetsRow, error) { var rows []sqlplugin.SignalsRequestedSetsRow err := pdb.conn.Select(&rows, getSignalsRequestedSetQuery, filter.ShardID, filter.NamespaceID, filter.WorkflowID, filter.RunID) for i := 0; i < len(rows); i++ { rows[i].ShardID = int64(filter.ShardID) rows[i].NamespaceID = filter.NamespaceID rows[i].WorkflowID = filter.WorkflowID rows[i].RunID = filter.RunID } return rows, err } // DeleteFromSignalsRequestedSets deletes one or more rows from signals_requested_sets table func (pdb *db) DeleteFromSignalsRequestedSets(filter *sqlplugin.SignalsRequestedSetsFilter) (sql.Result, error) { if filter.SignalID != nil { return pdb.conn.Exec(deleteSignalsRequestedSetQuery, filter.ShardID, filter.NamespaceID, filter.WorkflowID, filter.RunID, *filter.SignalID) } return pdb.conn.Exec(deleteAllSignalsRequestedSetQuery, filter.ShardID, filter.NamespaceID, filter.WorkflowID, filter.RunID) }