From 2be1bbcc111cb271d66d590ac66090d5debeddef Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Caol=C3=A1n=20McNamara?= Date: Tue, 18 Aug 2026 16:19:49 +0000 Subject: [PATCH] dbase: apply one consistent key-per-page limit in the .ndx index MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit db_maxkeys came from the index header, the key count of a page came from the page itself, and node access used neither. Now the header value is capped to the number of keys a DINDEX_PAGE_SIZE page can hold for the given key length, each page keeps its own capacity and caps its key count to it, and node positions are clamped to the same bound, so all three agree. ODbaseIndex::CreateImpl worked out db_maxkeys from a four byte page header while the reader counts eight bytes of page header, so use eight here too and write an index the reader agrees with. ONDXPage::Remove asserted on a position past the key count and then carried on regardless, so return early instead. ONDXPage::Merge added the key counts of two pages together in a sal_uInt16, so widen the sum to sal_uInt32. nValue was a file-level static shared by the page read and write operators for no reason, so make it a local in the reader and write rPage.nCount directly. The two GetChild overloads would load a child page that is already on the chain of parents the descent came down, so leave such a page unread rather than descending into a page we are inside. ONDXPage::Merge and ONDXPage::PrintPage can then be handed a page that was left unread, so both check for one with nothing loaded. Signed-off-by: Caolán McNamara Change-Id: Id51bf8368099447df41aaacbaebba7783c5769e2 Reviewed-on: https://gerrit.collaboraoffice.com/c/online/+/9518 Reviewed-by: Noel Grandin Tested-by: Jenkins CPCI (cherry picked from commit db92d0be756305ec0b1e122a1d00333477c17ba5) Reviewed-on: https://gerrit.libreoffice.org/c/core/+/209449 Tested-by: Xisco Fauli Reviewed-by: Xisco Fauli --- connectivity/source/drivers/dbase/DIndex.cxx | 9 ++- .../source/drivers/dbase/dindexnode.cxx | 76 ++++++++++++++----- connectivity/source/inc/dbase/dindexnode.hxx | 3 + 3 files changed, 70 insertions(+), 18 deletions(-) diff --git a/connectivity/source/drivers/dbase/DIndex.cxx b/connectivity/source/drivers/dbase/DIndex.cxx index 35a4087c0482..2c64b51b01e9 100644 --- a/connectivity/source/drivers/dbase/DIndex.cxx +++ b/connectivity/source/drivers/dbase/DIndex.cxx @@ -34,6 +34,8 @@ #include #include +#include + using namespace ::comphelper; using namespace connectivity; @@ -324,6 +326,11 @@ void connectivity::dbase::ReadHeader( rStream.ReadUChar(rHeader.db_unique); rStream.ReadBytes(&rHeader.db_name, 488); assert(rStream.GetError() || rStream.Tell() == nOldPos + DINDEX_PAGE_SIZE); + + // a page starts with a four byte key count and a four byte child page id, then a record id, a + // key and a child page id per key + const sal_uInt16 nKeysThatFitAPage = (DINDEX_PAGE_SIZE - 8) / (8 + rHeader.db_keylen); + rHeader.db_maxkeys = std::max(1, std::min(rHeader.db_maxkeys, nKeysThatFitAPage)); } SvStream& connectivity::dbase::operator >> (SvStream &rStream, ODbaseIndex& rIndex) @@ -518,7 +525,7 @@ void ODbaseIndex::CreateImpl() m_aHeader.db_keytype = (nType == DataType::VARCHAR || nType == DataType::CHAR) ? 0 : 1; m_aHeader.db_keylen = (m_aHeader.db_keytype) ? 8 : static_cast(getINT32(xTableCol->getPropertyValue(OMetaConnection::getPropMap().getNameByIndex(PROPERTY_ID_PRECISION)))); m_aHeader.db_keylen = (( m_aHeader.db_keylen - 1) / 4 + 1) * 4; - m_aHeader.db_maxkeys = (DINDEX_PAGE_SIZE - 4) / (8 + m_aHeader.db_keylen); + m_aHeader.db_maxkeys = (DINDEX_PAGE_SIZE - 8) / (8 + m_aHeader.db_keylen); if ( m_aHeader.db_maxkeys < 3 ) { impl_killFileAndthrowError_throw(STR_COULD_NOT_CREATE_INDEX_KEYSIZE,sFile); diff --git a/connectivity/source/drivers/dbase/dindexnode.cxx b/connectivity/source/drivers/dbase/dindexnode.cxx index 1b070b1aace1..5692475a3ff9 100644 --- a/connectivity/source/drivers/dbase/dindexnode.cxx +++ b/connectivity/source/drivers/dbase/dindexnode.cxx @@ -74,8 +74,8 @@ ONDXPage::ONDXPage(ODbaseIndex& rInd, sal_uInt32 nPos, ONDXPage* pParent) , aParent(pParent) , rIndex(rInd) { - sal_uInt16 nT = rIndex.getHeader().db_maxkeys; - ppNodes.reset( new ONDXNode[nT] ); + nNodeCapacity = std::max(1, rIndex.getHeader().db_maxkeys); + ppNodes.reset( new ONDXNode[nNodeCapacity] ); } ONDXPage::~ONDXPage() @@ -103,7 +103,7 @@ void ONDXPage::QueryDelete() if (aChild.Is()) aChild->Release(false); - for (sal_uInt16 i = 0; i < rIndex.getHeader().db_maxkeys;i++) + for (sal_uInt16 i = 0; i < nNodeCapacity;i++) { if (ppNodes[i].GetChild().Is()) ppNodes[i].GetChild()->Release(false); @@ -126,11 +126,30 @@ void ONDXPage::QueryDelete() } } +// true when the page is already somewhere on the chain of parents, so that loading it would +// descend into a page the descent is already inside +static bool lcl_isOnTheParentChain(sal_uInt32 nPagePos, const ONDXPage* pPage) +{ + for (const ONDXPage* p = pPage; p; p = p->GetParent()) + { + if (p->GetPagePos() == nPagePos) + return true; + } + return false; +} + ONDXPagePtr& ONDXPage::GetChild(ODbaseIndex const * pIndex) { if (!aChild.Is() && pIndex) { - aChild = rIndex.CreatePage(aChild.GetPagePos(),this,aChild.HasPage()); + const bool bLoad = aChild.HasPage(); + if (bLoad && lcl_isOnTheParentChain(aChild.GetPagePos(), this)) + { + SAL_WARN("connectivity.dbase", "page " << aChild.GetPagePos() + << " is its own ancestor, so it is left unread"); + } + else + aChild = rIndex.CreatePage(aChild.GetPagePos(),this,bLoad); } return aChild; } @@ -307,7 +326,7 @@ bool ONDXPage::Insert(ONDXNode& rNode, sal_uInt32 nRowsLeft) bool ONDXPage::Insert(sal_uInt16 nPos, ONDXNode& rNode) { - sal_uInt16 nMaxCount = rIndex.getHeader().db_maxkeys; + sal_uInt16 nMaxCount = nNodeCapacity; if (nPos >= nMaxCount) return false; @@ -352,7 +371,7 @@ void ONDXPage::Release(bool bSave) // free pointer aChild.Clear(); - for (sal_uInt16 i = 0; i < rIndex.getHeader().db_maxkeys;i++) + for (sal_uInt16 i = 0; i < nNodeCapacity;i++) { if (ppNodes[i].GetChild()) ppNodes[i].GetChild()->Release(bSave); @@ -482,6 +501,8 @@ void ONDXPage::Merge(sal_uInt16 nParentNodePos, const ONDXPagePtr& xPage) { DBG_ASSERT(HasParent(), "no parent existing"); DBG_ASSERT(nParentNodePos != NODE_NOTFOUND, "Wrong index setup"); + if (!xPage.Is()) + return; /* Merge 2 pages */ sal_uInt16 nMaxNodes = rIndex.GetMaxNodes(), @@ -489,7 +510,7 @@ void ONDXPage::Merge(sal_uInt16 nParentNodePos, const ONDXPagePtr& xPage) // Determine if page is right or left neighbour bool bRight = ((*xPage)[0].GetKey() > (*this)[0].GetKey()); // true when xPage is at the right side - sal_uInt16 nNewCount = (*xPage).Count() + Count(); + sal_uInt32 nNewCount = (*xPage).Count() + Count(); if (IsLeaf()) { @@ -731,7 +752,14 @@ ONDXPagePtr& ONDXNode::GetChild(ODbaseIndex* pIndex, ONDXPage* pParent) { if (!aChild.Is() && pIndex) { - aChild = pIndex->CreatePage(aChild.GetPagePos(),pParent,aChild.HasPage()); + const bool bLoad = aChild.HasPage(); + if (bLoad && lcl_isOnTheParentChain(aChild.GetPagePos(), pParent)) + { + SAL_WARN("connectivity.dbase", "page " << aChild.GetPagePos() + << " is its own ancestor, so it is left unread"); + } + else + aChild = pIndex->CreatePage(aChild.GetPagePos(),pParent,bLoad); } return aChild; } @@ -870,13 +898,13 @@ ONDXPagePtr& ONDXPagePtr::operator=(ONDXPagePtr && rOther) return *this; } -static sal_uInt32 nValue; - SvStream& connectivity::dbase::operator >> (SvStream &rStream, ONDXPage& rPage) { rStream.Seek(rPage.GetPagePos() * DINDEX_PAGE_SIZE); + sal_uInt32 nValue(0); rStream.ReadUInt32( nValue ) >> rPage.aChild; - rPage.nCount = sal_uInt16(nValue); + // a page holds at most as many keys as the node array has entries + rPage.nCount = std::min(sal_uInt16(nValue), rPage.GetIndex().getHeader().db_maxkeys); for (sal_uInt16 i = 0; i < rPage.nCount; i++) rPage[i].Read(rStream, rPage.GetIndex()); @@ -899,8 +927,7 @@ SvStream& connectivity::dbase::WriteONDXPage(SvStream &rStream, const ONDXPage& } rStream.Seek(rPage.GetPagePos() * DINDEX_PAGE_SIZE); - nValue = rPage.nCount; - rStream.WriteUInt32( nValue ); + rStream.WriteUInt32( rPage.nCount ); WriteONDXPagePtr( rStream, rPage.aChild ); sal_uInt16 i = 0; @@ -957,11 +984,15 @@ void ONDXPage::PrintPage() if (!IsLeaf()) { #if OSL_DEBUG_LEVEL > 1 - GetChild(&rIndex)->PrintPage(); + ONDXPagePtr aChildPage = GetChild(&rIndex); + if (aChildPage.Is()) + aChildPage->PrintPage(); for (sal_uInt16 i = 0; i < nCount; i++) { ONDXNode rNode = (*this)[i]; - rNode.GetChild(&rIndex,this)->PrintPage(); + ONDXPagePtr aNodePage = rNode.GetChild(&rIndex,this); + if (aNodePage.Is()) + aNodePage->PrintPage(); } #endif } @@ -1024,22 +1055,33 @@ void ONDXPage::SearchAndReplace(const ONDXKey& rSearch, } } +// a node position inside the node array +sal_uInt16 ONDXPage::ClampPos(sal_uInt16 nPos) const +{ + const sal_uInt16 nLast = nNodeCapacity - 1; + SAL_WARN_IF(nPos > nLast, "connectivity.dbase", "node " << nPos + << " asked for on a page whose last node is " << nLast); + return std::min(nPos, nLast); +} + ONDXNode& ONDXPage::operator[] (sal_uInt16 nPos) { DBG_ASSERT(nCount > nPos, "incorrect index access"); - return ppNodes[nPos]; + return ppNodes[ClampPos(nPos)]; } const ONDXNode& ONDXPage::operator[] (sal_uInt16 nPos) const { DBG_ASSERT(nCount > nPos, "incorrect index access"); - return ppNodes[nPos]; + return ppNodes[ClampPos(nPos)]; } void ONDXPage::Remove(sal_uInt16 nPos) { DBG_ASSERT(nCount > nPos, "incorrect index access"); + if (nPos >= nCount) + return; for (sal_uInt16 i = nPos; i < (nCount-1); i++) (*this)[i] = (*this)[i+1]; diff --git a/connectivity/source/inc/dbase/dindexnode.hxx b/connectivity/source/inc/dbase/dindexnode.hxx index 181c6ff74322..f5280c2a76a0 100644 --- a/connectivity/source/inc/dbase/dindexnode.hxx +++ b/connectivity/source/inc/dbase/dindexnode.hxx @@ -120,6 +120,7 @@ namespace connectivity::dbase sal_uInt32 nPagePos; // Position in the index file bool bModified : 1; sal_uInt16 nCount; + sal_uInt16 nNodeCapacity; // Number of entries allocated in ppNodes, one or more ONDXPagePtr aParent, // Parent page aChild; // Pointer to the right child page @@ -127,6 +128,8 @@ namespace connectivity::dbase std::unique_ptr ppNodes; // Array of nodes + sal_uInt16 ClampPos(sal_uInt16 nPos) const; + public: // Node operations sal_uInt16 Count() const {return nCount;}