Implemented change notification when a table changes by the query.

example/lib/task_dao.dart

@@ -18,6 +18,9 @@ abstract class TaskDao {
   @Query('SELECT * FROM task WHERE status = :status')
   Stream<List<Task>> findAllTasksByStatusAsStream(TaskStatus status);
 
+  @Query('UPDATE OR ABORT Task SET type = :type WHERE id = :id')
+  Future<int?> updateTypeById(TaskType type, int id);
+
   @insert
   Future<void> insertTask(Task task);
 

example/pubspec.lock

@@ -113,6 +113,14 @@ packages:
       url: "https://pub.dev"
     source: hosted
     version: "1.2.1"
+  charcode:
+    dependency: transitive
+    description:
+      name: charcode
+      sha256: fb98c0f6d12c920a02ee2d998da788bca066ca5f148492b7085ee23372b12306
+      url: "https://pub.dev"
+    source: hosted
+    version: "1.3.1"
   checked_yaml:
     dependency: transitive
     description:
@@ -453,6 +461,14 @@ packages:
       url: "https://pub.dev"
     source: hosted
     version: "1.11.0"
+  sqlparser:
+    dependency: transitive
+    description:
+      name: sqlparser
+      sha256: "91f47610aa54d8abf9d795a7b4e49b2a788f65d7493d5a68fbf180c3cbcc6f38"
+      url: "https://pub.dev"
+    source: hosted
+    version: "0.27.0"
   stack_trace:
     dependency: transitive
     description:

floor/lib/src/adapter/query_adapter.dart

@@ -1,12 +1,17 @@
 import 'dart:async';
 
+import 'package:collection/collection.dart';
 import 'package:floor/src/util/string_utils.dart';
 import 'package:sqflite/sqflite.dart';
+import 'package:sqlparser/sqlparser.dart';
+
+import '../util/constants.dart';
 
 /// This class knows how to execute database queries.
 class QueryAdapter {
   final DatabaseExecutor _database;
   final StreamController<String>? _changeListener;
+  late final SqlEngine _sqlEngine = SqlEngine();
 
   QueryAdapter(
     final DatabaseExecutor database, [
@@ -20,7 +25,7 @@ class QueryAdapter {
     final List<Object>? arguments,
     required final T Function(Map<String, Object?>) mapper,
   }) async {
-    final rows = await _database.rawQuery(sql, arguments);
+    final rows = await _preformQuery(sql, arguments);
 
     if (rows.isEmpty) {
       return null;
@@ -37,18 +42,23 @@ class QueryAdapter {
     final List<Object>? arguments,
     required final T Function(Map<String, Object?>) mapper,
   }) async {
-    final rows = await _database.rawQuery(sql, arguments);
-    return rows.map((row) => mapper(row)).toList();
+    final rootNode = _parseRootNode(sql);
+
+    if (rootNode is SelectStatement) {
+      return _database
+          .rawQuery(sql, arguments)
+          .then((rows) => rows.map((row) => mapper(row)).toList());
+    } else {
+      throw StateError(
+          'Unsupported query "$sql" for List return type. It should be SELECT, since DELETE, UPDATE, INSERT returns `int` type.');
+    }
   }
 
   Future<void> queryNoReturn(
     final String sql, {
     final List<Object>? arguments,
   }) async {
-    // TODO #94 differentiate between different query kinds (select, update, delete, insert)
-    //  this enables to notify the observers
-    //  also requires extracting the table name :(
-    await _database.rawQuery(sql, arguments);
+    await _preformQuery(sql, arguments);
   }
 
   /// Executes a SQLite query that returns a stream of single query results
@@ -118,4 +128,50 @@ class QueryAdapter {
 
     return controller.stream;
   }
+
+  /// Parses the SQL query to determine which method is declared and executes it.
+  Future<List<Map<String, Object?>>> _preformQuery(
+    String sql,
+    List<Object>? arguments,
+  ) async {
+    List<Map<String, Object?>> result = List.empty();
+    String tableName = '';
+    final rootNode = _parseRootNode(sql);
+
+    if (rootNode is SelectStatement) {
+      result = await _database.rawQuery(sql, arguments);
+    } else if (rootNode is InsertStatement) {
+      result = await _database.rawInsert(sql, arguments).then(_mapResult);
+      tableName = rootNode.table.tableName;
+    } else if (rootNode is UpdateStatement) {
+      result = await _database.rawUpdate(sql, arguments).then(_mapResult);
+      tableName = rootNode.table.tableName;
+    } else if (rootNode is DeleteStatement) {
+      result = await _database.rawDelete(sql, arguments).then(_mapResult);
+      tableName = rootNode.table.tableName;
+    }
+
+    _notifyIfChanged(tableName, result);
+
+    return result;
+  }
+
+  /// Checks the query result, if it is a table change, notifies by table name.
+  void _notifyIfChanged(
+    String tableName,
+    List<Map<String, Object?>> result,
+  ) {
+    final count = result.firstOrNull?[changedRowsKey];
+    if (tableName.isNotEmpty && count is int && count > 0) {
+      _changeListener?.add(tableName);
+    }
+  }
+
+  /// Converts the modification `int` result to a query result.
+  FutureOr<List<Map<String, Object?>>> _mapResult(int value) => [
+        {changedRowsKey: value}
+      ];
+
+  /// Parses a root node to validate SQL
+  AstNode _parseRootNode(String sql) => _sqlEngine.parse(sql).rootNode;
 }

floor/lib/src/util/constants.dart

@@ -0,0 +1 @@
+const String changedRowsKey = 'changed_rows_key';

floor/pubspec.lock

@@ -113,6 +113,14 @@ packages:
       url: "https://pub.dev"
     source: hosted
     version: "1.2.1"
+  charcode:
+    dependency: transitive
+    description:
+      name: charcode
+      sha256: fb98c0f6d12c920a02ee2d998da788bca066ca5f148492b7085ee23372b12306
+      url: "https://pub.dev"
+    source: hosted
+    version: "1.3.1"
   checked_yaml:
     dependency: transitive
     description:
@@ -454,6 +462,14 @@ packages:
       url: "https://pub.dev"
     source: hosted
     version: "1.11.0"
+  sqlparser:
+    dependency: "direct main"
+    description:
+      name: sqlparser
+      sha256: "91f47610aa54d8abf9d795a7b4e49b2a788f65d7493d5a68fbf180c3cbcc6f38"
+      url: "https://pub.dev"
+    source: hosted
+    version: "0.27.0"
   stack_trace:
     dependency: transitive
     description:

floor/pubspec.yaml

@@ -20,6 +20,7 @@ dependencies:
   path: ^1.8.1
   sqflite: ^2.0.0+4
   sqflite_common_ffi: ^2.0.0+3
+  sqlparser: ^0.27.0
 
 dev_dependencies:
   build_runner: ^2.1.2