feat: add Lumina vector index read infrastructure

.github/workflows/ci.yml

@@ -110,6 +110,11 @@ jobs:
           RUST_LOG: DEBUG
           RUST_BACKTRACE: full
 
+      - name: Install lumina native library
+        run: |
+          pip install lumina-data
+          echo "LUMINA_LIB_PATH=$(python3 -c 'import lumina_data; print(lumina_data.__path__[0])')/lib/liblumina_py.so" >> $GITHUB_ENV
+
       - name: DataFusion Integration Test
         run: cargo test -p paimon-datafusion --all-targets
         env:

crates/integrations/datafusion/src/full_text_search.rs

@@ -37,11 +37,16 @@ use datafusion::error::Result as DFResult;
 use datafusion::logical_expr::{Expr, TableProviderFilterPushDown};
 use datafusion::physical_plan::ExecutionPlan;
 use datafusion::prelude::SessionContext;
-use paimon::catalog::{Catalog, Identifier};
+use paimon::catalog::Catalog;
 
 use crate::error::to_datafusion_error;
 use crate::runtime::{await_with_runtime, block_on_with_runtime};
 use crate::table::{PaimonScanBuilder, PaimonTableProvider};
+use crate::table_function_args::{
+    extract_int_literal, extract_string_literal, parse_table_identifier,
+};
+
+const FUNCTION_NAME: &str = "full_text_search";
 
 /// Register the `full_text_search` table-valued function on a [`SessionContext`].
 pub fn register_full_text_search(
@@ -88,18 +93,19 @@ impl TableFunctionImpl for FullTextSearchFunction {
             ));
         }
 
-        let table_name = extract_string_literal(&args[0], "table_name")?;
-        let column_name = extract_string_literal(&args[1], "column_name")?;
-        let query_text = extract_string_literal(&args[2], "query_text")?;
-        let limit = extract_int_literal(&args[3], "limit")?;
+        let table_name = extract_string_literal(FUNCTION_NAME, &args[0], "table_name")?;
+        let column_name = extract_string_literal(FUNCTION_NAME, &args[1], "column_name")?;
+        let query_text = extract_string_literal(FUNCTION_NAME, &args[2], "query_text")?;
+        let limit = extract_int_literal(FUNCTION_NAME, &args[3], "limit")?;
 
         if limit <= 0 {
             return Err(datafusion::error::DataFusionError::Plan(
                 "full_text_search: limit must be positive".to_string(),
             ));
         }
 
-        let identifier = parse_table_identifier(&table_name, &self.default_database)?;
+        let identifier =
+            parse_table_identifier(FUNCTION_NAME, &table_name, &self.default_database)?;
 
         let catalog = Arc::clone(&self.catalog);
         let table = block_on_with_runtime(
@@ -201,58 +207,3 @@ impl TableProvider for FullTextSearchTableProvider {
         ])
     }
 }
-
-fn extract_string_literal(expr: &Expr, name: &str) -> DFResult<String> {
-    match expr {
-        Expr::Literal(scalar, _) => {
-            let s = scalar.try_as_str().flatten().ok_or_else(|| {
-                datafusion::error::DataFusionError::Plan(format!(
-                    "full_text_search: {name} must be a string literal, got: {expr}"
-                ))
-            })?;
-            Ok(s.to_string())
-        }
-        _ => Err(datafusion::error::DataFusionError::Plan(format!(
-            "full_text_search: {name} must be a literal, got: {expr}"
-        ))),
-    }
-}
-
-fn extract_int_literal(expr: &Expr, name: &str) -> DFResult<i64> {
-    use datafusion::common::ScalarValue;
-    match expr {
-        Expr::Literal(scalar, _) => match scalar {
-            ScalarValue::Int8(Some(v)) => Ok(*v as i64),
-            ScalarValue::Int16(Some(v)) => Ok(*v as i64),
-            ScalarValue::Int32(Some(v)) => Ok(*v as i64),
-            ScalarValue::Int64(Some(v)) => Ok(*v),
-            ScalarValue::UInt8(Some(v)) => Ok(*v as i64),
-            ScalarValue::UInt16(Some(v)) => Ok(*v as i64),
-            ScalarValue::UInt32(Some(v)) => Ok(*v as i64),
-            ScalarValue::UInt64(Some(v)) => i64::try_from(*v).map_err(|_| {
-                datafusion::error::DataFusionError::Plan(format!(
-                    "full_text_search: {name} value {v} exceeds i64 range"
-                ))
-            }),
-            _ => Err(datafusion::error::DataFusionError::Plan(format!(
-                "full_text_search: {name} must be an integer literal, got: {expr}"
-            ))),
-        },
-        _ => Err(datafusion::error::DataFusionError::Plan(format!(
-            "full_text_search: {name} must be a literal, got: {expr}"
-        ))),
-    }
-}
-
-fn parse_table_identifier(name: &str, default_database: &str) -> DFResult<Identifier> {
-    let parts: Vec<&str> = name.split('.').collect();
-    match parts.len() {
-        1 => Ok(Identifier::new(default_database, parts[0])),
-        2 => Ok(Identifier::new(parts[0], parts[1])),
-        // 3-part name: catalog.database.table — ignore catalog prefix
-        3 => Ok(Identifier::new(parts[1], parts[2])),
-        _ => Err(datafusion::error::DataFusionError::Plan(format!(
-            "full_text_search: invalid table name '{name}', expected 'table', 'database.table', or 'catalog.database.table'"
-        ))),
-    }
-}

crates/integrations/datafusion/src/lib.rs

@@ -48,7 +48,9 @@ pub mod runtime;
 mod sql_handler;
 mod system_tables;
 mod table;
+mod table_function_args;
 mod update;
+mod vector_search;
 
 pub use catalog::{PaimonCatalogProvider, PaimonSchemaProvider};
 pub use error::to_datafusion_error;
@@ -58,3 +60,4 @@ pub use physical_plan::PaimonTableScan;
 pub use relation_planner::PaimonRelationPlanner;
 pub use sql_handler::PaimonSqlHandler;
 pub use table::PaimonTableProvider;
+pub use vector_search::{register_vector_search, VectorSearchFunction};

crates/integrations/datafusion/src/table_function_args.rs

@@ -0,0 +1,82 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use datafusion::common::ScalarValue;
+use datafusion::error::{DataFusionError, Result as DFResult};
+use datafusion::logical_expr::Expr;
+use paimon::catalog::Identifier;
+
+pub(crate) fn extract_string_literal(
+    function_name: &str,
+    expr: &Expr,
+    name: &str,
+) -> DFResult<String> {
+    match expr {
+        Expr::Literal(scalar, _) => {
+            let s = scalar.try_as_str().flatten().ok_or_else(|| {
+                DataFusionError::Plan(format!(
+                    "{function_name}: {name} must be a string literal, got: {expr}"
+                ))
+            })?;
+            Ok(s.to_string())
+        }
+        _ => Err(DataFusionError::Plan(format!(
+            "{function_name}: {name} must be a literal, got: {expr}"
+        ))),
+    }
+}
+
+pub(crate) fn extract_int_literal(function_name: &str, expr: &Expr, name: &str) -> DFResult<i64> {
+    match expr {
+        Expr::Literal(scalar, _) => match scalar {
+            ScalarValue::Int8(Some(v)) => Ok(*v as i64),
+            ScalarValue::Int16(Some(v)) => Ok(*v as i64),
+            ScalarValue::Int32(Some(v)) => Ok(*v as i64),
+            ScalarValue::Int64(Some(v)) => Ok(*v),
+            ScalarValue::UInt8(Some(v)) => Ok(*v as i64),
+            ScalarValue::UInt16(Some(v)) => Ok(*v as i64),
+            ScalarValue::UInt32(Some(v)) => Ok(*v as i64),
+            ScalarValue::UInt64(Some(v)) => i64::try_from(*v).map_err(|_| {
+                DataFusionError::Plan(format!(
+                    "{function_name}: {name} value {v} exceeds i64 range"
+                ))
+            }),
+            _ => Err(DataFusionError::Plan(format!(
+                "{function_name}: {name} must be an integer literal, got: {expr}"
+            ))),
+        },
+        _ => Err(DataFusionError::Plan(format!(
+            "{function_name}: {name} must be a literal, got: {expr}"
+        ))),
+    }
+}
+
+pub(crate) fn parse_table_identifier(
+    function_name: &str,
+    name: &str,
+    default_database: &str,
+) -> DFResult<Identifier> {
+    let parts: Vec<&str> = name.split('.').collect();
+    match parts.len() {
+        1 => Ok(Identifier::new(default_database, parts[0])),
+        2 => Ok(Identifier::new(parts[0], parts[1])),
+        3 => Ok(Identifier::new(parts[1], parts[2])),
+        _ => Err(DataFusionError::Plan(format!(
+            "{function_name}: invalid table name '{name}', expected 'table', 'database.table', or 'catalog.database.table'"
+        ))),
+    }
+}