docs: def complete(self, stream):
              """Complete the pending stream.
  
              Any connections made to :py:attr:`stream` are connected to `stream` once
              this method returns.
  
              Args:
                  stream(Stream): Stream that completes the connection.
              """
              assert not self.is_complete()
              self._marker.addInputPort(outputPort=stream.oport)
              self.stream.oport.schema = stream.oport.schema
              # Update the pending schema to the actual schema
              # Any downstream filters that took the reference
              # will be automatically updated to the correct schema
              self._pending_schema._set(self.stream.oport.schema)
  
              # Mark the operator with the pending stream
              # a start point for graph travesal
              stream.oport.operator._start_op = True