Semantic meaning is taken from the wrong input when a multi-input transform feeds another transform — nondeterministic per process

A note for the community

• Please vote on this issue by adding a 👍 reaction to the original issue to help the community and maintainers prioritize this request
• If you are interested in working on this issue or have submitted a pull request, please leave a comment

Problem

When a transform has multiple inputs whose schema definitions disagree about a semantic meaning (different path for the same meaning, or one input declaring it and another not), the per-input definitions collapse into a single one at the next hop. From that point on, every event on that path is stamped with the same definition regardless of which input it came from, so find_key_by_meaning() — used by the datadog_* sinks — reads a field that the event's own source never declared for that meaning.

Which definition survives is:

• decided once per process,
• the same for every event that process handles,
• different across restarts of the same binary with the same config.

The topology needs two hops: sources → fan-in transform → sink keeps each event's own definition. The collapse happens when the fan-in transform feeds another transform.

Configuration

Two sources whose definitions declare meaning(host) at different paths (internal_logs takes the path from host_key), a fan-in transform, and one more transform before the sink. Every event carries both fields, so whichever value ends up in hostname shows which definition was used.

data_dir: "/tmp/dcollapse/data"

sources:
  src_a:
    type: internal_logs
    host_key: host_a          # definition: meaning(host) = .host_a
    pid_key: ""
  src_b:
    type: internal_logs
    host_key: host_b          # definition: meaning(host) = .host_b
    pid_key: ""

transforms:
  fanin:                      # multi-input: holds 2 definitions on one output
    type: remap
    inputs: [src_a, src_b]
    source: |
      .host_a = "FROM_A"
      .host_b = "FROM_B"
      .hostname = "ALREADY_SET"   # so the rename to _RESERVED_host is visible too

  passthrough:                # the extra hop
    type: remap
    inputs: [fanin]
    source: |
      .hop = "second"

sinks:
  dd:
    type: datadog_logs
    inputs: [passthrough]     # wiring `fanin` here directly instead => per-input definitions are kept
    default_api_key: "test"
    endpoint: http://127.0.0.1:8127
    compression: gzip

Both sources declare a host meaning here, so the sink always finds one and always logs the rename — the warning alone does not tell the variants apart. What differs is which field's value ends up in hostname, so the check needs the payload. Point the sink at a local receiver and restart Vector ~10 times:

import gzip, json
from http.server import BaseHTTPRequestHandler, HTTPServer
class H(BaseHTTPRequestHandler):
    def do_POST(self):
        b = self.rfile.read(int(self.headers.get("Content-Length") or 0))
        if "gzip" in (self.headers.get("Content-Encoding") or ""):
            b = gzip.decompress(b)
        for ev in json.loads(b):
            print("hostname=%s host_a=%s host_b=%s" % (ev.get("hostname"), ev.get("host_a"), ev.get("host_b")))
        self.send_response(200); self.end_headers(); self.wfile.write(b"{}")
    def log_message(self, *a): pass
HTTPServer(("127.0.0.1", 8127), H).serve_forever()

Version

vector 0.55.0 (x86_64-unknown-linux-gnu cf8de83 2026-04-22), schema.enabled at its default (false).
input_definitions / with_definitions / position_reserved_attr_event_root are identical in 0.51.0 – 0.56.0.

Debug Output

8 restarts per variant, same binary, the config above, only sinks.dd.inputs differing between the two. Counting hostname values in the sink payload:

A) sources -> fanin -> datadog_logs                 (inputs: [fanin])
   RUN 1: MIXED  FROM_A=23 FROM_B=22
   RUN 2: MIXED  FROM_A=23 FROM_B=22
   RUN 3: MIXED  FROM_A=23 FROM_B=23
   RUN 4: MIXED  FROM_A=23 FROM_B=22
   RUN 5: MIXED  FROM_A=22 FROM_B=23
   RUN 6: MIXED  FROM_A=23 FROM_B=23
   RUN 7: MIXED  FROM_A=23 FROM_B=23
   RUN 8: MIXED  FROM_A=23 FROM_B=23     <- every event keeps its own definition

B) sources -> fanin -> passthrough -> datadog_logs  (inputs: [passthrough])
   RUN 1: ALL FROM_A (40/40)
   RUN 2: ALL FROM_A (40/40)
   RUN 3: ALL FROM_B (40/40)
   RUN 4: ALL FROM_A (40/40)
   RUN 5: ALL FROM_B (40/40)
   RUN 6: ALL FROM_A (40/40)
   RUN 7: ALL FROM_B (39/39)
   RUN 8: ALL FROM_B (39/39)

In variant B, events produced by src_a get hostname from .host_b (and vice versa) — every event in the process is treated the same way.

The warning is rate limited, so its count says nothing about how many events were affected: each of the runs above logged 2 lines for 39–45 renamed events, the second line being
Internal log [Semantic meaning is defined, but the event path already exists. Renaming to not overwrite.] is being suppressed to avoid flooding.

Example Data

Verbatim payload received by the local receiver in variant B, from a run that used src_a's definition (meaning(host) = .host_a):

{
  "_RESERVED_host": "ALREADY_SET",
  "hop": "second",
  "host_b": "FROM_B",
  "hostname": "FROM_A",
  "level": "\"info\"",
  "message": "Log level is enabled.",
  "metadata": {
    "kind": "event",
    "level": "INFO",
    "module_path": "vector::app",
    "target": "vector::app"
  },
  "source_type": "internal_logs",
  "timestamp": 1787213726621
}

.host_a was moved into hostname and the existing hostname was renamed to _RESERVED_host, for every event in that process — including the events from src_b, whose definition points at .host_b; that field is left in the payload. On the next start the two can swap.

Additional Context

Code path:

1. src/topology/schema.rs → input_definitions() — for a transform whose input is another transform, it takes the upstream's TransformOutput::schema_definitions(...) (a HashMap<OutputId, Definition> with one entry per the upstream's own inputs), iterates .values() and passes them to OutputId::with_definitions():

   let mut transform_definitions = input.with_definitions(
       config.transform_output_for_port(key, &input.port, ...)?
           .schema_definitions(config.schema_enabled())
           .values()
           .cloned(),
   );
   

2. lib/vector-core/src/config/output_id.rs → with_definitions() re-keys every definition to the same OutputId (the immediate upstream component):

   definitions.into_iter().map(|definition| (self.clone(), definition)).collect()
   

3. src/transforms/remap.rs → outputs() inserts them into a HashMap keyed by that OutputId:

   default_definitions.insert(output_id.clone(), ...);
   

   The key is the same for all of them, so only one entry remains — the last one yielded by the .values() iteration in step 1. std::collections::HashMap iteration order is unspecified and seeded per instance.

4. At runtime, update_runtime_schema_definition() looks the definition up by upstream_id and finds only that entry.

In the datadog_logs sink (normalize_event → position_reserved_attr_event_root) this results in:

• a field from another input being promoted into a reserved attribute (hostname, service, ddsource, status, timestamp, ddtags);
• the value already present in that reserved attribute being renamed to _RESERVED_<meaning>.

References

• #22341 — same symptom (_RESERVED_service appearing randomly after restarts, not recovering until the next restart), closed without a root cause
• #16793 — the PR that introduced multiple definitions per transform output. Its description says: "Transforms can output multiple definitions (each will be tagged with a different OutputId signifying the definition that will be output based on what has been input - this will be done in a followup PR)"
• #11416 — states the requirement this breaks: "a sink can accept events from multiple inputs, and for it to be able to fetch field values based on the "semantic meaning" attached to an event schema, it has to know per event where the field belonging to a given semantic meaning can be found"

---

🤖 Investigated and written with Claude Code

Start with src/topology/schema.rs input_definitions(), then inspect lib/vector-core/src/config/output_id.rs with_definitions() and src/transforms/remap.rs outputs(). Reproduce the supplied two-hop configuration and local receiver, then trace update_runtime_schema_definition() into datadog_logs normalization. Done means each event retains its own semantic definition through the extra transform, with a regression test covering the differing source paths.