Add optimizer rule to change `LanceScanExec -> FilterExec` into `FilteredReadExec`

[wjones127]

There are query plans like this one that could benefit from pushing down the filter:

https://github.com/lancedb/lance/blob/35a0547426c47dd94425619c442dcd03ffd9860c/rust/lance/src/dataset/scanner.rs#L7157-L7158

Note: this will become obsolete if we change Scanner to construct a logical plan using the LanceTableProvider since pushdown can be handled at that layer. However, that's a ways off.

https://github.com/lancedb/lance/blob/35a0547426c47dd94425619c442dcd03ffd9860c/rust/lance/src/datafusion/dataframe.rs#L147-L158

[Maxwell-Guo]

please assign this issue to me , thanks . @wjones127

[wkalt]

@Maxwell-Guo in case it is useful, this script produces a few bad plans:

#!/bin/env python3

import lancedb
import numpy as np
import time

def add_rows(table, start, end, chunk_size=100000):
    for i in range(start, end, chunk_size):
        chunk = [{"scalar_id": j, "vector": np.random.random(4096)} for j in range(i, min(i + chunk_size, end))]
        table.add(chunk)

def create_scenario(db, scenario):
    table_name = f"test_random_vectors_{scenario.replace('-', '_')}"
    if table_name in db.table_names():
        return db.open_table(table_name)

    t = db.create_table(table_name, [{"scalar_id": i, "vector": np.random.random(4096)} for i in range(100000)])

    if scenario == "fully-indexed":
        add_rows(t, 100000, 700000)
        t.create_index(vector_column_name="vector", index_type="IVF_PQ")
        t.create_scalar_index("scalar_id", index_type="BTREE")
    elif scenario == "fully-unindexed":
        add_rows(t, 100000, 1000000)
    elif scenario == "partially-indexed":
        add_rows(t, 100000, 700000)
        t.create_index(vector_column_name="vector", index_type="IVF_PQ")
        t.create_scalar_index("scalar_id", index_type="BTREE")
        add_rows(t, 0, 300000)
    elif scenario == "full-vector-partial-btree":
        add_rows(t, 100000, 700000)
        t.create_index(vector_column_name="vector", index_type="IVF_PQ")
        t.create_scalar_index("scalar_id", index_type="BTREE")
        add_rows(t, 0, 300000)
        t.create_index(vector_column_name="vector", index_type="IVF_PQ")
    elif scenario == "partial-vector-partial-btree":
        add_rows(t, 100000, 700000)
        t.create_index(vector_column_name="vector", index_type="IVF_PQ")
        t.create_scalar_index("scalar_id", index_type="BTREE")
        add_rows(t, 0, 300000)
    return t

db = lancedb.connect(".")
vec = np.random.random(4096)

if __name__ == '__main__':
    scenarios = [
        "fully-indexed",
        "fully-unindexed",
        "partially-indexed",
        "full-vector-partial-btree",
        "partial-vector-partial-btree"
    ]
    for scenario in scenarios:
        t = create_scenario(db, scenario)
        print(f"\n=== {scenario} ===")
        start = time.time()
        print(t.search(vec).where("scalar_id in (9999)").limit(100).nprobes(20).analyze_plan())
        elapsed = time.time() - start
        print(f"Query time: {elapsed:.4f}s")

The output I get is this:

=== fully-indexed ===
AnalyzeExec verbose=true, metrics=[], cumulative_cpu=36.13771ms
  TracedExec, metrics=[], cumulative_cpu=36.13771ms
    ProjectionExec: expr=[scalar_id@2 as scalar_id, vector@3 as vector, _distance@0 as _distance], metrics=[output_rows=0, elapsed_compute=1ns], cumulative_cpu=36.13771ms
      Take: columns="_distance, _rowid, (scalar_id), (vector)", metrics=[output_rows=0, elapsed_compute=1ns, batches_processed=0, bytes_read=0, iops=0, requests=0], cumulative_cpu=36.137709ms
        CoalesceBatchesExec: target_batch_size=1024, metrics=[output_rows=0, elapsed_compute=11ns], cumulative_cpu=36.137708ms
          SortExec: TopK(fetch=100), expr=[_distance@0 ASC NULLS LAST, _rowid@1 ASC NULLS LAST], preserve_partitioning=[false], metrics=[output_rows=0, elapsed_compute=17.527µs, row_replacements=0], cumulative_cpu=36.137697ms
            ANNSubIndex: name=vector_idx, k=100, deltas=1, metrics=[output_rows=0, elapsed_compute=25.126056ms, index_comparisons=68181, indices_loaded=0, partitions_searched=20, parts_loaded=20], cumulative_cpu=36.12017ms
              ANNIvfPartition: uuid=6335819c-b15c-45b9-93eb-71bafc1141a0, minimum_nprobes=20, maximum_nprobes=Some(20), deltas=1, metrics=[output_rows=1, elapsed_compute=10.310017ms, deltas_searched=1, index_comparisons=0, indices_loaded=1, partitions_ranked=836, parts_loaded=0], cumulative_cpu=10.310017ms
              ScalarIndexQuery: query=[scalar_id = 9999]@scalar_id_idx, metrics=[output_rows=2, elapsed_compute=684.097µs, index_comparisons=4096, indices_loaded=1, output_batches=1, parts_loaded=1, search_time=1.044709ms, ser_time=17.342µs], cumulative_cpu=684.097µs

Query time: 0.0281s

=== fully-unindexed ===
AnalyzeExec verbose=true, metrics=[], cumulative_cpu=5.148648714s
  TracedExec, metrics=[], cumulative_cpu=5.148648714s
    ProjectionExec: expr=[scalar_id@0 as scalar_id, vector@1 as vector, _distance@3 as _distance], metrics=[output_rows=1, elapsed_compute=571ns], cumulative_cpu=5.148648714s
      GlobalLimitExec: skip=0, fetch=100, metrics=[output_rows=1, elapsed_compute=10ns], cumulative_cpu=5.148648143s
        FilterExec: _distance@3 IS NOT NULL, metrics=[output_rows=1, elapsed_compute=2.404µs], cumulative_cpu=5.148648133s
          SortExec: TopK(fetch=100), expr=[_distance@3 ASC NULLS LAST, _rowid@2 ASC NULLS LAST], preserve_partitioning=[false], metrics=[output_rows=1, elapsed_compute=54.323µs, row_replacements=1], cumulative_cpu=5.148645729s
            KNNVectorDistance: metric=l2, metrics=[output_rows=1, elapsed_compute=2.588792479s, output_batches=1], cumulative_cpu=5.148591406s
              FilterExec: scalar_id@0 = 9999, metrics=[output_rows=1, elapsed_compute=4.238µs], cumulative_cpu=2.559798927s
                LanceRead: uri=mnt/work/home/wyatt/reproducer/test_random_vectors_fully_unindexed.lance/data, projection=[scalar_id, vector], num_fragments=10, range_before=None, range_after=None, row_id=true, row_addr=false, full_filter=scalar_id = Int64(9999), refine_filter=scalar_id = Int64(9999), metrics=[output_rows=1, elapsed_compute=2.559794689s, fragments_scanned=10, ranges_scanned=10, rows_scanned=1000000, bytes_read=16392044130, iops=1010, requests=520, task_wait_time=2.559564762s], cumulative_cpu=2.559794689s

Query time: 2.6296s

=== partially-indexed ===
AnalyzeExec verbose=true, metrics=[], cumulative_cpu=1.544708124s
  TracedExec, metrics=[], cumulative_cpu=1.544708124s
    ProjectionExec: expr=[scalar_id@3 as scalar_id, vector@1 as vector, _distance@2 as _distance], metrics=[output_rows=1, elapsed_compute=1.633µs], cumulative_cpu=1.544708124s
      Take: columns="_rowid, vector, _distance, (scalar_id)", metrics=[output_rows=1, elapsed_compute=104.659µs, batches_processed=1, bytes_read=0, iops=0, requests=0], cumulative_cpu=1.544706491s
        CoalesceBatchesExec: target_batch_size=1024, metrics=[output_rows=1, elapsed_compute=3.035µs], cumulative_cpu=1.544601832s
          GlobalLimitExec: skip=0, fetch=100, metrics=[output_rows=1, elapsed_compute=10ns], cumulative_cpu=1.544598797s
            FilterExec: _distance@2 IS NOT NULL, metrics=[output_rows=1, elapsed_compute=2.314µs], cumulative_cpu=1.544598787s
              SortExec: TopK(fetch=100), expr=[_distance@2 ASC NULLS LAST, _rowid@0 ASC NULLS LAST], preserve_partitioning=[false], metrics=[output_rows=1, elapsed_compute=16.131µs, row_replacements=1], cumulative_cpu=1.544596473s
                KNNVectorDistance: metric=l2, metrics=[output_rows=1, elapsed_compute=22.854µs, output_batches=1], cumulative_cpu=1.544580342s
                  RepartitionExec: partitioning=RoundRobinBatch(1), input_partitions=2, metrics=[fetch_time=792.268144ms, repartition_time=2ns, send_time=2.075µs], cumulative_cpu=1.544557488s
                    UnionExec, metrics=[output_rows=1, elapsed_compute=14.859µs], cumulative_cpu=1.544557488s
                      ProjectionExec: expr=[_distance@3 as _distance, _rowid@2 as _rowid, vector@0 as vector], metrics=[output_rows=1, elapsed_compute=762ns], cumulative_cpu=1.532555941s
                        FilterExec: _distance@3 IS NOT NULL, metrics=[output_rows=1, elapsed_compute=2.324µs], cumulative_cpu=1.532555179s
                          SortExec: TopK(fetch=100), expr=[_distance@3 ASC NULLS LAST, _rowid@2 ASC NULLS LAST], preserve_partitioning=[false], metrics=[output_rows=1, elapsed_compute=25.969µs, row_replacements=1], cumulative_cpu=1.532552855s
                            KNNVectorDistance: metric=l2, metrics=[output_rows=1, elapsed_compute=768.705412ms, output_batches=1], cumulative_cpu=1.532526886s
                              FilterExec: scalar_id@1 = 9999, metrics=[output_rows=1, elapsed_compute=2.114738ms], cumulative_cpu=763.821474ms
                                LanceScan: uri=mnt/work/home/wyatt/reproducer/test_random_vectors_partially_indexed.lance/data, projection=[vector, scalar_id], row_id=true, row_addr=false, ordered=false, range=None, metrics=[output_rows=300000, elapsed_compute=761.706736ms, bytes_read=4917613239, iops=303, requests=156], cumulative_cpu=761.706736ms
                      Take: columns="_distance, _rowid, (vector)", metrics=[output_rows=0, elapsed_compute=1ns, batches_processed=0, bytes_read=0, iops=0, requests=0], cumulative_cpu=11.986688ms
                        CoalesceBatchesExec: target_batch_size=1024, metrics=[output_rows=0, elapsed_compute=240ns], cumulative_cpu=11.986687ms
                          SortExec: TopK(fetch=100), expr=[_distance@0 ASC NULLS LAST, _rowid@1 ASC NULLS LAST], preserve_partitioning=[false], metrics=[output_rows=0, elapsed_compute=108.014µs, row_replacements=0], cumulative_cpu=11.986447ms
                            ANNSubIndex: name=vector_idx, k=100, deltas=1, metrics=[output_rows=0, elapsed_compute=10.614492ms, index_comparisons=72927, indices_loaded=0, partitions_searched=20, parts_loaded=20], cumulative_cpu=11.878433ms
                              ANNIvfPartition: uuid=38a2dc14-fd91-4d0b-9425-494c6aa15485, minimum_nprobes=20, maximum_nprobes=Some(20), deltas=1, metrics=[output_rows=1, elapsed_compute=777.623µs, deltas_searched=1, index_comparisons=0, indices_loaded=0, partitions_ranked=836, parts_loaded=0], cumulative_cpu=777.623µs
                              ScalarIndexQuery: query=[scalar_id = 9999]@scalar_id_idx, metrics=[output_rows=2, elapsed_compute=486.318µs, index_comparisons=4096, indices_loaded=1, output_batches=1, parts_loaded=1, search_time=1.313898ms, ser_time=13.195µs], cumulative_cpu=486.318µs

Query time: 0.7914s

=== full-vector-partial-btree ===
AnalyzeExec verbose=true, metrics=[], cumulative_cpu=16.49465ms
  TracedExec, metrics=[], cumulative_cpu=16.49465ms
    ProjectionExec: expr=[scalar_id@2 as scalar_id, vector@3 as vector, _distance@0 as _distance], metrics=[output_rows=0, elapsed_compute=1ns], cumulative_cpu=16.49465ms
      Take: columns="_distance, _rowid, (scalar_id), (vector)", metrics=[output_rows=0, elapsed_compute=1ns, batches_processed=0, bytes_read=0, iops=0, requests=0], cumulative_cpu=16.494649ms
        CoalesceBatchesExec: target_batch_size=1024, metrics=[output_rows=0, elapsed_compute=80ns], cumulative_cpu=16.494648ms
          SortExec: TopK(fetch=100), expr=[_distance@0 ASC NULLS LAST, _rowid@1 ASC NULLS LAST], preserve_partitioning=[false], metrics=[output_rows=0, elapsed_compute=31.659µs, row_replacements=0], cumulative_cpu=16.494568ms
            ANNSubIndex: name=vector_idx, k=100, deltas=1, metrics=[output_rows=0, elapsed_compute=8.843018ms, index_comparisons=92874, indices_loaded=0, partitions_searched=20, parts_loaded=20], cumulative_cpu=16.462909ms
              ANNIvfPartition: uuid=4d5900bc-e30b-4b3b-9125-f6f3c19fdff5, minimum_nprobes=20, maximum_nprobes=Some(20), deltas=1, metrics=[output_rows=1, elapsed_compute=6.409689ms, deltas_searched=1, index_comparisons=0, indices_loaded=1, partitions_ranked=1000, parts_loaded=0], cumulative_cpu=6.409689ms
              LanceRead: uri=mnt/work/home/wyatt/reproducer/test_random_vectors_full_vector_partial_btree.lance/data, projection=[], num_fragments=10, range_before=None, range_after=None, row_id=true, row_addr=false, full_filter=scalar_id = Int64(9999), refine_filter=--, metrics=[output_rows=2, elapsed_compute=980.397µs, fragments_scanned=4, ranges_scanned=4, rows_scanned=300001, bytes_read=2413239, iops=9, requests=9, task_wait_time=968.067µs], cumulative_cpu=1.210202ms
                ScalarIndexQuery: query=[scalar_id = 9999]@scalar_id_idx, metrics=[output_rows=2, elapsed_compute=229.805µs, index_comparisons=4096, indices_loaded=1, output_batches=1, parts_loaded=1, search_time=363.829µs, ser_time=6.613µs], cumulative_cpu=229.805µs

Query time: 0.0109s

=== partial-vector-partial-btree ===
AnalyzeExec verbose=true, metrics=[], cumulative_cpu=1.540100347s
  TracedExec, metrics=[], cumulative_cpu=1.540100347s
    ProjectionExec: expr=[scalar_id@3 as scalar_id, vector@1 as vector, _distance@2 as _distance], metrics=[output_rows=1, elapsed_compute=832ns], cumulative_cpu=1.540100347s
      Take: columns="_rowid, vector, _distance, (scalar_id)", metrics=[output_rows=1, elapsed_compute=148.871µs, batches_processed=1, bytes_read=0, iops=0, requests=0], cumulative_cpu=1.540099515s
        CoalesceBatchesExec: target_batch_size=1024, metrics=[output_rows=1, elapsed_compute=2.104µs], cumulative_cpu=1.539950644s
          GlobalLimitExec: skip=0, fetch=100, metrics=[output_rows=1, elapsed_compute=20ns], cumulative_cpu=1.53994854s
            FilterExec: _distance@2 IS NOT NULL, metrics=[output_rows=1, elapsed_compute=1.483µs], cumulative_cpu=1.53994852s
              SortExec: TopK(fetch=100), expr=[_distance@2 ASC NULLS LAST, _rowid@0 ASC NULLS LAST], preserve_partitioning=[false], metrics=[output_rows=1, elapsed_compute=14.597µs, row_replacements=1], cumulative_cpu=1.539947037s
                KNNVectorDistance: metric=l2, metrics=[output_rows=1, elapsed_compute=33.505µs, output_batches=1], cumulative_cpu=1.53993244s
                  RepartitionExec: partitioning=RoundRobinBatch(1), input_partitions=2, metrics=[fetch_time=789.406346ms, repartition_time=2ns, send_time=2.426µs], cumulative_cpu=1.539898935s
                    UnionExec, metrics=[output_rows=1, elapsed_compute=26.61µs], cumulative_cpu=1.539898935s
                      ProjectionExec: expr=[_distance@3 as _distance, _rowid@2 as _rowid, vector@0 as vector], metrics=[output_rows=1, elapsed_compute=1.032µs], cumulative_cpu=1.52819678s
                        FilterExec: _distance@3 IS NOT NULL, metrics=[output_rows=1, elapsed_compute=2.054µs], cumulative_cpu=1.528195748s
                          SortExec: TopK(fetch=100), expr=[_distance@3 ASC NULLS LAST, _rowid@2 ASC NULLS LAST], preserve_partitioning=[false], metrics=[output_rows=1, elapsed_compute=35.707µs, row_replacements=1], cumulative_cpu=1.528193694s
                            KNNVectorDistance: metric=l2, metrics=[output_rows=1, elapsed_compute=765.693524ms, output_batches=1], cumulative_cpu=1.528157987s
                              FilterExec: scalar_id@1 = 9999, metrics=[output_rows=1, elapsed_compute=1.704055ms], cumulative_cpu=762.464463ms
                                LanceScan: uri=mnt/work/home/wyatt/reproducer/test_random_vectors_partial_vector_partial_btree.lance/data, projection=[vector, scalar_id], row_id=true, row_addr=false, ordered=false, range=None, metrics=[output_rows=300000, elapsed_compute=760.760408ms, bytes_read=4917613239, iops=303, requests=156], cumulative_cpu=760.760408ms
                      Take: columns="_distance, _rowid, (vector)", metrics=[output_rows=0, elapsed_compute=1ns, batches_processed=0, bytes_read=0, iops=0, requests=0], cumulative_cpu=11.675545ms
                        CoalesceBatchesExec: target_batch_size=1024, metrics=[output_rows=0, elapsed_compute=10ns], cumulative_cpu=11.675544ms
                          SortExec: TopK(fetch=100), expr=[_distance@0 ASC NULLS LAST, _rowid@1 ASC NULLS LAST], preserve_partitioning=[false], metrics=[output_rows=0, elapsed_compute=99.88µs, row_replacements=0], cumulative_cpu=11.675534ms
                            ANNSubIndex: name=vector_idx, k=100, deltas=1, metrics=[output_rows=0, elapsed_compute=8.898463ms, index_comparisons=75406, indices_loaded=0, partitions_searched=20, parts_loaded=20], cumulative_cpu=11.575654ms
                              ANNIvfPartition: uuid=7f22bcc3-3c14-4e96-a294-8fb2105b80de, minimum_nprobes=20, maximum_nprobes=Some(20), deltas=1, metrics=[output_rows=1, elapsed_compute=1.905088ms, deltas_searched=1, index_comparisons=0, indices_loaded=0, partitions_ranked=836, parts_loaded=0], cumulative_cpu=1.905088ms
                              ScalarIndexQuery: query=[scalar_id = 9999]@scalar_id_idx, metrics=[output_rows=2, elapsed_compute=772.103µs, index_comparisons=4096, indices_loaded=1, output_batches=1, parts_loaded=1, search_time=1.08257ms, ser_time=50.977µs], cumulative_cpu=772.103µs

Query time: 0.7889s

in principle all of these queries should be executable in milliseconds. There seem to be gaps in both,

• LanceScan is not optimized to LanceRead in all cases -- that's what this ticket describes
• LanceRead is executing these filters suboptimally - it's reading both columns when scanning the scalar column and taking vector rows would be much better.

Based on the way the "unindexed" case is working, we might not actually see improvement in these plans until both those issues are fixed. I'm not sure if it makes sense to bite both of them off at the same time though.

[Maxwell-Guo]

@wkalt thanks

[lichuang]

Is there any progress of this issue? @Maxwell-Guo