Mongo/BSON and compression question

[moores]

I have a large document (20-30KB) where one field contains nested data making up about 80% of the data in a document. I'd like to use GZIP compression on this one field, and do it somehow with Spring MongoDB mapping.

So basically, I'd like to convert this complex field to a String, manually marshal with BSON and GZIP it, and then set the value on the document. It's a nested field that is a GZIPed chunk of BSON.

This would reduce the size of my database, preventing me from considering sharding.

Just looking at the MappingMongoConverter code a bit- maybe that can be used manually?

--Michael

[Oliver Gierke]

If the field is of a special type you can simply use a Spring Converter<YourSpecialType, String> and Converter<String, YourSpecialType>. See [0] for details. If that's not the case have a look the lifecycle callbacks in AbstractMongoEventListener, onAfterLoad(…) and onBeforeSave(…) in particular.

[0] http://static.springsource.org/sprin...cit-converters

[moores]

Thanks Oliver,
So I suppose I could implement at Converter as you specified above, and the implementation of this converter could use a private MappingMongoConverter to just do the conversion to mongo DBObject, serialize the DBObject to BSON and then compress it, and finally return something with the compressed data in a byte array.
Does that make sense?

[Oliver Gierke]

If you need to convert a property of the root object and actually let the MappingMongoConverter do the rest, you're better of with the EventListener based approach. Injecting the MappingMongoConverter into the Converter implementation effectively creates a cyclic dependency which could cause trouble during bean instance creation.

[moores]

I finally got back to this task, and wrote an onBeforeSave() listener that overwrites the DBObject
with a byte[] field called "data" that contains the encoded BSON object. I set the "_class" field to be the @TypeAlias(value=..) I am using so that queries should map back to this listener class onBeforeConvert() method.

The onBeforeConvert() method is being called and I'm decoding the "data" field back into the original BSONObject that was handed to the onBeforeSaveMethod(), calling "dbo.putAll(...)" with the unzipped/decoded doc. The problem I'm seeing is that the MappingMongoConverter no longer understands how to map my @Field annotated attributes and that is surprising. See my event listener at the bottom.

WHY would spring mongodb not know how to map the doc during a find() if it's exactly the document that got written with a save()??

2012-10-28 19:11:48,416:ERROR:SimpleAsyncTaskExecutor-1:lcs.LCSDataImporterManager Task failed
org.springframework.core.convert.ConverterNotFoundException: No converter found capable of converting from type org.bson.BasicBSONObject<?, ?> to type com.expedia.www.domain.hotelSummaryConsumer.documents.PictureSet
       at org.springframework.core.convert.support.GenericConversionService.handleConverterNotFound(GenericConversionService.java:475)

The entity looks like this:

@Document(collection="pictureSetTestDocument")
@TypeAlias(value="pictureSetTestDocument")
public class PictureSetTestDocument {

	@Id
	private String id;
	
	@Field
	private PictureSet pictureSet;


	public String getId() {
		return id;
	}

	public void setId(String id) {
		this.id = id;
	}

	public PictureSet getPictureSet() {
		return pictureSet;
	}

	public void setPictureSet(PictureSet pictureSet) {
		this.pictureSet = pictureSet;
	}
}

The PictureSet is just another complex attribute with fields:

public class PictureSet 
{

	@Field("featured")
	private PictureGroup featured = PictureGroup.emptyPictureGroup();

	@Field("gallary")
	private List<PictureGroup> gallary = new ArrayList<PictureGroup>();

	@Field("maps")
	private List<PictureGroup> maps = new ArrayList<PictureGroup>();

	@Field("panorama")
	private List<PictureGroup> panorama = new ArrayList<PictureGroup>();
.....

Here is my AbstractMongoEventListener:

public class PictureSetEntityCompressingEventListener extends AbstractMongoEventListener<PictureSetTestDocument> 
{
	private boolean doCompressData = true;
	
	private static Logger LOG = Logger.getLogger(PictureSetEntityCompressingEventListener.class);
	private static final String DATA_FIELD_NAME = "_data";
	private static final String COMPRESSED_FLAG_FIELD_NAME = "_compressed";
	private static final String CLASS_FIELD_NAME = ChangeSetPersister.CLASS_KEY;
	private static final String ID_FIELD_NAME = ChangeSetPersister.ID_KEY;
	
	@Override
	public void onBeforeSave(PictureSetTestDocument source, DBObject dbo) {
		LOG.info("onBeforeSave:" + source);
		
		BasicDBObject dbObject = (BasicDBObject) dbo;
		String id = dbObject.getString(ID_FIELD_NAME);
		String classId = dbObject.getString(CLASS_FIELD_NAME);
		LOG.info("CLASS:" + classId);
		BSONEncoder encoder = new BasicBSONEncoder();
		
		byte[] encodedObject = encoder.encode(dbo);
		LOG.info("INPUTHEADER:" + Bits.readInt(encodedObject, 0));
		GZIPOutputStream gzip = null;
		ByteArrayOutputStream baos = null;
		byte[] encodedData = null;
		try 
		{
		    baos = new ByteArrayOutputStream();   
			gzip = new GZIPOutputStream(baos);
			//LOG.info("writing data to gzip stream:" + encodedObject.length);
			gzip.write(encodedObject);
			gzip.flush();
			gzip.finish();
		    encodedData = baos.toByteArray();
		    //LOG.info("encoded data:" + encodedData.length);
		} catch (IOException ioe) 
		{
			throw new DataImporterException("Error in data compression", ioe);
		} finally 
		{
			try 
			{
				if (gzip != null) 
				{
					gzip.close();
				}
				if (baos != null) 
				{
					baos.close();
				}
			} catch (IOException e) 
			{
				LOG.error("Error cleaning up after compression", e);
			}
		}
		// Clear out the newly marshalled object and add in the compressed data field.
		dbObject.clear(); 
		dbObject.put(ID_FIELD_NAME, id);
		dbObject.put(CLASS_FIELD_NAME, classId);
		dbObject.put(COMPRESSED_FLAG_FIELD_NAME, Boolean.valueOf(doCompressData));
		dbObject.put(DATA_FIELD_NAME, encodedData);	
	}

	@Override
	public void onBeforeConvert(PictureSetTestDocument source) 
	{
		// TODO Auto-generated method stub
		
		LOG.info("onBeforeConvert:" + source);
		super.onBeforeConvert(source);
	}



	@Override
	public void onAfterLoad(DBObject dbo) 
	{
		LOG.info("onAfterLoad:" + dbo);
		BSONDecoder decoder = new BasicBSONDecoder();
	
		byte[] zipData = (byte[]) dbo.get(DATA_FIELD_NAME);
		byte[] unzippedData = null;
		LOG.info("onAfterLoad: reading object:" + zipData.length);
		
		ByteArrayInputStream bais = new ByteArrayInputStream(zipData);
		ByteArrayOutputStream baos = new ByteArrayOutputStream(zipData.length);
		GZIPInputStream gzis = null;
		try
		{
		    gzis = new GZIPInputStream(bais);
		    final byte[] buffer = new byte[512];
		    int bytesRead = 0;
		    while (bytesRead != -1) {
	            bytesRead = gzis.read(buffer, 0, buffer.length);
	            if (bytesRead != -1) {
	                baos.write(buffer, 0, bytesRead);
	            }
	        }
		    baos.flush();
		    unzippedData = baos.toByteArray();
		    
		    
		} catch (IOException e) 
		{
			throw new DataImporterCompressionException("Failed to unzip payload" , e);
		} finally
		{
			try {
			if (baos != null)
			{
				baos.close();
			}
			if (gzis != null)
			{
			    gzis.close();
			}
			if (bais != null) 
			{
				bais.close();
			}
			} catch (IOException e) 
			{
				LOG.error("Failure to clean up streams", e);
			}
		}
		LOG.info("UZIPPED SIZE:" + unzippedData.length);
		LOG.info("INPUTHEADER:" + Bits.readInt(unzippedData, 0));
		BSONObject bsonObject = decoder.readObject(unzippedData);

		BasicDBObject dbObject = (BasicDBObject) dbo;
		// Clear out the compressed data and replace it with the BSON object to be mapped.
		dbObject.clear(); 
		dbo.putAll(bsonObject);
		LOG.info("CONVERTED DATA:" + dbo.toString());
	
	}
}