TidesDB.java

/**
 *
 * Copyright (C) TidesDB
 *
 * Original Author: Alex Gaetano Padula
 *
 * Licensed under the Mozilla Public License, v. 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
 *
 *     https://www.mozilla.org/en-US/MPL/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 com.tidesdb;

import java.io.Closeable;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

/**
 * TidesDB is the main database class providing access to TidesDB functionality.
 * This class wraps the native TidesDB library through JNI.
 */
public class TidesDB implements Closeable {
    
    static {
        NativeLibrary.load();
    }
    
    private long nativeHandle;
    private boolean closed = false;
    private final Set<ColumnFamily> hookBearingCFs = new HashSet<>();
    
    private TidesDB(long nativeHandle) {
        this.nativeHandle = nativeHandle;
    }
    
    /**
     * Opens a TidesDB instance with the given configuration.
     *
     * @param config the database configuration
     * @return a new TidesDB instance
     * @throws TidesDBException if the database cannot be opened
     */
    public static TidesDB open(Config config) throws TidesDBException {
        if (config == null) {
            throw new IllegalArgumentException("Config cannot be null");
        }
        if (config.getDbPath() == null || config.getDbPath().isEmpty()) {
            throw new IllegalArgumentException("Database path cannot be null or empty");
        }
        
        ObjectStoreConfig osc = config.getObjectStoreConfig();

        // Build an S3 connector up front, if requested. The native handle is owned by the
        // database after a successful open (released by close), mirroring the filesystem
        // connector path. Creation throws if the library was built without S3 support.
        S3Config s3 = config.getObjectStoreS3Config();
        long objStoreHandle = 0;
        if (s3 != null) {
            objStoreHandle = nativeObjstoreS3Create(
                s3.getEndpoint(),
                s3.getBucket(),
                s3.getPrefix(),
                s3.getAccessKey(),
                s3.getSecretKey(),
                s3.getRegion(),
                s3.isUseSsl(),
                s3.isUsePathStyle(),
                s3.getTlsCaPath(),
                s3.isTlsInsecureSkipVerify(),
                s3.getMultipartThreshold(),
                s3.getMultipartPartSize()
            );
        }

        long handle = nativeOpen(
            config.getDbPath(),
            config.getNumFlushThreads(),
            config.getNumCompactionThreads(),
            config.getLogLevel().getValue(),
            config.getBlockCacheSize(),
            config.getMaxOpenSSTables(),
            config.isLogToFile(),
            config.getLogTruncationAt(),
            config.getMaxMemoryUsage(),
            config.isUnifiedMemtable(),
            config.getUnifiedMemtableWriteBufferSize(),
            config.getUnifiedMemtableSkipListMaxLevel(),
            config.getUnifiedMemtableSkipListProbability(),
            config.getUnifiedMemtableSyncMode(),
            config.getUnifiedMemtableSyncIntervalUs(),
            config.getObjectStoreFsPath(),
            osc != null ? osc.getLocalCachePath() : null,
            osc != null ? osc.getLocalCacheMaxBytes() : 0,
            osc != null ? osc.isCacheOnRead() : true,
            osc != null ? osc.isCacheOnWrite() : true,
            osc != null ? osc.getMaxConcurrentUploads() : 4,
            osc != null ? osc.getMaxConcurrentDownloads() : 8,
            osc != null ? osc.getMultipartThreshold() : 64 * 1024 * 1024,
            osc != null ? osc.getMultipartPartSize() : 8 * 1024 * 1024,
            osc != null ? osc.isSyncManifestToObject() : true,
            osc != null ? osc.isReplicateWal() : true,
            osc != null ? osc.isWalUploadSync() : false,
            osc != null ? osc.getWalSyncThresholdBytes() : 1048576,
            osc != null ? osc.isWalSyncOnCommit() : false,
            osc != null ? osc.isReplicaMode() : false,
            osc != null ? osc.getReplicaSyncIntervalUs() : 5000000,
            osc != null ? osc.isReplicaReplayWal() : true,
            config.getMaxConcurrentFlushes(),
            config.isFinishCompactionsOnClose(),
            objStoreHandle
        );

        return new TidesDB(handle);
    }

    /**
     * Reports whether the native TidesDB library was built with S3 object store support
     * ({@code TIDESDB_WITH_S3=ON}). When false, configuring a {@link S3Config} and opening the
     * database throws a {@link TidesDBException}.
     *
     * @return true if an S3-compatible connector can be created, false otherwise
     */
    public static boolean isS3Available() {
        return nativeS3Available();
    }
    
    /**
     * Closes the database instance and releases all resources.
     */
    @Override
    public void close() {
        if (!closed && nativeHandle != 0) {
            closed = true;

            // Drain all registered hooks.  Loop because a concurrent setCommitHook
            // may have passed checkOwnerOpen before closed=true but hasn't
            // registered yet; its registration is rejected (closed is true), but a
            // retry loop guarantees the set is empty before calling nativeClose.
            while (true) {
                List<ColumnFamily> batch;
                synchronized (hookBearingCFs) {
                    if (hookBearingCFs.isEmpty()) break;
                    batch = new ArrayList<>(hookBearingCFs);
                    hookBearingCFs.clear();
                }
                for (ColumnFamily cf : batch) {
                    cf.clearHookOnClose();
                }
            }

            nativeClose(nativeHandle);
            nativeHandle = 0;
        }
    }
    
    /**
     * Creates a new column family with the given configuration.
     *
     * @param name the column family name
     * @param config the column family configuration
     * @throws TidesDBException if the column family cannot be created
     */
    public void createColumnFamily(String name, ColumnFamilyConfig config) throws TidesDBException {
        checkNotClosed();
        if (name == null || name.isEmpty()) {
            throw new IllegalArgumentException("Column family name cannot be null or empty");
        }
        if (config == null) {
            throw new IllegalArgumentException("Column family config cannot be null");
        }
        
        nativeCreateColumnFamily(nativeHandle, name,
            config.getWriteBufferSize(),
            config.getLevelSizeRatio(),
            config.getMinLevels(),
            config.getDividingLevelOffset(),
            config.getKlogValueThreshold(),
            config.getCompressionAlgorithm().getValue(),
            config.isEnableBloomFilter(),
            config.getBloomFPR(),
            config.isEnableBlockIndexes(),
            config.getIndexSampleRatio(),
            config.getBlockIndexPrefixLen(),
            config.getSyncMode().getValue(),
            config.getSyncIntervalUs(),
            config.getComparatorName(),
            config.getSkipListMaxLevel(),
            config.getSkipListProbability(),
            config.getDefaultIsolationLevel().getValue(),
            config.getMinDiskSpace(),
            config.getL1FileCountTrigger(),
            config.getL0QueueStallThreshold(),
            config.getTombstoneDensityTrigger(),
            config.getTombstoneDensityMinEntries(),
            config.isUseBtree(),
            config.isObjectLazyCompaction(),
            config.isObjectPrefetchCompaction()
        );
    }
    
    /**
     * Drops a column family and all associated data.
     *
     * @param name the column family name
     * @throws TidesDBException if the column family cannot be dropped
     */
    public void dropColumnFamily(String name) throws TidesDBException {
        checkNotClosed();
        if (name == null || name.isEmpty()) {
            throw new IllegalArgumentException("Column family name cannot be null or empty");
        }
        nativeDropColumnFamily(nativeHandle, name);
    }
    
    /**
     * Retrieves a column family by name.
     *
     * @param name the column family name
     * @return the column family
     * @throws TidesDBException if the column family is not found
     */
    public ColumnFamily getColumnFamily(String name) throws TidesDBException {
        checkNotClosed();
        if (name == null || name.isEmpty()) {
            throw new IllegalArgumentException("Column family name cannot be null or empty");
        }
        long cfHandle = nativeGetColumnFamily(nativeHandle, name);
        return new ColumnFamily(cfHandle, name, this);
    }
    
    /**
     * Lists all column families in the database.
     *
     * @return array of column family names
     * @throws TidesDBException if the list cannot be retrieved
     */
    public String[] listColumnFamilies() throws TidesDBException {
        checkNotClosed();
        return nativeListColumnFamilies(nativeHandle);
    }
    
    /**
     * Begins a new transaction with default isolation level.
     *
     * @return a new transaction
     * @throws TidesDBException if the transaction cannot be started
     */
    public Transaction beginTransaction() throws TidesDBException {
        checkNotClosed();
        long txnHandle = nativeBeginTransaction(nativeHandle);
        return new Transaction(txnHandle);
    }
    
    /**
     * Begins a new transaction with the specified isolation level.
     *
     * @param isolationLevel the isolation level
     * @return a new transaction
     * @throws TidesDBException if the transaction cannot be started
     */
    public Transaction beginTransaction(IsolationLevel isolationLevel) throws TidesDBException {
        checkNotClosed();
        if (isolationLevel == null) {
            throw new IllegalArgumentException("Isolation level cannot be null");
        }
        long txnHandle = nativeBeginTransactionWithIsolation(nativeHandle, isolationLevel.getValue());
        return new Transaction(txnHandle);
    }
    
    /**
     * Retrieves statistics about the block cache.
     *
     * @return cache statistics
     * @throws TidesDBException if the stats cannot be retrieved
     */
    public CacheStats getCacheStats() throws TidesDBException {
        checkNotClosed();
        return nativeGetCacheStats(nativeHandle);
    }
    
    /**
     * Registers a custom comparator with the database.
     *
     * @param name the comparator name
     * @param context optional context string
     * @throws TidesDBException if the comparator cannot be registered
     */
    public void registerComparator(String name, String context) throws TidesDBException {
        checkNotClosed();
        if (name == null || name.isEmpty()) {
            throw new IllegalArgumentException("Comparator name cannot be null or empty");
        }
        nativeRegisterComparator(nativeHandle, name, context);
    }
    
    /**
     * Creates an on-disk snapshot of the database without blocking normal reads/writes.
     *
     * @param dir the backup directory (must be non-existent or empty)
     * @throws TidesDBException if the backup fails
     */
    public void backup(String dir) throws TidesDBException {
        checkNotClosed();
        if (dir == null || dir.isEmpty()) {
            throw new IllegalArgumentException("Backup directory cannot be null or empty");
        }
        nativeBackup(nativeHandle, dir);
    }
    
    /**
     * Creates a lightweight, near-instant snapshot of an open database using hard links
     * instead of copying SSTable data.
     *
     * @param dir the checkpoint directory (must be non-existent or empty)
     * @throws TidesDBException if the checkpoint fails
     */
    public void checkpoint(String dir) throws TidesDBException {
        checkNotClosed();
        if (dir == null || dir.isEmpty()) {
            throw new IllegalArgumentException("Checkpoint directory cannot be null or empty");
        }
        nativeCheckpoint(nativeHandle, dir);
    }
    
    /**
     * Atomically renames a column family and its underlying directory.
     * The operation waits for any in-progress flush or compaction to complete before renaming.
     *
     * @param oldName the current column family name
     * @param newName the new column family name
     * @throws TidesDBException if the rename fails
     */
    public void renameColumnFamily(String oldName, String newName) throws TidesDBException {
        checkNotClosed();
        if (oldName == null || oldName.isEmpty()) {
            throw new IllegalArgumentException("Old column family name cannot be null or empty");
        }
        if (newName == null || newName.isEmpty()) {
            throw new IllegalArgumentException("New column family name cannot be null or empty");
        }
        nativeRenameColumnFamily(nativeHandle, oldName, newName);
    }
    
    /**
     * Creates a complete copy of an existing column family with a new name.
     * The clone contains all the data from the source at the time of cloning.
     * The clone is completely independent - modifications to one do not affect the other.
     *
     * @param sourceName the source column family name
     * @param destName the destination column family name
     * @throws TidesDBException if the clone fails
     */
    public void cloneColumnFamily(String sourceName, String destName) throws TidesDBException {
        checkNotClosed();
        if (sourceName == null || sourceName.isEmpty()) {
            throw new IllegalArgumentException("Source column family name cannot be null or empty");
        }
        if (destName == null || destName.isEmpty()) {
            throw new IllegalArgumentException("Destination column family name cannot be null or empty");
        }
        nativeCloneColumnFamily(nativeHandle, sourceName, destName);
    }
    
    /**
     * Forces a synchronous flush and aggressive compaction for all column families,
     * then drains both the global flush and compaction queues.
     * This blocks until all work is complete.
     *
     * @throws TidesDBException if the purge fails
     */
    public void purge() throws TidesDBException {
        checkNotClosed();
        nativePurge(nativeHandle);
    }
    
    /**
     * Deletes a column family using its handle.
     * This is an alternative to {@link #dropColumnFamily(String)} that takes a column family
     * object instead of a name.
     *
     * @param cf the column family to delete
     * @throws TidesDBException if the column family cannot be deleted
     */
    public void deleteColumnFamily(ColumnFamily cf) throws TidesDBException {
        checkNotClosed();
        if (cf == null) {
            throw new IllegalArgumentException("Column family cannot be null");
        }
        nativeDeleteColumnFamily(nativeHandle, cf.getNativeHandle());
    }

    /**
     * Switches a read-only replica database to primary mode.
     *
     * @throws TidesDBException if not in replica mode or promotion fails
     */
    public void promoteToPrimary() throws TidesDBException {
        checkNotClosed();
        nativePromoteToPrimary(nativeHandle);
    }

    /**
     * Retrieves aggregate statistics across the entire database instance.
     *
     * @return database-level statistics
     * @throws TidesDBException if the stats cannot be retrieved
     */
    public DbStats getDbStats() throws TidesDBException {
        checkNotClosed();
        return nativeGetDbStats(nativeHandle);
    }

    /**
     * Cancels background compaction database-wide. In-flight merges bail safely at their
     * next checkpoint (their uncommitted output is discarded, inputs are left intact, so no
     * data is lost) and any queued compaction is skipped. Flushes are unaffected, so
     * durability is preserved. Blocks (bounded) until compaction is idle.
     *
     * <p>The cancellation is sticky for the session and is reset on the next open. It is
     * intended to be called immediately before {@link #close()} for a fast shutdown.</p>
     *
     * @throws TidesDBException if the operation fails
     */
    public void cancelBackgroundWork() throws TidesDBException {
        checkNotClosed();
        nativeCancelBackgroundWork(nativeHandle);
    }

    /**
     * Raises this process's open-file ceiling toward {@code desired} descriptors so a database
     * can keep more SSTables open. The engine sizes {@code maxOpenSSTables} to fit this at open
     * time, so call it <strong>before</strong> {@link #open(Config)}. This is an explicit, opt-in
     * operator action; TidesDB never raises the limit itself.
     *
     * <p>On POSIX systems (Linux, macOS, the BSDs, illumos) this raises the {@code RLIMIT_NOFILE}
     * soft limit toward the hard limit; on Windows it raises the CRT stdio cap (max 8192). A
     * failed or partial raise is non-fatal.</p>
     *
     * @param desired target descriptor count; values &le; 0 just report the current ceiling
     * @return the open-file ceiling in effect after the attempt
     */
    public static long raiseOpenFileLimit(long desired) {
        return nativeRaiseOpenFileLimit(desired);
    }

    private void checkNotClosed() {
        if (closed) {
            throw new IllegalStateException("TidesDB instance is closed");
        }
    }
    
    /**
     * Reports whether this database instance has been closed.
     */
    boolean isClosed() {
        return closed;
    }
    
    /**
     * Registers a column family that has an installed commit hook.
     */
    void registerHookColumnFamily(ColumnFamily cf) {
        synchronized (hookBearingCFs) {
            if (!closed) hookBearingCFs.add(cf);
        }
    }
    
    /**
     * Unregisters a column family whose commit hook has been cleared.
     */
    void unregisterHookColumnFamily(ColumnFamily cf) {
        synchronized (hookBearingCFs) {
            hookBearingCFs.remove(cf);
        }
    }
    
    long getNativeHandle() {
        return nativeHandle;
    }
    
    private static native long nativeOpen(String dbPath, int numFlushThreads, int numCompactionThreads,
                                          int logLevel, long blockCacheSize, long maxOpenSSTables,
                                          boolean logToFile, long logTruncationAt,
                                          long maxMemoryUsage, boolean unifiedMemtable,
                                          long unifiedMemtableWriteBufferSize,
                                          int unifiedMemtableSkipListMaxLevel,
                                          float unifiedMemtableSkipListProbability,
                                          int unifiedMemtableSyncMode,
                                          long unifiedMemtableSyncIntervalUs,
                                          String objectStoreFsPath,
                                          String oscLocalCachePath, long oscLocalCacheMaxBytes,
                                          boolean oscCacheOnRead, boolean oscCacheOnWrite,
                                          int oscMaxConcurrentUploads, int oscMaxConcurrentDownloads,
                                          long oscMultipartThreshold, long oscMultipartPartSize,
                                          boolean oscSyncManifestToObject, boolean oscReplicateWal,
                                          boolean oscWalUploadSync, long oscWalSyncThresholdBytes,
                                          boolean oscWalSyncOnCommit, boolean oscReplicaMode,
                                          long oscReplicaSyncIntervalUs,
                                          boolean oscReplicaReplayWal,
                                          int maxConcurrentFlushes,
                                          boolean finishCompactionsOnClose,
                                          long objStoreHandle) throws TidesDBException;

    private static native long nativeObjstoreS3Create(String endpoint, String bucket, String prefix,
                                                      String accessKey, String secretKey, String region,
                                                      boolean useSsl, boolean usePathStyle,
                                                      String tlsCaPath, boolean tlsInsecureSkipVerify,
                                                      long multipartThreshold, long multipartPartSize)
                                                      throws TidesDBException;

    private static native boolean nativeS3Available();

    private static native void nativeClose(long handle);
    
    private static native void nativeCreateColumnFamily(long handle, String name,
        long writeBufferSize, long levelSizeRatio, int minLevels, int dividingLevelOffset,
        long klogValueThreshold, int compressionAlgorithm, boolean enableBloomFilter,
        double bloomFPR, boolean enableBlockIndexes, int indexSampleRatio, int blockIndexPrefixLen,
        int syncMode, long syncIntervalUs, String comparatorName, int skipListMaxLevel,
        float skipListProbability, int defaultIsolationLevel, long minDiskSpace,
        int l1FileCountTrigger, int l0QueueStallThreshold,
        double tombstoneDensityTrigger, long tombstoneDensityMinEntries,
        boolean useBtree,
        boolean objectLazyCompaction,
        boolean objectPrefetchCompaction) throws TidesDBException;
    
    private static native void nativeDropColumnFamily(long handle, String name) throws TidesDBException;
    
    private static native long nativeGetColumnFamily(long handle, String name) throws TidesDBException;
    
    private static native String[] nativeListColumnFamilies(long handle) throws TidesDBException;
    
    private static native long nativeBeginTransaction(long handle) throws TidesDBException;
    
    private static native long nativeBeginTransactionWithIsolation(long handle, int isolationLevel) throws TidesDBException;
    
    private static native CacheStats nativeGetCacheStats(long handle) throws TidesDBException;
    
    private static native void nativeRegisterComparator(long handle, String name, String context) throws TidesDBException;
    
    private static native void nativeBackup(long handle, String dir) throws TidesDBException;
    
    private static native void nativeCheckpoint(long handle, String dir) throws TidesDBException;
    
    private static native void nativeRenameColumnFamily(long handle, String oldName, String newName) throws TidesDBException;
    
    private static native void nativeCloneColumnFamily(long handle, String sourceName, String destName) throws TidesDBException;
    
    private static native void nativePurge(long handle) throws TidesDBException;

    private static native DbStats nativeGetDbStats(long handle) throws TidesDBException;

    private static native void nativeDeleteColumnFamily(long handle, long cfHandle) throws TidesDBException;

    private static native void nativePromoteToPrimary(long handle) throws TidesDBException;

    private static native void nativeCancelBackgroundWork(long handle) throws TidesDBException;

    private static native long nativeRaiseOpenFileLimit(long desired);
}