Skip to content

Commit e3d84a4

Browse files
realdatadrivenrealdatadriven
committed
Refactor GetColumnType method to remove table parameter and enhance data type conversion for DuckDB dialect
1 parent cc51e62 commit e3d84a4

2 files changed

Lines changed: 84 additions & 13 deletions

File tree

‎internal/db/duckdb.go‎

Lines changed: 49 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import (
55
"database/sql"
66
"fmt"
77
"os"
8+
"regexp"
89
"time"
910

1011
"github.com/jmoiron/sqlx"
@@ -321,12 +322,58 @@ func (db *DuckDB) QuerySingleRow(query string, params ...any) (*map[string]any,
321322
return &result, true, err
322323
}
323324

325+
func NamedToPositional(sql string, data map[string]any) (string, []any, error) {
326+
// Find all :param occurrences
327+
re := regexp.MustCompile(`:([a-zA-Z_][a-zA-Z0-9_]*)`)
328+
matches := re.FindAllStringSubmatch(sql, -1)
329+
if len(matches) == 0 {
330+
return sql, nil, nil
331+
}
332+
// Collect unique parameter names in order of appearance
333+
seen := make(map[string]bool)
334+
var paramOrder []string
335+
for _, match := range matches {
336+
name := match[1]
337+
if !seen[name] {
338+
seen[name] = true
339+
paramOrder = append(paramOrder, name)
340+
}
341+
}
342+
// Check for missing values
343+
var args []any
344+
for _, name := range paramOrder {
345+
val, ok := data[name]
346+
if !ok {
347+
return "", nil, fmt.Errorf("missing value for parameter: %s", name)
348+
}
349+
args = append(args, val)
350+
}
351+
// Replace :name → ?
352+
result := re.ReplaceAllStringFunc(sql, func(s string) string {
353+
// We could also do a map lookup here, but simpler to just replace with ?
354+
return "?"
355+
})
356+
return result, args, nil
357+
}
358+
324359
func (db *DuckDB) ExecuteNamedQuery(query string, data map[string]any) (int, error) {
325-
return 0, fmt.Errorf("not implemented yet %s", "_")
360+
ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
361+
defer cancel()
362+
_positionalQuery, args, err := NamedToPositional(query, data)
363+
if err != nil {
364+
return 0, fmt.Errorf("failed to convert named query to positional: %w", err)
365+
}
366+
// fmt.Println(_positionalQuery, args)
367+
result, err := db.ExecContext(ctx, _positionalQuery, args...)
368+
if err != nil {
369+
return 0, err
370+
}
371+
id, err := result.LastInsertId()
372+
return int(id), nil
326373
}
327374

328375
func (db *DuckDB) ExecuteQueryPGInsertWithLastInsertId(query string, data ...any) (int, error) {
329-
return 0, fmt.Errorf("not implemented yet %s", "_")
376+
return db.ExecuteNamedQuery(query, data[0].(map[string]any))
330377
}
331378

332379
func (db *DuckDB) FromParams(params map[string]any, extra_conf map[string]any) (*DB, string, string, error) {

‎internal/etlx/model.go‎

Lines changed: 35 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@ type SQLDialect interface {
2121
GetCreateTableIfNotExists(tableName string) (string, string)
2222
GetTableName(tableName string) string
2323
GetColumnName(fieldName string) string
24-
GetColumnType(field map[string]any, table string, dbCon db.DBInterface) string
24+
GetColumnType(field map[string]any, dbCon db.DBInterface) string
2525
GetPrimaryKey(field map[string]any) string
2626
GetAutoIncrement(field map[string]any) string
2727
GetNullable(field map[string]any) string
@@ -63,7 +63,7 @@ func (b *BaseDialect) GetTableName(tableName string) string {
6363
return tableName
6464
}
6565

66-
func (b *BaseDialect) GetColumnType(field map[string]any, table string, dbCon db.DBInterface) string {
66+
func (b *BaseDialect) GetColumnType(field map[string]any, dbCon db.DBInterface) string {
6767
sqlType := field["type"].(string)
6868
switch strings.ToUpper(sqlType) {
6969
case "INTEGER":
@@ -161,7 +161,7 @@ func (p *PostgresDialect) GetColumnName(fieldName string) string {
161161
return fmt.Sprintf(`"%s"`, fieldName)
162162
}
163163

164-
func (p *PostgresDialect) GetColumnType(field map[string]any, table string, dbCon db.DBInterface) string {
164+
func (p *PostgresDialect) GetColumnType(field map[string]any, dbCon db.DBInterface) string {
165165
sqlType := field["type"].(string)
166166
switch strings.ToUpper(sqlType) {
167167
case "INTEGER":
@@ -247,19 +247,19 @@ func (d *DuckDBDialect) GetColumnName(fieldName string) string {
247247
return fmt.Sprintf(`"%s"`, fieldName)
248248
}
249249

250-
func (d *DuckDBDialect) GetColumnType(field map[string]any, table string, dbCon db.DBInterface) string {
250+
func (d *DuckDBDialect) GetColumnType(field map[string]any, dbCon db.DBInterface) string {
251251
sqlType := field["type"].(string)
252252
switch strings.ToUpper(sqlType) {
253253
case "INTEGER":
254254
if autoincrement, ok := field["autoincrement"].(bool); ok && autoincrement {
255255
// GetColumnType(field map[string]any)
256256
if dbCon != nil {
257-
create_seq := fmt.Sprintf(`CREATE SEQUENCE IF NOT EXISTS "%s_%s_seq" START WITH 1 INCREMENT BY 1;`, table, field["name"].(string))
257+
create_seq := fmt.Sprintf(`CREATE SEQUENCE IF NOT EXISTS "%s_%s_seq" START WITH 1 INCREMENT BY 1;`, field["table"], field["name"])
258258
_, err := dbCon.ExecuteQuery(create_seq, []any{}...)
259259
if err != nil {
260260
fmt.Println("Err creating SEQUENCE:", err)
261261
}
262-
return fmt.Sprintf(`INTEGER PRIMARY KEY DEFAULT NEXTVAL('%s_%s_seq')`, table, field["name"].(string))
262+
return fmt.Sprintf(`INTEGER DEFAULT NEXTVAL('%s_%s_seq')`, field["table"], field["name"])
263263
}
264264
}
265265
return "INTEGER"
@@ -296,6 +296,28 @@ func (d *DuckDBDialect) GetTableComment(tableName, comment string) string {
296296

297297
func (d *DuckDBDialect) SupportsTableComment() bool { return true }
298298

299+
func (d *DuckDBDialect) DataTypeConversion(row map[string]any) map[string]any {
300+
converted := make(map[string]any)
301+
for k, v := range row {
302+
switch val := v.(type) {
303+
case bool:
304+
/*if v.(bool) {
305+
converted[k] = 1
306+
} else {
307+
converted[k] = 0
308+
}*/
309+
converted[k] = val
310+
case time.Time:
311+
// SELECT TIMESTAMPTZ '1992-09-20 11:30:00.123456-05:00'
312+
converted[k] = val.Format("2006-01-02 15:04:05.999999-07:00")
313+
// fmt.Printf("Converting time.Time value for DuckDB: %v -> %s\n", val, converted[k])
314+
default:
315+
converted[k] = val
316+
}
317+
}
318+
return converted
319+
}
320+
299321
// MySQLDialect implements SQLDialect for MySQL.
300322
type MySQLDialect struct{ BaseDialect }
301323

@@ -320,7 +342,7 @@ func (m *MySQLDialect) GetTableName(tableName string) string {
320342
return fmt.Sprintf("`%s`", tableName)
321343
}
322344

323-
func (m *MySQLDialect) GetColumnType(field map[string]any, table string, dbCon db.DBInterface) string {
345+
func (m *MySQLDialect) GetColumnType(field map[string]any, dbCon db.DBInterface) string {
324346
sqlType := field["type"].(string)
325347
switch strings.ToUpper(sqlType) {
326348
case "INTEGER":
@@ -403,7 +425,7 @@ func (s *SQLiteDialect) GetColumnName(fieldName string) string {
403425
return fmt.Sprintf(`"%s"`, fieldName)
404426
}
405427

406-
func (s *SQLiteDialect) GetColumnType(field map[string]any, table string, dbCon db.DBInterface) string {
428+
func (s *SQLiteDialect) GetColumnType(field map[string]any, dbCon db.DBInterface) string {
407429
sqlType := field["type"].(string)
408430
switch strings.ToUpper(sqlType) {
409431
case "INTEGER":
@@ -477,7 +499,7 @@ func (ms *MSSQLDialect) GetColumnName(fieldName string) string {
477499
return fmt.Sprintf(`[%s]`, fieldName)
478500
}
479501

480-
func (ms *MSSQLDialect) GetColumnType(field map[string]any, table string, dbCon db.DBInterface) string {
502+
func (ms *MSSQLDialect) GetColumnType(field map[string]any, dbCon db.DBInterface) string {
481503
sqlType := field["type"].(string)
482504
switch strings.ToUpper(sqlType) {
483505
case "INTEGER":
@@ -619,7 +641,9 @@ func generateCreateTableSQL(driver, tableName, tableComment, createAll string, f
619641
continue
620642
}
621643
name := dialect.GetColumnName(fieldName)
622-
columnType := dialect.GetColumnType(field, tableName, dbCon)
644+
field["name"] = fieldName // Add the original field name to the field definition for use in GetColumnType if needed
645+
field["table"] = tableName // Add the table name to the field definition for use in GetColumnType if needed
646+
columnType := dialect.GetColumnType(field, dbCon)
623647
primaryKey := dialect.GetPrimaryKey(field)
624648
autoincrement := dialect.GetAutoIncrement(field)
625649
nullable := dialect.GetNullable(field)
@@ -764,7 +788,6 @@ func (etlx *ETLX) InsertData(dbCon db.DBInterface, tableName string, columns map
764788
if _, ok := row["excluded"]; !ok {
765789
row["excluded"] = false
766790
}
767-
row = dialect.DataTypeConversion(row) // Convert data types as needed for the specific SQL dialect
768791
insertCols := []string{}
769792
insertVals := []string{} // For named parameters, e.g., ":colName"
770793
insertMap := make(map[string]any)
@@ -801,6 +824,7 @@ func (etlx *ETLX) InsertData(dbCon db.DBInterface, tableName string, columns map
801824
// If it's nullable and not provided, it will be omitted from the INSERT, allowing DB default/NULL
802825
}
803826
}
827+
insertMap = dialect.DataTypeConversion(insertMap) // Convert data types as needed for the specific SQL dialect
804828
// Construct the INSERT statement
805829
if len(insertCols) == 0 {
806830
return fmt.Errorf("row %d: no columns to insert", i)

0 commit comments

Comments
 (0)