feat: read instructions added

Author: HelloJowet

docs/data_engineering/data_lakehouse/delta_lake/rust.md

@@ -7,6 +7,7 @@ This guide provides an introduction to using the [Delta Lake Rust library (delta
 ```toml
 [dependencies]
 deltalake = { version = "0.22.3", features = ["datafusion", "s3"] }
+tokio = "1.42.0"
 ```
 
 ## Initial setup
@@ -30,8 +31,14 @@ fn configure_s3() {
 
 /// Builds a `DeltaOps` instance for the specified Delta table.
 /// Enabling operations such as creating, reading and writing data in the Delta Lake format.
-fn get_delta_ops(table_name: &str) -> Result<DeltaOps, DeltaTableError> {
-    let delta_table = DeltaTableBuilder::from_uri(format!("s3://data-lakehouse/{}", table_name)).build()?;
+async fn get_delta_ops(table_name: &str, load_state: bool) -> Result<DeltaOps, DeltaTableError> {
+    let delta_table_builder = DeltaTableBuilder::from_uri(format!("s3://data-lakehouse/{}", table_name));
+    let delta_table = match load_state {
+        // Load the existing table state
+        true => delta_table_builder.load().await?,
+        // Build the table without loading existing state
+        false => delta_table_builder.build()?,
+    };
 
     Ok(DeltaOps::from(delta_table))
 }
@@ -42,6 +49,8 @@ async fn main() {
 }
 ```
 
+If the table doesn't exist yet, the `load_state` parameter in `get_delta_ops` should be set to `false`, as setting it to `true` would attempt to read a non-existent state, resulting in an error. On the other hand, if you want to read from an existing table, `load_state` must be set to `true` to successfully load the data; otherwise, the load operation will fail.
+
 ## Create table
 
 ```rust
@@ -51,7 +60,7 @@ use deltalake::{DeltaTable, DeltaTableError};
 // ...
 
 async fn create_table(table_name: &str) -> Result<DeltaTable, DeltaTableError> {
-    let delta_ops = get_delta_ops(table_name)?;
+    let delta_ops = get_delta_ops(table_name, false)?;
 
     let table = delta_ops
         .create()
@@ -95,7 +104,7 @@ async fn insert(table_name: &str, save_mode: SaveMode) -> Result<DeltaTable, Del
     let names = StringArray::from(vec!["Tom", "Tim", "Titus"]);
     let employee_record = RecordBatch::try_new(schema, vec![Arc::new(ids), Arc::new(names)]).unwrap();
 
-    let delta_ops = get_delta_ops(table_name)?;
+    let delta_ops = get_delta_ops(table_name, false)?;
     // Insert record
     let table = delta_ops.write(vec![employee_record]).with_save_mode(save_mode).await?;
 
@@ -112,3 +121,31 @@ async fn main() {
 ```
 
 The default save mode for the `delta_ops.write` function is `SaveMode::Append`. To overwrite existing data instead of appending, use `SaveMode::Overwrite`.
+
+## Read
+
+```rust
+use deltalake::arrow::array::RecordBatch;
+use deltalake::operations::collect_sendable_stream;
+use deltalake::DeltaTableError;
+
+// ...
+
+async fn read(table_name: &str) -> Result<Vec<RecordBatch>, DeltaTableError> {
+    let delta_ops = get_delta_ops(table_name, true).await?;
+
+    let (_, stream) = delta_ops.load().await?;
+    let employee_records: Vec<RecordBatch> = collect_sendable_stream(stream).await?;
+
+    Ok(employee_records)
+}
+
+#[tokio::main()]
+async fn main() {
+    // ...
+
+    let table_name = "employee";
+    let employee_records = read(&table_name).await.expect("Read failed");
+    println!("employee_records: {:?}", employee_records);
+}
+```