Primary key tables (KV tables) support upsert, delete, and lookup operations.
use fluss::metadata::{DataTypes, Schema, TableDescriptor, TablePath}; let table_descriptor = TableDescriptor::builder() .schema( Schema::builder() .column("id", DataTypes::int()) .column("name", DataTypes::string()) .column("age", DataTypes::bigint()) .primary_key(vec!["id"]) .build()?, ) .build()?; let table_path = TablePath::new("fluss", "users"); admin.create_table(&table_path, &table_descriptor, true).await?;
use fluss::row::{GenericRow, InternalRow}; let table = conn.get_table(&table_path).await?; let table_upsert = table.new_upsert()?; let upsert_writer = table_upsert.create_writer()?; for (id, name, age) in [(1, "Alice", 25i64), (2, "Bob", 30), (3, "Charlie", 35)] { let mut row = GenericRow::new(3); row.set_field(0, id); row.set_field(1, name); row.set_field(2, age); upsert_writer.upsert(&row)?; } upsert_writer.flush().await?;
Upsert with the same primary key to update an existing record.
let mut row = GenericRow::new(3); row.set_field(0, 1); // id (primary key) row.set_field(1, "Alice"); row.set_field(2, 26i64); // updated age upsert_writer.upsert(&row)?; upsert_writer.flush().await?;
// Only primary key field needs to be set let mut row = GenericRow::new(3); row.set_field(0, 2); // id of record to delete upsert_writer.delete(&row)?; upsert_writer.flush().await?;
Update only specific columns while preserving others.
// By column indices let partial_upsert = table_upsert.partial_update(Some(vec![0, 2]))?; let partial_writer = partial_upsert.create_writer()?; let mut row = GenericRow::new(3); row.set_field(0, 1); // id (primary key, required) row.set_field(2, 27i64); // age (will be updated) // name will remain unchanged partial_writer.upsert(&row)?; partial_writer.flush().await?; // By column names let partial_upsert = table_upsert.partial_update_with_column_names(&["id", "age"])?; let partial_writer = partial_upsert.create_writer()?;
let mut lookuper = table.new_lookup()?.create_lookuper()?; let mut key = GenericRow::new(1); key.set_field(0, 1); // id to lookup let result = lookuper.lookup(&key).await?; if let Some(row) = result.get_single_row()? { println!( "Found: id={}, name={}, age={}", row.get_int(0)?, row.get_string(1)?, row.get_long(2)? ); } else { println!("Record not found"); }