From 9d3722bc2040bfbef170b75688e799619825ffcf Mon Sep 17 00:00:00 2001 From: vicennial Date: Tue, 4 Jul 2023 08:49:48 +0200 Subject: [PATCH 1/2] init --- .../jvm/src/test/resources/TestHelloV2.jar | Bin 0 -> 3784 bytes .../spark/sql/application/ReplE2ESuite.scala | 31 +++++++++++++++++- .../SparkConnectArtifactManager.scala | 2 +- 3 files changed, 31 insertions(+), 2 deletions(-) create mode 100755 connector/connect/client/jvm/src/test/resources/TestHelloV2.jar diff --git a/connector/connect/client/jvm/src/test/resources/TestHelloV2.jar b/connector/connect/client/jvm/src/test/resources/TestHelloV2.jar new file mode 100755 index 0000000000000000000000000000000000000000..d89cf6543a20090abd953af39ac97608aa02050e GIT binary patch literal 3784 zcmaJ^1yB^+{#{}T=|(_BWTh6Q5fD&;1tbJ^iKUSR0qJI?8zf|rkZvRei4|#2;p@B5!Scjn%?=l>4g8@sXZjwIL+)9)NY zpTEksbMCxo>xmwqGFyhGD?jPUw6|-Ts5UBK3@NocVv)6PoEt0KcrPS3^j0b)yF%smff7BB=h`z-=V_Gd3ACQ0(7gNkOv6HL zK(|2OfW51tljHtezW?=J!6Z|qdhYl+`n>wRIaswOXD*Y-4fH%CD&aS0w;WeRBtl!;^0z4(&e8Q>idz|6S0gU6`+07Odh!YjL>&0{ zoLi(FSPZ`S#xobiONS@_89k%96&`Tm!%fs-Daa=0uOPkR*f-;&sPMzMlm`GcjEC9(A@_ryiVSo$}GN9m24XskD}o< zkZJIv)FArmuNB7Xj#1=pPy(u=6w(hu?yiA(REmgIP;vFSD(Xz7LL@S#sds)}pSSIe zpHNJsZ!}Hsct+x`rz@>>C>5LaSjgSA#Et4Y+3O4Hw=J~huT2dH8?!72QuyU*g^`HO z(Cd2FglMHYC)$Z5ooUJZMUEMR8jG{u3;bxs9KY9G!B;#WYwS|i0>p(* z{p*kwr0f6hkzZCWNDq#3K75pWFD4j_t6P)Stixrb4J{=WZ z3nE#WEn6^CAhRR)kuWFw_l}@m_DWvi2?EQq>wR}6wN8zDC+ltUtY6&xpfCA(SZ&1U zru#*!&th-;v#O|t&1W;wC9>fRQ!s~3Ic4*E41WL+B7 zl@(Y}vv!+9$$6ODu3j(0lYT4gu{NoL%NTE0Mc2`sqeOb$aRrCb^uF>Wi;}G-!#zTF zC#KmrmBvmG;+!ko?!h!1L`dIC=D1i=5BW%-O4Staa3}vYYbn*lt}+>o5-}IwFcIa6 zCN)MaJy$6tMk(%lTgA!Ht;EW*lx$X~IquoiP5uQlMDL1r-F#46cy`;@*QzLzaL9{J znjV{IBD@TCaRxRjy0}S@eu6p_bu$;D7H)Al4TTdSLzdDKCvc;}M=dH(Z`C5rUUz(XPO@5m?kEi3M24(P7j z#gvy7jb@*yZe+;Vt&@C+FE|VW`Xj`X=lZgPcB7~2J5`rkAC08awBL?(#^S|k*w1x| z$6{SCNdn|J=-0-*e;|izJxUUEn6qE0N4~l*UqWUMI#S#Nm(oGFV|E)sVlrRn1k^ke zGeh{*3JVnPQG{|XfTR56o>snv$-1na`G*b9|4>&5>164rE+yI$U{J#DEC-Y~d>&K? zQ!ZJpGn`(R!GN3sb7j-|$WA{k>eCXyzO2xG#n!+h`aGy#w0oZ-o8C${q&P{O$9Qgj zc~g1YmSZ)T>eSSCqDtiXA~429mT^Wk@@a1pNKjuF#%mdB7S(Q8#oq+Alg#R;ez^+S zEEoN*VEpQHh%AcoaQNMu4v@^Dt6>rA8@*~4dX6hu^O=<-N8pZ%E=$C z${Z56--`Ytf0g~g02wgEJ1K@SbUCh_eAW9w&-Qn;{i}3DjyidVx#=MwL11tCl7sH@R#5Zn=4*@!0l5P;R zSYgKRi$XB9gP(0gy}}pVwp;LX8X(O@X4X3kXt^Ixxw^`CjFL6IlyFp8De6aVoi>0%pDQ9NUgxcoFpLYmCHHw~)8M;2T zz%T@65wbx{rjC|ftZl1{GZETii44qZ4K=fRr%$(^)trNK% z=46|u(eYrA(-Ku$%`I&)`Q#6Z(alj{e*R{9+P#69c8VgpOoTZFTlHpEBXic>m&jLKvb%gB~jX<+HQ1Y@U z)?#V!L>9@1*fN|xX?{HV%JACIZ~Jyo56>Q+jXR}PAhCs_00*{_hiau<6cQtNd9|IJ zmZbN{@-}-s>@n39D-Y!&`uCisbtm1b)hgeW6xz&=ue8>@uYo;?w-$+-Ef`7il4AH! zeXZ#8w{FX(pXFT^92{H%nkQ~J477~!Yf`e5`V#aHOKvv; z4-Ro}bsPhuJ(gT<2R5JFv=h85CJ1E#BA5;v1J?}sAiU}@v9KaW!Wuc+FZT^TQ3FVv z>I0j_P6EpLEGXq15$tEf#;p5MWf%BAEWG0aptG#%Mf}t{zo<&Q&-fa<> z!2_v*fYHO77C(%*XLI=CZ~ON#mdRxZAXYpti6>kKTiq#7`AOyz1^ay&_jq;rlb(}W z*l{}68oQbKbR@5@q>1Ws`NnHb%o_xO@V@@I1C)I+{<8B>fdvDAw83`)2G9Au2n)q# z%%oQ;v$jmDXWCaJb$uzAeSM49JPX$Z&WX%CroB!4r+_?=eVD;!yZQA?xIyHU;kP@n zp@bn;JC~rdo2x!vbe!6z$rFK zQ9=Y;17FSU88Wk_-;Q#exvMvFlYcj25p0;ra6UMaO$BY&FJYS;u#QtH5ob6~v4ld) zYdR(()1G+V_I~pYSTEz2$JkpB8B}{MVu+26-7(#Zd&Zc|&@@;R?yD8Dq>q(5DMnt2 z%_lT&L?lVe{koBNNz_wmyZ}ByaD(r6E(uO(F4#I6GTu@cQXgQBodyn(nOlYg5=JGY z6lh1p+tuVvNvLP&qP-QZF$M2~ z%bPl)$PgQ&WJ7VvTi1~rYrIclg46FH4Oc<=9F<11$lRZIi^OD)Zd6jw|1`d`T*1zX z&-jP|00+t|%Y}nW3;1oi{tP%*vFktaPa*88$~C}0hoRq9%KvTT{=XYn;pjJpel4%U z)bFVDpB?{cAg{vGZ=C&F{(1Ay==9(3{_{_+%;s<4{`!Rfsq&ZgR96NPToHl4h7=9} J;7WG2`xl*1oPhuU literal 0 HcmV?d00001 diff --git a/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/application/ReplE2ESuite.scala b/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/application/ReplE2ESuite.scala index 61959234c879..b8b55082a64e 100644 --- a/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/application/ReplE2ESuite.scala +++ b/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/application/ReplE2ESuite.scala @@ -17,12 +17,13 @@ package org.apache.spark.sql.application import java.io.{PipedInputStream, PipedOutputStream} +import java.nio.file.Paths import java.util.concurrent.{Executors, Semaphore, TimeUnit} import org.apache.commons.io.output.ByteArrayOutputStream import org.scalatest.BeforeAndAfterEach -import org.apache.spark.sql.connect.client.util.RemoteSparkSession +import org.apache.spark.sql.connect.client.util.{IntegrationTestUtils, RemoteSparkSession} class ReplE2ESuite extends RemoteSparkSession with BeforeAndAfterEach { @@ -151,4 +152,32 @@ class ReplE2ESuite extends RemoteSparkSession with BeforeAndAfterEach { assertContains("Array[java.lang.Long] = Array(0L, 2L, 4L, 6L, 8L)", output) } + test("Client-side JAR") { + // scalastyle:off classforname line.size.limit + val sparkHome = IntegrationTestUtils.sparkHome + val testJar = Paths + .get(s"$sparkHome/connector/connect/client/jvm/src/test/resources/TestHelloV2.jar") + .toFile + + assert(testJar.exists(), "Missing TestHelloV2 jar!") + val input = s""" + |import java.nio.file.Paths + |def classLoadingTest(x: Int): Int = { + | val classloader = + | Option(Thread.currentThread().getContextClassLoader).getOrElse(getClass.getClassLoader) + | val cls = Class.forName("com.example.Hello$$", true, classloader) + | val module = cls.getField("MODULE$$").get(null) + | cls.getMethod("test").invoke(module).asInstanceOf[Int] + |} + |val classLoaderUdf = udf(classLoadingTest _) + | + |val jarPath = Paths.get("$sparkHome/connector/connect/client/jvm/src/test/resources/TestHelloV2.jar").toUri + |spark.addArtifact(jarPath) + | + |spark.range(5).select(classLoaderUdf(col("id"))).as[Int].collect() + """.stripMargin + val output = runCommandsInShell(input) + assertContains("Array[Int] = Array(2, 2, 2, 2, 2)", output) + // scalastyle:on classforname line.size.limit + } } diff --git a/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/artifact/SparkConnectArtifactManager.scala b/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/artifact/SparkConnectArtifactManager.scala index 0a91c6b95502..9fd8e367e4aa 100644 --- a/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/artifact/SparkConnectArtifactManager.scala +++ b/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/artifact/SparkConnectArtifactManager.scala @@ -133,7 +133,7 @@ class SparkConnectArtifactManager(sessionHolder: SessionHolder) extends Logging Files.move(serverLocalStagingPath, target) if (remoteRelativePath.startsWith(s"jars${File.separator}")) { jarsList.add(target) - jarsURI.add(artifactURI + "/" + target.toString) + jarsURI.add(artifactURI + "/" + remoteRelativePath.toString) } else if (remoteRelativePath.startsWith(s"pyfiles${File.separator}")) { sessionHolder.session.sparkContext.addFile(target.toString) val stringRemotePath = remoteRelativePath.toString From b329123dbdc5b444dacf68ca2ae50a7b6c53b701 Mon Sep 17 00:00:00 2001 From: vicennial Date: Tue, 4 Jul 2023 17:40:46 +0200 Subject: [PATCH 2/2] use scala version to get right jar --- .../{TestHelloV2.jar => TestHelloV2_2.12.jar} | Bin .../src/test/resources/TestHelloV2_2.13.jar | Bin 0 -> 4118 bytes .../spark/sql/application/ReplE2ESuite.scala | 17 ++++++++++++++--- 3 files changed, 14 insertions(+), 3 deletions(-) rename connector/connect/client/jvm/src/test/resources/{TestHelloV2.jar => TestHelloV2_2.12.jar} (100%) mode change 100755 => 100644 create mode 100644 connector/connect/client/jvm/src/test/resources/TestHelloV2_2.13.jar diff --git a/connector/connect/client/jvm/src/test/resources/TestHelloV2.jar b/connector/connect/client/jvm/src/test/resources/TestHelloV2_2.12.jar old mode 100755 new mode 100644 similarity index 100% rename from connector/connect/client/jvm/src/test/resources/TestHelloV2.jar rename to connector/connect/client/jvm/src/test/resources/TestHelloV2_2.12.jar diff --git a/connector/connect/client/jvm/src/test/resources/TestHelloV2_2.13.jar b/connector/connect/client/jvm/src/test/resources/TestHelloV2_2.13.jar new file mode 100644 index 0000000000000000000000000000000000000000..6dee8fcd9c95702fb7909c909c6faf22fe93766d GIT binary patch literal 4118 zcmZ{n2Q*w;7sp2#MvYFikeLa>Ac$xqA$s&Kf*^>_h{Pa?&Ql_K8AOZTTlC($L?_Bb z@4ZC(czN$7dFy@q-n-Vl>;C`y>~q#SXYHTzBP?tR00;yE;QTSC3HT-O0N4NpNp&%p ztfCaxc_#pX1yFv33%XKZ{v}iXzh;U{#Xn{RF-2J^Ni}s&1*u(y?k@R9FwOy@N3h$y z-F?NX+=IL`^9W8k1rAw-0mm$CJng0H&2%U_w&wzx5eux?T2xV~3skGDP7Bag^Bu%C z0^`pzG(1^6BrgZKdalaA2LLWhJN5p34%nA}jcu*Dekx3Sepe9wr7&?aw6?P{`BV90 zJs$i^N%JqIjER+%Eriq9%Fw~VTU}3yT#N$#MKzK$H~c0OCZ=j3@EL~6pynxnP~kBX zm2B)0PdF*nr>dTp*)7v^SkH~fQAZZ&V4kU%ywoI#iM6%Ke1}cA!>c!E=jT}U`25$= z&~z(5AucTrLo2*41Q{*O!!@tBGd?ZrVB^IZiEt4jF`rdAOCc$-@>`#}l}_u)XvGO@ zZof==jHWW{-w~#>MWFEf+Z}Jo@a{d6i!?_~(LUsk88=(%WxeA8a(2Wf2};DBP>G+lz`OL5BUc2zq4$M6U$O-a6^jYWRHXNb(G;`kIp{9p-sv^$A)_vabc{R3f6w&s zB}*nU{E8mRGH3@gWIN@g5U9lZI4-{5*`Jr!wHoQfZTuR>E9)8Wz&awzVio$@goCTv zc+oDZs+vP&_A|XC0V7vU->&~A8E3RK|Dgrv`o4#n!y|MA8_{c^j>~Xb1*OhO{6q3o zKJ8@Yb<7}cI2d8EJh76dABOgW#zY9XE$1e@Q7&S1r7sp&-_j>cK0;L z`^`)&4|P-SqBn8XdC>TtkdAiPZK#ZG%^xA?C2cHYvnUb|e8Zyod2stzlGX^l^bneK zKeLi+xH1f2$w2u}++t!p9l9Q&3<}+6UI##4GX+S1LO0c7tzklfH%=f9@j!?Ex*) z+^#Pl789lRU69i6czQZza?lPh!xcx2Wg6M0h6>pb(Hx5P{s;PgYngMDnQ)(GC@_t*flC zj-sOMs2pE7KImoyN0K^jSrL@w1zz}p zpUO3deo~dtxf1a<3N}bpIUJnLiE5_;er=MtxXU)qoG$AkqjKOK9-Ry);O{hrqy2U9 zLvkQDG$@d&RIVz`*G46sTM^=Q@!zuFmSGb|lyuY@YjF?!>nmcfo+Z-PW4U#7#5 z*Nm^r{^6I%)f)~UR7%ooF_pI;Z21;NKXQw1+7a1c&Nfw%EZU%d9n?L5+ZcrM1 zz!-z4h&YoSh-%afd#Bt5yPC(Hj;+y$tG+ZK|IS#NWJNC2pkvT&IAjp9J zP!d{Akpan$I&L;dyLCrC3KoIM0z)01^21A3TnM8GzZ5 z>oJ*)CAR5rF+l^x7*wSnJ95vM=r$EX2AFSWuJbC{3Tp4!&H1Oww*i!;+xaX(cLNTzO#DQ~m2Yk0i> ztG&&vro2LJ->}ctJULdZ45FWCzAlqkjWBdsdIKUk5&m_A=e?lO+micR+`%;)wN?+U z6{IXp>^2H==p>(1o!8ZD3`=!U?$OkuYP4g+f+A0jf&p-%X~>Lq{k4fi%9!46xORu4 zPod8}RS=6_KhFBiquDHiJh!=eeOwF!@-poz%lyfys16B~Q+Xmmp1s#9Hi!0nS9>m3 zti3xSv(1LoOw)?A2~R#{i4js0Hq!>E{5~aHXgzoQ3TxiataH5ZG5K9+>pK{u&y!zUCq7C)9v{cIut53Z*VjlElU^;@nbqv5y<)Csv_a z)vL*f+DuV=@>Brdjo({7z|VcRmx zc&jv-kZ99?B4%X*(@);M%>)$xX7&u!Mt8%*g_W58*+s?)Fn=x0z+~E z5ojJ_t1ZBQL+f%;2wP++d_DTX1hOEOLDB`JMjY*BKDjA^R@H&6PC)5gy>c9z+RZuJ z#XB*ApZd6^w;sfwUbIz=?7BbLYK(MMcfphQ0}__4K^aX&IsPaBb!EIdUQ?ckx!>0b z%qDS6Y@)?DDSI3zl9hqXvjWF2;iQ@HR@4L)j@j{-ek=8yAWm?&SGp2NBr1j3H}rY8 zq%dV*bGvBKqslJ0t@j-PLd0Kvw51rfHG0RGOl@0m>B+`xksiI8c3Z?apHVrJK|Cwd z^RWyYu?LgR)0;NkaX4k10$yNFV#WlGZ=Fv&bDD5iH11@($562l7XU(we2}eAHJ#}S z1~!4!a}|g9A#XloO0tu__G1y9nv5Pz ze8KhdvicyE^4HucjKi((pecyzT=?KIQ-OAnzeY{!w;WXtS`pKro4!0?s9HDJD6QNw zQFSx#llCmOD->gSe0zD`1T%k1UmsGi@7^B3u6#r&O}fW4I(~ueetwU{HRv_whi_jr zN>kpll-+N-(e_=i%n5kN?A6ZBK`SVaIg4#2>9q{1b*J03AI~DcOI6*5Bq5ij=ywGY z(GwBy#@^>;Bl|Ms^W2u{v{aC(-?(U=*q+{=dia70M;|r##$74fJoSEZ-_tGWJ(q!n zyv-fPxjyc_eOxyZePR?x?Xh71xnpVCwi9dFyE=~o#v`{88Ti|RH> z3Qm8SnU~}~wllrHcj#>^1ui{zzFGztm$nR^gmv}$3`d=p)NVet&aCI0P|wI$XVUmG zSWl|OPE|c}jBDirurS0rD#cww^Hh6xTb@f*6V-Z;jojMe>Jdo4@XBsBwHz&?ttJO)6dDutYEJKo1-yQWDxwYTcqs_4b z-ru@#Te|kVt)WivA>D=T>|202r)}EhJ$8()$!q=4)zvpLa^n=o=euy7JL+t@@p@#t z(#5E{zXfu8X#4g44%$**pJR)iT*k}5cvV6R_dE_tv48J*y)M3xy|^ho3fRXXV`_LO zfkoY z{Tnr}X!Q5^2QRPK^xxKB@$ia8e+K;W^&jgmWO|hWAMpP%@@IzT%hw-J`P1=VSozbj z@R#E!Wd5A!6=we0PG4c>&mh70G1lMCU!n8YIbUV@bt$h}@1G%lxpr3-|CZ%PBd+`i Sh;t>xxx8Ln4i-p!_4Pme8xa%$ literal 0 HcmV?d00001 diff --git a/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/application/ReplE2ESuite.scala b/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/application/ReplE2ESuite.scala index b8b55082a64e..720f66680ee1 100644 --- a/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/application/ReplE2ESuite.scala +++ b/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/application/ReplE2ESuite.scala @@ -20,6 +20,8 @@ import java.io.{PipedInputStream, PipedOutputStream} import java.nio.file.Paths import java.util.concurrent.{Executors, Semaphore, TimeUnit} +import scala.util.Properties + import org.apache.commons.io.output.ByteArrayOutputStream import org.scalatest.BeforeAndAfterEach @@ -36,6 +38,11 @@ class ReplE2ESuite extends RemoteSparkSession with BeforeAndAfterEach { private var ammoniteIn: PipedInputStream = _ private val semaphore: Semaphore = new Semaphore(0) + private val scalaVersion = Properties.versionNumberString + .split("\\.") + .take(2) + .mkString(".") + private def getCleanString(out: ByteArrayOutputStream): String = { // Remove ANSI colour codes // Regex taken from https://stackoverflow.com/a/25189932 @@ -97,7 +104,10 @@ class ReplE2ESuite extends RemoteSparkSession with BeforeAndAfterEach { def assertContains(message: String, output: String): Unit = { val isContain = output.contains(message) - assert(isContain, "Ammonite output did not contain '" + message + "':\n" + output) + assert( + isContain, + "Ammonite output did not contain '" + message + "':\n" + output + + s"\nError Output: ${getCleanString(errorStream)}") } test("Simple query") { @@ -156,7 +166,8 @@ class ReplE2ESuite extends RemoteSparkSession with BeforeAndAfterEach { // scalastyle:off classforname line.size.limit val sparkHome = IntegrationTestUtils.sparkHome val testJar = Paths - .get(s"$sparkHome/connector/connect/client/jvm/src/test/resources/TestHelloV2.jar") + .get( + s"$sparkHome/connector/connect/client/jvm/src/test/resources/TestHelloV2_$scalaVersion.jar") .toFile assert(testJar.exists(), "Missing TestHelloV2 jar!") @@ -171,7 +182,7 @@ class ReplE2ESuite extends RemoteSparkSession with BeforeAndAfterEach { |} |val classLoaderUdf = udf(classLoadingTest _) | - |val jarPath = Paths.get("$sparkHome/connector/connect/client/jvm/src/test/resources/TestHelloV2.jar").toUri + |val jarPath = Paths.get("${testJar.toString}").toUri |spark.addArtifact(jarPath) | |spark.range(5).select(classLoaderUdf(col("id"))).as[Int].collect()