use reifydb_value::{error::Diagnostic, fragment::Fragment, value::value_type::ValueType};
pub fn flow_error(message: String) -> Diagnostic {
Diagnostic {
code: "FLOW_001".to_string(),
rql: None,
message: format!("Flow processing error: {}", message),
column: None,
fragment: Fragment::None,
label: None,
help: Some("Check view flow configuration".to_string()),
notes: vec![],
cause: None,
operator_chain: None,
}
}
pub fn flow_already_registered(flow_id: u64) -> Diagnostic {
Diagnostic {
code: "FLOW_003".to_string(),
rql: None,
message: format!("Flow {} is already registered", flow_id),
column: None,
fragment: Fragment::None,
label: None,
help: Some("Each flow can only be registered once. Check if the flow is already active.".to_string()),
notes: vec![],
cause: None,
operator_chain: None,
}
}
pub fn flow_version_corrupted(flow_id: u64, byte_count: usize) -> Diagnostic {
Diagnostic {
code: "FLOW_004".to_string(),
rql: None,
message: format!(
"Flow {} version data is corrupted: expected 8 bytes, found {} bytes",
flow_id, byte_count
),
column: None,
fragment: Fragment::None,
label: None,
help: Some("The flow version stored in the catalog is corrupted. \
This may indicate data corruption or a shape migration issue. \
Try dropping and recreating the flow."
.to_string()),
notes: vec![],
cause: None,
operator_chain: None,
}
}
pub fn flow_backfill_timeout(flow_id: u64, timeout_secs: u64) -> Diagnostic {
Diagnostic {
code: "FLOW_005".to_string(),
rql: None,
message: format!(
"Timeout waiting for flow {} backfill to complete after {} seconds",
flow_id, timeout_secs
),
column: None,
fragment: Fragment::None,
label: None,
help: Some("The flow backfill operation did not complete within the timeout period. \
This may indicate a large dataset, slow queries, or resource constraints. \
Try increasing the timeout or check for performance issues."
.to_string()),
notes: vec![],
cause: None,
operator_chain: None,
}
}
pub fn flow_dispatcher_unavailable() -> Diagnostic {
Diagnostic {
code: "FLOW_006".to_string(),
rql: None,
message: "Flow dispatcher is unavailable (channel closed)".to_string(),
column: None,
fragment: Fragment::None,
label: None,
help: Some("The flow dispatcher task has stopped or crashed. \
This may occur during shutdown or if the dispatcher encountered a fatal error. \
Check dispatcher logs for details."
.to_string()),
notes: vec![],
cause: None,
operator_chain: None,
}
}
pub fn flow_remote_source_unsupported() -> Diagnostic {
Diagnostic {
code: "FLOW_007".to_string(),
rql: None,
message: "Cannot create flow for remote source".to_string(),
column: None,
fragment: Fragment::None,
label: None,
help: Some("Remote tables do not support local flow graphs. Use remote subscription proxying instead."
.to_string()),
notes: vec![],
cause: None,
operator_chain: None,
}
}
pub fn flow_window_timestamp_column_not_found(column: &str) -> Diagnostic {
Diagnostic {
code: "FLOW_009".to_string(),
rql: None,
message: format!("Window timestamp column '{}' not found in input data", column),
column: None,
fragment: Fragment::None,
label: None,
help: Some(format!(
"The window operator is configured with ts: \"{}\" but no column with that name exists in the source table. \
Check the column name in the window WITH clause.",
column
)),
notes: vec![],
cause: None,
operator_chain: None,
}
}
pub fn flow_window_timestamp_column_type_mismatch(column: &str, found: ValueType) -> Diagnostic {
Diagnostic {
code: "FLOW_010".to_string(),
rql: None,
message: format!("Window timestamp column '{}' has type {:?}, expected DateTime", column, found),
column: None,
fragment: Fragment::None,
label: None,
help: Some("The timestamp column must be of type DateTime. \
If you have epoch milliseconds, convert with datetime::from_epoch_millis(column)."
.to_string()),
notes: vec![],
cause: None,
operator_chain: None,
}
}
pub fn flow_source_required() -> Diagnostic {
Diagnostic {
code: "FLOW_008".to_string(),
rql: None,
message: "Flow requires at least one source".to_string(),
column: None,
fragment: Fragment::None,
label: None,
help: Some("A view flow must read from a table, view, ring buffer, or series. \
Inline data (FROM [...]) cannot be used as the source for a view."
.to_string()),
notes: vec![],
cause: None,
operator_chain: None,
}
}
pub fn flow_window_requires_a_timed_source() -> Diagnostic {
Diagnostic {
code: "FLOW_014".to_string(),
rql: None,
message: "a windowed view requires a source that declares a time domain".to_string(),
column: None,
fragment: Fragment::None,
label: None,
help: Some("declare `time: event(<column>)` or `time: processing` on the source, \
or drop the window from the view"
.to_string()),
notes: vec![
"an undeclared source is `time: none`, whose rows carry no #time".to_string(),
"a window buckets and seals on #time, so it can never fire over rows that have none"
.to_string(),
],
cause: None,
operator_chain: None,
}
}
pub fn flow_sort_must_be_terminal() -> Diagnostic {
Diagnostic {
code: "FLOW_012".to_string(),
rql: None,
message: "sort is only supported as the final operator in a view".to_string(),
column: None,
fragment: Fragment::None,
label: None,
help: Some(
"Move the sort to the end of the pipeline so its output is not consumed by another operator. \
A view may sort its result, but cannot apply further operators after a sort."
.to_string(),
),
notes: vec![],
cause: None,
operator_chain: None,
}
}
pub fn flow_unsupported_aggregate_expression(output: &str) -> Diagnostic {
Diagnostic {
code: "FLOW_013".to_string(),
rql: None,
message: format!("aggregate output '{}' is not a supported aggregate expression in a view", output),
column: None,
fragment: Fragment::None,
label: None,
help: Some("Window and aggregate views support math::count, math::sum, math::avg, math::min and \
math::max over a column or scalar expression, optionally combined with arithmetic \
(for example math::max(x) - math::min(x)). Every output must reduce to such an aggregate."
.to_string()),
notes: vec![],
cause: None,
operator_chain: None,
}
}
pub fn flow_window_span_unavailable(output: &str, kind: &str) -> Diagnostic {
Diagnostic {
code: "FLOW_015".to_string(),
rql: None,
message: format!(
"aggregate output '{}' needs a window boundary, but a {} window has none",
output, kind
),
column: None,
fragment: Fragment::None,
label: None,
help: Some("window::start, window::end and window::duration are available on tumbling, sliding \
and session windows sized by a duration. A rolling window and a window sized by a row \
count have no boundary; window::last works on every window kind."
.to_string()),
notes: vec![],
cause: None,
operator_chain: None,
}
}
pub fn flow_supervisor_stopped() -> Diagnostic {
Diagnostic {
code: "FLOW_020".to_string(),
rql: None,
message: "flow supervisor actor has stopped".to_string(),
column: None,
fragment: Fragment::None,
label: None,
help: Some("The flow supervisor actor is no longer reachable and cannot consume CDC. \
This typically occurs during shutdown or after a supervisor panic."
.to_string()),
notes: vec![],
cause: None,
operator_chain: None,
}
}
fn flow_diagnostic(code: &str, message: String, help: &str) -> Diagnostic {
Diagnostic {
code: code.to_string(),
rql: None,
message,
column: None,
fragment: Fragment::None,
label: None,
help: Some(help.to_string()),
notes: vec![],
cause: None,
operator_chain: None,
}
}
pub fn flow_state_encode_failed(state: &str, cause: String) -> Diagnostic {
flow_diagnostic(
"FLOW_021",
format!("failed to serialize flow operator state '{}': {}", state, cause),
"An operator failed to encode its persistent state. This usually indicates a bug in the operator's \
state serialization, not user input.",
)
}
pub fn flow_state_decode_failed(state: &str, cause: String) -> Diagnostic {
flow_diagnostic(
"FLOW_022",
format!("failed to deserialize flow operator state '{}': {}", state, cause),
"An operator failed to decode its persistent state. This may indicate on-disk state corruption or a \
state-format change between versions.",
)
}
pub fn flow_unsupported_operator(kind: &str) -> Diagnostic {
flow_diagnostic(
"FLOW_023",
format!("operator kind '{}' is not supported in persistent flows", kind),
"This operator kind cannot appear in a persistent view flow. Rewrite the view without it.",
)
}
pub fn flow_operator_input_arity(operator: &str, expected: &str, found: usize) -> Diagnostic {
flow_diagnostic(
"FLOW_024",
format!("operator '{}' requires {} inputs, but the DAG provided {}", operator, expected, found),
"The compiled flow DAG has the wrong number of input edges for this operator. This indicates a flow \
compiler or catalog inconsistency.",
)
}
pub fn flow_parent_operator_not_found(input: String) -> Diagnostic {
flow_diagnostic(
"FLOW_025",
format!("parent operator not found while wiring operator input: {}", input),
"An operator references a parent operator that has not been registered. The flow DAG is incomplete \
or operators were registered out of order.",
)
}
pub fn flow_unknown_operator(operator: &str) -> Diagnostic {
flow_diagnostic(
"FLOW_026",
format!("unknown flow operator '{}'", operator),
"The flow references an operator that is not registered in this build. Check for a missing extern \
operator or a typo in the operator name.",
)
}
pub fn flow_extern_unsupported_on_wasm() -> Diagnostic {
flow_diagnostic(
"FLOW_027",
"extern operators are not supported on the wasm target".to_string(),
"Extern operators cannot be loaded in a wasm runtime. Use only built-in operators.",
)
}
pub fn flow_missing_input_edge() -> Diagnostic {
flow_diagnostic(
"FLOW_028",
"operator is missing a required input edge; the flow DAG is incomplete".to_string(),
"The compiled flow DAG is missing an edge that a operator requires. This indicates a flow compiler bug.",
)
}
pub fn flow_unknown_diff_origin(operator: &str, origin: Option<String>) -> Diagnostic {
let message = match origin {
Some(o) => format!("{} operator received a diff from an unknown operator: {}", operator, o),
None => format!("{} operator received a diff from an unknown operator", operator),
};
flow_diagnostic(
"FLOW_029",
message,
"An operator received change data tagged with an origin it does not have wired as an input. This \
indicates a flow routing or DAG inconsistency.",
)
}
pub fn extern_abi_tag_mismatch(plugin: u32, host: u32) -> Diagnostic {
flow_diagnostic(
"FLOW_031",
format!("extern operator ABI tag mismatch: plugin reports {:#06x}, host expects {:#06x}", plugin, host),
"The extern operator library was built against a different ABI than this host. Rebuild the extern \
operators against the current version.",
)
}
pub fn extern_library_not_loaded(path: &str) -> Diagnostic {
flow_diagnostic(
"FLOW_032",
format!("extern operator library not loaded: {}", path),
"The extern operator shared library could not be loaded. Check that the .so exists and is readable.",
)
}
pub fn extern_symbol_not_found(symbol: &str, cause: String) -> Diagnostic {
flow_diagnostic(
"FLOW_033",
format!("extern operator symbol '{}' not found: {}", symbol, cause),
"The extern operator library is missing an expected symbol. It may be built against a different ABI \
or be the wrong library.",
)
}
pub fn extern_operator_not_found(operator: &str) -> Diagnostic {
flow_diagnostic(
"FLOW_034",
format!("extern operator '{}' not found", operator),
"No loaded extern library provides this operator. Check the operators directory and the operator name.",
)
}
pub fn extern_create_failed(cause: String) -> Diagnostic {
flow_diagnostic(
"FLOW_035",
format!("failed to create extern operator: {}", cause),
"The extern operator's create function returned an error. See the underlying cause.",
)
}
pub fn flow_sink_missing_system_column(column: &str, row_idx: usize) -> Diagnostic {
flow_diagnostic(
"FLOW_036",
format!("row at index {} is missing the '{}' system column", row_idx, column),
"A view sink row is missing a required system timestamp column. This indicates an encoding bug \
upstream of the sink.",
)
}
pub fn flow_sink_dictionary_not_found(dictionary_id: String, column: &str) -> Diagnostic {
flow_diagnostic(
"FLOW_037",
format!("dictionary {} not found for view column '{}'", dictionary_id, column),
"A dictionary-encoded view column references a dictionary that no longer exists in the catalog.",
)
}
pub fn flow_sink_not_a_source_family(family: &str) -> Diagnostic {
flow_diagnostic(
"FLOW_047",
format!("a view sink cannot encode a row of the {} family", family),
"Only the table, series and ring buffer families reserve the created_at, updated_at and #time \
slots a view sink stamps. This indicates the sink was built over the wrong shape.",
)
}
pub fn flow_sink_missing_series_key(view: &str, column: &str, row_idx: usize) -> Diagnostic {
flow_diagnostic(
"FLOW_050",
format!("row at index {} of view '{}' has no series key in column '{}'", row_idx, view, column),
"A series view row key is built from the view's own key column, so the column must be present \
and hold a value convertible to an unsigned integer. A none value or an unsupported type here \
would collapse every row of the view onto the same key.",
)
}
pub fn flow_dictionary_source_unsupported() -> Diagnostic {
flow_diagnostic(
"FLOW_038",
"dictionaries cannot source deferred or transactional views".to_string(),
"Dictionary entries are single-version sequence-like state with no change feed. Query the \
dictionary directly with FROM namespace::dictionary instead.",
)
}
pub fn flow_queue_source_unsupported() -> Diagnostic {
flow_diagnostic(
"FLOW_039",
"queues cannot source deferred or transactional views".to_string(),
"Queue items are claimed and acknowledged by consumers rather than streamed. Query the \
queue directly with FROM namespace::queue instead.",
)
}
pub fn flow_rolling_lag_requires_event_time(flow: &str) -> Diagnostic {
Diagnostic {
code: "FLOW_043".to_string(),
rql: None,
message: format!("{flow} uses a rolling window with `lag` but its sources supply no event time"),
column: None,
fragment: Fragment::None,
label: None,
help: Some(
"lag holds a rolling window open for out-of-order arrivals, which only has meaning against a \
source-supplied event time. Declare `with { time: event(<column>) }` on the source \
object this flow reads, or remove the lag."
.to_string(),
),
notes: vec![],
cause: None,
operator_chain: None,
}
}
pub fn flow_join_retention_requires_event_time(flow: &str) -> Diagnostic {
Diagnostic {
code: "FLOW_049".to_string(),
rql: None,
message: format!("{flow} declares a join retention but its sources supply no event time"),
column: None,
fragment: Fragment::None,
label: None,
help: Some(
"a join retention frees a row once the watermark passes its own event time, which only has \
meaning against a source-supplied event time. Declare `with { time: event(<column>) }` on the \
source object this flow reads, or remove the retention."
.to_string(),
),
notes: vec![],
cause: None,
operator_chain: None,
}
}
pub fn flow_catch_up_read_failed(from: u64, up_to: u64, cause: &str) -> Diagnostic {
Diagnostic {
code: "FLOW_046".to_string(),
rql: None,
message: format!("cdc catch-up read for versions ({from}, {up_to}] failed: {cause}"),
column: None,
fragment: Fragment::None,
label: None,
help: Some("The flow catch-up loader could not read the durable CDC log on behalf of a \
lagging flow. The flow retries with backoff and is poisoned if the failure \
persists; check the CDC storage for corruption or exhaustion."
.to_string()),
notes: vec![],
cause: None,
operator_chain: None,
}
}
pub fn flow_span_on_unageable_node(flow: &str, operator: &str) -> Diagnostic {
Diagnostic {
code: "FLOW_045".to_string(),
rql: None,
message: format!("{operator} in {flow} declares a retention span but holds no state to age"),
column: None,
fragment: Fragment::None,
label: None,
help: Some(format!(
"Spans are only meaningful on operators that keep keyed state - join, distinct, append, \
apply and aggregate. {operator} keeps none, so the span would be accepted and never \
consulted. Move the span to the stateful operator downstream, or remove it."
)),
notes: vec![],
cause: None,
operator_chain: None,
}
}
pub fn flow_guest_key_too_wide(len: usize) -> Diagnostic {
flow_diagnostic(
"FLOW_051",
format!("a guest row mapping key is {} bytes, the key holds at most 16", len),
"A guest operator supplied a row mapping key wider than the key can hold. Shorten the key or hash it \
to sixteen bytes before passing it; the host refuses it rather than truncating it, because a \
truncated key silently collides with every other key sharing its first sixteen bytes.",
)
}