HPCC-30854 Ensure partitioning of KJ parts between workers in k8s

When mapping index parts to workers in containerized, treat all as if locally available, such that they will be round-robined into worker buckets. This will cause (when the default remoteKeyedLookup is true), workers to directly access a partition of the index parts and farm out request of the rest to other workers. The net result is to increase cache locality (both jhtree and within the Linux page cache). Also make some detailed logging conditional on logging level. Signed-off-by: Jake Smith <[email protected]>
hpcc-systems · Nov 23, 2023 · b1d0564 · b1d0564
1 parent e09e412
commit b1d0564
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 10 deletions.
diff --git a/thorlcr/activities/keyedjoin/thkeyedjoin.cpp b/thorlcr/activities/keyedjoin/thkeyedjoin.cpp
@@ -178,8 +178,14 @@ class CKeyedJoinMaster : public CMasterActivity
                             {
                                 do
                                 {
-                                    INode &groupNode = dfsGroup.queryNode(gn);
-                                    if (partNode->equals(&groupNode))
+                                    // NB: all parts are considered 'local' in containerized mode
+                                    // It will cause the parts to be striped across the group (and hence the workers),
+                                    // such that the parts will be partitioned, with each worker dealing with some parts
+                                    // locally via local key lookup handlers and the rest being handled remotely by other
+                                    // workers via remote key lookup handlers.
+                                    // remoteKeyedLookup=false will disabled this default behaviour, causing all parts
+                                    // to be handled locally by each worker.
+                                    if (isContainerized() || partNode->equals(&dfsGroup.queryNode(gn)))
                                     {
                                         /* NB: If there's >1 slave per node (e.g. slavesPerNode>1) then there are multiple matching node's in the dfsGroup
                                         * Which means a copy of a part may already be assigned to a cluster slave map. This check avoid handling it again if it has.

diff --git a/thorlcr/activities/keyedjoin/thkeyedjoinslave.cpp b/thorlcr/activities/keyedjoin/thkeyedjoinslave.cpp
@@ -1030,7 +1030,7 @@ class CKeyedJoinSlave : public CSlaveActivity, implements IJoinProcessor, implem
         {
             StringBuffer log;
             getInfo(log).append(": ").append(msg);
-            PROGLOG("%s", log.str());
+            LOG(MCthorDetailedDebugInfo, thorJob, "%s", log.str());
         }
         virtual StringBuffer &getInfo(StringBuffer &info) const
         {
@@ -1165,10 +1165,11 @@ class CKeyedJoinSlave : public CSlaveActivity, implements IJoinProcessor, implem
         }
         virtual void end()
         {
-#ifdef _DEBUG
-            VStringBuffer log("processed: %" I64F "u", total);
-            trace(log);
-#endif
+            if (!REJECTLOG(MCthorDetailedDebugInfo))
+            {
+                VStringBuffer log("processed: %" I64F "u", total);
+                trace(log);
+            }
         }
         virtual void process(CThorExpandingRowArray &processing, unsigned selected) = 0;
     // IThreaded
@@ -2931,9 +2932,8 @@ class CKeyedJoinSlave : public CSlaveActivity, implements IJoinProcessor, implem
             }
         }
         handlerContainer.init();
-#ifdef _DEBUG
-        handlerContainer.trace();
-#endif
+        if (!REJECTLOG(MCthorDetailedDebugInfo))
+            handlerContainer.trace();
     }
 public:
     IMPLEMENT_IINTERFACE_USING(PARENT);