nodedb 0.4.0

Local-first, real-time, edge-to-cloud hybrid database for multi-modal workloads
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149

// SPDX-License-Identifier: BUSL-1.1

//! `REWRITE PARTITIONS FOR <name>`
//!
//! Triggers an async background rewrite of all sealed partitions
//! for a timeseries collection. Non-blocking — returns immediately.
//! Useful for reclaiming space after column drops or applying new compression.

use nodedb_types::DatabaseId;

use crate::control::security::identity::AuthenticatedIdentity;
use crate::control::state::SharedState;

use super::super::super::result::{DdlError, DdlResult};
use super::helpers::ddl_err;

/// REWRITE PARTITIONS FOR <name>
pub fn rewrite_partitions(
    state: &SharedState,
    identity: &AuthenticatedIdentity,
    parts: &[&str],
) -> Result<Vec<DdlResult>, DdlError> {
    // REWRITE PARTITIONS FOR <name>
    if parts.len() < 4 {
        return Err(ddl_err(
            "42601",
            "syntax: REWRITE PARTITIONS FOR <collection>",
        ));
    }

    let name = parts[3].to_lowercase();
    let tenant_id = identity.tenant_id;

    // Verify collection exists and is timeseries.
    {
        let catalog = state.credentials.catalog();
        match catalog.get_collection(DatabaseId::DEFAULT, tenant_id.as_u64(), &name) {
            Ok(Some(coll)) if coll.collection_type.is_timeseries() => {}
            Ok(Some(_)) => {
                return Err(ddl_err(
                    "42809",
                    format!("'{name}' is not a timeseries collection"),
                ));
            }
            _ => {
                return Err(ddl_err(
                    "42P01",
                    format!("collection '{name}' does not exist"),
                ));
            }
        }
    }

    // Collect partition directories to rewrite.
    let partitions_to_rewrite: Vec<String> = if let Some(registries) = state.timeseries_registries()
    {
        let key = format!("{}:{}", tenant_id.as_u64(), name);
        let regs = crate::control::lock_utils::lock_or_recover(registries.lock(), "ts_registries");
        if let Some(registry) = regs.get(&key) {
            registry
                .iter()
                .filter(|(_, e)| {
                    e.meta.state == nodedb_types::timeseries::PartitionState::Sealed
                        || e.meta.state == nodedb_types::timeseries::PartitionState::Merged
                })
                .map(|(_, e)| e.dir_name.clone())
                .collect()
        } else {
            Vec::new()
        }
    } else {
        Vec::new()
    };

    let sealed_count = partitions_to_rewrite.len();
    tracing::info!(
        collection = name,
        sealed_partitions = sealed_count,
        "REWRITE PARTITIONS scheduled (async, non-blocking)"
    );

    if sealed_count > 0 {
        let wal_dir = state.wal.wal_dir();
        let ts_base = wal_dir
            .parent()
            .unwrap_or(wal_dir)
            .join("timeseries")
            .to_path_buf();
        let collection_name = name.clone();

        tokio::task::spawn_blocking(move || {
            let mut rewritten = 0usize;
            for dir_name in &partitions_to_rewrite {
                let partition_dir = ts_base.join(dir_name);
                if !partition_dir.exists() {
                    continue;
                }
                match crate::engine::timeseries::merge::merge_partitions(
                    &ts_base,
                    std::slice::from_ref(&partition_dir),
                    &format!("{dir_name}.rewrite"),
                ) {
                    Ok(result) => {
                        let rewrite_dir = ts_base.join(format!("{dir_name}.rewrite"));
                        let backup_dir = ts_base.join(format!("{dir_name}.old"));
                        if nodedb_wal::segment::atomic_swap_dirs_fsync(
                            &partition_dir,
                            &backup_dir,
                            &rewrite_dir,
                        )
                        .is_ok()
                        {
                            let _ = std::fs::remove_dir_all(&backup_dir);
                            // Write updated metadata to partition.meta (on-disk source of truth).
                            let meta_path = partition_dir.join("partition.meta");
                            let meta_tmp = partition_dir.join("partition.meta.tmp");
                            let meta_bytes =
                                sonic_rs::to_vec_pretty(&result.meta).unwrap_or_default();
                            let _ = nodedb_wal::segment::atomic_write_fsync(
                                &meta_tmp,
                                &meta_path,
                                &meta_bytes,
                            );
                            rewritten += 1;
                        }
                    }
                    Err(e) => {
                        tracing::warn!(
                            partition = dir_name,
                            error = %e,
                            "rewrite failed for partition"
                        );
                    }
                }
            }
            tracing::info!(
                collection = collection_name,
                rewritten,
                total = sealed_count,
                "REWRITE PARTITIONS completed"
            );
        });
    }

    Ok(vec![DdlResult::Status {
        command: "REWRITE PARTITIONS".to_string(),
        rows_affected: None,
    }])
}