class JoinNode extends PlanNode {

  @Override
  public SchemaKStream<?> buildStream(final KsqlQueryBuilder builder) {

    ensureMatchingPartitionCounts(builder.getServiceContext().getTopicClient());

    final JoinerFactory joinerFactory = new JoinerFactory(
        builder,
        this,
        builder.buildNodeContext(getId().toString()));

    return joinerFactory.getJoiner(left.getNodeOutputType(), right.getNodeOutputType()).join();
  }

  public  JoinNode(
      final PlanNodeId id,
      final JoinType joinType,
      final JoinKey joinKey,
      final boolean finalJoin,
      final PlanNode left,
      final PlanNode right,
      final Optional<WithinExpression> withinExpression
  );

}