Add integration test for doc mapper update (#5266)

* Add integration test for doc mapper update

* Add suggested test cases

* Refactor update tests to separate files

* Assert on entire hit content

* Ignore failing tests for now

quickwit/quickwit-integration-tests/Cargo.toml

@@ -25,6 +25,7 @@ tonic = { workspace = true }
 tracing = { workspace = true }
 
 quickwit-actors = { workspace = true, features = ["testsuite"] }
+quickwit-cli = { workspace = true }
 quickwit-common = { workspace = true, features = ["testsuite"] }
 quickwit-config = { workspace = true, features = ["testsuite"] }
 quickwit-metastore = { workspace = true, features = ["testsuite"] }

quickwit/quickwit-integration-tests/src/test_utils/cluster_sandbox.rs

@@ -18,6 +18,7 @@
 // along with this program. If not, see <http://www.gnu.org/licenses/>.
 
 use std::collections::{HashMap, HashSet};
+use std::io::Write;
 use std::net::SocketAddr;
 use std::path::PathBuf;
 use std::str::FromStr;
@@ -26,6 +27,7 @@ use std::time::{Duration, Instant};
 use futures_util::future;
 use itertools::Itertools;
 use quickwit_actors::ActorExitStatus;
+use quickwit_cli::tool::{local_ingest_docs_cli, LocalIngestDocsArgs};
 use quickwit_common::new_coolid;
 use quickwit_common::runtimes::RuntimesConfig;
 use quickwit_common::test_utils::{wait_for_server_ready, wait_until_predicate};
@@ -44,6 +46,7 @@ use quickwit_rest_client::rest_client::{
 use quickwit_serve::{serve_quickwit, ListSplitsQueryParams};
 use quickwit_storage::StorageResolver;
 use reqwest::Url;
+use serde_json::Value;
 use tempfile::TempDir;
 use tokio::sync::watch::{self, Receiver, Sender};
 use tokio::task::JoinHandle;
@@ -383,6 +386,45 @@ impl ClusterSandbox {
         Ok(())
     }
 
+    pub async fn local_ingest(&self, index_id: &str, json_data: &[Value]) -> anyhow::Result<()> {
+        let test_conf = self
+            .node_configs
+            .iter()
+            .find(|config| config.services.contains(&QuickwitService::Indexer))
+            .ok_or(anyhow::anyhow!("No indexer node found"))?;
+        // NodeConfig cannot be serialized, we write our own simplified config
+        let mut tmp_config_file = tempfile::Builder::new().suffix(".yaml").tempfile().unwrap();
+        let node_config = format!(
+            r#"
+                version: 0.8
+                metastore_uri: {}
+                data_dir: {:?}
+                "#,
+            test_conf.node_config.metastore_uri, test_conf.node_config.data_dir_path
+        );
+        tmp_config_file.write_all(node_config.as_bytes())?;
+        tmp_config_file.flush()?;
+
+        let mut tmp_data_file = tempfile::NamedTempFile::new().unwrap();
+        for line in json_data {
+            serde_json::to_writer(&mut tmp_data_file, line)?;
+            tmp_data_file.write_all(b"\n")?;
+        }
+        tmp_data_file.flush()?;
+
+        local_ingest_docs_cli(LocalIngestDocsArgs {
+            clear_cache: false,
+            config_uri: QuickwitUri::from_str(tmp_config_file.path().to_str().unwrap())?,
+            index_id: index_id.to_string(),
+            input_format: quickwit_config::SourceInputFormat::Json,
+            overwrite: false,
+            vrl_script: None,
+            input_path_opt: Some(tmp_data_file.path().to_path_buf()),
+        })
+        .await?;
+        Ok(())
+    }
+
     pub async fn shutdown(self) -> Result<Vec<HashMap<String, ActorExitStatus>>, anyhow::Error> {
         // We need to drop rest clients first because reqwest can hold connections open
         // preventing rest server's graceful shutdown.

quickwit/quickwit-integration-tests/src/tests/mod.rs

@@ -19,4 +19,4 @@
 
 mod basic_tests;
 mod index_tests;
-mod index_update_tests;
+mod update_tests;