create table from source: source limits are not applied

[nrainer-materialize]

What version of Materialize are you using?

e8e6737

What is the issue?

Given

$ postgres-execute connection=mz_system
ALTER SYSTEM SET max_sources = 2
ALTER SYSTEM SET max_tables = 2

> CREATE SOURCE mz_source FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source');
> CREATE SOURCE mz_source2 FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source');

the statement

> CREATE SOURCE mz_source3 FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source');

should fail with "creating source would violate max_sources limit (desired: 3, limit: 2, current: 0)" but does not.

[rjobanp]

@nrainer-materialize I don't know that this is an issue, or at least it's working as desired based on the current implementation:

The limit on sources actually appears to be a limit on the number of 'persist-shards' in-use by a source:

materialize/src/adapter/src/coord/ddl.rs

Lines 1562 to 1577 in 598b74a

	let current_sources: usize = self
	.catalog()
	.user_sources()
	.filter_map(|source| source.source())
	.map(|source| source.user_controllable_persist_shard_count())
	.sum::<i64>()
	.try_into()
	.expect("non-negative sum of sources");
	self.validate_resource_limit(
	current_sources,
	new_sources,
	SystemVars::max_sources,
	"source",
	MAX_SOURCES.name(),
	)?;

The limitation uses this method to determine how many 'persist shards' are actually needed by a given source:

materialize/src/catalog/src/memory/objects.rs

Lines 793 to 832 in 598b74a

	/// The expensive resource that each source consumes is persist shards. To
	/// prevent abuse, we want to prevent users from creating sources that use an
	/// unbounded number of persist shards. But we also don't want to count
	/// persist shards that are mandated by teh system (e.g., the progress
	/// shard) so that future versions of Materialize can introduce additional
	/// per-source shards (e.g., a per-source status shard) without impacting
	/// the limit calculation.
	pub fn user_controllable_persist_shard_count(&self) -> i64 {
	match &self.data_source {
	DataSourceDesc::Ingestion { ingestion_desc, .. } => {
	match &ingestion_desc.desc.connection {
	// These multi-output sources do not use their primary
	// source's data shard, so we don't include it in accounting
	// for users.
	GenericSourceConnection::Postgres(_) | GenericSourceConnection::MySql(_) => 0,
	GenericSourceConnection::LoadGenerator(lg) => {
	// TODO: make this a method on the load generator.
	if lg.load_generator.views().is_empty() {
	// Load generator writes directly to its persist shard
	1
	} else {
	// Load generator has 1 persist shard per output,
	// which will be accounted for by `SourceExport`.
	0
	}
	}
	GenericSourceConnection::Kafka(_) => 1,
	}
	}
	// DataSourceDesc::IngestionExport represents a subsource, which
	// use a data shard.
	DataSourceDesc::IngestionExport { .. } => 1,
	DataSourceDesc::Webhook { .. } => 1,
	// Introspection and progress subsources are not under the user's control, so shouldn't
	// count toward their quota.
	DataSourceDesc::Introspection(_) | DataSourceDesc::Progress => 0,
	}
	}
	}

Given that the CREATE SOURCE mz_source FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source'); statement does not create any subsources, there are no data shards being created for the statement, such that it doesn't affect the overall limit.

If we want to make this limit sources based on the number of source-statements, then we would have to change this logic, but since I did not implement or design the original behavior I am not sure that is something we want. @morsapaes do you have an opinion on this?