add CollectRowsWithFilter

[piusalfred]

Recently I have a case where after collecting rows using pgx.CollectRows, I needed to iterate over the result and filter out nil values which are basically the values that did not pass my custom filter after scan.

func (p *PostgresEventStore) Load(ctx context.Context, aggregateID uuid.UUID, predicate domain.EventPredicate) ([]domain.Event, error) {
    // omitted for brevity
    rows, err := p.pool.Query(ctx, stmt, args...)
    if err != nil {
        return nil, err
    }

    pgEvents, err := pgx.CollectRows[domain.Event](rows, func(row pgx.CollectableRow) (domain.Event, error) {
        var e PgEvent
        if scanErr := row.Scan(//....); scanErr != nil {
            return nil, err
        }

        // If it doesn't pass the predicate, we return nil
        if predicate != nil && !predicate(e) {
            return nil, nil
        }

        return e, nil
    })
    if err != nil {
        return nil, err
    }

    // SECOND PASS: filter out the nil items
    var events []domain.Event
    for _, e := range pgEvents {
        if e != nil {
            events = append(events, e)
        }
    }

    return events, nil
}

I believe it will be good to have a new function, for example pgx.CollectRowsWithFilter that extends pgx.CollectRows by also accepting a filter func func(T)bool meaning we’d handle everything in one pass. The above snippet becomes

func (p *PostgresEventStore) Load(ctx context.Context, aggregateID uuid.UUID, predicate domain.EventPredicate) ([]domain.Event, error) {
    // omitted for brevity
    rows, err := p.pool.Query(ctx, stmt, args...)
    if err != nil {
        return nil, fmt.Errorf("query events: %w", err)
    }

    pgEvents, err := pgx.CollectRowsWithFilter[domain.Event](
        rows,
        // Step 1: How to collect/scan each row
        func(row pgx.CollectableRow) (domain.Event, error) {
            var e PgEvent
            if scanErr := row.Scan(//...); scanErr != nil {
                return nil, err
            }
            return e, nil
        },
        // Step 2: Decide whether to keep the scanned event
        func(evt domain.Event) bool {return predicate(evt)},
    )
    if err != nil {
        return nil, err
    }

    // No second pass needed
    return pgEvents, nil
}

maybe we can have something like this

func AppendRowsWithFilter[T any, S ~[]T](slice S, rows Rows, fn RowToFunc[T], filter func(T)bool) (S, error){
	defer rows.Close()

	for rows.Next() {
		value, err := fn(rows)
		if err != nil {
			return nil, err
		}
                if filter == nil{
                    slice = append(slice,value)
                    continue
                }
		if filter(value) {
			slice = append(slice, value)
		}
	}

	if err := rows.Err(); err != nil {
		return nil, err
	}

	return slice, nil
}

func CollectRowsWithFilter[T any](rows Rows, fn RowToFunc[T], filter func(T)bool) ([]T, error){
	return AppendRowsWithFilter([]T{}, rows, fn, filter)
}

skipping items as soon as they fail a filter pass can improve efficiency by avoiding storing and then re-iterating the unwanted rows.

[guidog]

Forgive me, but why not just using a decent WHERE clause?

[piusalfred]

Honestly, a WHERE clause can work in some cases. But in this case domain.Event is an interface with multiple concrete implementations. Hence you can imagine there is a possibility for filters to be much more than some trivial checks, makes it hard to dynamically build the SQL query.

[jackc]

This seems to be a fairly rare use case. I think I'd like to avoid adding these to the public interface.

At some point, pgx will probably add iterator support for Rows. There is a working PR now, but it's on hold until I get a better feel for how custom iterators. Once that is in place application code will be able to use generic filters and maps without pgx needing to directly support it.

[piusalfred]

Thanks for the feedback @jackc Can you help direct me to the PR?

[jackc]

#2241

[piusalfred]

Thanks. I will go on and closes this.