Add stringLast and stringFirst aggregators extension

docs/content/development/extensions-core/first-last-string.md

@@ -0,0 +1,41 @@
+---
+layout: doc_page
+---
+
+# First/Last String Module
+
+To use these aggregators, make sure you [include](../../operations/including-extensions.html) the extension in your config file:
+
+```
+druid.extensions.loadList=["druid-first-last-string"]
+```
+
+## First String aggregator
+
+`firstString` computes the metric value with the minimum timestamp or `null` if no row exist
+
+```json
+{
+  "type" : "firstString",
+  "name" : <output_name>,
+  "fieldName" : <metric_name>,
+  "maxStringBytes" : <integer>
+}
+```
+
+## Last String aggregator
+
+```json
+{
+  "type" : "lastString",
+  "name" : <output_name>,
+  "fieldName" : <metric_name>,
+  "maxStringBytes" : <integer>
+}
+```
+
+`lastString` computes the metric value with the maximum timestamp or `null` if no row exist
+
+
+
+Note: The default value of `maxStringBytes` is 1024.

docs/content/development/extensions.md

@@ -72,6 +72,7 @@ All of these community extensions can be downloaded using *pull-deps* with the c
 |kafka-emitter|Kafka metrics emitter|[link](../development/extensions-contrib/kafka-emitter.html)|
 |druid-thrift-extensions|Support thrift ingestion |[link](../development/extensions-contrib/thrift.html)|
 |druid-opentsdb-emitter|OpenTSDB metrics emitter |[link](../development/extensions-contrib/opentsdb-emitter.html)|
+|druid-first-last-string|First/Last String Aggregators |[link](../development/extensions-contrib/first-last-string.html)|
 
 ## Promoting Community Extension to Core Extension
 

extensions-contrib/first-last-string/pom.xml

@@ -0,0 +1,81 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Druid - a distributed column store.
+  ~ Copyright 2012 - 2015 Metamarkets Group Inc.
+  ~
+  ~ Licensed 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.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>io.druid.extensions.contrib</groupId>
+    <artifactId>druid-first-last-string</artifactId>
+    <name>druid-first-last-string</name>
+    <description>druid-first-last-string</description>
+
+    <parent>
+        <groupId>io.druid</groupId>
+        <artifactId>druid</artifactId>
+        <version>0.13.0-SNAPSHOT</version>
+        <relativePath>../../pom.xml</relativePath>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.druid</groupId>
+            <artifactId>druid-processing</artifactId>
+            <version>${project.parent.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.druid</groupId>
+            <artifactId>druid-sql</artifactId>
+            <version>${project.parent.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <!-- Tests -->
+        <dependency>
+            <groupId>io.druid</groupId>
+            <artifactId>druid-processing</artifactId>
+            <version>${project.parent.version}</version>
+            <scope>test</scope>
+            <type>test-jar</type>
+        </dependency>
+        <dependency>
+            <groupId>io.druid</groupId>
+            <artifactId>druid-sql</artifactId>
+            <version>${project.parent.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.druid</groupId>
+            <artifactId>druid-server</artifactId>
+            <version>${project.parent.version}</version>
+            <scope>test</scope>
+            <type>test-jar</type>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.easymock</groupId>
+            <artifactId>easymock</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>

extensions-contrib/first-last-string/src/main/java/io/druid/query/aggregation/FirstLastStringDruidModule.java

@@ -0,0 +1,59 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets 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.
+ */
+
+package io.druid.query.aggregation;
+
+import com.fasterxml.jackson.databind.Module;
+import com.fasterxml.jackson.databind.jsontype.NamedType;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.google.common.collect.ImmutableList;
+import com.google.inject.Binder;
+import io.druid.initialization.DruidModule;
+import io.druid.query.aggregation.first.StringFirstAggregatorFactory;
+import io.druid.query.aggregation.last.StringLastAggregatorFactory;
+import io.druid.segment.serde.ComplexMetrics;
+
+import java.util.List;
+
+/**
+ */
+public class FirstLastStringDruidModule implements DruidModule
+{
+  public static final String STRING_LAST = "stringLast";
+  public static final String STRING_FIRST = "stringFirst";
+
+  @Override
+  public List<? extends Module> getJacksonModules()
+  {
+    return ImmutableList.of(
+        new SimpleModule("FirstLastStringModule").registerSubtypes(
+            new NamedType(StringLastAggregatorFactory.class, STRING_LAST),
+            new NamedType(StringFirstAggregatorFactory.class, STRING_FIRST)
+        )
+    );
+  }
+
+  @Override
+  public void configure(Binder binder)
+  {
+    if (ComplexMetrics.getSerdeForType("serializablePairLongString") == null) {
+      ComplexMetrics.registerSerde("serializablePairLongString", new SerializablePairSerde());
+    }
+  }
+}

extensions-contrib/first-last-string/src/main/java/io/druid/query/aggregation/SerializablePairSerde.java

@@ -0,0 +1,129 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets 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.
+ */
+
+package io.druid.query.aggregation;
+
+import io.druid.collections.SerializablePair;
+import io.druid.data.input.InputRow;
+import io.druid.segment.GenericColumnSerializer;
+import io.druid.segment.column.ColumnBuilder;
+import io.druid.segment.data.GenericIndexed;
+import io.druid.segment.data.ObjectStrategy;
+import io.druid.segment.serde.ComplexColumnPartSupplier;
+import io.druid.segment.serde.ComplexMetricExtractor;
+import io.druid.segment.serde.ComplexMetricSerde;
+import io.druid.segment.serde.LargeColumnSupportedComplexColumnSerializer;
+import io.druid.segment.writeout.SegmentWriteOutMedium;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+
+public class SerializablePairSerde extends ComplexMetricSerde
+{
+
+  public SerializablePairSerde()
+  {
+
+  }
+
+  @Override
+  public String getTypeName()
+  {
+    return "serializablePairLongString";
+  }
+
+  @Override
+  public ComplexMetricExtractor getExtractor()
+  {
+    return new ComplexMetricExtractor()
+    {
+      @Override
+      public Class<SerializablePair> extractedClass()
+      {
+        return SerializablePair.class;
+      }
+
+      @Override
+      public Object extractValue(InputRow inputRow, String metricName)
+      {
+        return inputRow.getRaw(metricName);
+      }
+    };
+  }
+
+  @Override
+  public void deserializeColumn(ByteBuffer buffer, ColumnBuilder columnBuilder)
+  {
+    final GenericIndexed column = GenericIndexed.read(buffer, getObjectStrategy(), columnBuilder.getFileMapper());
+    columnBuilder.setComplexColumn(new ComplexColumnPartSupplier(getTypeName(), column));
+  }
+
+  @Override
+  public ObjectStrategy getObjectStrategy()
+  {
+    return new ObjectStrategy<SerializablePair>()
+    {
+      @Override
+      public int compare(SerializablePair o1, SerializablePair o2)
+      {
+        return o1.lhs.equals(o2.lhs) && o1.rhs.equals(o2.rhs) ? 1 : 0;
+      }
+
+      @Override
+      public Class<? extends SerializablePair> getClazz()
+      {
+        return SerializablePair.class;
+      }
+
+      @Override
+      public SerializablePair<Long, String> fromByteBuffer(ByteBuffer buffer, int numBytes)
+      {
+        final ByteBuffer readOnlyBuffer = buffer.asReadOnlyBuffer();
+
+        Long lhs = readOnlyBuffer.getLong();
+        Integer stringSize = readOnlyBuffer.getInt();
+
+        byte[] stringBytes = new byte[stringSize];
+        readOnlyBuffer.get(stringBytes, 0, stringSize);
+
+        return new SerializablePair<>(lhs, new String(stringBytes, StandardCharsets.UTF_8));
+      }
+
+      @Override
+      public byte[] toBytes(SerializablePair val)
+      {
+        String rhsString = (String) val.rhs;
+
+        ByteBuffer bbuf = ByteBuffer.allocate(Long.BYTES + Integer.BYTES + rhsString.length());
+        bbuf.putLong((Long) val.lhs);
+        bbuf.putInt(Long.BYTES, rhsString.length());
+        bbuf.position(Long.BYTES + Integer.BYTES);
+        bbuf.put(rhsString.getBytes(StandardCharsets.UTF_8));
+
+        return bbuf.array();
+      }
+    };
+  }
+
+  @Override
+  public GenericColumnSerializer getSerializer(SegmentWriteOutMedium segmentWriteOutMedium, String column)
+  {
+    return LargeColumnSupportedComplexColumnSerializer.create(segmentWriteOutMedium, column, this.getObjectStrategy());
+  }
+}

extensions-contrib/first-last-string/src/main/java/io/druid/query/aggregation/first/StringFirstAggregateCombiner.java

@@ -0,0 +1,55 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets 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.
+ */
+
+package io.druid.query.aggregation.first;
+
+import io.druid.query.aggregation.ObjectAggregateCombiner;
+import io.druid.segment.ColumnValueSelector;
+
+import javax.annotation.Nullable;
+
+public class StringFirstAggregateCombiner extends ObjectAggregateCombiner<String>
+{
+  String lastString;
+
+  @Override
+  public void reset(ColumnValueSelector selector)
+  {
+    lastString = (String) selector.getObject();
+  }
+
+  @Override
+  public void fold(ColumnValueSelector selector)
+  {
+    lastString = (String) selector.getObject();
+  }
+
+  @Nullable
+  @Override
+  public String getObject()
+  {
+    return lastString;
+  }
+
+  @Override
+  public Class<String> classOfObject()
+  {
+    return String.class;
+  }
+}