SOLR-8556: Add ConcatOperation to be used with the SelectStream

git-svn-id: https://svn.apache.org/repos/asf/lucene/dev/trunk@1725769 13f79535-47bb-0310-9956-ffa450edef68

solr/CHANGES.txt

@@ -138,6 +138,8 @@ New Features
 * SOLR-8415: Provide command to switch between non/secure mode in ZK
   (Mike Drob, Gregory Chanan)
 
+* SOLR-8556: Add ConcatOperation to be used with the SelectStream (Joel Bernstein, Dennis Gove)
+
 Bug Fixes
 ----------------------
 * SOLR-8386: Add field option in the new admin UI schema page loads up even when no schemaFactory has been

solr/core/src/java/org/apache/solr/handler/StreamHandler.java

@@ -27,6 +27,7 @@ import java.util.Map.Entry;
 import org.apache.solr.client.solrj.io.SolrClientCache;
 import org.apache.solr.client.solrj.io.Tuple;
 import org.apache.solr.client.solrj.io.comp.StreamComparator;
+import org.apache.solr.client.solrj.io.ops.ConcatOperation;
 import org.apache.solr.client.solrj.io.ops.DistinctOperation;
 import org.apache.solr.client.solrj.io.ops.GroupOperation;
 import org.apache.solr.client.solrj.io.ops.ReplaceOperation;
@@ -132,6 +133,7 @@ public class StreamHandler extends RequestHandlerBase implements SolrCoreAware {
       
       // tuple manipulation operations
       .withFunctionName("replace", ReplaceOperation.class)
+      .withFunctionName("concat", ConcatOperation.class)
       
       // stream reduction operations
       .withFunctionName("group", GroupOperation.class)

solr/solrj/src/java/org/apache/solr/client/solrj/io/ops/ConcatOperation.java

@@ -0,0 +1,98 @@
+package org.apache.solr.client.solrj.io.ops;
+
+import java.io.IOException;
+import java.util.Locale;
+
+import org.apache.solr.client.solrj.io.Tuple;
+import org.apache.solr.client.solrj.io.stream.expr.StreamExpression;
+import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionNamedParameter;
+import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionParameter;
+import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
+import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Concatenates fields and adds them to the tuple. Example
+ * concat(fields="month,day,year", delim="-", as="id")
+ */
+public class ConcatOperation implements StreamOperation {
+
+  private static final long serialVersionUID = 1;
+
+  private String[] fields;
+  private String as;
+  private String delim;
+
+  public ConcatOperation(String[] fields, String as, String delim) {
+    this.fields = fields;
+    this.as = as;
+    this.delim = delim;
+  }
+
+  public ConcatOperation(StreamExpression expression, StreamFactory factory) throws IOException {
+
+    if(3 == expression.getParameters().size()){
+      StreamExpressionNamedParameter fieldsParam = factory.getNamedOperand(expression, "fields");
+      String fieldsStr = ((StreamExpressionValue)fieldsParam.getParameter()).getValue();
+      this.fields = fieldsStr.split(",");
+      for(int i=0; i<fields.length; i++) {
+        fields[i] = fields[i].trim();
+      }
+
+      StreamExpressionNamedParameter asParam = factory.getNamedOperand(expression, "as");
+      this.as = ((StreamExpressionValue)asParam.getParameter()).getValue();
+
+      StreamExpressionNamedParameter delim = factory.getNamedOperand(expression, "delim");
+      this.delim = ((StreamExpressionValue)delim.getParameter()).getValue();
+    } else{
+      throw new IOException(String.format(Locale.ROOT,"Invalid expression %s - unknown operands found", expression));
+    }
+  }
+
+  @Override
+  public void operate(Tuple tuple) {
+    StringBuilder buf = new StringBuilder();
+    for(String field : fields) {
+      if(buf.length() > 0) {
+        buf.append(delim);
+      }
+      Object value = tuple.get(field);
+      if(null == value){ value = "null"; }
+      buf.append(value);
+    }
+
+    tuple.put(as, buf.toString());
+  }
+
+  @Override
+  public StreamExpressionParameter toExpression(StreamFactory factory) throws IOException {
+    StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
+    
+    StringBuilder sb = new StringBuilder();
+    for(String field : fields){
+      if(sb.length() > 0){ sb.append(","); }
+      sb.append(field);
+    }
+    expression.addParameter(new StreamExpressionNamedParameter("fields",sb.toString()));
+    expression.addParameter(new StreamExpressionNamedParameter("delim",delim));
+    expression.addParameter(new StreamExpressionNamedParameter("as",as));
+    return expression;
+  }
+
+}

solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamExpressionTest.java

@@ -28,6 +28,7 @@ import org.apache.lucene.util.LuceneTestCase;
 import org.apache.lucene.util.LuceneTestCase.Slow;
 import org.apache.solr.client.solrj.impl.CloudSolrClient;
 import org.apache.solr.client.solrj.io.Tuple;
+import org.apache.solr.client.solrj.io.ops.ConcatOperation;
 import org.apache.solr.client.solrj.io.ops.GroupOperation;
 import org.apache.solr.client.solrj.io.comp.ComparatorOrder;
 import org.apache.solr.client.solrj.io.comp.FieldComparator;
@@ -1334,7 +1335,9 @@ public class StreamExpressionTest extends AbstractFullDistribZkTestBase {
       .withFunctionName("search", CloudSolrStream.class)
       .withFunctionName("innerJoin", InnerJoinStream.class)
       .withFunctionName("select", SelectStream.class)
-      .withFunctionName("replace", ReplaceOperation.class);
+      .withFunctionName("replace", ReplaceOperation.class)
+      .withFunctionName("concat", ConcatOperation.class)
+      ;
     
     // Basic test
     clause = "select("
@@ -1361,7 +1364,29 @@ public class StreamExpressionTest extends AbstractFullDistribZkTestBase {
     assertLong(tuples.get(2), "join1", 12);
     assertLong(tuples.get(7), "join1", 12);
     assertString(tuples.get(6), "join1", "d");
+    
 
+    // Basic with replacements and concat test
+    clause = "select("
+            +   "id, join1_i as join1, join2_s as join2, ident_s as identity,"
+            +   "replace(join1, 0, withValue=12), replace(join1, 3, withValue=12), replace(join1, 2, withField=join2),"
+            +   "concat(fields=\"identity,join1\", as=\"newIdentity\",delim=\"-\"),"
+            +   "search(collection1, q=\"side_s:left\", fl=\"id,join1_i,join2_s,ident_s\", sort=\"join1_i asc, join2_s asc, id asc\")"
+            + ")";
+    stream = factory.constructStream(clause);
+    tuples = getTuples(stream);
+    assertFields(tuples, "id", "join1", "join2", "identity", "newIdentity");
+    assertNotFields(tuples, "join1_i", "join2_s", "ident_s");
+    assertLong(tuples.get(0), "join1", 12);
+    assertString(tuples.get(0), "newIdentity", "left_1-12");
+    assertLong(tuples.get(1), "join1", 12); 
+    assertString(tuples.get(1), "newIdentity", "left_1-12");
+    assertLong(tuples.get(2), "join1", 12); 
+    assertString(tuples.get(2), "newIdentity", "left_2-12");
+    assertLong(tuples.get(7), "join1", 12); 
+    assertString(tuples.get(7), "newIdentity", "left_7-12");
+    assertString(tuples.get(6), "join1", "d");
+    assertString(tuples.get(6), "newIdentity", "left_6-d");
     
     // Inner stream test
     clause = "innerJoin("

solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/ops/ConcatOperationTest.java

@@ -0,0 +1,291 @@
+package org.apache.solr.client.solrj.io.stream.ops;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import java.util.Map;
+
+import junit.framework.Assert;
+
+import org.apache.commons.collections.map.HashedMap;
+import org.apache.lucene.util.LuceneTestCase;
+import org.apache.solr.client.solrj.io.Tuple;
+import org.apache.solr.client.solrj.io.ops.ConcatOperation;
+import org.apache.solr.client.solrj.io.ops.ReplaceOperation;
+import org.apache.solr.client.solrj.io.ops.StreamOperation;
+import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionParser;
+import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
+import org.junit.Test;
+
+/**
+ **/
+
+public class ConcatOperationTest extends LuceneTestCase {
+
+  StreamFactory factory;
+  Map<String, Object> values;
+  
+  public ConcatOperationTest() {
+    super();
+    
+    factory = new StreamFactory()
+      .withFunctionName("concat", ConcatOperation.class);
+    values = new HashedMap();
+  }
+    
+  @Test
+  public void concatSingleField() throws Exception{
+    Tuple tuple;
+    StreamOperation operation;
+        
+    operation = new ConcatOperation(new String[]{"fieldA"}, "fieldAConcat", "-");
+    
+    values.clear();
+    values.put("fieldA", "bar");
+    tuple = new Tuple(values);
+    operation.operate(tuple);
+    
+    Assert.assertNotNull(tuple.get("fieldA"));
+    Assert.assertEquals("bar", tuple.get("fieldA"));
+    
+    Assert.assertNotNull(tuple.get("fieldAConcat"));
+    Assert.assertEquals("bar", tuple.get("fieldAConcat"));
+  }
+
+  @Test
+  public void concatMultipleFields() throws Exception{
+    Tuple tuple;
+    StreamOperation operation;
+        
+    operation = new ConcatOperation(new String[]{"fieldA","fieldB"}, "fieldABConcat", "-");
+    values.clear();
+    values.put("fieldA", "bar");
+    values.put("fieldB", "baz");
+    tuple = new Tuple(values);
+    operation.operate(tuple);
+    
+    Assert.assertNotNull(tuple.get("fieldA"));
+    Assert.assertEquals("bar", tuple.get("fieldA"));
+    
+    Assert.assertNotNull(tuple.get("fieldB"));
+    Assert.assertEquals("baz", tuple.get("fieldB"));
+    
+    Assert.assertNotNull(tuple.get("fieldABConcat"));
+    Assert.assertEquals("bar-baz", tuple.get("fieldABConcat"));
+    
+    // do the same in oposite order
+    operation = new ConcatOperation(new String[]{"fieldB","fieldA"}, "fieldABConcat", "-");
+    tuple = new Tuple(values);
+    operation.operate(tuple);
+    
+    Assert.assertNotNull(tuple.get("fieldA"));
+    Assert.assertEquals("bar", tuple.get("fieldA"));
+    
+    Assert.assertNotNull(tuple.get("fieldB"));
+    Assert.assertEquals("baz", tuple.get("fieldB"));
+    
+    Assert.assertNotNull(tuple.get("fieldABConcat"));
+    Assert.assertEquals("baz-bar", tuple.get("fieldABConcat"));
+  }
+  
+  @Test
+  public void concatMultipleFieldsWithIgnoredFields() throws Exception{
+    Tuple tuple;
+    StreamOperation operation;
+        
+    operation = new ConcatOperation(new String[]{"fieldA","fieldB"}, "fieldABConcat", "-");
+    values.clear();
+    values.put("fieldA", "bar");
+    values.put("fieldB", "baz");
+    values.put("fieldC", "bab");
+    values.put("fieldD", "bat");
+    tuple = new Tuple(values);
+    operation.operate(tuple);
+    
+    Assert.assertNotNull(tuple.get("fieldA"));
+    Assert.assertEquals("bar", tuple.get("fieldA"));
+    
+    Assert.assertNotNull(tuple.get("fieldB"));
+    Assert.assertEquals("baz", tuple.get("fieldB"));
+    
+    Assert.assertNotNull(tuple.get("fieldC"));
+    Assert.assertEquals("bab", tuple.get("fieldC"));
+    
+    Assert.assertNotNull(tuple.get("fieldD"));
+    Assert.assertEquals("bat", tuple.get("fieldD"));
+    
+    Assert.assertNotNull(tuple.get("fieldABConcat"));
+    Assert.assertEquals("bar-baz", tuple.get("fieldABConcat"));
+    
+    // do the same in oposite order
+    operation = new ConcatOperation(new String[]{"fieldB","fieldA"}, "fieldABConcat", "-");
+    tuple = new Tuple(values);
+    operation.operate(tuple);
+    
+    Assert.assertNotNull(tuple.get("fieldA"));
+    Assert.assertEquals("bar", tuple.get("fieldA"));
+    
+    Assert.assertNotNull(tuple.get("fieldB"));
+    Assert.assertEquals("baz", tuple.get("fieldB"));
+    
+    Assert.assertNotNull(tuple.get("fieldABConcat"));
+    Assert.assertEquals("baz-bar", tuple.get("fieldABConcat"));
+  }
+
+  @Test
+  public void concatWithNullValues() throws Exception{
+    Tuple tuple;
+    StreamOperation operation;
+        
+    operation = new ConcatOperation(new String[]{"fieldA","fieldB"}, "fieldABConcat", "-");
+    values.clear();
+    values.put("fieldA", "bar");
+    tuple = new Tuple(values);
+    operation.operate(tuple);
+    
+    Assert.assertNotNull(tuple.get("fieldA"));
+    Assert.assertEquals("bar", tuple.get("fieldA"));
+    
+    Assert.assertNull(tuple.get("fieldB"));
+    
+    Assert.assertNotNull(tuple.get("fieldABConcat"));
+    Assert.assertEquals("bar-null", tuple.get("fieldABConcat"));
+    
+  }
+  
+///////////////////////////
+  @Test
+  public void concatSingleFieldExpression() throws Exception{
+    Tuple tuple;
+    StreamOperation operation;
+        
+    operation = new ConcatOperation(StreamExpressionParser.parse("concat(fields=\"fieldA\", as=\"fieldAConcat\", delim=\"-\")"), factory);
+    
+    values.clear();
+    values.put("fieldA", "bar");
+    tuple = new Tuple(values);
+    operation.operate(tuple);
+    
+    Assert.assertNotNull(tuple.get("fieldA"));
+    Assert.assertEquals("bar", tuple.get("fieldA"));
+    
+    Assert.assertNotNull(tuple.get("fieldAConcat"));
+    Assert.assertEquals("bar", tuple.get("fieldAConcat"));
+  }
+
+  @Test
+  public void concatMultipleFieldsExpression() throws Exception{
+    Tuple tuple;
+    StreamOperation operation;
+        
+    operation = new ConcatOperation(StreamExpressionParser.parse("concat(fields=\"fieldA,fieldB\", as=\"fieldABConcat\", delim=\"-\")"), factory);
+    values.clear();
+    values.put("fieldA", "bar");
+    values.put("fieldB", "baz");
+    tuple = new Tuple(values);
+    operation.operate(tuple);
+    
+    Assert.assertNotNull(tuple.get("fieldA"));
+    Assert.assertEquals("bar", tuple.get("fieldA"));
+    
+    Assert.assertNotNull(tuple.get("fieldB"));
+    Assert.assertEquals("baz", tuple.get("fieldB"));
+    
+    Assert.assertNotNull(tuple.get("fieldABConcat"));
+    Assert.assertEquals("bar-baz", tuple.get("fieldABConcat"));
+    
+    // do the same in oposite order
+    operation = new ConcatOperation(StreamExpressionParser.parse("concat(fields=\"fieldB,fieldA\", as=\"fieldABConcat\", delim=\"-\")"), factory);
+    tuple = new Tuple(values);
+    operation.operate(tuple);
+    
+    Assert.assertNotNull(tuple.get("fieldA"));
+    Assert.assertEquals("bar", tuple.get("fieldA"));
+    
+    Assert.assertNotNull(tuple.get("fieldB"));
+    Assert.assertEquals("baz", tuple.get("fieldB"));
+    
+    Assert.assertNotNull(tuple.get("fieldABConcat"));
+    Assert.assertEquals("baz-bar", tuple.get("fieldABConcat"));
+  }
+  
+  @Test
+  public void concatMultipleFieldsWithIgnoredFieldsExpression() throws Exception{
+    Tuple tuple;
+    StreamOperation operation;
+        
+    operation = new ConcatOperation(StreamExpressionParser.parse("concat(fields=\"fieldA,fieldB\", as=\"fieldABConcat\", delim=\"-\")"), factory);
+    values.clear();
+    values.put("fieldA", "bar");
+    values.put("fieldB", "baz");
+    values.put("fieldC", "bab");
+    values.put("fieldD", "bat");
+    tuple = new Tuple(values);
+    operation.operate(tuple);
+    
+    Assert.assertNotNull(tuple.get("fieldA"));
+    Assert.assertEquals("bar", tuple.get("fieldA"));
+    
+    Assert.assertNotNull(tuple.get("fieldB"));
+    Assert.assertEquals("baz", tuple.get("fieldB"));
+    
+    Assert.assertNotNull(tuple.get("fieldC"));
+    Assert.assertEquals("bab", tuple.get("fieldC"));
+    
+    Assert.assertNotNull(tuple.get("fieldD"));
+    Assert.assertEquals("bat", tuple.get("fieldD"));
+    
+    Assert.assertNotNull(tuple.get("fieldABConcat"));
+    Assert.assertEquals("bar-baz", tuple.get("fieldABConcat"));
+    
+    // do the same in oposite order
+    operation = new ConcatOperation(StreamExpressionParser.parse("concat(fields=\"fieldB,fieldA\", as=\"fieldABConcat\", delim=\"-\")"), factory);
+    tuple = new Tuple(values);
+    operation.operate(tuple);
+    
+    Assert.assertNotNull(tuple.get("fieldA"));
+    Assert.assertEquals("bar", tuple.get("fieldA"));
+    
+    Assert.assertNotNull(tuple.get("fieldB"));
+    Assert.assertEquals("baz", tuple.get("fieldB"));
+    
+    Assert.assertNotNull(tuple.get("fieldABConcat"));
+    Assert.assertEquals("baz-bar", tuple.get("fieldABConcat"));
+  }
+
+  @Test
+  public void concatWithNullValuesExpression() throws Exception{
+    Tuple tuple;
+    StreamOperation operation;
+        
+    operation = new ConcatOperation(StreamExpressionParser.parse("concat(fields=\"fieldA,fieldB\", as=\"fieldABConcat\", delim=\"-\")"), factory);
+    values.clear();
+    values.put("fieldA", "bar");
+    tuple = new Tuple(values);
+    operation.operate(tuple);
+    
+    Assert.assertNotNull(tuple.get("fieldA"));
+    Assert.assertEquals("bar", tuple.get("fieldA"));
+    
+    Assert.assertNull(tuple.get("fieldB"));
+    
+    Assert.assertNotNull(tuple.get("fieldABConcat"));
+    Assert.assertEquals("bar-null", tuple.get("fieldABConcat"));
+    
+  }
+  
+}