From b6424480c077411f8045b6374334fb196871d43b Mon Sep 17 00:00:00 2001
From: Andrew Hutchings <andrew@linuxjedi.co.uk>
Date: Tue, 29 May 2018 21:21:38 +0100
Subject: [PATCH] MCOL-1408 Fix HWM calculation for DML & API

HWM for DML and API was being calculated using the first column in a
table instead of the smallest column. This shifts the calculation to the
correct column.
---
 writeengine/wrapper/we_colop.cpp    |  8 +++----
 writeengine/wrapper/writeengine.cpp | 36 ++++++++++++++---------------
 2 files changed, 22 insertions(+), 22 deletions(-)

diff --git a/writeengine/wrapper/we_colop.cpp b/writeengine/wrapper/we_colop.cpp
index ac13a7bf0..fdaceef3e 100644
--- a/writeengine/wrapper/we_colop.cpp
+++ b/writeengine/wrapper/we_colop.cpp
@@ -208,7 +208,7 @@ int ColumnOp::allocRowId(const TxnID& txnid, bool useStartingExtent,
 				//Find out where the rest rows go
 				BRM::LBID_t startLbid;
 				//need to put in a loop until newExtent is true
-				newExtent = dbRootExtentTrackers[0]->nextSegFile(dbRoot, partition, segment, newHwm, startLbid);
+				newExtent = dbRootExtentTrackers[column.colNo]->nextSegFile(dbRoot, partition, segment, newHwm, startLbid);
 				TableMetaData* tableMetaData= TableMetaData::makeTableMetaData(tableOid);
 				while (!newExtent)
 				{
@@ -223,7 +223,7 @@ int ColumnOp::allocRowId(const TxnID& txnid, bool useStartingExtent,
 					
 					for (i=0; i < dbRootExtentTrackers.size(); i++)
 					{
-						if (i != 0)
+						if (i != column.colNo)
 							dbRootExtentTrackers[i]->nextSegFile(dbRoot, partition, segment, newHwm, startLbid);
 						
 						// Round up HWM to the end of the current extent
@@ -278,7 +278,7 @@ int ColumnOp::allocRowId(const TxnID& txnid, bool useStartingExtent,
 						}
 						tableMetaData->setColExtsInfo(newColStructList[i].dataOid, aColExtsInfo);
 					}
-					newExtent = dbRootExtentTrackers[0]->nextSegFile(dbRoot, partition, segment, newHwm, startLbid);	
+					newExtent = dbRootExtentTrackers[column.colNo]->nextSegFile(dbRoot, partition, segment, newHwm, startLbid);	
 				}
 			 }
 			 
@@ -297,7 +297,7 @@ int ColumnOp::allocRowId(const TxnID& txnid, bool useStartingExtent,
 				}
 				
 				rc = BRMWrapper::getInstance()->allocateStripeColExtents(cols, dbRoot, partition, segment, extents);
-				newHwm = extents[0].startBlkOffset;
+				newHwm = extents[column.colNo].startBlkOffset;
 				if (rc != NO_ERROR)
 					return rc;
 					
diff --git a/writeengine/wrapper/writeengine.cpp b/writeengine/wrapper/writeengine.cpp
index 50d846e7c..7cb3ca85e 100644
--- a/writeengine/wrapper/writeengine.cpp
+++ b/writeengine/wrapper/writeengine.cpp
@@ -1505,6 +1505,19 @@ int WriteEngineWrapper::insertColumnRecsBinary(const TxnID& txnid,
    for (i = 0; i < colStructList.size(); i++)
       Convertor::convertColType(&colStructList[i]);
 
+   // MCOL-984: find the smallest column width to calculate the RowID from so
+   // that all HWMs will be incremented by this operation
+   int32_t lowColLen = 8192;
+   int32_t colId = 0;
+   for (uint32_t colIt = 0; colIt < colStructList.size(); colIt++)
+   {
+        if (colStructList[colIt].colWidth < lowColLen)
+        {
+            colId = colIt;
+            lowColLen = colStructList[colId].colWidth;
+        }
+   }
+
   // rc = checkValid(txnid, colStructList, colValueList, ridList);
   // if (rc != NO_ERROR)
    //   return rc;
@@ -1531,8 +1544,8 @@ int WriteEngineWrapper::insertColumnRecsBinary(const TxnID& txnid,
     //--------------------------------------------------------------------------
     if (isFirstBatchPm)
     {
-        currentDBrootIdx = dbRootExtentTrackers[0]->getCurrentDBRootIdx();
-        extentInfo = dbRootExtentTrackers[0]->getDBRootExtentList();
+        currentDBrootIdx = dbRootExtentTrackers[colId]->getCurrentDBRootIdx();
+        extentInfo = dbRootExtentTrackers[colId]->getDBRootExtentList();
         dbRoot = extentInfo[currentDBrootIdx].fDbRoot;
         partitionNum = extentInfo[currentDBrootIdx].fPartition;
 
@@ -1698,7 +1711,7 @@ int WriteEngineWrapper::insertColumnRecsBinary(const TxnID& txnid,
     } // if (isFirstBatchPm)
     else //get the extent info from tableMetaData
     {
-        ColExtsInfo aColExtsInfo = tableMetaData->getColExtsInfo(colStructList[0].dataOid);
+        ColExtsInfo aColExtsInfo = tableMetaData->getColExtsInfo(colStructList[colId].dataOid);
         ColExtsInfo::iterator it = aColExtsInfo.begin();
         while (it != aColExtsInfo.end())
         {
@@ -1730,20 +1743,7 @@ int WriteEngineWrapper::insertColumnRecsBinary(const TxnID& txnid,
     //--------------------------------------------------------------------------
     // allocate row id(s)
     //--------------------------------------------------------------------------
-
-   // MCOL-984: find the smallest column width to calculate the RowID from so
-   // that all HWMs will be incremented by this operation
-   int32_t lowColLen = 8192;
-   int32_t colId = 0;
-   for (uint32_t colIt = 0; colIt < colStructList.size(); colIt++)
-   {
-        if (colStructList[colIt].colWidth < lowColLen)
-        {
-            colId = colIt;
-            lowColLen = colStructList[colId].colWidth;
-            curColStruct = colStructList[colId];
-        }
-   }
+   curColStruct = colStructList[colId];
    colOp = m_colOp[op(curColStruct.fCompressionType)];
 
    colOp->initColumn(curCol);
@@ -1765,7 +1765,7 @@ int WriteEngineWrapper::insertColumnRecsBinary(const TxnID& txnid,
     if (it != aColExtsInfo.end())
     {
         hwm = it->hwm;
-        //cout << "Got from colextinfo hwm for oid " << colStructList[0].dataOid << " is " << hwm << " and seg is " << colStructList[0].fColSegment << endl;
+        //cout << "Got from colextinfo hwm for oid " << colStructList[colId].dataOid << " is " << hwm << " and seg is " << colStructList[colId].fColSegment << endl;
     }
 
    oldHwm = hwm; //Save this info for rollback