fix(subscription): remove database semantics from payload api

Author: VGalaxies

iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionConsumer.java

@@ -724,7 +724,7 @@ private Optional<SubscriptionMessage> pollFile(
       // construct temporary message to nack
       nack(
           Collections.singletonList(
-              new SubscriptionMessage(commitContext, file.getAbsolutePath(), null)));
+              new SubscriptionMessage(commitContext, file.getAbsolutePath())));
       throw new SubscriptionRuntimeNonCriticalException(e.getMessage(), e);
     }
   }
@@ -874,8 +874,7 @@ private Optional<SubscriptionMessage> pollFileInternal(
 
             // generate subscription message
             inFlightFilesCommitContextSet.remove(commitContext);
-            return Optional.of(
-                new SubscriptionMessage(commitContext, file.getAbsolutePath(), null));
+            return Optional.of(new SubscriptionMessage(commitContext, file.getAbsolutePath()));
           }
         case ERROR:
           {
@@ -925,7 +924,7 @@ private Optional<SubscriptionMessage> pollTablets(
       // construct temporary message to nack
       nack(
           Collections.singletonList(
-              new SubscriptionMessage(response.getCommitContext(), Collections.emptyMap())));
+              new SubscriptionMessage(response.getCommitContext(), Collections.emptyList())));
       throw new SubscriptionRuntimeNonCriticalException(e.getMessage(), e);
     }
   }
@@ -946,8 +945,7 @@ private Optional<SubscriptionMessage> pollTabletsInternal(
           LOGGER.warn(errorMessage);
           throw new SubscriptionRuntimeNonCriticalException(errorMessage);
         }
-        return Optional.of(
-            new SubscriptionMessage(commitContext, Collections.singletonMap(null, tablets)));
+        return Optional.of(new SubscriptionMessage(commitContext, tablets));
       }
 
       timer.update();

iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionMessage.java

@@ -22,12 +22,10 @@
 import org.apache.iotdb.rpc.subscription.exception.SubscriptionIncompatibleHandlerException;
 import org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionCommitContext;
 
-import org.apache.thrift.annotation.Nullable;
 import org.apache.tsfile.write.record.Tablet;
 
 import java.util.Iterator;
 import java.util.List;
-import java.util.Map;
 import java.util.Objects;
 
 public class SubscriptionMessage implements Comparable<SubscriptionMessage> {
@@ -39,19 +37,17 @@ public class SubscriptionMessage implements Comparable<SubscriptionMessage> {
   private final SubscriptionMessageHandler handler;
 
   public SubscriptionMessage(
-      final SubscriptionCommitContext commitContext, final Map<String, List<Tablet>> tablets) {
+      final SubscriptionCommitContext commitContext, final List<Tablet> tablets) {
     this.commitContext = commitContext;
     this.messageType = SubscriptionMessageType.RECORD_HANDLER.getType();
     this.handler = new SubscriptionRecordHandler(tablets);
   }
 
   public SubscriptionMessage(
-      final SubscriptionCommitContext commitContext,
-      final String absolutePath,
-      @Nullable final String databaseName) {
+      final SubscriptionCommitContext commitContext, final String absolutePath) {
     this.commitContext = commitContext;
     this.messageType = SubscriptionMessageType.TS_FILE.getType();
-    this.handler = new SubscriptionTsFileHandler(absolutePath, databaseName);
+    this.handler = new SubscriptionTsFileHandler(absolutePath);
   }
 
   public SubscriptionCommitContext getCommitContext() {

iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionRecordHandler.java

@@ -21,7 +21,6 @@
 
 import org.apache.iotdb.isession.ISessionDataSet;
 
-import org.apache.thrift.annotation.Nullable;
 import org.apache.tsfile.enums.TSDataType;
 import org.apache.tsfile.read.common.Field;
 import org.apache.tsfile.read.common.RowRecord;
@@ -36,7 +35,6 @@
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
-import java.util.Map;
 import java.util.Objects;
 import java.util.stream.Collectors;
 
@@ -46,20 +44,13 @@ public class SubscriptionRecordHandler
 
   private final List<SubscriptionResultSet> resultSets;
 
-  public SubscriptionRecordHandler(final Map<String, List<Tablet>> tablets) {
+  public SubscriptionRecordHandler(final List<Tablet> tablets) {
     final List<SubscriptionResultSet> resultSets = new ArrayList<>();
-    for (final Map.Entry<String, List<Tablet>> entry : tablets.entrySet()) {
-      final String databaseName = entry.getKey();
-      final List<Tablet> tabletList = entry.getValue();
-      if (Objects.isNull(tabletList)) {
+    for (final Tablet tablet : tablets) {
+      if (Objects.isNull(tablet)) {
         continue;
       }
-      for (final Tablet tablet : tabletList) {
-        if (Objects.isNull(tablet)) {
-          continue;
-        }
-        resultSets.add(new SubscriptionResultSet(tablet, databaseName));
-      }
+      resultSets.add(new SubscriptionResultSet(tablet));
     }
     this.resultSets = Collections.unmodifiableList(resultSets);
   }
@@ -77,8 +68,6 @@ public static class SubscriptionResultSet implements ISessionDataSet {
 
     private Tablet tablet;
 
-    @Nullable private final String databaseName;
-
     private final List<RowPosition> sortedRowPositions;
 
     private int rowIndex = -1;
@@ -87,20 +76,15 @@ public static class SubscriptionResultSet implements ISessionDataSet {
 
     private List<String> columnTypeList;
 
-    private SubscriptionResultSet(final Tablet tablet, @Nullable final String databaseName) {
+    private SubscriptionResultSet(final Tablet tablet) {
       this.tablet = tablet;
-      this.databaseName = databaseName;
       this.sortedRowPositions = generateSortedRowPositions(tablet);
     }
 
     public Tablet getTablet() {
       return tablet;
     }
 
-    public String getDatabaseName() {
-      return databaseName;
-    }
-
     @Override
     public List<String> getColumnNames() {
       if (Objects.nonNull(columnNameList)) {
@@ -111,12 +95,6 @@ public List<String> getColumnNames() {
       columnNameList.add("Time");
 
       final List<MeasurementSchema> schemas = tablet.getSchemas();
-      if (Objects.nonNull(databaseName)) {
-        columnNameList.addAll(
-            schemas.stream().map(MeasurementSchema::getMeasurementId).collect(Collectors.toList()));
-        return columnNameList;
-      }
-
       final String deviceId = tablet.deviceId;
       columnNameList.addAll(
           schemas.stream()

iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionTsFileHandler.java

@@ -19,26 +19,18 @@
 
 package org.apache.iotdb.session.subscription.payload;
 
-import org.apache.thrift.annotation.Nullable;
 import org.apache.tsfile.read.TsFileReader;
 import org.apache.tsfile.read.TsFileSequenceReader;
 
 import java.io.IOException;
 
 public class SubscriptionTsFileHandler extends SubscriptionFileHandler {
 
-  @Nullable private final String databaseName;
-
-  public SubscriptionTsFileHandler(final String absolutePath, @Nullable final String databaseName) {
+  public SubscriptionTsFileHandler(final String absolutePath) {
     super(absolutePath);
-    this.databaseName = databaseName;
   }
 
   public TsFileReader openReader() throws IOException {
     return new TsFileReader(new TsFileSequenceReader(absolutePath));
   }
-
-  public String getDatabaseName() {
-    return databaseName;
-  }
 }