22.8.20 Pre-release (2)

docker/packager/binary/Dockerfile

@@ -55,7 +55,7 @@ RUN arch=${TARGETARCH:-amd64} \
   && dpkg -i /tmp/nfpm.deb \
   && rm /tmp/nfpm.deb
 
-ARG GO_VERSION=1.18.3
+ARG GO_VERSION=1.19.8
 # We need go for clickhouse-diagnostics
 RUN arch=${TARGETARCH:-amd64} \
   && curl -Lo /tmp/go.tgz "https://go.dev/dl/go${GO_VERSION}.linux-${arch}.tar.gz" \

docker/test/integration/runner/dockerd-entrypoint.sh

@@ -27,6 +27,7 @@ while true; do
     reties=$((reties+1))
     if [[ $reties -ge 100 ]]; then # 10 sec max
         echo "Can't start docker daemon, timeout exceeded." >&2
+        cat /ClickHouse/tests/integration/dockerd.log >&2
         exit 1;
     fi
     sleep 0.1

packages/clickhouse-common-static.yaml

@@ -33,8 +33,9 @@ deb:
 contents:
 - src: root/usr/bin/clickhouse
   dst: /usr/bin/clickhouse
-- src: root/usr/bin/clickhouse-diagnostics
-  dst: /usr/bin/clickhouse-diagnostics
+# Excluded due to CVEs in go runtime that popup constantly
+# - src: root/usr/bin/clickhouse-diagnostics
+#   dst: /usr/bin/clickhouse-diagnostics
 - src: root/usr/bin/clickhouse-extract-from-config
   dst: /usr/bin/clickhouse-extract-from-config
 - src: root/usr/bin/clickhouse-library-bridge

src/CMakeLists.txt

@@ -602,6 +602,7 @@ if (ENABLE_TESTS)
         dbms
         clickhouse_common_config
         clickhouse_common_zookeeper
+        ch_contrib::parquet
         string_utils)
 
     if (TARGET ch_contrib::simdjson)

src/Core/Settings.h

@@ -712,6 +712,7 @@ static constexpr UInt64 operator""_GiB(unsigned long long value)
     M(Bool, input_format_csv_use_best_effort_in_schema_inference, true, "Use some tweaks and heuristics to infer schema in CSV format", 0) \
     M(Bool, input_format_tsv_use_best_effort_in_schema_inference, true, "Use some tweaks and heuristics to infer schema in TSV format", 0) \
     M(Bool, input_format_parquet_skip_columns_with_unsupported_types_in_schema_inference, false, "Skip columns with unsupported types while schema inference for format Parquet", 0) \
+    M(UInt64, input_format_parquet_max_block_size, 8192, "Max block size for parquet reader.", 0) \
     M(Bool, input_format_protobuf_skip_fields_with_unsupported_types_in_schema_inference, false, "Skip fields with unsupported types while schema inference for format Protobuf", 0) \
     M(Bool, input_format_capn_proto_skip_fields_with_unsupported_types_in_schema_inference, false, "Skip columns with unsupported types while schema inference for format CapnProto", 0) \
     M(Bool, input_format_orc_skip_columns_with_unsupported_types_in_schema_inference, false, "Skip columns with unsupported types while schema inference for format ORC", 0) \

src/Disks/IDisk.h

@@ -71,6 +71,7 @@ class Space : public std::enable_shared_from_this<Space>
     virtual const String & getName() const = 0;
 
     /// Reserve the specified number of bytes.
+    /// Returns valid reservation or nullptr when failure.
     virtual ReservationPtr reserve(UInt64 bytes) = 0;
 
     virtual ~Space() = default;

src/Disks/VolumeJBOD.h

@@ -82,6 +82,9 @@ class VolumeJBOD : public IVolume
         ReservationPtr reserve(uint64_t bytes)
         {
             ReservationPtr reservation = disk->reserve(bytes);
+            if (!reservation)
+                return {};
+
             /// Not just subtract bytes, but update the value,
             /// since some reservations may be done directly via IDisk, or not by ClickHouse.
             free_size = reservation->getUnreservedSpace();

src/Formats/FormatFactory.cpp

@@ -104,6 +104,7 @@ FormatSettings getFormatSettings(ContextPtr context, const Settings & settings)
     format_settings.parquet.allow_missing_columns = settings.input_format_parquet_allow_missing_columns;
     format_settings.parquet.skip_columns_with_unsupported_types_in_schema_inference = settings.input_format_parquet_skip_columns_with_unsupported_types_in_schema_inference;
     format_settings.parquet.output_string_as_string = settings.output_format_parquet_string_as_string;
+    format_settings.parquet.max_block_size = settings.input_format_parquet_max_block_size;
     format_settings.pretty.charset = settings.output_format_pretty_grid_charset.toString() == "ASCII" ? FormatSettings::Pretty::Charset::ASCII : FormatSettings::Pretty::Charset::UTF8;
     format_settings.pretty.color = settings.output_format_pretty_color;
     format_settings.pretty.max_column_pad_width = settings.output_format_pretty_max_column_pad_width;

src/Formats/FormatSettings.h

@@ -160,6 +160,7 @@ struct FormatSettings
         bool case_insensitive_column_matching = false;
         std::unordered_set<int> skip_row_groups = {};
         bool output_string_as_string = false;
+        UInt64 max_block_size = 8192;
     } parquet;
 
     struct Pretty

src/Interpreters/convertFieldToType.cpp

@@ -234,10 +234,11 @@ Field convertFieldToTypeImpl(const Field & src, const IDataType & type, const ID
         }
 
         if (which_type.isDateTime64()
-            && (which_from_type.isNativeInt() || which_from_type.isNativeUInt() || which_from_type.isDate() || which_from_type.isDate32() || which_from_type.isDateTime() || which_from_type.isDateTime64()))
+            && (src.getType() == Field::Types::UInt64 || src.getType() == Field::Types::Int64 || src.getType() == Field::Types::Decimal64))
         {
             const auto scale = static_cast<const DataTypeDateTime64 &>(type).getScale();
-            const auto decimal_value = DecimalUtils::decimalFromComponents<DateTime64>(applyVisitor(FieldVisitorConvertToNumber<Int64>(), src), 0, scale);
+            const auto decimal_value
+                = DecimalUtils::decimalFromComponents<DateTime64>(applyVisitor(FieldVisitorConvertToNumber<Int64>(), src), 0, scale);
             return Field(DecimalField<DateTime64>(decimal_value, scale));
         }
     }

src/Processors/Formats/Impl/ArrowBlockInputFormat.cpp

@@ -71,7 +71,7 @@ Chunk ArrowBlockInputFormat::generate()
 
     ++record_batch_current;
 
-    arrow_column_to_ch_column->arrowTableToCHChunk(res, *table_result);
+    arrow_column_to_ch_column->arrowTableToCHChunk(res, *table_result, (*table_result)->num_rows());
 
     /// If defaults_for_omitted_fields is true, calculate the default values from default expression for omitted fields.
     /// Otherwise fill the missing columns with zero values of its type.

src/Processors/Formats/Impl/ArrowBufferedStreams.cpp

@@ -11,6 +11,7 @@
 #include <IO/copyData.h>
 #include <IO/PeekableReadBuffer.h>
 #include <arrow/buffer.h>
+#include <arrow/util/future.h>
 #include <arrow/io/memory.h>
 #include <arrow/result.h>
 #include <Core/Settings.h>
@@ -95,6 +96,12 @@ arrow::Result<std::shared_ptr<arrow::Buffer>> RandomAccessFileFromSeekableReadBu
     return buffer;
 }
 
+arrow::Future<std::shared_ptr<arrow::Buffer>> RandomAccessFileFromSeekableReadBuffer::ReadAsync(const arrow::io::IOContext &, int64_t position, int64_t nbytes)
+{
+    /// Just a stub to to avoid using internal arrow thread pool
+    return arrow::Future<std::shared_ptr<arrow::Buffer>>::MakeFinished(ReadAt(position, nbytes));
+}
+
 arrow::Status RandomAccessFileFromSeekableReadBuffer::Seek(int64_t position)
 {
     seekable_in.seek(position, SEEK_SET);

src/Processors/Formats/Impl/ArrowBufferedStreams.h

@@ -62,6 +62,11 @@ class RandomAccessFileFromSeekableReadBuffer : public arrow::io::RandomAccessFil
 
     arrow::Result<std::shared_ptr<arrow::Buffer>> Read(int64_t nbytes) override;
 
+    /// Override async reading to avoid using internal arrow thread pool.
+    /// In our code we don't use async reading, so implementation is sync,
+    /// we just call ReadAt and return future with ready value.
+    arrow::Future<std::shared_ptr<arrow::Buffer>> ReadAsync(const arrow::io::IOContext&, int64_t position, int64_t nbytes) override;
+
     arrow::Status Seek(int64_t position) override;
 
 private:

src/Processors/Formats/Impl/ArrowColumnToCHColumn.cpp

@@ -69,7 +69,6 @@ namespace ErrorCodes
     extern const int DUPLICATE_COLUMN;
     extern const int THERE_IS_NO_COLUMN;
     extern const int UNKNOWN_EXCEPTION;
-    extern const int INCORRECT_NUMBER_OF_COLUMNS;
     extern const int INCORRECT_DATA;
 }
 
@@ -90,7 +89,7 @@ static ColumnWithTypeAndName readColumnWithNumericData(std::shared_ptr<arrow::Ch
 
         /// buffers[0] is a null bitmap and buffers[1] are actual values
         std::shared_ptr<arrow::Buffer> buffer = chunk->data()->buffers[1];
-        const auto * raw_data = reinterpret_cast<const NumericType *>(buffer->data());
+        const auto * raw_data = reinterpret_cast<const NumericType *>(buffer->data()) + chunk->offset();
         column_data.insert_assume_reserved(raw_data, raw_data + chunk->length());
     }
     return {std::move(internal_column), std::move(internal_type), column_name};
@@ -347,7 +346,7 @@ static ColumnWithTypeAndName readColumnWithIndexesDataImpl(std::shared_ptr<arrow
 
         /// buffers[0] is a null bitmap and buffers[1] are actual values
         std::shared_ptr<arrow::Buffer> buffer = chunk->data()->buffers[1];
-        const auto * data = reinterpret_cast<const NumericType *>(buffer->data());
+        const auto * data = reinterpret_cast<const NumericType *>(buffer->data()) + chunk->offset();
 
         /// Check that indexes are correct (protection against corrupted files)
         for (int64_t i = 0; i != chunk->length(); ++i)
@@ -767,7 +766,7 @@ ArrowColumnToCHColumn::ArrowColumnToCHColumn(
 {
 }
 
-void ArrowColumnToCHColumn::arrowTableToCHChunk(Chunk & res, std::shared_ptr<arrow::Table> & table)
+void ArrowColumnToCHColumn::arrowTableToCHChunk(Chunk & res, std::shared_ptr<arrow::Table> & table, size_t num_rows)
 {
     NameToColumnPtr name_to_column_ptr;
     for (auto column_name : table->ColumnNames())
@@ -781,16 +780,12 @@ void ArrowColumnToCHColumn::arrowTableToCHChunk(Chunk & res, std::shared_ptr<arr
         name_to_column_ptr[std::move(column_name)] = arrow_column;
     }
 
-    arrowColumnsToCHChunk(res, name_to_column_ptr);
+    arrowColumnsToCHChunk(res, name_to_column_ptr, num_rows);
 }
 
-void ArrowColumnToCHColumn::arrowColumnsToCHChunk(Chunk & res, NameToColumnPtr & name_to_column_ptr)
+void ArrowColumnToCHColumn::arrowColumnsToCHChunk(Chunk & res, NameToColumnPtr & name_to_column_ptr, size_t num_rows)
 {
-    if (unlikely(name_to_column_ptr.empty()))
-        throw Exception(ErrorCodes::INCORRECT_NUMBER_OF_COLUMNS, "Columns is empty");
-
     Columns columns_list;
-    UInt64 num_rows = name_to_column_ptr.begin()->second->length();
     columns_list.reserve(header.columns());
     std::unordered_map<String, std::pair<BlockPtr, std::shared_ptr<NestedColumnExtractHelper>>> nested_tables;
     bool skipped = false;

src/Processors/Formats/Impl/ArrowColumnToCHColumn.h

@@ -28,9 +28,9 @@ class ArrowColumnToCHColumn
         bool allow_missing_columns_,
         bool case_insensitive_matching_ = false);
 
-    void arrowTableToCHChunk(Chunk & res, std::shared_ptr<arrow::Table> & table);
+    void arrowTableToCHChunk(Chunk & res, std::shared_ptr<arrow::Table> & table, size_t num_rows);
 
-    void arrowColumnsToCHChunk(Chunk & res, NameToColumnPtr & name_to_column_ptr);
+    void arrowColumnsToCHChunk(Chunk & res, NameToColumnPtr & name_to_column_ptr, size_t num_rows);
 
     /// Get missing columns that exists in header but not in arrow::Schema
     std::vector<size_t> getMissingColumns(const arrow::Schema & schema) const;

src/Processors/Formats/Impl/ORCBlockInputFormat.cpp

@@ -54,14 +54,19 @@ Chunk ORCBlockInputFormat::generate()
         throw ParsingException(
             ErrorCodes::CANNOT_READ_ALL_DATA, "Error while reading batch of ORC data: {}", table_result.status().ToString());
 
+    /// We should extract the number of rows directly from the stripe, because in case when
+    /// record batch contains 0 columns (for example if we requested only columns that
+    /// are not presented in data) the number of rows in record batch will be 0.
+    size_t num_rows = file_reader->GetRawORCReader()->getStripe(stripe_current)->getNumberOfRows();
+
     auto table = table_result.ValueOrDie();
-    if (!table || !table->num_rows())
+    if (!table || !num_rows)
         return {};
 
     ++stripe_current;
 
     Chunk res;
-    arrow_column_to_ch_column->arrowTableToCHChunk(res, table);
+    arrow_column_to_ch_column->arrowTableToCHChunk(res, table, num_rows);
     /// If defaults_for_omitted_fields is true, calculate the default values from default expression for omitted fields.
     /// Otherwise fill the missing columns with zero values of its type.
     if (format_settings.defaults_for_omitted_fields)

src/Processors/Formats/Impl/ParquetBlockInputFormat.cpp

@@ -16,6 +16,7 @@
 #include "ArrowColumnToCHColumn.h"
 #include <DataTypes/NestedUtils.h>
 
+
 namespace DB
 {
 
@@ -43,34 +44,38 @@ Chunk ParquetBlockInputFormat::generate()
     block_missing_values.clear();
 
     if (!file_reader)
+    {
         prepareReader();
+        file_reader->set_batch_size(format_settings.parquet.max_block_size);
+        std::vector<int> row_group_indices;
+        for (int i = 0; i < row_group_total; ++i)
+        {
+            if (!skip_row_groups.contains(i))
+                row_group_indices.emplace_back(i);
+        }
+        auto read_status = file_reader->GetRecordBatchReader(row_group_indices, column_indices, &current_record_batch_reader);
+        if (!read_status.ok())
+            throw DB::ParsingException(ErrorCodes::CANNOT_READ_ALL_DATA, "Error while reading Parquet data: {}", read_status.ToString());
+    }
 
     if (is_stopped)
         return {};
 
-    for (; row_group_current < row_group_total && skip_row_groups.contains(row_group_current); ++row_group_current)
-        ;
-
-    if (row_group_current >= row_group_total)
-        return res;
-
-    std::shared_ptr<arrow::Table> table;
-
-    std::unique_ptr<::arrow::RecordBatchReader> rbr;
-    std::vector<int> row_group_indices { row_group_current };
-    arrow::Status get_batch_reader_status = file_reader->GetRecordBatchReader(row_group_indices, column_indices, &rbr);
-
-    if (!get_batch_reader_status.ok())
-        throw ParsingException{"Error while reading Parquet data: " + get_batch_reader_status.ToString(), ErrorCodes::CANNOT_READ_ALL_DATA};
-
-    arrow::Status read_status = rbr->ReadAll(&table);
-
-    if (!read_status.ok())
-        throw ParsingException{"Error while reading Parquet data: " + read_status.ToString(), ErrorCodes::CANNOT_READ_ALL_DATA};
-
-    ++row_group_current;
-
-    arrow_column_to_ch_column->arrowTableToCHChunk(res, table);
+    auto batch = current_record_batch_reader->Next();
+    if (!batch.ok())
+    {
+        throw ParsingException(ErrorCodes::CANNOT_READ_ALL_DATA, "Error while reading Parquet data: {}",
+                               batch.status().ToString());
+    }
+    if (*batch)
+    {
+        auto tmp_table = arrow::Table::FromRecordBatches({*batch});
+        arrow_column_to_ch_column->arrowTableToCHChunk(res, *tmp_table, (*tmp_table)->num_rows());
+    }
+    else
+    {
+        return {};
+    }
 
     /// If defaults_for_omitted_fields is true, calculate the default values from default expression for omitted fields.
     /// Otherwise fill the missing columns with zero values of its type.
@@ -85,6 +90,7 @@ void ParquetBlockInputFormat::resetParser()
     IInputFormat::resetParser();
 
     file_reader.reset();
+    current_record_batch_reader.reset();
     column_indices.clear();
     row_group_current = 0;
     block_missing_values.clear();

src/Processors/Formats/Impl/ParquetBlockInputFormat.h

@@ -8,7 +8,7 @@
 
 namespace parquet::arrow { class FileReader; }
 
-namespace arrow { class Buffer; }
+namespace arrow { class Buffer; class RecordBatchReader;}
 
 namespace DB
 {
@@ -46,6 +46,7 @@ class ParquetBlockInputFormat : public IInputFormat
     BlockMissingValues block_missing_values;
     const FormatSettings format_settings;
     const std::unordered_set<int> & skip_row_groups;
+    std::shared_ptr<arrow::RecordBatchReader> current_record_batch_reader;
 
     std::atomic<int> is_stopped{0};
 };

src/Processors/tests/gtest_assert_arrow_log_does_not_abort.cpp

@@ -0,0 +1,18 @@
+#include <gtest/gtest.h>
+#include <arrow/chunked_array.h>
+#include <vector>
+#include <arrow/util/logging.h>
+
+using namespace DB;
+
+TEST(ChunkedArray, ChunkedArrayWithZeroChunksShouldNotAbort)
+{
+    std::vector<std::shared_ptr<::arrow::Array>> empty_chunks_vector;
+
+    EXPECT_ANY_THROW(::arrow::ChunkedArray{empty_chunks_vector});
+}
+
+TEST(ArrowLog, FatalLogShouldThrow)
+{
+    EXPECT_ANY_THROW(::arrow::util::ArrowLog(__FILE__, __LINE__, ::arrow::util::ArrowLogLevel::ARROW_FATAL));
+}

src/Storages/MergeTree/KeyCondition.cpp

@@ -1172,7 +1172,8 @@ bool KeyCondition::transformConstantWithValidFunctions(
 
             if (is_valid_chain)
             {
-                auto const_type = cur_node->result_type;
+                out_type = removeLowCardinality(out_type);
+                auto const_type = removeLowCardinality(cur_node->result_type);
                 auto const_column = out_type->createColumnConst(1, out_value);
                 auto const_value = (*castColumnAccurateOrNull({const_column, out_type, ""}, const_type))[0];
 

tests/ci/build_check.py

@@ -394,6 +394,18 @@ def main():
 
     print(f"::notice ::Log URL: {log_url}")
 
+    src_path = os.path.join(TEMP_PATH, "build_source.src.tar.gz")
+
+    if os.path.exists(src_path):
+        src_url = s3_helper.upload_build_file_to_s3(
+            src_path, s3_path_prefix + "/clickhouse-" + version.string + ".src.tar.gz"
+        )
+        logging.info("Source tar %s", src_url)
+    else:
+        logging.info("Source tar doesn't exist")
+
+    print(f"::notice ::Source tar URL: {src_url}")
+
     create_json_artifact(
         TEMP_PATH, build_name, log_url, build_urls, build_config, elapsed, success
     )

tests/ci/ci_config.py

@@ -335,6 +335,9 @@
             "required_build": "package_aarch64",
             "test_grep_exclude_filter": "constant_column_search",
         },
+        "Sign release (actions)": {
+            "required_build": "package_release"
+        }
     },
 }  # type: dict
 

tests/ci/clickhouse_helper.py

@@ -197,7 +197,7 @@ def mark_flaky_tests(clickhouse_helper, check_name, test_results):
     AND pull_request_number = 0
 """
 
-        tests_data = clickhouse_helper.select_json_each_row("default", query)
+        tests_data = clickhouse_helper.select_json_each_row("gh-data", query)
         master_failed_tests = {row["test_name"] for row in tests_data}
         logging.info("Found flaky tests: %s", ", ".join(master_failed_tests))