Add an extra delete mutation during rebuild for CDC index

haridsv · Jan 22, 2024 · 93d586e · 93d586e
1 parent da6ddad
commit 93d586e
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 3 deletions.
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/GlobalIndexRegionScanner.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/GlobalIndexRegionScanner.java
@@ -41,6 +41,7 @@
 import org.apache.phoenix.execute.MutationState;
 import org.apache.phoenix.filter.EmptyColumnOnlyFilter;
 import org.apache.phoenix.filter.PagingFilter;
+import org.apache.phoenix.hbase.index.IndexRegionObserver;
 import org.apache.phoenix.hbase.index.ValueGetter;
 import org.apache.phoenix.compile.ScanRanges;
 import org.apache.phoenix.filter.AllVersionsIndexRebuildFilter;
@@ -69,6 +70,7 @@
 import org.apache.phoenix.thirdparty.com.google.common.annotations.VisibleForTesting;
 import org.apache.phoenix.thirdparty.com.google.common.collect.Lists;
 import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
+import org.apache.phoenix.util.ByteUtil;
 import org.apache.phoenix.util.EnvironmentEdgeManager;
 import org.apache.phoenix.util.IndexUtil;
 import org.apache.phoenix.util.ServerUtil;
@@ -1351,6 +1353,17 @@ public static List<Mutation> prepareIndexMutationsForRebuild(IndexMaintainer ind
                                 indexMaintainer.buildRowDeleteMutation(indexRowKeyForCurrentDataRow,
                                         IndexMaintainer.DeleteType.ALL_VERSIONS, ts);
                         indexMutations.add(del);
+                        if (indexMaintainer.isCDCIndex()) {
+                            // CDC Index needs two delete markers one for deleting the index row,
+                            // and the other for referencing the data table delete mutation with
+                            // the right index row key, that is, the index row key starting with ts
+                            Put cdcDataRowState = new Put(currentDataRowState.getRow());
+                            cdcDataRowState.addColumn(indexMaintainer.getDataEmptyKeyValueCF(),
+                                    indexMaintainer.getEmptyKeyValueQualifierForDataTable(), ts,
+                                    ByteUtil.EMPTY_BYTE_ARRAY);
+                            indexMutations.add(IndexRegionObserver.getDeleteIndexMutation(
+                                    currentDataRowState, indexMaintainer, ts, rowKeyPtr));
+                        }
                     }
                     currentDataRowState = null;
                     indexRowKeyForCurrentDataRow = null;

diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
@@ -899,10 +899,10 @@ private void getCurrentRowStates(ObserverContext<RegionCoprocessorEnvironment> c
             }
         }
     }
-    private Mutation getDeleteIndexMutation(Put dataRowState, IndexMaintainer indexMaintainer,
+    public static Mutation getDeleteIndexMutation(Put dataRowState, IndexMaintainer indexMaintainer,
             long ts, ImmutableBytesPtr rowKeyPtr) {
-        ValueGetter cdcDataRowVG = new IndexUtil.SimpleValueGetter(dataRowState);
-        byte[] indexRowKey = indexMaintainer.buildRowKey(cdcDataRowVG, rowKeyPtr, null, null, ts);
+        ValueGetter dataRowVG = new IndexUtil.SimpleValueGetter(dataRowState);
+        byte[] indexRowKey = indexMaintainer.buildRowKey(dataRowVG, rowKeyPtr, null, null, ts);
         return indexMaintainer.buildRowDeleteMutation(indexRowKey,
                 IndexMaintainer.DeleteType.ALL_VERSIONS, ts);
     }
@@ -971,6 +971,9 @@ private void prepareIndexMutations(BatchMutateContext context, List<IndexMaintai
                             new Pair<Mutation, byte[]>(getDeleteIndexMutation(currentDataRowState,
                                     indexMaintainer, ts, rowKeyPtr), rowKeyPtr.get()));
                     if (indexMaintainer.isCDCIndex()) {
+                        // CDC Index needs two delete markers one for deleting the index row, and
+                        // the other for referencing the data table delete mutation with the
+                        // right index row key, that is, the index row key starting with ts
                         Put cdcDataRowState = new Put(currentDataRowState.getRow());
                         cdcDataRowState.addColumn(indexMaintainer.getDataEmptyKeyValueCF(),
                                 indexMaintainer.getEmptyKeyValueQualifierForDataTable(), ts,