remove arrow dependency from data skipping #132
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -24,21 +24,39 @@ struct LogReplayScanner { | |
| struct AddRemoveVisitor { | ||
| adds: Vec<Add>, | ||
| removes: Vec<Remove>, | ||
| selection_vector: Option<Vec<bool>>, | ||
| } | ||
|
|
||
| const ADD_FIELD_COUNT: usize = 15; | ||
|
|
||
| impl AddRemoveVisitor { | ||
| fn new(selection_vector: Option<Vec<bool>>) -> Self { | ||
| AddRemoveVisitor { | ||
| selection_vector, | ||
| ..Default::default() | ||
| } | ||
| } | ||
| } | ||
|
|
||
| impl DataVisitor for AddRemoveVisitor { | ||
| fn visit<'a>(&mut self, row_count: usize, getters: &[&'a dyn GetData<'a>]) -> DeltaResult<()> { | ||
| for i in 0..row_count { | ||
| // Add will have a path at index 0 if it is valid | ||
| if let Some(path) = getters[0].get_opt(i, "add.path")? { | ||
| // Keep the file unless the selection vector is present and is false for this row | ||
| if !self | ||
| .selection_vector | ||
| .as_ref() | ||
| .is_some_and(|selection| !selection[i]) | ||
| { | ||
| self.adds | ||
| .push(AddVisitor::visit_add(i, path, &getters[..ADD_FIELD_COUNT])?) | ||
|
Comment on lines
+50
to
+56
There was a problem hiding this comment. Gotta love functional languages, where the "simpler" approach actually produces more lines of code 🤦 |
||
| } | ||
| } | ||
| // Remove will have a path at index 15 if it is valid | ||
| // TODO(nick): Should count the fields in Add to ensure we don't get this wrong if more | ||
| // are added | ||
| // TODO(zach): add a check for selection vector that we never skip a remove | ||
| else if let Some(path) = getters[ADD_FIELD_COUNT].get_opt(i, "remove.path")? { | ||
| let remove_getters = &getters[ADD_FIELD_COUNT..]; | ||
| self.removes | ||
|
|
@@ -70,15 +88,13 @@ impl LogReplayScanner { | |
| actions: &dyn EngineData, | ||
| is_log_batch: bool, | ||
| ) -> DeltaResult<Vec<Add>> { | ||
|
There was a problem hiding this comment. If we applied the JVM kernel approach here, this method would take the engine data as input, and return a selection vector that covers both data skipping and log replay for rows of this batch. The engine data would not be modified at all, and engine is free to apply the filtering however it wishes. This also avoids the need to (pay the cost to) parse There was a problem hiding this comment.
Oh like the selection vector would just 'pick' which actions (which would all be
we still would need to inspect the path/dv in order to add to the 'seen' set and perform the log replay (remove action) filtering, correct? You're just suggesting to do this on-the-fly instead of parsing into structs? There was a problem hiding this comment. Yes, the visitor isn't required to parse the exploded fields into a struct! It could just ask for just the fields we need to examine, update the selection vector accordingly, and let the engine data continue providing the actual rows. |
||
| // apply data skipping to get back a selection vector for actions that passed skipping | ||
| // note: None implies all files passed data skipping. | ||
| let selection_vector = self | ||
| .filter | ||
| .as_ref() | ||
| .map(|filter| filter.apply(actions)) | ||
| .transpose()?; | ||
|
|
||
| let schema_to_use = StructType::new(if is_log_batch { | ||
| vec![ | ||
|
|
@@ -90,12 +106,12 @@ impl LogReplayScanner { | |
| // only serve as tombstones for vacuum jobs. So no need to load them here. | ||
| vec![crate::actions::schemas::ADD_FIELD.clone()] | ||
|
There was a problem hiding this comment. I don't think this works -- the visitor blindly accesses There was a problem hiding this comment. Ahh, great catch. Can't we just propagate There was a problem hiding this comment. That could work as a short-term mitigation, yes. Longer term, we want the schema filtering to happen higher up in the stack, so there isn't even a remove column and the There was a problem hiding this comment. fixed with short-term here :) |
||
| }); | ||
|
There was a problem hiding this comment. checkpoint bugfix: passing is_log_batch down to visitor |
||
| let mut visitor = AddRemoveVisitor::new(selection_vector); | ||
| actions.extract(Arc::new(schema_to_use), &mut visitor)?; | ||
|
|
||
| for remove in visitor.removes.into_iter() { | ||
|
There was a problem hiding this comment. NOTE: We don't need to process removes before adds within any given batch, because each batch is a subset of some commit, and a given commit cannot legally contain more than one action for a given path. We just need to remember them in There was a problem hiding this comment. ah yep +1 I can fix in a separate (small) PR? There was a problem hiding this comment. What are we fixing though? It's good to know that we don't need to look at the removes first, but it's also not wrong. The visitor has separated them for us already, so it's not any extra cost to do them first. There was a problem hiding this comment. small nit for something I noticed below on line 98 (sorry, github won't let me comment there). You can remove a for remove in visitor.removes.into_iter() {
let dv_id = remove.dv_unique_id();
self.seen.insert((remove.path, dv_id));
}There was a problem hiding this comment. removed the clone :) and can open an issue if we want to modify the way we process removes? There was a problem hiding this comment.
It requires materializing (arbitrarily large) lists of adds and removes, which kernel shouldn't be in the business of doing in the first place. We should really be returning the original engine data along with updated selection vector, since the engine already went to the trouble of allocating all the actions for us there. There was a problem hiding this comment. Got it. This is a more general "fix the way scan works" kind of change. If I understand correctly, your model would switch scan to an iterator, and not pass around I'd say we merge this PR just to get arrow out, and then look at changing the return type of @ryan-johnson-databricks does that make sense to you? |
||
| let dv_id = remove.dv_unique_id(); | ||
| self.seen.insert((remove.path, dv_id)); | ||
| } | ||
|
|
||
| visitor | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
note i did just a little bit of reorganization just to list things in the order they are used.