From 6da80e8d2cd0425fdebbe81f5bdf370fcd949991 Mon Sep 17 00:00:00 2001
From: Alberto Rodriguez <arodriguez@stratio.com>
Date: Wed, 29 Apr 2015 12:38:00 +0200
Subject: [PATCH] Add LIKE operator support. Fixes #METAMODEL-136

---
 .../metamodel/mongodb/MongoDbDataContext.java | 19 ++++++-
 .../mongodb/MongoDbDataContextTest.java       | 56 +++++++++++++++++++
 2 files changed, 74 insertions(+), 1 deletion(-)

diff --git a/mongodb/src/main/java/org/apache/metamodel/mongodb/MongoDbDataContext.java b/mongodb/src/main/java/org/apache/metamodel/mongodb/MongoDbDataContext.java
index 008dde92b..5e89b6d9b 100644
--- a/mongodb/src/main/java/org/apache/metamodel/mongodb/MongoDbDataContext.java
+++ b/mongodb/src/main/java/org/apache/metamodel/mongodb/MongoDbDataContext.java
@@ -25,6 +25,7 @@
 import java.util.Set;
 import java.util.SortedMap;
 import java.util.TreeMap;
+import java.util.regex.Pattern;
 
 import org.apache.metamodel.DataContext;
 import org.apache.metamodel.MetaModelException;
@@ -391,7 +392,12 @@ private void convertToCursorObject(BasicDBObject query, FilterItem item) {
             final BasicDBObject existingFilterObject = (BasicDBObject) query.get(columnName);
             if (existingFilterObject == null) {
                 if (operatorName == null) {
-                    query.put(columnName, operand);
+                    if (item.getOperator().equals(OperatorType.LIKE)) {
+                        query.put(columnName, turnOperandIntoRegExp(operand));
+                    }
+                    else {
+                        query.put(columnName, operand);
+                    }
                 } else {
                     query.put(columnName, new BasicDBObject(operatorName, operand));
                 }
@@ -409,6 +415,7 @@ private String getOperatorName(FilterItem item) {
         final String operatorName;
         switch (item.getOperator()) {
         case EQUALS_TO:
+        case LIKE:
             operatorName = null;
             break;
         case LESS_THAN:
@@ -435,6 +442,16 @@ private String getOperatorName(FilterItem item) {
         return operatorName;
     }
 
+    private Pattern turnOperandIntoRegExp(Object operand) {
+        StringBuilder operandAsRegExp = new StringBuilder(replaceWildCardLikeChars(operand.toString()));
+        operandAsRegExp.insert(0, "^").append("$");
+        return Pattern.compile(operandAsRegExp.toString(), Pattern.CASE_INSENSITIVE);
+    }
+
+    private String replaceWildCardLikeChars(String operand) {
+        return operand.replaceAll("%", ".*");
+    }
+
     @Override
     protected DataSet materializeMainSchemaTable(Table table, Column[] columns, int maxRows) {
         return materializeMainSchemaTableInternal(table, columns, null, 1, maxRows, true);
diff --git a/mongodb/src/test/java/org/apache/metamodel/mongodb/MongoDbDataContextTest.java b/mongodb/src/test/java/org/apache/metamodel/mongodb/MongoDbDataContextTest.java
index b2563d2cf..1a4a3af8d 100644
--- a/mongodb/src/test/java/org/apache/metamodel/mongodb/MongoDbDataContextTest.java
+++ b/mongodb/src/test/java/org/apache/metamodel/mongodb/MongoDbDataContextTest.java
@@ -490,4 +490,60 @@ public void run(UpdateCallback callback) {
         dc.refreshSchemas();
         assertEquals(0, defaultSchema.getTableCount());
     }
+
+    public void testSelectWithLikeOperator() throws Exception {
+        if (!isConfigured()) {
+            System.err.println(getInvalidConfigurationMessage());
+            return;
+        }
+
+        DBCollection col = db.createCollection(getCollectionName(), null);
+
+        // delete if already exists
+        {
+            col.drop();
+            col = db.createCollection(getCollectionName(), null);
+        }
+
+        final BasicDBObject dbRow = new BasicDBObject();
+        dbRow.append("name", new BasicDBObject().append("first", "John").append("last", "Doe"));
+        dbRow.append("gender", "MALE");
+        col.insert(dbRow);
+
+        final BasicDBObject dbRow2 = new BasicDBObject();
+        dbRow2.append("name", new BasicDBObject().append("first", "Mary").append("last", "Johnson"));
+        dbRow2.append("gender", "FEMALE");
+        col.insert(dbRow2);
+
+        final BasicDBObject dbRow3 = new BasicDBObject();
+        dbRow3.append("name", new BasicDBObject().append("first", "X").append("last", "Unknown"));
+        dbRow3.append("gender", "UNKNOWN");
+        col.insert(dbRow3);
+
+        final MongoDbDataContext dc = new MongoDbDataContext(db, new SimpleTableDef(getCollectionName(), new String[] {
+                "name.first", "name.last", "gender", "addresses", "addresses[0].city", "addresses[0].country",
+                "addresses[5].foobar" }));
+
+        final DataSet ds1 = dc.executeQuery("select * from my_collection where gender LIKE '%MALE%'");
+        final DataSet ds2 = dc.executeQuery("select * from my_collection where gender LIKE 'MALE%'");
+        final DataSet ds3 = dc.executeQuery("select * from my_collection where gender LIKE '%NK%OW%'");
+        final DataSet ds4 = dc.executeQuery("select * from my_collection where gender LIKE '%MALE'");
+        try {
+            assertTrue(ds1.next());
+            assertTrue(ds1.next());
+            assertFalse(ds1.next());
+            assertTrue(ds2.next());
+            assertFalse(ds2.next());
+            assertTrue(ds3.next());
+            assertFalse(ds3.next());
+            assertTrue(ds4.next());
+            assertTrue(ds4.next());
+            assertFalse(ds4.next());
+        } finally {
+            ds1.close();
+            ds2.close();
+            ds3.close();
+            ds4.close();
+        }
+    }
 }