Deduplicate filter projection with result projection (#668)

This was missing in #651
spiraldb · Aug 21, 2024 · a604fa8 · a604fa8
1 parent 1f6e81c
commit a604fa8
Show file tree

Hide file tree

Showing 3 changed files with 20 additions and 2 deletions.
diff --git a/vortex-serde/src/layouts/read/builder.rs b/vortex-serde/src/layouts/read/builder.rs
@@ -1,3 +1,4 @@
+use std::collections::HashSet;
 use std::sync::{Arc, RwLock};
 
 use bytes::BytesMut;
@@ -84,7 +85,12 @@ impl<R: VortexReadAt> LayoutReaderBuilder<R> {
                 Projection::All => (Projection::All, Projection::All),
                 Projection::Flat(mut v) => {
                     let original_len = v.len();
-                    v.extend(filter_columns.into_iter());
+                    let existing_fields: HashSet<Field> = v.iter().cloned().collect();
+                    v.extend(
+                        filter_columns
+                            .into_iter()
+                            .filter(|f| !existing_fields.contains(f)),
+                    );
                     (
                         Projection::Flat(v),
                         Projection::Flat((0..original_len).map(Field::from).collect()),

diff --git a/vortex-serde/src/layouts/read/layouts.rs b/vortex-serde/src/layouts/read/layouts.rs
@@ -61,7 +61,11 @@ impl Layout for FlatLayout {
             }
             FlatLayoutState::ReadBatch => {
                 let mut buf = self.cache.remove(&[]).ok_or_else(|| {
-                    vortex_err!("Wrong state transition, message should have been fetched")
+                    vortex_err!(
+                        "Wrong state transition, message {:?} (with range {}) should have been fetched",
+                        self.cache.absolute_id(&[]),
+                        self.range
+                    )
                 })?;
 
                 let mut array_reader = ArrayBufferReader::new();

diff --git a/vortex-serde/src/stream_writer/mod.rs b/vortex-serde/src/stream_writer/mod.rs
@@ -1,3 +1,5 @@
+use std::fmt::{Display, Formatter};
+
 use futures_util::{Stream, TryStreamExt};
 use vortex::array::ChunkedArray;
 use vortex::stream::ArrayStream;
@@ -98,6 +100,12 @@ pub struct ByteRange {
     pub end: u64,
 }
 
+impl Display for ByteRange {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "[{}, {})", self.begin, self.end)
+    }
+}
+
 #[allow(clippy::len_without_is_empty)]
 impl ByteRange {
     pub fn new(begin: u64, end: u64) -> Self {