ccdactivities.cpp 180 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023412342234323442345234623472348234923502351235223532354235523562357235823592360236123622363236423652366236723682369237023712372237323742375237623772378237923802381238223832384238523862387238823892390239123922393239423952396239723982399240024012402240324042405240624072408240924102411241224132414241524162417241824192420242124222423242424252426242724282429243024312432243324342435243624372438243924402441244224432444244524462447244824492450245124522453245424552456245724582459246024612462246324642465246624672468246924702471247224732474247524762477247824792480248124822483248424852486248724882489249024912492249324942495249624972498249925002501250225032504250525062507250825092510251125122513251425152516251725182519252025212522252325242525252625272528252925302531253225332534253525362537253825392540254125422543254425452546254725482549255025512552255325542555255625572558255925602561256225632564256525662567256825692570257125722573257425752576257725782579258025812582258325842585258625872588258925902591259225932594259525962597259825992600260126022603260426052606260726082609261026112612261326142615261626172618261926202621262226232624262526262627262826292630263126322633263426352636263726382639264026412642264326442645264626472648264926502651265226532654265526562657265826592660266126622663266426652666266726682669267026712672267326742675267626772678267926802681268226832684268526862687268826892690269126922693269426952696269726982699270027012702270327042705270627072708270927102711271227132714271527162717271827192720272127222723272427252726272727282729273027312732273327342735273627372738273927402741274227432744274527462747274827492750275127522753275427552756275727582759276027612762276327642765276627672768276927702771277227732774277527762777277827792780278127822783278427852786278727882789279027912792279327942795279627972798279928002801280228032804280528062807280828092810281128122813281428152816281728182819282028212822282328242825282628272828282928302831283228332834283528362837283828392840284128422843284428452846284728482849285028512852285328542855285628572858285928602861286228632864286528662867286828692870287128722873287428752876287728782879288028812882288328842885288628872888288928902891289228932894289528962897289828992900290129022903290429052906290729082909291029112912291329142915291629172918291929202921292229232924292529262927292829292930293129322933293429352936293729382939294029412942294329442945294629472948294929502951295229532954295529562957295829592960296129622963296429652966296729682969297029712972297329742975297629772978297929802981298229832984298529862987298829892990299129922993299429952996299729982999300030013002300330043005300630073008300930103011301230133014301530163017301830193020302130223023302430253026302730283029303030313032303330343035303630373038303930403041304230433044304530463047304830493050305130523053305430553056305730583059306030613062306330643065306630673068306930703071307230733074307530763077307830793080308130823083308430853086308730883089309030913092309330943095309630973098309931003101310231033104310531063107310831093110311131123113311431153116311731183119312031213122312331243125312631273128312931303131313231333134313531363137313831393140314131423143314431453146314731483149315031513152315331543155315631573158315931603161316231633164316531663167316831693170317131723173317431753176317731783179318031813182318331843185318631873188318931903191319231933194319531963197319831993200320132023203320432053206320732083209321032113212321332143215321632173218321932203221322232233224322532263227322832293230323132323233323432353236323732383239324032413242324332443245324632473248324932503251325232533254325532563257325832593260326132623263326432653266326732683269327032713272327332743275327632773278327932803281328232833284328532863287328832893290329132923293329432953296329732983299330033013302330333043305330633073308330933103311331233133314331533163317331833193320332133223323332433253326332733283329333033313332333333343335333633373338333933403341334233433344334533463347334833493350335133523353335433553356335733583359336033613362336333643365336633673368336933703371337233733374337533763377337833793380338133823383338433853386338733883389339033913392339333943395339633973398339934003401340234033404340534063407340834093410341134123413341434153416341734183419342034213422342334243425342634273428342934303431343234333434343534363437343834393440344134423443344434453446344734483449345034513452345334543455345634573458345934603461346234633464346534663467346834693470347134723473347434753476347734783479348034813482348334843485348634873488348934903491349234933494349534963497349834993500350135023503350435053506350735083509351035113512351335143515351635173518351935203521352235233524352535263527352835293530353135323533353435353536353735383539354035413542354335443545354635473548354935503551355235533554355535563557355835593560356135623563356435653566356735683569357035713572357335743575357635773578357935803581358235833584358535863587358835893590359135923593359435953596359735983599360036013602360336043605360636073608360936103611361236133614361536163617361836193620362136223623362436253626362736283629363036313632363336343635363636373638363936403641364236433644364536463647364836493650365136523653365436553656365736583659366036613662366336643665366636673668366936703671367236733674367536763677367836793680368136823683368436853686368736883689369036913692369336943695369636973698369937003701370237033704370537063707370837093710371137123713371437153716371737183719372037213722372337243725372637273728372937303731373237333734373537363737373837393740374137423743374437453746374737483749375037513752375337543755375637573758375937603761376237633764376537663767376837693770377137723773377437753776377737783779378037813782378337843785378637873788378937903791379237933794379537963797379837993800380138023803380438053806380738083809381038113812381338143815381638173818381938203821382238233824382538263827382838293830383138323833383438353836383738383839384038413842384338443845384638473848384938503851385238533854385538563857385838593860386138623863386438653866386738683869387038713872387338743875387638773878387938803881388238833884388538863887388838893890389138923893389438953896389738983899390039013902390339043905390639073908390939103911391239133914391539163917391839193920392139223923392439253926392739283929393039313932393339343935393639373938393939403941394239433944394539463947394839493950395139523953395439553956395739583959396039613962396339643965396639673968396939703971397239733974397539763977397839793980398139823983398439853986398739883989399039913992399339943995399639973998399940004001400240034004400540064007400840094010401140124013401440154016401740184019402040214022402340244025402640274028402940304031403240334034403540364037403840394040404140424043404440454046404740484049405040514052405340544055405640574058405940604061406240634064406540664067406840694070407140724073407440754076407740784079408040814082408340844085408640874088408940904091409240934094409540964097409840994100410141024103410441054106410741084109411041114112411341144115411641174118411941204121412241234124412541264127412841294130413141324133413441354136413741384139414041414142414341444145414641474148414941504151415241534154415541564157415841594160416141624163416441654166416741684169417041714172417341744175417641774178417941804181418241834184418541864187418841894190419141924193419441954196419741984199420042014202420342044205420642074208420942104211421242134214421542164217421842194220422142224223422442254226422742284229423042314232423342344235423642374238423942404241424242434244424542464247424842494250425142524253425442554256425742584259426042614262426342644265426642674268426942704271427242734274427542764277427842794280428142824283428442854286428742884289429042914292429342944295429642974298429943004301430243034304430543064307430843094310431143124313431443154316431743184319432043214322432343244325432643274328432943304331433243334334433543364337433843394340434143424343434443454346434743484349435043514352435343544355435643574358435943604361436243634364436543664367436843694370437143724373437443754376437743784379438043814382438343844385438643874388438943904391439243934394439543964397439843994400440144024403440444054406440744084409441044114412441344144415441644174418441944204421442244234424442544264427442844294430443144324433443444354436443744384439444044414442444344444445444644474448444944504451445244534454445544564457445844594460446144624463446444654466446744684469447044714472447344744475447644774478447944804481448244834484448544864487448844894490449144924493449444954496449744984499450045014502450345044505450645074508450945104511451245134514451545164517451845194520452145224523452445254526452745284529453045314532453345344535453645374538453945404541454245434544454545464547454845494550455145524553455445554556455745584559456045614562456345644565456645674568456945704571457245734574457545764577457845794580458145824583458445854586458745884589459045914592459345944595459645974598459946004601460246034604460546064607460846094610461146124613461446154616461746184619462046214622462346244625462646274628462946304631463246334634463546364637463846394640464146424643464446454646464746484649465046514652465346544655465646574658465946604661466246634664466546664667466846694670467146724673467446754676467746784679468046814682468346844685468646874688468946904691469246934694469546964697
  1. /*##############################################################################
  2. HPCC SYSTEMS software Copyright (C) 2012 HPCC Systems®.
  3. Licensed under the Apache License, Version 2.0 (the "License");
  4. you may not use this file except in compliance with the License.
  5. You may obtain a copy of the License at
  6. http://www.apache.org/licenses/LICENSE-2.0
  7. Unless required by applicable law or agreed to in writing, software
  8. distributed under the License is distributed on an "AS IS" BASIS,
  9. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  10. See the License for the specific language governing permissions and
  11. limitations under the License.
  12. ############################################################################## */
  13. #include "platform.h"
  14. #include "jlib.hpp"
  15. #include "ccd.hpp"
  16. #include "ccdquery.hpp"
  17. #include "ccdstate.hpp"
  18. #include "ccdserver.hpp"
  19. #include "ccdcontext.hpp"
  20. #include "ccddebug.hpp"
  21. #include "ccdactivities.hpp"
  22. #include "ccdqueue.ipp"
  23. #include "ccdsnmp.hpp"
  24. #include "ccdfile.hpp"
  25. #include "ccdkey.hpp"
  26. #include "rtlkey.hpp"
  27. #include "eclrtl_imp.hpp"
  28. #include "rtlread_imp.hpp"
  29. #include "rtlcommon.hpp"
  30. #include "rtldynfield.hpp"
  31. #include "jhtree.hpp"
  32. #include "jlog.hpp"
  33. #include "jmisc.hpp"
  34. #include "udplib.hpp"
  35. #include "csvsplitter.hpp"
  36. #include "thorxmlread.hpp"
  37. #include "thorcommon.ipp"
  38. #include "thorstrand.hpp"
  39. #include "jstats.h"
  40. size32_t diskReadBufferSize = 0x10000;
  41. using roxiemem::OwnedRoxieRow;
  42. using roxiemem::OwnedConstRoxieRow;
  43. using roxiemem::OwnedRoxieString;
  44. using roxiemem::IRowManager;
  45. #define maxContinuationSize 48000 // note - must fit in the 2-byte length field... but also needs to be possible to send back from Roxie server->slave in one packet
  46. size32_t serializeRow(IOutputRowSerializer * serializer, IMessagePacker *output, const void *unserialized)
  47. {
  48. CSizingSerializer sizer;
  49. serializer->serialize(sizer, (const byte *) unserialized);
  50. unsigned serializedLength = sizer.size();
  51. void *udpBuffer = output->getBuffer(serializedLength, true);
  52. CRawRowSerializer memSerializer(serializedLength, (byte *) udpBuffer);
  53. serializer->serialize(memSerializer, (const byte *) unserialized);
  54. assertex(memSerializer.size() == serializedLength);
  55. output->putBuffer(udpBuffer, serializedLength, true);
  56. return serializedLength;
  57. }
  58. inline void appendBuffer(IMessagePacker * output, size32_t size, const void * data, bool isVariable)
  59. {
  60. void *recBuffer = output->getBuffer(size, isVariable);
  61. memcpy(recBuffer, data, size);
  62. output->putBuffer(recBuffer, size, isVariable);
  63. }
  64. extern void putStatsValue(IPropertyTree *node, const char *statName, const char *statType, unsigned __int64 val)
  65. {
  66. if (val)
  67. {
  68. StringBuffer xpath;
  69. xpath.appendf("att[@name='%s']", statName);
  70. IPropertyTree *att = node->queryPropTree(xpath.str());
  71. if (!att)
  72. {
  73. att = node->addPropTree("att");
  74. att->setProp("@name", statName);
  75. }
  76. att->setProp("@type", statType);
  77. att->setPropInt64("@value", val);
  78. }
  79. }
  80. extern void putStatsValue(StringBuffer &reply, const char *statName, const char *statType, unsigned __int64 val)
  81. {
  82. if (val)
  83. {
  84. reply.appendf(" <att name='%s' type='%s' value='%" I64F "d'/>\n", statName, statType, val);
  85. }
  86. }
  87. CActivityFactory::CActivityFactory(unsigned _id, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory, ThorActivityKind _kind)
  88. : id(_id),
  89. subgraphId(_subgraphId),
  90. queryFactory(_queryFactory),
  91. helperFactory(_helperFactory),
  92. kind(_kind),
  93. mystats(allStatistics) // We COULD cut down this list but it would complicate the structure, and we do actually track more in the factory than in the activity
  94. {
  95. if (helperFactory)
  96. {
  97. Owned<IHThorArg> helper = helperFactory();
  98. meta.set(helper->queryOutputMeta());
  99. }
  100. }
  101. void CActivityFactory::addChildQuery(unsigned id, ActivityArray *childQuery)
  102. {
  103. childQueries.append(*childQuery);
  104. childQueryIndexes.append(id);
  105. }
  106. ActivityArray *CActivityFactory::queryChildQuery(unsigned idx, unsigned &id)
  107. {
  108. if (childQueries.isItem(idx))
  109. {
  110. id = childQueryIndexes.item(idx);
  111. return &childQueries.item(idx);
  112. }
  113. id = 0;
  114. return NULL;
  115. }
  116. class CSlaveActivityFactory : public CActivityFactory, implements ISlaveActivityFactory
  117. {
  118. public:
  119. IMPLEMENT_IINTERFACE
  120. CSlaveActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  121. : CActivityFactory(_graphNode.getPropInt("@id", 0), _subgraphId, _queryFactory, _helperFactory, getActivityKind(_graphNode))
  122. {
  123. }
  124. virtual IQueryFactory &queryQueryFactory() const
  125. {
  126. return CActivityFactory::queryQueryFactory();
  127. }
  128. void addChildQuery(unsigned id, ActivityArray *childQuery)
  129. {
  130. CActivityFactory::addChildQuery(id, childQuery);
  131. }
  132. StringBuffer &toString(StringBuffer &ret) const
  133. {
  134. return ret.appendf("%p", this);
  135. }
  136. const char *queryQueryName() const
  137. {
  138. return queryFactory.queryQueryName();
  139. }
  140. virtual ActivityArray *queryChildQuery(unsigned idx, unsigned &id)
  141. {
  142. return CActivityFactory::queryChildQuery(idx, id);
  143. }
  144. virtual void mergeStats(const CRuntimeStatisticCollection &from) const
  145. {
  146. CActivityFactory::mergeStats(from);
  147. }
  148. virtual unsigned queryId() const
  149. {
  150. return CActivityFactory::queryId();
  151. }
  152. virtual ThorActivityKind getKind() const
  153. {
  154. return CActivityFactory::getKind();
  155. }
  156. virtual void getEdgeProgressInfo(unsigned idx, IPropertyTree &edge) const
  157. {
  158. CActivityFactory::getEdgeProgressInfo(idx, edge);
  159. }
  160. virtual void getNodeProgressInfo(IPropertyTree &node) const
  161. {
  162. CActivityFactory::getNodeProgressInfo(node);
  163. }
  164. virtual void resetNodeProgressInfo()
  165. {
  166. CActivityFactory::resetNodeProgressInfo();
  167. }
  168. virtual void getActivityMetrics(StringBuffer &reply) const
  169. {
  170. CActivityFactory::getActivityMetrics(reply);
  171. }
  172. IRoxieSlaveContext *createSlaveContext(const SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const
  173. {
  174. return queryFactory.createSlaveContext(logctx, packet, childQueries.length()!=0);
  175. }
  176. virtual void getXrefInfo(IPropertyTree &reply, const IRoxieContextLogger &logctx) const
  177. {
  178. if (datafile)
  179. addXrefFileInfo(reply, datafile);
  180. }
  181. void createChildQueries(IRoxieSlaveContext *ctx, IArrayOf<IActivityGraph> &childGraphs, IHThorArg *colocalArg, IProbeManager *_probeManager, IRoxieSlaveContext *queryContext, const SlaveContextLogger &logctx) const
  182. {
  183. if (childQueries.length())
  184. {
  185. ForEachItemIn(idx, childQueries)
  186. {
  187. if (!_probeManager) // MORE - the probeAllRows is a hack!
  188. _probeManager = queryContext->queryProbeManager();
  189. IActivityGraph *childGraph = createActivityGraph(ctx, NULL, childQueryIndexes.item(idx), childQueries.item(idx), NULL, _probeManager, logctx, 1); // MORE - the parent is wrong!
  190. childGraphs.append(*childGraph);
  191. queryContext->noteChildGraph(childQueryIndexes.item(idx), childGraph);
  192. childGraph->onCreate(colocalArg); //NB: onCreate() on helper for activities in child graph are delayed, otherwise this would go wrong.
  193. }
  194. }
  195. }
  196. Owned<const IResolvedFile> datafile;
  197. protected:
  198. static IPropertyTree *queryStatsNode(IPropertyTree *parent, const char *xpath)
  199. {
  200. StringBuffer levelx;
  201. const char *sep = strchr(xpath, '/');
  202. if (sep)
  203. levelx.append(sep-xpath, xpath);
  204. else
  205. levelx.append(xpath);
  206. IPropertyTree *child = parent->queryPropTree(levelx);
  207. if (!child)
  208. {
  209. const char *id = strchr(levelx, '[');
  210. if (!id)
  211. {
  212. child = parent->addPropTree(levelx);
  213. }
  214. else
  215. {
  216. StringBuffer elem;
  217. elem.append(id-levelx, levelx);
  218. child = parent->addPropTree(elem);
  219. for (;;)
  220. {
  221. StringBuffer attr, val;
  222. id++;
  223. while (*id != '=')
  224. attr.append(*id++);
  225. id++;
  226. char qu = *id++;
  227. while (*id != qu)
  228. val.append(*id++);
  229. child->setProp(attr, val);
  230. id++;
  231. if (*id == ']')
  232. {
  233. if (id[1]!='[')
  234. break;
  235. id++;
  236. }
  237. else
  238. throwUnexpected();
  239. }
  240. }
  241. }
  242. if (sep)
  243. return queryStatsNode(child, sep+1);
  244. else
  245. return child;
  246. }
  247. };
  248. //================================================================================================
  249. class CRoxieSlaveActivity : implements CInterfaceOf<IRoxieSlaveActivity>, implements ICodeContext
  250. {
  251. protected:
  252. SlaveContextLogger &logctx;
  253. Linked<IRoxieQueryPacket> packet;
  254. mutable Owned<IRoxieSlaveContext> queryContext; // bit of a hack but easier than changing the ICodeContext callback interface to remove const
  255. const CSlaveActivityFactory *basefactory;
  256. IArrayOf<IActivityGraph> childGraphs;
  257. IHThorArg *basehelper;
  258. PartNoType lastPartNo;
  259. MemoryBuffer serializedCreate;
  260. MemoryBuffer resentInfo;
  261. CachedOutputMetaData meta;
  262. Owned<IOutputRowSerializer> serializer;
  263. Owned<IEngineRowAllocator> rowAllocator;
  264. #ifdef _DEBUG
  265. Owned<IProbeManager> probeManager;
  266. #endif
  267. bool aborted;
  268. bool resent;
  269. bool isOpt;
  270. bool variableFileName;
  271. RecordTranslationMode allowFieldTranslation;
  272. Owned<const IResolvedFile> varFileInfo;
  273. virtual void setPartNo(bool filechanged) = 0;
  274. inline void checkPartChanged(PartNoType &newPart)
  275. {
  276. if (newPart.partNo!=lastPartNo.partNo || newPart.fileNo!=lastPartNo.fileNo)
  277. {
  278. lastPartNo.partNo = newPart.partNo;
  279. bool filechanged = lastPartNo.fileNo != newPart.fileNo;
  280. lastPartNo.fileNo = newPart.fileNo;
  281. setPartNo(filechanged);
  282. }
  283. }
  284. virtual bool needsRowAllocator()
  285. {
  286. return meta.needsSerializeDisk() || meta.isVariableSize();
  287. }
  288. virtual void onCreate()
  289. {
  290. queryContext.setown(basefactory->createSlaveContext(logctx, packet));
  291. #ifdef _DEBUG
  292. // MORE - need to consider debugging....
  293. if (probeAllRows)
  294. probeManager.setown(createProbeManager());
  295. basefactory->createChildQueries(queryContext, childGraphs, basehelper, probeManager, queryContext, logctx);
  296. #else
  297. basefactory->createChildQueries(queryContext, childGraphs, basehelper, NULL, queryContext, logctx);
  298. #endif
  299. if (meta.needsSerializeDisk())
  300. serializer.setown(meta.createDiskSerializer(queryContext->queryCodeContext(), basefactory->queryId()));
  301. if (needsRowAllocator())
  302. rowAllocator.setown(getRowAllocator(meta.queryOriginal(), basefactory->queryId()));
  303. unsigned parentExtractSize;
  304. serializedCreate.read(parentExtractSize);
  305. const byte * parentExtract = serializedCreate.readDirect(parentExtractSize);
  306. basehelper->onCreate(this, NULL, &serializedCreate);
  307. basehelper->onStart(parentExtract, &serializedCreate);
  308. deserializeExtra(serializedCreate);
  309. if (variableFileName) // note - in keyed join with dependent index case, kj itself won't have variableFileName but indexread might
  310. {
  311. CDateTime cacheDate(serializedCreate);
  312. unsigned checksum;
  313. serializedCreate.read(checksum);
  314. OwnedRoxieString fname(queryDynamicFileName());
  315. varFileInfo.setown(querySlaveDynamicFileCache()->lookupDynamicFile(logctx, fname, cacheDate, checksum, &packet->queryHeader(), isOpt, true));
  316. setVariableFileInfo();
  317. }
  318. }
  319. virtual void deserializeExtra(MemoryBuffer &out)
  320. {
  321. }
  322. CRoxieSlaveActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CSlaveActivityFactory *_factory)
  323. : logctx(_logctx), packet(_packet), basefactory(_factory)
  324. {
  325. allowFieldTranslation = _factory->getEnableFieldTranslation();
  326. resent = packet->getContinuationLength() != 0;
  327. serializedCreate.setBuffer(packet->getContextLength(), (void *) packet->queryContextData(), false);
  328. if (resent)
  329. resentInfo.setBuffer(packet->getContinuationLength(), (void *) packet->queryContinuationData(), false);
  330. basehelper = _hFactory();
  331. lastPartNo.partNo = 0xffff;
  332. lastPartNo.fileNo = 0xffff;
  333. aborted = false;
  334. isOpt = false;
  335. variableFileName = false;
  336. meta.set(basehelper->queryOutputMeta());
  337. }
  338. ~CRoxieSlaveActivity()
  339. {
  340. ::Release(basehelper);
  341. }
  342. virtual void beforeDispose() override
  343. {
  344. CRuntimeStatisticCollection merged(allStatistics);
  345. logctx.gatherStats(merged);
  346. basefactory->mergeStats(merged);
  347. }
  348. public:
  349. IMPLEMENT_IINTERFACE_USING(CInterfaceOf<IRoxieSlaveActivity>)
  350. virtual const char *queryDynamicFileName() const = 0;
  351. virtual void setVariableFileInfo() = 0;
  352. virtual IIndexReadActivityInfo *queryIndexReadActivity() { throwUnexpected(); } // should only be called for index activity
  353. virtual unsigned queryId()
  354. {
  355. return basefactory->queryId();
  356. }
  357. virtual bool check()
  358. {
  359. Owned<IMessagePacker> output = ROQ->createOutputStream(packet->queryHeader(), false, logctx);
  360. doCheck(output);
  361. output->flush(true);
  362. return true;
  363. }
  364. virtual void doCheck(IMessagePacker *output)
  365. {
  366. // MORE - unsophisticated default - if this approach seems fruitful then we can add something more thorough
  367. void *recBuffer = output->getBuffer(sizeof(bool), false);
  368. bool ret = false;
  369. memcpy(recBuffer, &ret, sizeof(bool));
  370. }
  371. virtual void abort()
  372. {
  373. if (logctx.queryTraceLevel() > 2)
  374. {
  375. StringBuffer s;
  376. logctx.CTXLOG("Aborting running activity: %s", packet->queryHeader().toString(s).str());
  377. }
  378. aborted = true;
  379. logctx.abort();
  380. if (queryContext)
  381. {
  382. Owned<IException> E = MakeStringException(ROXIE_ABORT_ERROR, "Roxie server requested abort for running activity");
  383. queryContext->notifyAbort(E);
  384. }
  385. }
  386. virtual IRoxieQueryPacket *queryPacket() const
  387. {
  388. return packet;
  389. }
  390. void limitExceeded(bool keyed = false)
  391. {
  392. RoxiePacketHeader &header = packet->queryHeader();
  393. StringBuffer s;
  394. logctx.CTXLOG("%sLIMIT EXCEEDED: %s", keyed ? "KEYED " : "", header.toString(s).str());
  395. header.activityId = keyed ? ROXIE_KEYEDLIMIT_EXCEEDED : ROXIE_LIMIT_EXCEEDED;
  396. Owned<IMessagePacker> output = ROQ->createOutputStream(header, false, logctx);
  397. output->flush(true);
  398. aborted = true;
  399. logctx.abort();
  400. }
  401. virtual IThorChildGraph * resolveChildQuery(__int64 activityId, IHThorArg * colocal)
  402. {
  403. assertex(colocal == basehelper);
  404. return queryContext->queryCodeContext()->resolveChildQuery(activityId, colocal);
  405. }
  406. size32_t serializeRow(IMessagePacker *output, const void *unserialized) const
  407. {
  408. return ::serializeRow(serializer, output, unserialized);
  409. }
  410. virtual const char *loadResource(unsigned id)
  411. {
  412. return queryContext->queryCodeContext()->loadResource(id);
  413. }
  414. // Sets should not happen - they can only happen in the main Roxie server context
  415. virtual void setResultBool(const char *name, unsigned sequence, bool value) { throwUnexpected(); }
  416. virtual void setResultData(const char *name, unsigned sequence, int len, const void * data) { throwUnexpected(); }
  417. virtual void setResultDecimal(const char * stepname, unsigned sequence, int len, int precision, bool isSigned, const void *val) { throwUnexpected(); }
  418. virtual void setResultInt(const char *name, unsigned sequence, __int64 value, unsigned size) { throwUnexpected(); }
  419. virtual void setResultRaw(const char *name, unsigned sequence, int len, const void * data) { throwUnexpected(); }
  420. virtual void setResultReal(const char * stepname, unsigned sequence, double value) { throwUnexpected(); }
  421. virtual void setResultSet(const char *name, unsigned sequence, bool isAll, size32_t len, const void * data, ISetToXmlTransformer * transformer) { throwUnexpected(); }
  422. virtual void setResultString(const char *name, unsigned sequence, int len, const char * str) { throwUnexpected(); }
  423. virtual void setResultUInt(const char *name, unsigned sequence, unsigned __int64 value, unsigned size) { throwUnexpected(); }
  424. virtual void setResultUnicode(const char *name, unsigned sequence, int len, UChar const * str) { throwUnexpected(); }
  425. virtual void setResultVarString(const char * name, unsigned sequence, const char * value) { throwUnexpected(); }
  426. virtual void setResultVarUnicode(const char * name, unsigned sequence, UChar const * value) { throwUnexpected(); }
  427. // Some gets are allowed though (e.g. for ONCE values)
  428. virtual bool getResultBool(const char * name, unsigned sequence)
  429. {
  430. return queryContext->queryCodeContext()->getResultBool(name, sequence);
  431. }
  432. virtual void getResultData(unsigned & tlen, void * & tgt, const char * name, unsigned sequence)
  433. {
  434. queryContext->queryCodeContext()->getResultData(tlen, tgt, name, sequence);
  435. }
  436. virtual void getResultDecimal(unsigned tlen, int precision, bool isSigned, void * tgt, const char * stepname, unsigned sequence)
  437. {
  438. queryContext->queryCodeContext()->getResultDecimal(tlen, precision, isSigned, tgt, stepname, sequence);
  439. }
  440. virtual void getResultRaw(unsigned & tlen, void * & tgt, const char * name, unsigned sequence, IXmlToRowTransformer * xmlTransformer, ICsvToRowTransformer * csvTransformer)
  441. {
  442. queryContext->queryCodeContext()->getResultRaw(tlen, tgt, name, sequence, xmlTransformer, csvTransformer);
  443. }
  444. virtual void getResultSet(bool & isAll, size32_t & tlen, void * & tgt, const char * name, unsigned sequence, IXmlToRowTransformer * xmlTransformer, ICsvToRowTransformer * csvTransformer)
  445. {
  446. queryContext->queryCodeContext()->getResultSet(isAll, tlen, tgt, name, sequence, xmlTransformer, csvTransformer);
  447. }
  448. virtual __int64 getResultInt(const char * name, unsigned sequence)
  449. {
  450. return queryContext->queryCodeContext()->getResultInt(name, sequence);
  451. }
  452. virtual double getResultReal(const char * name, unsigned sequence)
  453. {
  454. return queryContext->queryCodeContext()->getResultReal(name, sequence);
  455. }
  456. virtual void getResultString(unsigned & tlen, char * & tgt, const char * name, unsigned sequence)
  457. {
  458. queryContext->queryCodeContext()->getResultString(tlen, tgt, name, sequence);
  459. }
  460. virtual void getResultStringF(unsigned tlen, char * tgt, const char * name, unsigned sequence)
  461. {
  462. queryContext->queryCodeContext()->getResultStringF(tlen, tgt, name, sequence);
  463. }
  464. virtual void getResultUnicode(unsigned & tlen, UChar * & tgt, const char * name, unsigned sequence)
  465. {
  466. queryContext->queryCodeContext()->getResultUnicode(tlen, tgt, name, sequence);
  467. }
  468. virtual char *getResultVarString(const char * name, unsigned sequence)
  469. {
  470. return queryContext->queryCodeContext()->getResultVarString(name, sequence);
  471. }
  472. virtual UChar *getResultVarUnicode(const char * name, unsigned sequence)
  473. {
  474. return queryContext->queryCodeContext()->getResultVarUnicode(name, sequence);
  475. }
  476. virtual void getResultRowset(size32_t & tcount, const byte * * & tgt, const char * name, unsigned sequence, IEngineRowAllocator * _rowAllocator, bool isGrouped, IXmlToRowTransformer * xmlTransformer, ICsvToRowTransformer * csvTransformer) override
  477. {
  478. return queryContext->queryCodeContext()->getResultRowset(tcount, tgt, name, sequence, _rowAllocator, isGrouped, xmlTransformer, csvTransformer);
  479. }
  480. virtual void getResultDictionary(size32_t & tcount, const byte * * & tgt, IEngineRowAllocator * _rowAllocator, const char * name, unsigned sequence, IXmlToRowTransformer * xmlTransformer, ICsvToRowTransformer * csvTransformer, IHThorHashLookupInfo * hasher) override
  481. {
  482. return queryContext->queryCodeContext()->getResultDictionary(tcount, tgt, _rowAllocator, name, sequence, xmlTransformer, csvTransformer, hasher);
  483. }
  484. virtual unsigned getResultHash(const char * name, unsigned sequence) { throwUnexpected(); }
  485. virtual unsigned getExternalResultHash(const char * wuid, const char * name, unsigned sequence) { throwUnexpected(); }
  486. virtual ISectionTimer * registerTimer(unsigned activityId, const char * name)
  487. {
  488. return queryNullSectionTimer();
  489. }
  490. // Not yet thought about these....
  491. virtual char *getWuid() { throwUnexpected(); } // caller frees return string.
  492. virtual void getExternalResultRaw(unsigned & tlen, void * & tgt, const char * wuid, const char * stepname, unsigned sequence, IXmlToRowTransformer * xmlTransformer, ICsvToRowTransformer * csvTransformer) { throwUnexpected(); } // shouldn't really be here, but it broke thor.
  493. virtual void executeGraph(const char * graphName, bool realThor, size32_t parentExtractSize, const void * parentExtract) { throwUnexpected(); }
  494. virtual unsigned __int64 getDatasetHash(const char * name, unsigned __int64 hash) { throwUnexpected(); return 0; }
  495. virtual char * getExpandLogicalName(const char * logicalName) { throwUnexpected(); }
  496. virtual void addWuException(const char * text, unsigned code, unsigned severity, const char * source) { throwUnexpected(); }
  497. virtual void addWuAssertFailure(unsigned code, const char * text, const char * filename, unsigned lineno, unsigned column, bool isAbort) { throwUnexpected(); }
  498. virtual IUserDescriptor *queryUserDescriptor() { throwUnexpected(); }
  499. virtual unsigned getNodes() { throwUnexpected(); }
  500. virtual unsigned getNodeNum() { throwUnexpected(); }
  501. virtual char *getFilePart(const char *logicalPart, bool create=false) { throwUnexpected(); } // caller frees return string.
  502. virtual unsigned __int64 getFileOffset(const char *logicalPart) { throwUnexpected(); }
  503. virtual IDistributedFileTransaction *querySuperFileTransaction() { throwUnexpected(); }
  504. virtual char *getJobName() { throwUnexpected(); } // caller frees return string.
  505. virtual char *getJobOwner() { throwUnexpected(); } // caller frees return string.
  506. virtual char *getClusterName() { throwUnexpected(); } // caller frees return str.
  507. virtual char *getGroupName() { throwUnexpected(); } // caller frees return string.
  508. virtual char * queryIndexMetaData(char const * lfn, char const * xpath) { throwUnexpected(); }
  509. virtual char *getDaliServers() { return queryContext->queryCodeContext()->getDaliServers(); }
  510. // The below are called on Roxie server and passed in context
  511. virtual unsigned getPriority() const { throwUnexpected(); }
  512. virtual char *getPlatform() { throwUnexpected(); }
  513. virtual char *getEnv(const char *name, const char *defaultValue) const { throwUnexpected(); }
  514. virtual char *getOS() { throwUnexpected(); }
  515. virtual unsigned logString(const char *text) const
  516. {
  517. if (text && *text)
  518. {
  519. logctx.CTXLOG("USER: %s", text);
  520. return strlen(text);
  521. }
  522. else
  523. return 0;
  524. }
  525. virtual const IContextLogger &queryContextLogger() const
  526. {
  527. return logctx;
  528. }
  529. virtual IEngineRowAllocator * getRowAllocator(IOutputMetaData * meta, unsigned activityId) const
  530. {
  531. return queryContext->queryCodeContext()->getRowAllocator(meta, activityId);
  532. }
  533. virtual IEngineRowAllocator * getRowAllocatorEx(IOutputMetaData * meta, unsigned activityId, unsigned heapFlags) const
  534. {
  535. return queryContext->queryCodeContext()->getRowAllocatorEx(meta, activityId, heapFlags);
  536. }
  537. virtual const char *cloneVString(const char *str) const
  538. {
  539. return queryContext->queryCodeContext()->cloneVString(str);
  540. }
  541. virtual const char *cloneVString(size32_t len, const char *str) const
  542. {
  543. return queryContext->queryCodeContext()->cloneVString(len, str);
  544. }
  545. virtual void getRowXML(size32_t & lenResult, char * & result, IOutputMetaData & info, const void * row, unsigned flags)
  546. {
  547. convertRowToXML(lenResult, result, info, row, flags);
  548. }
  549. virtual void getRowJSON(size32_t & lenResult, char * & result, IOutputMetaData & info, const void * row, unsigned flags)
  550. {
  551. convertRowToJSON(lenResult, result, info, row, flags);
  552. }
  553. const void * fromXml(IEngineRowAllocator * rowAllocator, size32_t len, const char * utf8, IXmlToRowTransformer * xmlTransformer, bool stripWhitespace)
  554. {
  555. return createRowFromXml(rowAllocator, len, utf8, xmlTransformer, stripWhitespace);
  556. }
  557. const void * fromJson(IEngineRowAllocator * rowAllocator, size32_t len, const char * utf8, IXmlToRowTransformer * xmlTransformer, bool stripWhitespace)
  558. {
  559. return createRowFromJson(rowAllocator, len, utf8, xmlTransformer, stripWhitespace);
  560. }
  561. virtual IEngineContext *queryEngineContext() { return NULL; }
  562. virtual IWorkUnit *updateWorkUnit() const { throwUnexpected(); }
  563. virtual unsigned getGraphLoopCounter() const override { return queryContext->queryCodeContext()->getGraphLoopCounter(); }
  564. virtual IEclGraphResults * resolveLocalQuery(__int64 activityId) override { return NULL; } // Only want to do this on the server
  565. virtual IDebuggableContext *queryDebugContext() const override { return queryContext->queryCodeContext()->queryDebugContext(); }
  566. };
  567. //================================================================================================
  568. class OptimizedRowBuilder : public ARowBuilder, public CInterface
  569. {
  570. public:
  571. OptimizedRowBuilder(IEngineRowAllocator * _rowAllocator, const CachedOutputMetaData & _meta, IMessagePacker * _output, IOutputRowSerializer * _serializer)
  572. : dynamicBuilder(_rowAllocator, false), meta(_meta), serializer(_serializer), output(_output)
  573. {
  574. useDynamic = serializer != NULL || meta.isVariableSize();
  575. }
  576. IMPLEMENT_IINTERFACE
  577. virtual IEngineRowAllocator *queryAllocator() const
  578. {
  579. return dynamicBuilder.queryAllocator();
  580. }
  581. virtual byte * createSelf()
  582. {
  583. if (useDynamic)
  584. {
  585. dynamicBuilder.ensureRow();
  586. self = dynamicBuilder.getSelf();
  587. }
  588. else
  589. self = static_cast<byte *>(output->getBuffer(meta.getFixedSize(), false));
  590. return self;
  591. }
  592. virtual byte * ensureCapacity(size32_t required, const char * fieldName)
  593. {
  594. if (useDynamic)
  595. {
  596. self = dynamicBuilder.ensureCapacity(required, fieldName);
  597. return static_cast<byte *>(self);
  598. }
  599. else
  600. {
  601. size32_t fixedLength = meta.getFixedSize();
  602. if (required <= fixedLength)
  603. return static_cast<byte *>(self);
  604. // This should never happen!
  605. rtlReportFieldOverflow(required, fixedLength, fieldName);
  606. return NULL;
  607. }
  608. }
  609. virtual void reportMissingRow() const
  610. {
  611. throw MakeStringException(MSGAUD_user, 1000, "OptimizedRowBuilder::row() is NULL");
  612. }
  613. inline void ensureRow()
  614. {
  615. if (!self)
  616. createSelf();
  617. }
  618. inline void clear()
  619. {
  620. if (useDynamic)
  621. dynamicBuilder.clear();
  622. self = NULL;
  623. }
  624. size_t writeToOutput(size32_t transformedSize, bool outputIfEmpty)
  625. {
  626. size32_t outputSize = transformedSize;
  627. if (transformedSize || outputIfEmpty)
  628. {
  629. if (useDynamic)
  630. {
  631. OwnedConstRoxieRow result = dynamicBuilder.finalizeRowClear(transformedSize);
  632. if (serializer)
  633. outputSize = serializeRow(serializer, output, result);
  634. else
  635. {
  636. self = static_cast<byte *>(output->getBuffer(transformedSize, true));
  637. memcpy(self, result, transformedSize);
  638. output->putBuffer(self, transformedSize, true);
  639. }
  640. }
  641. else
  642. {
  643. output->putBuffer(self, transformedSize, false);
  644. }
  645. }
  646. clear();
  647. return outputSize;
  648. }
  649. private:
  650. RtlDynamicRowBuilder dynamicBuilder;
  651. const CachedOutputMetaData & meta;
  652. IMessagePacker * output;
  653. IOutputRowSerializer * serializer;
  654. bool useDynamic;
  655. };
  656. class OptimizedKJRowBuilder : public ARowBuilder, public CInterface
  657. {
  658. // Rules are different enough that we can't easily derive from OptimizedRowBuilder
  659. public:
  660. IMPLEMENT_IINTERFACE;
  661. OptimizedKJRowBuilder(IEngineRowAllocator * _rowAllocator, const CachedOutputMetaData & _meta, IMessagePacker * _output)
  662. : dynamicBuilder(_rowAllocator, false), meta(_meta), output(_output)
  663. {
  664. useDynamic = meta.isVariableSize();
  665. }
  666. virtual IEngineRowAllocator *queryAllocator() const
  667. {
  668. return dynamicBuilder.queryAllocator();
  669. }
  670. virtual byte * createSelf()
  671. {
  672. if (useDynamic)
  673. {
  674. dynamicBuilder.ensureRow();
  675. self = dynamicBuilder.getSelf();
  676. return self;
  677. }
  678. else
  679. {
  680. self = static_cast<byte *>(output->getBuffer(KEYEDJOIN_RECORD_SIZE(meta.getFixedSize()), true)) + KEYEDJOIN_RECORD_SIZE(0);
  681. return self ;
  682. }
  683. }
  684. virtual byte * ensureCapacity(size32_t required, const char * fieldName)
  685. {
  686. if (useDynamic)
  687. {
  688. self = dynamicBuilder.ensureCapacity(required, fieldName);
  689. return self;
  690. }
  691. else
  692. {
  693. size32_t fixedLength = meta.getFixedSize();
  694. if (required <= fixedLength)
  695. return self;
  696. rtlReportFieldOverflow(required, fixedLength, fieldName);
  697. return NULL;
  698. }
  699. }
  700. virtual void reportMissingRow() const
  701. {
  702. throw MakeStringException(MSGAUD_user, 1000, "OptimizedKJRowBuilder::row() is NULL");
  703. }
  704. inline void ensureRow()
  705. {
  706. if (!self)
  707. createSelf();
  708. }
  709. void writeToOutput(size32_t transformedSize, offset_t recptr, CJoinGroup *jg, unsigned short partNo)
  710. {
  711. KeyedJoinHeader *rec;
  712. if (useDynamic)
  713. {
  714. OwnedConstRoxieRow result = dynamicBuilder.finalizeRowClear(transformedSize);
  715. rec = (KeyedJoinHeader *) (output->getBuffer(KEYEDJOIN_RECORD_SIZE(transformedSize), true));
  716. memcpy(&rec->rhsdata, result, transformedSize);
  717. }
  718. else
  719. {
  720. rec = (KeyedJoinHeader *)(self - KEYEDJOIN_RECORD_SIZE(0));
  721. }
  722. rec->fpos = recptr;
  723. rec->thisGroup = jg;
  724. rec->partNo = partNo;
  725. output->putBuffer(rec, KEYEDJOIN_RECORD_SIZE(transformedSize), true);
  726. self = NULL;
  727. }
  728. private:
  729. RtlDynamicRowBuilder dynamicBuilder;
  730. const CachedOutputMetaData & meta;
  731. IMessagePacker * output;
  732. bool useDynamic;
  733. };
  734. //================================================================================================
  735. class CRoxieDiskReadBaseActivity : public CRoxieSlaveActivity, implements IIndexReadContext//, implements IDiskReadActivity
  736. {
  737. protected:
  738. IHThorDiskReadBaseArg *helper;
  739. unsigned processed;
  740. unsigned parallelPartNo;
  741. unsigned numParallel;
  742. bool isGrouped = false;
  743. bool forceUnkeyed;
  744. offset_t readPos;
  745. CachedOutputMetaData diskSize;
  746. ScoredRowFilter postFilter;
  747. Owned<IDirectReader> reader;
  748. Linked<IInMemoryIndexManager> manager;
  749. Linked<ITranslatorSet> translators;
  750. Owned<IInMemoryFileProcessor> processor;
  751. CriticalSection pcrit;
  752. public:
  753. CRoxieDiskReadBaseActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CSlaveActivityFactory *_aFactory,
  754. IInMemoryIndexManager *_manager,
  755. ITranslatorSet *_translators,
  756. unsigned _parallelPartNo, unsigned _numParallel, bool _forceUnkeyed)
  757. : CRoxieSlaveActivity(_logctx, _packet, _hFactory, _aFactory),
  758. manager(_manager),
  759. translators(_translators),
  760. parallelPartNo(_parallelPartNo),
  761. numParallel(_numParallel),
  762. forceUnkeyed(_forceUnkeyed)
  763. {
  764. helper = (IHThorDiskReadBaseArg *) basehelper;
  765. variableFileName = allFilesDynamic || basefactory->queryQueryFactory().isDynamic() || ((helper->getFlags() & (TDXvarfilename|TDXdynamicfilename)) != 0);
  766. isOpt = (helper->getFlags() & TDRoptional) != 0;
  767. diskSize.set(helper->queryProjectedDiskRecordSize()->querySerializedDiskMeta());
  768. isGrouped = diskSize.isGrouped();
  769. processed = 0;
  770. readPos = 0;
  771. if (resent)
  772. {
  773. }
  774. }
  775. virtual void onCreate()
  776. {
  777. CRoxieSlaveActivity::onCreate();
  778. helper->createSegmentMonitors(this);
  779. const IKeyTranslator *keyTranslator = translators->queryKeyTranslator(0); // any part would do - in-memory requires all actuals to have same layout
  780. if (keyTranslator)
  781. keyTranslator->translate(postFilter);
  782. if (resent)
  783. {
  784. bool usedKey;
  785. resentInfo.read(processed);
  786. resentInfo.read(usedKey);
  787. if (usedKey)
  788. reader.setown(manager->selectKey(resentInfo, postFilter, translators));
  789. else
  790. {
  791. resentInfo.read(readPos);
  792. reader.setown(manager->createReader(postFilter, isGrouped, readPos, parallelPartNo, numParallel, translators));
  793. }
  794. assertex(resentInfo.remaining() == 0);
  795. }
  796. else
  797. {
  798. if (!forceUnkeyed && !isGrouped)
  799. reader.setown(manager->selectKey(postFilter, translators, logctx));
  800. if (!reader)
  801. reader.setown(manager->createReader(postFilter, isGrouped, readPos, parallelPartNo, numParallel, translators));
  802. }
  803. }
  804. virtual const char *queryDynamicFileName() const
  805. {
  806. return helper->getFileName();
  807. }
  808. virtual void setVariableFileInfo()
  809. {
  810. unsigned channel = packet->queryHeader().channel;
  811. unsigned formatCrc = basefactory->getFormatCrc(helper->getFormatCrc());
  812. translators.setown(varFileInfo->getTranslators(formatCrc, helper->queryProjectedDiskRecordSize(), helper->queryDiskRecordSize(), basefactory->getEnableFieldTranslation(), false)); // MORE - FormatCRC may be wrong here. Needs to be crc of projected not expected
  813. manager.setown(varFileInfo->getIndexManager(isOpt, channel, translators->queryActualLayout(0), false));
  814. }
  815. inline bool queryKeyed() const
  816. {
  817. return reader->isKeyed();
  818. }
  819. void setParallel(unsigned _partno, unsigned _numParallel)
  820. {
  821. assertex(!processor);
  822. parallelPartNo = _partno;
  823. numParallel = _numParallel;
  824. }
  825. virtual StringBuffer &toString(StringBuffer &ret) const
  826. {
  827. return ret.appendf("DiskRead %u", packet->queryHeader().activityId);
  828. }
  829. virtual void append(IKeySegmentMonitor *segment)
  830. {
  831. segment->Release();
  832. throwUnexpected();
  833. }
  834. virtual void append(FFoption option, const IFieldFilter * filter)
  835. {
  836. if (filter->isWild())
  837. filter->Release();
  838. else
  839. postFilter.addFilter(*filter);
  840. }
  841. virtual unsigned ordinality() const
  842. {
  843. throwUnexpected();
  844. }
  845. virtual IKeySegmentMonitor *item(unsigned idx) const
  846. {
  847. throwUnexpected();
  848. }
  849. virtual void abort()
  850. {
  851. CRoxieSlaveActivity::abort();
  852. CriticalBlock p(pcrit);
  853. if (processor)
  854. processor->abort();
  855. }
  856. virtual IMessagePacker *process()
  857. {
  858. MTIME_SECTION(queryActiveTimer(), "CRoxieDiskReadBaseActivity::process");
  859. diskReadStarted++;
  860. Owned<IMessagePacker> output = ROQ->createOutputStream(packet->queryHeader(), false, logctx);
  861. doProcess(output);
  862. helper->setCallback(NULL);
  863. if (aborted)
  864. return NULL;
  865. else
  866. {
  867. diskReadCompleted++;
  868. return output.getClear();
  869. }
  870. }
  871. virtual void doCheck(IMessagePacker *output)
  872. {
  873. // for in-memory diskread activities, not a lot to check. If it got this far the answer is 'true'...
  874. void *recBuffer = output->getBuffer(sizeof(bool), false);
  875. bool ret = true;
  876. memcpy(recBuffer, &ret, sizeof(bool));
  877. }
  878. virtual void doProcess(IMessagePacker * output) = 0;
  879. virtual void setPartNo(bool filechanged)
  880. {
  881. throwUnexpected();
  882. }
  883. };
  884. class CRoxieDiskBaseActivityFactory : public CSlaveActivityFactory
  885. {
  886. protected:
  887. Owned<ITranslatorSet> translators;
  888. Owned<IInMemoryIndexManager> manager;
  889. public:
  890. CRoxieDiskBaseActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  891. : CSlaveActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory)
  892. {
  893. Owned<IHThorDiskReadBaseArg> helper = (IHThorDiskReadBaseArg *) helperFactory();
  894. bool variableFileName = allFilesDynamic || queryFactory.isDynamic() || ((helper->getFlags() & (TDXvarfilename|TDXdynamicfilename)) != 0);
  895. if (!variableFileName)
  896. {
  897. bool isOpt = (helper->getFlags() & TDRoptional) != 0;
  898. OwnedRoxieString fileName(helper->getFileName());
  899. datafile.setown(_queryFactory.queryPackage().lookupFileName(fileName, isOpt, true, true, _queryFactory.queryWorkUnit(), true));
  900. if (datafile)
  901. {
  902. unsigned channel = queryFactory.queryChannel();
  903. unsigned formatCrc = getFormatCrc(helper->getFormatCrc());
  904. translators.setown(datafile->getTranslators(formatCrc, helper->queryProjectedDiskRecordSize(), helper->queryDiskRecordSize(), getEnableFieldTranslation(), false));
  905. manager.setown(datafile->getIndexManager(isOpt, channel, translators->queryActualLayout(0), _graphNode.getPropBool("att[@name=\"preload\"]/@value", false)));
  906. Owned<IPropertyTreeIterator> memKeyInfo = queryFactory.queryPackage().getInMemoryIndexInfo(_graphNode);
  907. Owned<IPropertyTree> memKeyHint;
  908. if (!memKeyInfo && _graphNode.hasProp("hint[@name=\"key\"]"))
  909. {
  910. memKeyHint.setown(createPTreeFromXMLString(_graphNode.queryProp("hint[@name=\"key\"]/@value")));
  911. memKeyInfo.setown(memKeyHint->getElements("MemIndex"));
  912. }
  913. if (memKeyInfo)
  914. {
  915. // There's a small potential flaw that I can end up with indexes from a query that was unloaded active on some nodes but not others
  916. // which could mess up continuation. I think I don't care.
  917. ForEach(*memKeyInfo)
  918. {
  919. IPropertyTree &info = memKeyInfo->query();
  920. manager->setKeyInfo(info);
  921. }
  922. }
  923. }
  924. else
  925. manager.setown(createInMemoryIndexManager(helper->queryProjectedDiskRecordSize()->queryRecordAccessor(true), true, nullptr));
  926. }
  927. }
  928. ~CRoxieDiskBaseActivityFactory()
  929. {
  930. }
  931. };
  932. //================================================================================================
  933. class CRoxieDiskReadActivity;
  934. class CRoxieCsvReadActivity;
  935. class CRoxieXmlReadActivity;
  936. IInMemoryFileProcessor *createReadRecordProcessor(CRoxieDiskReadActivity &owner, bool isGrouped, IDirectReader *reader);
  937. IInMemoryFileProcessor *createCsvRecordProcessor(CRoxieCsvReadActivity &owner, IDirectReader *reader, bool _skipHeader, const IResolvedFile *datafile, size32_t maxRowSize);
  938. IInMemoryFileProcessor *createXmlRecordProcessor(CRoxieXmlReadActivity &owner, IDirectReader *reader);
  939. class CRoxieDiskReadActivity : public CRoxieDiskReadBaseActivity
  940. {
  941. friend class ReadRecordProcessor;
  942. protected:
  943. IHThorDiskReadArg *helper;
  944. public:
  945. CRoxieDiskReadActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CSlaveActivityFactory *_aFactory,
  946. IInMemoryIndexManager *_manager, ITranslatorSet *_translators)
  947. : CRoxieDiskReadBaseActivity(_logctx, _packet, _hFactory, _aFactory, _manager, _translators, 0, 1, false)
  948. {
  949. onCreate();
  950. helper = (IHThorDiskReadArg *) basehelper;
  951. }
  952. virtual StringBuffer &toString(StringBuffer &ret) const
  953. {
  954. return ret.appendf("DiskRead %u", packet->queryHeader().activityId);
  955. }
  956. virtual void doProcess(IMessagePacker * output)
  957. {
  958. {
  959. CriticalBlock p(pcrit); // because of race with abort.
  960. processor.setown(createReadRecordProcessor(*this, isGrouped, reader));
  961. }
  962. unsigned __int64 rowLimit = helper->getRowLimit();
  963. unsigned __int64 stopAfter = helper->getChooseNLimit();
  964. processor->doQuery(output, processed, rowLimit, stopAfter);
  965. }
  966. size32_t doTransform(IMessagePacker * output, const void *src) const
  967. {
  968. OptimizedRowBuilder rowBuilder(rowAllocator, meta, output, serializer);
  969. unsigned transformedSize = helper->transform(rowBuilder, src);
  970. return rowBuilder.writeToOutput(transformedSize, false);
  971. }
  972. };
  973. class CRoxieCsvReadActivity : public CRoxieDiskReadBaseActivity
  974. {
  975. public:
  976. friend class CsvRecordProcessor;
  977. protected:
  978. IHThorCsvReadArg *helper;
  979. const IResolvedFile *datafile;
  980. size32_t maxRowSize;
  981. public:
  982. CRoxieCsvReadActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CSlaveActivityFactory *_aFactory,
  983. IInMemoryIndexManager *_manager, ITranslatorSet *_translators, const IResolvedFile *_datafile, size32_t _maxRowSize)
  984. : CRoxieDiskReadBaseActivity(_logctx, _packet, _hFactory, _aFactory, _manager, _translators, 0, 1, true), datafile(_datafile), maxRowSize(_maxRowSize)
  985. {
  986. onCreate();
  987. helper = (IHThorCsvReadArg *) basehelper;
  988. }
  989. virtual StringBuffer &toString(StringBuffer &ret) const
  990. {
  991. return ret.appendf("CsvRead %u", packet->queryHeader().activityId);
  992. }
  993. virtual void doProcess(IMessagePacker * output)
  994. {
  995. {
  996. CriticalBlock p(pcrit);
  997. processor.setown(
  998. createCsvRecordProcessor(*this,
  999. manager->createReader(postFilter, false, readPos, parallelPartNo, numParallel, translators),
  1000. packet->queryHeader().channel==1 && !resent,
  1001. varFileInfo ? varFileInfo.get() : datafile, maxRowSize));
  1002. }
  1003. unsigned __int64 rowLimit = helper->getRowLimit();
  1004. unsigned __int64 stopAfter = helper->getChooseNLimit();
  1005. processor->doQuery(output, processed, rowLimit, stopAfter);
  1006. }
  1007. size32_t doTransform(IMessagePacker * output, unsigned *srcLen, const char **src) const
  1008. {
  1009. OptimizedRowBuilder rowBuilder(rowAllocator, meta, output, serializer);
  1010. unsigned transformedSize = helper->transform(rowBuilder, srcLen, src);
  1011. return rowBuilder.writeToOutput(transformedSize, false);
  1012. }
  1013. };
  1014. class CRoxieXmlReadActivity : public CRoxieDiskReadBaseActivity
  1015. {
  1016. public:
  1017. friend class XmlRecordProcessor;
  1018. protected:
  1019. IHThorXmlReadArg *helper;
  1020. public:
  1021. CRoxieXmlReadActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CSlaveActivityFactory *_aFactory,
  1022. IInMemoryIndexManager *_manager, ITranslatorSet *_translators)
  1023. : CRoxieDiskReadBaseActivity(_logctx, _packet, _hFactory, _aFactory, _manager, _translators, 0, 1, true)
  1024. {
  1025. onCreate();
  1026. helper = (IHThorXmlReadArg *) basehelper;
  1027. }
  1028. virtual StringBuffer &toString(StringBuffer &ret) const
  1029. {
  1030. return ret.appendf("XmlRead %u", packet->queryHeader().activityId);
  1031. }
  1032. virtual void doProcess(IMessagePacker * output)
  1033. {
  1034. {
  1035. CriticalBlock p(pcrit);
  1036. processor.setown(createXmlRecordProcessor(*this, manager->createReader(postFilter, false, readPos, parallelPartNo, numParallel, translators)));
  1037. }
  1038. unsigned __int64 rowLimit = helper->getRowLimit();
  1039. unsigned __int64 stopAfter = helper->getChooseNLimit();
  1040. processor->doQuery(output, processed, rowLimit, stopAfter);
  1041. }
  1042. size32_t doTransform(IMessagePacker * output, IXmlToRowTransformer *rowTransformer, IColumnProvider *lastMatch, IThorDiskCallback *callback) const
  1043. {
  1044. OptimizedRowBuilder rowBuilder(rowAllocator, meta, output, serializer);
  1045. unsigned transformedSize = rowTransformer->transform(rowBuilder, lastMatch, callback);
  1046. return rowBuilder.writeToOutput(transformedSize, false);
  1047. }
  1048. };
  1049. class CRoxieDiskReadActivityFactory : public CRoxieDiskBaseActivityFactory
  1050. {
  1051. public:
  1052. CRoxieDiskReadActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  1053. : CRoxieDiskBaseActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory)
  1054. {
  1055. }
  1056. virtual IRoxieSlaveActivity *createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const
  1057. {
  1058. return new CRoxieDiskReadActivity(logctx, packet, helperFactory, this, manager, translators);
  1059. }
  1060. virtual StringBuffer &toString(StringBuffer &s) const
  1061. {
  1062. return CSlaveActivityFactory::toString(s.append("DiskRead "));
  1063. }
  1064. };
  1065. class CRoxieCsvReadActivityFactory : public CRoxieDiskBaseActivityFactory
  1066. {
  1067. size32_t maxRowSize;
  1068. public:
  1069. CRoxieCsvReadActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  1070. : CRoxieDiskBaseActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory)
  1071. {
  1072. maxRowSize = defaultMaxCsvRowSize * 1024 * 1024;
  1073. IConstWorkUnit *workunit = _queryFactory.queryWorkUnit();
  1074. if (workunit)
  1075. maxRowSize = workunit->getDebugValueInt(OPT_MAXCSVROWSIZE, defaultMaxCsvRowSize) * 1024 * 1024;
  1076. }
  1077. virtual IRoxieSlaveActivity *createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const
  1078. {
  1079. return new CRoxieCsvReadActivity(logctx, packet, helperFactory, this, manager, translators, datafile, maxRowSize);
  1080. }
  1081. virtual StringBuffer &toString(StringBuffer &s) const
  1082. {
  1083. return CSlaveActivityFactory::toString(s.append("DiskRead "));
  1084. }
  1085. };
  1086. class CRoxieXmlReadActivityFactory : public CRoxieDiskBaseActivityFactory
  1087. {
  1088. public:
  1089. CRoxieXmlReadActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  1090. : CRoxieDiskBaseActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory)
  1091. {
  1092. }
  1093. virtual IRoxieSlaveActivity *createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const
  1094. {
  1095. return new CRoxieXmlReadActivity(logctx, packet, helperFactory, this, manager, translators);
  1096. }
  1097. virtual StringBuffer &toString(StringBuffer &s) const
  1098. {
  1099. return CSlaveActivityFactory::toString(s.append("DiskRead "));
  1100. }
  1101. };
  1102. //================================================================================================
  1103. class RecordProcessor : implements IInMemoryFileProcessor, public CInterface
  1104. {
  1105. protected:
  1106. bool aborted = false;
  1107. IDirectReader *reader;
  1108. public:
  1109. IMPLEMENT_IINTERFACE
  1110. RecordProcessor(IDirectReader *_reader) : reader(_reader) {}
  1111. virtual void abort()
  1112. {
  1113. aborted = true;
  1114. }
  1115. };
  1116. //================================================================================================
  1117. // Disk read activity
  1118. class ReadRecordProcessor : public RecordProcessor
  1119. {
  1120. public:
  1121. ReadRecordProcessor(CRoxieDiskReadActivity &_owner, bool _isGrouped, IDirectReader *_reader)
  1122. : RecordProcessor(_reader), owner(_owner), isGrouped(_isGrouped)
  1123. {
  1124. helper = _owner.helper;
  1125. }
  1126. virtual void doQuery(IMessagePacker *output, unsigned processed, unsigned __int64 rowLimit, unsigned __int64 stopAfter)
  1127. {
  1128. unsigned totalSizeSent = 0;
  1129. helper->setCallback(reader);
  1130. unsigned lastGroupProcessed = processed;
  1131. bool someInGroup = false;
  1132. while (!aborted)
  1133. {
  1134. // This loop is the inner loop for memory diskreads - so keep it efficient!
  1135. const byte *nextRec = reader->nextRow();
  1136. if (nextRec)
  1137. {
  1138. someInGroup = true;
  1139. size32_t transformedSize = owner.doTransform(output, nextRec);
  1140. reader->finishedRow();
  1141. if (transformedSize)
  1142. {
  1143. processed++;
  1144. if (processed > rowLimit)
  1145. {
  1146. owner.limitExceeded();
  1147. return;
  1148. }
  1149. if (processed == stopAfter)
  1150. return;
  1151. totalSizeSent += transformedSize;
  1152. if (totalSizeSent > indexReadChunkSize && !isGrouped)
  1153. break;
  1154. }
  1155. }
  1156. else if (isGrouped)
  1157. {
  1158. if (processed>lastGroupProcessed)
  1159. break; // We return grouped data one whole group at a time
  1160. else if (!someInGroup) // eof
  1161. return;
  1162. }
  1163. else
  1164. return;
  1165. }
  1166. // Send continuation message indicating end of group or too much data
  1167. if (!aborted)
  1168. {
  1169. MemoryBuffer si;
  1170. unsigned short siLen = 0;
  1171. si.append(siLen);
  1172. si.append(processed);
  1173. si.append(reader->isKeyed());
  1174. reader->serializeCursorPos(si);
  1175. siLen = si.length() - sizeof(siLen);
  1176. si.writeDirect(0, sizeof(siLen), &siLen);
  1177. output->sendMetaInfo(si.toByteArray(), si.length());
  1178. }
  1179. }
  1180. protected:
  1181. CRoxieDiskReadActivity &owner;
  1182. IHThorDiskReadArg *helper;
  1183. bool isGrouped;
  1184. };
  1185. IInMemoryFileProcessor *createReadRecordProcessor(CRoxieDiskReadActivity &owner, bool isGrouped, IDirectReader *_reader)
  1186. {
  1187. return new ReadRecordProcessor(owner, isGrouped, _reader);
  1188. }
  1189. //================================================================================================
  1190. // RecordProcessor used by CSV read activity. We don't try to index these ...
  1191. class CsvRecordProcessor : public RecordProcessor
  1192. {
  1193. protected:
  1194. CRoxieCsvReadActivity &owner;
  1195. IHThorCsvReadArg *helper;
  1196. bool skipHeader;
  1197. const IResolvedFile *datafile;
  1198. size32_t maxRowSize;
  1199. public:
  1200. CsvRecordProcessor(CRoxieCsvReadActivity &_owner, IDirectReader *_reader, bool _skipHeader, const IResolvedFile *_datafile, size32_t _maxRowSize)
  1201. : RecordProcessor(_reader), owner(_owner), datafile(_datafile), maxRowSize(_maxRowSize)
  1202. {
  1203. // NOTE - postfilter not used at present - but may be in future
  1204. helper = _owner.helper;
  1205. skipHeader = _skipHeader;
  1206. helper->setCallback(reader);
  1207. }
  1208. virtual void doQuery(IMessagePacker *output, unsigned processed, unsigned __int64 rowLimit, unsigned __int64 stopAfter)
  1209. {
  1210. unsigned totalSizeSent = 0;
  1211. ICsvParameters * csvInfo = helper->queryCsvParameters();
  1212. unsigned headerLines = skipHeader ? csvInfo->queryHeaderLen() : 0;
  1213. const char *quotes = NULL;
  1214. const char *separators = NULL;
  1215. const char *terminators = NULL;
  1216. const char *escapes = NULL;
  1217. CSVSplitter csvSplitter;
  1218. if (datafile)
  1219. {
  1220. const IPropertyTree *options = datafile->queryProperties();
  1221. if (options)
  1222. {
  1223. quotes = options->queryProp("@csvQuote");
  1224. separators = options->queryProp("@csvSeparate");
  1225. terminators = options->queryProp("@csvTerminate");
  1226. escapes = options->queryProp("@csvEscape");
  1227. }
  1228. }
  1229. csvSplitter.init(helper->getMaxColumns(), csvInfo, quotes, separators, terminators, escapes);
  1230. ISerialStream *stream = reader->queryDirectStreamReader();
  1231. while (!aborted)
  1232. {
  1233. size32_t thisLineLength = csvSplitter.splitLine(stream, maxRowSize);
  1234. if (0 == thisLineLength)
  1235. break;
  1236. if (headerLines)
  1237. {
  1238. headerLines--;
  1239. stream->skip(thisLineLength);
  1240. }
  1241. else
  1242. {
  1243. unsigned transformedSize = owner.doTransform(output, csvSplitter.queryLengths(), (const char * *)csvSplitter.queryData());
  1244. stream->skip(thisLineLength);
  1245. if (transformedSize)
  1246. {
  1247. processed++;
  1248. if (processed > rowLimit)
  1249. {
  1250. owner.limitExceeded();
  1251. return;
  1252. }
  1253. if (processed == stopAfter)
  1254. return;
  1255. totalSizeSent += transformedSize;
  1256. if (totalSizeSent > indexReadChunkSize)
  1257. {
  1258. MemoryBuffer si;
  1259. unsigned short siLen = 0;
  1260. si.append(siLen);
  1261. si.append(processed);
  1262. si.append(reader->isKeyed());
  1263. reader->serializeCursorPos(si);
  1264. siLen = si.length() - sizeof(siLen);
  1265. si.writeDirect(0, sizeof(siLen), &siLen);
  1266. output->sendMetaInfo(si.toByteArray(), si.length());
  1267. return;
  1268. }
  1269. }
  1270. }
  1271. }
  1272. }
  1273. };
  1274. //================================================================================================
  1275. // RecordProcessor used by XML read activity. We don't try to index these...
  1276. class XmlRecordProcessor : public RecordProcessor, implements IXMLSelect, implements IThorDiskCallback
  1277. {
  1278. public:
  1279. IMPLEMENT_IINTERFACE;
  1280. XmlRecordProcessor(CRoxieXmlReadActivity &_owner, IDirectReader *_reader)
  1281. : RecordProcessor(_reader), owner(_owner), streamReader(_reader->queryDirectStreamReader()), fileposition(0)
  1282. {
  1283. helper = _owner.helper;
  1284. helper->setCallback(this);
  1285. }
  1286. //interface IThorDiskCallback
  1287. virtual unsigned __int64 getFilePosition(const void * row)
  1288. {
  1289. return fileposition;
  1290. }
  1291. virtual unsigned __int64 getLocalFilePosition(const void * row)
  1292. {
  1293. return streamReader->makeFilePositionLocal(fileposition);
  1294. }
  1295. virtual const char * queryLogicalFilename(const void * row)
  1296. {
  1297. return reader->queryLogicalFilename(row);
  1298. }
  1299. virtual void match(IColumnProvider &entry, offset_t startOffset, offset_t endOffset)
  1300. {
  1301. fileposition = startOffset;
  1302. lastMatch.set(&entry);
  1303. }
  1304. virtual void doQuery(IMessagePacker *output, unsigned processed, unsigned __int64 rowLimit, unsigned __int64 stopAfter)
  1305. {
  1306. // Note: xml read does not support continuation record stuff as too hard to serialize state of xml parser
  1307. Linked<IXmlToRowTransformer> rowTransformer = helper->queryTransformer();
  1308. OwnedRoxieString xmlIterator(helper->getXmlIteratorPath());
  1309. Owned<IXMLParse> xmlParser;
  1310. if (owner.basefactory->getKind() == TAKjsonread)
  1311. xmlParser.setown(createJSONParse(*streamReader, xmlIterator, *this, (0 != (TDRxmlnoroot & helper->getFlags()))?ptr_noRoot:ptr_none, (helper->getFlags() & TDRusexmlcontents) != 0));
  1312. else
  1313. xmlParser.setown(createXMLParse(*streamReader, xmlIterator, *this, (0 != (TDRxmlnoroot & helper->getFlags()))?ptr_noRoot:ptr_none, (helper->getFlags() & TDRusexmlcontents) != 0));
  1314. while (!aborted)
  1315. {
  1316. //call to next() will callback on the IXmlSelect interface
  1317. bool gotNext = false;
  1318. gotNext = xmlParser->next();
  1319. if(!gotNext)
  1320. break;
  1321. else if (lastMatch)
  1322. {
  1323. unsigned transformedSize = owner.doTransform(output, rowTransformer, lastMatch, this);
  1324. lastMatch.clear();
  1325. if (transformedSize)
  1326. {
  1327. processed++;
  1328. if (processed > rowLimit)
  1329. {
  1330. owner.limitExceeded();
  1331. return;
  1332. }
  1333. if (processed == stopAfter)
  1334. return;
  1335. }
  1336. }
  1337. }
  1338. }
  1339. protected:
  1340. CRoxieXmlReadActivity &owner;
  1341. IHThorXmlReadArg *helper;
  1342. Owned<IColumnProvider> lastMatch;
  1343. IDirectStreamReader *streamReader;
  1344. unsigned __int64 fileposition;
  1345. };
  1346. IInMemoryFileProcessor *createCsvRecordProcessor(CRoxieCsvReadActivity &owner, IDirectReader *_reader, bool _skipHeader, const IResolvedFile *datafile, size32_t maxRowSize)
  1347. {
  1348. return new CsvRecordProcessor(owner, _reader, _skipHeader, datafile, maxRowSize);
  1349. }
  1350. IInMemoryFileProcessor *createXmlRecordProcessor(CRoxieXmlReadActivity &owner, IDirectReader *_reader)
  1351. {
  1352. return new XmlRecordProcessor(owner, _reader);
  1353. }
  1354. ISlaveActivityFactory *createRoxieCsvReadActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  1355. {
  1356. return new CRoxieCsvReadActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory);
  1357. }
  1358. ISlaveActivityFactory *createRoxieXmlReadActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  1359. {
  1360. return new CRoxieXmlReadActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory);
  1361. }
  1362. ISlaveActivityFactory *createRoxieDiskReadActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  1363. {
  1364. return new CRoxieDiskReadActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory);
  1365. }
  1366. //================================================================================================
  1367. class CRoxieDiskNormalizeActivity;
  1368. IInMemoryFileProcessor *createNormalizeRecordProcessor(CRoxieDiskNormalizeActivity &owner, IDirectReader *_reader);
  1369. class CRoxieDiskNormalizeActivity : public CRoxieDiskReadBaseActivity
  1370. {
  1371. friend class NormalizeRecordProcessor;
  1372. protected:
  1373. IHThorDiskNormalizeArg *helper;
  1374. public:
  1375. CRoxieDiskNormalizeActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CSlaveActivityFactory *_aFactory,
  1376. IInMemoryIndexManager *_manager, ITranslatorSet *_translators)
  1377. : CRoxieDiskReadBaseActivity(_logctx, _packet, _hFactory, _aFactory, _manager, _translators, 0, 1, false)
  1378. {
  1379. onCreate();
  1380. helper = (IHThorDiskNormalizeArg *) basehelper;
  1381. }
  1382. virtual StringBuffer &toString(StringBuffer &ret) const
  1383. {
  1384. return ret.appendf("DiskNormalize %u", packet->queryHeader().activityId);
  1385. }
  1386. virtual void doProcess(IMessagePacker * output)
  1387. {
  1388. {
  1389. CriticalBlock p(pcrit);
  1390. processor.setown(createNormalizeRecordProcessor(*this, reader));
  1391. }
  1392. unsigned __int64 rowLimit = helper->getRowLimit();
  1393. unsigned __int64 stopAfter = helper->getChooseNLimit();
  1394. processor->doQuery(output, processed, rowLimit, stopAfter);
  1395. }
  1396. size32_t doNormalizeTransform(IMessagePacker * output) const
  1397. {
  1398. OptimizedRowBuilder rowBuilder(rowAllocator, meta, output, serializer);
  1399. unsigned transformedSize = helper->transform(rowBuilder);
  1400. return rowBuilder.writeToOutput(transformedSize, false);
  1401. }
  1402. };
  1403. class CRoxieDiskNormalizeActivityFactory : public CRoxieDiskBaseActivityFactory
  1404. {
  1405. public:
  1406. CRoxieDiskNormalizeActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  1407. : CRoxieDiskBaseActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory)
  1408. {
  1409. }
  1410. virtual IRoxieSlaveActivity *createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const
  1411. {
  1412. return new CRoxieDiskNormalizeActivity(logctx, packet, helperFactory, this, manager, translators);
  1413. }
  1414. virtual StringBuffer &toString(StringBuffer &s) const
  1415. {
  1416. return CSlaveActivityFactory::toString(s.append("DiskNormalize "));
  1417. }
  1418. };
  1419. ISlaveActivityFactory *createRoxieDiskNormalizeActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  1420. {
  1421. return new CRoxieDiskNormalizeActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory);
  1422. }
  1423. //================================================================================================
  1424. // RecordProcessors used by Disk Normalize activity.
  1425. class NormalizeRecordProcessor : public RecordProcessor
  1426. {
  1427. public:
  1428. NormalizeRecordProcessor(CRoxieDiskNormalizeActivity &_owner, IDirectReader *_reader)
  1429. : RecordProcessor(_reader), owner(_owner)
  1430. {
  1431. helper = _owner.helper;
  1432. }
  1433. virtual void doQuery(IMessagePacker *output, unsigned processed, unsigned __int64 rowLimit, unsigned __int64 stopAfter)
  1434. {
  1435. unsigned totalSizeSent = 0;
  1436. helper->setCallback(reader);
  1437. while (!aborted)
  1438. {
  1439. const byte *nextRec = reader->nextRow();
  1440. if (!nextRec)
  1441. break;
  1442. if (helper->first(nextRec))
  1443. {
  1444. do
  1445. {
  1446. size32_t transformedSize = owner.doNormalizeTransform(output);
  1447. if (transformedSize)
  1448. {
  1449. processed++;
  1450. if (processed > rowLimit)
  1451. {
  1452. owner.limitExceeded();
  1453. return;
  1454. }
  1455. totalSizeSent += transformedSize;
  1456. if (processed == stopAfter)
  1457. return;
  1458. }
  1459. } while (helper->next());
  1460. }
  1461. reader->finishedRow();
  1462. if (totalSizeSent > indexReadChunkSize)
  1463. {
  1464. MemoryBuffer si;
  1465. unsigned short siLen = 0;
  1466. si.append(siLen);
  1467. si.append(processed);
  1468. si.append(reader->isKeyed());
  1469. reader->serializeCursorPos(si);
  1470. siLen = si.length() - sizeof(siLen);
  1471. si.writeDirect(0, sizeof(siLen), &siLen);
  1472. output->sendMetaInfo(si.toByteArray(), si.length());
  1473. return;
  1474. }
  1475. }
  1476. }
  1477. protected:
  1478. CRoxieDiskNormalizeActivity &owner;
  1479. IHThorDiskNormalizeArg *helper;
  1480. };
  1481. IInMemoryFileProcessor *createNormalizeRecordProcessor(CRoxieDiskNormalizeActivity &owner, IDirectReader *_reader)
  1482. {
  1483. return new NormalizeRecordProcessor(owner, _reader);
  1484. }
  1485. //================================================================================================
  1486. class CRoxieDiskCountActivity;
  1487. IInMemoryFileProcessor *createCountRecordProcessor(CRoxieDiskCountActivity &owner, IDirectReader *reader);
  1488. class CRoxieDiskCountActivity : public CRoxieDiskReadBaseActivity
  1489. {
  1490. friend class CountRecordProcessor;
  1491. protected:
  1492. IHThorDiskCountArg *helper;
  1493. public:
  1494. CRoxieDiskCountActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CSlaveActivityFactory *_aFactory,
  1495. IInMemoryIndexManager *_manager, ITranslatorSet *_translators)
  1496. : CRoxieDiskReadBaseActivity(_logctx, _packet, _hFactory, _aFactory, _manager, _translators, 0, 1, false)
  1497. {
  1498. onCreate();
  1499. helper = (IHThorDiskCountArg *) basehelper;
  1500. }
  1501. virtual StringBuffer &toString(StringBuffer &ret) const
  1502. {
  1503. return ret.appendf("DiskCount %u", packet->queryHeader().activityId);
  1504. }
  1505. virtual void doProcess(IMessagePacker * output)
  1506. {
  1507. {
  1508. CriticalBlock p(pcrit);
  1509. processor.setown(createCountRecordProcessor(*this, reader));
  1510. }
  1511. unsigned __int64 stopAfter = helper->getChooseNLimit();
  1512. processor->doQuery(output, processed, (unsigned __int64) -1, stopAfter);
  1513. }
  1514. };
  1515. class CRoxieDiskCountActivityFactory : public CRoxieDiskBaseActivityFactory
  1516. {
  1517. public:
  1518. CRoxieDiskCountActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  1519. : CRoxieDiskBaseActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory)
  1520. {
  1521. }
  1522. virtual IRoxieSlaveActivity *createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const
  1523. {
  1524. return new CRoxieDiskCountActivity(logctx, packet, helperFactory, this, manager, translators);
  1525. }
  1526. virtual StringBuffer &toString(StringBuffer &s) const
  1527. {
  1528. return CSlaveActivityFactory::toString(s.append("DiskRead "));
  1529. }
  1530. };
  1531. //================================================================================================
  1532. // RecordProcessors used by Disk Count activity.
  1533. class CountRecordProcessor : public RecordProcessor
  1534. {
  1535. public:
  1536. CountRecordProcessor(CRoxieDiskCountActivity &_owner, IDirectReader *_reader)
  1537. : RecordProcessor(_reader), owner(_owner)
  1538. {
  1539. helper = _owner.helper;
  1540. }
  1541. virtual void doQuery(IMessagePacker *output, unsigned processed, unsigned __int64 rowLimit, unsigned __int64 stopAfter)
  1542. {
  1543. unsigned outputRecordSize = owner.meta.getFixedSize();
  1544. void *recBuffer = output->getBuffer(outputRecordSize, false);
  1545. helper->setCallback(reader);
  1546. unsigned __int64 totalCount = 0;
  1547. while (!aborted)
  1548. {
  1549. const byte *nextRec = reader->nextRow();
  1550. if (!nextRec)
  1551. break;
  1552. totalCount += helper->numValid(nextRec);
  1553. if (totalCount >= stopAfter)
  1554. {
  1555. totalCount = stopAfter;
  1556. break;
  1557. }
  1558. reader->finishedRow();
  1559. }
  1560. if (!aborted)
  1561. {
  1562. assert(!owner.serializer); // A count can never need serializing, surely!
  1563. if (outputRecordSize == 1)
  1564. *(byte *)recBuffer = (byte)totalCount;
  1565. else
  1566. {
  1567. assertex(outputRecordSize == sizeof(unsigned __int64));
  1568. *(unsigned __int64 *)recBuffer = totalCount;
  1569. }
  1570. output->putBuffer(recBuffer, outputRecordSize, false);
  1571. }
  1572. }
  1573. protected:
  1574. CRoxieDiskCountActivity &owner;
  1575. IHThorDiskCountArg *helper;
  1576. };
  1577. IInMemoryFileProcessor *createCountRecordProcessor(CRoxieDiskCountActivity &owner, IDirectReader *reader)
  1578. {
  1579. return new CountRecordProcessor(owner, reader);
  1580. }
  1581. ISlaveActivityFactory *createRoxieDiskCountActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  1582. {
  1583. return new CRoxieDiskCountActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory);
  1584. }
  1585. //================================================================================================
  1586. class CRoxieDiskAggregateActivity;
  1587. IInMemoryFileProcessor *createAggregateRecordProcessor(CRoxieDiskAggregateActivity &owner, IDirectReader *_reader);
  1588. class CRoxieDiskAggregateActivity : public CRoxieDiskReadBaseActivity
  1589. {
  1590. friend class AggregateRecordProcessor;
  1591. protected:
  1592. IHThorDiskAggregateArg *helper;
  1593. public:
  1594. CRoxieDiskAggregateActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CSlaveActivityFactory *_aFactory,
  1595. IInMemoryIndexManager *_manager,
  1596. ITranslatorSet *_translators,
  1597. unsigned _parallelPartNo, unsigned _numParallel, bool _forceUnkeyed)
  1598. : CRoxieDiskReadBaseActivity(_logctx, _packet, _hFactory, _aFactory, _manager, _translators, _parallelPartNo, _numParallel, _forceUnkeyed)
  1599. {
  1600. onCreate();
  1601. helper = (IHThorDiskAggregateArg *) basehelper;
  1602. }
  1603. virtual bool needsRowAllocator()
  1604. {
  1605. return true;
  1606. }
  1607. virtual StringBuffer &toString(StringBuffer &ret) const
  1608. {
  1609. return ret.appendf("DiskAggregate %u", packet->queryHeader().activityId);
  1610. }
  1611. virtual void doProcess(IMessagePacker * output)
  1612. {
  1613. {
  1614. CriticalBlock p(pcrit);
  1615. processor.setown(createAggregateRecordProcessor(*this, reader));
  1616. }
  1617. processor->doQuery(output, 0, 0, 0);
  1618. }
  1619. };
  1620. //================================================================================================
  1621. class CParallelRoxieActivity : public CRoxieSlaveActivity
  1622. {
  1623. protected:
  1624. IBasedArrayOf<CRoxieDiskReadBaseActivity, IRoxieSlaveActivity> parts;
  1625. unsigned numParallel;
  1626. CriticalSection parCrit;
  1627. Owned<IOutputRowDeserializer> deserializer;
  1628. public:
  1629. CParallelRoxieActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CSlaveActivityFactory *_factory, unsigned _numParallel)
  1630. : CRoxieSlaveActivity(_logctx, _packet, _hFactory, _factory), numParallel(_numParallel)
  1631. {
  1632. assertex(numParallel > 1);
  1633. }
  1634. virtual void abort()
  1635. {
  1636. ForEachItemIn(idx, parts)
  1637. {
  1638. parts.item(idx).abort();
  1639. }
  1640. }
  1641. virtual void setPartNo(bool filechanged) { throwUnexpected(); }
  1642. virtual StringBuffer &toString(StringBuffer &ret) const
  1643. {
  1644. return parts.item(0).toString(ret);
  1645. }
  1646. virtual const char *queryDynamicFileName() const
  1647. {
  1648. throwUnexpected();
  1649. }
  1650. virtual void setVariableFileInfo()
  1651. {
  1652. throwUnexpected();
  1653. }
  1654. virtual void doProcess(IMessagePacker * output) = 0;
  1655. virtual void processRow(CDummyMessagePacker &output) = 0;
  1656. virtual IMessagePacker *process()
  1657. {
  1658. if (numParallel == 1)
  1659. {
  1660. return parts.item(0).process();
  1661. }
  1662. else
  1663. {
  1664. MTIME_SECTION(queryActiveTimer(), "CParallelRoxieActivity::process");
  1665. diskReadStarted++;
  1666. Owned<IMessagePacker> output = ROQ->createOutputStream(packet->queryHeader(), false, logctx);
  1667. class casyncfor: public CAsyncFor
  1668. {
  1669. IBasedArrayOf<CRoxieDiskReadBaseActivity, IRoxieSlaveActivity> &parts;
  1670. CParallelRoxieActivity &parent;
  1671. public:
  1672. casyncfor(IBasedArrayOf<CRoxieDiskReadBaseActivity, IRoxieSlaveActivity> &_parts, CParallelRoxieActivity &_parent)
  1673. : parts(_parts), parent(_parent)
  1674. {
  1675. }
  1676. void Do(unsigned i)
  1677. {
  1678. try
  1679. {
  1680. CDummyMessagePacker d;
  1681. parts.item(i).doProcess(&d);
  1682. d.flush(true);
  1683. parent.processRow(d);
  1684. }
  1685. catch (IException *)
  1686. {
  1687. // if one throws exception, may as well abort the rest
  1688. parent.abort();
  1689. throw;
  1690. }
  1691. }
  1692. } afor(parts, *this);
  1693. afor.For(numParallel, numParallel);
  1694. //for (unsigned i = 0; i < numParallel; i++) afor.Do(i); // use this instead of line above to make them serial - handy for debugging!
  1695. if (aborted)
  1696. return NULL;
  1697. else
  1698. {
  1699. doProcess(output);
  1700. diskReadCompleted++;
  1701. return output.getClear();
  1702. }
  1703. }
  1704. }
  1705. };
  1706. class CParallelRoxieDiskAggregateActivity : public CParallelRoxieActivity
  1707. {
  1708. protected:
  1709. IHThorDiskAggregateArg *helper;
  1710. OwnedConstRoxieRow finalRow;
  1711. public:
  1712. CParallelRoxieDiskAggregateActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CSlaveActivityFactory *_aFactory,
  1713. IInMemoryIndexManager *_manager, ITranslatorSet *_translators, unsigned _numParallel) :
  1714. CParallelRoxieActivity(_logctx, _packet, _hFactory, _aFactory, _numParallel)
  1715. {
  1716. helper = (IHThorDiskAggregateArg *) basehelper;
  1717. onCreate();
  1718. if (meta.needsSerializeDisk())
  1719. {
  1720. // MORE - avoiding serializing to dummy would be more efficient...
  1721. deserializer.setown(meta.createDiskDeserializer(queryContext->queryCodeContext(), basefactory->queryId()));
  1722. }
  1723. CRoxieDiskAggregateActivity *part0 = new CRoxieDiskAggregateActivity(_logctx, _packet, _hFactory, _aFactory, _manager, _translators, 0, numParallel, false);
  1724. parts.append(*part0);
  1725. if (part0->queryKeyed())
  1726. {
  1727. numParallel = 1;
  1728. part0->setParallel(0, 1);
  1729. }
  1730. else
  1731. {
  1732. for (unsigned i = 1; i < numParallel; i++)
  1733. parts.append(*new CRoxieDiskAggregateActivity(_logctx, _packet, _hFactory, _aFactory, _manager, _translators, i, numParallel, true));
  1734. }
  1735. }
  1736. ~CParallelRoxieDiskAggregateActivity()
  1737. {
  1738. finalRow.clear();
  1739. }
  1740. virtual bool needsRowAllocator()
  1741. {
  1742. return true;
  1743. }
  1744. virtual void doProcess(IMessagePacker *output)
  1745. {
  1746. if (!aborted)
  1747. {
  1748. if (!finalRow)
  1749. {
  1750. RtlDynamicRowBuilder rowBuilder(rowAllocator);
  1751. size32_t size = helper->clearAggregate(rowBuilder);
  1752. finalRow.setown(rowBuilder.finalizeRowClear(size));
  1753. }
  1754. //GH-RKC: This can probably be cleaner and more efficient using the OptimizedRowBuilder class introduced since I fixed this code
  1755. if (serializer)
  1756. serializeRow(output, finalRow);
  1757. else
  1758. {
  1759. size32_t size = meta.getRecordSize(finalRow);
  1760. appendBuffer(output, size, finalRow, meta.isVariableSize());
  1761. }
  1762. }
  1763. finalRow.clear();
  1764. helper->setCallback(NULL);
  1765. }
  1766. virtual void processRow(CDummyMessagePacker &d)
  1767. {
  1768. CriticalBlock c(parCrit);
  1769. MemoryBuffer &m = d.data;
  1770. RtlDynamicRowBuilder finalBuilder(rowAllocator, false);
  1771. if (deserializer)
  1772. {
  1773. Owned<ISerialStream> stream = createMemoryBufferSerialStream(m);
  1774. CThorStreamDeserializerSource rowSource(stream);
  1775. while (m.remaining())
  1776. {
  1777. RecordLengthType *rowLen = (RecordLengthType *) m.readDirect(sizeof(RecordLengthType));
  1778. if (!*rowLen)
  1779. break;
  1780. RecordLengthType len = *rowLen;
  1781. RtlDynamicRowBuilder rowBuilder(rowAllocator);
  1782. size_t outsize = deserializer->deserialize(rowBuilder, rowSource);
  1783. if (!finalBuilder.exists())
  1784. finalBuilder.swapWith(rowBuilder);
  1785. else
  1786. {
  1787. const void * deserialized = rowBuilder.finalizeRowClear(outsize);
  1788. helper->mergeAggregate(finalBuilder, deserialized);
  1789. ReleaseRoxieRow(deserialized);
  1790. }
  1791. }
  1792. }
  1793. else
  1794. {
  1795. RecordLengthType len = meta.getFixedSize();
  1796. while (m.remaining())
  1797. {
  1798. const void *row;
  1799. if (!meta.isFixedSize())
  1800. {
  1801. RecordLengthType *rowLen = (RecordLengthType *) m.readDirect(sizeof(RecordLengthType));
  1802. if (!*rowLen)
  1803. break;
  1804. len = *rowLen;
  1805. }
  1806. row = m.readDirect(len);
  1807. if (!finalBuilder.exists())
  1808. cloneRow(finalBuilder, row, meta);
  1809. else
  1810. helper->mergeAggregate(finalBuilder, row);
  1811. }
  1812. }
  1813. if (finalBuilder.exists())
  1814. {
  1815. size32_t finalSize = meta.getRecordSize(finalBuilder.getSelf()); // MORE - can probably track it above...
  1816. finalRow.setown(finalBuilder.finalizeRowClear(finalSize));
  1817. }
  1818. }
  1819. };
  1820. class CRoxieDiskAggregateActivityFactory : public CRoxieDiskBaseActivityFactory
  1821. {
  1822. public:
  1823. CRoxieDiskAggregateActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  1824. : CRoxieDiskBaseActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory)
  1825. {
  1826. }
  1827. virtual IRoxieSlaveActivity *createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const
  1828. {
  1829. if (parallelAggregate > 1)
  1830. return new CParallelRoxieDiskAggregateActivity(logctx, packet, helperFactory, this, manager, translators, parallelAggregate);
  1831. else
  1832. return new CRoxieDiskAggregateActivity(logctx, packet, helperFactory, this, manager, translators, 0, 1, false);
  1833. }
  1834. virtual StringBuffer &toString(StringBuffer &s) const
  1835. {
  1836. return CSlaveActivityFactory::toString(s.append("DiskRead "));
  1837. }
  1838. };
  1839. //================================================================================================
  1840. // RecordProcessor used by Disk Aggregate activity.
  1841. class AggregateRecordProcessor : public RecordProcessor
  1842. {
  1843. public:
  1844. AggregateRecordProcessor(CRoxieDiskAggregateActivity &_owner, IDirectReader *_reader)
  1845. : RecordProcessor(_reader), owner(_owner)
  1846. {
  1847. helper = _owner.helper;
  1848. helper->setCallback(reader);
  1849. }
  1850. virtual void doQuery(IMessagePacker *output, unsigned processed, unsigned __int64 rowLimit, unsigned __int64 stopAfter)
  1851. {
  1852. OptimizedRowBuilder rowBuilder(owner.rowAllocator, owner.meta, output, owner.serializer);
  1853. helper->clearAggregate(rowBuilder);
  1854. while (!aborted)
  1855. {
  1856. const byte *nextRec = reader->nextRow();
  1857. if (!nextRec)
  1858. break;
  1859. helper->processRow(rowBuilder, nextRec);
  1860. reader->finishedRow();
  1861. }
  1862. if (!aborted)
  1863. {
  1864. if (helper->processedAnyRows())
  1865. {
  1866. size32_t finalSize = owner.meta.getRecordSize(rowBuilder.getSelf());
  1867. rowBuilder.writeToOutput(finalSize, true);
  1868. }
  1869. }
  1870. }
  1871. protected:
  1872. CRoxieDiskAggregateActivity &owner;
  1873. IHThorDiskAggregateArg *helper;
  1874. };
  1875. IInMemoryFileProcessor *createAggregateRecordProcessor(CRoxieDiskAggregateActivity &_owner, IDirectReader *_reader)
  1876. {
  1877. return new AggregateRecordProcessor(_owner, _reader);
  1878. }
  1879. ISlaveActivityFactory *createRoxieDiskAggregateActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  1880. {
  1881. return new CRoxieDiskAggregateActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory);
  1882. }
  1883. //================================================================================================
  1884. class CRoxieDiskGroupAggregateActivity : public CRoxieDiskReadBaseActivity
  1885. {
  1886. protected:
  1887. IHThorDiskGroupAggregateArg *helper;
  1888. RowAggregator results;
  1889. void outputResults(IMessagePacker *output)
  1890. {
  1891. if (!aborted)
  1892. {
  1893. for (;;)
  1894. {
  1895. Owned<AggregateRowBuilder> next = results.nextResult();
  1896. if (!next)
  1897. break;
  1898. unsigned rowSize = next->querySize();
  1899. OwnedConstRoxieRow row(next->finalizeRowClear());
  1900. if (serializer)
  1901. {
  1902. serializeRow(output, row);
  1903. }
  1904. else
  1905. {
  1906. void *recBuffer = output->getBuffer(rowSize, meta.isVariableSize());
  1907. memcpy(recBuffer, row, rowSize);
  1908. output->putBuffer(recBuffer, rowSize, meta.isVariableSize());
  1909. }
  1910. }
  1911. }
  1912. }
  1913. public:
  1914. CRoxieDiskGroupAggregateActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CSlaveActivityFactory *_aFactory,
  1915. IInMemoryIndexManager *_manager,
  1916. ITranslatorSet *_translators,
  1917. unsigned partNo, unsigned numParts, bool _forceUnkeyed)
  1918. : CRoxieDiskReadBaseActivity(_logctx, _packet, _hFactory, _aFactory, _manager, _translators, partNo, numParts, _forceUnkeyed),
  1919. helper((IHThorDiskGroupAggregateArg *) basehelper),
  1920. results(*helper, *helper)
  1921. {
  1922. onCreate();
  1923. results.start(rowAllocator, queryContext->queryCodeContext(), basefactory->queryId());
  1924. }
  1925. virtual bool needsRowAllocator()
  1926. {
  1927. return true;
  1928. }
  1929. virtual StringBuffer &toString(StringBuffer &ret) const
  1930. {
  1931. return ret.appendf("DiskGroupAggregate %u", packet->queryHeader().activityId);
  1932. }
  1933. virtual void doProcess(IMessagePacker * output)
  1934. {
  1935. {
  1936. CriticalBlock p(pcrit);
  1937. processor.setown(createGroupAggregateRecordProcessor(results, *helper, reader));
  1938. }
  1939. processor->doQuery(output, 0, 0, 0);
  1940. if (!aborted)
  1941. outputResults(output);
  1942. results.reset();
  1943. }
  1944. };
  1945. class CParallelRoxieDiskGroupAggregateActivity : public CParallelRoxieActivity
  1946. {
  1947. protected:
  1948. IHThorDiskGroupAggregateArg *helper;
  1949. RowAggregator resultAggregator;
  1950. Owned<IRowManager> rowManager;
  1951. public:
  1952. CParallelRoxieDiskGroupAggregateActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CSlaveActivityFactory *_aFactory,
  1953. IInMemoryIndexManager *_manager, ITranslatorSet *_translators, unsigned _numParallel) :
  1954. CParallelRoxieActivity(_logctx, _packet, _hFactory, _aFactory, _numParallel),
  1955. helper((IHThorDiskGroupAggregateArg *) basehelper),
  1956. resultAggregator(*helper, *helper)
  1957. {
  1958. onCreate();
  1959. resultAggregator.start(rowAllocator, queryContext->queryCodeContext(), basefactory->queryId());
  1960. if (meta.needsSerializeDisk())
  1961. {
  1962. // MORE - avoiding serializing to dummy would be more efficient...
  1963. deserializer.setown(meta.createDiskDeserializer(queryContext->queryCodeContext(), basefactory->queryId()));
  1964. }
  1965. CRoxieDiskGroupAggregateActivity *part0 = new CRoxieDiskGroupAggregateActivity(_logctx, _packet, _hFactory, _aFactory, _manager, _translators, 0, numParallel, false);
  1966. parts.append(*part0);
  1967. if (part0->queryKeyed())
  1968. {
  1969. numParallel = 1;
  1970. part0->setParallel(0, 1);
  1971. }
  1972. else
  1973. {
  1974. for (unsigned i = 1; i < numParallel; i++)
  1975. parts.append(*new CRoxieDiskGroupAggregateActivity(_logctx, _packet, _hFactory, _aFactory, _manager, _translators, i, numParallel, true));
  1976. }
  1977. }
  1978. virtual bool needsRowAllocator()
  1979. {
  1980. return true;
  1981. }
  1982. virtual void doProcess(IMessagePacker *output)
  1983. {
  1984. if (!aborted)
  1985. {
  1986. for (;;)
  1987. {
  1988. Owned<AggregateRowBuilder> next = resultAggregator.nextResult();
  1989. if (!next)
  1990. break;
  1991. unsigned rowSize = next->querySize();
  1992. OwnedConstRoxieRow row(next->finalizeRowClear());
  1993. if (serializer)
  1994. {
  1995. serializeRow(output, row);
  1996. }
  1997. else
  1998. {
  1999. void *recBuffer = output->getBuffer(rowSize, meta.isVariableSize());
  2000. memcpy(recBuffer, row, rowSize);
  2001. output->putBuffer(recBuffer, rowSize, meta.isVariableSize());
  2002. }
  2003. }
  2004. }
  2005. resultAggregator.reset();
  2006. helper->setCallback(NULL);
  2007. }
  2008. void processRow(CDummyMessagePacker &d)
  2009. {
  2010. CriticalBlock b(parCrit); // MORE - use a spinlock
  2011. MemoryBuffer &m = d.data;
  2012. Owned<ISerialStream> stream = createMemoryBufferSerialStream(m);
  2013. CThorStreamDeserializerSource rowSource(stream);
  2014. while (m.remaining())
  2015. {
  2016. const void *row;
  2017. if (meta.isFixedSize() && !deserializer)
  2018. {
  2019. row = m.readDirect(meta.getFixedSize());
  2020. resultAggregator.mergeElement(row);
  2021. }
  2022. else
  2023. {
  2024. RecordLengthType *rowLen = (RecordLengthType *) m.readDirect(sizeof(RecordLengthType));
  2025. if (!*rowLen)
  2026. break;
  2027. RecordLengthType len = *rowLen;
  2028. if (deserializer)
  2029. {
  2030. RtlDynamicRowBuilder rowBuilder(rowAllocator);
  2031. size_t outsize = deserializer->deserialize(rowBuilder, rowSource);
  2032. OwnedConstRoxieRow deserialized = rowBuilder.finalizeRowClear(outsize);
  2033. resultAggregator.mergeElement(deserialized);
  2034. }
  2035. else
  2036. {
  2037. row = m.readDirect(len);
  2038. resultAggregator.mergeElement(row);
  2039. }
  2040. }
  2041. }
  2042. }
  2043. };
  2044. class CRoxieDiskGroupAggregateActivityFactory : public CRoxieDiskBaseActivityFactory
  2045. {
  2046. public:
  2047. CRoxieDiskGroupAggregateActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  2048. : CRoxieDiskBaseActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory)
  2049. {
  2050. }
  2051. virtual IRoxieSlaveActivity *createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const
  2052. {
  2053. if (parallelAggregate > 1)
  2054. return new CParallelRoxieDiskGroupAggregateActivity(logctx, packet, helperFactory, this, manager, translators, parallelAggregate);
  2055. else
  2056. return new CRoxieDiskGroupAggregateActivity(logctx, packet, helperFactory, this, manager, translators, 0, 1, false);
  2057. }
  2058. virtual StringBuffer &toString(StringBuffer &s) const
  2059. {
  2060. return CSlaveActivityFactory::toString(s.append("DiskRead "));
  2061. }
  2062. };
  2063. //================================================================================================
  2064. // RecordProcessors used by Disk Group Aggregate activity.
  2065. class GroupAggregateRecordProcessor : public RecordProcessor, implements IHThorGroupAggregateCallback
  2066. {
  2067. public:
  2068. IMPLEMENT_IINTERFACE;
  2069. GroupAggregateRecordProcessor(RowAggregator &_results, IHThorDiskGroupAggregateArg &_helper, IDirectReader *_reader)
  2070. : RecordProcessor(_reader), results(_results), helper(_helper)
  2071. {
  2072. }
  2073. virtual void processRow(const void * next)
  2074. {
  2075. results.addRow(next);
  2076. }
  2077. virtual void doQuery(IMessagePacker *output, unsigned processed, unsigned __int64 rowLimit, unsigned __int64 stopAfter)
  2078. {
  2079. helper.setCallback(reader);
  2080. while (!aborted)
  2081. {
  2082. const byte *nextRec = reader->nextRow();
  2083. if (!nextRec)
  2084. break;
  2085. helper.processRow(nextRec, this);
  2086. reader->finishedRow();
  2087. }
  2088. }
  2089. protected:
  2090. RowAggregator &results;
  2091. IHThorDiskGroupAggregateArg &helper;
  2092. };
  2093. IInMemoryFileProcessor *createGroupAggregateRecordProcessor(RowAggregator &results, IHThorDiskGroupAggregateArg &helper, IDirectReader *reader)
  2094. {
  2095. return new GroupAggregateRecordProcessor(results, helper, reader);
  2096. }
  2097. ISlaveActivityFactory *createRoxieDiskGroupAggregateActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  2098. {
  2099. return new CRoxieDiskGroupAggregateActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory);
  2100. }
  2101. //================================================================================================
  2102. class CRoxieKeyedActivityFactory : public CSlaveActivityFactory
  2103. {
  2104. friend class CRoxieKeyedActivity;
  2105. protected:
  2106. Owned<IKeyArray> keyArray;
  2107. Owned<ITranslatorSet> translators;
  2108. unsigned formatCrc = 0;
  2109. IOutputMetaData *projectedMeta = nullptr;
  2110. IOutputMetaData *expectedMeta = nullptr;
  2111. CRoxieKeyedActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  2112. : CSlaveActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory)
  2113. {
  2114. }
  2115. };
  2116. class CRoxieIndexActivityFactory : public CRoxieKeyedActivityFactory
  2117. {
  2118. public:
  2119. CRoxieIndexActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  2120. : CRoxieKeyedActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory)
  2121. {
  2122. }
  2123. void init(IHThorIndexReadBaseArg * helper, IPropertyTree &graphNode)
  2124. {
  2125. formatCrc = getFormatCrc(helper->getFormatCrc());
  2126. projectedMeta = helper->queryProjectedDiskRecordSize();
  2127. expectedMeta = helper->queryDiskRecordSize();
  2128. bool variableFileName = allFilesDynamic || queryFactory.isDynamic() || ((helper->getFlags() & (TIRvarfilename|TIRdynamicfilename)) != 0);
  2129. if (!variableFileName)
  2130. {
  2131. bool isOpt = (helper->getFlags() & TIRoptional) != 0;
  2132. OwnedRoxieString indexName(helper->getFileName());
  2133. datafile.setown(queryFactory.queryPackage().lookupFileName(indexName, isOpt, true, true, queryFactory.queryWorkUnit(), true));
  2134. if (datafile)
  2135. {
  2136. translators.setown(datafile->getTranslators(formatCrc, projectedMeta, expectedMeta, queryFactory.queryOptions().enableFieldTranslation, true));
  2137. keyArray.setown(datafile->getKeyArray(isOpt, queryFactory.queryChannel()));
  2138. }
  2139. }
  2140. }
  2141. };
  2142. class CRoxieKeyedActivity : public CRoxieSlaveActivity
  2143. {
  2144. // Common base class for all activities that deal with keys - keyed join or indexread and its allies
  2145. protected:
  2146. Owned<IKeyManager> tlk;
  2147. Linked<ITranslatorSet> translators;
  2148. Linked<IKeyArray> keyArray;
  2149. const RtlRecord *keyRecInfo = nullptr;
  2150. bool createSegmentMonitorsPending;
  2151. virtual void createSegmentMonitors() = 0;
  2152. virtual void setPartNo(bool filechanged)
  2153. {
  2154. if (!lastPartNo.partNo) // Check for ,LOCAL indexes
  2155. {
  2156. assertex(filechanged);
  2157. Owned<IKeyIndexSet> allKeys = createKeyIndexSet();
  2158. for (unsigned subpart = 0; subpart < keyArray->length(); subpart++)
  2159. {
  2160. IKeyIndexBase *kib = keyArray->queryKeyPart(subpart);
  2161. if (kib)
  2162. {
  2163. IKeyIndex *k = kib->queryPart(lastPartNo.fileNo);
  2164. if (k)
  2165. {
  2166. assertex(!k->isTopLevelKey());
  2167. allKeys->addIndex(LINK(k));
  2168. }
  2169. }
  2170. }
  2171. if (allKeys->numParts())
  2172. {
  2173. tlk.setown(createKeyMerger(*keyRecInfo, allKeys, 0, &logctx));
  2174. createSegmentMonitorsPending = true;
  2175. }
  2176. else
  2177. tlk.clear();
  2178. }
  2179. else
  2180. {
  2181. IKeyIndexBase *kib = keyArray->queryKeyPart(lastPartNo.partNo);
  2182. assertex(kib != NULL);
  2183. IKeyIndex *k = kib->queryPart(lastPartNo.fileNo);
  2184. if (filechanged)
  2185. {
  2186. tlk.setown(createLocalKeyManager(*keyRecInfo, k, &logctx));
  2187. createSegmentMonitorsPending = true;
  2188. }
  2189. else
  2190. tlk->setKey(k);
  2191. }
  2192. }
  2193. virtual void setVariableFileInfo()
  2194. {
  2195. const CRoxieKeyedActivityFactory &aFactory = *static_cast<const CRoxieKeyedActivityFactory *>(basefactory);
  2196. translators.setown(varFileInfo->getTranslators(aFactory.formatCrc, aFactory.projectedMeta, aFactory.expectedMeta, allowFieldTranslation, true));
  2197. keyArray.setown(varFileInfo->getKeyArray(isOpt, packet->queryHeader().channel));
  2198. }
  2199. void noteStats(unsigned accepted, unsigned rejected)
  2200. {
  2201. // Note that the key-level statistics (seeks, scans, skips etc) are handled inside jhtree
  2202. logctx.noteStatistic(StNumIndexAccepted, accepted);
  2203. logctx.noteStatistic(StNumIndexRejected, rejected);
  2204. logctx.noteStatistic(StNumIndexRowsRead, accepted+rejected);
  2205. }
  2206. CRoxieKeyedActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CRoxieKeyedActivityFactory *_aFactory)
  2207. : CRoxieSlaveActivity(_logctx, _packet, _hFactory, _aFactory),
  2208. keyArray(_aFactory->keyArray),
  2209. translators(_aFactory->translators),
  2210. createSegmentMonitorsPending(true)
  2211. {
  2212. }
  2213. };
  2214. class CRoxieIndexActivity : public CRoxieKeyedActivity
  2215. {
  2216. // Common base class for indexread, indexcount and related activities
  2217. protected:
  2218. PartNoType *inputData; // list of channels
  2219. IHThorIndexReadBaseArg * indexHelper;
  2220. unsigned inputCount;
  2221. unsigned inputsDone;
  2222. unsigned processed;
  2223. unsigned keyprocessed;
  2224. unsigned steppingOffset;
  2225. unsigned steppingLength;
  2226. unsigned short numSkipFields;
  2227. unsigned numSeeks;
  2228. bool seeksAreEof;
  2229. bool lastRowCompleteMatch;
  2230. CIndexTransformCallback callback;
  2231. SmartStepExtra stepExtra; // just used for flags - a little unnecessary...
  2232. const byte *steppingRow;
  2233. bool checkLimit(unsigned __int64 limit)
  2234. {
  2235. assertex(!resent);
  2236. unsigned __int64 result = 0;
  2237. unsigned inputsDone = 0;
  2238. bool ret = true;
  2239. unsigned saveStepping = steppingOffset; // Avoid using a stepping keymerger for the checkCount - they don't support it (and would be inefficient)
  2240. steppingOffset = 0;
  2241. while (!aborted && inputsDone < inputCount)
  2242. {
  2243. checkPartChanged(inputData[inputsDone]);
  2244. if (tlk)
  2245. {
  2246. createSegmentMonitors();
  2247. result += tlk->checkCount(limit-result);
  2248. if (result > limit)
  2249. {
  2250. ret = false;
  2251. break;
  2252. }
  2253. }
  2254. inputsDone++;
  2255. }
  2256. if (saveStepping)
  2257. {
  2258. steppingOffset = saveStepping;
  2259. lastPartNo.partNo = 0xffff;
  2260. lastPartNo.fileNo = 0xffff;
  2261. tlk.clear();
  2262. }
  2263. return ret;
  2264. }
  2265. public:
  2266. CRoxieIndexActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CRoxieIndexActivityFactory *_aFactory, unsigned _steppingOffset)
  2267. : CRoxieKeyedActivity(_logctx, _packet, _hFactory, _aFactory),
  2268. steppingOffset(_steppingOffset),
  2269. stepExtra(SSEFreadAhead, NULL)
  2270. {
  2271. indexHelper = (IHThorIndexReadBaseArg *) basehelper;
  2272. keyRecInfo = &indexHelper->queryDiskRecordSize()->queryRecordAccessor(true);
  2273. variableFileName = allFilesDynamic || basefactory->queryQueryFactory().isDynamic() || ((indexHelper->getFlags() & (TIRvarfilename|TIRdynamicfilename)) != 0);
  2274. isOpt = (indexHelper->getFlags() & TDRoptional) != 0;
  2275. inputData = NULL;
  2276. inputCount = 0;
  2277. inputsDone = 0;
  2278. processed = 0;
  2279. keyprocessed = 0;
  2280. numSkipFields = 0;
  2281. lastRowCompleteMatch = true; // default is we only return complete matches....
  2282. seeksAreEof = false;
  2283. steppingLength = 0;
  2284. steppingRow = NULL;
  2285. numSeeks = 0;
  2286. if (packet->getSmartStepInfoLength())
  2287. {
  2288. const byte *smartStepInfoValue = packet->querySmartStepInfoData();
  2289. numSkipFields = * (unsigned short *) smartStepInfoValue;
  2290. smartStepInfoValue += sizeof(unsigned short);
  2291. steppingLength = * (unsigned short *) smartStepInfoValue;
  2292. smartStepInfoValue += sizeof(unsigned short);
  2293. unsigned flags = * (unsigned short *) smartStepInfoValue;
  2294. smartStepInfoValue += sizeof(unsigned short);
  2295. seeksAreEof = * (bool *) smartStepInfoValue;
  2296. smartStepInfoValue += sizeof(bool);
  2297. numSeeks = * (unsigned *) smartStepInfoValue;
  2298. smartStepInfoValue += sizeof(unsigned);
  2299. assertex(numSeeks); // Given that we put the first seek in here to there should always be at least one!
  2300. steppingRow = smartStepInfoValue; // the first of them...
  2301. stepExtra.set(flags, NULL);
  2302. if (logctx.queryTraceLevel() > 10)
  2303. {
  2304. logctx.CTXLOG("%d seek rows provided. mismatch(%d) readahead(%d) onlyfirst(%d)", numSeeks,
  2305. (int)stepExtra.returnMismatches(), (int)stepExtra.readAheadManyResults(), (int)stepExtra.onlyReturnFirstSeekMatch());
  2306. if (logctx.queryTraceLevel() > 15)
  2307. {
  2308. for (unsigned i = 0; i < numSeeks; i++)
  2309. {
  2310. StringBuffer b;
  2311. for (unsigned j = 0; j < steppingLength; j++)
  2312. b.appendf("%02x ", steppingRow[i*steppingLength + j]);
  2313. logctx.CTXLOG("Seek row %d: %s", i+1, b.str());
  2314. }
  2315. }
  2316. }
  2317. }
  2318. else
  2319. {
  2320. if (logctx.queryTraceLevel() > 10)
  2321. logctx.CTXLOG("0 seek rows provided.");
  2322. }
  2323. }
  2324. virtual void onCreate()
  2325. {
  2326. CRoxieKeyedActivity::onCreate();
  2327. inputData = (PartNoType *) serializedCreate.readDirect(0);
  2328. inputCount = (serializedCreate.length() - serializedCreate.getPos()) / sizeof(*inputData);
  2329. indexHelper->setCallback(&callback);
  2330. }
  2331. virtual const char *queryDynamicFileName() const
  2332. {
  2333. return indexHelper->getFileName();
  2334. }
  2335. virtual void createSegmentMonitors()
  2336. {
  2337. if (createSegmentMonitorsPending)
  2338. {
  2339. createSegmentMonitorsPending = false;
  2340. tlk->setLayoutTranslator(translators->queryTranslator(lastPartNo.fileNo));
  2341. indexHelper->createSegmentMonitors(tlk);
  2342. tlk->finishSegmentMonitors();
  2343. }
  2344. }
  2345. bool sendContinuation(IMessagePacker * output)
  2346. {
  2347. MemoryBuffer si;
  2348. unsigned short siLen = 0;
  2349. si.append(siLen);
  2350. si.append(lastRowCompleteMatch);
  2351. si.append(inputsDone);
  2352. si.append(processed);
  2353. si.append(keyprocessed);
  2354. si.append(lastPartNo.partNo);
  2355. si.append(lastPartNo.fileNo);
  2356. tlk->serializeCursorPos(si);
  2357. if (si.length() <= maxContinuationSize)
  2358. {
  2359. siLen = si.length() - sizeof(siLen);
  2360. si.writeDirect(0, sizeof(siLen), &siLen);
  2361. output->sendMetaInfo(si.toByteArray(), si.length());
  2362. return true;
  2363. }
  2364. else
  2365. return false;
  2366. }
  2367. void readContinuationInfo()
  2368. {
  2369. resentInfo.read(lastRowCompleteMatch);
  2370. resentInfo.read(inputsDone);
  2371. resentInfo.read(processed);
  2372. resentInfo.read(keyprocessed);
  2373. resentInfo.read(lastPartNo.partNo);
  2374. resentInfo.read(lastPartNo.fileNo);
  2375. setPartNo(true);
  2376. tlk->deserializeCursorPos(resentInfo);
  2377. assertex(resentInfo.remaining() == 0);
  2378. }
  2379. virtual void setPartNo(bool fileChanged)
  2380. {
  2381. // NOTE - may be used by both indexread and normalize...
  2382. if (steppingOffset) // MORE - may be other cases too - eg want output sorted and there are multiple subfiles...
  2383. {
  2384. unsigned i = 0;
  2385. Owned<IKeyIndexSet> allKeys = createKeyIndexSet();
  2386. while (!aborted && i < inputCount)
  2387. {
  2388. PartNoType &part = inputData[i];
  2389. lastPartNo.partNo = part.partNo;
  2390. lastPartNo.fileNo = part.fileNo; // This is a hack so that the translator can be retrieved. We don't support record translation properly when doing keymerging...)
  2391. // MORE - this code looks like it could be commoned up with code in CRoxieKeyedActivity::setPartNo
  2392. if (!lastPartNo.partNo)
  2393. {
  2394. for (unsigned subpart = 0; subpart < keyArray->length(); subpart++)
  2395. {
  2396. IKeyIndexBase *kib = keyArray->queryKeyPart(subpart);
  2397. if (kib)
  2398. {
  2399. IKeyIndex *k = kib->queryPart(lastPartNo.fileNo);
  2400. if (k)
  2401. {
  2402. assertex(!k->isTopLevelKey());
  2403. allKeys->addIndex(LINK(k));
  2404. }
  2405. }
  2406. }
  2407. }
  2408. else
  2409. {
  2410. IKeyIndexBase *kib = keyArray->queryKeyPart(part.partNo);
  2411. assertex(kib != NULL);
  2412. IKeyIndex *k = kib->queryPart(part.fileNo);
  2413. allKeys->addIndex(LINK(k));
  2414. }
  2415. i++;
  2416. }
  2417. if (allKeys->numParts())
  2418. tlk.setown(::createKeyMerger(*keyRecInfo, allKeys, steppingOffset, &logctx));
  2419. else
  2420. tlk.clear();
  2421. createSegmentMonitorsPending = true;
  2422. }
  2423. else
  2424. CRoxieKeyedActivity::setPartNo(fileChanged);
  2425. }
  2426. };
  2427. //================================================================================================
  2428. class CRoxieIndexReadActivity : public CRoxieIndexActivity, implements IIndexReadActivityInfo
  2429. {
  2430. protected:
  2431. IHThorCompoundReadExtra * readHelper;
  2432. public:
  2433. CRoxieIndexReadActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CRoxieIndexActivityFactory *_aFactory, unsigned _steppingOffset)
  2434. : CRoxieIndexActivity(_logctx, _packet, _hFactory, _aFactory, _steppingOffset)
  2435. {
  2436. onCreate();
  2437. readHelper = (IHThorIndexReadArg *) basehelper;
  2438. if (resent)
  2439. readContinuationInfo();
  2440. }
  2441. virtual StringBuffer &toString(StringBuffer &ret) const
  2442. {
  2443. return ret.appendf("IndexRead %u", packet->queryHeader().activityId);
  2444. }
  2445. /* Notes on Global smart stepping implementation:
  2446. When smart stepping, I get from the Roxie server:
  2447. 1 or more seek positions
  2448. some flags
  2449. I can read from the index in an order that matches the seek positions order (because of index merger), and which match my hard filter (they may not match my postfilter)
  2450. I can skip forwards in the index to the first record GE a skip position
  2451. I am not going to try to implement a (possible future) flag to return mismatches after any but the last of the seek positions (yet)
  2452. I want to return M matching (keyed matches, seek field matches, and postfilter matches) rows for all provided seek positions
  2453. where M is 1 if SSEFonlyReturnFirstSeekMatch flag set, otherwise all...
  2454. THEN once we are beyond the last provided seek:
  2455. if returnMismatches flag set,
  2456. current row = (next row matching keyed filter)
  2457. if someNewAsYetUnnamedAndUnimplementedFlag flag set
  2458. //if the post filter matches then there may be scope for returning all records which match the seek fields of that first match
  2459. // But not very likely to help unless terms correlated, and may well hinder
  2460. return current row and all following rows matching keyed filter, seek data from current row, and postfilter
  2461. return next row matching keyed filter (even if doesn't match postfilter)
  2462. else
  2463. return next N rows matching keyed filter and postfilter (where N depends on readAheadManyResults flag - 1 if not set)
  2464. */
  2465. inline void advanceToNextSeek()
  2466. {
  2467. assertex(numSeeks != 0);
  2468. if (--numSeeks)
  2469. steppingRow += steppingLength;
  2470. else
  2471. steppingRow = NULL;
  2472. }
  2473. virtual IMessagePacker *process()
  2474. {
  2475. MTIME_SECTION(queryActiveTimer(), "CRoxieIndexReadActivity ::process");
  2476. unsigned __int64 keyedLimit = readHelper->getKeyedLimit();
  2477. unsigned __int64 limit = readHelper->getRowLimit();
  2478. if (!resent && (keyedLimit != (unsigned __int64) -1) && ((keyedLimit > preabortIndexReadsThreshold) || (indexHelper->getFlags() & TIRcountkeyedlimit) != 0)) // Don't recheck the limit every time!
  2479. {
  2480. if (!checkLimit(keyedLimit))
  2481. {
  2482. limitExceeded(true);
  2483. return NULL;
  2484. }
  2485. }
  2486. unsigned __int64 stopAfter = readHelper->getChooseNLimit();
  2487. Owned<IMessagePacker> output = ROQ->createOutputStream(packet->queryHeader(), false, logctx);
  2488. OptimizedRowBuilder rowBuilder(rowAllocator, meta, output, serializer);
  2489. unsigned totalSizeSent = 0;
  2490. unsigned skipped = 0;
  2491. unsigned processedBefore = processed;
  2492. unsigned keyprocessedBefore = keyprocessed;
  2493. bool continuationFailed = false;
  2494. const byte *rawSeek = NULL;
  2495. if (steppingRow)
  2496. rawSeek = steppingRow;
  2497. bool continuationNeeded = false;
  2498. ScopedAtomic<unsigned> indexRecordsRead(::indexRecordsRead);
  2499. ScopedAtomic<unsigned> postFiltered(::postFiltered);
  2500. while (!aborted && inputsDone < inputCount)
  2501. {
  2502. if (!resent || !steppingOffset) // Bit of a hack... In the resent case, we have already set up the tlk, and all keys are processed at once in the steppingOffset case (which makes checkPartChanged gives a false positive in this case)
  2503. checkPartChanged(inputData[inputsDone]);
  2504. if (tlk)
  2505. {
  2506. createSegmentMonitors();
  2507. tlk->reset(resent);
  2508. resent = false;
  2509. {
  2510. TransformCallbackAssociation associate(callback, tlk); // want to destroy this before we advance to next key...
  2511. while (!aborted && (rawSeek ? tlk->lookupSkip(rawSeek, steppingOffset, steppingLength) : tlk->lookup(true)))
  2512. {
  2513. rawSeek = NULL; // only want to do the seek first time we look for a particular seek value
  2514. keyprocessed++;
  2515. if ((keyedLimit != (unsigned __int64) -1) && keyprocessed > keyedLimit)
  2516. {
  2517. noteStats(keyprocessed-keyprocessedBefore, skipped);
  2518. limitExceeded(true);
  2519. break;
  2520. }
  2521. indexRecordsRead++;
  2522. size32_t transformedSize;
  2523. const byte * keyRow = tlk->queryKeyBuffer();
  2524. int diff = 0;
  2525. if (steppingRow)
  2526. {
  2527. diff = memcmp(keyRow+steppingOffset, steppingRow, steppingLength);
  2528. assertex(diff >= 0);
  2529. }
  2530. while (diff > 0)
  2531. {
  2532. advanceToNextSeek();
  2533. if (!steppingRow)
  2534. break;
  2535. diff = memcmp(keyRow+steppingOffset, steppingRow, steppingLength);
  2536. if (diff < 0)
  2537. {
  2538. rawSeek = steppingRow;
  2539. break;
  2540. }
  2541. }
  2542. if (diff >= 0)
  2543. {
  2544. if (diff > 0 && seeksAreEof)
  2545. {
  2546. assertex(!steppingRow);
  2547. break;
  2548. }
  2549. rowBuilder.ensureRow();
  2550. transformedSize = readHelper->transform(rowBuilder, keyRow);
  2551. callback.finishedRow();
  2552. if (transformedSize)
  2553. {
  2554. if (logctx.queryTraceLevel() > 15)
  2555. {
  2556. StringBuffer b;
  2557. for (unsigned j = 0; j < (steppingLength ? steppingLength : 6); j++)
  2558. b.appendf("%02x ", keyRow[steppingOffset + j]);
  2559. logctx.CTXLOG("Returning seek row %s", b.str());
  2560. }
  2561. // Did get a match
  2562. processed++;
  2563. if (limit && processed > limit)
  2564. {
  2565. noteStats(keyprocessed-keyprocessedBefore, skipped);
  2566. limitExceeded(false);
  2567. break;
  2568. }
  2569. if (processed > stopAfter)
  2570. {
  2571. noteStats(keyprocessed-keyprocessedBefore, skipped);
  2572. return output.getClear();
  2573. }
  2574. rowBuilder.writeToOutput(transformedSize, true);
  2575. totalSizeSent += transformedSize;
  2576. if (totalSizeSent > indexReadChunkSize || (steppingOffset && !steppingRow && !stepExtra.readAheadManyResults()))
  2577. continuationNeeded = true;
  2578. lastRowCompleteMatch = true;
  2579. if (steppingRow && stepExtra.onlyReturnFirstSeekMatch())
  2580. advanceToNextSeek();
  2581. }
  2582. else
  2583. {
  2584. // Didn't get a match
  2585. if (steppingOffset && !steppingRow && stepExtra.returnMismatches())
  2586. {
  2587. transformedSize = readHelper->unfilteredTransform(rowBuilder, keyRow);
  2588. if (transformedSize) // will only be zero in odd situations where codegen can't work out how to transform (eg because of a skip)
  2589. {
  2590. callback.finishedRow();
  2591. rowBuilder.writeToOutput(transformedSize, true);
  2592. totalSizeSent += transformedSize;
  2593. continuationNeeded = true;
  2594. lastRowCompleteMatch = false;
  2595. }
  2596. }
  2597. else
  2598. {
  2599. postFiltered++;
  2600. skipped++;
  2601. }
  2602. }
  2603. }
  2604. if (continuationNeeded && !continuationFailed)
  2605. {
  2606. if (logctx.queryTraceLevel() > 10)
  2607. logctx.CTXLOG("Indexread returning partial result set %d rows from %d seeks, %d scans, %d skips", processed-processedBefore, tlk->querySeeks(), tlk->queryScans(), tlk->querySkips());
  2608. if (sendContinuation(output))
  2609. {
  2610. noteStats(keyprocessed-keyprocessedBefore, skipped);
  2611. return output.getClear();
  2612. }
  2613. else
  2614. {
  2615. // This is actually pretty fatal for smart-stepping case
  2616. if (logctx.queryTraceLevel())
  2617. logctx.CTXLOG("Indexread unable to return partial result set");
  2618. continuationFailed = true;
  2619. }
  2620. }
  2621. rowBuilder.clear();
  2622. }
  2623. }
  2624. }
  2625. if (steppingOffset)
  2626. inputsDone = inputCount;
  2627. else
  2628. inputsDone++;
  2629. }
  2630. if (tlk) // a very early abort can mean it is NULL.... MORE is this the right place to put it or should it be inside the loop??
  2631. {
  2632. if (logctx.queryTraceLevel() > 10 && !aborted)
  2633. {
  2634. logctx.CTXLOG("Indexread returning result set %d rows from %d seeks, %d scans, %d skips", processed-processedBefore, tlk->querySeeks(), tlk->queryScans(), tlk->querySkips());
  2635. if (steppingOffset)
  2636. logctx.CTXLOG("Indexread return: steppingOffset %d, steppingRow %p, stepExtra.returnMismatches() %d",steppingOffset, steppingRow, (int) stepExtra.returnMismatches());
  2637. }
  2638. noteStats(keyprocessed-keyprocessedBefore, skipped);
  2639. }
  2640. if (aborted)
  2641. return NULL;
  2642. else
  2643. return output.getClear();
  2644. }
  2645. virtual IIndexReadActivityInfo *queryIndexReadActivity()
  2646. {
  2647. return this;
  2648. }
  2649. virtual IKeyArray *getKeySet() const
  2650. {
  2651. return keyArray.getLink();
  2652. }
  2653. virtual const IResolvedFile *getVarFileInfo() const
  2654. {
  2655. return varFileInfo.getLink();
  2656. }
  2657. virtual ITranslatorSet *getTranslators() const override
  2658. {
  2659. return translators.getLink();
  2660. }
  2661. virtual void mergeSegmentMonitors(IIndexReadContext *irc) const
  2662. {
  2663. indexHelper->createSegmentMonitors(irc); // NOTE: they will merge;
  2664. }
  2665. virtual IRoxieServerActivity *queryActivity() { throwUnexpected(); }
  2666. virtual const RemoteActivityId &queryRemoteId() const { throwUnexpected(); }
  2667. };
  2668. //================================================================================================
  2669. class CRoxieIndexReadActivityFactory : public CRoxieIndexActivityFactory
  2670. {
  2671. unsigned steppingOffset;
  2672. public:
  2673. CRoxieIndexReadActivityFactory(IPropertyTree &graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  2674. : CRoxieIndexActivityFactory(graphNode, _subgraphId, _queryFactory, _helperFactory)
  2675. {
  2676. Owned<IHThorIndexReadArg> helper = (IHThorIndexReadArg *) helperFactory();
  2677. init(helper, graphNode);
  2678. ISteppingMeta *rawMeta = helper->queryRawSteppingMeta();
  2679. if (rawMeta)
  2680. {
  2681. // MORE - should check all keys in maxFields list can actually be keyed.
  2682. const CFieldOffsetSize * fields = rawMeta->queryFields();
  2683. steppingOffset = fields[0].offset;
  2684. }
  2685. else
  2686. {
  2687. steppingOffset = 0;
  2688. }
  2689. }
  2690. virtual IRoxieSlaveActivity *createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const
  2691. {
  2692. return new CRoxieIndexReadActivity(logctx, packet, helperFactory, this, steppingOffset);
  2693. }
  2694. virtual StringBuffer &toString(StringBuffer &s) const
  2695. {
  2696. return CSlaveActivityFactory::toString(s.append("INDEXREAD "));
  2697. }
  2698. };
  2699. ISlaveActivityFactory *createRoxieIndexReadActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  2700. {
  2701. return new CRoxieIndexReadActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory);
  2702. }
  2703. //================================================================================================
  2704. //MORE: Very similar to the indexRead code, but I'm not sure it's worth commoning up....
  2705. class CRoxieIndexNormalizeActivity : public CRoxieIndexActivity
  2706. {
  2707. protected:
  2708. IHThorCompoundNormalizeExtra * normalizeHelper;
  2709. public:
  2710. CRoxieIndexNormalizeActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CRoxieIndexActivityFactory *_aFactory)
  2711. : CRoxieIndexActivity(_logctx, _packet, _hFactory, _aFactory, 0) //MORE - stepping?
  2712. {
  2713. onCreate();
  2714. normalizeHelper = (IHThorIndexNormalizeArg *) basehelper;
  2715. if (resent)
  2716. readContinuationInfo();
  2717. }
  2718. virtual StringBuffer &toString(StringBuffer &ret) const
  2719. {
  2720. return ret.appendf("IndexNormalize %u", packet->queryHeader().activityId);
  2721. }
  2722. virtual IMessagePacker *process()
  2723. {
  2724. MTIME_SECTION(queryActiveTimer(), "CRoxieIndexNormalizeActivity ::process");
  2725. unsigned __int64 keyedLimit = normalizeHelper->getKeyedLimit();
  2726. unsigned __int64 rowLimit = normalizeHelper->getRowLimit();
  2727. if (!resent && (keyedLimit != (unsigned __int64) -1) && (indexHelper->getFlags() & TIRcountkeyedlimit) != 0) // Don't recheck the limit every time!
  2728. {
  2729. if (!checkLimit(keyedLimit))
  2730. {
  2731. limitExceeded(true);
  2732. return NULL;
  2733. }
  2734. }
  2735. unsigned __int64 stopAfter = normalizeHelper->getChooseNLimit();
  2736. Owned<IMessagePacker> output = ROQ->createOutputStream(packet->queryHeader(), false, logctx);
  2737. OptimizedRowBuilder rowBuilder(rowAllocator, meta, output, serializer);
  2738. unsigned totalSizeSent = 0;
  2739. unsigned skipped = 0;
  2740. unsigned processedBefore = processed;
  2741. ScopedAtomic<unsigned> indexRecordsRead(::indexRecordsRead);
  2742. ScopedAtomic<unsigned> postFiltered(::postFiltered);
  2743. bool continuationFailed = false;
  2744. while (!aborted && inputsDone < inputCount)
  2745. {
  2746. checkPartChanged(inputData[inputsDone]);
  2747. if (tlk)
  2748. {
  2749. createSegmentMonitors();
  2750. tlk->reset(resent);
  2751. resent = false;
  2752. TransformCallbackAssociation associate(callback, tlk);
  2753. while (!aborted && tlk->lookup(true))
  2754. {
  2755. keyprocessed++;
  2756. if (keyedLimit && processed > keyedLimit)
  2757. {
  2758. noteStats(processed-processedBefore, skipped);
  2759. limitExceeded(true);
  2760. break;
  2761. }
  2762. indexRecordsRead++;
  2763. if (normalizeHelper->first(tlk->queryKeyBuffer()))
  2764. {
  2765. do
  2766. {
  2767. rowBuilder.ensureRow();
  2768. size32_t transformedSize = normalizeHelper->transform(rowBuilder);
  2769. if (transformedSize)
  2770. {
  2771. processed++;
  2772. if (processed > rowLimit)
  2773. {
  2774. limitExceeded(false);
  2775. break;
  2776. }
  2777. if (processed > stopAfter)
  2778. {
  2779. noteStats(processed-processedBefore, skipped);
  2780. return output.getClear();
  2781. }
  2782. totalSizeSent += rowBuilder.writeToOutput(transformedSize, true);
  2783. }
  2784. } while (normalizeHelper->next());
  2785. callback.finishedRow();
  2786. if (totalSizeSent > indexReadChunkSize && !continuationFailed)
  2787. {
  2788. if (sendContinuation(output))
  2789. {
  2790. noteStats(processed-processedBefore, skipped);
  2791. return output.getClear();
  2792. }
  2793. else
  2794. continuationFailed = true;
  2795. }
  2796. }
  2797. else
  2798. {
  2799. postFiltered++;
  2800. skipped++;
  2801. }
  2802. }
  2803. }
  2804. inputsDone++;
  2805. }
  2806. noteStats(processed-processedBefore, skipped);
  2807. if (aborted)
  2808. return NULL;
  2809. else
  2810. return output.getClear();
  2811. }
  2812. };
  2813. //================================================================================================
  2814. class CRoxieIndexNormalizeActivityFactory : public CRoxieIndexActivityFactory
  2815. {
  2816. public:
  2817. CRoxieIndexNormalizeActivityFactory(IPropertyTree &graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  2818. : CRoxieIndexActivityFactory(graphNode, _subgraphId, _queryFactory, _helperFactory)
  2819. {
  2820. Owned<IHThorIndexNormalizeArg> helper = (IHThorIndexNormalizeArg *) helperFactory();
  2821. init(helper, graphNode);
  2822. }
  2823. virtual IRoxieSlaveActivity *createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const
  2824. {
  2825. return new CRoxieIndexNormalizeActivity(logctx, packet, helperFactory, this);
  2826. }
  2827. virtual StringBuffer &toString(StringBuffer &s) const
  2828. {
  2829. return CSlaveActivityFactory::toString(s.append("IndexNormalize "));
  2830. }
  2831. };
  2832. ISlaveActivityFactory *createRoxieIndexNormalizeActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  2833. {
  2834. return new CRoxieIndexNormalizeActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory);
  2835. }
  2836. //================================================================================================
  2837. class CRoxieIndexCountActivity : public CRoxieIndexActivity
  2838. {
  2839. protected:
  2840. IHThorIndexCountArg * countHelper;
  2841. unsigned __int64 choosenLimit;
  2842. unsigned __int64 rowLimit;
  2843. unsigned __int64 keyedLimit;
  2844. public:
  2845. CRoxieIndexCountActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CRoxieIndexActivityFactory *_aFactory)
  2846. : CRoxieIndexActivity(_logctx, _packet, _hFactory, _aFactory, 0)
  2847. {
  2848. onCreate();
  2849. countHelper = (IHThorIndexCountArg *) basehelper;
  2850. assertex(!resent);
  2851. choosenLimit = countHelper->getChooseNLimit();
  2852. rowLimit = countHelper->getRowLimit();
  2853. keyedLimit = countHelper->getKeyedLimit();
  2854. }
  2855. virtual StringBuffer &toString(StringBuffer &ret) const
  2856. {
  2857. return ret.appendf("IndexCount %u", packet->queryHeader().activityId);
  2858. }
  2859. virtual IMessagePacker *process()
  2860. {
  2861. MTIME_SECTION(queryActiveTimer(), "CRoxieIndexCountActivity ::process");
  2862. Owned<IMessagePacker> output = ROQ->createOutputStream(packet->queryHeader(), false, logctx);
  2863. unsigned skipped = 0;
  2864. unsigned processedBefore = processed;
  2865. unsigned __int64 count = 0;
  2866. ScopedAtomic<unsigned> indexRecordsRead(::indexRecordsRead);
  2867. while (!aborted && inputsDone < inputCount && count < choosenLimit)
  2868. {
  2869. checkPartChanged(inputData[inputsDone]);
  2870. if (tlk)
  2871. {
  2872. createSegmentMonitors();
  2873. tlk->reset(false);
  2874. if (countHelper->hasFilter())
  2875. {
  2876. callback.setManager(tlk);
  2877. while (!aborted && (count < choosenLimit) && tlk->lookup(true))
  2878. {
  2879. keyprocessed++;
  2880. indexRecordsRead++;
  2881. count += countHelper->numValid(tlk->queryKeyBuffer());
  2882. if (count > rowLimit)
  2883. limitExceeded(false);
  2884. else if (count > keyedLimit)
  2885. limitExceeded(true);
  2886. callback.finishedRow();
  2887. }
  2888. callback.setManager(NULL);
  2889. }
  2890. else
  2891. {
  2892. //MORE: GH->RKC There should be value in providing a choosenLimit to getCount()
  2893. //MORE: note that tlk->checkCount() is NOT suitable as it makes assumptions about the segmonitors (only checks leading ones)
  2894. count += tlk->getCount();
  2895. if (count > rowLimit)
  2896. limitExceeded(false);
  2897. else if (count > keyedLimit)
  2898. limitExceeded(true); // MORE - is this right?
  2899. }
  2900. }
  2901. inputsDone++;
  2902. }
  2903. if (!aborted && count)
  2904. {
  2905. if (count > choosenLimit)
  2906. count = choosenLimit;
  2907. processed++;
  2908. assertex(!serializer);
  2909. void *recBuffer = output->getBuffer(meta.getFixedSize(), false);
  2910. if (meta.getFixedSize() == 1)
  2911. *(byte *)recBuffer = (byte)count;
  2912. else
  2913. {
  2914. assertex(meta.getFixedSize() == sizeof(unsigned __int64));
  2915. *(unsigned __int64 *)recBuffer = count;
  2916. }
  2917. output->putBuffer(recBuffer, meta.getFixedSize(), false);
  2918. }
  2919. noteStats(processed-processedBefore, skipped);
  2920. if (aborted)
  2921. return NULL;
  2922. else
  2923. return output.getClear();
  2924. }
  2925. };
  2926. //================================================================================================
  2927. class CRoxieIndexCountActivityFactory : public CRoxieIndexActivityFactory
  2928. {
  2929. public:
  2930. CRoxieIndexCountActivityFactory(IPropertyTree &graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  2931. : CRoxieIndexActivityFactory(graphNode, _subgraphId, _queryFactory, _helperFactory)
  2932. {
  2933. Owned<IHThorIndexCountArg> helper = (IHThorIndexCountArg *) helperFactory();
  2934. init(helper, graphNode);
  2935. }
  2936. virtual IRoxieSlaveActivity *createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const
  2937. {
  2938. return new CRoxieIndexCountActivity(logctx, packet, helperFactory, this);
  2939. }
  2940. virtual StringBuffer &toString(StringBuffer &s) const
  2941. {
  2942. return CSlaveActivityFactory::toString(s.append("INDEXCOUNT "));
  2943. }
  2944. };
  2945. ISlaveActivityFactory *createRoxieIndexCountActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  2946. {
  2947. return new CRoxieIndexCountActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory);
  2948. }
  2949. //================================================================================================
  2950. class CRoxieIndexAggregateActivity : public CRoxieIndexActivity
  2951. {
  2952. protected:
  2953. IHThorCompoundAggregateExtra * aggregateHelper;
  2954. public:
  2955. CRoxieIndexAggregateActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CRoxieIndexActivityFactory *_aFactory)
  2956. : CRoxieIndexActivity(_logctx, _packet, _hFactory, _aFactory, 0)
  2957. {
  2958. onCreate();
  2959. aggregateHelper = (IHThorIndexAggregateArg *) basehelper;
  2960. assertex(!resent);
  2961. }
  2962. virtual StringBuffer &toString(StringBuffer &ret) const
  2963. {
  2964. return ret.appendf("IndexAggregate %u", packet->queryHeader().activityId);
  2965. }
  2966. virtual IMessagePacker *process()
  2967. {
  2968. MTIME_SECTION(queryActiveTimer(), "CRoxieIndexAggregateActivity ::process");
  2969. Owned<IMessagePacker> output = ROQ->createOutputStream(packet->queryHeader(), false, logctx);
  2970. OptimizedRowBuilder rowBuilder(rowAllocator, meta, output, serializer);
  2971. rowBuilder.ensureRow();
  2972. aggregateHelper->clearAggregate(rowBuilder);
  2973. unsigned skipped = 0;
  2974. unsigned processedBefore = processed;
  2975. ScopedAtomic<unsigned> indexRecordsRead(::indexRecordsRead);
  2976. while (!aborted && inputsDone < inputCount)
  2977. {
  2978. checkPartChanged(inputData[inputsDone]);
  2979. if (tlk)
  2980. {
  2981. createSegmentMonitors();
  2982. tlk->reset(false);
  2983. callback.setManager(tlk);
  2984. while (!aborted && tlk->lookup(true))
  2985. {
  2986. keyprocessed++;
  2987. indexRecordsRead++;
  2988. aggregateHelper->processRow(rowBuilder, tlk->queryKeyBuffer());
  2989. callback.finishedRow();
  2990. }
  2991. callback.setManager(NULL);
  2992. }
  2993. inputsDone++;
  2994. }
  2995. if (!aborted && aggregateHelper->processedAnyRows())
  2996. {
  2997. processed++;
  2998. size32_t transformedSize = meta.getRecordSize(rowBuilder.getSelf());
  2999. rowBuilder.writeToOutput(transformedSize, true);
  3000. }
  3001. noteStats(processed-processedBefore, skipped);
  3002. if (aborted)
  3003. return NULL;
  3004. else
  3005. return output.getClear();
  3006. }
  3007. };
  3008. //================================================================================================
  3009. class CRoxieIndexAggregateActivityFactory : public CRoxieIndexActivityFactory
  3010. {
  3011. public:
  3012. CRoxieIndexAggregateActivityFactory(IPropertyTree &graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  3013. : CRoxieIndexActivityFactory(graphNode, _subgraphId, _queryFactory, _helperFactory)
  3014. {
  3015. Owned<IHThorIndexAggregateArg> helper = (IHThorIndexAggregateArg *) helperFactory();
  3016. init(helper, graphNode);
  3017. }
  3018. virtual IRoxieSlaveActivity *createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const
  3019. {
  3020. return new CRoxieIndexAggregateActivity(logctx, packet, helperFactory, this);
  3021. }
  3022. virtual StringBuffer &toString(StringBuffer &s) const
  3023. {
  3024. return CSlaveActivityFactory::toString(s.append("INDEXAGGREGATE "));
  3025. }
  3026. };
  3027. ISlaveActivityFactory *createRoxieIndexAggregateActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  3028. {
  3029. return new CRoxieIndexAggregateActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory);
  3030. }
  3031. //================================================================================================
  3032. class CRoxieIndexGroupAggregateActivity : public CRoxieIndexActivity, implements IHThorGroupAggregateCallback
  3033. {
  3034. protected:
  3035. IHThorCompoundGroupAggregateExtra * aggregateHelper;
  3036. RowAggregator results;
  3037. unsigned groupSegCount;
  3038. ThorActivityKind kind;
  3039. public:
  3040. IMPLEMENT_IINTERFACE_USING(CRoxieIndexActivity)
  3041. CRoxieIndexGroupAggregateActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CRoxieIndexActivityFactory *_aFactory, ThorActivityKind _kind)
  3042. : CRoxieIndexActivity(_logctx, _packet, _hFactory, _aFactory, 0),
  3043. aggregateHelper((IHThorIndexGroupAggregateArg *) basehelper),
  3044. results(*aggregateHelper, *aggregateHelper), kind(_kind)
  3045. {
  3046. onCreate();
  3047. results.start(rowAllocator, queryContext->queryCodeContext(), basefactory->queryId());
  3048. assertex(!resent);
  3049. groupSegCount = 0;
  3050. }
  3051. virtual bool needsRowAllocator()
  3052. {
  3053. return true;
  3054. }
  3055. virtual StringBuffer &toString(StringBuffer &ret) const
  3056. {
  3057. return ret.appendf("IndexGroupAggregate %u", packet->queryHeader().activityId);
  3058. }
  3059. virtual void processRow(const void * next)
  3060. {
  3061. results.addRow(next);
  3062. }
  3063. virtual void createSegmentMonitors()
  3064. {
  3065. if (createSegmentMonitorsPending)
  3066. {
  3067. CRoxieIndexActivity::createSegmentMonitors();
  3068. if ((kind==TAKindexgroupcount || kind==TAKindexgroupexists))
  3069. groupSegCount = aggregateHelper->getGroupingMaxField();
  3070. else
  3071. groupSegCount = 0;
  3072. }
  3073. }
  3074. virtual IMessagePacker *process()
  3075. {
  3076. MTIME_SECTION(queryActiveTimer(), "CRoxieIndexGroupAggregateActivity ::process");
  3077. Owned<IRowManager> rowManager = roxiemem::createRowManager(0, NULL, logctx, NULL, true); // MORE - should not really use default limits
  3078. Owned<IMessagePacker> output = ROQ->createOutputStream(packet->queryHeader(), false, logctx);
  3079. unsigned processedBefore = processed;
  3080. ScopedAtomic<unsigned> indexRecordsRead(::indexRecordsRead);
  3081. try
  3082. {
  3083. while (!aborted && inputsDone < inputCount)
  3084. {
  3085. checkPartChanged(inputData[inputsDone]);
  3086. if (tlk)
  3087. {
  3088. createSegmentMonitors();
  3089. tlk->reset(false);
  3090. callback.setManager(tlk);
  3091. while (!aborted && tlk->lookup(true))
  3092. {
  3093. if (groupSegCount && !translators->queryTranslator(lastPartNo.fileNo))
  3094. {
  3095. AggregateRowBuilder &rowBuilder = results.addRow(tlk->queryKeyBuffer());
  3096. callback.finishedRow();
  3097. if (kind == TAKindexgroupcount)
  3098. {
  3099. unsigned __int64 count = tlk->getCurrentRangeCount(groupSegCount);
  3100. aggregateHelper->processCountGrouping(rowBuilder, count-1);
  3101. }
  3102. if (!tlk->nextRange(groupSegCount))
  3103. break;
  3104. }
  3105. else
  3106. {
  3107. keyprocessed++;
  3108. indexRecordsRead++;
  3109. aggregateHelper->processRow(tlk->queryKeyBuffer(), this);
  3110. callback.finishedRow();
  3111. }
  3112. }
  3113. callback.setManager(NULL);
  3114. }
  3115. inputsDone++;
  3116. }
  3117. if (!aborted)
  3118. {
  3119. for (;;)
  3120. {
  3121. Owned<AggregateRowBuilder> next = results.nextResult();
  3122. if (!next)
  3123. break;
  3124. unsigned rowSize = next->querySize();
  3125. OwnedConstRoxieRow row(next->finalizeRowClear());
  3126. if (serializer)
  3127. {
  3128. serializeRow(output, row);
  3129. }
  3130. else
  3131. {
  3132. void *recBuffer = output->getBuffer(rowSize, meta.isVariableSize());
  3133. memcpy(recBuffer, row, rowSize);
  3134. output->putBuffer(recBuffer, rowSize, meta.isVariableSize());
  3135. }
  3136. }
  3137. }
  3138. }
  3139. catch (...)
  3140. {
  3141. results.reset(); // kill entries before the rowManager dies.
  3142. throw;
  3143. }
  3144. results.reset();
  3145. noteStats(processed-processedBefore, 0);
  3146. if (aborted)
  3147. return NULL;
  3148. else
  3149. return output.getClear();
  3150. }
  3151. };
  3152. //================================================================================================
  3153. class CRoxieIndexGroupAggregateActivityFactory : public CRoxieIndexActivityFactory
  3154. {
  3155. ThorActivityKind kind;
  3156. public:
  3157. CRoxieIndexGroupAggregateActivityFactory(IPropertyTree &graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory, ThorActivityKind _kind)
  3158. : CRoxieIndexActivityFactory(graphNode, _subgraphId, _queryFactory, _helperFactory), kind(_kind)
  3159. {
  3160. Owned<IHThorIndexGroupAggregateArg> helper = (IHThorIndexGroupAggregateArg *) helperFactory();
  3161. init(helper, graphNode);
  3162. }
  3163. virtual IRoxieSlaveActivity *createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const
  3164. {
  3165. return new CRoxieIndexGroupAggregateActivity(logctx, packet, helperFactory, this, kind);
  3166. }
  3167. virtual StringBuffer &toString(StringBuffer &s) const
  3168. {
  3169. return CSlaveActivityFactory::toString(s.append("INDEXGROUPAGGREGATE "));
  3170. }
  3171. };
  3172. ISlaveActivityFactory *createRoxieIndexGroupAggregateActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory, ThorActivityKind _kind)
  3173. {
  3174. return new CRoxieIndexGroupAggregateActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory, _kind);
  3175. }
  3176. //================================================================================================
  3177. class CRoxieFetchActivityFactory : public CSlaveActivityFactory
  3178. {
  3179. public:
  3180. Owned<ITranslatorSet> translators;
  3181. Owned<IFileIOArray> fileArray;
  3182. CRoxieFetchActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  3183. : CSlaveActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory)
  3184. {
  3185. Owned<IHThorFetchBaseArg> helper = (IHThorFetchBaseArg *) helperFactory();
  3186. bool variableFileName = allFilesDynamic || queryFactory.isDynamic() || ((helper->getFetchFlags() & (FFvarfilename|FFdynamicfilename)) != 0);
  3187. if (!variableFileName)
  3188. {
  3189. bool isOpt = (helper->getFetchFlags() & FFdatafileoptional) != 0;
  3190. OwnedRoxieString fname(helper->getFileName());
  3191. datafile.setown(_queryFactory.queryPackage().lookupFileName(fname, isOpt, true, true, _queryFactory.queryWorkUnit(), true));
  3192. if (datafile)
  3193. {
  3194. unsigned formatCrc = getFormatCrc(helper->getDiskFormatCrc());
  3195. translators.setown(datafile->getTranslators(formatCrc, helper->queryProjectedDiskRecordSize(), helper->queryDiskRecordSize(), getEnableFieldTranslation(), false));
  3196. fileArray.setown(datafile->getIFileIOArray(isOpt, queryFactory.queryChannel()));
  3197. }
  3198. }
  3199. }
  3200. virtual IRoxieSlaveActivity *createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const;
  3201. virtual StringBuffer &toString(StringBuffer &s) const
  3202. {
  3203. return CSlaveActivityFactory::toString(s.append("FETCH "));
  3204. }
  3205. };
  3206. class CRoxieFetchActivityBase : public CRoxieSlaveActivity
  3207. {
  3208. protected:
  3209. IHThorFetchBaseArg *helper;
  3210. const CRoxieFetchActivityFactory *factory;
  3211. Owned<IFileIO> rawFile;
  3212. Owned<ISerialStream> rawStream;
  3213. offset_t base;
  3214. char *inputData;
  3215. char *inputLimit;
  3216. Linked<ITranslatorSet> translators;
  3217. Linked<IFileIOArray> files;
  3218. const IDynamicTransform *translator = nullptr;
  3219. bool needsRHS;
  3220. virtual size32_t doFetch(ARowBuilder & rowBuilder, offset_t pos, offset_t rawpos, void *inputData) = 0;
  3221. public:
  3222. CRoxieFetchActivityBase(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory,
  3223. const CRoxieFetchActivityFactory *_aFactory,
  3224. ITranslatorSet *_translators,
  3225. IFileIOArray *_files)
  3226. : CRoxieSlaveActivity(_logctx, _packet, _hFactory, _aFactory), factory(_aFactory), translators(_translators), files(_files)
  3227. {
  3228. helper = (IHThorFetchBaseArg *) basehelper;
  3229. base = 0;
  3230. variableFileName = allFilesDynamic || basefactory->queryQueryFactory().isDynamic() || ((helper->getFetchFlags() & (FFvarfilename|FFdynamicfilename)) != 0);
  3231. isOpt = (helper->getFetchFlags() & FFdatafileoptional) != 0;
  3232. onCreate();
  3233. inputData = (char *) serializedCreate.readDirect(0);
  3234. inputLimit = inputData + (serializedCreate.length() - serializedCreate.getPos());
  3235. needsRHS = helper->transformNeedsRhs();
  3236. }
  3237. virtual const char *queryDynamicFileName() const
  3238. {
  3239. return helper->getFileName();
  3240. }
  3241. virtual void setVariableFileInfo()
  3242. {
  3243. unsigned formatCrc = basefactory->getFormatCrc(helper->getDiskFormatCrc());
  3244. translators.setown(varFileInfo->getTranslators(formatCrc, helper->queryProjectedDiskRecordSize(), helper->queryDiskRecordSize(), basefactory->getEnableFieldTranslation(), false));
  3245. files.setown(varFileInfo->getIFileIOArray(isOpt, packet->queryHeader().channel));
  3246. }
  3247. virtual IMessagePacker *process();
  3248. virtual StringBuffer &toString(StringBuffer &ret) const
  3249. {
  3250. return ret.appendf("Fetch %u", packet->queryHeader().activityId);
  3251. }
  3252. virtual void setPartNo(bool filechanged);
  3253. };
  3254. IMessagePacker *CRoxieFetchActivityBase::process()
  3255. {
  3256. MTIME_SECTION(queryActiveTimer(), "CRoxieFetchActivityBase::process");
  3257. Owned<IMessagePacker> output = ROQ->createOutputStream(packet->queryHeader(), false, logctx);
  3258. unsigned accepted = 0;
  3259. unsigned rejected = 0;
  3260. unsigned __int64 rowLimit = helper->getRowLimit();
  3261. OptimizedRowBuilder rowBuilder(rowAllocator, meta, output, serializer);
  3262. while (!aborted && inputData < inputLimit)
  3263. {
  3264. checkPartChanged(*(PartNoType *) inputData);
  3265. inputData += sizeof(PartNoType);
  3266. offset_t rp = *(offset_t *)inputData;
  3267. inputData += sizeof(offset_t);
  3268. unsigned rhsSize;
  3269. if (needsRHS)
  3270. {
  3271. rhsSize = *(unsigned *)inputData;
  3272. inputData += sizeof(unsigned);
  3273. }
  3274. else
  3275. rhsSize = 0;
  3276. offset_t pos;
  3277. if (isLocalFpos(rp))
  3278. pos = getLocalFposOffset(rp);
  3279. else
  3280. pos = rp-base;
  3281. unsigned thisSize = doFetch(rowBuilder, pos, rp, inputData);
  3282. inputData += rhsSize;
  3283. if (thisSize)
  3284. {
  3285. rowBuilder.writeToOutput(thisSize, true);
  3286. accepted++;
  3287. if (accepted > rowLimit)
  3288. {
  3289. logctx.noteStatistic(StNumDiskSeeks, accepted+rejected);
  3290. logctx.noteStatistic(StNumDiskRowsRead, accepted+rejected);
  3291. logctx.noteStatistic(StNumDiskAccepted, accepted);
  3292. logctx.noteStatistic(StNumDiskRejected, rejected);
  3293. limitExceeded();
  3294. return NULL;
  3295. }
  3296. }
  3297. else
  3298. rejected++;
  3299. }
  3300. logctx.noteStatistic(StNumDiskSeeks, accepted+rejected);
  3301. logctx.noteStatistic(StNumDiskRowsRead, accepted+rejected);
  3302. logctx.noteStatistic(StNumDiskAccepted, accepted);
  3303. logctx.noteStatistic(StNumDiskRejected, rejected);
  3304. if (aborted)
  3305. return NULL;
  3306. else
  3307. return output.getClear();
  3308. }
  3309. class CRoxieFetchActivity : public CRoxieFetchActivityBase
  3310. {
  3311. Owned<IEngineRowAllocator> diskAllocator;
  3312. CThorContiguousRowBuffer prefetchSource;
  3313. Owned<ISourceRowPrefetcher> rowPrefetcher;
  3314. public:
  3315. CRoxieFetchActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory,
  3316. const CRoxieFetchActivityFactory *_aFactory,
  3317. ITranslatorSet *_translators,
  3318. IFileIOArray *_files)
  3319. : CRoxieFetchActivityBase(_logctx, _packet, _hFactory, _aFactory, _translators, _files)
  3320. {
  3321. // If we ever supported superfiles this would need to move to setPartNo, and pass proper subfile idx in
  3322. rowPrefetcher.setown(translators->getPrefetcher(0));
  3323. IOutputMetaData *diskMeta = helper->queryProjectedDiskRecordSize();
  3324. diskAllocator.setown(getRowAllocator(diskMeta, basefactory->queryId()));
  3325. }
  3326. virtual size32_t doFetch(ARowBuilder & rowBuilder, offset_t pos, offset_t rawpos, void *inputData)
  3327. {
  3328. prefetchSource.reset(pos);
  3329. rowPrefetcher->readAhead(prefetchSource);
  3330. const byte *diskRow = prefetchSource.queryRow();
  3331. if (translator)
  3332. {
  3333. MemoryBuffer buf;
  3334. MemoryBufferBuilder aBuilder(buf, 0);
  3335. translator->translate(aBuilder, diskRow);
  3336. // note the swapped parameters - left and right map to input and raw differently for JOIN vs FETCH
  3337. IHThorFetchArg *h = (IHThorFetchArg *) helper;
  3338. return h->transform(rowBuilder, buf.toByteArray(), inputData, rawpos);
  3339. }
  3340. else
  3341. {
  3342. // note the swapped parameters - left and right map to input and raw differently for JOIN vs FETCH
  3343. IHThorFetchArg *h = (IHThorFetchArg *) helper;
  3344. return h->transform(rowBuilder, diskRow, inputData, rawpos);
  3345. }
  3346. }
  3347. virtual void setPartNo(bool filechanged)
  3348. {
  3349. CRoxieFetchActivityBase::setPartNo(filechanged);
  3350. prefetchSource.setStream(rawStream);
  3351. }
  3352. };
  3353. IRoxieSlaveActivity *CRoxieFetchActivityFactory::createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const
  3354. {
  3355. return new CRoxieFetchActivity(logctx, packet, helperFactory, this, translators, fileArray);
  3356. }
  3357. //------------------------------------------------------------------------------------
  3358. class CRoxieCSVFetchActivity : public CRoxieFetchActivityBase
  3359. {
  3360. CSVSplitter csvSplitter;
  3361. size32_t maxRowSize;
  3362. CThorStreamDeserializerSource deserializeSource;
  3363. public:
  3364. CRoxieCSVFetchActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory,
  3365. const CRoxieFetchActivityFactory *_aFactory,
  3366. ITranslatorSet *_translators,
  3367. IFileIOArray *_files,
  3368. unsigned _maxColumns, size32_t _maxRowSize)
  3369. : CRoxieFetchActivityBase(_logctx, _packet, _hFactory, _aFactory, _translators, _files), maxRowSize(_maxRowSize)
  3370. {
  3371. const char * quotes = NULL;
  3372. const char * separators = NULL;
  3373. const char * terminators = NULL;
  3374. const char * escapes = NULL;
  3375. const IResolvedFile *fileInfo = varFileInfo ? varFileInfo : factory->datafile;
  3376. if (fileInfo)
  3377. {
  3378. const IPropertyTree *options = fileInfo->queryProperties();
  3379. if (options)
  3380. {
  3381. quotes = options->queryProp("@csvQuote");
  3382. separators = options->queryProp("@csvSeparate");
  3383. terminators = options->queryProp("@csvTerminate");
  3384. escapes = options->queryProp("@csvEscape");
  3385. }
  3386. }
  3387. IHThorCsvFetchArg *h = (IHThorCsvFetchArg *) helper;
  3388. ICsvParameters *csvInfo = h->queryCsvParameters();
  3389. csvSplitter.init(_maxColumns, csvInfo, quotes, separators, terminators, escapes);
  3390. if (translator)
  3391. UNIMPLEMENTED; // It's not obvious how we would support this.
  3392. }
  3393. virtual size32_t doFetch(ARowBuilder & rowBuilder, offset_t pos, offset_t rawpos, void *inputData)
  3394. {
  3395. IHThorCsvFetchArg *h = (IHThorCsvFetchArg *) helper;
  3396. rawStream->reset(pos);
  3397. size32_t rowSize = 4096; // MORE - make configurable
  3398. for (;;)
  3399. {
  3400. size32_t avail;
  3401. const void *peek = rawStream->peek(rowSize, avail);
  3402. if (csvSplitter.splitLine(avail, (const byte *)peek) < rowSize || avail < rowSize)
  3403. break;
  3404. if (rowSize == maxRowSize)
  3405. throw MakeStringException(0, "File contained a line of length greater than %d bytes.", maxRowSize);
  3406. if (rowSize >= maxRowSize/2)
  3407. rowSize = maxRowSize;
  3408. else
  3409. rowSize += rowSize;
  3410. }
  3411. return h->transform(rowBuilder, csvSplitter.queryLengths(), (const char * *)csvSplitter.queryData(), inputData, rawpos);
  3412. }
  3413. virtual void setPartNo(bool filechanged)
  3414. {
  3415. CRoxieFetchActivityBase::setPartNo(filechanged);
  3416. deserializeSource.setStream(rawStream);
  3417. }
  3418. };
  3419. class CRoxieXMLFetchActivity : public CRoxieFetchActivityBase, implements IXMLSelect
  3420. {
  3421. Owned<IXMLParse> parser;
  3422. Owned<IColumnProvider> lastMatch;
  3423. Owned<IFileIOStream> rawStreamX;
  3424. unsigned streamBufferSize;
  3425. CThorStreamDeserializerSource deserializeSource;
  3426. public:
  3427. IMPLEMENT_IINTERFACE_USING(CRoxieFetchActivityBase)
  3428. CRoxieXMLFetchActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory,
  3429. const CRoxieFetchActivityFactory *_aFactory,
  3430. ITranslatorSet *_translators,
  3431. IFileIOArray *_files,
  3432. unsigned _streamBufferSize)
  3433. : CRoxieFetchActivityBase(_logctx, _packet, _hFactory, _aFactory, _translators, _files),
  3434. streamBufferSize(_streamBufferSize)
  3435. {
  3436. if (translator)
  3437. UNIMPLEMENTED;
  3438. }
  3439. virtual size32_t doFetch(ARowBuilder & rowBuilder, offset_t pos, offset_t rawpos, void *inputData)
  3440. {
  3441. rawStreamX->seek(pos, IFSbegin);
  3442. try
  3443. {
  3444. while(!lastMatch)
  3445. if(!parser->next())
  3446. throw MakeStringException(ROXIE_RECORD_FETCH_ERROR, "XML parse error at position %" I64F "d", pos);
  3447. IHThorXmlFetchArg *h = (IHThorXmlFetchArg *) helper;
  3448. unsigned thisSize = h->transform(rowBuilder, lastMatch, inputData, rawpos);
  3449. lastMatch.clear();
  3450. parser->reset();
  3451. return thisSize;
  3452. }
  3453. catch (IException *E)
  3454. {
  3455. ::Release(E);
  3456. throw MakeStringException(ROXIE_RECORD_FETCH_ERROR, "XML parse error at position %" I64F "d", pos);
  3457. }
  3458. }
  3459. virtual void match(IColumnProvider & entry, offset_t startOffset, offset_t endOffset)
  3460. {
  3461. lastMatch.set(&entry);
  3462. }
  3463. virtual void setPartNo(bool filechanged)
  3464. {
  3465. CRoxieFetchActivityBase::setPartNo(filechanged);
  3466. deserializeSource.setStream(rawStream);
  3467. rawStreamX.setown(createBufferedIOStream(rawFile, streamBufferSize));
  3468. parser.setown((factory->getKind()==TAKjsonfetch) ? createJSONParse(*rawStreamX, "/", *this) : createXMLParse(*rawStreamX, "/", *this));
  3469. }
  3470. };
  3471. void CRoxieFetchActivityBase::setPartNo(bool filechanged)
  3472. {
  3473. rawFile.setown(files->getFilePart(lastPartNo.partNo, base)); // MORE - superfiles
  3474. translator = translators->queryTranslator(0); // MORE - superfiles
  3475. assertex(rawFile != NULL);
  3476. rawStream.setown(createFileSerialStream(rawFile, 0, -1, 0));
  3477. }
  3478. class CRoxieCSVFetchActivityFactory : public CRoxieFetchActivityFactory
  3479. {
  3480. unsigned maxColumns;
  3481. size32_t maxRowSize;
  3482. public:
  3483. CRoxieCSVFetchActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  3484. : CRoxieFetchActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory)
  3485. {
  3486. Owned<IHThorCsvFetchArg> helper = (IHThorCsvFetchArg*) helperFactory();
  3487. maxColumns = helper->getMaxColumns();
  3488. ICsvParameters *csvInfo = helper->queryCsvParameters();
  3489. assertex(!csvInfo->queryEBCDIC());
  3490. maxRowSize = defaultMaxCsvRowSize * 1024 * 1024;
  3491. IConstWorkUnit *workunit = _queryFactory.queryWorkUnit();
  3492. if (workunit)
  3493. maxRowSize = workunit->getDebugValueInt(OPT_MAXCSVROWSIZE, defaultMaxCsvRowSize) * 1024 * 1024;
  3494. }
  3495. virtual IRoxieSlaveActivity *createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const
  3496. {
  3497. return new CRoxieCSVFetchActivity(logctx, packet, helperFactory, this, translators, fileArray, maxColumns, maxRowSize);
  3498. }
  3499. };
  3500. class CRoxieXMLFetchActivityFactory : public CRoxieFetchActivityFactory
  3501. {
  3502. public:
  3503. CRoxieXMLFetchActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  3504. : CRoxieFetchActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory)
  3505. {
  3506. }
  3507. virtual IRoxieSlaveActivity *createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const
  3508. {
  3509. return new CRoxieXMLFetchActivity(logctx, packet, helperFactory, this, translators, fileArray, 4096);
  3510. }
  3511. };
  3512. ISlaveActivityFactory *createRoxieFetchActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  3513. {
  3514. return new CRoxieFetchActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory);
  3515. }
  3516. ISlaveActivityFactory *createRoxieCSVFetchActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  3517. {
  3518. return new CRoxieCSVFetchActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory);
  3519. }
  3520. ISlaveActivityFactory *createRoxieXMLFetchActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  3521. {
  3522. return new CRoxieXMLFetchActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory);
  3523. }
  3524. //================================================================================================
  3525. class CRoxieKeyedJoinIndexActivityFactory : public CRoxieKeyedActivityFactory
  3526. {
  3527. public:
  3528. CRoxieKeyedJoinIndexActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  3529. : CRoxieKeyedActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory)
  3530. {
  3531. Owned<IHThorKeyedJoinArg> helper = (IHThorKeyedJoinArg *) helperFactory();
  3532. bool variableFileName = allFilesDynamic || queryFactory.isDynamic() || ((helper->getJoinFlags() & (JFvarindexfilename|JFdynamicindexfilename|JFindexfromactivity)) != 0);
  3533. formatCrc = getFormatCrc(helper->getIndexFormatCrc());
  3534. projectedMeta = helper->queryProjectedIndexRecordSize();
  3535. expectedMeta = helper->queryIndexRecordSize();
  3536. if (!variableFileName)
  3537. {
  3538. bool isOpt = (helper->getJoinFlags() & JFindexoptional) != 0;
  3539. OwnedRoxieString indexFileName(helper->getIndexFileName());
  3540. datafile.setown(_queryFactory.queryPackage().lookupFileName(indexFileName, isOpt, true, true, _queryFactory.queryWorkUnit(), true));
  3541. if (datafile)
  3542. {
  3543. translators.setown(datafile->getTranslators(formatCrc, projectedMeta, expectedMeta, queryFactory.queryOptions().enableFieldTranslation, true));
  3544. keyArray.setown(datafile->getKeyArray(isOpt, queryFactory.queryChannel()));
  3545. }
  3546. }
  3547. }
  3548. virtual IRoxieSlaveActivity *createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const;
  3549. virtual StringBuffer &toString(StringBuffer &s) const
  3550. {
  3551. return CSlaveActivityFactory::toString(s.append("KEYEDJOIN INDEX "));
  3552. }
  3553. };
  3554. class CRoxieKeyedJoinIndexActivity : public CRoxieKeyedActivity
  3555. {
  3556. IHThorKeyedJoinArg *helper;
  3557. const CRoxieKeyedJoinIndexActivityFactory *factory;
  3558. unsigned inputLength;
  3559. char *inputData;
  3560. Owned<IRoxieSlaveActivity> rootIndexActivity;
  3561. IIndexReadActivityInfo *rootIndex;
  3562. unsigned processed;
  3563. unsigned candidateCount;
  3564. unsigned keepCount;
  3565. unsigned inputDone;
  3566. public:
  3567. CRoxieKeyedJoinIndexActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CRoxieKeyedJoinIndexActivityFactory *_aFactory)
  3568. : factory(_aFactory), CRoxieKeyedActivity(_logctx, _packet, _hFactory, _aFactory)
  3569. {
  3570. helper = (IHThorKeyedJoinArg *) basehelper;
  3571. keyRecInfo = &helper->queryIndexRecordSize()->queryRecordAccessor(true);
  3572. variableFileName = allFilesDynamic || basefactory->queryQueryFactory().isDynamic() || ((helper->getJoinFlags() & (JFvarindexfilename|JFdynamicindexfilename|JFindexfromactivity)) != 0);
  3573. inputDone = 0;
  3574. processed = 0;
  3575. candidateCount = 0;
  3576. keepCount = 0;
  3577. rootIndex = NULL;
  3578. onCreate();
  3579. inputData = (char *) serializedCreate.readDirect(0);
  3580. inputLength = (serializedCreate.length() - serializedCreate.getPos());
  3581. if (resent)
  3582. {
  3583. resentInfo.read(inputDone);
  3584. inputData += inputDone;
  3585. resentInfo.read(processed);
  3586. resentInfo.read(candidateCount);
  3587. resentInfo.read(keepCount);
  3588. resentInfo.read(lastPartNo.partNo);
  3589. resentInfo.read(lastPartNo.fileNo);
  3590. setPartNo(true);
  3591. tlk->deserializeCursorPos(resentInfo);
  3592. assertex(resentInfo.remaining() == 0);
  3593. }
  3594. }
  3595. ~CRoxieKeyedJoinIndexActivity()
  3596. {
  3597. }
  3598. virtual void deserializeExtra(MemoryBuffer &buff)
  3599. {
  3600. if (helper->getJoinFlags() & JFindexfromactivity)
  3601. {
  3602. RemoteActivityId indexActivityId(buff);
  3603. assertex(indexActivityId.activityId);
  3604. unsigned indexCtxLen;
  3605. buff.read(indexCtxLen);
  3606. const void *indexCtx = buff.readDirect(indexCtxLen);
  3607. //We create a packet for the index activity to use. Header, trace info and parentextract are clone of mine, context info is copied from buff
  3608. MemoryBuffer indexPacketData;
  3609. indexPacketData.append(sizeof(RoxiePacketHeader), &packet->queryHeader());
  3610. indexPacketData.append(packet->getTraceLength(), packet->queryTraceInfo());
  3611. const byte *parentExtract = (const byte *) packet->queryContextData();
  3612. unsigned parentExtractLen = *(unsigned*) parentExtract;
  3613. indexPacketData.append(parentExtractLen);
  3614. indexPacketData.append(parentExtractLen, parentExtract + sizeof(unsigned));
  3615. indexPacketData.append(indexCtxLen, indexCtx);
  3616. RoxiePacketHeader *newHeader = (RoxiePacketHeader *) indexPacketData.toByteArray();
  3617. newHeader->continueSequence = 0;
  3618. newHeader->activityId = indexActivityId.activityId;
  3619. newHeader->queryHash = indexActivityId.queryHash;
  3620. Owned<IRoxieQueryPacket> indexPacket = createRoxiePacket(indexPacketData);
  3621. Owned<ISlaveActivityFactory> indexActivityFactory = factory->queryQueryFactory().getSlaveActivityFactory(indexActivityId.activityId);
  3622. assertex(indexActivityFactory != NULL);
  3623. rootIndexActivity.setown(indexActivityFactory->createActivity(logctx, indexPacket));
  3624. rootIndex = rootIndexActivity->queryIndexReadActivity();
  3625. varFileInfo.setown(rootIndex->getVarFileInfo());
  3626. translators.setown(rootIndex->getTranslators());
  3627. keyArray.setown(rootIndex->getKeySet());
  3628. }
  3629. }
  3630. virtual const char *queryDynamicFileName() const
  3631. {
  3632. return helper->getIndexFileName();
  3633. }
  3634. virtual IMessagePacker *process();
  3635. virtual StringBuffer &toString(StringBuffer &ret) const
  3636. {
  3637. return ret.appendf("KeyedJoinIndex %u", packet->queryHeader().activityId);
  3638. }
  3639. virtual void createSegmentMonitors()
  3640. {
  3641. // This is called to create the segmonitors that apply to ALL rows - not the per-row ones
  3642. // At present there are none. However we should still set up the layout translation.
  3643. if (createSegmentMonitorsPending)
  3644. {
  3645. createSegmentMonitorsPending = false;
  3646. tlk->setLayoutTranslator(translators->queryTranslator(lastPartNo.fileNo));
  3647. }
  3648. }
  3649. };
  3650. IRoxieSlaveActivity *CRoxieKeyedJoinIndexActivityFactory::createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const
  3651. {
  3652. return new CRoxieKeyedJoinIndexActivity(logctx, packet, helperFactory, this);
  3653. }
  3654. IMessagePacker *CRoxieKeyedJoinIndexActivity::process()
  3655. {
  3656. MTIME_SECTION(queryActiveTimer(), "CRoxieKeyedJoinIndexActivity::process");
  3657. Owned<IMessagePacker> output = ROQ->createOutputStream(packet->queryHeader(), false, logctx);
  3658. CachedOutputMetaData joinFieldsMeta(helper->queryJoinFieldsRecordSize());
  3659. Owned<IEngineRowAllocator> joinFieldsAllocator = getRowAllocator(joinFieldsMeta, basefactory->queryId());
  3660. OptimizedKJRowBuilder rowBuilder(joinFieldsAllocator, joinFieldsMeta, output);
  3661. unsigned __int64 rowLimit = helper->getRowLimit();
  3662. unsigned atmost = helper->getJoinLimit();
  3663. if (!atmost) atmost = (unsigned) -1;
  3664. unsigned abortLimit = helper->getMatchAbortLimit();
  3665. if (!abortLimit) abortLimit = (unsigned) -1;
  3666. if (abortLimit < atmost)
  3667. atmost = abortLimit;
  3668. unsigned keepLimit = helper->getKeepLimit();
  3669. unsigned joinFlags = helper->getJoinFlags();
  3670. if (joinFlags & (JFtransformMaySkip | JFfetchMayFilter))
  3671. keepLimit = 0;
  3672. if ((joinFlags & (JFexclude|JFleftouter)) == (JFexclude|JFleftouter) && (!(joinFlags & JFfetchMayFilter))) // For left-only joins, all we care about is existence of a match. Return as soon as we know that there is one
  3673. keepLimit = 1;
  3674. unsigned processedBefore = processed;
  3675. unsigned rejected = 0;
  3676. CachedOutputMetaData inputFields(helper->queryIndexReadInputRecordSize());
  3677. unsigned inputSize = inputFields.getFixedSize();
  3678. unsigned totalSizeSent = 0;
  3679. // Now go fetch the records
  3680. bool continuationFailed = false;
  3681. while (!aborted && inputDone < inputLength)
  3682. {
  3683. checkPartChanged(*(PartNoType *) inputData);
  3684. CJoinGroup *jg = *(CJoinGroup **) (inputData + sizeof(PartNoType)); // NOTE - this is a pointer in Roxie server's address space - don't go following it!
  3685. char *inputRow = inputData+sizeof(PartNoType)+sizeof(const void *);
  3686. if (inputFields.isVariableSize())
  3687. {
  3688. inputSize = *(unsigned *)inputRow;
  3689. inputRow += sizeof(unsigned);
  3690. }
  3691. if (tlk)
  3692. {
  3693. createSegmentMonitors();
  3694. helper->createSegmentMonitors(tlk, inputRow);
  3695. if (rootIndex)
  3696. rootIndex->mergeSegmentMonitors(tlk);
  3697. tlk->finishSegmentMonitors();
  3698. if (logctx.queryTraceLevel() >= 20)
  3699. {
  3700. StringBuffer out;
  3701. printKeyedValues(out, tlk, helper->queryIndexRecordSize());
  3702. logctx.CTXLOG("Using filter %s", out.str());
  3703. }
  3704. if (!resent && (atmost != (unsigned) -1) && ((atmost > preabortKeyedJoinsThreshold) || (joinFlags & JFcountmatchabortlimit) || (keepLimit != 0)))
  3705. {
  3706. unsigned __int64 precount = tlk->checkCount(atmost);
  3707. if (precount > atmost)
  3708. {
  3709. candidateCount = atmost+1;
  3710. if (logctx.queryTraceLevel() > 5)
  3711. logctx.CTXLOG("Pre-aborting since candidate count is at least %" I64F "d", precount);
  3712. }
  3713. else
  3714. {
  3715. if (logctx.queryTraceLevel() > 10)
  3716. logctx.CTXLOG("NOT Pre-aborting since candidate count is %" I64F "d", precount);
  3717. tlk->reset(false);
  3718. }
  3719. }
  3720. else
  3721. tlk->reset(resent);
  3722. resent = false;
  3723. ScopedAtomic<unsigned> indexRecordsRead(::indexRecordsRead);
  3724. ScopedAtomic<unsigned> postFiltered(::postFiltered);
  3725. while (candidateCount <= atmost)
  3726. {
  3727. if (tlk->lookup(true))
  3728. {
  3729. candidateCount++;
  3730. indexRecordsRead++;
  3731. KLBlobProviderAdapter adapter(tlk);
  3732. const byte *indexRow = tlk->queryKeyBuffer();
  3733. size_t fposOffset = tlk->queryRowSize() - sizeof(offset_t);
  3734. offset_t fpos = rtlReadBigUInt8(indexRow + fposOffset);
  3735. if (helper->indexReadMatch(inputRow, indexRow, &adapter))
  3736. {
  3737. processed++;
  3738. if (keepLimit)
  3739. {
  3740. keepCount++;
  3741. if (keepCount > keepLimit)
  3742. break;
  3743. }
  3744. if (processed > rowLimit)
  3745. {
  3746. if (logctx.queryTraceLevel() > 1)
  3747. {
  3748. StringBuffer s;
  3749. logctx.CTXLOG("limit exceeded for %s", packet->queryHeader().toString(s).str());
  3750. }
  3751. noteStats(processed-processedBefore, rejected);
  3752. limitExceeded();
  3753. return NULL;
  3754. }
  3755. unsigned totalSize = 0;
  3756. if (helper->diskAccessRequired())
  3757. {
  3758. const void *self = output->getBuffer(KEYEDJOIN_RECORD_SIZE(0), true);
  3759. KeyedJoinHeader *rec = (KeyedJoinHeader *) self;
  3760. rec->fpos = fpos;
  3761. rec->thisGroup = jg;
  3762. rec->partNo = lastPartNo.partNo;
  3763. output->putBuffer(self, KEYEDJOIN_RECORD_SIZE(0), true);
  3764. }
  3765. else
  3766. {
  3767. KLBlobProviderAdapter adapter(tlk);
  3768. totalSize = helper->extractJoinFields(rowBuilder, indexRow, &adapter);
  3769. rowBuilder.writeToOutput(totalSize, fpos, jg, lastPartNo.partNo);
  3770. }
  3771. totalSizeSent += KEYEDJOIN_RECORD_SIZE(totalSize);
  3772. if (totalSizeSent > indexReadChunkSize && !continuationFailed)
  3773. {
  3774. MemoryBuffer si;
  3775. unsigned short siLen = 0;
  3776. si.append(siLen);
  3777. si.append(inputDone);
  3778. si.append(processed);
  3779. si.append(candidateCount);
  3780. si.append(keepCount);
  3781. si.append(lastPartNo.partNo);
  3782. si.append(lastPartNo.fileNo);
  3783. tlk->serializeCursorPos(si);
  3784. if (si.length() <= maxContinuationSize)
  3785. {
  3786. siLen = si.length() - sizeof(siLen);
  3787. si.writeDirect(0, sizeof(siLen), &siLen);
  3788. output->sendMetaInfo(si.toByteArray(), si.length());
  3789. noteStats(processed-processedBefore, rejected);
  3790. return output.getClear();
  3791. }
  3792. else
  3793. continuationFailed = true;
  3794. }
  3795. }
  3796. else
  3797. {
  3798. rejected++;
  3799. postFiltered++;
  3800. }
  3801. }
  3802. else
  3803. break;
  3804. }
  3805. tlk->releaseSegmentMonitors();
  3806. }
  3807. // output an end marker for the matches to this group
  3808. KeyedJoinHeader *rec = (KeyedJoinHeader *) output->getBuffer(KEYEDJOIN_RECORD_SIZE(0), true);
  3809. rec->fpos = candidateCount;
  3810. rec->thisGroup = jg;
  3811. rec->partNo = (unsigned short) -1;
  3812. output->putBuffer(rec, KEYEDJOIN_RECORD_SIZE(0), true);
  3813. totalSizeSent += KEYEDJOIN_RECORD_SIZE(0); // note - don't interrupt here though - too complicated.
  3814. candidateCount = 0;
  3815. keepCount = 0;
  3816. inputData = inputRow + inputSize;
  3817. inputDone += sizeof(PartNoType) + sizeof(const void *);
  3818. if (inputFields.isVariableSize())
  3819. inputDone += sizeof(unsigned);
  3820. inputDone += inputSize;
  3821. }
  3822. noteStats(processed-processedBefore, rejected);
  3823. if (aborted)
  3824. return NULL;
  3825. else
  3826. return output.getClear();
  3827. }
  3828. //================================================================================================
  3829. ISlaveActivityFactory *createRoxieKeyedJoinIndexActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  3830. {
  3831. return new CRoxieKeyedJoinIndexActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory);
  3832. }
  3833. //================================================================================================
  3834. class CRoxieKeyedJoinFetchActivityFactory : public CSlaveActivityFactory
  3835. {
  3836. public:
  3837. Owned<ITranslatorSet> translators;
  3838. Owned<IFileIOArray> files;
  3839. CRoxieKeyedJoinFetchActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  3840. : CSlaveActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory)
  3841. {
  3842. Owned<IHThorKeyedJoinArg> helper = (IHThorKeyedJoinArg *) helperFactory();
  3843. assertex(helper->diskAccessRequired());
  3844. bool variableFileName = allFilesDynamic || queryFactory.isDynamic() || ((helper->getFetchFlags() & (FFvarfilename|FFdynamicfilename)) != 0);
  3845. if (!variableFileName)
  3846. {
  3847. bool isOpt = (helper->getFetchFlags() & FFdatafileoptional) != 0;
  3848. OwnedRoxieString fileName(helper->getFileName());
  3849. datafile.setown(_queryFactory.queryPackage().lookupFileName(fileName, isOpt, true, true, _queryFactory.queryWorkUnit(), true));
  3850. if (datafile)
  3851. {
  3852. unsigned formatCrc = getFormatCrc(helper->getDiskFormatCrc());
  3853. translators.setown(datafile->getTranslators(formatCrc, helper->queryProjectedDiskRecordSize(), helper->queryDiskRecordSize(), getEnableFieldTranslation(), false));
  3854. files.setown(datafile->getIFileIOArray(isOpt, queryFactory.queryChannel()));
  3855. }
  3856. }
  3857. }
  3858. virtual IRoxieSlaveActivity *createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const;
  3859. virtual StringBuffer &toString(StringBuffer &s) const
  3860. {
  3861. return CSlaveActivityFactory::toString(s.append("KEYEDJOIN FETCH "));
  3862. }
  3863. };
  3864. class CRoxieKeyedJoinFetchActivity : public CRoxieSlaveActivity
  3865. {
  3866. IHThorKeyedJoinArg *helper;
  3867. Owned<IFileIO> rawFile;
  3868. offset_t base;
  3869. const char *inputLimit;
  3870. const char *inputData;
  3871. Linked<ITranslatorSet> translators;
  3872. Linked<IFileIOArray> files;
  3873. Owned<ISerialStream> rawStream;
  3874. CThorContiguousRowBuffer prefetchSource;
  3875. Owned<ISourceRowPrefetcher> prefetcher;
  3876. const IDynamicTransform *translator = nullptr;
  3877. virtual void setPartNo(bool filechanged)
  3878. {
  3879. rawFile.setown(files->getFilePart(lastPartNo.partNo, base)); // MORE - superfiles
  3880. translator = translators->queryTranslator(0); // MORE - superfiles
  3881. rawStream.setown(createFileSerialStream(rawFile, 0, -1, 0));
  3882. prefetcher.setown(translators->getPrefetcher(0));
  3883. prefetchSource.setStream(rawStream);
  3884. }
  3885. public:
  3886. CRoxieKeyedJoinFetchActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CRoxieKeyedJoinFetchActivityFactory *_aFactory,
  3887. IFileIOArray *_files, ITranslatorSet *_translators)
  3888. : files(_files),
  3889. translators(_translators),
  3890. CRoxieSlaveActivity(_logctx, _packet, _hFactory, _aFactory)
  3891. {
  3892. // MORE - no continuation row support?
  3893. base = 0;
  3894. helper = (IHThorKeyedJoinArg *) basehelper;
  3895. variableFileName = allFilesDynamic || basefactory->queryQueryFactory().isDynamic() || ((helper->getFetchFlags() & (FFvarfilename|FFdynamicfilename)) != 0);
  3896. onCreate();
  3897. inputData = (const char *) serializedCreate.readDirect(0);
  3898. inputLimit = inputData + (serializedCreate.length() - serializedCreate.getPos());
  3899. }
  3900. ~CRoxieKeyedJoinFetchActivity()
  3901. {
  3902. }
  3903. virtual const char *queryDynamicFileName() const
  3904. {
  3905. return helper->getFileName();
  3906. }
  3907. virtual void setVariableFileInfo()
  3908. {
  3909. unsigned formatCrc = basefactory->getFormatCrc(helper->getDiskFormatCrc());
  3910. translators.setown(varFileInfo->getTranslators(formatCrc, helper->queryProjectedDiskRecordSize(), helper->queryDiskRecordSize(), basefactory->getEnableFieldTranslation(), false));
  3911. files.setown(varFileInfo->getIFileIOArray(isOpt, packet->queryHeader().channel));
  3912. }
  3913. virtual IMessagePacker *process();
  3914. virtual StringBuffer &toString(StringBuffer &ret) const
  3915. {
  3916. return ret.appendf("KeyedJoinFetch %u", packet->queryHeader().activityId);
  3917. }
  3918. };
  3919. IMessagePacker *CRoxieKeyedJoinFetchActivity::process()
  3920. {
  3921. MTIME_SECTION(queryActiveTimer(), "CRoxieKeyedJoinFetchActivity::process");
  3922. // MORE - where we are returning everything there is an optimization or two to be had
  3923. Owned<IMessagePacker> output = ROQ->createOutputStream(packet->queryHeader(), false, logctx);
  3924. unsigned processed = 0;
  3925. unsigned skipped = 0;
  3926. unsigned __int64 rowLimit = helper->getRowLimit();
  3927. unsigned totalSizeSent = 0;
  3928. CachedOutputMetaData joinFieldsMeta(helper->queryJoinFieldsRecordSize());
  3929. Owned<IEngineRowAllocator> joinFieldsAllocator = getRowAllocator(joinFieldsMeta, basefactory->queryId());
  3930. OptimizedKJRowBuilder jfRowBuilder(joinFieldsAllocator, joinFieldsMeta, output);
  3931. CachedOutputMetaData inputFields(helper->queryFetchInputRecordSize());
  3932. size32_t inputSize = inputFields.getFixedSize();
  3933. while (!aborted && inputData < inputLimit)
  3934. {
  3935. checkPartChanged(*(PartNoType *) inputData);
  3936. inputData += sizeof(PartNoType);
  3937. offset_t rp;
  3938. memcpy(&rp, inputData, sizeof(rp));
  3939. offset_t pos;
  3940. if (isLocalFpos(rp))
  3941. pos = getLocalFposOffset(rp);
  3942. else
  3943. pos = rp-base;
  3944. prefetchSource.reset(pos);
  3945. prefetcher->readAhead(prefetchSource);
  3946. const byte *rawRHS = prefetchSource.queryRow();
  3947. MemoryBuffer buf;
  3948. if (translator)
  3949. {
  3950. MemoryBufferBuilder aBuilder(buf, 0);
  3951. translator->translate(aBuilder, rawRHS);
  3952. rawRHS = (const byte *) buf.toByteArray();
  3953. }
  3954. const KeyedJoinHeader *headerPtr = (KeyedJoinHeader *) inputData;
  3955. inputData = &headerPtr->rhsdata[0];
  3956. if (inputFields.isVariableSize())
  3957. {
  3958. memcpy(&inputSize, inputData, sizeof(inputSize));
  3959. inputData += sizeof(inputSize);
  3960. }
  3961. if (helper->fetchMatch(inputData, rawRHS))
  3962. {
  3963. unsigned thisSize = helper->extractJoinFields(jfRowBuilder, rawRHS, (IBlobProvider*)NULL);
  3964. jfRowBuilder.writeToOutput(thisSize, headerPtr->fpos, headerPtr->thisGroup, headerPtr->partNo);
  3965. totalSizeSent += KEYEDJOIN_RECORD_SIZE(thisSize);
  3966. processed++;
  3967. if (processed > rowLimit)
  3968. {
  3969. logctx.noteStatistic(StNumDiskSeeks, processed+skipped);
  3970. logctx.noteStatistic(StNumDiskRowsRead, processed+skipped);
  3971. logctx.noteStatistic(StNumDiskAccepted, processed);
  3972. logctx.noteStatistic(StNumDiskRejected, skipped);
  3973. limitExceeded();
  3974. return NULL;
  3975. }
  3976. }
  3977. else
  3978. {
  3979. skipped++;
  3980. KeyedJoinHeader *out = (KeyedJoinHeader *) output->getBuffer(KEYEDJOIN_RECORD_SIZE(0), true);
  3981. out->fpos = 0;
  3982. out->thisGroup = headerPtr->thisGroup;
  3983. out->partNo = (unsigned short) -1;
  3984. output->putBuffer(out, KEYEDJOIN_RECORD_SIZE(0), true);
  3985. totalSizeSent += KEYEDJOIN_RECORD_SIZE(0);
  3986. }
  3987. inputData += inputSize;
  3988. }
  3989. logctx.noteStatistic(StNumDiskSeeks, processed+skipped);
  3990. logctx.noteStatistic(StNumDiskRowsRead, processed+skipped);
  3991. logctx.noteStatistic(StNumDiskAccepted, processed);
  3992. logctx.noteStatistic(StNumDiskRejected, skipped);
  3993. if (aborted)
  3994. return NULL;
  3995. else
  3996. return output.getClear();
  3997. }
  3998. IRoxieSlaveActivity *CRoxieKeyedJoinFetchActivityFactory::createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const
  3999. {
  4000. return new CRoxieKeyedJoinFetchActivity(logctx, packet, helperFactory, this, files, translators);
  4001. }
  4002. ISlaveActivityFactory *createRoxieKeyedJoinFetchActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
  4003. {
  4004. return new CRoxieKeyedJoinFetchActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory);
  4005. }
  4006. //================================================================================================
  4007. class CRoxieRemoteActivity : public CRoxieSlaveActivity
  4008. {
  4009. protected:
  4010. IHThorRemoteArg * remoteHelper;
  4011. unsigned processed;
  4012. unsigned remoteId;
  4013. public:
  4014. CRoxieRemoteActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CSlaveActivityFactory *_aFactory, unsigned _remoteId)
  4015. : CRoxieSlaveActivity(_logctx, _packet, _hFactory, _aFactory),
  4016. remoteId(_remoteId)
  4017. {
  4018. remoteHelper = (IHThorRemoteArg *) basehelper;
  4019. processed = 0;
  4020. onCreate();
  4021. }
  4022. virtual StringBuffer &toString(StringBuffer &ret) const
  4023. {
  4024. return ret.appendf("Remote %u", packet->queryHeader().activityId);
  4025. }
  4026. virtual const char *queryDynamicFileName() const
  4027. {
  4028. throwUnexpected();
  4029. }
  4030. virtual void setVariableFileInfo()
  4031. {
  4032. throwUnexpected();
  4033. }
  4034. virtual IMessagePacker *process()
  4035. {
  4036. MTIME_SECTION(queryActiveTimer(), "CRoxieRemoteActivity ::process");
  4037. Owned<IMessagePacker> output = ROQ->createOutputStream(packet->queryHeader(), false, logctx);
  4038. unsigned __int64 rowLimit = remoteHelper->getRowLimit();
  4039. rtlRowBuilder remoteExtractBuilder;
  4040. remoteHelper->createParentExtract(remoteExtractBuilder);
  4041. Linked<IActivityGraph> remoteQuery = queryContext->queryChildGraph(remoteId);
  4042. Linked<IRoxieServerChildGraph> remoteGraph = remoteQuery->queryLoopGraph();
  4043. try
  4044. {
  4045. remoteGraph->beforeExecute();
  4046. Owned<IFinalRoxieInput> input = remoteGraph->startOutput(0, remoteExtractBuilder.size(), remoteExtractBuilder.getbytes(), false);
  4047. Owned<IStrandJunction> junction;
  4048. IEngineRowStream *stream = connectSingleStream(queryContext, input, 0, junction, 0);
  4049. while (!aborted)
  4050. {
  4051. const void * next = stream->ungroupedNextRow();
  4052. if (!next)
  4053. break;
  4054. size32_t nextSize = meta.getRecordSize(next);
  4055. //MORE - what about grouping?
  4056. processed++;
  4057. if (processed > rowLimit)
  4058. {
  4059. ReleaseRoxieRow(next);
  4060. limitExceeded();
  4061. break;
  4062. }
  4063. if (serializer)
  4064. serializeRow(output, next);
  4065. else
  4066. {
  4067. void * recBuffer = output->getBuffer(nextSize, meta.isVariableSize());
  4068. memcpy(recBuffer, next, nextSize);
  4069. output->putBuffer(recBuffer, nextSize, meta.isVariableSize());
  4070. }
  4071. ReleaseRoxieRow(next);
  4072. }
  4073. remoteGraph->afterExecute();
  4074. }
  4075. catch (IException *E)
  4076. {
  4077. remoteGraph->afterExecute();
  4078. if (aborted)
  4079. ::Release(E);
  4080. else
  4081. throw;
  4082. }
  4083. catch (...)
  4084. {
  4085. remoteGraph->afterExecute();
  4086. throw;
  4087. }
  4088. if (aborted)
  4089. return NULL;
  4090. else
  4091. return output.getClear();
  4092. }
  4093. virtual void setPartNo(bool filechanged)
  4094. {
  4095. throwUnexpected();
  4096. }
  4097. };
  4098. //================================================================================================
  4099. class CRoxieRemoteActivityFactory : public CSlaveActivityFactory
  4100. {
  4101. unsigned remoteId;
  4102. public:
  4103. CRoxieRemoteActivityFactory(IPropertyTree &graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory, unsigned _remoteId)
  4104. : CSlaveActivityFactory(graphNode, _subgraphId, _queryFactory, _helperFactory), remoteId(_remoteId)
  4105. {
  4106. }
  4107. virtual IRoxieSlaveActivity *createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const
  4108. {
  4109. return new CRoxieRemoteActivity(logctx, packet, helperFactory, this, remoteId);
  4110. }
  4111. virtual StringBuffer &toString(StringBuffer &s) const
  4112. {
  4113. return CSlaveActivityFactory::toString(s.append("Remote "));
  4114. }
  4115. };
  4116. ISlaveActivityFactory *createRoxieRemoteActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory, unsigned _remoteId)
  4117. {
  4118. return new CRoxieRemoteActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory, _remoteId);
  4119. }
  4120. //================================================================================================
  4121. class CRoxieDummyActivityFactory : public CSlaveActivityFactory // not a real activity - just used to properly link files
  4122. {
  4123. protected:
  4124. Owned<const IResolvedFile> indexfile;
  4125. Owned<IKeyArray> keyArray;
  4126. Owned<IFileIOArray> fileArray;
  4127. public:
  4128. CRoxieDummyActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, bool isLoadDataOnly)
  4129. : CSlaveActivityFactory(_graphNode, _subgraphId, _queryFactory, NULL)
  4130. {
  4131. if (_graphNode.getPropBool("att[@name='_isSpill']/@value", false) || _graphNode.getPropBool("att[@name='_isSpillGlobal']/@value", false))
  4132. return; // ignore 'spills'
  4133. try // operations does not want any missing file errors to be fatal, or throw traps - just log it
  4134. {
  4135. ThorActivityKind kind = getActivityKind(_graphNode);
  4136. if (kind != TAKdiskwrite && kind != TAKspillwrite && kind != TAKindexwrite && kind != TAKpiperead && kind != TAKpipewrite)
  4137. {
  4138. const char *fileName = queryNodeFileName(_graphNode, kind);
  4139. const char *indexName = queryNodeIndexName(_graphNode, kind);
  4140. if (indexName && !allFilesDynamic && !queryFactory.isDynamic())
  4141. {
  4142. bool isOpt = pretendAllOpt || _graphNode.getPropBool("att[@name='_isIndexOpt']/@value");
  4143. indexfile.setown(_queryFactory.queryPackage().lookupFileName(indexName, isOpt, true, true, _queryFactory.queryWorkUnit(), true));
  4144. if (indexfile)
  4145. keyArray.setown(indexfile->getKeyArray(isOpt, queryFactory.queryChannel()));
  4146. }
  4147. if (fileName && !allFilesDynamic && !queryFactory.isDynamic())
  4148. {
  4149. bool isOpt = pretendAllOpt || _graphNode.getPropBool("att[@name='_isOpt']/@value");
  4150. datafile.setown(_queryFactory.queryPackage().lookupFileName(fileName, isOpt, true, true, _queryFactory.queryWorkUnit(), true));
  4151. if (datafile)
  4152. {
  4153. fileArray.setown(datafile->getIFileIOArray(isOpt, queryFactory.queryChannel()));
  4154. }
  4155. }
  4156. }
  4157. }
  4158. catch(IException *E)
  4159. {
  4160. StringBuffer errors;
  4161. E->errorMessage(errors);
  4162. DBGLOG("%s File error = %s", (isLoadDataOnly) ? "LOADDATAONLY" : "SUSPENDED QUERY", errors.str());
  4163. E->Release();
  4164. }
  4165. }
  4166. virtual IRoxieSlaveActivity *createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const
  4167. {
  4168. throwUnexpected(); // don't actually want to create an activity
  4169. }
  4170. };
  4171. //================================================================================================
  4172. ISlaveActivityFactory *createRoxieDummyActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, bool isLoadDataOnly)
  4173. {
  4174. // MORE - bool isLoadDataOnly may need to be an enum if more than just LOADDATAONLY and suspended queries use this
  4175. return new CRoxieDummyActivityFactory(_graphNode, _subgraphId, _queryFactory, isLoadDataOnly);
  4176. }