AI slop

Author: fulmicoton-dd

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

@@ -441,139 +441,7 @@ impl IndexingPipeline {
             sequencer: sequencer_handle,
             publisher: publisher_handle,
             next_check_for_progress: Instant::now() + *HEARTBEAT,
-        }));
-        Ok(())
-    }
-
-    /// Spawn the parquet pipeline using ParquetDocProcessor and ParquetIndexer.
-    ///
-    /// This pipeline routes metrics data through the Parquet/DataFusion path instead of
-    /// the Tantivy path. The output is ParquetSplit files written to the indexing directory.
-    #[instrument(
-        name="spawn_parquet_pipeline",
-        level="info",
-        skip_all,
-        fields(
-            index=%self.params.pipeline_id.index_uid.index_id,
-            r#gen=self.generation()
-        ))]
-    async fn spawn_parquet_pipeline(&mut self, ctx: &ActorContext<Self>) -> anyhow::Result<()> {
-        let index_id = &self.params.pipeline_id.index_uid.index_id;
-        let source_id = &self.params.pipeline_id.source_id;
-
-        info!(
-            index_id,
-            source_id,
-            pipeline_uid=%self.params.pipeline_id.pipeline_uid,
-            root_dir=%self.params.indexing_directory.path().display(),
-            "spawning parquet indexing pipeline for metrics",
-        );
-
-        let (source_mailbox, source_inbox) = ctx
-            .spawn_ctx()
-            .create_mailbox::<SourceActor<ParquetDocProcessor>>(
-                "SourceActor",
-                QueueCapacity::Unbounded,
-            );
-
-        // ParquetPublisher
-        let parquet_publisher = ParquetPublisher::new(
-            PublisherType::ParquetPublisher,
-            self.params.metastore.clone(),
-            None,
-            Some(source_mailbox.clone()),
-        );
-        let (parquet_publisher_mailbox, parquet_publisher_handle) = ctx
-            .spawn_actor()
-            .set_kill_switch(self.kill_switch.clone())
-            .spawn(parquet_publisher);
-
-        // Sequencer for ordered delivery
-        let parquet_sequencer = Sequencer::new(parquet_publisher_mailbox);
-        let (parquet_sequencer_mailbox, parquet_sequencer_handle) = ctx
-            .spawn_actor()
-            .set_kill_switch(self.kill_switch.clone())
-            .spawn(parquet_sequencer);
-
-        // ParquetUploader
-        let parquet_uploader = ParquetUploader::new(
-            super::UploaderType::IndexUploader,
-            self.params.metastore.clone(),
-            self.params.storage.clone(),
-            SplitsUpdateMailbox::Sequencer(parquet_sequencer_mailbox),
-            self.params.max_concurrent_split_uploads_index,
-        );
-        let (parquet_uploader_mailbox, parquet_uploader_handle) = ctx
-            .spawn_actor()
-            .set_kill_switch(self.kill_switch.clone())
-            .spawn(parquet_uploader);
-
-        // 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
-            .spawn_actor()
-            .set_kill_switch(self.kill_switch.clone())
-            .spawn(parquet_packager);
-
-        // ParquetIndexer
-        let commit_timeout =
-            Duration::from_secs(self.params.indexing_settings.commit_timeout_secs as u64);
-        let parquet_indexer = ParquetIndexer::new(
-            self.params.pipeline_id.index_uid.clone(),
-            source_id.to_string(),
-            None,
-            parquet_packager_mailbox,
-            Some(commit_timeout),
-        );
-        let (parquet_indexer_mailbox, parquet_indexer_handle) = ctx
-            .spawn_actor()
-            .set_kill_switch(self.kill_switch.clone())
-            .spawn(parquet_indexer);
-
-        // ParquetDocProcessor
-        let parquet_doc_processor = ParquetDocProcessor::new(
-            index_id.to_string(),
-            source_id.to_string(),
-            parquet_indexer_mailbox,
-        );
-        let (parquet_doc_processor_mailbox, parquet_doc_processor_handle) = ctx
-            .spawn_actor()
-            .set_kill_switch(self.kill_switch.clone())
-            .spawn(parquet_doc_processor);
-
-        // Source (using parquet source loader)
-        let source_runtime = SourceRuntime {
-            pipeline_id: self.params.pipeline_id.clone(),
-            source_config: self.params.source_config.clone(),
-            metastore: self.params.metastore.clone(),
-            ingester_pool: self.params.ingester_pool.clone(),
-            queues_dir_path: self.params.queues_dir_path.clone(),
-            storage_resolver: self.params.source_storage_resolver.clone(),
-            event_broker: self.params.event_broker.clone(),
-            indexing_setting: self.params.indexing_settings.clone(),
-        };
-        let source = ctx
-            .protect_future(quickwit_supported_parquet_sources().load_source(source_runtime))
-            .await?;
-        let actor_source = SourceActor {
-            source,
-            processor_mailbox: parquet_doc_processor_mailbox,
-        };
-        let (source_mailbox, source_handle) = ctx
-            .spawn_actor()
-            .set_mailboxes(source_mailbox, source_inbox)
-            .set_kill_switch(self.kill_switch.clone())
-            .spawn(actor_source);
-        let assign_shards_message = AssignShards(Assignment {
-            shard_ids: self.shard_ids.clone(),
-        }));
+        });
         Ok(())
     }
 

quickwit/quickwit-indexing/src/actors/metrics_pipeline/pipeline.rs

@@ -343,9 +343,11 @@ impl MetricsPipeline {
 
         // 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 packager = ParquetPackager::new(split_writer, uploader_mailbox);
         let (packager_mailbox, packager_handle) = ctx

quickwit/quickwit-indexing/src/source/mod.rs

@@ -92,7 +92,7 @@ pub use source_sink::SourceSink;
 pub use pulsar_source::{PulsarSource, PulsarSourceFactory};
 #[cfg(feature = "sqs")]
 pub use queue_sources::sqs_queue;
-use quickwit_actors::{Actor, ActorContext, ActorExitStatus, DeferableReplyHandler, Handler, Mailbox};
+use quickwit_actors::{Actor, ActorContext, ActorExitStatus, Handler};
 use quickwit_common::metrics::{GaugeGuard, MEMORY_METRICS};
 use quickwit_common::pubsub::EventBroker;
 use quickwit_common::runtimes::RuntimeType;
@@ -118,7 +118,7 @@ pub use void_source::{VoidSource, VoidSourceFactory};
 
 use self::doc_file_reader::dir_and_filename;
 use self::stdin_source::StdinSourceFactory;
-use crate::models::{NewPublishLock, NewPublishToken, RawDocBatch};
+use crate::models::RawDocBatch;
 use crate::source::ingest::IngestSourceFactory;
 use crate::source::ingest_api_source::IngestApiSourceFactory;
 
@@ -318,15 +318,10 @@ pub struct SourceActor {
 }
 
 impl SourceActor {
-    pub fn new<A>(source: Box<dyn Source>, source_sink: Mailbox<A>) -> Self
-    where A: Actor
-            + DeferableReplyHandler<RawDocBatch>
-            + DeferableReplyHandler<NewPublishLock>
-            + DeferableReplyHandler<NewPublishToken>
-    {
+    pub fn new(source: Box<dyn Source>, source_sink: impl Into<SourceSink>) -> Self {
         SourceActor {
             source,
-            source_sink: SourceSink::new(source_sink),
+            source_sink: source_sink.into(),
         }
     }
 }
@@ -436,23 +431,6 @@ pub fn quickwit_supported_sources() -> &'static SourceLoader {
     &SOURCE_LOADER
 }
 
-/// Returns the source loader for parquet pipelines (ParquetDocProcessor).
-///
-/// Metrics pipelines currently only support IngestV2 sources, which is the
-/// production source type for metrics ingestion.
-pub fn quickwit_supported_parquet_sources() -> &'static ParquetSourceLoader {
-    static PARQUET_SOURCE_LOADER: LazyLock<ParquetSourceLoader> = LazyLock::new(|| {
-        let mut source_factory = ParquetSourceLoader::default();
-        // Only IngestV2 is currently used for metrics ingestion
-        source_factory.add_source(SourceType::IngestV2, IngestSourceFactory);
-        // Add other sources for testing/development
-        source_factory.add_source(SourceType::File, FileSourceFactory);
-        source_factory.add_source(SourceType::Vec, VecSourceFactory);
-        source_factory.add_source(SourceType::Void, VoidSourceFactory);
-        source_factory
-    });
-    &PARQUET_SOURCE_LOADER
-}
 
 pub async fn check_source_connectivity(
     storage_resolver: &StorageResolver,

quickwit/quickwit-indexing/src/source/source_sink.rs

@@ -71,19 +71,15 @@ pub struct SourceSink {
     inner: Arc<dyn SourceSinkTrait>,
 }
 
-impl SourceSink {
-    /// Create a `SourceSink` from any actor mailbox whose actor implements
-    /// the required message handlers.
-    pub fn new<A>(mailbox: Mailbox<A>) -> Self
-    where A: Actor
-            + DeferableReplyHandler<RawDocBatch>
-            + DeferableReplyHandler<NewPublishLock>
-            + DeferableReplyHandler<NewPublishToken> {
+impl<T: SourceSinkTrait> From<T> for SourceSink {
+    fn from(source_sink: T) -> Self {
         Self {
-            inner: Arc::new(mailbox),
+            inner: Arc::new(source_sink),
         }
     }
+}
 
+impl SourceSink {
     /// Send a `RawDocBatch` to the processor.
     ///
     /// The source context's protect zone is held while the send is in flight,