<commit_before>/*
 *    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.
 */

package org.kairosdb.core.aggregator;

import org.joda.time.Chronology;
import org.joda.time.DateTime;
import org.joda.time.DateTimeField;
import org.joda.time.DateTimeZone;
import org.joda.time.chrono.GregorianChronology;
import org.kairosdb.core.DataPoint;
import org.kairosdb.core.datastore.DataPointGroup;
import org.kairosdb.core.datastore.Sampling;
import org.kairosdb.core.datastore.TimeUnit;

import javax.validation.Valid;
import javax.validation.constraints.NotNull;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Iterator;
import java.util.TimeZone;

import static com.google.common.base.Preconditions.checkNotNull;

public abstract class RangeAggregator implements Aggregator, TimezoneAware
{
	private long m_startTime = 0L;
	private long m_queryStartTime = 0L;
	private long m_queryEndTime = 0L;
	private boolean m_started = false;
	private boolean m_alignSampling;
	private boolean m_exhaustive;
	private DateTimeZone m_timeZone = DateTimeZone.UTC;
	

	@NotNull
	@Valid
	protected Sampling m_sampling = new Sampling(1, TimeUnit.MILLISECONDS);
	protected boolean m_alignStartTime;

	public RangeAggregator()
	{
		this(false);
	}

	public RangeAggregator(boolean exhaustive)
	{
		m_exhaustive = exhaustive;
	}

	public DataPointGroup aggregate(DataPointGroup dataPointGroup)
	{
		checkNotNull(dataPointGroup);

		if (m_alignSampling)
			m_startTime = alignRangeBoundary(m_startTime);

		if (m_exhaustive)
			return (new ExhaustiveRangeDataPointAggregator(dataPointGroup, getSubAggregator()));
		else
			return (new RangeDataPointAggregator(dataPointGroup, getSubAggregator()));
	}

	/**
	 For YEARS, MONTHS, WEEKS, DAYS:
	 Computes the timestamp of the first millisecond of the day
	 of the timestamp.
	 For HOURS,
	 Computes the timestamp of the first millisecond of the hour
	 of the timestamp.
	 For MINUTES,
	 Computes the timestamp of the first millisecond of the minute
	 of the timestamp.
	 For SECONDS,
	 Computes the timestamp of the first millisecond of the second
	 of the timestamp.
	 For MILLISECONDS,
	 returns the timestamp

	 @param timestamp
	 @return
	 */
	@SuppressWarnings("fallthrough")
	private long alignRangeBoundary(long timestamp)
	{
		DateTime dt = new DateTime(timestamp, m_timeZone);
		TimeUnit tu = m_sampling.getUnit();
		switch (tu)
		{
			case YEARS:
				dt = dt.withDayOfYear(1).withMillisOfDay(0);
				break;
			case MONTHS:
				dt = dt.withDayOfMonth(1).withMillisOfDay(0);
				break;
			case WEEKS:
				dt = dt.withDayOfWeek(1).withMillisOfDay(0);
				break;
			case DAYS:
			case HOURS:
			case MINUTES:
			case SECONDS:
				dt = dt.withHourOfDay(0);
				dt = dt.withMinuteOfHour(0);
				dt = dt.withSecondOfMinute(0);
			default:
				dt = dt.withMillisOfSecond(0);
				break;
		}
		return dt.getMillis();
	}

	public void setSampling(Sampling sampling)
	{
		m_sampling = sampling;
	}

	/**
	 When set to true the time for the aggregated data point for each range will
	 fall on the start of the range instead of being the value for the first
	 data point within that range.
	 @param align
	 */
	public void setAlignStartTime(boolean align)
	{
		m_alignStartTime = align;
	}

	/**
	 Setting this to true will cause the aggregation range to be aligned based on
	 the sampling size.  For example if your sample size is either milliseconds,
	 seconds, minutes or hours then the start of the range will always be at the top
	 of the hour.  The effect of setting this to true is that your data will
	 take the same shape when graphed as you refresh the data.
	 @param align Set to true to align the range on fixed points instead of
	              the start of the query.
	 */
	public void setAlignSampling(boolean align)
	{
		m_alignSampling = align;
	}

	/**
	 Start time to calculate the ranges from.  Typically this is the start
	 of the query
	 @param startTime
	 */
	public void setStartTime(long startTime)
	{
		m_startTime = startTime;
		m_queryStartTime = startTime;
	}

	public void setEndTime(long endTime)
	{
		m_queryEndTime = endTime;
	}

	/**
	 Return a RangeSubAggregator that will be used to aggregate data over a
	 discrete range of data points.  This is called once per grouped data series.

	 For example, if one metric is queried and no grouping is done this method is
	 called once and the resulting object is called over and over for each range
	 within the results.

	 If the query were grouping by the host tag and host has values of 'A' and 'B'
	 this method will be called twice, once to aggregate results for 'A' and once
	 to aggregate results for 'B'.
	 @return
	 */
	protected abstract RangeSubAggregator getSubAggregator();

	/**
	 Sets the time zone to use for range calculations

	 @param timeZone
	 */
	public void setTimeZone(DateTimeZone timeZone)<commit_msg>Fix to keep exhaustive agg from going on into forever<commit_after>/*
 *    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.
 */

package org.kairosdb.core.aggregator;

import org.joda.time.Chronology;
import org.joda.time.DateTime;
import org.joda.time.DateTimeField;
import org.joda.time.DateTimeZone;
import org.joda.time.chrono.GregorianChronology;
import org.kairosdb.core.DataPoint;
import org.kairosdb.core.datastore.DataPointGroup;
import org.kairosdb.core.datastore.Sampling;
import org.kairosdb.core.datastore.TimeUnit;

import javax.validation.Valid;
import javax.validation.constraints.NotNull;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Iterator;
import java.util.TimeZone;

import static com.google.common.base.Preconditions.checkNotNull;

public abstract class RangeAggregator implements Aggregator, TimezoneAware
{
	private long m_startTime = 0L;
	private long m_queryStartTime = 0L;
	private long m_queryEndTime = 0L;
	private boolean m_started = false;
	private boolean m_alignSampling;
	private boolean m_exhaustive;
	private DateTimeZone m_timeZone = DateTimeZone.UTC;
	

	@NotNull
	@Valid
	protected Sampling m_sampling = new Sampling(1, TimeUnit.MILLISECONDS);
	protected boolean m_alignStartTime;

	public RangeAggregator()
	{
		this(false);
	}

	public RangeAggregator(boolean exhaustive)
	{
		m_exhaustive = exhaustive;
	}

	public DataPointGroup aggregate(DataPointGroup dataPointGroup)
	{
		checkNotNull(dataPointGroup);

		if (m_alignSampling)
			m_startTime = alignRangeBoundary(m_startTime);

		if (m_exhaustive)
			return (new ExhaustiveRangeDataPointAggregator(dataPointGroup, getSubAggregator()));
		else
			return (new RangeDataPointAggregator(dataPointGroup, getSubAggregator()));
	}

	/**
	 For YEARS, MONTHS, WEEKS, DAYS:
	 Computes the timestamp of the first millisecond of the day
	 of the timestamp.
	 For HOURS,
	 Computes the timestamp of the first millisecond of the hour
	 of the timestamp.
	 For MINUTES,
	 Computes the timestamp of the first millisecond of the minute
	 of the timestamp.
	 For SECONDS,
	 Computes the timestamp of the first millisecond of the second
	 of the timestamp.
	 For MILLISECONDS,
	 returns the timestamp

	 @param timestamp
	 @return
	 */
	@SuppressWarnings("fallthrough")
	private long alignRangeBoundary(long timestamp)
	{
		DateTime dt = new DateTime(timestamp, m_timeZone);
		TimeUnit tu = m_sampling.getUnit();
		switch (tu)
		{
			case YEARS:
				dt = dt.withDayOfYear(1).withMillisOfDay(0);
				break;
			case MONTHS:
				dt = dt.withDayOfMonth(1).withMillisOfDay(0);
				break;
			case WEEKS:
				dt = dt.withDayOfWeek(1).withMillisOfDay(0);
				break;
			case DAYS:
			case HOURS:
			case MINUTES:
			case SECONDS:
				dt = dt.withHourOfDay(0);
				dt = dt.withMinuteOfHour(0);
				dt = dt.withSecondOfMinute(0);
			default:
				dt = dt.withMillisOfSecond(0);
				break;
		}
		return dt.getMillis();
	}

	public void setSampling(Sampling sampling)
	{
		m_sampling = sampling;
	}

	/**
	 When set to true the time for the aggregated data point for each range will
	 fall on the start of the range instead of being the value for the first
	 data point within that range.
	 @param align
	 */
	public void setAlignStartTime(boolean align)
	{
		m_alignStartTime = align;
	}

	/**
	 Setting this to true will cause the aggregation range to be aligned based on
	 the sampling size.  For example if your sample size is either milliseconds,
	 seconds, minutes or hours then the start of the range will always be at the top
	 of the hour.  The effect of setting this to true is that your data will
	 take the same shape when graphed as you refresh the data.
	 @param align Set to true to align the range on fixed points instead of
	              the start of the query.
	 */
	public void setAlignSampling(boolean align)
	{
		m_alignSampling = align;
	}

	/**
	 Start time to calculate the ranges from.  Typically this is the start
	 of the query
	 @param startTime
	 */
	public void setStartTime(long startTime)
	{
		m_startTime = startTime;
		m_queryStartTime = startTime;
	}

	public void setEndTime(long endTime)
	{
		//This is to tell the exhaustive agg when to stop.
		//If the end time is not specified in the query the end time is
		//set to MAX_LONG which causes exhaustive agg to go on forever.
		long now = System.currentTimeMillis();
		if (endTime > now)
			m_queryEndTime = now;
		else
			m_queryEndTime = endTime;
	}

	/**
	 Return a RangeSubAggregator that will be used to aggregate data over a
	 discrete range of data points.  This is called once per grouped data series.

	 For example, if one metric is queried and no grouping is done this method is
	 called once and the resulting object is called over and over for each range
	 within the results.

	 If the query were grouping by the host tag and host has values of 'A' and 'B'
	 this method will be called twice, once to aggregate results for 'A' and once
	 to aggregate results for 'B'.
	 @return
	 */
	protected abstract RangeSubAggregator getSubAggregator();

	/**
	 Sets the time zone to use for range calculations

	 @param timeZone
	 */
	public void setTimeZone(DateTimeZone timeZone)
{'fraction_non_alphanumeric': 0.06853526220614828, 'fraction_numerical': 0.0032549728752260397, 'mean_word_length': 3.825916230366492, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3871552', 'n_tokens_mistral': 3578, 'n_tokens_neox': 3274, 'n_words': 1238}