[Pipe] Optimize OPC UA TsFile last-value transfer

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java

@@ -25,6 +25,7 @@
 import org.apache.iotdb.db.i18n.DataNodePipeMessages;
 import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
 import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
+import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
 import org.apache.iotdb.db.pipe.sink.protocol.opcua.client.ClientRunner;
 import org.apache.iotdb.db.pipe.sink.protocol.opcua.client.IoTDBOpcUaClient;
 import org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace;
@@ -39,11 +40,21 @@
 import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
 import org.apache.iotdb.pipe.api.event.Event;
 import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
 import org.apache.iotdb.pipe.api.exception.PipeException;
 
 import org.apache.tsfile.common.conf.TSFileConfig;
+import org.apache.tsfile.common.constant.TsFileConstant;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.file.metadata.TimeseriesMetadata;
+import org.apache.tsfile.read.TimeValuePair;
+import org.apache.tsfile.read.TsFileSequenceReader;
+import org.apache.tsfile.read.reader.TsFileLastReader;
 import org.apache.tsfile.utils.Pair;
 import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
 import org.eclipse.milo.opcua.sdk.client.identity.AnonymousProvider;
 import org.eclipse.milo.opcua.sdk.client.identity.IdentityProvider;
 import org.eclipse.milo.opcua.sdk.client.identity.UsernameProvider;
@@ -56,7 +67,11 @@
 import javax.annotation.Nullable;
 
 import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
@@ -497,6 +512,156 @@ public void heartbeat() throws Exception {
     // Server side, do nothing
   }
 
+  @Override
+  public void transfer(final TsFileInsertionEvent tsFileInsertionEvent) throws Exception {
+    if (!shouldTransferTsFileByMetadata(tsFileInsertionEvent)) {
+      PipeConnector.super.transfer(tsFileInsertionEvent);
+      return;
+    }
+
+    final PipeTsFileInsertionEvent pipeTsFileInsertionEvent =
+        (PipeTsFileInsertionEvent) tsFileInsertionEvent;
+    boolean delegatedToTabletTransfer = false;
+    try {
+      if (transferTsFileByMetadata(pipeTsFileInsertionEvent)
+          == TsFileTransferResult.FALLBACK_TO_TABLETS) {
+        delegatedToTabletTransfer = true;
+        PipeConnector.super.transfer(tsFileInsertionEvent);
+      }
+    } finally {
+      // PipeConnector.transfer(TsFileInsertionEvent) closes the event itself when it is used as a
+      // fallback. Keep the ownership here for the metadata fast path and exceptional exits.
+      if (!delegatedToTabletTransfer) {
+        tsFileInsertionEvent.close();
+      }
+    }
+  }
+
+  private boolean shouldTransferTsFileByMetadata(final TsFileInsertionEvent tsFileInsertionEvent) {
+    if (!isClientServerModel || !(tsFileInsertionEvent instanceof PipeTsFileInsertionEvent)) {
+      return false;
+    }
+
+    final PipeTsFileInsertionEvent pipeTsFileInsertionEvent =
+        (PipeTsFileInsertionEvent) tsFileInsertionEvent;
+    // Metadata contains the unfiltered last value. Deletions, path/time filters, and privilege
+    // filtering must use the normal parser so that the sink observes exactly the event payload.
+    return !pipeTsFileInsertionEvent.isWithMod()
+        && !pipeTsFileInsertionEvent.shouldParseTimeOrPattern()
+        && !pipeTsFileInsertionEvent.shouldParse4Privilege();
+  }
+
+  private TsFileTransferResult transferTsFileByMetadata(
+      final PipeTsFileInsertionEvent pipeTsFileInsertionEvent) throws Exception {
+    if (!pipeTsFileInsertionEvent.increaseReferenceCount(OpcUaSink.class.getName())) {
+      return TsFileTransferResult.SKIPPED;
+    }
+
+    try {
+      if (!pipeTsFileInsertionEvent.waitForTsFileClose()) {
+        return TsFileTransferResult.SKIPPED;
+      }
+
+      final Map<IDeviceID, List<Pair<IMeasurementSchema, TimeValuePair>>> deviceLastValues;
+      try {
+        deviceLastValues = readLastValues(pipeTsFileInsertionEvent.getTsFile());
+      } catch (final Exception e) {
+        // Keep the parser as a compatibility fallback when the TsFile metadata cannot be read.
+        return TsFileTransferResult.FALLBACK_TO_TABLETS;
+      }
+      if (Objects.isNull(deviceLastValues)) {
+        return TsFileTransferResult.FALLBACK_TO_TABLETS;
+      }
+
+      final boolean isTableModel = pipeTsFileInsertionEvent.isTableModelEvent();
+      if (Objects.nonNull(nameSpace)) {
+        for (final Map.Entry<IDeviceID, List<Pair<IMeasurementSchema, TimeValuePair>>> entry :
+            deviceLastValues.entrySet()) {
+          nameSpace.transferLastValues(entry.getKey(), entry.getValue(), isTableModel, this);
+        }
+      } else if (Objects.nonNull(client)) {
+        // Batch all devices into the same OPC UA write so that many-device TsFiles do not incur one
+        // network round trip per device.
+        client.transferLastValues(deviceLastValues, isTableModel, this);
+      } else {
+        throw new PipeException(DataNodePipeMessages.NO_OPC_CLIENT_OR_SERVER_IS_SPECIFIED);
+      }
+      return TsFileTransferResult.TRANSFERRED;
+    } finally {
+      pipeTsFileInsertionEvent.decreaseReferenceCount(OpcUaSink.class.getName(), false);
+    }
+  }
+
+  static @Nullable Map<IDeviceID, List<Pair<IMeasurementSchema, TimeValuePair>>> readLastValues(
+      final File tsFile) throws Exception {
+    final Map<IDeviceID, Map<String, TSDataType>> deviceToTimeseriesDataTypes =
+        readTimeseriesDataTypes(tsFile);
+    final long expectedTimeseriesCount =
+        deviceToTimeseriesDataTypes.values().stream().mapToLong(Map::size).sum();
+    long actualTimeseriesCount = 0;
+    final Map<IDeviceID, List<Pair<IMeasurementSchema, TimeValuePair>>> deviceLastValues =
+        new LinkedHashMap<>();
+    // Disable asynchronous IO here. The sink already runs in a pipe worker and a synchronous
+    // reader avoids leaving a background task behind when the event is cancelled or falls back to
+    // tablet parsing.
+    try (final TsFileLastReader lastReader = new TsFileLastReader(tsFile.getPath(), false, false)) {
+      while (lastReader.hasNext()) {
+        final Pair<IDeviceID, List<Pair<String, TimeValuePair>>> deviceLastValue =
+            lastReader.next();
+        final Map<String, TSDataType> timeseriesDataTypes =
+            deviceToTimeseriesDataTypes.get(deviceLastValue.getLeft());
+        if (Objects.isNull(timeseriesDataTypes)) {
+          return null;
+        }
+
+        final List<Pair<IMeasurementSchema, TimeValuePair>> typedLastValues =
+            deviceLastValues.computeIfAbsent(deviceLastValue.getLeft(), key -> new ArrayList<>());
+        for (final Pair<String, TimeValuePair> lastValue : deviceLastValue.getRight()) {
+          ++actualTimeseriesCount;
+          final TSDataType dataType = timeseriesDataTypes.get(lastValue.getLeft());
+          if (Objects.isNull(dataType)) {
+            return null;
+          }
+          if (!TsFileConstant.TIME_COLUMN_ID.equals(lastValue.getLeft())) {
+            typedLastValues.add(
+                new Pair<>(
+                    new MeasurementSchema(lastValue.getLeft(), dataType), lastValue.getRight()));
+          }
+        }
+      }
+    }
+
+    // TsFileLastReader logs and suppresses IOExceptions from Iterator#hasNext. Comparing against an
+    // independently read metadata count prevents a truncated result from being treated as EOF.
+    if (actualTimeseriesCount != expectedTimeseriesCount) {
+      return null;
+    }
+    return deviceLastValues;
+  }
+
+  private static Map<IDeviceID, Map<String, TSDataType>> readTimeseriesDataTypes(final File tsFile)
+      throws IOException {
+    try (final TsFileSequenceReader sequenceReader = new TsFileSequenceReader(tsFile.getPath())) {
+      final Map<IDeviceID, Map<String, TSDataType>> deviceToTimeseriesDataTypes =
+          new LinkedHashMap<>();
+      for (final Map.Entry<IDeviceID, List<TimeseriesMetadata>> entry :
+          sequenceReader.getAllTimeseriesMetadata(false).entrySet()) {
+        final Map<String, TSDataType> timeseriesDataTypes = new LinkedHashMap<>();
+        for (final TimeseriesMetadata metadata : entry.getValue()) {
+          timeseriesDataTypes.put(metadata.getMeasurementId(), metadata.getTsDataType());
+        }
+        deviceToTimeseriesDataTypes.put(entry.getKey(), timeseriesDataTypes);
+      }
+      return deviceToTimeseriesDataTypes;
+    }
+  }
+
+  private enum TsFileTransferResult {
+    TRANSFERRED,
+    SKIPPED,
+    FALLBACK_TO_TABLETS
+  }
+
   @Override
   public void transfer(final Event event) throws Exception {
     // Do nothing when receive heartbeat or other events

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java

@@ -28,6 +28,9 @@
 
 import org.apache.tsfile.common.constant.TsFileConstant;
 import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.read.TimeValuePair;
+import org.apache.tsfile.utils.Pair;
 import org.apache.tsfile.write.record.Tablet;
 import org.apache.tsfile.write.schema.IMeasurementSchema;
 import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
@@ -63,6 +66,7 @@
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
@@ -123,18 +127,49 @@ public void transfer(final Tablet tablet, final OpcUaSink sink) throws Exception
         tablet, false, sink, this::transferTabletRowForClientServerModel);
   }
 
+  public void transferLastValues(
+      final Map<IDeviceID, List<Pair<IMeasurementSchema, TimeValuePair>>> deviceLastValues,
+      final boolean isTableModel,
+      final OpcUaSink sink)
+      throws Exception {
+    final List<OpcUaWriteRequest> writeRequests = new ArrayList<>();
+    for (final Map.Entry<IDeviceID, List<Pair<IMeasurementSchema, TimeValuePair>>> entry :
+        deviceLastValues.entrySet()) {
+      OpcUaNameSpace.transferLastValues(
+          entry.getKey(),
+          entry.getValue(),
+          isTableModel,
+          sink,
+          (segments, measurementSchemas, timestamps, values, currentSink) ->
+              collectWriteRequests(
+                  segments, measurementSchemas, timestamps, values, currentSink, writeRequests));
+    }
+    writeValues(writeRequests);
+  }
+
   private void transferTabletRowForClientServerModel(
       final String[] segments,
       final List<IMeasurementSchema> measurementSchemas,
       final List<Long> timestamps,
       final List<Object> values,
       final OpcUaSink sink)
       throws Exception {
+    final List<OpcUaWriteRequest> writeRequests = new ArrayList<>();
+    collectWriteRequests(segments, measurementSchemas, timestamps, values, sink, writeRequests);
+    writeValues(writeRequests);
+  }
+
+  private void collectWriteRequests(
+      final String[] segments,
+      final List<IMeasurementSchema> measurementSchemas,
+      final List<Long> timestamps,
+      final List<Object> values,
+      final OpcUaSink sink,
+      final List<OpcUaWriteRequest> writeRequests) {
     StatusCode currentQuality = sink.getDefaultQuality();
     Object value = null;
     long timestamp = 0;
     NodeId opcDataType = null;
-    final List<OpcUaWriteRequest> writeRequests = new ArrayList<>();
 
     for (int i = 0; i < measurementSchemas.size(); ++i) {
       if (Objects.isNull(values.get(i))) {
@@ -177,8 +212,6 @@ private void transferTabletRowForClientServerModel(
       writeRequests.add(
           new OpcUaWriteRequest(value, timestamp, opcDataType, currentQuality, segments, null));
     }
-
-    writeValues(writeRequests);
   }
 
   private void writeValues(final List<OpcUaWriteRequest> writeRequests) throws Exception {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java

@@ -33,7 +33,11 @@
 import org.apache.tsfile.common.constant.TsFileConstant;
 import org.apache.tsfile.enums.ColumnCategory;
 import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.read.TimeValuePair;
 import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.DateUtils;
+import org.apache.tsfile.utils.Pair;
 import org.apache.tsfile.write.UnSupportedDataTypeException;
 import org.apache.tsfile.write.record.Tablet;
 import org.apache.tsfile.write.schema.IMeasurementSchema;
@@ -63,12 +67,12 @@
 import org.slf4j.LoggerFactory;
 
 import java.nio.file.Paths;
-import java.sql.Date;
 import java.time.LocalDate;
 import java.time.ZoneId;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Date;
 import java.util.List;
 import java.util.Objects;
 import java.util.Set;
@@ -130,6 +134,85 @@ public void transfer(final Tablet tablet, final boolean isTableModel, final OpcU
     }
   }
 
+  /**
+   * Transfers the last value of every measurement in a TsFile device without materializing a {@link
+   * Tablet}. The TsFile last-value reader obtains the values from metadata (and reads only the last
+   * chunk when a data type does not keep a value in statistics).
+   */
+  public void transferLastValues(
+      final IDeviceID deviceID,
+      final List<Pair<IMeasurementSchema, TimeValuePair>> lastValues,
+      final boolean isTableModel,
+      final OpcUaSink sink)
+      throws Exception {
+    transferLastValues(
+        deviceID, lastValues, isTableModel, sink, this::transferTabletRowForClientServerModel);
+  }
+
+  public static void transferLastValues(
+      final IDeviceID deviceID,
+      final List<Pair<IMeasurementSchema, TimeValuePair>> lastValues,
+      final boolean isTableModel,
+      final OpcUaSink sink,
+      final TabletRowConsumer consumer)
+      throws Exception {
+    final String[] segments;
+    if (!isTableModel) {
+      // IDeviceID may compact multiple tree nodes into one segment. Keep the same node layout as
+      // the Tablet path, which splits the complete device path.
+      segments = deviceID.toString().split("\\.");
+    } else {
+      final Object[] deviceSegments = deviceID.getSegments();
+      segments = new String[deviceSegments.length + 1];
+      segments[0] = sink.getDatabaseName();
+      for (int i = 0; i < deviceSegments.length; ++i) {
+        segments[i + 1] =
+            Objects.isNull(deviceSegments[i])
+                ? sink.getPlaceHolder4NullTag()
+                : String.valueOf(deviceSegments[i]);
+      }
+    }
+
+    final List<IMeasurementSchema> schemas = new ArrayList<>(lastValues.size());
+    final List<Long> timestamps = new ArrayList<>(lastValues.size());
+    final List<Object> values = new ArrayList<>(lastValues.size());
+    for (final Pair<IMeasurementSchema, TimeValuePair> lastValue : lastValues) {
+      if (Objects.isNull(lastValue)
+          || Objects.isNull(lastValue.getLeft())
+          || Objects.isNull(lastValue.getLeft().getMeasurementName())
+          || TsFileConstant.TIME_COLUMN_ID.equals(lastValue.getLeft().getMeasurementName())
+          || Objects.isNull(lastValue.getRight())
+          || Objects.isNull(lastValue.getRight().getValue())) {
+        continue;
+      }
+
+      final TimeValuePair timeValuePair = lastValue.getRight();
+      final TSDataType dataType = lastValue.getLeft().getType();
+      schemas.add(lastValue.getLeft());
+      timestamps.add(timeValuePair.getTimestamp());
+      values.add(getObjectValue4Opc(timeValuePair, dataType));
+    }
+
+    if (!schemas.isEmpty()) {
+      consumer.accept(segments, schemas, timestamps, values, sink);
+    }
+  }
+
+  private static Object getObjectValue4Opc(
+      final TimeValuePair timeValuePair, final TSDataType dataType) {
+    final Object value = timeValuePair.getValue().getValue();
+    return switch (dataType) {
+      case DATE ->
+          new DateTime(new Date(DateUtils.parseIntToDate(((Number) value).intValue()).getTime()));
+      case TIMESTAMP -> new DateTime(timestampToUtc(((Number) value).longValue()));
+      case TEXT, BLOB, STRING -> value instanceof Binary ? value.toString() : String.valueOf(value);
+      case BOOLEAN, INT32, INT64, FLOAT, DOUBLE -> value;
+      case VECTOR, OBJECT, UNKNOWN ->
+          throw new UnSupportedDataTypeException(
+              DataNodePipeMessages.UNSUPPORTED_DATATYPE + dataType);
+    };
+  }
+
   public static void transferTabletForClientServerModel(
       final Tablet tablet,
       final boolean isTableModel,
@@ -393,7 +476,9 @@ private static Object getTabletObjectValue4Opc(
       case INT32:
         return ((int[]) column)[rowIndex];
       case DATE:
-        return new DateTime(Date.valueOf(((LocalDate[]) column)[rowIndex]));
+        return new DateTime(
+            Date.from(
+                ((LocalDate[]) column)[rowIndex].atStartOfDay(ZoneId.systemDefault()).toInstant()));
       case INT64:
         return ((long[]) column)[rowIndex];
       case TIMESTAMP: