opensearch-project/OpenSearch

[BUG] Bulk upsert does not behave like a single Upsert, with an ingestion pipeline

Open

#10,864 opened on Oct 23, 2023

View on GitHub
 (4 comments) (2 reactions) (0 assignees)Java (1,505 forks)batch import
Indexingbuggood first issue

Repository metrics

Stars
 (8,123 stars)
PR merge metrics
 (Avg merge 5d 9h) (266 merged PRs in 30d)

Description

Describe the bug A single Upsert works as expected with an ingestion pipeline. But the same operation in a Bulk upsert doesn't give the same result.

To Reproduce Steps to reproduce the behavior:

  1. Create an ingestion pipeline that aims to calculate a duration between 2 dates
PUT _ingest/pipeline/pipeline-duration
{
  "description": "This pipeline complete data with begin, end and duration",
  "processors": [
    {
      "script": {
        "description": "Old duration",
        "lang": "painless",
        "source": "ctx.old_duration = 'Old duration was : '+ctx.event_duration;"
      }
    },
    {
      "script": {
        "description": "Begin computation",
        "lang": "painless",
        "params": { "_source": "" }, 
        "source": """
          if (ctx.event_begin == null || ((long) ctx.event_min) < ((long) ctx.event_begin)) { 
            ctx.event_begin = (long)ctx.event_min; 
          } else { 
            ctx.event_min = ctx.event_begin 
          } 
        """
      }
    },
    {
      "script": {
        "description": "End computation",
        "lang": "painless",
        "params": { "_source": "" }, 
        "source": """
          if (ctx.event_end == null || ((long) ctx.event_max) > ((long) ctx.event_end)) { 
            ctx.event_end = (long)ctx.event_max; 
          } else { 
            ctx.event_max = ctx.event_end 
          } 
        """
      }
    },
    {
      "script": {
        "description": "Duration computation",
        "lang": "painless",
        "source": "ctx.event_duration = (long)(ctx['event_max']-ctx['event_min']);"
      }
    }
  ]
}

  1. Create an index with the ingest pipeline
PUT index-duration
{
  "aliases": {},
  "mappings": {
    "properties": {
      "event_min": {
        "type": "float"
      },
      "event_max": {
        "type": "float"
      },
      "event_begin": {
        "type": "date",
        "format": "epoch_millis"
      },
      "event_end": {
        "type": "date",
        "format": "epoch_millis"
      },
      "event_duration": {
        "type": "long"
      },
      "event_name": {
        "type": "keyword"
      },
      "old_duration": {
        "type": "keyword"
      }
    }
  },
  "settings": {
    "index": {
      "final_pipeline": "pipeline-duration"
    }
  }
}

  1. Execute the 2 following Update operations :
POST index-duration/_update/doc_duration
{
  "doc" : {
    "event_min": 1,
    "event_max": 2,
	"event_name": "occurrence_1"
  },
  "doc_as_upsert": true
} 

second time

POST index-duration/_update/doc_duration
{
  "doc" : {
    "event_min": 3,
    "event_max": 5,
    "event_name": "occurrence_2"
  },
  "doc_as_upsert": true
} 

then check the document GET index-duration/_doc/doc_duration it gives :

  "_source": {
    "event_end": 5,
    "old_duration": "Old duration was : 1",
    "event_duration": 4,
    "event_min": 1,
    "event_name": "occurrence_2",
    "event_max": 5,
    "event_begin": 1
  }

  1. Now execute the same Update operations but embedded in a bulk
POST _bulk
{ "update": { "_index": "index-duration", "_id": "doc_duration_issue" } }
{ "doc" : { "event_min": 1, "event_max": 2, "event_name": "occurrence_1"},"doc_as_upsert": true} 

second time :

POST _bulk
{ "update": { "_index": "index-duration", "_id": "doc_duration_issue" } }
{ "doc" : { "event_min": 3, "event_max": 5, "event_name": "occurrence_2"},"doc_as_upsert": true} 

Then check the document with GET index-duration/_doc/doc_duration_issue it gives :

  "_source": {
    "event_end": 5,
    "old_duration": "Old duration was : null",
    "event_duration": 2,
    "event_min": 3,
    "event_name": "occurrence_2",
    "event_max": 5,
    "event_begin": 3
  }

Expected behavior Bulk Upsert and "Single" Upsert should have the same behavior when there is an ingestion pipeline. We expect to get the same values than doc_duration :

  "_source": {
    "event_end": 5,
    "old_duration": "Old duration was : 1",
    "event_duration": 4,
    "event_min": 1,
    "event_name": "occurrence_2",
    "event_max": 5,
    "event_begin": 1
  }

Plugins odfe-node1 opensearch-alerting 2.11.0.0 odfe-node1 opensearch-anomaly-detection 2.11.0.0 odfe-node1 opensearch-asynchronous-search 2.11.0.0 odfe-node1 opensearch-cross-cluster-replication 2.11.0.0 odfe-node1 opensearch-custom-codecs 2.11.0.0 odfe-node1 opensearch-geospatial 2.11.0.0 odfe-node1 opensearch-index-management 2.11.0.0 odfe-node1 opensearch-job-scheduler 2.11.0.0 odfe-node1 opensearch-knn 2.11.0.0 odfe-node1 opensearch-ml 2.11.0.0 odfe-node1 opensearch-neural-search 2.11.0.0 odfe-node1 opensearch-notifications 2.11.0.0 odfe-node1 opensearch-notifications-core 2.11.0.0 odfe-node1 opensearch-observability 2.11.0.0 odfe-node1 opensearch-performance-analyzer 2.11.0.0 odfe-node1 opensearch-reports-scheduler 2.11.0.0 odfe-node1 opensearch-security 2.11.0.0 odfe-node1 opensearch-security-analytics 2.11.0.0 odfe-node1 opensearch-sql 2.11.0.0

Host/Environment (please complete the following information):

  • docker images : opensearchproject/opensearch:2.11.0

Additional context This issue is not exactly the same than the one posted here : #2607 That's why I prefer open a new one.

Contributor guide