Author: nfilotto Date: 2010-12-30 06:29:48 -0500 (Thu, 30 Dec 2010) New Revision: 3757 Added: jcr/branches/1.14-IMPR/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/ChangesHolder.java jcr/branches/1.14-IMPR/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/impl/core/query/lucene/TestChangesHolder.java Modified: jcr/branches/1.14-IMPR/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/SearchIndex.java Log: EXOJCR-1080: Serializing and Deserializing have been improved Added: jcr/branches/1.14-IMPR/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/ChangesHolder.java =================================================================== --- jcr/branches/1.14-IMPR/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/ChangesHolder.java (rev 0) +++ jcr/branches/1.14-IMPR/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/ChangesHolder.java 2010-12-30 11:29:48 UTC (rev 3757) @@ -0,0 +1,377 @@ +/* + * Copyright (C) 2010 eXo Platform SAS. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.exoplatform.services.jcr.impl.core.query.lucene; + +import org.apache.lucene.analysis.TokenStream; +import org.apache.lucene.document.Document; +import org.apache.lucene.document.Field; + +import java.io.Externalizable; +import java.io.IOException; +import java.io.ObjectInput; +import java.io.ObjectOutput; +import java.util.ArrayList; +import java.util.Collection; +import java.util.LinkedList; +import java.util.List; + +/** + * This class is used to serialize and deserialize the changes to apply to the lucene index. + * + * @author <a href="mailto:nicolas.filotto@exoplatform.com">Nicolas Filotto</a> + * @version $Id$ + */ +public class ChangesHolder implements Externalizable +{ + + private static final int STORED_FLAG = 1; + + private static final int COMPRESSED_FLAG = 1 << 1; + + private static final int INDEXED_FLAG = 1 << 2; + + private static final int TOKENIZED_FLAG = 1 << 3; + + private static final int OMIT_NORMS_FLAG = 1 << 4; + + private static final int BINARY_FLAG = 1 << 5; + + private static final int STORE_TERM_VECTOR_FLAG = 1 << 6; + + private static final int STORE_POSITION_WITH_TERM_VECTOR_FLAG = 1 << 7; + + private static final int STORE_OFFSET_WITH_TERM_VECTOR_FLAG = 1 << 8; + + private static final int LAZY_FLAG = 1 << 9; + + private static final int OMIT_TF_FLAG = 1 << 10; + + private static final int BOOST_FLAG = 1 << 11; + + /** + * List of doc ids to remove from the index + */ + private List<String> remove; + + /** + * Collection of Lucene Documents to add to the index + */ + private Collection<Document> add; + + /** + * Default constructor used during the deserializing phase + */ + public ChangesHolder() + { + } + + /** + * @param remove Collection of doc ids to remove from the index + * @param add Lucene Documents to add to the index + */ + public ChangesHolder(Collection<String> remove, Collection<Document> add) + { + this.remove = new ArrayList<String>(remove); + this.add = add; + } + + /** + * @return the collection of doc id to remove + */ + public Collection<String> getRemove() + { + return remove; + } + + /** + * @return the collection of lucene document to add + */ + public Collection<Document> getAdd() + { + return add; + } + + /** + * {@inheritDoc} + */ + public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException + { + int length = in.readInt(); + this.remove = new ArrayList<String>(length); + for (int i = 0; i < length; i++) + { + remove.add(in.readUTF()); + } + this.add = new LinkedList<Document>(); + while (in.readBoolean()) + { + Document doc = new Document(); + doc.setBoost(in.readFloat()); + int l = in.readInt(); + for (int i = 0; i < l; i++) + { + doc.add(readField(in, doc)); + } + add.add(doc); + } + } + + /** + * Deserialize the field from the given {@link ObjectInput} + * @param in the stream from which we deserialize the Field + * @return the deserialized field + * @throws IOException + * @throws ClassNotFoundException + */ + private static Field readField(ObjectInput in, Document doc) throws IOException, ClassNotFoundException + { + String name = in.readUTF(); + int flags = in.readInt(); + float boost = (flags & BOOST_FLAG) > 0 ? in.readFloat() : 1.0f; + Object value = in.readObject(); + Field field; + if (value instanceof TokenStream) + { + field = new Field(name, (TokenStream)value); + } + else + { + // The value is a String + field = new Field(name, (String)value, getStoreParameter(flags), getIndexParameter(flags), getTermVectorParameter(flags)); + } + field.setBoost(boost); + field.setOmitNorms((flags & OMIT_NORMS_FLAG) > 0); + field.setOmitTf((flags & OMIT_TF_FLAG) > 0); + return field; + } + + /** + * Returns the index parameter extracted from the flags. + * + * @param flags the flags of the Lucene field. + * @return the index parameter corresponding to the given flags. + */ + private static Field.Index getIndexParameter(int flags) + { + if ((flags & INDEXED_FLAG) == 0) + { + return Field.Index.NO; + } + else if ((flags & TOKENIZED_FLAG) > 0) + { + return Field.Index.ANALYZED; + } + else + { + return Field.Index.NOT_ANALYZED; + } + } + + /** + * Returns the store parameter extracted from the flags. + * + * @param flags the flags of the Lucene field. + * @return the store parameter corresponding to the given flags. + */ + private static Field.Store getStoreParameter(int flags) + { + if ((flags & COMPRESSED_FLAG) > 0) + { + return Field.Store.COMPRESS; + } + else if ((flags & STORED_FLAG) > 0) + { + return Field.Store.YES; + } + else + { + return Field.Store.NO; + } + } + + /** + * Returns the term vector parameter extracted from the flags. + * + * @param flags the flags of the Lucene field. + * @return the term vector parameter corresponding to the given flags. + */ + private static Field.TermVector getTermVectorParameter(int flags) + { + if (((flags & STORE_POSITION_WITH_TERM_VECTOR_FLAG) > 0) + && ((flags & STORE_OFFSET_WITH_TERM_VECTOR_FLAG) > 0)) + { + return Field.TermVector.WITH_POSITIONS_OFFSETS; + } + else if ((flags & STORE_POSITION_WITH_TERM_VECTOR_FLAG) > 0) + { + return Field.TermVector.WITH_POSITIONS; + } + else if ((flags & STORE_OFFSET_WITH_TERM_VECTOR_FLAG) > 0) + { + return Field.TermVector.WITH_OFFSETS; + } + else if ((flags & STORE_TERM_VECTOR_FLAG) > 0) + { + return Field.TermVector.YES; + } + else + { + return Field.TermVector.NO; + } + } + + /** + * {@inheritDoc} + */ + @SuppressWarnings("unchecked") + public void writeExternal(ObjectOutput out) throws IOException + { + int length = remove.size(); + out.writeInt(length); + for (int i = 0; i < length; i++) + { + out.writeUTF(remove.get(i)); + } + out.flush(); + for (Document doc : add) + { + // Indicate that there is a doc to come + out.writeBoolean(true); + // boost + out.writeFloat(doc.getBoost()); + List<Field> fields = doc.getFields(); + int l = fields.size(); + out.writeInt(l); + for (int i = 0; i < l; i++) + { + writeField(out, fields.get(i)); + } + out.flush(); + } + // There is no doc anymore + out.writeBoolean(false); + } + + /** + * Serialize the Field into the given {@link ObjectOutput} + * @param out the stream in which we serialize the Field + * @param field the Field instance to serialize + * @throws IOException if the Field could not be serialized + */ + private static void writeField(ObjectOutput out, Field field) throws IOException + { + // Name + out.writeUTF(field.name()); + // Flags + writeFlags(out, field); + if (field.getBoost() != 1.0f) + { + // Boost + out.writeFloat(field.getBoost()); + } + // Value + writeValue(out, field); + } + + /** + * Serialize the value into the given {@link ObjectOutput} + * @param out the stream in which we serialize the value + * @param field the field from which we extract the value + * @throws IOException if the value could not be serialized + */ + private static void writeValue(ObjectOutput out, Field field) throws IOException + { + Object o = field.stringValue(); + if (o != null) + { + // Use writeObject instead of writeUTF because the value could contain unsupported + // characters + out.writeObject(o); + return; + } + o = field.tokenStreamValue(); + if (o != null) + { + out.writeObject(o); + return; + } + o = field.readerValue(); + throw new RuntimeException("Unsupported value " + o); + } + + /** + * Serialize the flags into the given {@link ObjectOutput} + * @param out the stream in which we serialize the flags + * @param field the field from which we extract the flags + * @throws IOException if the flags could not be serialized + */ + private static void writeFlags(ObjectOutput out, Field field) throws IOException + { + int flags = 0; + if (field.isStored()) + { + flags |= STORED_FLAG; + } + if (field.isCompressed()) + { + flags |= COMPRESSED_FLAG; + } + if (field.isIndexed()) + { + flags |= INDEXED_FLAG; + } + if (field.isTokenized()) + { + flags |= TOKENIZED_FLAG; + } + if (field.getOmitNorms()) + { + flags |= OMIT_NORMS_FLAG; + } + if (field.isBinary()) + { + flags |= BINARY_FLAG; + } + if (field.isTermVectorStored()) + { + flags |= STORE_TERM_VECTOR_FLAG; + } + if (field.isStorePositionWithTermVector()) + { + flags |= STORE_POSITION_WITH_TERM_VECTOR_FLAG; + } + if (field.isStoreOffsetWithTermVector()) + { + flags |= STORE_OFFSET_WITH_TERM_VECTOR_FLAG; + } + if (field.isLazy()) + { + flags |= LAZY_FLAG; + } + if (field.getOmitTf()) + { + flags |= OMIT_TF_FLAG; + } + if (field.getBoost() != 1.0f) + { + flags |= BOOST_FLAG; + } + out.writeInt(flags); + } +} Modified: jcr/branches/1.14-IMPR/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/SearchIndex.java =================================================================== --- jcr/branches/1.14-IMPR/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/SearchIndex.java 2010-12-30 11:01:59 UTC (rev 3756) +++ jcr/branches/1.14-IMPR/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/SearchIndex.java 2010-12-30 11:29:48 UTC (rev 3757) @@ -16,8 +16,6 @@ */ package org.exoplatform.services.jcr.impl.core.query.lucene; -import EDU.oswego.cs.dl.util.concurrent.ConcurrentHashMap; - import org.apache.commons.collections.IteratorUtils; import org.apache.commons.collections.Transformer; import org.apache.commons.collections.collection.TransformedCollection; @@ -64,13 +62,9 @@ import org.exoplatform.services.jcr.impl.util.SecurityHelper; import org.exoplatform.services.jcr.impl.util.io.PrivilegedFileHelper; import org.exoplatform.services.jcr.impl.util.io.PrivilegedSystemHelper; -import org.exoplatform.services.jcr.util.IdGenerator; import org.exoplatform.services.rpc.RPCException; import org.exoplatform.services.rpc.RPCService; import org.exoplatform.services.rpc.RemoteCommand; -import org.exoplatform.services.rpc.TopologyChangeEvent; -import org.exoplatform.services.rpc.TopologyChangeListener; -import org.exoplatform.services.rpc.impl.RPCServiceImpl; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.w3c.dom.Element; @@ -91,9 +85,7 @@ import java.util.Iterator; import java.util.List; import java.util.Map; -import java.util.Random; import java.util.Set; -import java.util.concurrent.atomic.AtomicInteger; import javax.jcr.RepositoryException; import javax.jcr.query.InvalidQueryException; @@ -665,9 +657,8 @@ //if (rpcService.isCoordinator()) if (modeHandler.getMode() == IndexerIoMode.READ_WRITE) { - ArrayList<String> remove = (ArrayList<String>)args[0]; - ArrayList<Document> add = (ArrayList<Document>)args[1]; - index.update(remove, add); + ChangesHolder changes = (ChangesHolder)args[0]; + index.update(changes.getRemove(), changes.getAdd()); } return null; } @@ -822,12 +813,9 @@ } else { - ArrayList<String> removeList = new ArrayList<String>(remove); - ArrayList<Document> addList = new ArrayList<Document>(add); - //log.info("sending docs to coordinator"); try { - rpcService.executeCommandOnCoordinator(sendDocs, true, removeList, addList); + rpcService.executeCommandOnCoordinator(sendDocs, true, new ChangesHolder(remove, add)); } catch (SecurityException e) { Added: jcr/branches/1.14-IMPR/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/impl/core/query/lucene/TestChangesHolder.java =================================================================== --- jcr/branches/1.14-IMPR/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/impl/core/query/lucene/TestChangesHolder.java (rev 0) +++ jcr/branches/1.14-IMPR/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/impl/core/query/lucene/TestChangesHolder.java 2010-12-30 11:29:48 UTC (rev 3757) @@ -0,0 +1,241 @@ +/* + * Copyright (C) 2010 eXo Platform SAS. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.exoplatform.services.jcr.impl.core.query.lucene; + +import junit.framework.TestCase; + +import org.apache.lucene.document.Document; +import org.apache.lucene.document.Field; +import org.apache.lucene.document.Field.Index; +import org.apache.lucene.document.Field.Store; +import org.apache.lucene.document.Field.TermVector; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.Iterator; +import java.util.List; +import java.util.UUID; + +/** + * @author <a href="mailto:nicolas.filotto@exoplatform.com">Nicolas Filotto</a> + * @version $Id$ + */ +public class TestChangesHolder extends TestCase +{ + + public void testSerNDeserializeDocs() throws Exception + { + System.out.println("### testSerNDeserializeDocs ###"); + Collection<Document> add = new ArrayList<Document>(3); + Document doc = new Document(); + doc.setBoost(2.0f); + Field fieldFull = new Field("full", "full-value", Store.COMPRESS, Index.ANALYZED_NO_NORMS, TermVector.WITH_POSITIONS_OFFSETS); + fieldFull.setBoost(2.0f); + fieldFull.setOmitTf(true); + doc.add(fieldFull); + Field fieldEmpty = new Field("empty", "empty-value", Store.NO, Index.NOT_ANALYZED, TermVector.NO); + doc.add(fieldEmpty); + add.add(doc); + doc = new Document(); + doc.add(fieldFull); + add.add(doc); + doc = new Document(); + doc.add(fieldEmpty); + add.add(doc); + + ByteArrayOutputStream baos = null; + + int total = 100000; + long start; + Collection<String> remove = Collections.emptyList(); + Collection<Document> addResult = null; + start = System.currentTimeMillis(); + for (int i = 0; i < total; i++) + { + baos = new ByteArrayOutputStream(); + ObjectOutputStream oos = new ObjectOutputStream(baos); + oos.writeObject(new ChangesHolder(remove, add)); + oos.close(); + } + System.out.println("Custom serialization: total time = " + (System.currentTimeMillis() - start) + ", size = " + baos.size()); + + start = System.currentTimeMillis(); + for (int i = 0; i < total; i++) + { + ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(baos.toByteArray())); + addResult = ((ChangesHolder)ois.readObject()).getAdd(); + ois.close(); + } + System.out.println("Custom deserialization: total time = " + (System.currentTimeMillis() - start)); + checkDocs(addResult); + start = System.currentTimeMillis(); + for (int i = 0; i < total; i++) + { + baos = new ByteArrayOutputStream(); + ObjectOutputStream oos = new ObjectOutputStream(baos); + oos.writeObject(add); + oos.close(); + } + System.out.println("Native serialization: total time = " + (System.currentTimeMillis() - start) + ", size = " + baos.size()); + start = System.currentTimeMillis(); + for (int i = 0; i < total; i++) + { + ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(baos.toByteArray())); + addResult = (Collection<Document>)ois.readObject(); + ois.close(); + } + System.out.println("Native deserialization: total time = " + (System.currentTimeMillis() - start)); + checkDocs(addResult); + } + + private void checkDocs(Collection<Document> addResult) + { + assertNotNull(addResult); + assertEquals(3, addResult.size()); + Iterator<Document> it = addResult.iterator(); + Document doc = it.next(); + assertEquals(2.0f, doc.getBoost()); + List<Field> fields = doc.getFields(); + assertNotNull(fields); + assertEquals(2, fields.size()); + checkFieldFull(fields.get(0)); + checkFieldEmpty(fields.get(1)); + doc = it.next(); + assertEquals(1.0f, doc.getBoost()); + fields = doc.getFields(); + assertNotNull(fields); + assertEquals(1, fields.size()); + checkFieldFull(fields.get(0)); + doc = it.next(); + assertEquals(1.0f, doc.getBoost()); + fields = doc.getFields(); + assertNotNull(fields); + assertEquals(1, fields.size()); + checkFieldEmpty(fields.get(0)); + } + + private void checkFieldFull(Field field) + { + assertEquals("full", field.name()); + assertEquals("full-value", field.stringValue()); + assertTrue(field.isStored()); + assertTrue(field.isCompressed()); + assertTrue(field.isIndexed()); + assertTrue(field.isTokenized()); + assertTrue(field.getOmitNorms()); + assertTrue(field.isTermVectorStored()); + assertTrue(field.isStoreOffsetWithTermVector()); + assertTrue(field.isStorePositionWithTermVector()); + assertTrue(field.getOmitTf()); + assertFalse(field.isBinary()); + assertFalse(field.isLazy()); + assertEquals(2.0f, field.getBoost()); + assertEquals(0, field.getBinaryLength()); + assertEquals(0, field.getBinaryOffset()); + } + + private void checkFieldEmpty(Field field) + { + assertEquals("empty", field.name()); + assertEquals("empty-value", field.stringValue()); + assertFalse(field.isStored()); + assertFalse(field.isCompressed()); + assertTrue(field.isIndexed()); + assertFalse(field.isTokenized()); + assertFalse(field.getOmitNorms()); + assertFalse(field.isTermVectorStored()); + assertFalse(field.isStoreOffsetWithTermVector()); + assertFalse(field.isStorePositionWithTermVector()); + assertFalse(field.getOmitTf()); + assertFalse(field.isBinary()); + assertFalse(field.isLazy()); + assertEquals(1.0f, field.getBoost()); + assertEquals(0, field.getBinaryLength()); + assertEquals(0, field.getBinaryOffset()); + } + + public void testSerNDeserializeIds() throws Exception + { + System.out.println("### testSerNDeserializeIds ###"); + Collection<String> remove = new ArrayList<String>(3); + remove.add(UUID.randomUUID().toString()); + remove.add(UUID.randomUUID().toString()); + remove.add(UUID.randomUUID().toString()); + ByteArrayOutputStream baos = null; + + int total = 100000; + long start; + Collection<Document> add = Collections.emptyList(); + Collection<String> addResult = null; + start = System.currentTimeMillis(); + for (int i = 0; i < total; i++) + { + baos = new ByteArrayOutputStream(); + ObjectOutputStream oos = new ObjectOutputStream(baos); + oos.writeObject(new ChangesHolder(remove, add)); + oos.close(); + } + System.out.println("Custom serialization: total time = " + (System.currentTimeMillis() - start) + ", size = " + baos.size()); + + start = System.currentTimeMillis(); + for (int i = 0; i < total; i++) + { + ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(baos.toByteArray())); + addResult = ((ChangesHolder)ois.readObject()).getRemove(); + ois.close(); + } + System.out.println("Custom deserialization: total time = " + (System.currentTimeMillis() - start)); + checkIds(remove, addResult); + start = System.currentTimeMillis(); + for (int i = 0; i < total; i++) + { + baos = new ByteArrayOutputStream(); + ObjectOutputStream oos = new ObjectOutputStream(baos); + oos.writeObject(remove); + oos.close(); + } + System.out.println("Native serialization: total time = " + (System.currentTimeMillis() - start) + ", size = " + baos.size()); + start = System.currentTimeMillis(); + for (int i = 0; i < total; i++) + { + ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(baos.toByteArray())); + addResult = (Collection<String>)ois.readObject(); + ois.close(); + } + System.out.println("Native deserialization: total time = " + (System.currentTimeMillis() - start)); + checkIds(remove, addResult); + } + + private void checkIds(Collection<String> remove, Collection<String> addResult) + { + assertNotNull(addResult); + assertEquals(remove.size(), addResult.size()); + Iterator<String> it1 = remove.iterator(); + Iterator<String> it2 = addResult.iterator(); + while (it1.hasNext()) + { + assertEquals(it1.next(), it2.next()); + } + } +}