mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
<!-- Describe what has changed in this PR --> **What changed?** I defined the persistence.QueueV2 interface, which supersedes the persistence.Queue interface. I also added an implementation of this interface for Cassandra. The implementations for delete and merge are left out for now and will be implemented later. As a result, there is no logic to read/write from the queues table (to read/write ack levels). Later, the ReadMessages method will need to be updated to use this as the minimum message id. <!-- Tell your future self why have you made these changes --> **Why?** We plan on using this for the upcoming history task DLQ project. <!-- How have you verified this change? Tested locally? Added a unit test? Checked in staging env? --> **How did you test it?** The new code has 100% test coverage, even including a test for CAS application failures due to concurrent writers. I tested that particular one locally 100 times to verify that it wasn't flaky. <!-- Assuming the worst case, what can be broken when deploying this change to production? --> **Potential risks** The biggest risk IMO is that because we're doing this piecemeal, we may realize there's an issue in the database schema and then have to change it later, which will be a pain if we need to ensure it's backwards compatible, but we can prevent this by ensuring that those changes are not split across multiple releases. <!-- Is this PR a hotfix candidate or require that a notification be sent to the broader community? (Yes/No) --> **Is hotfix candidate?** No.
212 lines
8.2 KiB
Cassandra CQL
212 lines
8.2 KiB
Cassandra CQL
CREATE TYPE serialized_event_batch (
|
|
encoding_type text,
|
|
version int,
|
|
data blob
|
|
);
|
|
|
|
CREATE TABLE executions (
|
|
shard_id int,
|
|
type int, -- enum RowType { Shard, Execution, TransferTask, TimerTask, ReplicationTask, VisibilityTask}
|
|
namespace_id uuid,
|
|
workflow_id text,
|
|
run_id uuid,
|
|
current_run_id uuid,
|
|
visibility_ts timestamp, -- unique identifier for timer tasks for an execution
|
|
task_id bigint, -- unique identifier for transfer and timer tasks for an execution
|
|
shard blob,
|
|
shard_encoding text,
|
|
execution blob,
|
|
execution_encoding text,
|
|
execution_state blob,
|
|
execution_state_encoding text,
|
|
transfer blob,
|
|
transfer_encoding text,
|
|
replication blob,
|
|
replication_encoding text,
|
|
timer blob,
|
|
timer_encoding text,
|
|
visibility_task_data blob,
|
|
visibility_task_encoding text,
|
|
task_data blob,
|
|
task_encoding text,
|
|
next_event_id bigint, -- This is needed to make conditional updates on session history
|
|
range_id bigint, -- Increasing sequence identifier for transfer queue, checkpointed into shard info
|
|
activity_map map<bigint, blob>,
|
|
activity_map_encoding text,
|
|
timer_map map<text, blob>,
|
|
timer_map_encoding text,
|
|
child_executions_map map<bigint, blob>,
|
|
child_executions_map_encoding text,
|
|
request_cancel_map map<bigint, blob>,
|
|
request_cancel_map_encoding text,
|
|
signal_map map<bigint, blob>,
|
|
signal_map_encoding text,
|
|
signal_requested set<uuid>,
|
|
buffered_events_list list<frozen<serialized_event_batch>>,
|
|
workflow_last_write_version bigint,
|
|
workflow_state int,
|
|
checksum blob,
|
|
checksum_encoding text,
|
|
db_record_version bigint,
|
|
PRIMARY KEY (shard_id, type, namespace_id, workflow_id, run_id, visibility_ts, task_id)
|
|
) WITH COMPACTION = {
|
|
'class': 'org.apache.cassandra.db.compaction.LeveledCompactionStrategy'
|
|
};
|
|
|
|
CREATE TABLE history_node (
|
|
tree_id uuid, -- run_id if no reset, otherwise run_id of first run
|
|
branch_id uuid, -- changes in case of reset workflow. Conflict resolution can also change branch id.
|
|
node_id bigint, -- == first eventID in a batch of events
|
|
txn_id bigint, -- in case of multiple transactions on same node, we utilize highest transaction ID. Unique.
|
|
prev_txn_id bigint, -- point to the previous node: event chaining
|
|
data blob, -- batch of workflow execution history events as a blob
|
|
data_encoding text, -- protocol used for history serialization
|
|
PRIMARY KEY ((tree_id), branch_id, node_id, txn_id )
|
|
) WITH CLUSTERING ORDER BY (branch_id ASC, node_id ASC, txn_id DESC)
|
|
AND COMPACTION = {
|
|
'class': 'org.apache.cassandra.db.compaction.LeveledCompactionStrategy'
|
|
};
|
|
|
|
CREATE TABLE history_tree (
|
|
tree_id uuid,
|
|
branch_id uuid,
|
|
branch blob,
|
|
branch_encoding text,
|
|
PRIMARY KEY ((tree_id), branch_id )
|
|
) WITH COMPACTION = {
|
|
'class': 'org.apache.cassandra.db.compaction.LeveledCompactionStrategy'
|
|
};
|
|
|
|
-- Stores activity or workflow tasks
|
|
CREATE TABLE tasks (
|
|
namespace_id uuid,
|
|
task_queue_name text,
|
|
task_queue_type int, -- enum TaskQueueType {ActivityTask, WorkflowTask}
|
|
type int, -- enum rowType {Task, TaskQueue}
|
|
task_id bigint, -- unique identifier for tasks, monotonically increasing
|
|
range_id bigint, -- Used to ensure that only one process can write to the table
|
|
task blob,
|
|
task_encoding text,
|
|
task_queue blob,
|
|
task_queue_encoding text,
|
|
PRIMARY KEY ((namespace_id, task_queue_name, task_queue_type), type, task_id)
|
|
) WITH COMPACTION = {
|
|
'class': 'org.apache.cassandra.db.compaction.LeveledCompactionStrategy'
|
|
};
|
|
|
|
-- Stores task queue information such as user provided versioning data
|
|
-- OR
|
|
-- Used as a mapping from build id to task queue
|
|
CREATE TABLE task_queue_user_data (
|
|
namespace_id uuid,
|
|
task_queue_name text,
|
|
build_id text, -- If this row is used as a mapping of build id to task queue, this will not be empty
|
|
data blob, -- temporal.server.api.persistence.v1.TaskQueueUserData
|
|
data_encoding text, -- Encoding type used for serialization, in practice this should always be proto3
|
|
version bigint, -- Version of this row, used for optimistic concurrency
|
|
-- task_queue_name is not a part of the parititioning key to allow cheaply iterating all task queues in a single
|
|
-- namespace. Access to this table should be infrequent enough that a single partition per namespace can be used.
|
|
-- Note that this imposes a limit on total task queue user data within one namespace (see the relevant single
|
|
-- partition Cassandra limits).
|
|
PRIMARY KEY ((namespace_id), build_id, task_queue_name)
|
|
) WITH COMPACTION = {
|
|
'class': 'org.apache.cassandra.db.compaction.LeveledCompactionStrategy'
|
|
};
|
|
|
|
-- this table is only used for storage of mapping of namespace uuid to namespace name
|
|
CREATE TABLE namespaces_by_id (
|
|
id uuid,
|
|
name text,
|
|
PRIMARY KEY (id)
|
|
) WITH COMPACTION = {
|
|
'class': 'org.apache.cassandra.db.compaction.LeveledCompactionStrategy'
|
|
};
|
|
|
|
CREATE TABLE namespaces (
|
|
namespaces_partition int,
|
|
name text,
|
|
id uuid,
|
|
detail blob,
|
|
detail_encoding text,
|
|
is_global_namespace boolean,
|
|
notification_version bigint,
|
|
PRIMARY KEY (namespaces_partition, name)
|
|
) WITH COMPACTION = {
|
|
'class': 'org.apache.cassandra.db.compaction.LeveledCompactionStrategy'
|
|
};
|
|
|
|
|
|
CREATE TABLE queue_metadata (
|
|
queue_type int,
|
|
cluster_ack_level map<text, bigint>, -- TODO remove once migration is finished
|
|
data blob,
|
|
data_encoding text,
|
|
version bigint,
|
|
PRIMARY KEY (queue_type)
|
|
) WITH COMPACTION = {
|
|
'class': 'org.apache.cassandra.db.compaction.LeveledCompactionStrategy'
|
|
};
|
|
|
|
CREATE TABLE queue (
|
|
queue_type int,
|
|
message_id bigint,
|
|
message_payload blob,
|
|
message_encoding text,
|
|
PRIMARY KEY (queue_type, message_id)
|
|
) WITH COMPACTION = {
|
|
'class': 'org.apache.cassandra.db.compaction.LeveledCompactionStrategy'
|
|
};
|
|
|
|
CREATE TABLE cluster_metadata_info (
|
|
metadata_partition int,
|
|
cluster_name text,
|
|
data blob,
|
|
data_encoding text,
|
|
version bigint,
|
|
PRIMARY KEY (metadata_partition, cluster_name)
|
|
) WITH COMPACTION = {
|
|
'class': 'org.apache.cassandra.db.compaction.LeveledCompactionStrategy'
|
|
};
|
|
|
|
CREATE TABLE cluster_membership
|
|
(
|
|
membership_partition tinyint,
|
|
host_id uuid,
|
|
rpc_address inet,
|
|
rpc_port smallint,
|
|
role tinyint,
|
|
session_start timestamp,
|
|
last_heartbeat timestamp,
|
|
PRIMARY KEY (membership_partition, role, host_id)
|
|
) WITH COMPACTION = {
|
|
'class': 'org.apache.cassandra.db.compaction.LeveledCompactionStrategy'
|
|
};
|
|
|
|
CREATE INDEX cm_lastheartbeat_idx on cluster_membership (last_heartbeat);
|
|
CREATE INDEX cm_sessionstart_idx on cluster_membership (session_start);
|
|
|
|
CREATE TABLE queues
|
|
(
|
|
queue_type int,
|
|
queue_name text,
|
|
metadata_payload blob,
|
|
metadata_encoding text,
|
|
version bigint,
|
|
PRIMARY KEY ((queue_type, queue_name))
|
|
) WITH COMPACTION = {
|
|
'class': 'org.apache.cassandra.db.compaction.LeveledCompactionStrategy'
|
|
};
|
|
|
|
CREATE TABLE queue_messages
|
|
(
|
|
queue_type int,
|
|
queue_name text,
|
|
queue_partition int,
|
|
message_id bigint,
|
|
message_payload blob,
|
|
message_encoding text,
|
|
PRIMARY KEY ((queue_type, queue_name, queue_partition), message_id)
|
|
) WITH COMPACTION = {
|
|
'class': 'org.apache.cassandra.db.compaction.LeveledCompactionStrategy'
|
|
};
|