MB-61890 - Introducing config at zap layer

build.go

@@ -160,7 +160,7 @@ func persistStoredFieldValues(fieldID int,
 func InitSegmentBase(mem []byte, memCRC uint32, chunkMode uint32,
 	fieldsMap map[string]uint16, fieldsInv []string, numDocs uint64,
 	storedIndexOffset uint64, dictLocs []uint64,
-	sectionsIndexOffset uint64) (*SegmentBase, error) {
+	sectionsIndexOffset uint64, config map[string]interface{}) (*SegmentBase, error) {
 	sb := &SegmentBase{
 		mem:                 mem,
 		memCRC:              memCRC,
@@ -175,6 +175,7 @@ func InitSegmentBase(mem []byte, memCRC uint32, chunkMode uint32,
 		dictLocs:            dictLocs,
 		fieldFSTs:           make(map[uint16]*vellum.FST),
 		vecIndexCache:       newVectorIndexCache(),
+		config:              config,
 	}
 	sb.updateSize()
 

build_test.go

@@ -48,28 +48,28 @@ func buildTestSegment() (*SegmentBase, uint64, error) {
 		doc,
 	}
 
-	seg, size, err := zapPlugin.newWithChunkMode(results, DefaultChunkMode)
+	seg, size, err := zapPlugin.newWithChunkMode(results, DefaultChunkMode, nil)
 	return seg.(*SegmentBase), size, err
 }
 
 func buildTestSegmentMulti() (*SegmentBase, uint64, error) {
 	results := buildTestAnalysisResultsMulti()
 
-	seg, size, err := zapPlugin.newWithChunkMode(results, DefaultChunkMode)
+	seg, size, err := zapPlugin.newWithChunkMode(results, DefaultChunkMode, nil)
 	return seg.(*SegmentBase), size, err
 }
 
 func buildTestSegmentMultiWithChunkFactor(chunkFactor uint32) (*SegmentBase, uint64, error) {
 	results := buildTestAnalysisResultsMulti()
 
-	seg, size, err := zapPlugin.newWithChunkMode(results, chunkFactor)
+	seg, size, err := zapPlugin.newWithChunkMode(results, chunkFactor, nil)
 	return seg.(*SegmentBase), size, err
 }
 
 func buildTestSegmentMultiWithDifferentFields(includeDocA, includeDocB bool) (*SegmentBase, uint64, error) {
 	results := buildTestAnalysisResultsMultiWithDifferentFields(includeDocA, includeDocB)
 
-	seg, size, err := zapPlugin.newWithChunkMode(results, DefaultChunkMode)
+	seg, size, err := zapPlugin.newWithChunkMode(results, DefaultChunkMode, nil)
 	return seg.(*SegmentBase), size, err
 }
 
@@ -152,7 +152,7 @@ func buildTestSegmentWithDefaultFieldMapping(chunkFactor uint32) (
 		doc,
 	}
 
-	sb, _, err := zapPlugin.newWithChunkMode(results, chunkFactor)
+	sb, _, err := zapPlugin.newWithChunkMode(results, chunkFactor, nil)
 
 	return sb.(*SegmentBase), fields, err
 }

dict_test.go

@@ -34,7 +34,7 @@ func buildTestSegmentForDict() (*SegmentBase, uint64, error) {
 		doc,
 	}
 
-	seg, size, err := zapPlugin.newWithChunkMode(results, 1024)
+	seg, size, err := zapPlugin.newWithChunkMode(results, 1024, nil)
 	return seg.(*SegmentBase), size, err
 }
 

merge.go

@@ -35,9 +35,21 @@ const docDropped = math.MaxUint64 // sentinel docNum to represent a deleted doc
 // Merge takes a slice of segments and bit masks describing which
 // documents may be dropped, and creates a new segment containing the
 // remaining data.  This new segment is built at the specified path.
-func (*ZapPlugin) Merge(segments []seg.Segment, drops []*roaring.Bitmap, path string,
+func (z *ZapPlugin) Merge(segments []seg.Segment, drops []*roaring.Bitmap, path string,
 	closeCh chan struct{}, s seg.StatsReporter) (
 	[][]uint64, uint64, error) {
+	return z.merge(segments, drops, path, closeCh, s, nil)
+}
+
+func (z *ZapPlugin) MergeEx(segments []seg.Segment, drops []*roaring.Bitmap, path string,
+	closeCh chan struct{}, s seg.StatsReporter, config map[string]interface{}) (
+	[][]uint64, uint64, error) {
+	return z.merge(segments, drops, path, closeCh, s, config)
+}
+
+func (*ZapPlugin) merge(segments []seg.Segment, drops []*roaring.Bitmap, path string,
+	closeCh chan struct{}, s seg.StatsReporter, config map[string]interface{}) (
+	[][]uint64, uint64, error) {
 	segmentBases := make([]*SegmentBase, len(segments))
 	for segmenti, segment := range segments {
 		switch segmentx := segment.(type) {

merge_test.go

@@ -146,7 +146,7 @@ func testMergeWithEmptySegments(t *testing.T, before bool, numEmptySegments int)
 
 		_ = os.RemoveAll("/tmp/" + fname)
 
-		emptySegment, _, err := zapPlugin.newWithChunkMode([]index.Document{}, 1024)
+		emptySegment, _, err := zapPlugin.newWithChunkMode([]index.Document{}, 1024, nil)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -644,7 +644,7 @@ func buildTestSegmentMultiHelper(docIds []string) (*SegmentBase, uint64, error)
 		doc2,
 	}
 
-	seg, size, err := zapPlugin.newWithChunkMode(results, 1024)
+	seg, size, err := zapPlugin.newWithChunkMode(results, 1024, nil)
 	return seg.(*SegmentBase), size, err
 }
 

new.go

@@ -42,11 +42,16 @@ var ValidateDocFields = func(field index.Field) error {
 // New creates an in-memory zap-encoded SegmentBase from a set of Documents
 func (z *ZapPlugin) New(results []index.Document) (
 	segment.Segment, uint64, error) {
-	return z.newWithChunkMode(results, DefaultChunkMode)
+	return z.newWithChunkMode(results, DefaultChunkMode, nil)
+}
+
+func (z *ZapPlugin) NewEx(results []index.Document, config map[string]interface{}) (
+	segment.Segment, uint64, error) {
+	return z.newWithChunkMode(results, DefaultChunkMode, config)
 }
 
 func (*ZapPlugin) newWithChunkMode(results []index.Document,
-	chunkMode uint32) (segment.Segment, uint64, error) {
+	chunkMode uint32, config map[string]interface{}) (segment.Segment, uint64, error) {
 	s := interimPool.Get().(*interim)
 
 	var br bytes.Buffer
@@ -73,7 +78,7 @@ func (*ZapPlugin) newWithChunkMode(results []index.Document,
 
 	sb, err := InitSegmentBase(br.Bytes(), s.w.Sum32(), chunkMode,
 		s.FieldsMap, s.FieldsInv, uint64(len(results)),
-		storedIndexOffset, dictOffsets, sectionsIndexOffset)
+		storedIndexOffset, dictOffsets, sectionsIndexOffset, config)
 
 	// get the bytes written before the interim's reset() call
 	// write it to the newly formed segment base.

segment.go

@@ -38,8 +38,18 @@ func init() {
 	reflectStaticSizeSegmentBase = int(unsafe.Sizeof(sb))
 }
 
+// OpenEx returns a zap impl of a segment which tracks some config values during
+// the its lifetime.
+func (z *ZapPlugin) OpenEx(path string, config map[string]interface{}) (segment.Segment, error) {
+	return z.open(path, config)
+}
+
 // Open returns a zap impl of a segment
-func (*ZapPlugin) Open(path string) (segment.Segment, error) {
+func (z *ZapPlugin) Open(path string) (segment.Segment, error) {
+	return z.open(path, nil)
+}
+
+func (*ZapPlugin) open(path string, config map[string]interface{}) (segment.Segment, error) {
 	f, err := os.Open(path)
 	if err != nil {
 		return nil, err
@@ -57,6 +67,7 @@ func (*ZapPlugin) Open(path string) (segment.Segment, error) {
 			fieldFSTs:      make(map[uint16]*vellum.FST),
 			vecIndexCache:  newVectorIndexCache(),
 			fieldDvReaders: make([]map[uint16]*docValueReader, len(segmentSections)),
+			config:         config,
 		},
 		f:    f,
 		mm:   mm,
@@ -108,6 +119,8 @@ type SegmentBase struct {
 	bytesRead    uint64
 	bytesWritten uint64
 
+	config map[string]interface{} // config for the segment
+
 	m         sync.Mutex
 	fieldFSTs map[uint16]*vellum.FST