Merge pull request ClickHouse#88827 from ClickHouse/pqe

Enable parquet reader v3 by default

Author: nikitamikhaylov

src/Common/threadPoolCallbackRunner.cpp

@@ -69,6 +69,7 @@ void ThreadPoolCallbackRunnerFast::operator()(std::function<void()> f)
     {
         std::unique_lock lock(mutex);
         queue.push_back(std::move(f));
+
         startMoreThreadsIfNeeded(active_tasks_, lock);
     }
 
@@ -92,23 +93,35 @@ void ThreadPoolCallbackRunnerFast::bulkSchedule(std::vector<std::function<void()
     if (mode == Mode::Disabled)
         throw Exception(ErrorCodes::LOGICAL_ERROR, "Thread pool runner is not initialized");
 
-    size_t active_tasks_ = fs.size() + active_tasks.fetch_add(fs.size(), std::memory_order_relaxed);
+    size_t n = fs.size();
+    size_t active_tasks_ = n + active_tasks.fetch_add(n, std::memory_order_relaxed);
 
     {
         std::unique_lock lock(mutex);
         queue.insert(queue.end(), std::move_iterator(fs.begin()), std::move_iterator(fs.end()));
-        startMoreThreadsIfNeeded(active_tasks_, lock);
+
+        try
+        {
+            startMoreThreadsIfNeeded(active_tasks_, lock);
+        }
+        catch (...)
+        {
+            /// Keep `queue` consistent with `queue_size`.
+            queue.erase(queue.end() - n, queue.end());
+            active_tasks.fetch_sub(n, std::memory_order_relaxed);
+            throw;
+        }
     }
 
     if (mode == Mode::ThreadPool)
     {
 #ifdef OS_LINUX
-        UInt32 prev_size = queue_size.fetch_add(fs.size(), std::memory_order_release);
+        UInt32 prev_size = queue_size.fetch_add(n, std::memory_order_release);
         if (prev_size < max_threads)
-            futexWake(&queue_size, fs.size());
+            futexWake(&queue_size, n);
 #else
-        if (fs.size() < 4)
-            for (size_t i = 0; i < fs.size(); ++i)
+        if (n < 4)
+            for (size_t i = 0; i < n; ++i)
                 queue_cv.notify_one();
         else
             queue_cv.notify_all();

src/Core/FormatFactorySettings.h

@@ -182,9 +182,9 @@ When reading Parquet files, parse JSON columns as ClickHouse JSON Column.
     DECLARE(Bool, input_format_parquet_use_native_reader, false, R"(
 Use native parquet reader v1. It's relatively fast but unfinished. Deprecated.
 )", 0) \
-    DECLARE(Bool, input_format_parquet_use_native_reader_v3, false, R"(
-Use Parquet reader v3. Experimental.
-)", EXPERIMENTAL) \
+    DECLARE(Bool, input_format_parquet_use_native_reader_v3, true, R"(
+Use Parquet reader v3.
+)", 0) \
     DECLARE(UInt64, input_format_parquet_memory_low_watermark, 2ul << 20, R"(
 Schedule prefetches more aggressively if memory usage is below than threshold. Potentially useful e.g. if there are many small bloom filters to read over network.
 )", 0) \

src/Core/SettingsChangesHistory.cpp

@@ -56,6 +56,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
             {"iceberg_timezone_for_timestamptz", "UTC", "UTC", "New setting."},
             {"hybrid_table_auto_cast_columns", true, true, "New setting to automatically cast Hybrid table columns when segments disagree on types. Default enabled."},
             {"allow_experimental_hybrid_table", false, false, "Added new setting to allow the Hybrid table engine."},
+            {"input_format_parquet_use_native_reader_v3", false, true, "Seems stable"},
             {"input_format_parquet_verify_checksums", true, true, "New setting."},
             {"output_format_parquet_write_checksums", false, true, "New setting."},
         });
@@ -82,7 +83,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
             {"distributed_cache_connect_max_tries", 20, 5, "Changed setting value"},
             {"opentelemetry_trace_cpu_scheduling", false, false, "New setting to trace `cpu_slot_preemption` feature."},
             {"output_format_parquet_max_dictionary_size", 1024 * 1024, 1024 * 1024, "New setting"},
-            {"input_format_parquet_use_native_reader_v3", false, true, "New setting"},
+            {"input_format_parquet_use_native_reader_v3", false, false, "New setting"},
             {"input_format_parquet_memory_low_watermark", 2ul << 20, 2ul << 20, "New setting"},
             {"input_format_parquet_memory_high_watermark", 4ul << 30, 4ul << 30, "New setting"},
             {"input_format_parquet_page_filter_push_down", true, true, "New setting (no effect when input_format_parquet_use_native_reader_v3 is disabled)"},

src/Processors/Formats/Impl/Parquet/Decoding.cpp

@@ -1003,7 +1003,7 @@ void Dictionary::index(const ColumnUInt32 & indexes_col, IColumn & out)
             c.reserve(c.size() + indexes.size());
             for (UInt32 idx : indexes)
             {
-                size_t start = offsets[size_t(idx) - 1] + 4; // offsets[-1] is ok because of padding
+                size_t start = offsets[ssize_t(idx) - 1] + 4; // offsets[-1] is ok because of padding
                 size_t len = offsets[idx] - start;
                 /// TODO [parquet]: Try optimizing short memcpy by taking advantage of padding (maybe memcpySmall.h helps). Also in PlainStringDecoder.
                 c.insertData(data.data() + start, len);
@@ -1219,7 +1219,7 @@ void TrivialStringConverter::convertColumn(std::span<const char> chars, const UI
     {
         col_str.getChars().reserve(col_str.getChars().size() + (offsets[num_values - 1] - offsets[-1]) - separator_bytes * num_values);
         for (size_t i = 0; i < num_values; ++i)
-            col_str.insertData(chars.data() + offsets[i - 1], offsets[i] - offsets[i - 1] - separator_bytes);
+            col_str.insertData(chars.data() + offsets[ssize_t(i) - 1], offsets[i] - offsets[ssize_t(i) - 1] - separator_bytes);
     }
 }
 
@@ -1345,8 +1345,8 @@ void BigEndianDecimalStringConverter<T>::convertColumn(std::span<const char> cha
 
     for (size_t i = 0; i < num_values; ++i)
     {
-        const char * data = chars.data() + offsets[i - 1];
-        size_t size = offsets[i] - offsets[i - 1] - separator_bytes;
+        const char * data = chars.data() + offsets[ssize_t(i) - 1];
+        size_t size = offsets[i] - offsets[ssize_t(i) - 1] - separator_bytes;
         if (size > sizeof(T))
             throw Exception(ErrorCodes::CANNOT_PARSE_NUMBER, "Unexpectedly wide Decimal value: {} > {} bytes", size, sizeof(T));
 

src/Processors/Formats/Impl/Parquet/Prefetcher.cpp

@@ -74,7 +74,7 @@ void Prefetcher::determineReadModeAndFileSize(ReadBuffer * reader_, const ReadOp
         if (!reader_->eof() && reader_->available() >= expected_prefix.size() &&
             memcmp(reader_->position(), expected_prefix.data(), expected_prefix.size()) != 0)
         {
-            throw Exception(ErrorCodes::INCORRECT_DATA, "Not a parquet file (wrong magic bytes at the start)");
+            throw Exception(ErrorCodes::INCORRECT_DATA, "Not a Parquet file (wrong magic bytes at the start)");
         }
 
         WriteBufferFromVector<PaddedPODArray<char>> out(entire_file);

src/Processors/Formats/Impl/Parquet/ReadManager.cpp

@@ -846,7 +846,7 @@ ReadManager::ReadResult ReadManager::read()
             bool thread_pool_was_idle = parser_shared_resources->parsing_runner.isIdle();
 
             if (exception)
-                std::rethrow_exception(exception);
+                std::rethrow_exception(copyMutableException(exception));
 
             /// If `preserve_order`, only deliver chunks from `first_incomplete_row_group`.
             /// This ensures that row groups are delivered in order. Within a row group, row

src/Processors/Formats/Impl/Parquet/Reader.cpp

@@ -15,6 +15,7 @@
 #include <Storages/SelectQueryInfo.h>
 
 #include <lz4.h>
+#include <arrow/util/crc32.h>
 
 #if USE_SNAPPY
 #include <snappy.h>
@@ -28,6 +29,7 @@ namespace DB::ErrorCodes
     extern const int INCORRECT_DATA;
     extern const int LOGICAL_ERROR;
     extern const int NOT_IMPLEMENTED;
+    extern const int CHECKSUM_DOESNT_MATCH;
 }
 
 namespace DB::Parquet
@@ -176,7 +178,7 @@ parq::FileMetaData Reader::readFileMetaData(Prefetcher & prefetcher)
     prefetcher.readSync(buf.data(), initial_read_size, file_size - initial_read_size);
 
     if (memcmp(buf.data() + initial_read_size - 4, "PAR1", 4) != 0)
-        throw Exception(ErrorCodes::INCORRECT_DATA, "Not a parquet file (wrong magic bytes at the end of file)");
+        throw Exception(ErrorCodes::INCORRECT_DATA, "Not a Parquet file (wrong magic bytes at the end of file)");
 
     int32_t metadata_size_i32;
     memcpy(&metadata_size_i32, buf.data() + initial_read_size - 8, 4);
@@ -216,7 +218,7 @@ parq::FileMetaData Reader::readFileMetaData(Prefetcher & prefetcher)
     ///      present. Instead, data_page_offset points to the dictionary page.
     ///  (2) Old DuckDB versions (<= 0.10.2) wrote incorrect data_page_offset when dictionary is
     ///      present.
-    /// We work around (1) in initializePage by allowing dictionary page in place of data page.
+    /// We work around (1) in initializeDataPage by allowing dictionary page in place of data page.
     /// We work around (2) here by converting it to case (1):
     ///   data_page_offset = dictionary_page_offset
     ///   dictionary_page_offset.reset()
@@ -756,8 +758,9 @@ void Reader::processBloomFilterHeader(ColumnChunk & column, const PrimitiveColum
 bool Reader::decodeDictionaryPage(ColumnChunk & column, const PrimitiveColumnInfo & column_info)
 {
     auto data = prefetcher.getRangeData(column.dictionary_page_prefetch);
-    parq::PageHeader header;
-    size_t header_size = deserializeThriftStruct(header, data.data(), data.size());
+    const char * data_ptr = data.data();
+    const char * data_end = data.data() + data.size();
+    auto [header, page_data] = decodeAndCheckPageHeader(data_ptr, data_end);
 
     if (header.type != parq::PageType::DICTIONARY_PAGE)
     {
@@ -768,15 +771,14 @@ bool Reader::decodeDictionaryPage(ColumnChunk & column, const PrimitiveColumnInf
         return false;
     }
 
-    decodeDictionaryPageImpl(header, data.subspan(header_size), column, column_info);
+    decodeDictionaryPageImpl(header, page_data, column, column_info);
     return true;
 }
 
 void Reader::decodeDictionaryPageImpl(const parq::PageHeader & header, std::span<const char> data, ColumnChunk & column, const PrimitiveColumnInfo & column_info)
 {
     chassert(header.type == parq::PageType::DICTIONARY_PAGE);
 
-    /// TODO [parquet]: Check checksum.
     size_t compressed_page_size = size_t(header.compressed_page_size);
     if (header.compressed_page_size < 0 || compressed_page_size > data.size())
         throw Exception(ErrorCodes::INCORRECT_DATA, "Dictionary page size out of bounds: {} > {}", header.compressed_page_size, data.size());
@@ -1381,7 +1383,7 @@ void Reader::skipToRow(size_t row_idx, ColumnChunk & column, const PrimitiveColu
 
         auto data = prefetcher.getRangeData(page_info.prefetch);
         const char * ptr = data.data();
-        if (!initializePage(ptr, ptr + data.size(), first_row_idx, page_info.end_row_idx, row_idx, column, column_info))
+        if (!initializeDataPage(ptr, ptr + data.size(), first_row_idx, page_info.end_row_idx, row_idx, column, column_info))
             throw Exception(ErrorCodes::LOGICAL_ERROR, "Page doesn't contain requested row");
         found_page = true;
     }
@@ -1403,12 +1405,33 @@ void Reader::skipToRow(size_t row_idx, ColumnChunk & column, const PrimitiveColu
         chassert(column.next_page_offset <= all_pages.size());
         const char * ptr = all_pages.data() + column.next_page_offset;
         const char * end = all_pages.data() + all_pages.size();
-        initializePage(ptr, end, page.next_row_idx, /*end_row_idx=*/ std::nullopt, row_idx, column, column_info);
+        initializeDataPage(ptr, end, page.next_row_idx, /*end_row_idx=*/ std::nullopt, row_idx, column, column_info);
         column.next_page_offset = ptr - all_pages.data();
     }
 }
 
-bool Reader::initializePage(const char * & data_ptr, const char * data_end, size_t next_row_idx, std::optional<size_t> end_row_idx, size_t target_row_idx, ColumnChunk & column, const PrimitiveColumnInfo & column_info)
+std::tuple<parq::PageHeader, std::span<const char>> Reader::decodeAndCheckPageHeader(const char * & data_ptr, const char * data_end) const
+{
+    parq::PageHeader header;
+    data_ptr += deserializeThriftStruct(header, data_ptr, data_end - data_ptr);
+    size_t compressed_page_size = size_t(header.compressed_page_size);
+    if (header.compressed_page_size < 0 || compressed_page_size > size_t(data_end - data_ptr))
+        throw Exception(ErrorCodes::INCORRECT_DATA, "Page size out of bounds: {} > {}", header.compressed_page_size, data_end - data_ptr);
+
+    std::span page_data(data_ptr, compressed_page_size);
+    data_ptr += compressed_page_size;
+
+    if (header.__isset.crc && options.format.parquet.verify_checksums)
+    {
+        uint32_t crc = arrow::internal::crc32(0, page_data.data(), page_data.size());
+        if (crc != uint32_t(header.crc))
+            throw Exception(ErrorCodes::CHECKSUM_DOESNT_MATCH, "Page CRC checksum verification failed");
+    }
+
+    return {header, page_data};
+}
+
+bool Reader::initializeDataPage(const char * & data_ptr, const char * data_end, size_t next_row_idx, std::optional<size_t> end_row_idx, size_t target_row_idx, ColumnChunk & column, const PrimitiveColumnInfo & column_info)
 {
     PageState & page = column.page;
     /// We reuse PageState instance across pages to reuse memory in buffers like decompressed_buf.
@@ -1425,13 +1448,7 @@ bool Reader::initializePage(const char * & data_ptr, const char * data_end, size
     /// Decode page header.
 
     parq::PageHeader header;
-    data_ptr += deserializeThriftStruct(header, data_ptr, data_end - data_ptr);
-    /// TODO [parquet]: Check checksum.
-    size_t compressed_page_size = size_t(header.compressed_page_size);
-    if (header.compressed_page_size < 0 || compressed_page_size > size_t(data_end - data_ptr))
-        throw Exception(ErrorCodes::INCORRECT_DATA, "Page size out of bounds: {} > {}", header.compressed_page_size, data_end - data_ptr);
-    page.data = std::span(data_ptr, compressed_page_size);
-    data_ptr += compressed_page_size;
+    std::tie(header, page.data) = decodeAndCheckPageHeader(data_ptr, data_end);
 
     /// Check if all rows of the page are filtered out, if we have enough information.
 
@@ -1525,7 +1542,7 @@ bool Reader::initializePage(const char * & data_ptr, const char * data_end, size
             page.codec = parq::CompressionCodec::UNCOMPRESSED;
         }
 
-        if (encoded_def_size + encoded_rep_size > compressed_page_size)
+        if (encoded_def_size + encoded_rep_size > page.data.size())
             throw Exception(ErrorCodes::INCORRECT_DATA, "Page data is too short (def+rep)");
         encoded_rep = page.data.data();
         encoded_def = page.data.data() + encoded_rep_size;

src/Processors/Formats/Impl/Parquet/Reader.h

@@ -61,7 +61,7 @@ namespace DB::Parquet
 //     - no columns to read outside prewhere
 //     - no columns to read, but not trivial count either
 //     - ROW POLICY, with and without prewhere, with old and new reader
-//     - prewhere with defaults (it probably doesn't fill them correctly, see MergeTreeRangeReader::executeActionsBeforePrewhere)
+//     - prewhere and other skipping with defaults (it probably doesn't fill them correctly, see MergeTreeRangeReader::executeActionsBeforePrewhere)
 //     - prewhere on virtual columns (do they end up in additional_columns?)
 //     - prewhere with weird filter type (LowCardinality(UInt8), Nullable(UInt8), const UInt8)
 //     - prewhere involving arrays and tuples
@@ -523,7 +523,8 @@ struct Reader
     double estimateAverageStringLengthPerRow(const ColumnChunk & column, const RowGroup & row_group) const;
     void decodeDictionaryPageImpl(const parq::PageHeader & header, std::span<const char> data, ColumnChunk & column, const PrimitiveColumnInfo & column_info);
     void skipToRow(size_t row_idx, ColumnChunk & column, const PrimitiveColumnInfo & column_info);
-    bool initializePage(const char * & data_ptr, const char * data_end, size_t next_row_idx, std::optional<size_t> end_row_idx, size_t target_row_idx, ColumnChunk & column, const PrimitiveColumnInfo & column_info);
+    std::tuple<parq::PageHeader, std::span<const char>> decodeAndCheckPageHeader(const char * & data_ptr, const char * data_end) const;
+    bool initializeDataPage(const char * & data_ptr, const char * data_end, size_t next_row_idx, std::optional<size_t> end_row_idx, size_t target_row_idx, ColumnChunk & column, const PrimitiveColumnInfo & column_info);
     void decompressPageIfCompressed(PageState & page);
     void createPageDecoder(PageState & page, ColumnChunk & column, const PrimitiveColumnInfo & column_info);
     bool skipRowsInPage(size_t target_row_idx, PageState & page, ColumnChunk & column, const PrimitiveColumnInfo & column_info);

src/Processors/Formats/Impl/Parquet/SchemaConverter.cpp

@@ -133,7 +133,12 @@ std::string_view SchemaConverter::useColumnMapperIfNeeded(const parq::SchemaElem
         return element.name;
     const auto & map = column_mapper->getFieldIdToClickHouseName();
     if (!element.__isset.field_id)
-        throw Exception(ErrorCodes::ICEBERG_SPECIFICATION_VIOLATION, "Missing field_id for column {}", element.name);
+    {
+        /// Does iceberg require that parquet files have field ids?
+        /// Our iceberg writer currently doesn't write them.
+        //throw Exception(ErrorCodes::ICEBERG_SPECIFICATION_VIOLATION, "Missing field_id for column {}", element.name);
+        return element.name;
+    }
     auto it = map.find(element.field_id);
     if (it == map.end())
         throw Exception(ErrorCodes::ICEBERG_SPECIFICATION_VIOLATION, "Parquet file has column {} with field_id {} that is not in datalake metadata", element.name, element.field_id);
@@ -270,15 +275,19 @@ void SchemaConverter::processSubtree(TraversalNode & node)
     }
 }
 
-bool SchemaConverter::processSubtreePrimitive(TraversalNode & node)
+static bool isPrimitiveNode(const parq::SchemaElement & elem)
 {
     /// `parquet.thrift` says "[num_children] is not set when the element is a primitive type".
-    /// If it's set but has value 0, logically it would make sense to interpret it as empty tuple/struct.
+    /// If it's set but has value 0, logically it should be an empty tuple/struct.
     /// But in practice some writers are sloppy about it and set this field to 0 (rather than unset)
     /// for primitive columns. E.g.
     /// tests/queries/0_stateless/data_hive/partitioning/non_existing_column=Elizabeth/sample.parquet
-    bool is_primitive = !node.element->__isset.num_children || (node.element->num_children == 0 && node.element->__isset.type);
-    if (!is_primitive)
+    return !elem.__isset.num_children || (elem.num_children == 0 && elem.__isset.type);
+}
+
+bool SchemaConverter::processSubtreePrimitive(TraversalNode & node)
+{
+    if (!isPrimitiveNode(*node.element))
         return false;
 
     primitive_column_idx += 1;
@@ -468,13 +477,18 @@ bool SchemaConverter::processSubtreeMap(TraversalNode & node)
 bool SchemaConverter::processSubtreeArrayOuter(TraversalNode & node)
 {
     /// Array:
-    ///   required group `name` (List):
+    ///   required/optional group `name` (List):
     ///     repeated group "list":
     ///       <recurse> "element"
     ///
     /// I.e. it's a double-wrapped burrito. To unwrap it into one Array, we have to coordinate
     /// across two levels of recursion: processSubtreeArrayOuter for the outer wrapper,
     /// processSubtreeArrayInner for the inner wrapper.
+    ///
+    /// But hudi writes arrays differently, without the inner group:
+    ///   required/optional group `name` (List):
+    ///     repeated <recurse> "array"
+    /// This probably makes it indinsinguishable from a single-element tuple.
 
     if (node.element->converted_type != parq::ConvertedType::LIST && !node.element->logicalType.__isset.LIST)
         return false;
@@ -483,10 +497,12 @@ bool SchemaConverter::processSubtreeArrayOuter(TraversalNode & node)
     if (node.element->num_children != 1)
         return false;
     const parq::SchemaElement & child = file_metadata.schema.at(schema_idx);
-    if (child.repetition_type != parq::FieldRepetitionType::REPEATED || child.num_children != 1)
+    if (child.repetition_type != parq::FieldRepetitionType::REPEATED)
         return false;
 
-    TraversalNode subnode = node.prepareToRecurse(SchemaContext::ListTuple, node.type_hint);
+    bool has_inner_group = child.num_children == 1;
+
+    TraversalNode subnode = node.prepareToRecurse(has_inner_group ? SchemaContext::ListTuple : SchemaContext::ListElement, node.type_hint);
     processSubtree(subnode);
 
     if (!node.requested || !subnode.output_idx.has_value())