feat: scale and compose probabilistic stream tools

cmd/bf/main_test.go

@@ -0,0 +1,161 @@
+package main
+
+import (
+	"bytes"
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"testing"
+
+	internalbf "github.com/jo-cube/toolbox/internal/bf"
+)
+
+func TestNULTestPreservesDelimiter(t *testing.T) {
+	t.Parallel()
+
+	f := newFilter(t, "aa", "bb")
+	path := writeFilter(t, "filter.bf", f)
+	var out bytes.Buffer
+	if err := test([]string{"-0", path}, bytes.NewReader([]byte("aa\x00cc\x00bb\x00")), &out); err != nil {
+		t.Fatal(err)
+	}
+	if want := []byte("aa\x00bb\x00"); !bytes.Equal(out.Bytes(), want) {
+		t.Fatalf("test output = %q, want %q", out.Bytes(), want)
+	}
+}
+
+func TestDedupeEmitsFirstProbablyUnseenItem(t *testing.T) {
+	t.Parallel()
+
+	var out, errOut bytes.Buffer
+	err := dedupe([]string{"--expected-items", "100", "--false-positive-rate", "0.000001"}, bytes.NewBufferString("a\nb\na\n"), &out, &errOut)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if out.String() != "a\nb\n" {
+		t.Fatalf("dedupe output = %q", out.String())
+	}
+	if errOut.Len() != 0 {
+		t.Fatalf("dedupe stderr = %q", errOut.String())
+	}
+}
+
+func TestFieldSelectionBuildsTestsAndDeduplicatesByField(t *testing.T) {
+	t.Parallel()
+
+	var built, errOut bytes.Buffer
+	buildInput := "1\talice\tcreated\n2\tbob\tcreated\n"
+	if err := build([]string{"--expected-items", "100", "--false-positive-rate", "0.000001", "-d", "\t", "-f", "2"}, bytes.NewBufferString(buildInput), &built, &errOut); err != nil {
+		t.Fatal(err)
+	}
+	f, err := internalbf.Read(bytes.NewReader(built.Bytes()))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !f.Test([]byte("alice")) || !f.Test([]byte("bob")) || f.Test([]byte("1\talice\tcreated")) {
+		t.Fatal("build did not hash only the selected field")
+	}
+
+	path := writeFilter(t, "fields.bf", f)
+	var tested bytes.Buffer
+	testInput := "3\t alice \tlogin\n4\tcarol\tlogin\n5\tbob\tlogout\n"
+	if err := test([]string{"--trim", "--delimiter", "\t", "--field", "2", path}, bytes.NewBufferString(testInput), &tested); err != nil {
+		t.Fatal(err)
+	}
+	if want := "3\t alice \tlogin\n5\tbob\tlogout\n"; tested.String() != want {
+		t.Fatalf("test output = %q, want %q", tested.String(), want)
+	}
+
+	var deduped bytes.Buffer
+	dedupeInput := "1\ta\n2\tb\n3\ta\n"
+	err = dedupe([]string{"--expected-items", "100", "--false-positive-rate", "0.000001", "-d", "\t", "-f", "2"}, bytes.NewBufferString(dedupeInput), &deduped, &errOut)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if want := "1\ta\n2\tb\n"; deduped.String() != want {
+		t.Fatalf("dedupe output = %q, want %q", deduped.String(), want)
+	}
+}
+
+func TestBuildWarnsWhenExpectedItemsAreExceeded(t *testing.T) {
+	t.Parallel()
+
+	var out, errOut bytes.Buffer
+	err := build([]string{"--expected-items", "1", "--false-positive-rate", "0.01"}, bytes.NewBufferString("a\nb\n"), &out, &errOut)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Contains(errOut.Bytes(), []byte("inserted items (2) exceed expected items (1)")) {
+		t.Fatalf("build stderr = %q", errOut.String())
+	}
+	if _, err := internalbf.Read(&out); err != nil {
+		t.Fatalf("build output is not a filter: %v", err)
+	}
+}
+
+func TestInspectReadsFilterFromStdin(t *testing.T) {
+	t.Parallel()
+
+	f := newFilter(t, "alpha")
+	var serialized, out bytes.Buffer
+	if err := internalbf.Write(&serialized, f); err != nil {
+		t.Fatal(err)
+	}
+	if err := inspect([]string{"--json", "-"}, &serialized, &out); err != nil {
+		t.Fatal(err)
+	}
+	var metadata internalbf.Metadata
+	if err := json.Unmarshal(out.Bytes(), &metadata); err != nil {
+		t.Fatal(err)
+	}
+	if metadata.SetBits == 0 || metadata.BitsetBytes != uint64(len(f.Bits)) {
+		t.Fatalf("metadata = %#v", metadata)
+	}
+}
+
+func TestUnionReadsOneFilterFromStdin(t *testing.T) {
+	t.Parallel()
+
+	a := newFilter(t, "alpha")
+	b := newFilter(t, "beta")
+	path := writeFilter(t, "a.bf", a)
+	var serializedB, combined bytes.Buffer
+	if err := internalbf.Write(&serializedB, b); err != nil {
+		t.Fatal(err)
+	}
+	if err := union([]string{path, "-"}, &serializedB, &combined); err != nil {
+		t.Fatal(err)
+	}
+	got, err := internalbf.Read(&combined)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !got.Test([]byte("alpha")) || !got.Test([]byte("beta")) {
+		t.Fatal("union lost an inserted item")
+	}
+}
+
+func newFilter(t *testing.T, items ...string) *internalbf.Filter {
+	t.Helper()
+	f, err := internalbf.New(100, 0.000001)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, item := range items {
+		f.Add([]byte(item))
+	}
+	return f
+}
+
+func writeFilter(t *testing.T, name string, f *internalbf.Filter) string {
+	t.Helper()
+	var data bytes.Buffer
+	if err := internalbf.Write(&data, f); err != nil {
+		t.Fatal(err)
+	}
+	path := filepath.Join(t.TempDir(), name)
+	if err := os.WriteFile(path, data.Bytes(), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	return path
+}

cmd/card/main.go

@@ -37,9 +37,10 @@ Examples:
   card --json .user_id .event_type events.jsonl
 
 Notes:
-  CSV mode selects header names.
-  Delimited mode selects 1-based field numbers.
-  JSON paths are simple dot paths; filters and array traversal are not supported.
+	  CSV mode selects header names.
+	  Delimited mode selects 1-based field numbers.
+	  JSON paths are simple dot paths; filters and array traversal are not supported.
+	  Prefix relative JSON input files with ./ so they are not parsed as selectors.
 
 Options:
 `, name)
@@ -80,7 +81,7 @@ Options:
 	var paths []string
 	var jsonPaths []string
 	if mode == "json" {
-		for len(args) > 0 && strings.HasPrefix(args[0], ".") {
+		for len(args) > 0 && isJSONSelector(args[0]) {
 			jsonPaths = append(jsonPaths, args[0])
 			args = args[1:]
 		}
@@ -132,6 +133,10 @@ Options:
 	}
 }
 
+func isJSONSelector(arg string) bool {
+	return strings.HasPrefix(arg, ".") && !strings.HasPrefix(arg, "./") && !strings.HasPrefix(arg, "../")
+}
+
 func splitList(raw string) []string {
 	if raw == "" {
 		return nil

cmd/card/main_test.go

@@ -0,0 +1,16 @@
+package main
+
+import "testing"
+
+func TestJSONSelectorDoesNotConsumeRelativePaths(t *testing.T) {
+	t.Parallel()
+
+	if !isJSONSelector(".user.id") {
+		t.Fatal("JSON selector was rejected")
+	}
+	for _, path := range []string{"./events.jsonl", "../events.jsonl"} {
+		if isJSONSelector(path) {
+			t.Fatalf("relative path %q was treated as a JSON selector", path)
+		}
+	}
+}

cmd/heavy/main.go

@@ -84,17 +84,17 @@ Options:
 	}
 
 	if *tsvOut {
-		fmt.Fprintln(os.Stdout, "rank\tcount_estimate\titem")
+		fmt.Fprintln(os.Stdout, "rank\tcount_estimate\tcount_lower_bound\titem")
 		for _, result := range results {
-			fmt.Fprintf(os.Stdout, "%d\t%d\t%s\n", result.Rank, result.CountEstimate, result.Item)
+			fmt.Fprintf(os.Stdout, "%d\t%d\t%d\t%s\n", result.Rank, result.CountEstimate, result.CountLowerBound, result.Item)
 		}
 		return
 	}
 
 	w := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0)
-	fmt.Fprintln(w, "rank\tcount_estimate\titem")
+	fmt.Fprintln(w, "rank\tcount_estimate\tcount_lower_bound\titem")
 	for _, result := range results {
-		fmt.Fprintf(w, "%d\t%d\t%s\n", result.Rank, result.CountEstimate, result.Item)
+		fmt.Fprintf(w, "%d\t%d\t%d\t%s\n", result.Rank, result.CountEstimate, result.CountLowerBound, result.Item)
 	}
 	if err := w.Flush(); err != nil {
 		fmt.Fprintf(os.Stderr, "heavy: %v\n", err)

cmd/hll/main.go

@@ -192,6 +192,9 @@ Sketches must use compatible precision, register count, version, and hash metada
 	if fs.NArg() < 2 {
 		return fmt.Errorf("usage: hll merge <file> <file>...")
 	}
+	if countStdin(fs.Args()) > 1 {
+		return fmt.Errorf("usage: hll merge accepts stdin only once")
+	}
 
 	merged, err := readSketch(fs.Arg(0))
 	if err != nil {
@@ -247,6 +250,9 @@ Options:
 }
 
 func readSketch(path string) (*hll.Sketch, error) {
+	if path == "-" {
+		return hll.Read(os.Stdin)
+	}
 	f, err := os.Open(path)
 	if err != nil {
 		return nil, fmt.Errorf("open %s: %w", path, err)
@@ -255,6 +261,16 @@ func readSketch(path string) (*hll.Sketch, error) {
 	return hll.Read(f)
 }
 
+func countStdin(paths []string) int {
+	count := 0
+	for _, path := range paths {
+		if path == "-" {
+			count++
+		}
+	}
+	return count
+}
+
 func writeEstimate(out *os.File, s *hll.Sketch, jsonOut bool) error {
 	m := s.Metadata()
 	if jsonOut {