+-------------------------------------------------------------------------+
|                              Client                                      |
|  - CreateCollection(schema with external_source)                         |
|  - AlterCollection (modify data source / trigger manual refresh)         |
+-----------------------------------+-------------------------------------+
                                    |
                                    v
+-------------------------------------------------------------------------+
|                             Proxy                                        |
|  - ValidateExternalCollectionSchema()                                    |
|  - Block write operations for external collections                       |
+-----------------------------------+-------------------------------------+
                                    |
                                    v
+-------------------------------------------------------------------------+
|                           RootCoord                                      |
|  - ValidateExternalCollectionSchema()                                    |
|  - Store ExternalSource/ExternalSpec in collection model                 |
+-----------------------------------+-------------------------------------+
                                    |
                                    v
+-------------------------------------------------------------------------+
|                           DataCoord                                      |
|  +---------------------------------------------------------------+      |
|  |              Compaction (DISABLED for external)                |      |
|  |  - Single/L0/Clustering compaction skipped                     |      |
|  +---------------------------------------------------------------+      |
|  +---------------------------------------------------------------+      |
|  |              Stats Inspector (LIMITED for external)            |      |
|  |  - TextIndexJob enabled; JSON enabled for StorageV3 manifests   |      |
|  |  - BM25 inspector jobs skipped                                 |      |
|  +---------------------------------------------------------------+      |
|  +---------------------------------------------------------------+      |
|  |              UpdateExternalCollectionTask                      |      |
|  |  - Handle manual refresh requests                              |      |
|  |  - Coordinate with DataNode for data sync                      |      |
|  +---------------------------------------------------------------+      |
+-----------------------------------+-------------------------------------+
                                    |
                                    v
+-------------------------------------------------------------------------+
|                           DataNode                                       |
|  +---------------------------------------------------------------+      |
|  |              ExternalCollectionManager                         |      |
|  |  - Task lifecycle management                                   |      |
|  |  - Worker pool for async execution                             |      |
|  +---------------------------------------------------------------+      |
|  +---------------------------------------------------------------+      |
|  |              UpdateExternalTask                                |      |
|  |  - Fetch fragments from external source                        |      |
|  |  - Compare with current segments                               |      |
|  |  - Organize orphan fragments into new segments                 |      |
|  |  - Create manifest files for segments                          |      |
|  +---------------------------------------------------------------+      |
+-----------------------------------+-------------------------------------+
                                    |
                                    v
+-------------------------------------------------------------------------+
|                           QueryNode                                      |
|  - Virtual PK generation: (segmentID << 32) | offset                    |
|  - ExternalFieldChunkedColumn: Lazy load via milvus-storage             |
|  - ExternalSegmentCandidate: PK-based segment matching                  |
|  - Skip delta logs and bloom filters for external collections           |
+-------------------------------------------------------------------------+

// Job state enumeration for external table refresh
enum RefreshExternalTableState {
    RefreshStatePending = 0;      // Job is queued, waiting to execute
    RefreshStateInProgress = 1;   // Job is currently executing
    RefreshStateCompleted = 2;    // Job completed successfully
    RefreshStateFailed = 3;       // Job failed with error
}

message RefreshExternalTableRequest {
    common.MsgBase base = 1;
    string db_name = 2;              // Database name
    string collection_name = 3;      // Collection name (required)
    string external_source = 4;      // Optional: new external source path
    string external_spec = 5;        // Optional: new external spec configuration
}

message RefreshExternalTableResponse {
    common.Status status = 1;
    string job_id = 2;               // Unique job identifier for tracking
}

# Refresh with current data source
response = client.refresh_external_table(
    collection_name="my_external_collection"
)
job_id = response.job_id

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

message GetRefreshExternalTableProgressRequest {
    common.MsgBase base = 1;
    string job_id = 2;               // Job ID from RefreshExternalTable response
}

message RefreshExternalTableJobInfo {
    string job_id = 1;                   // Job identifier
    string collection_name = 2;          // Collection name
    RefreshExternalTableState state = 3; // Current job state
    int64 progress = 4;                  // Progress percentage (0-100)
    string reason = 5;                   // Error message if failed
    string external_source = 6;          // External source used for this job
    int64 start_time = 7;                // Job start timestamp
    int64 end_time = 8;                  // Job end timestamp (0 if not completed)
}

message GetRefreshExternalTableProgressResponse {
    common.Status status = 1;
    RefreshExternalTableJobInfo job_info = 2;
}

# Get progress of a specific job
progress = client.get_refresh_external_table_progress(job_id="job_123456")
print(f"State: {progress.state}")
print(f"Progress: {progress.progress}%")

message ListRefreshExternalTableJobsRequest {
    common.MsgBase base = 1;
    string db_name = 2;              // Database name
    string collection_name = 3;      // Collection name (optional, if empty lists all)
    int64 limit = 4;                 // Max number of jobs to return (default: 100)
}

message ListRefreshExternalTableJobsResponse {
    common.Status status = 1;
    repeated RefreshExternalTableJobInfo jobs = 2;
}

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

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

schema := &schemapb.CollectionSchema{
    Name:           "my_external_collection",
    ExternalSource: "s3://bucket/path/to/data",
    ExternalSpec:   `{"format": "parquet"}`,
    Fields: []*schemapb.FieldSchema{
        {
            Name:          "text_field",
            DataType:      schemapb.DataType_VarChar,
            ExternalField: "source_text_column",  // Maps to external column
            TypeParams:    []*commonpb.KeyValuePair{{Key: "max_length", Value: "256"}},
        },
        {
            Name:          "vector_field",
            DataType:      schemapb.DataType_FloatVector,
            ExternalField: "source_embedding",
            TypeParams:    []*commonpb.KeyValuePair{{Key: "dim", Value: "128"}},
        },
    },
}

// IsExternalCollection returns true when schema describes an external collection.
// External collections are identified by having fields with ExternalField set,
// since ExternalSource can be null for empty external collections.
func IsExternalCollection(schema *schemapb.CollectionSchema) bool {
    if schema == nil {
        return false
    }
    for _, field := range schema.GetFields() {
        if field.GetExternalField() != "" {
            return true
        }
    }
    return false
}

// ValidateExternalCollectionSchema ensures unsupported features are disabled for external collections.
func ValidateExternalCollectionSchema(schema *schemapb.CollectionSchema) error {
    if !IsExternalCollection(schema) {
        return nil
    }

    if schema.GetEnableDynamicField() {
        return fmt.Errorf("external collection %s does not support dynamic field", schema.GetName())
    }

    if len(schema.GetStructArrayFields()) > 0 {
        return fmt.Errorf("external collection %s does not support struct fields", schema.GetName())
    }

    for _, field := range schema.GetFields() {
        // Skip system fields (RowID and Timestamp)
        if field.GetName() == common.RowIDFieldName || field.GetName() == common.TimeStampFieldName {
            continue
        }

        if field.GetIsPrimaryKey() {
            return fmt.Errorf("external collection %s does not support primary key field %s", schema.GetName(), field.GetName())
        }
        if field.GetIsPartitionKey() {
            return fmt.Errorf("external collection %s does not support partition key field %s", schema.GetName(), field.GetName())
        }
        if field.GetIsClusteringKey() {
            return fmt.Errorf("external collection %s does not support clustering key field %s", schema.GetName(), field.GetName())
        }
        if field.GetAutoID() {
            return fmt.Errorf("external collection %s does not support auto id on field %s", schema.GetName(), field.GetName())
        }

        helper := CreateFieldSchemaHelper(field)
        if helper.EnableMatch() {
            return fmt.Errorf("external collection %s does not support text match on field %s", schema.GetName(), field.GetName())
        }

        // Validate external_field mapping is set for all user fields
        if field.GetExternalField() == "" {
            return fmt.Errorf("field '%s' in external collection %s must have external_field mapping", field.GetName(), schema.GetName())
        }
    }

    return nil
}

func (t *createCollectionTask) PreExecute(ctx context.Context) error {
    // ...
    isExternalCollection := typeutil.IsExternalCollection(t.schema)
    if err := typeutil.ValidateExternalCollectionSchema(t.schema); err != nil {
        return err
    }

    // validate primary key definition when needed
    if !isExternalCollection {
        if err := validatePrimaryKey(t.schema); err != nil {
            return err
        }
    }
    // ...
}

message FieldSchema {
    // ... other fields ...
    string external_field = 17;  // external field name - maps to column name in external source
}

field := &schemapb.FieldSchema{
    Name:          "vector",           // Milvus field name
    DataType:      schemapb.DataType_FloatVector,
    ExternalField: "embedding_col",    // Column name in external Parquet file
}

message CollectionSchema {
    string name = 1;
    // ... other fields ...
    string external_source = 11;  // External data source (e.g., "s3://bucket/path")
    string external_spec = 12;    // External source config (JSON)
}

type Collection struct {
    // ... existing fields ...
    ExternalSource string
    ExternalSpec   string
}

{
    "format": "parquet"
}

type Fragment struct {
    FragmentID int64   // Unique fragment identifier
    FilePath   string  // File path in external storage
    StartRow   int64   // Start row index within the file (inclusive)
    EndRow     int64   // End row index within the file (exclusive)
    RowCount   int64   // Number of rows (EndRow - StartRow)
}

type SegmentRowMapping struct {
    SegmentID int64
    TotalRows int64
    Ranges    []FragmentRowRange
    Fragments []exttable.Fragment
}

type FragmentRowRange struct {
    FragmentID int64
    StartRow   int64  // inclusive
    EndRow     int64  // exclusive
}

// In each compaction policy/trigger
if collection.IsExternal() {
    log.Info("skip compaction for external collection", zap.Int64("collectionID", collection.ID))
    continue  // or return nil
}

func (si *statsInspector) SubmitStatsTask(..., subJobType indexpb.StatsSubJob, ...) {
    if si.isExternalCollection(segment.GetCollectionID()) {
        if subJobType == indexpb.StatsSubJob_JsonKeyIndexJob &&
            !canBuildExternalJSONKeyIndex(segment) {
            log.Info("skip submit external json stats task without v3 manifest")
            return nil
        }
        if subJobType != indexpb.StatsSubJob_TextIndexJob &&
            subJobType != indexpb.StatsSubJob_JsonKeyIndexJob {
            log.Info("skip submit stats task for external collection")
            return nil
        }
    }
    // ... submit task
}

inline int64_t GenerateVirtualPK(int64_t segment_id, int32_t offset) {
    return (segment_id << 32) | static_cast<int64_t>(offset);
}

inline std::pair<int64_t, int32_t> ParseVirtualPK(int64_t virtual_pk) {
    int64_t segment_id = virtual_pk >> 32;
    int32_t offset = static_cast<int32_t>(virtual_pk & 0xFFFFFFFF);
    return {segment_id, offset};
}

class VirtualPKChunkedColumn : public ChunkedColumnBase {
public:
    // Generates PKs based on segment ID and row offset
    // No actual data storage needed
    int64_t GetPK(int64_t row_offset) const {
        return GenerateVirtualPK(segment_id_, row_offset);
    }
};

class ExternalFieldChunkedColumn : public ChunkedColumnBase {
    // Loads data chunks from external source on demand
    // Uses milvus-storage library for S3/HDFS/etc. access
};

type ExternalSegmentCandidate struct {
    segmentID int64
    partition int64
    typ       commonpb.SegmentState
}

func (c *ExternalSegmentCandidate) MayPkExist(pk storage.PrimaryKey) bool {
    // Parse virtual PK to extract segment ID
    virtualPK := pk.GetValue().(int64)
    segmentID := virtualPK >> 32
    return segmentID == c.segmentID
}

func (loader *segmentLoader) Load(...) {
    if isExternalCollection(collectionSchema) {
        // 1. Virtual PK field already injected during creation

        // 2. Skip delta logs (no delete support)
        // 3. Skip bloom filter building
        // 4. Use ExternalFieldChunkedColumn for field data
        // 5. Use VirtualPKChunkedColumn for PK field
        // 6. Use ExternalSegmentCandidate instead of bloom filter
    }
}

                            Client
                              |
                              | RefreshExternalTable(collection_name)
                              v
                            Proxy
                              |
                              | Validate & Forward
                              v
                          DataCoord
                              |
                              | Create RefreshJob
                              v
                    ExternalCollectionTaskMeta
                    (Store job with Pending state)
                              |
                              v
                    ExternalCollectionScheduler
                              |
                              | Schedule job execution
                              v
                    CreateTaskOnWorker
                              |
                              v
                        DataNode
                (ExternalCollectionManager)
                              |
                              v
                    UpdateExternalTask
                              |
    +-------------------------+-------------------------+
    |                         |                         |
    v                         v                         v
Fetch fragments      Compare with          Organize orphan
from source         current segments       fragments to new
                                           segments
                              |
                              v
                    Create manifests
                              |
                              | Report progress
                              v
                        DataCoord
                  (UpdateJobProgress callback)
                              |
                              | Update job state & progress
                              v
                    ExternalCollectionTaskMeta
                              |
    +-------------------------+-------------------------+
    |                         |                         |
    v                         v                         v
Keep unchanged       Drop obsolete         Add new
segments             segments              segments
                              |
                              v
                    Mark job as Completed/Failed
                              |
                              v
                        Client
                              |
              GetRefreshExternalTableProgress(job_id)
                              |
                    [Poll until completed]

     RefreshExternalTable()
              |
              v
       +------+------+
       |   Pending   |  <-- Job created, queued for execution
       +------+------+
              |
              | (Scheduler picks up job)
              v
       +------+------+
       | InProgress  |  <-- DataNode executing refresh task
       +------+------+
              |
        +-----+-----+
        |           |
        v           v
  +-----+----+  +---+-----+
  | Completed|  | Failed  |
  +----------+  +---------+

type ExternalCollectionScheduler interface {
    Start()
    Stop()

    // SubmitRefreshJob creates a new refresh job for the collection
    // Returns job_id for tracking
    SubmitRefreshJob(ctx context.Context, req *RefreshJobRequest) (string, error)

    // GetJobProgress returns the current progress of a job
    GetJobProgress(ctx context.Context, jobID string) (*RefreshJobProgress, error)

    // ListJobs returns all jobs for a collection (or all if collectionID is 0)
    ListJobs(ctx context.Context, collectionID int64, limit int) ([]*RefreshJobInfo, error)
}

type RefreshJobRequest struct {
    CollectionID   int64
    CollectionName string
    ExternalSource string  // Optional: update source before refresh
    ExternalSpec   string  // Optional: update spec before refresh
}

func (s *externalCollectionScheduler) SubmitRefreshJob(ctx context.Context, req *RefreshJobRequest) (string, error) {
    // 1. Generate unique job ID
    jobID := fmt.Sprintf("refresh_%d_%d", req.CollectionID, time.Now().UnixNano())

    // 2. If external source/spec provided, update collection metadata
    if req.ExternalSource != "" || req.ExternalSpec != "" {
        if err := s.updateCollectionExternalConfig(ctx, req); err != nil {
            return "", err
        }
    }

    // 3. Create job record with Pending state
    job := &ExternalCollectionTask{
        JobID:          jobID,
        CollectionID:   req.CollectionID,
        CollectionName: req.CollectionName,
        ExternalSource: req.ExternalSource,
        ExternalSpec:   req.ExternalSpec,
        State:          RefreshStatePending,
        StartTime:      time.Now().UnixMilli(),
    }

    if err := s.taskMeta.AddTask(job); err != nil {
        return "", err
    }

    // 4. Enqueue for execution
    s.jobQueue <- job

    return jobID, nil
}

type ExternalCollectionTaskMeta interface {
    // Job management
    GetJob(jobID string) (*ExternalCollectionTask, error)
    AddTask(task *ExternalCollectionTask) error
    UpdateTask(task *ExternalCollectionTask) error

    // Query methods
    ListJobsByCollection(collectionID int64, limit int) ([]*ExternalCollectionTask, error)
    ListAllJobs(limit int) ([]*ExternalCollectionTask, error)

    // Cleanup
    CleanupCompletedJobs(olderThan time.Duration) error
}

type ExternalCollectionTask struct {
    JobID          string                     // Unique job identifier
    CollectionID   int64                      // Collection ID
    CollectionName string                     // Collection name
    ExternalSource string                     // External source path used for this job
    ExternalSpec   string                     // External spec used for this job
    State          RefreshExternalTableState  // Current job state

    // Progress tracking
    TotalFragments     int64  // Total fragments to process
    ProcessedFragments int64  // Fragments processed so far
    NewSegments        int64  // Number of new segments created
    DroppedSegments    int64  // Number of segments dropped
    KeptSegments       int64  // Number of segments kept unchanged

    // Timestamps
    StartTime int64  // Job start time (Unix epoch ms)
    EndTime   int64  // Job end time (0 if not completed)

    // Error info
    Reason string  // Error message if failed
}

type UpdateExternalCollectionTask struct {
    taskID         int64
    collectionID   int64
    externalSource string
    externalSpec   string
    // ...
}

// Task lifecycle methods
func (t *UpdateExternalCollectionTask) CreateTaskOnWorker() error
func (t *UpdateExternalCollectionTask) QueryTaskOnWorker() error
func (t *UpdateExternalCollectionTask) SetJobInfo() error      // Process results
func (t *UpdateExternalCollectionTask) DropTaskOnWorker() error

type ExternalCollectionManager struct {
    ctx       context.Context
    mu        sync.RWMutex
    tasks     map[TaskKey]*TaskInfo
    pool      *conc.Pool[any]
}

func (m *ExternalCollectionManager) SubmitTask(
    clusterID string,
    req *datapb.UpdateExternalCollectionRequest,
    taskFunc func(context.Context) (*datapb.UpdateExternalCollectionResponse, error),
) error

type UpdateExternalTask struct {
    ctx    context.Context
    req    *datapb.UpdateExternalCollectionRequest
    // ...
}

func (t *UpdateExternalTask) Execute(ctx context.Context) error {
    // 1. Fetch fragments from external source
    newFragments, err := t.fetchFragmentsFromExternalSource(ctx)

    // 2. Build current segment -> fragments mapping
    currentSegmentFragments := t.buildCurrentSegmentFragments()

    // 3. Compare and organize segments
    updatedSegments, err := t.organizeSegments(ctx, currentSegmentFragments, newFragments)

    return nil
}

Current Segments in Milvus:  [S1, S2, S3, S4, S5]

Worker Response:
  - keptSegments: [S1, S3]      (fragments unchanged)
  - updatedSegments: [S6', S7']  (new segments from orphan fragments)

Processing:
  1. Keep: S1, S3 (unchanged)
  2. Drop: S2, S4, S5 (mark as Dropped)
  3. Add:  S6, S7 (allocate new segment IDs)

Final Segments: [S1, S3, S6, S7]

func (t *UpdateExternalTask) balanceFragmentsToSegments(
    ctx context.Context,
    fragments []exttable.Fragment,
) ([]*datapb.SegmentInfo, error) {
    // 1. Calculate total rows
    // 2. Determine target rows per segment (default: 1M rows)
    // 3. Sort fragments by row count descending
    // 4. Greedy bin-packing: assign each fragment to bin with lowest row count
    // 5. Create manifest for each segment
    // 6. Return SegmentInfo list
}

func CreateManifestForSegment(
    basePath string,
    columns []string,
    format string,
    fragments []Fragment,
    storageConfig *indexpb.StorageConfig,
) (string, error) {
    // 1. Create column groups from fragments
    // 2. Begin transaction
    // 3. Commit transaction with column groups
    // 4. Return manifest path
}

func ReadFragmentsFromManifest(
    manifestPath string,
    storageConfig *indexpb.StorageConfig,
) ([]Fragment, error) {
    // 1. Parse manifest path to get base path
    // 2. Create properties from storage config
    // 3. Call exttable_read_column_groups FFI
    // 4. Extract fragments from column groups
    // 5. Return fragment list
}