Function CopyBlock

tempodb/encoding/vparquet3/copy.go:14–72

(ctx context.Context, fromMeta, toMeta *backend.BlockMeta, from backend.Reader, to backend.Writer)

)

func CopyBlock(ctx context.Context, fromMeta, toMeta *backend.BlockMeta, from backend.Reader, to backend.Writer) error {
	// Copy streams, efficient but can't cache.
	copyStream := func(name string) error {
		reader, size, err := from.StreamReader(ctx, name, (uuid.UUID)(fromMeta.BlockID), fromMeta.TenantID)
		if err != nil {
			return fmt.Errorf("error reading %s: %w", name, err)
		}
		defer reader.Close()

		return to.StreamWriter(ctx, name, (uuid.UUID)(toMeta.BlockID), toMeta.TenantID, reader, size)
	}

	// Read entire object and attempt to cache
	cpy := func(name string, cacheInfo *backend.CacheInfo) error {
		cacheInfo.Meta = fromMeta
		b, err := from.Read(ctx, name, (uuid.UUID)(fromMeta.BlockID), fromMeta.TenantID, cacheInfo)
		if err != nil {
			return fmt.Errorf("error reading %s: %w", name, err)
		}

		cacheInfo.Meta = toMeta
		return to.Write(ctx, name, (uuid.UUID)(toMeta.BlockID), toMeta.TenantID, b, cacheInfo)
	}

	// Data
	err := copyStream(DataFileName)
	if err != nil {
		return err
	}

	// Bloom
	cacheInfo := &backend.CacheInfo{Role: cache.RoleBloom}
	for i := 0; i < common.ValidateShardCount(int(fromMeta.BloomShardCount)); i++ {
		err = cpy(common.BloomName(i), cacheInfo)
		if err != nil {
			return err
		}
	}

	// Index (may not exist)
	err = cpy(common.NameIndex, &backend.CacheInfo{Role: cache.RoleTraceIDIdx})
	if err != nil && !errors.Is(err, backend.ErrDoesNotExist) {
		return err
	}

	// no-compact flag
	if hasNoCompactFlag, err := from.HasNoCompactFlag(ctx, (uuid.UUID)(fromMeta.BlockID), fromMeta.TenantID); err != nil {
		return err
	} else if hasNoCompactFlag {
		err = to.WriteNoCompactFlag(ctx, (uuid.UUID)(toMeta.BlockID), toMeta.TenantID)
		if err != nil {
			return err
		}
	}

	// Meta
	err = to.WriteBlockMeta(ctx, toMeta)
	return err