feat(datafusion): support the paimon_incremental_query table function

bindings/python/python/pypaimon_rust/datafusion.pyi

@@ -266,7 +266,8 @@ class TableScan:
     def with_row_position_slice(self, start: int, end: int) -> "TableScan": ...
     def with_row_position_shard(self, index: int, count: int) -> "TableScan": ...
     def with_chunk_shuffle(self, seed: str, chunk_size: int) -> "TableScan":
-        """Deterministically shuffle fixed-live-row chunks. ``seed`` is a decimal
+        """Deterministically shuffle fixed-live-row chunks; on Data Evolution
+        tables chunks count row ids. ``seed`` is a decimal
         integer string so arbitrarily large seeds keep Python ``random.Random``
         semantics."""
         ...

bindings/python/src/read.rs

@@ -767,7 +767,8 @@ impl PyTableScan {
         Ok(slf)
     }
 
-    /// Deterministically shuffle fixed-live-row chunks. `seed` is a decimal
+    /// Deterministically shuffle fixed-live-row chunks; on Data Evolution
+    /// tables chunks count row ids. `seed` is a decimal
     /// Python integer string so arbitrarily large seeds retain Python's
     /// `random.Random` semantics.
     fn with_chunk_shuffle(

crates/integrations/datafusion/src/catalog.rs

@@ -440,6 +440,22 @@ impl PaimonCatalogProvider {
         self
     }
 
+    pub(crate) fn catalog(&self) -> &Arc<dyn Catalog> {
+        &self.catalog
+    }
+
+    /// The session's dynamic options as they stand now.
+    pub(crate) fn session_options(&self) -> HashMap<String, String> {
+        self.dynamic_options
+            .read()
+            .unwrap_or_else(|e| e.into_inner())
+            .clone()
+    }
+
+    pub(crate) fn blob_reader_registry(&self) -> &BlobReaderRegistry {
+        &self.blob_reader_registry
+    }
+
     /// Register an engine for a table type the Paimon reader cannot serve
     /// (e.g. [`PaimonTableType::IcebergTable`]); everything else takes the
     /// Paimon path unchanged. Kept inside the provider so the registered

crates/integrations/datafusion/src/full_text_search.rs

@@ -24,6 +24,7 @@
 //!
 //! Reference: [PaimonTableValuedFunctions.scala](https://github.com/apache/paimon/blob/master/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonTableValuedFunctions.scala)
 
+use std::collections::HashMap;
 use std::fmt::Debug;
 use std::sync::Arc;
 
@@ -33,7 +34,7 @@ use datafusion::catalog::Session;
 use datafusion::catalog::TableFunctionImpl;
 use datafusion::common::project_schema;
 use datafusion::datasource::{TableProvider, TableType};
-use datafusion::error::Result as DFResult;
+use datafusion::error::{DataFusionError, Result as DFResult};
 use datafusion::logical_expr::{Expr, TableProviderFilterPushDown};
 use datafusion::physical_plan::empty::EmptyExec;
 use datafusion::physical_plan::ExecutionPlan;
@@ -46,7 +47,7 @@ use crate::table::{PaimonScanBuilder, PaimonTableProvider};
 use crate::table_function_args::{
     extract_int_literal, extract_string_literal, parse_table_identifier,
 };
-use crate::table_loader::load_data_table_for_read;
+use crate::table_loader::{ensure_no_incremental_window, load_data_table_for_read};
 
 const FUNCTION_NAME: &str = "full_text_search";
 
@@ -111,7 +112,19 @@ impl TableFunctionImpl for FullTextSearchFunction {
 
         let catalog = Arc::clone(&self.catalog);
         let table = block_on_with_runtime(
-            async move { load_data_table_for_read(&catalog, &identifier, FUNCTION_NAME).await },
+            async move {
+                let table = load_data_table_for_read(&catalog, &identifier, FUNCTION_NAME).await?;
+                // The index is searched at one snapshot, while a window stored
+                // on the table would make the read of its hits return that
+                // window's changes. Session options do not apply here.
+                ensure_no_incremental_window(
+                    &format!("{FUNCTION_NAME} searches one snapshot and cannot run"),
+                    &HashMap::new(),
+                    &identifier,
+                    &table,
+                )?;
+                Ok::<_, DataFusionError>(table)
+            },
             "full_text_search: catalog access thread panicked",
         )?;
 

crates/integrations/datafusion/src/hybrid_search.rs

@@ -59,7 +59,7 @@ use crate::table::{
 use crate::table_function_args::{
     extract_int_literal, extract_string_literal, parse_table_identifier,
 };
-use crate::table_loader::load_data_table_for_read;
+use crate::table_loader::{ensure_no_incremental_window, load_data_table_for_read};
 
 const FUNCTION_NAME: &str = "hybrid_search";
 const SEARCH_SCORE_COLUMN: &str = "__paimon_search_score";
@@ -133,7 +133,19 @@ impl TableFunctionImpl for HybridSearchFunction {
             parse_table_identifier(FUNCTION_NAME, &table_name, &self.default_database)?;
         let catalog = Arc::clone(&self.catalog);
         let table = block_on_with_runtime(
-            async move { load_data_table_for_read(&catalog, &identifier, FUNCTION_NAME).await },
+            async move {
+                let table = load_data_table_for_read(&catalog, &identifier, FUNCTION_NAME).await?;
+                // The indexes are searched at one snapshot, while a window
+                // stored on the table would make the read of their hits return
+                // that window's changes. Session options do not apply here.
+                ensure_no_incremental_window(
+                    &format!("{FUNCTION_NAME} searches one snapshot and cannot run"),
+                    &HashMap::new(),
+                    &identifier,
+                    &table,
+                )?;
+                Ok::<_, DataFusionError>(table)
+            },
             "hybrid_search: catalog access thread panicked",
         )?;