def params()

in spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/sql/Utils.scala [128:183]
45 lines of code
15 McCabe index (conditional complexity)

  def params(parameters: Map[String, String], logger: Logger) = {
    // '.' seems to be problematic when specifying the options
    val dottedParams = parameters.map { case (k, v) =>
      if (k.startsWith("sink.properties.") || k.startsWith("doris.sink.properties.")) {
        (k, v)
      } else {
        (k.replace('_', '.'), v)
      }
    }

    val preferredTableIdentifier = dottedParams.get(DorisOptions.DORIS_TABLE_IDENTIFIER.getName)
    logger.debug(s"preferred Table Identifier is '$preferredTableIdentifier'.")

    // Convert simple parameters into internal properties, and prefix other parameters
    // Convert password parameters from "password" into internal password properties
    // reuse credentials mask method in spark ExternalCatalogUtils#maskCredentials
    val processedParams = dottedParams.map {
      case ("doris.password", _) =>
        logger.error(s"${DorisOptions.DORIS_PASSWORD.getName} cannot use in Doris Datasource.")
        throw new DorisException(s"${DorisOptions.DORIS_PASSWORD.getName} cannot use in Doris Datasource," +
          s" use 'password' option to set password.")
      case ("doris.user", _) =>
        logger.error(s"${DorisOptions.DORIS_USER.getName} cannot use in Doris Datasource.")
        throw new DorisException(s"${DorisOptions.DORIS_USER.getName} cannot use in Doris Datasource," +
          s" use 'user' option to set user.")
      case (k, v) =>
        if (k.startsWith("doris.")) (k, v)
        else ("doris." + k, v)
    }.map {
      case (DorisOptions.DORIS_REQUEST_AUTH_PASSWORD, _) =>
        logger.error(s"${DorisOptions.DORIS_REQUEST_AUTH_PASSWORD} cannot use in Doris Datasource.")
        throw new DorisException(s"${DorisOptions.DORIS_REQUEST_AUTH_PASSWORD} cannot use in" +
          s" Doris Datasource, use 'password' option to set password.")
      case (DorisOptions.DORIS_REQUEST_AUTH_USER, _) =>
        logger.error(s"${DorisOptions.DORIS_REQUEST_AUTH_USER} cannot use in Doris Datasource.")
        throw new DorisException(s"${DorisOptions.DORIS_REQUEST_AUTH_USER} cannot use in" +
          s" Doris Datasource, use 'user' option to set user.")
      case ("doris.password", v) =>
        (DorisOptions.DORIS_REQUEST_AUTH_PASSWORD, v)
      case ("doris.user", v) =>
        (DorisOptions.DORIS_REQUEST_AUTH_USER, v)
      case (k, v) => (k, v)
    }

    // Set the preferred resource if it was specified originally
    val finalParams = preferredTableIdentifier match {
      case Some(tableIdentifier) => processedParams + (DorisOptions.DORIS_TABLE_IDENTIFIER.getName -> tableIdentifier)
      case None => processedParams
    }

    // validate path is available
    finalParams.getOrElse(DorisOptions.DORIS_TABLE_IDENTIFIER.getName,
      throw new DorisException("table identifier must be specified for doris table identifier."))

    finalParams
  }