Array handling

Author: pdelewski

platform/backend_connectors/hydrolix_backend_connector.go

@@ -115,7 +115,7 @@ func (p *HydrolixBackendConnector) makeRequest(ctx context.Context, method strin
 }
 
 // TODO hardcoded for now
-const token = "eyJhbGciOiJSUzI1NiIsInR5cCIgOiAiSldUIiwia2lkIiA6ICIybDZyTk1YV2hYQTA5M2tkRHA5ZFctaEMzM2NkOEtWUFhJdURZLWlLeUFjIn0.eyJleHAiOjE3NTMxNzUyNDAsImlhdCI6MTc1MzA4ODg0MCwianRpIjoiNWZkYmIyMzktMDk5ZS00M2VmLWFhMzctY2JiMDUxZDRkNDAyIiwiaXNzIjoiaHR0cHM6Ly9sb2NhbGhvc3Qva2V5Y2xvYWsvcmVhbG1zL2h5ZHJvbGl4LXVzZXJzIiwiYXVkIjpbImNvbmZpZy1hcGkiLCJhY2NvdW50Il0sInN1YiI6ImRiMWM1YTJiLTdhYjMtNGNmZi04NGU4LTQ3Yzc0YjRlZjAyMSIsInR5cCI6IkJlYXJlciIsImF6cCI6ImNvbmZpZy1hcGkiLCJzZXNzaW9uX3N0YXRlIjoiNDdmZDE2YjAtZjhkYi00NWI4LWIwNzUtOGYyNTRhNGUxNTFkIiwiYWNyIjoiMSIsImFsbG93ZWQtb3JpZ2lucyI6WyJodHRwOi8vbG9jYWxob3N0Il0sInJlYWxtX2FjY2VzcyI6eyJyb2xlcyI6WyJkZWZhdWx0LXJvbGVzLWh5ZHJvbGl4LXVzZXJzIiwib2ZmbGluZV9hY2Nlc3MiLCJ1bWFfYXV0aG9yaXphdGlvbiJdfSwicmVzb3VyY2VfYWNjZXNzIjp7ImFjY291bnQiOnsicm9sZXMiOlsibWFuYWdlLWFjY291bnQiLCJtYW5hZ2UtYWNjb3VudC1saW5rcyIsInZpZXctcHJvZmlsZSJdfX0sInNjb3BlIjoib3BlbmlkIGNvbmZpZy1hcGktc2VydmljZSBlbWFpbCBwcm9maWxlIiwic2lkIjoiNDdmZDE2YjAtZjhkYi00NWI4LWIwNzUtOGYyNTRhNGUxNTFkIiwiZW1haWxfdmVyaWZpZWQiOnRydWUsInByZWZlcnJlZF91c2VybmFtZSI6Im1lQGh5ZHJvbGl4LmlvIiwiZW1haWwiOiJtZUBoeWRyb2xpeC5pbyJ9.A-61oZS599AQd5HrECfNjVPXdKF2iVZmRGvmZuPiKh3JctL6F00V5InptaH4xZCHqOrda2gLTz5UL4hSx5Fdvk9yrpba5UFdmGhFpkW1jDaHDvOKK_UMGip3wGkAwp0EwEvfqcGhL0Iw96bTUOOlx6LAsSoKvOuLsL5ljBkDIi8vJadVfc01IuQu1AlwlDHpNn1EJbEFQi8cocDTYuG2U8FilFvhHLA77CFysimHxzq0n48vPpxCwtNpzQ9pbbX3ZPsxPoHFojXxvTf0yNWnUR76W3Zw2HlJrlRbiiRqQ4afkTWoxzIb6VAEImemKKtSLS4Ym3X9wuFp_fVLSokj-Q"
+const token = "eyJhbGciOiJSUzI1NiIsInR5cCIgOiAiSldUIiwia2lkIiA6ICIybDZyTk1YV2hYQTA5M2tkRHA5ZFctaEMzM2NkOEtWUFhJdURZLWlLeUFjIn0.eyJleHAiOjE3NTMyNjE3NjksImlhdCI6MTc1MzE3NTM2OSwianRpIjoiMzY2MTY2OGMtOGVjMi00YzkxLTg4ZTItOWQ1YmU3YTdiNGNkIiwiaXNzIjoiaHR0cHM6Ly9sb2NhbGhvc3Qva2V5Y2xvYWsvcmVhbG1zL2h5ZHJvbGl4LXVzZXJzIiwiYXVkIjpbImNvbmZpZy1hcGkiLCJhY2NvdW50Il0sInN1YiI6ImRiMWM1YTJiLTdhYjMtNGNmZi04NGU4LTQ3Yzc0YjRlZjAyMSIsInR5cCI6IkJlYXJlciIsImF6cCI6ImNvbmZpZy1hcGkiLCJzZXNzaW9uX3N0YXRlIjoiYzFjZmE2ZjktOWRhMS00ZDEzLThhYTItZmE3MjMzNWEzZjQxIiwiYWNyIjoiMSIsImFsbG93ZWQtb3JpZ2lucyI6WyJodHRwOi8vbG9jYWxob3N0Il0sInJlYWxtX2FjY2VzcyI6eyJyb2xlcyI6WyJkZWZhdWx0LXJvbGVzLWh5ZHJvbGl4LXVzZXJzIiwib2ZmbGluZV9hY2Nlc3MiLCJ1bWFfYXV0aG9yaXphdGlvbiJdfSwicmVzb3VyY2VfYWNjZXNzIjp7ImFjY291bnQiOnsicm9sZXMiOlsibWFuYWdlLWFjY291bnQiLCJtYW5hZ2UtYWNjb3VudC1saW5rcyIsInZpZXctcHJvZmlsZSJdfX0sInNjb3BlIjoib3BlbmlkIGNvbmZpZy1hcGktc2VydmljZSBlbWFpbCBwcm9maWxlIiwic2lkIjoiYzFjZmE2ZjktOWRhMS00ZDEzLThhYTItZmE3MjMzNWEzZjQxIiwiZW1haWxfdmVyaWZpZWQiOnRydWUsInByZWZlcnJlZF91c2VybmFtZSI6Im1lQGh5ZHJvbGl4LmlvIiwiZW1haWwiOiJtZUBoeWRyb2xpeC5pbyJ9.bSfYTJjJGE3S3IGG32qLwgBZrSZSAIImd4hyoN2KqjPARPhwYCqiW3luSlPc4YL-OHH0YW3YYPM1RyxYQn4CW0DvK9pwYE3qLu7Zt6uVd8Kb88KIHxrXN6mQrIP9-cuYUwyVNVbbGfEYi1aYWSWDJsX9IsmRyIh-aLVHzNXKF7f8I_ek6BdvVDFanoqXVytosWruH7avqJ2kswP99PHHOPJD_SpBhvO014p0-trLM5rWTR3YG6823EEpCSXZ8T4nPZXynXcC8cxTvg_1qVusG4_uBTnqw4LL2HlnYkYM1NPQShkcJvhN3-aygXdsiGgqlsljIpa6cuFkTpANeDbriw"
 const hdxHost = "3.20.203.177:8888"
 const orgID = "d9ce0431-f26f-44e3-b0ef-abc1653d04eb"
 const projectID = "27506b30-0c78-41fa-a059-048d687f1164"
@@ -149,6 +149,7 @@ func (p *HydrolixBackendConnector) ingestFun(ctx context.Context, ingest []map[s
 	emitRequest:
 		respJson, err := p.makeRequest(ctx, "POST", url, ingestJson, token, tableName)
 		if err != nil {
+			logger.DebugWithCtx(ctx).Msgf("Error ingesting table %s: %v", tableName, err)
 			var resp HydrolixResponse
 			if err := json.Unmarshal(respJson, &resp); err != nil {
 				if strings.Contains(resp.Message, "no table") {

platform/ingest/hydrolixlowerer.go

@@ -251,7 +251,16 @@ func CastToType(value any, typeName string) (any, error) {
 		case string:
 			return strconv.ParseBool(v)
 		}
-
+	case "int64":
+		if v, ok := value.(int64); ok {
+			return v, nil
+		}
+		switch v := value.(type) {
+		case float64:
+			return int64(v), nil
+		case string:
+			return strconv.Atoi(v)
+		}
 	default:
 		return nil, fmt.Errorf("unsupported target type: %s", typeName)
 	}
@@ -417,8 +426,17 @@ func (l *HydrolixLowerer) LowerToDDL(
 
 			case ArrayType:
 				elemType := typeInfo.Elements[0].Name
-				value = []interface{}{defaultForType(elemType)} // array with one sample element
-
+				value = []any{}
+				if events[colName] != nil {
+					for _, elem := range events[colName].([]any) {
+						castedElem, err := CastToType(elem, elemType)
+						if err != nil {
+							logger.ErrorWithCtx(context.Background()).Msgf("Error casting element %v to type %s: %v", elem, elemType, err)
+							continue
+						}
+						value = append(value.([]interface{}), castedElem)
+					}
+				}
 			case MapType:
 				keyType := typeInfo.Elements[0].Name
 				valType := typeInfo.Elements[1].Name