Merge pull request #13343 from azat/format-fixes-20.7

WithNames format fixes

src/Processors/Formats/IOutputFormat.cpp

@@ -46,6 +46,12 @@ IOutputFormat::Status IOutputFormat::prepare()
 
 void IOutputFormat::work()
 {
+    if (!prefix_written)
+    {
+        doWritePrefix();
+        prefix_written = true;
+    }
+
     if (finished && !finalized)
     {
         if (rows_before_limit_counter && rows_before_limit_counter->hasAppliedLimit())

src/Processors/Formats/IOutputFormat.h

@@ -87,6 +87,8 @@ private:
     /// Counters for consumed chunks. Are used for QueryLog.
     size_t result_rows = 0;
     size_t result_bytes = 0;
+
+    bool prefix_written = false;
 };
 }
 

src/Processors/Formats/Impl/TabSeparatedRowOutputFormat.cpp

@@ -18,7 +18,7 @@ TabSeparatedRowOutputFormat::TabSeparatedRowOutputFormat(
 }
 
 
-void TabSeparatedRowOutputFormat::writePrefix()
+void TabSeparatedRowOutputFormat::doWritePrefix()
 {
     const auto & header = getPort(PortKind::Main).getHeader();
     size_t columns = header.columns();

src/Processors/Formats/Impl/TabSeparatedRowOutputFormat.h

@@ -31,10 +31,11 @@ public:
     void writeField(const IColumn & column, const IDataType & type, size_t row_num) override;
     void writeFieldDelimiter() override;
     void writeRowEndDelimiter() override;
-    void writePrefix() override;
     void writeBeforeTotals() override;
     void writeBeforeExtremes() override;
 
+    void doWritePrefix() override;
+
     /// https://www.iana.org/assignments/media-types/text/tab-separated-values
     String getContentType() const override { return "text/tab-separated-values; charset=UTF-8"; }
 

src/Processors/Sources/SinkToOutputStream.cpp

@@ -8,15 +8,11 @@ SinkToOutputStream::SinkToOutputStream(BlockOutputStreamPtr stream_)
     : ISink(stream_->getHeader())
     , stream(std::move(stream_))
 {
+    stream->writePrefix();
 }
 
 void SinkToOutputStream::consume(Chunk chunk)
 {
-    if (!initialized)
-        stream->writePrefix();
-
-    initialized = true;
-
     stream->write(getPort().getHeader().cloneWithColumns(chunk.detachColumns()));
 }
 

src/Processors/Sources/SinkToOutputStream.h

@@ -22,7 +22,6 @@ protected:
 
 private:
     BlockOutputStreamPtr stream;
-    bool initialized = false;
 };
 
 }

tests/queries/0_stateless/01375_output_format_tsv_csv_with_names.reference

+TSVWithNames
+number
+0
+1
+TSVWithNamesAndTypes
+number
+UInt64
+0
+1
+CSVWithNames
+"number"
+0
+1

tests/queries/0_stateless/01375_output_format_tsv_csv_with_names.sh

+#!/usr/bin/env bash
+
+CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+. "$CURDIR"/../shell_config.sh
+
+opts=(
+    --input-format CSV
+    -q 'SELECT number FROM numbers(2)'
+)
+
+echo 'TSVWithNames'
+${CLICKHOUSE_LOCAL} "${opts[@]}" --format TSVWithNames
+
+echo 'TSVWithNamesAndTypes'
+${CLICKHOUSE_LOCAL} "${opts[@]}" --format TSVWithNamesAndTypes
+
+echo 'CSVWithNames'
+${CLICKHOUSE_LOCAL} "${opts[@]}" --format CSVWithNames

tests/queries/0_stateless/01375_storage_file_tsv_csv_with_names_write_prefix.reference

+zero rows
+TSVWithNames
+TSVWithNamesAndTypes
+CSVWithNames
+multi clickhouse-local one file
+TSVWithNames
+0
+0
+0
+TSVWithNamesAndTypes
+0
+0
+0
+CSVWithNames
+0
+0
+0

tests/queries/0_stateless/01375_storage_file_tsv_csv_with_names_write_prefix.sh

+#!/usr/bin/env bash
+
+CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+. "$CURDIR"/../shell_config.sh
+
+# zero rows
+echo 'zero rows'
+for format in TSVWithNames TSVWithNamesAndTypes CSVWithNames; do
+    echo $format
+    ${CLICKHOUSE_LOCAL} --query="
+        CREATE TABLE ${format}_01375 ENGINE File($format, '01375_$format.tsv') AS SELECT * FROM numbers(1) WHERE number < 0;
+        SELECT * FROM ${format}_01375;
+        DROP TABLE ${format}_01375;
+    "
+    rm 01375_$format.tsv
+done
+
+# run multiple times to the same file
+echo 'multi clickhouse-local one file'
+for format in TSVWithNames TSVWithNamesAndTypes CSVWithNames; do
+    echo $format
+    for _ in {1..2}; do
+        ${CLICKHOUSE_LOCAL} --query="
+            CREATE TABLE ${format}_01375 ENGINE File($format, '01375_$format.tsv') AS SELECT * FROM numbers(1);
+            SELECT * FROM ${format}_01375;
+            DROP TABLE ${format}_01375;
+        "
+    done
+    rm 01375_$format.tsv
+done

tests/queries/0_stateless/01375_storage_file_write_prefix_tsv_with_names.reference

+0	0
+1	1
+2	2
+3	3
+4	4
+5	5
+6	6
+7	7
+8	8
+9	9
+0	0
+1	1
+2	2
+3	3
+4	4
+5	5
+6	6
+7	7
+8	8
+9	9
+0	0
+1	1
+2	2
+3	3
+4	4
+5	5
+6	6
+7	7
+8	8
+9	9

tests/queries/0_stateless/01375_storage_file_write_prefix_tsv_with_names.sql

+DROP TABLE IF EXISTS tmp_01375;
+DROP TABLE IF EXISTS table_tsv_01375;
+
+CREATE TABLE tmp_01375 (n UInt32, s String) ENGINE = Memory;
+CREATE TABLE table_tsv_01375 AS tmp_01375 ENGINE = File(TSVWithNames);
+
+INSERT INTO table_tsv_01375 SELECT number as n, toString(n) as s FROM numbers(10);
+INSERT INTO table_tsv_01375 SELECT number as n, toString(n) as s FROM numbers(10);
+INSERT INTO table_tsv_01375 SELECT number as n, toString(n) as s FROM numbers(10);
+
+SELECT * FROM table_tsv_01375;
+
+DROP TABLE IF EXISTS tmp_01375;
+DROP TABLE IF EXISTS table_tsv_01375;