archive crate #8

tmcgroul · 2025-01-17T01:46:48Z

this crate supposed to be a replacement for already existing parquet writer - https://github.com/subsquid/archive.py/tree/master/sqa/writer. some things like logging and metrics are expected to be identical cause it would simplify transition from the old version

kalabukdima · 2025-01-20T12:06:50Z

crates/archive/src/archive.rs

+    let mut set = JoinSet::new();
+    set.spawn(async move {
+        sink.r#loop().await
+    });
+    set.spawn(async move {
+        writer.start().await
+    });
+    while let Some(res) = set.join_next().await {
+        res??;
+    }


Looks like a simple futures::join(sink.r#loop(), writer.start()).await

futures::join waits for both futures to complete. JoinSet supposed to catch an error from the writer task as early as possible.
we can easily imagine that smth happened during writing parquets and if we use join then the error will be propagated only when sink is trying to send a new value to already closed channel

If they both return Results, this one is even more convenient: https://docs.rs/tokio/latest/tokio/macro.try_join.html

kalabukdima · 2025-01-20T12:14:03Z

crates/archive/src/fs/local.rs

-        while let Some(entry) = read_dir.next_entry().await? {
-            if let Some(name) = entry.file_name().to_str() {
-                result.push(name.to_string());
+        match tokio::fs::read_dir(&self.root).await {


Can probably be simplified to let read_dir = read_dir(...).await.unwrap_or_default()

If clippy doesn't show it with the default config, you can try running cargo clippy -- -W clippy::pedantic — sometimes it gives some useful advices

here i can't really use unwrap_or_default() because Default trait isn't implemented for tokio::ReadDir. but yeah i will use recommendations from clippy

kalabukdima · 2025-01-20T12:17:12Z

crates/archive/src/fs/s3.rs

+            let dest_path = format!("{}/{}", dest, file_name);
+
+            if path.is_dir() {
+                Box::pin(self.upload_directory(&path, &dest_path)).await?;


Pinning on the stack is probably more efficient here: https://docs.rs/tokio/latest/tokio/macro.pin.html

recursion in an async fn requires boxing a recursive `async fn` call must introduce indirection such as `Box::pin` to avoid an infinitely sized future

seems like only Box::pin is available in this situation

🤔
Didn't know that, cool

kalabukdima · 2025-01-20T12:50:29Z

crates/archive/src/ingest.rs

+        let mut stream = res.bytes_stream();
+        let mut buf = bytes::BytesMut::new();
+        while let Some(bytes) = stream.try_next().await? {
+            buf.extend_from_slice(&bytes);
+            let pos = buf.iter().position(|b| *b == b'\n');
+            if let Some(pos) = pos {
+                let line = buf.split_to(pos).freeze();
+                buf.advance(1);
+                yield line;
+            }
+        }


It can probably be simplified with the StreamReader (less error-prone):

let res = client.post(url.as_str()).json(&range).send().await.map_err(|e| Error::new(ErrorKind::Other, e))?; let mut stream = res.bytes_stream().map(|result| result.map_err(|e| Error::new(ErrorKind::Other, e))); let mut reader = tokio_util::io::StreamReader::new(stream); loop { let mut buf = Vec::new(); reader.read_until(b'\n', &mut buf).await?; yield Bytes::from(buf); }

i remeber i was considering to use it but rejected the idea because it would require to map errors... let's try anyway

There is always an option to use anyhow::Result. This way in most situations you'll be able to just ?, and in worst case you'll have to do .map_err(Into::into)

kalabukdima · 2025-01-20T12:53:39Z

crates/archive/src/metrics.rs

+
+pub fn register_metrics(registry: &mut Registry) {
+    registry.register(
+        "sqd_progress_blocks_per_second",


Usually, it's both simpler and more flexible to expose raw counters (blocks processed since the start of the process) and then calculate useful functions like rate in the prometheus query. It will also remove ambiguity about what range the speed is averaged for.

metrics are identical to those that we have in current parquet writer and i would keep it like this so we can use existing charts and alerts

Then don't forget that this lib will add _total to the metric names of type Counter

kalabukdima · 2025-01-20T12:55:27Z

crates/archive/src/writer.rs

-                *spare = chunk.into_processor().ok()
-            }
-        });
+            metrics::LAST_SAVED_BLOCK.inc_by(last_block);


Should it be set?

Counter has no set method. i can switch to Gauge but i am not sure if won't affect charts 🤔

this way looks ugly but at least it should be correct

let val = last_block - metrics::LAST_SAVED_BLOCK.get(); metrics::LAST_SAVED_BLOCK.inc_by(val);

Yeah, it should probably be gauge if that's what you want. The way you proposed is not an atomic operation, so in case of concurrent access you can get a wrong value

i found a hack for that - prometheus/client_rust#252 (comment)

But why use a counter if it's always used as a gauge?

cause these metrics represent values that shouldn't go down

kalabukdima · 2025-01-20T13:05:54Z

crates/archive/src/sink.rs

+        if self.progress.has_news() {
+            self.report();
+        }


Can probably be moved inside Progress:

self.progress.maybe_report();

here progress only tracks numbers it doesn't print any information. so i can't add maybe_report to it

Yeah, I meant moving the logging and setting metrics to a new method in the Progress struct. But it doesn't really matter

kalabukdima · 2025-01-20T13:13:52Z

crates/archive/src/writer.rs

-        state.builder.chunk_builder().byte_size() + state.processor.byte_size()
-    }
+            let last_block = item.chunk.last_block;
+            let target_dir = tempfile::tempdir()?.into_path();


It's a bit strange that a generic FSRef is passed to the Writer, but it still uses local fs in this method

i believe it's parquet format limitation. we should write data to a local file first and then we can upload it to s3

kalabukdima · 2025-01-20T13:17:46Z

crates/archive/src/progress.rs

+        self.has_news
+    }
+
+    pub fn speed(&mut self) -> f64 {


Shouldn't it be enough to just report the total average? It would simplify things a lot

well... i just ported an algorithm that we already had in python version. i think this one is more accurate in comparison with the total average. it shows that current writing speed (due to data density or smth else) became faster or slower. the total average also would work

Is the python version located here? Let's add a link in the PR description

kalabukdima · 2025-01-20T13:18:49Z

crates/archive/src/sink.rs

+                        }
+                        if !data_ingested {
+                            tracing::info!("no blocks were found. waiting 5 min for a new try");
+                            tokio::time::sleep(Duration::from_secs(300)).await;


Shouldn't this delay be configurable? Sounds like 5 minutes may be a lot for some networks

eldargab · 2025-02-05T19:12:41Z

Good work, overall!

tmcgroul added 6 commits December 17, 2024 20:21

fix local fs

3ea6885

save work

0890fd8

add structured logging

b1f4040

add prometheus

ddc92a6

setup default logging env

e4c1d79

improve ingestion

4dca3b4

tmcgroul requested a review from eldargab January 17, 2025 01:46

kalabukdima reviewed Jan 20, 2025

View reviewed changes

tmcgroul added 4 commits January 20, 2025 22:31

refactoring

0dfd826

use tokio::try_join instead of JoinSet

51e1232

simplify metrics updates

4fdfa77

enable column statistics

09ba2bd

kalabukdima assigned tmcgroul Jan 28, 2025

tmcgroul added 3 commits January 28, 2025 21:06

add dictionary encoding

9bd9946

close writer channel

6ca09d9

disable default dictionary encoding

33d897c

eldargab merged commit 7c0ac7a into master Feb 5, 2025

eldargab deleted the dev branch February 5, 2025 19:12

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

archive crate #8

archive crate #8

tmcgroul commented Jan 17, 2025 •

edited

Loading

kalabukdima Jan 20, 2025

tmcgroul Jan 20, 2025

kalabukdima Jan 21, 2025

kalabukdima Jan 20, 2025

tmcgroul Jan 20, 2025

kalabukdima Jan 20, 2025

tmcgroul Jan 20, 2025

kalabukdima Jan 21, 2025

kalabukdima Jan 20, 2025

tmcgroul Jan 20, 2025

kalabukdima Jan 21, 2025

kalabukdima Jan 20, 2025

tmcgroul Jan 20, 2025

kalabukdima Jan 21, 2025

kalabukdima Jan 20, 2025

tmcgroul Jan 20, 2025

kalabukdima Jan 21, 2025

tmcgroul Jan 22, 2025

kalabukdima Jan 22, 2025

tmcgroul Jan 22, 2025

kalabukdima Jan 20, 2025

tmcgroul Jan 20, 2025

kalabukdima Jan 21, 2025

kalabukdima Jan 20, 2025

tmcgroul Jan 20, 2025

kalabukdima Jan 20, 2025

tmcgroul Jan 20, 2025

kalabukdima Jan 21, 2025

kalabukdima Jan 20, 2025

eldargab commented Feb 5, 2025

archive crate #8

archive crate #8

Conversation

tmcgroul commented Jan 17, 2025 • edited Loading

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

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

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

eldargab commented Feb 5, 2025

tmcgroul commented Jan 17, 2025 •

edited

Loading