mirror of
https://github.com/temporalio/temporal.git
synced 2026-08-30 18:41:49 -07:00
Combine PostgreSQL Visibility schema upgrades from v1.10 through v1.13 (#10371)
## What changed? Combine PostgreSQL Visibility schema upgrades from v1.10 through v1.13 ## Why? Adding one column at a time requires scanning the entire table. So, combining all the schema changes in v1.30.0 release into a single SQL upgrade file. https://github.com/temporalio/temporal/issues/10358 ## How did you test it? - [x] built - [x] run locally and tested manually - [x] covered by existing tests - [ ] added new unit test(s) - [ ] added new functional test(s) ## Potential risks
This commit is contained in:
Rodrigo Zhou
@@ -14,14 +14,22 @@ const (
|
||||
queryDelimiter = ';'
|
||||
querySliceDefaultSize = 100
|
||||
|
||||
sqlLeftParenthesis = '('
|
||||
sqlRightParenthesis = ')'
|
||||
sqlIfKeyword = "if"
|
||||
sqlBeginKeyword = "begin"
|
||||
sqlEndKeyword = "end"
|
||||
sqlLineComment = "--"
|
||||
sqlSingleQuote = '\''
|
||||
sqlDoubleQuote = '"'
|
||||
sqlLeftParenthesis = '('
|
||||
sqlRightParenthesis = ')'
|
||||
sqlCreateKeyword = "create"
|
||||
sqlTableKeyword = "table"
|
||||
sqlIndexKeyword = "index"
|
||||
sqlConcurrentlyKeyword = "concurrently"
|
||||
sqlAddKeyword = "add"
|
||||
sqlColumnKeyword = "column"
|
||||
sqlDoubleDollarKeyword = "$$"
|
||||
sqlIfKeyword = "if"
|
||||
sqlLoopKeyword = "loop"
|
||||
sqlBeginKeyword = "begin"
|
||||
sqlEndKeyword = "end"
|
||||
sqlLineComment = "--"
|
||||
sqlSingleQuote = '\''
|
||||
sqlDoubleQuote = '"'
|
||||
)
|
||||
|
||||
// LoadAndSplitQuery loads and split cql / sql query into one statement per string.
|
||||
@@ -78,11 +86,37 @@ func LoadAndSplitQueryFromReaders(
|
||||
}
|
||||
st = st[:len(st)-1]
|
||||
|
||||
case sqlIfKeyword[0]:
|
||||
if hasWordAt(contentStr, sqlIfKeyword, j) {
|
||||
st = append(st, sqlIfKeyword[0])
|
||||
j += len(sqlIfKeyword) - 1
|
||||
case sqlDoubleDollarKeyword[0]:
|
||||
if !hasWordAt(contentStr, sqlDoubleDollarKeyword, j) {
|
||||
continue
|
||||
}
|
||||
if len(st) == 0 || st[len(st)-1] != sqlDoubleDollarKeyword[0] {
|
||||
st = append(st, sqlDoubleDollarKeyword[0])
|
||||
j += len(sqlDoubleDollarKeyword) - 1
|
||||
} else {
|
||||
st = st[:len(st)-1]
|
||||
j += len(sqlDoubleDollarKeyword) - 1
|
||||
}
|
||||
|
||||
case sqlIfKeyword[0]:
|
||||
if !hasWordAt(contentStr, sqlIfKeyword, j) {
|
||||
continue
|
||||
}
|
||||
if hasWordsBefore(contentStr, j-1, sqlAddKeyword, sqlColumnKeyword) ||
|
||||
hasWordsBefore(contentStr, j-1, sqlCreateKeyword, sqlIndexKeyword) ||
|
||||
hasWordsBefore(contentStr, j-1, sqlCreateKeyword, sqlIndexKeyword, sqlConcurrentlyKeyword) ||
|
||||
hasWordsBefore(contentStr, j-1, sqlCreateKeyword, sqlTableKeyword) {
|
||||
continue
|
||||
}
|
||||
st = append(st, sqlIfKeyword[0])
|
||||
j += len(sqlIfKeyword) - 1
|
||||
|
||||
case sqlLoopKeyword[0]:
|
||||
if !hasWordAt(contentStr, sqlLoopKeyword, j) {
|
||||
continue
|
||||
}
|
||||
st = append(st, sqlLoopKeyword[0])
|
||||
j += len(sqlLoopKeyword) - 1
|
||||
|
||||
case sqlBeginKeyword[0]:
|
||||
if hasWordAt(contentStr, sqlBeginKeyword, j) {
|
||||
@@ -100,6 +134,13 @@ func LoadAndSplitQueryFromReaders(
|
||||
}
|
||||
st = st[:len(st)-1]
|
||||
j = after + len(sqlIfKeyword) - 1
|
||||
} else if ok, after := hasWordAfter(contentStr, sqlLoopKeyword, j+len(sqlEndKeyword)); ok {
|
||||
//nolint:revive
|
||||
if len(st) == 0 || st[len(st)-1] != sqlLoopKeyword[0] {
|
||||
return nil, errors.New("error reading contents: unmatched `END LOOP` keyword")
|
||||
}
|
||||
st = st[:len(st)-1]
|
||||
j = after + len(sqlLoopKeyword) - 1
|
||||
} else {
|
||||
if len(st) == 0 || st[len(st)-1] != sqlBeginKeyword[0] {
|
||||
return nil, errors.New("error reading contents: unmatched `END` keyword")
|
||||
@@ -183,6 +224,35 @@ func hasWordAfter(s, word string, pos int) (bool, int) {
|
||||
return hasWordAt(s, word, after), after
|
||||
}
|
||||
|
||||
// hasWordsBefore checks if the given words appears before position pos in s,
|
||||
// separated by at least one space, and each word is a whole word.
|
||||
// Words must appear is the given order.
|
||||
// Eg: hasWordsBefore("CREATE TABLE IF", 12, "CREATE", "TABLE") returns true.
|
||||
func hasWordsBefore(s string, pos int, words ...string) bool {
|
||||
if len(words) == 0 {
|
||||
return true
|
||||
}
|
||||
if pos <= 0 || !unicode.IsSpace(rune(s[pos])) {
|
||||
return false
|
||||
}
|
||||
for i := len(words) - 1; i >= 0; i-- {
|
||||
// skip spaces
|
||||
for pos >= 0 && unicode.IsSpace(rune(s[pos])) {
|
||||
pos--
|
||||
}
|
||||
// go to first char of word
|
||||
for pos > 0 && isAlphanumeric(s[pos-1]) {
|
||||
pos--
|
||||
}
|
||||
// check if current pos matches the word
|
||||
if pos < 0 || !hasWordAt(s, words[i], pos) {
|
||||
return false
|
||||
}
|
||||
pos--
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func isAlphanumeric(c byte) bool {
|
||||
return unicode.IsLetter(rune(c)) || unicode.IsDigit(rune(c))
|
||||
}
|
||||
|
||||
@@ -128,3 +128,54 @@ func (s *queryUtilSuite) TestHasWordAt() {
|
||||
s.False(hasWordAt("7BEGIN", "BEGIN", 1))
|
||||
s.False(hasWordAt("BEGIN7", "BEGIN", 0))
|
||||
}
|
||||
|
||||
func (s *queryUtilSuite) TestHasWordsBefore() {
|
||||
// example from the function's doc comment
|
||||
s.True(hasWordsBefore("CREATE TABLE IF", 12, "CREATE", "TABLE"))
|
||||
|
||||
// empty words slice always returns true, regardless of pos
|
||||
s.True(hasWordsBefore("CREATE TABLE", 6))
|
||||
s.True(hasWordsBefore("", 0))
|
||||
s.True(hasWordsBefore("anything", 3))
|
||||
|
||||
// pos <= 0 returns false when words are provided
|
||||
s.False(hasWordsBefore("CREATE TABLE", 0, "CREATE"))
|
||||
s.False(hasWordsBefore(" CREATE", 0, "CREATE"))
|
||||
s.False(hasWordsBefore(" CREATE", 2, "CREATE"))
|
||||
|
||||
// pos must point to a whitespace character
|
||||
s.False(hasWordsBefore("CREATE TABLE", 5, "CREATE")) // s[5]='E'
|
||||
s.False(hasWordsBefore("CREATE TABLE", 4, "CREATE")) // s[4]='T'
|
||||
|
||||
// single word matches
|
||||
s.True(hasWordsBefore("CREATE TABLE", 6, "CREATE"))
|
||||
s.True(hasWordsBefore(" CREATE ", 8, "CREATE")) // leading whitespace ok
|
||||
s.True(hasWordsBefore("(CREATE ", 7, "CREATE")) // preceded by non-alphanumeric
|
||||
s.True(hasWordsBefore("CREATE\tIF", 6, "CREATE")) // tab counts as whitespace
|
||||
s.True(hasWordsBefore("CREATE ", 8, "CREATE")) // multiple trailing spaces ok
|
||||
|
||||
// single word - wrong word at that position
|
||||
s.False(hasWordsBefore("CREATE TABLE", 6, "TABLE"))
|
||||
|
||||
// word must be a whole word (alphanumeric adjacency rejected)
|
||||
s.False(hasWordsBefore("XCREATE ", 7, "CREATE"))
|
||||
s.False(hasWordsBefore("7CREATE ", 7, "CREATE"))
|
||||
|
||||
// multiple words match in order
|
||||
s.True(hasWordsBefore("ADD COLUMN IF", 10, "ADD", "COLUMN"))
|
||||
s.True(hasWordsBefore("CREATE INDEX IF", 12, "CREATE", "INDEX"))
|
||||
|
||||
// multiple words with extra whitespace between them
|
||||
s.True(hasWordsBefore("CREATE TABLE IF", 14, "CREATE", "TABLE"))
|
||||
s.True(hasWordsBefore("CREATE\tTABLE\tIF", 12, "CREATE", "TABLE"))
|
||||
|
||||
// words in wrong order
|
||||
s.False(hasWordsBefore("CREATE TABLE IF", 12, "TABLE", "CREATE"))
|
||||
|
||||
// one of the words is missing / different
|
||||
s.False(hasWordsBefore("CREATE FOO IF", 10, "CREATE", "TABLE"))
|
||||
s.False(hasWordsBefore("DROP TABLE IF", 10, "CREATE", "TABLE"))
|
||||
|
||||
// more words requested than are available before pos
|
||||
s.False(hasWordsBefore("TABLE ", 5, "CREATE", "TABLE"))
|
||||
}
|
||||
|
||||
@@ -2,7 +2,6 @@
|
||||
"CurrVersion": "1.10",
|
||||
"MinCompatibleVersion": "0.1",
|
||||
"Description": "add TemporalReportedProblems column",
|
||||
"SchemaUpdateCqlFiles": [
|
||||
"add_temporal_reported_columns.sql"
|
||||
]
|
||||
}
|
||||
"SchemaUpdateCqlFiles": [],
|
||||
"AllowNoCqlFiles": true
|
||||
}
|
||||
|
||||
@@ -2,7 +2,6 @@
|
||||
"CurrVersion": "1.11",
|
||||
"MinCompatibleVersion": "0.1",
|
||||
"Description": "add CHASM search attributes columns and indices",
|
||||
"SchemaUpdateCqlFiles": [
|
||||
"add_chasm_search_attributes.sql"
|
||||
]
|
||||
"SchemaUpdateCqlFiles": [],
|
||||
"AllowNoCqlFiles": true
|
||||
}
|
||||
|
||||
@@ -2,7 +2,6 @@
|
||||
"CurrVersion": "1.12",
|
||||
"MinCompatibleVersion": "0.1",
|
||||
"Description": "add TemporalLowCardinalityKeyword01 to chasm_search_attributes",
|
||||
"SchemaUpdateCqlFiles": [
|
||||
"add_low_cardinality_keyword.sql"
|
||||
]
|
||||
"SchemaUpdateCqlFiles": [],
|
||||
"AllowNoCqlFiles": true
|
||||
}
|
||||
|
||||
@@ -0,0 +1,94 @@
|
||||
-- Add new columns
|
||||
ALTER TABLE executions_visibility
|
||||
-- v1.10
|
||||
ADD COLUMN IF NOT EXISTS TemporalReportedProblems JSONB GENERATED ALWAYS AS (search_attributes->'TemporalReportedProblems') STORED,
|
||||
|
||||
-- v1.11
|
||||
-- Pre-allocated CHASM search attributes
|
||||
ADD COLUMN IF NOT EXISTS TemporalBool01 BOOLEAN GENERATED ALWAYS AS ((search_attributes->'TemporalBool01')::boolean) STORED,
|
||||
ADD COLUMN IF NOT EXISTS TemporalBool02 BOOLEAN GENERATED ALWAYS AS ((search_attributes->'TemporalBool02')::boolean) STORED,
|
||||
ADD COLUMN IF NOT EXISTS TemporalDatetime01 TIMESTAMP GENERATED ALWAYS AS (convert_ts(search_attributes->>'TemporalDatetime01')) STORED,
|
||||
ADD COLUMN IF NOT EXISTS TemporalDatetime02 TIMESTAMP GENERATED ALWAYS AS (convert_ts(search_attributes->>'TemporalDatetime02')) STORED,
|
||||
ADD COLUMN IF NOT EXISTS TemporalDouble01 DECIMAL(20, 5) GENERATED ALWAYS AS ((search_attributes->'TemporalDouble01')::decimal) STORED,
|
||||
ADD COLUMN IF NOT EXISTS TemporalDouble02 DECIMAL(20, 5) GENERATED ALWAYS AS ((search_attributes->'TemporalDouble02')::decimal) STORED,
|
||||
ADD COLUMN IF NOT EXISTS TemporalInt01 BIGINT GENERATED ALWAYS AS ((search_attributes->'TemporalInt01')::bigint) STORED,
|
||||
ADD COLUMN IF NOT EXISTS TemporalInt02 BIGINT GENERATED ALWAYS AS ((search_attributes->'TemporalInt02')::bigint) STORED,
|
||||
ADD COLUMN IF NOT EXISTS TemporalKeyword01 VARCHAR(255) GENERATED ALWAYS AS (search_attributes->>'TemporalKeyword01') STORED,
|
||||
ADD COLUMN IF NOT EXISTS TemporalKeyword02 VARCHAR(255) GENERATED ALWAYS AS (search_attributes->>'TemporalKeyword02') STORED,
|
||||
ADD COLUMN IF NOT EXISTS TemporalKeyword03 VARCHAR(255) GENERATED ALWAYS AS (search_attributes->>'TemporalKeyword03') STORED,
|
||||
ADD COLUMN IF NOT EXISTS TemporalKeyword04 VARCHAR(255) GENERATED ALWAYS AS (search_attributes->>'TemporalKeyword04') STORED,
|
||||
ADD COLUMN IF NOT EXISTS TemporalKeywordList01 JSONB GENERATED ALWAYS AS (search_attributes->'TemporalKeywordList01') STORED,
|
||||
ADD COLUMN IF NOT EXISTS TemporalKeywordList02 JSONB GENERATED ALWAYS AS (search_attributes->'TemporalKeywordList02') STORED,
|
||||
|
||||
-- v1.12
|
||||
ADD COLUMN IF NOT EXISTS TemporalLowCardinalityKeyword01 VARCHAR(255) GENERATED ALWAYS AS (search_attributes->>'TemporalLowCardinalityKeyword01') STORED,
|
||||
|
||||
-- v1.13
|
||||
ADD COLUMN IF NOT EXISTS TemporalUsedWorkerDeploymentVersions JSONB GENERATED ALWAYS AS (search_attributes->'TemporalUsedWorkerDeploymentVersions') STORED;
|
||||
|
||||
-- Drop invalid indices
|
||||
DO LANGUAGE 'plpgsql' $$
|
||||
DECLARE
|
||||
r RECORD;
|
||||
BEGIN
|
||||
FOR r IN
|
||||
SELECT i.relname as indexname
|
||||
FROM
|
||||
pg_class i,
|
||||
pg_index ix
|
||||
WHERE
|
||||
i.oid = ix.indexrelid
|
||||
AND ix.indrelid = (SELECT oid FROM pg_class WHERE relname = 'executions_visibility')
|
||||
AND i.relname IN (
|
||||
'by_temporal_reported_problems',
|
||||
'by_temporal_bool_01',
|
||||
'by_temporal_bool_02',
|
||||
'by_temporal_datetime_01',
|
||||
'by_temporal_datetime_02',
|
||||
'by_temporal_double_01',
|
||||
'by_temporal_double_02',
|
||||
'by_temporal_int_01',
|
||||
'by_temporal_int_02',
|
||||
'by_temporal_keyword_01',
|
||||
'by_temporal_keyword_02',
|
||||
'by_temporal_keyword_03',
|
||||
'by_temporal_keyword_04',
|
||||
'by_temporal_keyword_list_01',
|
||||
'by_temporal_keyword_list_02',
|
||||
'by_temporal_low_cardinality_keyword_01',
|
||||
'by_used_deployment_versions'
|
||||
)
|
||||
AND NOT ix.indisvalid
|
||||
LOOP
|
||||
EXECUTE format('DROP INDEX %I', r.indexname);
|
||||
RAISE NOTICE 'Dropped invalid index %', r.indexname;
|
||||
END LOOP;
|
||||
END $$;
|
||||
|
||||
-- Create new indices
|
||||
|
||||
-- v1.10
|
||||
CREATE INDEX CONCURRENTLY IF NOT EXISTS by_temporal_reported_problems ON executions_visibility USING GIN (namespace_id, TemporalReportedProblems jsonb_path_ops);
|
||||
|
||||
-- v1.11
|
||||
-- Indexes for the pre-allocated CHASM search attributes
|
||||
CREATE INDEX CONCURRENTLY IF NOT EXISTS by_temporal_bool_01 ON executions_visibility (namespace_id, TemporalBool01, (COALESCE(close_time, '9999-12-31 23:59:59')) DESC, start_time DESC, run_id);
|
||||
CREATE INDEX CONCURRENTLY IF NOT EXISTS by_temporal_bool_02 ON executions_visibility (namespace_id, TemporalBool02, (COALESCE(close_time, '9999-12-31 23:59:59')) DESC, start_time DESC, run_id);
|
||||
CREATE INDEX CONCURRENTLY IF NOT EXISTS by_temporal_datetime_01 ON executions_visibility (namespace_id, TemporalDatetime01, (COALESCE(close_time, '9999-12-31 23:59:59')) DESC, start_time DESC, run_id);
|
||||
CREATE INDEX CONCURRENTLY IF NOT EXISTS by_temporal_datetime_02 ON executions_visibility (namespace_id, TemporalDatetime02, (COALESCE(close_time, '9999-12-31 23:59:59')) DESC, start_time DESC, run_id);
|
||||
CREATE INDEX CONCURRENTLY IF NOT EXISTS by_temporal_double_01 ON executions_visibility (namespace_id, TemporalDouble01, (COALESCE(close_time, '9999-12-31 23:59:59')) DESC, start_time DESC, run_id);
|
||||
CREATE INDEX CONCURRENTLY IF NOT EXISTS by_temporal_double_02 ON executions_visibility (namespace_id, TemporalDouble02, (COALESCE(close_time, '9999-12-31 23:59:59')) DESC, start_time DESC, run_id);
|
||||
CREATE INDEX CONCURRENTLY IF NOT EXISTS by_temporal_int_01 ON executions_visibility (namespace_id, TemporalInt01, (COALESCE(close_time, '9999-12-31 23:59:59')) DESC, start_time DESC, run_id);
|
||||
CREATE INDEX CONCURRENTLY IF NOT EXISTS by_temporal_int_02 ON executions_visibility (namespace_id, TemporalInt02, (COALESCE(close_time, '9999-12-31 23:59:59')) DESC, start_time DESC, run_id);
|
||||
CREATE INDEX CONCURRENTLY IF NOT EXISTS by_temporal_keyword_01 ON executions_visibility (namespace_id, TemporalKeyword01, (COALESCE(close_time, '9999-12-31 23:59:59')) DESC, start_time DESC, run_id);
|
||||
CREATE INDEX CONCURRENTLY IF NOT EXISTS by_temporal_keyword_02 ON executions_visibility (namespace_id, TemporalKeyword02, (COALESCE(close_time, '9999-12-31 23:59:59')) DESC, start_time DESC, run_id);
|
||||
CREATE INDEX CONCURRENTLY IF NOT EXISTS by_temporal_keyword_03 ON executions_visibility (namespace_id, TemporalKeyword03, (COALESCE(close_time, '9999-12-31 23:59:59')) DESC, start_time DESC, run_id);
|
||||
CREATE INDEX CONCURRENTLY IF NOT EXISTS by_temporal_keyword_04 ON executions_visibility (namespace_id, TemporalKeyword04, (COALESCE(close_time, '9999-12-31 23:59:59')) DESC, start_time DESC, run_id);
|
||||
CREATE INDEX CONCURRENTLY IF NOT EXISTS by_temporal_keyword_list_01 ON executions_visibility USING GIN (namespace_id, TemporalKeywordList01 jsonb_path_ops);
|
||||
CREATE INDEX CONCURRENTLY IF NOT EXISTS by_temporal_keyword_list_02 ON executions_visibility USING GIN (namespace_id, TemporalKeywordList02 jsonb_path_ops);
|
||||
|
||||
-- v1.12
|
||||
CREATE INDEX CONCURRENTLY IF NOT EXISTS by_temporal_low_cardinality_keyword_01 ON executions_visibility (namespace_id, TemporalLowCardinalityKeyword01, (COALESCE(close_time, '9999-12-31 23:59:59')) DESC, start_time DESC, run_id);
|
||||
|
||||
-- v1.13
|
||||
CREATE INDEX CONCURRENTLY IF NOT EXISTS by_used_deployment_versions ON executions_visibility USING GIN (namespace_id, TemporalUsedWorkerDeploymentVersions jsonb_path_ops);
|
||||
@@ -1,8 +1,8 @@
|
||||
{
|
||||
"CurrVersion": "1.13",
|
||||
"MinCompatibleVersion": "0.1",
|
||||
"Description": "add TemporalUsedWorkerDeploymentVersions search attribute",
|
||||
"Description": "combined changed from v1.10 to v1.13 (add TemporalUsedWorkerDeploymentVersions search attribute)",
|
||||
"SchemaUpdateCqlFiles": [
|
||||
"add_used_deployment_versions_search_attribute.sql"
|
||||
"combined_v1.10_v1.13.sql"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user