Feat/elastic like percentile aggrs

src/aggregate.h

@@ -25,6 +25,7 @@ class AggrFunc_i
 	virtual void	Update ( CSphMatch & tDst, const CSphMatch & tSrc, bool bGrouped, bool bMerge ) = 0;
 	virtual void	Setup ( CSphMatch & tDst, const CSphMatch & tSrc, bool bMerge ) {}
 	virtual void	Finalize ( CSphMatch & tDst ) {}
+	virtual void	Discard ( CSphMatch & tDst ) {}
 	virtual void	SetColumnar ( columnar::Columnar_i * pColumnar ) {}
 };
 
@@ -34,6 +35,10 @@ AggrFunc_i * CreateAggrAvg ( const CSphColumnInfo & tAttr, const CSphAttrLocator
 AggrFunc_i * CreateAggrMin ( const CSphColumnInfo & tAttr );
 AggrFunc_i * CreateAggrMax ( const CSphColumnInfo & tAttr );
 
+AggrFunc_i * CreateAggrPercentiles ( const CSphColumnInfo & tAttr );
+AggrFunc_i * CreateAggrPercentileRanks ( const CSphColumnInfo & tAttr );
+AggrFunc_i * CreateAggrMad ( const CSphColumnInfo & tAttr );
+
 AggrFunc_i * CreateAggrConcat ( const CSphColumnInfo & tAttr );
 
 #endif // _aggregate_

src/aggrexpr.h

@@ -63,7 +63,10 @@ enum class Aggr_e
 	MIN,
 	MAX,
 	SUM,
-	AVG
+	AVG,
+	PERCENTILES,
+	PERCENTILE_RANKS,
+	MAD
 };
 
 struct DateRangeSetting_t
@@ -85,13 +88,35 @@ struct AggrDateHistSetting_t
 	bool m_bFixed = false;
 };
 
+struct AggrPercentilesSetting_t
+{
+	CSphVector<float> m_dPercents;
+	bool m_bKeyed = false;
+	double m_fCompression = 200.0;
+};
+
+struct AggrPercentileRanksSetting_t
+{
+	CSphVector<double> m_dValues;
+	bool m_bKeyed = false;
+	double m_fCompression = 200.0;
+};
+
+struct AggrMadSetting_t
+{
+	double m_fCompression = 1000.0;
+};
+
 struct AggrSettings_t
 {
 	Aggr_e		m_eAggrFunc { Aggr_e::NONE };
 	AggrRangeSetting_t m_tRange;
 	AggrHistSetting_t m_tHist;
 	AggrDateRangeSetting_t m_tDateRange;
 	AggrDateHistSetting_t m_tDateHist;
+	AggrPercentilesSetting_t m_tPercentiles;
+	AggrPercentileRanksSetting_t m_tPercentileRanks;
+	AggrMadSetting_t m_tMad;
 };
 
 ISphExpr * CreateExprRange ( ISphExpr * pAttr, const AggrRangeSetting_t & tRanges );

src/attribute.cpp

@@ -9,6 +9,7 @@
 //
 
 #include "attribute.h"
+#include "std/tdigest_runtime.h"
 
 #include "sphinxint.h"
 #include "sphinxjson.h"
@@ -1139,6 +1140,34 @@ ByteBlob_t sphUnpackPtrAttr ( const BYTE * pData )
 	return { pData, iLen };
 }
 
+BYTE * sphCopyPackedAttr ( const BYTE * pData )
+{
+	ByteBlob_t dBlob = sphUnpackPtrAttr ( pData );
+	if ( sphIsTDigestRuntimeBlob ( dBlob ) )
+		return sphCloneTDigestRuntimeBlob ( dBlob );
+	return sphPackPtrAttr ( dBlob );
+}
+
+
+void sphDeallocatePacked ( const BYTE* pBlob )
+{
+	if ( !pBlob )
+		return;
+
+	const BYTE * pPayload = pBlob;
+	int iLen = (int)UnzipIntBE ( pPayload );
+	ByteBlob_t dBlob { pPayload, iLen };
+
+	if ( sphIsTDigestRuntimeBlob ( dBlob ) )
+		sphDestroyTDigestRuntimeBlob ( dBlob );
+
+#if WITH_SMALLALLOC
+	sphDeallocateSmall ( pBlob, sphCalcPackedLength ( iLen ) );
+#else
+	sphDeallocateSmall ( pBlob );
+#endif
+}
+
 
 ESphAttr sphPlainAttrToPtrAttr ( ESphAttr eAttrType )
 {

src/attribute.h

@@ -134,7 +134,7 @@ ESphAttr			sphPlainAttrToPtrAttr ( ESphAttr eAttrType );
 bool				sphIsDataPtrAttr ( ESphAttr eAttrType );
 
 // just repack (matter of optimizing)
-FORCE_INLINE BYTE * sphCopyPackedAttr ( const BYTE * pData ) { return sphPackPtrAttr ( sphUnpackPtrAttr ( pData ) ); }
+BYTE *				sphCopyPackedAttr ( const BYTE * pData );
 
 //////////////////////////////////////////////////////////////////////////
 // misc attribute-related
@@ -159,17 +159,7 @@ FORCE_INLINE DocID_t sphGetDocID ( const CSphRowitem * pData )
 #endif
 }
 
-FORCE_INLINE void sphDeallocatePacked ( const BYTE* pBlob )
-{
-	if ( !pBlob )
-		return;
-#if WITH_SMALLALLOC
-	const BYTE * pFoo = pBlob;
-	sphDeallocateSmall ( pBlob, sphCalcPackedLength ( UnzipIntBE ( pFoo ) ) );
-#else
-	sphDeallocateSmall ( pBlob );
-#endif
-}
+void				sphDeallocatePacked ( const BYTE* pBlob );
 
 const char * AttrType2Str ( ESphAttr eAttrType );
 

src/daemon/minimize_aggr_result.cpp

@@ -479,6 +479,7 @@ static void RecoverAggregateFunctions ( const CSphQuery & tQuery, const AggrResu
 			{
 				assert ( tCol.m_eAggrFunc==SPH_AGGR_NONE );
 				tCol.m_eAggrFunc = tItem.m_eAggrFunc;
+				tCol.m_fTdigestCompression = tItem.m_fTdigestCompression;
 			}
 		}
 	}

src/daemon/search_handler.cpp

@@ -2284,6 +2284,9 @@ static ESphAggrFunc GetAggr ( Aggr_e eAggrFunc )
 	case Aggr_e::MAX: return SPH_AGGR_MAX;
 	case Aggr_e::SUM: return SPH_AGGR_SUM;
 	case Aggr_e::AVG: return SPH_AGGR_AVG;
+	case Aggr_e::PERCENTILES: return SPH_AGGR_PERCENTILES;
+	case Aggr_e::PERCENTILE_RANKS: return SPH_AGGR_PERCENTILE_RANKS;
+	case Aggr_e::MAD: return SPH_AGGR_MAD;
 	default: return SPH_AGGR_NONE;
 	}
 }
@@ -2361,8 +2364,11 @@ SearchHandler_c CreateMsearchHandler ( std::unique_ptr<QueryParser_i> pQueryPars
 	{
 		const JsonAggr_t & tBucket = tQuery.m_dAggs[i];
 
-		// add only new items
-		if ( hAttrs[tBucket.m_sCol] )
+		const bool bHasAggrFunc = ( tBucket.m_eAggrFunc!=Aggr_e::NONE );
+		CSphString sAlias = bHasAggrFunc ? GetAggrName ( i, tBucket.m_sCol ) : tBucket.m_sCol;
+
+		// add only new items (by alias)
+		if ( hAttrs[sAlias] )
 			continue;
 
 		if ( tBucket.m_eAggrFunc==Aggr_e::COUNT )
@@ -2378,14 +2384,22 @@ SearchHandler_c CreateMsearchHandler ( std::unique_ptr<QueryParser_i> pQueryPars
 		if ( tBucket.m_eAggrFunc!=Aggr_e::NONE )
 		{
 			tItem.m_sExpr = DumpAggr ( tBucket.m_sCol.cstr(), tBucket );
-			tItem.m_sAlias = GetAggrName ( i, tBucket.m_sCol );
+			tItem.m_sAlias = sAlias;
 			tItem.m_eAggrFunc = GetAggr ( tBucket.m_eAggrFunc );
+			switch ( tBucket.m_eAggrFunc )
+			{
+			case Aggr_e::PERCENTILES: tItem.m_fTdigestCompression = tBucket.m_tPercentiles.m_fCompression; break;
+			case Aggr_e::PERCENTILE_RANKS: tItem.m_fTdigestCompression = tBucket.m_tPercentileRanks.m_fCompression; break;
+			case Aggr_e::MAD: tItem.m_fTdigestCompression = tBucket.m_tMad.m_fCompression; break;
+			default: break;
+			}
 		} else
 		{
 			tItem.m_sExpr = tBucket.m_sCol;
-			tItem.m_sAlias = tBucket.m_sCol;
-			hAttrs.Add ( tBucket.m_sCol );
+			tItem.m_sAlias = sAlias;
 		}
+
+		hAttrs.Add ( tItem.m_sAlias );
 	}
 
 	tQuery.m_bFacetHead = true;
@@ -2406,6 +2420,9 @@ SearchHandler_c CreateMsearchHandler ( std::unique_ptr<QueryParser_i> pQueryPars
 		tQuery.m_bFacet = true;
 
 		// select list to facet query
+		if ( tBucket.m_eAggrFunc==Aggr_e::PERCENTILES || tBucket.m_eAggrFunc==Aggr_e::PERCENTILE_RANKS || tBucket.m_eAggrFunc==Aggr_e::MAD )
+			tQuery.m_sSelect = DumpAggr ( tBucket.m_sCol.cstr(), tBucket );
+		else
 		tQuery.m_sSelect.SetSprintf ( "%s", tBucket.m_sCol.cstr() );
 
 		// ref items to facet query
@@ -2435,11 +2452,21 @@ SearchHandler_c CreateMsearchHandler ( std::unique_ptr<QueryParser_i> pQueryPars
 			case Aggr_e::MAX:
 			case Aggr_e::SUM:
 			case Aggr_e::AVG:
+			case Aggr_e::PERCENTILES:
+			case Aggr_e::PERCENTILE_RANKS:
+			case Aggr_e::MAD:
 			{
 				CSphQueryItem & tItem = tQuery.m_dRefItems.Add();
 				tItem.m_sExpr = DumpAggr ( tBucket.m_sCol.cstr(), tBucket );
 				tItem.m_sAlias = GetAggrName ( i, tBucket.m_sCol );
 				tItem.m_eAggrFunc = GetAggr ( tBucket.m_eAggrFunc );
+				switch ( tBucket.m_eAggrFunc )
+				{
+				case Aggr_e::PERCENTILES: tItem.m_fTdigestCompression = tBucket.m_tPercentiles.m_fCompression; break;
+				case Aggr_e::PERCENTILE_RANKS: tItem.m_fTdigestCompression = tBucket.m_tPercentileRanks.m_fCompression; break;
+				case Aggr_e::MAD: tItem.m_fTdigestCompression = tBucket.m_tMad.m_fCompression; break;
+				default: break;
+				}
 			}
 			break;
 
@@ -2473,6 +2500,9 @@ SearchHandler_c CreateMsearchHandler ( std::unique_ptr<QueryParser_i> pQueryPars
 			case Aggr_e::MAX:
 			case Aggr_e::SUM:
 			case Aggr_e::AVG:
+			case Aggr_e::PERCENTILES:
+			case Aggr_e::PERCENTILE_RANKS:
+			case Aggr_e::MAD:
 				break;
 
 			case Aggr_e::COMPOSITE:

src/queuecreator.cpp

@@ -277,6 +277,8 @@ class QueueCreator_c
 	bool						m_bHeadWOGroup;
 	bool						m_bGotDistinct;
 	bool						m_bExprsNeedDocids = false;
+	bool						m_bSlimTdigestCandidate = true;
+	bool						m_bHasTdigestAgg = false;
 
 	// for sorter to create pooled attributes
 	bool						m_bHaveStar = false;
@@ -352,6 +354,7 @@ class QueueCreator_c
 	std::unique_ptr<ISphFilter>	CreateAggrFilter() const;
 	void				SetupCollation();
 	bool				Err ( const char * sFmt, ... ) const;
+	bool				ShouldEnableSlimTdigest() const;
 };
 
 
@@ -613,6 +616,7 @@ bool QueueCreator_c::SetupGroupbySettings ( bool bHasImplicitGrouping )
 	if ( bHasImplicitGrouping )
 	{
 		m_tGroupSorterSettings.m_bImplicit = true;
+		m_tGroupSorterSettings.m_bSlimTdigest = ShouldEnableSlimTdigest();
 		return true;
 	}
 
@@ -809,6 +813,18 @@ void QueueCreator_c::PropagateEvalStage ( CSphColumnInfo & tExprCol, StrVec_t &
 
 bool QueueCreator_c::SetupAggregateExpr ( CSphColumnInfo & tExprCol, const CSphString & sExpr, DWORD uQueryPackedFactorFlags )
 {
+	const bool bIsSlimTdigest = ( tExprCol.m_eAggrFunc==SPH_AGGR_PERCENTILES
+		|| tExprCol.m_eAggrFunc==SPH_AGGR_PERCENTILE_RANKS
+		|| tExprCol.m_eAggrFunc==SPH_AGGR_MAD );
+
+	if ( tExprCol.m_eAggrFunc!=SPH_AGGR_NONE )
+	{
+		if ( bIsSlimTdigest )
+			m_bHasTdigestAgg = true;
+		else
+			m_bSlimTdigestCandidate = false;
+	}
+
 	// validate that MAX/MIN/SUM/AVG cannot be used on string/text columns
 	// This check must happen BEFORE the switch statement that may modify tExprCol.m_eAttrType
 	if ( tExprCol.m_eAggrFunc==SPH_AGGR_MAX || tExprCol.m_eAggrFunc==SPH_AGGR_MIN 
@@ -830,6 +846,13 @@ bool QueueCreator_c::SetupAggregateExpr ( CSphColumnInfo & tExprCol, const CSphS
 		tExprCol.m_eAttrType = SPH_ATTR_DOUBLE;
 		tExprCol.m_tLocator.m_iBitCount = 64;
 		break;
+	case SPH_AGGR_PERCENTILES:
+	case SPH_AGGR_PERCENTILE_RANKS:
+	case SPH_AGGR_MAD:
+		tExprCol.m_eAggrInputType = tExprCol.m_eAttrType;
+		tExprCol.m_eAttrType = SPH_ATTR_STRINGPTR;
+		tExprCol.m_tLocator.m_iBitCount = ROWITEMPTR_BITS;
+		break;
 
 	case SPH_AGGR_CAT:
 		// force GROUP_CONCAT() to be computed as strings
@@ -993,6 +1016,8 @@ bool QueueCreator_c::ParseQueryItem ( const CSphQueryItem & tItem )
 	const CSphString & sExpr = tItem.m_sExpr;
 	bool bIsCount = IsCount(sExpr);
 	m_bHasCount |= bIsCount;
+	if ( tItem.m_eAggrFunc==SPH_AGGR_NONE && !bIsCount && !IsGroupbyMagic ( sExpr ) )
+		m_bSlimTdigestCandidate = false;
 
 	if ( sExpr=="*" )
 	{
@@ -1098,6 +1123,7 @@ bool QueueCreator_c::ParseQueryItem ( const CSphQueryItem & tItem )
 	m_bZonespanlist |= bHasZonespanlist;
 	m_bExprsNeedDocids |= bExprsNeedDocids;
 	tExprCol.m_eAggrFunc = tItem.m_eAggrFunc;
+	tExprCol.m_fTdigestCompression = tItem.m_fTdigestCompression;
 	tExprCol.m_iIndex = iSorterAttr>= 0 ? m_pSorterSchema->GetAttrIndexOriginal ( tItem.m_sAlias.cstr() ) : -1;
 	if ( !tExprCol.m_pExpr )
 		return Err ( "parse error: %s", m_sError.cstr() );
@@ -2445,6 +2471,17 @@ ISphMatchSorter * QueueCreator_c::SpawnQueue()
 }
 
 
+bool QueueCreator_c::ShouldEnableSlimTdigest() const
+{
+	if ( !m_bSlimTdigestCandidate || !m_bHasTdigestAgg )
+		return false;
+	if ( m_tQuery.m_iLimit!=0 )
+		return false;
+	if ( m_tQuery.m_bFacet || m_tQuery.m_bFacetHead )
+		return false;
+	return true;
+}
+
 bool QueueCreator_c::SetupComputeQueue ()
 {
 	return AddJoinAttrs()

src/queuecreator.h

@@ -27,6 +27,7 @@ struct CSphGroupSorterSettings
 	CSphRefcountedPtr<CSphGrouper>		m_pGrouper;///< group key calculator
 	CSphRefcountedPtr<DistinctFetcher_i> m_pDistinctFetcher;
 	bool				m_bImplicit = false;///< for queries with aggregate functions but without group by clause
+	bool				m_bSlimTdigest = false; ///< implicit TDigest-only query, allow slim match buffer
 	SharedPtr_t<ISphFilter>	m_pAggrFilterTrait; ///< aggregate filter that got owned by grouper
 	bool				m_bJson = false;	///< whether we're grouping by Json attribute
 	int					m_iMaxMatches = 0;

src/schema/columninfo.cpp

@@ -29,6 +29,7 @@ bool CSphColumnInfo::operator== ( const CSphColumnInfo & rhs ) const
 {
 	return m_sName == rhs.m_sName
 		&& m_eAttrType == rhs.m_eAttrType
+		&& m_eAggrInputType == rhs.m_eAggrInputType
 		&& m_tLocator.m_iBitCount == rhs.m_tLocator.m_iBitCount
 		&& m_tLocator.m_iBitOffset == rhs.m_tLocator.m_iBitOffset
 		&& m_tLocator.m_bDynamic == rhs.m_tLocator.m_bDynamic;

src/schema/columninfo.h

@@ -52,6 +52,7 @@ struct CSphColumnInfo
 
 	CSphString		m_sName;							///< column name
 	ESphAttr		m_eAttrType;						///< attribute type
+	ESphAttr		m_eAggrInputType { SPH_ATTR_NONE };///< original attr type for aggregate inputs
 
 	ESphWordpart	m_eWordpart { SPH_WORDPART_WHOLE };	///< wordpart processing type
 	bool			m_bIndexed = false;					///< whether to index this column as fulltext field too
@@ -77,6 +78,8 @@ struct CSphColumnInfo
 	knn::ModelSettings_t m_tKNNModel;					///< knn model settings
 	CSphString		m_sKNNFrom;							///< fields/attrs used by the model
 
+	double			m_fTdigestCompression = 200.0;		///< tdigest compression for extended aggs
+
 	WORD			m_uNext = 0xFFFF;					///< next in linked list for hash in CSphSchema
 
 	/// handy ctor