Make node_id globally unique (again).

core/src/graph.jl

@@ -12,7 +12,19 @@ function create_graph(db::DB, config::Config, chunk_sizes::Vector{Int})::MetaGra
     )
     edge_rows = execute(
         db,
-        "SELECT fid, from_node_type, from_node_id, to_node_type, to_node_id, edge_type, subnetwork_id FROM Edge ORDER BY fid",
+        """
+        SELECT
+            Edge.fid,
+            FNode.node_id AS from_node_id,
+            FNode.node_type AS from_node_type,
+            TNode.node_id AS to_node_id,
+            TNode.node_type AS to_node_type,
+            Edge.edge_type,
+            Edge.subnetwork_id
+        FROM Edge
+        LEFT JOIN Node AS FNode ON FNode.node_id = Edge.from_node_id
+        LEFT JOIN Node AS TNode ON TNode.node_id = Edge.to_node_id
+        """,
     )
     # Node IDs per subnetwork
     node_ids = Dict{Int32, Set{NodeID}}()

core/src/parameter.jl

@@ -66,6 +66,16 @@ function NodeID(type::NodeType.T, value::Integer, db::DB)::NodeID
     return NodeID(type, value, idx)
 end
 
+function NodeID(value::Integer, db::DB)::NodeID
+    (idx, type) = execute(
+        columntable,
+        db,
+        "SELECT COUNT(*), node_type FROM Node WHERE node_type == (SELECT node_type FROM Node WHERE node_id == $value) AND node_id <= $value",
+    )
+    @assert idx[1] > 0
+    return NodeID(type[1], value, idx[1])
+end
+
 Base.Int32(id::NodeID) = id.value
 Base.convert(::Type{Int32}, id::NodeID) = id.value
 Base.broadcastable(id::NodeID) = Ref(id)

core/src/read.jl

@@ -661,7 +661,7 @@ function CompoundVariable(
     }[]
     # Each row defines a subvariable
     for row in compound_variable_data
-        listen_node_id = NodeID(row.listen_node_type, row.listen_node_id, db)
+        listen_node_id = NodeID(row.listen_node_id, db)
         # Placeholder until actual ref is known
         variable_ref = PreallocationRef(placeholder_vector, 0)
         variable = row.variable
@@ -900,14 +900,12 @@ function PidControl(
     end
     controlled_basins = collect(controlled_basins)
 
+    listen_node_id = NodeID.(parsed_parameters.listen_node_id, Ref(db))
+
     return PidControl(;
         node_id = node_ids,
         parsed_parameters.active,
-        listen_node_id = NodeID.(
-            parsed_parameters.listen_node_type,
-            parsed_parameters.listen_node_id,
-            Ref(db),
-        ),
+        listen_node_id,
         parsed_parameters.target,
         target_ref,
         parsed_parameters.proportional,

core/src/schema.jl

@@ -220,7 +220,6 @@ end
 @version DiscreteControlVariableV1 begin
     node_id::Int32
     compound_variable_id::Int32
-    listen_node_type::String
     listen_node_id::Int32
     variable::String
     weight::Union{Missing, Float64}
@@ -241,7 +240,6 @@ end
 
 @version ContinuousControlVariableV1 begin
     node_id::Int32
-    listen_node_type::String
     listen_node_id::Int32
     variable::String
     weight::Union{Missing, Float64}
@@ -258,7 +256,6 @@ end
 @version PidControlStaticV1 begin
     node_id::Int32
     active::Union{Missing, Bool}
-    listen_node_type::String
     listen_node_id::Int32
     target::Float64
     proportional::Float64
@@ -269,7 +266,6 @@ end
 
 @version PidControlTimeV1 begin
     node_id::Int32
-    listen_node_type::String
     listen_node_id::Int32
     time::DateTime
     target::Float64

core/src/validation.jl

@@ -110,8 +110,7 @@ sort_by_time_id_level(row) = (row.time, row.node_id, row.level)
 sort_by_priority(row) = (row.node_id, row.priority)
 sort_by_priority_time(row) = (row.node_id, row.priority, row.time)
 sort_by_subgrid_level(row) = (row.subgrid_id, row.basin_level)
-sort_by_variable(row) =
-    (row.node_id, row.listen_node_type, row.listen_node_id, row.variable)
+sort_by_variable(row) = (row.node_id, row.listen_node_id, row.variable)
 sort_by_condition(row) = (row.node_id, row.compound_variable_id, row.greater_than)
 sort_by_id_input(row) = (row.node_id, row.input)
 
@@ -174,13 +173,11 @@ end
 function valid_nodes(db::DB)::Bool
     errors = false
 
-    sql = "SELECT node_type, node_id FROM Node GROUP BY node_type, node_id HAVING COUNT(*) > 1"
-    node_type, node_id = execute(columntable, db, sql)
-
-    for (node_type, node_id) in zip(node_type, node_id)
+    sql = "SELECT node_id FROM Node GROUP BY node_id HAVING COUNT(*) > 1"
+    node_ids = only(execute(columntable, db, sql))
+    for node_id in node_ids
         errors = true
-        id = NodeID(node_type, node_id)
-        @error "Multiple occurrences of node $id found in Node table."
+        @error "Multiple occurrences of node_id $node_id found in Node table."
     end
 
     return !errors

core/src/write.jl

@@ -217,9 +217,7 @@ function flow_table(
 )::@NamedTuple{
     time::Vector{DateTime},
     edge_id::Vector{Union{Int32, Missing}},
-    from_node_type::Vector{String},
     from_node_id::Vector{Int32},
-    to_node_type::Vector{String},
     to_node_id::Vector{Int32},
     flow_rate::FlatVector{Float64},
 }
@@ -228,9 +226,7 @@ function flow_table(
     (; graph) = integrator.p
     (; flow_dict) = graph[]
 
-    from_node_type = String[]
     from_node_id = Int32[]
-    to_node_type = String[]
     to_node_id = Int32[]
     unique_edge_ids_flow = Union{Int32, Missing}[]
 
@@ -240,9 +236,7 @@ function flow_table(
     end
 
     for (from_id, to_id) in flow_edge_ids
-        push!(from_node_type, string(from_id.type))
         push!(from_node_id, from_id.value)
-        push!(to_node_type, string(to_id.type))
         push!(to_node_id, to_id.value)
         push!(unique_edge_ids_flow, graph[from_id, to_id].id)
     end
@@ -257,21 +251,11 @@ function flow_table(
     end
     time = repeat(datetime_since.(t_starts, config.starttime); inner = nflow)
     edge_id = repeat(unique_edge_ids_flow; outer = ntsteps)
-    from_node_type = repeat(from_node_type; outer = ntsteps)
     from_node_id = repeat(from_node_id; outer = ntsteps)
-    to_node_type = repeat(to_node_type; outer = ntsteps)
     to_node_id = repeat(to_node_id; outer = ntsteps)
     flow_rate = FlatVector(saveval, :flow)
 
-    return (;
-        time,
-        edge_id,
-        from_node_type,
-        from_node_id,
-        to_node_type,
-        to_node_id,
-        flow_rate,
-    )
+    return (; time, edge_id, from_node_id, to_node_id, flow_rate)
 end
 
 "Create a discrete control result table from the saved data"

core/test/control_test.jl

@@ -262,24 +262,20 @@ end
     model = Ribasim.run(toml_path)
     flow_data = DataFrame(Ribasim.flow_table(model))
 
-    function get_edge_flow(from_node_type, from_node_id, to_node_type, to_node_id)
+    function get_edge_flow(from_node_id, to_node_id)
         data = filter(
-            [:from_node_type, :from_node_id, :to_node_type, :to_node_id] =>
-                (a, b, c, d) ->
-                    (a == from_node_type) &&
-                        (b == from_node_id) &&
-                        (c == to_node_type) &&
-                        (d == to_node_id),
+            [:from_node_id, :to_node_id] =>
+                (a, b) -> (a == from_node_id) && (b == to_node_id),
             flow_data,
         )
         return data.flow_rate
     end
 
-    inflow = get_edge_flow("LinearResistance", 1, "Basin", 1)
-    @test get_edge_flow("Basin", 1, "Outlet", 1) ≈ max.(0.6 .* inflow, 0) rtol = 1e-4
-    @test get_edge_flow("Outlet", 1, "Terminal", 1) ≈ max.(0.6 .* inflow, 0) rtol = 1e-4
-    @test get_edge_flow("Basin", 1, "Outlet", 2) ≈ max.(0.4 .* inflow, 0) rtol = 1e-4
-    @test get_edge_flow("Outlet", 2, "Terminal", 2) ≈ max.(0.4 .* inflow, 0) rtol = 1e-4
+    inflow = get_edge_flow(2, 3)
+    @test get_edge_flow(3, 4) ≈ max.(0.6 .* inflow, 0) rtol = 1e-4
+    @test get_edge_flow(4, 6) ≈ max.(0.6 .* inflow, 0) rtol = 1e-4
+    @test get_edge_flow(3, 5) ≈ max.(0.4 .* inflow, 0) rtol = 1e-4
+    @test get_edge_flow(5, 7) ≈ max.(0.4 .* inflow, 0) rtol = 1e-4
 end
 
 @testitem "Concentration discrete control" begin

core/test/run_models_test.jl

@@ -38,16 +38,8 @@
 
     @testset "Schema" begin
         @test Tables.schema(flow) == Tables.Schema(
-            (
-                :time,
-                :edge_id,
-                :from_node_type,
-                :from_node_id,
-                :to_node_type,
-                :to_node_id,
-                :flow_rate,
-            ),
-            (DateTime, Union{Int32, Missing}, String, Int32, String, Int32, Float64),
+            (:time, :edge_id, :from_node_id, :to_node_id, :flow_rate),
+            (DateTime, Union{Int32, Missing}, Int32, Int32, Float64),
         )
         @test Tables.schema(basin) == Tables.Schema(
             (
@@ -103,8 +95,8 @@
     @testset "Results values" begin
         @test flow.time[1] == DateTime(2020)
         @test coalesce.(flow.edge_id[1:2], -1) == [0, 1]
-        @test flow.from_node_id[1:2] == [6, 6]
-        @test flow.to_node_id[1:2] == [6, 2147483647]
+        @test flow.from_node_id[1:2] == [6, 0]
+        @test flow.to_node_id[1:2] == [0, 2147483647]
 
         @test basin.storage[1] ≈ 1.0
         @test basin.level[1] ≈ 0.044711584