Merge pull request #66 from fschoell/fix/ch_check_nonexistent_columns

added a check for non-existent columns in Clickhouse

CHANGELOG.md

@@ -7,6 +7,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+* Added a check for non-existent columns in Clickhouse
+
 * Added support for `Nullable` types in Clickhouse.
 
 ## v4.2.2

db/dialect_clickhouse.go

@@ -11,10 +11,10 @@ import (
 	"time"
 
 	_ "github.com/ClickHouse/clickhouse-go/v2"
-
 	"github.com/streamingfast/cli"
 	sink "github.com/streamingfast/substreams-sink"
 	"go.uber.org/zap"
+	"golang.org/x/exp/maps"
 )
 
 type clickhouseDialect struct{}
@@ -169,11 +169,15 @@ func convertOpToClickhouseValues(o *Operation) ([]any, error) {
 	sort.Strings(columns)
 	values := make([]any, len(o.data))
 	for i, v := range columns {
-		convertedType, err := convertToType(o.data[v], o.table.columnsByName[v].scanType)
-		if err != nil {
-			return nil, fmt.Errorf("converting value %q to type %q in column %q: %w", o.data[v], o.table.columnsByName[v].scanType, v, err)
+		if col, exists := o.table.columnsByName[v]; exists {
+			convertedType, err := convertToType(o.data[v], col.scanType)
+			if err != nil {
+				return nil, fmt.Errorf("converting value %q to type %q in column %q: %w", o.data[v], col.scanType, v, err)
+			}
+			values[i] = convertedType
+		} else {
+			return nil, fmt.Errorf("cannot find column %q for table %q (valid columns are %q)", v, o.table.identifier, strings.Join(maps.Keys(o.table.columnsByName), ", "))
 		}
-		values[i] = convertedType
 	}
 	return values, nil
 }