Support Iceberg DROP SCHEMA CASCADE fallback (#638)

* Support Iceberg drop schema cascade fallback

* Address Iceberg drop schema fallback review

* Fix Iceberg drop schema lint errors

Author: EDsCODE

server/conn.go

@@ -1448,17 +1448,9 @@ func (c *clientConn) handleQuery(body []byte) error {
 		runExec := func() (ExecResult, error) {
 			execResult, err := c.executor.ExecContext(ctx, query)
 			if err != nil {
-				// Retry ALTER TABLE as ALTER VIEW if target is a view
-				if isAlterTableNotTableError(err) {
-					if alteredQuery, ok := transpiler.ConvertAlterTableToAlterView(query); ok {
-						return c.executor.ExecContext(ctx, alteredQuery)
-					}
-				}
-				// Retry DROP TABLE as DROP VIEW if target is a view
-				if isDropTableOnViewError(err) {
-					if alteredQuery, ok := transpiler.ConvertDropTableToDropView(query); ok {
-						return c.executor.ExecContext(ctx, alteredQuery)
-					}
+				fallbackResult, handled, fallbackErr := c.execCompatibilityFallback(ctx, query, err)
+				if handled {
+					return fallbackResult, fallbackErr
 				}
 			}
 			return execResult, err

server/exec_fallback.go

@@ -0,0 +1,34 @@
+package server
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/posthog/duckgres/transpiler"
+)
+
+func (c *clientConn) execCompatibilityFallback(ctx context.Context, query string, execErr error) (ExecResult, bool, error) {
+	if isAlterTableNotTableError(execErr) {
+		if alteredQuery, ok := transpiler.ConvertAlterTableToAlterView(query); ok {
+			result, err := c.executor.ExecContext(ctx, alteredQuery)
+			return result, true, err
+		}
+	}
+
+	if isDropTableOnViewError(execErr) {
+		if alteredQuery, ok := transpiler.ConvertDropTableToDropView(query); ok {
+			result, err := c.executor.ExecContext(ctx, alteredQuery)
+			return result, true, err
+		}
+	}
+
+	if isIcebergDropSchemaCascadeUnsupported(execErr) {
+		result, err := c.dropIcebergSchemaCascade(ctx, query)
+		if err != nil {
+			return nil, true, fmt.Errorf("iceberg DROP SCHEMA CASCADE fallback failed: %w", err)
+		}
+		return result, true, nil
+	}
+
+	return nil, false, nil
+}

server/exec_fallback_test.go

@@ -0,0 +1,53 @@
+package server
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+)
+
+func TestExecCompatibilityFallbackReturnsIcebergFallbackError(t *testing.T) {
+	execErr := errors.New("Not implemented Error: DROP SCHEMA <schema_name> CASCADE is not supported for Iceberg schemas currently")
+	c := &clientConn{executor: &failingFallbackExecutor{}}
+
+	_, handled, err := c.execCompatibilityFallback(context.Background(), "DROP SCHEMA IF EXISTS stripe CASCADE", execErr)
+	if !handled {
+		t.Fatal("expected Iceberg fallback to handle unsupported DROP SCHEMA CASCADE error")
+	}
+	if err == nil || !strings.Contains(err.Error(), "iceberg DROP SCHEMA CASCADE fallback failed") {
+		t.Fatalf("fallback error = %v, want contextual fallback failure", err)
+	}
+}
+
+type failingFallbackExecutor struct {
+	noopProfiling
+}
+
+func (e *failingFallbackExecutor) QueryContext(context.Context, string, ...any) (RowSet, error) {
+	return nil, errors.New("settings query failed")
+}
+
+func (e *failingFallbackExecutor) ExecContext(context.Context, string, ...any) (ExecResult, error) {
+	return nil, errors.New("unexpected exec")
+}
+
+func (e *failingFallbackExecutor) Query(string, ...any) (RowSet, error) {
+	return nil, errors.New("not implemented")
+}
+
+func (e *failingFallbackExecutor) Exec(string, ...any) (ExecResult, error) {
+	return nil, errors.New("not implemented")
+}
+
+func (e *failingFallbackExecutor) ConnContext(context.Context) (RawConn, error) {
+	return nil, errors.New("not implemented")
+}
+
+func (e *failingFallbackExecutor) PingContext(context.Context) error {
+	return errors.New("not implemented")
+}
+
+func (e *failingFallbackExecutor) Close() error {
+	return nil
+}

server/iceberg_drop_schema.go

@@ -0,0 +1,192 @@
+package server
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/posthog/duckgres/server/iceberg"
+	"github.com/posthog/duckgres/server/sqlcore"
+)
+
+type dropSchemaCascadeTarget struct {
+	Catalog string
+	Schema  string
+}
+
+func parseDropSchemaCascadeTarget(query string) (dropSchemaCascadeTarget, bool) {
+	s := strings.TrimSpace(sqlcore.StripLeadingComments(query))
+	s = strings.TrimSpace(strings.TrimSuffix(s, ";"))
+
+	pos, ok := consumeSQLKeyword(s, 0, "DROP")
+	if !ok {
+		return dropSchemaCascadeTarget{}, false
+	}
+	pos, ok = consumeSQLKeyword(s, pos, "SCHEMA")
+	if !ok {
+		return dropSchemaCascadeTarget{}, false
+	}
+
+	target := dropSchemaCascadeTarget{}
+	if next, ok := consumeSQLKeyword(s, pos, "IF"); ok {
+		next, ok = consumeSQLKeyword(s, next, "EXISTS")
+		if !ok {
+			return dropSchemaCascadeTarget{}, false
+		}
+		pos = next
+	}
+
+	targetSQL, ok := trimTrailingSQLKeyword(s[pos:], "CASCADE")
+	if !ok {
+		return dropSchemaCascadeTarget{}, false
+	}
+	parts, ok := sqlcore.ParseQualifiedIdentifier(targetSQL)
+	if !ok || len(parts) == 0 || len(parts) > 2 {
+		return dropSchemaCascadeTarget{}, false
+	}
+	if len(parts) == 1 {
+		target.Schema = parts[0]
+	} else {
+		target.Catalog = parts[0]
+		target.Schema = parts[1]
+	}
+	if target.Schema == "" || target.Catalog == "" && len(parts) == 2 {
+		return dropSchemaCascadeTarget{}, false
+	}
+	return target, true
+}
+
+func consumeSQLKeyword(s string, pos int, keyword string) (int, bool) {
+	pos = skipSQLSpace(s, pos)
+	end := pos + len(keyword)
+	if end > len(s) || !strings.EqualFold(s[pos:end], keyword) {
+		return pos, false
+	}
+	if end < len(s) && isSQLIdentChar(s[end]) {
+		return pos, false
+	}
+	return end, true
+}
+
+func trimTrailingSQLKeyword(s string, keyword string) (string, bool) {
+	s = strings.TrimSpace(s)
+	end := len(s) - len(keyword)
+	if end < 0 || !strings.EqualFold(s[end:], keyword) {
+		return "", false
+	}
+	if end > 0 && !isSQLSpace(s[end-1]) {
+		return "", false
+	}
+	target := strings.TrimSpace(s[:end])
+	return target, target != ""
+}
+
+func skipSQLSpace(s string, pos int) int {
+	for pos < len(s) && isSQLSpace(s[pos]) {
+		pos++
+	}
+	return pos
+}
+
+func isSQLSpace(ch byte) bool {
+	switch ch {
+	case ' ', '\t', '\n', '\r', '\f':
+		return true
+	default:
+		return false
+	}
+}
+
+func isSQLIdentChar(ch byte) bool {
+	return ch == '_' || ch >= '0' && ch <= '9' || ch >= 'A' && ch <= 'Z' || ch >= 'a' && ch <= 'z'
+}
+
+func isIcebergDropSchemaCascadeUnsupported(err error) bool {
+	if err == nil {
+		return false
+	}
+	return strings.Contains(err.Error(), "DROP SCHEMA <schema_name> CASCADE is not supported for Iceberg schemas currently")
+}
+
+func (c *clientConn) dropIcebergSchemaCascade(ctx context.Context, query string) (ExecResult, error) {
+	target, ok := parseDropSchemaCascadeTarget(query)
+	if !ok {
+		return nil, fmt.Errorf("not a DROP SCHEMA CASCADE statement")
+	}
+
+	catalog := target.Catalog
+	if catalog == "" {
+		defaultCatalog, err := c.currentSearchPathCatalog(ctx)
+		if err != nil {
+			return nil, err
+		}
+		catalog = defaultCatalog
+	}
+	if !strings.EqualFold(catalog, iceberg.CatalogName) {
+		return nil, fmt.Errorf("DROP SCHEMA CASCADE fallback only supports iceberg catalog, got %q", catalog)
+	}
+
+	rows, err := c.executor.QueryContext(ctx, `
+		SELECT table_name
+		FROM information_schema.tables
+		WHERE table_catalog = 'iceberg'
+		AND table_schema = ?
+		ORDER BY table_name
+	`, target.Schema)
+	if err != nil {
+		return nil, fmt.Errorf("list iceberg schema tables: %w", err)
+	}
+	defer func() { _ = rows.Close() }()
+
+	var tables []string
+	for rows.Next() {
+		var table string
+		if err := rows.Scan(&table); err != nil {
+			return nil, fmt.Errorf("scan iceberg schema table: %w", err)
+		}
+		tables = append(tables, table)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("list iceberg schema tables: %w", err)
+	}
+
+	for _, table := range tables {
+		dropTable := fmt.Sprintf("DROP TABLE IF EXISTS %s.%s.%s",
+			sqlcore.QuoteIdentifier(iceberg.CatalogName),
+			sqlcore.QuoteIdentifier(target.Schema),
+			sqlcore.QuoteIdentifier(table),
+		)
+		if _, err := c.executor.ExecContext(ctx, dropTable); err != nil {
+			return nil, fmt.Errorf("drop iceberg table %s.%s: %w", target.Schema, table, err)
+		}
+	}
+
+	dropSchema := fmt.Sprintf("DROP SCHEMA IF EXISTS %s.%s",
+		sqlcore.QuoteIdentifier(iceberg.CatalogName),
+		sqlcore.QuoteIdentifier(target.Schema),
+	)
+	return c.executor.ExecContext(ctx, dropSchema)
+}
+
+func (c *clientConn) currentSearchPathCatalog(ctx context.Context) (string, error) {
+	rows, err := c.executor.QueryContext(ctx, `
+		SELECT value
+		FROM duckdb_settings()
+		WHERE name = 'search_path'
+	`)
+	if err != nil {
+		return "", fmt.Errorf("read search_path: %w", err)
+	}
+	defer func() { _ = rows.Close() }()
+	if !rows.Next() {
+		return "", rows.Err()
+	}
+	var searchPath string
+	if err := rows.Scan(&searchPath); err != nil {
+		return "", fmt.Errorf("scan search_path: %w", err)
+	}
+	if err := rows.Err(); err != nil {
+		return "", err
+	}
+	return sqlcore.CatalogFromSearchPath(searchPath), nil
+}