Flink Connector Lance

Apache Flink Connector for Lance Vector Database

Features • Quick Start • Documentation • Contributing

📖 Overview

flink-connector-lance is a high-performance Apache Flink connector for Lance, a modern columnar data format optimized for machine learning workloads and vector search. This connector enables seamless integration between Flink's powerful stream/batch processing capabilities and Lance's efficient vector storage.

✨ Features

Feature	Description
🔄 Source & Sink	Full read/write support for Lance datasets
📊 Table API & SQL	Native Flink SQL DDL/DML support
🔍 Vector Search	KNN search with L2, Cosine, Dot metrics
📇 Index Building	IVF_PQ, IVF_HNSW, IVF_FLAT indexes
✅ Exactly-Once	Checkpoint-based exactly-once semantics
🎯 Predicate Pushdown	Filter pushdown for optimized reads
📁 Catalog Support	Lance Catalog for metadata management

🚀 Quick Start

Prerequisites

Java 8 or higher
Apache Flink 1.16.x
Maven 3.6+

Maven Dependency

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-lance</artifactId>
    <version>1.0.0-SNAPSHOT</version>
</dependency>

Flink SQL Example

-- Create a Lance table
CREATE TABLE vectors (
    id BIGINT,
    content STRING,
    embedding ARRAY<FLOAT>
) WITH (
    'connector' = 'lance',
    'path' = '/data/vectors',
    'write.batch-size' = '1024'
);

-- Insert data
INSERT INTO vectors VALUES 
    (1, 'Hello World', ARRAY[0.1, 0.2, 0.3, 0.4]);

-- Query data
SELECT * FROM vectors WHERE id > 0;

DataStream API Example

// Read from Lance
LanceSource source = LanceSource.builder()
    .path("/data/vectors")
    .batchSize(1024)
    .columns(Arrays.asList("id", "embedding"))
    .rowType(rowType)
    .build();

DataStream<RowData> stream = env.addSource(source);

// Write to Lance
LanceSink sink = LanceSink.builder()
    .path("/data/output")
    .batchSize(1024)
    .writeMode(WriteMode.APPEND)
    .rowType(rowType)
    .build();

stream.addSink(sink);

📚 Documentation

Table API / SQL

Create Catalog

CREATE CATALOG lance_catalog WITH (
    'type' = 'lance',
    'warehouse' = '/path/to/warehouse',
    'default-database' = 'default'
);

USE CATALOG lance_catalog;

Table Options

Option	Required	Default	Description
`connector`	✅	-	Must be `lance`
`path`	✅	-	Lance dataset path
`read.batch-size`	❌	1024	Read batch size
`write.batch-size`	❌	1024	Write batch size
`write.mode`	❌	append	`append` or `overwrite`
`write.max-rows-per-file`	❌	1000000	Max rows per file

Vector Index Building

LanceIndexBuilder builder = LanceIndexBuilder.builder()
    .datasetPath("/data/vectors")
    .columnName("embedding")
    .indexType(IndexType.IVF_PQ)
    .metricType(MetricType.L2)
    .numPartitions(256)
    .numSubVectors(16)
    .build();

builder.buildIndex();

Supported Index Types

Index Type	Use Case	Parameters
IVF_PQ	Large-scale, memory-efficient	`num_partitions`, `num_sub_vectors`, `num_bits`
IVF_HNSW	High recall, fast search	`num_partitions`, `m`, `ef_construction`
IVF_FLAT	Small datasets, exact search	`num_partitions`

📊 Index Selection Guide

Scenario	Recommended Index	Reason
< 100K vectors	IVF_FLAT	High accuracy, acceptable performance
100K - 10M vectors	IVF_PQ	Good balance of accuracy and memory
> 10M vectors	IVF_PQ (tuned)	Optimize `num_partitions` and `num_sub_vectors`
High recall required	IVF_HNSW	Best accuracy, higher memory usage
Memory constrained	IVF_PQ	Most memory efficient
Real-time search	IVF_HNSW	Fastest query latency

⚙️ Index Configuration Options

Option	Default	Description
`index.type`	IVF_PQ	Index type: `IVF_PQ`, `IVF_HNSW`, `IVF_FLAT`
`index.column`	-	Vector column name for indexing
`index.num-partitions`	256	Number of IVF partitions
`index.num-sub-vectors`	16	PQ sub-vectors (IVF_PQ only)
`index.num-bits`	8	Bits per sub-vector (IVF_PQ only)
`index.m`	16	HNSW max connections (IVF_HNSW only)
`index.ef-construction`	100	HNSW build quality (IVF_HNSW only)

🔧 Index Building SQL Example

-- Create table with IVF_PQ index
CREATE TABLE doc_embeddings (
    doc_id BIGINT,
    title STRING,
    embedding ARRAY<FLOAT>
) WITH (
    'connector' = 'lance',
    'path' = '/data/embeddings',
    'index.type' = 'IVF_PQ',
    'index.column' = 'embedding',
    'index.num-partitions' = '256',
    'index.num-sub-vectors' = '16',
    'vector.metric' = 'COSINE'
);

-- Create table with IVF_HNSW index (high accuracy)
CREATE TABLE high_accuracy_vectors (
    id BIGINT,
    vector ARRAY<FLOAT>
) WITH (
    'connector' = 'lance',
    'path' = '/data/ha_vectors',
    'index.type' = 'IVF_HNSW',
    'index.column' = 'vector',
    'index.num-partitions' = '128',
    'index.m' = '32',
    'index.ef-construction' = '200',
    'vector.metric' = 'L2'
);

Vector Search

LanceVectorSearch search = LanceVectorSearch.builder()
    .datasetPath("/data/vectors")
    .columnName("embedding")
    .metricType(MetricType.COSINE)
    .nprobes(20)
    .build();

search.open();
List<SearchResult> results = search.search(queryVector, 10);
search.close();

Distance Metrics

Metric	Description	Range
L2	Euclidean distance	[0, ∞)
Cosine	Cosine similarity	[-1, 1]
Dot	Inner product	(-∞, ∞)

Type Mapping

Lance/Arrow Type	Flink Type
Int8/16/32/64	TINYINT/SMALLINT/INT/BIGINT
Float32/64	FLOAT/DOUBLE
String	STRING
Boolean	BOOLEAN
Binary	BYTES
Date32	DATE
Timestamp	TIMESTAMP
FixedSizeList<Float>	ARRAY<FLOAT>

🏗️ Project Structure

flink-connector-lance/
├── src/main/java/org/apache/flink/connector/lance/
│   ├── LanceSource.java           # Source implementation
│   ├── LanceSink.java             # Sink with checkpointing
│   ├── LanceInputFormat.java      # Batch input format
│   ├── LanceIndexBuilder.java     # Vector index builder
│   ├── LanceVectorSearch.java     # KNN search
│   ├── config/
│   │   └── LanceOptions.java      # Configuration options
│   ├── converter/
│   │   ├── LanceTypeConverter.java
│   │   └── RowDataConverter.java
│   └── table/
│       ├── LanceDynamicTableFactory.java
│       ├── LanceDynamicTableSource.java
│       ├── LanceDynamicTableSink.java
│       ├── LanceCatalog.java
│       └── LanceVectorSearchFunction.java
└── src/test/java/                 # Unit & integration tests

🔧 Build

# Build without tests
mvn clean package -DskipTests

# Build with tests
mvn clean package

# Run tests only
mvn test

🤝 Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

Fork the repository
Create your feature branch (git checkout -b feature/AmazingFeature)
Commit your changes (git commit -m 'Add some AmazingFeature')
Push to the branch (git push origin feature/AmazingFeature)
Open a Pull Request

📋 Roadmap

📄 License

This project is licensed under the Apache License 2.0 - see the LICENSE file for details.

🙏 Acknowledgments

Apache Flink - Stateful computations over data streams
LanceDB - Modern columnar data format for ML
Apache Arrow - Cross-language development platform

Made with ❤️ by the Flink Connector Lance Contributors

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Flink Connector Lance

📖 Overview

✨ Features

🚀 Quick Start

Prerequisites

Maven Dependency

Flink SQL Example

DataStream API Example

📚 Documentation

Table API / SQL

Create Catalog

Table Options

Vector Index Building

Supported Index Types

📊 Index Selection Guide

⚙️ Index Configuration Options

🔧 Index Building SQL Example

Vector Search

Distance Metrics

Type Mapping

🏗️ Project Structure

🔧 Build

🤝 Contributing

📋 Roadmap

📄 License

🙏 Acknowledgments

FilesExpand file tree

README.md

Latest commit

History

README.md

File metadata and controls

Flink Connector Lance

📖 Overview

✨ Features

🚀 Quick Start

Prerequisites

Maven Dependency

Flink SQL Example

DataStream API Example

📚 Documentation

Table API / SQL

Create Catalog

Table Options

Vector Index Building

Supported Index Types

📊 Index Selection Guide

⚙️ Index Configuration Options

🔧 Index Building SQL Example

Vector Search

Distance Metrics

Type Mapping

🏗️ Project Structure

🔧 Build

🤝 Contributing

📋 Roadmap

📄 License

🙏 Acknowledgments