/*
 * 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.JavaConverters._
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

import org.apache.log4j.Logger
import org.apache.spark.rdd.RDD
import org.apache.spark.sql._
import org.apache.spark.sql.CarbonExpressions.{MatchCast => Cast}
import org.apache.spark.sql.carbondata.execution.datasources.{CarbonFileIndex, CarbonSparkDataSourceUtil}
import org.apache.spark.sql.catalyst.{expressions, InternalRow}
import org.apache.spark.sql.catalyst.expressions.{Attribute, _}
import org.apache.spark.sql.catalyst.planning.{ExtractEquiJoinKeys, PhysicalOperation}
import org.apache.spark.sql.catalyst.plans.{Inner, LeftSemi}
import org.apache.spark.sql.catalyst.plans.logical.{Filter => LogicalFilter, _}
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project}
import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning, Partitioning, UnknownPartitioning}
import org.apache.spark.sql.execution._
import org.apache.spark.sql.execution.datasources.{CatalogFileIndex, HadoopFsRelation, InMemoryFileIndex, LogicalRelation, SparkCarbonTableFormat}
import org.apache.spark.sql.execution.joins.{BuildLeft, BuildRight}
import org.apache.spark.sql.hive.MatchLogicalRelation
import org.apache.spark.sql.index.CarbonIndexUtil
import org.apache.spark.sql.optimizer.CarbonFilters
import org.apache.spark.sql.secondaryindex.joins.BroadCastSIFilterPushJoin
import org.apache.spark.sql.sources.{BaseRelation, Filter}
import org.apache.spark.sql.types._
import org.apache.spark.util.{CarbonReflectionUtils, SparkUtil}

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.metadata.schema.BucketingInfo
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.readcommitter.TableStatusReadCommittedScope
import org.apache.carbondata.core.statusmanager.SegmentUpdateStatusManager
import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.geo.{InPolygon, InPolygonUDF}
import org.apache.carbondata.index.{TextMatch, TextMatchLimit, TextMatchMaxDocUDF, TextMatchUDF}
import org.apache.carbondata.spark.rdd.CarbonScanRDD

/**
 * Carbon specific optimization
 * 1. filter push down
 * 2. count star
 */
private[sql] class CarbonLateDecodeStrategy extends SparkStrategy {
  val PUSHED_FILTERS = "PushedFilters"
  val READ_SCHEMA = "ReadSchema"

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

  /*
  Spark 2.3.1 plan there can be case of multiple projections like below
  Project [substring(name, 1, 2)#124, name#123, tupleId#117, cast(rand(-6778822102499951904)#125
  as string) AS rand(-6778822102499951904)#137]
   +- Project [substring(name#123, 1, 2) AS substring(name, 1, 2)#124, name#123, UDF:getTupleId()
    AS tupleId#117, (rand(-6778822102499951904) AS rand(-6778822102499951904)#125]
   +- Relation[imei#118,age#119,task#120L,num#121,level#122,name#123]
   CarbonDatasourceHadoopRelation []
 */
  def apply(plan: LogicalPlan): Seq[SparkPlan] = {
    val transformedPlan = makeDeterministic(plan)
    transformedPlan match {
      case PhysicalOperation(projects, filters, l: LogicalRelation)
        if l.relation.isInstanceOf[CarbonDatasourceHadoopRelation] =>
        val relation = l.relation.asInstanceOf[CarbonDatasourceHadoopRelation]
        // In Spark 2.3.1 there is case of multiple projections like below
        // if 1 projection is failed then need to continue to other
        try {
          pruneFilterProject(
            l,
            projects.filterNot(_.name.equalsIgnoreCase(CarbonCommonConstants.POSITION_ID)),
            filters,
            (a, f, p) =>
              setVectorReadSupport(
                l,
                a,
                relation.buildScan(a.map(_.name).toArray, filters, projects, f, p))
          ) :: Nil
        } catch {
          case e: CarbonPhysicalPlanException => Nil
        }
      case CountStarPlan(colAttr, PhysicalOperation(projectList, predicates, l: LogicalRelation))
        if l.relation.isInstanceOf[CarbonDatasourceHadoopRelation] && driverSideCountStar(l) =>
        val relation = l.relation.asInstanceOf[CarbonDatasourceHadoopRelation]
        CarbonCountStar(colAttr, relation.carbonTable, SparkSession.getActiveSession.get) :: Nil
      case ExtractEquiJoinKeys(Inner, leftKeys, rightKeys, condition,
      left, right)
        if isCarbonPlan(left) && CarbonIndexUtil.checkIsIndexTable(right) =>
        LOGGER.info(s"pushing down for ExtractEquiJoinKeys:right")
        val carbon = apply(left).head
        // in case of SI Filter push join remove projection list from the physical plan
        // no need to have the project list in the main table physical plan execution
        // only join uses the projection list
        var carbonChild = carbon match {
          case projectExec: ProjectExec =>
            projectExec.child
          case _ =>
            carbon
        }
        // check if the outer and the inner project are matching, only then remove project
        if (left.isInstanceOf[Project]) {
          val leftOutput = left.output
            .filterNot(attr => attr.name
              .equalsIgnoreCase(CarbonCommonConstants.POSITION_ID))
            .map(c => (c.name.toLowerCase, c.dataType))
          val childOutput = carbonChild.output
            .filterNot(attr => attr.name
              .equalsIgnoreCase(CarbonCommonConstants.POSITION_ID))
            .map(c => (c.name.toLowerCase, c.dataType))
          if (!leftOutput.equals(childOutput)) {
            // if the projection list and the scan list are different(in case of alias)
            // we should not skip the project, so we are taking the original plan with project
            carbonChild = carbon
          }
        }
        val pushedDownJoin = BroadCastSIFilterPushJoin(
          leftKeys: Seq[Expression],
          rightKeys: Seq[Expression],
          Inner,
          BuildRight,
          carbonChild,
          planLater(right),
          condition)
        condition.map(FilterExec(_, pushedDownJoin)).getOrElse(pushedDownJoin) :: Nil
      case ExtractEquiJoinKeys(Inner, leftKeys, rightKeys, condition, left,
      right)
        if isCarbonPlan(right) && CarbonIndexUtil.checkIsIndexTable(left) =>
        LOGGER.info(s"pushing down for ExtractEquiJoinKeys:left")
        val carbon = planLater(right)

        val pushedDownJoin =
          BroadCastSIFilterPushJoin(
            leftKeys: Seq[Expression],
            rightKeys: Seq[Expression],
            Inner,
            BuildLeft,
            planLater(left),
            carbon,
            condition)
        condition.map(FilterExec(_, pushedDownJoin)).getOrElse(pushedDownJoin) :: Nil
      case ExtractEquiJoinKeys(LeftSemi, leftKeys, rightKeys, condition,
      left, right)
        if isLeftSemiExistPushDownEnabled &&
            isAllCarbonPlan(left) && isAllCarbonPlan(right) =>
        LOGGER.info(s"pushing down for ExtractEquiJoinKeysLeftSemiExist:right")
        val carbon = planLater(left)
        val pushedDownJoin = BroadCastSIFilterPushJoin(
          leftKeys: Seq[Expression],
          rightKeys: Seq[Expression],
          LeftSemi,
          BuildRight,
          carbon,
          planLater(right),
          condition)
        condition.map(FilterExec(_, pushedDownJoin)).getOrElse(pushedDownJoin) :: Nil
      case _ => Nil
    }
  }

  private def isAllCarbonPlan(plan: LogicalPlan): Boolean = {
    val allRelations = plan.collect { case logicalRelation: LogicalRelation => logicalRelation }
    allRelations.forall(x => x.relation.isInstanceOf[CarbonDatasourceHadoopRelation])
  }

  private def isCarbonPlan(plan: LogicalPlan): Boolean = {
    plan match {
      case PhysicalOperation(_, _,
      MatchLogicalRelation(_: CarbonDatasourceHadoopRelation, _, _)) =>
        true
      case LogicalFilter(_, MatchLogicalRelation(_: CarbonDatasourceHadoopRelation, _, _)) =>
        true
      case _ => false
    }
  }

  private def isLeftSemiExistPushDownEnabled: Boolean = {
    CarbonProperties.getInstance.getProperty(
      CarbonCommonConstants.CARBON_PUSH_LEFTSEMIEXIST_JOIN_AS_IN_FILTER,
      CarbonCommonConstants.CARBON_PUSH_LEFTSEMIEXIST_JOIN_AS_IN_FILTER_DEFAULT).toBoolean
  }

  /**
   * get deterministic expression for NamedExpression
   */
  private def makeDeterministicExp(exp: NamedExpression): Expression = {
    exp match {
      case alias: Alias if alias.child.isInstanceOf[CustomDeterministicExpression] =>
        alias
      case _ =>
        CustomDeterministicExpression(exp)
    }
  }

  /**
   * Convert all Expression to deterministic Expression
   */
  private def makeDeterministic(plan: LogicalPlan): LogicalPlan = {
    val transformedPlan = plan transform {
      case p@Project(projectList: Seq[NamedExpression], cd) =>
        if (cd.isInstanceOf[org.apache.spark.sql.catalyst.plans.logical.Filter] ||
            cd.isInstanceOf[LogicalRelation]) {
          p.transformAllExpressions {
            case a@Alias(exp, _)
              if !exp.deterministic && !exp.isInstanceOf[CustomDeterministicExpression] =>
              CarbonToSparkAdapter.createAliasRef(
                CustomDeterministicExpression(exp),
                a.name,
                a.exprId,
                a.qualifier,
                a.explicitMetadata,
                Some(a))
            case exp: NamedExpression
              if !exp.deterministic && !exp.isInstanceOf[CustomDeterministicExpression] =>
              makeDeterministicExp(exp)
          }
        } else {
          p
        }
      case f@org.apache.spark.sql.catalyst.plans.logical.Filter(condition: Expression, cd) =>
        if (cd.isInstanceOf[Project] || cd.isInstanceOf[LogicalRelation]) {
          f.transformAllExpressions {
            case a@Alias(exp, _)
              if !exp.deterministic && !exp.isInstanceOf[CustomDeterministicExpression] =>
              CarbonToSparkAdapter.createAliasRef(
                CustomDeterministicExpression(exp),
                a.name,
                a.exprId,
                a.qualifier,
                a.explicitMetadata,
                Some(a))
            case exp: NamedExpression
              if !exp.deterministic && !exp.isInstanceOf[CustomDeterministicExpression] =>
              makeDeterministicExp(exp)
          }
        } else {
          f
        }
    }
    transformedPlan
  }

  /**
   * Return true if driver-side count star optimization can be used.
   * Following case can't use driver-side count star:
   * 1. There is data update and delete
   * 2. It is streaming table
   */
  private def driverSideCountStar(logicalRelation: LogicalRelation): Boolean = {
    val relation = logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation]
    val segmentUpdateStatusManager = new SegmentUpdateStatusManager(
      relation.carbonRelation.carbonTable)
    val updateDeltaMetadata = segmentUpdateStatusManager.readLoadMetadata()
    val hasNonCarbonSegment =
      segmentUpdateStatusManager.getLoadMetadataDetails.exists(!_.isCarbonFormat)
    if (hasNonCarbonSegment || updateDeltaMetadata != null && updateDeltaMetadata.nonEmpty) {
      false
    } else if (relation.carbonTable.isStreamingSink) {
      false
    } else {
      true
    }
  }

  /**
   * Converts to physical RDD of carbon after pushing down applicable filters.
   * @return
   */
  def pruneFilterProject(
      relation: LogicalRelation,
      projects: Seq[NamedExpression],
      filterPredicates: Seq[Expression],
      scanBuilder: (Seq[Attribute], Array[Filter], Seq[PartitionSpec]) => RDD[InternalRow])
  : CodegenSupport = {
    val names = relation.catalogTable match {
      case Some(table) => table.partitionColumnNames
      case _ => Seq.empty
    }
    // Get the current partitions from table.
    var partitions: Seq[PartitionSpec] = null
    if (names.nonEmpty) {
      val partitionSet = AttributeSet(names
        .map(p => relation.output.find(_.name.equalsIgnoreCase(p)).get))
      val partitionKeyFilters = CarbonToSparkAdapter
        .getPartitionKeyFilter(partitionSet, filterPredicates)
      // Update the name with lower case as it is case sensitive while getting partition info.
      val updatedPartitionFilters = partitionKeyFilters.map { exp =>
        exp.transform {
          case attr: AttributeReference =>
            CarbonToSparkAdapter.createAttributeReference(
              attr.name.toLowerCase,
              attr.dataType,
              attr.nullable,
              attr.metadata,
              attr.exprId,
              attr.qualifier)
        }
      }
      partitions =
        CarbonFilters.getPartitions(
          updatedPartitionFilters.toSeq,
          SparkSession.getActiveSession.get,
          relation.catalogTable.get.identifier).orNull
    }
    pruneFilterProjectRaw(
      relation,
      projects,
      filterPredicates,
      partitions,
      (requestedColumns, _, pushedFilters, p) => {
        scanBuilder(requestedColumns, pushedFilters.toArray, p)
      })
  }

  def setVectorReadSupport(
      relation: LogicalRelation,
      output: Seq[Attribute],
      rdd: RDD[InternalRow]): RDD[InternalRow] = {
    rdd.asInstanceOf[CarbonScanRDD[InternalRow]]
      .setVectorReaderSupport(supportBatchedDataSource(relation.relation.sqlContext, output))
    rdd
  }

  protected def pruneFilterProjectRaw(
      relation: LogicalRelation,
      rawProjects: Seq[NamedExpression],
      filterPredicates: Seq[Expression],
      partitions: Seq[PartitionSpec],
      scanBuilder: (Seq[Attribute], Seq[Expression], Seq[Filter], Seq[PartitionSpec])
        => RDD[InternalRow]): CodegenSupport = {
    val table = relation.relation.asInstanceOf[CarbonDatasourceHadoopRelation]
    val readCommittedScope = new TableStatusReadCommittedScope(
      table.identifier,
      FileFactory.getConfiguration)
    val extraRdd = MixedFormatHandler.extraRDD(
      relation,
      rawProjects,
      filterPredicates,
      readCommittedScope,
      table.identifier)
    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 candidatePredicates = filterPredicates.map {
      _ transform {
        case a: AttributeReference => relation.attributeMap(a) // Match original case of attributes.
      }
    }

    val (unhandledPredicates, pushedFilters, handledFilters ) =
      selectFilters(relation.relation, candidatePredicates)

    // A set of column attributes that are only referenced by pushed down filters.  We can eliminate
    // them from requested columns.
    val handledSet = {
      val handledPredicates = filterPredicates.filterNot(unhandledPredicates.contains)
      val unhandledSet = AttributeSet(unhandledPredicates.flatMap(_.references))
      try {
        AttributeSet(handledPredicates.flatMap(_.references)) --
        (projectSet ++ unhandledSet).map(relation.attributeMap)
      } catch {
        case e: Throwable => throw new CarbonPhysicalPlanException
      }
    }
    // Combines all Catalyst filter `Expression`s that are either not convertible to data source
    // `Filter`s or cannot be handled by `relation`.
    val filterCondition = unhandledPredicates.reduceLeftOption(expressions.And)

    val metadata: Map[String, String] = {
      val pairs = ArrayBuffer.empty[(String, String)]

      if (pushedFilters.nonEmpty) {
        pairs += (PUSHED_FILTERS -> pushedFilters.mkString("[", ", ", "]"))
      }
      pairs += (READ_SCHEMA -> projectSet.++(filterSet).toSeq.toStructType.catalogString)
      pairs.toMap
    }

    var vectorPushRowFilters = CarbonProperties.getInstance().isPushRowFiltersForVector

    // Spark cannot filter the rows from the pages in case of polygon query. So, we do the row
    // level filter at carbon and return the rows directly.
    if (candidatePredicates
      .exists(exp => exp.isInstanceOf[ScalaUDF] &&
                     exp.asInstanceOf[ScalaUDF].function.isInstanceOf[InPolygonUDF])) {
      vectorPushRowFilters = true
    }

    // In case of mixed format, make the vectorPushRowFilters always false as other formats
    // filtering happens in spark layer.
    if (vectorPushRowFilters && extraRdd.isDefined) {
      vectorPushRowFilters = false
    }
    if (projects.map(_.toAttribute) == projects &&
        projectSet.size == projects.size &&
        filterSet.subsetOf(projectSet)) {
      // When it is possible to just use column pruning to get the right projection and
      // when the columns of this projection are enough to evaluate all filter conditions,
      // just do a scan followed by a filter, with no extra project.
      val requestedColumns = projects
        // Safe due to if above.
        .asInstanceOf[Seq[Attribute]]
        // Match original case of attributes.
        .map(relation.attributeMap)
        // Don't request columns that are only referenced by pushed filters.
        .filterNot(handledSet.contains)
      val updateRequestedColumns = requestedColumns

      val updateProject = projects.map { expr =>
        expr.toAttribute.asInstanceOf[AttributeReference]
      }
      val scan = getDataSourceScan(
        relation,
        (updateProject, partitions),
        scanBuilder,
        candidatePredicates,
        pushedFilters,
        handledFilters,
        metadata,
        updateRequestedColumns.asInstanceOf[Seq[Attribute]],
        extraRdd)
      if (scan.isInstanceOf[CarbonDataSourceScan] && table.carbonTable.isTransactionalTable) {
        // set the read committed scope and reuse in the scanRDD
        scan.inputRDDs().head match {
          case rdd: CarbonScanRDD[InternalRow] =>
            rdd.setReadCommittedScope(readCommittedScope)
          case _ =>
        }
      }
      // Check whether spark should handle row filters in case of vector flow.
      if (!vectorPushRowFilters && scan.isInstanceOf[CarbonDataSourceScan]) {
        // Here carbon only do page pruning and row level pruning will be done by spark.
        scan.inputRDDs().head match {
          case rdd: CarbonScanRDD[InternalRow] =>
            rdd.setDirectScanSupport(true)
          case _ =>
        }
        filterPredicates.reduceLeftOption(expressions.And).map(execution.FilterExec(_, scan))
          .getOrElse(scan)
      } else if (extraRdd.isDefined) {
        filterPredicates.reduceLeftOption(expressions.And).map(execution.FilterExec(_, scan))
          .getOrElse(scan)
      } else {
        filterCondition.map(execution.FilterExec(_, scan)).getOrElse(scan)
      }
    } else {

      var newProjectList: Seq[Attribute] = Seq.empty
      // In case of implicit exist we should disable vectorPushRowFilters as it goes in IUD flow
      // to get the positionId or tupleID
      var implicitExisted = false
      var updatedProjects = projects.map {
          case a@Alias(s: ScalaUDF, name)
            if name.equalsIgnoreCase(CarbonCommonConstants.POSITION_ID) ||
                name.equalsIgnoreCase(CarbonCommonConstants.CARBON_IMPLICIT_COLUMN_TUPLEID) =>
            val reference = AttributeReference(name, StringType, true)().withExprId(a.exprId)
            newProjectList :+= reference
            implicitExisted = true
            reference
          case a@Alias(s: ScalaUDF, name)
            if name.equalsIgnoreCase(CarbonCommonConstants.CARBON_IMPLICIT_COLUMN_SEGMENTID) =>
            implicitExisted = true
            val reference =
              AttributeReference(CarbonCommonConstants.CARBON_IMPLICIT_COLUMN_TUPLEID,
                StringType, true)().withExprId(a.exprId)
            newProjectList :+= reference
            a.transform {
              case s: ScalaUDF =>
                CarbonToSparkAdapter.createScalaUDF(s, reference)
            }
          case other => other
      }
      val updatedColumns: (Seq[Attribute], Seq[Expression]) = getRequestedColumns(relation,
        projectsAttr,
        filterSet,
        handledSet,
        newProjectList,
        updatedProjects)
      // Don't request columns that are only referenced by pushed filters.
      val requestedColumns = updatedColumns._1
      updatedProjects = updatedColumns._2

      var updateRequestedColumns =
        if (!vectorPushRowFilters && !implicitExisted) {
          (projectsAttr.to[mutable.LinkedHashSet] ++ filterSet).map(relation.attributeMap).toSeq
        } else {
          requestedColumns
        }
      val supportBatch =
        supportBatchedDataSource(relation.relation.sqlContext,
          updateRequestedColumns) && extraRdd.getOrElse((null, true))._2
      if (!vectorPushRowFilters && !supportBatch && !implicitExisted && filterSet.nonEmpty &&
          !filterSet.baseSet.exists(_.a.dataType.isInstanceOf[ArrayType])) {
        // revert for row scan
        updateRequestedColumns = requestedColumns
      }
      val newRequestedColumns = if (!vectorPushRowFilters && extraRdd.isDefined) {
        extractUniqueAttributes(projectsAttr, filterSet.toSeq)
      } else {
        updateRequestedColumns
      }
      val scan = getDataSourceScan(
        relation,
        (newRequestedColumns, partitions),
        scanBuilder,
        candidatePredicates,
        pushedFilters,
        handledFilters,
        metadata,
        newRequestedColumns,
        extraRdd)
      if (scan.isInstanceOf[CarbonDataSourceScan] && table.carbonTable.isTransactionalTable) {
        // set the read committed scope and reuse in the scanRDD
        scan.inputRDDs().head match {
          case rdd: CarbonScanRDD[InternalRow] =>
            rdd.setReadCommittedScope(readCommittedScope)
          case _ =>
        }
      }
      // Check whether spark should handle row filters in case of vector flow.
      if (!vectorPushRowFilters &&
          scan.isInstanceOf[CarbonDataSourceScan] &&
          !implicitExisted) {
        // Here carbon only do page pruning and row level pruning will be done by spark.
        scan.inputRDDs().head match {
          case rdd: CarbonScanRDD[InternalRow] =>
            rdd.setDirectScanSupport(true)
          case _ =>
        }
        execution.ProjectExec(
          updatedProjects.asInstanceOf[Seq[NamedExpression]],
          filterPredicates.reduceLeftOption(expressions.And).map(
            execution.FilterExec(_, scan)).getOrElse(scan))
      } else if (extraRdd.isDefined) {
        execution.ProjectExec(
          updatedProjects.asInstanceOf[Seq[NamedExpression]],
          filterPredicates.reduceLeftOption(expressions.And).map(
            execution.FilterExec(_, scan)).getOrElse(scan))
      } else {
        execution.ProjectExec(
          updatedProjects.asInstanceOf[Seq[NamedExpression]],
          filterCondition.map(execution.FilterExec(_, scan)).getOrElse(scan))
      }

    }
  }

  /*
    This function is used to get the Unique attributes from filter set
    and projection set based on their semantics.
   */
  private def extractUniqueAttributes(projections: Seq[Attribute],
      filter: Seq[Attribute]): Seq[Attribute] = {
    def checkSemanticEquals(filter: Attribute): Option[Attribute] = {
      projections.find(_.semanticEquals(filter))
    }

    filter.toList match {
      case head :: tail =>
        checkSemanticEquals(head) match {
          case Some(_) => extractUniqueAttributes(projections, tail)
          case None => extractUniqueAttributes(projections :+ head, tail)
        }
      case Nil => projections
    }
  }

  protected def getRequestedColumns(relation: LogicalRelation,
      projectsAttr: Seq[Attribute],
      filterSet: AttributeSet,
      handledSet: AttributeSet,
      newProjectList: Seq[Attribute],
      updatedProjects: Seq[Expression]): (Seq[Attribute], Seq[Expression]) = {
    ((projectsAttr.to[mutable.LinkedHashSet] ++ filterSet -- handledSet)
       .map(relation.attributeMap).toSeq ++ newProjectList, updatedProjects)
  }

  private def getDataSourceScan(relation: LogicalRelation,
      outputAndPartitions: (Seq[Attribute], Seq[PartitionSpec]),
      scanBuilder: (Seq[Attribute], Seq[Expression], Seq[Filter],
        Seq[PartitionSpec]) => RDD[InternalRow],
      candidatePredicates: Seq[Expression],
      pushedFilters: Seq[Filter], handledFilters: Seq[Filter],
      metadata: Map[String, String],
      updateRequestedColumns: Seq[Attribute],
      extraRDD: Option[(RDD[InternalRow], Boolean)]): DataSourceScanExec = {
    val table = relation.relation.asInstanceOf[CarbonDatasourceHadoopRelation]
    var rdd = scanBuilder(updateRequestedColumns, candidatePredicates,
      pushedFilters, outputAndPartitions._2)
    if (supportBatchedDataSource(relation.relation.sqlContext, updateRequestedColumns) &&
      extraRDD.getOrElse((rdd, true))._2) {
      rdd = extraRDD.map(_._1.union(rdd)).getOrElse(rdd)
      new CarbonDataSourceScan(
        outputAndPartitions._1,
        rdd,
        createHadoopFSRelation(relation),
        getPartitioning(table.carbonTable, updateRequestedColumns),
        metadata,
        relation.catalogTable.map(_.identifier), relation)
    } else {
      rdd match {
        case cs: CarbonScanRDD[InternalRow] => cs.setVectorReaderSupport(false)
        case _ =>
      }
      rdd = extraRDD.map(_._1.union(rdd)).getOrElse(rdd)
      val partition = getPartitioning(table.carbonTable, updateRequestedColumns)
      CarbonReflectionUtils.getRowDataSourceScanExecObj(relation, outputAndPartitions._1,
        pushedFilters, handledFilters,
        rdd, partition, metadata)
    }
  }

  private def getPartitioning(carbonTable: CarbonTable,
      output: Seq[Attribute]): Partitioning = {
    val info: BucketingInfo = carbonTable.getBucketingInfo()
    if (info != null) {
      val cols = info.getListOfColumns.asScala
      val numBuckets = info.getNumOfRanges
      val bucketColumns = cols.flatMap { n =>
        val attrRef = output.find(_.name.equalsIgnoreCase(n.getColumnName))
        attrRef match {
          case Some(attr: AttributeReference) =>
            Some(AttributeReference(attr.name,
              CarbonSparkDataSourceUtil.convertCarbonToSparkDataType(n.getDataType),
              attr.nullable,
              attr.metadata)(attr.exprId, attr.qualifier))
          case _ => None
        }
      }
      if (bucketColumns.size == cols.size) {
        // use HashPartitioning will not shuffle
        HashPartitioning(bucketColumns, numBuckets)
      } else {
        UnknownPartitioning(0)
      }
    } else {
      UnknownPartitioning(0)
    }
  }

  private def isComplexAttribute(attribute: Attribute) = attribute.dataType match {
    case ArrayType(dataType, _) => true
    case StructType(_) => true
    case MapType(_, _, _) => true
    case _ => false
  }

  protected[sql] def selectFilters(
      relation: BaseRelation,
      predicates: Seq[Expression]): (Seq[Expression], Seq[Filter], Seq[Filter]) = {

    // In case of ComplexType dataTypes no filters should be pushed down. IsNotNull is being
    // explicitly added by spark and pushed. That also has to be handled and pushed back to
    // Spark for handling.
    // allow array_contains() push down
    val filteredPredicates = predicates.filter { predicate =>
      predicate.isInstanceOf[ArrayContains] ||
      predicate.collect {
        case a: Attribute if isComplexAttribute(a) => a
      }.isEmpty
    }

    // For conciseness, all Catalyst filter expressions of type `expressions.Expression` below are
    // called `predicate`s, while all data source filters of type `sources.Filter` are simply called
    // `filter`s. And block filters for lucene with more than one text_match udf
    // Todo: handle when lucene and normal query filter is supported

    var count = 0
    val translated: Seq[(Expression, Filter)] = filteredPredicates.flatMap {
      predicate =>
        if (predicate.isInstanceOf[ScalaUDF]) {
          predicate match {
            case u: ScalaUDF if u.function.isInstanceOf[TextMatchUDF] ||
                                u.function.isInstanceOf[TextMatchMaxDocUDF] => count = count + 1
            case _ =>
          }
        }
        if (count > 1) {
          throw new MalformedCarbonCommandException(
            "Specify all search filters for Lucene within a single text_match UDF")
        }
        val filter = translateFilter(predicate)
        if (filter.isDefined) {
          Some(predicate, filter.get)
        } else {
          None
        }
    }

    // A map from original Catalyst expressions to corresponding translated data source filters.
    val translatedMap: Map[Expression, Filter] = translated.toMap


    // Catalyst predicate expressions that cannot be translated to data source filters.
    val unrecognizedPredicates = predicates.filterNot(translatedMap.contains)

    // Data source filters that cannot be handled by `relation`. The semantic of a unhandled filter
    // at here is that a data source may not be able to apply this filter to every row
    // of the underlying dataset.
    val unhandledFilters = relation.unhandledFilters(translatedMap.values.toArray).toSet

    val (unhandled, handled) = translated.partition {
      case (predicate, filter) =>
        unhandledFilters.contains(filter)
    }

    // Catalyst predicate expressions that can be translated to data source filters, but cannot be
    // handled by `relation`.
    val (unhandledPredicates, _) = unhandled.unzip

    // Translated data source filters that can be handled by `relation`
    val (_, handledFilters) = handled.unzip

    // translated contains all filters that have been converted to the public Filter interface.
    // We should always push them to the data source no matter whether the data source can apply
    // a filter to every row or not.
    val (_, translatedFilters) = translated.unzip

    (unrecognizedPredicates ++ unhandledPredicates, translatedFilters, handledFilters)
  }

  /**
   * Tries to translate a Catalyst [[Expression]] into data source [[Filter]].
   *
   * @return a `Some[Filter]` if the input [[Expression]] is convertible, otherwise a `None`.
   */
  protected[sql] def translateFilter(predicate: Expression, or: Boolean = false): Option[Filter] = {
    predicate match {
      case u: ScalaUDF if u.function.isInstanceOf[TextMatchUDF] =>
        if (u.children.size > 1) {
          throw new MalformedCarbonCommandException(
            "TEXT_MATCH UDF syntax: TEXT_MATCH('luceneQuerySyntax')")
        }
        Some(TextMatch(u.children.head.toString()))

      case u: ScalaUDF if u.function.isInstanceOf[TextMatchMaxDocUDF] =>
        if (u.children.size > 2) {
          throw new MalformedCarbonCommandException(
            "TEXT_MATCH UDF syntax: TEXT_MATCH_LIMIT('luceneQuerySyntax')")
        }
        Some(TextMatchLimit(u.children.head.toString(), u.children.last.toString()))

      case u: ScalaUDF if u.function.isInstanceOf[InPolygonUDF] =>
        if (u.children.size > 1) {
          throw new MalformedCarbonCommandException("Expect one string in polygon")
        }
        Some(InPolygon(u.children.head.toString()))

      case or@Or(left, right) =>

        val leftFilter = translateFilter(left, true)
        val rightFilter = translateFilter(right, true)
        if (leftFilter.isDefined && rightFilter.isDefined) {
          Some(sources.Or(leftFilter.get, rightFilter.get))
        } else {
          None
        }

      case And(left, right) =>
        val leftFilter = translateFilter(left, or)
        val rightFilter = translateFilter(right, or)
        if (or) {
          if (leftFilter.isDefined && rightFilter.isDefined) {
            (translateFilter(left) ++ translateFilter(right)).reduceOption(sources.And)
          } else {
            None
          }
        } else {
          (translateFilter(left) ++ translateFilter(right)).reduceOption(sources.And)
        }
      case EqualTo(a: Attribute, Literal(v, t)) =>
        Some(sources.EqualTo(a.name, v))
      case EqualTo(l@Literal(v, t), a: Attribute) =>
        Some(sources.EqualTo(a.name, v))
      case c@EqualTo(Cast(a: Attribute, _), Literal(v, t)) =>
        CastExpressionOptimization.checkIfCastCanBeRemove(c)
      case c@EqualTo(Literal(v, t), Cast(a: Attribute, _)) =>
        CastExpressionOptimization.checkIfCastCanBeRemove(c)
      case Not(EqualTo(a: Attribute, Literal(v, t))) =>
        Some(sources.Not(sources.EqualTo(a.name, v)))
      case Not(EqualTo(Literal(v, t), a: Attribute)) =>
        Some(sources.Not(sources.EqualTo(a.name, v)))
      case c@Not(EqualTo(Cast(a: Attribute, _), Literal(v, t))) =>
        CastExpressionOptimization.checkIfCastCanBeRemove(c)
      case c@Not(EqualTo(Literal(v, t), Cast(a: Attribute, _))) =>
        CastExpressionOptimization.checkIfCastCanBeRemove(c)
      case IsNotNull(a: Attribute) => Some(sources.IsNotNull(a.name))
      case IsNull(a: Attribute) => Some(sources.IsNull(a.name))
      case Not(In(a: Attribute, list)) if !list.exists(!_.isInstanceOf[Literal]) =>
        val hSet = list.map(e => e.eval(EmptyRow))
        Some(sources.Not(sources.In(a.name, hSet.toArray)))
      case In(a: Attribute, list) if !list.exists(!_.isInstanceOf[Literal]) =>
        val hSet = list.map(e => e.eval(EmptyRow))
        Some(sources.In(a.name, hSet.toArray))
      case c@Not(In(Cast(a: Attribute, _), list))
        if !list.exists(!_.isInstanceOf[Literal]) =>
        Some(CastExpr(c))
      case c@In(Cast(a: Attribute, _), list) if !list.exists(!_.isInstanceOf[Literal]) =>
        Some(CastExpr(c))
      case InSet(a: Attribute, set) =>
        Some(sources.In(a.name, set.toArray))
      case Not(InSet(a: Attribute, set)) =>
        Some(sources.Not(sources.In(a.name, set.toArray)))
      case GreaterThan(a: Attribute, Literal(v, t)) =>
        Some(sources.GreaterThan(a.name, v))
      case GreaterThan(Literal(v, t), a: Attribute) =>
        Some(sources.LessThan(a.name, v))
      case c@GreaterThan(Cast(a: Attribute, _), Literal(v, t)) =>
        CastExpressionOptimization.checkIfCastCanBeRemove(c)
      case c@GreaterThan(Literal(v, t), Cast(a: Attribute, _)) =>
        CastExpressionOptimization.checkIfCastCanBeRemove(c)
      case LessThan(a: Attribute, Literal(v, t)) =>
        Some(sources.LessThan(a.name, v))
      case LessThan(Literal(v, t), a: Attribute) =>
        Some(sources.GreaterThan(a.name, v))
      case c@LessThan(Cast(a: Attribute, _), Literal(v, t)) =>
        CastExpressionOptimization.checkIfCastCanBeRemove(c)
      case c@LessThan(Literal(v, t), Cast(a: Attribute, _)) =>
        CastExpressionOptimization.checkIfCastCanBeRemove(c)
      case GreaterThanOrEqual(a: Attribute, Literal(v, t)) =>
        Some(sources.GreaterThanOrEqual(a.name, v))
      case GreaterThanOrEqual(Literal(v, t), a: Attribute) =>
        Some(sources.LessThanOrEqual(a.name, v))
      case c@GreaterThanOrEqual(Cast(a: Attribute, _), Literal(v, t)) =>
        CastExpressionOptimization.checkIfCastCanBeRemove(c)
      case c@GreaterThanOrEqual(Literal(v, t), Cast(a: Attribute, _)) =>
        CastExpressionOptimization.checkIfCastCanBeRemove(c)
      case LessThanOrEqual(a: Attribute, Literal(v, t)) =>
        Some(sources.LessThanOrEqual(a.name, v))
      case LessThanOrEqual(Literal(v, t), a: Attribute) =>
        Some(sources.GreaterThanOrEqual(a.name, v))
      case c@LessThanOrEqual(Cast(a: Attribute, _), Literal(v, t)) =>
        CastExpressionOptimization.checkIfCastCanBeRemove(c)
      case c@LessThanOrEqual(Literal(v, t), Cast(a: Attribute, _)) =>
        CastExpressionOptimization.checkIfCastCanBeRemove(c)
      case s@StartsWith(a: Attribute, Literal(v, t)) =>
        Some(sources.StringStartsWith(a.name, v.toString))
      case c@EndsWith(a: Attribute, Literal(v, t)) =>
        Some(CarbonEndsWith(c))
      case c@Contains(a: Attribute, Literal(v, t)) =>
        Some(CarbonContainsWith(c))
      case c@Literal(v, t) if (v == null) =>
        Some(FalseExpr())
      case c@ArrayContains(a: Attribute, Literal(v, t)) =>
        a.dataType match {
          case arrayType: ArrayType =>
            if (SparkUtil.isPrimitiveType(arrayType.elementType)) {
              Some(sources.EqualTo(a.name, v))
            } else {
              None
            }
          case _ =>
            None
        }
      case c@ArrayContains(Cast(a: Attribute, _), Literal(v, t)) =>
        CastExpressionOptimization.checkIfCastCanBeRemove(
          EqualTo(
            predicate.asInstanceOf[ArrayContains].left,
            predicate.asInstanceOf[ArrayContains].right))
      case _ => None
    }
  }

  def supportBatchedDataSource(sqlContext: SQLContext, cols: Seq[Attribute]): Boolean = {
    val vectorizedReader = {
      if (sqlContext.sparkSession.conf.contains(CarbonCommonConstants.ENABLE_VECTOR_READER)) {
        sqlContext.sparkSession.conf.get(CarbonCommonConstants.ENABLE_VECTOR_READER)
      } else if (System.getProperty(CarbonCommonConstants.ENABLE_VECTOR_READER) != null) {
        System.getProperty(CarbonCommonConstants.ENABLE_VECTOR_READER)
      } else {
        CarbonProperties.getInstance().getProperty(CarbonCommonConstants.ENABLE_VECTOR_READER,
          CarbonCommonConstants.ENABLE_VECTOR_READER_DEFAULT)
      }
    }
    val supportCodegen =
      sqlContext.conf.wholeStageEnabled && sqlContext.conf.wholeStageMaxNumFields >= cols.size
    supportCodegen && vectorizedReader.toBoolean &&
    cols.forall(_.dataType.isInstanceOf[AtomicType])
  }

  private def createHadoopFSRelation(relation: LogicalRelation): HadoopFsRelation = {
    val sparkSession = relation.relation.sqlContext.sparkSession
    relation.catalogTable match {
      case Some(catalogTable) =>
        val fileIndex = CarbonFileIndex(sparkSession,
          catalogTable.schema,
          catalogTable.storage.properties,
          new CatalogFileIndex(
          sparkSession,
          catalogTable,
          sizeInBytes = relation.relation.sizeInBytes))
        fileIndex.setDummy(true)
        HadoopFsRelation(
          fileIndex,
          catalogTable.partitionSchema,
          catalogTable.schema,
          catalogTable.bucketSpec,
          new SparkCarbonTableFormat,
          catalogTable.storage.properties)(sparkSession)
      case _ =>
        HadoopFsRelation(
          new InMemoryFileIndex(sparkSession, Seq.empty, Map.empty, None),
          new StructType(),
          relation.relation.schema,
          None,
          new SparkCarbonTableFormat,
          null)(sparkSession)
    }
  }
}

class CarbonPhysicalPlanException extends Exception