Initial version of read_delta

Author: Bidek56

Cargo.toml

@@ -27,6 +27,7 @@ thiserror = "1"
 smartstring = { version = "1" }
 serde_json = { version = "1" }
 either = "1.9"
+deltalake = "0.17.1"
 
 [dependencies.polars]
 features = [

__tests__/examples/delta/sample.table/0-07808ab0-6b13-421c-bb67-ca09945eb281-0.parquet

@@ -0,0 +1,4 @@
+{"protocol":{"minReaderVersion":1,"minWriterVersion":2}}
+{"metaData":{"id":"b27252b0-26cc-49c5-b79d-12dd6647ffba","name":null,"description":null,"format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"foo\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"fruits\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"B\",\"type\":\"long\",\"nullable\":true,\"metadata\":{}},{\"name\":\"cars\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"createdTime":1711300448856,"configuration":{}}}
+{"add":{"path":"0-07808ab0-6b13-421c-bb67-ca09945eb281-0.parquet","partitionValues":{},"size":1550,"modificationTime":1711300448855,"dataChange":true,"stats":"{\"numRecords\": 5, \"minValues\": {\"foo\": \"5\", \"fruits\": \"apple\", \"B\": 1, \"cars\": \"audi\"}, \"maxValues\": {\"foo\": \"you have a,b,c\", \"fruits\": \"banana\", \"B\": 5, \"cars\": \"beetle\"}, \"nullCount\": {\"foo\": 0, \"fruits\": 0, \"B\": 0, \"cars\": 0}}","tags":null,"deletionVector":null,"baseRowId":null,"defaultRowCommitVersion":null,"clusteringProvider":null}}
+{"commitInfo":{"timestamp":1711300448856,"operation":"CREATE TABLE","operationParameters":{"location":"file:///examples/delta/sample.table","protocol":"{\"minReaderVersion\":1,\"minWriterVersion\":2}","metadata":"{\"configuration\":{},\"createdTime\":1711300448856,\"description\":null,\"format\":{\"options\":{},\"provider\":\"parquet\"},\"id\":\"b27252b0-26cc-49c5-b79d-12dd6647ffba\",\"name\":null,\"partitionColumns\":[],\"schemaString\":\"{\\\"type\\\":\\\"struct\\\",\\\"fields\\\":[{\\\"name\\\":\\\"foo\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}},{\\\"name\\\":\\\"fruits\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}},{\\\"name\\\":\\\"B\\\",\\\"type\\\":\\\"long\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}},{\\\"name\\\":\\\"cars\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}]}\"}","mode":"ErrorIfExists"},"clientVersion":"delta-rs.0.17.1"}}

__tests__/examples/delta/sample.table/_delta_log/00000000000000000000.json

@@ -18,6 +18,8 @@ const ipcpath = path.resolve(__dirname, "./examples/foods.ipc");
 const jsonpath = path.resolve(__dirname, "./examples/foods.json");
 // eslint-disable-next-line no-undef
 const singlejsonpath = path.resolve(__dirname, "./examples/single_foods.json");
+// eslint-disable-next-line no-undef
+const deltapath = path.resolve(__dirname, "./examples/delta/sample.table");
 describe("read:csv", () => {
   it("can read from a csv file", () => {
     const df = pl.readCSV(csvpath);
@@ -235,6 +237,13 @@ describe("scan", () => {
   });
 });
 
+describe("delta", () => {
+  test("delta:scan", async () => {
+    const df = await pl.readDelta(deltapath).collect();
+    expect(df.shape).toEqual({ height: 5, width: 4 });
+  });
+});
+
 describe("parquet", () => {
   beforeEach(() => {
     pl.readCSV(csvpath).writeParquet(parquetpath);

__tests__/io.test.ts

@@ -1,5 +1,5 @@
 {
-  "$schema": "https://biomejs.dev/schemas/1.0.0/schema.json",
+  "$schema": "https://biomejs.dev/schemas/1.6.2/schema.json",
   "organizeImports": {
     "enabled": false
   },
@@ -24,7 +24,8 @@
     "ignore": [
       "polars/native-polars.js",
       "./docs/*",
-      "./bin/*"
+      "./bin/*",
+      "**/delta/sample.table/*"
     ]
   },
   "formatter": {
@@ -33,7 +34,8 @@
     "ignore": [
       "polars/native-polars.js",
       "./docs/*",
-      "./bin/*"
+      "./bin/*",
+      "**/delta/sample.table/*"
     ]
   },
   "javascript": {

biome.json

@@ -54,7 +54,7 @@
     "precommit": "yarn lint && yarn test"
   },
   "devDependencies": {
-    "@biomejs/biome": "^1.6.1",
+    "@biomejs/biome": "^1.6.2",
     "@napi-rs/cli": "^2.18.0",
     "@types/chance": "^1.1.6",
     "@types/jest": "^29.5.12",
@@ -65,7 +65,7 @@
     "ts-jest": "^29.1.2",
     "ts-node": "^10.9.2",
     "typedoc": "^0.25.12",
-    "typescript": "5.4.2"
+    "typescript": "5.4.3"
   },
   "packageManager": "[email protected]",
   "workspaces": [

bun.lockb

@@ -47,6 +47,7 @@ export namespace pl {
   export import readJSON = io.readJSON;
   export import readParquet = io.readParquet;
   export import readAvro = io.readAvro;
+  export import readDelta = io.readDelta;
 
   export import readCSVStream = io.readCSVStream;
   export import readJSONStream = io.readJSONStream;

package.json

@@ -436,6 +436,32 @@ export function readParquet(
   throw new Error("must supply either a path or body");
 }
 
+interface ReadDeltaOptions {
+  version?: number;
+  columns?: string[];
+  parallel?: "auto" | "columns" | "row_groups" | "none";
+}
+
+/**
+ * Reads into a DataFrame from a Delta lake table.
+ * @param path - Path or URI to the root of the Delta lake table.
+ * @param option.version - Version of the Delta lake table. Note: If `version` is not provided, the latest version of delta lake
+ *   table is read.
+ * @param option.columns - Columns to select. Accepts a list of column names.
+ * @param options.parallel - Any of  'auto' | 'columns' |  'row_groups' | 'none'
+ *     This determines the direction of parallelism. 'auto' will try to determine the optimal direction.
+ *     Defaults to 'auto'
+ */
+
+export function readDelta(
+  path: string,
+  options: ReadDeltaOptions = {},
+): LazyDataFrame {
+  const parallel = options?.parallel ?? "auto";
+  options = { parallel, ...options };
+  return _LazyDataFrame(pli.readDelta(path, options));
+}
+
 export interface ReadAvroOptions {
   columns: string[] | Array<string> | number[];
   projection: number;

polars/index.ts

@@ -1,9 +1,12 @@
+use crate::delta::read_delta_table;
+use crate::export::JsLazyFrame;
 use crate::file::*;
 use crate::prelude::*;
 use crate::series::JsSeries;
 use napi::JsUnknown;
 use polars::frame::row::{infer_schema, Row};
 use polars::frame::NullStrategy;
+use polars_io::pl_async::get_runtime;
 use polars_io::RowIndex;
 
 use std::borrow::Borrow;
@@ -271,6 +274,29 @@ pub fn read_json(
     Ok(df.into())
 }
 
+#[napi(object)]
+pub struct ReadDeltaOptions {
+    pub version: Option<i64>,
+    pub columns: Option<Vec<String>>,
+    pub parallel: Wrap<ParallelStrategy>,
+}
+
+#[napi(catch_unwind)]
+pub fn read_delta(path: String,
+    options: ReadDeltaOptions
+) -> napi::Result<JsLazyFrame> {
+    let table: std::prelude::v1::Result<LazyFrame, deltalake::DeltaTableError> = get_runtime().block_on(async {
+        read_delta_table(&path, options).await
+    });
+
+    let ldf:LazyFrame = match table {
+        Ok(table) => table,
+        Err(err) => return Err(napi::Error::from_reason(err.to_string()))
+    };
+
+    Ok(LazyFrame::from(ldf).into())
+}
+
 #[napi(object)]
 pub struct ReadParquetOptions {
     pub columns: Option<Vec<String>>,

polars/io.ts

@@ -0,0 +1,52 @@
+use polars_core::prelude::DataFrame;
+use polars_io::{parquet::ParquetReader, SerReader};
+use polars_lazy::{dsl::UnionArgs, frame::{IntoLazy, LazyFrame}};
+use polars::prelude::concat;
+use deltalake::{DeltaTableBuilder, DeltaTableError};
+use crate::dataframe::ReadDeltaOptions;
+
+pub async fn read_delta_table(path: &str, 
+    options: ReadDeltaOptions,
+    ) -> Result<LazyFrame, DeltaTableError> {
+    let mut db = DeltaTableBuilder::from_uri(path)
+                .with_allow_http(false);
+
+    // if version specified, add it
+    if options.version.is_some() {
+        db = db.with_version(options.version.unwrap());
+    }
+
+    let dt = db.load().await?;
+
+    // show all active files in the table
+    let files: Vec<_> = dt.get_file_uris()?.collect();
+
+    let mut df_collection: Vec<DataFrame> = vec![];
+
+    for file in files.into_iter() {
+        let base = std::path::Path::new(path);
+        let file_path = std::path::Path::new(&file);
+        let full_path = base.join(file_path);
+        let mut file = std::fs::File::open(full_path).unwrap();
+
+        let columns = options.columns.clone();
+        let parallel = options.parallel.0;
+
+        let df = ParquetReader::new(&mut file)
+            .with_columns(columns)
+            .read_parallel(parallel)
+            .finish().unwrap();
+
+        df_collection.push(df);
+    }
+
+    let empty_head = df_collection[0].clone().lazy().limit(0);
+
+    Ok(df_collection.into_iter().fold(empty_head, |acc, df| concat([acc, df.lazy()], 
+            UnionArgs {
+                rechunk: false,
+                parallel: false,
+                ..Default::default()
+            }).unwrap()))
+
+}

src/dataframe.rs

@@ -28,6 +28,7 @@ pub mod series;
 pub mod set;
 pub mod utils;
 pub mod sql;
+pub mod delta;
 
 pub use polars_core;