implement custom boolean parsing logic for CSV and JSON

integration_tests/src/test/resources/boolean.json

@@ -1,3 +1,11 @@
 { "number": true }
+{ "number": True }
+{ "number": TRUE }
 { "number": false }
-{ "number": null }
+{ "number": False }
+{ "number": FALSE }
+{ "number": null }
+{ "number": y }
+{ "number": n }
+{ "number": 0 }
+{ "number": 1 }

integration_tests/src/test/resources/simple_boolean_values.csv

@@ -7,3 +7,11 @@ False
 TRUE
 FALSE
 BAD
+y
+n
+yes
+no
+1
+0
+t
+f

sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuBatchScanExec.scala

@@ -21,7 +21,7 @@ import java.nio.charset.StandardCharsets
 import scala.collection.JavaConverters._
 
 import ai.rapids.cudf
-import ai.rapids.cudf.{HostMemoryBuffer, Schema, Table}
+import ai.rapids.cudf.{ColumnVector, DType, HostMemoryBuffer, Scalar, Schema, Table}
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.Path
 
@@ -423,4 +423,28 @@ class CSVPartitionReader(
    * @return the file format short name
    */
   override def getFileFormatShortName: String = "CSV"
+
+  /**
+   * CSV supports "true" and "false" (case-insensitive) as valid boolean values.
+   */
+  override def castStringToBool(input: ColumnVector): ColumnVector = {
+    withResource(input.strip()) { stripped =>
+      withResource(stripped.lower()) { lower =>
+        withResource(Scalar.fromString("true")) { t =>
+          withResource(Scalar.fromString("false")) { f =>
+            withResource(lower.equalTo(t)) { isTrue =>
+              withResource(lower.equalTo(f)) { isFalse =>
+                withResource(isTrue.or(isFalse)) { isValidBool =>
+                  withResource(Scalar.fromNull(DType.BOOL8)) { nullBool =>
+                    isValidBool.ifElse(isTrue, nullBool)
+                  }
+                }
+              }
+            }
+          }
+        }
+      }
+    }
+  }
+
 }

sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuCast.scala

@@ -849,7 +849,7 @@ object GpuCast extends Arm {
     }
   }
 
-  def castStringToBool(input: ColumnVector, ansiEnabled: Boolean): ColumnVector = {
+  private def castStringToBool(input: ColumnVector, ansiEnabled: Boolean): ColumnVector = {
     val trueStrings = Seq("t", "true", "y", "yes", "1")
     val falseStrings = Seq("f", "false", "n", "no", "0")
     val boolStrings = trueStrings ++ falseStrings

sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuTextBasedPartitionReader.scala

@@ -199,7 +199,7 @@ abstract class GpuTextBasedPartitionReader(
             for (i <- 0 until table.getNumberOfColumns) {
               val castColumn = dataSchema.fields(i).dataType match {
                 case DataTypes.BooleanType =>
-                  GpuCast.castStringToBool(table.getColumn(i), ansiEnabled)
+                  castStringToBool(table.getColumn(i))
                 case DataTypes.FloatType =>
                   GpuCast.castStringToFloats(table.getColumn(i), ansiEnabled, DType.FLOAT32)
                 case DataTypes.DoubleType =>
@@ -220,6 +220,8 @@ abstract class GpuTextBasedPartitionReader(
     }
   }
 
+  def castStringToBool(input: ColumnVector): ColumnVector
+
   /**
    * Read the host buffer to GPU table
    * @param dataBuffer host buffer to be read

sql-plugin/src/main/scala/org/apache/spark/sql/catalyst/json/rapids/GpuJsonScan.scala

@@ -21,7 +21,7 @@ import java.nio.charset.StandardCharsets
 import scala.collection.JavaConverters._
 
 import ai.rapids.cudf
-import ai.rapids.cudf.{HostMemoryBuffer, Schema, Table}
+import ai.rapids.cudf.{ColumnVector, DType, HostMemoryBuffer, Scalar, Schema, Table}
 import com.nvidia.spark.rapids._
 import org.apache.hadoop.conf.Configuration
 
@@ -334,4 +334,24 @@ class JsonPartitionReader(
       Some(new Table(prunedColumnVectors: _*))
     }
   }
+
+  /**
+   * JSON only supports unquoted lower-case "true" and "false" as valid boolean values.
+   */
+  override def castStringToBool(input: ColumnVector): ColumnVector = {
+    withResource(Scalar.fromString("true")) { t =>
+      withResource(Scalar.fromString("false")) { f =>
+        withResource(input.equalTo(t)) { isTrue =>
+          withResource(input.equalTo(f)) { isFalse =>
+            withResource(isTrue.or(isFalse)) { isValidBool =>
+              withResource(Scalar.fromNull(DType.BOOL8)) { nullBool =>
+                isValidBool.ifElse(isTrue, nullBool)
+              }
+            }
+          }
+        }
+      }
+    }
+  }
+
 }