Page History

/**
 * Paimon can allocate heap memory itself, and when the operator exits, JVM will release the allocated heap memory. 
 * But for off-heap memory, Paimon cannot release the memory itself. Engines such as Flink and Spark should allocate 
 * off-heap memory for Paimon, and release the memory when the task goes to termination.
 */ 
@Public
public final class MemorySegment {
    /* Remove this method, Paimon cannot free memory itself. */
    public static MemorySegment allocateOffHeapMemory(int size);

    /* Remote this method, Paimon cannot free memory itself. */
    public void free();
}

/**
 * Memory segment allocator will allocate segments from memory manager in Flink.
 **/
@Public
public class MemorySegmentAllocator {
    /* Memory manager in Flink. */
    private final MemoryManager memoryManager;

    /**
     * Allocates a set of memory segments for memory pool.
     *
     * @param target The list into which to put the allocated memory pages.
     * @param numberOfPages The number of pages to allocate.
     * @throws MemoryAllocationException Thrown, if this memory manager does not
     *     have the requested amount of memory pages any more.
     */
    void allocate(
        Collection<MemorySegment> target,
        int numberOfPages) throws MemoryAllocationException;
    
    /* Release the segments allocated by the allocator if the task is closed. */
    void release();
}

/* Add memory allocator related options in flink connector. */
public class FlinkConnectorOptions {
    /* Sink writer allocate segments from managed memory. */
    public static final ConfigOption<Boolean> SINK_USE_MANAGED_MEMORY =
        ConfigOptions.key("sink.use-managed-memory-allocator")
                .booleanType()
                .defaultValue(false)
                .withDescription(
                        "If true, flink sink will use managed memory for merge tree; otherwise, it will create an independent memory allocator.");

    /**
     * Weight of writer buffer in managed memory, Flink will compute the memory size for writer according 
     * to the weight, the actual memory used depends on the running environment.
     */
    public static final ConfigOption<MemorySize> SINK_MANAGED_WRITER_BUFFER_MEMORY =
        ConfigOptions.key("sink.managed.writer-buffer-memory")
                .memoryType()
                .defaultValue(MemorySize.ofMebiBytes(256))
                .withDescription("Weight of writer buffer in managed memory, Flink will compute the memory size " +
                     "for writer according to the weight, the actual memory used depends on the running environment.");
}

/* Memory allocator for flink task which allocates segments from MemoryManager. */
public class FlinkMemorySegmentAllocator implements MemoryAllocator {
    /* The flink task which will be used to allocate and release segments. */
    private final Object owner;
    /* The memory manager from flink task. */
    private final MemoryManager memoryManager;
    
    /* Allocate segments from memory manager. */
    void allocate(
            Collection<MemorySegment> target,
            int numberOfPages) throws MemoryAllocationException {
        ...;
        memoryManager.allocatePages(owner, target, numOfPages);
        ...;
    }
    
    /* Release the segments allocated by the allocator if the task is closed. */
    void release();
}