package main

import (
	"fmt"
	"log"
	"os"
	"path/filepath"
	"time"

	deltalake "github.com/tidbcloud/metering/delta-rs-golang"
)

func main() {
	// Create a temporary directory for the demo
	tempDir, err := os.MkdirTemp("/tmp", "delta_optimize_example")
	if err != nil {
		log.Fatal(err)
	}
	defer os.RemoveAll(tempDir)

	tablePath := filepath.Join(tempDir, "optimize_example_table")

	// Define the table schema
	columns := []deltalake.ColumnDefinition{
		{Name: "id", Type: "integer", Nullable: false},
		{Name: "name", Type: "string", Nullable: true},
		{Name: "value", Type: "double", Nullable: true},
		{Name: "created_at", Type: "long", Nullable: false},
	}

	// Create table options
	options := deltalake.CreateTableOptions{
		TablePath: tablePath,
		SaveMode:  deltalake.WriteModeOverwrite,
		Columns:   columns,
	}

	// Create the Delta Lake table
	fmt.Printf("Creating Delta Lake table at: %s\n", tablePath)
	table, err := deltalake.CreateDeltaLakeTable(options)
	if err != nil {
		log.Fatal("Failed to create table:", err)
	}
	defer table.Close()

	fmt.Println("✅ Delta Lake table created successfully!")

	// Write multiple small batches to create many small files
	// This simulates the real-world scenario where small files accumulate
	numBatches := 20      // Create 20 small files to demonstrate optimization effect
	recordsPerBatch := 50 // Small batches to create small files

	fmt.Printf("📝 Writing %d batches of %d records each to create small files...\n", numBatches, recordsPerBatch)

	for i := 0; i < numBatches; i++ {
		// Create a batch of data
		var batchData [][]interface{}
		for j := 0; j < recordsPerBatch; j++ {
			batchData = append(batchData, []interface{}{
				i*recordsPerBatch + j + 1,
				fmt.Sprintf("User_%d_%d", i, j),
				float64(i*recordsPerBatch+j) * 10.5,
				time.Now().Unix() + int64(i*3600), // Different timestamps
			})
		}

		writeData := deltalake.WriteData{
			Schema: columns,
			Data:   batchData,
		}

		// Write data in append mode
		updatedTable, err := table.Write(writeData, deltalake.WriteModeAppend)
		if err != nil {
			log.Fatalf("Failed to write data batch %d: %v", i+1, err)
		}

		// Close the old table and use the updated one
		table.Close()
		table = updatedTable

		if (i+1)%5 == 0 || i == numBatches-1 {
			fmt.Printf("  ✓ Written %d/%d batches\n", i+1, numBatches)
		}
	}

	totalRecords := numBatches * recordsPerBatch
	fmt.Printf("📋 Total records written: %d\n", totalRecords)

	// Check table files before optimization
	fmt.Println("\n🔍 Table status before optimization:")
	files, err := table.GetFiles()
	if err != nil {
		log.Fatalf("Failed to get table files: %v", err)
	}
	fmt.Printf("  📁 Number of files: %d\n", len(files))

	// Show some example file paths to demonstrate the small file problem
	if len(files) > 0 {
		fmt.Printf("  📄 Sample file paths:\n")
		for i, file := range files {
			if i < 3 { // Show first 3 files as examples
				fmt.Printf("    • %s\n", file)
			} else if i == 3 {
				fmt.Printf("    • ... and %d more files\n", len(files)-3)
				break
			}
		}
		fmt.Printf("  ⚠️  Multiple small files detected: This is a good candidate for optimization!\n")
	}

	// Get table version before optimization
	versionBefore, err := table.Version()
	if err != nil {
		log.Fatalf("Failed to get table version: %v", err)
	}
	fmt.Printf("  📋 Table version: %d\n", versionBefore)

	// Perform compact optimization
	fmt.Println("\n🔧 Starting compact optimization...")
	optimizeBuilder := table.Optimize()
	if optimizeBuilder == nil {
		log.Fatalf("Failed to create optimize builder")
	}
	defer optimizeBuilder.Free()

	// Set optimization parameters
	optimizeBuilder.SetOptimizeType(deltalake.OptimizeTypeCompact).
		SetTargetSize(128 * 1024 * 1024). // 128MB target size
		SetMaxConcurrentTasks(4)

	// Execute optimization
	optimizeResult, err := optimizeBuilder.Execute()
	if err != nil {
		log.Fatalf("Failed to execute optimization: %v", err)
	}

	fmt.Println("✅ Optimization completed successfully!")

	// Print optimization metrics
	metrics := optimizeResult.Metrics
	fmt.Printf("\n📊 Optimization Metrics:\n")
	fmt.Printf("  📈 Files Added: %d\n", metrics.NumFilesAdded)
	fmt.Printf("  📉 Files Removed: %d\n", metrics.NumFilesRemoved)
	fmt.Printf("  🗂️  Partitions Optimized: %d\n", metrics.PartitionsOptimized)
	fmt.Printf("  🔍 Total Files Considered: %d\n", metrics.TotalConsideredFiles)
	fmt.Printf("  ⏭️  Total Files Skipped: %d\n", metrics.TotalFilesSkipped)
	fmt.Printf("  📋 Preserve Insertion Order: %t\n", metrics.PreserveInsertionOrder)

	// Print file details
	fmt.Printf("\n📊 Added Files Details:\n")
	fmt.Printf("  📄 Count: %d\n", metrics.FilesAdded.TotalFiles)
	fmt.Printf("  💾 Total Size: %d bytes\n", metrics.FilesAdded.TotalSize)
	fmt.Printf("  📊 Average Size: %.2f bytes\n", metrics.FilesAdded.Avg)
	fmt.Printf("  📉 Min Size: %d bytes\n", metrics.FilesAdded.Min)
	fmt.Printf("  📈 Max Size: %d bytes\n", metrics.FilesAdded.Max)

	fmt.Printf("\n📊 Removed Files Details:\n")
	fmt.Printf("  📄 Count: %d\n", metrics.FilesRemoved.TotalFiles)
	fmt.Printf("  💾 Total Size: %d bytes\n", metrics.FilesRemoved.TotalSize)
	fmt.Printf("  📊 Average Size: %.2f bytes\n", metrics.FilesRemoved.Avg)
	fmt.Printf("  📉 Min Size: %d bytes\n", metrics.FilesRemoved.Min)
	fmt.Printf("  📈 Max Size: %d bytes\n", metrics.FilesRemoved.Max)

	// Update table reference
	table = optimizeResult.Table

	// Check table files after optimization
	fmt.Println("\n🔍 Table status after optimization:")
	filesAfter, err := table.GetFiles()
	if err != nil {
		log.Fatalf("Failed to get table files after optimization: %v", err)
	}
	fmt.Printf("  📁 Number of files: %d\n", len(filesAfter))

	// Get table version after optimization
	versionAfter, err := table.Version()
	if err != nil {
		log.Fatalf("Failed to get table version after optimization: %v", err)
	}
	fmt.Printf("  📋 Table version: %d\n", versionAfter)

	// Show optimization summary
	fmt.Println("\n🎯 Optimization Summary:")
	fileReduction := len(files) - len(filesAfter)
	reductionPercent := float64(fileReduction) / float64(len(files)) * 100
	fmt.Printf("  📉 File count reduced: %d → %d (-%d files, %.1f%% reduction)\n",
		len(files), len(filesAfter), fileReduction, reductionPercent)

	if metrics.FilesRemoved.TotalFiles > 0 && metrics.FilesAdded.TotalFiles > 0 {
		avgSizeIncrease := metrics.FilesAdded.Avg / metrics.FilesRemoved.Avg
		fmt.Printf("  📈 Average file size increased: %.1fx larger (%.0f → %.0f bytes)\n",
			avgSizeIncrease, metrics.FilesRemoved.Avg, metrics.FilesAdded.Avg)
	}

	fmt.Printf("  💡 Benefits: Reduced metadata overhead, improved query performance\n")
	fmt.Printf("  ✨ Delta table version: %d → %d\n", versionBefore, versionAfter)

	// Verify data integrity by reading all data
	fmt.Println("\n🔍 Verifying data integrity...")
	data, err := table.ReadData(0) // Read all data
	if err != nil {
		log.Fatalf("Failed to read table data: %v", err)
	}

	fmt.Printf("✅ Data integrity verified: %d records found (expected: %d)\n", len(data), totalRecords)

	if len(data) != totalRecords {
		log.Fatalf("❌ Data integrity check failed: expected %d records, got %d", totalRecords, len(data))
	}

	// Show a few sample records to demonstrate data is accessible
	fmt.Println("\n📋 Sample records after optimization:")
	for i, record := range data {
		if i >= 3 { // Show only first 3 records
			break
		}
		fmt.Printf("  🔸 Record %d: ID=%v, Name=%v, Value=%v\n",
			i+1, record["id"], record["name"], record["value"])
	}

	fmt.Printf("\n🎉 Optimize example completed successfully! Delta-rs version: %s\n", deltalake.Version())
}