From 16e75aaa571aa0fb8fec36747e55680988334e75 Mon Sep 17 00:00:00 2001 From: Tim Brooks Date: Fri, 24 Jul 2026 00:09:07 -0500 Subject: [PATCH 1/3] Introduce EngineBatch abstraction MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the two-argument indexBatch(List, SourceBatch) on Engine and IndexShard with a single EngineBatch record. No behavior change — callers and overrides are updated mechanically. This unblocks swapping the internal batch representation in a follow-on PR. --- .../XContentMeteringParserDecoratorIT.java | 8 ++--- .../action/bulk/ShardBatchIndexer.java | 5 +-- .../elasticsearch/index/engine/Engine.java | 7 ++-- .../index/engine/EngineBatch.java | 23 +++++++++++++ .../index/engine/InternalEngine.java | 4 ++- .../elasticsearch/index/shard/IndexShard.java | 15 ++++---- .../index/engine/InternalEngineTests.java | 34 +++++++++---------- .../index/engine/InternalTestEngine.java | 7 ++-- .../ccr/index/engine/FollowingEngine.java | 8 ++--- 9 files changed, 68 insertions(+), 43 deletions(-) create mode 100644 server/src/main/java/org/elasticsearch/index/engine/EngineBatch.java diff --git a/server/src/internalClusterTest/java/org/elasticsearch/plugins/internal/XContentMeteringParserDecoratorIT.java b/server/src/internalClusterTest/java/org/elasticsearch/plugins/internal/XContentMeteringParserDecoratorIT.java index 9df2dbafcfdcd..47f4572a7b0b3 100644 --- a/server/src/internalClusterTest/java/org/elasticsearch/plugins/internal/XContentMeteringParserDecoratorIT.java +++ b/server/src/internalClusterTest/java/org/elasticsearch/plugins/internal/XContentMeteringParserDecoratorIT.java @@ -12,6 +12,7 @@ import org.elasticsearch.action.index.IndexRequest; import org.elasticsearch.index.Index; import org.elasticsearch.index.IndexSettings; +import org.elasticsearch.index.engine.EngineBatch; import org.elasticsearch.index.engine.EngineFactory; import org.elasticsearch.index.engine.InternalEngine; import org.elasticsearch.index.mapper.MapperService; @@ -20,7 +21,6 @@ import org.elasticsearch.plugins.EnginePlugin; import org.elasticsearch.plugins.IngestPlugin; import org.elasticsearch.plugins.Plugin; -import org.elasticsearch.sourcebatch.SourceBatch; import org.elasticsearch.test.ESIntegTestCase; import org.elasticsearch.xcontent.FilterXContentParserWrapper; import org.elasticsearch.xcontent.XContentParser; @@ -110,9 +110,9 @@ public IndexResult index(Index index) throws IOException { } @Override - public List indexBatch(List operations, SourceBatch batch) throws IOException { - List results = super.indexBatch(operations, batch); - for (Index op : operations) { + public List indexBatch(EngineBatch batch) throws IOException { + List results = super.indexBatch(batch); + for (Index op : batch.operations()) { reportDocumentSize(op.parsedDoc()); } return results; diff --git a/server/src/main/java/org/elasticsearch/action/bulk/ShardBatchIndexer.java b/server/src/main/java/org/elasticsearch/action/bulk/ShardBatchIndexer.java index ffc3892c749e9..63f928fce8284 100644 --- a/server/src/main/java/org/elasticsearch/action/bulk/ShardBatchIndexer.java +++ b/server/src/main/java/org/elasticsearch/action/bulk/ShardBatchIndexer.java @@ -19,6 +19,7 @@ import org.elasticsearch.core.Nullable; import org.elasticsearch.eirf.EirfRowXContentParser; import org.elasticsearch.index.engine.Engine; +import org.elasticsearch.index.engine.EngineBatch; import org.elasticsearch.index.mapper.ShardBatchMapper; import org.elasticsearch.index.mapper.SourceToParse; import org.elasticsearch.index.seqno.SequenceNumbers; @@ -135,7 +136,7 @@ private static void doBatchIndexOnPrimary( // The chunk's operations map 1:1 to the rows [chunkStart, chunkEnd); pass the matching slice so the // engine can write them as a single Translog.IndexBatch record. final SourceBatch chunkBatch = batch.slice(chunkStart, chunkEnd); - final List results = primary.applyIndexOperationBatchOnPrimary(operations, chunkBatch); + final List results = primary.applyIndexOperationBatchOnPrimary(new EngineBatch(operations, chunkBatch)); for (Engine.IndexResult result : results) { assert context.hasMoreOperationsToExecute(); @@ -225,7 +226,7 @@ static ReplicaBatchResult performBatchIndexOnReplica(BulkItemRequest[] items, So // operations are the contiguous run [chunkStart, chunkStart + operations.size()); pass the matching slice // so the engine writes them as a single Translog.IndexBatch record. final SourceBatch chunkBatch = batch.slice(chunkStart, chunkStart + operations.size()); - final List results = replica.applyIndexOperationBatchOnReplica(operations, chunkBatch); + final List results = replica.applyIndexOperationBatchOnReplica(new EngineBatch(operations, chunkBatch)); for (Engine.IndexResult result : results) { if (result.getFailure() != null) { throw result.getFailure(); diff --git a/server/src/main/java/org/elasticsearch/index/engine/Engine.java b/server/src/main/java/org/elasticsearch/index/engine/Engine.java index b0e0879bd818e..341c20b7303be 100644 --- a/server/src/main/java/org/elasticsearch/index/engine/Engine.java +++ b/server/src/main/java/org/elasticsearch/index/engine/Engine.java @@ -102,7 +102,6 @@ import org.elasticsearch.indices.IndexingMemoryController; import org.elasticsearch.indices.recovery.RecoverySettings; import org.elasticsearch.search.suggest.completion.CompletionStats; -import org.elasticsearch.sourcebatch.SourceBatch; import org.elasticsearch.threadpool.ThreadPool; import org.elasticsearch.transport.Transports; @@ -744,9 +743,9 @@ public Condition newCondition() { */ public abstract IndexResult index(Index index) throws IOException; - public List indexBatch(List operations, SourceBatch batch) throws IOException { - ArrayList results = new ArrayList<>(operations.size()); - for (Index index : operations) { + public List indexBatch(EngineBatch batch) throws IOException { + ArrayList results = new ArrayList<>(batch.operations().size()); + for (Index index : batch.operations()) { results.add(index(index)); } return results; diff --git a/server/src/main/java/org/elasticsearch/index/engine/EngineBatch.java b/server/src/main/java/org/elasticsearch/index/engine/EngineBatch.java new file mode 100644 index 0000000000000..cabc2890e21f8 --- /dev/null +++ b/server/src/main/java/org/elasticsearch/index/engine/EngineBatch.java @@ -0,0 +1,23 @@ +/* + * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one + * or more contributor license agreements. Licensed under the "Elastic License + * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side + * Public License v 1"; you may not use this file except in compliance with, at + * your election, the "Elastic License 2.0", the "GNU Affero General Public + * License v3.0 only", or the "Server Side Public License, v 1". + */ + +package org.elasticsearch.index.engine; + +import org.elasticsearch.sourcebatch.SourceBatch; + +import java.util.List; + +/** + * A batch of index operations ready for engine-level processing, produced by the bulk batch + * indexing path and consumed by {@link Engine#indexBatch}. + * + * @param operations per-document {@link Engine.Index} operations for this batch + * @param sourceBatch the raw encoded source data backing the operations + */ +public record EngineBatch(List operations, SourceBatch sourceBatch) {} diff --git a/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java b/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java index fa0d5ecaef0b3..192dc567a7b17 100644 --- a/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java +++ b/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java @@ -1385,7 +1385,9 @@ public IndexResult index(Index index) throws IOException { } @Override - public List indexBatch(List operations, SourceBatch batch) throws IOException { + public List indexBatch(EngineBatch engineBatch) throws IOException { + final List operations = engineBatch.operations(); + final SourceBatch batch = engineBatch.sourceBatch(); assert operations.size() == batch.docCount() : "operations [" + operations.size() + "] must map 1:1 to batch rows [" + batch.docCount() + "]"; try (var ignored = acquireEnsureOpenRef()) { diff --git a/server/src/main/java/org/elasticsearch/index/shard/IndexShard.java b/server/src/main/java/org/elasticsearch/index/shard/IndexShard.java index 14f7084f15980..9b572d550b2d2 100644 --- a/server/src/main/java/org/elasticsearch/index/shard/IndexShard.java +++ b/server/src/main/java/org/elasticsearch/index/shard/IndexShard.java @@ -96,6 +96,7 @@ import org.elasticsearch.index.engine.CommitStats; import org.elasticsearch.index.engine.Engine; import org.elasticsearch.index.engine.Engine.GetResult; +import org.elasticsearch.index.engine.EngineBatch; import org.elasticsearch.index.engine.EngineConfig; import org.elasticsearch.index.engine.EngineException; import org.elasticsearch.index.engine.EngineFactory; @@ -165,7 +166,6 @@ import org.elasticsearch.search.internal.FieldUsageTrackingDirectoryReader; import org.elasticsearch.search.suggest.completion.CompletionStats; import org.elasticsearch.snapshots.Snapshot; -import org.elasticsearch.sourcebatch.SourceBatch; import org.elasticsearch.threadpool.ThreadPool; import org.elasticsearch.transport.Transports; @@ -1144,29 +1144,30 @@ public static Engine.Index prepareIndex( * Applies a batch of index operations on the primary. Returns null if any operation requires a mapping update, * signaling the caller to fall back to the item-by-item path. */ - public List applyIndexOperationBatchOnPrimary(List operations, SourceBatch batch) throws IOException { + public List applyIndexOperationBatchOnPrimary(EngineBatch batch) throws IOException { ensureWriteAllowed(Engine.Operation.Origin.PRIMARY); final Engine engine = getEngine(); - return indexBatch(engine, operations, batch); + return indexBatch(engine, batch); } /** * Applies a batch of index operations on a replica. */ - public List applyIndexOperationBatchOnReplica(List operations, SourceBatch batch) throws IOException { + public List applyIndexOperationBatchOnReplica(EngineBatch batch) throws IOException { ensureWriteAllowed(Engine.Operation.Origin.REPLICA); final Engine engine = getEngine(); - return indexBatch(engine, operations, batch); + return indexBatch(engine, batch); } - private List indexBatch(Engine engine, List operations, SourceBatch batch) throws IOException { + private List indexBatch(Engine engine, EngineBatch batch) throws IOException { + final List operations = batch.operations(); List preIndexOps = new ArrayList<>(operations.size()); // TODO: Right now the only production users are stats. Should add batch listener. for (Engine.Index op : operations) { preIndexOps.add(indexingOperationListeners.preIndex(shardId, op)); } try { - List results = engine.indexBatch(preIndexOps, batch); + List results = engine.indexBatch(new EngineBatch(preIndexOps, batch.sourceBatch())); // TODO: Look at if these can be batch optimized for (int i = 0; i < results.size(); i++) { indexingOperationListeners.postIndex(shardId, preIndexOps.get(i), results.get(i)); diff --git a/server/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java b/server/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java index c70c753988b4e..eb502e16fffc9 100644 --- a/server/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java +++ b/server/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java @@ -237,7 +237,7 @@ private static Engine.IndexResult indexDoc(Engine engine, Engine.Index operation if (randomBoolean()) { SourceBatch batch = tryEncodeAsBatch(List.of(operation)); if (batch != null) { - return engine.indexBatch(List.of(operation), batch).getFirst(); + return engine.indexBatch(new EngineBatch(List.of(operation), batch)).getFirst(); } } return engine.index(operation); @@ -245,7 +245,7 @@ private static Engine.IndexResult indexDoc(Engine engine, Engine.Index operation /** * Encodes the given ops' sources into a {@link SourceBatch} that can be passed to - * {@link Engine#indexBatch(List, SourceBatch)}. + * {@link Engine#indexBatch(EngineBatch)}. */ private static SourceBatch encodeAsBatch(List operations) throws IOException { List sources = new ArrayList<>(operations.size()); @@ -8077,7 +8077,7 @@ public void testIndexBatchMultipleDocuments() throws IOException { ParsedDocument doc = createParsedDoc(Integer.toString(i), null); ops.add(indexForDoc(doc)); } - List results = engine.indexBatch(ops, encodeAsBatch(ops)); + List results = engine.indexBatch(new EngineBatch(ops, encodeAsBatch(ops))); assertThat(results, hasSize(ops.size())); for (int i = 0; i < results.size(); i++) { Engine.IndexResult result = results.get(i); @@ -8096,7 +8096,7 @@ public void testIndexBatchSequenceNumbersAreMonotonic() throws IOException { for (int i = 0; i < batchSize; i++) { ops.add(indexForDoc(createParsedDoc(Integer.toString(i), null))); } - List results = engine.indexBatch(ops, encodeAsBatch(ops)); + List results = engine.indexBatch(new EngineBatch(ops, encodeAsBatch(ops))); long prevSeqNo = -1; for (Engine.IndexResult result : results) { assertThat(result.getResultType(), equalTo(Engine.Result.Type.SUCCESS)); @@ -8109,7 +8109,7 @@ public void testIndexBatchSingletonEquivalentToIndex() throws IOException { ParsedDocument doc = createParsedDoc("1", null); Engine.Index op = indexForDoc(doc); List ops = List.of(op); - List batchResults = engine.indexBatch(ops, encodeAsBatch(ops)); + List batchResults = engine.indexBatch(new EngineBatch(ops, encodeAsBatch(ops))); assertThat(batchResults, hasSize(1)); Engine.IndexResult result = batchResults.getFirst(); assertThat(result.getResultType(), equalTo(Engine.Result.Type.SUCCESS)); @@ -8143,7 +8143,7 @@ public void testIndexBatchVersionConflict() throws IOException { firstResult.getTerm() ); List ops = List.of(conflictingOp); - List results = engine.indexBatch(ops, encodeAsBatch(ops)); + List results = engine.indexBatch(new EngineBatch(ops, encodeAsBatch(ops))); assertThat(results, hasSize(1)); assertThat(results.getFirst().getResultType(), equalTo(Engine.Result.Type.FAILURE)); assertThat(results.getFirst().getFailure(), instanceOf(VersionConflictEngineException.class)); @@ -8159,7 +8159,7 @@ public void testIndexBatchMixedNewAndExisting() throws IOException { indexForDoc(createParsedDoc("1", null)), indexForDoc(createParsedDoc("2", null)) ); - List results = engine.indexBatch(ops, encodeAsBatch(ops)); + List results = engine.indexBatch(new EngineBatch(ops, encodeAsBatch(ops))); assertThat(results, hasSize(3)); for (Engine.IndexResult result : results) { assertThat(result.getResultType(), equalTo(Engine.Result.Type.SUCCESS)); @@ -8187,7 +8187,7 @@ public void testIndexBatchVersionLookupFromLucene() throws IOException { for (ParsedDocument doc : docs) { updates.add(indexForDoc(doc)); } - List results = engine.indexBatch(updates, encodeAsBatch(updates)); + List results = engine.indexBatch(new EngineBatch(updates, encodeAsBatch(updates))); assertThat(results, hasSize(count)); for (Engine.IndexResult result : results) { assertThat(result.getResultType(), equalTo(Engine.Result.Type.SUCCESS)); @@ -8217,7 +8217,7 @@ public void testIndexBatchVersionConflictFromLucene() throws IOException { firstResult.getTerm() ); List ops = List.of(conflictingOp); - List results = engine.indexBatch(ops, encodeAsBatch(ops)); + List results = engine.indexBatch(new EngineBatch(ops, encodeAsBatch(ops))); assertThat(results, hasSize(1)); assertThat(results.getFirst().getResultType(), equalTo(Engine.Result.Type.FAILURE)); assertThat(results.getFirst().getFailure(), instanceOf(VersionConflictEngineException.class)); @@ -8234,7 +8234,7 @@ public void testIndexBatchMixedLuceneAndVersionMap() throws IOException { indexDoc(engine, indexForDoc(doc2)); List ops = List.of(indexForDoc(doc1), indexForDoc(doc2)); - List results = engine.indexBatch(ops, encodeAsBatch(ops)); + List results = engine.indexBatch(new EngineBatch(ops, encodeAsBatch(ops))); assertThat(results, hasSize(2)); for (Engine.IndexResult result : results) { assertThat(result.getResultType(), equalTo(Engine.Result.Type.SUCCESS)); @@ -8273,7 +8273,7 @@ public void testIndexBatchGcExpiredTombstoneTreatedAsNotFound() throws IOExcepti // where the stale live document is found, and the operation is incorrectly treated as an // update rather than a create. List ops = List.of(indexForDoc(doc)); - List results = engine.indexBatch(ops, encodeAsBatch(ops)); + List results = engine.indexBatch(new EngineBatch(ops, encodeAsBatch(ops))); assertThat(results, hasSize(1)); assertThat(results.getFirst().getResultType(), equalTo(Engine.Result.Type.SUCCESS)); assertThat(results.getFirst().isCreated(), equalTo(true)); @@ -8288,7 +8288,7 @@ public void testIndexBatchFastPathOnly() throws IOException { for (int i = 0; i < count; i++) { ops.add(appendOnlyPrimary(createParsedDoc(Integer.toString(i), null), false, timestamp + i)); } - List results = engine.indexBatch(ops, encodeAsBatch(ops)); + List results = engine.indexBatch(new EngineBatch(ops, encodeAsBatch(ops))); assertThat(results, hasSize(count)); for (Engine.IndexResult result : results) { assertThat(result.getResultType(), equalTo(Engine.Result.Type.SUCCESS)); @@ -8338,7 +8338,7 @@ public void testIndexBatchTimeSeriesPhase2() throws IOException { for (ParsedDocument doc : docs) { updates.add(indexForDoc(doc)); } - List results = engine.indexBatch(updates, encodeAsBatch(updates)); + List results = engine.indexBatch(new EngineBatch(updates, encodeAsBatch(updates))); assertThat(results, hasSize(count)); for (Engine.IndexResult result : results) { assertThat(result.getResultType(), equalTo(Engine.Result.Type.SUCCESS)); @@ -8354,7 +8354,7 @@ public void testMixedPrimaryTermThrows() throws IOException { Engine.Index op1 = new Engine.Index(newUid(doc1), primaryTerm.get(), doc1); Engine.Index op2 = new Engine.Index(newUid(doc2), primaryTerm.get() + 1, doc2); var updates = List.of(op1, op2); - expectThrows(AssertionError.class, () -> engine.indexBatch(updates, encodeAsBatch(updates))); + expectThrows(AssertionError.class, () -> engine.indexBatch(new EngineBatch(updates, encodeAsBatch(updates)))); } public void testIndexBatchSeqNosAreContiguous() throws IOException { @@ -8365,7 +8365,7 @@ public void testIndexBatchSeqNosAreContiguous() throws IOException { ops.add(indexForDoc(createParsedDoc(Integer.toString(i), null))); } long seqNoBefore = engine.getLocalCheckpointTracker().getMaxSeqNo(); - List results = engine.indexBatch(ops, encodeAsBatch(ops)); + List results = engine.indexBatch(new EngineBatch(ops, encodeAsBatch(ops))); assertThat(results, hasSize(batchSize)); long firstSeqNo = seqNoBefore + 1; @@ -8382,7 +8382,7 @@ public void testIndexBatchProcessedCheckpointAdvancesAfterBatch() throws IOExcep ops.add(indexForDoc(createParsedDoc(Integer.toString(i), null))); } long checkpointBefore = engine.getProcessedLocalCheckpoint(); - List results = engine.indexBatch(ops, encodeAsBatch(ops)); + List results = engine.indexBatch(new EngineBatch(ops, encodeAsBatch(ops))); assertThat(results, hasSize(batchSize)); long expectedCheckpoint = checkpointBefore + batchSize; @@ -8421,7 +8421,7 @@ public void testIndexBatchCheckpointWithVersionConflicts() throws IOException { long checkpointBefore = engine.getProcessedLocalCheckpoint(); List ops = List.of(conflicting, goodOp); - List results = engine.indexBatch(ops, encodeAsBatch(ops)); + List results = engine.indexBatch(new EngineBatch(ops, encodeAsBatch(ops))); assertThat(results, hasSize(2)); // conflicting op: failure, no seq no assigned diff --git a/test/framework/src/main/java/org/elasticsearch/index/engine/InternalTestEngine.java b/test/framework/src/main/java/org/elasticsearch/index/engine/InternalTestEngine.java index 6a118e6ee241d..0c5b9848978f0 100644 --- a/test/framework/src/main/java/org/elasticsearch/index/engine/InternalTestEngine.java +++ b/test/framework/src/main/java/org/elasticsearch/index/engine/InternalTestEngine.java @@ -12,7 +12,6 @@ import org.elasticsearch.common.util.concurrent.ConcurrentCollections; import org.elasticsearch.index.seqno.LocalCheckpointTracker; import org.elasticsearch.index.seqno.SequenceNumbers; -import org.elasticsearch.sourcebatch.SourceBatch; import java.io.IOException; import java.util.List; @@ -54,8 +53,8 @@ public IndexResult index(Index index) throws IOException { } @Override - public List indexBatch(List operations, SourceBatch batch) throws IOException { - for (Index index : operations) { + public List indexBatch(EngineBatch batch) throws IOException { + for (Index index : batch.operations()) { if (index.seqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) { idToMaxSeqNo.compute(index.id(), (id, existing) -> { if (existing == null) { @@ -68,7 +67,7 @@ public List indexBatch(List operations, SourceBatch batch) t }); } } - return super.indexBatch(operations, batch); + return super.indexBatch(batch); } @Override diff --git a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/index/engine/FollowingEngine.java b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/index/engine/FollowingEngine.java index e344133cc848b..6bb1d807bd0ef 100644 --- a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/index/engine/FollowingEngine.java +++ b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/index/engine/FollowingEngine.java @@ -20,12 +20,12 @@ import org.elasticsearch.common.lucene.Lucene; import org.elasticsearch.core.Assertions; import org.elasticsearch.index.VersionType; +import org.elasticsearch.index.engine.EngineBatch; import org.elasticsearch.index.engine.EngineConfig; import org.elasticsearch.index.engine.InternalEngine; import org.elasticsearch.index.mapper.SeqNoFieldMapper; import org.elasticsearch.index.seqno.SequenceNumbers; import org.elasticsearch.rest.RestStatus; -import org.elasticsearch.sourcebatch.SourceBatch; import org.elasticsearch.xpack.ccr.CcrSettings; import java.io.IOException; @@ -180,11 +180,11 @@ protected boolean assertPrimaryCanOptimizeAddDocument(final Index index) { } @Override - public List indexBatch(List operations, SourceBatch batch) throws IOException { + public List indexBatch(EngineBatch batch) throws IOException { // CCR following engine has special versioning semantics that are not compatible with // the optimized batch indexing path in InternalEngine. Fall back to sequential indexing. - List results = new ArrayList<>(operations.size()); - for (Index op : operations) { + List results = new ArrayList<>(batch.operations().size()); + for (Index op : batch.operations()) { results.add(index(op)); } return results; From e259425b33d4c4f8ff06a03f77a1bfcfc53733e2 Mon Sep 17 00:00:00 2001 From: Tim Brooks Date: Fri, 24 Jul 2026 23:43:34 -0500 Subject: [PATCH 2/3] Change --- .../elasticsearch/xpack/stateless/engine/IndexEngine.java | 6 ++++-- .../xpack/stateless/engine/IndexEngineTests.java | 7 ++++--- 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/x-pack/plugin/stateless/src/main/java/org/elasticsearch/xpack/stateless/engine/IndexEngine.java b/x-pack/plugin/stateless/src/main/java/org/elasticsearch/xpack/stateless/engine/IndexEngine.java index cd4db06b512d2..0a12211d48016 100644 --- a/x-pack/plugin/stateless/src/main/java/org/elasticsearch/xpack/stateless/engine/IndexEngine.java +++ b/x-pack/plugin/stateless/src/main/java/org/elasticsearch/xpack/stateless/engine/IndexEngine.java @@ -43,6 +43,7 @@ import org.elasticsearch.index.engine.ElasticsearchMergeScheduler; import org.elasticsearch.index.engine.ElasticsearchReaderManager; import org.elasticsearch.index.engine.Engine; +import org.elasticsearch.index.engine.EngineBatch; import org.elasticsearch.index.engine.EngineConfig; import org.elasticsearch.index.engine.EngineCreationFailureException; import org.elasticsearch.index.engine.EngineException; @@ -522,12 +523,13 @@ public IndexResult index(Index index) throws IOException { } @Override - public List indexBatch(List operations, SourceBatch batch) throws IOException { + public List indexBatch(EngineBatch engineBatch) throws IOException { checkNoNewOperationsWhileHollow(); + List operations = engineBatch.operations(); for (Index operation : operations) { documentParsingReporter.onParsingCompleted(operation.parsedDoc()); } - List results = super.indexBatch(operations, batch); + List results = super.indexBatch(engineBatch); for (int i = 0; i < results.size(); i++) { if (results.get(i).getResultType() == Result.Type.SUCCESS) { documentParsingReporter.onIndexingCompleted(operations.get(i).parsedDoc()); diff --git a/x-pack/plugin/stateless/src/test/java/org/elasticsearch/xpack/stateless/engine/IndexEngineTests.java b/x-pack/plugin/stateless/src/test/java/org/elasticsearch/xpack/stateless/engine/IndexEngineTests.java index 7cdefce96619b..df7a91c02ee80 100644 --- a/x-pack/plugin/stateless/src/test/java/org/elasticsearch/xpack/stateless/engine/IndexEngineTests.java +++ b/x-pack/plugin/stateless/src/test/java/org/elasticsearch/xpack/stateless/engine/IndexEngineTests.java @@ -24,6 +24,7 @@ import org.elasticsearch.index.IndexSettings; import org.elasticsearch.index.VersionType; import org.elasticsearch.index.engine.Engine; +import org.elasticsearch.index.engine.EngineBatch; import org.elasticsearch.index.engine.EngineConfig; import org.elasticsearch.index.engine.EngineException; import org.elasticsearch.index.engine.MergeMemoryEstimator; @@ -470,7 +471,7 @@ public void testDocSizeIsReportedUponBatchIndex() throws IOException { ) { // Success case: all docs in the batch succeed. List ops = List.of(randomDoc("id1"), randomDoc("id2"), randomDoc("id3")); - List results = engine.indexBatch(ops, encodeAsEscfBatch(ops)); + List results = engine.indexBatch(new EngineBatch(ops, encodeAsEscfBatch(ops))); for (int i = 0; i < results.size(); i++) { assertThat(results.get(i).getResultType(), equalTo(Engine.Result.Type.SUCCESS)); verify(documentSizeReporter).onParsingCompleted(eq(ops.get(i).parsedDoc())); @@ -480,7 +481,7 @@ public void testDocSizeIsReportedUponBatchIndex() throws IOException { // Failure case: a version-conflicting op does not get onIndexingCompleted. Engine.Index conflictingOp = versionConflictingIndexOperation(randomDoc("id1")); List failOps = List.of(conflictingOp); - List failResults = engine.indexBatch(failOps, encodeAsEscfBatch(failOps)); + List failResults = engine.indexBatch(new EngineBatch(failOps, encodeAsEscfBatch(failOps))); assertThat(failResults.get(0).getResultType(), equalTo(Engine.Result.Type.FAILURE)); verify(documentSizeReporter).onParsingCompleted(eq(conflictingOp.parsedDoc())); verify(documentSizeReporter, never()).onIndexingCompleted(eq(conflictingOp.parsedDoc())); @@ -497,7 +498,7 @@ public void testHollowEngineCannotIngestBatch() throws Exception { final var maxSeqNo = engine.getMaxSeqNo(); List batchOps = List.of(randomDoc(String.valueOf(1))); - expectThrows(IllegalStateException.class, () -> engine.indexBatch(batchOps, encodeAsEscfBatch(batchOps))); + expectThrows(IllegalStateException.class, () -> engine.indexBatch(new EngineBatch(batchOps, encodeAsEscfBatch(batchOps)))); assertThat(engine.getMaxSeqNo(), equalTo(maxSeqNo)); } } From 3440f65ff55ac7134d67926cf6e24a39933763c0 Mon Sep 17 00:00:00 2001 From: elasticsearchmachine Date: Sat, 25 Jul 2026 04:50:29 +0000 Subject: [PATCH 3/3] [CI] Auto commit changes from spotless --- .../org/elasticsearch/xpack/stateless/engine/IndexEngine.java | 1 - 1 file changed, 1 deletion(-) diff --git a/x-pack/plugin/stateless/src/main/java/org/elasticsearch/xpack/stateless/engine/IndexEngine.java b/x-pack/plugin/stateless/src/main/java/org/elasticsearch/xpack/stateless/engine/IndexEngine.java index 0a12211d48016..602ee499d32fd 100644 --- a/x-pack/plugin/stateless/src/main/java/org/elasticsearch/xpack/stateless/engine/IndexEngine.java +++ b/x-pack/plugin/stateless/src/main/java/org/elasticsearch/xpack/stateless/engine/IndexEngine.java @@ -63,7 +63,6 @@ import org.elasticsearch.plugins.internal.DocumentParsingProvider; import org.elasticsearch.plugins.internal.DocumentSizeAccumulator; import org.elasticsearch.plugins.internal.DocumentSizeReporter; -import org.elasticsearch.sourcebatch.SourceBatch; import org.elasticsearch.threadpool.ThreadPool; import org.elasticsearch.xpack.stateless.cache.SharedBlobCacheWarmingService; import org.elasticsearch.xpack.stateless.commits.BatchedCompoundCommit;