//
// Copyright (c) 2020, Manticore Software LTD (https://manticoresearch.com)
// Copyright (c) 2001-2016, Andrew Aksyonoff
// Copyright (c) 2008-2016, Sphinx Technologies Inc
// All rights reserved
//
// This program is free software; you can redistribute it and/or modify
// it under the terms of the GNU General Public License. You should have
// received a copy of the GPL license along with this program; if you
// did not, you can find it at http://www.gnu.org/
//

#include "dynamic_idx.h"
#include "sphinxsort.h"
#include "querycontext.h"

using namespace Threads;

class GenericFeeder_c : public RowBuffer_i
{
protected:
	enum class FeedState_e : BYTE
	{
		Idle,
		SchemaReady,
		RowReady
	};

	TableFeeder_fn				m_fnFeed;
    bool 						m_bProducerStarted = false;
	std::atomic<bool>			m_bProducerFinished { false };
	std::atomic<FeedState_e>	m_eState { FeedState_e::Idle };
	Coro::Event_c				m_tCanProduce;
	Coro::Event_c				m_tCanConsume;

	// feeder stuff
	CSphSchema* 	m_pSchema = nullptr;
	CSphMatch*		m_pMatch = nullptr;
	bool			m_bHaveMoreMatches = true;
	int				m_iCurMatch = 1;

	void StartProducer()
	{
		if ( m_bProducerStarted )
			return;

		m_bProducerStarted = true;

		auto * pScheduler = Coro::CurrentScheduler();
		if ( !pScheduler )
			pScheduler = GlobalWorkPool();

		StartJob ( [this, fnFeed = std::move ( m_fnFeed )] () mutable {
	 		m_tCanProduce.WaitEvent();
	 		fnFeed ( this );
	 		m_bProducerFinished.store ( true, std::memory_order_release );
	 		m_eState.store ( FeedState_e::Idle, std::memory_order_release );
	 		m_tCanConsume.SetEvent();
	 	}, pScheduler );
	}

	void PublishAndWait ( FeedState_e eState )
	{
		m_eState.store ( eState, std::memory_order_release );
		m_tCanConsume.SetEvent();
		m_tCanProduce.WaitEvent();
	}

	FeedState_e ReleaseProducerAndWaitChunk()
	{
		m_tCanProduce.SetEvent();
		m_tCanConsume.WaitEvent();
		return m_eState.load ( std::memory_order_acquire );
	}

	void ReleaseProducerAndWaitForFinish()
	{
		m_bHaveMoreMatches = false;
		m_tCanProduce.SetEvent();
		while ( !m_bProducerFinished.load ( std::memory_order_acquire ) )
			m_tCanConsume.WaitEvent();
	}

	void DrainProducer() noexcept
	{
		if ( !m_bProducerStarted || m_bProducerFinished.load ( std::memory_order_acquire ) )
			return;

		m_pMatch = nullptr; // ignore any payload while draining
		while ( !m_bProducerFinished.load ( std::memory_order_acquire ) )
			ReleaseProducerAndWaitChunk();
	}

	void ResetToIdle () noexcept
	{
		m_bHaveMoreMatches = false;
		m_pMatch = nullptr; // that should stop any further feeding
		PublishAndWait ( FeedState_e::Idle );
	}

	void PutStrt ( const CSphColumnInfo & tCol, Str_t sMsg )
	{
		assert ( m_pMatch );
		assert ( tCol.m_eAttrType == SPH_ATTR_STRINGPTR );
		BYTE * pData = nullptr;
		m_pMatch->SetAttr ( tCol.m_tLocator, (SphAttr_t) sphPackPtrAttr ( sMsg.second, &pData ) );
		memcpy ( pData, sMsg.first, sMsg.second );
	}

public:
	explicit GenericFeeder_c ( TableFeeder_fn fnFeed )
		: m_fnFeed ( std::move ( fnFeed ) )
	{}

	~GenericFeeder_c() override
	{
		DrainProducer();
	}

	// set upstream match
	void SetSorterStuff ( CSphMatch * pMatch )
	{
		m_pMatch = pMatch;
	}

	virtual void PrepareFillNextMatch() noexcept = 0;

	bool FillNextMatch()
	{
		if ( !m_bHaveMoreMatches )
			return false;

		PrepareFillNextMatch();

		auto eState = ReleaseProducerAndWaitChunk();
		if ( eState==FeedState_e::RowReady )
			return m_bHaveMoreMatches;

		ReleaseProducerAndWaitForFinish();
		return false;
	}

	void Error ( const char * sError, EMYSQL_ERR ) override
	{
		m_bError = true;
		m_sError = sError;
		ResetToIdle ();
	}
	void Ok ( int, int, const char *, bool, int64_t ) override {}
	void Add ( BYTE ) override {}
};

class Feeder_c final : public GenericFeeder_c
{
	bool 			m_bAutoID = true;
	int 			m_iCurCol = 0;


	const CSphColumnInfo & GetNextCol()
	{
		assert ( m_pMatch );
		assert ( m_iCurCol>=0 );

		const CSphColumnInfo & dColQuery = m_pSchema->GetAttr ( m_iCurCol );
		++m_iCurCol;
		return dColQuery;
	}

	// insert column into schema
	void ColSchema ( const char * szName, MysqlColumnType_e uType )
	{
		ESphAttr eType = SPH_ATTR_STRINGPTR;
		switch ( uType )
		{
		case MYSQL_COL_LONGLONG:	eType = SPH_ATTR_BIGINT; break;
		case MYSQL_COL_LONG:		eType = SPH_ATTR_INTEGER; break;
		case MYSQL_COL_FLOAT:		eType = SPH_ATTR_FLOAT;	break;
		case MYSQL_COL_DOUBLE:		eType = SPH_ATTR_DOUBLE; break;
		default: break;
		}
		CSphString sName ( szName );
		sName.ToLower();
		if ( m_pSchema->GetAttrIndex ( sName.cstr() )<0 )
			m_pSchema->AddAttr ( CSphColumnInfo ( szName, eType ), true );
		else {
			assert ( sName=="id");
			m_bAutoID = false;
		}
	}

public:
	explicit Feeder_c ( TableFeeder_fn fnFeed )
		: GenericFeeder_c ( std::move ( fnFeed ) )
	{
	}

	// collecting schema
	void SetSchema ( CSphSchema * pSchema )
	{
		m_pSchema = pSchema;
		StartProducer();
		auto eState = ReleaseProducerAndWaitChunk();
		assert ( eState==FeedState_e::SchemaReady || eState==FeedState_e::Idle );
		if ( eState!=FeedState_e::SchemaReady )
			ReleaseProducerAndWaitForFinish();
	}

	void PrepareFillNextMatch() noexcept override
	{
		m_iCurCol = 0;
		if ( m_bAutoID )
		{
			auto * pID = m_pSchema->GetAttr ( sphGetDocidName () );
			m_pMatch->SetAttr ( pID->m_tLocator, m_iCurMatch );
			++m_iCurCol;
		}
		++m_iCurMatch;
	}

	void PutStr ( const CSphColumnInfo & tCol, const StringBuilder_c & sMsg )
	{
		assert ( m_pMatch );
		assert ( tCol.m_eAttrType == SPH_ATTR_STRINGPTR );
		PutStrt ( tCol, Str_t ( sMsg ) );
	}

public:
	// Header of the table with defined num of columns
	inline void HeadBegin () override
	{
		if ( !m_pSchema )
			return;

		// add id column
		ColSchema ( sphGetDocidName (), MYSQL_COL_LONGLONG );
	}

	// add the next column.
	void HeadColumn ( const char * sName, MysqlColumnType_e uType ) override
	{
		if ( m_pSchema )
			ColSchema ( sName, uType );
	}

	bool HeadEnd ( bool bMoreResults, int iWarns ) override
	{
		if ( !m_pSchema )
		{
			assert (false && "dynamic table invoked without parent schema");
			return false;
		}

		PublishAndWait ( FeedState_e::SchemaReady );
		return true;
	}

	// match constructing routines
	void PutFloatAsString ( float fVal, const char * sFormat ) override
	{
		if ( !m_pMatch )
			return;

		auto & tCol = GetNextCol ();
		auto & tMatch = *m_pMatch;
		if ( tCol.m_eAttrType!=SPH_ATTR_STRINGPTR )
			tMatch.SetAttrFloat ( tCol.m_tLocator, fVal );
		else
		{
			StringBuilder_c sData;
			sData.Appendf ( "%f", fVal );
			PutStr ( tCol, sData );
		}
	}

	void PutDoubleAsString ( double fVal, const char * szFormat ) override
	{
		if ( !m_pMatch )
			return;

		auto & tCol = GetNextCol ();
		auto & tMatch = *m_pMatch;
		if ( tCol.m_eAttrType!=SPH_ATTR_STRINGPTR )
			tMatch.SetAttrDouble ( tCol.m_tLocator, fVal );
		else
		{
			StringBuilder_c sData;
			sData.Appendf ( "%f", fVal );
			PutStr ( tCol, sData );
		}
	}

	void PutPercentAsString ( int64_t iVal, int64_t iBase ) override
	{
		if ( iBase )
			PutNumAsString ( iVal * 100 / iBase );
		else
			PutNumAsString ( 100 );
	}

	void PutNumAsString ( int64_t iVal ) override
	{
		if ( !m_pMatch )
			return;
		auto& tCol = GetNextCol();
		auto & tMatch = *m_pMatch;

		if ( tCol.m_eAttrType!=SPH_ATTR_STRINGPTR )
			tMatch.SetAttr ( tCol.m_tLocator, iVal );
		else
		{
			StringBuilder_c sData;
			sData << iVal;
			PutStr ( tCol, sData );
		}
	}

	void PutNumAsString ( uint64_t uVal ) override
	{
		if ( !m_pMatch )
			return;
		auto & tCol = GetNextCol ();
		auto & tMatch = *m_pMatch;

		if ( tCol.m_eAttrType!=SPH_ATTR_STRINGPTR )
			tMatch.SetAttr ( tCol.m_tLocator, uVal );
		else
		{
			StringBuilder_c sData;
			sData << uVal;
			PutStr ( tCol, sData );
		}
	}

	void PutNumAsString ( int iVal ) override
	{
		if ( !m_pMatch )
			return;
		auto & tCol = GetNextCol ();
		auto & tMatch = *m_pMatch;

		if ( tCol.m_eAttrType!=SPH_ATTR_STRINGPTR )
			tMatch.SetAttr ( tCol.m_tLocator, iVal );
		else
		{
			StringBuilder_c sData;
			sData << iVal;
			PutStr ( tCol, sData );
		}
	}

	void PutNumAsString ( DWORD uVal ) override
	{
		if ( !m_pMatch )
			return;
		auto & tCol = GetNextCol ();
		auto & tMatch = *m_pMatch;

		if ( tCol.m_eAttrType!=SPH_ATTR_STRINGPTR )
			tMatch.SetAttr ( tCol.m_tLocator, uVal );
		else
		{
			StringBuilder_c sData;
			sData << uVal;
			PutStr ( tCol, sData );
		}
	}

	void PutFloat ( float fVal ) final
	{
		PutFloatAsString ( fVal, nullptr );
	}

	void PutDouble ( double fVal ) final
	{
		PutDoubleAsString ( fVal, nullptr );
	}

	void PutInt ( int iVal ) final
	{
		PutNumAsString ( iVal );
	}

	void PutInt64 ( int64_t iVal ) final
	{
		PutNumAsString ( iVal );
	}

	void PutDWORD ( DWORD uVal ) final
	{
		PutNumAsString ( uVal );
	}

	void PutUint64 ( uint64_t uVal ) final
	{
		PutNumAsString ( uVal );
	}

	void PutArray ( const ByteBlob_t&, bool ) override {}

	// pack string
	void PutString ( Str_t sMsg ) override
	{
		if ( !m_pMatch )
			return;

		auto & tCol = GetNextCol ();
		auto & tMatch = *m_pMatch;

		BYTE * pData = nullptr;
		tMatch.SetAttr ( tCol.m_tLocator, (SphAttr_t) sphPackPtrAttr ( sMsg.second, &pData ) );
		memcpy ( pData, sMsg.first, sMsg.second );
	}

	void PutMicrosec ( int64_t iUsec ) override
	{
		if ( !m_pMatch )
			return;
		auto & tCol = GetNextCol ();
		auto & tMatch = *m_pMatch;

		if ( tCol.m_eAttrType == SPH_ATTR_FLOAT )
		{
			auto fSec = (float)iUsec / 1000000.0f;
			tMatch.SetAttr ( tCol.m_tLocator, sphF2DW ( fSec ) );
		} else if ( tCol.m_eAttrType==SPH_ATTR_STRINGPTR )
		{
			StringBuilder_c sData;
			sData << iUsec;
			PutStr ( tCol, sData );
		} else
			tMatch.SetAttr ( tCol.m_tLocator, iUsec );
	}

	void PutNULL () override
	{
		if ( !m_pMatch )
			return;
		auto & tCol = GetNextCol ();
		auto & tMatch = *m_pMatch;

		if ( tCol.m_eAttrType!=SPH_ATTR_STRINGPTR )
			tMatch.SetAttr ( tCol.m_tLocator, 0 );
		else
		{
			StringBuilder_c sData;
			sData << 0;
			PutStr ( tCol, sData );
		}
	}

public:
	/// more high level. Processing the whole tables.
	// sends collected data, then reset
	bool Commit() override
	{
		PublishAndWait ( FeedState_e::RowReady );
		return m_bHaveMoreMatches; // true for continue iteration, false to stop
	}

	// wrappers for popular packets
	void Eof ( bool bMoreResults, int iWarns, const char* ) override
	{
		ResetToIdle();
	}
	using RowBuffer_i::Eof;
};

// feed schema only and skip all the data
class FeederSchema_c final : public GenericFeeder_c
{
	void PutString ( int iCol, const char * sMsg )
	{
		if ( !m_pMatch )
			return;
		int iLen = ( sMsg && *sMsg ) ? (int) strlen ( sMsg ) : 0;

		if ( !sMsg )
			sMsg = "";

		PutStrt (m_pSchema->GetAttr ( iCol ), {sMsg,iLen});
	}

public:
	explicit FeederSchema_c ( TableFeeder_fn fnFeed )
		: GenericFeeder_c ( std::move ( fnFeed ) )
	{}


	// collecting schema
	void SetSchema ( CSphSchema * pSchema )
	{
		m_pSchema = pSchema;
		m_pSchema->AddAttr ( CSphColumnInfo ( sphGetDocidName (), SPH_ATTR_BIGINT ), true );
		m_pSchema->AddAttr ( CSphColumnInfo ( "Field", SPH_ATTR_STRINGPTR ), true );
		m_pSchema->AddAttr ( CSphColumnInfo ( "Type", SPH_ATTR_STRINGPTR ), true );
		m_pSchema->AddAttr ( CSphColumnInfo ( "Properties", SPH_ATTR_STRINGPTR ), true );
	}

	void PrepareFillNextMatch() noexcept override
	{
		StartProducer();
	}

public:
	void HeadBegin () override {}

	// add the next column.
	void HeadColumn ( const char * sName, MysqlColumnType_e uType ) override
	{
		if ( !m_pSchema )
			return;

		if ( !m_pMatch )
			return;

		// docid
		m_pMatch->SetAttr ( m_pSchema->GetAttr ( 0 ).m_tLocator, m_iCurMatch );
		++m_iCurMatch;

		PutString ( 1, sName );
		switch ( uType )
		{
			case MYSQL_COL_LONGLONG:	PutString ( 2, "bigint" ); break;
			case MYSQL_COL_LONG:		PutString ( 2, "uint" ); break;
			case MYSQL_COL_FLOAT:		PutString ( 2, "float" ); break;
			case MYSQL_COL_DOUBLE:		PutString ( 2, "double" ); break;
			default:					PutString ( 2, "string" ); break;
		}
		PutString ( 3, "" );

		PublishAndWait ( FeedState_e::RowReady );
	}

	bool HeadEnd ( bool bMoreResults, int iWarns ) override
	{
		if ( !m_pSchema )
		{
			assert (false && "dynamic table invoked without parent schema");
			return false;
		}

		ResetToIdle();
		return false;
	}

	// match constructing routines (empty for schema only)
	void PutFloatAsString ( float, const char * ) override {}
	void PutDoubleAsString ( double, const char * ) override {}
	void PutPercentAsString ( int64_t, int64_t ) override {}
	void PutNumAsString ( int64_t ) override {}
	void PutNumAsString ( uint64_t ) override {}
	void PutNumAsString ( int ) override {}
	void PutNumAsString ( DWORD ) override {}
	void PutFloat ( float fVal ) override {}
	void PutDouble ( double fVal ) override {}
	void PutInt ( int iVal ) override {}
	void PutInt64 ( int64_t iVal ) override {}
	void PutDWORD ( DWORD uVal ) override {}
	void PutUint64 ( uint64_t uVal ) override {}
	void PutArray ( const ByteBlob_t&, bool ) override {}
	void PutString ( Str_t ) override {}
	void PutMicrosec ( int64_t ) override {}
	void PutNULL () override {}
	bool Commit() override { return false;}
	void Eof ( bool, int, const char* ) override {}
	using RowBuffer_i::Eof;
};

GenericTableIndex_c::GenericTableIndex_c()
	: CSphIndexStub ( "dynamic", nullptr )
{}

bool GenericTableIndex_c::MultiQuery ( CSphQueryResult & tResult, const CSphQuery & tQuery,
		const VecTraits_T<ISphMatchSorter *> & dAllSorters, const CSphMultiQueryArgs &tArgs ) const
{
	MEMORY ( MEM_DISK_QUERY );

	// to avoid the checking of a ppSorters's element for NULL on every next step, just filter out all nulls right here
	CSphVector<ISphMatchSorter *> dSorters;
	dSorters.Reserve ( dAllSorters.GetLength() );
	dAllSorters.Apply ([&dSorters] ( ISphMatchSorter* p) { if ( p ) dSorters.Add(p); });

	// if we have anything to work with
	if ( dSorters.IsEmpty() )
		return false;

	// non-random at the start, random at the end
	dSorters.Sort ( CmpPSortersByRandom_fn () );

	const QueryParser_i * pQueryParser = tQuery.m_pQueryParser;
	assert ( pQueryParser );

	// fast path for scans
	if ( pQueryParser->IsFullscan ( tQuery ) )
		return MultiScan ( tResult, tQuery, dSorters, tArgs );

	return false;
}

class DynMatchProcessor_c : public MatchProcessor_i, ISphNoncopyable
{
public:
	DynMatchProcessor_c ( int iTag, const CSphQueryContext &tCtx )
		: m_iTag ( iTag )
		, m_tCtx ( tCtx )
	{}

	void Process ( CSphMatch * pMatch ) final			{ ProcessMatch(pMatch); }
	bool ProcessInRowIdOrder() const final				{ return false;	}
	void Process ( VecTraits_T<CSphMatch *> & dMatches ) final { dMatches.for_each ( [this]( CSphMatch * pMatch ){ ProcessMatch(pMatch); } ); }

private:
	int							m_iTag;
	const CSphQueryContext &	m_tCtx;

	inline void ProcessMatch ( CSphMatch * pMatch )
	{
		if ( pMatch->m_iTag>=0 )
			return;

		m_tCtx.CalcFinal ( *pMatch );
		pMatch->m_iTag = m_iTag;
	}
};


bool GenericTableIndex_c::MultiScan ( CSphQueryResult & tResult, const CSphQuery & tQuery, const VecTraits_T<ISphMatchSorter *> & dSorters, const CSphMultiQueryArgs & tArgs ) const
{
	assert ( tArgs.m_iTag>=0 );
	auto & tMeta = *tResult.m_pMeta;

	QueryProfile_c * pProfiler = tMeta.m_pProfile;

	if ( tArgs.m_uPackedFactorFlags & SPH_FACTOR_ENABLE )
		tMeta.m_sWarning.SetSprintf ( "packedfactors() will not work with a fullscan; you need to specify a query" );

	// start counting
	int64_t tmQueryStart = sphMicroTimer ();
	MiniTimer_c dTimerGuard;
	int64_t tmMaxTimer = dTimerGuard.Engage ( tQuery.m_uMaxQueryMsec ); // max_query_time

	// select the sorter with max schema
	// uses GetAttrsCount to get working facets (was GetRowSize)
	int iMaxSchemaIndex = GetMaxSchemaIndexAndMatchCapacity ( dSorters ).first;
	const ISphSchema & tMaxSorterSchema = *( dSorters[iMaxSchemaIndex]->GetSchema ());
	auto dSorterSchemas = SorterSchemas ( dSorters, iMaxSchemaIndex );

	// setup calculations and result schema
	CSphQueryContext tCtx ( tQuery );

	if ( !tCtx.SetupCalc ( tMeta, tMaxSorterSchema, m_tSchema, nullptr, nullptr, dSorterSchemas ) )
		return false;

	// setup filters
	CreateFilterContext_t tFlx;
	tFlx.m_pFilters = &tQuery.m_dFilters;
	tFlx.m_pFilterTree = &tQuery.m_dFilterTree;
	tFlx.m_pMatchSchema = &tMaxSorterSchema;
	tFlx.m_pIndexSchema = &m_tSchema;
	tFlx.m_eCollation = tQuery.m_eCollation;
	tFlx.m_bScan = true;

	if ( !tCtx.CreateFilters ( tFlx , tMeta.m_sError, tMeta.m_sWarning ) )
		return false;

	// prepare to work them rows
	bool bRandomize = dSorters[0]->IsRandom();

	CSphMatch tMatch;
	// note: we reserve dynamic area in match using max sorter schema, but then fill it by locators from index schema.
	// that works relying that sorter always includes all attrs from index, leaving final selection of cols
	// to result minimizer. Once we try to pre-optimize sorter schema by select list, it will cause crashes here.
	tMatch.Reset ( tMaxSorterSchema.GetDynamicSize () );
	tMatch.m_iWeight = tArgs.m_iIndexWeight;
	tMatch.m_iTag = tCtx.m_dCalcFinal.GetLength () ? -1 : tArgs.m_iTag;

	CSphScopedProfile tProf ( pProfiler, SPH_QSTATE_FULLSCAN );

	int iCutoff = ( tQuery.m_iCutoff<=0 ) ? -1 : tQuery.m_iCutoff;

	SetSorterStuff ( &tMatch );

	Threads::Coro::HighFreqChecker_c fnHeavyCheck;
	const int64_t& iCheckTimePoint { Threads::Coro::GetNextTimePointUS() };

	while ( FillNextMatch() )
	{
		++tMeta.m_tStats.m_iFetchedDocs;

		tCtx.CalcFilter ( tMatch );
		if ( tCtx.m_pFilter && !tCtx.m_pFilter->Eval ( tMatch ) )
		{
			tCtx.FreeDataFilter ( tMatch );
			m_tSchema.FreeDataPtrs ( tMatch );
			continue;
		}

		if ( bRandomize )
			tMatch.m_iWeight = ( sphRand () & 0xffff ) * tArgs.m_iIndexWeight;

		// submit match to sorters
		tCtx.CalcSort ( tMatch );

		bool bNewMatch = false;
		dSorters.Apply ( [&tMatch, &bNewMatch] ( ISphMatchSorter * p ) { bNewMatch |= p->Push ( tMatch ); } );

		// stringptr expressions should be duplicated (or taken over) at this point
		tCtx.FreeDataFilter ( tMatch );
		tCtx.FreeDataSort ( tMatch );
		m_tSchema.FreeDataPtrs ( tMatch );

		// handle cutoff
		if ( bNewMatch && --iCutoff==0 )
			break;

		// handle timer
		if ( sph::TimeExceeded ( tmMaxTimer ) )
		{
			tMeta.m_sWarning = "query time exceeded max_query_time";
			break;
		}

		if ( fnHeavyCheck() && sph::TimeExceeded ( iCheckTimePoint ) )
		{
			if ( session::GetKilled() )
			{
				tMeta.m_sWarning = "query was killed";
				break;
			}
			Threads::Coro::RescheduleAndKeepCrashQuery();
		}
	}

	auto sErrors = GetErrors();
	bool bOk = IsEmpty ( sErrors );
	if ( !bOk )
		tMeta.m_sError = sErrors;

	SwitchProfile ( pProfiler, SPH_QSTATE_FINALIZE );

	// do final expression calculations
	// even if no final stage calc to finilize doc collector
	if ( tCtx.m_dCalcFinal.GetLength() || m_bForceFinalize )
	{
		DynMatchProcessor_c tFinal ( tArgs.m_iTag, tCtx );
		dSorters.Apply ( [&] ( ISphMatchSorter * p ) { p->Finalize ( tFinal, false, tArgs.m_bFinalizeSorters ); } );
	}

	tMeta.AddQueryTimeUs ( sphMicroTimer() - tmQueryStart );

	return bOk;
}

///////////////
/// Index for data flow
class DynamicIndex_c final: public GenericTableIndex_c
{
	mutable Feeder_c		m_tFeeder;
	mutable bool m_bSchemaCreated = false;

public:
	explicit DynamicIndex_c ( TableFeeder_fn fnFeed)
		: m_tFeeder ( std::move ( fnFeed ) )
	{}

	const CSphSchema & GetMatchSchema () const final;

protected:
	~DynamicIndex_c() final = default;

private:
	void SetSorterStuff ( CSphMatch * pMatch ) const final;
	bool FillNextMatch () const final;
	Str_t GetErrors () const final;
};


const CSphSchema & DynamicIndex_c::GetMatchSchema () const
{
	if ( !m_bSchemaCreated )
	{
		m_tFeeder.SetSchema ( const_cast<CSphSchema *> (&m_tSchema) );
		m_bSchemaCreated = true;
	}
	return m_tSchema;
}

void DynamicIndex_c::SetSorterStuff ( CSphMatch * pMatch ) const
{
	assert ( m_bSchemaCreated );
	m_tFeeder.SetSorterStuff(pMatch);
}

bool DynamicIndex_c::FillNextMatch () const
{
	return m_tFeeder.FillNextMatch();
}

Str_t DynamicIndex_c::GetErrors () const
{
	return FromStr ( m_tFeeder.GetError() );
}

///////////////
/// Index for schema data flow
class DynamicIndexSchema_c final : public GenericTableIndex_c
{
	mutable FeederSchema_c		m_tFeeder;
	mutable bool m_bSchemaCreated = false;

public:
	explicit DynamicIndexSchema_c ( TableFeeder_fn fnFeed)
		: m_tFeeder ( std::move ( fnFeed ) )
	{}

	const CSphSchema & GetMatchSchema () const final;

protected:
	~DynamicIndexSchema_c() final = default;

private:
	void SetSorterStuff ( CSphMatch * pMatch ) const final;
	bool FillNextMatch () const final;
	Str_t GetErrors () const final;
};


const CSphSchema & DynamicIndexSchema_c::GetMatchSchema () const
{
	if ( !m_bSchemaCreated )
	{
		m_tFeeder.SetSchema ( const_cast<CSphSchema *> (&m_tSchema) );
		m_bSchemaCreated = true;
	}
	return m_tSchema;
}

void DynamicIndexSchema_c::SetSorterStuff ( CSphMatch * pMatch ) const
{
	assert ( m_bSchemaCreated );
	m_tFeeder.SetSorterStuff(pMatch);
}

bool DynamicIndexSchema_c::FillNextMatch () const
{
	return m_tFeeder.FillNextMatch();
}

Str_t DynamicIndexSchema_c::GetErrors () const
{
	return FromStr ( m_tFeeder.GetError() );
}

ServedIndexRefPtr_c MakeServed ( CSphIndex* pIndex )
{
	auto pServed = MakeServedIndex();
	pServed->SetIdx ( std::unique_ptr<CSphIndex> ( pIndex ) );
	return pServed;
}

/// external functions
ServedIndexRefPtr_c MakeDynamicIndex ( TableFeeder_fn fnFeed )
{
	return MakeServed ( new DynamicIndex_c ( std::move ( fnFeed ) ) );
}

ServedIndexRefPtr_c MakeDynamicIndexSchema ( TableFeeder_fn fnFeed )
{
	return MakeServed ( new DynamicIndexSchema_c ( std::move ( fnFeed ) ) );
}
