Updating read_lance to support pushdowns.

[everettVT]

Currently daft.read_lance implements reading fragements from lance.LanceDataset but there is a LanceDataset.scanner that has sophisticated support for pushdowns and other optimizations. The LanceDataset.get_fragements() call inside the LanceScanOperator isn't actually filtering anything since the method remains unimplemented on the Lance side as seen here .

This means daft is getting all fragments for the entire dataset regardless of the filters from the LanceScanOperator.

On the rust side however, get_fragements IS implemented here :

    /// Get fragments.
    ///
    /// If `filter` is provided, only fragments with the given name will be returned.
    pub fn get_fragments(&self) -> Vec<FileFragment> {
        let dataset = Arc::new(self.clone());
        self.manifest
            .fragments
            .iter()
            .map(|f| FileFragment::new(dataset.clone(), f.clone()))
            .collect()
    }

I've got no idea how this would need to translate on the daft side, but I figured we could start a discussion.
@rchowell @jackye1995 @Jay-ju

[rchowell]

The first step is to convert the daft expression into a PyArrow Expression, then we will be able to use this in our call to get_fragments.

 # TODO: figure out how to translate Pushdowns into LanceDB filters
filters = None
fragments = self._ds.get_fragments(filter=filters)

I actually have some outdated code to do this already, and implemented the appropriate introspection classes. Here are the PR's that laid the groundwork to make this possible, now it's time to apply it.

• feat: adds expression visitor #4278
• feat: defines pushdown terms for python #4203
• docs: adds end-to-end custom scan with pushdowns example #4243

@Jay-ju was looking into this, and I believe I can best assist here by implementing the translation. I'll keep y'all posted! 🤙

[rchowell]

Going to add some tests now. #4616

[rchowell]

Added some basic tests and hooked it up to the lance source. https://github.com/Eventual-Inc/Daft/pull/4616/files#diff-3c73499823a5ef5a19d3206cad6ef74fb422cc3b804d70d13651d8b4616d477d

[Jay-ju]

@everettVT I'm really glad that you're also working on this. Actually, I'm currently dealing with some matters of interfacing with Lance. I noticed that @rchowell has already submitted a PR #4616 for the pyarrow expression. I also submitted a PR for expression conversion at #4610 before. However, now I want to continue pushing down filter, limit, and count at https://github.com/Eventual-Inc/Daft/pull/4612，I want to know which part you want to work on?

[rchowell]

@Jay-ju please continue your work on the lance pushdowns! I will focus on the pyarrow expression translation only, so that we can progress together.

Interestingly I ran into this error from lance.

>         if filter is not None:
>           raise ValueError("get_fragments() does not support filter yet")
E           ValueError: get_fragments() does not support filter yet

Seems I'll need to check the Lance version to see if pushdowns are even possible! Maybe @jackye1995 can give a pointer to which version this became possible. Thanks!

[Jay-ju]

@Jay-ju please continue your work on the lance pushdowns! I will focus on the pyarrow expression translation only, so that we can progress together.

Interestingly I ran into this error from lance.

>         if filter is not None:
>           raise ValueError("get_fragments() does not support filter yet")
E           ValueError: get_fragments() does not support filter yet

Seems I'll need to check the Lance version to see if pushdowns are even possible! Maybe @jackye1995 can give a pointer to which version this became possible. Thanks!

Maybe I can solve this. I think the task here can be generated by the scanner at the granularity of the fragment. @rchowell

just like this

@jackye1995 , wdyt ？

[everettVT]

@Jay-ju I definitely agree with the use of the scanner to pass the pushdowns through.

In the LanceScanOperator.to_scan_tasks() method I imagine it would look something like:

class LanceScanOperator(ScanOperator):
    ...
    def to_scan_tasks(self, pushdowns: PyPushdowns) -> Iterator[ScanTask]:
        ...

        # convert to the python pushdowns
        pds = Pushdowns._from_pypushdowns(pushdowns)
        fragments = self._ds.get_fragments()
        
        scanner_ = self._ds.scanner(
            columns = self._get_select_pushdown(pds) if pushdowns.columns else None
            filter  = self._get_filter_pushdown(pds)  if pushdowns.filters else None
            limit   = self._get_limit_pushdown(pds)   if pushdowns.limit   else None
            batch_size = self.batch_size # or something
            use_stats = True, 
            scan_stats_callback = SomeDaft
            ... # rest of *args/**kwargs (includes support for stats which could address note below)
        )
        
        for i, fragment in enumerate(fragments):
            # TODO: figure out how if we can get this metadata from LanceDB fragments cheaply
            size_bytes = None
            stats = None

            # NOTE: `fragment.count_rows()` should result in 1 IO call for the data file
            # (1 fragment = 1 data file) and 1 more IO call for the deletion file (if present).
            # This could potentially be expensive to perform serially if there are thousands of files.
            # Given that num_rows isn't leveraged for much at the moment, and without statistics
            # we will probably end up materializing the data anyways for any operations, we leave this
            # as None.
            num_rows = None

            yield ScanTask.python_factory_func_scan_task(
                module=_lancedb_table_factory_function.__module__,
                func_name=_lancedb_table_factory_function.__name__,
                func_args=(self._ds, fragment.fragment_id, required_columns),
                schema=self.schema()._schema,
                num_rows=num_rows,
                size_bytes=size_bytes,
                pushdowns=pushdowns,
                stats=stats,
            )
    

            
    def _get_select_pushdown(self, pushdowns: Pushdowns) -> pc.Expression:
        try:
            return pushdowns.select.to_arrow_expr()
        except ValueError:
            return None
            
    def _get_filter_pushdown(self, pushdowns: Pushdowns) -> pc.Expression:
        try:
            # best attempt convert to pyarrow.compute.Expression
            return pushdowns.filters.to_arrow_expr()
        except ValueError:
            # don't error, just accept that we couldn't push this down.
            return None
    
    def _get_limit_pushdown(self, pushdowns: Pushdowns) -> pc.Expression:
        try:
            # best attempt convert to pyarrow.compute.Expression
            return pushdowns.filters.to_arrow_expr()
        except ValueError:
            # don't error, just accept that we couldn't push this down.
            return None

This is notional, since I know you'll be working on this @Jay-ju .
The full args/kwargs supported in LanceDataset.scanner() are worth reviewing. Particularly if daft intends to support lance's FTS and stats callback.

Another consideration is building in support for args like batch_size and io_buffer_size. I'm not as sure about others like offset (Fetch starting with this row) or fast_search(search will only be performed on the indexed data). The arg for late_materialization is an interesting one as well given daft's lazy evaluation.

One thing I am thinking about your approach above @Jay-ju , is that we are still scanning all fragments even with our pushdowns.

I am struck by a more fundamental question of whether it is more appropriate to use the LanceDataset.scanner() or LanceFragment.scanner() methods, or possibly even both here. LanceFragment

---

Concerning coordinating the work surrounding these updates. My initial concern that sparked this larger conversation was concerned with adding support for LanceTable abstractions in daft's catalog feature. Naturally, I think there are several dependencies here that need to be solved before I add a high order feature like that.

I'm happy to help as much as I can. I don't think we've quite figured this one out, so I am not sure how the work should be divided.

I do know that once we get something working here, I would suggest a few housekeeping edits concerning the organization of these lower level lance connector features into a dedicated lance folder, similar to iceberg and other formats.

Overall I can certainly help with the discussions and research here. Once we get to testing I'd be happy to divide that up and take on the housekeeping items.

[Jay-ju]

@everettVT, I kind of understand what you mean. Here, I think we should give priority to using lancedataset.scanner. Because we do need to push down the statements of fts and fast_search, and these are not available on the fragment interface. However, it should be emphasized that the scanner interface of the Lance Dataset also has the parameter fragments: Optional[Iterable[LanceFragment]] = None, so this is not contradictory.

[Jay-ju]

[everettVT]

I see! This helps me understand. Thank you.

[Jay-ju]

@everettVT So Sorry, there should be a problem with the content I screenshot for you just now. Here you can see that we still use the dataset interface, but the scanner_options contains the key of fragments.

 fragments = [lance_ds.get_fragment(id) for id in fragment_ids]
    scanner_options["fragments"] = fragments
    scanner = lance_ds.scanner(**scanner_options)

[everettVT]

Oh interesting. So the fragments argument in LanceDataset.scanner is comprised of the id's? Not the raw fragments directly?

[Jay-ju]

@everettVT is raw fragment， https://github.com/Eventual-Inc/Daft/pull/4622/files

[everettVT]

I think this pushdown discussion deserves some deep work focus. I'm trying to finish up a few other things in another project at the moment, but would like to dive into the internals of profiling different approaches here.

It may be trivial to fetch all fragments.