diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBAttributeUpdateOrderingIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBAttributeUpdateOrderingIT.java new file mode 100644 index 000000000000..df6bc662199c --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBAttributeUpdateOrderingIT.java @@ -0,0 +1,82 @@ +/* + * 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.relational.it.schema; + +import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.TableClusterIT; +import org.apache.iotdb.relational.it.schema.fault.MetadataFaultController.Node; + +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.util.Map; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +@RunWith(IoTDBTestRunner.class) +@Category(TableClusterIT.class) +public class IoTDBAttributeUpdateOrderingIT extends TableMetadataPropagationTestBase { + /** a delayed batch belongs to the dropped column's earlier lifetime. */ + @Test + public void testOldAttributeBatchAfterDropAndReadd() throws Exception { + create("metadata_attribute_lifetime", 2); + DataNodeWrapper remote = remote(); + Node target = NODES.get(remote); + warm(remote, 2); + Map lease = target.lease(); + target.arm(UPDATE, "HOLD_FIRST", database, "changing=late-old;"); + update("UPDATE sensor SET changing='late-old' WHERE device='d0'"); + try { + target.awaitHeld(UPDATE); + Map delayed = target.status(UPDATE); + assertEquals("detail", delayed.get("batchType")); + assertTrue(delayed.toString(), delayed.get("version").matches("\\d+=\\d+")); + assertTrue(delayed.toString(), delayed.get("attributes").contains("changing=late-old")); + update("ALTER TABLE sensor DROP COLUMN changing"); + update("ALTER TABLE sensor ADD COLUMN changing STRING ATTRIBUTE"); + assertTwoDevices(null, null, true); + update("UPDATE sensor SET changing='new-lifetime' WHERE device='d0'"); + awaitTwoDevices("new-lifetime", null, true); + assertEquals("0", target.status(UPDATE).get("finished")); + assertLease(target, lease); + target.release(UPDATE); + awaitFinished(target, UPDATE); + try { + assertTwoDevices("new-lifetime", null, true); + } catch (AssertionError e) { + throw new AssertionError( + "Completed old batch " + + delayed.get("version") + + " (" + + delayed.get("attributes") + + ") changed confirmed current data at " + + remote.getIpAndPortString(), + e); + } + update("UPDATE sensor SET changing='newer-lifetime' WHERE device='d0'"); + awaitTwoDevices("newer-lifetime", null, true); + assertLease(target, lease); + } finally { + target.release(UPDATE); + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceCacheAttributeGuard.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceCacheAttributeGuard.java index 5710aeee7246..7176c1279008 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceCacheAttributeGuard.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceCacheAttributeGuard.java @@ -54,7 +54,8 @@ public boolean isRegionFetched(final Integer schemaRegionId) { } public void addFetchedRegion(final Integer schemaRegionId) { - fetchedSchemaRegionIds2LargestVersionAndDatabaseMap.put( + // Later schema reads must not let delayed updates bypass the accepted version. + fetchedSchemaRegionIds2LargestVersionAndDatabaseMap.putIfAbsent( schemaRegionId, new Pair<>(Long.MIN_VALUE, null)); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceCacheAttributeGuardTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceCacheAttributeGuardTest.java new file mode 100644 index 000000000000..30c6c04d913e --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceCacheAttributeGuardTest.java @@ -0,0 +1,64 @@ +/* + * 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.queryengine.plan.relational.metadata.fetcher; + +import org.apache.iotdb.db.schemaengine.schemaregion.attribute.update.UpdateDetailContainer; +import org.apache.iotdb.mpp.rpc.thrift.TAttributeUpdateReq; +import org.apache.iotdb.mpp.rpc.thrift.TSchemaRegionAttributeInfo; + +import org.junit.Test; + +import java.nio.ByteBuffer; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class TableDeviceCacheAttributeGuardTest { + @Test + public void testSchemaFetchPreservesAcceptedVersion() { + final TableDeviceCacheAttributeGuard guard = new TableDeviceCacheAttributeGuard(); + guard.addFetchedRegion(1); + final TAttributeUpdateReq accepted = update(2); + guard.handleAttributeUpdate(accepted); + assertFalse(accepted.getAttributeUpdateMap().isEmpty()); + + guard.addFetchedRegion(1); + for (final long version : new long[] {1, 2}) { + final TAttributeUpdateReq stale = update(version); + guard.handleAttributeUpdate(stale); + assertTrue(stale.getAttributeUpdateMap().isEmpty()); + } + final TAttributeUpdateReq newer = update(3); + guard.handleAttributeUpdate(newer); + assertFalse(newer.getAttributeUpdateMap().isEmpty()); + } + + private static TAttributeUpdateReq update(final long version) { + final byte[] body = + new UpdateDetailContainer().getUpdateContent(new AtomicInteger(1024), new AtomicBoolean()); + final Map updates = new HashMap<>(); + updates.put(1, new TSchemaRegionAttributeInfo(version, "db", ByteBuffer.wrap(body))); + return new TAttributeUpdateReq(updates); + } +}