diff --git a/cmd/badloader/badloader.go b/cmd/badloader/badloader.go new file mode 100644 index 000000000..719a256f5 --- /dev/null +++ b/cmd/badloader/badloader.go @@ -0,0 +1,154 @@ +// Copyright 2020 Pilosa Corp. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "archive/tar" + "compress/gzip" + "context" + "time" + //"fmt" + "fmt" + "github.com/pilosa/pilosa/v2" + "github.com/pilosa/pilosa/v2/http" + "io" + "io/ioutil" + gohttp "net/http" + //"log" + "os" + //"path/filepath" + //"sort" + "strconv" + "strings" +) + +func UploadTar(srcFile string, client *http.InternalClient) error { + t0 := time.Now() + + f, err := os.Open(srcFile) + if err != nil { + return (err) + } + defer f.Close() + var tarReader *tar.Reader + if strings.HasSuffix(srcFile, "gz") { + gzf, err := gzip.NewReader(f) + if err != nil { + return err + } + tarReader = tar.NewReader(gzf) + } else { + tarReader = tar.NewReader(f) + } + viewData := make(map[string][]byte) + //given ordered by index/field/view + //trait_store/product_count__commercial_cd_or_share_certificate/views/bsig_product_count__commercial_cd_or_share_certificate/fragments/255 + lastIndex := "" + lastField := "" + lastShard := uint64(0) + //vv("top of tar loop") + n := 0 + for { + header, err := tarReader.Next() + if err == io.EOF { + if header != nil { + panic("header should not be nil on err io.EOF") + } + //submit any stuff we have left + if len(viewData) > 0 { + request := &pilosa.ImportRoaringRequest{ + Views: viewData, + } + // Submit(lastIndex, lastField, lastShard, request) + //vv("about to submit lastIndex='%v' lastShard='%v'", lastIndex, lastShard) + uri := GetImportRoaringURI(lastIndex, lastShard) + err := client.ImportRoaring(context.Background(), uri, lastIndex, lastField, lastShard, false, request) + panicOn(err) + //vv("done with submit lastIndex='%v' lastShard='%v'", lastIndex, lastShard) + } + return nil + } + //vv("got header '%v'", header.Name) + n++ + if n%500 == 0 { + vv("n = %v, progress, elapsed '%v'", n, time.Since(t0)) + } + parts := strings.Split(header.Name, "/") + //vv("parts = '%#v'", parts) + index := parts[1] + field := parts[2] + view := parts[4] + shard, err := strconv.ParseUint(parts[6], 10, 64) + if err != nil { + return err + } + // TODO: shards can be loaded in parallel, so maybe farm out to a worker set of goro. + if index != lastIndex || field != lastField || shard != lastShard { + if len(viewData) > 0 { + request := &pilosa.ImportRoaringRequest{ + Views: viewData, + } + //vv("about to submit lastIndex='%v' lastShard='%v'", lastIndex, lastShard) + uri := GetImportRoaringURI(lastIndex, lastShard) + panicOn(client.ImportRoaring(context.Background(), uri, lastIndex, lastField, lastShard, false, request)) + viewData = make(map[string][]byte) + //vv("done with submit lastIndex='%v' lastShard='%v'; took='%v'", lastIndex, lastShard, time.Since(t0)) + + } + } + roaringData, err := ioutil.ReadAll(tarReader) + if err != nil { + return err + } + if _, already := viewData[view]; already { + panic(fmt.Sprintf("view '%v' already present!", view)) + } + viewData[view] = roaringData + lastIndex = index + lastField = field + + //lastShard = shard + //vv("bottom of loop") + } +} + +// badloader reproduce a union in place issue for us. slurp is +// the new "good" loader, and should always be preferred now +// when not trying to repro that bug. pulled from 85fa67e8 +func main() { + + host := "127.0.0.1:10101" + h := &gohttp.Client{} + c, err := http.NewInternalClient(host, h) + panicOn(err) + + tarSrcPath := "q2.tar.gz" + t0 := time.Now() + panicOn(UploadTar(tarSrcPath, c)) + vv("total elapsed '%v'", time.Since(t0)) +} + +var globURI *pilosa.URI + +func init() { + var err error + globURI, err = pilosa.NewURIFromHostPort("127.0.0.1", 10101) + panicOn(err) +} + +// get correct node to go to. +func GetImportRoaringURI(index string, shard uint64) *pilosa.URI { + return globURI +} diff --git a/cmd/badloader/vprint.go b/cmd/badloader/vprint.go new file mode 100644 index 000000000..e6bea8d55 --- /dev/null +++ b/cmd/badloader/vprint.go @@ -0,0 +1,177 @@ +// home: https://github.com/glyerine/vprint +// Copyright 2019 Jason E. Aten, Ph.D. All rights reserved. +// License: MIT +// +// MIT License +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in all +// copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. + +package main + +import ( + "fmt" + "io" + "os" + "path" + "runtime" + "runtime/debug" + "sync" + "time" +) + +const RFC3339MsecTz0 = "2006-01-02T15:04:05.000Z07:00" +const RFC3339UsecTz0 = "2006-01-02T15:04:05.000000Z07:00" + +// for tons of debug output +var VerboseVerbose bool = false + +// convience functions for . import +var pp = PP +var vv = VV + +var panicOn = PanicOn + +func init() { + // keeper linter happy + _ = pp + _ = vv +} + +func PanicOn(err error) { + if err != nil { + panic(err) + } +} + +func PP(format string, a ...interface{}) { + if VerboseVerbose { + TSPrintf(format, a...) + } +} + +func VV(format string, a ...interface{}) { + TSPrintf(format, a...) +} + +func AlwaysPrintf(format string, a ...interface{}) { + TSPrintf(format, a...) +} + +var tsPrintfMut sync.Mutex + +// time-stamped printf +func TSPrintf(format string, a ...interface{}) { + tsPrintfMut.Lock() + Printf("\n%s %s ", FileLine(3), ts()) + Printf(format+"\n", a...) + tsPrintfMut.Unlock() +} + +// get timestamp for logging purposes +func ts() string { + return time.Now().Format(RFC3339UsecTz0) +} + +// so we can multi write easily, use our own printf +var OurStdout io.Writer = os.Stdout + +// Printf formats according to a format specifier and writes to standard output. +// It returns the number of bytes written and any write error encountered. +func Printf(format string, a ...interface{}) (n int, err error) { + return fmt.Fprintf(OurStdout, format, a...) +} + +func FileLine(depth int) string { + _, fileName, fileLine, ok := runtime.Caller(depth) + var s string + if ok { + s = fmt.Sprintf("%s:%d", path.Base(fileName), fileLine) + } else { + s = "" + } + return s +} + +func stack() string { + return string(debug.Stack()) +} + +func FileExists(name string) bool { + fi, err := os.Stat(name) + if err != nil { + return false + } + if fi.IsDir() { + return false + } + return true +} + +func DirExists(name string) bool { + fi, err := os.Stat(name) + if err != nil { + return false + } + if fi.IsDir() { + return true + } + return false +} + +func FileSize(name string) (int64, error) { + fi, err := os.Stat(name) + if err != nil { + return -1, err + } + return fi.Size(), nil +} + +// Caller returns the name of the calling function. +func Caller(upStack int) string { + // elide ourself and runtime.Callers + target := upStack + 2 + + pc := make([]uintptr, target+2) + n := runtime.Callers(0, pc) + + f := runtime.Frame{Function: "unknown"} + if n > 0 { + frames := runtime.CallersFrames(pc[:n]) + for i := 0; i <= target; i++ { + contender, more := frames.Next() + if i == target { + f = contender + } + if !more { + break + } + } + } + return f.Function +} + +// happy linter: +var _ = DirExists +var _ = FileExists +var _ = Caller +var _ = stack +var _ = RFC3339MsecTz0 +var _ = RFC3339UsecTz0 +var _ = AlwaysPrintf +var _ = FileSize diff --git a/license.exceptions b/license.exceptions index 9f6914859..639e3558f 100644 --- a/license.exceptions +++ b/license.exceptions @@ -12,6 +12,7 @@ ./vprint.go ./rbf/vprint.go ./cmd/slurp/vprint.go +./cmd/badloader/vprint.go ./cmd/demo-lmdb/vprint.go ./gid.go ./cmd/lmdb-keydump/vprint.go diff --git a/roaring/roaring.go b/roaring/roaring.go index 49e5b08e5..d82ef3869 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -6891,7 +6891,7 @@ func (c *Container) CountRange(start, end int32) (n int32) { return c.countRange(start, end) } -func (c *Container) UnionInPlace(other *Container) *Container { +func (c *Container) UnionInPlace(other *Container) (r *Container) { return c.unionInPlace(other) } diff --git a/roaring/roaring_test.go b/roaring/roaring_test.go index 217d24bb7..18b9ce8ab 100644 --- a/roaring/roaring_test.go +++ b/roaring/roaring_test.go @@ -2260,3 +2260,51 @@ func TestRunAddRemoveAddRemove(t *testing.T) { } } } + +// confirm that two fairly full array containers, when +// unioned in place, do not produce a new invalid array container that +// has more array elements than can fit in a bitmap; such +// was seen at one point by a Container.UnionInPlace operation. +func TestContainer_UnionInPlace_TwoBigArrays(t *testing.T) { + var ( + bm0 = roaring.NewBitmap() + bm1 = roaring.NewBitmap() + ) + for i := uint64(0); i < 8192; i++ { + if i%3 == 0 { + if _, err := bm0.Add(i); err != nil { + t.Fatalf("adding bits: %v", err) + } + } + } + for i := uint64(0); i < 8192; i++ { + if i%3 == 1 { + if _, err := bm1.Add(i); err != nil { + t.Fatalf("adding bits: %v", err) + } + } + } + it, ok := bm0.Containers.Iterator(0) + if !ok || it == nil { + panic("empty iterator!") + } + if !it.Next() { + panic("no container???") + } + _, ct0 := it.Value() + + it, ok = bm1.Containers.Iterator(0) + if !ok || it == nil { + panic("empty iterator!") + } + if !it.Next() { + panic("no container???") + } + _, ct1 := it.Value() + + resCt := ct0.UnionInPlace(ct1) + typ := roaring.ContainerType(resCt) + if typ == roaring.ContainerArray { + panic("should be NOT be an array now") + } +}