ClosingFuture.whenAllComplete(ImmutableList.of(input1, input2Failed))
            .call(
                new CombiningCallable<TestCloseable>() {
                  @Override
                  public TestCloseable call(DeferredCloser closer, Peeker peeker) throws Exception {
                    closer.eventuallyClose(closeable1, closingExecutor);
                    assertThat(peeker.getDone(input1)).isSameInstanceAs("value1");

  // CHECK-DAG:  [[READ1:%.+]] = "tf.ReadVariableOp"([[ARG1]])
  // CHECK:      [[INPUT0:%.+]] = "tf.TPUPartitionedInputV2"([[READ0]], [[READ1]])
  // CHECK-DAG:  [[READ2:%.+]] = "tf.ReadVariableOp"([[ARG0]])
  // CHECK-DAG:  [[READ3:%.+]] = "tf.ReadVariableOp"([[ARG1]])
  // CHECK:      [[INPUT1:%.+]] = "tf.TPUPartitionedInputV2"([[READ2]], [[READ3]])

        PathSensitivity.NAME_ONLY | [['first/input', 'foo'], ['second/input', 'foo'], ['third/input1', 'foo']]
        PathSensitivity.RELATIVE  | [['first/input', 'foo'], ['second/input', 'foo'], ['third/input1', 'foo']]
        PathSensitivity.ABSOLUTE  | [['first/input', 'foo'], ['first/input', 'foo'], ['third/input', 'foo']]
    }

  params.op_kernel = kernel.get();
  AllocatorAttributes alloc_attrs;
  params.output_attr_array = &alloc_attrs;
  gtl::InlinedVector<TensorValue, 4> inputs;
  inputs.emplace_back(tags);
  inputs.emplace_back(values);
  params.inputs = inputs;
  OpKernelContext ctx(&params, 1);
  kernel->Compute(&ctx);
  ASSERT_EQ(expected_code, ctx.status().code());
  if (expected_code == error::OK) {
    Summary summary;

  EXPECT_THAT(input2->type, Eq(TensorType_INT8));

  // Check if the quantization params of the minimum/maximum inputs match
  // after requantization
  EXPECT_THAT(input1->quantization->scale, Eq(input2->quantization->scale));
  EXPECT_THAT(input1->quantization->zero_point,
              Eq(input2->quantization->zero_point));

  // Check the input quantization params match the output ones.

  // CHECK-SAME: %[[INPUT1:.*]]: tensor<1024x3xf32>, %[[INPUT2:.*]]: tensor<1024x3xf32>
  // CHECK: %[[CONSTANT2:.*]] = stablehlo.constant dense<1.000000e+03> : tensor<1024x3xf32>
  // CHECK: %[[ADD:.*]] = stablehlo.add %[[INPUT1]], %[[CONSTANT2]] : tensor<1024x3xf32>
  // CHECK: %[[MUL:.*]] = stablehlo.multiply %[[INPUT1]], %[[INPUT2]] : tensor<1024x3xf32>
  // CHECK: return %[[ADD]], %[[MUL]]

  func.return %r : tensor<i32>
}

// -----

// Test inputs to the child streams are merged to the parent streams

// CHECK-LABEL: func private @main_stream_1
// CHECK-SAME: ([[INPUT0:%.*]]: tensor<i32>, [[INPUT1:%.*]]: tensor<i32>
// CHECK: tf.Sub
// CHECK: tf.Sub
// CHECK: mlrt.async({{%.*}}, [[INPUT1]]

// CHECK-LABEL: func @main

            test {
                inputs1 = files("input1.txt")
                inputs2 = files("input2.txt", "input3.txt")
            }
        """

        when:
        succeeds "test", "--info"

        then:
        executedAndNotSkipped ':test'
        outputContains "Input property 'inputs1' file ${file("input2.txt")} has been removed."

    %0 = func.call @func_0_CPU_QUANTIZED_INT8(%arg0, %arg1, %arg2) {tac.device = "CPU", tac.inference_type = "QUANTIZED_INT8", tac.interface_name = "func_0"} : (tensor<100x!quant.uniform<i8:f32, 2.000000e-01:-3>>,...

In general, these are built by providing some `func(inputs...) outputs...` (called "transformation" functions).
While more could be expressed, there are currently three forms implemented.

* `func() *O` via `NewSingleton`
    * This generates a collection that has a single value. An example would be some global configuration.
* `func(input I) *O` via `NewCollection`