memory table untestable unless test data is given to direct input

[jiping-s]

A note for the community

In unit tests, the transforms referenced by memory tables need input data directly on their inputs, otherwise test fails. It's impractical for anything but simplest setup.

Problem

See example from #24081

If cache_generator does not use demo_logs_processor directly, but:

  demo_logs_processor_2:
    type: "remap"
    inputs: ["demo_logs_processor"]
    source: ""

  cache_generator:
    type: "remap"
    inputs: ["demo_logs_processor_2"]

then vector test throws error:

Failed to build test 'Syslog merge out':
  Transform "cache_generator" has no inputs.

While test data may be injected directly to demo_logs_processor_2 to fix the problem, it's not practical in larger pipelines, because inputs for all memory tables must be injected for all tests, even if some of the tables are not actually touched.

Configuration

Version

0.52.0

Debug Output

Example Data

No response

Additional Context

No response

References

No response

[jiping-s]

I made a temporary workaround by adding all dependencies used by memory table transforms:

• transforms are added as-is
• sources and sinks are added as synthetic sources of the "unit_test" type

However, it only allows normal unit tests to pass, not actually testing enrichment with memory tables:

1. Newly added nodes are not connected to test outputs - memory table access is still not part of topology
2. Regardless whether 1 is addressed, updating memory table could not affect test outputs (due to timing?)

in src/config/unit_test/mod.rs:

// collect dependencies outside of main test inputs => sinks graph. Required for memory table.
fn add_input_dependencies(
    input_id: &str,
    config_builder: &ConfigBuilder,
    synthetic_sources: &mut IndexMap<ComponentKey, SourceOuter>,
    valid_components: &mut HashSet<String>,
) {
    // input id might be "component_id.route_name"
    let component_id = input_id.split_once('.').map(|(name, _)| name).unwrap_or(input_id);

    if valid_components.contains(component_id) {
        return;
    }
    valid_components.insert(component_id.to_string());

    let key = ComponentKey::from(component_id);
    if let Some(transform) = config_builder.transforms.get(&key) {
        debug!(message = "Adding out-of-graph transform node: ", node = component_id, internal_log_rate_limit = false);
        for input in &transform.inputs {
            add_input_dependencies(
                &input.to_string(),
                config_builder,
                synthetic_sources,
                valid_components,
            );
        }
    } else if component_id.contains("-source-") {
        // unit test sources
    } else if !config_builder.sources.contains_key(&key) && !config_builder.sinks.contains_key(&key) {
        debug!(message = "Adding out-of-graph synthetic source node: ", node = component_id, internal_log_rate_limit = false);
        let empty_source_config = UnitTestSourceConfig::default();
        synthetic_sources.insert(key, SourceOuter::new(empty_source_config));
    } else {
        warn!(message = "Ignoring unknown out-of-graph node: ", node = component_id, internal_log_rate_limit = false);
    }
}

...
async fn build_unit_test(
...
    // Enrichment tables consume inputs but are referenced dynamically in VRL transforms
    // (via get_enrichment_table_record). Since we can't statically analyze VRL usage,
    // we conservatively include all enrichment table inputs as valid components.
    let mut synthetic_sources = IndexMap::new();
    config_builder
        .enrichment_tables
        .iter()
        .filter_map(|(key, c)| c.as_sink(key).map(|(_, sink)| sink.inputs))
        .for_each(|inputs| {
            // Recursively add all dependencies of enrichment table inputs
            for input in inputs {
                add_input_dependencies(
                    &input.to_string(),
                    &config_builder,
                    &mut synthetic_sources,
                    &mut valid_components,
                );
            }
        });
    config_builder.sources.extend(synthetic_sources);

...