feat: enabled sorting in compressed files

Cargo.toml

@@ -4,7 +4,7 @@ members = [".", "crate/encstr"]
 [workspace.package]
 repository = "https://github.com/pamburus/hl"
 authors = ["Pavel Ivanov <[email protected]>"]
-version = "0.30.0-alpha.3"
+version = "0.30.0-alpha.4"
 edition = "2021"
 license = "MIT"
 
@@ -43,27 +43,29 @@ crossbeam-channel = "0"
 crossbeam-queue = "0"
 crossbeam-utils = "0"
 deko = "0"
-derive_deref = "1"
+derive_more = {version="1", features = ["deref", "from"]}
 dirs = "5"
 dirs-sys = "0"
 encstr = { path = "./crate/encstr" }
 enum-map = "2"
 enumset = "1"
 enumset-ext = { path = "./crate/enumset-ext" }
+env_logger = "0"
 flate2 = "1"
 heapless = "0"
 hex = "0"
 htp = { git = "https://github.com/pamburus/htp.git" }
 humantime = "2"
 itertools = "0.13"
 itoa = { version = "1", default-features = false }
+log = "0"
+nonzero_ext = "0"
 notify = { version = "7", features = ["macos_kqueue"] }
 nu-ansi-term = "0"
 num_cpus = "1"
 once_cell = "1"
 pest = "2"
 pest_derive = "2"
-phf = { version = "0", features = ["macros"] }
 regex = "1"
 rust-embed = "8"
 serde = { version = "1", features = ["derive"] }
@@ -88,8 +90,10 @@ kqueue = "1"
 
 [dev-dependencies]
 byte-strings = "0"
+clean-path = "0"
 criterion = "0"
 maplit = "1"
+mockall = "0"
 stats_alloc = "0"
 regex = "1"
 wildmatch = "2"

build/ci/coverage.sh

@@ -41,18 +41,15 @@ function clean() {
 function test() {
     cargo test --tests --workspace
     cargo build
-    ${MAIN_EXECUTABLE:?} > /dev/null
+    ${MAIN_EXECUTABLE:?} --config= > /dev/null
     ${MAIN_EXECUTABLE:?} --config= --help > /dev/null
     ${MAIN_EXECUTABLE:?} --config=etc/defaults/config-k8s.yaml > /dev/null
     ${MAIN_EXECUTABLE:?} --config=etc/defaults/config-ecs.yaml > /dev/null
-    ${MAIN_EXECUTABLE:?} --shell-completions bash > /dev/null
-    ${MAIN_EXECUTABLE:?} --man-page > /dev/null
-    ${MAIN_EXECUTABLE:?} --list-themes > /dev/null
-    echo "" | ${MAIN_EXECUTABLE:?} --concurrency 4 > /dev/null
-    if ${MAIN_EXECUTABLE:?} -s test.log.gz 2>/dev/null > /dev/null; then
-        echo "Expected combination of options `-s 1234.log.gz` to fail"
-        exit 1
-    fi
+    ${MAIN_EXECUTABLE:?} --config= --shell-completions bash > /dev/null
+    ${MAIN_EXECUTABLE:?} --config= --man-page > /dev/null
+    ${MAIN_EXECUTABLE:?} --config= --list-themes > /dev/null
+    ${MAIN_EXECUTABLE:?} --config= sample/prometheus.log -P > /dev/null
+    echo "" | ${MAIN_EXECUTABLE:?} --config= --concurrency 4 > /dev/null
 } 
 
 function merge() {

sample/test.log

@@ -0,0 +1,2 @@
+{"ts":"2024-10-01T01:02:04Z","level":"info","msg":"latest test message"}
+{"ts":"2024-10-01T01:02:03Z","level":"info","msg":"earlier test message"}

src/app.rs

@@ -40,6 +40,7 @@
     settings::{FieldShowOption, Fields, Formatting},
     theme::{Element, StylingPush, Theme},
     timezone::Tz,
+    vfs::LocalFileSystem,
     IncludeExcludeKeyFilter,
     {error::*, QueryNone},
 };
@@ -219,7 +220,7 @@
                 let mut tx = StripedSender::new(txi);
                 let scanner = Scanner::new(sfi, &self.options.delimiter);
                 for (i, mut input) in inputs.into_iter().enumerate() {
-                    for item in scanner.items(&mut input.stream).with_max_segment_size(self.options.max_message_size.into()) {
+                    for item in scanner.items(&mut input.stream.as_sequential()).with_max_segment_size(self.options.max_message_size.into()) {
                         if tx.send((i, item?)).is_none() {
                             break;
                         }
@@ -272,6 +273,7 @@
     fn sort(&self, inputs: Vec<InputHolder>, output: &mut Output) -> Result<()> {
         let mut output = BufWriter::new(output);
         let indexer_settings = IndexerSettings::new(
+            LocalFileSystem,
             self.options.buffer_size.try_into()?,
             self.options.max_message_size.try_into()?,
             &self.options.fields.settings.predefined,
@@ -499,14 +501,14 @@
                 let reader = scope.spawn(closure!(clone sfi, clone txi, |_| -> Result<()> {
                     let scanner = Scanner::new(sfi.clone(), &self.options.delimiter);
                     let mut meta = None;
-                    if let InputReference::File(filename) = &input_ref {
-                        meta = Some(fs::metadata(filename)?);
+                    if let InputReference::File(path) = &input_ref {
+                        meta = Some(fs::metadata(&path.canonical)?);
                     }
-                    let mut input = Some(input_ref.open_tail(self.options.tail)?);
+                    let mut input = Some(input_ref.open()?.tail(self.options.tail)?);
                     let is_file = |meta: &Option<fs::Metadata>| meta.as_ref().map(|m|m.is_file()).unwrap_or(false);
                     let process = |input: &mut Option<Input>, is_file: bool| {
                         if let Some(input) = input {
-                            for (j, item) in scanner.items(&mut input.stream).with_max_segment_size(self.options.max_message_size.into()).enumerate() {
+                            for (j, item) in scanner.items(&mut input.stream.as_sequential()).with_max_segment_size(self.options.max_message_size.into()).enumerate() {
                                 if txi.send((i, j, item?)).is_err() {
                                     break;
                                 }
@@ -516,14 +518,14 @@
                             Ok(false)
                         }
                     };
-                    if let InputReference::File(filename) = &input_ref {
+                    if let InputReference::File(path) = &input_ref {
                         if process(&mut input, is_file(&meta))? {
                             return Ok(())
                         }
-                        fsmon::run(vec![filename.clone()], |event| {
+                        fsmon::run(vec![path.canonical.clone()], |event| {
                             match event.kind {
                                 EventKind::Modify(_) | EventKind::Create(_) | EventKind::Any | EventKind::Other => {
-                                    if let (Some(old_meta), Ok(new_meta)) = (&meta, fs::metadata(&filename)) {
+                                    if let (Some(old_meta), Ok(new_meta)) = (&meta, fs::metadata(&path.canonical)) {
                                         if old_meta.len() > new_meta.len() {
                                             input = None;
                                         }
@@ -691,7 +693,7 @@
     fn input_badges<'a, I: IntoIterator<Item = &'a InputReference>>(&self, inputs: I) -> Option<Vec<String>> {
         let name = |input: &InputReference| match input {
             InputReference::Stdin => "<stdin>".to_owned(),
-            InputReference::File(path) => path.to_string_lossy().to_string(),
+            InputReference::File(path) => path.original.to_string_lossy().to_string(),
         };
 
         let mut badges = inputs.into_iter().map(|x| name(x).chars().collect_vec()).collect_vec();

src/error.rs

@@ -102,8 +102,6 @@ pub enum Error {
     },
     #[error("invalid index header")]
     InvalidIndexHeader,
-    #[error("requested sorting of messages in {}-compressed file '{}' that is not currently supported", HILITE.paint(.format), HILITE.paint(.path.to_string_lossy()))]
-    UnsupportedFormatForIndexing { path: PathBuf, format: String },
     #[error("failed to parse json: {0}")]
     JsonParseError(#[from] serde_json::Error),
     #[error("failed to parse logfmt: {0}")]