[phase-31 3/4] Writer + pipeline wiring

quickwit/quickwit-control-plane/src/indexing_scheduler/mod.rs

@@ -68,7 +68,7 @@ pub struct IndexingSchedulerState {
 ///
 /// Scheduling executes the following steps:
 /// 1. Builds a [`PhysicalIndexingPlan`] from the list of logical indexing tasks. See
-///    [`build_physical_indexing_plan`] for the implementation details.
+///    `build_physical_indexing_plan` for the implementation details.
 /// 2. Apply the [`PhysicalIndexingPlan`]: for each indexer, the scheduler send the indexing tasks
 ///    by gRPC. An indexer immediately returns an Ok and apply asynchronously the received plan. Any
 ///    errors (network) happening in this step are ignored. The scheduler runs a control loop that
@@ -98,7 +98,7 @@ pub struct IndexingSchedulerState {
 /// Concretely, it will send the faulty nodes of the plan they are supposed to follow.
 //
 /// Finally, in order to give the time for each indexer to run their indexing tasks, the control
-/// plane will wait at least [`MIN_DURATION_BETWEEN_SCHEDULING`] before comparing the desired
+/// plane will wait at least `MIN_DURATION_BETWEEN_SCHEDULING` before comparing the desired
 /// plan with the running plan.
 pub struct IndexingScheduler {
     cluster_id: String,

quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs

@@ -638,9 +638,11 @@ impl IndexingPipeline {
 
         // ParquetPackager
         let writer_config = quickwit_parquet_engine::storage::ParquetWriterConfig::default();
+        let table_config = quickwit_parquet_engine::table_config::TableConfig::default();
         let split_writer = quickwit_parquet_engine::storage::ParquetSplitWriter::new(
             writer_config,
             self.params.indexing_directory.path(),
+            &table_config,
         );
         let parquet_packager = ParquetPackager::new(split_writer, parquet_uploader_mailbox);
         let (parquet_packager_mailbox, parquet_packager_handle) = ctx

quickwit/quickwit-indexing/src/actors/parquet_doc_processor.rs

@@ -549,7 +549,8 @@ mod tests {
 
         // Create ParquetPackager
         let writer_config = ParquetWriterConfig::default();
-        let split_writer = ParquetSplitWriter::new(writer_config, temp_dir.path());
+        let table_config = quickwit_parquet_engine::table_config::TableConfig::default();
+        let split_writer = ParquetSplitWriter::new(writer_config, temp_dir.path(), &table_config);
         let packager = ParquetPackager::new(split_writer, uploader_mailbox);
         let (packager_mailbox, packager_handle) = universe.spawn_builder().spawn(packager);
 

quickwit/quickwit-indexing/src/actors/parquet_e2e_test.rs

@@ -172,7 +172,8 @@ async fn test_metrics_pipeline_e2e() {
 
     // ParquetPackager between indexer and uploader
     let writer_config = ParquetWriterConfig::default();
-    let split_writer = ParquetSplitWriter::new(writer_config, temp_dir.path());
+    let table_config = quickwit_parquet_engine::table_config::TableConfig::default();
+    let split_writer = ParquetSplitWriter::new(writer_config, temp_dir.path(), &table_config);
     let packager = ParquetPackager::new(split_writer, uploader_mailbox);
     let (packager_mailbox, packager_handle) = universe.spawn_builder().spawn(packager);