Merge pull request #634 from molecula/unionfix

Follow roaring.Union() with optimize() to avoid overly large containers.

badger.go

@@ -296,6 +296,7 @@ func (r *badgerRegistrar) openBadgerDBWrapper(bpath string) (*BadgerDBWrapper, e
 
 	opt.Compression = badgeroptions.None // turn off compression.
 	opt.ZSTDCompressionLevel = 0         // really, just in case.
+	opt.SyncWrites = true                // default is true, safe.
 
 	// MaxCacheSize docs:
 	//
@@ -1535,7 +1536,7 @@ func (tx *BadgerTx) ImportRoaringBits(index, field, view string, shard uint64, i
 				continue
 			}
 
-			newC := oldC.UnionInPlace(synthC)
+			newC := roaring.Union(oldC, synthC) // UnionInPlace was giving us crashes on overly large containers.
 
 			if roaring.ContainerType(newC) == containerBitmap {
 				newC.Repair() // update the bit-count so .n is valid. b/c UnionInPlace doesn't update it.
@@ -1634,6 +1635,9 @@ func fromArray16(a []uint16) []byte {
 	if len(a) == 0 {
 		return []byte{}
 	}
+	if len(a) > 4096 {
+		panic(fmt.Sprintf("cannot put more than 4096 integers into an array container: %v too big", len(a)))
+	}
 	return (*[8192]byte)(unsafe.Pointer(&a[0]))[: len(a)*2 : len(a)*2]
 }
 
@@ -1650,6 +1654,9 @@ func fromInterval16(a []roaring.Interval16) []byte {
 	if len(a) == 0 {
 		return []byte{}
 	}
+	if len(a) > 2048 {
+		panic(fmt.Sprintf("cannot put more than 2048 roaring.Interval16 into a container: %v too big", len(a)))
+	}
 	return (*[8192]byte)(unsafe.Pointer(&a[0]))[: len(a)*4 : len(a)*4]
 }
 
@@ -1769,6 +1776,23 @@ func asInts(a []uint64) (r []int) {
 	return
 }
 
+var _ = zeroKeyContainerAsString // happy linter
+
+// for debugging
+func zeroKeyContainerAsString(ct *roaring.Container) (r string) {
+	cts := roaring.NewSliceContainers()
+	cts.Put(0, ct)
+	rbm := &roaring.Bitmap{Containers: cts}
+	r = fmt.Sprintf("[%v]:", containerTypeNames[roaring.ContainerType(ct)]) + bitmapAsString(rbm)
+	return
+}
+
+var containerTypeNames = map[byte]string{
+	containerArray:  "array",
+	containerBitmap: "bitmap",
+	containerRun:    "run",
+}
+
 func bitmapAsString(rbm *roaring.Bitmap) (r string) {
 	r = "c("
 	slc := rbm.Slice()

cmd/loader/loader.go

@@ -0,0 +1,150 @@
+// 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, "/")
+		index := parts[0]
+		field := parts[1]
+		view := parts[3]
+		shard, err := strconv.ParseUint(parts[5], 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")
+	}
+}
+
+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/loader/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

@@ -11,3 +11,4 @@
 ./logger/filewriter_test.go
 ./vprint.go
 ./rbf/vprint.go
+./cmd/loader/vprint.go

roaring/roaring.go

@@ -4368,6 +4368,7 @@ func unionArrayArray(a, b *Container) *Container {
 			break
 		}
 	}
+	// note: len(output) CAN be > 4096
 	return NewContainerArray(output)
 }
 
@@ -6865,8 +6866,12 @@ func ConvertRunToBitmap(c *Container) {
 func Optimize(c *Container) {
 	c.optimize()
 }
-func Union(a, b *Container) *Container {
-	return union(a, b)
+func Union(a, b *Container) (c *Container) {
+	c = union(a, b)
+	// c can be have arrays that are too big, and need
+	// to be optimized into raw bitmaps.
+	c.optimize()
+	return c
 }
 
 func Difference(a, b *Container) *Container {