Merge branch 'master' into skip_cluster_tests_bg

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
+}

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

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

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)
 }
 

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")
+	}
+}