From 05453ff7761a0fee32be8475d380a6c74bf4e5b0 Mon Sep 17 00:00:00 2001 From: Roman Khachatryan Date: Tue, 27 Oct 2026 15:50:29 +0100 Subject: [PATCH 1/2] [FLINK-38199][benchmark] Add SinkUpsertMaterializerBenchmark Squashed from apache/flink-benchmarks#106 and rebased onto master, with spotless and checkstyle fixes applied. --- pom.xml | 7 + .../SinkUpsertMaterializerBenchmark.java | 318 ++++++++++++++++++ 2 files changed, 325 insertions(+) create mode 100644 src/main/java/org/apache/flink/benchmark/SinkUpsertMaterializerBenchmark.java diff --git a/pom.xml b/pom.xml index fc06d98..36db2cd 100644 --- a/pom.xml +++ b/pom.xml @@ -140,6 +140,13 @@ under the License. test-jar + + org.apache.flink + flink-table-runtime + ${flink.version} + test-jar + + org.apache.flink flink-test-utils-junit diff --git a/src/main/java/org/apache/flink/benchmark/SinkUpsertMaterializerBenchmark.java b/src/main/java/org/apache/flink/benchmark/SinkUpsertMaterializerBenchmark.java new file mode 100644 index 0000000..4709817 --- /dev/null +++ b/src/main/java/org/apache/flink/benchmark/SinkUpsertMaterializerBenchmark.java @@ -0,0 +1,318 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 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 + * + * http://www.apache.org/licenses/LICENSE-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 org.apache.flink.benchmark; + +import org.apache.flink.api.common.state.StateTtlConfig; +import org.apache.flink.runtime.state.StateBackend; +import org.apache.flink.runtime.state.hashmap.HashMapStateBackend; +import org.apache.flink.state.rocksdb.EmbeddedRocksDBStateBackend; +import org.apache.flink.streaming.api.operators.OneInputStreamOperator; +import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.runtime.generated.GeneratedRecordEqualiser; +import org.apache.flink.table.runtime.generated.HashFunction; +import org.apache.flink.table.runtime.generated.RecordEqualiser; +import org.apache.flink.table.runtime.keyselector.RowDataKeySelector; +import org.apache.flink.table.runtime.operators.multipleinput.output.BlackHoleOutput; +import org.apache.flink.table.runtime.operators.sink.SinkUpsertMaterializer; +import org.apache.flink.table.runtime.typeutils.InternalSerializers; +import org.apache.flink.table.runtime.util.StateConfigUtil; +import org.apache.flink.table.types.logical.BigIntType; +import org.apache.flink.table.types.logical.IntType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.table.types.logical.VarCharType; +import org.apache.flink.table.utils.HandwrittenSelectorUtil; + +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.OperationsPerInvocation; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.TearDown; +import org.openjdk.jmh.runner.Runner; +import org.openjdk.jmh.runner.RunnerException; +import org.openjdk.jmh.runner.options.OptionsBuilder; +import org.openjdk.jmh.runner.options.VerboseMode; + +import java.nio.charset.StandardCharsets; +import java.util.Objects; +import java.util.Random; + +import static org.apache.flink.table.runtime.util.StreamRecordUtils.deleteRecord; +import static org.apache.flink.table.runtime.util.StreamRecordUtils.insertRecord; +import static org.apache.flink.util.Preconditions.checkState; + +/** + * Benchmark for {@link SinkUpsertMaterializer}. + * + *

It inserts a fixed number of records (inserts and deletes) in a loop and measures the time it + * takes to handle all the records. It uses a test harness, not the full Flink job. + * + *

The benchmark has the following parameters: + * + *

+ * + *

Results on M3Pro: + * + * 

{@code
+ * Benchmark                            (hasUpsertKey)  (payloadSize)  (retractDelay)  (retractPercentage)  (stateBackend)  (sumVersion)   Mode  Cnt     Score   Error   Units
+ * SinkUpsertMaterializerBenchmark.run           false            100               1                  100            HEAP            V1  thrpt       4138.808          ops/ms
+ * SinkUpsertMaterializerBenchmark.run           false            100               1                  100         ROCKSDB            V1  thrpt        284.055          ops/ms
+ * SinkUpsertMaterializerBenchmark.run           false            100              10                  100            HEAP            V1  thrpt       3729.824          ops/ms
+ * SinkUpsertMaterializerBenchmark.run           false            100              10                  100         ROCKSDB            V1  thrpt        205.047          ops/ms
+ * SinkUpsertMaterializerBenchmark.run           false            100             100                  100            HEAP            V1  thrpt       4137.591          ops/ms
+ * SinkUpsertMaterializerBenchmark.run           false            100             100                  100         ROCKSDB            V1  thrpt         80.406          ops/ms
+ * SinkUpsertMaterializerBenchmark.run           false            100             200                  100            HEAP            V1  thrpt       1886.574          ops/ms
+ * SinkUpsertMaterializerBenchmark.run           false            100             200                  100         ROCKSDB            V1  thrpt         30.935          ops/ms
+ * SinkUpsertMaterializerBenchmark.run           false            100            1000                  100            HEAP            V1  thrpt        546.826          ops/ms
+ * SinkUpsertMaterializerBenchmark.run           false            100            1000                  100         ROCKSDB            V1  thrpt          7.081          ops/ms
+ * SinkUpsertMaterializerBenchmark.run            true            100               1                  100            HEAP            V1  thrpt       4006.263          ops/ms
+ * SinkUpsertMaterializerBenchmark.run            true            100               1                  100         ROCKSDB            V1  thrpt        297.556          ops/ms
+ * SinkUpsertMaterializerBenchmark.run            true            100              10                  100            HEAP            V1  thrpt       3240.089          ops/ms
+ * SinkUpsertMaterializerBenchmark.run            true            100              10                  100         ROCKSDB            V1  thrpt        209.375          ops/ms
+ * SinkUpsertMaterializerBenchmark.run            true            100             100                  100            HEAP            V1  thrpt       2131.445          ops/ms
+ * SinkUpsertMaterializerBenchmark.run            true            100             100                  100         ROCKSDB            V1  thrpt         78.209          ops/ms
+ * SinkUpsertMaterializerBenchmark.run            true            100             200                  100            HEAP            V1  thrpt        652.936          ops/ms
+ * SinkUpsertMaterializerBenchmark.run            true            100             200                  100         ROCKSDB            V1  thrpt         29.674          ops/ms
+ * SinkUpsertMaterializerBenchmark.run            true            100            1000                  100            HEAP            V1  thrpt        118.567          ops/ms
+ * SinkUpsertMaterializerBenchmark.run            true            100            1000                  100         ROCKSDB            V1  thrpt          6.426          ops/ms
+ * }
+ */ +@OperationsPerInvocation(value = SinkUpsertMaterializerBenchmark.RECORDS_PER_INVOCATION) +@SuppressWarnings("ConstantValue") +public class SinkUpsertMaterializerBenchmark extends BenchmarkBase { + + private static final int STREAM_KEY = 0; + + public static void main(String[] args) throws RunnerException { + new Runner( + new OptionsBuilder() + .verbosity(VerboseMode.NORMAL) + .include( + ".*" + + SinkUpsertMaterializerBenchmark.class + .getCanonicalName() + + ".*") + .build()) + .run(); + } + + @Benchmark + public void run(SumBmState state) throws Exception { + for (long record = 0; record < state.numRecordsTotal; record++) { + state.harness.processElement(insertRecord(record, STREAM_KEY, state.payload)); + if (state.shouldRetract(record)) { + state.harness.processElement( + deleteRecord(record - state.retractOffset, STREAM_KEY, state.payload)); + } + } + } + + protected static final int RECORDS_PER_INVOCATION = 10_000; + + @State(Scope.Thread) + public static class SumBmState { + + @Param({"false", "true"}) + public boolean hasUpsertKey; + + @Param({"HEAP", "ROCKSDB"}) + public SumStateBackend stateBackend; + + public int numRecordsTotal; + + // larger payload amplifies any inefficiencies but slows down the benchmark; mostly affects + // rocksdb + @Param({"20"}) + public int payloadSize; + + // lower retraction percentage implies longer history, making retractions even harder + // (unless percentage = 0) + @Param("100") + public int retractPercentage; + + // higher retraction delay leaves longer history, making retractions even harder (unless + // percentage = 0) + // for automated runs, reduce the run time (and the data points) to the most common cases + @Param({"1", "1000"}) + // for comparison, the following values might be useful: + // @Param({"1", "10", "100", "200", "1000"}) + public int retractDelay; + + // the lower the value, the closer to the end of the list is the element to retract, the + // harder for V1 to find the element + public long retractOffset; + + public String payload; + + public KeyedOneInputStreamOperatorTestHarness harness; + + @Setup(Level.Invocation) + public void initSumBmState() throws Exception { + harness = getHarness(createSum(hasUpsertKey), stateBackend); + payload = generatePayload(payloadSize); + numRecordsTotal = RECORDS_PER_INVOCATION; + retractOffset = (1 + retractDelay) / 2; + checkState(numRecordsTotal > retractDelay); + checkState(retractPercentage >= 0 && retractPercentage <= 100); + } + + @TearDown(Level.Invocation) + public void teardown() throws Exception { + this.harness.close(); + this.harness = null; + } + + public boolean shouldRetract(long record) { + return retractEnabled() && (retractEverything() || retractRecord(record)); + } + + private boolean retractEnabled() { + return retractPercentage > 0; + } + + private boolean retractEverything() { + return retractPercentage == 100; + } + + private boolean retractRecord(long index) { + return index >= retractDelay && index % 100 < retractPercentage; + } + } + + private static String generatePayload(int size) { + final byte[] bytes = new byte[size]; + new Random().nextBytes(bytes); + return new String(bytes, StandardCharsets.UTF_8); + } + + private static final int UPSERT_KEY_POS = 0; + private static final int STREAM_KEY_POS = 1; + private static final int PAYLOAD_POS = 2; + private static final LogicalType[] types = + new LogicalType[] {new BigIntType(), new IntType(), new VarCharType()}; + + private static KeyedOneInputStreamOperatorTestHarness getHarness( + OneInputStreamOperator materializer, SumStateBackend stateBackend) + throws Exception { + RowDataKeySelector rowDataSelector = + HandwrittenSelectorUtil.getRowDataSelector(new int[] {STREAM_KEY_POS}, types); + KeyedOneInputStreamOperatorTestHarness testHarness = + new KeyedOneInputStreamOperatorTestHarness<>( + materializer, rowDataSelector, rowDataSelector.getProducedType()); + testHarness.getExecutionConfig().setMaxParallelism(2048); + testHarness.setStateBackend(stateBackend.create(true)); + testHarness.setOutputCreator(ign -> new BlackHoleOutput()); // requires change in Flink + testHarness.open(); + return testHarness; + } + + private static OneInputStreamOperator createSum(boolean hasUpsertKey) { + StateTtlConfig ttlConfig = StateConfigUtil.createTtlConfig(0); // no ttl + RowType physicalRowType = RowType.of(types); + int[] inputUpsertKey = hasUpsertKey ? new int[] {UPSERT_KEY_POS} : null; + return new SinkUpsertMaterializer( + ttlConfig, + InternalSerializers.create(physicalRowType), + equalizer(), + hasUpsertKey ? upsertEqualizer() : null, + inputUpsertKey); + } + + private static class TestRecordEqualiser implements RecordEqualiser, HashFunction { + @Override + public boolean equals(RowData row1, RowData row2) { + return row1.getRowKind() == row2.getRowKind() + && row1.getLong(UPSERT_KEY_POS) == row2.getLong(UPSERT_KEY_POS) + && row1.getInt(STREAM_KEY_POS) == row2.getInt(STREAM_KEY_POS) + && row1.getString(PAYLOAD_POS).equals(row2.getString(PAYLOAD_POS)); + } + + @Override + public int hashCode(Object data) { + RowData rd = (RowData) data; + return Objects.hash( + rd.getLong(UPSERT_KEY_POS), + rd.getInt(STREAM_KEY_POS), + rd.getString(PAYLOAD_POS)); + } + } + + private static class TestUpsertKeyEqualiser implements RecordEqualiser, HashFunction { + @Override + public boolean equals(RowData row1, RowData row2) { + return row1.getRowKind() == row2.getRowKind() + && row1.getLong(UPSERT_KEY_POS) == row2.getLong(UPSERT_KEY_POS); + } + + @Override + public int hashCode(Object data) { + return Long.hashCode(((RowData) data).getLong(UPSERT_KEY_POS)); + } + } + + private static GeneratedRecordEqualiser equalizer() { + return new GeneratedRecordEqualiser("", "", new Object[0]) { + + @Override + public RecordEqualiser newInstance(ClassLoader classLoader) { + return new TestRecordEqualiser(); + } + }; + } + + private static GeneratedRecordEqualiser upsertEqualizer() { + return new GeneratedRecordEqualiser("", "", new Object[0]) { + + @Override + public RecordEqualiser newInstance(ClassLoader classLoader) { + return new TestUpsertKeyEqualiser(); + } + }; + } + + public enum SumStateBackend { + HEAP { + + public StateBackend create(boolean incrementalIfSupported) { + return new HashMapStateBackend(); + } + }, + ROCKSDB { + + public StateBackend create(boolean incrementalIfSupported) { + return new EmbeddedRocksDBStateBackend(incrementalIfSupported); + } + }; + + public abstract StateBackend create(boolean incrementalIfSupported); + } +} From 781379b2385273656f7f5dd3eff939d669af9381 Mon Sep 17 00:00:00 2001 From: Martijn Visser <2989614+martijnvisser@users.noreply.github.com> Date: Tue, 27 Oct 2026 15:54:23 +0100 Subject: [PATCH 2/2] [FLINK-38199][benchmark] Benchmark SinkUpsertMaterializerV2 and measure steady state Build the history in the iteration setup and have each invocation insert one row and retract the middle row, so the measured method no longer creates a harness and the history length stays constant. Add the table.exec.sink.upsert-materialize-strategy.type strategies as a parameter and replace retractDelay and retractPercentage with historyLength. Generated-by: Claude Code (Claude Opus 5.5) --- .../SinkUpsertMaterializerBenchmark.java | 258 +++++++++--------- 1 file changed, 126 insertions(+), 132 deletions(-) diff --git a/src/main/java/org/apache/flink/benchmark/SinkUpsertMaterializerBenchmark.java b/src/main/java/org/apache/flink/benchmark/SinkUpsertMaterializerBenchmark.java index 4709817..1dd03f1 100644 --- a/src/main/java/org/apache/flink/benchmark/SinkUpsertMaterializerBenchmark.java +++ b/src/main/java/org/apache/flink/benchmark/SinkUpsertMaterializerBenchmark.java @@ -19,19 +19,23 @@ package org.apache.flink.benchmark; import org.apache.flink.api.common.state.StateTtlConfig; +import org.apache.flink.configuration.Configuration; import org.apache.flink.runtime.state.StateBackend; import org.apache.flink.runtime.state.hashmap.HashMapStateBackend; import org.apache.flink.state.rocksdb.EmbeddedRocksDBStateBackend; +import org.apache.flink.streaming.api.TimeDomain; import org.apache.flink.streaming.api.operators.OneInputStreamOperator; import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness; import org.apache.flink.table.data.RowData; +import org.apache.flink.table.runtime.generated.GeneratedHashFunction; import org.apache.flink.table.runtime.generated.GeneratedRecordEqualiser; import org.apache.flink.table.runtime.generated.HashFunction; import org.apache.flink.table.runtime.generated.RecordEqualiser; import org.apache.flink.table.runtime.keyselector.RowDataKeySelector; import org.apache.flink.table.runtime.operators.multipleinput.output.BlackHoleOutput; import org.apache.flink.table.runtime.operators.sink.SinkUpsertMaterializer; -import org.apache.flink.table.runtime.typeutils.InternalSerializers; +import org.apache.flink.table.runtime.operators.sink.SinkUpsertMaterializerV2; +import org.apache.flink.table.runtime.sequencedmultisetstate.SequencedMultiSetStateConfig; import org.apache.flink.table.runtime.util.StateConfigUtil; import org.apache.flink.table.types.logical.BigIntType; import org.apache.flink.table.types.logical.IntType; @@ -39,6 +43,7 @@ import org.apache.flink.table.types.logical.RowType; import org.apache.flink.table.types.logical.VarCharType; import org.apache.flink.table.utils.HandwrittenSelectorUtil; +import org.apache.flink.util.StringUtils; import org.openjdk.jmh.annotations.Benchmark; import org.openjdk.jmh.annotations.Level; @@ -48,172 +53,93 @@ import org.openjdk.jmh.annotations.Setup; import org.openjdk.jmh.annotations.State; import org.openjdk.jmh.annotations.TearDown; -import org.openjdk.jmh.runner.Runner; -import org.openjdk.jmh.runner.RunnerException; -import org.openjdk.jmh.runner.options.OptionsBuilder; -import org.openjdk.jmh.runner.options.VerboseMode; -import java.nio.charset.StandardCharsets; import java.util.Objects; import java.util.Random; import static org.apache.flink.table.runtime.util.StreamRecordUtils.deleteRecord; import static org.apache.flink.table.runtime.util.StreamRecordUtils.insertRecord; -import static org.apache.flink.util.Preconditions.checkState; /** - * Benchmark for {@link SinkUpsertMaterializer}. + * Benchmark for {@link SinkUpsertMaterializer} and {@link SinkUpsertMaterializerV2}. * - *

It inserts a fixed number of records (inserts and deletes) in a loop and measures the time it - * takes to handle all the records. It uses a test harness, not the full Flink job. + *

It uses a test harness, not the full Flink job. All records share the same stream key. The + * setup fills the state of that key with {@code historyLength} rows; each invocation then inserts + * one row and retracts the row in the middle of the history, so the history length stays constant + * during the measurement. * *

The benchmark has the following parameters: * *

    - *
  • total number of records - fixed to 10K; all records are inserted under the same stream key + *
  • strategy: LEGACY/VALUE/MAP/ADAPTIVE, as in {@code + * table.exec.sink.upsert-materialize-strategy.type}; LEGACY is {@link + * SinkUpsertMaterializer}, the others are {@link SinkUpsertMaterializerV2} with the + * corresponding {@link SequencedMultiSetStateConfig} *
  • hasUpsertKey: true/false - *
  • stateBackend type: HEAP/RocksDB - *
  • payload size - fixed to 100 bytes; bigger payload degrades faster on RocksDB - *
  • retractPercentage, retractDelay - how many records to retract, and from which record to - * start retracting; both parameters control how frequently retraction happens and how long - * the history is. Retraction is performed from the middle of the list. Retraction is - * inefficient in the current implementation on long lists. + *
  • stateBackend: HEAP/ROCKSDB + *
  • historyLength: number of rows kept for the stream key; longer histories make retractions + * more expensive + *
  • payloadSize: fixed to 20; a bigger payload amplifies inefficiencies, mostly on RocksDB *
- * - *

Results on M3Pro: - * - * 

{@code
- * Benchmark                            (hasUpsertKey)  (payloadSize)  (retractDelay)  (retractPercentage)  (stateBackend)  (sumVersion)   Mode  Cnt     Score   Error   Units
- * SinkUpsertMaterializerBenchmark.run           false            100               1                  100            HEAP            V1  thrpt       4138.808          ops/ms
- * SinkUpsertMaterializerBenchmark.run           false            100               1                  100         ROCKSDB            V1  thrpt        284.055          ops/ms
- * SinkUpsertMaterializerBenchmark.run           false            100              10                  100            HEAP            V1  thrpt       3729.824          ops/ms
- * SinkUpsertMaterializerBenchmark.run           false            100              10                  100         ROCKSDB            V1  thrpt        205.047          ops/ms
- * SinkUpsertMaterializerBenchmark.run           false            100             100                  100            HEAP            V1  thrpt       4137.591          ops/ms
- * SinkUpsertMaterializerBenchmark.run           false            100             100                  100         ROCKSDB            V1  thrpt         80.406          ops/ms
- * SinkUpsertMaterializerBenchmark.run           false            100             200                  100            HEAP            V1  thrpt       1886.574          ops/ms
- * SinkUpsertMaterializerBenchmark.run           false            100             200                  100         ROCKSDB            V1  thrpt         30.935          ops/ms
- * SinkUpsertMaterializerBenchmark.run           false            100            1000                  100            HEAP            V1  thrpt        546.826          ops/ms
- * SinkUpsertMaterializerBenchmark.run           false            100            1000                  100         ROCKSDB            V1  thrpt          7.081          ops/ms
- * SinkUpsertMaterializerBenchmark.run            true            100               1                  100            HEAP            V1  thrpt       4006.263          ops/ms
- * SinkUpsertMaterializerBenchmark.run            true            100               1                  100         ROCKSDB            V1  thrpt        297.556          ops/ms
- * SinkUpsertMaterializerBenchmark.run            true            100              10                  100            HEAP            V1  thrpt       3240.089          ops/ms
- * SinkUpsertMaterializerBenchmark.run            true            100              10                  100         ROCKSDB            V1  thrpt        209.375          ops/ms
- * SinkUpsertMaterializerBenchmark.run            true            100             100                  100            HEAP            V1  thrpt       2131.445          ops/ms
- * SinkUpsertMaterializerBenchmark.run            true            100             100                  100         ROCKSDB            V1  thrpt         78.209          ops/ms
- * SinkUpsertMaterializerBenchmark.run            true            100             200                  100            HEAP            V1  thrpt        652.936          ops/ms
- * SinkUpsertMaterializerBenchmark.run            true            100             200                  100         ROCKSDB            V1  thrpt         29.674          ops/ms
- * SinkUpsertMaterializerBenchmark.run            true            100            1000                  100            HEAP            V1  thrpt        118.567          ops/ms
- * SinkUpsertMaterializerBenchmark.run            true            100            1000                  100         ROCKSDB            V1  thrpt          6.426          ops/ms
- * }
*/ -@OperationsPerInvocation(value = SinkUpsertMaterializerBenchmark.RECORDS_PER_INVOCATION) -@SuppressWarnings("ConstantValue") public class SinkUpsertMaterializerBenchmark extends BenchmarkBase { private static final int STREAM_KEY = 0; - public static void main(String[] args) throws RunnerException { - new Runner( - new OptionsBuilder() - .verbosity(VerboseMode.NORMAL) - .include( - ".*" - + SinkUpsertMaterializerBenchmark.class - .getCanonicalName() - + ".*") - .build()) - .run(); - } - @Benchmark - public void run(SumBmState state) throws Exception { - for (long record = 0; record < state.numRecordsTotal; record++) { - state.harness.processElement(insertRecord(record, STREAM_KEY, state.payload)); - if (state.shouldRetract(record)) { - state.harness.processElement( - deleteRecord(record - state.retractOffset, STREAM_KEY, state.payload)); - } - } + @OperationsPerInvocation(2) + public void sinkUpsertMaterializer(SumBmState state) throws Exception { + long record = state.nextRecord++; + state.harness.processElement(insertRecord(record, STREAM_KEY, state.payload)); + state.harness.processElement( + deleteRecord(record - state.retractOffset, STREAM_KEY, state.payload)); } - protected static final int RECORDS_PER_INVOCATION = 10_000; - @State(Scope.Thread) public static class SumBmState { + @Param({"LEGACY", "VALUE", "MAP", "ADAPTIVE"}) + public SumStrategy strategy; + @Param({"false", "true"}) public boolean hasUpsertKey; @Param({"HEAP", "ROCKSDB"}) public SumStateBackend stateBackend; - public int numRecordsTotal; + @Param({"1", "500"}) + public int historyLength; - // larger payload amplifies any inefficiencies but slows down the benchmark; mostly affects - // rocksdb @Param({"20"}) public int payloadSize; - // lower retraction percentage implies longer history, making retractions even harder - // (unless percentage = 0) - @Param("100") - public int retractPercentage; - - // higher retraction delay leaves longer history, making retractions even harder (unless - // percentage = 0) - // for automated runs, reduce the run time (and the data points) to the most common cases - @Param({"1", "1000"}) - // for comparison, the following values might be useful: - // @Param({"1", "10", "100", "200", "1000"}) - public int retractDelay; - - // the lower the value, the closer to the end of the list is the element to retract, the - // harder for V1 to find the element - public long retractOffset; - public String payload; - public KeyedOneInputStreamOperatorTestHarness harness; - - @Setup(Level.Invocation) - public void initSumBmState() throws Exception { - harness = getHarness(createSum(hasUpsertKey), stateBackend); - payload = generatePayload(payloadSize); - numRecordsTotal = RECORDS_PER_INVOCATION; - retractOffset = (1 + retractDelay) / 2; - checkState(numRecordsTotal > retractDelay); - checkState(retractPercentage >= 0 && retractPercentage <= 100); - } - - @TearDown(Level.Invocation) - public void teardown() throws Exception { - this.harness.close(); - this.harness = null; - } + // the rows inserted before the first retracted one are never retracted, which keeps the + // retracted row in the middle of the history + public long retractOffset; - public boolean shouldRetract(long record) { - return retractEnabled() && (retractEverything() || retractRecord(record)); - } + public long nextRecord; - private boolean retractEnabled() { - return retractPercentage > 0; - } + public KeyedOneInputStreamOperatorTestHarness harness; - private boolean retractEverything() { - return retractPercentage == 100; + @Setup(Level.Iteration) + public void setUp() throws Exception { + harness = getHarness(strategy.create(hasUpsertKey), stateBackend); + payload = StringUtils.generateRandomAlphanumericString(new Random(42), payloadSize); + retractOffset = (historyLength + 1) / 2; + for (nextRecord = 0; nextRecord < historyLength; nextRecord++) { + harness.processElement(insertRecord(nextRecord, STREAM_KEY, payload)); + } } - private boolean retractRecord(long index) { - return index >= retractDelay && index % 100 < retractPercentage; + @TearDown(Level.Iteration) + public void tearDown() throws Exception { + harness.close(); + harness = null; } } - private static String generatePayload(int size) { - final byte[] bytes = new byte[size]; - new Random().nextBytes(bytes); - return new String(bytes, StandardCharsets.UTF_8); - } - private static final int UPSERT_KEY_POS = 0; private static final int STREAM_KEY_POS = 1; private static final int PAYLOAD_POS = 2; @@ -230,21 +156,69 @@ private static KeyedOneInputStreamOperatorTestHarness materializer, rowDataSelector, rowDataSelector.getProducedType()); testHarness.getExecutionConfig().setMaxParallelism(2048); testHarness.setStateBackend(stateBackend.create(true)); - testHarness.setOutputCreator(ign -> new BlackHoleOutput()); // requires change in Flink + testHarness.setOutputCreator(ign -> new BlackHoleOutput()); testHarness.open(); return testHarness; } - private static OneInputStreamOperator createSum(boolean hasUpsertKey) { - StateTtlConfig ttlConfig = StateConfigUtil.createTtlConfig(0); // no ttl - RowType physicalRowType = RowType.of(types); - int[] inputUpsertKey = hasUpsertKey ? new int[] {UPSERT_KEY_POS} : null; - return new SinkUpsertMaterializer( - ttlConfig, - InternalSerializers.create(physicalRowType), - equalizer(), - hasUpsertKey ? upsertEqualizer() : null, - inputUpsertKey); + public enum SumStrategy { + LEGACY { + @Override + OneInputStreamOperator create(boolean hasUpsertKey) { + return SinkUpsertMaterializer.create( + TTL_CONFIG, + RowType.of(types), + equalizer(), + hasUpsertKey ? upsertEqualizer() : null, + upsertKey(hasUpsertKey)); + } + }, + VALUE { + @Override + OneInputStreamOperator create(boolean hasUpsertKey) { + return createV2( + hasUpsertKey, + SequencedMultiSetStateConfig.forValue(TimeDomain.EVENT_TIME, TTL_CONFIG)); + } + }, + MAP { + @Override + OneInputStreamOperator create(boolean hasUpsertKey) { + return createV2( + hasUpsertKey, + SequencedMultiSetStateConfig.forMap(TimeDomain.EVENT_TIME, TTL_CONFIG)); + } + }, + ADAPTIVE { + @Override + OneInputStreamOperator create(boolean hasUpsertKey) { + return createV2( + hasUpsertKey, + SequencedMultiSetStateConfig.adaptive( + TimeDomain.EVENT_TIME, null, null, TTL_CONFIG)); + } + }; + + // no ttl, V2 does not support it yet (FLINK-38463) + private static final StateTtlConfig TTL_CONFIG = StateConfigUtil.createTtlConfig(0); + + abstract OneInputStreamOperator create(boolean hasUpsertKey); + + private static OneInputStreamOperator createV2( + boolean hasUpsertKey, SequencedMultiSetStateConfig stateConfig) { + return SinkUpsertMaterializerV2.create( + RowType.of(types), + equalizer(), + hasUpsertKey ? upsertEqualizer() : null, + hashFunction(), + hasUpsertKey ? upsertHashFunction() : null, + upsertKey(hasUpsertKey), + stateConfig); + } + + private static int[] upsertKey(boolean hasUpsertKey) { + return hasUpsertKey ? new int[] {UPSERT_KEY_POS} : null; + } } private static class TestRecordEqualiser implements RecordEqualiser, HashFunction { @@ -299,6 +273,26 @@ public RecordEqualiser newInstance(ClassLoader classLoader) { }; } + private static GeneratedHashFunction hashFunction() { + return new GeneratedHashFunction("", "", new Object[0], new Configuration()) { + + @Override + public HashFunction newInstance(ClassLoader classLoader) { + return new TestRecordEqualiser(); + } + }; + } + + private static GeneratedHashFunction upsertHashFunction() { + return new GeneratedHashFunction("", "", new Object[0], new Configuration()) { + + @Override + public HashFunction newInstance(ClassLoader classLoader) { + return new TestUpsertKeyEqualiser(); + } + }; + } + public enum SumStateBackend { HEAP {