[dbsql] FilterFieldOrColumns for Column Index Scans rather than COALESCE Full Scans

pkg/dbsql/crud.go

@@ -148,6 +148,14 @@ type CrudBase[T Resource] struct {
 	ReadOnlyColumns   []string
 	ReadQueryModifier QueryModifier
 	AfterLoad         func(ctx context.Context, inst T) error // perform final validation/formatting after an instance is loaded from db
+
+	// FilterFieldOrColumns enables OR-expansion for filter fields that span multiple
+	// database columns (e.g. asset lookups that resolve through joined tables).
+	// When a filter operation targets one of these fields, the single condition is
+	// expanded into an OR across all listed columns, allowing the database to use
+	// individual column indexes instead of evaluating a COALESCE expression.
+	// The FilterFieldMap entry for the same field is still used for ORDER BY / GROUP BY.
+	FilterFieldOrColumns map[string][]string
 }
 
 func (c *CrudBase[T]) Scoped(scope sq.Eq) CRUD[T] {
@@ -156,6 +164,45 @@ func (c *CrudBase[T]) Scoped(scope sq.Eq) CRUD[T] {
 	return &cScoped
 }
 
+// expandOrFields rewrites the finalized filter tree so that any leaf operation
+// on a field listed in FilterFieldOrColumns is replaced with an OR across all
+// the listed columns. This lets the database use per-column indexes instead of
+// evaluating a COALESCE/expression wrapper.
+func (c *CrudBase[T]) expandOrFields(fi *ffapi.FilterInfo) *ffapi.FilterInfo {
+	if len(c.FilterFieldOrColumns) == 0 || fi == nil {
+		return fi
+	}
+	return expandOrFieldsWalk(fi, c.FilterFieldOrColumns)
+}
+
+func expandOrFieldsWalk(fi *ffapi.FilterInfo, orMap map[string][]string) *ffapi.FilterInfo {
+	if fi.Op == ffapi.FilterOpAnd || fi.Op == ffapi.FilterOpOr {
+		newChildren := make([]*ffapi.FilterInfo, len(fi.Children))
+		for i, child := range fi.Children {
+			newChildren[i] = expandOrFieldsWalk(child, orMap)
+		}
+		expanded := *fi
+		expanded.Children = newChildren
+		return &expanded
+	}
+
+	cols, ok := orMap[fi.Field]
+	if !ok || len(cols) == 0 {
+		return fi
+	}
+
+	children := make([]*ffapi.FilterInfo, len(cols))
+	for i, col := range cols {
+		child := *fi
+		child.Field = col
+		children[i] = &child
+	}
+	return &ffapi.FilterInfo{
+		Op:       ffapi.FilterOpOr,
+		Children: children,
+	}
+}
+
 func (c *CrudBase[T]) TableAlias() string {
 	if c.ReadTableAlias != "" {
 		return c.ReadTableAlias
@@ -709,6 +756,7 @@ func (c *CrudBase[T]) GetMany(ctx context.Context, filter ffapi.Filter) (instanc
 	if err != nil {
 		return nil, nil, err
 	}
+	fi = c.expandOrFields(fi)
 	tableFrom, cols, readCols := c.getReadCols(fi)
 	var preconditions []sq.Sqlizer
 	if c.ScopedFilter != nil {
@@ -819,6 +867,7 @@ func (c *CrudBase[T]) Count(ctx context.Context, filter ffapi.Filter) (count int
 	var fop sq.Sqlizer
 	fi, err := filter.Finalize()
 	if err == nil {
+		fi = c.expandOrFields(fi)
 		fop, err = c.DB.filterOp(ctx, c.Table, fi, c.FilterFieldMap)
 	}
 	if err != nil {

pkg/dbsql/crud_test.go

@@ -1525,3 +1525,69 @@ func TestCustomIDColumn(t *testing.T) {
 	}
 	tc.Validate()
 }
+
+func TestExpandOrFieldsLeafExpansion(t *testing.T) {
+	sf := (&ffapi.StringField{}).GetSerialization()
+	_ = sf.Scan("abc")
+
+	fi := &ffapi.FilterInfo{
+		Op:    ffapi.FilterOpEq,
+		Field: "asset",
+		Value: sf,
+	}
+	orMap := map[string][]string{
+		"asset": {"data.asset_id", "pools.asset", "nfts.asset"},
+	}
+	result := expandOrFieldsWalk(fi, orMap)
+	assert.Equal(t, ffapi.FilterOpOr, result.Op)
+	require.Len(t, result.Children, 3)
+	assert.Equal(t, "data.asset_id", result.Children[0].Field)
+	assert.Equal(t, "pools.asset", result.Children[1].Field)
+	assert.Equal(t, "nfts.asset", result.Children[2].Field)
+	for _, child := range result.Children {
+		assert.Equal(t, ffapi.FilterOpEq, child.Op)
+		v, _ := child.Value.Value()
+		assert.Equal(t, "abc", v)
+	}
+}
+
+func TestExpandOrFieldsInsideAnd(t *testing.T) {
+	sf1 := (&ffapi.StringField{}).GetSerialization()
+	_ = sf1.Scan("x")
+	sf2 := (&ffapi.StringField{}).GetSerialization()
+	_ = sf2.Scan("y")
+
+	fi := &ffapi.FilterInfo{
+		Op: ffapi.FilterOpAnd,
+		Children: []*ffapi.FilterInfo{
+			{Op: ffapi.FilterOpEq, Field: "asset", Value: sf1},
+			{Op: ffapi.FilterOpEq, Field: "name", Value: sf2},
+		},
+	}
+	orMap := map[string][]string{"asset": {"col1", "col2"}}
+	result := expandOrFieldsWalk(fi, orMap)
+	assert.Equal(t, ffapi.FilterOpAnd, result.Op)
+	require.Len(t, result.Children, 2)
+	assert.Equal(t, ffapi.FilterOpOr, result.Children[0].Op)
+	require.Len(t, result.Children[0].Children, 2)
+	assert.Equal(t, ffapi.FilterOpEq, result.Children[1].Op)
+	assert.Equal(t, "name", result.Children[1].Field)
+}
+
+func TestExpandOrFieldsNilAndEmpty(t *testing.T) {
+	crud := &CrudBase[*TestCRUDable]{}
+	assert.Nil(t, crud.expandOrFields(nil))
+
+	fi := &ffapi.FilterInfo{Op: ffapi.FilterOpEq, Field: "f1"}
+	assert.Same(t, fi, crud.expandOrFields(fi))
+}
+
+func TestExpandOrFieldsNoMatch(t *testing.T) {
+	sf := (&ffapi.StringField{}).GetSerialization()
+	_ = sf.Scan("z")
+
+	fi := &ffapi.FilterInfo{Op: ffapi.FilterOpEq, Field: "name", Value: sf}
+	orMap := map[string][]string{"asset": {"col1", "col2"}}
+	result := expandOrFieldsWalk(fi, orMap)
+	assert.Same(t, fi, result)
+}