Fixing compilation errors and adding tests

data-jdbc/src/main/java/io/micronaut/data/jdbc/operations/DefaultJdbcRepositoryOperations.java

@@ -88,10 +88,7 @@
 import io.micronaut.data.runtime.operations.internal.OperationContext;
 import io.micronaut.data.runtime.operations.internal.SyncCascadeOperations;
 import io.micronaut.data.runtime.operations.internal.query.BindableParametersStoredQuery;
-import io.micronaut.data.runtime.operations.internal.sql.AbstractSqlRepositoryOperations;
-import io.micronaut.data.runtime.operations.internal.sql.SqlJsonColumnMapperProvider;
-import io.micronaut.data.runtime.operations.internal.sql.SqlPreparedQuery;
-import io.micronaut.data.runtime.operations.internal.sql.SqlStoredQuery;
+import io.micronaut.data.runtime.operations.internal.sql.*;
 import io.micronaut.data.runtime.support.AbstractConversionContext;
 import io.micronaut.json.JsonMapper;
 import io.micronaut.transaction.TransactionOperations;
@@ -198,7 +195,8 @@ public final class DefaultJdbcRepositoryOperations extends AbstractSqlRepository
                                     JdbcSchemaHandler schemaHandler,
                                     @Nullable JsonMapper jsonMapper,
                                     SqlJsonColumnMapperProvider<ResultSet> sqlJsonColumnMapperProvider,
-                                    List<SqlExceptionMapper> sqlExceptionMapperList) {
+                                    List<SqlExceptionMapper> sqlExceptionMapperList,
+                                    List<SqlExecutionObserver> observers) {
         super(
             dataSourceName,
             new ColumnNameResultSetReader(conversionService),
@@ -210,7 +208,8 @@ public final class DefaultJdbcRepositoryOperations extends AbstractSqlRepository
             conversionService,
             attributeConverterRegistry,
             jsonMapper,
-            sqlJsonColumnMapperProvider);
+            sqlJsonColumnMapperProvider,
+            observers);
         this.schemaTenantResolver = schemaTenantResolver;
         this.schemaHandler = schemaHandler;
         this.connectionOperations = connectionOperations;
@@ -538,8 +537,8 @@ public Optional<Number> executeUpdate(@NonNull PreparedQuery<?, Number> pq) {
             try (PreparedStatement ps = prepareStatement(connection::prepareStatement, preparedQuery, true, false)) {
                 preparedQuery.bindParameters(new JdbcParameterBinder(connection, ps, preparedQuery));
                 int result = ps.executeUpdate();
-                if (QUERY_LOG.isTraceEnabled()) {
-                    QUERY_LOG.trace("Update operation updated {} records", result);
+                for (SqlExecutionObserver observer : observers) {
+                    observer.updatedRecords(result);
                 }
                 if (preparedQuery.isOptimisticLock()) {
                     checkOptimisticLocking(1, result);
@@ -847,8 +846,8 @@ public <R> R execute(@NonNull ConnectionCallback<R> callback) {
     public <R> R prepareStatement(@NonNull String sql, @NonNull PreparedStatementCallback<R> callback) {
         ArgumentUtils.requireNonNull("sql", sql);
         ArgumentUtils.requireNonNull("callback", callback);
-        if (QUERY_LOG.isDebugEnabled()) {
-            QUERY_LOG.debug("Executing Query: {}", sql);
+        for (SqlExecutionObserver observer : observers) {
+            observer.query(sql);
         }
         ConnectionContext connectionCtx = getConnectionCtx();
         try {
@@ -1169,8 +1168,8 @@ private PreparedStatement prepare(Connection connection, SqlStoredQuery<T, ?> st
 
         @Override
         protected void execute() throws SQLException {
-            if (QUERY_LOG.isDebugEnabled()) {
-                QUERY_LOG.debug("Executing SQL query: {}", storedQuery.getQuery());
+            for (SqlExecutionObserver observer : observers) {
+                observer.query(storedQuery.getQuery());
             }
             try {
                 if (storedQuery.getOperationType() == StoredQuery.OperationType.INSERT_RETURNING
@@ -1292,8 +1291,8 @@ private void setParameters(PreparedStatement stmt, SqlStoredQuery<T, ?> storedQu
 
         @Override
         protected void execute() {
-            if (QUERY_LOG.isDebugEnabled()) {
-                QUERY_LOG.debug("Executing SQL query: {}", storedQuery.getQuery());
+            for (SqlExecutionObserver observer : observers) {
+                observer.query(storedQuery.getQuery());
             }
             if (storedQuery.getOperationType() == StoredQuery.OperationType.INSERT_RETURNING
                 || storedQuery.getOperationType() == StoredQuery.OperationType.UPDATE_RETURNING) {

data-r2dbc/src/main/java/io/micronaut/data/r2dbc/operations/DefaultR2dbcRepositoryOperations.java

@@ -77,10 +77,7 @@
 import io.micronaut.data.runtime.operations.internal.OperationContext;
 import io.micronaut.data.runtime.operations.internal.ReactiveCascadeOperations;
 import io.micronaut.data.runtime.operations.internal.query.BindableParametersStoredQuery;
-import io.micronaut.data.runtime.operations.internal.sql.AbstractSqlRepositoryOperations;
-import io.micronaut.data.runtime.operations.internal.sql.SqlJsonColumnMapperProvider;
-import io.micronaut.data.runtime.operations.internal.sql.SqlPreparedQuery;
-import io.micronaut.data.runtime.operations.internal.sql.SqlStoredQuery;
+import io.micronaut.data.runtime.operations.internal.sql.*;
 import io.micronaut.data.runtime.support.AbstractConversionContext;
 import io.micronaut.json.JsonMapper;
 import io.micronaut.transaction.exceptions.TransactionSystemException;
@@ -188,7 +185,8 @@ final class DefaultR2dbcRepositoryOperations extends AbstractSqlRepositoryOperat
         SqlJsonColumnMapperProvider<Row> sqlJsonColumnMapperProvider,
         List<R2dbcExceptionMapper> r2dbcExceptionMapperList,
         @Parameter R2dbcReactorTransactionOperations transactionOperations,
-        @Parameter ReactorConnectionOperations<Connection> connectionOperations) {
+        @Parameter ReactorConnectionOperations<Connection> connectionOperations,
+        List<SqlExecutionObserver> observers) {
         super(
             dataSourceName,
             new ColumnNameR2dbcResultReader(conversionService),
@@ -200,7 +198,8 @@ final class DefaultR2dbcRepositoryOperations extends AbstractSqlRepositoryOperat
             conversionService,
             attributeConverterRegistry,
             jsonMapper,
-            sqlJsonColumnMapperProvider);
+            sqlJsonColumnMapperProvider,
+            observers);
         this.connectionFactory = connectionFactory;
         this.ioExecutorService = executorService;
         this.schemaTenantResolver = schemaTenantResolver;
@@ -545,8 +544,8 @@ public Mono<Number> executeUpdate(@NonNull PreparedQuery<?, Number> pq) {
                 preparedQuery.bindParameters(new R2dbcParameterBinder(connection, statement, preparedQuery));
                 return executeAndGetRowsUpdatedSingle(statement, dialect)
                     .flatMap((Number rowsUpdated) -> {
-                        if (QUERY_LOG.isTraceEnabled()) {
-                            QUERY_LOG.trace("Update operation updated {} records", rowsUpdated);
+                        for (SqlExecutionObserver observer : observers) {
+                            observer.updatedRecords(rowsUpdated);
                         }
                         if (preparedQuery.isOptimisticLock()) {
                             checkOptimisticLocking(1, rowsUpdated);
@@ -970,8 +969,8 @@ private <T> Mono<T> executeAndMapEachRowSingle(Statement statement, Dialect dial
 
         @Override
         protected void execute() throws RuntimeException {
-            if (QUERY_LOG.isDebugEnabled()) {
-                QUERY_LOG.debug("Executing SQL query: {}", storedQuery.getQuery());
+            for (SqlExecutionObserver observer : observers) {
+                observer.query(storedQuery.getQuery());
             }
             Statement statement = prepare(ctx.connection);
             setParameters(statement, storedQuery);
@@ -1065,8 +1064,8 @@ private void setParameters(Statement stmt, SqlStoredQuery<T, ?> storedQuery) {
 
         @Override
         protected void execute() throws RuntimeException {
-            if (QUERY_LOG.isDebugEnabled()) {
-                QUERY_LOG.debug("Executing SQL query: {}", storedQuery.getQuery());
+            for (SqlExecutionObserver observer : observers) {
+                observer.query(storedQuery.getQuery());
             }
             Statement statement;
             if (hasGeneratedId) {

data-runtime/src/main/java/io/micronaut/data/runtime/operations/internal/sql/AbstractSqlRepositoryOperations.java

@@ -113,7 +113,7 @@ public abstract class AbstractSqlRepositoryOperations<RS, PS, Exc extends Except
     private final Map<QueryKey, SqlStoredQuery> entityInserts = new ConcurrentHashMap<>(10);
     private final Map<QueryKey, SqlStoredQuery> entityUpdates = new ConcurrentHashMap<>(10);
     private final Map<Association, String> associationInserts = new ConcurrentHashMap<>(10);
-    private final List<SqlExecutionObserver> listeners;
+    protected final List<SqlExecutionObserver> observers;
 
     /**
      * Default constructor.
@@ -142,15 +142,15 @@ protected AbstractSqlRepositoryOperations(
         AttributeConverterRegistry attributeConverterRegistry,
         JsonMapper jsonMapper,
         SqlJsonColumnMapperProvider<RS> sqlJsonColumnMapperProvider,
-            List<SqlExecutionObserver> listeners) {
+            List<SqlExecutionObserver> observers) {
         super(dateTimeProvider, runtimeEntityRegistry, conversionService, attributeConverterRegistry);
         this.dataSourceName = dataSourceName;
         this.columnNameResultSetReader = columnNameResultSetReader;
         this.columnIndexResultSetReader = columnIndexResultSetReader;
         this.preparedStatementWriter = preparedStatementWriter;
         this.jsonMapper = jsonMapper;
         this.sqlJsonColumnMapperProvider = sqlJsonColumnMapperProvider;
-        this.listeners = listeners;
+        this.observers = observers;
         Collection<BeanDefinition<Object>> beanDefinitions = beanContext
             .getBeanDefinitions(Object.class, Qualifiers.byStereotype(Repository.class));
         for (BeanDefinition<Object> beanDefinition : beanDefinitions) {
@@ -200,7 +200,7 @@ protected <T, R> PS prepareStatement(StatementSupplier<PS> statementFunction,
         }
 
         String query = sqlPreparedQuery.getQuery();
-        listeners.forEach(listener -> listener.query(query));
+        observers.forEach(listener -> listener.query(query));
         final PS ps;
         try {
             ps = statementFunction.create(query);
@@ -248,7 +248,7 @@ protected void setStatementParameter(PS preparedStatement, int index, DataType d
 
         dataType = dialect.getDataType(dataType);
 
-        for (SqlExecutionObserver listener : listeners) {
+        for (SqlExecutionObserver listener : observers) {
             listener.parameter(index, value, dataType);
         }
 

data-runtime/src/main/java/io/micronaut/data/runtime/operations/internal/sql/LoggingSqlExecutionObserver.java

@@ -38,4 +38,11 @@ public void parameter(int index, Object value, DataType dataType) {
             QUERY_LOG.trace("Binding parameter at position {} to value {} with data type: {}", index, value, dataType);
         }
     }
+
+    @Override
+    public void updatedRecords(Number result) {
+        if (QUERY_LOG.isTraceEnabled()) {
+            QUERY_LOG.trace("Update operation updated {} records", result);
+        }
+    }
 }

data-runtime/src/main/java/io/micronaut/data/runtime/operations/internal/sql/SqlExecutionObserver.java

@@ -22,4 +22,6 @@ public interface SqlExecutionObserver {
     void query(String query);
 
     void parameter(int index, Object value, DataType datatype);
+
+    void updatedRecords(Number result);
 }

data-tck/src/main/groovy/io/micronaut/data/tck/TestSqlExecutionObserver.groovy

@@ -0,0 +1,39 @@
+package io.micronaut.data.tck
+
+import io.micronaut.data.model.DataType
+import io.micronaut.data.runtime.operations.internal.sql.SqlExecutionObserver
+import jakarta.inject.Singleton
+
+@Singleton
+class TestSqlExecutionObserver implements SqlExecutionObserver {
+    public List<Invocation> invocations = new ArrayList<>()
+
+    @Override
+    void query(String query) {
+        invocations.add(new Invocation(query))
+    }
+
+    @Override
+    void parameter(int index, Object value, DataType datatype) {
+        invocations.last().parameters[index] = value
+    }
+
+    @Override
+    void updatedRecords(Number result) {
+        invocations.last().affected = result
+    }
+
+    void clear() {
+        invocations.clear()
+    }
+
+    class Invocation {
+        String query
+        Map<Integer, Object> parameters = [:]
+        Number affected
+
+        Invocation(String query) {
+            this.query = query
+        }
+    }
+}