oak-store-document/src/test/java/org/apache/jackrabbit/oak/plugins/document/DocumentBatchSplitTest.java - jackrabbit-oak - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF 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.apache.jackrabbit.oak.plugins.document;

 import static org.apache.jackrabbit.oak.plugins.document.Collection.NODES;
 import static org.apache.jackrabbit.oak.plugins.memory.BinaryPropertyState.binaryProperty;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;

 import java.io.IOException;
 import java.util.List;
 import java.util.Random;
 import java.util.concurrent.TimeUnit;

 import org.apache.jackrabbit.oak.api.PropertyState;
 import org.apache.jackrabbit.oak.plugins.document.VersionGarbageCollector.VersionGCStats;
 import org.apache.jackrabbit.oak.plugins.document.util.TimingDocumentStoreWrapper;
 import org.apache.jackrabbit.oak.spi.commit.CommitInfo;
 import org.apache.jackrabbit.oak.spi.commit.EmptyHook;
 import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import com.google.common.collect.Lists;

 import ch.qos.logback.classic.Level;
 import ch.qos.logback.classic.LoggerContext;

 /**
  * Check correct splitting of documents (OAK-926 & OAK-1342).
  */
 @RunWith(Parameterized.class)
 public class DocumentBatchSplitTest {

     private static final Logger LOG = LoggerFactory.getLogger(DocumentBatchSplitTest.class);

     private String createOrUpdateBatchSize;
     private boolean createOrUpdateBatchSizeIsNull;

     private DocumentStoreFixture fixture;
     protected DocumentMK mk;

     public DocumentBatchSplitTest(DocumentStoreFixture fixture) {
         this.fixture = fixture;
     }

     @Parameterized.Parameters(name="{0}")
     public static java.util.Collection<Object[]> fixtures() throws IOException {
         List<Object[]> fixtures = Lists.newArrayList();
         fixtures.add(new Object[] {new DocumentStoreFixture.MemoryFixture()});

         DocumentStoreFixture mongo = new DocumentStoreFixture.MongoFixture();
         if(mongo.isAvailable()){
             fixtures.add(new Object[] {mongo});
         }
         return fixtures;
     }

     @After
     public void tearDown() throws Exception {
         if (mk != null) {
             mk.dispose();
             mk = null;
         }
         fixture.dispose();
         // reset log level to default
         enableLevel("org", null);
     }

     @Before
     public void backupProperty() {
         createOrUpdateBatchSize = System.getProperty("oak.documentMK.createOrUpdateBatchSize");
         if (createOrUpdateBatchSize == null) {
             createOrUpdateBatchSizeIsNull = true;
         }
     }

     @After
     public void restoreProperty() {
         if (createOrUpdateBatchSize != null) {
             System.setProperty("oak.documentMK.createOrUpdateBatchSize", createOrUpdateBatchSize);
         } else if (createOrUpdateBatchSizeIsNull) {
             System.clearProperty("oak.documentMK.createOrUpdateBatchSize");
         }
     }

     @Test
     @Ignore(value = "useful for benchmarking only, long execution duration")
     public void batchSplitBenchmark() throws Exception {
         int[] batchSizes = new int[] {1,2,10,30,50,75,100,200,300,400,500,1000,2000,5000,10000};
         for (int batchSize : batchSizes) {
             batchSplitTest(batchSize, 10000);
             batchSplitTest(batchSize, 10000);
         }
     }

     @Test
     public void largeBatchSplit() throws Exception {
         batchSplitTest(200, 1000);
     }

     @Test
     public void mediumBatchSplit() throws Exception {
         batchSplitTest(50, 1000);
     }

     @Test
     public void smallBatchSplit() throws Exception {
         batchSplitTest(2, 1000);
     }

     @Test
     public void noBatchSplit() throws Exception {
         batchSplitTest(1, 1000);
     }

     /** Make sure we have a test that has log level set to DEBUG */
     @Test
     public void debugLogLevelBatchSplit() throws Exception {
         enableLevel("org", Level.DEBUG);
         batchSplitTest(50, 1000);
     }

     private void batchSplitTest(int batchSize, int splitDocCnt) throws Exception {
         LOG.info("batchSplitTest: batchSize = " + batchSize+ ", splitDocCnt = " + splitDocCnt +
                 ", fixture = " + fixture);
         // this tests wants to use CountingDocumentStore
         // plus it wants to set the batchSize
         if (mk != null) {
             mk.dispose();
             mk = null;
         }
         if (fixture.getName().equals("MongoDB")) {
             MongoUtils.dropCollections(MongoUtils.DB);
         }

         System.setProperty("oak.documentMK.createOrUpdateBatchSize", String.valueOf(batchSize));

         DocumentMK.Builder mkBuilder = new DocumentMK.Builder();
         DocumentStore delegateStore = fixture.createDocumentStore();
         TimingDocumentStoreWrapper timingStore = new TimingDocumentStoreWrapper(delegateStore);
         CountingDocumentStore store = new CountingDocumentStore(timingStore);
         mkBuilder.setDocumentStore(store);
         // disable automatic background operations
         mkBuilder.setAsyncDelay(0);
         mk = mkBuilder.open();
         DocumentNodeStore ns = mk.getNodeStore();
         assertEquals(batchSize, ns.getCreateOrUpdateBatchSize());

         NodeBuilder builder = ns.getRoot().builder();
         for(int child = 0; child < 100; child++) {
             builder.child("testchild-" + child);
         }
         ns.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);

         for(int i=0; i<2; i++) {
             builder = ns.getRoot().builder();
             for(int child = 0; child < splitDocCnt; child++) {
                 PropertyState binary = binaryProperty("prop", randomBytes(5 * 1024));
                 builder.child("testchild-" + child).setProperty(binary);
             }
             ns.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
         }
         timingStore.reset();
         store.resetCounters();
         final long start = System.currentTimeMillis();
         ns.runBackgroundUpdateOperations();
         int createOrUpdateCalls = store.getNumCreateOrUpdateCalls(NODES);
         final long remoteCallMeasurement = timingStore.getAndResetOverallTime();
         final long totalSplitDuration = (System.currentTimeMillis() - start);
         final long localSplitPart = totalSplitDuration - remoteCallMeasurement;
         LOG.info("batchSplitTest: batchSize = " + batchSize +
                 ", splitDocCnt = " + splitDocCnt +
                 ", createOrUpdateCalls = " + createOrUpdateCalls +
                 ", fixture = " + fixture.getName() +
                 ", split total ms = " + (System.currentTimeMillis() - start) +
                 " (thereof local = " + localSplitPart +
                 ", remote = " + remoteCallMeasurement + ")");
         int expected = 2 * (splitDocCnt / batchSize)       /* 2 calls per batch */
                 + 2 * Math.min(1, splitDocCnt % batchSize) /* 1 additional pair for the last batch */
                 + 1;                                       /* 1 more for backgroundWrite's update to root */
         assertTrue("batchSize = " + batchSize
                 + ", splitDocCnt = " + splitDocCnt
                 + ", expected=" + expected
                 + ", createOrUpdates=" + createOrUpdateCalls,
                 createOrUpdateCalls >= expected && createOrUpdateCalls <= expected + 2);
         VersionGarbageCollector gc = ns.getVersionGarbageCollector();

         int actualSplitDocGCCount = 0;
         long timeout = ns.getClock().getTime() + 20000;
         while(actualSplitDocGCCount < splitDocCnt && ns.getClock().getTime() < timeout) {
             VersionGCStats stats = gc.gc(1, TimeUnit.MILLISECONDS);
             actualSplitDocGCCount += stats.splitDocGCCount;
             if (actualSplitDocGCCount != splitDocCnt) {
                 LOG.info("batchSplitTest: Expected " + splitDocCnt + ", actual " + actualSplitDocGCCount);
                 // advance time a bit to ensure gc does clean up the split docs
                 ns.getClock().waitUntil(ns.getClock().getTime() + 1000);
                 ns.runBackgroundUpdateOperations();
             }
         }

         // make sure those splitDocCnt split docs are deleted
         assertTrue("gc not as expected: expected " + splitDocCnt
                 + ", got " + actualSplitDocGCCount, splitDocCnt <= actualSplitDocGCCount);

         mk.dispose();
         mk = null;
     }

     private byte[] randomBytes(int num) {
         Random random = new Random(42);
         byte[] data = new byte[num];
         random.nextBytes(data);
         return data;
     }

     // TODO: from DocumentStoreStatsTest
     // but there are various places such as RevisionsCommand, BroadcastTest that have similar code.
     // we might want to move this to a new common util/helper
     private static void enableLevel(String logName, Level level){
         ((LoggerContext)LoggerFactory.getILoggerFactory())
                 .getLogger(logName).setLevel(level);
     }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF 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.apache.jackrabbit.oak.plugins.document;

	import static org.apache.jackrabbit.oak.plugins.document.Collection.NODES;
	import static org.apache.jackrabbit.oak.plugins.memory.BinaryPropertyState.binaryProperty;
	import static org.junit.Assert.assertEquals;
	import static org.junit.Assert.assertTrue;

	import java.io.IOException;
	import java.util.List;
	import java.util.Random;
	import java.util.concurrent.TimeUnit;

	import org.apache.jackrabbit.oak.api.PropertyState;
	import org.apache.jackrabbit.oak.plugins.document.VersionGarbageCollector.VersionGCStats;
	import org.apache.jackrabbit.oak.plugins.document.util.TimingDocumentStoreWrapper;
	import org.apache.jackrabbit.oak.spi.commit.CommitInfo;
	import org.apache.jackrabbit.oak.spi.commit.EmptyHook;
	import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
	import org.junit.After;
	import org.junit.Before;
	import org.junit.Ignore;
	import org.junit.Test;
	import org.junit.runner.RunWith;
	import org.junit.runners.Parameterized;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import com.google.common.collect.Lists;

	import ch.qos.logback.classic.Level;
	import ch.qos.logback.classic.LoggerContext;

	/**
	* Check correct splitting of documents (OAK-926 & OAK-1342).
	*/
	@RunWith(Parameterized.class)
	public class DocumentBatchSplitTest {

	private static final Logger LOG = LoggerFactory.getLogger(DocumentBatchSplitTest.class);

	private String createOrUpdateBatchSize;
	private boolean createOrUpdateBatchSizeIsNull;

	private DocumentStoreFixture fixture;
	protected DocumentMK mk;

	public DocumentBatchSplitTest(DocumentStoreFixture fixture) {
	this.fixture = fixture;
	}

	@Parameterized.Parameters(name="{0}")
	public static java.util.Collection<Object[]> fixtures() throws IOException {
	List<Object[]> fixtures = Lists.newArrayList();
	fixtures.add(new Object[] {new DocumentStoreFixture.MemoryFixture()});

	DocumentStoreFixture mongo = new DocumentStoreFixture.MongoFixture();
	if(mongo.isAvailable()){
	fixtures.add(new Object[] {mongo});
	}
	return fixtures;
	}

	@After
	public void tearDown() throws Exception {
	if (mk != null) {
	mk.dispose();
	mk = null;
	}
	fixture.dispose();
	// reset log level to default
	enableLevel("org", null);
	}

	@Before
	public void backupProperty() {
	createOrUpdateBatchSize = System.getProperty("oak.documentMK.createOrUpdateBatchSize");
	if (createOrUpdateBatchSize == null) {
	createOrUpdateBatchSizeIsNull = true;
	}
	}

	@After
	public void restoreProperty() {
	if (createOrUpdateBatchSize != null) {
	System.setProperty("oak.documentMK.createOrUpdateBatchSize", createOrUpdateBatchSize);
	} else if (createOrUpdateBatchSizeIsNull) {
	System.clearProperty("oak.documentMK.createOrUpdateBatchSize");
	}
	}

	@Test
	@Ignore(value = "useful for benchmarking only, long execution duration")
	public void batchSplitBenchmark() throws Exception {
	int[] batchSizes = new int[] {1,2,10,30,50,75,100,200,300,400,500,1000,2000,5000,10000};
	for (int batchSize : batchSizes) {
	batchSplitTest(batchSize, 10000);
	batchSplitTest(batchSize, 10000);
	}
	}

	@Test
	public void largeBatchSplit() throws Exception {
	batchSplitTest(200, 1000);
	}

	@Test
	public void mediumBatchSplit() throws Exception {
	batchSplitTest(50, 1000);
	}

	@Test
	public void smallBatchSplit() throws Exception {
	batchSplitTest(2, 1000);
	}

	@Test
	public void noBatchSplit() throws Exception {
	batchSplitTest(1, 1000);
	}

	/** Make sure we have a test that has log level set to DEBUG */
	@Test
	public void debugLogLevelBatchSplit() throws Exception {
	enableLevel("org", Level.DEBUG);
	batchSplitTest(50, 1000);
	}

	private void batchSplitTest(int batchSize, int splitDocCnt) throws Exception {
	LOG.info("batchSplitTest: batchSize = " + batchSize+ ", splitDocCnt = " + splitDocCnt +
	", fixture = " + fixture);
	// this tests wants to use CountingDocumentStore
	// plus it wants to set the batchSize
	if (mk != null) {
	mk.dispose();
	mk = null;
	}
	if (fixture.getName().equals("MongoDB")) {
	MongoUtils.dropCollections(MongoUtils.DB);
	}

	System.setProperty("oak.documentMK.createOrUpdateBatchSize", String.valueOf(batchSize));

	DocumentMK.Builder mkBuilder = new DocumentMK.Builder();
	DocumentStore delegateStore = fixture.createDocumentStore();
	TimingDocumentStoreWrapper timingStore = new TimingDocumentStoreWrapper(delegateStore);
	CountingDocumentStore store = new CountingDocumentStore(timingStore);
	mkBuilder.setDocumentStore(store);
	// disable automatic background operations
	mkBuilder.setAsyncDelay(0);
	mk = mkBuilder.open();
	DocumentNodeStore ns = mk.getNodeStore();
	assertEquals(batchSize, ns.getCreateOrUpdateBatchSize());

	NodeBuilder builder = ns.getRoot().builder();
	for(int child = 0; child < 100; child++) {
	builder.child("testchild-" + child);
	}
	ns.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);

	for(int i=0; i<2; i++) {
	builder = ns.getRoot().builder();
	for(int child = 0; child < splitDocCnt; child++) {
	PropertyState binary = binaryProperty("prop", randomBytes(5 * 1024));
	builder.child("testchild-" + child).setProperty(binary);
	}
	ns.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
	}
	timingStore.reset();
	store.resetCounters();
	final long start = System.currentTimeMillis();
	ns.runBackgroundUpdateOperations();
	int createOrUpdateCalls = store.getNumCreateOrUpdateCalls(NODES);
	final long remoteCallMeasurement = timingStore.getAndResetOverallTime();
	final long totalSplitDuration = (System.currentTimeMillis() - start);
	final long localSplitPart = totalSplitDuration - remoteCallMeasurement;
	LOG.info("batchSplitTest: batchSize = " + batchSize +
	", splitDocCnt = " + splitDocCnt +
	", createOrUpdateCalls = " + createOrUpdateCalls +
	", fixture = " + fixture.getName() +
	", split total ms = " + (System.currentTimeMillis() - start) +
	" (thereof local = " + localSplitPart +
	", remote = " + remoteCallMeasurement + ")");
	int expected = 2 * (splitDocCnt / batchSize) /* 2 calls per batch */
	+ 2 * Math.min(1, splitDocCnt % batchSize) /* 1 additional pair for the last batch */
	+ 1; /* 1 more for backgroundWrite's update to root */
	assertTrue("batchSize = " + batchSize
	+ ", splitDocCnt = " + splitDocCnt
	+ ", expected=" + expected
	+ ", createOrUpdates=" + createOrUpdateCalls,
	createOrUpdateCalls >= expected && createOrUpdateCalls <= expected + 2);
	VersionGarbageCollector gc = ns.getVersionGarbageCollector();

	int actualSplitDocGCCount = 0;
	long timeout = ns.getClock().getTime() + 20000;
	while(actualSplitDocGCCount < splitDocCnt && ns.getClock().getTime() < timeout) {
	VersionGCStats stats = gc.gc(1, TimeUnit.MILLISECONDS);
	actualSplitDocGCCount += stats.splitDocGCCount;
	if (actualSplitDocGCCount != splitDocCnt) {
	LOG.info("batchSplitTest: Expected " + splitDocCnt + ", actual " + actualSplitDocGCCount);
	// advance time a bit to ensure gc does clean up the split docs
	ns.getClock().waitUntil(ns.getClock().getTime() + 1000);
	ns.runBackgroundUpdateOperations();
	}
	}

	// make sure those splitDocCnt split docs are deleted
	assertTrue("gc not as expected: expected " + splitDocCnt
	+ ", got " + actualSplitDocGCCount, splitDocCnt <= actualSplitDocGCCount);

	mk.dispose();
	mk = null;
	}

	private byte[] randomBytes(int num) {
	Random random = new Random(42);
	byte[] data = new byte[num];
	random.nextBytes(data);
	return data;
	}

	// TODO: from DocumentStoreStatsTest
	// but there are various places such as RevisionsCommand, BroadcastTest that have similar code.
	// we might want to move this to a new common util/helper
	private static void enableLevel(String logName, Level level){
	((LoggerContext)LoggerFactory.getILoggerFactory())
	.getLogger(logName).setLevel(level);
	}
	}