Fix

Author: ilyario

internal/connectors/trino.go

@@ -893,6 +893,16 @@ func quoteTrinoIdentifier(name string) string {
 	return name
 }
 
+// unwrapMessageDataForColumns returns the map to use for column lookups.
+// If msgData is {"value": {...}, "_metadata": {...}}, returns the inner value map.
+// Otherwise returns msgData as-is (plain columnar format).
+func unwrapMessageDataForColumns(msgData map[string]interface{}) map[string]interface{} {
+	if v, ok := msgData["value"].(map[string]interface{}); ok && len(msgData) <= 2 {
+		return v
+	}
+	return msgData
+}
+
 // extractDataAndMetadata extracts data and _metadata from a rawMode message.
 // If msg.Data is {"value": ..., "_metadata": ...}, uses those; otherwise uses whole data and msg.Metadata.
 func extractDataAndMetadata(msg *types.Message) (dataStr, metaStr string) {
@@ -1103,13 +1113,24 @@ func (t *TrinoSinkConnector) Write(ctx context.Context, messages <-chan *types.M
 	}
 }
 
+// hasRawModeColumns checks if the table has data and _metadata columns (required for raw mode).
+func (t *TrinoSinkConnector) hasRawModeColumns(columns []TableColumnInfo) bool {
+	hasData, hasMeta := false, false
+	for _, col := range columns {
+		if col.Name == "data" {
+			hasData = true
+		}
+		if col.Name == "_metadata" {
+			hasMeta = true
+		}
+	}
+	return hasData && hasMeta
+}
+
 func (t *TrinoSinkConnector) executeBatch(ctx context.Context, batch []*types.Message) error {
 	if len(batch) == 0 {
 		return nil
 	}
-	if t.rawMode() {
-		return t.executeBatchRaw(ctx, batch)
-	}
 
 	// Get table columns (use cached if available, otherwise fetch)
 	t.columnsMu.RLock()
@@ -1128,6 +1149,16 @@ func (t *TrinoSinkConnector) executeBatch(ctx context.Context, batch []*types.Me
 		t.columnsMu.Unlock()
 	}
 
+	// Use raw mode only when configured AND table has data/_metadata columns.
+	// If rawMode is true but table has different schema (e.g. columnar), fall back to schema-based insert.
+	if t.rawMode() && t.hasRawModeColumns(tableColumns) {
+		return t.executeBatchRaw(ctx, batch)
+	}
+	if t.rawMode() && !t.hasRawModeColumns(tableColumns) {
+		t.logger.Info("rawMode is true but target table does not have data/_metadata columns, using schema-based insert",
+			"table", fmt.Sprintf("%s.%s.%s", t.config.Catalog, t.config.Schema, t.config.Table))
+	}
+
 	// Create a map for fast lookup of table columns by name
 	tableColumnsMap := make(map[string]TableColumnInfo)
 	for _, col := range tableColumns {
@@ -1146,8 +1177,9 @@ func (t *TrinoSinkConnector) executeBatch(ctx context.Context, batch []*types.Me
 			return fmt.Errorf("failed to parse message JSON at index %d: %w", i, err)
 		}
 
-		// Collect all keys from all messages
-		for k := range msgData {
+		// Collect all keys from all messages (unwrap {"value": {...}} format if present)
+		dataForColumns := unwrapMessageDataForColumns(msgData)
+		for k := range dataForColumns {
 			allMessageKeys[k] = true
 		}
 	}
@@ -1167,13 +1199,14 @@ func (t *TrinoSinkConnector) executeBatch(ctx context.Context, batch []*types.Me
 		var firstMsgData map[string]interface{}
 		if err := json.Unmarshal(batch[0].Data, &firstMsgData); err == nil {
 			// Log first message structure (limit size to avoid huge logs)
-			firstMsgJSON, _ := json.Marshal(firstMsgData)
+			dataForPreview := unwrapMessageDataForColumns(firstMsgData)
+			firstMsgJSON, _ := json.Marshal(dataForPreview)
 			msgPreview := string(firstMsgJSON)
 			if len(msgPreview) > 500 {
 				msgPreview = msgPreview[:500] + "..."
 			}
-			firstMsgKeys := make([]string, 0, len(firstMsgData))
-			for k := range firstMsgData {
+			firstMsgKeys := make([]string, 0, len(dataForPreview))
+			for k := range dataForPreview {
 				firstMsgKeys = append(firstMsgKeys, k)
 			}
 			t.logger.Info("First message in batch (preview)",
@@ -1244,6 +1277,9 @@ func (t *TrinoSinkConnector) executeBatch(ctx context.Context, batch []*types.Me
 			return fmt.Errorf("failed to parse message JSON: %w", err)
 		}
 
+		// Unwrap {"value": {...}} format if present (e.g. from Kafka with rawMode source)
+		data = unwrapMessageDataForColumns(data)
+
 		// Build values for this row - use values from message or NULL
 		values := make([]string, len(columnsToUse))
 		for j, col := range columnsToUse {

internal/connectors/trino_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package connectors
 
 import (
+	"encoding/json"
 	"strings"
 	"testing"
 	"time"
@@ -402,6 +403,94 @@ func TestExtractDataAndMetadata(t *testing.T) {
 	}
 }
 
+func TestUnwrapMessageDataForColumns(t *testing.T) {
+	tests := []struct {
+		name     string
+		msgData  string
+		wantKeys []string
+	}{
+		{
+			name:     "wrapped_value_format",
+			msgData:  `{"value":{"id":1,"name":"foo"},"_metadata":{"offset":10}}`,
+			wantKeys: []string{"id", "name"},
+		},
+		{
+			name:     "plain_columnar_format",
+			msgData:  `{"id":2,"name":"bar","amount":100}`,
+			wantKeys: []string{"id", "name", "amount"},
+		},
+		{
+			name:     "value_not_map_returns_as_is",
+			msgData:  `{"value":"string","_metadata":{}}`,
+			wantKeys: []string{"value", "_metadata"},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var m map[string]interface{}
+			require.NoError(t, json.Unmarshal([]byte(tt.msgData), &m))
+			got := unwrapMessageDataForColumns(m)
+			gotKeys := make([]string, 0, len(got))
+			for k := range got {
+				gotKeys = append(gotKeys, k)
+			}
+			assert.ElementsMatch(t, tt.wantKeys, gotKeys)
+		})
+	}
+}
+
+func TestTrinoSinkConnector_hasRawModeColumns(t *testing.T) {
+	connector := NewTrinoSinkConnector(&v1.TrinoSinkSpec{
+		ServerURL: "http://localhost:8080",
+		Catalog:   "test",
+		Schema:    "test",
+		Table:     "test",
+	})
+	tests := []struct {
+		name    string
+		columns []TableColumnInfo
+		want    bool
+	}{
+		{
+			name: "has_both",
+			columns: []TableColumnInfo{
+				{Name: "data", Type: "varchar"},
+				{Name: "_metadata", Type: "varchar"},
+			},
+			want: true,
+		},
+		{
+			name: "missing_data",
+			columns: []TableColumnInfo{
+				{Name: "_metadata", Type: "varchar"},
+			},
+			want: false,
+		},
+		{
+			name: "missing_metadata",
+			columns: []TableColumnInfo{
+				{Name: "data", Type: "varchar"},
+			},
+			want: false,
+		},
+		{
+			name: "columnar_schema",
+			columns: []TableColumnInfo{
+				{Name: "id", Type: "bigint"},
+				{Name: "name", Type: "varchar"},
+				{Name: "amount", Type: "double"},
+			},
+			want: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := connector.hasRawModeColumns(tt.columns)
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}
+
 func TestQuoteTrinoIdentifier(t *testing.T) {
 	tests := []struct {
 		name string