// ĠCopyright Ġ 2 0 2 0 ĠAnd y ĠGro ve Ċ // Ċ // Ġ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 Ġorg . bal list ac omp ute . execution Ċ Ċ import Ġjava . io . File Ċ import Ġkotlin . test . assertEquals Ċ import Ġorg . apache . arrow . vector . types . pojo . Arrow Type Ċ import Ġorg . bal list ac omp ute . datasource . InMemory DataSource Ċ import Ġorg . bal list ac omp ute . datat ypes . Arrow Types Ċ import Ġorg . bal list ac omp ute . datat ypes . Field Ċ import Ġorg . bal list ac omp ute . datat ypes . Schema Ċ import Ġorg . bal list ac omp ute . fu z zer . F uz zer Ċ import Ġorg . bal list ac omp ute . logical .* Ċ import Ġorg . junit . Test Ċ import Ġorg . junit . jupiter . api . Test Instance Ċ Ċ @ Test Instance ( Test Instance . Lifecycle . PER _ CLASS ) Ċ class ĠExecution Test Ġ{ ĊĊĠ Ġval Ġdir Ġ= Ġ"../ testdata " ĊĊĠ Ġval Ġemployee Csv Ġ= ĠFile ( dir , Ġ" employee . csv "). absolute Path ĊĊĠ Ġ@ Test ĊĠ Ġfun Ġ` employees Ġin ĠCO Ġusing ĠDataFrame `() Ġ{ ĊĠĠĠ Ġ// ĠCreate Ġa Ġcontext ĊĠĠĠ Ġval Ġctx Ġ= ĠExecutionContext ( map Of ()) ĊĊĠĠĠ Ġ// ĠConstruct Ġa Ġquery Ġusing Ġthe ĠDataFrame ĠAPI ĊĠĠĠ Ġval Ġdf Ġ= ĊĠĠĠĠĠĠĠ Ġctx . csv ( employee Csv ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter ( col (" state ") Ġeq Ġlit (" CO ")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. project ( listOf ( col (" id "), Ġcol (" first _ name "), Ġcol (" last _ name "))) ĊĊĠĠĠ Ġval Ġbatches Ġ= Ġctx . execute ( df ). as Sequence (). toList () ĊĠĠĠ ĠassertEquals ( 1 , Ġbatches . size ) ĊĊĠĠĠ Ġval Ġbatch Ġ= Ġbatches . first () ĊĠĠĠ ĠassertEquals (" 2 , G reg g , Lang ford \ n " Ġ+ Ġ" 3 , John , Tr avis \ n ", Ġbatch . to CSV ()) ĊĠ Ġ} ĊĊĠ Ġ@ Test ĊĠ Ġfun Ġ` employees Ġin ĠCA Ġusing ĠSQL `() Ġ{ ĊĠĠĠ Ġ// ĠCreate Ġa Ġcontext ĊĠĠĠ Ġval Ġctx Ġ= ĠExecutionContext ( map Of ()) ĊĊĠĠĠ Ġval Ġemployee Ġ= Ġctx . csv ( employee Csv ) ĊĠĠĠ Ġctx . register (" employee ", Ġemployee ) ĊĊĠĠĠ Ġ// ĠConstruct Ġa Ġquery Ġusing Ġthe ĠDataFrame ĠAPI ĊĠĠĠ Ġval Ġdf Ġ= Ġctx . sql (" SELECT Ġid , Ġfirst _ name , Ġlast _ name ĠFROM Ġemployee ĠWHERE Ġstate Ġ= Ġ' CA '") ĊĊĠĠĠ Ġval Ġbatches Ġ= Ġctx . execute ( df ). as Sequence (). toList () ĊĠĠĠ ĠassertEquals ( 1 , Ġbatches . size ) ĊĊĠĠĠ Ġval Ġbatch Ġ= Ġbatches . first () ĊĠĠĠ ĠassertEquals (" 1 , Bill , Hop k ins \ n ", Ġbatch . to CSV ()) ĊĠ Ġ} ĊĊĠ Ġ@ Test ĊĠ Ġfun Ġ` aggregate Ġquery `() Ġ{ ĊĠĠĠ Ġ// ĠCreate Ġa Ġcontext ĊĠĠĠ Ġval Ġctx Ġ= ĠExecutionContext ( map Of ()) ĊĊĠĠĠ Ġ// Ġconstruct Ġa Ġquery Ġusing Ġthe ĠDataFrame ĠAPI ĊĠĠĠ Ġval Ġdf Ġ= ĊĠĠĠĠĠĠĠ Ġctx . csv ( employee Csv ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. aggregate ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠlistOf ( col (" state ")), ĠlistOf ( Max ( cast ( col (" salary "), ĠArrow Type . Int ( 3 2 , Ġtrue ))))) ĊĊĠĠĠ Ġval Ġbatches Ġ= Ġctx . execute ( df ). as Sequence (). toList () ĊĠĠĠ ĠassertEquals ( 1 , Ġbatches . size ) ĊĊĠĠĠ Ġval Ġbatch Ġ= Ġbatches . first () ĊĠĠĠ Ġval Ġexpected Ġ= Ġ" CO , 1 1 5 0 0 \ n " Ġ+ Ġ" CA , 1 2 0 0 0 \ n " Ġ+ Ġ", 1 1 5 0 0 \ n " ĊĠĠĠ ĠassertEquals ( expected , Ġbatch . to CSV ()) ĊĠ Ġ} ĊĊĠ Ġ@ Test ĊĠ Ġfun Ġ` bon uses Ġin ĠCA Ġusing ĠSQL Ġand ĠDataFrame `() Ġ{ ĊĠĠĠ Ġ// ĠCreate Ġa Ġcontext ĊĠĠĠ Ġval Ġctx Ġ= ĠExecutionContext ( map Of ()) ĊĊĠĠĠ Ġ// Ġconstruct Ġa Ġquery Ġusing Ġthe ĠDataFrame ĠAPI ĊĠĠĠ Ġval Ġca Employees Ġ= ĊĠĠĠĠĠĠĠ Ġctx . csv ( employee Csv ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter ( col (" state ") Ġeq Ġlit (" CA ")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. project ( listOf ( col (" id "), Ġcol (" first _ name "), Ġcol (" last _ name "), Ġcol (" salary "))) ĊĊĠĠĠ Ġ// Ġregister Ġthe ĠDataFrame Ġas Ġa Ġtable ĊĠĠĠ Ġctx . register (" ca _ employees ", Ġca Employees ) ĊĊĠĠĠ Ġ// ĠConstruct Ġa Ġquery Ġusing Ġthe ĠDataFrame ĠAPI ĊĠĠĠ Ġval Ġdf Ġ= Ġctx . sql (" SELECT Ġid , Ġfirst _ name , Ġlast _ name , Ġsalary ĠFROM Ġca _ employees ") ĊĠĠĠ Ġ// Ġval Ġdf Ġ= Ġctx . sql (" SELECT Ġid , Ġfirst _ name , Ġlast _ name , Ġsalary Ġ* Ġ 0 . 1 ĠAS Ġbonus ĠFROM Ġca _ employees ") ĊĊĠĠĠ Ġval Ġbatches Ġ= Ġctx . execute ( df ). as Sequence (). toList () ĊĠĠĠ ĠassertEquals ( 1 , Ġbatches . size ) ĊĊĠĠĠ Ġval Ġbatch Ġ= Ġbatches . first () ĊĠĠĠ ĠassertEquals (" 1 , Bill , Hop k ins , 1 2 0 0 0 \ n ", Ġbatch . to CSV ()) ĊĠ Ġ} ĊĊĠ Ġ@ Test ĊĠ Ġfun Ġ` min Ġmax Ġsum Ġfloat `() Ġ{ ĊĠĠĠ Ġval Ġschema Ġ= ĠSchema ( listOf ( Field (" a ", ĠArrow Types . StringType ), ĠField (" b ", ĠArrow Types . Float Type ))) ĊĊĠĠĠ Ġ// Ġval Ġbatch Ġ= ĠF uz zer (). create Record Batch ( schema , Ġ 1 0 2 4 ) ĊĠĠĠ Ġval Ġinput Ġ= ĊĠĠĠĠĠĠĠ ĠF uz zer () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. create Record Batch ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġschema , ĠlistOf ( listOf (" a ", Ġ" a ", Ġ" b ", Ġ" b "), ĠlistOf ( 1 . 0 f , Ġ 2 . 0 f , Ġ 4 . 0 f , Ġ 3 . 0 f ))) ĊĊĠĠĠ Ġval ĠdataSource Ġ= ĠIn Memory DataSource ( schema , ĠlistOf ( input )) ĊĊĠĠĠ Ġval Ġctx Ġ= ĠExecutionContext ( map Of ()) ĊĠĠĠ Ġval Ġlogical Plan Ġ= ĊĠĠĠĠĠĠĠ ĠDataFrame Impl ( Scan ("", ĠdataSource , ĠlistOf ())) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. aggregate ( listOf ( col (" a ")), ĠlistOf ( Min ( col (" b ")), ĠMax ( col (" b ")), ĠSum ( col (" b ")) )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. logical Plan () ĊĊĠĠĠ Ġval Ġbatches Ġ= Ġctx . execute ( logical Plan ). as Sequence (). toList () ĊĠĠĠ ĠassertEquals ( 1 , Ġbatches . size ) ĊĊĠĠĠ Ġval Ġbatch Ġ= Ġbatches . first () ĊĠĠĠ ĠassertEquals (" a , 1 . 0 , 2 . 0 , 3 . 0 \ n " Ġ+ Ġ" b , 3 . 0 , 4 . 0 , 7 . 0 \ n ", Ġbatch . to CSV ()) ĊĠ Ġ} ĊĊĠ Ġ@ Test ĊĠ Ġfun Ġ` float Ġmath `() Ġ{ ĊĠĠĠ Ġval Ġschema Ġ= ĠSchema ( listOf ( Field (" a ", ĠArrow Types . Float Type ), ĠField (" b ", ĠArrow Types . Float Type ))) ĊĊĠĠĠ Ġ// Ġval Ġbatch Ġ= ĠF uz zer (). create Record Batch ( schema , Ġ 1 0 2 4 ) ĊĠĠĠ Ġval Ġinput Ġ= ĊĠĠĠĠĠĠĠ ĠF uz zer () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. create Record Batch ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġschema , ĠlistOf ( listOf ( 1 . 0 f , Ġ 2 . 0 f , Ġ 4 . 0 f , Ġ 3 . 0 f ), ĠlistOf ( 1 1 . 0 f , Ġ 2 2 . 0 f , Ġ 4 4 . 0 f , Ġ 3 3 . 0 f ))) ĊĊĠĠĠ Ġval ĠdataSource Ġ= ĠIn Memory DataSource ( schema , ĠlistOf ( input )) ĊĊĠĠĠ Ġval Ġctx Ġ= ĠExecutionContext ( map Of ()) ĊĠĠĠ Ġval Ġlogical Plan Ġ= ĊĠĠĠĠĠĠĠ ĠDataFrame Impl ( Scan ("", ĠdataSource , ĠlistOf ())) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. project ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠAdd ( col (" a "), Ġcol (" b ")), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠSubtract ( col (" a "), Ġcol (" b ")), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMultiply ( col (" a "), Ġcol (" b ")), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠDiv ide ( col (" a "), Ġcol (" b ")) )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. logical Plan () ĊĊĠĠĠ Ġval Ġbatches Ġ= Ġctx . execute ( logical Plan ). as Sequence (). toList () ĊĠĠĠ ĠassertEquals ( 1 , Ġbatches . size ) ĊĊĠĠĠ Ġval Ġbatch Ġ= Ġbatches . first () ĊĠĠĠ ĠassertEquals ( ĊĠĠĠĠĠĠĠ Ġ" 1 2 . 0 ,- 1 0 . 0 , 1 1 . 0 , 0 . 0 9 0 9 0 9 0 9 \ n " Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" 2 4 . 0 ,- 2 0 . 0 , 4 4 . 0 , 0 . 0 9 0 9 0 9 0 9 \ n " Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" 4 8 . 0 ,- 4 0 . 0 , 1 7 6 . 0 , 0 . 0 9 0 9 0 9 0 9 \ n " Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" 3 6 . 0 ,- 3 0 . 0 , 9 9 . 0 , 0 . 0 9 0 9 0 9 0 9 \ n ", ĊĠĠĠĠĠĠĠ Ġbatch . to CSV ()) ĊĠ Ġ} ĊĊĠ Ġ@ Test ĊĠ Ġfun Ġ` boolean Ġexpressions `() Ġ{ ĊĠĠĠ Ġval Ġschema Ġ= ĊĠĠĠĠĠĠĠ ĠSchema ( listOf ( Field (" a ", ĠArrow Types . Boolean Type ), ĠField (" b ", ĠArrow Types . Boolean Type ))) ĊĊĠĠĠ Ġval Ġinput Ġ= ĊĠĠĠĠĠĠĠ ĠF uz zer () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. create Record Batch ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġschema , ĠlistOf ( listOf ( false , Ġfalse , Ġtrue , Ġtrue ), ĠlistOf ( false , Ġtrue , Ġfalse , Ġtrue ))) ĊĊĠĠĠ Ġval ĠdataSource Ġ= ĠIn Memory DataSource ( schema , ĠlistOf ( input )) ĊĊĠĠĠ Ġval Ġctx Ġ= ĠExecutionContext ( map Of ()) ĊĠĠĠ Ġval Ġlogical Plan Ġ= ĊĠĠĠĠĠĠĠ ĠDataFrame Impl ( Scan ("", ĠdataSource , ĠlistOf ())) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. project ( listOf ( And ( col (" a "), Ġcol (" b ")), ĠOr ( col (" a "), Ġcol (" b ")) )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. logical Plan () ĊĊĠĠĠ Ġval Ġbatches Ġ= Ġctx . execute ( logical Plan ). as Sequence (). toList () ĊĠĠĠ ĠassertEquals ( 1 , Ġbatches . size ) ĊĊĠĠĠ Ġval Ġbatch Ġ= Ġbatches . first () ĊĠĠĠ ĠassertEquals (" false , false \ n " Ġ+ Ġ" false , true \ n " Ġ+ Ġ" false , true \ n " Ġ+ Ġ" true , true \ n ", Ġbatch . to CSV ()) ĊĠ Ġ} Ċ } Ċ