cleanup README and use compressed CityLots

Author: timbray

README.md

@@ -124,17 +124,15 @@ but an AND on the field names.
 
 Note that the `shellstyle` Patterns can include only
 one `*` character. The architecture probably allows
-support for a larger subset of regular expressions
-but currently,  the testing for just the single -`*`
-patterns is a bit lacking.
+support for a larger subset of regular expressions, 
+eventually.
 
 Number matching is weak - the number has to appear 
 exactly the same in the pattern and the event. I.e.,
 Quamina doesn't know that 35, 35.000, and 3.5e1 are the
-same number.
-
-There's a fix for this in the code which is commented
-out because it causes a significant performance penalty.
+same number. There's a fix for this in the code which 
+is commented out because it causes a 
+significant performance penalty.
 
 ## Flattening and Matching
 
@@ -222,8 +220,9 @@ The `error` return is used to signal invalid Pattern
 structure, which could be bad UTF-8 or malformed JSON 
 or leaf values which are not provided as arrays.
 
-As many Patterns as desired can be added to a Matcher
-but at this time there is no capability of removing any.
+As many Patterns as desired can be added to a Matcher. 
+The `CoreMatcher` type does not support `DeletePattern()`
+but `pruner.Matcher` does.
 
 The `AddPattern` call is single-threaded; if multiple
 threads call it, they will block and execute sequentially.
@@ -242,15 +241,20 @@ The `[]X` return slice may be empty if none of the Patterns
 match the provided Event. 
 
 ```go
-func (m *Matcher) MatchesForFields([]Field) []X
+func (m *Matcher) MatchesForFields([]Field) ([]X, error)
 ```
 Performs the functions of `MatchesForJSON` on an 
 Event which has been flattened into a list of `Field`
-instances.
+instances.  At the moment, `CoreMatcher` only returns
+an error if the `[]Field` argument is nil. `pruner.Matcher`
+can return an error if it suffers a failure in its 
+Pattern storage.
 
-`MatchesForJSONEvent` is thread-safe. Many threads may
+These matching calls are thread-safe. Many threads may
 be executing it concurrently, even while `AddPattern` is
-also executing.
+also executing.  There is a significant performance 
+penalty if there is a high rate of `AddPattern` in
+combination with matching.
 
 ```go
 func NewFJ(*Matcher) Flattener
@@ -260,7 +264,7 @@ Creates a new JSON-specific Flattener.
 func (fj *FJ) Flatten([]byte event) []Field
 ```
 Transforms an event, which must be JSON object
-encoded in UTF-8 into a list of `Field` instances.
+encoded in UTF-8, into a list of `Field` instances.
 
 ```go
 func (fj *FJ) FlattenAndMatch([]byte event) ([]X, error)
@@ -305,7 +309,7 @@ colonies before slavery was abolished.
 
 ### Credits
 
-@timbray: v1.0 and patches.
+@timbray: v0.1 and patches.
 
 @jsmorph: `Pruner` and concurrency testing.
 

core/benchmarks_test.go

@@ -2,16 +2,50 @@ package core
 
 import (
 	"bufio"
+	"compress/gzip"
 	"fmt"
 	"os"
 	"runtime"
 	"runtime/debug"
+	"sync"
 	"testing"
 	"time"
 )
 
 const oneMeg = 1024 * 1024
 
+var cityLotsLock sync.Mutex
+var cityLotsLines [][]byte
+var cityLotsLineCount int
+
+func getCityLotsLines(t *testing.T) [][]byte {
+	cityLotsLock.Lock()
+	defer cityLotsLock.Unlock()
+	if cityLotsLines != nil {
+		return cityLotsLines
+	}
+	file, err := os.Open("../testdata/citylots.jlines.gz")
+	if err != nil {
+		t.Error("Can't open citlots.jlines.gz: " + err.Error())
+	}
+	defer func(file *os.File) {
+		_ = file.Close()
+	}(file)
+	zr, err := gzip.NewReader(file)
+	if err != nil {
+		t.Error("Can't open zip reader: " + err.Error())
+	}
+
+	scanner := bufio.NewScanner(zr)
+	buf := make([]byte, oneMeg)
+	scanner.Buffer(buf, oneMeg)
+	for scanner.Scan() {
+		cityLotsLineCount++
+		cityLotsLines = append(cityLotsLines, []byte(scanner.Text()))
+	}
+	return cityLotsLines
+}
+
 func TestCRANLEIGH(t *testing.T) {
 
 	jCranleigh := `{ "type": "Feature", "properties": { "MAPBLKLOT": "7222001", "BLKLOT": "7222001", "BLOCK_NUM": "7222", "LOT_NUM": "001", "FROM_ST": "1", "TO_ST": "1", "STREET": "CRANLEIGH", "ST_TYPE": "DR", "ODD_EVEN": "O" }, "geometry": { "type": "Polygon", "coordinates": [ [ [ -122.472773074480756, 37.73439178240811, 0.0 ], [ -122.47278111723567, 37.73451247621523, 0.0 ], [ -122.47242608711845, 37.73452184591072, 0.0 ], [ -122.472418368113281, 37.734401143064396, 0.0 ], [ -122.472773074480756, 37.73439178240811, 0.0 ] ] ] } }`
@@ -55,14 +89,6 @@ const thresholdPerformance = 1.0
 //  that it uses geometry/co-ordintes, which will force the fj flattener to process the big arrays of numbers in
 //  each line.  A high proportion of typical Quamina workloads should run faster.
 func TestCityLots(t *testing.T) {
-	file, err := os.Open("../testdata/citylots.jlines")
-	if err != nil {
-		t.Error("Can't open file: " + err.Error())
-	}
-	defer func(file *os.File) {
-		_ = file.Close()
-	}(file)
-
 	patterns := []string{
 		`{ "properties": { "STREET": [ "CRANLEIGH" ] } }`,
 		`{ "properties": { "STREET": [ "17TH" ], "ODD_EVEN": [ "E"] } }`,
@@ -82,10 +108,7 @@ func TestCityLots(t *testing.T) {
 		"0011008":   1,
 	}
 
-	scanner := bufio.NewScanner(file)
-	buf := make([]byte, oneMeg)
-	scanner.Buffer(buf, oneMeg)
-
+	var err error
 	m := NewCoreMatcher()
 	for i := range names {
 		err = m.AddPattern(names[i], patterns[i])
@@ -96,20 +119,13 @@ func TestCityLots(t *testing.T) {
 	fj := NewFJ(m)
 	results := make(map[X]int)
 
-	lineCount := 0
-	var lines [][]byte
-	for scanner.Scan() {
-		lineCount++
-		lines = append(lines, []byte(scanner.Text()))
-	}
-	lineCount = 0
+	lines := getCityLotsLines(t)
 	before := time.Now()
 	for _, line := range lines {
 		matches, err := fj.FlattenAndMatch(line)
 		if err != nil {
 			t.Error("Matches4JSON: " + err.Error())
 		}
-		lineCount++
 		for _, match := range matches {
 			count, ok := results[match]
 			if !ok {
@@ -121,7 +137,7 @@ func TestCityLots(t *testing.T) {
 	fmt.Println()
 
 	elapsed := float64(time.Since(before).Milliseconds())
-	perSecond := float64(lineCount) / (elapsed / 1000.0)
+	perSecond := float64(cityLotsLineCount) / (elapsed / 1000.0)
 	fmt.Printf("%.2f matches/second\n\n", perSecond)
 
 	if perSecond < thresholdPerformance {
@@ -135,11 +151,6 @@ func TestCityLots(t *testing.T) {
 		t.Errorf(message1 + message2)
 	}
 
-	err = scanner.Err()
-	if err != nil {
-		t.Error("Scanner error: " + err.Error())
-	}
-
 	if len(results) != len(wanted) {
 		t.Errorf("got %d results, wanted %d", len(results), len(wanted))
 	}
@@ -185,18 +196,7 @@ func TestMySoftwareHatesMe(t *testing.T) {
 
 // exercise shellstyle matching a little, is much faster than TestCityLots because it's only working wth one field
 func TestBigShellStyle(t *testing.T) {
-	file, err := os.Open("../testdata/citylots.jlines")
-	if err != nil {
-		t.Error("Can't open file: " + err.Error())
-	}
-	defer func(file *os.File) {
-		_ = file.Close()
-	}(file)
-
-	scanner := bufio.NewScanner(file)
-	buf := make([]byte, oneMeg)
-	scanner.Buffer(buf, oneMeg)
-
+	lines := getCityLotsLines(t)
 	m := NewCoreMatcher()
 
 	wanted := map[X]int{
@@ -231,12 +231,6 @@ func TestBigShellStyle(t *testing.T) {
 	*/
 	fmt.Println(matcherStats(m))
 
-	lineCount := 0
-	var lines [][]byte
-	for scanner.Scan() {
-		lineCount++
-		lines = append(lines, []byte(scanner.Text()))
-	}
 	lCounts := make(map[X]int)
 	before := time.Now()
 	fj := NewFJ(m)
@@ -256,7 +250,7 @@ func TestBigShellStyle(t *testing.T) {
 		}
 	}
 	elapsed := float64(time.Since(before).Milliseconds())
-	perSecond := float64(lineCount) / (elapsed / 1000.0)
+	perSecond := float64(cityLotsLineCount) / (elapsed / 1000.0)
 	fmt.Printf("%.2f matches/second with letter patterns\n\n", perSecond)
 
 	for k, wc := range wanted {

core/concurrency_test.go

@@ -1,10 +1,8 @@
 package core
 
 import (
-	"bufio"
 	"fmt"
 	"math/rand"
-	"os"
 	"testing"
 	"time"
 )
@@ -39,13 +37,7 @@ func TestConcurrency(t *testing.T) {
 	// I was initially surprised that adding 860 or so changes to the automaton while it's running doesn't seem to
 	//  cause any decrease in performance. But I guess it splits out very cleanly onto another core and really
 	//  doesn't steal any resources from the thread doing the Match calls
-	file, err := os.Open("../testdata/citylots.jlines")
-	if err != nil {
-		t.Error("Can't open file: " + err.Error())
-	}
-	defer func(file *os.File) {
-		_ = file.Close()
-	}(file)
+	lines := getCityLotsLines(t)
 
 	patterns := []string{
 		`{ "properties": { "STREET": [ "CRANLEIGH" ] } }`,
@@ -69,10 +61,7 @@ func TestConcurrency(t *testing.T) {
 		"0011008":    1,
 	}
 
-	scanner := bufio.NewScanner(file)
-	buf := make([]byte, oneMeg)
-	scanner.Buffer(buf, oneMeg)
-
+	var err error
 	m := NewCoreMatcher()
 	for i := range names {
 		err = m.AddPattern(names[i], patterns[i])
@@ -82,14 +71,8 @@ func TestConcurrency(t *testing.T) {
 	}
 	results := make(map[X]int)
 
-	lineCount := 0
-	var lines [][]byte
-	for scanner.Scan() {
-		lineCount++
-		lines = append(lines, []byte(scanner.Text()))
-	}
 	use37 := true
-	lineCount = 0
+	lineCount := 0
 	before := time.Now()
 	ch := make(chan string, 1000)
 	sent := 0
@@ -117,11 +100,6 @@ func TestConcurrency(t *testing.T) {
 	perSecond := float64(lineCount) / (elapsed / 1000.0)
 	fmt.Printf("\n%.2f matches/second with updates\n\n", perSecond)
 
-	err = scanner.Err()
-	if err != nil {
-		t.Error("Scanner error: " + err.Error())
-	}
-
 	if len(results) != len(wanted) {
 		t.Errorf("got %d results, wanted %d", len(results), len(wanted))
 	}

core/core_matcher.go

@@ -33,7 +33,7 @@ type coreStart struct {
 }
 
 // X for anything, should eventually be a generic?
-type X interface{}
+type X any
 
 func NewCoreMatcher() *CoreMatcher {
 	m := CoreMatcher{}

core/fj.go

@@ -28,7 +28,7 @@ type FJ struct {
 }
 
 // Reset an FJ struct so it can be re-used and won't need to be reconstructed for each event to be flattened
-func (fj *FJ) Reset() {
+func (fj *FJ) reset() {
 	fj.eventIndex = 0
 	fj.fields = fj.fields[:0]
 	fj.skipping = 0
@@ -80,7 +80,7 @@ func (fj *FJ) Flatten(event []byte) ([]Field, error) {
 	if fj.cleanSheet {
 		fj.cleanSheet = false
 	} else {
-		fj.Reset()
+		fj.reset()
 	}
 	if len(event) == 0 {
 		return nil, fj.error("empty event")
@@ -471,7 +471,7 @@ func (fj *FJ) readStringValue() ([]byte, error) {
 		} else if ch == '\\' {
 			val, err := fj.readStringValWithEscapes(valStart)
 			return val, err
-		} else if ch <= 0x1f || ch >= byte(ByteCeiling) {
+		} else if ch <= 0x1f || ch >= byte(byteCeiling) {
 			return nil, fj.error(fmt.Sprintf("illegal UTF-8 byte %x in string value", ch))
 		}
 		if fj.step() != nil {
@@ -501,7 +501,7 @@ func (fj *FJ) readStringValWithEscapes(nameStart int) ([]byte, error) {
 				return nil, err
 			}
 			val = append(val, unescaped...)
-		} else if ch <= 0x1f || ch >= byte(ByteCeiling) {
+		} else if ch <= 0x1f || ch >= byte(byteCeiling) {
 			return nil, fj.error(fmt.Sprintf("illegal UTF-8 byte %x in string value", ch))
 		} else {
 			val = append(val, ch)
@@ -529,7 +529,7 @@ func (fj *FJ) readMemberName() ([]byte, error) {
 		} else if ch == '\\' {
 			name, err := fj.readMemberNameWithEscapes(nameStart)
 			return name, err
-		} else if ch <= 0x1f || ch >= byte(ByteCeiling) {
+		} else if ch <= 0x1f || ch >= byte(byteCeiling) {
 			return nil, fj.error(fmt.Sprintf("illegal UTF-8 byte %x in field name", ch))
 		}
 		if fj.step() != nil {
@@ -547,7 +547,7 @@ func (fj *FJ) readMemberNameWithEscapes(nameStart int) ([]byte, error) {
 		if ch == '"' {
 			fj.eventIndex = from
 			return memberName, nil
-		} else if ch <= 0x1f || ch >= byte(ByteCeiling) {
+		} else if ch <= 0x1f || ch >= byte(byteCeiling) {
 			return nil, fj.error(fmt.Sprintf("illegal UTF-8 byte %x in field name", ch))
 		} else if ch == '\\' {
 			var unescaped []byte

core/flattener.go

@@ -35,5 +35,5 @@ type Field struct {
 type Flattener interface {
 	Flatten(event []byte) ([]Field, error)
 	FlattenAndMatch(event []byte) ([]X, error)
-	Reset()
+	reset()
 }