You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: count.rs
    path of file: ./repos/cnosdb/tskv/src/compute
    
    the code of the file until where you have to start completion: // use std::collections::HashSet;
// use std::sync::Arc;
//
// use models::predicate::domain::{TimeRange, TimeRanges};
// use models::{utils as model_utils, ColumnId, FieldId, SeriesId, Timestamp};
// use tokio::runtime::Runtime;
// use tokio::sync::Semaphore;
// use trace::trace;
//
// use crate::tseries_family::{ColumnFile, SuperVersion};
// use crate::tsm::{self, BlockMeta, TsmReader};
// use crate::{Error, Result};
//
// #[derive(PartialEq, Eq)]
// pub enum TimeRangeCmp {
//     /// A has no intersection with B
//     Exclude,
//     /// A includes B
//     Include,
//     /// A overlaps with B
//     Intersect,
// }
//
// enum CountingObject {
//     Field(FieldId),
//     Series(SeriesId),
// }
//
// /// Compute pushed down aggregate:
// ///
// /// `SELECT count(<column>) FROM <table> WHERE <time_range_predicates>`
// pub async fn count_column_non_null_values(
//     runtime: Arc<Runtime>,
//     super_version: Arc<SuperVersion>,
//     series_ids: Arc<Vec<SeriesId>>,
//     column_id: Option<ColumnId>,
//     time_ranges: Arc<TimeRanges>,
// ) -> Result<u64> {
//     let column_files = Arc::new(super_version.column_files(&time_ranges));
//     let mut jh_vec = Vec::with_capacity(series_ids.len());
//
//     // Limit the number of concurrent tasks.
//     let max_count_tasks = num_cpus::get().min(series_ids.len());
//     let count_tasks_limit = Arc::new(Semaphore::new(max_count_tasks));
//     for series_id in series_ids.iter() {
//         let count_object = if let Some(column_id) = 
    
//     // fn test_super_version_count_memcache() {
//     //     let dir = "/tmp/test/ts_family/super_version_count_memcache";
//     //     let mut global_config = config::get_config_for_test();
//     //     global_config.storage.path = dir.to_string();
//     //
//     //     let pool: MemoryPoolRef = Arc::new(GreedyMemoryPool::default());
//     //     #[rustfmt::skip]
//     //     let cache_group = {
//     //         let caches = vec![MemCache::new(1, 16, 2, 0, &pool), MemCache::new(1, 16, 2, 0, &pool), MemCache::new(1, 16, 2, 0, &pool)];
//     //         // cache, sid, schema_id, schema, time_range, put_none
//     //         put_rows_to_cache(&caches[0], 1, 1, default_table_schema(vec![1]), (1, 4), false);
//     //         put_rows_to_cache(&caches[0], 2, 1, default_table_schema(vec![1]), (101, 104), false);
//     //         put_rows_to_cache(&caches[1], 1, 1, default_table_schema(vec![1]), (4, 6), false);
//     //         put_rows_to_cache(&caches[1], 2, 1, default_table_schema(vec![1]), (104, 106), false);
//     //         put_rows_to_cache(&caches[2], 1, 1, default_table_schema(vec![1]), (7, 9), false);
//     //         put_rows_to_cache(&caches[2], 2, 1, default_table_schema(vec![1]), (107, 109), false);
//     //         put_rows_to_cache(&caches[2], 1, 1, default_table_schema(vec![1]), (11, 15), false);
//     //         put_rows_to_cache(&caches[2], 2, 1, default_table_schema(vec![1]), (111, 115), false);
//     //         let cache = MemCache::new(1, 16, 2, 0, &pool);
//     //         put_rows_to_cache(&cache, 1, 1, default_table_schema(vec![1]), (11, 15), false);
//     //         put_rows_to_cache(&cache, 2, 1, default_table_schema(vec![1]), (111, 115), false);
//     //         CacheGroup {
//     //             mut_cache: Arc::new(RwLock::new(cache)),
//     //             immut_cache: caches.into_iter().map(|c| Arc::new(RwLock::new(c))).collect(),
//     //         }
//     //     };
//     //
//     //     let opt = Arc::new(Options::from(&global_config));
//     //     let database = Arc::new("dba".to_string());
//     //     let ts_family_id = 1;
//     //
//     //     let version =
//     //         build_version_by_column_files(opt.storage.clone(), database, ts_family_id, vec![]);
//     //     let test_helper = TestHelper {
//     //         runtime: Arc::new(
//     //             tokio::runtime::Builder::new_multi_thread()
//     //                 .enable_all()
//     //                 .build()
//     //                 .unwrap(),
//     //         ),
//     //         super_version: Arc::new(SuperVersion::new(
//     //             ts_family_id,
//     //             opt.storage.clone(),
//     //             cache_group,
//     //             Arc::new(version),
//     //             1,
//     //         )),
//     //     };
//     //
//     //     #[rustfmt::skip]
//     //     let _skip_fmt = {
//     //         assert_eq!(test_helper.run(&[1], None, (i64::MIN, i64::MAX)).unwrap(), 14);
//     //         assert_eq!(test_helper.run(&[1], None, (i64::MIN, i64::MAX)).unwrap(), 14);
//     //
//     //         assert_eq!(test_helper.run(&[1], Some(1), (i64::MIN, i64::MAX)).unwrap(), 14);
//     //         assert_eq!(test_helper.run(&[1, 2], Some(1), (i64::MIN, i64::MAX)).unwrap(), 28);
//     //         assert_eq!(test_helper.run(&[1], Some(1), (0, 0)).unwrap(), 0);
//     //         assert_eq!(test_helper.run(&[1], Some(1), (1, 1)).unwrap(), 1);
//     //         assert_eq!(test_helper.run(&[1], Some(1), (1, 4)).unwrap(), 4);
//     //         assert_eq!(test_helper.run(&[1], Some(1), (1, 5)).unwrap(), 5);
//     //         assert_eq!(test_helper.run(&[1], Some(1), (4, 4)).unwrap(), 1);
//     //         assert_eq!(test_helper.run(&[1], Some(1), (4, 5)).unwrap(), 2);
//     //         assert_eq!(test_helper.run(&[1], Some(1), (4, 7)).unwrap(), 4);
//     //         assert_eq!(test_helper.run(&[1], Some(1), (8, 10)).unwrap(), 2);
//     //         assert_eq!(test_helper.run(&[1], Some(1), (10, 10)).unwrap(), 0);
//     //
//     //         assert_eq!(test_helper.run(&[2], Some(1), (101, 104)).unwrap(), 4);
//     //         assert_eq!(test_helper.run(&[2], Some(1), (104, 105)).unwrap(), 2);
//     //         "skip_fmt"
//     //     };
//     // }