code: @Override
      final StreamShape getSourceShape() {
          @SuppressWarnings("rawtypes")
          AbstractPipeline p = AbstractPipeline.this;
          while (p.depth > 0) {
              p = p.previousStage;
          }
          return p.getOutputShape();
      }
docs: PipelineHelper