opensearch-project/OpenSearch

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

Open

#10.864 aberto em 23 de out. de 2023

Ver no GitHub
 (4 comments) (2 reactions) (0 assignees)Java (1.505 forks)batch import
Indexingbuggood first issue

Métricas do repositório

Stars
 (8.123 stars)
Métricas de merge de PR
 (Mesclagem média 5d 9h) (266 fundiu PRs em 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.

Guia do colaborador