Parse partition values before handing to PyArrow

python/Cargo.toml

@@ -29,4 +29,4 @@ features = ["extension-module", "abi3", "abi3-py36"]
 [dependencies.deltalake]
 path = "../rust"
 version = "0"
-features = ["s3", "azure", "glue", "gcs"]
+features = ["s3", "azure", "glue", "gcs", "python"]

python/deltalake/table.py

@@ -281,7 +281,9 @@ def to_pyarrow_dataset(
                 filesystem=filesystem,
                 partition_expression=part_expression,
             )
-            for file, part_expression in self._table.dataset_partitions(partitions)
+            for file, part_expression in self._table.dataset_partitions(
+                partitions, self.pyarrow_schema()
+            )
         ]
 
         return FileSystemDataset(fragments, self.pyarrow_schema(), format, filesystem)

python/src/lib.rs

@@ -248,6 +248,7 @@ impl RawDeltaTable {
         &mut self,
         py: Python<'py>,
         partition_filters: Option<Vec<(&str, &str, PartitionFilterValue)>>,
+        schema: ArrowSchema,
     ) -> PyResult<Vec<(String, Option<&'py PyAny>)>> {
         let path_set = match partition_filters {
             Some(filters) => Some(HashSet::<_>::from_iter(
@@ -266,7 +267,7 @@ impl RawDeltaTable {
             })
             .map(|((path, partition_values), stats)| {
                 let stats = stats.map_err(PyDeltaTableError::from_raw)?;
-                let expression = filestats_to_expression(py, partition_values, stats)?;
+                let expression = filestats_to_expression(py, &schema, partition_values, stats)?;
                 Ok((path, expression))
             })
             .collect()
@@ -297,19 +298,31 @@ fn json_value_to_py(value: &serde_json::Value, py: Python) -> PyObject {
 /// skipped during a scan.
 fn filestats_to_expression<'py>(
     py: Python<'py>,
+    schema: &ArrowSchema,
     partitions_values: &HashMap<String, Option<String>>,
     stats: Option<Stats>,
 ) -> PyResult<Option<&'py PyAny>> {
     let ds = PyModule::import(py, "pyarrow.dataset")?;
     let field = ds.getattr("field")?;
+    let pa = PyModule::import(py, "pyarrow")?;
     let mut expressions: Vec<PyResult<&PyAny>> = Vec::new();
 
     for (column, value) in partitions_values.iter() {
         if let Some(value) = value {
+            // value is a string, but needs to be parsed into appropriate type
+            let column_type = schema
+                .field_with_name(column)
+                .map_err(|_| PyDeltaTableError::new_err("Partition column not found in schema"))?
+                .data_type()
+                .clone()
+                .into_py(py);
+            let converted_value = pa
+                .call_method1("scalar", (value,))?
+                .call_method1("cast", (column_type,))?;
             expressions.push(
                 field
                     .call1((column,))?
-                    .call_method1("__eq__", (value.to_object(py),)),
+                    .call_method1("__eq__", (converted_value,)),
             );
         }
     }

python/tests/data/date_partitioned_df/_delta_log/00000000000000000000.json

@@ -0,0 +1,4 @@
+{"protocol":{"minReaderVersion":1,"minWriterVersion":2}}
+{"metaData":{"id":"588135b2-b298-4d9f-aab6-6dd9bf90d575","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"long\",\"nullable\":true,\"metadata\":{}},{\"name\":\"date\",\"type\":\"date\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["date"],"configuration":{},"createdTime":1645893400586}}
+{"add":{"path":"date=2021-01-01/part-00000-6ae76612-3903-4007-b5ec-511b932751ea.c000.snappy.parquet","partitionValues":{"date":"2021-01-01"},"size":500,"modificationTime":1645893404567,"dataChange":true}}
+{"commitInfo":{"timestamp":1645893404671,"operation":"WRITE","operationParameters":{"mode":"ErrorIfExists","partitionBy":"[\"date\"]"},"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{"numFiles":"1","numOutputRows":"5","numOutputBytes":"500"},"engineInfo":"Apache-Spark/3.2.1 Delta-Lake/1.1.0"}}

python/tests/data/timestamp_partitioned_df/_delta_log/00000000000000000000.json

@@ -0,0 +1,4 @@
+{"protocol":{"minReaderVersion":1,"minWriterVersion":2}}
+{"metaData":{"id":"e79e0060-0670-46ed-9e93-9dff7ac96b07","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"long\",\"nullable\":true,\"metadata\":{}},{\"name\":\"date\",\"type\":\"timestamp\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["date"],"configuration":{},"createdTime":1645893413372}}
+{"add":{"path":"date=2021-01-01%2000%253A00%253A00/part-00000-6177a755-69ce-44ea-bbfa-0479c4c1e704.c000.snappy.parquet","partitionValues":{"date":"2021-01-01 00:00:00"},"size":500,"modificationTime":1645893413662,"dataChange":true}}
+{"commitInfo":{"timestamp":1645893413668,"operation":"WRITE","operationParameters":{"mode":"ErrorIfExists","partitionBy":"[\"date\"]"},"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{"numFiles":"1","numOutputRows":"5","numOutputBytes":"500"},"engineInfo":"Apache-Spark/3.2.1 Delta-Lake/1.1.0"}}

python/tests/test_table_read.py

@@ -1,12 +1,14 @@
 import os
+from datetime import date
 from threading import Barrier, Thread
 
 import pandas as pd
+import pyarrow as pa
 import pyarrow.dataset as ds
 import pytest
 from pyarrow.fs import LocalFileSystem
 
-from deltalake import DeltaTable, Metadata
+from deltalake import DeltaTable
 
 
 def test_read_simple_table_to_dict():
@@ -93,6 +95,20 @@ def test_read_partitioned_table_to_dict():
     assert dt.to_pyarrow_dataset().to_table().to_pydict() == expected
 
 
+@pytest.mark.parametrize(
+    "name,partition_type", [("date", pa.date32()), ("timestamp", pa.timestamp("us"))]
+)
+def test_read_date_partitioned_table(name, partition_type):
+    table_path = f"tests/data/{name}_partitioned_df"
+    dt = DeltaTable(table_path)
+    table = dt.to_pyarrow_table()
+    assert table["date"].type == partition_type
+    date_expected = pa.array([date(2021, 1, 1)] * 5).cast(partition_type)
+    assert table["date"] == pa.chunked_array([date_expected])
+    id_expected = pa.array(range(5))
+    assert table["id"] == pa.chunked_array([id_expected])
+
+
 def test_read_partitioned_table_with_partitions_filters_to_dict():
     table_path = "../rust/tests/data/delta-0.8.0-partitioned"
     dt = DeltaTable(table_path)

rust/Cargo.toml

@@ -76,6 +76,7 @@ s3 = ["rusoto_core/native-tls", "rusoto_credential", "rusoto_s3/native-tls", "ru
 s3-rustls = ["rusoto_core/rustls", "rusoto_credential", "rusoto_s3/rustls", "rusoto_sts/rustls", "rusoto_dynamodb/rustls", "maplit", "dynamodb_lock/rustls"]
 gcs = ["async-stream", "tame-gcs", "tame-oauth", "reqwest"]
 glue = ["s3", "rusoto_glue"]
+python = ["arrow/pyarrow"]
 
 [build-dependencies]
 glibc_version = "0"