Fix prepared statement caching using observability.

We now no longer create a statement proxy when preparing CQL statements as prepared statements do not use RequestTracker for completion callbacks. Also, ObservationStatement now implements equals and hashCode to report equality for its underlying statement in case the statement was used as cache key.

Previously, we created a proxy without implementing equals and hashCode resulting in re-preparation as the prepared statement cache kept growing because the input statement did not provide means to serve as cache key.

Closes #1601

Author: mp911de

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionObservationInterceptor.java

@@ -93,7 +93,7 @@ final class CqlSessionObservationInterceptor implements MethodInterceptor {
 			Observation observation = startObservation(statement, true, "prepare");
 
 			try {
-				return this.delegate.prepare((SimpleStatement) ObservationStatement.createProxy(observation, statement));
+				return this.delegate.prepare((SimpleStatement) statement);
 			} catch (RuntimeException e) {
 
 				observation.error(e);
@@ -113,7 +113,7 @@ final class CqlSessionObservationInterceptor implements MethodInterceptor {
 
 			Observation observation = startObservation(statement, true, "prepareAsync");
 
-			return this.delegate.prepareAsync((SimpleStatement) ObservationStatement.createProxy(observation, statement))
+			return this.delegate.prepareAsync((SimpleStatement) statement)
 					.whenComplete((preparedStatement, throwable) -> {
 
 						if (throwable != null) {

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservableReactiveSession.java

@@ -167,7 +167,7 @@ public Mono<PreparedStatement> prepare(SimpleStatement statement) {
 		return Mono.deferContextual(contextView -> {
 
 			Observation observation = startObservation(getParentObservation(contextView), statement, true, "prepare");
-			return this.delegate.prepare(ObservationStatement.createProxy(observation, statement)) //
+			return this.delegate.prepare(statement) //
 					.doOnError(observation::error) //
 					.doFinally(ignore -> observation.stop());
 		});

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservationStatement.java

@@ -19,13 +19,13 @@
 
 import java.lang.reflect.Method;
 
-import javax.annotation.Nonnull;
-
 import org.aopalliance.intercept.MethodInterceptor;
 import org.aopalliance.intercept.MethodInvocation;
 import org.jspecify.annotations.Nullable;
+
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.util.ClassUtils;
+import org.springframework.util.ObjectUtils;
 
 import com.datastax.oss.driver.api.core.cql.Statement;
 
@@ -76,23 +76,51 @@ public static boolean isObservationStatement(Statement<?> statement) {
 	}
 
 	@Override
-	public @Nullable Object invoke(@Nonnull MethodInvocation invocation) throws Throwable {
+	public @Nullable Object invoke(MethodInvocation invocation) throws Throwable {
 
 		Method method = invocation.getMethod();
-
-		if (method.getName().equals("getTargetClass")) {
-			return this.delegate.getClass();
-		}
-
-		if (method.getName().equals("getObservation")) {
-			return this.observation;
+		@Nullable
+		Object[] args = invocation.getArguments();
+
+		String name = method.getName();
+
+		switch (name) {
+			case "equals" -> {
+				if (args.length == 1) {
+					return equals(args[0]);
+				}
+			}
+			case "hashCode" -> {
+				return hashCode();
+			}
+
+			case "getTargetClass" -> {
+				return this.delegate.getClass();
+			}
+
+			case "getObservation" -> {
+				return this.observation;
+			}
 		}
 
 		Object result = invocation.proceed();
 		if (result instanceof Statement<?>) {
 			this.delegate = (Statement<?>) result;
 		}
+
 		return result;
 	}
 
+	@Override
+	public boolean equals(Object o) {
+		if (!(o instanceof ObservationStatement that)) {
+			return false;
+		}
+		return ObjectUtils.nullSafeEquals(delegate, that.delegate);
+	}
+
+	@Override
+	public int hashCode() {
+		return ObjectUtils.nullSafeHash(delegate);
+	}
 }

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/ImperativeIntegrationTests.java

@@ -31,6 +31,7 @@
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 
 import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.api.core.cql.PreparedStatement;
 
 /**
  * Collection of tests that log metrics and tracing.
@@ -74,6 +75,13 @@ public SampleTestRunnerConsumer yourCode() {
 
 			CqlTemplate template = new CqlTemplate(observableSession);
 
+			PreparedStatement prepare1 = observableSession
+					.prepare("INSERT INTO person (id, firstName, lastName) VALUES (?, ?, ?);");
+			PreparedStatement prepare2 = observableSession
+					.prepare("INSERT INTO person (id, firstName, lastName) VALUES (?, ?, ?);");
+
+			assertThat(prepare1).isSameAs(prepare2);
+
 			template.execute("INSERT INTO person (id,firstName,lastName) VALUES(?,?,?)", 1, "Walter", "White");
 
 			assertThat(tracer.getFinishedSpans()).hasSizeGreaterThanOrEqualTo(5);

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/ObservationStatementUnitTests.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2025 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.cassandra.observability;
+
+import static org.assertj.core.api.Assertions.*;
+
+import io.micrometer.observation.Observation;
+import io.micrometer.observation.tck.TestObservationRegistry;
+
+import org.junit.jupiter.api.Test;
+
+import com.datastax.oss.driver.api.core.cql.SimpleStatement;
+
+/**
+ * Unit test for {@link ObservationStatement}
+ *
+ * @author Mark Paluch
+ */
+class ObservationStatementUnitTests {
+
+	@Test // GH-1601
+	void equalsAndHashCodeShouldBeEqual() {
+
+		TestObservationRegistry registry = TestObservationRegistry.create();
+
+		Observation observation1 = Observation.start("foo", registry);
+		Observation observation2 = Observation.start("bar", registry);
+
+		SimpleStatement statement1 = ObservationStatement.createProxy(observation1,
+				SimpleStatement.newInstance("SELECT * FROM foo"));
+		SimpleStatement statement2 = ObservationStatement.createProxy(observation2,
+				SimpleStatement.newInstance("SELECT * FROM foo"));
+		SimpleStatement statement3 = ObservationStatement.createProxy(observation2,
+				SimpleStatement.newInstance("SELECT * FROM bar"));
+
+		assertThat(statement1).isEqualTo(statement2).hasSameHashCodeAs(statement2);
+		assertThat(statement1).isNotEqualTo(statement3);
+		assertThat(statement1.hashCode()).isNotEqualTo(statement3.hashCode());
+	}
+}

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/ReactiveIntegrationTests.java

@@ -36,6 +36,8 @@
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 
+import com.datastax.oss.driver.api.core.cql.PreparedStatement;
+
 /**
  * Collection of tests that log metrics and tracing.
  *
@@ -68,7 +70,6 @@ public SampleTestRunnerConsumer yourCode() {
 
 			Observation intermediate = Observation.start("intermediate", createObservationRegistry());
 
-
 			Mono<ReactiveResultSet> drop = observableSession.execute("DROP KEYSPACE IF EXISTS ObservationTest");
 			Mono<ReactiveResultSet> create = observableSession.execute("CREATE KEYSPACE ObservationTest " + "WITH "
 					+ "REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 };");
@@ -87,6 +88,13 @@ public SampleTestRunnerConsumer yourCode() {
 						.verifyComplete();
 			});
 
+			PreparedStatement prepare1 = observableSession
+					.prepare("INSERT INTO person (id, firstName, lastName) VALUES (?, ?, ?);").block();
+			PreparedStatement prepare2 = observableSession
+					.prepare("INSERT INTO person (id, firstName, lastName) VALUES (?, ?, ?);").block();
+
+			assertThat(prepare1).isSameAs(prepare2);
+
 			assertThat(tracer.getFinishedSpans()).hasSizeGreaterThanOrEqualTo(5);
 		};
 	}