dual query returns from CH

Author: mieciu

ci/it/configs/quesma-with-dual-writes-and-common-table.yml.template

@@ -30,6 +30,8 @@ processors:
           target: [ c ]
         logs-3:
           target: [ c, e ]
+        logs-dual-query:
+          target: [ c, e ]
         logs-4:
           useCommonTable: true
           target: [ c ]
@@ -48,6 +50,8 @@ processors:
           target: [ c ]
         logs-3:
           target: [ c, e ]
+        logs-dual-query:
+          target: [ c, e ]
         logs-4:
           useCommonTable: true
           target: [ c ]

ci/it/testcases/test_dual_write_and_common_table.go

@@ -34,7 +34,7 @@ func (a *DualWriteAndCommonTableTestcase) RunTests(ctx context.Context, t *testi
 	a.testBasicRequest(ctx, t)
 	a.testIngestToClickHouseWorks(ctx, t)
 	//a.testIngestToCommonTableWorks(ctx, t)
-	//a.testDualQueryReturnsDataFromClickHouse(ctx, t)
+	a.testDualQueryReturnsDataFromClickHouse(ctx, t)
 	a.testDualWritesWork(ctx, t)
 	a.testWildcardGoesToElastic(ctx, t)
 	return nil
@@ -49,6 +49,71 @@ func (a *DualWriteAndCommonTableTestcase) testBasicRequest(ctx context.Context,
 	assert.Equal(t, http.StatusOK, resp.StatusCode)
 }
 
+func (a *DualWriteAndCommonTableTestcase) testDualQueryReturnsDataFromClickHouse(ctx context.Context, t *testing.T) {
+	resp, err := a.RequestToQuesma(ctx, "POST", "/logs-dual-query/_doc", []byte(`{"name": "Przemyslaw", "age": 31337}`))
+	if err != nil {
+		t.Fatalf("Failed to insert document: %s", err)
+	}
+	defer resp.Body.Close()
+	assert.Equal(t, http.StatusOK, resp.StatusCode)
+
+	chQuery := "SELECT * FROM 'logs-dual-query'"
+	rows, err := a.ExecuteClickHouseQuery(ctx, chQuery)
+	if err != nil {
+		t.Fatalf("Failed to execute query: %s", err)
+	}
+	columnTypes, err := rows.ColumnTypes()
+	values := make([]interface{}, len(columnTypes))
+	valuePtrs := make([]interface{}, len(columnTypes))
+	for i := range values {
+		valuePtrs[i] = &values[i]
+	}
+	var name string
+	var age int
+	for rows.Next() {
+		if err := rows.Scan(valuePtrs...); err != nil {
+			t.Fatalf("Failed to scan row: %s", err)
+		}
+		for i, col := range values {
+			switch columnTypes[i].Name() {
+			case "name":
+				if v, ok := col.(*string); ok {
+					name = *v
+				}
+			case "age":
+				if v, ok := col.(*int64); ok {
+					age = int(*v)
+				}
+			}
+		}
+		if name == "Przemyslaw" && age == 31337 {
+			break
+		}
+	}
+	assert.Equal(t, "Przemyslaw", name)
+	assert.Equal(t, 31337, age)
+
+	// In the meantime let's delete the index from Elasticsearch
+	_, _ = a.RequestToElasticsearch(ctx, "DELETE", "/logs-dual-query", nil)
+	if err != nil {
+		t.Fatalf("Failed to make DELETE request: %s", err)
+	}
+	// FINAL TEST - WHETHER QUESMA RETURNS DATA FROM CLICKHOUSE
+	resp, err = a.RequestToQuesma(ctx, "GET", "/logs-dual-query/_search", []byte(`{"query": {"match_all": {}}}`))
+	if err != nil {
+		t.Fatalf("Failed to make GET request: %s", err)
+	}
+	defer resp.Body.Close()
+	bodyBytes, err := io.ReadAll(resp.Body)
+	if err != nil {
+		t.Fatalf("Failed to read response body: %s", err)
+	}
+
+	assert.Equal(t, http.StatusOK, resp.StatusCode)
+	assert.Contains(t, string(bodyBytes), "Przemyslaw")
+	assert.Contains(t, "Clickhouse", resp.Header.Get("X-Quesma-Source"))
+}
+
 func (a *DualWriteAndCommonTableTestcase) testIngestToClickHouseWorks(ctx context.Context, t *testing.T) {
 	resp, err := a.RequestToQuesma(ctx, "POST", "/logs-2/_doc", []byte(`{"name": "Przemyslaw", "age": 31337}`))
 	if err != nil {