much faster bitmap scan with BitScanForward/__builtin_ctzll

secondary/blockreader.cpp

@@ -21,6 +21,7 @@
 #include "iterator.h"
 #include "delta.h"
 #include "interval.h"
+#include "bitvec.h"
 
 #include <functional>
 
@@ -66,6 +67,9 @@ class SplitBitmap_c
 	inline int	Scan ( int iStart );
 	inline int	GetLength() const { return m_iSize; }
 
+	template <typename RESULT>
+	void		Fetch ( int & iIterator, RESULT * & pRes, RESULT * pMax );
+
 private:
 	using BITMAP_TYPE = BitVec_T<uint64_t>;
 	const int	BITMAP_BITS = 16;
@@ -123,6 +127,41 @@ int SplitBitmap_c::Scan ( int iStart )
 	return iRes + iBitmapStart;
 }
 
+template <typename RESULT>
+void SplitBitmap_c::Fetch ( int & iIterator, RESULT * & pRes, RESULT * pMax )
+{
+	int iBitmap = iIterator >> ( BITMAP_BITS-6 );
+	if ( iBitmap>=m_dBitmaps.size() )
+		return;
+
+	auto & pBitmap = m_dBitmaps[iBitmap];
+	if ( !pBitmap )
+	{
+		iBitmap++;
+		while ( iBitmap<m_dBitmaps.size() && !m_dBitmaps[iBitmap] )
+			iBitmap++;
+
+		if ( iBitmap==m_dBitmaps.size() )
+			return;
+
+		// since this bitmap exists, its guaranteed to have set bits
+		iIterator = iBitmap << ( BITMAP_BITS-6 );
+		m_dBitmaps[iBitmap]->Fetch ( iIterator, pRes, pMax );
+	}
+	else
+	{
+		// we have a valid bitmap, but there's no guarantee we still have set bits
+		auto * pResStart = pRes;
+		iIterator = iBitmap << ( BITMAP_BITS-6 );
+		m_dBitmaps[iBitmap]->Fetch ( iIterator, pRes, pMax );
+		if ( pRes==pResStart )
+		{
+			iIterator += VALUES_PER_BITMAP>>6;
+			Fetch ( iIterator, pRes, pMax );
+		}
+	}
+}
+
 /////////////////////////////////////////////////////////////////////
 class BitmapIterator_i : public BlockIterator_i
 {
@@ -151,7 +190,6 @@ class BitmapIterator_T : public BitmapIterator_i
 	BITMAP						m_tBitmap;
 	std::vector<IteratorDesc_t> m_dDesc;
 	int64_t						m_iNumProcessed = 0;
-	bool						m_bFirst = true;
 	int							m_iIndex = 0;
 	int							m_iRowsLeft = INT_MAX;
 	SpanResizeable_T<uint32_t>	m_dRows;
@@ -195,32 +233,19 @@ void BitmapIterator_T<BITMAP>::Add ( BlockIterator_i * pIterator )
 template <typename BITMAP>
 bool BitmapIterator_T<BITMAP>::HintRowID ( uint32_t tRowID )
 {
-	m_bFirst = false;
-	m_iIndex = m_tBitmap.Scan(tRowID);
+	m_iIndex = tRowID >> 6;
 	return m_iIndex < m_tBitmap.GetLength();
 }
 
 template <typename BITMAP>
 bool BitmapIterator_T<BITMAP>::GetNextRowIdBlock ( Span_T<uint32_t> & dRowIdBlock )
 {
-	if ( m_bFirst )
-	{
-		m_iIndex = m_tBitmap.Scan(m_iIndex);
-		m_bFirst = false;
-	}
-
 	uint32_t * pData = m_dRows.data();
 	uint32_t * pPtr = pData;
-	uint32_t * pMax = m_dRows.end();
-
-	int iLen = m_tBitmap.GetLength();
-	while ( m_iIndex<iLen && pPtr<pMax )
-	{
-		*pPtr++= m_iIndex;
-		m_iIndex = m_tBitmap.Scan(m_iIndex+1);
-	}
-
+
+	m_tBitmap.Fetch ( m_iIndex, pPtr, m_dRows.end() );
 	dRowIdBlock = Span_T<uint32_t>( pData, pPtr-pData );
+
 	return !dRowIdBlock.empty();
 }
 

secondary/builder.cpp

@@ -19,6 +19,7 @@
 #include "codec.h"
 #include "delta.h"
 #include "pgm.h"
+#include "bitvec.h"
 
 #include <queue>
 

secondary/iterator.cpp

@@ -18,6 +18,7 @@
 #include "secondary.h"
 #include "delta.h"
 #include "interval.h"
+#include "bitvec.h"
 
 namespace SI
 {

secondary/secondary.cpp

@@ -20,6 +20,7 @@
 #include "delta.h"
 #include "codec.h"
 #include "blockreader.h"
+#include "bitvec.h"
 
 #include <unordered_map>
 

util/CMakeLists.txt

@@ -27,6 +27,7 @@ add_library ( util OBJECT
 		delta.h
 		reader.h
 		codec.h
+		bitvec.h
 		)
 
 include ( CheckFunctionExists )

util/bitvec.h

@@ -0,0 +1,140 @@
+// Copyright (c) 2020-2022, Manticore Software LTD (https://manticoresearch.com)
+// All rights reserved
+//
+//
+// 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.
+
+#pragma once
+
+#ifdef _MSC_VER
+	#include <intrin.h>
+#endif
+
+namespace util
+{
+
+template <typename T=uint32_t>
+class BitVec_T
+{
+public:
+	explicit BitVec_T ( int iSize ) { Resize(iSize); }
+
+	inline bool BitGet ( int iBit )
+	{
+		if ( !m_dData.size() )
+			return false;
+
+		assert ( iBit>=0 && iBit<m_iSize );
+		return ( ( m_dData [ iBit>>SHIFT ] & ( ( (T)1 )<<( iBit&MASK ) ) )!=0 );
+	}
+
+	inline void BitSet ( int iBit )
+	{
+		if ( !m_dData.size() )
+			return;
+
+		assert ( iBit>=0 && iBit<m_iSize );
+		m_dData [ iBit>>SHIFT ] |= ( ( (T)1 )<<( iBit&MASK ) );
+	}
+
+	template <typename RESULT>
+	void Fetch ( int & iIterator, RESULT * & pRes, RESULT * pMax )
+	{
+		assert ( sizeof(T)==8 );	// this func should not be used with 32-bit based bitmaps
+
+		const T * pDataStart = &m_dData.front();
+		const T * pData = pDataStart + iIterator;
+		const T * pDataMax = pDataStart + m_dData.size();
+
+		pMax -= SIZEBITS;
+		assert ( pMax>=pRes );
+
+		RESULT tOutStart = iIterator << SHIFT;
+		for ( ; pRes<pMax && pData<pDataMax; pData++ )
+		{
+			uint64_t tVal = *pData;
+			while ( tVal )
+			{
+				unsigned long uIdx;
+#ifdef _MSC_VER
+				::_BitScanForward64 ( &uIdx, tVal );
+#else
+				uIdx = __builtin_ctzll(tVal);
+#endif
+				*pRes++ = uIdx + tOutStart;
+
+				tVal ^= tVal & (~tVal + 1);
+			}
+
+			tOutStart += 64;
+		}
+
+		iIterator = pData - pDataStart;
+	}
+
+	int Scan ( int iStart )
+	{
+		if ( iStart>=m_iSize )
+			return m_iSize;
+
+		const T * pData = &m_dData.front();
+		int iIndex = iStart>>SHIFT;
+		T uMask = ~( ( T(1)<<( iStart&MASK ) )-1 );
+		if ( pData[iIndex] & uMask )
+			return (iIndex<<SHIFT) + ScanBit ( pData[iIndex], iStart&MASK );
+
+		iIndex++;
+		while ( iIndex<(int)m_dData.size() && !pData[iIndex] )
+			iIndex++;
+
+		if ( iIndex>=(int)m_dData.size() )
+			return m_iSize;
+
+		return (iIndex<<SHIFT) + ScanBit ( pData[iIndex], 0 );
+	}
+
+	void SetAllBits() { std::fill ( m_dData.begin(), m_dData.end(), (T)0xffffffffffffffffULL ); }
+	void Resize ( int iSize )
+	{
+		m_iSize = iSize;
+		if ( iSize )
+		{
+			int iCount = ( iSize+SIZEBITS-1 )/SIZEBITS;
+			m_dData = std::vector<T> ( iCount, 0 );
+		}
+	}
+
+	int	GetLength() const { return m_iSize; }
+	const std::vector<T> & GetData() const { return m_dData; }
+
+private:
+	static const size_t	SIZEBITS = sizeof(T)*8;
+	static const T		MASK = T(sizeof(T)*8 - 1);
+	static constexpr T	SHIFT = T(Log2(SIZEBITS)-1);
+
+	std::vector<T>	m_dData;
+	int				m_iSize = 0;
+
+	inline int ScanBit ( T tData, int iStart )
+	{
+		for ( int i = iStart; i < SIZEBITS; i++ )
+			if ( tData & ( (T)1<<i ) )
+				return i;
+
+		return -1;
+	}
+};
+
+using BitVec_c = BitVec_T<>;
+
+} // namespace util

util/util.h

@@ -448,83 +448,4 @@ constexpr int Log2 ( T tValue )
 	return iBits;
 }
 
-template <typename T=uint32_t>
-class BitVec_T
-{
-public:
-	explicit BitVec_T ( int iSize ) { Resize(iSize); }
-
-	inline bool BitGet ( int iBit )
-	{
-		if ( !m_dData.size() )
-			return false;
-
-		assert ( iBit>=0 && iBit<m_iSize );
-		return ( ( m_dData [ iBit>>SHIFT ] & ( ( (T)1 )<<( iBit&MASK ) ) )!=0 );
-	}
-
-	inline void BitSet ( int iBit )
-	{
-		if ( !m_dData.size() )
-			return;
-
-		assert ( iBit>=0 && iBit<m_iSize );
-		m_dData [ iBit>>SHIFT ] |= ( ( (T)1 )<<( iBit&MASK ) );
-	}
-
-	int Scan ( int iStart )
-	{
-		if ( iStart>=m_iSize )
-			return m_iSize;
-
-		const T * pData = &m_dData.front();
-		int iIndex = iStart>>SHIFT;
-		T uMask = ~( ( T(1)<<( iStart&MASK ) )-1 );
-		if ( pData[iIndex] & uMask )
-			return (iIndex<<SHIFT) + ScanBit ( pData[iIndex], iStart&MASK );
-
-		iIndex++;
-		while ( iIndex<(int)m_dData.size() && !pData[iIndex] )
-			iIndex++;
-
-		if ( iIndex>=(int)m_dData.size() )
-			return m_iSize;
-
-		return (iIndex<<SHIFT) + ScanBit ( pData[iIndex], 0 );
-	}
-
-	void SetAllBits() { std::fill ( m_dData.begin(), m_dData.end(), (T)0xffffffffffffffffULL ); }
-	void Resize ( int iSize )
-	{
-		m_iSize = iSize;
-		if ( iSize )
-		{
-			int iCount = ( iSize+SIZEBITS-1 )/SIZEBITS;
-			m_dData = std::vector<T> ( iCount, 0 );
-		}
-	}
-
-	int	GetLength() const { return m_iSize; }
-	const std::vector<T> & GetData() const { return m_dData; }
-
-private:
-	static const size_t	SIZEBITS = sizeof(T)*8;
-	static const T		MASK = T(sizeof(T)*8 - 1);
-	static constexpr T	SHIFT = T(Log2(SIZEBITS)-1);
-
-	std::vector<T>	m_dData;
-	int				m_iSize = 0;
-
-	inline int ScanBit ( T tData, int iStart )
-	{
-		for ( int i = iStart; i < SIZEBITS; i++ )
-			if ( tData & ( (T)1<<i ) )
-				return i;
-
-		return -1;
-	}
-};
-
-using BitVec_c = BitVec_T<>;
-
 } // namespace util