You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@horaedb.apache.org by "zealchen (via GitHub)" <gi...@apache.org> on 2024/12/06 21:35:11 UTC
[PR] feat: another format for manifest [horaedb]
zealchen opened a new pull request, #1604:
URL: https://github.com/apache/horaedb/pull/1604
## Rationale
#1600
## Detailed Changes
Creating a new data format to represent the manifest snapshot file.
```text
| magic(u32) | version(u8) | flags(u8) | length(u64) | Record(N) ... |
The Magic field (u32) is used to ensure the validity of the data source.
The Flags field (u8) is reserved for future extensibility, such as enabling compression or supporting additional features.
The length field (u64) represents the total length of the subsequent records and serves as a straightforward method for verifying their integrity. (length = record_length * record_count)
# Record is a self-descriptive message
| id(u64) | time_range(i64*2)| size(u32) | num_rows(u32)|
```
In do_merge, the snapshot data handle is like:
```text
Old data flow in do_merge:
delta_sstmetas
| (extend vec)
V
object_store -> org_bytes -> org_pb -> Vec<sstmeta> -> dst_pb -> dst_bytes -> object_store
New data flow in do_merge:
delta_sstmetas -> bytes
| (append)
V
object_store -> org_bytes -> dst_bytes -> object_store
````
Specifically, I create the SnapshotHeader and SnapshotRecordV1 to represent the corresponding data in snapshot bytes. Before merge delta sstfiles into new bytes, we allocate a larger Vec `<u8>` and copy each segments (header, old records, new records) into it.
This RP DOES NOT address format upgrade logic which can be resolved in another PR. As for the upgrade, we could define new SnapshotRecord format and do data migration before Manifest::try_new.
## Test Plan
UT
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@horaedb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@horaedb.apache.org
For additional commands, e-mail: commits-help@horaedb.apache.org
Re: [PR] feat: another format for manifest [horaedb]
Posted by "zealchen (via GitHub)" <gi...@apache.org>.
zealchen commented on code in PR #1604:
URL: https://github.com/apache/horaedb/pull/1604#discussion_r1881357718
##########
horaedb/metric_engine/src/manifest.rs:
##########
@@ -184,6 +203,264 @@ impl Manifest {
}
}
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotHeader {
+ pub magic: u32,
+ pub version: u8,
+ pub flag: u8,
+ pub length: u64,
+}
+
+impl SnapshotHeader {
+ // format: | magic(u32) | version(u8) | flags(u8) | length(u64) |
+ // The Magic field (u32) is used to ensure the validity of the data source.
+ // The Flags field (u8) is reserved for future extensibility, such as enabling
+ // compression or supporting additional features.
+ // The length field (u64) represents the total length of the subsequent records
+ // and serves as a straightforward method for verifying their integrity.
+ // (length = record_length
+ // * record_count)
+
+ // use #[repr(packed)] to force unalignment and avoid hard code internal types
+ // here, better solutions?
+ pub const LENGTH: usize = size_of::<SnapshotHeader>();
+ pub const MAGIC: u32 = 0x32A489BF;
+
+ pub fn new(length: u64) -> Self {
+ Self {
+ magic: SnapshotHeader::MAGIC,
+ version: SnapshotRecordV1::VERSION,
+ flag: 0,
+ length,
+ }
+ }
+
+ pub fn try_from_bytes(bytes: &[u8]) -> Result<Self> {
+ if bytes.len() < Self::LENGTH {
+ return Err(anyhow!("invalid bytes, length: {}", bytes.len()).into());
+ } else {
+ let magic = u32::from_le_bytes(bytes[0..4].try_into().unwrap());
Review Comment:
Adding an ensure! to check the magic number.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@horaedb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@horaedb.apache.org
For additional commands, e-mail: commits-help@horaedb.apache.org
Re: [PR] feat: another format for manifest [horaedb]
Posted by "jiacai2050 (via GitHub)" <gi...@apache.org>.
jiacai2050 commented on code in PR #1604:
URL: https://github.com/apache/horaedb/pull/1604#discussion_r1877716319
##########
horaedb/metric_engine/src/manifest.rs:
##########
@@ -184,6 +203,264 @@ impl Manifest {
}
}
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotHeader {
+ pub magic: u32,
+ pub version: u8,
+ pub flag: u8,
+ pub length: u64,
+}
+
+impl SnapshotHeader {
+ // format: | magic(u32) | version(u8) | flags(u8) | length(u64) |
+ // The Magic field (u32) is used to ensure the validity of the data source.
+ // The Flags field (u8) is reserved for future extensibility, such as enabling
+ // compression or supporting additional features.
+ // The length field (u64) represents the total length of the subsequent records
+ // and serves as a straightforward method for verifying their integrity.
+ // (length = record_length
+ // * record_count)
+
+ // use #[repr(packed)] to force unalignment and avoid hard code internal types
+ // here, better solutions?
+ pub const LENGTH: usize = size_of::<SnapshotHeader>();
+ pub const MAGIC: u32 = 0x32A489BF;
+
+ pub fn new(length: u64) -> Self {
+ Self {
+ magic: SnapshotHeader::MAGIC,
+ version: SnapshotRecordV1::VERSION,
+ flag: 0,
+ length,
+ }
+ }
+
+ pub fn try_from_bytes(bytes: &[u8]) -> Result<Self> {
+ if bytes.len() < Self::LENGTH {
Review Comment:
You can use `ensure!` to do this kinds of sanity check.
##########
horaedb/metric_engine/src/manifest.rs:
##########
@@ -184,6 +203,264 @@ impl Manifest {
}
}
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotHeader {
+ pub magic: u32,
+ pub version: u8,
+ pub flag: u8,
+ pub length: u64,
+}
+
+impl SnapshotHeader {
+ // format: | magic(u32) | version(u8) | flags(u8) | length(u64) |
+ // The Magic field (u32) is used to ensure the validity of the data source.
+ // The Flags field (u8) is reserved for future extensibility, such as enabling
+ // compression or supporting additional features.
+ // The length field (u64) represents the total length of the subsequent records
+ // and serves as a straightforward method for verifying their integrity.
+ // (length = record_length
+ // * record_count)
+
+ // use #[repr(packed)] to force unalignment and avoid hard code internal types
+ // here, better solutions?
+ pub const LENGTH: usize = size_of::<SnapshotHeader>();
+ pub const MAGIC: u32 = 0x32A489BF;
+
+ pub fn new(length: u64) -> Self {
+ Self {
+ magic: SnapshotHeader::MAGIC,
+ version: SnapshotRecordV1::VERSION,
+ flag: 0,
+ length,
+ }
+ }
+
+ pub fn try_from_bytes(bytes: &[u8]) -> Result<Self> {
+ if bytes.len() < Self::LENGTH {
+ return Err(anyhow!("invalid bytes, length: {}", bytes.len()).into());
+ } else {
+ let magic = u32::from_le_bytes(bytes[0..4].try_into().unwrap());
+ let version = bytes[5];
+ let flag = bytes[6];
+ let length = u64::from_le_bytes(bytes[6..14].try_into().unwrap());
+ Ok(Self {
+ magic,
+ version,
+ flag,
+ length,
+ })
+ }
+ }
+
+ pub fn write_to(&self, writter: &mut &mut [u8]) -> Result<()> {
+ if writter.len() < SnapshotHeader::LENGTH {
Review Comment:
`ensure!`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@horaedb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@horaedb.apache.org
For additional commands, e-mail: commits-help@horaedb.apache.org
Re: [PR] feat: another format for manifest [horaedb]
Posted by "jiacai2050 (via GitHub)" <gi...@apache.org>.
jiacai2050 commented on code in PR #1604:
URL: https://github.com/apache/horaedb/pull/1604#discussion_r1877756109
##########
horaedb/metric_engine/src/manifest.rs:
##########
@@ -184,6 +203,264 @@ impl Manifest {
}
}
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotHeader {
+ pub magic: u32,
+ pub version: u8,
+ pub flag: u8,
+ pub length: u64,
+}
+
+impl SnapshotHeader {
+ // format: | magic(u32) | version(u8) | flags(u8) | length(u64) |
Review Comment:
Could you write this as box comments like here:
https://github.com/apache/horaedb/blob/6cdf2e66a04f11c60c04cacae0111a14d344a501/src/components/codec/src/columnar/bytes.rs#L34-L36
##########
horaedb/metric_engine/src/manifest.rs:
##########
@@ -184,6 +203,264 @@ impl Manifest {
}
}
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotHeader {
+ pub magic: u32,
+ pub version: u8,
+ pub flag: u8,
+ pub length: u64,
+}
+
+impl SnapshotHeader {
+ // format: | magic(u32) | version(u8) | flags(u8) | length(u64) |
+ // The Magic field (u32) is used to ensure the validity of the data source.
+ // The Flags field (u8) is reserved for future extensibility, such as enabling
+ // compression or supporting additional features.
+ // The length field (u64) represents the total length of the subsequent records
+ // and serves as a straightforward method for verifying their integrity.
+ // (length = record_length
+ // * record_count)
+
+ // use #[repr(packed)] to force unalignment and avoid hard code internal types
+ // here, better solutions?
+ pub const LENGTH: usize = size_of::<SnapshotHeader>();
+ pub const MAGIC: u32 = 0x32A489BF;
+
+ pub fn new(length: u64) -> Self {
+ Self {
+ magic: SnapshotHeader::MAGIC,
+ version: SnapshotRecordV1::VERSION,
+ flag: 0,
+ length,
+ }
+ }
+
+ pub fn try_from_bytes(bytes: &[u8]) -> Result<Self> {
+ if bytes.len() < Self::LENGTH {
+ return Err(anyhow!("invalid bytes, length: {}", bytes.len()).into());
+ } else {
+ let magic = u32::from_le_bytes(bytes[0..4].try_into().unwrap());
+ let version = bytes[5];
+ let flag = bytes[6];
+ let length = u64::from_le_bytes(bytes[6..14].try_into().unwrap());
+ Ok(Self {
+ magic,
+ version,
+ flag,
+ length,
+ })
+ }
+ }
+
+ pub fn write_to(&self, writter: &mut &mut [u8]) -> Result<()> {
+ if writter.len() < SnapshotHeader::LENGTH {
+ return Err(anyhow!(
+ "writter buf is too small for writing the header, length: {}",
+ writter.len()
+ )
+ .into());
+ }
+ writter.write(self.magic.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.version.to_le_bytes().as_slice())
+ .unwrap();
+ writter.write(self.flag.to_le_bytes().as_slice()).unwrap();
+ writter.write(self.length.to_le_bytes().as_slice()).unwrap();
+ Ok(())
+ }
+}
+
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotRecordV1 {
+ id: u64,
+ time_range_start: i64,
+ time_range_end: i64,
+ size: u32,
+ num_rows: u32,
+}
+
+impl SnapshotRecordV1 {
+ // format: | id(u64) | time_range(i64*2)| size(u32) | num_rows(u32)|
+ const LENGTH: usize = size_of::<SnapshotRecordV1>();
+ pub const VERSION: u8 = 1;
+
+ pub fn write_to(&self, writter: &mut &mut [u8]) -> Result<()> {
+ if writter.len() < SnapshotRecordV1::LENGTH {
+ return Err(anyhow!(
+ "writter buf is too small for writing the record, length: {}",
+ writter.len()
+ )
+ .into());
+ }
+ writter.write(self.id.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.time_range_start.to_le_bytes().as_slice())
+ .unwrap();
+ writter
+ .write(self.time_range_end.to_le_bytes().as_slice())
+ .unwrap();
+ writter.write(self.size.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.num_rows.to_le_bytes().as_slice())
+ .unwrap();
+ Ok(())
+ }
+
+ pub fn id(&self) -> u64 {
+ self.id
+ }
+}
+
+impl From<SstFile> for SnapshotRecordV1 {
+ fn from(value: SstFile) -> Self {
+ SnapshotRecordV1 {
+ id: value.id(),
+ time_range_start: value.meta().time_range.start.0,
+ time_range_end: value.meta().time_range.end.0,
+ size: value.meta().size,
+ num_rows: value.meta().num_rows,
+ }
+ }
+}
+
+impl TryFrom<&[u8]> for SnapshotRecordV1 {
+ type Error = Error;
+
+ fn try_from(value: &[u8]) -> Result<Self> {
+ if value.len() < SnapshotRecordV1::LENGTH {
+ return Err(anyhow!("invalid value len: {}", value.len()).into());
+ }
+ let id = u64::from_le_bytes(value[0..8].try_into().unwrap());
+ let time_range_start = i64::from_le_bytes(value[8..16].try_into().unwrap());
+ let time_range_end = i64::from_le_bytes(value[16..24].try_into().unwrap());
+ let size = u32::from_le_bytes(value[24..28].try_into().unwrap());
+ let num_rows = u32::from_le_bytes(value[28..32].try_into().unwrap());
+ Ok(SnapshotRecordV1 {
+ id,
+ time_range_start,
+ time_range_end,
+ size,
+ num_rows,
+ })
+ }
+}
+
+impl Into<SstFile> for SnapshotRecordV1 {
+ fn into(self) -> SstFile {
+ let file_meta = FileMeta {
+ max_sequence: self.id(),
+ num_rows: self.num_rows,
+ size: self.size,
+ time_range: TimeRange::new(
+ Timestamp::from(self.time_range_start),
+ Timestamp::from(self.time_range_end),
+ ),
+ };
+ SstFile::new(self.id(), file_meta)
+ }
+}
+
+struct Snapshot {
+ header: SnapshotHeader,
+ inner: Bytes,
+}
+
+impl Default for Snapshot {
+ // create an empty Snapshot
+ fn default() -> Self {
+ let header = SnapshotHeader::new(0);
+ Self {
+ header,
+ inner: Bytes::new(),
+ }
+ }
+}
+
+impl Snapshot {
+ pub fn try_from_bytes(bytes: Bytes) -> Result<Self> {
+ if bytes.is_empty() {
+ return Ok(Snapshot::default());
+ }
+ let header = SnapshotHeader::try_from_bytes(bytes.as_bytes())?;
+ let header_length = header.length as usize;
+ if header_length > 0
+ && (header_length % SnapshotRecordV1::LENGTH != 0
+ || header_length + SnapshotHeader::LENGTH != bytes.len())
+ {
+ return Err(anyhow!(
+ "create snapshot from bytes failed, invalid bytes, header length = {}, total length: {}",
+ header_length,
+ bytes.len()
+ ).into());
+ }
+ Ok(Self {
+ header,
+ inner: bytes,
+ })
+ }
+
+ pub fn to_sstfiles(&self) -> Result<Vec<SstFile>> {
+ if self.header.length == 0 {
+ return Ok(Vec::new());
+ } else {
+ let buf = self.inner.as_bytes();
+ let mut result: Vec<SstFile> = Vec::new();
Review Comment:
We can set initial capacity from header.
##########
horaedb/metric_engine/src/manifest.rs:
##########
@@ -184,6 +203,264 @@ impl Manifest {
}
}
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotHeader {
+ pub magic: u32,
+ pub version: u8,
+ pub flag: u8,
+ pub length: u64,
+}
+
+impl SnapshotHeader {
+ // format: | magic(u32) | version(u8) | flags(u8) | length(u64) |
+ // The Magic field (u32) is used to ensure the validity of the data source.
+ // The Flags field (u8) is reserved for future extensibility, such as enabling
+ // compression or supporting additional features.
+ // The length field (u64) represents the total length of the subsequent records
+ // and serves as a straightforward method for verifying their integrity.
+ // (length = record_length
+ // * record_count)
+
+ // use #[repr(packed)] to force unalignment and avoid hard code internal types
+ // here, better solutions?
+ pub const LENGTH: usize = size_of::<SnapshotHeader>();
+ pub const MAGIC: u32 = 0x32A489BF;
+
+ pub fn new(length: u64) -> Self {
+ Self {
+ magic: SnapshotHeader::MAGIC,
+ version: SnapshotRecordV1::VERSION,
+ flag: 0,
+ length,
+ }
+ }
+
+ pub fn try_from_bytes(bytes: &[u8]) -> Result<Self> {
+ if bytes.len() < Self::LENGTH {
+ return Err(anyhow!("invalid bytes, length: {}", bytes.len()).into());
+ } else {
+ let magic = u32::from_le_bytes(bytes[0..4].try_into().unwrap());
+ let version = bytes[5];
+ let flag = bytes[6];
+ let length = u64::from_le_bytes(bytes[6..14].try_into().unwrap());
+ Ok(Self {
+ magic,
+ version,
+ flag,
+ length,
+ })
+ }
+ }
+
+ pub fn write_to(&self, writter: &mut &mut [u8]) -> Result<()> {
+ if writter.len() < SnapshotHeader::LENGTH {
+ return Err(anyhow!(
+ "writter buf is too small for writing the header, length: {}",
+ writter.len()
+ )
+ .into());
+ }
+ writter.write(self.magic.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.version.to_le_bytes().as_slice())
+ .unwrap();
+ writter.write(self.flag.to_le_bytes().as_slice()).unwrap();
+ writter.write(self.length.to_le_bytes().as_slice()).unwrap();
+ Ok(())
+ }
+}
+
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotRecordV1 {
+ id: u64,
+ time_range_start: i64,
Review Comment:
Use TimeRange here? so we know if it's inclusive or not.
##########
horaedb/metric_engine/src/manifest.rs:
##########
@@ -184,6 +203,264 @@ impl Manifest {
}
}
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotHeader {
+ pub magic: u32,
+ pub version: u8,
+ pub flag: u8,
+ pub length: u64,
+}
+
+impl SnapshotHeader {
+ // format: | magic(u32) | version(u8) | flags(u8) | length(u64) |
+ // The Magic field (u32) is used to ensure the validity of the data source.
+ // The Flags field (u8) is reserved for future extensibility, such as enabling
+ // compression or supporting additional features.
+ // The length field (u64) represents the total length of the subsequent records
+ // and serves as a straightforward method for verifying their integrity.
+ // (length = record_length
+ // * record_count)
+
+ // use #[repr(packed)] to force unalignment and avoid hard code internal types
+ // here, better solutions?
+ pub const LENGTH: usize = size_of::<SnapshotHeader>();
+ pub const MAGIC: u32 = 0x32A489BF;
+
+ pub fn new(length: u64) -> Self {
+ Self {
+ magic: SnapshotHeader::MAGIC,
+ version: SnapshotRecordV1::VERSION,
+ flag: 0,
+ length,
+ }
+ }
+
+ pub fn try_from_bytes(bytes: &[u8]) -> Result<Self> {
+ if bytes.len() < Self::LENGTH {
+ return Err(anyhow!("invalid bytes, length: {}", bytes.len()).into());
+ } else {
+ let magic = u32::from_le_bytes(bytes[0..4].try_into().unwrap());
+ let version = bytes[5];
+ let flag = bytes[6];
+ let length = u64::from_le_bytes(bytes[6..14].try_into().unwrap());
+ Ok(Self {
+ magic,
+ version,
+ flag,
+ length,
+ })
+ }
+ }
+
+ pub fn write_to(&self, writter: &mut &mut [u8]) -> Result<()> {
+ if writter.len() < SnapshotHeader::LENGTH {
+ return Err(anyhow!(
+ "writter buf is too small for writing the header, length: {}",
+ writter.len()
+ )
+ .into());
+ }
+ writter.write(self.magic.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.version.to_le_bytes().as_slice())
+ .unwrap();
+ writter.write(self.flag.to_le_bytes().as_slice()).unwrap();
+ writter.write(self.length.to_le_bytes().as_slice()).unwrap();
+ Ok(())
+ }
+}
+
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotRecordV1 {
+ id: u64,
+ time_range_start: i64,
+ time_range_end: i64,
+ size: u32,
+ num_rows: u32,
+}
+
+impl SnapshotRecordV1 {
+ // format: | id(u64) | time_range(i64*2)| size(u32) | num_rows(u32)|
+ const LENGTH: usize = size_of::<SnapshotRecordV1>();
+ pub const VERSION: u8 = 1;
+
+ pub fn write_to(&self, writter: &mut &mut [u8]) -> Result<()> {
+ if writter.len() < SnapshotRecordV1::LENGTH {
+ return Err(anyhow!(
+ "writter buf is too small for writing the record, length: {}",
+ writter.len()
+ )
+ .into());
+ }
+ writter.write(self.id.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.time_range_start.to_le_bytes().as_slice())
+ .unwrap();
+ writter
+ .write(self.time_range_end.to_le_bytes().as_slice())
+ .unwrap();
+ writter.write(self.size.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.num_rows.to_le_bytes().as_slice())
+ .unwrap();
+ Ok(())
+ }
+
+ pub fn id(&self) -> u64 {
+ self.id
+ }
+}
+
+impl From<SstFile> for SnapshotRecordV1 {
+ fn from(value: SstFile) -> Self {
+ SnapshotRecordV1 {
+ id: value.id(),
+ time_range_start: value.meta().time_range.start.0,
+ time_range_end: value.meta().time_range.end.0,
+ size: value.meta().size,
+ num_rows: value.meta().num_rows,
+ }
+ }
+}
+
+impl TryFrom<&[u8]> for SnapshotRecordV1 {
+ type Error = Error;
+
+ fn try_from(value: &[u8]) -> Result<Self> {
+ if value.len() < SnapshotRecordV1::LENGTH {
+ return Err(anyhow!("invalid value len: {}", value.len()).into());
+ }
+ let id = u64::from_le_bytes(value[0..8].try_into().unwrap());
+ let time_range_start = i64::from_le_bytes(value[8..16].try_into().unwrap());
+ let time_range_end = i64::from_le_bytes(value[16..24].try_into().unwrap());
+ let size = u32::from_le_bytes(value[24..28].try_into().unwrap());
+ let num_rows = u32::from_le_bytes(value[28..32].try_into().unwrap());
+ Ok(SnapshotRecordV1 {
+ id,
+ time_range_start,
+ time_range_end,
+ size,
+ num_rows,
+ })
+ }
+}
+
+impl Into<SstFile> for SnapshotRecordV1 {
+ fn into(self) -> SstFile {
+ let file_meta = FileMeta {
+ max_sequence: self.id(),
+ num_rows: self.num_rows,
+ size: self.size,
+ time_range: TimeRange::new(
+ Timestamp::from(self.time_range_start),
+ Timestamp::from(self.time_range_end),
+ ),
+ };
+ SstFile::new(self.id(), file_meta)
+ }
+}
+
+struct Snapshot {
+ header: SnapshotHeader,
+ inner: Bytes,
+}
+
+impl Default for Snapshot {
+ // create an empty Snapshot
+ fn default() -> Self {
+ let header = SnapshotHeader::new(0);
+ Self {
+ header,
+ inner: Bytes::new(),
+ }
+ }
+}
+
+impl Snapshot {
+ pub fn try_from_bytes(bytes: Bytes) -> Result<Self> {
+ if bytes.is_empty() {
+ return Ok(Snapshot::default());
+ }
+ let header = SnapshotHeader::try_from_bytes(bytes.as_bytes())?;
+ let header_length = header.length as usize;
+ if header_length > 0
Review Comment:
`ensure`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@horaedb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@horaedb.apache.org
For additional commands, e-mail: commits-help@horaedb.apache.org
Re: [PR] feat: another format for manifest [horaedb]
Posted by "jiacai2050 (via GitHub)" <gi...@apache.org>.
jiacai2050 commented on code in PR #1604:
URL: https://github.com/apache/horaedb/pull/1604#discussion_r1877759763
##########
horaedb/metric_engine/src/manifest.rs:
##########
@@ -184,6 +203,264 @@ impl Manifest {
}
}
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotHeader {
+ pub magic: u32,
+ pub version: u8,
+ pub flag: u8,
+ pub length: u64,
+}
+
+impl SnapshotHeader {
+ // format: | magic(u32) | version(u8) | flags(u8) | length(u64) |
+ // The Magic field (u32) is used to ensure the validity of the data source.
+ // The Flags field (u8) is reserved for future extensibility, such as enabling
+ // compression or supporting additional features.
+ // The length field (u64) represents the total length of the subsequent records
+ // and serves as a straightforward method for verifying their integrity.
+ // (length = record_length
+ // * record_count)
+
+ // use #[repr(packed)] to force unalignment and avoid hard code internal types
+ // here, better solutions?
+ pub const LENGTH: usize = size_of::<SnapshotHeader>();
Review Comment:
I think you can write it directly: `length = 4 /*magic*/ + 1 /*version*/ + 1 + 8 `
##########
horaedb/metric_engine/src/manifest.rs:
##########
@@ -184,6 +203,264 @@ impl Manifest {
}
}
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotHeader {
+ pub magic: u32,
+ pub version: u8,
+ pub flag: u8,
+ pub length: u64,
+}
+
+impl SnapshotHeader {
+ // format: | magic(u32) | version(u8) | flags(u8) | length(u64) |
+ // The Magic field (u32) is used to ensure the validity of the data source.
+ // The Flags field (u8) is reserved for future extensibility, such as enabling
+ // compression or supporting additional features.
+ // The length field (u64) represents the total length of the subsequent records
+ // and serves as a straightforward method for verifying their integrity.
+ // (length = record_length
+ // * record_count)
Review Comment:
No newline here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@horaedb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@horaedb.apache.org
For additional commands, e-mail: commits-help@horaedb.apache.org
Re: [PR] feat: another format for manifest [horaedb]
Posted by "jiacai2050 (via GitHub)" <gi...@apache.org>.
jiacai2050 commented on code in PR #1604:
URL: https://github.com/apache/horaedb/pull/1604#discussion_r1877763252
##########
horaedb/metric_engine/src/manifest.rs:
##########
@@ -184,6 +203,264 @@ impl Manifest {
}
}
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotHeader {
+ pub magic: u32,
+ pub version: u8,
+ pub flag: u8,
+ pub length: u64,
+}
+
+impl SnapshotHeader {
+ // format: | magic(u32) | version(u8) | flags(u8) | length(u64) |
+ // The Magic field (u32) is used to ensure the validity of the data source.
+ // The Flags field (u8) is reserved for future extensibility, such as enabling
+ // compression or supporting additional features.
+ // The length field (u64) represents the total length of the subsequent records
+ // and serves as a straightforward method for verifying their integrity.
+ // (length = record_length
+ // * record_count)
+
+ // use #[repr(packed)] to force unalignment and avoid hard code internal types
+ // here, better solutions?
+ pub const LENGTH: usize = size_of::<SnapshotHeader>();
+ pub const MAGIC: u32 = 0x32A489BF;
+
+ pub fn new(length: u64) -> Self {
+ Self {
+ magic: SnapshotHeader::MAGIC,
+ version: SnapshotRecordV1::VERSION,
+ flag: 0,
+ length,
+ }
+ }
+
+ pub fn try_from_bytes(bytes: &[u8]) -> Result<Self> {
+ if bytes.len() < Self::LENGTH {
+ return Err(anyhow!("invalid bytes, length: {}", bytes.len()).into());
+ } else {
+ let magic = u32::from_le_bytes(bytes[0..4].try_into().unwrap());
Review Comment:
Why unwrap is safe here? what if the file contains unexpected bytes, such as `FFFFFFFF`.
##########
horaedb/metric_engine/src/manifest.rs:
##########
@@ -184,6 +203,264 @@ impl Manifest {
}
}
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotHeader {
+ pub magic: u32,
+ pub version: u8,
+ pub flag: u8,
+ pub length: u64,
+}
+
+impl SnapshotHeader {
+ // format: | magic(u32) | version(u8) | flags(u8) | length(u64) |
+ // The Magic field (u32) is used to ensure the validity of the data source.
+ // The Flags field (u8) is reserved for future extensibility, such as enabling
+ // compression or supporting additional features.
+ // The length field (u64) represents the total length of the subsequent records
+ // and serves as a straightforward method for verifying their integrity.
+ // (length = record_length
+ // * record_count)
+
+ // use #[repr(packed)] to force unalignment and avoid hard code internal types
+ // here, better solutions?
+ pub const LENGTH: usize = size_of::<SnapshotHeader>();
+ pub const MAGIC: u32 = 0x32A489BF;
+
+ pub fn new(length: u64) -> Self {
+ Self {
+ magic: SnapshotHeader::MAGIC,
+ version: SnapshotRecordV1::VERSION,
+ flag: 0,
+ length,
+ }
+ }
+
+ pub fn try_from_bytes(bytes: &[u8]) -> Result<Self> {
+ if bytes.len() < Self::LENGTH {
Review Comment:
ensure!
##########
horaedb/metric_engine/src/manifest.rs:
##########
@@ -184,6 +203,264 @@ impl Manifest {
}
}
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotHeader {
+ pub magic: u32,
+ pub version: u8,
+ pub flag: u8,
+ pub length: u64,
+}
+
+impl SnapshotHeader {
+ // format: | magic(u32) | version(u8) | flags(u8) | length(u64) |
+ // The Magic field (u32) is used to ensure the validity of the data source.
+ // The Flags field (u8) is reserved for future extensibility, such as enabling
+ // compression or supporting additional features.
+ // The length field (u64) represents the total length of the subsequent records
+ // and serves as a straightforward method for verifying their integrity.
+ // (length = record_length
+ // * record_count)
+
+ // use #[repr(packed)] to force unalignment and avoid hard code internal types
+ // here, better solutions?
+ pub const LENGTH: usize = size_of::<SnapshotHeader>();
+ pub const MAGIC: u32 = 0x32A489BF;
+
+ pub fn new(length: u64) -> Self {
+ Self {
+ magic: SnapshotHeader::MAGIC,
+ version: SnapshotRecordV1::VERSION,
+ flag: 0,
+ length,
+ }
+ }
+
+ pub fn try_from_bytes(bytes: &[u8]) -> Result<Self> {
+ if bytes.len() < Self::LENGTH {
+ return Err(anyhow!("invalid bytes, length: {}", bytes.len()).into());
+ } else {
+ let magic = u32::from_le_bytes(bytes[0..4].try_into().unwrap());
+ let version = bytes[5];
+ let flag = bytes[6];
+ let length = u64::from_le_bytes(bytes[6..14].try_into().unwrap());
+ Ok(Self {
+ magic,
+ version,
+ flag,
+ length,
+ })
+ }
+ }
+
+ pub fn write_to(&self, writter: &mut &mut [u8]) -> Result<()> {
+ if writter.len() < SnapshotHeader::LENGTH {
+ return Err(anyhow!(
+ "writter buf is too small for writing the header, length: {}",
+ writter.len()
+ )
+ .into());
+ }
+ writter.write(self.magic.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.version.to_le_bytes().as_slice())
+ .unwrap();
+ writter.write(self.flag.to_le_bytes().as_slice()).unwrap();
+ writter.write(self.length.to_le_bytes().as_slice()).unwrap();
Review Comment:
Remove those unwrap.
##########
horaedb/metric_engine/src/manifest.rs:
##########
@@ -184,6 +203,264 @@ impl Manifest {
}
}
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotHeader {
+ pub magic: u32,
+ pub version: u8,
+ pub flag: u8,
+ pub length: u64,
+}
+
+impl SnapshotHeader {
+ // format: | magic(u32) | version(u8) | flags(u8) | length(u64) |
+ // The Magic field (u32) is used to ensure the validity of the data source.
+ // The Flags field (u8) is reserved for future extensibility, such as enabling
+ // compression or supporting additional features.
+ // The length field (u64) represents the total length of the subsequent records
+ // and serves as a straightforward method for verifying their integrity.
+ // (length = record_length
+ // * record_count)
+
+ // use #[repr(packed)] to force unalignment and avoid hard code internal types
+ // here, better solutions?
+ pub const LENGTH: usize = size_of::<SnapshotHeader>();
+ pub const MAGIC: u32 = 0x32A489BF;
+
+ pub fn new(length: u64) -> Self {
+ Self {
+ magic: SnapshotHeader::MAGIC,
+ version: SnapshotRecordV1::VERSION,
+ flag: 0,
+ length,
+ }
+ }
+
+ pub fn try_from_bytes(bytes: &[u8]) -> Result<Self> {
+ if bytes.len() < Self::LENGTH {
+ return Err(anyhow!("invalid bytes, length: {}", bytes.len()).into());
+ } else {
+ let magic = u32::from_le_bytes(bytes[0..4].try_into().unwrap());
+ let version = bytes[5];
+ let flag = bytes[6];
+ let length = u64::from_le_bytes(bytes[6..14].try_into().unwrap());
+ Ok(Self {
+ magic,
+ version,
+ flag,
+ length,
+ })
+ }
+ }
+
+ pub fn write_to(&self, writter: &mut &mut [u8]) -> Result<()> {
+ if writter.len() < SnapshotHeader::LENGTH {
+ return Err(anyhow!(
+ "writter buf is too small for writing the header, length: {}",
+ writter.len()
+ )
+ .into());
+ }
+ writter.write(self.magic.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.version.to_le_bytes().as_slice())
+ .unwrap();
+ writter.write(self.flag.to_le_bytes().as_slice()).unwrap();
+ writter.write(self.length.to_le_bytes().as_slice()).unwrap();
+ Ok(())
+ }
+}
+
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotRecordV1 {
+ id: u64,
+ time_range_start: i64,
+ time_range_end: i64,
+ size: u32,
+ num_rows: u32,
+}
+
+impl SnapshotRecordV1 {
+ // format: | id(u64) | time_range(i64*2)| size(u32) | num_rows(u32)|
+ const LENGTH: usize = size_of::<SnapshotRecordV1>();
+ pub const VERSION: u8 = 1;
+
+ pub fn write_to(&self, writter: &mut &mut [u8]) -> Result<()> {
+ if writter.len() < SnapshotRecordV1::LENGTH {
+ return Err(anyhow!(
+ "writter buf is too small for writing the record, length: {}",
+ writter.len()
+ )
+ .into());
+ }
+ writter.write(self.id.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.time_range_start.to_le_bytes().as_slice())
+ .unwrap();
+ writter
+ .write(self.time_range_end.to_le_bytes().as_slice())
+ .unwrap();
+ writter.write(self.size.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.num_rows.to_le_bytes().as_slice())
+ .unwrap();
+ Ok(())
+ }
+
+ pub fn id(&self) -> u64 {
+ self.id
+ }
+}
+
+impl From<SstFile> for SnapshotRecordV1 {
+ fn from(value: SstFile) -> Self {
+ SnapshotRecordV1 {
+ id: value.id(),
+ time_range_start: value.meta().time_range.start.0,
+ time_range_end: value.meta().time_range.end.0,
+ size: value.meta().size,
+ num_rows: value.meta().num_rows,
+ }
+ }
+}
+
+impl TryFrom<&[u8]> for SnapshotRecordV1 {
+ type Error = Error;
+
+ fn try_from(value: &[u8]) -> Result<Self> {
+ if value.len() < SnapshotRecordV1::LENGTH {
+ return Err(anyhow!("invalid value len: {}", value.len()).into());
+ }
+ let id = u64::from_le_bytes(value[0..8].try_into().unwrap());
+ let time_range_start = i64::from_le_bytes(value[8..16].try_into().unwrap());
+ let time_range_end = i64::from_le_bytes(value[16..24].try_into().unwrap());
+ let size = u32::from_le_bytes(value[24..28].try_into().unwrap());
+ let num_rows = u32::from_le_bytes(value[28..32].try_into().unwrap());
+ Ok(SnapshotRecordV1 {
+ id,
+ time_range_start,
+ time_range_end,
+ size,
+ num_rows,
+ })
+ }
+}
+
+impl Into<SstFile> for SnapshotRecordV1 {
+ fn into(self) -> SstFile {
+ let file_meta = FileMeta {
+ max_sequence: self.id(),
+ num_rows: self.num_rows,
+ size: self.size,
+ time_range: TimeRange::new(
+ Timestamp::from(self.time_range_start),
+ Timestamp::from(self.time_range_end),
+ ),
+ };
+ SstFile::new(self.id(), file_meta)
+ }
+}
+
+struct Snapshot {
+ header: SnapshotHeader,
+ inner: Bytes,
+}
+
+impl Default for Snapshot {
+ // create an empty Snapshot
+ fn default() -> Self {
+ let header = SnapshotHeader::new(0);
+ Self {
+ header,
+ inner: Bytes::new(),
+ }
+ }
+}
+
+impl Snapshot {
+ pub fn try_from_bytes(bytes: Bytes) -> Result<Self> {
+ if bytes.is_empty() {
+ return Ok(Snapshot::default());
+ }
+ let header = SnapshotHeader::try_from_bytes(bytes.as_bytes())?;
+ let header_length = header.length as usize;
+ if header_length > 0
Review Comment:
ensure
##########
horaedb/metric_engine/src/manifest.rs:
##########
@@ -184,6 +203,264 @@ impl Manifest {
}
}
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotHeader {
+ pub magic: u32,
+ pub version: u8,
+ pub flag: u8,
+ pub length: u64,
+}
+
+impl SnapshotHeader {
+ // format: | magic(u32) | version(u8) | flags(u8) | length(u64) |
+ // The Magic field (u32) is used to ensure the validity of the data source.
+ // The Flags field (u8) is reserved for future extensibility, such as enabling
+ // compression or supporting additional features.
+ // The length field (u64) represents the total length of the subsequent records
+ // and serves as a straightforward method for verifying their integrity.
+ // (length = record_length
+ // * record_count)
+
+ // use #[repr(packed)] to force unalignment and avoid hard code internal types
+ // here, better solutions?
+ pub const LENGTH: usize = size_of::<SnapshotHeader>();
+ pub const MAGIC: u32 = 0x32A489BF;
+
+ pub fn new(length: u64) -> Self {
+ Self {
+ magic: SnapshotHeader::MAGIC,
+ version: SnapshotRecordV1::VERSION,
+ flag: 0,
+ length,
+ }
+ }
+
+ pub fn try_from_bytes(bytes: &[u8]) -> Result<Self> {
+ if bytes.len() < Self::LENGTH {
+ return Err(anyhow!("invalid bytes, length: {}", bytes.len()).into());
+ } else {
+ let magic = u32::from_le_bytes(bytes[0..4].try_into().unwrap());
+ let version = bytes[5];
+ let flag = bytes[6];
+ let length = u64::from_le_bytes(bytes[6..14].try_into().unwrap());
+ Ok(Self {
+ magic,
+ version,
+ flag,
+ length,
+ })
+ }
+ }
+
+ pub fn write_to(&self, writter: &mut &mut [u8]) -> Result<()> {
+ if writter.len() < SnapshotHeader::LENGTH {
Review Comment:
ensure
##########
horaedb/metric_engine/src/manifest.rs:
##########
@@ -184,6 +203,264 @@ impl Manifest {
}
}
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotHeader {
+ pub magic: u32,
+ pub version: u8,
+ pub flag: u8,
+ pub length: u64,
+}
+
+impl SnapshotHeader {
+ // format: | magic(u32) | version(u8) | flags(u8) | length(u64) |
+ // The Magic field (u32) is used to ensure the validity of the data source.
+ // The Flags field (u8) is reserved for future extensibility, such as enabling
+ // compression or supporting additional features.
+ // The length field (u64) represents the total length of the subsequent records
+ // and serves as a straightforward method for verifying their integrity.
+ // (length = record_length
+ // * record_count)
+
+ // use #[repr(packed)] to force unalignment and avoid hard code internal types
+ // here, better solutions?
+ pub const LENGTH: usize = size_of::<SnapshotHeader>();
+ pub const MAGIC: u32 = 0x32A489BF;
+
+ pub fn new(length: u64) -> Self {
+ Self {
+ magic: SnapshotHeader::MAGIC,
+ version: SnapshotRecordV1::VERSION,
+ flag: 0,
+ length,
+ }
+ }
+
+ pub fn try_from_bytes(bytes: &[u8]) -> Result<Self> {
+ if bytes.len() < Self::LENGTH {
+ return Err(anyhow!("invalid bytes, length: {}", bytes.len()).into());
+ } else {
+ let magic = u32::from_le_bytes(bytes[0..4].try_into().unwrap());
+ let version = bytes[5];
+ let flag = bytes[6];
+ let length = u64::from_le_bytes(bytes[6..14].try_into().unwrap());
+ Ok(Self {
+ magic,
+ version,
+ flag,
+ length,
+ })
+ }
+ }
+
+ pub fn write_to(&self, writter: &mut &mut [u8]) -> Result<()> {
+ if writter.len() < SnapshotHeader::LENGTH {
+ return Err(anyhow!(
+ "writter buf is too small for writing the header, length: {}",
+ writter.len()
+ )
+ .into());
+ }
+ writter.write(self.magic.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.version.to_le_bytes().as_slice())
+ .unwrap();
+ writter.write(self.flag.to_le_bytes().as_slice()).unwrap();
+ writter.write(self.length.to_le_bytes().as_slice()).unwrap();
+ Ok(())
+ }
+}
+
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotRecordV1 {
+ id: u64,
+ time_range_start: i64,
+ time_range_end: i64,
+ size: u32,
+ num_rows: u32,
+}
+
+impl SnapshotRecordV1 {
+ // format: | id(u64) | time_range(i64*2)| size(u32) | num_rows(u32)|
+ const LENGTH: usize = size_of::<SnapshotRecordV1>();
+ pub const VERSION: u8 = 1;
+
+ pub fn write_to(&self, writter: &mut &mut [u8]) -> Result<()> {
+ if writter.len() < SnapshotRecordV1::LENGTH {
Review Comment:
ensure
##########
horaedb/metric_engine/src/manifest.rs:
##########
@@ -184,6 +203,264 @@ impl Manifest {
}
}
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotHeader {
+ pub magic: u32,
+ pub version: u8,
+ pub flag: u8,
+ pub length: u64,
+}
+
+impl SnapshotHeader {
+ // format: | magic(u32) | version(u8) | flags(u8) | length(u64) |
+ // The Magic field (u32) is used to ensure the validity of the data source.
+ // The Flags field (u8) is reserved for future extensibility, such as enabling
+ // compression or supporting additional features.
+ // The length field (u64) represents the total length of the subsequent records
+ // and serves as a straightforward method for verifying their integrity.
+ // (length = record_length
+ // * record_count)
+
+ // use #[repr(packed)] to force unalignment and avoid hard code internal types
+ // here, better solutions?
+ pub const LENGTH: usize = size_of::<SnapshotHeader>();
+ pub const MAGIC: u32 = 0x32A489BF;
+
+ pub fn new(length: u64) -> Self {
+ Self {
+ magic: SnapshotHeader::MAGIC,
+ version: SnapshotRecordV1::VERSION,
+ flag: 0,
+ length,
+ }
+ }
+
+ pub fn try_from_bytes(bytes: &[u8]) -> Result<Self> {
+ if bytes.len() < Self::LENGTH {
+ return Err(anyhow!("invalid bytes, length: {}", bytes.len()).into());
+ } else {
+ let magic = u32::from_le_bytes(bytes[0..4].try_into().unwrap());
+ let version = bytes[5];
+ let flag = bytes[6];
+ let length = u64::from_le_bytes(bytes[6..14].try_into().unwrap());
+ Ok(Self {
+ magic,
+ version,
+ flag,
+ length,
+ })
+ }
+ }
+
+ pub fn write_to(&self, writter: &mut &mut [u8]) -> Result<()> {
Review Comment:
typo: writer
##########
horaedb/metric_engine/src/manifest.rs:
##########
@@ -184,6 +203,264 @@ impl Manifest {
}
}
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotHeader {
+ pub magic: u32,
+ pub version: u8,
+ pub flag: u8,
+ pub length: u64,
+}
+
+impl SnapshotHeader {
+ // format: | magic(u32) | version(u8) | flags(u8) | length(u64) |
+ // The Magic field (u32) is used to ensure the validity of the data source.
+ // The Flags field (u8) is reserved for future extensibility, such as enabling
+ // compression or supporting additional features.
+ // The length field (u64) represents the total length of the subsequent records
+ // and serves as a straightforward method for verifying their integrity.
+ // (length = record_length
+ // * record_count)
+
+ // use #[repr(packed)] to force unalignment and avoid hard code internal types
+ // here, better solutions?
+ pub const LENGTH: usize = size_of::<SnapshotHeader>();
+ pub const MAGIC: u32 = 0x32A489BF;
+
+ pub fn new(length: u64) -> Self {
+ Self {
+ magic: SnapshotHeader::MAGIC,
+ version: SnapshotRecordV1::VERSION,
+ flag: 0,
+ length,
+ }
+ }
+
+ pub fn try_from_bytes(bytes: &[u8]) -> Result<Self> {
+ if bytes.len() < Self::LENGTH {
+ return Err(anyhow!("invalid bytes, length: {}", bytes.len()).into());
+ } else {
+ let magic = u32::from_le_bytes(bytes[0..4].try_into().unwrap());
+ let version = bytes[5];
+ let flag = bytes[6];
+ let length = u64::from_le_bytes(bytes[6..14].try_into().unwrap());
+ Ok(Self {
+ magic,
+ version,
+ flag,
+ length,
+ })
+ }
+ }
+
+ pub fn write_to(&self, writter: &mut &mut [u8]) -> Result<()> {
+ if writter.len() < SnapshotHeader::LENGTH {
+ return Err(anyhow!(
+ "writter buf is too small for writing the header, length: {}",
+ writter.len()
+ )
+ .into());
+ }
+ writter.write(self.magic.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.version.to_le_bytes().as_slice())
+ .unwrap();
+ writter.write(self.flag.to_le_bytes().as_slice()).unwrap();
+ writter.write(self.length.to_le_bytes().as_slice()).unwrap();
+ Ok(())
+ }
+}
+
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotRecordV1 {
+ id: u64,
+ time_range_start: i64,
+ time_range_end: i64,
+ size: u32,
+ num_rows: u32,
+}
+
+impl SnapshotRecordV1 {
+ // format: | id(u64) | time_range(i64*2)| size(u32) | num_rows(u32)|
+ const LENGTH: usize = size_of::<SnapshotRecordV1>();
+ pub const VERSION: u8 = 1;
+
+ pub fn write_to(&self, writter: &mut &mut [u8]) -> Result<()> {
+ if writter.len() < SnapshotRecordV1::LENGTH {
+ return Err(anyhow!(
+ "writter buf is too small for writing the record, length: {}",
+ writter.len()
+ )
+ .into());
+ }
+ writter.write(self.id.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.time_range_start.to_le_bytes().as_slice())
+ .unwrap();
+ writter
+ .write(self.time_range_end.to_le_bytes().as_slice())
+ .unwrap();
+ writter.write(self.size.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.num_rows.to_le_bytes().as_slice())
+ .unwrap();
Review Comment:
Remove those unwrap.
##########
horaedb/metric_engine/src/manifest.rs:
##########
@@ -184,6 +203,264 @@ impl Manifest {
}
}
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotHeader {
+ pub magic: u32,
+ pub version: u8,
+ pub flag: u8,
+ pub length: u64,
+}
+
+impl SnapshotHeader {
+ // format: | magic(u32) | version(u8) | flags(u8) | length(u64) |
+ // The Magic field (u32) is used to ensure the validity of the data source.
+ // The Flags field (u8) is reserved for future extensibility, such as enabling
+ // compression or supporting additional features.
+ // The length field (u64) represents the total length of the subsequent records
+ // and serves as a straightforward method for verifying their integrity.
+ // (length = record_length
+ // * record_count)
+
+ // use #[repr(packed)] to force unalignment and avoid hard code internal types
+ // here, better solutions?
+ pub const LENGTH: usize = size_of::<SnapshotHeader>();
+ pub const MAGIC: u32 = 0x32A489BF;
+
+ pub fn new(length: u64) -> Self {
+ Self {
+ magic: SnapshotHeader::MAGIC,
+ version: SnapshotRecordV1::VERSION,
+ flag: 0,
+ length,
+ }
+ }
+
+ pub fn try_from_bytes(bytes: &[u8]) -> Result<Self> {
+ if bytes.len() < Self::LENGTH {
+ return Err(anyhow!("invalid bytes, length: {}", bytes.len()).into());
+ } else {
+ let magic = u32::from_le_bytes(bytes[0..4].try_into().unwrap());
+ let version = bytes[5];
+ let flag = bytes[6];
+ let length = u64::from_le_bytes(bytes[6..14].try_into().unwrap());
+ Ok(Self {
+ magic,
+ version,
+ flag,
+ length,
+ })
+ }
+ }
+
+ pub fn write_to(&self, writter: &mut &mut [u8]) -> Result<()> {
+ if writter.len() < SnapshotHeader::LENGTH {
+ return Err(anyhow!(
+ "writter buf is too small for writing the header, length: {}",
+ writter.len()
+ )
+ .into());
+ }
+ writter.write(self.magic.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.version.to_le_bytes().as_slice())
+ .unwrap();
+ writter.write(self.flag.to_le_bytes().as_slice()).unwrap();
+ writter.write(self.length.to_le_bytes().as_slice()).unwrap();
+ Ok(())
+ }
+}
+
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotRecordV1 {
+ id: u64,
+ time_range_start: i64,
+ time_range_end: i64,
+ size: u32,
+ num_rows: u32,
+}
+
+impl SnapshotRecordV1 {
+ // format: | id(u64) | time_range(i64*2)| size(u32) | num_rows(u32)|
+ const LENGTH: usize = size_of::<SnapshotRecordV1>();
+ pub const VERSION: u8 = 1;
+
+ pub fn write_to(&self, writter: &mut &mut [u8]) -> Result<()> {
+ if writter.len() < SnapshotRecordV1::LENGTH {
+ return Err(anyhow!(
+ "writter buf is too small for writing the record, length: {}",
+ writter.len()
+ )
+ .into());
+ }
+ writter.write(self.id.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.time_range_start.to_le_bytes().as_slice())
+ .unwrap();
+ writter
+ .write(self.time_range_end.to_le_bytes().as_slice())
+ .unwrap();
+ writter.write(self.size.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.num_rows.to_le_bytes().as_slice())
+ .unwrap();
+ Ok(())
+ }
+
+ pub fn id(&self) -> u64 {
+ self.id
+ }
+}
+
+impl From<SstFile> for SnapshotRecordV1 {
+ fn from(value: SstFile) -> Self {
+ SnapshotRecordV1 {
+ id: value.id(),
+ time_range_start: value.meta().time_range.start.0,
+ time_range_end: value.meta().time_range.end.0,
+ size: value.meta().size,
+ num_rows: value.meta().num_rows,
+ }
+ }
+}
+
+impl TryFrom<&[u8]> for SnapshotRecordV1 {
+ type Error = Error;
+
+ fn try_from(value: &[u8]) -> Result<Self> {
+ if value.len() < SnapshotRecordV1::LENGTH {
+ return Err(anyhow!("invalid value len: {}", value.len()).into());
+ }
+ let id = u64::from_le_bytes(value[0..8].try_into().unwrap());
+ let time_range_start = i64::from_le_bytes(value[8..16].try_into().unwrap());
+ let time_range_end = i64::from_le_bytes(value[16..24].try_into().unwrap());
+ let size = u32::from_le_bytes(value[24..28].try_into().unwrap());
+ let num_rows = u32::from_le_bytes(value[28..32].try_into().unwrap());
+ Ok(SnapshotRecordV1 {
+ id,
+ time_range_start,
+ time_range_end,
+ size,
+ num_rows,
+ })
+ }
+}
+
+impl Into<SstFile> for SnapshotRecordV1 {
+ fn into(self) -> SstFile {
+ let file_meta = FileMeta {
+ max_sequence: self.id(),
+ num_rows: self.num_rows,
+ size: self.size,
+ time_range: TimeRange::new(
+ Timestamp::from(self.time_range_start),
+ Timestamp::from(self.time_range_end),
+ ),
+ };
+ SstFile::new(self.id(), file_meta)
+ }
+}
+
+struct Snapshot {
+ header: SnapshotHeader,
+ inner: Bytes,
+}
+
+impl Default for Snapshot {
+ // create an empty Snapshot
+ fn default() -> Self {
+ let header = SnapshotHeader::new(0);
+ Self {
+ header,
+ inner: Bytes::new(),
+ }
+ }
+}
+
+impl Snapshot {
+ pub fn try_from_bytes(bytes: Bytes) -> Result<Self> {
+ if bytes.is_empty() {
+ return Ok(Snapshot::default());
+ }
+ let header = SnapshotHeader::try_from_bytes(bytes.as_bytes())?;
+ let header_length = header.length as usize;
+ if header_length > 0
+ && (header_length % SnapshotRecordV1::LENGTH != 0
+ || header_length + SnapshotHeader::LENGTH != bytes.len())
+ {
+ return Err(anyhow!(
+ "create snapshot from bytes failed, invalid bytes, header length = {}, total length: {}",
+ header_length,
+ bytes.len()
+ ).into());
+ }
+ Ok(Self {
+ header,
+ inner: bytes,
+ })
+ }
+
+ pub fn to_sstfiles(&self) -> Result<Vec<SstFile>> {
+ if self.header.length == 0 {
+ return Ok(Vec::new());
+ } else {
+ let buf = self.inner.as_bytes();
+ let mut result: Vec<SstFile> = Vec::new();
+ let mut index = SnapshotHeader::LENGTH;
+ while index < buf.len() {
+ let record =
+ SnapshotRecordV1::try_from(&buf[index..index + SnapshotRecordV1::LENGTH])?;
+ index += SnapshotRecordV1::LENGTH;
+ result.push(record.into());
+ }
+
+ Ok(result)
+ }
+ }
+
+ pub fn dedup_sstfiles(&self, sstfiles: &mut Vec<SstFile>) -> Result<()> {
+ let buf = self.inner.as_bytes();
+ let mut ids = HashSet::new();
+ let mut index = SnapshotHeader::LENGTH;
+ while index < buf.len() {
+ let record = SnapshotRecordV1::try_from(&buf[index..index + SnapshotRecordV1::LENGTH])?;
+ index += SnapshotRecordV1::LENGTH;
+ ids.insert(record.id());
+ }
+ sstfiles.retain(|item| !ids.contains(&item.id()));
+
+ Ok(())
+ }
+
+ pub fn merge_sstfiles(&mut self, sstfiles: Vec<SstFile>) {
+ // update header
+ self.header.length += (sstfiles.len() * SnapshotRecordV1::LENGTH) as u64;
+ // final snapshot
+ let mut snapshot = vec![1u8; SnapshotHeader::LENGTH + self.header.length as usize];
Review Comment:
Why to init all bytes to 1?
##########
horaedb/metric_engine/src/manifest.rs:
##########
@@ -184,6 +203,264 @@ impl Manifest {
}
}
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotHeader {
+ pub magic: u32,
+ pub version: u8,
+ pub flag: u8,
+ pub length: u64,
+}
+
+impl SnapshotHeader {
+ // format: | magic(u32) | version(u8) | flags(u8) | length(u64) |
+ // The Magic field (u32) is used to ensure the validity of the data source.
+ // The Flags field (u8) is reserved for future extensibility, such as enabling
+ // compression or supporting additional features.
+ // The length field (u64) represents the total length of the subsequent records
+ // and serves as a straightforward method for verifying their integrity.
+ // (length = record_length
+ // * record_count)
+
+ // use #[repr(packed)] to force unalignment and avoid hard code internal types
+ // here, better solutions?
+ pub const LENGTH: usize = size_of::<SnapshotHeader>();
+ pub const MAGIC: u32 = 0x32A489BF;
+
+ pub fn new(length: u64) -> Self {
+ Self {
+ magic: SnapshotHeader::MAGIC,
+ version: SnapshotRecordV1::VERSION,
+ flag: 0,
+ length,
+ }
+ }
+
+ pub fn try_from_bytes(bytes: &[u8]) -> Result<Self> {
+ if bytes.len() < Self::LENGTH {
+ return Err(anyhow!("invalid bytes, length: {}", bytes.len()).into());
+ } else {
+ let magic = u32::from_le_bytes(bytes[0..4].try_into().unwrap());
+ let version = bytes[5];
+ let flag = bytes[6];
+ let length = u64::from_le_bytes(bytes[6..14].try_into().unwrap());
+ Ok(Self {
+ magic,
+ version,
+ flag,
+ length,
+ })
+ }
+ }
+
+ pub fn write_to(&self, writter: &mut &mut [u8]) -> Result<()> {
+ if writter.len() < SnapshotHeader::LENGTH {
+ return Err(anyhow!(
+ "writter buf is too small for writing the header, length: {}",
+ writter.len()
+ )
+ .into());
+ }
+ writter.write(self.magic.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.version.to_le_bytes().as_slice())
+ .unwrap();
+ writter.write(self.flag.to_le_bytes().as_slice()).unwrap();
+ writter.write(self.length.to_le_bytes().as_slice()).unwrap();
+ Ok(())
+ }
+}
+
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotRecordV1 {
+ id: u64,
+ time_range_start: i64,
+ time_range_end: i64,
+ size: u32,
+ num_rows: u32,
+}
+
+impl SnapshotRecordV1 {
+ // format: | id(u64) | time_range(i64*2)| size(u32) | num_rows(u32)|
+ const LENGTH: usize = size_of::<SnapshotRecordV1>();
+ pub const VERSION: u8 = 1;
+
+ pub fn write_to(&self, writter: &mut &mut [u8]) -> Result<()> {
+ if writter.len() < SnapshotRecordV1::LENGTH {
+ return Err(anyhow!(
+ "writter buf is too small for writing the record, length: {}",
+ writter.len()
+ )
+ .into());
+ }
+ writter.write(self.id.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.time_range_start.to_le_bytes().as_slice())
+ .unwrap();
+ writter
+ .write(self.time_range_end.to_le_bytes().as_slice())
+ .unwrap();
+ writter.write(self.size.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.num_rows.to_le_bytes().as_slice())
+ .unwrap();
+ Ok(())
+ }
+
+ pub fn id(&self) -> u64 {
+ self.id
+ }
+}
+
+impl From<SstFile> for SnapshotRecordV1 {
+ fn from(value: SstFile) -> Self {
+ SnapshotRecordV1 {
+ id: value.id(),
+ time_range_start: value.meta().time_range.start.0,
+ time_range_end: value.meta().time_range.end.0,
+ size: value.meta().size,
+ num_rows: value.meta().num_rows,
+ }
+ }
+}
+
+impl TryFrom<&[u8]> for SnapshotRecordV1 {
+ type Error = Error;
+
+ fn try_from(value: &[u8]) -> Result<Self> {
+ if value.len() < SnapshotRecordV1::LENGTH {
+ return Err(anyhow!("invalid value len: {}", value.len()).into());
+ }
+ let id = u64::from_le_bytes(value[0..8].try_into().unwrap());
+ let time_range_start = i64::from_le_bytes(value[8..16].try_into().unwrap());
+ let time_range_end = i64::from_le_bytes(value[16..24].try_into().unwrap());
+ let size = u32::from_le_bytes(value[24..28].try_into().unwrap());
+ let num_rows = u32::from_le_bytes(value[28..32].try_into().unwrap());
+ Ok(SnapshotRecordV1 {
+ id,
+ time_range_start,
+ time_range_end,
+ size,
+ num_rows,
+ })
+ }
+}
+
+impl Into<SstFile> for SnapshotRecordV1 {
+ fn into(self) -> SstFile {
+ let file_meta = FileMeta {
+ max_sequence: self.id(),
+ num_rows: self.num_rows,
+ size: self.size,
+ time_range: TimeRange::new(
+ Timestamp::from(self.time_range_start),
+ Timestamp::from(self.time_range_end),
+ ),
+ };
+ SstFile::new(self.id(), file_meta)
+ }
+}
+
+struct Snapshot {
+ header: SnapshotHeader,
+ inner: Bytes,
+}
+
+impl Default for Snapshot {
+ // create an empty Snapshot
+ fn default() -> Self {
+ let header = SnapshotHeader::new(0);
+ Self {
+ header,
+ inner: Bytes::new(),
+ }
+ }
+}
+
+impl Snapshot {
+ pub fn try_from_bytes(bytes: Bytes) -> Result<Self> {
+ if bytes.is_empty() {
+ return Ok(Snapshot::default());
+ }
+ let header = SnapshotHeader::try_from_bytes(bytes.as_bytes())?;
+ let header_length = header.length as usize;
+ if header_length > 0
+ && (header_length % SnapshotRecordV1::LENGTH != 0
+ || header_length + SnapshotHeader::LENGTH != bytes.len())
+ {
+ return Err(anyhow!(
+ "create snapshot from bytes failed, invalid bytes, header length = {}, total length: {}",
+ header_length,
+ bytes.len()
+ ).into());
+ }
+ Ok(Self {
+ header,
+ inner: bytes,
+ })
+ }
+
+ pub fn to_sstfiles(&self) -> Result<Vec<SstFile>> {
+ if self.header.length == 0 {
+ return Ok(Vec::new());
+ } else {
+ let buf = self.inner.as_bytes();
+ let mut result: Vec<SstFile> = Vec::new();
+ let mut index = SnapshotHeader::LENGTH;
+ while index < buf.len() {
+ let record =
+ SnapshotRecordV1::try_from(&buf[index..index + SnapshotRecordV1::LENGTH])?;
+ index += SnapshotRecordV1::LENGTH;
+ result.push(record.into());
+ }
+
+ Ok(result)
+ }
+ }
+
+ pub fn dedup_sstfiles(&self, sstfiles: &mut Vec<SstFile>) -> Result<()> {
+ let buf = self.inner.as_bytes();
+ let mut ids = HashSet::new();
+ let mut index = SnapshotHeader::LENGTH;
+ while index < buf.len() {
+ let record = SnapshotRecordV1::try_from(&buf[index..index + SnapshotRecordV1::LENGTH])?;
+ index += SnapshotRecordV1::LENGTH;
+ ids.insert(record.id());
+ }
+ sstfiles.retain(|item| !ids.contains(&item.id()));
+
+ Ok(())
+ }
+
+ pub fn merge_sstfiles(&mut self, sstfiles: Vec<SstFile>) {
+ // update header
+ self.header.length += (sstfiles.len() * SnapshotRecordV1::LENGTH) as u64;
+ // final snapshot
+ let mut snapshot = vec![1u8; SnapshotHeader::LENGTH + self.header.length as usize];
+ let mut writter = snapshot.as_mut_slice();
Review Comment:
typo: writer
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@horaedb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@horaedb.apache.org
For additional commands, e-mail: commits-help@horaedb.apache.org
Re: [PR] feat: another format for manifest [horaedb]
Posted by "zealchen (via GitHub)" <gi...@apache.org>.
zealchen commented on code in PR #1604:
URL: https://github.com/apache/horaedb/pull/1604#discussion_r1881350493
##########
horaedb/metric_engine/src/manifest.rs:
##########
@@ -184,6 +203,264 @@ impl Manifest {
}
}
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotHeader {
+ pub magic: u32,
+ pub version: u8,
+ pub flag: u8,
+ pub length: u64,
+}
+
+impl SnapshotHeader {
+ // format: | magic(u32) | version(u8) | flags(u8) | length(u64) |
+ // The Magic field (u32) is used to ensure the validity of the data source.
+ // The Flags field (u8) is reserved for future extensibility, such as enabling
+ // compression or supporting additional features.
+ // The length field (u64) represents the total length of the subsequent records
+ // and serves as a straightforward method for verifying their integrity.
+ // (length = record_length
+ // * record_count)
+
+ // use #[repr(packed)] to force unalignment and avoid hard code internal types
+ // here, better solutions?
+ pub const LENGTH: usize = size_of::<SnapshotHeader>();
+ pub const MAGIC: u32 = 0x32A489BF;
+
+ pub fn new(length: u64) -> Self {
+ Self {
+ magic: SnapshotHeader::MAGIC,
+ version: SnapshotRecordV1::VERSION,
+ flag: 0,
+ length,
+ }
+ }
+
+ pub fn try_from_bytes(bytes: &[u8]) -> Result<Self> {
+ if bytes.len() < Self::LENGTH {
+ return Err(anyhow!("invalid bytes, length: {}", bytes.len()).into());
+ } else {
+ let magic = u32::from_le_bytes(bytes[0..4].try_into().unwrap());
+ let version = bytes[5];
+ let flag = bytes[6];
+ let length = u64::from_le_bytes(bytes[6..14].try_into().unwrap());
+ Ok(Self {
+ magic,
+ version,
+ flag,
+ length,
+ })
+ }
+ }
+
+ pub fn write_to(&self, writter: &mut &mut [u8]) -> Result<()> {
+ if writter.len() < SnapshotHeader::LENGTH {
+ return Err(anyhow!(
+ "writter buf is too small for writing the header, length: {}",
+ writter.len()
+ )
+ .into());
+ }
+ writter.write(self.magic.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.version.to_le_bytes().as_slice())
+ .unwrap();
+ writter.write(self.flag.to_le_bytes().as_slice()).unwrap();
+ writter.write(self.length.to_le_bytes().as_slice()).unwrap();
+ Ok(())
+ }
+}
+
+#[repr(packed)]
+#[derive(Debug)]
+struct SnapshotRecordV1 {
+ id: u64,
+ time_range_start: i64,
+ time_range_end: i64,
+ size: u32,
+ num_rows: u32,
+}
+
+impl SnapshotRecordV1 {
+ // format: | id(u64) | time_range(i64*2)| size(u32) | num_rows(u32)|
+ const LENGTH: usize = size_of::<SnapshotRecordV1>();
+ pub const VERSION: u8 = 1;
+
+ pub fn write_to(&self, writter: &mut &mut [u8]) -> Result<()> {
+ if writter.len() < SnapshotRecordV1::LENGTH {
+ return Err(anyhow!(
+ "writter buf is too small for writing the record, length: {}",
+ writter.len()
+ )
+ .into());
+ }
+ writter.write(self.id.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.time_range_start.to_le_bytes().as_slice())
+ .unwrap();
+ writter
+ .write(self.time_range_end.to_le_bytes().as_slice())
+ .unwrap();
+ writter.write(self.size.to_le_bytes().as_slice()).unwrap();
+ writter
+ .write(self.num_rows.to_le_bytes().as_slice())
+ .unwrap();
+ Ok(())
+ }
+
+ pub fn id(&self) -> u64 {
+ self.id
+ }
+}
+
+impl From<SstFile> for SnapshotRecordV1 {
+ fn from(value: SstFile) -> Self {
+ SnapshotRecordV1 {
+ id: value.id(),
+ time_range_start: value.meta().time_range.start.0,
+ time_range_end: value.meta().time_range.end.0,
+ size: value.meta().size,
+ num_rows: value.meta().num_rows,
+ }
+ }
+}
+
+impl TryFrom<&[u8]> for SnapshotRecordV1 {
+ type Error = Error;
+
+ fn try_from(value: &[u8]) -> Result<Self> {
+ if value.len() < SnapshotRecordV1::LENGTH {
+ return Err(anyhow!("invalid value len: {}", value.len()).into());
+ }
+ let id = u64::from_le_bytes(value[0..8].try_into().unwrap());
+ let time_range_start = i64::from_le_bytes(value[8..16].try_into().unwrap());
+ let time_range_end = i64::from_le_bytes(value[16..24].try_into().unwrap());
+ let size = u32::from_le_bytes(value[24..28].try_into().unwrap());
+ let num_rows = u32::from_le_bytes(value[28..32].try_into().unwrap());
+ Ok(SnapshotRecordV1 {
+ id,
+ time_range_start,
+ time_range_end,
+ size,
+ num_rows,
+ })
+ }
+}
+
+impl Into<SstFile> for SnapshotRecordV1 {
+ fn into(self) -> SstFile {
+ let file_meta = FileMeta {
+ max_sequence: self.id(),
+ num_rows: self.num_rows,
+ size: self.size,
+ time_range: TimeRange::new(
+ Timestamp::from(self.time_range_start),
+ Timestamp::from(self.time_range_end),
+ ),
+ };
+ SstFile::new(self.id(), file_meta)
+ }
+}
+
+struct Snapshot {
+ header: SnapshotHeader,
+ inner: Bytes,
+}
+
+impl Default for Snapshot {
+ // create an empty Snapshot
+ fn default() -> Self {
+ let header = SnapshotHeader::new(0);
+ Self {
+ header,
+ inner: Bytes::new(),
+ }
+ }
+}
+
+impl Snapshot {
+ pub fn try_from_bytes(bytes: Bytes) -> Result<Self> {
+ if bytes.is_empty() {
+ return Ok(Snapshot::default());
+ }
+ let header = SnapshotHeader::try_from_bytes(bytes.as_bytes())?;
+ let header_length = header.length as usize;
+ if header_length > 0
+ && (header_length % SnapshotRecordV1::LENGTH != 0
+ || header_length + SnapshotHeader::LENGTH != bytes.len())
+ {
+ return Err(anyhow!(
+ "create snapshot from bytes failed, invalid bytes, header length = {}, total length: {}",
+ header_length,
+ bytes.len()
+ ).into());
+ }
+ Ok(Self {
+ header,
+ inner: bytes,
+ })
+ }
+
+ pub fn to_sstfiles(&self) -> Result<Vec<SstFile>> {
+ if self.header.length == 0 {
+ return Ok(Vec::new());
+ } else {
+ let buf = self.inner.as_bytes();
+ let mut result: Vec<SstFile> = Vec::new();
+ let mut index = SnapshotHeader::LENGTH;
+ while index < buf.len() {
+ let record =
+ SnapshotRecordV1::try_from(&buf[index..index + SnapshotRecordV1::LENGTH])?;
+ index += SnapshotRecordV1::LENGTH;
+ result.push(record.into());
+ }
+
+ Ok(result)
+ }
+ }
+
+ pub fn dedup_sstfiles(&self, sstfiles: &mut Vec<SstFile>) -> Result<()> {
+ let buf = self.inner.as_bytes();
+ let mut ids = HashSet::new();
+ let mut index = SnapshotHeader::LENGTH;
+ while index < buf.len() {
+ let record = SnapshotRecordV1::try_from(&buf[index..index + SnapshotRecordV1::LENGTH])?;
+ index += SnapshotRecordV1::LENGTH;
+ ids.insert(record.id());
+ }
+ sstfiles.retain(|item| !ids.contains(&item.id()));
+
+ Ok(())
+ }
+
+ pub fn merge_sstfiles(&mut self, sstfiles: Vec<SstFile>) {
+ // update header
+ self.header.length += (sstfiles.len() * SnapshotRecordV1::LENGTH) as u64;
+ // final snapshot
+ let mut snapshot = vec![1u8; SnapshotHeader::LENGTH + self.header.length as usize];
Review Comment:
It should be 0u8. It's simply a mistake.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@horaedb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@horaedb.apache.org
For additional commands, e-mail: commits-help@horaedb.apache.org
Re: [PR] feat: another format for manifest [horaedb]
Posted by "jiacai2050 (via GitHub)" <gi...@apache.org>.
jiacai2050 merged PR #1604:
URL: https://github.com/apache/horaedb/pull/1604
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@horaedb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@horaedb.apache.org
For additional commands, e-mail: commits-help@horaedb.apache.org