Improve stream #15

maximedion2 · 2024-04-05T04:57:29Z

No description provided.

maximedion2 · 2024-04-05T04:58:03Z

src/async_reader/zarr_read_async.rs

-            let s = s.join(".");
-
-            let p = self.location.child(var.to_string()).child(s);
+            let separator = separators


This is an actual bug fix, that needs to go in, unrelated to the async stream work.

maximedion2 · 2024-04-05T05:01:00Z

src/datafusion/scanner.rs

@@ -185,7 +185,10 @@ mod tests {
        let schema = first_batch.schema();

        let names = schema.fields().iter().map(|f| f.name()).collect::<Vec<_>>();
-        assert_eq!(names, vec!["lon", "lat"]);
+        let target_names = ["lon", "lat", "float_data"];


So, in the case or zarr stores, there is no clear order for the "columns", unlike e.g. a parquet file. So while this test as is might have run fine when you tried it, when I did it was picking up "lat" and "float_data", I think it might actually be non-deterministic? Do you think that this could potentially be a problem? When multiple record batches are combined, we could maybe run into issues? If that's the case I could enforce that columns are read in alphabetical order, that would fix it.

Sorry about that, but yeah, would be good to give it some order or way to make it deterministic between schema inference and scanning.

no worries, we found an issue with a test, that's part of the process! I'll add a fix to this PR, probably tomorrow, and I'll update some tests to explicitly check for column order in the output.

maximedion2 · 2024-04-07T05:32:13Z

src/reader/mod.rs

@@ -192,12 +192,8 @@ impl<T: ZarrIterator> ZarrRecordBatchReader<T> {
        let mut arrs: Vec<ArrayRef> = Vec::with_capacity(self.meta.get_num_columns());
        let mut fields: Vec<FieldRef> = Vec::with_capacity(self.meta.get_num_columns());

-        // the sort below is important, because within a zarr store, the different arrays are


Turns out this is a proper bug, because I already thought about this before (and then I forgot about it), but the way I tried to enforce ordering wasn't quite right. It should be fixed now.

maximedion2 · 2024-04-07T05:35:51Z

src/datafusion/scanner.rs

@@ -185,7 +185,7 @@ mod tests {
        let schema = first_batch.schema();

        let names = schema.fields().iter().map(|f| f.name()).collect::<Vec<_>>();
-        assert_eq!(names, vec!["lon", "lat"]);
+        assert_eq!(names, vec!["lat", "lon"]);


I undid my change, restored the test to what it was before except that now it's checking for the last 2 columns in alphabetical order (so the columns are "float_data", "lat", "lon", and it's now picking up the last 2, always).

tshauck

Generally LGTM, just left a few nits, but looks like a good place to stop before adding the filter pushdown.

tshauck · 2024-04-07T17:20:11Z

src/reader/metadata.rs

@@ -675,6 +675,10 @@ impl ZarrStoreMetadata {
            }
        }

+        // the sort below is important, because within a zarr store, the different arrays are
+        // not ordered, so there is no predefined order for the different columns. we effectively
+        // define one here, my ordering the columns alphabetically.


Suggested change

// define one here, my ordering the columns alphabetically.

// define one here, by ordering the columns alphabetically.

tshauck · 2024-04-07T17:21:09Z

src/reader/zarr_read.rs

+
+        // the sort below is important, because within a zarr store, the different arrays are
+        // not ordered, so there is no predefined order for the different columns. we effectively
+        // define one here, my ordering the columns alphabetically.


Suggested change

// define one here, my ordering the columns alphabetically.

// define one here, by ordering the columns alphabetically.

tshauck · 2024-04-07T17:22:04Z

src/bin/async-benchmark.rs

@@ -0,0 +1,28 @@
+use arrow_zarr::async_reader::{ZarrPath, ZarrRecordBatchStreamBuilderNonBlocking};


nit, but missing copyright. DataFusion at least has a checker as part of its CI, which may make sense to use here.

Added a copyright here and in several other files.

tshauck · 2024-04-07T17:26:59Z

src/async_reader/mod.rs

+    T: ZarrStream + Unpin + Send + 'static,
+{
+    type Item = ZarrResult<RecordBatch>;
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {


This function has a few .expects and .unwraps. I get some aren't possible, but in those cases perhaps we can follow something more idiomatic?

fixed some of those .unwraps, I'll keep cleaning up the code as I make more changes.

tshauck · 2024-04-07T17:28:25Z

src/async_reader/mod.rs

+                    self.store_wrapper = Some(wrapper);
+
+                    // if store returns none, it's the end and it's time to return
+                    if chunk.is_none() {


e.g.

let chunk = if let Some(chunk) = chunk { chunk // or maybe do more here } else { self.state = ZarrStreamStateNonBlocking::Done; return Poll::Ready(None); }

yeah you're right, I've been following a few patterns that are not really rust-idiomatic, I've been learning rust on-and-off for the last 2 years or so and I sometimes get sloppy, like these .is_none followed by an unwrap call. I'll do a quick check tonight and clean it up a bit.

tshauck · 2024-04-07T17:41:05Z

.gitignore

+target
+.vscode


FWIW, you can make a ~/.gitignore and put your editor specific stuff there, then you won't need to worry about adding these things to individual repos.

https://gist.github.com/subfuzion/db7f57fff2fb6998a16c

sounds good, done.

maximedion2 added 2 commits April 3, 2024 23:21

first draft for a non blocking async stream

6080dbd

fixed a bug and a test

3efdff8

maximedion2 requested a review from tshauck April 5, 2024 04:57

maximedion2 commented Apr 5, 2024

View reviewed changes

fixed an issue with colummn ordering in zarr chunks

c5988ed

maximedion2 commented Apr 7, 2024

View reviewed changes

linter fix

666f1ed

tshauck approved these changes Apr 7, 2024

View reviewed changes

tshauck reviewed Apr 7, 2024

View reviewed changes

addressing PR comments

869bc74

maximedion2 merged commit 43a5edb into main Apr 8, 2024
3 checks passed

maximedion2 linked an issue May 2, 2024 that may be closed by this pull request

Improve async reader stream implementation #6

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Improve stream #15

Improve stream #15

maximedion2 commented Apr 5, 2024

maximedion2 Apr 5, 2024

maximedion2 Apr 5, 2024

tshauck Apr 5, 2024

maximedion2 Apr 5, 2024

maximedion2 Apr 7, 2024 •

edited

Loading

maximedion2 Apr 7, 2024

tshauck left a comment

tshauck Apr 7, 2024

tshauck Apr 7, 2024

tshauck Apr 7, 2024

maximedion2 Apr 8, 2024

tshauck Apr 7, 2024

maximedion2 Apr 8, 2024

tshauck Apr 7, 2024

maximedion2 Apr 7, 2024

tshauck Apr 7, 2024

maximedion2 Apr 8, 2024

	// define one here, my ordering the columns alphabetically.
	// define one here, by ordering the columns alphabetically.

		@@ -0,0 +1,28 @@
		use arrow_zarr::async_reader::{ZarrPath, ZarrRecordBatchStreamBuilderNonBlocking};

		target
		.vscode

Improve stream #15

Improve stream #15

Conversation

maximedion2 commented Apr 5, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

maximedion2 Apr 7, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tshauck left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

maximedion2 Apr 7, 2024 •

edited

Loading