Fix consensus subscription replay dedup

Author: VGalaxies

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java

@@ -2850,7 +2850,8 @@ public SettableFuture<ConfigTaskResult> createTopic(
         PipeDataNodeAgent.plugin()
             .validate(
                 "fakePipeName",
-                // TODO: currently use root to create topic
+                // Topic validation uses the admin name because topic creation has no user context
+                // here.
                 temporaryTopicMeta.generateExtractorAttributes(
                     CommonDescriptor.getInstance().getConfig().getDefaultAdminName()),
                 temporaryTopicMeta.generateProcessorAttributes(),

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/ProgressWALReader.java

@@ -70,6 +70,10 @@ public long getCurrentEntryLocalSeq() {
     return delegate.getCurrentEntryLocalSeq();
   }
 
+  public long getCurrentEntrySearchIndex() {
+    return delegate.getCurrentEntrySearchIndex();
+  }
+
   public int getCurrentEntryIndex() {
     return delegate.getCurrentEntryIndex();
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALByteBufReader.java

@@ -19,6 +19,7 @@
 
 package org.apache.iotdb.db.storageengine.dataregion.wal.io;
 
+import org.apache.iotdb.consensus.iot.log.ConsensusReqReader;
 import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry;
 
 import java.io.Closeable;
@@ -144,6 +145,14 @@ public long getCurrentEntryLocalSeq() {
     return metaData.getFirstSearchIndex() + currentEntryIndex;
   }
 
+  public long getCurrentEntrySearchIndex() {
+    final long firstSearchIndex = metaData.getFirstSearchIndex();
+    if (firstSearchIndex == ConsensusReqReader.DEFAULT_SEARCH_INDEX) {
+      return ConsensusReqReader.DEFAULT_SEARCH_INDEX;
+    }
+    return firstSearchIndex + currentEntryIndex;
+  }
+
   /** Returns the current entry index (0-based). */
   public int getCurrentEntryIndex() {
     return currentEntryIndex;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALFileUtils.java

@@ -306,6 +306,10 @@ public static long findSearchIndexAfterWriterProgress(
   }
 
   private interface SearchableRequestVisitor {
+    /**
+     * @return {@code true} to continue scanning the following searchable requests, {@code false} to
+     *     stop immediately.
+     */
     boolean onRequest(SearchableRequestMeta request);
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java

@@ -734,13 +734,7 @@ public void unbindPrefetchingQueue(final String consumerGroupId, final String to
           "Subscription: broker bound to consumer group [{}] does not exist", consumerGroupId);
       return;
     }
-    if (broker instanceof ConsensusSubscriptionBroker) {
-      ((ConsensusSubscriptionBroker) broker).unbindConsensusPrefetchingQueue(topicName);
-    } else if (broker instanceof SubscriptionBroker) {
-      ((SubscriptionBroker) broker).unbindPrefetchingQueue(topicName);
-    } else {
-      broker.removeQueue(topicName);
-    }
+    broker.unbind(topicName);
     prefetchingQueueCount.invalidate();
   }
 
@@ -789,11 +783,7 @@ public Map<String, Long> getConsensusLagSummary() {
     for (final Map.Entry<String, List<ISubscriptionBroker>> entry :
         consumerGroupIdToBrokers.entrySet()) {
       for (final ISubscriptionBroker broker : entry.getValue()) {
-        if (!(broker instanceof ConsensusSubscriptionBroker)) {
-          continue;
-        }
-        for (final Map.Entry<String, Long> lag :
-            ((ConsensusSubscriptionBroker) broker).getLagSummary().entrySet()) {
+        for (final Map.Entry<String, Long> lag : broker.getLagSummary().entrySet()) {
           result.put(entry.getKey() + "/" + lag.getKey(), lag.getValue());
         }
       }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ConsensusSubscriptionBroker.java

@@ -210,9 +210,7 @@ public List<SubscriptionEvent> pollTablets(
       return Collections.emptyList();
     }
 
-    final ConsensusPrefetchingQueue assignedQueue =
-        getAssignedQueueForConsumer(
-            queues, topicName, consumerId, commitContext.getRegionId(), "pollTablets");
+    final ConsensusPrefetchingQueue assignedQueue = getQueueForCommitContext(queues, commitContext);
     if (Objects.isNull(assignedQueue)) {
       return Collections.emptyList();
     }
@@ -245,8 +243,7 @@ public List<SubscriptionCommitContext> commit(
       }
 
       final ConsensusPrefetchingQueue assignedQueue =
-          getAssignedQueueForConsumer(
-              queues, topicName, consumerId, commitContext.getRegionId(), nack ? "nack" : "ack");
+          getQueueForCommitContext(queues, commitContext);
       boolean handled = false;
       if (Objects.nonNull(assignedQueue)) {
         final boolean success;
@@ -285,8 +282,7 @@ public int refreshInFlightEventLeases(
       }
 
       final ConsensusPrefetchingQueue assignedQueue =
-          getAssignedQueueForConsumer(
-              queues, topicName, consumerId, commitContext.getRegionId(), "refresh lease");
+          getQueueForCommitContext(queues, commitContext);
       if (Objects.nonNull(assignedQueue)
           && assignedQueue.refreshInFlightEventLease(consumerId, commitContext)) {
         refreshedCount++;
@@ -537,34 +533,19 @@ private List<ConsensusPrefetchingQueue> buildPollOrderForAssignedQueues(
     return orderedQueues;
   }
 
-  private ConsensusPrefetchingQueue getAssignedQueueForConsumer(
-      final List<ConsensusPrefetchingQueue> queues,
-      final String topicName,
-      final String consumerId,
-      final String regionId,
-      final String action) {
-    final TopicOwnershipSnapshot ownershipSnapshot =
-        refreshAndGetTopicOwnership(topicName, queues, consumerId);
+  private ConsensusPrefetchingQueue getQueueForCommitContext(
+      final List<ConsensusPrefetchingQueue> queues, final SubscriptionCommitContext commitContext) {
+    final String regionId = commitContext.getRegionId();
     for (final ConsensusPrefetchingQueue queue : queues) {
       if (queue.isClosed()) {
         continue;
       }
-      if (!regionId.isEmpty() && !regionId.equals(queue.getConsensusGroupId().toString())) {
+      if (Objects.nonNull(regionId)
+          && !regionId.isEmpty()
+          && !regionId.equals(queue.getConsensusGroupId().toString())) {
         continue;
       }
-      if (consumerId.equals(
-          ownershipSnapshot.getOwnerConsumerId(queue.getConsensusGroupId().toString()))) {
-        return queue;
-      }
-      LOGGER.debug(
-          "ConsensusSubscriptionBroker [{}]: consumer [{}] skipped {} on topic [{}], region [{}] is currently owned by [{}]",
-          brokerId,
-          consumerId,
-          action,
-          topicName,
-          queue.getConsensusGroupId(),
-          ownershipSnapshot.getOwnerConsumerId(queue.getConsensusGroupId().toString()));
-      return null;
+      return queue;
     }
     return null;
   }
@@ -657,6 +638,11 @@ public void unbindConsensusPrefetchingQueue(final String topicName) {
     closeAndRemoveConsensusPrefetchingQueues(topicName, true);
   }
 
+  @Override
+  public void unbind(final String topicName) {
+    unbindConsensusPrefetchingQueue(topicName);
+  }
+
   public int unbindByRegion(final ConsensusGroupId regionId) {
     final List<ConsensusPrefetchingQueue> queuesToClose = new ArrayList<>();
     final List<String> topicNamesToLog = new ArrayList<>();

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ISubscriptionBroker.java

@@ -79,6 +79,12 @@ default boolean acceptsCommitContext(final SubscriptionCommitContext commitConte
 
   int getQueueCount();
 
+  default Map<String, Long> getLagSummary() {
+    return Collections.emptyMap();
+  }
+
+  void unbind(String topicName);
+
   void removeQueue(String topicName);
 
   boolean isEmpty();

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionBroker.java

@@ -496,6 +496,11 @@ public void unbindPrefetchingQueue(final String topicName) {
         brokerId);
   }
 
+  @Override
+  public void unbind(final String topicName) {
+    unbindPrefetchingQueue(topicName);
+  }
+
   @Override
   public void removeQueue(final String topicName) {
     removePrefetchingQueue(topicName);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java

@@ -297,25 +297,17 @@ private List<Tablet> convertInsertTabletNode(final InsertTabletNode node) {
       schemas.add(new MeasurementSchema(measurements[colIdx], dataTypes[colIdx]));
     }
 
-    // Reuse complete tablets on the server-side read path. Column filtering changes tablet shape
-    // and still needs copies for the selected columns.
-    final long[] newTimes = allColumnsMatch ? times : Arrays.copyOf(times, rowCount);
+    // Column filtering changes only the tablet shape. The selected value arrays come from WAL
+    // InsertNodes and are reused by the subscription read path.
+    final long[] newTimes = times;
     final Object[] newColumns = new Object[columnCount];
     final BitMap[] newBitMaps = new BitMap[columnCount];
 
     for (int i = 0; i < columnCount; i++) {
       final int originalColIdx = allColumnsMatch ? i : matchedColumnIndices.get(i);
-      newColumns[i] =
-          allColumnsMatch
-              ? columns[originalColIdx]
-              : copyColumnArray(dataTypes[originalColIdx], columns[originalColIdx], rowCount);
+      newColumns[i] = columns[originalColIdx];
       if (bitMaps != null && bitMaps[originalColIdx] != null) {
-        if (allColumnsMatch) {
-          newBitMaps[i] = bitMaps[originalColIdx];
-        } else {
-          newBitMaps[i] = new BitMap(rowCount);
-          BitMap.copyOfRange(bitMaps[originalColIdx], 0, newBitMaps[i], 0, rowCount);
-        }
+        newBitMaps[i] = bitMaps[originalColIdx];
       }
     }
 
@@ -534,25 +526,17 @@ private List<Tablet> convertRelationalInsertTabletNode(final RelationalInsertTab
       columnTypes.add(toTsFileColumnCategory(node.getColumnCategories(), originalColIdx));
     }
 
-    // Reuse complete tablets on the server-side read path. Column filtering changes tablet shape
-    // and still needs copies for the selected columns.
-    final long[] newTimes = allColumnsMatch ? times : Arrays.copyOf(times, rowCount);
+    // Column filtering changes only the tablet shape. The selected value arrays come from WAL
+    // InsertNodes and are reused by the subscription read path.
+    final long[] newTimes = times;
     final Object[] newColumns = new Object[columnCount];
     final BitMap[] newBitMaps = new BitMap[columnCount];
 
     for (int colIdx = 0; colIdx < columnCount; colIdx++) {
       final int originalColIdx = allColumnsMatch ? colIdx : matchedColumnIndices.get(colIdx);
-      newColumns[colIdx] =
-          allColumnsMatch
-              ? columns[originalColIdx]
-              : copyColumnArray(dataTypes[originalColIdx], columns[originalColIdx], rowCount);
+      newColumns[colIdx] = columns[originalColIdx];
       if (bitMaps != null && bitMaps[originalColIdx] != null) {
-        if (allColumnsMatch) {
-          newBitMaps[colIdx] = bitMaps[originalColIdx];
-        } else {
-          newBitMaps[colIdx] = new BitMap(rowCount);
-          BitMap.copyOfRange(bitMaps[originalColIdx], 0, newBitMaps[colIdx], 0, rowCount);
-        }
+        newBitMaps[colIdx] = bitMaps[originalColIdx];
       }
     }
 
@@ -655,65 +639,6 @@ private ColumnCategory toTsFileColumnCategory(
         : ColumnCategory.FIELD;
   }
 
-  /**
-   * Bulk-copies a typed column array using System.arraycopy. Returns a new array of the same type
-   * containing the first {@code rowCount} elements.
-   */
-  private Object copyColumnArray(
-      final TSDataType dataType, final Object sourceColumn, final int rowCount) {
-    switch (dataType) {
-      case BOOLEAN:
-        {
-          final boolean[] src = (boolean[]) sourceColumn;
-          final boolean[] dst = new boolean[rowCount];
-          System.arraycopy(src, 0, dst, 0, rowCount);
-          return dst;
-        }
-      case INT32:
-      case DATE:
-        {
-          final int[] src = (int[]) sourceColumn;
-          final int[] dst = new int[rowCount];
-          System.arraycopy(src, 0, dst, 0, rowCount);
-          return dst;
-        }
-      case INT64:
-      case TIMESTAMP:
-        {
-          final long[] src = (long[]) sourceColumn;
-          final long[] dst = new long[rowCount];
-          System.arraycopy(src, 0, dst, 0, rowCount);
-          return dst;
-        }
-      case FLOAT:
-        {
-          final float[] src = (float[]) sourceColumn;
-          final float[] dst = new float[rowCount];
-          System.arraycopy(src, 0, dst, 0, rowCount);
-          return dst;
-        }
-      case DOUBLE:
-        {
-          final double[] src = (double[]) sourceColumn;
-          final double[] dst = new double[rowCount];
-          System.arraycopy(src, 0, dst, 0, rowCount);
-          return dst;
-        }
-      case TEXT:
-      case BLOB:
-      case STRING:
-        {
-          final Binary[] src = (Binary[]) sourceColumn;
-          final Binary[] dst = new Binary[rowCount];
-          System.arraycopy(src, 0, dst, 0, rowCount);
-          return dst;
-        }
-      default:
-        LOGGER.warn("Unsupported data type for bulk copy: {}", dataType);
-        return sourceColumn;
-    }
-  }
-
   /**
    * Adds a single value to the tablet at the specified position.
    *