You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/01/09 08:32:08 UTC
(paimon) branch master updated: [vector] add faiss index and e2e test to support vector index (#6991)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 03f7c277ba [vector] add faiss index and e2e test to support vector index (#6991)
03f7c277ba is described below
commit 03f7c277ba07b30b1a73bd2d03314e9ca4a42e80
Author: jerry <li...@alibaba-inc.com>
AuthorDate: Fri Jan 9 16:32:02 2026 +0800
[vector] add faiss index and e2e test to support vector index (#6991)
---
.github/workflows/faiss-vector-index-tests.yml | 66 ---
.github/workflows/publish-faiss_snapshot.yml | 186 ++++++++
.github/workflows/utitcase.yml | 42 +-
paimon-faiss/paimon-faiss-e2e-test/pom.xml | 292 ++++++++++++
.../paimon/spark/sql/FaissVectorIndexE2ETest.scala | 395 ++++++++++++++++
paimon-faiss/paimon-faiss-index/pom.xml | 111 +++++
.../org/apache/paimon/faiss/index/FaissIndex.java | 404 ++++++++++++++++
.../apache/paimon/faiss/index/FaissIndexMeta.java | 111 +++++
.../apache/paimon/faiss/index/FaissIndexType.java | 60 +++
.../faiss/index/FaissVectorGlobalIndexReader.java | 491 +++++++++++++++++++
.../faiss/index/FaissVectorGlobalIndexWriter.java | 307 ++++++++++++
.../faiss/index/FaissVectorGlobalIndexer.java | 53 +++
.../index/FaissVectorGlobalIndexerFactory.java | 40 ++
.../faiss/index/FaissVectorIndexOptions.java | 210 +++++++++
.../paimon/faiss/index/FaissVectorMetric.java | 78 ++++
.../index/FaissVectorSearchGlobalIndexResult.java | 48 ++
....apache.paimon.globalindex.GlobalIndexerFactory | 18 +
.../index/FaissVectorGlobalIndexScanTest.java | 343 ++++++++++++++
.../faiss/index/FaissVectorGlobalIndexTest.java | 520 +++++++++++++++++++++
paimon-faiss/pom.xml | 2 +
20 files changed, 3708 insertions(+), 69 deletions(-)
diff --git a/.github/workflows/faiss-vector-index-tests.yml b/.github/workflows/faiss-vector-index-tests.yml
deleted file mode 100644
index 16a1d386fb..0000000000
--- a/.github/workflows/faiss-vector-index-tests.yml
+++ /dev/null
@@ -1,66 +0,0 @@
-################################################################################
-# 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.
-################################################################################
-
-name: Faiss Vector Index Tests
-
-on:
- push:
- paths:
- - 'paimon-faiss/**'
- - '.github/workflows/faiss-vector-index-tests.yml'
- pull_request:
- paths:
- - 'paimon-faiss/**'
- - '.github/workflows/faiss-vector-index-tests.yml'
-
-env:
- JDK_VERSION: 8
- MAVEN_OPTS: -Dmaven.wagon.httpconnectionManager.ttlSeconds=30 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true
-
-concurrency:
- group: ${{ github.workflow }}-${{ github.event_name }}-${{ github.event.number || github.run_id }}
- cancel-in-progress: true
-
-jobs:
- build_test:
- runs-on: ubuntu-latest
- timeout-minutes: 90
- steps:
- - name: Checkout code
- uses: actions/checkout@v4
-
- - name: Set up JDK ${{ env.JDK_VERSION }}
- uses: actions/setup-java@v4
- with:
- java-version: ${{ env.JDK_VERSION }}
- distribution: 'temurin'
-
- - name: Build FAISS native library
- uses: ./.github/actions/build-faiss-native
- with:
- platform: linux-amd64
-
- - name: Build paimon-faiss-jni
- shell: bash
- run: mvn -B clean install -pl paimon-faiss/paimon-faiss-jni -am -DskipTests -Ppaimon-faiss
-
- - name: Test paimon-faiss-jni
- timeout-minutes: 10
- run: mvn -T 1C -B test -pl paimon-faiss/paimon-faiss-jni -DskipFaissTests=false -Ppaimon-faiss
- env:
- MAVEN_OPTS: -Xmx2048m
diff --git a/.github/workflows/publish-faiss_snapshot.yml b/.github/workflows/publish-faiss_snapshot.yml
new file mode 100644
index 0000000000..e192846fd9
--- /dev/null
+++ b/.github/workflows/publish-faiss_snapshot.yml
@@ -0,0 +1,186 @@
+################################################################################
+# 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.
+################################################################################
+
+name: Publish Faiss Snapshot
+
+on:
+ schedule:
+ # At the end of every day
+ - cron: '0 1 * * *'
+ workflow_dispatch:
+ push:
+ paths:
+ - 'paimon-faiss/**'
+ branches:
+ - master
+
+env:
+ JDK_VERSION: 8
+ MAVEN_OPTS: -Dmaven.wagon.httpconnectionManager.ttlSeconds=30 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true
+
+concurrency:
+ group: ${{ github.workflow }}-${{ github.event_name }}-${{ github.event.number || github.run_id }}
+ cancel-in-progress: true
+
+jobs:
+ # Build native library for Linux AMD64
+ build-native-linux-amd64:
+ runs-on: ubuntu-latest
+ steps:
+ - name: Checkout code
+ uses: actions/checkout@v4
+
+ - name: Set up JDK ${{ env.JDK_VERSION }}
+ uses: actions/setup-java@v4
+ with:
+ java-version: ${{ env.JDK_VERSION }}
+ distribution: 'temurin'
+
+ - name: Build FAISS native library
+ uses: ./.github/actions/build-faiss-native
+ with:
+ platform: linux-amd64
+
+ - name: Upload native library
+ uses: actions/upload-artifact@v4
+ with:
+ name: native-linux-amd64
+ path: paimon-faiss/paimon-faiss-jni/src/main/resources/linux/amd64/
+ retention-days: 1
+
+ # Build native library for Linux AARCH64
+ build-native-linux-aarch64:
+ runs-on: ubuntu-24.04-arm
+ steps:
+ - name: Checkout code
+ uses: actions/checkout@v4
+
+ - name: Set up JDK ${{ env.JDK_VERSION }}
+ uses: actions/setup-java@v4
+ with:
+ java-version: ${{ env.JDK_VERSION }}
+ distribution: 'temurin'
+
+ - name: Build FAISS native library
+ uses: ./.github/actions/build-faiss-native
+ with:
+ platform: linux-aarch64
+
+ - name: Upload native library
+ uses: actions/upload-artifact@v4
+ with:
+ name: native-linux-aarch64
+ path: paimon-faiss/paimon-faiss-jni/src/main/resources/linux/aarch64/
+ retention-days: 1
+
+ # Build native library for macOS ARM (Apple Silicon)
+ build-native-macos-arm:
+ runs-on: macos-14
+ steps:
+ - name: Checkout code
+ uses: actions/checkout@v4
+
+ - name: Set up JDK ${{ env.JDK_VERSION }}
+ uses: actions/setup-java@v4
+ with:
+ java-version: ${{ env.JDK_VERSION }}
+ distribution: 'temurin'
+
+ - name: Build FAISS native library
+ uses: ./.github/actions/build-faiss-native
+ with:
+ platform: darwin-aarch64
+ use-homebrew: 'true'
+
+ - name: Upload native library
+ uses: actions/upload-artifact@v4
+ with:
+ name: native-darwin-aarch64
+ path: paimon-faiss/paimon-faiss-jni/src/main/resources/darwin/aarch64/
+ retention-days: 1
+
+ # Package and publish
+ package-and-publish:
+ if: github.repository == 'apache/paimon'
+ needs: [build-native-linux-amd64, build-native-linux-aarch64, build-native-macos-arm]
+ runs-on: ubuntu-latest
+ steps:
+ - name: Checkout code
+ uses: actions/checkout@v4
+
+ - name: Set up JDK ${{ env.JDK_VERSION }}
+ uses: actions/setup-java@v4
+ with:
+ java-version: ${{ env.JDK_VERSION }}
+ distribution: 'temurin'
+
+ - name: Download Linux AMD64 native library
+ uses: actions/download-artifact@v4
+ with:
+ name: native-linux-amd64
+ path: paimon-faiss/paimon-faiss-jni/src/main/resources/linux/amd64/
+
+ - name: Download Linux AARCH64 native library
+ uses: actions/download-artifact@v4
+ with:
+ name: native-linux-aarch64
+ path: paimon-faiss/paimon-faiss-jni/src/main/resources/linux/aarch64/
+
+ - name: Download macOS ARM native library
+ uses: actions/download-artifact@v4
+ with:
+ name: native-darwin-aarch64
+ path: paimon-faiss/paimon-faiss-jni/src/main/resources/darwin/aarch64/
+
+ - name: List all native libraries
+ run: |
+ echo "=== All native libraries ==="
+ find paimon-faiss/paimon-faiss-jni/src/main/resources -type f \( -name "*.so" -o -name "*.so.*" -o -name "*.dylib" \) -exec ls -la {} \;
+
+ - name: Cache local Maven repository
+ uses: actions/cache@v4
+ with:
+ path: ~/.m2/repository
+ key: faiss-snapshot-maven-${{ hashFiles('**/pom.xml') }}
+ restore-keys: |
+ faiss-snapshot-maven-
+
+ - name: Build and package paimon-faiss-jni
+ run: |
+ mvn -B clean install -pl paimon-faiss/paimon-faiss-jni -am -DskipTests -Ppaimon-faiss -Drat.skip
+
+ - name: Build and package paimon-faiss-index
+ run: |
+ mvn -B clean install -pl paimon-faiss/paimon-faiss-index -am -DskipTests -Ppaimon-faiss -Drat.skip
+
+ - name: Publish snapshot
+ env:
+ ASF_USERNAME: ${{ secrets.NEXUS_USER }}
+ ASF_PASSWORD: ${{ secrets.NEXUS_PW }}
+ MAVEN_OPTS: -Xmx4096m
+ run: |
+ tmp_settings="tmp-settings.xml"
+ echo "<settings><servers><server>" > $tmp_settings
+ echo "<id>apache.snapshots.https</id><username>$ASF_USERNAME</username>" >> $tmp_settings
+ echo "<password>$ASF_PASSWORD</password>" >> $tmp_settings
+ echo "</server></servers></settings>" >> $tmp_settings
+
+ mvn --settings $tmp_settings deploy -pl paimon-faiss/paimon-faiss-jni,paimon-faiss/paimon-faiss-index -Dgpg.skip -Drat.skip -DskipTests -Ppaimon-faiss
+
+ rm $tmp_settings
+
diff --git a/.github/workflows/utitcase.yml b/.github/workflows/utitcase.yml
index 62939d90cf..218e89c654 100644
--- a/.github/workflows/utitcase.yml
+++ b/.github/workflows/utitcase.yml
@@ -39,8 +39,33 @@ concurrency:
cancel-in-progress: true
jobs:
- build:
+ build_native:
runs-on: ubuntu-latest
+ steps:
+ - name: Checkout code
+ uses: actions/checkout@v4
+
+ - name: Set up JDK ${{ env.JDK_VERSION }}
+ uses: actions/setup-java@v4
+ with:
+ java-version: ${{ env.JDK_VERSION }}
+ distribution: 'temurin'
+
+ - name: Build FAISS native library
+ uses: ./.github/actions/build-faiss-native
+ with:
+ platform: linux-amd64
+
+ - name: Upload native library artifact
+ uses: actions/upload-artifact@v4
+ with:
+ name: faiss-native-linux-amd64
+ path: paimon-faiss/paimon-faiss-jni/src/main/resources/linux/amd64/
+ retention-days: 1
+
+ build_test:
+ runs-on: ubuntu-latest
+ needs: build_native
steps:
- name: Checkout code
@@ -52,10 +77,21 @@ jobs:
java-version: ${{ env.JDK_VERSION }}
distribution: 'temurin'
+ - name: Download native library artifact
+ uses: actions/download-artifact@v4
+ with:
+ name: faiss-native-linux-amd64
+ path: paimon-faiss/paimon-faiss-jni/src/main/resources/linux/amd64/
+
+ - name: List downloaded native library
+ run: |
+ echo "=== Downloaded native libraries ==="
+ ls -la paimon-faiss/paimon-faiss-jni/src/main/resources/linux/amd64/
+
- name: Build Others
run: |
echo "Start compiling modules"
- mvn -T 2C -B clean install -DskipTests -Pflink1,spark3
+ mvn -T 2C -B clean install -DskipTests -Pflink1,spark3,paimon-faiss
- name: Test Others
timeout-minutes: 60
@@ -64,7 +100,7 @@ jobs:
jvm_timezone=$(random_timezone)
echo "JVM timezone is set to $jvm_timezone"
- TEST_MODULES="!paimon-e2e-tests,!paimon-faiss/paimon-faiss-jni,"
+ TEST_MODULES="!paimon-e2e-tests,"
for suffix in ut 3.5 3.4 3.3 3.2; do
TEST_MODULES+="!org.apache.paimon:paimon-spark-${suffix}_2.12,"
done
diff --git a/paimon-faiss/paimon-faiss-e2e-test/pom.xml b/paimon-faiss/paimon-faiss-e2e-test/pom.xml
new file mode 100644
index 0000000000..f1719057b5
--- /dev/null
+++ b/paimon-faiss/paimon-faiss-e2e-test/pom.xml
@@ -0,0 +1,292 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-faiss</artifactId>
+ <version>1.4-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>paimon-faiss-e2e-test</artifactId>
+ <name>Paimon : Faiss End to End Tests</name>
+
+ <properties>
+ <test.java.version>java8</test.java.version>
+ <test.spark.main.version>3.5</test.spark.main.version>
+ <test.spark.version>3.5.7</test.spark.version>
+ </properties>
+
+ <dependencies>
+ <!-- Add dependencies to affect the order of mvn build -->
+
+ <!-- Jackson dependencies - must come first to ensure correct version -->
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ <version>2.15.2</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>2.15.2</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
+ <version>2.15.2</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-format</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-spark-${test.spark.main.version}_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-faiss-index</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-test-utils</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-spark-ut_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+
+ <!-- Spark dependencies -->
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-sql_${scala.binary.version}</artifactId>
+ <version>${test.spark.version}</version>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j2-impl</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jul-to-slf4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jcl-over-slf4j</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-sql_${scala.binary.version}</artifactId>
+ <version>${test.spark.version}</version>
+ <classifier>tests</classifier>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j2-impl</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jul-to-slf4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jcl-over-slf4j</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-catalyst_${scala.binary.version}</artifactId>
+ <version>${test.spark.version}</version>
+ <classifier>tests</classifier>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-core_${scala.binary.version}</artifactId>
+ <version>${test.spark.version}</version>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j2-impl</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jul-to-slf4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jcl-over-slf4j</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-core_${scala.binary.version}</artifactId>
+ <version>${test.spark.version}</version>
+ <classifier>tests</classifier>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j2-impl</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jul-to-slf4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jcl-over-slf4j</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-hive_${scala.binary.version}</artifactId>
+ <version>${test.spark.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- Scala dependencies -->
+ <dependency>
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scala-library</artifactId>
+ <version>${scala.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- Test dependencies -->
+ <dependency>
+ <groupId>org.scalatest</groupId>
+ <artifactId>scalatest_${scala.binary.version}</artifactId>
+ <version>3.2.14</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-deploy-plugin</artifactId>
+ <configuration>
+ <skip>true</skip>
+ </configuration>
+ </plugin>
+
+ <plugin>
+ <groupId>net.alchim31.maven</groupId>
+ <artifactId>scala-maven-plugin</artifactId>
+ <version>${scala-maven-plugin.version}</version>
+ <executions>
+ <execution>
+ <id>scala-compile-first</id>
+ <phase>process-resources</phase>
+ <goals>
+ <goal>add-source</goal>
+ <goal>compile</goal>
+ </goals>
+ </execution>
+ <execution>
+ <id>scala-test-compile</id>
+ <phase>process-test-resources</phase>
+ <goals>
+ <goal>testCompile</goal>
+ </goals>
+ </execution>
+ </executions>
+ <configuration>
+ <scalaVersion>${scala.version}</scalaVersion>
+ <args>
+ <arg>-nobootcp</arg>
+ </args>
+ </configuration>
+ </plugin>
+
+ <plugin>
+ <groupId>org.scalatest</groupId>
+ <artifactId>scalatest-maven-plugin</artifactId>
+ <version>${scalatest-maven-plugin.version}</version>
+ <configuration>
+ <reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory>
+ <junitxml>.</junitxml>
+ <filereports>TestSuiteReport.txt</filereports>
+ <suites>org.apache.paimon.spark.sql.FaissVectorIndexE2ETest</suites>
+ </configuration>
+ <executions>
+ <execution>
+ <id>test</id>
+ <goals>
+ <goal>test</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
+ <profiles>
+ <profile>
+ <id>java11</id>
+ <activation>
+ <jdk>[11,)</jdk>
+ </activation>
+ <properties>
+ <test.java.version>java11</test.java.version>
+ </properties>
+ </profile>
+ </profiles>
+</project>
diff --git a/paimon-faiss/paimon-faiss-e2e-test/src/test/scala/org/apache/paimon/spark/sql/FaissVectorIndexE2ETest.scala b/paimon-faiss/paimon-faiss-e2e-test/src/test/scala/org/apache/paimon/spark/sql/FaissVectorIndexE2ETest.scala
new file mode 100644
index 0000000000..36b7ed2620
--- /dev/null
+++ b/paimon-faiss/paimon-faiss-e2e-test/src/test/scala/org/apache/paimon/spark/sql/FaissVectorIndexE2ETest.scala
@@ -0,0 +1,395 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.spark.sql
+
+import org.apache.paimon.spark.PaimonSparkTestBase
+
+import scala.collection.JavaConverters._
+
+/** End-to-end tests for FAISS vector index read/write operations on Spark 3.5. */
+class FaissVectorIndexE2ETest extends PaimonSparkTestBase {
+
+ // ========== Index Creation Tests ==========
+
+ test("create faiss vector index - basic") {
+ withTable("T") {
+ spark.sql("""
+ |CREATE TABLE T (id INT, v ARRAY<FLOAT>)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'global-index.row-count-per-shard' = '10000',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true')
+ |""".stripMargin)
+
+ val values = (0 until 100)
+ .map(
+ i => s"($i, array(cast($i as float), cast(${i + 1} as float), cast(${i + 2} as float)))")
+ .mkString(",")
+ spark.sql(s"INSERT INTO T VALUES $values")
+
+ // Create FAISS vector index
+ val output = spark
+ .sql("CALL sys.create_global_index(table => 'test.T', index_column => 'v', index_type => 'faiss-vector-ann', options => 'vector.dim=3')")
+ .collect()
+ .head
+ assert(output.getBoolean(0))
+
+ // Verify index was created
+ val table = loadTable("T")
+ val indexEntries = table
+ .store()
+ .newIndexFileHandler()
+ .scanEntries()
+ .asScala
+ .filter(_.indexFile().indexType() == "faiss-vector-ann")
+
+ assert(indexEntries.nonEmpty)
+ val totalRowCount = indexEntries.map(_.indexFile().rowCount()).sum
+ assert(totalRowCount == 100L)
+ }
+ }
+
+ test("create faiss vector index - with different index types") {
+ withTable("T") {
+ spark.sql("""
+ |CREATE TABLE T (id INT, v ARRAY<FLOAT>)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'global-index.row-count-per-shard' = '10000',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true')
+ |""".stripMargin)
+
+ val values = (0 until 50)
+ .map(
+ i => s"($i, array(cast($i as float), cast(${i + 1} as float), cast(${i + 2} as float)))")
+ .mkString(",")
+ spark.sql(s"INSERT INTO T VALUES $values")
+
+ // Test with HNSW index type
+ val output = spark
+ .sql("CALL sys.create_global_index(table => 'test.T', index_column => 'v', index_type => 'faiss-vector-ann', options => 'vector.dim=3,vector.index-type=HNSW')")
+ .collect()
+ .head
+ assert(output.getBoolean(0))
+
+ val table = loadTable("T")
+ val indexEntries = table
+ .store()
+ .newIndexFileHandler()
+ .scanEntries()
+ .asScala
+ .filter(_.indexFile().indexType() == "faiss-vector-ann")
+
+ assert(indexEntries.nonEmpty)
+ }
+ }
+
+ test("create faiss vector index - with partitioned table") {
+ withTable("T") {
+ spark.sql("""
+ |CREATE TABLE T (id INT, v ARRAY<FLOAT>, pt STRING)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'global-index.row-count-per-shard' = '10000',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true')
+ | PARTITIONED BY (pt)
+ |""".stripMargin)
+
+ var values = (0 until 500)
+ .map(
+ i =>
+ s"($i, array(cast($i as float), cast(${i + 1} as float), cast(${i + 2} as float)), 'p0')")
+ .mkString(",")
+ spark.sql(s"INSERT INTO T VALUES $values")
+
+ values = (0 until 300)
+ .map(
+ i =>
+ s"($i, array(cast($i as float), cast(${i + 1} as float), cast(${i + 2} as float)), 'p1')")
+ .mkString(",")
+ spark.sql(s"INSERT INTO T VALUES $values")
+
+ val output = spark
+ .sql("CALL sys.create_global_index(table => 'test.T', index_column => 'v', index_type => 'faiss-vector-ann', options => 'vector.dim=3')")
+ .collect()
+ .head
+ assert(output.getBoolean(0))
+
+ val table = loadTable("T")
+ val indexEntries = table
+ .store()
+ .newIndexFileHandler()
+ .scanEntries()
+ .asScala
+ .filter(_.indexFile().indexType() == "faiss-vector-ann")
+
+ assert(indexEntries.nonEmpty)
+ val totalRowCount = indexEntries.map(_.indexFile().rowCount()).sum
+ assert(totalRowCount == 800L)
+ }
+ }
+
+ // ========== Index Write Tests ==========
+
+ test("write vectors - large dataset") {
+ withTable("T") {
+ spark.sql("""
+ |CREATE TABLE T (id INT, v ARRAY<FLOAT>)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'global-index.row-count-per-shard' = '10000',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true')
+ |""".stripMargin)
+
+ // Insert 10000 rows
+ val values = (0 until 10000)
+ .map(
+ i => s"($i, array(cast($i as float), cast(${i + 1} as float), cast(${i + 2} as float)))")
+ .mkString(",")
+ spark.sql(s"INSERT INTO T VALUES $values")
+
+ val output = spark
+ .sql("CALL sys.create_global_index(table => 'test.T', index_column => 'v', index_type => 'faiss-vector-ann', options => 'vector.dim=3')")
+ .collect()
+ .head
+ assert(output.getBoolean(0))
+
+ val table = loadTable("T")
+ val indexEntries = table
+ .store()
+ .newIndexFileHandler()
+ .scanEntries()
+ .asScala
+ .filter(_.indexFile().indexType() == "faiss-vector-ann")
+
+ val totalRowCount = indexEntries.map(_.indexFile().rowCount()).sum
+ assert(totalRowCount == 10000L)
+ }
+ }
+
+ // ========== Index Read/Search Tests ==========
+
+ test("read vectors - basic search") {
+ withTable("T") {
+ spark.sql("""
+ |CREATE TABLE T (id INT, v ARRAY<FLOAT>)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'global-index.row-count-per-shard' = '10000',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true')
+ |""".stripMargin)
+
+ val values = (0 until 100)
+ .map(
+ i => s"($i, array(cast($i as float), cast(${i + 1} as float), cast(${i + 2} as float)))")
+ .mkString(",")
+ spark.sql(s"INSERT INTO T VALUES $values")
+
+ spark
+ .sql("CALL sys.create_global_index(table => 'test.T', index_column => 'v', index_type => 'faiss-vector-ann', options => 'vector.dim=3')")
+ .collect()
+
+ // Test vector search
+ val result = spark
+ .sql("""
+ |SELECT * FROM vector_search('T', 'v', array(50.0f, 51.0f, 52.0f), 5)
+ |""".stripMargin)
+ .collect()
+ assert(result.map(_.getInt(0)).contains(50))
+ }
+ }
+
+ test("read vectors - top-k search with different k values") {
+ withTable("T") {
+ spark.sql("""
+ |CREATE TABLE T (id INT, v ARRAY<FLOAT>)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'global-index.row-count-per-shard' = '10000',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true')
+ |""".stripMargin)
+
+ val values = (0 until 200)
+ .map(
+ i => s"($i, array(cast($i as float), cast(${i + 1} as float), cast(${i + 2} as float)))")
+ .mkString(",")
+ spark.sql(s"INSERT INTO T VALUES $values")
+
+ spark
+ .sql("CALL sys.create_global_index(table => 'test.T', index_column => 'v', index_type => 'faiss-vector-ann', options => 'vector.dim=3')")
+ .collect()
+
+ // Test with k=1
+ var result = spark
+ .sql("""
+ |SELECT * FROM vector_search('T', 'v', array(100.0f, 101.0f, 102.0f), 1)
+ |""".stripMargin)
+ .collect()
+ assert(result.length == 1)
+ assert(result.head.getInt(0) == 100)
+
+ // Test with k=10
+ result = spark
+ .sql("""
+ |SELECT * FROM vector_search('T', 'v', array(100.0f, 101.0f, 102.0f), 10)
+ |""".stripMargin)
+ .collect()
+ assert(result.map(_.getInt(0)).contains(100))
+ }
+ }
+
+ test("read vectors - multiple concurrent searches") {
+ withTable("T") {
+ spark.sql("""
+ |CREATE TABLE T (id INT, v ARRAY<FLOAT>)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'global-index.row-count-per-shard' = '10000',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true')
+ |""".stripMargin)
+
+ val values = (0 until 500)
+ .map(
+ i => s"($i, array(cast($i as float), cast(${i + 1} as float), cast(${i + 2} as float)))")
+ .mkString(",")
+ spark.sql(s"INSERT INTO T VALUES $values")
+
+ spark
+ .sql("CALL sys.create_global_index(table => 'test.T', index_column => 'v', index_type => 'faiss-vector-ann', options => 'vector.dim=3')")
+ .collect()
+
+ // Perform multiple searches
+ val result1 = spark
+ .sql("""
+ |SELECT * FROM vector_search('T', 'v', array(10.0f, 11.0f, 12.0f), 3)
+ |""".stripMargin)
+ .collect()
+ assert(result1.length == 3)
+ assert(result1.map(_.getInt(0)).contains(10))
+
+ val result2 = spark
+ .sql("""
+ |SELECT * FROM vector_search('T', 'v', array(250.0f, 251.0f, 252.0f), 5)
+ |""".stripMargin)
+ .collect()
+ assert(result2.map(_.getInt(0)).contains(250))
+
+ val result3 = spark
+ .sql("""
+ |SELECT * FROM vector_search('T', 'v', array(450.0f, 451.0f, 452.0f), 7)
+ |""".stripMargin)
+ .collect()
+ assert(result3.map(_.getInt(0)).contains(450))
+ }
+ }
+
+ test("read vectors - normalized vectors search") {
+ withTable("T") {
+ spark.sql("""
+ |CREATE TABLE T (id INT, v ARRAY<FLOAT>)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'global-index.row-count-per-shard' = '10000',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true')
+ |""".stripMargin)
+
+ // Insert rows with normalized vectors
+ val values = (1 to 100)
+ .map {
+ i =>
+ val v = math.sqrt(3.0 * i * i)
+ val normalized = i.toFloat / v.toFloat
+ s"($i, array($normalized, $normalized, $normalized))"
+ }
+ .mkString(",")
+ spark.sql(s"INSERT INTO T VALUES $values")
+
+ spark.sql(
+ "CALL sys.create_global_index(table => 'test.T', index_column => 'v', index_type => 'faiss-vector-ann', options => 'vector.dim=3')")
+
+ // Query for top 10 similar to (1, 1, 1) normalized
+ val result = spark
+ .sql("""
+ |SELECT * FROM vector_search('T', 'v', array(0.577f, 0.577f, 0.577f), 10)
+ |""".stripMargin)
+ .collect()
+
+ assert(result.length == 10)
+ }
+ }
+
+ // ========== Integration Tests ==========
+
+ test("end-to-end: write, index, read cycle") {
+ withTable("T") {
+ // Step 1: Create table
+ spark.sql("""
+ |CREATE TABLE T (id INT, name STRING, embedding ARRAY<FLOAT>)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'global-index.row-count-per-shard' = '10000',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true')
+ |""".stripMargin)
+
+ // Step 2: Write data
+ val values = (0 until 1000)
+ .map(
+ i =>
+ s"($i, 'item_$i', array(cast($i as float), cast(${i + 1} as float), cast(${i + 2} as float)))")
+ .mkString(",")
+ spark.sql(s"INSERT INTO T VALUES $values")
+
+ // Step 3: Create index
+ val indexResult = spark
+ .sql("CALL sys.create_global_index(table => 'test.T', index_column => 'embedding', index_type => 'faiss-vector-ann', options => 'vector.dim=3')")
+ .collect()
+ .head
+ assert(indexResult.getBoolean(0))
+
+ // Step 4: Verify index creation
+ val table = loadTable("T")
+ val indexEntries = table
+ .store()
+ .newIndexFileHandler()
+ .scanEntries()
+ .asScala
+ .filter(_.indexFile().indexType() == "faiss-vector-ann")
+ assert(indexEntries.nonEmpty)
+
+ // Step 5: Read using vector search
+ val searchResult = spark
+ .sql(
+ """
+ |SELECT id, name FROM vector_search('T', 'embedding', array(500.0f, 501.0f, 502.0f), 10)
+ |""".stripMargin)
+ .collect()
+
+ assert(searchResult.exists(row => row.getInt(0) == 500 && row.getString(1) == "item_500"))
+ }
+ }
+}
diff --git a/paimon-faiss/paimon-faiss-index/pom.xml b/paimon-faiss/paimon-faiss-index/pom.xml
new file mode 100644
index 0000000000..04fab6a9c9
--- /dev/null
+++ b/paimon-faiss/paimon-faiss-index/pom.xml
@@ -0,0 +1,111 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-faiss</artifactId>
+ <version>1.4-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>paimon-faiss-index</artifactId>
+ <name>Paimon : Faiss Index</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-faiss-jni</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <!-- test dependencies -->
+
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <version>${junit5.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-core</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-format</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-test-utils</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client</artifactId>
+ <version>${hadoop.version}</version>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <artifactId>maven-jar-plugin</artifactId>
+ <executions>
+ <execution>
+ <goals>
+ <goal>test-jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
+
+
diff --git a/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissIndex.java b/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissIndex.java
new file mode 100644
index 0000000000..2aa9312a5f
--- /dev/null
+++ b/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissIndex.java
@@ -0,0 +1,404 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.faiss.index;
+
+import org.apache.paimon.faiss.Index;
+import org.apache.paimon.faiss.IndexFactory;
+import org.apache.paimon.faiss.IndexHNSW;
+import org.apache.paimon.faiss.IndexIVF;
+import org.apache.paimon.faiss.MetricType;
+
+import java.io.Closeable;
+import java.io.File;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+/**
+ * A wrapper class for FAISS index with zero-copy support.
+ *
+ * <p>This class provides a safe Java API for interacting with native FAISS indices using direct
+ * ByteBuffers for zero-copy data transfer. This eliminates memory duplication and improves
+ * performance for large-scale vector operations.
+ */
+public class FaissIndex implements Closeable {
+
+ private final Index index;
+ private final int dimension;
+ private final FaissVectorMetric metric;
+ private final FaissIndexType indexType;
+ private volatile boolean closed = false;
+
+ private FaissIndex(
+ Index index, int dimension, FaissVectorMetric metric, FaissIndexType indexType) {
+ this.index = index;
+ this.dimension = dimension;
+ this.metric = metric;
+ this.indexType = indexType;
+ }
+
+ /**
+ * Create a flat index (exact search).
+ *
+ * @param dimension the dimension of vectors
+ * @param metric the distance metric
+ * @return the created index
+ */
+ public static FaissIndex createFlatIndex(int dimension, FaissVectorMetric metric) {
+ MetricType metricType = toMetricType(metric);
+ Index index = IndexFactory.create(dimension, "IDMap,Flat", metricType);
+ return new FaissIndex(index, dimension, metric, FaissIndexType.FLAT);
+ }
+
+ /**
+ * Create an HNSW index.
+ *
+ * @param dimension the dimension of vectors
+ * @param m the number of connections per layer
+ * @param efConstruction the size of the dynamic candidate list for construction
+ * @param metric the distance metric
+ * @return the created index
+ */
+ public static FaissIndex createHnswIndex(
+ int dimension, int m, int efConstruction, FaissVectorMetric metric) {
+ MetricType metricType = toMetricType(metric);
+ String description = String.format("IDMap2,HNSW%d", m);
+ Index index = IndexFactory.create(dimension, description, metricType);
+ // Set efConstruction before adding any vectors
+ IndexHNSW.setEfConstruction(index, efConstruction);
+ return new FaissIndex(index, dimension, metric, FaissIndexType.HNSW);
+ }
+
+ /**
+ * Create an IVF index.
+ *
+ * @param dimension the dimension of vectors
+ * @param nlist the number of inverted lists (clusters)
+ * @param metric the distance metric
+ * @return the created index
+ */
+ public static FaissIndex createIvfIndex(int dimension, int nlist, FaissVectorMetric metric) {
+ MetricType metricType = toMetricType(metric);
+ String description = String.format("IDMap,IVF%d,Flat", nlist);
+ Index index = IndexFactory.create(dimension, description, metricType);
+ return new FaissIndex(index, dimension, metric, FaissIndexType.IVF);
+ }
+
+ /**
+ * Create an IVF-PQ index.
+ *
+ * @param dimension the dimension of vectors
+ * @param nlist the number of inverted lists (clusters)
+ * @param m the number of sub-quantizers
+ * @param nbits the number of bits per sub-quantizer
+ * @param metric the distance metric
+ * @return the created index
+ */
+ public static FaissIndex createIvfPqIndex(
+ int dimension, int nlist, int m, int nbits, FaissVectorMetric metric) {
+ MetricType metricType = toMetricType(metric);
+ String description = String.format("IDMap,IVF%d,PQ%dx%d", nlist, m, nbits);
+ Index index = IndexFactory.create(dimension, description, metricType);
+ return new FaissIndex(index, dimension, metric, FaissIndexType.IVF_PQ);
+ }
+
+ /**
+ * Create an IVF-SQ8 index (Inverted File with Scalar Quantization 8-bit).
+ *
+ * @param dimension the dimension of vectors
+ * @param nlist the number of inverted lists (clusters)
+ * @param metric the distance metric
+ * @return the created index
+ */
+ public static FaissIndex createIvfSq8Index(int dimension, int nlist, FaissVectorMetric metric) {
+ MetricType metricType = toMetricType(metric);
+ String description = String.format("IDMap,IVF%d,SQ8", nlist);
+ Index index = IndexFactory.create(dimension, description, metricType);
+ return new FaissIndex(index, dimension, metric, FaissIndexType.IVF_SQ8);
+ }
+
+ /**
+ * Load an index from a local file using memory-mapped I/O.
+ *
+ * <p>This method uses FAISS's native file reading which can leverage mmap for efficient memory
+ * usage with large indices.
+ *
+ * @param file the local file containing the FAISS index
+ * @return the loaded index
+ */
+ public static FaissIndex fromFile(File file) {
+ Index index = Index.readFromFile(file);
+ return new FaissIndex(
+ index, index.getDimension(), FaissVectorMetric.L2, FaissIndexType.UNKNOWN);
+ }
+
+ /**
+ * Search for k nearest neighbors.
+ *
+ * @param queryVectors array containing query vectors (n * dimension floats)
+ * @param n the number of queries
+ * @param k the number of nearest neighbors
+ * @param distances output array for distances (n * k floats)
+ * @param labels output array for labels (n * k longs)
+ */
+ public void search(float[] queryVectors, int n, int k, float[] distances, long[] labels) {
+ ensureOpen();
+ if (queryVectors.length < n * dimension) {
+ throw new IllegalArgumentException(
+ "Query vectors array too small: required "
+ + (n * dimension)
+ + ", got "
+ + queryVectors.length);
+ }
+ if (distances.length < n * k) {
+ throw new IllegalArgumentException(
+ "Distances array too small: required " + (n * k) + ", got " + distances.length);
+ }
+ if (labels.length < n * k) {
+ throw new IllegalArgumentException(
+ "Labels array too small: required " + (n * k) + ", got " + labels.length);
+ }
+ index.search(n, queryVectors, k, distances, labels);
+ }
+
+ /**
+ * Train the index (zero-copy).
+ *
+ * <p>Required for IVF-based indices before adding vectors.
+ *
+ * @param vectorBuffer direct ByteBuffer containing training vectors (n * dimension floats)
+ * @param n the number of training vectors
+ */
+ public void train(ByteBuffer vectorBuffer, int n) {
+ ensureOpen();
+ validateVectorBuffer(vectorBuffer, n);
+ index.train(n, vectorBuffer);
+ }
+
+ /**
+ * Add vectors to the index (zero-copy).
+ *
+ * @param vectorBuffer direct ByteBuffer containing vectors (n * dimension floats)
+ * @param n the number of vectors
+ */
+ public void add(ByteBuffer vectorBuffer, int n) {
+ ensureOpen();
+ validateVectorBuffer(vectorBuffer, n);
+ index.add(n, vectorBuffer);
+ }
+
+ /**
+ * Add vectors with IDs to the index (zero-copy).
+ *
+ * @param vectorBuffer direct ByteBuffer containing vectors (n * dimension floats)
+ * @param idBuffer direct ByteBuffer containing IDs (n longs)
+ * @param n the number of vectors
+ */
+ public void addWithIds(ByteBuffer vectorBuffer, ByteBuffer idBuffer, int n) {
+ ensureOpen();
+ validateVectorBuffer(vectorBuffer, n);
+ validateIdBuffer(idBuffer, n);
+ index.addWithIds(n, vectorBuffer, idBuffer);
+ }
+
+ /**
+ * Check if the index is trained.
+ *
+ * @return true if the index is trained
+ */
+ public boolean isTrained() {
+ ensureOpen();
+ return index.isTrained();
+ }
+
+ /**
+ * Get the number of vectors in the index.
+ *
+ * @return the number of vectors
+ */
+ public long size() {
+ ensureOpen();
+ return index.getCount();
+ }
+
+ /** Reset the index (remove all vectors). */
+ public void reset() {
+ ensureOpen();
+ index.reset();
+ }
+
+ // ==================== Index Configuration ====================
+
+ /**
+ * Set HNSW search parameter efSearch.
+ *
+ * @param efSearch the size of the dynamic candidate list for search
+ */
+ public void setHnswEfSearch(int efSearch) {
+ ensureOpen();
+ IndexHNSW.setEfSearch(index, efSearch);
+ }
+
+ /**
+ * Set IVF search parameter nprobe.
+ *
+ * @param nprobe the number of clusters to visit during search
+ */
+ public void setIvfNprobe(int nprobe) {
+ ensureOpen();
+ IndexIVF.setNprobe(index, nprobe);
+ }
+
+ // ==================== Index Properties ====================
+
+ /**
+ * Get the dimension of vectors in the index.
+ *
+ * @return the dimension
+ */
+ public int dimension() {
+ return dimension;
+ }
+
+ /**
+ * Get the metric used by this index.
+ *
+ * @return the metric
+ */
+ public FaissVectorMetric metric() {
+ return metric;
+ }
+
+ /**
+ * Get the type of this index.
+ *
+ * @return the index type
+ */
+ public FaissIndexType indexType() {
+ return indexType;
+ }
+
+ // ==================== Serialization ====================
+
+ /**
+ * Get the size in bytes needed to serialize this index.
+ *
+ * @return the serialization size
+ */
+ public long serializeSize() {
+ ensureOpen();
+ return index.serializeSize();
+ }
+
+ /**
+ * Serialize the index to a direct ByteBuffer (zero-copy).
+ *
+ * @param buffer direct ByteBuffer to write to (must have sufficient capacity)
+ * @return the number of bytes written
+ */
+ public long serialize(ByteBuffer buffer) {
+ ensureOpen();
+ if (!buffer.isDirect()) {
+ throw new IllegalArgumentException("Buffer must be a direct buffer");
+ }
+ return index.serialize(buffer);
+ }
+
+ // ==================== Buffer Allocation Utilities ====================
+
+ /**
+ * Allocate a direct ByteBuffer suitable for vector data.
+ *
+ * @param numVectors number of vectors
+ * @param dimension vector dimension
+ * @return a direct ByteBuffer in native byte order
+ */
+ public static ByteBuffer allocateVectorBuffer(int numVectors, int dimension) {
+ return ByteBuffer.allocateDirect(numVectors * dimension * Float.BYTES)
+ .order(ByteOrder.nativeOrder());
+ }
+
+ /**
+ * Allocate a direct ByteBuffer suitable for ID data.
+ *
+ * @param numIds number of IDs
+ * @return a direct ByteBuffer in native byte order
+ */
+ public static ByteBuffer allocateIdBuffer(int numIds) {
+ return ByteBuffer.allocateDirect(numIds * Long.BYTES).order(ByteOrder.nativeOrder());
+ }
+
+ // ==================== Internal Methods ====================
+
+ private void validateVectorBuffer(ByteBuffer buffer, int numVectors) {
+ if (!buffer.isDirect()) {
+ throw new IllegalArgumentException("Vector buffer must be a direct buffer");
+ }
+ int requiredBytes = numVectors * dimension * Float.BYTES;
+ if (buffer.capacity() < requiredBytes) {
+ throw new IllegalArgumentException(
+ "Vector buffer too small: required "
+ + requiredBytes
+ + " bytes, got "
+ + buffer.capacity());
+ }
+ }
+
+ private void validateIdBuffer(ByteBuffer buffer, int numIds) {
+ if (!buffer.isDirect()) {
+ throw new IllegalArgumentException("ID buffer must be a direct buffer");
+ }
+ int requiredBytes = numIds * Long.BYTES;
+ if (buffer.capacity() < requiredBytes) {
+ throw new IllegalArgumentException(
+ "ID buffer too small: required "
+ + requiredBytes
+ + " bytes, got "
+ + buffer.capacity());
+ }
+ }
+
+ private void ensureOpen() {
+ if (closed) {
+ throw new IllegalStateException("Index has been closed");
+ }
+ }
+
+ private static MetricType toMetricType(FaissVectorMetric metric) {
+ switch (metric) {
+ case L2:
+ return MetricType.L2;
+ case INNER_PRODUCT:
+ return MetricType.INNER_PRODUCT;
+ default:
+ throw new IllegalArgumentException("Unknown metric: " + metric);
+ }
+ }
+
+ @Override
+ public void close() {
+ if (!closed) {
+ synchronized (this) {
+ if (!closed) {
+ index.close();
+ closed = true;
+ }
+ }
+ }
+ }
+}
diff --git a/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissIndexMeta.java b/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissIndexMeta.java
new file mode 100644
index 0000000000..78a8362606
--- /dev/null
+++ b/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissIndexMeta.java
@@ -0,0 +1,111 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.faiss.index;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.Serializable;
+
+/** Metadata for FAISS vector index. */
+public class FaissIndexMeta implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final int VERSION = 1;
+
+ private final int dim;
+ private final int metricValue;
+ private final int indexTypeOrdinal;
+ private final long numVectors;
+ private final long minId;
+ private final long maxId;
+
+ public FaissIndexMeta(
+ int dim,
+ int metricValue,
+ int indexTypeOrdinal,
+ long numVectors,
+ long minId,
+ long maxId) {
+ this.dim = dim;
+ this.metricValue = metricValue;
+ this.indexTypeOrdinal = indexTypeOrdinal;
+ this.numVectors = numVectors;
+ this.minId = minId;
+ this.maxId = maxId;
+ }
+
+ public int dim() {
+ return dim;
+ }
+
+ public int metricValue() {
+ return metricValue;
+ }
+
+ public int indexTypeOrdinal() {
+ return indexTypeOrdinal;
+ }
+
+ public long numVectors() {
+ return numVectors;
+ }
+
+ public long minId() {
+ return minId;
+ }
+
+ public long maxId() {
+ return maxId;
+ }
+
+ /** Serialize metadata to byte array. */
+ public byte[] serialize() throws IOException {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(baos);
+ out.writeInt(VERSION);
+ out.writeInt(dim);
+ out.writeInt(metricValue);
+ out.writeInt(indexTypeOrdinal);
+ out.writeLong(numVectors);
+ out.writeLong(minId);
+ out.writeLong(maxId);
+ out.flush();
+ return baos.toByteArray();
+ }
+
+ /** Deserialize metadata from byte array. */
+ public static FaissIndexMeta deserialize(byte[] data) throws IOException {
+ DataInputStream in = new DataInputStream(new ByteArrayInputStream(data));
+ int version = in.readInt();
+ if (version != VERSION) {
+ throw new IOException("Unsupported FAISS index meta version: " + version);
+ }
+ int dim = in.readInt();
+ int metricValue = in.readInt();
+ int indexTypeOrdinal = in.readInt();
+ long numVectors = in.readLong();
+ long minId = in.readLong();
+ long maxId = in.readLong();
+ return new FaissIndexMeta(dim, metricValue, indexTypeOrdinal, numVectors, minId, maxId);
+ }
+}
diff --git a/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissIndexType.java b/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissIndexType.java
new file mode 100644
index 0000000000..dbe615897b
--- /dev/null
+++ b/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissIndexType.java
@@ -0,0 +1,60 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.faiss.index;
+
+/** Enumeration of supported FAISS index types. */
+public enum FaissIndexType {
+
+ /** Flat index - exact brute-force search. */
+ FLAT("Flat"),
+
+ /** HNSW (Hierarchical Navigable Small World) graph-based index. */
+ HNSW("HNSW"),
+
+ /** IVF (Inverted File) index with flat vectors. */
+ IVF("IVF"),
+
+ /** IVF-PQ (Inverted File with Product Quantization) index. */
+ IVF_PQ("IVF_PQ"),
+
+ /** IVF-SQ8 (Inverted File with Scalar Quantization 8-bit) index. */
+ IVF_SQ8("IVF_SQ8"),
+
+ /** Unknown index type (e.g., loaded from serialized data). */
+ UNKNOWN("Unknown");
+
+ private final String name;
+
+ FaissIndexType(String name) {
+ this.name = name;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public static FaissIndexType fromString(String name) {
+ for (FaissIndexType type : values()) {
+ if (type.name.equalsIgnoreCase(name)) {
+ return type;
+ }
+ }
+ throw new IllegalArgumentException("Unknown index type: " + name);
+ }
+}
diff --git a/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissVectorGlobalIndexReader.java b/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissVectorGlobalIndexReader.java
new file mode 100644
index 0000000000..3aa1fca16c
--- /dev/null
+++ b/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissVectorGlobalIndexReader.java
@@ -0,0 +1,491 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.faiss.index;
+
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.globalindex.GlobalIndexIOMeta;
+import org.apache.paimon.globalindex.GlobalIndexReader;
+import org.apache.paimon.globalindex.GlobalIndexResult;
+import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
+import org.apache.paimon.predicate.FieldRef;
+import org.apache.paimon.predicate.VectorSearch;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.FloatType;
+import org.apache.paimon.utils.IOUtils;
+import org.apache.paimon.utils.RoaringNavigableMap64;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Optional;
+import java.util.PriorityQueue;
+import java.util.UUID;
+
+/**
+ * Vector global index reader using FAISS.
+ *
+ * <p>This implementation uses FAISS for efficient approximate nearest neighbor search.
+ */
+public class FaissVectorGlobalIndexReader implements GlobalIndexReader {
+
+ private final List<FaissIndex> indices;
+ private final List<FaissIndexMeta> indexMetas;
+ private final List<File> localIndexFiles;
+ private final List<GlobalIndexIOMeta> ioMetas;
+ private final GlobalIndexFileReader fileReader;
+ private final DataType fieldType;
+ private final FaissVectorIndexOptions options;
+ private volatile boolean metasLoaded = false;
+ private volatile boolean indicesLoaded = false;
+
+ public FaissVectorGlobalIndexReader(
+ GlobalIndexFileReader fileReader,
+ List<GlobalIndexIOMeta> ioMetas,
+ DataType fieldType,
+ FaissVectorIndexOptions options) {
+ this.fileReader = fileReader;
+ this.ioMetas = ioMetas;
+ this.fieldType = fieldType;
+ this.options = options;
+ this.indices = new ArrayList<>();
+ this.indexMetas = new ArrayList<>();
+ this.localIndexFiles = new ArrayList<>();
+ }
+
+ @Override
+ public Optional<GlobalIndexResult> visitVectorSearch(VectorSearch vectorSearch) {
+ try {
+ // First load only metadata
+ ensureLoadMetas();
+
+ RoaringNavigableMap64 includeRowIds = vectorSearch.includeRowIds();
+
+ // If includeRowIds is specified, check which indices contain matching rows
+ if (includeRowIds != null) {
+ List<Integer> matchingIndices = new ArrayList<>();
+ for (int i = 0; i < indexMetas.size(); i++) {
+ FaissIndexMeta meta = indexMetas.get(i);
+ // Check if the index's rowId range overlaps with includeRowIds
+ if (hasOverlap(meta.minId(), meta.maxId(), includeRowIds)) {
+ matchingIndices.add(i);
+ }
+ }
+ // If no index contains matching rowIds, return empty
+ if (matchingIndices.isEmpty()) {
+ return Optional.empty();
+ }
+ // Load only matching indices
+ ensureLoadIndices(matchingIndices);
+ } else {
+ // Load all indices
+ ensureLoadAllIndices();
+ }
+
+ return Optional.ofNullable(search(vectorSearch));
+ } catch (IOException e) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to search FAISS vector index with fieldName=%s, limit=%d",
+ vectorSearch.fieldName(), vectorSearch.limit()),
+ e);
+ }
+ }
+
+ /** Check if the range [minId, maxId] has any overlap with includeRowIds. */
+ private boolean hasOverlap(long minId, long maxId, RoaringNavigableMap64 includeRowIds) {
+ // Check if any ID in includeRowIds falls within [minId, maxId]
+ for (Long id : includeRowIds) {
+ if (id >= minId && id <= maxId) {
+ return true;
+ }
+ // Since iterator returns sorted values, we can stop early if we've passed maxId
+ if (id > maxId) {
+ break;
+ }
+ }
+ return false;
+ }
+
+ private GlobalIndexResult search(VectorSearch vectorSearch) throws IOException {
+ validateVectorType(vectorSearch.vector());
+ float[] queryVector = ((float[]) vectorSearch.vector()).clone();
+ // L2 normalize the query vector if enabled
+ if (options.normalize()) {
+ normalizeL2(queryVector);
+ }
+ int limit = vectorSearch.limit();
+
+ // Collect results from all indices using a min-heap
+ PriorityQueue<ScoredRow> result =
+ new PriorityQueue<>(Comparator.comparingDouble(sr -> sr.score));
+
+ RoaringNavigableMap64 includeRowIds = vectorSearch.includeRowIds();
+
+ // When filtering is enabled, we need to fetch more results to ensure
+ // we have enough after filtering. Use a multiplier based on index size.
+ int searchK = limit;
+ if (includeRowIds != null) {
+ // Fetch more results when filtering - up to searchFactor * limit or all filtered IDs
+ searchK =
+ Math.max(
+ limit * options.searchFactor(),
+ (int) includeRowIds.getLongCardinality());
+ }
+
+ for (FaissIndex index : indices) {
+ // Skip null indices (not loaded)
+ if (index == null) {
+ continue;
+ }
+
+ // Configure search parameters based on index type
+ configureSearchParams(index);
+
+ // Limit searchK to the index size
+ int effectiveK = (int) Math.min(searchK, index.size());
+ if (effectiveK <= 0) {
+ continue;
+ }
+
+ // Allocate result arrays
+ float[] distances = new float[effectiveK];
+ long[] labels = new long[effectiveK];
+
+ // Perform search
+ index.search(queryVector, 1, effectiveK, distances, labels);
+
+ for (int i = 0; i < effectiveK; i++) {
+ long rowId = labels[i];
+ if (rowId < 0) {
+ // Invalid result (not enough neighbors)
+ continue;
+ }
+
+ // Filter by include row IDs if specified
+ if (includeRowIds != null && !includeRowIds.contains(rowId)) {
+ continue;
+ }
+
+ // Convert distance to score (higher is better for similarity)
+ float score = convertDistanceToScore(distances[i]);
+
+ if (result.size() < limit) {
+ result.offer(new ScoredRow(rowId, score));
+ } else {
+ if (result.peek() != null && score > result.peek().score) {
+ result.poll();
+ result.offer(new ScoredRow(rowId, score));
+ }
+ }
+ }
+ }
+
+ RoaringNavigableMap64 roaringBitmap64 = new RoaringNavigableMap64();
+ HashMap<Long, Float> id2scores = new HashMap<>(result.size());
+ for (ScoredRow scoredRow : result) {
+ id2scores.put(scoredRow.rowId, scoredRow.score);
+ roaringBitmap64.add(scoredRow.rowId);
+ }
+ return new FaissVectorSearchGlobalIndexResult(roaringBitmap64, id2scores);
+ }
+
+ private void configureSearchParams(FaissIndex index) {
+ switch (index.indexType()) {
+ case HNSW:
+ index.setHnswEfSearch(options.efSearch());
+ break;
+ case IVF:
+ case IVF_PQ:
+ case IVF_SQ8:
+ // For small indices, use higher nprobe to ensure enough results
+ // Use at least nprobe or 10% of index size, whichever is larger
+ int effectiveNprobe =
+ Math.max(options.nprobe(), (int) Math.max(1, index.size() / 10));
+ index.setIvfNprobe(effectiveNprobe);
+ break;
+ default:
+ // No special configuration needed
+ break;
+ }
+ }
+
+ private float convertDistanceToScore(float distance) {
+ // For L2 distance, smaller is better, so we invert it
+ // For inner product, larger is better (already a similarity)
+ if (options.metric() == FaissVectorMetric.L2) {
+ // Convert L2 distance to similarity score
+ return 1.0f / (1.0f + distance);
+ } else {
+ // Inner product is already a similarity
+ return distance;
+ }
+ }
+
+ private void validateVectorType(Object vector) {
+ if (!(vector instanceof float[])) {
+ throw new IllegalArgumentException(
+ "Expected float[] vector but got: " + vector.getClass());
+ }
+ if (!(fieldType instanceof ArrayType)
+ || !(((ArrayType) fieldType).getElementType() instanceof FloatType)) {
+ throw new IllegalArgumentException(
+ "FAISS currently only supports float arrays, but field type is: " + fieldType);
+ }
+ }
+
+ /** Load only metadata from all index files. */
+ private void ensureLoadMetas() throws IOException {
+ if (!metasLoaded) {
+ synchronized (this) {
+ if (!metasLoaded) {
+ for (GlobalIndexIOMeta ioMeta : ioMetas) {
+ byte[] metaBytes = ioMeta.metadata();
+ FaissIndexMeta meta = FaissIndexMeta.deserialize(metaBytes);
+ indexMetas.add(meta);
+ }
+ metasLoaded = true;
+ }
+ }
+ }
+ }
+
+ /** Load all indices. */
+ private void ensureLoadAllIndices() throws IOException {
+ if (!indicesLoaded) {
+ synchronized (this) {
+ if (!indicesLoaded) {
+ for (int i = 0; i < ioMetas.size(); i++) {
+ loadIndexAt(i);
+ }
+ indicesLoaded = true;
+ }
+ }
+ }
+ }
+
+ /** Load only the specified indices by their positions. */
+ private void ensureLoadIndices(List<Integer> positions) throws IOException {
+ synchronized (this) {
+ // Ensure indices list is large enough
+ while (indices.size() < ioMetas.size()) {
+ indices.add(null);
+ }
+ for (int pos : positions) {
+ if (indices.get(pos) == null) {
+ loadIndexAt(pos);
+ }
+ }
+ }
+ }
+
+ /** Load a single index at the specified position. */
+ private void loadIndexAt(int position) throws IOException {
+ GlobalIndexIOMeta ioMeta = ioMetas.get(position);
+ FaissIndex index = null;
+ try (SeekableInputStream in = fileReader.getInputStream(ioMeta.fileName())) {
+ index = loadIndex(in);
+ if (indices.size() <= position) {
+ while (indices.size() < position) {
+ indices.add(null);
+ }
+ indices.add(index);
+ } else {
+ indices.set(position, index);
+ }
+ } catch (Exception e) {
+ IOUtils.closeQuietly(index);
+ throw e;
+ }
+ }
+
+ private FaissIndex loadIndex(SeekableInputStream in) throws IOException {
+
+ // Create a temp file for the raw FAISS index data (for mmap loading)
+ File rawIndexFile =
+ Files.createTempFile("paimon-faiss-" + UUID.randomUUID(), ".faiss").toFile();
+ localIndexFiles.add(rawIndexFile);
+
+ // Copy raw FAISS index data to the temp file
+ try (FileOutputStream fos = new FileOutputStream(rawIndexFile)) {
+ byte[] buffer = new byte[32768];
+ int bytesRead;
+ while ((bytesRead = in.read(buffer)) != -1) {
+ fos.write(buffer, 0, bytesRead);
+ }
+ }
+ // Load via mmap from the raw index file
+ return FaissIndex.fromFile(rawIndexFile);
+ }
+
+ /**
+ * L2 normalize the vector in place.
+ *
+ * @param vector the vector to normalize
+ */
+ private void normalizeL2(float[] vector) {
+ float norm = 0.0f;
+ for (float v : vector) {
+ norm += v * v;
+ }
+ norm = (float) Math.sqrt(norm);
+ if (norm > 0) {
+ for (int i = 0; i < vector.length; i++) {
+ vector[i] /= norm;
+ }
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ Throwable firstException = null;
+
+ // Close all FAISS indices
+ for (FaissIndex index : indices) {
+ if (index == null) {
+ continue;
+ }
+ try {
+ index.close();
+ } catch (Throwable t) {
+ if (firstException == null) {
+ firstException = t;
+ } else {
+ firstException.addSuppressed(t);
+ }
+ }
+ }
+ indices.clear();
+
+ // Delete local temporary files
+ for (File localFile : localIndexFiles) {
+ try {
+ if (localFile != null && localFile.exists()) {
+ localFile.delete();
+ }
+ } catch (Throwable t) {
+ if (firstException == null) {
+ firstException = t;
+ } else {
+ firstException.addSuppressed(t);
+ }
+ }
+ }
+ localIndexFiles.clear();
+
+ if (firstException != null) {
+ if (firstException instanceof IOException) {
+ throw (IOException) firstException;
+ } else if (firstException instanceof RuntimeException) {
+ throw (RuntimeException) firstException;
+ } else {
+ throw new RuntimeException(
+ "Failed to close FAISS vector global index reader", firstException);
+ }
+ }
+ }
+
+ /** Helper class to store row ID with its score. */
+ private static class ScoredRow {
+ final long rowId;
+ final float score;
+
+ ScoredRow(long rowId, float score) {
+ this.rowId = rowId;
+ this.score = score;
+ }
+ }
+
+ // =================== unsupported =====================
+
+ @Override
+ public Optional<GlobalIndexResult> visitIsNotNull(FieldRef fieldRef) {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<GlobalIndexResult> visitIsNull(FieldRef fieldRef) {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<GlobalIndexResult> visitStartsWith(FieldRef fieldRef, Object literal) {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<GlobalIndexResult> visitEndsWith(FieldRef fieldRef, Object literal) {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<GlobalIndexResult> visitContains(FieldRef fieldRef, Object literal) {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<GlobalIndexResult> visitLike(FieldRef fieldRef, Object literal) {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<GlobalIndexResult> visitLessThan(FieldRef fieldRef, Object literal) {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<GlobalIndexResult> visitGreaterOrEqual(FieldRef fieldRef, Object literal) {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<GlobalIndexResult> visitNotEqual(FieldRef fieldRef, Object literal) {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<GlobalIndexResult> visitLessOrEqual(FieldRef fieldRef, Object literal) {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<GlobalIndexResult> visitEqual(FieldRef fieldRef, Object literal) {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<GlobalIndexResult> visitGreaterThan(FieldRef fieldRef, Object literal) {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<GlobalIndexResult> visitIn(FieldRef fieldRef, List<Object> literals) {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<GlobalIndexResult> visitNotIn(FieldRef fieldRef, List<Object> literals) {
+ return Optional.empty();
+ }
+}
diff --git a/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissVectorGlobalIndexWriter.java b/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissVectorGlobalIndexWriter.java
new file mode 100644
index 0000000000..aed49f4cc3
--- /dev/null
+++ b/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissVectorGlobalIndexWriter.java
@@ -0,0 +1,307 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.faiss.index;
+
+import org.apache.paimon.data.InternalArray;
+import org.apache.paimon.globalindex.GlobalIndexSingletonWriter;
+import org.apache.paimon.globalindex.ResultEntry;
+import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.FloatType;
+
+import java.io.BufferedOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.FloatBuffer;
+import java.nio.LongBuffer;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Vector global index writer using FAISS.
+ *
+ * <p>This implementation uses FAISS for efficient approximate nearest neighbor search with support
+ * for multiple index types including Flat, HNSW, IVF, and IVF-PQ.
+ *
+ * <p>Vectors are added to the index in batches. When the current index reaches {@code sizePerIndex}
+ * vectors, it is serialized to a file and a new index is created.
+ */
+public class FaissVectorGlobalIndexWriter implements GlobalIndexSingletonWriter {
+
+ private static final int DEFAULT_BATCH_SIZE = 10000;
+
+ private final GlobalIndexFileWriter fileWriter;
+ private final FaissVectorIndexOptions options;
+ private final int sizePerIndex;
+ private final int batchSize;
+ private final int dim;
+ private final DataType fieldType;
+
+ private long count = 0;
+ private long currentIndexCount = 0;
+ private long currentIndexMinId = Long.MAX_VALUE;
+ private long currentIndexMaxId = Long.MIN_VALUE;
+ private final List<VectorEntry> pendingBatch;
+ private final List<ResultEntry> results;
+ private FaissIndex currentIndex;
+ private boolean trained = false;
+
+ public FaissVectorGlobalIndexWriter(
+ GlobalIndexFileWriter fileWriter, DataType fieldType, FaissVectorIndexOptions options) {
+ this.fileWriter = fileWriter;
+ this.fieldType = fieldType;
+ this.options = options;
+ this.sizePerIndex = options.sizePerIndex();
+ this.batchSize = Math.min(DEFAULT_BATCH_SIZE, sizePerIndex);
+ this.dim = options.dimension();
+ this.pendingBatch = new ArrayList<>(batchSize);
+ this.results = new ArrayList<>();
+
+ validateFieldType(fieldType);
+ }
+
+ private void validateFieldType(DataType dataType) {
+ if (!(dataType instanceof ArrayType)) {
+ throw new IllegalArgumentException(
+ "FAISS vector index requires ArrayType, but got: " + dataType);
+ }
+ DataType elementType = ((ArrayType) dataType).getElementType();
+ if (!(elementType instanceof FloatType)) {
+ throw new IllegalArgumentException(
+ "FAISS vector index requires float array, but got: " + elementType);
+ }
+ }
+
+ @Override
+ public void write(Object fieldData) {
+ float[] vector;
+ if (fieldData instanceof float[]) {
+ vector = (float[]) fieldData;
+ } else if (fieldData instanceof InternalArray) {
+ vector = ((InternalArray) fieldData).toFloatArray();
+ } else {
+ throw new RuntimeException(
+ "Unsupported vector type: " + fieldData.getClass().getName());
+ }
+ checkDimension(vector);
+ // L2 normalize the vector if enabled
+ if (options.normalize()) {
+ normalizeL2(vector);
+ }
+ // Track min/max IDs for current index
+ currentIndexMinId = Math.min(currentIndexMinId, count);
+ currentIndexMaxId = Math.max(currentIndexMaxId, count);
+ pendingBatch.add(new VectorEntry(count, vector));
+ count++;
+
+ try {
+ // When batch is full, add to current index
+ if (pendingBatch.size() >= batchSize) {
+ addBatchToIndex();
+ }
+ // When current index reaches sizePerIndex, flush to file
+ if (currentIndexCount >= sizePerIndex) {
+ flushCurrentIndex();
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public List<ResultEntry> finish() {
+ try {
+ // Add any remaining pending batch to current index
+ if (!pendingBatch.isEmpty()) {
+ addBatchToIndex();
+ }
+ // Flush current index if it has data
+ if (currentIndex != null && currentIndexCount > 0) {
+ flushCurrentIndex();
+ }
+ return results;
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to write FAISS vector global index", e);
+ }
+ }
+
+ private void addBatchToIndex() throws IOException {
+ if (pendingBatch.isEmpty()) {
+ return;
+ }
+
+ // Create index if not exists
+ if (currentIndex == null) {
+ currentIndex = createIndex();
+ trained = false;
+ }
+
+ int n = pendingBatch.size();
+
+ // Train if necessary (for IVF-based indices) - only once per index
+ if (!trained && !currentIndex.isTrained()) {
+ int trainingSize = Math.min(n, options.trainingSize());
+ ByteBuffer trainingBuffer = FaissIndex.allocateVectorBuffer(trainingSize, dim);
+ FloatBuffer trainingFloatView = trainingBuffer.asFloatBuffer();
+ for (int i = 0; i < trainingSize; i++) {
+ float[] vector = pendingBatch.get(i).vector;
+ for (int j = 0; j < dim; j++) {
+ trainingFloatView.put(i * dim + j, vector[j]);
+ }
+ }
+ currentIndex.train(trainingBuffer, trainingSize);
+ trained = true;
+ }
+
+ // Allocate buffers for this batch only
+ ByteBuffer vectorBuffer = FaissIndex.allocateVectorBuffer(n, dim);
+ ByteBuffer idBuffer = FaissIndex.allocateIdBuffer(n);
+ FloatBuffer floatView = vectorBuffer.asFloatBuffer();
+ LongBuffer longView = idBuffer.asLongBuffer();
+
+ // Fill buffers
+ for (int i = 0; i < n; i++) {
+ VectorEntry entry = pendingBatch.get(i);
+ float[] vector = entry.vector;
+ for (int j = 0; j < dim; j++) {
+ floatView.put(i * dim + j, vector[j]);
+ }
+ longView.put(i, entry.id);
+ }
+
+ // Add to index
+ currentIndex.addWithIds(vectorBuffer, idBuffer, n);
+ currentIndexCount += n;
+ pendingBatch.clear();
+ }
+
+ private void flushCurrentIndex() throws IOException {
+ if (currentIndex == null || currentIndexCount == 0) {
+ return;
+ }
+
+ String fileName = fileWriter.newFileName(FaissVectorGlobalIndexerFactory.IDENTIFIER);
+ try (OutputStream out = new BufferedOutputStream(fileWriter.newOutputStream(fileName))) {
+ // Serialize the index
+ long serializeSize = currentIndex.serializeSize();
+ if (serializeSize > Integer.MAX_VALUE) {
+ throw new IOException(
+ "Index too large to serialize: "
+ + serializeSize
+ + " bytes exceeds maximum buffer size");
+ }
+ ByteBuffer serializeBuffer =
+ ByteBuffer.allocateDirect((int) serializeSize).order(ByteOrder.nativeOrder());
+ long bytesWritten = currentIndex.serialize(serializeBuffer);
+
+ // Write only the raw serialized index data to file (no header)
+ byte[] indexData = new byte[(int) bytesWritten];
+ serializeBuffer.rewind();
+ serializeBuffer.get(indexData);
+ out.write(indexData);
+ out.flush();
+ }
+
+ // Create metadata and add to results
+ FaissIndexMeta meta =
+ new FaissIndexMeta(
+ dim,
+ options.metric().getValue(),
+ options.indexType().ordinal(),
+ currentIndexCount,
+ currentIndexMinId,
+ currentIndexMaxId);
+ results.add(new ResultEntry(fileName, currentIndexCount, meta.serialize()));
+
+ // Close and reset
+ currentIndex.close();
+ currentIndex = null;
+ currentIndexCount = 0;
+ currentIndexMinId = Long.MAX_VALUE;
+ currentIndexMaxId = Long.MIN_VALUE;
+ trained = false;
+ }
+
+ private FaissIndex createIndex() {
+ int dim = options.dimension();
+ FaissVectorMetric metric = options.metric();
+ // Auto-calculate nlist: min(configured nlist, sqrt(sizePerIndex))
+ // This ensures nlist is appropriate for the expected data size
+ int effectiveNlist = Math.max(1, Math.min(options.nlist(), (int) Math.sqrt(sizePerIndex)));
+
+ switch (options.indexType()) {
+ case FLAT:
+ return FaissIndex.createFlatIndex(dim, metric);
+ case HNSW:
+ return FaissIndex.createHnswIndex(
+ dim, options.m(), options.efConstruction(), metric);
+ case IVF:
+ return FaissIndex.createIvfIndex(dim, effectiveNlist, metric);
+ case IVF_PQ:
+ return FaissIndex.createIvfPqIndex(
+ dim, effectiveNlist, options.pqM(), options.pqNbits(), metric);
+ case IVF_SQ8:
+ return FaissIndex.createIvfSq8Index(dim, effectiveNlist, metric);
+ default:
+ throw new IllegalArgumentException(
+ "Unsupported index type: " + options.indexType());
+ }
+ }
+
+ private void checkDimension(float[] vector) {
+ if (vector.length != options.dimension()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Vector dimension mismatch: expected %d, but got %d",
+ options.dimension(), vector.length));
+ }
+ }
+
+ /**
+ * L2 normalize the vector in place.
+ *
+ * @param vector the vector to normalize
+ */
+ private void normalizeL2(float[] vector) {
+ float norm = 0.0f;
+ for (float v : vector) {
+ norm += v * v;
+ }
+ norm = (float) Math.sqrt(norm);
+ if (norm > 0) {
+ for (int i = 0; i < vector.length; i++) {
+ vector[i] /= norm;
+ }
+ }
+ }
+
+ /** Entry holding a vector and its row ID. */
+ private static class VectorEntry {
+ final long id;
+ final float[] vector;
+
+ VectorEntry(long id, float[] vector) {
+ this.id = id;
+ this.vector = vector;
+ }
+ }
+}
diff --git a/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissVectorGlobalIndexer.java b/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissVectorGlobalIndexer.java
new file mode 100644
index 0000000000..e554f2e7ee
--- /dev/null
+++ b/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissVectorGlobalIndexer.java
@@ -0,0 +1,53 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.faiss.index;
+
+import org.apache.paimon.globalindex.GlobalIndexIOMeta;
+import org.apache.paimon.globalindex.GlobalIndexReader;
+import org.apache.paimon.globalindex.GlobalIndexWriter;
+import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
+import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataType;
+
+import java.util.List;
+
+/** FAISS vector global indexer. */
+public class FaissVectorGlobalIndexer implements GlobalIndexer {
+
+ private final DataType fieldType;
+ private final FaissVectorIndexOptions options;
+
+ public FaissVectorGlobalIndexer(DataType fieldType, Options options) {
+ this.fieldType = fieldType;
+ this.options = new FaissVectorIndexOptions(options);
+ }
+
+ @Override
+ public GlobalIndexWriter createWriter(GlobalIndexFileWriter fileWriter) {
+ return new FaissVectorGlobalIndexWriter(fileWriter, fieldType, options);
+ }
+
+ @Override
+ public GlobalIndexReader createReader(
+ GlobalIndexFileReader fileReader, List<GlobalIndexIOMeta> files) {
+ return new FaissVectorGlobalIndexReader(fileReader, files, fieldType, options);
+ }
+}
diff --git a/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissVectorGlobalIndexerFactory.java b/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissVectorGlobalIndexerFactory.java
new file mode 100644
index 0000000000..d9c3bcd8b9
--- /dev/null
+++ b/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissVectorGlobalIndexerFactory.java
@@ -0,0 +1,40 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.faiss.index;
+
+import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.GlobalIndexerFactory;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataField;
+
+/** Factory for creating FAISS vector index. */
+public class FaissVectorGlobalIndexerFactory implements GlobalIndexerFactory {
+
+ public static final String IDENTIFIER = "faiss-vector-ann";
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public GlobalIndexer create(DataField field, Options options) {
+ return new FaissVectorGlobalIndexer(field.type(), options);
+ }
+}
diff --git a/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissVectorIndexOptions.java b/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissVectorIndexOptions.java
new file mode 100644
index 0000000000..9a1e56192b
--- /dev/null
+++ b/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissVectorIndexOptions.java
@@ -0,0 +1,210 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.faiss.index;
+
+import org.apache.paimon.options.ConfigOption;
+import org.apache.paimon.options.ConfigOptions;
+import org.apache.paimon.options.Options;
+
+/** Options for FAISS vector index. */
+public class FaissVectorIndexOptions {
+
+ public static final ConfigOption<Integer> VECTOR_DIM =
+ ConfigOptions.key("vector.dim")
+ .intType()
+ .defaultValue(128)
+ .withDescription("The dimension of the vector");
+
+ public static final ConfigOption<FaissVectorMetric> VECTOR_METRIC =
+ ConfigOptions.key("vector.metric")
+ .enumType(FaissVectorMetric.class)
+ .defaultValue(FaissVectorMetric.L2)
+ .withDescription(
+ "The similarity metric for vector search (L2, INNER_PRODUCT), and L2 is the default");
+
+ public static final ConfigOption<FaissIndexType> VECTOR_INDEX_TYPE =
+ ConfigOptions.key("vector.index-type")
+ .enumType(FaissIndexType.class)
+ .defaultValue(FaissIndexType.IVF_SQ8)
+ .withDescription(
+ "The type of FAISS index to use (FLAT, HNSW, IVF, IVF_PQ, IVF_SQ8), and IVF_SQ8 is the default");
+
+ public static final ConfigOption<Integer> VECTOR_M =
+ ConfigOptions.key("vector.m")
+ .intType()
+ .defaultValue(32)
+ .withDescription(
+ "The maximum number of connections for each element in HNSW index");
+
+ public static final ConfigOption<Integer> VECTOR_EF_CONSTRUCTION =
+ ConfigOptions.key("vector.ef-construction")
+ .intType()
+ .defaultValue(40)
+ .withDescription(
+ "The size of the dynamic candidate list during HNSW index construction");
+
+ public static final ConfigOption<Integer> VECTOR_EF_SEARCH =
+ ConfigOptions.key("vector.ef-search")
+ .intType()
+ .defaultValue(16)
+ .withDescription("The size of the dynamic candidate list during HNSW search");
+
+ public static final ConfigOption<Integer> VECTOR_NLIST =
+ ConfigOptions.key("vector.nlist")
+ .intType()
+ .defaultValue(100)
+ .withDescription("The number of inverted lists (clusters) for IVF index");
+
+ public static final ConfigOption<Integer> VECTOR_NPROBE =
+ ConfigOptions.key("vector.nprobe")
+ .intType()
+ .defaultValue(64)
+ .withDescription("The number of clusters to visit during IVF search");
+
+ public static final ConfigOption<Integer> VECTOR_PQ_M =
+ ConfigOptions.key("vector.pq-m")
+ .intType()
+ .defaultValue(8)
+ .withDescription("The number of sub-quantizers for IVF-PQ index");
+
+ public static final ConfigOption<Integer> VECTOR_PQ_NBITS =
+ ConfigOptions.key("vector.pq-nbits")
+ .intType()
+ .defaultValue(8)
+ .withDescription("The number of bits per sub-quantizer for IVF-PQ index");
+
+ public static final ConfigOption<Integer> VECTOR_SIZE_PER_INDEX =
+ ConfigOptions.key("vector.size-per-index")
+ .intType()
+ .defaultValue(200_0000)
+ .withDescription("The size of vectors stored in each vector index file");
+
+ public static final ConfigOption<Integer> VECTOR_TRAINING_SIZE =
+ ConfigOptions.key("vector.training-size")
+ .intType()
+ .defaultValue(50_0000)
+ .withDescription("The number of vectors to use for training IVF-based indices");
+
+ public static final ConfigOption<Integer> VECTOR_SEARCH_FACTOR =
+ ConfigOptions.key("vector.search-factor")
+ .intType()
+ .defaultValue(10)
+ .withDescription(
+ "The multiplier for the search limit when filtering is applied. "
+ + "This is used to fetch more results to ensure enough records after filtering.");
+
+ public static final ConfigOption<Boolean> VECTOR_NORMALIZE =
+ ConfigOptions.key("vector.normalize")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to L2 normalize vectors before indexing and searching. "
+ + "Enabling this converts L2 distance to cosine similarity.");
+
+ private final int dimension;
+ private final FaissVectorMetric metric;
+ private final FaissIndexType indexType;
+ private final int m;
+ private final int efConstruction;
+ private final int efSearch;
+ private final int nlist;
+ private final int nprobe;
+ private final int pqM;
+ private final int pqNbits;
+ private final int sizePerIndex;
+ private final int trainingSize;
+ private final int searchFactor;
+ private final boolean normalize;
+
+ public FaissVectorIndexOptions(Options options) {
+ this.dimension = options.get(VECTOR_DIM);
+ this.metric = options.get(VECTOR_METRIC);
+ this.indexType = options.get(VECTOR_INDEX_TYPE);
+ this.m = options.get(VECTOR_M);
+ this.efConstruction = options.get(VECTOR_EF_CONSTRUCTION);
+ this.efSearch = options.get(VECTOR_EF_SEARCH);
+ this.nlist = options.get(VECTOR_NLIST);
+ this.nprobe = options.get(VECTOR_NPROBE);
+ this.pqM = options.get(VECTOR_PQ_M);
+ this.pqNbits = options.get(VECTOR_PQ_NBITS);
+ this.sizePerIndex =
+ options.get(VECTOR_SIZE_PER_INDEX) > 0
+ ? options.get(VECTOR_SIZE_PER_INDEX)
+ : VECTOR_SIZE_PER_INDEX.defaultValue();
+ this.trainingSize = options.get(VECTOR_TRAINING_SIZE);
+ this.searchFactor = options.get(VECTOR_SEARCH_FACTOR);
+ this.normalize = options.get(VECTOR_NORMALIZE);
+ }
+
+ public int dimension() {
+ return dimension;
+ }
+
+ public FaissVectorMetric metric() {
+ return metric;
+ }
+
+ public FaissIndexType indexType() {
+ return indexType;
+ }
+
+ public int m() {
+ return m;
+ }
+
+ public int efConstruction() {
+ return efConstruction;
+ }
+
+ public int efSearch() {
+ return efSearch;
+ }
+
+ public int nlist() {
+ return nlist;
+ }
+
+ public int nprobe() {
+ return nprobe;
+ }
+
+ public int pqM() {
+ return pqM;
+ }
+
+ public int pqNbits() {
+ return pqNbits;
+ }
+
+ public int sizePerIndex() {
+ return sizePerIndex;
+ }
+
+ public int trainingSize() {
+ return trainingSize;
+ }
+
+ public int searchFactor() {
+ return searchFactor;
+ }
+
+ public boolean normalize() {
+ return normalize;
+ }
+}
diff --git a/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissVectorMetric.java b/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissVectorMetric.java
new file mode 100644
index 0000000000..49b492a1d9
--- /dev/null
+++ b/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissVectorMetric.java
@@ -0,0 +1,78 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.faiss.index;
+
+/** Enumeration of supported FAISS vector similarity metrics. */
+public enum FaissVectorMetric {
+
+ /**
+ * L2 (Euclidean) distance metric.
+ *
+ * <p>This is the default metric for FAISS. Lower values indicate more similar vectors.
+ */
+ L2(0),
+
+ /**
+ * Inner product (dot product) metric.
+ *
+ * <p>Higher values indicate more similar vectors. For normalized vectors, this is equivalent to
+ * cosine similarity.
+ */
+ INNER_PRODUCT(1);
+
+ private final int value;
+
+ FaissVectorMetric(int value) {
+ this.value = value;
+ }
+
+ /**
+ * Get the numeric value used by FAISS.
+ *
+ * @return the metric value
+ */
+ public int getValue() {
+ return value;
+ }
+
+ /**
+ * Convert a string to a metric.
+ *
+ * @param name the metric name
+ * @return the metric
+ */
+ public static FaissVectorMetric fromString(String name) {
+ return valueOf(name.toUpperCase());
+ }
+
+ /**
+ * Convert a numeric value to a metric.
+ *
+ * @param value the metric value
+ * @return the metric
+ */
+ public static FaissVectorMetric fromValue(int value) {
+ for (FaissVectorMetric metric : values()) {
+ if (metric.value == value) {
+ return metric;
+ }
+ }
+ throw new IllegalArgumentException("Unknown metric value: " + value);
+ }
+}
diff --git a/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissVectorSearchGlobalIndexResult.java b/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissVectorSearchGlobalIndexResult.java
new file mode 100644
index 0000000000..68de17ec6e
--- /dev/null
+++ b/paimon-faiss/paimon-faiss-index/src/main/java/org/apache/paimon/faiss/index/FaissVectorSearchGlobalIndexResult.java
@@ -0,0 +1,48 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.faiss.index;
+
+import org.apache.paimon.globalindex.ScoreGetter;
+import org.apache.paimon.globalindex.VectorSearchGlobalIndexResult;
+import org.apache.paimon.utils.RoaringNavigableMap64;
+
+import java.util.HashMap;
+
+/** Vector search global index result for FAISS vector index. */
+public class FaissVectorSearchGlobalIndexResult implements VectorSearchGlobalIndexResult {
+
+ private final HashMap<Long, Float> id2scores;
+ private final RoaringNavigableMap64 results;
+
+ public FaissVectorSearchGlobalIndexResult(
+ RoaringNavigableMap64 results, HashMap<Long, Float> id2scores) {
+ this.id2scores = id2scores;
+ this.results = results;
+ }
+
+ @Override
+ public ScoreGetter scoreGetter() {
+ return id2scores::get;
+ }
+
+ @Override
+ public RoaringNavigableMap64 results() {
+ return this.results;
+ }
+}
diff --git a/paimon-faiss/paimon-faiss-index/src/main/resources/META-INF/services/org.apache.paimon.globalindex.GlobalIndexerFactory b/paimon-faiss/paimon-faiss-index/src/main/resources/META-INF/services/org.apache.paimon.globalindex.GlobalIndexerFactory
new file mode 100644
index 0000000000..bbf3981d4e
--- /dev/null
+++ b/paimon-faiss/paimon-faiss-index/src/main/resources/META-INF/services/org.apache.paimon.globalindex.GlobalIndexerFactory
@@ -0,0 +1,18 @@
+# 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.
+
+org.apache.paimon.faiss.index.FaissVectorGlobalIndexerFactory
+
+
diff --git a/paimon-faiss/paimon-faiss-index/src/test/java/org/apache/paimon/faiss/index/FaissVectorGlobalIndexScanTest.java b/paimon-faiss/paimon-faiss-index/src/test/java/org/apache/paimon/faiss/index/FaissVectorGlobalIndexScanTest.java
new file mode 100644
index 0000000000..d0b9fe68d4
--- /dev/null
+++ b/paimon-faiss/paimon-faiss-index/src/test/java/org/apache/paimon/faiss/index/FaissVectorGlobalIndexScanTest.java
@@ -0,0 +1,343 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.faiss.index;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.faiss.Faiss;
+import org.apache.paimon.faiss.FaissException;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.globalindex.ResultEntry;
+import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.VectorSearch;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamTableWrite;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Assumptions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for scanning FAISS vector global index. */
+public class FaissVectorGlobalIndexScanTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ private FileStoreTable table;
+ private String commitUser;
+ private FileIO fileIO;
+ private RowType rowType;
+ private final String vectorFieldName = "vec";
+
+ @BeforeEach
+ public void before() throws Exception {
+ // Skip tests if FAISS native library is not available
+ if (!Faiss.isLibraryLoaded()) {
+ try {
+ Faiss.loadLibrary();
+ } catch (FaissException e) {
+ StringBuilder errorMsg = new StringBuilder("FAISS native library not available.");
+ errorMsg.append("\nError: ").append(e.getMessage());
+ if (e.getCause() != null) {
+ errorMsg.append("\nCause: ").append(e.getCause().getMessage());
+ }
+ errorMsg.append(
+ "\n\nTo run FAISS tests, ensure the paimon-faiss-jni JAR"
+ + " with native libraries is available in the classpath.");
+ Assumptions.assumeTrue(false, errorMsg.toString());
+ }
+ }
+
+ Path tablePath = new Path(tempDir.toString());
+ fileIO = new LocalFileIO();
+ SchemaManager schemaManager = new SchemaManager(fileIO, tablePath);
+
+ String similarityMetric = "L2";
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column(vectorFieldName, new ArrayType(DataTypes.FLOAT()))
+ .option(CoreOptions.BUCKET.key(), "-1")
+ .option("vector.dim", "2")
+ .option("vector.metric", similarityMetric)
+ .option("vector.index-type", "HNSW")
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true")
+ .build();
+
+ TableSchema tableSchema = schemaManager.createTable(schema);
+ table = FileStoreTableFactory.create(fileIO, tablePath, tableSchema);
+ rowType = table.rowType();
+ commitUser = UUID.randomUUID().toString();
+ }
+
+ @Test
+ public void testVectorIndexScanEndToEnd() throws Exception {
+ float[][] vectors =
+ new float[][] {
+ new float[] {1.0f, 0.0f}, new float[] {0.95f, 0.1f}, new float[] {0.1f, 0.95f},
+ new float[] {0.98f, 0.05f}, new float[] {0.0f, 1.0f}, new float[] {0.05f, 0.98f}
+ };
+
+ // 1. Write data using Paimon API
+ writeVectors(vectors);
+
+ // 2. Manually build vector index
+ List<IndexFileMeta> indexFiles = buildIndexManually(vectors);
+
+ // 3. Commit index files to the Table (Update Index Manifest)
+ commitIndex(indexFiles);
+
+ // 4. Verify results
+ float[] queryVector = new float[] {0.85f, 0.15f};
+ VectorSearch vectorSearch = new VectorSearch(queryVector, 2, vectorFieldName);
+ ReadBuilder readBuilder = table.newReadBuilder().withVectorSearch(vectorSearch);
+ TableScan scan = readBuilder.newScan();
+ List<Integer> ids = new ArrayList<>();
+ readBuilder
+ .newRead()
+ .createReader(scan.plan())
+ .forEachRemaining(
+ row -> {
+ ids.add(row.getInt(0));
+ });
+ // With L2 distance, the closest vectors to [0.85, 0.15] should be [0.95, 0.1] and [0.98,
+ // 0.05]
+ assertThat(ids).containsExactlyInAnyOrder(1, 3);
+ }
+
+ @Test
+ public void testVectorIndexScanWithDifferentMetrics() throws Exception {
+ // Re-create table with INNER_PRODUCT metric
+ Path tablePath = new Path(tempDir.toString(), "inner_product");
+ fileIO.mkdirs(tablePath);
+ SchemaManager schemaManager = new SchemaManager(fileIO, tablePath);
+
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column(vectorFieldName, new ArrayType(DataTypes.FLOAT()))
+ .option(CoreOptions.BUCKET.key(), "-1")
+ .option("vector.dim", "2")
+ .option("vector.metric", "INNER_PRODUCT")
+ .option("vector.index-type", "FLAT")
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true")
+ .build();
+
+ TableSchema tableSchema = schemaManager.createTable(schema);
+ FileStoreTable ipTable = FileStoreTableFactory.create(fileIO, tablePath, tableSchema);
+ String ipCommitUser = UUID.randomUUID().toString();
+
+ // Normalized vectors for inner product
+ float[][] vectors =
+ new float[][] {
+ new float[] {1.0f, 0.0f},
+ new float[] {0.707f, 0.707f},
+ new float[] {0.0f, 1.0f},
+ };
+
+ // Write data
+ StreamTableWrite write = ipTable.newWrite(ipCommitUser);
+ for (int i = 0; i < vectors.length; i++) {
+ write.write(GenericRow.of(i, new GenericArray(vectors[i])));
+ }
+ List<CommitMessage> messages = write.prepareCommit(false, 0);
+ StreamTableCommit commit = ipTable.newCommit(ipCommitUser);
+ commit.commit(0, messages);
+ write.close();
+
+ // Build index
+ Options options = new Options(ipTable.options());
+ FaissVectorIndexOptions indexOptions = new FaissVectorIndexOptions(options);
+ Path indexDir = ipTable.store().pathFactory().indexPath();
+ if (!fileIO.exists(indexDir)) {
+ fileIO.mkdirs(indexDir);
+ }
+
+ GlobalIndexFileWriter fileWriter =
+ new GlobalIndexFileWriter() {
+ @Override
+ public String newFileName(String prefix) {
+ return prefix + "-" + UUID.randomUUID();
+ }
+
+ @Override
+ public PositionOutputStream newOutputStream(String fileName)
+ throws IOException {
+ return fileIO.newOutputStream(new Path(indexDir, fileName), false);
+ }
+ };
+
+ FaissVectorGlobalIndexWriter indexWriter =
+ new FaissVectorGlobalIndexWriter(
+ fileWriter, new ArrayType(DataTypes.FLOAT()), indexOptions);
+ for (float[] vec : vectors) {
+ indexWriter.write(vec);
+ }
+
+ List<ResultEntry> entries = indexWriter.finish();
+ List<IndexFileMeta> metas = new ArrayList<>();
+ int fieldId = ipTable.rowType().getFieldIndex(vectorFieldName);
+
+ for (ResultEntry entry : entries) {
+ long fileSize = fileIO.getFileSize(new Path(indexDir, entry.fileName()));
+ GlobalIndexMeta globalMeta =
+ new GlobalIndexMeta(0, vectors.length - 1, fieldId, null, entry.meta());
+
+ metas.add(
+ new IndexFileMeta(
+ FaissVectorGlobalIndexerFactory.IDENTIFIER,
+ entry.fileName(),
+ fileSize,
+ entry.rowCount(),
+ globalMeta));
+ }
+
+ // Commit index
+ DataIncrement dataIncrement = DataIncrement.indexIncrement(metas);
+ CommitMessage message =
+ new CommitMessageImpl(
+ BinaryRow.EMPTY_ROW,
+ 0,
+ 1,
+ dataIncrement,
+ CompactIncrement.emptyIncrement());
+ ipTable.newCommit(ipCommitUser).commit(1, Collections.singletonList(message));
+
+ // Query - for inner product, [1, 0] should be closest to [1, 0]
+ float[] queryVector = new float[] {1.0f, 0.0f};
+ VectorSearch vectorSearch = new VectorSearch(queryVector, 1, vectorFieldName);
+ ReadBuilder readBuilder = ipTable.newReadBuilder().withVectorSearch(vectorSearch);
+ TableScan scan = readBuilder.newScan();
+ List<Integer> ids = new ArrayList<>();
+ readBuilder
+ .newRead()
+ .createReader(scan.plan())
+ .forEachRemaining(
+ row -> {
+ ids.add(row.getInt(0));
+ });
+ assertThat(ids).containsExactly(0);
+ }
+
+ private void writeVectors(float[][] vectors) throws Exception {
+ StreamTableWrite write = table.newWrite(commitUser);
+ for (int i = 0; i < vectors.length; i++) {
+ write.write(GenericRow.of(i, new GenericArray(vectors[i])));
+ }
+ List<CommitMessage> messages = write.prepareCommit(false, 0);
+ StreamTableCommit commit = table.newCommit(commitUser);
+ commit.commit(0, messages);
+ write.close();
+ }
+
+ private List<IndexFileMeta> buildIndexManually(float[][] vectors) throws Exception {
+ Options options = new Options(table.options());
+ FaissVectorIndexOptions indexOptions = new FaissVectorIndexOptions(options);
+ Path indexDir = table.store().pathFactory().indexPath();
+ if (!fileIO.exists(indexDir)) {
+ fileIO.mkdirs(indexDir);
+ }
+
+ GlobalIndexFileWriter fileWriter =
+ new GlobalIndexFileWriter() {
+ @Override
+ public String newFileName(String prefix) {
+ return prefix + "-" + UUID.randomUUID();
+ }
+
+ @Override
+ public PositionOutputStream newOutputStream(String fileName)
+ throws IOException {
+ return fileIO.newOutputStream(new Path(indexDir, fileName), false);
+ }
+ };
+
+ FaissVectorGlobalIndexWriter writer =
+ new FaissVectorGlobalIndexWriter(
+ fileWriter, new ArrayType(DataTypes.FLOAT()), indexOptions);
+ for (float[] vec : vectors) {
+ writer.write(vec);
+ }
+
+ List<ResultEntry> entries = writer.finish();
+
+ List<IndexFileMeta> metas = new ArrayList<>();
+ int fieldId = rowType.getFieldIndex(vectorFieldName);
+
+ for (ResultEntry entry : entries) {
+ long fileSize = fileIO.getFileSize(new Path(indexDir, entry.fileName()));
+ GlobalIndexMeta globalMeta =
+ new GlobalIndexMeta(0, vectors.length - 1, fieldId, null, entry.meta());
+
+ metas.add(
+ new IndexFileMeta(
+ FaissVectorGlobalIndexerFactory.IDENTIFIER,
+ entry.fileName(),
+ fileSize,
+ entry.rowCount(),
+ globalMeta));
+ }
+ return metas;
+ }
+
+ private void commitIndex(List<IndexFileMeta> indexFiles) {
+ StreamTableCommit commit = table.newCommit(commitUser);
+ DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles);
+ CommitMessage message =
+ new CommitMessageImpl(
+ BinaryRow.EMPTY_ROW,
+ 0,
+ 1,
+ dataIncrement,
+ CompactIncrement.emptyIncrement());
+ commit.commit(1, Collections.singletonList(message));
+ }
+}
diff --git a/paimon-faiss/paimon-faiss-index/src/test/java/org/apache/paimon/faiss/index/FaissVectorGlobalIndexTest.java b/paimon-faiss/paimon-faiss-index/src/test/java/org/apache/paimon/faiss/index/FaissVectorGlobalIndexTest.java
new file mode 100644
index 0000000000..7311d95322
--- /dev/null
+++ b/paimon-faiss/paimon-faiss-index/src/test/java/org/apache/paimon/faiss/index/FaissVectorGlobalIndexTest.java
@@ -0,0 +1,520 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.faiss.index;
+
+import org.apache.paimon.faiss.Faiss;
+import org.apache.paimon.faiss.FaissException;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.globalindex.GlobalIndexIOMeta;
+import org.apache.paimon.globalindex.GlobalIndexResult;
+import org.apache.paimon.globalindex.ResultEntry;
+import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
+import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.VectorSearch;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.FloatType;
+import org.apache.paimon.utils.RoaringNavigableMap64;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assumptions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Random;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for {@link FaissVectorGlobalIndexWriter} and {@link FaissVectorGlobalIndexReader}. */
+public class FaissVectorGlobalIndexTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ private FileIO fileIO;
+ private Path indexPath;
+ private DataType vectorType;
+ private final String fieldName = "vec";
+
+ @BeforeEach
+ public void setup() {
+ // Skip tests if FAISS native library is not available
+ if (!Faiss.isLibraryLoaded()) {
+ try {
+ Faiss.loadLibrary();
+ } catch (FaissException e) {
+ StringBuilder errorMsg = new StringBuilder("FAISS native library not available.");
+ errorMsg.append("\nError: ").append(e.getMessage());
+ if (e.getCause() != null) {
+ errorMsg.append("\nCause: ").append(e.getCause().getMessage());
+ }
+ errorMsg.append(
+ "\n\nTo run FAISS tests, ensure the paimon-faiss-jni JAR"
+ + " with native libraries is available in the classpath.");
+ Assumptions.assumeTrue(false, errorMsg.toString());
+ }
+ }
+
+ fileIO = new LocalFileIO();
+ indexPath = new Path(tempDir.toString());
+ vectorType = new ArrayType(new FloatType());
+ }
+
+ @AfterEach
+ public void cleanup() throws IOException {
+ if (fileIO != null) {
+ fileIO.delete(indexPath, true);
+ }
+ }
+
+ private GlobalIndexFileWriter createFileWriter(Path path) {
+ return new GlobalIndexFileWriter() {
+ @Override
+ public String newFileName(String prefix) {
+ return prefix + "-" + UUID.randomUUID();
+ }
+
+ @Override
+ public PositionOutputStream newOutputStream(String fileName) throws IOException {
+ return fileIO.newOutputStream(new Path(path, fileName), false);
+ }
+ };
+ }
+
+ private GlobalIndexFileReader createFileReader(Path path) {
+ return new GlobalIndexFileReader() {
+ @Override
+ public SeekableInputStream getInputStream(String fileName) throws IOException {
+ return fileIO.newInputStream(new Path(path, fileName));
+ }
+
+ @Override
+ public Path filePath(String fileName) {
+ return new Path(path, fileName);
+ }
+ };
+ }
+
+ @Test
+ public void testDifferentMetrics() throws IOException {
+ int dimension = 32;
+ int numVectors = 20;
+
+ String[] metrics = {"L2", "INNER_PRODUCT"};
+
+ for (String metric : metrics) {
+ Options options = createDefaultOptions(dimension);
+ options.setString("vector.metric", metric);
+ FaissVectorIndexOptions indexOptions = new FaissVectorIndexOptions(options);
+ Path metricIndexPath = new Path(indexPath, metric.toLowerCase());
+ GlobalIndexFileWriter fileWriter = createFileWriter(metricIndexPath);
+ FaissVectorGlobalIndexWriter writer =
+ new FaissVectorGlobalIndexWriter(fileWriter, vectorType, indexOptions);
+
+ List<float[]> testVectors = generateRandomVectors(numVectors, dimension);
+ testVectors.forEach(writer::write);
+
+ List<ResultEntry> results = writer.finish();
+ assertThat(results).hasSize(1);
+
+ ResultEntry result = results.get(0);
+ GlobalIndexFileReader fileReader = createFileReader(metricIndexPath);
+ List<GlobalIndexIOMeta> metas = new ArrayList<>();
+ metas.add(
+ new GlobalIndexIOMeta(
+ result.fileName(),
+ fileIO.getFileSize(new Path(metricIndexPath, result.fileName())),
+ result.meta()));
+
+ try (FaissVectorGlobalIndexReader reader =
+ new FaissVectorGlobalIndexReader(fileReader, metas, vectorType, indexOptions)) {
+ VectorSearch vectorSearch = new VectorSearch(testVectors.get(0), 3, fieldName);
+ GlobalIndexResult searchResult = reader.visitVectorSearch(vectorSearch).get();
+ assertThat(searchResult).isNotNull();
+ }
+ }
+ }
+
+ @Test
+ public void testDifferentIndexTypes() throws IOException {
+ int dimension = 32;
+ int numVectors = 100; // More vectors for IVF training
+
+ String[] indexTypes = {"FLAT", "HNSW"};
+ // IVF and IVF_PQ require more training data, tested separately
+
+ for (String indexType : indexTypes) {
+ Options options = createDefaultOptions(dimension);
+ options.setString("vector.index-type", indexType);
+ FaissVectorIndexOptions indexOptions = new FaissVectorIndexOptions(options);
+ Path typeIndexPath = new Path(indexPath, indexType.toLowerCase());
+ GlobalIndexFileWriter fileWriter = createFileWriter(typeIndexPath);
+ FaissVectorGlobalIndexWriter writer =
+ new FaissVectorGlobalIndexWriter(fileWriter, vectorType, indexOptions);
+
+ List<float[]> testVectors = generateRandomVectors(numVectors, dimension);
+ testVectors.forEach(writer::write);
+
+ List<ResultEntry> results = writer.finish();
+ assertThat(results).hasSize(1);
+
+ ResultEntry result = results.get(0);
+ GlobalIndexFileReader fileReader = createFileReader(typeIndexPath);
+ List<GlobalIndexIOMeta> metas = new ArrayList<>();
+ metas.add(
+ new GlobalIndexIOMeta(
+ result.fileName(),
+ fileIO.getFileSize(new Path(typeIndexPath, result.fileName())),
+ result.meta()));
+
+ try (FaissVectorGlobalIndexReader reader =
+ new FaissVectorGlobalIndexReader(fileReader, metas, vectorType, indexOptions)) {
+ VectorSearch vectorSearch = new VectorSearch(testVectors.get(0), 5, fieldName);
+ GlobalIndexResult searchResult = reader.visitVectorSearch(vectorSearch).get();
+ assertThat(searchResult).isNotNull();
+ }
+ }
+ }
+
+ @Test
+ public void testDifferentDimensions() throws IOException {
+ int[] dimensions = {8, 32, 128, 256};
+
+ for (int dimension : dimensions) {
+ Options options = createDefaultOptions(dimension);
+ FaissVectorIndexOptions indexOptions = new FaissVectorIndexOptions(options);
+ Path dimIndexPath = new Path(indexPath, "dim_" + dimension);
+ GlobalIndexFileWriter fileWriter = createFileWriter(dimIndexPath);
+ FaissVectorGlobalIndexWriter writer =
+ new FaissVectorGlobalIndexWriter(fileWriter, vectorType, indexOptions);
+
+ int numVectors = 10;
+ List<float[]> testVectors = generateRandomVectors(numVectors, dimension);
+ testVectors.forEach(writer::write);
+
+ List<ResultEntry> results = writer.finish();
+ assertThat(results).hasSize(1);
+
+ ResultEntry result = results.get(0);
+ GlobalIndexFileReader fileReader = createFileReader(dimIndexPath);
+ List<GlobalIndexIOMeta> metas = new ArrayList<>();
+ metas.add(
+ new GlobalIndexIOMeta(
+ result.fileName(),
+ fileIO.getFileSize(new Path(dimIndexPath, result.fileName())),
+ result.meta()));
+
+ try (FaissVectorGlobalIndexReader reader =
+ new FaissVectorGlobalIndexReader(fileReader, metas, vectorType, indexOptions)) {
+ VectorSearch vectorSearch = new VectorSearch(testVectors.get(0), 5, fieldName);
+ GlobalIndexResult searchResult = reader.visitVectorSearch(vectorSearch).get();
+ assertThat(searchResult).isNotNull();
+ }
+ }
+ }
+
+ @Test
+ public void testDimensionMismatch() throws IOException {
+ Options options = createDefaultOptions(64);
+
+ GlobalIndexFileWriter fileWriter = createFileWriter(indexPath);
+ FaissVectorIndexOptions indexOptions = new FaissVectorIndexOptions(options);
+ FaissVectorGlobalIndexWriter writer =
+ new FaissVectorGlobalIndexWriter(fileWriter, vectorType, indexOptions);
+
+ // Try to write vector with wrong dimension
+ float[] wrongDimVector = new float[32]; // Wrong dimension
+ assertThatThrownBy(() -> writer.write(wrongDimVector))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("dimension mismatch");
+ }
+
+ @Test
+ public void testFloatVectorIndexEndToEnd() throws IOException {
+ int dimension = 2;
+ Options options = createDefaultOptions(dimension);
+ int sizePerIndex = 3;
+ options.setInteger("vector.size-per-index", sizePerIndex);
+
+ float[][] vectors =
+ new float[][] {
+ new float[] {1.0f, 0.0f}, new float[] {0.95f, 0.1f}, new float[] {0.1f, 0.95f},
+ new float[] {0.98f, 0.05f}, new float[] {0.0f, 1.0f}, new float[] {0.05f, 0.98f}
+ };
+
+ GlobalIndexFileWriter fileWriter = createFileWriter(indexPath);
+ FaissVectorIndexOptions indexOptions = new FaissVectorIndexOptions(options);
+ FaissVectorGlobalIndexWriter writer =
+ new FaissVectorGlobalIndexWriter(fileWriter, vectorType, indexOptions);
+ Arrays.stream(vectors).forEach(writer::write);
+
+ List<ResultEntry> results = writer.finish();
+ assertThat(results).hasSize(2);
+
+ GlobalIndexFileReader fileReader = createFileReader(indexPath);
+ List<GlobalIndexIOMeta> metas = new ArrayList<>();
+ for (ResultEntry result : results) {
+ metas.add(
+ new GlobalIndexIOMeta(
+ result.fileName(),
+ fileIO.getFileSize(new Path(indexPath, result.fileName())),
+ result.meta()));
+ }
+
+ try (FaissVectorGlobalIndexReader reader =
+ new FaissVectorGlobalIndexReader(fileReader, metas, vectorType, indexOptions)) {
+ VectorSearch vectorSearch = new VectorSearch(vectors[0], 1, fieldName);
+ FaissVectorSearchGlobalIndexResult result =
+ (FaissVectorSearchGlobalIndexResult)
+ reader.visitVectorSearch(vectorSearch).get();
+ assertThat(result.results().getLongCardinality()).isEqualTo(1);
+ long expectedRowId = 0;
+ assertThat(containsRowId(result, expectedRowId)).isTrue();
+
+ // Test with filter
+ expectedRowId = 1;
+ RoaringNavigableMap64 filterResults = new RoaringNavigableMap64();
+ filterResults.add(expectedRowId);
+ vectorSearch =
+ new VectorSearch(vectors[0], 1, fieldName).withIncludeRowIds(filterResults);
+ result =
+ (FaissVectorSearchGlobalIndexResult)
+ reader.visitVectorSearch(vectorSearch).get();
+ assertThat(containsRowId(result, expectedRowId)).isTrue();
+
+ // Test with multiple results
+ float[] queryVector = new float[] {0.85f, 0.15f};
+ vectorSearch = new VectorSearch(queryVector, 2, fieldName);
+ result =
+ (FaissVectorSearchGlobalIndexResult)
+ reader.visitVectorSearch(vectorSearch).get();
+ assertThat(result.results().getLongCardinality()).isEqualTo(2);
+ }
+ }
+
+ @Test
+ public void testInvalidTopK() {
+ assertThatThrownBy(() -> new VectorSearch(new float[] {0.1f}, 0, fieldName))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Limit must be positive");
+ }
+
+ @Test
+ public void testMultipleIndexFiles() throws IOException {
+ int dimension = 32;
+ Options options = createDefaultOptions(dimension);
+ options.setInteger("vector.size-per-index", 5); // Small batch size
+
+ GlobalIndexFileWriter fileWriter = createFileWriter(indexPath);
+ FaissVectorIndexOptions indexOptions = new FaissVectorIndexOptions(options);
+ FaissVectorGlobalIndexWriter writer =
+ new FaissVectorGlobalIndexWriter(fileWriter, vectorType, indexOptions);
+
+ int numVectors = 15; // Will create 3 index files
+ List<float[]> testVectors = generateRandomVectors(numVectors, dimension);
+ testVectors.forEach(writer::write);
+
+ List<ResultEntry> results = writer.finish();
+ assertThat(results).hasSize(3);
+
+ GlobalIndexFileReader fileReader = createFileReader(indexPath);
+ List<GlobalIndexIOMeta> metas = new ArrayList<>();
+ for (ResultEntry result : results) {
+ metas.add(
+ new GlobalIndexIOMeta(
+ result.fileName(),
+ fileIO.getFileSize(new Path(indexPath, result.fileName())),
+ result.meta()));
+ }
+
+ try (FaissVectorGlobalIndexReader reader =
+ new FaissVectorGlobalIndexReader(fileReader, metas, vectorType, indexOptions)) {
+ // Search should work across all index files
+ VectorSearch vectorSearch = new VectorSearch(testVectors.get(10), 3, fieldName);
+ GlobalIndexResult searchResult = reader.visitVectorSearch(vectorSearch).get();
+ assertThat(searchResult).isNotNull();
+ assertThat(searchResult.results().getLongCardinality()).isGreaterThan(0);
+ }
+ }
+
+ @Test
+ public void testBatchWriteMultipleFiles() throws IOException {
+ int dimension = 8;
+ Options options = createDefaultOptions(dimension);
+ int sizePerIndex = 100;
+ options.setInteger("vector.size-per-index", sizePerIndex);
+
+ GlobalIndexFileWriter fileWriter = createFileWriter(indexPath);
+ FaissVectorIndexOptions indexOptions = new FaissVectorIndexOptions(options);
+ FaissVectorGlobalIndexWriter writer =
+ new FaissVectorGlobalIndexWriter(fileWriter, vectorType, indexOptions);
+
+ // Write 350 vectors, should create 4 files (100 + 100 + 100 + 50)
+ int numVectors = 350;
+ List<float[]> testVectors = generateRandomVectors(numVectors, dimension);
+ testVectors.forEach(writer::write);
+
+ List<ResultEntry> results = writer.finish();
+ assertThat(results).hasSize(4);
+
+ // Verify files are created
+ GlobalIndexFileReader fileReader = createFileReader(indexPath);
+ List<GlobalIndexIOMeta> metas = new ArrayList<>();
+ for (ResultEntry result : results) {
+ Path filePath = new Path(indexPath, result.fileName());
+ assertThat(fileIO.exists(filePath)).isTrue();
+ assertThat(fileIO.getFileSize(filePath)).isGreaterThan(0);
+ metas.add(
+ new GlobalIndexIOMeta(
+ result.fileName(), fileIO.getFileSize(filePath), result.meta()));
+ }
+
+ // Search for vectors from different files
+ try (FaissVectorGlobalIndexReader reader =
+ new FaissVectorGlobalIndexReader(fileReader, metas, vectorType, indexOptions)) {
+ // Search for vector in first file (index 0-99)
+ VectorSearch vectorSearch = new VectorSearch(testVectors.get(50), 3, fieldName);
+ GlobalIndexResult searchResult = reader.visitVectorSearch(vectorSearch).get();
+ assertThat(searchResult).isNotNull();
+ assertThat(searchResult.results().getLongCardinality()).isGreaterThan(0);
+
+ // Search for vector in second file (index 100-199)
+ vectorSearch = new VectorSearch(testVectors.get(150), 3, fieldName);
+ searchResult = reader.visitVectorSearch(vectorSearch).get();
+ assertThat(searchResult).isNotNull();
+ assertThat(searchResult.results().getLongCardinality()).isGreaterThan(0);
+
+ // Search for vector in last file (index 300-349)
+ vectorSearch = new VectorSearch(testVectors.get(320), 3, fieldName);
+ searchResult = reader.visitVectorSearch(vectorSearch).get();
+ assertThat(searchResult).isNotNull();
+ assertThat(searchResult.results().getLongCardinality()).isGreaterThan(0);
+
+ // Verify that search returns vectors from the correct file
+ // The exact match should have rowId equal to the vector index
+ vectorSearch = new VectorSearch(testVectors.get(200), 1, fieldName);
+ FaissVectorSearchGlobalIndexResult result =
+ (FaissVectorSearchGlobalIndexResult)
+ reader.visitVectorSearch(vectorSearch).get();
+ assertThat(containsRowId(result, 200)).isTrue();
+ }
+ }
+
+ @Test
+ public void testBatchWriteWithRemainder() throws IOException {
+ int dimension = 16;
+ Options options = createDefaultOptions(dimension);
+ int sizePerIndex = 50;
+ options.setInteger("vector.size-per-index", sizePerIndex);
+
+ GlobalIndexFileWriter fileWriter = createFileWriter(indexPath);
+ FaissVectorIndexOptions indexOptions = new FaissVectorIndexOptions(options);
+ FaissVectorGlobalIndexWriter writer =
+ new FaissVectorGlobalIndexWriter(fileWriter, vectorType, indexOptions);
+
+ // Write 73 vectors: 50 in first file, 23 in second file
+ int numVectors = 73;
+ List<float[]> testVectors = generateRandomVectors(numVectors, dimension);
+ testVectors.forEach(writer::write);
+
+ List<ResultEntry> results = writer.finish();
+ assertThat(results).hasSize(2);
+
+ GlobalIndexFileReader fileReader = createFileReader(indexPath);
+ List<GlobalIndexIOMeta> metas = new ArrayList<>();
+ for (ResultEntry result : results) {
+ metas.add(
+ new GlobalIndexIOMeta(
+ result.fileName(),
+ fileIO.getFileSize(new Path(indexPath, result.fileName())),
+ result.meta()));
+ }
+
+ try (FaissVectorGlobalIndexReader reader =
+ new FaissVectorGlobalIndexReader(fileReader, metas, vectorType, indexOptions)) {
+ // Search for vector in the remainder file (second file)
+ VectorSearch vectorSearch = new VectorSearch(testVectors.get(60), 1, fieldName);
+ FaissVectorSearchGlobalIndexResult result =
+ (FaissVectorSearchGlobalIndexResult)
+ reader.visitVectorSearch(vectorSearch).get();
+ assertThat(result).isNotNull();
+ assertThat(containsRowId(result, 60)).isTrue();
+
+ // Search for the last vector
+ vectorSearch = new VectorSearch(testVectors.get(72), 1, fieldName);
+ result =
+ (FaissVectorSearchGlobalIndexResult)
+ reader.visitVectorSearch(vectorSearch).get();
+ assertThat(result).isNotNull();
+ assertThat(containsRowId(result, 72)).isTrue();
+ }
+ }
+
+ private Options createDefaultOptions(int dimension) {
+ Options options = new Options();
+ options.setInteger("vector.dim", dimension);
+ options.setString("vector.metric", "L2");
+ options.setString("vector.index-type", "HNSW");
+ options.setInteger("vector.m", 32);
+ options.setInteger("vector.ef-construction", 40);
+ options.setInteger("vector.ef-search", 16);
+ return options;
+ }
+
+ private List<float[]> generateRandomVectors(int count, int dimension) {
+ Random random = new Random(42);
+ List<float[]> vectors = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ float[] vector = new float[dimension];
+ for (int j = 0; j < dimension; j++) {
+ vector[j] = random.nextFloat() * 2 - 1; // Range [-1, 1]
+ }
+ // Normalize for inner product similarity
+ float norm = 0;
+ for (float v : vector) {
+ norm += v * v;
+ }
+ norm = (float) Math.sqrt(norm);
+ if (norm > 0) {
+ for (int m = 0; m < vector.length; m++) {
+ vector[m] /= norm;
+ }
+ }
+ vectors.add(vector);
+ }
+ return vectors;
+ }
+
+ private boolean containsRowId(GlobalIndexResult result, long rowId) {
+ List<Long> resultIds = new ArrayList<>();
+ result.results().iterator().forEachRemaining(resultIds::add);
+ return resultIds.contains(rowId);
+ }
+}
diff --git a/paimon-faiss/pom.xml b/paimon-faiss/pom.xml
index 43e22504a6..f4234c927c 100644
--- a/paimon-faiss/pom.xml
+++ b/paimon-faiss/pom.xml
@@ -34,5 +34,7 @@ under the License.
<modules>
<module>paimon-faiss-jni</module>
+ <module>paimon-faiss-index</module>
+ <module>paimon-faiss-e2e-test</module>
</modules>
</project>