<commit_before>/**
 * Copyright (C) 2015 Stratio (http://stratio.com)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *         http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.stratio.crossdata.connector.cassandra


import com.datastax.driver.core.{ProtocolVersion, ResultSet}
import com.stratio.crossdata.connector.cassandra.CassandraAttributeRole._
import org.apache.spark.Logging
import org.apache.spark.sql.cassandra.{CassandraSQLRow, CassandraXDSourceRelation}
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.crossdata.catalyst.planning.ExtendedPhysicalOperation
import org.apache.spark.sql.crossdata.execution.{NativeUDF, NativeUDFAttribute}
import org.apache.spark.sql.sources.CatalystToCrossdataAdapter.{AggregationLogicalPlan, BaseLogicalPlan, SimpleLogicalPlan}
import org.apache.spark.sql.sources.{CatalystToCrossdataAdapter, Filter => SourceFilter}
import org.apache.spark.sql.types.DataTypes
import org.apache.spark.sql.{Row, sources}

object CassandraQueryProcessor {

  val DefaultLimit = 10000
  type ColumnName = String

  case class CassandraPlan(basePlan: BaseLogicalPlan, limit: Option[Int])


  def apply(cassandraRelation: CassandraXDSourceRelation, logicalPlan: LogicalPlan) = new CassandraQueryProcessor(cassandraRelation, logicalPlan)

  def buildNativeQuery(
                        tableQN: String,
                        requiredColumns: Seq[String],
                        filters: Array[SourceFilter],
                        limit: Int,
                        udfs: Map[String, NativeUDF] = Map.empty): String = {

    def quoteString(in: Any): String = in match {
      case s: String => s"'$s'"
      case a: Attribute => expandAttribute(a.toString)
      case other => other.toString
    }

    // UDFs are string references in both filters and projects => lookup in udfsMap
    def expandAttribute(att: String): String = {
      udfs get (att) map { udf =>
        val actualParams = udf.children.collect { // TODO: Add type checker (maybe not here)
          case at: NativeUDFAttribute => expandAttribute(at.toString)
          case at: AttributeReference => at.name
          case lit @ Literal(_, DataTypes.StringType) => quoteString(lit.toString)
          case lit: Literal => lit.toString
        } mkString ","
        s"${udf.name}($actualParams)"
      } getOrElse att.split("#").head.trim // TODO: Try a more sophisticated way...
    }

    def filterToCQL(filter: SourceFilter): String = filter match {

      case sources.EqualTo(attribute, value) => s"${expandAttribute(attribute)} = ${quoteString(value)}"
      case sources.In(attribute, values) => s"${expandAttribute(attribute)} IN ${values.map(quoteString).mkString("(", ",", ")")}"
      case sources.LessThan(attribute, value) => s"${expandAttribute(attribute)} < ${quoteString(value)}"
      case sources.GreaterThan(attribute, value) => s"${expandAttribute(attribute)} > ${quoteString(value)}"
      case sources.LessThanOrEqual(attribute, value) => s"${expandAttribute(attribute)} <= ${quoteString(value)}"
      case sources.GreaterThanOrEqual(attribute, value) => s"${expandAttribute(attribute)} >= ${quoteString(value)}"
      case sources.And(leftFilter, rightFilter) => s"${filterToCQL(leftFilter)} AND ${filterToCQL(rightFilter)}"

    }

    val filter = if (filters.nonEmpty) filters.map(filterToCQL).mkString("WHERE ", " AND ", "") else ""
    val columns = requiredColumns.map(expandAttribute).mkString(", ")

    s"SELECT $columns FROM $tableQN $filter LIMIT $limit ALLOW FILTERING"
  }

}

// TODO logs, doc, tests
class CassandraQueryProcessor(cassandraRelation: CassandraXDSourceRelation, logicalPlan: LogicalPlan) extends Logging {

  import CassandraQueryProcessor._

  def execute(): Option[Array[Row]] = {
    def annotateRepeatedNames(names: Seq[String]): Seq[String] = {
      val indexedNames = names zipWithIndex
      val m = indexedNames.groupBy(_._1).values.flatMap(_.zipWithIndex.map(x => x._1._2 -> x._2)).toMap
      indexedNames map { case (name, index) => val c = m(index); if (c > 0) s"$name$c" else name }
    }
<commit_msg>rename variable<commit_after>/**
 * Copyright (C) 2015 Stratio (http://stratio.com)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *         http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.stratio.crossdata.connector.cassandra


import com.datastax.driver.core.{ProtocolVersion, ResultSet}
import com.stratio.crossdata.connector.cassandra.CassandraAttributeRole._
import org.apache.spark.Logging
import org.apache.spark.sql.cassandra.{CassandraSQLRow, CassandraXDSourceRelation}
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.crossdata.catalyst.planning.ExtendedPhysicalOperation
import org.apache.spark.sql.crossdata.execution.{NativeUDF, NativeUDFAttribute}
import org.apache.spark.sql.sources.CatalystToCrossdataAdapter.{AggregationLogicalPlan, BaseLogicalPlan, SimpleLogicalPlan}
import org.apache.spark.sql.sources.{CatalystToCrossdataAdapter, Filter => SourceFilter}
import org.apache.spark.sql.types.DataTypes
import org.apache.spark.sql.{Row, sources}

object CassandraQueryProcessor {

  val DefaultLimit = 10000
  type ColumnName = String

  case class CassandraPlan(basePlan: BaseLogicalPlan, limit: Option[Int])


  def apply(cassandraRelation: CassandraXDSourceRelation, logicalPlan: LogicalPlan) = new CassandraQueryProcessor(cassandraRelation, logicalPlan)

  def buildNativeQuery(
                        tableQN: String,
                        requiredColumns: Seq[String],
                        filters: Array[SourceFilter],
                        limit: Int,
                        udfs: Map[String, NativeUDF] = Map.empty): String = {

    def quoteString(in: Any): String = in match {
      case s: String => s"'$s'"
      case a: Attribute => expandAttribute(a.toString)
      case other => other.toString
    }

    // UDFs are string references in both filters and projects => lookup in udfsMap
    def expandAttribute(att: String): String = {
      udfs get (att) map { udf =>
        val actualParams = udf.children.collect { // TODO: Add type checker (maybe not here)
          case at: NativeUDFAttribute => expandAttribute(at.toString)
          case at: AttributeReference => at.name
          case lit @ Literal(_, DataTypes.StringType) => quoteString(lit.toString)
          case lit: Literal => lit.toString
        } mkString ","
        s"${udf.name}($actualParams)"
      } getOrElse att.split("#").head.trim // TODO: Try a more sophisticated way...
    }

    def filterToCQL(filter: SourceFilter): String = filter match {

      case sources.EqualTo(attribute, value) => s"${expandAttribute(attribute)} = ${quoteString(value)}"
      case sources.In(attribute, values) => s"${expandAttribute(attribute)} IN ${values.map(quoteString).mkString("(", ",", ")")}"
      case sources.LessThan(attribute, value) => s"${expandAttribute(attribute)} < ${quoteString(value)}"
      case sources.GreaterThan(attribute, value) => s"${expandAttribute(attribute)} > ${quoteString(value)}"
      case sources.LessThanOrEqual(attribute, value) => s"${expandAttribute(attribute)} <= ${quoteString(value)}"
      case sources.GreaterThanOrEqual(attribute, value) => s"${expandAttribute(attribute)} >= ${quoteString(value)}"
      case sources.And(leftFilter, rightFilter) => s"${filterToCQL(leftFilter)} AND ${filterToCQL(rightFilter)}"

    }

    val filter = if (filters.nonEmpty) filters.map(filterToCQL).mkString("WHERE ", " AND ", "") else ""
    val columns = requiredColumns.map(expandAttribute).mkString(", ")

    s"SELECT $columns FROM $tableQN $filter LIMIT $limit ALLOW FILTERING"
  }

}

// TODO logs, doc, tests
class CassandraQueryProcessor(cassandraRelation: CassandraXDSourceRelation, logicalPlan: LogicalPlan) extends Logging {

  import CassandraQueryProcessor._

  def execute(): Option[Array[Row]] = {
    def annotateRepeatedNames(names: Seq[String]): Seq[String] = {
      val indexedNames = names zipWithIndex
      val name2pos = indexedNames.groupBy(_._1).values.flatMap(_.zipWithIndex.map(x => x._1._2 -> x._2)).toMap
      indexedNames map { case (name, index) => val c = name2pos(index); if (c > 0) s"$name$c" else name }
    }

{'fraction_non_alphanumeric': 0.11852331606217617, 'fraction_numerical': 0.004101899827288429, 'mean_word_length': 3.9229543039319874, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 49, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5111952', 'n_tokens_mistral': 2726, 'n_tokens_neox': 2622, 'n_words': 671}