package testcases

import (
	"bytes"
	"context"
	"fmt"
	"net"
	"net/url"
	"os"
	"os/exec"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/apache/arrow/go/v17/arrow"
	"github.com/apache/arrow/go/v17/arrow/array"
	"github.com/apache/arrow/go/v17/arrow/memory"
	"github.com/apache/arrow/go/v17/parquet/pqarrow"
	miniogo "github.com/minio/minio-go/v7"
	miniocreds "github.com/minio/minio-go/v7/pkg/credentials"
	"github.com/stretchr/testify/require"

	"github.com/milvus-io/milvus/client/v2/column"
	"github.com/milvus-io/milvus/client/v2/entity"
	"github.com/milvus-io/milvus/client/v2/index"
	client "github.com/milvus-io/milvus/client/v2/milvusclient"
	"github.com/milvus-io/milvus/tests/go_client/base"
	"github.com/milvus-io/milvus/tests/go_client/common"
	hp "github.com/milvus-io/milvus/tests/go_client/testcases/helper"
)

// --- MinIO helpers (configurable via environment variables) ---

func envOrDefault(key, defaultVal string) string {
	if v := os.Getenv(key); v != "" {
		return v
	}
	switch key {
	case "MINIO_ADDRESS":
		if address, _, ok := inferGoSDKMinIOConfig(); ok {
			return address
		}
	case "MINIO_BUCKET":
		if _, bucket, ok := inferGoSDKMinIOConfig(); ok {
			return bucket
		}
	case "MINIO_ROOT_PATH":
		if _, _, ok := inferGoSDKMinIOConfig(); ok {
			// The go-sdk Helm deployment writes Milvus object data under
			// minio.rootPath=file. Keep the inferred MinIO config complete so
			// tests that inspect persisted objects use the same prefix.
			return "file"
		}
	}
	return defaultVal
}

func inferGoSDKMinIOConfig() (string, string, bool) {
	return inferGoSDKMinIOConfigFromAddr(hp.GetAddr())
}

func inferGoSDKMinIOConfigFromAddr(addr string) (string, string, bool) {
	host := hostFromMilvusAddr(addr)
	serviceName, namespaceSuffix, _ := strings.Cut(host, ".")
	if !strings.HasPrefix(serviceName, "gosdk-") || !strings.HasSuffix(serviceName, "-milvus") {
		return "", "", false
	}
	minioServiceName := strings.TrimSuffix(serviceName, "-milvus") + "-minio"
	minioHost := minioServiceName
	if namespaceSuffix != "" {
		minioHost += "." + namespaceSuffix
	}
	return net.JoinHostPort(minioHost, "9000"), "milvus-bucket", true
}

func hostFromMilvusAddr(addr string) string {
	addr = strings.TrimSpace(addr)
	if addr == "" {
		return ""
	}
	if u, err := url.Parse(addr); err == nil && u.Host != "" {
		return u.Hostname()
	}

	addr = strings.TrimPrefix(addr, "//")
	if idx := strings.IndexByte(addr, '/'); idx >= 0 {
		addr = addr[:idx]
	}
	if host, _, err := net.SplitHostPort(addr); err == nil {
		return strings.Trim(host, "[]")
	}
	if idx := strings.LastIndexByte(addr, ':'); idx >= 0 {
		addr = addr[:idx]
	}
	return strings.Trim(addr, "[]")
}

type minioConfig struct {
	address   string
	accessKey string
	secretKey string
	bucket    string
	rootPath  string
}

func getMinIOConfig() minioConfig {
	return minioConfig{
		address:   envOrDefault("MINIO_ADDRESS", "localhost:9000"),
		accessKey: envOrDefault("MINIO_ACCESS_KEY", "minioadmin"),
		secretKey: envOrDefault("MINIO_SECRET_KEY", "minioadmin"),
		bucket:    envOrDefault("MINIO_BUCKET", "a-bucket"),
		rootPath:  envOrDefault("MINIO_ROOT_PATH", "files"),
	}
}

func newMinIOClient(cfg minioConfig) (*miniogo.Client, error) {
	return miniogo.New(cfg.address, &miniogo.Options{
		Creds:  miniocreds.NewStaticV4(cfg.accessKey, cfg.secretKey, ""),
		Secure: false,
	})
}

// extTestURI wraps a MinIO-relative object path into a full MinIO URI pointing
// at the configured bucket. The minio:// scheme is unambiguous, so tests do
// not need to carry AWS-family region or provider fields just to disambiguate
// s3:// from AWS S3.
func extTestURI(cfg minioConfig, relPath string) string {
	return fmt.Sprintf("minio://%s/%s/%s", cfg.address, cfg.bucket, relPath)
}

// extTestSpec returns an ExternalSpec JSON fragment for external-table tests
// that target Milvus's own MinIO.
func extTestSpec(cfg minioConfig, format string) string {
	return fmt.Sprintf(
		`{"format":%q,"extfs":{"cloud_provider":"minio","access_key_id":%q,"access_key_value":%q}}`,
		format, cfg.accessKey, cfg.secretKey)
}

// cleanupMinIOPrefix removes all objects under a prefix in the given bucket.
func cleanupMinIOPrefix(ctx context.Context, mc *miniogo.Client, bucket, prefix string) {
	for obj := range mc.ListObjects(ctx, bucket, miniogo.ListObjectsOptions{
		Prefix:    prefix,
		Recursive: true,
	}) {
		if obj.Err != nil {
			continue
		}
		_ = mc.RemoveObject(ctx, bucket, obj.Key, miniogo.RemoveObjectOptions{})
	}
}

func skipIfMinIOUnreachable(ctx context.Context, t *testing.T, mc *miniogo.Client, bucket string) {
	t.Helper()
	checkCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
	defer cancel()
	if _, err := mc.BucketExists(checkCtx, bucket); err != nil {
		t.Skipf("MinIO unreachable (bucket=%s): %v", bucket, err)
	}
}

// --- Parquet generation ---

const (
	testVecDim     = 4
	perfTestVecDim = 128
)

const (
	externalDataSchemaBasic           = "basic"
	externalDataSchemaMulti           = "multi"
	externalDataSchemaLarge           = "large"
	externalDataSchemaNullableVector  = "nullable_vector"
	externalDataSchemaSnapshotRestore = "snapshot_restore"
)

func generateParquetBytes(schema string, numRows, startID int64, dim int) ([]byte, error) {
	switch schema {
	case externalDataSchemaBasic, externalDataSchemaMulti,
		externalDataSchemaLarge, externalDataSchemaNullableVector,
		externalDataSchemaSnapshotRestore:
		return generateParquetBytesWithCompression(schema, numRows, startID, dim, "")
	default:
		return nil, fmt.Errorf("unknown parquet data schema: %s", schema)
	}
}

// generateSnapshotRestoreParquetBytes creates a compact Parquet file for the
// snapshot restore e2e. It stores float vectors as FixedSizeBinary so the test
// focuses on the snapshot restore path instead of list-vector decoding.
func generateSnapshotRestoreParquetBytes(numRows int64, startID int64) ([]byte, error) {
	return generateParquetBytes(
		externalDataSchemaSnapshotRestore, numRows, startID, testVecDim)
}

func generateParquetBytesWithCompression(schema string, numRows, startID int64, dim int, compression string) ([]byte, error) {
	tmp, err := os.CreateTemp("", "external-table-*.parquet")
	if err != nil {
		return nil, fmt.Errorf("create temp parquet file: %w", err)
	}
	tmpPath := tmp.Name()
	if err := tmp.Close(); err != nil {
		return nil, fmt.Errorf("close temp parquet file: %w", err)
	}
	defer os.Remove(tmpPath)

	args := []string{
		"generate_parquet_data.py",
		"--schema", schema,
		tmpPath,
		strconv.FormatInt(numRows, 10),
		"--start-id", strconv.FormatInt(startID, 10),
		"--vec-dim", strconv.Itoa(dim),
		"--bin-vec-dim", strconv.Itoa(testBinVecDim),
	}
	if compression != "" {
		args = append(args, "--compression", compression)
	}

	output, err := exec.Command("python3", args...).CombinedOutput() // #nosec G204
	if err != nil {
		return nil, fmt.Errorf("generate parquet data via python: %w: %s", err, strings.TrimSpace(string(output)))
	}

	data, err := os.ReadFile(tmpPath)
	if err != nil {
		return nil, fmt.Errorf("read generated parquet file: %w", err)
	}
	if len(data) == 0 {
		return nil, fmt.Errorf("generated empty parquet file: %s", strings.TrimSpace(string(output)))
	}

	return data, nil
}

func externalTakeDenseBytes(row int64, byteWidth int, seed int64) []byte {
	data := make([]byte, byteWidth)
	for i := range data {
		data[i] = byte((seed + row*int64(byteWidth) + int64(i)) % 251)
	}
	return data
}

func externalTakeInt8Vector(row int64) []int8 {
	data := make([]int8, testVecDim)
	for i := range data {
		data[i] = int8((row*3 + int64(i)) % 127)
	}
	return data
}

func externalTakeInt8VectorBytes(row int64) []byte {
	values := externalTakeInt8Vector(row)
	data := make([]byte, len(values))
	for i, value := range values {
		data[i] = byte(value)
	}
	return data
}

func generateExternalTakeVectorParquetBytes(numRows int64) ([]byte, error) {
	const binVecDim = 16
	binVecByteWidth := binVecDim / 8
	fp16ByteWidth := testVecDim * 2
	bf16ByteWidth := testVecDim * 2

	arrowSchema := arrow.NewSchema(
		[]arrow.Field{
			{Name: "id", Type: arrow.PrimitiveTypes.Int64},
			{Name: "bin_vec", Type: &arrow.FixedSizeBinaryType{ByteWidth: binVecByteWidth}},
			{Name: "fp16_vec", Type: &arrow.FixedSizeBinaryType{ByteWidth: fp16ByteWidth}},
			{Name: "bf16_vec", Type: &arrow.FixedSizeBinaryType{ByteWidth: bf16ByteWidth}},
			{Name: "int8_vec", Type: &arrow.FixedSizeBinaryType{ByteWidth: testVecDim}},
		},
		nil,
	)

	var buf bytes.Buffer
	writer, err := pqarrow.NewFileWriter(arrowSchema, &buf, nil, pqarrow.DefaultWriterProps())
	if err != nil {
		return nil, fmt.Errorf("create parquet writer: %w", err)
	}

	pool := memory.NewGoAllocator()
	builder := array.NewRecordBuilder(pool, arrowSchema)
	defer builder.Release()

	idBuilder := builder.Field(0).(*array.Int64Builder)
	binBuilder := builder.Field(1).(*array.FixedSizeBinaryBuilder)
	fp16Builder := builder.Field(2).(*array.FixedSizeBinaryBuilder)
	bf16Builder := builder.Field(3).(*array.FixedSizeBinaryBuilder)
	int8Builder := builder.Field(4).(*array.FixedSizeBinaryBuilder)

	for row := int64(0); row < numRows; row++ {
		idBuilder.Append(row)
		binBuilder.Append(externalTakeDenseBytes(row, binVecByteWidth, 11))
		fp16Builder.Append(externalTakeDenseBytes(row, fp16ByteWidth, 31))
		bf16Builder.Append(externalTakeDenseBytes(row, bf16ByteWidth, 51))
		int8Builder.Append(externalTakeInt8VectorBytes(row))
	}

	record := builder.NewRecord()
	defer record.Release()

	if err := writer.Write(record); err != nil {
		return nil, fmt.Errorf("write record: %w", err)
	}
	if err := writer.Close(); err != nil {
		return nil, fmt.Errorf("close writer: %w", err)
	}

	return buf.Bytes(), nil
}

// --- Multi-type parquet generation ---

// generateMultiTypeParquetBytes creates a Parquet file with multiple data types:
//   - id (Int64), bool_val (Boolean), int8_val (Int8), int16_val (Int16),
//   - int32_val (Int32), float_val (Float32), double_val (Float64),
//   - embedding (FixedSizeList[Float32, testVecDim])
//
// Data formulas for row i (startID+i):
//
//	bool_val = (i is even), int8_val = i%100, int16_val = i*10,
//	int32_val = i*100, float_val = i*1.5, double_val = i*0.01
const testBinVecDim = 8 // BinaryVector dimension must be multiple of 8

// --- Helpers ---

func uploadParquetToMinIO(ctx context.Context, t *testing.T, mc *miniogo.Client, bucket, objectKey string, data []byte) {
	t.Helper()
	_, err := mc.PutObject(ctx, bucket, objectKey,
		bytes.NewReader(data), int64(len(data)),
		miniogo.PutObjectOptions{ContentType: "application/octet-stream"})
	require.NoError(t, err, "upload parquet file to MinIO: %s", objectKey)
	t.Logf("Uploaded %s/%s (%d bytes)", bucket, objectKey, len(data))
}

func waitForRefreshComplete(ctx context.Context, t *testing.T, mc *base.MilvusClient, jobID int64, timeout time.Duration) {
	t.Helper()
	deadline := time.After(timeout)
	ticker := time.NewTicker(2 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-deadline:
			t.Fatalf("Refresh job %d timed out after %s", jobID, timeout)
		case <-ticker.C:
			progress, err := mc.GetRefreshExternalCollectionProgress(ctx,
				client.NewGetRefreshExternalCollectionProgressOption(jobID))
			require.NoError(t, err)
			t.Logf("Job %d: state=%s progress=%d%%", jobID, progress.State, progress.Progress)

			if progress.State == entity.RefreshStateCompleted {
				return
			}
			if progress.State == entity.RefreshStateFailed {
				t.Fatalf("Refresh job %d failed: %s", jobID, progress.Reason)
			}
		}
	}
}

func refreshAndWait(ctx context.Context, t *testing.T, mc *base.MilvusClient, collName string) int64 {
	t.Helper()
	refreshResult, err := mc.RefreshExternalCollection(ctx,
		client.NewRefreshExternalCollectionOption(collName))
	common.CheckErr(t, err, true)
	require.Greater(t, refreshResult.JobID, int64(0))
	t.Logf("Started refresh job: %d", refreshResult.JobID)

	waitForRefreshComplete(ctx, t, mc, refreshResult.JobID, 120*time.Second)
	return refreshResult.JobID
}

func indexAndLoadCollection(ctx context.Context, t *testing.T, mc *base.MilvusClient, collName, vecFieldName string) {
	t.Helper()
	idx := index.NewAutoIndex(entity.L2)
	idxTask, err := mc.CreateIndex(ctx, client.NewCreateIndexOption(collName, vecFieldName, idx))
	common.CheckErr(t, err, true)
	err = idxTask.Await(ctx)
	common.CheckErr(t, err, true)
	t.Log("Vector index created on " + vecFieldName)

	loadTask, err := mc.LoadCollection(ctx, client.NewLoadCollectionOption(collName))
	common.CheckErr(t, err, true)
	err = loadTask.Await(ctx)
	common.CheckErr(t, err, true)
	t.Log("Collection loaded")
}

// indexAndLoadCollectionWithScalarAndVector creates scalar indexes on id and value fields,
// a vector index on the embedding field, verifies all indexes are built successfully,
// and then loads the collection.
func indexAndLoadCollectionWithScalarAndVector(ctx context.Context, t *testing.T, mc *base.MilvusClient, collName string, totalRows int64) {
	t.Helper()

	// Create scalar index on "id" field
	idIdx := index.NewAutoIndex(entity.IP)
	idIdxTask, err := mc.CreateIndex(ctx, client.NewCreateIndexOption(collName, "id", idIdx))
	common.CheckErr(t, err, true)
	err = idIdxTask.Await(ctx)
	common.CheckErr(t, err, true)
	t.Log("Scalar index created on id field")

	// NOTE: Skip scalar index on "value" field — InvertedIndexTantivy::Load
	// crashes on ARM64 macOS (Tantivy FFI bug, not related to external table).
	// Re-enable when running on Linux CI.
	// valIdx := index.NewAutoIndex(entity.IP)
	// valIdxTask, err := mc.CreateIndex(ctx, client.NewCreateIndexOption(collName, "value", valIdx))
	// common.CheckErr(t, err, true)
	// err = valIdxTask.Await(ctx)
	// common.CheckErr(t, err, true)
	t.Log("Scalar index on value field: SKIPPED (ARM64 Tantivy bug)")

	// Create vector index on "embedding" field
	vecIdx := index.NewAutoIndex(entity.L2)
	vecIdxTask, err := mc.CreateIndex(ctx, client.NewCreateIndexOption(collName, "embedding", vecIdx))
	common.CheckErr(t, err, true)
	err = vecIdxTask.Await(ctx)
	common.CheckErr(t, err, true)
	t.Log("Vector index created on embedding field")

	// Verify all indexes via DescribeIndex (skip "value" — ARM64 Tantivy bug)
	for _, fieldName := range []string{"id", "embedding"} {
		descIdx, err := mc.DescribeIndex(ctx, client.NewDescribeIndexOption(collName, fieldName))
		common.CheckErr(t, err, true)
		require.Equal(t, index.IndexState(3), descIdx.State,
			"index on %s should be Finished (state=3)", fieldName) // 3 = IndexState_Finished
		require.Equal(t, totalRows, descIdx.TotalRows,
			"index on %s: TotalRows should be %d", fieldName, totalRows)
		require.Equal(t, totalRows, descIdx.IndexedRows,
			"index on %s: IndexedRows should be %d", fieldName, totalRows)
		require.Equal(t, int64(0), descIdx.PendingIndexRows,
			"index on %s: PendingIndexRows should be 0", fieldName)
		t.Logf("Index on %s: state=%d totalRows=%d indexedRows=%d pendingRows=%d",
			fieldName, descIdx.State, descIdx.TotalRows, descIdx.IndexedRows, descIdx.PendingIndexRows)
	}

	// Load collection
	loadTask, err := mc.LoadCollection(ctx, client.NewLoadCollectionOption(collName))
	common.CheckErr(t, err, true)
	err = loadTask.Await(ctx)
	common.CheckErr(t, err, true)
	t.Log("Collection loaded")
}

// --- E2E Tests ---

func TestExternalCollectionSnapshotRestoreAndAccess(t *testing.T) {
	t.Parallel()

	ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	minioCfg := getMinIOConfig()
	minioClient, err := newMinIOClient(minioCfg)
	require.NoError(t, err)
	skipIfMinIOUnreachable(ctx, t, minioClient, minioCfg.bucket)

	collName := common.GenRandomString("ext_snapshot_restore", 6)
	restoredCollName := fmt.Sprintf("restored_%s", collName)
	snapshotName := fmt.Sprintf("snapshot_%s", common.GenRandomString("ext", 6))
	extPath := fmt.Sprintf("external-e2e-test/%s", collName)
	collectionsToClean := []string{collName, restoredCollName}

	t.Cleanup(func() {
		_ = mc.DropSnapshot(context.Background(), client.NewDropSnapshotOption(snapshotName, collName))
		for _, name := range collectionsToClean {
			_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(name))
		}
		cleanupMinIOPrefix(context.Background(), minioClient, minioCfg.bucket, fmt.Sprintf("%s/", extPath))
	})

	const numRows = int64(100)
	data, err := generateSnapshotRestoreParquetBytes(numRows, 0)
	require.NoError(t, err)
	objectKey := fmt.Sprintf("%s/data.parquet", extPath)
	uploadParquetToMinIO(ctx, t, minioClient, minioCfg.bucket, objectKey, data)

	schema := entity.NewSchema().
		WithName(collName).
		WithExternalSource(extTestURI(minioCfg, extPath)).
		WithExternalSpec(extTestSpec(minioCfg, "parquet")).
		WithField(entity.NewField().
			WithName("id").
			WithDataType(entity.FieldTypeInt64).
			WithExternalField("id")).
		WithField(entity.NewField().
			WithName("value").
			WithDataType(entity.FieldTypeFloat).
			WithExternalField("value")).
		WithField(entity.NewField().
			WithName("embedding").
			WithDataType(entity.FieldTypeFloatVector).
			WithDim(testVecDim).
			WithExternalField("embedding"))

	err = mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
	common.CheckErr(t, err, true)

	refreshAndWait(ctx, t, mc, collName)
	indexAndLoadCollection(ctx, t, mc, collName, "embedding")

	err = mc.CreateSnapshot(ctx, client.NewCreateSnapshotOption(snapshotName, collName).
		WithDescription("external collection snapshot restore e2e"))
	common.CheckErr(t, err, true)

	jobID, err := mc.RestoreSnapshot(ctx, client.NewRestoreSnapshotOption(snapshotName, collName, restoredCollName))
	common.CheckErr(t, err, true)
	_, err = waitForRestoreComplete(ctx, mc, jobID, 3*time.Minute)
	common.CheckErr(t, err, true)

	loadTask, err := mc.LoadCollection(ctx, client.NewLoadCollectionOption(restoredCollName).WithReplica(1))
	common.CheckErr(t, err, true)
	err = loadTask.Await(ctx)
	common.CheckErr(t, err, true)

	countRes, err := mc.Query(ctx, client.NewQueryOption(restoredCollName).
		WithConsistencyLevel(entity.ClStrong).
		WithOutputFields(common.QueryCountFieldName))
	common.CheckErr(t, err, true)
	count, err := countRes.GetColumn(common.QueryCountFieldName).GetAsInt64(0)
	require.NoError(t, err)
	require.Equal(t, numRows, count)

	filterRes, err := mc.Query(ctx, client.NewQueryOption(restoredCollName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("id < 10").
		WithOutputFields("id", "value"))
	common.CheckErr(t, err, true)
	require.Equal(t, 10, filterRes.GetColumn("id").Len())

	vec := make([]float32, testVecDim)
	for i := range vec {
		vec[i] = float32(i) * 0.1
	}
	searchRes, err := mc.Search(ctx, client.NewSearchOption(restoredCollName, 5,
		[]entity.Vector{entity.FloatVector(vec)}).
		WithConsistencyLevel(entity.ClStrong).
		WithANNSField("embedding").
		WithOutputFields("id", "value"))
	common.CheckErr(t, err, true)
	require.Equal(t, 1, len(searchRes))
	require.Greater(t, searchRes[0].ResultCount, 0)
}

// TestRefreshExternalCollectionAndVerifySegments tests the full external collection workflow:
//  1. Upload parquet test data to MinIO
//  2. Create an external collection with matching schema
//  3. Trigger a refresh job
//  4. Wait for refresh to complete
//  5. Verify the generated segment count and total row count
func TestRefreshExternalCollectionAndVerifySegments(t *testing.T) {
	t.Parallel()

	ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	// Connect to MinIO (skip if unavailable)
	minioCfg := getMinIOConfig()
	minioClient, err := newMinIOClient(minioCfg)
	require.NoError(t, err)

	exists, err := minioClient.BucketExists(ctx, minioCfg.bucket)
	if err != nil || !exists {
		t.Skipf("MinIO bucket %q not accessible (exists=%v, err=%v), skipping",
			minioCfg.bucket, exists, err)
	}

	collName := common.GenRandomString("ext_refresh", 6)
	extPath := fmt.Sprintf("external-e2e-test/%s", collName)

	// ---------------------------------------------------------------
	// Step 1: Generate and upload parquet data to MinIO
	// ---------------------------------------------------------------
	const numFiles = 2
	const rowsPerFile = int64(1000)
	totalExpectedRows := int64(numFiles) * rowsPerFile

	for i := 0; i < numFiles; i++ {
		data, err := generateParquetBytes(externalDataSchemaBasic, rowsPerFile, int64(i)*rowsPerFile, testVecDim)
		require.NoError(t, err, "generate parquet file %d", i)

		objectKey := fmt.Sprintf("%s/data%d.parquet", extPath, i)
		_, err = minioClient.PutObject(ctx, minioCfg.bucket, objectKey,
			bytes.NewReader(data), int64(len(data)),
			miniogo.PutObjectOptions{ContentType: "application/octet-stream"})
		require.NoError(t, err, "upload parquet file %d to MinIO", i)
		t.Logf("Uploaded %s/%s (%d bytes, %d rows)", minioCfg.bucket, objectKey, len(data), rowsPerFile)
	}

	t.Cleanup(func() {
		cleanupMinIOPrefix(context.Background(), minioClient, minioCfg.bucket,
			fmt.Sprintf("%s/", extPath))
	})

	// ---------------------------------------------------------------
	// Step 2: Create external collection
	// ---------------------------------------------------------------
	schema := entity.NewSchema().
		WithName(collName).
		WithExternalSource(extTestURI(minioCfg, extPath)).
		WithExternalSpec(extTestSpec(minioCfg, "parquet")).
		WithField(
			entity.NewField().
				WithName("id").
				WithDataType(entity.FieldTypeInt64).
				WithExternalField("id"),
		).
		WithField(
			entity.NewField().
				WithName("value").
				WithDataType(entity.FieldTypeFloat).
				WithExternalField("value"),
		).
		WithField(
			entity.NewField().
				WithName("embedding").
				WithDataType(entity.FieldTypeFloatVector).
				WithDim(testVecDim).
				WithExternalField("embedding"),
		)

	err = mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
	common.CheckErr(t, err, true)
	t.Logf("Created external collection: %s", collName)

	t.Cleanup(func() {
		_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(collName))
	})

	// Verify collection exists with correct external source
	coll, err := mc.DescribeCollection(ctx, client.NewDescribeCollectionOption(collName))
	common.CheckErr(t, err, true)
	require.Equal(t, extTestURI(minioCfg, extPath), coll.Schema.ExternalSource)

	// ---------------------------------------------------------------
	// Step 3: Trigger refresh
	// ---------------------------------------------------------------
	refreshResult, err := mc.RefreshExternalCollection(ctx,
		client.NewRefreshExternalCollectionOption(collName))
	common.CheckErr(t, err, true)
	require.Greater(t, refreshResult.JobID, int64(0))
	t.Logf("Started refresh job: %d", refreshResult.JobID)

	// ---------------------------------------------------------------
	// Step 4: Poll until refresh completes (timeout 120s)
	// ---------------------------------------------------------------
	jobID := refreshResult.JobID
	deadline := time.After(120 * time.Second)
	ticker := time.NewTicker(2 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-deadline:
			t.Fatalf("Refresh job %d timed out after 120s", jobID)
		case <-ticker.C:
			progress, err := mc.GetRefreshExternalCollectionProgress(ctx,
				client.NewGetRefreshExternalCollectionProgressOption(jobID))
			require.NoError(t, err)
			t.Logf("Job %d: state=%s progress=%d%%", jobID, progress.State, progress.Progress)

			if progress.State == entity.RefreshStateCompleted {
				goto verify
			}
			if progress.State == entity.RefreshStateFailed {
				t.Fatalf("Refresh job %d failed: %s", jobID, progress.Reason)
			}
		}
	}
verify:

	// ---------------------------------------------------------------
	// Step 5: Verify row count via collection statistics
	// ---------------------------------------------------------------
	// NOTE: GetPersistentSegmentInfo uses channel-based segment lookup which does
	// not work for external collections (external segments lack InsertChannel).
	// GetCollectionStats uses collection-ID-based lookup (coll2Segments index)
	// which correctly includes external segments.
	stats, err := mc.GetCollectionStats(ctx, client.NewGetCollectionStatsOption(collName))
	common.CheckErr(t, err, true)

	rowCountStr, ok := stats["row_count"]
	require.True(t, ok, "collection stats should contain row_count")

	rowCount, err := strconv.ParseInt(rowCountStr, 10, 64)
	require.NoError(t, err, "parse row_count")
	require.Equal(t, totalExpectedRows, rowCount,
		"total row count should match uploaded parquet data")

	t.Logf("Verified: collection stats row_count=%d (expected %d)", rowCount, totalExpectedRows)

	// ---------------------------------------------------------------
	// Step 6: Verify via ListRefreshExternalCollectionJobs
	// ---------------------------------------------------------------
	jobs, err := mc.ListRefreshExternalCollectionJobs(ctx,
		client.NewListRefreshExternalCollectionJobsOption(collName))
	common.CheckErr(t, err, true)
	require.Greater(t, len(jobs), 0, "should have at least one refresh job")

	found := false
	for _, job := range jobs {
		if job.JobID == jobID {
			require.Equal(t, entity.RefreshStateCompleted, job.State)
			found = true
			break
		}
	}
	require.True(t, found, "should find the completed refresh job in job list")

	// ---------------------------------------------------------------
	// Step 7: Regression for issue #48626 — explore temp dir must be
	// reclaimed after the refresh job reaches a terminal state, not
	// left as orphaned 0-byte placeholders on object storage.
	// The datacoord Finished handler fires cleanupExploreTempForJob
	// via defer, so by the time GetRefreshExternalCollectionProgress
	// reports Completed + schema updated, the prefix must be gone.
	// ---------------------------------------------------------------
	exploreTempPrefix := fmt.Sprintf("__explore_temp__/coord_%d/", jobID)
	var leaked []string
	for obj := range minioClient.ListObjects(ctx, minioCfg.bucket, miniogo.ListObjectsOptions{
		Prefix:    exploreTempPrefix,
		Recursive: true,
	}) {
		if obj.Err != nil {
			continue
		}
		leaked = append(leaked, obj.Key)
	}
	require.Empty(t, leaked,
		"explore temp prefix %q should be fully reclaimed after refresh completes (issue #48626)",
		exploreTempPrefix)
	t.Logf("Verified issue #48626: explore temp prefix %q is clean", exploreTempPrefix)
}

// TestExternalCollectionParquetCompressionCodecs is the regression test for
// issue #48869. Part 8 added `arrow:with_snappy=True` and `arrow:with_lz4=True`
// to the conan recipe; if either gets dropped or the bundled Arrow is rebuilt
// without codec support, external collections backed by compressed parquet
// silently break at query time. This test generates a tiny parquet file with
// each codec via pyarrow (Go arrow can't write LZ4 in a C++-readable format)
// and verifies refresh + row count match.
func TestExternalCollectionParquetCompressionCodecs(t *testing.T) {
	ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	minioCfg := getMinIOConfig()
	minioClient, err := newMinIOClient(minioCfg)
	require.NoError(t, err)
	exists, err := minioClient.BucketExists(ctx, minioCfg.bucket)
	if err != nil || !exists {
		t.Skipf("MinIO bucket %q not accessible, skipping", minioCfg.bucket)
	}

	const numRows = 256

	for _, codec := range []string{"snappy", "lz4"} {
		codec := codec
		t.Run(codec, func(t *testing.T) {
			data, err := generateParquetBytesWithCompression(externalDataSchemaBasic, numRows, 0, testVecDim, codec)
			require.NoError(t, err, "generate %s-compressed parquet", codec)

			collName := common.GenRandomString("ext_codec_"+codec, 6)
			extPath := fmt.Sprintf("external-e2e-test/%s", collName)
			objectKey := fmt.Sprintf("%s/data.parquet", extPath)

			_, err = minioClient.PutObject(ctx, minioCfg.bucket, objectKey,
				bytes.NewReader(data), int64(len(data)),
				miniogo.PutObjectOptions{ContentType: "application/octet-stream"})
			require.NoError(t, err, "upload %s-compressed parquet", codec)
			t.Cleanup(func() {
				cleanupMinIOPrefix(context.Background(), minioClient, minioCfg.bucket, extPath+"/")
			})

			schema := entity.NewSchema().
				WithName(collName).
				WithExternalSource(extTestURI(minioCfg, extPath)).
				WithExternalSpec(extTestSpec(minioCfg, "parquet")).
				WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithExternalField("id")).
				WithField(entity.NewField().WithName("value").WithDataType(entity.FieldTypeFloat).WithExternalField("value")).
				WithField(entity.NewField().WithName("embedding").WithDataType(entity.FieldTypeFloatVector).WithDim(testVecDim).WithExternalField("embedding"))

			err = mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
			common.CheckErr(t, err, true)
			t.Cleanup(func() {
				_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(collName))
			})

			refreshResult, err := mc.RefreshExternalCollection(ctx,
				client.NewRefreshExternalCollectionOption(collName))
			common.CheckErr(t, err, true)
			jobID := refreshResult.JobID

			deadline := time.After(120 * time.Second)
			ticker := time.NewTicker(2 * time.Second)
			defer ticker.Stop()
			for loop := true; loop; {
				select {
				case <-deadline:
					t.Fatalf("refresh for %s codec timed out", codec)
				case <-ticker.C:
					p, err := mc.GetRefreshExternalCollectionProgress(ctx,
						client.NewGetRefreshExternalCollectionProgressOption(jobID))
					require.NoError(t, err)
					if p.State == entity.RefreshStateCompleted {
						loop = false
					}
					if p.State == entity.RefreshStateFailed {
						t.Fatalf("refresh for %s codec failed: %s", codec, p.Reason)
					}
				}
			}

			stats, err := mc.GetCollectionStats(ctx, client.NewGetCollectionStatsOption(collName))
			common.CheckErr(t, err, true)
			rowCountStr, ok := stats["row_count"]
			require.True(t, ok)
			rowCount, err := strconv.ParseInt(rowCountStr, 10, 64)
			require.NoError(t, err)
			require.Equal(t, int64(numRows), rowCount,
				"%s-compressed parquet row count mismatch — codec likely not linked into Arrow", codec)
			t.Logf("Verified issue #48869 codec=%s: refresh + row count match", codec)
		})
	}
}

// TestExternalCollectionLoadAndQuery tests the full external collection lifecycle:
//  1. Upload parquet test data to MinIO
//  2. Create an external collection
//  3. Refresh and wait for completion
//  4. Create index on vector field
//  5. Load collection
//  6. Query count(*)
//  7. Query with filter and output fields
//  8. Search with vector
func TestExternalCollectionLoadAndQuery(t *testing.T) {
	t.Parallel()

	ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	// Connect to MinIO (skip if unavailable)
	minioCfg := getMinIOConfig()
	minioClient, err := newMinIOClient(minioCfg)
	require.NoError(t, err)

	exists, err := minioClient.BucketExists(ctx, minioCfg.bucket)
	if err != nil || !exists {
		t.Skipf("MinIO bucket %q not accessible (exists=%v, err=%v), skipping",
			minioCfg.bucket, exists, err)
	}

	collName := common.GenRandomString("ext_load_query", 6)
	extPath := fmt.Sprintf("external-e2e-test/%s", collName)

	// ---------------------------------------------------------------
	// Step 1: Generate and upload parquet data to MinIO
	// ---------------------------------------------------------------
	const numFiles = 2
	const rowsPerFile = int64(1000)
	totalExpectedRows := int64(numFiles) * rowsPerFile

	for i := 0; i < numFiles; i++ {
		data, err := generateParquetBytes(externalDataSchemaBasic, rowsPerFile, int64(i)*rowsPerFile, testVecDim)
		require.NoError(t, err, "generate parquet file %d", i)

		objectKey := fmt.Sprintf("%s/data%d.parquet", extPath, i)
		_, err = minioClient.PutObject(ctx, minioCfg.bucket, objectKey,
			bytes.NewReader(data), int64(len(data)),
			miniogo.PutObjectOptions{ContentType: "application/octet-stream"})
		require.NoError(t, err, "upload parquet file %d to MinIO", i)
		t.Logf("Uploaded %s/%s (%d bytes, %d rows)", minioCfg.bucket, objectKey, len(data), rowsPerFile)
	}

	t.Cleanup(func() {
		cleanupMinIOPrefix(context.Background(), minioClient, minioCfg.bucket,
			fmt.Sprintf("%s/", extPath))
	})

	// ---------------------------------------------------------------
	// Step 2: Create external collection
	// ---------------------------------------------------------------
	schema := entity.NewSchema().
		WithName(collName).
		WithExternalSource(extTestURI(minioCfg, extPath)).
		WithExternalSpec(extTestSpec(minioCfg, "parquet")).
		WithField(
			entity.NewField().
				WithName("id").
				WithDataType(entity.FieldTypeInt64).
				WithExternalField("id"),
		).
		WithField(
			entity.NewField().
				WithName("value").
				WithDataType(entity.FieldTypeFloat).
				WithExternalField("value"),
		).
		WithField(
			entity.NewField().
				WithName("embedding").
				WithDataType(entity.FieldTypeFloatVector).
				WithDim(testVecDim).
				WithExternalField("embedding"),
		)

	err = mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
	common.CheckErr(t, err, true)
	t.Logf("Created external collection: %s", collName)

	t.Cleanup(func() {
		_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(collName))
	})

	// Verify collection exists with correct external source
	coll, err := mc.DescribeCollection(ctx, client.NewDescribeCollectionOption(collName))
	common.CheckErr(t, err, true)
	require.Equal(t, extTestURI(minioCfg, extPath), coll.Schema.ExternalSource)

	// ---------------------------------------------------------------
	// Step 3: Trigger refresh and poll until complete
	// ---------------------------------------------------------------
	refreshResult, err := mc.RefreshExternalCollection(ctx,
		client.NewRefreshExternalCollectionOption(collName))
	common.CheckErr(t, err, true)
	require.Greater(t, refreshResult.JobID, int64(0))
	t.Logf("Started refresh job: %d", refreshResult.JobID)

	jobID := refreshResult.JobID
	deadline := time.After(120 * time.Second)
	ticker := time.NewTicker(2 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-deadline:
			t.Fatalf("Refresh job %d timed out after 120s", jobID)
		case <-ticker.C:
			progress, err := mc.GetRefreshExternalCollectionProgress(ctx,
				client.NewGetRefreshExternalCollectionProgressOption(jobID))
			require.NoError(t, err)
			t.Logf("Job %d: state=%s progress=%d%%", jobID, progress.State, progress.Progress)

			if progress.State == entity.RefreshStateCompleted {
				goto indexAndLoad
			}
			if progress.State == entity.RefreshStateFailed {
				t.Fatalf("Refresh job %d failed: %s", jobID, progress.Reason)
			}
		}
	}
indexAndLoad:

	// ---------------------------------------------------------------
	// Step 4: Create vector index and load collection
	// (skip scalar indexes to avoid ARM64 Tantivy/HNSW crashes)
	// ---------------------------------------------------------------
	indexAndLoadCollection(ctx, t, mc, collName, "embedding")

	// ---------------------------------------------------------------
	// Step 6: Query count(*)
	// ---------------------------------------------------------------
	countRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithOutputFields(common.QueryCountFieldName))
	common.CheckErr(t, err, true)

	count, err := countRes.GetColumn(common.QueryCountFieldName).GetAsInt64(0)
	require.NoError(t, err)
	require.Equal(t, totalExpectedRows, count,
		"count(*) should match total uploaded rows")
	t.Logf("Query count(*) = %d (expected %d)", count, totalExpectedRows)

	// ---------------------------------------------------------------
	// Step 7: Query with filter and output fields
	// ---------------------------------------------------------------
	filterRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("id < 100").
		WithOutputFields("id", "value"))
	common.CheckErr(t, err, true)

	idCol := filterRes.GetColumn("id")
	require.Equal(t, 100, idCol.Len(),
		"filter id < 100 should return exactly 100 rows")
	t.Logf("Query with filter returned %d rows", idCol.Len())

	// ---------------------------------------------------------------
	// Step 8: Search with vector
	// ---------------------------------------------------------------
	vec := make([]float32, testVecDim)
	for i := range vec {
		vec[i] = float32(i) * 0.1
	}
	searchRes, err := mc.Search(ctx, client.NewSearchOption(collName, 5,
		[]entity.Vector{entity.FloatVector(vec)}).
		WithConsistencyLevel(entity.ClStrong).
		WithANNSField("embedding").
		WithOutputFields("id", "value"))
	common.CheckErr(t, err, true)

	require.Equal(t, 1, len(searchRes), "should have 1 result set for 1 query vector")
	require.Greater(t, searchRes[0].ResultCount, 0, "search should return results")
	t.Logf("Search returned %d results", searchRes[0].ResultCount)

	// ---------------------------------------------------------------
	// Step 9: Search with filter (hybrid search)
	// ---------------------------------------------------------------
	hybridRes, err := mc.Search(ctx, client.NewSearchOption(collName, 5,
		[]entity.Vector{entity.FloatVector(vec)}).
		WithConsistencyLevel(entity.ClStrong).
		WithANNSField("embedding").
		WithFilter("id >= 500 && id < 1000").
		WithOutputFields("id", "value"))
	common.CheckErr(t, err, true)

	require.Equal(t, 1, len(hybridRes))
	require.Greater(t, hybridRes[0].ResultCount, 0, "hybrid search should return results")
	// Verify all returned IDs satisfy the filter
	hybridIDCol := hybridRes[0].GetColumn("id")
	for i := 0; i < hybridIDCol.Len(); i++ {
		val, _ := hybridIDCol.GetAsInt64(i)
		require.GreaterOrEqual(t, val, int64(500), "filtered search result id should be >= 500")
		require.Less(t, val, int64(1000), "filtered search result id should be < 1000")
	}
	t.Logf("Hybrid search (filter id in [500,1000)) returned %d results", hybridRes[0].ResultCount)

	// ---------------------------------------------------------------
	// Step 10: Query with pagination (offset + limit)
	// ---------------------------------------------------------------
	pageRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("id >= 0").
		WithOutputFields("id").
		WithOffset(10).
		WithLimit(20))
	common.CheckErr(t, err, true)

	pageIDCol := pageRes.GetColumn("id")
	require.Equal(t, 20, pageIDCol.Len(), "pagination should return exactly 20 rows")
	firstID, _ := pageIDCol.GetAsInt64(0)
	require.Equal(t, int64(10), firstID, "first row after offset=10 should have id=10")
	t.Logf("Pagination query (offset=10, limit=20) returned %d rows, first id=%d", pageIDCol.Len(), firstID)

	// ---------------------------------------------------------------
	// Step 11: Search with multiple query vectors
	// ---------------------------------------------------------------
	vec2 := make([]float32, testVecDim)
	for i := range vec2 {
		vec2[i] = float32(999-i) * 0.1
	}
	multiRes, err := mc.Search(ctx, client.NewSearchOption(collName, 3,
		[]entity.Vector{entity.FloatVector(vec), entity.FloatVector(vec2)}).
		WithConsistencyLevel(entity.ClStrong).
		WithANNSField("embedding").
		WithOutputFields("id"))
	common.CheckErr(t, err, true)

	require.Equal(t, 2, len(multiRes), "should have 2 result sets for 2 query vectors")
	require.Greater(t, multiRes[0].ResultCount, 0, "first query should return results")
	require.Greater(t, multiRes[1].ResultCount, 0, "second query should return results")
	t.Logf("Multi-vector search returned %d and %d results", multiRes[0].ResultCount, multiRes[1].ResultCount)

	// ---------------------------------------------------------------
	// Step 12: Query with output embedding field (vector retrieval)
	// ---------------------------------------------------------------
	vecRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("id == 0").
		WithOutputFields("id", "embedding"))
	common.CheckErr(t, err, true)

	vecIDCol := vecRes.GetColumn("id")
	require.Equal(t, 1, vecIDCol.Len(), "should return exactly 1 row for id==0")
	embCol := vecRes.GetColumn("embedding")
	require.NotNil(t, embCol, "embedding column should be present in output")
	require.Equal(t, 1, embCol.Len(), "embedding column should have 1 row")
	t.Log("Query with embedding output field succeeded")
}

func TestExternalCollectionNullableFloatVectorTakeOutput(t *testing.T) {
	t.Parallel()

	ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	minioCfg := getMinIOConfig()
	minioClient, err := newMinIOClient(minioCfg)
	require.NoError(t, err)

	exists, err := minioClient.BucketExists(ctx, minioCfg.bucket)
	if err != nil || !exists {
		t.Skipf("MinIO bucket %q not accessible (exists=%v, err=%v), skipping",
			minioCfg.bucket, exists, err)
	}

	collName := common.GenRandomString("ext_nullable_vec", 6)
	extPath := fmt.Sprintf("external-e2e-test/%s", collName)

	data, err := generateParquetBytes(externalDataSchemaNullableVector, 3, 0, testVecDim)
	require.NoError(t, err, "generate nullable vector parquet")

	objectKey := fmt.Sprintf("%s/data.parquet", extPath)
	uploadParquetToMinIO(ctx, t, minioClient, minioCfg.bucket, objectKey, data)

	t.Cleanup(func() {
		cleanupMinIOPrefix(context.Background(), minioClient, minioCfg.bucket,
			fmt.Sprintf("%s/", extPath))
	})

	schema := entity.NewSchema().
		WithName(collName).
		WithExternalSource(extTestURI(minioCfg, extPath)).
		WithExternalSpec(extTestSpec(minioCfg, "parquet")).
		WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithExternalField("id")).
		WithField(entity.NewField().WithName("embedding").WithDataType(entity.FieldTypeFloatVector).
			WithDim(testVecDim).WithNullable(true).WithExternalField("embedding"))

	err = mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
	common.CheckErr(t, err, true)
	t.Cleanup(func() {
		_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(collName))
	})

	refreshAndWait(ctx, t, mc, collName)
	indexAndLoadCollection(ctx, t, mc, collName, "embedding")

	queryRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("id >= 0").
		WithOutputFields("id", "embedding"))
	common.CheckErr(t, err, true)
	require.Equal(t, 3, queryRes.ResultCount)

	idCol := queryRes.GetColumn("id")
	embeddingCol := queryRes.GetColumn("embedding")
	require.NotNil(t, idCol)
	require.NotNil(t, embeddingCol)

	type vectorRow struct {
		isNull bool
		vector []float32
	}
	rows := make(map[int64]vectorRow, queryRes.ResultCount)
	for i := 0; i < queryRes.ResultCount; i++ {
		id, err := idCol.GetAsInt64(i)
		require.NoError(t, err)
		isNull, err := embeddingCol.IsNull(i)
		require.NoError(t, err)

		row := vectorRow{isNull: isNull}
		if !isNull {
			raw, err := embeddingCol.Get(i)
			require.NoError(t, err)
			vec, ok := raw.(entity.FloatVector)
			require.Truef(t, ok, "embedding row %d has unexpected type %T", i, raw)
			row.vector = []float32(vec)
		}
		rows[id] = row
	}

	require.Len(t, rows, 3)

	assertVector := func(id int64, expected []float32) {
		t.Helper()
		row, ok := rows[id]
		require.True(t, ok, "missing row id=%d", id)
		require.False(t, row.isNull, "embedding should not be null for id=%d", id)
		require.Equal(t, len(expected), len(row.vector), "embedding dim mismatch for id=%d", id)
		for i := range expected {
			require.InDelta(t, expected[i], row.vector[i], 1e-6,
				"embedding[%d] mismatch for id=%d", i, id)
		}
	}

	assertVector(0, []float32{0, 1, 2, 3})
	require.True(t, rows[1].isNull, "embedding should be null for id=1")
	assertVector(2, []float32{8, 9, 10, 11})
}

func TestExternalCollectionAdditionalVectorTakeOutput(t *testing.T) {
	ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	minioCfg := getMinIOConfig()
	minioClient, err := newMinIOClient(minioCfg)
	require.NoError(t, err)

	exists, err := minioClient.BucketExists(ctx, minioCfg.bucket)
	if err != nil || !exists {
		t.Skipf("MinIO bucket %q not accessible (exists=%v, err=%v), skipping",
			minioCfg.bucket, exists, err)
	}

	collName := common.GenRandomString("ext_take_vec", 6)
	extPath := fmt.Sprintf("external-e2e-test/%s", collName)

	const numRows = int64(8)
	data, err := generateExternalTakeVectorParquetBytes(numRows)
	require.NoError(t, err, "generate vector take parquet")

	objectKey := fmt.Sprintf("%s/data.parquet", extPath)
	uploadParquetToMinIO(ctx, t, minioClient, minioCfg.bucket, objectKey, data)

	t.Cleanup(func() {
		cleanupMinIOPrefix(context.Background(), minioClient, minioCfg.bucket,
			fmt.Sprintf("%s/", extPath))
		_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(collName))
	})

	schema := entity.NewSchema().
		WithName(collName).
		WithExternalSource(extTestURI(minioCfg, extPath)).
		WithExternalSpec(extTestSpec(minioCfg, "parquet")).
		WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithExternalField("id")).
		WithField(entity.NewField().WithName("bin_vec").WithDataType(entity.FieldTypeBinaryVector).
			WithDim(16).WithExternalField("bin_vec")).
		WithField(entity.NewField().WithName("fp16_vec").WithDataType(entity.FieldTypeFloat16Vector).
			WithDim(testVecDim).WithExternalField("fp16_vec")).
		WithField(entity.NewField().WithName("bf16_vec").WithDataType(entity.FieldTypeBFloat16Vector).
			WithDim(testVecDim).WithExternalField("bf16_vec")).
		WithField(entity.NewField().WithName("int8_vec").WithDataType(entity.FieldTypeInt8Vector).
			WithDim(testVecDim).WithExternalField("int8_vec"))

	err = mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
	common.CheckErr(t, err, true)

	refreshAndWait(ctx, t, mc, collName)

	indexes := map[string]index.Index{
		"bin_vec":  index.NewBinFlatIndex(entity.HAMMING),
		"fp16_vec": index.NewFlatIndex(entity.L2),
		"bf16_vec": index.NewFlatIndex(entity.L2),
		"int8_vec": index.NewGenericIndex("int8_hnsw", map[string]string{
			index.MetricTypeKey: string(entity.COSINE),
			index.IndexTypeKey:  "HNSW",
			"M":                 "8",
			"efConstruction":    "64",
		}),
	}
	for fieldName, idx := range indexes {
		idxTask, idxErr := mc.CreateIndex(ctx, client.NewCreateIndexOption(collName, fieldName, idx))
		common.CheckErr(t, idxErr, true)
		require.NoError(t, idxTask.Await(ctx), "create index on %s", fieldName)
	}

	loadTask, loadErr := mc.LoadCollection(ctx, client.NewLoadCollectionOption(collName))
	common.CheckErr(t, loadErr, true)
	require.NoError(t, loadTask.Await(ctx))

	assertVectorRows := func(result client.ResultSet, context string) {
		t.Helper()
		idCol := result.GetColumn("id")
		require.NotNil(t, idCol, "%s id column", context)
		for _, fieldName := range []string{"bin_vec", "fp16_vec", "bf16_vec", "int8_vec"} {
			require.NotNil(t, result.GetColumn(fieldName), "%s %s column", context, fieldName)
		}
		for i := 0; i < result.ResultCount; i++ {
			id, err := idCol.GetAsInt64(i)
			require.NoError(t, err)

			rawBin, err := result.GetColumn("bin_vec").Get(i)
			require.NoError(t, err)
			require.Equal(t, externalTakeDenseBytes(id, 2, 11), []byte(rawBin.(entity.BinaryVector)),
				"%s bin_vec row id=%d", context, id)

			rawFP16, err := result.GetColumn("fp16_vec").Get(i)
			require.NoError(t, err)
			require.Equal(t, externalTakeDenseBytes(id, testVecDim*2, 31), []byte(rawFP16.(entity.Float16Vector)),
				"%s fp16_vec row id=%d", context, id)

			rawBF16, err := result.GetColumn("bf16_vec").Get(i)
			require.NoError(t, err)
			require.Equal(t, externalTakeDenseBytes(id, testVecDim*2, 51), []byte(rawBF16.(entity.BFloat16Vector)),
				"%s bf16_vec row id=%d", context, id)

			rawInt8, err := result.GetColumn("int8_vec").Get(i)
			require.NoError(t, err)
			require.Equal(t, externalTakeInt8Vector(id), []int8(rawInt8.(entity.Int8Vector)),
				"%s int8_vec row id=%d", context, id)
		}
	}

	queryRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("id in [1, 3, 5]").
		WithOutputFields("id", "bin_vec", "fp16_vec", "bf16_vec", "int8_vec"))
	common.CheckErr(t, err, true)
	require.Equal(t, 3, queryRes.ResultCount)
	assertVectorRows(queryRes, "query output")

	searchRes, err := mc.Search(ctx, client.NewSearchOption(collName, 3,
		[]entity.Vector{entity.Int8Vector(externalTakeInt8Vector(3))}).
		WithConsistencyLevel(entity.ClStrong).
		WithANNSField("int8_vec").
		WithOutputFields("id", "bin_vec", "fp16_vec", "bf16_vec", "int8_vec"))
	common.CheckErr(t, err, true)
	require.Equal(t, 1, len(searchRes))
	require.Greater(t, searchRes[0].ResultCount, 0)
	assertVectorRows(searchRes[0], "int8 search output")
}

func TestExternalCollectionSparseVectorCurrentlyRejected(t *testing.T) {
	ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	minioCfg := getMinIOConfig()
	collName := common.GenRandomString("ext_sparse_reject", 6)
	extPath := fmt.Sprintf("external-e2e-test/%s", collName)

	schema := entity.NewSchema().
		WithName(collName).
		WithExternalSource(extTestURI(minioCfg, extPath)).
		WithExternalSpec(extTestSpec(minioCfg, "parquet")).
		WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithExternalField("id")).
		WithField(entity.NewField().WithName("sparse_vec").WithDataType(entity.FieldTypeSparseVector).
			WithExternalField("sparse_vec"))

	err := mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
	require.Error(t, err)
	require.Contains(t, err.Error(), "does not support field type SparseFloatVector")
}

// TestExternalCollectionIncrementalRefresh tests that refreshing an external collection
// after modifying the underlying parquet files correctly updates segments:
//  1. Initial files: data0 (ids 0-499) + data1 (ids 500-999) → 1000 rows
//  2. Remove data1, add data2 (ids 2000-2299) → 800 rows
//  3. Verify: data0 intact, data1 gone, data2 present
func TestExternalCollectionIncrementalRefresh(t *testing.T) {
	t.Parallel()

	ctx := hp.CreateContext(t, time.Second*600) // 10 min for two refresh+load cycles
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	minioCfg := getMinIOConfig()
	minioClient, err := newMinIOClient(minioCfg)
	require.NoError(t, err)

	exists, err := minioClient.BucketExists(ctx, minioCfg.bucket)
	if err != nil || !exists {
		t.Skipf("MinIO bucket %q not accessible, skipping", minioCfg.bucket)
	}

	collName := common.GenRandomString("ext_incr_refresh", 6)
	extPath := fmt.Sprintf("external-e2e-test/%s", collName)

	t.Cleanup(func() {
		cleanupMinIOPrefix(context.Background(), minioClient, minioCfg.bucket, extPath+"/")
		_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(collName))
	})

	// ---------------------------------------------------------------
	// Phase 1: Upload initial data files
	// ---------------------------------------------------------------
	data0, err := generateParquetBytes(externalDataSchemaBasic, 500, 0, testVecDim) // ids 0-499
	require.NoError(t, err)
	data1, err := generateParquetBytes(externalDataSchemaBasic, 500, 500, testVecDim) // ids 500-999
	require.NoError(t, err)

	uploadParquetToMinIO(ctx, t, minioClient, minioCfg.bucket, extPath+"/data0.parquet", data0)
	uploadParquetToMinIO(ctx, t, minioClient, minioCfg.bucket, extPath+"/data1.parquet", data1)

	// ---------------------------------------------------------------
	// Create external collection
	// ---------------------------------------------------------------
	schema := entity.NewSchema().
		WithName(collName).
		WithExternalSource(extTestURI(minioCfg, extPath)).
		WithExternalSpec(extTestSpec(minioCfg, "parquet")).
		WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithExternalField("id")).
		WithField(entity.NewField().WithName("value").WithDataType(entity.FieldTypeFloat).WithExternalField("value")).
		WithField(entity.NewField().WithName("embedding").WithDataType(entity.FieldTypeFloatVector).WithDim(testVecDim).WithExternalField("embedding"))

	err = mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
	common.CheckErr(t, err, true)
	t.Logf("Created external collection: %s", collName)

	// ---------------------------------------------------------------
	// First refresh + load
	// ---------------------------------------------------------------
	refreshAndWait(ctx, t, mc, collName)
	indexAndLoadCollection(ctx, t, mc, collName, "embedding")

	// Verify: count = 1000
	countRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithOutputFields(common.QueryCountFieldName))
	common.CheckErr(t, err, true)
	count, _ := countRes.GetColumn(common.QueryCountFieldName).GetAsInt64(0)
	require.Equal(t, int64(1000), count, "initial count should be 1000")
	t.Logf("Phase 1: count(*) = %d", count)

	// Verify: data1 range (ids 500-999) is present
	res1, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("id >= 500 && id < 1000").
		WithOutputFields("id"))
	common.CheckErr(t, err, true)
	require.Equal(t, 500, res1.GetColumn("id").Len(), "should find 500 rows in data1 range")

	// ---------------------------------------------------------------
	// Phase 2: Release, modify files, refresh again
	// ---------------------------------------------------------------
	err = mc.ReleaseCollection(ctx, client.NewReleaseCollectionOption(collName))
	require.NoError(t, err, "release collection")
	t.Log("Collection released")

	// Remove data1.parquet (ids 500-999)
	err = minioClient.RemoveObject(ctx, minioCfg.bucket, extPath+"/data1.parquet", miniogo.RemoveObjectOptions{})
	require.NoError(t, err, "remove data1.parquet")
	t.Log("Removed data1.parquet from MinIO")

	// Upload data2.parquet (ids 2000-2299, 300 rows)
	data2, err := generateParquetBytes(externalDataSchemaBasic, 300, 2000, testVecDim)
	require.NoError(t, err)
	uploadParquetToMinIO(ctx, t, minioClient, minioCfg.bucket, extPath+"/data2.parquet", data2)

	// Second refresh
	refreshAndWait(ctx, t, mc, collName)

	// Verify collection stats: count = 500 + 300 = 800
	stats, err := mc.GetCollectionStats(ctx, client.NewGetCollectionStatsOption(collName))
	common.CheckErr(t, err, true)
	rowCountStr, ok := stats["row_count"]
	require.True(t, ok, "stats should contain row_count")
	rowCount, _ := strconv.ParseInt(rowCountStr, 10, 64)
	require.Equal(t, int64(800), rowCount, "after file change: 500 (data0) + 300 (data2) = 800")
	t.Logf("Phase 2: collection stats row_count = %d", rowCount)

	// ---------------------------------------------------------------
	// Phase 3: Reload and verify data correctness
	// ---------------------------------------------------------------
	indexAndLoadCollection(ctx, t, mc, collName, "embedding")

	// Verify total count = 800
	countRes2, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithOutputFields(common.QueryCountFieldName))
	common.CheckErr(t, err, true)
	count2, _ := countRes2.GetColumn(common.QueryCountFieldName).GetAsInt64(0)
	require.Equal(t, int64(800), count2, "count should be 800 after file change")

	// Verify: data0 range (ids 0-499) still intact
	res0, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("id >= 0 && id < 500").
		WithOutputFields("id"))
	common.CheckErr(t, err, true)
	require.Equal(t, 500, res0.GetColumn("id").Len(), "data0 (ids 0-499) should be intact")

	// Verify: data1 range (ids 500-999) is gone
	res1Gone, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("id >= 500 && id < 1000").
		WithOutputFields("id"))
	common.CheckErr(t, err, true)
	data1Len := 0
	if col := res1Gone.GetColumn("id"); col != nil {
		data1Len = col.Len()
	}
	require.Equal(t, 0, data1Len, "data1 (ids 500-999) should be removed after refresh")

	// Verify: data2 range (ids 2000-2299) is present
	res2, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("id >= 2000 && id < 2300").
		WithOutputFields("id"))
	common.CheckErr(t, err, true)
	require.Equal(t, 300, res2.GetColumn("id").Len(), "data2 (ids 2000-2299) should be present")

	// Verify: search still works after incremental refresh
	vec := make([]float32, testVecDim)
	for i := range vec {
		vec[i] = float32(i) * 0.1
	}
	searchRes, err := mc.Search(ctx, client.NewSearchOption(collName, 5,
		[]entity.Vector{entity.FloatVector(vec)}).
		WithConsistencyLevel(entity.ClStrong).
		WithANNSField("embedding").
		WithOutputFields("id"))
	common.CheckErr(t, err, true)
	require.Greater(t, searchRes[0].ResultCount, 0, "search should return results after incremental refresh")

	t.Log("Phase 3: all verifications passed — incremental refresh works correctly")
}

// TestRefreshExternalCollectionAfterAddColumnReturnsCorrectData verifies that
// adding an external field to an already refreshed external collection patches
// existing same-fragment segments and returns correct values for the new field.
func TestRefreshExternalCollectionAfterAddColumnReturnsCorrectData(t *testing.T) {
	t.Parallel()

	ctx := hp.CreateContext(t, time.Second*600)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	minioCfg := getMinIOConfig()
	minioClient, err := newMinIOClient(minioCfg)
	require.NoError(t, err)
	skipIfMinIOUnreachable(ctx, t, minioClient, minioCfg.bucket)

	collName := common.GenRandomString("ext_add_col", 6)
	extPath := fmt.Sprintf("external-e2e-test/%s", collName)

	t.Cleanup(func() {
		cleanupMinIOPrefix(context.Background(), minioClient, minioCfg.bucket, extPath+"/")
		_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(collName))
	})

	const rowsPerFile = int64(25)
	totalExpectedRows := rowsPerFile * 2
	for _, file := range []struct {
		name    string
		startID int64
	}{
		{name: "data0.parquet", startID: 0},
		{name: "data1.parquet", startID: 100},
	} {
		data, genErr := generateParquetBytes(externalDataSchemaLarge, rowsPerFile, file.startID, testVecDim)
		require.NoError(t, genErr)
		uploadParquetToMinIO(ctx, t, minioClient, minioCfg.bucket, extPath+"/"+file.name, data)
	}

	// Create the collection without the parquet "score" column. The first
	// refresh creates external segments whose manifests and fake binlogs cover
	// only id, value, and embedding.
	schema := entity.NewSchema().
		WithName(collName).
		WithExternalSource(extTestURI(minioCfg, extPath)).
		WithExternalSpec(extTestSpec(minioCfg, "parquet")).
		WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithExternalField("id")).
		WithField(entity.NewField().WithName("value").WithDataType(entity.FieldTypeFloat).WithExternalField("value")).
		WithField(entity.NewField().WithName("embedding").WithDataType(entity.FieldTypeFloatVector).
			WithDim(testVecDim).WithExternalField("embedding"))

	err = mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
	common.CheckErr(t, err, true)

	refreshAndWait(ctx, t, mc, collName)
	stats, err := mc.GetCollectionStats(ctx, client.NewGetCollectionStatsOption(collName))
	common.CheckErr(t, err, true)
	rowCount, err := strconv.ParseInt(stats["row_count"], 10, 64)
	require.NoError(t, err)
	require.Equal(t, totalExpectedRows, rowCount)

	scoreField := entity.NewField().
		WithName("score").
		WithDataType(entity.FieldTypeDouble).
		WithNullable(true).
		WithExternalField("score")
	err = mc.AddCollectionField(ctx, client.NewAddCollectionFieldOption(collName, scoreField))
	common.CheckErr(t, err, true)

	described, err := mc.DescribeCollection(ctx, client.NewDescribeCollectionOption(collName))
	common.CheckErr(t, err, true)
	hasScore := false
	for _, field := range described.Schema.Fields {
		if field.Name == "score" {
			hasScore = true
			require.Equal(t, entity.FieldTypeDouble, field.DataType)
			require.Equal(t, "score", field.ExternalField)
			require.True(t, field.Nullable)
		}
	}
	require.True(t, hasScore, "DescribeCollection should include the added external score field")

	refreshAndWait(ctx, t, mc, collName)
	indexAndLoadCollection(ctx, t, mc, collName, "embedding")

	countRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithOutputFields(common.QueryCountFieldName))
	common.CheckErr(t, err, true)
	count, err := countRes.GetColumn(common.QueryCountFieldName).GetAsInt64(0)
	require.NoError(t, err)
	require.Equal(t, totalExpectedRows, count)

	lowIDRows, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("id < 5").
		WithOutputFields("id", "score"))
	common.CheckErr(t, err, true)
	assertExternalScoreRows(t, lowIDRows, map[int64]float64{
		0: 0.00,
		1: 0.01,
		2: 0.02,
		3: 0.03,
		4: 0.04,
	})

	scoreFilterRows, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("score >= 1.0 && score < 1.05").
		WithOutputFields("id", "score"))
	common.CheckErr(t, err, true)
	assertExternalScoreRows(t, scoreFilterRows, map[int64]float64{
		100: 1.00,
		101: 1.01,
		102: 1.02,
		103: 1.03,
		104: 1.04,
	})
}

func assertExternalScoreRows(t *testing.T, result client.ResultSet, expected map[int64]float64) {
	t.Helper()

	idCol := result.GetColumn("id")
	scoreCol := result.GetColumn("score")
	require.NotNil(t, idCol, "query result should contain id")
	require.NotNil(t, scoreCol, "query result should contain score")
	require.Equal(t, len(expected), idCol.Len())
	require.Equal(t, len(expected), scoreCol.Len())

	actual := make(map[int64]float64, idCol.Len())
	for i := 0; i < idCol.Len(); i++ {
		id, err := idCol.GetAsInt64(i)
		require.NoError(t, err)
		isNull, err := scoreCol.IsNull(i)
		require.NoError(t, err)
		require.False(t, isNull, "score should be non-null for id %d", id)
		score, err := scoreCol.GetAsDouble(i)
		require.NoError(t, err)
		actual[id] = score
	}

	require.Len(t, actual, len(expected))
	for id, expectedScore := range expected {
		score, ok := actual[id]
		require.True(t, ok, "missing row for id %d in result %v", id, actual)
		require.InDelta(t, expectedScore, score, 1e-9, "unexpected score for id %d", id)
	}
}

// TestExternalCollectionMultipleDataTypes tests external collections with various data types:
// Bool, Int8, Int16, Int32, Int64, Float, Double, VarChar, FloatVector
func TestExternalCollectionMultipleDataTypes(t *testing.T) {
	t.Parallel()

	ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	minioCfg := getMinIOConfig()
	minioClient, err := newMinIOClient(minioCfg)
	require.NoError(t, err)

	exists, err := minioClient.BucketExists(ctx, minioCfg.bucket)
	if err != nil || !exists {
		t.Skipf("MinIO bucket %q not accessible, skipping", minioCfg.bucket)
	}

	collName := common.GenRandomString("ext_multi_type", 6)
	extPath := fmt.Sprintf("external-e2e-test/%s", collName)

	t.Cleanup(func() {
		cleanupMinIOPrefix(context.Background(), minioClient, minioCfg.bucket, extPath+"/")
		_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(collName))
	})

	// ---------------------------------------------------------------
	// Step 1: Generate and upload multi-type parquet data
	// ---------------------------------------------------------------
	const numRows = int64(100)
	data, err := generateParquetBytes(externalDataSchemaMulti, numRows, 0, testVecDim)
	require.NoError(t, err)
	uploadParquetToMinIO(ctx, t, minioClient, minioCfg.bucket, extPath+"/data.parquet", data)

	// ---------------------------------------------------------------
	// Step 2: Create external collection with multi-type schema
	// ---------------------------------------------------------------
	schema := buildMultiTypeExternalSchema(collName,
		extTestURI(minioCfg, extPath),
		extTestSpec(minioCfg, "parquet"))

	err = mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
	common.CheckErr(t, err, true)
	t.Logf("Created multi-type external collection: %s", collName)

	runMultiTypeRefreshIndexLoadVerify(ctx, t, mc, collName, numRows)
}

// runMultiTypeRefreshIndexLoadVerify drives the post-create stages of the
// multi-type external-collection test (refresh -> index all vector fields ->
// load -> exhaustive query/search verification). Shared between the parquet
// and vortex multi-type tests so the same schema + verification plan covers
// both formats.  Assumes the dataset shape produced by
// generateParquetBytes(..., externalDataSchemaMulti, ...) /
// generate_vortex_data.py --schema multi with
// the given numRows.
func runMultiTypeRefreshIndexLoadVerify(ctx context.Context, t *testing.T,
	mc *base.MilvusClient, collName string, numRows int64,
) {
	runMultiTypeRefreshIndexLoadVerifyWithVectorFields(ctx, t, mc, collName, numRows, "", "",
		[]string{"embedding", "bin_vec", "fp16_vec", "bf16_vec", "int8_vec"})
}

// runMultiTypeRefreshIndexLoadVerifyWithSourceSpec accepts optional refresh
// source+spec (used by iceberg paths that re-supply extfs credentials on
// refresh; proxy now enforces both-or-neither).
func runMultiTypeRefreshIndexLoadVerifyWithSourceSpec(ctx context.Context, t *testing.T,
	mc *base.MilvusClient, collName string, numRows int64, refreshSource, refreshSpec string,
) {
	runMultiTypeRefreshIndexLoadVerifyWithVectorFields(ctx, t, mc, collName, numRows, refreshSource, refreshSpec,
		[]string{"embedding", "bin_vec", "fp16_vec", "bf16_vec", "int8_vec"})
}

func runMultiTypeRefreshIndexLoadVerifyWithVectorFields(ctx context.Context, t *testing.T,
	mc *base.MilvusClient, collName string, numRows int64, refreshSource, refreshSpec string, vectorFields []string,
) {
	// ---------------------------------------------------------------
	// Step 3: Refresh + index all vector fields + load
	// ---------------------------------------------------------------
	if refreshSource == "" && refreshSpec == "" {
		refreshAndWait(ctx, t, mc, collName)
	} else {
		opt := client.NewRefreshExternalCollectionOption(collName).
			WithExternalSource(refreshSource).
			WithExternalSpec(refreshSpec)
		refreshResult, err := mc.RefreshExternalCollection(ctx, opt)
		common.CheckErr(t, err, true)
		require.Greater(t, refreshResult.JobID, int64(0))
		waitForRefreshComplete(ctx, t, mc, refreshResult.JobID, 120*time.Second)
	}

	// Create index on all vector fields
	for _, vecField := range vectorFields {
		var metricType entity.MetricType
		switch vecField {
		case "bin_vec":
			metricType = entity.HAMMING
		default:
			metricType = entity.L2
		}
		idx := index.NewAutoIndex(metricType)
		idxTask, idxErr := mc.CreateIndex(ctx, client.NewCreateIndexOption(collName, vecField, idx))
		common.CheckErr(t, idxErr, true)
		require.NoError(t, idxTask.Await(ctx))
		t.Logf("Index created on %s (metric=%s)", vecField, metricType)
	}

	// Load collection
	loadTask, loadErr := mc.LoadCollection(ctx, client.NewLoadCollectionOption(collName))
	common.CheckErr(t, loadErr, true)
	require.NoError(t, loadTask.Await(ctx))
	t.Log("Collection loaded")

	// ---------------------------------------------------------------
	// Step 4: Verify count(*)
	// ---------------------------------------------------------------
	countRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithOutputFields(common.QueryCountFieldName))
	common.CheckErr(t, err, true)
	count, _ := countRes.GetColumn(common.QueryCountFieldName).GetAsInt64(0)
	require.Equal(t, numRows, count, "count should match uploaded rows")
	t.Logf("count(*) = %d", count)

	// ---------------------------------------------------------------
	// Step 5: Bool filter — bool_val == true → even ids → 50 rows
	// ---------------------------------------------------------------
	boolRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("bool_val == true").
		WithOutputFields("id", "bool_val"))
	common.CheckErr(t, err, true)
	require.Equal(t, int(numRows/2), boolRes.GetColumn("id").Len(),
		"bool_val==true should match even-id rows")
	t.Logf("Bool filter: %d rows", boolRes.GetColumn("id").Len())

	// ---------------------------------------------------------------
	// Step 6: Int8 filter — int8_val < 10 → ids 0-9 → 10 rows
	// ---------------------------------------------------------------
	int8Res, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("int8_val < 10").
		WithOutputFields("id", "int8_val"))
	common.CheckErr(t, err, true)
	require.Equal(t, 10, int8Res.GetColumn("id").Len(), "int8_val < 10 should be 10 rows")
	t.Logf("Int8 filter: %d rows", int8Res.GetColumn("id").Len())

	// ---------------------------------------------------------------
	// Step 7: Int16 filter — int16_val >= 500 → ids >= 50 → 50 rows
	// ---------------------------------------------------------------
	int16Res, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("int16_val >= 500").
		WithOutputFields("id", "int16_val"))
	common.CheckErr(t, err, true)
	require.Equal(t, 50, int16Res.GetColumn("id").Len(), "int16_val >= 500 should be 50 rows")
	t.Logf("Int16 filter: %d rows", int16Res.GetColumn("id").Len())

	// ---------------------------------------------------------------
	// Step 8: Int32 filter — int32_val < 5000 → ids < 50 → 50 rows
	// ---------------------------------------------------------------
	int32Res, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("int32_val < 5000").
		WithOutputFields("id", "int32_val"))
	common.CheckErr(t, err, true)
	require.Equal(t, 50, int32Res.GetColumn("id").Len(), "int32_val < 5000 should be 50 rows")
	t.Logf("Int32 filter: %d rows", int32Res.GetColumn("id").Len())

	// ---------------------------------------------------------------
	// Step 9: Float filter — float_val < 15.0 → id*1.5 < 15.0 → ids 0-9 → 10 rows
	// ---------------------------------------------------------------
	floatRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("float_val < 15.0").
		WithOutputFields("id", "float_val"))
	common.CheckErr(t, err, true)
	require.Equal(t, 10, floatRes.GetColumn("id").Len(), "float_val < 15.0 should be 10 rows")
	t.Logf("Float filter: %d rows", floatRes.GetColumn("id").Len())

	// ---------------------------------------------------------------
	// Step 10: Double filter — double_val < 0.5 → id*0.01 < 0.5 → ids 0-49 → 50 rows
	// ---------------------------------------------------------------
	doubleRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("double_val < 0.5").
		WithOutputFields("id", "double_val"))
	common.CheckErr(t, err, true)
	require.Equal(t, 50, doubleRes.GetColumn("id").Len(), "double_val < 0.5 should be 50 rows")
	t.Logf("Double filter: %d rows", doubleRes.GetColumn("id").Len())

	// ---------------------------------------------------------------
	// Step 10b: VarChar prefix filter — varchar_val like "str_004%" → ids 40-49 → 10 rows
	// ---------------------------------------------------------------
	varcharRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter(`varchar_val like "str_004%"`).
		WithOutputFields("id", "varchar_val"))
	common.CheckErr(t, err, true)
	require.Equal(t, 10, varcharRes.GetColumn("id").Len(), `varchar_val like "str_004%" should be 10 rows`)
	t.Logf("VarChar prefix filter: %d rows", varcharRes.GetColumn("id").Len())

	// ---------------------------------------------------------------
	// Step 10c: VarChar equality filter — varchar_val == "str_0042" → id=42 → 1 row
	// ---------------------------------------------------------------
	varcharEqRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter(`varchar_val == "str_0042"`).
		WithOutputFields("id", "varchar_val"))
	common.CheckErr(t, err, true)
	require.Equal(t, 1, varcharEqRes.GetColumn("id").Len(), `varchar_val == "str_0042" should be 1 row`)
	eqID, _ := varcharEqRes.GetColumn("id").GetAsInt64(0)
	require.Equal(t, int64(42), eqID, "matched row should be id=42")
	t.Logf("VarChar equality filter: matched id=%d", eqID)

	// ---------------------------------------------------------------
	// Step 11: Verify specific row values for id=42
	// ---------------------------------------------------------------
	row42, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("id == 42").
		WithOutputFields("id", "bool_val", "int8_val", "int16_val", "int32_val", "float_val", "double_val", "varchar_val", "json_val", "ts_val"))
	common.CheckErr(t, err, true)
	require.Equal(t, 1, row42.GetColumn("id").Len(), "should find exactly 1 row with id==42")

	boolVal, err := row42.GetColumn("bool_val").GetAsBool(0)
	require.NoError(t, err)
	require.Equal(t, true, boolVal, "id=42 (even) → bool_val should be true")

	int8Val, err := row42.GetColumn("int8_val").GetAsInt64(0)
	require.NoError(t, err)
	require.Equal(t, int64(42), int8Val, "int8_val for id=42 should be 42")

	int16Val, err := row42.GetColumn("int16_val").GetAsInt64(0)
	require.NoError(t, err)
	require.Equal(t, int64(420), int16Val, "int16_val for id=42 should be 420 (42*10)")

	int32Val, err := row42.GetColumn("int32_val").GetAsInt64(0)
	require.NoError(t, err)
	require.Equal(t, int64(4200), int32Val, "int32_val for id=42 should be 4200 (42*100)")

	floatVal, err := row42.GetColumn("float_val").GetAsDouble(0)
	require.NoError(t, err)
	require.InDelta(t, 63.0, floatVal, 0.1, "float_val for id=42 should be ~63.0 (42*1.5)")

	doubleVal, err := row42.GetColumn("double_val").GetAsDouble(0)
	require.NoError(t, err)
	require.InDelta(t, 0.42, doubleVal, 0.001, "double_val for id=42 should be ~0.42 (42*0.01)")

	varcharVal, err := row42.GetColumn("varchar_val").GetAsString(0)
	require.NoError(t, err)
	require.Equal(t, "str_0042", varcharVal, "varchar_val for id=42 should be 'str_0042'")

	jsonRaw, err := row42.GetColumn("json_val").Get(0)
	require.NoError(t, err)
	jsonVal := string(jsonRaw.([]byte))
	require.Contains(t, jsonVal, `"key":42`, "json_val for id=42 should contain key=42")
	require.Contains(t, jsonVal, `"name":"item_42"`, "json_val for id=42 should contain name=item_42")

	// Timestamptz: verify it's returned (as string representation)
	tsRaw, err := row42.GetColumn("ts_val").Get(0)
	require.NoError(t, err)
	require.NotNil(t, tsRaw, "ts_val for id=42 should not be nil")
	t.Logf("ts_val for id=42: %v (type=%T)", tsRaw, tsRaw)

	t.Log("All field values verified for id=42 (including JSON, Timestamptz)")

	// ---------------------------------------------------------------
	// Step 12: Search — verify nearest neighbor matches exact vector
	// ---------------------------------------------------------------
	vec := make([]float32, testVecDim)
	for i := range vec {
		vec[i] = float32(42)*0.1 + float32(i) // same formula as id=42's embedding
	}
	searchRes, err := mc.Search(ctx, client.NewSearchOption(collName, 5,
		[]entity.Vector{entity.FloatVector(vec)}).
		WithConsistencyLevel(entity.ClStrong).
		WithANNSField("embedding").
		WithOutputFields("id"))
	common.CheckErr(t, err, true)
	require.Equal(t, 1, len(searchRes))
	require.Greater(t, searchRes[0].ResultCount, 0, "search should return at least 1 result")
	nearestID, _ := searchRes[0].GetColumn("id").GetAsInt64(0)
	t.Logf("Search: nearest neighbor for id=42's vector is id=%d (top-5 returned %d results)", nearestID, searchRes[0].ResultCount)

	// ---------------------------------------------------------------
	// Step 12b: JSON filter — json_val["key"] < 10 → ids 0-9 → 10 rows
	// ---------------------------------------------------------------
	jsonRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter(`json_val["key"] < 10`).
		WithOutputFields("id", "json_val"))
	common.CheckErr(t, err, true)
	require.Equal(t, 10, jsonRes.GetColumn("id").Len(), `json_val["key"] < 10 should be 10 rows`)
	t.Logf("JSON filter: %d rows", jsonRes.GetColumn("id").Len())

	// ---------------------------------------------------------------
	// Step 12c: Geometry value check — query geo_val for id=42
	// ---------------------------------------------------------------
	geoRow, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("id == 42").
		WithOutputFields("id", "geo_val"))
	common.CheckErr(t, err, true)
	geoVal, err := geoRow.GetColumn("geo_val").GetAsString(0)
	require.NoError(t, err)
	require.Contains(t, geoVal, "42", "geo_val for id=42 should contain coordinate 42")
	t.Logf("Geometry value verified: %s", geoVal)

	// ---------------------------------------------------------------
	// Step 12e: Array filter — array_contains(array_int, 297) → id=99 (99*3=297) → 1 row
	// Use 297 because only id=99 has array [99, 198, 297]
	// ---------------------------------------------------------------
	arrayRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("array_contains(array_int, 297)").
		WithOutputFields("id"))
	common.CheckErr(t, err, true)
	require.Equal(t, 1, arrayRes.GetColumn("id").Len(), "array_contains(array_int, 297) should match id=99")
	arrayMatchID, _ := arrayRes.GetColumn("id").GetAsInt64(0)
	require.Equal(t, int64(99), arrayMatchID)
	t.Logf("Array filter: matched id=%d", arrayMatchID)

	// ---------------------------------------------------------------
	// Step 12f: Array as OUTPUT field — regression for issue #48619
	// Querying with array_int / array_str in output_fields used to crash
	// segcore (SIGSEGV in ArrayChunkWriter::calculate_size) because the
	// external chunk loader passed an arrow::ListArray where Milvus
	// expected a BinaryArray of serialized ScalarFieldProto bytes.
	// ---------------------------------------------------------------
	arrayIntOut, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("id == 42").
		WithOutputFields("id", "array_int"))
	common.CheckErr(t, err, true)
	require.Equal(t, 1, arrayIntOut.GetColumn("id").Len(), "should find id==42")
	t.Logf("Array(Int32) as output field returned %d row(s) without crash",
		arrayIntOut.GetColumn("id").Len())

	arrayStrOut, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("id == 42").
		WithOutputFields("id", "array_str"))
	common.CheckErr(t, err, true)
	require.Equal(t, 1, arrayStrOut.GetColumn("id").Len(), "should find id==42")
	t.Logf("Array(VarChar) as output field returned %d row(s) without crash",
		arrayStrOut.GetColumn("id").Len())

	// ---------------------------------------------------------------
	// Step 12g: Geometry as OUTPUT field across many rows — exercise
	// chunked-load path (single-row check at step 12c only loads one
	// chunk; querying many rows pulls more cells through the translator).
	// ---------------------------------------------------------------
	geoBulkRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("id < 50").
		WithOutputFields("id", "geo_val"))
	common.CheckErr(t, err, true)
	require.Equal(t, 50, geoBulkRes.GetColumn("id").Len(), "id<50 should be 50 rows")
	t.Logf("Geometry as output field across %d rows without crash",
		geoBulkRes.GetColumn("id").Len())

	// ---------------------------------------------------------------
	// Step 13: Compound filter across types — bool AND int range AND varchar
	// ---------------------------------------------------------------
	compoundRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter(`bool_val == true && int32_val < 5000 && varchar_val like "str_00%"`).
		WithOutputFields("id"))
	common.CheckErr(t, err, true)
	// ids 0-49 have int32_val < 5000, of which 25 are even (bool_val==true)
	// varchar_val like "str_00%" matches ids 0-99 (all have str_00xx format), so no extra filter
	require.Equal(t, 25, compoundRes.GetColumn("id").Len(),
		`compound filter: bool_val==true AND int32_val<5000 AND varchar like "str_00%" → 25 rows`)
	t.Log("Compound filter across types (incl. varchar) passed")
}

// --- Lance format helpers ---

// generateLanceDataOnMinIO uses the Python lance library to write a Lance dataset
// directly to MinIO/S3. This is necessary because lance format writing requires the
// Rust-based lance library, which is not accessible from Go directly.
func generateLanceDataOnMinIO(t *testing.T, s3URI, schema string, numRows, startID int) {
	t.Helper()
	minioCfg := getMinIOConfig()
	args := []string{
		"--schema", schema,
		s3URI,
		strconv.Itoa(numRows),
		"--start-id", strconv.Itoa(startID),
		"--vec-dim", strconv.Itoa(testVecDim),
	}
	if schema == externalDataSchemaMulti {
		args = append(args, "--bin-vec-dim", strconv.Itoa(testBinVecDim))
	}

	// Secrets go through the child process environment, not argv, so
	// `ps -ef` and CI logs don't expose them — see runPythonScript.
	out, err := runPythonScript(t, "python3", "generate_lance_data.py",
		map[string]string{
			"MINIO_ADDRESS":    minioCfg.address,
			"MINIO_ACCESS_KEY": minioCfg.accessKey,
			"MINIO_SECRET_KEY": minioCfg.secretKey,
		},
		args...)
	require.NoError(t, err, "generate lance data: %s", out)
	require.Contains(t, out, "OK", "lance data generation should succeed")
	t.Logf("Generated lance data: %s", out)
}

// runPythonScript runs a python script with the given positional arguments and
// extra environment variables. Secrets MUST be passed via extraEnv rather than
// embedded in args — argv is visible in `ps -ef`, `/proc/<pid>/cmdline`, and
// most CI log collectors, whereas envp is only readable by the process owner.
// This helper replaces an earlier runShellCommand wrapper that built a single
// `bash -c "KEY=val ... python3 ..."` string and leaked every caller's AKSK
// to the process command line.
func runPythonScript(t *testing.T, pythonBin, scriptPath string, extraEnv map[string]string, args ...string) (string, error) {
	t.Helper()
	cmd := exec.Command(pythonBin, append([]string{scriptPath}, args...)...) // #nosec G204
	cmd.Env = append(os.Environ(), mapToEnvSlice(extraEnv)...)
	output, err := cmd.CombinedOutput()
	return string(output), err
}

// mapToEnvSlice converts a string map into the "KEY=VALUE" slice form
// expected by exec.Cmd.Env.
func mapToEnvSlice(m map[string]string) []string {
	out := make([]string, 0, len(m))
	for k, v := range m {
		out = append(out, k+"="+v)
	}
	return out
}

// TestExternalCollectionLanceFormat tests external collections with lance-table format:
//  1. Generate Lance dataset on MinIO using Python
//  2. Create external collection with lance-table format
//  3. Refresh and wait for completion
//  4. Create index and load collection
//  5. Query count(*) and verify row count
//  6. Query with filter
//  7. Search with vector
func TestExternalCollectionLanceFormat(t *testing.T) {
	t.Parallel()

	ctx := hp.CreateContext(t, time.Second*600) // 10 min for lance operations
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	// Connect to MinIO
	minioCfg := getMinIOConfig()
	minioClient, err := newMinIOClient(minioCfg)
	require.NoError(t, err)

	exists, err := minioClient.BucketExists(ctx, minioCfg.bucket)
	if err != nil || !exists {
		t.Skipf("MinIO bucket %q not accessible, skipping", minioCfg.bucket)
	}

	collName := common.GenRandomString("ext_lance", 6)
	extPath := fmt.Sprintf("external-e2e-test/%s", collName)

	// ---------------------------------------------------------------
	// Step 1: Generate Lance dataset on MinIO
	// ---------------------------------------------------------------
	const totalRows = 100000
	s3URI := fmt.Sprintf("s3://%s/%s", minioCfg.bucket, extPath)
	generateLanceDataOnMinIO(t, s3URI, externalDataSchemaBasic, totalRows, 0)

	t.Cleanup(func() {
		cleanupMinIOPrefix(context.Background(), minioClient, minioCfg.bucket,
			fmt.Sprintf("%s/%s/", minioCfg.rootPath, extPath))
	})

	// ---------------------------------------------------------------
	// Step 2: Create external collection with lance-table format
	// ---------------------------------------------------------------
	schema := entity.NewSchema().
		WithName(collName).
		WithExternalSource(extTestURI(minioCfg, extPath)).
		WithExternalSpec(extTestSpec(minioCfg, "lance-table")).
		WithField(
			entity.NewField().
				WithName("id").
				WithDataType(entity.FieldTypeInt64).
				WithExternalField("id"),
		).
		WithField(
			entity.NewField().
				WithName("value").
				WithDataType(entity.FieldTypeFloat).
				WithExternalField("value"),
		).
		WithField(
			entity.NewField().
				WithName("embedding").
				WithDataType(entity.FieldTypeFloatVector).
				WithDim(testVecDim).
				WithExternalField("embedding"),
		)

	err = mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
	common.CheckErr(t, err, true)
	t.Logf("Created external collection with lance-table format: %s", collName)

	t.Cleanup(func() {
		_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(collName))
	})

	// Verify collection
	coll, err := mc.DescribeCollection(ctx, client.NewDescribeCollectionOption(collName))
	common.CheckErr(t, err, true)
	require.Equal(t, extTestURI(minioCfg, extPath), coll.Schema.ExternalSource)
	// Spec redacted at proxy edge.
	require.Contains(t, coll.Schema.ExternalSpec, `"format":"lance-table"`)
	require.Contains(t, coll.Schema.ExternalSpec, `"access_key_id":"***"`)

	// ---------------------------------------------------------------
	// Step 3: Refresh and wait for completion
	// ---------------------------------------------------------------
	refreshAndWait(ctx, t, mc, collName)

	// Verify row count via stats
	stats, err := mc.GetCollectionStats(ctx, client.NewGetCollectionStatsOption(collName))
	common.CheckErr(t, err, true)
	rowCountStr, ok := stats["row_count"]
	require.True(t, ok, "stats should contain row_count")
	rowCount, err := strconv.ParseInt(rowCountStr, 10, 64)
	require.NoError(t, err)
	require.Equal(t, int64(totalRows), rowCount,
		"collection stats row_count should match lance dataset")
	t.Logf("Refresh complete: row_count=%d", rowCount)

	// ---------------------------------------------------------------
	// Step 4: Create scalar + vector indexes and load collection
	// ---------------------------------------------------------------
	indexAndLoadCollection(ctx, t, mc, collName, "embedding")

	// ---------------------------------------------------------------
	// Step 5: Query count(*)
	// ---------------------------------------------------------------
	countRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithOutputFields(common.QueryCountFieldName))
	common.CheckErr(t, err, true)

	count, err := countRes.GetColumn(common.QueryCountFieldName).GetAsInt64(0)
	require.NoError(t, err)
	require.Equal(t, int64(totalRows), count,
		"count(*) should match lance dataset rows")
	t.Logf("Query count(*) = %d", count)

	// ---------------------------------------------------------------
	// Step 6: Query with filter
	// ---------------------------------------------------------------
	filterRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("id < 100").
		WithOutputFields("id", "value"))
	common.CheckErr(t, err, true)

	idCol := filterRes.GetColumn("id")
	require.Equal(t, 100, idCol.Len(),
		"filter id < 100 should return exactly 100 rows")
	t.Logf("Query with filter returned %d rows", idCol.Len())

	// Verify specific value
	val42, err := filterRes.GetColumn("value").GetAsDouble(42)
	require.NoError(t, err)
	expected42 := float64(42) * 1.5
	require.InDelta(t, expected42, val42, 0.01,
		"value for id=42 should be 42*1.5=63.0")

	// ---------------------------------------------------------------
	// Step 6b: Verify data correctness — sample rows across the range
	// ---------------------------------------------------------------
	sampleIDs := []int64{0, 1, 999, 50000, 99999}
	for _, sid := range sampleIDs {
		sampleRes, err := mc.Query(ctx, client.NewQueryOption(collName).
			WithConsistencyLevel(entity.ClStrong).
			WithFilter(fmt.Sprintf("id == %d", sid)).
			WithOutputFields("id", "value"))
		common.CheckErr(t, err, true)
		require.Equal(t, 1, sampleRes.GetColumn("id").Len(),
			"should find exactly 1 row for id=%d", sid)
		sampleVal, err := sampleRes.GetColumn("value").GetAsDouble(0)
		require.NoError(t, err)
		require.InDelta(t, float64(sid)*1.5, sampleVal, 0.01,
			"value for id=%d should be %f", sid, float64(sid)*1.5)
	}
	t.Logf("Verified %d sample rows across id range", len(sampleIDs))

	// ---------------------------------------------------------------
	// Step 6c: Verify large range query
	// ---------------------------------------------------------------
	largeRangeRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("id >= 50000 && id < 60000").
		WithOutputFields(common.QueryCountFieldName))
	common.CheckErr(t, err, true)
	rangeCount, err := largeRangeRes.GetColumn(common.QueryCountFieldName).GetAsInt64(0)
	require.NoError(t, err)
	require.Equal(t, int64(10000), rangeCount,
		"id in [50000, 60000) should return 10000 rows")
	t.Logf("Large range query [50000, 60000) returned %d rows", rangeCount)

	// ---------------------------------------------------------------
	// Step 7: Search with vector — verify nearest neighbor is id=0
	// ---------------------------------------------------------------
	vec := make([]float32, testVecDim)
	for i := range vec {
		vec[i] = float32(i) * 0.1 // same as id=0's embedding
	}
	searchRes, err := mc.Search(ctx, client.NewSearchOption(collName, 5,
		[]entity.Vector{entity.FloatVector(vec)}).
		WithConsistencyLevel(entity.ClStrong).
		WithANNSField("embedding").
		WithOutputFields("id", "value"))
	common.CheckErr(t, err, true)

	require.Equal(t, 1, len(searchRes), "should have 1 result set")
	require.Greater(t, searchRes[0].ResultCount, 0, "search should return results")
	nearestID, _ := searchRes[0].GetColumn("id").GetAsInt64(0)
	nearestVal, _ := searchRes[0].GetColumn("value").GetAsDouble(0)
	t.Logf("Search: nearest neighbor id=%d value=%.2f", nearestID, nearestVal)

	// ---------------------------------------------------------------
	// Step 8: Search with filter (hybrid search)
	// ---------------------------------------------------------------
	hybridRes, err := mc.Search(ctx, client.NewSearchOption(collName, 5,
		[]entity.Vector{entity.FloatVector(vec)}).
		WithConsistencyLevel(entity.ClStrong).
		WithANNSField("embedding").
		WithFilter("id >= 50000 && id < 60000").
		WithOutputFields("id", "value"))
	common.CheckErr(t, err, true)

	require.Equal(t, 1, len(hybridRes))
	require.Greater(t, hybridRes[0].ResultCount, 0, "hybrid search should return results")
	hybridIDCol := hybridRes[0].GetColumn("id")
	for i := 0; i < hybridIDCol.Len(); i++ {
		val, _ := hybridIDCol.GetAsInt64(i)
		require.GreaterOrEqual(t, val, int64(50000))
		require.Less(t, val, int64(60000))
	}
	t.Logf("Hybrid search (id in [50000,60000)) returned %d results with correct values", hybridRes[0].ResultCount)

	t.Log("Lance format external collection test passed!")
}

// generateVortexDataOnMinIO uses the Python vortex library to write a Vortex
// dataset directly to a MinIO/S3 bucket. This mirrors the Lance test pattern
// where a Python script handles format-specific data generation.
func generateVortexDataOnMinIO(t *testing.T, outputPath, schema string, numRows int, bucket string) {
	t.Helper()
	minioCfg := getMinIOConfig()
	args := []string{
		"--schema", schema,
		outputPath,
		strconv.Itoa(numRows),
		"--vec-dim", strconv.Itoa(testVecDim),
	}
	if schema == externalDataSchemaMulti {
		args = append(args, "--bin-vec-dim", strconv.Itoa(testBinVecDim))
	}

	out, err := runPythonScript(t, "python3", "generate_vortex_data.py",
		map[string]string{
			"MINIO_ADDRESS":    minioCfg.address,
			"MINIO_ACCESS_KEY": minioCfg.accessKey,
			"MINIO_SECRET_KEY": minioCfg.secretKey,
			"MINIO_BUCKET":     bucket,
		},
		args...)
	require.NoError(t, err, "generate vortex data: %s", out)
	require.Contains(t, out, "OK", "vortex data generation should succeed")
	t.Logf("Generated vortex data: %s", out)
}

// buildMultiTypeExternalSchema constructs the schema mirroring the one used
// by TestExternalCollectionMultipleDataTypes. Shared across format tests so
// that the same field set + index plan can be exercised with parquet, vortex,
// lance, etc.
func buildMultiTypeExternalSchema(collName, externalSource, externalSpec string) *entity.Schema {
	return entity.NewSchema().
		WithName(collName).
		WithExternalSource(externalSource).
		WithExternalSpec(externalSpec).
		WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithExternalField("id")).
		WithField(entity.NewField().WithName("bool_val").WithDataType(entity.FieldTypeBool).WithExternalField("bool_val")).
		WithField(entity.NewField().WithName("int8_val").WithDataType(entity.FieldTypeInt8).WithExternalField("int8_val")).
		WithField(entity.NewField().WithName("int16_val").WithDataType(entity.FieldTypeInt16).WithExternalField("int16_val")).
		WithField(entity.NewField().WithName("int32_val").WithDataType(entity.FieldTypeInt32).WithExternalField("int32_val")).
		WithField(entity.NewField().WithName("float_val").WithDataType(entity.FieldTypeFloat).WithExternalField("float_val")).
		WithField(entity.NewField().WithName("double_val").WithDataType(entity.FieldTypeDouble).WithExternalField("double_val")).
		WithField(entity.NewField().WithName("varchar_val").WithDataType(entity.FieldTypeVarChar).WithMaxLength(64).WithExternalField("varchar_val")).
		WithField(entity.NewField().WithName("json_val").WithDataType(entity.FieldTypeJSON).WithExternalField("json_val")).
		WithField(entity.NewField().WithName("array_int").WithDataType(entity.FieldTypeArray).WithElementType(entity.FieldTypeInt32).WithMaxCapacity(16).WithExternalField("array_int")).
		WithField(entity.NewField().WithName("array_str").WithDataType(entity.FieldTypeArray).WithElementType(entity.FieldTypeVarChar).WithMaxLength(64).WithMaxCapacity(16).WithExternalField("array_str")).
		WithField(entity.NewField().WithName("ts_val").WithDataType(entity.FieldTypeTimestamptz).WithExternalField("ts_val")).
		WithField(entity.NewField().WithName("geo_val").WithDataType(entity.FieldTypeGeometry).WithExternalField("geo_val")).
		WithField(entity.NewField().WithName("embedding").WithDataType(entity.FieldTypeFloatVector).WithDim(testVecDim).WithExternalField("embedding")).
		WithField(entity.NewField().WithName("bin_vec").WithDataType(entity.FieldTypeBinaryVector).WithDim(testBinVecDim).WithExternalField("bin_vec")).
		WithField(entity.NewField().WithName("fp16_vec").WithDataType(entity.FieldTypeFloat16Vector).WithDim(testVecDim).WithExternalField("fp16_vec")).
		WithField(entity.NewField().WithName("bf16_vec").WithDataType(entity.FieldTypeBFloat16Vector).WithDim(testVecDim).WithExternalField("bf16_vec")).
		WithField(entity.NewField().WithName("int8_vec").WithDataType(entity.FieldTypeInt8Vector).WithDim(testVecDim).WithExternalField("int8_vec"))
}

func buildVortexMultiTypeExternalSchema(collName, externalSource, externalSpec string) *entity.Schema {
	return entity.NewSchema().
		WithName(collName).
		WithExternalSource(externalSource).
		WithExternalSpec(externalSpec).
		WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithExternalField("id")).
		WithField(entity.NewField().WithName("bool_val").WithDataType(entity.FieldTypeBool).WithExternalField("bool_val")).
		WithField(entity.NewField().WithName("int8_val").WithDataType(entity.FieldTypeInt8).WithExternalField("int8_val")).
		WithField(entity.NewField().WithName("int16_val").WithDataType(entity.FieldTypeInt16).WithExternalField("int16_val")).
		WithField(entity.NewField().WithName("int32_val").WithDataType(entity.FieldTypeInt32).WithExternalField("int32_val")).
		WithField(entity.NewField().WithName("float_val").WithDataType(entity.FieldTypeFloat).WithExternalField("float_val")).
		WithField(entity.NewField().WithName("double_val").WithDataType(entity.FieldTypeDouble).WithExternalField("double_val")).
		WithField(entity.NewField().WithName("varchar_val").WithDataType(entity.FieldTypeVarChar).WithMaxLength(64).WithExternalField("varchar_val")).
		WithField(entity.NewField().WithName("json_val").WithDataType(entity.FieldTypeJSON).WithExternalField("json_val")).
		WithField(entity.NewField().WithName("array_int").WithDataType(entity.FieldTypeArray).WithElementType(entity.FieldTypeInt32).WithMaxCapacity(16).WithExternalField("array_int")).
		WithField(entity.NewField().WithName("array_str").WithDataType(entity.FieldTypeArray).WithElementType(entity.FieldTypeVarChar).WithMaxLength(64).WithMaxCapacity(16).WithExternalField("array_str")).
		WithField(entity.NewField().WithName("ts_val").WithDataType(entity.FieldTypeTimestamptz).WithExternalField("ts_val")).
		WithField(entity.NewField().WithName("geo_val").WithDataType(entity.FieldTypeGeometry).WithExternalField("geo_val")).
		WithField(entity.NewField().WithName("embedding").WithDataType(entity.FieldTypeFloatVector).WithDim(testVecDim).WithExternalField("embedding"))
}

// TestExternalCollectionVortexFormat tests external collections with vortex format:
//  1. Generate vortex dataset on MinIO using the generate_vortex_data tool
//  2. Create external collection with vortex format
//  3. Refresh and wait for completion
//  4. Create index and load collection
//  5. Query count(*) and verify row count
//  6. Query with filter
//  7. Search with vector
func TestExternalCollectionVortexFormat(t *testing.T) {
	t.Parallel()

	ctx := hp.CreateContext(t, time.Second*600)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	minioCfg := getMinIOConfig()
	minioClient, err := newMinIOClient(minioCfg)
	require.NoError(t, err)

	exists, err := minioClient.BucketExists(ctx, minioCfg.bucket)
	if err != nil || !exists {
		t.Skipf("MinIO bucket %q not accessible, skipping", minioCfg.bucket)
	}

	collName := common.GenRandomString("ext_vortex", 6)
	extPath := fmt.Sprintf("external-e2e-test/%s", collName)

	// ---------------------------------------------------------------
	// Step 1: Generate vortex dataset on MinIO
	// ---------------------------------------------------------------
	const totalRows = 100000
	generateVortexDataOnMinIO(t, extPath, externalDataSchemaBasic, totalRows, minioCfg.bucket)

	t.Cleanup(func() {
		cleanupMinIOPrefix(context.Background(), minioClient, minioCfg.bucket,
			fmt.Sprintf("%s/", extPath))
	})

	// ---------------------------------------------------------------
	// Step 2: Create external collection with vortex format
	// ---------------------------------------------------------------
	schema := entity.NewSchema().
		WithName(collName).
		WithExternalSource(extTestURI(minioCfg, extPath)).
		WithExternalSpec(extTestSpec(minioCfg, "vortex")).
		WithField(
			entity.NewField().
				WithName("id").
				WithDataType(entity.FieldTypeInt64).
				WithExternalField("id"),
		).
		WithField(
			entity.NewField().
				WithName("value").
				WithDataType(entity.FieldTypeFloat).
				WithExternalField("value"),
		).
		WithField(
			entity.NewField().
				WithName("embedding").
				WithDataType(entity.FieldTypeFloatVector).
				WithDim(testVecDim).
				WithExternalField("embedding"),
		).
		// VARCHAR / JSON / GEOMETRY trigger vortex-specific bugs (#49352, #49353)
		// that are not covered by the int/float/vector-only baseline.
		WithField(
			entity.NewField().
				WithName("varchar_val").
				WithDataType(entity.FieldTypeVarChar).
				WithMaxLength(64).
				WithExternalField("varchar_val"),
		).
		WithField(
			entity.NewField().
				WithName("json_val").
				WithDataType(entity.FieldTypeJSON).
				WithExternalField("json_val"),
		).
		WithField(
			entity.NewField().
				WithName("geo_val").
				WithDataType(entity.FieldTypeGeometry).
				WithExternalField("geo_val"),
		)

	err = mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
	common.CheckErr(t, err, true)
	t.Logf("Created external collection with vortex format: %s", collName)

	t.Cleanup(func() {
		_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(collName))
	})

	coll, err := mc.DescribeCollection(ctx, client.NewDescribeCollectionOption(collName))
	common.CheckErr(t, err, true)
	require.Equal(t, extTestURI(minioCfg, extPath), coll.Schema.ExternalSource)
	// Spec redacted at proxy edge.
	require.Contains(t, coll.Schema.ExternalSpec, `"format":"vortex"`)
	require.Contains(t, coll.Schema.ExternalSpec, `"access_key_id":"***"`)

	// ---------------------------------------------------------------
	// Step 3: Refresh and wait for completion
	// ---------------------------------------------------------------
	refreshAndWait(ctx, t, mc, collName)

	stats, err := mc.GetCollectionStats(ctx, client.NewGetCollectionStatsOption(collName))
	common.CheckErr(t, err, true)
	rowCountStr, ok := stats["row_count"]
	require.True(t, ok, "stats should contain row_count")
	rowCount, err := strconv.ParseInt(rowCountStr, 10, 64)
	require.NoError(t, err)
	require.Equal(t, int64(totalRows), rowCount,
		"collection stats row_count should match vortex dataset")
	t.Logf("Refresh complete: row_count=%d", rowCount)

	// ---------------------------------------------------------------
	// Step 4: Create scalar + vector indexes and load collection
	// ---------------------------------------------------------------
	indexAndLoadCollection(ctx, t, mc, collName, "embedding")

	// ---------------------------------------------------------------
	// Step 5: Query count(*)
	// ---------------------------------------------------------------
	countRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithOutputFields(common.QueryCountFieldName))
	common.CheckErr(t, err, true)

	count, err := countRes.GetColumn(common.QueryCountFieldName).GetAsInt64(0)
	require.NoError(t, err)
	require.Equal(t, int64(totalRows), count,
		"count(*) should match vortex dataset rows")
	t.Logf("Query count(*) = %d", count)

	// ---------------------------------------------------------------
	// Step 6: Query with filter
	// ---------------------------------------------------------------
	filterRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("id < 100").
		WithOutputFields("id", "value"))
	common.CheckErr(t, err, true)

	idCol := filterRes.GetColumn("id")
	require.Equal(t, 100, idCol.Len(),
		"filter id < 100 should return exactly 100 rows")
	t.Logf("Query with filter returned %d rows", idCol.Len())

	val42, err := filterRes.GetColumn("value").GetAsDouble(42)
	require.NoError(t, err)
	expected42 := float64(42) * 1.5
	require.InDelta(t, expected42, val42, 0.01,
		"value for id=42 should be 42*1.5=63.0")

	// ---------------------------------------------------------------
	// Step 6b: Verify data correctness — sample rows across the range
	// ---------------------------------------------------------------
	sampleIDs := []int64{0, 1, 999, 50000, 99999}
	for _, sid := range sampleIDs {
		sampleRes, err := mc.Query(ctx, client.NewQueryOption(collName).
			WithConsistencyLevel(entity.ClStrong).
			WithFilter(fmt.Sprintf("id == %d", sid)).
			WithOutputFields("id", "value"))
		common.CheckErr(t, err, true)
		require.Equal(t, 1, sampleRes.GetColumn("id").Len(),
			"should find exactly 1 row for id=%d", sid)
		sampleVal, err := sampleRes.GetColumn("value").GetAsDouble(0)
		require.NoError(t, err)
		require.InDelta(t, float64(sid)*1.5, sampleVal, 0.01,
			"value for id=%d should be %f", sid, float64(sid)*1.5)
	}
	t.Logf("Verified %d sample rows across id range", len(sampleIDs))

	// ---------------------------------------------------------------
	// Step 6c: Verify large range query
	// ---------------------------------------------------------------
	largeRangeRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithFilter("id >= 50000 && id < 60000").
		WithOutputFields(common.QueryCountFieldName))
	common.CheckErr(t, err, true)
	rangeCount, err := largeRangeRes.GetColumn(common.QueryCountFieldName).GetAsInt64(0)
	require.NoError(t, err)
	require.Equal(t, int64(10000), rangeCount,
		"id in [50000, 60000) should return 10000 rows")
	t.Logf("Large range query [50000, 60000) returned %d rows", rangeCount)

	// ---------------------------------------------------------------
	// Step 7: Search with vector — verify nearest neighbor is id=0
	// ---------------------------------------------------------------
	vec := make([]float32, testVecDim)
	for i := range vec {
		vec[i] = float32(i) * 0.1 // same as id=0's embedding
	}
	searchRes, err := mc.Search(ctx, client.NewSearchOption(collName, 5,
		[]entity.Vector{entity.FloatVector(vec)}).
		WithConsistencyLevel(entity.ClStrong).
		WithANNSField("embedding").
		WithOutputFields("id", "value"))
	common.CheckErr(t, err, true)

	require.Equal(t, 1, len(searchRes), "should have 1 result set")
	require.Greater(t, searchRes[0].ResultCount, 0, "search should return results")
	nearestID, _ := searchRes[0].GetColumn("id").GetAsInt64(0)
	nearestVal, _ := searchRes[0].GetColumn("value").GetAsDouble(0)
	t.Logf("Search: nearest neighbor id=%d value=%.2f", nearestID, nearestVal)

	// ---------------------------------------------------------------
	// Step 8: Search with filter (hybrid search)
	// ---------------------------------------------------------------
	hybridRes, err := mc.Search(ctx, client.NewSearchOption(collName, 5,
		[]entity.Vector{entity.FloatVector(vec)}).
		WithConsistencyLevel(entity.ClStrong).
		WithANNSField("embedding").
		WithFilter("id >= 50000 && id < 60000").
		WithOutputFields("id", "value"))
	common.CheckErr(t, err, true)

	require.Equal(t, 1, len(hybridRes))
	require.Greater(t, hybridRes[0].ResultCount, 0, "hybrid search should return results")
	hybridIDCol := hybridRes[0].GetColumn("id")
	for i := 0; i < hybridIDCol.Len(); i++ {
		val, _ := hybridIDCol.GetAsInt64(i)
		require.GreaterOrEqual(t, val, int64(50000))
		require.Less(t, val, int64(60000))
	}
	t.Logf("Hybrid search (id in [50000,60000)) returned %d results with correct values", hybridRes[0].ResultCount)

	t.Log("Vortex format external collection test passed!")
}

// TestExternalCollectionMultipleDataTypesVortex mirrors the scalar/array/json/
// geometry/timestamp coverage from TestExternalCollectionMultipleDataTypes.
// Vortex 0.56.0 cannot write Arrow FixedSizeBinary, so binary/half/int8 vector
// fields remain covered by the parquet and lance multi-type tests.
func TestExternalCollectionMultipleDataTypesVortex(t *testing.T) {
	t.Parallel()

	ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	minioCfg := getMinIOConfig()
	minioClient, err := newMinIOClient(minioCfg)
	require.NoError(t, err)

	exists, err := minioClient.BucketExists(ctx, minioCfg.bucket)
	if err != nil || !exists {
		t.Skipf("MinIO bucket %q not accessible, skipping", minioCfg.bucket)
	}

	collName := common.GenRandomString("ext_multi_vortex", 6)
	extPath := fmt.Sprintf("external-e2e-test/%s", collName)

	t.Cleanup(func() {
		cleanupMinIOPrefix(context.Background(), minioClient, minioCfg.bucket, extPath+"/")
		_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(collName))
	})

	// Step 1: Generate multi-type vortex dataset on MinIO
	const numRows = int64(100)
	generateVortexDataOnMinIO(t, extPath, externalDataSchemaMulti, int(numRows), minioCfg.bucket)

	// Step 2: Create external collection with Vortex-compatible multi-type schema
	schema := buildVortexMultiTypeExternalSchema(collName,
		extTestURI(minioCfg, extPath),
		extTestSpec(minioCfg, "vortex"))

	err = mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
	common.CheckErr(t, err, true)
	t.Logf("Created multi-type vortex external collection: %s", collName)

	// Step 3 onwards: shared refresh + index + load + verify
	runMultiTypeRefreshIndexLoadVerifyWithVectorFields(ctx, t, mc, collName, numRows, "", "", []string{"embedding"})
}

// TestExternalCollectionMultipleDataTypesLance mirrors
// TestExternalCollectionMultipleDataTypesVortex but writes the dataset in lance
// format. Exercises the same multi-type schema and shared
// refresh/index/load/query/search verification across the lance bridge.
func TestExternalCollectionMultipleDataTypesLance(t *testing.T) {
	t.Parallel()

	ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	minioCfg := getMinIOConfig()
	minioClient, err := newMinIOClient(minioCfg)
	require.NoError(t, err)

	exists, err := minioClient.BucketExists(ctx, minioCfg.bucket)
	if err != nil || !exists {
		t.Skipf("MinIO bucket %q not accessible, skipping", minioCfg.bucket)
	}

	collName := common.GenRandomString("ext_multi_lance", 6)
	extPath := fmt.Sprintf("external-e2e-test/%s", collName)

	t.Cleanup(func() {
		cleanupMinIOPrefix(context.Background(), minioClient, minioCfg.bucket, extPath+"/")
		_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(collName))
	})

	const numRows = int64(100)
	// Lance writer takes a full s3:// URI (different from vortex's relative path).
	s3URI := fmt.Sprintf("s3://%s/%s", minioCfg.bucket, extPath)
	generateLanceDataOnMinIO(t, s3URI, externalDataSchemaMulti, int(numRows), 0)

	schema := buildMultiTypeExternalSchema(collName,
		extTestURI(minioCfg, extPath),
		extTestSpec(minioCfg, "lance-table"))

	err = mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
	common.CheckErr(t, err, true)
	t.Logf("Created multi-type lance external collection: %s", collName)

	runMultiTypeRefreshIndexLoadVerify(ctx, t, mc, collName, numRows)
}

// TestExternalCollectionFloat32ListVector verifies that external collections
// whose files store vectors as FixedSizeList<Float32, dim> (native float list)
// can be refreshed, indexed, loaded, and queried correctly.
// Currently only Parquet is tested because Parquet readers return native Arrow
// types regardless of the output schema.  Vortex and Lance require schema-level
// changes to support float32 list vectors (tracked separately).
func TestExternalCollectionFloat32ListVector(t *testing.T) {
	t.Parallel()

	ctx := hp.CreateContext(t, time.Second*600)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	// Connect to MinIO
	minioCfg := getMinIOConfig()
	minioClient, err := newMinIOClient(minioCfg)
	require.NoError(t, err)
	exists, err := minioClient.BucketExists(ctx, minioCfg.bucket)
	if err != nil || !exists {
		t.Skipf("MinIO bucket %q not accessible, skipping", minioCfg.bucket)
	}

	const numRows = 2000

	type formatCase struct {
		name      string
		specJSON  string
		setupData func(t *testing.T, extPath string)
		cleanup   func(extPath string)
	}

	formats := []formatCase{
		{
			name:     "parquet",
			specJSON: extTestSpec(minioCfg, "parquet"),
			setupData: func(t *testing.T, extPath string) {
				t.Helper()
				data, genErr := generateParquetBytes(externalDataSchemaBasic, numRows, 0, testVecDim)
				require.NoError(t, genErr, "generate parquet")
				objectKey := fmt.Sprintf("%s/data.parquet", extPath)
				_, putErr := minioClient.PutObject(ctx, minioCfg.bucket, objectKey,
					bytes.NewReader(data), int64(len(data)),
					miniogo.PutObjectOptions{ContentType: "application/octet-stream"})
				require.NoError(t, putErr, "upload parquet to MinIO")
				t.Logf("Uploaded %s/%s (%d bytes, %d rows)", minioCfg.bucket, objectKey, len(data), numRows)
			},
			cleanup: func(extPath string) {
				cleanupMinIOPrefix(context.Background(), minioClient, minioCfg.bucket, extPath+"/")
			},
		},
	}

	for _, fc := range formats {
		fc := fc // capture for subtest closure
		t.Run(fc.name, func(t *testing.T) {
			// Replace hyphens with underscores for valid collection names
			safePrefix := strings.ReplaceAll(fc.name, "-", "_")
			collName := common.GenRandomString("ext_f32_"+safePrefix, 6)
			extPath := fmt.Sprintf("external-e2e-test/%s", collName)

			// Step 1: Generate and upload data
			fc.setupData(t, extPath)
			t.Cleanup(func() { fc.cleanup(extPath) })

			// Step 2: Create external collection
			schema := entity.NewSchema().
				WithName(collName).
				WithExternalSource(extTestURI(minioCfg, extPath)).
				WithExternalSpec(fc.specJSON).
				WithField(entity.NewField().
					WithName("id").
					WithDataType(entity.FieldTypeInt64).
					WithExternalField("id")).
				WithField(entity.NewField().
					WithName("value").
					WithDataType(entity.FieldTypeFloat).
					WithExternalField("value")).
				WithField(entity.NewField().
					WithName("embedding").
					WithDataType(entity.FieldTypeFloatVector).
					WithDim(testVecDim).
					WithExternalField("embedding"))

			err := mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
			common.CheckErr(t, err, true)
			t.Logf("Created external collection [%s]: %s", fc.name, collName)

			t.Cleanup(func() {
				_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(collName))
			})

			// Step 3: Refresh and wait
			refreshAndWait(ctx, t, mc, collName)

			// Verify row count via stats
			stats, err := mc.GetCollectionStats(ctx, client.NewGetCollectionStatsOption(collName))
			common.CheckErr(t, err, true)
			rowCountStr, ok := stats["row_count"]
			require.True(t, ok, "stats should contain row_count")
			rowCount, err := strconv.ParseInt(rowCountStr, 10, 64)
			require.NoError(t, err)
			require.Equal(t, int64(numRows), rowCount,
				"row_count should match uploaded data")
			t.Logf("Refresh complete: row_count=%d", rowCount)

			// Step 4: Index + Load
			indexAndLoadCollection(ctx, t, mc, collName, "embedding")

			// Step 5: Query count(*)
			countRes, err := mc.Query(ctx, client.NewQueryOption(collName).
				WithConsistencyLevel(entity.ClStrong).
				WithOutputFields(common.QueryCountFieldName))
			common.CheckErr(t, err, true)
			count, err := countRes.GetColumn(common.QueryCountFieldName).GetAsInt64(0)
			require.NoError(t, err)
			require.Equal(t, int64(numRows), count,
				"count(*) should match uploaded rows")
			t.Logf("count(*) = %d", count)

			// Step 6: Query with filter + output fields
			filterRes, err := mc.Query(ctx, client.NewQueryOption(collName).
				WithConsistencyLevel(entity.ClStrong).
				WithFilter("id < 100").
				WithOutputFields("id", "value"))
			common.CheckErr(t, err, true)
			require.Equal(t, 100, filterRes.GetColumn("id").Len(),
				"filter id < 100 should return 100 rows")
			t.Logf("Query with filter returned %d rows", filterRes.GetColumn("id").Len())

			// Verify a specific value (id=42 → value=42*1.5=63.0)
			val42, err := filterRes.GetColumn("value").GetAsDouble(42)
			require.NoError(t, err)
			require.InDelta(t, float64(42)*1.5, val42, 0.01,
				"value for id=42 should be 63.0")

			// Step 7: Vector search
			vec := make([]float32, testVecDim)
			for i := range vec {
				vec[i] = float32(i) * 0.1 // matches id=0's embedding
			}
			searchRes, err := mc.Search(ctx, client.NewSearchOption(collName, 5,
				[]entity.Vector{entity.FloatVector(vec)}).
				WithConsistencyLevel(entity.ClStrong).
				WithANNSField("embedding").
				WithOutputFields("id", "value"))
			common.CheckErr(t, err, true)
			require.Equal(t, 1, len(searchRes), "1 result set for 1 query vector")
			require.Greater(t, searchRes[0].ResultCount, 0, "search should return results")
			nearestID, _ := searchRes[0].GetColumn("id").GetAsInt64(0)
			t.Logf("Search nearest: id=%d", nearestID)

			// Step 8: Hybrid search (vector + filter)
			hybridRes, err := mc.Search(ctx, client.NewSearchOption(collName, 5,
				[]entity.Vector{entity.FloatVector(vec)}).
				WithConsistencyLevel(entity.ClStrong).
				WithANNSField("embedding").
				WithFilter("id >= 500 && id < 1500").
				WithOutputFields("id"))
			common.CheckErr(t, err, true)
			require.Equal(t, 1, len(hybridRes))
			require.Greater(t, hybridRes[0].ResultCount, 0, "hybrid search should return results")
			hybridIDCol := hybridRes[0].GetColumn("id")
			for i := 0; i < hybridIDCol.Len(); i++ {
				val, _ := hybridIDCol.GetAsInt64(i)
				require.GreaterOrEqual(t, val, int64(500), "id should be >= 500")
				require.Less(t, val, int64(1500), "id should be < 1500")
			}
			t.Logf("Hybrid search (id in [500,1500)) returned %d results", hybridRes[0].ResultCount)

			t.Logf("[%s] All checks passed!", fc.name)
		})
	}
}

// TestExternalCollectionDifferentBucket tests external collection with data in a
// different MinIO bucket than what Milvus uses (a-bucket).
// This verifies that external_source with s3://external-bucket/path/ URI works
// when the data resides in a separate bucket on the same MinIO instance.
func TestExternalCollectionDifferentBucket(t *testing.T) {
	ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	// Connect to MinIO
	minioCfg := getMinIOConfig()
	minioClient, err := newMinIOClient(minioCfg)
	require.NoError(t, err)

	skipIfMinIOUnreachable(ctx, t, minioClient, minioCfg.bucket)

	// Use a different bucket than Milvus's default (a-bucket)
	externalBucket := "external-bucket"

	// Create the external bucket if it doesn't exist
	exists, err := minioClient.BucketExists(ctx, externalBucket)
	require.NoError(t, err, "check bucket existence")
	if !exists {
		err = minioClient.MakeBucket(ctx, externalBucket, miniogo.MakeBucketOptions{})
		require.NoError(t, err, "create external bucket")
		t.Logf("Created bucket: %s", externalBucket)
	}

	collName := common.GenRandomString("ext_diff_bucket", 6)
	extPath := fmt.Sprintf("external-e2e-test/%s", collName)

	// ---------------------------------------------------------------
	// Step 1: Generate and upload parquet data to the EXTERNAL bucket
	// ---------------------------------------------------------------
	const numFiles = 2
	const rowsPerFile = int64(500)
	totalExpectedRows := int64(numFiles) * rowsPerFile

	for i := 0; i < numFiles; i++ {
		data, err := generateParquetBytes(externalDataSchemaBasic, rowsPerFile, int64(i)*rowsPerFile, testVecDim)
		require.NoError(t, err, "generate parquet file %d", i)

		objectKey := fmt.Sprintf("%s/data%d.parquet", extPath, i)
		uploadParquetToMinIO(ctx, t, minioClient, externalBucket, objectKey, data)
	}

	t.Cleanup(func() {
		cleanupMinIOPrefix(context.Background(), minioClient, externalBucket,
			fmt.Sprintf("%s/", extPath))
	})

	// ---------------------------------------------------------------
	// Step 2: Create external collection with minio://external-bucket/path URI
	// ---------------------------------------------------------------
	// The key difference: ExternalSource uses a full minio:// URI pointing to
	// a different bucket than Milvus's own bucket (a-bucket).
	// URI format: minio://host:port/bucket/path — post-refactor validator
	// requires explicit non-empty scheme + host. Empty-host shorthand
	// (s3:///bucket/path) was dropped.
	externalSource := fmt.Sprintf("minio://%s/%s/%s", minioCfg.address, externalBucket, extPath)
	t.Logf("ExternalSource: %s (Milvus bucket: %s)", externalSource, minioCfg.bucket)

	schema := entity.NewSchema().
		WithName(collName).
		WithExternalSource(externalSource).
		WithExternalSpec(extTestSpec(minioCfg, "parquet")).
		WithField(
			entity.NewField().
				WithName("id").
				WithDataType(entity.FieldTypeInt64).
				WithExternalField("id"),
		).
		WithField(
			entity.NewField().
				WithName("value").
				WithDataType(entity.FieldTypeFloat).
				WithExternalField("value"),
		).
		WithField(
			entity.NewField().
				WithName("embedding").
				WithDataType(entity.FieldTypeFloatVector).
				WithDim(testVecDim).
				WithExternalField("embedding"),
		)

	err = mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
	common.CheckErr(t, err, true)
	t.Logf("Created external collection: %s", collName)

	t.Cleanup(func() {
		_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(collName))
	})

	// ---------------------------------------------------------------
	// Step 3: Trigger refresh and wait for completion
	// ---------------------------------------------------------------
	refreshAndWait(ctx, t, mc, collName)

	// ---------------------------------------------------------------
	// Step 4: Verify row count
	// ---------------------------------------------------------------
	stats, err := mc.GetCollectionStats(ctx, client.NewGetCollectionStatsOption(collName))
	common.CheckErr(t, err, true)

	rowCountStr, ok := stats["row_count"]
	require.True(t, ok, "collection stats should contain row_count")

	rowCount, err := strconv.ParseInt(rowCountStr, 10, 64)
	require.NoError(t, err, "parse row_count")
	require.Equal(t, totalExpectedRows, rowCount,
		"total row count should match uploaded parquet data")
	t.Logf("Verified: row_count=%d (expected %d)", rowCount, totalExpectedRows)

	// ---------------------------------------------------------------
	// Step 5: Create index, load, and query
	// ---------------------------------------------------------------
	indexAndLoadCollection(ctx, t, mc, collName, "embedding")

	// Query count(*)
	countRes, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithConsistencyLevel(entity.ClStrong).
		WithOutputFields(common.QueryCountFieldName))
	common.CheckErr(t, err, true)

	count, err := countRes.GetColumn(common.QueryCountFieldName).GetAsInt64(0)
	require.NoError(t, err)
	require.Equal(t, totalExpectedRows, count,
		"count(*) should match total uploaded rows")
	t.Logf("Query count(*) = %d", count)

	// Search
	vec := make([]float32, testVecDim)
	for i := range vec {
		vec[i] = float32(i) * 0.1
	}
	searchRes, err := mc.Search(ctx, client.NewSearchOption(collName, 5,
		[]entity.Vector{entity.FloatVector(vec)}).
		WithConsistencyLevel(entity.ClStrong).
		WithANNSField("embedding").
		WithOutputFields("id", "value"))
	common.CheckErr(t, err, true)

	require.Equal(t, 1, len(searchRes), "should have 1 result set")
	require.Greater(t, searchRes[0].ResultCount, 0, "search should return results")
	t.Logf("Search returned %d results", searchRes[0].ResultCount)

	t.Log("External collection with different bucket test completed!")
}

// TestRefreshExternalCollectionUpdatesSchema verifies that when RefreshExternalCollection
// is called with new ExternalSource and ExternalSpec, the collection schema is updated
// and subsequent DescribeCollection reflects the new values.
func TestRefreshExternalCollectionUpdatesSchema(t *testing.T) {
	ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	// Connect to MinIO
	minioCfg := getMinIOConfig()
	minioClient, err := newMinIOClient(minioCfg)
	require.NoError(t, err)
	exists, err := minioClient.BucketExists(ctx, minioCfg.bucket)
	if err != nil || !exists {
		t.Skipf("MinIO bucket %q not accessible, skipping", minioCfg.bucket)
	}

	collName := common.GenRandomString("ext_schema_upd", 6)

	// Prepare two data paths with different parquet files. Validator requires
	// fully-qualified scheme://host/bucket/key URIs — bare relative paths
	// are rejected post-refactor.
	pathV1 := extTestURI(minioCfg, fmt.Sprintf("external-e2e-test/%s/v1", collName))
	pathV2 := extTestURI(minioCfg, fmt.Sprintf("external-e2e-test/%s/v2", collName))
	relV1 := fmt.Sprintf("external-e2e-test/%s/v1", collName)
	relV2 := fmt.Sprintf("external-e2e-test/%s/v2", collName)

	const rowsPerFile = int64(100)
	for _, p := range []struct {
		path    string
		startID int64
	}{
		{relV1, 0},
		{relV2, 1000},
	} {
		data, genErr := generateParquetBytes(externalDataSchemaBasic, rowsPerFile, p.startID, testVecDim)
		require.NoError(t, genErr)
		objectKey := fmt.Sprintf("%s/data.parquet", p.path)
		_, putErr := minioClient.PutObject(ctx, minioCfg.bucket, objectKey,
			bytes.NewReader(data), int64(len(data)),
			miniogo.PutObjectOptions{ContentType: "application/octet-stream"})
		require.NoError(t, putErr)
		t.Logf("Uploaded %s/%s (%d rows)", minioCfg.bucket, objectKey, rowsPerFile)
	}

	t.Cleanup(func() {
		cleanupMinIOPrefix(context.Background(), minioClient, minioCfg.bucket,
			fmt.Sprintf("external-e2e-test/%s/", collName))
	})

	// ---------------------------------------------------------------
	// Step 1: Create external collection with v1 source and parquet spec
	// ---------------------------------------------------------------
	specV1 := extTestSpec(minioCfg, "parquet")
	schema := entity.NewSchema().
		WithName(collName).
		WithExternalSource(pathV1).
		WithExternalSpec(specV1).
		WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithExternalField("id")).
		WithField(entity.NewField().WithName("value").WithDataType(entity.FieldTypeFloat).WithExternalField("value")).
		WithField(entity.NewField().WithName("embedding").WithDataType(entity.FieldTypeFloatVector).WithDim(testVecDim).WithExternalField("embedding"))

	err = mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
	common.CheckErr(t, err, true)
	t.Logf("Created external collection: %s (source=%s)", collName, pathV1)

	t.Cleanup(func() {
		_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(collName))
	})

	// Verify initial schema
	coll, err := mc.DescribeCollection(ctx, client.NewDescribeCollectionOption(collName))
	common.CheckErr(t, err, true)
	require.Equal(t, pathV1, coll.Schema.ExternalSource, "initial ExternalSource should be pathV1")
	// DescribeCollection redacts secret extfs values at the proxy edge;
	// assert format + non-secret keys + redaction markers, not byte-equality.
	require.Contains(t, coll.Schema.ExternalSpec, `"format":"parquet"`)
	require.Contains(t, coll.Schema.ExternalSpec, `"access_key_id":"***"`)
	require.Contains(t, coll.Schema.ExternalSpec, `"access_key_value":"***"`)
	require.NotContains(t, coll.Schema.ExternalSpec, `"region":`)

	// ---------------------------------------------------------------
	// Step 2: First refresh (with default source/spec) — establishes baseline
	// ---------------------------------------------------------------
	refreshResult, err := mc.RefreshExternalCollection(ctx,
		client.NewRefreshExternalCollectionOption(collName))
	common.CheckErr(t, err, true)
	waitRefreshComplete(t, ctx, mc, refreshResult.JobID, 120*time.Second)
	t.Logf("First refresh completed (jobID=%d)", refreshResult.JobID)

	// ---------------------------------------------------------------
	// Step 3: Second refresh with NEW source and spec
	// ---------------------------------------------------------------
	specV2 := extTestSpec(minioCfg, "parquet")
	refreshResult2, err := mc.RefreshExternalCollection(ctx,
		client.NewRefreshExternalCollectionOption(collName).
			WithExternalSource(pathV2).
			WithExternalSpec(specV2))
	common.CheckErr(t, err, true)
	require.Greater(t, refreshResult2.JobID, int64(0))
	t.Logf("Second refresh started (jobID=%d, newSource=%s, newSpec=%s)",
		refreshResult2.JobID, pathV2, specV2)

	waitRefreshComplete(t, ctx, mc, refreshResult2.JobID, 120*time.Second)
	t.Logf("Second refresh completed (jobID=%d)", refreshResult2.JobID)

	// ---------------------------------------------------------------
	// Step 4: Verify schema updated via DescribeCollection
	// ---------------------------------------------------------------
	coll2, err := mc.DescribeCollection(ctx, client.NewDescribeCollectionOption(collName))
	common.CheckErr(t, err, true)
	require.Equal(t, pathV2, coll2.Schema.ExternalSource,
		"ExternalSource should be updated to pathV2 after refresh")
	// Redacted via proxy edge; verify shape + non-secret values.
	require.Contains(t, coll2.Schema.ExternalSpec, `"format":"parquet"`)
	require.Contains(t, coll2.Schema.ExternalSpec, `"access_key_id":"***"`)
	require.Contains(t, coll2.Schema.ExternalSpec, `"access_key_value":"***"`)
	require.NotContains(t, coll2.Schema.ExternalSpec, `"region":`)
	t.Logf("Verified: schema updated — source=%s, spec=%s", coll2.Schema.ExternalSource, coll2.Schema.ExternalSpec)
	_ = specV2 // value used to drive the refresh; final spec compared structurally above.
}

// TestExternalSourceFormats is a table-driven test covering the 3 supported
// data formats (parquet / lance-table / vortex) against a fully qualified
// external_source URI (scheme://endpoint/bucket/key).
//
// Legacy bare-relative-path and empty-host shorthand forms were removed in
// the extfs-isolation refactor — the validator now requires an explicit URI
// with non-empty scheme+host. Bucket-mode variation is therefore collapsed
// into a single cross-endpoint form targeting a dedicated external bucket
// (different from Milvus's own bucket) with explicit credentials in
// spec.extfs. This is the only URI shape the new validator accepts.
func TestExternalSourceFormats(t *testing.T) {
	ctx := hp.CreateContext(t, time.Second*600)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	minioCfg := getMinIOConfig()
	minioClient, err := newMinIOClient(minioCfg)
	require.NoError(t, err)
	exists, err := minioClient.BucketExists(ctx, minioCfg.bucket)
	if err != nil || !exists {
		t.Skipf("MinIO bucket %q not accessible", minioCfg.bucket)
	}

	// Ensure external-bucket target exists
	const externalBucket = "external-bucket"
	if ok, _ := minioClient.BucketExists(ctx, externalBucket); !ok {
		require.NoError(t, minioClient.MakeBucket(ctx, externalBucket, miniogo.MakeBucketOptions{}))
	}

	const rowsPerFile = int64(100)
	const numRows = 100

	// Build extfs credentials JSON fragment once. The minio:// external source
	// carries the endpoint and bucket, so tests only need credentials here.
	extfsFragment := fmt.Sprintf(
		`"extfs":{"access_key_id":%q,"access_key_value":%q}`,
		minioCfg.accessKey, minioCfg.secretKey)

	type testCase struct {
		name   string
		format string // parquet, lance-table, vortex
	}

	cases := []testCase{
		{"parquet", "parquet"},
		{"lance_table", "lance-table"},
		{"vortex", "vortex"},
	}

	for _, tc := range cases {
		tc := tc
		t.Run(tc.name, func(t *testing.T) {
			collName := common.GenRandomString("ext_fmt", 6)
			dataPath := fmt.Sprintf("external-e2e-test/%s", collName)

			// Fully qualified URI (scheme://endpoint/bucket/key) targeting
			// the dedicated external bucket.
			targetBucket := externalBucket
			externalSource := fmt.Sprintf("minio://%s/%s/%s", minioCfg.address, externalBucket, dataPath)

			// Generate and upload test data
			switch tc.format {
			case "parquet":
				data, genErr := generateParquetBytes(externalDataSchemaBasic, rowsPerFile, 0, testVecDim)
				require.NoError(t, genErr)
				objectKey := fmt.Sprintf("%s/data.parquet", dataPath)
				uploadParquetToMinIO(ctx, t, minioClient, targetBucket, objectKey, data)
			case "lance-table":
				s3URI := fmt.Sprintf("s3://%s/%s", targetBucket, dataPath)
				generateLanceDataOnMinIO(t, s3URI, externalDataSchemaBasic, numRows, 0)
			case "vortex":
				generateVortexDataOnMinIO(t, dataPath, externalDataSchemaBasic, numRows, targetBucket)
			}

			t.Cleanup(func() {
				cleanupMinIOPrefix(context.Background(), minioClient, targetBucket,
					fmt.Sprintf("%s/", dataPath))
			})

			// Create external collection
			specJSON := fmt.Sprintf(`{"format":%q,%s}`, tc.format, extfsFragment)
			schema := entity.NewSchema().
				WithName(collName).
				WithExternalSource(externalSource).
				WithExternalSpec(specJSON).
				WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithExternalField("id")).
				WithField(entity.NewField().WithName("value").WithDataType(entity.FieldTypeFloat).WithExternalField("value")).
				WithField(entity.NewField().WithName("embedding").WithDataType(entity.FieldTypeFloatVector).WithDim(testVecDim).WithExternalField("embedding"))

			err := mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
			common.CheckErr(t, err, true)
			t.Logf("Created collection: %s (source=%s, format=%s)", collName, externalSource, tc.format)

			t.Cleanup(func() {
				_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(collName))
			})

			// Refresh and wait
			refreshAndWait(ctx, t, mc, collName)

			// Verify row count via stats
			stats, statErr := mc.GetCollectionStats(ctx, client.NewGetCollectionStatsOption(collName))
			common.CheckErr(t, statErr, true)
			rowCountStr := stats["row_count"]
			rowCount, parseErr := strconv.ParseInt(rowCountStr, 10, 64)
			require.NoError(t, parseErr)
			require.Equal(t, int64(numRows), rowCount,
				"expected %d rows, got %d", numRows, rowCount)
			t.Logf("Verified stats: %d rows", rowCount)

			// Index + Load + Query
			indexAndLoadCollection(ctx, t, mc, collName, "embedding")

			// Query count(*)
			countRes, queryErr := mc.Query(ctx, client.NewQueryOption(collName).
				WithOutputFields("count(*)"))
			common.CheckErr(t, queryErr, true)
			count, _ := countRes.Fields[0].GetAsInt64(0)
			require.Equal(t, int64(numRows), count,
				"query count(*) expected %d, got %d", numRows, count)

			// Query with filter: id < 10
			filterRes, filterErr := mc.Query(ctx, client.NewQueryOption(collName).
				WithFilter("id < 10").
				WithOutputFields("id", "value"))
			common.CheckErr(t, filterErr, true)
			require.Equal(t, 10, filterRes.Fields[0].Len(),
				"expected 10 rows with id < 10, got %d", filterRes.Fields[0].Len())

			t.Logf("Verified: index + load + query count(*)=%d, filter=10 rows", count)
		})
	}
}

// waitRefreshComplete polls refresh job progress until it reaches Completed or fails with timeout.
func waitRefreshComplete(t *testing.T, ctx context.Context, mc *base.MilvusClient, jobID int64, timeout time.Duration) {
	t.Helper()
	deadline := time.After(timeout)
	ticker := time.NewTicker(2 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-deadline:
			t.Fatalf("Refresh job %d timed out after %s", jobID, timeout)
		case <-ticker.C:
			progress, err := mc.GetRefreshExternalCollectionProgress(ctx,
				client.NewGetRefreshExternalCollectionProgressOption(jobID))
			require.NoError(t, err)
			t.Logf("Job %d: state=%s progress=%d%%", jobID, progress.State, progress.Progress)

			if progress.State == entity.RefreshStateCompleted {
				return
			}
			if progress.State == entity.RefreshStateFailed {
				t.Fatalf("Refresh job %d failed: %s", jobID, progress.Reason)
			}
		}
	}
}

// TestExternalCollectionLoadPerfProfile writes 500K rows with 768d vectors + scalar fields,
// triggers refresh + index + load, and measures load performance with timing logs.
func TestExternalCollectionLoadPerfProfile(t *testing.T) {
	if !isTruthyEnv("GO_CLIENT_ENABLE_EXTERNAL_TABLE_PERF_PROFILE") {
		t.Skip("set GO_CLIENT_ENABLE_EXTERNAL_TABLE_PERF_PROFILE=1 to run the external table perf profile")
	}

	ctx := hp.CreateContext(t, time.Second*600)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	minioCfg := getMinIOConfig()
	minioClient, err := newMinIOClient(minioCfg)
	require.NoError(t, err)
	skipIfMinIOUnreachable(ctx, t, minioClient, minioCfg.bucket)

	collName := common.GenRandomString("ext_perf_profile", 6)
	extPath := "external-e2e-test/" + collName
	t.Cleanup(func() {
		cleanupMinIOPrefix(context.Background(), minioClient, minioCfg.bucket, extPath+"/")
	})

	const totalRows = 500000
	const vecDim = 128

	t.Logf("Generating %d rows with %d-dim vectors + scalars...", totalRows, vecDim)
	genStart := time.Now()
	data, err := generateParquetBytes(externalDataSchemaLarge, totalRows, 0, vecDim)
	require.NoError(t, err)
	t.Logf("Generated %d bytes (%.1f MB) in %v", len(data), float64(len(data))/1048576.0, time.Since(genStart))

	uploadStart := time.Now()
	uploadParquetToMinIO(ctx, t, minioClient, minioCfg.bucket, extPath+"/data.parquet", data)
	t.Logf("Uploaded to MinIO in %v", time.Since(uploadStart))

	schema := entity.NewSchema().
		WithName(collName).
		WithExternalSource(extTestURI(minioCfg, extPath)).
		WithExternalSpec(extTestSpec(minioCfg, "parquet")).
		WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithExternalField("id")).
		WithField(entity.NewField().WithName("score").WithDataType(entity.FieldTypeDouble).WithExternalField("score")).
		WithField(entity.NewField().WithName("label").WithDataType(entity.FieldTypeInt32).WithExternalField("label")).
		WithField(entity.NewField().WithName("tag").WithDataType(entity.FieldTypeVarChar).WithMaxLength(256).WithExternalField("tag")).
		WithField(entity.NewField().WithName("value").WithDataType(entity.FieldTypeFloat).WithExternalField("value")).
		WithField(entity.NewField().WithName("embedding").WithDataType(entity.FieldTypeFloatVector).WithDim(vecDim).WithExternalField("embedding"))

	err = mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
	common.CheckErr(t, err, true)
	t.Cleanup(func() {
		_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(collName))
	})

	t.Log("Starting refresh...")
	refreshStart := time.Now()
	refreshAndWait(ctx, t, mc, collName)
	t.Logf("Refresh done in %v", time.Since(refreshStart))

	t.Log("Creating index...")
	indexStart := time.Now()
	idxTask, err := mc.CreateIndex(ctx, client.NewCreateIndexOption(collName, "embedding",
		index.NewAutoIndex(entity.L2)))
	common.CheckErr(t, err, true)
	err = idxTask.Await(ctx)
	common.CheckErr(t, err, true)
	t.Logf("Index created in %v", time.Since(indexStart))

	// Wait for index metadata to fully propagate.
	time.Sleep(5 * time.Second)

	// NOTE: Load is intentionally skipped here.
	// The test script will measure memory baseline after index,
	// then trigger Load via pymilvus to capture the true memory delta.

	t.Log("=== PERF PROFILE COMPLETE (index only, no load) ===")
	t.Logf("Total rows: %d, Vector dim: %d", totalRows, vecDim)
	t.Logf("Raw vector data: %.1f MB", float64(totalRows)*float64(vecDim)*4/1048576.0)
	t.Logf("Estimated total row size: %d bytes (vec=%d + scalars~30)", vecDim*4+30, vecDim*4)
}

// TestRefreshExternalCollectionZeroRowParquet covers issue #49225:
//
// A zero-row parquet file in the external bucket previously crashed
// datanode with "integer divide by zero" inside balanceFragmentsToSegments,
// putting the standalone Milvus pod into CrashLoopBackOff. The fix
// short-circuits at DataCoord.createTasksForJob, marking the refresh job
// Failed with a non-retriable reason instead.
//
// Pass criteria:
//  1. refresh_external_collection accepts the request and returns a job_id.
//  2. The job transitions to a terminal state (Failed expected) within the
//     poll deadline — i.e. it does NOT hang in Pending forever AND the
//     datanode does NOT crash mid-poll.
//  3. The Failed reason mentions zero rows so operators have actionable
//     signal.
func TestRefreshExternalCollectionZeroRowParquet(t *testing.T) {
	ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	minioCfg := getMinIOConfig()
	minioClient, err := newMinIOClient(minioCfg)
	require.NoError(t, err)

	exists, err := minioClient.BucketExists(ctx, minioCfg.bucket)
	if err != nil || !exists {
		t.Skipf("MinIO bucket %q not accessible (exists=%v, err=%v), skipping",
			minioCfg.bucket, exists, err)
	}

	collName := common.GenRandomString("ext_zero_row", 6)
	extPath := fmt.Sprintf("external-e2e-test/%s", collName)

	// Upload a single well-formed parquet file with zero rows.
	data, err := generateParquetBytes(externalDataSchemaBasic, 0, 0, testVecDim)
	require.NoError(t, err, "generate zero-row parquet")
	require.NotEmpty(t, data, "zero-row parquet must still be a valid file with header/footer")

	objectKey := fmt.Sprintf("%s/empty.parquet", extPath)
	_, err = minioClient.PutObject(ctx, minioCfg.bucket, objectKey,
		bytes.NewReader(data), int64(len(data)),
		miniogo.PutObjectOptions{ContentType: "application/octet-stream"})
	require.NoError(t, err, "upload zero-row parquet to MinIO")
	t.Logf("Uploaded zero-row parquet %s/%s (%d bytes)",
		minioCfg.bucket, objectKey, len(data))

	t.Cleanup(func() {
		cleanupMinIOPrefix(context.Background(), minioClient, minioCfg.bucket,
			fmt.Sprintf("%s/", extPath))
	})

	schema := entity.NewSchema().
		WithName(collName).
		WithExternalSource(extTestURI(minioCfg, extPath)).
		WithExternalSpec(extTestSpec(minioCfg, "parquet")).
		WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithExternalField("id")).
		WithField(entity.NewField().WithName("value").WithDataType(entity.FieldTypeFloat).WithExternalField("value")).
		WithField(entity.NewField().WithName("embedding").WithDataType(entity.FieldTypeFloatVector).
			WithDim(testVecDim).WithExternalField("embedding"))

	err = mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
	common.CheckErr(t, err, true)
	t.Cleanup(func() {
		_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(collName))
	})

	refreshResult, err := mc.RefreshExternalCollection(ctx,
		client.NewRefreshExternalCollectionOption(collName))
	common.CheckErr(t, err, true)
	require.Greater(t, refreshResult.JobID, int64(0))
	jobID := refreshResult.JobID
	t.Logf("Started refresh job: %d", jobID)

	// Must reach a terminal state within 60s. Before the fix this loop hit
	// the deadline because the datanode CrashLoopBackOff prevented any task
	// from running to completion.
	deadline := time.After(60 * time.Second)
	ticker := time.NewTicker(2 * time.Second)
	defer ticker.Stop()

	var finalState entity.RefreshExternalCollectionState
	var finalReason string
	for done := false; !done; {
		select {
		case <-deadline:
			t.Fatalf("Refresh job %d did not reach terminal state within 60s "+
				"(zero-row parquet likely panicked datanode)", jobID)
		case <-ticker.C:
			progress, err := mc.GetRefreshExternalCollectionProgress(ctx,
				client.NewGetRefreshExternalCollectionProgressOption(jobID))
			require.NoError(t, err)
			t.Logf("Job %d: state=%s reason=%q", jobID, progress.State, progress.Reason)
			if progress.State == entity.RefreshStateCompleted ||
				progress.State == entity.RefreshStateFailed {
				finalState = progress.State
				finalReason = progress.Reason
				done = true
			}
		}
	}

	require.Equal(t, entity.RefreshStateFailed, finalState,
		"zero-row external source must surface as RefreshFailed, not silent Completed")
	require.Contains(t, strings.ToLower(finalReason), "zero",
		"failure reason should mention zero rows so operators can act; got %q", finalReason)
}

// Regression for issue #48637: a typo in external_field mapping (name
// points to a column that does not exist in the parquet) must surface
// the C++ "Column 'xxx' not found in schema" root cause in the
// RefreshFailed reason returned to the client, not the generic
// "sampling failed for all N segment(s)" message that forces operators
// to SSH into datanode logs.
func TestRefreshExternalCollectionWrongExternalField(t *testing.T) {
	ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	minioCfg := getMinIOConfig()
	minioClient, err := newMinIOClient(minioCfg)
	require.NoError(t, err)
	exists, err := minioClient.BucketExists(ctx, minioCfg.bucket)
	if err != nil || !exists {
		t.Skipf("MinIO bucket %q not accessible, skipping", minioCfg.bucket)
	}

	collName := common.GenRandomString("ext_wrong_field", 6)
	extPath := fmt.Sprintf("external-e2e-test/%s", collName)

	// Well-formed parquet with real columns id/value/embedding.
	data, err := generateParquetBytes(externalDataSchemaBasic, 8, 0, testVecDim)
	require.NoError(t, err)
	_, err = minioClient.PutObject(ctx, minioCfg.bucket,
		fmt.Sprintf("%s/data.parquet", extPath),
		bytes.NewReader(data), int64(len(data)),
		miniogo.PutObjectOptions{ContentType: "application/octet-stream"})
	require.NoError(t, err)
	t.Cleanup(func() {
		cleanupMinIOPrefix(context.Background(), minioClient, minioCfg.bucket,
			fmt.Sprintf("%s/", extPath))
	})

	// Schema maps "embedding" to a column "wrong_col_a" that does not exist.
	schema := entity.NewSchema().
		WithName(collName).
		WithExternalSource(extTestURI(minioCfg, extPath)).
		WithExternalSpec(extTestSpec(minioCfg, "parquet")).
		WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithExternalField("id")).
		WithField(entity.NewField().WithName("embedding").WithDataType(entity.FieldTypeFloatVector).
			WithDim(testVecDim).WithExternalField("wrong_col_a"))

	err = mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
	common.CheckErr(t, err, true)
	t.Cleanup(func() {
		_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(collName))
	})

	refreshResult, err := mc.RefreshExternalCollection(ctx,
		client.NewRefreshExternalCollectionOption(collName))
	common.CheckErr(t, err, true)
	jobID := refreshResult.JobID

	deadline := time.After(60 * time.Second)
	ticker := time.NewTicker(2 * time.Second)
	defer ticker.Stop()

	var finalState entity.RefreshExternalCollectionState
	var finalReason string
	for done := false; !done; {
		select {
		case <-deadline:
			t.Fatalf("Refresh job %d did not reach terminal state within 60s", jobID)
		case <-ticker.C:
			progress, err := mc.GetRefreshExternalCollectionProgress(ctx,
				client.NewGetRefreshExternalCollectionProgressOption(jobID))
			require.NoError(t, err)
			t.Logf("Job %d: state=%s reason=%q", jobID, progress.State, progress.Reason)
			if progress.State == entity.RefreshStateCompleted ||
				progress.State == entity.RefreshStateFailed {
				finalState = progress.State
				finalReason = progress.Reason
				done = true
			}
		}
	}

	require.Equal(t, entity.RefreshStateFailed, finalState,
		"wrong external_field must surface as RefreshFailed")
	// Root cause from C++ layer must reach the client.
	require.Contains(t, finalReason, "wrong_col_a",
		"reason must name the offending column; got %q", finalReason)
	require.Contains(t, strings.ToLower(finalReason), "not found",
		"reason must state the column was not found; got %q", finalReason)
	require.Contains(t, finalReason, "external_field mappings",
		"reason must hint at the real fix; got %q", finalReason)
}

// Security regression: DescribeCollection must redact secret extfs keys
// (access_key_id / access_key_value / ssl_ca_cert) in the ExternalSpec
// JSON returned to the client. Any caller with Describe privilege would
// otherwise be able to read another tenant's object-storage credentials.
func TestDescribeExternalCollectionRedactsCredentials(t *testing.T) {
	ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	minioCfg := getMinIOConfig()
	if _, err := newMinIOClient(minioCfg); err != nil {
		t.Skipf("MinIO unavailable, skipping: %v", err)
	}

	collName := common.GenRandomString("ext_redact", 6)
	extPath := fmt.Sprintf("external-e2e-test/%s", collName)

	// Embed secret credentials in extfs; they are persisted into etcd and
	// must NOT flow back out unredacted through DescribeCollection.
	rawSpec := fmt.Sprintf(
		`{"format":"parquet","extfs":{"access_key_id":%q,"access_key_value":%q}}`,
		"AKIAEXAMPLELEAKME", "supersecretvaluemustnotleak")

	schema := entity.NewSchema().
		WithName(collName).
		WithExternalSource(extTestURI(minioCfg, extPath)).
		WithExternalSpec(rawSpec).
		WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithExternalField("id")).
		WithField(entity.NewField().WithName("embedding").WithDataType(entity.FieldTypeFloatVector).
			WithDim(testVecDim).WithExternalField("embedding"))

	err := mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
	common.CheckErr(t, err, true)
	t.Cleanup(func() {
		_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(collName))
	})

	desc, err := mc.DescribeCollection(ctx, client.NewDescribeCollectionOption(collName))
	require.NoError(t, err)

	returnedSpec := desc.Schema.ExternalSpec
	t.Logf("returned ExternalSpec: %s", returnedSpec)

	require.NotContains(t, returnedSpec, "AKIAEXAMPLELEAKME",
		"access_key_id plaintext leaked via DescribeCollection")
	require.NotContains(t, returnedSpec, "supersecretvaluemustnotleak",
		"access_key_value plaintext leaked via DescribeCollection")
	require.Contains(t, returnedSpec, `"access_key_id":"***"`,
		"access_key_id must be redacted to ***")
	require.Contains(t, returnedSpec, `"access_key_value":"***"`,
		"access_key_value must be redacted to ***")
	// Non-secret values remain visible for operator troubleshooting.
	require.Contains(t, returnedSpec, `"format":"parquet"`,
		"non-secret spec values must remain readable")
}

// Regression for issue #49335: a refresh_external_collection call carrying
// an explicit (external_source, external_spec) override must persist the
// new tuple back into the collection schema, so that subsequent
// describe_collection reflects the new source AND a later reuse-path
// refresh (no override) targets the new source rather than silently
// reverting to the original one.
//
// Pre-fix behavior: the post-completion schema-sync path
// (server_external_schema_updater.updateExternalSchemaViaWAL) went through
// AlterCollection, which the atomic-tuple guard introduced in PR #49302
// rejected — the data load against NEW succeeded but the persisted
// schema stayed pointing at OLD forever.
func TestRefreshExternalCollectionOverridePersistsNewSource(t *testing.T) {
	ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	minioCfg := getMinIOConfig()
	minioClient, err := newMinIOClient(minioCfg)
	require.NoError(t, err)
	exists, err := minioClient.BucketExists(ctx, minioCfg.bucket)
	if err != nil || !exists {
		t.Skipf("MinIO bucket %q not accessible, skipping", minioCfg.bucket)
	}

	collName := common.GenRandomString("ext_override", 6)
	pathA := fmt.Sprintf("external-e2e-test/%s/a", collName)
	pathB := fmt.Sprintf("external-e2e-test/%s/b", collName)

	// Upload distinct parquet payloads to bucket/A and bucket/B.
	dataA, err := generateParquetBytes(externalDataSchemaBasic, 8, 0, testVecDim)
	require.NoError(t, err)
	dataB, err := generateParquetBytes(externalDataSchemaBasic, 8, 1000, testVecDim)
	require.NoError(t, err)
	for _, p := range []struct {
		path  string
		bytes []byte
	}{{pathA, dataA}, {pathB, dataB}} {
		_, err = minioClient.PutObject(ctx, minioCfg.bucket,
			fmt.Sprintf("%s/data.parquet", p.path),
			bytes.NewReader(p.bytes), int64(len(p.bytes)),
			miniogo.PutObjectOptions{ContentType: "application/octet-stream"})
		require.NoError(t, err)
	}
	t.Cleanup(func() {
		cleanupMinIOPrefix(context.Background(), minioClient, minioCfg.bucket,
			fmt.Sprintf("external-e2e-test/%s/", collName))
	})

	uriA := extTestURI(minioCfg, pathA)
	uriB := extTestURI(minioCfg, pathB)
	spec := extTestSpec(minioCfg, "parquet")

	// Create with SRC_A.
	schema := entity.NewSchema().
		WithName(collName).
		WithExternalSource(uriA).
		WithExternalSpec(spec).
		WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithExternalField("id")).
		WithField(entity.NewField().WithName("value").WithDataType(entity.FieldTypeFloat).WithExternalField("value")).
		WithField(entity.NewField().WithName("embedding").WithDataType(entity.FieldTypeFloatVector).
			WithDim(testVecDim).WithExternalField("embedding"))
	err = mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
	common.CheckErr(t, err, true)
	t.Cleanup(func() {
		_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(collName))
	})

	// Initial refresh against SRC_A (reuse path).
	refreshA, err := mc.RefreshExternalCollection(ctx,
		client.NewRefreshExternalCollectionOption(collName))
	common.CheckErr(t, err, true)
	waitRefreshTerminal(t, ctx, mc, refreshA.JobID, entity.RefreshStateCompleted)

	// Override refresh to SRC_B with explicit external_source.
	refreshB, err := mc.RefreshExternalCollection(ctx,
		client.NewRefreshExternalCollectionOption(collName).
			WithExternalSource(uriB).
			WithExternalSpec(spec))
	common.CheckErr(t, err, true)
	waitRefreshTerminal(t, ctx, mc, refreshB.JobID, entity.RefreshStateCompleted)

	// Persisted schema must now reflect SRC_B. Pre-fix this still showed pathA.
	desc, err := mc.DescribeCollection(ctx, client.NewDescribeCollectionOption(collName))
	require.NoError(t, err)
	require.Equal(t, uriB, desc.Schema.ExternalSource,
		"override refresh must persist new external_source; pre-fix #49335 left it stale")
}

func waitRefreshTerminal(t *testing.T, ctx context.Context, mc *base.MilvusClient, jobID int64, expect entity.RefreshExternalCollectionState) {
	t.Helper()
	deadline := time.After(60 * time.Second)
	ticker := time.NewTicker(2 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-deadline:
			t.Fatalf("refresh job %d did not reach terminal state within 60s", jobID)
		case <-ticker.C:
			progress, err := mc.GetRefreshExternalCollectionProgress(ctx,
				client.NewGetRefreshExternalCollectionProgressOption(jobID))
			require.NoError(t, err)
			t.Logf("Job %d: state=%s reason=%q", jobID, progress.State, progress.Reason)
			if progress.State == entity.RefreshStateCompleted ||
				progress.State == entity.RefreshStateFailed {
				require.Equal(t, expect, progress.State,
					"refresh terminal state mismatch (reason=%s)", progress.Reason)
				return
			}
		}
	}
}

func TestRefreshExternalCollectionMilvusTableSnapshot(t *testing.T) {
	ctx := hp.CreateContext(t, 5*time.Minute)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	const rowCount = 5000
	const deletedCount = 500
	sourceName := common.GenRandomString("mt_src", 6)
	sourceSchema := entity.NewSchema().
		WithName(sourceName).
		WithField(entity.NewField().
			WithName("id").
			WithDataType(entity.FieldTypeInt64).
			WithIsPrimaryKey(true)).
		WithField(entity.NewField().
			WithName("embedding").
			WithDataType(entity.FieldTypeFloatVector).
			WithDim(testVecDim))

	require.NoError(t, mc.CreateCollection(ctx, client.NewCreateCollectionOption(sourceName, sourceSchema)))
	t.Cleanup(func() {
		_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(sourceName))
	})

	ids := make([]int64, rowCount)
	vectors := make([][]float32, rowCount)
	for i := 0; i < rowCount; i++ {
		ids[i] = int64(i)
		vectors[i] = []float32{float32(i), float32(i) + 1, float32(i) + 2, float32(i) + 3}
	}
	_, err := mc.Insert(ctx, client.NewColumnBasedInsertOption(sourceName,
		column.NewColumnInt64("id", ids),
		column.NewColumnFloatVector("embedding", testVecDim, vectors)))
	require.NoError(t, err)

	flushTask, err := mc.Flush(ctx, client.NewFlushOption(sourceName))
	require.NoError(t, err)
	require.NoError(t, flushTask.Await(ctx))

	deletedIDs := make([]int64, deletedCount)
	for i := 0; i < deletedCount; i++ {
		deletedIDs[i] = int64(i)
	}
	delRes, err := mc.Delete(ctx, client.NewDeleteOption(sourceName).WithInt64IDs("id", deletedIDs))
	require.NoError(t, err)
	require.Equal(t, int64(deletedCount), delRes.DeleteCount)

	// Wait for the standalone flush rate limiter to reset before flushing
	// the delete log. The local default is 0.1 flush requests per second.
	time.Sleep(10 * time.Second)

	flushTask, err = mc.Flush(ctx, client.NewFlushOption(sourceName))
	require.NoError(t, err)
	require.NoError(t, flushTask.Await(ctx))

	sourceDesc, err := mc.DescribeCollection(ctx, client.NewDescribeCollectionOption(sourceName))
	require.NoError(t, err)

	snapshotName := common.GenRandomString("mt_snap", 6)
	require.NoError(t, mc.CreateSnapshot(ctx, client.NewCreateSnapshotOption(snapshotName, sourceName).
		WithDescription("milvus-table external snapshot e2e")))
	t.Cleanup(func() {
		_ = mc.DropSnapshot(context.Background(), client.NewDropSnapshotOption(snapshotName, sourceName))
	})

	snapshotInfo, err := mc.DescribeSnapshot(ctx, client.NewDescribeSnapshotOption(snapshotName, sourceName))
	require.NoError(t, err)
	require.NotEmpty(t, snapshotInfo.GetS3Location())

	minioCfg := getMinIOConfig()
	externalSource := extTestURI(minioCfg, snapshotInfo.GetS3Location())
	externalSpec := extTestSpec(minioCfg, "milvus-table")

	externalName := common.GenRandomString("mt_ext", 6)
	externalSchema := entity.NewSchema().
		WithName(externalName).
		WithExternalSource(externalSource).
		WithExternalSpec(externalSpec).
		WithField(entity.NewField().
			WithName("id").
			WithDataType(entity.FieldTypeInt64).
			WithIsPrimaryKey(true).
			WithExternalField("id")).
		WithField(entity.NewField().
			WithName("embedding").
			WithDataType(entity.FieldTypeFloatVector).
			WithDim(testVecDim).
			WithExternalField("embedding"))

	require.NoError(t, mc.CreateCollection(ctx, client.NewCreateCollectionOption(externalName, externalSchema)))
	t.Cleanup(func() {
		_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(externalName))
	})

	refresh, err := mc.RefreshExternalCollection(ctx, client.NewRefreshExternalCollectionOption(externalName))
	require.NoError(t, err)
	waitRefreshTerminal(t, ctx, mc, refresh.JobID, entity.RefreshStateCompleted)

	idxTask, err := mc.CreateIndex(ctx,
		client.NewCreateIndexOption(externalName, "embedding", index.NewFlatIndex(entity.L2)))
	require.NoError(t, err)
	require.NoError(t, idxTask.Await(ctx))

	loadTask, err := mc.LoadCollection(ctx, client.NewLoadCollectionOption(externalName))
	require.NoError(t, err)
	require.NoError(t, loadTask.Await(ctx))

	countRes, err := mc.Query(ctx, client.NewQueryOption(externalName).
		WithOutputFields(common.QueryCountFieldName).
		WithConsistencyLevel(entity.ClStrong))
	require.NoError(t, err)
	count, err := countRes.GetColumn(common.QueryCountFieldName).GetAsInt64(0)
	require.NoError(t, err)
	require.Equal(t, int64(rowCount-deletedCount), count)

	searchRes, err := mc.Search(ctx,
		client.NewSearchOption(externalName, 5, []entity.Vector{entity.FloatVector(vectors[deletedCount])}).
			WithOutputFields("id"))
	require.NoError(t, err)
	require.NotEmpty(t, searchRes)
	require.Equal(t, 5, searchRes[0].ResultCount)
	idCol := searchRes[0].GetColumn("id")
	require.NotNil(t, idCol)
	require.Equal(t, searchRes[0].ResultCount, idCol.Len())
	for i := 0; i < idCol.Len(); i++ {
		id, err := idCol.GetAsInt64(i)
		require.NoError(t, err)
		require.GreaterOrEqual(t, id, int64(deletedCount))
	}
	nearestID, err := idCol.GetAsInt64(0)
	require.NoError(t, err)
	require.Equal(t, int64(deletedCount), nearestID)

	t.Logf("milvus-table snapshot e2e passed: sourceCollectionID=%d snapshot=%s externalSource=%s",
		sourceDesc.ID, snapshotName, externalSource)
}

func TestRefreshExternalCollectionMilvusTableSnapshotVirtualPK(t *testing.T) {
	ctx := hp.CreateContext(t, 5*time.Minute)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	const rowCount = 5000
	const deletedCount = 500
	const targetRow = 1234
	sourceName := common.GenRandomString("mt_vpk_src", 6)
	sourceSchema := entity.NewSchema().
		WithName(sourceName).
		WithField(entity.NewField().
			WithName("id").
			WithDataType(entity.FieldTypeInt64).
			WithIsPrimaryKey(true)).
		WithField(entity.NewField().
			WithName("embedding").
			WithDataType(entity.FieldTypeFloatVector).
			WithDim(testVecDim))

	require.NoError(t, mc.CreateCollection(ctx, client.NewCreateCollectionOption(sourceName, sourceSchema)))
	t.Cleanup(func() {
		_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(sourceName))
	})

	ids := make([]int64, rowCount)
	vectors := make([][]float32, rowCount)
	for i := 0; i < rowCount; i++ {
		ids[i] = int64(i)
		vectors[i] = []float32{float32(i), float32(i) + 1, float32(i) + 2, float32(i) + 3}
	}
	_, err := mc.Insert(ctx, client.NewColumnBasedInsertOption(sourceName,
		column.NewColumnInt64("id", ids),
		column.NewColumnFloatVector("embedding", testVecDim, vectors)))
	require.NoError(t, err)

	flushTask, err := mc.Flush(ctx, client.NewFlushOption(sourceName))
	require.NoError(t, err)
	require.NoError(t, flushTask.Await(ctx))

	deletedIDs := make([]int64, deletedCount)
	for i := 0; i < deletedCount; i++ {
		deletedIDs[i] = int64(i)
	}
	delRes, err := mc.Delete(ctx, client.NewDeleteOption(sourceName).WithInt64IDs("id", deletedIDs))
	require.NoError(t, err)
	require.Equal(t, int64(deletedCount), delRes.DeleteCount)

	// Wait for the standalone flush rate limiter to reset before flushing
	// the delete log. The local default is 0.1 flush requests per second.
	time.Sleep(10 * time.Second)

	flushTask, err = mc.Flush(ctx, client.NewFlushOption(sourceName))
	require.NoError(t, err)
	require.NoError(t, flushTask.Await(ctx))

	sourceDesc, err := mc.DescribeCollection(ctx, client.NewDescribeCollectionOption(sourceName))
	require.NoError(t, err)

	snapshotName := common.GenRandomString("mt_vpk_snap", 6)
	require.NoError(t, mc.CreateSnapshot(ctx, client.NewCreateSnapshotOption(snapshotName, sourceName).
		WithDescription("milvus-table external snapshot virtual pk e2e")))
	t.Cleanup(func() {
		_ = mc.DropSnapshot(context.Background(), client.NewDropSnapshotOption(snapshotName, sourceName))
	})

	snapshotInfo, err := mc.DescribeSnapshot(ctx, client.NewDescribeSnapshotOption(snapshotName, sourceName))
	require.NoError(t, err)
	require.NotEmpty(t, snapshotInfo.GetS3Location())

	minioCfg := getMinIOConfig()
	externalSource := extTestURI(minioCfg, snapshotInfo.GetS3Location())
	externalSpec := extTestSpec(minioCfg, "milvus-table")

	externalName := common.GenRandomString("mt_vpk_ext", 6)
	externalSchema := entity.NewSchema().
		WithName(externalName).
		WithExternalSource(externalSource).
		WithExternalSpec(externalSpec).
		WithField(entity.NewField().
			WithName("source_id").
			WithDataType(entity.FieldTypeInt64).
			WithExternalField("id")).
		WithField(entity.NewField().
			WithName("embedding").
			WithDataType(entity.FieldTypeFloatVector).
			WithDim(testVecDim).
			WithExternalField("embedding"))

	require.NoError(t, mc.CreateCollection(ctx, client.NewCreateCollectionOption(externalName, externalSchema)))
	t.Cleanup(func() {
		_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(externalName))
	})

	refresh, err := mc.RefreshExternalCollection(ctx, client.NewRefreshExternalCollectionOption(externalName))
	require.NoError(t, err)
	waitRefreshTerminal(t, ctx, mc, refresh.JobID, entity.RefreshStateCompleted)

	idxTask, err := mc.CreateIndex(ctx,
		client.NewCreateIndexOption(externalName, "embedding", index.NewFlatIndex(entity.L2)))
	require.NoError(t, err)
	require.NoError(t, idxTask.Await(ctx))

	loadTask, err := mc.LoadCollection(ctx, client.NewLoadCollectionOption(externalName))
	require.NoError(t, err)
	require.NoError(t, loadTask.Await(ctx))

	countRes, err := mc.Query(ctx, client.NewQueryOption(externalName).
		WithOutputFields(common.QueryCountFieldName).
		WithConsistencyLevel(entity.ClStrong))
	require.NoError(t, err)
	count, err := countRes.GetColumn(common.QueryCountFieldName).GetAsInt64(0)
	require.NoError(t, err)
	require.Equal(t, int64(rowCount-deletedCount), count)

	deletedQueryRes, err := mc.Query(ctx, client.NewQueryOption(externalName).
		WithFilter("source_id in [0,1,499]").
		WithOutputFields("source_id").
		WithConsistencyLevel(entity.ClStrong))
	require.NoError(t, err)
	require.Zero(t, deletedQueryRes.ResultCount)
	deletedQuerySourceIDCol := deletedQueryRes.GetColumn("source_id")
	require.NotNil(t, deletedQuerySourceIDCol)
	require.Zero(t, deletedQuerySourceIDCol.Len())

	deletedSearchRes, err := mc.Search(ctx,
		client.NewSearchOption(externalName, 5, []entity.Vector{entity.FloatVector(vectors[0])}).
			WithOutputFields("source_id"))
	require.NoError(t, err)
	require.Len(t, deletedSearchRes, 1)
	require.Equal(t, 5, deletedSearchRes[0].ResultCount)
	deletedSearchSourceIDCol := deletedSearchRes[0].GetColumn("source_id")
	require.NotNil(t, deletedSearchSourceIDCol)
	for i := 0; i < deletedSearchSourceIDCol.Len(); i++ {
		sourceID, err := deletedSearchSourceIDCol.GetAsInt64(i)
		require.NoError(t, err)
		require.GreaterOrEqual(t, sourceID, int64(deletedCount))
	}

	searchRes, err := mc.Search(ctx,
		client.NewSearchOption(externalName, 1, []entity.Vector{entity.FloatVector(vectors[targetRow])}).
			WithOutputFields("source_id"))
	require.NoError(t, err)
	require.Len(t, searchRes, 1)
	require.Equal(t, 1, searchRes[0].ResultCount)
	require.NotNil(t, searchRes[0].IDs)
	virtualID, err := searchRes[0].IDs.GetAsInt64(0)
	require.NoError(t, err)
	sourceID, err := searchRes[0].GetColumn("source_id").GetAsInt64(0)
	require.NoError(t, err)
	require.Equal(t, int64(targetRow), sourceID)
	require.NotEqual(t, sourceID, virtualID,
		"search IDs should be synthesized virtual PKs, not source real PKs")

	t.Logf("milvus-table virtual pk snapshot e2e passed: sourceCollectionID=%d snapshot=%s externalSource=%s",
		sourceDesc.ID, snapshotName, externalSource)
}

// Regression for the explore-manifest index-drift bug: the source
// directory is intentionally polluted with non-parquet strays
// (`_SUCCESS`, `.crc`, `README.md`) interleaved with valid parquet
// shards. Pre-fix, DataCoord filtered them out before slicing tasks
// while DataNode read the unfiltered manifest and sliced against the
// stale index space — picking `_SUCCESS` as a "parquet" file and
// failing with "Invalid parquet magic" or silently dropping a real
// shard. Post-fix, both layers normalize (sort + format-filter) the
// manifest identically, so refresh completes and total row count
// matches the sum of the parquet shards only.
func TestRefreshExternalCollectionFiltersNonParquetStrays(t *testing.T) {
	ctx := hp.CreateContext(t, time.Second*common.DefaultTimeout)
	mc := hp.CreateDefaultMilvusClient(ctx, t)

	minioCfg := getMinIOConfig()
	minioClient, err := newMinIOClient(minioCfg)
	require.NoError(t, err)
	exists, err := minioClient.BucketExists(ctx, minioCfg.bucket)
	if err != nil || !exists {
		t.Skipf("MinIO bucket %q not accessible, skipping", minioCfg.bucket)
	}

	collName := common.GenRandomString("ext_strays", 6)
	extPath := fmt.Sprintf("external-e2e-test/%s", collName)

	// 3 valid parquet shards, varying sizes to detect index drift.
	const rowsPerShard = int64(8)
	const numShards = 3
	for i := 0; i < numShards; i++ {
		data, err := generateParquetBytes(externalDataSchemaBasic, rowsPerShard, int64(i)*rowsPerShard, testVecDim)
		require.NoError(t, err)
		_, err = minioClient.PutObject(ctx, minioCfg.bucket,
			fmt.Sprintf("%s/part-%05d.parquet", extPath, i),
			bytes.NewReader(data), int64(len(data)),
			miniogo.PutObjectOptions{ContentType: "application/octet-stream"})
		require.NoError(t, err)
	}
	// Stray non-parquet files interleaved by lex order with the parquets.
	strays := map[string][]byte{
		"_SUCCESS":    []byte(""),
		"_committed":  []byte("xyz"),
		"part-00.crc": []byte("crc-checksum-bytes"),
		"README.md":   []byte("# this dir\n"),
		"metadata":    []byte(`{"v":1}`),
	}
	for name, payload := range strays {
		_, err = minioClient.PutObject(ctx, minioCfg.bucket,
			fmt.Sprintf("%s/%s", extPath, name),
			bytes.NewReader(payload), int64(len(payload)),
			miniogo.PutObjectOptions{ContentType: "application/octet-stream"})
		require.NoError(t, err)
	}
	t.Cleanup(func() {
		cleanupMinIOPrefix(context.Background(), minioClient, minioCfg.bucket,
			fmt.Sprintf("%s/", extPath))
	})

	schema := entity.NewSchema().
		WithName(collName).
		WithExternalSource(extTestURI(minioCfg, extPath)).
		WithExternalSpec(extTestSpec(minioCfg, "parquet")).
		WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithExternalField("id")).
		WithField(entity.NewField().WithName("value").WithDataType(entity.FieldTypeFloat).WithExternalField("value")).
		WithField(entity.NewField().WithName("embedding").WithDataType(entity.FieldTypeFloatVector).
			WithDim(testVecDim).WithExternalField("embedding"))
	err = mc.CreateCollection(ctx, client.NewCreateCollectionOption(collName, schema))
	common.CheckErr(t, err, true)
	t.Cleanup(func() {
		_ = mc.DropCollection(context.Background(), client.NewDropCollectionOption(collName))
	})

	refresh, err := mc.RefreshExternalCollection(ctx,
		client.NewRefreshExternalCollectionOption(collName))
	common.CheckErr(t, err, true)
	waitRefreshTerminal(t, ctx, mc, refresh.JobID, entity.RefreshStateCompleted)

	// Load + count to assert no shard was dropped and no stray was read
	// (a stray would either fail load or — worse — succeed silently with
	// the wrong row count if the bug let mis-sliced indices slip
	// through).
	idxOpt := client.NewCreateIndexOption(collName, "embedding", index.NewAutoIndex(entity.MetricType("L2")))
	idxTask, err := mc.CreateIndex(ctx, idxOpt)
	common.CheckErr(t, err, true)
	require.NoError(t, idxTask.Await(ctx))
	loadTask, err := mc.LoadCollection(ctx, client.NewLoadCollectionOption(collName))
	common.CheckErr(t, err, true)
	require.NoError(t, loadTask.Await(ctx))

	res, err := mc.Query(ctx, client.NewQueryOption(collName).
		WithFilter("").
		WithOutputFields(common.QueryCountFieldName).
		WithConsistencyLevel(entity.ClStrong))
	require.NoError(t, err)
	require.Greater(t, len(res.Fields), 0)
	got := res.Fields[0].FieldData().GetScalars().GetLongData().GetData()
	require.Len(t, got, 1)
	expected := rowsPerShard * numShards
	require.Equal(t, expected, got[0],
		"row count must equal sum of parquet shards (%d); strays must be filtered, no shard may be dropped",
		expected)
}