Repository: oom-ai/oomstore
Pull request #1172 (2022-02-10 08:22:20)

docs(sdk/rust): add rust docs


Patch:
diff --git a/internal/database/dbutil/data_loader.go b/internal/database/dbutil/data_loader.go
index 1b92d131..1d48757d 100644
--- a/internal/database/dbutil/data_loader.go
+++ b/internal/database/dbutil/data_loader.go
@@ -73,8 +73,6 @@ func ReadLine(opt ReadLineOpt) ([]interface{}, error) {
 	// so here we use csv.Reader to read data directly.
 	// If there are other data formats in the future, we need to choose different decoders according to the different formats.
 	reader := csv.NewReader(opt.Source.Reader)
-	reader.Comma = opt.Source.Delimiter
-
 	row, err := reader.Read()
 	if err != nil {
 		return nil, err
diff --git a/internal/database/offline/types.go b/internal/database/offline/types.go
index 1f199023..9b04c439 100644
--- a/internal/database/offline/types.go
+++ b/internal/database/offline/types.go
@@ -51,7 +51,7 @@ type PushOpt struct {
 
 type CSVSource struct {
 	Reader    *bufio.Reader
-	Delimiter rune
+	Delimiter string
 }
 
 type SnapshotOpt struct {
diff --git a/oomagent/server.go b/oomagent/server.go
index 5569d7d6..2ef3d00e 100644
--- a/oomagent/server.go
+++ b/oomagent/server.go
@@ -97,12 +97,11 @@ func (s *server) ChannelImport(stream codegen.OomAgent_ChannelImportServer) erro
 	if firstReq.Group == nil {
 		return status.Errorf(codes.InvalidArgument, "group is required in first request")
 	}
-	var description string
-	var delimiter rune
+	var description, delimiter string
 	if firstReq.Description != nil {
 		description = *firstReq.Description
 	}
-	delimiter = ','
+	delimiter = ","
 	reader, writer := io.Pipe()
 
 	go func() {
@@ -166,16 +165,14 @@ func (s *server) Snapshot(ctx context.Context, re *codegen.SnapshotRequest) (*co
 }
 
 func (s *server) Import(ctx context.Context, req *codegen.ImportRequest) (*codegen.ImportResponse, error) {
-	var description string
-	var delimiter rune
-
+	var description, delimiter string
 	if req.Description != nil {
 		description = *req.Description
 	}
 	if req.Delimiter != nil {
-		delimiter = []rune(*req.Delimiter)[0]
+		delimiter = *req.Delimiter
 	} else {
-		delimiter = ','
+		delimiter = ","
 	}
 	revisionID, err := s.oomstore.Import(ctx, types.ImportOpt{
 		GroupName:      req.Group,
diff --git a/oomcli/cmd/import.go b/oomcli/cmd/import.go
index bd9d19c1..c500d3de 100644
--- a/oomcli/cmd/import.go
+++ b/oomcli/cmd/import.go
@@ -12,14 +12,11 @@ import (
 var importOpt types.ImportOpt
 var importCSVFileDataSource types.CsvFileDataSource
 var importTableLinkDataSource types.TableLinkDataSource
-var delimitre string
 
 var importCmd = &cobra.Command{
 	Use:   "import",
 	Short: "Import feature data from a data source",
 	PreRunE: func(cmd *cobra.Command, args []string) error {
-		importCSVFileDataSource.Delimiter = []rune(delimitre)[0]
-
 		if !cmd.Flags().Changed("revision") {
 			importOpt.Revision = nil
 		}
@@ -64,6 +61,6 @@ func init() {
 	flags.StringVar(&importCSVFileDataSource.InputFilePath, "input-file", "", "input csv file")
 	flags.StringVar(&importTableLinkDataSource.TableName, "table-link", "", "link to a existing data table")
 
-	flags.StringVar(&delimitre, "delimiter", ",", "specify field delimiter")
+	flags.StringVar(&importCSVFileDataSource.Delimiter, "delimiter", ",", "specify field delimiter")
 	importOpt.Revision = flags.Int64P("revision", "r", 0, "user-defined revision")
 }
diff --git a/pkg/oomstore/types/import.go b/pkg/oomstore/types/import.go
index 003fd1b1..c0f2f43b 100644
--- a/pkg/oomstore/types/import.go
+++ b/pkg/oomstore/types/import.go
@@ -23,12 +23,12 @@ type ImportOpt struct {
 
 type CsvReaderDataSource struct {
 	Reader    io.Reader
-	Delimiter rune
+	Delimiter string
 }
 
 type CsvFileDataSource struct {
 	InputFilePath string
-	Delimiter     rune
+	Delimiter     string
 }
 
 type TableLinkDataSource struct {
diff --git a/sdk/rust/Cargo.toml b/sdk/rust/Cargo.toml
index 999bf01b..f6b60fc5 100644
--- a/sdk/rust/Cargo.toml
+++ b/sdk/rust/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "oomclient"
-version = "0.1.0-rc.8"
+version = "0.1.0-rc.9"
 edition = "2021"
 authors = ["Wenxuan Zhang <wenxuangm@gmail.com>"]
 description = "oomstore client for rust"
diff --git a/sdk/rust/src/client.rs b/sdk/rust/src/client.rs
index 638461d1..3c936970 100644
--- a/sdk/rust/src/client.rs
+++ b/sdk/rust/src/client.rs
@@ -30,6 +30,8 @@ use futures_core::stream::Stream;
 use std::{collections::HashMap, path::Path, sync::Arc};
 use tonic::{codegen::StdError, transport, Request};
 
+/// A rust client for [oomstore](https://github.com/oom-ai/oomstore),
+/// using the grpc protocol to communicate with oomagent server under the hood.
 #[derive(Debug, Clone)]
 pub struct Client {
     client: OomAgentClient<transport::Channel>,
@@ -38,6 +40,7 @@ pub struct Client {
 
 // TODO: Add a Builder to create the client
 impl Client {
+    /// Connect to an oomagent instance running on the given endpoint.
     pub async fn connect<D>(dst: D) -> Result<Self>
     where
         D: std::convert::TryInto<tonic::transport::Endpoint>,
@@ -46,6 +49,7 @@ impl Client {
         Ok(Self { client: OomAgentClient::connect(dst).await?, _agent: None })
     }
 
+    /// Connect to an oomagent instance embedded with the client.
     pub async fn with_embedded_oomagent<P1, P2>(bin_path: Option<P1>, cfg_path: Option<P2>) -> Result<Self>
     where
         P1: AsRef<Path>,
@@ -58,14 +62,17 @@ impl Client {
         })
     }
 
+    /// Connect to an embedded oomagent instance with default config.
     pub async fn with_default_embedded_oomagent() -> Result<Self> {
         Self::with_embedded_oomagent(None::<String>, None::<String>).await
     }
 
+    /// Check if oomagent is ready to serve requests.
     pub async fn health_check(&mut self) -> Result<()> {
         Ok(self.client.health_check(HealthCheckRequest {}).await.map(|_| ())?)
     }
 
+    /// Get online features for an entity without further conversion.
     pub async fn online_get_raw(
         &mut self,
         entity_key: impl Into<String>,
@@ -82,6 +89,7 @@ impl Client {
         })
     }
 
+    /// Get online features for an entity.
     pub async fn online_get(
         &mut self,
         key: impl Into<String>,
@@ -91,6 +99,7 @@ impl Client {
         Ok(parse_raw_feature_values(rs))
     }
 
+    /// Get online features for multiple entities without further conversion.
     pub async fn online_multi_get_raw(
         &mut self,
         entity_keys: Vec<String>,
@@ -104,6 +113,7 @@ impl Client {
         Ok(res.result)
     }
 
+    /// Get online features for multiple entities.
     pub async fn online_multi_get(
         &mut self,
         keys: Vec<String>,
@@ -113,6 +123,7 @@ impl Client {
         Ok(rs.into_iter().map(|(k, v)| (k, parse_raw_feature_values(v))).collect())
     }
 
+    /// Sync a certain revision of batch features from offline to online store.
     pub async fn sync(
         &mut self,
         group: impl Into<String>,
@@ -128,6 +139,7 @@ impl Client {
         Ok(())
     }
 
+    /// Import features from external (batch and stream) data sources to offline store through channels.
     pub async fn channel_import(
         &mut self,
         group: impl Into<String>,
@@ -147,6 +159,7 @@ impl Client {
         Ok(res.revision_id as u32)
     }
 
+    /// Import features from external (batch and stream) data sources to offline store through files.
     pub async fn import(
         &mut self,
         group: impl Into<String>,
@@ -169,6 +182,7 @@ impl Client {
         Ok(res.revision_id as u32)
     }
 
+    /// Push stream features from stream data source to both offline and online stores.
     pub async fn push(
         &mut self,
         entity_key: impl Into<String>,
@@ -190,6 +204,7 @@ impl Client {
         Ok(())
     }
 
+    /// Point-in-Time Join features against labeled entity rows through channels.
     pub async fn channel_join(
         &mut self,
         join_features: Vec<String>,
@@ -230,6 +245,7 @@ impl Client {
         Ok((header, outbound))
     }
 
+    /// Point-in-Time Join features against labeled entity rows through files.
     pub async fn join(
         &mut self,
         features: Vec<String>,
@@ -246,6 +262,7 @@ impl Client {
         Ok(())
     }
 
+    /// Export certain features to a channel.
     pub async fn channel_export(
         &mut self,
         features: Vec<String>,
@@ -275,6 +292,7 @@ impl Client {
         Ok((header, outbound))
     }
 
+    /// Export certain features to a file.
     pub async fn export(
         &mut self,
         features: Vec<String>,
@@ -291,6 +309,7 @@ impl Client {
         Ok(())
     }
 
+    /// Take snapshot for a stream feature group in offline store.
     pub async fn snapshot(&mut self, group: impl Into<String>) -> Result<()> {
         self.client.snapshot(SnapshotRequest { group: group.into() }).await?;
         Ok(())
diff --git a/sdk/rust/src/error.rs b/sdk/rust/src/error.rs
index b06272c8..294108ea 100644
--- a/sdk/rust/src/error.rs
+++ b/sdk/rust/src/error.rs
@@ -1,6 +1,7 @@
 use std::{io, net::AddrParseError, num};
 use thiserror::Error;
 
+/// An error originating from the oomclient or it's dependencies.
 #[derive(Error, Debug)]
 pub enum OomError {
     #[error("a grpc status describing the result of an rpc call")]
diff --git a/sdk/rust/src/lib.rs b/sdk/rust/src/lib.rs
index ba8d7b4f..974a56ee 100644
--- a/sdk/rust/src/lib.rs
+++ b/sdk/rust/src/lib.rs
@@ -1,3 +1,26 @@
+//! This crate provides an easy-to-use client for [oomstore](https://github.com/oom-ai/oomstore), a
+//! lightweight and fast feature store powered by go.
+//!
+//! ## Examples
+//!
+//! ```rust,no_run
+//! use oomclient::Client;
+//!
+//! #[tokio::main]
+//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
+//!     let mut client = Client::connect("http://localhost:50051").await?;
+//!
+//!     let features = vec!["account.state".into(), "txn_stats.count_7d".into()];
+//!     let response = client.online_get_raw("48", features.clone()).await?;
+//!     println!("RESPONSE={:#?}", response);
+//!
+//!     let response = client.online_get("48", features).await?;
+//!     println!("RESPONSE={:#?}", response);
+//!
+//!     Ok(())
+//! }
+//! ```
+
 mod client;
 mod error;
 mod server;
@@ -9,7 +32,13 @@ mod oomagent {
 
 pub use client::Client;
 pub use error::OomError;
-pub use oomagent::{value::Value, EntityRow};
+pub use oomagent::EntityRow;
 pub use server::ServerWrapper;
 
+/// Represents a dynamically typed value which can be either
+/// an int64, a double, a string, a bool, a unix milliseconds, or a
+/// bytes array.
+pub type Value = oomagent::value::Value;
+
+/// A result holding an Error.
 pub type Result<T> = std::result::Result<T, OomError>;
diff --git a/sdk/rust/src/server.rs b/sdk/rust/src/server.rs
index 4840ee4d..5e9d3774 100644
--- a/sdk/rust/src/server.rs
+++ b/sdk/rust/src/server.rs
@@ -14,6 +14,7 @@ use tokio_stream::StreamExt;
 
 use crate::Result;
 
+/// A rust oomagent server wrapper, mainly for [oomclient](crate::Client) with embedded oomagent server.
 #[derive(Debug, Clone)]
 pub struct ServerWrapper {
     addr:   SocketAddr,
@@ -21,8 +22,8 @@ pub struct ServerWrapper {
     pid:    u32,
 }
 
-// TODO: Add a Builder to create the server wrapper
 impl ServerWrapper {
+    // TODO: Add a Builder to create the server wrapper
     pub async fn new<P1, P2>(bin_path: Option<P1>, cfg_path: Option<P2>, port: Option<u16>) -> Result<Self>
     where
         P1: AsRef<Path>,


Test patch:
diff --git a/internal/database/offline/test_impl/export.go b/internal/database/offline/test_impl/export.go
index 1ad95865..18174bd2 100644
--- a/internal/database/offline/test_impl/export.go
+++ b/internal/database/offline/test_impl/export.go
@@ -29,7 +29,7 @@ func TestExport(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyS
 1236,huawei,300
 1237,oneplus,240
 `)),
-		Delimiter: ',',
+		Delimiter: ",",
 	})
 	buildTestSnapshotTable(ctx, t, store, streamFeatures, 1, streamSnapshotTable, &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,1000,true
@@ -37,7 +37,7 @@ func TestExport(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyS
 1236,1560,true
 1237,4000,false
 `)),
-		Delimiter: ',',
+		Delimiter: ",",
 	})
 	buildTestSnapshotTable(ctx, t, store, append(streamFeatures, unixMilli), 1, streamCdcTable, &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,1200,true,2
@@ -46,7 +46,7 @@ func TestExport(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyS
 1238,1500,true,11
 1239,2700,false,12
 `)),
-		Delimiter: ',',
+		Delimiter: ",",
 	})
 
 	testCases := []struct {
diff --git a/internal/database/offline/test_impl/import.go b/internal/database/offline/test_impl/import.go
index 235d35aa..a01d729e 100644
--- a/internal/database/offline/test_impl/import.go
+++ b/internal/database/offline/test_impl/import.go
@@ -37,7 +37,7 @@ func TestImport(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyS
 1236,huawei,5999
 1237,oneplus,3999
 `)),
-			Delimiter: ',',
+			Delimiter: ",",
 		},
 	}
 
diff --git a/internal/database/offline/test_impl/join.go b/internal/database/offline/test_impl/join.go
index 008387df..00b609c6 100644
--- a/internal/database/offline/test_impl/join.go
+++ b/internal/database/offline/test_impl/join.go
@@ -35,19 +35,19 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 		Reader: bufio.NewReader(strings.NewReader(`1234,xiaomi,100
 1235,apple,200
 `)),
-		Delimiter: ',',
+		Delimiter: ",",
 	})
 	buildTestSnapshotTable(ctx, t, store, oneGroupFeatures, 2, "offline_snapshot_1_2", &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,galaxy,300
 1235,oneplus,240
 `)),
-		Delimiter: ',',
+		Delimiter: ",",
 	})
 	buildTestSnapshotTable(ctx, t, store, twoGroupFeatures[2:], 1, "offline_snapshot_2_1", &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,1
 1235,0
 `)),
-		Delimiter: ',',
+		Delimiter: ",",
 	})
 
 	buildTestSnapshotTable(ctx, t, store, append(oneGroupFeatures, unixMilli), 1, "offline_cdc_1_1", &offline.CSVSource{
@@ -55,12 +55,12 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 1235,apple-2,115,14
 1234,xiaomi-1,130,10
 `)),
-		Delimiter: ',',
+		Delimiter: ",",
 	})
 	buildTestSnapshotTable(ctx, t, store, append(oneGroupFeatures, unixMilli), 2, "offline_cdc_1_2", &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,galaxy-1,320,18
 `)),
-		Delimiter: ',',
+		Delimiter: ",",
 	})
 
 	testCases := []struct {
diff --git a/internal/database/offline/test_impl/snapshot.go b/internal/database/offline/test_impl/snapshot.go
index e765bc74..5b39645d 100644
--- a/internal/database/offline/test_impl/snapshot.go
+++ b/internal/database/offline/test_impl/snapshot.go
@@ -33,7 +33,7 @@ func TestSnapshot(t *testing.T, prepareStore PrepareStoreFn, destroyStore Destro
 1235,apple,200,5
 1236,oneplus,155,8
 `)),
-		Delimiter: ',',
+		Delimiter: ",",
 	})
 
 	buildTestSnapshotTable(ctx, t, store, append(features, unixMilli), 2, "offline_stream_cdc_1_2", &offline.CSVSource{
@@ -43,7 +43,7 @@ func TestSnapshot(t *testing.T, prepareStore PrepareStoreFn, destroyStore Destro
 1237,pixel,200,11
 1235,apple-1,113,15
 `)),
-		Delimiter: ',',
+		Delimiter: ",",
 	})
 
 	err := store.Snapshot(ctx, offline.SnapshotOpt{