diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java index 1263603bd491..af00b88ffa6d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java +++ b/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>> 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>> 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>> readLastValues( + final File tsFile) throws Exception { + final Map> deviceToTimeseriesDataTypes = + readTimeseriesDataTypes(tsFile); + final long expectedTimeseriesCount = + deviceToTimeseriesDataTypes.values().stream().mapToLong(Map::size).sum(); + long actualTimeseriesCount = 0; + final Map>> 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>> deviceLastValue = + lastReader.next(); + final Map timeseriesDataTypes = + deviceToTimeseriesDataTypes.get(deviceLastValue.getLeft()); + if (Objects.isNull(timeseriesDataTypes)) { + return null; + } + + final List> typedLastValues = + deviceLastValues.computeIfAbsent(deviceLastValue.getLeft(), key -> new ArrayList<>()); + for (final Pair 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> readTimeseriesDataTypes(final File tsFile) + throws IOException { + try (final TsFileSequenceReader sequenceReader = new TsFileSequenceReader(tsFile.getPath())) { + final Map> deviceToTimeseriesDataTypes = + new LinkedHashMap<>(); + for (final Map.Entry> entry : + sequenceReader.getAllTimeseriesMetadata(false).entrySet()) { + final Map 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 diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java index 8c6c30dfebab..83fc94e2ea2f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java +++ b/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,6 +127,26 @@ public void transfer(final Tablet tablet, final OpcUaSink sink) throws Exception tablet, false, sink, this::transferTabletRowForClientServerModel); } + public void transferLastValues( + final Map>> deviceLastValues, + final boolean isTableModel, + final OpcUaSink sink) + throws Exception { + final List writeRequests = new ArrayList<>(); + for (final Map.Entry>> 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 measurementSchemas, @@ -130,11 +154,22 @@ private void transferTabletRowForClientServerModel( final List values, final OpcUaSink sink) throws Exception { + final List writeRequests = new ArrayList<>(); + collectWriteRequests(segments, measurementSchemas, timestamps, values, sink, writeRequests); + writeValues(writeRequests); + } + + private void collectWriteRequests( + final String[] segments, + final List measurementSchemas, + final List timestamps, + final List values, + final OpcUaSink sink, + final List writeRequests) { StatusCode currentQuality = sink.getDefaultQuality(); Object value = null; long timestamp = 0; NodeId opcDataType = null; - final List 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 writeRequests) throws Exception { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java index 1d6262c2c6a1..421ba27f77f0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java +++ b/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> 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> 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 schemas = new ArrayList<>(lastValues.size()); + final List timestamps = new ArrayList<>(lastValues.size()); + final List values = new ArrayList<>(lastValues.size()); + for (final Pair 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: diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSinkTsFileMetadataTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSinkTsFileMetadataTest.java new file mode 100644 index 000000000000..ad26debe4167 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSinkTsFileMetadataTest.java @@ -0,0 +1,167 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.iotdb.db.pipe.sink.protocol.opcua; + +import org.apache.tsfile.common.conf.TSFileConfig; +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.file.metadata.TableSchema; +import org.apache.tsfile.read.TimeValuePair; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.Pair; +import org.apache.tsfile.write.TsFileWriter; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.Assert; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import java.io.File; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +public class OpcUaSinkTsFileMetadataTest { + + @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + + @Test + public void testReadLastValuesFromTreeTsFile() throws Exception { + final File tsFile = new File(temporaryFolder.getRoot(), "tree.tsfile"); + final String device = "root.sg.d1"; + final List schemas = + Arrays.asList( + new MeasurementSchema("s1", TSDataType.INT64), + new MeasurementSchema("blob", TSDataType.BLOB)); + final Tablet tablet = new Tablet(device, schemas, 3); + for (int i = 0; i < 3; ++i) { + tablet.addTimestamp(i, i + 1L); + } + tablet.addValue("s1", 0, 10L); + tablet.addValue("s1", 1, 20L); + tablet.addValue("s1", 2, null); + tablet.addValue("blob", 0, new Binary("old", TSFileConfig.STRING_CHARSET)); + tablet.addValue("blob", 1, null); + tablet.addValue("blob", 2, new Binary("last", TSFileConfig.STRING_CHARSET)); + + try (final TsFileWriter writer = new TsFileWriter(tsFile)) { + for (final IMeasurementSchema schema : schemas) { + writer.registerTimeseries(device, schema); + } + writer.writeTree(tablet); + } + + final Map>> deviceLastValues = + OpcUaSink.readLastValues(tsFile); + Assert.assertEquals(1, deviceLastValues.size()); + final Map lastValues = + toMeasurementMap(deviceLastValues.values().iterator().next()); + + Assert.assertFalse(lastValues.containsKey(TsFileConstant.TIME_COLUMN_ID)); + assertLongLastValue(lastValues.get("s1"), 2L, 20L); + assertBinaryLastValue(lastValues.get("blob"), 3L, "last"); + } + + @Test + public void testReadLastValuesFromTableTsFile() throws Exception { + final File tsFile = new File(temporaryFolder.getRoot(), "table.tsfile"); + final List columnNames = Arrays.asList("tag", "s1", "blob", "timestamp"); + final List dataTypes = + Arrays.asList(TSDataType.STRING, TSDataType.INT64, TSDataType.BLOB, TSDataType.TIMESTAMP); + final List columnCategories = + Arrays.asList( + ColumnCategory.TAG, ColumnCategory.FIELD, ColumnCategory.FIELD, ColumnCategory.FIELD); + final List schemas = + Arrays.asList( + new MeasurementSchema("tag", TSDataType.STRING), + new MeasurementSchema("s1", TSDataType.INT64), + new MeasurementSchema("blob", TSDataType.BLOB), + new MeasurementSchema("timestamp", TSDataType.TIMESTAMP)); + final Tablet tablet = new Tablet("table", columnNames, dataTypes, columnCategories, 3); + for (int i = 0; i < 3; ++i) { + tablet.addTimestamp(i, i + 1L); + tablet.addValue(i, 0, "tag-value"); + } + tablet.addValue(0, 1, 10L); + tablet.addValue(1, 1, 20L); + tablet.addValue("s1", 2, null); + tablet.addValue("blob", 0, new Binary("old", TSFileConfig.STRING_CHARSET)); + tablet.addValue("blob", 1, null); + tablet.addValue("blob", 2, new Binary("last", TSFileConfig.STRING_CHARSET)); + tablet.addValue(0, 3, 1_700_000_000_000L); + tablet.addValue(1, 3, 1_700_000_001_000L); + tablet.addValue(2, 3, 1_700_000_002_000L); + + try (final TsFileWriter writer = new TsFileWriter(tsFile)) { + writer.registerTableSchema(new TableSchema("table", schemas, columnCategories)); + writer.writeTable(tablet); + } + + final Map>> deviceLastValues = + OpcUaSink.readLastValues(tsFile); + Assert.assertEquals(1, deviceLastValues.size()); + final Map.Entry>> entry = + deviceLastValues.entrySet().iterator().next(); + Assert.assertArrayEquals(new Object[] {"table", "tag-value"}, entry.getKey().getSegments()); + + final Map lastValues = toMeasurementMap(entry.getValue()); + Assert.assertFalse(lastValues.containsKey("tag")); + Assert.assertFalse(lastValues.containsKey(TsFileConstant.TIME_COLUMN_ID)); + assertLongLastValue(lastValues.get("s1"), 2L, 20L); + assertBinaryLastValue(lastValues.get("blob"), 3L, "last"); + assertLongLastValue(lastValues.get("timestamp"), 3L, 1_700_000_002_000L); + Assert.assertEquals(TSDataType.TIMESTAMP, getSchema(entry.getValue(), "timestamp").getType()); + } + + private static Map toMeasurementMap( + final List> lastValues) { + final Map result = new LinkedHashMap<>(); + lastValues.forEach( + lastValue -> result.put(lastValue.getLeft().getMeasurementName(), lastValue.getRight())); + return result; + } + + private static IMeasurementSchema getSchema( + final List> lastValues, final String measurement) { + return lastValues.stream() + .map(Pair::getLeft) + .filter(schema -> measurement.equals(schema.getMeasurementName())) + .findFirst() + .orElseThrow(AssertionError::new); + } + + private static void assertLongLastValue( + final TimeValuePair lastValue, final long timestamp, final long value) { + Assert.assertNotNull(lastValue); + Assert.assertEquals(timestamp, lastValue.getTimestamp()); + Assert.assertEquals(value, lastValue.getValue().getLong()); + } + + private static void assertBinaryLastValue( + final TimeValuePair lastValue, final long timestamp, final String value) { + Assert.assertNotNull(lastValue); + Assert.assertEquals(timestamp, lastValue.getTimestamp()); + Assert.assertEquals(value, lastValue.getValue().getBinary().toString()); + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSinkTsFilePerformanceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSinkTsFilePerformanceTest.java new file mode 100644 index 000000000000..59fdd76f73a6 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSinkTsFilePerformanceTest.java @@ -0,0 +1,415 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.iotdb.db.pipe.sink.protocol.opcua; + +import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.commons.pipe.config.PipeConfig; +import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixTreePattern; +import org.apache.iotdb.db.pipe.event.common.tsfile.parser.scan.TsFileInsertionEventScanParser; +import org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace; +import org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace.TabletRowConsumer; +import org.apache.iotdb.db.utils.ManualPerformanceTestUtils; +import org.apache.iotdb.db.utils.ManualPerformanceTestUtils.Measurement; +import org.apache.iotdb.db.utils.ManualPerformanceTestUtils.Summary; + +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.file.metadata.enums.CompressionType; +import org.apache.tsfile.file.metadata.enums.TSEncoding; +import org.apache.tsfile.read.TimeValuePair; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.Pair; +import org.apache.tsfile.write.TsFileWriter; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.Assert; +import org.junit.Assume; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import java.io.File; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; + +/** + * Manual benchmark for the OPC UA client-server TsFile paths. Enable it with {@code + * -Diotdb.opcua.tsfile.last-value.perf.enabled=true}; the remaining properties below tune the + * generated file and measurement rounds. + */ +public class OpcUaSinkTsFilePerformanceTest { + + private static final String ENABLED_PROPERTY = "iotdb.opcua.tsfile.last-value.perf.enabled"; + private static final String DEVICE_COUNT_PROPERTY = + "iotdb.opcua.tsfile.last-value.perf.device.count"; + private static final String MEASUREMENT_COUNT_PROPERTY = + "iotdb.opcua.tsfile.last-value.perf.measurement.count"; + private static final String BLOB_MEASUREMENT_COUNT_PROPERTY = + "iotdb.opcua.tsfile.last-value.perf.blob.measurement.count"; + private static final String ROW_COUNT_PROPERTY = "iotdb.opcua.tsfile.last-value.perf.row.count"; + private static final String TABLET_ROW_COUNT_PROPERTY = + "iotdb.opcua.tsfile.last-value.perf.tablet.row.count"; + private static final String WARMUP_ITERATIONS_PROPERTY = + "iotdb.opcua.tsfile.last-value.perf.warmup.iterations"; + private static final String ITERATIONS_PROPERTY = "iotdb.opcua.tsfile.last-value.perf.iterations"; + private static final String ROUNDS_PROPERTY = "iotdb.opcua.tsfile.last-value.perf.rounds"; + + private static volatile long benchmarkBlackhole; + + @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + + @Test + public void compareTabletAndMetadataLastValuePaths() throws Exception { + Assume.assumeTrue( + String.format( + Locale.ROOT, + "Manual performance UT. Enable with -D%s=true; use the benchmark-specific properties to tune its workload.", + ENABLED_PROPERTY), + Boolean.getBoolean(ENABLED_PROPERTY)); + Assume.assumeTrue( + "Current-thread CPU time and allocation metrics are required.", + ManualPerformanceTestUtils.enableThreadMetrics()); + + final int deviceCount = Integer.getInteger(DEVICE_COUNT_PROPERTY, 2); + final int measurementCount = Integer.getInteger(MEASUREMENT_COUNT_PROPERTY, 32); + final int blobMeasurementCount = Integer.getInteger(BLOB_MEASUREMENT_COUNT_PROPERTY, 1); + final int rowCount = Integer.getInteger(ROW_COUNT_PROPERTY, 50_000); + final int tabletRowCount = Integer.getInteger(TABLET_ROW_COUNT_PROPERTY, 1024); + final int warmupIterations = Integer.getInteger(WARMUP_ITERATIONS_PROPERTY, 1); + final int iterations = Integer.getInteger(ITERATIONS_PROPERTY, 1); + final int rounds = Integer.getInteger(ROUNDS_PROPERTY, 5); + assertPositive( + deviceCount, + measurementCount, + rowCount, + tabletRowCount, + warmupIterations, + iterations, + rounds); + Assert.assertTrue(blobMeasurementCount >= 0); + Assert.assertTrue(blobMeasurementCount <= measurementCount); + + final File tsFile = new File(temporaryFolder.getRoot(), "opcua-last-value-performance.tsfile"); + generateAlignedTreeTsFile( + tsFile, deviceCount, measurementCount, blobMeasurementCount, rowCount, tabletRowCount); + + final boolean originalPipeMemoryManagementEnabled = + PipeConfig.getInstance().getPipeMemoryManagementEnabled(); + CommonDescriptor.getInstance().getConfig().setPipeMemoryManagementEnabled(false); + try { + final Map tabletLastValues = + captureLastValues(consumer -> transferByTabletPath(tsFile, consumer)); + final Map metadataLastValues = + captureLastValues(consumer -> transferByMetadataPath(tsFile, consumer)); + Assert.assertEquals((long) deviceCount * measurementCount, tabletLastValues.size()); + Assert.assertEquals(tabletLastValues, metadataLastValues); + + compare( + tsFile, + deviceCount, + measurementCount, + blobMeasurementCount, + rowCount, + warmupIterations, + iterations, + rounds); + } finally { + CommonDescriptor.getInstance() + .getConfig() + .setPipeMemoryManagementEnabled(originalPipeMemoryManagementEnabled); + } + } + + private static void compare( + final File tsFile, + final int deviceCount, + final int measurementCount, + final int blobMeasurementCount, + final int rowCount, + final int warmupIterations, + final int iterations, + final int rounds) { + final Runnable tabletPath = () -> benchmark(() -> transferByTabletPath(tsFile, null)); + final Runnable metadataPath = () -> benchmark(() -> transferByMetadataPath(tsFile, null)); + + for (int i = 0; i < warmupIterations; ++i) { + if ((i & 1) == 0) { + tabletPath.run(); + metadataPath.run(); + } else { + metadataPath.run(); + tabletPath.run(); + } + } + + final Measurement[] tabletMeasurements = new Measurement[rounds]; + final Measurement[] metadataMeasurements = new Measurement[rounds]; + for (int round = 0; round < rounds; ++round) { + if ((round & 1) == 0) { + tabletMeasurements[round] = ManualPerformanceTestUtils.measure(iterations, tabletPath); + metadataMeasurements[round] = ManualPerformanceTestUtils.measure(iterations, metadataPath); + } else { + metadataMeasurements[round] = ManualPerformanceTestUtils.measure(iterations, metadataPath); + tabletMeasurements[round] = ManualPerformanceTestUtils.measure(iterations, tabletPath); + } + } + + final Summary tabletSummary = + ManualPerformanceTestUtils.summarize(tabletMeasurements, iterations); + final Summary metadataSummary = + ManualPerformanceTestUtils.summarize(metadataMeasurements, iterations); + final long pointCount = (long) deviceCount * measurementCount * rowCount; + System.out.printf( + Locale.ROOT, + "%nOPC UA TsFile last-value benchmark: file=%.2f MiB, devices=%d, measurements/device=%d, BLOB measurements/device=%d, rows/device=%d, points=%d, warmups=%d, iterations/round=%d, rounds=%d%n", + tsFile.length() / 1024.0 / 1024.0, + deviceCount, + measurementCount, + blobMeasurementCount, + rowCount, + pointCount, + warmupIterations, + iterations, + rounds); + printSummary("tablet path", tabletSummary); + printSummary("metadata path", metadataSummary); + System.out.printf( + Locale.ROOT, + " change: CPU speedup=%.2fx, allocation reduction=%.1f%%, peak-heap reduction=%.1f%%%n", + ratio(tabletSummary.getCpuNanosPerOperation(), metadataSummary.getCpuNanosPerOperation()), + reduction( + tabletSummary.getAllocatedBytesPerOperation(), + metadataSummary.getAllocatedBytesPerOperation()), + reduction(tabletSummary.getPeakHeapDeltaBytes(), metadataSummary.getPeakHeapDeltaBytes())); + } + + private static void transferByTabletPath( + final File tsFile, final TabletRowConsumer suppliedConsumer) throws Exception { + final BenchmarkConsumer benchmarkConsumer = + Objects.isNull(suppliedConsumer) ? new BenchmarkConsumer() : null; + final TabletRowConsumer consumer = + Objects.isNull(suppliedConsumer) ? benchmarkConsumer : suppliedConsumer; + try (final TsFileInsertionEventScanParser parser = + new TsFileInsertionEventScanParser( + tsFile, + new PrefixTreePattern("root"), + Long.MIN_VALUE, + Long.MAX_VALUE, + null, + null, + false)) { + for (final Pair tabletWithIsAligned : parser.toTabletWithIsAligneds()) { + OpcUaNameSpace.transferTabletForClientServerModel( + tabletWithIsAligned.getLeft(), false, null, consumer); + } + } + if (Objects.nonNull(benchmarkConsumer)) { + benchmarkBlackhole = benchmarkConsumer.result(); + } + } + + private static void transferByMetadataPath( + final File tsFile, final TabletRowConsumer suppliedConsumer) throws Exception { + final BenchmarkConsumer benchmarkConsumer = + Objects.isNull(suppliedConsumer) ? new BenchmarkConsumer() : null; + final TabletRowConsumer consumer = + Objects.isNull(suppliedConsumer) ? benchmarkConsumer : suppliedConsumer; + for (final Map.Entry>> entry : + OpcUaSink.readLastValues(tsFile).entrySet()) { + OpcUaNameSpace.transferLastValues(entry.getKey(), entry.getValue(), false, null, consumer); + } + if (Objects.nonNull(benchmarkConsumer)) { + benchmarkBlackhole = benchmarkConsumer.result(); + } + } + + private static Map captureLastValues( + final ThrowingConsumerRunner runner) throws Exception { + final Map lastValues = new LinkedHashMap<>(); + runner.run( + (segments, schemas, timestamps, values, sink) -> { + final String device = String.join(".", segments); + for (int i = 0; i < schemas.size(); ++i) { + lastValues.put( + device + "." + schemas.get(i).getMeasurementName(), + new CapturedLastValue(timestamps.get(i), values.get(i))); + } + }); + return lastValues; + } + + private static void benchmark(final ThrowingRunnable operation) { + try { + operation.run(); + } catch (final Exception e) { + throw new AssertionError(e); + } + } + + private static void generateAlignedTreeTsFile( + final File tsFile, + final int deviceCount, + final int measurementCount, + final int blobMeasurementCount, + final int rowCount, + final int tabletRowCount) + throws Exception { + final List schemas = new ArrayList<>(measurementCount); + for (int measurement = 0; measurement < measurementCount; ++measurement) { + final TSDataType dataType = + measurement < blobMeasurementCount ? TSDataType.BLOB : TSDataType.INT64; + schemas.add( + new MeasurementSchema( + "s" + measurement, dataType, TSEncoding.PLAIN, CompressionType.LZ4)); + } + + try (final TsFileWriter writer = new TsFileWriter(tsFile)) { + for (int device = 0; device < deviceCount; ++device) { + final String deviceId = "root.opcua_perf.d" + device; + writer.registerAlignedTimeseries(new PartialPath(deviceId), schemas); + final Tablet tablet = new Tablet(deviceId, schemas, tabletRowCount); + for (int row = 0; row < rowCount; ++row) { + if (tablet.getRowSize() == tablet.getMaxRowNumber()) { + writer.writeAligned(tablet); + tablet.reset(); + } + + final int rowIndex = tablet.getRowSize(); + tablet.addTimestamp(rowIndex, row); + for (int measurement = 0; measurement < measurementCount; ++measurement) { + if (measurement < blobMeasurementCount) { + tablet.addValue( + schemas.get(measurement).getMeasurementName(), + rowIndex, + new Binary( + "d" + device + "s" + measurement + "r" + row, TSFileConfig.STRING_CHARSET)); + } else { + tablet.addValue( + rowIndex, + measurement, + ((long) device * measurementCount + measurement) * rowCount + row); + } + } + } + if (tablet.getRowSize() > 0) { + writer.writeAligned(tablet); + } + } + } + } + + private static void assertPositive(final int... values) { + for (final int value : values) { + Assert.assertTrue(value > 0); + } + } + + private static void printSummary(final String label, final Summary summary) { + System.out.printf( + Locale.ROOT, + " %-13s CPU=%.3f ms/file, allocated=%.3f MiB/file, peak heap delta=%.3f MiB%n", + label, + summary.getCpuNanosPerOperation() / 1_000_000.0, + summary.getAllocatedBytesPerOperation() / 1024.0 / 1024.0, + summary.getPeakHeapDeltaBytes() / 1024.0 / 1024.0); + } + + private static double ratio(final double baseline, final double optimized) { + return optimized == 0 ? Double.POSITIVE_INFINITY : baseline / optimized; + } + + private static double reduction(final double baseline, final double optimized) { + return baseline == 0 ? 0 : (baseline - optimized) * 100.0 / baseline; + } + + @FunctionalInterface + private interface ThrowingConsumerRunner { + void run(TabletRowConsumer consumer) throws Exception; + } + + @FunctionalInterface + private interface ThrowingRunnable { + void run() throws Exception; + } + + private static final class BenchmarkConsumer implements TabletRowConsumer { + + private long hash = 1; + private long callbackCount; + private long valueCount; + + @Override + public void accept( + final String[] segments, + final List measurementSchemas, + final List timestamps, + final List values, + final OpcUaSink sink) { + ++callbackCount; + for (final String segment : segments) { + hash = 31 * hash + Objects.hashCode(segment); + } + for (int i = 0; i < measurementSchemas.size(); ++i) { + hash = 31 * hash + measurementSchemas.get(i).getMeasurementName().hashCode(); + hash = 31 * hash + Long.hashCode(timestamps.get(i)); + hash = 31 * hash + Objects.hashCode(values.get(i)); + ++valueCount; + } + } + + private long result() { + return hash ^ callbackCount ^ valueCount; + } + } + + private static final class CapturedLastValue { + + private final long timestamp; + private final Object value; + + private CapturedLastValue(final long timestamp, final Object value) { + this.timestamp = timestamp; + this.value = value; + } + + @Override + public boolean equals(final Object obj) { + if (this == obj) { + return true; + } + if (!(obj instanceof CapturedLastValue)) { + return false; + } + final CapturedLastValue that = (CapturedLastValue) obj; + return timestamp == that.timestamp && Objects.equals(value, that.value); + } + + @Override + public int hashCode() { + return Objects.hash(timestamp, value); + } + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClientTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClientTest.java index 5cb881939d40..8f8333143e31 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClientTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClientTest.java @@ -23,7 +23,12 @@ import org.apache.iotdb.pipe.api.exception.PipeException; 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.utils.TsPrimitiveType; 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.OpcUaClient; import org.eclipse.milo.opcua.sdk.client.identity.AnonymousProvider; @@ -42,7 +47,9 @@ import java.util.Arrays; import java.util.Collections; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; import java.util.concurrent.CompletableFuture; public class IoTDBOpcUaClientTest { @@ -63,6 +70,30 @@ public void testTransferWritesAllMeasurementsInOneRequest() throws Exception { Mockito.argThat(listWithSize(2))); } + @Test + public void testTransferLastValuesBatchesDevicesInOneRequest() throws Exception { + final OpcUaClient miloClient = Mockito.mock(OpcUaClient.class); + Mockito.when(miloClient.writeValuesAsync(Mockito.anyList(), Mockito.anyList())) + .thenReturn( + CompletableFuture.completedFuture(Arrays.asList(StatusCode.GOOD, StatusCode.GOOD))); + final IoTDBOpcUaClient client = createClient(miloClient); + final Map>> deviceLastValues = + new LinkedHashMap<>(); + deviceLastValues.put( + IDeviceID.Factory.DEFAULT_FACTORY.create("root.db.d1"), + Collections.singletonList(lastValue("s1", 1L, 11L))); + deviceLastValues.put( + IDeviceID.Factory.DEFAULT_FACTORY.create("root.db.d2"), + Collections.singletonList(lastValue("s1", 2L, 22L))); + + client.transferLastValues(deviceLastValues, false, createSink()); + + Mockito.verify(miloClient) + .writeValuesAsync( + Mockito.argThat(nodeIds("root/db/d1/s1", "root/db/d2/s1")), + Mockito.argThat(listWithSize(2))); + } + @Test public void testTransferCreatesAndRetriesOnlyMissingNodes() throws Exception { final OpcUaClient miloClient = Mockito.mock(OpcUaClient.class); @@ -157,6 +188,13 @@ private static Tablet createTablet() { return tablet; } + private static Pair lastValue( + final String measurement, final long timestamp, final long value) { + return new Pair<>( + new MeasurementSchema(measurement, TSDataType.INT64), + new TimeValuePair(timestamp, TsPrimitiveType.getByType(TSDataType.INT64, value))); + } + private static ArgumentMatcher> nodeIds(final String... identifiers) { return nodeIds -> { if (nodeIds.size() != identifiers.length) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpaceMetadataTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpaceMetadataTest.java new file mode 100644 index 000000000000..a7410aea71ac --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpaceMetadataTest.java @@ -0,0 +1,167 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.iotdb.db.pipe.sink.protocol.opcua.server; + +import org.apache.iotdb.db.pipe.sink.protocol.opcua.OpcUaSink; + +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.StringArrayDeviceID; +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.utils.TsPrimitiveType; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.eclipse.milo.opcua.stack.core.types.builtin.DateTime; +import org.junit.Assert; +import org.junit.Test; +import org.mockito.Mockito; + +import java.time.LocalDate; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Date; +import java.util.List; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; + +public class OpcUaNameSpaceMetadataTest { + + @Test + public void testTransferLastValuesForTreeModel() throws Exception { + final CapturedRow capturedRow = new CapturedRow(); + + OpcUaNameSpace.transferLastValues( + IDeviceID.Factory.DEFAULT_FACTORY.create("root.sg.d1"), + Arrays.asList( + lastValue("s1", TSDataType.INT64, 5L, 50L), + lastValue("s2", TSDataType.TEXT, 6L, "last"), + new Pair<>(new MeasurementSchema("empty", TSDataType.INT64), null), + new Pair<>( + new MeasurementSchema(TsFileConstant.TIME_COLUMN_ID, TSDataType.INT64), + timeValue(TSDataType.INT64, 7L, 8L))), + false, + createSink(), + capturedRow::capture); + + Assert.assertArrayEquals(new String[] {"root", "sg", "d1"}, capturedRow.segments.get()); + Assert.assertEquals(Arrays.asList("s1", "s2"), capturedRow.getMeasurementNames()); + Assert.assertEquals(Arrays.asList(5L, 6L), capturedRow.timestamps.get()); + Assert.assertEquals(Arrays.asList(50L, "last"), capturedRow.values.get()); + } + + @Test + public void testTransferLastValuesForTableModel() throws Exception { + final int lastDate = DateUtils.parseDateExpressionToInt(LocalDate.of(2024, 1, 2)); + final long lastTimestamp = 1_700_000_001_000L; + final CapturedRow capturedRow = new CapturedRow(); + final OpcUaSink sink = createSink(); + Mockito.when(sink.getDatabaseName()).thenReturn("database"); + Mockito.when(sink.getPlaceHolder4NullTag()).thenReturn("null_tag"); + + OpcUaNameSpace.transferLastValues( + new StringArrayDeviceID("table", "tag", null, "tag2"), + Arrays.asList( + lastValue("date", TSDataType.DATE, 2L, lastDate), + lastValue("timestamp", TSDataType.TIMESTAMP, 4L, lastTimestamp)), + true, + sink, + capturedRow::capture); + + Assert.assertArrayEquals( + new String[] {"database", "table", "tag", "null_tag", "tag2"}, capturedRow.segments.get()); + Assert.assertEquals(Arrays.asList("date", "timestamp"), capturedRow.getMeasurementNames()); + Assert.assertEquals(Arrays.asList(2L, 4L), capturedRow.timestamps.get()); + Assert.assertEquals( + new DateTime(new Date(DateUtils.parseIntToDate(lastDate).getTime())).getUtcTime(), + ((DateTime) capturedRow.values.get().get(0)).getUtcTime()); + Assert.assertEquals( + OpcUaNameSpace.timestampToUtc(lastTimestamp), + ((DateTime) capturedRow.values.get().get(1)).getUtcTime()); + } + + @Test + public void testTransferLastValuesSupportsBinaryValues() throws Exception { + final CapturedRow capturedRow = new CapturedRow(); + + OpcUaNameSpace.transferLastValues( + IDeviceID.Factory.DEFAULT_FACTORY.create("root.sg.d1"), + Arrays.asList(lastValue("blob", TSDataType.BLOB, 1L, "payload")), + false, + createSink(), + capturedRow::capture); + + Assert.assertEquals(Arrays.asList("blob"), capturedRow.getMeasurementNames()); + Assert.assertEquals(TSDataType.BLOB, capturedRow.schemas.get().get(0).getType()); + Assert.assertEquals(Arrays.asList("payload"), capturedRow.values.get()); + } + + private static OpcUaSink createSink() { + final OpcUaSink sink = Mockito.mock(OpcUaSink.class); + Mockito.when(sink.getPlaceHolder4NullTag()).thenReturn("null"); + return sink; + } + + private static Pair lastValue( + final String measurement, + final TSDataType dataType, + final long timestamp, + final Object value) { + return new Pair<>( + new MeasurementSchema(measurement, dataType), timeValue(dataType, timestamp, value)); + } + + private static TimeValuePair timeValue( + final TSDataType dataType, final long timestamp, final Object value) { + final Object primitiveValue = + dataType == TSDataType.TEXT || dataType == TSDataType.BLOB || dataType == TSDataType.STRING + ? new Binary( + String.valueOf(value), org.apache.tsfile.common.conf.TSFileConfig.STRING_CHARSET) + : value; + return new TimeValuePair(timestamp, TsPrimitiveType.getByType(dataType, primitiveValue)); + } + + private static class CapturedRow { + private final AtomicReference segments = new AtomicReference<>(); + private final AtomicReference> schemas = new AtomicReference<>(); + private final AtomicReference> timestamps = new AtomicReference<>(); + private final AtomicReference> values = new AtomicReference<>(); + + private void capture( + final String[] segments, + final List schemas, + final List timestamps, + final List values, + final OpcUaSink sink) { + this.segments.set(segments); + this.schemas.set(new ArrayList<>(schemas)); + this.timestamps.set(new ArrayList<>(timestamps)); + this.values.set(new ArrayList<>(values)); + } + + private List getMeasurementNames() { + return schemas.get().stream() + .map(IMeasurementSchema::getMeasurementName) + .collect(Collectors.toList()); + } + } +}