| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * License); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| |
| #include "bench_read.h" |
| |
| #include <fcntl.h> |
| #include <sys/stat.h> |
| |
| #include <chrono> |
| #include <iomanip> |
| #include <iostream> |
| #include <memory> |
| #include <numeric> |
| #include <string> |
| #include <vector> |
| |
| #include <arrow/api.h> |
| #include <arrow/io/api.h> |
| #include <parquet/arrow/reader.h> |
| #include <parquet/arrow/writer.h> |
| #include <parquet/metadata.h> |
| #include <parquet/properties.h> |
| #include <parquet/statistics.h> |
| |
| #include "common/schema.h" |
| #include "common/tablet.h" |
| #include "common/tsblock/tsblock.h" |
| #include "common/tsblock/vector/fixed_length_vector.h" |
| #include "common/tsblock/vector/vector.h" |
| #include "file/write_file.h" |
| #include "reader/filter/tag_filter.h" |
| #include "reader/result_set.h" |
| #include "reader/table_result_set.h" |
| #include "reader/tsfile_reader.h" |
| #include "utils/util_define.h" |
| #include "writer/tsfile_table_writer.h" |
| |
| #define BENCH_HANDLE_ERROR(err_no) \ |
| do { \ |
| if ((err_no) != 0) { \ |
| std::cerr << "tsfile err " \ |
| << (err_no) << "\n"; \ |
| return (err_no); \ |
| } \ |
| } while (0) |
| |
| #define BENCH_CHECK_RET_NEG1(expr) \ |
| do { \ |
| int _ts_err = (expr); \ |
| if (_ts_err != 0) { \ |
| std::cerr << "tsfile err " << _ts_err << "\n"; \ |
| return -1; \ |
| } \ |
| } while (0) |
| |
| namespace { |
| |
| static const char* kTable = "bench_table"; |
| static const char* kTag2Val = "tag_b"; |
| static const int kNumDevices = 10; |
| static const char* kFilterDevice = "device_0"; |
| |
| static const std::vector<std::string> kReadCols{ |
| "id1", "id2", "s1", "s2", "s3", "s4"}; |
| |
| static std::string device_name(int i) { |
| return "device_" + std::to_string(i); |
| } |
| |
| // ─── Cache drop ────────────────────────────────────────────────────────────── |
| |
| void bench_drop_cache() { |
| #if defined(__APPLE__) |
| if (system("sudo purge") != 0) { |
| std::cerr << "[bench] purge failed or not available " |
| "(run `sudo purge` manually before bench_read)\n"; |
| } |
| #elif defined(__linux__) |
| if (system("sync && sudo sh -c 'echo 3 > /proc/sys/vm/drop_caches'") != 0) { |
| std::cerr << "[bench] drop_caches failed " |
| "(run `sudo sh -c 'echo 3 > /proc/sys/vm/drop_caches'` manually)\n"; |
| } |
| #else |
| std::cerr << "[bench] bench_drop_cache not supported on this platform\n"; |
| #endif |
| } |
| |
| // ─── Write ──────────────────────────────────────────────────────────────────── |
| |
| int write_tsfile(const std::string& path, int64_t row_count) { |
| storage::libtsfile_init(); |
| storage::WriteFile file; |
| int flags = O_WRONLY | O_CREAT | O_TRUNC; |
| #ifdef _WIN32 |
| flags |= O_BINARY; |
| #endif |
| BENCH_HANDLE_ERROR(file.create(path.c_str(), flags, 0666)); |
| |
| auto* schema = new storage::TableSchema( |
| std::string(kTable), |
| { |
| common::ColumnSchema("id1", common::STRING, |
| common::UNCOMPRESSED, common::PLAIN, |
| common::ColumnCategory::TAG), |
| common::ColumnSchema("id2", common::STRING, |
| common::UNCOMPRESSED, common::PLAIN, |
| common::ColumnCategory::TAG), |
| common::ColumnSchema("s1", common::INT64, |
| common::SNAPPY, common::PLAIN, |
| common::ColumnCategory::FIELD), |
| common::ColumnSchema("s2", common::DOUBLE, |
| common::SNAPPY, common::PLAIN, |
| common::ColumnCategory::FIELD), |
| common::ColumnSchema("s3", common::FLOAT, |
| common::SNAPPY, common::PLAIN, |
| common::ColumnCategory::FIELD), |
| common::ColumnSchema("s4", common::INT32, |
| common::SNAPPY, common::PLAIN, |
| common::ColumnCategory::FIELD), |
| }); |
| |
| auto* writer = new storage::TsFileTableWriter(&file, schema); |
| const uint32_t batch_cap = 65536; |
| int64_t rows_per_dev = row_count / kNumDevices; |
| |
| for (int dev = 0; dev < kNumDevices; dev++) { |
| std::string dev_id = device_name(dev); |
| int64_t dev_base = dev * rows_per_dev; |
| |
| for (int64_t off = 0; off < rows_per_dev;) { |
| uint32_t n = static_cast<uint32_t>( |
| std::min<int64_t>(batch_cap, rows_per_dev - off)); |
| storage::Tablet tablet( |
| kTable, |
| {"id1", "id2", "s1", "s2", "s3", "s4"}, |
| {common::STRING, common::STRING, common::INT64, |
| common::DOUBLE, common::FLOAT, common::INT32}, |
| {common::ColumnCategory::TAG, common::ColumnCategory::TAG, |
| common::ColumnCategory::FIELD, common::ColumnCategory::FIELD, |
| common::ColumnCategory::FIELD, common::ColumnCategory::FIELD}, |
| std::max(n, 1u)); |
| for (uint32_t i = 0; i < n; i++) { |
| int64_t ts = dev_base + off + i; |
| BENCH_HANDLE_ERROR(tablet.add_timestamp(i, ts)); |
| BENCH_HANDLE_ERROR(tablet.add_value(i, "id1", dev_id.c_str())); |
| BENCH_HANDLE_ERROR(tablet.add_value(i, "id2", kTag2Val)); |
| BENCH_HANDLE_ERROR(tablet.add_value(i, "s1", ts)); |
| BENCH_HANDLE_ERROR(tablet.add_value(i, "s2", ts * 1.1)); |
| BENCH_HANDLE_ERROR(tablet.add_value(i, "s3", static_cast<float>(ts % 10000))); |
| BENCH_HANDLE_ERROR(tablet.add_value(i, "s4", static_cast<int32_t>(ts % 100000))); |
| } |
| BENCH_HANDLE_ERROR(writer->write_table(tablet)); |
| off += n; |
| } |
| } |
| BENCH_HANDLE_ERROR(writer->flush()); |
| BENCH_HANDLE_ERROR(writer->close()); |
| delete writer; |
| delete schema; |
| return 0; |
| } |
| |
| int write_parquet(const std::string& path, int64_t row_count) { |
| try { |
| auto schema = arrow::schema({ |
| arrow::field("time", arrow::int64()), |
| arrow::field("id1", arrow::utf8()), |
| arrow::field("id2", arrow::utf8()), |
| arrow::field("s1", arrow::int64()), |
| arrow::field("s2", arrow::float64()), |
| arrow::field("s3", arrow::float32()), |
| arrow::field("s4", arrow::int32()), |
| }); |
| |
| auto writer_props = parquet::WriterProperties::Builder() |
| .compression(parquet::Compression::SNAPPY) |
| ->build(); |
| auto arrow_props = parquet::ArrowWriterProperties::Builder().build(); |
| |
| const int64_t batch_cap = 65536; |
| int64_t rows_per_dev = row_count / kNumDevices; |
| arrow::MemoryPool* pool = arrow::default_memory_pool(); |
| |
| PARQUET_ASSIGN_OR_THROW(auto out, |
| arrow::io::FileOutputStream::Open(path)); |
| PARQUET_ASSIGN_OR_THROW( |
| std::unique_ptr<parquet::arrow::FileWriter> pw, |
| parquet::arrow::FileWriter::Open(*schema, pool, out, |
| writer_props, arrow_props)); |
| |
| for (int dev = 0; dev < kNumDevices; dev++) { |
| std::string dev_id = device_name(dev); |
| int64_t dev_base = dev * rows_per_dev; |
| |
| arrow::Int64Builder time_b; |
| arrow::StringBuilder id1_b; |
| arrow::StringBuilder id2_b; |
| arrow::Int64Builder s1_b; |
| arrow::DoubleBuilder s2_b; |
| arrow::FloatBuilder s3_b; |
| arrow::Int32Builder s4_b; |
| |
| for (int64_t off = 0; off < rows_per_dev;) { |
| int64_t n = std::min(batch_cap, rows_per_dev - off); |
| time_b.Reset(); id1_b.Reset(); id2_b.Reset(); |
| s1_b.Reset(); s2_b.Reset(); s3_b.Reset(); s4_b.Reset(); |
| for (int64_t i = 0; i < n; i++) { |
| int64_t ts = dev_base + off + i; |
| PARQUET_THROW_NOT_OK(time_b.Append(ts)); |
| PARQUET_THROW_NOT_OK(id1_b.Append(dev_id)); |
| PARQUET_THROW_NOT_OK(id2_b.Append(kTag2Val)); |
| PARQUET_THROW_NOT_OK(s1_b.Append(ts)); |
| PARQUET_THROW_NOT_OK(s2_b.Append(ts * 1.1)); |
| PARQUET_THROW_NOT_OK(s3_b.Append(static_cast<float>(ts % 10000))); |
| PARQUET_THROW_NOT_OK(s4_b.Append(static_cast<int32_t>(ts % 100000))); |
| } |
| PARQUET_ASSIGN_OR_THROW(auto a_time, time_b.Finish()); |
| PARQUET_ASSIGN_OR_THROW(auto a_id1, id1_b.Finish()); |
| PARQUET_ASSIGN_OR_THROW(auto a_id2, id2_b.Finish()); |
| PARQUET_ASSIGN_OR_THROW(auto a_s1, s1_b.Finish()); |
| PARQUET_ASSIGN_OR_THROW(auto a_s2, s2_b.Finish()); |
| PARQUET_ASSIGN_OR_THROW(auto a_s3, s3_b.Finish()); |
| PARQUET_ASSIGN_OR_THROW(auto a_s4, s4_b.Finish()); |
| auto batch = arrow::RecordBatch::Make( |
| schema, n, |
| {a_time, a_id1, a_id2, a_s1, a_s2, a_s3, a_s4}); |
| PARQUET_THROW_NOT_OK(pw->WriteRecordBatch(*batch)); |
| off += n; |
| } |
| } |
| PARQUET_THROW_NOT_OK(pw->Close()); |
| PARQUET_THROW_NOT_OK(out->Close()); |
| return 0; |
| } catch (const std::exception& e) { |
| std::cerr << "parquet write: " << e.what() << "\n"; |
| return 1; |
| } |
| } |
| |
| // ─── Helpers ────────────────────────────────────────────────────────────────── |
| |
| static void print_result(const char* engine, double secs, |
| int64_t result_rows, int64_t checksum) { |
| std::cout << " " << std::left << std::setw(16) << engine |
| << std::fixed << std::setprecision(4) << secs << " s | " |
| << std::right << std::setw(12) |
| << static_cast<int64_t>(result_rows / secs) << " rows/s" |
| << " | sum_s1=" << checksum << "\n"; |
| } |
| |
| // ─── Scenario 1: Tag Filter ─────────────────────────────────────────────────── |
| |
| int64_t tsfile_tag_filter(const std::string& path, int64_t row_count) { |
| storage::libtsfile_init(); |
| storage::TsFileReader reader; |
| BENCH_CHECK_RET_NEG1(reader.open(path)); |
| |
| auto table_schema = reader.get_table_schema(std::string(kTable)); |
| storage::Filter* tag_filter = |
| storage::TagFilterBuilder(table_schema.get()).eq("id1", kFilterDevice); |
| |
| storage::ResultSet* rs = nullptr; |
| BENCH_CHECK_RET_NEG1( |
| reader.query(kTable, kReadCols, 0, row_count, rs, tag_filter)); |
| |
| int64_t sum = 0; |
| bool has_next = false; |
| int ret = common::E_OK; |
| while (IS_SUCC(ret = rs->next(has_next)) && has_next) { |
| if (!rs->is_null("s1")) { |
| sum += rs->get_value<int64_t>("s1"); |
| } |
| } |
| rs->close(); |
| reader.close(); |
| delete tag_filter; |
| return sum; |
| } |
| |
| // Collect row group indices whose statistics overlap the given string equality. |
| // Equivalent to TsFile's device-level chunk pruning. |
| static std::vector<int> rg_prune_string_eq( |
| const parquet::FileMetaData& meta, int col_idx, const std::string& target) { |
| std::vector<int> result; |
| for (int rg = 0; rg < meta.num_row_groups(); ++rg) { |
| auto stats = meta.RowGroup(rg)->ColumnChunk(col_idx)->statistics(); |
| if (stats && stats->HasMinMax()) { |
| auto s = std::static_pointer_cast<parquet::ByteArrayStatistics>(stats); |
| std::string mn(reinterpret_cast<const char*>(s->min().ptr), s->min().len); |
| std::string mx(reinterpret_cast<const char*>(s->max().ptr), s->max().len); |
| if (target < mn || target > mx) continue; // prune |
| } |
| result.push_back(rg); |
| } |
| return result; |
| } |
| |
| // Collect row group indices whose time range overlaps [ts_start, ts_end). |
| // Equivalent to TsFile's page-level time statistics pruning. |
| static std::vector<int> rg_prune_time_range( |
| const parquet::FileMetaData& meta, int col_idx, |
| int64_t ts_start, int64_t ts_end) { |
| std::vector<int> result; |
| for (int rg = 0; rg < meta.num_row_groups(); ++rg) { |
| auto stats = meta.RowGroup(rg)->ColumnChunk(col_idx)->statistics(); |
| if (stats && stats->HasMinMax()) { |
| auto s = std::static_pointer_cast<parquet::Int64Statistics>(stats); |
| if (s->max() < ts_start || s->min() >= ts_end) continue; // prune |
| } |
| result.push_back(rg); |
| } |
| return result; |
| } |
| |
| int64_t parquet_tag_filter(const std::string& path) { |
| try { |
| std::vector<std::string> cols{"time", "id1", "id2", "s1", "s2", "s3", "s4"}; |
| arrow::MemoryPool* pool = arrow::default_memory_pool(); |
| PARQUET_ASSIGN_OR_THROW(auto infile, |
| arrow::io::ReadableFile::Open(path)); |
| PARQUET_ASSIGN_OR_THROW( |
| std::unique_ptr<parquet::arrow::FileReader> reader, |
| parquet::arrow::OpenFile(infile, pool)); |
| |
| std::shared_ptr<arrow::Schema> file_schema; |
| PARQUET_THROW_NOT_OK(reader->GetSchema(&file_schema)); |
| std::vector<int> indices; |
| for (const auto& name : cols) |
| indices.push_back(file_schema->GetFieldIndex(name)); |
| |
| // Row group pruning via min/max statistics on id1 column. |
| auto& meta = *reader->parquet_reader()->metadata(); |
| int id1_col = meta.schema()->ColumnIndex("id1"); |
| auto matching_rgs = rg_prune_string_eq(meta, id1_col, kFilterDevice); |
| |
| PARQUET_ASSIGN_OR_THROW(auto batch_reader, |
| reader->GetRecordBatchReader(matching_rgs, indices)); |
| |
| int64_t sum = 0; |
| std::shared_ptr<arrow::RecordBatch> batch; |
| while (batch_reader->ReadNext(&batch).ok() && batch) { |
| auto id1_arr = std::static_pointer_cast<arrow::StringArray>( |
| batch->GetColumnByName("id1")); |
| auto s1_arr = std::static_pointer_cast<arrow::Int64Array>( |
| batch->GetColumnByName("s1")); |
| for (int64_t i = 0; i < batch->num_rows(); ++i) { |
| if (!id1_arr->IsNull(i) && |
| id1_arr->GetString(i) == kFilterDevice && |
| !s1_arr->IsNull(i)) { |
| sum += s1_arr->Value(i); |
| } |
| } |
| } |
| return sum; |
| } catch (const std::exception& e) { |
| std::cerr << "parquet tag filter: " << e.what() << "\n"; |
| return -1; |
| } |
| } |
| |
| // ─── Scenario 2: Time Range Filter ─────────────────────────────────────────── |
| |
| // TsFile query(start, end) is inclusive on both sides: [start, end]. |
| // Pass (ts_end - 1) to match Parquet's half-open [ts_start, ts_end) semantics. |
| int64_t tsfile_time_filter(const std::string& path, |
| int64_t ts_start, int64_t ts_end) { |
| storage::libtsfile_init(); |
| storage::TsFileReader reader; |
| BENCH_CHECK_RET_NEG1(reader.open(path)); |
| |
| storage::ResultSet* rs = nullptr; |
| BENCH_CHECK_RET_NEG1( |
| reader.query(kTable, kReadCols, ts_start, ts_end - 1, rs, nullptr)); |
| |
| int64_t sum = 0; |
| bool has_next = false; |
| int ret = common::E_OK; |
| while (IS_SUCC(ret = rs->next(has_next)) && has_next) { |
| if (!rs->is_null("s1")) sum += rs->get_value<int64_t>("s1"); |
| } |
| rs->close(); |
| reader.close(); |
| return sum; |
| } |
| |
| int64_t parquet_time_filter(const std::string& path, |
| int64_t ts_start, int64_t ts_end) { |
| try { |
| std::vector<std::string> cols{"time", "id1", "id2", "s1", "s2", "s3", "s4"}; |
| arrow::MemoryPool* pool = arrow::default_memory_pool(); |
| PARQUET_ASSIGN_OR_THROW(auto infile, |
| arrow::io::ReadableFile::Open(path)); |
| PARQUET_ASSIGN_OR_THROW( |
| std::unique_ptr<parquet::arrow::FileReader> reader, |
| parquet::arrow::OpenFile(infile, pool)); |
| |
| std::shared_ptr<arrow::Schema> file_schema; |
| PARQUET_THROW_NOT_OK(reader->GetSchema(&file_schema)); |
| std::vector<int> indices; |
| for (const auto& name : cols) |
| indices.push_back(file_schema->GetFieldIndex(name)); |
| |
| // Row group pruning via min/max statistics on time column. |
| auto& meta = *reader->parquet_reader()->metadata(); |
| int time_col = meta.schema()->ColumnIndex("time"); |
| auto matching_rgs = rg_prune_time_range(meta, time_col, ts_start, ts_end); |
| |
| PARQUET_ASSIGN_OR_THROW(auto batch_reader, |
| reader->GetRecordBatchReader(matching_rgs, indices)); |
| |
| int64_t sum = 0; |
| std::shared_ptr<arrow::RecordBatch> batch; |
| while (batch_reader->ReadNext(&batch).ok() && batch) { |
| auto time_arr = std::static_pointer_cast<arrow::Int64Array>( |
| batch->GetColumnByName("time")); |
| auto s1_arr = std::static_pointer_cast<arrow::Int64Array>( |
| batch->GetColumnByName("s1")); |
| for (int64_t i = 0; i < batch->num_rows(); ++i) { |
| int64_t t = time_arr->Value(i); |
| if (t >= ts_start && t < ts_end && !s1_arr->IsNull(i)) |
| sum += s1_arr->Value(i); |
| } |
| } |
| return sum; |
| } catch (const std::exception& e) { |
| std::cerr << "parquet time filter: " << e.what() << "\n"; |
| return -1; |
| } |
| } |
| |
| // ─── Optimized: Batch columnar read ────────────────────────────────────────── |
| |
| // Find the 0-based TsBlock vector index for a named column. |
| // ResultSetMetadata prepends "time" as column 1 (1-indexed), so |
| // TsBlock vector index = metadata column index - 1. |
| static int find_vec_idx(storage::ResultSet* rs, const std::string& name) { |
| auto meta = rs->get_metadata(); |
| for (int i = 1; i <= static_cast<int>(meta->get_column_count()); ++i) { |
| if (meta->get_column_name(i) == name) return i - 1; |
| } |
| return -1; |
| } |
| |
| // Sum all INT64 values in a Vector, using direct buffer access for the |
| // common no-null case to avoid per-element overhead. |
| static int64_t sum_vec_int64(common::Vector* vec, uint32_t rows) { |
| int64_t sum = 0; |
| if (!vec->has_null()) { |
| // Fast path: dense int64_t array, single pointer scan. |
| const int64_t* p = reinterpret_cast<const int64_t*>( |
| vec->get_value_data().get_data()); |
| for (uint32_t r = 0; r < rows; ++r) sum += p[r]; |
| } else { |
| // Slow path: skip null rows; advance sequential cursor manually. |
| vec->reset_offset(); |
| for (uint32_t r = 0; r < rows; ++r) { |
| if (!vec->is_null(r)) { |
| uint32_t len = 0; |
| bool null = false; |
| char* val = vec->read(&len, &null, r); |
| sum += *reinterpret_cast<int64_t*>(val); |
| vec->update_offset(); |
| } |
| } |
| } |
| return sum; |
| } |
| |
| // batch_size controls TsBlock capacity; 65536 rows/block matches write batches. |
| static const int kBatchSize = 65536; |
| |
| int64_t tsfile_tag_filter_batch(const std::string& path, int64_t row_count) { |
| storage::libtsfile_init(); |
| storage::TsFileReader reader; |
| BENCH_CHECK_RET_NEG1(reader.open(path)); |
| |
| auto table_schema = reader.get_table_schema(std::string(kTable)); |
| storage::Filter* tag_filter = |
| storage::TagFilterBuilder(table_schema.get()).eq("id1", kFilterDevice); |
| |
| storage::ResultSet* rs = nullptr; |
| BENCH_CHECK_RET_NEG1( |
| reader.query(kTable, kReadCols, 0, row_count, rs, tag_filter, kBatchSize)); |
| |
| const int s1_idx = find_vec_idx(rs, "s1"); |
| int64_t sum = 0; |
| common::TsBlock* block = nullptr; |
| while (rs->get_next_tsblock(block) == common::E_OK && block) { |
| sum += sum_vec_int64(block->get_vector(s1_idx), block->get_row_count()); |
| } |
| rs->close(); |
| reader.close(); |
| delete tag_filter; |
| return sum; |
| } |
| |
| int64_t tsfile_time_filter_batch(const std::string& path, |
| int64_t ts_start, int64_t ts_end) { |
| storage::libtsfile_init(); |
| storage::TsFileReader reader; |
| BENCH_CHECK_RET_NEG1(reader.open(path)); |
| |
| storage::ResultSet* rs = nullptr; |
| BENCH_CHECK_RET_NEG1( |
| reader.query(kTable, kReadCols, ts_start, ts_end - 1, rs, kBatchSize)); |
| |
| const int s1_idx = find_vec_idx(rs, "s1"); |
| int64_t sum = 0; |
| common::TsBlock* block = nullptr; |
| while (rs->get_next_tsblock(block) == common::E_OK && block) { |
| sum += sum_vec_int64(block->get_vector(s1_idx), block->get_row_count()); |
| } |
| rs->close(); |
| reader.close(); |
| return sum; |
| } |
| |
| } // namespace |
| |
| // ─── Entry point ───────────────────────────────────────────────────────────── |
| |
| int bench_write(int64_t row_count, bool run_parquet) { |
| const std::string ts_path = "read_perf_bench.tsfile"; |
| const std::string pq_path = "read_perf_bench.parquet"; |
| |
| std::cout << "rows_total=" << row_count |
| << " devices=" << kNumDevices |
| << " rows_per_device=" << row_count / kNumDevices |
| << "\ncolumns: time, id1, id2, s1(INT64), s2(DOUBLE)," |
| " s3(FLOAT), s4(INT32)\ncompression: SNAPPY\n"; |
| |
| { |
| using clock = std::chrono::high_resolution_clock; |
| auto t0 = clock::now(); |
| if (write_tsfile(ts_path, row_count) != 0) return 1; |
| double s = std::chrono::duration<double>(clock::now() - t0).count(); |
| std::cout << "write TsFile : " << std::fixed << std::setprecision(3) |
| << s << " s\n"; |
| } |
| if (run_parquet) { |
| using clock = std::chrono::high_resolution_clock; |
| auto t0 = clock::now(); |
| if (write_parquet(pq_path, row_count) != 0) return 1; |
| double s = std::chrono::duration<double>(clock::now() - t0).count(); |
| std::cout << "write Parquet : " << std::fixed << std::setprecision(3) |
| << s << " s\n"; |
| } |
| std::cout << "\n"; |
| return 0; |
| } |
| |
| int bench_read(int64_t row_count, bool run_parquet) { |
| int64_t rows_per_device = row_count / kNumDevices; |
| // TIME_FILTER: query the first 1/3 of the total time range. |
| // Timestamps are laid out as [0, row_count) across all devices. |
| int64_t time_range_start = 0; |
| int64_t time_range_end = row_count / 3; // ~333K rows for 1M total |
| int64_t time_result_rows = time_range_end - time_range_start; |
| |
| const std::string ts_path = "read_perf_bench.tsfile"; |
| const std::string pq_path = "read_perf_bench.parquet"; |
| |
| std::cout << "\n"; |
| |
| using clock = std::chrono::high_resolution_clock; |
| |
| // ── Scenario 1: Tag Filter ──────────────────────────────────────────────── |
| std::cout << "[TAG_FILTER] id1=\"" << kFilterDevice |
| << "\" result_rows=" << rows_per_device << "\n"; |
| |
| auto t0 = clock::now(); |
| int64_t sum_ts_tag_row = tsfile_tag_filter(ts_path, row_count); |
| double sec_ts_tag_row = std::chrono::duration<double>(clock::now() - t0).count(); |
| if (sum_ts_tag_row < 0) return 1; |
| |
| auto t1 = clock::now(); |
| int64_t sum_ts_tag_bat = tsfile_tag_filter_batch(ts_path, row_count); |
| double sec_ts_tag_bat = std::chrono::duration<double>(clock::now() - t1).count(); |
| if (sum_ts_tag_bat < 0) return 1; |
| |
| print_result("TsFile (row)", sec_ts_tag_row, rows_per_device, sum_ts_tag_row); |
| print_result("TsFile (batch)", sec_ts_tag_bat, rows_per_device, sum_ts_tag_bat); |
| if (run_parquet) { |
| auto t2 = clock::now(); |
| int64_t sum_pq_tag = parquet_tag_filter(pq_path); |
| double sec_pq_tag = std::chrono::duration<double>(clock::now() - t2).count(); |
| if (sum_pq_tag < 0) return 1; |
| print_result("Parquet+Arrow", sec_pq_tag, rows_per_device, sum_pq_tag); |
| if (sum_ts_tag_row != sum_pq_tag || sum_ts_tag_bat != sum_pq_tag) |
| std::cerr << " warning: tag filter checksum mismatch\n"; |
| } |
| std::cout << "\n"; |
| |
| // ── Scenario 2: Time Range Filter ───────────────────────────────────────── |
| // Both TsFile and Parquet query the identical half-open interval |
| // [time_range_start, time_range_end). TsFile query() is inclusive on |
| // both ends, so pass (time_range_end - 1) as the upper bound. |
| std::cout << "[TIME_FILTER] time in [" << time_range_start |
| << ", " << time_range_end << ")" |
| << " result_rows=" << time_result_rows << "\n"; |
| |
| auto t3 = clock::now(); |
| int64_t sum_ts_time_row = tsfile_time_filter(ts_path, time_range_start, time_range_end); |
| double sec_ts_time_row = std::chrono::duration<double>(clock::now() - t3).count(); |
| if (sum_ts_time_row < 0) return 1; |
| |
| auto t4 = clock::now(); |
| int64_t sum_ts_time_bat = tsfile_time_filter_batch(ts_path, time_range_start, time_range_end); |
| double sec_ts_time_bat = std::chrono::duration<double>(clock::now() - t4).count(); |
| if (sum_ts_time_bat < 0) return 1; |
| |
| print_result("TsFile (row)", sec_ts_time_row, time_result_rows, sum_ts_time_row); |
| print_result("TsFile (batch)", sec_ts_time_bat, time_result_rows, sum_ts_time_bat); |
| if (run_parquet) { |
| auto t5 = clock::now(); |
| int64_t sum_pq_time = parquet_time_filter(pq_path, time_range_start, time_range_end); |
| double sec_pq_time = std::chrono::duration<double>(clock::now() - t5).count(); |
| if (sum_pq_time < 0) return 1; |
| print_result("Parquet+Arrow", sec_pq_time, time_result_rows, sum_pq_time); |
| if (sum_ts_time_row != sum_pq_time || sum_ts_time_bat != sum_pq_time) |
| std::cerr << " warning: time filter checksum mismatch\n"; |
| } |
| |
| return 0; |
| } |
