diff --git a/slo-workload/AGENTS.md b/slo-workload/AGENTS.md index 4d21e97..cf82adb 100644 --- a/slo-workload/AGENTS.md +++ b/slo-workload/AGENTS.md @@ -147,9 +147,12 @@ pushing to CI. Locally that's fine; in CI it would merge current and baseline series. The action always sets it, but watch out when reproducing CI behaviour manually. -- **Prefill failure threshold is 50%**. If half the writes during prefill - fail, the runner refuses to start the read loop — empty key-space means - meaningless read latency. Don't lower the threshold to "fix" a flaky test. +- **Any final operation failure fails the workload**, including prefill. + Recovered retries stay successful. The first terminal cause survives peer + shutdown and cleanup; successful later operations cannot erase it. +- **Each launcher owns a process-private table for its whole lifecycle.** + One owner checks a reserved control key's exact generation and payload before, + during and after load; random load reads do not assert committed-key presence. - **`acceptUnknownOptions(false)`** on JCommander is intentional. A typo in the ydb-slo-action `workload_*_command` input should fail loudly, not silently fall back to defaults. diff --git a/slo-workload/core/pom.xml b/slo-workload/core/pom.xml index fb3ddf5..424a0ad 100644 --- a/slo-workload/core/pom.xml +++ b/slo-workload/core/pom.xml @@ -21,6 +21,12 @@ + + org.junit.jupiter + junit-jupiter + 5.10.1 + test + com.beust jcommander diff --git a/slo-workload/core/src/main/java/tech/ydb/slo/core/Config.java b/slo-workload/core/src/main/java/tech/ydb/slo/core/Config.java index 2e15ab2..5e5b4cd 100644 --- a/slo-workload/core/src/main/java/tech/ydb/slo/core/Config.java +++ b/slo-workload/core/src/main/java/tech/ydb/slo/core/Config.java @@ -9,7 +9,7 @@ public final class Config { private final int durationSeconds; private final String otlpEndpoint; - private Config( + Config( String connectionString, String jdbcUrl, String token, diff --git a/slo-workload/core/src/main/java/tech/ydb/slo/core/Launcher.java b/slo-workload/core/src/main/java/tech/ydb/slo/core/Launcher.java index c2103bc..cd24ee8 100644 --- a/slo-workload/core/src/main/java/tech/ydb/slo/core/Launcher.java +++ b/slo-workload/core/src/main/java/tech/ydb/slo/core/Launcher.java @@ -111,7 +111,8 @@ public static int run( } public static String tablePathFor(Config config) { - return sanitize(config.workloadName()) + "_" + sanitize(config.ref()); + return sanitize(config.workloadName()) + "_" + sanitize(config.ref()) + + "_" + java.util.UUID.randomUUID().toString().replace("-", ""); } private static void closeQuietly(AutoCloseable closeable, String name) { diff --git a/slo-workload/core/src/main/java/tech/ydb/slo/core/kv/ControlRows.java b/slo-workload/core/src/main/java/tech/ydb/slo/core/kv/ControlRows.java new file mode 100644 index 0000000..38acd0d --- /dev/null +++ b/slo-workload/core/src/main/java/tech/ydb/slo/core/kv/ControlRows.java @@ -0,0 +1,42 @@ +package tech.ydb.slo.core.kv; + +import java.time.Instant; +import java.time.temporal.ChronoUnit; + +/** One runner owns this key in its process-private table; load IDs start at zero. */ +final class ControlRows { + private Row published; + private long generation; + + void verify(KvSession session, KvWorkloadParams params) { + if (published != null) { + requireRow(session.read(published.id(), params.readTimeoutMs()), published); + } + Row next = new Row(Long.MAX_VALUE, "control/" + generation, generation + 0.25, + Instant.now().truncatedTo(ChronoUnit.MICROS), generation); + requireSuccess(session.write(next, params.writeTimeoutMs())); + published = next; + requireRow(session.read(next.id(), params.readTimeoutMs()), next); + generation++; + } + + private static void requireSuccess(OpOutcome outcome) { + if (!outcome.isSuccess()) { + throw new IllegalStateException("Control operation failed: " + outcome.errorKind(), outcome.cause()); + } + } + + static void requireRow(OpOutcome outcome, Row expected) { + requireSuccess(outcome); + if (outcome.rows().size() != 1) { + throw new IllegalStateException("Control row cardinality: " + outcome.rows().size()); + } + Row actual = outcome.rows().get(0); + if (actual.id() != expected.id() || actual.payloadHash() != expected.payloadHash() + || !expected.payloadStr().equals(actual.payloadStr()) + || Double.compare(expected.payloadDouble(), actual.payloadDouble()) != 0 + || !expected.payloadTimestamp().equals(actual.payloadTimestamp())) { + throw new IllegalStateException("Control row mismatch at generation " + expected.payloadHash()); + } + } +} diff --git a/slo-workload/core/src/main/java/tech/ydb/slo/core/kv/OpOutcome.java b/slo-workload/core/src/main/java/tech/ydb/slo/core/kv/OpOutcome.java index 6edc914..6c980aa 100644 --- a/slo-workload/core/src/main/java/tech/ydb/slo/core/kv/OpOutcome.java +++ b/slo-workload/core/src/main/java/tech/ydb/slo/core/kv/OpOutcome.java @@ -4,19 +4,40 @@ public final class OpOutcome { private final boolean success; private final int retryAttempts; private final String errorKind; + private final Throwable cause; + private final java.util.List rows; - private OpOutcome(boolean success, int retryAttempts, String errorKind) { + private OpOutcome(boolean success, int retryAttempts, String errorKind, Throwable cause, + java.util.List rows) { this.success = success; this.retryAttempts = Math.max(0, retryAttempts); this.errorKind = errorKind; + this.cause = cause; + this.rows = java.util.List.copyOf(rows); } public static OpOutcome success(int retryAttempts) { - return new OpOutcome(true, retryAttempts, null); + return success(retryAttempts, java.util.List.of()); + } + + public static OpOutcome success(int retryAttempts, java.util.List rows) { + return new OpOutcome(true, retryAttempts, null, null, rows); + } + + public java.util.List rows() { + return rows; } public static OpOutcome error(int retryAttempts, String errorKind) { - return new OpOutcome(false, retryAttempts, errorKind); + return error(retryAttempts, errorKind, null); + } + + public static OpOutcome error(int retryAttempts, String errorKind, Throwable cause) { + return new OpOutcome(false, retryAttempts, errorKind, cause, java.util.List.of()); + } + + public Throwable cause() { + return cause; } public boolean isSuccess() { diff --git a/slo-workload/core/src/main/java/tech/ydb/slo/core/kv/RowGenerator.java b/slo-workload/core/src/main/java/tech/ydb/slo/core/kv/RowGenerator.java index c2dcd30..6c6aa07 100644 --- a/slo-workload/core/src/main/java/tech/ydb/slo/core/kv/RowGenerator.java +++ b/slo-workload/core/src/main/java/tech/ydb/slo/core/kv/RowGenerator.java @@ -21,10 +21,13 @@ public Row generate() { } public static Row generate(long id) { + if (id == Long.MAX_VALUE) { + throw new IllegalStateException("Load writer reached the reserved control key"); + } long payloadHash = ThreadLocalRandom.current().nextLong(); double payloadDouble = ThreadLocalRandom.current().nextDouble(); String payloadStr = randomPayloadString(); - Instant payloadTimestamp = Instant.now(); + Instant payloadTimestamp = Instant.now().truncatedTo(java.time.temporal.ChronoUnit.MICROS); return new Row(id, payloadStr, payloadDouble, payloadTimestamp, payloadHash); } diff --git a/slo-workload/core/src/main/java/tech/ydb/slo/core/kv/WorkloadRunner.java b/slo-workload/core/src/main/java/tech/ydb/slo/core/kv/WorkloadRunner.java index 8e4a195..bc7a126 100644 --- a/slo-workload/core/src/main/java/tech/ydb/slo/core/kv/WorkloadRunner.java +++ b/slo-workload/core/src/main/java/tech/ydb/slo/core/kv/WorkloadRunner.java @@ -7,6 +7,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; import com.google.common.util.concurrent.RateLimiter; import org.slf4j.Logger; @@ -17,13 +18,13 @@ public final class WorkloadRunner { private static final Logger logger = LoggerFactory.getLogger(WorkloadRunner.class); - private static final double PREFILL_SUCCESS_THRESHOLD = 0.5; - private final KvClient client; private final Metrics metrics; private final KvWorkloadParams params; private final String tablePath; private final RowGenerator generator; + private final AtomicReference firstFailure = new AtomicReference<>(); + private final ControlRows control = new ControlRows(); public WorkloadRunner(KvClient client, Metrics metrics, KvWorkloadParams params, String tablePath) { this.client = client; @@ -38,6 +39,8 @@ public void setup() throws Exception { client.createTable(params, tablePath); logger.info("table {} created", tablePath); + verifyControl(); + if (params.prefillCount() <= 0) { logger.info("prefill count <= 0, skipping prefill"); return; @@ -49,29 +52,22 @@ public void setup() throws Exception { parallelism, namedThreadFactory("slo-prefill-") ); AtomicLong nextId = new AtomicLong(0); - AtomicInteger failed = new AtomicInteger(); - AtomicInteger sessionOpenFailures = new AtomicInteger(); try { for (int w = 0; w < parallelism; w++) { prefillPool.execute(() -> { try (KvSession session = client.openSession()) { long id; - while ((id = nextId.getAndIncrement()) < params.prefillCount()) { + while (firstFailure.get() == null && + (id = nextId.getAndIncrement()) < params.prefillCount()) { OpOutcome outcome = session.write( RowGenerator.generate(id), params.writeTimeoutMs()); if (!outcome.isSuccess()) { - int f = failed.incrementAndGet(); - if (f <= 5) { - logger.warn("prefill row {} failed: {}", id, outcome.errorKind()); - } + recordFailure(outcome); + return; } } - } catch (Exception e) { - sessionOpenFailures.incrementAndGet(); - long firstUnclaimed = nextId.getAndSet(params.prefillCount()); - if (firstUnclaimed < params.prefillCount()) { - failed.addAndGet((int) (params.prefillCount() - firstUnclaimed)); - } + } catch (Throwable e) { + firstFailure.compareAndSet(null, e); logger.error("prefill worker failed to open session: {}", e.toString()); } }); @@ -79,31 +75,13 @@ parallelism, namedThreadFactory("slo-prefill-") } finally { prefillPool.shutdown(); if (!prefillPool.awaitTermination(5, TimeUnit.MINUTES)) { + firstFailure.compareAndSet(null, new IllegalStateException("prefill workers did not finish")); prefillPool.shutdownNow(); } } - int total = params.prefillCount(); - int failedCount = failed.get(); - int succeeded = total - failedCount; - if (sessionOpenFailures.get() == parallelism) { - throw new IllegalStateException( - "all " + parallelism + " prefill workers failed to open a session — " - + "check YDB connectivity and credentials" - ); - } - if (succeeded < total * PREFILL_SUCCESS_THRESHOLD) { - throw new IllegalStateException( - "prefill completed with " + failedCount + " failed rows out of " + total - + " (success rate < " + (int) (PREFILL_SUCCESS_THRESHOLD * 100) - + "%); reads would target an empty key-space, refusing to run" - ); - } - if (failedCount > 0) { - logger.warn("prefill completed with {} failed rows out of {}", failedCount, total); - } else { - logger.info("prefill completed"); - } + throwIfFailed(); + logger.info("prefill completed"); } public void run() throws InterruptedException { @@ -119,6 +97,7 @@ public void run() throws InterruptedException { if (readWorkers == 0 && writeWorkers == 0) { logger.warn("both read and write RPS are <= 0, run phase has nothing to do"); + verifyControl(); return; } @@ -145,6 +124,18 @@ public void run() throws InterruptedException { logger.info("write RPS <= 0, skipping write workers"); } + RateLimiter controlLimiter = RateLimiter.create(1.0); + while (firstFailure.get() == null && !Thread.currentThread().isInterrupted()) { + long remaining = endNanos - System.nanoTime(); + if (remaining <= 0 || !controlLimiter.tryAcquire(remaining, TimeUnit.NANOSECONDS)) { + break; + } + if (firstFailure.get() != null || System.nanoTime() >= endNanos) { + break; + } + verifyControl(); + } + long graceNanos = TimeUnit.SECONDS.toNanos(params.shutdownTimeSeconds()); long waitNanos = durationSeconds > 0 ? Math.max(0L, endNanos - System.nanoTime()) + graceNanos @@ -160,6 +151,7 @@ public void run() throws InterruptedException { if (readPool != null) { long started = System.nanoTime(); if (!readPool.awaitTermination(waitNanos, TimeUnit.NANOSECONDS)) { + firstFailure.compareAndSet(null, new IllegalStateException("read pool did not drain")); logger.warn("read pool did not drain within deadline, forcing shutdown"); readPool.shutdownNow(); } @@ -167,14 +159,20 @@ public void run() throws InterruptedException { } if (writePool != null) { if (!writePool.awaitTermination(waitNanos, TimeUnit.NANOSECONDS)) { + firstFailure.compareAndSet(null, new IllegalStateException("write pool did not drain")); logger.warn("write pool did not drain within deadline, forcing shutdown"); writePool.shutdownNow(); } } + } catch (InterruptedException e) { + throwIfFailed(); + throw e; } finally { forceShutdown(readPool, "read pool"); forceShutdown(writePool, "write pool"); } + throwIfFailed(); + verifyControl(); } public void teardown() { @@ -182,9 +180,18 @@ public void teardown() { client.dropTable(tablePath); } + private void verifyControl() { + try (KvSession session = client.openSession()) { + control.verify(session, params); + } catch (Throwable failure) { + firstFailure.compareAndSet(null, failure); + throwIfFailed(); + } + } + private void readWorkerLoop(long endNanos, RateLimiter limiter, AtomicLong writesIssued) { try (KvSession session = client.openSession()) { - while (!Thread.currentThread().isInterrupted()) { + while (firstFailure.get() == null && !Thread.currentThread().isInterrupted()) { long remaining = endNanos - System.nanoTime(); if (remaining <= 0) { return; @@ -193,20 +200,26 @@ private void readWorkerLoop(long endNanos, RateLimiter limiter, AtomicLong write if (!limiter.tryAcquire(remaining, TimeUnit.NANOSECONDS)) { return; } + if (firstFailure.get() != null || System.nanoTime() >= endNanos) { + return; + } try { readOnce(session, writesIssued.get()); } catch (Throwable t) { + firstFailure.compareAndSet(null, t); logger.warn("read op threw unexpectedly: {}", t.toString()); + return; } } - } catch (Exception e) { + } catch (Throwable e) { + firstFailure.compareAndSet(null, e); logger.warn("read worker failed to open session: {}", e.toString()); } } private void writeWorkerLoop(long endNanos, RateLimiter limiter, AtomicLong writesIssued) { try (KvSession session = client.openSession()) { - while (!Thread.currentThread().isInterrupted()) { + while (firstFailure.get() == null && !Thread.currentThread().isInterrupted()) { long remaining = endNanos - System.nanoTime(); if (remaining <= 0) { return; @@ -214,14 +227,20 @@ private void writeWorkerLoop(long endNanos, RateLimiter limiter, AtomicLong writ if (!limiter.tryAcquire(remaining, TimeUnit.NANOSECONDS)) { return; } + if (firstFailure.get() != null || System.nanoTime() >= endNanos) { + return; + } try { writeOnce(session, generator.generate()); writesIssued.incrementAndGet(); } catch (Throwable t) { + firstFailure.compareAndSet(null, t); logger.warn("write op threw unexpectedly: {}", t.toString()); + return; } } - } catch (Exception e) { + } catch (Throwable e) { + firstFailure.compareAndSet(null, e); logger.warn("write worker failed to open session: {}", e.toString()); } } @@ -235,6 +254,7 @@ private void readOnce(KvSession session, long writesObserved) { if (outcome.isSuccess()) { span.finishSuccess(outcome.retryAttempts()); } else { + recordFailure(outcome); span.finishError(outcome.retryAttempts(), outcome.errorKind()); logger.debug("read {} failed: {}", id, outcome.errorKind()); } @@ -246,11 +266,24 @@ private void writeOnce(KvSession session, Row row) { if (outcome.isSuccess()) { span.finishSuccess(outcome.retryAttempts()); } else { + recordFailure(outcome); span.finishError(outcome.retryAttempts(), outcome.errorKind()); logger.debug("write {} failed: {}", row.id(), outcome.errorKind()); } } + private void recordFailure(OpOutcome outcome) { + firstFailure.compareAndSet(null, outcome.cause() != null ? outcome.cause() + : new IllegalStateException("Operation failed: " + outcome.errorKind())); + } + + private void throwIfFailed() { + Throwable failure = firstFailure.get(); + if (failure != null) { + throw new IllegalStateException("Workload failed", failure); + } + } + private int workerCount(int rps) { if (rps <= 0) { return 0; diff --git a/slo-workload/core/src/test/java/tech/ydb/slo/core/WorkloadRunnerTest.java b/slo-workload/core/src/test/java/tech/ydb/slo/core/WorkloadRunnerTest.java new file mode 100644 index 0000000..b7ab813 --- /dev/null +++ b/slo-workload/core/src/test/java/tech/ydb/slo/core/WorkloadRunnerTest.java @@ -0,0 +1,85 @@ +package tech.ydb.slo.core; + +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import com.beust.jcommander.JCommander; +import org.junit.jupiter.api.Test; + +import tech.ydb.slo.core.kv.KvClient; +import tech.ydb.slo.core.kv.KvSession; +import tech.ydb.slo.core.kv.KvWorkloadParams; +import tech.ydb.slo.core.kv.OpOutcome; +import tech.ydb.slo.core.kv.Row; +import tech.ydb.slo.core.kv.WorkloadRunner; + +import static org.junit.jupiter.api.Assertions.*; + +class WorkloadRunnerTest { + @Test + void checksControlGenerationsWhileLoadIsActive() throws Exception { + KvWorkloadParams params = new KvWorkloadParams(); + JCommander.newBuilder().addObject(params).build().parse( + "--read-rps", "0", "--write-rps", "1", "--prefill-count", "0", + "--duration", "0", "--max-workers", "1"); + AtomicReference stored = new AtomicReference<>(); + CountDownLatch loadStarted = new CountDownLatch(1); + CountDownLatch checkedDuringLoad = new CountDownLatch(1); + KvClient client = new KvClient() { + public void createTable(KvWorkloadParams options, String tablePath) {} + public void dropTable(String tablePath) {} + public KvSession openSession() { + return new KvSession() { + public OpOutcome write(Row row, int timeoutMs) { + if (row.id() == Long.MAX_VALUE) { + stored.set(row); + } else { + loadStarted.countDown(); + } + return OpOutcome.success(0); + } + public OpOutcome read(long id, int timeoutMs) { + Row row = stored.get(); + if (row.payloadHash() >= 2) { + try { + assertTrue(loadStarted.await(10, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + checkedDuringLoad.countDown(); + row = new Row(row.id(), row.payloadStr(), row.payloadDouble(), + row.payloadTimestamp(), row.payloadHash() - 1); + } + return OpOutcome.success(0, List.of(row)); + } + }; + } + }; + Config config = new Config("grpc://localhost:1/local", "", "", "unit", + "control-runner", 0, ""); + try (Metrics metrics = Metrics.create(config)) { + WorkloadRunner runner = new WorkloadRunner(client, metrics, params, "unused"); + runner.setup(); + ExecutorService executor = Executors.newSingleThreadExecutor(); + Future result = executor.submit( + () -> assertThrows(IllegalStateException.class, runner::run)); + try { + assertTrue(checkedDuringLoad.await(10, TimeUnit.SECONDS)); + IllegalStateException failure = result.get(10, TimeUnit.SECONDS); + assertInstanceOf(IllegalStateException.class, failure.getCause()); + assertEquals(0, loadStarted.getCount()); + assertEquals(2, stored.get().payloadHash()); + } finally { + result.cancel(true); + executor.shutdownNow(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } + } + } +} diff --git a/slo-workload/core/src/test/java/tech/ydb/slo/core/kv/ControlRowsTest.java b/slo-workload/core/src/test/java/tech/ydb/slo/core/kv/ControlRowsTest.java new file mode 100644 index 0000000..8e2c38e --- /dev/null +++ b/slo-workload/core/src/test/java/tech/ydb/slo/core/kv/ControlRowsTest.java @@ -0,0 +1,79 @@ +package tech.ydb.slo.core.kv; + +import java.time.Instant; +import java.util.List; +import org.junit.jupiter.api.Test; +import static org.junit.jupiter.api.Assertions.*; + +class ControlRowsTest { + private final Row expected = new Row(Long.MAX_VALUE, "control/7", 7.25, Instant.ofEpochSecond(1, 123000), 7); + + @Test + void exactRowsOnly() { + assertThrows(IllegalStateException.class, () -> RowGenerator.generate(Long.MAX_VALUE)); + assertDoesNotThrow(() -> ControlRows.requireRow(OpOutcome.success(3, List.of(expected)), expected)); + assertThrows(IllegalStateException.class, () -> ControlRows.requireRow(OpOutcome.success(0), expected)); + assertThrows(IllegalStateException.class, () -> ControlRows.requireRow( + OpOutcome.success(0, List.of(expected, expected)), expected)); + for (Row bad : List.of( + new Row(1, expected.payloadStr(), 7.25, expected.payloadTimestamp(), 7), + new Row(expected.id(), expected.payloadStr(), 7.25, expected.payloadTimestamp(), 6), + new Row(expected.id(), "corrupt", 7.25, expected.payloadTimestamp(), 7), + new Row(expected.id(), expected.payloadStr(), 8.25, expected.payloadTimestamp(), 7), + new Row(expected.id(), expected.payloadStr(), 7.25, expected.payloadTimestamp().plusNanos(1), 7))) { + assertThrows(IllegalStateException.class, () -> ControlRows.requireRow( + OpOutcome.success(0, List.of(bad)), expected)); + } + } + + @Test + void preservesSdkFailureBeforeCheckingPresence() { + Throwable cause = new RuntimeException("sdk"); + assertSame(cause, assertThrows(IllegalStateException.class, () -> ControlRows.requireRow( + OpOutcome.error(2, "sdk", cause), expected)).getCause()); + } + + @Test + void sequentialOwnershipAndRetainedGeneration() { + ControlRows control = new ControlRows(); + KvWorkloadParams params = new KvWorkloadParams(); + class Session implements KvSession { + Row stored; + int writes; + int reads; + public OpOutcome write(Row row, int timeoutMs) { + assertEquals(writes++, row.payloadHash()); + assertEquals(0, row.payloadTimestamp().getNano() % 1000); + stored = row; + return OpOutcome.success(2); + } + public OpOutcome read(long id, int timeoutMs) { + reads++; + assertEquals(stored.id(), id); + return OpOutcome.success(1, List.of(stored)); + } + } + Session session = new Session(); + control.verify(session, params); + assertEquals(1, session.writes); + control.verify(session, params); + assertEquals(2, session.writes); + assertEquals(3, session.reads); + session.stored = expected; + assertThrows(IllegalStateException.class, () -> control.verify(session, params)); + assertEquals(2, session.writes); + } + + @Test + void failedWriteDoesNotPublishReadableRow() { + int[] reads = {0}; + Throwable cause = new RuntimeException("write"); + KvSession session = new KvSession() { + public OpOutcome write(Row row, int timeoutMs) { return OpOutcome.error(1, "sdk", cause); } + public OpOutcome read(long id, int timeoutMs) { reads[0]++; return OpOutcome.success(0); } + }; + assertSame(cause, assertThrows(IllegalStateException.class, + () -> new ControlRows().verify(session, new KvWorkloadParams())).getCause()); + assertEquals(0, reads[0]); + } +} diff --git a/slo-workload/jdbc/src/main/java/tech/ydb/slo/jdbc/JdbcKvClient.java b/slo-workload/jdbc/src/main/java/tech/ydb/slo/jdbc/JdbcKvClient.java index 45c7982..c5a7322 100644 --- a/slo-workload/jdbc/src/main/java/tech/ydb/slo/jdbc/JdbcKvClient.java +++ b/slo-workload/jdbc/src/main/java/tech/ydb/slo/jdbc/JdbcKvClient.java @@ -98,8 +98,7 @@ public OpOutcome read(long id, int timeoutMs) { while (attempts < maxAttempts) { attempts++; try { - read(id, hash, timeoutSeconds(timeoutMs)); - return OpOutcome.success(attempts - 1); + return OpOutcome.success(attempts - 1, read(id, hash, timeoutSeconds(timeoutMs))); } catch (SQLException e) { last = e; if (!isRetryable(e) || attempts >= maxAttempts) { @@ -111,7 +110,7 @@ public OpOutcome read(long id, int timeoutMs) { } } } - return OpOutcome.error(attempts - 1, classifyError(last)); + return OpOutcome.error(attempts - 1, classifyError(last), last); } @Override @@ -135,7 +134,7 @@ public OpOutcome write(Row row, int timeoutMs) { } } } - return OpOutcome.error(attempts - 1, classifyError(last)); + return OpOutcome.error(attempts - 1, classifyError(last), last); } @Override @@ -173,16 +172,21 @@ private PreparedStatement writeStmt() throws SQLException { return writeStmt; } - private void read(long id, long hash, int timeoutSeconds) throws SQLException { + private java.util.List read(long id, long hash, int timeoutSeconds) throws SQLException { PreparedStatement stmt = readStmt(); stmt.setQueryTimeout(timeoutSeconds); stmt.setLong(1, id); stmt.setLong(2, hash); + java.util.List rows = new java.util.ArrayList<>(); try (ResultSet rs = stmt.executeQuery()) { while (rs.next()) { - rs.getLong("id"); + rows.add(new Row(((Number) rs.getObject("id")).longValue(), rs.getString("payload_str"), + ((Number) rs.getObject("payload_double")).doubleValue(), + rs.getTimestamp("payload_timestamp").toInstant(), + ((Number) rs.getObject("payload_hash")).longValue())); } } + return rows; } private void write(Row row, long hash, int timeoutSeconds) throws SQLException { diff --git a/slo-workload/query/src/main/java/tech/ydb/slo/query/QueryKvClient.java b/slo-workload/query/src/main/java/tech/ydb/slo/query/QueryKvClient.java index 0a9181d..663adca 100644 --- a/slo-workload/query/src/main/java/tech/ydb/slo/query/QueryKvClient.java +++ b/slo-workload/query/src/main/java/tech/ydb/slo/query/QueryKvClient.java @@ -9,6 +9,7 @@ import tech.ydb.common.transaction.TxMode; import tech.ydb.core.Result; import tech.ydb.core.Status; +import tech.ydb.core.UnexpectedResultException; import tech.ydb.core.grpc.GrpcTransport; import tech.ydb.query.QueryClient; import tech.ydb.query.settings.ExecuteQuerySettings; @@ -154,17 +155,21 @@ public OpOutcome read(long id, int timeoutMs) { int retryAttempts = Math.max(0, attempts.get() - 1); if (!result.getStatus().isSuccess()) { - return OpOutcome.error(retryAttempts, classifyStatus(result.getStatus())); + return OpOutcome.error(retryAttempts, classifyStatus(result.getStatus()), + new UnexpectedResultException("Final read failed", result.getStatus())); } QueryReader reader = result.getValue(); - if (reader.getResultSetCount() > 0) { - ResultSetReader rs = reader.getResultSet(0); + java.util.List rows = new java.util.ArrayList<>(); + for (int i = 0; i < reader.getResultSetCount(); i++) { + ResultSetReader rs = reader.getResultSet(i); while (rs.next()) { - rs.getColumn("id").getUint64(); + rows.add(new Row(rs.getColumn("id").getUint64(), + rs.getColumn("payload_str").getText(), rs.getColumn("payload_double").getDouble(), + rs.getColumn("payload_timestamp").getTimestamp(), rs.getColumn("payload_hash").getUint64())); } } - return OpOutcome.success(retryAttempts); + return OpOutcome.success(retryAttempts, rows); } @Override @@ -196,7 +201,8 @@ public OpOutcome write(Row row, int timeoutMs) { if (status.isSuccess()) { return OpOutcome.success(retryAttempts); } - return OpOutcome.error(retryAttempts, classifyStatus(status)); + return OpOutcome.error(retryAttempts, classifyStatus(status), + new UnexpectedResultException("Final write failed", status)); } } diff --git a/slo-workload/spring-data-jdbc/src/main/java/tech/ydb/slo/springjdbc/KvOperationService.java b/slo-workload/spring-data-jdbc/src/main/java/tech/ydb/slo/springjdbc/KvOperationService.java index 1dd2368..d7f9031 100644 --- a/slo-workload/spring-data-jdbc/src/main/java/tech/ydb/slo/springjdbc/KvOperationService.java +++ b/slo-workload/spring-data-jdbc/src/main/java/tech/ydb/slo/springjdbc/KvOperationService.java @@ -28,13 +28,13 @@ public KvOperationService(JdbcTemplate jdbc) { } @YdbTransactional(readOnly = true) - public void read(String tablePath, long id, long hash) { + public java.util.List read(String tablePath, long id, long hash) { ATTEMPTS.get()[0]++; - jdbc.query(String.format(KvSchema.SELECT_TEMPLATE, tablePath), rs -> { - while (rs.next()) { - rs.getLong("id"); - } - }, id, hash); + return jdbc.query(String.format(KvSchema.SELECT_TEMPLATE, tablePath), (rs, index) -> + new Row(((Number) rs.getObject("id")).longValue(), rs.getString("payload_str"), + ((Number) rs.getObject("payload_double")).doubleValue(), + rs.getTimestamp("payload_timestamp").toInstant(), + ((Number) rs.getObject("payload_hash")).longValue()), id, hash); } @YdbTransactional(idempotent = true) diff --git a/slo-workload/spring-data-jdbc/src/main/java/tech/ydb/slo/springjdbc/SpringJdbcKvClient.java b/slo-workload/spring-data-jdbc/src/main/java/tech/ydb/slo/springjdbc/SpringJdbcKvClient.java index 949bf0a..b054da1 100644 --- a/slo-workload/spring-data-jdbc/src/main/java/tech/ydb/slo/springjdbc/SpringJdbcKvClient.java +++ b/slo-workload/spring-data-jdbc/src/main/java/tech/ydb/slo/springjdbc/SpringJdbcKvClient.java @@ -88,12 +88,12 @@ public OpOutcome read(long id, int timeoutMs) { KvOperationService.resetAttempts(); try { long hash = RowGenerator.numericHash(id); - operations.read(tablePath, id, hash); - return OpOutcome.success(Math.max(0, KvOperationService.currentAttempts() - 1)); + java.util.List rows = operations.read(tablePath, id, hash); + return OpOutcome.success(Math.max(0, KvOperationService.currentAttempts() - 1), rows); } catch (RuntimeException e) { return OpOutcome.error( Math.max(0, KvOperationService.currentAttempts() - 1), - classifyError(e) + classifyError(e), e ); } } @@ -108,7 +108,7 @@ public OpOutcome write(Row row, int timeoutMs) { } catch (RuntimeException e) { return OpOutcome.error( Math.max(0, KvOperationService.currentAttempts() - 1), - classifyError(e) + classifyError(e), e ); } } diff --git a/slo-workload/spring-data-jpa/src/main/java/tech/ydb/slo/springjpa/KvOperationService.java b/slo-workload/spring-data-jpa/src/main/java/tech/ydb/slo/springjpa/KvOperationService.java index 39621c7..9d7049f 100644 --- a/slo-workload/spring-data-jpa/src/main/java/tech/ydb/slo/springjpa/KvOperationService.java +++ b/slo-workload/spring-data-jpa/src/main/java/tech/ydb/slo/springjpa/KvOperationService.java @@ -31,7 +31,7 @@ public static int currentAttempts() { private EntityManager entityManager; @YdbTransactional(readOnly = true) - public void read(String tablePath, long id, long hash) { + public List read(String tablePath, long id, long hash) { ATTEMPTS.get()[0]++; Query query = entityManager.createNativeQuery(String.format(KvSchema.SELECT_TEMPLATE, tablePath)) .setParameter(1, id) @@ -40,11 +40,13 @@ public void read(String tablePath, long id, long hash) { query.setFlushMode(FlushModeType.COMMIT); @SuppressWarnings("unchecked") List rows = query.getResultList(); + List decoded = new java.util.ArrayList<>(); for (Object[] row : rows) { - if (row[0] != null) { - ((Number) row[0]).longValue(); - } + decoded.add(new Row(((Number) row[0]).longValue(), (String) row[1], + ((Number) row[2]).doubleValue(), ((Timestamp) row[3]).toInstant(), + ((Number) row[4]).longValue())); } + return decoded; } @YdbTransactional(idempotent = true) diff --git a/slo-workload/spring-data-jpa/src/main/java/tech/ydb/slo/springjpa/SpringJpaKvClient.java b/slo-workload/spring-data-jpa/src/main/java/tech/ydb/slo/springjpa/SpringJpaKvClient.java index 0c3326d..c213e75 100644 --- a/slo-workload/spring-data-jpa/src/main/java/tech/ydb/slo/springjpa/SpringJpaKvClient.java +++ b/slo-workload/spring-data-jpa/src/main/java/tech/ydb/slo/springjpa/SpringJpaKvClient.java @@ -83,12 +83,12 @@ public OpOutcome read(long id, int timeoutMs) { KvOperationService.resetAttempts(); try { long hash = RowGenerator.numericHash(id); - operations.read(tablePath, id, hash); - return OpOutcome.success(Math.max(0, KvOperationService.currentAttempts() - 1)); + java.util.List rows = operations.read(tablePath, id, hash); + return OpOutcome.success(Math.max(0, KvOperationService.currentAttempts() - 1), rows); } catch (RuntimeException e) { return OpOutcome.error( Math.max(0, KvOperationService.currentAttempts() - 1), - classifyError(e) + classifyError(e), e ); } } @@ -103,7 +103,7 @@ public OpOutcome write(Row row, int timeoutMs) { } catch (RuntimeException e) { return OpOutcome.error( Math.max(0, KvOperationService.currentAttempts() - 1), - classifyError(e) + classifyError(e), e ); } }