Optimize API: Allow to optimize index/indices. Closes #9.

Author: kimchy

modules/benchmark/micro/src/main/java/org/elasticsearch/benchmark/index/engine/SimpleEngineBenchmark.java

@@ -154,7 +154,7 @@ public SimpleEngineBenchmark build() {
                     .add(field("content", contentItem)).build();
             engine.index(new Engine.Index(new Term("_id", sId), doc, Lucene.STANDARD_ANALYZER, "type", sId, "{ ... }"));
         }
-        engine.refresh(true);
+        engine.refresh(new Engine.Refresh(true));
         stopWatch.stop();
         System.out.println("Warmup of [" + contentItems.length + "] content items, took " + stopWatch.totalTime());
 
@@ -192,7 +192,7 @@ public void run() throws Exception {
 
         scheduledExecutorService.shutdown();
 
-        engine.refresh(true);
+        engine.refresh(new Engine.Refresh(true));
         stopWatch = new StopWatch();
         stopWatch.start();
         Engine.Searcher searcher = engine.searcher();
@@ -212,7 +212,7 @@ private class Flusher implements Runnable {
 
         @Override public void run() {
             stopWatch.start("" + ++id);
-            engine.flush();
+            engine.flush(new Engine.Flush());
             stopWatch.stop();
         }
     }
@@ -224,7 +224,7 @@ private class Refresher implements Runnable {
         @Override public synchronized void run() {
             stopWatch.start("" + ++id);
             int lastId = idGenerator.get();
-            engine.refresh(true);
+            engine.refresh(new Engine.Refresh(true));
             lastRefreshedId = lastId;
             stopWatch.stop();
         }

modules/elasticsearch/src/main/java/org/elasticsearch/action/TransportActionModule.java

@@ -34,6 +34,7 @@
 import org.elasticsearch.action.admin.indices.gateway.snapshot.TransportIndexGatewaySnapshotAction;
 import org.elasticsearch.action.admin.indices.gateway.snapshot.TransportShardGatewaySnapshotAction;
 import org.elasticsearch.action.admin.indices.mapping.create.TransportCreateMappingAction;
+import org.elasticsearch.action.admin.indices.optimize.TransportOptimizeAction;
 import org.elasticsearch.action.admin.indices.refresh.TransportRefreshAction;
 import org.elasticsearch.action.admin.indices.status.TransportIndicesStatusAction;
 import org.elasticsearch.action.count.TransportCountAction;
@@ -74,6 +75,7 @@ public class TransportActionModule extends AbstractModule {
 
         bind(TransportRefreshAction.class).asEagerSingleton();
         bind(TransportFlushAction.class).asEagerSingleton();
+        bind(TransportOptimizeAction.class).asEagerSingleton();
 
         bind(TransportIndexAction.class).asEagerSingleton();
         bind(TransportGetAction.class).asEagerSingleton();

modules/elasticsearch/src/main/java/org/elasticsearch/action/TransportActions.java

@@ -45,6 +45,7 @@ public static class Indices {
             public static final String DELETE = "indices/deleteIndex";
             public static final String FLUSH = "indices/flush";
             public static final String REFRESH = "indices/refresh";
+            public static final String OPTIMIZE = "indices/optimize";
             public static final String STATUS = "indices/status";
 
             public static class Gateway {

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/ping/broadcast/BroadcastPingRequest.java

@@ -34,10 +34,6 @@ public class BroadcastPingRequest extends BroadcastOperationRequest {
     BroadcastPingRequest() {
     }
 
-    public BroadcastPingRequest(String index) {
-        super(new String[]{index}, null);
-    }
-
     public BroadcastPingRequest(String... indices) {
         super(indices, null);
     }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/ping/replication/ReplicationPingRequest.java

@@ -27,10 +27,6 @@
  */
 public class ReplicationPingRequest extends IndicesReplicationOperationRequest {
 
-    public ReplicationPingRequest(String index) {
-        this(new String[]{index});
-    }
-
     public ReplicationPingRequest(String... indices) {
         this.indices = indices;
     }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/flush/FlushRequest.java

@@ -31,10 +31,6 @@ public class FlushRequest extends BroadcastOperationRequest {
 
     }
 
-    public FlushRequest(String index) {
-        this(new String[]{index});
-    }
-
     public FlushRequest(String... indices) {
         super(indices, null);
         // we want to do the refresh in parallel on local shards...

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/flush/TransportFlushAction.java

@@ -27,6 +27,7 @@
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.routing.GroupShardsIterator;
 import org.elasticsearch.cluster.routing.ShardRouting;
+import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -84,7 +85,7 @@ public class TransportFlushAction extends TransportBroadcastOperationAction<Flus
 
     @Override protected ShardFlushResponse shardOperation(ShardFlushRequest request) throws ElasticSearchException {
         IndexShard indexShard = indicesService.indexServiceSafe(request.index()).shardSafe(request.shardId());
-        indexShard.flush();
+        indexShard.flush(new Engine.Flush());
         return new ShardFlushResponse(request.index(), request.shardId());
     }
 

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/gateway/snapshot/GatewaySnapshotRequest.java

@@ -27,10 +27,6 @@
  */
 public class GatewaySnapshotRequest extends IndicesReplicationOperationRequest {
 
-    public GatewaySnapshotRequest(String index) {
-        this(new String[]{index});
-    }
-
     public GatewaySnapshotRequest(String... indices) {
         this.indices = indices;
     }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/optimize/OptimizeRequest.java

@@ -0,0 +1,115 @@
+/*
+ * Licensed to Elastic Search and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Elastic Search licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.action.admin.indices.optimize;
+
+import org.elasticsearch.action.support.broadcast.BroadcastOperationRequest;
+import org.elasticsearch.action.support.broadcast.BroadcastOperationThreading;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+/**
+ * A request to optimize one or more indices. In order to optimize on all the indices, pass an empty array or
+ * <tt>null</tt> for the indices.
+ *
+ * <p>{@link #waitForMerge(boolean)} allows to control if the call will block until the optimize completes and
+ * defaults to <tt>true</tt>.
+ *
+ * <p>{@link #maxNumSegments(int)} allows to control the number of segments to optimize down to. By default, will
+ * cause the optimize process to optimize down to half the configured number of segments.
+ *
+ * @author kimchy (Shay Banon)
+ */
+public class OptimizeRequest extends BroadcastOperationRequest {
+
+    private boolean waitForMerge = true;
+
+    private int maxNumSegments = -1;
+
+    /**
+     * Constructs an optimization request over one or more indices.
+     *
+     * @param indices The indices to optimize, no indices passed means all indices will be optimized.
+     */
+    public OptimizeRequest(String... indices) {
+        super(indices, null);
+        // we want to do the optimize in parallel on local shards...
+        operationThreading(BroadcastOperationThreading.THREAD_PER_SHARD);
+    }
+
+    OptimizeRequest() {
+
+    }
+
+    @Override public OptimizeRequest listenerThreaded(boolean threadedListener) {
+        super.listenerThreaded(threadedListener);
+        return this;
+    }
+
+    @Override public OptimizeRequest operationThreading(BroadcastOperationThreading operationThreading) {
+        super.operationThreading(operationThreading);
+        return this;
+    }
+
+    /**
+     * Should the call block until the optimize completes. Defaults to <tt>true</tt>.
+     */
+    public boolean waitForMerge() {
+        return waitForMerge;
+    }
+
+    /**
+     * Should the call block until the optimize completes. Defaults to <tt>true</tt>.
+     */
+    public OptimizeRequest waitForMerge(boolean waitForMerge) {
+        this.waitForMerge = waitForMerge;
+        return this;
+    }
+
+    /**
+     * Will optimize the index down to <= maxNumSegments. By default, will cause the optimize
+     * process to optimize down to half the configured number of segments.
+     */
+    public int maxNumSegments() {
+        return maxNumSegments;
+    }
+
+    /**
+     * Will optimize the index down to <= maxNumSegments. By default, will cause the optimize
+     * process to optimize down to half the configured number of segments.
+     */
+    public OptimizeRequest maxNumSegments(int maxNumSegments) {
+        this.maxNumSegments = maxNumSegments;
+        return this;
+    }
+
+    public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+        super.readFrom(in);
+        waitForMerge = in.readBoolean();
+        maxNumSegments = in.readInt();
+    }
+
+    public void writeTo(DataOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeBoolean(waitForMerge);
+        out.writeInt(maxNumSegments);
+    }
+}

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/optimize/OptimizeResponse.java

@@ -0,0 +1,48 @@
+/*
+ * Licensed to Elastic Search and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Elastic Search licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.action.admin.indices.optimize;
+
+import org.elasticsearch.action.support.broadcast.BroadcastOperationResponse;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+/**
+ * @author kimchy (Shay Banon)
+ */
+public class OptimizeResponse extends BroadcastOperationResponse {
+
+    OptimizeResponse() {
+
+    }
+
+    OptimizeResponse(int successfulShards, int failedShards) {
+        super(successfulShards, failedShards);
+    }
+
+    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+        super.readFrom(in);
+    }
+
+    @Override public void writeTo(DataOutput out) throws IOException {
+        super.writeTo(out);
+    }
+}

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/optimize/ShardOptimizeRequest.java

@@ -0,0 +1,65 @@
+/*
+ * Licensed to Elastic Search and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Elastic Search licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.action.admin.indices.optimize;
+
+import org.elasticsearch.action.support.broadcast.BroadcastShardOperationRequest;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+/**
+ * @author kimchy (Shay Banon)
+ */
+public class ShardOptimizeRequest extends BroadcastShardOperationRequest {
+
+    private boolean waitForMerge = true;
+
+    private int maxNumSegments = -1;
+
+    ShardOptimizeRequest() {
+    }
+
+    public ShardOptimizeRequest(String index, int shardId, OptimizeRequest request) {
+        super(index, shardId);
+        waitForMerge = request.waitForMerge();
+        maxNumSegments = request.maxNumSegments();
+    }
+
+    boolean waitForMerge() {
+        return waitForMerge;
+    }
+
+    int maxNumSegments() {
+        return maxNumSegments;
+    }
+
+    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+        super.readFrom(in);
+        waitForMerge = in.readBoolean();
+        maxNumSegments = in.readInt();
+    }
+
+    @Override public void writeTo(DataOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeBoolean(waitForMerge);
+        out.writeInt(maxNumSegments);
+    }
+}