Support append row kinds and historical row updates

crates/paimon/src/spec/schema.rs

@@ -23,7 +23,7 @@ use crate::spec::core_options::{
     QUERY_AUTH_ENABLED_OPTION, SEQUENCE_FIELD_OPTION, TABLE_READ_SEQUENCE_NUMBER_ENABLED_OPTION,
     TABLE_TYPE_OPTION,
 };
-use crate::spec::types::{ArrayType, DataType, MapType, MultisetType, RowType, VarCharType};
+use crate::spec::types::{ArrayType, DataType, MapType, MultisetType, RowType};
 use crate::spec::{
     remove_field_scoped_options, rename_field_scoped_options,
     validate_no_aggregation_on_sequence_field, AggregationConfig, BlobType, ColumnMove,
@@ -1416,7 +1416,7 @@ impl Schema {
         Self::validate_first_row_changelog_producer(options)?;
         Self::validate_changelog_producer_requires_primary_keys(options, primary_keys)?;
         Self::validate_read_sequence_number_requires_primary_keys(options, primary_keys)?;
-        Self::validate_rowkind_field(options, primary_keys, fields)?;
+        Self::validate_rowkind_field(options, fields)?;
         Self::validate_deletion_vectors(options)?;
         Self::validate_bucket_keys(options, fields, partition_keys, primary_keys)?;
         Self::validate_sequence_field(options, fields, partition_keys, primary_keys)?;
@@ -2120,42 +2120,25 @@ impl Schema {
 
     fn validate_rowkind_field(
         options: &HashMap<String, String>,
-        primary_keys: &[String],
         fields: &[DataField],
     ) -> crate::Result<()> {
         let core = CoreOptions::new(options);
         let Some(field_name) = core.rowkind_field() else {
             return Ok(());
         };
-
-        if primary_keys.is_empty() {
-            return Err(crate::Error::ConfigInvalid {
-                message: "rowkind.field requires a primary-key table".to_string(),
-            });
-        }
-
-        let producer = core
-            .try_changelog_producer()
+        core.try_changelog_producer()
             .map_err(Self::options_error_to_config_invalid)?;
-        if producer == ChangelogProducer::Input {
-            return Err(crate::Error::ConfigInvalid {
-                message: "rowkind.field cannot be used with changelog-producer=input".to_string(),
-            });
-        }
 
         let Some(field) = fields.iter().find(|f| f.name() == field_name) else {
             return Err(crate::Error::ConfigInvalid {
                 message: format!("Rowkind field '{field_name}' can not be found in table schema."),
             });
         };
 
-        if !matches!(
-            field.data_type(),
-            DataType::VarChar(v) if v.length() == VarCharType::MAX_LENGTH
-        ) {
+        if !matches!(field.data_type(), DataType::Char(_) | DataType::VarChar(_)) {
             return Err(crate::Error::ConfigInvalid {
                 message: format!(
-                    "rowkind.field '{field_name}' must be STRING (unbounded VARCHAR) type"
+                    "rowkind.field '{field_name}' must be CHARACTER_STRING (CHAR or VARCHAR) type"
                 ),
             });
         }
@@ -6591,17 +6574,13 @@ mod tests {
     }
 
     #[test]
-    fn rowkind_field_requires_primary_key_table() {
-        let err = Schema::builder()
+    fn rowkind_field_accepts_append_tables() {
+        Schema::builder()
             .column("id", DataType::Int(IntType::new()))
             .column("op", DataType::VarChar(VarCharType::string_type()))
             .option("rowkind.field", "op")
             .build()
-            .unwrap_err();
-        assert!(
-            matches!(err, crate::Error::ConfigInvalid { ref message } if message.contains("rowkind.field")),
-            "got {err:?}"
-        );
+            .unwrap();
     }
 
     #[test]
@@ -6690,20 +6669,15 @@ mod tests {
     }
 
     #[test]
-    fn rowkind_field_rejects_changelog_producer_input() {
-        let err = Schema::builder()
+    fn rowkind_field_accepts_changelog_producer_input() {
+        Schema::builder()
             .column("id", DataType::Int(IntType::new()))
             .column("op", DataType::VarChar(VarCharType::string_type()))
             .primary_key(["id"])
             .option("changelog-producer", "input")
             .option("rowkind.field", "op")
             .build()
-            .unwrap_err();
-        assert!(
-            matches!(err, crate::Error::ConfigInvalid { ref message }
-                if message.contains("changelog-producer=input")),
-            "got {err:?}"
-        );
+            .unwrap();
     }
 
     #[test]

crates/paimon/src/table/mod.rs

@@ -302,7 +302,8 @@ pub struct Table {
     query_auth_session: Option<u64>,
     rest_env: Option<RESTEnv>,
     /// True when this table copy was switched to a historical schema by
-    /// [`Table::copy_with_time_travel`]. Such a copy is read-only.
+    /// [`Table::copy_with_time_travel`]. Ordinary writes and publication reject
+    /// this copy; row-level updates may use it as their read snapshot.
     time_traveled: bool,
     /// Captured snapshot or empty view, so scans don't resolve latest again.
     /// Cleared when [`Table::copy_with_options`] changes the selector.
@@ -801,7 +802,7 @@ impl Table {
         })
     }
 
-    /// Create a read-only copy pinned to an already resolved snapshot.
+    /// Create a read view pinned to an already resolved snapshot.
     ///
     /// Replaces any selector that originally resolved the snapshot with an
     /// explicit `scan.snapshot-id`, so every subsequent scan stage observes the
@@ -817,7 +818,7 @@ impl Table {
         Ok(table)
     }
 
-    /// Create a read-only copy pinned to a snapshot from this table and branch,
+    /// Create a read view pinned to a snapshot from this table and branch,
     /// preserving the current read schema and options other than scan selectors.
     ///
     /// Unlike time travel, pinning must not change the fields used by an already
@@ -1133,8 +1134,9 @@ impl Table {
         })
     }
 
-    /// Whether this copy is read-only because its schema was time-traveled or
-    /// its read view was explicitly pinned (including an empty view).
+    /// Whether this copy has a time-traveled schema or an explicitly pinned
+    /// read view (including an empty view). Ordinary writes and commits reject
+    /// it; row-level updates retain its snapshot as their conflict baseline.
     pub fn is_time_traveled(&self) -> bool {
         self.time_traveled
     }

crates/paimon/src/table/row_kind_generator.rs

@@ -19,17 +19,17 @@
 //!
 //! Reference: Java `org.apache.paimon.table.sink.RowKindGenerator`.
 
-use arrow_array::{Array, LargeStringArray, RecordBatch, StringArray};
+use arrow_array::{LargeStringArray, RecordBatch, StringArray};
 
 use crate::spec::{RowKind, TableSchema};
 
 pub struct RowKindGenerator {
-    index: usize,
+    field_name: String,
 }
 
 impl RowKindGenerator {
     pub fn create(schema: &TableSchema, field_name: &str) -> crate::Result<Self> {
-        let index = schema
+        schema
             .fields()
             .iter()
             .position(|f| f.name() == field_name)
@@ -40,28 +40,42 @@ impl RowKindGenerator {
                 ),
                 source: None,
             })?;
-        Ok(Self { index })
+        Ok(Self {
+            field_name: field_name.to_owned(),
+        })
     }
 
-    pub fn generate(&self, batch: &RecordBatch, row: usize) -> crate::Result<RowKind> {
-        let col = batch.column(self.index);
-        if col.is_null(row) {
-            return Err(crate::Error::DataInvalid {
-                message: "Row kind cannot be null.".to_string(),
-                source: None,
-            });
-        }
-        let value = if let Some(strings) = col.as_any().downcast_ref::<StringArray>() {
-            strings.value(row)
+    pub fn generate_batch(&self, batch: &RecordBatch) -> crate::Result<Vec<RowKind>> {
+        // with_write_type may reorder the physical columns. Resolve once for
+        // the batch, then parse values without a per-row field-name search.
+        let col =
+            batch
+                .column_by_name(&self.field_name)
+                .ok_or_else(|| crate::Error::DataInvalid {
+                    message: format!(
+                        "Cannot find rowkind field '{}' in write input",
+                        self.field_name
+                    ),
+                    source: None,
+                })?;
+        let parse = |value: Option<&str>| {
+            value
+                .ok_or_else(|| crate::Error::DataInvalid {
+                    message: "Row kind cannot be null.".into(),
+                    source: None,
+                })
+                .and_then(RowKind::from_short_string)
+        };
+        if let Some(strings) = col.as_any().downcast_ref::<StringArray>() {
+            strings.iter().map(parse).collect()
         } else if let Some(strings) = col.as_any().downcast_ref::<LargeStringArray>() {
-            strings.value(row)
+            strings.iter().map(parse).collect()
         } else {
-            return Err(crate::Error::DataInvalid {
-                message: "rowkind.field column must be Utf8/String".to_string(),
+            Err(crate::Error::DataInvalid {
+                message: "rowkind.field column must be Utf8/String".into(),
                 source: None,
-            });
-        };
-        RowKind::from_short_string(value)
+            })
+        }
     }
 }
 
@@ -104,6 +118,6 @@ mod tests {
             ],
         )
         .unwrap();
-        assert_eq!(gen.generate(&batch, 0).unwrap(), RowKind::Delete);
+        assert_eq!(gen.generate_batch(&batch).unwrap(), vec![RowKind::Delete]);
     }
 }