From a6323b8685b444b2353ae8737a8e9881ff18edf6 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 5 Nov 2026 13:18:34 +0800 Subject: [PATCH 1/2] Preserve accepted table attribute versions across schema fetches --- integration-test/pom.xml | 6 + .../it/env/cluster/config/MppJVMConfig.java | 15 + .../env/cluster/node/AbstractNodeWrapper.java | 1 + .../IoTDBAttributeUpdateOrderingIT.java | 82 +++ .../TableMetadataPropagationTestBase.java | 431 +++++++++++++ .../it/schema/fault/MetadataFaultAgent.java | 600 ++++++++++++++++++ .../schema/fault/MetadataFaultController.java | 244 +++++++ .../TableDeviceCacheAttributeGuard.java | 3 +- .../TableDeviceCacheAttributeGuardTest.java | 64 ++ 9 files changed, 1445 insertions(+), 1 deletion(-) create mode 100644 integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBAttributeUpdateOrderingIT.java create mode 100644 integration-test/src/test/java/org/apache/iotdb/relational/it/schema/TableMetadataPropagationTestBase.java create mode 100644 integration-test/src/test/java/org/apache/iotdb/relational/it/schema/fault/MetadataFaultAgent.java create mode 100644 integration-test/src/test/java/org/apache/iotdb/relational/it/schema/fault/MetadataFaultController.java create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceCacheAttributeGuardTest.java diff --git a/integration-test/pom.xml b/integration-test/pom.xml index 779c059e2d649..c98043a37505a 100644 --- a/integration-test/pom.xml +++ b/integration-test/pom.xml @@ -92,6 +92,12 @@ 3 + + org.ow2.asm + asm + + runtime + org.apache.iotdb iotdb-server diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppJVMConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppJVMConfig.java index d9c51717cf8e1..7304e3024cd14 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppJVMConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppJVMConfig.java @@ -23,12 +23,26 @@ import javax.annotation.Nullable; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + public class MppJVMConfig implements JVMConfig { private int initHeapSize; private int maxHeapSize; private int maxDirectMemorySize; private String timezone; + private List jvmOptions = Collections.emptyList(); + + /** Additional arguments for child test-node JVMs, for example a test-only Java agent. */ + public void setJvmOptions(List options) { + this.jvmOptions = new ArrayList<>(options); + } + + public List getJvmOptions() { + return Collections.unmodifiableList(jvmOptions); + } @Override public JVMConfig setInitHeapSize(int initSize) { @@ -94,6 +108,7 @@ public void override(@Nullable MppJVMConfig config) { this.setMaxHeapSize(config.getMaxHeapSize()); this.setMaxDirectMemorySize(config.getMaxDirectMemorySize()); this.setTimezone(config.getTimezone()); + this.setJvmOptions(config.getJvmOptions()); } public static Builder builder() { diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/AbstractNodeWrapper.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/AbstractNodeWrapper.java index 8d3f5f2cc9d9e..a5bc823eafda2 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/AbstractNodeWrapper.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/AbstractNodeWrapper.java @@ -483,6 +483,7 @@ public void start() { File stdoutFile = new File(getLogPath()); List startCmd = new ArrayList<>(); startCmd.add(JAVA_CMD); + startCmd.addAll(jvmConfig.getJvmOptions()); if (!SystemUtils.IS_JAVA_1_8) { startCmd.add("--add-opens=java.base/java.util.concurrent=ALL-UNNAMED"); startCmd.add("--add-opens=java.base/java.lang=ALL-UNNAMED"); 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 0000000000000..df6bc662199c5 --- /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/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/TableMetadataPropagationTestBase.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/TableMetadataPropagationTestBase.java new file mode 100644 index 0000000000000..04fc106807547 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/TableMetadataPropagationTestBase.java @@ -0,0 +1,431 @@ +/* + * 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.common.rpc.thrift.TConsensusGroupType; +import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant; +import org.apache.iotdb.confignode.i18n.ProcedureMessages; +import org.apache.iotdb.consensus.ConsensusFactory; +import org.apache.iotdb.db.it.utils.TestUtils; +import org.apache.iotdb.isession.SessionConfig; +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.env.cluster.config.MppCommonConfig; +import org.apache.iotdb.it.env.cluster.config.MppJVMConfig; +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.itbase.env.BaseEnv; +import org.apache.iotdb.relational.it.schema.fault.MetadataFaultController; +import org.apache.iotdb.relational.it.schema.fault.MetadataFaultController.Node; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.awaitility.Awaitility; +import org.junit.After; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +/** + * Real remote attribute pulls, selective invalidation failures and delayed propagation batches. The + * two DataNodes share one Memory cluster; the non-replica is always the cache under test. + */ +@RunWith(IoTDBTestRunner.class) +@Category(TableClusterIT.class) +public abstract class TableMetadataPropagationTestBase { + protected static final String TABLE = "sensor"; + protected static final String FILL = "putAttributes:enter"; + protected static final String UPDATE = "updateAttribute:enter"; + protected static final String INVALIDATE = "invalidateColumnCache:enter"; + protected static final String ACK = "invalidateColumnCache:exit"; + protected static final String DELETE = "deleteColumnData:enter"; + protected static final String ACCEPT = "handleAttributeUpdate:exit"; + protected static final String CONTAINER = "handleContainer:enter"; + protected static final String OLD_COMMIT = "sendUpdateRequestAndMayShrink:exit"; + protected static final Map NODES = new HashMap<>(); + protected static MetadataFaultController faults; + protected String database; + + @BeforeClass + public static void setUp() throws Exception { + EnvFactory.getEnv() + .getConfig() + .getCommonConfig() + .setSchemaEngineMode("Memory") + .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS) + .setSchemaReplicationFactor(1) + .setDataReplicationFactor(1) + .setSchemaRegionGroupExtensionPolicy("CUSTOM") + .setDataRegionGroupExtensionPolicy("CUSTOM") + .setDefaultSchemaRegionGroupNumPerDatabase(1) + .setDefaultDataRegionGroupNumPerDatabase(1) + .setMetadataLeaseFenceMs(8_000) + // At a fixed 512 MiB heap this gives SchemaRegion about 256 KiB. Cache memory remains + // sufficient for exact-device reads, so cache eviction cannot hide the propagation race. + .setSchemaMemoryAllocate("1:199:1"); + EnvFactory.getEnv().getConfig().getDataNodeJVMConfig().setMaxHeapSize(512); + Properties properties = new Properties(); + properties.setProperty("general_region_attribute_security_service_interval_seconds", "1"); + properties.setProperty("general_region_attribute_security_service_timeout_seconds", "2"); + properties.setProperty("detail_container_min_degrade_memory_in_bytes", "8192"); + ((MppCommonConfig) EnvFactory.getEnv().getConfig().getDataNodeCommonConfig()) + .updateProperties(properties); + faults = + new MetadataFaultController( + (MppJVMConfig) EnvFactory.getEnv().getConfig().getDataNodeJVMConfig()); + try { + EnvFactory.getEnv().initClusterEnvironment(1, 2); + for (DataNodeWrapper wrapper : EnvFactory.getEnv().getDataNodeWrapperList()) { + NODES.put(wrapper, faults.connect(wrapper)); + } + } finally { + faults.restoreLaunchEnvironment(); + } + } + + @AfterClass + public static void tearDown() throws Exception { + try { + EnvFactory.getEnv().cleanClusterEnvironment(); + } finally { + if (faults != null) { + faults.close(); + } + NODES.clear(); + } + } + + @After + public void cleanUp() throws Exception { + faults.reset(); + if (database != null) { + try (Connection connection = EnvFactory.getEnv().getTableConnection(); + Statement statement = connection.createStatement()) { + statement.execute("DROP DATABASE " + database); + } + } + } + + protected void create(String name, int devices) throws Exception { + database = name; + try (Connection connection = connection(EnvFactory.getEnv().getDataNodeWrapperList().get(0)); + Statement statement = connection.createStatement()) { + statement.execute("CREATE DATABASE " + database); + statement.execute("USE " + database); + statement.execute( + "CREATE TABLE sensor (device STRING TAG, changing STRING ATTRIBUTE, kept STRING ATTRIBUTE, reading INT32 FIELD)"); + statement.execute( + "INSERT INTO sensor(time,device,changing,kept,reading) VALUES(1,'d0','old-0','keep-0',10)"); + } + DataNodeWrapper remote = remote(); + DataNodeWrapper owner = + EnvFactory.getEnv().getDataNodeWrapperList().stream() + .filter(wrapper -> wrapper != remote) + .findFirst() + .get(); + // The remaining devices have never been fetched by the remote coordinator before its gate. + try (Connection connection = connection(owner); + Statement statement = connection.createStatement()) { + statement.execute("USE " + database); + for (int i = 1; i < devices; i++) { + statement.execute( + "INSERT INTO sensor(time,device,changing,kept,reading) VALUES(" + + (i + 1) + + ",'d" + + i + + "','old-" + + i + + "','keep-" + + i + + "'," + + (i + 10) + + ")"); + } + statement.execute("FLUSH"); + } + } + + protected DataNodeWrapper remote() throws SQLException { + DataNodeWrapper replica = null; + int regions = 0; + try (Connection connection = EnvFactory.getEnv().getTableConnection(); + Statement statement = connection.createStatement(); + ResultSet result = statement.executeQuery("SHOW REGIONS")) { + while (result.next()) { + if (TConsensusGroupType.SchemaRegion.toString() + .equals(result.getString(ColumnHeaderConstant.TYPE)) + && database.equals(result.getString(ColumnHeaderConstant.DATABASE))) { + regions++; + replica = + EnvFactory.getEnv() + .dataNodeIdToWrapper(result.getInt(ColumnHeaderConstant.DATA_NODE_ID)) + .get(); + } + } + } + assertEquals("One sole SchemaRegion replica is required", 1, regions); + assertNotNull(replica); + for (DataNodeWrapper wrapper : EnvFactory.getEnv().getDataNodeWrapperList()) { + if (wrapper != replica) { + return wrapper; + } + } + throw new AssertionError("No remote cache node"); + } + + protected static Connection connection(DataNodeWrapper wrapper) throws SQLException { + return EnvFactory.getEnv() + .getConnection( + wrapper, + SessionConfig.DEFAULT_USER, + SessionConfig.DEFAULT_PASSWORD, + BaseEnv.TABLE_SQL_DIALECT); + } + + protected void warm(DataNodeWrapper wrapper, int devices) throws Exception { + Node node = NODES.get(wrapper); + node.arm(FILL, "OBSERVE", database); + try (Connection connection = connection(wrapper); + Statement statement = connection.createStatement()) { + for (int i = 0; i < devices; i++) { + try (ResultSet result = + statement.executeQuery( + "SELECT changing, kept, reading FROM " + + database + + ".sensor WHERE device='d" + + i + + "'")) { + TestUtils.assertResultSetEqualUnordered( + result, + "changing,kept,reading,", + java.util.Collections.singletonList( + "old-" + i + ",keep-" + i + "," + (i + 10) + ",")); + } + } + } + long fills = Long.parseLong(node.status(FILL).get("matches")); + assertTrue("The remote queries must actually fetch their attributes", fills >= devices - 1); + // The same complete keys must now hit the cache whose staleness is under test. + try (Connection connection = connection(wrapper); + Statement statement = connection.createStatement()) { + for (int i = 0; i < devices; i++) { + try (ResultSet result = + statement.executeQuery( + "SELECT changing FROM " + database + ".sensor WHERE device='d" + i + "'")) { + assertTrue(result.next()); + assertEquals("old-" + i, result.getString(1)); + assertFalse(result.next()); + } + } + } + assertEquals(Long.toString(fills), node.status(FILL).get("matches")); + } + + protected void update(String sql) throws Exception { + try (Connection connection = connection(EnvFactory.getEnv().getDataNodeWrapperList().get(0)); + Statement statement = connection.createStatement()) { + statement.execute("USE " + database); + statement.execute(sql); + } + } + + protected void assertTwoDevices(String first, String second, boolean readded) { + List attributes = new ArrayList<>(); + attributes.add(first); + attributes.add(second); + assertDevices(attributes, readded); + } + + protected void awaitTwoDevices(String first, String second, boolean readded) { + Awaitility.await() + .atMost(20, TimeUnit.SECONDS) + .untilAsserted(() -> assertTwoDevices(first, second, readded)); + } + + protected void assertDevices(List attributes, boolean readded) { + List data = new ArrayList<>(); + List devices = new ArrayList<>(); + int nulls = 0; + for (int i = 0; i < attributes.size(); i++) { + String attribute = attributes.get(i); + if (attribute == null) { + nulls++; + } + data.add((i + 1) + ",d" + i + "," + attribute + ",keep-" + i + "," + (i + 10) + ","); + devices.add( + "d" + + i + + "," + + (readded ? "keep-" + i + "," + attribute : attribute + ",keep-" + i) + + ","); + TestUtils.tableResultSetEqualTestUnordered( + "SELECT device, changing, kept, reading FROM sensor WHERE device='d" + i + "'", + new String[] {"device", "changing", "kept", "reading"}, + new String[] {"d" + i + "," + attribute + ",keep-" + i + "," + (i + 10) + ","}, + database); + } + TestUtils.tableResultSetEqualTestUnordered( + "SELECT CAST(time AS INT64) AS time, device, changing, kept, reading FROM sensor", + new String[] {"time", "device", "changing", "kept", "reading"}, + data.toArray(new String[0]), + database); + TestUtils.tableResultSetEqualTestUnordered( + "SHOW DEVICES FROM sensor", + readded + ? new String[] {"device", "kept", "changing"} + : new String[] {"device", "changing", "kept"}, + devices.toArray(new String[0]), + database); + TestUtils.tableResultSetEqualTestUnordered( + "COUNT DEVICES FROM sensor", + new String[] {"count(devices)"}, + new String[] {attributes.size() + ","}, + database); + TestUtils.tableResultSetEqualTestUnordered( + "COUNT DEVICES FROM sensor WHERE changing IS NULL", + new String[] {"count(devices)"}, + new String[] {nulls + ","}, + database); + if (attributes.get(0) != null) { + assertAttributeFilter(0, attributes.get(0)); + } + TestUtils.tableResultSetEqualTestUnordered( + "COUNT DEVICES FROM sensor WHERE changing='late-old'", + new String[] {"count(devices)"}, + new String[] {"0,"}, + database); + TestUtils.tableResultSetEqualTest( + "DESCRIBE sensor", + new String[] {"ColumnName", "DataType", "Category"}, + readded + ? new String[] { + "time,TIMESTAMP,TIME,", + "device,STRING,TAG,", + "kept,STRING,ATTRIBUTE,", + "reading,INT32,FIELD,", + "changing,STRING,ATTRIBUTE," + } + : new String[] { + "time,TIMESTAMP,TIME,", + "device,STRING,TAG,", + "changing,STRING,ATTRIBUTE,", + "kept,STRING,ATTRIBUTE,", + "reading,INT32,FIELD," + }, + database); + } + + protected void assertInvalidationFailure(SQLException failure) { + assertNotNull("DDL crossed the live node without an ACK", failure); + assertEquals( + failure.toString(), TSStatusCode.METADATA_ERROR.getStatusCode(), failure.getErrorCode()); + assertTrue( + failure.toString(), + failure + .getMessage() + .contains( + String.format( + ProcedureMessages.INVALIDATE_COLUMN_CACHE_FAILED_FOR_TABLE, + "changing", + database, + TABLE))); + } + + protected void assertAttributeFilter(int device, String attribute) { + String predicate = + "device='d" + + device + + "' AND " + + (attribute == null ? "changing IS NULL" : "changing='" + attribute + "'"); + TestUtils.tableResultSetEqualTestUnordered( + "COUNT DEVICES FROM sensor WHERE " + predicate, + new String[] {"count(devices)"}, + new String[] {"1,"}, + database); + // Check nullable data predicates after the complete fault/recovery sequence. A query failure + // must still fail the test, without preventing the delayed batch or later DDL from being + // tested. + TestUtils.tableResultSetEqualTestUnordered( + "SELECT device, kept, reading FROM sensor WHERE " + predicate, + new String[] {"device", "kept", "reading"}, + new String[] {"d" + device + ",keep-" + device + "," + (device + 10) + ","}, + database); + } + + protected static void assertLease(Node node, Map initial) throws Exception { + Map current = node.lease(); + assertEquals(current.toString(), "false", current.get("fenced")); + assertEquals( + "Lease recovery must not clear the cache", initial.get("clears"), current.get("clears")); + assertTrue( + current.toString(), + Long.parseLong(current.get("elapsed")) < Long.parseLong(current.get("threshold"))); + assertTrue( + current.toString(), + Long.parseLong(current.get("heartbeats")) >= Long.parseLong(initial.get("heartbeats"))); + } + + protected static void awaitMatches(Node node, String hook) { + Awaitility.await() + .atMost(20, TimeUnit.SECONDS) + .until(() -> Boolean.parseBoolean(node.status(hook).get("captured"))); + } + + protected static void awaitFinished(Node node, String hook) { + Awaitility.await() + .atMost(10, TimeUnit.SECONDS) + .until(() -> Long.parseLong(node.status(hook).get("completed")) > 0); + } + + protected void awaitEmpty(Node sender) { + Awaitility.await() + .atMost(20, TimeUnit.SECONDS) + .untilAsserted( + () -> + assertEquals( + sender.updaters(database).toString(), + "0", + sender.updaters(database).get("pending"))); + } + + protected static void finish(ExecutorService executor) throws InterruptedException { + executor.shutdown(); + if (!executor.awaitTermination(35, TimeUnit.SECONDS)) { + executor.shutdownNow(); + assertTrue("Test executor did not stop", executor.awaitTermination(5, TimeUnit.SECONDS)); + } + } +} diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/fault/MetadataFaultAgent.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/fault/MetadataFaultAgent.java new file mode 100644 index 0000000000000..fdcde9766bc9a --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/fault/MetadataFaultAgent.java @@ -0,0 +1,600 @@ +/* + * 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.fault; + +import org.objectweb.asm.ClassReader; +import org.objectweb.asm.ClassVisitor; +import org.objectweb.asm.ClassWriter; +import org.objectweb.asm.MethodVisitor; +import org.objectweb.asm.Opcodes; +import org.objectweb.asm.Type; + +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.lang.instrument.ClassFileTransformer; +import java.lang.instrument.Instrumentation; +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.security.ProtectionDomain; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +/** + * Launch-time instrumentation for metadata IT child processes only. No production classes are + * replaced on disk. The gates run before the receiver takes its locks, or after its finally block. + * The control channel only arms gates and reads state; it never edits product caches or versions. + */ +public final class MetadataFaultAgent { + public static final String INVALIDATE = "invalidateColumnCache"; + public static final String UPDATE = "updateAttribute"; + public static final String FILL = "putAttributes"; + private static final String RPC = + "org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl"; + private static final String FETCHER = + "org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/"; + private static final String UPDATER = + "org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/"; + private static final String LEASE = "org/apache/iotdb/db/schemaengine/lease/MetadataLeaseManager"; + private static final Map RULES = new ConcurrentHashMap<>(); + private static final Map COUNTS = new ConcurrentHashMap<>(); + private static final Set UPDATERS = ConcurrentHashMap.newKeySet(); + + private MetadataFaultAgent() {} + + public static void premain(String directory, Instrumentation instrumentation) throws IOException { + instrumentation.addTransformer(new Transformer()); + ServerSocket server = new ServerSocket(0, 16, InetAddress.getLoopbackAddress()); + Thread controller = new Thread(() -> serve(server), "metadata-it-fault-controller"); + controller.setDaemon(true); + controller.start(); + Path endpoint = Paths.get(directory, ProcessHandle.current().pid() + ".port"); + Files.write(endpoint, Collections.singletonList(Integer.toString(server.getLocalPort()))); + } + + private static void serve(ServerSocket server) { + while (!server.isClosed()) { + try (Socket socket = server.accept(); + DataInputStream input = new DataInputStream(socket.getInputStream()); + DataOutputStream output = new DataOutputStream(socket.getOutputStream())) { + socket.setSoTimeout(5_000); + Map response; + try { + String command = input.readUTF(); + String hook = input.readUTF(); + String mode = input.readUTF(); + String match = input.readUTF(); + String attributes = input.readUTF(); + response = command(command, hook, mode, match, attributes); + } catch (Throwable e) { + response = Collections.singletonMap("error", e.toString()); + } + output.writeInt(response.size()); + for (Map.Entry entry : response.entrySet()) { + output.writeUTF(entry.getKey()); + output.writeUTF(entry.getValue()); + } + } catch (IOException e) { + // A failed control connection must not disable subsequent release/cleanup requests. + System.err.println("Metadata IT control connection failed: " + e); + } + } + } + + private static Map command( + String command, String hook, String mode, String match, String attributes) throws Exception { + switch (command) { + case "arm": + Rule old = RULES.putIfAbsent(hook, new Rule(mode, match, attributes)); + if (old != null) { + throw new IllegalStateException("Gate already armed: " + hook); + } + return Collections.emptyMap(); + case "status": + Rule rule = RULES.get(hook); + return rule == null ? Collections.emptyMap() : rule.snapshot(); + case "release": + Rule toRelease = RULES.get(hook); + if (toRelease != null) { + toRelease.release.countDown(); + } + return Collections.emptyMap(); + case "reset": + RULES.values().forEach(value -> value.release.countDown()); + RULES.clear(); + return Collections.emptyMap(); + case "lease": + Object lease = singleton(LEASE.replace('/', '.')); + Map state = new HashMap<>(); + state.put("heartbeats", Long.toString(count("triggerCheckWithHeartBeat:exit"))); + state.put("clears", Long.toString(count("clearCacheAndPullMetaData:enter"))); + state.put("fenced", invoke(lease, "isFenced").toString()); + state.put("elapsed", invoke(lease, "getMillisSinceLastConfigNodeHeartbeat").toString()); + state.put("threshold", field(lease, "fenceThresholdMs").toString()); + return state; + case "updaters": + return updaterState(match); + case "count": + return Collections.singletonMap("count", Long.toString(count(hook))); + default: + throw new IllegalArgumentException(command); + } + } + + private static long count(String hook) { + AtomicLong count = COUNTS.get(hook); + return count == null ? 0 : count.get(); + } + + private static Map updaterState(String database) throws Exception { + Map state = new HashMap<>(); + int pending = 0; + int detail = 0; + int clear = 0; + int found = 0; + for (Object updater : UPDATERS) { + if (!field(updater, "databaseName").toString().equals(database)) { + continue; + } + found++; + Map containers = (Map) field(updater, "attributeUpdateMap"); + pending += containers.size(); + for (Object container : containers.values()) { + if (container.getClass().getSimpleName().equals("UpdateDetailContainer")) { + detail++; + } else { + clear++; + } + } + Object statistics = field(field(updater, "regionStatistics"), "schemaEngineStatistics"); + state.put("usage", invoke(statistics, "getMemoryUsage").toString()); + state.put("capacity", invoke(statistics, "getMemoryCapacity").toString()); + state.put("allowCreate", invoke(statistics, "isAllowToCreateNewSeries").toString()); + } + state.put("found", Integer.toString(found)); + state.put("pending", Integer.toString(pending)); + state.put("detail", Integer.toString(detail)); + state.put("clear", Integer.toString(clear)); + state.put("degrades", Long.toString(count("degrade:exit"))); + return state; + } + + private static Object singleton(String name) throws Exception { + return Class.forName(name).getMethod("getInstance").invoke(null); + } + + private static Object invoke(Object object, String method) throws Exception { + return object.getClass().getMethod(method).invoke(object); + } + + private static Object field(Object object, String name) throws Exception { + Class type = object.getClass(); + while (type != null) { + try { + Field field = type.getDeclaredField(name); + field.setAccessible(true); + return field.get(object); + } catch (NoSuchFieldException e) { + type = type.getSuperclass(); + } + } + throw new NoSuchFieldException(name); + } + + /** Called only by the instrumented child process, never by the test's control connection. */ + public static void hit(String hook, Object receiver, Object[] arguments, Object result) + throws Throwable { + COUNTS.computeIfAbsent(hook, key -> new AtomicLong()).incrementAndGet(); + if (receiver.getClass().getSimpleName().equals("DeviceAttributeCacheUpdater")) { + UPDATERS.add(receiver); + } + if (hook.endsWith(":exit") && arguments.length > 0) { + Rule entryRule = RULES.get(hook.substring(0, hook.length() - 5) + ":enter"); + if (entryRule != null && arguments[0] == entryRule.firstArgument) { + entryRule.completed.incrementAndGet(); + } + } + Rule rule = RULES.get(hook); + if (rule == null) { + return; + } + String description = Arrays.toString(arguments); + if (receiver.getClass().getSimpleName().equals("TableDeviceSchemaCache") + && arguments.length > 2 + && arguments[2] instanceof Map) { + StringBuilder values = new StringBuilder(); + ((Map) arguments[2]) + .forEach( + (key, value) -> + values.append(key).append('=').append(binaryString(value)).append(';')); + description += " fetchedAttributes=" + values; + } + if (receiver.getClass().getSimpleName().equals("GeneralRegionAttributeSecurityService")) { + Map databases = (Map) field(receiver, "regionId2DatabaseMap"); + for (Object region : ((Map) arguments[0]).keySet()) { + description += " database=" + databases.get(region); + } + } + if (!description.contains(rule.match)) { + return; + } + String attributes = batchAttributes(arguments); + if (!attributes.contains(rule.expectedAttributes)) { + return; + } + final long occurrence; + // Publish the first batch's identity before a concurrent retry can decide whether to pass. + // Do not retain this monitor while waiting at the gate. + synchronized (rule) { + occurrence = rule.matches.incrementAndGet(); + if (occurrence == 1) { + rule.description = description.substring(0, Math.min(description.length(), 4_096)); + rule.stack = Arrays.toString(Thread.currentThread().getStackTrace()); + rule.version = version(arguments); + rule.batchType = batchType(arguments); + rule.attributes = attributes; + rule.firstArgument = arguments.length == 0 ? null : arguments[0]; + rule.captured = true; + } + } + for (Object argument : arguments) { + if (argument != null && argument.getClass().getSimpleName().equals("UpdateClearContainer")) { + rule.clear.incrementAndGet(); + } + } + if (rule.release.getCount() == 0) { + rule.passed.incrementAndGet(); + return; + } + if (rule.mode.equals("HOLD_FIRST") && occurrence == 1) { + rule.held.countDown(); + if (!rule.release.await(120, TimeUnit.SECONDS)) { + throw new AssertionError("Metadata IT gate was not released: " + hook); + } + rule.finished.incrementAndGet(); + return; + } + if (rule.mode.equals("DROP") + || (rule.mode.equals("HOLD_FIRST") + && rule.version != null + && rule.version.equals(version(arguments)))) { + rule.dropped.incrementAndGet(); + // Thrift's processor rethrows TTransportException and closes this connection. In particular, + // this does not manufacture an application error status that would bypass transport retries. + throw (Throwable) + Class.forName("org.apache.thrift.transport.TTransportException") + .getConstructor(String.class) + .newInstance("Metadata IT discarded " + hook); + } + rule.passed.incrementAndGet(); + } + + private static String version(Object[] arguments) throws Exception { + if (arguments.length == 0) { + return null; + } + if (arguments[0] instanceof Map) { + StringBuilder versions = new StringBuilder(); + for (Map.Entry entry : ((Map) arguments[0]).entrySet()) { + versions.append(entry.getKey()).append('=').append(invoke(entry.getValue(), "getLeft")); + } + return versions.toString(); + } + if (!arguments[0].getClass().getSimpleName().equals("TAttributeUpdateReq")) { + return null; + } + Map updates = (Map) invoke(arguments[0], "getAttributeUpdateMap"); + StringBuilder versions = new StringBuilder(); + for (Map.Entry entry : updates.entrySet()) { + versions.append(entry.getKey()).append('=').append(invoke(entry.getValue(), "getVersion")); + } + return versions.toString(); + } + + private static String binaryString(Object value) { + if (value == null) { + return "null"; + } + try { + for (Method method : value.getClass().getMethods()) { + if (method.getName().equals("getStringValue") + && method.getParameterCount() == 1 + && method.getParameterTypes()[0].getName().equals("java.nio.charset.Charset")) { + return method.invoke(value, java.nio.charset.StandardCharsets.UTF_8).toString(); + } + } + } catch (ReflectiveOperationException ignored) { + // Fall back to the value's diagnostic form. + } + return String.valueOf(value); + } + + private static String batchType(Object[] arguments) throws Exception { + for (byte[] body : batchBodies(arguments)) { + return body[0] == 0 ? "clear" : "detail"; + } + return "empty"; + } + + private static String batchAttributes(Object[] arguments) throws Exception { + List bodies = batchBodies(arguments); + if (bodies.isEmpty()) { + return ""; + } + Method decode = + Class.forName((UPDATER + "DeviceAttributeCacheUpdater").replace('/', '.')) + .getMethod("getContainer", byte[].class); + StringBuilder values = new StringBuilder(); + for (byte[] body : bodies) { + // Deserialize a separate diagnostic copy; never alter the request or product cache. + Object container = decode.invoke(null, body); + if (!container.getClass().getSimpleName().equals("UpdateDetailContainer")) { + continue; + } + for (Map.Entry table : ((Map) invoke(container, "getUpdateMap")).entrySet()) { + for (Map.Entry device : ((Map) table.getValue()).entrySet()) { + for (Map.Entry attribute : ((Map) device.getValue()).entrySet()) { + values + .append(table.getKey()) + .append('/') + .append(device.getKey()) + .append(':') + .append(attribute.getKey()) + .append('=') + .append(binaryString(attribute.getValue())) + .append(';'); + } + } + } + } + return values.substring(0, Math.min(values.length(), 4_096)); + } + + private static List batchBodies(Object[] arguments) throws Exception { + List bodies = new ArrayList<>(); + if (arguments.length == 0) { + return bodies; + } + if (arguments[0].getClass().getSimpleName().equals("TAttributeUpdateReq")) { + for (Object update : ((Map) invoke(arguments[0], "getAttributeUpdateMap")).values()) { + bodies.add((byte[]) invoke(update, "getBody")); + } + } else if (arguments[0] instanceof Map) { + // At sendUpdateRequestAndMayShrink's exit this map contains only acknowledged batches. + for (Object pair : ((Map) arguments[0]).values()) { + for (Object body : ((Map) invoke(pair, "getRight")).values()) { + bodies.add((byte[]) body); + } + } + } + return bodies; + } + + private static final class Rule { + private final String mode; + private final String match; + private final String expectedAttributes; + private final AtomicLong matches = new AtomicLong(); + private final AtomicLong dropped = new AtomicLong(); + private final AtomicLong passed = new AtomicLong(); + private final AtomicLong finished = new AtomicLong(); + private final AtomicLong clear = new AtomicLong(); + private final AtomicLong completed = new AtomicLong(); + private final CountDownLatch held = new CountDownLatch(1); + private final CountDownLatch release = new CountDownLatch(1); + private volatile String description = ""; + private volatile String stack = ""; + private volatile String version; + private volatile String batchType = ""; + private volatile String attributes = ""; + private volatile Object firstArgument; + private volatile boolean captured; + + private Rule(String mode, String match, String expectedAttributes) { + this.mode = mode; + this.match = match; + this.expectedAttributes = expectedAttributes; + } + + private Map snapshot() { + Map state = new HashMap<>(); + state.put("matches", Long.toString(matches.get())); + state.put("dropped", Long.toString(dropped.get())); + state.put("passed", Long.toString(passed.get())); + state.put("finished", Long.toString(finished.get())); + state.put("held", Boolean.toString(held.getCount() == 0)); + state.put("description", description); + state.put("stack", stack); + state.put("version", String.valueOf(version)); + state.put("batchType", batchType); + state.put("attributes", attributes); + state.put("clear", Long.toString(clear.get())); + state.put("completed", Long.toString(completed.get())); + state.put("captured", Boolean.toString(captured)); + return state; + } + } + + private static boolean selected(String owner, String method, boolean exit) { + if (owner.equals(RPC)) { + return method.equals(INVALIDATE) + || method.equals(UPDATE) + || method.equals("deleteColumnData"); + } + if (owner.equals(FETCHER + "cache/TableDeviceSchemaCache")) { + return method.equals(FILL) && !exit; + } + if (owner.equals(FETCHER + "TableDeviceCacheAttributeGuard")) { + return (method.equals("handleAttributeUpdate") && exit) + || (method.equals("handleContainer") && !exit); + } + if (owner.equals(LEASE)) { + return (method.equals("triggerCheckWithHeartBeat") && exit) + || (method.equals("clearCacheAndPullMetaData") && !exit); + } + if (owner.equals(UPDATER + "DeviceAttributeCacheUpdater")) { + return exit && (method.equals("afterUpdate") || method.equals("commit")); + } + if (owner.equals(UPDATER + "GeneralRegionAttributeSecurityService")) { + return method.equals("sendUpdateRequestAndMayShrink") && exit; + } + return owner.equals(UPDATER + "UpdateDetailContainer") && method.equals("degrade") && exit; + } + + private static final class Transformer implements ClassFileTransformer { + @Override + public byte[] transform( + ClassLoader loader, + String name, + Class redefining, + ProtectionDomain domain, + byte[] bytes) { + if (!(name.equals(RPC) + || name.startsWith(FETCHER) + || name.startsWith(UPDATER) + || name.equals(LEASE))) { + return null; + } + ClassReader reader = new ClassReader(bytes); + // javac can discard this/parameter locals from a return frame when the original code no + // longer reads them. Exit hooks do read them, so retaining the original frames is invalid. + ClassWriter writer = new LoaderClassWriter(reader, loader); + reader.accept(new HookVisitor(writer, name), ClassReader.SKIP_FRAMES); + return writer.toByteArray(); + } + } + + private static final class LoaderClassWriter extends ClassWriter { + private final ClassLoader loader; + + private LoaderClassWriter(ClassReader reader, ClassLoader loader) { + super(reader, ClassWriter.COMPUTE_FRAMES); + this.loader = loader; + } + + @Override + protected ClassLoader getClassLoader() { + return loader; + } + } + + private static final class HookVisitor extends ClassVisitor { + private final String owner; + + private HookVisitor(ClassVisitor visitor, String owner) { + super(Opcodes.ASM9, visitor); + this.owner = owner; + } + + @Override + public MethodVisitor visitMethod( + int access, String name, String descriptor, String signature, String[] exceptions) { + MethodVisitor visitor = super.visitMethod(access, name, descriptor, signature, exceptions); + if (!selected(owner, name, false) && !selected(owner, name, true)) { + return visitor; + } + return new HookMethodVisitor(visitor, owner, name, descriptor); + } + } + + private static final class HookMethodVisitor extends MethodVisitor { + private final String owner; + private final String method; + private final String descriptor; + + private HookMethodVisitor( + MethodVisitor visitor, String owner, String method, String descriptor) { + super(Opcodes.ASM9, visitor); + this.owner = owner; + this.method = method; + this.descriptor = descriptor; + } + + @Override + public void visitCode() { + super.visitCode(); + if (selected(owner, method, false)) { + hook(false, false); + } + } + + @Override + public void visitInsn(int opcode) { + if ((opcode == Opcodes.RETURN || opcode == Opcodes.ARETURN) + && selected(owner, method, true)) { + hook(true, opcode == Opcodes.ARETURN); + } + super.visitInsn(opcode); + } + + private void hook(boolean exit, boolean objectReturn) { + if (objectReturn) { + super.visitInsn(Opcodes.DUP); + } + super.visitLdcInsn(method + (exit ? ":exit" : ":enter")); + if (objectReturn) { + super.visitInsn(Opcodes.SWAP); + } + super.visitVarInsn(Opcodes.ALOAD, 0); + if (objectReturn) { + super.visitInsn(Opcodes.SWAP); + } + Type[] arguments = Type.getArgumentTypes(descriptor); + super.visitLdcInsn(arguments.length); + super.visitTypeInsn(Opcodes.ANEWARRAY, "java/lang/Object"); + int local = 1; + for (int i = 0; i < arguments.length; i++) { + super.visitInsn(Opcodes.DUP); + super.visitLdcInsn(i); + // Selected entry/exit methods have reference parameters only. + if (arguments[i].getSort() != Type.OBJECT && arguments[i].getSort() != Type.ARRAY) { + throw new IllegalStateException("Unexpected hook signature: " + descriptor); + } + super.visitVarInsn(Opcodes.ALOAD, local++); + super.visitInsn(Opcodes.AASTORE); + } + if (objectReturn) { + super.visitInsn(Opcodes.SWAP); + } else { + super.visitInsn(Opcodes.ACONST_NULL); + } + super.visitMethodInsn( + Opcodes.INVOKESTATIC, + Type.getInternalName(MetadataFaultAgent.class), + "hit", + "(Ljava/lang/String;Ljava/lang/Object;[Ljava/lang/Object;Ljava/lang/Object;)V", + false); + } + } +} diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/fault/MetadataFaultController.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/fault/MetadataFaultController.java new file mode 100644 index 0000000000000..db7cd1e0d1625 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/fault/MetadataFaultController.java @@ -0,0 +1,244 @@ +/* + * 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.fault; + +import org.apache.iotdb.it.env.cluster.config.MppJVMConfig; +import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; + +import org.awaitility.Awaitility; +import org.objectweb.asm.ClassReader; + +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import java.util.jar.Attributes; +import java.util.jar.JarEntry; +import java.util.jar.JarFile; +import java.util.jar.JarOutputStream; +import java.util.jar.Manifest; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +/** Owns the agent's temporary jar, launch environment and bounded loopback control connections. */ +public final class MetadataFaultController implements AutoCloseable { + private final Path directory; + private final List nodes = new ArrayList<>(); + private final MppJVMConfig dataNodeJvmConfig; + private final List previousJvmOptions; + private boolean launchEnvironmentRestored; + + public MetadataFaultController(MppJVMConfig dataNodeJvmConfig) throws Exception { + this.dataNodeJvmConfig = dataNodeJvmConfig; + previousJvmOptions = new ArrayList<>(dataNodeJvmConfig.getJvmOptions()); + directory = Files.createTempDirectory(Paths.get("target").toAbsolutePath(), "metadata-fault-"); + Path jar = directory.resolve("agent.jar"); + Manifest manifest = new Manifest(); + manifest.getMainAttributes().put(Attributes.Name.MANIFEST_VERSION, "1.0"); + manifest.getMainAttributes().putValue("Premain-Class", MetadataFaultAgent.class.getName()); + try (JarOutputStream output = new JarOutputStream(Files.newOutputStream(jar), manifest)) { + addClass(output, MetadataFaultAgent.class); + // Keep ASM inside the temporary agent so child-process classpaths need no test artifacts. + Path asm = + Paths.get(ClassReader.class.getProtectionDomain().getCodeSource().getLocation().toURI()); + try (JarFile source = new JarFile(asm.toFile())) { + for (JarEntry entry : source.stream().collect(Collectors.toList())) { + if (entry.getName().startsWith("org/objectweb/asm/") && !entry.isDirectory()) { + output.putNextEntry(new JarEntry(entry.getName())); + try (InputStream input = source.getInputStream(entry)) { + copy(input, output); + } + output.closeEntry(); + } + } + } + } + List launchOptions = new ArrayList<>(previousJvmOptions); + launchOptions.add("-javaagent:" + jar + "=" + directory); + dataNodeJvmConfig.setJvmOptions(launchOptions); + } + + /** Called immediately after cluster startup, also on startup failure. */ + public void restoreLaunchEnvironment() { + if (launchEnvironmentRestored) { + return; + } + dataNodeJvmConfig.setJvmOptions(previousJvmOptions); + launchEnvironmentRestored = true; + } + + public Node connect(DataNodeWrapper wrapper) throws Exception { + long pid = wrapper.getInstance().pid(); + Path endpoint = directory.resolve(pid + ".port"); + Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() -> Files.isRegularFile(endpoint)); + Node node = new Node(Integer.parseInt(Files.readAllLines(endpoint).get(0))); + nodes.add(node); + return node; + } + + private static void addClass(JarOutputStream output, Class type) throws IOException { + String name = type.getName().replace('.', '/') + ".class"; + output.putNextEntry(new JarEntry(name)); + try (InputStream input = type.getClassLoader().getResourceAsStream(name)) { + if (input == null) { + throw new IOException("Missing agent class: " + name); + } + copy(input, output); + } + output.closeEntry(); + for (Class nested : type.getDeclaredClasses()) { + addClass(output, nested); + } + } + + private static void copy(InputStream input, JarOutputStream output) throws IOException { + byte[] buffer = new byte[8_192]; + int length; + while ((length = input.read(buffer)) != -1) { + output.write(buffer, 0, length); + } + } + + /** Release all in-flight calls before shutting down the cluster. */ + public void reset() throws IOException { + IOException failure = null; + for (Node node : nodes) { + try { + node.reset(); + } catch (IOException e) { + if (failure == null) { + failure = e; + } else { + failure.addSuppressed(e); + } + } + } + if (failure != null) { + throw failure; + } + } + + /** Called after the child processes have stopped. */ + @Override + public void close() throws IOException { + restoreLaunchEnvironment(); + try (Stream paths = Files.walk(directory)) { + for (Path path : paths.sorted(Comparator.reverseOrder()).collect(Collectors.toList())) { + Files.delete(path); + } + } + } + + public static final class Node { + private final int port; + + private Node(int port) { + this.port = port; + } + + public void arm(String hook, String mode, String database) throws IOException { + arm(hook, mode, database, ""); + } + + public void arm(String hook, String mode, String database, String attributes) + throws IOException { + request("arm", hook, mode, database, attributes); + } + + public Map status(String hook) throws IOException { + return request("status", hook, "", ""); + } + + public void release(String hook) throws IOException { + request("release", hook, "", ""); + } + + public void reset() throws IOException { + request("reset", "", "", ""); + } + + public Map lease() throws IOException { + return request("lease", "", "", ""); + } + + public Map updaters(String database) throws IOException { + return request("updaters", "", "", database); + } + + public long count(String hook) throws IOException { + return Long.parseLong(request("count", hook, "", "").get("count")); + } + + public void awaitHeld(String hook) { + Awaitility.await() + .atMost(30, TimeUnit.SECONDS) + .untilAsserted( + () -> { + Map state = status(hook); + if (!Boolean.parseBoolean(state.get("held"))) { + throw new AssertionError(hook + " did not reach the gate: " + state); + } + }); + } + + private Map request(String command, String hook, String mode, String match) + throws IOException { + return request(command, hook, mode, match, ""); + } + + private Map request( + String command, String hook, String mode, String match, String attributes) + throws IOException { + try (Socket socket = new Socket()) { + socket.connect(new InetSocketAddress(InetAddress.getLoopbackAddress(), port), 5_000); + socket.setSoTimeout(5_000); + try (DataOutputStream output = new DataOutputStream(socket.getOutputStream()); + DataInputStream input = new DataInputStream(socket.getInputStream())) { + output.writeUTF(command); + output.writeUTF(hook); + output.writeUTF(mode); + output.writeUTF(match); + output.writeUTF(attributes); + output.flush(); + Map result = new HashMap<>(); + int size = input.readInt(); + for (int i = 0; i < size; i++) { + result.put(input.readUTF(), input.readUTF()); + } + if (result.containsKey("error")) { + throw new IOException("Metadata IT controller: " + result.get("error")); + } + return result; + } + } + } + } +} 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 5710aeee7246a..7176c1279008f 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 0000000000000..30c6c04d913eb --- /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); + } +} From 6b5d8ca0be623b2da96f2fb6ae5fcf5bb356b535 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 5 Nov 2026 16:42:20 +0800 Subject: [PATCH 2/2] Remove redundant ASM integration-test dependency --- integration-test/pom.xml | 6 ------ 1 file changed, 6 deletions(-) diff --git a/integration-test/pom.xml b/integration-test/pom.xml index c98043a37505a..779c059e2d649 100644 --- a/integration-test/pom.xml +++ b/integration-test/pom.xml @@ -92,12 +92,6 @@ 3 - - org.ow2.asm - asm - - runtime - org.apache.iotdb iotdb-server