pub fn close(mut self) -> Result<(), Error> {
        info!("flushing {} partition segment {}", self.partition, self.id);
        if self.batch.len() > 0 {
            &self.flush()?;
        }
        let mut partition_registry: PartitionRegistry;
        let (start_ts, end_ts) = (self.start_ts, self.end_ts);
        let segment_file = SegmentFile {
            id: self.id,
            start_ts: start_ts,
            end_ts: end_ts,
        };
        if segment_file.id == 1 {
            // This is new segment so, create a new partition registry and save it.
            let segment_files = vec![segment_file];
            partition_registry = PartitionRegistry {
                last_assigned: 1,
                segment_files: segment_files,
            };
        } else {
            let registry_buf = self
                .store
                .get(format!("{}_{}", PARTITION_PREFIX, self.partition).as_bytes())?
                .unwrap();
            let mut registry_buf = Deserializer::new(&registry_buf[..]);
            partition_registry = Deserialize::deserialize(&mut registry_buf)?;
            partition_registry.segment_files.push(segment_file);
            partition_registry.last_assigned = self.id;
        }
        let mut buf = Vec::with_capacity(1024);
        partition_registry
            .serialize(&mut Serializer::new(&mut buf))
            .unwrap();
        self.store.set(
            format!("{}_{}", PARTITION_PREFIX, self.partition).as_bytes(),
            buf,
        );
        // insert posting list.
        let mut indices = Vec::new();
        let mut wb = Batch::new();
        let posting_list = self.index_posting_list.drain();
        for (index, list) in posting_list {
            wb.set(
                format!(
                    "{}_{}_{}_{}",
                    SEGMENT_PREFIX, self.partition, self.id, &index
                )
                .into_bytes(),
                list,
            )
            .unwrap();
            indices.push(index);
        }
        // Insert json key posting list.
        let posting_list = self.json_key_posting_list.drain();
        for (key, list) in posting_list {
            wb.set(
                format!(
                    "{}_{}_{}_{}",
                    SEGEMENT_JSON_KEY_PREFIX, self.partition, self.id, &key
                )
                .into_bytes(),
                list,
            )
            .unwrap();
        }
        wb.set(
            format!(
                "{}_{}_{}_{}",
                SEGMENT_PREFIX, self.partition, self.id, POSTING_LIST_ALL,
            )
            .into_bytes(),
            self.entry_offsets,
        )
        .unwrap();
        self.store.flush_batch(wb)?;
        indices.sort();
        let indices = indices.drain(..);
        let mut index_writer = SetBuilder::memory();
        for index in indices {
            index_writer.insert(index).unwrap();
        }
        let mut index_file = File::create(
            Path::new(&self.config.dir)
                .join("partition")
                .join(self.partition.to_string())
                .join(format!("segment_index_{}.fst", self.id)),
        )?;
        let buf = index_writer.into_inner().unwrap();
        index_file.write(&buf)?;
        index_file.flush()?;
        let info = self.segment_file.metadata().unwrap();
        assert_eq!(info.len(), self.file_offset);
        Ok(())
    }
/// close will flush all the remaining batched line.
close will flush all the remaining batched line.
close will flush all the remaining batched line .
pub fn close(mut self) -> Result<(), Error> {
        info!("flushing {} partition segment {}", self.partition, self.id);
        if self.batch.len() > 0 {
            &self.flush()?;
        }
        let mut partition_registry: PartitionRegistry;
        let (start_ts, end_ts) = (self.start_ts, self.end_ts);
        let segment_file = SegmentFile {
            id: self.id,
            start_ts: start_ts,
            end_ts: end_ts,
        };
        if segment_file.id == 1 {
            // This is new segment so, create a new partition registry and save it.
            let segment_files = vec![segment_file];
            partition_registry = PartitionRegistry {
                last_assigned: 1,
                segment_files: segment_files,
            };
        } else {
            let registry_buf = self
                .store
                .get(format!("{}_{}", PARTITION_PREFIX, self.partition).as_bytes())?
                .unwrap();
            let mut registry_buf = Deserializer::new(&registry_buf[..]);
            partition_registry = Deserialize::deserialize(&mut registry_buf)?;
            partition_registry.segment_files.push(segment_file);
            partition_registry.last_assigned = self.id;
        }
        let mut buf = Vec::with_capacity(1024);
        partition_registry
            .serialize(&mut Serializer::new(&mut buf))
            .unwrap();
        self.store.set(
            format!("{}_{}", PARTITION_PREFIX, self.partition).as_bytes(),
            buf,
        );
        // insert posting list.
        let mut indices = Vec::new();
        let mut wb = Batch::new();
        let posting_list = self.index_posting_list.drain();
        for (index, list) in posting_list {
            wb.set(
                format!(
                    "{}_{}_{}_{}",
                    SEGMENT_PREFIX, self.partition, self.id, &index
                )
                .into_bytes(),
                list,
            )
            .unwrap();
            indices.push(index);
        }
        // Insert json key posting list.
        let posting_list = self.json_key_posting_list.drain();
        for (key, list) in posting_list {
            wb.set(
                format!(
                    "{}_{}_{}_{}",
                    SEGEMENT_JSON_KEY_PREFIX, self.partition, self.id, &key
                )
                .into_bytes(),
                list,
            )
            .unwrap();
        }
        wb.set(
            format!(
                "{}_{}_{}_{}",
                SEGMENT_PREFIX, self.partition, self.id, POSTING_LIST_ALL,
            )
            .into_bytes(),
            self.entry_offsets,
        )
        .unwrap();
        self.store.flush_batch(wb)?;
        indices.sort();
        let indices = indices.drain(..);
        let mut index_writer = SetBuilder::memory();
        for index in indices {
            index_writer.insert(index).unwrap();
        }
        let mut index_file = File::create(
            Path::new(&self.config.dir)
                .join("partition")
                .join(self.partition.to_string())
                .join(format!("segment_index_{}.fst", self.id)),
        )?;
        let buf = index_writer.into_inner().unwrap();
        index_file.write(&buf)?;
        index_file.flush()?;
        let info = self.segment_file.metadata().unwrap();
        assert_eq!(info.len(), self.file_offset);
        Ok(())
    }
pub fn close ( mut self ) -> Result < ( ) , Error > { info ! ( " " self partition self id ) ; if self . batch . len ( ) > 0 { & self . flush ( ) ? ; } let mut partition_registry : PartitionRegistry ; let ( start_ts , end_ts ) = ( self . start_ts , self . end_ts ) ; let segment_file = SegmentFile { id : self . id , start_ts : start_ts , end_ts : end_ts , } ; if segment_file . id == 1 { // This is new segment so, create a new partition registry and save it. let segment_files = vec ! [ segment_file ] ; partition_registry = PartitionRegistry { last_assigned : 1 , segment_files : segment_files , } ; } else { let registry_buf = self . store . get ( format ! ( " " PARTITION_PREFIX self partition ) . as_bytes ( ) ) ? . unwrap ( ) ; let mut registry_buf = Deserializer :: new ( & registry_buf [ .. ] ) ; partition_registry = Deserialize :: deserialize ( & mut registry_buf ) ? ; partition_registry . segment_files . push ( segment_file ) ; partition_registry . last_assigned = self . id ; } let mut buf = Vec :: with_capacity ( 1024 ) ; partition_registry . serialize ( & mut Serializer :: new ( & mut buf ) ) . unwrap ( ) ; self . store . set ( format ! ( " " PARTITION_PREFIX self partition ) . as_bytes ( ) , buf , ) ; // insert posting list. let mut indices = Vec :: new ( ) ; let mut wb = Batch :: new ( ) ; let posting_list = self . index_posting_list . drain ( ) ; for ( index , list ) in posting_list { wb . set ( format ! ( " " SEGMENT_PREFIX self partition self id index ) . into_bytes ( ) , list , ) . unwrap ( ) ; indices . push ( index ) ; } // Insert json key posting list. let posting_list = self . json_key_posting_list . drain ( ) ; for ( key , list ) in posting_list { wb . set ( format ! ( " " SEGEMENT_JSON_KEY_PREFIX self partition self id key ) . into_bytes ( ) , list , ) . unwrap ( ) ; } wb . set ( format ! ( " " SEGMENT_PREFIX self partition self id POSTING_LIST_ALL ) . into_bytes ( ) , self . entry_offsets , ) . unwrap ( ) ; self . store . flush_batch ( wb ) ? ; indices . sort ( ) ; let indices = indices . drain ( .. ) ; let mut index_writer = SetBuilder :: memory ( ) ; for index in indices { index_writer . insert ( index ) . unwrap ( ) ; } let mut index_file = File :: create ( Path :: new ( & self . config . dir ) . join ( " " ) . join ( self . partition . to_string ( ) ) . join ( format ! ( " " self id ) ) , ) ? ; let buf = index_writer . into_inner ( ) . unwrap ( ) ; index_file . write ( & buf ) ? ; index_file . flush ( ) ? ; let info = self . segment_file . metadata ( ) . unwrap ( ) ; assert_eq ! ( info len ( ) self file_offset ) ; Ok ( ( ) ) }
close will flush all the remaining batched line.
close will flush all the remaining batched line .
mut self