comp_factory.cpp 223 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397239823992400240124022403240424052406240724082409241024112412241324142415241624172418241924202421242224232424242524262427242824292430243124322433243424352436243724382439244024412442244324442445244624472448244924502451245224532454245524562457245824592460246124622463246424652466246724682469247024712472247324742475247624772478247924802481248224832484248524862487248824892490249124922493249424952496249724982499250025012502250325042505250625072508250925102511251225132514251525162517251825192520252125222523252425252526252725282529253025312532253325342535253625372538253925402541254225432544254525462547254825492550255125522553255425552556255725582559256025612562256325642565256625672568256925702571257225732574257525762577257825792580258125822583258425852586258725882589259025912592259325942595259625972598259926002601260226032604260526062607260826092610261126122613261426152616261726182619262026212622262326242625262626272628262926302631263226332634263526362637263826392640264126422643264426452646264726482649265026512652265326542655265626572658265926602661266226632664266526662667266826692670267126722673267426752676267726782679268026812682268326842685268626872688268926902691269226932694269526962697269826992700270127022703270427052706270727082709271027112712271327142715271627172718271927202721272227232724272527262727272827292730273127322733273427352736273727382739274027412742274327442745274627472748274927502751275227532754275527562757275827592760276127622763276427652766276727682769277027712772277327742775277627772778277927802781278227832784278527862787278827892790279127922793279427952796279727982799280028012802280328042805280628072808280928102811281228132814281528162817281828192820282128222823282428252826282728282829283028312832283328342835283628372838283928402841284228432844284528462847284828492850285128522853285428552856285728582859286028612862286328642865286628672868286928702871287228732874287528762877287828792880288128822883288428852886288728882889289028912892289328942895289628972898289929002901290229032904290529062907290829092910291129122913291429152916291729182919292029212922292329242925292629272928292929302931293229332934293529362937293829392940294129422943294429452946294729482949295029512952295329542955295629572958295929602961296229632964296529662967296829692970297129722973297429752976297729782979298029812982298329842985298629872988298929902991299229932994299529962997299829993000300130023003300430053006300730083009301030113012301330143015301630173018301930203021302230233024302530263027302830293030303130323033303430353036303730383039304030413042304330443045304630473048304930503051305230533054305530563057305830593060306130623063306430653066306730683069307030713072307330743075307630773078307930803081308230833084308530863087308830893090309130923093309430953096309730983099310031013102310331043105310631073108310931103111311231133114311531163117311831193120312131223123312431253126312731283129313031313132313331343135313631373138313931403141314231433144314531463147314831493150315131523153315431553156315731583159316031613162316331643165316631673168316931703171317231733174317531763177317831793180318131823183318431853186318731883189319031913192319331943195319631973198319932003201320232033204320532063207320832093210321132123213321432153216321732183219322032213222322332243225322632273228322932303231323232333234323532363237323832393240324132423243324432453246324732483249325032513252325332543255325632573258325932603261326232633264326532663267326832693270327132723273327432753276327732783279328032813282328332843285328632873288328932903291329232933294329532963297329832993300330133023303330433053306330733083309331033113312331333143315331633173318331933203321332233233324332533263327332833293330333133323333333433353336333733383339334033413342334333443345334633473348334933503351335233533354335533563357335833593360336133623363336433653366336733683369337033713372337333743375337633773378337933803381338233833384338533863387338833893390339133923393339433953396339733983399340034013402340334043405340634073408340934103411341234133414341534163417341834193420342134223423342434253426342734283429343034313432343334343435343634373438343934403441344234433444344534463447344834493450345134523453345434553456345734583459346034613462346334643465346634673468346934703471347234733474347534763477347834793480348134823483348434853486348734883489349034913492349334943495349634973498349935003501350235033504350535063507350835093510351135123513351435153516351735183519352035213522352335243525352635273528352935303531353235333534353535363537353835393540354135423543354435453546354735483549355035513552355335543555355635573558355935603561356235633564356535663567356835693570357135723573357435753576357735783579358035813582358335843585358635873588358935903591359235933594359535963597359835993600360136023603360436053606360736083609361036113612361336143615361636173618361936203621362236233624362536263627362836293630363136323633363436353636363736383639364036413642364336443645364636473648364936503651365236533654365536563657365836593660366136623663366436653666366736683669367036713672367336743675367636773678367936803681368236833684368536863687368836893690369136923693369436953696369736983699370037013702370337043705370637073708370937103711371237133714371537163717371837193720372137223723372437253726372737283729373037313732373337343735373637373738373937403741374237433744374537463747374837493750375137523753375437553756375737583759376037613762376337643765376637673768376937703771377237733774377537763777377837793780378137823783378437853786378737883789379037913792379337943795379637973798379938003801380238033804380538063807380838093810381138123813381438153816381738183819382038213822382338243825382638273828382938303831383238333834383538363837383838393840384138423843384438453846384738483849385038513852385338543855385638573858385938603861386238633864386538663867386838693870387138723873387438753876387738783879388038813882388338843885388638873888388938903891389238933894389538963897389838993900390139023903390439053906390739083909391039113912391339143915391639173918391939203921392239233924392539263927392839293930393139323933393439353936393739383939394039413942394339443945394639473948394939503951395239533954395539563957395839593960396139623963396439653966396739683969397039713972397339743975397639773978397939803981398239833984398539863987398839893990399139923993399439953996399739983999400040014002400340044005400640074008400940104011401240134014401540164017401840194020402140224023402440254026402740284029403040314032403340344035403640374038403940404041404240434044404540464047404840494050405140524053405440554056405740584059406040614062406340644065406640674068406940704071407240734074407540764077407840794080408140824083408440854086408740884089409040914092409340944095409640974098409941004101410241034104410541064107410841094110411141124113411441154116411741184119412041214122412341244125412641274128412941304131413241334134413541364137413841394140414141424143414441454146414741484149415041514152415341544155415641574158415941604161416241634164416541664167416841694170417141724173417441754176417741784179418041814182418341844185418641874188418941904191419241934194419541964197419841994200420142024203420442054206420742084209421042114212421342144215421642174218421942204221422242234224422542264227422842294230423142324233423442354236423742384239424042414242424342444245424642474248424942504251425242534254425542564257425842594260426142624263426442654266426742684269427042714272427342744275427642774278427942804281428242834284428542864287428842894290429142924293429442954296429742984299430043014302430343044305430643074308430943104311431243134314431543164317431843194320432143224323432443254326432743284329433043314332433343344335433643374338433943404341434243434344434543464347434843494350435143524353435443554356435743584359436043614362436343644365436643674368436943704371437243734374437543764377437843794380438143824383438443854386438743884389439043914392439343944395439643974398439944004401440244034404440544064407440844094410441144124413441444154416441744184419442044214422442344244425442644274428442944304431443244334434443544364437443844394440444144424443444444454446444744484449445044514452445344544455445644574458445944604461446244634464446544664467446844694470447144724473447444754476447744784479448044814482448344844485448644874488448944904491449244934494449544964497449844994500450145024503450445054506450745084509451045114512451345144515451645174518451945204521452245234524452545264527452845294530453145324533453445354536453745384539454045414542454345444545454645474548454945504551455245534554455545564557455845594560456145624563456445654566456745684569457045714572457345744575457645774578457945804581458245834584458545864587458845894590459145924593459445954596459745984599460046014602460346044605460646074608460946104611461246134614461546164617461846194620462146224623462446254626462746284629463046314632463346344635463646374638463946404641464246434644464546464647464846494650465146524653465446554656465746584659466046614662466346644665466646674668466946704671467246734674467546764677467846794680468146824683468446854686468746884689469046914692469346944695469646974698469947004701470247034704470547064707470847094710471147124713471447154716471747184719472047214722472347244725472647274728472947304731473247334734473547364737473847394740474147424743474447454746474747484749475047514752475347544755475647574758475947604761476247634764476547664767476847694770477147724773477447754776477747784779478047814782478347844785478647874788478947904791479247934794479547964797479847994800480148024803480448054806480748084809481048114812481348144815481648174818481948204821482248234824482548264827482848294830483148324833483448354836483748384839484048414842484348444845484648474848484948504851485248534854485548564857485848594860486148624863486448654866486748684869487048714872487348744875487648774878487948804881488248834884488548864887488848894890489148924893489448954896489748984899490049014902490349044905490649074908490949104911491249134914491549164917491849194920492149224923492449254926492749284929493049314932493349344935493649374938493949404941494249434944494549464947494849494950495149524953495449554956495749584959496049614962496349644965496649674968496949704971497249734974497549764977497849794980498149824983498449854986498749884989499049914992499349944995499649974998499950005001500250035004500550065007500850095010501150125013501450155016501750185019502050215022502350245025502650275028502950305031503250335034503550365037503850395040504150425043504450455046504750485049505050515052505350545055505650575058505950605061506250635064506550665067506850695070507150725073507450755076507750785079508050815082508350845085508650875088508950905091509250935094509550965097509850995100510151025103510451055106510751085109511051115112511351145115511651175118511951205121512251235124512551265127512851295130513151325133513451355136513751385139514051415142514351445145514651475148514951505151515251535154515551565157515851595160516151625163516451655166516751685169517051715172517351745175517651775178517951805181518251835184518551865187518851895190519151925193519451955196519751985199520052015202520352045205520652075208520952105211521252135214521552165217521852195220522152225223522452255226522752285229523052315232523352345235523652375238523952405241524252435244524552465247524852495250525152525253525452555256525752585259526052615262526352645265526652675268526952705271527252735274527552765277527852795280528152825283528452855286528752885289529052915292529352945295529652975298529953005301530253035304530553065307530853095310531153125313531453155316531753185319532053215322532353245325532653275328532953305331533253335334533553365337533853395340534153425343534453455346534753485349535053515352535353545355535653575358535953605361536253635364536553665367536853695370537153725373537453755376537753785379538053815382538353845385538653875388538953905391539253935394539553965397539853995400540154025403540454055406540754085409541054115412541354145415541654175418541954205421542254235424542554265427542854295430543154325433543454355436543754385439544054415442544354445445544654475448544954505451545254535454545554565457545854595460546154625463546454655466546754685469547054715472547354745475547654775478547954805481548254835484548554865487548854895490549154925493549454955496549754985499550055015502550355045505550655075508550955105511551255135514551555165517551855195520552155225523552455255526552755285529553055315532553355345535553655375538553955405541554255435544554555465547554855495550555155525553555455555556555755585559556055615562556355645565556655675568556955705571557255735574557555765577557855795580558155825583558455855586558755885589559055915592559355945595559655975598559956005601560256035604560556065607560856095610561156125613561456155616561756185619562056215622562356245625562656275628562956305631563256335634563556365637563856395640564156425643564456455646564756485649565056515652565356545655565656575658565956605661566256635664566556665667566856695670567156725673567456755676567756785679568056815682568356845685568656875688568956905691569256935694569556965697569856995700570157025703570457055706570757085709571057115712571357145715571657175718571957205721572257235724572557265727572857295730573157325733573457355736573757385739574057415742574357445745574657475748574957505751575257535754575557565757575857595760576157625763576457655766576757685769577057715772577357745775577657775778577957805781578257835784578557865787578857895790579157925793579457955796579757985799580058015802580358045805580658075808580958105811581258135814581558165817581858195820582158225823582458255826582758285829583058315832583358345835583658375838583958405841584258435844584558465847584858495850585158525853585458555856585758585859586058615862586358645865586658675868586958705871587258735874587558765877587858795880588158825883588458855886588758885889589058915892589358945895589658975898
  1. #include "pg_compat.h"
  2. #define TypeName PG_TypeName
  3. #define SortBy PG_SortBy
  4. #define Sort PG_Sort
  5. #define Unique PG_Unique
  6. #undef SIZEOF_SIZE_T
  7. extern "C" {
  8. #include "postgres.h"
  9. #include "access/xact.h"
  10. #include "catalog/pg_am_d.h"
  11. #include "catalog/pg_collation_d.h"
  12. #include "catalog/pg_conversion_d.h"
  13. #include "catalog/pg_database_d.h"
  14. #include "catalog/pg_operator_d.h"
  15. #include "catalog/pg_proc_d.h"
  16. #include "catalog/pg_namespace_d.h"
  17. #include "catalog/pg_tablespace_d.h"
  18. #include "catalog/pg_type_d.h"
  19. #include "datatype/timestamp.h"
  20. #include "utils/builtins.h"
  21. #include "utils/memutils.h"
  22. #include "utils/array.h"
  23. #include "utils/arrayaccess.h"
  24. #include "utils/datum.h"
  25. #include "utils/lsyscache.h"
  26. #include "utils/datetime.h"
  27. #include "utils/numeric.h"
  28. #include "utils/typcache.h"
  29. #include "utils/memutils_internal.h"
  30. #include "mb/pg_wchar.h"
  31. #include "nodes/execnodes.h"
  32. #include "executor/executor.h"
  33. #include "lib/stringinfo.h"
  34. #include "miscadmin.h"
  35. #include "funcapi.h"
  36. #include "thread_inits.h"
  37. #undef Abs
  38. #undef Min
  39. #undef Max
  40. #undef TypeName
  41. #undef SortBy
  42. #undef Sort
  43. #undef Unique
  44. #undef LOG
  45. #undef INFO
  46. #undef NOTICE
  47. #undef WARNING
  48. //#undef ERROR
  49. #undef FATAL
  50. #undef PANIC
  51. #undef open
  52. #undef fopen
  53. #undef bind
  54. #undef locale_t
  55. constexpr auto PG_DAY = DAY;
  56. constexpr auto PG_SECOND = SECOND;
  57. constexpr auto PG_ERROR = ERROR;
  58. #undef DAY
  59. #undef SECOND
  60. #undef ERROR
  61. }
  62. #include <yql/essentials/core/pg_settings/guc_settings.h>
  63. #include <yql/essentials/parser/pg_wrapper/interface/interface.h>
  64. #include <yql/essentials/parser/pg_wrapper/memory_context.h>
  65. #include <yql/essentials/parser/pg_wrapper/pg_catalog_consts.h>
  66. #include <yql/essentials/minikql/computation/mkql_block_impl.h>
  67. #include <yql/essentials/minikql/computation/mkql_computation_node_impl.h>
  68. #include <yql/essentials/minikql/computation/mkql_computation_node_holders.h>
  69. #include <yql/essentials/minikql/computation/mkql_computation_node_pack_impl.h>
  70. #include <yql/essentials/minikql/computation/mkql_custom_list.h>
  71. #include <yql/essentials/minikql/computation/presort_impl.h>
  72. #include <yql/essentials/minikql/mkql_node_cast.h>
  73. #include <yql/essentials/minikql/mkql_alloc.h>
  74. #include <yql/essentials/minikql/mkql_buffer.h>
  75. #include <yql/essentials/minikql/mkql_node_builder.h>
  76. #include <yql/essentials/minikql/mkql_string_util.h>
  77. #include <yql/essentials/minikql/mkql_type_builder.h>
  78. #include <yql/essentials/types/binary_json/read.h>
  79. #include <yql/essentials/types/uuid/uuid.h>
  80. #include <yql/essentials/public/udf/arrow/block_reader.h>
  81. #include <yql/essentials/public/udf/arrow/block_builder.cpp>
  82. #include <yql/essentials/parser/pg_catalog/catalog.h>
  83. #include <yql/essentials/providers/common/codec/yql_codec_buf.h>
  84. #include <yql/essentials/public/result_format/yql_codec_results.h>
  85. #include <yql/essentials/public/udf/udf_value_builder.h>
  86. #include <yql/essentials/utils/fp_bits.h>
  87. #include <library/cpp/yson/detail.h>
  88. #include <util/string/split.h>
  89. #include <util/system/getpid.h>
  90. #include "arrow.h"
  91. #include "arrow_impl.h"
  92. #define DAY PG_DAY
  93. #define SECOND PG_SECOND
  94. #define ERROR PG_ERROR
  95. extern "C" {
  96. extern void *MkqlAlloc(MemoryContext context, Size size);
  97. extern void MkqlFree(void *pointer);
  98. extern void *MkqlRealloc(void *pointer, Size size);
  99. extern void MkqlReset(MemoryContext context);
  100. extern void MkqlDelete(MemoryContext context);
  101. extern MemoryContext MkqlGetChunkContext(void *pointer);
  102. extern Size MkqlGetChunkSpace(void *pointer);
  103. extern bool MkqlIsEmpty(MemoryContext context);
  104. extern void MkqlStats(MemoryContext context,
  105. MemoryStatsPrintFunc printfunc, void *passthru,
  106. MemoryContextCounters *totals,
  107. bool print_to_stderr);
  108. #ifdef MEMORY_CONTEXT_CHECKING
  109. extern void MkqlCheck(MemoryContext context);
  110. #endif
  111. }
  112. namespace NYql {
  113. using namespace NKikimr::NMiniKQL;
  114. TVPtrHolder TVPtrHolder::Instance;
  115. // use 'false' for native format
  116. static __thread bool NeedCanonizeFp = false;
  117. NUdf::TUnboxedValue CreatePgString(i32 typeLen, ui32 targetTypeId, TStringBuf data) {
  118. // typname => 'cstring', typlen => '-2'
  119. // typname = > 'text', typlen => '-1'
  120. // typname => 'name', typlen => NAMEDATALEN
  121. Y_UNUSED(targetTypeId); // todo: verify typeLen
  122. switch (typeLen) {
  123. case -1:
  124. return PointerDatumToPod((Datum)MakeVar(data));
  125. case -2:
  126. return PointerDatumToPod((Datum)MakeCString(data));
  127. default:
  128. return PointerDatumToPod((Datum)MakeFixedString(data, typeLen));
  129. }
  130. }
  131. extern "C" void *MkqlAlloc(MemoryContext context, Size size) {
  132. Y_UNUSED(context);
  133. auto fullSize = size + sizeof(TMkqlPAllocHeader);
  134. auto header = (TMkqlPAllocHeader*)MKQLAllocWithSize(fullSize, EMemorySubPool::Default);
  135. header->Size = size;
  136. header->U.Entry.Link(TlsAllocState->CurrentPAllocList);
  137. Y_ENSURE((ui64(context) & MEMORY_CONTEXT_METHODID_MASK) == 0);
  138. header->Self = ui64(context) | MCTX_UNUSED3_ID;
  139. return header + 1;
  140. }
  141. extern "C" void MkqlFree(void* pointer) {
  142. if (pointer) {
  143. auto header = ((TMkqlPAllocHeader*)pointer) - 1;
  144. // remove this block from list
  145. header->U.Entry.Unlink();
  146. auto fullSize = header->Size + sizeof(TMkqlPAllocHeader);
  147. MKQLFreeWithSize(header, fullSize, EMemorySubPool::Default);
  148. }
  149. }
  150. extern "C" void* MkqlRealloc(void* pointer, Size size) {
  151. if (!size) {
  152. MkqlFree(pointer);
  153. return nullptr;
  154. }
  155. auto ret = MkqlAlloc(nullptr, size);
  156. if (pointer) {
  157. auto header = ((TMkqlPAllocHeader*)pointer) - 1;
  158. memmove(ret, pointer, header->Size);
  159. MkqlFree(pointer);
  160. }
  161. return ret;
  162. }
  163. extern "C" void MkqlReset(MemoryContext context) {
  164. Y_UNUSED(context);
  165. }
  166. extern "C" void MkqlDelete(MemoryContext context) {
  167. Y_UNUSED(context);
  168. }
  169. extern "C" MemoryContext MkqlGetChunkContext(void *pointer) {
  170. return (MemoryContext)(((ui64*)pointer)[-1] & ~MEMORY_CONTEXT_METHODID_MASK);
  171. }
  172. extern "C" Size MkqlGetChunkSpace(void* pointer) {
  173. Y_UNUSED(pointer);
  174. return 0;
  175. }
  176. extern "C" bool MkqlIsEmpty(MemoryContext context) {
  177. Y_UNUSED(context);
  178. return false;
  179. }
  180. extern "C" void MkqlStats(MemoryContext context,
  181. MemoryStatsPrintFunc printfunc, void *passthru,
  182. MemoryContextCounters *totals,
  183. bool print_to_stderr) {
  184. Y_UNUSED(context);
  185. Y_UNUSED(printfunc);
  186. Y_UNUSED(passthru);
  187. Y_UNUSED(totals);
  188. Y_UNUSED(print_to_stderr);
  189. }
  190. extern "C" void MkqlCheck(MemoryContext context) {
  191. Y_UNUSED(context);
  192. }
  193. Datum MakeArrayOfText(const TVector<TString>& arr) {
  194. TVector<Datum> elems(arr.size());
  195. for (size_t i = 0; i < elems.size(); ++i) {
  196. elems[i] = (Datum)MakeVar(arr[i]);
  197. }
  198. auto ret = construct_array(elems.data(), (int)arr.size(), TEXTOID, -1, false, 'i');
  199. for (size_t i = 0; i < elems.size(); ++i) {
  200. pfree((void*)elems[i]);
  201. }
  202. return (Datum)ret;
  203. }
  204. class TPgConst : public TMutableComputationNode<TPgConst> {
  205. typedef TMutableComputationNode<TPgConst> TBaseComputation;
  206. public:
  207. TPgConst(TComputationMutables& mutables, ui32 typeId, const std::string_view& value, IComputationNode* typeMod)
  208. : TBaseComputation(mutables)
  209. , TypeId(typeId)
  210. , Value(value)
  211. , TypeMod(typeMod)
  212. , TypeDesc(NPg::LookupType(TypeId))
  213. {
  214. Zero(FInfo);
  215. ui32 inFuncId = TypeDesc.InFuncId;
  216. if (TypeDesc.TypeId == TypeDesc.ArrayTypeId) {
  217. inFuncId = NPg::LookupProc("array_in", { 0,0,0 }).ProcId;
  218. }
  219. Y_ENSURE(inFuncId);
  220. GetPgFuncAddr(inFuncId, FInfo);
  221. Y_ENSURE(!FInfo.fn_retset);
  222. Y_ENSURE(FInfo.fn_addr);
  223. Y_ENSURE(FInfo.fn_nargs >=1 && FInfo.fn_nargs <= 3);
  224. TypeIOParam = MakeTypeIOParam(TypeDesc);
  225. }
  226. NUdf::TUnboxedValuePod DoCalculate(TComputationContext& compCtx) const {
  227. i32 typeMod = -1;
  228. if (TypeMod) {
  229. typeMod = DatumGetInt32(ScalarDatumFromPod(TypeMod->GetValue(compCtx)));
  230. }
  231. LOCAL_FCINFO(callInfo, 3);
  232. Zero(*callInfo);
  233. FmgrInfo copyFmgrInfo = FInfo;
  234. callInfo->flinfo = &copyFmgrInfo;
  235. callInfo->nargs = 3;
  236. callInfo->fncollation = DEFAULT_COLLATION_OID;
  237. callInfo->isnull = false;
  238. callInfo->args[0] = { (Datum)Value.c_str(), false };
  239. callInfo->args[1] = { ObjectIdGetDatum(TypeIOParam), false };
  240. callInfo->args[2] = { Int32GetDatum(typeMod), false };
  241. TPAllocScope call;
  242. auto ret = FInfo.fn_addr(callInfo);
  243. Y_ENSURE(!callInfo->isnull);
  244. return AnyDatumToPod(ret, TypeDesc.PassByValue);
  245. }
  246. private:
  247. void RegisterDependencies() const final {
  248. if (TypeMod) {
  249. DependsOn(TypeMod);
  250. }
  251. }
  252. const ui32 TypeId;
  253. const TString Value;
  254. IComputationNode* const TypeMod;
  255. const NPg::TTypeDesc TypeDesc;
  256. FmgrInfo FInfo;
  257. ui32 TypeIOParam;
  258. };
  259. class TPgInternal0 : public TMutableComputationNode<TPgInternal0> {
  260. typedef TMutableComputationNode<TPgInternal0> TBaseComputation;
  261. public:
  262. TPgInternal0(TComputationMutables& mutables)
  263. : TBaseComputation(mutables)
  264. {
  265. }
  266. NUdf::TUnboxedValuePod DoCalculate(TComputationContext& compCtx) const {
  267. return ScalarDatumToPod(PointerGetDatum(nullptr));
  268. }
  269. private:
  270. void RegisterDependencies() const final {
  271. }
  272. };
  273. class TPgTableContent : public TMutableComputationNode<TPgTableContent> {
  274. typedef TMutableComputationNode<TPgTableContent> TBaseComputation;
  275. private:
  276. static NUdf::TUnboxedValuePod MakePgDatabaseDatnameColumn(ui32 index) {
  277. std::string content;
  278. switch (index) {
  279. case 1: {
  280. content = "template1";
  281. break;
  282. }
  283. case 2: {
  284. content = "template0";
  285. break;
  286. }
  287. case PG_POSTGRES_DATABASE_ID: {
  288. content = "postgres";
  289. break;
  290. }
  291. case PG_CURRENT_DATABASE_ID: {
  292. Y_ENSURE(PGGetGUCSetting("ydb_database"));
  293. content = *PGGetGUCSetting("ydb_database");
  294. break;
  295. }
  296. }
  297. return PointerDatumToPod((Datum)(MakeFixedString(content, NAMEDATALEN)));
  298. }
  299. public:
  300. TPgTableContent(
  301. TComputationMutables& mutables,
  302. const std::string_view& cluster,
  303. const std::string_view& table,
  304. TType* returnType)
  305. : TBaseComputation(mutables)
  306. , Cluster_(cluster)
  307. , Table_(table)
  308. , ItemType_(AS_TYPE(TStructType, AS_TYPE(TListType, returnType)->GetItemType()))
  309. {
  310. YQL_ENSURE(Cluster_ == "pg_catalog" || Cluster_ == "information_schema");
  311. if (Cluster_ == "pg_catalog") {
  312. if (Table_ == "pg_type") {
  313. static const std::pair<const char*, TPgTypeFiller> AllPgTypeFillers[] = {
  314. {"oid", [](const NPg::TTypeDesc& desc) { return ScalarDatumToPod(ObjectIdGetDatum(desc.TypeId)); }},
  315. {"typname", [](const NPg::TTypeDesc& desc) { return PointerDatumToPod((Datum)(MakeFixedString(desc.Name, NAMEDATALEN))); }},
  316. {"typinput", [](const NPg::TTypeDesc& desc) { return ScalarDatumToPod(ObjectIdGetDatum(desc.InFuncId)); }},
  317. {"typnamespace", [](const NPg::TTypeDesc& desc) { return ScalarDatumToPod(ObjectIdGetDatum(PG_CATALOG_NAMESPACE)); }},
  318. {"typtype", [](const NPg::TTypeDesc& desc) { return ScalarDatumToPod(CharGetDatum((char)desc.TypType)); }},
  319. {"typrelid", [](const NPg::TTypeDesc&) { return ScalarDatumToPod(ObjectIdGetDatum(0)); }},
  320. {"typelem", [](const NPg::TTypeDesc& desc) { return ScalarDatumToPod(ObjectIdGetDatum(desc.ElementTypeId)); }},
  321. };
  322. ApplyFillers(AllPgTypeFillers, Y_ARRAY_SIZE(AllPgTypeFillers), PgTypeFillers_);
  323. } else if (Table_ == "pg_database") {
  324. static const std::pair<const char*, TPgDatabaseFiller> AllPgDatabaseFillers[] = {
  325. {"oid", [](ui32 index) { return ScalarDatumToPod(ObjectIdGetDatum(index)); }},
  326. {"datdba", [](ui32) { return ScalarDatumToPod(ObjectIdGetDatum(1)); }},
  327. {"datistemplate", [](ui32 index) { return ScalarDatumToPod(BoolGetDatum(index < PG_POSTGRES_DATABASE_ID)); }},
  328. {"datallowconn", [](ui32 index) { return ScalarDatumToPod(BoolGetDatum(index != 2)); }},
  329. {"datname", MakePgDatabaseDatnameColumn},
  330. {"encoding", [](ui32) { return ScalarDatumToPod(Int32GetDatum(PG_UTF8)); }},
  331. {"datcollate", [](ui32) { return PointerDatumToPod((Datum)(MakeFixedString("C", NAMEDATALEN))); }},
  332. {"datctype", [](ui32) { return PointerDatumToPod((Datum)(MakeFixedString("C", NAMEDATALEN))); }},
  333. };
  334. ApplyFillers(AllPgDatabaseFillers, Y_ARRAY_SIZE(AllPgDatabaseFillers), PgDatabaseFillers_);
  335. } else if (Table_ == "pg_tablespace") {
  336. static const std::pair<const char*, TPgTablespaceFiller> AllPgTablespaceFillers[] = {
  337. {"oid", [](ui32 index) { return ScalarDatumToPod(ObjectIdGetDatum(index == 1 ? DEFAULTTABLESPACE_OID : GLOBALTABLESPACE_OID)); }},
  338. {"spcname", [](ui32 index) { return PointerDatumToPod((Datum)(MakeFixedString(index == 1 ? "pg_default" : "pg_global", NAMEDATALEN))); }},
  339. };
  340. ApplyFillers(AllPgTablespaceFillers, Y_ARRAY_SIZE(AllPgTablespaceFillers), PgTablespaceFillers_);
  341. } else if (Table_ == "pg_shdescription") {
  342. static const std::pair<const char*, TPgShDescriptionFiller> AllPgShDescriptionFillers[] = {
  343. {"objoid", [](ui32 index) { return ScalarDatumToPod(ObjectIdGetDatum(index)); }},
  344. {"classoid", [](ui32) { return ScalarDatumToPod(ObjectIdGetDatum(DatabaseRelationId)); }},
  345. {"description", [](ui32 index) { return PointerDatumToPod((Datum)MakeVar(
  346. index == 1 ? "default template for new databases" :
  347. (index == 2 ? "unmodifiable empty database" :
  348. "default administrative connection database")
  349. )); }},
  350. };
  351. ApplyFillers(AllPgShDescriptionFillers, Y_ARRAY_SIZE(AllPgShDescriptionFillers), PgShDescriptionFillers_);
  352. } else if (Table_ == "pg_stat_gssapi") {
  353. static const std::pair<const char*, TPgStatGssapiFiller> AllPgStatGssapiFillers[] = {
  354. {"encrypted", []() { return ScalarDatumToPod(BoolGetDatum(false)); }},
  355. {"gss_authenticated", []() { return ScalarDatumToPod(BoolGetDatum(false)); }},
  356. {"pid", []() { return ScalarDatumToPod(Int32GetDatum(GetPID())); }}
  357. };
  358. ApplyFillers(AllPgStatGssapiFillers, Y_ARRAY_SIZE(AllPgStatGssapiFillers), PgStatGssapiFillers_);
  359. } else if (Table_ == "pg_namespace") {
  360. static const std::pair<const char*, TPgNamespaceFiller> AllPgNamespaceFillers[] = {
  361. {"nspname", [](const NPg::TNamespaceDesc& desc) {return PointerDatumToPod((Datum)MakeFixedString(desc.Name, NAMEDATALEN));}},
  362. {"oid", [](const NPg::TNamespaceDesc& desc) { return ScalarDatumToPod(ObjectIdGetDatum(desc.Oid)); }},
  363. {"nspowner", [](const NPg::TNamespaceDesc&) { return ScalarDatumToPod(ObjectIdGetDatum(1)); }},
  364. };
  365. ApplyFillers(AllPgNamespaceFillers, Y_ARRAY_SIZE(AllPgNamespaceFillers), PgNamespaceFillers_);
  366. } else if (Table_ == "pg_am") {
  367. static const std::pair<const char*, TPgAmFiller> AllPgAmFillers[] = {
  368. {"oid", [](const NPg::TAmDesc& desc) { return ScalarDatumToPod(ObjectIdGetDatum(desc.Oid)); }},
  369. {"amname", [](const NPg::TAmDesc& desc) { return PointerDatumToPod((Datum)MakeFixedString(desc.AmName, NAMEDATALEN)); }},
  370. {"amtype", [](const NPg::TAmDesc& desc) { return ScalarDatumToPod(CharGetDatum((char)desc.AmType)); }},
  371. };
  372. ApplyFillers(AllPgAmFillers, Y_ARRAY_SIZE(AllPgAmFillers), PgAmFillers_);
  373. } else if (Table_ == "pg_description") {
  374. static const std::pair<const char*, TPgDescriptionFiller> AllPgDescriptionFillers[] = {
  375. {"objoid", [](const TDescriptionDesc& desc) { return ScalarDatumToPod(ObjectIdGetDatum(desc.Objoid)); }},
  376. {"classoid", [](const TDescriptionDesc& desc) { return ScalarDatumToPod(ObjectIdGetDatum(desc.Classoid)); }},
  377. {"objsubid", [](const TDescriptionDesc& desc) { return ScalarDatumToPod(Int32GetDatum(desc.Objsubid)); }},
  378. {"description", [](const TDescriptionDesc& desc) { return PointerDatumToPod((Datum)MakeVar(desc.Description)); }}
  379. };
  380. ApplyFillers(AllPgDescriptionFillers, Y_ARRAY_SIZE(AllPgDescriptionFillers), PgDescriptionFillers_);
  381. } else if (Table_ == "pg_tables") {
  382. static const std::pair<const char*, TTablesFiller> AllPgTablesFillers[] = {
  383. {"schemaname", [](const NPg::TTableInfo& desc) { return PointerDatumToPod((Datum)MakeFixedString(desc.Schema, NAMEDATALEN)); }},
  384. {"tablename", [](const NPg::TTableInfo& desc) { return PointerDatumToPod((Datum)MakeFixedString(desc.Name, NAMEDATALEN)); }},
  385. };
  386. ApplyFillers(AllPgTablesFillers, Y_ARRAY_SIZE(AllPgTablesFillers), PgTablesFillers_);
  387. } else if (Table_ == "pg_roles") {
  388. static const std::pair<const char*, TPgRolesFiller> AllPgRolesFillers[] = {
  389. {"rolname", [](ui32 index) {
  390. return PointerDatumToPod((Datum)MakeFixedString(index == 1 ? "postgres" : *PGGetGUCSetting("ydb_user"), NAMEDATALEN));
  391. }},
  392. {"oid", [](ui32 index) { return ScalarDatumToPod(ObjectIdGetDatum(index)); }},
  393. {"rolbypassrls", [](ui32) { return ScalarDatumToPod(BoolGetDatum(true)); }},
  394. {"rolsuper", [](ui32) { return ScalarDatumToPod(BoolGetDatum(true)); }},
  395. {"rolinherit", [](ui32) { return ScalarDatumToPod(BoolGetDatum(true)); }},
  396. {"rolcreaterole", [](ui32) { return ScalarDatumToPod(BoolGetDatum(true)); }},
  397. {"rolcreatedb", [](ui32) { return ScalarDatumToPod(BoolGetDatum(true)); }},
  398. {"rolcanlogin", [](ui32) { return ScalarDatumToPod(BoolGetDatum(true)); }},
  399. {"rolreplication", [](ui32) { return ScalarDatumToPod(BoolGetDatum(true)); }},
  400. {"rolconnlimit", [](ui32) { return ScalarDatumToPod(Int32GetDatum(-1)); }},
  401. {"rolvaliduntil", [](ui32) { return NUdf::TUnboxedValuePod(); }},
  402. {"rolconfig", [](ui32) { return PointerDatumToPod(MakeArrayOfText({
  403. "search_path=public",
  404. "default_transaction_isolation=serializable",
  405. "standard_conforming_strings=on",
  406. })); }},
  407. };
  408. ApplyFillers(AllPgRolesFillers, Y_ARRAY_SIZE(AllPgRolesFillers), PgRolesFillers_);
  409. } else if (Table_ == "pg_user") {
  410. static const std::pair<const char*, TPgUserFiller> AllPgUserFillers[] = {
  411. {"usename", [](ui32 index) {
  412. return PointerDatumToPod((Datum)MakeFixedString(index == 1 ? "postgres" : *PGGetGUCSetting("ydb_user"), NAMEDATALEN));
  413. }},
  414. {"usesysid", [](ui32 index) { return ScalarDatumToPod(ObjectIdGetDatum(index)); }},
  415. {"usecreatedb", [](ui32) { return ScalarDatumToPod(BoolGetDatum(true)); }},
  416. {"usesuper", [](ui32) { return ScalarDatumToPod(BoolGetDatum(true)); }},
  417. {"userepl", [](ui32) { return ScalarDatumToPod(BoolGetDatum(true)); }},
  418. {"usebypassrls", [](ui32) { return ScalarDatumToPod(BoolGetDatum(true)); }},
  419. {"passwd", [](ui32) { return NUdf::TUnboxedValuePod(); }},
  420. {"valuntil", [](ui32) { return NUdf::TUnboxedValuePod(); }},
  421. {"useconfig", [](ui32) { return PointerDatumToPod(MakeArrayOfText({
  422. "search_path=public",
  423. "default_transaction_isolation=serializable",
  424. "standard_conforming_strings=on",
  425. })); }},
  426. };
  427. ApplyFillers(AllPgUserFillers, Y_ARRAY_SIZE(AllPgUserFillers), PgUserFillers_);
  428. } else if (Table_ == "pg_stat_database") {
  429. static const std::pair<const char*, TPgDatabaseStatFiller> AllPgDatabaseStatFillers[] = {
  430. {"datid", [](ui32 index) { return ScalarDatumToPod(ObjectIdGetDatum(index ? 3 : 0)); }},
  431. {"blks_hit", [](ui32) { return ScalarDatumToPod(Int64GetDatum(0)); }},
  432. {"blks_read", [](ui32) { return ScalarDatumToPod(Int64GetDatum(0)); }},
  433. {"tup_deleted", [](ui32) { return ScalarDatumToPod(Int64GetDatum(0)); }},
  434. {"tup_fetched", [](ui32) { return ScalarDatumToPod(Int64GetDatum(0)); }},
  435. {"tup_inserted", [](ui32) { return ScalarDatumToPod(Int64GetDatum(0)); }},
  436. {"tup_returned", [](ui32) { return ScalarDatumToPod(Int64GetDatum(0)); }},
  437. {"tup_updated", [](ui32) { return ScalarDatumToPod(Int64GetDatum(0)); }},
  438. {"xact_commit", [](ui32) { return ScalarDatumToPod(Int64GetDatum(0)); }},
  439. {"xact_rollback", [](ui32) { return ScalarDatumToPod(Int64GetDatum(0)); }},
  440. };
  441. ApplyFillers(AllPgDatabaseStatFillers, Y_ARRAY_SIZE(AllPgDatabaseStatFillers), PgDatabaseStatFillers_);
  442. } else if (Table_ == "pg_class") {
  443. static const std::pair<const char*, TPgClassFiller> AllPgClassFillers[] = {
  444. {"oid", [](const NPg::TTableInfo& desc, ui32, ui32) { return ScalarDatumToPod(ObjectIdGetDatum(desc.Oid)); }},
  445. {"relispartition", [](const NPg::TTableInfo&, ui32, ui32) { return ScalarDatumToPod(BoolGetDatum(false)); }},
  446. {"relkind", [](const NPg::TTableInfo& desc, ui32, ui32) { return ScalarDatumToPod(CharGetDatum((char)desc.Kind)); }},
  447. {"relname", [](const NPg::TTableInfo& desc, ui32, ui32) { return PointerDatumToPod((Datum)MakeFixedString(desc.Name, NAMEDATALEN)); }},
  448. {"relnamespace", [](const NPg::TTableInfo&, ui32 namespaceOid,ui32) { return ScalarDatumToPod(ObjectIdGetDatum(namespaceOid)); }},
  449. {"relowner", [](const NPg::TTableInfo&, ui32, ui32) { return ScalarDatumToPod(ObjectIdGetDatum(1)); }},
  450. {"relam", [](const NPg::TTableInfo&, ui32, ui32 amOid) { return ScalarDatumToPod(ObjectIdGetDatum(amOid)); }},
  451. };
  452. ApplyFillers(AllPgClassFillers, Y_ARRAY_SIZE(AllPgClassFillers), PgClassFillers_);
  453. } else if (Table_ == "pg_proc") {
  454. static const std::pair<const char*, TPgProcFiller> AllPgProcFillers[] = {
  455. {"oid", [](const NPg::TProcDesc& desc) { return ScalarDatumToPod(ObjectIdGetDatum(desc.ProcId)); }},
  456. {"proname", [](const NPg::TProcDesc& desc) { return PointerDatumToPod((Datum)MakeFixedString(desc.Name, NAMEDATALEN)); }},
  457. {"pronamespace", [](const NPg::TProcDesc&) { return ScalarDatumToPod(ObjectIdGetDatum(PG_CATALOG_NAMESPACE)); }},
  458. {"proowner", [](const NPg::TProcDesc&) { return ScalarDatumToPod(ObjectIdGetDatum(1)); }},
  459. {"prorettype", [](const NPg::TProcDesc& desc) { return ScalarDatumToPod(ObjectIdGetDatum(desc.ResultType)); }},
  460. {"prolang", [](const NPg::TProcDesc& desc) { return ScalarDatumToPod(ObjectIdGetDatum(desc.Lang)); }},
  461. {"prokind", [](const NPg::TProcDesc& desc) { return ScalarDatumToPod(CharGetDatum((char)desc.Kind)); }},
  462. };
  463. ApplyFillers(AllPgProcFillers, Y_ARRAY_SIZE(AllPgProcFillers), PgProcFillers_);
  464. } else if (Table_ == "pg_operator") {
  465. static const std::pair<const char*, TPgOperFiller> AllPgOperFillers[] = {
  466. {"oid", [](const NPg::TOperDesc& desc) { return ScalarDatumToPod(ObjectIdGetDatum(desc.OperId)); }},
  467. {"oprcom", [](const NPg::TOperDesc& desc) { return ScalarDatumToPod(ObjectIdGetDatum(desc.ComId)); }},
  468. {"oprleft", [](const NPg::TOperDesc& desc) { return ScalarDatumToPod(ObjectIdGetDatum(desc.LeftType)); }},
  469. {"oprname", [](const NPg::TOperDesc& desc) { return PointerDatumToPod((Datum)MakeFixedString(desc.Name, NAMEDATALEN)); }},
  470. {"oprnamespace", [](const NPg::TOperDesc& desc) { return ScalarDatumToPod(ObjectIdGetDatum(PG_CATALOG_NAMESPACE)); }},
  471. {"oprnegate", [](const NPg::TOperDesc& desc) { return ScalarDatumToPod(ObjectIdGetDatum(desc.NegateId)); }},
  472. {"oprowner", [](const NPg::TOperDesc& desc) { return ScalarDatumToPod(ObjectIdGetDatum(1)); }},
  473. {"oprresult", [](const NPg::TOperDesc& desc) { return ScalarDatumToPod(ObjectIdGetDatum(desc.ResultType)); }},
  474. {"oprright", [](const NPg::TOperDesc& desc) { return ScalarDatumToPod(ObjectIdGetDatum(desc.RightType)); }},
  475. };
  476. ApplyFillers(AllPgOperFillers, Y_ARRAY_SIZE(AllPgOperFillers), PgOperFillers_);
  477. } else if (Table_ == "pg_aggregate") {
  478. static const std::pair<const char*, TPgAggregateFiller> AllPgAggregateFillers[] = {
  479. {"aggfnoid", [](const NPg::TAggregateDesc& desc) { return ScalarDatumToPod(ObjectIdGetDatum(desc.AggId)); }},
  480. {"aggkind", [](const NPg::TAggregateDesc& desc) { return ScalarDatumToPod(CharGetDatum((char)desc.Kind)); }},
  481. {"aggtranstype", [](const NPg::TAggregateDesc& desc) { return ScalarDatumToPod(ObjectIdGetDatum(desc.TransTypeId)); }},
  482. };
  483. ApplyFillers(AllPgAggregateFillers, Y_ARRAY_SIZE(AllPgAggregateFillers), PgAggregateFillers_);
  484. } else if (Table_ == "pg_language") {
  485. static const std::pair<const char*, TPgLanguageFiller> AllPgLanguageFillers[] = {
  486. {"oid", [](const NPg::TLanguageDesc& desc) { return ScalarDatumToPod(ObjectIdGetDatum(desc.LangId)); }},
  487. {"lanname", [](const NPg::TLanguageDesc& desc) { return PointerDatumToPod((Datum)MakeFixedString(desc.Name, NAMEDATALEN)); }},
  488. {"lanowner", [](const NPg::TLanguageDesc&) { return ScalarDatumToPod(ObjectIdGetDatum(1)); }},
  489. };
  490. ApplyFillers(AllPgLanguageFillers, Y_ARRAY_SIZE(AllPgLanguageFillers), PgLanguageFillers_);
  491. } else if (Table_ == "pg_extension") {
  492. static const std::pair<const char*, TPgExtensionFiller> AllPgExtensionFillers[] = {
  493. {"oid", [](ui32 oid,const NPg::TExtensionDesc& desc) { return ScalarDatumToPod(ObjectIdGetDatum(oid)); }},
  494. {"extname", [](ui32,const NPg::TExtensionDesc& desc) { return PointerDatumToPod((Datum)MakeFixedString(desc.Name, NAMEDATALEN)); }},
  495. {"extowner", [](ui32,const NPg::TExtensionDesc&) { return ScalarDatumToPod(ObjectIdGetDatum(1)); }},
  496. {"extnamespace", [](ui32,const NPg::TExtensionDesc&) { return ScalarDatumToPod(ObjectIdGetDatum(PG_CATALOG_NAMESPACE)); }},
  497. {"extversion", [](ui32,const NPg::TExtensionDesc& desc) { return PointerDatumToPod((Datum)MakeVar(desc.Version)); }},
  498. {"extrelocatable", [](ui32,const NPg::TExtensionDesc&) { return ScalarDatumToPod(BoolGetDatum(false)); }},
  499. };
  500. ApplyFillers(AllPgExtensionFillers, Y_ARRAY_SIZE(AllPgExtensionFillers), PgExtensionFillers_);
  501. }
  502. } else {
  503. if (Table_ == "tables") {
  504. static const std::pair<const char*, TTablesFiller> AllTablesFillers[] = {
  505. {"table_schema", [](const NPg::TTableInfo& desc) { return PointerDatumToPod((Datum)MakeFixedString(desc.Schema, NAMEDATALEN)); }},
  506. {"table_name", [](const NPg::TTableInfo& desc) { return PointerDatumToPod((Datum)MakeFixedString(desc.Name, NAMEDATALEN)); }},
  507. };
  508. ApplyFillers(AllTablesFillers, Y_ARRAY_SIZE(AllTablesFillers), TablesFillers_);
  509. } else if (Table_ == "columns") {
  510. static const std::pair<const char*, TColumnsFiller> AllColumnsFillers[] = {
  511. {"table_schema", [](const NPg::TColumnInfo& desc) { return PointerDatumToPod((Datum)MakeFixedString(desc.Schema, NAMEDATALEN)); }},
  512. {"table_name", [](const NPg::TColumnInfo& desc) { return PointerDatumToPod((Datum)MakeFixedString(desc.TableName, NAMEDATALEN)); }},
  513. {"column_name", [](const NPg::TColumnInfo& desc) { return PointerDatumToPod((Datum)MakeFixedString(desc.Name, NAMEDATALEN)); }},
  514. {"udt_name", [](const NPg::TColumnInfo& desc) { return PointerDatumToPod((Datum)MakeFixedString(desc.UdtType, NAMEDATALEN)); }},
  515. };
  516. ApplyFillers(AllColumnsFillers, Y_ARRAY_SIZE(AllColumnsFillers), ColumnsFillers_);
  517. }
  518. }
  519. }
  520. template <typename T, typename F>
  521. void ApplyFillers(const T* allFillers, size_t n, TVector<F>& fillers) {
  522. fillers.resize(ItemType_->GetMembersCount());
  523. for (size_t i = 0; i < n; ++i) {
  524. const auto& [name, func] = allFillers[i];
  525. if (auto pos = ItemType_->FindMemberIndex(name)) {
  526. fillers[*pos] = func;
  527. }
  528. }
  529. }
  530. class TSystemColumnFiller {
  531. public:
  532. TSystemColumnFiller(TStructType* itemType, const TString& cluster, const TString& table) {
  533. const auto& info = NPg::LookupStaticTable(NPg::TTableInfoKey{cluster, table});
  534. TableOid = info.Oid;
  535. if (info.Kind != NPg::ERelKind::Relation) {
  536. return;
  537. }
  538. TableOidPos = itemType->FindMemberIndex("_yql_virtual_tableoid");
  539. }
  540. void Fill(NUdf::TUnboxedValue* items) {
  541. if (TableOidPos) {
  542. items[*TableOidPos] = ScalarDatumToPod(Int32GetDatum(TableOid));
  543. }
  544. }
  545. private:
  546. ui32 TableOid = 0;
  547. TMaybe<ui32> TableOidPos;
  548. };
  549. NUdf::TUnboxedValuePod DoCalculate(TComputationContext& compCtx) const {
  550. TUnboxedValueVector rows;
  551. if (Cluster_ == "pg_catalog") {
  552. TSystemColumnFiller sysFiller(ItemType_, TString(Cluster_), TString(Table_));
  553. if (Table_ == "pg_type") {
  554. NPg::EnumTypes([&](ui32 oid, const NPg::TTypeDesc& desc) {
  555. NUdf::TUnboxedValue* items;
  556. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(PgTypeFillers_.size(), items);
  557. for (ui32 i = 0; i < PgTypeFillers_.size(); ++i) {
  558. if (PgTypeFillers_[i]) {
  559. items[i] = PgTypeFillers_[i](desc);
  560. }
  561. }
  562. sysFiller.Fill(items);
  563. rows.emplace_back(row);
  564. });
  565. } else if (Table_ == "pg_database") {
  566. TVector <ui32> dbOids = {1, 2, 3};
  567. if (PGGetGUCSetting("ydb_database")) {
  568. dbOids.emplace_back(PG_CURRENT_DATABASE_ID);
  569. }
  570. for (ui32 index : dbOids) {
  571. NUdf::TUnboxedValue* items;
  572. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(PgDatabaseFillers_.size(), items);
  573. for (ui32 i = 0; i < PgDatabaseFillers_.size(); ++i) {
  574. if (PgDatabaseFillers_[i]) {
  575. items[i] = PgDatabaseFillers_[i](index);
  576. }
  577. }
  578. sysFiller.Fill(items);
  579. rows.emplace_back(row);
  580. }
  581. } else if (Table_ == "pg_tablespace") {
  582. for (ui32 index = 1; index <= 2; ++index) {
  583. NUdf::TUnboxedValue* items;
  584. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(PgTablespaceFillers_.size(), items);
  585. for (ui32 i = 0; i < PgTablespaceFillers_.size(); ++i) {
  586. if (PgTablespaceFillers_[i]) {
  587. items[i] = PgTablespaceFillers_[i](index);
  588. }
  589. }
  590. sysFiller.Fill(items);
  591. rows.emplace_back(row);
  592. }
  593. } else if (Table_ == "pg_shdescription") {
  594. for (ui32 index = 1; index <= 3; ++index) {
  595. NUdf::TUnboxedValue* items;
  596. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(PgShDescriptionFillers_.size(), items);
  597. for (ui32 i = 0; i < PgShDescriptionFillers_.size(); ++i) {
  598. if (PgShDescriptionFillers_[i]) {
  599. items[i] = PgShDescriptionFillers_[i](index);
  600. }
  601. }
  602. sysFiller.Fill(items);
  603. rows.emplace_back(row);
  604. }
  605. } else if (Table_ == "pg_stat_gssapi") {
  606. NUdf::TUnboxedValue* items;
  607. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(PgStatGssapiFillers_.size(), items);
  608. for (ui32 i = 0; i < PgStatGssapiFillers_.size(); ++i) {
  609. if (PgStatGssapiFillers_[i]) {
  610. items[i] = PgStatGssapiFillers_[i]();
  611. }
  612. }
  613. sysFiller.Fill(items);
  614. rows.emplace_back(row);
  615. } else if (Table_ == "pg_namespace") {
  616. NPg::EnumNamespace([&](ui32 oid, const NPg::TNamespaceDesc& desc) {
  617. NUdf::TUnboxedValue* items;
  618. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(ItemType_->GetMembersCount(), items);
  619. for (ui32 i = 0; i < PgNamespaceFillers_.size(); ++i) {
  620. if (PgNamespaceFillers_[i]) {
  621. items[i] = PgNamespaceFillers_[i](desc);
  622. }
  623. }
  624. sysFiller.Fill(items);
  625. rows.emplace_back(row);
  626. });
  627. } else if (Table_ == "pg_am") {
  628. NPg::EnumAm([&](ui32 oid, const NPg::TAmDesc& desc) {
  629. NUdf::TUnboxedValue* items;
  630. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(ItemType_->GetMembersCount(), items);
  631. for (ui32 i = 0; i < PgAmFillers_.size(); ++i) {
  632. if (PgAmFillers_[i]) {
  633. items[i] = PgAmFillers_[i](desc);
  634. }
  635. }
  636. sysFiller.Fill(items);
  637. rows.emplace_back(row);
  638. });
  639. } else if (Table_ == "pg_description") {
  640. TDescriptionDesc desc;
  641. desc.Classoid = AccessMethodRelationId;
  642. NPg::EnumAm([&](ui32 oid, const NPg::TAmDesc& desc_) {
  643. NUdf::TUnboxedValue* items;
  644. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(PgDescriptionFillers_.size(), items);
  645. for (ui32 i = 0; i < PgDescriptionFillers_.size(); ++i) {
  646. desc.Objoid = oid;
  647. desc.Description = desc_.Descr;
  648. if (PgDescriptionFillers_[i]) {
  649. items[i] = PgDescriptionFillers_[i](desc);
  650. }
  651. }
  652. sysFiller.Fill(items);
  653. rows.emplace_back(row);
  654. });
  655. desc.Classoid = TypeRelationId;
  656. NPg::EnumTypes([&](ui32 oid, const NPg::TTypeDesc& desc_) {
  657. NUdf::TUnboxedValue* items;
  658. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(PgDescriptionFillers_.size(), items);
  659. for (ui32 i = 0; i < PgDescriptionFillers_.size(); ++i) {
  660. desc.Objoid = oid;
  661. desc.Description = desc_.Descr;
  662. if (PgDescriptionFillers_[i]) {
  663. items[i] = PgDescriptionFillers_[i](desc);
  664. }
  665. }
  666. sysFiller.Fill(items);
  667. rows.emplace_back(row);
  668. });
  669. desc.Classoid = NamespaceRelationId;
  670. NPg::EnumNamespace([&](ui32 oid, const NPg::TNamespaceDesc& desc_) {
  671. NUdf::TUnboxedValue* items;
  672. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(PgDescriptionFillers_.size(), items);
  673. for (ui32 i = 0; i < PgDescriptionFillers_.size(); ++i) {
  674. desc.Objoid = oid;
  675. desc.Description = desc_.Descr;
  676. if (PgDescriptionFillers_[i]) {
  677. items[i] = PgDescriptionFillers_[i](desc);
  678. }
  679. }
  680. sysFiller.Fill(items);
  681. rows.emplace_back(row);
  682. });
  683. desc.Classoid = ConversionRelationId;
  684. NPg::EnumConversions([&](const NPg::TConversionDesc& desc_) {
  685. NUdf::TUnboxedValue* items;
  686. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(PgDescriptionFillers_.size(), items);
  687. for (ui32 i = 0; i < PgDescriptionFillers_.size(); ++i) {
  688. desc.Objoid = desc_.ConversionId;
  689. desc.Description = desc_.Descr;
  690. if (PgDescriptionFillers_[i]) {
  691. items[i] = PgDescriptionFillers_[i](desc);
  692. }
  693. }
  694. sysFiller.Fill(items);
  695. rows.emplace_back(row);
  696. });
  697. desc.Classoid = OperatorRelationId;
  698. NPg::EnumOperators([&](const NPg::TOperDesc& desc_) {
  699. NUdf::TUnboxedValue* items;
  700. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(PgDescriptionFillers_.size(), items);
  701. for (ui32 i = 0; i < PgDescriptionFillers_.size(); ++i) {
  702. desc.Objoid = desc_.OperId;
  703. desc.Description = desc_.Descr;
  704. if (PgDescriptionFillers_[i]) {
  705. items[i] = PgDescriptionFillers_[i](desc);
  706. }
  707. }
  708. sysFiller.Fill(items);
  709. rows.emplace_back(row);
  710. });
  711. desc.Classoid = ProcedureRelationId;
  712. NPg::EnumProc([&](ui32, const NPg::TProcDesc& desc_) {
  713. NUdf::TUnboxedValue* items;
  714. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(PgDescriptionFillers_.size(), items);
  715. for (ui32 i = 0; i < PgDescriptionFillers_.size(); ++i) {
  716. desc.Objoid = desc_.ProcId;
  717. desc.Description = desc_.Descr;
  718. if (PgDescriptionFillers_[i]) {
  719. items[i] = PgDescriptionFillers_[i](desc);
  720. }
  721. }
  722. sysFiller.Fill(items);
  723. rows.emplace_back(row);
  724. });
  725. } else if (Table_ == "pg_tables") {
  726. const auto& tables = NPg::GetStaticTables();
  727. for (const auto& t : tables) {
  728. NUdf::TUnboxedValue* items;
  729. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(PgTablesFillers_.size(), items);
  730. for (ui32 i = 0; i < PgTablesFillers_.size(); ++i) {
  731. if (PgTablesFillers_[i]) {
  732. items[i] = PgTablesFillers_[i](t);
  733. }
  734. }
  735. sysFiller.Fill(items);
  736. rows.emplace_back(row);
  737. }
  738. } else if (Table_ == "pg_roles") {
  739. ui32 tableSize = PGGetGUCSetting("ydb_user") ? 2 : 1;
  740. for (ui32 index = 1; index <= tableSize; ++index) {
  741. NUdf::TUnboxedValue* items;
  742. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(PgRolesFillers_.size(), items);
  743. for (ui32 i = 0; i < PgRolesFillers_.size(); ++i) {
  744. if (PgRolesFillers_[i]) {
  745. items[i] = PgRolesFillers_[i](index);
  746. }
  747. }
  748. sysFiller.Fill(items);
  749. rows.emplace_back(row);
  750. }
  751. } else if (Table_ == "pg_user") {
  752. ui32 tableSize = PGGetGUCSetting("ydb_user") ? 2 : 1;
  753. for (ui32 index = 1; index <= tableSize; ++index) {
  754. NUdf::TUnboxedValue* items;
  755. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(PgUserFillers_.size(), items);
  756. for (ui32 i = 0; i < PgUserFillers_.size(); ++i) {
  757. if (PgUserFillers_[i]) {
  758. items[i] = PgUserFillers_[i](index);
  759. }
  760. }
  761. sysFiller.Fill(items);
  762. rows.emplace_back(row);
  763. }
  764. } else if (Table_ == "pg_stat_database") {
  765. for (ui32 index = 0; index <= 1; ++index) {
  766. NUdf::TUnboxedValue* items;
  767. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(PgDatabaseStatFillers_.size(), items);
  768. for (ui32 i = 0; i < PgDatabaseStatFillers_.size(); ++i) {
  769. if (PgDatabaseStatFillers_[i]) {
  770. items[i] = PgDatabaseStatFillers_[i](index);
  771. }
  772. }
  773. sysFiller.Fill(items);
  774. rows.emplace_back(row);
  775. }
  776. } else if (Table_ == "pg_class") {
  777. const auto& tables = NPg::GetStaticTables();
  778. THashMap<TString, ui32> namespaces;
  779. NPg::EnumNamespace([&](ui32 oid, const NPg::TNamespaceDesc& desc) {
  780. namespaces[desc.Name] = oid;
  781. });
  782. ui32 btreeAmOid = 0;
  783. NPg::EnumAm([&](ui32 oid, const NPg::TAmDesc& desc) {
  784. if (desc.AmName == "btree") {
  785. btreeAmOid = oid;
  786. }
  787. });
  788. for (const auto& t : tables) {
  789. const ui32 amOid = (t.Kind == NPg::ERelKind::Relation) ? btreeAmOid : 0;
  790. NUdf::TUnboxedValue* items;
  791. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(PgClassFillers_.size(), items);
  792. for (ui32 i = 0; i < PgClassFillers_.size(); ++i) {
  793. if (PgClassFillers_[i]) {
  794. items[i] = PgClassFillers_[i](t, namespaces[t.Schema], amOid);
  795. }
  796. }
  797. sysFiller.Fill(items);
  798. rows.emplace_back(row);
  799. }
  800. } else if (Table_ == "pg_proc") {
  801. NPg::EnumProc([&](ui32, const NPg::TProcDesc& desc) {
  802. NUdf::TUnboxedValue* items;
  803. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(PgProcFillers_.size(), items);
  804. for (ui32 i = 0; i < PgProcFillers_.size(); ++i) {
  805. if (PgProcFillers_[i]) {
  806. items[i] = PgProcFillers_[i](desc);
  807. }
  808. }
  809. sysFiller.Fill(items);
  810. rows.emplace_back(row);
  811. });
  812. } else if (Table_ == "pg_operator") {
  813. NPg::EnumOperators([&](const NPg::TOperDesc& desc) {
  814. NUdf::TUnboxedValue* items;
  815. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(PgOperFillers_.size(), items);
  816. for (ui32 i = 0; i < PgOperFillers_.size(); ++i) {
  817. if (PgOperFillers_[i]) {
  818. items[i] = PgOperFillers_[i](desc);
  819. }
  820. }
  821. sysFiller.Fill(items);
  822. rows.emplace_back(row);
  823. });
  824. } else if (Table_ == "pg_aggregate") {
  825. NPg::EnumAggregation([&](ui32, const NPg::TAggregateDesc& desc) {
  826. NUdf::TUnboxedValue* items;
  827. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(PgAggregateFillers_.size(), items);
  828. for (ui32 i = 0; i < PgAggregateFillers_.size(); ++i) {
  829. if (PgAggregateFillers_[i]) {
  830. items[i] = PgAggregateFillers_[i](desc);
  831. }
  832. }
  833. sysFiller.Fill(items);
  834. rows.emplace_back(row);
  835. });
  836. } else if (Table_ == "pg_language") {
  837. NPg::EnumLanguages([&](ui32, const NPg::TLanguageDesc& desc) {
  838. NUdf::TUnboxedValue* items;
  839. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(PgLanguageFillers_.size(), items);
  840. for (ui32 i = 0; i < PgLanguageFillers_.size(); ++i) {
  841. if (PgLanguageFillers_[i]) {
  842. items[i] = PgLanguageFillers_[i](desc);
  843. }
  844. }
  845. sysFiller.Fill(items);
  846. rows.emplace_back(row);
  847. });
  848. } else if (Table_ == "pg_extension") {
  849. NPg::EnumExtensions([&](ui32 oid, const NPg::TExtensionDesc& desc) {
  850. NUdf::TUnboxedValue* items;
  851. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(PgExtensionFillers_.size(), items);
  852. for (ui32 i = 0; i < PgExtensionFillers_.size(); ++i) {
  853. if (PgExtensionFillers_[i]) {
  854. items[i] = PgExtensionFillers_[i](oid, desc);
  855. }
  856. }
  857. sysFiller.Fill(items);
  858. rows.emplace_back(row);
  859. });
  860. }
  861. } else {
  862. if (Table_ == "tables") {
  863. const auto& tables = NPg::GetStaticTables();
  864. for (const auto& t : tables) {
  865. NUdf::TUnboxedValue* items;
  866. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(TablesFillers_.size(), items);
  867. for (ui32 i = 0; i < TablesFillers_.size(); ++i) {
  868. if (TablesFillers_[i]) {
  869. items[i] = TablesFillers_[i](t);
  870. }
  871. }
  872. rows.emplace_back(row);
  873. }
  874. } else if (Table_ == "columns") {
  875. const auto& columns = NPg::GetStaticColumns();
  876. for (const auto& t : columns) {
  877. for (const auto& c : t.second) {
  878. NUdf::TUnboxedValue* items;
  879. auto row = compCtx.HolderFactory.CreateDirectArrayHolder(ColumnsFillers_.size(), items);
  880. for (ui32 i = 0; i < ColumnsFillers_.size(); ++i) {
  881. if (ColumnsFillers_[i]) {
  882. items[i] = ColumnsFillers_[i](c);
  883. }
  884. }
  885. rows.emplace_back(row);
  886. }
  887. }
  888. }
  889. }
  890. return compCtx.HolderFactory.VectorAsVectorHolder(std::move(rows));
  891. }
  892. private:
  893. void RegisterDependencies() const final {
  894. }
  895. const std::string_view Cluster_;
  896. const std::string_view Table_;
  897. TStructType* const ItemType_;
  898. using TPgTypeFiller = NUdf::TUnboxedValuePod(*)(const NPg::TTypeDesc& desc);
  899. TVector<TPgTypeFiller> PgTypeFillers_;
  900. using TPgDatabaseFiller = NUdf::TUnboxedValuePod(*)(ui32 index);
  901. TVector<TPgDatabaseFiller> PgDatabaseFillers_;
  902. using TPgTablespaceFiller = NUdf::TUnboxedValuePod(*)(ui32 index);
  903. TVector<TPgTablespaceFiller> PgTablespaceFillers_;
  904. using TPgShDescriptionFiller = NUdf::TUnboxedValuePod(*)(ui32 index);
  905. TVector<TPgShDescriptionFiller> PgShDescriptionFillers_;
  906. using TPgStatGssapiFiller = NUdf::TUnboxedValuePod(*)();
  907. TVector<TPgStatGssapiFiller> PgStatGssapiFillers_;
  908. using TPgNamespaceFiller = NUdf::TUnboxedValuePod(*)(const NPg::TNamespaceDesc&);
  909. TVector<TPgNamespaceFiller> PgNamespaceFillers_;
  910. using TPgAmFiller = NUdf::TUnboxedValuePod(*)(const NPg::TAmDesc&);
  911. TVector<TPgAmFiller> PgAmFillers_;
  912. using TPgRolesFiller = NUdf::TUnboxedValuePod(*)(ui32 index);
  913. TVector<TPgRolesFiller> PgRolesFillers_;
  914. using TPgUserFiller = NUdf::TUnboxedValuePod(*)(ui32 index);
  915. TVector<TPgUserFiller> PgUserFillers_;
  916. using TPgDatabaseStatFiller = NUdf::TUnboxedValuePod(*)(ui32 index);
  917. TVector<TPgDatabaseStatFiller> PgDatabaseStatFillers_;
  918. struct TDescriptionDesc {
  919. ui32 Objoid = 0;
  920. ui32 Classoid = 0;
  921. i32 Objsubid = 0;
  922. TString Description;
  923. };
  924. using TPgDescriptionFiller = NUdf::TUnboxedValuePod(*)(const TDescriptionDesc&);
  925. TVector<TPgDescriptionFiller> PgDescriptionFillers_;
  926. using TTablesFiller = NUdf::TUnboxedValuePod(*)(const NPg::TTableInfo&);
  927. TVector<TTablesFiller> PgTablesFillers_;
  928. TVector<TTablesFiller> TablesFillers_;
  929. using TColumnsFiller = NUdf::TUnboxedValuePod(*)(const NPg::TColumnInfo&);
  930. TVector<TColumnsFiller> ColumnsFillers_;
  931. using TPgClassFiller = NUdf::TUnboxedValuePod(*)(const NPg::TTableInfo&, ui32 namespaceOid, ui32 amOid);
  932. TVector<TPgClassFiller> PgClassFillers_;
  933. using TPgProcFiller = NUdf::TUnboxedValuePod(*)(const NPg::TProcDesc&);
  934. TVector<TPgProcFiller> PgProcFillers_;
  935. using TPgAggregateFiller = NUdf::TUnboxedValuePod(*)(const NPg::TAggregateDesc&);
  936. TVector<TPgAggregateFiller> PgAggregateFillers_;
  937. using TPgLanguageFiller = NUdf::TUnboxedValuePod(*)(const NPg::TLanguageDesc&);
  938. TVector<TPgLanguageFiller> PgLanguageFillers_;
  939. using TPgOperFiller = NUdf::TUnboxedValuePod(*)(const NPg::TOperDesc&);
  940. TVector<TPgOperFiller> PgOperFillers_;
  941. using TPgExtensionFiller = NUdf::TUnboxedValuePod(*)(ui32,const NPg::TExtensionDesc&);
  942. TVector<TPgExtensionFiller> PgExtensionFillers_;
  943. };
  944. class TFunctionCallInfo {
  945. public:
  946. TFunctionCallInfo(ui32 numArgs, const FmgrInfo* finfo)
  947. : NumArgs(numArgs)
  948. , CopyFmgrInfo(*finfo)
  949. {
  950. if (!finfo->fn_addr) {
  951. return;
  952. }
  953. MemSize = SizeForFunctionCallInfo(numArgs);
  954. Ptr = TWithDefaultMiniKQLAlloc::AllocWithSize(MemSize);
  955. auto& callInfo = Ref();
  956. Zero(callInfo);
  957. callInfo.flinfo = &CopyFmgrInfo; // client may mutate fn_extra
  958. callInfo.nargs = NumArgs;
  959. callInfo.fncollation = DEFAULT_COLLATION_OID;
  960. }
  961. FunctionCallInfoBaseData& Ref() {
  962. Y_ENSURE(Ptr);
  963. return *(FunctionCallInfoBaseData*)Ptr;
  964. }
  965. ~TFunctionCallInfo() {
  966. if (Ptr) {
  967. TWithDefaultMiniKQLAlloc::FreeWithSize(Ptr, MemSize);
  968. }
  969. }
  970. TFunctionCallInfo(const TFunctionCallInfo&) = delete;
  971. void operator=(const TFunctionCallInfo&) = delete;
  972. private:
  973. const ui32 NumArgs = 0;
  974. ui32 MemSize = 0;
  975. void* Ptr = nullptr;
  976. FmgrInfo CopyFmgrInfo;
  977. };
  978. class TReturnSetInfo {
  979. public:
  980. TReturnSetInfo() {
  981. Ptr = TWithDefaultMiniKQLAlloc::AllocWithSize(sizeof(ReturnSetInfo));
  982. Zero(Ref());
  983. Ref().type = T_ReturnSetInfo;
  984. }
  985. ~TReturnSetInfo() {
  986. Free();
  987. }
  988. void Free() {
  989. if (!Ptr) {
  990. return;
  991. }
  992. if (Ref().expectedDesc) {
  993. FreeTupleDesc(Ref().expectedDesc);
  994. }
  995. if (Ref().setResult) {
  996. tuplestore_end(Ref().setResult);
  997. }
  998. TWithDefaultMiniKQLAlloc::FreeWithSize(Ptr, sizeof(ReturnSetInfo));
  999. Ptr = nullptr;
  1000. }
  1001. ReturnSetInfo& Ref() {
  1002. Y_ENSURE(Ptr, "ReturnSetInfo is dead");
  1003. return *static_cast<ReturnSetInfo*>(Ptr);
  1004. }
  1005. private:
  1006. void* Ptr = nullptr;
  1007. };
  1008. class TExprContextHolder {
  1009. public:
  1010. TExprContextHolder() {
  1011. Ptr = CreateStandaloneExprContext();
  1012. }
  1013. ExprContext& Ref() {
  1014. Y_ENSURE(Ptr, "TExprContextHolder is dead");
  1015. return *Ptr;
  1016. }
  1017. ~TExprContextHolder() {
  1018. Free();
  1019. }
  1020. void Free() {
  1021. if (!Ptr) {
  1022. return;
  1023. }
  1024. FreeExprContext(Ptr, true);
  1025. Ptr = nullptr;
  1026. }
  1027. private:
  1028. ExprContext* Ptr;
  1029. };
  1030. class TPgArgsExprBuilder {
  1031. public:
  1032. TPgArgsExprBuilder()
  1033. : PgFuncArgsList(nullptr, &free)
  1034. {}
  1035. void Add(ui32 argOid)
  1036. {
  1037. PgArgNodes.emplace_back();
  1038. auto& p = PgArgNodes.back();
  1039. Zero(p);
  1040. p.xpr.type = T_Param;
  1041. p.paramkind = PARAM_EXTERN;
  1042. p.paramtype = argOid;
  1043. p.paramcollid = DEFAULT_COLLATION_OID;
  1044. p.paramtypmod = -1;
  1045. p.paramid = PgArgNodes.size();
  1046. }
  1047. Node* Build(const NPg::TProcDesc& procDesc) {
  1048. PgFuncArgsList.reset((List*)malloc(offsetof(List, initial_elements) + PgArgNodes.size() * sizeof(ListCell)));
  1049. PgFuncArgsList->type = T_List;
  1050. PgFuncArgsList->elements = PgFuncArgsList->initial_elements;
  1051. PgFuncArgsList->length = PgFuncArgsList->max_length = PgArgNodes.size();
  1052. for (size_t i = 0; i < PgArgNodes.size(); ++i) {
  1053. PgFuncArgsList->elements[i].ptr_value = &PgArgNodes[i];
  1054. }
  1055. Zero(PgFuncNode);
  1056. PgFuncNode.xpr.type = T_FuncExpr;
  1057. PgFuncNode.funcid = procDesc.ProcId;
  1058. PgFuncNode.funcresulttype = procDesc.ResultType;
  1059. PgFuncNode.funcretset = procDesc.ReturnSet;
  1060. PgFuncNode.funcvariadic = procDesc.VariadicArgType && procDesc.VariadicArgType != procDesc.VariadicType;
  1061. PgFuncNode.args = PgFuncArgsList.get();
  1062. return (Node*)&PgFuncNode;
  1063. }
  1064. private:
  1065. TVector<Param> PgArgNodes;
  1066. std::unique_ptr<List, decltype(&free)> PgFuncArgsList;
  1067. FuncExpr PgFuncNode;
  1068. };
  1069. template <typename TDerived>
  1070. class TPgResolvedCallBase : public TMutableComputationNode<TDerived> {
  1071. typedef TMutableComputationNode<TDerived> TBaseComputation;
  1072. public:
  1073. TPgResolvedCallBase(TComputationMutables& mutables, const std::string_view& name, ui32 id,
  1074. TComputationNodePtrVector&& argNodes, TVector<TType*>&& argTypes, TType* returnType,
  1075. bool isList, const TStructType* structType)
  1076. : TBaseComputation(mutables)
  1077. , Name(name)
  1078. , Id(id)
  1079. , ProcDesc(NPg::LookupProc(id))
  1080. , RetTypeDesc(NPg::LookupType(returnType->IsStruct() ? RECORDOID : AS_TYPE(TPgType, returnType)->GetTypeId()))
  1081. , ArgNodes(std::move(argNodes))
  1082. , ArgTypes(std::move(argTypes))
  1083. , StructType(structType)
  1084. {
  1085. Zero(FInfo);
  1086. Y_ENSURE(Id);
  1087. GetPgFuncAddr(Id, FInfo);
  1088. Y_ENSURE(FInfo.fn_retset == isList);
  1089. Y_ENSURE(FInfo.fn_addr);
  1090. Y_ENSURE(ArgNodes.size() <= FUNC_MAX_ARGS);
  1091. ArgDesc.reserve(ArgTypes.size());
  1092. for (ui32 i = 0; i < ArgTypes.size(); ++i) {
  1093. ui32 type;
  1094. // extract real type from input args
  1095. auto argType = ArgTypes[i];
  1096. if (argType->IsPg()) {
  1097. type = static_cast<TPgType*>(argType)->GetTypeId();
  1098. } else {
  1099. // keep original description for nulls
  1100. type = ProcDesc.ArgTypes[i];
  1101. }
  1102. ArgDesc.emplace_back(NPg::LookupType(type));
  1103. }
  1104. Y_ENSURE(ArgDesc.size() == ArgNodes.size());
  1105. for (size_t i = 0; i < ArgDesc.size(); ++i) {
  1106. ArgsExprBuilder.Add(ArgDesc[i].TypeId);
  1107. }
  1108. FInfo.fn_expr = ArgsExprBuilder.Build(ProcDesc);
  1109. if (StructType) {
  1110. StructTypeDesc.reserve(StructType->GetMembersCount());
  1111. for (ui32 i = 0; i < StructType->GetMembersCount(); ++i) {
  1112. auto itemType = StructType->GetMemberType(i);
  1113. auto type = AS_TYPE(TPgType, itemType)->GetTypeId();
  1114. StructTypeDesc.emplace_back(NPg::LookupType(type));
  1115. }
  1116. }
  1117. }
  1118. private:
  1119. void RegisterDependencies() const final {
  1120. for (const auto node : ArgNodes) {
  1121. this->DependsOn(node);
  1122. }
  1123. }
  1124. protected:
  1125. const std::string_view Name;
  1126. const ui32 Id;
  1127. FmgrInfo FInfo;
  1128. const NPg::TProcDesc ProcDesc;
  1129. const NPg::TTypeDesc RetTypeDesc;
  1130. const TComputationNodePtrVector ArgNodes;
  1131. const TVector<TType*> ArgTypes;
  1132. const TStructType* StructType;
  1133. TVector<NPg::TTypeDesc> ArgDesc;
  1134. TVector<NPg::TTypeDesc> StructTypeDesc;
  1135. TPgArgsExprBuilder ArgsExprBuilder;
  1136. };
  1137. struct TPgResolvedCallState : public TComputationValue<TPgResolvedCallState> {
  1138. TPgResolvedCallState(TMemoryUsageInfo* memInfo, ui32 numArgs, const FmgrInfo* finfo)
  1139. : TComputationValue(memInfo)
  1140. , CallInfo(numArgs, finfo)
  1141. , Args(numArgs)
  1142. {
  1143. }
  1144. TFunctionCallInfo CallInfo;
  1145. TUnboxedValueVector Args;
  1146. };
  1147. template <bool UseContext>
  1148. class TPgResolvedCall : public TPgResolvedCallBase<TPgResolvedCall<UseContext>> {
  1149. typedef TPgResolvedCallBase<TPgResolvedCall<UseContext>> TBaseComputation;
  1150. public:
  1151. TPgResolvedCall(TComputationMutables& mutables, const std::string_view& name, ui32 id,
  1152. TComputationNodePtrVector&& argNodes, TVector<TType*>&& argTypes, TType* returnType)
  1153. : TBaseComputation(mutables, name, id, std::move(argNodes), std::move(argTypes), returnType, false, nullptr)
  1154. , StateIndex(mutables.CurValueIndex++)
  1155. {
  1156. }
  1157. NUdf::TUnboxedValuePod DoCalculate(TComputationContext& compCtx) const {
  1158. auto& state = this->GetState(compCtx);
  1159. auto& callInfo = state.CallInfo.Ref();
  1160. auto& args = state.Args;
  1161. if constexpr (UseContext) {
  1162. callInfo.context = (Node*)TlsAllocState->CurrentContext;
  1163. }
  1164. callInfo.isnull = false;
  1165. for (ui32 i = 0; i < this->ArgNodes.size(); ++i) {
  1166. args[i] = std::move(this->ArgNodes[i]->GetValue(compCtx));
  1167. auto& value = args[i];
  1168. NullableDatum argDatum = { 0, false };
  1169. if (!value) {
  1170. if (this->FInfo.fn_strict) {
  1171. return NUdf::TUnboxedValuePod();
  1172. }
  1173. argDatum.isnull = true;
  1174. } else {
  1175. argDatum.value = this->ArgDesc[i].PassByValue ?
  1176. ScalarDatumFromPod(value) :
  1177. PointerDatumFromPod(value);
  1178. }
  1179. callInfo.args[i] = argDatum;
  1180. }
  1181. const bool needToFree = PrepareVariadicArray(callInfo, this->ProcDesc);
  1182. NUdf::TUnboxedValuePod res;
  1183. if constexpr (!UseContext) {
  1184. TPAllocScope call;
  1185. res = this->DoCall(callInfo);
  1186. } else {
  1187. res = this->DoCall(callInfo);
  1188. }
  1189. if (needToFree) {
  1190. FreeVariadicArray(callInfo, this->ArgNodes.size());
  1191. }
  1192. return res;
  1193. }
  1194. private:
  1195. NUdf::TUnboxedValuePod DoCall(FunctionCallInfoBaseData& callInfo) const {
  1196. auto ret = this->FInfo.fn_addr(&callInfo);
  1197. if (callInfo.isnull) {
  1198. return NUdf::TUnboxedValuePod();
  1199. }
  1200. return AnyDatumToPod(ret, this->RetTypeDesc.PassByValue);
  1201. }
  1202. TPgResolvedCallState& GetState(TComputationContext& compCtx) const {
  1203. auto& result = compCtx.MutableValues[this->StateIndex];
  1204. if (!result.HasValue()) {
  1205. result = compCtx.HolderFactory.Create<TPgResolvedCallState>(this->ArgNodes.size(), &this->FInfo);
  1206. }
  1207. return *static_cast<TPgResolvedCallState*>(result.AsBoxed().Get());
  1208. }
  1209. const ui32 StateIndex;
  1210. };
  1211. class TPgResolvedMultiCall : public TPgResolvedCallBase<TPgResolvedMultiCall> {
  1212. typedef TPgResolvedCallBase<TPgResolvedMultiCall> TBaseComputation;
  1213. private:
  1214. class TListValue : public TCustomListValue {
  1215. public:
  1216. class TIterator : public TComputationValue<TIterator> {
  1217. public:
  1218. TIterator(TMemoryUsageInfo* memInfo, const std::string_view& name, const TUnboxedValueVector& args,
  1219. const TVector<NPg::TTypeDesc>& argDesc, const NPg::TTypeDesc& retTypeDesc, const NPg::TProcDesc& procDesc,
  1220. const FmgrInfo* fInfo, const TStructType* structType, const TVector<NPg::TTypeDesc>& structTypeDesc, const THolderFactory& holderFactory)
  1221. : TComputationValue<TIterator>(memInfo)
  1222. , Name(name)
  1223. , Args(args)
  1224. , ArgDesc(argDesc)
  1225. , RetTypeDesc(retTypeDesc)
  1226. , ProcDesc(procDesc)
  1227. , CallInfo(argDesc.size(), fInfo)
  1228. , StructType(structType)
  1229. , StructTypeDesc(structTypeDesc)
  1230. , HolderFactory(holderFactory)
  1231. {
  1232. auto& callInfo = CallInfo.Ref();
  1233. callInfo.resultinfo = (fmNodePtr)&RSInfo.Ref();
  1234. auto& rsInfo = *(ReturnSetInfo*)callInfo.resultinfo;
  1235. rsInfo.econtext = &ExprContextHolder.Ref();
  1236. rsInfo.allowedModes = (int) (SFRM_ValuePerCall | SFRM_Materialize);
  1237. rsInfo.returnMode = SFRM_ValuePerCall;
  1238. rsInfo.setResult = nullptr;
  1239. rsInfo.setDesc = nullptr;
  1240. if (RetTypeDesc.TypeId != RECORDOID) {
  1241. rsInfo.expectedDesc = CreateTemplateTupleDesc(1);
  1242. TupleDescInitEntry(rsInfo.expectedDesc, (AttrNumber) 1, nullptr, RetTypeDesc.TypeId, -1, 0);
  1243. } else {
  1244. if (StructType) {
  1245. YQL_ENSURE(ProcDesc.OutputArgNames.size() == ProcDesc.OutputArgTypes.size());
  1246. YQL_ENSURE(ProcDesc.OutputArgNames.size() == StructType->GetMembersCount());
  1247. StructIndicies.resize(StructType->GetMembersCount());
  1248. }
  1249. rsInfo.expectedDesc = CreateTemplateTupleDesc(ProcDesc.OutputArgTypes.size());
  1250. for (size_t i = 0; i < ProcDesc.OutputArgTypes.size(); ++i) {
  1251. auto attrName = ProcDesc.OutputArgNames.empty() ? nullptr : ProcDesc.OutputArgNames[i].c_str();
  1252. TupleDescInitEntry(rsInfo.expectedDesc, (AttrNumber) 1 + i, attrName, ProcDesc.OutputArgTypes[i], -1, 0);
  1253. if (StructType) {
  1254. auto index = StructType->FindMemberIndex(ProcDesc.OutputArgNames[i]);
  1255. YQL_ENSURE(index);
  1256. StructIndicies[i] = *index;
  1257. }
  1258. }
  1259. rsInfo.expectedDesc = BlessTupleDesc(rsInfo.expectedDesc);
  1260. }
  1261. TupleSlot = MakeSingleTupleTableSlot(rsInfo.expectedDesc, &TTSOpsMinimalTuple);
  1262. for (ui32 i = 0; i < args.size(); ++i) {
  1263. const auto& value = args[i];
  1264. NullableDatum argDatum = { 0, false };
  1265. if (!value) {
  1266. argDatum.isnull = true;
  1267. if (callInfo.flinfo->fn_strict) {
  1268. IsFinished = true;
  1269. break;
  1270. }
  1271. } else {
  1272. argDatum.value = ArgDesc[i].PassByValue ?
  1273. ScalarDatumFromPod(value) :
  1274. PointerDatumFromPod(value);
  1275. }
  1276. callInfo.args[i] = argDatum;
  1277. }
  1278. }
  1279. ~TIterator() {
  1280. FinishAndFree();
  1281. }
  1282. private:
  1283. bool Next(NUdf::TUnboxedValue& value) final {
  1284. if (IsFinished) {
  1285. return false;
  1286. }
  1287. if (RSInfo.Ref().setResult) {
  1288. return CopyTuple(value);
  1289. }
  1290. auto& callInfo = CallInfo.Ref();
  1291. callInfo.isnull = false;
  1292. auto ret = callInfo.flinfo->fn_addr(&callInfo);
  1293. if (RSInfo.Ref().returnMode == SFRM_Materialize) {
  1294. Y_ENSURE(RSInfo.Ref().isDone == ExprSingleResult);
  1295. Y_ENSURE(RSInfo.Ref().setResult);
  1296. auto readPtr = tuplestore_alloc_read_pointer(RSInfo.Ref().setResult, 0);
  1297. tuplestore_select_read_pointer(RSInfo.Ref().setResult, readPtr);
  1298. return CopyTuple(value);
  1299. } else {
  1300. if (RSInfo.Ref().isDone == ExprEndResult) {
  1301. FinishAndFree();
  1302. return false;
  1303. }
  1304. if (StructType) {
  1305. YQL_ENSURE(!callInfo.isnull);
  1306. auto tuple = DatumGetHeapTupleHeader(ret);
  1307. YQL_ENSURE(HeapTupleHeaderGetNatts(tuple) == StructType->GetMembersCount());
  1308. HeapTupleData tmptup;
  1309. tmptup.t_len = HeapTupleHeaderGetDatumLength(tuple);
  1310. ItemPointerSetInvalid(&(tmptup.t_self));
  1311. tmptup.t_tableOid = InvalidOid;
  1312. tmptup.t_data = tuple;
  1313. NUdf::TUnboxedValue* itemsPtr;
  1314. value = HolderFactory.CreateDirectArrayHolder(StructType->GetMembersCount(), itemsPtr);
  1315. for (ui32 i = 0; i < StructType->GetMembersCount(); ++i) {
  1316. bool isNull;
  1317. auto datum = heap_getattr(&tmptup,i + 1,RSInfo.Ref().expectedDesc,&isNull);
  1318. itemsPtr[StructIndicies[i]] = CloneTupleItem(i, isNull, datum);
  1319. }
  1320. } else {
  1321. if (callInfo.isnull) {
  1322. value = NUdf::TUnboxedValuePod();
  1323. } else {
  1324. if (RetTypeDesc.PassByValue) {
  1325. value = ScalarDatumToPod(ret);
  1326. } else {
  1327. auto cloned = datumCopy(ret, false, RetTypeDesc.TypeLen);
  1328. value = PointerDatumToPod(cloned);
  1329. }
  1330. }
  1331. }
  1332. return true;
  1333. }
  1334. }
  1335. bool CopyTuple(NUdf::TUnboxedValue& value) {
  1336. if (!tuplestore_gettupleslot(RSInfo.Ref().setResult, true, false, TupleSlot)) {
  1337. FinishAndFree();
  1338. return false;
  1339. }
  1340. slot_getallattrs(TupleSlot);
  1341. if (RetTypeDesc.TypeId == RECORDOID) {
  1342. if (StructType) {
  1343. Y_ENSURE(TupleSlot->tts_nvalid == StructType->GetMembersCount());
  1344. NUdf::TUnboxedValue* itemsPtr;
  1345. value = HolderFactory.CreateDirectArrayHolder(StructType->GetMembersCount(), itemsPtr);
  1346. for (ui32 i = 0; i < StructType->GetMembersCount(); ++i) {
  1347. itemsPtr[StructIndicies[i]] = CloneTupleItem(i);
  1348. }
  1349. } else {
  1350. // whole record value
  1351. auto tupleDesc = RSInfo.Ref().expectedDesc;
  1352. auto tuple = ExecCopySlotHeapTuple(TupleSlot);
  1353. auto result = (HeapTupleHeader) palloc(tuple->t_len);
  1354. memcpy(result, tuple->t_data, tuple->t_len);
  1355. HeapTupleHeaderSetDatumLength(result, tuple->t_len);
  1356. HeapTupleHeaderSetTypeId(result, tupleDesc->tdtypeid);
  1357. HeapTupleHeaderSetTypMod(result, tupleDesc->tdtypmod);
  1358. heap_freetuple(tuple);
  1359. value = PointerDatumToPod(HeapTupleHeaderGetDatum(result));
  1360. }
  1361. } else {
  1362. Y_ENSURE(TupleSlot->tts_nvalid == 1);
  1363. value = CloneTupleItem(0);
  1364. }
  1365. return true;
  1366. }
  1367. NUdf::TUnboxedValuePod CloneTupleItem(ui32 index, bool isNull, Datum datum) {
  1368. if (isNull) {
  1369. return NUdf::TUnboxedValuePod();
  1370. } else {
  1371. const auto& desc = StructType ? StructTypeDesc[StructIndicies[index]] : RetTypeDesc;
  1372. if (desc.PassByValue) {
  1373. return ScalarDatumToPod(datum);
  1374. } else if (desc.TypeLen == -1) {
  1375. const text* orig = (const text*)datum;
  1376. return PointerDatumToPod((Datum)MakeVar(GetVarBuf(orig)));
  1377. } else if(desc.TypeLen == -2) {
  1378. const char* orig = (const char*)datum;
  1379. return PointerDatumToPod((Datum)MakeCString(orig));
  1380. } else {
  1381. const char* orig = (const char*)datum;
  1382. return PointerDatumToPod((Datum)MakeFixedString(orig, desc.TypeLen));
  1383. }
  1384. }
  1385. }
  1386. NUdf::TUnboxedValuePod CloneTupleItem(ui32 index) {
  1387. return CloneTupleItem(index, TupleSlot->tts_isnull[index], TupleSlot->tts_values[index]);
  1388. }
  1389. void FinishAndFree() {
  1390. if (TupleSlot) {
  1391. ExecDropSingleTupleTableSlot(TupleSlot);
  1392. TupleSlot = nullptr;
  1393. }
  1394. RSInfo.Free();
  1395. ExprContextHolder.Free();
  1396. IsFinished = true;
  1397. }
  1398. const std::string_view Name;
  1399. TUnboxedValueVector Args;
  1400. const TVector<NPg::TTypeDesc>& ArgDesc;
  1401. const NPg::TTypeDesc& RetTypeDesc;
  1402. const NPg::TProcDesc& ProcDesc;
  1403. TExprContextHolder ExprContextHolder;
  1404. TFunctionCallInfo CallInfo;
  1405. const TStructType* StructType;
  1406. const TVector<NPg::TTypeDesc>& StructTypeDesc;
  1407. const THolderFactory& HolderFactory;
  1408. TReturnSetInfo RSInfo;
  1409. bool IsFinished = false;
  1410. TupleTableSlot* TupleSlot = nullptr;
  1411. TVector<ui32> StructIndicies;
  1412. };
  1413. TListValue(TMemoryUsageInfo* memInfo, TComputationContext& compCtx,
  1414. const std::string_view& name, TUnboxedValueVector&& args, const TVector<NPg::TTypeDesc>& argDesc,
  1415. const NPg::TTypeDesc& retTypeDesc, const NPg::TProcDesc& procDesc, const FmgrInfo* fInfo,
  1416. const TStructType* structType, const TVector<NPg::TTypeDesc>& structTypeDesc, const THolderFactory& holderFactory)
  1417. : TCustomListValue(memInfo)
  1418. , CompCtx(compCtx)
  1419. , Name(name)
  1420. , Args(args)
  1421. , ArgDesc(argDesc)
  1422. , RetTypeDesc(retTypeDesc)
  1423. , ProcDesc(procDesc)
  1424. , FInfo(fInfo)
  1425. , StructType(structType)
  1426. , StructTypeDesc(structTypeDesc)
  1427. , HolderFactory(holderFactory)
  1428. {
  1429. }
  1430. private:
  1431. NUdf::TUnboxedValue GetListIterator() const final {
  1432. return CompCtx.HolderFactory.Create<TIterator>(Name, Args, ArgDesc, RetTypeDesc, ProcDesc, FInfo, StructType, StructTypeDesc, CompCtx.HolderFactory);
  1433. }
  1434. TComputationContext& CompCtx;
  1435. const std::string_view Name;
  1436. TUnboxedValueVector Args;
  1437. const TVector<NPg::TTypeDesc>& ArgDesc;
  1438. const NPg::TTypeDesc& RetTypeDesc;
  1439. const NPg::TProcDesc& ProcDesc;
  1440. const FmgrInfo* FInfo;
  1441. const TStructType* StructType;
  1442. const TVector<NPg::TTypeDesc>& StructTypeDesc;
  1443. const THolderFactory& HolderFactory;
  1444. };
  1445. public:
  1446. TPgResolvedMultiCall(TComputationMutables& mutables, const std::string_view& name, ui32 id,
  1447. TComputationNodePtrVector&& argNodes, TVector<TType*>&& argTypes, TType* returnType, const TStructType* structType)
  1448. : TBaseComputation(mutables, name, id, std::move(argNodes), std::move(argTypes), returnType, true, structType)
  1449. {
  1450. }
  1451. NUdf::TUnboxedValuePod DoCalculate(TComputationContext& compCtx) const {
  1452. TUnboxedValueVector args;
  1453. args.reserve(ArgNodes.size());
  1454. for (ui32 i = 0; i < ArgNodes.size(); ++i) {
  1455. auto value = ArgNodes[i]->GetValue(compCtx);
  1456. args.push_back(value);
  1457. }
  1458. return compCtx.HolderFactory.Create<TListValue>(compCtx, Name, std::move(args), ArgDesc, RetTypeDesc, ProcDesc, &FInfo, StructType, StructTypeDesc, compCtx.HolderFactory);
  1459. }
  1460. };
  1461. class TPgToRecord : public TMutableComputationNode<TPgToRecord> {
  1462. typedef TMutableComputationNode<TPgToRecord> TBaseComputation;
  1463. public:
  1464. TPgToRecord(
  1465. TComputationMutables& mutables,
  1466. IComputationNode* arg,
  1467. TStructType* structType,
  1468. TVector<std::pair<TString,TString>>&& members
  1469. )
  1470. : TBaseComputation(mutables)
  1471. , Arg(arg)
  1472. , StructType(structType)
  1473. , Members(std::move(members))
  1474. , StateIndex(mutables.CurValueIndex++)
  1475. {
  1476. StructIndicies.resize(Members.size());
  1477. FieldTypes.resize(Members.size());
  1478. for (ui32 i = 0; i < Members.size(); ++i) {
  1479. StructIndicies[i] = structType->GetMemberIndex(Members[i].second);
  1480. auto itemType = structType->GetMemberType(StructIndicies[i]);
  1481. ui32 oid;
  1482. if (itemType->IsNull()) {
  1483. oid = UNKNOWNOID;
  1484. } else {
  1485. oid = AS_TYPE(TPgType, itemType)->GetTypeId();
  1486. }
  1487. FieldTypes[i] = &NPg::LookupType(oid);
  1488. }
  1489. }
  1490. NUdf::TUnboxedValuePod DoCalculate(TComputationContext& compCtx) const {
  1491. auto input = Arg->GetValue(compCtx);
  1492. auto& state = GetState(compCtx);
  1493. auto elements = input.GetElements();
  1494. TVector<NUdf::TUnboxedValue> elemValues;
  1495. if (!elements) {
  1496. elemValues.reserve(StructType->GetMembersCount());
  1497. for (ui32 i = 0; i < StructType->GetMembersCount(); ++i) {
  1498. elemValues.push_back(input.GetElement(i));
  1499. }
  1500. elements = elemValues.data();
  1501. }
  1502. for (ui32 i = 0; i < Members.size(); ++i) {
  1503. auto index = StructIndicies[i];
  1504. if (!elements[index]) {
  1505. state.Nulls[i] = true;
  1506. } else {
  1507. state.Nulls[i] = false;
  1508. if (FieldTypes[i]->PassByValue) {
  1509. state.Values[i] = ScalarDatumFromPod(elements[index]);
  1510. } else {
  1511. state.Values[i] = PointerDatumFromPod(elements[index]);
  1512. }
  1513. }
  1514. }
  1515. HeapTuple tuple = heap_form_tuple(state.Desc, state.Values.get(), state.Nulls.get());
  1516. auto result = (HeapTupleHeader) palloc(tuple->t_len);
  1517. memcpy(result, tuple->t_data, tuple->t_len);
  1518. heap_freetuple(tuple);
  1519. return PointerDatumToPod((Datum)result);
  1520. }
  1521. private:
  1522. void RegisterDependencies() const final {
  1523. DependsOn(Arg);
  1524. }
  1525. struct TPgToRecordState : public TComputationValue<TPgToRecordState> {
  1526. TPgToRecordState(TMemoryUsageInfo* memInfo, const TVector<std::pair<TString,TString>>& members,
  1527. const TVector<const NPg::TTypeDesc*>& fieldTypes)
  1528. : TComputationValue(memInfo)
  1529. {
  1530. Values.reset(new Datum[members.size()]);
  1531. Nulls.reset(new bool[members.size()]);
  1532. Desc = CreateTemplateTupleDesc(members.size());
  1533. for (ui32 i = 0; i < members.size(); ++i) {
  1534. TupleDescInitEntry(Desc, (AttrNumber) 1 + i, members[i].first.c_str(), fieldTypes[i]->TypeId, -1, 0);
  1535. }
  1536. Desc = BlessTupleDesc(Desc);
  1537. }
  1538. ~TPgToRecordState()
  1539. {
  1540. FreeTupleDesc(Desc);
  1541. }
  1542. std::unique_ptr<Datum[]> Values;
  1543. std::unique_ptr<bool[]> Nulls;
  1544. TupleDesc Desc;
  1545. };
  1546. TPgToRecordState& GetState(TComputationContext& compCtx) const {
  1547. auto& result = compCtx.MutableValues[StateIndex];
  1548. if (!result.HasValue()) {
  1549. result = compCtx.HolderFactory.Create<TPgToRecordState>(Members, FieldTypes);
  1550. }
  1551. return *static_cast<TPgToRecordState*>(result.AsBoxed().Get());
  1552. }
  1553. IComputationNode* const Arg;
  1554. TStructType* const StructType;
  1555. const TVector<std::pair<TString,TString>> Members;
  1556. const ui32 StateIndex;
  1557. TVector<ui32> StructIndicies;
  1558. TVector<const NPg::TTypeDesc*> FieldTypes;
  1559. };
  1560. class TPgCast : public TMutableComputationNode<TPgCast> {
  1561. typedef TMutableComputationNode<TPgCast> TBaseComputation;
  1562. public:
  1563. TPgCast(TComputationMutables& mutables, ui32 sourceId, ui32 targetId, IComputationNode* arg, IComputationNode* typeMod)
  1564. : TBaseComputation(mutables)
  1565. , StateIndex(mutables.CurValueIndex++)
  1566. , SourceId(sourceId)
  1567. , TargetId(targetId)
  1568. , Arg(arg)
  1569. , TypeMod(typeMod)
  1570. , SourceTypeDesc(SourceId ? NPg::LookupType(SourceId) : NPg::TTypeDesc())
  1571. , TargetTypeDesc(NPg::LookupType(targetId))
  1572. , IsSourceArray(SourceId && SourceTypeDesc.TypeId == SourceTypeDesc.ArrayTypeId)
  1573. , IsTargetArray(TargetTypeDesc.TypeId == TargetTypeDesc.ArrayTypeId)
  1574. , SourceElemDesc(SourceId ? NPg::LookupType(IsSourceArray ? SourceTypeDesc.ElementTypeId : SourceTypeDesc.TypeId) : NPg::TTypeDesc())
  1575. , TargetElemDesc(NPg::LookupType(IsTargetArray ? TargetTypeDesc.ElementTypeId : TargetTypeDesc.TypeId))
  1576. {
  1577. TypeIOParam = MakeTypeIOParam(TargetTypeDesc);
  1578. Zero(FInfo1);
  1579. Zero(FInfo2);
  1580. if (TypeMod && SourceId == TargetId && NPg::HasCast(TargetElemDesc.TypeId, TargetElemDesc.TypeId)) {
  1581. const auto& cast = NPg::LookupCast(TargetElemDesc.TypeId, TargetElemDesc.TypeId);
  1582. Y_ENSURE(cast.FunctionId);
  1583. GetPgFuncAddr(cast.FunctionId, FInfo1);
  1584. Y_ENSURE(!FInfo1.fn_retset);
  1585. Y_ENSURE(FInfo1.fn_addr);
  1586. Y_ENSURE(FInfo1.fn_nargs >= 2 && FInfo1.fn_nargs <= 3);
  1587. ConvertLength = true;
  1588. ArrayCast = IsSourceArray;
  1589. return;
  1590. }
  1591. if (SourceId == 0 || SourceId == TargetId) {
  1592. return;
  1593. }
  1594. ui32 funcId;
  1595. ui32 funcId2 = 0;
  1596. if (!NPg::HasCast(SourceElemDesc.TypeId, TargetElemDesc.TypeId) || (IsSourceArray != IsTargetArray)) {
  1597. ArrayCast = IsSourceArray && IsTargetArray;
  1598. if (IsSourceArray && !IsTargetArray) {
  1599. Y_ENSURE(TargetTypeDesc.Category == 'S' || TargetId == UNKNOWNOID);
  1600. funcId = NPg::LookupProc("array_out", { 0 }).ProcId;
  1601. } else if (IsTargetArray && !IsSourceArray) {
  1602. Y_ENSURE(SourceElemDesc.Category == 'S' || SourceId == UNKNOWNOID);
  1603. funcId = NPg::LookupProc("array_in", { 0,0,0 }).ProcId;
  1604. } else if (SourceElemDesc.Category == 'S' || SourceId == UNKNOWNOID) {
  1605. funcId = TargetElemDesc.InFuncId;
  1606. } else {
  1607. Y_ENSURE(TargetTypeDesc.Category == 'S' || TargetId == UNKNOWNOID);
  1608. funcId = SourceElemDesc.OutFuncId;
  1609. }
  1610. } else {
  1611. Y_ENSURE(IsSourceArray == IsTargetArray);
  1612. ArrayCast = IsSourceArray;
  1613. const auto& cast = NPg::LookupCast(SourceElemDesc.TypeId, TargetElemDesc.TypeId);
  1614. switch (cast.Method) {
  1615. case NPg::ECastMethod::Binary:
  1616. return;
  1617. case NPg::ECastMethod::Function: {
  1618. Y_ENSURE(cast.FunctionId);
  1619. funcId = cast.FunctionId;
  1620. break;
  1621. }
  1622. case NPg::ECastMethod::InOut: {
  1623. funcId = SourceElemDesc.OutFuncId;
  1624. funcId2 = TargetElemDesc.InFuncId;
  1625. break;
  1626. }
  1627. }
  1628. }
  1629. Y_ENSURE(funcId);
  1630. GetPgFuncAddr(funcId, FInfo1);
  1631. Y_ENSURE(!FInfo1.fn_retset);
  1632. Y_ENSURE(FInfo1.fn_addr);
  1633. Y_ENSURE(FInfo1.fn_nargs >= 1 && FInfo1.fn_nargs <= 3);
  1634. Func1Lookup = NPg::LookupProc(funcId);
  1635. Y_ENSURE(Func1Lookup.ArgTypes.size() >= 1 && Func1Lookup.ArgTypes.size() <= 3);
  1636. if (NPg::LookupType(Func1Lookup.ArgTypes[0]).TypeLen == -2 && SourceElemDesc.Category == 'S') {
  1637. ConvertArgToCString = true;
  1638. }
  1639. if (funcId2) {
  1640. Y_ENSURE(funcId2);
  1641. GetPgFuncAddr(funcId2, FInfo2);
  1642. Y_ENSURE(!FInfo2.fn_retset);
  1643. Y_ENSURE(FInfo2.fn_addr);
  1644. Y_ENSURE(FInfo2.fn_nargs == 1);
  1645. Func2Lookup = NPg::LookupProc(funcId2);
  1646. Y_ENSURE(Func2Lookup.ArgTypes.size() == 1);
  1647. }
  1648. if (!funcId2) {
  1649. if (NPg::LookupType(Func1Lookup.ResultType).TypeLen == -2 && TargetElemDesc.Category == 'S') {
  1650. ConvertResFromCString = true;
  1651. }
  1652. } else {
  1653. const auto& Func2ArgType = NPg::LookupType(Func2Lookup.ArgTypes[0]);
  1654. if (NPg::LookupType(Func1Lookup.ResultType).TypeLen == -2 && Func2ArgType.Category == 'S') {
  1655. ConvertResFromCString = true;
  1656. }
  1657. if (NPg::LookupType(Func2Lookup.ResultType).TypeLen == -2 && TargetElemDesc.Category == 'S') {
  1658. ConvertResFromCString2 = true;
  1659. }
  1660. }
  1661. }
  1662. NUdf::TUnboxedValuePod DoCalculate(TComputationContext& compCtx) const {
  1663. auto value = Arg->GetValue(compCtx);
  1664. if (!value) {
  1665. return value.Release();
  1666. }
  1667. i32 typeMod = -1;
  1668. if (TypeMod) {
  1669. typeMod = DatumGetInt32(ScalarDatumFromPod(TypeMod->GetValue(compCtx)));
  1670. }
  1671. if (!FInfo1.fn_addr) {
  1672. // binary compatible
  1673. if (!ArrayCast) {
  1674. return value.Release();
  1675. } else {
  1676. // clone array with new target type in the header
  1677. auto datum = PointerDatumFromPod(value);
  1678. ArrayType* arr = DatumGetArrayTypePCopy(datum);
  1679. ARR_ELEMTYPE(arr) = TargetElemDesc.TypeId;
  1680. return PointerDatumToPod(PointerGetDatum(arr));
  1681. }
  1682. }
  1683. TPAllocScope call;
  1684. auto& state = GetState(compCtx);
  1685. if (ArrayCast) {
  1686. auto arr = (ArrayType*)DatumGetPointer(PointerDatumFromPod(value));
  1687. auto ndim = ARR_NDIM(arr);
  1688. auto dims = ARR_DIMS(arr);
  1689. auto lb = ARR_LBOUND(arr);
  1690. auto nitems = ArrayGetNItems(ndim, dims);
  1691. Datum* elems = (Datum*)TWithDefaultMiniKQLAlloc::AllocWithSize(nitems * sizeof(Datum));
  1692. Y_DEFER {
  1693. TWithDefaultMiniKQLAlloc::FreeWithSize(elems, nitems * sizeof(Datum));
  1694. };
  1695. bool* nulls = (bool*)TWithDefaultMiniKQLAlloc::AllocWithSize(nitems);
  1696. Y_DEFER {
  1697. TWithDefaultMiniKQLAlloc::FreeWithSize(nulls, nitems);
  1698. };
  1699. array_iter iter;
  1700. array_iter_setup(&iter, (AnyArrayType*)arr);
  1701. for (ui32 i = 0; i < nitems; ++i) {
  1702. bool isNull;
  1703. auto datum = array_iter_next(&iter, &isNull, i, SourceElemDesc.TypeLen,
  1704. SourceElemDesc.PassByValue, SourceElemDesc.TypeAlign);
  1705. if (isNull) {
  1706. nulls[i] = true;
  1707. continue;
  1708. } else {
  1709. nulls[i] = false;
  1710. elems[i] = ConvertDatum(datum, state, typeMod);
  1711. }
  1712. }
  1713. auto ret = construct_md_array(elems, nulls, ndim, dims, lb, TargetElemDesc.TypeId,
  1714. TargetElemDesc.TypeLen, TargetElemDesc.PassByValue, TargetElemDesc.TypeAlign);
  1715. return PointerDatumToPod(PointerGetDatum(ret));
  1716. } else {
  1717. auto datum = SourceTypeDesc.PassByValue ?
  1718. ScalarDatumFromPod(value) :
  1719. PointerDatumFromPod(value);
  1720. auto ret = ConvertDatum(datum, state, typeMod);
  1721. return AnyDatumToPod(ret, TargetTypeDesc.PassByValue);
  1722. }
  1723. }
  1724. private:
  1725. void RegisterDependencies() const final {
  1726. DependsOn(Arg);
  1727. if (TypeMod) {
  1728. DependsOn(TypeMod);
  1729. }
  1730. }
  1731. struct TState : public TComputationValue<TState> {
  1732. TState(TMemoryUsageInfo* memInfo, const FmgrInfo* finfo1, const FmgrInfo* finfo2)
  1733. : TComputationValue(memInfo)
  1734. , CallInfo1(3, finfo1)
  1735. , CallInfo2(1, finfo2)
  1736. {
  1737. }
  1738. TFunctionCallInfo CallInfo1, CallInfo2;
  1739. };
  1740. TState& GetState(TComputationContext& compCtx) const {
  1741. auto& result = compCtx.MutableValues[StateIndex];
  1742. if (!result.HasValue()) {
  1743. result = compCtx.HolderFactory.Create<TState>(&FInfo1, &FInfo2);
  1744. }
  1745. return *static_cast<TState*>(result.AsBoxed().Get());
  1746. }
  1747. Datum ConvertDatum(Datum datum, TState& state, i32 typeMod) const {
  1748. auto& callInfo1 = state.CallInfo1.Ref();
  1749. callInfo1.isnull = false;
  1750. NullableDatum argDatum = { datum, false };
  1751. void* freeCString = nullptr;
  1752. Y_DEFER {
  1753. if (freeCString) {
  1754. pfree(freeCString);
  1755. }
  1756. };
  1757. if (ConvertArgToCString) {
  1758. argDatum.value = (Datum)MakeCString(GetVarBuf((const text*)argDatum.value));
  1759. freeCString = (void*)argDatum.value;
  1760. }
  1761. callInfo1.args[0] = argDatum;
  1762. if (ConvertLength) {
  1763. callInfo1.args[1] = { Int32GetDatum(typeMod), false };
  1764. callInfo1.args[2] = { BoolGetDatum(true), false };
  1765. } else {
  1766. if (FInfo1.fn_nargs == 2) {
  1767. callInfo1.args[1] = { Int32GetDatum(typeMod), false };
  1768. } else {
  1769. callInfo1.args[1] = { ObjectIdGetDatum(TypeIOParam), false };
  1770. callInfo1.args[2] = { Int32GetDatum(typeMod), false };
  1771. }
  1772. }
  1773. void* freeMem = nullptr;
  1774. void* freeMem2 = nullptr;
  1775. Y_DEFER {
  1776. if (freeMem) {
  1777. pfree(freeMem);
  1778. }
  1779. if (freeMem2) {
  1780. pfree(freeMem2);
  1781. }
  1782. };
  1783. {
  1784. auto ret = FInfo1.fn_addr(&callInfo1);
  1785. Y_ENSURE(!callInfo1.isnull);
  1786. if (ConvertResFromCString) {
  1787. freeMem = (void*)ret;
  1788. ret = (Datum)MakeVar((const char*)ret);
  1789. }
  1790. if (FInfo2.fn_addr) {
  1791. auto& callInfo2 = state.CallInfo1.Ref();
  1792. callInfo2.isnull = false;
  1793. NullableDatum argDatum2 = { ret, false };
  1794. callInfo2.args[0] = argDatum2;
  1795. auto ret2 = FInfo2.fn_addr(&callInfo2);
  1796. pfree((void*)ret);
  1797. Y_ENSURE(!callInfo2.isnull);
  1798. ret = ret2;
  1799. }
  1800. if (ConvertResFromCString2) {
  1801. freeMem2 = (void*)ret;
  1802. ret = (Datum)MakeVar((const char*)ret);
  1803. }
  1804. return ret;
  1805. }
  1806. }
  1807. const ui32 StateIndex;
  1808. const ui32 SourceId;
  1809. const ui32 TargetId;
  1810. IComputationNode* const Arg;
  1811. IComputationNode* const TypeMod;
  1812. const NPg::TTypeDesc SourceTypeDesc;
  1813. const NPg::TTypeDesc TargetTypeDesc;
  1814. const bool IsSourceArray;
  1815. const bool IsTargetArray;
  1816. const NPg::TTypeDesc SourceElemDesc;
  1817. const NPg::TTypeDesc TargetElemDesc;
  1818. FmgrInfo FInfo1, FInfo2;
  1819. NPg::TProcDesc Func1Lookup, Func2Lookup;
  1820. bool ConvertArgToCString = false;
  1821. bool ConvertResFromCString = false;
  1822. bool ConvertResFromCString2 = false;
  1823. ui32 TypeIOParam = 0;
  1824. bool ArrayCast = false;
  1825. bool ConvertLength = false;
  1826. };
  1827. const i32 PgDateShift = UNIX_EPOCH_JDATE - POSTGRES_EPOCH_JDATE;
  1828. const i64 PgTimestampShift = USECS_PER_DAY * (UNIX_EPOCH_JDATE - POSTGRES_EPOCH_JDATE);
  1829. inline i32 Date2Pg(i32 value) {
  1830. return value + PgDateShift;
  1831. }
  1832. inline i64 Timestamp2Pg(i64 value) {
  1833. return value + PgTimestampShift;
  1834. }
  1835. inline Interval* Interval2Pg(i64 value) {
  1836. auto ret = (Interval*)palloc(sizeof(Interval));
  1837. ret->time = value % 86400000000ll;
  1838. ret->day = value / 86400000000ll;
  1839. ret->month = 0;
  1840. return ret;
  1841. }
  1842. template <NUdf::EDataSlot Slot>
  1843. NUdf::TUnboxedValuePod ConvertToPgValue(NUdf::TUnboxedValuePod value, TMaybe<NUdf::EDataSlot> actualSlot = {}) {
  1844. #ifndef NDEBUG
  1845. // todo: improve checks
  1846. if (actualSlot && Slot != *actualSlot) {
  1847. throw yexception() << "Invalid data slot in ConvertToPgValue, expected " << Slot << ", but actual: " << *actualSlot;
  1848. }
  1849. #else
  1850. Y_UNUSED(actualSlot);
  1851. #endif
  1852. switch (Slot) {
  1853. case NUdf::EDataSlot::Bool:
  1854. return ScalarDatumToPod(BoolGetDatum(value.Get<bool>()));
  1855. case NUdf::EDataSlot::Int8:
  1856. return ScalarDatumToPod(Int16GetDatum(value.Get<i8>()));
  1857. case NUdf::EDataSlot::Uint8:
  1858. return ScalarDatumToPod(Int16GetDatum(value.Get<ui8>()));
  1859. case NUdf::EDataSlot::Int16:
  1860. return ScalarDatumToPod(Int16GetDatum(value.Get<i16>()));
  1861. case NUdf::EDataSlot::Uint16:
  1862. return ScalarDatumToPod(Int32GetDatum(value.Get<ui16>()));
  1863. case NUdf::EDataSlot::Int32:
  1864. return ScalarDatumToPod(Int32GetDatum(value.Get<i32>()));
  1865. case NUdf::EDataSlot::Uint32:
  1866. return ScalarDatumToPod(Int64GetDatum(value.Get<ui32>()));
  1867. case NUdf::EDataSlot::Int64:
  1868. return ScalarDatumToPod(Int64GetDatum(value.Get<i64>()));
  1869. case NUdf::EDataSlot::Uint64:
  1870. return PointerDatumToPod(NumericGetDatum(Uint64ToPgNumeric(value.Get<ui64>())));
  1871. case NUdf::EDataSlot::DyNumber:
  1872. return PointerDatumToPod(NumericGetDatum(DyNumberToPgNumeric(value)));
  1873. case NUdf::EDataSlot::Float:
  1874. return ScalarDatumToPod(Float4GetDatum(value.Get<float>()));
  1875. case NUdf::EDataSlot::Double:
  1876. return ScalarDatumToPod(Float8GetDatum(value.Get<double>()));
  1877. case NUdf::EDataSlot::String:
  1878. case NUdf::EDataSlot::Yson:
  1879. case NUdf::EDataSlot::Utf8: {
  1880. const auto& ref = value.AsStringRef();
  1881. return PointerDatumToPod((Datum)MakeVar(ref));
  1882. }
  1883. case NUdf::EDataSlot::Date: {
  1884. auto res = Date2Pg(value.Get<ui16>());
  1885. return ScalarDatumToPod(res);
  1886. }
  1887. case NUdf::EDataSlot::Datetime: {
  1888. auto res = Timestamp2Pg(value.Get<ui32>() * 1000000ull);
  1889. return ScalarDatumToPod(res);
  1890. }
  1891. case NUdf::EDataSlot::Timestamp: {
  1892. auto res = Timestamp2Pg(value.Get<ui64>());
  1893. return ScalarDatumToPod(res);
  1894. }
  1895. case NUdf::EDataSlot::Interval:
  1896. case NUdf::EDataSlot::Interval64: {
  1897. auto res = Interval2Pg(value.Get<i64>());
  1898. return PointerDatumToPod(PointerGetDatum(res));
  1899. }
  1900. case NUdf::EDataSlot::Date32: {
  1901. auto res = Date2Pg(value.Get<i32>());
  1902. return ScalarDatumToPod(res);
  1903. }
  1904. case NUdf::EDataSlot::Datetime64: {
  1905. auto res = Timestamp2Pg(value.Get<i64>() * 1000000ull);
  1906. return ScalarDatumToPod(res);
  1907. }
  1908. case NUdf::EDataSlot::Timestamp64: {
  1909. auto res = Timestamp2Pg(value.Get<i64>());
  1910. return ScalarDatumToPod(res);
  1911. }
  1912. case NUdf::EDataSlot::Json: {
  1913. auto input = MakeCString(value.AsStringRef());
  1914. auto res = DirectFunctionCall1Coll(json_in, DEFAULT_COLLATION_OID, PointerGetDatum(input));
  1915. pfree(input);
  1916. return PointerDatumToPod(PointerGetDatum((void*)res));
  1917. }
  1918. case NUdf::EDataSlot::JsonDocument: {
  1919. auto str = NKikimr::NBinaryJson::SerializeToJson(value.AsStringRef());
  1920. auto res = (text*)DirectFunctionCall1Coll(jsonb_in, DEFAULT_COLLATION_OID, PointerGetDatum(str.c_str()));
  1921. return PointerDatumToPod(PointerGetDatum(res));
  1922. }
  1923. case NUdf::EDataSlot::Uuid: {
  1924. TString str;
  1925. str.reserve(36);
  1926. ui16 dw[8];
  1927. std::memcpy(dw, value.AsStringRef().Data(), sizeof(dw));
  1928. TStringOutput out(str);
  1929. NKikimr::NUuid::UuidToString(dw, out);
  1930. auto res = DirectFunctionCall1Coll(uuid_in, DEFAULT_COLLATION_OID, PointerGetDatum(str.c_str()));
  1931. return PointerDatumToPod(PointerGetDatum((void*)res));
  1932. }
  1933. case NUdf::EDataSlot::TzDate:
  1934. case NUdf::EDataSlot::TzDatetime:
  1935. case NUdf::EDataSlot::TzTimestamp:
  1936. case NUdf::EDataSlot::TzDate32:
  1937. case NUdf::EDataSlot::TzDatetime64:
  1938. case NUdf::EDataSlot::TzTimestamp64: {
  1939. NUdf::TUnboxedValue str = ValueToString(Slot, value);
  1940. return PointerDatumToPod(PointerGetDatum(MakeVar(str.AsStringRef())));
  1941. }
  1942. default:
  1943. ythrow yexception() << "Unexpected data slot in ConvertToPgValue: " << Slot;
  1944. }
  1945. }
  1946. template <NUdf::EDataSlot Slot, bool IsCString>
  1947. NUdf::TUnboxedValuePod ConvertFromPgValue(NUdf::TUnboxedValuePod value, TMaybe<NUdf::EDataSlot> actualSlot = {}) {
  1948. #ifndef NDEBUG
  1949. // todo: improve checks
  1950. if (actualSlot && Slot != *actualSlot) {
  1951. throw yexception() << "Invalid data slot in ConvertFromPgValue, expected " << Slot << ", but actual: " << *actualSlot;
  1952. }
  1953. #else
  1954. Y_UNUSED(actualSlot);
  1955. #endif
  1956. switch (Slot) {
  1957. case NUdf::EDataSlot::Bool:
  1958. return NUdf::TUnboxedValuePod((bool)DatumGetBool(ScalarDatumFromPod(value)));
  1959. case NUdf::EDataSlot::Int16:
  1960. return NUdf::TUnboxedValuePod((i16)DatumGetInt16(ScalarDatumFromPod(value)));
  1961. case NUdf::EDataSlot::Int32:
  1962. return NUdf::TUnboxedValuePod((i32)DatumGetInt32(ScalarDatumFromPod(value)));
  1963. case NUdf::EDataSlot::Int64:
  1964. return NUdf::TUnboxedValuePod((i64)DatumGetInt64(ScalarDatumFromPod(value)));
  1965. case NUdf::EDataSlot::Float:
  1966. return NUdf::TUnboxedValuePod((float)DatumGetFloat4(ScalarDatumFromPod(value)));
  1967. case NUdf::EDataSlot::Double:
  1968. return NUdf::TUnboxedValuePod((double)DatumGetFloat8(ScalarDatumFromPod(value)));
  1969. case NUdf::EDataSlot::String:
  1970. case NUdf::EDataSlot::Utf8:
  1971. if (IsCString) {
  1972. auto x = (const char*)PointerDatumFromPod(value);
  1973. return MakeString(TStringBuf(x));
  1974. } else {
  1975. auto x = (const text*)PointerDatumFromPod(value);
  1976. return MakeString(GetVarBuf(x));
  1977. }
  1978. case NUdf::EDataSlot::Date32: {
  1979. auto res = (i32)DatumGetInt32(ScalarDatumFromPod(value)) - PgDateShift;
  1980. if (res < NUdf::MIN_DATE32 || res > NUdf::MAX_DATE32) {
  1981. return NUdf::TUnboxedValuePod();
  1982. }
  1983. return NUdf::TUnboxedValuePod(res);
  1984. }
  1985. case NUdf::EDataSlot::Timestamp64: {
  1986. auto res = (i64)DatumGetInt64(ScalarDatumFromPod(value)) - PgTimestampShift;
  1987. if (res < NUdf::MIN_TIMESTAMP64 || res > NUdf::MAX_TIMESTAMP64) {
  1988. return NUdf::TUnboxedValuePod();
  1989. }
  1990. return NUdf::TUnboxedValuePod(res);
  1991. }
  1992. case NUdf::EDataSlot::Uuid: {
  1993. auto str = (char*)DirectFunctionCall1Coll(uuid_out, DEFAULT_COLLATION_OID, PointerDatumFromPod(value));
  1994. auto res = ParseUuid(NUdf::TStringRef(TStringBuf(str)));
  1995. pfree(str);
  1996. return res;
  1997. }
  1998. default:
  1999. ythrow yexception() << "Unexpected data slot in ConvertFromPgValue: " << Slot;
  2000. }
  2001. }
  2002. NUdf::TUnboxedValuePod ConvertFromPgValue(NUdf::TUnboxedValuePod source, ui32 sourceTypeId, NKikimr::NMiniKQL::TType* targetType) {
  2003. TMaybe<NUdf::EDataSlot> targetDataTypeSlot;
  2004. #ifndef NDEBUG
  2005. bool isOptional = false;
  2006. auto targetDataType = UnpackOptionalData(targetType, isOptional);
  2007. YQL_ENSURE(targetDataType);
  2008. targetDataTypeSlot = targetDataType->GetDataSlot();
  2009. if (!source && !isOptional) {
  2010. throw yexception() << "Null value is not allowed for non-optional data type " << *targetType;
  2011. }
  2012. #else
  2013. Y_UNUSED(targetType);
  2014. #endif
  2015. if (!source) {
  2016. return source;
  2017. }
  2018. switch (sourceTypeId) {
  2019. case BOOLOID:
  2020. return ConvertFromPgValue<NUdf::EDataSlot::Bool, false>(source, targetDataTypeSlot);
  2021. case INT2OID:
  2022. return ConvertFromPgValue<NUdf::EDataSlot::Int16, false>(source, targetDataTypeSlot);
  2023. case INT4OID:
  2024. return ConvertFromPgValue<NUdf::EDataSlot::Int32, false>(source, targetDataTypeSlot);
  2025. case INT8OID:
  2026. return ConvertFromPgValue<NUdf::EDataSlot::Int64, false>(source, targetDataTypeSlot);
  2027. case FLOAT4OID:
  2028. return ConvertFromPgValue<NUdf::EDataSlot::Float, false>(source, targetDataTypeSlot);
  2029. case FLOAT8OID:
  2030. return ConvertFromPgValue<NUdf::EDataSlot::Double, false>(source, targetDataTypeSlot);
  2031. case TEXTOID:
  2032. case VARCHAROID:
  2033. return ConvertFromPgValue<NUdf::EDataSlot::Utf8, false>(source, targetDataTypeSlot);
  2034. case BYTEAOID:
  2035. return ConvertFromPgValue<NUdf::EDataSlot::String, false>(source, targetDataTypeSlot);
  2036. case CSTRINGOID:
  2037. return ConvertFromPgValue<NUdf::EDataSlot::Utf8, true>(source, targetDataTypeSlot);
  2038. default:
  2039. ythrow yexception() << "Unsupported type: " << NPg::LookupType(sourceTypeId).Name;
  2040. }
  2041. }
  2042. NUdf::TUnboxedValuePod ConvertToPgValue(NUdf::TUnboxedValuePod source, NKikimr::NMiniKQL::TType* sourceType, ui32 targetTypeId) {
  2043. TMaybe<NUdf::EDataSlot> sourceDataTypeSlot;
  2044. #ifndef NDEBUG
  2045. bool isOptional = false;
  2046. auto sourceDataType = UnpackOptionalData(sourceType, isOptional);
  2047. YQL_ENSURE(sourceDataType);
  2048. sourceDataTypeSlot = sourceDataType->GetDataSlot();
  2049. if (!source && !isOptional) {
  2050. throw yexception() << "Null value is not allowed for non-optional data type " << *sourceType;
  2051. }
  2052. #else
  2053. Y_UNUSED(sourceType);
  2054. #endif
  2055. if (!source) {
  2056. return source;
  2057. }
  2058. switch (targetTypeId) {
  2059. case BOOLOID:
  2060. return ConvertToPgValue<NUdf::EDataSlot::Bool>(source, sourceDataTypeSlot);
  2061. case INT2OID:
  2062. return ConvertToPgValue<NUdf::EDataSlot::Int16>(source, sourceDataTypeSlot);
  2063. case INT4OID:
  2064. return ConvertToPgValue<NUdf::EDataSlot::Int32>(source, sourceDataTypeSlot);
  2065. case INT8OID:
  2066. return ConvertToPgValue<NUdf::EDataSlot::Int64>(source, sourceDataTypeSlot);
  2067. case FLOAT4OID:
  2068. return ConvertToPgValue<NUdf::EDataSlot::Float>(source, sourceDataTypeSlot);
  2069. case FLOAT8OID:
  2070. return ConvertToPgValue<NUdf::EDataSlot::Double>(source, sourceDataTypeSlot);
  2071. case TEXTOID:
  2072. return ConvertToPgValue<NUdf::EDataSlot::Utf8>(source, sourceDataTypeSlot);
  2073. case BYTEAOID:
  2074. return ConvertToPgValue<NUdf::EDataSlot::String>(source, sourceDataTypeSlot);
  2075. default:
  2076. ythrow yexception() << "Unsupported type: " << NPg::LookupType(targetTypeId).Name;
  2077. }
  2078. }
  2079. template <NUdf::EDataSlot Slot, bool IsCString>
  2080. class TFromPg : public TMutableComputationNode<TFromPg<Slot, IsCString>> {
  2081. typedef TMutableComputationNode<TFromPg<Slot, IsCString>> TBaseComputation;
  2082. public:
  2083. TFromPg(TComputationMutables& mutables, IComputationNode* arg)
  2084. : TBaseComputation(mutables)
  2085. , Arg(arg)
  2086. {
  2087. }
  2088. NUdf::TUnboxedValuePod DoCalculate(TComputationContext& compCtx) const {
  2089. auto value = Arg->GetValue(compCtx);
  2090. if (!value) {
  2091. return value.Release();
  2092. }
  2093. return ConvertFromPgValue<Slot, IsCString>(value);
  2094. }
  2095. private:
  2096. void RegisterDependencies() const final {
  2097. this->DependsOn(Arg);
  2098. }
  2099. IComputationNode* const Arg;
  2100. };
  2101. template <NUdf::EDataSlot Slot>
  2102. class TToPg : public TMutableComputationNode<TToPg<Slot>> {
  2103. typedef TMutableComputationNode<TToPg<Slot>> TBaseComputation;
  2104. public:
  2105. TToPg(TComputationMutables& mutables, IComputationNode* arg, TDataType* argType)
  2106. : TBaseComputation(mutables)
  2107. , Arg(arg)
  2108. , ArgType(argType)
  2109. {
  2110. }
  2111. NUdf::TUnboxedValuePod DoCalculate(TComputationContext& compCtx) const {
  2112. auto value = Arg->GetValue(compCtx);
  2113. if (!value) {
  2114. return value.Release();
  2115. }
  2116. if constexpr (Slot == NUdf::EDataSlot::Decimal) {
  2117. auto decimalType = static_cast<TDataDecimalType*>(ArgType);
  2118. return PointerDatumToPod(NumericGetDatum(DecimalToPgNumeric(value,
  2119. decimalType->GetParams().first, decimalType->GetParams().second)));
  2120. } else {
  2121. return ConvertToPgValue<Slot>(value);
  2122. }
  2123. }
  2124. private:
  2125. void RegisterDependencies() const final {
  2126. this->DependsOn(Arg);
  2127. }
  2128. IComputationNode* const Arg;
  2129. TDataType* ArgType;
  2130. };
  2131. class TPgArray : public TMutableComputationNode<TPgArray> {
  2132. typedef TMutableComputationNode<TPgArray> TBaseComputation;
  2133. public:
  2134. TPgArray(TComputationMutables& mutables, TComputationNodePtrVector&& argNodes, const TVector<TType*>&& argTypes, ui32 arrayType)
  2135. : TBaseComputation(mutables)
  2136. , ArgNodes(std::move(argNodes))
  2137. , ArgTypes(std::move(argTypes))
  2138. , ArrayTypeDesc(NPg::LookupType(arrayType))
  2139. , ElemTypeDesc(NPg::LookupType(ArrayTypeDesc.ElementTypeId))
  2140. {
  2141. ArgDescs.resize(ArgNodes.size());
  2142. for (ui32 i = 0; i < ArgNodes.size(); ++i) {
  2143. if (!ArgTypes[i]->IsNull()) {
  2144. auto type = static_cast<TPgType*>(ArgTypes[i])->GetTypeId();
  2145. ArgDescs[i] = NPg::LookupType(type);
  2146. if (ArgDescs[i].TypeId == ArgDescs[i].ArrayTypeId) {
  2147. MultiDims = true;
  2148. }
  2149. }
  2150. }
  2151. }
  2152. NUdf::TUnboxedValuePod DoCalculate(TComputationContext& compCtx) const {
  2153. TUnboxedValueVector args;
  2154. ui32 nelems = ArgNodes.size();
  2155. args.reserve(nelems);
  2156. for (ui32 i = 0; i < nelems; ++i) {
  2157. auto value = ArgNodes[i]->GetValue(compCtx);
  2158. args.push_back(value);
  2159. }
  2160. Datum* dvalues = (Datum*)TWithDefaultMiniKQLAlloc::AllocWithSize(nelems * sizeof(Datum));
  2161. Y_DEFER {
  2162. TWithDefaultMiniKQLAlloc::FreeWithSize(dvalues, nelems * sizeof(Datum));
  2163. };
  2164. bool *dnulls = (bool*)TWithDefaultMiniKQLAlloc::AllocWithSize(nelems);
  2165. Y_DEFER {
  2166. TWithDefaultMiniKQLAlloc::FreeWithSize(dnulls, nelems);
  2167. };
  2168. TPAllocScope call;
  2169. for (ui32 i = 0; i < nelems; ++i) {
  2170. const auto& value = args[i];
  2171. if (value) {
  2172. dnulls[i] = false;
  2173. dvalues[i] = ArgDescs[i].PassByValue ?
  2174. ScalarDatumFromPod(value) :
  2175. PointerDatumFromPod(value);
  2176. } else {
  2177. dnulls[i] = true;
  2178. }
  2179. }
  2180. {
  2181. int ndims = 0;
  2182. int dims[MAXDIM];
  2183. int lbs[MAXDIM];
  2184. if (!MultiDims) {
  2185. // 1D array
  2186. ndims = 1;
  2187. dims[0] = nelems;
  2188. lbs[0] = 1;
  2189. auto result = construct_md_array(dvalues, dnulls, ndims, dims, lbs,
  2190. ElemTypeDesc.TypeId,
  2191. ElemTypeDesc.TypeLen,
  2192. ElemTypeDesc.PassByValue,
  2193. ElemTypeDesc.TypeAlign);
  2194. return PointerDatumToPod(PointerGetDatum(result));
  2195. }
  2196. else {
  2197. /* Must be nested array expressions */
  2198. auto element_type = ElemTypeDesc.TypeId;
  2199. int nbytes = 0;
  2200. int nitems = 0;
  2201. int outer_nelems = 0;
  2202. int elem_ndims = 0;
  2203. int *elem_dims = NULL;
  2204. int *elem_lbs = NULL;
  2205. bool firstone = true;
  2206. bool havenulls = false;
  2207. bool haveempty = false;
  2208. char **subdata;
  2209. bits8 **subbitmaps;
  2210. int *subbytes;
  2211. int *subnitems;
  2212. int32 dataoffset;
  2213. char *dat;
  2214. int iitem;
  2215. subdata = (char **)palloc(nelems * sizeof(char *));
  2216. subbitmaps = (bits8 **)palloc(nelems * sizeof(bits8 *));
  2217. subbytes = (int *)palloc(nelems * sizeof(int));
  2218. subnitems = (int *)palloc(nelems * sizeof(int));
  2219. /* loop through and get data area from each element */
  2220. for (int elemoff = 0; elemoff < nelems; elemoff++)
  2221. {
  2222. Datum arraydatum;
  2223. bool eisnull;
  2224. ArrayType *array;
  2225. int this_ndims;
  2226. arraydatum = dvalues[elemoff];
  2227. eisnull = dnulls[elemoff];
  2228. /* temporarily ignore null subarrays */
  2229. if (eisnull)
  2230. {
  2231. haveempty = true;
  2232. continue;
  2233. }
  2234. array = DatumGetArrayTypeP(arraydatum);
  2235. /* run-time double-check on element type */
  2236. if (element_type != ARR_ELEMTYPE(array))
  2237. ereport(ERROR,
  2238. (errcode(ERRCODE_DATATYPE_MISMATCH),
  2239. errmsg("cannot merge incompatible arrays"),
  2240. errdetail("Array with element type %s cannot be "
  2241. "included in ARRAY construct with element type %s.",
  2242. format_type_be(ARR_ELEMTYPE(array)),
  2243. format_type_be(element_type))));
  2244. this_ndims = ARR_NDIM(array);
  2245. /* temporarily ignore zero-dimensional subarrays */
  2246. if (this_ndims <= 0)
  2247. {
  2248. haveempty = true;
  2249. continue;
  2250. }
  2251. if (firstone)
  2252. {
  2253. /* Get sub-array details from first member */
  2254. elem_ndims = this_ndims;
  2255. ndims = elem_ndims + 1;
  2256. if (ndims <= 0 || ndims > MAXDIM)
  2257. ereport(ERROR,
  2258. (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
  2259. errmsg("number of array dimensions (%d) exceeds the maximum allowed (%d)",
  2260. ndims, MAXDIM)));
  2261. elem_dims = (int *)palloc(elem_ndims * sizeof(int));
  2262. memcpy(elem_dims, ARR_DIMS(array), elem_ndims * sizeof(int));
  2263. elem_lbs = (int *)palloc(elem_ndims * sizeof(int));
  2264. memcpy(elem_lbs, ARR_LBOUND(array), elem_ndims * sizeof(int));
  2265. firstone = false;
  2266. }
  2267. else
  2268. {
  2269. /* Check other sub-arrays are compatible */
  2270. if (elem_ndims != this_ndims ||
  2271. memcmp(elem_dims, ARR_DIMS(array),
  2272. elem_ndims * sizeof(int)) != 0 ||
  2273. memcmp(elem_lbs, ARR_LBOUND(array),
  2274. elem_ndims * sizeof(int)) != 0)
  2275. ereport(ERROR,
  2276. (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
  2277. errmsg("multidimensional arrays must have array "
  2278. "expressions with matching dimensions")));
  2279. }
  2280. subdata[outer_nelems] = ARR_DATA_PTR(array);
  2281. subbitmaps[outer_nelems] = ARR_NULLBITMAP(array);
  2282. subbytes[outer_nelems] = ARR_SIZE(array) - ARR_DATA_OFFSET(array);
  2283. nbytes += subbytes[outer_nelems];
  2284. subnitems[outer_nelems] = ArrayGetNItems(this_ndims,
  2285. ARR_DIMS(array));
  2286. nitems += subnitems[outer_nelems];
  2287. havenulls |= ARR_HASNULL(array);
  2288. outer_nelems++;
  2289. }
  2290. /*
  2291. * If all items were null or empty arrays, return an empty array;
  2292. * otherwise, if some were and some weren't, raise error. (Note: we
  2293. * must special-case this somehow to avoid trying to generate a 1-D
  2294. * array formed from empty arrays. It's not ideal...)
  2295. */
  2296. if (haveempty)
  2297. {
  2298. if (ndims == 0) /* didn't find any nonempty array */
  2299. {
  2300. return PointerDatumToPod(PointerGetDatum(construct_empty_array(element_type)));
  2301. }
  2302. ereport(ERROR,
  2303. (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
  2304. errmsg("multidimensional arrays must have array "
  2305. "expressions with matching dimensions")));
  2306. }
  2307. /* setup for multi-D array */
  2308. dims[0] = outer_nelems;
  2309. lbs[0] = 1;
  2310. for (int i = 1; i < ndims; i++)
  2311. {
  2312. dims[i] = elem_dims[i - 1];
  2313. lbs[i] = elem_lbs[i - 1];
  2314. }
  2315. /* check for subscript overflow */
  2316. (void)ArrayGetNItems(ndims, dims);
  2317. ArrayCheckBounds(ndims, dims, lbs);
  2318. if (havenulls)
  2319. {
  2320. dataoffset = ARR_OVERHEAD_WITHNULLS(ndims, nitems);
  2321. nbytes += dataoffset;
  2322. }
  2323. else
  2324. {
  2325. dataoffset = 0; /* marker for no null bitmap */
  2326. nbytes += ARR_OVERHEAD_NONULLS(ndims);
  2327. }
  2328. ArrayType* result = (ArrayType *)palloc(nbytes);
  2329. SET_VARSIZE(result, nbytes);
  2330. result->ndim = ndims;
  2331. result->dataoffset = dataoffset;
  2332. result->elemtype = element_type;
  2333. memcpy(ARR_DIMS(result), dims, ndims * sizeof(int));
  2334. memcpy(ARR_LBOUND(result), lbs, ndims * sizeof(int));
  2335. dat = ARR_DATA_PTR(result);
  2336. iitem = 0;
  2337. for (int i = 0; i < outer_nelems; i++)
  2338. {
  2339. memcpy(dat, subdata[i], subbytes[i]);
  2340. dat += subbytes[i];
  2341. if (havenulls)
  2342. array_bitmap_copy(ARR_NULLBITMAP(result), iitem,
  2343. subbitmaps[i], 0,
  2344. subnitems[i]);
  2345. iitem += subnitems[i];
  2346. }
  2347. return PointerDatumToPod(PointerGetDatum(result));
  2348. }
  2349. }
  2350. }
  2351. private:
  2352. void RegisterDependencies() const final {
  2353. for (auto arg : ArgNodes) {
  2354. DependsOn(arg);
  2355. }
  2356. }
  2357. TComputationNodePtrVector ArgNodes;
  2358. TVector<TType*> ArgTypes;
  2359. const NPg::TTypeDesc& ArrayTypeDesc;
  2360. const NPg::TTypeDesc& ElemTypeDesc;
  2361. TVector<NPg::TTypeDesc> ArgDescs;
  2362. bool MultiDims = false;
  2363. };
  2364. template <bool PassByValue>
  2365. class TPgClone : public TMutableComputationNode<TPgClone<PassByValue>> {
  2366. typedef TMutableComputationNode<TPgClone<PassByValue>> TBaseComputation;
  2367. public:
  2368. TPgClone(TComputationMutables& mutables, IComputationNode* input, TComputationNodePtrVector&& dependentNodes, i32 typeLen)
  2369. : TBaseComputation(mutables)
  2370. , Input(input)
  2371. , DependentNodes(std::move(dependentNodes))
  2372. , TypeLen(typeLen)
  2373. {
  2374. }
  2375. NUdf::TUnboxedValuePod DoCalculate(TComputationContext& compCtx) const {
  2376. auto value = Input->GetValue(compCtx);
  2377. if constexpr (PassByValue) {
  2378. return value.Release();
  2379. }
  2380. auto datum = PointerDatumFromPod(value);
  2381. if (TypeLen == -1) {
  2382. return PointerDatumToPod((Datum)MakeVar(GetVarBuf((const text*)datum)));
  2383. } else if (TypeLen == -2) {
  2384. return PointerDatumToPod((Datum)MakeCString(TStringBuf((const char*)datum)));
  2385. } else {
  2386. return PointerDatumToPod((Datum)MakeFixedString(TStringBuf((const char*)datum), TypeLen));
  2387. }
  2388. }
  2389. private:
  2390. void RegisterDependencies() const final {
  2391. this->DependsOn(Input);
  2392. for (auto arg : DependentNodes) {
  2393. this->DependsOn(arg);
  2394. }
  2395. }
  2396. IComputationNode* const Input;
  2397. TComputationNodePtrVector DependentNodes;
  2398. const i32 TypeLen;
  2399. };
  2400. struct TFromPgExec {
  2401. TFromPgExec(ui32 sourceId)
  2402. : SourceId(sourceId)
  2403. , IsCString(NPg::LookupType(sourceId).TypeLen == -2)
  2404. {}
  2405. arrow::Status Exec(arrow::compute::KernelContext* ctx, const arrow::compute::ExecBatch& batch, arrow::Datum* res) const {
  2406. arrow::Datum inputDatum = batch.values[0];
  2407. Y_ENSURE(inputDatum.is_array());
  2408. const auto& array= *inputDatum.array();
  2409. size_t length = array.length;
  2410. switch (SourceId) {
  2411. case BOOLOID: {
  2412. auto inputPtr = array.GetValues<ui64>(1);
  2413. auto outputPtr = res->array()->GetMutableValues<ui8>(1);
  2414. for (size_t i = 0; i < length; ++i) {
  2415. outputPtr[i] = DatumGetBool(inputPtr[i]) ? 1 : 0;
  2416. }
  2417. break;
  2418. }
  2419. case INT2OID: {
  2420. auto inputPtr = array.GetValues<ui64>(1);
  2421. auto outputPtr = res->array()->GetMutableValues<i16>(1);
  2422. for (size_t i = 0; i < length; ++i) {
  2423. outputPtr[i] = DatumGetInt16(inputPtr[i]);
  2424. }
  2425. break;
  2426. }
  2427. case INT4OID: {
  2428. auto inputPtr = array.GetValues<ui64>(1);
  2429. auto outputPtr = res->array()->GetMutableValues<i32>(1);
  2430. for (size_t i = 0; i < length; ++i) {
  2431. outputPtr[i] = DatumGetInt32(inputPtr[i]);
  2432. }
  2433. break;
  2434. }
  2435. case INT8OID: {
  2436. auto inputPtr = array.GetValues<ui64>(1);
  2437. auto outputPtr = res->array()->GetMutableValues<i64>(1);
  2438. for (size_t i = 0; i < length; ++i) {
  2439. outputPtr[i] = DatumGetInt64(inputPtr[i]);
  2440. }
  2441. break;
  2442. }
  2443. case FLOAT4OID: {
  2444. auto inputPtr = array.GetValues<ui64>(1);
  2445. auto outputPtr = res->array()->GetMutableValues<float>(1);
  2446. for (size_t i = 0; i < length; ++i) {
  2447. outputPtr[i] = DatumGetFloat4(inputPtr[i]);
  2448. }
  2449. break;
  2450. }
  2451. case FLOAT8OID: {
  2452. auto inputPtr = array.GetValues<ui64>(1);
  2453. auto outputPtr = res->array()->GetMutableValues<double>(1);
  2454. for (size_t i = 0; i < length; ++i) {
  2455. outputPtr[i] = DatumGetFloat8(inputPtr[i]);
  2456. }
  2457. break;
  2458. }
  2459. case TEXTOID:
  2460. case VARCHAROID:
  2461. case BYTEAOID:
  2462. case CSTRINGOID: {
  2463. NUdf::TStringBlockReader<arrow::BinaryType, true> reader;
  2464. NUdf::TStringArrayBuilder<arrow::BinaryType, true> builder(NKikimr::NMiniKQL::TTypeInfoHelper(), SourceId == BYTEAOID ? arrow::binary() : arrow::utf8(), *ctx->memory_pool(), length);
  2465. for (size_t i = 0; i < length; ++i) {
  2466. auto item = reader.GetItem(array, i);
  2467. if (!item) {
  2468. builder.Add(NUdf::TBlockItem());
  2469. continue;
  2470. }
  2471. ui32 len;
  2472. const char* ptr = item.AsStringRef().Data() + sizeof(void*);
  2473. if (IsCString) {
  2474. len = strlen(ptr);
  2475. } else {
  2476. len = GetCleanVarSize((const text*)ptr);
  2477. Y_ENSURE(len + VARHDRSZ + sizeof(void*) <= item.AsStringRef().Size());
  2478. ptr += VARHDRSZ;
  2479. }
  2480. builder.Add(NUdf::TBlockItem(NUdf::TStringRef(ptr, len)));
  2481. }
  2482. *res = builder.Build(true);
  2483. break;
  2484. }
  2485. case DATEOID: {
  2486. NUdf::TFixedSizeBlockReader<ui64, true> reader;
  2487. NUdf::TFixedSizeArrayBuilder<i32, true> builder(NKikimr::NMiniKQL::TTypeInfoHelper(), arrow::int32(), *ctx->memory_pool(), length);
  2488. for (size_t i = 0; i < length; ++i) {
  2489. auto item = reader.GetItem(array, i);
  2490. if (!item) {
  2491. builder.Add(NUdf::TBlockItem());
  2492. continue;
  2493. }
  2494. auto res = (i32)DatumGetInt32((Datum)item.Get<ui64>()) - PgDateShift;
  2495. if (res < NUdf::MIN_DATE32 || res > NUdf::MAX_DATE32) {
  2496. builder.Add(NUdf::TBlockItem());
  2497. continue;
  2498. }
  2499. builder.Add(NUdf::TBlockItem(res));
  2500. }
  2501. *res = builder.Build(true);
  2502. break;
  2503. }
  2504. case TIMESTAMPOID: {
  2505. NUdf::TFixedSizeBlockReader<ui64, true> reader;
  2506. NUdf::TFixedSizeArrayBuilder<i64, true> builder(NKikimr::NMiniKQL::TTypeInfoHelper(), arrow::int64(), *ctx->memory_pool(), length);
  2507. for (size_t i = 0; i < length; ++i) {
  2508. auto item = reader.GetItem(array, i);
  2509. if (!item) {
  2510. builder.Add(NUdf::TBlockItem());
  2511. continue;
  2512. }
  2513. auto res = (i64)DatumGetInt64((Datum)item.Get<ui64>()) - PgTimestampShift;
  2514. if (res < NUdf::MIN_TIMESTAMP64 || res > NUdf::MAX_TIMESTAMP64) {
  2515. builder.Add(NUdf::TBlockItem());
  2516. continue;
  2517. }
  2518. builder.Add(NUdf::TBlockItem(res));
  2519. }
  2520. *res = builder.Build(true);
  2521. break;
  2522. }
  2523. default:
  2524. ythrow yexception() << "Unsupported type: " << NPg::LookupType(SourceId).Name;
  2525. }
  2526. return arrow::Status::OK();
  2527. }
  2528. const ui32 SourceId;
  2529. const bool IsCString;
  2530. };
  2531. std::shared_ptr<arrow::compute::ScalarKernel> MakeFromPgKernel(TType* inputType, TType* resultType, ui32 sourceId) {
  2532. const TVector<TType*> argTypes = { inputType };
  2533. std::shared_ptr<arrow::DataType> returnArrowType;
  2534. MKQL_ENSURE(ConvertArrowType(AS_TYPE(TBlockType, resultType)->GetItemType(), returnArrowType), "Unsupported arrow type");
  2535. auto exec = std::make_shared<TFromPgExec>(sourceId);
  2536. auto kernel = std::make_shared<arrow::compute::ScalarKernel>(ConvertToInputTypes(argTypes), ConvertToOutputType(resultType),
  2537. [exec](arrow::compute::KernelContext* ctx, const arrow::compute::ExecBatch& batch, arrow::Datum* res) {
  2538. return exec->Exec(ctx, batch, res);
  2539. });
  2540. switch (sourceId) {
  2541. case BOOLOID:
  2542. case INT2OID:
  2543. case INT4OID:
  2544. case INT8OID:
  2545. case FLOAT4OID:
  2546. case FLOAT8OID:
  2547. break;
  2548. case TEXTOID:
  2549. case VARCHAROID:
  2550. case BYTEAOID:
  2551. case CSTRINGOID:
  2552. case DATEOID:
  2553. case TIMESTAMPOID:
  2554. kernel->null_handling = arrow::compute::NullHandling::COMPUTED_NO_PREALLOCATE;
  2555. break;
  2556. default:
  2557. ythrow yexception() << "Unsupported type: " << NPg::LookupType(sourceId).Name;
  2558. }
  2559. return kernel;
  2560. }
  2561. struct TToPgExec {
  2562. TToPgExec(NUdf::EDataSlot sourceDataSlot)
  2563. : SourceDataSlot(sourceDataSlot)
  2564. {}
  2565. arrow::Status Exec(arrow::compute::KernelContext* ctx, const arrow::compute::ExecBatch& batch, arrow::Datum* res) const {
  2566. arrow::Datum inputDatum = batch.values[0];
  2567. Y_ENSURE(inputDatum.is_array());
  2568. const auto& array= *inputDatum.array();
  2569. size_t length = array.length;
  2570. switch (SourceDataSlot) {
  2571. case NUdf::EDataSlot::Bool: {
  2572. auto inputPtr = array.GetValues<ui8>(1);
  2573. auto outputPtr = res->array()->GetMutableValues<ui64>(1);
  2574. for (size_t i = 0; i < length; ++i) {
  2575. outputPtr[i] = BoolGetDatum(inputPtr[i]);
  2576. }
  2577. break;
  2578. }
  2579. case NUdf::EDataSlot::Int8: {
  2580. auto inputPtr = array.GetValues<i8>(1);
  2581. auto outputPtr = res->array()->GetMutableValues<ui64>(1);
  2582. for (size_t i = 0; i < length; ++i) {
  2583. outputPtr[i] = Int16GetDatum(inputPtr[i]);
  2584. }
  2585. break;
  2586. }
  2587. case NUdf::EDataSlot::Uint8: {
  2588. auto inputPtr = array.GetValues<ui8>(1);
  2589. auto outputPtr = res->array()->GetMutableValues<ui64>(1);
  2590. for (size_t i = 0; i < length; ++i) {
  2591. outputPtr[i] = Int16GetDatum(inputPtr[i]);
  2592. }
  2593. break;
  2594. }
  2595. case NUdf::EDataSlot::Int16: {
  2596. auto inputPtr = array.GetValues<i16>(1);
  2597. auto outputPtr = res->array()->GetMutableValues<ui64>(1);
  2598. for (size_t i = 0; i < length; ++i) {
  2599. outputPtr[i] = Int16GetDatum(inputPtr[i]);
  2600. }
  2601. break;
  2602. }
  2603. case NUdf::EDataSlot::Uint16: {
  2604. auto inputPtr = array.GetValues<ui16>(1);
  2605. auto outputPtr = res->array()->GetMutableValues<ui64>(1);
  2606. for (size_t i = 0; i < length; ++i) {
  2607. outputPtr[i] = Int32GetDatum(inputPtr[i]);
  2608. }
  2609. break;
  2610. }
  2611. case NUdf::EDataSlot::Int32: {
  2612. auto inputPtr = array.GetValues<i32>(1);
  2613. auto outputPtr = res->array()->GetMutableValues<ui64>(1);
  2614. for (size_t i = 0; i < length; ++i) {
  2615. outputPtr[i] = Int32GetDatum(inputPtr[i]);
  2616. }
  2617. break;
  2618. }
  2619. case NUdf::EDataSlot::Uint32: {
  2620. auto inputPtr = array.GetValues<ui32>(1);
  2621. auto outputPtr = res->array()->GetMutableValues<ui64>(1);
  2622. for (size_t i = 0; i < length; ++i) {
  2623. outputPtr[i] = Int64GetDatum(inputPtr[i]);
  2624. }
  2625. break;
  2626. }
  2627. case NUdf::EDataSlot::Int64: {
  2628. auto inputPtr = array.GetValues<i64>(1);
  2629. auto outputPtr = res->array()->GetMutableValues<ui64>(1);
  2630. for (size_t i = 0; i < length; ++i) {
  2631. outputPtr[i] = Int64GetDatum(inputPtr[i]);
  2632. }
  2633. break;
  2634. }
  2635. case NUdf::EDataSlot::Uint64: {
  2636. NUdf::TFixedSizeBlockReader<ui64, true> reader;
  2637. NUdf::TStringArrayBuilder<arrow::BinaryType, true> builder(NKikimr::NMiniKQL::TTypeInfoHelper(), arrow::binary(), *ctx->memory_pool(), length);
  2638. for (size_t i = 0; i < length; ++i) {
  2639. auto item = reader.GetItem(array, i);
  2640. if (!item) {
  2641. builder.Add(NUdf::TBlockItem());
  2642. continue;
  2643. }
  2644. auto res = Uint64ToPgNumeric(item.Get<ui64>());
  2645. auto ref = NUdf::TStringRef((const char*)res, GetFullVarSize((const text*)res));
  2646. auto ptr = builder.AddPgItem<false, 0>(ref);
  2647. UpdateCleanVarSize((text*)(ptr + sizeof(void*)), GetCleanVarSize((const text*)res));
  2648. pfree(res);
  2649. }
  2650. *res = builder.Build(true);
  2651. break;
  2652. }
  2653. case NUdf::EDataSlot::Float: {
  2654. auto inputPtr = array.GetValues<float>(1);
  2655. auto outputPtr = res->array()->GetMutableValues<ui64>(1);
  2656. for (size_t i = 0; i < length; ++i) {
  2657. outputPtr[i] = Float4GetDatum(inputPtr[i]);
  2658. }
  2659. break;
  2660. }
  2661. case NUdf::EDataSlot::Double: {
  2662. auto inputPtr = array.GetValues<double>(1);
  2663. auto outputPtr = res->array()->GetMutableValues<ui64>(1);
  2664. for (size_t i = 0; i < length; ++i) {
  2665. outputPtr[i] = Float8GetDatum(inputPtr[i]);
  2666. }
  2667. break;
  2668. }
  2669. case NUdf::EDataSlot::Utf8:
  2670. case NUdf::EDataSlot::String:
  2671. case NUdf::EDataSlot::Yson: {
  2672. NUdf::TStringBlockReader<arrow::BinaryType, true> reader;
  2673. NUdf::TStringArrayBuilder<arrow::BinaryType, true> builder(NKikimr::NMiniKQL::TTypeInfoHelper(), arrow::binary(), *ctx->memory_pool(), length);
  2674. for (size_t i = 0; i < length; ++i) {
  2675. auto item = reader.GetItem(array, i);
  2676. if (!item) {
  2677. builder.Add(NUdf::TBlockItem());
  2678. continue;
  2679. }
  2680. auto ref = item.AsStringRef();
  2681. auto ptr = builder.AddPgItem<false, VARHDRSZ>(ref);
  2682. UpdateCleanVarSize((text*)(ptr + sizeof(void*)), ref.Size());
  2683. }
  2684. *res = builder.Build(true);
  2685. break;
  2686. }
  2687. case NUdf::EDataSlot::Date: {
  2688. auto inputPtr = array.GetValues<ui16>(1);
  2689. auto outputPtr = res->array()->GetMutableValues<ui64>(1);
  2690. for (size_t i = 0; i < length; ++i) {
  2691. outputPtr[i] = Int32GetDatum(Date2Pg(inputPtr[i]));
  2692. }
  2693. break;
  2694. }
  2695. case NUdf::EDataSlot::Datetime: {
  2696. auto inputPtr = array.GetValues<ui32>(1);
  2697. auto outputPtr = res->array()->GetMutableValues<ui64>(1);
  2698. for (size_t i = 0; i < length; ++i) {
  2699. outputPtr[i] = Int64GetDatum(Timestamp2Pg(inputPtr[i] * 1000000ull));
  2700. }
  2701. break;
  2702. }
  2703. case NUdf::EDataSlot::Timestamp: {
  2704. auto inputPtr = array.GetValues<ui64>(1);
  2705. auto outputPtr = res->array()->GetMutableValues<ui64>(1);
  2706. for (size_t i = 0; i < length; ++i) {
  2707. outputPtr[i] = Int64GetDatum(Timestamp2Pg(inputPtr[i]));
  2708. }
  2709. break;
  2710. }
  2711. case NUdf::EDataSlot::Date32: {
  2712. auto inputPtr = array.GetValues<i32>(1);
  2713. auto outputPtr = res->array()->GetMutableValues<ui64>(1);
  2714. for (size_t i = 0; i < length; ++i) {
  2715. outputPtr[i] = Int32GetDatum(Date2Pg(inputPtr[i]));
  2716. }
  2717. break;
  2718. }
  2719. case NUdf::EDataSlot::Datetime64: {
  2720. auto inputPtr = array.GetValues<i64>(1);
  2721. auto outputPtr = res->array()->GetMutableValues<ui64>(1);
  2722. for (size_t i = 0; i < length; ++i) {
  2723. outputPtr[i] = Int64GetDatum(Timestamp2Pg(inputPtr[i] * 1000000ull));
  2724. }
  2725. break;
  2726. }
  2727. case NUdf::EDataSlot::Timestamp64: {
  2728. auto inputPtr = array.GetValues<i64>(1);
  2729. auto outputPtr = res->array()->GetMutableValues<ui64>(1);
  2730. for (size_t i = 0; i < length; ++i) {
  2731. outputPtr[i] = Int64GetDatum(Timestamp2Pg(inputPtr[i]));
  2732. }
  2733. break;
  2734. }
  2735. case NUdf::EDataSlot::Interval:
  2736. case NUdf::EDataSlot::Interval64: {
  2737. NUdf::TFixedSizeBlockReader<i64, true> reader;
  2738. NUdf::TStringArrayBuilder<arrow::BinaryType, true> builder(NKikimr::NMiniKQL::TTypeInfoHelper(), arrow::binary(), *ctx->memory_pool(), length);
  2739. for (size_t i = 0; i < length; ++i) {
  2740. auto item = reader.GetItem(array, i);
  2741. if (!item) {
  2742. builder.Add(NUdf::TBlockItem());
  2743. continue;
  2744. }
  2745. Interval pgInterval;
  2746. pgInterval.time = item.Get<i64>() % 86400000000ll;
  2747. pgInterval.day = item.Get<i64>() / 86400000000ll;
  2748. pgInterval.month = 0;
  2749. auto ref = NUdf::TStringRef((const char*)&pgInterval, sizeof(Interval));
  2750. builder.AddPgItem<false, 0>(ref);
  2751. }
  2752. *res = builder.Build(true);
  2753. break;
  2754. }
  2755. case NUdf::EDataSlot::Json:
  2756. {
  2757. NUdf::TStringBlockReader<arrow::BinaryType, true> reader;
  2758. NUdf::TStringArrayBuilder<arrow::BinaryType, true> builder(NKikimr::NMiniKQL::TTypeInfoHelper(), arrow::binary(), *ctx->memory_pool(), length);
  2759. for (size_t i = 0; i < length; ++i) {
  2760. auto item = reader.GetItem(array, i);
  2761. if (!item) {
  2762. builder.Add(NUdf::TBlockItem());
  2763. continue;
  2764. }
  2765. auto input = MakeCString(item.AsStringRef());
  2766. auto res = (text*)DirectFunctionCall1Coll(json_in, DEFAULT_COLLATION_OID, PointerGetDatum(input));
  2767. pfree(input);
  2768. auto ref = NUdf::TStringRef((const char*)res, GetFullVarSize(res));
  2769. auto ptr = builder.AddPgItem<false, 0>(ref);
  2770. UpdateCleanVarSize((text*)(ptr + sizeof(void*)), GetCleanVarSize(res));
  2771. pfree(res);
  2772. }
  2773. *res = builder.Build(true);
  2774. break;
  2775. }
  2776. case NUdf::EDataSlot::JsonDocument:
  2777. {
  2778. NUdf::TStringBlockReader<arrow::BinaryType, true> reader;
  2779. NUdf::TStringArrayBuilder<arrow::BinaryType, true> builder(NKikimr::NMiniKQL::TTypeInfoHelper(), arrow::binary(), *ctx->memory_pool(), length);
  2780. for (size_t i = 0; i < length; ++i) {
  2781. auto item = reader.GetItem(array, i);
  2782. if (!item) {
  2783. builder.Add(NUdf::TBlockItem());
  2784. continue;
  2785. }
  2786. auto str = NKikimr::NBinaryJson::SerializeToJson(item.AsStringRef());
  2787. auto res = (text*)DirectFunctionCall1Coll(jsonb_in, DEFAULT_COLLATION_OID, PointerGetDatum(str.c_str()));
  2788. auto ref = NUdf::TStringRef((const char*)res, GetFullVarSize(res));
  2789. auto ptr = builder.AddPgItem<false, 0>(ref);
  2790. UpdateCleanVarSize((text*)(ptr + sizeof(void*)), GetCleanVarSize(res));
  2791. pfree(res);
  2792. }
  2793. *res = builder.Build(true);
  2794. break;
  2795. }
  2796. default:
  2797. ythrow yexception() << "Unsupported type: " << NUdf::GetDataTypeInfo(SourceDataSlot).Name;
  2798. }
  2799. return arrow::Status::OK();
  2800. }
  2801. const NUdf::EDataSlot SourceDataSlot;
  2802. };
  2803. std::shared_ptr<arrow::compute::ScalarKernel> MakeToPgKernel(TType* inputType, TType* resultType, NUdf::EDataSlot dataSlot) {
  2804. const TVector<TType*> argTypes = { inputType };
  2805. std::shared_ptr<arrow::DataType> returnArrowType;
  2806. MKQL_ENSURE(ConvertArrowType(AS_TYPE(TBlockType, resultType)->GetItemType(), returnArrowType), "Unsupported arrow type");
  2807. auto exec = std::make_shared<TToPgExec>(dataSlot);
  2808. auto kernel = std::make_shared<arrow::compute::ScalarKernel>(ConvertToInputTypes(argTypes), ConvertToOutputType(resultType),
  2809. [exec](arrow::compute::KernelContext* ctx, const arrow::compute::ExecBatch& batch, arrow::Datum* res) {
  2810. return exec->Exec(ctx, batch, res);
  2811. });
  2812. switch (dataSlot) {
  2813. case NUdf::EDataSlot::Bool:
  2814. case NUdf::EDataSlot::Int8:
  2815. case NUdf::EDataSlot::Uint8:
  2816. case NUdf::EDataSlot::Int16:
  2817. case NUdf::EDataSlot::Uint16:
  2818. case NUdf::EDataSlot::Int32:
  2819. case NUdf::EDataSlot::Uint32:
  2820. case NUdf::EDataSlot::Int64:
  2821. case NUdf::EDataSlot::Float:
  2822. case NUdf::EDataSlot::Double:
  2823. case NUdf::EDataSlot::Date:
  2824. case NUdf::EDataSlot::Datetime:
  2825. case NUdf::EDataSlot::Timestamp:
  2826. case NUdf::EDataSlot::Date32:
  2827. case NUdf::EDataSlot::Datetime64:
  2828. case NUdf::EDataSlot::Timestamp64:
  2829. break;
  2830. case NUdf::EDataSlot::String:
  2831. case NUdf::EDataSlot::Utf8:
  2832. case NUdf::EDataSlot::Interval:
  2833. case NUdf::EDataSlot::Interval64:
  2834. case NUdf::EDataSlot::Uint64:
  2835. case NUdf::EDataSlot::Yson:
  2836. case NUdf::EDataSlot::Json:
  2837. case NUdf::EDataSlot::JsonDocument:
  2838. kernel->null_handling = arrow::compute::NullHandling::COMPUTED_NO_PREALLOCATE;
  2839. break;
  2840. default:
  2841. ythrow yexception() << "Unsupported type: " << NUdf::GetDataTypeInfo(dataSlot).Name;
  2842. }
  2843. return kernel;
  2844. }
  2845. std::shared_ptr<arrow::compute::ScalarKernel> MakePgKernel(TVector<TType*> argTypes, TType* resultType, TExecFunc execFunc, ui32 procId) {
  2846. std::shared_ptr<arrow::DataType> returnArrowType;
  2847. MKQL_ENSURE(ConvertArrowType(AS_TYPE(TBlockType, resultType)->GetItemType(), returnArrowType), "Unsupported arrow type");
  2848. auto kernel = std::make_shared<arrow::compute::ScalarKernel>(ConvertToInputTypes(argTypes), ConvertToOutputType(resultType),
  2849. [execFunc](arrow::compute::KernelContext* ctx, const arrow::compute::ExecBatch& batch, arrow::Datum* res) {
  2850. return execFunc(ctx, batch, res);
  2851. });
  2852. TVector<ui32> pgArgTypes;
  2853. for (const auto& t : argTypes) {
  2854. auto itemType = AS_TYPE(TBlockType, t)->GetItemType();
  2855. ui32 oid;
  2856. if (itemType->IsNull()) {
  2857. oid = UNKNOWNOID;
  2858. } else {
  2859. oid = AS_TYPE(TPgType, itemType)->GetTypeId();
  2860. }
  2861. pgArgTypes.push_back(oid);
  2862. }
  2863. kernel->null_handling = arrow::compute::NullHandling::COMPUTED_NO_PREALLOCATE;
  2864. kernel->init = [procId, pgArgTypes](arrow::compute::KernelContext*, const arrow::compute::KernelInitArgs&) {
  2865. auto state = std::make_unique<TPgKernelState>();
  2866. Zero(state->flinfo);
  2867. GetPgFuncAddr(procId, state->flinfo);
  2868. YQL_ENSURE(state->flinfo.fn_addr);
  2869. state->resultinfo = nullptr;
  2870. state->context = nullptr;
  2871. state->fncollation = DEFAULT_COLLATION_OID;
  2872. const auto& procDesc = NPg::LookupProc(procId);
  2873. const auto& retTypeDesc = NPg::LookupType(procDesc.ResultType);
  2874. state->Name = procDesc.Name;
  2875. state->IsFixedResult = retTypeDesc.PassByValue;
  2876. state->TypeLen = retTypeDesc.TypeLen;
  2877. auto fmgrDataHolder = std::make_shared<TPgArgsExprBuilder>();
  2878. for (const auto& argTypeId : pgArgTypes) {
  2879. const auto& argTypeDesc = NPg::LookupType(argTypeId);
  2880. state->IsFixedArg.push_back(argTypeDesc.PassByValue);
  2881. fmgrDataHolder->Add(argTypeId);
  2882. }
  2883. state->flinfo.fn_expr = fmgrDataHolder->Build(procDesc);
  2884. state->FmgrDataHolder = fmgrDataHolder;
  2885. state->ProcDesc = &procDesc;
  2886. return arrow::Result(std::move(state));
  2887. };
  2888. return kernel;
  2889. }
  2890. TComputationNodeFactory GetPgFactory() {
  2891. return [] (TCallable& callable, const TComputationNodeFactoryContext& ctx) -> IComputationNode* {
  2892. TStringBuf name = callable.GetType()->GetName();
  2893. if (name == "PgConst") {
  2894. const auto typeIdData = AS_VALUE(TDataLiteral, callable.GetInput(0));
  2895. const auto valueData = AS_VALUE(TDataLiteral, callable.GetInput(1));
  2896. ui32 typeId = typeIdData->AsValue().Get<ui32>();
  2897. auto value = valueData->AsValue().AsStringRef();
  2898. IComputationNode* typeMod = nullptr;
  2899. if (callable.GetInputsCount() >= 3) {
  2900. typeMod = LocateNode(ctx.NodeLocator, callable, 2);
  2901. }
  2902. return new TPgConst(ctx.Mutables, typeId, value, typeMod);
  2903. }
  2904. if (name == "PgInternal0") {
  2905. return new TPgInternal0(ctx.Mutables);
  2906. }
  2907. if (name == "PgTableContent") {
  2908. const auto clusterData = AS_VALUE(TDataLiteral, callable.GetInput(0));
  2909. const auto tableData = AS_VALUE(TDataLiteral, callable.GetInput(1));
  2910. const auto cluster = clusterData->AsValue().AsStringRef();
  2911. const auto table = tableData->AsValue().AsStringRef();
  2912. const auto returnType = callable.GetType()->GetReturnType();
  2913. return new TPgTableContent(ctx.Mutables, cluster, table, returnType);
  2914. }
  2915. if (name == "PgToRecord") {
  2916. auto structType = AS_TYPE(TStructType, callable.GetInput(0).GetStaticType());
  2917. auto input = LocateNode(ctx.NodeLocator, callable, 0);
  2918. TVector<std::pair<TString, TString>> members;
  2919. auto tuple = AS_VALUE(TTupleLiteral, callable.GetInput(1));
  2920. MKQL_ENSURE(tuple->GetValuesCount() % 2 == 0, "Malformed names");
  2921. for (ui32 i = 0; i < tuple->GetValuesCount(); i += 2) {
  2922. const auto recordFieldData = AS_VALUE(TDataLiteral, tuple->GetValue(i));
  2923. const auto struсtMemberData = AS_VALUE(TDataLiteral, tuple->GetValue(i + 1));
  2924. const TString recordField(recordFieldData->AsValue().AsStringRef());
  2925. const TString struсtMember(struсtMemberData->AsValue().AsStringRef());
  2926. members.push_back({recordField, struсtMember});
  2927. }
  2928. return new TPgToRecord(ctx.Mutables, input, structType, std::move(members));
  2929. }
  2930. if (name == "PgResolvedCall") {
  2931. const auto useContextData = AS_VALUE(TDataLiteral, callable.GetInput(0));
  2932. const auto rangeFunctionData = AS_VALUE(TDataLiteral, callable.GetInput(1));
  2933. const auto nameData = AS_VALUE(TDataLiteral, callable.GetInput(2));
  2934. const auto idData = AS_VALUE(TDataLiteral, callable.GetInput(3));
  2935. auto useContext = useContextData->AsValue().Get<bool>();
  2936. auto rangeFunction = rangeFunctionData->AsValue().Get<bool>();
  2937. auto name = nameData->AsValue().AsStringRef();
  2938. auto id = idData->AsValue().Get<ui32>();
  2939. TComputationNodePtrVector argNodes;
  2940. TVector<TType*> argTypes;
  2941. for (ui32 i = 4; i < callable.GetInputsCount(); ++i) {
  2942. argNodes.emplace_back(LocateNode(ctx.NodeLocator, callable, i));
  2943. argTypes.emplace_back(callable.GetInput(i).GetStaticType());
  2944. }
  2945. const auto returnType = callable.GetType()->GetReturnType();
  2946. const bool isList = returnType->IsList();
  2947. const auto itemType = isList ? AS_TYPE(TListType, returnType)->GetItemType() : returnType;
  2948. const TStructType* structType = nullptr;
  2949. if (rangeFunction) {
  2950. if (itemType->IsStruct()) {
  2951. structType = AS_TYPE(TStructType, itemType);
  2952. }
  2953. }
  2954. if (isList) {
  2955. YQL_ENSURE(!useContext);
  2956. return new TPgResolvedMultiCall(ctx.Mutables, name, id, std::move(argNodes), std::move(argTypes), itemType, structType);
  2957. } else {
  2958. YQL_ENSURE(!structType);
  2959. if (useContext) {
  2960. return new TPgResolvedCall<true>(ctx.Mutables, name, id, std::move(argNodes), std::move(argTypes), returnType);
  2961. } else {
  2962. return new TPgResolvedCall<false>(ctx.Mutables, name, id, std::move(argNodes), std::move(argTypes), returnType);
  2963. }
  2964. }
  2965. }
  2966. if (name == "BlockPgResolvedCall") {
  2967. const auto nameData = AS_VALUE(TDataLiteral, callable.GetInput(0));
  2968. const auto idData = AS_VALUE(TDataLiteral, callable.GetInput(1));
  2969. auto name = nameData->AsValue().AsStringRef();
  2970. auto id = idData->AsValue().Get<ui32>();
  2971. TComputationNodePtrVector argNodes;
  2972. TVector<TType*> argTypes;
  2973. for (ui32 i = 2; i < callable.GetInputsCount(); ++i) {
  2974. argNodes.emplace_back(LocateNode(ctx.NodeLocator, callable, i));
  2975. argTypes.emplace_back(callable.GetInput(i).GetStaticType());
  2976. }
  2977. auto returnType = callable.GetType()->GetReturnType();
  2978. auto execFunc = FindExec(id);
  2979. YQL_ENSURE(execFunc);
  2980. auto kernel = MakePgKernel(argTypes, returnType, execFunc, id);
  2981. return new TBlockFuncNode(ctx.Mutables, callable.GetType()->GetName(), std::move(argNodes), argTypes, *kernel, kernel);
  2982. }
  2983. if (name == "PgCast") {
  2984. auto arg = LocateNode(ctx.NodeLocator, callable, 0);
  2985. auto inputType = callable.GetInput(0).GetStaticType();
  2986. ui32 sourceId = 0;
  2987. if (!inputType->IsNull()) {
  2988. sourceId = AS_TYPE(TPgType, inputType)->GetTypeId();
  2989. }
  2990. auto returnType = callable.GetType()->GetReturnType();
  2991. auto targetId = AS_TYPE(TPgType, returnType)->GetTypeId();
  2992. IComputationNode* typeMod = nullptr;
  2993. if (callable.GetInputsCount() >= 2) {
  2994. typeMod = LocateNode(ctx.NodeLocator, callable, 1);
  2995. }
  2996. return new TPgCast(ctx.Mutables, sourceId, targetId, arg, typeMod);
  2997. }
  2998. if (name == "FromPg") {
  2999. auto arg = LocateNode(ctx.NodeLocator, callable, 0);
  3000. auto inputType = callable.GetInput(0).GetStaticType();
  3001. ui32 sourceId = AS_TYPE(TPgType, inputType)->GetTypeId();
  3002. switch (sourceId) {
  3003. case BOOLOID:
  3004. return new TFromPg<NUdf::EDataSlot::Bool, false>(ctx.Mutables, arg);
  3005. case INT2OID:
  3006. return new TFromPg<NUdf::EDataSlot::Int16, false>(ctx.Mutables, arg);
  3007. case INT4OID:
  3008. return new TFromPg<NUdf::EDataSlot::Int32, false>(ctx.Mutables, arg);
  3009. case INT8OID:
  3010. return new TFromPg<NUdf::EDataSlot::Int64, false>(ctx.Mutables, arg);
  3011. case FLOAT4OID:
  3012. return new TFromPg<NUdf::EDataSlot::Float, false>(ctx.Mutables, arg);
  3013. case FLOAT8OID:
  3014. return new TFromPg<NUdf::EDataSlot::Double, false>(ctx.Mutables, arg);
  3015. case TEXTOID:
  3016. case VARCHAROID:
  3017. return new TFromPg<NUdf::EDataSlot::Utf8, false>(ctx.Mutables, arg);
  3018. case BYTEAOID:
  3019. return new TFromPg<NUdf::EDataSlot::String, false>(ctx.Mutables, arg);
  3020. case CSTRINGOID:
  3021. return new TFromPg<NUdf::EDataSlot::Utf8, true>(ctx.Mutables, arg);
  3022. case DATEOID:
  3023. return new TFromPg<NUdf::EDataSlot::Date32, true>(ctx.Mutables, arg);
  3024. case TIMESTAMPOID:
  3025. return new TFromPg<NUdf::EDataSlot::Timestamp64, true>(ctx.Mutables, arg);
  3026. case UUIDOID:
  3027. return new TFromPg<NUdf::EDataSlot::Uuid, true>(ctx.Mutables, arg);
  3028. default:
  3029. ythrow yexception() << "Unsupported type: " << NPg::LookupType(sourceId).Name;
  3030. }
  3031. }
  3032. if (name == "BlockFromPg") {
  3033. auto arg = LocateNode(ctx.NodeLocator, callable, 0);
  3034. auto inputType = callable.GetInput(0).GetStaticType();
  3035. auto returnType = callable.GetType()->GetReturnType();
  3036. ui32 sourceId = AS_TYPE(TPgType, AS_TYPE(TBlockType, inputType)->GetItemType())->GetTypeId();
  3037. auto kernel = MakeFromPgKernel(inputType, returnType, sourceId);
  3038. return new TBlockFuncNode(ctx.Mutables, callable.GetType()->GetName(), { arg }, { inputType }, *kernel, kernel);
  3039. }
  3040. if (name == "ToPg") {
  3041. auto arg = LocateNode(ctx.NodeLocator, callable, 0);
  3042. auto inputType = callable.GetInput(0).GetStaticType();
  3043. auto argType = inputType;
  3044. if (argType->IsOptional()) {
  3045. argType = AS_TYPE(TOptionalType, argType)->GetItemType();
  3046. }
  3047. auto dataType = AS_TYPE(TDataType, argType);
  3048. auto sourceDataSlot = dataType->GetDataSlot();
  3049. switch (*sourceDataSlot) {
  3050. case NUdf::EDataSlot::Bool:
  3051. return new TToPg<NUdf::EDataSlot::Bool>(ctx.Mutables, arg, dataType);
  3052. case NUdf::EDataSlot::Int8:
  3053. return new TToPg<NUdf::EDataSlot::Int8>(ctx.Mutables, arg, dataType);
  3054. case NUdf::EDataSlot::Uint8:
  3055. return new TToPg<NUdf::EDataSlot::Uint8>(ctx.Mutables, arg, dataType);
  3056. case NUdf::EDataSlot::Int16:
  3057. return new TToPg<NUdf::EDataSlot::Int16>(ctx.Mutables, arg, dataType);
  3058. case NUdf::EDataSlot::Uint16:
  3059. return new TToPg<NUdf::EDataSlot::Uint16>(ctx.Mutables, arg, dataType);
  3060. case NUdf::EDataSlot::Int32:
  3061. return new TToPg<NUdf::EDataSlot::Int32>(ctx.Mutables, arg, dataType);
  3062. case NUdf::EDataSlot::Uint32:
  3063. return new TToPg<NUdf::EDataSlot::Uint32>(ctx.Mutables, arg, dataType);
  3064. case NUdf::EDataSlot::Int64:
  3065. return new TToPg<NUdf::EDataSlot::Int64>(ctx.Mutables, arg, dataType);
  3066. case NUdf::EDataSlot::Uint64:
  3067. return new TToPg<NUdf::EDataSlot::Uint64>(ctx.Mutables, arg, dataType);
  3068. case NUdf::EDataSlot::Float:
  3069. return new TToPg<NUdf::EDataSlot::Float>(ctx.Mutables, arg, dataType);
  3070. case NUdf::EDataSlot::Double:
  3071. return new TToPg<NUdf::EDataSlot::Double>(ctx.Mutables, arg, dataType);
  3072. case NUdf::EDataSlot::Utf8:
  3073. return new TToPg<NUdf::EDataSlot::Utf8>(ctx.Mutables, arg, dataType);
  3074. case NUdf::EDataSlot::String:
  3075. return new TToPg<NUdf::EDataSlot::String>(ctx.Mutables, arg, dataType);
  3076. case NUdf::EDataSlot::Date:
  3077. return new TToPg<NUdf::EDataSlot::Date>(ctx.Mutables, arg, dataType);
  3078. case NUdf::EDataSlot::Datetime:
  3079. return new TToPg<NUdf::EDataSlot::Datetime>(ctx.Mutables, arg, dataType);
  3080. case NUdf::EDataSlot::Timestamp:
  3081. return new TToPg<NUdf::EDataSlot::Timestamp>(ctx.Mutables, arg, dataType);
  3082. case NUdf::EDataSlot::Interval:
  3083. return new TToPg<NUdf::EDataSlot::Interval>(ctx.Mutables, arg, dataType);
  3084. case NUdf::EDataSlot::TzDate:
  3085. return new TToPg<NUdf::EDataSlot::TzDate>(ctx.Mutables, arg, dataType);
  3086. case NUdf::EDataSlot::TzDatetime:
  3087. return new TToPg<NUdf::EDataSlot::TzDatetime>(ctx.Mutables, arg, dataType);
  3088. case NUdf::EDataSlot::TzTimestamp:
  3089. return new TToPg<NUdf::EDataSlot::TzTimestamp>(ctx.Mutables, arg, dataType);
  3090. case NUdf::EDataSlot::Date32:
  3091. return new TToPg<NUdf::EDataSlot::Date32>(ctx.Mutables, arg, dataType);
  3092. case NUdf::EDataSlot::Datetime64:
  3093. return new TToPg<NUdf::EDataSlot::Datetime64>(ctx.Mutables, arg, dataType);
  3094. case NUdf::EDataSlot::Timestamp64:
  3095. return new TToPg<NUdf::EDataSlot::Timestamp64>(ctx.Mutables, arg, dataType);
  3096. case NUdf::EDataSlot::Interval64:
  3097. return new TToPg<NUdf::EDataSlot::Interval64>(ctx.Mutables, arg, dataType);
  3098. case NUdf::EDataSlot::TzDate32:
  3099. return new TToPg<NUdf::EDataSlot::TzDate32>(ctx.Mutables, arg, dataType);
  3100. case NUdf::EDataSlot::TzDatetime64:
  3101. return new TToPg<NUdf::EDataSlot::TzDatetime64>(ctx.Mutables, arg, dataType);
  3102. case NUdf::EDataSlot::TzTimestamp64:
  3103. return new TToPg<NUdf::EDataSlot::TzTimestamp64>(ctx.Mutables, arg, dataType);
  3104. case NUdf::EDataSlot::Uuid:
  3105. return new TToPg<NUdf::EDataSlot::Uuid>(ctx.Mutables, arg, dataType);
  3106. case NUdf::EDataSlot::Yson:
  3107. return new TToPg<NUdf::EDataSlot::Yson>(ctx.Mutables, arg, dataType);
  3108. case NUdf::EDataSlot::Json:
  3109. return new TToPg<NUdf::EDataSlot::Json>(ctx.Mutables, arg, dataType);
  3110. case NUdf::EDataSlot::JsonDocument:
  3111. return new TToPg<NUdf::EDataSlot::JsonDocument>(ctx.Mutables, arg, dataType);
  3112. case NUdf::EDataSlot::Decimal:
  3113. return new TToPg<NUdf::EDataSlot::Decimal>(ctx.Mutables, arg, dataType);
  3114. case NUdf::EDataSlot::DyNumber:
  3115. return new TToPg<NUdf::EDataSlot::DyNumber>(ctx.Mutables, arg, dataType);
  3116. default:
  3117. ythrow yexception() << "Unsupported type: " << NUdf::GetDataTypeInfo(*sourceDataSlot).Name;
  3118. }
  3119. }
  3120. if (name == "BlockToPg") {
  3121. auto arg = LocateNode(ctx.NodeLocator, callable, 0);
  3122. auto inputType = callable.GetInput(0).GetStaticType();
  3123. auto argType = AS_TYPE(TBlockType, inputType)->GetItemType();
  3124. if (argType->IsOptional()) {
  3125. argType = AS_TYPE(TOptionalType, argType)->GetItemType();
  3126. }
  3127. auto sourceDataSlot = AS_TYPE(TDataType, argType)->GetDataSlot();
  3128. auto returnType = callable.GetType()->GetReturnType();
  3129. auto targetId = AS_TYPE(TPgType, AS_TYPE(TBlockType, returnType)->GetItemType())->GetTypeId();
  3130. auto kernel = MakeToPgKernel(inputType, returnType, *sourceDataSlot);
  3131. return new TBlockFuncNode(ctx.Mutables, callable.GetType()->GetName(), { arg }, { inputType }, *kernel, kernel);
  3132. }
  3133. if (name == "PgArray") {
  3134. TComputationNodePtrVector argNodes;
  3135. TVector<TType*> argTypes;
  3136. for (ui32 i = 0; i < callable.GetInputsCount(); ++i) {
  3137. argNodes.emplace_back(LocateNode(ctx.NodeLocator, callable, i));
  3138. argTypes.emplace_back(callable.GetInput(i).GetStaticType());
  3139. }
  3140. auto returnType = callable.GetType()->GetReturnType();
  3141. auto arrayTypeId = AS_TYPE(TPgType, returnType)->GetTypeId();
  3142. return new TPgArray(ctx.Mutables, std::move(argNodes), std::move(argTypes), arrayTypeId);
  3143. }
  3144. if (name == "PgClone") {
  3145. auto input = LocateNode(ctx.NodeLocator, callable, 0);
  3146. TComputationNodePtrVector dependentNodes;
  3147. for (ui32 i = 1; i < callable.GetInputsCount(); ++i) {
  3148. dependentNodes.emplace_back(LocateNode(ctx.NodeLocator, callable, i));
  3149. }
  3150. auto returnType = callable.GetType()->GetReturnType();
  3151. auto typeId = AS_TYPE(TPgType, returnType)->GetTypeId();
  3152. const auto& desc = NPg::LookupType(typeId);
  3153. if (desc.PassByValue) {
  3154. return new TPgClone<true>(ctx.Mutables, input, std::move(dependentNodes), desc.TypeLen);
  3155. } else if (desc.TypeLen == -1) {
  3156. return new TPgClone<false>(ctx.Mutables, input, std::move(dependentNodes), desc.TypeLen);
  3157. } else {
  3158. return new TPgClone<false>(ctx.Mutables, input, std::move(dependentNodes), desc.TypeLen);
  3159. }
  3160. }
  3161. return nullptr;
  3162. };
  3163. }
  3164. namespace NCommon {
  3165. TString PgValueToNativeText(const NUdf::TUnboxedValuePod& value, ui32 pgTypeId) {
  3166. YQL_ENSURE(value); // null could not be represented as text
  3167. TPAllocScope call;
  3168. const auto& typeInfo = NPg::LookupType(pgTypeId);
  3169. auto outFuncId = typeInfo.OutFuncId;
  3170. if (typeInfo.TypeId == typeInfo.ArrayTypeId) {
  3171. outFuncId = NPg::LookupProc("array_out", { 0 }).ProcId;
  3172. }
  3173. char* str = nullptr;
  3174. Y_DEFER {
  3175. if (str) {
  3176. pfree(str);
  3177. }
  3178. };
  3179. {
  3180. FmgrInfo finfo;
  3181. Zero(finfo);
  3182. Y_ENSURE(outFuncId);
  3183. GetPgFuncAddr(outFuncId, finfo);
  3184. Y_ENSURE(!finfo.fn_retset);
  3185. Y_ENSURE(finfo.fn_addr);
  3186. Y_ENSURE(finfo.fn_nargs == 1);
  3187. LOCAL_FCINFO(callInfo, 1);
  3188. Zero(*callInfo);
  3189. callInfo->flinfo = &finfo;
  3190. callInfo->nargs = 1;
  3191. callInfo->fncollation = DEFAULT_COLLATION_OID;
  3192. callInfo->isnull = false;
  3193. callInfo->args[0] = { typeInfo.PassByValue ?
  3194. ScalarDatumFromPod(value) :
  3195. PointerDatumFromPod(value), false };
  3196. str = (char*)finfo.fn_addr(callInfo);
  3197. Y_ENSURE(!callInfo->isnull);
  3198. return TString(str);
  3199. }
  3200. }
  3201. template <typename F>
  3202. void PgValueToNativeBinaryImpl(const NUdf::TUnboxedValuePod& value, ui32 pgTypeId, bool needCanonizeFp, F f) {
  3203. YQL_ENSURE(value); // null could not be represented as binary
  3204. if (!NPg::HasType(pgTypeId)) {
  3205. f(TStringBuf(value.AsStringRef()));
  3206. return;
  3207. }
  3208. const bool oldNeedCanonizeFp = NeedCanonizeFp;
  3209. NeedCanonizeFp = needCanonizeFp;
  3210. Y_DEFER {
  3211. NeedCanonizeFp = oldNeedCanonizeFp;
  3212. };
  3213. TPAllocScope call;
  3214. const auto& typeInfo = NPg::LookupType(pgTypeId);
  3215. auto sendFuncId = typeInfo.SendFuncId;
  3216. if (typeInfo.TypeId == typeInfo.ArrayTypeId) {
  3217. sendFuncId = NPg::LookupProc("array_send", { 0 }).ProcId;
  3218. }
  3219. text* x = nullptr;
  3220. Y_DEFER {
  3221. if (x) {
  3222. pfree(x);
  3223. }
  3224. };
  3225. {
  3226. FmgrInfo finfo;
  3227. Zero(finfo);
  3228. Y_ENSURE(sendFuncId);
  3229. GetPgFuncAddr(sendFuncId, finfo);
  3230. Y_ENSURE(!finfo.fn_retset);
  3231. Y_ENSURE(finfo.fn_addr);
  3232. Y_ENSURE(finfo.fn_nargs == 1);
  3233. LOCAL_FCINFO(callInfo, 1);
  3234. Zero(*callInfo);
  3235. callInfo->flinfo = &finfo;
  3236. callInfo->nargs = 1;
  3237. callInfo->fncollation = DEFAULT_COLLATION_OID;
  3238. callInfo->isnull = false;
  3239. callInfo->args[0] = { typeInfo.PassByValue ?
  3240. ScalarDatumFromPod(value) :
  3241. PointerDatumFromPod(value), false };
  3242. x = (text*)finfo.fn_addr(callInfo);
  3243. Y_ENSURE(!callInfo->isnull);
  3244. auto s = GetVarBuf(x);
  3245. ui32 len = s.size();
  3246. f(TStringBuf(s.data(), s.size()));
  3247. }
  3248. }
  3249. TString PgValueToNativeBinary(const NUdf::TUnboxedValuePod& value, ui32 pgTypeId) {
  3250. TString result;
  3251. PgValueToNativeBinaryImpl(value, pgTypeId, false, [&result](TStringBuf b) {
  3252. result = b;
  3253. });
  3254. return result;
  3255. }
  3256. TString PgValueToString(const NUdf::TUnboxedValuePod& value, ui32 pgTypeId) {
  3257. YQL_ENSURE(value); // null could not be represented as text
  3258. switch (pgTypeId) {
  3259. case BOOLOID:
  3260. return DatumGetBool(ScalarDatumFromPod(value)) ? "true" : "false";
  3261. case INT2OID:
  3262. return ToString(DatumGetInt16(ScalarDatumFromPod(value)));
  3263. case INT4OID:
  3264. return ToString(DatumGetInt32(ScalarDatumFromPod(value)));
  3265. case INT8OID:
  3266. return ToString(DatumGetInt64(ScalarDatumFromPod(value)));
  3267. case FLOAT4OID:
  3268. return ::FloatToString(DatumGetFloat4(ScalarDatumFromPod(value)));
  3269. case FLOAT8OID:
  3270. return ::FloatToString(DatumGetFloat8(ScalarDatumFromPod(value)));
  3271. case BYTEAOID:
  3272. case VARCHAROID:
  3273. case TEXTOID: {
  3274. const auto x = (const text*)PointerDatumFromPod(value);
  3275. return TString(GetVarBuf(x));
  3276. }
  3277. case CSTRINGOID: {
  3278. return TString((const char*)PointerDatumFromPod(value));
  3279. }
  3280. default:
  3281. return PgValueToNativeText(value, pgTypeId);
  3282. }
  3283. }
  3284. void WriteYsonValueInTableFormatPg(TOutputBuf& buf, TPgType* type, const NUdf::TUnboxedValuePod& value, bool topLevel) {
  3285. using namespace NYson::NDetail;
  3286. if (!value) {
  3287. if (topLevel) {
  3288. buf.Write(BeginListSymbol);
  3289. buf.Write(EndListSymbol);
  3290. } else {
  3291. buf.Write(EntitySymbol);
  3292. }
  3293. return;
  3294. }
  3295. switch (type->GetTypeId()) {
  3296. case BOOLOID:
  3297. buf.Write(DatumGetBool(ScalarDatumFromPod(value)) ? TrueMarker : FalseMarker);
  3298. break;
  3299. case INT2OID:
  3300. buf.Write(Int64Marker);
  3301. buf.WriteVarI64(DatumGetInt16(ScalarDatumFromPod(value)));
  3302. break;
  3303. case INT4OID:
  3304. buf.Write(Int64Marker);
  3305. buf.WriteVarI64(DatumGetInt32(ScalarDatumFromPod(value)));
  3306. break;
  3307. case INT8OID:
  3308. buf.Write(Int64Marker);
  3309. buf.WriteVarI64(DatumGetInt64(ScalarDatumFromPod(value)));
  3310. break;
  3311. case FLOAT4OID: {
  3312. buf.Write(DoubleMarker);
  3313. double val = DatumGetFloat4(ScalarDatumFromPod(value));
  3314. buf.WriteMany((const char*)&val, sizeof(val));
  3315. break;
  3316. }
  3317. case FLOAT8OID: {
  3318. buf.Write(DoubleMarker);
  3319. double val = DatumGetFloat8(ScalarDatumFromPod(value));
  3320. buf.WriteMany((const char*)&val, sizeof(val));
  3321. break;
  3322. }
  3323. case BYTEAOID:
  3324. case VARCHAROID:
  3325. case TEXTOID: {
  3326. const auto x = (const text*)PointerDatumFromPod(value);
  3327. auto s = GetVarBuf(x);
  3328. buf.Write(StringMarker);
  3329. buf.WriteVarI32(s.size());
  3330. buf.WriteMany(s.data(), s.size());
  3331. break;
  3332. }
  3333. case CSTRINGOID: {
  3334. auto s = (const char*)PointerDatumFromPod(value);
  3335. auto len = strlen(s);
  3336. buf.Write(StringMarker);
  3337. buf.WriteVarI32(len);
  3338. buf.WriteMany(s, len);
  3339. break;
  3340. }
  3341. default:
  3342. buf.Write(StringMarker);
  3343. PgValueToNativeBinaryImpl(value, type->GetTypeId(), true, [&buf](TStringBuf b) {
  3344. buf.WriteVarI32(b.size());
  3345. buf.WriteMany(b.data(), b.size());
  3346. });
  3347. break;
  3348. }
  3349. }
  3350. void WriteYsonValuePg(NResult::TYsonResultWriter& writer, const NUdf::TUnboxedValuePod& value, TPgType* type,
  3351. const TVector<ui32>* structPositions) {
  3352. if (!value) {
  3353. writer.OnNull();
  3354. return;
  3355. }
  3356. writer.OnStringScalar(PgValueToString(value, type->GetTypeId()));
  3357. }
  3358. NUdf::TUnboxedValue ReadYsonValueInTableFormatPg(TPgType* type, char cmd, TInputBuf& buf) {
  3359. using namespace NYson::NDetail;
  3360. if (cmd == EntitySymbol) {
  3361. return NUdf::TUnboxedValuePod();
  3362. }
  3363. if (cmd == BeginListSymbol) {
  3364. cmd = buf.Read();
  3365. if (cmd == ListItemSeparatorSymbol) {
  3366. cmd = buf.Read();
  3367. }
  3368. YQL_ENSURE(cmd == EndListSymbol);
  3369. return NUdf::TUnboxedValuePod();
  3370. }
  3371. switch (type->GetTypeId()) {
  3372. case BOOLOID: {
  3373. YQL_ENSURE(cmd == FalseMarker || cmd == TrueMarker, "Expected either true or false, but got: " << TString(cmd).Quote());
  3374. return ScalarDatumToPod(BoolGetDatum(cmd == TrueMarker));
  3375. }
  3376. case INT2OID: {
  3377. CHECK_EXPECTED(cmd, Int64Marker);
  3378. auto x = i16(buf.ReadVarI64());
  3379. return ScalarDatumToPod(Int16GetDatum(x));
  3380. }
  3381. case INT4OID: {
  3382. CHECK_EXPECTED(cmd, Int64Marker);
  3383. auto x = i32(buf.ReadVarI64());
  3384. return ScalarDatumToPod(Int32GetDatum(x));
  3385. }
  3386. case INT8OID: {
  3387. CHECK_EXPECTED(cmd, Int64Marker);
  3388. auto x = buf.ReadVarI64();
  3389. return ScalarDatumToPod(Int64GetDatum(x));
  3390. }
  3391. case FLOAT4OID: {
  3392. CHECK_EXPECTED(cmd, DoubleMarker);
  3393. double x;
  3394. buf.ReadMany((char*)&x, sizeof(x));
  3395. return ScalarDatumToPod(Float4GetDatum(x));
  3396. }
  3397. case FLOAT8OID: {
  3398. CHECK_EXPECTED(cmd, DoubleMarker);
  3399. double x;
  3400. buf.ReadMany((char*)&x, sizeof(x));
  3401. return ScalarDatumToPod(Float8GetDatum(x));
  3402. }
  3403. case BYTEAOID:
  3404. case VARCHAROID:
  3405. case TEXTOID: {
  3406. CHECK_EXPECTED(cmd, StringMarker);
  3407. auto s = buf.ReadYtString();
  3408. auto ret = MakeVar(s);
  3409. return PointerDatumToPod((Datum)ret);
  3410. }
  3411. case CSTRINGOID: {
  3412. CHECK_EXPECTED(cmd, StringMarker);
  3413. auto s = buf.ReadYtString();
  3414. auto ret = MakeCString(s);
  3415. return PointerDatumToPod((Datum)ret);
  3416. }
  3417. default:
  3418. TPAllocScope call;
  3419. auto s = buf.ReadYtString();
  3420. return PgValueFromNativeBinary(s, type->GetTypeId());
  3421. }
  3422. }
  3423. NUdf::TUnboxedValue PgValueFromNativeBinary(const TStringBuf binary, ui32 pgTypeId) {
  3424. if (!NPg::HasType(pgTypeId)) {
  3425. return MakeString(binary);
  3426. }
  3427. TPAllocScope call;
  3428. StringInfoData stringInfo;
  3429. stringInfo.data = (char*)binary.data();
  3430. stringInfo.len = binary.size();
  3431. stringInfo.maxlen = binary.size();
  3432. stringInfo.cursor = 0;
  3433. const auto& typeInfo = NPg::LookupType(pgTypeId);
  3434. auto typeIOParam = MakeTypeIOParam(typeInfo);
  3435. auto receiveFuncId = typeInfo.ReceiveFuncId;
  3436. if (typeInfo.TypeId == typeInfo.ArrayTypeId) {
  3437. receiveFuncId = NPg::LookupProc("array_recv", { 0,0,0 }).ProcId;
  3438. }
  3439. {
  3440. FmgrInfo finfo;
  3441. Zero(finfo);
  3442. Y_ENSURE(receiveFuncId);
  3443. GetPgFuncAddr(receiveFuncId, finfo);
  3444. Y_ENSURE(!finfo.fn_retset);
  3445. Y_ENSURE(finfo.fn_addr);
  3446. Y_ENSURE(finfo.fn_nargs >= 1 && finfo.fn_nargs <= 3);
  3447. LOCAL_FCINFO(callInfo, 3);
  3448. Zero(*callInfo);
  3449. callInfo->flinfo = &finfo;
  3450. callInfo->nargs = 3;
  3451. callInfo->fncollation = DEFAULT_COLLATION_OID;
  3452. callInfo->isnull = false;
  3453. callInfo->args[0] = { (Datum)&stringInfo, false };
  3454. callInfo->args[1] = { ObjectIdGetDatum(typeIOParam), false };
  3455. callInfo->args[2] = { Int32GetDatum(-1), false };
  3456. auto x = finfo.fn_addr(callInfo);
  3457. Y_ENSURE(!callInfo->isnull);
  3458. if (stringInfo.cursor != stringInfo.len) {
  3459. TStringBuilder errMsg;
  3460. errMsg << "Not all data has been consumed by 'recv' function: " << NPg::LookupProc(receiveFuncId).Name << ", data size: " << stringInfo.len << ", consumed size: " << stringInfo.cursor;
  3461. UdfTerminate(errMsg.c_str());
  3462. }
  3463. return AnyDatumToPod(x, typeInfo.PassByValue);
  3464. }
  3465. }
  3466. NUdf::TUnboxedValue PgValueFromNativeText(const TStringBuf text, ui32 pgTypeId) {
  3467. TString str{ text };
  3468. TPAllocScope call;
  3469. const auto& typeInfo = NPg::LookupType(pgTypeId);
  3470. auto typeIOParam = MakeTypeIOParam(typeInfo);
  3471. auto inFuncId = typeInfo.InFuncId;
  3472. if (typeInfo.TypeId == typeInfo.ArrayTypeId) {
  3473. inFuncId = NPg::LookupProc("array_in", { 0,0,0 }).ProcId;
  3474. }
  3475. {
  3476. FmgrInfo finfo;
  3477. Zero(finfo);
  3478. Y_ENSURE(inFuncId);
  3479. GetPgFuncAddr(inFuncId, finfo);
  3480. Y_ENSURE(!finfo.fn_retset);
  3481. Y_ENSURE(finfo.fn_addr);
  3482. Y_ENSURE(finfo.fn_nargs >= 1 && finfo.fn_nargs <= 3);
  3483. LOCAL_FCINFO(callInfo, 3);
  3484. Zero(*callInfo);
  3485. callInfo->flinfo = &finfo;
  3486. callInfo->nargs = 3;
  3487. callInfo->fncollation = DEFAULT_COLLATION_OID;
  3488. callInfo->isnull = false;
  3489. callInfo->args[0] = { (Datum)str.c_str(), false };
  3490. callInfo->args[1] = { ObjectIdGetDatum(typeIOParam), false };
  3491. callInfo->args[2] = { Int32GetDatum(-1), false };
  3492. auto x = finfo.fn_addr(callInfo);
  3493. Y_ENSURE(!callInfo->isnull);
  3494. return AnyDatumToPod(x, typeInfo.PassByValue);
  3495. }
  3496. }
  3497. NUdf::TUnboxedValue PgValueFromString(const TStringBuf s, ui32 pgTypeId) {
  3498. switch (pgTypeId) {
  3499. case BOOLOID: {
  3500. return ScalarDatumToPod(BoolGetDatum(FromString<bool>(s)));
  3501. }
  3502. case INT2OID: {
  3503. return ScalarDatumToPod(Int16GetDatum(FromString<i16>(s)));
  3504. }
  3505. case INT4OID: {
  3506. return ScalarDatumToPod(Int32GetDatum(FromString<i32>(s)));
  3507. }
  3508. case INT8OID: {
  3509. return ScalarDatumToPod(Int64GetDatum(FromString<i64>(s)));
  3510. }
  3511. case FLOAT4OID: {
  3512. return ScalarDatumToPod(Float4GetDatum(FromString<float>(s)));
  3513. }
  3514. case FLOAT8OID: {
  3515. return ScalarDatumToPod(Float8GetDatum(FromString<double>(s)));
  3516. }
  3517. case BYTEAOID:
  3518. case VARCHAROID:
  3519. case TEXTOID: {
  3520. auto ret = MakeVar(s);
  3521. return PointerDatumToPod((Datum)ret);
  3522. }
  3523. case CSTRINGOID: {
  3524. auto ret = MakeCString(s);
  3525. return PointerDatumToPod((Datum)ret);
  3526. }
  3527. default:
  3528. return PgValueFromNativeText(s, pgTypeId);
  3529. }
  3530. }
  3531. NUdf::TUnboxedValue ReadYsonValuePg(TPgType* type, char cmd, TInputBuf& buf) {
  3532. using namespace NYson::NDetail;
  3533. if (cmd == EntitySymbol) {
  3534. return NUdf::TUnboxedValuePod();
  3535. }
  3536. CHECK_EXPECTED(cmd, StringMarker);
  3537. auto s = buf.ReadYtString();
  3538. return PgValueFromString(s, type->GetTypeId());
  3539. }
  3540. void SkipSkiffPg(TPgType* type, NCommon::TInputBuf& buf) {
  3541. auto marker = buf.Read();
  3542. if (!marker) {
  3543. return;
  3544. }
  3545. switch (type->GetTypeId()) {
  3546. case BOOLOID: {
  3547. buf.Read();
  3548. return;
  3549. }
  3550. case INT2OID:
  3551. case INT4OID:
  3552. case INT8OID: {
  3553. buf.SkipMany(sizeof(i64));
  3554. return;
  3555. }
  3556. case FLOAT4OID:
  3557. case FLOAT8OID: {
  3558. buf.SkipMany(sizeof(double));
  3559. return;
  3560. }
  3561. default: {
  3562. ui32 size;
  3563. buf.ReadMany((char*)&size, sizeof(size));
  3564. CHECK_STRING_LENGTH_UNSIGNED(size);
  3565. buf.SkipMany(size);
  3566. return;
  3567. }
  3568. }
  3569. }
  3570. NUdf::TUnboxedValue ReadSkiffPg(TPgType* type, NCommon::TInputBuf& buf) {
  3571. auto marker = buf.Read();
  3572. if (!marker) {
  3573. return NUdf::TUnboxedValue();
  3574. }
  3575. switch (type->GetTypeId()) {
  3576. case BOOLOID: {
  3577. auto x = buf.Read();
  3578. return ScalarDatumToPod(BoolGetDatum(x != 0));
  3579. }
  3580. case INT2OID: {
  3581. i64 x;
  3582. buf.ReadMany((char*)&x, sizeof(x));
  3583. return ScalarDatumToPod(Int16GetDatum((i16)x));
  3584. }
  3585. case INT4OID: {
  3586. i64 x;
  3587. buf.ReadMany((char*)&x, sizeof(x));
  3588. return ScalarDatumToPod(Int32GetDatum((i32)x));
  3589. }
  3590. case INT8OID: {
  3591. i64 x;
  3592. buf.ReadMany((char*)&x, sizeof(x));
  3593. return ScalarDatumToPod(Int64GetDatum(x));
  3594. }
  3595. case FLOAT4OID: {
  3596. double x;
  3597. buf.ReadMany((char*)&x, sizeof(x));
  3598. return ScalarDatumToPod(Float4GetDatum((float)x));
  3599. }
  3600. case FLOAT8OID: {
  3601. double x;
  3602. buf.ReadMany((char*)&x, sizeof(x));
  3603. return ScalarDatumToPod(Float8GetDatum(x));
  3604. }
  3605. case BYTEAOID:
  3606. case VARCHAROID:
  3607. case TEXTOID: {
  3608. ui32 size;
  3609. buf.ReadMany((char*)&size, sizeof(size));
  3610. CHECK_STRING_LENGTH_UNSIGNED(size);
  3611. text* s = (text*)palloc(size + VARHDRSZ);
  3612. auto mem = s;
  3613. Y_DEFER {
  3614. if (mem) {
  3615. pfree(mem);
  3616. }
  3617. };
  3618. UpdateCleanVarSize(s, size);
  3619. buf.ReadMany(GetMutableVarData(s), size);
  3620. mem = nullptr;
  3621. return PointerDatumToPod((Datum)s);
  3622. }
  3623. case CSTRINGOID: {
  3624. ui32 size;
  3625. buf.ReadMany((char*)&size, sizeof(size));
  3626. CHECK_STRING_LENGTH_UNSIGNED(size);
  3627. char* s = (char*)palloc(size + 1);
  3628. auto mem = s;
  3629. Y_DEFER {
  3630. if (mem) {
  3631. pfree(mem);
  3632. }
  3633. };
  3634. buf.ReadMany(s, size);
  3635. mem = nullptr;
  3636. s[size] = '\0';
  3637. return PointerDatumToPod((Datum)s);
  3638. }
  3639. default:
  3640. TPAllocScope call;
  3641. ui32 size;
  3642. buf.ReadMany((char*)&size, sizeof(size));
  3643. CHECK_STRING_LENGTH_UNSIGNED(size);
  3644. char* s = (char*)TWithDefaultMiniKQLAlloc::AllocWithSize(size);
  3645. Y_DEFER {
  3646. TWithDefaultMiniKQLAlloc::FreeWithSize(s, size);
  3647. };
  3648. buf.ReadMany(s, size);
  3649. return PgValueFromNativeBinary(TStringBuf(s, size), type->GetTypeId());
  3650. }
  3651. }
  3652. void WriteSkiffPg(TPgType* type, const NUdf::TUnboxedValuePod& value, NCommon::TOutputBuf& buf) {
  3653. if (!value) {
  3654. buf.Write('\0');
  3655. return;
  3656. }
  3657. buf.Write('\1');
  3658. switch (type->GetTypeId()) {
  3659. case BOOLOID: {
  3660. char x = DatumGetBool(ScalarDatumFromPod(value));
  3661. buf.Write(x);
  3662. break;
  3663. }
  3664. case INT2OID: {
  3665. i64 x = DatumGetInt16(ScalarDatumFromPod(value));
  3666. buf.WriteMany((const char*)&x, sizeof(x));
  3667. break;
  3668. }
  3669. case INT4OID: {
  3670. i64 x = DatumGetInt32(ScalarDatumFromPod(value));
  3671. buf.WriteMany((const char*)&x, sizeof(x));
  3672. break;
  3673. }
  3674. case INT8OID: {
  3675. i64 x = DatumGetInt64(ScalarDatumFromPod(value));
  3676. buf.WriteMany((const char*)&x, sizeof(x));
  3677. break;
  3678. }
  3679. case FLOAT4OID: {
  3680. double x = DatumGetFloat4(ScalarDatumFromPod(value));
  3681. buf.WriteMany((const char*)&x, sizeof(x));
  3682. break;
  3683. }
  3684. case FLOAT8OID: {
  3685. double x = DatumGetFloat8(ScalarDatumFromPod(value));
  3686. buf.WriteMany((const char*)&x, sizeof(x));
  3687. break;
  3688. }
  3689. case BYTEAOID:
  3690. case VARCHAROID:
  3691. case TEXTOID: {
  3692. const auto x = (const text*)PointerDatumFromPod(value);
  3693. auto s = GetVarBuf(x);
  3694. ui32 len = s.size();
  3695. buf.WriteMany((const char*)&len, sizeof(len));
  3696. buf.WriteMany(s.data(), len);
  3697. break;
  3698. }
  3699. case CSTRINGOID: {
  3700. const auto x = (const char*)PointerDatumFromPod(value);
  3701. ui32 len = strlen(x);
  3702. buf.WriteMany((const char*)&len, sizeof(len));
  3703. buf.WriteMany(x, len);
  3704. break;
  3705. }
  3706. default:
  3707. PgValueToNativeBinaryImpl(value, type->GetTypeId(), true, [&buf](TStringBuf b) {
  3708. ui32 len = b.size();
  3709. buf.WriteMany((const char*)&len, sizeof(len));
  3710. buf.WriteMany(b.data(), len);
  3711. });
  3712. }
  3713. }
  3714. extern "C" void ReadSkiffPgValue(TPgType* type, NUdf::TUnboxedValue& value, NCommon::TInputBuf& buf) {
  3715. value = ReadSkiffPg(type, buf);
  3716. }
  3717. extern "C" void WriteSkiffPgValue(TPgType* type, const NUdf::TUnboxedValuePod& value, NCommon::TOutputBuf& buf) {
  3718. WriteSkiffPg(type, value, buf);
  3719. }
  3720. } // namespace NCommon
  3721. namespace {
  3722. template<typename TScalarGetter, typename TPointerGetter>
  3723. arrow::Datum DoMakePgScalar(const NPg::TTypeDesc& desc, arrow::MemoryPool& pool, const TScalarGetter& getScalar, const TPointerGetter& getPtr) {
  3724. if (desc.PassByValue) {
  3725. return arrow::MakeScalar(getScalar());
  3726. } else {
  3727. const char* ptr = getPtr();
  3728. ui32 size;
  3729. if (desc.TypeLen == -1) {
  3730. size = GetCleanVarSize((const text*)ptr) + VARHDRSZ;
  3731. } else if (desc.TypeLen == -2) {
  3732. size = strlen(ptr) + 1;
  3733. } else {
  3734. size = desc.TypeLen;
  3735. }
  3736. std::shared_ptr<arrow::Buffer> buffer(ARROW_RESULT(arrow::AllocateBuffer(size + sizeof(void*), &pool)));
  3737. NUdf::ZeroMemoryContext(buffer->mutable_data() + sizeof(void*));
  3738. std::memcpy(buffer->mutable_data() + sizeof(void*), ptr, size);
  3739. return arrow::Datum(std::make_shared<arrow::BinaryScalar>(buffer));
  3740. }
  3741. }
  3742. } // namespace
  3743. arrow::Datum MakePgScalar(NKikimr::NMiniKQL::TPgType* type, const NKikimr::NUdf::TUnboxedValuePod& value, arrow::MemoryPool& pool) {
  3744. return DoMakePgScalar(
  3745. NPg::LookupType(type->GetTypeId()), pool,
  3746. [&value]() { return (uint64_t)ScalarDatumFromPod(value); },
  3747. [&value]() { return (const char*)PointerDatumFromPod(value); }
  3748. );
  3749. }
  3750. arrow::Datum MakePgScalar(NKikimr::NMiniKQL::TPgType* type, const NUdf::TBlockItem& value, arrow::MemoryPool& pool) {
  3751. return DoMakePgScalar(
  3752. NPg::LookupType(type->GetTypeId()), pool,
  3753. [&value]() { return (uint64_t)ScalarDatumFromItem(value); },
  3754. [&value]() { return (const char*)PointerDatumFromItem(value); }
  3755. );
  3756. }
  3757. ui32 ConvertToPgType(NUdf::EDataSlot slot) {
  3758. switch (slot) {
  3759. case NUdf::EDataSlot::Bool:
  3760. return BOOLOID;
  3761. case NUdf::EDataSlot::Int8:
  3762. return INT2OID;
  3763. case NUdf::EDataSlot::Uint8:
  3764. return INT2OID;
  3765. case NUdf::EDataSlot::Int16:
  3766. return INT2OID;
  3767. case NUdf::EDataSlot::Uint16:
  3768. return INT4OID;
  3769. case NUdf::EDataSlot::Int32:
  3770. return INT4OID;
  3771. case NUdf::EDataSlot::Uint32:
  3772. return INT8OID;
  3773. case NUdf::EDataSlot::Int64:
  3774. return INT8OID;
  3775. case NUdf::EDataSlot::Uint64:
  3776. return NUMERICOID;
  3777. case NUdf::EDataSlot::Float:
  3778. return FLOAT4OID;
  3779. case NUdf::EDataSlot::Double:
  3780. return FLOAT8OID;
  3781. case NUdf::EDataSlot::String:
  3782. return BYTEAOID;
  3783. case NUdf::EDataSlot::Utf8:
  3784. return TEXTOID;
  3785. case NUdf::EDataSlot::Yson:
  3786. return BYTEAOID;
  3787. case NUdf::EDataSlot::Json:
  3788. return JSONOID;
  3789. case NUdf::EDataSlot::Uuid:
  3790. return UUIDOID;
  3791. case NUdf::EDataSlot::Date:
  3792. return DATEOID;
  3793. case NUdf::EDataSlot::Datetime:
  3794. return TIMESTAMPOID;
  3795. case NUdf::EDataSlot::Timestamp:
  3796. return TIMESTAMPOID;
  3797. case NUdf::EDataSlot::Interval:
  3798. return INTERVALOID;
  3799. case NUdf::EDataSlot::TzDate:
  3800. return TEXTOID;
  3801. case NUdf::EDataSlot::TzDatetime:
  3802. return TEXTOID;
  3803. case NUdf::EDataSlot::TzTimestamp:
  3804. return TEXTOID;
  3805. case NUdf::EDataSlot::Decimal:
  3806. return NUMERICOID;
  3807. case NUdf::EDataSlot::DyNumber:
  3808. return NUMERICOID;
  3809. case NUdf::EDataSlot::JsonDocument:
  3810. return JSONBOID;
  3811. case NUdf::EDataSlot::Date32:
  3812. return DATEOID;
  3813. case NUdf::EDataSlot::Datetime64:
  3814. return TIMESTAMPOID;
  3815. case NUdf::EDataSlot::Timestamp64:
  3816. return TIMESTAMPOID;
  3817. case NUdf::EDataSlot::Interval64:
  3818. return INTERVALOID;
  3819. case NUdf::EDataSlot::TzDate32:
  3820. return TEXTOID;
  3821. case NUdf::EDataSlot::TzDatetime64:
  3822. return TEXTOID;
  3823. case NUdf::EDataSlot::TzTimestamp64:
  3824. return TEXTOID;
  3825. }
  3826. }
  3827. TMaybe<NUdf::EDataSlot> ConvertFromPgType(ui32 typeId) {
  3828. switch (typeId) {
  3829. case BOOLOID:
  3830. return NUdf::EDataSlot::Bool;
  3831. case INT2OID:
  3832. return NUdf::EDataSlot::Int16;
  3833. case INT4OID:
  3834. return NUdf::EDataSlot::Int32;
  3835. case INT8OID:
  3836. return NUdf::EDataSlot::Int64;
  3837. case FLOAT4OID:
  3838. return NUdf::EDataSlot::Float;
  3839. case FLOAT8OID:
  3840. return NUdf::EDataSlot::Double;
  3841. case BYTEAOID:
  3842. return NUdf::EDataSlot::String;
  3843. case TEXTOID:
  3844. case VARCHAROID:
  3845. case CSTRINGOID:
  3846. return NUdf::EDataSlot::Utf8;
  3847. case DATEOID:
  3848. return NUdf::EDataSlot::Date32;
  3849. case TIMESTAMPOID:
  3850. return NUdf::EDataSlot::Timestamp64;
  3851. case UUIDOID:
  3852. return NUdf::EDataSlot::Uuid;
  3853. }
  3854. return Nothing();
  3855. }
  3856. bool ParsePgIntervalModifier(const TString& str, i32& ret) {
  3857. auto ustr = to_upper(str);
  3858. if (ustr == "YEAR") {
  3859. ret = INTERVAL_MASK(YEAR);
  3860. } else if (ustr == "MONTH") {
  3861. ret = INTERVAL_MASK(MONTH);
  3862. } else if (ustr == "DAY") {
  3863. ret = INTERVAL_MASK(DAY);
  3864. } else if (ustr == "HOUR") {
  3865. ret = INTERVAL_MASK(HOUR);
  3866. } else if (ustr == "MINUTE") {
  3867. ret = INTERVAL_MASK(MINUTE);
  3868. } else if (ustr == "SECOND") {
  3869. ret = INTERVAL_MASK(SECOND);
  3870. } else if (ustr == "YEAR TO MONTH") {
  3871. ret = INTERVAL_MASK(YEAR) | INTERVAL_MASK(MONTH);
  3872. } else if (ustr == "DAY TO HOUR") {
  3873. ret = INTERVAL_MASK(DAY) | INTERVAL_MASK(HOUR);
  3874. } else if (ustr == "DAY TO MINUTE") {
  3875. ret = INTERVAL_MASK(DAY) | INTERVAL_MASK(HOUR) | INTERVAL_MASK(MINUTE);
  3876. } else if (ustr == "DAY TO SECOND") {
  3877. ret = INTERVAL_MASK(DAY) | INTERVAL_MASK(HOUR) | INTERVAL_MASK(MINUTE) | INTERVAL_MASK(SECOND);
  3878. } else if (ustr == "HOUR TO MINUTE") {
  3879. ret = INTERVAL_MASK(HOUR) | INTERVAL_MASK(MINUTE);
  3880. } else if (ustr == "HOUR TO SECOND") {
  3881. ret = INTERVAL_MASK(HOUR) | INTERVAL_MASK(MINUTE) | INTERVAL_MASK(SECOND);
  3882. } else if (ustr == "MINUTE TO SECOND") {
  3883. ret = INTERVAL_MASK(MINUTE) | INTERVAL_MASK(SECOND);
  3884. } else {
  3885. return false;
  3886. }
  3887. return true;
  3888. }
  3889. template<typename TBuf>
  3890. void DoPGPack(bool stable, const TPgType* type, const NUdf::TUnboxedValuePod& value, TBuf& buf) {
  3891. switch (type->GetTypeId()) {
  3892. case BOOLOID: {
  3893. const auto x = DatumGetBool(ScalarDatumFromPod(value)) != 0;
  3894. NDetails::PutRawData(x, buf);
  3895. break;
  3896. }
  3897. case INT2OID: {
  3898. const auto x = DatumGetInt16(ScalarDatumFromPod(value));
  3899. NDetails::PackInt16(x, buf);
  3900. break;
  3901. }
  3902. case INT4OID: {
  3903. const auto x = DatumGetInt32(ScalarDatumFromPod(value));
  3904. NDetails::PackInt32(x, buf);
  3905. break;
  3906. }
  3907. case INT8OID: {
  3908. const auto x = DatumGetInt64(ScalarDatumFromPod(value));
  3909. NDetails::PackInt64(x, buf);
  3910. break;
  3911. }
  3912. case FLOAT4OID: {
  3913. auto x = DatumGetFloat4(ScalarDatumFromPod(value));
  3914. if (stable) {
  3915. NYql::CanonizeFpBits<float>(&x);
  3916. }
  3917. NDetails::PutRawData(x, buf);
  3918. break;
  3919. }
  3920. case FLOAT8OID: {
  3921. auto x = DatumGetFloat8(ScalarDatumFromPod(value));
  3922. if (stable) {
  3923. NYql::CanonizeFpBits<double>(&x);
  3924. }
  3925. NDetails::PutRawData(x, buf);
  3926. break;
  3927. }
  3928. case BYTEAOID:
  3929. case VARCHAROID:
  3930. case TEXTOID: {
  3931. const auto x = (const text*)PointerDatumFromPod(value);
  3932. auto s = GetVarBuf(x);
  3933. NDetails::PackUInt32(s.size(), buf);
  3934. buf.Append(s.data(), s.size());
  3935. break;
  3936. }
  3937. case CSTRINGOID: {
  3938. const auto x = (const char*)PointerDatumFromPod(value);
  3939. const auto len = strlen(x);
  3940. NDetails::PackUInt32(len, buf);
  3941. buf.Append(x, len);
  3942. break;
  3943. }
  3944. default:
  3945. NYql::NCommon::PgValueToNativeBinaryImpl(value, type->GetTypeId(), stable, [&buf](TStringBuf b) {
  3946. NDetails::PackUInt32(b.size(), buf);
  3947. buf.Append(b.data(), b.size());
  3948. });
  3949. }
  3950. }
  3951. } // NYql
  3952. namespace NKikimr {
  3953. namespace NMiniKQL {
  3954. using namespace NYql;
  3955. ui64 PgValueSize(const NUdf::TUnboxedValuePod& value, i32 typeLen) {
  3956. if (typeLen == -1) {
  3957. auto datum = PointerDatumFromPod(value);
  3958. const auto x = (const text*)PointerDatumFromPod(value);
  3959. return GetCleanVarSize(x);
  3960. } else if (typeLen == -2) {
  3961. auto datum = PointerDatumFromPod(value);
  3962. const auto x = (const char*)PointerDatumFromPod(value);
  3963. return strlen(x);
  3964. } else {
  3965. return typeLen;
  3966. }
  3967. }
  3968. ui64 PgValueSize(ui32 pgTypeId, const NUdf::TUnboxedValuePod& value) {
  3969. const auto& typeDesc = NYql::NPg::LookupType(pgTypeId);
  3970. return PgValueSize(value, typeDesc.TypeLen);
  3971. }
  3972. ui64 PgValueSize(const TPgType* type, const NUdf::TUnboxedValuePod& value) {
  3973. return PgValueSize(type->GetTypeId(), value);
  3974. }
  3975. void PGPackImpl(bool stable, const TPgType* type, const NUdf::TUnboxedValuePod& value, TBuffer& buf) {
  3976. DoPGPack(stable, type, value, buf);
  3977. }
  3978. void PGPackImpl(bool stable, const TPgType* type, const NUdf::TUnboxedValuePod& value, TPagedBuffer& buf) {
  3979. DoPGPack(stable, type, value, buf);
  3980. }
  3981. NUdf::TUnboxedValue PGUnpackImpl(const TPgType* type, TStringBuf& buf) {
  3982. NDetails::TChunkedInputBuffer chunked(buf);
  3983. return PGUnpackImpl(type, chunked);
  3984. }
  3985. NUdf::TUnboxedValue PGUnpackImpl(const TPgType* type, NDetails::TChunkedInputBuffer& buf) {
  3986. switch (type->GetTypeId()) {
  3987. case BOOLOID: {
  3988. const auto x = NDetails::GetRawData<bool>(buf);
  3989. return ScalarDatumToPod(BoolGetDatum(x));
  3990. }
  3991. case INT2OID: {
  3992. const auto x = NDetails::UnpackInt16(buf);
  3993. return ScalarDatumToPod(Int16GetDatum(x));
  3994. }
  3995. case INT4OID: {
  3996. const auto x = NDetails::UnpackInt32(buf);
  3997. return ScalarDatumToPod(Int32GetDatum(x));
  3998. }
  3999. case INT8OID: {
  4000. const auto x = NDetails::UnpackInt64(buf);
  4001. return ScalarDatumToPod(Int64GetDatum(x));
  4002. }
  4003. case FLOAT4OID: {
  4004. const auto x = NDetails::GetRawData<float>(buf);
  4005. return ScalarDatumToPod(Float4GetDatum(x));
  4006. }
  4007. case FLOAT8OID: {
  4008. const auto x = NDetails::GetRawData<double>(buf);
  4009. return ScalarDatumToPod(Float8GetDatum(x));
  4010. }
  4011. case BYTEAOID:
  4012. case VARCHAROID:
  4013. case TEXTOID: {
  4014. auto size = NDetails::UnpackUInt32(buf);
  4015. auto deleter = [](text* ptr) { pfree(ptr); };
  4016. std::unique_ptr<text, decltype(deleter)> ret(MakeVarNotFilled(size));
  4017. buf.CopyTo(GetMutableVarData(ret.get()), size);
  4018. return PointerDatumToPod((Datum)ret.release());
  4019. }
  4020. case CSTRINGOID: {
  4021. auto size = NDetails::UnpackUInt32(buf);
  4022. auto deleter = [](char* ptr) { pfree(ptr); };
  4023. std::unique_ptr<char, decltype(deleter)> ret(MakeCStringNotFilled(size));
  4024. buf.CopyTo(ret.get(), size);
  4025. return PointerDatumToPod((Datum)ret.release());
  4026. }
  4027. default:
  4028. TPAllocScope call;
  4029. auto size = NDetails::UnpackUInt32(buf);
  4030. std::unique_ptr<char[]> tmpBuf(new char[size]);
  4031. buf.CopyTo(tmpBuf.get(), size);
  4032. TStringBuf s{tmpBuf.get(), size};
  4033. return NYql::NCommon::PgValueFromNativeBinary(s, type->GetTypeId());
  4034. }
  4035. }
  4036. void EncodePresortPGValue(TPgType* type, const NUdf::TUnboxedValue& value, TVector<ui8>& output) {
  4037. switch (type->GetTypeId()) {
  4038. case BOOLOID: {
  4039. const auto x = DatumGetBool(ScalarDatumFromPod(value)) != 0;
  4040. NDetail::EncodeBool<false>(output, x);
  4041. break;
  4042. }
  4043. case INT2OID: {
  4044. const auto x = DatumGetInt16(ScalarDatumFromPod(value));
  4045. NDetail::EncodeSigned<i16, false>(output, x);
  4046. break;
  4047. }
  4048. case INT4OID: {
  4049. const auto x = DatumGetInt32(ScalarDatumFromPod(value));
  4050. NDetail::EncodeSigned<i32, false>(output, x);
  4051. break;
  4052. }
  4053. case INT8OID: {
  4054. const auto x = DatumGetInt64(ScalarDatumFromPod(value));
  4055. NDetail::EncodeSigned<i64, false>(output, x);
  4056. break;
  4057. }
  4058. case FLOAT4OID: {
  4059. const auto x = DatumGetFloat4(ScalarDatumFromPod(value));
  4060. NDetail::EncodeFloating<float, false>(output, x);
  4061. break;
  4062. }
  4063. case FLOAT8OID: {
  4064. const auto x = DatumGetFloat8(ScalarDatumFromPod(value));
  4065. NDetail::EncodeFloating<double, false>(output, x);
  4066. break;
  4067. }
  4068. case BYTEAOID:
  4069. case VARCHAROID:
  4070. case TEXTOID: {
  4071. const auto x = (const text*)PointerDatumFromPod(value);
  4072. auto s = GetVarBuf(x);
  4073. NDetail::EncodeString<false>(output, s);
  4074. break;
  4075. }
  4076. case CSTRINGOID: {
  4077. const auto x = (const char*)PointerDatumFromPod(value);
  4078. NDetail::EncodeString<false>(output, x);
  4079. break;
  4080. }
  4081. default:
  4082. NYql::NCommon::PgValueToNativeBinaryImpl(value, type->GetTypeId(), true, [&output](TStringBuf b) {
  4083. NDetail::EncodeString<false>(output, b);
  4084. });
  4085. }
  4086. }
  4087. NUdf::TUnboxedValue DecodePresortPGValue(TPgType* type, TStringBuf& input, TVector<ui8>& buffer) {
  4088. switch (type->GetTypeId()) {
  4089. case BOOLOID: {
  4090. const auto x = NDetail::DecodeBool<false>(input);
  4091. return ScalarDatumToPod(BoolGetDatum(x));
  4092. }
  4093. case INT2OID: {
  4094. const auto x = NDetail::DecodeSigned<i16, false>(input);
  4095. return ScalarDatumToPod(Int16GetDatum(x));
  4096. }
  4097. case INT4OID: {
  4098. const auto x = NDetail::DecodeSigned<i32, false>(input);
  4099. return ScalarDatumToPod(Int32GetDatum(x));
  4100. }
  4101. case INT8OID: {
  4102. const auto x = NDetail::DecodeSigned<i64, false>(input);
  4103. return ScalarDatumToPod(Int64GetDatum(x));
  4104. }
  4105. case FLOAT4OID: {
  4106. const auto x = NDetail::DecodeFloating<float, false>(input);
  4107. return ScalarDatumToPod(Float4GetDatum(x));
  4108. }
  4109. case FLOAT8OID: {
  4110. const auto x = NDetail::DecodeFloating<double, false>(input);
  4111. return ScalarDatumToPod(Float8GetDatum(x));
  4112. }
  4113. case BYTEAOID:
  4114. case VARCHAROID:
  4115. case TEXTOID: {
  4116. buffer.clear();
  4117. const auto s = NDetail::DecodeString<false>(input, buffer);
  4118. auto ret = MakeVar(s);
  4119. return PointerDatumToPod((Datum)ret);
  4120. }
  4121. case CSTRINGOID: {
  4122. buffer.clear();
  4123. const auto s = NDetail::DecodeString<false>(input, buffer);
  4124. auto ret = MakeCString(s);
  4125. return PointerDatumToPod((Datum)ret);
  4126. }
  4127. default:
  4128. buffer.clear();
  4129. const auto s = NDetail::DecodeString<false>(input, buffer);
  4130. return NYql::NCommon::PgValueFromNativeBinary(s, type->GetTypeId());
  4131. }
  4132. }
  4133. void* PgInitializeContext(const std::string_view& contextType) {
  4134. if (contextType == "Agg") {
  4135. auto ctx = (AggState*)TWithDefaultMiniKQLAlloc::AllocWithSize(sizeof(AggState));
  4136. Zero(*ctx);
  4137. *(NodeTag*)ctx = T_AggState;
  4138. ctx->curaggcontext = (ExprContext*)TWithDefaultMiniKQLAlloc::AllocWithSize(sizeof(ExprContext));
  4139. Zero(*ctx->curaggcontext);
  4140. ctx->curaggcontext->ecxt_per_tuple_memory = (MemoryContext)&((TMainContext*)TlsAllocState->MainContext)->Data;
  4141. return ctx;
  4142. } else if (contextType == "WinAgg") {
  4143. auto ctx = (WindowAggState*)TWithDefaultMiniKQLAlloc::AllocWithSize(sizeof(WindowAggState));
  4144. Zero(*ctx);
  4145. *(NodeTag*)ctx = T_WindowAggState;
  4146. ctx->curaggcontext = (MemoryContext)&((TMainContext*)TlsAllocState->MainContext)->Data;
  4147. return ctx;
  4148. } else {
  4149. ythrow yexception() << "Unsupported context type: " << contextType;
  4150. }
  4151. }
  4152. void PgDestroyContext(const std::string_view& contextType, void* ctx) {
  4153. if (contextType == "Agg") {
  4154. TWithDefaultMiniKQLAlloc::FreeWithSize(((AggState*)ctx)->curaggcontext, sizeof(ExprContext));
  4155. TWithDefaultMiniKQLAlloc::FreeWithSize(ctx, sizeof(AggState));
  4156. } else if (contextType == "WinAgg") {
  4157. TWithDefaultMiniKQLAlloc::FreeWithSize(ctx, sizeof(WindowAggState));
  4158. } else {
  4159. Y_ABORT("Unsupported context type");
  4160. }
  4161. }
  4162. template <bool PassByValue, bool IsArray>
  4163. class TPgHashBase {
  4164. public:
  4165. TPgHashBase(const NYql::NPg::TTypeDesc& typeDesc)
  4166. : TypeDesc(typeDesc)
  4167. {
  4168. auto hashProcId = TypeDesc.HashProcId;
  4169. if constexpr (IsArray) {
  4170. const auto& elemDesc = NYql::NPg::LookupType(TypeDesc.ElementTypeId);
  4171. Y_ENSURE(elemDesc.HashProcId);
  4172. hashProcId = NYql::NPg::LookupProc("hash_array", { 0, 0 }).ProcId;
  4173. }
  4174. Y_ENSURE(hashProcId);;
  4175. Zero(FInfoHash);
  4176. GetPgFuncAddr(hashProcId, FInfoHash);
  4177. Y_ENSURE(!FInfoHash.fn_retset);
  4178. Y_ENSURE(FInfoHash.fn_addr);
  4179. Y_ENSURE(FInfoHash.fn_nargs == 1);
  4180. }
  4181. protected:
  4182. const NYql::NPg::TTypeDesc TypeDesc;
  4183. FmgrInfo FInfoHash;
  4184. };
  4185. template <bool PassByValue, bool IsArray>
  4186. class TPgHash : public TPgHashBase<PassByValue, IsArray>, public NUdf::IHash {
  4187. public:
  4188. using TBase = TPgHashBase<PassByValue, IsArray>;
  4189. TPgHash(const NYql::NPg::TTypeDesc& typeDesc)
  4190. : TBase(typeDesc)
  4191. {}
  4192. ui64 Hash(NUdf::TUnboxedValuePod lhs) const override {
  4193. LOCAL_FCINFO(callInfo, 1);
  4194. Zero(*callInfo);
  4195. callInfo->flinfo = const_cast<FmgrInfo*>(&this->FInfoHash); // don't copy becase of IHash isn't threadsafe
  4196. callInfo->nargs = 1;
  4197. callInfo->fncollation = DEFAULT_COLLATION_OID;
  4198. callInfo->isnull = false;
  4199. if (!lhs) {
  4200. return 0;
  4201. }
  4202. callInfo->args[0] = { PassByValue ?
  4203. ScalarDatumFromPod(lhs) :
  4204. PointerDatumFromPod(lhs), false };
  4205. auto x = this->FInfoHash.fn_addr(callInfo);
  4206. Y_ENSURE(!callInfo->isnull);
  4207. return DatumGetUInt32(x);
  4208. }
  4209. };
  4210. template <bool PassByValue, bool IsArray>
  4211. class TPgHashItem : public TPgHashBase<PassByValue, IsArray>, public NUdf::TBlockItemHasherBase<TPgHashItem<PassByValue, IsArray>, true> {
  4212. public:
  4213. using TBase = TPgHashBase<PassByValue, IsArray>;
  4214. TPgHashItem(const NYql::NPg::TTypeDesc& typeDesc)
  4215. : TBase(typeDesc)
  4216. {}
  4217. ui64 DoHash(NUdf::TBlockItem value) const {
  4218. LOCAL_FCINFO(callInfo, 1);
  4219. Zero(*callInfo);
  4220. callInfo->flinfo = const_cast<FmgrInfo*>(&this->FInfoHash); // don't copy becase of IHash isn't threadsafe
  4221. callInfo->nargs = 1;
  4222. callInfo->fncollation = DEFAULT_COLLATION_OID;
  4223. callInfo->isnull = false;
  4224. callInfo->args[0] = { PassByValue ?
  4225. ScalarDatumFromItem(value) :
  4226. PointerDatumFromItem(value), false };
  4227. auto x = this->FInfoHash.fn_addr(callInfo);
  4228. Y_ENSURE(!callInfo->isnull);
  4229. return DatumGetUInt32(x);
  4230. }
  4231. };
  4232. NUdf::IHash::TPtr MakePgHash(const NMiniKQL::TPgType* type) {
  4233. const auto& typeDesc = NYql::NPg::LookupType(type->GetTypeId());
  4234. if (typeDesc.PassByValue) {
  4235. return new TPgHash<true, false>(typeDesc);
  4236. } else if (typeDesc.TypeId == typeDesc.ArrayTypeId) {
  4237. return new TPgHash<false, true>(typeDesc);
  4238. } else {
  4239. return new TPgHash<false, false>(typeDesc);
  4240. }
  4241. }
  4242. NUdf::IBlockItemHasher::TPtr MakePgItemHasher(ui32 typeId) {
  4243. const auto& typeDesc = NYql::NPg::LookupType(typeId);
  4244. if (typeDesc.PassByValue) {
  4245. return new TPgHashItem<true, false>(typeDesc);
  4246. } else if (typeDesc.TypeId == typeDesc.ArrayTypeId) {
  4247. return new TPgHashItem<false, true>(typeDesc);
  4248. } else {
  4249. return new TPgHashItem<false, false>(typeDesc);
  4250. }
  4251. }
  4252. template <bool PassByValue, bool IsArray>
  4253. class TPgCompareBase {
  4254. public:
  4255. TPgCompareBase(const NYql::NPg::TTypeDesc& typeDesc)
  4256. : TypeDesc(typeDesc)
  4257. {
  4258. Zero(FInfoLess);
  4259. Zero(FInfoCompare);
  4260. Zero(FInfoEquals);
  4261. auto lessProcId = TypeDesc.LessProcId;
  4262. auto compareProcId = TypeDesc.CompareProcId;
  4263. auto equalProcId = TypeDesc.EqualProcId;
  4264. if constexpr (IsArray) {
  4265. const auto& elemDesc = NYql::NPg::LookupType(TypeDesc.ElementTypeId);
  4266. Y_ENSURE(elemDesc.CompareProcId);
  4267. compareProcId = NYql::NPg::LookupProc("btarraycmp", { 0, 0 }).ProcId;
  4268. } else {
  4269. Y_ENSURE(lessProcId);
  4270. Y_ENSURE(equalProcId);
  4271. GetPgFuncAddr(lessProcId, FInfoLess);
  4272. Y_ENSURE(!FInfoLess.fn_retset);
  4273. Y_ENSURE(FInfoLess.fn_addr);
  4274. Y_ENSURE(FInfoLess.fn_nargs == 2);
  4275. GetPgFuncAddr(equalProcId, FInfoEquals);
  4276. Y_ENSURE(!FInfoEquals.fn_retset);
  4277. Y_ENSURE(FInfoEquals.fn_addr);
  4278. Y_ENSURE(FInfoEquals.fn_nargs == 2);
  4279. }
  4280. Y_ENSURE(compareProcId);
  4281. GetPgFuncAddr(compareProcId, FInfoCompare);
  4282. Y_ENSURE(!FInfoCompare.fn_retset);
  4283. Y_ENSURE(FInfoCompare.fn_addr);
  4284. Y_ENSURE(FInfoCompare.fn_nargs == 2);
  4285. }
  4286. protected:
  4287. const NYql::NPg::TTypeDesc TypeDesc;
  4288. FmgrInfo FInfoLess, FInfoCompare, FInfoEquals;
  4289. };
  4290. template <bool PassByValue, bool IsArray>
  4291. class TPgCompare : public TPgCompareBase<PassByValue, IsArray>, public NUdf::ICompare {
  4292. public:
  4293. using TBase = TPgCompareBase<PassByValue, IsArray>;
  4294. TPgCompare(const NYql::NPg::TTypeDesc& typeDesc)
  4295. : TBase(typeDesc)
  4296. {}
  4297. bool Less(NUdf::TUnboxedValuePod lhs, NUdf::TUnboxedValuePod rhs) const override {
  4298. if constexpr (IsArray) {
  4299. return Compare(lhs, rhs) < 0;
  4300. }
  4301. LOCAL_FCINFO(callInfo, 2);
  4302. Zero(*callInfo);
  4303. callInfo->flinfo = const_cast<FmgrInfo*>(&this->FInfoLess); // don't copy becase of ICompare isn't threadsafe
  4304. callInfo->nargs = 2;
  4305. callInfo->fncollation = DEFAULT_COLLATION_OID;
  4306. callInfo->isnull = false;
  4307. if (!lhs) {
  4308. if (!rhs) {
  4309. return false;
  4310. }
  4311. return true;
  4312. }
  4313. if (!rhs) {
  4314. return false;
  4315. }
  4316. callInfo->args[0] = { PassByValue ?
  4317. ScalarDatumFromPod(lhs) :
  4318. PointerDatumFromPod(lhs), false };
  4319. callInfo->args[1] = { PassByValue ?
  4320. ScalarDatumFromPod(rhs) :
  4321. PointerDatumFromPod(rhs), false };
  4322. auto x = this->FInfoLess.fn_addr(callInfo);
  4323. Y_ENSURE(!callInfo->isnull);
  4324. return DatumGetBool(x);
  4325. }
  4326. int Compare(NUdf::TUnboxedValuePod lhs, NUdf::TUnboxedValuePod rhs) const override {
  4327. LOCAL_FCINFO(callInfo, 2);
  4328. Zero(*callInfo);
  4329. callInfo->flinfo = const_cast<FmgrInfo*>(&this->FInfoCompare); // don't copy becase of ICompare isn't threadsafe
  4330. callInfo->nargs = 2;
  4331. callInfo->fncollation = DEFAULT_COLLATION_OID;
  4332. callInfo->isnull = false;
  4333. if (!lhs) {
  4334. if (!rhs) {
  4335. return 0;
  4336. }
  4337. return -1;
  4338. }
  4339. if (!rhs) {
  4340. return 1;
  4341. }
  4342. callInfo->args[0] = { PassByValue ?
  4343. ScalarDatumFromPod(lhs) :
  4344. PointerDatumFromPod(lhs), false };
  4345. callInfo->args[1] = { PassByValue ?
  4346. ScalarDatumFromPod(rhs) :
  4347. PointerDatumFromPod(rhs), false };
  4348. auto x = this->FInfoCompare.fn_addr(callInfo);
  4349. Y_ENSURE(!callInfo->isnull);
  4350. return DatumGetInt32(x);
  4351. }
  4352. };
  4353. template <bool PassByValue, bool IsArray>
  4354. class TPgCompareItem : public TPgCompareBase<PassByValue, IsArray>, public NUdf::TBlockItemComparatorBase<TPgCompareItem<PassByValue, IsArray>, true> {
  4355. public:
  4356. using TBase = TPgCompareBase<PassByValue, IsArray>;
  4357. TPgCompareItem(const NYql::NPg::TTypeDesc& typeDesc)
  4358. : TBase(typeDesc)
  4359. {}
  4360. i64 DoCompare(NUdf::TBlockItem lhs, NUdf::TBlockItem rhs) const {
  4361. LOCAL_FCINFO(callInfo, 2);
  4362. Zero(*callInfo);
  4363. callInfo->flinfo = const_cast<FmgrInfo*>(&this->FInfoCompare); // don't copy becase of ICompare isn't threadsafe
  4364. callInfo->nargs = 2;
  4365. callInfo->fncollation = DEFAULT_COLLATION_OID;
  4366. callInfo->isnull = false;
  4367. callInfo->args[0] = { PassByValue ?
  4368. ScalarDatumFromItem(lhs) :
  4369. PointerDatumFromItem(lhs), false };
  4370. callInfo->args[1] = { PassByValue ?
  4371. ScalarDatumFromItem(rhs) :
  4372. PointerDatumFromItem(rhs), false };
  4373. auto x = this->FInfoCompare.fn_addr(callInfo);
  4374. Y_ENSURE(!callInfo->isnull);
  4375. return DatumGetInt32(x);
  4376. }
  4377. bool DoEquals(NUdf::TBlockItem lhs, NUdf::TBlockItem rhs) const {
  4378. if constexpr (IsArray) {
  4379. return DoCompare(lhs, rhs) == 0;
  4380. }
  4381. LOCAL_FCINFO(callInfo, 2);
  4382. Zero(*callInfo);
  4383. callInfo->flinfo = const_cast<FmgrInfo*>(&this->FInfoEquals); // don't copy becase of ICompare isn't threadsafe
  4384. callInfo->nargs = 2;
  4385. callInfo->fncollation = DEFAULT_COLLATION_OID;
  4386. callInfo->isnull = false;
  4387. callInfo->args[0] = { PassByValue ?
  4388. ScalarDatumFromItem(lhs) :
  4389. PointerDatumFromItem(lhs), false };
  4390. callInfo->args[1] = { PassByValue ?
  4391. ScalarDatumFromItem(rhs) :
  4392. PointerDatumFromItem(rhs), false };
  4393. auto x = this->FInfoEquals.fn_addr(callInfo);
  4394. Y_ENSURE(!callInfo->isnull);
  4395. return DatumGetBool(x);
  4396. }
  4397. bool DoLess(NUdf::TBlockItem lhs, NUdf::TBlockItem rhs) const {
  4398. if constexpr (IsArray) {
  4399. return DoCompare(lhs, rhs) < 0;
  4400. }
  4401. LOCAL_FCINFO(callInfo, 2);
  4402. Zero(*callInfo);
  4403. callInfo->flinfo = const_cast<FmgrInfo*>(&this->FInfoLess); // don't copy becase of ICompare isn't threadsafe
  4404. callInfo->nargs = 2;
  4405. callInfo->fncollation = DEFAULT_COLLATION_OID;
  4406. callInfo->isnull = false;
  4407. callInfo->args[0] = { PassByValue ?
  4408. ScalarDatumFromItem(lhs) :
  4409. PointerDatumFromItem(lhs), false };
  4410. callInfo->args[1] = { PassByValue ?
  4411. ScalarDatumFromItem(rhs) :
  4412. PointerDatumFromItem(rhs), false };
  4413. auto x = this->FInfoLess.fn_addr(callInfo);
  4414. Y_ENSURE(!callInfo->isnull);
  4415. return DatumGetBool(x);
  4416. }
  4417. };
  4418. NUdf::ICompare::TPtr MakePgCompare(const NMiniKQL::TPgType* type) {
  4419. const auto& typeDesc = NYql::NPg::LookupType(type->GetTypeId());
  4420. if (typeDesc.PassByValue) {
  4421. return new TPgCompare<true, false>(typeDesc);
  4422. } else if (typeDesc.TypeId == typeDesc.ArrayTypeId) {
  4423. return new TPgCompare<false, true>(typeDesc);
  4424. } else {
  4425. return new TPgCompare<false, false>(typeDesc);
  4426. }
  4427. }
  4428. NUdf::IBlockItemComparator::TPtr MakePgItemComparator(ui32 typeId) {
  4429. const auto& typeDesc = NYql::NPg::LookupType(typeId);
  4430. if (typeDesc.PassByValue) {
  4431. return new TPgCompareItem<true, false>(typeDesc);
  4432. } else if (typeDesc.TypeId == typeDesc.ArrayTypeId) {
  4433. return new TPgCompareItem<false, true>(typeDesc);
  4434. } else {
  4435. return new TPgCompareItem<false, false>(typeDesc);
  4436. }
  4437. }
  4438. template <bool PassByValue, bool IsArray>
  4439. class TPgEquate: public NUdf::IEquate {
  4440. public:
  4441. TPgEquate(const NYql::NPg::TTypeDesc& typeDesc)
  4442. : TypeDesc(typeDesc)
  4443. {
  4444. auto equalProcId = TypeDesc.EqualProcId;
  4445. if constexpr (IsArray) {
  4446. const auto& elemDesc = NYql::NPg::LookupType(TypeDesc.ElementTypeId);
  4447. Y_ENSURE(elemDesc.CompareProcId);
  4448. equalProcId = NYql::NPg::LookupProc("btarraycmp", { 0, 0 }).ProcId;
  4449. }
  4450. Y_ENSURE(equalProcId);
  4451. Zero(FInfoEquate);
  4452. GetPgFuncAddr(equalProcId, FInfoEquate);
  4453. Y_ENSURE(!FInfoEquate.fn_retset);
  4454. Y_ENSURE(FInfoEquate.fn_addr);
  4455. Y_ENSURE(FInfoEquate.fn_nargs == 2);
  4456. }
  4457. bool Equals(NUdf::TUnboxedValuePod lhs, NUdf::TUnboxedValuePod rhs) const override {
  4458. LOCAL_FCINFO(callInfo, 2);
  4459. Zero(*callInfo);
  4460. callInfo->flinfo = const_cast<FmgrInfo*>(&FInfoEquate); // don't copy becase of IEquate isn't threadsafe
  4461. callInfo->nargs = 2;
  4462. callInfo->fncollation = DEFAULT_COLLATION_OID;
  4463. callInfo->isnull = false;
  4464. if (!lhs) {
  4465. if (!rhs) {
  4466. return true;
  4467. }
  4468. return false;
  4469. }
  4470. if (!rhs) {
  4471. return false;
  4472. }
  4473. callInfo->args[0] = { PassByValue ?
  4474. ScalarDatumFromPod(lhs) :
  4475. PointerDatumFromPod(lhs), false };
  4476. callInfo->args[1] = { PassByValue ?
  4477. ScalarDatumFromPod(rhs) :
  4478. PointerDatumFromPod(rhs), false };
  4479. auto x = FInfoEquate.fn_addr(callInfo);
  4480. Y_ENSURE(!callInfo->isnull);
  4481. if constexpr (IsArray) {
  4482. return DatumGetInt32(x) == 0;
  4483. }
  4484. return DatumGetBool(x);
  4485. }
  4486. private:
  4487. const NYql::NPg::TTypeDesc TypeDesc;
  4488. FmgrInfo FInfoEquate;
  4489. };
  4490. NUdf::IEquate::TPtr MakePgEquate(const TPgType* type) {
  4491. const auto& typeDesc = NYql::NPg::LookupType(type->GetTypeId());
  4492. if (typeDesc.PassByValue) {
  4493. return new TPgEquate<true, false>(typeDesc);
  4494. } else if (typeDesc.TypeId == typeDesc.ArrayTypeId) {
  4495. return new TPgEquate<false, true>(typeDesc);
  4496. } else {
  4497. return new TPgEquate<false, false>(typeDesc);
  4498. }
  4499. }
  4500. void* PgInitializeMainContext() {
  4501. auto ctx = new TMainContext();
  4502. static_assert(MEMORY_CONTEXT_METHODID_MASK < alignof(decltype(TMainContext::Data)));
  4503. MemoryContextCreate((MemoryContext)&ctx->Data,
  4504. T_AllocSetContext,
  4505. MCTX_UNUSED3_ID,
  4506. nullptr,
  4507. "mkql");
  4508. static_assert(MEMORY_CONTEXT_METHODID_MASK < alignof(decltype(TMainContext::ErrorData)));
  4509. MemoryContextCreate((MemoryContext)&ctx->ErrorData,
  4510. T_AllocSetContext,
  4511. MCTX_UNUSED3_ID,
  4512. nullptr,
  4513. "mkql-err");
  4514. ctx->StartTimestamp = GetCurrentTimestamp();
  4515. return ctx;
  4516. }
  4517. void PgDestroyMainContext(void* ctx) {
  4518. auto typedCtx = (TMainContext*)ctx;
  4519. MemoryContextDeleteChildren((MemoryContext)&typedCtx->Data);
  4520. MemoryContextDeleteChildren((MemoryContext)&typedCtx->ErrorData);
  4521. delete typedCtx;
  4522. }
  4523. void PgAcquireThreadContext(void* ctx) {
  4524. if (ctx) {
  4525. pg_thread_init();
  4526. TExtensionsRegistry::Instance().InitThread();
  4527. auto main = (TMainContext*)ctx;
  4528. main->PrevCurrentMemoryContext = CurrentMemoryContext;
  4529. main->PrevErrorContext = ErrorContext;
  4530. main->PrevCacheMemoryContext = CacheMemoryContext;
  4531. SaveRecordCacheState(&main->PrevRecordCacheState);
  4532. LoadRecordCacheState(&main->CurrentRecordCacheState);
  4533. CurrentMemoryContext = CacheMemoryContext = (MemoryContext)&main->Data;
  4534. ErrorContext = (MemoryContext)&main->ErrorData;
  4535. SetParallelStartTimestamps(main->StartTimestamp, main->StartTimestamp);
  4536. main->PrevStackBase = set_stack_base();
  4537. yql_error_report_active = true;
  4538. if (main->GUCSettings && main->GUCSettings->Get("ydb_database")) {
  4539. MyDatabaseId = PG_CURRENT_DATABASE_ID;
  4540. }
  4541. }
  4542. }
  4543. void PgReleaseThreadContext(void* ctx) {
  4544. if (ctx) {
  4545. auto main = (TMainContext*)ctx;
  4546. CurrentMemoryContext = main->PrevCurrentMemoryContext;
  4547. ErrorContext = main->PrevErrorContext;
  4548. CacheMemoryContext = main->PrevCacheMemoryContext;
  4549. SaveRecordCacheState(&main->CurrentRecordCacheState);
  4550. LoadRecordCacheState(&main->PrevRecordCacheState);
  4551. restore_stack_base(main->PrevStackBase);
  4552. yql_error_report_active = false;
  4553. MyDatabaseId = PG_POSTGRES_DATABASE_ID;
  4554. }
  4555. }
  4556. class TExtensionLoader : public NYql::NPg::IExtensionLoader {
  4557. public:
  4558. void Load(ui32 extensionIndex, const TString& name, const TString& path) final {
  4559. RebuildSysCache();
  4560. TExtensionsRegistry::Instance().Load(extensionIndex, name, path);
  4561. }
  4562. };
  4563. std::unique_ptr<NYql::NPg::IExtensionLoader> CreateExtensionLoader() {
  4564. return std::make_unique<TExtensionLoader>();
  4565. }
  4566. void PgSetGUCSettings(void* ctx, const TGUCSettings::TPtr& GUCSettings) {
  4567. if (ctx && GUCSettings) {
  4568. auto main = (TMainContext*)ctx;
  4569. main->GUCSettings = GUCSettings;
  4570. if (main->GUCSettings->Get("ydb_database")) {
  4571. MyDatabaseId = PG_CURRENT_DATABASE_ID;
  4572. }
  4573. }
  4574. PgCreateSysCacheEntries(ctx);
  4575. }
  4576. std::optional<std::string> PGGetGUCSetting(const std::string& key) {
  4577. if (TlsAllocState) {
  4578. auto ctx = (TMainContext*)TlsAllocState->MainContext;
  4579. if (ctx && ctx->GUCSettings) {
  4580. return ctx->GUCSettings->Get(key);
  4581. }
  4582. }
  4583. return std::nullopt;
  4584. }
  4585. extern "C" void yql_prepare_error(const char* msg) {
  4586. auto ctx = (TMainContext*)TlsAllocState->MainContext;
  4587. ctx->LastError = msg;
  4588. }
  4589. extern "C" void yql_raise_error() {
  4590. auto ctx = (TMainContext*)TlsAllocState->MainContext;
  4591. UdfTerminate(ctx->LastError.c_str());
  4592. }
  4593. } // namespace NMiniKQL
  4594. } // namespace NKikimr
  4595. namespace NYql {
  4596. class TPgBuilderImpl : public NUdf::IPgBuilder {
  4597. public:
  4598. NUdf::TUnboxedValue ValueFromText(ui32 typeId, const NUdf::TStringRef& value, NUdf::TStringValue& error) const override {
  4599. try {
  4600. return NCommon::PgValueFromNativeText(static_cast<TStringBuf>(value), typeId);
  4601. } catch (const std::exception& e) {
  4602. error = NUdf::TStringValue(TStringBuf(e.what()));
  4603. }
  4604. return NUdf::TUnboxedValue();
  4605. }
  4606. NUdf::TUnboxedValue ValueFromBinary(ui32 typeId, const NUdf::TStringRef& value, NUdf::TStringValue& error) const override {
  4607. try {
  4608. return NCommon::PgValueFromNativeBinary(static_cast<TStringBuf>(value), typeId);
  4609. } catch (const std::exception& e) {
  4610. error = NUdf::TStringValue(TStringBuf(e.what()));
  4611. }
  4612. return NUdf::TUnboxedValue();
  4613. }
  4614. NUdf::TUnboxedValue ConvertFromPg(NUdf::TUnboxedValue source, ui32 sourceTypeId, const NUdf::TType* targetType) const override {
  4615. auto t = static_cast<const NKikimr::NMiniKQL::TType*>(targetType);
  4616. return ConvertFromPgValue(source, sourceTypeId, const_cast<NKikimr::NMiniKQL::TType*>(t));
  4617. }
  4618. NUdf::TUnboxedValue ConvertToPg(NUdf::TUnboxedValue source, const NUdf::TType* sourceType, ui32 targetTypeId) const override {
  4619. auto t = static_cast<const NKikimr::NMiniKQL::TType*>(sourceType);
  4620. return ConvertToPgValue(source, const_cast<NKikimr::NMiniKQL::TType*>(t), targetTypeId);
  4621. }
  4622. NUdf::TUnboxedValue NewString(i32 typeLen, ui32 targetTypeId, NUdf::TStringRef data) const override {
  4623. return CreatePgString(typeLen, targetTypeId, data);
  4624. }
  4625. NUdf::TStringRef AsCStringBuffer(const NUdf::TUnboxedValue& value) const override {
  4626. auto x = (const char*)PointerDatumFromPod(value);
  4627. return { x, ui32(strlen(x) + 1)};
  4628. }
  4629. NUdf::TStringRef AsTextBuffer(const NUdf::TUnboxedValue& value) const override {
  4630. auto x = (const text*)PointerDatumFromPod(value);
  4631. return { (const char*)x, GetFullVarSize(x) };
  4632. }
  4633. NUdf::TUnboxedValue MakeCString(const char* value) const override {
  4634. auto len = 1 + strlen(value);
  4635. char* ret = (char*)palloc(len);
  4636. memcpy(ret, value, len);
  4637. return PointerDatumToPod((Datum)ret);
  4638. }
  4639. NUdf::TUnboxedValue MakeText(const char* value) const override {
  4640. auto len = GetFullVarSize((const text*)value);
  4641. char* ret = (char*)palloc(len);
  4642. memcpy(ret, value, len);
  4643. return PointerDatumToPod((Datum)ret);
  4644. }
  4645. NUdf::TStringRef AsFixedStringBuffer(const NUdf::TUnboxedValue& value, ui32 length) const override {
  4646. auto x = (const char*)PointerDatumFromPod(value);
  4647. return { x, length };
  4648. }
  4649. };
  4650. std::unique_ptr<NUdf::IPgBuilder> CreatePgBuilder() {
  4651. return std::make_unique<TPgBuilderImpl>();
  4652. }
  4653. } // namespace NYql
  4654. extern "C" {
  4655. void yql_canonize_float4(float4* x) {
  4656. if (NYql::NeedCanonizeFp) {
  4657. NYql::CanonizeFpBits<float>(x);
  4658. }
  4659. }
  4660. extern void yql_canonize_float8(float8* x) {
  4661. if (NYql::NeedCanonizeFp) {
  4662. NYql::CanonizeFpBits<double>(x);
  4663. }
  4664. }
  4665. void get_type_io_data(Oid typid,
  4666. IOFuncSelector which_func,
  4667. int16 *typlen,
  4668. bool *typbyval,
  4669. char *typalign,
  4670. char *typdelim,
  4671. Oid *typioparam,
  4672. Oid *func) {
  4673. const auto& typeDesc = NYql::NPg::LookupType(typid);
  4674. *typlen = typeDesc.TypeLen;
  4675. *typbyval = typeDesc.PassByValue;
  4676. *typalign = typeDesc.TypeAlign;
  4677. *typdelim = typeDesc.TypeDelim;
  4678. *typioparam = NYql::MakeTypeIOParam(typeDesc);
  4679. switch (which_func) {
  4680. case IOFunc_input:
  4681. *func = typeDesc.InFuncId;
  4682. break;
  4683. case IOFunc_output:
  4684. *func = typeDesc.OutFuncId;
  4685. break;
  4686. case IOFunc_receive:
  4687. *func = typeDesc.ReceiveFuncId;
  4688. break;
  4689. case IOFunc_send:
  4690. *func = typeDesc.SendFuncId;
  4691. break;
  4692. }
  4693. }
  4694. } // extern "C"
  4695. namespace NKikimr::NPg {
  4696. constexpr char INTERNAL_TYPE_AND_MOD_SEPARATOR = ':';
  4697. struct ITypeDesc {
  4698. virtual ~ITypeDesc() = default;
  4699. };
  4700. class TPgTypeDescriptor
  4701. : public NYql::NPg::TTypeDesc
  4702. , public ITypeDesc
  4703. {
  4704. public:
  4705. explicit TPgTypeDescriptor(const NYql::NPg::TTypeDesc& desc)
  4706. : NYql::NPg::TTypeDesc(desc)
  4707. {
  4708. if (TypeId == ArrayTypeId) {
  4709. const auto& typeDesc = NYql::NPg::LookupType(ElementTypeId);
  4710. YdbTypeName = TString("_pg") + desc.Name.substr(1);
  4711. if (typeDesc.CompareProcId) {
  4712. CompareProcId = NYql::NPg::LookupProc("btarraycmp", { 0, 0 }).ProcId;
  4713. }
  4714. if (typeDesc.HashProcId) {
  4715. HashProcId = NYql::NPg::LookupProc("hash_array", { 0 }).ProcId;
  4716. }
  4717. if (typeDesc.ReceiveFuncId) {
  4718. ReceiveFuncId = NYql::NPg::LookupProc("array_recv", { 0, 0, 0 }).ProcId;
  4719. }
  4720. if (typeDesc.SendFuncId) {
  4721. SendFuncId = NYql::NPg::LookupProc("array_send", { 0 }).ProcId;
  4722. }
  4723. if (typeDesc.InFuncId) {
  4724. InFuncId = NYql::NPg::LookupProc("array_in", { 0, 0, 0 }).ProcId;
  4725. }
  4726. if (typeDesc.OutFuncId) {
  4727. OutFuncId = NYql::NPg::LookupProc("array_out", { 0 }).ProcId;
  4728. }
  4729. if (NYql::NPg::HasCast(ElementTypeId, ElementTypeId) && typeDesc.TypeModInFuncId) {
  4730. NeedsCoercion = true;
  4731. TypeModInFuncId = typeDesc.TypeModInFuncId;
  4732. }
  4733. } else {
  4734. YdbTypeName = TString("pg") + desc.Name;
  4735. StoredSize = TypeLen < 0 ? 0 : TypeLen;
  4736. if (TypeId == NAMEOID) {
  4737. StoredSize = 0; // store 'name' as usual string
  4738. }
  4739. if (NYql::NPg::HasCast(TypeId, TypeId) && TypeModInFuncId) {
  4740. NeedsCoercion = true;
  4741. }
  4742. }
  4743. }
  4744. int Compare(const char* dataL, size_t sizeL, const char* dataR, size_t sizeR) const {
  4745. NMiniKQL::TScopedAlloc alloc(__LOCATION__);
  4746. NMiniKQL::TPAllocScope scope;
  4747. Datum datumL = 0, datumR = 0;
  4748. Y_DEFER {
  4749. if (!PassByValue) {
  4750. if (datumL)
  4751. pfree((void*)datumL);
  4752. if (datumR)
  4753. pfree((void*)datumR);
  4754. }
  4755. };
  4756. datumL = Receive(dataL, sizeL);
  4757. datumR = Receive(dataR, sizeR);
  4758. FmgrInfo finfo;
  4759. InitFunc(CompareProcId, &finfo, 2, 2);
  4760. LOCAL_FCINFO(callInfo, 2);
  4761. Zero(*callInfo);
  4762. callInfo->flinfo = &finfo;
  4763. callInfo->nargs = 2;
  4764. callInfo->fncollation = DEFAULT_COLLATION_OID;
  4765. callInfo->isnull = false;
  4766. callInfo->args[0] = { datumL, false };
  4767. callInfo->args[1] = { datumR, false };
  4768. auto result = finfo.fn_addr(callInfo);
  4769. Y_ENSURE(!callInfo->isnull);
  4770. return DatumGetInt32(result);
  4771. }
  4772. ui64 Hash(const char* data, size_t size) const {
  4773. NMiniKQL::TScopedAlloc alloc(__LOCATION__);
  4774. NMiniKQL::TPAllocScope scope;
  4775. Datum datum = 0;
  4776. Y_DEFER {
  4777. if (!PassByValue && datum) {
  4778. pfree((void*)datum);
  4779. }
  4780. };
  4781. datum = Receive(data, size);
  4782. FmgrInfo finfo;
  4783. InitFunc(HashProcId, &finfo, 1, 1);
  4784. LOCAL_FCINFO(callInfo, 1);
  4785. Zero(*callInfo);
  4786. callInfo->flinfo = &finfo;
  4787. callInfo->nargs = 1;
  4788. callInfo->fncollation = DEFAULT_COLLATION_OID;
  4789. callInfo->isnull = false;
  4790. callInfo->args[0] = { datum, false };
  4791. auto result = finfo.fn_addr(callInfo);
  4792. Y_ENSURE(!callInfo->isnull);
  4793. return DatumGetUInt32(result);
  4794. }
  4795. TConvertResult NativeBinaryFromNativeText(const TString& str) const {
  4796. NMiniKQL::TOnlyThrowingBindTerminator bind;
  4797. NMiniKQL::TScopedAlloc alloc(__LOCATION__);
  4798. NMiniKQL::TPAllocScope scope;
  4799. Datum datum = 0;
  4800. text* serialized = nullptr;
  4801. Y_DEFER {
  4802. if (!PassByValue && datum) {
  4803. pfree((void*)datum);
  4804. }
  4805. if (serialized) {
  4806. pfree(serialized);
  4807. }
  4808. };
  4809. try {
  4810. {
  4811. FmgrInfo finfo;
  4812. InitFunc(InFuncId, &finfo, 1, 3);
  4813. LOCAL_FCINFO(callInfo, 3);
  4814. Zero(*callInfo);
  4815. callInfo->flinfo = &finfo;
  4816. callInfo->nargs = 3;
  4817. callInfo->fncollation = DEFAULT_COLLATION_OID;
  4818. callInfo->isnull = false;
  4819. callInfo->args[0] = { (Datum)str.data(), false };
  4820. callInfo->args[1] = { ObjectIdGetDatum(NMiniKQL::MakeTypeIOParam(*this)), false };
  4821. callInfo->args[2] = { Int32GetDatum(-1), false };
  4822. datum = finfo.fn_addr(callInfo);
  4823. Y_ENSURE(!callInfo->isnull);
  4824. }
  4825. FmgrInfo finfo;
  4826. InitFunc(SendFuncId, &finfo, 1, 1);
  4827. LOCAL_FCINFO(callInfo, 1);
  4828. Zero(*callInfo);
  4829. callInfo->flinfo = &finfo;
  4830. callInfo->nargs = 1;
  4831. callInfo->fncollation = DEFAULT_COLLATION_OID;
  4832. callInfo->isnull = false;
  4833. callInfo->args[0] = { datum, false };
  4834. serialized = (text*)finfo.fn_addr(callInfo);
  4835. Y_ENSURE(!callInfo->isnull);
  4836. return {TString(NMiniKQL::GetVarBuf(serialized)), {}};
  4837. } catch (const yexception& e) {
  4838. TStringBuilder errMsg;
  4839. errMsg << "Error while converting text to binary: " << e.what();
  4840. return {"", errMsg};
  4841. }
  4842. }
  4843. TConvertResult NativeTextFromNativeBinary(const TStringBuf binary) const {
  4844. NMiniKQL::TOnlyThrowingBindTerminator bind;
  4845. NMiniKQL::TScopedAlloc alloc(__LOCATION__);
  4846. NMiniKQL::TPAllocScope scope;
  4847. Datum datum = 0;
  4848. char* str = nullptr;
  4849. Y_DEFER {
  4850. if (!PassByValue && datum) {
  4851. pfree((void*)datum);
  4852. }
  4853. if (str) {
  4854. pfree(str);
  4855. }
  4856. };
  4857. try {
  4858. datum = Receive(binary.data(), binary.size());
  4859. FmgrInfo finfo;
  4860. InitFunc(OutFuncId, &finfo, 1, 1);
  4861. LOCAL_FCINFO(callInfo, 1);
  4862. Zero(*callInfo);
  4863. callInfo->flinfo = &finfo;
  4864. callInfo->nargs = 1;
  4865. callInfo->fncollation = DEFAULT_COLLATION_OID;
  4866. callInfo->isnull = false;
  4867. callInfo->args[0] = { datum, false };
  4868. str = (char*)finfo.fn_addr(callInfo);
  4869. Y_ENSURE(!callInfo->isnull);
  4870. return {TString(str), {}};
  4871. } catch (const yexception& e) {
  4872. TStringBuilder errMsg;
  4873. errMsg << "Error while converting binary to text: " << e.what();
  4874. return {"", errMsg};
  4875. }
  4876. }
  4877. TTypeModResult ReadTypeMod(const TString& str) const {
  4878. TVector<TString> params;
  4879. ::Split(str, ",", params);
  4880. if (params.size() > 2) {
  4881. TStringBuilder errMsg;
  4882. errMsg << "Error in 'typemodin' function: "
  4883. << NYql::NPg::LookupProc(TypeModInFuncId).Name
  4884. << ", reason: too many parameters";
  4885. return {-1, errMsg};
  4886. }
  4887. TVector<Datum> dvalues;
  4888. TVector<bool> dnulls;
  4889. dnulls.resize(params.size(), false);
  4890. dvalues.reserve(params.size());
  4891. TString textNumberParam;
  4892. if (TypeId == INTERVALOID || TypeId == INTERVALARRAYOID) {
  4893. i32 typmod = -1;
  4894. auto ok = NYql::ParsePgIntervalModifier(params[0], typmod);
  4895. if (!ok) {
  4896. TStringBuilder errMsg;
  4897. errMsg << "Error in 'typemodin' function: "
  4898. << NYql::NPg::LookupProc(TypeModInFuncId).Name
  4899. << ", reason: invalid parameter '" << params[0]
  4900. << "' for type pginterval";
  4901. return {-1, errMsg};
  4902. }
  4903. textNumberParam = Sprintf("%d", typmod);
  4904. dvalues.push_back(PointerGetDatum(textNumberParam.data()));
  4905. if (params.size() > 1) {
  4906. dvalues.push_back(PointerGetDatum(params[1].data()));
  4907. }
  4908. } else {
  4909. for (size_t i = 0; i < params.size(); ++i) {
  4910. dvalues.push_back(PointerGetDatum(params[i].data()));
  4911. }
  4912. }
  4913. NMiniKQL::TOnlyThrowingBindTerminator bind;
  4914. NMiniKQL::TScopedAlloc alloc(__LOCATION__);
  4915. NMiniKQL::TPAllocScope scope;
  4916. ArrayType* paramsArray = nullptr;
  4917. Y_DEFER {
  4918. if (paramsArray) {
  4919. pfree(paramsArray);
  4920. }
  4921. };
  4922. try {
  4923. int ndims = 0;
  4924. int dims[MAXDIM];
  4925. int lbs[MAXDIM];
  4926. ndims = 1;
  4927. dims[0] = params.size();
  4928. lbs[0] = 1;
  4929. const auto& cstringDesc = NYql::NPg::LookupType(CSTRINGOID);
  4930. paramsArray = construct_md_array(dvalues.data(), dnulls.data(), ndims, dims, lbs,
  4931. cstringDesc.TypeId,
  4932. cstringDesc.TypeLen,
  4933. cstringDesc.PassByValue,
  4934. cstringDesc.TypeAlign);
  4935. FmgrInfo finfo;
  4936. InitFunc(TypeModInFuncId, &finfo, 1, 1);
  4937. LOCAL_FCINFO(callInfo, 1);
  4938. Zero(*callInfo);
  4939. callInfo->flinfo = &finfo;
  4940. callInfo->nargs = 1;
  4941. callInfo->fncollation = DEFAULT_COLLATION_OID;
  4942. callInfo->isnull = false;
  4943. callInfo->args[0] = { PointerGetDatum(paramsArray), false };
  4944. auto result = finfo.fn_addr(callInfo);
  4945. Y_ENSURE(!callInfo->isnull);
  4946. return {DatumGetInt32(result), {}};
  4947. } catch (const yexception& e) {
  4948. TStringBuilder errMsg;
  4949. errMsg << "Error in 'typemodin' function: "
  4950. << NYql::NPg::LookupProc(TypeModInFuncId).Name
  4951. << ", reason: " << e.what();
  4952. return {-1, errMsg};
  4953. }
  4954. }
  4955. TMaybe<TString> Validate(const TStringBuf binary) const {
  4956. NMiniKQL::TOnlyThrowingBindTerminator bind;
  4957. NMiniKQL::TScopedAlloc alloc(__LOCATION__);
  4958. NMiniKQL::TPAllocScope scope;
  4959. Datum datum = 0;
  4960. Y_DEFER {
  4961. if (!PassByValue && datum) {
  4962. pfree((void*)datum);
  4963. }
  4964. };
  4965. try {
  4966. datum = Receive(binary.data(), binary.size());
  4967. return {};
  4968. } catch (const yexception& e) {
  4969. TStringBuilder errMsg;
  4970. errMsg << "Error in 'recv' function: "
  4971. << NYql::NPg::LookupProc(ReceiveFuncId).Name
  4972. << ", reason: " << e.what();
  4973. return errMsg;
  4974. }
  4975. }
  4976. TCoerceResult Coerce(const TStringBuf binary, i32 typmod) const {
  4977. return Coerce(true, binary, 0, typmod);
  4978. }
  4979. TCoerceResult Coerce(const NUdf::TUnboxedValuePod& value, i32 typmod) const {
  4980. Datum datum = PassByValue ?
  4981. NMiniKQL::ScalarDatumFromPod(value) :
  4982. NMiniKQL::PointerDatumFromPod(value);
  4983. return Coerce(false, {}, datum, typmod);
  4984. }
  4985. private:
  4986. TCoerceResult Coerce(bool isSourceBinary, const TStringBuf binary, Datum datum, i32 typmod) const {
  4987. NMiniKQL::TOnlyThrowingBindTerminator bind;
  4988. NMiniKQL::TScopedAlloc alloc(__LOCATION__);
  4989. NMiniKQL::TPAllocScope scope;
  4990. Datum datumCasted = 0;
  4991. TVector<Datum> elems;
  4992. TVector<bool> nulls;
  4993. TVector<Datum> castedElements;
  4994. bool passByValueElem = false;
  4995. text* serialized = nullptr;
  4996. Y_DEFER {
  4997. if (!PassByValue) {
  4998. if (datum && isSourceBinary) {
  4999. pfree((void*)datum);
  5000. }
  5001. if (datumCasted) {
  5002. pfree((void*)datumCasted);
  5003. }
  5004. }
  5005. if (IsArray() && !passByValueElem) {
  5006. for (ui32 i = 0; i < castedElements.size(); ++i) {
  5007. pfree((void*)castedElements[i]);
  5008. }
  5009. }
  5010. if (serialized) {
  5011. pfree(serialized);
  5012. }
  5013. };
  5014. try {
  5015. if (isSourceBinary) {
  5016. datum = Receive(binary.data(), binary.size());
  5017. }
  5018. if (IsArray()) {
  5019. const auto& typeDesc = NYql::NPg::LookupType(ElementTypeId);
  5020. passByValueElem = typeDesc.PassByValue;
  5021. auto arr = (ArrayType*)DatumGetPointer(datum);
  5022. auto ndim = ARR_NDIM(arr);
  5023. auto dims = ARR_DIMS(arr);
  5024. auto lb = ARR_LBOUND(arr);
  5025. auto nitems = ArrayGetNItems(ndim, dims);
  5026. elems.resize(nitems);
  5027. nulls.resize(nitems);
  5028. castedElements.reserve(nitems);
  5029. array_iter iter;
  5030. array_iter_setup(&iter, (AnyArrayType*)arr);
  5031. for (ui32 i = 0; i < nitems; ++i) {
  5032. bool isNull;
  5033. auto datum = array_iter_next(&iter, &isNull, i,
  5034. typeDesc.TypeLen, typeDesc.PassByValue, typeDesc.TypeAlign);
  5035. if (isNull) {
  5036. elems[i] = 0;
  5037. nulls[i] = true;
  5038. continue;
  5039. }
  5040. elems[i] = CoerceOne(ElementTypeId, datum, typmod);
  5041. nulls[i] = false;
  5042. if (elems[i] != datum) {
  5043. castedElements.push_back(elems[i]);
  5044. }
  5045. }
  5046. if (!castedElements.empty()) {
  5047. auto newArray = construct_md_array(elems.data(), nulls.data(), ndim, dims, lb,
  5048. typeDesc.TypeId, typeDesc.TypeLen, typeDesc.PassByValue, typeDesc.TypeAlign);
  5049. datumCasted = PointerGetDatum(newArray);
  5050. }
  5051. } else {
  5052. datumCasted = CoerceOne(TypeId, datum, typmod);
  5053. if (datumCasted == datum) {
  5054. datumCasted = 0;
  5055. }
  5056. }
  5057. if (!datumCasted && isSourceBinary) {
  5058. return {{}, {}};
  5059. } else {
  5060. FmgrInfo finfo;
  5061. InitFunc(SendFuncId, &finfo, 1, 1);
  5062. LOCAL_FCINFO(callInfo, 1);
  5063. Zero(*callInfo);
  5064. callInfo->flinfo = &finfo;
  5065. callInfo->nargs = 1;
  5066. callInfo->fncollation = DEFAULT_COLLATION_OID;
  5067. callInfo->isnull = false;
  5068. callInfo->args[0] = { datumCasted ? datumCasted : datum, false };
  5069. serialized = (text*)finfo.fn_addr(callInfo);
  5070. Y_ENSURE(!callInfo->isnull);
  5071. return {TString(NMiniKQL::GetVarBuf(serialized)), {}};
  5072. }
  5073. } catch (const yexception& e) {
  5074. TStringBuilder errMsg;
  5075. errMsg << "Error while coercing value, reason: " << e.what();
  5076. return {{}, errMsg};
  5077. }
  5078. }
  5079. Datum CoerceOne(ui32 typeId, Datum datum, i32 typmod) const {
  5080. const auto& cast = NYql::NPg::LookupCast(typeId, typeId);
  5081. FmgrInfo finfo;
  5082. InitFunc(cast.FunctionId, &finfo, 2, 3);
  5083. LOCAL_FCINFO(callInfo, 3);
  5084. Zero(*callInfo);
  5085. callInfo->flinfo = &finfo;
  5086. callInfo->nargs = 3;
  5087. callInfo->fncollation = DEFAULT_COLLATION_OID;
  5088. callInfo->isnull = false;
  5089. callInfo->args[0] = { datum, false };
  5090. callInfo->args[1] = { Int32GetDatum(typmod), false };
  5091. callInfo->args[2] = { BoolGetDatum(false), false };
  5092. auto result = finfo.fn_addr(callInfo);
  5093. Y_ENSURE(!callInfo->isnull);
  5094. return result;
  5095. }
  5096. Datum Receive(const char* data, size_t size) const {
  5097. StringInfoData stringInfo;
  5098. stringInfo.data = (char*)data;
  5099. stringInfo.len = size;
  5100. stringInfo.maxlen = size;
  5101. stringInfo.cursor = 0;
  5102. FmgrInfo finfo;
  5103. InitFunc(ReceiveFuncId, &finfo, 1, 3);
  5104. LOCAL_FCINFO(callInfo, 3);
  5105. Zero(*callInfo);
  5106. callInfo->flinfo = &finfo;
  5107. callInfo->nargs = 3;
  5108. callInfo->fncollation = DEFAULT_COLLATION_OID;
  5109. callInfo->isnull = false;
  5110. callInfo->args[0] = { (Datum)&stringInfo, false };
  5111. callInfo->args[1] = { ObjectIdGetDatum(NMiniKQL::MakeTypeIOParam(*this)), false };
  5112. callInfo->args[2] = { Int32GetDatum(-1), false };
  5113. auto result = finfo.fn_addr(callInfo);
  5114. Y_ENSURE(!callInfo->isnull);
  5115. return result;
  5116. }
  5117. bool IsArray() const {
  5118. return TypeId == ArrayTypeId;
  5119. }
  5120. static inline void InitFunc(ui32 funcId, FmgrInfo* info, ui32 argCountMin, ui32 argCountMax) {
  5121. Zero(*info);
  5122. Y_ENSURE(funcId);
  5123. NYql::GetPgFuncAddr(funcId, *info);
  5124. Y_ENSURE(info->fn_addr);
  5125. Y_ENSURE(info->fn_nargs >= argCountMin && info->fn_nargs <= argCountMax);
  5126. }
  5127. public:
  5128. TString YdbTypeName;
  5129. ui32 StoredSize = 0; // size in local db, 0 for variable size
  5130. bool NeedsCoercion = false;
  5131. };
  5132. class TPgTypeDescriptors {
  5133. public:
  5134. static const TPgTypeDescriptors& Instance() {
  5135. return *Singleton<TPgTypeDescriptors>();
  5136. }
  5137. TPgTypeDescriptors() {
  5138. auto initType = [this] (ui32 pgTypeId, const NYql::NPg::TTypeDesc& type) {
  5139. this->InitType(pgTypeId, type);
  5140. };
  5141. NYql::NPg::EnumTypes(initType);
  5142. }
  5143. const TPgTypeDescriptor* Find(ui32 pgTypeId) const {
  5144. return PgTypeDescriptors.FindPtr(pgTypeId);
  5145. }
  5146. const TPgTypeDescriptor* Find(const TStringBuf name) const {
  5147. auto* id = ByName.FindPtr(name);
  5148. if (id) {
  5149. return Find(*id);
  5150. }
  5151. return {};
  5152. }
  5153. private:
  5154. void InitType(ui32 pgTypeId, const NYql::NPg::TTypeDesc& type) {
  5155. Y_ENSURE(pgTypeId);
  5156. auto desc = TPgTypeDescriptor(type);
  5157. Y_ENSURE(ByName.emplace(desc.YdbTypeName, pgTypeId).second);
  5158. Y_ENSURE(PgTypeDescriptors.emplace(pgTypeId, desc).second);
  5159. }
  5160. private:
  5161. THashMap<ui32, TPgTypeDescriptor> PgTypeDescriptors;
  5162. THashMap<TString, ui32> ByName;
  5163. };
  5164. ui32 PgTypeIdFromTypeDesc(const ITypeDesc* typeDesc) {
  5165. if (!typeDesc) {
  5166. return 0;
  5167. }
  5168. return static_cast<const TPgTypeDescriptor*>(typeDesc)->TypeId;
  5169. }
  5170. const ITypeDesc* TypeDescFromPgTypeId(ui32 pgTypeId) {
  5171. if (!pgTypeId) {
  5172. return {};
  5173. }
  5174. return (const ITypeDesc*)TPgTypeDescriptors::Instance().Find(pgTypeId);
  5175. }
  5176. TString PgTypeNameFromTypeDesc(const ITypeDesc* typeDesc, const TString& typeMod) {
  5177. if (!typeDesc) {
  5178. return "";
  5179. }
  5180. auto* pgTypeDesc = static_cast<const TPgTypeDescriptor*>(typeDesc);
  5181. if (typeMod.empty()) {
  5182. return pgTypeDesc->YdbTypeName;
  5183. }
  5184. return pgTypeDesc->YdbTypeName + INTERNAL_TYPE_AND_MOD_SEPARATOR + typeMod;
  5185. }
  5186. const ITypeDesc* TypeDescFromPgTypeName(const TStringBuf name) {
  5187. auto space = name.find_first_of(INTERNAL_TYPE_AND_MOD_SEPARATOR);
  5188. if (space != TStringBuf::npos) {
  5189. return (const ITypeDesc*)TPgTypeDescriptors::Instance().Find(name.substr(0, space));
  5190. }
  5191. return (const ITypeDesc*)TPgTypeDescriptors::Instance().Find(name);
  5192. }
  5193. TString TypeModFromPgTypeName(const TStringBuf name) {
  5194. auto space = name.find_first_of(INTERNAL_TYPE_AND_MOD_SEPARATOR);
  5195. if (space != TStringBuf::npos) {
  5196. return TString(name.substr(space + 1));
  5197. }
  5198. return {};
  5199. }
  5200. bool TypeDescIsComparable(const ITypeDesc* typeDesc) {
  5201. if (!typeDesc) {
  5202. return false;
  5203. }
  5204. return static_cast<const TPgTypeDescriptor*>(typeDesc)->CompareProcId != 0;
  5205. }
  5206. i32 TypeDescGetTypeLen(const ITypeDesc* typeDesc) {
  5207. if (!typeDesc) {
  5208. return 0;
  5209. }
  5210. return static_cast<const TPgTypeDescriptor*>(typeDesc)->TypeLen;
  5211. }
  5212. ui32 TypeDescGetStoredSize(const ITypeDesc* typeDesc) {
  5213. if (!typeDesc) {
  5214. return 0;
  5215. }
  5216. return static_cast<const TPgTypeDescriptor*>(typeDesc)->StoredSize;
  5217. }
  5218. bool TypeDescNeedsCoercion(const ITypeDesc* typeDesc) {
  5219. if (!typeDesc) {
  5220. return false;
  5221. }
  5222. return static_cast<const TPgTypeDescriptor*>(typeDesc)->NeedsCoercion;
  5223. }
  5224. int PgNativeBinaryCompare(const char* dataL, size_t sizeL, const char* dataR, size_t sizeR, const ITypeDesc* typeDesc) {
  5225. return static_cast<const TPgTypeDescriptor*>(typeDesc)->Compare(dataL, sizeL, dataR, sizeR);
  5226. }
  5227. ui64 PgNativeBinaryHash(const char* data, size_t size, const ITypeDesc* typeDesc) {
  5228. return static_cast<const TPgTypeDescriptor*>(typeDesc)->Hash(data, size);
  5229. }
  5230. TTypeModResult BinaryTypeModFromTextTypeMod(const TString& str, const ITypeDesc* typeDesc) {
  5231. if (!typeDesc) {
  5232. return {-1, "invalid type descriptor"};
  5233. }
  5234. return static_cast<const TPgTypeDescriptor*>(typeDesc)->ReadTypeMod(str);
  5235. }
  5236. TMaybe<TString> PgNativeBinaryValidate(const TStringBuf binary, const ITypeDesc* typeDesc) {
  5237. if (!typeDesc) {
  5238. return "invalid type descriptor";
  5239. }
  5240. return static_cast<const TPgTypeDescriptor*>(typeDesc)->Validate(binary);
  5241. }
  5242. TCoerceResult PgNativeBinaryCoerce(const TStringBuf binary, const ITypeDesc* typeDesc, i32 typmod) {
  5243. if (!typeDesc) {
  5244. return {{}, "invalid type descriptor"};
  5245. }
  5246. return static_cast<const TPgTypeDescriptor*>(typeDesc)->Coerce(binary, typmod);
  5247. }
  5248. TConvertResult PgNativeBinaryFromNativeText(const TString& str, const ITypeDesc* typeDesc) {
  5249. if (!typeDesc) {
  5250. return {{}, "invalid type descriptor"};
  5251. }
  5252. return static_cast<const TPgTypeDescriptor*>(typeDesc)->NativeBinaryFromNativeText(str);
  5253. }
  5254. TConvertResult PgNativeBinaryFromNativeText(const TString& str, ui32 pgTypeId) {
  5255. return PgNativeBinaryFromNativeText(str, TypeDescFromPgTypeId(pgTypeId));
  5256. }
  5257. TConvertResult PgNativeTextFromNativeBinary(const TStringBuf binary, const ITypeDesc* typeDesc) {
  5258. if (!typeDesc) {
  5259. return {{}, "invalid type descriptor"};
  5260. }
  5261. return static_cast<const TPgTypeDescriptor*>(typeDesc)->NativeTextFromNativeBinary(binary);
  5262. }
  5263. TConvertResult PgNativeTextFromNativeBinary(const TStringBuf binary, ui32 pgTypeId) {
  5264. return PgNativeTextFromNativeBinary(binary, TypeDescFromPgTypeId(pgTypeId));
  5265. }
  5266. } // namespace NKikimr::NPg
  5267. namespace NYql::NCommon {
  5268. TString PgValueCoerce(const NUdf::TUnboxedValuePod& value, ui32 pgTypeId, i32 typMod, TMaybe<TString>* error) {
  5269. auto typeDesc = NKikimr::NPg::TypeDescFromPgTypeId(pgTypeId);
  5270. if (!typeDesc) {
  5271. if (error) {
  5272. *error = "invalid type descriptor";
  5273. }
  5274. return {};
  5275. }
  5276. auto result = static_cast<const NKikimr::NPg::TPgTypeDescriptor*>(typeDesc)->Coerce(value, typMod);
  5277. if (result.Error) {
  5278. if (error) {
  5279. *error = result.Error;
  5280. }
  5281. return {};
  5282. }
  5283. return *result.NewValue;
  5284. }
  5285. } // namespace NYql::NCommon