milvus-io--milvus/docs/user_guides/external_table.md

Milvus External Table User Guide

1. Overview

1.1 What is External Table

External Table (External Collection) is a special type of data collection in Milvus that allows users to directly access data stored in external storage systems (such as S3, HDFS, etc.) without copying the data into Milvus local storage.

This enables Milvus to serve as a query layer over existing data lakes while maintaining compatibility with standard Milvus query interfaces.

1.2 Core Benefits

• Zero Data Copy: Query data directly from external storage without ETL process
• Unified Query Interface: Use standard Search/Query APIs to query external data
• Vector Index Support: Build vector indexes on external data for efficient similarity search
• Data Lake Integration: Seamlessly integrate with existing data lake infrastructure

1.3 Use Cases

• Large amounts of vector data already stored in S3 or other object storage
• Need to perform vector search on data lake data
• Want to maintain separation between data storage and query engine
• Need to query periodically updated external data

---

2. Quick Start

2.1 Create External Collection

External Collection is created through the standard CreateCollection API by setting external_source in the schema.

Python SDK Example

from pymilvus import MilvusClient, DataType

client = MilvusClient("http://localhost:19530")

# Define Schema
schema = client.create_schema()

# Add fields - must specify external_field to map to column names in external data source
schema.add_field(
    field_name="text",
    datatype=DataType.VARCHAR,
    max_length=256,
    external_field="source_text_column"  # Maps to column name in external Parquet file
)

schema.add_field(
    field_name="vector",
    datatype=DataType.FLOAT_VECTOR,
    dim=128,
    external_field="embedding_column"    # Maps to column name in external Parquet file
)

# Set external data source
schema.external_source = "s3://my-bucket/path/to/data"
schema.external_spec = '{"format": "parquet"}'

# Create Collection
client.create_collection(
    collection_name="my_external_collection",
    schema=schema
)

Go SDK Example

package main

import (
    "context"
    "log"

    "github.com/milvus-io/milvus/client/v2"
    "github.com/milvus-io/milvus/client/v2/entity"
)

func main() {
    ctx := context.Background()

    // Connect to Milvus
    cli, err := client.New(ctx, &client.ClientConfig{
        Address: "localhost:19530",
    })
    if err != nil {
        log.Fatal(err)
    }
    defer cli.Close(ctx)

    // Define Schema with external source
    schema := entity.NewSchema().
        WithName("my_external_collection").
        WithExternalSource("s3://my-bucket/path/to/data").
        WithExternalSpec(`{"format": "parquet"}`).
        WithField(entity.NewField().
            WithName("text").
            WithDataType(entity.FieldTypeVarChar).
            WithMaxLength(256).
            WithExternalField("source_text_column")).  // Maps to external column
        WithField(entity.NewField().
            WithName("vector").
            WithDataType(entity.FieldTypeFloatVector).
            WithDim(128).
            WithExternalField("embedding_column"))     // Maps to external column

    // Create Collection
    err = cli.CreateCollection(ctx, client.NewCreateCollectionOption(
        "my_external_collection",
        schema,
    ))
    if err != nil {
        log.Fatal(err)
    }
}

2.2 Field Mapping Rules

Schema Parameter	Description	Example
external_source	External data source path	s3://bucket/path
external_spec	Data source configuration (JSON format)	{"format": "parquet"}
external_field	Maps field to external column name	Must be specified for each field

Note: All user-defined fields must set external_field to map to column names in the external data source.

---

3. Supported Operations

3.1 Load Collection

# Load External Collection into memory
client.load_collection("my_external_collection")

3.2 Vector Search

# Execute vector search
results = client.search(
    collection_name="my_external_collection",
    data=[[0.1, 0.2, ...]],  # Query vector
    anns_field="vector",
    limit=10,
    output_fields=["text"]
)

3.3 Scalar Query

# Execute scalar query
results = client.query(
    collection_name="my_external_collection",
    filter="text like 'hello%'",
    output_fields=["text", "vector"],
    limit=10
)

3.4 Create Index

# Create index on vector field
index_params = client.prepare_index_params()
index_params.add_index(
    field_name="vector",
    index_type="HNSW",
    metric_type="L2",
    params={"M": 16, "efConstruction": 200}
)

client.create_index(
    collection_name="my_external_collection",
    index_params=index_params
)

3.5 Drop Collection

# Drop External Collection
client.drop_collection("my_external_collection")

---

4. Unsupported Operations

External Collection is read-only. The following operations are not supported:

Operation	Status	Description
Insert	Not Supported	Data must be modified at external source
Delete	Not Supported	Data must be modified at external source
Upsert	Not Supported	Data must be modified at external source
Import	Not Supported	Data comes directly from external source
Flush	Not Supported	No local data cache
Add Field	Not Supported	Schema is fixed after creation
Alter Field	Not Supported	Schema is fixed after creation
Create/Drop Partition	Not Supported	Partitions not supported
Manual Compaction	Not Supported	Not needed

4.1 Schema Restrictions

When creating an External Collection, the following features cannot be used:

Feature	Status	Reason
Primary Key Field	Not Allowed	System auto-generates virtual PK
Dynamic Field	Not Allowed	Schema must be fixed
Partition Key	Not Allowed	External data partitioning not supported
Clustering Key	Not Allowed	No clustering compaction
Auto ID	Not Allowed	Uses virtual PK
Text Match	Not Allowed	Requires internal indexing
Namespace Field	Not Allowed	External isolation not supported

---

5. Data Updates

External table data refresh is manually triggered using the RefreshExternalTable API. This design gives you full control over when data synchronization occurs and allows you to track progress.

5.1 Refresh APIs

5.1.1 RefreshExternalTable

Triggers a data refresh job for an external collection.

# Basic refresh - re-scan current data source
response = client.refresh_external_table(
    collection_name="my_external_collection"
)
job_id = response.job_id
print(f"Refresh job started: {job_id}")

# Refresh with updated data source path
response = client.refresh_external_table(
    collection_name="my_external_collection",
    external_source="s3://my-bucket/path/to/new_data",
    external_spec='{"format": "parquet"}'
)

Parameters:

Parameter	Type	Required	Description
collection_name	str	Yes	Name of the external collection
external_source	str	No	New external source path (optional)
external_spec	str	No	New external spec configuration (optional)

Returns: job_id for tracking progress

5.1.2 GetRefreshExternalTableProgress

Gets the current progress and status of a refresh job.

# Get progress of a specific job
progress = client.get_refresh_external_table_progress(job_id="job_123456")

print(f"State: {progress.state}")           # Pending/InProgress/Completed/Failed
print(f"Progress: {progress.progress}%")
print(f"New segments: {progress.new_segments}")
print(f"Dropped segments: {progress.dropped_segments}")
print(f"Kept segments: {progress.kept_segments}")

if progress.state == "Failed":
    print(f"Error: {progress.reason}")

Progress States:

State	Description
Pending	Job is queued, waiting to execute
InProgress	Job is currently executing
Completed	Job completed successfully
Failed	Job failed with error

5.1.3 ListRefreshExternalTableJobs

Lists all refresh jobs for a collection.

# List all jobs for a specific collection
jobs = client.list_refresh_external_table_jobs(
    collection_name="my_external_collection",
    limit=10
)

for job in jobs:
    print(f"Job: {job.job_id}")
    print(f"  State: {job.state}")
    print(f"  Progress: {job.progress}%")
    print(f"  Started: {job.start_time}")
    print(f"  Source: {job.external_source}")

# List all external table refresh jobs across all collections
all_jobs = client.list_refresh_external_table_jobs()

5.2 Complete Refresh Workflow

from pymilvus import MilvusClient
import time

client = MilvusClient("http://localhost:19530")

# Step 1: Trigger refresh
response = client.refresh_external_table(
    collection_name="my_external_collection"
)
job_id = response.job_id
print(f"Refresh job started: {job_id}")

# Step 2: Poll for completion
while True:
    progress = client.get_refresh_external_table_progress(job_id=job_id)

    print(f"Progress: {progress.progress}% ({progress.state})")

    if progress.state == "Completed":
        print("Refresh completed successfully!")
        print(f"  New segments: {progress.new_segments}")
        print(f"  Dropped segments: {progress.dropped_segments}")
        print(f"  Kept segments: {progress.kept_segments}")
        break
    elif progress.state == "Failed":
        print(f"Refresh failed: {progress.reason}")
        break

    time.sleep(5)  # Poll every 5 seconds

# Step 3: Re-load collection to query refreshed data
client.load_collection("my_external_collection")

5.3 Incremental Update Strategy

The system uses segment-level incremental update strategy:

1. Keep: Segments whose external fragments are unchanged remain intact
2. Drop: Segments whose corresponding external fragments are deleted/modified are removed
3. Add: New external fragments are organized into new segments

This strategy minimizes data reloading during updates.

Note: Current version does not support automatic detection of external data source changes. Users must manually trigger refresh using refresh_external_table.

---

6. Supported Data Formats

Format	Status	Description
Parquet	Supported	Apache Parquet format

---

7. Storage Configuration

7.1 S3 Configuration Example

schema.external_source = "s3://my-bucket/vector-data/"
schema.external_spec = '''
{
    "format": "parquet"
}
'''

External Collection reuses storage configuration from Milvus configuration file (minio.* or s3.* configuration items).

---

8. Important Notes

1. Immutable Schema: Schema cannot be modified after creation. Plan carefully before creation.
2. Read-Only Mode: All data modifications must be done at the external data source.
3. Manual Refresh: External data changes require manual trigger using refresh_external_table API. Use get_refresh_external_table_progress to track progress.
4. Field Mapping: Each field must correctly map to column names in the external data source.
5. Data Type Matching: Ensure Milvus field types are compatible with external data column types.
6. Re-load After Refresh: After refresh job completes, call load_collection to make the updated data available for queries.

---

9. Complete Example

9.1 Python SDK Complete Example

from pymilvus import MilvusClient, DataType

# Connect to Milvus
client = MilvusClient("http://localhost:19530")

# Create Schema
schema = client.create_schema()

# Add text field
schema.add_field(
    field_name="title",
    datatype=DataType.VARCHAR,
    max_length=512,
    external_field="doc_title"
)

# Add vector field
schema.add_field(
    field_name="embedding",
    datatype=DataType.FLOAT_VECTOR,
    dim=768,
    external_field="text_embedding"
)

# Configure external data source
schema.external_source = "s3://my-data-lake/documents/"
schema.external_spec = '{"format": "parquet"}'

# Create External Collection
client.create_collection(
    collection_name="document_search",
    schema=schema
)

# Create vector index
index_params = client.prepare_index_params()
index_params.add_index(
    field_name="embedding",
    index_type="HNSW",
    metric_type="COSINE",
    params={"M": 32, "efConstruction": 256}
)
client.create_index("document_search", index_params)

# ============================================
# Refresh data when external source changes
# ============================================
import time

# Step 1: Trigger refresh job
response = client.refresh_external_table(
    collection_name="document_search",
    external_source="s3://my-data-lake/documents/v1",
    external_spec='{"format": "parquet"}',
)
job_id = response.job_id
print(f"Refresh job started: {job_id}")

# Step 2: Poll for completion
while True:
    progress = client.get_refresh_external_table_progress(job_id=job_id)
    print(f"Progress: {progress.progress}% ({progress.state})")

    if progress.state == "Completed":
        print("Refresh completed!")
        break
    elif progress.state == "Failed":
        print(f"Refresh failed: {progress.reason}")
        break

    time.sleep(5)

# Step 3: Re-load collection to query refreshed data
client.load_collection("document_search")

# Now search will use the refreshed data
results = client.search(
    collection_name="document_search",
    data=[query_embedding],
    anns_field="embedding",
    limit=10,
    output_fields=["title"]
)

# ============================================
# List all refresh jobs for this collection
# ============================================
jobs = client.list_refresh_external_table_jobs(
    collection_name="document_search"
)
for job in jobs:
    print(f"Job {job.job_id}: {job.state} ({job.progress}%)")

9.2 Go SDK Complete Example

package main

import (
    "context"
    "fmt"
    "log"

    "github.com/milvus-io/milvus/client/v2"
    "github.com/milvus-io/milvus/client/v2/entity"
    "github.com/milvus-io/milvus/client/v2/index"
)

func main() {
    ctx := context.Background()

    // Connect to Milvus
    cli, err := client.New(ctx, &client.ClientConfig{
        Address: "localhost:19530",
    })
    if err != nil {
        log.Fatal(err)
    }
    defer cli.Close(ctx)

    collectionName := "document_search"

    // ============================================
    // Create External Collection
    // ============================================

    // Define Schema with external source
    schema := entity.NewSchema().
        WithName(collectionName).
        WithExternalSource("s3://my-data-lake/documents/").
        WithExternalSpec(`{"format": "parquet"}`).
        WithField(entity.NewField().
            WithName("title").
            WithDataType(entity.FieldTypeVarChar).
            WithMaxLength(512).
            WithExternalField("doc_title")).
        WithField(entity.NewField().
            WithName("embedding").
            WithDataType(entity.FieldTypeFloatVector).
            WithDim(768).
            WithExternalField("text_embedding"))

    // Create Collection
    err = cli.CreateCollection(ctx, client.NewCreateCollectionOption(collectionName, schema))
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println("External collection created successfully")

    // ============================================
    // Create Vector Index
    // ============================================

    indexTask, err := cli.CreateIndex(ctx, client.NewCreateIndexOption(
        collectionName,
        "embedding",
        index.NewHNSWIndex(entity.COSINE, 32, 256),
    ))
    if err != nil {
        log.Fatal(err)
    }
    err = indexTask.Await(ctx)
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println("Index created successfully")

    // ============================================
    // Load Collection
    // ============================================

    loadTask, err := cli.LoadCollection(ctx, client.NewLoadCollectionOption(collectionName))
    if err != nil {
        log.Fatal(err)
    }
    err = loadTask.Await(ctx)
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println("Collection loaded successfully")

    // ============================================
    // Search
    // ============================================

    // Query embedding (replace with actual query vector)
    queryEmbedding := make([]float32, 768)
    for i := range queryEmbedding {
        queryEmbedding[i] = 0.1
    }

    results, err := cli.Search(ctx, client.NewSearchOption(
        collectionName,
        10, // limit
        []entity.Vector{entity.FloatVector(queryEmbedding)},
    ).WithANNSField("embedding").WithOutputFields("title"))
    if err != nil {
        log.Fatal(err)
    }

    for _, result := range results {
        for i := 0; i < result.ResultCount; i++ {
            title, _ := result.Fields.GetColumn("title").Get(i)
            fmt.Printf("Result %d: title=%v, score=%f\n", i, title, result.Scores[i])
        }
    }

    // ============================================
    // Drop Collection (cleanup)
    // ============================================

    err = cli.DropCollection(ctx, client.NewDropCollectionOption(collectionName))
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println("Collection dropped successfully")
}

---

10. Future Plans (Roadmap)

The following features are planned for future releases:

10.1 Scalar Index Support

Support creating scalar indexes on external collections to accelerate filtering queries:

# Future: Create scalar index on external collection
index_params.add_index(
    field_name="category",
    index_type="INVERTED"
)

10.2 Function Support

Support embedding functions and other built-in transformation functions for external collections:

# Future: Use embedding function with external collection
schema.add_function(
    name="text_to_vector",
    function_type=FunctionType.EMBEDDING,
    input_field="text",
    output_field="vector",
    params={"model": "text-embedding-3-small"}
)

10.3 Schema Evolution (Add/Drop Fields)

Support adding or removing fields from external collections after creation:

# Future: Add new field to external collection
client.add_field(
    collection_name="my_external_collection",
    field_name="new_column",
    datatype=DataType.VARCHAR,
    max_length=128,
    external_field="source_new_column"
)

# Future: Drop field from external collection
client.drop_field(
    collection_name="my_external_collection",
    field_name="old_column"
)

10.4 Additional Planned Features

Feature	Description	Priority
More Data Formats	Support Apache Iceberg, Delta Lake, ORC formats	High
Auto Data Sync	Automatic detection of external data source changes with scheduled refresh	Low
Partition Mapping	Map external data partitions to Milvus partitions	Medium
Text Match	Support full-text search on external collections	Medium
Cross-source Query	Query across multiple external data sources	Low
Change Data Capture	Support CDC-based incremental updates	Low

---

11. Related Documentation

• Design Document: External Table