POC: separate-index sync MV — write path, product UX, commit sync, native read

sandbox/libs/analytics-api/src/main/java/org/opensearch/analytics/schema/IpType.java

@@ -12,7 +12,6 @@
 import org.apache.calcite.sql.type.SqlTypeName;
 
 import java.util.Locale;
-
 /**
  * Calcite type marker for an OpenSearch {@code ip} column. Backed by
  * {@link SqlTypeName#VARBINARY} so planner coercion is unchanged; the subclass exists

sandbox/libs/dataformat-native/rust/Cargo.toml

@@ -84,8 +84,12 @@ opensearch-repository-azure = { path = "../../../plugins/native-repository-azure
 opensearch-repository-fs = { path = "../../../plugins/native-repository-fs/src/main/rust" }
 
 [profile.release]
-lto = true
-codegen-units = 1
+# POC iteration speed: LTO off + parallel codegen. Fat LTO with
+# codegen-units=1 is the *benchmarking* configuration — restore it
+# (lto = true, codegen-units = 1) before taking performance numbers;
+# expect noticeably slower query/fold paths without it.
+lto = false
+codegen-units = 16
 incremental = true
 debug = "line-tables-only"
 strip = false

sandbox/plugins/analytics-backend-datafusion/rust/src/api.rs

@@ -924,7 +924,8 @@ pub async unsafe fn execute_query(
     let shard_view = &*(shard_view_ptr as *const ShardView);
     let runtime = &*(runtime_ptr as *const DataFusionRuntime);
     let cpu_executor = manager.cpu_executor();
-
+    //  a > 5 && name = 'arpit' --> ?
+    // substrait a > 5 && (indexed_filter)
     // Create per-query context (auto-registers in the global registry) and extract
     // its per-query memory pool overlaying the global pool.
     let global_pool = runtime.runtime_env.memory_pool.clone();

sandbox/plugins/analytics-backend-datafusion/rust/src/ffm.rs

@@ -579,6 +579,232 @@ pub unsafe extern "C" fn df_query_registry_top_n_by_current(
     Ok(written as i64)
 }
 
+/// POC(mv) v2 search: Final-fold via a caller-provided SQL template
+/// (placeholder __MV_STATES__ replaced with the UNION ALL of state files).
+#[ffm_safe]
+#[no_mangle]
+pub unsafe extern "C" fn df_mv_search_v2(
+    files_ptr: *const *const u8,
+    files_lens: *const i64,
+    files_count: i64,
+    sql_ptr: *const u8,
+    sql_len: i64,
+    out_ptr: *mut u8,
+    out_cap: i64,
+    out_len: *mut i64,
+) -> i64 {
+    let mut files = Vec::with_capacity(files_count as usize);
+    for i in 0..files_count as usize {
+        let f_ptr = *files_ptr.add(i);
+        let f_len = *files_lens.add(i);
+        files.push(
+            str_from_raw(f_ptr, f_len)
+                .map_err(|e| format!("df_mv_search_v2: file[{}]: {}", i, e))?
+                .to_string(),
+        );
+    }
+    let sql = str_from_raw(sql_ptr, sql_len).map_err(|e| format!("df_mv_search_v2: sql: {}", e))?;
+    let text = crate::mv_writer::mv_search_v2(&files, sql)?;
+    write_out_buffer(
+        text.as_bytes(),
+        out_ptr,
+        out_cap,
+        out_len,
+        "mv search v2 result",
+    )?;
+    Ok(0)
+}
+
+/// POC(mv) streaming writer lifecycle — create/feed/finalize/abort.
+#[ffm_safe]
+#[no_mangle]
+pub unsafe extern "C" fn df_mv_writer_create(
+    sql_ptr: *const u8,
+    sql_len: i64,
+    num_group_cols: i64,
+) -> i64 {
+    let sql =
+        str_from_raw(sql_ptr, sql_len).map_err(|e| format!("df_mv_writer_create: sql: {}", e))?;
+    Ok(crate::mv_writer::mv_writer_create(sql, num_group_cols))
+}
+
+#[ffm_safe]
+#[no_mangle]
+pub unsafe extern "C" fn df_mv_writer_feed(writer_id: i64, array_ptr: i64, schema_ptr: i64) -> i64 {
+    use arrow_array::ffi::{FFI_ArrowArray, FFI_ArrowSchema};
+    let ffi_array = FFI_ArrowArray::from_raw(array_ptr as *mut FFI_ArrowArray);
+    let ffi_schema = FFI_ArrowSchema::from_raw(schema_ptr as *mut FFI_ArrowSchema);
+    let mut array_data = arrow_array::ffi::from_ffi(ffi_array, &ffi_schema)
+        .map_err(|e| format!("df_mv_writer_feed: import: {}", e))?;
+    array_data.align_buffers();
+    let struct_array = arrow_array::StructArray::from(array_data);
+    let batch = arrow_array::RecordBatch::from(struct_array);
+    crate::mv_writer::mv_writer_feed(writer_id, &batch)?;
+    Ok(0)
+}
+
+#[ffm_safe]
+#[no_mangle]
+pub unsafe extern "C" fn df_mv_writer_finalize(
+    writer_id: i64,
+    out_ptr: *const u8,
+    out_len: i64,
+) -> i64 {
+    let output =
+        str_from_raw(out_ptr, out_len).map_err(|e| format!("df_mv_writer_finalize: out: {}", e))?;
+    let rows = crate::mv_writer::mv_writer_finalize(writer_id, output)?;
+    Ok(rows)
+}
+
+/// Finalizes the MV writer and exports the sorted state batch via Arrow
+/// C-Data into caller-allocated `FFI_ArrowArray`/`FFI_ArrowSchema` structs
+/// (zero copy — the JVM imports the same buffers; Rust's release callback
+/// frees them when the consumer closes). Returns the state row count.
+///
+/// # Safety
+/// `array_addr` / `schema_addr` must point to caller-allocated, uninitialized
+/// ArrowArray / ArrowSchema C structs that the caller will import exactly once.
+#[ffm_safe]
+#[no_mangle]
+pub unsafe extern "C" fn df_mv_writer_finalize_arrow(
+    writer_id: i64,
+    array_addr: i64,
+    schema_addr: i64,
+) -> i64 {
+    crate::mv_writer::mv_writer_finalize_arrow(writer_id, array_addr, schema_addr)
+}
+
+#[no_mangle]
+pub unsafe extern "C" fn df_mv_writer_abort(writer_id: i64) {
+    crate::mv_writer::mv_writer_abort(writer_id);
+}
+
+/// POC(mv): Final-aggregate over MV state files. Writes result text (svc\tcount lines)
+/// into the caller buffer via write_out_buffer.
+#[ffm_safe]
+#[no_mangle]
+pub unsafe extern "C" fn df_mv_search_poc(
+    files_ptr: *const *const u8,
+    files_lens: *const i64,
+    files_count: i64,
+    group_ptr: *const u8,
+    group_len: i64,
+    state_ptr: *const u8,
+    state_len: i64,
+    out_ptr: *mut u8,
+    out_cap: i64,
+    out_len: *mut i64,
+) -> i64 {
+    let mut files = Vec::with_capacity(files_count as usize);
+    for i in 0..files_count as usize {
+        let f_ptr = *files_ptr.add(i);
+        let f_len = *files_lens.add(i);
+        files.push(
+            str_from_raw(f_ptr, f_len)
+                .map_err(|e| format!("df_mv_search_poc: file[{}]: {}", i, e))?
+                .to_string(),
+        );
+    }
+    let group = str_from_raw(group_ptr, group_len)
+        .map_err(|e| format!("df_mv_search_poc: group: {}", e))?;
+    let state = str_from_raw(state_ptr, state_len)
+        .map_err(|e| format!("df_mv_search_poc: state: {}", e))?;
+    let text = crate::mv_poc::mv_search_poc(&files, group, state)?;
+    write_out_buffer(
+        text.as_bytes(),
+        out_ptr,
+        out_cap,
+        out_len,
+        "mv search result",
+    )?;
+    Ok(0)
+}
+
+/// POC(mv): build the hardcoded MV state file from a primary parquet file.
+/// Blocking; returns state-row count (>=0) on success.
+#[ffm_safe]
+#[no_mangle]
+pub unsafe extern "C" fn df_mv_build_poc(
+    input_ptr: *const u8,
+    input_len: i64,
+    table_ptr: *const u8,
+    table_len: i64,
+    sql_ptr: *const u8,
+    sql_len: i64,
+    output_ptr: *const u8,
+    output_len: i64,
+) -> i64 {
+    let input =
+        str_from_raw(input_ptr, input_len).map_err(|e| format!("df_mv_build_poc: input: {}", e))?;
+    let table =
+        str_from_raw(table_ptr, table_len).map_err(|e| format!("df_mv_build_poc: table: {}", e))?;
+    let sql = str_from_raw(sql_ptr, sql_len).map_err(|e| format!("df_mv_build_poc: sql: {}", e))?;
+    let output = str_from_raw(output_ptr, output_len)
+        .map_err(|e| format!("df_mv_build_poc: output: {}", e))?;
+    let rows = crate::mv_poc::mv_build_poc(input, table, sql, output)?;
+    Ok(rows)
+}
+
+/// Refresh-time ship build: run the definition's Partial over one parquet
+/// file and export the sorted state batch via Arrow C-Data (no scratch file).
+#[ffm_safe]
+#[no_mangle]
+pub unsafe extern "C" fn df_session_attach_mv(
+    handle_ptr: i64,
+    mv_paths_ptr: *const u8,
+    mv_paths_len: i64,
+    covered_names_ptr: *const u8,
+    covered_names_len: i64,
+    strict: u8,
+) -> i64 {
+    let handle = &mut *(handle_ptr as *mut crate::session_context::SessionContextHandle);
+    let mv_paths = str_from_raw(mv_paths_ptr, mv_paths_len)
+        .map_err(|e| format!("df_session_attach_mv: mv_paths: {}", e))?;
+    let covered_names = str_from_raw(covered_names_ptr, covered_names_len)
+        .map_err(|e| format!("df_session_attach_mv: covered_names: {}", e))?;
+    let mv_file_paths: Vec<String> = mv_paths
+        .split('\n')
+        .filter(|s| !s.is_empty())
+        .map(str::to_string)
+        .collect();
+    let covered_raw_file_names: std::collections::HashSet<String> = covered_names
+        .split('\n')
+        .filter(|s| !s.is_empty())
+        .map(str::to_string)
+        .collect();
+    if mv_file_paths.is_empty() {
+        return Ok(0); // nothing covered — leave the session raw-only
+    }
+    handle.mv_binding = Some(crate::mv_read::MVBinding {
+        mv_file_paths,
+        covered_raw_file_names,
+        strict: strict != 0,
+    });
+    Ok(0)
+}
+
+#[ffm_safe]
+#[no_mangle]
+pub unsafe extern "C" fn df_mv_build_arrow(
+    input_ptr: *const u8,
+    input_len: i64,
+    table_ptr: *const u8,
+    table_len: i64,
+    sql_ptr: *const u8,
+    sql_len: i64,
+    array_addr: i64,
+    schema_addr: i64,
+) -> i64 {
+    let input = str_from_raw(input_ptr, input_len)
+        .map_err(|e| format!("df_mv_build_arrow: input: {}", e))?;
+    let table = str_from_raw(table_ptr, table_len)
+        .map_err(|e| format!("df_mv_build_arrow: table: {}", e))?;
+    let sql =
+        str_from_raw(sql_ptr, sql_len).map_err(|e| format!("df_mv_build_arrow: sql: {}", e))?;
+    let rows = crate::mv_poc::mv_build_arrow(input, table, sql, array_addr, schema_addr)?;
+    Ok(rows)
+}
+
 #[ffm_safe]
 #[no_mangle]
 pub unsafe extern "C" fn df_sql_to_substrait(

sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_executor.rs

@@ -145,6 +145,7 @@ pub async fn execute_indexed_query(
         has_topk: false,
         prepared_plan: None,
         phantom_reservation: None,
+        mv_binding: None,
     };
     let ptr = Box::into_raw(Box::new(handle)) as i64;
 
@@ -1431,6 +1432,13 @@ async unsafe fn execute_indexed_with_context_inner(
     } else {
         physical_plan
     };
+    // MV read path (indexed executor = the live shard fragment path on this
+    // branch): a session binding replaces the Partial with the mv-state file
+    // scan (STRICT — misalignment throws, never a silent fallback).
+    let physical_plan = match handle.mv_binding.as_ref() {
+        Some(binding) => crate::mv_read::apply_mv_binding(&ctx, physical_plan, binding).await?,
+        None => physical_plan,
+    };
     let target_schema = crate::schema_coerce::coerce_inferred_schema(physical_plan.schema());
     let physical_plan = crate::relabel_exec::wrap_if_relabel_needed(physical_plan, target_schema)?;
     log_debug!(

sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/stream.rs

@@ -853,7 +853,7 @@ impl IndexedStream {
 
 impl Stream for IndexedStream {
     type Item = Result<RecordBatch>;
-
+    // 10 > A > 5 OR Text = 'Arpit'
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         // Manual timer for `elapsed_compute`: total wall time spent
         // inside this poll. Attributed to the operator for EXPLAIN

sandbox/plugins/analytics-backend-datafusion/rust/src/lib.rs

@@ -34,6 +34,9 @@ pub mod indexed_table;
 pub mod local_executor;
 pub mod memory;
 pub mod memory_guard;
+pub mod mv_poc;
+pub mod mv_read;
+pub mod mv_writer;
 pub mod native_error;
 pub mod partition_stream;
 pub mod patterns;