CASSANDRA-20975 Add capability to load pluggable compression service providers #4513
Conversation
| * @param options compression options of baseCompressor | ||
| * @return returns a decorated compressor, if service available, otherwise baseCompressor | ||
| */ | ||
| private static ICompressor decorateCompressor(ICompressor baseCompressor, Map<String, String> options) |
There was a problem hiding this comment.
I think this might be way simplified
private static ICompressor decorateCompressor(ICompressor baseCompressor, Map<String, String> options)
{
return Optional.ofNullable(baseCompressor)
.flatMap(c -> ServiceLoader.load(ICompressorFactory.class)
.stream()
.map(ServiceLoader.Provider::get)
.filter(factory-> c.getClass().getSimpleName()
.equals(factory.getSupportedCompressorName()))
.findFirst())
.flatMap(factory -> factory.createCompressor(options))
.map(pluginCompressor -> new CompressorDecorator(baseCompressor, pluginCompressor))
.orElse(null);
}
We can get rid of getServiceProviderFactory completely. Just return Optional<ICompressor> from factory's createCompressor. You are just catching that exception here and logging, you can do same in the implementation and return empty optional instead if not possible to instantiate.
| this.otherOptions = ImmutableMap.copyOf(otherOptions); | ||
| this.minCompressRatio = minCompressRatio; | ||
| this.maxCompressedLength = maxCompressedLength; | ||
| this.decoratedSstableCompressor = decorateCompressor(sstableCompressor, otherOptions); |
|
|
||
| public final class CompressionParams | ||
| { | ||
| private static final Logger logger = LoggerFactory.getLogger(CompressionParams.class); |
There was a problem hiding this comment.
if you log in the implementation as suggested then this might go away (with imports too)
|
|
||
| public interface ICompressorFactory | ||
| { | ||
| public static final ImmutableMap<String, String> COMPRESSOR_NAME_MAP = ImmutableMap.of( |
There was a problem hiding this comment.
this is unused and can be removed, also not correct as such, we would need to be way more robust than this in general.
There was a problem hiding this comment.
public static final redundant too. Would you mind to shed some light at the logic behind this?
We would need to use some reflection to scan classes in org.apache.cassandra.io.compress package which are not abstract and implement ICompressor otherwise we would need to update this every time we add a new compressor implementation. (you are already missing ZstdDictionaryCompressor here)
I am also not completely sure if all such compression algorithms support your hardware speedup. In that case we would need to some filter them more. Not sure, waiting on your feedback.
There was a problem hiding this comment.
Intention behind using the map was to have a simple name to represent the algorithm like, deflate, lz4 etc. and map them to concrete classes in Cassandra. Another way I can think of is make the plugin factory return concrete class names like DeflateCompressor when getSupportedCompressorName is invoked, to specify which compressor it is accelerating. Will that be alright?
| /** | ||
| * This class uses a plugin compressor to perform compress/decompress, if available, otherwise reverts to default SstableCompressor. | ||
| */ | ||
| public class CompressorDecorator extends AbstractCompressorDecorator |
There was a problem hiding this comment.
I do not think that controlling the execution flow with try/catch in below methods is a good idea. Perhaps you might decide if it is possible to use your plugin as you are getting it from the factory? This is very uncomfortable and probably also not performance friendly when we do uncompress on plugin every single time and then falling back to base compressor.
There was a problem hiding this comment.
Based on the discussion on ML it seems we will go with the fallback logic after all. In that case we need to log (in a non-spamming manner) + introduce metrics and update them on each failure.
There was a problem hiding this comment.
sounds good. I will look into adding the metrics
| .bufferType(parameters.getSstableCompressor().preferredBufferType()) | ||
| .finishOnClose(option.finishOnClose()) | ||
| .build()); | ||
| ICompressor compressor = parameters.getSstableCompressor(); |
There was a problem hiding this comment.
alignement, plus we should most probably not use it like this. We should still use getSstableCompressor() as it was before. It is implementation detail if the compressor is decorated or not.
There was a problem hiding this comment.
Agree. Reasoning for doing it this way, sstableCompressor being final it can be changed only in the createCompressor function. But if I change it there, CompressionMetadata also gets updated and stored.
Any thoughts how I could overcome this issue?
smiklosovic
left a comment
smiklosovic
left a comment
There was a problem hiding this comment.
i think we need to iron out some design issues
| if (result == null) | ||
| { | ||
| result = resolveCompressor(parameters.getSstableCompressor(), compressionDictionary); | ||
| result = resolveCompressor(parameters.getDecoratedSstableCompressor(), compressionDictionary); |
| this.otherOptions = ImmutableMap.copyOf(otherOptions); | ||
| this.minCompressRatio = minCompressRatio; | ||
| this.maxCompressedLength = maxCompressedLength; | ||
| this.decoratedSstableCompressor = decorateCompressor(sstableCompressor, otherOptions); |
There was a problem hiding this comment.
do we actually need decoratedSstableCompressor? I propose to do something like
this.sstableCompressor = maybeDecorateCompressor(sstableCompressor, otherOptions);
Then sstableCompressor would be either decorated or not.
There was a problem hiding this comment.
Ah..I will try this. I think you answered one of my questions
2 participants
Updated the code based on feedback from #4420.
CompressionParams,getDecoratedSstableCompressorso that plugin compressor does not change the compression metadata