Github/temporal
1
0
Fork 0
mirror of https://github.com/temporalio/temporal.git synced 2026-08-30 18:41:49 -07:00
Code Issues Packages Projects Releases Wiki Activity

Add external payload size and count to Visibility Schema (#9007)

Browse Source 
## What changed?
Added TemporalExternalPayloadSizeBytes and TemporalExternalPayloadCount
to Visibility Schema as pre-defined search attribute.

## Why?
We are planning to expose external payload size and count as the search
attributes. They are pre-defined, rather than system, because they won't
be set on every single workflow, but only ones which store large
payloads externally (e.g. in S3).

## How did you test it?
- [ ] built
- [ ] run locally and tested manually
- [ ] covered by existing tests
- [ ] added new unit test(s)
- [ ] added new functional test(s)

## Potential risks
Incorrectly modified schema
This commit is contained in:
Vladyslav Simonenko
2026-01-23 11:42:08 -08:00
committed by GitHub
parent 8f60a1aac4
commit 400e2ad37f
13 changed files with 291 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
schema/elasticsearch/visibility/index_template_v7.json
View File

@@ -1 +1 @@
./versioned/v13/index_template_v7.json
./versioned/v14/index_template_v7.json

170
schema/elasticsearch/visibility/versioned/v14/index_template_v7.json Normal file
View File

@@ -0,0 +1,170 @@
{
"order": 0,
"index_patterns": ["temporal_visibility_v1*"],
"settings": {
"index": {
"number_of_shards": "1",
"number_of_replicas": "0",
"auto_expand_replicas": "0-2",
"search.idle.after": "365d",
"sort.field": ["CloseTime", "StartTime", "RunId"],
"sort.order": ["desc", "desc", "desc"],
"sort.missing": ["_first", "_first", "_first"]
}
},
"mappings": {
"dynamic": "false",
"properties": {
"NamespaceId": {
"type": "keyword"
},
"TemporalNamespaceDivision": {
"type": "keyword"
},
"WorkflowId": {
"type": "keyword"
},
"RunId": {
"type": "keyword"
},
"WorkflowType": {
"type": "keyword"
},
"StartTime": {
"type": "date_nanos"
},
"ExecutionTime": {
"type": "date_nanos"
},
"CloseTime": {
"type": "date_nanos"
},
"ExecutionDuration": {
"type": "long"
},
"ExecutionStatus": {
"type": "keyword"
},
"TaskQueue": {
"type": "keyword"
},
"TemporalChangeVersion": {
"type": "keyword"
},
"BatcherNamespace": {
"type": "keyword"
},
"BatcherUser": {
"type": "keyword"
},
"BinaryChecksums": {
"type": "keyword"
},
"HistoryLength": {
"type": "long"
},
"StateTransitionCount": {
"type": "long"
},
"TemporalScheduledStartTime": {
"type": "date_nanos"
},
"TemporalScheduledById": {
"type": "keyword"
},
"TemporalSchedulePaused": {
"type": "boolean"
},
"HistorySizeBytes": {
"type": "long"
},
"BuildIds": {
"type": "keyword"
},
"ParentWorkflowId": {
"type": "keyword"
},
"ParentRunId": {
"type": "keyword"
},
"RootWorkflowId": {
"type": "keyword"
},
"RootRunId": {
"type": "keyword"
},
"TemporalPauseInfo": {
"type": "keyword"
},
"TemporalWorkerDeploymentVersion": {
"type": "keyword"
},
"TemporalWorkflowVersioningBehavior": {
"type": "keyword"
},
"TemporalWorkerDeployment": {
"type": "keyword"
},
"TemporalReportedProblems": {
"type": "keyword"
},
"TemporalUsedWorkerDeploymentVersions": {
"type": "keyword"
},
"TemporalExternalPayloadSizeBytes": {
"type": "long"
},
"TemporalExternalPayloadCount": {
"type": "long"
},
"TemporalBool01": {
"type": "boolean"
},
"TemporalBool02": {
"type": "boolean"
},
"TemporalDatetime01": {
"type": "date_nanos"
},
"TemporalDatetime02": {
"type": "date_nanos"
},
"TemporalDouble01": {
"type": "scaled_float",
"scaling_factor": 10000
},
"TemporalDouble02": {
"type": "scaled_float",
"scaling_factor": 10000
},
"TemporalInt01": {
"type": "long"
},
"TemporalInt02": {
"type": "long"
},
"TemporalKeyword01": {
"type": "keyword"
},
"TemporalKeyword02": {
"type": "keyword"
},
"TemporalKeyword03": {
"type": "keyword"
},
"TemporalKeyword04": {
"type": "keyword"
},
"TemporalLowCardinalityKeyword01": {
"type": "keyword"
},
"TemporalKeywordList01": {
"type": "keyword"
},
"TemporalKeywordList02": {
"type": "keyword"
}
}
},
"aliases": {}
}

57
schema/elasticsearch/visibility/versioned/v14/upgrade.sh Executable file
View File

@@ -0,0 +1,57 @@
#!/usr/bin/env bash
set -eu -o pipefail
# Prerequisites:
# - jq
# - curl
# Input parameters.
: "${ES_SCHEME:=http}"
: "${ES_SERVER:=127.0.0.1}"
: "${ES_PORT:=9200}"
: "${ES_USER:=}"
: "${ES_PWD:=}"
: "${ES_VERSION:=v7}"
: "${ES_VIS_INDEX_V1:=temporal_visibility_v1_dev}"
: "${AUTO_CONFIRM:=}"
: "${SLICES_COUNT:=auto}"
es_endpoint="${ES_SCHEME}://${ES_SERVER}:${ES_PORT}"
echo "=== Step 0. Sanity check if Elasticsearch index is accessible ==="
if ! curl --silent --fail --user "${ES_USER}":"${ES_PWD}" "${es_endpoint}/${ES_VIS_INDEX_V1}/_stats/docs" --write-out "\n"; then
echo "Elasticsearch index ${ES_VIS_INDEX_V1} is not accessible at ${es_endpoint}."
exit 1
fi
echo "=== Step 1. Add TemporalExternalPayloadSizeBytes and TemporalExternalPayloadCount builtin search attributes ==="
new_mapping='
{
"properties": {
"TemporalExternalPayloadSizeBytes": {
"type": "long"
},
"TemporalExternalPayloadCount": {
"type": "long"
}
}
}
'
if [ -z "${AUTO_CONFIRM}" ]; then
read -p "Add TemporalExternalPayloadSizeBytes and TemporalExternalPayloadCount builtin search attributes to the index ${ES_VIS_INDEX_V1}? (N/y)" -n 1 -r
echo
else
REPLY="y"
fi
if [ "${REPLY}" = "y" ]; then
curl --silent --fail --user "${ES_USER}":"${ES_PWD}" -X PUT "${es_endpoint}/${ES_VIS_INDEX_V1}/_mapping" -H "Content-Type: application/json" --data-binary "$new_mapping" | jq
# Wait for mapping changes to go through.
until curl --silent --user "${ES_USER}":"${ES_PWD}" "${es_endpoint}/_cluster/health/${ES_VIS_INDEX_V1}" | jq --exit-status '.status=="green" | .'; do
echo "Waiting for Elasticsearch index ${ES_VIS_INDEX_V1} become green."
sleep 1
done
fi

2
schema/embed.go
View File

@@ -58,7 +58,7 @@ func ElasticsearchClusterSettings() (string, error) {
// ElasticsearchIndexTemplate returns the embedded index template for Elasticsearch v7 (latest version)
func ElasticsearchIndexTemplate() (string, error) {
data, err := assets.ReadFile("elasticsearch/visibility/versioned/v13/index_template_v7.json")
data, err := assets.ReadFile("elasticsearch/visibility/versioned/v14/index_template_v7.json")
if err != nil {
return "", err
}

2
schema/mysql/v8/version.go
View File

@@ -6,4 +6,4 @@ package v8
const Version = "1.19"
// VisibilityVersion is the MySQL visibility database release version
const VisibilityVersion = "1.13"
const VisibilityVersion = "1.14"

6
schema/mysql/v8/visibility/schema.sql
View File

@@ -51,7 +51,8 @@ CREATE TABLE executions_visibility (
TemporalWorkflowVersioningBehavior VARCHAR(255) GENERATED ALWAYS AS (search_attributes->>"$.TemporalWorkflowVersioningBehavior"),
TemporalWorkerDeployment VARCHAR(255) GENERATED ALWAYS AS (search_attributes->>"$.TemporalWorkerDeployment"),
TemporalUsedWorkerDeploymentVersions JSON GENERATED ALWAYS AS (search_attributes->'$.TemporalUsedWorkerDeploymentVersions'),
TemporalExternalPayloadSizeBytes BIGINT GENERATED ALWAYS AS (search_attributes->"$.TemporalExternalPayloadSizeBytes"),
TemporalExternalPayloadCount BIGINT GENERATED ALWAYS AS (search_attributes->"$.TemporalExternalPayloadCount"),
PRIMARY KEY (namespace_id, run_id)
);
@@ -85,7 +86,8 @@ CREATE INDEX by_temporal_scheduled_start_time ON executions_visibility (namespac
CREATE INDEX by_temporal_scheduled_by_id ON executions_visibility (namespace_id, TemporalScheduledById, (COALESCE(close_time, CAST('9999-12-31 23:59:59' AS DATETIME))) DESC, start_time DESC, run_id);
CREATE INDEX by_temporal_schedule_paused ON executions_visibility (namespace_id, TemporalSchedulePaused, (COALESCE(close_time, CAST('9999-12-31 23:59:59' AS DATETIME))) DESC, start_time DESC, run_id);
CREATE INDEX by_temporal_namespace_division ON executions_visibility (namespace_id, TemporalNamespaceDivision, (COALESCE(close_time, CAST('9999-12-31 23:59:59' AS DATETIME))) DESC, start_time DESC, run_id);
CREATE INDEX by_temporal_external_payload_size_bytes ON executions_visibility (namespace_id, TemporalExternalPayloadSizeBytes, (COALESCE(close_time, CAST('9999-12-31 23:59:59' AS DATETIME))) DESC, start_time DESC, run_id);
CREATE INDEX by_temporal_external_payload_count ON executions_visibility (namespace_id, TemporalExternalPayloadCount, (COALESCE(close_time, CAST('9999-12-31 23:59:59' AS DATETIME))) DESC, start_time DESC, run_id);
CREATE TABLE custom_search_attributes (
namespace_id CHAR(64) NOT NULL,

25
schema/mysql/v8/visibility/versioned/v1.14/add_external_payload_size_and_count_search_attributes.sql Normal file
View File

@@ -0,0 +1,25 @@
ALTER TABLE executions_visibility
ADD COLUMN TemporalExternalPayloadSizeBytes BIGINT
GENERATED ALWAYS AS (search_attributes->"$.TemporalExternalPayloadSizeBytes");
ALTER TABLE executions_visibility
ADD COLUMN TemporalExternalPayloadCount BIGINT
GENERATED ALWAYS AS (search_attributes->"$.TemporalExternalPayloadCount");
CREATE INDEX by_temporal_external_payload_size_bytes
ON executions_visibility (
namespace_id,
TemporalExternalPayloadSizeBytes,
(COALESCE(close_time, CAST('9999-12-31 23:59:59' AS DATETIME))) DESC,
start_time DESC,
run_id
);
CREATE INDEX by_temporal_external_payload_count
ON executions_visibility (
namespace_id,
TemporalExternalPayloadCount,
(COALESCE(close_time, CAST('9999-12-31 23:59:59' AS DATETIME))) DESC,
start_time DESC,
run_id
);

8
schema/mysql/v8/visibility/versioned/v1.14/manifest.json Normal file
View File

@@ -0,0 +1,8 @@
{
"CurrVersion": "1.14",
"MinCompatibleVersion": "0.1",
"Description": "add TemporalExternalPayloadSizeBytes and TemporalExternalPayloadCount builtin search attributes",
"SchemaUpdateCqlFiles": [
"add_external_payload_size_and_count_search_attributes.sql"
]
}

2
schema/postgresql/v12/version.go
View File

@@ -8,4 +8,4 @@ const Version = "1.19"
// VisibilityVersion is the Postgres visibility database release version
// Temporal supports both MySQL and Postgres officially, so upgrade should be performed for both MySQL and Postgres
const VisibilityVersion = "1.13"
const VisibilityVersion = "1.14"

4
schema/postgresql/v12/visibility/schema.sql
View File

@@ -56,6 +56,8 @@ CREATE TABLE executions_visibility (
TemporalWorkflowVersioningBehavior VARCHAR(255) GENERATED ALWAYS AS (search_attributes->>'TemporalWorkflowVersioningBehavior') STORED,
TemporalWorkerDeployment VARCHAR(255) GENERATED ALWAYS AS (search_attributes->>'TemporalWorkerDeployment') STORED,
TemporalUsedWorkerDeploymentVersions JSONB GENERATED ALWAYS AS (search_attributes->'TemporalUsedWorkerDeploymentVersions') STORED,
TemporalExternalPayloadSizeBytes BIGINT GENERATED ALWAYS AS ((search_attributes->'TemporalExternalPayloadSizeBytes')::bigint) STORED,
TemporalExternalPayloadCount BIGINT GENERATED ALWAYS AS ((search_attributes->'TemporalExternalPayloadCount')::bigint) STORED,
-- Pre-allocated custom search attributes
Bool01 BOOLEAN GENERATED ALWAYS AS ((search_attributes->'Bool01')::boolean) STORED,
@@ -137,6 +139,8 @@ CREATE INDEX by_temporal_scheduled_start_time ON executions_visibility (namespac
CREATE INDEX by_temporal_scheduled_by_id ON executions_visibility (namespace_id, TemporalScheduledById, (COALESCE(close_time, '9999-12-31 23:59:59')) DESC, start_time DESC, run_id);
CREATE INDEX by_temporal_schedule_paused ON executions_visibility (namespace_id, TemporalSchedulePaused, (COALESCE(close_time, '9999-12-31 23:59:59')) DESC, start_time DESC, run_id);
CREATE INDEX by_temporal_namespace_division ON executions_visibility (namespace_id, TemporalNamespaceDivision, (COALESCE(close_time, '9999-12-31 23:59:59')) DESC, start_time DESC, run_id);
CREATE INDEX by_temporal_external_payload_size_bytes ON executions_visibility (namespace_id, TemporalExternalPayloadSizeBytes, (COALESCE(close_time, '9999-12-31 23:59:59')) DESC, start_time DESC, run_id);
CREATE INDEX by_temporal_external_payload_count ON executions_visibility (namespace_id, TemporalExternalPayloadCount, (COALESCE(close_time, '9999-12-31 23:59:59')) DESC, start_time DESC, run_id);
-- Indexes for the pre-allocated custom search attributes
CREATE INDEX by_bool_01 ON executions_visibility (namespace_id, Bool01, (COALESCE(close_time, '9999-12-31 23:59:59')) DESC, start_time DESC, run_id);

9
schema/postgresql/v12/visibility/versioned/v1.14/add_external_payload_size_and_count_search_attributes.sql Normal file
View File

@@ -0,0 +1,9 @@
ALTER TABLE executions_visibility
ADD COLUMN TemporalExternalPayloadSizeBytes BIGINT GENERATED ALWAYS AS ((search_attributes->'TemporalExternalPayloadSizeBytes')::bigint) STORED;
ALTER TABLE executions_visibility
ADD COLUMN TemporalExternalPayloadCount BIGINT GENERATED ALWAYS AS ((search_attributes->'TemporalExternalPayloadCount')::bigint) STORED;
CREATE INDEX by_temporal_external_payload_size_bytes ON executions_visibility (namespace_id, TemporalExternalPayloadSizeBytes, (COALESCE(close_time, '9999-12-31 23:59:59')) DESC, start_time DESC, run_id);
CREATE INDEX by_temporal_external_payload_count ON executions_visibility (namespace_id, TemporalExternalPayloadCount, (COALESCE(close_time, '9999-12-31 23:59:59')) DESC, start_time DESC, run_id);

8
schema/postgresql/v12/visibility/versioned/v1.14/manifest.json Normal file
View File

@@ -0,0 +1,8 @@
{
"CurrVersion": "1.14",
"MinCompatibleVersion": "0.1",
"Description": "add TemporalExternalPayloadSizeBytes and TemporalExternalPayloadCount builtin search attributes",
"SchemaUpdateCqlFiles": [
"add_external_payload_size_and_count_search_attributes.sql"
]
}

2
schema/sqlite/v3/visibility/schema.sql
View File

@@ -36,6 +36,8 @@ CREATE TABLE executions_visibility (
TemporalWorkflowVersioningBehavior VARCHAR(255) GENERATED ALWAYS AS (JSON_EXTRACT(search_attributes, "$.TemporalWorkflowVersioningBehavior")),
TemporalWorkerDeployment VARCHAR(255) GENERATED ALWAYS AS (JSON_EXTRACT(search_attributes, "$.TemporalWorkerDeployment")),
TemporalUsedWorkerDeploymentVersions TEXT GENERATED ALWAYS AS (JSON_EXTRACT(search_attributes, "$.TemporalUsedWorkerDeploymentVersions")) STORED,
TemporalExternalPayloadSizeBytes BIGINT GENERATED ALWAYS AS (JSON_EXTRACT(search_attributes, "$.TemporalExternalPayloadSizeBytes")) STORED,
TemporalExternalPayloadCount BIGINT GENERATED ALWAYS AS (JSON_EXTRACT(search_attributes, "$.TemporalExternalPayloadCount")) STORED,
-- Pre-allocated custom search attributes
Bool01 BOOLEAN GENERATED ALWAYS AS (JSON_EXTRACT(search_attributes, "$.Bool01")),