Merge pull request ClickHouse#80591 from ClickHouse/chesema-storage-url-test

storage url does not call server on empty data

Author: CheSema

src/Common/ProfileEvents.cpp

@@ -985,6 +985,9 @@ The server successfully detected this situation and will download merged part fr
     M(ReadWriteBufferFromHTTPRequestsSent, "Number of HTTP requests sent by ReadWriteBufferFromHTTP", ValueType::Number) \
     M(ReadWriteBufferFromHTTPBytes, "Total size of payload bytes received and sent by ReadWriteBufferFromHTTP. Doesn't include HTTP headers.", ValueType::Bytes) \
     \
+    M(WriteBufferFromHTTPRequestsSent, "Number of HTTP requests sent by WriteBufferFromHTTP", ValueType::Number) \
+    M(WriteBufferFromHTTPBytes, "Total size of payload bytes received and sent by WriteBufferFromHTTP. Doesn't include HTTP headers.", ValueType::Bytes) \
+    \
     M(ConcurrencyControlSlotsGranted, "Number of CPU slot granted according to guarantee of 1 thread per query and for queries with setting 'use_concurrency_control' = 0", ValueType::Number) \
     M(ConcurrencyControlSlotsDelayed, "Number of CPU slot not granted initially and required to wait for a free CPU slot", ValueType::Number) \
     M(ConcurrencyControlSlotsAcquired, "Total number of CPU slot acquired", ValueType::Number) \

src/IO/ReadWriteBufferFromHTTP.cpp

@@ -1,6 +1,4 @@
 #include "ReadWriteBufferFromHTTP.h"
-#include <sstream>
-#include <string_view>
 
 #include <IO/HTTPCommon.h>
 #include <Common/NetException.h>

src/IO/WriteBufferFromHTTP.cpp

@@ -5,6 +5,12 @@
 #include <Interpreters/Context.h>
 
 
+namespace ProfileEvents
+{
+    extern const Event WriteBufferFromHTTPRequestsSent;
+    extern const Event WriteBufferFromHTTPBytes;
+}
+
 namespace DB
 {
 
@@ -39,9 +45,16 @@ WriteBufferFromHTTP::WriteBufferFromHTTP(
 
     LOG_TRACE((getLogger("WriteBufferToHTTP")), "Sending request to {}", uri.toString());
 
+    ProfileEvents::increment(ProfileEvents::WriteBufferFromHTTPRequestsSent);
     ostr = &session->sendRequest(request);
 }
 
+void WriteBufferFromHTTP::nextImpl()
+{
+    ProfileEvents::increment(ProfileEvents::WriteBufferFromHTTPBytes, offset());
+    WriteBufferFromOStream::nextImpl();
+}
+
 void WriteBufferFromHTTP::finalizeImpl()
 {
     // Make sure the content in the buffer has been flushed

src/IO/WriteBufferFromHTTP.h

@@ -29,6 +29,9 @@ class WriteBufferFromHTTP : public WriteBufferFromOStream
                                  size_t buffer_size_ = DBMS_DEFAULT_BUFFER_SIZE,
                                  ProxyConfiguration proxy_configuration = {});
 
+    // Counts the counter WriteBufferFromHTTPBytes
+    void nextImpl() override;
+
     /// Receives response from the server after sending all data.
     void finalizeImpl() override;
 

tests/queries/0_stateless/03519_storage_url.reference

@@ -0,0 +1,42 @@
+dst_remote	2	1
+dst_remote	4	1
+dst_remote	100	1
+dst_remote	200	1
+dst_remote	300	1
+dst_remote	400	1
+Row 1:
+──────
+query:           insert into dst_remote format TSV
+
+written_rows:    2
+HTTPRequests:    0
+HTTPConnections: 0
+
+Row 2:
+──────
+query:           insert into dst_remote format TSV
+
+written_rows:    4
+HTTPRequests:    0
+HTTPConnections: 0
+
+Row 3:
+──────
+query:           insert into src values 
+written_rows:    4
+HTTPRequests:    0
+HTTPConnections: 0
+
+Row 4:
+──────
+query:           insert into src values 
+written_rows:    6
+HTTPRequests:    1
+HTTPConnections: 1
+
+Row 5:
+──────
+query:           insert into src values 
+written_rows:    8
+HTTPRequests:    1
+HTTPConnections: 1

tests/queries/0_stateless/03519_storage_url.sh

@@ -0,0 +1,59 @@
+#!/usr/bin/env bash
+
+CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+# shellcheck source=../shell_config.sh
+. "$CURDIR"/../shell_config.sh
+
+set -eu
+
+$CLICKHOUSE_CLIENT -q "
+drop table if exists src;
+drop table if exists dst_remote;
+drop table if exists dst;
+drop table if exists mv;
+"
+
+$CLICKHOUSE_CLIENT -q "
+create table src (key Int, value Int)
+engine=MergeTree()
+ORDER BY key;
+
+create table dst_remote (key Int, value Int)
+engine=MergeTree()
+ORDER BY key;
+
+create table dst (key Int, value Int)
+engine=URL('${CLICKHOUSE_URL}&async_insert=0&query=insert+into+dst_remote+format+TSV', TSV);
+
+create materialized view mv to dst
+as select * from src where value > 0;
+"
+
+$CLICKHOUSE_CLIENT --async_insert=0 -q "
+insert into src values (1, 0), (2, 1), (3, 0), (4, 1);
+
+insert into src values (10, 0), (20, 0), (30, 0), (40, 0);
+
+insert into src values (100, 1), (200, 1), (300, 1), (400, 1);
+"
+
+$CLICKHOUSE_CLIENT -q "
+select 'dst_remote', * from dst_remote order by ALL;
+"
+
+$CLICKHOUSE_CLIENT -q "
+system flush logs query_log;
+
+select
+    query,
+    written_rows,
+    ProfileEvents['ReadWriteBufferFromHTTPRequestsSent'] + ProfileEvents['WriteBufferFromHTTPRequestsSent'] as HTTPRequests,
+    ProfileEvents['StorageConnectionsCreated'] + ProfileEvents['StorageConnectionsReused'] as HTTPConnections
+from system.query_log
+where
+    current_database = currentDatabase() and
+    type = 'QueryFinish' and
+    query_kind = 'Insert'
+order by ALL
+format Vertical;
+"