package deltalake

import (
	"os"
	"path/filepath"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

func TestVersion(t *testing.T) {
	version := Version()
	assert.NotEmpty(t, version)
	t.Logf("Delta-rs version: %s", version)
}

func TestCreateDeltaLakeTable(t *testing.T) {
	// Create temporary directory
	tempDir, err := os.MkdirTemp("", "delta_test")
	require.NoError(t, err)
	defer os.RemoveAll(tempDir)

	tablePath := filepath.Join(tempDir, "test_table")
	tableName := "test_table"

	// Define table schema
	columns := []ColumnDefinition{
		{Name: "id", Type: "integer", Nullable: false},
		{Name: "name", Type: "string", Nullable: true},
		{Name: "age", Type: "integer", Nullable: true},
		{Name: "active", Type: "boolean", Nullable: true},
	}

	options := CreateTableOptions{
		TablePath: tablePath,
		SaveMode:  WriteModeOverwrite,
		Columns:   columns,
		TableName: &tableName,
	}

	// Create the table
	table, err := CreateDeltaLakeTable(options)
	require.NoError(t, err)
	require.NotNil(t, table)

	// Clean up
	table.Close()

	// Verify the table directory was created
	_, err = os.Stat(tablePath)
	assert.NoError(t, err, "Table directory should exist")
}

func TestCreateDeltaLakeTableWithoutName(t *testing.T) {
	// Create temporary directory
	tempDir, err := os.MkdirTemp("", "delta_test")
	require.NoError(t, err)
	defer os.RemoveAll(tempDir)

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

	// Define table schema
	columns := []ColumnDefinition{
		{Name: "message", Type: "string", Nullable: false},
		{Name: "timestamp", Type: "timestamp", Nullable: false},
	}

	options := CreateTableOptions{
		TablePath: tablePath,
		SaveMode:  WriteModeError,
		Columns:   columns,
		TableName: nil, // No table name
	}

	// Create the table
	table, err := CreateDeltaLakeTable(options)
	require.NoError(t, err)
	require.NotNil(t, table)

	// Clean up
	table.Close()

	// Verify the table directory was created
	_, err = os.Stat(tablePath)
	assert.NoError(t, err, "Table directory should exist")
}

func TestCreateDeltaLakeTableInvalidSaveMode(t *testing.T) {
	// Create temporary directory
	tempDir, err := os.MkdirTemp("", "delta_test")
	require.NoError(t, err)
	defer os.RemoveAll(tempDir)

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

	// Define table schema
	columns := []ColumnDefinition{
		{Name: "id", Type: "integer", Nullable: false},
	}

	options := CreateTableOptions{
		TablePath: tablePath,
		SaveMode:  WriteMode("invalid_mode"),
		Columns:   columns,
		TableName: nil,
	}

	// Create the table - should fail
	table, err := CreateDeltaLakeTable(options)
	assert.Error(t, err)
	assert.Nil(t, table)
	t.Logf("Error received: %v", err)
	assert.Contains(t, err.Error(), "Invalid save mode")
}

func TestCreateDeltaLakeTableUnsupportedDataType(t *testing.T) {
	// Create temporary directory
	tempDir, err := os.MkdirTemp("", "delta_test")
	require.NoError(t, err)
	defer os.RemoveAll(tempDir)

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

	// Define table schema with unsupported type
	columns := []ColumnDefinition{
		{Name: "id", Type: "unsupported_type", Nullable: false},
	}

	options := CreateTableOptions{
		TablePath: tablePath,
		SaveMode:  WriteModeOverwrite,
		Columns:   columns,
		TableName: nil,
	}

	// Create the table - should fail
	table, err := CreateDeltaLakeTable(options)
	assert.Error(t, err)
	assert.Nil(t, table)
	t.Logf("Error received: %v", err)
	assert.Contains(t, err.Error(), "Unsupported data type")
}

func TestOpenDeltaLakeTable(t *testing.T) {
	// Create temporary directory
	tempDir, err := os.MkdirTemp("", "delta_open_test")
	require.NoError(t, err)
	defer os.RemoveAll(tempDir)

	tablePath := filepath.Join(tempDir, "test_open_table")
	tableName := "test_open_table"

	// First create a table
	columns := []ColumnDefinition{
		{Name: "id", Type: "integer", Nullable: false},
		{Name: "name", Type: "string", Nullable: true},
		{Name: "score", Type: "double", Nullable: true},
	}

	options := CreateTableOptions{
		TablePath: tablePath,
		SaveMode:  WriteModeOverwrite,
		Columns:   columns,
		TableName: &tableName,
	}

	createdTable, err := CreateDeltaLakeTable(options)
	require.NoError(t, err)
	require.NotNil(t, createdTable)
	createdTable.Close()

	// Now test opening the table
	storageOptions := map[string]string{}
	openedTable, err := OpenDeltaLakeTable(tablePath, storageOptions)
	require.NoError(t, err)
	require.NotNil(t, openedTable)
	defer openedTable.Close()

	// Test Version method
	version, err := openedTable.Version()
	require.NoError(t, err)
	assert.Equal(t, int64(0), version, "Initial table version should be 0")

	// Test GetFiles method - empty table may not have files yet
	files, err := openedTable.GetFiles()
	require.NoError(t, err)
	// Empty tables may have 0 files, which is valid
	t.Logf("Table files: %v", files)
}

func TestOpenDeltaTableWithStorageOptions(t *testing.T) {
	// Create temporary directory
	tempDir, err := os.MkdirTemp("", "delta_storage_test")
	require.NoError(t, err)
	defer os.RemoveAll(tempDir)

	tablePath := filepath.Join(tempDir, "storage_options_table")
	tableName := "storage_options_table"

	// Create table with storage options
	columns := []ColumnDefinition{
		{Name: "id", Type: "integer", Nullable: false},
		{Name: "data", Type: "string", Nullable: true},
	}

	storageOptions := map[string]string{
		"custom_option":              "custom_value",
		"AWS_REGION":                 "us-west-2",
		"AWS_S3_ALLOW_UNSAFE_RENAME": "true",
	}

	options := CreateTableOptions{
		TablePath:      tablePath,
		SaveMode:       WriteModeOverwrite,
		Columns:        columns,
		TableName:      &tableName,
		StorageOptions: storageOptions,
	}

	createdTable, err := CreateDeltaLakeTable(options)
	require.NoError(t, err)
	require.NotNil(t, createdTable)
	createdTable.Close()

	// Open table with same storage options
	openedTable, err := OpenDeltaLakeTable(tablePath, storageOptions)
	require.NoError(t, err)
	require.NotNil(t, openedTable)
	defer openedTable.Close()

	// Test table operations
	version, err := openedTable.Version()
	require.NoError(t, err)
	assert.Equal(t, int64(0), version)

	files, err := openedTable.GetFiles()
	require.NoError(t, err)
	// Empty tables may not have files yet
	t.Logf("Table files: %v", files)
}

func TestDeltaTableVersionAndFilesIntegration(t *testing.T) {
	// Integration test for Version and GetFiles methods
	tempDir, err := os.MkdirTemp("", "delta_integration_test")
	require.NoError(t, err)
	defer os.RemoveAll(tempDir)

	tablePath := filepath.Join(tempDir, "integration_table")
	tableName := "integration_table"

	// Create table
	columns := []ColumnDefinition{
		{Name: "user_id", Type: "integer", Nullable: false},
		{Name: "username", Type: "string", Nullable: false},
		{Name: "email", Type: "string", Nullable: true},
		{Name: "created_at", Type: "timestamp", Nullable: false},
		{Name: "is_active", Type: "boolean", Nullable: true},
	}

	options := CreateTableOptions{
		TablePath: tablePath,
		SaveMode:  WriteModeOverwrite,
		Columns:   columns,
		TableName: &tableName,
	}

	createdTable, err := CreateDeltaLakeTable(options)
	require.NoError(t, err)

	// Test Version on created table
	version, err := createdTable.Version()
	require.NoError(t, err)
	assert.Equal(t, int64(0), version)

	// Test GetFiles on created table
	files, err := createdTable.GetFiles()
	require.NoError(t, err)
	// Empty tables may not have files yet
	t.Logf("Initial table files: %v", files)
	initialFiles := files

	createdTable.Close()

	// Reopen and test consistency
	storageOptions := map[string]string{}
	reopenedTable, err := OpenDeltaLakeTable(tablePath, storageOptions)
	require.NoError(t, err)
	defer reopenedTable.Close()

	// Test Version on reopened table
	reopenedVersion, err := reopenedTable.Version()
	require.NoError(t, err)
	assert.Equal(t, version, reopenedVersion, "Version should be consistent after reopening")

	// Test GetFiles on reopened table
	reopenedFiles, err := reopenedTable.GetFiles()
	require.NoError(t, err)
	assert.Equal(t, len(initialFiles), len(reopenedFiles), "File count should be consistent")

	// Verify file contents match
	fileMap := make(map[string]bool)
	for _, file := range initialFiles {
		fileMap[file] = true
	}

	for _, file := range reopenedFiles {
		assert.True(t, fileMap[file], "File %s should exist in both file lists", file)
	}
}

func TestWriteDataBasic(t *testing.T) {
	// Create temporary directory
	tempDir, err := os.MkdirTemp("", "delta_write_test")
	require.NoError(t, err)
	defer os.RemoveAll(tempDir)

	tablePath := filepath.Join(tempDir, "write_test_table")
	tableName := "write_test_table"

	// Create table
	columns := []ColumnDefinition{
		{Name: "id", Type: "integer", Nullable: false},
		{Name: "name", Type: "string", Nullable: true},
		{Name: "age", Type: "integer", Nullable: true},
	}

	options := CreateTableOptions{
		TablePath: tablePath,
		SaveMode:  WriteModeOverwrite,
		Columns:   columns,
		TableName: &tableName,
	}

	table, err := CreateDeltaLakeTable(options)
	require.NoError(t, err)
	require.NotNil(t, table)
	defer table.Close()

	// Check initial version
	version, err := table.Version()
	require.NoError(t, err)
	assert.Equal(t, int64(0), version)

	// Prepare data to write
	writeData := WriteData{
		Schema: columns,
		Data: [][]interface{}{
			{1, "Alice", 25},
			{2, "Bob", 30},
			{3, "Charlie", 35},
		},
	}

	// Write data to table
	updatedTable, err := table.Write(writeData, WriteModeAppend)
	require.NoError(t, err)
	require.NotNil(t, updatedTable)
	defer updatedTable.Close()

	// Check updated version
	newVersion, err := updatedTable.Version()
	require.NoError(t, err)
	assert.Equal(t, int64(1), newVersion, "Version should increment after write")

	// Check files were created
	files, err := updatedTable.GetFiles()
	require.NoError(t, err)
	assert.Greater(t, len(files), 0, "Table should have data files after write")
	t.Logf("Files after write: %v", files)
}

func TestWriteDataMultipleBatches(t *testing.T) {
	// Create temporary directory
	tempDir, err := os.MkdirTemp("", "delta_write_batches_test")
	require.NoError(t, err)
	defer os.RemoveAll(tempDir)

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

	// Create table
	columns := []ColumnDefinition{
		{Name: "id", Type: "integer", Nullable: false},
		{Name: "message", Type: "string", Nullable: true},
		{Name: "value", Type: "double", Nullable: true},
	}

	options := CreateTableOptions{
		TablePath: tablePath,
		SaveMode:  WriteModeOverwrite,
		Columns:   columns,
	}

	table, err := CreateDeltaLakeTable(options)
	require.NoError(t, err)
	defer table.Close()

	// First batch
	batch1 := WriteData{
		Schema: columns,
		Data: [][]interface{}{
			{1, "Hello", 10.5},
			{2, "World", 20.0},
		},
	}

	table1, err := table.Write(batch1, WriteModeAppend)
	require.NoError(t, err)
	defer table1.Close()

	version1, err := table1.Version()
	require.NoError(t, err)
	assert.Equal(t, int64(1), version1)

	files1, err := table1.GetFiles()
	require.NoError(t, err)
	assert.Greater(t, len(files1), 0)

	// Second batch
	batch2 := WriteData{
		Schema: columns,
		Data: [][]interface{}{
			{3, "Delta", 30.5},
			{4, "Lake", 40.0},
			{5, "Test", 50.25},
		},
	}

	table2, err := table1.Write(batch2, WriteModeAppend)
	require.NoError(t, err)
	defer table2.Close()

	version2, err := table2.Version()
	require.NoError(t, err)
	assert.Equal(t, int64(2), version2)

	files2, err := table2.GetFiles()
	require.NoError(t, err)
	assert.Greater(t, len(files2), len(files1), "Should have more files after second write")
	t.Logf("Files after first batch: %d, after second batch: %d", len(files1), len(files2))
}

func TestQueryTableExecutesSQL(t *testing.T) {
	// Create temporary directory
	tempDir, err := os.MkdirTemp("", "delta_query_test")
	require.NoError(t, err)
	defer os.RemoveAll(tempDir)

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

	columns := []ColumnDefinition{
		{Name: "id", Type: "integer", Nullable: false},
		{Name: "name", Type: "string", Nullable: false},
		{Name: "age", Type: "integer", Nullable: false},
		{Name: "year", Type: "integer", Nullable: false},
	}

	options := CreateTableOptions{
		TablePath: tablePath,
		SaveMode:  WriteModeOverwrite,
		Columns:   columns,
	}

	table, err := CreateDeltaLakeTable(options)
	require.NoError(t, err)
	require.NotNil(t, table)
	defer table.Close()

	writeData := WriteData{
		Schema: columns,
		Data: [][]interface{}{
			{1, "Alice", 25, 2020},
			{2, "Bob", 30, 2021},
			{3, "Charlie", 35, 2021},
			{4, "Diana", 40, 2022},
		},
	}

	tableWithData, err := table.Write(writeData, WriteModeAppend)
	require.NoError(t, err)
	require.NotNil(t, tableWithData)
	defer tableWithData.Close()

	rows, err := tableWithData.Query("SELECT id, name FROM delta_table WHERE age >= 30 ORDER BY id", nil)
	require.NoError(t, err)
	require.Len(t, rows, 3, "Expected three rows matching age >= 30")

	assert.Equal(t, "Bob", rows[0]["name"])
	assert.Equal(t, "Charlie", rows[1]["name"])
	assert.Equal(t, "Diana", rows[2]["name"])

	getNumeric := func(value interface{}) float64 {
		switch v := value.(type) {
		case float64:
			return v
		case int:
			return float64(v)
		case int64:
			return float64(v)
		default:
			t.Fatalf("unexpected numeric type: %T", value)
		}
		return 0
	}

	assert.Equal(t, float64(2), getNumeric(rows[0]["id"]))
	assert.Equal(t, float64(3), getNumeric(rows[1]["id"]))
	assert.Equal(t, float64(4), getNumeric(rows[2]["id"]))

	aliasRows, err := tableWithData.Query(
		"SELECT COUNT(1) AS cnt FROM people WHERE year = 2021",
		&QueryOptions{TableAlias: "people"},
	)
	require.NoError(t, err)
	require.Len(t, aliasRows, 1)

	cntValue := getNumeric(aliasRows[0]["cnt"])
	assert.Equal(t, float64(2), cntValue, "Expected two rows for year 2021")
}

func TestWriteDataDifferentTypes(t *testing.T) {
	// Create temporary directory
	tempDir, err := os.MkdirTemp("", "delta_write_types_test")
	require.NoError(t, err)
	defer os.RemoveAll(tempDir)

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

	// Create table with various data types
	columns := []ColumnDefinition{
		{Name: "int_col", Type: "integer", Nullable: false},
		{Name: "long_col", Type: "long", Nullable: true},
		{Name: "string_col", Type: "string", Nullable: true},
		{Name: "bool_col", Type: "boolean", Nullable: true},
		{Name: "float_col", Type: "float", Nullable: true},
		{Name: "double_col", Type: "double", Nullable: true},
	}

	options := CreateTableOptions{
		TablePath: tablePath,
		SaveMode:  WriteModeOverwrite,
		Columns:   columns,
	}

	table, err := CreateDeltaLakeTable(options)
	require.NoError(t, err)
	defer table.Close()

	// Write data with different types
	writeData := WriteData{
		Schema: columns,
		Data: [][]interface{}{
			{123, int64(9876543210), "Hello", true, float32(3.14), 2.718},
			{456, int64(1234567890), "World", false, float32(1.41), 1.414},
		},
	}

	updatedTable, err := table.Write(writeData, WriteModeAppend)
	require.NoError(t, err)
	require.NotNil(t, updatedTable)
	defer updatedTable.Close()

	// Verify write was successful
	version, err := updatedTable.Version()
	require.NoError(t, err)
	assert.Equal(t, int64(1), version)

	files, err := updatedTable.GetFiles()
	require.NoError(t, err)
	assert.Greater(t, len(files), 0)
}

func TestWriteDataInvalidSchema(t *testing.T) {
	// Create temporary directory
	tempDir, err := os.MkdirTemp("", "delta_write_invalid_test")
	require.NoError(t, err)
	defer os.RemoveAll(tempDir)

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

	// Create table
	columns := []ColumnDefinition{
		{Name: "id", Type: "integer", Nullable: false},
		{Name: "name", Type: "string", Nullable: true},
	}

	options := CreateTableOptions{
		TablePath: tablePath,
		SaveMode:  WriteModeOverwrite,
		Columns:   columns,
	}

	table, err := CreateDeltaLakeTable(options)
	require.NoError(t, err)
	defer table.Close()

	// Try to write data with mismatched schema
	invalidData := WriteData{
		Schema: []ColumnDefinition{
			{Name: "id", Type: "string", Nullable: false}, // Wrong type
			{Name: "name", Type: "string", Nullable: true},
		},
		Data: [][]interface{}{
			{"invalid", "test"},
		},
	}

	_, err = table.Write(invalidData, WriteModeAppend)
	assert.Error(t, err, "Should fail with schema mismatch")
	t.Logf("Expected error: %v", err)
}

func TestWriteDataEmptyData(t *testing.T) {
	// Create temporary directory
	tempDir, err := os.MkdirTemp("", "delta_write_empty_test")
	require.NoError(t, err)
	defer os.RemoveAll(tempDir)

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

	// Create table
	columns := []ColumnDefinition{
		{Name: "id", Type: "integer", Nullable: false},
		{Name: "value", Type: "string", Nullable: true},
	}

	options := CreateTableOptions{
		TablePath: tablePath,
		SaveMode:  WriteModeOverwrite,
		Columns:   columns,
	}

	table, err := CreateDeltaLakeTable(options)
	require.NoError(t, err)
	defer table.Close()

	// Write empty data
	emptyData := WriteData{
		Schema: columns,
		Data:   [][]interface{}{}, // Empty data
	}

	updatedTable, err := table.Write(emptyData, WriteModeAppend)
	require.NoError(t, err)
	require.NotNil(t, updatedTable)
	defer updatedTable.Close()

	// Version should still increment even with empty data
	version, err := updatedTable.Version()
	require.NoError(t, err)
	assert.Equal(t, int64(1), version)
}

func TestWriteDataOverwriteMode(t *testing.T) {
	// Create temporary directory
	tempDir, err := os.MkdirTemp("", "delta_write_overwrite_test")
	require.NoError(t, err)
	defer os.RemoveAll(tempDir)

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

	// Create table
	columns := []ColumnDefinition{
		{Name: "id", Type: "integer", Nullable: false},
		{Name: "data", Type: "string", Nullable: true},
	}

	options := CreateTableOptions{
		TablePath: tablePath,
		SaveMode:  WriteModeOverwrite,
		Columns:   columns,
	}

	table, err := CreateDeltaLakeTable(options)
	require.NoError(t, err)
	defer table.Close()

	// First write
	data1 := WriteData{
		Schema: columns,
		Data: [][]interface{}{
			{1, "first"},
			{2, "second"},
		},
	}

	table1, err := table.Write(data1, WriteModeAppend)
	require.NoError(t, err)
	defer table1.Close()

	files1, err := table1.GetFiles()
	require.NoError(t, err)
	assert.Greater(t, len(files1), 0)

	// Overwrite with new data
	data2 := WriteData{
		Schema: columns,
		Data: [][]interface{}{
			{3, "third"},
		},
	}

	table2, err := table1.Write(data2, WriteModeOverwrite)
	require.NoError(t, err)
	defer table2.Close()

	version2, err := table2.Version()
	require.NoError(t, err)
	assert.Equal(t, int64(2), version2)

	files2, err := table2.GetFiles()
	require.NoError(t, err)
	// After overwrite, we should have files (could be different from append)
	assert.Greater(t, len(files2), 0)
	t.Logf("Files before overwrite: %d, after overwrite: %d", len(files1), len(files2))
}

func TestWriteDataNullValues(t *testing.T) {
	// Create temporary directory
	tempDir, err := os.MkdirTemp("", "delta_write_null_test")
	require.NoError(t, err)
	defer os.RemoveAll(tempDir)

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

	// Create table with nullable columns
	columns := []ColumnDefinition{
		{Name: "id", Type: "integer", Nullable: false},
		{Name: "optional_name", Type: "string", Nullable: true},
		{Name: "optional_value", Type: "double", Nullable: true},
	}

	options := CreateTableOptions{
		TablePath: tablePath,
		SaveMode:  WriteModeOverwrite,
		Columns:   columns,
	}

	table, err := CreateDeltaLakeTable(options)
	require.NoError(t, err)
	defer table.Close()

	// Write data with some nullable fields (skip null values for now)
	writeData := WriteData{
		Schema: columns,
		Data: [][]interface{}{
			{1, "Alice", 100.0},
			{2, "Bob", 200.0},
		},
	}

	updatedTable, err := table.Write(writeData, WriteModeAppend)
	require.NoError(t, err)
	require.NotNil(t, updatedTable)
	defer updatedTable.Close()

	// Verify write was successful
	version, err := updatedTable.Version()
	require.NoError(t, err)
	assert.Equal(t, int64(1), version)

	files, err := updatedTable.GetFiles()
	require.NoError(t, err)
	assert.Greater(t, len(files), 0)
}

func TestWriteDataInvalidSaveMode(t *testing.T) {
	// Create temporary directory
	tempDir, err := os.MkdirTemp("", "delta_write_mode_test")
	require.NoError(t, err)
	defer os.RemoveAll(tempDir)

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

	// Create table
	columns := []ColumnDefinition{
		{Name: "id", Type: "integer", Nullable: false},
	}

	options := CreateTableOptions{
		TablePath: tablePath,
		SaveMode:  WriteModeOverwrite,
		Columns:   columns,
	}

	table, err := CreateDeltaLakeTable(options)
	require.NoError(t, err)
	defer table.Close()

	// Try to write with invalid save mode
	writeData := WriteData{
		Schema: columns,
		Data: [][]interface{}{
			{1},
		},
	}

	_, err = table.Write(writeData, WriteMode("invalid_mode"))
	assert.Error(t, err, "Should fail with invalid save mode")
	assert.Contains(t, err.Error(), "Invalid save mode", "Error should mention invalid save mode")
}

func TestDeleteWhereRemovesRows(t *testing.T) {
	// Create temporary directory
	tempDir, err := os.MkdirTemp("", "delta_delete_where_test")
	require.NoError(t, err)
	defer os.RemoveAll(tempDir)

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

	// Create table schema
	columns := []ColumnDefinition{
		{Name: "id", Type: "integer", Nullable: false},
		{Name: "name", Type: "string", Nullable: true},
	}

	options := CreateTableOptions{
		TablePath: tablePath,
		SaveMode:  WriteModeOverwrite,
		Columns:   columns,
	}

	table, err := CreateDeltaLakeTable(options)
	require.NoError(t, err)
	require.NotNil(t, table)
	defer table.Close()

	// Seed table with rows
	writeData := WriteData{
		Schema: columns,
		Data: [][]interface{}{
			{1, "Alice"},
			{2, "Bob"},
			{3, "Charlie"},
		},
	}

	tableWithData, err := table.Write(writeData, WriteModeAppend)
	require.NoError(t, err)
	require.NotNil(t, tableWithData)
	defer tableWithData.Close()

	versionBefore, err := tableWithData.Version()
	require.NoError(t, err)
	assert.Equal(t, int64(1), versionBefore, "Version should increment after initial write")

	deleteResult, err := tableWithData.DeleteWhere("id = 2")
	require.NoError(t, err)
	require.NotNil(t, deleteResult)
	require.NotNil(t, deleteResult.Table)
	defer deleteResult.Table.Close()

	assert.Equal(t, int64(1), deleteResult.Metrics.NumDeletedRows, "Should delete exactly one row")
	assert.GreaterOrEqual(t, deleteResult.Metrics.NumRemovedFiles, int64(0))
	assert.GreaterOrEqual(t, deleteResult.Metrics.NumAddedFiles, int64(0))

	versionAfter, err := deleteResult.Table.Version()
	require.NoError(t, err)
	assert.Equal(t, versionBefore+1, versionAfter, "Delete should advance the table version")

	remainingRows, err := deleteResult.Table.ReadData(10)
	require.NoError(t, err)
	assert.Equal(t, 2, len(remainingRows), "Two rows should remain after deletion")

	for _, row := range remainingRows {
		idValue, exists := row["id"]
		require.True(t, exists, "Row should contain id column")

		switch v := idValue.(type) {
		case float64:
			assert.NotEqual(t, float64(2), v, "Deleted row should not be present")
		case int:
			assert.NotEqual(t, 2, v, "Deleted row should not be present")
		case int64:
			assert.NotEqual(t, int64(2), v, "Deleted row should not be present")
		default:
			t.Fatalf("unexpected id type: %T", idValue)
		}
	}
}

func TestTableExists(t *testing.T) {
	// Create temporary directory
	tempDir, err := os.MkdirTemp("", "delta_test_exists")
	require.NoError(t, err)
	defer os.RemoveAll(tempDir)

	existingTablePath := filepath.Join(tempDir, "existing_table")
	nonExistentTablePath := filepath.Join(tempDir, "nonexistent_table")

	// Test 1: Check non-existent table
	t.Run("NonExistentTable", func(t *testing.T) {
		exists, err := TableExists(nonExistentTablePath, nil)
		require.NoError(t, err)
		assert.False(t, exists, "Non-existent table should not exist")
	})

	// Test 2: Create a table and then check if it exists
	t.Run("ExistingTable", func(t *testing.T) {
		// Define table schema
		columns := []ColumnDefinition{
			{Name: "id", Type: "integer", Nullable: true},
			{Name: "name", Type: "string", Nullable: true},
		}

		options := CreateTableOptions{
			TablePath: existingTablePath,
			SaveMode:  WriteModeOverwrite,
			Columns:   columns,
		}

		// Create the table
		table, err := CreateDeltaLakeTable(options)
		require.NoError(t, err)
		require.NotNil(t, table)
		defer table.Close()

		// Now check if the table exists
		exists, err := TableExists(existingTablePath, nil)
		require.NoError(t, err)
		assert.True(t, exists, "Created table should exist")
	})

	// Test 3: Check with storage options
	t.Run("WithStorageOptions", func(t *testing.T) {
		// Define table schema
		columns := []ColumnDefinition{
			{Name: "id", Type: "integer", Nullable: true},
			{Name: "message", Type: "string", Nullable: true},
		}

		// Use storage options
		storageOptions := map[string]string{
			"AWS_REGION": "us-west-2",
		}

		options := CreateTableOptions{
			TablePath:      existingTablePath,
			SaveMode:       WriteModeOverwrite,
			Columns:        columns,
			StorageOptions: storageOptions,
		}

		// Create the table with storage options
		table, err := CreateDeltaLakeTable(options)
		require.NoError(t, err)
		require.NotNil(t, table)
		defer table.Close()

		// Check if table exists with storage options
		exists, err := TableExists(existingTablePath, storageOptions)
		require.NoError(t, err)
		assert.True(t, exists, "Table should exist with storage options")

		// Check if table exists without storage options (should still work)
		exists, err = TableExists(existingTablePath, nil)
		require.NoError(t, err)
		assert.True(t, exists, "Table should exist even without storage options")
	})

	// Test 4: Error cases
	t.Run("ErrorCases", func(t *testing.T) {
		// Test empty path
		exists, err := TableExists("", nil)
		assert.Error(t, err, "Empty path should return error")
		assert.False(t, exists, "Empty path should not exist")
		assert.Contains(t, err.Error(), "table path cannot be empty", "Error should mention empty path")
	})

	// Test 5: Non-Delta directory (regular directory without Delta files)
	t.Run("NonDeltaDirectory", func(t *testing.T) {
		nonDeltaPath := filepath.Join(tempDir, "non_delta_dir")
		err := os.MkdirAll(nonDeltaPath, 0755)
		require.NoError(t, err)

		// This should return false, not error, since it's not a Delta table
		exists, err := TableExists(nonDeltaPath, nil)
		require.NoError(t, err)
		assert.False(t, exists, "Non-Delta directory should not exist as Delta table")
	})
}

func TestTableExistsIntegrationWithOpen(t *testing.T) {
	// Create temporary directory
	tempDir, err := os.MkdirTemp("", "delta_test_integration")
	require.NoError(t, err)
	defer os.RemoveAll(tempDir)

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

	// First check that table doesn't exist
	exists, err := TableExists(tablePath, nil)
	require.NoError(t, err)
	assert.False(t, exists, "Table should not exist initially")

	// Try to open non-existent table (should fail)
	_, err = OpenDeltaLakeTable(tablePath, nil)
	assert.Error(t, err, "Opening non-existent table should fail")

	// Create the table
	columns := []ColumnDefinition{
		{Name: "id", Type: "integer", Nullable: false},
		{Name: "data", Type: "string", Nullable: true},
	}

	options := CreateTableOptions{
		TablePath: tablePath,
		SaveMode:  WriteModeOverwrite,
		Columns:   columns,
	}

	createdTable, err := CreateDeltaLakeTable(options)
	require.NoError(t, err)
	require.NotNil(t, createdTable)
	createdTable.Close()

	// Now check that table exists
	exists, err = TableExists(tablePath, nil)
	require.NoError(t, err)
	assert.True(t, exists, "Table should exist after creation")

	// Now try to open existing table (should succeed)
	openedTable, err := OpenDeltaLakeTable(tablePath, nil)
	require.NoError(t, err)
	require.NotNil(t, openedTable)
	defer openedTable.Close()

	// Verify it's the same table by checking version
	version, err := openedTable.Version()
	require.NoError(t, err)
	assert.GreaterOrEqual(t, version, int64(0), "Table version should be valid")
}