diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/source/IoTDBConfigRegionSource.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/source/IoTDBConfigRegionSource.java index 656f517c4e08a..4a73009195c35 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/source/IoTDBConfigRegionSource.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/source/IoTDBConfigRegionSource.java @@ -65,9 +65,11 @@ import java.nio.file.Paths; import java.util.Collections; import java.util.HashSet; +import java.util.List; import java.util.Objects; import java.util.Optional; import java.util.Set; +import java.util.stream.Collectors; import static org.apache.iotdb.commons.conf.IoTDBConstant.MULTI_LEVEL_PATH_WILDCARD; import static org.apache.iotdb.commons.conf.IoTDBConstant.PATH_ROOT; @@ -262,26 +264,29 @@ protected PipeWritePlanEvent getNextEventInCurrentSnapshot() { } @Override - protected Optional trimRealtimeEventByPrivilege( - final PipeWritePlanEvent event) { + protected List trimRealtimeEventByPrivilege(final PipeWritePlanEvent event) { final ConfigPhysicalPlan plan = ((PipeConfigRegionWritePlanEvent) event).getConfigPhysicalPlan(); final Boolean isTableDatabasePlan = isTableDatabasePlan(plan); - Optional result = Optional.of(plan); - if (!Boolean.TRUE.equals(isTableDatabasePlan)) { - result = treePrivilegeParseVisitor.process(plan, userEntity); - } - if (result.isPresent() && !Boolean.FALSE.equals(isTableDatabasePlan)) { - result = TABLE_PRIVILEGE_PARSE_VISITOR.process(result.get(), userEntity); - } - if (result.isPresent()) { - return Optional.of( - new PipeConfigRegionWritePlanEvent(result.get(), event.isGeneratedByPipe())); + List result = + Boolean.TRUE.equals(isTableDatabasePlan) + ? Collections.singletonList(plan) + : treePrivilegeParseVisitor.process(plan, userEntity); + if (!Boolean.FALSE.equals(isTableDatabasePlan)) { + result = + result.stream() + .map(parsedPlan -> TABLE_PRIVILEGE_PARSE_VISITOR.process(parsedPlan, userEntity)) + .filter(Optional::isPresent) + .map(Optional::get) + .collect(Collectors.toList()); } - if (skipIfNoPrivileges) { - return Optional.empty(); + if (result.isEmpty() && !skipIfNoPrivileges) { + throw new AccessDeniedException(ManagerMessages.NOT_HAS_PRIVILEGE_TO_TRANSFER_PLAN + plan); } - throw new AccessDeniedException(ManagerMessages.NOT_HAS_PRIVILEGE_TO_TRANSFER_PLAN + plan); + return result.stream() + .map( + parsedPlan -> new PipeConfigRegionWritePlanEvent(parsedPlan, event.isGeneratedByPipe())) + .collect(Collectors.toList()); } @Override diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/source/PipeConfigTreePatternParseVisitor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/source/PipeConfigTreePatternParseVisitor.java index 859772fa78aed..330ec10069fd4 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/source/PipeConfigTreePatternParseVisitor.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/source/PipeConfigTreePatternParseVisitor.java @@ -61,6 +61,8 @@ import java.util.stream.IntStream; import java.util.stream.Stream; +import static org.apache.iotdb.commons.conf.IoTDBConstant.MULTI_LEVEL_PATH_WILDCARD; + /** * The {@link PipeConfigTreePatternParseVisitor} will transform the schema {@link * ConfigPhysicalPlan}s using {@link IoTDBTreePatternOperations}. Rule: @@ -303,16 +305,25 @@ public Optional visitPipeDeactivateTemplate( @Override public Optional visitTTL( final SetTTLPlan setTTLPlan, final IoTDBTreePatternOperations pattern) { - final PartialPath partialPath = new PartialPath(setTTLPlan.getPathPattern()); + final PartialPath originalPath = new PartialPath(setTTLPlan.getPathPattern()); + final PartialPath partialPath = + setTTLPlan.isDataBase() ? originalPath.concatNode(MULTI_LEVEL_PATH_WILDCARD) : originalPath; final List intersectionList = pattern.matchPrefixPath(partialPath.getFullPath()) ? Collections.singletonList(partialPath) : pattern.getIntersection(partialPath); - // The intersectionList is either a singleton list or an empty list, because the pipe - // pattern and TTL path are each either a prefix path or a full path - return !intersectionList.isEmpty() - ? Optional.of(new SetTTLPlan(intersectionList.get(0).getNodes(), setTTLPlan.getTTL())) - : Optional.empty(); + // Both the pipe pattern and the TTL path are prefix paths or full paths. + if (intersectionList.isEmpty()) { + return Optional.empty(); + } + final PartialPath intersection = intersectionList.get(0); + final boolean retainDatabase = setTTLPlan.isDataBase() && intersection.equals(partialPath); + final SetTTLPlan result = + new SetTTLPlan( + retainDatabase ? originalPath.getNodes() : intersection.getNodes(), + setTTLPlan.getTTL()); + result.setDataBase(retainDatabase); + return Optional.of(result); } @Override diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/source/PipeConfigTreePrivilegeParseVisitor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/source/PipeConfigTreePrivilegeParseVisitor.java index aa81c0681ba7d..c9b49227ed27a 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/source/PipeConfigTreePrivilegeParseVisitor.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/source/PipeConfigTreePrivilegeParseVisitor.java @@ -66,12 +66,13 @@ import java.util.Map; import java.util.Objects; import java.util.Optional; +import java.util.stream.Collectors; import static org.apache.iotdb.commons.conf.IoTDBConstant.MULTI_LEVEL_PATH_WILDCARD; import static org.apache.iotdb.commons.schema.SchemaConstant.ALL_MATCH_SCOPE; public class PipeConfigTreePrivilegeParseVisitor - extends ConfigPhysicalPlanVisitor, IAuditEntity> { + extends ConfigPhysicalPlanVisitor, IAuditEntity> { private static final Logger LOGGER = LoggerFactory.getLogger(PipeConfigTreePrivilegeParseVisitor.class); private final boolean skip; @@ -81,68 +82,68 @@ public class PipeConfigTreePrivilegeParseVisitor } @Override - public Optional visitPlan( + public List visitPlan( final ConfigPhysicalPlan plan, final IAuditEntity context) { - return Optional.of(plan); + return Collections.singletonList(plan); } @Override - public Optional visitCreateDatabase( + public List visitCreateDatabase( final DatabaseSchemaPlan createDatabasePlan, final IAuditEntity userEntity) { return canReadSysSchema(createDatabasePlan.getSchema().getName(), userEntity, true) - ? Optional.of(createDatabasePlan) - : Optional.empty(); + ? Collections.singletonList(createDatabasePlan) + : Collections.emptyList(); } @Override - public Optional visitAlterDatabase( + public List visitAlterDatabase( final DatabaseSchemaPlan alterDatabasePlan, final IAuditEntity userEntity) { return canReadSysSchema(alterDatabasePlan.getSchema().getName(), userEntity, true) - ? Optional.of(alterDatabasePlan) - : Optional.empty(); + ? Collections.singletonList(alterDatabasePlan) + : Collections.emptyList(); } @Override - public Optional visitDeleteDatabase( + public List visitDeleteDatabase( final DeleteDatabasePlan deleteDatabasePlan, final IAuditEntity userEntity) { return canReadSysSchema(deleteDatabasePlan.getName(), userEntity, true) - ? Optional.of(deleteDatabasePlan) - : Optional.empty(); + ? Collections.singletonList(deleteDatabasePlan) + : Collections.emptyList(); } @Override - public Optional visitCreateSchemaTemplate( + public List visitCreateSchemaTemplate( final CreateSchemaTemplatePlan createSchemaTemplatePlan, final IAuditEntity userEntity) { return canShowSchemaTemplate(createSchemaTemplatePlan.getTemplate().getName(), userEntity) - ? Optional.of(createSchemaTemplatePlan) - : Optional.empty(); + ? Collections.singletonList(createSchemaTemplatePlan) + : Collections.emptyList(); } @Override - public Optional visitCommitSetSchemaTemplate( + public List visitCommitSetSchemaTemplate( final CommitSetSchemaTemplatePlan commitSetSchemaTemplatePlan, final IAuditEntity userEntity) { return canReadSysSchema(commitSetSchemaTemplatePlan.getPath(), userEntity, false) - ? Optional.of(commitSetSchemaTemplatePlan) - : Optional.empty(); + ? Collections.singletonList(commitSetSchemaTemplatePlan) + : Collections.emptyList(); } @Override - public Optional visitPipeUnsetSchemaTemplate( + public List visitPipeUnsetSchemaTemplate( final PipeUnsetSchemaTemplatePlan pipeUnsetSchemaTemplatePlan, final IAuditEntity userEntity) { return canReadSysSchema(pipeUnsetSchemaTemplatePlan.getPath(), userEntity, false) - ? Optional.of(pipeUnsetSchemaTemplatePlan) - : Optional.empty(); + ? Collections.singletonList(pipeUnsetSchemaTemplatePlan) + : Collections.emptyList(); } @Override - public Optional visitExtendSchemaTemplate( + public List visitExtendSchemaTemplate( final ExtendSchemaTemplatePlan extendSchemaTemplatePlan, final IAuditEntity userEntity) { return canShowSchemaTemplate( extendSchemaTemplatePlan.getTemplateExtendInfo().getTemplateName(), userEntity) - ? Optional.of(extendSchemaTemplatePlan) - : Optional.empty(); + ? Collections.singletonList(extendSchemaTemplatePlan) + : Collections.emptyList(); } public boolean canShowSchemaTemplate(final String templateName, final IAuditEntity userEntity) { @@ -171,39 +172,51 @@ public boolean canReadSysSchema( } @Override - public Optional visitGrantUser( + public List visitGrantUser( final AuthorTreePlan grantUserPlan, final IAuditEntity userEntity) { - return visitUserPlan(grantUserPlan, userEntity); + return visitUserPlan(grantUserPlan, userEntity) + .map(Collections::singletonList) + .orElseGet(Collections::emptyList); } @Override - public Optional visitRevokeUser( + public List visitRevokeUser( final AuthorTreePlan revokeUserPlan, final IAuditEntity userEntity) { - return visitUserPlan(revokeUserPlan, userEntity); + return visitUserPlan(revokeUserPlan, userEntity) + .map(Collections::singletonList) + .orElseGet(Collections::emptyList); } @Override - public Optional visitGrantRole( + public List visitGrantRole( final AuthorTreePlan grantRolePlan, final IAuditEntity userEntity) { - return visitRolePlan(grantRolePlan, userEntity); + return visitRolePlan(grantRolePlan, userEntity) + .map(Collections::singletonList) + .orElseGet(Collections::emptyList); } @Override - public Optional visitRevokeRole( + public List visitRevokeRole( final AuthorTreePlan revokeRolePlan, final IAuditEntity userEntity) { - return visitRolePlan(revokeRolePlan, userEntity); + return visitRolePlan(revokeRolePlan, userEntity) + .map(Collections::singletonList) + .orElseGet(Collections::emptyList); } @Override - public Optional visitGrantRoleToUser( + public List visitGrantRoleToUser( final AuthorTreePlan grantRoleToUserPlan, final IAuditEntity userEntity) { - return visitUserRolePlan(grantRoleToUserPlan, userEntity); + return visitUserRolePlan(grantRoleToUserPlan, userEntity) + .map(Collections::singletonList) + .orElseGet(Collections::emptyList); } @Override - public Optional visitRevokeRoleFromUser( + public List visitRevokeRoleFromUser( final AuthorTreePlan revokeRoleFromUserPlan, final IAuditEntity userEntity) { - return visitUserRolePlan(revokeRoleFromUserPlan, userEntity); + return visitUserRolePlan(revokeRoleFromUserPlan, userEntity) + .map(Collections::singletonList) + .orElseGet(Collections::emptyList); } public static Optional visitUserRolePlan( @@ -260,7 +273,7 @@ public static Optional visitRolePlan( } @Override - public Optional visitPipeDeleteTimeSeries( + public List visitPipeDeleteTimeSeries( final PipeDeleteTimeSeriesPlan pipeDeleteTimeSeriesPlan, final IAuditEntity userEntity) { final CNAuditLogger logger = ConfigNode.getInstance().getConfigManager().getAuditLogger(); final PathPatternTree originalTree = @@ -278,16 +291,16 @@ public Optional visitPipeDeleteTimeSeries( final boolean result = !intersectedTree.isEmpty(); logger.recordObjectAuthenticationAuditLog(userEntity.setResult(result), () -> auditObject); return result - ? Optional.of(new PipeDeleteTimeSeriesPlan(intersectedTree.serialize())) - : Optional.empty(); + ? Collections.singletonList(new PipeDeleteTimeSeriesPlan(intersectedTree.serialize())) + : Collections.emptyList(); } catch (final IOException e) { LOGGER.warn(ManagerMessages.SERIALIZATION_FAILED_FOR_THE_DELETE_TIME_SERIES_PLAN_IN_PIPE, e); logger.recordObjectAuthenticationAuditLog(userEntity.setResult(false), () -> auditObject); - return Optional.empty(); + return Collections.emptyList(); } catch (final AuthException e) { logger.recordObjectAuthenticationAuditLog(userEntity.setResult(false), () -> auditObject); if (skip) { - return Optional.empty(); + return Collections.emptyList(); } else { throw new AccessDeniedException( ManagerMessages.NOT_HAS_PRIVILEGE_TO_TRANSFER_PLAN + pipeDeleteTimeSeriesPlan); @@ -296,7 +309,7 @@ public Optional visitPipeDeleteTimeSeries( } @Override - public Optional visitPipeDeleteLogicalView( + public List visitPipeDeleteLogicalView( final PipeDeleteLogicalViewPlan pipeDeleteLogicalViewPlan, final IAuditEntity userEntity) { final CNAuditLogger logger = ConfigNode.getInstance().getConfigManager().getAuditLogger(); final PathPatternTree originalTree = @@ -314,16 +327,16 @@ public Optional visitPipeDeleteLogicalView( final boolean result = !intersectedTree.isEmpty(); logger.recordObjectAuthenticationAuditLog(userEntity.setResult(result), () -> auditObject); return result - ? Optional.of(new PipeDeleteLogicalViewPlan(intersectedTree.serialize())) - : Optional.empty(); + ? Collections.singletonList(new PipeDeleteLogicalViewPlan(intersectedTree.serialize())) + : Collections.emptyList(); } catch (final IOException e) { LOGGER.warn(ManagerMessages.SERIALIZATION_FAILED_FOR_THE_DELETE_TIME_SERIES_PLAN_IN_PIPE, e); logger.recordObjectAuthenticationAuditLog(userEntity.setResult(false), () -> auditObject); - return Optional.empty(); + return Collections.emptyList(); } catch (final AuthException e) { logger.recordObjectAuthenticationAuditLog(userEntity.setResult(false), () -> auditObject); if (skip) { - return Optional.empty(); + return Collections.emptyList(); } else { throw new AccessDeniedException( ManagerMessages.NOT_HAS_PRIVILEGE_TO_TRANSFER_PLAN + pipeDeleteLogicalViewPlan); @@ -332,7 +345,7 @@ public Optional visitPipeDeleteLogicalView( } @Override - public Optional visitPipeAlterEncodingCompressor( + public List visitPipeAlterEncodingCompressor( final PipeAlterEncodingCompressorPlan pipeAlterEncodingCompressor, final IAuditEntity userEntity) { final CNAuditLogger logger = ConfigNode.getInstance().getConfigManager().getAuditLogger(); @@ -351,21 +364,21 @@ public Optional visitPipeAlterEncodingCompressor( final boolean result = !intersectedTree.isEmpty(); logger.recordObjectAuthenticationAuditLog(userEntity.setResult(result), () -> auditObject); return result - ? Optional.of( + ? Collections.singletonList( new PipeAlterEncodingCompressorPlan( intersectedTree.serialize(), pipeAlterEncodingCompressor.getEncoding(), pipeAlterEncodingCompressor.getCompressor(), pipeAlterEncodingCompressor.isMayAlterAudit())) - : Optional.empty(); + : Collections.emptyList(); } catch (final IOException e) { LOGGER.warn(ManagerMessages.SERIALIZATION_FAILED_FOR_THE_DELETE_TIME_SERIES_PLAN_IN_PIPE, e); logger.recordObjectAuthenticationAuditLog(userEntity.setResult(false), () -> auditObject); - return Optional.empty(); + return Collections.emptyList(); } catch (final AuthException e) { logger.recordObjectAuthenticationAuditLog(userEntity.setResult(false), () -> auditObject); if (skip) { - return Optional.empty(); + return Collections.emptyList(); } else { throw new AccessDeniedException( ManagerMessages.NOT_HAS_PRIVILEGE_TO_TRANSFER_PLAN + pipeAlterEncodingCompressor); @@ -374,7 +387,7 @@ public Optional visitPipeAlterEncodingCompressor( } @Override - public Optional visitPipeDeactivateTemplate( + public List visitPipeDeactivateTemplate( final PipeDeactivateTemplatePlan pipeDeactivateTemplatePlan, final IAuditEntity userEntity) { final CNAuditLogger logger = ConfigNode.getInstance().getConfigManager().getAuditLogger(); userEntity.setPrivilegeType(PrivilegeType.READ_SCHEMA); @@ -396,12 +409,12 @@ public Optional visitPipeDeactivateTemplate( final boolean result = !newTemplateSetInfo.isEmpty(); logger.recordObjectAuthenticationAuditLog(userEntity.setResult(result), () -> auditObject); return !newTemplateSetInfo.isEmpty() - ? Optional.of(new PipeDeactivateTemplatePlan(newTemplateSetInfo)) - : Optional.empty(); + ? Collections.singletonList(new PipeDeactivateTemplatePlan(newTemplateSetInfo)) + : Collections.emptyList(); } catch (final AuthException e) { logger.recordObjectAuthenticationAuditLog(userEntity.setResult(false), () -> auditObject); if (skip) { - return Optional.empty(); + return Collections.emptyList(); } else { throw new AccessDeniedException( ManagerMessages.NOT_HAS_PRIVILEGE_TO_TRANSFER_PLAN + pipeDeactivateTemplatePlan); @@ -410,27 +423,41 @@ public Optional visitPipeDeactivateTemplate( } @Override - public Optional visitTTL( + public List visitTTL( final SetTTLPlan setTTLPlan, final IAuditEntity userEntity) { final CNAuditLogger logger = ConfigNode.getInstance().getConfigManager().getAuditLogger(); userEntity.setPrivilegeType(PrivilegeType.READ_SCHEMA); final String auditObject = Arrays.toString(setTTLPlan.getPathPattern()); try { + // Database TTL also applies to descendants for compatibility with device TTL. + final PartialPath originalPath = new PartialPath(setTTLPlan.getPathPattern()); + final PartialPath effectivePath = + setTTLPlan.isDataBase() + ? originalPath.concatNode(MULTI_LEVEL_PATH_WILDCARD) + : originalPath; final List paths = - getAllIntersectedPatterns( - new PartialPath(setTTLPlan.getPathPattern()), userEntity, setTTLPlan); - // The intersectionList is either a singleton list or an empty list, because the pipe - // pattern and TTL path are each either a prefix path or a full path - final boolean result = - !paths.isEmpty() && paths.get(0).getNodeLength() == setTTLPlan.getPathPattern().length; - logger.recordObjectAuthenticationAuditLog(userEntity.setResult(result), () -> auditObject); - return result - ? Optional.of(new SetTTLPlan(paths.get(0).getNodes(), setTTLPlan.getTTL())) - : Optional.empty(); + getAllIntersectedPatterns(effectivePath, userEntity, setTTLPlan).stream() + .filter(effectivePath::include) + .collect(Collectors.toList()); + logger.recordObjectAuthenticationAuditLog( + userEntity.setResult(!paths.isEmpty()), () -> auditObject); + return paths.stream() + .map( + path -> { + final boolean retainDatabase = + setTTLPlan.isDataBase() && path.equals(effectivePath); + final SetTTLPlan result = + new SetTTLPlan( + retainDatabase ? originalPath.getNodes() : path.getNodes(), + setTTLPlan.getTTL()); + result.setDataBase(retainDatabase); + return result; + }) + .collect(Collectors.toList()); } catch (final AuthException e) { logger.recordObjectAuthenticationAuditLog(userEntity.setResult(false), () -> auditObject); if (skip) { - return Optional.empty(); + return Collections.emptyList(); } else { throw new AccessDeniedException( ManagerMessages.NOT_HAS_PRIVILEGE_TO_TRANSFER_PLAN + setTTLPlan); @@ -439,7 +466,7 @@ public Optional visitTTL( } @Override - public Optional visitPipeAlterTimeSeries( + public List visitPipeAlterTimeSeries( final PipeAlterTimeSeriesPlan pipeAlterTimeSeriesPlan, final IAuditEntity userEntity) { final CNAuditLogger logger = ConfigNode.getInstance().getConfigManager().getAuditLogger(); userEntity.setPrivilegeType(PrivilegeType.READ_SCHEMA); @@ -457,15 +484,15 @@ public Optional visitPipeAlterTimeSeries( == pipeAlterTimeSeriesPlan.getMeasurementPath().getNodeLength(); logger.recordObjectAuthenticationAuditLog(userEntity.setResult(result), () -> auditObject); return result - ? Optional.of( + ? Collections.singletonList( new PipeAlterTimeSeriesPlan( new MeasurementPath(paths.get(0).getNodes()), pipeAlterTimeSeriesPlan.getOperationType(), pipeAlterTimeSeriesPlan.getDataType())) - : Optional.empty(); + : Collections.emptyList(); } catch (final AuthException e) { if (skip) { - return Optional.empty(); + return Collections.emptyList(); } else { throw new AccessDeniedException( ManagerMessages.NOT_HAS_PRIVILEGE_TO_TRANSFER_PLAN + pipeAlterTimeSeriesPlan); @@ -479,8 +506,19 @@ private List getAllIntersectedPatterns( final PathPatternTree thisPatternTree = new PathPatternTree(); thisPatternTree.appendPathPattern(partialPath); thisPatternTree.constructTree(); - final PathPatternTree intersectedTree = - thisPatternTree.intersectWithFullPathPrefixTree(getAuthorizedPTree(userEntity)); + final PathPatternTree intersectedTree = new PathPatternTree(); + // Intersect each authorization separately so an unmatched prefix cannot leave an ancestor + // path in the result tree. + for (final PartialPath authorizedPath : getAuthorizedPTree(userEntity).getAllPathPatterns()) { + if (authorizedPath.endWithMultiLevelWildcard()) { + partialPath + .intersectWithPrefixPattern(authorizedPath) + .forEach(intersectedTree::appendPathPattern); + } else if (partialPath.matchFullPath(authorizedPath)) { + intersectedTree.appendPathPattern(authorizedPath); + } + } + intersectedTree.constructTree(); if (!skip && !thisPatternTree.equals(intersectedTree)) { throw new AccessDeniedException(ManagerMessages.NOT_HAS_PRIVILEGE_TO_TRANSFER_PLAN + plan); } diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/source/PipeConfigTreePatternParseVisitorTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/source/PipeConfigTreePatternParseVisitorTest.java index cdd2a1f484067..7f68578a02d1d 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/source/PipeConfigTreePatternParseVisitorTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/source/PipeConfigTreePatternParseVisitorTest.java @@ -476,6 +476,37 @@ public void testSetTTL() throws IllegalPathException { Assert.assertEquals(Long.MAX_VALUE, plan.getTTL()); } + @Test + public void testSetDatabaseTTLTrimsToPipePattern() { + final SetTTLPlan plan = new SetTTLPlan(new String[] {"root", "db"}, 100); + plan.setDataBase(true); + final SetTTLPlan trimmed = + (SetTTLPlan) + IoTDBConfigRegionSource.TREE_PATTERN_PARSE_VISITOR + .visitTTL(plan, prefixPathPattern) + .orElseThrow(AssertionError::new); + + Assert.assertArrayEquals(new String[] {"root", "db", "device", "**"}, trimmed.getPathPattern()); + Assert.assertEquals(100, trimmed.getTTL()); + Assert.assertFalse(trimmed.isDataBase()); + Assert.assertArrayEquals(new String[] {"root", "db"}, plan.getPathPattern()); + Assert.assertTrue(plan.isDataBase()); + } + + @Test + public void testSetDatabaseTTLRetainsCompatibilityFlag() { + final SetTTLPlan plan = new SetTTLPlan(new String[] {"root", "db"}, 100); + plan.setDataBase(true); + final SetTTLPlan parsed = + (SetTTLPlan) + IoTDBConfigRegionSource.TREE_PATTERN_PARSE_VISITOR + .visitTTL(plan, new IoTDBTreePattern("root.**")) + .orElseThrow(AssertionError::new); + + Assert.assertEquals(plan, parsed); + Assert.assertNotSame(plan, parsed); + } + @Test public void testPipeAlterTimeSeries() throws IllegalPathException { final PipeAlterTimeSeriesPlan plan = diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/source/PipeConfigTreePrivilegeParseVisitorTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/source/PipeConfigTreePrivilegeParseVisitorTest.java index d88a13a251759..d760f5f14cc29 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/source/PipeConfigTreePrivilegeParseVisitorTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/source/PipeConfigTreePrivilegeParseVisitorTest.java @@ -27,8 +27,10 @@ import org.apache.iotdb.commons.exception.auth.AccessDeniedException; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.path.PathPatternTree; +import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern; import org.apache.iotdb.commons.schema.template.Template; import org.apache.iotdb.commons.utils.StatusUtils; +import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan; import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType; import org.apache.iotdb.confignode.consensus.request.write.auth.AuthorTreePlan; import org.apache.iotdb.confignode.consensus.request.write.database.DatabaseSchemaPlan; @@ -55,11 +57,14 @@ import java.io.IOException; import java.lang.reflect.Field; import java.nio.ByteBuffer; +import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Objects; import java.util.function.BiFunction; +import java.util.stream.Collectors; public class PipeConfigTreePrivilegeParseVisitorTest { private static final UserEntity FAKE_USER_ENTITY = new UserEntity(0L, "", ""); @@ -98,13 +103,13 @@ public void testCanReadSysSchema() { && privilegeUnion.getPaths().stream().allMatch(path -> path.equals("root.db"))); Assert.assertTrue(skipVisitor.canReadSysSchema("root.db", FAKE_USER_ENTITY, true)); Assert.assertFalse(skipVisitor.canReadSysSchema("root.db", FAKE_USER_ENTITY, false)); - Assert.assertFalse( + Assert.assertTrue( throwVisitor .visitCreateDatabase( new DatabaseSchemaPlan( ConfigPhysicalPlanType.CreateDatabase, new TDatabaseSchema("root.db1")), FAKE_USER_ENTITY) - .isPresent()); + .isEmpty()); permissionManager.setUserPrivilege( (userName, privilegeUnion) -> @@ -119,64 +124,64 @@ public void testAuthPrivilege() { permissionManager.setUserPrivilege( (userName, privilegeUnion) -> privilegeUnion.getPrivilegeType() == PrivilegeType.MANAGE_USER); - Assert.assertTrue( + Assert.assertFalse( skipVisitor .visitGrantUser(new AuthorTreePlan(ConfigPhysicalPlanType.GrantUser), FAKE_USER_ENTITY) - .isPresent()); - Assert.assertTrue( + .isEmpty()); + Assert.assertFalse( skipVisitor .visitRevokeUser( new AuthorTreePlan(ConfigPhysicalPlanType.RevokeUser), FAKE_USER_ENTITY) - .isPresent()); - Assert.assertFalse( + .isEmpty()); + Assert.assertTrue( skipVisitor .visitGrantRole(new AuthorTreePlan(ConfigPhysicalPlanType.GrantRole), FAKE_USER_ENTITY) - .isPresent()); - Assert.assertFalse( + .isEmpty()); + Assert.assertTrue( skipVisitor .visitRevokeRole( new AuthorTreePlan(ConfigPhysicalPlanType.RevokeRole), FAKE_USER_ENTITY) - .isPresent()); + .isEmpty()); permissionManager.setUserPrivilege((userName, privilegeUnion) -> false); AuthorTreePlan plan = new AuthorTreePlan(ConfigPhysicalPlanType.GrantUser); plan.setUserName(""); - Assert.assertTrue(skipVisitor.visitGrantUser(plan, FAKE_USER_ENTITY).isPresent()); - Assert.assertTrue(skipVisitor.visitRevokeUser(plan, FAKE_USER_ENTITY).isPresent()); + Assert.assertFalse(skipVisitor.visitGrantUser(plan, FAKE_USER_ENTITY).isEmpty()); + Assert.assertFalse(skipVisitor.visitRevokeUser(plan, FAKE_USER_ENTITY).isEmpty()); plan.setUserName("another"); - Assert.assertFalse(skipVisitor.visitGrantUser(plan, FAKE_USER_ENTITY).isPresent()); - Assert.assertFalse(skipVisitor.visitRevokeUser(plan, FAKE_USER_ENTITY).isPresent()); + Assert.assertTrue(skipVisitor.visitGrantUser(plan, FAKE_USER_ENTITY).isEmpty()); + Assert.assertTrue(skipVisitor.visitRevokeUser(plan, FAKE_USER_ENTITY).isEmpty()); permissionManager.setUserPrivilege( (userName, privilegeUnion) -> privilegeUnion.getPrivilegeType() == PrivilegeType.MANAGE_ROLE); - Assert.assertFalse( + Assert.assertTrue( skipVisitor .visitGrantUser(new AuthorTreePlan(ConfigPhysicalPlanType.GrantUser), FAKE_USER_ENTITY) - .isPresent()); - Assert.assertFalse( + .isEmpty()); + Assert.assertTrue( skipVisitor .visitRevokeUser( new AuthorTreePlan(ConfigPhysicalPlanType.RevokeUser), FAKE_USER_ENTITY) - .isPresent()); - Assert.assertTrue( + .isEmpty()); + Assert.assertFalse( skipVisitor .visitGrantRole(new AuthorTreePlan(ConfigPhysicalPlanType.GrantRole), FAKE_USER_ENTITY) - .isPresent()); - Assert.assertTrue( + .isEmpty()); + Assert.assertFalse( skipVisitor .visitRevokeRole( new AuthorTreePlan(ConfigPhysicalPlanType.RevokeRole), FAKE_USER_ENTITY) - .isPresent()); + .isEmpty()); permissionManager.setUserPrivilege((userName, privilegeUnion) -> false); plan = new AuthorTreePlan(ConfigPhysicalPlanType.GrantUser); plan.setRoleName(""); - Assert.assertTrue(skipVisitor.visitGrantRole(plan, FAKE_USER_ENTITY).isPresent()); - Assert.assertTrue(skipVisitor.visitRevokeRole(plan, FAKE_USER_ENTITY).isPresent()); + Assert.assertFalse(skipVisitor.visitGrantRole(plan, FAKE_USER_ENTITY).isEmpty()); + Assert.assertFalse(skipVisitor.visitRevokeRole(plan, FAKE_USER_ENTITY).isEmpty()); plan.setRoleName("another"); - Assert.assertFalse(skipVisitor.visitGrantRole(plan, FAKE_USER_ENTITY).isPresent()); - Assert.assertFalse(skipVisitor.visitRevokeRole(plan, FAKE_USER_ENTITY).isPresent()); + Assert.assertTrue(skipVisitor.visitGrantRole(plan, FAKE_USER_ENTITY).isEmpty()); + Assert.assertTrue(skipVisitor.visitRevokeRole(plan, FAKE_USER_ENTITY).isEmpty()); } @Test @@ -201,7 +206,7 @@ public void testPatternRelatedPrivilege() throws IOException { skipVisitor .visitPipeDeleteTimeSeries( new PipeDeleteTimeSeriesPlan(buffer), FAKE_USER_ENTITY) - .get()) + .get(0)) .getPatternTreeBytes()) .getAllPathPatterns()); Assert.assertEquals( @@ -211,7 +216,7 @@ public void testPatternRelatedPrivilege() throws IOException { skipVisitor .visitPipeDeleteLogicalView( new PipeDeleteLogicalViewPlan(buffer), FAKE_USER_ENTITY) - .get()) + .get(0)) .getPatternTreeBytes()) .getAllPathPatterns()); Assert.assertEquals( @@ -223,7 +228,7 @@ public void testPatternRelatedPrivilege() throws IOException { new PipeAlterEncodingCompressorPlan( buffer, (byte) 0, (byte) 0, false), FAKE_USER_ENTITY) - .get()) + .get(0)) .getPatternTreeBytes()) .getAllPathPatterns()); Assert.assertThrows( @@ -256,7 +261,7 @@ public void testPatternRelatedPrivilege() throws IOException { } }), FAKE_USER_ENTITY) - .get()) + .get(0)) .getTemplateSetInfo() .keySet()); @@ -267,14 +272,118 @@ public void testPatternRelatedPrivilege() throws IOException { .visitTTL( new SetTTLPlan(new String[] {"root", "*", "device", "measurement"}, 100), FAKE_USER_ENTITY) - .get()) + .get(0)) .getPathPattern()); - Assert.assertFalse( + Assert.assertTrue( skipVisitor .visitTTL( new SetTTLPlan(new String[] {"root", "db2", "device", "measurement"}, 100), FAKE_USER_ENTITY) - .isPresent()); + .isEmpty()); + } + + @Test + public void testTTLTrimToNarrowerAuthorizedPath() { + final SetTTLPlan trimmedPlan = + (SetTTLPlan) + skipVisitor + .visitTTL(new SetTTLPlan(new String[] {"root", "db", "**"}, 100), FAKE_USER_ENTITY) + .get(0); + + Assert.assertArrayEquals( + new String[] {"root", "db", "device", "**"}, trimmedPlan.getPathPattern()); + Assert.assertEquals(100, trimmedPlan.getTTL()); + } + + @Test + public void testTTLTransfersMultipleAuthorizedPaths() { + permissionManager.setAuthorizedPaths( + List.of( + new PartialPath(new String[] {"root", "db", "device1"}), + new PartialPath(new String[] {"root", "db", "device2"}))); + final SetTTLPlan plan = new SetTTLPlan(new String[] {"root", "db", "**"}, 100); + final List plans = skipVisitor.visitTTL(plan, FAKE_USER_ENTITY); + + Assert.assertEquals(2, plans.size()); + Assert.assertEquals( + new HashSet<>(Arrays.asList("root.db.device1", "root.db.device2")), + plans.stream() + .map(parsedPlan -> String.join(".", ((SetTTLPlan) parsedPlan).getPathPattern())) + .collect(Collectors.toSet())); + plans.forEach(parsedPlan -> Assert.assertEquals(100, ((SetTTLPlan) parsedPlan).getTTL())); + Assert.assertThrows( + AccessDeniedException.class, () -> throwVisitor.visitTTL(plan, FAKE_USER_ENTITY)); + } + + @Test + public void testTTLTrimsDatabaseCompatibilityPath() { + for (final long ttl : new long[] {100, -1}) { + final SetTTLPlan plan = new SetTTLPlan(new String[] {"root", "db"}, ttl); + plan.setDataBase(true); + final List plans = skipVisitor.visitTTL(plan, FAKE_USER_ENTITY); + + Assert.assertEquals(1, plans.size()); + final SetTTLPlan trimmed = (SetTTLPlan) plans.get(0); + Assert.assertArrayEquals( + new String[] {"root", "db", "device", "**"}, trimmed.getPathPattern()); + Assert.assertEquals(ttl, trimmed.getTTL()); + Assert.assertFalse(trimmed.isDataBase()); + Assert.assertThrows( + AccessDeniedException.class, () -> throwVisitor.visitTTL(plan, FAKE_USER_ENTITY)); + Assert.assertArrayEquals(new String[] {"root", "db"}, plan.getPathPattern()); + Assert.assertTrue(plan.isDataBase()); + } + } + + @Test + public void testTTLRetainsFullyAuthorizedDatabasePlan() { + permissionManager.setAuthorizedPaths( + Collections.singletonList(new PartialPath(new String[] {"root", "db", "**"}))); + final SetTTLPlan plan = new SetTTLPlan(new String[] {"root", "db"}, 100); + plan.setDataBase(true); + + for (final PipeConfigTreePrivilegeParseVisitor visitor : List.of(skipVisitor, throwVisitor)) { + Assert.assertEquals( + Collections.singletonList(plan), visitor.visitTTL(plan, FAKE_USER_ENTITY)); + } + } + + @Test + public void testTTLDeduplicatesOverlappingAuthorizations() { + permissionManager.setAuthorizedPaths( + List.of( + new PartialPath(new String[] {"root", "db", "device", "**"}), + new PartialPath(new String[] {"root", "db", "device", "s1"}))); + final List plans = + skipVisitor.visitTTL( + new SetTTLPlan(new String[] {"root", "db", "**"}, 100), FAKE_USER_ENTITY); + + Assert.assertEquals(1, plans.size()); + Assert.assertArrayEquals( + new String[] {"root", "db", "device", "**"}, ((SetTTLPlan) plans.get(0)).getPathPattern()); + } + + @Test + public void testTTLSkipsAncestorOfAuthorizedPath() { + final SetTTLPlan plan = new SetTTLPlan(new String[] {"root", "db"}, 100); + + Assert.assertTrue(skipVisitor.visitTTL(plan, FAKE_USER_ENTITY).isEmpty()); + Assert.assertThrows( + AccessDeniedException.class, () -> throwVisitor.visitTTL(plan, FAKE_USER_ENTITY)); + } + + @Test + public void testTTLWithFullPathAuthorization() { + final String[] path = new String[] {"root", "db", "device"}; + permissionManager.setAuthorizedPaths(Collections.singletonList(new PartialPath(path))); + final SetTTLPlan plan = new SetTTLPlan(path, 100); + + for (final PipeConfigTreePrivilegeParseVisitor visitor : List.of(skipVisitor, throwVisitor)) { + final SetTTLPlan authorizedPlan = + (SetTTLPlan) visitor.visitTTL(plan, FAKE_USER_ENTITY).get(0); + Assert.assertArrayEquals(path, authorizedPlan.getPathPattern()); + Assert.assertEquals(100, authorizedPlan.getTTL()); + } } @Test @@ -284,7 +393,7 @@ public void testSourcePrivilegeTrimDoesNotFallbackToTableDefaultForTreePlan() th new SetTTLPlan(new String[] {"root", "db2", "device", "measurement"}, 100), false); final TestIoTDBConfigRegionSource skipSource = new TestIoTDBConfigRegionSource(true); - Assert.assertFalse(skipSource.trimRealtimeEventByPrivilege(unauthorizedTreeEvent).isPresent()); + Assert.assertTrue(skipSource.trimRealtimeEventByPrivilege(unauthorizedTreeEvent).isEmpty()); final PipeConfigRegionWritePlanEvent trimmedTreeEvent = (PipeConfigRegionWritePlanEvent) @@ -293,7 +402,7 @@ public void testSourcePrivilegeTrimDoesNotFallbackToTableDefaultForTreePlan() th new PipeConfigRegionWritePlanEvent( new SetTTLPlan(new String[] {"root", "*", "device", "measurement"}, 100), false)) - .get(); + .get(0); Assert.assertArrayEquals( new String[] {"root", "db", "device", "measurement"}, ((SetTTLPlan) trimmedTreeEvent.getConfigPhysicalPlan()).getPathPattern()); @@ -304,6 +413,42 @@ public void testSourcePrivilegeTrimDoesNotFallbackToTableDefaultForTreePlan() th () -> throwSource.trimRealtimeEventByPrivilege(unauthorizedTreeEvent)); } + @Test + public void testSourceTransfersAllAuthorizedTTLPlans() throws Exception { + permissionManager.setAuthorizedPaths( + List.of( + new PartialPath(new String[] {"root", "db", "device1"}), + new PartialPath(new String[] {"root", "db", "device2"}))); + final TestIoTDBConfigRegionSource source = new TestIoTDBConfigRegionSource(true); + Assert.assertEquals( + 2, + source + .trimRealtimeEventByPrivilege( + new PipeConfigRegionWritePlanEvent( + new SetTTLPlan(new String[] {"root", "db", "**"}, 100), false)) + .size()); + } + + @Test + public void testSourceTrimsDatabaseTTLAfterPipePattern() throws Exception { + final SetTTLPlan plan = new SetTTLPlan(new String[] {"root", "db"}, 100); + plan.setDataBase(true); + final ConfigPhysicalPlan parsedPlan = + IoTDBConfigRegionSource.TREE_PATTERN_PARSE_VISITOR + .visitTTL(plan, new IoTDBTreePattern("root.**")) + .orElseThrow(AssertionError::new); + Assert.assertTrue(((SetTTLPlan) parsedPlan).isDataBase()); + final PipeConfigRegionWritePlanEvent trimmed = + (PipeConfigRegionWritePlanEvent) + new TestIoTDBConfigRegionSource(true) + .trimRealtimeEventByPrivilege(new PipeConfigRegionWritePlanEvent(parsedPlan, false)) + .get(0); + Assert.assertArrayEquals( + new String[] {"root", "db", "device", "**"}, + ((SetTTLPlan) trimmed.getConfigPhysicalPlan()).getPathPattern()); + Assert.assertFalse(((SetTTLPlan) trimmed.getConfigPhysicalPlan()).isDataBase()); + } + private static class TestIoTDBConfigRegionSource extends IoTDBConfigRegionSource { private TestIoTDBConfigRegionSource(final boolean skip) @@ -322,6 +467,8 @@ private static class TestPermissionManager extends PermissionManager { private BiFunction checkUserPrivileges = (userName, privilegeUnion) -> true; + private List authorizedPaths = + Collections.singletonList(new PartialPath(new String[] {"root", "db", "device", "**"})); public TestPermissionManager() { super(null, null); @@ -344,11 +491,15 @@ private void setUserPrivilege( public PathPatternTree fetchRawAuthorizedPTree( final String userName, final PrivilegeType type) { final PathPatternTree tree = new PathPatternTree(); - tree.appendPathPattern(new PartialPath(new String[] {"root", "db", "device", "**"})); + authorizedPaths.forEach(tree::appendPathPattern); tree.constructTree(); return tree; } + private void setAuthorizedPaths(final List authorizedPaths) { + this.authorizedPaths = authorizedPaths; + } + @Override public TPermissionInfoResp checkRoleOfUser(String username, String rolename) { return Objects.equals(username, rolename) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/schemaregion/IoTDBSchemaRegionSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/schemaregion/IoTDBSchemaRegionSource.java index 8e33be13bdbe1..f98590ad66b06 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/schemaregion/IoTDBSchemaRegionSource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/schemaregion/IoTDBSchemaRegionSource.java @@ -75,6 +75,7 @@ import java.util.Collections; import java.util.HashSet; import java.util.Iterator; +import java.util.List; import java.util.Objects; import java.util.Optional; import java.util.Set; @@ -291,18 +292,18 @@ protected PipeWritePlanEvent getNextEventInCurrentSnapshot() { } @Override - protected Optional trimRealtimeEventByPrivilege( - final PipeWritePlanEvent event) throws AccessDeniedException { + protected List trimRealtimeEventByPrivilege(final PipeWritePlanEvent event) + throws AccessDeniedException { final Optional result = treePrivilegeParseVisitor .process(((PipeSchemaRegionWritePlanEvent) event).getPlanNode(), userEntity) .flatMap(planNode -> tablePrivilegeParseVisitor.process(planNode, userEntity)); if (result.isPresent()) { - return Optional.of( + return Collections.singletonList( new PipeSchemaRegionWritePlanEvent(result.get(), event.isGeneratedByPipe())); } if (skipIfNoPrivileges) { - return Optional.empty(); + return Collections.emptyList(); } throw new AccessDeniedException( DataNodePipeMessages.NOT_HAS_PRIVILEGE_TO_TRANSFER_EVENT diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/source/IoTDBNonDataRegionSource.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/source/IoTDBNonDataRegionSource.java index d4d55ba7dac7a..8551696fd6d50 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/source/IoTDBNonDataRegionSource.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/source/IoTDBNonDataRegionSource.java @@ -46,6 +46,9 @@ import org.slf4j.LoggerFactory; import java.io.IOException; +import java.util.ArrayDeque; +import java.util.Collections; +import java.util.Deque; import java.util.LinkedList; import java.util.List; import java.util.Objects; @@ -75,6 +78,8 @@ public abstract class IoTDBNonDataRegionSource extends IoTDBSource { private final AtomicBoolean hasWarnedUnexpectedHybridProgressIndex = new AtomicBoolean(false); protected PipeWritePlanEvent lastEvent = null; + private final Deque pendingEvents = new ArrayDeque<>(); + private ProgressIndex pendingEventsProgressIndex; protected abstract AbstractPipeListeningQueue getListeningQueue(); @@ -172,9 +177,13 @@ public EnrichedEvent supply() throws Exception { } } + if (!pendingEvents.isEmpty()) { + return supplyPendingEvent(); + } + // Check whether snapshot being parsed exists PipeWritePlanEvent realtimeEvent = lastEvent; - if (hasNextEventInCurrentSnapshot()) { + if (Objects.isNull(realtimeEvent) && hasNextEventInCurrentSnapshot()) { realtimeEvent = getNextEventInCurrentSnapshot(); } @@ -228,14 +237,18 @@ public EnrichedEvent supply() throws Exception { } lastEvent = realtimeEvent; - realtimeEvent = + final List trimmedEvents = trimRealtimeEventByPipePattern(realtimeEvent) - .flatMap(this::trimRealtimeEventByPrivilege) - .orElse(null); + .map(this::trimRealtimeEventByPrivilege) + .orElseGet(Collections::emptyList); + for (final PipeWritePlanEvent trimmedEvent : trimmedEvents) { + if (isTypeListened(trimmedEvent) + && (isForwardingPipeRequests || !trimmedEvent.isGeneratedByPipe())) { + pendingEvents.addLast(trimmedEvent); + } + } - if (Objects.isNull(realtimeEvent) - || !isTypeListened(realtimeEvent) - || (!isForwardingPipeRequests && realtimeEvent.isGeneratedByPipe())) { + if (pendingEvents.isEmpty()) { final ProgressReportEvent event = new ProgressReportEvent(pipeName, creationTime, pipeTaskMeta); if (shouldBindIndex) { @@ -246,25 +259,35 @@ public EnrichedEvent supply() throws Exception { return event; } - realtimeEvent = + pendingEventsProgressIndex = + shouldBindIndex ? new MetaProgressIndex(iterator.getNextIndex() - 1) : null; + lastEvent = null; + return supplyPendingEvent(); + } + + private PipeWritePlanEvent supplyPendingEvent() { + final PipeWritePlanEvent realtimeEvent = (PipeWritePlanEvent) - realtimeEvent.shallowCopySelfAndBindPipeTaskMetaForProgressReport( - pipeName, - creationTime, - pipeTaskMeta, - treePattern, - tablePattern, - userId, - userName, - cliHostname, - skipIfNoPrivileges, - Long.MIN_VALUE, - Long.MAX_VALUE); - if (shouldBindIndex) { - realtimeEvent.bindProgressIndex(new MetaProgressIndex(iterator.getNextIndex() - 1)); + pendingEvents + .getFirst() + .shallowCopySelfAndBindPipeTaskMetaForProgressReport( + pipeName, + creationTime, + pipeTaskMeta, + treePattern, + tablePattern, + userId, + userName, + cliHostname, + skipIfNoPrivileges, + Long.MIN_VALUE, + Long.MAX_VALUE); + // The original event is complete only after all authorized intersections have been sent. + if (pendingEvents.size() == 1 && Objects.nonNull(pendingEventsProgressIndex)) { + realtimeEvent.bindProgressIndex(pendingEventsProgressIndex); } realtimeEvent.increaseReferenceCount(IoTDBNonDataRegionSource.class.getName()); - lastEvent = null; + pendingEvents.removeFirst(); return realtimeEvent; } @@ -279,7 +302,7 @@ protected abstract void initSnapshotGenerator(final PipeSnapshotEvent event) protected abstract PipeWritePlanEvent getNextEventInCurrentSnapshot(); - protected abstract Optional trimRealtimeEventByPrivilege( + protected abstract List trimRealtimeEventByPrivilege( final PipeWritePlanEvent event) throws AccessDeniedException; // The trimmed event shall be non-null. @@ -294,6 +317,9 @@ protected abstract Optional trimRealtimeEventByPipePattern( public void close() throws Exception { getListeningQueue().returnIterator(iterator); historicalEvents.clear(); + pendingEvents.clear(); + pendingEventsProgressIndex = null; + lastEvent = null; } //////////////////////////// APIs provided for metric framework //////////////////////////// diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/source/IoTDBNonDataRegionSourceTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/source/IoTDBNonDataRegionSourceTest.java index 3bfbd6dcefddb..3c988abd1f125 100644 --- a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/source/IoTDBNonDataRegionSourceTest.java +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/source/IoTDBNonDataRegionSourceTest.java @@ -27,9 +27,11 @@ import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.exception.auth.AccessDeniedException; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta; +import org.apache.iotdb.commons.pipe.datastructure.queue.ConcurrentIterableLinkedQueue; import org.apache.iotdb.commons.pipe.datastructure.queue.listening.AbstractPipeListeningQueue; import org.apache.iotdb.commons.pipe.event.PipeSnapshotEvent; import org.apache.iotdb.commons.pipe.event.PipeWritePlanEvent; +import org.apache.iotdb.pipe.api.event.Event; import org.junit.Assert; import org.junit.Test; @@ -38,8 +40,13 @@ import javax.annotation.Nonnull; import java.io.IOException; +import java.util.ArrayDeque; import java.util.Collections; +import java.util.Deque; +import java.util.List; import java.util.Optional; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Function; public class IoTDBNonDataRegionSourceTest { @@ -91,9 +98,110 @@ public void testGetUnTransferredEventCountWithHybridProgressIndexWithoutMetaInde Assert.assertEquals(7L, source.getUnTransferredEventCount()); } + @Test + public void testSupplySplitEventsBeforeAdvancingRealtimeProgress() throws Exception { + final PipeWritePlanEvent original = newWritePlanEvent(); + final PipeWritePlanEvent first = newWritePlanEvent(); + final PipeWritePlanEvent second = newWritePlanEvent(); + final PipeWritePlanEvent next = newWritePlanEvent(); + final ConcurrentIterableLinkedQueue queue = new ConcurrentIterableLinkedQueue<>(); + queue.add(original); + queue.add(next); + final TestNonDataRegionSource source = newSource(queue); + source.privilegeTrimmer = + event -> event == original ? List.of(first, second) : Collections.singletonList(event); + + Assert.assertSame(first, source.supply()); + Mockito.verify(first, Mockito.never()).bindProgressIndex(Mockito.any()); + Assert.assertSame(second, source.supply()); + Mockito.verify(second).bindProgressIndex(new MetaProgressIndex(0)); + Assert.assertSame(next, source.supply()); + Mockito.verify(next).bindProgressIndex(new MetaProgressIndex(1)); + source.close(); + } + + @Test + public void testSnapshotSplitsBindProgressOnlyToFinalPlan() throws Exception { + final PipeWritePlanEvent original = newWritePlanEvent(); + final PipeWritePlanEvent next = newWritePlanEvent(); + final PipeWritePlanEvent first = newWritePlanEvent(); + final PipeWritePlanEvent second = newWritePlanEvent(); + final PipeWritePlanEvent third = newWritePlanEvent(); + final PipeWritePlanEvent fourth = newWritePlanEvent(); + final TestNonDataRegionSource source = newSource(new ConcurrentIterableLinkedQueue<>()); + source.snapshotPlans.add(original); + source.snapshotPlans.add(next); + source.privilegeTrimmer = + event -> event == original ? List.of(first, second) : List.of(third, fourth); + + Assert.assertSame(first, source.supply()); + Assert.assertSame(second, source.supply()); + Assert.assertSame(third, source.supply()); + Assert.assertSame(fourth, source.supply()); + for (final PipeWritePlanEvent event : List.of(first, second, third)) { + Mockito.verify(event, Mockito.never()).bindProgressIndex(Mockito.any()); + } + Mockito.verify(fourth).bindProgressIndex(new MetaProgressIndex(-1)); + source.close(); + } + + @Test + public void testSnapshotPrivilegeRetryDoesNotConsumeNextPlan() throws Exception { + final PipeWritePlanEvent original = newWritePlanEvent(); + final PipeWritePlanEvent next = newWritePlanEvent(); + final TestNonDataRegionSource source = newSource(new ConcurrentIterableLinkedQueue<>()); + source.snapshotPlans.add(original); + source.snapshotPlans.add(next); + final AtomicBoolean firstAttempt = new AtomicBoolean(true); + source.privilegeTrimmer = + event -> { + if (firstAttempt.getAndSet(false)) { + throw new AccessDeniedException("test"); + } + return Collections.singletonList(event); + }; + + Assert.assertThrows(AccessDeniedException.class, source::supply); + Assert.assertSame(original, source.supply()); + Assert.assertSame(next, source.supply()); + source.close(); + } + + private static TestNonDataRegionSource newSource( + final ConcurrentIterableLinkedQueue queue) { + final AbstractPipeListeningQueue listeningQueue = + Mockito.mock(AbstractPipeListeningQueue.class); + Mockito.when(listeningQueue.isGivenNextIndexValid(0L)).thenReturn(true); + Mockito.when(listeningQueue.newIterator(0L)).thenReturn(queue.iterateFrom(0)); + return new TestNonDataRegionSource( + listeningQueue, new PipeTaskMeta(new MetaProgressIndex(-1), 0)); + } + + private static PipeWritePlanEvent newWritePlanEvent() { + final PipeWritePlanEvent event = Mockito.mock(PipeWritePlanEvent.class); + Mockito.when( + event.shallowCopySelfAndBindPipeTaskMetaForProgressReport( + Mockito.any(), + Mockito.anyLong(), + Mockito.any(), + Mockito.any(), + Mockito.any(), + Mockito.any(), + Mockito.any(), + Mockito.any(), + Mockito.anyBoolean(), + Mockito.anyLong(), + Mockito.anyLong())) + .thenReturn(event); + return event; + } + private static final class TestNonDataRegionSource extends IoTDBNonDataRegionSource { private final AbstractPipeListeningQueue listeningQueue; + private final Deque snapshotPlans = new ArrayDeque<>(); + private Function> privilegeTrimmer = + Collections::singletonList; private TestNonDataRegionSource( final AbstractPipeListeningQueue listeningQueue, final PipeTaskMeta pipeTaskMeta) { @@ -134,18 +242,18 @@ protected void initSnapshotGenerator(final PipeSnapshotEvent event) @Override protected boolean hasNextEventInCurrentSnapshot() { - return false; + return !snapshotPlans.isEmpty(); } @Override protected PipeWritePlanEvent getNextEventInCurrentSnapshot() { - return null; + return snapshotPlans.removeFirst(); } @Override - protected Optional trimRealtimeEventByPrivilege( - final PipeWritePlanEvent event) throws AccessDeniedException { - return Optional.of(event); + protected List trimRealtimeEventByPrivilege(final PipeWritePlanEvent event) + throws AccessDeniedException { + return privilegeTrimmer.apply(event); } @Override