Skip to content

Commit 0039ea5

Browse files
liikeuforeverSubsegment
liikeuforever
authored and
Subsegment
committed
refactor: Remove or comment out useless code in the tskv directory.
1 parent ac6ecd4 commit 0039ea5

36 files changed

Lines changed: 74 additions & 975 deletions

‎tskv/src/compaction/check.rs‎

Lines changed: 0 additions & 117 deletions
Original file line numberDiff line numberDiff line change
@@ -1,86 +1,22 @@
11
use std::sync::Arc;
22

3-
use blake3::Hasher;
43
use datafusion::arrow::datatypes::{
54
DataType as ArrowDataType, Field as ArrowField, Schema, SchemaRef,
65
};
76
use datafusion::arrow::record_batch::RecordBatch;
8-
use models::predicate::domain::TimeRange;
97
use models::{utils as model_utils, ColumnId, FieldId, SeriesId, Timestamp};
108
use tokio::sync::RwLock;
119

1210
use crate::error::{TskvError, TskvResult};
1311
use crate::tseries_family::TseriesFamily;
1412
use crate::TseriesFamilyId;
1513

16-
/// Duration of each TimeRangeHashTreeNode, 24 hour.
17-
const DEFAULT_DURATION: i64 = 24 * 60 * 60 * 1_000_000_000;
18-
1914
pub type Hash = [u8; 32];
2015

21-
pub fn hash_to_string(hash: Hash) -> String {
22-
let mut s = String::with_capacity(32);
23-
for v in hash {
24-
s.push_str(format!("{:x}", v).as_str());
25-
}
26-
s
27-
}
28-
2916
#[derive(Default, Debug)]
3017
pub struct VnodeHashTreeNode {
3118
pub vnode_id: TseriesFamilyId,
3219
pub fields: Vec<FieldHashTreeNode>,
33-
min_ts: Timestamp,
34-
max_ts: Timestamp,
35-
}
36-
37-
impl VnodeHashTreeNode {
38-
pub fn with_capacity(vnode_id: TseriesFamilyId, capacity: usize) -> Self {
39-
Self {
40-
vnode_id,
41-
fields: Vec::with_capacity(capacity),
42-
min_ts: Timestamp::MAX,
43-
max_ts: Timestamp::MIN,
44-
}
45-
}
46-
47-
pub fn push(&mut self, value: FieldHashTreeNode) {
48-
self.min_ts = self.min_ts.min(value.min_ts);
49-
self.max_ts = self.max_ts.max(value.max_ts);
50-
self.fields.push(value);
51-
}
52-
53-
/// Calculate checksum with checksums of all field hash trees.
54-
pub fn checksum(&self) -> Hash {
55-
let mut hasher = Hasher::new();
56-
for f in self.fields.iter() {
57-
hasher.update(&f.checksum());
58-
}
59-
hasher.finalize().into()
60-
}
61-
62-
/// Calculate checksum with calculated hash values of all time range hash trees.
63-
///
64-
/// TODO(zipper): This method has double memory cost.
65-
pub fn into_checksum(self) -> Hash {
66-
let mut hashes = Vec::with_capacity(self.len());
67-
for f in self.fields {
68-
for t in f.time_ranges {
69-
hashes.push(t.hash);
70-
}
71-
}
72-
hashes.sort();
73-
74-
let mut hasher = Hasher::new();
75-
for h in hashes {
76-
hasher.update(&h);
77-
}
78-
hasher.finalize().into()
79-
}
80-
81-
pub fn len(&self) -> usize {
82-
self.fields.iter().map(|f| f.len()).sum()
83-
}
8420
}
8521

8622
impl std::fmt::Display for VnodeHashTreeNode {
@@ -101,41 +37,12 @@ impl std::fmt::Display for VnodeHashTreeNode {
10137
pub struct FieldHashTreeNode {
10238
pub field_id: FieldId,
10339
pub time_ranges: Vec<TimeRangeHashTreeNode>,
104-
min_ts: Timestamp,
105-
max_ts: Timestamp,
10640
}
10741

10842
impl FieldHashTreeNode {
109-
pub fn with_capacity(field_id: FieldId, capacity: usize) -> Self {
110-
Self {
111-
field_id,
112-
time_ranges: Vec::with_capacity(capacity),
113-
min_ts: Timestamp::MAX,
114-
max_ts: Timestamp::MIN,
115-
}
116-
}
117-
118-
pub fn push(&mut self, value: TimeRangeHashTreeNode) {
119-
self.min_ts = self.min_ts.min(value.min_ts);
120-
self.max_ts = self.max_ts.max(value.max_ts);
121-
self.time_ranges.push(value);
122-
}
123-
124-
pub fn checksum(&self) -> Hash {
125-
let mut hasher = Hasher::new();
126-
for v in self.time_ranges.iter() {
127-
hasher.update(&v.hash);
128-
}
129-
hasher.finalize().into()
130-
}
131-
13243
pub fn column_series(&self) -> (ColumnId, SeriesId) {
13344
model_utils::split_id(self.field_id)
13445
}
135-
136-
pub fn len(&self) -> usize {
137-
self.time_ranges.len()
138-
}
13946
}
14047

14148
impl std::fmt::Display for FieldHashTreeNode {
@@ -163,20 +70,6 @@ pub struct TimeRangeHashTreeNode {
16370
pub hash: Hash,
16471
}
16572

166-
impl TimeRangeHashTreeNode {
167-
pub fn new(time_range: TimeRange, hash: Hash) -> Self {
168-
Self {
169-
min_ts: time_range.min_ts,
170-
max_ts: time_range.max_ts,
171-
hash,
172-
}
173-
}
174-
175-
pub fn checksum(&self) -> Hash {
176-
self.hash
177-
}
178-
}
179-
18073
impl std::fmt::Display for TimeRangeHashTreeNode {
18174
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
18275
write!(
@@ -198,16 +91,6 @@ pub fn vnode_table_checksum_schema() -> SchemaRef {
19891
]))
19992
}
20093

201-
pub fn vnode_field_time_range_table_checksum_schema() -> SchemaRef {
202-
Arc::new(Schema::new(vec![
203-
ArrowField::new("vnode_id", ArrowDataType::UInt32, false),
204-
ArrowField::new("field_Id", ArrowDataType::UInt32, false),
205-
ArrowField::new("min_ts", ArrowDataType::UInt32, false),
206-
ArrowField::new("max_ts", ArrowDataType::UInt32, false),
207-
ArrowField::new("checksum", ArrowDataType::Utf8, false),
208-
]))
209-
}
210-
21194
/// Get checksum of all data of a vnode, returns RecordBatch with columns of vnode_id and it's checksum, for example:
21295
///
21396
/// | vnode_id | checksum |

‎tskv/src/compaction/compact.rs‎

Lines changed: 2 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -333,10 +333,6 @@ impl CompactingBlockMetaGroup {
333333
pub fn is_empty(&self) -> bool {
334334
self.blk_metas.is_empty()
335335
}
336-
337-
pub fn len(&self) -> usize {
338-
self.blk_metas.len()
339-
}
340336
}
341337

342338
/// Temporary compacting data block.
@@ -497,11 +493,6 @@ impl CompactingFile {
497493
fn series_id(&self) -> Option<SeriesId> {
498494
self.series_ids.get(self.series_idx).copied()
499495
}
500-
501-
fn chunk(&self) -> Option<Arc<Chunk>> {
502-
let sid = self.series_id()?;
503-
self.tsm_reader.chunk().get(&sid).cloned()
504-
}
505496
}
506497

507498
impl Eq for CompactingFile {}
@@ -534,15 +525,9 @@ impl PartialOrd for CompactingFile {
534525
pub(crate) struct CompactIterator {
535526
tsm_readers: Vec<Arc<TsmReader>>,
536527
compacting_files: BinaryHeap<Pin<Box<CompactingFile>>>,
537-
/// Maximum values in generated CompactingBlock
538-
max_data_block_size: usize,
539528
// /// The time range of data to be merged of level-0 data blocks.
540529
// /// The level-0 data that out of the thime range will write back to level-0.
541530
// level_time_range: TimeRange,
542-
/// Decode a data block even though it doesn't need to merge with others,
543-
/// return CompactingBlock::DataBlock rather than CompactingBlock::Raw .
544-
decode_non_overlap_blocks: bool,
545-
546531
tmp_tsm_blk_meta_iters: Vec<(Arc<Chunk>, ColumnGroupID, usize)>,
547532
/// Index to mark `Peekable<BlockMetaIterator>` in witch `TsmReader`,
548533
/// tmp_tsm_blks[i] is in self.tsm_readers[ tmp_tsm_blk_tsm_reader_idx[i] ]
@@ -562,8 +547,6 @@ impl Default for CompactIterator {
562547
Self {
563548
tsm_readers: Default::default(),
564549
compacting_files: Default::default(),
565-
max_data_block_size: 0,
566-
decode_non_overlap_blocks: false,
567550
tmp_tsm_blk_meta_iters: Default::default(),
568551
tmp_tsm_blk_tsm_reader_idx: Default::default(),
569552
finished_readers: Default::default(),
@@ -575,11 +558,7 @@ impl Default for CompactIterator {
575558
}
576559

577560
impl CompactIterator {
578-
pub(crate) fn new(
579-
tsm_readers: Vec<Arc<TsmReader>>,
580-
max_data_block_size: usize,
581-
decode_non_overlap_blocks: bool,
582-
) -> Self {
561+
pub(crate) fn new(tsm_readers: Vec<Arc<TsmReader>>) -> Self {
583562
let compacting_files: BinaryHeap<Pin<Box<CompactingFile>>> = tsm_readers
584563
.iter()
585564
.enumerate()
@@ -590,8 +569,6 @@ impl CompactIterator {
590569
Self {
591570
tsm_readers,
592571
compacting_files,
593-
max_data_block_size,
594-
decode_non_overlap_blocks,
595572
finished_readers: vec![false; compacting_files_cnt],
596573
..Default::default()
597574
}
@@ -757,11 +734,6 @@ impl CompactIterator {
757734
}
758735
}
759736

760-
/// Returns if r1 (min_ts, max_ts) overlaps r2 (min_ts, max_ts)
761-
fn overlaps_tuples(r1: (i64, i64), r2: (i64, i64)) -> bool {
762-
r1.0 <= r2.1 && r1.1 >= r2.0
763-
}
764-
765737
pub async fn run_compaction_job(
766738
request: CompactReq,
767739
kernel: Arc<GlobalContext>,
@@ -799,7 +771,7 @@ pub async fn run_compaction_job(
799771
}
800772

801773
let max_block_size = TseriesFamily::MAX_DATA_BLOCK_SIZE as usize;
802-
let mut iter = CompactIterator::new(tsm_readers, max_block_size, false);
774+
let mut iter = CompactIterator::new(tsm_readers);
803775
let tsm_dir = request.storage_opt.tsm_dir(&request.database, tsf_id);
804776
let max_file_size = request.storage_opt.level_max_file_size(request.out_level);
805777
let mut tsm_writer =
@@ -1044,30 +1016,6 @@ pub mod test {
10441016
col
10451017
}
10461018

1047-
fn u64_some_column(data: Vec<Option<u64>>, col: TableColumn) -> MutableColumn {
1048-
let mut col = MutableColumn::empty_with_cap(col, data.len()).unwrap();
1049-
for datum in data {
1050-
col.push(datum.map(FieldVal::Unsigned)).unwrap()
1051-
}
1052-
col
1053-
}
1054-
1055-
fn f64_some_column(data: Vec<Option<f64>>, col: TableColumn) -> MutableColumn {
1056-
let mut col = MutableColumn::empty_with_cap(col, data.len()).unwrap();
1057-
for datum in data {
1058-
col.push(datum.map(FieldVal::Float)).unwrap()
1059-
}
1060-
col
1061-
}
1062-
1063-
fn bool_some_column(data: Vec<Option<bool>>, col: TableColumn) -> MutableColumn {
1064-
let mut col = MutableColumn::empty_with_cap(col, data.len()).unwrap();
1065-
for datum in data {
1066-
col.push(datum.map(FieldVal::Boolean)).unwrap()
1067-
}
1068-
col
1069-
}
1070-
10711019
fn get_result_file_path(dir: impl AsRef<Path>, version_edit: VersionEdit) -> PathBuf {
10721020
if !version_edit.add_files.is_empty() {
10731021
let file_id = version_edit.add_files.first().unwrap().file_id;

‎tskv/src/compaction/flush.rs‎

Lines changed: 1 addition & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,6 @@ use crate::{ColumnFileId, TsKvContext, TseriesFamilyId, TskvError};
2222
pub struct FlushTask {
2323
ts_family_id: TseriesFamilyId,
2424
mem_caches: Vec<Arc<RwLock<MemCache>>>,
25-
low_seq_no: u64,
26-
high_seq_no: u64,
2725
global_context: Arc<GlobalContext>,
2826
path_tsm: PathBuf,
2927
path_delta: PathBuf,
@@ -33,17 +31,13 @@ impl FlushTask {
3331
pub fn new(
3432
ts_family_id: TseriesFamilyId,
3533
mem_caches: Vec<Arc<RwLock<MemCache>>>,
36-
low_seq_no: u64,
37-
high_seq_no: u64,
3834
global_context: Arc<GlobalContext>,
3935
path_tsm: impl AsRef<Path>,
4036
path_delta: impl AsRef<Path>,
4137
) -> Self {
4238
Self {
4339
ts_family_id,
4440
mem_caches,
45-
low_seq_no,
46-
high_seq_no,
4741
global_context,
4842
path_tsm: path_tsm.as_ref().into(),
4943
path_delta: path_delta.as_ref().into(),
@@ -166,8 +160,6 @@ pub async fn run_flush_memtable_job(
166160
let flush_task = FlushTask::new(
167161
req.ts_family_id,
168162
req.mems.clone(),
169-
req.low_seq_no,
170-
req.high_seq_no,
171163
ctx.global_ctx.clone(),
172164
path_tsm,
173165
path_delta,
@@ -231,7 +223,7 @@ pub mod flush_tests {
231223
use models::field_value::FieldVal;
232224
use models::predicate::domain::TimeRange;
233225
use models::schema::{ColumnType, TableColumn, TskvTableSchema};
234-
use models::{ColumnId, SeriesKey, ValueType};
226+
use models::{SeriesKey, ValueType};
235227
use parking_lot::lock_api::RwLock;
236228
use utils::dedup_front_by_key;
237229

@@ -244,20 +236,6 @@ pub mod flush_tests {
244236
use crate::tsm::writer::TsmWriter;
245237
use crate::{Options, VersionEdit};
246238

247-
fn i64_column(data: Vec<i64>) -> MutableColumn {
248-
let mut col = MutableColumn::empty(TableColumn::new(
249-
1,
250-
"f1".to_string(),
251-
ColumnType::Field(ValueType::Integer),
252-
Encoding::default(),
253-
))
254-
.unwrap();
255-
for datum in data {
256-
col.push(Some(FieldVal::Integer(datum))).unwrap()
257-
}
258-
col
259-
}
260-
261239
fn f64_column(data: Vec<f64>) -> MutableColumn {
262240
let mut col = MutableColumn::empty(TableColumn::new(
263241
4,
@@ -286,25 +264,6 @@ pub mod flush_tests {
286264
col
287265
}
288266

289-
pub fn default_table_schema(ids: Vec<ColumnId>) -> TskvTableSchema {
290-
let fields = ids
291-
.iter()
292-
.map(|i| TableColumn {
293-
id: *i,
294-
name: i.to_string(),
295-
column_type: ColumnType::Field(ValueType::Unknown),
296-
encoding: Encoding::Default,
297-
})
298-
.collect();
299-
300-
TskvTableSchema::new(
301-
"cnosdb".to_string(),
302-
"public".to_string(),
303-
"".to_string(),
304-
fields,
305-
)
306-
}
307-
308267
#[test]
309268
fn test_sort_dedup() {
310269
{
@@ -452,8 +411,6 @@ pub mod flush_tests {
452411
let flush_task = FlushTask::new(
453412
1,
454413
mem_caches,
455-
0,
456-
1,
457414
Arc::new(GlobalContext::new()),
458415
path_tsm.clone(),
459416
path_delta.clone(),

0 commit comments

Comments
 (0)