Update By Query API

Specifying the refresh parameter refreshes all shards once the request completes. This is different than the update API’s refresh parameter, which causes just the shard that received the request to be refreshed. Unlike the update API, it does not support wait_for.

If the request contains wait_for_completion=false, Elasticsearch performs some preflight checks, launches the request, and returns a task you can use to cancel or get the status of the task. Elasticsearch creates a record of this task as a document at .tasks/task/${taskId}.

wait_for_active_shards controls how many copies of a shard must be active before proceeding with the request. See Active shards for details. timeout controls how long each write request waits for unavailable shards to become available. Both work exactly the way they work in the Bulk API. Update by query uses scrolled searches, so you can also specify the scroll parameter to control how long it keeps the search context alive, for example ?scroll=10m. The default is 5 minutes.

To control the rate at which update by query issues batches of update operations, you can set requests_per_second to any positive decimal number. This pads each batch with a wait time to throttle the rate. Set requests_per_second to -1 to disable throttling.

Throttling uses a wait time between batches so that the internal scroll requests can be given a timeout that takes the request padding into account. The padding time is the difference between the batch size divided by the requests_per_second and the time spent writing. By default the batch size is 1000, so if requests_per_second is set to 500:

target_time = 1000 / 500 per second = 2 seconds
wait_time = target_time - write_time = 2 seconds - .5 seconds = 1.5 seconds

Since the batch is issued as a single _bulk request, large batch sizes cause Elasticsearch to create many requests and wait before starting the next set. This is "bursty" instead of "smooth".

Update by query supports sliced scroll to parallelize the update process. This can improve efficiency and provide a convenient way to break the request down into smaller parts.

Setting slices to auto chooses a reasonable number for most data streams and indices. If you’re slicing manually or otherwise tuning automatic slicing, keep in mind that:

Whether query or update performance dominates the runtime depends on the documents being reindexed and cluster resources.

Update by query supports scripts to update the document source. For example, the following request increments the count field for all documents with a user.id of kimchy in my-index-000001:

POST my-index-000001/_update_by_query
{
  "script": {
    "source": "ctx._source.count++",
    "lang": "painless"
  },
  "query": {
    "term": {
      "user.id": "kimchy"
    }
  }
}

Note that conflicts=proceed is not specified in this example. In this case, a version conflict should halt the process so you can handle the failure.

As with the Update API, you can set ctx.op to change the operation that is performed:

Update by query only supports update, noop, and delete. Setting ctx.op to anything else is an error. Setting any other field in ctx is an error. This API only enables you to modify the source of matching documents, you cannot move them.

Update by query can use the Ingest pipelines feature by specifying a pipeline:

PUT _ingest/pipeline/set-foo
{
  "description" : "sets foo",
  "processors" : [ {
      "set" : {
        "field": "foo",
        "value": "bar"
      }
  } ]
}
POST my-index-000001/_update_by_query?pipeline=set-foo

You can fetch the status of all running update by query requests with the Task API:

$response = $client->tasks()->list();

resp = client.tasks.list(detailed="true", actions="*byquery")
print(resp)

response = client.tasks.list(
  detailed: true,
  actions: '*byquery'
)
puts response

res, err := es.Tasks.List(
	es.Tasks.List.WithActions("*byquery"),
	es.Tasks.List.WithDetailed(true),
)
fmt.Println(res, err)

const response = await client.tasks.list({
  detailed: 'true',
  actions: '*byquery'
})
console.log(response)

GET _tasks?detailed=true&actions=*byquery

The responses looks like:

{
  "nodes" : {
    "r1A2WoRbTwKZ516z6NEs5A" : {
      "name" : "r1A2WoR",
      "transport_address" : "127.0.0.1:9300",
      "host" : "127.0.0.1",
      "ip" : "127.0.0.1:9300",
      "attributes" : {
        "testattr" : "test",
        "portsfile" : "true"
      },
      "tasks" : {
        "r1A2WoRbTwKZ516z6NEs5A:36619" : {
          "node" : "r1A2WoRbTwKZ516z6NEs5A",
          "id" : 36619,
          "type" : "transport",
          "action" : "indices:data/write/update/byquery",
          "status" : {    
            "total" : 6154,
            "updated" : 3500,
            "created" : 0,
            "deleted" : 0,
            "batches" : 4,
            "version_conflicts" : 0,
            "noops" : 0,
            "retries": {
              "bulk": 0,
              "search": 0
            },
            "throttled_millis": 0
          },
          "description" : ""
        }
      }
    }
  }
}

With the task id you can look up the task directly. The following example retrieves information about task r1A2WoRbTwKZ516z6NEs5A:36619:

$params = [
    'task_id' => 'r1A2WoRbTwKZ516z6NEs5A:36619',
];
$response = $client->tasks()->get($params);

resp = client.tasks.get(task_id="r1A2WoRbTwKZ516z6NEs5A:36619")
print(resp)

response = client.tasks.get(
  task_id: 'r1A2WoRbTwKZ516z6NEs5A:36619'
)
puts response

res, err := es.Tasks.Get(
	"r1A2WoRbTwKZ516z6NEs5A:36619",
)
fmt.Println(res, err)

const response = await client.tasks.get({
  task_id: 'r1A2WoRbTwKZ516z6NEs5A:36619'
})
console.log(response)

GET /_tasks/r1A2WoRbTwKZ516z6NEs5A:36619

The advantage of this API is that it integrates with wait_for_completion=false to transparently return the status of completed tasks. If the task is completed and wait_for_completion=false was set on it, then it’ll come back with a results or an error field. The cost of this feature is the document that wait_for_completion=false creates at .tasks/task/${taskId}. It is up to you to delete that document.

Any update by query can be cancelled using the Task Cancel API:

$params = [
    'task_id' => 'r1A2WoRbTwKZ516z6NEs5A:36619',
];
$response = $client->tasks()->cancel($params);

resp = client.tasks.cancel(task_id="r1A2WoRbTwKZ516z6NEs5A:36619")
print(resp)

response = client.tasks.cancel(
  task_id: 'r1A2WoRbTwKZ516z6NEs5A:36619'
)
puts response

res, err := es.Tasks.Cancel(
	es.Tasks.Cancel.WithTaskID("r1A2WoRbTwKZ516z6NEs5A:36619"),
)
fmt.Println(res, err)

const response = await client.tasks.cancel({
  task_id: 'r1A2WoRbTwKZ516z6NEs5A:36619'
})
console.log(response)

POST _tasks/r1A2WoRbTwKZ516z6NEs5A:36619/_cancel

The task ID can be found using the tasks API.

Cancellation should happen quickly but might take a few seconds. The task status API above will continue to list the update by query task until this task checks that it has been cancelled and terminates itself.

The value of requests_per_second can be changed on a running update by query using the _rethrottle API:

$params = [
    'task_id' => 'r1A2WoRbTwKZ516z6NEs5A:36619',
];
$response = $client->updateByQueryRethrottle($params);

resp = client.update_by_query_rethrottle(
    task_id="r1A2WoRbTwKZ516z6NEs5A:36619", requests_per_second="-1",
)
print(resp)

response = client.update_by_query_rethrottle(
  task_id: 'r1A2WoRbTwKZ516z6NEs5A:36619',
  requests_per_second: -1
)
puts response

res, err := es.UpdateByQueryRethrottle(
	"r1A2WoRbTwKZ516z6NEs5A:36619",
	esapi.IntPtr(-1),
)
fmt.Println(res, err)

const response = await client.updateByQueryRethrottle({
  task_id: 'r1A2WoRbTwKZ516z6NEs5A:36619',
  requests_per_second: '-1'
})
console.log(response)

POST _update_by_query/r1A2WoRbTwKZ516z6NEs5A:36619/_rethrottle?requests_per_second=-1

The task ID can be found using the tasks API.

Just like when setting it on the _update_by_query API, requests_per_second can be either -1 to disable throttling or any decimal number like 1.7 or 12 to throttle to that level. Rethrottling that speeds up the query takes effect immediately, but rethrotting that slows down the query will take effect after completing the current batch. This prevents scroll timeouts.

Slice an update by query manually by providing a slice id and total number of slices to each request:

POST my-index-000001/_update_by_query
{
  "slice": {
    "id": 0,
    "max": 2
  },
  "script": {
    "source": "ctx._source['extra'] = 'test'"
  }
}
POST my-index-000001/_update_by_query
{
  "slice": {
    "id": 1,
    "max": 2
  },
  "script": {
    "source": "ctx._source['extra'] = 'test'"
  }
}

Which you can verify works with:

GET _refresh
POST my-index-000001/_search?size=0&q=extra:test&filter_path=hits.total

Which results in a sensible total like this one:

{
  "hits": {
    "total": {
        "value": 120,
        "relation": "eq"
    }
  }
}

You can also let update by query automatically parallelize using Sliced scroll to slice on _id. Use slices to specify the number of slices to use:

POST my-index-000001/_update_by_query?refresh&slices=5
{
  "script": {
    "source": "ctx._source['extra'] = 'test'"
  }
}

Which you also can verify works with:

POST my-index-000001/_search?size=0&q=extra:test&filter_path=hits.total

Which results in a sensible total like this one:

{
  "hits": {
    "total": {
        "value": 120,
        "relation": "eq"
    }
  }
}

Setting slices to auto will let Elasticsearch choose the number of slices to use. This setting will use one slice per shard, up to a certain limit. If there are multiple source data streams or indices, it will choose the number of slices based on the index or backing index with the smallest number of shards.

Adding slices to _update_by_query just automates the manual process used in the section above, creating sub-requests which means it has some quirks:

Say you created an index without dynamic mapping, filled it with data, and then added a mapping value to pick up more fields from the data:

$params = [
    'index' => 'test',
    'body' => [
        'mappings' => [
            'dynamic' => false,
            'properties' => [
                'text' => [
                    'type' => 'text',
                ],
            ],
        ],
    ],
];
$response = $client->indices()->create($params);
$params = [
    'index' => 'test',
    'body' => [
        'text' => 'words words',
        'flag' => 'bar',
    ],
];
$response = $client->index($params);
$params = [
    'index' => 'test',
    'body' => [
        'text' => 'words words',
        'flag' => 'foo',
    ],
];
$response = $client->index($params);
$params = [
    'index' => 'test',
    'body' => [
        'properties' => [
            'text' => [
                'type' => 'text',
            ],
            'flag' => [
                'type' => 'text',
                'analyzer' => 'keyword',
            ],
        ],
    ],
];
$response = $client->indices()->putMapping($params);

resp = client.indices.create(
    index="test",
    body={
        "mappings": {
            "dynamic": False,
            "properties": {"text": {"type": "text"}},
        }
    },
)
print(resp)

resp = client.index(
    index="test",
    refresh=True,
    body={"text": "words words", "flag": "bar"},
)
print(resp)

resp = client.index(
    index="test",
    refresh=True,
    body={"text": "words words", "flag": "foo"},
)
print(resp)

resp = client.indices.put_mapping(
    index="test",
    body={
        "properties": {
            "text": {"type": "text"},
            "flag": {"type": "text", "analyzer": "keyword"},
        }
    },
)
print(resp)

response = client.indices.create(
  index: 'test',
  body: {
    mappings: {
      dynamic: false,
      properties: {
        text: {
          type: 'text'
        }
      }
    }
  }
)
puts response

response = client.indices.create(
  index: 'test',
  refresh: true,
  body: {
    text: 'words words',
    flag: 'bar'
  }
)
puts response

response = client.indices.create(
  index: 'test',
  refresh: true,
  body: {
    text: 'words words',
    flag: 'foo'
  }
)
puts response

response = client.indices.create(
  index: 'test',
  body: {
    properties: {
      text: {
        type: 'text'
      },
      flag: {
        type: 'text',
        analyzer: 'keyword'
      }
    }
  }
)
puts response

{
	res, err := es.Indices.Create(
		"test",
		es.Indices.Create.WithBody(strings.NewReader(`{
	  "mappings": {
	    "dynamic": false,
	    "properties": {
	      "text": {
	        "type": "text"
	      }
	    }
	  }
	}`)),
	)
	fmt.Println(res, err)
}

{
	res, err := es.Index(
		"test",
		strings.NewReader(`{
	  "text": "words words",
	  "flag": "bar"
	}`),
		es.Index.WithRefresh("true"),
		es.Index.WithPretty(),
	)
	fmt.Println(res, err)
}

{
	res, err := es.Index(
		"test",
		strings.NewReader(`{
	  "text": "words words",
	  "flag": "foo"
	}`),
		es.Index.WithRefresh("true"),
		es.Index.WithPretty(),
	)
	fmt.Println(res, err)
}

{
	res, err := es.Indices.PutMapping(
		[]string{"test"},
		strings.NewReader(`{
	  "properties": {
	    "text": {
	      "type": "text"
	    },
	    "flag": {
	      "type": "text",
	      "analyzer": "keyword"
	    }
	  }
	}`),
	)
	fmt.Println(res, err)
}

const response0 = await client.indices.create({
  index: 'test',
  body: {
    mappings: {
      dynamic: false,
      properties: {
        text: {
          type: 'text'
        }
      }
    }
  }
})
console.log(response0)

const response1 = await client.index({
  index: 'test',
  refresh: true,
  body: {
    text: 'words words',
    flag: 'bar'
  }
})
console.log(response1)

const response2 = await client.index({
  index: 'test',
  refresh: true,
  body: {
    text: 'words words',
    flag: 'foo'
  }
})
console.log(response2)

const response3 = await client.indices.putMapping({
  index: 'test',
  body: {
    properties: {
      text: {
        type: 'text'
      },
      flag: {
        type: 'text',
        analyzer: 'keyword'
      }
    }
  }
})
console.log(response3)

PUT test
{
  "mappings": {
    "dynamic": false,   
    "properties": {
      "text": {"type": "text"}
    }
  }
}

POST test/_doc?refresh
{
  "text": "words words",
  "flag": "bar"
}
POST test/_doc?refresh
{
  "text": "words words",
  "flag": "foo"
}
PUT test/_mapping   
{
  "properties": {
    "text": {"type": "text"},
    "flag": {"type": "text", "analyzer": "keyword"}
  }
}

Searching for the data won’t find anything:

$params = [
    'index' => 'test',
    'body' => [
        'query' => [
            'match' => [
                'flag' => 'foo',
            ],
        ],
    ],
];
$response = $client->search($params);

resp = client.search(
    index="test",
    filter_path="hits.total",
    body={"query": {"match": {"flag": "foo"}}},
)
print(resp)

response = client.search(
  index: 'test',
  filter_path: 'hits.total',
  body: {
    query: {
      match: {
        flag: 'foo'
      }
    }
  }
)
puts response

res, err := es.Search(
	es.Search.WithIndex("test"),
	es.Search.WithBody(strings.NewReader(`{
	  "query": {
	    "match": {
	      "flag": "foo"
	    }
	  }
	}`)),
	es.Search.WithFilterPath("hits.total"),
	es.Search.WithPretty(),
)
fmt.Println(res, err)

const response = await client.search({
  index: 'test',
  filter_path: 'hits.total',
  body: {
    query: {
      match: {
        flag: 'foo'
      }
    }
  }
})
console.log(response)

POST test/_search?filter_path=hits.total
{
  "query": {
    "match": {
      "flag": "foo"
    }
  }
}

{
  "hits" : {
    "total": {
        "value": 0,
        "relation": "eq"
    }
  }
}

But you can issue an _update_by_query request to pick up the new mapping:

$params = [
    'index' => 'test',
];
$response = $client->updateByQuery($params);
$params = [
    'index' => 'test',
    'body' => [
        'query' => [
            'match' => [
                'flag' => 'foo',
            ],
        ],
    ],
];
$response = $client->search($params);

resp = client.update_by_query(
    index="test", refresh=True, conflicts="proceed",
)
print(resp)

resp = client.search(
    index="test",
    filter_path="hits.total",
    body={"query": {"match": {"flag": "foo"}}},
)
print(resp)

response = client.update_by_query(
  index: 'test',
  refresh: true,
  conflicts: 'proceed'
)
puts response

response = client.update_by_query(
  index: 'test',
  filter_path: 'hits.total',
  body: {
    query: {
      match: {
        flag: 'foo'
      }
    }
  }
)
puts response

{
	res, err := es.UpdateByQuery(
		[]string{"test"},
		es.UpdateByQuery.WithConflicts("proceed"),
		es.UpdateByQuery.WithRefresh(true),
	)
	fmt.Println(res, err)
}

{
	res, err := es.Search(
		es.Search.WithIndex("test"),
		es.Search.WithBody(strings.NewReader(`{
	  "query": {
	    "match": {
	      "flag": "foo"
	    }
	  }
	}`)),
		es.Search.WithFilterPath("hits.total"),
		es.Search.WithPretty(),
	)
	fmt.Println(res, err)
}

const response0 = await client.updateByQuery({
  index: 'test',
  refresh: true,
  conflicts: 'proceed'
})
console.log(response0)

const response1 = await client.search({
  index: 'test',
  filter_path: 'hits.total',
  body: {
    query: {
      match: {
        flag: 'foo'
      }
    }
  }
})
console.log(response1)

POST test/_update_by_query?refresh&conflicts=proceed
POST test/_search?filter_path=hits.total
{
  "query": {
    "match": {
      "flag": "foo"
    }
  }
}

{
  "hits" : {
    "total": {
        "value": 1,
        "relation": "eq"
    }
  }
}

You can do the exact same thing when adding a field to a multifield.