TSDB: Support GET and DELETE and doc versioning

benchmarks/src/main/java/org/elasticsearch/benchmark/search/QueryParserHelperBenchmark.java

@@ -31,10 +31,10 @@
 import org.elasticsearch.index.analysis.IndexAnalyzers;
 import org.elasticsearch.index.analysis.NamedAnalyzer;
 import org.elasticsearch.index.fielddata.IndexFieldDataCache;
-import org.elasticsearch.index.mapper.IdFieldMapper;
 import org.elasticsearch.index.mapper.MapperRegistry;
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.mapper.ParsedDocument;
+import org.elasticsearch.index.mapper.ProvidedIdFieldMapper;
 import org.elasticsearch.index.mapper.SourceToParse;
 import org.elasticsearch.index.query.SearchExecutionContext;
 import org.elasticsearch.index.search.QueryParserHelper;
@@ -184,7 +184,7 @@ protected final MapperService createMapperService(String mappings) {
             similarityService,
             mapperRegistry,
             () -> { throw new UnsupportedOperationException(); },
-            new IdFieldMapper(() -> true),
+            new ProvidedIdFieldMapper(() -> true),
             new ScriptCompiler() {
                 @Override
                 public <T> T compile(Script script, ScriptContext<T> scriptContext) {

docs/changelog/82633.yaml

@@ -0,0 +1,5 @@
+pr: 82633
+summary: "TSDB: Support GET and DELETE and doc versioning"
+area: TSDB
+type: feature
+issues: []

modules/data-streams/src/javaRestTest/java/org/elasticsearch/datastreams/TsdbDataStreamRestIT.java

@@ -17,7 +17,9 @@
 import java.io.IOException;
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
+import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
 
 import static org.elasticsearch.cluster.metadata.DataStreamTestHelper.backingIndexEqualTo;
 import static org.hamcrest.Matchers.aMapWithSize;
@@ -301,10 +303,15 @@ public void testMigrateRegularDataStreamToTsdbDataStream() throws Exception {
         int numDocs = 32;
         var currentTime = Instant.now();
         var currentMinus30Days = currentTime.minus(30, ChronoUnit.DAYS);
+        Set<Instant> times = new HashSet<>();
         for (int i = 0; i < numRollovers; i++) {
             for (int j = 0; j < numDocs; j++) {
                 var indexRequest = new Request("POST", "/k8s/_doc");
-                var time = Instant.ofEpochMilli(randomLongBetween(currentMinus30Days.toEpochMilli(), currentTime.toEpochMilli()));
+                var time = randomValueOtherThanMany(
+                    times::contains,
+                    () -> Instant.ofEpochMilli(randomLongBetween(currentMinus30Days.toEpochMilli(), currentTime.toEpochMilli()))
+                );
+                times.add(time);
                 indexRequest.setJsonEntity(DOC.replace("$time", formatInstant(time)));
                 var response = client().performRequest(indexRequest);
                 assertOK(response);
@@ -350,13 +357,15 @@ public void testMigrateRegularDataStreamToTsdbDataStream() throws Exception {
         assertThat(newIndex, backingIndexEqualTo("k8s", 6));
 
         // Ingest documents that will land in the new tsdb backing index:
+        var t = currentTime;
         for (int i = 0; i < numDocs; i++) {
             var indexRequest = new Request("POST", "/k8s/_doc");
-            indexRequest.setJsonEntity(DOC.replace("$time", formatInstant(currentTime)));
+            indexRequest.setJsonEntity(DOC.replace("$time", formatInstant(t)));
             var response = client().performRequest(indexRequest);
             assertOK(response);
             var responseBody = entityAsMap(response);
             assertThat((String) responseBody.get("_index"), backingIndexEqualTo("k8s", 6));
+            t = t.plusMillis(1000);
         }
 
         // Fail if documents target older non tsdb backing index:

modules/lang-painless/src/main/java/org/elasticsearch/painless/action/PainlessExecuteAction.java

@@ -48,8 +48,7 @@
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.IndexService;
 import org.elasticsearch.index.mapper.DateFieldMapper;
-import org.elasticsearch.index.mapper.DocumentParser;
-import org.elasticsearch.index.mapper.MappingLookup;
+import org.elasticsearch.index.mapper.DocumentMapper;
 import org.elasticsearch.index.mapper.ParsedDocument;
 import org.elasticsearch.index.mapper.SourceToParse;
 import org.elasticsearch.index.query.AbstractQueryBuilder;
@@ -701,22 +700,22 @@ private static Response prepareRamIndex(
             CheckedBiFunction<SearchExecutionContext, LeafReaderContext, Response, IOException> handler,
             IndexService indexService
         ) throws IOException {
-
             Analyzer defaultAnalyzer = indexService.getIndexAnalyzers().getDefaultIndexAnalyzer();
 
             try (Directory directory = new ByteBuffersDirectory()) {
                 try (IndexWriter indexWriter = new IndexWriter(directory, new IndexWriterConfig(defaultAnalyzer))) {
-                    String index = indexService.index().getName();
                     BytesReference document = request.contextSetup.document;
                     XContentType xContentType = request.contextSetup.xContentType;
                     SourceToParse sourceToParse = new SourceToParse("_id", document, xContentType);
-                    MappingLookup mappingLookup = indexService.mapperService().mappingLookup();
-                    DocumentParser documentParser = indexService.mapperService().documentParser();
+                    DocumentMapper documentMapper = indexService.mapperService().documentMapper();
+                    if (documentMapper == null) {
+                        documentMapper = DocumentMapper.createEmpty(indexService.mapperService());
+                    }
                     // Note that we are not doing anything with dynamic mapping updates, hence fields that are not mapped but are present
                     // in the sample doc are not accessible from the script through doc['field'].
                     // This is a problem especially for indices that have no mappings, as no fields will be accessible, neither through doc
                     // nor _source (if there are no mappings there are no metadata fields).
-                    ParsedDocument parsedDocument = documentParser.parseDocument(sourceToParse, mappingLookup);
+                    ParsedDocument parsedDocument = documentMapper.parse(sourceToParse);
                     indexWriter.addDocuments(parsedDocument.docs());
                     try (IndexReader indexReader = DirectoryReader.open(indexWriter)) {
                         final IndexSearcher searcher = new IndexSearcher(indexReader);

modules/parent-join/src/main/java/org/elasticsearch/join/mapper/ParentJoinFieldMapper.java

@@ -70,10 +70,12 @@ public static class Defaults {
     }
 
     private static void checkIndexCompatibility(IndexSettings settings, String name) {
+        String indexName = settings.getIndex().getName();
         if (settings.getIndexMetadata().isRoutingPartitionedIndex()) {
-            throw new IllegalStateException(
-                "cannot create join field [" + name + "] " + "for the partitioned index " + "[" + settings.getIndex().getName() + "]"
-            );
+            throw new IllegalStateException("cannot create join field [" + name + "] for the partitioned index [" + indexName + "]");
+        }
+        if (settings.getIndexMetadata().getRoutingPaths().isEmpty() == false) {
+            throw new IllegalStateException("cannot create join field [" + name + "] for the index [" + indexName + "] with routing_path");
         }
     }
 
@@ -141,7 +143,7 @@ public ParentJoinFieldMapper build(MapperBuilderContext context) {
         }
     }
 
-    public static TypeParser PARSER = new TypeParser((n, c) -> {
+    public static final TypeParser PARSER = new TypeParser((n, c) -> {
         checkIndexCompatibility(c.getIndexSettings(), n);
         return new Builder(n);
     });
@@ -293,7 +295,7 @@ public void parse(DocumentParserContext context) throws IOException {
         if (fieldType().joiner.parentTypeExists(name)) {
             // Index the document as a parent
             String fieldName = fieldType().joiner.childJoinField(name);
-            parentIdFields.get(fieldName).indexValue(context, context.sourceToParse().id());
+            parentIdFields.get(fieldName).indexValue(context, context.id());
         }
 
         BytesRef binaryValue = new BytesRef(name);

modules/reindex/src/yamlRestTest/resources/rest-api-spec/test/delete_by_query/10_basic.yml

@@ -499,12 +499,9 @@ tsdb:
           - '{"@timestamp": "2021-04-28T18:50:03.142Z", "metricset": "pod", "k8s": {"pod": {"name": "cow", "uid":"1c4fc7b8-93b7-4ba8-b609-2a48af2f8e39", "ip": "10.10.55.4", "network": {"tx": 1434521831, "rx": 530575198}}}}'
 
   - do:
-      catch: bad_request
       delete_by_query:
         index: tsdb
         body:
           query:
             match_all: {}
-
-  - match: {failures.0.status: 400}
-  - match: {failures.0.cause.reason: "delete is not supported because the destination index [tsdb] is in time series mode"}
+  - match: {deleted: 1}