add support for configurable dispatch chunk size

we've observed improvements on LR workloads when
increasing the dispatch chunk size

internal/caveats/run.go

@@ -221,19 +221,11 @@ func runExpressionWithCaveats(
 	var exprStringPieces []string
 
 	var currentResult ExpressionResult = syntheticResult{
-		value:                false,
+		value:                cop.Op == core.CaveatOperation_AND,
 		contextValues:        map[string]any{},
 		exprString:           "",
 		missingContextParams: []string{},
 	}
-	if cop.Op == core.CaveatOperation_AND {
-		currentResult = syntheticResult{
-			value:                true,
-			contextValues:        map[string]any{},
-			exprString:           "",
-			missingContextParams: []string{},
-		}
-	}
 
 	buildExprString := func() (string, error) {
 		if debugOption != RunCaveatExpressionWithDebugInformation {

internal/datastore/common/sql.go

@@ -102,14 +102,16 @@ type SchemaQueryFilterer struct {
 	schema                SchemaInformation
 	queryBuilder          sq.SelectBuilder
 	filteringColumnCounts map[string]int
+	filterMaximumIDCount  uint16
 }
 
 // NewSchemaQueryFilterer creates a new SchemaQueryFilterer object.
-func NewSchemaQueryFilterer(schema SchemaInformation, initialQuery sq.SelectBuilder) SchemaQueryFilterer {
+func NewSchemaQueryFilterer(schema SchemaInformation, initialQuery sq.SelectBuilder, filterMaximumIDCount uint16) SchemaQueryFilterer {
 	return SchemaQueryFilterer{
 		schema:                schema,
 		queryBuilder:          initialQuery,
 		filteringColumnCounts: map[string]int{},
+		filterMaximumIDCount:  filterMaximumIDCount,
 	}
 }
 
@@ -306,8 +308,8 @@ func (sqf SchemaQueryFilterer) MustFilterWithResourceIDPrefix(prefix string) Sch
 // FilterToResourceIDs returns a new SchemaQueryFilterer that is limited to resources with any of the
 // specified IDs.
 func (sqf SchemaQueryFilterer) FilterToResourceIDs(resourceIds []string) (SchemaQueryFilterer, error) {
-	if len(resourceIds) > int(datastore.FilterMaximumIDCount) {
-		return sqf, spiceerrors.MustBugf("cannot have more than %d resources IDs in a single filter", datastore.FilterMaximumIDCount)
+	if len(resourceIds) > int(sqf.filterMaximumIDCount) {
+		return sqf, spiceerrors.MustBugf("cannot have more than %d resources IDs in a single filter", sqf.filterMaximumIDCount)
 	}
 
 	var builder strings.Builder
@@ -422,8 +424,8 @@ func (sqf SchemaQueryFilterer) FilterWithSubjectsSelectors(selectors ...datastor
 		}
 
 		if len(selector.OptionalSubjectIds) > 0 {
-			if len(selector.OptionalSubjectIds) > int(datastore.FilterMaximumIDCount) {
-				return sqf, spiceerrors.MustBugf("cannot have more than %d subject IDs in a single filter", datastore.FilterMaximumIDCount)
+			if len(selector.OptionalSubjectIds) > int(sqf.filterMaximumIDCount) {
+				return sqf, spiceerrors.MustBugf("cannot have more than %d subject IDs in a single filter", sqf.filterMaximumIDCount)
 			}
 
 			var builder strings.Builder

internal/datastore/common/sql_test.go

@@ -669,7 +669,7 @@ func TestSchemaQueryFilterer(t *testing.T) {
 				"caveat",
 				TupleComparison,
 			)
-			filterer := NewSchemaQueryFilterer(schema, base)
+			filterer := NewSchemaQueryFilterer(schema, base, 100)
 
 			ran := test.run(filterer)
 			require.Equal(t, test.expectedColumnCounts, ran.filteringColumnCounts)
@@ -693,7 +693,7 @@ func BenchmarkSchemaFilterer(b *testing.B) {
 		"caveat_name",
 		TupleComparison,
 	)
-	sqf := NewSchemaQueryFilterer(si, sq.Select("*"))
+	sqf := NewSchemaQueryFilterer(si, sq.Select("*"), 100)
 	var names []string
 	for i := 0; i < 500; i++ {
 		names = append(names, uuid.NewString())

internal/datastore/crdb/crdb.go

@@ -275,9 +275,10 @@ type crdbDatastore struct {
 
 	featureGroup singleflight.Group[string, *datastore.Features]
 
-	pruneGroup *errgroup.Group
-	ctx        context.Context
-	cancel     context.CancelFunc
+	pruneGroup           *errgroup.Group
+	ctx                  context.Context
+	cancel               context.CancelFunc
+	filterMaximumIDCount uint16
 }
 
 func (cds *crdbDatastore) SnapshotReader(rev datastore.Revision) datastore.Reader {
@@ -289,7 +290,7 @@ func (cds *crdbDatastore) SnapshotReader(rev datastore.Revision) datastore.Reade
 		return query.From(fromStr + " AS OF SYSTEM TIME " + rev.String())
 	}
 
-	return &crdbReader{cds.readPool, executor, noOverlapKeyer, nil, fromBuilder}
+	return &crdbReader{cds.readPool, executor, noOverlapKeyer, nil, fromBuilder, cds.filterMaximumIDCount}
 }
 
 func (cds *crdbDatastore) ReadWriteTx(
@@ -319,6 +320,7 @@ func (cds *crdbDatastore) ReadWriteTx(
 				func(query sq.SelectBuilder, fromStr string) sq.SelectBuilder {
 					return query.From(fromStr)
 				},
+				cds.filterMaximumIDCount,
 			},
 			tx,
 			0,

internal/datastore/crdb/options.go

@@ -22,8 +22,8 @@ type crdbOptions struct {
 	overlapKey                  string
 	enableConnectionBalancing   bool
 	analyzeBeforeStatistics     bool
-
-	enablePrometheusStats bool
+	filterMaximumIDCount        uint16
+	enablePrometheusStats       bool
 }
 
 const (
@@ -48,6 +48,7 @@ const (
 	defaultEnablePrometheusStats     = false
 	defaultEnableConnectionBalancing = true
 	defaultConnectRate               = 100 * time.Millisecond
+	defaultFilterMaximumIDCount      = 100
 )
 
 // Option provides the facility to configure how clients within the CRDB
@@ -68,6 +69,7 @@ func generateConfig(options []Option) (crdbOptions, error) {
 		enablePrometheusStats:       defaultEnablePrometheusStats,
 		enableConnectionBalancing:   defaultEnableConnectionBalancing,
 		connectRate:                 defaultConnectRate,
+		filterMaximumIDCount:        defaultFilterMaximumIDCount,
 	}
 
 	for _, option := range options {
@@ -306,3 +308,8 @@ func WithEnableConnectionBalancing(connectionBalancing bool) Option {
 func DebugAnalyzeBeforeStatistics() Option {
 	return func(po *crdbOptions) { po.analyzeBeforeStatistics = true }
 }
+
+// FilterMaximumIDCount is the maximum number of IDs that can be used to filter IDs in queries
+func FilterMaximumIDCount(filterMaximumIDCount uint16) Option {
+	return func(po *crdbOptions) { po.filterMaximumIDCount = filterMaximumIDCount }
+}

internal/datastore/crdb/reader.go

@@ -60,11 +60,12 @@ var (
 )
 
 type crdbReader struct {
-	query         pgxcommon.DBFuncQuerier
-	executor      common.QueryExecutor
-	keyer         overlapKeyer
-	overlapKeySet keySet
-	fromBuilder   func(query sq.SelectBuilder, fromStr string) sq.SelectBuilder
+	query                pgxcommon.DBFuncQuerier
+	executor             common.QueryExecutor
+	keyer                overlapKeyer
+	overlapKeySet        keySet
+	fromBuilder          func(query sq.SelectBuilder, fromStr string) sq.SelectBuilder
+	filterMaximumIDCount uint16
 }
 
 func (cr *crdbReader) CountRelationships(ctx context.Context, name string) (int, error) {
@@ -83,7 +84,7 @@ func (cr *crdbReader) CountRelationships(ctx context.Context, name string) (int,
 	}
 
 	query := cr.fromBuilder(countTuples, tableTuple)
-	builder, err := common.NewSchemaQueryFilterer(schema, query).FilterWithRelationshipsFilter(relFilter)
+	builder, err := common.NewSchemaQueryFilterer(schema, query, cr.filterMaximumIDCount).FilterWithRelationshipsFilter(relFilter)
 	if err != nil {
 		return 0, err
 	}
@@ -208,7 +209,7 @@ func (cr *crdbReader) QueryRelationships(
 	opts ...options.QueryOptionsOption,
 ) (iter datastore.RelationshipIterator, err error) {
 	query := cr.fromBuilder(queryTuples, tableTuple)
-	qBuilder, err := common.NewSchemaQueryFilterer(schema, query).FilterWithRelationshipsFilter(filter)
+	qBuilder, err := common.NewSchemaQueryFilterer(schema, query, cr.filterMaximumIDCount).FilterWithRelationshipsFilter(filter)
 	if err != nil {
 		return nil, err
 	}
@@ -222,7 +223,7 @@ func (cr *crdbReader) ReverseQueryRelationships(
 	opts ...options.ReverseQueryOptionsOption,
 ) (iter datastore.RelationshipIterator, err error) {
 	query := cr.fromBuilder(queryTuples, tableTuple)
-	qBuilder, err := common.NewSchemaQueryFilterer(schema, query).
+	qBuilder, err := common.NewSchemaQueryFilterer(schema, query, cr.filterMaximumIDCount).
 		FilterWithSubjectsSelectors(subjectsFilter.AsSelector())
 	if err != nil {
 		return nil, err

internal/datastore/mysql/datastore.go

@@ -319,6 +319,7 @@ func (mds *Datastore) SnapshotReader(rev datastore.Revision) datastore.Reader {
 		createTxFunc,
 		executor,
 		buildLivingObjectFilterForRevision(rev),
+		mds.filterMaximumIDCount,
 	}
 }
 
@@ -356,6 +357,7 @@ func (mds *Datastore) ReadWriteTx(
 					longLivedTx,
 					executor,
 					currentlyLivingObjects,
+					mds.filterMaximumIDCount,
 				},
 				mds.driver.RelationTuple(),
 				tx,
@@ -484,6 +486,7 @@ type Datastore struct {
 	watchBufferLength       uint16
 	watchBufferWriteTimeout time.Duration
 	maxRetries              uint8
+	filterMaximumIDCount    uint16
 
 	optimizedRevisionQuery string
 	validTransactionQuery  string

internal/datastore/mysql/options.go

@@ -22,6 +22,7 @@ const (
 	defaultMaxRetries                        = 8
 	defaultGCEnabled                         = true
 	defaultCredentialsProviderName           = ""
+	defaultFilterMaximumIDCount              = 100
 )
 
 type mysqlOptions struct {
@@ -42,6 +43,7 @@ type mysqlOptions struct {
 	lockWaitTimeoutSeconds      *uint8
 	gcEnabled                   bool
 	credentialsProviderName     string
+	filterMaximumIDCount        uint16
 }
 
 // Option provides the facility to configure how clients within the
@@ -64,6 +66,7 @@ func generateConfig(options []Option) (mysqlOptions, error) {
 		maxRetries:                  defaultMaxRetries,
 		gcEnabled:                   defaultGCEnabled,
 		credentialsProviderName:     defaultCredentialsProviderName,
+		filterMaximumIDCount:        defaultFilterMaximumIDCount,
 	}
 
 	for _, option := range options {
@@ -247,3 +250,8 @@ func GCMaxOperationTime(time time.Duration) Option {
 func CredentialsProviderName(credentialsProviderName string) Option {
 	return func(mo *mysqlOptions) { mo.credentialsProviderName = credentialsProviderName }
 }
+
+// FilterMaximumIDCount is the maximum number of IDs that can be used to filter IDs in queries
+func FilterMaximumIDCount(filterMaximumIDCount uint16) Option {
+	return func(mo *mysqlOptions) { mo.filterMaximumIDCount = filterMaximumIDCount }
+}

internal/datastore/mysql/reader.go

@@ -22,9 +22,10 @@ type txFactory func(context.Context) (*sql.Tx, txCleanupFunc, error)
 type mysqlReader struct {
 	*QueryBuilder
 
-	txSource txFactory
-	executor common.QueryExecutor
-	filterer queryFilterer
+	txSource             txFactory
+	executor             common.QueryExecutor
+	filterer             queryFilterer
+	filterMaximumIDCount uint16
 }
 
 type queryFilterer func(original sq.SelectBuilder) sq.SelectBuilder
@@ -65,7 +66,7 @@ func (mr *mysqlReader) CountRelationships(ctx context.Context, name string) (int
 		return 0, err
 	}
 
-	qBuilder, err := common.NewSchemaQueryFilterer(schema, mr.filterer(mr.CountTupleQuery)).FilterWithRelationshipsFilter(relFilter)
+	qBuilder, err := common.NewSchemaQueryFilterer(schema, mr.filterer(mr.CountTupleQuery), mr.filterMaximumIDCount).FilterWithRelationshipsFilter(relFilter)
 	if err != nil {
 		return 0, err
 	}
@@ -174,7 +175,7 @@ func (mr *mysqlReader) QueryRelationships(
 	filter datastore.RelationshipsFilter,
 	opts ...options.QueryOptionsOption,
 ) (iter datastore.RelationshipIterator, err error) {
-	qBuilder, err := common.NewSchemaQueryFilterer(schema, mr.filterer(mr.QueryTuplesQuery)).FilterWithRelationshipsFilter(filter)
+	qBuilder, err := common.NewSchemaQueryFilterer(schema, mr.filterer(mr.QueryTuplesQuery), mr.filterMaximumIDCount).FilterWithRelationshipsFilter(filter)
 	if err != nil {
 		return nil, err
 	}
@@ -187,7 +188,7 @@ func (mr *mysqlReader) ReverseQueryRelationships(
 	subjectsFilter datastore.SubjectsFilter,
 	opts ...options.ReverseQueryOptionsOption,
 ) (iter datastore.RelationshipIterator, err error) {
-	qBuilder, err := common.NewSchemaQueryFilterer(schema, mr.filterer(mr.QueryTuplesQuery)).
+	qBuilder, err := common.NewSchemaQueryFilterer(schema, mr.filterer(mr.QueryTuplesQuery), mr.filterMaximumIDCount).
 		FilterWithSubjectsSelectors(subjectsFilter.AsSelector())
 	if err != nil {
 		return nil, err

internal/datastore/postgres/options.go

@@ -21,6 +21,7 @@ type postgresOptions struct {
 	gcInterval              time.Duration
 	gcMaxOperationTime      time.Duration
 	maxRetries              uint8
+	filterMaximumIDCount    uint16
 
 	enablePrometheusStats   bool
 	analyzeBeforeStatistics bool
@@ -63,6 +64,7 @@ const (
 	defaultGCEnabled                         = true
 	defaultCredentialsProviderName           = ""
 	defaultReadStrictMode                    = false
+	defaultFilterMaximumIDCount              = 100
 )
 
 // Option provides the facility to configure how clients within the
@@ -84,6 +86,7 @@ func generateConfig(options []Option) (postgresOptions, error) {
 		credentialsProviderName:     defaultCredentialsProviderName,
 		readStrictMode:              defaultReadStrictMode,
 		queryInterceptor:            nil,
+		filterMaximumIDCount:        defaultFilterMaximumIDCount,
 	}
 
 	for _, option := range options {
@@ -356,3 +359,8 @@ func MigrationPhase(phase string) Option {
 func CredentialsProviderName(credentialsProviderName string) Option {
 	return func(po *postgresOptions) { po.credentialsProviderName = credentialsProviderName }
 }
+
+// FilterMaximumIDCount is the maximum number of IDs that can be used to filter IDs in queries
+func FilterMaximumIDCount(filterMaximumIDCount uint16) Option {
+	return func(po *postgresOptions) { po.filterMaximumIDCount = filterMaximumIDCount }
+}

internal/datastore/postgres/postgres.go

@@ -330,6 +330,7 @@ func newPostgresDatastore(
 		credentialsProvider:     credentialsProvider,
 		isPrimary:               isPrimary,
 		inStrictReadMode:        config.readStrictMode,
+		filterMaximumIDCount:    config.filterMaximumIDCount,
 	}
 
 	if isPrimary && config.readStrictMode {
@@ -384,10 +385,11 @@ type pgDatastore struct {
 
 	credentialsProvider datastore.CredentialsProvider
 
-	gcGroup  *errgroup.Group
-	gcCtx    context.Context
-	cancelGc context.CancelFunc
-	gcHasRun atomic.Bool
+	gcGroup              *errgroup.Group
+	gcCtx                context.Context
+	cancelGc             context.CancelFunc
+	gcHasRun             atomic.Bool
+	filterMaximumIDCount uint16
 }
 
 func (pgd *pgDatastore) IsStrictReadModeEnabled() bool {
@@ -410,6 +412,7 @@ func (pgd *pgDatastore) SnapshotReader(revRaw datastore.Revision) datastore.Read
 		queryFuncs,
 		executor,
 		buildLivingObjectFilterForRevision(rev),
+		pgd.filterMaximumIDCount,
 	}
 }
 
@@ -447,6 +450,7 @@ func (pgd *pgDatastore) ReadWriteTx(
 					queryFuncs,
 					executor,
 					currentlyLivingObjects,
+					pgd.filterMaximumIDCount,
 				},
 				tx,
 				newXID,