class PhysicalPlanBuilder {

  public QueryMetadata buildPhysicalPlan(final Pair<String, PlanNode> statementPlanPair) {
    final SchemaKStream resultStream = statementPlanPair
        .getRight()
        .buildStream(
            builder,
            ksqlConfig,
            kafkaTopicClient,
            functionRegistry,
            overriddenStreamsProperties,
            schemaRegistryClient
        );
    final OutputNode outputNode = resultStream.outputNode();
    boolean isBareQuery = outputNode instanceof KsqlBareOutputNode;

    // Check to make sure the logical and physical plans match up;
    // important to do this BEFORE actually starting up
    // the corresponding Kafka Streams job
    if (isBareQuery && !(resultStream instanceof QueuedSchemaKStream)) {
      throw new KsqlException(String.format(
          "Mismatch between logical and physical output; "
          + "expected a QueuedSchemaKStream based on logical "
          + "KsqlBareOutputNode, found a %s instead",
          resultStream.getClass().getCanonicalName()
      ));
    }
    String serviceId = getServiceId();
    String
        persistanceQueryPrefix =
        ksqlConfig.get(KsqlConfig.KSQL_PERSISTENT_QUERY_NAME_PREFIX_CONFIG).toString();
    String
        transientQueryPrefix =
        ksqlConfig.get(KsqlConfig.KSQL_TRANSIENT_QUERY_NAME_PREFIX_CONFIG).toString();

    if (isBareQuery) {

      return buildPlanForBareQuery(
          (QueuedSchemaKStream) resultStream,
          (KsqlBareOutputNode) outputNode,
          serviceId,
          transientQueryPrefix,
          statementPlanPair.getLeft()
      );

    } else if (outputNode instanceof KsqlStructuredDataOutputNode) {

      return buildPlanForStructuredOutputNode(
          statementPlanPair.getLeft(),
          resultStream,
          (KsqlStructuredDataOutputNode) outputNode,
          serviceId,
          persistanceQueryPrefix,
          statementPlanPair.getLeft()
      );

    } else {
      throw new KsqlException(
          "Sink data source of type: "
          + outputNode.getClass()
          + " is not supported.");
    }
  }

  public  PhysicalPlanBuilder(
      final StreamsBuilder builder,
      final KsqlConfig ksqlConfig,
      final KafkaTopicClient kafkaTopicClient,
      final FunctionRegistry functionRegistry,
      final Map<String, Object> overriddenStreamsProperties,
      final boolean updateMetastore,
      final MetaStore metaStore,
      final SchemaRegistryClient schemaRegistryClient,
      final KafkaStreamsBuilder kafkaStreamsBuilder
  );
  public  PhysicalPlanBuilder(
      final StreamsBuilder builder,
      final KsqlConfig ksqlConfig,
      final KafkaTopicClient kafkaTopicClient,
      final FunctionRegistry functionRegistry,
      final Map<String, Object> overriddenStreamsProperties,
      final boolean updateMetastore,
      final MetaStore metaStore,
      final SchemaRegistryClient schemaRegistryClient
  );

}