Prepare initial release

Author: nevi-me

Cargo.toml

@@ -5,8 +5,7 @@ authors = ["Neville Dipale <[email protected]>"]
 edition = "2018"
 
 [dependencies]
-bson = { version = "0.14", features = ["decimal128"] }
-mongodb = { git = "https://github.com/nevi-me/mongo-rust-driver", branch = "decimal-128-hack" }
+bson = "0.14"
+mongodb = "0.9.2"
 arrow = "0.16.0"
-mongodb-schema-parser = { git = "https://github.com/nevi-me/mongodb-schema-parser", branch = "write-bson", default-features = false }
 chrono = "0.4"

README.md

@@ -11,6 +11,7 @@ We are curently writing this library due to a need to read MongoDB data into dat
 ## Features
 
 - [X] Read from a collection to batches
+- [X] Write from batches to a collection
 - [ ] Infer collection schema
 - [ ] Projection predicate push-down
 - [ ] Data types

src/lib.rs

@@ -1,5 +1,12 @@
+//!
+//! MongoDB to Apache Arrow Connector
+//!
+//! This crate allows reading and writing MongoDB data in the Apache Arrow format.
+//! Data is read as `RecordBatch`es from a MongoDB database using the aggregation
+//! framework.
+//! Arrow `RecordBatch`es are written to MongoDB using an insert_many into a collection.
+
+/// MongoDB reader
 pub mod reader;
+/// MongoDB writer
 pub mod writer;
-
-#[macro_use(bson)]
-extern crate bson;

src/reader.rs

@@ -10,22 +10,35 @@ use mongodb::{
     options::{AggregateOptions, ClientOptions, StreamAddress},
     Client,
 };
-// use mongodb_schema_parser::SchemaParser;
 
+/// Configuration for the MongoDB reader
 pub struct ReaderConfig<'a> {
+    /// The hostname to connect to
     pub hostname: &'a str,
+    /// An optional port, defaults to 27017
     pub port: Option<u16>,
-    // read_preference,
+    /// The name of the database to read from
     pub database: &'a str,
+    /// The name of the collection to read from
     pub collection: &'a str,
 }
 
+/// Database reader
 pub struct Reader {
+    /// The MongoDB client, with a connection established
     client: Client,
+    /// The name of the database to read from
     database: String,
+    /// The name of the collection to read from
     collection: String,
+    /// The schema of the data to read
     schema: Schema,
+    /// An internal tracker of the current index that has been read
     current_index: usize,
+    /// The preferred batch size per document.
+    /// If the documents being read are fairly small, or can fit in memory,
+    ///   a larger batch size is more performant as it would result in
+    ///   less roundtrips to the database.
     batch_size: usize,
 }
 

src/writer.rs

@@ -11,28 +11,49 @@ use mongodb::{
     Client,
 };
 
+/// Configuration for the MongoDB writer
 pub struct WriterConfig<'a> {
+    /// The hostname to connect to
     pub hostname: &'a str,
+    /// An optional port, defaults to 27017
     pub port: Option<u16>,
+    /// The name of the database to write to
     pub database: &'a str,
+    /// The name of the collection to write to
     pub collection: &'a str,
+    /// The write mode, whether an existing collection should
+    ///  be appended to or overwritten
     pub write_mode: WriteMode,
+    /// Whether compatible types should be coerced, for example
+    ///  an Int8 type will be written to an Int32 as BSON doesn't have Int8
     pub coerce_types: bool,
 }
 
+/// The mode to write to the collection in
 pub enum WriteMode {
+    /// Do not drop collection, but append to an existing collection.
+    /// If the collection does not exist, a new one is created.
     Append,
+    /// Try to drop the collection if it exists.
+    /// MongoDB returns an error if a collection that does not exist
+    ///  is dropped. We log this to the console, but do not return an error.
     Overwrite,
 }
 
+/// Database writer
 pub struct Writer {
+    /// The MongoDB client, with a connection established
     client: Client,
+    /// The name of the database to write to
     database: String,
+    /// The name of the collection to write to
     collection: String,
+    /// The schema of the data to write
     schema: Schema,
 }
 
 impl Writer {
+    /// Try to create a new writer, with provided writer options and a schema
     pub fn try_new(config: &WriterConfig, schema: Schema) -> Result<Self, ()> {
         // check if data types can be written
         Writer::check_supported_schema(schema.fields(), config.coerce_types)?;
@@ -63,6 +84,7 @@ impl Writer {
         })
     }
 
+    /// Write a batch to the database
     pub fn write(&self, batch: &RecordBatch) -> Result<(), ()> {
         if batch.schema().as_ref() != &self.schema {
             eprintln!("Schema of record batch does not match writer");
@@ -140,6 +162,7 @@ impl Writer {
     }
 }
 
+/// A private struct that uses a newtype pattern, holds the documents to be written
 struct Documents(Vec<bson::Document>);
 
 impl From<&RecordBatch> for Documents {