package deltalake

/*
#cgo LDFLAGS: -L ./target/x86_64-unknown-linux-gnu/release -L ./target/x86_64-unknown-linux-gnu/debug -L ./target/release -L ./target/debug -ldeltalake_golang
#include <stdlib.h>
#include <stdint.h>
#include <stdbool.h>

// Error handling
typedef struct {
    int error_type;
    char* message;
} DeltaGoError;

// Result types
typedef struct {
    bool success;
    void* value;
    DeltaGoError* error;
} CResult_TableHandle;

typedef struct {
    bool success;
    char* value;
    DeltaGoError* error;
} CResult_String;

typedef struct {
    bool success;
    long value;
    DeltaGoError* error;
} CResult_Long;

typedef struct {
    bool success;
    bool value;
    DeltaGoError* error;
} CResult_Bool;

typedef struct {
    bool success;
    void* value;
    DeltaGoError* error;
} CResult_Void;

// Table handle
typedef struct {
    void* ptr;
} DeltaTableHandle;

// Core functions
void deltalake_init();
char* deltalake_version();

// Table management
CResult_TableHandle deltalake_create_table(char* table_path, char* save_mode, char* columns_json, char* partition_columns_json, char* table_name, char* storage_options_json, char* table_configuration_json);
CResult_TableHandle deltalake_open_table(char* table_path, char* storage_options_json);
CResult_TableHandle deltalake_open_table_with_config(char* table_path, char* storage_options_json, int require_files, int log_buffer_size);
CResult_Bool deltalake_table_exists(char* table_path, char* storage_options_json);
CResult_TableHandle deltalake_write_table(void* table, char* records_json, char* save_mode);
CResult_Long deltalake_table_version(void* table);
CResult_String deltalake_table_files(void* table);
CResult_String deltalake_read_table_data(void* table, long limit);
CResult_String deltalake_query_table(void* table, char* sql_query, char* table_alias);
void deltalake_free_table(void* table);


// Delete operations
typedef struct DeleteResult {
	void* table_handle;
	char* metrics_json;
} DeleteResult;

typedef struct {
	bool success;
	DeleteResult* value;
	DeltaGoError* error;
} CResult_DeleteResult;

CResult_DeleteResult deltalake_delete_where(void* table_handle, char* predicate);
void deltalake_free_delete_result(void* result);

// Vacuum operations
typedef struct VacuumResult {
    void* table_handle;
    char* metrics_json;
} VacuumResult;

void* deltalake_vacuum_table(void* table_handle);
void deltalake_vacuum_set_retention_hours(void* vacuum_handle, int hours);
void deltalake_vacuum_set_dry_run(void* vacuum_handle, int dry_run);
VacuumResult* deltalake_vacuum_execute(void* vacuum_handle);
void deltalake_free_vacuum_builder(void* vacuum_handle);
void deltalake_free_vacuum_result(VacuumResult* result);

// Metadata cleanup operations
long deltalake_cleanup_metadata(void* table_handle);
long deltalake_cleanup_metadata_by_key(char* ops_key);

// Optimize operations
typedef struct OptimizeResult {
    void* table_handle;
    char* metrics_json;
} OptimizeResult;

void* deltalake_optimize_table(void* table_handle);
bool deltalake_optimize_set_type(void* optimize_handle, char* optimize_type, char* zorder_columns_json);
bool deltalake_optimize_set_target_size(void* optimize_handle, long target_size);
bool deltalake_optimize_set_max_concurrent_tasks(void* optimize_handle, int max_tasks);
OptimizeResult* deltalake_optimize_execute(void* optimize_handle);
void deltalake_free_optimize_builder(void* optimize_handle);
void deltalake_free_optimize_result(OptimizeResult* result);

// Memory management
void deltalake_free_string(char* ptr);
void deltalake_free_error(DeltaGoError* error);

// Error handling
char* deltalake_error_message(const DeltaGoError* error);
int deltalake_error_type(const DeltaGoError* error);
*/
import "C"
import (
	"encoding/json"
	"fmt"
	"strings"
	"unsafe"
)

// Error types
const (
	ErrorTypeDeltaTable            = 0
	ErrorTypeObjectStore           = 1
	ErrorTypeArrow                 = 2
	ErrorTypeDataFusion            = 3
	ErrorTypeInvalidUri            = 4
	ErrorTypeInvalidSchema         = 5
	ErrorTypeInvalidConfiguration  = 6
	ErrorTypeInvalidStorageOptions = 7
	ErrorTypeNullPointer           = 8
	ErrorTypeInvalidUtf8           = 9
	ErrorTypeIoError               = 10
	ErrorTypeSerializationError    = 11
)

// DeltaError represents an error from the Delta Lake library
type DeltaError struct {
	ErrorType int
	Message   string
}

func (e *DeltaError) Error() string {
	return fmt.Sprintf("Delta error (type %d): %s", e.ErrorType, e.Message)
}

// OptimizeMetrics represents optimization results
type OptimizeMetrics struct {
	NumFilesAdded          int64         `json:"numFilesAdded"`
	NumFilesRemoved        int64         `json:"numFilesRemoved"`
	FilesAdded             MetricDetails `json:"-"` // Will be parsed from string
	FilesRemoved           MetricDetails `json:"-"` // Will be parsed from string
	PartitionsOptimized    int64         `json:"partitionsOptimized"`
	NumBatches             int64         `json:"numBatches"`
	TotalConsideredFiles   int           `json:"totalConsideredFiles"`
	TotalFilesSkipped      int           `json:"totalFilesSkipped"`
	PreserveInsertionOrder bool          `json:"preserveInsertionOrder"`
}

// Custom type for parsing the JSON response
type optimizeMetricsRaw struct {
	NumFilesAdded          int64  `json:"numFilesAdded"`
	NumFilesRemoved        int64  `json:"numFilesRemoved"`
	FilesAddedStr          string `json:"filesAdded"`
	FilesRemovedStr        string `json:"filesRemoved"`
	PartitionsOptimized    int64  `json:"partitionsOptimized"`
	NumBatches             int64  `json:"numBatches"`
	TotalConsideredFiles   int    `json:"totalConsideredFiles"`
	TotalFilesSkipped      int    `json:"totalFilesSkipped"`
	PreserveInsertionOrder bool   `json:"preserveInsertionOrder"`
}

// WriteMode represents the write mode for table operations
type WriteMode string

const (
	WriteModeAppend    WriteMode = "append"
	WriteModeOverwrite WriteMode = "overwrite"
	WriteModeError     WriteMode = "error"
	WriteModeIgnore    WriteMode = "ignore"
)

// ColumnDefinition represents a column definition for creating Delta tables
type ColumnDefinition struct {
	Name     string `json:"name"`
	Type     string `json:"type"`
	Nullable bool   `json:"nullable"`
}

// WriteData represents data to be written to a Delta table
type WriteData struct {
	Schema []ColumnDefinition `json:"schema"`
	Data   [][]interface{}    `json:"data"`
}

// QueryOptions represents optional parameters for SQL queries
type QueryOptions struct {
	// TableAlias allows overriding the table name used within the SQL statement.
	// When empty, the table is registered as "delta_table".
	TableAlias string
}

// DeltaTable represents a Delta Lake table handle
type DeltaTable struct {
	handle unsafe.Pointer
}

// TableConfig represents configuration options for opening Delta tables
type TableConfig struct {
	// RequireFiles controls whether file metadata is loaded into memory.
	// When false, significantly reduces memory usage for write-only workloads.
	// Default: true
	RequireFiles bool

	// LogBufferSize controls concurrent log file reading during table loading.
	// Higher values may improve performance but increase memory usage.
	// Default: 4 * number of CPUs
	LogBufferSize int
}

// DefaultTableConfig returns the default table configuration
func DefaultTableConfig() TableConfig {
	return TableConfig{
		RequireFiles:  true,
		LogBufferSize: 0, // 0 means use delta-rs default
	}
}

// MergeBuilder represents a Delta Lake merge operation builder
type MergeBuilder struct {
	handle unsafe.Pointer
	table  *DeltaTable
}

// MergeUpdate represents column updates for merge operations
type MergeUpdate struct {
	Column string `json:"column"`
	Value  string `json:"value"`
}

// MergeOptions represents options for merge operations
type MergeOptions struct {
	SourceAlias string
	TargetAlias string
	Updates     map[string]string // column -> expression mapping
}

// VacuumBuilder represents a Delta Lake vacuum operation builder
type VacuumBuilder struct {
	handle unsafe.Pointer
	table  *DeltaTable
}

// VacuumMetrics represents the results of a vacuum operation
type VacuumMetrics struct {
	FilesDeleted      []string `json:"files_deleted"`
	TotalFilesDeleted int64    `json:"total_files_deleted"`
	DryRun            bool     `json:"dry_run"`
	Error             string   `json:"error,omitempty"`
}

// VacuumResult represents the complete result of a vacuum operation
type VacuumResult struct {
	Table   *DeltaTable
	Metrics VacuumMetrics
}

// DeleteMetrics represents the results of a delete operation
type DeleteMetrics struct {
	NumAddedFiles   int64 `json:"num_added_files"`
	NumRemovedFiles int64 `json:"num_removed_files"`
	NumDeletedRows  int64 `json:"num_deleted_rows"`
	NumCopiedRows   int64 `json:"num_copied_rows"`
	ExecutionTimeMS int64 `json:"execution_time_ms"`
	ScanTimeMS      int64 `json:"scan_time_ms"`
	RewriteTimeMS   int64 `json:"rewrite_time_ms"`
}

// DeleteResult represents the complete result of a delete operation
type DeleteResult struct {
	Table   *DeltaTable
	Metrics DeleteMetrics
}

type cDeleteResult struct {
	table_handle unsafe.Pointer
	metrics_json *C.char
}

// OptimizeBuilder represents a Delta Lake optimize operation builder
type OptimizeBuilder struct {
	handle unsafe.Pointer
	table  *DeltaTable
}

// OptimizeType represents the type of optimization
type OptimizeType string

const (
	OptimizeTypeCompact OptimizeType = "compact"
	OptimizeTypeZOrder  OptimizeType = "zorder"
)

// MetricDetails represents detailed metrics for file operations
type MetricDetails struct {
	Avg        float64 `json:"avg"`
	Max        int64   `json:"max"`
	Min        int64   `json:"min"`
	TotalFiles int     `json:"totalFiles"`
	TotalSize  int64   `json:"totalSize"`
}

// OptimizeResult represents the complete result of an optimize operation
type OptimizeResult struct {
	Table   *DeltaTable
	Metrics OptimizeMetrics
}

// Close frees the Delta table handle
func (dt *DeltaTable) Close() {
	if dt.handle != nil {
		C.deltalake_free_table(dt.handle)
		dt.handle = nil
	}
}

// TriggerGC asks delta-rs to run internal garbage collection for deferred resources.

// Version returns the current version of the Delta table
func (dt *DeltaTable) Version() (int64, error) {
	if dt.handle == nil {
		return 0, fmt.Errorf("table handle is nil")
	}

	result := C.deltalake_table_version(dt.handle)
	if !result.success {
		return 0, convertError(result.error)
	}

	return int64(result.value), nil
}

// GetFiles returns the list of files in the Delta table
func (dt *DeltaTable) GetFiles() ([]string, error) {
	if dt.handle == nil {
		return nil, fmt.Errorf("table handle is nil")
	}

	result := C.deltalake_table_files(dt.handle)
	if !result.success {
		return nil, convertError(result.error)
	}

	defer C.deltalake_free_string(result.value)
	filesJSON := C.GoString(result.value)

	var files []string
	if err := json.Unmarshal([]byte(filesJSON), &files); err != nil {
		return nil, fmt.Errorf("failed to parse files JSON: %w", err)
	}

	return files, nil
}

// ReadData reads data from the Delta table and returns it as a slice of maps
func (dt *DeltaTable) ReadData(limit int64) ([]map[string]interface{}, error) {
	if dt.handle == nil {
		return nil, fmt.Errorf("table handle is nil")
	}

	result := C.deltalake_read_table_data(dt.handle, C.long(limit))
	if !result.success {
		return nil, convertError(result.error)
	}

	defer C.deltalake_free_string(result.value)
	dataJSON := C.GoString(result.value)

	var data []map[string]interface{}
	if err := json.Unmarshal([]byte(dataJSON), &data); err != nil {
		return nil, fmt.Errorf("failed to parse data JSON: %w", err)
	}

	return data, nil
}

// Query executes an arbitrary SQL query against the Delta table and returns the resulting rows.
// The table is registered in the query context as "delta_table" by default, or using the alias
// provided through QueryOptions.
func (dt *DeltaTable) Query(sql string, options *QueryOptions) ([]map[string]interface{}, error) {
	if dt.handle == nil {
		return nil, fmt.Errorf("table handle is nil")
	}

	if strings.TrimSpace(sql) == "" {
		return nil, fmt.Errorf("sql query cannot be empty")
	}

	sqlCStr := C.CString(sql)
	defer C.free(unsafe.Pointer(sqlCStr))

	var aliasCStr *C.char
	if options != nil {
		if alias := strings.TrimSpace(options.TableAlias); alias != "" {
			aliasCStr = C.CString(alias)
			defer C.free(unsafe.Pointer(aliasCStr))
		}
	}

	result := C.deltalake_query_table(dt.handle, sqlCStr, aliasCStr)
	if !result.success {
		return nil, convertError(result.error)
	}

	defer C.deltalake_free_string(result.value)
	resultJSON := C.GoString(result.value)

	var rows []map[string]interface{}
	if err := json.Unmarshal([]byte(resultJSON), &rows); err != nil {
		return nil, fmt.Errorf("failed to parse query result JSON: %w", err)
	}

	return rows, nil
}

// Write writes data to the Delta table
//
// CRITICAL MEMORY MANAGEMENT:
// This function transfers ownership of dt.handle to Rust.
// After calling this function, dt.handle is ALWAYS invalidated.
//
// Return value semantics from Rust:
// - On success: result.value = new table pointer
// - On early error (before consumption): result.value = original table pointer (reusable)
// - On late error (after consumption): result.value = null (pointer is gone)
//
// Caller MUST always use the returned DeltaTable, never the original dt.
func (dt *DeltaTable) Write(writeData WriteData, saveMode WriteMode) (*DeltaTable, error) {
	if dt.handle == nil {
		return nil, fmt.Errorf("table handle is nil")
	}

	// Convert write data to JSON
	recordsJSON, err := json.Marshal(writeData)
	if err != nil {
		return nil, fmt.Errorf("failed to marshal write data to JSON: %w", err)
	}

	// Convert parameters to C strings
	recordsCStr := C.CString(string(recordsJSON))
	defer C.free(unsafe.Pointer(recordsCStr))

	modeCStr := C.CString(string(saveMode))
	defer C.free(unsafe.Pointer(modeCStr))

	// Call C function - this ALWAYS takes ownership of dt.handle
	result := C.deltalake_write_table(dt.handle, recordsCStr, modeCStr)

	// CRITICAL: dt.handle is now owned by Rust, invalidate it to prevent double-free
	dt.handle = nil

	// Check for errors
	if !result.success {
		dt.handle = result.value
		return dt, convertError(result.error)
	}

	// Success: validate the new pointer
	if result.value == nil || result.error != nil {
		// This should never happen on success
		panic("write succeeded but returned nil table handle")
	}

	// Create Go wrapper for the new table
	newTable := &DeltaTable{
		handle: result.value,
	}

	return newTable, nil
}

// DeleteWhere removes rows from the Delta table that satisfy the given predicate.
// An empty predicate deletes all data from the table.
func (dt *DeltaTable) DeleteWhere(predicate string) (*DeleteResult, error) {
	if dt.handle == nil {
		return nil, fmt.Errorf("table handle is nil")
	}

	var predicateCStr *C.char
	if predicate != "" {
		predicateCStr = C.CString(predicate)
		defer C.free(unsafe.Pointer(predicateCStr))
	}

	result := C.deltalake_delete_where(dt.handle, predicateCStr)
	if !result.success {
		return nil, convertError(result.error)
	}

	if result.value == nil {
		return nil, fmt.Errorf("delete operation returned nil result")
	}

	deleteResultPtr := unsafe.Pointer(result.value)
	defer C.deltalake_free_delete_result(deleteResultPtr)

	deleteResult := (*cDeleteResult)(deleteResultPtr)

	if deleteResult.table_handle == nil {
		return nil, fmt.Errorf("delete operation returned nil table handle")
	}

	// Parse metrics JSON
	metrics := DeleteMetrics{}
	if deleteResult.metrics_json != nil {
		metricsJSON := C.GoString(deleteResult.metrics_json)
		if metricsJSON != "" {
			if err := json.Unmarshal([]byte(metricsJSON), &metrics); err != nil {
				// Free the table handle on error path
				C.deltalake_free_table(deleteResult.table_handle)
				return nil, fmt.Errorf("failed to parse delete metrics: %w", err)
			}
		}
	}

	// Extract table handle (ownership transferred from DeleteResult to DeltaTable)
	// The defer above will free the DeleteResult struct and metrics_json, but NOT table_handle
	newTable := &DeltaTable{
		handle: deleteResult.table_handle,
	}

	return &DeleteResult{
		Table:   newTable,
		Metrics: metrics,
	}, nil
}

// CreateTableOptions represents options for creating a Delta table
type CreateTableOptions struct {
	TablePath          string
	SaveMode           WriteMode
	Columns            []ColumnDefinition
	PartitionColumns   []string          // Partition columns for creating partitioned tables
	TableName          *string           // Optional table name
	StorageOptions     map[string]string // Optional storage options for object storage configuration
	TableConfiguration map[string]string // Optional table configuration properties (delta.*)
}

func init() {
	C.deltalake_init()
}

// Version returns the version of the delta-rs core library
func Version() string {
	cstr := C.deltalake_version()
	defer C.deltalake_free_string(cstr)
	return C.GoString(cstr)
}

// Helper function to convert Go error from C result
func convertError(cerr *C.DeltaGoError) error {
	if cerr == nil {
		return nil
	}
	defer C.deltalake_free_error(cerr)

	errType := int(C.deltalake_error_type(cerr))
	message := C.GoString(C.deltalake_error_message(cerr))

	return &DeltaError{
		ErrorType: errType,
		Message:   message,
	}
}

// Helper function to convert Go strings to C string array
func stringSliceToC(strs []string) (**C.char, C.int) {
	if len(strs) == 0 {
		return nil, 0
	}

	cstrs := make([]*C.char, len(strs))
	for i, s := range strs {
		cstrs[i] = C.CString(s)
	}

	return &cstrs[0], C.int(len(strs))
}

// Helper function to free C string array
func freeCStringArray(cstrs **C.char, length C.int) {
	if cstrs == nil || length == 0 {
		return
	}

	slice := (*[1 << 30]*C.char)(unsafe.Pointer(cstrs))[:length:length]
	for _, cstr := range slice {
		C.free(unsafe.Pointer(cstr))
	}
}

// OpenDeltaLakeTable opens an existing Delta Lake table
func OpenDeltaLakeTable(tablePath string, storageOptions map[string]string) (*DeltaTable, error) {
	// Validate input parameters
	if tablePath == "" {
		return nil, &DeltaError{
			ErrorType: ErrorTypeInvalidConfiguration,
			Message:   "table path cannot be empty",
		}
	}

	// Convert table path to C string
	pathCStr := C.CString(tablePath)
	defer C.free(unsafe.Pointer(pathCStr))

	// Convert storage options to JSON (optional)
	var storageOptionsCStr *C.char
	if len(storageOptions) > 0 {
		storageOptionsJSON, marshalErr := json.Marshal(storageOptions)
		if marshalErr != nil {
			return nil, fmt.Errorf("failed to marshal storage options to JSON: %w", marshalErr)
		}
		storageOptionsCStr = C.CString(string(storageOptionsJSON))
		defer C.free(unsafe.Pointer(storageOptionsCStr))
	}

	// Call C function
	result := C.deltalake_open_table(pathCStr, storageOptionsCStr)

	// Check for errors
	if !result.success {
		return nil, convertError(result.error)
	}

	// Validate the result
	if result.value == nil {
		return nil, &DeltaError{
			ErrorType: ErrorTypeNullPointer,
			Message:   "received null table handle from C library",
		}
	}

	// Create Go wrapper
	table := &DeltaTable{
		handle: result.value,
	}

	return table, nil
}

// OpenDeltaLakeTableWithConfig opens an existing Delta Lake table with custom configuration
func OpenDeltaLakeTableWithConfig(tablePath string, storageOptions map[string]string, config TableConfig) (*DeltaTable, error) {
	// Validate input parameters
	if tablePath == "" {
		return nil, &DeltaError{
			ErrorType: ErrorTypeInvalidConfiguration,
			Message:   "table path cannot be empty",
		}
	}

	// Convert table path to C string
	pathCStr := C.CString(tablePath)
	defer C.free(unsafe.Pointer(pathCStr))

	// Convert storage options to JSON (optional)
	var storageOptionsCStr *C.char
	if len(storageOptions) > 0 {
		storageOptionsJSON, marshalErr := json.Marshal(storageOptions)
		if marshalErr != nil {
			return nil, fmt.Errorf("failed to marshal storage options to JSON: %w", marshalErr)
		}
		storageOptionsCStr = C.CString(string(storageOptionsJSON))
		defer C.free(unsafe.Pointer(storageOptionsCStr))
	}

	// Convert config to C values
	requireFiles := 1
	if !config.RequireFiles {
		requireFiles = 0
	}
	logBufferSize := C.int(config.LogBufferSize)

	// Call C function with config
	result := C.deltalake_open_table_with_config(pathCStr, storageOptionsCStr, C.int(requireFiles), logBufferSize)

	// Check for errors
	if !result.success {
		return nil, convertError(result.error)
	}

	// Validate the result
	if result.value == nil {
		return nil, &DeltaError{
			ErrorType: ErrorTypeNullPointer,
			Message:   "received null table handle from C library",
		}
	}

	// Create Go wrapper
	table := &DeltaTable{
		handle: result.value,
	}

	return table, nil
}

// TableExists checks if a Delta Lake table exists at the specified path
func TableExists(tablePath string, storageOptions map[string]string) (bool, error) {
	// Validate input parameters
	if tablePath == "" {
		return false, &DeltaError{
			ErrorType: ErrorTypeInvalidConfiguration,
			Message:   "table path cannot be empty",
		}
	}

	// Convert table path to C string
	pathCStr := C.CString(tablePath)
	defer C.free(unsafe.Pointer(pathCStr))

	// Convert storage options to JSON (optional)
	var storageOptionsCStr *C.char
	if len(storageOptions) > 0 {
		storageOptionsJSON, marshalErr := json.Marshal(storageOptions)
		if marshalErr != nil {
			return false, fmt.Errorf("failed to marshal storage options to JSON: %w", marshalErr)
		}
		storageOptionsCStr = C.CString(string(storageOptionsJSON))
		defer C.free(unsafe.Pointer(storageOptionsCStr))
	}

	// Call C function
	result := C.deltalake_table_exists(pathCStr, storageOptionsCStr)

	// Check for errors
	if !result.success {
		return false, convertError(result.error)
	}

	return bool(result.value), nil
}

// CreateDeltaLakeTable creates a new Delta Lake table with the specified options
func CreateDeltaLakeTable(options CreateTableOptions) (*DeltaTable, error) {
	// Convert table path to C string
	pathCStr := C.CString(options.TablePath)
	defer C.free(unsafe.Pointer(pathCStr))

	// Convert save mode to C string
	modeCStr := C.CString(string(options.SaveMode))
	defer C.free(unsafe.Pointer(modeCStr))

	// Convert columns to JSON
	columnsJSON, err := json.Marshal(options.Columns)
	if err != nil {
		return nil, fmt.Errorf("failed to marshal columns to JSON: %w", err)
	}
	columnsCStr := C.CString(string(columnsJSON))
	defer C.free(unsafe.Pointer(columnsCStr))

	// Convert partition columns to JSON
	var partitionColumnsCStr *C.char
	if len(options.PartitionColumns) > 0 {
		partitionColumnsJSON, err := json.Marshal(options.PartitionColumns)
		if err != nil {
			return nil, fmt.Errorf("failed to marshal partition columns to JSON: %w", err)
		}
		partitionColumnsCStr = C.CString(string(partitionColumnsJSON))
		defer C.free(unsafe.Pointer(partitionColumnsCStr))
	}

	// Convert table name to C string (optional)
	var nameCStr *C.char
	if options.TableName != nil {
		nameCStr = C.CString(*options.TableName)
		defer C.free(unsafe.Pointer(nameCStr))
	}

	// Convert storage options to JSON (optional)
	var storageOptionsCStr *C.char
	if len(options.StorageOptions) > 0 {
		storageOptionsJSON, err := json.Marshal(options.StorageOptions)
		if err != nil {
			return nil, fmt.Errorf("failed to marshal storage options to JSON: %w", err)
		}
		storageOptionsCStr = C.CString(string(storageOptionsJSON))
		defer C.free(unsafe.Pointer(storageOptionsCStr))
	}

	// Convert table configuration to JSON (optional)
	var tableConfigurationCStr *C.char
	if len(options.TableConfiguration) > 0 {
		tableConfigJSON, err := json.Marshal(options.TableConfiguration)
		if err != nil {
			return nil, fmt.Errorf("failed to marshal table configuration to JSON: %w", err)
		}
		tableConfigurationCStr = C.CString(string(tableConfigJSON))
		defer C.free(unsafe.Pointer(tableConfigurationCStr))
	}

	// Call C function
	result := C.deltalake_create_table(pathCStr, modeCStr, columnsCStr, partitionColumnsCStr, nameCStr, storageOptionsCStr, tableConfigurationCStr)

	// Check for errors
	if !result.success {
		return nil, convertError(result.error)
	}

	// Create Go wrapper
	table := &DeltaTable{
		handle: result.value,
	}

	return table, nil
}

// Vacuum starts a vacuum operation on the Delta table
func (dt *DeltaTable) Vacuum() *VacuumBuilder {
	if dt == nil || dt.handle == nil {
		return nil
	}

	handle := C.deltalake_vacuum_table(dt.handle)

	if handle == nil {
		return nil
	}

	return &VacuumBuilder{
		handle: handle,
		table:  dt,
	}
}

// SetRetentionHours sets the retention period in hours for the vacuum operation
func (vb *VacuumBuilder) SetRetentionHours(hours int) *VacuumBuilder {
	if vb.handle != nil {
		C.deltalake_vacuum_set_retention_hours(vb.handle, C.int(hours))
	}
	return vb
}

// SetDryRun sets whether the vacuum operation should be a dry run
func (vb *VacuumBuilder) SetDryRun(dryRun bool) *VacuumBuilder {
	if vb.handle != nil {
		var cDryRun C.int
		if dryRun {
			cDryRun = 1
		} else {
			cDryRun = 0
		}
		C.deltalake_vacuum_set_dry_run(vb.handle, cDryRun)
	}
	return vb
}

// Execute executes the vacuum operation and returns the result
func (vb *VacuumBuilder) Execute() (*VacuumResult, error) {
	if vb.handle == nil {
		return nil, fmt.Errorf("vacuum builder is nil")
	}

	result := C.deltalake_vacuum_execute(vb.handle)

	if result == nil {
		return nil, fmt.Errorf("vacuum execution failed")
	}

	defer C.deltalake_free_vacuum_result(result)

	// Convert C result to Go struct
	metricsJson := C.GoString(result.metrics_json)

	var metrics VacuumMetrics
	if err := json.Unmarshal([]byte(metricsJson), &metrics); err != nil {
		return nil, fmt.Errorf("failed to parse vacuum metrics: %v", err)
	}

	// Check for errors first
	if metrics.Error != "" {
		return &VacuumResult{
			Table:   vb.table,
			Metrics: metrics,
		}, fmt.Errorf("vacuum operation error: %s", metrics.Error)
	}

	// Update the table handle if execution was successful
	if result.table_handle != nil {
		if vb.table.handle != nil {
			C.deltalake_free_table(vb.table.handle)
		}
		vb.table.handle = result.table_handle
	}

	return &VacuumResult{
		Table:   vb.table,
		Metrics: metrics,
	}, nil
}

// Free frees the vacuum builder resources
func (vb *VacuumBuilder) Free() {
	if vb.handle != nil {
		C.deltalake_free_vacuum_builder(vb.handle)
		vb.handle = nil
	}
}

// Optimize creates a new optimize builder for this table
func (dt *DeltaTable) Optimize() *OptimizeBuilder {
	if dt.handle == nil {
		return nil
	}

	handle := C.deltalake_optimize_table(dt.handle)
	if handle == nil {
		return nil
	}

	return &OptimizeBuilder{
		handle: handle,
		table:  dt,
	}
}

// SetOptimizeType sets the optimization type (compact or zorder)
// For zorder optimization, provide the column names in zorderColumns
func (ob *OptimizeBuilder) SetOptimizeType(optimizeType OptimizeType, zorderColumns ...string) *OptimizeBuilder {
	if ob.handle == nil {
		return ob
	}

	var zorderJSON string
	if optimizeType == OptimizeTypeZOrder && len(zorderColumns) > 0 {
		zorderData, _ := json.Marshal(zorderColumns)
		zorderJSON = string(zorderData)
	}

	typeCStr := C.CString(string(optimizeType))
	defer C.free(unsafe.Pointer(typeCStr))

	var zorderCStr *C.char
	if zorderJSON != "" {
		zorderCStr = C.CString(zorderJSON)
		defer C.free(unsafe.Pointer(zorderCStr))
	}

	C.deltalake_optimize_set_type(ob.handle, typeCStr, zorderCStr)
	return ob
}

// SetTargetSize sets the target file size for optimization
func (ob *OptimizeBuilder) SetTargetSize(targetSize int64) *OptimizeBuilder {
	if ob.handle == nil {
		return ob
	}

	C.deltalake_optimize_set_target_size(ob.handle, C.long(targetSize))
	return ob
}

// SetMaxConcurrentTasks sets the maximum number of concurrent tasks for optimization
func (ob *OptimizeBuilder) SetMaxConcurrentTasks(maxTasks int) *OptimizeBuilder {
	if ob.handle == nil {
		return ob
	}

	C.deltalake_optimize_set_max_concurrent_tasks(ob.handle, C.int(maxTasks))
	return ob
}

// Execute runs the optimize operation and returns the result
func (ob *OptimizeBuilder) Execute() (*OptimizeResult, error) {
	if ob.handle == nil {
		return nil, fmt.Errorf("optimize builder handle is nil")
	}

	result := C.deltalake_optimize_execute(ob.handle)
	if result == nil {
		return nil, fmt.Errorf("optimize operation failed")
	}
	defer C.deltalake_free_optimize_result(result)

	// Parse metrics JSON
	metricsJSON := C.GoString(result.metrics_json)

	// First parse the raw structure
	var rawMetrics optimizeMetricsRaw
	if err := json.Unmarshal([]byte(metricsJSON), &rawMetrics); err != nil {
		return nil, fmt.Errorf("failed to parse optimize metrics: %v", err)
	}

	// Parse nested JSON strings for file details
	var filesAdded, filesRemoved MetricDetails
	if rawMetrics.FilesAddedStr != "" {
		if err := json.Unmarshal([]byte(rawMetrics.FilesAddedStr), &filesAdded); err != nil {
			return nil, fmt.Errorf("failed to parse files added details: %v", err)
		}
	}
	if rawMetrics.FilesRemovedStr != "" {
		if err := json.Unmarshal([]byte(rawMetrics.FilesRemovedStr), &filesRemoved); err != nil {
			return nil, fmt.Errorf("failed to parse files removed details: %v", err)
		}
	}

	// Build the final metrics structure
	metrics := OptimizeMetrics{
		NumFilesAdded:          rawMetrics.NumFilesAdded,
		NumFilesRemoved:        rawMetrics.NumFilesRemoved,
		FilesAdded:             filesAdded,
		FilesRemoved:           filesRemoved,
		PartitionsOptimized:    rawMetrics.PartitionsOptimized,
		NumBatches:             rawMetrics.NumBatches,
		TotalConsideredFiles:   rawMetrics.TotalConsideredFiles,
		TotalFilesSkipped:      rawMetrics.TotalFilesSkipped,
		PreserveInsertionOrder: rawMetrics.PreserveInsertionOrder,
	}

	// Update table handle if it changed
	if result.table_handle != nil {
		if ob.table.handle != nil {
			C.deltalake_free_table(ob.table.handle)
		}
		ob.table.handle = result.table_handle
	}

	return &OptimizeResult{
		Table:   ob.table,
		Metrics: metrics,
	}, nil
}

// Free frees the optimize builder resources
func (ob *OptimizeBuilder) Free() {
	if ob.handle != nil {
		C.deltalake_free_optimize_builder(ob.handle)
		ob.handle = nil
	}
}