From d4d196d11856f6bc9f16898a27824dece5e95c3a Mon Sep 17 00:00:00 2001 From: Max Burke Date: Thu, 20 Oct 2022 12:55:52 -0700 Subject: [PATCH 1/3] implement ScalarValue for FixedSizeBinary --- datafusion/common/src/scalar.rs | 79 +++++++++++++++++++++++++++++++- datafusion/proto/src/to_proto.rs | 1 + 2 files changed, 79 insertions(+), 1 deletion(-) diff --git a/datafusion/common/src/scalar.rs b/datafusion/common/src/scalar.rs index c3f91dd9b1d1..7e871fa94f0e 100644 --- a/datafusion/common/src/scalar.rs +++ b/datafusion/common/src/scalar.rs @@ -78,6 +78,8 @@ pub enum ScalarValue { LargeUtf8(Option), /// binary Binary(Option>), + /// fixed size binary + FixedSizeBinary(i32, Option>), /// large binary LargeBinary(Option>), /// list of nested ScalarValue @@ -160,6 +162,8 @@ impl PartialEq for ScalarValue { (LargeUtf8(_), _) => false, (Binary(v1), Binary(v2)) => v1.eq(v2), (Binary(_), _) => false, + (FixedSizeBinary(_, v1), FixedSizeBinary(_, v2)) => v1.eq(v2), + (FixedSizeBinary(_, _), _) => false, (LargeBinary(v1), LargeBinary(v2)) => v1.eq(v2), (LargeBinary(_), _) => false, (List(v1, t1), List(v2, t2)) => v1.eq(v2) && t1.eq(t2), @@ -248,6 +252,8 @@ impl PartialOrd for ScalarValue { (LargeUtf8(_), _) => None, (Binary(v1), Binary(v2)) => v1.partial_cmp(v2), (Binary(_), _) => None, + (FixedSizeBinary(_, v1), FixedSizeBinary(_, v2)) => v1.partial_cmp(v2), + (FixedSizeBinary(_, _), _) => None, (LargeBinary(v1), LargeBinary(v2)) => v1.partial_cmp(v2), (LargeBinary(_), _) => None, (List(v1, t1), List(v2, t2)) => { @@ -537,6 +543,7 @@ impl std::hash::Hash for ScalarValue { Utf8(v) => v.hash(state), LargeUtf8(v) => v.hash(state), Binary(v) => v.hash(state), + FixedSizeBinary(_, v) => v.hash(state), LargeBinary(v) => v.hash(state), List(v, t) => { v.hash(state); @@ -901,6 +908,7 @@ impl ScalarValue { ScalarValue::Utf8(_) => DataType::Utf8, ScalarValue::LargeUtf8(_) => DataType::LargeUtf8, ScalarValue::Binary(_) => DataType::Binary, + ScalarValue::FixedSizeBinary(sz, _) => DataType::FixedSizeBinary(*sz), ScalarValue::LargeBinary(_) => DataType::LargeBinary, ScalarValue::List(_, field) => DataType::List(Box::new(Field::new( "item", @@ -988,6 +996,7 @@ impl ScalarValue { ScalarValue::Utf8(v) => v.is_none(), ScalarValue::LargeUtf8(v) => v.is_none(), ScalarValue::Binary(v) => v.is_none(), + ScalarValue::FixedSizeBinary(_, v) => v.is_none(), ScalarValue::LargeBinary(v) => v.is_none(), ScalarValue::List(v, _) => v.is_none(), ScalarValue::Date32(v) => v.is_none(), @@ -1357,13 +1366,30 @@ impl ScalarValue { _ => unreachable!("Invalid dictionary keys type: {:?}", key_type), } } + DataType::FixedSizeBinary(_) => { + let array = scalars + .map(|sv| { + if let ScalarValue::FixedSizeBinary(_, v) = sv { + Ok(v) + } else { + Err(DataFusionError::Internal(format!( + "Inconsistent types in ScalarValue::iter_to_array. \ + Expected {:?}, got {:?}", + data_type, sv + ))) + } + }) + .collect::>>()?; + let array = + FixedSizeBinaryArray::try_from_sparse_iter(array.into_iter())?; + Arc::new(array) + } // explicitly enumerate unsupported types so newly added // types must be aknowledged DataType::Float16 | DataType::Time32(_) | DataType::Time64(_) | DataType::Duration(_) - | DataType::FixedSizeBinary(_) | DataType::FixedSizeList(_, _) | DataType::Interval(_) | DataType::LargeList(_) @@ -1566,6 +1592,20 @@ impl ScalarValue { Arc::new(repeat(None::<&str>).take(size).collect::()) } }, + ScalarValue::FixedSizeBinary(_, e) => match e { + Some(value) => Arc::new( + FixedSizeBinaryArray::try_from_sparse_iter( + repeat(Some(value.as_slice())).take(size), + ) + .unwrap(), + ), + None => Arc::new( + FixedSizeBinaryArray::try_from_sparse_iter( + repeat(None::<&[u8]>).take(size), + ) + .unwrap(), + ), + }, ScalarValue::LargeBinary(e) => match e { Some(value) => Arc::new( repeat(Some(value.as_slice())) @@ -1851,6 +1891,23 @@ impl ScalarValue { }; ScalarValue::new_list(value, nested_type.data_type().clone()) } + DataType::FixedSizeBinary(_) => { + let array = array + .as_any() + .downcast_ref::() + .unwrap(); + let size = match array.data_type() { + DataType::FixedSizeBinary(size) => *size, + _ => unreachable!(), + }; + ScalarValue::FixedSizeBinary( + size, + match array.is_null(index) { + true => None, + false => Some(array.value(index).into()), + }, + ) + } other => { return Err(DataFusionError::NotImplemented(format!( "Can't create a scalar from array of type \"{:?}\"", @@ -1941,6 +1998,9 @@ impl ScalarValue { ScalarValue::Binary(val) => { eq_array_primitive!(array, index, BinaryArray, val) } + ScalarValue::FixedSizeBinary(_, val) => { + eq_array_primitive!(array, index, FixedSizeBinaryArray, val) + } ScalarValue::LargeBinary(val) => { eq_array_primitive!(array, index, LargeBinaryArray, val) } @@ -2285,6 +2345,17 @@ impl fmt::Display for ScalarValue { )?, None => write!(f, "NULL")?, }, + ScalarValue::FixedSizeBinary(_, e) => match e { + Some(l) => write!( + f, + "{}", + l.iter() + .map(|v| format!("{}", v)) + .collect::>() + .join(",") + )?, + None => write!(f, "NULL")?, + }, ScalarValue::LargeBinary(e) => match e { Some(l) => write!( f, @@ -2365,6 +2436,12 @@ impl fmt::Debug for ScalarValue { ScalarValue::LargeUtf8(Some(_)) => write!(f, "LargeUtf8(\"{}\")", self), ScalarValue::Binary(None) => write!(f, "Binary({})", self), ScalarValue::Binary(Some(_)) => write!(f, "Binary(\"{}\")", self), + ScalarValue::FixedSizeBinary(size, None) => { + write!(f, "FixedSizeBinary({}, {})", size, self) + } + ScalarValue::FixedSizeBinary(size, Some(_)) => { + write!(f, "FixedSizeBinary({}, \"{}\")", size, self) + } ScalarValue::LargeBinary(None) => write!(f, "LargeBinary({})", self), ScalarValue::LargeBinary(Some(_)) => write!(f, "LargeBinary(\"{}\")", self), ScalarValue::List(_, _) => write!(f, "List([{}])", self), diff --git a/datafusion/proto/src/to_proto.rs b/datafusion/proto/src/to_proto.rs index 47b779fffc74..ef2f27a5c074 100644 --- a/datafusion/proto/src/to_proto.rs +++ b/datafusion/proto/src/to_proto.rs @@ -1187,6 +1187,7 @@ impl TryFrom<&ScalarValue> for protobuf::ScalarValue { Value::LargeBinaryValue(s.to_owned()) }) } + scalar::ScalarValue::FixedSizeBinary(_, _) => todo!(), datafusion::scalar::ScalarValue::Time64(v) => { create_proto_scalar(v, PrimitiveScalarType::Time64, |v| { From 82638a4aa313fac41ce130d705a791ffd3a555f3 Mon Sep 17 00:00:00 2001 From: Max Burke Date: Fri, 21 Oct 2022 19:50:14 -0700 Subject: [PATCH 2/3] Return an error instead of panicking --- datafusion/proto/src/to_proto.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/datafusion/proto/src/to_proto.rs b/datafusion/proto/src/to_proto.rs index ef2f27a5c074..5e47bbfe9f83 100644 --- a/datafusion/proto/src/to_proto.rs +++ b/datafusion/proto/src/to_proto.rs @@ -1187,7 +1187,11 @@ impl TryFrom<&ScalarValue> for protobuf::ScalarValue { Value::LargeBinaryValue(s.to_owned()) }) } - scalar::ScalarValue::FixedSizeBinary(_, _) => todo!(), + scalar::ScalarValue::FixedSizeBinary(_, _) => { + return Err(Error::General( + "FixedSizeBinary is not yet implemented".to_owned(), + )) + } datafusion::scalar::ScalarValue::Time64(v) => { create_proto_scalar(v, PrimitiveScalarType::Time64, |v| { From dc1d62feae37f2c8b35eb11b76d91d1c08e011c7 Mon Sep 17 00:00:00 2001 From: Max Burke Date: Fri, 21 Oct 2022 20:50:20 -0700 Subject: [PATCH 3/3] add tests for scalar value fixedsizebinary variant --- .../core/tests/parquet/test_binary.parquet | Bin 0 -> 8558 bytes datafusion/core/tests/sql/parquet.rs | 22 ++++++++++++++++++ 2 files changed, 22 insertions(+) create mode 100644 datafusion/core/tests/parquet/test_binary.parquet diff --git a/datafusion/core/tests/parquet/test_binary.parquet b/datafusion/core/tests/parquet/test_binary.parquet new file mode 100644 index 0000000000000000000000000000000000000000..9d906bc490d0624c0223b35f51fe13bfa40477da GIT binary patch literal 8558 zcmcI~MOYj_)8^nBd~k=r;1(b_gS)#1cNiQ(f(CaB?(Xgo+}+(JcyO0L`M%xD?qx6g z9_sCquIlQiZ`Dd*6*eF&u*(HF;vxskM*sq_0Dx@X8!{3=gyYXU-_SBEW=Wp@SKJ0b z0ijp_dsL0FPqe_W_Szwo`2>jsR_ZkDlXT1g+_zYm{7@nGz6WUqX$D~q4L>30J5$8(t#!<${X{WmG(tY- zOB$)^1RBfRifCmp+9dgEw>|_%NtTcB7wBvST~{EbLakEJE0k_WOGzvbJ-{_&n#W;Q zMSz`TP>7-fJ>S=z(BGgbFb0!F`#DJh)MIwSW}~Wn-KW zGu(!%l{L@<7}HPESA-ITlC66-0rf zF>b$)P6mGuv}e_&r#2{{m{^WEs^x3|mV?+@obVib{M;0Sso(B#MSHNgTZpSj&DD53 zt>{hguod=J%pb@TB~@t328e?3xRgO3Z``p<%nVw-vgFF)2@kMgD*0M=#Tslx@N7fj zb^Z2nS4Yi<5{e-#tg?5;+kwE3Shc3nP(Jb;Qt(FN6I*#nP>5mB>gq9nD;EIZ3n6$1 zgK)TCScb;`eABPPcf7^MPK$JM7^dc=&-C%uJ@D(x5Xc!LT!WGwZ#7H>O8L#<)CD>nf)|rtgxr7 zEZBr0+U1|?(XqfI$Y;h7`XUL0f0#*;@Mjl2 zPd}McF+m+lAl_?yJ1(^i(Y3mzhBDN@I5M)(=ZGUH|Bf<5?O(t3un}0l>C~UcpHFi1 zQMJy473Q1f^hjQ)ks14|7qys${u@AB`j1xMDOik*IDbaX z2}fw!(sxEr=3F=J>WJ|ZnF3Kg>G!<)KXl)&>GDB)to_MfVU)}A`RY{((+{R{Fo2Kz z%tTfLzbf1HG@k6lpVuDt!eqaBVA5J%H!aVmD-@V!&2;o34^ef0DCCEa3C16YCo?_{ zE8i^aK6Qs-mJnTM3A4gigfQ0XonY8EW|db{D;6xLncnDM6Z>7WpznPFvMf9#Vx;Yl zrmZi}QM>U?h^%5Zq@gR7leZ(C@M&C7GHj$)AfhhV38bz)3ADV*DK;`LuJWf2l?(C= ze5RSxV>W(a%I6g)Q#%`2V)b8#NHbT~53HQd^ z=TJz@;$8_TFy9@g{j1&J||kp zGt!yx!oSh8Fx;YPJ%is z!>p!*&34Gc9`|Y~fKt(fxKxESmyPsv{xcDklN|x>Bj~ki|Ncf}?(@wSOf34$i22AI zCZ#;TCa?MjlzY2xn!V}WbPcRR!@SNXcgLqmR>!`$XTVS2KJiJ+=c?~^jT?HhEG|e; z{@@#)3;oQR{=$tE(J{kH^ukYo@@qp}lHN!nqwxkike)x)ON2(d+;oZf{o~Km^LGBR z@q=h(VLeNKiH^bFHpl`ZTB^e4+@+T^vKqf+Z65_1kfXN^ zeUm6Dv*zBl&#-A!Mp@lv(n4s7UI)#YS!Z3&*2)0MQG*6DgUBaGzQ)7U$y&~dU^6Jl-4TN&gg z*HzSJV_L6MTFgD>u8HNrJO^j8%@V2@<_ftR{Mhn;yn|xi?<`(Ed-|n#8jV z!b+XpEnk)79SI`r4CauTwXD1LSdXSiPxWJE2qo(VZ!4(_=UDVJ4`Va)PSH$|zI(oA(nj9cXFBXL+iz zICIubSouL?l8`UypF~u%;hLZX#K?AJHeV{W7}(STI1!S%bqs9BKHUT13TxE@8BNRG zKi>!d`q_^^%9*JZehn+$!OqfvxA9GDTPs(q{gt$-#4omTdg5m#TVtnoJwyfCaYKgv zjbx36!6K@Ts2b-otVBRor#kNb)*_~!CE*UaqBqDzl^etL((qDmq17Pc7Sg-tm1^u{ zuv^yPRtR%nzq0}Zjb#*$t+zCy(Mlph7*8UUyKu@y2Rt;6Z7igRF~3lyH~4sFg|})t z%&$r~9>#L}4My4V0)#Zv5*Nc@sFcO<^iXC&;(MC?QqyJvs^Zq+ciP*AtIuMzKzd=hmJlC6^6lHnd#v55ad{1O`VF5Ckb}y)uuZsFeODDG5 zn-EEjXgiw-%4GdvqAG2m@$qaY681#Z-np@|!u51_{20#qO(Awn~P3=8IZ~t5pj)r9TjI z5jejXsq24WTME-!Xnmd>lNbLIrn7l`UP+v>U;eINtwQqU_n__xBH)b4mhNz4L^#KY zsFOcz)%d;+CSwO=O{8G#q{)rp@EE;*%W~hpA!TC{pWt*c#;!odh1?8>$fIuxrfqX_ zo-6x+id#^+6`EqI8_OwBjp67i1|aX|?in*pa*O3m&>j?1a8BFXir_axKgK5%g1;^4@VAB-Zh$&HagPEHwHe)qCx%C zxW2Ja9g#xA{7`iJGKi2Iu{3L;Q%mRdb}>H#cf5=41b1X5IEIYgPG0TO#sRKxBW$2ro0np6zKb3}PlIQ)sW$MbX4 zU}AakarFh+K?IREn{^x?=y2Bzixs&s$aa=?urr<6c+ztNBrtJ6+5HAs4u6~*Iq9le zVSP0jR$$M^HxFDe-9!euAhm zJh5bxv^%(p)shk{*ENR2_X_?Y6qIxgW@C5>Cz90FA7%e@9D?GS!-v#xF5RDoKRD)g zVzcIkMq^kNNEN+|B5J~7c!1^RVwwFukz5yHR5(JFp%CqU+h%^#M>1n%yVqegpn>I) zC?h(Qpns$`@sfO$*gEc!p6J}6C~Xg@vcd`{%z=19w`1t-1^#}2oqW)SWEy;RiZJ*r zd{(OqZSwQsungoS7|}}5pPUPl5Bv07>f`Ap2C30&QsXk9X-sj3G0yR_@g`dCzWCvJ zC+JnQewXYOv~a5B!Tyj|<&3sVX&CCY3w1hWIylPVm`e6xGEV(la%uCR;Y7pFb)14N z?Tf4L!|>DPc|1J1Xfw>ZQ>h!?1elf&3T!gaiwQZs_4>2kl1gqp)Z5oDmC0T5-Cr`i zkDZr7H{q!8t`B9Wuf7`lnEvv@H=mdcUIDM0#yF_>eghhA&k2_NqU1sD75`*ZT8l=ctMz07+U60D34z7UAX^;S&^xB!Qlwq(CO2I z?@N!c3GA!pYKR7OVUu^c6cC#`#k};(v@>6^2>oHaQ0XIXNwM#)6*_K~k>BNvXdtyD zZ^Mi)#{m{nQ83>)i#SSD_ynidd}QJ}qOK7HT@TMiQZ)nC1`*$uT;dP-4D^)i^aP_zP){3;5-pS8lxYzO5kN#&R&%{F2$27!#ega zOE5wCHJiR)UZ4CeQN^KV3cmou5*zU~R)KLoBXHXH%$eEy;YtO;E90&`Gm4rn zVCeKzMdnsmqAMZ;Wu&FOYqPj0U0F*3sOx6-n$zibb%K(hOf5&6{J3!`TIx;h~lAO%DtG}^0!-wtj3umc7niw^E z&7S@g2QV{rq-IyE<-8t!xPFp-7C5wSe8^q~NBODmR*fp7Qc6T$ape^Fs+|eQRvrb+ zDf!-VsOE|i3g5d_=i=|&MBo=KT;&Fzt9c&!y7Q7uu>(9GadHn7YFvIvl6^I?7X0v| z%_?sZH_u5uoR}S&-@rxPajXE?liAH>Pi|`rh)`5NMk=t>RpacYGgIFRqPb=dK`Ho( zu2;w~1+fsh!e3Y|0~LHXr&m$>aBLQ?6jS#4$aYJ zW`E0yS$D?tf-8~^B#9790vk0&cx|$YS{8MmUJrJQ6;~X%*%db3s}>6%=td-auHTHy z^zUT#`4vZ?d_;e|JnVKz1YGgB>6gSCPNn=Bgn#QgJ0;wRpDB6jlVz4ilnW-qZiRqB zT%{OrMHw%b8eyhZ$Td}xab@J6i5V~t*&@VdUhT5%cgnajE_a_@M-fq8{XdmjvX_N0 zA$&7J)4ot)tQ;q6Jt$8XqZewZ-rLB?XxR3!53DdN_B@A(FY~}`Ewf-kJ5?N1?R!5v z7li^-ghbw@oh#{L&vIL+QWui2O#}&6&nq=PRJX;rFJbte6iO+ee%^46122YCVyvs z^{m=l`#Ar}l4|z^9-Qw6Z1*~Bbt`8KBZf)>)!y*lbkv4~qwtZ8 zrQUY;tr?%!6x@b^Ds2u#hw-@(oQ(l6+F`@Q2ou-mi<5&iI7wBcLbirufF@qelW{mU zBD=+LUSA{1d|!Q^Cd?F{Cd6&h!m*KLf4ybuNC09T^~V(N%T*&~utu|C`+Qqh zH>5ehTvOn}d-J{JXt8jre7!}tk+&F-Lur#LbLp$b4{9;_ng9Z^@N-VMek1EFB5twg zj89Gh5?vgd0e3Yh9$H-WuIl(aTDJ)}RCxnZ9b3-Zp!^M`$F_C_a6=v<#aaQLF1hAU z{(2l;a>?k1+pP#)k^bSqG)C^ITss|tldX|csAjdpz%|A!blj_`cU08+>uOYJ~f7Q4gGo^-Dvom6^uaK8Yi z;#Q&NgeHz5alm}M++WsBcv4jmbV6OG=c0KApH9Wcd(I3%o0S)vCn|z&eW#K(;L|$} zKINXrEiRxzw@dxQ$J}Y-tzadz2R557=Pu1I;z%zXj!k8RwKtK*pe{SZH8}w~$*eDB z*u0%ik3Cd4X4ii4zboU9eK>BDAA3C8qLVzP9og6)HyDjHuX@%;@0t~o^rE@HbsqYm z-)&u;CTnY)XD5JnfA^7`-*ih2L7v7%Ut1UR-6v@+d1c3+k@o+vcw{f+Zrpyr z`1Br)rZ}*-sLX|rWZ;}kG13{jZ#eL>`fy6SL}PO3@+LN1;yZ}*XU+13CoW+%;>z9` z8g=k7sPPOL>Lzzgs~z^BTTd@Ix9l=2%)7vr59K(DFNjIK@%2pJT$2ewSo)cXocoWDcWsIs!xb#LCM* zD)8{j#1BWCZp2ET`)@jRnt+$K6!oKmNED zlTxA)1t(`^ge>gafps$dJE`8T$k|cmd|uvvi+JP z^Pp+acI*WRNqMknHta|qO&yrd85RaC{ zjo1pObl;xCQ^LUAGzrwDL3vf2#Vkiey+xW z5|caI;-;g$-BPLVtr!k)i@|0IN-CoJ{xyqQw!na9h5bM+Oh>)_aQK)4pO$XJvRu<@ zVZ)yN{6nS%!)TXh=$Yhu6d}<5s$<$Nce4HKdQ}Ry9OlJ=|8*xs0%op5Tjgu@PhYx2 z`DVXgRVd^AyViBoUK9<71HW;FX!V@`h_f(qfhS^*G7$!-1j00k{cg|u)=BD*${J5N zkKjqu2TzmF$gt#&JK)E{AK2Wca?vjE}K5efq;jTlj(h&)KsOkOWjsB6D%R zLiT=o!4^z&oNJJcb>rgbHy>0-d;&2O%6^u&}q;cDvwSX?4bduA7cmozoX7v^Jxz8RJd} z55pyXl_!WThw0khm=#i14C6C!GKmGmh02BBq$I{C)x$=1!t2kkRO))&WTc!6uf{@) z&FrFu9FLOT4Qz|qVY|YWawXo9beu%ovvTPIf@eJOcrs6T+8krKQ zdl6%5AXvz#D2 zy4wDlh}_E5X60*Nd}Z5bI{f=JmFQ&}S!JNm0Nu*)hXmqV>JyQACqC;o0?((;eqhPTmTTp{~3ml@qZ0c0f2Pa8zN{J005AW z0EOfZfJVd@&i3_#LBs-Sb!GbnAYf2RB@5^H2cqD!TP}6w1O%g#i-h2VbAN&q~bu@J^7*Wv`V>BDIx`7i40n` zmdiZ_;YloCdO`?93nNlEEN6(x%9CqMhT`aS2g+0GET(d$(sQj#dY&vO0YLFv!GG)!E$vh}iUcL)AIG!5CCB z=~6Yh{o(i=)@wsGd4tj9qG80+wfV#GAk|X6;o5@HBvzA=bm_Xn@wCs*$7{oNMUz>= z0mvjW^~KY9Qt|ZqBlRV-MM`-x88Qu}^JQ9f*6Sk;Ws6l`dc#O$8_SpLEN4sgM;j|v zn;^R*8L~~4>uny_$Lpg_Rhyl@Fes#Q&DGnzA=nHCW6d?YgE3UHnQ|?)`=iMmHXCCt zb-yRGMZ-zuTkDTzi&e`E##;17 zeE<*|@K<2`B>(^jOTfqpz=wA>u{5(Y`sYCdk^`atQw8w12mvtx@&Cv2|9Z$rph5iG z2o3;1K#Y$HR#8#Z-QrHFIGyfk2pyoSYoo$=%GHoUI(}$$6MrnOO;l RiT~cY3lspL^!KO5{{Z~AGf4md literal 0 HcmV?d00001 diff --git a/datafusion/core/tests/sql/parquet.rs b/datafusion/core/tests/sql/parquet.rs index 8bec4f1dd3db..c70466737f40 100644 --- a/datafusion/core/tests/sql/parquet.rs +++ b/datafusion/core/tests/sql/parquet.rs @@ -48,6 +48,28 @@ async fn parquet_query() { assert_batches_eq!(expected, &actual); } +#[tokio::test] +async fn fixed_size_binary_columns() { + let ctx = SessionContext::new(); + ctx.register_parquet( + "t0", + "tests/parquet/test_binary.parquet", + ParquetReadOptions::default(), + ) + .await + .unwrap(); + let sql = "SELECT ids FROM t0 ORDER BY ids"; + let plan = ctx.create_logical_plan(sql).unwrap(); + let plan = ctx.optimize(&plan).unwrap(); + let plan = ctx.create_physical_plan(&plan).await.unwrap(); + let task_ctx = ctx.task_ctx(); + let results = collect(plan, task_ctx).await.unwrap(); + for batch in results { + assert_eq!(466, batch.num_rows()); + assert_eq!(1, batch.num_columns()); + } +} + #[tokio::test] async fn parquet_single_nan_schema() { let ctx = SessionContext::new();