diff --git a/docs/src/.pages b/docs/src/.pages new file mode 100644 index 0000000..ba6e6dc --- /dev/null +++ b/docs/src/.pages @@ -0,0 +1,6 @@ +nav: + - Welcome: index.md + - Install: install.md + - Config: config.md + - Performance: performance.md + - Operations: operations diff --git a/docs/src/config.md b/docs/src/config.md new file mode 100644 index 0000000..fc78309 --- /dev/null +++ b/docs/src/config.md @@ -0,0 +1,79 @@ +# Config + +Configuration options are grouped by area. The table connector uses `'connector' = 'lance'`; the +catalog types are `'lance'` (directory/S3) and `'lance-namespace'` (dir/rest). + +## Table connector options (`connector = 'lance'`) + +### Common + +| Option | Required | Default | Description | +|---|---|---|---| +| `path` | ✅ | — | Path to the Lance dataset | +| `hadoop.*` | ❌ | — | Prefix for Hadoop-family filesystem config (e.g. `hadoop.tbdsfs.meta`); stripped and injected into the Hadoop `Configuration` used for path resolution | + +### Read (Source) + +| Option | Required | Default | Description | +|---|---|---|---| +| `read.batch-size` | ❌ | 1024 | Read batch size | +| `read.limit` | ❌ | — | Maximum rows to read (limit pushdown) | +| `read.columns` | ❌ | — | Columns to read, comma separated | +| `read.filter` | ❌ | — | SQL `WHERE`-style filter predicate | +| `read.version` | ❌ | — | Time travel: read a specific dataset version | +| `read.as-of-timestamp` | ❌ | — | Time travel: read as of an ISO-8601 timestamp (ignored when `read.version` is set) | + +### Write (Sink) + +| Option | Required | Default | Description | +|---|---|---|---| +| `write.batch-size` | ❌ | 1024 | Write batch size | +| `write.mode` | ❌ | append | `append` or `overwrite` | +| `write.max-rows-per-file` | ❌ | 1000000 | Maximum rows per data file | + +### Vector index + +| Option | Required | Default | Description | +|---|---|---|---| +| `index.type` | ❌ | IVF_PQ | `IVF_PQ`, `IVF_HNSW`, or `IVF_FLAT` | +| `index.column` | ❌ | — | Vector column name to index | +| `index.num-partitions` | ❌ | 256 | IVF partition count | +| `index.num-sub-vectors` | ❌ | — | PQ sub-vector count (auto if unset) | +| `index.num-bits` | ❌ | 8 | PQ quantization bits (1–16) | +| `index.max-level` | ❌ | 7 | HNSW max level | +| `index.m` | ❌ | 16 | HNSW connections per level | +| `index.ef-construction` | ❌ | 100 | HNSW construction search width | + +### Vector search + +| Option | Required | Default | Description | +|---|---|---|---| +| `vector.column` | ❌ | — | Vector search column name | +| `vector.metric` | ❌ | L2 | `L2`, `Cosine`, or `Dot` | +| `vector.nprobes` | ❌ | 20 | IVF search probe count | +| `vector.ef` | ❌ | 100 | HNSW search width | +| `vector.refine-factor` | ❌ | — | Refine factor for recall | + +## Catalog options (`type = 'lance'`) + +Directory or S3 warehouse. + +| Option | Required | Default | Description | +|---|---|---|---| +| `warehouse` | ✅ | — | Warehouse path (local or `s3://…`) | +| `default-database` | ❌ | default | Default database name | +| `s3-access-key` | ❌ | — | S3 access key ID | +| `s3-secret-key` | ❌ | — | S3 secret access key | +| `s3-region` | ❌ | — | S3 region (e.g. `us-east-1`) | +| `s3-endpoint` | ❌ | — | S3 endpoint (for S3-compatible storage like MinIO) | +| `s3-virtual-hosted-style` | ❌ | true | Virtual-hosted-style URLs | +| `s3-allow-http` | ❌ | false | Allow HTTP (default HTTPS only) | + +## Namespace catalog options (`type = 'lance-namespace'`) + +| Option | Required | Default | Description | +|---|---|---|---| +| `impl` | ✅ | — | Namespace implementation: `dir` or `rest` | +| `root` | ❌ | — | Root path for directory namespace | +| `uri` | ❌ | — | URI for REST namespace | +| `default-database` | ❌ | default | Default database name | diff --git a/docs/src/index.md b/docs/src/index.md new file mode 100644 index 0000000..a52f225 --- /dev/null +++ b/docs/src/index.md @@ -0,0 +1,60 @@ +# Flink Lance Connector + +## Introduction + +The Apache Flink Connector for Lance allows Apache Flink to read and write datasets stored in the +[Lance](https://lance.org/) columnar format — an open lakehouse format optimized for multimodal AI +and vector search workloads. + +By using the Flink Connector for Lance, you can run Flink's stream/batch processing, SQL querying, +and stateful pipelines directly on Lance datasets, including native vector search. + +## Features + +The connector is built on the Flink Table API (`DynamicTableSource` / `DynamicTableSink`) plus +`CatalogFactory`. Specifically, you can use the Flink Connector for Lance to: + +* **Read & Write Lance Datasets**: append and overwrite datasets via Flink SQL or the DataStream API. +* **Column, Filter, Limit & Aggregate Pushdown**: push projections, `WHERE` predicates, limits and + aggregations down to Lance for efficient scans. +* **Vector Search**: KNN search over `ARRAY` columns with `L2`, `Cosine`, and `Dot` metrics, + via the `LanceVectorSearchFunction` table function. +* **Vector Index Building**: create `IVF_PQ`, `IVF_HNSW`, and `IVF_FLAT` indexes. +* **Time Travel**: read a historical version via `read.version` or `read.as-of-timestamp`. +* **Catalog Support**: a directory/S3 `lance` catalog and a `lance-namespace` catalog (dir / rest). + +## Quick Start + +Create a catalog and a table, then insert and query: + +```sql +-- Create a directory-based catalog +CREATE CATALOG lance_catalog WITH ( + 'type' = 'lance', + 'warehouse' = '/path/to/warehouse', + 'default-database' = 'default' +); + +USE CATALOG lance_catalog; + +-- Create a Lance table +CREATE TABLE vectors ( + id BIGINT, + content STRING, + embedding ARRAY +) 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; +``` + +See [Install](install.md) for dependency setup and [Operations](operations/) for the full SQL +surface. diff --git a/docs/src/install.md b/docs/src/install.md new file mode 100644 index 0000000..c8b17a6 --- /dev/null +++ b/docs/src/install.md @@ -0,0 +1,49 @@ +# Install + +## Requirements + +* JDK 11 or higher +* Maven 3.6+ +* Apache Flink 1.18 / 1.19 / 1.20 + +The connector ships one artifact per supported Flink minor version: + +| Flink version | Artifact | +|---|---| +| 1.18 | `lance-flink-1.18` | +| 1.19 | `lance-flink-1.19` | +| 1.20 | `lance-flink-1.20` | + +## Dependencies + +The connector depends on `org.lance:lance-core` (7.0.0) and Apache Arrow (18.3.0). These are pulled +in transitively; you only need to add the connector artifact for your Flink version. + +Lance's Java bindings ship a platform-specific JNI native library +(`liblance_jni.so` / `liblance_jni.dylib`) inside the `lance-core` jar. Ensure you run on a +supported platform (linux-x86-64, linux-aarch64, darwin-aarch64). + +## Maven + +```xml + + org.apache.flink + lance-flink-1.18 + 0.1.0 + +``` + +## Build from source + +```bash +mvn clean verify +``` + +The build produces a fat jar per module (e.g. `lance-flink-1.18/target/lance-flink-1.18-*.jar`). +Add the jar to your Flink cluster or job classpath, then use the `lance` / `lance-namespace` +catalog types in SQL. + +## Note on Arrow and Netty + +The Arrow allocator defaults are set at runtime; if you see classloader-related SPI issues in +tests, set the system property `arrow.memory.allocator.type=Netty`. diff --git a/docs/src/operations/.pages b/docs/src/operations/.pages new file mode 100644 index 0000000..582cf82 --- /dev/null +++ b/docs/src/operations/.pages @@ -0,0 +1,4 @@ +nav: + - DDL: ddl + - DQL: dql + - DML: dml diff --git a/docs/src/operations/ddl/.pages b/docs/src/operations/ddl/.pages new file mode 100644 index 0000000..b95d1c3 --- /dev/null +++ b/docs/src/operations/ddl/.pages @@ -0,0 +1,3 @@ +nav: + - CREATE CATALOG: create-catalog.md + - CREATE TABLE: create-table.md diff --git a/docs/src/operations/ddl/create-catalog.md b/docs/src/operations/ddl/create-catalog.md new file mode 100644 index 0000000..ee2e5a0 --- /dev/null +++ b/docs/src/operations/ddl/create-catalog.md @@ -0,0 +1,80 @@ +# CREATE CATALOG + +The Lance Flink connector ships two catalog types, registered via SPI: + +| `type` | Class | Description | +|---|---|---| +| `lance` | `LanceCatalogFactory` | Directory-based catalog over a warehouse path (local or S3) | +| `lance-namespace` | `LanceNamespaceCatalogFactory` | Catalog backed by a Lance namespace (dir or REST) | + +## Directory catalog (`type = 'lance'`) + +```sql +CREATE CATALOG lance_catalog WITH ( + 'type' = 'lance', + 'warehouse' = '/path/to/warehouse', + 'default-database' = 'default' +); + +USE CATALOG lance_catalog; +``` + +### S3 warehouse + +```sql +CREATE CATALOG lance_s3_catalog WITH ( + 'type' = 'lance', + 'warehouse' = 's3://bucket-name/warehouse', + 'default-database' = 'default', + 's3-access-key' = 'your-access-key', + 's3-secret-key' = 'your-secret-key', + 's3-region' = 'us-east-1', + 's3-endpoint' = 'https://s3.amazonaws.com' +); +``` + +| Option | Required | Default | Description | +|---|---|---|---| +| `warehouse` | ✅ | — | Warehouse path (local or `s3://`) | +| `default-database` | ❌ | `default` | Default database | +| `s3-access-key` | ❌ | — | S3 access key | +| `s3-secret-key` | ❌ | — | S3 secret key | +| `s3-region` | ❌ | — | S3 region | +| `s3-endpoint` | ❌ | — | S3 endpoint (for MinIO etc.) | +| `s3-virtual-hosted-style` | ❌ | `true` | Virtual-hosted-style URLs | +| `s3-allow-http` | ❌ | `false` | Allow HTTP | + +## Namespace catalog (`type = 'lance-namespace'`) + +```sql +-- Directory-based namespace +CREATE CATALOG my_lance WITH ( + 'type' = 'lance-namespace', + 'impl' = 'dir', + 'root' = '/tmp/lance-warehouse' +); + +-- REST-based namespace +CREATE CATALOG my_lance WITH ( + 'type' = 'lance-namespace', + 'impl' = 'rest', + 'uri' = 'http://localhost:8080' +); +``` + +| Option | Required | Default | Description | +|---|---|---|---| +| `impl` | ✅ | — | `dir` or `rest` | +| `root` | ❌ | — | Root path for `dir` impl | +| `uri` | ❌ | — | URI for `rest` impl | +| `default-database` | ❌ | `default` | Default database | + +## Supported DDL + +| Statement | Status | +|---|---| +| `CREATE DATABASE` / `DROP DATABASE` / `ALTER DATABASE` | ✅ | +| `SHOW DATABASES` / `SHOW TABLES` | ✅ | +| `CREATE TABLE` / `DROP TABLE` / `RENAME TABLE` | ✅ | +| `ALTER TABLE` | ❌ — not supported (structure immutable) | +| `CREATE INDEX` | ❌ — no SQL DDL; configure `index.*` on the table | diff --git a/docs/src/operations/ddl/create-table.md b/docs/src/operations/ddl/create-table.md new file mode 100644 index 0000000..54702d5 --- /dev/null +++ b/docs/src/operations/ddl/create-table.md @@ -0,0 +1,56 @@ +# CREATE TABLE + +Lance tables are created through the dynamic table factory (`connector = 'lance'`). +The actual Lance dataset is created lazily on first write. + +## Minimal example + +```sql +CREATE TABLE vectors ( + id BIGINT, + content STRING, + embedding ARRAY +) WITH ( + 'connector' = 'lance', + 'path' = '/data/vectors' +); +``` + +## With a vector index + +```sql +CREATE TABLE doc_embeddings ( + doc_id BIGINT, + title STRING, + embedding ARRAY +) 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' +); +``` + +## Required options + +| Option | Description | +|---|---| +| `path` | Path to the Lance dataset | + +## Type mapping + +| Lance / Arrow type | Flink type | +|---|---| +| Int8 / Int16 / Int32 / Int64 | TINYINT / SMALLINT / INT / BIGINT | +| Float32 / Float64 | FLOAT / DOUBLE | +| String | STRING | +| Boolean | BOOLEAN | +| Binary | BYTES | +| Date32 | DATE | +| Timestamp | TIMESTAMP | +| FixedSizeList\ | ARRAY\ | + +See [Config](../../config.md) for the full option reference. diff --git a/docs/src/operations/dml/.pages b/docs/src/operations/dml/.pages new file mode 100644 index 0000000..e9a5257 --- /dev/null +++ b/docs/src/operations/dml/.pages @@ -0,0 +1,2 @@ +nav: + - INSERT INTO: insert-into.md diff --git a/docs/src/operations/dml/insert-into.md b/docs/src/operations/dml/insert-into.md new file mode 100644 index 0000000..861a656 --- /dev/null +++ b/docs/src/operations/dml/insert-into.md @@ -0,0 +1,39 @@ +# INSERT INTO + +The Lance Flink sink appends rows to a Lance dataset. Write mode is controlled by +the `write.mode` option. + +## Write modes + +| `write.mode` | Behaviour | +|---|---| +| `append` (default) | Append rows to the existing dataset | +| `overwrite` | Replace the dataset on first write | + +## Example + +```sql +INSERT INTO vectors VALUES + (1, 'Hello World', ARRAY[0.1, 0.2, 0.3, 0.4]); +``` + +## Sink options + +| Option | Default | Description | +|---|---|---| +| `write.batch-size` | 1024 | Rows buffered before a flush | +| `write.mode` | `append` | `append` or `overwrite` | +| `write.max-rows-per-file` | 1000000 | Rows per data file | + +## Current limitations + +| Statement | Status | +|---|---| +| `INSERT INTO` (append) | ✅ | +| `INSERT OVERWRITE` | ✅ | +| `UPDATE` | ❌ — not implemented | +| `DELETE` | ❌ — in progress (see issue #63 / #74) | +| Primary key / upsert | ❌ — PK declaration and CDC changelog not yet supported | + +> The sink currently declares insert-only changelog mode. CDC `UPDATE` / `DELETE` +> support is tracked in the connector roadmap. diff --git a/docs/src/operations/dql/.pages b/docs/src/operations/dql/.pages new file mode 100644 index 0000000..35f1b23 --- /dev/null +++ b/docs/src/operations/dql/.pages @@ -0,0 +1,4 @@ +nav: + - SELECT: select.md + - Vector Search: vector-search.md + - Time Travel: time-travel.md diff --git a/docs/src/operations/dql/select.md b/docs/src/operations/dql/select.md new file mode 100644 index 0000000..efbc7ac --- /dev/null +++ b/docs/src/operations/dql/select.md @@ -0,0 +1,48 @@ +# SELECT + +The Lance Flink connector supports reading Lance datasets via Flink SQL `SELECT`. +Read optimizations are pushed down to Lance natively to reduce I/O. + +## Supported pushdowns + +| Ability | Interface | Notes | +|---|---|---| +| Column projection | `SupportsProjectionPushDown` | Only projected columns are read | +| Predicate (filter) | `SupportsFilterPushDown` | `WHERE` clauses are pushed to Lance | +| Limit | `SupportsLimitPushDown` | `LIMIT` is pushed down | +| Aggregate | `SupportsAggregatePushDown` | Eligible aggregates run natively | + +## Example + +```sql +-- Projection + filter + limit are all pushed down +SELECT id, content +FROM lance_table +WHERE id > 100 +LIMIT 10; +``` + +## Static read options + +The same read behaviour can be configured declaratively on the table DDL +without relying on planner pushdown: + +```sql +CREATE TABLE lance_table ( + id BIGINT, + content STRING, + embedding ARRAY +) WITH ( + 'connector' = 'lance', + 'path' = '/data/vectors', + 'read.columns' = 'id,content', + 'read.filter' = 'id > 100', + 'read.limit' = '10' +); +``` + +| Option | Description | +|---|---| +| `read.columns` | Comma-separated columns to read | +| `read.filter` | SQL `WHERE`-style filter string | +| `read.limit` | Maximum rows to read | diff --git a/docs/src/operations/dql/time-travel.md b/docs/src/operations/dql/time-travel.md new file mode 100644 index 0000000..5e9efea --- /dev/null +++ b/docs/src/operations/dql/time-travel.md @@ -0,0 +1,44 @@ +# Time Travel + +The Lance Flink connector supports reading historical versions of a dataset via +time-travel options, resolved uniformly through `LanceOpener`. + +## Options + +| Option | Type | Description | +|---|---|---| +| `read.version` | LONG | Read the given dataset version (highest precedence) | +| `read.as-of-timestamp` | STRING | Read as of an ISO-8601 timestamp; resolves to the newest version whose creation time is ≤ the timestamp | + +When both are set, `read.version` takes precedence. + +## Examples + +```sql +-- Read a specific version +SELECT * FROM lance_table /*+ OPTIONS('read.version' = '3') */; + +-- Read as of a timestamp (resolves to the newest version <= the timestamp) +SELECT * FROM lance_table /*+ OPTIONS('read.as-of-timestamp' = '2026-07-01T00:00:00Z') */; +``` + +Declaratively on the table DDL: + +```sql +CREATE TABLE lance_table ( + id BIGINT, + content STRING +) WITH ( + 'connector' = 'lance', + 'path' = '/data/vectors', + 'read.version' = '3' +); +``` + +## Semantics + +- `read.version` opens exactly that version. +- `read.as-of-timestamp` accepts any string parseable by + `Instant.parse`, `OffsetDateTime.parse`, or `ZonedDateTime.parse`; + a bare `yyyy-MM-ddTHH:mm:ss` is assumed UTC. +- A timestamp predating the oldest version raises a clear error. diff --git a/docs/src/operations/dql/vector-search.md b/docs/src/operations/dql/vector-search.md new file mode 100644 index 0000000..a0155b0 --- /dev/null +++ b/docs/src/operations/dql/vector-search.md @@ -0,0 +1,76 @@ +# Vector Search + +The Lance Flink connector exposes a table function for KNN vector search over a +Lance dataset, powered by Lance's IVF / HNSW indexes. + +## Function signature + +The function is `LanceVectorSearchFunction` — a Flink `TableFunction` registered as a +temporary function before use: + +```sql +CREATE TEMPORARY FUNCTION vector_search AS + 'org.apache.flink.connector.lance.table.LanceVectorSearchFunction' + LANGUAGE JAVA USING JAR '/path/to/lance-flink-1.18-0.1.0.jar'; +``` + +The registered function name (here `vector_search`) is user-chosen. Its `eval` overloads are: + +``` +vector_search(dataset_path, column_name, query_vector [, k [, metric]]) +``` + +- `dataset_path` — path to the Lance dataset +- `column_name` — the vector column to search +- `query_vector` — the query vector (`ARRAY`; also accepts `DECIMAL[]` / `DOUBLE[]` / `float[]`) +- `k` — number of nearest neighbours (default `10`) +- `metric` — distance metric (`L2` / `Cosine` / `Dot`, default `L2`) + +The emitted row is the source row plus a `_distance DOUBLE` column. + +## Distance metrics + +| Metric | Description | Range | +|---|---|---| +| `L2` | Euclidean distance | [0, ∞) | +| `Cosine` | Cosine similarity | [-1, 1] | +| `Dot` | Inner product | (-∞, ∞) | + +## Search options + +Configured on the table DDL or via `LanceOptions`: + +| Option | Default | Description | +|---|---|---| +| `vector.column` | — | Vector column name | +| `vector.metric` | `L2` | Distance metric | +| `vector.nprobes` | 20 | IVF probe count | +| `vector.ef` | 100 | HNSW search width | +| `vector.refine-factor` | — | Re-rank factor for recall | + +## Example + +```sql +CREATE TABLE vectors ( + id BIGINT, + content STRING, + embedding ARRAY +) WITH ( + 'connector' = 'lance', + 'path' = '/data/vectors', + 'index.type' = 'IVF_PQ', + 'index.column' = 'embedding', + 'vector.metric' = 'COSINE' +); + +-- KNN search for the 10 nearest vectors +SELECT * +FROM vectors, + LATERAL TABLE(vector_search( + '/data/vectors', + 'embedding', + ARRAY[0.1, 0.2, 0.3, 0.4], + 10, + 'COSINE' + )); +``` diff --git a/docs/src/performance.md b/docs/src/performance.md new file mode 100644 index 0000000..6bc839f --- /dev/null +++ b/docs/src/performance.md @@ -0,0 +1,49 @@ +# Performance + +This page covers vector index selection and the tuning knobs that most affect +read/search throughput and write amplification in the Lance Flink connector. + +## Vector index types + +| Index type | Best for | Memory | Recall | Key parameters | +|---|---|---|---|---| +| `IVF_FLAT` | Small datasets (< 100K vectors), exact-ish search | High | Highest | `index.num-partitions` | +| `IVF_PQ` | Large datasets, memory-constrained | Lowest | Good | `index.num-partitions`, `index.num-sub-vectors`, `index.num-bits` | +| `IVF_HNSW` | High recall, fast query latency | High | High | `index.num-partitions`, `index.m`, `index.ef-construction` | + +## Index selection guide + +| Scenario | Recommended index | Reason | +|---|---|---| +| < 100K vectors | `IVF_FLAT` | Highest accuracy, acceptable latency | +| 100K – 10M vectors | `IVF_PQ` | Good accuracy/memory trade-off | +| > 10M vectors | `IVF_PQ` (tuned) | Tune `index.num-partitions` and `index.num-sub-vectors` | +| High recall required | `IVF_HNSW` | Best accuracy, higher memory | +| Memory constrained | `IVF_PQ` | Most memory efficient | +| Real-time search | `IVF_HNSW` | Fastest query latency | + +## Search tuning + +| Option | Default | Effect | +|---|---|---| +| `vector.nprobes` | 20 | Number of IVF partitions probed per query. Higher = better recall, slower query | +| `vector.ef` | 100 | HNSW search width. Higher = better recall, slower query | +| `vector.refine-factor` | — | Refines top-k results by re-ranking candidates. Higher = better recall | + +## Write amplification + +| Option | Default | Effect | +|---|---|---| +| `write.batch-size` | 1024 | Rows buffered before a flush; larger = fewer, bigger writes | +| `write.max-rows-per-file` | 1000000 | Rows per data file; larger = fewer files, larger compaction units | + +## Distance metrics + +| Metric | Description | Range | +|---|---|---| +| `L2` | Euclidean distance | [0, ∞) | +| `Cosine` | Cosine similarity | [-1, 1] | +| `Dot` | Inner product | (-∞, ∞) | + +> **Note:** No benchmark numbers are published yet. These knobs are the exposed +> surface; measure against your own workload to tune them.