support produce partitioned table

bindings/cpp/src/lib.rs

@@ -508,7 +508,10 @@ impl Table {
             Err(e) => return Err(format!("Failed to create append: {e}")),
         };
 
-        let writer = table_append.create_writer();
+        let writer = match table_append.create_writer() {
+            Ok(w) => w,
+            Err(e) => return Err(format!("Failed to create writer: {e}")),
+        };
         let writer = Box::into_raw(Box::new(AppendWriter { inner: writer }));
         Ok(writer)
     }
@@ -580,7 +583,7 @@ impl AppendWriter {
     fn append(&mut self, row: &ffi::FfiGenericRow) -> ffi::FfiResult {
         let generic_row = types::ffi_row_to_core(row);
 
-        let result = RUNTIME.block_on(async { self.inner.append(generic_row).await });
+        let result = RUNTIME.block_on(async { self.inner.append(&generic_row).await });
 
         match result {
             Ok(_) => ok_result(),

bindings/python/src/table.rs

@@ -68,7 +68,9 @@ impl FlussTable {
                 .new_append()
                 .map_err(|e| FlussError::new_err(e.to_string()))?;
 
-            let rust_writer = table_append.create_writer();
+            let rust_writer = table_append
+                .create_writer()
+                .map_err(|e| FlussError::new_err(e.to_string()))?;
 
             let py_writer = AppendWriter::from_core(rust_writer, table_info);
 
@@ -251,7 +253,7 @@ impl AppendWriter {
 
         future_into_py(py, async move {
             inner
-                .append(generic_row)
+                .append(&generic_row)
                 .await
                 .map_err(|e| FlussError::new_err(e.to_string()))
         })

crates/examples/src/example_kv_table.rs

@@ -16,7 +16,7 @@
 // under the License.
 
 use clap::Parser;
-use fluss::client::{FlussConnection, UpsertWriter};
+use fluss::client::FlussConnection;
 use fluss::config::Config;
 use fluss::error::Result;
 use fluss::metadata::{DataTypes, Schema, TableDescriptor, TablePath};

crates/examples/src/example_partitioned_kv_table.rs

@@ -16,7 +16,7 @@
 // under the License.
 
 use clap::Parser;
-use fluss::client::{FlussAdmin, FlussConnection, UpsertWriter};
+use fluss::client::{FlussAdmin, FlussConnection};
 use fluss::config::Config;
 use fluss::error::Result;
 use fluss::metadata::{DataTypes, PartitionSpec, Schema, TableDescriptor, TablePath};

crates/examples/src/example_table.rs

@@ -63,13 +63,13 @@ pub async fn main() -> Result<()> {
     row.set_field(2, 123_456_789_123i64);
 
     let table = conn.get_table(&table_path).await?;
-    let append_writer = table.new_append()?.create_writer();
-    let f1 = append_writer.append(row);
-    row = GenericRow::new(3);
+    let append_writer = table.new_append()?.create_writer()?;
+    let f1 = append_writer.append(&row);
+    let mut row = GenericRow::new(3);
     row.set_field(0, 233333);
     row.set_field(1, "tt44");
     row.set_field(2, 987_654_321_987i64);
-    let f2 = append_writer.append(row);
+    let f2 = append_writer.append(&row);
     try_join!(f1, f2, append_writer.flush())?;
 
     // scan rows

crates/fluss/src/client/table/append.rs

@@ -15,16 +15,16 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use crate::client::table::partition_getter::{PartitionGetter, get_physical_path};
 use crate::client::{WriteRecord, WriterClient};
 use crate::error::Result;
 use crate::metadata::{PhysicalTablePath, TableInfo, TablePath};
-use crate::row::GenericRow;
+use crate::row::{ColumnarRow, InternalRow};
 use arrow::array::RecordBatch;
 use std::sync::Arc;
 
-#[allow(dead_code)]
 pub struct TableAppend {
-    table_path: TablePath,
+    table_path: Arc<TablePath>,
     table_info: Arc<TableInfo>,
     writer_client: Arc<WriterClient>,
 }
@@ -36,32 +36,48 @@ impl TableAppend {
         writer_client: Arc<WriterClient>,
     ) -> Self {
         Self {
-            table_path,
+            table_path: Arc::new(table_path),
             table_info,
             writer_client,
         }
     }
 
-    pub fn create_writer(&self) -> AppendWriter {
-        AppendWriter {
-            physical_table_path: Arc::new(PhysicalTablePath::of(Arc::new(self.table_path.clone()))),
+    pub fn create_writer(&self) -> Result<AppendWriter> {
+        let partition_getter = if self.table_info.is_partitioned() {
+            Some(PartitionGetter::new(
+                self.table_info.row_type(),
+                Arc::clone(self.table_info.get_partition_keys()),
+            )?)
+        } else {
+            None
+        };
+
+        Ok(AppendWriter {
+            table_path: Arc::clone(&self.table_path),
+            partition_getter,
             writer_client: self.writer_client.clone(),
             table_info: Arc::clone(&self.table_info),
-        }
+        })
     }
 }
 
 pub struct AppendWriter {
-    physical_table_path: Arc<PhysicalTablePath>,
+    table_path: Arc<TablePath>,
+    partition_getter: Option<PartitionGetter>,
     writer_client: Arc<WriterClient>,
     table_info: Arc<TableInfo>,
 }
 
 impl AppendWriter {
-    pub async fn append(&self, row: GenericRow<'_>) -> Result<()> {
+    pub async fn append<R: InternalRow>(&self, row: &R) -> Result<()> {
+        let physical_table_path = Arc::new(get_physical_path(
+            &self.table_path,
+            self.partition_getter.as_ref(),
+            row,
+        )?);
         let record = WriteRecord::for_append(
             Arc::clone(&self.table_info),
-            Arc::clone(&self.physical_table_path),
+            physical_table_path,
             self.table_info.schema_id,
             row,
         );
@@ -70,10 +86,25 @@ impl AppendWriter {
         result_handle.result(result)
     }
 
+    /// Appends an Arrow RecordBatch to the table.
+    ///
+    /// For partitioned tables, the partition is derived from the **first row** of the batch.
+    /// Callers must ensure all rows in the batch belong to the same partition.
     pub async fn append_arrow_batch(&self, batch: RecordBatch) -> Result<()> {
+        let physical_table_path = if self.partition_getter.is_some() && batch.num_rows() > 0 {
+            let first_row = ColumnarRow::new(Arc::new(batch.clone()));
+            Arc::new(get_physical_path(
+                &self.table_path,
+                self.partition_getter.as_ref(),
+                &first_row,
+            )?)
+        } else {
+            Arc::new(PhysicalTablePath::of(Arc::clone(&self.table_path)))
+        };
+
         let record = WriteRecord::for_append_record_batch(
             Arc::clone(&self.table_info),
-            Arc::clone(&self.physical_table_path),
+            physical_table_path,
             self.table_info.schema_id,
             batch,
         );

crates/fluss/src/client/table/log_fetch_buffer.rs

@@ -917,7 +917,7 @@ mod tests {
         let mut row = GenericRow::new(2);
         row.set_field(0, 1_i32);
         row.set_field(1, "alice");
-        let record = WriteRecord::for_append(table_info, physical_table_path, 1, row);
+        let record = WriteRecord::for_append(table_info, physical_table_path, 1, &row);
         builder.append(&record)?;
 
         let data = builder.build()?;

crates/fluss/src/client/table/mod.rs

@@ -31,16 +31,14 @@ mod partition_getter;
 mod remote_log;
 mod scanner;
 mod upsert;
-mod writer;
 
-use crate::client::table::upsert::TableUpsert;
 pub use append::{AppendWriter, TableAppend};
 pub use lookup::{LookupResult, Lookuper, TableLookup};
 pub use remote_log::{
     DEFAULT_SCANNER_REMOTE_LOG_DOWNLOAD_THREADS, DEFAULT_SCANNER_REMOTE_LOG_PREFETCH_NUM,
 };
 pub use scanner::{LogScanner, RecordBatchLogScanner, TableScan};
-pub use writer::{TableWriter, UpsertWriter};
+pub use upsert::{TableUpsert, UpsertWriter};
 
 #[allow(dead_code)]
 pub struct FlussTable<'a> {

crates/fluss/src/client/table/partition_getter.rs

@@ -17,12 +17,29 @@
 
 use crate::error::Error::IllegalArgument;
 use crate::error::Result;
-use crate::metadata::{DataType, ResolvedPartitionSpec, RowType};
+use crate::metadata::{DataType, PhysicalTablePath, ResolvedPartitionSpec, RowType, TablePath};
 use crate::row::InternalRow;
 use crate::row::field_getter::FieldGetter;
 use crate::util::partition;
 use std::sync::Arc;
 
+/// Get the physical table path for a row, handling partitioned vs non-partitioned tables.
+pub fn get_physical_path<R: InternalRow>(
+    table_path: &Arc<TablePath>,
+    partition_getter: Option<&PartitionGetter>,
+    row: &R,
+) -> Result<PhysicalTablePath> {
+    if let Some(getter) = partition_getter {
+        let partition = getter.get_partition(row)?;
+        Ok(PhysicalTablePath::of_partitioned(
+            Arc::clone(table_path),
+            Some(partition),
+        ))
+    } else {
+        Ok(PhysicalTablePath::of(Arc::clone(table_path)))
+    }
+}
+
 /// A getter to get partition name from a row.
 #[allow(dead_code)]
 pub struct PartitionGetter {

crates/fluss/src/client/table/scanner.rs

@@ -1516,14 +1516,11 @@ mod tests {
             },
         )?;
         let physical_table_path = Arc::new(PhysicalTablePath::of(table_path));
-        let record = WriteRecord::for_append(
-            Arc::new(table_info.clone()),
-            physical_table_path,
-            1,
-            GenericRow {
-                values: vec![Datum::Int32(1)],
-            },
-        );
+        let row = GenericRow {
+            values: vec![Datum::Int32(1)],
+        };
+        let record =
+            WriteRecord::for_append(Arc::new(table_info.clone()), physical_table_path, 1, &row);
         builder.append(&record)?;
         builder.build()
     }