public static void optimize(Pipeline pipeline) {
    // Compute which Schema fields are (or conversely, are not) accessed in a pipeline.
    FieldAccessVisitor fieldAccessVisitor = new FieldAccessVisitor();
    pipeline.traverseTopologically(fieldAccessVisitor);

    // Find transforms in this pipeline which both: 1. support projection pushdown and 2. output
    // unused fields.
    ProjectionProducerVisitor pushdownProjectorVisitor =
        new ProjectionProducerVisitor(fieldAccessVisitor.getPCollectionFieldAccess());
    pipeline.traverseTopologically(pushdownProjectorVisitor);
    Map<ProjectionProducer<PTransform<?, ?>>, Map<PCollection<?>, FieldAccessDescriptor>>
        pushdownOpportunities = pushdownProjectorVisitor.getPushdownOpportunities();

    // Translate target PCollections to their output TupleTags.
    PCollectionOutputTagVisitor outputTagVisitor =
        new PCollectionOutputTagVisitor(pushdownOpportunities);
    pipeline.traverseTopologically(outputTagVisitor);
    Map<ProjectionProducer<PTransform<?, ?>>, Map<TupleTag<?>, FieldAccessDescriptor>>
        taggedFieldAccess = outputTagVisitor.getTaggedFieldAccess();

    // For each eligible transform, replace it with a modified transform that omits the unused
    // fields.
    for (Entry<ProjectionProducer<PTransform<?, ?>>, Map<TupleTag<?>, FieldAccessDescriptor>>
        entry : taggedFieldAccess.entrySet()) {
      for (Entry<TupleTag<?>, FieldAccessDescriptor> outputFields : entry.getValue().entrySet()) {
        LOG.info(
            "Optimizing transform {}: output {} will contain reduced field set {}",
            entry.getKey(),
            outputFields.getKey(),
            outputFields.getValue().fieldNamesAccessed());
      }
      PTransformMatcher matcher = application -> application.getTransform() == entry.getKey();
      PushdownOverrideFactory<?, ?> overrideFactory =
          new PushdownOverrideFactory<>(entry.getValue());
      pipeline.replaceAll(ImmutableList.of(PTransformOverride.of(matcher, overrideFactory)));
    }
  }
@Test
  public void testMultipleOutputs() {
    Pipeline p = Pipeline.create();
    MultipleOutputSourceWithPushdown originalSource =
        new MultipleOutputSourceWithPushdown(
            FieldAccessDescriptor.withFieldNames("foo", "bar", "baz"),
            FieldAccessDescriptor.withFieldNames("qux", "quux", "quuz"));
    PCollectionTuple inputs = p.apply(originalSource);
    inputs
        .get(originalSource.tag1)
        .apply(new FieldAccessTransform(FieldAccessDescriptor.withFieldNames("foo")));
    inputs
        .get(originalSource.tag1)
        .apply(new FieldAccessTransform(FieldAccessDescriptor.withFieldNames("bar")));
    inputs
        .get(originalSource.tag2)
        .apply(new FieldAccessTransform(FieldAccessDescriptor.withFieldNames("qux")));
    inputs
        .get(originalSource.tag2)
        .apply(new FieldAccessTransform(FieldAccessDescriptor.withFieldNames("quux")));

    MultipleOutputSourceWithPushdown expectedSource =
        new MultipleOutputSourceWithPushdown(
            FieldAccessDescriptor.withFieldNames("foo", "bar"),
            FieldAccessDescriptor.withFieldNames("qux", "quux"));

    ProjectionPushdownOptimizer.optimize(p);
    Assert.assertTrue(pipelineHasTransform(p, expectedSource));
    Assert.assertFalse(pipelineHasTransform(p, originalSource));
  }