Fix skipping index IT for char column and re-enable Iceberg IT (#349)

* enable Iceberg IT Signed-off-by: Peng Huo <[email protected]> * push down read-padding on char type Signed-off-by: Peng Huo <[email protected]> --------- Signed-off-by: Peng Huo <[email protected]>
opensearch-project · May 21, 2024 · b5eb552 · b5eb552
1 parent 624b488
commit b5eb552
Show file tree

Hide file tree

Showing 4 changed files with 16 additions and 12 deletions.
diff --git a/...ation/src/main/scala/org/opensearch/flint/spark/skipping/FlintSparkSkippingStrategy.scala b/...ation/src/main/scala/org/opensearch/flint/spark/skipping/FlintSparkSkippingStrategy.scala
@@ -11,7 +11,10 @@ import org.opensearch.flint.spark.skipping.FlintSparkSkippingStrategy.SkippingKi
 
 import org.apache.spark.sql.Column
 import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, GetStructField}
+import org.apache.spark.sql.catalyst.expressions.objects.StaticInvoke
+import org.apache.spark.sql.catalyst.util.CharVarcharCodegenUtils
 import org.apache.spark.sql.functions.col
+import org.apache.spark.sql.types.StringType
 
 /**
  * Skipping index strategy that defines skipping data structure building and reading logic.
@@ -115,6 +118,17 @@ object FlintSparkSkippingStrategy {
           Seq(attr.name)
         case GetStructField(child, _, Some(name)) =>
           extractColumnName(child) :+ name
+        /**
+         * Since Spark 3.4 add read-side padding, char_col = "sample char" became
+         * (staticinvoke(class org.apache.spark.sql.catalyst.util.CharVarcharCodegenUtils,
+         * StringType, readSidePadding, char_col#47, 20, true, false, true) = sample char )
+         *
+         * When create skipping index, Spark did write-side padding. So read-side push down can be
+         * ignored. More reading, https://issues.apache.org/jira/browse/SPARK-40697
+         */
+        case StaticInvoke(staticObject, StringType, "readSidePadding", arguments, _, _, _, _)
+            if classOf[CharVarcharCodegenUtils].isAssignableFrom(staticObject) =>
+          extractColumnName(arguments.head)
         case _ => Seq.empty
       }
     }

diff --git a/integ-test/src/test/scala/org/opensearch/flint/spark/FlintSparkSkippingIndexITSuite.scala b/integ-test/src/test/scala/org/opensearch/flint/spark/FlintSparkSkippingIndexITSuite.scala
@@ -797,13 +797,10 @@ class FlintSparkSkippingIndexITSuite extends FlintSparkSuite {
     // CharType column is padded to a fixed length with whitespace
     val paddedChar = "sample char".padTo(20, ' ')
     checkAnswer(query, Row("sample varchar", paddedChar))
-    /*
-     * todo Spark 3.4 add read-side padding, SkippingIndex rule can not push down char_col plan now.
-     *  https://issues.apache.org/jira/browse/SPARK-40697
-     */
     query.queryExecution.executedPlan should
       useFlintSparkSkippingFileIndex(
-        hasIndexFilter(isnull(col("varchar_col")) || col("varchar_col") === "sample varchar"))
+        hasIndexFilter((isnull(col("varchar_col")) || col("varchar_col") === "sample varchar") &&
+          (isnull(col("char_col")) || col("char_col") === paddedChar)))
 
     deleteTestIndex(testIndex)
   }

diff --git a/...t/scala/org/opensearch/flint/spark/iceberg/FlintSparkIcebergMaterializedViewITSuite.scala b/...t/scala/org/opensearch/flint/spark/iceberg/FlintSparkIcebergMaterializedViewITSuite.scala
@@ -7,9 +7,6 @@ package org.opensearch.flint.spark.iceberg
 
 import org.opensearch.flint.spark.FlintSparkMaterializedViewSqlITSuite
 
-// FIXME: https://github.com/opensearch-project/opensearch-spark/issues/331#issuecomment-2110948494
-/*
 class FlintSparkIcebergMaterializedViewITSuite
     extends FlintSparkMaterializedViewSqlITSuite
     with FlintSparkIcebergSuite {}
- */
diff --git a/...test/scala/org/opensearch/flint/spark/iceberg/FlintSparkIcebergSkippingIndexITSuite.scala b/...test/scala/org/opensearch/flint/spark/iceberg/FlintSparkIcebergSkippingIndexITSuite.scala
@@ -5,12 +5,8 @@
 
 package org.opensearch.flint.spark.iceberg
 
-import org.junit.Ignore
 import org.opensearch.flint.spark.FlintSparkSkippingIndexSqlITSuite
 
-// FIXME: https://github.com/opensearch-project/opensearch-spark/issues/331#issuecomment-2110948494
-/*
 class FlintSparkIcebergSkippingIndexITSuite
     extends FlintSparkSkippingIndexSqlITSuite
     with FlintSparkIcebergSuite {}
- */