cobble_table/
write.rs

1use crate::catalog::{CatalogResult, CatalogSchemaId, CatalogTable, TableId, TableIdentifier};
2use crate::{Result, TableError, TableSchema, TableWriterBuilder};
3use cobble::{Config, VolumeDescriptor, VolumeUsageKind};
4use serde::{Deserialize, Serialize};
5
6pub(crate) const TABLE_WRITE_PLAN_FORMAT: &str = "cobble-table-write-plan";
7pub(crate) const TABLE_WRITE_PLAN_VERSION: u32 = 1;
8
9/// Builds a portable writer initialization plan from one catalog table definition.
10pub struct TableWriteBuilder {
11    table: CatalogTable,
12    total_buckets: u32,
13}
14
15impl TableWriteBuilder {
16    pub(crate) fn new(table: CatalogTable, total_buckets: u32) -> Self {
17        Self {
18            table,
19            total_buckets,
20        }
21    }
22
23    /// Set the table-wide bucket count captured in the plan.
24    pub fn total_buckets(mut self, total_buckets: u32) -> Self {
25        self.total_buckets = total_buckets;
26        self
27    }
28
29    /// Freeze this table definition and shared-storage locations for worker use.
30    pub fn build(self) -> CatalogResult<TableWritePlan> {
31        crate::catalog::build_write_plan(self.table, self.total_buckets)
32    }
33}
34
35/// Serializable, catalog-independent initialization for one table writer shard.
36///
37/// The plan carries only shared metadata, snapshot, and WAL locations. Workers supply their own
38/// primary-data, cache, and read-only runtime volumes when creating a writer.
39#[derive(Clone, Serialize, Deserialize)]
40pub struct TableWritePlan {
41    pub(crate) format: String,
42    pub(crate) version: u32,
43    pub(crate) identifier: TableIdentifier,
44    pub(crate) table_id: TableId,
45    pub(crate) catalog_schema_id: CatalogSchemaId,
46    pub(crate) schema: TableSchema,
47    pub(crate) storage_id: String,
48    pub(crate) shared_volumes: Vec<VolumeDescriptor>,
49    pub(crate) total_buckets: u32,
50    #[serde(skip)]
51    pub(crate) auth_source: Option<Config>,
52}
53
54impl TableWritePlan {
55    pub(crate) fn validate(&self) -> Result<()> {
56        if self.format != TABLE_WRITE_PLAN_FORMAT {
57            return Err(TableError::InvalidSchema(format!(
58                "unsupported table write plan format: {}",
59                self.format
60            )));
61        }
62        if self.version != TABLE_WRITE_PLAN_VERSION {
63            return Err(TableError::InvalidSchema(format!(
64                "unsupported table write plan version: {}",
65                self.version
66            )));
67        }
68        if self.identifier.namespace().is_empty()
69            || self.identifier.name().is_empty()
70            || self.identifier.name() != self.identifier.name().trim()
71        {
72            return Err(TableError::InvalidSchema(
73                "table write plan has an invalid identifier".to_string(),
74            ));
75        }
76        self.schema.validate()?;
77        if self.storage_id.is_empty()
78            || self.storage_id == "."
79            || self.storage_id == ".."
80            || self.storage_id.contains(['/', '\\'])
81        {
82            return Err(TableError::InvalidSchema(
83                "table write plan has an invalid storage id".to_string(),
84            ));
85        }
86        if !(1..=u16::MAX as u32 + 1).contains(&self.total_buckets) {
87            return Err(TableError::InvalidSchema(
88                "table write plan total_buckets must be in range 1..=65536".to_string(),
89            ));
90        }
91        if self.shared_volumes.is_empty()
92            || !self
93                .shared_volumes
94                .iter()
95                .any(|volume| volume.supports(VolumeUsageKind::Meta))
96            || self.shared_volumes.iter().any(|volume| {
97                volume.supports(VolumeUsageKind::PrimaryDataPriorityHigh)
98                    || volume.supports(VolumeUsageKind::PrimaryDataPriorityMedium)
99                    || volume.supports(VolumeUsageKind::PrimaryDataPriorityLow)
100                    || volume.supports(VolumeUsageKind::Cache)
101                    || volume.supports(VolumeUsageKind::Readonly)
102            })
103        {
104            return Err(TableError::InvalidSchema(
105                "table write plan has invalid shared volumes".to_string(),
106            ));
107        }
108        Ok(())
109    }
110
111    pub(crate) fn table_id(&self) -> TableId {
112        self.table_id
113    }
114
115    /// Initialize a shard writer builder using this fixed table definition.
116    pub fn writer_builder(&self, runtime: Config) -> CatalogResult<TableWriterBuilder> {
117        crate::catalog::writer_builder_from_write_plan(self, runtime)
118    }
119}
cobble_table/write.rs

cobble_table/
write.rs
