optimise first and latest event fetch

src/storage/azure_blob.rs

@@ -698,6 +698,53 @@ impl ObjectStorage for BlobStore {
  Ok(streams)
  }
 
+ async fn list_hours(
+ &self,
+ stream_name: &str,
+ date: &str,
+ ) -> Result<Vec<String>, ObjectStorageError> {
+ let pre = object_store::path::Path::from(format!("{}/{}/", stream_name, date));
+ let resp = self.client.list_with_delimiter(Some(&pre)).await?;
+
+ let hours = resp
+ .common_prefixes
+ .iter()
+ .filter_map(|path| {
+ path.as_ref()
+ .strip_prefix(&format!("{}/{}/", stream_name, date))
+ .and_then(|s| s.strip_suffix('/'))
+ .map(String::from)
+ })
+ .filter(|dir| dir.starts_with("hour="))
+ .collect();
+
+ Ok(hours)
+ }
+
+ async fn list_minutes(
+ &self,
+ stream_name: &str,
+ date: &str,
+ hour: &str,
+ ) -> Result<Vec<String>, ObjectStorageError> {
+ let pre = object_store::path::Path::from(format!("{}/{}/{}/", stream_name, date, hour));
+ let resp = self.client.list_with_delimiter(Some(&pre)).await?;
+
+ let minutes = resp
+ .common_prefixes
+ .iter()
+ .filter_map(|path| {
+ path.as_ref()
+ .strip_prefix(&format!("{}/{}/{}/", stream_name, date, hour))
+ .and_then(|s| s.strip_suffix('/'))
+ .map(String::from)
+ })
+ .filter(|dir| dir.starts_with("minute="))
+ .collect();
+
+ Ok(minutes)
+ }
+
  async fn list_manifest_files(
  &self,
  stream_name: &str,

src/storage/gcs.rs

@@ -605,6 +605,53 @@ impl ObjectStorage for Gcs {
  Ok(streams)
  }
 
+ async fn list_hours(
+ &self,
+ stream_name: &str,
+ date: &str,
+ ) -> Result<Vec<String>, ObjectStorageError> {
+ let pre = object_store::path::Path::from(format!("{}/{}/", stream_name, date));
+ let resp = self.client.list_with_delimiter(Some(&pre)).await?;
+
+ let hours = resp
+ .common_prefixes
+ .iter()
+ .filter_map(|path| {
+ path.as_ref()
+ .strip_prefix(&format!("{}/{}/", stream_name, date))
+ .and_then(|s| s.strip_suffix('/'))
+ .map(String::from)
+ })
+ .filter(|dir| dir.starts_with("hour="))
+ .collect();
+
+ Ok(hours)
+ }
+
+ async fn list_minutes(
+ &self,
+ stream_name: &str,
+ date: &str,
+ hour: &str,
+ ) -> Result<Vec<String>, ObjectStorageError> {
+ let pre = object_store::path::Path::from(format!("{}/{}/{}/", stream_name, date, hour));
+ let resp = self.client.list_with_delimiter(Some(&pre)).await?;
+
+ let minutes = resp
+ .common_prefixes
+ .iter()
+ .filter_map(|path| {
+ path.as_ref()
+ .strip_prefix(&format!("{}/{}/{}/", stream_name, date, hour))
+ .and_then(|s| s.strip_suffix('/'))
+ .map(String::from)
+ })
+ .filter(|dir| dir.starts_with("minute="))
+ .collect();
+
+ Ok(minutes)
+ }
+
  async fn list_manifest_files(
  &self,
  stream_name: &str,

src/storage/localfs.rs

@@ -422,6 +422,43 @@ impl ObjectStorage for LocalFS {
  Ok(dates.into_iter().flatten().collect())
  }
 
+ async fn list_hours(
+ &self,
+ stream_name: &str,
+ date: &str,
+ ) -> Result<Vec<String>, ObjectStorageError> {
+ let path = self.root.join(stream_name).join(date);
+ let directories = ReadDirStream::new(fs::read_dir(&path).await?);
+ let entries: Vec<DirEntry> = directories.try_collect().await?;
+ let entries = entries.into_iter().map(dir_name);
+ let hours: Vec<_> = FuturesUnordered::from_iter(entries).try_collect().await?;
+ Ok(hours
+ .into_iter()
+ .flatten()
+ .filter(|dir| dir.starts_with("hour="))
+ .collect())
+ }
+
+ async fn list_minutes(
+ &self,
+ stream_name: &str,
+ date: &str,
+ hour: &str,
+ ) -> Result<Vec<String>, ObjectStorageError> {
+ let path = self.root.join(stream_name).join(date).join(hour);
+ // Propagate any read_dir errors instead of swallowing them
+ let directories = ReadDirStream::new(fs::read_dir(&path).await?);
+ let entries: Vec<DirEntry> = directories.try_collect().await?;
+ let entries = entries.into_iter().map(dir_name);
+ let minutes: Vec<_> = FuturesUnordered::from_iter(entries).try_collect().await?;
+ // Filter down to only the "minute=" prefixed directories
+ Ok(minutes
+ .into_iter()
+ .flatten()
+ .filter(|dir| dir.starts_with("minute="))
+ .collect())
+ }
+
  async fn list_manifest_files(
  &self,
  _stream_name: &str,