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 testIntermediateProducer() {
    Pipeline p = Pipeline.create();
    SimpleSource source =
        new SimpleSource(FieldAccessDescriptor.withFieldNames("foo", "bar", "baz"));
    IntermediateTransformWithPushdown originalT =
        new IntermediateTransformWithPushdown(
            FieldAccessDescriptor.withFieldNames("foo", "bar", "baz"));
    FieldAccessDescriptor downstreamFieldAccess =
        FieldAccessDescriptor.withFieldNames("foo", "bar");
    p.apply(source).apply(originalT).apply(new FieldAccessTransform(downstreamFieldAccess));

    // TODO(https://github.com/apache/beam/issues/21359) Support pushdown on intermediate
    // transforms. For now, test that the pushdown optimizer ignores immediate transforms.
    ProjectionPushdownOptimizer.optimize(p);
    Assert.assertTrue(pipelineHasTransform(p, originalT));
  }