Fixing filters so they work for qualified column names

Author: rjrudin

docs/reading.md

@@ -71,9 +71,10 @@ down the following operations to MarkLogic:
 - `count`
 - `drop` and `select`
 - `filter` and `where`
-- `orderBy`
+- `groupBy` when followed by `count`
 - `limit`
 - `offset`
+- `orderBy`
 
 For each of the above operations, the Optic pipeline associated with the user's Optic DSL query is modified to include
 the associated Optic function. Note that if multiple partitions are used to perform the `read` operation, each 

src/main/java/com/marklogic/spark/reader/PlanUtil.java

@@ -20,7 +20,7 @@
  * Methods for modifying a serialized Optic plan. These were moved here both to facilitate unit testing for some of them
  * and to simplify {@code ReadContext}.
  */
-abstract class PlanUtil {
+public abstract class PlanUtil {
 
     private final static Logger logger = LoggerFactory.getLogger(PlanUtil.class);
 
@@ -73,9 +73,20 @@ static ObjectNode buildSelect(StructType schema) {
         });
     }
 
-    private static void populateSchemaCol(ObjectNode node, String columnName) {
+    /**
+     * Knows how to handle any of 3 variations of a column name: 1) no qualifier - "CitationID"; 2) view qualifier -
+     * "myView.CitationID"; and 3) schema+view qualifier - "mySchema.myView.CitationID".
+     * <p>
+     * This should always be used whenever a push down operation involves constructing a column, as we need to handle
+     * all 3 of the variations above. And op.schemaCol is required in order to achieve that.
+     *
+     * @param node
+     * @param columnName
+     */
+    public static void populateSchemaCol(ObjectNode node, String columnName) {
+        final String[] parts = removeTickMarksFromColumnName(columnName).split("\\.");
+
         ArrayNode colArgs = node.put("ns", "op").put("fn", "schema-col").putArray("args");
-        String[] parts = columnName.split("\\.");
         if (parts.length == 3) {
             colArgs.add(parts[0]).add(parts[1]).add(parts[2]);
         } else if (parts.length == 2) {
@@ -85,6 +96,22 @@ private static void populateSchemaCol(ObjectNode node, String columnName) {
         }
     }
 
+    /**
+     * For some push down operations, the tick marks that a user must use when a column name contains a period will
+     * still be present.
+     *
+     * @param columnName
+     * @return
+     */
+    private static String removeTickMarksFromColumnName(String columnName) {
+        if (columnName.startsWith("`")) {
+            columnName = columnName.substring(1);
+        }
+        return columnName.endsWith("`") ?
+            columnName.substring(0, columnName.length() - 1) :
+            columnName;
+    }
+
     static ObjectNode buildWhere(List<OpticFilter> opticFilters) {
         return newOperation("where", args -> {
             // If there's only one filter, can toss it into the "where" clause. Else, toss an "and" into the "where" and

src/main/java/com/marklogic/spark/reader/filter/IsNotNullFilter.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.marklogic.client.expression.PlanBuilder;
+import com.marklogic.spark.reader.PlanUtil;
 import org.apache.spark.sql.sources.IsNotNull;
 
 class IsNotNullFilter implements OpticFilter {
@@ -16,11 +17,11 @@ class IsNotNullFilter implements OpticFilter {
 
     @Override
     public void populateArg(ObjectNode arg) {
-        arg
+        ObjectNode colArg = arg
             .put("ns", "op").put("fn", "is-defined")
-            .putArray("args").addObject()
-            .put("ns", "op").put("fn", "col")
-            .putArray("args").add(filter.attribute());
+            .putArray("args").addObject();
+
+        PlanUtil.populateSchemaCol(colArg, filter.attribute());
     }
 
     @Override

src/main/java/com/marklogic/spark/reader/filter/IsNullFilter.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.marklogic.client.expression.PlanBuilder;
+import com.marklogic.spark.reader.PlanUtil;
 import org.apache.spark.sql.sources.IsNull;
 
 class IsNullFilter implements OpticFilter {
@@ -16,13 +17,13 @@ class IsNullFilter implements OpticFilter {
 
     @Override
     public void populateArg(ObjectNode arg) {
-        arg
+        ObjectNode colArg = arg
             .put("ns", "op").put("fn", "not")
             .putArray("args").addObject()
             .put("ns", "op").put("fn", "is-defined")
-            .putArray("args").addObject()
-            .put("ns", "op").put("fn", "col")
-            .putArray("args").add(filter.attribute());
+            .putArray("args").addObject();
+
+        PlanUtil.populateSchemaCol(colArg, filter.attribute());
     }
 
     @Override

src/main/java/com/marklogic/spark/reader/filter/SingleValueFilter.java

@@ -3,6 +3,7 @@
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.marklogic.client.expression.PlanBuilder;
+import com.marklogic.spark.reader.PlanUtil;
 
 import java.util.UUID;
 
@@ -29,14 +30,11 @@ class SingleValueFilter implements OpticFilter {
     public void populateArg(ObjectNode arg) {
         arg.put("ns", "op");
         arg.put("fn", this.functionName);
-        ArrayNode equalArgs = arg.putArray("args");
+        ArrayNode functionArgs = arg.putArray("args");
 
-        ObjectNode equalArg = equalArgs.addObject();
-        equalArg.put("ns", "op");
-        equalArg.put("fn", "col");
-        equalArg.putArray("args").add(this.columnName);
+        PlanUtil.populateSchemaCol(functionArgs.addObject(), this.columnName);
 
-        ObjectNode paramArg = equalArgs.addObject();
+        ObjectNode paramArg = functionArgs.addObject();
         paramArg.put("ns", "op");
         paramArg.put("fn", "param");
         paramArg.putArray("args").add(this.paramName);

src/test/java/com/marklogic/spark/reader/PushDownFilterTest.java

@@ -22,6 +22,27 @@ void equalToWithFilter() {
         assertEquals(4, countOfRowsReadFromMarkLogic);
     }
 
+    @Test
+    void equalToWithSchemaAndViewQualifier() {
+        assertEquals(4, newDefaultReader()
+            .load()
+            .filter("`Medical.Authors.CitationID` == 1")
+            .collectAsList()
+            .size(), "Verifying that a filter with a fully-qualified column name still works correctly.");
+        assertEquals(4, countOfRowsReadFromMarkLogic);
+    }
+
+    @Test
+    void equalToWithViewQualifier() {
+        assertEquals(4, newDefaultReader()
+            .option(Options.READ_OPTIC_DSL, "op.fromView('Medical', 'Authors', 'myView')")
+            .load()
+            .filter("`myView.CitationID` == 1")
+            .collectAsList()
+            .size(), "Verifying that a filter with a view-qualified column name still works correctly.");
+        assertEquals(4, countOfRowsReadFromMarkLogic);
+    }
+
     @Test
     void equalToWithWhere() {
         assertEquals(2, getCountOfRowsWithFilter("CitationID = 5"));
@@ -124,17 +145,40 @@ void inWithNoMatches() {
     @Test
     void isNotNull() {
         assertEquals(2, newDataset().filter(new Column("BooleanValue").isNotNull()).collectAsList().size());
+        assertEquals(2, countOfRowsReadFromMarkLogic);
+    }
+
+    @Test
+    void isNotNullQualified() {
+        assertEquals(2, newDefaultReader()
+            .load()
+            .filter(new Column("`Medical.Authors.BooleanValue`").isNotNull())
+            .collectAsList()
+            .size());
+
         assertEquals(2, countOfRowsReadFromMarkLogic,
             "2 of the authors are expected to have a BooleanValue column.");
     }
 
     @Test
     void isNull() {
-        assertEquals(13, newDataset().filter(new Column("BooleanValue").isNull()).collectAsList().size());
+        assertEquals(13, newDataset()
+            .filter(new Column("BooleanValue").isNull())
+            .collectAsList()
+            .size());
         assertEquals(13, countOfRowsReadFromMarkLogic,
             "13 of the authors are expected to have a null BooleanValue column.");
     }
 
+    @Test
+    void isNullQualified() {
+        assertEquals(13, newDefaultReader()
+            .load()
+            .filter(new Column("`Medical.Authors.BooleanValue`").isNull())
+            .collectAsList().size());
+        assertEquals(13, countOfRowsReadFromMarkLogic);
+    }
+
     @Test
     void stringContains() {
         List<Row> rows = newDataset().filter(new Column("LastName").contains("umbe")).collectAsList();