Merge pull request #2 from grcevski/spacetime_transactions

WIP: Transaction ops in translog

Author: henningandersen

server/src/internalClusterTest/java/org/elasticsearch/index/shard/IndexShardIT.java

@@ -362,7 +362,8 @@ public void testMaybeFlush() throws Exception {
             SequenceNumbers.UNASSIGNED_SEQ_NO,
             0,
             IndexRequest.UNSET_AUTO_GENERATED_TIMESTAMP,
-            false
+            false,
+            IndexShard.NO_TRANSACTION_ID
         );
         assertTrue(shard.shouldPeriodicallyFlush());
         final Translog translog = getTranslog(shard);
@@ -447,7 +448,8 @@ public void testMaybeRollTranslogGeneration() throws Exception {
                 SequenceNumbers.UNASSIGNED_SEQ_NO,
                 0,
                 IndexRequest.UNSET_AUTO_GENERATED_TIMESTAMP,
-                false
+                false,
+                IndexShard.NO_TRANSACTION_ID
             );
             final Translog.Location location = result.getTranslogLocation();
             shard.afterWriteOperation();

server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java

@@ -33,6 +33,7 @@
 import org.elasticsearch.cluster.action.index.MappingUpdatedAction;
 import org.elasticsearch.cluster.action.shard.ShardStateAction;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.compress.CompressedXContent;
 import org.elasticsearch.common.inject.Inject;
@@ -181,20 +182,33 @@ public static void performOnPrimary(
 
             @Override
             protected void doRun() throws Exception {
-                while (context.hasMoreOperationsToExecute()) {
-                    if (executeBulkItemRequest(
-                        context,
-                        updateHelper,
-                        nowInMillisSupplier,
-                        mappingUpdater,
-                        waitForMappingUpdate,
-                        ActionListener.wrap(v -> executor.execute(this), this::onRejection)
-                    ) == false) {
-                        // We are waiting for a mapping update on another thread, that will invoke this action again once its done
-                        // so we just break out here.
-                        return;
+                String uid = UUIDs.base64UUID();
+                long transactionId = -1L;
+                try {
+                    transactionId = primary.startTransaction(uid);
+                    while (context.hasMoreOperationsToExecute()) {
+                        if (executeBulkItemRequest(
+                            context,
+                            updateHelper,
+                            nowInMillisSupplier,
+                            mappingUpdater,
+                            waitForMappingUpdate,
+                            ActionListener.wrap(v -> executor.execute(this), this::onRejection),
+                            transactionId
+                        ) == false) {
+                            // We are waiting for a mapping update on another thread, that will invoke this action again once its done
+                            // so we just break out here.
+                            return;
+                        }
+                        assert context.isInitial(); // either completed and moved to next or reset
                     }
-                    assert context.isInitial(); // either completed and moved to next or reset
+
+                    primary.commitTransaction(uid, transactionId);
+                } catch (Exception x) {
+                    logger.warn("Encountered an error while executing bulk transaction", x);
+                    primary.rollbackTransaction(uid, transactionId);
+                } finally {
+                    primary.closeTransaction(uid, transactionId);
                 }
                 primary.getBulkOperationListener().afterBulk(request.totalSizeInBytes(), System.nanoTime() - startBulkTime);
                 // We're done, there's no more operations to execute so we resolve the wrapped listener
@@ -206,7 +220,6 @@ public void onRejection(Exception e) {
                 // We must finish the outstanding request. Finishing the outstanding request can include
                 // refreshing and fsyncing. Therefore, we must force execution on the WRITE thread.
                 executor.execute(new ActionRunnable<>(listener) {
-
                     @Override
                     protected void doRun() {
                         // Fail all operations after a bulk rejection hit an action that waited for a mapping update and finish the request
@@ -250,6 +263,25 @@ private void finishRequest() {
         }.run();
     }
 
+    static boolean executeBulkItemRequest(
+        BulkPrimaryExecutionContext context,
+        UpdateHelper updateHelper,
+        LongSupplier nowInMillisSupplier,
+        MappingUpdatePerformer mappingUpdater,
+        Consumer<ActionListener<Void>> waitForMappingUpdate,
+        ActionListener<Void> itemDoneListener
+    ) throws Exception {
+        return executeBulkItemRequest(
+            context,
+            updateHelper,
+            nowInMillisSupplier,
+            mappingUpdater,
+            waitForMappingUpdate,
+            itemDoneListener,
+            IndexShard.NO_TRANSACTION_ID
+        );
+    }
+
     /**
      * Executes bulk item requests and handles request execution exceptions.
      * @return {@code true} if request completed on this thread and the listener was invoked, {@code false} if the request triggered
@@ -261,7 +293,8 @@ static boolean executeBulkItemRequest(
         LongSupplier nowInMillisSupplier,
         MappingUpdatePerformer mappingUpdater,
         Consumer<ActionListener<Void>> waitForMappingUpdate,
-        ActionListener<Void> itemDoneListener
+        ActionListener<Void> itemDoneListener,
+        long transactionId
     ) throws Exception {
         final DocWriteRequest.OpType opType = context.getCurrent().opType();
 
@@ -306,7 +339,8 @@ static boolean executeBulkItemRequest(
                 request.id(),
                 request.versionType(),
                 request.ifSeqNo(),
-                request.ifPrimaryTerm()
+                request.ifPrimaryTerm(),
+                transactionId
             );
         } else {
             final IndexRequest request = context.getRequestToExecute();
@@ -324,7 +358,8 @@ static boolean executeBulkItemRequest(
                 request.ifSeqNo(),
                 request.ifPrimaryTerm(),
                 request.getAutoGeneratedTimestamp(),
-                request.isRetry()
+                request.isRetry(),
+                transactionId
             );
         }
         if (result.getResultType() == Engine.Result.Type.MAPPING_UPDATE_REQUIRED) {
@@ -509,7 +544,7 @@ private static BulkItemResponse processUpdateResponse(
     protected void dispatchedShardOperationOnReplica(BulkShardRequest request, IndexShard replica, ActionListener<ReplicaResult> listener) {
         ActionListener.completeWith(listener, () -> {
             final long startBulkTime = System.nanoTime();
-            final Translog.Location location = performOnReplica(request, replica);
+            final Translog.Location location = performOnReplica(request, replica, IndexShard.NO_TRANSACTION_ID);
             replica.getBulkOperationListener().afterBulk(request.totalSizeInBytes(), System.nanoTime() - startBulkTime);
             return new WriteReplicaResult<>(request, location, null, replica, logger);
         });
@@ -525,7 +560,7 @@ protected int replicaOperationCount(BulkShardRequest request) {
         return request.items().length;
     }
 
-    public static Translog.Location performOnReplica(BulkShardRequest request, IndexShard replica) throws Exception {
+    public static Translog.Location performOnReplica(BulkShardRequest request, IndexShard replica, long transactionId) throws Exception {
         Translog.Location location = null;
         for (int i = 0; i < request.items().length; i++) {
             final BulkItemRequest item = request.items()[i];
@@ -553,7 +588,7 @@ public static Translog.Location performOnReplica(BulkShardRequest request, Index
                     continue; // ignore replication as it's a noop
                 }
                 assert response.getResponse().getSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO;
-                operationResult = performOpOnReplica(response.getResponse(), item.request(), replica);
+                operationResult = performOpOnReplica(response.getResponse(), item.request(), replica, transactionId);
             }
             assert operationResult != null : "operation result must never be null when primary response has no failure";
             location = syncOperationResultOrThrow(operationResult, location);
@@ -564,7 +599,8 @@ public static Translog.Location performOnReplica(BulkShardRequest request, Index
     private static Engine.Result performOpOnReplica(
         DocWriteResponse primaryResponse,
         DocWriteRequest<?> docWriteRequest,
-        IndexShard replica
+        IndexShard replica,
+        long transactionId
     ) throws Exception {
         final Engine.Result result;
         switch (docWriteRequest.opType()) {
@@ -584,7 +620,8 @@ private static Engine.Result performOpOnReplica(
                     primaryResponse.getVersion(),
                     indexRequest.getAutoGeneratedTimestamp(),
                     indexRequest.isRetry(),
-                    sourceToParse
+                    sourceToParse,
+                    transactionId
                 );
                 break;
             case DELETE:
@@ -593,7 +630,8 @@ private static Engine.Result performOpOnReplica(
                     primaryResponse.getSeqNo(),
                     primaryResponse.getPrimaryTerm(),
                     primaryResponse.getVersion(),
-                    deleteRequest.id()
+                    deleteRequest.id(),
+                    transactionId
                 );
                 break;
             default:

server/src/main/java/org/elasticsearch/index/engine/Engine.java

@@ -321,6 +321,14 @@ public Condition newCondition() {
         }
     }
 
+    public abstract long startTransaction(String id) throws IOException;
+
+    public abstract boolean commitTransaction(String id, long transactionId) throws IOException;
+
+    public abstract boolean rollbackTransaction(String id, long transactionId) throws IOException;
+
+    public abstract boolean closeTransaction(String id, long transactionId) throws IOException;
+
     /**
      * Perform document index operation on the engine
      * @param index operation to perform
@@ -1271,7 +1279,8 @@ public abstract static class Operation {
         public enum TYPE {
             INDEX,
             DELETE,
-            NO_OP;
+            NO_OP,
+            TX_OP;
 
             private final String lowercase;
 
@@ -1363,6 +1372,7 @@ public static class Index extends Operation {
         private final boolean isRetry;
         private final long ifSeqNo;
         private final long ifPrimaryTerm;
+        private final long transactionId;
 
         public Index(
             Term uid,
@@ -1377,6 +1387,38 @@ public Index(
             boolean isRetry,
             long ifSeqNo,
             long ifPrimaryTerm
+        ) {
+            this(
+                uid,
+                doc,
+                seqNo,
+                primaryTerm,
+                version,
+                versionType,
+                origin,
+                startTime,
+                autoGeneratedIdTimestamp,
+                isRetry,
+                ifSeqNo,
+                ifPrimaryTerm,
+                -1L
+            );
+        }
+
+        public Index(
+            Term uid,
+            ParsedDocument doc,
+            long seqNo,
+            long primaryTerm,
+            long version,
+            VersionType versionType,
+            Origin origin,
+            long startTime,
+            long autoGeneratedIdTimestamp,
+            boolean isRetry,
+            long ifSeqNo,
+            long ifPrimaryTerm,
+            long transactionId
         ) {
             super(uid, seqNo, primaryTerm, version, versionType, origin, startTime);
             assert (origin == Origin.PRIMARY) == (versionType != null) : "invalid version_type=" + versionType + " for origin=" + origin;
@@ -1389,6 +1431,7 @@ public Index(
             this.autoGeneratedIdTimestamp = autoGeneratedIdTimestamp;
             this.ifSeqNo = ifSeqNo;
             this.ifPrimaryTerm = ifPrimaryTerm;
+            this.transactionId = transactionId;
         }
 
         public Index(Term uid, long primaryTerm, ParsedDocument doc) {
@@ -1467,13 +1510,18 @@ public long getIfSeqNo() {
         public long getIfPrimaryTerm() {
             return ifPrimaryTerm;
         }
+
+        public long getTransactionId() {
+            return transactionId;
+        }
     }
 
     public static class Delete extends Operation {
 
         private final String id;
         private final long ifSeqNo;
         private final long ifPrimaryTerm;
+        private final long transactionId;
 
         public Delete(
             String id,
@@ -1486,6 +1534,22 @@ public Delete(
             long startTime,
             long ifSeqNo,
             long ifPrimaryTerm
+        ) {
+            this(id, uid, seqNo, primaryTerm, version, versionType, origin, startTime, ifSeqNo, ifPrimaryTerm, -1L);
+        }
+
+        public Delete(
+            String id,
+            Term uid,
+            long seqNo,
+            long primaryTerm,
+            long version,
+            VersionType versionType,
+            Origin origin,
+            long startTime,
+            long ifSeqNo,
+            long ifPrimaryTerm,
+            long transactionId
         ) {
             super(uid, seqNo, primaryTerm, version, versionType, origin, startTime);
             assert (origin == Origin.PRIMARY) == (versionType != null) : "invalid version_type=" + versionType + " for origin=" + origin;
@@ -1496,6 +1560,7 @@ public Delete(
             this.id = Objects.requireNonNull(id);
             this.ifSeqNo = ifSeqNo;
             this.ifPrimaryTerm = ifPrimaryTerm;
+            this.transactionId = transactionId;
         }
 
         public Delete(String id, Term uid, long primaryTerm) {
@@ -1550,6 +1615,10 @@ public long getIfSeqNo() {
         public long getIfPrimaryTerm() {
             return ifPrimaryTerm;
         }
+
+        public long getTransactionId() {
+            return transactionId;
+        }
     }
 
     public static class NoOp extends Operation {
@@ -1597,6 +1666,43 @@ public int estimatedSizeInBytes() {
 
     }
 
+    public static class TxOp extends Operation {
+        public TxOp(final long startTime) {
+            super(null, UNASSIGNED_SEQ_NO, UNASSIGNED_PRIMARY_TERM, Versions.NOT_FOUND, null, null, startTime);
+        }
+
+        @Override
+        public Term uid() {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public long version() {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public VersionType versionType() {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        String id() {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public TYPE operationType() {
+            return TYPE.TX_OP;
+        }
+
+        @Override
+        public int estimatedSizeInBytes() {
+            return 2 * Long.BYTES;
+        }
+
+    }
+
     public static class Get {
         private final boolean realtime;
         private final Term uid;