Golang function that reads S3 files and populates maps with strings as keys

Question

I have a below read function which is called by multiple go routines to read s3 files and it populates two concurrent map as shown below.

• During server startup, it calls read function below to populate two concurrent map.
• And also periodically every 30 seconds, it calls read function again to read new s3 files and populate two concurrent map again with some new data.

So basically at a given state of time during the whole lifecycle of this app, both my concurrent map have some data and also periodically being updated too.

func (r *clientRepository) read(file string, bucket string) error {
    var err error
    //... read s3 file

    for {
        rows, err := pr.ReadByNumber(r.cfg.RowsToRead)
        if err != nil {
            return errs.Wrap(err)
        }
        if len(rows) <= 0 {
            break
        }

        byteSlice, err := json.Marshal(rows)
        if err != nil {
            return errs.Wrap(err)
        }
        var productRows []ParquetData
        err = json.Unmarshal(byteSlice, &productRows)
        if err != nil {
            return errs.Wrap(err)
        }

        for i := range productRows {
            var flatProduct definitions.CustomerInfo
            err = r.ConvertData(spn, &productRows[i], &flatProduct)
            if err != nil {
                return errs.Wrap(err)
            }

            // populate first concurrent map here
            r.products.Set(strconv.FormatInt(flatProduct.ProductId, 10), &flatProduct)
            for _, catalogId := range flatProduct.Catalogs {
                strCatalogId := strconv.FormatInt(int64(catalogId), 10)
                // upsert second concurrent map here
                r.productCatalog.Upsert(strCatalogId, flatProduct.ProductId, func(exists bool, valueInMap interface{}, newValue interface{}) interface{} {
                    productID := newValue.(int64)
                    if valueInMap == nil {
                        return map[int64]struct{}{productID: {}}
                    }
                    oldIDs := valueInMap.(map[int64]struct{})
                    // value is irrelevant, no need to check if key exists
                    oldIDs[productID] = struct{}{}
                    return oldIDs
                })
            }
        }
    }
    return nil
}

And then I have below three functions which is used by my main application threads to get data from the concurrent map populated above.

func (r *clientRepository) GetProductMap() *cmap.ConcurrentMap {
    return r.products
}

func (r *clientRepository) GetProductCatalogMap() *cmap.ConcurrentMap {
    return r.productCatalog
}

func (r *clientRepository) GetProductData(pid string) *definitions.CustomerInfo {
    pd, ok := r.products.Get(pid)
    if ok {
        return pd.(*definitions.CustomerInfo)
    }
    return nil
}

I have a use case where I need to populate map from multiple go routines and then read data from those maps from bunch of main application threads so it needs to be thread safe and it should be fast enough as well without much locking.

Problem Statement

I am dealing with lots of data like 30-40 GB worth of data from all these files which I am reading into memory. I am using concurrent map here which solves most of my concurrency issues but the key for the concurrent map is string and it doesn't have any implementation where key can be integer. In my case key is just a product id which can be int32 so is it worth it storing all those product id's as string in this concurrent map? I think string allocation takes more memory compare to storing all those keys as integer? At least it does in c/c++ so I am assuming it should be same case here in golang too.

Is there anything I can to improve here w.r.t map usage so that I can reduce memory utilization plus I don't lose performance as well while reading data from these maps from main threads?

I am using concurrent map from this repo which doesn't have implementation for key as integer.

Answer 1

I'd start by breaking up clientRepository.read into the S3 part and the actual Parquet reading part, with the latter (maybe called readParquet) taking only an io.Reader or at most a source.ParquetFile if that's what the Parquet library requires. That would make it easier to test the Parquet reading separate from the S3 data fetching, including testing performance using a local Parquet file.

The main thing that jumps out to me is that it's inefficient to JSON-marshal the rows for each read, only to unmarshal them back into Go structs again. Can that be avoided by passing a pointer to productRows []ParquetData directly to pr.Read? I'm not sure how your ParquetData struct is defined, but from this example from the Parquet library's docs, it looks like you could read directly to a Go struct slice (untested):

productRows := make([]ParquetData, r.cfg.RowsToRead)
err := pr.Read(&productRows)
if err != nil { ... }

That's much simpler and should be significantly more efficient, especially on 30-40GB of data.

Regarding the concurrent map, I'd suggest avoiding a library and using a plain Go map with sync.RWMutex locking around reads and writes. This would allow you to use an integer type for the map key too, addressing that inefficiency. Something like this:

type clientRepository struct {
    productsLock sync.RWMutex
    products     map[int]*definitions.CustomerInfo
}

func (r *clientRepository) read(file string, bucket string) error {
    // ...
    r.productsLock.Lock() // or you could lock it only once per productRow loop
    r.products[flatProduct.ProductId] = &flatProduct
    // ... update productCatalog here ...
    r.productsLock.Unlock()
    // ...
}

func (r *clientRepository) GetProductData(pid string) *definitions.CustomerInfo {
    r.productsLock.RLock()
    defer r.productsLock.RUnlock()

    pd, ok := r.products[pid]
    if ok {
        return pd.(*definitions.CustomerInfo)
    }
    return nil
}

However, that would require you to change the clientRepository GetProductMap and GetProductCatalogMap methods to return simple Go types -- but I think that's a better and more maintainable API anyway, as it avoids callers having to learn and use the ConcurrentMap API.

So I'd suggest changing GetProductMap to avoid returning the entire map. For example, what do you need in addition to GetProductData -- the list of product IDs, maybe? You can already check if an ID exists by calling GetProductData and checking that it returns non-nil. If you really need a list of product IDs, either add GetProductIDs and build it each call, or return a pre-built slice if performance is a concern (but be careful with concurrency). In any case, I'd probably try to avoid an API signature which returned essentially the entire cache.

Similar for GetProductCatalog: change it to return a single value or simple Go values like GetProductData does.

Other things

One question that comes to mind when updating the products and catalog maps: what if products are removed -- they should be removed from the map, right? If that's the case, it might be simplest to build a new map each time, and lock the mutex when you're updating the clientRepository.products field to the new map.

You might be able to save a bit of memory and pointer shuffling if you use a map[int32]CustomerInfo (without the pointer). But I'm not sure; you'd want to test it. In addition, you haven't shown the definition of your CustomerInfo struct, but you might be able to optimize the field sizes and field order in that struct, as well as change your product catalog map to use int32 keys as well if that's acceptable.

Style point: you can get both the loop index and slice value using a two-value range, like so:

for i, row := range productRows {
var flatProduct definitions.CustomerInfo
err = r.ConvertData(spn, &row, &flatProduct)
if err != nil { ... }

Style point: strconv.Itoa(catalogId) is a shortcut for strconv.FormatInt(int64(catalogId), 10)).