[GitHub] [druid] clintropolis commented on a change in pull request #9317: ANY Aggregator should not skip null values implementation

[GitBox <gi...@apache.org>]

clintropolis commented on a change in pull request #9317: ANY Aggregator should not skip null values implementation
URL: https://github.com/apache/druid/pull/9317#discussion_r376653240
 
 

 ##########
 File path: processing/src/main/java/org/apache/druid/query/aggregation/any/DoubleAnyAggregatorFactory.java
 ##########
 @@ -19,103 +19,214 @@
 
 package org.apache.druid.query.aggregation.any;
 
-import com.fasterxml.jackson.annotation.JacksonInject;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import org.apache.druid.math.expr.ExprMacroTable;
+import com.google.common.base.Preconditions;
+import org.apache.druid.java.util.common.UOE;
+import org.apache.druid.query.aggregation.AggregateCombiner;
 import org.apache.druid.query.aggregation.Aggregator;
 import org.apache.druid.query.aggregation.AggregatorFactory;
 import org.apache.druid.query.aggregation.AggregatorUtil;
 import org.apache.druid.query.aggregation.BufferAggregator;
-import org.apache.druid.query.aggregation.SimpleDoubleAggregatorFactory;
+import org.apache.druid.query.aggregation.DoubleSumAggregator;
 import org.apache.druid.query.cache.CacheKeyBuilder;
 import org.apache.druid.segment.BaseDoubleColumnValueSelector;
+import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.NilColumnValueSelector;
+import org.apache.druid.segment.column.ColumnHolder;
 
 import javax.annotation.Nullable;
+import java.nio.ByteBuffer;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.List;
+import java.util.Objects;
 
-public class DoubleAnyAggregatorFactory extends SimpleDoubleAggregatorFactory
+public class DoubleAnyAggregatorFactory extends AggregatorFactory
 {
+  private static final Comparator<Double> VALUE_COMPARATOR = Comparator.nullsFirst(Double::compare);
+
+  private static final Aggregator NIL_AGGREGATOR = new DoubleAnyAggregator(
+      NilColumnValueSelector.instance()
+  )
+  {
+    @Override
+    public void aggregate()
+    {
+      // no-op
+    }
+  };
+
+  private static final BufferAggregator NIL_BUFFER_AGGREGATOR = new DoubleAnyBufferAggregator(
+      NilColumnValueSelector.instance()
+  )
+  {
+    @Override
+    public void aggregate(ByteBuffer buf, int position)
+    {
+      // no-op
+    }
+  };
+
+  private final String fieldName;
+  private final String name;
+  private final boolean storeDoubleAsFloat;
+
   @JsonCreator
   public DoubleAnyAggregatorFactory(
       @JsonProperty("name") String name,
-      @JsonProperty("fieldName") final String fieldName,
-      @JsonProperty("expression") @Nullable String expression,
-      @JacksonInject ExprMacroTable macroTable
+      @JsonProperty("fieldName") final String fieldName
   )
   {
-    super(macroTable, name, fieldName, expression);
-  }
+    Preconditions.checkNotNull(name, "Must have a valid, non-null aggregator name");
+    Preconditions.checkNotNull(fieldName, "Must have a valid, non-null fieldName");
 
-  public DoubleAnyAggregatorFactory(String name, String fieldName)
-  {
-    this(name, fieldName, null, ExprMacroTable.nil());
+    this.name = name;
+    this.fieldName = fieldName;
+    this.storeDoubleAsFloat = ColumnHolder.storeDoubleAsFloat();
   }
 
   @Override
-  protected double nullValue()
+  public Aggregator factorize(ColumnSelectorFactory metricFactory)
   {
-    return Double.NaN;
+    final BaseDoubleColumnValueSelector valueSelector = metricFactory.makeColumnValueSelector(fieldName);
+    if (valueSelector instanceof NilColumnValueSelector) {
+      return NIL_AGGREGATOR;
+    } else {
+      return new DoubleAnyAggregator(
+          valueSelector
+      );
+    }
   }
 
   @Override
-  protected Aggregator buildAggregator(BaseDoubleColumnValueSelector selector)
+  public BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory)
   {
-    return new DoubleAnyAggregator(selector);
+    final BaseDoubleColumnValueSelector valueSelector = metricFactory.makeColumnValueSelector(fieldName);
+    if (valueSelector instanceof NilColumnValueSelector) {
+      return NIL_BUFFER_AGGREGATOR;
+    } else {
+      return new DoubleAnyBufferAggregator(
+          valueSelector
+      );
+    }
   }
 
   @Override
-  protected BufferAggregator buildBufferAggregator(BaseDoubleColumnValueSelector selector)
+  public Comparator getComparator()
   {
-    return new DoubleAnyBufferAggregator(selector);
+    return VALUE_COMPARATOR;
   }
 
   @Override
   @Nullable
   public Object combine(@Nullable Object lhs, @Nullable Object rhs)
   {
-    if (lhs != null) {
-      return lhs;
-    } else {
-      return rhs;
-    }
+    return lhs;
+  }
+
+  @Override
+  public AggregateCombiner makeAggregateCombiner()
+  {
+    throw new UOE("DoubleAnyAggregatorFactory is not supported during ingestion for rollup");
   }
 
   @Override
   public AggregatorFactory getCombiningFactory()
   {
-    return new DoubleAnyAggregatorFactory(name, name, null, macroTable);
+    return new DoubleAnyAggregatorFactory(name, name);
   }
 
   @Override
   public List<AggregatorFactory> getRequiredColumns()
   {
-    return Collections.singletonList(new DoubleAnyAggregatorFactory(fieldName, fieldName, expression, macroTable));
+    return Collections.singletonList(new DoubleAnyAggregatorFactory(fieldName, fieldName));
+  }
+
+  @Override
+  public Object deserialize(Object object)
+  {
+    // handle "NaN" / "Infinity" values serialized as strings in JSON
+    if (object instanceof String) {
+      return Double.parseDouble((String) object);
+    }
+    return object;
+  }
+
+  @Override
+  @Nullable
+  public Object finalizeComputation(@Nullable Object object)
+  {
+    return object;
+  }
+
+  @Override
+  @JsonProperty
+  public String getName()
+  {
+    return name;
+  }
+
+  @JsonProperty
+  public String getFieldName()
+  {
+    return fieldName;
+  }
+
+  @Override
+  public List<String> requiredFields()
+  {
+    return Collections.singletonList(fieldName);
   }
 
   @Override
   public byte[] getCacheKey()
   {
     return new CacheKeyBuilder(AggregatorUtil.DOUBLE_ANY_CACHE_TYPE_ID)
         .appendString(fieldName)
-        .appendString(expression)
         .build();
   }
 
+  @Override
+  public String getTypeName()
+  {
+    // if we don't pretend to be a primitive, group by v1 gets sad and doesn't work because no complex type serde
 
 Review comment:
   this comment seems not applicable since it is a primitive

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org