Fix spiceai#10951: UdtfExec invariant Vec lengths must match children count (spiceai#10953)

* udtf: fix UdtfExec invariant vec lengths to match children count

`UdtfExec::children()` reports one child (the inner plan), but
`maintains_input_order()`, `required_input_ordering()`,
`required_input_distribution()`, and `benefits_from_input_partitioning()`
all returned empty Vecs, violating the DataFusion ExecutionPlan contract
that these vectors must have one entry per child.

When `UdtfExec` got composed under another plan (e.g. `rrf(text_search(...),
vector_search(...))` without an explicit `join_key`), an optimizer pass
invoked `check_default_invariants` and the query failed with:

  Internal error: Assertion failed: actual_len == children_len
  (left: 0, right: 1): UdtfExec::maintains_input_order returned Vec with
  incorrect size: 0 != 1.

Since `UdtfExec::execute()` delegates straight to the inner plan, the
correct values are: `maintains_input_order = [true]`,
`benefits_from_input_partitioning = [false]`, `required_input_ordering = [None]`,
`required_input_distribution = [UnspecifiedDistribution]`.

Add a regression test that builds a UdtfExec and asserts both the per-method
Vec lengths and `check_invariants` itself.

Fixes spiceai#10951

* cayenne: invalidate scan_file_statistics cache after position-based delete

The position-based delete path updates the per-file deletion bitmap in
`cached_deleted_row_ids` but did not invalidate the per-file
`scan_file_statistics` cache that `CayenneTableProvider::list_files_for_snapshot_scan`
populates from `infer_stats`. Because `infer_stats` applies the
`VortexAccessPlanProvider::adjust_statistics` hook at the time it runs, the
cached entry froze the row count as of the *previous* delete. The next
`COUNT(*)` (or any other stats-driven query) hit the cache and returned the
stale count — even though the deletion bitmap itself was up to date.

Also keep the `cayenne::metastore::sqlite` docblock backtick fix (`SQLite`)
that the trunk lint failure pointed out — clippy::doc_markdown was failing
on that line under `clippy::pedantic`.

`PkKeysetInvalidatingDeletionSink` already wraps every `delete_using_deletion_vectors`
sink, so dropping the cache there covers every position-based deletion path
without touching the position-based sink internals. PK-strategy callers also
flow through this sink — clearing for them is harmless because they don't
populate the access-plan stats path.

Author: phillipleblanc

crates/cayenne/src/metastore/sqlite.rs

@@ -85,7 +85,7 @@ async fn configure_sqlite_connection(
 ///
 /// Pool size is `min(available_parallelism, 32)` (minimum 2). If
 /// `available_parallelism()` fails (rare — e.g. seccomp-restricted
-/// environments), K falls back to 4. SQLite WAL mode allows many
+/// environments), K falls back to 4. `SQLite` WAL mode allows many
 /// concurrent readers per database file (read-only operations don't take
 /// the WAL write lock), so a larger pool lifts the read-side concurrency
 /// ceiling for metadata-heavy workloads — e.g. 64-core deployments running

crates/cayenne/src/provider/table.rs

@@ -1485,6 +1485,12 @@ impl DeletionSink for PkKeysetInvalidatingDeletionSink {
         let deleted = self.inner.delete_from().await?;
         if deleted > 0 {
             self.table.clear_cached_pk_keyset();
+            // Drop the per-file stats `CayenneTableProvider::collect_scan_file_statistics`
+            // caches. Without this, a follow-up `COUNT(*)` (or any other stats-driven
+            // query) is served the row count we computed *before* this delete added
+            // its rows to the position-based deletion vector, so the count is stale —
+            // see `tests/position_based_deletion_test.rs::test_position_based_sequential_deletes`.
+            self.table.invalidate_scan_file_statistics();
         }
         Ok(deleted)
     }
@@ -6178,6 +6184,16 @@ impl CayenneTableProvider {
         self.refresh_listing_table_under_held_fence()
     }
 
+    /// Drop every entry in [`Self::scan_file_statistics`].
+    ///
+    /// Calls must follow any operation that adds, removes, or updates
+    /// position-based deletion vectors so the next stats-driven query (e.g.
+    /// `COUNT(*)`) reinvokes `infer_stats`, which in turn reapplies the
+    /// `VortexAccessPlanProvider` and observes the fresh deletion bitmap.
+    pub(crate) fn invalidate_scan_file_statistics(&self) {
+        self.scan_file_statistics.clear();
+    }
+
     /// Refresh the listing table, ASSUMING the caller already holds
     /// [`Self::listing_fence`] for write.
     ///

crates/runtime/src/execution_plan/udtf_exec.rs

@@ -168,19 +168,19 @@ impl ExecutionPlan for UdtfExec {
     }
 
     fn required_input_distribution(&self) -> Vec<Distribution> {
-        vec![]
+        vec![Distribution::UnspecifiedDistribution]
     }
 
     fn required_input_ordering(&self) -> Vec<Option<OrderingRequirements>> {
-        vec![]
+        vec![None]
     }
 
     fn maintains_input_order(&self) -> Vec<bool> {
-        vec![]
+        vec![true]
     }
 
     fn benefits_from_input_partitioning(&self) -> Vec<bool> {
-        vec![]
+        vec![false]
     }
 
     fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
@@ -477,3 +477,52 @@ impl ExecutionPlan for PlaceholderExec {
         Ok(SortOrderPushdownResult::Unsupported)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use arrow_schema::{DataType, Field, Schema};
+    use datafusion::physical_plan::empty::EmptyExec;
+    use runtime_proto::ListUdfsArgs;
+    use runtime_proto::udtf_args::Args;
+
+    fn test_udtf_args() -> UdtfArgs {
+        UdtfArgs {
+            args: Some(Args::ListUdfs(ListUdfsArgs {})),
+        }
+    }
+
+    fn test_inner() -> Arc<dyn ExecutionPlan> {
+        let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)]));
+        Arc::new(EmptyExec::new(schema))
+    }
+
+    /// Regression test for issue #10951.
+    ///
+    /// `check_default_invariants` (invoked by optimizer rules like `EnforceSorting`
+    /// when `UdtfExec` is composed under another plan, e.g. `rrf(text_search(...),
+    /// vector_search(...))`) asserts that `maintains_input_order`,
+    /// `required_input_ordering`, `required_input_distribution`, and
+    /// `benefits_from_input_partitioning` each return a Vec with one entry per
+    /// child. `UdtfExec` reports one child (the inner plan), so each of these
+    /// must return a single-element Vec.
+    #[test]
+    fn invariant_vec_lengths_match_children_count() {
+        let exec = UdtfExec::new(test_udtf_args(), test_inner());
+        let children_len = exec.children().len();
+        assert_eq!(children_len, 1);
+        assert_eq!(exec.maintains_input_order().len(), children_len);
+        assert_eq!(exec.required_input_ordering().len(), children_len);
+        assert_eq!(exec.required_input_distribution().len(), children_len);
+        assert_eq!(exec.benefits_from_input_partitioning().len(), children_len);
+    }
+
+    #[test]
+    fn check_default_invariants_passes() {
+        let exec = UdtfExec::new(test_udtf_args(), test_inner());
+        exec.check_invariants(InvariantLevel::Always)
+            .expect("default invariants should pass for UdtfExec");
+        exec.check_invariants(InvariantLevel::Executable)
+            .expect("default invariants should pass for UdtfExec");
+    }
+}