fix: quote tdengine history inserts
This commit is contained in:
@@ -63,7 +63,7 @@ func (w *Writer) AppendRawFrame(ctx context.Context, env envelope.FrameEnvelope)
|
|||||||
_, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
|
_, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
|
||||||
(ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes,
|
(ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes,
|
||||||
raw_hex, raw_text, parsed_json, fields_json, parse_status, parse_error, source_endpoint)
|
raw_hex, raw_text, parsed_json, fields_json, parse_status, parse_error, source_endpoint)
|
||||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, table), rawValues(env)...)
|
VALUES (%s)`, table, joinLiterals(rawValues(env))))
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -80,7 +80,7 @@ func (w *Writer) AppendLocation(ctx context.Context, env envelope.FrameEnvelope)
|
|||||||
_, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
|
_, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
|
||||||
(ts, event_id, frame_id, received_at, longitude, latitude, altitude_m, speed_kmh,
|
(ts, event_id, frame_id, received_at, longitude, latitude, altitude_m, speed_kmh,
|
||||||
direction_deg, alarm_flag, status_flag, total_mileage_km)
|
direction_deg, alarm_flag, status_flag, total_mileage_km)
|
||||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, table), locationValues(env, longitude, latitude)...)
|
VALUES (%s)`, table, joinLiterals(locationValues(env, longitude, latitude))))
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -95,7 +95,7 @@ func (w *Writer) AppendMileagePoint(ctx context.Context, env envelope.FrameEnvel
|
|||||||
}
|
}
|
||||||
_, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
|
_, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
|
||||||
(ts, event_id, frame_id, received_at, total_mileage_km, speed_kmh, longitude, latitude)
|
(ts, event_id, frame_id, received_at, total_mileage_km, speed_kmh, longitude, latitude)
|
||||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, table), mileageValues(env, totalMileage)...)
|
VALUES (%s)`, table, joinLiterals(mileageValues(env, totalMileage))))
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -298,3 +298,39 @@ func millis(value int64) time.Time {
|
|||||||
func quote(value string) string {
|
func quote(value string) string {
|
||||||
return strings.ReplaceAll(value, "'", "''")
|
return strings.ReplaceAll(value, "'", "''")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func joinLiterals(values []any) string {
|
||||||
|
out := make([]string, 0, len(values))
|
||||||
|
for _, value := range values {
|
||||||
|
out = append(out, literal(value))
|
||||||
|
}
|
||||||
|
return strings.Join(out, ", ")
|
||||||
|
}
|
||||||
|
|
||||||
|
func literal(value any) string {
|
||||||
|
if value == nil {
|
||||||
|
return "NULL"
|
||||||
|
}
|
||||||
|
switch typed := value.(type) {
|
||||||
|
case time.Time:
|
||||||
|
return "'" + typed.UTC().Format("2006-01-02 15:04:05.000") + "'"
|
||||||
|
case string:
|
||||||
|
return "'" + quote(typed) + "'"
|
||||||
|
case envelope.ParseStatus:
|
||||||
|
return "'" + quote(string(typed)) + "'"
|
||||||
|
case int:
|
||||||
|
return strconv.FormatInt(int64(typed), 10)
|
||||||
|
case int64:
|
||||||
|
return strconv.FormatInt(typed, 10)
|
||||||
|
case uint16:
|
||||||
|
return strconv.FormatUint(uint64(typed), 10)
|
||||||
|
case uint32:
|
||||||
|
return strconv.FormatUint(uint64(typed), 10)
|
||||||
|
case float64:
|
||||||
|
return strconv.FormatFloat(typed, 'f', -1, 64)
|
||||||
|
case float32:
|
||||||
|
return strconv.FormatFloat(float64(typed), 'f', -1, 64)
|
||||||
|
default:
|
||||||
|
return "'" + quote(fmt.Sprint(typed)) + "'"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -45,6 +45,13 @@ func TestWriterAppendsRawLocationAndMileage(t *testing.T) {
|
|||||||
if got := countSQL(exec.calls, "INSERT INTO mil_"); got != 1 {
|
if got := countSQL(exec.calls, "INSERT INTO mil_"); got != 1 {
|
||||||
t.Fatalf("mileage insert count = %d", got)
|
t.Fatalf("mileage insert count = %d", got)
|
||||||
}
|
}
|
||||||
|
rawInsert := findSQL(exec.calls, "INSERT INTO raw_")
|
||||||
|
if !strings.Contains(rawInsert, "'go_") || !strings.Contains(rawInsert, "'2e") && !strings.Contains(rawInsert, "'") {
|
||||||
|
t.Fatalf("raw insert should quote string literals: %s", rawInsert)
|
||||||
|
}
|
||||||
|
if strings.Contains(rawInsert, "VALUES (?,") {
|
||||||
|
t.Fatalf("raw insert should not use placeholders for tdengine websocket plain insert: %s", rawInsert)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestWriterSkipsSparseDerivedRows(t *testing.T) {
|
func TestWriterSkipsSparseDerivedRows(t *testing.T) {
|
||||||
@@ -103,6 +110,15 @@ func countSQL(calls []execCall, pattern string) int {
|
|||||||
return count
|
return count
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func findSQL(calls []execCall, pattern string) string {
|
||||||
|
for _, call := range calls {
|
||||||
|
if strings.Contains(call.query, pattern) {
|
||||||
|
return call.query
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
|
||||||
type execCall struct {
|
type execCall struct {
|
||||||
query string
|
query string
|
||||||
args []any
|
args []any
|
||||||
|
|||||||
Reference in New Issue
Block a user