go.temporal.io/server/schema/sqlite/setup.go
200 LOC · 66 covered · 134 uncovered · 15 ranges · 471 concepts · 2 introducers · 415 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
package sqlite
2
3
import (
4
"bytes"
5
"context"
6
_ "embed"
7
"fmt"
8
"io"
9
10
enumspb "go.temporal.io/api/enums/v1"
11
persistencespb "go.temporal.io/server/api/persistence/v1"
12
"go.temporal.io/server/common"
13
"go.temporal.io/server/common/config"
14
"go.temporal.io/server/common/log"
15
"go.temporal.io/server/common/metrics"
16
p "go.temporal.io/server/common/persistence"
17
"go.temporal.io/server/common/persistence/serialization"
18
"go.temporal.io/server/common/persistence/sql"
19
"go.temporal.io/server/common/persistence/sql/sqlplugin"
20
"go.temporal.io/server/common/primitives"
21
"go.temporal.io/server/common/primitives/timestamp"
22
"go.temporal.io/server/common/resolver"
23
"go.temporal.io/server/common/searchattribute/sadefs"
24
)
25
26
var (
27
//go:embed v3/temporal/schema.sql
28
executionSchema []byte
29
//go:embed v3/visibility/schema.sql
30
visibilitySchema []byte
31
)
32
33
// SetupSchema initializes the SQLite schema in an empty database.
34
//
35
// Note: this function may receive breaking changes or be removed in the future.
36
func SetupSchema(cfg *config.SQL) error {
37
db, err := sql.NewSQLAdminDB(sqlplugin.DbKindUnknown, cfg, resolver.NewNoopResolver(), log.NewNoopLogger(), metrics.NoopMetricsHandler)
38
if err != nil {
39
return fmt.Errorf("unable to create SQLite admin DB: %w", err)
40
}
41
defer func() { _ = db.Close() }()
42
43
return SetupSchemaOnDB(db)
44
}
45
46
// SetupSchemaOnDB initializes the SQLite schema in an empty database using existing DB connection.
47
//
48
// Note: this function may receive breaking changes or be removed in the future.
50
>
statements, err := p.LoadAndSplitQueryFromReaders([]io.Reader{bytes.NewBuffer(executionSchema)})
51
>
if err != nil {
52
return fmt.Errorf("error loading execution schema: %w", err)
53
}
54
56
>
if err = db.Exec(stmt); err != nil {
57
return fmt.Errorf("error executing statement %q: %w", stmt, err)
58
}
59
}
60
61
>
statements, err = p.LoadAndSplitQueryFromReaders([]io.Reader{bytes.NewBuffer(visibilitySchema)})
plugin.go ×5
62
>
if err != nil {
63
return fmt.Errorf("error loading visibility schema: %w", err)
64
}
65
67
>
if err = db.Exec(stmt); err != nil {
68
return fmt.Errorf("error executing statement %q: %w", stmt, err)
69
}
70
}
71
73
}
74
75
// NamespaceConfig determines how namespaces should be configured during registration.
76
//
77
// Note: this struct may receive breaking changes or be removed in the future.
78
type NamespaceConfig struct {
79
// Low level representation of a Namespace used by Temporal persistence drivers.
80
Detail *persistencespb.NamespaceDetail
81
// Global Namespaces provide support for replication of Workflow execution across clusters.
82
IsGlobal bool
83
}
84
85
// CreateNamespaces creates namespaces in the target database without the need to have a running Temporal server.
86
//
87
// This exists primarily as a workaround for https://github.com/temporalio/temporal/issues/1336. Namespaces should
88
// typically be created through the Temporal API either via `tctl` or an SDK client.
89
//
90
// Attempting to create a namespace that already exists will be a no-op.
91
//
92
// Note: this function may receive breaking changes or be removed in the future.
93
>
func CreateNamespaces(cfg *config.SQL, namespaces ...*NamespaceConfig) error {
lite_server.go ×25
94
>
db, err := sql.NewSQLDB(sqlplugin.DbKindUnknown, cfg, resolver.NewNoopResolver(), log.NewNoopLogger(), metrics.NoopMetricsHandler)
95
>
if err != nil {
96
return fmt.Errorf("unable to create SQLite admin DB: %w", err)
97
}
99
101
>
if err := createNamespaceIfNotExists(db, ns); err != nil {
102
return fmt.Errorf("error creating namespace %q: %w", ns.Detail.Info.Name, err)
103
}
104
}
105
107
}
108
109
// NewNamespaceConfig initializes a NamespaceConfig with the field values needed to pre-register
110
// the namespace via the CreateNamespaces function.
111
//
112
// Note: this function may receive breaking changes or be removed in the future.
113
func NewNamespaceConfig(
114
activeClusterName string,
115
namespace string,
116
global bool,
117
customSearchAttributes map[string]enumspb.IndexedValueType,
119
>
dbCustomSearchAttributes := sadefs.GetDBIndexSearchAttributes(nil).CustomSearchAttributes
120
>
fieldToAliasMap := map[string]string{}
121
>
for saName, saType := range customSearchAttributes {
122
var targetFieldName string
123
var cntUsed int
124
for fieldName, fieldType := range dbCustomSearchAttributes {
125
if fieldType != saType {
126
continue
127
}
128
if _, ok := fieldToAliasMap[fieldName]; !ok {
129
targetFieldName = fieldName
130
break
131
}
132
cntUsed++
133
}
134
if targetFieldName == "" {
135
return nil, fmt.Errorf(
136
"cannot have more than %d search attributes of type %s",
137
cntUsed,
138
saType,
139
)
140
}
141
fieldToAliasMap[targetFieldName] = saName
142
}
143
145
>
Info: &persistencespb.NamespaceInfo{
146
>
Id: primitives.NewUUID().String(),
147
>
State: enumspb.NAMESPACE_STATE_REGISTERED,
148
>
Name: namespace,
149
>
},
150
>
Config: &persistencespb.NamespaceConfig{
151
>
Retention: timestamp.DurationFromHours(24),
152
>
HistoryArchivalState: enumspb.ARCHIVAL_STATE_DISABLED,
153
>
VisibilityArchivalState: enumspb.ARCHIVAL_STATE_DISABLED,
154
>
CustomSearchAttributeAliases: fieldToAliasMap,
155
>
},
156
>
ReplicationConfig: &persistencespb.NamespaceReplicationConfig{
157
>
ActiveClusterName: activeClusterName,
158
>
Clusters: []string{activeClusterName},
159
>
},
160
>
FailoverVersion: common.EmptyVersion,
161
>
FailoverNotificationVersion: -1,
162
>
}
163
>
return &NamespaceConfig{
164
>
Detail: &detail,
165
>
IsGlobal: global,
166
>
}, nil
167
}
168
169
>
func createNamespaceIfNotExists(db sqlplugin.DB, namespace *NamespaceConfig) error {
lite_server.go ×25
170
>
var (
171
>
name = namespace.Detail.GetInfo().GetName()
172
>
id = primitives.MustParseUUID(namespace.Detail.GetInfo().GetId())
173
>
)
174
>
175
>
// Return early if namespace already exists
176
>
rows, err := db.SelectFromNamespace(context.Background(), sqlplugin.NamespaceFilter{
177
>
Name: &name,
178
>
})
179
>
if err == nil && len(rows) > 0 {
180
return nil
181
}
182
183
>
blob, err := serialization.NewSerializer().NamespaceDetailToBlob(namespace.Detail)
lite_server.go ×25
184
>
if err != nil {
185
return err
186
}
187
188
>
if _, err := db.InsertIntoNamespace(context.Background(), &sqlplugin.NamespaceRow{
lite_server.go ×25
189
>
ID: id,
190
>
Name: name,
191
>
Data: blob.GetData(),
192
>
DataEncoding: blob.GetEncodingType().String(),
193
>
IsGlobal: namespace.IsGlobal,
194
>
NotificationVersion: 0,
195
>
}); err != nil {
196
return err
197
}
198
200
}