flow.proto

syntax = "proto3";

import "google/protobuf/timestamp.proto";
import "google/protobuf/descriptor.proto";
import "peers.proto";

extend google.protobuf.EnumValueOptions {
  // If the flow is in this state, then maintenance should wait for it to transition to another non-wait state
  // This is to make sure we have only either terminal states or running state which we can pause
  optional bool peerdb_maintenance_wait = 16551843;
}

package peerdb_flow;

message AlertInput {
  string flow_name = 1;
  string message = 2;
}

message TableNameMapping {
  string source_table_name = 1;
  string destination_table_name = 2;
}

message ColumnSetting {
  string source_name = 1;
  string destination_name = 2;
  string destination_type = 3;
  int32 ordering = 4;
  int32 partitioning = 6;
  bool nullable_enabled = 5;
}

message TableMapping {
  string source_table_identifier = 1;
  string destination_table_identifier = 2;
  string partition_key = 3;
  repeated string exclude = 4;
  repeated ColumnSetting columns = 5;
  TableEngine engine = 6;
  string sharding_key = 7;
  string policy_name = 8;
  string partition_by_expr = 9;
}

message SetupInput {
  map<string, string> env = 1;
  string flow_name = 2;
  string peer_name = 3;
}

// FlowConnectionConfigs is for external use by the API, maintaining backwards compatibility
// When adding fields here, add them to FlowConnectionConfigsCore too
message FlowConnectionConfigs {
  reserved 2,3,17;
  string flow_job_name = 1;

  // config for the CDC flow itself
  // currently, TableMappings, MaxBatchSize and IdleTimeoutSeconds are dynamic via Temporal signals
  repeated TableMapping table_mappings = 4;
  uint32 max_batch_size = 5;
  uint64 idle_timeout_seconds = 6;
  string cdc_staging_path = 7;
  string publication_name = 8;
  string replication_slot_name = 9;

  // config for the initial load feature, along with interactions like resync and initial_snapshot_only
  bool do_initial_snapshot = 10;
  uint32 snapshot_num_rows_per_partition = 11;
  uint32 snapshot_num_partitions_override = 26;
  string snapshot_staging_path = 12;
  // max parallel workers is per table
  uint32 snapshot_max_parallel_workers = 13;
  uint32 snapshot_num_tables_in_parallel = 14;
  // if true, then the flow will be resynced
  // create new tables with "_resync" suffix, perform initial load and then swap the new tables with the old ones
  // to only be used after the old mirror is dropped
  bool resync = 15;
  bool initial_snapshot_only = 16;

  // configurations for soft delete and synced at columns, affects both initial snapshot and CDC
  string soft_delete_col_name = 18;
  string synced_at_col_name = 19;

  string script = 20;
  TypeSystem system = 21;

  // source and destination peer
  string source_name = 22;
  string destination_name = 23;

  map<string, string> env = 24;
  uint32 version = 25;

  // Connector capabilities detected at flow creation time.
  // Used to control type mapping backwards compatibility.
  repeated string flags = 27;

  optional bool skip_validation = 28;
}

// FlowConnectionConfigsCore is used internally in the codebase, it is safe to remove (mark reserved) fields from it
// It still needs to be a proto because Temporal
// When adding fields here, add them to FlowConnectionConfigs too
// Equivalence with FlowConnectionConfigs is enforced during codegen
message FlowConnectionConfigsCore {
  reserved 2,3,17;
  string flow_job_name = 1;

  // config for the CDC flow itself
  // currently, TableMappings, MaxBatchSize and IdleTimeoutSeconds are dynamic via Temporal signals
  repeated TableMapping table_mappings = 4;
  uint32 max_batch_size = 5;
  uint64 idle_timeout_seconds = 6;
  string cdc_staging_path = 7;
  string publication_name = 8;
  string replication_slot_name = 9;

  // config for the initial load feature, along with interactions like resync and initial_snapshot_only
  bool do_initial_snapshot = 10;
  uint32 snapshot_num_rows_per_partition = 11;
  uint32 snapshot_num_partitions_override = 26;
  string snapshot_staging_path = 12;
  // max parallel workers is per table
  uint32 snapshot_max_parallel_workers = 13;
  uint32 snapshot_num_tables_in_parallel = 14;
  // if true, then the flow will be resynced
  // create new tables with "_resync" suffix, perform initial load and then swap the new tables with the old ones
  // to only be used after the old mirror is dropped
  bool resync = 15;
  bool initial_snapshot_only = 16;

  // configurations for soft delete and synced at columns, affects both initial snapshot and CDC
  string soft_delete_col_name = 18;
  string synced_at_col_name = 19;

  string script = 20;
  TypeSystem system = 21;

  // source and destination peer
  string source_name = 22;
  string destination_name = 23;

  map<string, string> env = 24;
  uint32 version = 25;

  // Connector capabilities detected at flow creation time.
  // Used to control type mapping backwards compatibility.
  repeated string flags = 27;

  optional bool skip_validation = 28;
}

message RenameTableOption {
  string current_name = 1;
  string new_name = 2;
}

message RenameTablesInput {
  reserved 2,4,5;
  string flow_job_name = 1;
  repeated RenameTableOption rename_table_options = 3;
  string peer_name = 6;
  string soft_delete_col_name = 7;
  string synced_at_col_name = 8;
}

message RemoveTablesFromRawTableInput {
  string flow_job_name = 1;
  repeated string destination_table_names = 2;
  int64 sync_batch_id = 3;
  int64 normalize_batch_id = 4;
}

message RenameTablesOutput {
  string flow_job_name = 1;
}

message CreateTablesFromExistingInput {
  string flow_job_name = 1;
  map<string, string> new_to_existing_table_mapping = 3;
  string peer_name = 4;
}

message CreateTablesFromExistingOutput {
  string flow_job_name = 2;
}

message SyncFlowOptions {
  reserved 5;
  uint32 batch_size = 1;
  uint64 idle_timeout_seconds = 3;
  map<uint32, string> src_table_id_name_mapping = 4;
  repeated TableMapping table_mappings = 6;
  int32 number_of_syncs = 7;
}

message EnsurePullabilityBatchInput {
  string flow_job_name = 2;
  repeated string source_table_identifiers = 3;
  bool check_constraints = 4;
  string peer_name = 5;
}

message PostgresTableIdentifier {
  uint32 rel_id = 1;
}

message EnsurePullabilityBatchOutput {
  map<string, PostgresTableIdentifier> table_identifier_mapping = 1;
}

message SetupReplicationInput {
  string flow_job_name = 2;
  map<string, string> table_name_mapping = 3;
  map<string, string> env = 4;

  // replicate to destination using ctid
  bool do_initial_snapshot = 5;
  string existing_publication_name = 6;
  string existing_replication_slot_name = 7;
  string peer_name = 8;
  string destination_name = 9;
}

message SetupReplicationOutput {
  reserved 3;
  string slot_name = 1;
  string snapshot_name = 2;
}

message CreateRawTableInput {
  string flow_job_name = 2;
  map<string, string> table_name_mapping = 3;
  string peer_name = 4;
}

message CreateRawTableOutput { string table_identifier = 1; }

// stored in catalog table table_schema_mapping, be wary of breaking changes
message TableSchema {
  string table_identifier = 1;
  repeated string primary_key_columns = 2;
  bool is_replica_identity_full = 3;
  TypeSystem system = 4;
  bool nullable_enabled = 5;
  repeated FieldDescription columns = 6;
  uint32 table_oid = 7;
}

message FieldDescription {
  string name = 1;
  string type = 2;
  int32 type_modifier = 3;
  bool nullable = 4;
  string type_schema_name = 5;
}

message SetupTableSchemaBatchInput {
  reserved 2;
  map<string, string> env = 1;
  string flow_name = 3;
  TypeSystem system = 4;
  string peer_name = 5;
  repeated TableMapping table_mappings = 6;
  uint32 version = 7;
}

message SetupNormalizedTableBatchInput {
  map<string, string> env = 1;
  repeated TableMapping table_mappings = 3;

  // migration related columns
  string soft_delete_col_name = 4;
  string synced_at_col_name = 5;
  string flow_name = 6;
  string peer_name = 7;
  bool is_resync = 8;
  uint32 version = 9;
  repeated string flags = 10;
}

message SetupNormalizedTableOutput {
  string table_identifier = 1;
  bool already_exists = 2;
}

message SetupNormalizedTableBatchOutput {
  map<string, bool> table_exists_mapping = 1;
}

// partition ranges [start, end] inclusive
message IntPartitionRange {
  int64 start = 1;
  int64 end = 2;
}

message TimestampPartitionRange {
  google.protobuf.Timestamp start = 1;
  google.protobuf.Timestamp end = 2;
}

message TID {
  uint32 block_number = 1;
  // Maps to pgtype.TID.OffsetNumber (uint16). Protobuf lacks uint16
  uint32 offset_number = 2;
}

message TIDPartitionRange {
  TID start = 1;
  TID end = 2;
}

message UIntPartitionRange {
  uint64 start = 1;
  uint64 end = 2;
}

message NullPartitionRange {
}

message ObjectIdPartitionRange {
  string start = 1;
  string end = 2;
}

message PartitionRange {
  // can be a timestamp range or an integer range
  oneof range {
    IntPartitionRange int_range = 1;
    TimestampPartitionRange timestamp_range = 2;
    TIDPartitionRange tid_range = 3;
    UIntPartitionRange uint_range = 4;
    ObjectIdPartitionRange object_id_range = 5;
    NullPartitionRange null_range = 6;
  }
}

enum TableEngine {
  CH_ENGINE_REPLACING_MERGE_TREE = 0;
  CH_ENGINE_MERGE_TREE = 1;
  CH_ENGINE_NULL = 2;
  CH_ENGINE_REPLICATED_REPLACING_MERGE_TREE = 3;
  CH_ENGINE_REPLICATED_MERGE_TREE = 4;
  CH_ENGINE_COALESCING_MERGE_TREE = 5;
}

// protos for qrep
enum QRepWriteType {
  QREP_WRITE_MODE_APPEND = 0;
  QREP_WRITE_MODE_UPSERT = 1;
  // only valid when initial_copy_true is set to true. TRUNCATES tables before reverting to APPEND.
  QREP_WRITE_MODE_OVERWRITE = 2;
}

message QRepWriteMode {
  QRepWriteType write_type = 1;
  repeated string upsert_key_columns = 2;
}

enum TypeSystem {
  Q = 0;
  PG = 1;
}

message QRepConfig {
  reserved 2, 3;
  string flow_job_name = 1;

  string destination_table_identifier = 4;

  // If present, the query provided by the user. If empty, a full-table query determined by the connector.
  string query = 5;

  string watermark_table = 6;
  string watermark_column = 7;

  bool initial_copy_only = 8;

  uint32 max_parallel_workers = 9;

  // time to wait between getting partitions to process
  uint32 wait_between_batches_seconds = 10;

  QRepWriteMode write_mode = 11;

  // The location where the avro files will be written
  // if this starts with gs:// then it will be written to GCS
  // if this starts with s3:// then it will be written to S3, only supported in Snowflake
  // if nothing is specified then it will be written to local disk
  // if using GCS or S3 make sure your instance has the correct permissions.
  string staging_path = 12;

  uint32 num_rows_per_partition = 13;
  // overrides num_rows_per_partition to skip count(*) query
  uint32 num_partitions_override = 29;

  // Creates the watermark table on the destination as-is, can be used for some queries.
  bool setup_watermark_table_on_destination = 14;

  // create new tables with "_peerdb_resync" suffix, perform initial load and then swap the new table with the old ones
  // to be used after the old mirror is dropped
  bool dst_table_full_resync = 15;

  string synced_at_col_name = 16;
  string soft_delete_col_name = 17;

  TypeSystem system = 18;
  string script = 19;

  string source_name = 20;
  string destination_name = 21;
  string snapshot_name = 23;

  map<string, string> env = 24;

  string parent_mirror_name = 25; // internal
  repeated string exclude = 26;

  repeated ColumnSetting columns = 27;
  uint32 version = 28; // internal

  peerdb_peers.DBType source_type = 30;

  // Connector capabilities detected at flow creation time.
  // Used to control type mapping backwards compatibility.
  repeated string flags = 31; // internal
  // if true, then a separate null partition will be created for rows with null values in the watermark column
  bool add_null_partition = 32; // internal
}

message ChildTableRange {
  string table = 1;
  uint32 start = 2;
  uint32 end = 3;
}

message QRepPartition {
  string partition_id = 2;
  PartitionRange range = 3;
  bool full_table_partition = 4;
  // For partitioned tables with CTID partitioning: each entry specifies a
  // child table and block range to query instead of the parent watermark table.
  // When populated, the top-level `range` field is unused.
  repeated ChildTableRange child_table_ranges = 5;
}

message QRepPartitionBatch {
  int32 batch_id = 1;
  repeated QRepPartition partitions = 2;
}

message QRepParitionResult {
  repeated QRepPartition partitions = 1;
}

message DropFlowInput {
  reserved 2,3;
  string flow_job_name = 1;
  bool drop_flow_stats = 4;
  FlowConnectionConfigsCore flow_connection_configs = 5;
  bool skip_destination_drop = 6;
  bool skip_source_drop = 7;
  bool resync = 8;
}

message TableSchemaDelta {
  string src_table_name = 1;
  string dst_table_name = 2;
  repeated FieldDescription added_columns = 3;
  TypeSystem system = 4;
  bool nullable_enabled = 5;
}

message QRepFlowState {
  reserved 4;
  QRepPartition last_partition = 1;
  uint64 num_partitions_processed = 2;
  bool needs_resync = 3;
  FlowStatus current_flow_status = 5;
}

message PeerDBColumns {
  reserved 3;
  string soft_delete_col_name = 1;
  string synced_at_col_name = 2;
}

message GetOpenConnectionsForUserResult {
  string user_name = 1;
  int64 current_open_connections = 2;
}

// UI reads current workflow status and also requests status changes using same enum
// see flow/cmd/handler.go FlowStateChange
enum FlowStatus {
  STATUS_UNKNOWN = 0 [(peerdb_maintenance_wait) = true];
  STATUS_RUNNING = 1;
  STATUS_PAUSED = 2;
  STATUS_PAUSING = 3 [(peerdb_maintenance_wait) = true];;
  // not reachable in QRep mirrors
  STATUS_SETUP = 4 [(peerdb_maintenance_wait) = true];
  // not reachable in QRep mirrors
  STATUS_SNAPSHOT = 5 [(peerdb_maintenance_wait) = true];;
  STATUS_TERMINATING = 6;
  STATUS_TERMINATED = 7;
  STATUS_COMPLETED = 8;
  STATUS_RESYNC = 9 [(peerdb_maintenance_wait) = true];;
  STATUS_FAILED = 10;
  STATUS_MODIFYING = 11 [(peerdb_maintenance_wait) = true];;
}

message CDCFlowConfigUpdate {
  repeated TableMapping additional_tables = 1;
  uint32 batch_size = 2;
  uint64 idle_timeout = 3;
  int32 number_of_syncs = 4;
  repeated TableMapping removed_tables = 5;
  // updates keys in the env map, existing keys left unchanged
  map<string, string> updated_env = 6;
  uint32 snapshot_num_rows_per_partition = 7;
  uint32 snapshot_num_partitions_override = 10;
  uint32 snapshot_max_parallel_workers = 8;
  uint32 snapshot_num_tables_in_parallel = 9;
  bool skip_initial_snapshot_for_table_additions = 11;
}

message QRepFlowConfigUpdate {
}

message FlowConfigUpdate {
  oneof update {
    CDCFlowConfigUpdate cdc_flow_config_update = 1;
    QRepFlowConfigUpdate qrep_flow_config_update = 2;
  }
}

message SetupFlowOutput {
  map<uint32, string> src_table_id_name_mapping = 1;
}

message AddTablesToPublicationInput {
  string flow_job_name = 1;
  string publication_name = 2;
  repeated TableMapping additional_tables = 3;
}

message RemoveTablesFromPublicationInput {
  string flow_job_name = 1;
  string publication_name = 2;
  repeated TableMapping tables_to_remove = 3;
}

message IsQRepPartitionSyncedInput {
  string flow_job_name = 1;
  string partition_id = 2;
}

message ExportTxSnapshotOutput {
  reserved 2;
  string snapshot_name = 1;
}

enum DynconfValueType {
  UNKNOWN = 0;
  STRING = 1;
  INT = 2;
  UINT = 3;
  BOOL = 4;
}

enum DynconfApplyMode {
  APPLY_MODE_UNKNOWN = 0;
  // should apply immediately
  APPLY_MODE_IMMEDIATE = 1;
  // should apply after the mirror is paused and resumed
  APPLY_MODE_AFTER_RESUME = 2;
  // should apply after pod is restarted
  APPLY_MODE_RESTART = 3;
  // only applies to newly created mirrors
  APPLY_MODE_NEW_MIRROR = 4;
}

enum DynconfTarget {
  ALL = 0;
  BIGQUERY = 1;
  SNOWFLAKE = 2;
  CLICKHOUSE = 3;
  QUEUES = 4;
  POSTGRES = 5;
  MYSQL = 6;
}

message DropFlowActivityInput {
  string flow_job_name = 1;
  string peer_name = 2;
}

message StartMaintenanceFlowInput {
}

message StartMaintenanceFlowOutput {
  string version = 1;
}

message StartMaintenanceSignal {
  repeated string skipped_snapshot_wait_flows = 1;
}

message EndMaintenanceFlowInput {
}

message EndMaintenanceFlowOutput {
  string version = 1;
}

message MaintenanceMirror {
  int64 mirror_id = 1;
  string mirror_name = 2;
  string workflow_id = 3;
  bool is_cdc = 4;
  google.protobuf.Timestamp mirror_created_at = 5;
  google.protobuf.Timestamp mirror_updated_at = 6;
}

message MaintenanceMirrors {
  repeated MaintenanceMirror mirrors = 1;
}

message PeerContextMetadata {
  string name = 1;
  peerdb_peers.DBType type = 2;
  peerdb_peers.DatabaseVariant variant = 3;
  string hostname = 4;
}

message FlowContextMetadataInput {
  string flow_name = 1;
  string source_name = 2;
  string destination_name = 3;
  FlowStatus status = 4;
  bool is_resync = 5;
  bool fetch_source_variant = 6;
}

enum FlowOperation {
  FLOW_OPERATION_UNKNOWN = 0;
  FLOW_OPERATION_SYNC = 1;
  FLOW_OPERATION_NORMALIZE = 2;
}

// FlowContextMetadata has contextual information of a flow and is universal at the flow level, it cannot be different for children context
// it is referenced via pointer
message FlowContextMetadata{
  string flow_name = 1;
  PeerContextMetadata source = 2;
  PeerContextMetadata destination = 3;
  FlowStatus status = 4;
  bool is_resync = 5;
  map<string, string> tags = 6;
}

message AdditionalContextMetadata{
  FlowOperation operation = 1;
}

message GetDefaultPartitionKeyForTablesInput {
  repeated TableMapping table_mappings = 1;
}

message GetDefaultPartitionKeyForTablesOutput {
  map<string, string> table_default_partition_key_mapping = 1;
}

message GetFlowInfoToCancelFromCatalogOutput {
  FlowConnectionConfigsCore flow_connection_configs = 1;
  string workflow_id = 2;
  peerdb_peers.DBType source_peer_type = 3;
}

message RunPgDumpSchemaInput {
  string source_name = 1;
  string destination_name = 2;
  string flow_name = 3;
  map<string, string> env = 4;
}