Ingest processor can be nil (#1457)

Fixes panic introduced here (#1448).

Author: nablaone

ci/it/testcases/test_reading_clickhouse_tables.go

@@ -93,12 +93,29 @@ func (a *ReadingClickHouseTablesIntegrationTestcase) testWildcardGoesToElastic(c
 func (a *ReadingClickHouseTablesIntegrationTestcase) testIngestIsDisabled(ctx context.Context, t *testing.T) {
 	// There is no ingest pipeline, so Quesma should reject all ingest requests
 	for _, tt := range []string{"test_table", "extra_index"} {
-		t.Run(tt, func(t *testing.T) {
+		t.Run(tt+"_doc", func(t *testing.T) {
 			resp, bodyBytes := a.RequestToQuesma(ctx, t, "POST", fmt.Sprintf("/%s/_doc", tt), []byte(`{"name": "Piotr", "age": 11111}`))
 			assert.Contains(t, string(bodyBytes), "index_closed_exception")
 			assert.Equal(t, http.StatusOK, resp.StatusCode)
 			assert.Equal(t, "Clickhouse", resp.Header.Get("X-Quesma-Source"))
 			assert.Equal(t, "Elasticsearch", resp.Header.Get("X-Elastic-Product"))
 		})
 	}
+
+	for _, tt := range []string{"test_table", "extra_index"} {
+		t.Run(tt+"_bulk", func(t *testing.T) {
+
+			bulkPayload := []byte(fmt.Sprintf(`
+				{ "index": { "_index": "%s", "_id": "1" } }
+				{ "name": "Alice", "age": 30 }
+`, tt))
+
+			resp, bodyBytes := a.RequestToQuesma(ctx, t, "POST", "/_bulk", bulkPayload)
+			assert.Contains(t, string(bodyBytes), "index_closed_exception")
+			assert.Equal(t, http.StatusOK, resp.StatusCode)
+			assert.Equal(t, "Clickhouse", resp.Header.Get("X-Quesma-Source"))
+			assert.Equal(t, "Elasticsearch", resp.Header.Get("X-Elastic-Product"))
+		})
+	}
+
 }

platform/functionality/bulk/bulk.go

@@ -75,7 +75,13 @@ func Write(ctx context.Context, defaultIndex *string, bulk types.NDJSON, ip *ing
 
 	// The returned results should be in the same order as the input request, however splitting the bulk might change the order.
 	// Therefore, each BulkRequestEntry has a corresponding pointer to the result entry, allowing us to freely split and reshuffle the bulk.
-	results, clickhouseBulkEntries, elasticRequestBody, elasticBulkEntries, err := SplitBulk(ctx, defaultIndex, bulk, maxBulkSize, tableResolver, ip.GetIndexNameRewriter())
+
+	var indexNameRewriter ingest.IndexNameRewriter
+	if ip != nil {
+		indexNameRewriter = ip.GetIndexNameRewriter()
+	}
+
+	results, clickhouseBulkEntries, elasticRequestBody, elasticBulkEntries, err := SplitBulk(ctx, defaultIndex, bulk, maxBulkSize, tableResolver, indexNameRewriter)
 	if err != nil {
 		return []BulkItem{}, err
 	}
@@ -129,7 +135,9 @@ func SplitBulk(ctx context.Context, defaultIndex *string, bulk types.NDJSON, max
 		index := op.GetIndex()
 		operation := op.GetOperation()
 
-		index = rewriter.RewriteIndex(index)
+		if rewriter != nil {
+			index = rewriter.RewriteIndex(index)
+		}
 
 		entryWithResponse := BulkRequestEntry{
 			operation: operation,