refactor(rust): Move new-streaming parquet and CSV sources to under `…

…io_sources/` (#20160)
pola-rs · Dec 5, 2024 · 300340a · 300340a
1 parent b019e42
commit 300340a
Show file tree

Hide file tree

Showing 11 changed files with 9 additions and 9 deletions.
diff --git a/crates/polars-stream/src/nodes/csv_source.rs → ...polars-stream/src/nodes/io_sources/csv.rs b/crates/polars-stream/src/nodes/csv_source.rs → ...polars-stream/src/nodes/io_sources/csv.rs
@@ -24,12 +24,12 @@ use polars_utils::mmap::MemSlice;
 use polars_utils::pl_str::PlSmallStr;
 use polars_utils::IdxSize;
 
-use super::compute_node_prelude::*;
-use super::{MorselSeq, TaskPriority};
 use crate::async_executor;
 use crate::async_primitives::connector::connector;
 use crate::async_primitives::wait_group::{IndexedWaitGroup, WaitToken};
 use crate::morsel::SourceToken;
+use crate::nodes::compute_node_prelude::*;
+use crate::nodes::{MorselSeq, TaskPriority};
 
 struct LineBatch {
     bytes: MemSlice,

diff --git a/crates/polars-stream/src/nodes/io_sources/mod.rs b/crates/polars-stream/src/nodes/io_sources/mod.rs
@@ -1 +1,4 @@
+pub mod csv;
 pub mod ipc;
+#[cfg(feature = "parquet")]
+pub mod parquet;
diff --git a/...s-stream/src/nodes/parquet_source/init.rs → ...ream/src/nodes/io_sources/parquet/init.rs b/...s-stream/src/nodes/parquet_source/init.rs → ...ream/src/nodes/io_sources/parquet/init.rs
diff --git a/...odes/parquet_source/mem_prefetch_funcs.rs → .../io_sources/parquet/mem_prefetch_funcs.rs b/...odes/parquet_source/mem_prefetch_funcs.rs → .../io_sources/parquet/mem_prefetch_funcs.rs
diff --git a/...rc/nodes/parquet_source/metadata_fetch.rs → ...odes/io_sources/parquet/metadata_fetch.rs b/...rc/nodes/parquet_source/metadata_fetch.rs → ...odes/io_sources/parquet/metadata_fetch.rs
diff --git a/...rc/nodes/parquet_source/metadata_utils.rs → ...odes/io_sources/parquet/metadata_utils.rs b/...rc/nodes/parquet_source/metadata_utils.rs → ...odes/io_sources/parquet/metadata_utils.rs
diff --git a/...rs-stream/src/nodes/parquet_source/mod.rs → ...tream/src/nodes/io_sources/parquet/mod.rs b/...rs-stream/src/nodes/parquet_source/mod.rs → ...tream/src/nodes/io_sources/parquet/mod.rs
@@ -17,10 +17,10 @@ use polars_plan::prelude::FileScanOptions;
 use polars_utils::index::AtomicIdxSize;
 use polars_utils::pl_str::PlSmallStr;
 
-use super::compute_node_prelude::*;
-use super::{MorselSeq, TaskPriority};
 use crate::async_primitives::wait_group::WaitToken;
 use crate::morsel::SourceToken;
+use crate::nodes::compute_node_prelude::*;
+use crate::nodes::{MorselSeq, TaskPriority};
 use crate::utils::task_handles_ext;
 
 mod init;

diff --git a/...es/parquet_source/row_group_data_fetch.rs → ...o_sources/parquet/row_group_data_fetch.rs b/...es/parquet_source/row_group_data_fetch.rs → ...o_sources/parquet/row_group_data_fetch.rs
diff --git a/.../nodes/parquet_source/row_group_decode.rs → ...es/io_sources/parquet/row_group_decode.rs b/.../nodes/parquet_source/row_group_decode.rs → ...es/io_sources/parquet/row_group_decode.rs
diff --git a/crates/polars-stream/src/nodes/mod.rs b/crates/polars-stream/src/nodes/mod.rs
@@ -1,4 +1,3 @@
-pub mod csv_source;
 pub mod filter;
 pub mod group_by;
 pub mod in_memory_map;
@@ -11,8 +10,6 @@ pub mod joins;
 pub mod map;
 pub mod multiplexer;
 pub mod ordered_union;
-#[cfg(feature = "parquet")]
-pub mod parquet_source;
 pub mod reduce;
 pub mod select;
 pub mod simple_projection;

diff --git a/crates/polars-stream/src/physical_plan/to_graph.rs b/crates/polars-stream/src/physical_plan/to_graph.rs
@@ -352,7 +352,7 @@ fn to_graph_rec<'a>(
                     } => {
                         if std::env::var("POLARS_DISABLE_PARQUET_SOURCE").as_deref() != Ok("1") {
                             ctx.graph.add_node(
-                                nodes::parquet_source::ParquetSourceNode::new(
+                                nodes::io_sources::parquet::ParquetSourceNode::new(
                                     scan_sources,
                                     file_info,
                                     hive_parts,
@@ -395,7 +395,7 @@ fn to_graph_rec<'a>(
                         }
 
                         ctx.graph.add_node(
-                            nodes::csv_source::CsvSourceNode::new(
+                            nodes::io_sources::csv::CsvSourceNode::new(
                                 scan_sources,
                                 file_info,
                                 file_options,