Partitioning

rust/Cargo.toml

@@ -157,7 +157,7 @@ required-features = ["mmap"]
 default_categorical_8 = ["minarrow/default_categorical_8"]
 extended_categorical = ["default_categorical_8", "minarrow/extended_categorical"]
 large_string = ["minarrow/large_string"]
-datetime = ["minarrow/datetime"]
+datetime = ["minarrow/datetime", "minarrow/datetime_ops"]
 decimal = ["minarrow/decimal"]
 extended_numeric_types = ["minarrow/extended_numeric_types"]
 lbuffer = ["minarrow/lbuffer"]

rust/benches/file/chunked_throughput.rs

@@ -124,7 +124,7 @@ fn bench_chunked_arrow(c: &mut Criterion) {
     // throughput measurements.
     let read_dir = fresh_dir("arrow_read");
     {
-        let mut w = ChunkedArrowWriter::new(&read_dir, BASE).unwrap();
+        let mut w = ChunkedArrowWriter::new(&read_dir, BASE, None).unwrap();
         for t in &tables {
             w.write_chunk(t).unwrap();
         }
@@ -142,14 +142,14 @@ fn bench_chunked_arrow(c: &mut Criterion) {
     //
     // Directory cleanup runs in the next iteration's setup and is not timed.
     let write_serial_logical = |dir: PathBuf| {
-        let mut w = ChunkedArrowWriter::new(&dir, BASE).unwrap();
+        let mut w = ChunkedArrowWriter::new(&dir, BASE, None).unwrap();
         for t in &tables {
             w.write_chunk(t).unwrap();
         }
     };
 
     let write_serial_physical = |dir: PathBuf| {
-        let mut w = ChunkedArrowWriter::new(&dir, BASE).unwrap();
+        let mut w = ChunkedArrowWriter::new(&dir, BASE, None).unwrap();
         let mut paths: Vec<PathBuf> = Vec::with_capacity(N_CHUNKS);
 
         for t in &tables {
@@ -183,12 +183,12 @@ fn bench_chunked_arrow(c: &mut Criterion) {
     });
 
     let par_write_logical = |dir: PathBuf| {
-        let w = ChunkedArrowWriter::new(&dir, BASE).unwrap();
+        let w = ChunkedArrowWriter::new(&dir, BASE, None).unwrap();
         let paths = w.par_write_all(&table_refs, None).unwrap();
         assert_eq!(paths.len(), N_CHUNKS);
     };
     let par_write_physical = |dir: PathBuf| {
-        let w = ChunkedArrowWriter::new(&dir, BASE).unwrap();
+        let w = ChunkedArrowWriter::new(&dir, BASE, None).unwrap();
         let paths = w.par_write_all(&table_refs, None).unwrap();
         assert_eq!(paths.len(), N_CHUNKS);
         for p in &paths {
@@ -269,7 +269,7 @@ fn bench_chunked_parquet(c: &mut Criterion) {
     // Pre-write the dataset once to learn the real physical byte count.
     let read_dir = fresh_dir("parquet_read");
     {
-        let mut w = ChunkedParquetWriter::new(&read_dir, BASE, None).unwrap();
+        let mut w = ChunkedParquetWriter::new(&read_dir, BASE, None, None).unwrap();
         for t in &tables {
             w.write_chunk(t).unwrap();
         }
@@ -285,13 +285,13 @@ fn bench_chunked_parquet(c: &mut Criterion) {
     // dir. Denominator = output file bytes. Cleanup of the chunk dir
     // is NOT in the timed region.
     let write_serial_logical = |dir: PathBuf| {
-        let mut w = ChunkedParquetWriter::new(&dir, BASE, None).unwrap();
+        let mut w = ChunkedParquetWriter::new(&dir, BASE, None, None).unwrap();
         for t in &tables {
             w.write_chunk(t).unwrap();
         }
     };
     let write_serial_physical = |dir: PathBuf| {
-        let mut w = ChunkedParquetWriter::new(&dir, BASE, None).unwrap();
+        let mut w = ChunkedParquetWriter::new(&dir, BASE, None, None).unwrap();
         let mut paths: Vec<PathBuf> = Vec::with_capacity(N_CHUNKS);
         for t in &tables {
             paths.push(w.write_chunk(t).unwrap());
@@ -321,12 +321,12 @@ fn bench_chunked_parquet(c: &mut Criterion) {
     });
 
     let par_write_logical = |dir: PathBuf| {
-        let w = ChunkedParquetWriter::new(&dir, BASE, None).unwrap();
+        let w = ChunkedParquetWriter::new(&dir, BASE, None, None).unwrap();
         let paths = w.par_write_all(&table_refs, None).unwrap();
         assert_eq!(paths.len(), N_CHUNKS);
     };
     let par_write_physical = |dir: PathBuf| {
-        let w = ChunkedParquetWriter::new(&dir, BASE, None).unwrap();
+        let w = ChunkedParquetWriter::new(&dir, BASE, None, None).unwrap();
         let paths = w.par_write_all(&table_refs, None).unwrap();
         assert_eq!(paths.len(), N_CHUNKS);
         for p in &paths {
@@ -405,7 +405,7 @@ fn bench_chunked_csv(c: &mut Criterion) {
     // Pre-write the dataset once to learn the real physical byte count.
     let read_dir = fresh_dir("csv_read");
     {
-        let mut w = ChunkedCsvWriter::new(&read_dir, BASE, CsvEncodeOptions::default()).unwrap();
+        let mut w = ChunkedCsvWriter::new(&read_dir, BASE, CsvEncodeOptions::default(), None).unwrap();
         for t in &tables {
             w.write_chunk(t).unwrap();
         }
@@ -420,14 +420,14 @@ fn bench_chunked_csv(c: &mut Criterion) {
     // Physical writes synchronise each chunk and the parent directory, and use the
     // encoded file size. Directory cleanup runs outside the timed region.
     let write_serial_logical = |dir: PathBuf| {
-        let mut w = ChunkedCsvWriter::new(&dir, BASE, CsvEncodeOptions::default()).unwrap();
+        let mut w = ChunkedCsvWriter::new(&dir, BASE, CsvEncodeOptions::default(), None).unwrap();
         for t in &tables {
             w.write_chunk(t).unwrap();
         }
     };
 
     let write_serial_physical = |dir: PathBuf| {
-        let mut w = ChunkedCsvWriter::new(&dir, BASE, CsvEncodeOptions::default()).unwrap();
+        let mut w = ChunkedCsvWriter::new(&dir, BASE, CsvEncodeOptions::default(), None).unwrap();
         let mut paths: Vec<PathBuf> = Vec::with_capacity(N_CHUNKS);
         for t in &tables {
             paths.push(w.write_chunk(t).unwrap());
@@ -457,12 +457,12 @@ fn bench_chunked_csv(c: &mut Criterion) {
     });
 
     let par_write_logical = |dir: PathBuf| {
-        let w = ChunkedCsvWriter::new(&dir, BASE, CsvEncodeOptions::default()).unwrap();
+        let w = ChunkedCsvWriter::new(&dir, BASE, CsvEncodeOptions::default(), None).unwrap();
         let paths = w.par_write_all(&table_refs, None).unwrap();
         assert_eq!(paths.len(), N_CHUNKS);
     };
     let par_write_physical = |dir: PathBuf| {
-        let w = ChunkedCsvWriter::new(&dir, BASE, CsvEncodeOptions::default()).unwrap();
+        let w = ChunkedCsvWriter::new(&dir, BASE, CsvEncodeOptions::default(), None).unwrap();
         let paths = w.par_write_all(&table_refs, None).unwrap();
         assert_eq!(paths.len(), N_CHUNKS);
         for p in &paths {

rust/scripts/run-object-storage-s3-tests.sh

@@ -23,4 +23,4 @@ export AWS_REGION="${AWS_REGION:-$(aws configure get region || true)}"
 : "${AWS_REGION:?set AWS_REGION to the region of the bucket}"
 
 echo "Running object storage tests against ${LIGHTSTREAM_S3_URL} in ${AWS_REGION}"
-cargo test --features "aws parquet csv json" --test object_storage_s3 -- --ignored "$@"
+cargo test --features "aws datetime parquet csv json" --test object_storage_s3 -- --ignored "$@"

rust/src/lib.rs

@@ -250,7 +250,8 @@ pub mod models {
         pub mod parquet;
 
         /// Chunked file readers - glob a directory of `<base>-NNNNN.<ext>`
-        /// files and present them as an ordered iterator of `Table`s.
+        /// files, including its partition directories, and present them as
+        /// an ordered iterator of `Table`s.
         pub mod chunked {
             /// Chunked CSV reader.
             #[cfg(feature = "csv")]
@@ -360,6 +361,9 @@ pub mod models {
             /// Chunked Arrow IPC writer. Holds a tokio runtime internally
             /// to drive the async IPC file writer.
             pub mod arrow;
+
+            /// Partition paths for the chunked writers.
+            pub mod partition;
         }
 
         /// TCP table writer