chore(metastore): use column reader for blooms (#20053)

This PR is a continuation of #19992. I use `pointers.Reader` to refine the sections that match bloom filters.

I also fix minor issues that I introduced in the previous PR:
- convert the columns only once per column (instead of doing it for every row)
- ignore the sections that are of a wrong "type"

Author: ivkalita

pkg/dataobj/metastore/iter.go

@@ -90,10 +90,8 @@ func forEachStreamSectionPointer(
 		}
 
 		if colStreamID == nil || colMinTimestamp == nil || colMaxTimestamp == nil {
-			return fmt.Errorf(
-				"one of mandatory columns is missing: (streamID=%t, minTimestamp=%t, maxTimestamp=%t)",
-				colStreamID == nil, colMinTimestamp == nil, colMaxTimestamp == nil,
-			)
+			// the section has no rows with stream-based indices and can be ignored completely
+			continue
 		}
 
 		reader.Reset(pointers.ReaderOptions{
@@ -123,60 +121,68 @@ func forEachStreamSectionPointer(
 
 				switch pointerCol.Type {
 				case pointers.ColumnTypePath:
+					values := col.(*array.String)
 					for rIdx := range numRows {
 						if col.IsNull(rIdx) {
 							continue
 						}
-						buf[rIdx].Path = col.(*array.String).Value(rIdx)
+						buf[rIdx].Path = values.Value(rIdx)
 					}
 				case pointers.ColumnTypeSection:
+					values := col.(*array.Int64)
 					for rIdx := range numRows {
 						if col.IsNull(rIdx) {
 							continue
 						}
-						buf[rIdx].Section = col.(*array.Int64).Value(rIdx)
+						buf[rIdx].Section = values.Value(rIdx)
 					}
 				case pointers.ColumnTypeStreamID:
+					values := col.(*array.Int64)
 					for rIdx := range numRows {
 						if col.IsNull(rIdx) {
 							continue
 						}
-						buf[rIdx].StreamID = col.(*array.Int64).Value(rIdx)
+						buf[rIdx].StreamID = values.Value(rIdx)
 					}
 				case pointers.ColumnTypeStreamIDRef:
+					values := col.(*array.Int64)
 					for rIdx := range numRows {
 						if col.IsNull(rIdx) {
 							continue
 						}
-						buf[rIdx].StreamIDRef = col.(*array.Int64).Value(rIdx)
+						buf[rIdx].StreamIDRef = values.Value(rIdx)
 					}
 				case pointers.ColumnTypeMinTimestamp:
+					values := col.(*array.Timestamp)
 					for rIdx := range numRows {
 						if col.IsNull(rIdx) {
 							continue
 						}
-						buf[rIdx].StartTs = time.Unix(0, int64(col.(*array.Timestamp).Value(rIdx)))
+						buf[rIdx].StartTs = time.Unix(0, int64(values.Value(rIdx)))
 					}
 				case pointers.ColumnTypeMaxTimestamp:
+					values := col.(*array.Timestamp)
 					for rIdx := range numRows {
 						if col.IsNull(rIdx) {
 							continue
 						}
-						buf[rIdx].EndTs = time.Unix(0, int64(col.(*array.Timestamp).Value(rIdx)))
+						buf[rIdx].EndTs = time.Unix(0, int64(values.Value(rIdx)))
 					}
 				case pointers.ColumnTypeRowCount:
+					values := col.(*array.Int64)
 					for rIdx := range numRows {
 						if col.IsNull(rIdx) {
 							continue
 						}
-						buf[rIdx].LineCount = col.(*array.Int64).Value(rIdx)
+						buf[rIdx].LineCount = values.Value(rIdx)
 					}
 				case pointers.ColumnTypeUncompressedSize:
+					values := col.(*array.Int64)
 					for rIdx := range numRows {
 						if col.IsNull(rIdx) {
 							continue
 						}
-						buf[rIdx].UncompressedSize = col.(*array.Int64).Value(rIdx)
+						buf[rIdx].UncompressedSize = values.Value(rIdx)
 					}
 				default:
 					continue
@@ -195,3 +201,149 @@ func forEachStreamSectionPointer(
 
 	return nil
 }
+
+func forEachMatchedPointerSectionKey(
+	ctx context.Context,
+	object *dataobj.Object,
+	columnName scalar.Scalar,
+	matchColumnValue string,
+	f func(key SectionKey),
+) error {
+	targetTenant, err := user.ExtractOrgID(ctx)
+	if err != nil {
+		return fmt.Errorf("extracting org ID: %w", err)
+	}
+
+	var reader pointers.Reader
+	defer reader.Close()
+
+	const batchSize = 128
+	buf := make([]SectionKey, batchSize)
+
+	for _, section := range object.Sections().Filter(pointers.CheckSection) {
+		if section.Tenant != targetTenant {
+			continue
+		}
+
+		sec, err := pointers.Open(ctx, section)
+		if err != nil {
+			return fmt.Errorf("opening section: %w", err)
+		}
+
+		pointerCols, err := findPointersColumnsByTypes(
+			sec.Columns(),
+			pointers.ColumnTypePath,
+			pointers.ColumnTypeSection,
+			pointers.ColumnTypeColumnName,
+			pointers.ColumnTypeValuesBloomFilter,
+		)
+		if err != nil {
+			return fmt.Errorf("finding pointers columns: %w", err)
+		}
+
+		var (
+			colColumnName *pointers.Column
+			colBloom      *pointers.Column
+		)
+
+		for _, c := range pointerCols {
+			if c.Type == pointers.ColumnTypeColumnName {
+				colColumnName = c
+			}
+			if c.Type == pointers.ColumnTypeValuesBloomFilter {
+				colBloom = c
+			}
+			if colColumnName != nil && colBloom != nil {
+				break
+			}
+		}
+
+		if colColumnName == nil || colBloom == nil {
+			// the section has no rows for blooms and can be ignored completely
+			continue
+		}
+
+		reader.Reset(
+			pointers.ReaderOptions{
+				Columns: pointerCols,
+				Predicates: []pointers.Predicate{
+					pointers.WhereBloomFilterMatches(colColumnName, colBloom, columnName, matchColumnValue),
+				},
+			},
+		)
+
+		for {
+			rec, readErr := reader.Read(ctx, batchSize)
+			if readErr != nil && !errors.Is(readErr, io.EOF) {
+				return fmt.Errorf("reading record batch: %w", readErr)
+			}
+			if rec == nil {
+				if errors.Is(readErr, io.EOF) {
+					break
+				}
+				continue
+			}
+
+			numRows := int(rec.NumRows())
+			if numRows == 0 {
+				rec.Release()
+				if errors.Is(readErr, io.EOF) {
+					break
+				}
+				continue
+			}
+
+			for colIdx := range int(rec.NumCols()) {
+				col := rec.Column(colIdx)
+				pointerCol := pointerCols[colIdx]
+
+				switch pointerCol.Type {
+				case pointers.ColumnTypePath:
+					values := col.(*array.String)
+					for rowIdx := range numRows {
+						if col.IsNull(rowIdx) {
+							continue
+						}
+						buf[rowIdx].ObjectPath = values.Value(rowIdx)
+					}
+				case pointers.ColumnTypeSection:
+					values := col.(*array.Int64)
+					for rowIdx := range numRows {
+						if col.IsNull(rowIdx) {
+							continue
+						}
+						buf[rowIdx].SectionIdx = values.Value(rowIdx)
+					}
+				default:
+					continue
+				}
+			}
+
+			for _, sectionKey := range buf {
+				f(sectionKey)
+			}
+
+			if errors.Is(readErr, io.EOF) {
+				break
+			}
+		}
+	}
+
+	return nil
+}
+
+func findPointersColumnsByTypes(allColumns []*pointers.Column, columnTypes ...pointers.ColumnType) ([]*pointers.Column, error) {
+	result := make([]*pointers.Column, 0, len(columnTypes))
+
+	for _, c := range allColumns {
+		for _, neededType := range columnTypes {
+			if neededType != c.Type {
+				continue
+			}
+
+			result = append(result, c)
+		}
+	}
+
+	return result, nil
+}