Q&A : In a reduce transform, How to access array or sum values in the ends_when vrl condition ?

[htam-net]

Question

Hello everyone,

I try to reconstruct a longer message splited in several chunks (all chunks have an unique chunk_id, and all chunks of the same message share a message_id to use in group_by). Also, all chunks include a total_chunk variable.

in the merge_strategies, I place the variable chunk_id in array, so I can manage the possible out-of-order chunks in a later remap transform.
To optimize, I'm looking for a solution to be able to count the already grouped_by chunks and when it equals the total_chunks, the ends_when clause.

I also try to add a .chunk_count = 1 variable and sum it in the reduce, the .chunk_count variable in the ends_when is not updated either.

How can I access the value of these "reduced" variables in the ends_when statement ?

Vector Config

reduce_chunks:
    type: reduce
    inputs:
      - udp_source
    group_by: 
      - "message_id" # all chunks with the message_id (UUID) are part of the same original message
    merge_strategies:
      chunk: "array"       # We place all chunk in a array
      chunk_id: "array"     # We place all chunk_id in a array to be able to reorder in a later remap transform
      total_chunks: "discard" # We keep the first one (because all value are 
    ends_when: 
      type: vrl
      source: |
        log("Start Verifying last message", level: "info", rate_limit_secs: 0)
        log(.chunk_id, level: "info", rate_limit_secs: 0) # We can see in the logs that .chunk_id is still an int and not an array
        if is_array(.chunk_id) {
          if exists(.total_chunks) {
            log("array chunk_id is present and so is .total_chunks", level: "info", rate_limit_secs: 0)
            if length(array!(.chunk_id)) >= to_int!(.total_chunks) {
              log("this is last message", level: "info", rate_limit_secs: 0)
              true
            } else {
              log("message is not the last", level: "info", rate_limit_secs: 0)
              false
            }
          } else {
            log(".total_chunks is missing", level: "info", rate_limit_secs: 0)
            false
          }
        } else {                
          log("no array of .chunk_id", level: "info", rate_limit_secs: 0)
          false
        }

Vector Logs

2026-07-06T18:22:16.865912Z  INFO transform{component_kind="transform" component_id=reduce_chunks component_type=reduce}: vrl::stdlib::log::implementation: Start Verifying last message internal_log_rate_secs=0 vrl_position=0
2026-07-06T18:22:16.865921Z  INFO transform{component_kind="transform" component_id=reduce_chunks component_type=reduce}: vrl::stdlib::log::implementation: 0 internal_log_rate_secs=0 vrl_position=82
2026-07-06T18:22:16.865928Z  INFO transform{component_kind="transform" component_id=reduce_chunks component_type=reduce}: vrl::stdlib::log::implementation: no array of .chunk_id internal_log_rate_secs=0 vrl_position=657
2026-07-06T18:22:16.865966Z TRACE transform{component_kind="transform" component_id=reduce_chunks component_type=reduce}: vector_common::internal_event::events_received: Internal log [Events received.] is being suppressed to avoid flooding.
2026-07-06T18:22:16.865975Z  INFO transform{component_kind="transform" component_id=reduce_chunks component_type=reduce}: vrl::stdlib::log::implementation: Start Verifying last message internal_log_rate_secs=0 vrl_position=0
2026-07-06T18:22:16.865981Z  INFO transform{component_kind="transform" component_id=reduce_chunks component_type=reduce}: vrl::stdlib::log::implementation: 1 internal_log_rate_secs=0 vrl_position=82
2026-07-06T18:22:16.865987Z  INFO transform{component_kind="transform" component_id=reduce_chunks component_type=reduce}: vrl::stdlib::log::implementation: no array of .chunk_id internal_log_rate_secs=0 vrl_position=657
2026-07-06T18:22:16.866213Z  INFO transform{component_kind="transform" component_id=reduce_chunks component_type=reduce}: vrl::stdlib::log::implementation: Start Verifying last message internal_log_rate_secs=0 vrl_position=0
2026-07-06T18:22:16.866226Z  INFO transform{component_kind="transform" component_id=reduce_chunks component_type=reduce}: vrl::stdlib::log::implementation: 2 internal_log_rate_secs=0 vrl_position=82
2026-07-06T18:22:16.866295Z  INFO transform{component_kind="transform" component_id=reduce_chunks component_type=reduce}: vrl::stdlib::log::implementation: no array of .chunk_id internal_log_rate_secs=0 vrl_position=657
2026-07-06T18:22:16.866360Z  INFO transform{component_kind="transform" component_id=reduce_chunks component_type=reduce}: vrl::stdlib::log::implementation: Start Verifying last message internal_log_rate_secs=0 vrl_position=0
2026-07-06T18:22:16.866390Z  INFO transform{component_kind="transform" component_id=reduce_chunks component_type=reduce}: vrl::stdlib::log::implementation: 3 internal_log_rate_secs=0 vrl_position=82
2026-07-06T18:22:16.866406Z  INFO transform{component_kind="transform" component_id=reduce_chunks component_type=reduce}: vrl::stdlib::log::implementation: no array of .chunk_id internal_log_rate_secs=0 vrl_position=657
2026-07-06T18:22:16.866445Z  INFO transform{component_kind="transform" component_id=reduce_chunks component_type=reduce}: vrl::stdlib::log::implementation: Start Verifying last message internal_log_rate_secs=0 vrl_position=0
2026-07-06T18:22:16.866494Z  INFO transform{component_kind="transform" component_id=reduce_chunks component_type=reduce}: vrl::stdlib::log::implementation: 4 internal_log_rate_secs=0 vrl_position=82
2026-07-06T18:22:16.866516Z  INFO transform{component_kind="transform" component_id=reduce_chunks component_type=reduce}: vrl::stdlib::log::implementation: no array of .chunk_id internal_log_rate_secs=0 vrl_position=657
2026-07-06T18:22:16.866589Z  INFO transform{component_kind="transform" component_id=reduce_chunks component_type=reduce}: vrl::stdlib::log::implementation: Start Verifying last message internal_log_rate_secs=0 vrl_position=0
2026-07-06T18:22:16.866636Z  INFO transform{component_kind="transform" component_id=reduce_chunks component_type=reduce}: vrl::stdlib::log::implementation: 5 internal_log_rate_secs=0 vrl_position=82
2026-07-06T18:22:16.866665Z  INFO transform{component_kind="transform" component_id=reduce_chunks component_type=reduce}: vrl::stdlib::log::implementation: no array of .chunk_id internal_log_rate_secs=0 vrl_position=657
2026-07-06T18:22:16.866778Z  INFO transform{component_kind="transform" component_id=reduce_chunks component_type=reduce}: vrl::stdlib::log::implementation: Start Verifying last message internal_log_rate_secs=0 vrl_position=0
2026-07-06T18:22:16.866799Z  INFO transform{component_kind="transform" component_id=reduce_chunks component_type=reduce}: vrl::stdlib::log::implementation: 6 internal_log_rate_secs=0 vrl_position=82
2026-07-06T18:22:16.866814Z  INFO transform{component_kind="transform" component_id=reduce_chunks component_type=reduce}: vrl::stdlib::log::implementation: no array of .chunk_id internal_log_rate_secs=0 vrl_position=657

[s3onghyun]

The reason .chunk_id is still a scalar in ends_when is that ends_when (and starts_when) is evaluated against the incoming event, before it is merged into the accumulated reduce state. In src/transforms/reduce/transform.rs the transform_one path checks the condition first and only then calls push_or_new_reduce_state, so your array merge strategy has not been applied yet at the moment the condition runs. There is no way to reference the accumulator from inside ends_when.

So counting up to total_chunks in ends_when cannot work by design. Practical options:

• Rely on expire_after_ms to flush each group, then validate / reorder the chunks in a downstream remap.
• Or compute a per-event counter upstream and end on that, rather than on the merged array length.

The merged array only exists on the flushed output event, not during the ends_when check.

[htam-net]

Hello, are your sure you posted your answer in the right discussion thread ? ;-)

[s3onghyun]

You're right, and apologies for the noise. I pasted the wrong snippet here in a batch mix-up on my end, and I've edited the comment above to the actual answer for your question.

Short version: ends_when is evaluated against the incoming event before it is merged into the reduce state, so your array/sum merge results don't exist yet at that point. That is exactly why the log shows .chunk_id as a plain int rather than the accumulated array, and why .chunk_count never grows. There is no way to read the accumulator from inside ends_when. Practical options: let expire_after_ms flush the group and validate/reorder the chunks in a downstream remap, or carry the count on each event differently. Thanks for the nudge.

[htam-net]

OK. Thank you very much for the confirmation.

expire_after_ms is already in place as safety belt in case some packets have been dropped.

The challenge I try to overcome is that packet could be delivered out of order. So, I can't rely on a "last marked packet". And I try to "flush" as early as possible once all packets have been received.

Thank again.
Sincerely yours, Mathieu.

[htam-net]

Answer : not possible as of 2026-07-08