[HUDI-2814] Addressing issues w/ Z-order Layout Optimization (#4060)

* `ZCurveOptimizeHelper` > `ZOrderingIndexHelper`; Moved Z-index helper under `hudi.index.zorder` package * Tidying up `ZOrderingIndexHelper` * Fixing compilation * Fixed index new/original table merging sequence to always prefer values from new index; Cleaned up `HoodieSparkUtils` * Added test for `mergeIndexSql` * Abstracted Z-index name composition w/in `ZOrderingIndexHelper`; * Fixed `DataSkippingUtils` to interrupt prunning in case data filter contains non-indexed column reference * Properly handle exceptions origination during pruning in `HoodieFileIndex` * Make sure no errors are logged upon encountering `AnalysisException` * Cleaned up Z-index updating sequence; Tidying up comments, java-docs; * Fixed Z-index to properly handle changes of the list of clustered columns * Tidying up * `lint` * Suppressing `JavaDocStyle` first sentence check * Fixed compilation * Fixing incorrect `DecimalType` conversion * Refactored test `TestTableLayoutOptimization` - Added Z-index table composition test (against fixtures) - Separated out GC test; Tidying up * Fixed tests re-shuffling column order for Z-Index table `DataFrame` to align w/ the one by one loaded from JSON * Scaffolded `DataTypeUtils` to do basic checks of Spark types; Added proper compatibility checking b/w old/new index-tables * Added test for Z-index tables merging * Fixed import being shaded by creating internal `hudi.util` package * Fixed packaging for `TestOptimizeTable` * Revised `updateMetadataIndex` seq to provide Z-index updating process w/ source table schema * Make sure existing Z-index table schema is sync'd to source table's one * Fixed shaded refs * Fixed tests * Fixed type conversion of Parquet provided metadata values into Spark expected schemas * Fixed `composeIndexSchema` utility to propose proper schema * Added more tests for Z-index: - Checking that Z-index table is built correctly - Checking that Z-index tables are merged correctly (during update) * Fixing source table * Fixing tests to read from Parquet w/ proper schema * Refactored `ParquetUtils` utility reading stats from Parquet footers * Fixed incorrect handling of Decimals extracted from Parquet footers * Worked around issues in javac failign to compile stream's collection * Fixed handling of `Date` type * Fixed handling of `DateType` to be parsed as `LocalDate` * Updated fixture; Make sure test loads Z-index fixture using proper schema * Removed superfluous scheme adjusting when reading from Parquet, since Spark is actually able to perfectly restore schema (given Parquet was previously written by Spark as well) * Fixing race-condition in Parquet's `DateStringifier` trying to share `SimpleDataFormat` object which is inherently not thread-safe * Tidying up * Make sure schema is used upon reading to validate input files are in the appropriate format; Tidying up; * Worked around javac (1.8) inability to infer expression type properly * Updated fixtures; Tidying up * Fixing compilation after rebase * Assert clustering have in Z-order layout optimization testing * Tidying up exception messages * XXX * Added test validating Z-index lookup filter correctness * Added more test-cases; Tidying up * Added tests for string expressions * Fixed incorrect Z-index filter lookup translations * Added more test-cases * Added proper handling on complex negations of AND/OR expressions by pushing NOT operator down into inner expressions for appropriate handling * Added `-target:jvm-1.8` for `hudi-spark` module * Adding more tests * Added tests for non-indexed columns * Properly handle non-indexed columns by falling back to a re-write of containing expression as `TrueLiteral` instead * Fixed tests * Removing the parquet test files and disabling corresponding tests Co-authored-by: Vinoth Chandar <vinoth@apache.org>
2021-11-26 10:02:15 -08:00
parent 3d75aca40d
commit 5755ff25a4
28 changed files with 1955 additions and 932 deletions
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/hudi/HoodieFileIndex.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/hudi/HoodieFileIndex.scala
@@ -30,21 +30,24 @@ import org.apache.hudi.common.table.{HoodieTableMetaClient, TableSchemaResolver}

 import org.apache.spark.api.java.JavaSparkContext
 import org.apache.spark.internal.Logging
+import org.apache.spark.sql.avro.SchemaConverters
 import org.apache.spark.sql.{Column, SparkSession}
 import org.apache.spark.sql.catalyst.expressions.{And, AttributeReference, BoundReference, Expression, InterpretedPredicate}
 import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateTimeUtils}
 import org.apache.spark.sql.catalyst.{InternalRow, expressions}
 import org.apache.spark.sql.execution.datasources.{FileIndex, FileStatusCache, NoopCache, PartitionDirectory}
-import org.apache.spark.sql.hudi.{DataSkippingUtils, HoodieSqlUtils}
+import org.apache.spark.sql.hudi.DataSkippingUtils.createZIndexLookupFilter
+import org.apache.spark.sql.hudi.HoodieSqlUtils
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{AnalysisException, Column, SparkSession}
 import org.apache.spark.unsafe.types.UTF8String

 import java.util.Properties
-
-import scala.collection.JavaConverters._
 import scala.collection.JavaConversions._
+import scala.collection.JavaConverters._
 import scala.collection.mutable
+import scala.util.{Failure, Success, Try}

 /**
 * A file index which support partition prune for hoodie snapshot and read-optimized query.
@@ -169,16 +172,16 @@ case class HoodieFileIndex(
   *       ultimately be scanned as part of query execution. Hence, this method has to maintain the
   *       invariant of conservatively including every base-file's name, that is NOT referenced in its index.
   *
-   * @param dataFilters list of original data filters passed down from querying engine
+   * @param queryFilters list of original data filters passed down from querying engine
   * @return list of pruned (data-skipped) candidate base-files' names
   */
-  private def lookupCandidateFilesNamesInZIndex(dataFilters: Seq[Expression]): Option[Set[String]] = {
+  private def lookupCandidateFilesInZIndex(queryFilters: Seq[Expression]): Try[Option[Set[String]]] = Try {
    val indexPath = metaClient.getZindexPath
    val fs = metaClient.getFs

-    if (!enableDataSkipping() || !fs.exists(new Path(indexPath)) || dataFilters.isEmpty) {
+    if (!enableDataSkipping() || !fs.exists(new Path(indexPath)) || queryFilters.isEmpty) {
      // scalastyle:off return
-      return Option.empty
+      return Success(Option.empty)
      // scalastyle:on return
    }

@@ -192,7 +195,7 @@ case class HoodieFileIndex(

    if (candidateIndexTables.isEmpty) {
      // scalastyle:off return
-      return Option.empty
+      return Success(Option.empty)
      // scalastyle:on return
    }

@@ -207,7 +210,7 @@ case class HoodieFileIndex(
    dataFrameOpt.map(df => {
      val indexSchema = df.schema
      val indexFilter =
-        dataFilters.map(DataSkippingUtils.createZIndexLookupFilter(_, indexSchema))
+        queryFilters.map(createZIndexLookupFilter(_, indexSchema))
          .reduce(And)

      logInfo(s"Index filter condition: $indexFilter")
@@ -221,7 +224,7 @@ case class HoodieFileIndex(
          .toSet

      val prunedCandidateFileNames =
-        df.filter(new Column(indexFilter))
+        df.where(new Column(indexFilter))
          .select("file")
          .collect()
          .map(_.getString(0))
@@ -261,11 +264,22 @@ case class HoodieFileIndex(
    //    - Data-skipping is enabled
    //    - Z-index is present
    //    - List of predicates (filters) is present
-    val candidateFilesNamesOpt: Option[Set[String]] = lookupCandidateFilesNamesInZIndex(dataFilters)
+    val candidateFilesNamesOpt: Option[Set[String]] =
+      lookupCandidateFilesInZIndex(dataFilters) match {
+        case Success(opt) => opt
+        case Failure(e) =>
+          if (e.isInstanceOf[AnalysisException]) {
+            logDebug("Failed to relay provided data filters to Z-index lookup", e)
+          } else {
+            logError("Failed to lookup candidate files in Z-index", e)
+          }
+          Option.empty
+      }

    logDebug(s"Overlapping candidate files (from Z-index): ${candidateFilesNamesOpt.getOrElse(Set.empty)}")

-    if (queryAsNonePartitionedTable) { // Read as Non-Partitioned table.
+    if (queryAsNonePartitionedTable) {
+      // Read as Non-Partitioned table
      // Filter in candidate files based on the Z-index lookup
      val candidateFiles =
        allFiles.filter(fileStatus =>
@@ -273,9 +287,10 @@ case class HoodieFileIndex(
          candidateFilesNamesOpt.forall(_.contains(fileStatus.getPath.getName))
        )

-      logInfo(s"Total files : ${allFiles.size}," +
-        s" candidate files after data skipping: ${candidateFiles.size} " +
-        s" skipping percent ${if (allFiles.length != 0) (allFiles.size - candidateFiles.size) / allFiles.size.toDouble else 0}")
+      logInfo(s"Total files : ${allFiles.size}; " +
+        s"candidate files after data skipping: ${candidateFiles.size}; " +
+        s"skipping percent ${if (allFiles.nonEmpty) (allFiles.size - candidateFiles.size) / allFiles.size.toDouble else 0}")
+
      Seq(PartitionDirectory(InternalRow.empty, candidateFiles))
    } else {
      // Prune the partition path by the partition filters
@@ -284,27 +299,27 @@ case class HoodieFileIndex(
      var candidateFileSize = 0

      val result = prunedPartitions.map { partition =>
-        val baseFileStatuses = cachedAllInputFileSlices(partition).map(fileSlice => {
-          if (fileSlice.getBaseFile.isPresent) {
-            fileSlice.getBaseFile.get().getFileStatus
-          } else {
-            null
-          }
-        }).filterNot(_ == null)
+        val baseFileStatuses: Seq[FileStatus] =
+          cachedAllInputFileSlices(partition)
+            .map(fs => fs.getBaseFile.orElse(null))
+            .filter(_ != null)
+            .map(_.getFileStatus)

        // Filter in candidate files based on the Z-index lookup
        val candidateFiles =
-          baseFileStatuses.filter(fileStatus =>
+          baseFileStatuses.filter(fs =>
            // NOTE: This predicate is true when {@code Option} is empty
-            candidateFilesNamesOpt.forall(_.contains(fileStatus.getPath.getName)))
+            candidateFilesNamesOpt.forall(_.contains(fs.getPath.getName)))

        totalFileSize += baseFileStatuses.size
        candidateFileSize += candidateFiles.size
        PartitionDirectory(partition.values, candidateFiles)
      }
-      logInfo(s"Total files: ${totalFileSize}," +
-        s" Candidate files after data skipping : ${candidateFileSize} " +
-        s"skipping percent ${if (allFiles.length != 0) (totalFileSize - candidateFileSize) / totalFileSize.toDouble else 0}")
+
+      logInfo(s"Total base files: ${totalFileSize}; " +
+        s"candidate files after data skipping : ${candidateFileSize}; " +
+        s"skipping percent ${if (allFiles.nonEmpty) (totalFileSize - candidateFileSize) / totalFileSize.toDouble else 0}")
+
      result
    }
  }
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
@@ -54,7 +54,6 @@ import org.apache.spark.sql.types.StructType
 import org.apache.spark.sql.{DataFrame, Dataset, Row, SQLContext, SaveMode, SparkSession}
 import org.apache.spark.{SPARK_VERSION, SparkContext}

-import java.util
 import java.util.Properties

 import scala.collection.JavaConversions._
@@ -289,7 +288,7 @@ object HoodieSparkSqlWriter {
  }

  def generateSchemaWithoutPartitionColumns(partitionParam: String, schema: Schema): Schema = {
-    val fieldsToRemove =  new util.ArrayList[String]()
+    val fieldsToRemove =  new java.util.ArrayList[String]()
    partitionParam.split(",").map(partitionField => partitionField.trim)
      .filter(s => !s.isEmpty).map(field => fieldsToRemove.add(field))
    HoodieAvroUtils.removeFields(schema, fieldsToRemove)
@@ -629,7 +628,7 @@ object HoodieSparkSqlWriter {
        kv._1.startsWith(parameters(COMMIT_METADATA_KEYPREFIX.key)))
      val commitSuccess =
        client.commit(tableInstantInfo.instantTime, writeResult.getWriteStatuses,
-          common.util.Option.of(new util.HashMap[String, String](mapAsJavaMap(metaMap))),
+          common.util.Option.of(new java.util.HashMap[String, String](mapAsJavaMap(metaMap))),
          tableInstantInfo.commitActionType,
          writeResult.getPartitionToReplaceFileIds)

@@ -643,7 +642,7 @@ object HoodieSparkSqlWriter {
      val asyncCompactionEnabled = isAsyncCompactionEnabled(client, tableConfig, parameters, jsc.hadoopConfiguration())
      val compactionInstant: common.util.Option[java.lang.String] =
        if (asyncCompactionEnabled) {
-          client.scheduleCompaction(common.util.Option.of(new util.HashMap[String, String](mapAsJavaMap(metaMap))))
+          client.scheduleCompaction(common.util.Option.of(new java.util.HashMap[String, String](mapAsJavaMap(metaMap))))
        } else {
          common.util.Option.empty()
        }
@@ -653,7 +652,7 @@ object HoodieSparkSqlWriter {
      val asyncClusteringEnabled = isAsyncClusteringEnabled(client, parameters)
      val clusteringInstant: common.util.Option[java.lang.String] =
        if (asyncClusteringEnabled) {
-          client.scheduleClustering(common.util.Option.of(new util.HashMap[String, String](mapAsJavaMap(metaMap))))
+          client.scheduleClustering(common.util.Option.of(new java.util.HashMap[String, String](mapAsJavaMap(metaMap))))
        } else {
          common.util.Option.empty()
        }
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/DataSkippingUtils.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/DataSkippingUtils.scala
@@ -19,9 +19,11 @@ package org.apache.spark.sql.hudi

 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.{FileStatus, Path}
-import org.apache.spark.sql.{AnalysisException, SparkSession}
+import org.apache.hudi.index.zorder.ZOrderingIndexHelper.{getMaxColumnNameFor, getMinColumnNameFor, getNumNullsColumnNameFor}
+import org.apache.spark.internal.Logging
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
+import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
 import org.apache.spark.sql.catalyst.expressions.{Alias, And, Attribute, AttributeReference, EqualNullSafe, EqualTo, Expression, ExtractValue, GetStructField, GreaterThan, GreaterThanOrEqual, In, IsNotNull, IsNull, LessThan, LessThanOrEqual, Literal, Not, Or, StartsWith}
 import org.apache.spark.sql.execution.datasources.PartitionedFile
 import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
@@ -29,181 +31,230 @@ import org.apache.spark.sql.functions.col
 import org.apache.spark.sql.sources.Filter
 import org.apache.spark.sql.types.{StringType, StructType}
 import org.apache.spark.sql.vectorized.ColumnarBatch
+import org.apache.spark.sql.{AnalysisException, SparkSession}
 import org.apache.spark.unsafe.types.UTF8String

 import scala.collection.JavaConverters._

-object DataSkippingUtils {
+object DataSkippingUtils extends Logging {

  /**
   * Translates provided {@link filterExpr} into corresponding filter-expression for Z-index index table
   * to filter out candidate files that would hold records matching the original filter
   *
-   * @param filterExpr  original filter from query
+   * @param sourceFilterExpr  original filter from query
   * @param indexSchema index table schema
   * @return filter for Z-index table
   */
-  def createZIndexLookupFilter(filterExpr: Expression, indexSchema: StructType): Expression = {
-
-    def rewriteCondition(colName: Seq[String], conditionExpress: Expression): Expression = {
-      val stats = Set.apply(
-        UnresolvedAttribute(colName).name + "_minValue",
-        UnresolvedAttribute(colName).name + "_maxValue",
-        UnresolvedAttribute(colName).name + "_num_nulls"
-      )
-
-      if (stats.forall(stat => indexSchema.exists(_.name == stat))) {
-        conditionExpress
-      } else {
-        Literal.TrueLiteral
-      }
+  def createZIndexLookupFilter(sourceFilterExpr: Expression, indexSchema: StructType): Expression = {
+    // Try to transform original Source Table's filter expression into
+    // Column-Stats Index filter expression
+    tryComposeIndexFilterExpr(sourceFilterExpr, indexSchema) match {
+      case Some(e) => e
+      // NOTE: In case we can't transform source filter expression, we fallback
+      // to {@code TrueLiteral}, to essentially avoid pruning any indexed files from scanning
+      case None => TrueLiteral
    }
+  }

-    def refColExpr(colName: Seq[String], statisticValue: String): Expression =
-      col(UnresolvedAttribute(colName).name + statisticValue).expr
+  private def tryComposeIndexFilterExpr(sourceExpr: Expression, indexSchema: StructType): Option[Expression] = {
+    def minValue(colName: String) = col(getMinColumnNameFor(colName)).expr
+    def maxValue(colName: String) = col(getMaxColumnNameFor(colName)).expr
+    def numNulls(colName: String) = col(getNumNullsColumnNameFor(colName)).expr

-    def minValue(colName: Seq[String]) = refColExpr(colName, "_minValue")
-    def maxValue(colName: Seq[String]) = refColExpr(colName, "_maxValue")
-    def numNulls(colName: Seq[String]) = refColExpr(colName, "_num_nulls")
-
-    def colContainsValuesEqualToLiteral(colName: Seq[String], value: Literal) =
+    def colContainsValuesEqualToLiteral(colName: String, value: Literal): Expression =
+    // Only case when column C contains value V is when min(C) <= V <= max(c)
      And(LessThanOrEqual(minValue(colName), value), GreaterThanOrEqual(maxValue(colName), value))

-    def colContainsValuesEqualToLiterals(colName: Seq[String], list: Seq[Literal]) =
-      list.map { lit => colContainsValuesEqualToLiteral(colName, lit) }.reduce(Or)
+    def colContainsOnlyValuesEqualToLiteral(colName: String, value: Literal) =
+    // Only case when column C contains _only_ value V is when min(C) = V AND max(c) = V
+      And(EqualTo(minValue(colName), value), EqualTo(maxValue(colName), value))

-    filterExpr match {
+    sourceExpr match {
      // Filter "colA = b"
      // Translates to "colA_minValue <= b AND colA_maxValue >= b" condition for index lookup
      case EqualTo(attribute: AttributeReference, value: Literal) =>
-        val colName = getTargetColNameParts(attribute)
-        rewriteCondition(colName, colContainsValuesEqualToLiteral(colName, value))
+        getTargetIndexedColName(attribute, indexSchema)
+          .map(colName => colContainsValuesEqualToLiteral(colName, value))
+
      // Filter "b = colA"
      // Translates to "colA_minValue <= b AND colA_maxValue >= b" condition for index lookup
      case EqualTo(value: Literal, attribute: AttributeReference) =>
-        val colName = getTargetColNameParts(attribute)
-        rewriteCondition(colName, colContainsValuesEqualToLiteral(colName, value))
+        getTargetIndexedColName(attribute, indexSchema)
+          .map(colName => colContainsValuesEqualToLiteral(colName, value))
+
+      // Filter "colA != b"
+      // Translates to "NOT(colA_minValue = b AND colA_maxValue = b)"
+      // NOTE: This is NOT an inversion of `colA = b`
+      case Not(EqualTo(attribute: AttributeReference, value: Literal)) =>
+        getTargetIndexedColName(attribute, indexSchema)
+          .map(colName => Not(colContainsOnlyValuesEqualToLiteral(colName, value)))
+
+      // Filter "b != colA"
+      // Translates to "NOT(colA_minValue = b AND colA_maxValue = b)"
+      // NOTE: This is NOT an inversion of `colA = b`
+      case Not(EqualTo(value: Literal, attribute: AttributeReference)) =>
+        getTargetIndexedColName(attribute, indexSchema)
+          .map(colName => Not(colContainsOnlyValuesEqualToLiteral(colName, value)))
+
      // Filter "colA = null"
      // Translates to "colA_num_nulls = null" for index lookup
      case equalNullSafe @ EqualNullSafe(_: AttributeReference, _ @ Literal(null, _)) =>
-        val colName = getTargetColNameParts(equalNullSafe.left)
-        rewriteCondition(colName, EqualTo(numNulls(colName), equalNullSafe.right))
+        getTargetIndexedColName(equalNullSafe.left, indexSchema)
+          .map(colName => EqualTo(numNulls(colName), equalNullSafe.right))
+
      // Filter "colA < b"
      // Translates to "colA_minValue < b" for index lookup
      case LessThan(attribute: AttributeReference, value: Literal) =>
-        val colName = getTargetColNameParts(attribute)
-        rewriteCondition(colName, LessThan(minValue(colName), value))
-      // Filter "b < colA"
-      // Translates to "b < colA_maxValue" for index lookup
-      case LessThan(value: Literal, attribute: AttributeReference) =>
-        val colName = getTargetColNameParts(attribute)
-        rewriteCondition(colName, GreaterThan(maxValue(colName), value))
-      // Filter "colA > b"
-      // Translates to "colA_maxValue > b" for index lookup
-      case GreaterThan(attribute: AttributeReference, value: Literal) =>
-        val colName = getTargetColNameParts(attribute)
-        rewriteCondition(colName, GreaterThan(maxValue(colName), value))
+        getTargetIndexedColName(attribute, indexSchema)
+          .map(colName => LessThan(minValue(colName), value))
+
      // Filter "b > colA"
      // Translates to "b > colA_minValue" for index lookup
      case GreaterThan(value: Literal, attribute: AttributeReference) =>
-        val colName = getTargetColNameParts(attribute)
-        rewriteCondition(colName, LessThan(minValue(colName), value))
+        getTargetIndexedColName(attribute, indexSchema)
+          .map(colName => LessThan(minValue(colName), value))
+
+      // Filter "b < colA"
+      // Translates to "b < colA_maxValue" for index lookup
+      case LessThan(value: Literal, attribute: AttributeReference) =>
+        getTargetIndexedColName(attribute, indexSchema)
+          .map(colName => GreaterThan(maxValue(colName), value))
+
+      // Filter "colA > b"
+      // Translates to "colA_maxValue > b" for index lookup
+      case GreaterThan(attribute: AttributeReference, value: Literal) =>
+        getTargetIndexedColName(attribute, indexSchema)
+          .map(colName => GreaterThan(maxValue(colName), value))
+
      // Filter "colA <= b"
      // Translates to "colA_minValue <= b" for index lookup
      case LessThanOrEqual(attribute: AttributeReference, value: Literal) =>
-        val colName = getTargetColNameParts(attribute)
-        rewriteCondition(colName, LessThanOrEqual(minValue(colName), value))
-      // Filter "b <= colA"
-      // Translates to "b <= colA_maxValue" for index lookup
-      case LessThanOrEqual(value: Literal, attribute: AttributeReference) =>
-        val colName = getTargetColNameParts(attribute)
-        rewriteCondition(colName, GreaterThanOrEqual(maxValue(colName), value))
-      // Filter "colA >= b"
-      // Translates to "colA_maxValue >= b" for index lookup
-      case GreaterThanOrEqual(attribute: AttributeReference, right: Literal) =>
-        val colName = getTargetColNameParts(attribute)
-        rewriteCondition(colName, GreaterThanOrEqual(maxValue(colName), right))
+        getTargetIndexedColName(attribute, indexSchema)
+          .map(colName => LessThanOrEqual(minValue(colName), value))
+
      // Filter "b >= colA"
      // Translates to "b >= colA_minValue" for index lookup
      case GreaterThanOrEqual(value: Literal, attribute: AttributeReference) =>
-        val colName = getTargetColNameParts(attribute)
-        rewriteCondition(colName, LessThanOrEqual(minValue(colName), value))
+        getTargetIndexedColName(attribute, indexSchema)
+          .map(colName => LessThanOrEqual(minValue(colName), value))
+
+      // Filter "b <= colA"
+      // Translates to "b <= colA_maxValue" for index lookup
+      case LessThanOrEqual(value: Literal, attribute: AttributeReference) =>
+        getTargetIndexedColName(attribute, indexSchema)
+          .map(colName => GreaterThanOrEqual(maxValue(colName), value))
+
+      // Filter "colA >= b"
+      // Translates to "colA_maxValue >= b" for index lookup
+      case GreaterThanOrEqual(attribute: AttributeReference, right: Literal) =>
+        getTargetIndexedColName(attribute, indexSchema)
+          .map(colName => GreaterThanOrEqual(maxValue(colName), right))
+
      // Filter "colA is null"
      // Translates to "colA_num_nulls > 0" for index lookup
      case IsNull(attribute: AttributeReference) =>
-        val colName = getTargetColNameParts(attribute)
-        rewriteCondition(colName, GreaterThan(numNulls(colName), Literal(0)))
+        getTargetIndexedColName(attribute, indexSchema)
+          .map(colName => GreaterThan(numNulls(colName), Literal(0)))
+
      // Filter "colA is not null"
      // Translates to "colA_num_nulls = 0" for index lookup
      case IsNotNull(attribute: AttributeReference) =>
-        val colName = getTargetColNameParts(attribute)
-        rewriteCondition(colName, EqualTo(numNulls(colName), Literal(0)))
+        getTargetIndexedColName(attribute, indexSchema)
+          .map(colName => EqualTo(numNulls(colName), Literal(0)))
+
      // Filter "colA in (a, b, ...)"
      // Translates to "(colA_minValue <= a AND colA_maxValue >= a) OR (colA_minValue <= b AND colA_maxValue >= b)" for index lookup
+      // NOTE: This is equivalent to "colA = a OR colA = b OR ..."
      case In(attribute: AttributeReference, list: Seq[Literal]) =>
-        val colName = getTargetColNameParts(attribute)
-        rewriteCondition(colName, colContainsValuesEqualToLiterals(colName, list))
-      // Filter "colA like xxx"
+        getTargetIndexedColName(attribute, indexSchema)
+          .map(colName =>
+            list.map { lit => colContainsValuesEqualToLiteral(colName, lit) }.reduce(Or)
+          )
+
+      // Filter "colA not in (a, b, ...)"
+      // Translates to "NOT((colA_minValue = a AND colA_maxValue = a) OR (colA_minValue = b AND colA_maxValue = b))" for index lookup
+      // NOTE: This is NOT an inversion of `in (a, b, ...)` expr, this is equivalent to "colA != a AND colA != b AND ..."
+      case Not(In(attribute: AttributeReference, list: Seq[Literal])) =>
+        getTargetIndexedColName(attribute, indexSchema)
+          .map(colName =>
+            Not(
+              list.map { lit => colContainsOnlyValuesEqualToLiteral(colName, lit) }.reduce(Or)
+            )
+          )
+
+      // Filter "colA like 'xxx%'"
      // Translates to "colA_minValue <= xxx AND colA_maxValue >= xxx" for index lookup
      // NOTE: That this operator only matches string prefixes, and this is
      //       essentially equivalent to "colA = b" expression
      case StartsWith(attribute, v @ Literal(_: UTF8String, _)) =>
-        val colName = getTargetColNameParts(attribute)
-        rewriteCondition(colName, colContainsValuesEqualToLiteral(colName, v))
-      // Filter "colA not in (a, b, ...)"
-      // Translates to "(colA_minValue > a OR colA_maxValue < a) AND (colA_minValue > b OR colA_maxValue < b)" for index lookup
-      // NOTE: This is an inversion of `in (a, b, ...)` expr
-      case Not(In(attribute: AttributeReference, list: Seq[Literal])) =>
-        val colName = getTargetColNameParts(attribute)
-        rewriteCondition(colName, Not(colContainsValuesEqualToLiterals(colName, list)))
-      // Filter "colA != b"
-      // Translates to "colA_minValue > b OR colA_maxValue < b" (which is an inversion of expr for "colA = b") for index lookup
-      // NOTE: This is an inversion of `colA = b` expr
-      case Not(EqualTo(attribute: AttributeReference, value: Literal)) =>
-        val colName = getTargetColNameParts(attribute)
-        rewriteCondition(colName, Not(colContainsValuesEqualToLiteral(colName, value)))
-      // Filter "b != colA"
-      // Translates to "colA_minValue > b OR colA_maxValue < b" (which is an inversion of expr for "colA = b") for index lookup
-      // NOTE: This is an inversion of `colA != b` expr
-      case Not(EqualTo(value: Literal, attribute: AttributeReference)) =>
-        val colName = getTargetColNameParts(attribute)
-        rewriteCondition(colName, Not(colContainsValuesEqualToLiteral(colName, value)))
-      // Filter "colA not like xxx"
-      // Translates to "!(colA_minValue <= xxx AND colA_maxValue >= xxx)" for index lookup
-      // NOTE: This is a inversion of "colA like xxx" assuming that colA is a string-based type
+        getTargetIndexedColName(attribute, indexSchema)
+          .map(colName => colContainsValuesEqualToLiteral(colName, v))
+
+      // Filter "colA not like 'xxx%'"
+      // Translates to "NOT(colA_minValue like 'xxx%' AND colA_maxValue like 'xxx%')" for index lookup
+      // NOTE: This is NOT an inversion of "colA like xxx"
      case Not(StartsWith(attribute, value @ Literal(_: UTF8String, _))) =>
-        val colName = getTargetColNameParts(attribute)
-        rewriteCondition(colName, Not(colContainsValuesEqualToLiteral(colName, value)))
+        getTargetIndexedColName(attribute, indexSchema)
+          .map(colName =>
+            Not(And(StartsWith(minValue(colName), value), StartsWith(maxValue(colName), value)))
+          )

      case or: Or =>
        val resLeft = createZIndexLookupFilter(or.left, indexSchema)
        val resRight = createZIndexLookupFilter(or.right, indexSchema)
-        Or(resLeft, resRight)
+
+        Option(Or(resLeft, resRight))

      case and: And =>
        val resLeft = createZIndexLookupFilter(and.left, indexSchema)
        val resRight = createZIndexLookupFilter(and.right, indexSchema)
-        And(resLeft, resRight)

-      case expr: Expression =>
-        Literal.TrueLiteral
+        Option(And(resLeft, resRight))
+
+      //
+      // Pushing Logical NOT inside the AND/OR expressions
+      // NOTE: This is required to make sure we're properly handling negations in
+      //       cases like {@code NOT(colA = 0)}, {@code NOT(colA in (a, b, ...)}
+      //
+
+      case Not(And(left: Expression, right: Expression)) =>
+        Option(createZIndexLookupFilter(Or(Not(left), Not(right)), indexSchema))
+
+      case Not(Or(left: Expression, right: Expression)) =>
+        Option(createZIndexLookupFilter(And(Not(left), Not(right)), indexSchema))
+
+      case _: Expression => None
    }
  }

-  /**
-    * Extracts name from a resolved expression referring to a nested or non-nested column.
-    */
-  def getTargetColNameParts(resolvedTargetCol: Expression): Seq[String] = {
+  private def checkColIsIndexed(colName: String, indexSchema: StructType): Boolean = {
+    Set.apply(
+      getMinColumnNameFor(colName),
+      getMaxColumnNameFor(colName),
+      getNumNullsColumnNameFor(colName)
+    )
+      .forall(stat => indexSchema.exists(_.name == stat))
+  }
+
+  private def getTargetIndexedColName(resolvedExpr: Expression, indexSchema: StructType): Option[String] = {
+    val colName = UnresolvedAttribute(getTargetColNameParts(resolvedExpr)).name
+
+    // Verify that the column is indexed
+    if (checkColIsIndexed(colName, indexSchema)) {
+      Option.apply(colName)
+    } else {
+      None
+    }
+  }
+
+  private def getTargetColNameParts(resolvedTargetCol: Expression): Seq[String] = {
    resolvedTargetCol match {
      case attr: Attribute => Seq(attr.name)
-
      case Alias(c, _) => getTargetColNameParts(c)
-
      case GetStructField(c, _, Some(name)) => getTargetColNameParts(c) :+ name
-
      case ex: ExtractValue =>
        throw new AnalysisException(s"convert reference to name failed, Updating nested fields is only supported for StructType: ${ex}.")
-
      case other =>
        throw new AnalysisException(s"convert reference to name failed,  Found unsupported expression ${other}")
    }