Feat/sample overwrite ipv4

crates/big-cluster/src/ddl.rs

@@ -108,6 +108,15 @@ impl<P: PagerMut + Sync> Cluster<P> {
         })
     }
 
+    /// A mutex field that also declares which values it may hold.
+    ///
+    /// Travels the same leader-then-fan-out path every schema change takes, and it has to: a
+    /// node that missed the member list would accept a write the others refuse, which is the
+    /// kind of disagreement that shows up as a record existing on some nodes and not others.
+    pub fn create_enum(&self, table: &str, field: &str, members: Vec<String>) -> Result<u64> {
+        self.ddl(&Ddl::CreateEnum { table: table.to_string(), field: field.to_string(), members })
+    }
+
     pub fn create_database(&self, name: &str) -> Result<bool> {
         self.ddl(&Ddl::CreateDatabase { name: name.to_string() }).map(|n| n == 1)
     }
@@ -327,6 +336,9 @@ pub fn apply_ddl<P: PagerMut + Sync>(api: &Api<P>, op: &Ddl) -> big_embed::Resul
         Ddl::CreateTimeQuantum { table, field, granularity } => {
             api.create_time_quantum(table, field, granularity.clone())? as u64
         }
+        Ddl::CreateEnum { table, field, members } => {
+            api.create_enum(table, field, members.clone())? as u64
+        }
         Ddl::DropTable { table } => api.drop_table(table)? as u64,
         // A table that is not here counts as nothing truncated rather than as a failure: the
         // leader has already ruled on whether it exists, so a peer without it is a peer that

crates/big-cluster/src/query.rs

@@ -327,6 +327,11 @@ impl<P: PagerMut + Sync> Cluster<P> {
             big_embed::FieldKind::TimeQuantum => {
                 self.create_time_quantum(table, &column.name, Vec::new())
             }
+            // Decided on the declaration rather than on the kind, because an enum and a plain
+            // mutex are one `FieldKind` and only the member list tells them apart.
+            _ if !column.members.is_empty() => {
+                self.create_enum(table, &column.name, column.members.clone())
+            }
             _ => self.create_field(table, &column.name, kind, column.bit_depth),
         }
     }
@@ -936,6 +941,20 @@ impl<P: PagerMut + Sync> Cluster<P> {
             Some(select) => self.read_source(select, &name, opts)?,
         };
 
+        // **`OVERWRITE` empties here, and the position is the whole of what it promises.**
+        // After the source query has been read whole and merged, so a `SELECT` that turns out
+        // to be unreadable leaves the table alone; before a single fact is written, so what
+        // lands is these rows and nothing older. It is still two operations with no transaction
+        // around them - a crash in between leaves the table empty, which `Insert::overwrite`
+        // says out loud rather than hiding. The atomic replacement is `EXCHANGE TABLES`.
+        //
+        // Row ids are allocated *after* this, because a truncate keeps the row keys but frees
+        // the fragments: allocating first would hand out ids against a record count that is
+        // about to become zero.
+        if insert.overwrite {
+            self.sql_truncate_table(&name, false)?;
+        }
+
         // Asked for once for the whole statement rather than once per row: a round trip per
         // row would make a thousand-row insert a thousand round trips to one node.
         let allocated = match insert.id_at {

crates/big-cluster/src/wire/fact.rs

@@ -210,6 +210,16 @@ pub enum Ddl {
         field: String,
         granularity: Vec<Granularity>,
     },
+    /// A `MUTEX` field that also declares which values it may hold.
+    ///
+    /// Its own variant rather than a `CreateField` with a list bolted on, for the reason
+    /// [`Ddl::CreateDecimal`] is one: the extra thing it carries is not optional, and a shape
+    /// where it might be absent is a shape where a peer can create the field without it.
+    CreateEnum {
+        table: String,
+        field: String,
+        members: Vec<String>,
+    },
     DropTable {
         table: String,
     },
@@ -346,6 +356,15 @@ impl Ddl {
                 put_u8(&mut out, 13);
                 put_str(&mut out, table);
             }
+            Self::CreateEnum { table, field, members } => {
+                put_u8(&mut out, 17);
+                put_str(&mut out, table);
+                put_str(&mut out, field);
+                put_count(&mut out, members.len());
+                for m in members {
+                    put_str(&mut out, m);
+                }
+            }
             Self::RenameTable { table, to } => {
                 put_u8(&mut out, 15);
                 put_str(&mut out, table);
@@ -452,6 +471,18 @@ impl Ddl {
             10 => Self::CreateRole { role: r.str()? },
             11 => Self::DropRole { role: r.str()? },
             13 => Self::TruncateTable { table: r.str()? },
+            17 => Self::CreateEnum {
+                table: r.str()?,
+                field: r.str()?,
+                members: {
+                    let n = r.count()?;
+                    let mut out = Vec::with_capacity(n.min(1024));
+                    for _ in 0..n {
+                        out.push(r.str()?);
+                    }
+                    out
+                },
+            },
             15 => Self::RenameTable { table: r.str()?, to: r.str()? },
             16 => Self::ExchangeTables { a: r.str()?, b: r.str()? },
             14 => Self::DropViewsBefore {

crates/big-cluster/src/wire/plan.rs

@@ -63,6 +63,7 @@ mod rows_tag {
     pub const ALL: u8 = 9;
     pub const COMPARE_FLOAT: u8 = 10;
     pub const KEY_LIKE: u8 = 11;
+    pub const SAMPLE: u8 = 12;
 }
 
 pub fn put_rows(out: &mut Vec<u8>, rows: &Rows) {
@@ -123,6 +124,11 @@ pub fn put_rows(out: &mut Vec<u8>, rows: &Rows) {
             put_rows(out, inner);
         }
         Rows::All => put_u8(out, rows_tag::ALL),
+        Rows::Sample { inner, stride } => {
+            put_u8(out, rows_tag::SAMPLE);
+            put_u32(out, *stride);
+            put_rows(out, inner);
+        }
     }
 }
 
@@ -168,6 +174,12 @@ fn get_rows_at(r: &mut Reader<'_>, depth: usize) -> Result<Rows> {
         ),
         rows_tag::NOT => Rows::Not(Box::new(get_rows_at(r, depth + 1)?)),
         rows_tag::ALL => Rows::All,
+        rows_tag::SAMPLE => {
+            // The stride before the inner set, so it is read before the recursion rather than
+            // after it - the same order it was written in.
+            let stride = r.u32()?;
+            Rows::Sample { inner: Box::new(get_rows_at(r, depth + 1)?), stride }
+        }
         tag => return Err(WireError::BadTag { what: "row set", tag }),
     })
 }

crates/big-cluster/src/wire/request.rs

@@ -637,6 +637,13 @@ pub fn put_schema(
             for g in &field.granularity {
                 put_u8(out, g.as_char() as u8);
             }
+            // The declared enum members, which a repair needs for the same reason it needs the
+            // scale: without them the field comes back as the mutex it is stored as, and the
+            // check on the way in would be gone.
+            put_count(out, field.members.len());
+            for m in &field.members {
+                put_str(out, m);
+            }
         }
     }
     put_count(out, views.len());
@@ -675,7 +682,19 @@ pub fn get_schema(bytes: &[u8]) -> Result<(Vec<big_embed::TableInfo>, Vec<big_em
                     granularity_of(c).ok_or(WireError::BadTag { what: "granularity", tag: c })?,
                 );
             }
-            fields.push(big_embed::FieldInfo { name: field, kind, bit_depth, scale, granularity });
+            let n = r.count()?;
+            let mut members = reserve(n);
+            for _ in 0..n {
+                members.push(r.str()?);
+            }
+            fields.push(big_embed::FieldInfo {
+                name: field,
+                kind,
+                bit_depth,
+                scale,
+                granularity,
+                members,
+            });
         }
         let r = big_db::TableRef::parse(&name);
         tables.push(big_embed::TableInfo {

crates/big-cluster/tests/logic/insert_select.test

@@ -206,3 +206,54 @@ error
 INSERT INTO floaty2 (temp) SELECT temp FROM floaty
 ----
 error: engine_cannot_answer
+
+## `INSERT OVERWRITE`: the rows that were there are gone, and only these are left.
+#
+# The claim worth running against real records is the *order*. The source is read whole and
+# merged before anything is cleared, so the count below is the source's count and not the sum of
+# both — and a table that was not empty beforehand does not contribute a single record.
+
+statement
+CREATE TABLE ow_src (amount UINT(32))
+
+statement
+INSERT INTO ow_src (_record_id, amount) VALUES (1, 7), (2, 8)
+
+statement
+CREATE TABLE ow_dst (amount UINT(32))
+
+statement
+INSERT INTO ow_dst (_record_id, amount) VALUES (1, 100), (2, 200), (3, 300)
+
+query
+SELECT count(*), sum(amount) FROM ow_dst
+----
+count  sum
+3      600
+
+statement
+INSERT OVERWRITE ow_dst (amount) SELECT amount FROM ow_src
+
+# Three rows went, two arrived. A statement that wrote without emptying would say 5 and 615.
+query
+SELECT count(*), sum(amount) FROM ow_dst
+----
+count  sum
+2      15
+
+# The source is untouched.
+query
+SELECT count(*) FROM ow_src
+----
+count
+2
+
+# Overwriting with literals empties just as much.
+statement
+INSERT OVERWRITE TABLE ow_dst (amount) VALUES (1)
+
+query
+SELECT count(*), sum(amount) FROM ow_dst
+----
+count  sum
+1      1

crates/big-cluster/tests/logic/records.test

@@ -142,3 +142,66 @@ SHOW TABLES FORMAT TSV
 ----
 name  type        engine           fields
 tx    BASE TABLE  bitmap+columnar  4
+
+## `SAMPLE`: one record in every n, by record id — against real records.
+#
+# The claim these cases pin is that the sample is on the *id* and not on a block of ids, which
+# is only visible with ids whose values are known. Ten records, ids 1..10: one in two takes the
+# even ids, one in five takes 5 and 10. A sample that took a block would take 1..5 instead, and
+# `sum` is what tells the two apart — a block would sum to 15 where an id sample sums to 30.
+
+statement
+CREATE TABLE smp (amount UINT(32))
+
+statement
+INSERT INTO smp (_record_id, amount) VALUES
+  (1, 1), (2, 2), (3, 3), (4, 4), (5, 5),
+  (6, 6), (7, 7), (8, 8), (9, 9), (10, 10)
+
+query
+SELECT count(*), sum(amount) FROM smp
+----
+count  sum
+10     55
+
+# Every second id: 2, 4, 6, 8, 10.
+query
+SELECT count(*), sum(amount) FROM smp SAMPLE 2
+----
+count  sum
+5      30
+
+# Every fifth: 5 and 10. A block of two would have been 1 and 2.
+query
+SELECT count(*), sum(amount) FROM smp SAMPLE 5
+----
+count  sum
+2      15
+
+# The fraction and the ratio are the same stride as the integer.
+query
+SELECT count(*) FROM smp SAMPLE 0.5
+----
+count
+5
+
+query
+SELECT count(*) FROM smp SAMPLE 1/5
+----
+count
+2
+
+# `SAMPLE` wraps the filter, so this is half of the records over five rather than half of every
+# record of which some are over five: ids 6..10 are over five, and 6, 8, 10 are even.
+query
+SELECT count(*), sum(amount) FROM smp SAMPLE 2 WHERE amount > 5
+----
+count  sum
+3      24
+
+# A stride past the record count is not an error, it is a small answer.
+query
+SELECT count(*) FROM smp SAMPLE 64
+----
+count
+0

crates/big-cluster/tests/logic/schema.test

@@ -445,3 +445,60 @@ error
 EXCHANGE TABLES live AND typo
 ----
 error: unknown_table
+
+## `ENUM`: the declaration is enforced, and it survives.
+#
+# Two claims, and neither is visible in translation. First that a value outside the list is
+# refused on the way in — which is the whole of what declaring the members buys, because the
+# mutex underneath would take any string. Second that `SHOW CREATE TABLE` says `ENUM(...)` and
+# not `MUTEX`: the members live in the catalog beside the field, so the declared type is still
+# there to print.
+
+statement
+CREATE TABLE orders (status ENUM('new', 'paid', 'shipped'), amount UINT(32))
+
+statement
+INSERT INTO orders (_record_id, status, amount) VALUES (1, 'new', 10), (2, 'paid', 20)
+
+query
+SELECT count(*) FROM orders WHERE status = 'paid'
+----
+count
+1
+
+# A value the column was not declared to hold. The mutex underneath would have taken it.
+error
+INSERT INTO orders (_record_id, status, amount) VALUES (3, 'refunded', 30)
+----
+error: malformed_line
+
+# And nothing was written: the statement failed before the first fact went out.
+query
+SELECT count(*) FROM orders
+----
+count
+2
+
+# The declared type comes back, not the storage one.
+query
+SHOW CREATE TABLE orders
+----
+statement
+CREATE TABLE orders (
+    status ENUM('new', 'paid', 'shipped'),
+    amount UINT(32)
+) ENGINE = 'bitmap+columnar'
+
+# A plain mutex still takes anything, which is what says the check belongs to the declaration
+# rather than to the kind.
+statement
+CREATE TABLE loose (status MUTEX)
+
+statement
+INSERT INTO loose (_record_id, status) VALUES (1, 'anything at all')
+
+query
+SELECT count(*) FROM loose WHERE status = 'anything at all'
+----
+count
+1