Review remarks #2

Author: pdelewski

platform/backend_connectors/hydrolix_backend_connector.go

@@ -136,6 +136,8 @@ func (p *HydrolixBackendConnector) ingestFun(ctx context.Context, ingestSlice []
 	}
 
 	url := fmt.Sprintf("%s/ingest/event", p.cfg.Url.String())
+	// Sleep duration is arbitrarily chosen.
+	// It seems that the Hydrolix API needs some time to process the table creation before ingesting data.
 	const sleepDuration = 5 * time.Second
 	const maxRetries = 5
 	for retries := 0; retries < maxRetries; retries++ {

platform/ingest/hydrolixlowerer.go

@@ -26,6 +26,7 @@ type HydrolixLowerer struct {
 	ingestFieldStatistics      IngestFieldStatistics
 	ingestFieldStatisticsLock  sync.Mutex
 	tableCreteStatementMapping map[*chLib.Table]CreateTableStatement // cache for table creation statements
+	tableCreationLock          sync.Mutex
 }
 
 func NewHydrolixLowerer(virtualTableStorage persistence.JSONDatabase) *HydrolixLowerer {
@@ -451,11 +452,13 @@ func (l *HydrolixLowerer) LowerToDDL(
 	createTableCmd CreateTableStatement,
 ) ([]string, error) {
 
+	l.tableCreationLock.Lock()
 	if _, exists := l.tableCreteStatementMapping[table]; !exists {
 		l.tableCreteStatementMapping[table] = createTableCmd
 	} else {
 		createTableCmd = l.tableCreteStatementMapping[table]
 	}
+	l.tableCreationLock.Unlock()
 
 	// --- Create Table Section ---
 	createTable := map[string]interface{}{