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
)

package main

import (
    "context"
    "log"

    "github.com/milvus-io/milvus/client/v3"
    "github.com/milvus-io/milvus/client/v3/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)
    }
}

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

# 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"]
)

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

# 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
)

# Drop External Collection
client.drop_collection("my_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"}'
)

# 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}")

# 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()

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")

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

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}%)")

package main

import (
    "context"
    "fmt"
    "log"

    "github.com/milvus-io/milvus/client/v3"
    "github.com/milvus-io/milvus/client/v3/entity"
    "github.com/milvus-io/milvus/client/v3/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")
}

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

# 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"}
)

# 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"
)