diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/payload/PipeTransferConfigSnapshotSealReq.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/payload/PipeTransferConfigSnapshotSealReq.java index 90a13f32e49f8..87efd140a016c 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/payload/PipeTransferConfigSnapshotSealReq.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/payload/PipeTransferConfigSnapshotSealReq.java @@ -26,9 +26,9 @@ import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq; import java.io.IOException; -import java.util.Arrays; -import java.util.Collections; +import java.util.ArrayList; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.Objects; @@ -78,12 +78,8 @@ public static PipeTransferConfigSnapshotSealReq toTPipeTransferReq( return (PipeTransferConfigSnapshotSealReq) new PipeTransferConfigSnapshotSealReq() .convertToTPipeTransferReq( - Objects.nonNull(templateFileName) - ? Arrays.asList(snapshotName, templateFileName) - : Collections.singletonList(snapshotName), - Objects.nonNull(templateFileName) - ? Arrays.asList(snapshotLength, templateFileLength) - : Collections.singletonList(snapshotLength), + buildFileNameList(snapshotName, templateFileName), + buildFileLengthList(snapshotLength, templateFileName, templateFileLength), parameters); } @@ -123,15 +119,31 @@ public static byte[] toTPipeTransferBytes( parameters.put("authUserName", authUserName); return new PipeTransferConfigSnapshotSealReq() .convertToTPipeTransferSnapshotSealBytes( - Objects.nonNull(templateFileName) - ? Arrays.asList(snapshotName, templateFileName) - : Collections.singletonList(snapshotName), - Objects.nonNull(templateFileName) - ? Arrays.asList(snapshotLength, templateFileLength) - : Collections.singletonList(snapshotLength), + buildFileNameList(snapshotName, templateFileName), + buildFileLengthList(snapshotLength, templateFileName, templateFileLength), parameters); } + private static List buildFileNameList( + final String snapshotName, final String templateFileName) { + final List fileNameList = new ArrayList<>(); + fileNameList.add(snapshotName); + if (Objects.nonNull(templateFileName)) { + fileNameList.add(templateFileName); + } + return fileNameList; + } + + private static List buildFileLengthList( + final long snapshotLength, final String templateFileName, final long templateFileLength) { + final List fileLengthList = new ArrayList<>(); + fileLengthList.add(snapshotLength); + if (Objects.nonNull(templateFileName)) { + fileLengthList.add(templateFileLength); + } + return fileLengthList; + } + /////////////////////////////// Object /////////////////////////////// @Override diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java index ba4ad87b1fdf9..d86031c9919e6 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java @@ -1373,6 +1373,11 @@ public void cleanupRemoteSnapshot(Peer targetPeer) throws ConsensusGroupModifyPe } public void cleanupSnapshot(String snapshotId) throws ConsensusGroupModifyPeerException { + cleanupSnapshotWithoutLock(snapshotId); + } + + private void cleanupSnapshotWithoutLock(String snapshotId) + throws ConsensusGroupModifyPeerException { snapshotReceiveFolderMap.remove(snapshotId); List allDirs = new ArrayList<>(Collections.singletonList(storageDir)); allDirs.addAll(recvFolderManager.getFolders()); diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/client/DispatchLogHandler.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/client/DispatchLogHandler.java index 615ecd6b07c46..8190b6b0ece89 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/client/DispatchLogHandler.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/client/DispatchLogHandler.java @@ -68,11 +68,10 @@ public DispatchLogHandler( @Override public void onComplete(TSyncLogEntriesRes response) { recordTransferAttempt(response); - if (response.getStatuses().stream() - .anyMatch(status -> RetryUtils.needRetryForWrite(status.getCode()))) { + if (response.getStatuses().stream().anyMatch(status -> needRetryForSyncLog(status.getCode()))) { List retryStatusMessages = response.getStatuses().stream() - .filter(status -> RetryUtils.needRetryForWrite(status.getCode())) + .filter(status -> needRetryForSyncLog(status.getCode())) .map(TSStatus::getMessage) .collect(Collectors.toList()); @@ -116,6 +115,10 @@ public void onComplete(TSyncLogEntriesRes response) { logDispatcherThreadMetrics.recordSyncLogTimePerRequest(System.nanoTime() - createTime); } + private boolean needRetryForSyncLog(int statusCode) { + return RetryUtils.needRetryForWrite(statusCode); + } + @Override public void onError(Exception exception) { recordTransferAttempt(false, null, exception); diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java index 82394b6a244a1..6c9924677eaad 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java @@ -90,9 +90,7 @@ public TSyncLogEntriesRes syncLogEntries(TSyncLogEntriesReq req) { ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_SYNC_LOG, groupId, req.getLogEntries().size()); - LOGGER.error(message); - TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()); - status.setMessage(message); + TSStatus status = getConsensusGroupNotExistStatus(message); return new TSyncLogEntriesRes(Collections.singletonList(status)); } if (impl.isReadOnly()) { @@ -154,9 +152,7 @@ public TSyncWriterSafeTimeBarrierRes syncWriterSafeTimeBarrier( ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST, groupId, "TSyncWriterSafeTimeBarrierReq"); - LOGGER.error(message); - final TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()); - status.setMessage(message); + final TSStatus status = getConsensusGroupNotExistStatus(message); return new TSyncWriterSafeTimeBarrierRes().setStatus(status); } impl.observeRemoteWriterSafeTimeBarrier( @@ -180,9 +176,7 @@ public TInactivatePeerRes inactivatePeer(TInactivatePeerReq req) throws TExcepti ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST, groupId, "inactivatePeer"); - LOGGER.error(message); - TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()); - status.setMessage(message); + TSStatus status = getConsensusGroupNotExistStatus(message); return new TInactivatePeerRes(status); } impl.setActive(false); @@ -203,9 +197,7 @@ public TActivatePeerRes activatePeer(TActivatePeerReq req) throws TException { ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST, groupId, "inactivatePeer"); - LOGGER.error(message); - TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()); - status.setMessage(message); + TSStatus status = getConsensusGroupNotExistStatus(message); return new TActivatePeerRes(status); } KillPoint.setKillPoint(DataNodeKillPoints.DESTINATION_ADD_PEER_DONE); @@ -225,9 +217,7 @@ public TBuildSyncLogChannelRes buildSyncLogChannel(TBuildSyncLogChannelReq req) ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST, groupId, "buildSyncLogChannel"); - LOGGER.error(message); - TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()); - status.setMessage(message); + TSStatus status = getConsensusGroupNotExistStatus(message); return new TBuildSyncLogChannelRes(status); } TSStatus responseStatus; @@ -248,9 +238,7 @@ public TRemoveSyncLogChannelRes removeSyncLogChannel(TRemoveSyncLogChannelReq re ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST, groupId, "buildSyncLogChannel"); - LOGGER.error(message); - TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()); - status.setMessage(message); + TSStatus status = getConsensusGroupNotExistStatus(message); return new TRemoveSyncLogChannelRes(status); } TSStatus responseStatus; @@ -310,9 +298,7 @@ public TSendSnapshotFragmentRes sendSnapshotFragment(TSendSnapshotFragmentReq re ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST, groupId, "buildSyncLogChannel"); - LOGGER.error(message); - TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()); - status.setMessage(message); + TSStatus status = getConsensusGroupNotExistStatus(message); return new TSendSnapshotFragmentRes(status); } TSStatus responseStatus; @@ -338,9 +324,7 @@ public TTriggerSnapshotLoadRes triggerSnapshotLoad(TTriggerSnapshotLoadReq req) ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST, groupId, "buildSyncLogChannel"); - LOGGER.error(message); - TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()); - status.setMessage(message); + TSStatus status = getConsensusGroupNotExistStatus(message); return new TTriggerSnapshotLoadRes(status); } if (!impl.loadSnapshot(req.snapshotId)) { @@ -370,9 +354,7 @@ public TCleanupTransferredSnapshotRes cleanupTransferredSnapshot( ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST, groupId, "buildSyncLogChannel"); - LOGGER.error(message); - TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()); - status.setMessage(message); + TSStatus status = getConsensusGroupNotExistStatus(message); return new TCleanupTransferredSnapshotRes(status); } TSStatus responseStatus; @@ -387,5 +369,10 @@ public TCleanupTransferredSnapshotRes cleanupTransferredSnapshot( return new TCleanupTransferredSnapshotRes(responseStatus); } + private static TSStatus getConsensusGroupNotExistStatus(String message) { + LOGGER.error(message); + return new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()).setMessage(message); + } + public void handleClientExit() {} } diff --git a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessorTest.java b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessorTest.java new file mode 100644 index 0000000000000..5eeb1bd4157ea --- /dev/null +++ b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessorTest.java @@ -0,0 +1,185 @@ +/* + * 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.consensus.iot.service; + +import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId; +import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.consensus.EmptyStateMachine; +import org.apache.iotdb.consensus.config.ConsensusConfig; +import org.apache.iotdb.consensus.i18n.ConsensusMessages; +import org.apache.iotdb.consensus.iot.IoTConsensus; +import org.apache.iotdb.consensus.iot.thrift.TActivatePeerReq; +import org.apache.iotdb.consensus.iot.thrift.TBuildSyncLogChannelReq; +import org.apache.iotdb.consensus.iot.thrift.TCleanupTransferredSnapshotReq; +import org.apache.iotdb.consensus.iot.thrift.TInactivatePeerReq; +import org.apache.iotdb.consensus.iot.thrift.TRemoveSyncLogChannelReq; +import org.apache.iotdb.consensus.iot.thrift.TSendSnapshotFragmentReq; +import org.apache.iotdb.consensus.iot.thrift.TSyncLogEntriesReq; +import org.apache.iotdb.consensus.iot.thrift.TSyncWriterSafeTimeBarrierReq; +import org.apache.iotdb.consensus.iot.thrift.TTriggerSnapshotLoadReq; +import org.apache.iotdb.consensus.iot.thrift.TWaitReleaseAllRegionRelatedResourceReq; +import org.apache.iotdb.consensus.iot.thrift.TWaitSyncLogCompleteReq; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import java.io.IOException; +import java.util.Collections; + +public class IoTConsensusRPCServiceProcessorTest { + + private static final TConsensusGroupId MISSING_GROUP_ID = + new TConsensusGroupId(TConsensusGroupType.DataRegion, 1); + + @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + + private IoTConsensus consensus; + private IoTConsensusRPCServiceProcessor processor; + + @Before + public void setUp() throws IOException { + consensus = + new IoTConsensus( + ConsensusConfig.newBuilder() + .setThisNodeId(1) + .setThisNode(new TEndPoint("127.0.0.1", 0)) + .setStorageDir(temporaryFolder.newFolder("storage").getAbsolutePath()) + .setRecvSnapshotDirs( + Collections.singletonList( + temporaryFolder.newFolder("snapshot").getAbsolutePath())) + .setConsensusGroupType(TConsensusGroupType.DataRegion) + .build(), + groupId -> new EmptyStateMachine()); + processor = new IoTConsensusRPCServiceProcessor(consensus); + } + + @After + public void tearDown() { + consensus.stop(); + } + + @Test + public void testMissingConsensusGroupPreservesStatusAndMessage() throws Exception { + assertMissingGroupStatus( + processor + .syncLogEntries( + new TSyncLogEntriesReq() + .setConsensusGroupId(MISSING_GROUP_ID) + .setLogEntries(Collections.emptyList())) + .getStatuses() + .get(0), + String.format( + ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_SYNC_LOG, "DataRegion[1]", 0)); + assertMissingGroupStatus( + processor + .syncWriterSafeTimeBarrier( + new TSyncWriterSafeTimeBarrierReq().setConsensusGroupId(MISSING_GROUP_ID)) + .getStatus(), + String.format( + ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST, + "DataRegion[1]", + "TSyncWriterSafeTimeBarrierReq")); + assertMissingGroupStatus( + processor + .inactivatePeer(new TInactivatePeerReq().setConsensusGroupId(MISSING_GROUP_ID)) + .getStatus(), + String.format( + ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST, + "DataRegion[1]", + "inactivatePeer")); + assertMissingGroupStatus( + processor + .activatePeer(new TActivatePeerReq().setConsensusGroupId(MISSING_GROUP_ID)) + .getStatus(), + String.format( + ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST, + "DataRegion[1]", + "inactivatePeer")); + assertMissingGroupStatus( + processor + .buildSyncLogChannel( + new TBuildSyncLogChannelReq().setConsensusGroupId(MISSING_GROUP_ID)) + .getStatus(), + String.format( + ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST, + "DataRegion[1]", + "buildSyncLogChannel")); + assertMissingGroupStatus( + processor + .removeSyncLogChannel( + new TRemoveSyncLogChannelReq().setConsensusGroupId(MISSING_GROUP_ID)) + .getStatus(), + String.format( + ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST, + "DataRegion[1]", + "buildSyncLogChannel")); + assertMissingGroupStatus( + processor + .sendSnapshotFragment( + new TSendSnapshotFragmentReq().setConsensusGroupId(MISSING_GROUP_ID)) + .getStatus(), + String.format( + ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST, + "DataRegion[1]", + "buildSyncLogChannel")); + assertMissingGroupStatus( + processor + .triggerSnapshotLoad( + new TTriggerSnapshotLoadReq().setConsensusGroupId(MISSING_GROUP_ID)) + .getStatus(), + String.format( + ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST, + "DataRegion[1]", + "buildSyncLogChannel")); + assertMissingGroupStatus( + processor + .cleanupTransferredSnapshot( + new TCleanupTransferredSnapshotReq().setConsensusGroupId(MISSING_GROUP_ID)) + .getStatus(), + String.format( + ConsensusMessages.UNEXPECTED_CONSENSUS_GROUP_ID_FOR_REQUEST, + "DataRegion[1]", + "buildSyncLogChannel")); + } + + @Test + public void testMissingConsensusGroupCompletesIdempotentWaits() throws Exception { + Assert.assertTrue( + processor.waitSyncLogComplete( + new TWaitSyncLogCompleteReq().setConsensusGroupId(MISSING_GROUP_ID)) + .complete); + Assert.assertTrue( + processor.waitReleaseAllRegionRelatedResource( + new TWaitReleaseAllRegionRelatedResourceReq().setConsensusGroupId(MISSING_GROUP_ID)) + .releaseAllResource); + } + + private static void assertMissingGroupStatus(TSStatus status, String expectedMessage) { + Assert.assertEquals(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode(), status.getCode()); + Assert.assertEquals(expectedMessage, status.getMessage()); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java index 9cd0eb59f384e..891340bd91a68 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java @@ -29,6 +29,7 @@ import org.apache.iotdb.commons.pipe.event.EnrichedEvent; import org.apache.iotdb.commons.pipe.resource.PipeResourceFailureType; import org.apache.iotdb.commons.pipe.sink.protocol.IoTDBSink; +import org.apache.iotdb.commons.pipe.sink.protocol.PipeBatchMetricsSettable; import org.apache.iotdb.commons.pipe.sink.protocol.PipeConnectorWithEventDiscard; import org.apache.iotdb.commons.pipe.sink.protocol.PipeSinkWithSchedulingDelay; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType; @@ -648,8 +649,9 @@ public double getTotalCompressedSize() { } public void setTabletBatchSizeHistogram(Histogram tabletBatchSizeHistogram) { - if (outputPipeSink instanceof IoTDBSink) { - ((IoTDBSink) outputPipeSink).setTabletBatchSizeHistogram(tabletBatchSizeHistogram); + if (outputPipeSink instanceof PipeBatchMetricsSettable) { + ((PipeBatchMetricsSettable) outputPipeSink) + .setTabletBatchSizeHistogram(tabletBatchSizeHistogram); } } @@ -660,14 +662,15 @@ public void setSchemaBatchSizeHistogram(Histogram schemaBatchSizeHistogram) { } public void setTsFileBatchSizeHistogram(Histogram tsFileBatchSizeHistogram) { - if (outputPipeSink instanceof IoTDBSink) { - ((IoTDBSink) outputPipeSink).setTsFileBatchSizeHistogram(tsFileBatchSizeHistogram); + if (outputPipeSink instanceof PipeBatchMetricsSettable) { + ((PipeBatchMetricsSettable) outputPipeSink) + .setTsFileBatchSizeHistogram(tsFileBatchSizeHistogram); } } public void setTabletBatchTimeIntervalHistogram(Histogram tabletBatchTimeIntervalHistogram) { - if (outputPipeSink instanceof IoTDBSink) { - ((IoTDBSink) outputPipeSink) + if (outputPipeSink instanceof PipeBatchMetricsSettable) { + ((PipeBatchMetricsSettable) outputPipeSink) .setTabletBatchTimeIntervalHistogram(tabletBatchTimeIntervalHistogram); } } @@ -680,15 +683,15 @@ public void setSchemaBatchTimeIntervalHistogram(Histogram schemaBatchTimeInterva } public void setTsFileBatchTimeIntervalHistogram(Histogram tsFileBatchTimeIntervalHistogram) { - if (outputPipeSink instanceof IoTDBSink) { - ((IoTDBSink) outputPipeSink) + if (outputPipeSink instanceof PipeBatchMetricsSettable) { + ((PipeBatchMetricsSettable) outputPipeSink) .setTsFileBatchTimeIntervalHistogram(tsFileBatchTimeIntervalHistogram); } } public void setEventSizeHistogram(Histogram eventSizeHistogram) { - if (outputPipeSink instanceof IoTDBSink) { - ((IoTDBSink) outputPipeSink).setBatchEventSizeHistogram(eventSizeHistogram); + if (outputPipeSink instanceof PipeBatchMetricsSettable) { + ((PipeBatchMetricsSettable) outputPipeSink).setBatchEventSizeHistogram(eventSizeHistogram); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/util/ModsOperationUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/util/ModsOperationUtil.java index 38487b693e928..e8c173292e4eb 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/util/ModsOperationUtil.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/util/ModsOperationUtil.java @@ -56,18 +56,21 @@ private ModsOperationUtil() { */ public static PatternTreeMap loadModificationsFromTsFile(File tsFile) { - PatternTreeMap modifications = - PatternTreeMapFactory.getModsPatternTreeMap(); - try { - ModificationFile.readAllModifications(tsFile, true) - .forEach( - modification -> modifications.append(modification.keyOfPatternTree(), modification)); + return buildModificationsPatternTreeMap(ModificationFile.readAllModifications(tsFile, true)); } catch (Exception e) { throw new PipeException( DataNodePipeMessages.FAILED_TO_LOAD_MODIFICATIONS_FROM_TSFILE + tsFile.getPath(), e); } + } + public static PatternTreeMap + buildModificationsPatternTreeMap(final List modEntries) { + final PatternTreeMap modifications = + PatternTreeMapFactory.getModsPatternTreeMap(); + for (final ModEntry modEntry : modEntries) { + modifications.append(modEntry.keyOfPatternTree(), modEntry); + } return modifications; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java index 8f2e81098238b..44f126e90874d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java @@ -968,10 +968,14 @@ private TSStatus loadSchemaSnapShot( statement, databasePath.getNodes()[1]))); } } + flushSchemaSnapshotStatementBatches(results); + return PipeReceiverStatusHandler.getPriorStatus(results); + } + + private void flushSchemaSnapshotStatementBatches(final List results) { batchVisitor.getRemainBatches().stream() .filter(Optional::isPresent) .forEach(statement -> results.add(executeStatementAndClassifyExceptions(statement.get()))); - return PipeReceiverStatusHandler.getPriorStatus(results); } static boolean shouldLoadTreeSchemaSnapshotDatabase( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilder.java index 45ec6505390fa..ff290cea46d42 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilder.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilder.java @@ -22,6 +22,7 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey; import org.apache.iotdb.commons.pipe.event.EnrichedEvent; +import org.apache.iotdb.commons.pipe.sink.protocol.PipeBatchMetricsSettable; 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; @@ -63,7 +64,7 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_BATCH_SIZE_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_LEADER_CACHE_ENABLE_KEY; -public class PipeTransferBatchReqBuilder implements AutoCloseable { +public class PipeTransferBatchReqBuilder implements AutoCloseable, PipeBatchMetricsSettable { private static final Logger LOGGER = LoggerFactory.getLogger(PipeTransferBatchReqBuilder.class); @@ -212,19 +213,11 @@ public synchronized void onEvent(final TabletInsertionEvent event) */ public synchronized List> getAllNonEmptyAndShouldEmitBatchesAndDetach() { - final List> batches = - new ArrayList<>(endPointToBatch.size() + 1); - if (!defaultBatch.isEmpty() && defaultBatch.shouldEmit()) { - batches.add(new Pair<>(null, defaultBatch)); - } - - for (final Map.Entry entry : endPointToBatch.entrySet()) { - final PipeTabletEventPlainBatch batch = entry.getValue(); - if (!batch.isEmpty() && batch.shouldEmit()) { - batches.add(new Pair<>(entry.getKey(), batch)); - } - } + return detachBatches(getAllNonEmptyAndShouldEmitBatches()); + } + private List> detachBatches( + final List> batches) { // Construct all replacement batches before changing the builder mappings. If construction of // one replacement fails (for example while creating a TsFile batch directory), the old // batches remain owned by this builder and can still be retried or closed by the caller. @@ -329,24 +322,28 @@ public void recordTsFileMetric(long timeInterval, long bufferSize, long eventSiz this.eventSizeHistogram.update(eventSize); } + @Override public void setTabletBatchSizeHistogram(Histogram tabletBatchSizeHistogram) { if (tabletBatchSizeHistogram != null) { this.tabletBatchSizeHistogram = tabletBatchSizeHistogram; } } + @Override public void setTsFileBatchSizeHistogram(Histogram tsFileBatchSizeHistogram) { if (tsFileBatchSizeHistogram != null) { this.tsFileBatchSizeHistogram = tsFileBatchSizeHistogram; } } + @Override public void setTabletBatchTimeIntervalHistogram(Histogram tabletBatchTimeIntervalHistogram) { if (tabletBatchTimeIntervalHistogram != null) { this.tabletBatchTimeIntervalHistogram = tabletBatchTimeIntervalHistogram; } } + @Override public void setTsFileBatchTimeIntervalHistogram(Histogram tsFileBatchTimeIntervalHistogram) { if (tsFileBatchTimeIntervalHistogram != null) { this.tsFileBatchTimeIntervalHistogram = tsFileBatchTimeIntervalHistogram; @@ -354,6 +351,11 @@ public void setTsFileBatchTimeIntervalHistogram(Histogram tsFileBatchTimeInterva } public void setEventSizeHistogram(Histogram eventSizeHistogram) { + setBatchEventSizeHistogram(eventSizeHistogram); + } + + @Override + public void setBatchEventSizeHistogram(Histogram eventSizeHistogram) { if (eventSizeHistogram != null) { this.eventSizeHistogram = eventSizeHistogram; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java index 2a0000f572c56..c8cb2b2621c70 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java @@ -189,7 +189,7 @@ private synchronized boolean handleSinkClosed(final AsyncPipeDataTransferService return true; } - private void returnClientToPool(final AsyncPipeDataTransferServiceClient client) { + protected final void returnClientToPool(final AsyncPipeDataTransferServiceClient client) { client.setShouldReturnSelf(true); client.returnSelf( e -> { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java index e542fc5171915..0a4945cf83a63 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java @@ -146,7 +146,7 @@ public class PipeHistoricalDataRegionTsFileAndDeletionSource private int dataRegionId; private TreePattern treePattern; - private TablePattern tablePattern; + private TablePattern dataTablePattern; private boolean isModelDetected = false; private boolean isTableModel; @@ -368,7 +368,7 @@ public void customize( + Integer.toHexString(System.identityHashCode(environment)); treePattern = TreePattern.parsePipePatternFromSourceParameters(parameters); - tablePattern = TablePattern.parsePipePatternFromSourceParameters(parameters); + dataTablePattern = TablePattern.parsePipePatternFromSourceParameters(parameters); tsFileParser = parameters.getStringByKeys(EXTRACTOR_TSFILE_PARSER_KEY, SOURCE_TSFILE_PARSER_KEY); @@ -380,7 +380,7 @@ public void customize( isTableModel = PathUtils.isTableModelDatabase(databaseName); isModelDetected = true; if (isTableModel) { - isDbNameCoveredByPattern = tablePattern.coversDb(databaseName); + isDbNameCoveredByPattern = dataTablePattern.coversDb(databaseName); } else { isDbNameCoveredByPattern = treePattern.coversDb(databaseName); } @@ -1031,9 +1031,9 @@ private boolean mayTsFileResourceOverlappedWithPattern(final TsFileResource reso } return isTableModel - ? (tablePattern.isTableModelDataAllowedToBeCaptured() - && tablePattern.matchesDatabase(resource.getDatabaseName()) - && tablePattern.matchesTable(deviceID.getTableName())) + ? (dataTablePattern.isTableModelDataAllowedToBeCaptured() + && dataTablePattern.matchesDatabaseAndTable( + resource.getDatabaseName(), deviceID.getTableName())) : (treePattern.isTreeModelDataAllowedToBeCaptured() && treePattern.mayOverlapWithDevice(deviceID)); }); @@ -1048,8 +1048,8 @@ private void detectModel(final TsFileResource resource, final IDeviceID deviceID final String databaseName = resource.getDatabaseName(); isDbNameCoveredByPattern = isTableModel - ? tablePattern.isTableModelDataAllowedToBeCaptured() - && tablePattern.coversDb(databaseName) + ? dataTablePattern.isTableModelDataAllowedToBeCaptured() + && dataTablePattern.coversDb(databaseName) : treePattern.isTreeModelDataAllowedToBeCaptured() && treePattern.coversDb(databaseName); } @@ -1283,7 +1283,7 @@ protected Event supplyTsFileEvent(final TsFileResource resource) { creationTime, pipeTaskMeta, treePattern, - tablePattern, + dataTablePattern, userId, userName, cliHostname, @@ -1360,15 +1360,15 @@ private boolean isTsFileResourceCoveredByTablePattern( final TsFileResource resource, final Set tableNames) { return isModelDetected && isTableModel - && tablePattern.isTableModelDataAllowedToBeCaptured() + && dataTablePattern.isTableModelDataAllowedToBeCaptured() && Objects.nonNull(resource) && Objects.nonNull(tableNames) && !tableNames.isEmpty() && tableNames.stream() .allMatch( tableName -> - tablePattern.matchesDatabase(resource.getDatabaseName()) - && tablePattern.matchesTable(tableName)); + dataTablePattern.matchesDatabaseAndTable( + resource.getDatabaseName(), tableName)); } private Event supplyDeletionEvent(final DeletionResource deletionResource) { @@ -1379,7 +1379,7 @@ private Event supplyDeletionEvent(final DeletionResource deletionResource) { creationTime, pipeTaskMeta, treePattern, - tablePattern, + dataTablePattern, userId, userName, cliHostname, @@ -1452,6 +1452,10 @@ public int getPendingQueueSize() { @Override public synchronized void close() { + doClose(); + } + + private void doClose() { if (!isTerminateSignalSent) { PipeTerminateEvent.clearHistoricalTransferSummary(pipeName, creationTime, dataRegionId); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeTimePartitionListener.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeTimePartitionListener.java index 7c674a3bd1d7c..d8fd295c9cafc 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeTimePartitionListener.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeTimePartitionListener.java @@ -40,25 +40,29 @@ public class PipeTimePartitionListener { //////////////////////////// start & stop //////////////////////////// - public synchronized void startListen(int dataRegionId, PipeRealtimeDataRegionSource extractor) { + public synchronized void startListen( + final int dataRegionId, final PipeRealtimeDataRegionSource source) { dataRegionId2Sources .computeIfAbsent(dataRegionId, o -> new HashMap<>()) - .put(extractor.getTaskID(), extractor); - // Assign the previously recorded upper and lower bounds of time partition to the extractor that + .put(source.getTaskID(), source); + // Assign the previously recorded upper and lower bounds of time partition to the source that // has just started listening to the growth of time partition. - Pair timePartitionIdBound = dataRegionId2TimePartitionIdBound.get(dataRegionId); + final Pair timePartitionIdBound = + dataRegionId2TimePartitionIdBound.get(dataRegionId); if (Objects.nonNull(timePartitionIdBound)) { - extractor.setDataRegionTimePartitionIdBound(timePartitionIdBound); + source.setDataRegionTimePartitionIdBound(timePartitionIdBound); } } - public synchronized void stopListen(int dataRegionId, PipeRealtimeDataRegionSource extractor) { - Map extractors = dataRegionId2Sources.get(dataRegionId); - if (Objects.isNull(extractors)) { + public synchronized void stopListen( + final int dataRegionId, final PipeRealtimeDataRegionSource source) { + final Map sources = + dataRegionId2Sources.get(dataRegionId); + if (Objects.isNull(sources)) { return; } - extractors.remove(extractor.getTaskID()); - if (extractors.isEmpty()) { + sources.remove(source.getTaskID()); + if (sources.isEmpty()) { dataRegionId2Sources.remove(dataRegionId); } } @@ -66,9 +70,10 @@ public synchronized void stopListen(int dataRegionId, PipeRealtimeDataRegionSour //////////////////////////// listen to changes //////////////////////////// public synchronized void listenToTimePartitionGrow( - int dataRegionId, Pair newTimePartitionIdBound) { + final int dataRegionId, final Pair newTimePartitionIdBound) { boolean shouldBroadcastTimePartitionChange = false; - Pair oldTimePartitionIdBound = dataRegionId2TimePartitionIdBound.get(dataRegionId); + final Pair oldTimePartitionIdBound = + dataRegionId2TimePartitionIdBound.get(dataRegionId); if (Objects.isNull(oldTimePartitionIdBound)) { dataRegionId2TimePartitionIdBound.put(dataRegionId, newTimePartitionIdBound); @@ -84,13 +89,15 @@ public synchronized void listenToTimePartitionGrow( } if (shouldBroadcastTimePartitionChange) { - Map extractors = dataRegionId2Sources.get(dataRegionId); - if (Objects.isNull(extractors)) { + final Map sources = + dataRegionId2Sources.get(dataRegionId); + if (Objects.isNull(sources)) { return; } - Pair timePartitionIdBound = dataRegionId2TimePartitionIdBound.get(dataRegionId); - extractors.forEach( - (id, extractor) -> extractor.setDataRegionTimePartitionIdBound(timePartitionIdBound)); + final Pair timePartitionIdBound = + dataRegionId2TimePartitionIdBound.get(dataRegionId); + sources.forEach( + (id, source) -> source.setDataRegionTimePartitionIdBound(timePartitionIdBound)); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java index 882860cebbc77..910e721ce215f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java @@ -497,8 +497,7 @@ private List convertRelationalInsertRowNode(final RelationalInsertRowNod return Collections.emptyList(); } - final Tablet prunedTablet = - TabletColumnPruner.pruneTableModelTablet(tablet, databaseName, getColumnFilterMatcher()); + final Tablet prunedTablet = projectAndPruneTableModelTablet(tablet); return Objects.nonNull(prunedTablet) ? Collections.singletonList(prunedTablet) : Collections.emptyList(); @@ -569,8 +568,7 @@ private List convertRelationalInsertTabletNode(final RelationalInsertTab Objects.nonNull(validBitMaps) ? validBitMaps.toArray(new BitMap[0]) : null, node.getRowCount()); - final Tablet prunedTablet = - TabletColumnPruner.pruneTableModelTablet(tablet, databaseName, getColumnFilterMatcher()); + final Tablet prunedTablet = projectAndPruneTableModelTablet(tablet); return Objects.nonNull(prunedTablet) ? Collections.singletonList(prunedTablet) : Collections.emptyList(); @@ -584,6 +582,11 @@ private List convertRelationalInsertRowsNode(final RelationalInsertRowsN return tablets; } + private Tablet projectAndPruneTableModelTablet(final Tablet sourceTablet) { + return TabletColumnPruner.pruneTableModelTablet( + sourceTablet, databaseName, getColumnFilterMatcher()); + } + private Tablet buildTableModelTabletFromRow( final String tableName, final long time, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeEventBatches.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeEventBatches.java index 0cb81ca43e699..8a64b2eec2d36 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeEventBatches.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeEventBatches.java @@ -108,6 +108,8 @@ public boolean onEvent(final EnrichedEvent event, final Consumer generateSubscriptionEvents() { - if (!prepareTreeViewTabletProjectorForEmission()) { + if (!tableViewTabletProjectorProvider.prepareForEmission()) { return null; } resetForIteration(); @@ -217,16 +225,19 @@ private Pair> convertToTablets( return null; } - return pruneTablets(projectTreeViewTabletsIfNecessary(result)); + return pruneTablets(projectTableViewTabletsIfNecessary(result)); } - private Pair> projectTreeViewTabletsIfNecessary( + private Pair> projectTableViewTabletsIfNecessary( final Pair> tablets) { if (Objects.isNull(tablets) || Objects.nonNull(tablets.left) || Objects.isNull(tablets.right)) { return tablets; } - final TreeViewTabletProjector projector = getTreeViewTabletProjector(); + final TreeViewTabletProjector projector = + tableViewTabletProjectorProvider.prepareForEmission() + ? tableViewTabletProjectorProvider.getTreeViewTabletProjector() + : null; if (Objects.isNull(projector)) { return tablets; } @@ -243,90 +254,6 @@ private Pair> projectTreeViewTabletsIfNecessary( : new Pair<>(projector.getDatabaseName(), projectedTablets); } - private TreeViewTabletProjector getTreeViewTabletProjector() { - return prepareTreeViewTabletProjectorForEmission() ? treeViewTabletProjector : null; - } - - private boolean prepareTreeViewTabletProjectorForEmission() { - if (treeViewTabletProjectorInitialized) { - return true; - } - - synchronized (this) { - if (treeViewTabletProjectorInitialized) { - return true; - } - - final TopicConfig topicConfig = - SubscriptionAgent.topic() - .getTopicConfigs( - Collections.singleton(prefetchingQueue.getTopicName()), - SubscriptionAgent.consumer().isTableModel(prefetchingQueue.getConsumerGroupId())) - .get(prefetchingQueue.getTopicName()); - if (Objects.isNull(topicConfig)) { - return false; - } - if (!topicConfig.isTableTopic()) { - treeViewTabletProjectorInitialized = true; - return true; - } - - final String database = - topicConfig.getStringOrDefault( - TopicConstant.DATABASE_KEY, TopicConstant.DATABASE_DEFAULT_VALUE); - final String tableName = - topicConfig.getStringOrDefault( - TopicConstant.TABLE_KEY, TopicConstant.TABLE_DEFAULT_VALUE); - if (isDefaultTopicPattern(database, TopicConstant.DATABASE_DEFAULT_VALUE) - || isDefaultTopicPattern(tableName, TopicConstant.TABLE_DEFAULT_VALUE) - || !isLiteralTopicPattern(database) - || !isLiteralTopicPattern(tableName)) { - treeViewTabletProjectorInitialized = true; - return true; - } - - if (!isTreeCapturedByTopic(topicConfig) && topicConfig.isColumnFilterTrivial()) { - treeViewTabletProjectorInitialized = true; - return true; - } - - final TsTable table = DataNodeTableCache.getInstance().getTable(database, tableName, false); - if (Objects.isNull(table)) { - LOGGER.debug( - DataNodePipeMessages - .PIPE_LOG_SUBSCRIPTIONPIPETABLETEVENTBATCH_POSTPONE_EMITTING_SUBSCRIPTION_TABLET_BATCH_FOR_TOPIC_ARG_BECAUSE_TABLE_SCHEMA_ARG_ARG_IS_NOT_AVAILABLE_LOCALLY_996C618D, - prefetchingQueue.getTopicName(), - database, - tableName); - return false; - } - if (TreeViewSchema.isTreeViewTable(table)) { - treeViewTabletProjector = new TreeViewTabletProjector(database, table); - } - - treeViewTabletProjectorInitialized = true; - return true; - } - } - - private static boolean isDefaultTopicPattern(final String pattern, final String defaultPattern) { - return Objects.isNull(pattern) || defaultPattern.equals(pattern.trim()); - } - - private static boolean isLiteralTopicPattern(final String pattern) { - final String regexMetaCharacters = ".*+?[](){}\\|^$"; - return Objects.nonNull(pattern) - && pattern.chars().noneMatch(c -> regexMetaCharacters.indexOf((char) c) >= 0); - } - - private static boolean isTreeCapturedByTopic(final TopicConfig topicConfig) { - return topicConfig.getBooleanOrDefault( - Arrays.asList( - PipeSourceConstant.EXTRACTOR_CAPTURE_TREE_KEY, - PipeSourceConstant.SOURCE_CAPTURE_TREE_KEY), - false); - } - private Pair> pruneTablets(final Pair> tablets) { if (Objects.isNull(tablets) || Objects.isNull(tablets.left) || Objects.isNull(tablets.right)) { return tablets; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/TableViewTabletProjectorProvider.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/TableViewTabletProjectorProvider.java new file mode 100644 index 0000000000000..c76868cfc3547 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/TableViewTabletProjectorProvider.java @@ -0,0 +1,140 @@ +/* + * 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.subscription.event.batch; + +import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant; +import org.apache.iotdb.commons.schema.table.TreeViewSchema; +import org.apache.iotdb.commons.schema.table.TsTable; +import org.apache.iotdb.db.i18n.DataNodePipeMessages; +import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; +import org.apache.iotdb.db.subscription.agent.SubscriptionAgent; +import org.apache.iotdb.db.subscription.broker.SubscriptionPrefetchingTabletQueue; +import org.apache.iotdb.db.subscription.columnfilter.TreeViewTabletProjector; +import org.apache.iotdb.rpc.subscription.config.TopicConfig; +import org.apache.iotdb.rpc.subscription.config.TopicConstant; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Arrays; +import java.util.Collections; +import java.util.Objects; + +/** + * Resolves and caches the tree-view projector for one tablet batch. Keeping the provider scoped to + * the batch preserves its schema snapshot and the initialization checks of subsequent batches. + */ +class TableViewTabletProjectorProvider { + + private static final Logger LOGGER = + LoggerFactory.getLogger(SubscriptionPipeTabletEventBatch.class); + + private final SubscriptionPrefetchingTabletQueue prefetchingQueue; + private volatile boolean initialized; + private volatile TreeViewTabletProjector treeViewTabletProjector; + + TableViewTabletProjectorProvider(final SubscriptionPrefetchingTabletQueue prefetchingQueue) { + this.prefetchingQueue = prefetchingQueue; + } + + boolean prepareForEmission() { + if (initialized) { + return true; + } + + synchronized (this) { + if (initialized) { + return true; + } + + final TopicConfig topicConfig = + SubscriptionAgent.topic() + .getTopicConfigs( + Collections.singleton(prefetchingQueue.getTopicName()), + SubscriptionAgent.consumer().isTableModel(prefetchingQueue.getConsumerGroupId())) + .get(prefetchingQueue.getTopicName()); + if (Objects.isNull(topicConfig)) { + return false; + } + if (!topicConfig.isTableTopic()) { + initialized = true; + return true; + } + + final String database = + topicConfig.getStringOrDefault( + TopicConstant.DATABASE_KEY, TopicConstant.DATABASE_DEFAULT_VALUE); + final String tableName = + topicConfig.getStringOrDefault( + TopicConstant.TABLE_KEY, TopicConstant.TABLE_DEFAULT_VALUE); + if (isDefaultTopicPattern(database, TopicConstant.DATABASE_DEFAULT_VALUE) + || isDefaultTopicPattern(tableName, TopicConstant.TABLE_DEFAULT_VALUE) + || !isLiteralTopicPattern(database) + || !isLiteralTopicPattern(tableName)) { + initialized = true; + return true; + } + + if (!isTreeCapturedByTopic(topicConfig) && topicConfig.isColumnFilterTrivial()) { + initialized = true; + return true; + } + + final TsTable table = DataNodeTableCache.getInstance().getTable(database, tableName, false); + if (Objects.isNull(table)) { + LOGGER.debug( + DataNodePipeMessages + .PIPE_LOG_SUBSCRIPTIONPIPETABLETEVENTBATCH_POSTPONE_EMITTING_SUBSCRIPTION_TABLET_BATCH_FOR_TOPIC_ARG_BECAUSE_TABLE_SCHEMA_ARG_ARG_IS_NOT_AVAILABLE_LOCALLY_996C618D, + prefetchingQueue.getTopicName(), + database, + tableName); + return false; + } + if (TreeViewSchema.isTreeViewTable(table)) { + treeViewTabletProjector = new TreeViewTabletProjector(database, table); + } + + initialized = true; + return true; + } + } + + private static boolean isDefaultTopicPattern(final String pattern, final String defaultPattern) { + return Objects.isNull(pattern) || defaultPattern.equals(pattern.trim()); + } + + private static boolean isLiteralTopicPattern(final String pattern) { + final String regexMetaCharacters = ".*+?[](){}\\|^$"; + return Objects.nonNull(pattern) + && pattern.chars().noneMatch(c -> regexMetaCharacters.indexOf((char) c) >= 0); + } + + private static boolean isTreeCapturedByTopic(final TopicConfig topicConfig) { + return topicConfig.getBooleanOrDefault( + Arrays.asList( + PipeSourceConstant.EXTRACTOR_CAPTURE_TREE_KEY, + PipeSourceConstant.SOURCE_CAPTURE_TREE_KEY), + false); + } + + TreeViewTabletProjector getTreeViewTabletProjector() { + return treeViewTabletProjector; + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSourceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSourceTest.java index 991dab795f94a..e89678d4ce97b 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSourceTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSourceTest.java @@ -86,7 +86,7 @@ public void testDeletionOnlyCustomizeInitializesSourceContext() throws Exception Assert.assertEquals(false, getPrivateField(source, "shouldExtractInsertion")); Assert.assertEquals(true, getPrivateField(source, "shouldExtractDeletion")); Assert.assertNotNull(getPrivateField(source, "treePattern")); - Assert.assertNotNull(getPrivateField(source, "tablePattern")); + Assert.assertNotNull(getPrivateField(source, "dataTablePattern")); } @Test diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/event/batch/TableViewTabletProjectorProviderTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/event/batch/TableViewTabletProjectorProviderTest.java new file mode 100644 index 0000000000000..882f8f762e952 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/event/batch/TableViewTabletProjectorProviderTest.java @@ -0,0 +1,151 @@ +/* + * 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.subscription.event.batch; + +import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant; +import org.apache.iotdb.commons.pipe.config.constant.SystemConstant; +import org.apache.iotdb.commons.schema.table.TreeViewSchema; +import org.apache.iotdb.commons.schema.table.TsTable; +import org.apache.iotdb.commons.schema.table.column.FieldColumnSchema; +import org.apache.iotdb.commons.schema.table.column.TimeColumnSchema; +import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerGroupMeta; +import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerMeta; +import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta; +import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; +import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; +import org.apache.iotdb.db.subscription.agent.SubscriptionAgent; +import org.apache.iotdb.db.subscription.broker.SubscriptionPrefetchingTabletQueue; +import org.apache.iotdb.rpc.subscription.config.ConsumerConstant; +import org.apache.iotdb.rpc.subscription.config.TopicConstant; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.utils.Pair; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mockito; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; + +public class TableViewTabletProjectorProviderTest { + + private final String topicName = "projector_topic_" + UUID.randomUUID(); + private final String groupId = "projector_group_" + UUID.randomUUID(); + private final String database = "projector_db_" + UUID.randomUUID(); + private final List batches = new ArrayList<>(); + private SubscriptionPrefetchingTabletQueue queue; + + @Before + public void setUp() { + final ConsumerMeta consumer = + new ConsumerMeta( + "consumer", + 1L, + Collections.singletonMap( + ConsumerConstant.SQL_DIALECT_KEY, SystemConstant.SQL_DIALECT_TABLE_VALUE)); + Assert.assertNull( + SubscriptionAgent.consumer() + .handleSingleConsumerGroupMetaChanges(new ConsumerGroupMeta(groupId, 1L, consumer))); + Assert.assertTrue(SubscriptionAgent.consumer().isTableModel(groupId)); + queue = Mockito.mock(SubscriptionPrefetchingTabletQueue.class); + Mockito.when(queue.getTopicName()).thenReturn(topicName); + Mockito.when(queue.getConsumerGroupId()).thenReturn(groupId); + } + + @After + public void tearDown() { + for (final SubscriptionPipeTabletEventBatch batch : batches) { + batch.cleanUp(true); + } + SubscriptionAgent.topic().handleDropTopic(topicName); + SubscriptionAgent.consumer().handleDropConsumerGroup(groupId); + DataNodeTableCache.getInstance().invalid(database); + } + + @Test + public void testBatchWaitsForTopicAndSchemaThenRetainsItsProjector() { + final SubscriptionPipeTabletEventBatch batch = createBatch(); + Assert.assertNull(batch.generateSubscriptionEvents()); + + installTopic(true); + Assert.assertNull(batch.generateSubscriptionEvents()); + + final TsTable table = new TsTable("view"); + table.addProp(TreeViewSchema.TREE_PATH_PATTERN, "root.test.**"); + table.addColumnSchema(new TimeColumnSchema("time", TSDataType.TIMESTAMP)); + final FieldColumnSchema field = new FieldColumnSchema("temperature", TSDataType.DOUBLE); + TreeViewSchema.setOriginalName(field, "temperature"); + table.addColumnSchema(field); + DataNodeTableCache.getInstance().preUpdateTable(database, table, null); + DataNodeTableCache.getInstance().commitUpdateTable(database, table.getTableName(), null); + Assert.assertNotNull(batch.generateSubscriptionEvents()); + + DataNodeTableCache.getInstance().invalid(database); + final Tablet sourceTablet = + new Tablet( + "root.test.d1", + Collections.singletonList(new MeasurementSchema("temperature", TSDataType.DOUBLE)), + 1); + sourceTablet.addTimestamp(0, 1L); + sourceTablet.addValue("temperature", 0, 2.5); + batch.enrichedEvents.add( + new PipeRawTabletInsertionEvent( + false, "root.test", null, "root.test", sourceTablet, true, null, 0, null, null, false)); + Assert.assertNotNull(batch.generateSubscriptionEvents()); + final Pair> emittedTablets = batch.next(); + Assert.assertEquals(database, emittedTablets.left); + Assert.assertEquals(1, emittedTablets.right.size()); + Assert.assertEquals("view", emittedTablets.right.get(0).getTableName()); + Assert.assertEquals(2.5, ((double[]) emittedTablets.right.get(0).getValues()[0])[0], 0.0); + Assert.assertNull(createBatch().generateSubscriptionEvents()); + } + + @Test + public void testOrdinaryTableTopicDoesNotWaitForSchema() { + installTopic(false); + Assert.assertNotNull(createBatch().generateSubscriptionEvents()); + } + + private void installTopic(final boolean captureTree) { + final Map attributes = new HashMap<>(); + attributes.put(SystemConstant.SQL_DIALECT_KEY, "table"); + attributes.put(TopicConstant.DATABASE_KEY, database); + attributes.put(TopicConstant.TABLE_KEY, "view"); + attributes.put(PipeSourceConstant.SOURCE_CAPTURE_TREE_KEY, Boolean.toString(captureTree)); + Assert.assertNull( + SubscriptionAgent.topic() + .handleSingleTopicMetaChanges(new TopicMeta(topicName, 1L, attributes))); + } + + private SubscriptionPipeTabletEventBatch createBatch() { + final SubscriptionPipeTabletEventBatch batch = + new SubscriptionPipeTabletEventBatch(1, queue, 20, 1024); + batches.add(batch); + return batch; + } +} diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeSourceTreePatternUtils.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeSourceTreePatternUtils.java new file mode 100644 index 0000000000000..43643e5e10fb1 --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeSourceTreePatternUtils.java @@ -0,0 +1,60 @@ +/* + * 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.commons.pipe.config; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +public final class PipeSourceTreePatternUtils { + + private PipeSourceTreePatternUtils() { + // Utility class. + } + + /** Splits comma-separated patterns without splitting commas inside backquoted path nodes. */ + public static List splitPatternList(final String patternList) { + if (patternList == null || patternList.trim().isEmpty()) { + return Collections.emptyList(); + } + final List segments = new ArrayList<>(); + final StringBuilder current = new StringBuilder(); + boolean inBackticks = false; + for (final char c : patternList.toCharArray()) { + if (c == '`') { + inBackticks = !inBackticks; + current.append(c); + } else if (c == ',' && !inBackticks) { + final String singlePattern = current.toString().trim(); + if (!singlePattern.isEmpty()) { + segments.add(singlePattern); + } + current.setLength(0); + } else { + current.append(c); + } + } + final String lastPattern = current.toString().trim(); + if (!lastPattern.isEmpty()) { + segments.add(lastPattern); + } + return segments; + } +} diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TablePattern.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TablePattern.java index 578a1e8afe171..3487de2398312 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TablePattern.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TablePattern.java @@ -85,6 +85,10 @@ public boolean matchesTable(final String table) { return tablePattern == null || tablePattern.matcher(table).matches(); } + public boolean matchesDatabaseAndTable(final String database, final String table) { + return matchesDatabase(database) && matchesTable(table); + } + public String getDatabasePattern() { return databasePattern == null ? EXTRACTOR_DATABASE_NAME_DEFAULT_VALUE diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TreePattern.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TreePattern.java index 3c093b68c1723..92c150f988687 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TreePattern.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TreePattern.java @@ -23,6 +23,7 @@ import org.apache.iotdb.commons.i18n.PipeMessages; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.path.PathPatternUtil; +import org.apache.iotdb.commons.pipe.config.PipeSourceTreePatternUtils; import org.apache.iotdb.commons.pipe.datastructure.visibility.VisibilityUtils; import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; @@ -181,16 +182,25 @@ public static TreePattern parsePipePatternFromSourceParametersInternal( List exclusionPatterns = parseExclusionPatternList(sourceParameters, isTreeModelDataAllowedToBeCaptured); - // 3. Optimize the lists: remove redundant patterns (e.g., if "root.**" exists, "root.db" is + return buildPatternWithExclusion( + isTreeModelDataAllowedToBeCaptured, sourceParameters, inclusionPatterns, exclusionPatterns); + } + + private static TreePattern buildPatternWithExclusion( + final boolean isTreeModelDataAllowedToBeCaptured, + final PipeParameters sourceParameters, + List inclusionPatterns, + List exclusionPatterns) { + // Optimize the lists: remove redundant patterns (e.g., if "root.**" exists, "root.db" is // redundant) inclusionPatterns = optimizePatterns(inclusionPatterns); exclusionPatterns = optimizePatterns(exclusionPatterns); - // 4. Prune inclusion patterns: if an inclusion pattern is fully covered by an exclusion + // Prune inclusion patterns: if an inclusion pattern is fully covered by an exclusion // pattern, remove it inclusionPatterns = pruneInclusionPatterns(inclusionPatterns, exclusionPatterns); - // 5. Check if the resulting inclusion pattern is empty + // Check if the resulting inclusion pattern is empty if (inclusionPatterns.isEmpty()) { final String msg = String.format( @@ -213,11 +223,11 @@ public static TreePattern parsePipePatternFromSourceParametersInternal( throw new PipeException(msg); } - // 6. Prune exclusion patterns: if an exclusion pattern does not overlap with + // Prune exclusion patterns: if an exclusion pattern does not overlap with // ANY of the remaining inclusion patterns, it is useless and should be removed. exclusionPatterns = pruneIrrelevantExclusions(inclusionPatterns, exclusionPatterns); - // 7. Build final patterns + // Build final patterns final TreePattern finalInclusionPattern = buildUnionPattern(isTreeModelDataAllowedToBeCaptured, inclusionPatterns); @@ -228,7 +238,7 @@ public static TreePattern parsePipePatternFromSourceParametersInternal( final TreePattern finalExclusionPattern = buildUnionPattern(isTreeModelDataAllowedToBeCaptured, exclusionPatterns); - // 8. Combine inclusion and exclusion + // Combine inclusion and exclusion if (finalInclusionPattern instanceof IoTDBTreePatternOperations && finalExclusionPattern instanceof IoTDBTreePatternOperations) { return new WithExclusionIoTDBTreePattern( @@ -689,31 +699,9 @@ private static List parseMultiplePatterns( return Collections.singletonList(patternSupplier.apply(pattern)); } - final List patterns = new ArrayList<>(); - final StringBuilder currentPattern = new StringBuilder(); - boolean inBackticks = false; - - for (final char c : pattern.toCharArray()) { - if (c == '`') { - inBackticks = !inBackticks; - currentPattern.append(c); - } else if (c == ',' && !inBackticks) { - final String singlePattern = currentPattern.toString().trim(); - if (!singlePattern.isEmpty()) { - patterns.add(patternSupplier.apply(singlePattern)); - } - currentPattern.setLength(0); - } else { - currentPattern.append(c); - } - } - - final String lastPattern = currentPattern.toString().trim(); - if (!lastPattern.isEmpty()) { - patterns.add(patternSupplier.apply(lastPattern)); - } - - return patterns; + return PipeSourceTreePatternUtils.splitPatternList(pattern).stream() + .map(patternSupplier) + .collect(Collectors.toList()); } /** diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSink.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSink.java index 601ad31a0f42e..2bbd87d19b451 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSink.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSink.java @@ -154,7 +154,9 @@ @TreeModel @TableModel -public abstract class IoTDBSink implements PipeConnector, PipeConnectorWithEventDiscard { +@SuppressWarnings("java:S100") +public abstract class IoTDBSink + implements PipeConnector, PipeBatchMetricsSettable, PipeConnectorWithEventDiscard { private static final String PARSE_URL_ERROR_FORMATTER = PipeMessages.PARSE_URL_ERROR; private static final String PARSE_URL_ERROR_MESSAGE = PipeMessages.PARSE_URL_ERROR_MESSAGE; @@ -714,6 +716,7 @@ public String getSinkTaskId() { return sinkTaskId; } + @Override public void setTabletBatchSizeHistogram(Histogram tabletBatchSizeHistogram) { // do nothing by default } @@ -722,10 +725,12 @@ public void setSchemaBatchSizeHistogram(Histogram schemaBatchSizeHistogram) { // do nothing by default } + @Override public void setTsFileBatchSizeHistogram(Histogram tsFileBatchSizeHistogram) { // do nothing by default } + @Override public void setTabletBatchTimeIntervalHistogram(Histogram tabletBatchTimeIntervalHistogram) { // do nothing by default } @@ -734,11 +739,13 @@ public void setSchemaBatchTimeIntervalHistogram(Histogram schemaBatchTimeInterva // do nothing by default } + @Override public void setTsFileBatchTimeIntervalHistogram(Histogram tsFileBatchTimeIntervalHistogram) { // do nothing by default } - public void setBatchEventSizeHistogram(Histogram tsFileBatchTimeIntervalHistogram) { + @Override + public void setBatchEventSizeHistogram(Histogram eventSizeHistogram) { // do nothing by default } } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/PipeBatchMetricsSettable.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/PipeBatchMetricsSettable.java new file mode 100644 index 0000000000000..9cfed40e2c8d5 --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/PipeBatchMetricsSettable.java @@ -0,0 +1,39 @@ +/* + * 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.commons.pipe.sink.protocol; + +import org.apache.iotdb.metrics.type.Histogram; + +/** + * Interface for components that can accept pipe batch metrics (histograms). Implemented by sinks + * and builders that need to record tablet/tsfile batch size and time interval metrics. + */ +public interface PipeBatchMetricsSettable { + + void setTabletBatchSizeHistogram(Histogram tabletBatchSizeHistogram); + + void setTsFileBatchSizeHistogram(Histogram tsFileBatchSizeHistogram); + + void setTabletBatchTimeIntervalHistogram(Histogram tabletBatchTimeIntervalHistogram); + + void setTsFileBatchTimeIntervalHistogram(Histogram tsFileBatchTimeIntervalHistogram); + + void setBatchEventSizeHistogram(Histogram eventSizeHistogram); +}