feat: support medium sync policy in CCR tasks

[Ryan19929]

Overview

This PR implements storage medium sync policy support for ccr-syncer tasks. The enhancement introduces retry mechanisms for CREATE TABLE operations and storage medium synchronization for MODIFY PARTITION operations.

Background & Problem

In CCR scenarios, source and destination clusters often have different storage medium configurations:

• Source cluster: May have diverse medium configurations (SSD/HDD/Mixed)
• Destination cluster: May have different medium availability or cost considerations
• Current limitation: No flexible medium sync strategy support
• Challenges:
  • CREATE TABLE failures when specified medium unavailable in destination
  • Inconsistent storage medium between source and destination

Solution: Medium Sync Policy

HTPP Server Change

• create_ccr: Add medium_sync_policy attribute for policy configuration
• UpdateMediumSyncPolicy: New interface for policy management

Two Medium Sync Policies

Policies 1: hdd (Force HDD)

• Force all tables/partitions to use HDD storage medium

Policies 2: same_with_upstream (Preserve Original)

• Maintain identical storage medium configuration as source cluster

Technical Implementation

Restore(full/partial sync)

Please jump into this pr: apache/doris#52391

CREATE TABLE Binlog Enhancement

Multi-retry mechanism with intelligent fallback:

1. First attempt: Apply medium sync policy (hdd/same_with_upstream)
2. Second attempt: Switch medium (SSD ↔ HDD) if storage-related error
3. Final attempt: Remove storage_medium if still failing (this step should not be triggered and may be removed)

// Example: CREATE TABLE with medium retry
func createTableWithMediumRetry(j *ccr.Job, createTable *record.CreateTable, srcDb string) error {
    originalSql := createTable.Sql
    
    // Process SQL according to medium policy
    if err := processCreateTableSqlByMediumPolicy(j, createTable); err != nil {
        return err
    }
    
    // First attempt with policy-specified medium
    err := j.IDest.CreateTableOrView(createTable, srcDb)
    if err == nil {
        return nil // Success
    }
    
    // Check if it's storage related error and should retry
    if !isStorageMediumError(err.Error()) {
        return err // Not storage related, no retry
    }
    
    // Second attempt: switch medium if storage-related error
    currentMedium := extractStorageMediumFromCreateTableSql(createTable.Sql)
    if currentMedium == "" {
        currentMedium = "ssd" // default
    }
    switchedMedium := switchStorageMedium(currentMedium)
    createTable.Sql = setStorageMediumInCreateTableSql(originalSql, switchedMedium)
    
    err = j.IDest.CreateTableOrView(createTable, srcDb)
    if err == nil {
        return nil // Success with switched medium
    }
    
    // Final attempt: remove storage_medium only if still storage related error
    if isStorageMediumError(err.Error()) {
        createTable.Sql = ccr.FilterStorageMediumFromCreateTableSql(originalSql)
        err = j.IDest.CreateTableOrView(createTable, srcDb)
        if err == nil {
            return nil // Success without storage_medium
        }
    }
    
    return err // All attempts failed
}

MODIFY PARTITION Binlog Enhancement

Storage medium property synchronization:

// Example: ALTER PARTITION storage_medium sync
func (s *Spec) ModifyPartitionProperty(destTableName string, batchInfo *record.BatchModifyPartitionsInfo) error {
    for _, partitionInfo := range batchInfo.Infos {
        sql := fmt.Sprintf("ALTER TABLE %s.%s MODIFY PARTITION %s SET (\"storage_medium\" = \"%s\")",
            dbName, destTableName, partitionInfo.PartitionName, partitionInfo.DataProperty.StorageMedium)
        s.Exec(sql)
    }
}

Strategy Behavior Comparison

Operation	hdd Strategy	same_with_upstream Strategy
CREATE TABLE	Force set storage_medium = "hdd"	Preserve original storage_medium
MODIFY PARTITION	Skip partition modifications	Sync storage_medium from source
Fallback	HDD → Switch -> Remove medium	Original → Switch → Remove medium

Examples

Full/Paritial Sync examples in doris's pr: apache/doris#52391

Scenario 1: Source SSD Table → Destination HDD-only Cluster

-- CCR task with hdd strategy
curl -X POST -H "Content-Type: application/json" -d '{
    "name": "ccr_test_ssd",
    "src": {...},
    "dest": {...},
    "medium_sync_policy": "hdd/same_with_upstream"
}' http://127.0.0.1:9190/create_ccr

CREATE TABLE binlog processing:

• Original SQL: CREATE TABLE orders (...) PROPERTIES ("storage_medium" = "ssd")
• Processed SQL:
  • hdd: storage_medium=hdd
  • same_with_upstream:storage_medium=hdd(ssd -> hdd)

MODIFY PARTITION binlog processing:

• Action:
  • hdd: Skip all partition medium modifications
  • same_with_upstream: Alter storage meidum

Scenario 2: Source Mixed → Destination Mixed Cluster

-- CCR task with same_with_upstream strategy  
curl -X POST -H "Content-Type: application/json" -d '{
    "name": "ccr_test_ssd",
    "src": {...},
    "dest": {...},
    "medium_sync_policy": "same_with_upstream"
}' http://127.0.0.1:9190/create_ccr

CREATE TABLE binlog processing:

Source Table	Original Medium	First Attempt	Fallback	Final Result
hot_data	SSD	SSD	-	SSD
cold_data	HDD	HDD	-	HDD

MODIFY PARTITION binlog processing:

-- Source: 
ALTER TALBE hot_data MODIFY PARTITION (*) SET ("storage_medium" = "hdd")
-- Destination: 
ALTER TABLE hot_data MODIFY PARTITION (*) SET ("storage_medium" = "hdd")