From 0276c13ceab3f104bcc810f25e56724642a105a7 Mon Sep 17 00:00:00 2001 From: Ben Johnson Date: Fri, 28 Apr 2017 12:59:39 -0600 Subject: [PATCH 1/5] Prevent row labels that match the column label. --- index.go | 5 +++ index_test.go | 88 +++++++++++++++++++++++++++++++++------------------ pilosa.go | 1 + 3 files changed, 64 insertions(+), 30 deletions(-) diff --git a/index.go b/index.go index e15013023..f81ca810d 100644 --- a/index.go +++ b/index.go @@ -368,6 +368,11 @@ func (i *Index) createFrame(name string, opt FrameOptions) (*Frame, error) { return nil, ErrInvalidCacheType } + // Validate that row label does not match column label. + if i.columnLabel == opt.RowLabel || (opt.RowLabel == "" && i.columnLabel == DefaultRowLabel) { + return nil, ErrColumnRowLabelEqual + } + // Initialize frame. f, err := i.newFrame(i.FramePath(name), name) if err != nil { diff --git a/index_test.go b/index_test.go index 8299f5068..b98c8f791 100644 --- a/index_test.go +++ b/index_test.go @@ -34,42 +34,70 @@ func TestIndex_CreateFrameIfNotExists(t *testing.T) { } } -// Ensure index is assigned the correct time quantum on creation. -func TestIndex_CreateFrame_TimeQuantum(t *testing.T) { - t.Run("Explicit", func(t *testing.T) { - index := MustOpenIndex() - defer index.Close() +func TestIndex_CreateFrame(t *testing.T) { + // Ensure time quantum can be set appropriately on a new frame. + t.Run("TimeQuantum", func(t *testing.T) { + t.Run("Explicit", func(t *testing.T) { + index := MustOpenIndex() + defer index.Close() - // Set index time quantum. - if err := index.SetTimeQuantum(pilosa.TimeQuantum("YM")); err != nil { - t.Fatal(err) - } + // Set index time quantum. + if err := index.SetTimeQuantum(pilosa.TimeQuantum("YM")); err != nil { + t.Fatal(err) + } - // Create frame with explicit quantum. - f, err := index.CreateFrame("f", pilosa.FrameOptions{TimeQuantum: pilosa.TimeQuantum("YMDH")}) - if err != nil { - t.Fatal(err) - } else if q := f.TimeQuantum(); q != pilosa.TimeQuantum("YMDH") { - t.Fatalf("unexpected frame time quantum: %s", q) - } + // Create frame with explicit quantum. + f, err := index.CreateFrame("f", pilosa.FrameOptions{TimeQuantum: pilosa.TimeQuantum("YMDH")}) + if err != nil { + t.Fatal(err) + } else if q := f.TimeQuantum(); q != pilosa.TimeQuantum("YMDH") { + t.Fatalf("unexpected frame time quantum: %s", q) + } + }) + + t.Run("Inherited", func(t *testing.T) { + index := MustOpenIndex() + defer index.Close() + + // Set index time quantum. + if err := index.SetTimeQuantum(pilosa.TimeQuantum("YM")); err != nil { + t.Fatal(err) + } + + // Create frame. + f, err := index.CreateFrame("f", pilosa.FrameOptions{}) + if err != nil { + t.Fatal(err) + } else if q := f.TimeQuantum(); q != pilosa.TimeQuantum("YM") { + t.Fatalf("unexpected frame time quantum: %s", q) + } + }) }) - t.Run("Inherited", func(t *testing.T) { - index := MustOpenIndex() - defer index.Close() + // Ensure frame cannot be created with a matching row label. + t.Run("ErrColumnRowLabelEqual", func(t *testing.T) { + t.Run("Explicit", func(t *testing.T) { + index := MustOpenIndex() + defer index.Close() - // Set index time quantum. - if err := index.SetTimeQuantum(pilosa.TimeQuantum("YM")); err != nil { - t.Fatal(err) - } + _, err := index.CreateFrame("f", pilosa.FrameOptions{RowLabel: pilosa.DefaultColumnLabel}) + if err != pilosa.ErrColumnRowLabelEqual { + t.Fatalf("unexpected error: %s", err) + } + }) - // Create frame. - f, err := index.CreateFrame("f", pilosa.FrameOptions{}) - if err != nil { - t.Fatal(err) - } else if q := f.TimeQuantum(); q != pilosa.TimeQuantum("YM") { - t.Fatalf("unexpected frame time quantum: %s", q) - } + t.Run("Default", func(t *testing.T) { + index := MustOpenIndex() + defer index.Close() + if err := index.SetColumnLabel(pilosa.DefaultRowLabel); err != nil { + t.Fatal(err) + } + + _, err := index.CreateFrame("f", pilosa.FrameOptions{}) + if err != pilosa.ErrColumnRowLabelEqual { + t.Fatalf("unexpected error: %s", err) + } + }) }) } diff --git a/pilosa.go b/pilosa.go index 1736cccb9..72ed4a49d 100644 --- a/pilosa.go +++ b/pilosa.go @@ -20,6 +20,7 @@ var ( ErrFrameExists = errors.New("frame already exists") ErrFrameNotFound = errors.New("frame not found") ErrFrameInverseDisabled = errors.New("frame inverse disabled") + ErrColumnRowLabelEqual = errors.New("column and row labels cannot be equal") ErrInvalidView = errors.New("invalid view") ErrInvalidCacheType = errors.New("invalid cache type") From 11b84cec94a56958ba3a94e9e653983d71607f8b Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Tue, 2 May 2017 16:58:03 -0500 Subject: [PATCH 2/5] change default cache type to ranked All tests which use MustOpenFragment now explicitly pass a cacheType parameter. If this parameter is an empty string, it means that whether the test works should not depend on the cache type of the fragment. Otherwise the test should explicitly set the cache type it needs rather than relying on the default. --- fragment.go | 14 ++++++------ fragment_test.go | 58 +++++++++++++++++++++++++++++------------------- frame.go | 2 +- view.go | 4 ++-- 4 files changed, 45 insertions(+), 33 deletions(-) diff --git a/fragment.go b/fragment.go index c9c400d79..8c0964da1 100644 --- a/fragment.go +++ b/fragment.go @@ -82,9 +82,9 @@ type Fragment struct { opN int // number of ops since snapshot // Cache for row counts. - cacheType string // passed in by frame + CacheType string // passed in by frame cache Cache - cacheSize uint32 + CacheSize uint32 // Cache containing full rows (not just counts). rowCache BitmapCache @@ -115,8 +115,8 @@ func NewFragment(path, index, frame, view string, slice uint64) *Fragment { frame: frame, view: view, slice: slice, - cacheType: DefaultCacheType, - cacheSize: DefaultCacheSize, + CacheType: DefaultCacheType, + CacheSize: DefaultCacheSize, LogOutput: ioutil.Discard, MaxOpN: DefaultFragmentMaxOpN, @@ -236,11 +236,11 @@ func (f *Fragment) openStorage() error { // openCache initializes the cache from row ids persisted to disk. func (f *Fragment) openCache() error { // Determine cache type from frame name. - switch f.cacheType { + switch f.CacheType { case CacheTypeRanked: - f.cache = NewRankCache(f.cacheSize) + f.cache = NewRankCache(f.CacheSize) case CacheTypeLRU: - f.cache = NewLRUCache(f.cacheSize) + f.cache = NewLRUCache(f.CacheSize) default: return ErrInvalidCacheType } diff --git a/fragment_test.go b/fragment_test.go index 5f3da53e2..36e9361ab 100644 --- a/fragment_test.go +++ b/fragment_test.go @@ -17,6 +17,7 @@ package pilosa_test import ( "bytes" "flag" + "fmt" "io/ioutil" "math" "os" @@ -37,7 +38,7 @@ const SliceWidth = pilosa.SliceWidth // Ensure a fragment can set a bit and retrieve it. func TestFragment_SetBit(t *testing.T) { - f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0) + f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0, "") defer f.Close() // Set bits on the fragment. @@ -68,7 +69,7 @@ func TestFragment_SetBit(t *testing.T) { // Ensure a fragment can clear a set bit. func TestFragment_ClearBit(t *testing.T) { - f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0) + f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0, "") defer f.Close() // Set and then clear bits on the fragment. @@ -95,7 +96,7 @@ func TestFragment_ClearBit(t *testing.T) { // Ensure a fragment can snapshot correctly. func TestFragment_Snapshot(t *testing.T) { - f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0) + f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0, "") defer f.Close() // Set and then clear bits on the fragment. @@ -124,7 +125,7 @@ func TestFragment_Snapshot(t *testing.T) { // Ensure a fragment can iterate over all bits in order. func TestFragment_ForEachBit(t *testing.T) { - f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0) + f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0, "") defer f.Close() // Set bits on the fragment. @@ -153,13 +154,13 @@ func TestFragment_ForEachBit(t *testing.T) { // Ensure a fragment can return the top n results. func TestFragment_Top(t *testing.T) { - f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0) + f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0, pilosa.CacheTypeRanked) defer f.Close() - // Set bits on the rows 100, 101, & 102. f.MustSetBits(100, 1, 3, 200) f.MustSetBits(101, 1) f.MustSetBits(102, 1, 2) + f.RecalculateCache() // Retrieve top rows. if pairs, err := f.Top(pilosa.TopOptions{N: 2}); err != nil { @@ -175,14 +176,14 @@ func TestFragment_Top(t *testing.T) { // Ensure a fragment can filter rows when retrieving the top n rows. func TestFragment_Top_Filter(t *testing.T) { - f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0) + f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0, pilosa.CacheTypeRanked) defer f.Close() // Set bits on the rows 100, 101, & 102. f.MustSetBits(100, 1, 3, 200) f.MustSetBits(101, 1) f.MustSetBits(102, 1, 2) - + f.RecalculateCache() // Assign attributes. f.RowAttrStore.SetAttrs(101, map[string]interface{}{"x": uint64(10)}) f.RowAttrStore.SetAttrs(102, map[string]interface{}{"x": uint64(20)}) @@ -205,7 +206,7 @@ func TestFragment_Top_Filter(t *testing.T) { // Ensure a fragment can return top rows that intersect with an input row. func TestFragment_TopN_Intersect(t *testing.T) { - f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0) + f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0, pilosa.CacheTypeRanked) defer f.Close() // Create an intersecting input row. @@ -216,6 +217,7 @@ func TestFragment_TopN_Intersect(t *testing.T) { f.MustSetBits(101, 1, 2, 3, 4) // three intersections f.MustSetBits(102, 1, 2, 4, 5, 6) // two intersections f.MustSetBits(103, 1000, 1001, 1002) // no intersection + f.RecalculateCache() // Retrieve top rows. if pairs, err := f.Top(pilosa.TopOptions{N: 3, Src: src}); err != nil { @@ -235,7 +237,7 @@ func TestFragment_TopN_Intersect_Large(t *testing.T) { t.Skip("short mode") } - f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0) + f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0, pilosa.CacheTypeRanked) defer f.Close() // Create an intersecting input row. @@ -250,6 +252,7 @@ func TestFragment_TopN_Intersect_Large(t *testing.T) { f.MustSetBits(i, j) } } + f.RecalculateCache() // Retrieve top rows. if pairs, err := f.Top(pilosa.TopOptions{N: 10, Src: src}); err != nil { @@ -272,7 +275,7 @@ func TestFragment_TopN_Intersect_Large(t *testing.T) { // Ensure a fragment can return top rows when specified by ID. func TestFragment_TopN_IDs(t *testing.T) { - f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0) + f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0, pilosa.CacheTypeRanked) defer f.Close() // Set bits on various rows. @@ -360,7 +363,7 @@ func TestFragment_TopN_CacheSize(t *testing.T) { // Ensure fragment can return a checksum for its blocks. func TestFragment_Checksum(t *testing.T) { - f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0) + f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0, "") defer f.Close() // Retrieve checksum and set bits. @@ -379,7 +382,7 @@ func TestFragment_Checksum(t *testing.T) { // Ensure fragment can return a checksum for a given block. func TestFragment_Blocks(t *testing.T) { - f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0) + f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0, "") defer f.Close() // Retrieve initial checksum. @@ -417,7 +420,7 @@ func TestFragment_Blocks(t *testing.T) { // Ensure fragment returns an empty checksum if no data exists for a block. func TestFragment_Blocks_Empty(t *testing.T) { - f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0) + f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0, "") defer f.Close() // Set bits on a different block. @@ -435,7 +438,7 @@ func TestFragment_Blocks_Empty(t *testing.T) { // Ensure a fragment's cache can be persisted between restarts. func TestFragment_LRUCache_Persistence(t *testing.T) { - f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0) + f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0, pilosa.CacheTypeLRU) defer f.Close() // Set bits on the fragment. @@ -520,7 +523,7 @@ func TestFragment_RankCache_Persistence(t *testing.T) { // Ensure a fragment can be copied to another fragment. func TestFragment_WriteTo_ReadFrom(t *testing.T) { - f0 := MustOpenFragment("i", "f", pilosa.ViewStandard, 0) + f0 := MustOpenFragment("i", "f", pilosa.ViewStandard, 0, "") defer f0.Close() // Set and then clear bits on the fragment. @@ -545,7 +548,7 @@ func TestFragment_WriteTo_ReadFrom(t *testing.T) { } // Read into another fragment. - f1 := MustOpenFragment("i", "f", pilosa.ViewStandard, 0) + f1 := MustOpenFragment("i", "f", pilosa.ViewStandard, 0, "") if rn, err := f1.ReadFrom(&buf); err != nil { t.Fatal(err) } else if wn != rn { @@ -594,7 +597,7 @@ func BenchmarkFragment_Blocks(b *testing.B) { } func BenchmarkFragment_IntersectionCount(b *testing.B) { - f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0) + f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0, "") defer f.Close() f.MaxOpN = math.MaxInt32 @@ -631,7 +634,7 @@ type Fragment struct { } // NewFragment returns a new instance of Fragment with a temporary path. -func NewFragment(index, frame, view string, slice uint64) *Fragment { +func NewFragment(index, frame, view string, slice uint64, cacheType string) *Fragment { file, err := ioutil.TempFile("", "pilosa-fragment-") if err != nil { panic(err) @@ -642,13 +645,18 @@ func NewFragment(index, frame, view string, slice uint64) *Fragment { Fragment: pilosa.NewFragment(file.Name(), index, frame, view, slice), RowAttrStore: MustOpenAttrStore(), } + f.Fragment.CacheType = cacheType f.Fragment.RowAttrStore = f.RowAttrStore.AttrStore return f } // MustOpenFragment creates and opens an fragment at a temporary path. Panic on error. -func MustOpenFragment(index, frame, view string, slice uint64) *Fragment { - f := NewFragment(index, frame, view, slice) +func MustOpenFragment(index, frame, view string, slice uint64, cacheType string) *Fragment { + if cacheType == "" { + cacheType = pilosa.DefaultCacheType + } + f := NewFragment(index, frame, view, slice, cacheType) + fmt.Println("CacheType", f.CacheType) if err := f.Open(); err != nil { panic(err) } @@ -665,12 +673,14 @@ func (f *Fragment) Close() error { // Reopen closes the fragment and reopens it as a new instance. func (f *Fragment) Reopen() error { + cacheType := f.Fragment.CacheType path := f.Path() if err := f.Fragment.Close(); err != nil { return err } f.Fragment = pilosa.NewFragment(path, f.Index(), f.Frame(), f.View(), f.Slice()) + f.Fragment.CacheType = cacheType f.Fragment.RowAttrStore = f.RowAttrStore.AttrStore if err := f.Open(); err != nil { return err @@ -734,7 +744,7 @@ func GenerateImportFill(rowN int, pct float64) (rowIDs, columnIDs []uint64) { } func TestFragment_Tanimoto(t *testing.T) { - f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0) + f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0, pilosa.CacheTypeRanked) defer f.Close() src := pilosa.NewBitmap(1, 2, 3) @@ -743,6 +753,7 @@ func TestFragment_Tanimoto(t *testing.T) { f.MustSetBits(100, 1, 3, 2, 200) f.MustSetBits(101, 1, 3) f.MustSetBits(102, 1, 2, 10, 12) + f.RecalculateCache() if pairs, err := f.Top(pilosa.TopOptions{TanimotoThreshold: 50, Src: src}); err != nil { t.Fatal(err) @@ -756,7 +767,7 @@ func TestFragment_Tanimoto(t *testing.T) { } func TestFragment_Zero_Tanimoto(t *testing.T) { - f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0) + f := MustOpenFragment("i", "f", pilosa.ViewStandard, 0, pilosa.CacheTypeRanked) defer f.Close() src := pilosa.NewBitmap(1, 2, 3) @@ -765,6 +776,7 @@ func TestFragment_Zero_Tanimoto(t *testing.T) { f.MustSetBits(100, 1, 3, 2, 200) f.MustSetBits(101, 1, 3) f.MustSetBits(102, 1, 2, 10, 12) + f.RecalculateCache() if pairs, err := f.Top(pilosa.TopOptions{TanimotoThreshold: 0, Src: src}); err != nil { t.Fatal(err) diff --git a/frame.go b/frame.go index 1fd470258..333dbfd36 100644 --- a/frame.go +++ b/frame.go @@ -32,7 +32,7 @@ import ( // Default frame settings. const ( DefaultRowLabel = "rowID" - DefaultCacheType = CacheTypeLRU + DefaultCacheType = CacheTypeRanked DefaultInverseEnabled = false // Default ranked frame cache diff --git a/view.go b/view.go index 3f26185e4..2e2ca32ab 100644 --- a/view.go +++ b/view.go @@ -255,8 +255,8 @@ func (v *View) createFragmentIfNotExists(slice uint64) (*Fragment, error) { func (v *View) newFragment(path string, slice uint64) *Fragment { frag := NewFragment(path, v.index, v.frame, v.name, slice) - frag.cacheType = v.cacheType - frag.cacheSize = v.cacheSize + frag.CacheType = v.cacheType + frag.CacheSize = v.cacheSize frag.LogOutput = v.LogOutput frag.stats = v.stats.WithTags(fmt.Sprintf("slice:%d", slice)) return frag From 36e11e4ba881d96f609098a60f8af9a0cc2a9231 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Tue, 2 May 2017 17:53:22 -0500 Subject: [PATCH 3/5] remove print --- fragment_test.go | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/fragment_test.go b/fragment_test.go index 36e9361ab..f6b36f11e 100644 --- a/fragment_test.go +++ b/fragment_test.go @@ -17,7 +17,6 @@ package pilosa_test import ( "bytes" "flag" - "fmt" "io/ioutil" "math" "os" @@ -656,7 +655,7 @@ func MustOpenFragment(index, frame, view string, slice uint64, cacheType string) cacheType = pilosa.DefaultCacheType } f := NewFragment(index, frame, view, slice, cacheType) - fmt.Println("CacheType", f.CacheType) + if err := f.Open(); err != nil { panic(err) } From e3d6f966572afb8aa7ed50b5ab0b59482ee428f0 Mon Sep 17 00:00:00 2001 From: Ben Johnson Date: Wed, 3 May 2017 08:53:20 -0600 Subject: [PATCH 4/5] Add max-writes-per-requests limit. A configurable limit has been added to restrict the number of mutating calls in a `pql.Query`. This is to prevents requests from timing out from large queries. The default is set to 5000 writes per request and is configurable through the configuration file and the command line flags. --- cmd/server.go | 1 + config.go | 10 +++++++++- ctl/config.go | 1 + executor.go | 8 ++++++++ executor_test.go | 11 +++++++++++ handler.go | 7 ++++++- pilosa.go | 1 + pql/ast.go | 12 ++++++++++++ server.go | 4 ++++ server/server.go | 3 +++ 10 files changed, 56 insertions(+), 2 deletions(-) diff --git a/cmd/server.go b/cmd/server.go index 999193ba0..54603806f 100644 --- a/cmd/server.go +++ b/cmd/server.go @@ -89,6 +89,7 @@ on the configured port.`, flags.StringVarP(&Server.Config.DataDir, "data-dir", "d", "~/.pilosa", "Directory to store pilosa data files.") flags.StringVarP(&Server.Config.Host, "bind", "b", ":10101", "Default URI on which pilosa should listen.") + flags.IntVarP(&Server.Config.MaxWritesPerRequest, "max-writes-per-request", "", Server.Config.MaxWritesPerRequest, "Number of write commands per request.") flags.IntVarP(&Server.Config.Cluster.ReplicaN, "cluster.replicas", "", 1, "Number of hosts each piece of data should be stored on.") flags.StringSliceVarP(&Server.Config.Cluster.Hosts, "cluster.hosts", "", []string{}, "Comma separated list of hosts in cluster.") flags.StringSliceVarP(&Server.Config.Cluster.InternalHosts, "cluster.internal-hosts", "", []string{}, "Comma separated list of hosts in cluster used for internal communication.") diff --git a/config.go b/config.go index 0711e8dba..3e58a2d13 100644 --- a/config.go +++ b/config.go @@ -28,6 +28,9 @@ const ( // DefaultInternalPort the port the nodes intercommunicate on. DefaultInternalPort = "14000" + + // DefaultMaxWritesPerRequest is the default number of writes per request. + DefaultMaxWritesPerRequest = 5000 ) // Config represents the configuration for the command. @@ -53,13 +56,18 @@ type Config struct { Interval Duration `toml:"interval"` } `toml:"anti-entropy"` + // Limits the number of mutating commands that can be in a single request to + // the server. This includes SetBit, ClearBit, SetRowAttrs & SetColumnAttrs. + MaxWritesPerRequest int `toml:"max-writes-per-request"` + LogPath string `toml:"log-path"` } // NewConfig returns an instance of Config with default options. func NewConfig() *Config { c := &Config{ - Host: DefaultHost + ":" + DefaultPort, + Host: DefaultHost + ":" + DefaultPort, + MaxWritesPerRequest: DefaultMaxWritesPerRequest, } c.Cluster.ReplicaN = DefaultReplicaN c.Cluster.Type = DefaultClusterType diff --git a/ctl/config.go b/ctl/config.go index 75cf6adf3..d084c1e2c 100644 --- a/ctl/config.go +++ b/ctl/config.go @@ -40,6 +40,7 @@ func (cmd *ConfigCommand) Run(ctx context.Context) error { fmt.Fprintln(cmd.Stdout, strings.TrimSpace(` data-dir = "~/.pilosa" bind = "localhost:10101" +max-writes-per-request = 5000 [cluster] poll-interval = "2m0s" diff --git a/executor.go b/executor.go index a16c3f139..39753787e 100644 --- a/executor.go +++ b/executor.go @@ -49,6 +49,9 @@ type Executor struct { // Client used for remote HTTP requests. HTTPClient *http.Client + + // Maximum number of SetBit() or ClearBit() commands per request. + MaxWritesPerRequest int } // NewExecutor returns a new instance of Executor. @@ -65,6 +68,11 @@ func (e *Executor) Execute(ctx context.Context, index string, q *pql.Query, slic return nil, ErrIndexRequired } + // Verify that the number of writes do not exceed the maximum. + if e.MaxWritesPerRequest > 0 && q.WriteCallN() > e.MaxWritesPerRequest { + return nil, ErrTooManyWrites + } + // Default options. if opt == nil { opt = &ExecOptions{} diff --git a/executor_test.go b/executor_test.go index 1e0cd8343..a89911f38 100644 --- a/executor_test.go +++ b/executor_test.go @@ -699,6 +699,17 @@ func TestExecutor_Execute_Remote_TopN(t *testing.T) { } } +// Ensure executor returns an error if too many writes are in a single request. +func TestExecutor_Execute_ErrMaxWritesPerRequest(t *testing.T) { + hldr := MustOpenHolder() + defer hldr.Close() + e := NewExecutor(hldr.Holder, NewCluster(1)) + e.MaxWritesPerRequest = 3 + if _, err := e.Execute(context.Background(), "i", MustParse(`SetBit() ClearBit() SetBit() SetBit()`), nil, nil); err != pilosa.ErrTooManyWrites { + t.Fatalf("unexpected error: %s", err) + } +} + // Executor represents a test wrapper for pilosa.Executor. type Executor struct { *pilosa.Executor diff --git a/handler.go b/handler.go index fb2a35a0f..3e837bef3 100644 --- a/handler.go +++ b/handler.go @@ -228,7 +228,12 @@ func (h *Handler) handlePostQuery(w http.ResponseWriter, r *http.Request) { // Set appropriate status code, if there is an error. if resp.Err != nil { - w.WriteHeader(http.StatusInternalServerError) + switch resp.Err { + case ErrTooManyWrites: + w.WriteHeader(http.StatusRequestEntityTooLarge) + default: + w.WriteHeader(http.StatusInternalServerError) + } } // Write response back to client. diff --git a/pilosa.go b/pilosa.go index dd7402677..b50ebbf78 100644 --- a/pilosa.go +++ b/pilosa.go @@ -44,6 +44,7 @@ var ( // ErrFragmentNotFound is returned when a fragment does not exist. ErrFragmentNotFound = errors.New("fragment not found") ErrQueryRequired = errors.New("query required") + ErrTooManyWrites = errors.New("too many write commands") ) // Regular expression to validate index and frame names. diff --git a/pql/ast.go b/pql/ast.go index b83b35731..19b5381ed 100644 --- a/pql/ast.go +++ b/pql/ast.go @@ -28,6 +28,18 @@ type Query struct { Calls []*Call } +// WriteCallN returns the number of mutating calls. +func (q *Query) WriteCallN() int { + var n int + for _, call := range q.Calls { + switch call.Name { + case "SetBit", "ClearBit", "SetRowAttrs", "SetColumnAttrs": + n++ + } + } + return n +} + // String returns a string representation of the query. func (q *Query) String() string { a := make([]string, len(q.Calls)) diff --git a/server.go b/server.go index 4a5b4192c..c283b3d0e 100644 --- a/server.go +++ b/server.go @@ -61,6 +61,9 @@ type Server struct { AntiEntropyInterval time.Duration PollingInterval time.Duration + // Misc options. + MaxWritesPerRequest int + LogOutput io.Writer } @@ -129,6 +132,7 @@ func (s *Server) Open() error { e.Holder = s.Holder e.Host = s.Host e.Cluster = s.Cluster + e.MaxWritesPerRequest = s.MaxWritesPerRequest // Initialize HTTP handler. s.Handler.Broadcaster = s.Broadcaster diff --git a/server/server.go b/server/server.go index 73522b0fa..b8475718a 100644 --- a/server/server.go +++ b/server/server.go @@ -135,6 +135,9 @@ func (m *Command) SetupServer() error { m.Server.Holder.Path = m.Config.DataDir m.Server.Holder.Stats = pilosa.NewExpvarStatsClient() + // Copy configuration flags. + m.Server.MaxWritesPerRequest = m.Config.MaxWritesPerRequest + var err error m.Server.Host, err = normalizeHost(m.Config.Host) if err != nil { From 9fdb2e1460394af09464bbfe428f2f6998df7df0 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Wed, 3 May 2017 13:52:04 -0500 Subject: [PATCH 5/5] differentiate error messages --- executor.go | 2 +- httpbroadcast/messenger.go | 2 +- server.go | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/executor.go b/executor.go index a16c3f139..e0094eb1f 100644 --- a/executor.go +++ b/executor.go @@ -1039,7 +1039,7 @@ func (e *Executor) exec(ctx context.Context, node *Node, index string, q *pql.Qu // Check status code. if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("invalid status: code=%d, err=%s", resp.StatusCode, body) + return nil, fmt.Errorf("invalid status Executor.exec: code=%d, err=%s, req: %v", resp.StatusCode, body, req) } // Decode response object. diff --git a/httpbroadcast/messenger.go b/httpbroadcast/messenger.go index 387f28585..2b6bb599c 100644 --- a/httpbroadcast/messenger.go +++ b/httpbroadcast/messenger.go @@ -114,7 +114,7 @@ func (h *HTTPBroadcaster) sendNodeMessage(node *pilosa.Node, msg []byte) error { // Check status code. if resp.StatusCode != http.StatusOK { - return fmt.Errorf("invalid status: code=%d, err=%s", resp.StatusCode, body) + return fmt.Errorf("invalid status sendNodeMessage: code=%d, err=%s, req=%v", resp.StatusCode, body, req) } return nil diff --git a/server.go b/server.go index 4a5b4192c..77686bea3 100644 --- a/server.go +++ b/server.go @@ -417,7 +417,7 @@ func checkMaxSlices(hostport string) (map[string]uint64, error) { // Check status code. if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("invalid status: code=%d, err=%s", resp.StatusCode, body) + return nil, fmt.Errorf("invalid status checkMaxSlices: code=%d, err=%s, req=%v", resp.StatusCode, body, req) } // Decode response object.