Merge pull request #49120 from CurtizJ/add-field-async-insert-log

Add field with number of rows to `system.asynchronous_insert_log`

src/Common/ProfileEvents.cpp

@@ -10,6 +10,7 @@
     M(InsertQuery, "Same as Query, but only for INSERT queries.") \
     M(AsyncInsertQuery, "Same as InsertQuery, but only for asynchronous INSERT queries.") \
     M(AsyncInsertBytes, "Data size in bytes of asynchronous INSERT queries.") \
+    M(AsyncInsertRows, "Number of rows inserted by asynchronous INSERT queries.") \
     M(AsyncInsertCacheHits, "Number of times a duplicate hash id has been found in asynchronous INSERT hash id cache.") \
     M(FailedQuery, "Number of failed queries.") \
     M(FailedSelectQuery, "Same as FailedQuery, but only for SELECT queries.") \

src/Interpreters/AsynchronousInsertLog.cpp

@@ -36,6 +36,7 @@ NamesAndTypesList AsynchronousInsertLogElement::getNamesAndTypes()
         {"format", std::make_shared<DataTypeLowCardinality>(std::make_shared<DataTypeString>())},
         {"query_id", std::make_shared<DataTypeString>()},
         {"bytes", std::make_shared<DataTypeUInt64>()},
+        {"rows", std::make_shared<DataTypeUInt64>()},
         {"exception", std::make_shared<DataTypeString>()},
         {"status", type_status},
 
@@ -71,6 +72,7 @@ void AsynchronousInsertLogElement::appendToBlock(MutableColumns & columns) const
     columns[i++]->insert(insert_query.format);
     columns[i++]->insert(query_id);
     columns[i++]->insert(bytes);
+    columns[i++]->insert(rows);
     columns[i++]->insert(exception);
     columns[i++]->insert(status);
 

src/Interpreters/AsynchronousInsertLog.h

@@ -24,6 +24,7 @@ struct AsynchronousInsertLogElement
     ASTPtr query;
     String query_id;
     UInt64 bytes{};
+    UInt64 rows{};
     String exception;
     Status status{};
 

src/Interpreters/AsynchronousInsertQueue.cpp

@@ -40,6 +40,7 @@ namespace ProfileEvents
 {
     extern const Event AsyncInsertQuery;
     extern const Event AsyncInsertBytes;
+    extern const Event AsyncInsertRows;
     extern const Event FailedAsyncInsertQuery;
 }
 
@@ -444,7 +445,8 @@ try
     {
         auto buffer = std::make_unique<ReadBufferFromString>(entry->bytes);
         current_entry = entry;
-        total_rows += executor.execute(*buffer);
+        size_t num_rows = executor.execute(*buffer);
+        total_rows += num_rows;
         chunk_info->offsets.push_back(total_rows);
 
         /// Keep buffer, because it still can be used
@@ -459,6 +461,7 @@ try
             elem.query = key.query;
             elem.query_id = entry->query_id;
             elem.bytes = entry->bytes.size();
+            elem.rows = num_rows;
             elem.exception = current_exception;
             current_exception.clear();
 
@@ -479,6 +482,7 @@ try
 
     format->addBuffer(std::move(last_buffer));
     auto insert_query_id = insert_context->getCurrentQueryId();
+    ProfileEvents::increment(ProfileEvents::AsyncInsertRows, total_rows);
 
     auto finish_entries = [&]
     {

tests/queries/0_stateless/02456_async_inserts_logs.reference

@@ -1,7 +1,10 @@
 5
-	Values	21	1	Ok	1
-t_async_inserts_logs	JSONEachRow	39	1	Ok	1
-t_async_inserts_logs	Values	8	1	Ok	1
-t_async_inserts_logs	JSONEachRow	6	0	ParsingError	1
-t_async_inserts_logs	Values	6	0	ParsingError	1
-t_async_inserts_logs	Values	8	0	FlushError	1
+	Values	21	2	1	Ok	1
+t_async_inserts_logs	JSONEachRow	39	2	1	Ok	1
+t_async_inserts_logs	Values	8	1	1	Ok	1
+t_async_inserts_logs	JSONEachRow	6	0	0	ParsingError	1
+t_async_inserts_logs	Values	6	0	0	ParsingError	1
+t_async_inserts_logs	Values	8	1	0	FlushError	1
+AsyncInsertBytes	1
+AsyncInsertQuery	1
+AsyncInsertRows	1

tests/queries/0_stateless/02456_async_inserts_logs.sh

@@ -30,10 +30,15 @@ ${CLICKHOUSE_CLIENT} -q "SELECT count() FROM t_async_inserts_logs"
 
 ${CLICKHOUSE_CLIENT} -q "SYSTEM FLUSH LOGS"
 ${CLICKHOUSE_CLIENT} -q "
-    SELECT table, format, bytes, empty(exception), status,
+    SELECT table, format, bytes, rows, empty(exception), status,
     status = 'ParsingError' ? flush_time_microseconds = 0 : flush_time_microseconds > event_time_microseconds AS time_ok
     FROM system.asynchronous_insert_log
     WHERE database = '$CLICKHOUSE_DATABASE' OR query ILIKE 'INSERT INTO FUNCTION%$CLICKHOUSE_DATABASE%'
     ORDER BY table, status, format"
 
 ${CLICKHOUSE_CLIENT} -q "DROP TABLE t_async_inserts_logs"
+
+${CLICKHOUSE_CLIENT} -q "
+SELECT event, value > 0 FROM system.events
+WHERE event IN ('AsyncInsertQuery', 'AsyncInsertBytes', 'AsyncInsertRows')
+ORDER BY event"