<commit_before>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 org.apache.spark.sql.execution.strategy

import scala.collection.mutable

import org.apache.log4j.Logger
import org.apache.spark.rdd.RDD
import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.{expressions, InternalRow, TableIdentifier}
import org.apache.spark.sql.catalyst.catalog.CatalogTablePartition
import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, _}
import org.apache.spark.sql.catalyst.planning.PhysicalOperation
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, _}
import org.apache.spark.sql.execution._
import org.apache.spark.sql.execution.datasources.LogicalRelation
import org.apache.spark.sql.execution.strategy.CarbonPlanHelper.vectorReaderEnabled
import org.apache.spark.sql.optimizer.CarbonFilters
import org.apache.spark.sql.types._

import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.indexstore.PartitionSpec
import org.apache.carbondata.core.readcommitter.TableStatusReadCommittedScope
import org.apache.carbondata.core.scan.expression.{Expression => CarbonFilter}
import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.geo.{GeoHashUtils, GeoIdToGridXyUDF, GeoIdToLatLngUDF, InPolygonListUDF, InPolygonRangeListUDF, InPolygonUDF, InPolylineListUDF, LatLngToGeoIdUDF, ToRangeListUDF, ToUpperLayerGeoIdUDF}
import org.apache.carbondata.hadoop.CarbonProjection
import org.apache.carbondata.index.{TextMatchMaxDocUDF, TextMatchUDF}

private[sql] object CarbonSourceStrategy extends SparkStrategy {
  val PUSHED_FILTERS = "PushedFilters"
  val READ_SCHEMA = "ReadSchema"

  val LOGGER: Logger = LogServiceFactory.getLogService(this.getClass.getName)

  def apply(plan: LogicalPlan): Seq[SparkPlan] = {
    val transformedPlan = CarbonPlanHelper.makeDeterministic(plan)
    transformedPlan match {
      case GlobalLimit(IntegerLiteral(limit), LocalLimit(IntegerLiteral(limitValue),
      _@PhysicalOperation(projects, filters, l: LogicalRelation))) if isCarbonRelation(l) =>
        GlobalLimitExec(limit, LocalLimitExec(limitValue, pruneFilterProject(l,
          projects.filterNot(_.name.equalsIgnoreCase(CarbonCommonConstants.POSITION_ID)),
          filters))) :: Nil
      case PhysicalOperation(projects, filters, l: LogicalRelation) if isCarbonRelation(l) =>
        try {
          pruneFilterProject(l, projects, filters) :: Nil
        } catch {
          case _: CarbonPhysicalPlanException => Nil
        }
      case Project(projectList, _: OneRowRelation) if validateUdf(projectList) => Nil
      case _ => Nil
    }
  }

  private def validateUdf(projects: Seq[NamedExpression]): Boolean = {
    projects foreach {
      case alias: Alias if alias.child.isInstanceOf[Expression] =>
        alias.child match {
          case Cast(s: ScalaUDF, _, _) => validateGeoUtilUDFs(s)
          case s: ScalaUDF => validateGeoUtilUDFs(s)
          case _ =>
        }
    }
    true
  }

  def validateGeoUtilUDFs(s: ScalaUDF): Boolean = {
    s.function match {
      case _: ToRangeListUDF =>
        val inputNames = List("polygon", "oriLatitude", "gridSize")
        for (i <- s.children.indices) {
          GeoHashUtils.validateUDFInputValue(s.children(i),
            inputNames(i),
            CarbonToSparkAdapter.getTypeName(s.inputTypes(i)))
        }
      case _: LatLngToGeoIdUDF =>
        val inputNames = List("latitude", "longitude", "oriLatitude", "gridSize")
        for (i <- s.children.indices) {
          GeoHashUtils.validateUDFInputValue(s.children(i),
            inputNames(i),
            CarbonToSparkAdapter.getTypeName(s.inputTypes(i)))
        }
      case _: GeoIdToGridXyUDF =>
        GeoHashUtils.validateUDFInputValue(s.children.head,
          "geoId",
          CarbonToSparkAdapter.getTypeName(s.inputTypes.head))
      case _: GeoIdToLatLngUDF =>
        val inputNames = List("geoId", "oriLatitude", "gridSize")
        for (i <- s.children.indices) {
          GeoHashUtils.validateUDFInputValue(s.children(i),
            inputNames(i),
            CarbonToSparkAdapter.getTypeName(s.inputTypes(i)))
        }
      case _: ToUpperLayerGeoIdUDF =>
        GeoHashUtils.validateUDFInputValue(s.children.head,
          "geoId",
          CarbonToSparkAdapter.getTypeName(s.inputTypes.head))
      case _ =>
    }
    true
  }

  private def isCarbonRelation(logicalRelation: LogicalRelation): Boolean = {
    logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation]
  }

  private def getPartitionFilter(relation: LogicalRelation,
      filterPredicates: Seq[Expression], names: Seq[String]): (Seq[Expression], AttributeSet) = {
    // Get the current partitions from table.
    if (names.nonEmpty) {
      val partitionSet = AttributeSet(names
        .map(p => relation.output.find(_.name.equalsIgnoreCase(p)).get))
      // Update the name with lower case as it is case sensitive while getting partition info.
      (CarbonToSparkAdapter
        .getPartitionFilter(partitionSet, filterPredicates)
        .map(CarbonToSparkAdapter.lowerCaseAttribute), partitionSet)
    } else {
      (Seq.empty, AttributeSet.empty)
    }
  }

  /**
   * Converts to physical RDD of carbon after pushing down applicable filters.
   * @return
   */
  private def pruneFilterProject(
      relation: LogicalRelation,
      rawProjects: Seq[NamedExpression],
      allPredicates: Seq[Expression]): SparkPlan = {
    // get partition column names
    val names = relation.catalogTable.map(_.partitionColumnNames).getOrElse(Seq.empty)
    // get partition set from filter expression
    val (partitionsFilter, partitionSet) = getPartitionFilter(relation, allPredicates, names)
    var partitions : (Seq[CatalogTablePartition], Seq[PartitionSpec], Seq[Expression]) =
      (null, null, Seq.empty)
    var filterPredicates = allPredicates
    if(names.nonEmpty && partitionsFilter.nonEmpty) {
      partitions = CarbonFilters.getCatalogTablePartitions(
        partitionsFilter.filterNot(e => e.find(_.isInstanceOf[PlanExpression[_]]).isDefined),
        SparkSession.getActiveSession.get,
        relation.catalogTable.get.identifier
      )
    }
    // remove dynamic partition filter from predicates
    filterPredicates = CarbonToSparkAdapter.getDataFilter(partitionSet,
      allPredicates, partitionsFilter)
    val table = relation.relation.asInstanceOf[CarbonDatasourceHadoopRelation]
    val projects = rawProjects.map {p =>
      p.transform {
        case CustomDeterministicExpression(exp) => exp
      }
    }.asInstanceOf[Seq[NamedExpression]]

    // contains the original order of the projection requested
    val projectsAttr = projects.flatMap(_.references)
    val projectSet = AttributeSet(projectsAttr)
    val filterSet = AttributeSet(filterPredicates.flatMap(_.references))

    val relationPredicates = filterPredicates.map {
      _ transform {
        case a: AttributeReference => relation.attributeMap(a) // Match original case of attributes.
      }
    }

    val (unhandledPredicates, handledPredicates, pushedFilters) =
      selectFilters(table, relationPredicates)<commit_msg>[CARBONDATA-4306] Fix Query Performance issue for Spark 3.1<commit_after>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 org.apache.spark.sql.execution.strategy

import scala.collection.mutable

import org.apache.log4j.Logger
import org.apache.spark.rdd.RDD
import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.{expressions, InternalRow, TableIdentifier}
import org.apache.spark.sql.catalyst.catalog.CatalogTablePartition
import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, _}
import org.apache.spark.sql.catalyst.planning.PhysicalOperation
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, _}
import org.apache.spark.sql.execution._
import org.apache.spark.sql.execution.datasources.LogicalRelation
import org.apache.spark.sql.execution.strategy.CarbonPlanHelper.vectorReaderEnabled
import org.apache.spark.sql.optimizer.CarbonFilters
import org.apache.spark.sql.types._

import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.indexstore.PartitionSpec
import org.apache.carbondata.core.readcommitter.TableStatusReadCommittedScope
import org.apache.carbondata.core.scan.expression.{Expression => CarbonFilter}
import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.geo.{GeoHashUtils, GeoIdToGridXyUDF, GeoIdToLatLngUDF, InPolygonListUDF, InPolygonRangeListUDF, InPolygonUDF, InPolylineListUDF, LatLngToGeoIdUDF, ToRangeListUDF, ToUpperLayerGeoIdUDF}
import org.apache.carbondata.hadoop.CarbonProjection
import org.apache.carbondata.index.{TextMatchMaxDocUDF, TextMatchUDF}

private[sql] object CarbonSourceStrategy extends SparkStrategy {
  val PUSHED_FILTERS = "PushedFilters"
  val READ_SCHEMA = "ReadSchema"

  val LOGGER: Logger = LogServiceFactory.getLogService(this.getClass.getName)

  def apply(plan: LogicalPlan): Seq[SparkPlan] = {
    val transformedPlan = CarbonPlanHelper.makeDeterministic(plan)
    transformedPlan match {
      case GlobalLimit(IntegerLiteral(limit), LocalLimit(IntegerLiteral(limitValue),
      _@PhysicalOperation(projects, filters, l: LogicalRelation))) if isCarbonRelation(l) =>
        GlobalLimitExec(limit, LocalLimitExec(limitValue, pruneFilterProject(l,
          projects.filterNot(_.name.equalsIgnoreCase(CarbonCommonConstants.POSITION_ID)),
          filters))) :: Nil
      case PhysicalOperation(projects, filters, l: LogicalRelation) if isCarbonRelation(l) =>
        try {
          pruneFilterProject(l, projects, filters) :: Nil
        } catch {
          case _: CarbonPhysicalPlanException => Nil
        }
      case Project(projectList, _: OneRowRelation) if validateUdf(projectList) => Nil
      case _ => Nil
    }
  }

  private def validateUdf(projects: Seq[NamedExpression]): Boolean = {
    projects foreach {
      case alias: Alias if alias.child.isInstanceOf[Expression] =>
        alias.child match {
          case Cast(s: ScalaUDF, _, _) => validateGeoUtilUDFs(s)
          case s: ScalaUDF => validateGeoUtilUDFs(s)
          case _ =>
        }
    }
    true
  }

  def validateGeoUtilUDFs(s: ScalaUDF): Boolean = {
    s.function match {
      case _: ToRangeListUDF =>
        val inputNames = List("polygon", "oriLatitude", "gridSize")
        for (i <- s.children.indices) {
          GeoHashUtils.validateUDFInputValue(s.children(i),
            inputNames(i),
            CarbonToSparkAdapter.getTypeName(s.inputTypes(i)))
        }
      case _: LatLngToGeoIdUDF =>
        val inputNames = List("latitude", "longitude", "oriLatitude", "gridSize")
        for (i <- s.children.indices) {
          GeoHashUtils.validateUDFInputValue(s.children(i),
            inputNames(i),
            CarbonToSparkAdapter.getTypeName(s.inputTypes(i)))
        }
      case _: GeoIdToGridXyUDF =>
        GeoHashUtils.validateUDFInputValue(s.children.head,
          "geoId",
          CarbonToSparkAdapter.getTypeName(s.inputTypes.head))
      case _: GeoIdToLatLngUDF =>
        val inputNames = List("geoId", "oriLatitude", "gridSize")
        for (i <- s.children.indices) {
          GeoHashUtils.validateUDFInputValue(s.children(i),
            inputNames(i),
            CarbonToSparkAdapter.getTypeName(s.inputTypes(i)))
        }
      case _: ToUpperLayerGeoIdUDF =>
        GeoHashUtils.validateUDFInputValue(s.children.head,
          "geoId",
          CarbonToSparkAdapter.getTypeName(s.inputTypes.head))
      case _ =>
    }
    true
  }

  private def isCarbonRelation(logicalRelation: LogicalRelation): Boolean = {
    logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation]
  }

  private def getPartitionFilter(relation: LogicalRelation,
      filterPredicates: Seq[Expression], names: Seq[String]): (Seq[Expression], AttributeSet) = {
    // Get the current partitions from table.
    if (names.nonEmpty) {
      val partitionSet = AttributeSet(names
        .map(p => relation.output.find(_.name.equalsIgnoreCase(p)).get))
      // Update the name with lower case as it is case sensitive while getting partition info.
      (CarbonToSparkAdapter
        .getPartitionFilter(partitionSet, filterPredicates)
        .map(CarbonToSparkAdapter.lowerCaseAttribute), partitionSet)
    } else {
      (Seq.empty, AttributeSet.empty)
    }
  }

  /**
   * Converts to physical RDD of carbon after pushing down applicable filters.
   * @return
   */
  private def pruneFilterProject(
      relation: LogicalRelation,
      rawProjects: Seq[NamedExpression],
      allPredicates: Seq[Expression]): SparkPlan = {
    // get partition column names
    val names = relation.catalogTable.map(_.partitionColumnNames).getOrElse(Seq.empty)
    // get partition set from filter expression
    val (partitionsFilter, partitionSet) = getPartitionFilter(relation, allPredicates, names)
    var partitions : (Seq[CatalogTablePartition], Seq[PartitionSpec], Seq[Expression]) =
      (null, null, Seq.empty)
    var filterPredicates = allPredicates
    if(names.nonEmpty && partitionsFilter.nonEmpty) {
      partitions = CarbonFilters.getCatalogTablePartitions(
        partitionsFilter.filterNot(e => e.find(_.isInstanceOf[PlanExpression[_]]).isDefined),
        SparkSession.getActiveSession.get,
        relation.catalogTable.get.identifier
      )
      // remove dynamic partition filter from predicates
      filterPredicates = CarbonToSparkAdapter.getDataFilter(partitionSet,
        allPredicates, partitionsFilter)
    }
    val table = relation.relation.asInstanceOf[CarbonDatasourceHadoopRelation]
    val projects = rawProjects.map {p =>
      p.transform {
        case CustomDeterministicExpression(exp) => exp
      }
    }.asInstanceOf[Seq[NamedExpression]]

    // contains the original order of the projection requested
    val projectsAttr = projects.flatMap(_.references)
    val projectSet = AttributeSet(projectsAttr)
    val filterSet = AttributeSet(filterPredicates.flatMap(_.references))

    val relationPredicates = filterPredicates.map {
      _ transform {
        case a: AttributeReference => relation.attributeMap(a) // Match original case of attributes.
      }
    }

    val (unhandledPredicates, handledPredicates, pushedFilters) =
      selectFilters(table, relationPredicates)
{'fraction_non_alphanumeric': 0.09293771456596371, 'fraction_numerical': 0.000980872976949485, 'mean_word_length': 3.820626477541371, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 45, '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': '6894539', 'n_tokens_mistral': 4824, 'n_tokens_neox': 4709, 'n_words': 994}