test: verify column family 49 moves

engine/src/test/java/io/camunda/zeebe/engine/state/migration/to_8_4/ColumnFamilyPrefixCorrectionMigrationTest.java

@@ -13,9 +13,11 @@
 import io.camunda.zeebe.db.impl.DbCompositeKey;
 import io.camunda.zeebe.db.impl.DbInt;
 import io.camunda.zeebe.db.impl.DbLong;
+import io.camunda.zeebe.db.impl.DbNil;
 import io.camunda.zeebe.db.impl.DbString;
 import io.camunda.zeebe.engine.state.message.DbMessageState;
 import io.camunda.zeebe.engine.state.migration.to_8_4.corrections.ColumnFamily48Corrector;
+import io.camunda.zeebe.engine.state.migration.to_8_4.corrections.ColumnFamily49Corrector;
 import io.camunda.zeebe.engine.state.mutable.MutableProcessingState;
 import io.camunda.zeebe.engine.util.ProcessingStateExtension;
 import io.camunda.zeebe.protocol.ZbColumnFamilies;
@@ -164,4 +166,128 @@ void shouldIgnoreDecisionKeyEntries() {
           .isEqualTo(123L);
     }
   }
+
+  /**
+   * Test correction from
+   * DEPRECATED_DMN_DECISION_REQUIREMENTS_KEY_BY_DECISION_REQUIREMENT_ID_AND_VERSION ->
+   * PROCESS_INSTANCE_KEY_BY_DEFINITION_KEY
+   */
+  @Nested
+  @ExtendWith(ProcessingStateExtension.class)
+  class ColumnFamily49CorrectorTestTest {
+    private ZeebeDb<ZbColumnFamilies> zeebeDb;
+    private MutableProcessingState processingState;
+    private TransactionContext transactionContext;
+
+    private ColumnFamily49Corrector sut;
+
+    private DbLong elementInstanceKey;
+    private DbLong processDefinitionKey;
+    private DbCompositeKey<DbLong, DbLong> processInstanceKeyByProcessDefinitionKey;
+
+    private ColumnFamily<DbCompositeKey<DbLong, DbLong>, DbNil> wrongPiKeyByProcDefKeyColumnFamily;
+    private ColumnFamily<DbCompositeKey<DbLong, DbLong>, DbNil>
+        correctPiKeyByProcDefKeyColumnFamily;
+
+    private DbString decisionRequirementsId;
+    private DbInt decisionRequirementsVersion;
+    private DbCompositeKey<DbString, DbInt> decisionRequirementsIdAndVersion;
+    private DbLong decisionRequirementsKey;
+
+    private ColumnFamily<DbCompositeKey<DbString, DbInt>, DbLong>
+        correctDecisionRequirementsKeyColumnFamily;
+
+    @BeforeEach
+    void setup() {
+      sut = new ColumnFamily49Corrector(zeebeDb, transactionContext);
+
+      elementInstanceKey = new DbLong();
+      processDefinitionKey = new DbLong();
+      processInstanceKeyByProcessDefinitionKey =
+          new DbCompositeKey<>(processDefinitionKey, elementInstanceKey);
+      wrongPiKeyByProcDefKeyColumnFamily =
+          zeebeDb.createColumnFamily(
+              ZbColumnFamilies
+                  .DEPRECATED_DMN_DECISION_REQUIREMENTS_KEY_BY_DECISION_REQUIREMENT_ID_AND_VERSION,
+              transactionContext,
+              processInstanceKeyByProcessDefinitionKey,
+              DbNil.INSTANCE);
+
+      correctPiKeyByProcDefKeyColumnFamily =
+          zeebeDb.createColumnFamily(
+              ZbColumnFamilies.PROCESS_INSTANCE_KEY_BY_DEFINITION_KEY,
+              transactionContext,
+              processInstanceKeyByProcessDefinitionKey,
+              DbNil.INSTANCE);
+
+      decisionRequirementsId = new DbString();
+      decisionRequirementsVersion = new DbInt();
+      decisionRequirementsIdAndVersion =
+          new DbCompositeKey<>(decisionRequirementsId, decisionRequirementsVersion);
+      decisionRequirementsKey = new DbLong();
+      correctDecisionRequirementsKeyColumnFamily =
+          zeebeDb.createColumnFamily(
+              ZbColumnFamilies
+                  .DEPRECATED_DMN_DECISION_REQUIREMENTS_KEY_BY_DECISION_REQUIREMENT_ID_AND_VERSION,
+              transactionContext,
+              decisionRequirementsIdAndVersion,
+              decisionRequirementsKey);
+    }
+
+    @Test
+    void shouldMovePiKeyByProcDefKeyToCorrectColumnFamily() {
+      // given
+      elementInstanceKey.wrapLong(123);
+      processDefinitionKey.wrapLong(456);
+      wrongPiKeyByProcDefKeyColumnFamily.insert(
+          processInstanceKeyByProcessDefinitionKey, DbNil.INSTANCE);
+
+      // when
+      sut.correctColumnFamilyPrefix();
+
+      // then
+      Assertions.assertThat(wrongPiKeyByProcDefKeyColumnFamily.isEmpty()).isTrue();
+      Assertions.assertThat(
+              correctPiKeyByProcDefKeyColumnFamily.exists(processInstanceKeyByProcessDefinitionKey))
+          .isTrue();
+    }
+
+    @Test
+    void shouldIgnoreProcessInstanceKeyByDefinitionKeyEntries() {
+      // given
+      decisionRequirementsId.wrapString("decisionRequirements");
+      decisionRequirementsVersion.wrapInt(1);
+      decisionRequirementsKey.wrapLong(543);
+      correctDecisionRequirementsKeyColumnFamily.insert(
+          decisionRequirementsIdAndVersion, decisionRequirementsKey);
+
+      elementInstanceKey.wrapLong(123);
+      processDefinitionKey.wrapLong(456);
+      wrongPiKeyByProcDefKeyColumnFamily.insert(
+          processInstanceKeyByProcessDefinitionKey, DbNil.INSTANCE);
+
+      decisionRequirementsId.wrapString("decisionRequirements2");
+      decisionRequirementsVersion.wrapInt(2);
+      decisionRequirementsKey.wrapLong(987);
+      correctDecisionRequirementsKeyColumnFamily.insert(
+          decisionRequirementsIdAndVersion, decisionRequirementsKey);
+
+      Assertions.assertThat(correctDecisionRequirementsKeyColumnFamily.count()).isEqualTo(3);
+
+      // when
+      sut.correctColumnFamilyPrefix();
+
+      // then
+      // we can no longer use wrongPiKeyByProcDefKeyColumnFamily.isEmpty() as there are entries in
+      // there just no longer process instance keys by process definition key entries, but we can
+      // simply count the entries
+      Assertions.assertThat(correctDecisionRequirementsKeyColumnFamily.count()).isEqualTo(2);
+
+      elementInstanceKey.wrapLong(123);
+      processDefinitionKey.wrapLong(456);
+      Assertions.assertThat(
+              correctPiKeyByProcDefKeyColumnFamily.exists(processInstanceKeyByProcessDefinitionKey))
+          .isTrue();
+    }
+  }
 }