public Stream<Tuple2<K, V>> prefixSearch(String columnFamilyName, byte[] prefix) {
        Preconditions.checkArgument(!closed);
        final RocksIterator it = getRocksDB().newIterator(managedHandlesMap.get(columnFamilyName));
        Iterator<Tuple2<K, V>> conditionalIt = new ConditionalIteratorWrapper<>(it, this, prefix);
        return StreamSupport
                .stream(Spliterators.spliteratorUnknownSize(conditionalIt, 0), false)
                .onClose(() -> it.close());
    }
Perform a prefix search and return stream of key-value pairs retrieved.
@note This stream must be closed after use, otherwise it will cause a memory leak

@param columnFamilyName Column Family Name
@param prefix Prefix Key
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/collections/RocksDBDAO.java