HPCC-30974 Copying super files via fileservices/dfu does not track re…

…ad/cost stats. Signed-off-by: Shamser Ahmed <[email protected]>
hpcc-systems · Dec 19, 2023 · b7613c3 · b7613c3
1 parent 8d8e3f1
commit b7613c3
Showing 1 changed file with 37 additions and 31 deletions.
diff --git a/dali/ft/filecopy.cpp b/dali/ft/filecopy.cpp
@@ -3384,25 +3384,9 @@ void FileSprayer::spray()
     afterTransfer();
 
     //If got here then we have succeeded
+    //(note: if we don't get here, file access costs will not be updated)
     updateTargetProperties();
 
-    //Calculate and store file access cost
-    double fileAccessCost = 0.0;
-    if (distributedTarget)
-    {
-        StringBuffer cluster;
-        distributedTarget->getClusterName(0, cluster);
-        if (!cluster.isEmpty())
-            fileAccessCost += calcFileAccessCost(cluster, totalNumWrites, 0);
-    }
-    if (distributedSource && distributedSource->querySuperFile()==nullptr)
-    {
-        StringBuffer cluster;
-        distributedSource->getClusterName(0, cluster);
-        if (!cluster.isEmpty())
-            fileAccessCost += calcFileAccessCost(cluster, 0, totalNumReads);
-    }
-    progressReport->setFileAccessCost(fileAccessCost);
     StringBuffer copyEventText;     // [logical-source] > [logical-target]
     if (distributedSource)
         copyEventText.append(distributedSource->queryLogicalName());
@@ -3455,6 +3439,7 @@ void FileSprayer::updateTargetProperties()
 {
     TimeSection timer("FileSprayer::updateTargetProperties() time");
     Owned<IException> error;
+    cost_type totalCost = 0;
     if (distributedTarget)
     {
         StringBuffer failedParts;
@@ -3468,6 +3453,8 @@ void FileSprayer::updateTargetProperties()
         bool sameSizeSourceTarget = (sources.ordinality() == distributedTarget->numParts());
         offset_t partCompressedLength = 0;
 
+        IDistributedSuperFile * superTgt = distributedTarget->querySuperFile();
+        IDistributedSuperFile * superSrc = distributedSource->querySuperFile();
         ForEachItemIn(idx, partition)
         {
             PartitionPoint & cur = partition.item(idx);
@@ -3585,16 +3572,44 @@ void FileSprayer::updateTargetProperties()
                 partLength = 0;
                 partCompressedLength = 0;
             }
+
+            // Update cost/numWrites for target file or subfile (if superfile)
+            IDistributedFile * tgtDistFile = nullptr;
+            if (superTgt)
+            {
+                unsigned subfilepartidx = 0;
+                tgtDistFile = superTgt->querySubPart(cur.whichOutput, subfilepartidx);
+            }
+            else
+                tgtDistFile = distributedTarget;
+            DistributedFilePropertyLock lock(tgtDistFile);
+            IPropertyTree &tgtProps = lock.queryAttributes();
+            cost_type writeCost = money2cost_type(calcFileAccessCost(tgtDistFile, curProgress.numWrites, 0));
+            tgtProps.setPropInt64(getDFUQResultFieldName(DFUQRFwriteCost), writeCost);
+            tgtProps.setPropInt64(getDFUQResultFieldName(DFUQRFnumDiskWrites), totalNumWrites);
+
+            // Update cost/numReads for source file or subfile (if superfile)
+            IDistributedFile * srcDistFile = nullptr;
+            if (superSrc)
+            {
+                unsigned subfilepartidx = 0;
+                srcDistFile = superSrc->querySubPart(cur.whichInput, subfilepartidx);
+            }
+            else
+                srcDistFile = distributedSource;
+            cost_type legacyReadCost = getLegacyReadCost(srcDistFile->queryAttributes(), srcDistFile);
+            cost_type readCost = money2cost_type(calcFileAccessCost(srcDistFile, curProgress.numReads, 0));
+            srcDistFile->addAttrValue(getDFUQResultFieldName(DFUQRFreadCost), legacyReadCost+readCost);
+            srcDistFile->addAttrValue(getDFUQResultFieldName(DFUQRFnumDiskReads), curProgress.numReads);
+
+            totalCost += (writeCost+readCost);
         }
 
         if (failedParts.length())
             error.setown(MakeStringException(DFTERR_InputOutputCrcMismatch, "%s", failedParts.str()));
 
         DistributedFilePropertyLock lock(distributedTarget);
         IPropertyTree &curProps = lock.queryAttributes();
-        cost_type writeCost = money2cost_type(calcFileAccessCost(distributedTarget, totalNumWrites, 0));
-        curProps.setPropInt64(getDFUQResultFieldName(DFUQRFwriteCost), writeCost);
-        curProps.setPropInt64(getDFUQResultFieldName(DFUQRFnumDiskWrites), totalNumWrites);
 
         if (calcCRC())
             curProps.setPropInt(FAcrc, totalCRC.get());
@@ -3771,17 +3786,8 @@ void FileSprayer::updateTargetProperties()
         if (expireDays != -1)
             curProps.setPropInt("@expireDays", expireDays);
     }
-    if (distributedSource)
-    {
-        if (distributedSource->querySuperFile()==nullptr)
-        {
-            IPropertyTree & fileAttr = distributedSource->queryAttributes();
-            cost_type legacyReadCost = getLegacyReadCost(fileAttr, distributedSource);
-            cost_type curReadCost = money2cost_type(calcFileAccessCost(distributedSource, 0, totalNumReads));
-            distributedSource->addAttrValue(getDFUQResultFieldName(DFUQRFreadCost), legacyReadCost+curReadCost);
-            distributedSource->addAttrValue(getDFUQResultFieldName(DFUQRFnumDiskReads), totalNumReads);
-        }
-    }
+
+    progressReport->setFileAccessCost(cost_type2money(totalCost));
     if (error)
         throw error.getClear();
 }