Merge pull request #9949 from ghalliday/issue16649

HPCC-16649 Restrict moving projects over distributes if row size increases

Reviewed-By: Shamser Ahmed <shamser.ahmed@lexisnexis.co.uk>
Reviewed-By: Richard Chapman <rchapman@hpccsystems.com>

ecl/hql/hqlattr.cpp

@@ -57,7 +57,7 @@ MODULE_EXIT()
 // getCachedEclCRC(), cacheTablesUsed(), isIndependentOfScope()
 // logic inside createDataset
 
-//Originally the idea was to have a class instance for each kind of opcode, and to call opcode[op]->evalautePropXXXXXX(this);
+//Originally the idea was to have a class instance for each kind of opcode, and to call opcode[op]->evaluatePropXXXXXX(this);
 //to evaluate the property.  However because there are so many opcodes I'm not convinced this is the best way.
 //Better may be to model it more on the way queryRecordCount() is implemented.
 
@@ -1239,7 +1239,7 @@ static IHqlExpression * evaluateRecordAttrSize(IHqlExpression * expr)
 }
 
 
-static IHqlExpression * evalautePropSize(IHqlExpression * expr)
+static IHqlExpression * evaluatePropSize(IHqlExpression * expr)
 {
     switch (expr->getOperator())
     {
@@ -1434,7 +1434,7 @@ IHqlExpression * HqlUnadornedNormalizer::createTransformed(IHqlExpression * expr
     return HqlCachedPropertyTransformer::createTransformed(expr);
 }
 
-static IHqlExpression * evalautePropUnadorned(IHqlExpression * expr)
+static IHqlExpression * evaluatePropUnadorned(IHqlExpression * expr)
 {
     HqlUnadornedNormalizer normalizer;
     //NB: Also has the side-effect of adding any missing attributes
@@ -1581,7 +1581,7 @@ static bool optimizeFieldOrder(HqlExprArray & out, const HqlExprCopyArray & in)
     return true;
 }
 
-static IHqlExpression * evalautePropAligned(IHqlExpression * expr)
+static IHqlExpression * evaluatePropAligned(IHqlExpression * expr)
 {
     bool same = true;
     HqlExprArray result;
@@ -2046,35 +2046,56 @@ static unsigned estimateRowSize(IHqlExpression * record)
 }
 
 
-bool reducesRowSize(IHqlExpression * expr)
+//compare the expected size converting from oldRecord to newRecord
+static bool increasesRowSize(IHqlExpression * newRecord, IHqlExpression * oldRecord, bool assumeIncrease)
 {
-    //More: This should be improved...., but slightly tricky without doing lots more processing.
-    IHqlExpression * transform = queryNewColumnProvider(expr);
-    IHqlExpression * prevRecord = expr->queryChild(0)->queryRecord();
-    unsigned newRowSize = estimateRowSize(transform->queryRecord());
-    unsigned prevRowSize = estimateRowSize(prevRecord);
-    if ((newRowSize != UNKNOWN_LENGTH) && (prevRowSize != UNKNOWN_LENGTH))
-        return newRowSize < prevRowSize;
+    unsigned newRowSize = estimateRowSize(newRecord);
+    unsigned oldRowSize = estimateRowSize(oldRecord);
+    //Fixed size records
+    if ((newRowSize != UNKNOWN_LENGTH) && (oldRowSize != UNKNOWN_LENGTH))
+        return newRowSize > oldRowSize;
 
-    IHqlExpression * record = expr->queryRecord();
-    if (getFlatFieldCount(record) < getFlatFieldCount(prevRecord))
+    //Fixed size record compared with the minimum size of a variable size record.
+    size32_t oldMinSize = getMinRecordSize(oldRecord);
+    size32_t newMinSize = getMinRecordSize(newRecord);
+    if (isFixedSizeRecord(newRecord) && newRowSize <= oldMinSize)
+        return false;
+    if (isFixedSizeRecord(oldRecord) && oldRowSize <= newMinSize)
         return true;
-    return false;
+
+    HqlRecordStats oldStats;
+    HqlRecordStats newStats;
+    gatherRecordStats(oldStats, oldRecord);
+    gatherRecordStats(newStats, newRecord);
+
+    if (newStats.unknownSizeFields != oldStats.unknownSizeFields)
+    {
+        //Assume that unknown size fields are larger - the rows are sometimes smaller, but almost definitely more painful.
+        return (newStats.unknownSizeFields > oldStats.unknownSizeFields);
+    }
+
+    //Check the minimum size - essentially the size of the fixed size fields.
+    if (newMinSize != oldMinSize)
+        return newMinSize > oldMinSize;
+    if (newStats.fields > oldStats.fields)
+        return true;
+    return assumeIncrease;
 }
 
-bool increasesRowSize(IHqlExpression * expr)
+// Does this operation decrease the size of the row?  False negatives preferred.
+bool reducesRowSize(IHqlExpression * expr)
 {
-    IHqlExpression * transform = queryNewColumnProvider(expr);
-    IHqlExpression * prevRecord = expr->queryChild(0)->queryRecord();
-    unsigned newRowSize = estimateRowSize(transform);
-    unsigned prevRowSize = estimateRowSize(prevRecord);
-    if ((newRowSize != UNKNOWN_LENGTH) && (prevRowSize != UNKNOWN_LENGTH))
-        return newRowSize > prevRowSize;
+    OwnedHqlExpr newRecord = getSerializedForm(expr->queryRecord(), diskAtom);
+    OwnedHqlExpr oldRecord = getSerializedForm(expr->queryChild(0)->queryRecord(), diskAtom);
+    return increasesRowSize(oldRecord, newRecord, false);
+}
 
-    IHqlExpression * record = expr->queryRecord();
-    if (getFlatFieldCount(record) > getFlatFieldCount(prevRecord))
-        return true;
-    return false;
+// Does this operation increase the size of the row?  False negatives preferred.
+bool increasesRowSize(IHqlExpression * expr)
+{
+    OwnedHqlExpr newRecord = getSerializedForm(expr->queryRecord(), diskAtom);
+    OwnedHqlExpr oldRecord = getSerializedForm(expr->queryChild(0)->queryRecord(), diskAtom);
+    return increasesRowSize(newRecord, oldRecord, false);
 }
 
 bool isLimitedDataset(IHqlExpression * expr, bool onFailOnly)
@@ -3068,7 +3089,7 @@ IHqlExpression * calcRowInformation(IHqlExpression * expr)
     return info.createRecordCountAttr();
 }
 
-static IHqlExpression * evalautePropRecordCount(IHqlExpression * expr)
+static IHqlExpression * evaluatePropRecordCount(IHqlExpression * expr)
 {
     OwnedHqlExpr info = calcRowInformation(expr);
     meta.addProperty(expr, EPrecordCount, info);
@@ -3598,7 +3619,7 @@ IHqlExpression * HqlLocationIndependentNormalizer::createTransformed(IHqlExpress
     return transformed.getClear();
 }
 
-IHqlExpression * evalautePropLocationIndependent(IHqlExpression * expr)
+IHqlExpression * evaluatePropLocationIndependent(IHqlExpression * expr)
 {
     if (isAlwaysLocationIndependent(expr))
         return expr->queryBody();
@@ -3993,19 +4014,19 @@ IHqlExpression * CHqlRealExpression::queryProperty(ExprPropKind kind)
     switch (kind)
     {
     case EPrecordCount:
-        return evalautePropRecordCount(this);
+        return evaluatePropRecordCount(this);
     case EPdiskserializedForm:
         return evaluatePropSerializedForm(this, kind, diskAtom);
     case EPinternalserializedForm:
         return evaluatePropSerializedForm(this, kind, internalAtom);
     case EPsize:
-        return evalautePropSize(this);
+        return evaluatePropSize(this);
     case EPaligned:
-        return evalautePropAligned(this);
+        return evaluatePropAligned(this);
     case EPunadorned:
-        return evalautePropUnadorned(this);
+        return evaluatePropUnadorned(this);
     case EPlocationIndependent:
-        return evalautePropLocationIndependent(this);
+        return evaluatePropLocationIndependent(this);
     case EPlikelihood:
         return evaluateLikelihood(this);
     }

ecl/hql/hqlopt.cpp

@@ -3325,14 +3325,14 @@ IHqlExpression * CTreeOptimizer::doCreateTransformed(IHqlExpression * transforme
             case no_subsort:
                 if (transformedCountProject)
                     break;
-                if (increasesRowSize(transformed))
+                if ((options & HOOminimizeNetworkAndMemory) && increasesRowSize(transformed))
                     break;
                 return moveProjectionOverSimple(transformed, true, false);
             case no_distribute:
                 //Cannot move a count project over anything that changes the order of the records.
                 if (transformedCountProject)
                     break;
-                if (increasesRowSize(transformed))
+                if ((options & HOOminimizeNetworkAndMemory) && increasesRowSize(transformed))
                     break;
                 return moveProjectionOverSimple(transformed, true, false);
             case no_distributed:
@@ -3524,7 +3524,7 @@ IHqlExpression * CTreeOptimizer::doCreateTransformed(IHqlExpression * transforme
             case no_distribute:
             case no_sort:
             case no_subsort:
-                if (increasesRowSize(transformed))
+                if ((options & HOOminimizeNetworkAndMemory) && increasesRowSize(transformed))
                     break;
                 return moveProjectionOverSimple(transformed, true, false);
             case no_distributed:

ecl/hql/hqlopt.hpp

@@ -33,6 +33,7 @@ enum
     HOOalwayslocal              = 0x0100,
     HOOexpensive                = 0x0200,   // include potentially expensive optimizations
     HOOexpandselectcreaterow    = 0x0400,
+    HOOminimizeNetworkAndMemory = 0x0800,
 };
 
 extern HQL_API IHqlExpression * optimizeHqlExpression(IErrorReceiver & errorProcessor, IHqlExpression * expr, unsigned options);

ecl/hql/hqlutil.cpp

@@ -2250,6 +2250,53 @@ unsigned getFlatFieldCount(IHqlExpression * expr)
     }
 }
 
+//This function gathers information about the number of fields etc. in a record.
+//MORE: Should this calculate numFields, numExpandedFields, numDatasetFields all in a single function and
+//then cache it as a property of a no_record?   Only if it becomes a significant bottleneck.
+void gatherRecordStats(HqlRecordStats & stats, IHqlExpression * expr)
+{
+    switch (expr->getOperator())
+    {
+    case no_record:
+        {
+            ForEachChild(i, expr)
+                gatherRecordStats(stats, expr->queryChild(i));
+            break;
+        }
+    case no_ifblock:
+        return gatherRecordStats(stats, expr->queryChild(1));
+    case no_field:
+        {
+            ITypeInfo * type = expr->queryType();
+            switch (type->getTypeCode())
+            {
+            case type_table:
+            case type_groupedtable:
+                stats.fields++;
+                stats.unknownSizeFields++;
+                break;
+            case type_dictionary:
+                stats.fields++;
+                stats.unknownSizeFields++;
+                break;
+            case type_row:
+                gatherRecordStats(stats, expr->queryRecord());
+                break;
+            default:
+                stats.fields++;
+                //HPCC-17606 add: if (type->getSize() == UNKNOWN_LENGTH) stats.unknownSizeFields++;
+                break;
+            }
+            break;
+        }
+    case no_attr:
+    case no_attr_link:
+    case no_attr_expr:
+        break;
+    default:
+        UNIMPLEMENTED;
+    }
+}
 
 unsigned isEmptyRecord(IHqlExpression * record)
 {

ecl/hql/hqlutil.hpp

@@ -94,6 +94,13 @@ extern HQL_API unsigned isEmptyRecord(IHqlExpression * record);
 extern HQL_API unsigned isSimpleRecord(IHqlExpression * record);
 extern HQL_API void getSimpleFields(HqlExprArray &out, IHqlExpression *record);
 
+struct HqlRecordStats
+{
+    unsigned fields = 0;
+    unsigned unknownSizeFields = 0;
+};
+extern HQL_API void gatherRecordStats(HqlRecordStats & stats, IHqlExpression * expr);
+
 extern HQL_API bool isTrivialSelectN(IHqlExpression * expr);
 
 extern HQL_API IHqlExpression * queryConvertChoosenNSort(IHqlExpression * expr, unsigned __int64 topNlimit);

ecl/hqlcpp/hqlcpp.cpp

@@ -1910,7 +1910,7 @@ void HqlCppTranslator::postProcessOptions()
         wu()->setCloneable(true);
 }
 
-unsigned HqlCppTranslator::getOptimizeFlags() const
+unsigned HqlCppTranslator::getOptimizeFlags(bool insideChildQuery) const
 {
     unsigned optFlags = HOOfold;
     switch (targetClusterType)
@@ -1934,6 +1934,12 @@ unsigned HqlCppTranslator::getOptimizeFlags() const
         optFlags |= HOOexpensive;
     if (options.expandSelectCreateRow)
         optFlags  |= HOOexpandselectcreaterow;
+
+    //The following flag should possibly be dependent on this condition:
+    //if (targetThor() && !insideChildQuery)
+    //However it may be better to err on the side of caution - see HPCC-10144 since moving a project over a barrier may
+    //cause it in some circumstances to be evaluated many more times.  See HPCC-17439 to revisit later.
+    optFlags |= HOOminimizeNetworkAndMemory;
     return optFlags;
 }
 

ecl/hqlcpp/hqlcpp.ipp

@@ -986,7 +986,7 @@ public:
     inline bool hasDynamicFilename(IHqlExpression * expr) const { return options.allFilenamesDynamic || hasDynamic(expr); }
     inline bool canGenerateStringInline(unsigned len)       { return ((options.inlineStringThreshold == 0) || (len <= options.inlineStringThreshold)); }
 
-    unsigned getOptimizeFlags() const;
+    unsigned getOptimizeFlags(bool insideChildQuery) const;
     unsigned getSourceAggregateOptimizeFlags() const;
     void addGlobalOnWarning(IHqlExpression * setMetaExpr);
 
@@ -1928,7 +1928,7 @@ protected:
     IHqlExpression * getResourcedGraph(IHqlExpression * expr, IHqlExpression * graphIdExpr);
     IHqlExpression * getResourcedChildGraph(BuildCtx & ctx, IHqlExpression * childQuery, unsigned numResults, node_operator graphKind, bool unlimitedResources);
     IHqlExpression * optimizeCompoundSource(IHqlExpression * expr, unsigned flags);
-    IHqlExpression * optimizeGraphPostResource(IHqlExpression * expr, unsigned csfFlags, bool projectBeforeSpill);
+    IHqlExpression * optimizeGraphPostResource(IHqlExpression * expr, unsigned csfFlags, bool projectBeforeSpill, bool insideChildQuery);
     bool isInlineOk();
     GraphLocalisation getGraphLocalisation(IHqlExpression * expr, bool isInsideChildQuery);
     bool isAlwaysCoLocal();

ecl/hqlcpp/hqlcppds.cpp

@@ -1512,7 +1512,7 @@ void ChildGraphBuilder::generateGraph(BuildCtx & ctx)
     //Remove this line once all engines use the new child queries exclusively
     if (numResults == 0) numResults++;
 
-    OwnedHqlExpr resourced = translator.getResourcedChildGraph(graphctx, childQuery, numResults, no_none, false);
+    OwnedHqlExpr resourced = translator.getResourcedChildGraph(graphctx, childQuery, numResults, no_childquery, false);
 
     Owned<ParentExtract> extractBuilder = translator.createExtractBuilder(graphctx, PETchild, represents, resourced, true);
     if (!translator.queryOptions().serializeRowsetInExtract)
@@ -1560,7 +1560,7 @@ void ChildGraphBuilder::generatePrefetchGraph(BuildCtx & _ctx, OwnedHqlExpr * re
     BuildCtx aliasctx(ctx);
     aliasctx.addGroup();
 
-    OwnedHqlExpr resourced = translator.getResourcedChildGraph(ctx, childQuery, numResults, no_none, false);
+    OwnedHqlExpr resourced = translator.getResourcedChildGraph(ctx, childQuery, numResults, no_hqlproject, false);
 
     Owned<ParentExtract> extractBuilder = translator.createExtractBuilder(ctx, PETchild, represents, resourced, false);
     createBuilderAlias(aliasctx, extractBuilder);
@@ -1818,11 +1818,12 @@ IHqlExpression * HqlCppTranslator::getResourcedChildGraph(BuildCtx & ctx, IHqlEx
         noteFinishedTiming("workunit;tree transform: optimize disk read", startCycles);
     }
 
+    bool isInsideChildQuery = (graphKind == no_childquery) || insideChildQuery(ctx);
     if (options.optimizeGraph)
     {
         cycle_t startCycles = get_cycles_now();
         traceExpression("BeforeOptimizeSub", resourced);
-        resourced.setown(optimizeHqlExpression(queryErrorProcessor(), resourced, getOptimizeFlags()|HOOcompoundproject));
+        resourced.setown(optimizeHqlExpression(queryErrorProcessor(), resourced, getOptimizeFlags(isInsideChildQuery)|HOOcompoundproject));
         traceExpression("AfterOptimizeSub", resourced);
         noteFinishedTiming("workunit;optimize graph", startCycles);
     }
@@ -1835,7 +1836,7 @@ IHqlExpression * HqlCppTranslator::getResourcedChildGraph(BuildCtx & ctx, IHqlEx
     if (graphKind == no_loop)
     {
         bool insideChild = insideChildQuery(ctx);
-        resourced.setown(resourceLoopGraph(*this, activeRows, resourced, targetClusterType, graphIdExpr, numResults, insideChild, unlimitedResources));
+        resourced.setown(resourceLoopGraph(*this, activeRows, resourced, targetClusterType, graphIdExpr, numResults, isInsideChildQuery, unlimitedResources));
     }
     else
         resourced.setown(resourceNewChildGraph(*this, activeRows, resourced, targetClusterType, graphIdExpr, numResults));
@@ -1844,11 +1845,11 @@ IHqlExpression * HqlCppTranslator::getResourcedChildGraph(BuildCtx & ctx, IHqlEx
     checkNormalized(ctx, resourced);
     traceExpression("AfterResourcing Child", resourced);
     
-    resourced.setown(optimizeGraphPostResource(resourced, csfFlags, false));
+    resourced.setown(optimizeGraphPostResource(resourced, csfFlags, false, isInsideChildQuery));
     if (options.optimizeSpillProject)
     {
         resourced.setown(convertSpillsToActivities(resourced, true));
-        resourced.setown(optimizeGraphPostResource(resourced, csfFlags, false));
+        resourced.setown(optimizeGraphPostResource(resourced, csfFlags, false, isInsideChildQuery));
     }
 
     if (options.paranoidCheckNormalized || options.paranoidCheckDependencies)

ecl/hqlcpp/hqlhtcpp.cpp

@@ -1996,6 +1996,23 @@ void ActivityInstance::setInternalSink(bool value)
 }
 
 
+static void getRecordSizeText(StringBuffer & out, IHqlExpression * record)
+{
+    size32_t minSize = getMinRecordSize(record);
+    if (isVariableSizeRecord(record))
+    {
+        size32_t expectedSize = getExpectedRecordSize(record);
+        out.append(minSize).append("..");
+        if (maxRecordSizeUsesDefault(record))
+            out.append("?");
+        else
+            out.append(getMaxRecordSize(record, 0));
+        out.append("(").append(expectedSize).append(")");
+    }
+    else
+        out.append(minSize);
+}
+
 void ActivityInstance::createGraphNode(IPropertyTree * defaultSubGraph, bool alwaysExecuted)
 {
     IPropertyTree * parentGraphNode = subgraph ? subgraph->tree.get() : defaultSubGraph;
@@ -2118,26 +2135,17 @@ void ActivityInstance::createGraphNode(IPropertyTree * defaultSubGraph, bool alw
 
         if (options.noteRecordSizeInGraph)
         {
-            IHqlExpression * record = dataset->queryRecord();
+            LinkedHqlExpr record = dataset->queryRecord();
             if (!record && (getNumChildTables(dataset) == 1))
-                record = dataset->queryChild(0)->queryRecord();
+                record.set(dataset->queryChild(0)->queryRecord());
             if (record)
             {
-                size32_t minSize = getMinRecordSize(record);
-                if (isVariableSizeRecord(record))
-                {
-                    size32_t expectedSize = getExpectedRecordSize(record);
-                    StringBuffer temp;
-                    temp.append(minSize).append("..");
-                    if (maxRecordSizeUsesDefault(record))
-                        temp.append("?");
-                    else
-                        temp.append(getMaxRecordSize(record, translator.getDefaultMaxRecordSize()));
-                    temp.append("(").append(expectedSize).append(")");
-                    addAttribute("recordSize", temp.str());
-                }
-                else
-                    addAttributeInt("recordSize", minSize);
+                //In Thor the serialized record is the interesting value, so include that in the graph
+                if (translator.targetThor())
+                    record.setown(getSerializedForm(record, diskAtom));
+                StringBuffer temp;
+                getRecordSizeText(temp, record);
+                addAttribute("recordSize", temp.str());
             }
         }
 
@@ -9372,7 +9380,7 @@ IHqlExpression * HqlCppTranslator::optimizeCompoundSource(IHqlExpression * expr,
     return ret.getClear();
 }
 
-IHqlExpression * HqlCppTranslator::optimizeGraphPostResource(IHqlExpression * expr, unsigned csfFlags, bool projectBeforeSpill)
+IHqlExpression * HqlCppTranslator::optimizeGraphPostResource(IHqlExpression * expr, unsigned csfFlags, bool projectBeforeSpill, bool insideChildQuery)
 {
     LinkedHqlExpr resourced = expr;
     // Second attempt to spot compound disk reads - this time of spill files for thor.
@@ -9399,7 +9407,7 @@ IHqlExpression * HqlCppTranslator::optimizeGraphPostResource(IHqlExpression * ex
     {
         cycle_t startCycles = get_cycles_now();
         traceExpression("BeforeOptimize2", resourced);
-        resourced.setown(optimizeHqlExpression(queryErrorProcessor(), resourced, getOptimizeFlags()|HOOcompoundproject));
+        resourced.setown(optimizeHqlExpression(queryErrorProcessor(), resourced, getOptimizeFlags(insideChildQuery)|HOOcompoundproject));
         traceExpression("AfterOptimize2", resourced);
         noteFinishedTiming("compile:optimize graph", startCycles);
     }
@@ -9430,7 +9438,7 @@ IHqlExpression * HqlCppTranslator::getResourcedGraph(IHqlExpression * expr, IHql
     // Call optimizer before resourcing so items get moved over conditions, and remove other items
     // which would otherwise cause extra spills.
     traceExpression("BeforeOptimize", resourced);
-    unsigned optFlags = getOptimizeFlags();
+    unsigned optFlags = getOptimizeFlags(false);
 
     checkNormalized(resourced);
     if (options.optimizeGraph)
@@ -9480,11 +9488,11 @@ IHqlExpression * HqlCppTranslator::getResourcedGraph(IHqlExpression * expr, IHql
     checkNormalized(resourced);
 
     bool createGraphResults = (outputLibraryId != 0) || options.alwaysUseGraphResults;
-    resourced.setown(optimizeGraphPostResource(resourced, csfFlags, options.optimizeSpillProject && !createGraphResults));
+    resourced.setown(optimizeGraphPostResource(resourced, csfFlags, options.optimizeSpillProject && !createGraphResults, false));
     if (options.optimizeSpillProject)
     {
         resourced.setown(convertSpillsToActivities(resourced, createGraphResults));
-        resourced.setown(optimizeGraphPostResource(resourced, csfFlags, false));
+        resourced.setown(optimizeGraphPostResource(resourced, csfFlags, false, false));
     }
 
     checkNormalized(resourced);

ecl/hqlcpp/hqlsource.cpp

@@ -75,7 +75,8 @@ void HqlCppTranslator::addGlobalOnWarning(IHqlExpression * setMetaExpr)
 
 unsigned HqlCppTranslator::getSourceAggregateOptimizeFlags() const
 {
-    return getOptimizeFlags()|HOOfold|HOOinsidecompound;
+    const bool insideChildQuery = false; // value does not currently matter
+    return getOptimizeFlags(insideChildQuery)|HOOfold|HOOinsidecompound;
 }
 
 void HqlCppTranslator::doBuildExprFilepos(BuildCtx & ctx, IHqlExpression * expr, CHqlBoundExpr & tgt)