improved support for multiple statement execution and recordset paging

Data/src/RecordSet.cpp

@@ -51,13 +51,19 @@ namespace Poco {
 namespace Data {
 
 
-RecordSet::RecordSet(const Statement& rStatement): 
+const std::size_t RecordSet::UNKNOWN_TOTAL_ROW_COUNT = std::numeric_limits<std::size_t>::max();
+
+
+RecordSet::RecordSet(const Statement& rStatement,
+	RowFormatter* pRowFormatter): 
 	Statement(rStatement),
 	_currentRow(0),
 	_pBegin(new RowIterator(this, 0 == rowsExtracted())),
 	_pEnd(new RowIterator(this, true)),
-	_pFilter(0)
+	_pFilter(0),
+	_totalRowCount(UNKNOWN_TOTAL_ROW_COUNT)
 {
+	if (pRowFormatter) setRowFormatter(pRowFormatter);
 }
 
 
@@ -68,7 +74,8 @@ RecordSet::RecordSet(Session& rSession,
 	_currentRow(0),
 	_pBegin(new RowIterator(this, 0 == rowsExtracted())),
 	_pEnd(new RowIterator(this, true)),
-	_pFilter(0)
+	_pFilter(0),
+	_totalRowCount(UNKNOWN_TOTAL_ROW_COUNT)
 {
 	if (pRowFormatter) setRowFormatter(pRowFormatter);
 }
@@ -79,7 +86,8 @@ RecordSet::RecordSet(const RecordSet& other):
 	_currentRow(other._currentRow),
 	_pBegin(new RowIterator(this, 0 == rowsExtracted())),
 	_pEnd(new RowIterator(this, true)),
-	_pFilter(other._pFilter)
+	_pFilter(other._pFilter),
+	_totalRowCount(other._totalRowCount)
 {
 }
 
@@ -203,7 +211,7 @@ Row& RecordSet::row(std::size_t pos)
 std::size_t RecordSet::rowCount() const
 {
 	poco_assert (extractions().size());
-	std::size_t rc = totalRowCount();
+	std::size_t rc = subTotalRowCount();
 	if (!isFiltered()) return rc;
 
 	std::size_t counter = 0;
@@ -225,7 +233,7 @@ bool RecordSet::isAllowed(std::size_t row) const
 
 bool RecordSet::moveFirst()
 {
-	if (totalRowCount() > 0)
+	if (subTotalRowCount() > 0)
 	{
 		if (!isFiltered())
 		{
@@ -237,7 +245,7 @@ bool RecordSet::moveFirst()
 		currentRow = 0;
 		while (!isAllowed(currentRow))
 		{
-			if (currentRow >= totalRowCount() - 1) return false;
+			if (currentRow >= subTotalRowCount() - 1) return false;
 			++currentRow;
 		}
 
@@ -253,7 +261,7 @@ bool RecordSet::moveNext()
 	std::size_t currentRow = _currentRow;
 	do
 	{
-		if (currentRow >= totalRowCount() - 1) return false;
+		if (currentRow >= subTotalRowCount() - 1) return false;
 		++currentRow;
 	} while (isFiltered() && !isAllowed(currentRow));
 
@@ -278,10 +286,10 @@ bool RecordSet::movePrevious()
 
 bool RecordSet::moveLast()
 {
-	if (totalRowCount() > 0)
+	if (subTotalRowCount() > 0)
 	{
 		std::size_t currentRow = _currentRow;
-		currentRow = totalRowCount() - 1;
+		currentRow = subTotalRowCount() - 1;
 		if (!isFiltered())
 		{
 			_currentRow = currentRow;
@@ -303,26 +311,21 @@ bool RecordSet::moveLast()
 
 std::ostream& RecordSet::copyValues(std::ostream& os, std::size_t offset, std::size_t length) const
 {
-	if (length == RowIterator::POSITION_END) 
-	{
-		if (0 != offset) throw RangeException("Invalid range.");
-		length = rowCount();
-	}
-
 	RowIterator itBegin = *_pBegin + offset;
-	RowIterator itEnd = itBegin + length;
+	RowIterator itEnd = (RowIterator::POSITION_END != length) ? itBegin + length : *_pEnd;
 	std::copy(itBegin, itEnd, std::ostream_iterator<Row>(os));
 	return os;
 }
 
 
-std::ostream& RecordSet::copy(std::ostream& os) const
+std::ostream& RecordSet::copy(std::ostream& os, std::size_t offset, std::size_t length) const
 {
-	const RowFormatter& ri = (*_pBegin)->getFormatter();
-	os << ri.prefix();
+	RowFormatter& rf = const_cast<RowFormatter&>((*_pBegin)->getFormatter());
+	rf.setTotalRowCount(getTotalRowCount());
+	os << rf.prefix();
 	copyNames(os);
-	copyValues(os);
-	os << ri.postfix();
+	copyValues(os, offset, length);
+	os << rf.postfix();
 	return os;
 }
 
@@ -341,4 +344,10 @@ bool RecordSet::isFiltered() const
 }
 
 
+void RecordSet::setTotalRowCount(const std::string& sql)
+{
+	session() << sql, into(_totalRowCount), now;
+}
+
+
 } } // namespace Poco::Data

Data/src/RowFormatter.cpp

@@ -45,7 +45,8 @@ namespace Data {
 
 RowFormatter::RowFormatter(const std::string& prefix, const std::string& postfix):
 	_prefix(prefix), 
-	_postfix(postfix)
+	_postfix(postfix),
+	_totalRowCount(0)
 {
 }
 
@@ -55,4 +56,12 @@ RowFormatter::~RowFormatter()
 }
 
 
+void RowFormatter::reset()
+{
+	_prefix = "";
+	_postfix = "";
+	_totalRowCount = INVALID_ROW_COUNT;
+}
+
+
 } } // namespace Poco::Data

Data/src/RowIterator.cpp

@@ -89,7 +89,7 @@ void RowIterator::increment() const
 	if (POSITION_END == _position)
 		throw RangeException("End of iterator reached.");
 
-	if (_position < _pRecordSet->totalRowCount() - 1)
+	if (_position < _pRecordSet->subTotalRowCount() - 1)
 		++_position;
 	else
 		_position = POSITION_END;
@@ -110,7 +110,7 @@ void RowIterator::decrement() const
 	if (0 == _position)
 		throw RangeException("Beginning of iterator reached.");
 	else if (POSITION_END == _position)
-		_position = _pRecordSet->totalRowCount() - 1;
+		_position = _pRecordSet->subTotalRowCount() - 1;
 	else
 		--_position;
 
@@ -146,16 +146,15 @@ void RowIterator::setPosition(std::size_t pos) const
 			std::size_t end = pos - _position;
 			for (; start < end; ++start)
 			{
-				if (_pRecordSet->totalRowCount() != pos) ++pos;
+				if (_pRecordSet->subTotalRowCount() != pos) ++pos;
 				else throw RangeException("Invalid position argument.");
 			}
 		}
-			
 	}
 
-	if (pos < _pRecordSet->totalRowCount())
+	if (pos < _pRecordSet->subTotalRowCount())
 		_position = pos;
-	else if (pos == _pRecordSet->totalRowCount())
+	else if (pos == _pRecordSet->subTotalRowCount())
 		_position = POSITION_END;
 	else
 		throw RangeException("Invalid position argument.");

Data/src/SimpleRowFormatter.cpp

@@ -44,7 +44,7 @@ namespace Data {
 
 
 SimpleRowFormatter::SimpleRowFormatter(std::streamsize columnWidth):
-	_colWidth(columnWidth)
+	_colWidth(columnWidth), _rowCount(0)
 {
 }
 
@@ -81,6 +81,8 @@ void SimpleRowFormatter::swap(SimpleRowFormatter& other)
 
 std::string& SimpleRowFormatter::formatNames(const NameVecPtr pNames, std::string& formattedNames) const
 {
+	_rowCount = 0;
+
 	std::ostringstream str;
 	std::string line(_colWidth * pNames->size(), '-');
 
@@ -112,10 +114,15 @@ std::string& SimpleRowFormatter::formatValues(const ValueVec& vals, std::string&
 		}
 		else str << std::left;
 
-		str << std::setw(_colWidth) << it->convert<std::string>();
+		if (!it->isEmpty())
+			str << std::setw(_colWidth) << it->convert<std::string>();
+		else
+			str << std::setw(_colWidth) << "null";
 	}
 	str << std::endl;
 
+	++_rowCount;
+
 	return formattedValues = str.str();
 }
 

Data/src/Statement.cpp

@@ -109,7 +109,7 @@ Statement& Statement::reset(Session& session)
 }
 
 
-std::size_t Statement::execute()
+std::size_t Statement::execute(bool reset)
 {
 	Mutex::ScopedLock lock(_mutex);
 	bool isDone = done();
@@ -124,7 +124,7 @@ std::size_t Statement::execute()
 		if (!isAsync())
 		{
 			if (isDone) _pImpl->reset();
-			return _pImpl->execute();
+			return _pImpl->execute(reset);
 		}
 		else
 		{
@@ -136,22 +136,22 @@ std::size_t Statement::execute()
 }
 
 
-const Statement::Result& Statement::executeAsync()
+const Statement::Result& Statement::executeAsync(bool reset)
 {
 	Mutex::ScopedLock lock(_mutex);
 	if (initialized() || paused() || done())
-		return doAsyncExec();
+		return doAsyncExec(reset);
 	else
 		throw InvalidAccessException("Statement still executing.");
 }
 
 
-const Statement::Result& Statement::doAsyncExec()
+const Statement::Result& Statement::doAsyncExec(bool reset)
 {
 	if (done()) _pImpl->reset();
 	if (!_pAsyncExec)
 		_pAsyncExec = new AsyncExecMethod(_pImpl, &StatementImpl::execute);
-	_pResult = new Result((*_pAsyncExec)());
+	_pResult = new Result((*_pAsyncExec)(reset));
 	return *_pResult;
 }
 
@@ -308,4 +308,11 @@ Statement& Statement::operator , (BulkFnType)
 }
 
 
+Session Statement::session()
+{
+	Poco::AutoPtr<SessionImpl> ps(&impl()->session(), true); 
+	return Session(ps);
+}
+
+
 } } // namespace Poco::Data

Data/src/StatementImpl.cpp

@@ -80,6 +80,7 @@ StatementImpl::StatementImpl(SessionImpl& rSession):
 
 	_extractors.resize(1);
 	_columnsExtracted.resize(1, 0);
+	_subTotalRowCount.resize(1, 0);
 }
 
 
@@ -88,9 +89,9 @@ StatementImpl::~StatementImpl()
 }
 
 
-std::size_t StatementImpl::execute()
+std::size_t StatementImpl::execute(const bool& reset)
 {
-	resetExtraction();
+	if (reset) resetExtraction();
 
 	if (!_rSession.isConnected())
 	{
@@ -117,10 +118,32 @@ std::size_t StatementImpl::execute()
 	if (lim < _lowerLimit)
 		throw LimitException("Did not receive enough data.");
 
+	assignSubTotal(reset);
+
 	return lim;
 }
 
 
+void StatementImpl::assignSubTotal(bool reset)
+{
+	if (_extractors.size() == _subTotalRowCount.size())
+	{
+		CountVec::iterator it  = _subTotalRowCount.begin();
+		CountVec::iterator end = _subTotalRowCount.end();
+		for (int counter = 0; it != end; ++it, ++counter)
+		{
+			if (_extractors[counter].size())
+			{
+				if (reset)
+					*it += _extractors[counter][0]->numOfRowsHandled();
+				else
+					*it = _extractors[counter][0]->numOfRowsHandled();
+			}
+		}
+	}
+}
+
+
 std::size_t StatementImpl::executeWithLimit()
 {
 	poco_assert (_state != ST_DONE);
@@ -229,13 +252,19 @@ void StatementImpl::setBulkExtraction(const Bulk& b)
 
 void StatementImpl::fixupExtraction()
 {
+	CountVec::iterator sIt  = _subTotalRowCount.begin();
+	CountVec::iterator sEnd = _subTotalRowCount.end();
+	for (; sIt != sEnd; ++sIt) *sIt = 0;
+
+	if (_curDataSet >= _columnsExtracted.size())
+	{
+		_columnsExtracted.resize(_curDataSet + 1, 0);
+		_subTotalRowCount.resize(_curDataSet + 1, 0);
+	}
+
 	Poco::Data::AbstractExtractionVec::iterator it    = extractions().begin();
 	Poco::Data::AbstractExtractionVec::iterator itEnd = extractions().end();
 	AbstractExtractor& ex = extractor();
-	
-	if (_curDataSet >= _columnsExtracted.size()) 
-		_columnsExtracted.resize(_curDataSet + 1, 0);
-	
 	for (; it != itEnd; ++it)
 	{
 		(*it)->setExtractor(&ex);
@@ -432,6 +461,19 @@ std::size_t StatementImpl::rowsExtracted(int dataSet) const
 }
 
 
+std::size_t StatementImpl::subTotalRowCount(int dataSet) const
+{
+	if (USE_CURRENT_DATA_SET == dataSet) dataSet = _curDataSet;
+	if (_subTotalRowCount.size() > 0)
+	{
+		poco_assert (dataSet >= 0 && dataSet < _subTotalRowCount.size());
+		return _subTotalRowCount[dataSet];
+	}
+	
+	return 0;
+}
+
+
 void StatementImpl::formatSQL(std::vector<Any>& arguments)
 {
 	std::string sql;