milvus-io--milvus/tests/python_client/milvus_client/test_milvus_client_optimize.py

"""
Optimize API Test Cases

The optimize() method is a high-level sugar API wrapping force merge compaction.
It performs: wait for indexes -> force merge compaction -> wait for compaction ->
wait for index rebuild -> refresh load (if loaded).

L3 tests require Milvus configuration changes:
    dataCoord:
      segment:
        maxSize: 64  # MB, default is 1024
      compaction:
        enableAutoCompaction: false
"""

import time

import numpy as np
import pytest
from base.client_v2_base import TestMilvusClientV2Base
from common import common_func as cf
from common import common_type as ct
from common.common_type import CaseLabel, CheckTasks
from common.constants import *  # noqa: F403
from pymilvus import DataType
from utils.util_log import test_log as log
from utils.util_pymilvus import *  # noqa: F403

prefix = "client_optimize"
epsilon = ct.epsilon
default_nb = ct.default_nb
default_nb_medium = ct.default_nb_medium
default_nq = ct.default_nq
default_dim = ct.default_dim
default_limit = ct.default_limit
default_search_exp = "id >= 0"
exp_res = "exp_res"
default_search_field = ct.default_float_vec_field_name
default_search_params = ct.default_search_params
default_primary_key_field_name = "id"
default_vector_field_name = "vector"
default_float_field_name = ct.default_float_field_name
default_string_field_name = ct.default_string_field_name


class TestMilvusClientOptimizeInvalid(TestMilvusClientV2Base):
    """Test cases for optimize() with invalid parameters"""

    """
    ******************************************************************
    #  The following are invalid base cases
    ******************************************************************
    """

    @pytest.mark.tags(CaseLabel.L1)
    @pytest.mark.parametrize("target_size", ["abc", "1XB", "MB100", "1.2.3GB", "--1GB"])
    def test_optimize_invalid_target_size_format(self, target_size):
        """
        target: test optimize with invalid target_size string format
        method: create collection, call optimize with malformed target_size
        expected: Raise ParamError from client-side parse_target_size
        """
        client = self._client()
        collection_name = cf.gen_unique_str(prefix)
        self.create_collection(client, collection_name, default_dim)
        error = {ct.err_code: 1, ct.err_msg: "Invalid"}
        self.optimize(
            client,
            collection_name,
            target_size=target_size,
            check_task=CheckTasks.err_res,
            check_items=error,
        )

    @pytest.mark.tags(CaseLabel.L1)
    @pytest.mark.parametrize("target_size", ["0MB", "0GB", "0B", "100B", "500KB"])
    def test_optimize_target_size_too_small(self, target_size):
        """
        target: test optimize with target_size that resolves to less than 1MB
        method: create collection, call optimize with tiny target_size
        expected: Raise ParamError (target size too small)
        """
        client = self._client()
        collection_name = cf.gen_unique_str(prefix)
        self.create_collection(client, collection_name, default_dim)
        error = {ct.err_code: 1, ct.err_msg: "target size too small"}
        self.optimize(
            client,
            collection_name,
            target_size=target_size,
            check_task=CheckTasks.err_res,
            check_items=error,
        )

    @pytest.mark.tags(CaseLabel.L1)
    def test_optimize_nonexistent_collection(self):
        """
        target: test optimize on a non-existent collection
        method: call optimize on a collection that doesn't exist
        expected: Raise exception (collection not found)
        """
        client = self._client()
        collection_name = cf.gen_unique_str(prefix)
        error = {ct.err_code: 0, ct.err_msg: "can't find collection"}
        self.optimize(
            client,
            collection_name,
            target_size="1GB",
            check_task=CheckTasks.err_res,
            check_items=error,
        )

    @pytest.mark.tags(CaseLabel.L1)
    def test_optimize_empty_collection_name(self):
        """
        target: test optimize with empty collection_name
        method: call optimize with empty string
        expected: Raise exception
        """
        client = self._client()
        error = {ct.err_code: 1, ct.err_msg: "collection_name"}
        self.optimize(
            client,
            "",
            target_size="1GB",
            check_task=CheckTasks.err_res,
            check_items=error,
        )

    @pytest.mark.tags(CaseLabel.L1)
    def test_optimize_invalid_collection_name(self):
        """
        target: test optimize with invalid collection_name (blank space)
        method: call optimize with whitespace collection_name
        expected: Raise exception (invalid collection name)
        """
        client = self._client()
        error = {ct.err_code: 1100, ct.err_msg: "Invalid collection name"}
        self.optimize(
            client,
            " ",
            target_size="1GB",
            check_task=CheckTasks.err_res,
            check_items=error,
        )

    @pytest.mark.tags(CaseLabel.L1)
    @pytest.mark.parametrize("target_size", [[], {}, (1, 2)])
    def test_optimize_invalid_target_size_type(self, target_size):
        """
        target: test optimize with invalid target_size type
        method: call optimize with non-string/non-numeric target_size
        expected: Raise ParamError
        """
        client = self._client()
        collection_name = cf.gen_unique_str(prefix)
        self.create_collection(client, collection_name, default_dim)
        error = {ct.err_code: 1, ct.err_msg: "target_size must be a string or number"}
        self.optimize(
            client,
            collection_name,
            target_size=target_size,
            check_task=CheckTasks.err_res,
            check_items=error,
        )


class TestMilvusClientOptimizeValid(TestMilvusClientV2Base):
    """Test cases for optimize() with valid parameters"""

    """
    ******************************************************************
    #  The following are valid base cases
    ******************************************************************
    """

    @pytest.mark.tags(CaseLabel.L1)
    def test_optimize_empty_collection(self):
        """
        target: test optimize on an empty collection
        method: create collection, call optimize with wait=True
        expected: Returns OptimizeResult with status="success"
        """
        client = self._client()
        collection_name = cf.gen_unique_str(prefix)
        self.create_collection(client, collection_name, default_dim)
        result = self.optimize(client, collection_name, target_size="1GB")[0]
        assert result.status == "success"
        assert result.collection_name == collection_name
        # Empty collection may return compaction_id=-1 (no segments to compact)
        assert isinstance(result.compaction_id, int)
        log.info(f"Optimize on empty collection completed: {result}")

    @pytest.mark.tags(CaseLabel.L3)
    def test_optimize_default_target_size(self):
        """
        target: test optimize with default target_size (None)
        method: create collection, insert data, flush, optimize without target_size
        expected: Compaction completes successfully with auto-calculated size
        note: L3 - requires config change (segment.maxSize=64MB)
        """
        client = self._client()
        collection_name = cf.gen_unique_str(prefix)
        dim = 128
        self.create_collection(client, collection_name, dim)
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, dim))[0]),
            }
            for i in range(default_nb)
        ]
        self.insert(client, collection_name, rows)
        self.flush(client, collection_name)
        result = self.optimize(client, collection_name, timeout=300)[0]
        assert result.status == "success"
        assert result.collection_name == collection_name
        assert result.compaction_id > 0
        log.info(f"Optimize with default target_size completed: {result}")

    @pytest.mark.tags(CaseLabel.L3)
    @pytest.mark.parametrize("target_size", ["512MB", "1GB", "2GB"])
    def test_optimize_explicit_target_size(self, target_size):
        """
        target: test optimize with various explicit target_size strings
        method: create collection, insert data, flush, optimize with target_size
        expected: Compaction completes successfully
        note: L3 - requires config change (segment.maxSize=64MB)
        """
        client = self._client()
        collection_name = cf.gen_unique_str(prefix)
        dim = 128
        self.create_collection(client, collection_name, dim)
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, dim))[0]),
            }
            for i in range(default_nb)
        ]
        self.insert(client, collection_name, rows)
        self.flush(client, collection_name)
        result = self.optimize(client, collection_name, target_size=target_size, timeout=300)[0]
        assert result.status == "success"
        assert result.collection_name == collection_name
        assert result.target_size == target_size
        log.info(f"Optimize with target_size={target_size} completed: {result}")

    @pytest.mark.tags(CaseLabel.L3)
    def test_optimize_result_fields(self):
        """
        target: test optimize result contains all expected fields
        method: create collection, insert data, flush, optimize, verify result fields
        expected: OptimizeResult has status, collection_name, compaction_id, target_size, progress
        note: L3 - requires config change (segment.maxSize=64MB)
        """
        client = self._client()
        collection_name = cf.gen_unique_str(prefix)
        dim = 128
        self.create_collection(client, collection_name, dim)
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, dim))[0]),
            }
            for i in range(default_nb)
        ]
        self.insert(client, collection_name, rows)
        self.flush(client, collection_name)
        target_size = "1GB"
        result = self.optimize(client, collection_name, target_size=target_size, timeout=300)[0]
        # Verify all result fields
        assert result.status == "success"
        assert result.collection_name == collection_name
        assert isinstance(result.compaction_id, int)
        assert result.compaction_id > 0
        assert result.target_size == target_size
        assert isinstance(result.progress, list)
        assert len(result.progress) > 0
        log.info(f"Optimize result fields verified: progress={result.progress}")

    @pytest.mark.tags(CaseLabel.L3)
    def test_optimize_with_multiple_segments(self):
        """
        target: test optimize merges multiple segments
        method: create collection, insert in batches with flush to create segments, optimize
        expected: Compaction completes, segment count reduced
        note: L3 - requires config change (segment.maxSize=64MB)
        """
        client = self._client()
        collection_name = cf.gen_unique_str(prefix)
        dim = 128
        self.create_collection(client, collection_name, dim)
        rng = np.random.default_rng(seed=19530)
        num_batches = 5
        batch_size = default_nb
        for batch in range(num_batches):
            rows = [
                {
                    default_primary_key_field_name: batch * batch_size + i,
                    default_vector_field_name: list(rng.random((1, dim))[0]),
                }
                for i in range(batch_size)
            ]
            self.insert(client, collection_name, rows)
            self.flush(client, collection_name)
            log.info(f"Inserted batch {batch + 1}/{num_batches}")

        result = self.optimize(client, collection_name, target_size="2GB", timeout=600)[0]
        assert result.status == "success"
        log.info(f"Optimize with {num_batches} batches completed: compaction_id={result.compaction_id}")

    @pytest.mark.tags(CaseLabel.L3)
    def test_optimize_search_after(self):
        """
        target: test search works correctly after optimize
        method: create collection, insert data, flush, optimize, search
        expected: Search returns correct results
        note: L3 - requires config change (segment.maxSize=64MB)
        """
        client = self._client()
        collection_name = cf.gen_unique_str(prefix)
        dim = 128
        nb = default_nb
        self.create_collection(client, collection_name, dim)
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, dim))[0]),
            }
            for i in range(nb)
        ]
        self.insert(client, collection_name, rows)
        self.flush(client, collection_name)
        # Optimize (includes refresh_load if loaded)
        result = self.optimize(client, collection_name, target_size="2GB", timeout=300)[0]
        assert result.status == "success"
        # Search after optimize
        search_vectors = rng.random((1, dim))
        search_res = self.search(
            client,
            collection_name,
            list(search_vectors),
            limit=10,
            output_fields=[default_primary_key_field_name],
        )[0]
        log.info(f"Search results: {search_res}")
        assert len(search_res) == 1
        assert len(search_res[0]) == 10
        log.info(f"Search after optimize returned {len(search_res[0])} results")

    @pytest.mark.tags(CaseLabel.L3)
    def test_optimize_wait_false_task_tracking(self):
        """
        target: test optimize with wait=False returns OptimizeTask with progress tracking
        method: create collection, insert data, flush, optimize with wait=False, track progress
        expected: Task completes, progress stages are recorded
        note: L3 - requires config change (segment.maxSize=64MB)
        """
        client = self._client()
        collection_name = cf.gen_unique_str(prefix)
        dim = 128
        self.create_collection(client, collection_name, dim)
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, dim))[0]),
            }
            for i in range(default_nb)
        ]
        self.insert(client, collection_name, rows)
        self.flush(client, collection_name)
        # Call optimize with wait=False to get OptimizeTask
        task = self.optimize(client, collection_name, target_size="1GB", wait=False)[0]
        # Track progress
        cost = 300
        start = time.time()
        while not task.done():
            progress = task.progress()
            log.info(f"Optimize progress: {progress}")
            time.sleep(2)
            if time.time() - start > cost:
                raise Exception(f"Optimize task cost more than {cost}s")
        # Get result
        result = task.result(timeout=10)
        assert result.status == "success"
        assert result.collection_name == collection_name
        # Verify progress history
        history = task.progress_history()
        assert len(history) > 0
        log.info(f"Optimize task completed with progress history: {history}")

    @pytest.mark.tags(CaseLabel.L3)
    def test_optimize_wait_false_cancel(self):
        """
        target: test cancelling an optimize task
        method: create collection, insert data, flush, start optimize with wait=False, cancel it
        expected: Task is cancelled, result raises MilvusException
        note: L3 - requires config change (segment.maxSize=64MB)
        """
        client = self._client()
        collection_name = cf.gen_unique_str(prefix)
        dim = 128
        self.create_collection(client, collection_name, dim)
        rng = np.random.default_rng(seed=19530)
        num_batches = 5
        for batch in range(num_batches):
            rows = [
                {
                    default_primary_key_field_name: batch * 1000 + i,
                    default_vector_field_name: list(rng.random((1, dim))[0]),
                }
                for i in range(1000)
            ]
            self.insert(client, collection_name, rows)
            self.flush(client, collection_name)
        # Start optimize with wait=False
        task = self.optimize(client, collection_name, target_size="2GB", wait=False)[0]
        # Wait a moment then cancel
        time.sleep(2)
        cancelled = task.cancel()
        log.info(f"Task cancel result: {cancelled}")
        assert task.cancelled()
        log.info("Optimize task cancelled successfully")

    @pytest.mark.tags(CaseLabel.L3)
    def test_optimize_verify_segment_count(self):
        """
        target: test optimize reduces segment count
        method: create collection, insert in batches, check segments before/after optimize
        expected: Fewer segments after optimize
        note: L3 - requires config change (segment.maxSize=64MB)
        """
        client = self._client()
        collection_name = cf.gen_unique_str(prefix)
        dim = 128
        self.create_collection(client, collection_name, dim)
        rng = np.random.default_rng(seed=19530)
        num_batches = 5
        batch_size = default_nb
        for batch in range(num_batches):
            rows = [
                {
                    default_primary_key_field_name: batch * batch_size + i,
                    default_vector_field_name: list(rng.random((1, dim))[0]),
                }
                for i in range(batch_size)
            ]
            self.insert(client, collection_name, rows)
            self.flush(client, collection_name)

        # Get stable segment count before optimize
        assert self.wait_for_index_ready(client, collection_name, default_vector_field_name, timeout=300)
        self.release_collection(client, collection_name)
        self.load_collection(client, collection_name)
        segments_before = client.list_loaded_segments(collection_name)
        segment_count_before = len(segments_before)
        log.info(f"Segment count before optimize: {segment_count_before}")

        # Optimize (handles compaction + index rebuild + refresh load)
        result = self.optimize(client, collection_name, target_size="2GB", timeout=600)[0]
        assert result.status == "success"

        # Release and reload to get updated segment info
        assert self.wait_for_index_ready(client, collection_name, default_vector_field_name, timeout=300)
        self.release_collection(client, collection_name)
        self.load_collection(client, collection_name)
        segments_after = client.list_loaded_segments(collection_name)
        segment_count_after = len(segments_after)
        log.info(f"Segment count after optimize: {segment_count_after}")

        assert segment_count_after <= segment_count_before, (
            f"Expected fewer segments after optimize, got {segment_count_after} >= {segment_count_before}"
        )
        log.info(f"Optimize reduced segments from {segment_count_before} to {segment_count_after}")

    @pytest.mark.tags(CaseLabel.L3)
    def test_optimize_numeric_target_size(self):
        """
        target: test optimize with numeric target_size (bytes)
        method: create collection, insert data, flush, optimize with int target_size
        expected: Compaction completes, parse_target_size treats number as bytes
        note: L3 - requires config change (segment.maxSize=64MB)
        """
        client = self._client()
        collection_name = cf.gen_unique_str(prefix)
        dim = 128
        self.create_collection(client, collection_name, dim)
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, dim))[0]),
            }
            for i in range(default_nb)
        ]
        self.insert(client, collection_name, rows)
        self.flush(client, collection_name)
        # 1GB in bytes
        target_size_bytes = 1024 * 1024 * 1024
        result = self.optimize(client, collection_name, target_size=target_size_bytes, timeout=300)[0]
        assert result.status == "success"
        log.info(f"Optimize with numeric target_size={target_size_bytes} completed")

    @pytest.mark.tags(CaseLabel.L3)
    def test_optimize_with_clustering_key(self):
        """
        target: test optimize on collection with clustering key
        method: create collection with clustering key, insert data, optimize
        expected: Optimize completes successfully
        note: L3 - requires config change (segment.maxSize=64MB)
        """
        client = self._client()
        collection_name = cf.gen_unique_str(prefix)
        dim = 128
        schema = self.create_schema(client, enable_dynamic_field=False)[0]
        schema.add_field(
            default_primary_key_field_name,
            DataType.INT64,
            is_primary=True,
            auto_id=False,
        )
        schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=dim)
        schema.add_field(
            default_string_field_name,
            DataType.VARCHAR,
            max_length=64,
            is_clustering_key=True,
        )
        index_params = self.prepare_index_params(client)[0]
        index_params.add_index(default_vector_field_name, metric_type="COSINE")
        self.create_collection(
            client,
            collection_name,
            dimension=dim,
            schema=schema,
            index_params=index_params,
        )
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, dim))[0]),
                default_string_field_name: f"str_{i}",
            }
            for i in range(default_nb)
        ]
        self.insert(client, collection_name, rows)
        self.flush(client, collection_name)
        result = self.optimize(client, collection_name, target_size="2GB", timeout=300)[0]
        assert result.status == "success"
        log.info(f"Optimize with clustering key completed: {result}")