Merge #144955 #145288

144955: logictest: add test cases for RLS interaction with triggers r=rafiss a=rafiss

Epic: CRDB-48807
Release note: None

145288: storage: minor cleanups around pebble compaction concurrency r=RaduBerinde a=RaduBerinde

#### storage: remove AdjustCompactionConcurrency

This is not used.

Epic: none
Release note: None

#### storage: remove unnecessary initialization in newPebble

We now only call this function from Open and we always have options
set.

Epic: none
Release note: None

#### pebble: separate compaction concurrency override logic

Separate the logic that allows temporarily overriding compaction
concurrency.

Epic: none
Release note: None

#### storage: cleanup default max concurrent compactions code

Refactor this code so we statically initialize the default value.

Epic: none
Release note: None

Co-authored-by: Rafi Shamim <[email protected]>
Co-authored-by: Radu Berinde <[email protected]>

Author: 3 people

pkg/kv/kvserver/stores_server.go

@@ -191,19 +191,21 @@ func (is Server) ScanStorageInternalKeys(
 // SetCompactionConcurrency implements PerStoreServer. It changes the compaction
 // concurrency of a store. While SetCompactionConcurrency is safe for concurrent
 // use, it adds uncertainty about the compaction concurrency actually set on
-// the store. It also adds uncertainty about the compaction concurrency set on
-// the store once the request is cancelled.
+// the store. We do guarantee that once all SetCompactionConcurrency requests
+// are finished (cancelled), the override is removed and the original
+// concurrency is restored.
 func (is Server) SetCompactionConcurrency(
 	ctx context.Context, req *CompactionConcurrencyRequest,
 ) (*CompactionConcurrencyResponse, error) {
 	resp := &CompactionConcurrencyResponse{}
 	err := is.execStoreCommand(ctx, req.StoreRequestHeader,
 		func(ctx context.Context, s *Store) error {
-			prevConcurrency := s.TODOEngine().SetCompactionConcurrency(req.CompactionConcurrency)
+			s.TODOEngine().SetCompactionConcurrency(req.CompactionConcurrency)
 
-			// Wait for cancellation, and once cancelled, reset the compaction concurrency.
+			// Wait for cancellation, and once cancelled, reset the compaction
+			// concurrency.
 			<-ctx.Done()
-			s.TODOEngine().SetCompactionConcurrency(prevConcurrency)
+			s.TODOEngine().SetCompactionConcurrency(0)
 			return nil
 		})
 	return resp, err

pkg/sql/logictest/testdata/logic_test/row_level_security

@@ -4485,3 +4485,267 @@ statement ok
 DROP ROLE parent_role
 
 subtest end
+
+subtest triggers_with_rls
+
+statement ok
+CREATE TABLE trigger_rls_table (id INT PRIMARY KEY, value INT, owner STRING);
+
+statement ok
+INSERT INTO trigger_rls_table VALUES (1, 100, 'alice'), (2, 200, 'bob'), (3, 300, 'alice'), (4, 400, 'bob');
+
+statement ok
+CREATE USER alice;
+
+statement ok
+CREATE USER bob;
+
+statement ok
+GRANT ALL ON trigger_rls_table TO alice, bob;
+
+# Enable RLS on the table
+statement ok
+ALTER TABLE trigger_rls_table ENABLE ROW LEVEL SECURITY;
+
+# Create a policy that only allows users to see their own data.
+statement ok
+CREATE POLICY owner_policy ON trigger_rls_table
+  USING (owner = current_user())
+  WITH CHECK (owner = current_user());
+
+# Create a trigger function that logs changes to the table.
+statement ok
+CREATE FUNCTION log_value_changes() RETURNS TRIGGER LANGUAGE PLPGSQL AS $$
+BEGIN
+  IF TG_OP = 'UPDATE' THEN
+    RAISE NOTICE 'User % updated value for id % from % to %', current_user(), (NEW).id, (OLD).value, (NEW).value;
+  ELSIF TG_OP = 'INSERT' THEN
+    RAISE NOTICE 'User % inserted new row with id % and value %', current_user(), (NEW).id, (NEW).value;
+  ELSIF TG_OP = 'DELETE' THEN
+    RAISE NOTICE 'User % deleted row with id % and value %', current_user(), (OLD).id, (OLD).value;
+  END IF;
+  RETURN NEW;
+END;
+$$;
+
+# Create a trigger function that enforces a business rule: value must be positive.
+statement ok
+CREATE FUNCTION enforce_positive_value() RETURNS TRIGGER LANGUAGE PLPGSQL AS $$
+BEGIN
+  IF (NEW).value <= 0 THEN
+    RAISE EXCEPTION 'Value must be positive';
+  END IF;
+  RETURN NEW;
+END;
+$$;
+
+# Create a trigger that logs changes.
+statement ok
+CREATE TRIGGER log_changes
+  AFTER INSERT OR UPDATE OR DELETE ON trigger_rls_table
+  FOR EACH ROW
+  EXECUTE FUNCTION log_value_changes();
+
+# Create a trigger that enforces the business rule.
+statement ok
+CREATE TRIGGER check_positive_value
+  BEFORE INSERT OR UPDATE ON trigger_rls_table
+  FOR EACH ROW
+  EXECUTE FUNCTION enforce_positive_value();
+
+# Test as alice.
+statement ok
+SET ROLE alice;
+
+# Alice should only see her own rows.
+query IIT
+SELECT * FROM trigger_rls_table ORDER BY id;
+----
+1  100  alice
+3  300  alice
+
+# Update a row that alice owns.
+query T noticetrace
+UPDATE trigger_rls_table SET value = 150 WHERE id = 1;
+----
+NOTICE: User alice updated value for id 1 from 100 to 150
+
+# Try to update a row that bob owns (should not error, but should not affect any rows).
+query T noticetrace
+UPDATE trigger_rls_table SET value = 250 WHERE id = 2;
+----
+
+# Test negative value should trigger error.
+statement error pq: Value must be positive
+UPDATE trigger_rls_table SET value = -100 WHERE id = 1;
+
+# Insert a new row that alice owns.
+query T noticetrace
+INSERT INTO trigger_rls_table VALUES (5, 500, 'alice');
+----
+NOTICE: User alice inserted new row with id 5 and value 500
+
+# Try to insert a row that bob owns (should fail due to RLS policy).
+statement error pq: new row violates row-level security policy for table "trigger_rls_table"
+INSERT INTO trigger_rls_table VALUES (6, 600, 'bob');
+
+# Delete a row that alice owns.
+query T noticetrace
+DELETE FROM trigger_rls_table WHERE id = 3;
+----
+NOTICE: User alice deleted row with id 3 and value 300
+
+# Test as bob.
+statement ok
+SET ROLE bob;
+
+# Bob should only see his own rows.
+query IIT
+SELECT * FROM trigger_rls_table ORDER BY id;
+----
+2  200  bob
+4  400  bob
+
+# Update a row that bob owns.
+query T noticetrace
+UPDATE trigger_rls_table SET value = 250 WHERE id = 2;
+----
+NOTICE: User bob updated value for id 2 from 200 to 250
+
+# Try to update a row that alice owns (should not error, but should not affect any rows).
+query T noticetrace
+UPDATE trigger_rls_table SET value = 175 WHERE id = 1;
+----
+
+# Insert a new row that bob owns.
+query T noticetrace
+INSERT INTO trigger_rls_table VALUES (7, 700, 'bob');
+----
+NOTICE: User bob inserted new row with id 7 and value 700
+
+# Test if bob can bypass RLS with a trigger function that modifies other rows.
+statement ok
+CREATE FUNCTION try_bypass_rls() RETURNS TRIGGER LANGUAGE PLPGSQL AS $$
+BEGIN
+  UPDATE trigger_rls_table SET value = 999 WHERE owner = 'alice';
+  RETURN NEW;
+END;
+$$;
+
+statement ok
+CREATE TRIGGER bypass_attempt
+  AFTER INSERT ON trigger_rls_table
+  FOR EACH ROW
+  EXECUTE FUNCTION try_bypass_rls();
+
+# Try to trigger the bypass (this insert will succeed but the trigger's UPDATE should be limited by RLS).
+statement ok
+INSERT INTO trigger_rls_table VALUES (8, 800, 'bob');
+
+# Check as root what happened.
+statement ok
+SET ROLE root;
+
+# Alice's rows should not have been affected by bob's trigger.
+query IIT
+SELECT * FROM trigger_rls_table WHERE owner = 'alice' ORDER BY id;
+----
+1  150  alice
+5  500  alice
+
+# Now test a BEFORE trigger that modifies a column in a way that would violate an RLS policy.
+statement ok
+SET ROLE alice;
+
+# Create a trigger function that tries to change the owner column to 'bob'.
+statement ok
+CREATE FUNCTION change_owner_to_bob() RETURNS TRIGGER LANGUAGE PLPGSQL AS $$
+BEGIN
+  NEW.owner = 'bob';
+  RETURN NEW;
+END;
+$$;
+
+# Create a trigger that applies the function before insert.
+statement ok
+CREATE TRIGGER force_bob_ownership
+  BEFORE INSERT ON trigger_rls_table
+  FOR EACH ROW
+  EXECUTE FUNCTION change_owner_to_bob();
+
+# Try to insert a row - this should fail because the trigger modifies 
+# the owner to 'bob' which violates alice's RLS policy.
+statement error pq: new row violates row-level security policy for table "trigger_rls_table"
+INSERT INTO trigger_rls_table VALUES (9, 900, 'alice');
+
+# Now do the same with update.
+statement ok
+CREATE FUNCTION change_updated_owner_to_bob() RETURNS TRIGGER LANGUAGE PLPGSQL AS $$
+BEGIN
+  IF TG_OP = 'UPDATE' THEN
+    NEW.owner = 'bob';
+  END IF;
+  RETURN NEW;
+END;
+$$;
+
+statement ok
+CREATE TRIGGER force_update_bob_ownership
+  BEFORE UPDATE ON trigger_rls_table
+  FOR EACH ROW
+  EXECUTE FUNCTION change_updated_owner_to_bob();
+
+# Try to update a row - this should fail because the trigger modifies
+# the owner to 'bob' which violates alice's RLS policy.
+statement error pq: new row violates row-level security policy for table "trigger_rls_table"
+UPDATE trigger_rls_table SET value = 600 WHERE id = 5;
+
+# Let's try with both bob and root to see how it works for them.
+statement ok
+SET ROLE bob;
+
+# For bob, changing owner to bob is allowed by the policy.
+query T noticetrace
+INSERT INTO trigger_rls_table VALUES (10, 1000, 'alice');
+----
+NOTICE: User bob inserted new row with id 10 and value 1000
+
+# Verify the owner was changed to bob by the trigger.
+query IIT
+SELECT * FROM trigger_rls_table WHERE id = 10;
+----
+10  1000  bob
+
+# For completeness, let's see from the root perspective.
+statement ok
+SET ROLE root;
+
+# Check all rows in the table to see what happened.
+query IIT
+SELECT * FROM trigger_rls_table ORDER BY id;
+----
+1  150  alice
+2  250  bob
+4  400  bob
+5  500  alice
+7  700  bob
+8  800  bob
+10  1000  bob
+
+# Clean up the additional triggers and functions.
+statement ok
+DROP TRIGGER force_bob_ownership ON trigger_rls_table;
+DROP TRIGGER force_update_bob_ownership ON trigger_rls_table;
+DROP FUNCTION change_owner_to_bob;
+DROP FUNCTION change_updated_owner_to_bob;
+DROP TRIGGER log_changes ON trigger_rls_table;
+DROP TRIGGER check_positive_value ON trigger_rls_table;
+DROP TRIGGER bypass_attempt ON trigger_rls_table;
+DROP FUNCTION log_value_changes;
+DROP FUNCTION enforce_positive_value;
+DROP FUNCTION try_bypass_rls;
+DROP TABLE trigger_rls_table;
+DROP USER alice;
+DROP USER bob;
+
+subtest end

pkg/storage/engine.go

@@ -1102,13 +1102,9 @@ type Engine interface {
 	// version that it must maintain compatibility with.
 	SetMinVersion(version roachpb.Version) error
 
-	// SetCompactionConcurrency is used to set the engine's compaction
-	// concurrency. It returns the previous compaction concurrency.
-	SetCompactionConcurrency(n uint64) uint64
-
-	// AdjustCompactionConcurrency adjusts the compaction concurrency up or down by
-	// the passed delta, down to a minimum of 1.
-	AdjustCompactionConcurrency(delta int64) uint64
+	// SetCompactionConcurrency is used to override the engine's max compaction
+	// concurrency. A value of 0 removes any existing override.
+	SetCompactionConcurrency(n uint64)
 
 	// SetStoreID informs the engine of the store ID, once it is known.
 	// Used to show the store ID in logs and to initialize the shared object

pkg/storage/mvcc.go

@@ -101,46 +101,32 @@ var TargetBytesPerLockConflictError = settings.RegisterIntSetting(
 	settings.WithName("storage.mvcc.target_bytes_per_lock_conflict_error"),
 )
 
-// getMaxConcurrentCompactions wraps the maxConcurrentCompactions env var in a
-// func that may be installed on Options.MaxConcurrentCompactions. It also
-// imposes a floor on the max, so that an engine is always created with at least
-// 1 slot for a compactions.
-//
-// NB: This function inspects the environment every time it's called. This is
-// okay, because Engine construction in NewPebble will invoke it and store the
-// value on the Engine itself.
-func getMaxConcurrentCompactions() int {
-	n := envutil.EnvOrDefaultInt(
-		"COCKROACH_CONCURRENT_COMPACTIONS", func() int {
-			// The old COCKROACH_ROCKSDB_CONCURRENCY environment variable was never
-			// documented, but customers were told about it and use today in
-			// production. We don't want to break them, so if the new env var
-			// is unset but COCKROACH_ROCKSDB_CONCURRENCY is set, use the old env
-			// var's value. This old env var has a wart in that it's expressed as a
-			// number of concurrency slots to make available to both flushes and
-			// compactions (a vestige of the corresponding RocksDB option's
-			// mechanics). We need to adjust it to be in terms of just compaction
-			// concurrency by subtracting the flushing routine's dedicated slot.
-			//
-			// TODO(jackson): Should envutil expose its `getEnv` internal func for
-			// cases like this where we actually want to know whether it's present
-			// or not; not just fallback to a default?
-			if oldV := envutil.EnvOrDefaultInt("COCKROACH_ROCKSDB_CONCURRENCY", 0); oldV > 0 {
-				return oldV - 1
-			}
-
-			// By default use up to min(numCPU-1, 3) threads for background
-			// compactions per store (reserving the final process for flushes).
-			const max = 3
-			if n := runtime.GOMAXPROCS(0); n-1 < max {
-				return n - 1
-			}
-			return max
-		}())
-	if n < 1 {
-		return 1
-	}
-	return n
+var defaultMaxConcurrentCompactions = getDefaultMaxConcurrentCompactions()
+
+func getDefaultMaxConcurrentCompactions() int {
+	if v := envutil.EnvOrDefaultInt("COCKROACH_CONCURRENT_COMPACTIONS", 0); v > 0 {
+		return v
+	}
+	// The old COCKROACH_ROCKSDB_CONCURRENCY environment variable was never
+	// documented, but customers were told about it and use today in
+	// production. We don't want to break them, so if the new env var
+	// is unset but COCKROACH_ROCKSDB_CONCURRENCY is set, use the old env
+	// var's value. This old env var has a wart in that it's expressed as a
+	// number of concurrency slots to make available to both flushes and
+	// compactions (a vestige of the corresponding RocksDB option's
+	// mechanics). We need to adjust it to be in terms of just compaction
+	// concurrency by subtracting the flushing routine's dedicated slot.
+	if oldV := envutil.EnvOrDefaultInt("COCKROACH_ROCKSDB_CONCURRENCY", 0); oldV > 0 {
+		return max(oldV-1, 1)
+	}
+
+	// By default use up to min(GOMAXPROCS-1, 3) threads for background
+	// compactions per store (reserving the final process for flushes).
+	const upperLimit = 3
+	if n := runtime.GOMAXPROCS(0); n-1 < upperLimit {
+		return max(n-1, 1)
+	}
+	return upperLimit
 }
 
 // l0SubLevelCompactionConcurrency is the sub-level threshold at which to