use bitmask-oriented id abstractions - #139
Conversation
| #[test] | ||
| fn op_type_code_roundtrip_covers_all_variants() { | ||
| // Exhaustively iterate via from_u8 across the assigned range. | ||
| let all = [ |
There was a problem hiding this comment.
Can we somehow iterate OperatorTypeCode values itself instead of copying into separate all (ie if we add new OperatorTypeCode we may forget to update the test) ? Is it possible to iterate enum values in Rust?
| ) | ||
| }); | ||
| *counter = next; | ||
| node.operator_id = OperatorId::new(op_type_code, next); |
| 0x11 => Some(Self::Chained), | ||
| _ => None, | ||
| } | ||
| } |
There was a problem hiding this comment.
My concern is the hard binding of the type to the byte—adding/removing operator types can lead to confusion: it'll be necessary to update not only the enum but also the from_u8 function. Also right now, everything is in order (first sources, then regular, then sinks) which looks good and pretty, but what if we add another source? It'll go down after the sink - looks awkward. Functionally it's ok, just curious if we can do better
There was a problem hiding this comment.
I researched the possibilities and came up with the following alternatives:
To get rid of additional efforts for from_u8 update I found the cleanest way with new crate "num_enum", which is just a codegen library, which will do basically the same thing in the code when calling it's method.
The code would look like this:
use num_enum::TryFromPrimitive;
#[repr(u8)]
#[derive(Copy, Clone, Eq, PartialEq, Hash, Debug, Serialize, Deserialize, TryFromPrimitive)]
pub enum OperatorTypeCode {
Invalid = 0x00,
SourceVector = 0x01,
// ...
Chained = 0x11,
}
impl OperatorTypeCode {
pub fn from_u8(byte: u8) -> Option<Self> {
Self::try_from(byte).ok()
}
// ...
}
For ordering types I think the only way we have is introducing strict intervals for source, regular and sinks.
My proposal is to reserve 32 values for both source and sinks and the rest for regular transforms. What do you think?
| 0x11 => Some(Self::Chained), | ||
| _ => None, | ||
| } | ||
| } |
There was a problem hiding this comment.
Asked Gemini, suggested looking at num-derive and num-traits crates and strum crate (this one can turn enums to iterators, can be relevant) - may be worth exploring. In general im fine with the way it is now, just if it can be improved with not too much work then great, if it's too much work - dont bother, keep as is.
There was a problem hiding this comment.
I think I can fix it quickly
| 8 + 24 + 24 + len // u64 + Option<String> + Option<HashMap> + content | ||
|
|
||
| 8 + 8 + 24 + len // ingest_timestamp + upstream_vertex_id + Option<HashMap> + content |
| base_prefix: &ObjectPath, | ||
| partition_path: &str, | ||
| task_index: i32, | ||
| task_index: u16, |
| // Group task metrics by operator_id. | ||
| // The HashMap is keyed by vertex_id rendered via Display; iterate vertices in the graph | ||
| // and reverse-match by Display form to recover their operator_id. | ||
| let by_label: HashMap<VertexId, &TaskMetrics> = tasks_metrics |
There was a problem hiding this comment.
Can you explain why we need separate by_label map here?
There was a problem hiding this comment.
Totally agree there is no need for it - looks like Claude overthought it...
Refactored this part.
| .runtime_context | ||
| .as_ref() | ||
| .map(|ctx| ctx.vertex_id().to_string()) | ||
| .unwrap_or_else(|| "unknown_vertex".to_string()); |
|
|
||
| use anyhow::Result; | ||
| use datafusion::common::ScalarValue; | ||
| use kameo::Reply; |
| .as_deref() | ||
| .and_then(parse_task_index_from_vertex_id); | ||
| .unwrap() | ||
| .task_index(); |
| pub target_vertex_id: VertexId, | ||
| pub edge_id: String, | ||
| pub target_operator_id: String, | ||
| pub edge_id: ChannelId, |
anovv
left a comment
anovv
left a comment
There was a problem hiding this comment.
Thanks for working on this, makes things more consistent and structured.
Looks good to me, just few questions/suggestions
| fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { | ||
| f.write_str(self.short_name()) | ||
| } | ||
| } |
There was a problem hiding this comment.
we can also simplify short_name() and custom Display impl using strum crate - smth like
use strum::{Display, IntoStaticStr};
#[derive(Display, IntoStaticStr)] // implements Display and &'static str conversion
pub enum OperatorTypeCode {
...
then short name is simply
pub fn short_name(self) -> &'static str {
self.into() // refers to IntoStaticStr
}
(the syntax may not be exact - it is AI generated, but you get the point)
and custom Display won't be needed either (unless we want to use short_name() directly).
2 participants
The point of the PR is:
OperatorTypeCode — a #[repr(u8)] enum that assigns a stable numeric code to each concrete operator variant. Restricts the system to at most 255 distinct operator types. Codes must only be appended (never renumbered) so existing checkpoints and serialized IDs stay interpretable.
OperatorId — a u16 identifier for a logical operator node within the execution graph. Layout: [op_type: u8 | instance: u8], where the upper 8 bits hold the OperatorTypeCode and the lower 8 bits hold the instance index of that operator type within the graph. All parallel partitions of the same operator share one OperatorId.
VertexId — a u32 identifier for a single execution vertex (one parallel partition of an operator). Layout: [op_type: u8 | instance: u8 | task_index: u16], where the upper 16 bits form the OperatorId and the lower 16 bits identify the parallel partition within that operator.
ChannelId — a u64 globally unique identifier for a directed channel between two vertices. Layout: [source: VertexId | target: VertexId], upper 32 bits for the source and lower 32 bits for the target.
Also added: Display/Debug formatters, accessor methods, and unit tests covering round-trips, layout packing, and display output.
Also added formatters, helper methods and tests.