C++: optimize batch read/write paths and aligned table null handling#823
Open
ColinLeeo wants to merge 2 commits into
Open
C++: optimize batch read/write paths and aligned table null handling#823ColinLeeo wants to merge 2 commits into
ColinLeeo wants to merge 2 commits into
Conversation
ColinLeeo
force-pushed
the
colin_all_opts
branch
2 times, most recently
from
69e1658 to
7db1a3a
Compare
ColinLeeo
changed the title
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## develop #823 +/- ##
===========================================
- Coverage 61.57% 58.67% -2.91%
===========================================
Files 731 733 +2
Lines 45874 47855 +1981
Branches 6880 7476 +596
===========================================
- Hits 28249 28078 -171
- Misses 16614 18577 +1963
- Partials 1011 1200 +189 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Contributor
There was a problem hiding this comment.
Pull request overview
This PR consolidates a large optimization branch into develop: it adds batch decode/write APIs across the C++ Decoder/Encoder hierarchy, multi-value aligned read paths with optional parallel decode, columnar tablet write helpers, SIMD fast paths, and a set of correctness fixes for aligned/table null handling (null TAG segments, null FIELD writes, all-null value pages, sparse aligned columns, repeated logical devices, ValuePageWriter::reset state). It also trims the C wrapper API (drops unused metadata export/tag-filter symbols, then re-adds tag-filter helpers in a different section) and removes several regression tests for behaviors it claims to fix.
Changes:
- Add batch decode/encode/write paths through
Decoder/Encoder/page/chunk writers and aMultiAlignedTimeseriesIndexplus single-device aligned fast-path reader. - Several aligned table fixes (null TAG/FIELD, all-null pages, single-device tablet flag,
ValuePageWriter::reset, double-free of first-page buffers viarelease_cur_page_data). - Build/infra: SIMD option, optional
BUILD_EXAMPLES, mem-stat counters widened to 64-bit, newBlockingQueue, removal of several existing regression tests, license-header punctuation churn in multiple CMake files.
Reviewed changes
Copilot reviewed 118 out of 119 changed files in this pull request and generated 12 comments.
Show a summary per file
| File | Description |
|---|---|
| cpp/src/reader/filter/{and,or}_filter.h, filter.h, time_operator.h | Adds satisfy_batch_time (uses fixed 129-element stack buffer — flagged). |
| cpp/src/encoding/{plain,decoder,encoder,plain_decoder,dictionary_encoder}.h | Batch encode/decode API + dictionary index assignment change (flagged). |
| cpp/src/writer/{value_,time_,}{chunk,page}_writer.{h,cc} | Batch write paths, first-page ownership transfer, larger page buffers. |
| cpp/src/writer/tsfile_table_writer.{h,cc}, tsfile_writer.h | Memoized lowercasing, idempotent close, optional parallel write pool. |
| cpp/src/reader/* | Aligned multi-value batch path, bloom-filter contains, table result-set lifecycle. |
| cpp/src/file/tsfile_io_writer.{h,cc}, restorable_tsfile_io_writer.cc | Recovery cleanup simplified; conditional sync_on_close_ (flagged); chunk-group index for O(1) lookup. |
| cpp/src/file/tsfile_io_reader.h | Device-node cache + multi-SSI alloc. |
| cpp/src/common/allocator/byte_stream.h, alloc_base.h, mem_alloc.cc | Page-mask bitwise modulo + power-of-2 rounding for wrapped buffers (flagged), 64-bit stat counters. |
| cpp/src/common/{tablet,schema,path,global,thread_pool}.* | Single-device flag, string-column uint32_t offsets, Path inlined, config knobs reshuffled. |
| cpp/src/common/container/{bit_map,blocking_queue,byte_buffer}.* | New BlockingQueue, BitMap::may_have_set_bits, bounds asserts. |
| cpp/src/compress/{snappy,lz4,uncompressed}_compressor.* | Safer after_compress ownership handling; Uncompressed now copies. |
| cpp/src/cwrapper/{tsfile_cwrapper.h,arrow_c.cc} | Tag-filter API moved, sliced-Arrow handling reverted (loses prior bug-fix paths). |
| cpp/test/** | Deletes several regression tests (deep path, missing measurement, aligned NULL boundary, dictionary RLE run counts, Arrow slice-with-offset, etc.) and adds new batch/page-boundary tests. |
| python/tsfile/dataset/reader.py + tests | Switches row reads to read_arrow_batch(). |
| cpp/{CMakeLists.txt,examples/**,src/CMakeLists.txt,src/common/CMakeLists.txt,test/CMakeLists.txt} | Build flags, SIMD option, Arrow/Parquet-dependent examples, license-header punctuation regressions. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+53
to
+63
| int satisfy_batch_time(const int64_t* times, int count, bool* mask) { | ||
| bool mask_right[129]; | ||
| left_->satisfy_batch_time(times, count, mask); | ||
| right_->satisfy_batch_time(times, count, mask_right); | ||
| int pass = 0; | ||
| for (int i = 0; i < count; ++i) { | ||
| mask[i] = mask[i] || mask_right[i]; | ||
| if (mask[i]) ++pass; | ||
| } | ||
| return pass; | ||
| } |
Comment on lines
+53
to
+63
| int satisfy_batch_time(const int64_t* times, int count, bool* mask) { | ||
| bool mask_right[129]; | ||
| left_->satisfy_batch_time(times, count, mask); | ||
| right_->satisfy_batch_time(times, count, mask_right); | ||
| int pass = 0; | ||
| for (int i = 0; i < count; ++i) { | ||
| mask[i] = mask[i] && mask_right[i]; | ||
| if (mask[i]) ++pass; | ||
| } | ||
| return pass; | ||
| } |
| index_entry_.push_back(value); | ||
| map_size_ = map_size_ + value.length(); | ||
| entry_index_[value] = static_cast<int>(index_entry_.size()) - 1; | ||
| entry_index_[value] = entry_index_.size(); |
Comment on lines
+295
to
303
| // page_mask_ is used as a bitmask and only works correctly for | ||
| // power-of-2 page sizes. Round up to the next power-of-2 so that | ||
| // (read_pos_ & page_mask_) gives the correct within-page offset and | ||
| // the page-crossing check doesn't misfire on arbitrary buffer sizes. | ||
| uint32_t ps = 1; | ||
| while (ps < (uint32_t)buf_len) ps <<= 1; | ||
| page_size_ = ps; | ||
| page_mask_ = ps - 1; | ||
| head_.store(&wrapped_page_); |
| } else if (RET_FAIL(write_file_footer())) { | ||
| std::cout << "writer file footer error, ret = " << ret << std::endl; | ||
| } else if (RET_FAIL(sync_file())) { | ||
| } else if (g_config_value_.sync_on_close_ && RET_FAIL(sync_file())) { |
| @@ -1,5 +1,5 @@ | |||
| #[[ | |||
| Licensed to the Apache Software Foundation (ASF) under one | |||
| Licensed to the Apache Software Foundation(ASF) under one | |||
|
|
||
| Unless required by applicable law or agreed to in writing, | ||
| software distributed under the License is distributed on an | ||
| software distributed under the LICENSE is distributed on an |
| file = write_file_new("test/", &error_no); | ||
| ASSERT_TRUE(error_no == RET_FILRET_OPEN_ERR || | ||
| error_no == RET_ALREADY_EXIST); | ||
| ASSERT_EQ(RET_FILRET_OPEN_ERR, error_no); |
Comment on lines
+1105
to
+1115
| @@ -1320,7 +1112,7 @@ TEST_F(TreeQueryByRowTest, DISABLED_QueryByRowFasterThanManualNext) { | |||
| write_test_file(devices, measurements, num_rows); | |||
|
|
|||
| const int num_iters = 5; | |||
| const double tolerance = 0.2; | |||
| const double tolerance = 0.05; | |||
Comment on lines
+119
to
+128
| uint32_t cur_points = value_page_writer_.get_point_numer(); | ||
| uint32_t page_remaining = | ||
| common::g_config_value_.page_writer_max_point_num_ - cur_points; | ||
| if (page_remaining == 0) { | ||
| if (RET_FAIL(seal_cur_page(false))) { | ||
| return ret; | ||
| } | ||
| page_remaining = | ||
| common::g_config_value_.page_writer_max_point_num_; | ||
| } |
Squashed PR snapshot of the long-lived `final` work, rebased on top of current develop (2a864c5). Combines the original "TsFile C++ batch read/write optimization" (5f12115) snapshot with subsequent build / platform fixes and a follow-up read-path optimization commit (c902b2b). ═════════════════════════════════════════════════════════════════════ Read path ═════════════════════════════════════════════════════════════════════ - Decoder base gains batch APIs (read_batch_int32 / int64 / float / double, skip_*); PLAIN, TS2DIFF, Gorilla decoders implement them. TS2DIFF has block-level peeking so time filters can skip blocks without decoding. Gorilla adds a raw-pointer GorillaBitReader that bypasses ByteStream overhead. - ChunkReader / AlignedChunkReader add *_DECODE_TV_BATCH methods that decode time + value into a TsBlock in one pass, applying batch time filters before append. - AlignedChunkReader supports a multi-value mode: one time chunk + N value chunks decoded in a single pass, sharing the decoded timestamps and filter mask. SingleDeviceTsBlockReader auto-detects same-device measurements via VectorMeasurementColumnContext. - Optional page-level parallel decompression via a DecodeThreadPool when ENABLE_THREADS is set. Page-plan classification (SKIP / FULL_PASS / BOUNDARY) lets a scatter-free memcpy fast path fire when every row passes and no column has nulls. Additional optimizations (from c902b2b, ported from `final`): - Aligned fast path: enable_dense_aligned_fast_path defaults true and compute_dense_row_count falls back to the TimeseriesIndex top-level statistic for single-chunk timeseries (chunk-level stat is omitted during serialization for those). Re-enables the bulk-copy SSI --> caller path that was defensively disabled. - Chunk-level parallel decode: per-column tasks own all that column's pages and write into a per-(col,page) PageDecodedState slot; one wait_all per chunk amortizes thread-pool overhead. Hybrid dispatch in get_next_page_multi — chunk-level for narrow chunks (<= 6 value columns), 4/6 thesis path otherwise to avoid cache thrash. - Per-worker time decoder/compressor pool (via ThreadPool:: current_worker_id) parallelizes the previously-serial time-page decode loop. - Pre-decode int64/float/double values in the parallel worker into ValueColumnState::pending_decoded_values; multi_DECODE_TV_BATCH then memcpys the per-batch slice instead of calling the decoder inline. - Partial-page bulk scatter: bulk-memcpy path now copies min(budget, remaining_in_page) rows from page_time_cursor_ so the tail page of every SSI tsblock takes the memcpy fast path instead of bleeding into the row-by-row scatter loop. - tsblock_max_memory_ 64KB -> 2MB so a 10K-row page fits in one SSI tsblock and bulk_copy_into doesn't fragment into many tiny batches. ═════════════════════════════════════════════════════════════════════ Write path ═════════════════════════════════════════════════════════════════════ - ValuePageWriter gains write_batch / write_string_batch that take timestamp + value + nullness arrays directly, removing the per-value append loop. Tablet exposes set_timestamps / set_column_values / set_column_string_repeated / reset for bulk reuse and switches StringColumn to an Arrow-compatible offset+buffer layout. - TS2DIFFEncoder::flush packs all deltas with a single pack_bits_msb + write_buf instead of per-value write_bits, falling back to the scalar path for the rare bit_width > 56 case. - Int64Statistic::update_batch (NEON-accelerated min/max/sum). ═════════════════════════════════════════════════════════════════════ Encoding / SIMD ═════════════════════════════════════════════════════════════════════ - TS2DIFF batch decode adds AVX2 helpers via SIMDe (already on develop) for both i32 and i64; scalar fallback unchanged. - PLAIN byte-swap path uses ARM NEON (vrev64q_u8 / vrev32q_u8) when available, falling back to __builtin_bswap. - CMakeLists adds ENABLE_SIMD; Release builds turn on -O3 -march=native -flto (off when ASan is on or on Windows/MinGW). ═════════════════════════════════════════════════════════════════════ Allocator / ByteStream / ThreadPool ═════════════════════════════════════════════════════════════════════ - ByteStream caches page_mask_ (= page_size - 1) so the hot path uses a bitmask instead of modulo; wrap_from rounds buffer sizes up to a power of two for correctness. - common::ThreadPool gets a thread_local current_worker_id() accessor (set by worker_loop) and a num_threads() getter, letting callers attach per-worker state without contention. ═════════════════════════════════════════════════════════════════════ Build / platform ═════════════════════════════════════════════════════════════════════ - Linux Release: -march=native + -flto by default, automatically dropped under ASan to keep leak detection accurate. - MSVC / MinGW: replace GCC-only intrinsics, restore lost includes, disable LTO + -march=native there. - Restore tag_filter_create/between, metadata test, and segment behavior; restore cwrapper metadata + tag_filter/batch_size args on table query C APIs that the batch-opt snapshot had dropped. - Disable QueryByRowPerformanceTest and the flaky QueryByRowFasterThanManualNext test. ═════════════════════════════════════════════════════════════════════ Python binding ═════════════════════════════════════════════════════════════════════ - read_series_by_row: pull TsBlocks via Arrow IPC instead of the row-by-row Python loop. Aligns reader query plumbing with develop so the binding sees the same parameter set. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
ColinLeeo
force-pushed
the
colin_all_opts
branch
2 times, most recently
from
043ce0e to
651f4af
Compare
The squash carried implicit reverts of every develop commit landed between the old branch base (e3cdf87) and current HEAD (2a864c5) — typo fixes (21988e7), the get_timeseries_metadata N+1 optimization (324acba), the TS_2DIFF float/double overflow-page fix (2a864c5), the 2.3.1 release-prep poms, and several regression tests in develop's reader/writer test suites. This commit: - restores 16 files the optimization branch never touched - 3-way merges 15 files both sides modified, keeping develop's typo fixes / N+1 optimization / TS_2DIFF overflow bitmaps / regression tests on top of the read/write batch optimization changes - keeps the small Windows-only compile fix in query_by_row_performance_test.cc and the zlib-1.2.13 -> 1.3.1 bump - restores the cpp/CMakeLists.txt TSFILE_OPTIMIZATION_FLAGS knob while keeping -O3 -march=native -flto as the Linux/macOS Release default
Comment on lines
836
to
+840
| case common::INT32: | ||
| case common::INT64: | ||
| case common::FLOAT: | ||
| case common::DOUBLE: { | ||
| size_t elem_size = | ||
| (dtype == common::INT64 || dtype == common::DOUBLE) ? 8 : 4; | ||
| const void* data = | ||
| static_cast<const char*>(col_arr->buffers[1]) + | ||
| off * elem_size; | ||
| uint8_t* null_bm = InvertArrowBitmap( | ||
| validity, off, static_cast<uint32_t>(n_rows)); | ||
| if (validity != nullptr && null_bm == nullptr) { | ||
| delete tablet; | ||
| return common::E_OOM; | ||
| // Invert Arrow bitmap (1=valid) to TsFile bitmap (1=null) |
Comment on lines
593
to
600
| virtual common::String get_measurement_name() const { | ||
| return value_ts_idx_->get_measurement_name(); | ||
| } | ||
| virtual common::TSDataType get_data_type() const { | ||
| return value_ts_idx_ == nullptr ? common::INVALID_DATATYPE | ||
| : value_ts_idx_->get_data_type(); | ||
| return time_ts_idx_->get_data_type(); | ||
| } | ||
| virtual bool is_aligned() const { return true; } | ||
| virtual Statistic* get_statistic() const { | ||
| return value_ts_idx_->get_statistic(); |
Comment on lines
674
to
678
| // timeseries measurenemnt chunk meta info | ||
| // map <device_name, <measurement_name, vector<chunk_meta>>> | ||
| std::map<std::shared_ptr<IDeviceID>, | ||
| std::map<common::String, std::vector<ChunkMeta*>>, | ||
| IDeviceIDComparator> | ||
| std::map<common::String, std::vector<ChunkMeta*>>> | ||
| tsm_chunk_meta_info_; |
Comment on lines
105
to
108
| if (!tmp.empty()) { | ||
| auto& merged = | ||
| tsm_chunk_meta_info_[chunk_group_meta_iter_.get()->device_id_]; | ||
| for (auto& m_entry : tmp) { | ||
| auto& vec = merged[m_entry.first]; | ||
| vec.insert(vec.end(), m_entry.second.begin(), | ||
| m_entry.second.end()); | ||
| } | ||
| tsm_chunk_meta_info_[chunk_group_meta_iter_.get()->device_id_] = | ||
| tmp; | ||
| } |
Comment on lines
+71
to
+75
| void after_uncompress(char* uncompressed_buf) { | ||
| if (uncompressed_buf != nullptr) { | ||
| common::mem_free(uncompressed_buf_); | ||
| uncompressed_buf_ = nullptr; | ||
| } |
Comment on lines
+69
to
+73
| if (!names_lowered_) { | ||
| tablet.set_table_name(to_lower(tablet.get_table_name())); | ||
| for (size_t i = 0; i < tablet.get_column_count(); i++) { | ||
| tablet.set_column_name(i, to_lower(tablet.get_column_name(i))); | ||
| } |
Comment on lines
+54
to
+66
| g_config_value_.float_encoding_type_ = PLAIN; | ||
| g_config_value_.double_encoding_type_ = PLAIN; | ||
| g_config_value_.string_encoding_type_ = PLAIN; | ||
| // Default compression type is LZ4 | ||
| #ifdef ENABLE_LZ4 | ||
| g_config_value_.default_compression_type_ = LZ4; | ||
| g_config_value_.default_compression_type_ = SNAPPY; | ||
| #else | ||
| g_config_value_.default_compression_type_ = UNCOMPRESSED; | ||
| #endif | ||
| unsigned int hw_cores = std::thread::hardware_concurrency(); | ||
| if (hw_cores == 0) hw_cores = 1; // fallback if detection fails | ||
| g_config_value_.parallel_write_enabled_ = (hw_cores > 1); | ||
| g_config_value_.write_thread_count_ = | ||
| static_cast<int32_t>(std::min(hw_cores, 64u)); | ||
| // Enforce aligned page size limits strictly by default. | ||
| g_config_value_.strict_page_size_ = true; | ||
| g_config_value_.parallel_read_enabled_ = true; | ||
| g_config_value_.parallel_write_enabled_ = true; | ||
| g_config_value_.read_thread_count_ = 4; | ||
| g_config_value_.write_thread_count_ = 6; |
Comment on lines
58
to
77
| void TsFileIOWriter::destroy() { | ||
| if (destroyed_) { | ||
| return; | ||
| } | ||
| // Recovery attaches a prefix of ChunkGroupMeta; device_id and chunk stats | ||
| // in that snapshot live in reader/recovery memory. After open, new chunks | ||
| // may be pushed into the same ChunkGroupMeta (same device); only those | ||
| // appended ChunkMeta need statistic_->destroy() (see | ||
| // recovery_chunk_meta_prefix_). | ||
| for (auto iter = chunk_group_meta_list_.begin(); | ||
| iter != chunk_group_meta_list_.end(); iter++) { | ||
| ChunkGroupMeta* cgm = iter.get(); | ||
| auto prefix_it = recovery_chunk_meta_prefix_.find(cgm); | ||
| const bool is_recovery_cgm = | ||
| chunk_group_meta_from_recovery_ && cgm != nullptr && | ||
| prefix_it != recovery_chunk_meta_prefix_.end(); | ||
| uint32_t recovered_cm_count = is_recovery_cgm ? prefix_it->second : 0; | ||
|
|
||
| if (!is_recovery_cgm) { | ||
| if (cgm != nullptr && cgm->device_id_) { | ||
| cgm->device_id_.reset(); | ||
| // When meta came from RestorableTsFileIOWriter recovery, entries live in | ||
| // an arena there; do not release device_id_/statistic_ here. | ||
| if (!chunk_group_meta_from_recovery_) { | ||
| for (auto iter = chunk_group_meta_list_.begin(); | ||
| iter != chunk_group_meta_list_.end(); iter++) { | ||
| if (iter.get() && iter.get()->device_id_) { | ||
| iter.get()->device_id_.reset(); | ||
| } | ||
| } | ||
|
|
||
| if (cgm == nullptr) { | ||
| continue; | ||
| } | ||
| uint32_t cm_idx = 0; | ||
| for (auto chunk_meta = cgm->chunk_meta_list_.begin(); | ||
| chunk_meta != cgm->chunk_meta_list_.end(); | ||
| chunk_meta++, cm_idx++) { | ||
| if (chunk_meta.get() == nullptr || | ||
| chunk_meta.get()->statistic_ == nullptr) { | ||
| continue; | ||
| } | ||
| if (is_recovery_cgm && cm_idx < recovered_cm_count) { | ||
| continue; | ||
| if (iter.get()) { | ||
| for (auto chunk_meta = iter.get()->chunk_meta_list_.begin(); | ||
| chunk_meta != iter.get()->chunk_meta_list_.end(); | ||
| chunk_meta++) { | ||
| if (chunk_meta.get()) { | ||
| chunk_meta.get()->statistic_->destroy(); | ||
| } | ||
| } | ||
| } | ||
| chunk_meta.get()->statistic_->destroy(); | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
This PR optimizes the C++ TsFile read/write paths for batch and columnar workloads, and fixes several aligned table null-handling issues uncovered while validating the optimized path.
It consolidates the batch decode/write work from the long-lived optimization branch into a reviewable change for
develop.Supersedes #749, #754, and #774.
Main Changes
Decoderhierarchy and implement batch paths for PLAIN, TS2DIFF, and Gorilla.ChunkReaderandAlignedChunkReader, including shared timestamp decoding for aligned multi-value reads.ByteStream, compressor, and page/chunk writer internals used by the optimized paths.Correctness Fixes
ValuePageWriter::reset()so row count and null bitmap state are reset together.Compatibility Notes
cpp/third_party/is left atdevelopstate so existing platform compatibility fixes are preserved.Verification
cmake --build cpp/target/build --target TsFile_Test -j1ctest --test-dir cpp/target/build/test --output-on-failure -R '^TsFileTableReaderTest\.TestNullInTable4$'ctest --test-dir cpp/target/build/test --output-on-failure -j4cd cpp && mvn spotless:checkcd cpp && mvn apache-rat:checkCurrent full C++ test result:
496/496tests pass.