fix filter_agg

dig/dig.go

@@ -961,34 +961,29 @@ func (lwc *logWithCtx) get(name string) any {
 
 type filterResults struct {
 	kind string
-	used bool
+	set  bool
 	val  bool
 }
 
 func (fr *filterResults) add(b bool) {
-	fr.used = true
+	if !fr.set {
+		fr.set = true
+		fr.val = b
+		return
+	}
 	switch fr.kind {
 	case "and":
-		if !b {
-			fr.val = false
-		}
+		fr.val = fr.val && b
 	default:
-		if b {
-			fr.val = true
-		}
+		fr.val = fr.val || b
 	}
 }
 
 func (fr *filterResults) accept() bool {
-	if !fr.used {
+	if !fr.set {
 		return true
 	}
-	switch fr.kind {
-	case "and":
-		return !fr.val
-	default:
-		return fr.val
-	}
+	return fr.val
 }
 
 func (ig Integration) processTx(rows [][]any, lwc *logWithCtx, pgmut *sync.Mutex, pg wpg.Conn) ([][]any, bool, error) {

dig/dig_test.go

@@ -463,3 +463,25 @@ func TestFilter(t *testing.T) {
 		tc.WantGot(t, c.want, frs.accept())
 	}
 }
+
+func TestFilterResults(t *testing.T) {
+	cases := []struct {
+		kind  string
+		want  bool
+		input []bool
+	}{
+		{"and", false, []bool{true, false}},
+		{"and", false, []bool{false, false}},
+		{"and", true, []bool{true, true}},
+		{"or", true, []bool{true, false}},
+		{"or", true, []bool{true, true}},
+		{"or", false, []bool{false, false}},
+	}
+	for _, c := range cases {
+		frs := filterResults{kind: c.kind}
+		for _, b := range c.input {
+			frs.add(b)
+		}
+		tc.WantGot(t, c.want, frs.accept())
+	}
+}