Function bodies 231 total

public class BestSuitCodec implements Codec {

    public static final String SERIALIZE_AS_STRING = "serializeAsString";

    private final ObjectMapper objectMapper;

    public BestSuitCodec() {
        this(defaultObjectMapper());
    }

    public BestSuitCodec(ObjectMapper objectMapper) {
        this.objectMapper = objectMapper;
    }

    private static ObjectMapper defaultObjectMapper() {
        ObjectMapper om = new ObjectMapper();
        om.disable(FAIL_ON_UNKNOWN_PROPERTIES);
        om.registerModule(new JavaTimeModule());
        om.registerModule(new Jdk8Module());
        return om;
    }

    @Override
    public byte[] serialize(Object object, Map<String, String> flags) {
        if (object == null) {
            return null;
        }
        try {
            if (isSerializeAsString(flags)) {
                return Bytes.toBytes(String.valueOf(object));
            }
            if (object instanceof String s) return Bytes.toBytes(s);
            if (object instanc

    public BestSuitCodec() {
        this(defaultObjectMapper());
    }

    public BestSuitCodec(ObjectMapper objectMapper) {
        this.objectMapper = objectMapper;
    }

    private static ObjectMapper defaultObjectMapper() {
        ObjectMapper om = new ObjectMapper();
        om.disable(FAIL_ON_UNKNOWN_PROPERTIES);
        om.registerModule(new JavaTimeModule());
        om.registerModule(new Jdk8Module());
        return om;
    }

    public byte[] serialize(Object object, Map<String, String> flags) {
        if (object == null) {
            return null;
        }
        try {
            if (isSerializeAsString(flags)) {
                return Bytes.toBytes(String.valueOf(object));
            }
            if (object instanceof String s) return Bytes.toBytes(s);
            if (object instanceof Integer i) return Bytes.toBytes(i);
            if (object instanceof Short s) return Bytes.toBytes(s);
            if (object instanceof Long l) return Bytes.toBytes(l);
            if (object instanceof Float f) return Bytes.toBytes(f);
            if (object instanceof Double d) return Bytes.toBytes(d);
            if (object instanceof BigDecimal bd) return Bytes.toBytes(bd);
            if (object instanceof Boolean b) return Bytes.toBytes(b);
            // Fallback: Jackson JSON
            return objectMapper.writeValueAsBytes(object);
        } catch (Exception e) {
            throw new CodecException("Fail

    public Object deserialize(byte[] bytes, Type type, Map<String, String> flags) {
        if (bytes == null) {
            return null;
        }
        try {
            Class<?> rawType = getRawType(type);
            if (isSerializeAsString(flags)) {
                return deserializeFromString(Bytes.toString(bytes), rawType);
            }
            if (rawType == String.class) return Bytes.toString(bytes);
            if (rawType == Integer.class) return Bytes.toInt(bytes);
            if (rawType == Short.class) return Bytes.toShort(bytes);
            if (rawType == Long.class) return Bytes.toLong(bytes);
            if (rawType == Float.class) return Bytes.toFloat(bytes);
            if (rawType == Double.class) return Bytes.toDouble(bytes);
            if (rawType == BigDecimal.class) return Bytes.toBigDecimal(bytes);
            if (rawType == Boolean.class) return Bytes.toBoolean(bytes);
            // Fallback: Jackson JSON
            JavaType javaType = objectMapper.

    public boolean canDeserialize(Type type) {
        try {
            Class<?> rawType = getRawType(type);
            if (rawType == String.class || rawType == Integer.class || rawType == Short.class
                    || rawType == Long.class || rawType == Float.class || rawType == Double.class
                    || rawType == BigDecimal.class || rawType == Boolean.class) {
                return true;
            }
            return objectMapper.canDeserialize(objectMapper.constructType(type));
        } catch (Exception e) {
            return false;
        }
    }

    private Object deserializeFromString(String str, Class<?> rawType) {
        if (rawType == String.class) return str;
        if (rawType == Integer.class) return Integer.valueOf(str);
        if (rawType == Short.class) return Short.valueOf(str);
        if (rawType == Long.class) return Long.valueOf(str);
        if (rawType == Float.class) return Float.valueOf(str);
        if (rawType == Double.class) return Double.valueOf(str);
        if (rawType == BigDecimal.class) return new BigDecimal(str);
        if (rawType == Boolean.class) return Boolean.valueOf(str);
        throw new CodecException("Cannot deserialize string '" + str + "' to type " + rawType.getName());
    }

    private boolean isSerializeAsString(Map<String, String> flags) {
        if (flags == null) return false;
        String val = flags.get(SERIALIZE_AS_STRING);
        return "true".equalsIgnoreCase(val);
    }

    private Class<?> getRawType(Type type) {
        if (type instanceof Class<?> c) return c;
        if (type instanceof java.lang.reflect.ParameterizedType pt) {
            return (Class<?>) pt.getRawType();
        }
        return Object.class;
    }

    public static class CodecException extends RuntimeException {
        public CodecException(String message) { super(message); }
        public CodecException(String message, Throwable cause) { super(message, cause); }
    }

public class AsyncHBaseDAO<R, T> {

    private final AsyncConnection connection;
    private final HBaseMapper<R, T> mapper;
    private final TableName tableName;
    private final Executor executor;

    public AsyncHBaseDAO(AsyncConnection connection, HBaseMapper<R, T> mapper) {
        this(connection, mapper, null);
    }

    public AsyncHBaseDAO(AsyncConnection connection, HBaseMapper<R, T> mapper, Executor executor) {
        this.connection = connection;
        this.mapper = mapper;
        this.tableName = TableName.valueOf(mapper.getTableName());
        this.executor = executor;
    }

    // ─── Read Operations ─────────────────────────────────────────────

    public CompletableFuture<T> get(R rowKey) {
        return applyMapping(getAsyncTable().get(createGet(rowKey)), this::resultToEntity);
    }

    public CompletableFuture<T> get(R rowKey, int numVersionsToFetch) throws IOException {
        Get get = createGet(rowKey);
        if (numVersionsToFetch > 1) {
       

    public AsyncHBaseDAO(AsyncConnection connection, HBaseMapper<R, T> mapper) {
        this(connection, mapper, null);
    }

    public AsyncHBaseDAO(AsyncConnection connection, HBaseMapper<R, T> mapper, Executor executor) {
        this.connection = connection;
        this.mapper = mapper;
        this.tableName = TableName.valueOf(mapper.getTableName());
        this.executor = executor;
    }

    public CompletableFuture<T> get(R rowKey) {
        return applyMapping(getAsyncTable().get(createGet(rowKey)), this::resultToEntity);
    }

    public CompletableFuture<T> get(R rowKey, int numVersionsToFetch) throws IOException {
        Get get = createGet(rowKey);
        if (numVersionsToFetch > 1) {
            get.readVersions(numVersionsToFetch);
        }
        return applyMapping(getAsyncTable().get(get), this::resultToEntity);
    }

    public CompletableFuture<T> get(R rowKey, int numVersionsToFetch, Executor exec) throws IOException {
        Get get = createGet(rowKey);
        if (numVersionsToFetch > 1) {
            get.readVersions(numVersionsToFetch);
        }
        return applyMapping(getAsyncTable().get(get), this::resultToEntity, exec);
    }

    public List<CompletableFuture<T>> get(List<R> rowKeys) throws IOException {
        return getBatch(rowKeys, 1, null);
    }

    public List<CompletableFuture<T>> get(List<R> rowKeys, int numVersionsToFetch) throws IOException {
        return getBatch(rowKeys, numVersionsToFetch, null);
    }

    public List<CompletableFuture<T>> get(List<R> rowKeys, int numVersionsToFetch, Executor exec) throws IOException {
        return getBatch(rowKeys, numVersionsToFetch, exec);
    }

    public CompletableFuture<List<T>> getAll(List<R> rowKeys) throws IOException {
        return collectNonNull(get(rowKeys));
    }

    public CompletableFuture<List<T>> getAll(List<R> rowKeys, int numVersionsToFetch) throws IOException {
        return collectNonNull(get(rowKeys, numVersionsToFetch));
    }

    public CompletableFuture<List<T>> scanAll(Scan scan) {
        return applyMapping(
                getAsyncTable().scanAll(scan),
                this::resultsToEntities);
    }

    public CompletableFuture<List<T>> scanAll(Scan scan, Executor exec) {
        return applyMapping(
                getAsyncTable().scanAll(scan),
                this::resultsToEntities,
                exec);
    }

    public CompletableFuture<List<T>> getByPrefix(byte[] rowPrefix) {
        Scan scan = new Scan();
        scan.setRowPrefixFilter(rowPrefix);
        return scanAll(scan);
    }

    public CompletableFuture<List<T>> getByPrefix(byte[] rowPrefix, int numVersionsToFetch) throws IOException {
        Scan scan = new Scan();
        scan.setRowPrefixFilter(rowPrefix);
        if (numVersionsToFetch > 1) {
            scan.readVersions(numVersionsToFetch);
        }
        return scanAll(scan);
    }

    public CompletableFuture<List<T>> scanRange(R startRowKey, R endRowKey) {
        return scanRange(startRowKey, true, endRowKey, false);
    }

    public CompletableFuture<List<T>> scanRange(R startRowKey, boolean startInclusive,
                                                R endRowKey, boolean endInclusive) {
        Scan scan = new Scan();
        scan.withStartRow(rowKeyToBytes(startRowKey), startInclusive);
        scan.withStopRow(rowKeyToBytes(endRowKey), endInclusive);
        return scanAll(scan);
    }

    public CompletableFuture<List<T>> scanRange(R startRowKey, boolean startInclusive,
                                                R endRowKey, boolean endInclusive,
                                                int numVersionsToFetch) throws IOException {
        Scan scan = new Scan();
        scan.withStartRow(rowKeyToBytes(startRowKey), startInclusive);
        scan.withStopRow(rowKeyToBytes(endRowKey), endInclusive);
        if (numVersionsToFetch > 1) {
            scan.readVersions(numVersionsToFetch);
        }
        return scanAll(scan);
    }

    public void scan(Scan scan, AdvancedScanResultConsumer consumer) {
        getAsyncTable().scan(scan, consumer);
    }

    public CompletableFuture<List<T>> scanStreaming(Scan scan) {
        return scanStreaming(scan, executor);
    }

    public CompletableFuture<List<T>> scanStreaming(Scan scan, Executor exec) {
        CompletableFuture<List<T>> resultFuture = new CompletableFuture<>();
        List<T> entities = Collections.synchronizedList(new ArrayList<>());

        getAsyncTable().scan(scan, new AdvancedScanResultConsumer() {
            @Override
            public void onNext(Result[] results, ScanController controller) {
                Runnable mapTask = () -> {
                    for (Result r : results) {
                        if (r != null && !r.isEmpty()) {
                            entities.add(mapper.readFromResult(r));
                        }
                    }
                };
                if (exec != null) {
                    exec.execute(mapTask);
                } else {
                    mapTask.run();
                }
            }

            @Override
            public void onError(Throwable error) {
                resultFuture.completeExceptionally(error);
          

            public void onNext(Result[] results, ScanController controller) {
                Runnable mapTask = () -> {
                    for (Result r : results) {
                        if (r != null && !r.isEmpty()) {
                            entities.add(mapper.readFromResult(r));
                        }
                    }
                };
                if (exec != null) {
                    exec.execute(mapTask);
                } else {
                    mapTask.run();
                }
            }

            public void onError(Throwable error) {
                resultFuture.completeExceptionally(error);
            }

            public void onComplete() {
                resultFuture.complete(entities);
            }

    public CompletableFuture<R> persist(T entity) {
        Put put = mapper.writeAsPut(entity);
        R rowKey = mapper.getRowKey(entity);
        return getAsyncTable().put(put).thenApply(v -> rowKey);
    }

    public List<CompletableFuture<R>> persist(List<T> entities) {
        AsyncTable<AdvancedScanResultConsumer> table = getAsyncTable();
        List<CompletableFuture<R>> futures = new ArrayList<>(entities.size());
        for (T entity : entities) {
            Put put = mapper.writeAsPut(entity);
            R rowKey = mapper.getRowKey(entity);
            futures.add(table.put(put).thenApply(v -> rowKey));
        }
        return futures;
    }

    public CompletableFuture<List<R>> persistAll(List<T> entities) {
        List<CompletableFuture<R>> futures = persist(entities);
        return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
                .thenApply(v -> {
                    List<R> keys = new ArrayList<>(futures.size());
                    for (CompletableFuture<R> f : futures) {
                        keys.add(f.join());
                    }
                    return keys;
                });
    }

    public CompletableFuture<Void> delete(R rowKey) {
        Delete delete = new Delete(rowKeyToBytes(rowKey));
        return getAsyncTable().delete(delete);
    }

    public List<CompletableFuture<Void>> deleteByKeys(R... rowKeys) {
        AsyncTable<AdvancedScanResultConsumer> table = getAsyncTable();
        List<Delete> deletes = new ArrayList<>(rowKeys.length);
        for (R rk : rowKeys) {
            deletes.add(new Delete(rowKeyToBytes(rk)));
        }
        return table.delete(deletes);
    }

    public CompletableFuture<Void> deleteEntity(T entity) {
        return delete(mapper.getRowKey(entity));
    }

    public CompletableFuture<Long> increment(R rowKey, String fieldName, long amount) {
        byte[][] col = mapper.getColumn(fieldName);
        Increment inc = new Increment(rowKeyToBytes(rowKey));
        inc.addColumn(col[0], col[1], amount);
        return applyMapping(
                getAsyncTable().increment(inc),
                result -> Bytes.toLong(result.getValue(col[0], col[1])));
    }

    public CompletableFuture<T> increment(Increment increment) {
        return applyMapping(
                getAsyncTable().increment(increment),
                this::resultToEntity);
    }

    public CompletableFuture<T> append(Append append) {
        return applyMapping(
                getAsyncTable().append(append),
                this::resultToEntity);
    }

    public CompletableFuture<Boolean> exists(R rowKey) {
        return getAsyncTable().exists(createGet(rowKey));
    }

    public List<CompletableFuture<Boolean>> exists(R... rowKeys) {
        AsyncTable<AdvancedScanResultConsumer> table = getAsyncTable();
        List<Get> gets = new ArrayList<>(rowKeys.length);
        for (R rk : rowKeys) {
            gets.add(createGet(rk));
        }
        return table.exists(gets);
    }

    public AsyncTable<AdvancedScanResultConsumer> getAsyncTable() {
        return connection.getTable(tableName);
    }

    public Get createGet(R rowKey) {
        return new Get(rowKeyToBytes(rowKey));
    }

    public Increment createIncrement(R rowKey) {
        return new Increment(rowKeyToBytes(rowKey));
    }

    public Append createAppend(R rowKey) {
        return new Append(rowKeyToBytes(rowKey));
    }