diff --git a/integration-test/src/test/java/org/apache/iotdb/tools/it/SchemaTagsRoundTripIT.java b/integration-test/src/test/java/org/apache/iotdb/tools/it/SchemaTagsRoundTripIT.java
new file mode 100644
index 0000000000000..a8f33bc59f454
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/tools/it/SchemaTagsRoundTripIT.java
@@ -0,0 +1,655 @@
+/*
+ * 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.tools.it;
+
+import org.apache.iotdb.cli.it.AbstractScriptIT;
+import org.apache.iotdb.isession.ISession;
+import org.apache.iotdb.isession.SessionConfig;
+import org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.tool.common.Constants;
+
+import com.google.gson.Gson;
+import com.google.gson.reflect.TypeToken;
+import org.apache.commons.csv.CSVFormat;
+import org.apache.commons.csv.CSVParser;
+import org.apache.commons.csv.CSVRecord;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.common.Field;
+import org.awaitility.Awaitility;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+
+import java.io.File;
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Proves that the tree model schema tools restore the tags and attributes of every time series:
+ * {@code export-schema} writes them as RFC4180 quoted JSON, {@code import-schema} reads that format
+ * back, and awkward characters survive the round trip unchanged.
+ *
+ * <p>The schema of one database and one exported file mixes everything the tools have to keep
+ * apart: every data type/encoding/compression combination, all four tag/attribute combinations on
+ * different devices, and several combinations across the measurements of one device.
+ */
+@RunWith(IoTDBTestRunner.class)
+@Category({LocalStandaloneIT.class, ClusterIT.class})
+public class SchemaTagsRoundTripIT extends AbstractScriptIT {
+
+  private static final String DATABASE = "root.rt";
+  private static final String EXPORT_PREFIX = "schema_tags";
+  private static final String HEADER =
+      "Timeseries,Alias,DataType,Encoding,Compression,Tags,Attributes";
+  private static final String ALIAS = "aliased";
+
+  /** The row whose tag value carries a double quote, i.e. the row that must be RFC4180 quoted. */
+  private static final String QUOTE_ROW = DATABASE + ".quote.s1";
+
+  /** The JSON object the export writes for the tag value {@code a"b}. */
+  private static final String QUOTE_ROW_TAGS_JSON = "{\"k\":\"a\\\"b\"}";
+
+  private static final String NO_ATTRIBUTES_ROW = DATABASE + ".none.s1";
+
+  /** How long the metadata drop may take to become visible on every node. */
+  private static final int SCHEMA_CLEANUP_TIMEOUT_SECONDS = 60;
+
+  private static final Type JSON_MAP_TYPE = new TypeToken<Map<String, String>>() {}.getType();
+  private static final Gson GSON = new Gson();
+  private static final Map<String, String> NO_VALUES = Collections.emptyMap();
+
+  /** Time series path to the tags, attributes and alias the test creates for it. */
+  private static final Map<String, Series> CREATED = new LinkedHashMap<>();
+
+  static {
+    // Awkward tag values, one device per value because tags belong to the device.
+    singleTag("v1", "v1", Format.BOOLEAN_PLAIN_SNAPPY);
+    singleTag("comma", "a,b", Format.INT32_TS_2DIFF_SNAPPY);
+    singleTag("quote", "a\"b", Format.DOUBLE_GORILLA_LZ4);
+    singleTag("backslash", "a\\b", Format.INT64_GORILLA_LZ4);
+    singleTag("backslashn", "a\\nb", Format.FLOAT_GORILLA_GZIP);
+    singleTag("tab", "a\tb", Format.DOUBLE_GORILLA_LZ4);
+    singleTag("newline", "a\nb", Format.TEXT_DICTIONARY_SNAPPY);
+    singleTag("cjk", "中文😀", Format.INT64_GORILLA_LZ4);
+    singleTag("nulllit", "null", Format.BOOLEAN_PLAIN_SNAPPY);
+    singleTag("empty", "", Format.INT32_TS_2DIFF_SNAPPY);
+    singleTag("backtick", "`x`", Format.FLOAT_GORILLA_GZIP);
+    tagged("spacedkey", "k e y", "v", Format.TEXT_DICTIONARY_SNAPPY);
+
+    // The four tag/attribute combinations, each on a device of its own.
+    series(
+        "both",
+        "s1",
+        Format.BOOLEAN_PLAIN_SNAPPY,
+        mapOf("t1", "v1", "t2", "a,b"),
+        mapOf("a1", "one", "a2", "he said \"hi\""),
+        null);
+    series(
+        "multi",
+        "s1",
+        Format.TEXT_DICTIONARY_SNAPPY,
+        mapOf("a", "1", "b", "a,b", "c", "a\"b"),
+        NO_VALUES,
+        null);
+    series(
+        "attr",
+        "s1",
+        Format.TEXT_DICTIONARY_SNAPPY,
+        NO_VALUES,
+        mapOf("note", "he said \"hi\"", "plain", "v"),
+        null);
+    series("none", "s1", Format.TEXT_DICTIONARY_SNAPPY, NO_VALUES, NO_VALUES, null);
+    series("alias", "s1", Format.FLOAT_GORILLA_GZIP, mapOf("aliasTag", "v"), NO_VALUES, ALIAS);
+
+    // One device mixing three of the combinations across its measurements.
+    series(
+        "mixed",
+        "tagsOnly",
+        Format.INT32_TS_2DIFF_SNAPPY,
+        mapOf("m1", "x", "m2", "y"),
+        NO_VALUES,
+        null);
+    series(
+        "mixed",
+        "attrsOnly",
+        Format.INT64_GORILLA_LZ4,
+        NO_VALUES,
+        mapOf("m1", "x", "m2", "y"),
+        null);
+    series("mixed", "neither", Format.DOUBLE_GORILLA_LZ4, NO_VALUES, NO_VALUES, null);
+  }
+
+  @Rule public TemporaryFolder folder = new TemporaryFolder();
+
+  private static String ip;
+  private static String port;
+  private static String toolsPath;
+  private static String homePath;
+
+  @BeforeClass
+  public static void setUp() {
+    EnvFactory.getEnv().initClusterEnvironment();
+    ip = EnvFactory.getEnv().getIP();
+    port = EnvFactory.getEnv().getPort();
+    toolsPath = EnvFactory.getEnv().getToolsPath();
+    final String libPath = EnvFactory.getEnv().getLibPath();
+    homePath =
+        libPath.substring(0, libPath.lastIndexOf(File.separator + "lib" + File.separator + "*"));
+  }
+
+  @AfterClass
+  public static void tearDown() {
+    EnvFactory.getEnv().cleanClusterEnvironment();
+  }
+
+  @Test
+  public void test() throws IOException {
+    String os = System.getProperty("os.name").toLowerCase();
+    if (os.startsWith("windows")) {
+      testOnWindows();
+    } else {
+      testOnUnix();
+    }
+  }
+
+  @Override
+  protected void testOnWindows() throws IOException {
+    try {
+      roundTrip(true);
+    } catch (IOException e) {
+      throw e;
+    } catch (Exception e) {
+      throw new IOException(e);
+    }
+  }
+
+  @Override
+  protected void testOnUnix() throws IOException {
+    try {
+      roundTrip(false);
+    } catch (IOException e) {
+      throw e;
+    } catch (Exception e) {
+      throw new IOException(e);
+    }
+  }
+
+  /**
+   * Exports the schema, drops it, imports the exported file again and compares the tags and
+   * attributes read back from the server with the ones that were created.
+   */
+  private void roundTrip(boolean windows) throws Exception {
+    final Path outputDirectory = folder.newFolder("export").toPath();
+
+    createSchema();
+
+    testOutput(
+        exportProcessBuilder(windows, outputDirectory),
+        new String[] {Constants.EXPORT_COMPLETELY},
+        0);
+
+    final Path csv = singleExportFile(outputDirectory);
+    assertExportFormat(csv);
+
+    dropSchema();
+
+    testOutput(importProcessBuilder(windows, csv), new String[] {Constants.IMPORT_COMPLETELY}, 0);
+
+    assertSchemaRestored();
+  }
+
+  private static void createSchema() throws IoTDBConnectionException, StatementExecutionException {
+    try (ISession session = EnvFactory.getEnv().getSessionConnection()) {
+      session.createDatabase(DATABASE);
+      for (Map.Entry<String, Series> entry : CREATED.entrySet()) {
+        final Series series = entry.getValue();
+        // Set the tags and attributes through the session API, so that no SQL escaping can hide a
+        // character that the export or the import tool mangles.
+        session.createTimeseries(
+            entry.getKey(),
+            series.format.dataType,
+            series.format.encoding,
+            series.format.compression,
+            null,
+            series.tags.isEmpty() ? null : series.tags,
+            series.attributes.isEmpty() ? null : series.attributes,
+            series.alias);
+      }
+    }
+  }
+
+  private static void dropSchema() throws IoTDBConnectionException, StatementExecutionException {
+    try (ISession session = EnvFactory.getEnv().getSessionConnection()) {
+      session.executeNonQueryStatement("DROP DATABASE " + DATABASE);
+    }
+    // The drop reaches the DataNodes asynchronously. Waiting until it is observable keeps a
+    // leftover device from making the import pass without restoring anything.
+    Awaitility.await()
+        .atMost(SCHEMA_CLEANUP_TIMEOUT_SECONDS, TimeUnit.SECONDS)
+        .pollInterval(500, TimeUnit.MILLISECONDS)
+        .until(() -> !databaseExists() && readSchema().isEmpty());
+    try (ISession session = EnvFactory.getEnv().getSessionConnection()) {
+      // Importing a file only creates time series, so the database has to exist beforehand.
+      session.createDatabase(DATABASE);
+    }
+  }
+
+  private static boolean databaseExists()
+      throws IoTDBConnectionException, StatementExecutionException {
+    try (ISession session = EnvFactory.getEnv().getSessionConnection();
+        SessionDataSet dataSet = session.executeQueryStatement("SHOW DATABASES")) {
+      while (dataSet.hasNext()) {
+        if (DATABASE.equals(cell(dataSet.next().getFields(), 0))) {
+          return true;
+        }
+      }
+    }
+    return false;
+  }
+
+  private Path singleExportFile(Path outputDirectory) throws IOException {
+    try (Stream<Path> files = Files.list(outputDirectory)) {
+      final List<Path> csvFiles =
+          files
+              .filter(file -> file.getFileName().toString().startsWith(EXPORT_PREFIX))
+              .filter(file -> file.getFileName().toString().endsWith(".csv"))
+              .collect(Collectors.toList());
+      assertEquals("expected exactly one exported CSV file, found " + csvFiles, 1, csvFiles.size());
+      return csvFiles.get(0);
+    }
+  }
+
+  private static void assertExportFormat(Path csv) throws IOException {
+    final List<String> lines = Files.readAllLines(csv, StandardCharsets.UTF_8);
+    assertTrue("the exported schema file is empty", lines.size() > 1);
+    assertEquals(HEADER, lines.get(0));
+
+    final String content = Files.readString(csv, StandardCharsets.UTF_8);
+    // RFC4180 uses LF as the record separator; a CR would come from a platform dependent writer.
+    assertFalse("the export must use LF line endings", content.contains("\r"));
+    // RFC4180: the JSON object of the value containing a double quote is wrapped in a quoted cell
+    // and every inner quote is doubled, instead of being escaped with a backslash.
+    final String quotedCell = "\"" + QUOTE_ROW_TAGS_JSON.replace("\"", "\"\"") + "\"";
+    assertTrue(
+        "the Tags cell of "
+            + QUOTE_ROW
+            + " is not RFC4180 quoted: "
+            + quotedCell
+            + " in\n"
+            + content,
+        content.contains(quotedCell));
+
+    try (CSVParser parser = CSVParser.parse(csv, StandardCharsets.UTF_8, csvFormat())) {
+      final Map<String, CSVRecord> records = new TreeMap<>();
+      for (CSVRecord record : parser) {
+        records.put(record.get("Timeseries"), record);
+      }
+      assertEquals(
+          "unexpected number of exported rows: " + records, CREATED.size(), records.size());
+
+      final Set<Format> formats = EnumSet.noneOf(Format.class);
+      for (Map.Entry<String, Series> entry : CREATED.entrySet()) {
+        final String path = entry.getKey();
+        final Series series = entry.getValue();
+        final CSVRecord record = records.get(path);
+        assertNotNull("the export is missing " + path, record);
+        assertEquals(path + " DataType", series.format.dataType.name(), record.get("DataType"));
+        assertEquals(path + " Encoding", series.format.encoding.name(), record.get("Encoding"));
+        assertEquals(
+            path + " Compression", series.format.compression.name(), record.get("Compression"));
+        assertEquals(
+            path + " Alias", series.alias == null ? "" : series.alias, record.get("Alias"));
+        assertEquals(path + " Tags", series.tags, parseJsonObject(record.get("Tags")));
+        assertEquals(
+            path + " Attributes", series.attributes, parseJsonObject(record.get("Attributes")));
+        // The three assertions above proved that the file carries this combination.
+        formats.add(series.format);
+      }
+      assertEquals(
+          "the file must carry every data type, encoding and compression combination",
+          EnumSet.allOf(Format.class),
+          formats);
+
+      // Nothing to write is written as an empty cell, not as {} and not as null.
+      assertEquals("", records.get(NO_ATTRIBUTES_ROW).get("Tags"));
+      assertEquals("", records.get(NO_ATTRIBUTES_ROW).get("Attributes"));
+
+      assertPresenceCombinations(records);
+    }
+  }
+
+  /**
+   * The one exported file must carry all four tag/attribute combinations, spread over several
+   * devices, and at least one device must mix several of them across its measurements.
+   */
+  private static void assertPresenceCombinations(Map<String, CSVRecord> records) {
+    final Set<Presence> presences = EnumSet.noneOf(Presence.class);
+    final Map<String, Set<Presence>> presencesByDevice = new TreeMap<>();
+    for (Map.Entry<String, CSVRecord> entry : records.entrySet()) {
+      final String path = entry.getKey();
+      final CSVRecord record = entry.getValue();
+      final Presence presence =
+          Presence.of(
+              parseJsonObject(record.get("Tags")), parseJsonObject(record.get("Attributes")));
+      presences.add(presence);
+      presencesByDevice
+          .computeIfAbsent(
+              path.substring(0, path.lastIndexOf('.')), device -> EnumSet.noneOf(Presence.class))
+          .add(presence);
+    }
+    assertEquals(
+        "the file must carry every tag/attribute combination",
+        EnumSet.allOf(Presence.class),
+        presences);
+    assertTrue(
+        "the combinations must be spread over at least four devices: " + presencesByDevice,
+        presencesByDevice.size() >= 4);
+    assertTrue(
+        "one device must mix several combinations across its measurements: " + presencesByDevice,
+        presencesByDevice.values().stream().mapToInt(Set::size).max().orElse(0) >= 3);
+  }
+
+  private static void assertSchemaRestored()
+      throws IoTDBConnectionException, StatementExecutionException {
+    final Map<String, Series> restored = readSchema();
+    assertEquals("the import must restore every time series", CREATED.keySet(), restored.keySet());
+    for (Map.Entry<String, Series> entry : CREATED.entrySet()) {
+      final String path = entry.getKey();
+      final Series expected = entry.getValue();
+      final Series actual = restored.get(path);
+      assertEquals(path + " data type/encoding/compression", expected.format, actual.format);
+      assertEquals(path + " tags", expected.tags, actual.tags);
+      assertEquals(path + " attributes", expected.attributes, actual.attributes);
+      assertEquals(path + " alias", expected.alias, actual.alias);
+    }
+  }
+
+  /** Reads the tags, attributes and alias of every time series of the test database. */
+  private static Map<String, Series> readSchema()
+      throws IoTDBConnectionException, StatementExecutionException {
+    final Map<String, Series> schema = new TreeMap<>();
+    try (ISession session = EnvFactory.getEnv().getSessionConnection();
+        SessionDataSet dataSet =
+            session.executeQueryStatement("SHOW TIMESERIES " + DATABASE + ".**")) {
+      final List<String> columns = dataSet.getColumnNames();
+      final int timeseriesIndex = columns.indexOf("Timeseries");
+      final int aliasIndex = columns.indexOf("Alias");
+      final int dataTypeIndex = columns.indexOf("DataType");
+      final int encodingIndex = columns.indexOf("Encoding");
+      final int compressionIndex = columns.indexOf("Compression");
+      final int tagsIndex = columns.indexOf("Tags");
+      final int attributesIndex = columns.indexOf("Attributes");
+      assertTrue(
+          "SHOW TIMESERIES must expose the Timeseries, Alias, DataType, Encoding, Compression, "
+              + "Tags and Attributes columns: "
+              + columns,
+          timeseriesIndex >= 0
+              && aliasIndex >= 0
+              && dataTypeIndex >= 0
+              && encodingIndex >= 0
+              && compressionIndex >= 0
+              && tagsIndex >= 0
+              && attributesIndex >= 0);
+      while (dataSet.hasNext()) {
+        final List<Field> fields = dataSet.next().getFields();
+        final String path = cell(fields, timeseriesIndex);
+        schema.put(
+            path,
+            new Series(
+                Format.of(
+                    cell(fields, dataTypeIndex),
+                    cell(fields, encodingIndex),
+                    cell(fields, compressionIndex)),
+                parseJsonObject(cell(fields, tagsIndex)),
+                parseJsonObject(cell(fields, attributesIndex)),
+                cell(fields, aliasIndex)));
+      }
+    }
+    return schema;
+  }
+
+  /** Returns a cell of a schema row, or {@code null} when the server sent a NULL there. */
+  private static String cell(List<Field> fields, int index) {
+    if (index < 0 || index >= fields.size()) {
+      return null;
+    }
+    final Field field = fields.get(index);
+    if (field == null) {
+      return null;
+    }
+    // A NULL field is rendered as the text "null" by getStringValue(), which is how the export
+    // tool distinguishes a missing value as well.
+    final String value = field.getStringValue();
+    return "null".equals(value) ? null : value;
+  }
+
+  /** Parses a JSON object cell into a map; an empty cell carries no tags and no attributes. */
+  private static Map<String, String> parseJsonObject(String cell) {
+    if (cell == null || cell.trim().isEmpty()) {
+      return new TreeMap<>();
+    }
+    final Map<String, String> parsed = GSON.fromJson(cell, JSON_MAP_TYPE);
+    return parsed == null ? new TreeMap<>() : new TreeMap<>(parsed);
+  }
+
+  private static CSVFormat csvFormat() {
+    return CSVFormat.Builder.create(CSVFormat.DEFAULT)
+        .setHeader()
+        .setSkipHeaderRecord(true)
+        .build();
+  }
+
+  private ProcessBuilder exportProcessBuilder(boolean windows, Path outputDirectory) {
+    final List<String> command = new ArrayList<>();
+    addScriptInvocation(command, windows, "export-schema");
+    addConnectionArguments(command);
+    command.addAll(
+        Arrays.asList(
+            "-t",
+            outputDirectory.toAbsolutePath().toString(),
+            "-pfn",
+            EXPORT_PREFIX,
+            "-path",
+            DATABASE + ".**"));
+    return processBuilder(command, windows);
+  }
+
+  private ProcessBuilder importProcessBuilder(boolean windows, Path csv) {
+    final List<String> command = new ArrayList<>();
+    addScriptInvocation(command, windows, "import-schema");
+    addConnectionArguments(command);
+    command.addAll(Arrays.asList("-db", DATABASE, "-s", csv.toAbsolutePath().toString()));
+    return processBuilder(command, windows);
+  }
+
+  private static void addConnectionArguments(List<String> command) {
+    command.addAll(
+        Arrays.asList(
+            "-h",
+            ip,
+            "-p",
+            port,
+            "-u",
+            SessionConfig.DEFAULT_USER,
+            "-pw",
+            SessionConfig.DEFAULT_PASSWORD));
+  }
+
+  private static void addScriptInvocation(List<String> command, boolean windows, String script) {
+    if (windows) {
+      command.add("cmd.exe");
+      command.add("/c");
+      command.add(
+          toolsPath
+              + File.separator
+              + "windows"
+              + File.separator
+              + "schema"
+              + File.separator
+              + script
+              + ".bat");
+    } else {
+      command.add("bash");
+      command.add(toolsPath + File.separator + "schema" + File.separator + script + ".sh");
+    }
+  }
+
+  private static ProcessBuilder processBuilder(List<String> command, boolean windows) {
+    if (windows) {
+      // cmd.exe exits with the error level of the script it ran.
+      command.add("&");
+      command.add("exit");
+      command.add("%^errorlevel%");
+    }
+    final ProcessBuilder builder = new ProcessBuilder(command);
+    builder.environment().put("IOTDB_HOME", homePath);
+    return builder;
+  }
+
+  private static void singleTag(String device, String value, Format format) {
+    tagged(device, "k", value, format);
+  }
+
+  private static void tagged(String device, String key, String value, Format format) {
+    series(device, "s1", format, mapOf(key, value), NO_VALUES, null);
+  }
+
+  private static void series(
+      String device,
+      String measurement,
+      Format format,
+      Map<String, String> tags,
+      Map<String, String> attributes,
+      String alias) {
+    CREATED.put(
+        DATABASE + "." + device + "." + measurement, new Series(format, tags, attributes, alias));
+  }
+
+  private static Map<String, String> mapOf(String... keysAndValues) {
+    final Map<String, String> map = new LinkedHashMap<>();
+    for (int i = 0; i < keysAndValues.length; i += 2) {
+      map.put(keysAndValues[i], keysAndValues[i + 1]);
+    }
+    return map;
+  }
+
+  /**
+   * A data type, encoding and compression combination of the test schema. Every combination has to
+   * appear in the exported file, and the server renders them with these names.
+   */
+  private enum Format {
+    BOOLEAN_PLAIN_SNAPPY(TSDataType.BOOLEAN, TSEncoding.PLAIN, CompressionType.SNAPPY),
+    INT32_TS_2DIFF_SNAPPY(TSDataType.INT32, TSEncoding.TS_2DIFF, CompressionType.SNAPPY),
+    INT64_GORILLA_LZ4(TSDataType.INT64, TSEncoding.GORILLA, CompressionType.LZ4),
+    DOUBLE_GORILLA_LZ4(TSDataType.DOUBLE, TSEncoding.GORILLA, CompressionType.LZ4),
+    FLOAT_GORILLA_GZIP(TSDataType.FLOAT, TSEncoding.GORILLA, CompressionType.GZIP),
+    TEXT_DICTIONARY_SNAPPY(TSDataType.TEXT, TSEncoding.DICTIONARY, CompressionType.SNAPPY);
+
+    private final TSDataType dataType;
+    private final TSEncoding encoding;
+    private final CompressionType compression;
+
+    Format(TSDataType dataType, TSEncoding encoding, CompressionType compression) {
+      this.dataType = dataType;
+      this.encoding = encoding;
+      this.compression = compression;
+    }
+
+    /** Resolves the combination rendered by {@code SHOW TIMESERIES}, i.e. by the export. */
+    private static Format of(String dataType, String encoding, String compression) {
+      for (Format format : values()) {
+        if (format.dataType.name().equals(dataType)
+            && format.encoding.name().equals(encoding)
+            && format.compression.name().equals(compression)) {
+          return format;
+        }
+      }
+      throw new AssertionError(
+          "unexpected data type/encoding/compression combination: "
+              + dataType
+              + "/"
+              + encoding
+              + "/"
+              + compression);
+    }
+  }
+
+  /** Whether a time series carries tags and attributes; all four must coexist in one file. */
+  private enum Presence {
+    BOTH,
+    TAGS_ONLY,
+    ATTRIBUTES_ONLY,
+    NEITHER;
+
+    private static Presence of(Map<String, String> tags, Map<String, String> attributes) {
+      if (!tags.isEmpty()) {
+        return attributes.isEmpty() ? TAGS_ONLY : BOTH;
+      }
+      return attributes.isEmpty() ? NEITHER : ATTRIBUTES_ONLY;
+    }
+  }
+
+  /** One time series of the test schema, i.e. one row of the exported file. */
+  private static final class Series {
+    private final Format format;
+    private final Map<String, String> tags;
+    private final Map<String, String> attributes;
+    private final String alias;
+
+    private Series(
+        Format format, Map<String, String> tags, Map<String, String> attributes, String alias) {
+      this.format = format;
+      this.tags = tags;
+      this.attributes = attributes;
+      this.alias = alias;
+    }
+  }
+}
diff --git a/iotdb-client/cli/pom.xml b/iotdb-client/cli/pom.xml
index a1e00508740bd..5302c86b8fd81 100644
--- a/iotdb-client/cli/pom.xml
+++ b/iotdb-client/cli/pom.xml
@@ -144,6 +144,10 @@
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-csv</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.google.code.gson</groupId>
+            <artifactId>gson</artifactId>
+        </dependency>
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>
diff --git a/iotdb-client/cli/src/main/i18n/en/org/apache/iotdb/cli/i18n/CliMessages.java b/iotdb-client/cli/src/main/i18n/en/org/apache/iotdb/cli/i18n/CliMessages.java
index 8cd97e9b07884..5ac63fff12c65 100644
--- a/iotdb-client/cli/src/main/i18n/en/org/apache/iotdb/cli/i18n/CliMessages.java
+++ b/iotdb-client/cli/src/main/i18n/en/org/apache/iotdb/cli/i18n/CliMessages.java
@@ -157,6 +157,13 @@ public final class CliMessages {
       MESSAGE_THE_HEADER_OF_THE_CSV_FILE_TO_BE_IMPORTED_IS_ILLEGAL_THE_CORRECT_FORMAT_IS_TIMESERIES_ALIBABA_DATATYPE_ENCODING_COMPRESSION_C429E7C6 =
           "The header of the CSV file to be imported is illegal. The correct format is"
               + " \"Timeseries, Alibaba, DataType, Encoding, Compression\"!";
+  public static final String
+      MESSAGE_THE_HEADER_OF_THE_CSV_FILE_TO_BE_IMPORTED_IS_ILLEGAL_THE_CORRECT_FORMAT_IS_TIMESERIES_ALIAS_DATATYPE_ENCODING_COMPRESSION_TAGS_ATTRIBUTES_14EDBFB9 =
+          "The header of the CSV file to be imported is illegal. The correct format is"
+              + " \"Timeseries, Alias, DataType, Encoding, Compression, Tags, Attributes\"!";
+  public static final String
+      MESSAGE_FAILED_TO_PARSE_THE_ARG_COLUMN_OF_TIMESERIES_ARG_AS_JSON_THE_RECORD_IS_SKIPPED_ARG_35D898C8 =
+          "Failed to parse the %s column of timeseries %s as JSON, the record is skipped: %s";
 
   private CliMessages() {}
 
diff --git a/iotdb-client/cli/src/main/i18n/zh/org/apache/iotdb/cli/i18n/CliMessages.java b/iotdb-client/cli/src/main/i18n/zh/org/apache/iotdb/cli/i18n/CliMessages.java
index 6123237db8d74..7efdc8d83ec7e 100644
--- a/iotdb-client/cli/src/main/i18n/zh/org/apache/iotdb/cli/i18n/CliMessages.java
+++ b/iotdb-client/cli/src/main/i18n/zh/org/apache/iotdb/cli/i18n/CliMessages.java
@@ -145,6 +145,13 @@ public final class CliMessages {
   public static final String
       MESSAGE_THE_HEADER_OF_THE_CSV_FILE_TO_BE_IMPORTED_IS_ILLEGAL_THE_CORRECT_FORMAT_IS_TIMESERIES_ALIBABA_DATATYPE_ENCODING_COMPRESSION_C429E7C6 =
           "待导入的 CSV 文件表头不合法，正确格式为 \"Timeseries, Alibaba, DataType, Encoding, Compression\"！";
+  public static final String
+      MESSAGE_THE_HEADER_OF_THE_CSV_FILE_TO_BE_IMPORTED_IS_ILLEGAL_THE_CORRECT_FORMAT_IS_TIMESERIES_ALIAS_DATATYPE_ENCODING_COMPRESSION_TAGS_ATTRIBUTES_14EDBFB9 =
+          "待导入的 CSV 文件表头不合法，正确格式为"
+              + " \"Timeseries, Alias, DataType, Encoding, Compression, Tags, Attributes\"！";
+  public static final String
+      MESSAGE_FAILED_TO_PARSE_THE_ARG_COLUMN_OF_TIMESERIES_ARG_AS_JSON_THE_RECORD_IS_SKIPPED_ARG_35D898C8 =
+          "时序 JSON 解析失败，列：%s，时序：%s，已跳过该记录，原因：%s";
 
   private CliMessages() {}
 
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/common/Constants.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/common/Constants.java
index ec4d2c7632cbc..e2c96f4d2e744 100644
--- a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/common/Constants.java
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/common/Constants.java
@@ -204,8 +204,31 @@ public class Constants {
   public static final String COLON = ": ";
   public static final String MINUS = "-";
 
+  public static final String HEADER_TAGS = "Tags";
+  public static final String HEADER_ATTRIBUTES = "Attributes";
+  public static final String HEADER_ALIAS = "Alias";
+  public static final String HEADER_DATATYPE = "DataType";
+  public static final String HEADER_ENCODING = "Encoding";
+  public static final String HEADER_COMPRESSION = "Compression";
+
   public static final List<String> HEAD_COLUMNS =
-      Arrays.asList("Timeseries", "Alias", "DataType", "Encoding", "Compression");
+      Arrays.asList(
+          "Timeseries", HEADER_ALIAS, HEADER_DATATYPE, HEADER_ENCODING, HEADER_COMPRESSION);
+
+  /**
+   * Column layout written by the tree model schema export. The Tags and Attributes cells hold a
+   * JSON object, so files carrying them are parsed with the RFC4180 settings instead of the legacy
+   * ones.
+   */
+  public static final List<String> HEAD_COLUMNS_V2 =
+      Arrays.asList(
+          "Timeseries",
+          HEADER_ALIAS,
+          HEADER_DATATYPE,
+          HEADER_ENCODING,
+          HEADER_COMPRESSION,
+          HEADER_TAGS,
+          HEADER_ATTRIBUTES);
 
   // export constants
   public static final String EXPORT_CLI_PREFIX = CliMessages.EXCEPTION_EXPORT_DATA_2F1FB3D9;
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/schema/AbstractSchemaTool.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/schema/AbstractSchemaTool.java
index b88cfda2c0a32..f9998491a1f41 100644
--- a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/schema/AbstractSchemaTool.java
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/schema/AbstractSchemaTool.java
@@ -30,6 +30,9 @@
 import org.apache.iotdb.session.pool.TableSessionPoolBuilder;
 import org.apache.iotdb.tool.common.Constants;
 
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.gson.reflect.TypeToken;
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.csv.CSVFormat;
 import org.apache.commons.csv.CSVPrinter;
@@ -41,8 +44,11 @@
 
 import java.io.IOException;
 import java.io.PrintWriter;
+import java.lang.reflect.Type;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
 import java.util.concurrent.atomic.LongAdder;
 
 public abstract class AbstractSchemaTool {
@@ -77,6 +83,9 @@ public abstract class AbstractSchemaTool {
   private static final IoTPrinter ioTPrinter = new IoTPrinter(System.out);
   private static final Logger LOGGER = LoggerFactory.getLogger(AbstractSchemaTool.class);
 
+  private static final Gson JSON_GSON = new GsonBuilder().disableHtmlEscaping().create();
+  private static final Type JSON_CELL_TYPE = new TypeToken<Map<String, String>>() {}.getType();
+
   protected AbstractSchemaTool() {}
 
   protected static Session.Builder configureSsl(Session.Builder builder) {
@@ -218,6 +227,51 @@ public static Boolean writeCsvFile(List<List<Object>> records, String filePath)
     }
   }
 
+  /**
+   * The RFC4180 CSV layout of the schema files, i.e. the delimiter is {@code ','}, the quote char
+   * is {@code '"'} and no escape char is configured. A JSON object cell is therefore always quoted
+   * and its inner quotes are doubled, which is exactly what RFC4180 asks for.
+   */
+  static CSVFormat.Builder createRfc4180CsvFormat() {
+    return CSVFormat.Builder.create(CSVFormat.DEFAULT)
+        .setQuote('"')
+        .setQuoteMode(QuoteMode.MINIMAL)
+        .setRecordSeparator('\n');
+  }
+
+  /**
+   * Serializes tags or attributes into the compact JSON object cell of the schema file. Keys are
+   * sorted so that repeated exports of the same schema are identical, and HTML escaping is disabled
+   * so that printable characters such as {@code '}, {@code =}, {@code <} and {@code &} stay
+   * readable.
+   *
+   * @return an empty string when there is nothing to write, never {@code {}} or {@code null}
+   */
+  static String toJsonCell(Map<String, String> values) {
+    if (values == null || values.isEmpty()) {
+      return "";
+    }
+    return JSON_GSON.toJson(new TreeMap<>(values));
+  }
+
+  /**
+   * Parses a Tags / Attributes cell written by {@link #toJsonCell(Map)}.
+   *
+   * @return {@code null} when the cell carries no value, never an empty map
+   * @throws com.google.gson.JsonParseException when the cell is not a JSON object
+   */
+  static Map<String, String> parseJsonCell(String cell) {
+    if (cell == null) {
+      return null;
+    }
+    String trimmed = cell.trim();
+    if (trimmed.isEmpty() || "null".equals(trimmed)) {
+      return null;
+    }
+    Map<String, String> parsed = JSON_GSON.fromJson(trimmed, JSON_CELL_TYPE);
+    return parsed == null || parsed.isEmpty() ? null : new TreeMap<>(parsed);
+  }
+
   static class CSVPrinterWrapper {
     private final String filePath;
     private final CSVFormat csvFormat;
@@ -225,13 +279,7 @@ static class CSVPrinterWrapper {
 
     public CSVPrinterWrapper(String filePath) {
       this.filePath = filePath;
-      this.csvFormat =
-          CSVFormat.Builder.create(CSVFormat.DEFAULT)
-              .setHeader()
-              .setSkipHeaderRecord(true)
-              .setEscape('\\')
-              .setQuoteMode(QuoteMode.NONE)
-              .build();
+      this.csvFormat = createRfc4180CsvFormat().setHeader().setSkipHeaderRecord(true).build();
     }
 
     public void printRecord(final Iterable<?> values) throws IOException {
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/schema/ExportSchemaTree.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/schema/ExportSchemaTree.java
index b8cd60323e22c..12e3b5066485a 100644
--- a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/schema/ExportSchemaTree.java
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/schema/ExportSchemaTree.java
@@ -32,7 +32,9 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 
 import static org.apache.iotdb.commons.schema.SchemaConstant.AUDIT_DATABASE;
 import static org.apache.iotdb.commons.schema.SchemaConstant.SYSTEM_DATABASE;
@@ -98,25 +100,19 @@ private static void writeCsvFile(
       while (i++ < linesPerFile) {
         if (sessionDataSet.hasNext()) {
           if (i == 1) {
-            csvPrinterWrapper.printRecord(Constants.HEAD_COLUMNS);
+            csvPrinterWrapper.printRecord(Constants.HEAD_COLUMNS_V2);
           }
           RowRecord rowRecord = sessionDataSet.next();
           List<Field> fields = rowRecord.getFields();
-          if (fields.get(timeseriesIndex).getStringValue().startsWith(SYSTEM_DATABASE + ".")
-              || fields.get(timeseriesIndex).getStringValue().startsWith(AUDIT_DATABASE + ".")
-              || !fields.get(viewTypeIndex).getStringValue().equals(Constants.BASE_VIEW_TYPE)) {
+          String timeseries = fields.get(timeseriesIndex).getStringValue();
+          if (timeseries.startsWith(SYSTEM_DATABASE + ".")
+              || timeseries.startsWith(AUDIT_DATABASE + ".")
+              || !Constants.BASE_VIEW_TYPE.equals(fields.get(viewTypeIndex).getStringValue())) {
+            continue;
+          }
+          if (!printRecord(csvPrinterWrapper, headers, fields, timeseries)) {
             continue;
           }
-          Constants.HEAD_COLUMNS.forEach(
-              column -> {
-                Field field = fields.get(headers.indexOf(column));
-                String fieldStringValue = field.getStringValue();
-                if (!"null".equals(field.getStringValue())) {
-                  csvPrinterWrapper.print(fieldStringValue);
-                } else {
-                  csvPrinterWrapper.print("");
-                }
-              });
           csvPrinterWrapper.println();
         } else {
           hasNext = false;
@@ -128,4 +124,44 @@ private static void writeCsvFile(
       csvPrinterWrapper.close();
     }
   }
+
+  /**
+   * Writes the cells of one {@code SHOW TIMESERIES} record.
+   *
+   * @return {@code false} when the record must be skipped because a Tags or Attributes cell could
+   *     not be parsed, in which case nothing has been written for it
+   */
+  static boolean printRecord(
+      CSVPrinterWrapper csvPrinterWrapper,
+      List<String> headers,
+      List<Field> fields,
+      String timeseries) {
+    List<String> cells = new ArrayList<>(Constants.HEAD_COLUMNS_V2.size());
+    for (String column : Constants.HEAD_COLUMNS_V2) {
+      int index = headers.indexOf(column);
+      Field field = index < 0 ? null : fields.get(index);
+      String fieldStringValue = field == null ? null : field.getStringValue();
+      if (Constants.HEADER_TAGS.equals(column) || Constants.HEADER_ATTRIBUTES.equals(column)) {
+        Map<String, String> tags;
+        try {
+          tags = parseJsonCell(fieldStringValue);
+        } catch (RuntimeException e) {
+          ioTPrinter.println(
+              String.format(
+                  CliMessages
+                      .MESSAGE_FAILED_TO_PARSE_THE_ARG_COLUMN_OF_TIMESERIES_ARG_AS_JSON_THE_RECORD_IS_SKIPPED_ARG_35D898C8,
+                  column,
+                  timeseries,
+                  String.valueOf(e.getMessage())));
+          return false;
+        }
+        cells.add(toJsonCell(tags));
+      } else {
+        cells.add(
+            "null".equals(fieldStringValue) || fieldStringValue == null ? "" : fieldStringValue);
+      }
+    }
+    cells.forEach(csvPrinterWrapper::print);
+    return true;
+  }
 }
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/schema/ImportSchemaTree.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/schema/ImportSchemaTree.java
index 231467b33fadc..9e0471a80d605 100644
--- a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/schema/ImportSchemaTree.java
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/schema/ImportSchemaTree.java
@@ -38,11 +38,13 @@
 import org.apache.tsfile.file.metadata.enums.CompressionType;
 import org.apache.tsfile.file.metadata.enums.TSEncoding;
 
+import java.io.BufferedReader;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStreamReader;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
@@ -58,7 +60,7 @@
 public class ImportSchemaTree extends AbstractImportSchema {
 
   private static final IoTPrinter ioTPrinter = new IoTPrinter(System.out);
-  private static SessionPool sessionPool;
+  static SessionPool sessionPool;
 
   public void init()
       throws InterruptedException, IoTDBConnectionException, StatementExecutionException {
@@ -131,18 +133,65 @@ protected void importSchemaFromCsvFile(File file) {
     }
   }
 
-  private static CSVParser readCsvFile(String path) throws IOException {
-    return CSVFormat.Builder.create(CSVFormat.DEFAULT)
-        .setHeader()
-        .setSkipHeaderRecord(true)
-        .setQuote('`')
-        .setEscape('\\')
-        .setIgnoreEmptyLines(true)
-        .build()
-        .parse(new InputStreamReader(new FileInputStream(path)));
+  static CSVParser readCsvFile(String path) throws IOException {
+    CSVFormat.Builder builder =
+        CSVFormat.Builder.create(CSVFormat.DEFAULT)
+            .setHeader()
+            .setSkipHeaderRecord(true)
+            .setIgnoreEmptyLines(true);
+    if (isV2Format(readHeaderNames(path))) {
+      builder.setQuote('"');
+    } else {
+      builder.setQuote('`').setEscape('\\');
+    }
+    return builder.build().parse(new InputStreamReader(new FileInputStream(path)));
+  }
+
+  /**
+   * Reads the column names of a schema file without parsing it, so that the CSV settings of the
+   * file can be selected before it is read.
+   */
+  static List<String> readHeaderNames(String path) throws IOException {
+    try (BufferedReader reader =
+        new BufferedReader(new InputStreamReader(new FileInputStream(path)))) {
+      String headerLine = reader.readLine();
+      if (headerLine == null) {
+        return Collections.emptyList();
+      }
+      return Arrays.stream(headerLine.split(","))
+          .map(ImportSchemaTree::unquoteColumnName)
+          .collect(Collectors.toList());
+    }
   }
 
-  private static boolean checkHeader(List<String> headerNames) {
+  /** Accepts a header written by a spreadsheet tool, which quotes every column name. */
+  private static String unquoteColumnName(String columnName) {
+    String name = columnName.trim();
+    if (name.length() > 1 && name.startsWith("\"") && name.endsWith("\"")) {
+      return name.substring(1, name.length() - 1).trim();
+    }
+    return name;
+  }
+
+  /**
+   * A file whose header carries a Tags or an Attributes column is a v2 file, which is RFC4180 and
+   * holds JSON cells. Everything else is a legacy file and keeps the settings it was written with.
+   */
+  static boolean isV2Format(List<String> headerNames) {
+    return headerNames.contains(Constants.HEADER_TAGS)
+        || headerNames.contains(Constants.HEADER_ATTRIBUTES);
+  }
+
+  static boolean checkHeader(List<String> headerNames) {
+    if (isV2Format(headerNames)) {
+      if (headerNames.containsAll(Constants.HEAD_COLUMNS_V2)) {
+        return true;
+      }
+      ioTPrinter.println(
+          CliMessages
+              .MESSAGE_THE_HEADER_OF_THE_CSV_FILE_TO_BE_IMPORTED_IS_ILLEGAL_THE_CORRECT_FORMAT_IS_TIMESERIES_ALIAS_DATATYPE_ENCODING_COMPRESSION_TAGS_ATTRIBUTES_14EDBFB9);
+      return false;
+    }
     if (CollectionUtils.isNotEmpty(headerNames)
         && new HashSet<>(headerNames).size() == Constants.HEAD_COLUMNS.size()) {
       List<String> strangers =
@@ -159,18 +208,25 @@ private static boolean checkHeader(List<String> headerNames) {
     return true;
   }
 
-  private static void writeScheme(
+  static void writeScheme(
       String fileName, List<String> headerNames, Stream<CSVRecord> records, String failedFilePath)
       throws IllegalPathException {
+    boolean v2 = isV2Format(headerNames);
+    int tagsIndex = headerNames.indexOf(Constants.HEADER_TAGS);
+    int attributesIndex = headerNames.indexOf(Constants.HEADER_ATTRIBUTES);
     List<String> paths = new ArrayList<>();
     List<TSDataType> dataTypes = new ArrayList<>();
     List<TSEncoding> encodings = new ArrayList<>();
     List<CompressionType> compressors = new ArrayList<>();
+    List<Map<String, String>> tags = v2 ? new ArrayList<>() : null;
+    List<Map<String, String>> attributes = v2 ? new ArrayList<>() : null;
     List<String> pathsWithAlias = new ArrayList<>();
     List<TSDataType> dataTypesWithAlias = new ArrayList<>();
     List<TSEncoding> encodingsWithAlias = new ArrayList<>();
     List<CompressionType> compressorsWithAlias = new ArrayList<>();
     List<String> measurementAlias = new ArrayList<>();
+    List<Map<String, String>> tagsWithAlias = v2 ? new ArrayList<>() : null;
+    List<Map<String, String>> attributesWithAlias = v2 ? new ArrayList<>() : null;
 
     AtomicReference<Boolean> hasStarted = new AtomicReference<>(false);
     AtomicInteger pointSize = new AtomicInteger(0);
@@ -185,7 +241,7 @@ private static void writeScheme(
               try {
                 if (CollectionUtils.isNotEmpty(paths)) {
                   writeAndEmptyDataSet(
-                      paths, dataTypes, encodings, compressors, null, null, null, null, 3);
+                      paths, dataTypes, encodings, compressors, null, tags, attributes, null, 3);
                 }
               } catch (Exception e) {
                 paths.forEach(t -> failedRecords.add(Collections.singletonList(t)));
@@ -198,8 +254,8 @@ private static void writeScheme(
                       encodingsWithAlias,
                       compressorsWithAlias,
                       null,
-                      null,
-                      null,
+                      tagsWithAlias,
+                      attributesWithAlias,
                       measurementAlias,
                       3);
                 }
@@ -210,7 +266,17 @@ private static void writeScheme(
               dataTypes.clear();
               encodings.clear();
               compressors.clear();
+              pathsWithAlias.clear();
+              dataTypesWithAlias.clear();
+              encodingsWithAlias.clear();
+              compressorsWithAlias.clear();
               measurementAlias.clear();
+              if (v2) {
+                tags.clear();
+                attributes.clear();
+                tagsWithAlias.clear();
+                attributesWithAlias.clear();
+              }
               pointSize.set(0);
             }
           } else {
@@ -218,7 +284,17 @@ private static void writeScheme(
             dataTypes.clear();
             encodings.clear();
             compressors.clear();
+            pathsWithAlias.clear();
+            dataTypesWithAlias.clear();
+            encodingsWithAlias.clear();
+            compressorsWithAlias.clear();
             measurementAlias.clear();
+            if (v2) {
+              tags.clear();
+              attributes.clear();
+              tagsWithAlias.clear();
+              attributesWithAlias.clear();
+            }
           }
           String path = recordObj.get(headerNames.indexOf(Constants.HEAD_COLUMNS.get(0)));
           String alias = recordObj.get(headerNames.indexOf(Constants.HEAD_COLUMNS.get(1)));
@@ -230,6 +306,23 @@ private static void writeScheme(
           String compressionTypeRaw =
               recordObj.get(headerNames.indexOf(Constants.HEAD_COLUMNS.get(4)));
           CompressionType compressionType = compressInfer(compressionTypeRaw);
+          Map<String, String> recordTags = null;
+          Map<String, String> recordAttributes = null;
+          boolean tagsParsed = true;
+          if (v2) {
+            try {
+              recordTags = parseJsonCell(recordObj.get(tagsIndex));
+            } catch (RuntimeException e) {
+              ioTPrinter.println(invalidJsonMessage(Constants.HEADER_TAGS, path, e));
+              tagsParsed = false;
+            }
+            try {
+              recordAttributes = parseJsonCell(recordObj.get(attributesIndex));
+            } catch (RuntimeException e) {
+              ioTPrinter.println(invalidJsonMessage(Constants.HEADER_ATTRIBUTES, path, e));
+              tagsParsed = false;
+            }
+          }
           if (StringUtils.isBlank(path)
               || path.trim().startsWith(SYSTEM_DATABASE)
               || path.trim().startsWith(AUDIT_DATABASE)) {
@@ -268,31 +361,56 @@ private static void writeScheme(
                     compressionTypeRaw));
             failedRecords.add(recordObj.stream().collect(Collectors.toList()));
             failed = true;
+          } else if (!tagsParsed) {
+            failedRecords.add(recordObj.stream().collect(Collectors.toList()));
+            failed = true;
           } else {
             if (StringUtils.isBlank(alias)) {
               paths.add(path);
               dataTypes.add(dataType);
               encodings.add(encodingType);
               compressors.add(compressionType);
+              if (v2) {
+                tags.add(recordTags);
+                attributes.add(recordAttributes);
+              }
             } else {
               pathsWithAlias.add(path);
               dataTypesWithAlias.add(dataType);
               encodingsWithAlias.add(encodingType);
               compressorsWithAlias.add(compressionType);
               measurementAlias.add(alias);
+              if (v2) {
+                tagsWithAlias.add(recordTags);
+                attributesWithAlias.add(recordAttributes);
+              }
             }
             pointSize.getAndIncrement();
           }
           if (!failed && aligned) {
             String deviceId = path.substring(0, path.lastIndexOf("."));
             paths.add(0, path.substring(deviceId.length() + 1));
+            if (v2) {
+              // keep the maps aligned with the measurements list, whose head now holds this one
+              tags.add(0, recordTags);
+              attributes.add(0, recordAttributes);
+            }
             writeAndEmptyDataSetAligned(
-                deviceId, paths, dataTypes, encodings, compressors, measurementAlias, 3);
+                deviceId,
+                paths,
+                dataTypes,
+                encodings,
+                compressors,
+                measurementAlias,
+                tags,
+                attributes,
+                3);
           }
         });
     try {
       if (CollectionUtils.isNotEmpty(paths)) {
-        writeAndEmptyDataSet(paths, dataTypes, encodings, compressors, null, null, null, null, 3);
+        writeAndEmptyDataSet(
+            paths, dataTypes, encodings, compressors, null, tags, attributes, null, 3);
       }
     } catch (Exception e) {
       paths.forEach(t -> failedRecords.add(Collections.singletonList(t)));
@@ -305,8 +423,8 @@ private static void writeScheme(
             encodingsWithAlias,
             compressorsWithAlias,
             null,
-            null,
-            null,
+            tagsWithAlias,
+            attributesWithAlias,
             measurementAlias,
             3);
       }
@@ -380,10 +498,24 @@ private static void writeAndEmptyDataSetAligned(
       List<TSEncoding> encodings,
       List<CompressionType> compressors,
       List<String> measurementAliasList,
+      List<Map<String, String>> tagsList,
+      List<Map<String, String>> attributesList,
       int retryTime) {
     try {
-      sessionPool.createAlignedTimeseries(
-          deviceId, measurements, dataTypes, encodings, compressors, measurementAliasList);
+      if (tagsList == null && attributesList == null) {
+        sessionPool.createAlignedTimeseries(
+            deviceId, measurements, dataTypes, encodings, compressors, measurementAliasList);
+      } else {
+        sessionPool.createAlignedTimeseries(
+            deviceId,
+            measurements,
+            dataTypes,
+            encodings,
+            compressors,
+            measurementAliasList,
+            tagsList,
+            attributesList);
+      }
     } catch (IoTDBConnectionException e) {
       if (retryTime > 0) {
         writeAndEmptyDataSetAligned(
@@ -393,6 +525,8 @@ private static void writeAndEmptyDataSetAligned(
             encodings,
             compressors,
             measurementAliasList,
+            tagsList,
+            attributesList,
             --retryTime);
       }
     } catch (StatementExecutionException e) {
@@ -405,9 +539,24 @@ private static void writeAndEmptyDataSetAligned(
       encodings.clear();
       compressors.clear();
       measurementAliasList.clear();
+      if (tagsList != null) {
+        tagsList.clear();
+      }
+      if (attributesList != null) {
+        attributesList.clear();
+      }
     }
   }
 
+  private static String invalidJsonMessage(String column, String path, Exception e) {
+    return String.format(
+        CliMessages
+            .MESSAGE_FAILED_TO_PARSE_THE_ARG_COLUMN_OF_TIMESERIES_ARG_AS_JSON_THE_RECORD_IS_SKIPPED_ARG_35D898C8,
+        column,
+        path,
+        String.valueOf(e.getMessage()));
+  }
+
   private static TSDataType typeInfer(String typeStr) {
     try {
       if (StringUtils.isNotBlank(typeStr)) {
diff --git a/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/schema/SchemaTagsFidelityTest.java b/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/schema/SchemaTagsFidelityTest.java
new file mode 100644
index 0000000000000..85fd629ddbbbb
--- /dev/null
+++ b/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/schema/SchemaTagsFidelityTest.java
@@ -0,0 +1,318 @@
+/*
+ * 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.tool.schema;
+
+import org.apache.iotdb.tool.common.Constants;
+
+import org.apache.commons.csv.CSVParser;
+import org.apache.commons.csv.CSVRecord;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.common.Field;
+import org.apache.tsfile.utils.Binary;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+/** Guards the lossless Tags / Attributes round trip of the tree model schema tools. */
+public class SchemaTagsFidelityTest {
+
+  @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+  @Test
+  public void jsonCellSortsKeysAndKeepsPrintableCharactersReadable() {
+    Map<String, String> tags = new LinkedHashMap<>();
+    tags.put("type", "sensor");
+    tags.put("room", "101");
+    tags.put("a-b", "it's = <v> & w");
+
+    assertEquals(
+        "{\"a-b\":\"it's = <v> & w\",\"room\":\"101\",\"type\":\"sensor\"}",
+        AbstractSchemaTool.toJsonCell(tags));
+  }
+
+  @Test
+  public void jsonCellEscapesOnlyWhatJsonRequires() {
+    Map<String, String> tags = new LinkedHashMap<>();
+    tags.put("quote", "a\"b");
+    tags.put("backslash", "a\\b");
+    tags.put("tab", "a\tb");
+    tags.put("newline", "a\nb");
+    tags.put("comma", "a,b");
+    tags.put("cjk", "中文😀");
+    tags.put("backtick", "`x`");
+    tags.put("null", "null");
+    tags.put("empty", "");
+
+    assertEquals(
+        "{\"backslash\":\"a\\\\b\",\"backtick\":\"`x`\",\"cjk\":\"中文😀\",\"comma\":\"a,b\","
+            + "\"empty\":\"\",\"newline\":\"a\\nb\",\"null\":\"null\",\"quote\":\"a\\\"b\","
+            + "\"tab\":\"a\\tb\"}",
+        AbstractSchemaTool.toJsonCell(tags));
+  }
+
+  @Test
+  public void jsonCellOfMissingValuesIsAnEmptyCell() {
+    assertEquals("", AbstractSchemaTool.toJsonCell(null));
+    assertEquals("", AbstractSchemaTool.toJsonCell(new LinkedHashMap<>()));
+  }
+
+  @Test
+  public void jsonCellRoundTripsEveryCharacter() {
+    Map<String, String> tags = new LinkedHashMap<>();
+    tags.put("quote", "a\"b");
+    tags.put("backslash", "a\\b");
+    tags.put("tab", "a\tb");
+    tags.put("newline", "a\nb");
+    tags.put("comma", "a,b");
+    tags.put("cjk", "中文😀");
+    tags.put("backtick", "`x`");
+    tags.put("null", "null");
+    tags.put("empty", "");
+    tags.put("key with space", "[  C  ]");
+
+    assertEquals(tags, AbstractSchemaTool.parseJsonCell(AbstractSchemaTool.toJsonCell(tags)));
+  }
+
+  @Test
+  public void jsonCellOfNothingParsesToNoValue() {
+    assertNull(AbstractSchemaTool.parseJsonCell(null));
+    assertNull(AbstractSchemaTool.parseJsonCell(""));
+    assertNull(AbstractSchemaTool.parseJsonCell("   "));
+    assertNull(AbstractSchemaTool.parseJsonCell("null"));
+    assertNull(AbstractSchemaTool.parseJsonCell("{}"));
+  }
+
+  @Test
+  public void invalidJsonCellIsRejected() {
+    for (String cell : Arrays.asList("not json", "[1,2]", "\"abc\"", "{\"k\":\"v\"} trailing")) {
+      try {
+        AbstractSchemaTool.parseJsonCell(cell);
+        throw new AssertionError("expected a parse failure for " + cell);
+      } catch (RuntimeException expected) {
+        // the export reports this and skips the record
+      }
+    }
+  }
+
+  @Test
+  public void csvRoundTripKeepsEveryFieldByteIdentical() throws IOException {
+    Map<String, String> tags = new LinkedHashMap<>();
+    tags.put("quote", "a\"b");
+    tags.put("backslash", "a\\b");
+    tags.put("comma", "a,b");
+    tags.put("tab", "a\tb");
+    tags.put("newline", "a\nb");
+    Map<String, String> attributes = new LinkedHashMap<>();
+    attributes.put("owner", "team a");
+    attributes.put("note", "q\"w\\e,r\ty\nz");
+
+    List<String> row =
+        Arrays.asList(
+            "root.t.d1.s1",
+            "aliasA",
+            "INT64",
+            "TS_2DIFF",
+            "SNAPPY",
+            AbstractSchemaTool.toJsonCell(tags),
+            AbstractSchemaTool.toJsonCell(attributes));
+    File csv = writeV2File("roundtrip.csv", row);
+
+    CSVParser parser = ImportSchemaTree.readCsvFile(csv.getAbsolutePath());
+    List<CSVRecord> records = parser.getRecords();
+    assertEquals(1, records.size());
+    CSVRecord record = records.get(0);
+    for (int i = 0; i < row.size(); i++) {
+      assertEquals(row.get(i), record.get(Constants.HEAD_COLUMNS_V2.get(i)));
+    }
+    assertEquals(tags, AbstractSchemaTool.parseJsonCell(record.get(Constants.HEADER_TAGS)));
+    assertEquals(
+        attributes, AbstractSchemaTool.parseJsonCell(record.get(Constants.HEADER_ATTRIBUTES)));
+  }
+
+  @Test
+  public void emptyTagsAndAttributesAreWrittenAsEmptyCells() throws IOException {
+    File csv =
+        writeV2File(
+            "empty.csv", Arrays.asList("root.t.d1.s1", "", "INT64", "TS_2DIFF", "SNAPPY", "", ""));
+
+    assertEquals(
+        "Timeseries,Alias,DataType,Encoding,Compression,Tags,Attributes\n"
+            + "root.t.d1.s1,,INT64,TS_2DIFF,SNAPPY,,\n",
+        new String(Files.readAllBytes(csv.toPath()), StandardCharsets.UTF_8));
+    CSVRecord record = ImportSchemaTree.readCsvFile(csv.getAbsolutePath()).getRecords().get(0);
+    assertNull(AbstractSchemaTool.parseJsonCell(record.get(Constants.HEADER_TAGS)));
+    assertNull(AbstractSchemaTool.parseJsonCell(record.get(Constants.HEADER_ATTRIBUTES)));
+  }
+
+  @Test
+  public void jsonCellsAreRfc4180Quoted() throws IOException {
+    Map<String, String> tags = new LinkedHashMap<>();
+    tags.put("room", "101");
+    tags.put("type", "sensor");
+    File csv =
+        writeV2File(
+            "quoted.csv",
+            Arrays.asList(
+                "root.t.d1.s1",
+                "",
+                "INT64",
+                "TS_2DIFF",
+                "SNAPPY",
+                AbstractSchemaTool.toJsonCell(tags),
+                ""));
+
+    assertEquals(
+        "Timeseries,Alias,DataType,Encoding,Compression,Tags,Attributes\n"
+            + "root.t.d1.s1,,INT64,TS_2DIFF,SNAPPY,\"{\"\"room\"\":\"\"101\"\","
+            + "\"\"type\"\":\"\"sensor\"\"}\",\n",
+        new String(Files.readAllBytes(csv.toPath()), StandardCharsets.UTF_8));
+  }
+
+  @Test
+  public void legacyFiveColumnHeaderSelectsTheLegacyPath() {
+    assertFalse(ImportSchemaTree.isV2Format(Constants.HEAD_COLUMNS));
+    assertTrue(ImportSchemaTree.isV2Format(Constants.HEAD_COLUMNS_V2));
+  }
+
+  @Test
+  public void quotedHeaderStillSelectsTheV2Path() throws IOException {
+    File csv = temporaryFolder.newFile("quoted_header.csv");
+    Files.write(
+        csv.toPath(),
+        ("\"Timeseries\",\"Alias\",\"DataType\",\"Encoding\",\"Compression\",\"Tags\","
+                + "\"Attributes\"\nroot.t.d1.s1,,INT64,TS_2DIFF,SNAPPY,,\n")
+            .getBytes(StandardCharsets.UTF_8));
+
+    assertTrue(
+        ImportSchemaTree.isV2Format(ImportSchemaTree.readHeaderNames(csv.getAbsolutePath())));
+  }
+
+  @Test
+  public void legacyFileKeepsItsBackslashEscapes() throws IOException {
+    File csv = temporaryFolder.newFile("legacy.csv");
+    Files.write(
+        csv.toPath(),
+        ("Timeseries,Alias,DataType,Encoding,Compression\n"
+                + "root.sg.d.s1,a\\,b,INT64,TS_2DIFF,SNAPPY\n")
+            .getBytes(StandardCharsets.UTF_8));
+
+    CSVRecord record = ImportSchemaTree.readCsvFile(csv.getAbsolutePath()).getRecords().get(0);
+    assertEquals(5, record.size());
+    assertEquals("a,b", record.get(Constants.HEADER_ALIAS));
+  }
+
+  @Test
+  public void exportedRecordCarriesTagsAndAttributes() throws IOException {
+    File csv = temporaryFolder.newFile("export.csv");
+    AbstractSchemaTool.CSVPrinterWrapper printer =
+        new AbstractSchemaTool.CSVPrinterWrapper(csv.getAbsolutePath());
+    printer.printRecord(Constants.HEAD_COLUMNS_V2);
+
+    List<String> headers =
+        Arrays.asList(
+            Constants.HEADER_TIMESERIES,
+            Constants.HEADER_ALIAS,
+            "Database",
+            Constants.HEADER_DATATYPE,
+            Constants.HEADER_ENCODING,
+            Constants.HEADER_COMPRESSION,
+            Constants.HEADER_TAGS,
+            Constants.HEADER_ATTRIBUTES,
+            Constants.HEADER_VIEW_TYPE);
+    List<Field> fields =
+        Arrays.asList(
+            text("root.t.d1.s1"),
+            emptyField(),
+            text("root.t"),
+            text("INT64"),
+            text("TS_2DIFF"),
+            text("SNAPPY"),
+            text("{\"room\": \"101\", \"type\": \"sensor\"}"),
+            emptyField(),
+            text(Constants.BASE_VIEW_TYPE));
+    assertTrue(ExportSchemaTree.printRecord(printer, headers, fields, "root.t.d1.s1"));
+    printer.println();
+    printer.close();
+
+    CSVRecord record = ImportSchemaTree.readCsvFile(csv.getAbsolutePath()).getRecords().get(0);
+    assertEquals("root.t.d1.s1", record.get(Constants.HEADER_TIMESERIES));
+    assertEquals("", record.get(Constants.HEADER_ALIAS));
+    assertEquals("INT64", record.get(Constants.HEADER_DATATYPE));
+    assertEquals("TS_2DIFF", record.get(Constants.HEADER_ENCODING));
+    assertEquals("SNAPPY", record.get(Constants.HEADER_COMPRESSION));
+    assertEquals("{\"room\":\"101\",\"type\":\"sensor\"}", record.get(Constants.HEADER_TAGS));
+    assertEquals("", record.get(Constants.HEADER_ATTRIBUTES));
+  }
+
+  @Test
+  public void exportedRecordWithBrokenJsonIsSkipped() throws IOException {
+    File csv = temporaryFolder.newFile("broken.csv");
+    AbstractSchemaTool.CSVPrinterWrapper printer =
+        new AbstractSchemaTool.CSVPrinterWrapper(csv.getAbsolutePath());
+    List<String> headers =
+        Arrays.asList(
+            Constants.HEADER_TIMESERIES, Constants.HEADER_TAGS, Constants.HEADER_ATTRIBUTES);
+    List<Field> fields = Arrays.asList(text("root.t.d1.s1"), text("not json"), emptyField());
+
+    assertFalse(ExportSchemaTree.printRecord(printer, headers, fields, "root.t.d1.s1"));
+    printer.close();
+    assertEquals(0, csv.length());
+  }
+
+  /**
+   * Writes a v2 file with the writer of the schema tools so that the reader is exercised against
+   * the bytes the export really produces.
+   */
+  private File writeV2File(String name, List<String> row) throws IOException {
+    File csv = temporaryFolder.newFile(name);
+    AbstractSchemaTool.CSVPrinterWrapper printer =
+        new AbstractSchemaTool.CSVPrinterWrapper(csv.getAbsolutePath());
+    printer.printRecord(Constants.HEAD_COLUMNS_V2);
+    row.forEach(printer::print);
+    printer.println();
+    printer.close();
+    return csv;
+  }
+
+  private static Field text(String value) {
+    Field field = new Field(TSDataType.TEXT);
+    field.setBinaryV(new Binary(value, StandardCharsets.UTF_8));
+    return field;
+  }
+
+  /** The shape SHOW TIMESERIES uses for a cell without a value. */
+  private static Field emptyField() {
+    return new Field(null);
+  }
+}
diff --git a/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/schema/SchemaTagsImportTest.java b/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/schema/SchemaTagsImportTest.java
new file mode 100644
index 0000000000000..4b4065c8ccf02
--- /dev/null
+++ b/iotdb-client/cli/src/test/java/org/apache/iotdb/tool/schema/SchemaTagsImportTest.java
@@ -0,0 +1,288 @@
+/*
+ * 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.tool.schema;
+
+import org.apache.iotdb.session.pool.SessionPool;
+import org.apache.iotdb.tool.common.Constants;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.external.commons.collections4.CollectionUtils;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.junit.After;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+/** Checks that the import passes the parsed tags and attributes on to the client API. */
+public class SchemaTagsImportTest {
+
+  @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+  @After
+  public void reset() {
+    ImportSchemaTree.sessionPool = null;
+    ImportSchemaTree.aligned = false;
+    ImportSchemaTree.batchPointSize = Constants.BATCH_POINT_SIZE;
+  }
+
+  @Test
+  public void legacyFileKeepsTheLegacyCreateCall() throws Exception {
+    SessionPool pool = mock(SessionPool.class);
+    ImportSchemaTree.sessionPool = pool;
+
+    runImport(
+        "Timeseries,Alias,DataType,Encoding,Compression\n"
+            + "root.sg.d.s1,,INT64,TS_2DIFF,SNAPPY\n"
+            + "root.sg.d.s2,aliasB,INT64,TS_2DIFF,SNAPPY\n");
+
+    verify(pool)
+        .createMultiTimeseries(
+            Arrays.asList("root.sg.d.s1"),
+            Arrays.asList(TSDataType.INT64),
+            Arrays.asList(TSEncoding.TS_2DIFF),
+            Arrays.asList(CompressionType.SNAPPY),
+            null,
+            null,
+            null,
+            null);
+    verify(pool)
+        .createMultiTimeseries(
+            Arrays.asList("root.sg.d.s2"),
+            Arrays.asList(TSDataType.INT64),
+            Arrays.asList(TSEncoding.TS_2DIFF),
+            Arrays.asList(CompressionType.SNAPPY),
+            null,
+            null,
+            null,
+            Arrays.asList("aliasB"));
+  }
+
+  @Test
+  public void v2FilePassesTagsAndAttributesToCreateMultiTimeseries() throws Exception {
+    SessionPool pool = mock(SessionPool.class);
+    ImportSchemaTree.sessionPool = pool;
+    Map<String, String> tags = new LinkedHashMap<>();
+    tags.put("room", "101");
+    tags.put("type", "sensor");
+    Map<String, String> attributes = new LinkedHashMap<>();
+    attributes.put("owner", "team a");
+
+    runImport(
+        Constants.HEAD_COLUMNS_V2.stream().collect(Collectors.joining(","))
+            + "\nroot.sg.d.s1,,INT64,TS_2DIFF,SNAPPY,\""
+            + AbstractSchemaTool.toJsonCell(tags).replace("\"", "\"\"")
+            + "\",\""
+            + AbstractSchemaTool.toJsonCell(attributes).replace("\"", "\"\"")
+            + "\"\n"
+            + "root.sg.d.s2,,INT64,TS_2DIFF,SNAPPY,,\n");
+
+    verify(pool)
+        .createMultiTimeseries(
+            Arrays.asList("root.sg.d.s1", "root.sg.d.s2"),
+            Arrays.asList(TSDataType.INT64, TSDataType.INT64),
+            Arrays.asList(TSEncoding.TS_2DIFF, TSEncoding.TS_2DIFF),
+            Arrays.asList(CompressionType.SNAPPY, CompressionType.SNAPPY),
+            null,
+            Arrays.asList(new LinkedHashMap<>(tags), null),
+            Arrays.asList(new LinkedHashMap<>(attributes), null),
+            null);
+  }
+
+  @Test
+  public void v2AlignedFilePassesTagsAndAttributesToCreateAlignedTimeseries() throws Exception {
+    SessionPool pool = mock(SessionPool.class);
+    ImportSchemaTree.sessionPool = pool;
+    ImportSchemaTree.aligned = true;
+    Map<String, String> tags = new LinkedHashMap<>();
+    tags.put("room", "101");
+    Map<String, String> attributes = new LinkedHashMap<>();
+    attributes.put("owner", "team a");
+    // the aligned writer empties its lists in a finally block, so snapshot the arguments
+    List<Object> alignedCall = new ArrayList<>();
+    doAnswer(
+            call -> {
+              for (Object argument : call.getArguments()) {
+                alignedCall.add(
+                    argument instanceof List ? new ArrayList<>((List<?>) argument) : argument);
+              }
+              return null;
+            })
+        .when(pool)
+        .createAlignedTimeseries(any(), any(), any(), any(), any(), any(), any(), any());
+
+    runImport(
+        Constants.HEAD_COLUMNS_V2.stream().collect(Collectors.joining(","))
+            + "\nroot.sg.d.s1,,INT64,TS_2DIFF,SNAPPY,\""
+            + AbstractSchemaTool.toJsonCell(tags).replace("\"", "\"\"")
+            + "\",\""
+            + AbstractSchemaTool.toJsonCell(attributes).replace("\"", "\"\"")
+            + "\"\n");
+
+    // the pre-existing aligned path hands over both the measurement and the full path, so the maps
+    // must be built with the same shape to stay positionally aligned
+    assertEquals(
+        Arrays.asList(
+            "root.sg.d",
+            Arrays.asList("s1", "root.sg.d.s1"),
+            Arrays.asList(TSDataType.INT64),
+            Arrays.asList(TSEncoding.TS_2DIFF),
+            Arrays.asList(CompressionType.SNAPPY),
+            new ArrayList<String>(),
+            Arrays.asList(new LinkedHashMap<>(tags), new LinkedHashMap<>(tags)),
+            Arrays.asList(new LinkedHashMap<>(attributes), new LinkedHashMap<>(attributes))),
+        alignedCall);
+    verify(pool, never())
+        .createMultiTimeseries(any(), any(), any(), any(), any(), any(), any(), any());
+  }
+
+  @Test
+  public void v2HeaderWithoutAllColumnsCreatesNothing() throws Exception {
+    SessionPool pool = mock(SessionPool.class);
+    ImportSchemaTree.sessionPool = pool;
+    assertFalse(
+        ImportSchemaTree.checkHeader(
+            Arrays.asList("Timeseries", "Alias", "DataType", "Encoding", "Compression", "Tags")));
+
+    verify(pool, never())
+        .createMultiTimeseries(any(), any(), any(), any(), any(), any(), any(), any());
+  }
+
+  @Test
+  public void brokenJsonCellSkipsTheRecord() throws Exception {
+    SessionPool pool = mock(SessionPool.class);
+    ImportSchemaTree.sessionPool = pool;
+
+    runImport(
+        Constants.HEAD_COLUMNS_V2.stream().collect(Collectors.joining(","))
+            + "\nroot.sg.d.s1,,INT64,TS_2DIFF,SNAPPY,\"{not json}\",\n");
+
+    verify(pool, never())
+        .createMultiTimeseries(any(), any(), any(), any(), any(), any(), any(), any());
+  }
+
+  /**
+   * Guards the batch-flush boundary: every aliased list must be emptied together, otherwise the
+   * next batch pairs the aliases still buffered with the wrong paths.
+   */
+  @Test
+  public void batchedAliasedRecordsKeepAliasTagsAndPathsPaired() throws Exception {
+    SessionPool pool = mock(SessionPool.class);
+    ImportSchemaTree.sessionPool = pool;
+    ImportSchemaTree.batchPointSize = 2;
+    List<List<List<?>>> calls = new ArrayList<>();
+    doAnswer(
+            call -> {
+              List<List<?>> snapshot = new ArrayList<>();
+              for (int i = 0; i < call.getArguments().length; i++) {
+                List<?> argument = call.getArgument(i);
+                snapshot.add(argument == null ? null : new ArrayList<>(argument));
+              }
+              calls.add(snapshot);
+              return null;
+            })
+        .when(pool)
+        .createMultiTimeseries(any(), any(), any(), any(), any(), any(), any(), any());
+
+    File csv = temporaryFolder.newFile("batch.csv");
+    AbstractSchemaTool.CSVPrinterWrapper printer =
+        new AbstractSchemaTool.CSVPrinterWrapper(csv.getAbsolutePath());
+    printer.printRecord(Constants.HEAD_COLUMNS_V2);
+    for (int i = 1; i <= 5; i++) {
+      printer.print("root.sg.d.s" + i);
+      printer.print("alias" + i);
+      printer.print("INT64");
+      printer.print("TS_2DIFF");
+      printer.print("SNAPPY");
+      printer.print(AbstractSchemaTool.toJsonCell(Collections.singletonMap("k", "v" + i)));
+      printer.print("");
+      printer.println();
+    }
+    printer.close();
+    runImportCsv(csv);
+
+    // 3 flushes with batchPointSize 2: [s1,s2], [s3,s4], [s5]
+    assertEquals(3, calls.size());
+    List<String> sentPaths = new ArrayList<>();
+    for (List<List<?>> call : calls) {
+      List<?> paths = call.get(0);
+      List<?> dataTypes = call.get(1);
+      List<?> encodings = call.get(2);
+      List<?> compressors = call.get(3);
+      List<?> tags = call.get(5);
+      List<?> attributes = call.get(6);
+      List<?> measurementAlias = call.get(7);
+      assertTrue(CollectionUtils.isNotEmpty(paths));
+      // every list that is indexed by the path position must have the same length
+      for (List<?> parallel :
+          Arrays.asList(dataTypes, encodings, compressors, tags, attributes, measurementAlias)) {
+        assertEquals("lists must stay positionally aligned", paths.size(), parallel.size());
+      }
+      for (int i = 0; i < paths.size(); i++) {
+        String path = (String) paths.get(i);
+        String seq = path.substring(path.length() - 1);
+        sentPaths.add(path);
+        // s<n> must travel with alias<n> and with the tag built from the same n
+        assertEquals(path, "alias" + seq, measurementAlias.get(i));
+        assertEquals("v" + seq, ((Map<?, ?>) tags.get(i)).get("k"));
+      }
+    }
+    assertEquals(
+        Arrays.asList(
+            "root.sg.d.s1", "root.sg.d.s2", "root.sg.d.s3", "root.sg.d.s4", "root.sg.d.s5"),
+        sentPaths);
+  }
+
+  private void runImport(String csvContent) throws Exception {
+    File csv = temporaryFolder.newFile("schema_" + System.nanoTime() + ".csv");
+    Files.write(csv.toPath(), csvContent.getBytes(StandardCharsets.UTF_8));
+    runImportCsv(csv);
+  }
+
+  private void runImportCsv(File csv) throws Exception {
+    String failedFilePath = temporaryFolder.getRoot().getAbsolutePath() + "/failed";
+    try (Stream<org.apache.commons.csv.CSVRecord> records =
+        ImportSchemaTree.readCsvFile(csv.getAbsolutePath()).stream()) {
+      List<String> headerNames = ImportSchemaTree.readHeaderNames(csv.getAbsolutePath());
+      ImportSchemaTree.writeScheme(csv.getName(), headerNames, records, failedFilePath);
+    }
+  }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/LogicalViewSchemaSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/LogicalViewSchemaSource.java
index 144ec16d4a37e..6d2a6f061b864 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/LogicalViewSchemaSource.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/LogicalViewSchemaSource.java
@@ -38,8 +38,8 @@
 
 import java.util.Collections;
 import java.util.List;
-import java.util.Map;
-import java.util.stream.Collectors;
+
+import static org.apache.iotdb.db.queryengine.execution.operator.schema.source.TimeSeriesSchemaSource.mapToString;
 
 public class LogicalViewSchemaSource implements ISchemaSource<ITimeSeriesSchemaInfo> {
 
@@ -118,15 +118,4 @@ public boolean hasSchemaStatistic(ISchemaRegion schemaRegion) {
   public long getSchemaStatistic(ISchemaRegion schemaRegion) {
     return 0;
   }
-
-  private String mapToString(Map<String, String> map) {
-    if (map == null || map.isEmpty()) {
-      return null;
-    }
-    String content =
-        map.entrySet().stream()
-            .map(e -> "\"" + e.getKey() + "\"" + ":" + "\"" + e.getValue() + "\"")
-            .collect(Collectors.joining(","));
-    return "{" + content + "}";
-  }
 }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/TimeSeriesSchemaSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/TimeSeriesSchemaSource.java
index 0b89998052b00..832cf46100def 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/TimeSeriesSchemaSource.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/TimeSeriesSchemaSource.java
@@ -42,7 +42,6 @@
 
 import java.util.List;
 import java.util.Map;
-import java.util.stream.Collectors;
 
 import static org.apache.iotdb.commons.schema.SchemaConstant.ALL_MATCH_PATTERN;
 
@@ -161,14 +160,68 @@ public boolean shouldSkipSchemaRegion(final ISchemaRegion schemaRegion) {
     return false;
   }
 
+  /**
+   * Renders a tag/attribute map as a JSON object (RFC 8259). Keys and values are escaped as JSON
+   * strings, while non-ASCII characters are kept raw for readability.
+   *
+   * @return the JSON object text, or {@code null} when the map is null or empty
+   */
   public static String mapToString(Map<String, String> map) {
     if (map == null || map.isEmpty()) {
       return null;
     }
-    String content =
-        map.entrySet().stream()
-            .map(e -> "\"" + e.getKey() + "\"" + ":" + "\"" + e.getValue() + "\"")
-            .collect(Collectors.joining(","));
-    return "{" + content + "}";
+    final StringBuilder builder = new StringBuilder("{");
+    boolean first = true;
+    for (final Map.Entry<String, String> entry : map.entrySet()) {
+      if (!first) {
+        builder.append(',');
+      }
+      first = false;
+      appendJsonString(builder, entry.getKey());
+      builder.append(':');
+      appendJsonString(builder, entry.getValue());
+    }
+    return builder.append('}').toString();
+  }
+
+  private static final char[] HEX_DIGITS = "0123456789abcdef".toCharArray();
+
+  private static void appendJsonString(final StringBuilder builder, final String raw) {
+    // a null key or value is rendered as the literal text "null", as before
+    final String value = raw == null ? "null" : raw;
+    builder.append('"');
+    for (int i = 0, length = value.length(); i < length; i++) {
+      final char c = value.charAt(i);
+      switch (c) {
+        case '"':
+          builder.append("\\\"");
+          break;
+        case '\\':
+          builder.append("\\\\");
+          break;
+        case '\b':
+          builder.append("\\b");
+          break;
+        case '\f':
+          builder.append("\\f");
+          break;
+        case '\n':
+          builder.append("\\n");
+          break;
+        case '\r':
+          builder.append("\\r");
+          break;
+        case '\t':
+          builder.append("\\t");
+          break;
+        default:
+          if (c < 0x20) {
+            builder.append("\\u00").append(HEX_DIGITS[(c >> 4) & 0xF]).append(HEX_DIGITS[c & 0xF]);
+          } else {
+            builder.append(c);
+          }
+      }
+    }
+    builder.append('"');
   }
 }
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/TimeSeriesSchemaSourceMapToStringTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/TimeSeriesSchemaSourceMapToStringTest.java
new file mode 100644
index 0000000000000..1fa137d5d5a8c
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/TimeSeriesSchemaSourceMapToStringTest.java
@@ -0,0 +1,221 @@
+/*
+ * 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.execution.operator.schema.source;
+
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.iotdb.db.queryengine.execution.operator.schema.source.TimeSeriesSchemaSource.mapToString;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+public class TimeSeriesSchemaSourceMapToStringTest {
+
+  /** The pre-escaping implementation, kept here to prove backward compatibility. */
+  private static String legacyMapToString(final Map<String, String> map) {
+    if (map == null || map.isEmpty()) {
+      return null;
+    }
+    final String content =
+        map.entrySet().stream()
+            .map(e -> "\"" + e.getKey() + "\"" + ":" + "\"" + e.getValue() + "\"")
+            .collect(Collectors.joining(","));
+    return "{" + content + "}";
+  }
+
+  private static Map<String, String> mapOf(final String key, final String value) {
+    final Map<String, String> map = new LinkedHashMap<>();
+    map.put(key, value);
+    return map;
+  }
+
+  private static JsonObject parse(final String json) {
+    final JsonElement element = JsonParser.parseString(json);
+    assertEquals("not a JSON object: " + json, true, element.isJsonObject());
+    return element.getAsJsonObject();
+  }
+
+  /** Renders, parses with a standard JSON parser, and checks the round-trip to the source map. */
+  private static void assertJsonRoundTrip(final Map<String, String> map) {
+    final JsonObject json = parse(mapToString(map));
+    assertEquals(map.size(), json.size());
+    for (final Map.Entry<String, String> entry : map.entrySet()) {
+      assertEquals(true, json.has(entry.getKey()));
+      assertEquals(entry.getValue(), json.get(entry.getKey()).getAsString());
+    }
+  }
+
+  @Test
+  public void testPlainValue() {
+    final Map<String, String> map = mapOf("room", "101");
+    assertEquals("{\"room\":\"101\"}", mapToString(map));
+    assertJsonRoundTrip(map);
+  }
+
+  @Test
+  public void testCommaIsNotEscaped() {
+    final Map<String, String> map = mapOf("k", "a,b");
+    assertEquals("{\"k\":\"a,b\"}", mapToString(map));
+    assertJsonRoundTrip(map);
+  }
+
+  @Test
+  public void testDoubleQuote() {
+    final Map<String, String> map = mapOf("k", "a\"b");
+    assertEquals("{\"k\":\"a\\\"b\"}", mapToString(map));
+    assertJsonRoundTrip(map);
+    // the escaped quote must not split the document into extra keys
+    assertEquals(1, parse(mapToString(map)).size());
+  }
+
+  @Test
+  public void testBackslash() {
+    final Map<String, String> map = mapOf("k", "a\\b");
+    assertEquals("{\"k\":\"a\\\\b\"}", mapToString(map));
+    assertJsonRoundTrip(map);
+  }
+
+  @Test
+  public void testLiteralBackslashAndN() {
+    final Map<String, String> map = mapOf("k", "a\\nb");
+    assertEquals("{\"k\":\"a\\\\nb\"}", mapToString(map));
+    assertJsonRoundTrip(map);
+    assertEquals("a\\nb", parse(mapToString(map)).get("k").getAsString());
+  }
+
+  @Test
+  public void testDoubleBackslash() {
+    final Map<String, String> map = mapOf("k", "a\\\\b");
+    assertEquals("{\"k\":\"a\\\\\\\\b\"}", mapToString(map));
+    assertJsonRoundTrip(map);
+    assertEquals("a\\\\b", parse(mapToString(map)).get("k").getAsString());
+  }
+
+  @Test
+  public void testTab() {
+    final Map<String, String> map = mapOf("k", "\t");
+    assertEquals("{\"k\":\"\\t\"}", mapToString(map));
+    assertJsonRoundTrip(map);
+  }
+
+  @Test
+  public void testLineFeed() {
+    final Map<String, String> map = mapOf("k", "a\nb");
+    assertEquals("{\"k\":\"a\\nb\"}", mapToString(map));
+    assertJsonRoundTrip(map);
+  }
+
+  @Test
+  public void testCarriageReturnAndLineFeed() {
+    final Map<String, String> map = mapOf("k", "a\r\nb");
+    assertEquals("{\"k\":\"a\\r\\nb\"}", mapToString(map));
+    assertJsonRoundTrip(map);
+  }
+
+  @Test
+  public void testBackspaceAndFormFeed() {
+    final Map<String, String> map = mapOf("k", "a\bb\fc");
+    assertEquals("{\"k\":\"a\\bb\\fc\"}", mapToString(map));
+    assertJsonRoundTrip(map);
+  }
+
+  @Test
+  public void testOtherControlCharacterUsesUnicodeEscape() {
+    final Map<String, String> map = mapOf("k", "a\u0001b");
+    assertEquals("{\"k\":\"a\\u0001b\"}", mapToString(map));
+    assertJsonRoundTrip(map);
+    assertEquals("a\u0001b", parse(mapToString(map)).get("k").getAsString());
+  }
+
+  @Test
+  public void testNonAsciiIsKeptRaw() {
+    final Map<String, String> map = mapOf("cn", "中文\uD83D\uDE00");
+    assertEquals("{\"cn\":\"中文\uD83D\uDE00\"}", mapToString(map));
+    assertJsonRoundTrip(map);
+    // no \\uXXXX escaping of non-ASCII characters
+    assertEquals(false, mapToString(map).contains("\\u"));
+  }
+
+  @Test
+  public void testEmptyStringValue() {
+    final Map<String, String> map = mapOf("k", "");
+    assertEquals("{\"k\":\"\"}", mapToString(map));
+    assertJsonRoundTrip(map);
+  }
+
+  @Test
+  public void testEscapingInKey() {
+    final Map<String, String> map = mapOf("a\"b\\c\td", "v");
+    assertEquals("{\"a\\\"b\\\\c\\td\":\"v\"}", mapToString(map));
+    assertJsonRoundTrip(map);
+  }
+
+  @Test
+  public void testMultipleKeys() {
+    final Map<String, String> map = new LinkedHashMap<>();
+    map.put("root.sg", "101");
+    map.put("unit", "celsius");
+    map.put("note", "a\"b\\c");
+    assertEquals(
+        "{\"root.sg\":\"101\",\"unit\":\"celsius\",\"note\":\"a\\\"b\\\\c\"}", mapToString(map));
+    assertJsonRoundTrip(map);
+  }
+
+  @Test
+  public void testNullAndEmptyMap() {
+    assertNull(mapToString(null));
+    assertNull(mapToString(Collections.emptyMap()));
+  }
+
+  @Test
+  public void testIdenticalToLegacyOutputWhenNothingNeedsEscaping() {
+    final Map<String, String> plain = new LinkedHashMap<>();
+    plain.put("room", "101");
+    plain.put("sensor", "temperature");
+    plain.put("path", "root.sg.d1.s1");
+    plain.put("empty", "");
+    plain.put("punctuation", "a,b;c=d (e) [f] {g} /h@i#j$k%l^m&n*o-p+q_r");
+    plain.put("cjk", "中文😀");
+    plain.put("single", "x");
+
+    assertEquals(legacyMapToString(plain), mapToString(plain));
+    assertJsonRoundTrip(plain);
+
+    final Map<String, String> single = mapOf("room", "101");
+    assertEquals(legacyMapToString(single), mapToString(single));
+  }
+
+  @Test
+  public void testLegacyComparisonForMultiplePlainMaps() {
+    final Map<String, String> map = new LinkedHashMap<>();
+    map.put("a", "1");
+    map.put("b", "2");
+    map.put("c", "3");
+    assertEquals(legacyMapToString(map), mapToString(map));
+    assertEquals("{\"a\":\"1\",\"b\":\"2\",\"c\":\"3\"}", mapToString(map));
+  }
+}