fix: schema mismatch during write (apache#966)

## Rationale
Close apache#308 

## Detailed Changes


## Test Plan
This is a corner case between write and alter schema, the testcase is
not easy to construct, so we plan to test this in test env first, then
add tests later.

---------

Co-authored-by: jiacai2050 <dev@liujiacai.net>

analytic_engine/src/instance/write.rs

@@ -281,9 +281,7 @@ impl<'a> MemTableWriter<'a> {
     // inserting memtable? RocksDB checks memtable size in MemTableInserter
     /// Write data into memtable.
     ///
-    /// The data in `encoded_rows` will be moved to memtable.
-    ///
-    /// The len of `row_group` and `encoded_rows` must be equal.
+    /// index_in_writer must match the schema in table_data.
     pub fn write(
         &self,
         sequence: SequenceNumber,
@@ -295,7 +293,7 @@ impl<'a> MemTableWriter<'a> {
             return Ok(());
         }
 
-        let schema = row_group.schema();
+        let schema = &self.table_data.schema();
         // Store all memtables we wrote and update their last sequence later.
         let mut wrote_memtables: SmallVec<[_; 4]> = SmallVec::new();
         let mut last_mutable_mem: Option<MemTableForWrite> = None;

analytic_engine/src/table/data.rs

@@ -368,12 +368,11 @@ impl TableData {
         timestamp: Timestamp,
         table_schema: &Schema,
     ) -> Result<MemTableForWrite> {
-        let schema_version = table_schema.version();
         let last_sequence = self.last_sequence();
 
         if let Some(mem) = self
             .current_version
-            .memtable_for_write(timestamp, schema_version)
+            .memtable_for_write(timestamp, table_schema.version())
             .context(FindMemTable)?
         {
             return Ok(mem);

common_types/src/row/contiguous.rs

@@ -286,8 +286,6 @@ impl<'a, T: RowBuffer + 'a> ContiguousRowWriter<'a, T> {
         self.inner
             .reset(datum_buffer_len, DatumKind::Null.into_u8());
 
-        assert_eq!(row.num_columns(), self.table_schema.num_columns());
-
         // Offset to next string in string buffer.
         let mut next_string_offset: OffsetSize = 0;
         for index_in_table in 0..self.table_schema.num_columns() {