Index: src/main/java/org/apache/jackrabbit/oak/plugins/segment/AbstractStore.java =================================================================== --- src/main/java/org/apache/jackrabbit/oak/plugins/segment/AbstractStore.java (revision 1559096) +++ src/main/java/org/apache/jackrabbit/oak/plugins/segment/AbstractStore.java (working copy) @@ -23,6 +23,7 @@ import java.util.UUID; import javax.annotation.Nonnull; +import javax.annotation.Nullable; import org.apache.jackrabbit.oak.cache.CacheLIRS; @@ -160,4 +161,9 @@ return type.isInstance(object) && ((Record) object).getStore() == this; } + @Nullable + @Override + public ExternalBlob readBlob(String reference) { + return null; + } } Index: src/main/java/org/apache/jackrabbit/oak/plugins/segment/ExternalBlob.java =================================================================== --- src/main/java/org/apache/jackrabbit/oak/plugins/segment/ExternalBlob.java (revision 0) +++ src/main/java/org/apache/jackrabbit/oak/plugins/segment/ExternalBlob.java (working copy) @@ -0,0 +1,34 @@ +/* + * 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.segment; + +import org.apache.jackrabbit.oak.api.Blob; + +/** + * Marks a blob that is external. + */ +public interface ExternalBlob extends Blob { + + /** + * Return a reference to this external blob. + * + * @return reference + */ + public String getReference(); +} Index: src/main/java/org/apache/jackrabbit/oak/plugins/segment/Segment.java =================================================================== --- src/main/java/org/apache/jackrabbit/oak/plugins/segment/Segment.java (revision 1559096) +++ src/main/java/org/apache/jackrabbit/oak/plugins/segment/Segment.java (working copy) @@ -438,6 +438,12 @@ } } + SegmentBlob createBlob(int offset) { + RecordId id = new RecordId(uuid, offset); + int n = readByte(offset) & 0xff; + return new SegmentBlob(this, id, (n & 0xe0) == 0xe0); + } + SegmentStream readStream(int offset) { RecordId id = new RecordId(uuid, offset); int pos = pos(offset, 1); @@ -460,6 +466,25 @@ } } + String readBlobReference(int offset) { + int pos = pos(offset, 1); + + int length = (data.get(pos++) & 0x1f) << 8 + | (data.get(pos++) & 0xff); + + byte[] bytes = new byte[length]; + ByteBuffer buffer = data.duplicate(); + buffer.position(pos + 8); // skip blob length + buffer.get(bytes); + return new String(bytes, Charsets.UTF_8); + } + + long readBlobLength(int offset) { + long high = readInt(offset + 2); + long low = readInt(offset + 6); + return high << 32 | low; + } + //------------------------------------------------------------< Object >-- @Override Index: src/main/java/org/apache/jackrabbit/oak/plugins/segment/SegmentBlob.java =================================================================== --- src/main/java/org/apache/jackrabbit/oak/plugins/segment/SegmentBlob.java (revision 1559096) +++ src/main/java/org/apache/jackrabbit/oak/plugins/segment/SegmentBlob.java (working copy) @@ -21,20 +21,34 @@ import org.apache.jackrabbit.oak.api.Blob; import org.apache.jackrabbit.oak.plugins.memory.AbstractBlob; +import java.io.InputStream; + class SegmentBlob extends Record implements Blob { - SegmentBlob(Segment segment, RecordId id) { + private boolean external; + + SegmentBlob(Segment segment, RecordId id, boolean external) { super(segment, id); + + this.external = external; } @Override @Nonnull - public SegmentStream getNewStream() { + public InputStream getNewStream() { + if (external) { + String refererence = getSegment().readBlobReference(getOffset()); + return getStore().readBlob(refererence).getNewStream(); + } return getSegment().readStream(getOffset()); } @Override public long length() { - SegmentStream stream = getNewStream(); + if (external) { + return getSegment().readBlobLength(getOffset()); + } + + SegmentStream stream = (SegmentStream) getNewStream(); try { return stream.getLength(); } finally { Index: src/main/java/org/apache/jackrabbit/oak/plugins/segment/SegmentPropertyState.java =================================================================== --- src/main/java/org/apache/jackrabbit/oak/plugins/segment/SegmentPropertyState.java (revision 1559096) +++ src/main/java/org/apache/jackrabbit/oak/plugins/segment/SegmentPropertyState.java (working copy) @@ -162,7 +162,7 @@ @SuppressWarnings("unchecked") private T getValue(Segment segment, RecordId id, Type type) { if (type == BINARY) { - return (T) new SegmentBlob(segment, id); // load binaries lazily + return (T) segment.createBlob(id.getOffset()); // load binaries lazily } String value = segment.readString(id); Index: src/main/java/org/apache/jackrabbit/oak/plugins/segment/SegmentStore.java =================================================================== --- src/main/java/org/apache/jackrabbit/oak/plugins/segment/SegmentStore.java (revision 1559096) +++ src/main/java/org/apache/jackrabbit/oak/plugins/segment/SegmentStore.java (working copy) @@ -16,6 +16,7 @@ */ package org.apache.jackrabbit.oak.plugins.segment; +import javax.annotation.Nullable; import java.util.UUID; public interface SegmentStore { @@ -51,4 +52,11 @@ */ boolean isInstance(Object object, Class type); + /** + * Read a blob from external storage. + * + * @param reference blob reference + * @return external blob + */ + ExternalBlob readBlob(String reference); } Index: src/main/java/org/apache/jackrabbit/oak/plugins/segment/SegmentWriter.java =================================================================== --- src/main/java/org/apache/jackrabbit/oak/plugins/segment/SegmentWriter.java (revision 1559096) +++ src/main/java/org/apache/jackrabbit/oak/plugins/segment/SegmentWriter.java (working copy) @@ -484,6 +484,31 @@ } /** + * Write a reference to an external blob. + * + * @param reference reference + * @param blobLength blob length + * @return record id + */ + private synchronized RecordId writeValueRecord(String reference, long blobLength) { + byte[] data = reference.getBytes(Charsets.UTF_8); + int length = data.length; + + checkArgument(length < 8192); + + RecordId id = prepare(RecordType.VALUE, 2 + 8 + length); + int len = length | 0xE000; + buffer[position++] = (byte) (len >> 8); + buffer[position++] = (byte) len; + + writeLong(blobLength); + + System.arraycopy(data, 0, buffer, position, length); + position += length; + return id; + } + + /** * Writes a block record containing the given block of bytes. * * @param bytes source buffer @@ -640,13 +665,20 @@ } public SegmentBlob writeBlob(Blob blob) throws IOException { - if (store.isInstance(blob, SegmentBlob.class)) { + if (blob instanceof ExternalBlob) { + return writeBlob((ExternalBlob) blob); + } else if (store.isInstance(blob, SegmentBlob.class)) { return (SegmentBlob) blob; } else { return writeStream(blob.getNewStream()); } } + private SegmentBlob writeBlob(ExternalBlob blob) { + RecordId id = writeValueRecord(blob.getReference(), blob.length()); + return new SegmentBlob(dummySegment, id, true); + } + /** * Writes a stream value record. The given stream is consumed * and closed by this method. @@ -666,7 +698,7 @@ Closeables.close(stream, threw); } } - return new SegmentBlob(dummySegment, id); + return new SegmentBlob(dummySegment, id, false); } private RecordId internalWriteStream(InputStream stream) Index: src/main/java/org/apache/jackrabbit/oak/plugins/segment/file/FileBlob.java =================================================================== --- src/main/java/org/apache/jackrabbit/oak/plugins/segment/file/FileBlob.java (revision 0) +++ src/main/java/org/apache/jackrabbit/oak/plugins/segment/file/FileBlob.java (working copy) @@ -0,0 +1,71 @@ +/* + * 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.segment.file; + +import org.apache.jackrabbit.oak.plugins.segment.ExternalBlob; + +import javax.annotation.Nonnull; +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; + +public class FileBlob implements ExternalBlob { + + private final String path; + + public FileBlob(String path) { + this.path = path; + } + + public String getReference() { + return path; + } + + @Nonnull + @Override + public InputStream getNewStream() { + try { + return new FileInputStream(getFile()); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + @Override + public long length() { + return getFile().length(); + } + + private File getFile() { + return new File(path); + } + + @Override + public boolean equals(Object obj) { + if (obj instanceof FileBlob) { + FileBlob other = (FileBlob) obj; + return this.path.equals(other.path); + } + return super.equals(obj); + } + + @Override + public int hashCode() { + return path.hashCode(); + } +} Index: src/main/java/org/apache/jackrabbit/oak/plugins/segment/file/FileStore.java =================================================================== --- src/main/java/org/apache/jackrabbit/oak/plugins/segment/file/FileStore.java (revision 1559096) +++ src/main/java/org/apache/jackrabbit/oak/plugins/segment/file/FileStore.java (working copy) @@ -36,11 +36,13 @@ import java.util.concurrent.atomic.AtomicReference; import javax.annotation.Nonnull; +import javax.annotation.Nullable; import org.apache.jackrabbit.oak.plugins.segment.AbstractStore; import org.apache.jackrabbit.oak.plugins.segment.Journal; import org.apache.jackrabbit.oak.plugins.segment.RecordId; import org.apache.jackrabbit.oak.plugins.segment.Segment; +import org.apache.jackrabbit.oak.plugins.segment.ExternalBlob; import org.apache.jackrabbit.oak.spi.state.NodeBuilder; import org.apache.jackrabbit.oak.spi.state.NodeState; import org.slf4j.Logger; @@ -114,7 +116,7 @@ String name = String.format(FILE_NAME_FORMAT, "bulk", i); File file = new File(directory, name); if (file.isFile()) { - bulkFiles.add(new TarFile(file, maxFileSizeMB, memoryMapping)); + bulkFiles.add(new TarFile(file, maxFileSize, memoryMapping)); } else { break; } @@ -124,7 +126,7 @@ String name = String.format(FILE_NAME_FORMAT, "data", i); File file = new File(directory, name); if (file.isFile()) { - dataFiles.add(new TarFile(file, maxFileSizeMB, memoryMapping)); + dataFiles.add(new TarFile(file, maxFileSize, memoryMapping)); } else { break; } @@ -338,4 +340,9 @@ super.deleteSegment(segmentId); } + @Nullable + @Override + public ExternalBlob readBlob(String reference) { + return new FileBlob(reference); + } } Index: src/test/java/org/apache/jackrabbit/oak/plugins/segment/ExternalBlobTest.java =================================================================== --- src/test/java/org/apache/jackrabbit/oak/plugins/segment/ExternalBlobTest.java (revision 0) +++ src/test/java/org/apache/jackrabbit/oak/plugins/segment/ExternalBlobTest.java (working copy) @@ -0,0 +1,111 @@ +/* + * 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.segment; + +import org.apache.commons.io.FileUtils; +import org.apache.commons.io.IOUtils; +import org.apache.jackrabbit.api.JackrabbitRepository; +import org.apache.jackrabbit.commons.jackrabbit.authorization.AccessControlUtils; +import org.apache.jackrabbit.oak.api.Blob; +import org.apache.jackrabbit.oak.plugins.memory.AbstractBlob; +import org.apache.jackrabbit.oak.plugins.segment.file.FileBlob; +import org.apache.jackrabbit.oak.api.PropertyState; +import org.apache.jackrabbit.oak.api.Type; +import org.apache.jackrabbit.oak.plugins.segment.SegmentNodeStore; +import org.apache.jackrabbit.oak.plugins.segment.SegmentStore; +import org.apache.jackrabbit.oak.plugins.segment.file.FileStore; +import org.apache.jackrabbit.oak.spi.commit.EmptyHook; +import org.apache.jackrabbit.oak.spi.security.principal.EveryonePrincipal; +import org.apache.jackrabbit.oak.spi.state.NodeBuilder; +import org.apache.jackrabbit.oak.spi.state.NodeState; +import org.junit.After; +import org.junit.Test; + +import javax.jcr.GuestCredentials; +import javax.jcr.Node; +import javax.jcr.Repository; +import javax.jcr.RepositoryException; +import javax.jcr.Session; +import javax.jcr.SimpleCredentials; +import javax.jcr.security.Privilege; +import java.io.ByteArrayInputStream; +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.util.Random; + +import static junit.framework.Assert.assertEquals; +import static junit.framework.Assert.assertTrue; + +public class ExternalBlobTest { + + private SegmentStore store; + private SegmentNodeStore nodeStore; + private FileBlob fileBlob; + + @Test + public void testCreateAndRead() throws Exception { + SegmentNodeStore nodeStore = getNodeStore(); + + NodeState state = nodeStore.getRoot().getChildNode("hello"); + if (!state.exists()) { + NodeBuilder builder = nodeStore.getRoot().builder(); + builder.child("hello"); + nodeStore.merge(builder, EmptyHook.INSTANCE, null); + } + + Blob blob = getFileBlob(); + NodeBuilder builder = nodeStore.getRoot().builder(); + builder.getChildNode("hello").setProperty("world", blob); + nodeStore.merge(builder, EmptyHook.INSTANCE, null); + + state = nodeStore.getRoot().getChildNode("hello"); + blob = state.getProperty("world").getValue(Type.BINARY); + + assertTrue("Blob written and read must be equal", + AbstractBlob.equal(blob, getFileBlob())); + } + + @After + public void close() { + if (store != null) { + store.close(); + } + } + + protected SegmentNodeStore getNodeStore() throws IOException { + if (nodeStore == null) { + store = new FileStore(new File("/tmp"), 256, false); + nodeStore = new SegmentNodeStore(store); + } + return nodeStore; + } + + private FileBlob getFileBlob() throws IOException { + if (fileBlob == null) { + File file = File.createTempFile("blob", "tmp"); + file.deleteOnExit(); + + byte[] data = new byte[2345]; + new Random().nextBytes(data); + FileUtils.writeByteArrayToFile(file, data); + + fileBlob = new FileBlob(file.getPath()); + } + return fileBlob; + } +}