Tests for MetricStream and RollupStream and many other tests

solr/solrj/src/java/org/apache/solr/client/solrj/streaming/AscBucketComp.java

@@ -0,0 +1,41 @@
+package org.apache.solr.client.solrj.streaming;
+
+/*
+ * 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.Comparator;
+
+public class AscBucketComp implements Comparator<BucketMetrics> {
+
+  private int ord;
+
+  public AscBucketComp(int ord) {
+    this.ord = ord;
+  }
+
+  public int compare(BucketMetrics b1, BucketMetrics b2) {
+    double d1 = b1.getMetrics()[ord].getValue();
+    double d2 = b2.getMetrics()[ord].getValue();
+    if(d1 > d2) {
+      return 1;
+    } else if(d1 < d2) {
+      return -1;
+    } else {
+      return 0;
+    }
+  }
+}

solr/solrj/src/java/org/apache/solr/client/solrj/streaming/AscMetricComp.java

@@ -0,0 +1,39 @@
+package org.apache.solr.client.solrj.streaming;
+
+/*
+ * 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.io.Serializable;
+import java.util.Comparator;
+import java.util.List;
+
+public class AscMetricComp implements Comparator<Tuple>, Serializable {
+
+  private static final long serialVersionUID = 1;
+
+  private int ord;
+
+  public AscMetricComp(int ord) {
+    this.ord = ord;
+  }
+
+  public int compare(Tuple t1, Tuple t2) {
+    List<Double> values1 = (List<Double>)t1.get("metricValues");
+    List<Double> values2 = (List<Double>)t2.get("metricValues");
+    return values1.get(ord).compareTo(values2.get(ord));
+  }
+}

solr/solrj/src/java/org/apache/solr/client/solrj/streaming/Bucket.java

@@ -23,23 +23,17 @@ public class Bucket implements Serializable {
 
   private static final long serialVersionUID = 1;
 
-  private String bucketValue;
+  private String bucketKey;
 
   public Bucket() {
 
   }
 
-  public Bucket(String bucketValue) {
-    this.bucketValue = bucketValue;
+  public Bucket(String bucketKey) {
+    this.bucketKey = bucketKey;
   }
 
   public String getBucketValue(Tuple tuple) {
-    return bucketValue;
-
+    return tuple.get(bucketKey).toString();
   }
-
-
-
-
-
 }

solr/solrj/src/java/org/apache/solr/client/solrj/streaming/BucketMetrics.java

@@ -0,0 +1,43 @@
+package org.apache.solr.client.solrj.streaming;
+
+/*
+ * 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.io.Serializable;
+
+public class BucketMetrics implements Serializable {
+
+  private static final long serialVersionUID = 1;
+
+  private HashKey key;
+  private Metric[] metrics;
+
+  public BucketMetrics(HashKey key, Metric[] metrics) {
+    this.key = key;
+    this.metrics = metrics;
+  }
+
+  public Metric[] getMetrics() {
+    return metrics;
+  }
+
+  public HashKey getKey() {
+    return key;
+  }
+
+
+}

solr/solrj/src/java/org/apache/solr/client/solrj/streaming/CloudSolrStream.java

@@ -242,6 +242,10 @@ public void close() throws IOException {
   }
 
   public Tuple read() throws IOException {
+    return _read();
+  }
+
+  protected Tuple _read() throws IOException {
     TupleWrapper tw = tuples.pollFirst();
     if(tw != null) {
       Tuple t = tw.getTuple();

solr/solrj/src/java/org/apache/solr/client/solrj/streaming/CountMetric.java

@@ -36,6 +36,10 @@ public void update(Tuple tuple) {
    ++count;
   }
 
+  public double getValue() {
+    return count;
+  }
+
   public Metric newInstance() {
     return new CountMetric();
   }

solr/solrj/src/java/org/apache/solr/client/solrj/streaming/DescBucketComp.java

@@ -0,0 +1,41 @@
+package org.apache.solr.client.solrj.streaming;
+
+/*
+ * 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.Comparator;
+
+public class DescBucketComp implements Comparator<BucketMetrics> {
+
+  private int ord;
+
+  public DescBucketComp(int ord) {
+    this.ord = ord;
+  }
+
+  public int compare(BucketMetrics b1, BucketMetrics b2) {
+    double d1 = b1.getMetrics()[ord].getValue();
+    double d2 = b2.getMetrics()[ord].getValue();
+    if(d1 > d2) {
+      return -1;
+    } else if(d1 < d2) {
+      return 1;
+    } else {
+      return 0;
+    }
+  }
+}

solr/solrj/src/java/org/apache/solr/client/solrj/streaming/DescMetricComp.java

@@ -0,0 +1,39 @@
+package org.apache.solr.client.solrj.streaming;
+
+/*
+ * 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.io.Serializable;
+import java.util.Comparator;
+import java.util.List;
+
+public class DescMetricComp implements Comparator<Tuple>, Serializable {
+
+  private static final long serialVersionUID = 1;
+
+  private int ord;
+
+  public DescMetricComp(int ord) {
+    this.ord = ord;
+  }
+
+  public int compare(Tuple t1, Tuple t2) {
+    List<Double> values1 = (List<Double>)t1.get("metricValues");
+    List<Double> values2 = (List<Double>)t2.get("metricValues");
+    return values1.get(ord).compareTo(values2.get(ord))*-1;
+  }
+}

solr/solrj/src/java/org/apache/solr/client/solrj/streaming/HashKey.java

@@ -33,10 +33,15 @@ public HashKey(String value) {
   public HashKey(Tuple t, String[] keys) {
     this.parts = new Object[keys.length];
     for(int i=0; i<keys.length; i++) {
+      System.out.println("#################################################:"+keys[i]+":"+t.get(keys[i]));
       parts[i] = t.get(keys[i]);
     }
   }
 
+  public HashKey(String[] parts) {
+    this.parts = parts;
+  }
+
   public Object getParts() {
     return parts;
   }

solr/solrj/src/java/org/apache/solr/client/solrj/streaming/MaxMetric.java

@@ -38,6 +38,14 @@ public String getName() {
     return "mix:"+column;
   }
 
+  public double getValue() {
+    if(isDouble) {
+      return doubleMax;
+    } else {
+      return longMax;
+    }
+  }
+
   public void update(Tuple tuple) {
     if(isDouble) {
       double d = (double)tuple.get(column);

solr/solrj/src/java/org/apache/solr/client/solrj/streaming/MeanMetric.java

@@ -59,6 +59,18 @@ public Metric newInstance() {
     return new MeanMetric(column, isDouble);
   }
 
+  public double getValue() {
+    double dcount = (double)count;
+    if(isDouble) {
+      double ave = doubleSum/dcount;
+      return ave;
+
+    } else {
+      double ave = longSum/dcount;
+      return ave;
+    }
+  }
+
   public Map<String, Double> metricValues() {
     Map m = new HashMap();
     double dcount = (double)count;

solr/solrj/src/java/org/apache/solr/client/solrj/streaming/Metric.java

@@ -22,6 +22,7 @@
 
 public interface Metric extends Serializable {
   public String getName();
+  public double getValue();
   public void update(Tuple tuple);
   public Metric newInstance();
   public Map<String, Double> metricValues();