shyjsarah commented on code in PR #760:
URL: https://github.com/apache/paimon-rust/pull/760#discussion_r3892803733
##########
crates/integrations/datafusion/src/lateral_vector_search.rs:
##########
@@ -352,9 +450,57 @@ struct LateralVectorSearchExec {
query_vector_expr: Arc<dyn PhysicalExpr>,
limit: usize,
output_schema: ArrowSchemaRef,
+ filter: Option<Predicate>,
+ prepared_filter_cache: Arc<ExecutionPreparedFilterCache>,
plan_properties: Arc<PlanProperties>,
}
+#[derive(Debug)]
+struct ExecutionPreparedFilterEntry {
+ context: Weak<TaskContext>,
+ prepared_filter: Weak<OnceCell<PreparedVectorSearchFilter>>,
+}
+
+#[derive(Debug, Default)]
+struct ExecutionPreparedFilterCache {
+ // DataFusion passes the same TaskContext Arc to every partition of one
+ // execution. Weak references let those partitions share one prepared
filter
+ // while ensuring a later execution of the reusable plan starts fresh.
+ entries: Mutex<Vec<ExecutionPreparedFilterEntry>>,
+}
+
+impl ExecutionPreparedFilterCache {
+ fn for_execution(
+ &self,
+ context: &Arc<TaskContext>,
+ ) -> Arc<OnceCell<PreparedVectorSearchFilter>> {
+ let mut entries = self
+ .entries
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ entries.retain(|entry| {
+ entry.context.strong_count() > 0 &&
entry.prepared_filter.strong_count() > 0
+ });
+ for entry in entries.iter() {
+ let Some(entry_context) = entry.context.upgrade() else {
+ continue;
+ };
+ if Arc::ptr_eq(&entry_context, context) {
+ if let Some(prepared_filter) = entry.prepared_filter.upgrade()
{
+ return prepared_filter;
+ }
+ }
+ }
+
+ let prepared_filter = Arc::new(OnceCell::new());
+ entries.push(ExecutionPreparedFilterEntry {
+ context: Arc::downgrade(context),
+ prepared_filter: Arc::downgrade(&prepared_filter),
+ });
Review Comment:
Fixed in 339321c. The execution cache now strongly owns the prepared filter
while the TaskContext is alive, so sequential partitions share the same
resolved snapshot. I also updated the regression test to execute two partitions
sequentially, mutate/reindex the target between them, verify the second
partition still sees the original snapshot, and verify a new TaskContext sees
the refreshed snapshot.
##########
crates/paimon/src/lumina/reader.rs:
##########
@@ -369,7 +369,7 @@ fn search_lumina_batch(
}
if vector_searches
.iter()
- .any(|vector_search| vector_search.include_row_ids.is_some())
+ .any(|vector_search|
vector_search.effective_include_row_ids().is_some())
{
Review Comment:
Fixed in 66516d1. Lumina now detects a shared filter by Arc identity,
materializes the bitmap once, and calls search_with_filter once for the full
query batch. Differing/mixed filters and differing limits retain the scalar
fallback; empty shared filters return without native calls. Added mock-backed
tests for the batch call and fallback cases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
