UDF / UDA execution time in trace

patch by Christopher Batey; reviewed by Robert Stupp for CASSANDRA-9723

CHANGES.txt

@@ -1,3 +1,6 @@
+2.2.1
+ * UDF / UDA execution time in trace (CASSANDRA-9723)
+
 2.2.0
  * Fix cqlsh copy methods and other windows specific issues (CASSANDRA-9795) 
  * Don't wrap byte arrays in SequentialWriter (CASSANDRA-9797)

doc/cql3/CQL.textile

@@ -1938,7 +1938,7 @@ User-defined aggregates can be used in "@SELECT@":#selectStmt statement.
 A complete working example for user-defined aggregates (assuming that a keyspace has been selected using the "@USE@":#useStmt statement):
 
 bc(sample).. 
-CREATE FUNCTION averageState ( state tuple<int,bigint>, val int )
+CREATE OR REPLACE FUNCTION averageState ( state tuple<int,bigint>, val int )
   CALLED ON NULL INPUT
   RETURNS tuple<int,bigint>
   LANGUAGE java
@@ -1950,7 +1950,7 @@ CREATE FUNCTION averageState ( state tuple<int,bigint>, val int )
     return state;
   ';
 
-CREATE FUNCTION averageFinal ( state tuple<int,bigint> )
+CREATE OR REPLACE FUNCTION averageFinal ( state tuple<int,bigint> )
   CALLED ON NULL INPUT
   RETURNS double
   LANGUAGE java
@@ -1962,7 +1962,7 @@ CREATE FUNCTION averageFinal ( state tuple<int,bigint> )
     return Double.valueOf(r);
   ';
 
-CREATE AGGREGATE average ( int )
+CREATE OR REPLACE AGGREGATE average ( int )
   SFUNC averageState
   STYPE tuple<int,bigint>
   FINALFUNC averageFinal

src/java/org/apache/cassandra/cql3/functions/UDAggregate.java

@@ -27,6 +27,7 @@ import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.tracing.Tracing;
 
 /**
  * Base class for user-defined-aggregates.
@@ -142,6 +143,9 @@ public class UDAggregate extends AbstractFunction implements AggregateFunction
     {
         return new Aggregate()
         {
+            private long stateFunctionCount;
+            private long stateFunctionDuration;
+
             private ByteBuffer state;
             {
                 reset();
@@ -149,6 +153,8 @@ public class UDAggregate extends AbstractFunction implements AggregateFunction
 
             public void addInput(int protocolVersion, List<ByteBuffer> values) throws InvalidRequestException
             {
+                long startTime = System.nanoTime();
+                stateFunctionCount++;
                 List<ByteBuffer> fArgs = new ArrayList<>(values.size() + 1);
                 fArgs.add(state);
                 fArgs.addAll(values);
@@ -162,19 +168,26 @@ public class UDAggregate extends AbstractFunction implements AggregateFunction
                 {
                     state = stateFunction.execute(protocolVersion, fArgs);
                 }
+                stateFunctionDuration += (System.nanoTime() - startTime) / 1000;
             }
 
             public ByteBuffer compute(int protocolVersion) throws InvalidRequestException
             {
+                // final function is traced in UDFunction
+                Tracing.trace("Executed UDA {}: {} call(s) to state function {} in {}\u03bcs", name(), stateFunctionCount, stateFunction.name(), stateFunctionDuration);
                 if (finalFunction == null)
                     return state;
+
                 List<ByteBuffer> fArgs = Collections.singletonList(state);
-                return finalFunction.execute(protocolVersion, fArgs);
+                ByteBuffer result = finalFunction.execute(protocolVersion, fArgs);
+                return result;
             }
 
             public void reset()
             {
                 state = initcond != null ? initcond.duplicate() : null;
+                stateFunctionDuration = 0;
+                stateFunctionCount = 0;
             }
         };
     }

src/java/org/apache/cassandra/cql3/functions/UDFunction.java

@@ -34,6 +34,7 @@ import org.apache.cassandra.cql3.*;
 import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.exceptions.*;
 import org.apache.cassandra.service.MigrationManager;
+import org.apache.cassandra.tracing.Tracing;
 import org.apache.cassandra.utils.ByteBufferUtil;
 
 /**
@@ -140,7 +141,11 @@ public abstract class UDFunction extends AbstractFunction implements ScalarFunct
 
         if (!isCallableWrtNullable(parameters))
             return null;
-        return executeUserDefined(protocolVersion, parameters);
+
+        long tStart = System.nanoTime();
+        ByteBuffer result = executeUserDefined(protocolVersion, parameters);
+        Tracing.trace("Executed UDF {} in {}\u03bcs", name(), (System.nanoTime() - tStart) / 1000);
+        return result;
     }
 
     public boolean isCallableWrtNullable(List<ByteBuffer> parameters)