FilteredReadExec, Plan, Split, Execute with physical offsets

[wjones127]

Support being able to provide a mask of physical row addresses. This can be used to define granular splits for distributed execution. For example, read first 500 rows on one node, and 500 rows on second node.

[LuQQiu]

The RowIdTreeMap can be on different levels

• FilteredReadExec still do it's own planning, but with this RowIdTreeMap in mind, those are the rows that are allowed to run on top
• FilteredReadExec skips its planning, the RowIdTreeMap is the planned result

Would want to discuss these two versions. First one would be the easier and safer one while may have duplicate planning work.

[LuQQiu]

The FilteredReadExec can be splited to three phase

1. Plan: Optionally execute index, apply scan_range_before_filter, find out what fragments need to be read, what rows need to be read from this fragment, and what left over optional predicate needs to be applied to this fragment
2. Split: This can be independent of this exec. This can be query engine's logic, about based on what to read and what to filter, decide how to split the job. The job split logic can be lance metadata related (e.g. based on split io size like 8MB, or can be based on compute overhead, or can be based on caching logics etc). We can provide some general methods for split but this is highly related to the query / compute engine, can make it extensible.
3. Execute, for the original planned result, or splitted result, can execute on that split/partition.

I will mainly work on 1 and 3. The 2 Split is compute/query engine specific, we can add that later.

[LuQQiu]

 Design Summary



 FilteredReadExec
 │
 ├── Mode: PlanAndExecute (original)
 │   ├── has index_input
 │   ├── can call plan() to get FilteredReadPlan
 │   └── execute() does planning internally
 │
 └── Mode: ExecuteOnly (new)
     ├── has precomputed FilteredReadPlan
     ├── no index_input needed
     └── execute() skips planning, uses plan directly

 Usage Modes

 // Mode 1: All-in-one (DataFusion compatible, current behavior)
 let exec = FilteredReadExec::new(dataset, options, index_input);
 let stream = exec.execute(0, ctx)?;  // plans + executes


 // Mode 2: Plan then execute locally
 let exec = FilteredReadExec::new(dataset, options, index_input);
 let plan = exec.plan(&ctx).await?;
 let exec2 = exec.with_plan(plan);
 let stream = exec2.execute(0, ctx)?;  // skips planning


 // Mode 3: Distributed (plan on coordinator, execute on workers)
 // Coordinator:
 let exec = FilteredReadExec::new(dataset, options, index_input);
 let plan = exec.plan(&ctx).await?;
 let splits = user_split_logic(&plan);
 for split in splits {
     let worker_exec = exec.with_plan(split);
     send_to_worker(worker_exec);  // serialize and send
 }
 // Worker:
 let stream = worker_exec.execute(0, ctx)?;  // skips planning

  pub struct FilteredReadPlan {
      pub fragments: Vec<FragmentPlan>,
  }

  pub struct FragmentPlan {
      pub fragment_id: u32,
      pub rows: RowAddrSelection,
      pub filter: Option<Arc<Expr>>,
  }

  pub enum RowAddrSelection {
      Full,
      Partial(RoaringBitmap),
  }

For separate plan / exec mode, caller needs to deal with metrics and dependencies properly.

  let exec = FilteredReadExec::new(dataset, options, index_input);
  let plan = exec.plan(&ctx).await?;

  // Traverse exec's children to get index metrics
  fn collect_metrics(exec: &dyn ExecutionPlan) -> Vec<(&str,
  Option<MetricsSet>)> {
      let mut result = vec![(exec.name(), exec.metrics())];
      for child in exec.children() {
          result.extend(collect_metrics(child.as_ref()));
      }
      result
  }

  let all_metrics = collect_metrics(&exec);
  // [("FilteredReadExec", Some(...)), ("ScalarIndexExec", Some(...)), ...]

you can still get the FilteredReadExec plan metrics if any and the inner index exec's metrics if any, in distributed mode, please propagate the metrics correctly.

[LuQQiu]

#5843