From e865f6da2601121d8e0c52b2f4452e9a6907b958 Mon Sep 17 00:00:00 2001 From: vikasrathee-cs Date: Mon, 27 Jul 2026 13:41:24 +0530 Subject: [PATCH] Addition of Databricks Plugin --- .../AbstractDBSpecificConnector.java | 21 +- .../docs/Databricks-batchsource.md | 15 + .../docs/Databricks-connector.md | 15 + .../icons/Databricks-batchsource.png | Bin 0 -> 28613 bytes databricks-plugin/pom.xml | 127 ++++++++ .../databricks/DatabricksConnector.java | 162 ++++++++++ .../databricks/DatabricksConnectorConfig.java | 129 ++++++++ .../databricks/DatabricksConstants.java | 31 ++ .../plugin/databricks/DatabricksDBRecord.java | 65 ++++ .../databricks/DatabricksSchemaReader.java | 79 +++++ .../plugin/databricks/DatabricksSource.java | 153 ++++++++++ .../DatabricksConnectorUnitTest.java | 70 +++++ .../DatabricksSchemaReaderTest.java | 90 ++++++ .../widgets/Databricks-batchsource.json | 279 ++++++++++++++++++ .../widgets/Databricks-connector.json | 114 +++++++ pom.xml | 1 + 16 files changed, 1350 insertions(+), 1 deletion(-) create mode 100644 databricks-plugin/docs/Databricks-batchsource.md create mode 100644 databricks-plugin/docs/Databricks-connector.md create mode 100644 databricks-plugin/icons/Databricks-batchsource.png create mode 100644 databricks-plugin/pom.xml create mode 100644 databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConnector.java create mode 100644 databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConnectorConfig.java create mode 100644 databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConstants.java create mode 100644 databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksDBRecord.java create mode 100644 databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksSchemaReader.java create mode 100644 databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksSource.java create mode 100644 databricks-plugin/src/test/java/io/cdap/plugin/databricks/DatabricksConnectorUnitTest.java create mode 100644 databricks-plugin/src/test/java/io/cdap/plugin/databricks/DatabricksSchemaReaderTest.java create mode 100644 databricks-plugin/widgets/Databricks-batchsource.json create mode 100644 databricks-plugin/widgets/Databricks-connector.json diff --git a/database-commons/src/main/java/io/cdap/plugin/db/connector/AbstractDBSpecificConnector.java b/database-commons/src/main/java/io/cdap/plugin/db/connector/AbstractDBSpecificConnector.java index 0308cf7a4..e0407203b 100644 --- a/database-commons/src/main/java/io/cdap/plugin/db/connector/AbstractDBSpecificConnector.java +++ b/database-commons/src/main/java/io/cdap/plugin/db/connector/AbstractDBSpecificConnector.java @@ -104,8 +104,12 @@ public InputFormatProvider getInputFormatProvider(ConnectorContext context, Samp String tableQuery = getTableQuery(path.getDatabase(), path.getSchema(), path.getTable(), request.getLimit(), request.getProperties().get("sampleType"), request.getProperties().get("strata"), sessionID); DataDrivenETLDBInputFormat.setInput(connectionConfigAccessor.getConfiguration(), getDBRecordType(), - tableQuery, null, false); + tableQuery, null, isAutoCommitEnabled()); connectionConfigAccessor.setConnectionArguments(Maps.fromProperties(config.getConnectionArgumentsProperties())); + String isolationLevel = getTransactionIsolationLevel(); + if (isolationLevel != null) { + connectionConfigAccessor.setTransactionIsolationLevel(isolationLevel); + } connectionConfigAccessor.getConfiguration().setInt(MRJobConfig.NUM_MAPS, 1); Map additionalArguments = config.getAdditionalArguments(); for (Map.Entry argument : additionalArguments.entrySet()) { @@ -221,4 +225,19 @@ protected Schema getTableSchema(Connection connection, String database, protected String generateSessionID() { return UUID.randomUUID().toString().replace('-', '_'); } + + /** + * Returns whether auto-commit should be enabled for this connector. + * By default, it is false. + */ + protected boolean isAutoCommitEnabled() { + return false; + } + /** + * Returns the default transaction isolation level for this connector. + * If null, it falls back to the database driver's default or serializable. + */ + protected String getTransactionIsolationLevel() { + return null; + } } diff --git a/databricks-plugin/docs/Databricks-batchsource.md b/databricks-plugin/docs/Databricks-batchsource.md new file mode 100644 index 000000000..f259e6e9d --- /dev/null +++ b/databricks-plugin/docs/Databricks-batchsource.md @@ -0,0 +1,15 @@ +# Databricks Batch Source + +Description +----------- +Reads data from a Databricks table using a configurable SQL query. + +Properties +---------- +* **Use Connection**: Whether to use an existing Databricks connection. +* **Host**: Server Hostname of the Databricks cluster or SQL warehouse. +* **Port**: Database port (default is 443). +* **HTTP Path**: The HTTP Path for the Databricks cluster or SQL warehouse. +* **Reference Name**: Name used to identify this source for lineage. +* **Database / Catalog**: Optional catalog or database name. +* **Import Query**: SQL query to execute against Databricks. diff --git a/databricks-plugin/docs/Databricks-connector.md b/databricks-plugin/docs/Databricks-connector.md new file mode 100644 index 000000000..73d7ca1fd --- /dev/null +++ b/databricks-plugin/docs/Databricks-connector.md @@ -0,0 +1,15 @@ +# Databricks Database Connector + +Description +----------- +Connects to Databricks database / Lakehouse via JDBC. + +Properties +---------- +* **Host**: Server Hostname of the Databricks cluster or SQL warehouse. +* **Port**: Database port (default is 443). +* **HTTP Path**: The HTTP Path for the Databricks cluster or SQL warehouse. +* **Database / Catalog**: Optional catalog or database name to connect to. +* **Username**: Username / token user. +* **Password / Token**: Personal Access Token (PAT) or password. +* **Connection Arguments**: Arbitrary key-value pairs to pass as connection arguments to the JDBC driver (e.g. `AuthMech=11;Auth_Flow=2`). diff --git a/databricks-plugin/icons/Databricks-batchsource.png b/databricks-plugin/icons/Databricks-batchsource.png new file mode 100644 index 0000000000000000000000000000000000000000..e27f31a4c6774e6349ae3983828a8e5561636f55 GIT binary patch literal 28613 zcmbq)bzGEP*YyAb0s;aOlF}tzGL&?;N=ONk5(0yCw-VCbB_aqCN`q1&-60`4LrTL? z-x==vd7tjb-in)6OIvP?TU-)(LX!@&vUBZ&xk{}C5Qnqrx0 z{HX-o-g2{X!CgGsT3b9i_X8S+%m7YE$a~kjhLI0?Jt*Yw6D~Wi%^dJkHJP}LjN`{^ z6A*v+cvtJT0zDHGV&`bq9~RsODgC&!QS$8Q@~r#UknvGt{EpX7?Zm-vkw@#9t^=a~ z|Nn609*?=ymL!J{{K>@oA%p#TAo}4d|88du(L;$-R_2$gpR}4p?fKV#EA6AV)oaU4 zkBJ-FQ|WiV^XYtp6k71={Td7YYINHK@@R5p%y~(0td3~Uug?VOZbCdI5=p>QyG^)F zn-?)q4{3tDu28leNW_VvX}vefY^M11Y)+HjfqrMsOoEW^gQm>7krHZ{m$wwWHwsdX zBd33bADbg-DuZMYoLH3EAp`XaiT>mculU@8n(b0e&oo}*ebTy^{@YUF%6*Rq zlKBdKBUSj_H#a;hf71n#krR3on^0rIu_ND1SGRTWVVvTvD2vFYakB%y8HF`<(Wq1L zFG3RsZOe@N*#q@2ZXPKVu(k9@`b7O|V+qB{lX}pYxwAxm2O0eG^}u)^9%P+Zp}y~V zAQweR$N>Licam3mZYlnh&`8$`7cl*lyBn!~l+HEy)svwE`#~LCa+i4a=M{}Iip~Qx zY2=Urp-25`F5=5wpJkD3R@!iJ9`(z%6u4B=h!?HjLm1`y?{^1arLl@&Uew*2#1YBp zis1vgMXZlEr)Irsw|{%G!6?>$`=cg`uXHvx&1M+OCjm>&(%|{ zK3Rm6E)|lfcp?DbijlA0);DAj-;eqj4{9z^f;q$B(yxVN5O>&=EoJ|A9SXR7+`EbBxM zbQrVu^Ur(IvIFA-U>)oVeUOY1_^Hj}u;dxI6!z8R6^nfgWu$R#=QB26G1$HJ_#W`s z$3mlx@l>}UOIhnu?KFMW^2S3hNNvf2;!CQ4CyPe4H{*4H2RRCVAKushrLZ^$cEg5l z#OrFxYy7||i!+P1`=znkJuXZZtX@4a<$Db%`mFN(&W8BX1oANTf=eIZ60+0qXxU$C1z{oq?oLUKcm zkm8l&6y2c%x+Knc+Sll_Et?e|TgokvwjIO5eI|OJZQPh8qN*}0MkY~Fz=Mi?lK>Y) z>yIhiH-_+sLUB8tOQ93Bo87WrX+~@1ZDdipVeHel55Q72Q-H(hO;&4dWnuJgFKo!b zj^PrmdOV89uidf8N#A5&&2Ty=JGKhfe&lu2T0}w%_8tQW(BC;Abg6T<#(vgB)uW$2 z(pQCxKP72SmDKdFX)uw6C1yE4)NZ1*msEHzF<(%Dn%3YS%mAu!LDx0?t9*YnBF;by znf21>E6mQ6sIl>KXmLM-<6O}6LTtx5zStf`M;Tk<=M!mJM?rAPq7ChGt{n4iNc2IV z%)tM_5`*{dgO(=mN=de!($D+DVIo!BRXh8wVH!&)l>4Tw2p};3gFqFAcVT!S={*%w z{N^9VW-Geu51waC#Y9Nj^hE2Ro|C0v)nKHpracUlBiOQC6xc}@utZ<(;=wS!^X3U$ zd>pKjT6oh(%gL*~*?ZpEMC zZ5YYlQsBNMY%}aD4ja%-2sy6TLmhLC${4W31q{wjQw{FEtN;WyI`JTriNvka>Ryme z)%X?&dx|7pb@XX2{W@g~Ii#ZW&Yi8ux}au&9qAF>M0&!bTsHlgqRA;o#!=J!@!~cg zq)f~uRu@8JmVdwALvItpw6vyHRc*qZ#*DPr2}Tl`GD-RvR6EL0Q3wh;a!u7o(g?n^ zFTVa;tS}pbFyaEIGqPB~cJ=X^u2LEA`L7{Q5CK$tIaS#s5_tttA%W7eTv<4dU$=t8 zjC27iF(>bjE(GQGr)E(?Rn-f?yd`MtD zF5mNio^lWDPFTvR@d`$ZEDzpxjJYW6y@IgE+hvZ*ub~4vBp?il!NHsy8q4;hi*tBT zy56HrID+o4VU*Qz;{4d1p1Vp0u^_|hyweHbgi>c;>}M6#5l=zoCRK}WlQ4RL#F2L2 zg!vm8xVR8FG}X}AGkVuds}u$2Eem4wr|}0?EvJ|_FvV3< zD@ey;r>Olr4{%ereVzV$zunel8ANmXJq%Ok3EW*WJ-XmgYLS2e4J+qIQ>)sYW#!@X zL(ZA1^E`2{ID#U4)~L<6c$s0@=}KRRo$$7fmtuyr6)-Gy4{WL&B7Mb~Bm!!t8h$O{ z({x5Vf>kg}{MAGTGSw)+sA1pS&WWeUlp$#a#(xO_4cG{9!tlVk7g-wPb0Bk`eHQgK zY(R(VlIUB|)yVqgGe`Y&QP+nFNN)7!1k@x@&2gp|K3RTTM~aG$h2XUnq6SihBU>h( zruJh4fG?G%>oCnNx^_8V-D%GI@l8ti&Bn`oK0WhHya{6PVbqs_RHNq(O?Z~AX8O3p z(u@Z$tAu5#TdtXW_IU0jG-k=fXlYiiAiOPk^yx(d9i7|J8fHKlCjeH3ml=VmgeHf? zFUBlzGt<;z&+!uY!WBt#Zxz4r_iYekCP36|y$)#_>yL<<5vJ9IC^1E^*lf7GD=fhEiztTz@E$r?V2LY|VwHWO*9W z3Xa4(1b-ABZ&%2Gt`G>E4G=gY4-++FmmfX9ZTNON1cZG-M&*2G-DTcfUc$FOB>?R2 zI<#-mpk^wL)zH_i)c9u!AQ*9AoN5;Suvb(93cl({X3;mf--K!`ZSd>`hoO$c?zzry zEwH9`!jCGgsq}?m$&^WvwPOvc3%E2L(Vqgp3tbW|s@%W=31T7QEb1t62ZyGUIRPAY zyYvj@;V!F1iT|dUq<^RcdCV8xZjxHAPWwENS_-1U!{#@cg|?Y^I+Vf!Y+Cxg;Fi9; zU`~f_#@Mi!Wy(WRtB~q|&%?~U`ONgCGt(@*;6|s!ay*C~h?%dpZhLq`c{weUt1};-c3V>}g z1c1&UAMX>(ExV`#rKL7nXNo<~m#;pp2X@PXCrkd5Ry=~R@g4H7|5||^%LdsHJK#$V zktO%7XXeyYP)+s1UD>9g+2RH-EpIE~u3CUPLU=q7YCFwl0HCQ^)KFCi80#0zXKYsw zVZ@UcW7%GeaSoW?^6-bKtnL^<4k^$l+4AtC(3j3TeA=4t%xJs#Qf(taA12zh0s>+w zI7~r;2ckf4smnTKPIZ?~Wi8E`22yvIZXP};Xspyb@F-a+1wBV&E0_M==RbRwdZ)R0 z!>pckR~1&h8Z?@}QadB$Q=Tg9Pl2D=X4Lo#WW7A*x~o-|w&| znx%In%#!=Neq7JQY<$3HqE~5jKh?Q;AjCGWAN@33wD?J_@tt06^6zQFtLC!^s(!36 zr)UEKULVfL)?YKRRnyq`sH1_y|0w&Ap>mbUSsW4IUz{TUMzM+-8sQq9&yI!PSr@gl zB<1;hXXP*Y19||K8kGoulkmU z3&6@XTB}Wkl89Ae?=VgdfZbhBp9U2|U-;wAc(G~SYn8sv`8@$~9tgTlca1?u z`E%a$a}@KCBCZZAP|f-J9D$eBRX=S0dT5BDV@ZyX(r&wX%&*?t3C>%PL=Hhiu?`+= zF}eaSST^M$y+c8*Q|wUh2Tdp>h<5My2--%eW7i= zw+Jbb997BZ+DpiN^jr-=6se|aB|sFU%I6(C(Zp`ASXvBpt6Pq{%o}~Q{!xI}#%5~+ zFq;~{&4g^FDkCNA_pQv2{$RH8>Ov)AmfdjVR09i8n93C8^rHEl!?|oDK&Aow=r0J} z$ZxpSA|SgaJd&Jj_$3k+GaMDKM=Hz)RwV*VU12tkt5T+l6IJB>r4VZ@fP&((mCRIuMBhjBN_xdxVl3^i`n&rom-@xj~Q zS{>Q69hFGU0#SD~7JDd9K$~hCI)XJ@KPq)AlS+PTzWdlf&lbnG537X)8jJUX4(*Bb z-QnyjinwyLVvr+Ig+^0b!<#u$=v!Crhm8gLSn~;|fF!Y)K%UsV#!$a0AaxP~m5uPq zwXT%mYy@6zLhJ zfce*xTRqt1_RCTf!m}M;6!n?xXo7eUp-RnFfgD)}cQhvg0G&&`e0oHRtk2ztOO8a> zIn0Yjq&X8O?!F=$(ZdousApM;g=8xl%-AUVrb{;7TL$%KY2=}A1i9f!1vVWf zC=g5nNqhK2X^HUbA@7JHf}{Vab3s&(aN_BetyVL8 z0wqS!%<%U9L{A+24Ft5$wIKM0UeyUvR=M;4rF|9mNvrDgss*F3fi$s{sR0N7HP))6AGkOUQ9#Qa*VDNH9N20w5> zZY7_(R@~?Yo9qj;V7CU^Umq(E6T){}JG^L{{5|69Q~2$013KIr;%Ofn+Rv@-hQ%yp zx(k!Z>hgl)({zij zo4mEs%8xborBT{3s`^NE?`X3dZR;TfT)Xdp3BMcU&{*`f$q1Nn4GdE@DBUIF;fYaMQ1 zvb{g1hJ!yEWvrbs((@tz5CjS*6x^E=7XN~C%$W_*L1SD8>eo`>=#QsM6C%4lwLc>K z1W5VENZHhiKYOf`GtIHvc&PXO4Ug+FfS?$(|8OhB0<)_s?_dm`gRqF3)8+^#0ILZq z#NRZKT1|fu?HZhiB!GF01=qBBAB(ML;<58#27D>?l+(60SJZinO>d%*ccG4v^Labd zmHR`O|E=5@l9D{W%>Y7FEi7uo5si#XBXk!YkhlT!?vH^H%5y+CQnL-Gmen z@&A=aqx!I z7lpf-R3_;Hs~%l6wX}PLsq*8#)yEe0e-?E!0AoGhyjNw{jt5yS&JKxIcYLLUuroZb zs5IS~r$WVa{htyaJIpVusnK*Zc!-lWO3uU=r)Gn zvU(wm184+M;ev&sa^6KOmNWbjYPb}T`E082BM1XATV+k0Od!;81VIj~L%N$(wg+LT zDdwT1z$yxTbYaPhAA4~io4kt)aVD8)?5LBf5d#SetlF#GU9}DS$|DL^Ali)pypx`Z zu@5ADYn+#WS|C5W!Ij7w{D)LfRW=idc(_VL(T!L|jQXsTS&P~%%i<;e>$}_^{$d8u zklEp7nFoBEA@^U;d?k59Ij2@5Nb{;8LsWho_HJ`g8`2h(CAS0`4eJj-CuufI0@W+i zRhE@^b(D^t(v?oIXNho#?CMzd6g_u1HH%=FtJl)6FvhkT5cUC4Jt{y!S#r1oe3`3$ ztZww+1&~`bg?`%>=`H&F>$fF#jqcu7wV&+8K3?Eiaa+>NI+Plm*Bm0EeSpMy=UWu4v;f{ZlX9fMoS5Ko`=E zA)et!%Bh27?SGC8-iXE#lP8R#wRE#3+4Fbn2 z(TrLVeKii%3Eq#Rdt=LI03=l<4$TX3r^d*_~11E>ts=#k?+C+1H&jXRr9)fanSFEiSu3 zqsR9EpOIH@JQa!P_>JZvY-!8bYjpclhmuBy$(}1R&FL$-dk89(M9XcLiV#mftJDKR zI|Jn+!D9xzfD-V^{KKh26zhStfKU)Aw@pUW_IdTT-XN-s&P~aNLS#!hPc&ZfMD-k% z4r{*i#vua^mTm0W9$JSw9fTR`=!wL`x@wdUYs=eDZ$*6<FlqkA4*rmo zgwT9XXOqKU02SJ0B3a6un#DX1miQ}ELI+5q0Mi1Jf#l`hiqN~EH|EdH<~l2el))C! zf+~S@D#%#{{;=H>O-t&(wRB|K9cG+mmQU`H4*oXoHxr;97scQTCD!J>J=S&YojRucu^Ji~DE;buf06 zKtKhv`HB!_5RaC~BWav{xQ&bIZ|dO8I6wv<5DbGClTv#htmwI^9=%Pubd!uFsx|PK4*UItkW;J)@3*9HQ)XfmiCy>} zH7Qu4{T19~zaGFXa)p1bR|2Y@E|JAUE7kiVSAMGkbfaw7BCvtbyflGsjuexwZ+l{a zu6(LJ)^T+tWplWlgfYjX&v4GL%Xk`_zx}{15|G}71nu;G`hIT_%LPHxYiU9S4uRvL z7WaV(%h6o$G}O!-E}o=~S&Knhk@<2FV0$Dd#D?RuT;l^)zyKkvk zxF9=}_~-wd9fTxnt@nFNu1WHp%rfsmHRU@j*d%i*MmWnGFFxljGlh~vp7KVNYc&N1 zV{0fF5<)4Ux|Un&AGWzyb+Z7+FC4N3$e{={bY00ye6?$w2_n=WFFMGwr$m0>t~y{{ z<}osZ#TBZgiQW1tV=LPHhGXO_VE0;_ayiAVZY3@VJU5o|JPgtG??9jQSRE{px9Egq6m5nUEdQ2es&W4nNQyaE$*7sU zxUF{97arO5JAB&GOfhue$^E0;Q=w;UEwi|dw70WZeuAnMYr=TzOU1~--A((B3J?e8 z!c>NSg+^=sWkC2qB}0W6kVg_U$pbzU+@VoLm$>B5@eJ87gp^HkZM)vHGJTVI@3h05+rHvy?kr8rM|IiiG`{JZ=1OeP7*MHe3o?gt@ zP)+G-0_YO}vV)v2uQuSkkYDPm=3~A^}ZMbS+lX;%<%+1|jba zYS;*-!q3F7irtc)Jx76*qMfh_WGoU8Ms3b}QX2PK`K5!yVmhbP5?h|1jftT$KYn!E zST;1gN^!fx{{GZA5Y7yK8m*VLq$<~DH?4-zPat~>fIbvAXX1d|g~UiRD#o^6MzoX| zA)odC>cau#Rj;{y{lfjOylL0gD)_Tp16}9t7m#(fQF6C)I!IoO9k`uze0hg@O5{Pu zpcE&no*d2>OC0o32FQV;=Q}-#8IrQ~hwj)p2dy07Xe#;3?8z*RfX%%!fj=q$LB`nQ zSTu-Rq=78BucXK5C2I}M54^vy&hL0}*;|xiyVf_EEur!sFF;iuqv-mTSbw&-_SqPV zO+8h$JNVi>$&OlZbb;*kY*V?-+MHw${Q+dtbJ3YIQBroy8w$q;(}&yiP4JcO(ysFV zoRVAl&GsN-Ec@zKw^|cjzC4A6jP=JTKb2M!+t4M<<6m<>zj4mIY=YeGsoQNCY*2Ah zY$AYJ1MI01?E0Fdd0hg^0YHdF(%Dlyx#|xI7#azdzkz^axWrl#(bq%tPv|zV!ZL^Z zfYxIMq!BnrKXc5Wyh=}u-sXcZj;jKuy25TUaKV9S7SKgkt?wh`bNsP&h=L}zDoN=) z!S06MYa(dW0ebEWiXMR3F7Q%=h}zr|8Pw`z|Dj{}lGR(H$DXCpMZ~yHtTKJSnW(JqLLHmI5kDm`{e9Bu~_0iwwV5B3Rf+$Acd=!K(Rd-17NYNr$S z6lB1I`#*=%y^UMA#2 z_|)2S_ZxjAUIRcY+k8k6DI=MI7a^8Z(JUy>3qkP|sZ{4Qq`ukQiYO|Fi^u8!omzQ0r zgSPw~Fhm3Z>7E7^Dufh_AMT~)4_8yPS`HwMlGu=39icnAL<8DYmU0=uB!tr(ju8YX zOE5W%9|J9Y?HB;$&s1*I!WE8n%*W_}E5sV;Arkb(v7JHRl$1{lt(xlR|K#siOud@^ zc;}<7U0Zf27Nbi9=;CP-0D;thsyw5kCWulckgqzgrKwx*!`xBJq4#2VFzJ_%qN_kD zz%x8$no&(%?t|96RYg=!4(#dIj;4N)Bppy5hi>Ou5F0r9?q?;iBok$7(*Tc$;B-;! z<|lb4X*!K<{UiZd!jTDZxG(z%o?IL3E-HY$QO69%9U)7I3FC_wa6v~v;cbYqE+>pz zlBcW~>KF*Z!8>>9S9Vg1IhV|guK)C>HBJB$-6}s`WY^BEb0*LwR#Rm1imS+zN`DbL zB7~|78SDT2q5Gz8Wj3VcOn=yl&dzQj6wCT$^JkGeTZ*RxoOfV6cpveVq-eKmD$4l= z75$XPycb(G_D`gPIY<}wISomqE4DN!U`(QEi31o080*t}4 z(XT0D zejnHjBrYfY`u&*tk{rbWOBhCWBbHr(9t1dmb>8wfqzb5%j}xv`>JitBu$V%=_ijlo zD%XE>;{|LYgewP!h{?yZ>mLA&|Cndwb>AYE!`$ai^-G_Dx3$z>cFArjPKOGTjJYBg z3(gGbUp_Djgz9ew#+s4Q-;}fq=^vy2Q+(oU9KP}ykw_gl;mfaxW6$v^NnHRccB_lFmO}p#uk+6Ib)3l%sM3?1G)lpMK#^eWdIg01 z)I+BZpDr(z6Ts$wkN)GyVBT=YaVRl8C@=BxhBHNj+VAl#qO(3xB|Wp_eMX-nlgeNB zO)rAzG)tprba%hoTet|b4PzTAs_t_9+cX>otO-Z1j`sAm`A%^j@fYT2Z1dDmWkxD| z%*Q_@GIS&_TYD`1Fn_;Z>7;98LCf0GyiS?Xb^7SuV)YV;RKzJJf%+jz-{%saEute0 zclErH3|VOb*@%i#wk+BbBP-^(x$tR4;xDICXokJ>6S}^8?n95=AvcADo9sLn=f8fE zaQ2xG`JkgDXQ5}2sYo0a*Y^08j=oLkQY=sh`!%-_Cl+;R7GCg&N!n6>nURE_YqG-= zhS_V-qQb-=K#>5_T#04hdfUMLHNBRYs~;&yLLUb~o+DgC2R!|tosaNq#vN0$Il?O_ zXfpXEN|n+WN)iGxquSZgN^)enB5_`(cY>b#**$r##oKz$q!s-S9a@b7E5Y>=p;-|ueRQz9e$i9FE+W5(vW|)*s6=|=sZS<_xEv}VOI>kCHMs~bS zl3Y#%kM9JktnFfsMYR*74Zh=L$=E$v;=5<6;eG}Uz8Z3lSE?T?jUkSH z+aKGEDK#rfe&H$ifj{-FeDy>Y@QQmD~& zz1L)GqRzOiQkc#6$is=AM>-U{rF=f>Z1n;yPWF^D0qP)kt8nhjh(286=}n7-q;lWj%o2# zhoRLCaZ8Z9adH{zy%5e(uzfuNaR+uk4{0x4TRi7otWD3;$8bSqL0)S4ES&UU?nqOZ_e6ecmzbg%t-d$*dGJ&@Lamiq}KIZY4Ay!!jj$Ksc_ioX3{t{0bmb<5aR8IvY2DDQdMsgm%iu{!|-}X`X z9~WQwpIu2WwD*bVX75kAeT~fB@BPXvV2VS@?pMjb=#lrL_o;Ux#$@de${VLhp^;^o zE@_iNb<8#edUQ)(hTc2}M3c4mDP9V%sYJQ98_KMUGJp={GYBN&|7gLik^k5$o>DmB zWEGi@d_`0gZMh2O0_**>FNMD*o+L$}hl*?i&$SZ3g_u~af zc_-?p3>!DJ149E#WW0M)aj==5^WQe;?4A;Fj93gMXe*PAT2kb6+1z9rCtfa8TPP$Z z{z%SL~+%CtdDw zAEumPLh2>+2p#V;P@^^sX~(M^6kFRm)Ki|$qU>H00Y_`)n|V?pQ9yipO(ey#(fkM| zpYjA}yXVb+y65r3HkBE(wKtw0`40M_aq+#Cq^r|P;of46+)m*Kk|73vkIL-+PsbnU zihN2C6A}Q&bYP5m_D`-(klT3VOzAvp)fYdSr-_v8!1JZ})uRPElpl`lDn)<*&WEr< z+E?w;g9N49J<4xIWbVvREFhWl=#Ebed{3y(cb6n@S-c-Fpkc_Q!;U4fPDmyxMJ`x( z8rlv8sjR1s$AeXK>-n;t`_Z2gC54%>0 z&@>e5uFNbOr@N70xR9m)nfNftAYYGqDFBM^& zeLKAcB4j0jCEVt<-~me?(6&|8JA#K=Grw0j33~0VW41;7Hwn?1&rZw|6d)xBLX0bge?{$M1Zm6t@#|uE_2=C=YDE*v4$k-0S24!>#Lpe(V@} z1b62nVL&G%78uA3-1ox4NqoAW9w-qD1HLjfJ90%Hh|Ga=j%@xvwzJ${?Wl$^5rVc* z^;6@;=#VyL`(j5A7s@QkQP3PAq{e2v8EWK!H52&NQTf@gHS4&Y_9c0gpE0lC4k!Vm zAA|Hk^s0U^|8q1DJd2o3LbrleYJyhDI?g7HsgtCxBhzM%1^p8RUPfL>_|{>)S`l~x zaEI7dYATG$fx%X=S3M_C{-36WaFIek)KK(IF)*x zHJnbwUxEZ_yGRt<479;?!PksAd7xdLXDe34WwW9{9BrTbP9I4Zz{w{@56i}#Q~V@9 zVT_c9Klq`%J-VQN&q1YcOc-&Z#0lrVwkXJ1b$k*j)_dlIV*_624_*KeX0&FVHczmS zl0%`>F2t_mdp->~o5uFDbTc$iWnz>Dtv##LM;3(Y9LX2XMl78vJJA+-ris>6;x7w< z`b$9*KgP#omoev6LQg^FU~;a4q^uAnZh+JCyhF)V%?nkVM&eU_27k6;-JUuZcpKKW z*+-NynQv2?PN{>5j~&Z5O%_{^rM$kc&Go847@bbJXQ7tY@w-=wN1ky;U_oXYUSLwB zXz!U(Bx)1d{rNVpZa;1S%1m|kSt~}jrlV5d!18*6ANHqQEu>1dUdd6uy&bLgSXyS8 z&T(MHOQ+58I@k5Qp%*4^M-a}`pH&|i;!@K9L8m6mum3A35%Ut5l3YsNgB-qZtpIVB zrPBz5hGS@6Y1EAf+isDoh9vul$V4+#&f^GlU zEbu#dB~K)81{_0+KJw(qOQ8bYxzh6|=*~Wo6OX9rXiyay z-{sOV#xn ztNyNaldr726B~@ow~J-KhFqkG#n=Dn+H{kx`OA;FlKB#_(b)6^%Qy# z-Hkokqt*5NaJ#=szkyq*gE}(!Uklcqn=1U{Dw~y}om_UIuDk|y7CkN_<-udd2);uLtNrycH z8b~$^IsSv{1C?X9*&8e}Jsp~aZ<|`Vi=V!RGbtAt+4+6Zs1tv8h+X&;Xb-}$M60_k z5T(B~0oTXRTs6~GUTv}rP5-oKD#U(I_T2rf5tVJ!AH)Bgm(L-JW)+2cvpI_EI|DFTE1k^&)Ogx`=kXJRWk3gdwpBchDg z+XA?;_7x=E;+yCRaLkN5#mdVLMLI5brB&wRn3jcV;CjHXlW)&KFV;@=j!dfQI z1M!p$7l`*Byp>YQ6C7+j9S3rW+i@SmYu;=$pX@>|bSo0xiPxEz&%h-WyMS&5WsBC| z&6BvDc<@P1>G^~!c|^(KX3{e@lSO0BS14w0^Ya30i=*#~0hY}nfwQM_48I3f=Z~V; zCeLZOr@#&0@uoOU zUK7Zzd~g-xu5s+19?H{!B-7r_&pS@1E5Ajwjff>dLF|wxy63;_xm#~i0jR`-Q+s+r z1MLqz-C@mfnp!{418s?)v0VqrI1_-b*T^ow-}LEoty$k*>j^t>rwP!cXleVoqD@bq z1+~KH2-!S=XbR`swv4g68MffpzI&gutp=#efJ(F6wcBy3zYj`nb?R1-p#*&Q2w7i9 z3HEnB24NQ`tJyEJN2-LgY4;x?u(o7i?-Fgi+M90HI&?$_nHmjqg^Rnf&Z*H=muner ztk=IPFhV|LIa+x4PcQ1}eTq>ZNn>D#a-Tl=af|E%!@EWj-Ge^ed*sQCp~wSY-ni%x zAGTlbZ_3^}zyLp{?$X@n_}3`D#lt}G!kIO8xAi+-YZsAE+#dR3tEwL(y)WHTsAO`j zFb2t)jaTo72B-n?$Y5kzTkm=A0mx7E*ymX!fsP>M?on6>zscjpC!g6SZ0b`n(jg4Gs;v z?WZSDpvM>z&_22kTW)hM#S(eH_lsKm^%u9K&o!tnu#z(4e=G|?QLyPiJr5>7j5SOk$G@WqA2}pSV_+`^c^j zO45M+lFqp)uzQPx=OP}c+acP=g$cH{)40cVD3@f8xi9(j@!Sy=+T04bT|gD*fMpdr zfF5wlWgh27?oR=bz}AsPp6YpkJ4iGVM~rhy3wfmTJ*);`{l4oD2A@3FLu|A8;V(^s zMI4+Dnc2~i94{3pR>uXL{!`;2Mi;pzk5K=*FDLuI*O)1Of zR`^$6Q|3`sc>+(nsm=s3{BI520=@MDxRY^| ze&#oPS>D%p_sI4(T&r2FuSpjikP{ z7VVQB50qP_dPIrruOo`jsInHLyQf9Y$0PKgR;DsjM341=bOU;I3gLF=p|=_*sK4OW zXRiuwE(Z!lO0+M7QJ(Er5YG{Tj%~Jd|b)KZjGu)Sk<1I{? zEp2by$>R(!O>2W{;VXVQy!VY`CaQFClWaeAmsos z^hg7D1hyJmb_|T(#tBqKP%L|l1mw=M^TUTnK9@gjZ+PDYG%4se^Kp6HbZPR&mI4vawT$* z#bbDXa*(t@{sK+*cGPhncBDSG8aU}#joM9#K?@ZyM$La}B=le4>Vm|aJgRoOvITtM z$*}WtWr_uv@D#xf78P(mK&7PHIg<;WtKA;l{2A9Jtm;4cY2nDk1}x$C+*l4$xqvGl z2wCuGpO2_pAo?|=TiZ$gB;f&rBoZxMb-TX#eu$6cek8egj<2`2NDpcHn5?$8T=kP9 zwn5S_s!w|Y7pxEkL`J&)P%q+>7zn0Ew`MwkOSeu2f!A`Xr&!IZ`GmXrzTr-V-5xe? zm$^|;3_E*DgIoz;u?An+nmgpTe}O&n3B$S0lH^K2pup}MQf$Ic2aNWPkM8`Yx8SRM z6zASknDOWDcZFvBxgM-^yP`qaI;`byFTBQyZ7_{jGi0P14#9m4w$+kWO!*8xF(KeZ zQEpWtdLq?O$4wkQ*MRQ*C| z>Vn92b_=uZHW|dIaJG}#1)1Ds- z{5nz)V&6un&Yt{pJ4nt@BHtmS8FE`2B7WJ05}7XFRtf`S9iMURv?;)Ap&Z*CEC~To zqKW+rD{kcna|(T|<6X&coG$r7(}iZ&&B=46)eWWzpU z!UE-Y82|6px6bIMx9RnQ4?)m{f}ooOVw}=;+BC(ac@w?)Ym5b6D}kbVp+b{!$O^e=h$qe);+8(SMw*$hqDP zgqoq~`zygEeiU<~1=qK1D$y4Q{eq7CSPrE$bnpdi-|NduH}_``f&w4kF6Fqh6z!_0%p>~f#94fqZm>KC3gr$K_!eY7Wc9OU z*BhOJ=}Tlm%-dg4$jsrr-zWXzOQIVIJA${@(b^+nPea(=i#XzE{>KY|hBNI|y0Cs$ z0A2hZZ~D~T)XmM;)Sq|@=>dP@yV>A3xlSoRBEf_ZpRK;m8`A#$v+P}X(?>mqaUGuh z(ip=M%_7_1Sd04&^KTE)9)zKpMI7EPJ_8MEX*e6-NT{x+WiLDDd*f(6Nm#a8;k`q7 zWX$JE>}(Hy81>1>7}S^J=QlTsoKKq~*W`Na`=UC(Z3&0zplbGGDtOPng@ur|+2$QI zG0;I6C?Ohm-5D#sG&k?PtYEau@uQ8J4V=aXV&`{Pr3{baGpyOKpFz{$1if}dCVn&I zC39Eekmr{@h8)9Ei4&7bhty`!A52)ocpQQOfpB1a)!i|*mB1zl(;#8&Pn%Y89p!oq zVs5(hc3n?{S1S!P62om}KG@#-iPIIUN1Ud*Uc*HS`=?80q_iz{HZWytOv zahrJBAPGy?Dm<6`5ThIQZm**HCm}T_Rn6xnfOM2 ze;3pNx9SQ2-aEWrL~lH)@}+0IMF%nakQ)3_%xV`DWgH`)`M7E-M%-l;BjdQusS^$+ zU(_X19PeDcXNvl)GeNw`eklygRLGqk9WcM>;A&IkRh3!)+40WDF5jTK%N4okmdb?z z!n1^>1^sXLShhEbRbzfr5DCXGPs`03Z*MwqSyf@<(aHGGmpCtfuaizCMZP-iodrh; z?3l0^0W$WSoHm1W!k@(E-UsBqxm2Psj{@?uCyLvlgl03i$%I*!`lpb;>$emX|L_m7 zRJzKp|5Q`qyNX=yPDYX}k85|IeU?Y8^PcB`m;A%`#6;!f>hoFc=nqAQGgZD>;`L{7 zEjX8W%{?VhyZ(N9g5p`|mI>&v;9YFR=JPNbzeR7;1Bq8B&7lc3~oJqd-!YN=N{VP%F`Hc8@;&+XbZ!3-X~Ac z>=1#)Qy{*VmblEY|KhDc%K)BIVQ9>^z!nRt?LgL;5Dncy5c0%7Pm5C5g4OS2u@n$vAz)8|PLCiBOY?MY5 zLTZPZ1qRG2J%y%93K|!vAQBpOs?)RQgQz)$#b7z0afS@UXSKwRY<}{-F33v=x$AZv ze)yoWt>K*%E{L+E^6X^hRIZ5EW!4_25MFUX_b$vrE-}9O`%qQ4i^x4gI71g->1&~` zeYn@6ywuV`4=|Dz41y@6@44S(ez<%HSOHJMxm#Ph&jSrJ|-nrosCJ z?W37_;L}jelCJ77oRTaH>2%kBR$z4uDgd5MwGQdTiWg3Z1XRU?rZy|=XzkCqSo&7?v{1GsZzZ9ip)ALIr-~c2+U2+0mGJ^X? z$sd&#FmY3xojXi^IAKsr44){gE`-9Q3nckWIQz4O({K`!gDT!R$}1ryijNXiMGOcr z@|K&Zn?tP#)}>^>f4m(FrzQ1Cf;ek+GC&(P7 zgL#B%+g}F45lf(e12V%*pD5+ zr%^sy5+7ws1%{%i){Kf}sa?GysCF1GzpczZm32sCOuI2-+n@c0Y?;;JmalljyN0Ho z5|%;J5z9Ks1~6KwrO3H;FJdX>qvZ)&uJi|?YbbF=_m%?yGX(ZjTWy|ynt9PPLnW6i zGfrObr*QH665%yz*6E+=%3qZTy_)dQplc!nov+e@z4RY&!<0717&~dtQ0hk_Rvr;c|=$rUp`a(m1ZVoq4Bdc-fPW zDPMzD75CkM-~DXtA+qZ!Vlo%i8JSE#PWnP@#D^E8VzZ?`y>M~mFt}%3VwfQm^Cn?}ee#is%VdKNm1v?Xhva&3{$yFXHVS{07tCy@RCUUd9B<3T+sxK?KUbA{N)wIznd-t9Qpf$*Ldm}Wn~7WM`&$N?H*>-l2#Lf%YvlNBFgpECfO9R>SN<0LPlcc z6Nm&G%CEPp?p7-^k~+Wq5TS2A*Xb0dL0tBsECa2hHSUg)-Y@J=gpeIiJ5Jbp8@)u; z+$7h5|DC>#`7IcN)m7Wq6i)*^MkY1%9A9x7Va?onvtz?d%@Y1Nk8l4p=UE)h}7HF@@hB`&Gv@$ZJb_FQNF&f{|B546ddOMzt+Axkm~>Y|4N9gkeL)R5)wsMM&>1v%7~1zcgiSR zR(57(i>T~bL`Fz5uf2=xy|?>2FM7W}pU?O2-=CLzU$575o%1-4$2sRUkRQ#(;$bv7 zPou4%u@Fx9DUc9rcy zNnh#dJvD$x0cE8Otpd3-SKaJZ9t|f<<9j&Yi(-xRpApzGL=3__UUuI62)~p(a>5s` z7c&z}&M>1Kg4x1!2dmCQvDdTFn(S4a&exN@2TPz$Q zGnnf}ZB&$!O=uCv|kyfgDGC;5XJ zXAe}_F;F8XFzVw>b!TM&7r#VC+X_M=)~K_S(qW#lKrVC(eTAJKdjpQ4~N~r@bZ|~VW z2L7@EkX0%IwyINaVpsy_1B~grz!{`$qJH__d8b~y7|L#W#MrwzV(^D>LkV4aiH})o z-w_v)2I0vQuxW|{b_0x`{h2cWLH&}uJC=62|CkGVgN>IvmTMyaj`nXI z`hEdO^#3`D$C6o(C4X}f23{3uY?cWd@4LqgcBlNNgFerJzZ z(l{M&I`yOFnquw_JBk>zKMTtja^hDB6gvSGf>+DX-~4YJS6h_+++k}f}+ z4PMXjx+N-<)1B49cS3_>Zfy4bLylhF_)xw$-fH{M&^bZeP^Wf=zSReVV??*)hRsXu zYlX8%KYkw z-RdgGwE-@C2e>dzTL950)WF&?ez+{`$I;I`mK71jn#hlJv@{CZ2okC$@t{E2D?c&0 z?-BD!{vHp{wr8x!>VOt*Tq9m2H00w8%RKuHHy*oiO|$tDLCkk6JXfv;S2x6rwI4X7 zNb@b!uW8@D<2H4L$7jh9TSER^y-laA$>vc;^U6*|9qTU$M0BOz73@7)lKUGpu`+z3 zH#pk^c@g>T{0c>t>(OrXXoV~n)FXRY{lW#*ckIgbLs^7q??0nbo$@>6tt!N*Ss3goL z+xy8PSTwFDU(H(;8ls2`IoAzGV68)a@=V*jh(-0(p$j*#ifti6X><#13^14uaHP5i zJxY8;iREv-t7{g`Dh^*Ct2J1= zuE&D9PcveDv|aE;krLmKrG)WCyBPW@UZE!<8{8M}c?sBi>&+|N&N8%FB1h)KSS7a6 zTj8f#T;XOfAO++?kuhel)#8*69WgZ1T&v zM6~h#O+dN%EP|%aH-sau?J(AaBpdZUNE^k(UUge_fgRW+v0Fg>y6-WI$!?8nW&Wi2 zZmrF&txKf=a|V0gDpWT~D(O?ekCSNN^IU|I?}%P{5pkNBq1P>f%>eN-_WTs+9h3KK zyIg+>7o}S$Nq2iLge=ogoRYH~U3*l`-=_0h=M+5PUg^XAgCbBfs^%eq8UryIJ= zHrB-{(TZHb7g3vnD1*2)8kXYh5z#ZQJXo*f6~l*0XCE*r?7uQ^sofy^NDc|{ih2{= zI+z$fBHj7-U1wV#lGEu;u|lmrH=dGqZlch-J*>!ELHwW4VC0 z<9#*#!F60ac}gy*;#mASQ8E1DzEUR}lK52JAXtW9l>|r!B)Qp%L~kxaRMFUQaPO6% zZuNanx8oO5zHAy*I9-8~g)48Cs9zb}XBv*E{~K*y8TtC>QCl34g+IQcPOY&f_LaS-bu{b`oB%t5 zs@Vmo_O5<_wG$+o+m>^>)Z0l>lpx`gW_LNiMdq17`l=3wE~pbA>=ugh1)sPLx~ZxE z(t!6GGohyb##*%8wE=peyRxM`%{=nlVdnnBg7b2LN&dwHQLnRVXFFjLCiLQzeJM~W zN_5I%QSHk2d~I)zzcXtiDh-qN6Amd-o^E@)zDor40)G&~z?sV;0ovk_#n%MuOxN#S z&X1L^aC#;(X{&!#r#5G^Yl_Fh8!w4Z8Shw3M?&m(v-Ew&f^%wfH<(5Rv>PMufMtE; zGyMmrt(5s?^Jao|DWw@VKd$g8bAAbXvBsQFjyae#sdQU|!13nKl5<-uQjZ;x1y^ll zYyiv4=Wg9lerA)zG+Va9HHu($3v<1Ge|B(K6~mi)T&n1} zeWxw>!()4gZ0hhRTiy69T&&?E&kA=ndFGOhGyNgG7SE3^;+Lqsc~H!MBnk|WF9FSm9!=Qqb5tr>yy4ZUTxj9G}1?gbXo@hLpsxQVf4%B%ZfF{#L9yM&1?)Ga^!O5zdg@W<(My|`NdusE7StlB^usPnn8^|MCt|<+W zxcp(a2`Bqb;PKTNA7Q7Swe?+Zx^l!sfw1-Nx*Gj=HI1J#~Pnn9XXnefY@JrP=66vv^B(vH%!-8<9>c${y zmPH4cMc7AOff4Z(P2fJ>X0Mq=tTT?Op<8QXpGaEXj6FuMAS&$-MSZ@z+$tgSsf9x2j5>C3N7hTo?sX4r?t4L+-o20@ z@R8L`@#D`Kqt5jC{Ud4#wB75NoRQU0(aX1IRuzPTwY@dzf9owvr&>yu3vYYf|Hd;v ze7wTrBW)BvQ3=I^AGG=aqSZdM^L{I3Lg`PMEv5LVuw~U97=I4p(TvEV8YjgHGG

zC?oOy=btZ>+BjiTl0Y!o?uIy<^I{(f4)eAN`W-b#D2%!O;kd=fJ zlH2@-79#I9qMck!KW7hXUfu|5=Tf;Byu=V>3(LvDj90uw)1Rq%?KE0Bs3Rs;I(#dy#0vVO84p8Gld7Hu?BO;f?r782%(cO7e}N6!HU zwO1=2z!kL`n=?9!j#D#k{_^&~CiC|c>uWfw-%#}AZ+EoxPo%wiHOh5An)$WsMX)D# zWiCM2Dhv)lbu4)ifug8bBI}9=e`6>iTViAS3;TaoqJ1Bi={nT7g@ZJ9nEmue|n) z6wwDdX--LT#j&IHS78!IH+(u2=SqatLh!zS?)=tDV)Kk)aHoQ{pTx!v%!~A#-f;oi z6F1gIS=s~)MHP0g_=z89AM&V}$!BIbwUMJZCr1ieQjMykxpLwOULN9}v7WT=PRSty zdTjkZy)$s~Qay8=v&yy#s7DtNdf16l6UzBnd_t&FlCu(wl4e*3rk^DfIQ;6l{bj)s zdPdv{oFwv~u9FOF-NURv(nSFTN9>qME|*h1DA0~Msi(Mdy$aoDH1){-uV7Pi=e2s9 zTG?RIDYUZX`TkEOp%02r6chl#6JSTdt)V+=p&-tk6TcV)Kgp<0>Z|cc4=cis`j~7cI>Z4gxr|?+47|oaY31C%l2_DKf zz?+;dyQYw%nf2_pq%Z_(<-&eyCX=U@)N9_gog_{^o!Xq2&zQPi9ee#tO8;BDmk4#g z> z9ok})H<3ZC!XVyZ>K$aR`G_C8(o*W|PK9u=x|^u9Ed8z4d-?u3^14|QDpiFfqXqQ5 zh=W=;n7Uo6I81NRe=+6kbXEd2>-psmYSN@A&LB_L>6p?!}?~PtHSY*sK zLewn7)uX9Dr_K-F%VAM^&e8WI@wp*yT?fD)rx2mR>g!*`yV!J+Wy?G9RcW_6J|+{Q zL{!n2R#~~opuH7hyH{8_T8AeETU|b1xzi|U}$E+kUCT`KO73` zoq@d}qgXYcx-j`T$2?nkKwC`6PR;on8H$tiBr$KLrc#is)X16Cj48gsY5DMv{{w2o zBAY47837{kNj3>X(^W$Vguf^>#21_3q(LCT9HLj9n`JGUmtDjhzroJE8{o}hEw`D?SPkAxTLP#8O&_Ri$HI;7i78y&Zl=;kUebrmd%|Y z-=5wCY^DF9b1x3lCBtf)j$_}1OI14PFB)Ey*5vIX;79W9D4+&!1ywzeqND8jxu ztI+MCb8;|D`F&;xx6ZdAEbrIpl?jl$7}`L93Duu^o~N#NHRdoPKbCVo>)B;f(?oat zI!?B0yaH)`H1e?!NK~h%>`~plx7I7WJiDED|6m}>QoFKJZdPWzU;yWDM?yO}e(-wv zFn}o9n8%5-#Z(0v!FJUT*6F!__JC*;?2u~lS*D_eVmMjvI8!oAu{g?*(lP91L*-oL zjs`?^Udin>n13-vf1BzOPv$jTl)Bde|1*a&4!)iL`z$kFf5V-=q4)Bo=4#IUgXrjG+ls7|kX67!|Cuv~)M~R6{W0)*F=BHU>QwrUZ9lb>m)9Ga{W9KM_ClyP zxL407K>Z{)!B?e^V_kQ)ttLeTu!k%w?dR~G5IZSpIM@wD9_MTI*#8S0@gM+zU08l2 znb^bn-9W_=k5nV-9?lUaV>^ofsGQRB=p05g_L&p7X4B1s2fW2rG|@t+9cue*@ae3(6?z!YVw0uYA&&+w z+V>5}PnPd7rp~DTBu!6(<9Z7DzhHkSK7J8!r`L%P@mP^E8#?u`QDtU9y$fqk^A z_E{>mWW`ZlAb)<3;a47`$Cs6o&6Vxy+?hiduk|+*()suW8{W-S!mcb#_a{$wcdGL6 zvX>}h)E_Dl$8!quFEgl$qk@t)1HCu7&HBv5-h~sZ(W6il+Q>Tspy{}ORNoVuYD?iB zCMgsD8?L&}ry@*4?W&K_?Vw2taoA!tddUKSPmG;3EXf;yG z%1q$!a^0pV0bgGFyT7VYq6C&L{+Fyc9>{AAi087&weyEs;?sypi4payxq{D9mzjqn z-zK@!biT7`%c3fpdN@p(a_6khxU6_ePTsJ~k$|t5V{>PVloPu$(y#jT3Ylrt+$btq zGEp*5TOV+)DmHL%18Pu2DaLv{NoOG=;K3$}@-?6Mj*`}aA&dO!Fv-+Y$=biNX^x=Rf|@n-K51NSKdd`(Am1pp9uGHyOoY%R@-`GZ~jB8Eh8+jJ?k{< zSv%5RWr&Zzx@*TL8x6Fec-XRxT#c{`0$J(zOMfgRzX$qDG?4fpMj4|c1d@nH9M$pW zeV}(LK0S13bF1B0{&=V4^*b{%_C$`lz`Cq%ZBrForN25_sh}IYr~m4io@?;137{%r!gXxt zSK9sFeR;ZG?*xC;gCOM0{J-v7Y_TQ`Ts>fpdRNk3!p3Iu_z5G0if@I_v$3y8vG6Y; z!VrHqrnRO|^Y|6S%r)g@^zJ2$>afWckLEtl9unR2N=CV) zvnMv&n#3mo8<;LiwLWa*0NU*h-Iff*Q1vx>?0;^2E5Z&uF?ku%e`NdCqRor&%CE}U ziki^p>M9302#gcMW`CXdqM!Td2)<-$WQ)qL3r&88#J1umcU>cOO$K*Z$YAEGHeP@; zG6q6D`Gi8 zn@4Avr$?t#UaD35L&+#rw7j0vEo^V3m^^w5NPwB}^|=|Qux|QMUp{Xw2tUSrKot;7gQHRR{=i4E@=`g`x2u604TTUwt>lFD06IwJ z`DLL2ea7CV zHyYllL<08P=1OblcXoOE4~jvNU-+s7JwpzLgrN!&}16GIRu$sE} z@~a3W7eiySJCVIpah-Ny@vHCjd;4wn@TD695I#UM#Yqqsu5Y4RCExN!~IA+eW*skW81?+Y=xp?FN$*7B{Zh$-BRZN1gQ{4M+|zlze$YE;*gUD znU2WX-83a#|L}~Iojp3&*W%v`{De32ls9g}m2$AoPzK5upEgkXJ;M#l?*c;qcGZO?PtrlIB-zpv%AyAi#}~v%vm~ zCsoT_?A8q!(#t9>YI5CILMG!{V;ZcY6r>H=b6|d?roH1NP{Y@Gg?w@_vl)7xYXwgq5`31cAs`2aMSyJ(e8n$qTfnDWt|an2WiS zD3p>HJd=d-h&NBeLYc0^eNWseao_JyL$E$`rKX(55|5j#dE6>(%vG&TAycv*28bX= z;HosBJIdYvBP+GR2FW|v7U~$)H#e>*~h8Y?}f*NZ1@s%5xjRN~te7pJ>o8 zoXyO6?!T}VtkVXb`#G$P>KF)~YOpduK_h8Ovq+j!$f#}-#K~3qKf#Qg=ogoW#1wVF zBAKq>RT7-A52L(>mzk8Yp^Q)`a_YSkE$*~=Bn43L8KL^Fegb2gE+*qXc`UG=vB3<(AIqkwIN4Ad)5#{`CvM5Fkc5^6+1p!i(gTJ!=nTF)7?D-F_+s}G7<12R%wSXem)55C70LOsFD6tMt;5b(K zS6Nv!7T{6ZUVO%wQkn@jd0EU&go-hvmBiX-%}fN`uytH2KCt3}Kriy0Ka#@c$gOzA zpb;{{n8boLfBqS<-Vp#4-a-8@Pe~rf_}_emE#Gkv*&@MR>3_^-Q;4SQC-VJI;^ms8 zmEf>^xFtT4m(>R8ks`OthF~&-;F(l2Xo!P{kFEPpMhDdnzR0k&Szt>}@lHYzQXbPN zKM)N6lS0<4_7nCIM~yan>ZOq@*nPD_MJpvazeU3K_+r?m?-kf=28gn(q^dnX$6m}< z{Xe!%km~JM+;w0n0MA)pfaMInstT||ycIGd-1zNY{!aam;sM!<1N7stB@0;jsY?CW z<4v5`C1}u!yaC^0;PZ*sxm^QC$gcI$*? znEjfYs9iW+O2~lUt$c#fGVprKtSsFxVgAUt0ob@H#yDuHObxZF>YEy!Ys;Qo-}+67 zHHL`@ZXkkJaU}3lUD|ni*Rk=&k*`u8-ys3b%VYm$6U-fU6$LIhHa=llSR$y?0 zZ1nL1ty)GyJ6BWq`2Lu7sW*y?++ULOgB$_a^*3R<;T=!#bq)ABCEB;p%T7NK07cA{ z`&CGM<51EfEG37g80Su%!0NPC)Wgu6pNYdOf5L@KjCpSUUoO)+>wr2B7Yta++(KiBJ>1s)e#rYs$MMpQ{hZ zpnxX^J`|{CvH}S%DlIQt7bE37ONa*$)w9Q@KW(0dx}sfaKZ9?{ynR-2vX1@iGc z@-b3In_b8W9CnH79~f#1*F0--7&3AnWaM~UkZAknzNj6cL@{gf!Hf_fU&EV1xWH%C zgh47Em!V58`Yxe0yuqhyZPOj;^pyf0vjp9GQC12W@*p_367<{RfqtOR_(Yt!6u)Wjz@lQ zwMa#RqHeAd;V8>k4>fS)My|+Cz?e2Ss>j4y + + + + database-plugins-parent + io.cdap.plugin + 1.13.0-SNAPSHOT + + + Databricks plugin + databricks-plugin + 4.0.0 + + + 3.4.1 + + + + + io.cdap.cdap + cdap-etl-api + + + io.cdap.plugin + database-commons + ${project.version} + + + io.cdap.plugin + hydrator-common + + + com.google.guava + guava + + + + + com.databricks + databricks-jdbc + ${databricks-jdbc.version} + test + + + io.cdap.plugin + database-commons + ${project.version} + test-jar + test + + + io.cdap.cdap + hydrator-test + + + io.cdap.cdap + cdap-data-pipeline3_2.12 + + + junit + junit + + + org.mockito + mockito-core + test + + + io.cdap.cdap + cdap-api + provided + + + + + + + io.cdap + cdap-maven-plugin + + + org.apache.felix + maven-bundle-plugin + 5.1.2 + true + + + <_exportcontents> + io.cdap.plugin.databricks.*; + io.cdap.plugin.db.source.*; + org.apache.commons.lang; + org.apache.commons.logging.*; + org.codehaus.jackson.* + + *;inline=false;scope=compile + true + lib + + + + + package + + bundle + + + + + + + diff --git a/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConnector.java b/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConnector.java new file mode 100644 index 000000000..fc54dd9e3 --- /dev/null +++ b/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConnector.java @@ -0,0 +1,162 @@ +/* + * Copyright © 2026 Cask Data, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not + * use this file except in compliance with the License. You may obtain a copy of + * the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations under + * the License. + */ + +package io.cdap.plugin.databricks; + +import io.cdap.cdap.api.annotation.Category; +import io.cdap.cdap.api.annotation.Description; +import io.cdap.cdap.api.annotation.Name; +import io.cdap.cdap.api.annotation.Plugin; +import io.cdap.cdap.api.data.format.StructuredRecord; +import io.cdap.cdap.etl.api.batch.BatchSource; +import io.cdap.cdap.etl.api.connector.Connector; +import io.cdap.cdap.etl.api.connector.ConnectorSpec; +import io.cdap.cdap.etl.api.connector.ConnectorSpecRequest; +import io.cdap.cdap.etl.api.connector.PluginSpec; +import io.cdap.plugin.common.Constants; +import io.cdap.plugin.common.ReferenceNames; +import io.cdap.plugin.common.db.DBConnectorPath; +import io.cdap.plugin.db.NoOpCommitConnection; +import io.cdap.plugin.db.SchemaReader; +import io.cdap.plugin.db.TransactionIsolationLevel; +import io.cdap.plugin.db.connector.AbstractDBSpecificConnector; +import io.cdap.plugin.db.connector.DBSpecificPath; +import org.apache.hadoop.io.LongWritable; +import org.apache.hadoop.mapreduce.lib.db.DBWritable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.sql.Connection; +import java.sql.SQLException; +import java.util.HashMap; +import java.util.Map; + +/** + * Databricks Database Connector that connects to Databricks database via JDBC. + */ +@Plugin(type = Connector.PLUGIN_TYPE) +@Name(DatabricksConstants.PLUGIN_NAME) +@Description("Connection to access data in Databricks using JDBC.") +@Category("Database") +public class DatabricksConnector extends AbstractDBSpecificConnector { + public static final String NAME = DatabricksConstants.PLUGIN_NAME; + private final DatabricksConnectorConfig config; + + private static final Logger LOG = LoggerFactory.getLogger(DatabricksConnector.class); + + public DatabricksConnector(DatabricksConnectorConfig config) { + super(config); + this.config = config; + } + + @Override + protected DBConnectorPath getDBConnectorPath(String path) throws IOException { + return DBSpecificPath.of(path, supportSchema()); + } + + @Override + protected Connection getConnection(DBConnectorPath path) { + Connection connection = super.getConnection(path); + try { + connection.setTransactionIsolation(Connection.TRANSACTION_READ_UNCOMMITTED); + } catch (SQLException e) { + LOG.warn("Failed to set transaction isolation level to READ_UNCOMMITTED", e); + } + return new NoOpCommitConnection(connection); + } + + @Override + protected Connection getConnection() { + Connection connection = super.getConnection(); + try { + connection.setTransactionIsolation(Connection.TRANSACTION_READ_UNCOMMITTED); + } catch (SQLException e) { + LOG.warn("Failed to set transaction isolation level to READ_UNCOMMITTED", e); + } + return new NoOpCommitConnection(connection); + } + + @Override + public boolean supportSchema() { + return true; + } + + @Override + protected Class getDBRecordType() { + return DatabricksDBRecord.class; + } + + @Override + public StructuredRecord transform(LongWritable longWritable, DatabricksDBRecord record) { + return record.getRecord(); + } + + @Override + protected SchemaReader getSchemaReader(String sessionID) { + return new DatabricksSchemaReader(sessionID); + } + + @Override + protected String getTableName(String database, String schema, String table) { + if (database == null && schema == null) { + return String.format("`%s`", table); + } + if (database == null) { + return String.format("`%s`.`%s`", schema, table); + } + if (schema == null) { + return String.format("`%s`.`%s`", database, table); + } + return String.format("`%s`.`%s`.`%s`", database, schema, table); + } + + @Override + protected String getRandomQuery(String tableName, int limit) { + return String.format("SELECT * FROM %s LIMIT %d", tableName, limit); + } + + @Override + protected void setConnectorSpec(ConnectorSpecRequest request, DBConnectorPath path, + ConnectorSpec.Builder builder) { + Map sourceProperties = new HashMap<>(); + setConnectionProperties(sourceProperties, request); + builder.addRelatedPlugin(new PluginSpec(DatabricksConstants.PLUGIN_NAME, + BatchSource.PLUGIN_TYPE, sourceProperties)); + + String schema = path.getSchema(); + sourceProperties.put(DatabricksSource.DatabricksSourceConfig.NUM_SPLITS, "1"); + sourceProperties.put(DatabricksSource.DatabricksSourceConfig.FETCH_SIZE, + DatabricksSource.DatabricksSourceConfig.DEFAULT_FETCH_SIZE); + String table = path.getTable(); + if (table == null) { + return; + } + sourceProperties.put(DatabricksSource.DatabricksSourceConfig.IMPORT_QUERY, + getTableQuery(path.getDatabase(), schema, table)); + sourceProperties.put(Constants.Reference.REFERENCE_NAME, ReferenceNames.cleanseReferenceName(table)); + } + + @Override + protected boolean isAutoCommitEnabled() { + return true; + } + + @Override + protected String getTransactionIsolationLevel() { + return TransactionIsolationLevel.Level.TRANSACTION_READ_UNCOMMITTED.name(); + } +} diff --git a/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConnectorConfig.java b/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConnectorConfig.java new file mode 100644 index 000000000..88756dc06 --- /dev/null +++ b/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConnectorConfig.java @@ -0,0 +1,129 @@ +/* + * Copyright © 2026 Cask Data, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not + * use this file except in compliance with the License. You may obtain a copy of + * the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations under + * the License. + */ + +package io.cdap.plugin.databricks; + +import com.google.common.base.Strings; +import io.cdap.cdap.api.annotation.Description; +import io.cdap.cdap.api.annotation.Macro; +import io.cdap.cdap.api.annotation.Name; +import io.cdap.plugin.db.ConnectionConfig; +import io.cdap.plugin.db.connector.AbstractDBConnectorConfig; + +import javax.annotation.Nullable; + +/** + * Configuration for Databricks connector + */ +public class DatabricksConnectorConfig extends AbstractDBConnectorConfig { + + public static final String HTTP_PATH = "httpPath"; + + @Name(ConnectionConfig.HOST) + @Description("The server hostname of the Databricks cluster or SQL warehouse.") + @Macro + private String host; + + @Name(ConnectionConfig.PORT) + @Description("Database port number. Default is 443.") + @Macro + @Nullable + private Integer port; + + @Name(HTTP_PATH) + @Description("The HTTP Path for the Databricks cluster or SQL warehouse.") + @Macro + private String httpPath; + + @Name(ConnectionConfig.DATABASE) + @Description("Database or Catalog name to connect to.") + @Macro + @Nullable + private String database; + + public DatabricksConnectorConfig(@Nullable @Name(ConnectionConfig.USER) String user, + @Nullable @Name(ConnectionConfig.PASSWORD) String password, + @Name(ConnectionConfig.JDBC_PLUGIN_NAME) String jdbcPluginName, + @Nullable @Name(ConnectionConfig.CONNECTION_ARGUMENTS) String connectionArguments, + @Name(ConnectionConfig.HOST) String host, + @Name(DatabricksConnectorConfig.HTTP_PATH) String httpPath, + @Nullable @Name(ConnectionConfig.DATABASE) String database, + @Nullable @Name(ConnectionConfig.PORT) Integer port) { + this.user = user; + this.password = password; + this.jdbcPluginName = jdbcPluginName; + this.connectionArguments = connectionArguments; + this.host = host; + this.httpPath = httpPath; + this.database = database; + this.port = port; + } + + @Nullable + @Override + public String getUser() { + if (Strings.isNullOrEmpty(user) && !Strings.isNullOrEmpty(password)) { + return "token"; + } + return user; + } + + @Override + public java.util.Properties getConnectionArgumentsProperties() { + return getConnectionArgumentsProperties(connectionArguments, getUser(), getPassword()); + } + + @Nullable + public String getDatabase() { + return database; + } + + public String getHost() { + return host; + } + + public int getPort() { + return port == null ? 443 : port; + } + + public String getHttpPath() { + return httpPath; + } + + @Override + public String getConnectionString() { + if (database != null && !database.trim().isEmpty()) { + return String.format( + DatabricksConstants.DATABRICKS_DB_CONNECTION_STRING_FORMAT, + host, + getPort(), + database, + httpPath); + } + return String.format( + DatabricksConstants.DATABRICKS_CONNECTION_STRING_FORMAT, + host, + getPort(), + httpPath); + } + + @Override + public boolean canConnect() { + return super.canConnect() && !containsMacro(ConnectionConfig.HOST) && + !containsMacro(ConnectionConfig.PORT) && !containsMacro(HTTP_PATH) && + !containsMacro(ConnectionConfig.DATABASE); + } +} diff --git a/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConstants.java b/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConstants.java new file mode 100644 index 000000000..b82d88234 --- /dev/null +++ b/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConstants.java @@ -0,0 +1,31 @@ +/* + * Copyright © 2026 Cask Data, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not + * use this file except in compliance with the License. You may obtain a copy of + * the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations under + * the License. + */ + +package io.cdap.plugin.databricks; + +/** Databricks constants. */ +public final class DatabricksConstants { + + private DatabricksConstants() { + } + + public static final String PLUGIN_NAME = "Databricks"; + public static final String DRIVER_CLASS_NAME = "com.databricks.client.jdbc.Driver"; + public static final String DATABRICKS_CONNECTION_STRING_FORMAT = + "jdbc:databricks://%s:%d;HttpPath=%s;"; + public static final String DATABRICKS_DB_CONNECTION_STRING_FORMAT = + "jdbc:databricks://%s:%d/%s;HttpPath=%s;"; +} diff --git a/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksDBRecord.java b/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksDBRecord.java new file mode 100644 index 000000000..2940c7db6 --- /dev/null +++ b/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksDBRecord.java @@ -0,0 +1,65 @@ +/* + * Copyright © 2026 Cask Data, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not + * use this file except in compliance with the License. You may obtain a copy of + * the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations under + * the License. + */ + +package io.cdap.plugin.databricks; + +import io.cdap.cdap.api.data.format.StructuredRecord; +import io.cdap.cdap.api.data.schema.Schema; +import io.cdap.plugin.db.DBRecord; +import io.cdap.plugin.db.SchemaReader; + +import java.sql.ResultSet; +import java.sql.ResultSetMetaData; +import java.sql.SQLException; + +/** + * Writable class for Databricks Source + */ +public class DatabricksDBRecord extends DBRecord { + + /** + * Used in map-reduce. Do not remove. + */ + @SuppressWarnings("unused") + public DatabricksDBRecord() { + } + + @Override + protected SchemaReader getSchemaReader() { + return new DatabricksSchemaReader(); + } + + @Override + protected void handleField(ResultSet resultSet, StructuredRecord.Builder recordBuilder, Schema.Field field, + int columnIndex, int sqlType, int sqlPrecision, int sqlScale) throws SQLException { + ResultSetMetaData metadata = resultSet.getMetaData(); + String columnTypeName = metadata.getColumnTypeName(columnIndex); + + if (columnTypeName != null && (columnTypeName.equalsIgnoreCase("VARIANT") || + columnTypeName.equalsIgnoreCase("ARRAY") || columnTypeName.equalsIgnoreCase("MAP") || + columnTypeName.equalsIgnoreCase("STRUCT") || columnTypeName.equalsIgnoreCase("JSON"))) { + Object value = resultSet.getObject(columnIndex); + if (value != null) { + recordBuilder.set(field.getName(), value.toString()); + } else { + recordBuilder.set(field.getName(), null); + } + return; + } + + setField(resultSet, recordBuilder, field, columnIndex, sqlType, sqlPrecision, sqlScale); + } +} diff --git a/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksSchemaReader.java b/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksSchemaReader.java new file mode 100644 index 000000000..9dbf8bfd6 --- /dev/null +++ b/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksSchemaReader.java @@ -0,0 +1,79 @@ +/* + * Copyright © 2026 Cask Data, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not + * use this file except in compliance with the License. You may obtain a copy of + * the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations under + * the License. + */ + +package io.cdap.plugin.databricks; + +import io.cdap.cdap.api.data.schema.Schema; +import io.cdap.plugin.db.CommonSchemaReader; + +import java.sql.ResultSetMetaData; +import java.sql.SQLException; +import java.sql.Types; + +/** + * Databricks Schema Reader class + */ +public class DatabricksSchemaReader extends CommonSchemaReader { + + private final String sessionID; + + public DatabricksSchemaReader() { + this(null); + } + + public DatabricksSchemaReader(String sessionID) { + super(); + this.sessionID = sessionID; + } + + @Override + public Schema getSchema(ResultSetMetaData metadata, int index) throws SQLException { + String typeName = metadata.getColumnTypeName(index); + + if (typeName != null) { + if (typeName.equalsIgnoreCase("INT") || typeName.equalsIgnoreCase("INTEGER") || + typeName.equalsIgnoreCase("SMALLINT") || typeName.equalsIgnoreCase("TINYINT")) { + return Schema.of(Schema.Type.INT); + } + if (typeName.equalsIgnoreCase("BIGINT")) { + return Schema.of(Schema.Type.LONG); + } + if (typeName.equalsIgnoreCase("TIMESTAMP") || typeName.equalsIgnoreCase("TIMESTAMP_NTZ") || + typeName.equalsIgnoreCase("TIMESTAMPTZ")) { + return Schema.of(Schema.LogicalType.DATETIME); + } + if (typeName.equalsIgnoreCase("DATE")) { + return Schema.of(Schema.LogicalType.DATE); + } + if (typeName.equalsIgnoreCase("VARIANT") || typeName.equalsIgnoreCase("ARRAY") || + typeName.equalsIgnoreCase("MAP") || typeName.equalsIgnoreCase("STRUCT") || + typeName.equalsIgnoreCase("JSON")) { + return Schema.of(Schema.Type.STRING); + } + } + + return super.getSchema(metadata, index); + } + + @Override + public boolean shouldIgnoreColumn(ResultSetMetaData metadata, int index) throws SQLException { + if (sessionID == null) { + return false; + } + String columnName = metadata.getColumnName(index); + return ("c_" + sessionID).equals(columnName) || ("sqn_" + sessionID).equals(columnName); + } +} diff --git a/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksSource.java b/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksSource.java new file mode 100644 index 000000000..51aa810f8 --- /dev/null +++ b/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksSource.java @@ -0,0 +1,153 @@ +/* + * Copyright © 2026 Cask Data, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not + * use this file except in compliance with the License. You may obtain a copy of + * the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations under + * the License. + */ + +package io.cdap.plugin.databricks; + +import com.google.common.annotations.VisibleForTesting; +import io.cdap.cdap.api.annotation.Description; +import io.cdap.cdap.api.annotation.Macro; +import io.cdap.cdap.api.annotation.Metadata; +import io.cdap.cdap.api.annotation.MetadataProperty; +import io.cdap.cdap.api.annotation.Name; +import io.cdap.cdap.api.annotation.Plugin; +import io.cdap.cdap.api.data.schema.Schema; +import io.cdap.cdap.etl.api.FailureCollector; +import io.cdap.cdap.etl.api.batch.BatchSource; +import io.cdap.cdap.etl.api.batch.BatchSourceContext; +import io.cdap.cdap.etl.api.connector.Connector; +import io.cdap.plugin.common.Asset; +import io.cdap.plugin.common.ConfigUtil; +import io.cdap.plugin.common.LineageRecorder; +import io.cdap.plugin.db.ConnectionConfigAccessor; +import io.cdap.plugin.db.SchemaReader; +import io.cdap.plugin.db.TransactionIsolationLevel; +import io.cdap.plugin.db.config.AbstractDBSpecificSourceConfig; +import io.cdap.plugin.db.source.AbstractDBSource; +import io.cdap.plugin.util.DBUtils; +import org.apache.hadoop.mapreduce.lib.db.DBWritable; + +import java.io.IOException; +import java.util.Collections; +import java.util.Map; +import javax.annotation.Nullable; + +/** + * Batch source to read from a Databricks database. + */ +@Plugin(type = BatchSource.PLUGIN_TYPE) +@Name(DatabricksConstants.PLUGIN_NAME) +@Description( + "Reads from a Databricks table using a configurable SQL query." + + " Outputs one record for each row returned by the query.") +@Metadata(properties = {@MetadataProperty(key = Connector.PLUGIN_TYPE, value = DatabricksConnector.NAME)}) +public class DatabricksSource extends AbstractDBSource { + + private final DatabricksSourceConfig databricksSourceConfig; + + public DatabricksSource(DatabricksSourceConfig databricksSourceConfig) { + super(databricksSourceConfig); + this.databricksSourceConfig = databricksSourceConfig; + } + + @Override + protected SchemaReader getSchemaReader() { + return new DatabricksSchemaReader(); + } + + @Override + protected Class getDBRecordType() { + return DatabricksDBRecord.class; + } + + @Override + protected String createConnectionString() { + DatabricksConnectorConfig connection = databricksSourceConfig.getConnection(); + return connection == null ? null : connection.getConnectionString(); + } + + @Override + protected LineageRecorder getLineageRecorder(BatchSourceContext context) { + DatabricksConnectorConfig connection = databricksSourceConfig.getConnection(); + String host = connection == null ? null : connection.getHost(); + int port = connection == null ? 443 : connection.getPort(); + String database = connection == null ? null : connection.getDatabase(); + String fqn = DBUtils.constructFQN("databricks", host, port, database, + databricksSourceConfig.getReferenceName()); + Asset.Builder assetBuilder = Asset.builder(databricksSourceConfig.getReferenceName()).setFqn(fqn); + return new LineageRecorder(context, assetBuilder.build()); + } + + @Override + public ConnectionConfigAccessor getConnectionConfigAccessor(String driverClassName, + Schema schemaFromDB, + FailureCollector collector) throws IOException { + ConnectionConfigAccessor configAccessor = + super.getConnectionConfigAccessor(driverClassName, schemaFromDB, collector); + configAccessor.setAutoCommitEnabled(true); + return configAccessor; + } + + /** + * Databricks source config. + */ + public static class DatabricksSourceConfig extends AbstractDBSpecificSourceConfig { + + @Name(ConfigUtil.NAME_USE_CONNECTION) + @Nullable + @Description("Whether to use an existing connection.") + private Boolean useConnection; + + @Name(ConfigUtil.NAME_CONNECTION) + @Macro + @Nullable + @Description("The existing connection to use.") + private DatabricksConnectorConfig connection; + + @Override + public Map getDBSpecificArguments() { + return Collections.emptyMap(); + } + + @VisibleForTesting + public DatabricksSourceConfig(@Nullable Boolean useConnection, + @Nullable DatabricksConnectorConfig connection) { + this.useConnection = useConnection; + this.connection = connection; + } + + @Override + public String getTransactionIsolationLevel() { + return TransactionIsolationLevel.Level.TRANSACTION_READ_UNCOMMITTED.name(); + } + + @Override + public Integer getFetchSize() { + Integer fetchSize = super.getFetchSize(); + return fetchSize == null ? Integer.parseInt(DEFAULT_FETCH_SIZE) : fetchSize; + } + + @Override + protected DatabricksConnectorConfig getConnection() { + return connection; + } + + @Override + public void validate(FailureCollector collector) { + ConfigUtil.validateConnection(this, useConnection, connection, collector); + super.validate(collector); + } + } +} diff --git a/databricks-plugin/src/test/java/io/cdap/plugin/databricks/DatabricksConnectorUnitTest.java b/databricks-plugin/src/test/java/io/cdap/plugin/databricks/DatabricksConnectorUnitTest.java new file mode 100644 index 000000000..8a71e8679 --- /dev/null +++ b/databricks-plugin/src/test/java/io/cdap/plugin/databricks/DatabricksConnectorUnitTest.java @@ -0,0 +1,70 @@ +/* + * Copyright © 2026 Cask Data, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not + * use this file except in compliance with the License. You may obtain a copy of + * the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations under + * the License. + */ + +package io.cdap.plugin.databricks; + +import org.junit.Assert; +import org.junit.Test; + +/** + * Unit tests for {@link DatabricksConnector} + */ +public class DatabricksConnectorUnitTest { + + private static final DatabricksConnector CONNECTOR = new DatabricksConnector(new DatabricksConnectorConfig( + "token", "password", "jdbc", "", "dbc-xxx.cloud.databricks.com", + "sql/1.0/warehouses/xxx", "main", 443)); + + @Test + public void testGetTableName() { + Assert.assertEquals("`main`.`default`.`my_table`", + CONNECTOR.getTableName("main", "default", "my_table")); + Assert.assertEquals("`default`.`my_table`", + CONNECTOR.getTableName(null, "default", "my_table")); + Assert.assertEquals("`my_table`", + CONNECTOR.getTableName(null, null, "my_table")); + } + + @Test + public void testGetRandomQuery() { + Assert.assertEquals("SELECT * FROM `main`.`default`.`my_table` LIMIT 10", + CONNECTOR.getRandomQuery("`main`.`default`.`my_table`", 10)); + } + + @Test + public void testGetDBRecordType() { + Assert.assertEquals("class io.cdap.plugin.databricks.DatabricksDBRecord", + CONNECTOR.getDBRecordType().toString()); + } + + @Test + public void testConnectionString() { + DatabricksConnectorConfig config = new DatabricksConnectorConfig( + "token", "secret", "jdbc", "", "dbc-xxx.cloud.databricks.com", + "sql/1.0/warehouses/xxx", "main", 443); + Assert.assertEquals( + "jdbc:databricks://dbc-xxx.cloud.databricks.com:443/main;HttpPath=sql/1.0/warehouses/xxx;", + config.getConnectionString()); + + DatabricksConnectorConfig configNoDb = new DatabricksConnectorConfig( + "token", "secret", "jdbc", "", "dbc-xxx.cloud.databricks.com", + "sql/1.0/warehouses/xxx", null, 443); + Assert.assertEquals( + "jdbc:databricks://dbc-xxx.cloud.databricks.com:443;HttpPath=sql/1.0/warehouses/xxx;", + configNoDb.getConnectionString()); + } + +} diff --git a/databricks-plugin/src/test/java/io/cdap/plugin/databricks/DatabricksSchemaReaderTest.java b/databricks-plugin/src/test/java/io/cdap/plugin/databricks/DatabricksSchemaReaderTest.java new file mode 100644 index 000000000..2bf486099 --- /dev/null +++ b/databricks-plugin/src/test/java/io/cdap/plugin/databricks/DatabricksSchemaReaderTest.java @@ -0,0 +1,90 @@ +/* + * Copyright © 2026 Cask Data, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not + * use this file except in compliance with the License. You may obtain a copy of + * the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations under + * the License. + */ + +package io.cdap.plugin.databricks; + +import io.cdap.cdap.api.data.schema.Schema; +import org.junit.Assert; +import org.junit.Test; + +import java.lang.reflect.Proxy; +import java.sql.ResultSetMetaData; +import java.sql.SQLException; +import java.util.Map; + +public class DatabricksSchemaReaderTest { + + private ResultSetMetaData createMockMetadata(Map columnTypeNames, + Map columnNames) { + return (ResultSetMetaData) Proxy.newProxyInstance( + ResultSetMetaData.class.getClassLoader(), + new Class[]{ResultSetMetaData.class}, + (proxy, method, args) -> { + if ("getColumnTypeName".equals(method.getName())) { + int index = (Integer) args[0]; + return columnTypeNames.get(index); + } + if ("getColumnName".equals(method.getName())) { + int index = (Integer) args[0]; + return columnNames.get(index); + } + return null; + } + ); + } + + @Test + public void testGetSchemaDatabricksTypes() throws SQLException { + DatabricksSchemaReader schemaReader = new DatabricksSchemaReader(); + Map typeNames = new java.util.HashMap<>(); + typeNames.put(1, "INT"); + typeNames.put(2, "BIGINT"); + typeNames.put(3, "TIMESTAMP"); + typeNames.put(4, "TIMESTAMP_NTZ"); + typeNames.put(5, "DATE"); + typeNames.put(6, "VARIANT"); + typeNames.put(7, "STRUCT"); + typeNames.put(8, "ARRAY"); + typeNames.put(9, "MAP"); + + ResultSetMetaData metadata = createMockMetadata(typeNames, java.util.Collections.emptyMap()); + + Assert.assertEquals(Schema.of(Schema.Type.INT), schemaReader.getSchema(metadata, 1)); + Assert.assertEquals(Schema.of(Schema.Type.LONG), schemaReader.getSchema(metadata, 2)); + Assert.assertEquals(Schema.of(Schema.LogicalType.DATETIME), schemaReader.getSchema(metadata, 3)); + Assert.assertEquals(Schema.of(Schema.LogicalType.DATETIME), schemaReader.getSchema(metadata, 4)); + Assert.assertEquals(Schema.of(Schema.LogicalType.DATE), schemaReader.getSchema(metadata, 5)); + Assert.assertEquals(Schema.of(Schema.Type.STRING), schemaReader.getSchema(metadata, 6)); + Assert.assertEquals(Schema.of(Schema.Type.STRING), schemaReader.getSchema(metadata, 7)); + Assert.assertEquals(Schema.of(Schema.Type.STRING), schemaReader.getSchema(metadata, 8)); + Assert.assertEquals(Schema.of(Schema.Type.STRING), schemaReader.getSchema(metadata, 9)); + } + + @Test + public void testShouldIgnoreColumn() throws SQLException { + DatabricksSchemaReader schemaReader = new DatabricksSchemaReader("sessionID"); + Map names = new java.util.HashMap<>(); + names.put(1, "c_sessionID"); + names.put(2, "sqn_sessionID"); + names.put(3, "columnName"); + + ResultSetMetaData metadata = createMockMetadata(java.util.Collections.emptyMap(), names); + + Assert.assertTrue(schemaReader.shouldIgnoreColumn(metadata, 1)); + Assert.assertTrue(schemaReader.shouldIgnoreColumn(metadata, 2)); + Assert.assertFalse(schemaReader.shouldIgnoreColumn(metadata, 3)); + } +} diff --git a/databricks-plugin/widgets/Databricks-batchsource.json b/databricks-plugin/widgets/Databricks-batchsource.json new file mode 100644 index 000000000..d6e0de95d --- /dev/null +++ b/databricks-plugin/widgets/Databricks-batchsource.json @@ -0,0 +1,279 @@ +{ + "metadata": { + "spec-version": "1.5" + }, + "display-name": "Databricks", + "configuration-groups": [ + { + "label": "Connection", + "properties": [ + { + "widget-type": "toggle", + "label": "Use connection", + "name": "useConnection", + "widget-attributes": { + "on": { + "value": "true", + "label": "YES" + }, + "off": { + "value": "false", + "label": "NO" + }, + "default": "false" + } + }, + { + "widget-type": "connection-select", + "label": "Connection", + "name": "connection", + "widget-attributes": { + "connectionType": "Databricks" + } + }, + { + "widget-type": "plugin-list", + "label": "JDBC Driver name", + "name": "jdbcPluginName", + "widget-attributes": { + "plugin-type": "jdbc" + } + }, + { + "widget-type": "textbox", + "label": "Host", + "name": "host", + "widget-attributes": { + "placeholder": "Databricks server hostname." + } + }, + { + "widget-type": "number", + "label": "Port", + "name": "port", + "widget-attributes": { + "default": "443" + } + }, + { + "widget-type": "textbox", + "label": "HTTP Path", + "name": "httpPath", + "widget-attributes": { + "placeholder": "e.g., sql/1.0/warehouses/xxxx" + } + }, + { + "widget-type": "textbox", + "label": "Username", + "name": "user" + }, + { + "widget-type": "password", + "label": "Password / Token", + "name": "password" + }, + { + "widget-type": "keyvalue", + "label": "Connection Arguments", + "name": "connectionArguments", + "widget-attributes": { + "showDelimiter": "false", + "key-placeholder": "Key", + "value-placeholder": "Value", + "kv-delimiter": "=", + "delimiter": ";" + } + } + ] + }, + { + "label": "Basic", + "properties": [ + { + "widget-type": "textbox", + "label": "Reference Name", + "name": "referenceName", + "widget-attributes": { + "placeholder": "Name used to identify this source for lineage. Typically, the name of the table/view." + } + }, + { + "widget-type": "textbox", + "label": "Database / Catalog", + "name": "database" + }, + { + "widget-type": "connection-browser", + "widget-category": "plugin", + "widget-attributes": { + "connectionType": "Databricks", + "label": "Browse Database" + } + } + ] + }, + { + "label": "SQL Query", + "properties": [ + { + "widget-type": "textarea", + "label": "Import Query", + "name": "importQuery", + "widget-attributes": { + "rows": "4" + } + }, + { + "widget-type": "get-schema", + "widget-category": "plugin" + } + ] + }, + { + "label": "Advanced", + "properties": [ + { + "widget-type": "textarea", + "label": "Bounding Query", + "name": "boundingQuery", + "widget-attributes": { + "rows": "4" + } + }, + { + "widget-type": "textbox", + "label": "Split-By Field Name", + "name": "splitBy" + }, + { + "widget-type": "textbox", + "label": "Number of Splits", + "name": "numSplits", + "widget-attributes": { + "default": "1" + } + }, + { + "widget-type": "number", + "label": "Fetch Size", + "name": "fetchSize", + "widget-attributes": { + "default": "1000", + "minimum": "0" + } + } + ] + }, + { + "properties": [ + { + "widget-type": "hidden", + "label": "Initial Retry Duration (sec)", + "name": "initialRetryDuration", + "widget-attributes": { + "default": 5, + "minimum": 0 + } + }, + { + "widget-type": "hidden", + "label": "Maximum Retry Duration (sec)", + "name": "maxRetryDuration", + "widget-attributes": { + "default": 80, + "minimum": 0 + } + }, + { + "widget-type": "hidden", + "label": "Maximum Retry Count", + "name": "maxRetryCount", + "widget-attributes": { + "default": 5, + "minimum": 0 + } + } + ] + } + ], + "outputs": [ + { + "name": "schema", + "widget-type": "schema", + "widget-attributes": { + "schema-types": [ + "boolean", + "int", + "long", + "float", + "double", + "bytes", + "string" + ], + "schema-default-type": "string" + } + } + ], + "filters": [ + { + "name": "showConnectionProperties", + "condition": { + "expression": "useConnection == false" + }, + "show": [ + { + "type": "property", + "name": "jdbcPluginName" + }, + { + "type": "property", + "name": "host" + }, + { + "type": "property", + "name": "port" + }, + { + "type": "property", + "name": "httpPath" + }, + { + "type": "property", + "name": "user" + }, + { + "type": "property", + "name": "password" + }, + { + "type": "property", + "name": "database" + }, + { + "type": "property", + "name": "connectionArguments" + } + ] + }, + { + "name": "showConnectionId", + "condition": { + "expression": "useConnection == true" + }, + "show": [ + { + "type": "property", + "name": "connection" + } + ] + } + ], + "jump-config": { + "datasets": [ + { + "ref-property-name": "referenceName" + } + ] + } +} diff --git a/databricks-plugin/widgets/Databricks-connector.json b/databricks-plugin/widgets/Databricks-connector.json new file mode 100644 index 000000000..15325289c --- /dev/null +++ b/databricks-plugin/widgets/Databricks-connector.json @@ -0,0 +1,114 @@ +{ + "metadata": { + "spec-version": "1.0" + }, + "display-name": "Databricks", + "configuration-groups": [ + { + "label": "Basic", + "properties": [ + { + "widget-type": "plugin-list", + "label": "JDBC Driver name", + "name": "jdbcPluginName", + "widget-attributes": { + "plugin-type": "jdbc" + } + }, + { + "widget-type": "textbox", + "label": "Host", + "name": "host", + "widget-attributes": { + "placeholder": "e.g., dbc-xxxx.cloud.databricks.com" + } + }, + { + "widget-type": "number", + "label": "Port", + "name": "port", + "widget-attributes": { + "default": "443" + } + }, + { + "widget-type": "textbox", + "label": "HTTP Path", + "name": "httpPath", + "widget-attributes": { + "placeholder": "e.g., sql/1.0/warehouses/xxxx" + } + }, + { + "widget-type": "textbox", + "label": "Database / Catalog", + "name": "database" + } + ] + }, + { + "label": "Credentials", + "properties": [ + { + "widget-type": "textbox", + "label": "Username", + "name": "user" + }, + { + "widget-type": "password", + "label": "Password / Token", + "name": "password" + } + ] + }, + { + "label": "Advanced", + "properties": [ + { + "widget-type": "keyvalue", + "label": "Connection Arguments", + "name": "connectionArguments", + "widget-attributes": { + "showDelimiter": "false", + "key-placeholder": "Key", + "value-placeholder": "Value", + "kv-delimiter": "=", + "delimiter": ";" + } + } + ] + }, + { + "properties": [ + { + "widget-type": "hidden", + "label": "Initial Retry Duration (sec)", + "name": "initialRetryDuration", + "widget-attributes": { + "default": 5, + "minimum": 0 + } + }, + { + "widget-type": "hidden", + "label": "Maximum Retry Duration (sec)", + "name": "maxRetryDuration", + "widget-attributes": { + "default": 80, + "minimum": 0 + } + }, + { + "widget-type": "hidden", + "label": "Maximum Retry Count", + "name": "maxRetryCount", + "widget-attributes": { + "default": 5, + "minimum": 0 + } + } + ] + } + ], + "outputs": [] +} diff --git a/pom.xml b/pom.xml index 54e6ef09e..c739d8855 100644 --- a/pom.xml +++ b/pom.xml @@ -45,6 +45,7 @@ teradata-plugin generic-db-argument-setter amazon-redshift-plugin + databricks-plugin