private RocksDBDAO<T, R> lazyGetRocksDb() {
        if (null == rocksDb) {
            synchronized (this) {
                if (null == rocksDb) {
                    diskMapPath = String.format("%s%s%s-%s",
                            rocksDbStoragePath, File.separator, "rocksdb", UUID.randomUUID());
                    LOG.info("Open rocksdb dir in {}", diskMapPath);
                    try {
                        File diskMapPathFile = new File(diskMapPath);
                        FileIOUtils.deleteDirectory(diskMapPathFile);
                        FileIOUtils.mkdir(diskMapPathFile);
                    } catch (IOException e) {
                        LOG.warn("Open rocksdb dir occur error", e);
                        throw new RuntimeException(e);
                    }
                    rocksDb = new RocksDBDAO<>(
                            ROCKSDB_COL_FAMILY, diskMapPath, keySerializer, valueSerializer);
                    rocksDb.addColumnFamily(ROCKSDB_COL_FAMILY);
                    this.addShutDownHook();
                }
            }
        }
        return rocksDb;
    }
Perform a range search according to the prefix of the key, and the KV data of all keys matching the
prefix will be returned
@note This stream must be closed after use, otherwise it will cause a memory leak

@param keyPrefix key prefix binary data
@return
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/collections/RocksDBKVBuffer.java