@@ -7,8 +7,9 @@ use std::path::PathBuf;
77use anyhow:: anyhow;
88use async_trait:: async_trait;
99use lance:: Dataset ;
10- use lance:: dataset:: ProjectionRequest ;
1110use lance:: dataset:: WriteParams ;
11+ use lance:: dataset:: fragment:: FileFragment ;
12+ use lance:: datatypes:: Schema ;
1213use lance_file:: version:: LanceFileVersion ;
1314use parquet:: arrow:: arrow_reader:: ParquetRecordBatchReaderBuilder ;
1415use vortex_bench:: Format ;
@@ -30,7 +31,12 @@ async fn parquet_to_lance_file(parquet_path: PathBuf, lance_path: &str) -> anyho
3031 let builder = ParquetRecordBatchReaderBuilder :: try_new ( file) ?;
3132 let reader = builder. build ( ) ?;
3233
33- let write_params = WriteParams :: with_storage_version ( LanceFileVersion :: V2_1 ) ;
34+ // Write Lance's current stable format into a single fragment, so a take reads one file
35+ // like the Vortex and Parquet accessors do instead of opening a reader per fragment.
36+ let write_params = WriteParams {
37+ max_rows_per_file : usize:: MAX ,
38+ ..WriteParams :: with_storage_version ( LanceFileVersion :: Stable )
39+ } ;
3440 Dataset :: write (
3541 reader,
3642 output_fname
@@ -79,11 +85,14 @@ pub async fn nested_structs_lance() -> anyhow::Result<PathBuf> {
7985
8086/// Random accessor for Lance format files.
8187///
82- /// The dataset handle is opened at construction time and reused across `take()` calls.
88+ /// Like the Vortex and Parquet accessors, this reads the dataset's single file: rows are taken
89+ /// straight from its one fragment, opened once at construction and reused across `take()` calls.
90+ /// Going through `Dataset::take` would instead rebuild a projection plan and map dataset offsets
91+ /// to fragments on every call.
8392pub struct LanceRandomAccessor {
8493 name : String ,
85- dataset : Dataset ,
86- projection : ProjectionRequest ,
94+ fragment : FileFragment ,
95+ schema : Schema ,
8796}
8897
8998impl LanceRandomAccessor {
@@ -100,11 +109,18 @@ impl LanceRandomAccessor {
100109 /// Open a Lance dataset from any URI (local path, `s3://...`, ...).
101110 pub async fn open_uri ( uri : & str , name : impl Into < String > ) -> anyhow:: Result < Self > {
102111 let dataset = Dataset :: open ( uri) . await ?;
103- let projection = ProjectionRequest :: from_schema ( dataset. schema ( ) . clone ( ) ) ;
112+ let schema = dataset. schema ( ) . clone ( ) ;
113+ let fragment = match <[ _ ; 1 ] >:: try_from ( dataset. get_fragments ( ) ) {
114+ Ok ( [ fragment] ) => fragment,
115+ Err ( fragments) => anyhow:: bail!(
116+ "Lance random access expects one fragment, found {}" ,
117+ fragments. len( )
118+ ) ,
119+ } ;
104120 Ok ( Self {
105121 name : name. into ( ) ,
106- dataset ,
107- projection ,
122+ fragment ,
123+ schema ,
108124 } )
109125 }
110126}
@@ -120,7 +136,11 @@ impl RandomAccessor for LanceRandomAccessor {
120136 }
121137
122138 async fn take ( & self , indices : & [ u64 ] ) -> anyhow:: Result < RandomAccessorRet > {
123- let batch = self . dataset . take ( indices, self . projection . clone ( ) ) . await ?;
139+ let indices = indices
140+ . iter ( )
141+ . map ( |& idx| u32:: try_from ( idx) )
142+ . collect :: < Result < Vec < _ > , _ > > ( ) ?;
143+ let batch = self . fragment . take ( & indices, & self . schema ) . await ?;
124144 Ok ( RandomAccessorRet :: Native ( Box :: new ( batch) ) )
125145 }
126146}
0 commit comments