Introducing the Observer pattern to SQL execution

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

@@ -49,7 +49,6 @@
 import io.micronaut.data.model.runtime.RuntimePersistentProperty;
 import io.micronaut.data.model.runtime.StoredQuery;
 import io.micronaut.data.operations.HintsCapableRepository;
-import io.micronaut.data.runtime.config.DataSettings;
 import io.micronaut.data.runtime.convert.DataConversionService;
 import io.micronaut.data.runtime.criteria.RuntimeCriteriaBuilder;
 import io.micronaut.data.runtime.date.DateTimeProvider;
@@ -68,7 +67,6 @@
 import io.micronaut.inject.annotation.AnnotationMetadataHierarchy;
 import io.micronaut.inject.qualifiers.Qualifiers;
 import io.micronaut.json.JsonMapper;
-import org.slf4j.Logger;
 
 import java.io.IOException;
 import java.util.AbstractMap;
@@ -101,8 +99,6 @@ public abstract class AbstractSqlRepositoryOperations<RS, PS, Exc extends Except
     MethodContextAwareStoredQueryDecorator,
     HintsCapableRepository {
 
-    protected static final Logger QUERY_LOG = DataSettings.QUERY_LOG;
-
     protected final String dataSourceName;
     @SuppressWarnings("WeakerAccess")
     protected final ResultReader<RS, String> columnNameResultSetReader;
@@ -117,6 +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);
+    protected final List<SqlExecutionObserver> observers;
 
     /**
      * Default constructor.
@@ -144,14 +141,16 @@ protected AbstractSqlRepositoryOperations(
         DataConversionService conversionService,
         AttributeConverterRegistry attributeConverterRegistry,
         JsonMapper jsonMapper,
-        SqlJsonColumnMapperProvider<RS> sqlJsonColumnMapperProvider) {
+        SqlJsonColumnMapperProvider<RS> sqlJsonColumnMapperProvider,
+            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.observers = observers;
         Collection<BeanDefinition<Object>> beanDefinitions = beanContext
             .getBeanDefinitions(Object.class, Qualifiers.byStereotype(Repository.class));
         for (BeanDefinition<Object> beanDefinition : beanDefinitions) {
@@ -201,9 +200,7 @@ protected <T, R> PS prepareStatement(StatementSupplier<PS> statementFunction,
         }
 
         String query = sqlPreparedQuery.getQuery();
-        if (QUERY_LOG.isDebugEnabled()) {
-            QUERY_LOG.debug("Executing Query: {}", query);
-        }
+        observers.forEach(listener -> listener.query(query));
         final PS ps;
         try {
             ps = statementFunction.create(query);
@@ -251,8 +248,8 @@ protected void setStatementParameter(PS preparedStatement, int index, DataType d
 
         dataType = dialect.getDataType(dataType);
 
-        if (QUERY_LOG.isTraceEnabled()) {
-            QUERY_LOG.trace("Binding parameter at position {} to value {} with data type: {}", index, value, dataType);
+        for (SqlExecutionObserver listener : observers) {
+            listener.parameter(index, value, dataType);
         }
 
         // We want to avoid potential conversion for JSON because mapper already returned value ready to be set as statement parameter

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

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2017-2020 original 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 io.micronaut.data.runtime.operations.internal.sql;
+
+import io.micronaut.data.model.DataType;
+import io.micronaut.data.runtime.config.DataSettings;
+import jakarta.inject.Singleton;
+import org.slf4j.Logger;
+
+@Singleton
+public class LoggingSqlExecutionObserver implements SqlExecutionObserver {
+
+    protected static final Logger QUERY_LOG = DataSettings.QUERY_LOG;
+
+    @Override
+    public void query(String query) {
+        if (QUERY_LOG.isDebugEnabled()) {
+            QUERY_LOG.debug("Executing Query: {}", query);
+        }
+    }
+
+    @Override
+    public void parameter(int index, Object value, DataType dataType) {
+        if (QUERY_LOG.isTraceEnabled()) {
+            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

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2017-2020 original 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 io.micronaut.data.runtime.operations.internal.sql;
+
+import io.micronaut.data.model.DataType;
+
+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
+        }
+    }
+}