diff --git a/core/src/main/resources/help/commands/ddl/alter_table.json b/core/src/main/resources/help/commands/ddl/alter_table.json
index d1200a3a..e2f41d33 100644
--- a/core/src/main/resources/help/commands/ddl/alter_table.json
+++ b/core/src/main/resources/help/commands/ddl/alter_table.json
@@ -23,7 +23,7 @@
     "ALTER COLUMN [IF EXISTS] column_name DROP NOT NULL",
     "ALTER COLUMN [IF EXISTS] column_name SET COMMENT 'text'",
     "ALTER COLUMN [IF EXISTS] column_name DROP COMMENT",
-    "ALTER COLUMN [IF EXISTS] column_name SET TYPE new_type",
+    "ALTER COLUMN [IF EXISTS] column_name SET DATA TYPE new_type",
     "ALTER COLUMN [IF EXISTS] column_name SET SCRIPT AS (expression)",
     "ALTER COLUMN [IF EXISTS] column_name DROP SCRIPT",
     "ALTER COLUMN [IF EXISTS] column_name SET OPTIONS (option = value, ...)",
@@ -87,7 +87,7 @@
       "optional": false
     },
     {
-      "name": "SET TYPE",
+      "name": "SET DATA TYPE",
       "description": "Change column data type (requires reindex)",
       "optional": false
     },
diff --git a/documentation/sql/ddl_statements.md b/documentation/sql/ddl_statements.md
index 2dd33b95..865f727f 100644
--- a/documentation/sql/ddl_statements.md
+++ b/documentation/sql/ddl_statements.md
@@ -307,6 +307,32 @@ CREATE TABLE users (
 PARTITIONED BY (birthdate MONTH);
 ```
 
+### Table Options (index settings)
+
+Index settings are declared with a table-level `OPTIONS (…)` clause after the column list:
+
+```sql
+CREATE TABLE users (
+  id INT,
+  name VARCHAR,
+  PRIMARY KEY (id)
+) OPTIONS (number_of_shards = 1, number_of_replicas = 0);
+```
+
+Any index setting can be passed — for example `default_pipeline` to route inserts through a
+custom ingest pipeline (see [Using Enrich Policies in Pipelines](#using-enrich-policies-in-pipelines)):
+
+```sql
+CREATE TABLE events (
+  id INT,
+  user_id KEYWORD,
+  PRIMARY KEY (id)
+) OPTIONS (default_pipeline = 'events_enriched');
+```
+
+> ⚠️ `OPTIONS`, not `WITH` — `WITH (…)` belongs to `CREATE MATERIALIZED VIEW`. A
+> `CREATE TABLE … WITH (…)` is rejected by the parser (the whole statement must parse).
+
 ---
 
 ## CREATE TABLE AS SELECT
@@ -346,10 +372,32 @@ The gateway:
 - `ALTER COLUMN column_name SET|ADD FIELD field_definition`
 - `ALTER COLUMN column_name DROP FIELD field_name`
 - `ALTER COLUMN column_name SET FIELDS (...)`
-- `SET|ADD MAPPING (key = value)`
+- `SET|ADD MAPPING key = value`
 - `DROP MAPPING key`
-- `SET|ADD SETTING (key = value)`
+- `SET|ADD SETTING key = value`
 - `DROP SETTING key`
+- `SET|ADD ALIAS alias_name = value`
+- `DROP ALIAS alias_name`
+
+### Table-level clauses take no parentheses
+
+The table-level `MAPPING`, `SETTING` and `ALIAS` clauses are written as a bare `key = value`,
+**without** parentheses — unlike the column-level `ALTER COLUMN … SET|ADD OPTION (key = value)`:
+
+```sql
+ALTER TABLE orders (
+  SET MAPPING _meta.owner = 'analytics',
+  SET SETTING index.refresh_interval = '1s',
+  SET ALIAS recent_orders = (filter = (range = (order_date = (gte = 'now-7d'))))
+);
+```
+
+The parentheses around the statement list are optional and independent of any single clause —
+`ALTER TABLE orders SET SETTING index.refresh_interval = '1s'` is equally valid.
+
+A value may be a scalar (`'1s'`, `true`, `2`), an array (`['a', 'b']`), or a nested object
+written with **parentheses**: `(key = value, key = (nested = value))`. The `{…}` brace form is
+for `STRUCT` column values in `INSERT`, and is not accepted here.
 
 ### Type Changes and Safety
 
@@ -1391,7 +1439,7 @@ CREATE TABLE events (
   event_type KEYWORD,
   timestamp TIMESTAMP,
   PRIMARY KEY (id)
-) WITH (default_pipeline = "events_enriched");
+) OPTIONS (default_pipeline = 'events_enriched');
 ```
 
 #### 6. Insert data
diff --git a/sql/src/main/scala/app/softnetwork/elastic/sql/package.scala b/sql/src/main/scala/app/softnetwork/elastic/sql/package.scala
index d539e3d3..a89950f7 100644
--- a/sql/src/main/scala/app/softnetwork/elastic/sql/package.scala
+++ b/sql/src/main/scala/app/softnetwork/elastic/sql/package.scala
@@ -55,6 +55,14 @@ package object sql {
     case _       => ""
   }
 
+  /** Escapes a bare string for a single-quoted SQL literal, exactly reversing how the grammar's
+    * `'([^'\\]|\\.)*'` unescapes it — backslash first, then the quote. Renderers that hold a plain
+    * `String` rather than a `StringValue` (a column COMMENT, for one) need it too, which is why it
+    * lives here rather than on the value.
+    */
+  def escapeStringLiteral(value: String): String =
+    value.replace("\\", "\\\\").replace("'", "\\'")
+
   /** Base trait for all tokens
     */
   trait Token extends Serializable with Validation {
@@ -438,42 +446,37 @@ package object sql {
       }
       .mkString("(", ", ", ")")
 
+    /** Recursion is on the HEAD of the path, one level at a time. Descending to the LEAF's parent
+      * and then re-attaching it under `keys.head` collapsed the intermediate levels: for a path of
+      * depth 3 or more, `_meta.columns.<c>.default_value` replaced the whole of `_meta` with the
+      * innermost object, dropping every sibling key. Depth 1 and 2 happened to be correct, which is
+      * why it survived — the metadata paths the schema writes are exactly the deeper ones.
+      */
     def set(path: String, newValue: Value[_]): ObjectValue = {
-      val keys = path.split("\\.")
-      val updatedValue = {
-        if (keys.length == 1) {
-          value + (keys.head -> newValue)
-        } else {
-          val parentPath = keys.dropRight(1).mkString(".")
-          val parentKey = keys.last
-          val parentObject = find(parentPath) match {
+      path.split("\\.").toList match {
+        case Nil      => this
+        case k :: Nil => ObjectValue(value + (k -> newValue))
+        case k :: rest =>
+          val child = value.get(k) match {
             case Some(obj: ObjectValue) => obj
             case _                      => ObjectValue.empty
           }
-          val updatedParent = parentObject.set(parentKey, newValue)
-          value + (keys.head -> updatedParent)
-        }
+          ObjectValue(value + (k -> child.set(rest.mkString("."), newValue)))
       }
-      ObjectValue(updatedValue)
     }
 
     def remove(path: String): ObjectValue = {
-      val keys = path.split("\\.")
-      val updatedValue = {
-        if (keys.length == 1) {
-          value - keys.head
-        } else {
-          val parentPath = keys.dropRight(1).mkString(".")
-          val parentKey = keys.last
-          val parentObject = find(parentPath) match {
-            case Some(obj: ObjectValue) => obj
-            case _                      => ObjectValue.empty
+      path.split("\\.").toList match {
+        case Nil      => this
+        case k :: Nil => ObjectValue(value - k)
+        case k :: rest =>
+          value.get(k) match {
+            case Some(obj: ObjectValue) =>
+              ObjectValue(value + (k -> obj.remove(rest.mkString("."))))
+            // nothing at that path: removing is a no-op, NOT a reason to overwrite `k`
+            case _ => this
           }
-          val updatedParent = parentObject.remove(parentKey)
-          value + (keys.head -> updatedParent)
-        }
       }
-      ObjectValue(updatedValue)
     }
 
     def find(path: String): Option[Value[_]] = {
@@ -575,7 +578,10 @@ package object sql {
   }
 
   case class StringValue(override val value: String) extends Value[String](value) {
-    override def sql: String = s"""'$value'"""
+    // Escaped exactly as the grammar's literal (`'([^'\\]|\\.)*'`) unescapes it — backslash first,
+    // then the quote, so the composition reverses. Without this a value holding an apostrophe
+    // rendered `'it's'`, which no longer parses.
+    override def sql: String = s"""'${escapeStringLiteral(value)}'"""
     override def baseType: SQLType = SQLTypes.Varchar
 
     override def ddl: String = s""""${value.replace("\\", "\\\\").replace("\"", "\\\"")}""""
diff --git a/sql/src/main/scala/app/softnetwork/elastic/sql/parser/Parser.scala b/sql/src/main/scala/app/softnetwork/elastic/sql/parser/Parser.scala
index c61304d7..f5e91847 100644
--- a/sql/src/main/scala/app/softnetwork/elastic/sql/parser/Parser.scala
+++ b/sql/src/main/scala/app/softnetwork/elastic/sql/parser/Parser.scala
@@ -188,19 +188,29 @@ object Parser
     (keyword("ALTER") ~ keyword("PIPELINE")) ~ ifExists ~ ident ~ start.? ~ repsep(
       alterPipelineStatement,
       separator
-    ) ~ end.? ^^ { case _ ~ ie ~ pipeline ~ s ~ stmts ~ e =>
+    ) ~ end.? >> { case _ ~ ie ~ pipeline ~ s ~ stmts ~ e =>
       if (s.isDefined && e.isEmpty) {
-        throw new Exception("Mismatched closing parentheses in ALTER PIPELINE statement")
+        err("Mismatched closing parentheses in ALTER PIPELINE statement")
       } else if (s.isEmpty && e.isDefined) {
-        throw new Exception("Mismatched opening parentheses in ALTER PIPELINE statement")
+        err("Mismatched opening parentheses in ALTER PIPELINE statement")
       } else if (s.isEmpty && e.isEmpty && stmts.size > 1) {
-        throw new Exception("Multiple ALTER PIPELINE statements require parentheses")
+        err("Multiple ALTER PIPELINE statements require parentheses")
       } else
-        AlterPipeline(pipeline, ie, stmts)
+        success(AlterPipeline(pipeline, ie, stmts))
     }
 
+  /** `FIELDS (…)` — required. The empty fallback belongs to `optionalMultiFields`, whose only
+    * caller is a column definition (a column need not declare sub-fields). Folding it in here made
+    * every consumer optional, and `alterColumnFields` (`ALTER COLUMN c SET <multiFields>`) then
+    * matched a bare `SET` with an empty field list: `ALTER COLUMN c SET` parsed as a no-op, and
+    * `ALTER COLUMN c SET FIELD raw KEYWORD` matched that same alternative and left `FIELD raw
+    * KEYWORD` unconsumed — so `SET FIELD` silently did nothing before #213 made trailing input an
+    * error, and could not parse at all afterwards.
+    */
   def multiFields: PackratParser[List[Column]] =
-    keyword("FIELDS") ~ start ~> repsep(column, separator) <~ end ^^ (cols => cols) | success(Nil)
+    keyword("FIELDS") ~ start ~> repsep(column, separator) <~ end ^^ (cols => cols)
+
+  def optionalMultiFields: PackratParser[List[Column]] = multiFields | success(Nil)
 
   def ifExists: PackratParser[Boolean] =
     opt(keyword("IF") ~ keyword("EXISTS")) ^^ {
@@ -220,10 +230,6 @@ object Parser
       case None    => false
     }
 
-  def ingest_id: PackratParser[Value[_]] = "_id" ^^ (_ => IdValue)
-
-  def ingest_timestamp: PackratParser[Value[_]] = "_ingest.timestamp" ^^ (_ => IngestTimestampValue)
-
   def defaultVal: PackratParser[Option[Value[_]]] =
     opt(keyword("DEFAULT") ~ (value | ingest_id | ingest_timestamp)) ^^ {
       case Some(_ ~ v) => Some(v)
@@ -245,7 +251,7 @@ object Parser
     (keyword("SCRIPT") ~ keyword("AS")) ~ start ~ scriptValue ~ end ^^ { case _ ~ _ ~ s ~ _ => s }
 
   def column: PackratParser[Column] =
-    ident ~ extension_type ~ (script | multiFields) ~ defaultVal ~ notNull ~ comment ~ (options | success(
+    ident ~ extension_type ~ (script | optionalMultiFields) ~ defaultVal ~ notNull ~ comment ~ (options | success(
       ListMap.empty[String, Value[_]]
     )) ^^ { case name ~ dt ~ mfs ~ dv ~ nn ~ ct ~ opts =>
       mfs match {
@@ -543,10 +549,17 @@ object Parser
       DropColumnScript(name, ifExists = ie)
     }
 
+  /** The value grammar must match `defaultVal`'s: a column declares `DEFAULT _ingest.timestamp` at
+    * CREATE time, so an ALTER that sets the same default on an existing column has to accept it
+    * too. It did not, which broke the `_last_updated` column the materialized-view machinery adds
+    * whenever that column already exists (`TableDiff` renders `ColumnDefaultSet` and the extension
+    * runs the rendered SQL).
+    */
   def alterColumnDefault: PackratParser[AlterColumnDefault] =
-    alterColumnIfExists ~ ident ~ (keyword("SET") ~ keyword("DEFAULT")) ~ value ^^ {
-      case ie ~ name ~ _ ~ dv =>
-        AlterColumnDefault(name, dv, ifExists = ie)
+    alterColumnIfExists ~ ident ~ (keyword("SET") ~ keyword(
+      "DEFAULT"
+    )) ~ (value | ingest_id | ingest_timestamp) ^^ { case ie ~ name ~ _ ~ dv =>
+      AlterColumnDefault(name, dv, ifExists = ie)
     }
 
   def dropColumnDefault: PackratParser[DropColumnDefault] =
@@ -631,15 +644,18 @@ object Parser
     (keyword("ALTER") ~ keyword("TABLE")) ~ ifExists ~ ident ~ start.? ~ repsep(
       alterTableStatement,
       separator
-    ) ~ end.? ^^ { case _ ~ ie ~ table ~ s ~ stmts ~ e =>
+    ) ~ end.? >> { case _ ~ ie ~ table ~ s ~ stmts ~ e =>
+      // `err`, not `throw`: these run inside a combinator, and `Parser.apply` is typed
+      // `Either[ParserError, Statement]` — a raw exception escapes that signature and only
+      // `GatewayApi` happens to wrap the call in `ElasticResult.attempt`.
       if (s.isDefined && e.isEmpty) {
-        throw new Exception("Mismatched closing parentheses in ALTER TABLE statement")
+        err("Mismatched closing parentheses in ALTER TABLE statement")
       } else if (s.isEmpty && e.isDefined) {
-        throw new Exception("Mismatched opening parentheses in ALTER TABLE statement")
+        err("Mismatched opening parentheses in ALTER TABLE statement")
       } else if (s.isEmpty && e.isEmpty && stmts.size > 1) {
-        throw new Exception("Multiple ALTER TABLE statements require parentheses")
+        err("Multiple ALTER TABLE statements require parentheses")
       } else
-        AlterTable(table, ie, stmts)
+        success(AlterTable(table, ie, stmts))
     }
 
   // Watcher parsers
@@ -661,7 +677,7 @@ object Parser
   def compareWatcherCondition: PackratParser[CompareWatcherCondition] =
     keyword("WHEN") ~> opt(not) ~ ident ~ comparison_operator ~ opt(value) ~ opt(
       dateMathScript
-    ) ^^ { case n ~ field ~ op ~ v ~ fun =>
+    ) >> { case n ~ field ~ op ~ v ~ fun =>
       val target_op =
         n match {
           case Some(_) => op.not
@@ -669,23 +685,23 @@ object Parser
         }
       v match {
         case Some(value) =>
-          CompareWatcherCondition(field, target_op, Left(value))
+          success(CompareWatcherCondition(field, target_op, Left(value)))
         case None =>
           fun match {
             case Some(f) if f.identifier.dependencies.isEmpty =>
-              CompareWatcherCondition(
-                field,
-                target_op,
-                Right(f.identifier.withFunctions(f +: f.identifier.functions))
+              success(
+                CompareWatcherCondition(
+                  field,
+                  target_op,
+                  Right(f.identifier.withFunctions(f +: f.identifier.functions))
+                )
               )
             case Some(_) =>
-              throw new Exception(
-                s"Date/datetime functions with field dependencies are not supported for comparison"
+              err(
+                "Date/datetime functions with field dependencies are not supported for comparison"
               )
             case None =>
-              throw new Exception(
-                s"A value or a date/datetime function must be provided for comparison"
-              )
+              err("A value or a date/datetime function must be provided for comparison")
           }
       }
     }
@@ -865,13 +881,10 @@ object Parser
     }
 
   def watcherAction: PackratParser[(String, WatcherAction)] =
-    ident ~ opt(keyword("AS")) ~ (loggingAction | webhookAction) ^^ { case name ~ _ ~ wa =>
+    ident ~ opt(keyword("AS")) ~ (loggingAction | webhookAction) >> { case name ~ _ ~ wa =>
       wa match {
-        case Some(wa) => (name, wa)
-        case _ =>
-          throw new Exception(
-            s"Unsupported watcher action type in action '$name'"
-          )
+        case Some(wa) => success((name, wa))
+        case _        => err(s"Unsupported watcher action type in action '$name'")
       }
     }
 
@@ -1324,12 +1337,16 @@ trait Parser
       ObjectValues(ovs)
     }
 
+  // `ingest_id | ingest_timestamp` for the same reason as `alterColumnDefault`: the mapping
+  // metadata a column's DEFAULT is mirrored into (`_meta.columns.<c>.default_value`) is written
+  // through this production.
   def option: PackratParser[(String, Value[_])] =
-    (ident | literal) ~ "=" ~ (objectValues | objectValue | value) ^^ { case key ~ _ ~ value =>
-      key match {
-        case lit: StringValue => (lit.value, value)
-        case id: String       => (id, value)
-      }
+    (ident | literal) ~ "=" ~ (objectValues | objectValue | value | ingest_id | ingest_timestamp) ^^ {
+      case key ~ _ ~ value =>
+        key match {
+          case lit: StringValue => (lit.value, value)
+          case id: String       => (id, value)
+        }
     }
 
   def options: PackratParser[ListMap[String, Value[_]]] =
diff --git a/sql/src/main/scala/app/softnetwork/elastic/sql/parser/type/package.scala b/sql/src/main/scala/app/softnetwork/elastic/sql/parser/type/package.scala
index 564bf915..e8c2d25b 100644
--- a/sql/src/main/scala/app/softnetwork/elastic/sql/parser/type/package.scala
+++ b/sql/src/main/scala/app/softnetwork/elastic/sql/parser/type/package.scala
@@ -21,7 +21,9 @@ import app.softnetwork.elastic.sql.{
   BooleanValues,
   DoubleValue,
   DoubleValues,
+  IdValue,
   Identifier,
+  IngestTimestampValue,
   LongValue,
   LongValues,
   Null,
@@ -91,6 +93,16 @@ package object `type` {
     def value: PackratParser[Value[_]] =
       literal | pi | random | double | long | boolean | nullValue | param | array
 
+    /** The two ingest-time placeholders a column DEFAULT may carry. They live beside `value` so
+      * every value position can opt into them — `object Parser`'s `defaultVal` and `trait Parser`'s
+      * `option` both need them, and only the former could see them while they were declared in the
+      * object.
+      */
+    def ingest_id: PackratParser[Value[_]] = "_id" ^^ (_ => IdValue)
+
+    def ingest_timestamp: PackratParser[Value[_]] =
+      "_ingest.timestamp" ^^ (_ => IngestTimestampValue)
+
     def identifierWithValue: Parser[Identifier] = (value ^^ functionAsIdentifier) >> cast
 
     def char_type: PackratParser[SQLTypes.Char.type] =
diff --git a/sql/src/main/scala/app/softnetwork/elastic/sql/query/Where.scala b/sql/src/main/scala/app/softnetwork/elastic/sql/query/Where.scala
index 8f2d41e1..3b5f47dd 100644
--- a/sql/src/main/scala/app/softnetwork/elastic/sql/query/Where.scala
+++ b/sql/src/main/scala/app/softnetwork/elastic/sql/query/Where.scala
@@ -344,12 +344,16 @@ sealed trait Expression extends FunctionChain with ElasticFilter with Criteria {
     identifier.bucket.find(_.name == bucket.name) match {
       case Some(_) =>
         operator match {
+          // These feed the ES `terms` include/exclude and its regex form, so they need the value
+          // itself, never its SQL rendering: `.sql` carries the quote delimiters (which used to be
+          // stripped for EQ/NE but not for LIKE/RLIKE, embedding `'` in the regex) and, since
+          // string literals are escaped, would also carry `\\` for a value holding one backslash.
           case EQ =>
             if ((!not && maybeNot.isEmpty) || (not && maybeNot.isDefined))
               maybeValue match {
-                case Some(v: Value[_]) if v.sql.nonEmpty =>
+                case Some(v: Value[_]) if v.value.toString.nonEmpty =>
                   bucketIncludesExcludes.copy(values =
-                    bucketIncludesExcludes.values ++ Set(v.sql.replaceAll("'", ""))
+                    bucketIncludesExcludes.values ++ Set(v.value.toString)
                   )
                 case _ => bucketIncludesExcludes
               }
@@ -357,9 +361,9 @@ sealed trait Expression extends FunctionChain with ElasticFilter with Criteria {
           case NE | DIFF =>
             if ((not && maybeNot.isEmpty) || (!not && maybeNot.isDefined))
               maybeValue match {
-                case Some(v: Value[_]) if v.sql.nonEmpty =>
+                case Some(v: Value[_]) if v.value.toString.nonEmpty =>
                   bucketIncludesExcludes.copy(values =
-                    bucketIncludesExcludes.values ++ Set(v.sql.replaceAll("'", ""))
+                    bucketIncludesExcludes.values ++ Set(v.value.toString)
                   )
                 case _ => bucketIncludesExcludes
               }
@@ -367,9 +371,9 @@ sealed trait Expression extends FunctionChain with ElasticFilter with Criteria {
           case LIKE =>
             if ((!not && maybeNot.isEmpty) || (not && maybeNot.isDefined))
               maybeValue match {
-                case Some(v: StringValue) if v.sql.nonEmpty =>
+                case Some(v: StringValue) if v.value.nonEmpty =>
                   bucketIncludesExcludes.copy(regex =
-                    bucketIncludesExcludes.regex.orElse(Option(v.sql.replaceAll("%", ".*")))
+                    bucketIncludesExcludes.regex.orElse(Option(v.value.replaceAll("%", ".*")))
                   )
                 case _ => bucketIncludesExcludes
               }
@@ -377,9 +381,9 @@ sealed trait Expression extends FunctionChain with ElasticFilter with Criteria {
           case RLIKE =>
             if ((!not && maybeNot.isEmpty) || (not && maybeNot.isDefined))
               maybeValue match {
-                case Some(v: StringValue) if v.sql.nonEmpty =>
+                case Some(v: StringValue) if v.value.nonEmpty =>
                   bucketIncludesExcludes.copy(regex =
-                    bucketIncludesExcludes.regex.orElse(Option(v.sql))
+                    bucketIncludesExcludes.regex.orElse(Option(v.value))
                   )
                 case _ => bucketIncludesExcludes
               }
@@ -892,8 +896,9 @@ case class MatchCriteria(
     identifier.bucket.find(_.name == bucket.name) match {
       case Some(_) =>
         if ((!not && maybeNot.isEmpty) || (not && maybeNot.isDefined))
+          // The value, not its rendering — see the include/exclude note above.
           bucketIncludesExcludes.copy(regex =
-            bucketIncludesExcludes.regex.orElse(Option(value.sql))
+            bucketIncludesExcludes.regex.orElse(Option(value.value.toString))
           )
         else bucketIncludesExcludes
       case _ => bucketIncludesExcludes
diff --git a/sql/src/main/scala/app/softnetwork/elastic/sql/query/package.scala b/sql/src/main/scala/app/softnetwork/elastic/sql/query/package.scala
index 8d4cadd6..674de21c 100644
--- a/sql/src/main/scala/app/softnetwork/elastic/sql/query/package.scala
+++ b/sql/src/main/scala/app/softnetwork/elastic/sql/query/package.scala
@@ -668,14 +668,14 @@ package object query {
   ) extends PipelineStatement
       with DdlStatement {
     override def sql: String = {
-      val ifExistsClause = if (ifExists) " IF EXISTS " else ""
+      val ifExistsClause = if (ifExists) " IF EXISTS" else ""
       val parenthesesNeeded = statements.size > 1
       val statementsSql = if (parenthesesNeeded) {
         statements.map(_.sql).mkString("(\n\t", ",\n\t", "\n)")
       } else {
         statements.map(_.sql).mkString("")
       }
-      s"ALTER PIPELINE $name$ifExistsClause $statementsSql"
+      s"ALTER PIPELINE$ifExistsClause $name $statementsSql"
     }
 
     lazy val ddlProcessors: Seq[IngestProcessor] = statements.flatMap(_.ddlProcessor)
@@ -963,14 +963,14 @@ package object query {
       extends TableStatement
       with DdlStatement {
     override def sql: String = {
-      val ifExistsClause = if (ifExists) " IF EXISTS " else ""
+      val ifExistsClause = if (ifExists) " IF EXISTS" else ""
       val parenthesesNeeded = statements.size > 1
       val statementsSql = if (parenthesesNeeded) {
         statements.map(_.sql).mkString("(\n\t", ",\n\t", "\n)")
       } else {
         statements.map(_.sql).mkString("")
       }
-      s"ALTER TABLE $table$ifExistsClause $statementsSql"
+      s"ALTER TABLE$ifExistsClause $table $statementsSql"
     }
 
     lazy val processors: Seq[IngestProcessor] = statements.flatMap(_.ddlProcessor)
@@ -1040,7 +1040,7 @@ package object query {
       extends AlterTableStatement {
     override def sql: String = {
       val ifExistsClause = if (ifExists) " IF EXISTS" else ""
-      s"ALTER COLUMN$ifExistsClause $columnName SET TYPE $newType"
+      s"ALTER COLUMN$ifExistsClause $columnName SET DATA TYPE $newType"
     }
   }
   case class AlterColumnScript(
@@ -1091,7 +1091,7 @@ package object query {
   ) extends AlterTableStatement {
     override def sql: String = {
       val ifExistsClause = if (ifExists) " IF EXISTS" else ""
-      s"ALTER COLUMN$ifExistsClause $columnName SET COMMENT '$comment'"
+      s"ALTER COLUMN$ifExistsClause $columnName SET COMMENT '${escapeStringLiteral(comment)}'"
     }
   }
   case class DropColumnComment(columnName: String, ifExists: Boolean = false)
@@ -1133,7 +1133,7 @@ package object query {
   ) extends AlterTableStatement {
     override def sql: String = {
       val ifExistsClause = if (ifExists) " IF EXISTS" else ""
-      s"ALTER COLUMN$ifExistsClause $columnName SET FIELD $field"
+      s"ALTER COLUMN$ifExistsClause $columnName SET FIELD ${field.sql}"
     }
   }
   case class DropColumnField(
diff --git a/sql/src/main/scala/app/softnetwork/elastic/sql/schema/package.scala b/sql/src/main/scala/app/softnetwork/elastic/sql/schema/package.scala
index 5e832247..00c07e1c 100644
--- a/sql/src/main/scala/app/softnetwork/elastic/sql/schema/package.scala
+++ b/sql/src/main/scala/app/softnetwork/elastic/sql/schema/package.scala
@@ -1013,7 +1013,7 @@ package object schema {
       }
       val defaultOpt = defaultValue.map(v => s" DEFAULT ${v.sql}").getOrElse("")
       val notNullOpt = if (notNull) " NOT NULL" else ""
-      val commentOpt = comment.map(c => s" COMMENT '$c'").getOrElse("")
+      val commentOpt = comment.map(c => s" COMMENT '${escapeStringLiteral(c)}'").getOrElse("")
       val fieldsOpt = if (multiFields.nonEmpty) {
         s" FIELDS (\n\t${multiFields.mkString(s",\n\t")}\n\t)"
       } else {
@@ -1577,7 +1577,7 @@ package object schema {
                 table.copy(columns = table.columns.filterNot(_.name == columnName))
               else throw ColumnNotFound(columnName, table.name)
             case RenameColumn(oldName, newName) =>
-              if (cols.contains(oldName))
+              if (table.cols.contains(oldName))
                 table.copy(
                   columns = table.columns.map { col =>
                     if (col.name == oldName) col.copy(name = newName) else col
@@ -1731,7 +1731,7 @@ package object schema {
               else throw ColumnNotFound(columnName, table.name)
             // multi-fields
             case AlterColumnFields(columnName, newFields, ifExists) =>
-              val col = find(columnName)
+              val col = table.find(columnName)
               val exists = col.isDefined
               if (ifExists && !exists) table
               else {
@@ -1755,15 +1755,25 @@ package object schema {
                   field,
                   ifExists
                 ) =>
-              val col = find(columnName)
+              val col = table.find(columnName)
               val exists = col.isDefined
               if (ifExists && !exists) table
               else {
                 col match {
                   case Some(c) =>
-                    val updatedFields = c.multiFields.filterNot(_.name == field.name) :+ field
-                    c.copy(multiFields = updatedFields)
-                    table
+                    // The copy is the whole point of the branch: assigning it to nothing and
+                    // returning `table` made SET|ADD FIELD a no-op that still reported success.
+                    val updated = c
+                      .copy(
+                        multiFields =
+                          c.multiFields.filterNot(_.name == field.name) :+ field.update(Some(c))
+                      )
+                      .updateStruct()
+                    table.copy(
+                      columns = table.columns.map { existing =>
+                        if (existing.name == updated.name) updated else existing
+                      }
+                    )
                   case _ => throw ColumnNotFound(columnName, table.name)
                 }
               }
@@ -1772,16 +1782,22 @@ package object schema {
                   fieldName,
                   ifExists
                 ) =>
-              val col = find(columnName)
+              val col = table.find(columnName)
               val exists = col.isDefined
               if (ifExists && !exists) table
               else {
                 col match {
                   case Some(c) =>
-                    c.copy(
-                      multiFields = c.multiFields.filterNot(_.name == fieldName)
+                    val updated = c
+                      .copy(
+                        multiFields = c.multiFields.filterNot(_.name == fieldName)
+                      )
+                      .updateStruct()
+                    table.copy(
+                      columns = table.columns.map { existing =>
+                        if (existing.name == updated.name) updated else existing
+                      }
                     )
-                    table
                   case _ => throw ColumnNotFound(columnName, table.name)
                 }
               }
diff --git a/sql/src/test/scala/app/softnetwork/elastic/sql/ObjectValuePathSpec.scala b/sql/src/test/scala/app/softnetwork/elastic/sql/ObjectValuePathSpec.scala
new file mode 100644
index 00000000..72caf7e1
--- /dev/null
+++ b/sql/src/test/scala/app/softnetwork/elastic/sql/ObjectValuePathSpec.scala
@@ -0,0 +1,62 @@
+package app.softnetwork.elastic.sql
+
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import scala.collection.immutable.ListMap
+
+/** `ObjectValue.set`/`remove` carry the table's `_meta`, whose paths are three and four levels deep
+  * (`_meta.columns.<column>.default_value`). Descending to the leaf's parent and re-attaching it
+  * under the FIRST key collapsed everything in between — the deeper the path, the more siblings
+  * disappeared — so a metadata update silently replaced `_meta` with its innermost object.
+  */
+class ObjectValuePathSpec extends AnyFlatSpec with Matchers {
+
+  private def meta: ObjectValue =
+    ObjectValue(
+      ListMap(
+        "type" -> StringValue("regular"),
+        "columns" -> ObjectValue(
+          ListMap(
+            "id"   -> ObjectValue(ListMap("data_type" -> StringValue("INT"))),
+            "name" -> ObjectValue(ListMap("data_type" -> StringValue("VARCHAR")))
+          )
+        )
+      )
+    )
+
+  "set" should "reach a depth-3 path without dropping its siblings" in {
+    val updated = meta.set("columns.id.default_value", IngestTimestampValue)
+    updated.find("columns.id.default_value") shouldBe Some(IngestTimestampValue)
+    updated.find("columns.id.data_type") shouldBe Some(StringValue("INT"))
+    updated.find("columns.name.data_type") shouldBe Some(StringValue("VARCHAR"))
+    updated.find("type") shouldBe Some(StringValue("regular"))
+  }
+
+  it should "create the intermediate levels of a path that does not exist yet" in {
+    val updated = meta.set("columns.age.data_type", StringValue("INT"))
+    updated.find("columns.age.data_type") shouldBe Some(StringValue("INT"))
+    updated.find("columns.id.data_type") shouldBe Some(StringValue("INT"))
+    updated.find("type") shouldBe Some(StringValue("regular"))
+  }
+
+  it should "still handle depth 1 and 2" in {
+    meta.set("type", StringValue("view")).find("type") shouldBe Some(StringValue("view"))
+    val d2 = meta.set("columns.extra", StringValue("x"))
+    d2.find("columns.extra") shouldBe Some(StringValue("x"))
+    d2.find("columns.id.data_type") shouldBe Some(StringValue("INT"))
+  }
+
+  "remove" should "reach a depth-3 path without dropping its siblings" in {
+    val updated = meta.remove("columns.id.data_type")
+    updated.find("columns.id.data_type") shouldBe None
+    updated.find("columns.name.data_type") shouldBe Some(StringValue("VARCHAR"))
+    updated.find("type") shouldBe Some(StringValue("regular"))
+  }
+
+  it should "leave the object untouched when the path is absent" in {
+    // Used to overwrite the head key with an empty object, deleting everything under it.
+    meta.remove("columns.absent.data_type") shouldBe meta
+    meta.remove("absent.deeper.still") shouldBe meta
+  }
+}
diff --git a/sql/src/test/scala/app/softnetwork/elastic/sql/parser/AlterTableRoundTripSpec.scala b/sql/src/test/scala/app/softnetwork/elastic/sql/parser/AlterTableRoundTripSpec.scala
new file mode 100644
index 00000000..81e3e8cc
--- /dev/null
+++ b/sql/src/test/scala/app/softnetwork/elastic/sql/parser/AlterTableRoundTripSpec.scala
@@ -0,0 +1,120 @@
+package app.softnetwork.elastic.sql.parser
+
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+/** Every statement the engine can RENDER must parse back — `MaterializedViewExtension` executes
+  * `client.run(alter.sql)` and writes the same string into a user-runnable `.sql` artifact, so a
+  * rendering the grammar rejects is a runtime failure, not a cosmetic one.
+  *
+  * The family this pins down was found one member at a time: `AlterColumnField` rendered `SET
+  * FIELD` while only `ADD FIELD` parsed, `AlterColumnType` rendered `SET TYPE` against the
+  * grammar's `SET DATA TYPE`, `AlterTable`/`AlterPipeline` put `IF EXISTS` after the name, and `SET
+  * DEFAULT _ingest.timestamp` — the shape the MV machinery emits for `_last_updated` — could be
+  * rendered but not parsed. Enumerating the whole surface is the only way this stays closed.
+  */
+class AlterTableRoundTripSpec extends AnyFlatSpec with Matchers {
+
+  private val statements = Seq(
+    "ALTER TABLE users ADD COLUMN last_login TIMESTAMP",
+    "ALTER TABLE users ADD COLUMN IF NOT EXISTS last_login TIMESTAMP",
+    "ALTER TABLE users DROP COLUMN old_field",
+    "ALTER TABLE users DROP COLUMN IF EXISTS old_field",
+    "ALTER TABLE users RENAME COLUMN old_name TO new_name",
+    "ALTER TABLE users ALTER COLUMN name SET DATA TYPE KEYWORD",
+    "ALTER TABLE users ALTER COLUMN age SET SCRIPT AS (DATE_DIFF(birthdate, CURRENT_DATE, YEAR))",
+    "ALTER TABLE users ALTER COLUMN age DROP SCRIPT",
+    "ALTER TABLE users ALTER COLUMN name SET DEFAULT 'unknown'",
+    "ALTER TABLE users ALTER COLUMN name DROP DEFAULT",
+    "ALTER TABLE users ALTER COLUMN name SET NOT NULL",
+    "ALTER TABLE users ALTER COLUMN name DROP NOT NULL",
+    "ALTER TABLE users ALTER COLUMN name SET COMMENT 'Full name'",
+    "ALTER TABLE users ALTER COLUMN name DROP COMMENT",
+    "ALTER TABLE users ALTER COLUMN name SET OPTION (fielddata = true)",
+    "ALTER TABLE users ALTER COLUMN name ADD OPTION (fielddata = true)",
+    "ALTER TABLE users ALTER COLUMN name DROP OPTION fielddata",
+    "ALTER TABLE users ALTER COLUMN name SET FIELDS (raw KEYWORD)",
+    "ALTER TABLE users ALTER COLUMN name SET FIELD raw KEYWORD",
+    "ALTER TABLE users ALTER COLUMN name ADD FIELD raw KEYWORD",
+    "ALTER TABLE users ALTER COLUMN name DROP FIELD raw",
+    "ALTER TABLE users SET MAPPING _meta.owner = 'analytics'",
+    "ALTER TABLE users ADD MAPPING _meta.owner = 'analytics'",
+    "ALTER TABLE users DROP MAPPING _meta.owner",
+    "ALTER TABLE users SET SETTING index.refresh_interval = '1s'",
+    "ALTER TABLE users DROP SETTING index.refresh_interval",
+    "ALTER TABLE users SET ALIAS recent = (routing = 'user1')",
+    "ALTER TABLE users DROP ALIAS recent",
+    // the shapes TableDiff renders for a materialized view
+    "ALTER TABLE users ALTER COLUMN _last_updated SET DEFAULT _ingest.timestamp",
+    "ALTER TABLE users SET MAPPING _meta.columns._last_updated.default_value = _ingest.timestamp",
+    // IF EXISTS on the table itself, and the multi-statement (parenthesised) form
+    "ALTER TABLE IF EXISTS users DROP COLUMN old_field",
+    """ALTER TABLE users (
+      |	ADD COLUMN a INT,
+      |	DROP COLUMN b
+      |)""".stripMargin,
+    """ALTER TABLE IF EXISTS users (
+      |	ADD COLUMN a INT,
+      |	DROP COLUMN b
+      |)""".stripMargin
+  )
+
+  "every ALTER TABLE statement" should "re-parse the SQL it renders" in {
+    statements.foreach { sql =>
+      val parsed = Parser(sql)
+      withClue(s"did not parse: $sql -> ") { parsed.isRight shouldBe true }
+      val stmt = parsed.toOption.get
+      withClue(s"rendering of [$sql] was [${stmt.sql}] which ") {
+        Parser(stmt.sql) shouldBe Right(stmt)
+      }
+    }
+  }
+
+  "an ALTER PIPELINE statement" should "re-parse the SQL it renders" in {
+    Seq(
+      """ALTER PIPELINE user_pipeline ADD PROCESSOR SET (field = "status", value = "active")""",
+      """ALTER PIPELINE IF EXISTS user_pipeline ADD PROCESSOR SET (field = "status", value = "active")"""
+    ).foreach { sql =>
+      val parsed = Parser(sql)
+      withClue(s"did not parse: $sql -> ") { parsed.isRight shouldBe true }
+      val rendered = parsed.toOption.get.sql
+      // Rendering fixed point rather than AST equality: two `AddPipelineProcessor`s that render
+      // identically still compare unequal (a processor field that never reaches the SQL), which
+      // is pre-existing and out of this fix's scope. What must hold is that the emitted SQL
+      // parses and survives a second round.
+      withClue(s"rendering of [$sql] was [$rendered] which ") {
+        Parser(rendered).map(_.sql) shouldBe Right(rendered)
+      }
+    }
+  }
+
+  "a string carrying quotes or backslashes" should "survive rendering" in {
+    // `SET COMMENT 'it's here'` used to be emitted verbatim and could not be parsed back.
+    Seq(
+      "ALTER TABLE users ALTER COLUMN name SET COMMENT 'it\\'s here'",
+      "ALTER TABLE users ALTER COLUMN name SET DEFAULT 'C:\\\\data'",
+      "ALTER TABLE users SET MAPPING _meta.note = 'don\\'t drop'"
+    ).foreach { sql =>
+      val parsed = Parser(sql)
+      withClue(s"did not parse: $sql -> ") { parsed.isRight shouldBe true }
+      val stmt = parsed.toOption.get
+      withClue(s"rendering of [$sql] was [${stmt.sql}] which ") {
+        Parser(stmt.sql) shouldBe Right(stmt)
+      }
+    }
+  }
+
+  "a parser guard" should "return a Left rather than throw" in {
+    // These run inside a combinator action; they used to `throw` straight out of `Parser.apply`,
+    // whose signature promises Either[ParserError, Statement].
+    Seq(
+      "ALTER TABLE users (ADD COLUMN a INT",
+      "ALTER TABLE users ADD COLUMN a INT)",
+      "ALTER TABLE users ADD COLUMN a INT, ADD COLUMN b INT",
+      "ALTER PIPELINE p (ADD PROCESSOR SET (field = \"a\", value = \"b\")"
+    ).foreach { sql =>
+      withClue(s"[$sql] ") { noException should be thrownBy Parser(sql) }
+      withClue(s"[$sql] ") { Parser(sql).isLeft shouldBe true }
+    }
+  }
+}
diff --git a/sql/src/test/scala/app/softnetwork/elastic/sql/parser/ParserSpec.scala b/sql/src/test/scala/app/softnetwork/elastic/sql/parser/ParserSpec.scala
index b6e9c76c..0e101a66 100644
--- a/sql/src/test/scala/app/softnetwork/elastic/sql/parser/ParserSpec.scala
+++ b/sql/src/test/scala/app/softnetwork/elastic/sql/parser/ParserSpec.scala
@@ -1952,6 +1952,54 @@ class ParserSpec extends AnyFlatSpec with Matchers {
     }
   }
 
+  // `ALTER COLUMN … SET|ADD FIELD` is documented, and only `ADD` ever worked: `multiFields`
+  // carried its own empty fallback, so `alterColumnFields` (`ALTER COLUMN c SET <multiFields>`)
+  // matched a bare `SET` with no fields and left `FIELD raw KEYWORD` unconsumed. Before #213 that
+  // trailing input was discarded — the user's SET FIELD silently did nothing; afterwards it was a
+  // parse error. `AlterColumnField.sql` renders `SET FIELD`, so the AST could not re-parse its own
+  // rendering either.
+
+  it should "parse ALTER COLUMN ... SET FIELD as well as ADD FIELD" in {
+    val setField = Parser("ALTER TABLE users ALTER COLUMN profile SET FIELD followers INT")
+    val addField = Parser("ALTER TABLE users ALTER COLUMN profile ADD FIELD followers INT")
+    setField.isRight shouldBe true
+    // SET and ADD are synonyms here — the grammar keeps neither, so both yield the same AST.
+    setField shouldBe addField
+    setField.toOption.get match {
+      case AlterTable("users", _, Seq(AlterColumnField("profile", field, false))) =>
+        field.name shouldBe "followers"
+        field.dataType.typeId shouldBe "INT"
+      case other => fail(s"Expected a single AlterColumnField, got $other")
+    }
+  }
+
+  it should "re-parse the SQL an AlterColumnField renders" in {
+    val stmt = Parser("ALTER TABLE users ALTER COLUMN profile ADD FIELD followers INT").toOption.get
+    Parser(stmt.sql) shouldBe Right(stmt)
+  }
+
+  it should "reject an ALTER COLUMN whose SET says nothing" in {
+    // Used to parse as AlterColumnFields(c, Nil) — a statement that silently did nothing.
+    Parser("ALTER TABLE users ALTER COLUMN profile SET").isLeft shouldBe true
+  }
+
+  it should "still parse ALTER COLUMN ... SET FIELDS (...)" in {
+    val result = Parser(
+      "ALTER TABLE users ALTER COLUMN profile SET FIELDS (city VARCHAR, followers INT)"
+    )
+    result.isRight shouldBe true
+    result.toOption.get match {
+      case AlterTable("users", _, Seq(AlterColumnFields("profile", fields, false))) =>
+        fields.map(_.name) should contain inOrder ("city", "followers")
+      case other => fail(s"Expected a single AlterColumnFields, got $other")
+    }
+  }
+
+  it should "still parse a column that declares neither FIELDS nor a script" in {
+    // optionalMultiFields: the empty fallback the column definition depends on.
+    Parser("CREATE TABLE t (id INT, name VARCHAR)").isRight shouldBe true
+  }
+
   it should "parse ALTER TABLE MAPPINGS" in {
     val sql = """ALTER TABLE orders (
                 |  ADD COLUMN _last_updated TIMESTAMP DEFAULT _ingest.timestamp,
diff --git a/sql/src/test/scala/app/softnetwork/elastic/sql/query/BucketIncludesSpec.scala b/sql/src/test/scala/app/softnetwork/elastic/sql/query/BucketIncludesSpec.scala
new file mode 100644
index 00000000..6d69bfe0
--- /dev/null
+++ b/sql/src/test/scala/app/softnetwork/elastic/sql/query/BucketIncludesSpec.scala
@@ -0,0 +1,55 @@
+package app.softnetwork.elastic.sql.query
+
+import app.softnetwork.elastic.sql.parser.Parser
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+/** The `terms` aggregation's `include` / `exclude` (and their regex form) are derived from a HAVING
+  * criterion's value. They must come from the VALUE, never from its SQL rendering: `.sql` carries
+  * the quote delimiters and, since string literals are escaped, a value holding one backslash would
+  * reach Elasticsearch holding two — an include that matches nothing.
+  */
+class BucketIncludesSpec extends AnyFlatSpec with Matchers {
+
+  private def includesOf(sql: String): BucketIncludesExcludes =
+    Parser(sql) match {
+      case Right(s: SingleSearch) =>
+        val bucket = s.buckets.headOption.getOrElse(fail(s"no bucket in [$sql]"))
+        s.having
+          .flatMap(_.criteria)
+          .map(_.includes(bucket, not = false, BucketIncludesExcludes()))
+          .getOrElse(fail(s"no HAVING criteria in [$sql]"))
+      case other => fail(s"Expected a SingleSearch, got $other")
+    }
+
+  "a HAVING equality on the grouped field" should "include the value itself" in {
+    includesOf(
+      "SELECT category, COUNT(*) AS c FROM t GROUP BY category HAVING category = 'books'"
+    ).values shouldBe Set("books")
+  }
+
+  it should "not escape a backslash the value carries" in {
+    // SQL source `'a\\b'` is the two-character value `a\b`.
+    includesOf(
+      "SELECT category, COUNT(*) AS c FROM t GROUP BY category HAVING category = 'a\\\\b'"
+    ).values shouldBe Set("a\\b")
+  }
+
+  it should "keep an apostrophe the value carries" in {
+    includesOf(
+      "SELECT category, COUNT(*) AS c FROM t GROUP BY category HAVING category = 'it\\'s'"
+    ).values shouldBe Set("it's")
+  }
+
+  "a HAVING LIKE on the grouped field" should "build a regex without the quote delimiters" in {
+    includesOf(
+      "SELECT category, COUNT(*) AS c FROM t GROUP BY category HAVING category LIKE 'book%'"
+    ).regex shouldBe Some("book.*")
+  }
+
+  "a HAVING RLIKE on the grouped field" should "pass the pattern through verbatim" in {
+    includesOf(
+      "SELECT category, COUNT(*) AS c FROM t GROUP BY category HAVING category RLIKE 'boo.+'"
+    ).regex shouldBe Some("boo.+")
+  }
+}
diff --git a/sql/src/test/scala/app/softnetwork/elastic/sql/schema/AlterTableMergeSpec.scala b/sql/src/test/scala/app/softnetwork/elastic/sql/schema/AlterTableMergeSpec.scala
new file mode 100644
index 00000000..70e8e6f1
--- /dev/null
+++ b/sql/src/test/scala/app/softnetwork/elastic/sql/schema/AlterTableMergeSpec.scala
@@ -0,0 +1,159 @@
+package app.softnetwork.elastic.sql.schema
+
+import app.softnetwork.elastic.sql.{IngestTimestampValue, ObjectValue, StringValue}
+import app.softnetwork.elastic.sql.`type`.SQLTypes
+import app.softnetwork.elastic.sql.parser.Parser
+import app.softnetwork.elastic.sql.query.{AlterTable, AlterTableStatement}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+/** `Table.merge` is what turns an ALTER TABLE into the schema the gateway diffs against
+  * Elasticsearch (`GatewayApi.run`: merge -> diff -> push). The single-field branches used to
+  * compute their `copy` and discard it, returning the table unchanged: `SET|ADD FIELD` and `DROP
+  * FIELD` produced an EMPTY diff, so nothing reached the cluster and the DDL still reported
+  * success. Parser-level tests cannot see that — only asserting the merged schema can.
+  */
+class AlterTableMergeSpec extends AnyFlatSpec with Matchers {
+
+  private def statementsOf(sql: String): Seq[AlterTableStatement] =
+    Parser(sql) match {
+      case Right(AlterTable(_, _, statements)) => statements
+      case other                               => fail(s"Expected an AlterTable, got $other")
+    }
+
+  private val users: Table =
+    Table(
+      name = "users",
+      columns = List(
+        Column("id", SQLTypes.Int),
+        Column("name", SQLTypes.Varchar)
+      ),
+      primaryKey = List("id")
+    ).update()
+
+  private def multiFieldsOf(table: Table, column: String): List[String] =
+    table.columns.find(_.name == column).map(_.multiFields.map(_.name)).getOrElse(Nil)
+
+  "merge" should "apply ALTER COLUMN ... SET FIELD" in {
+    val merged =
+      users.merge(statementsOf("ALTER TABLE users ALTER COLUMN name SET FIELD raw KEYWORD"))
+    multiFieldsOf(merged, "name") should contain("raw")
+  }
+
+  it should "apply ALTER COLUMN ... ADD FIELD" in {
+    val merged =
+      users.merge(statementsOf("ALTER TABLE users ALTER COLUMN name ADD FIELD raw KEYWORD"))
+    multiFieldsOf(merged, "name") should contain("raw")
+  }
+
+  it should "produce a non-empty diff for SET FIELD" in {
+    // The gateway pushes `schema.diff(merged)`; an unchanged merge means an empty diff, i.e. a
+    // DDL that silently does nothing to the index.
+    val merged =
+      users.merge(statementsOf("ALTER TABLE users ALTER COLUMN name SET FIELD raw KEYWORD"))
+    users.diff(merged).columns should not be empty
+  }
+
+  it should "replace a sub-field of the same name rather than duplicate it" in {
+    val withRaw =
+      users.merge(statementsOf("ALTER TABLE users ALTER COLUMN name SET FIELD raw KEYWORD"))
+    val reAdded =
+      withRaw.merge(statementsOf("ALTER TABLE users ALTER COLUMN name SET FIELD raw VARCHAR"))
+    multiFieldsOf(reAdded, "name").count(_ == "raw") shouldBe 1
+    reAdded.columns
+      .find(_.name == "name")
+      .flatMap(_.multiFields.find(_.name == "raw"))
+      .map(_.dataType.typeId) shouldBe Some("VARCHAR")
+  }
+
+  it should "apply ALTER COLUMN ... DROP FIELD" in {
+    val withRaw =
+      users.merge(statementsOf("ALTER TABLE users ALTER COLUMN name SET FIELD raw KEYWORD"))
+    multiFieldsOf(withRaw, "name") should contain("raw")
+
+    val dropped = withRaw.merge(statementsOf("ALTER TABLE users ALTER COLUMN name DROP FIELD raw"))
+    multiFieldsOf(dropped, "name") should not contain "raw"
+    withRaw.diff(dropped).columns should not be empty
+  }
+
+  // `merge` folds the statements, so each one must see the table the previous one produced. The
+  // column branches resolved against `this` — the table as it was BEFORE the fold — so a statement
+  // could not touch a column an earlier statement in the same ALTER had just added.
+
+  it should "see a column added earlier in the same ALTER TABLE" in {
+    val merged = users.merge(
+      statementsOf(
+        "ALTER TABLE users (ADD COLUMN profile VARCHAR, ALTER COLUMN profile SET FIELDS (raw KEYWORD))"
+      )
+    )
+    multiFieldsOf(merged, "profile") should contain("raw")
+  }
+
+  it should "rename a column added earlier in the same ALTER TABLE" in {
+    val merged = users.merge(
+      statementsOf("ALTER TABLE users (ADD COLUMN profile VARCHAR, RENAME COLUMN profile TO bio)")
+    )
+    merged.columns.map(_.name) should contain("bio")
+    merged.columns.map(_.name) should not contain "profile"
+  }
+
+  it should "converge when the ALTER a diff produced is applied back" in {
+    // What a materialized-view reconcile does: diff -> render -> run -> diff again. The second
+    // diff has to be empty, or the loop keeps re-emitting the same statements. A column gaining
+    // `DEFAULT _ingest.timestamp` is the shape the MV machinery adds, and it exercises both the
+    // ingest value and the `_meta.columns.<c>.default_value` metadata mirror.
+    val desired =
+      users
+        .copy(columns =
+          users.columns :+ Column(
+            "_last_updated",
+            SQLTypes.Timestamp,
+            defaultValue = Some(IngestTimestampValue)
+          )
+        )
+        .update()
+
+    val firstDiff = users.diff(desired)
+    firstDiff.isEmpty shouldBe false
+
+    val alter = firstDiff
+      .alterTable("users", ifExists = false)
+      .getOrElse(fail("expected the diff to produce an ALTER TABLE"))
+    val merged = users.merge(alter.statements)
+
+    withClue(s"residual diff after applying [${alter.sql}]: ") {
+      merged.diff(desired).isEmpty shouldBe true
+    }
+  }
+
+  it should "keep the rest of _meta when a column's metadata changes" in {
+    val desired =
+      users
+        .copy(columns = users.columns.map {
+          case c if c.name == "name" => c.copy(defaultValue = Some(StringValue("anonymous")))
+          case c                     => c
+        })
+        .update()
+
+    val merged = users.merge(
+      users.diff(desired).alterTable("users", ifExists = false).map(_.statements).getOrElse(Nil)
+    )
+
+    // `_meta.columns.<c>.…` is three levels deep — the depth at which the metadata update used to
+    // replace the whole of `_meta` with its innermost object.
+    merged.mappings.get("_meta") match {
+      case Some(m: ObjectValue) =>
+        m.find("columns.id.data_type") should not be empty
+        m.find("columns.name.data_type") should not be empty
+      case other => fail(s"Expected an ObjectValue _meta, got $other")
+    }
+  }
+
+  it should "still apply ALTER COLUMN ... SET FIELDS (...)" in {
+    // Control for the plural branch this fix was modelled on.
+    val merged = users.merge(
+      statementsOf("ALTER TABLE users ALTER COLUMN name SET FIELDS (raw KEYWORD, en VARCHAR)")
+    )
+    multiFieldsOf(merged, "name") should contain allOf ("raw", "en")
+  }
+}