feat(workflow-operator): answer the filter on an empty cell instead of throwing

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/substringSearch/SubstringSearchOpExec.scala

@@ -30,11 +30,19 @@ class SubstringSearchOpExec(descString: String) extends FilterOpExec {
   this.setFilterFunc(findSubstring)
 
   private def findSubstring(tuple: Tuple): Boolean = {
-    val content = tuple.getField(desc.attribute).toString
-    if (desc.isCaseSensitive) {
-      content.contains(desc.substring)
+    val field = tuple.getField[Any](desc.attribute)
+    // A row with nothing in the column matches nothing. FilterPredicate answers the
+    // same way: once a field is null, every condition but IS_NULL / IS_NOT_NULL is
+    // false. An empty cell is ordinary input, since a blank in a CSV arrives as null.
+    if (field == null) {
+      false
     } else {
-      content.toLowerCase.contains(desc.substring.toLowerCase)
+      val content = field.toString
+      if (desc.isCaseSensitive) {
+        content.contains(desc.substring)
+      } else {
+        content.toLowerCase.contains(desc.substring.toLowerCase)
+      }
     }
   }
 }

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/unneststring/UnnestStringOpExec.scala

@@ -29,10 +29,18 @@ class UnnestStringOpExec(descString: String) extends FlatMapOpExec {
   setFlatMapFunc(splitByDelimiter)
 
   private def splitByDelimiter(tuple: Tuple): Iterator[TupleLike] = {
-    desc.delimiter.r
-      .split(tuple.getField(desc.attribute).toString)
-      .filter(_.nonEmpty)
-      .iterator
-      .map(split => TupleLike(tuple.getFields ++ Seq(split)))
+    val field = tuple.getField[Any](desc.attribute)
+    // Nothing in the column unnests to nothing, the same way the filter below drops
+    // the empty pieces a run of delimiters produces. An empty cell is ordinary input,
+    // since a blank in a CSV arrives as null.
+    if (field == null) {
+      Iterator.empty
+    } else {
+      desc.delimiter.r
+        .split(field.toString)
+        .filter(_.nonEmpty)
+        .iterator
+        .map(split => TupleLike(tuple.getFields ++ Seq(split)))
+    }
   }
 }

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/substringSearch/SubstringSearchOpExecSpec.scala

@@ -94,6 +94,24 @@ class SubstringSearchOpExecSpec extends AnyFlatSpec {
     assert(exec.processTuple(t, port = 0).toList == List(t))
   }
 
+  // ---------------------------------------------------------------------------
+  // Edge: empty cell
+  // ---------------------------------------------------------------------------
+
+  it should "yield nothing when the column is empty" in {
+    val exec = new SubstringSearchOpExec(descJson(substring = "hello"))
+    // A blank CSV cell arrives as null. This used to throw a NullPointerException on
+    // the toString instead of answering the filter.
+    assert(exec.processTuple(tuple(null), port = 0).toList.isEmpty)
+  }
+
+  it should "yield nothing when the column is empty and the substring is empty too" in {
+    val exec = new SubstringSearchOpExec(descJson(substring = ""))
+    // The empty substring matches every value, but a row with no value has none to
+    // match, so it is filtered out rather than kept.
+    assert(exec.processTuple(tuple(null), port = 0).toList.isEmpty)
+  }
+
   // ---------------------------------------------------------------------------
   // Edge: empty substring
   // ---------------------------------------------------------------------------

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/unneststring/UnnestStringOpExecSpec.scala

@@ -108,6 +108,24 @@ class UnnestStringOpExecSpec extends AnyFlatSpec with BeforeAndAfter {
     opExec.close()
   }
 
+  it should "produce no rows when the attribute is empty" in {
+    opDesc.attribute = "field1"
+    opDesc.delimiter = "-"
+    opExec = new UnnestStringOpExec(objectMapper.writeValueAsString(opDesc))
+    // A blank CSV cell arrives as null. This used to throw a NullPointerException on
+    // the toString instead of unnesting to nothing.
+    val tuple: Tuple = Tuple
+      .builder(tupleSchema)
+      .add(new Attribute("field1", AttributeType.STRING), null)
+      .add(new Attribute("field2", AttributeType.INTEGER), 1)
+      .add(new Attribute("field3", AttributeType.STRING), "a")
+      .build()
+
+    opExec.open()
+    assert(opExec.processTuple(tuple, 0).isEmpty)
+    opExec.close()
+  }
+
   it should "split by regex delimiter" in {
     opDesc.attribute = "field1"
     opDesc.delimiter = "<\\d*>"