diff --git a/data-loader/cli/src/main/java/com/scalar/db/dataloader/cli/command/dataexport/ExportCommand.java b/data-loader/cli/src/main/java/com/scalar/db/dataloader/cli/command/dataexport/ExportCommand.java
index 1a72319fc3..8331c45f87 100755
--- a/data-loader/cli/src/main/java/com/scalar/db/dataloader/cli/command/dataexport/ExportCommand.java
+++ b/data-loader/cli/src/main/java/com/scalar/db/dataloader/cli/command/dataexport/ExportCommand.java
@@ -55,6 +55,7 @@ public class ExportCommand extends ExportCommandOptions implements Callable<Inte
   public Integer call() throws Exception {
     validateDeprecatedOptions();
     applyDeprecatedOptions();
+    warnAboutIgnoredDeprecatedOptions();
     String scalarDbPropertiesFilePath = getScalarDbPropertiesFilePath();
 
     try {
@@ -141,6 +142,28 @@ private void validateDeprecatedOptions() {
         MAX_THREADS_OPTION_SHORT);
   }
 
+  /** Warns about deprecated options that are no longer used and have been completely ignored. */
+  private void warnAboutIgnoredDeprecatedOptions() {
+    CommandLine.ParseResult parseResult = spec.commandLine().getParseResult();
+    boolean hasIncludeMetadata =
+        parseResult.hasMatchedOption(DEPRECATED_INCLUDE_METADATA_OPTION)
+            || parseResult.hasMatchedOption(DEPRECATED_INCLUDE_METADATA_OPTION_SHORT);
+
+    if (hasIncludeMetadata) {
+      // Use picocli's ANSI support for colored warning output
+      CommandLine.Help.Ansi ansi = CommandLine.Help.Ansi.AUTO;
+      String warning =
+          ansi.string(
+              "@|bold,yellow The "
+                  + DEPRECATED_INCLUDE_METADATA_OPTION
+                  + " option is deprecated and no longer has any effect. "
+                  + "Use the 'scalar.db.consensus_commit.include_metadata.enabled' configuration property "
+                  + "in your ScalarDB properties file to control whether transaction metadata is included in scan operations.|@");
+
+      logger.warn(warning);
+    }
+  }
+
   private String getScalarDbPropertiesFilePath() {
     if (StringUtils.isBlank(configFilePath)) {
       throw new IllegalArgumentException(DataLoaderError.CONFIG_FILE_PATH_BLANK.buildMessage());
@@ -160,8 +183,7 @@ private void validateOutputDirectory() throws DirectoryValidationException {
 
   private ExportManager createExportManager(
       TransactionFactory transactionFactory, ScalarDbDao scalarDbDao, FileFormat fileFormat) {
-    ProducerTaskFactory taskFactory =
-        new ProducerTaskFactory(delimiter, includeTransactionMetadata, prettyPrintJson);
+    ProducerTaskFactory taskFactory = new ProducerTaskFactory(delimiter, prettyPrintJson);
     DistributedTransactionManager manager = transactionFactory.getTransactionManager();
     switch (fileFormat) {
       case JSON:
@@ -180,7 +202,6 @@ private ExportOptions buildExportOptions(Key partitionKey, ScanRange scanRange)
         ExportOptions.builder(namespace, table, partitionKey, outputFormat)
             .sortOrders(sortOrders)
             .excludeHeaderRow(excludeHeader)
-            .includeTransactionMetadata(includeTransactionMetadata)
             .delimiter(delimiter)
             .limit(limit)
             .maxThreadCount(maxThreads)
diff --git a/data-loader/cli/src/main/java/com/scalar/db/dataloader/cli/command/dataexport/ExportCommandOptions.java b/data-loader/cli/src/main/java/com/scalar/db/dataloader/cli/command/dataexport/ExportCommandOptions.java
index 4492f36259..00451a8889 100755
--- a/data-loader/cli/src/main/java/com/scalar/db/dataloader/cli/command/dataexport/ExportCommandOptions.java
+++ b/data-loader/cli/src/main/java/com/scalar/db/dataloader/cli/command/dataexport/ExportCommandOptions.java
@@ -19,6 +19,8 @@ public class ExportCommandOptions {
   public static final String MAX_THREADS_OPTION = "--max-threads";
   public static final String MAX_THREADS_OPTION_SHORT = "-mt";
   public static final String DEPRECATED_THREADS_OPTION = "--threads";
+  public static final String DEPRECATED_INCLUDE_METADATA_OPTION = "--include-metadata";
+  public static final String DEPRECATED_INCLUDE_METADATA_OPTION_SHORT = "-m";
 
   @CommandLine.Option(
       names = {"--config", "-c"},
@@ -69,10 +71,19 @@ public class ExportCommandOptions {
       defaultValue = "json")
   protected FileFormat outputFormat;
 
+  /**
+   * @deprecated As of release 3.17.0 This option is no longer used and will be removed in release
+   *     4.0.0. The option is not fully removed as users who might already have their scripts or
+   *     commands pre-set might pass the argument and when passed if not supported, picocli will
+   *     throw an error. We want to avoid that and instead just show a warning.
+   */
+  @Deprecated
   @CommandLine.Option(
       names = {"--include-metadata", "-m"},
-      description = "Include transaction metadata in the exported data (default: false)",
-      defaultValue = "false")
+      description =
+          "Deprecated: This option is no longer used. Please use scalar.db.consensus_commit.include_metadata.enabled to control whether transaction metadata is included in scan operations.",
+      defaultValue = "false",
+      hidden = true)
   protected boolean includeTransactionMetadata;
 
   @CommandLine.Option(
diff --git a/data-loader/cli/src/main/java/com/scalar/db/dataloader/cli/command/dataimport/ImportCommand.java b/data-loader/cli/src/main/java/com/scalar/db/dataloader/cli/command/dataimport/ImportCommand.java
index 56d2e84428..7de29db0cd 100755
--- a/data-loader/cli/src/main/java/com/scalar/db/dataloader/cli/command/dataimport/ImportCommand.java
+++ b/data-loader/cli/src/main/java/com/scalar/db/dataloader/cli/command/dataimport/ImportCommand.java
@@ -8,13 +8,10 @@
 import com.scalar.db.api.TableMetadata;
 import com.scalar.db.dataloader.core.DataLoaderError;
 import com.scalar.db.dataloader.core.FileFormat;
-import com.scalar.db.dataloader.core.ScalarDbMode;
 import com.scalar.db.dataloader.core.dataimport.ImportManager;
 import com.scalar.db.dataloader.core.dataimport.ImportOptions;
 import com.scalar.db.dataloader.core.dataimport.controlfile.ControlFile;
 import com.scalar.db.dataloader.core.dataimport.controlfile.ControlFileTable;
-import com.scalar.db.dataloader.core.dataimport.dao.ScalarDbStorageManager;
-import com.scalar.db.dataloader.core.dataimport.dao.ScalarDbTransactionManager;
 import com.scalar.db.dataloader.core.dataimport.log.ImportLoggerConfig;
 import com.scalar.db.dataloader.core.dataimport.log.LogMode;
 import com.scalar.db.dataloader.core.dataimport.log.SingleFileImportLogger;
@@ -26,7 +23,6 @@
 import com.scalar.db.dataloader.core.tablemetadata.TableMetadataException;
 import com.scalar.db.dataloader.core.tablemetadata.TableMetadataService;
 import com.scalar.db.dataloader.core.util.TableMetadataUtil;
-import com.scalar.db.service.StorageFactory;
 import com.scalar.db.service.TransactionFactory;
 import java.io.BufferedReader;
 import java.io.File;
@@ -150,32 +146,14 @@ private ImportManager createImportManager(
       throws IOException {
     File configFile = new File(configFilePath);
     ImportProcessorFactory importProcessorFactory = new DefaultImportProcessorFactory();
-    ImportManager importManager;
-    if (scalarDbMode == ScalarDbMode.TRANSACTION) {
-      ScalarDbTransactionManager scalarDbTransactionManager =
-          new ScalarDbTransactionManager(TransactionFactory.create(configFile));
-      importManager =
-          new ImportManager(
-              tableMetadataMap,
-              reader,
-              importOptions,
-              importProcessorFactory,
-              ScalarDbMode.TRANSACTION,
-              null,
-              scalarDbTransactionManager.getDistributedTransactionManager());
-    } else {
-      ScalarDbStorageManager scalarDbStorageManager =
-          new ScalarDbStorageManager(StorageFactory.create(configFile));
-      importManager =
-          new ImportManager(
-              tableMetadataMap,
-              reader,
-              importOptions,
-              importProcessorFactory,
-              ScalarDbMode.STORAGE,
-              scalarDbStorageManager.getDistributedStorage(),
-              null);
-    }
+    ImportManager importManager =
+        new ImportManager(
+            tableMetadataMap,
+            reader,
+            importOptions,
+            importProcessorFactory,
+            scalarDbMode,
+            TransactionFactory.create(configFile).getTransactionManager());
     if (importOptions.getLogMode().equals(LogMode.SPLIT_BY_DATA_CHUNK)) {
       importManager.addListener(new SplitByDataChunkImportLogger(config, logWriterFactory));
     } else {
diff --git a/data-loader/cli/src/test/java/com/scalar/db/dataloader/cli/command/dataexport/ExportCommandTest.java b/data-loader/cli/src/test/java/com/scalar/db/dataloader/cli/command/dataexport/ExportCommandTest.java
index 9ecc616e25..4c60109856 100755
--- a/data-loader/cli/src/test/java/com/scalar/db/dataloader/cli/command/dataexport/ExportCommandTest.java
+++ b/data-loader/cli/src/test/java/com/scalar/db/dataloader/cli/command/dataexport/ExportCommandTest.java
@@ -1,6 +1,7 @@
 package com.scalar.db.dataloader.cli.command.dataexport;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -152,14 +153,14 @@ void call_withOnlyDeprecatedStartExclusive_shouldApplyInvertedValue() {
     cmd.parseArgs(args);
 
     // Verify the deprecated value was parsed
-    assertEquals(true, command.startExclusiveDeprecated);
+    assertTrue(command.startExclusiveDeprecated);
 
     // Apply deprecated options (this is what the command does after validation)
     command.applyDeprecatedOptions();
 
     // Verify the value was applied with inverted logic
     // start-exclusive=true should become start-inclusive=false
-    assertEquals(false, command.scanStartInclusive);
+    assertFalse(command.scanStartInclusive);
   }
 
   @Test
@@ -181,14 +182,14 @@ void call_withOnlyDeprecatedEndExclusive_shouldApplyInvertedValue() {
     cmd.parseArgs(args);
 
     // Verify the deprecated value was parsed
-    assertEquals(false, command.endExclusiveDeprecated);
+    assertFalse(command.endExclusiveDeprecated);
 
     // Apply deprecated options (this is what the command does after validation)
     command.applyDeprecatedOptions();
 
     // Verify the value was applied with inverted logic
     // end-exclusive=false should become end-inclusive=true
-    assertEquals(true, command.scanEndInclusive);
+    assertTrue(command.scanEndInclusive);
   }
 
   @Test
@@ -273,4 +274,83 @@ void call_withoutMaxThreads_shouldDefaultToAvailableProcessors() {
     // Verify it was set to available processors
     assertEquals(Runtime.getRuntime().availableProcessors(), command.maxThreads);
   }
+
+  @Test
+  void call_withDeprecatedIncludeMetadataOption_shouldParseAndDetectOption() {
+    // Test that the deprecated option can be parsed without crashing
+    // and is detected for warning purposes (both long and short forms)
+    String[] argsWithLongForm = {
+      "--config",
+      "scalardb.properties",
+      "--namespace",
+      "scalar",
+      "--table",
+      "asset",
+      "--format",
+      "JSON",
+      "--include-metadata"
+    };
+
+    String[] argsWithShortForm = {
+      "--config",
+      "scalardb.properties",
+      "--namespace",
+      "scalar",
+      "--table",
+      "asset",
+      "--format",
+      "JSON",
+      "-m"
+    };
+
+    // Test long form
+    ExportCommand commandLong = new ExportCommand();
+    CommandLine cmdLong = new CommandLine(commandLong);
+    cmdLong.parseArgs(argsWithLongForm);
+    commandLong.spec = cmdLong.getCommandSpec();
+
+    // Verify the option is detected (so warning will trigger)
+    assertTrue(
+        cmdLong
+            .getParseResult()
+            .hasMatchedOption(ExportCommandOptions.DEPRECATED_INCLUDE_METADATA_OPTION));
+
+    // Test short form
+    ExportCommand commandShort = new ExportCommand();
+    CommandLine cmdShort = new CommandLine(commandShort);
+    cmdShort.parseArgs(argsWithShortForm);
+    commandShort.spec = cmdShort.getCommandSpec();
+
+    // Verify the short option is detected (so warning will trigger)
+    assertTrue(
+        cmdShort
+            .getParseResult()
+            .hasMatchedOption(ExportCommandOptions.DEPRECATED_INCLUDE_METADATA_OPTION_SHORT));
+  }
+
+  @Test
+  void call_withoutDeprecatedIncludeMetadataOption_shouldNotDetectOption() {
+    // Verify that when the deprecated option is NOT used, it's not detected
+    // (so warning won't trigger incorrectly)
+    String[] args = {
+      "--config",
+      "scalardb.properties",
+      "--namespace",
+      "scalar",
+      "--table",
+      "asset",
+      "--format",
+      "JSON"
+    };
+
+    ExportCommand command = new ExportCommand();
+    CommandLine cmd = new CommandLine(command);
+    cmd.parseArgs(args);
+    command.spec = cmd.getCommandSpec();
+
+    // Verify the option is NOT detected (so warning won't trigger)
+    assertFalse(
+        cmd.getParseResult()
+            .hasMatchedOption(ExportCommandOptions.DEPRECATED_INCLUDE_METADATA_OPTION));
+  }
 }
diff --git a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/CsvExportManager.java b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/CsvExportManager.java
index 9589e405ac..eeeece1428 100644
--- a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/CsvExportManager.java
+++ b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/CsvExportManager.java
@@ -5,7 +5,6 @@
 import com.scalar.db.dataloader.core.dataexport.producer.ProducerTaskFactory;
 import com.scalar.db.dataloader.core.dataimport.dao.ScalarDbDao;
 import com.scalar.db.dataloader.core.util.CsvUtil;
-import com.scalar.db.transaction.consensuscommit.ConsensusCommitUtils;
 import java.io.IOException;
 import java.io.Writer;
 import java.util.Iterator;
@@ -77,8 +76,7 @@ private String createCsvHeaderRow(ExportOptions exportOptions, TableMetadata tab
     Iterator<String> iterator = tableMetadata.getColumnNames().iterator();
     while (iterator.hasNext()) {
       String columnName = iterator.next();
-      if (shouldIgnoreColumn(
-          exportOptions.isIncludeTransactionMetadata(), columnName, tableMetadata, projections)) {
+      if (!projections.isEmpty() && !projections.contains(columnName)) {
         continue;
       }
       headerRow.append(columnName);
@@ -90,24 +88,4 @@ private String createCsvHeaderRow(ExportOptions exportOptions, TableMetadata tab
     headerRow.append("\n");
     return headerRow.toString();
   }
-
-  /**
-   * To ignore a column or not based on conditions such as if it is a metadata column or if it is
-   * not include in selected projections
-   *
-   * @param isIncludeTransactionMetadata to include transaction metadata or not
-   * @param columnName column name
-   * @param tableMetadata table metadata
-   * @param projections selected columns for projection
-   * @return ignore the column or not
-   */
-  private boolean shouldIgnoreColumn(
-      boolean isIncludeTransactionMetadata,
-      String columnName,
-      TableMetadata tableMetadata,
-      List<String> projections) {
-    return (!isIncludeTransactionMetadata
-            && ConsensusCommitUtils.isTransactionMetaColumn(columnName, tableMetadata))
-        || (!projections.isEmpty() && !projections.contains(columnName));
-  }
 }
diff --git a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/ExportManager.java b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/ExportManager.java
index 4a6a41596e..0d60b9e073 100644
--- a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/ExportManager.java
+++ b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/ExportManager.java
@@ -11,7 +11,6 @@
 import com.scalar.db.dataloader.core.dataexport.validation.ExportOptionsValidator;
 import com.scalar.db.dataloader.core.dataimport.dao.ScalarDbDao;
 import com.scalar.db.dataloader.core.dataimport.dao.ScalarDbDaoException;
-import com.scalar.db.dataloader.core.util.TableMetadataUtil;
 import com.scalar.db.exception.transaction.CrudException;
 import com.scalar.db.exception.transaction.UnknownTransactionStatusException;
 import com.scalar.db.io.DataType;
@@ -76,7 +75,6 @@ public ExportReport startExport(
     try {
       validateExportOptions(exportOptions, tableMetadata);
       Map<String, DataType> dataTypeByColumnName = tableMetadata.getColumnDataTypes();
-      handleTransactionMetadata(exportOptions, tableMetadata);
       processHeader(exportOptions, tableMetadata, writer);
 
       ExecutorService executorService =
@@ -198,22 +196,6 @@ private void validateExportOptions(ExportOptions exportOptions, TableMetadata ta
     ExportOptionsValidator.validate(exportOptions, tableMetadata);
   }
 
-  /**
-   * To update projection columns of export options if include metadata options is enabled
-   *
-   * @param exportOptions export options
-   * @param tableMetadata metadata of the table
-   */
-  private void handleTransactionMetadata(ExportOptions exportOptions, TableMetadata tableMetadata) {
-    if (exportOptions.isIncludeTransactionMetadata()
-        && !exportOptions.getProjectionColumns().isEmpty()) {
-      List<String> projectionMetadata =
-          TableMetadataUtil.populateProjectionsWithMetadata(
-              tableMetadata, exportOptions.getProjectionColumns());
-      exportOptions.setProjectionColumns(projectionMetadata);
-    }
-  }
-
   /**
    * To create a scanner object
    *
diff --git a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/ExportOptions.java b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/ExportOptions.java
index fac6df7b4e..54feeaeca1 100644
--- a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/ExportOptions.java
+++ b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/ExportOptions.java
@@ -27,7 +27,6 @@ public class ExportOptions {
   @Builder.Default private final int maxThreadCount = Runtime.getRuntime().availableProcessors();
   @Builder.Default private final String delimiter = ";";
   @Builder.Default private final boolean excludeHeaderRow = false;
-  @Builder.Default private final boolean includeTransactionMetadata = false;
   @Builder.Default private List<String> projectionColumns = Collections.emptyList();
   private List<Scan.Ordering> sortOrders;
 
diff --git a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/producer/CsvProducerTask.java b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/producer/CsvProducerTask.java
index 2a37c7c507..fe48106b29 100644
--- a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/producer/CsvProducerTask.java
+++ b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/producer/CsvProducerTask.java
@@ -6,7 +6,6 @@
 import com.scalar.db.dataloader.core.util.CsvUtil;
 import com.scalar.db.dataloader.core.util.DecimalUtil;
 import com.scalar.db.io.DataType;
-import com.scalar.db.transaction.consensuscommit.ConsensusCommitUtils;
 import java.nio.charset.Charset;
 import java.time.Instant;
 import java.time.LocalDate;
@@ -33,19 +32,17 @@ public class CsvProducerTask extends ProducerTask {
   /**
    * Class constructor
    *
-   * @param includeMetadata Include metadata in the exported data
    * @param projectColumns list of columns that is required in export data
    * @param tableMetadata Metadata for a single ScalarDB table
    * @param columnDataTypes Map of data types for the all columns in a ScalarDB table
    * @param delimiter Delimiter used in csv content
    */
   public CsvProducerTask(
-      boolean includeMetadata,
       List<String> projectColumns,
       TableMetadata tableMetadata,
       Map<String, DataType> columnDataTypes,
       String delimiter) {
-    super(includeMetadata, projectColumns, tableMetadata, columnDataTypes);
+    super(projectColumns, tableMetadata, columnDataTypes);
     this.delimiter = delimiter;
   }
 
@@ -82,12 +79,10 @@ private String convertResultToCsv(Result result) {
       while (iterator.hasNext()) {
         String columnName = iterator.next();
 
-        // Skip the field if it can be ignored based on check
+        // Skip the field if it's not in the projection list (when projections are specified)
         boolean columnNotProjected =
             !projectedColumnsSet.isEmpty() && !projectedColumnsSet.contains(columnName);
-        boolean isMetadataColumn =
-            ConsensusCommitUtils.isTransactionMetaColumn(columnName, tableMetadata);
-        if (columnNotProjected || (!includeMetadata && isMetadataColumn)) {
+        if (columnNotProjected) {
           continue;
         }
 
diff --git a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/producer/JsonLineProducerTask.java b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/producer/JsonLineProducerTask.java
index aabbba205c..1b3b757563 100644
--- a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/producer/JsonLineProducerTask.java
+++ b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/producer/JsonLineProducerTask.java
@@ -6,7 +6,6 @@
 import com.scalar.db.api.TableMetadata;
 import com.scalar.db.dataloader.core.DataLoaderObjectMapper;
 import com.scalar.db.io.DataType;
-import com.scalar.db.transaction.consensuscommit.ConsensusCommitUtils;
 import java.nio.charset.Charset;
 import java.time.Instant;
 import java.time.LocalDate;
@@ -28,17 +27,15 @@ public class JsonLineProducerTask extends ProducerTask {
   /**
    * Class constructor
    *
-   * @param includeMetadata Include metadata in the exported data
    * @param projectionColumns list of columns that is required in export data
    * @param tableMetadata Metadata for a single ScalarDB table
    * @param columnDataTypes Map of data types for the all columns in a ScalarDB table
    */
   public JsonLineProducerTask(
-      boolean includeMetadata,
       List<String> projectionColumns,
       TableMetadata tableMetadata,
       Map<String, DataType> columnDataTypes) {
-    super(includeMetadata, projectionColumns, tableMetadata, columnDataTypes);
+    super(projectionColumns, tableMetadata, columnDataTypes);
   }
 
   /**
@@ -72,12 +69,10 @@ private ObjectNode generateJsonForResult(Result result) {
 
     // Loop through all the columns and to the json object
     for (String columnName : tableColumns) {
-      // Skip the field if it can be ignored based on check
+      // Skip the field if it's not in the projection list (when projections are specified)
       boolean columnNotProjected =
           !projectedColumnsSet.isEmpty() && !projectedColumnsSet.contains(columnName);
-      boolean isMetadataColumn =
-          ConsensusCommitUtils.isTransactionMetaColumn(columnName, tableMetadata);
-      if (columnNotProjected || (!includeMetadata && isMetadataColumn)) {
+      if (columnNotProjected) {
         continue;
       }
 
diff --git a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/producer/JsonProducerTask.java b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/producer/JsonProducerTask.java
index 742911e211..b167ec9b8f 100644
--- a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/producer/JsonProducerTask.java
+++ b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/producer/JsonProducerTask.java
@@ -7,7 +7,6 @@
 import com.scalar.db.api.TableMetadata;
 import com.scalar.db.dataloader.core.DataLoaderObjectMapper;
 import com.scalar.db.io.DataType;
-import com.scalar.db.transaction.consensuscommit.ConsensusCommitUtils;
 import java.nio.charset.Charset;
 import java.time.Instant;
 import java.time.LocalDate;
@@ -30,19 +29,17 @@ public class JsonProducerTask extends ProducerTask {
   /**
    * Class constructor
    *
-   * @param includeMetadata Include metadata in the exported data
    * @param projectionColumns list of columns that is required in export data
    * @param tableMetadata Metadata for a single ScalarDB table
    * @param columnDataTypes Map of data types for the all columns in a ScalarDB table
    * @param prettyPrintJson Json data should be formatted or not
    */
   public JsonProducerTask(
-      boolean includeMetadata,
       List<String> projectionColumns,
       TableMetadata tableMetadata,
       Map<String, DataType> columnDataTypes,
       boolean prettyPrintJson) {
-    super(includeMetadata, projectionColumns, tableMetadata, columnDataTypes);
+    super(projectionColumns, tableMetadata, columnDataTypes);
     this.prettyPrintJson = prettyPrintJson;
   }
 
@@ -84,12 +81,10 @@ private ObjectNode generateJsonForResult(Result result) {
 
     // Loop through all the columns and to the json object
     for (String columnName : tableColumns) {
-      // Skip the field if it can be ignored based on check
+      // Skip the field if it's not in the projection list (when projections are specified)
       boolean columnNotProjected =
           !projectedColumnsSet.isEmpty() && !projectedColumnsSet.contains(columnName);
-      boolean isMetadataColumn =
-          ConsensusCommitUtils.isTransactionMetaColumn(columnName, tableMetadata);
-      if (columnNotProjected || (!includeMetadata && isMetadataColumn)) {
+      if (columnNotProjected) {
         continue;
       }
 
diff --git a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/producer/ProducerTask.java b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/producer/ProducerTask.java
index acb9a90519..48dcc58850 100644
--- a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/producer/ProducerTask.java
+++ b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/producer/ProducerTask.java
@@ -32,12 +32,6 @@ public abstract class ProducerTask {
    */
   protected final Map<String, DataType> dataTypeByColumnName;
 
-  /**
-   * Flag indicating whether to include internal metadata columns (e.g., transaction metadata) in
-   * the output.
-   */
-  protected final boolean includeMetadata;
-
   /**
    * A set of column names to include in the exported output. If empty, all columns are included by
    * default.
@@ -47,17 +41,14 @@ public abstract class ProducerTask {
   /**
    * Class constructor
    *
-   * @param includeMetadata Include metadata in the exported data
    * @param projectionColumns List of column name for projection
    * @param tableMetadata Metadata of the ScalarDB table
    * @param columnDataTypes Map of data types for the all columns in a ScalarDB table
    */
   protected ProducerTask(
-      boolean includeMetadata,
       List<String> projectionColumns,
       TableMetadata tableMetadata,
       Map<String, DataType> columnDataTypes) {
-    this.includeMetadata = includeMetadata;
     this.projectedColumnsSet = new HashSet<>(projectionColumns);
     this.tableMetadata = tableMetadata;
     this.dataTypeByColumnName = columnDataTypes;
diff --git a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/producer/ProducerTaskFactory.java b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/producer/ProducerTaskFactory.java
index 21268e016d..b632e5428a 100644
--- a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/producer/ProducerTaskFactory.java
+++ b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/producer/ProducerTaskFactory.java
@@ -14,7 +14,6 @@
 public class ProducerTaskFactory {
 
   private final String delimiter;
-  private final boolean includeMetadata;
   private final boolean prettyPrintJson;
 
   /**
@@ -36,21 +35,15 @@ public ProducerTask createProducerTask(
       case JSON:
         producerTask =
             new JsonProducerTask(
-                includeMetadata,
-                projectionColumns,
-                tableMetadata,
-                dataTypeByColumnName,
-                prettyPrintJson);
+                projectionColumns, tableMetadata, dataTypeByColumnName, prettyPrintJson);
         break;
       case JSONL:
         producerTask =
-            new JsonLineProducerTask(
-                includeMetadata, projectionColumns, tableMetadata, dataTypeByColumnName);
+            new JsonLineProducerTask(projectionColumns, tableMetadata, dataTypeByColumnName);
         break;
       case CSV:
         producerTask =
-            new CsvProducerTask(
-                includeMetadata, projectionColumns, tableMetadata, dataTypeByColumnName, delimiter);
+            new CsvProducerTask(projectionColumns, tableMetadata, dataTypeByColumnName, delimiter);
         break;
       default:
         throw new IllegalArgumentException(
diff --git a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/ImportManager.java b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/ImportManager.java
index 07ef2dd756..89b836c750 100644
--- a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/ImportManager.java
+++ b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/ImportManager.java
@@ -1,6 +1,5 @@
 package com.scalar.db.dataloader.core.dataimport;
 
-import com.scalar.db.api.DistributedStorage;
 import com.scalar.db.api.DistributedTransactionManager;
 import com.scalar.db.api.TableMetadata;
 import com.scalar.db.dataloader.core.ScalarDbMode;
@@ -44,7 +43,6 @@ public class ImportManager implements ImportEventListener {
   private final ImportProcessorFactory importProcessorFactory;
   private final List<ImportEventListener> listeners = new ArrayList<>();
   private final ScalarDbMode scalarDbMode;
-  private final DistributedStorage distributedStorage;
   private final DistributedTransactionManager distributedTransactionManager;
 
   /**
@@ -62,7 +60,6 @@ public void startImport() {
             .tableMetadataByTableName(tableMetadata)
             .dao(new ScalarDbDao())
             .distributedTransactionManager(distributedTransactionManager)
-            .distributedStorage(distributedStorage)
             .tableColumnDataTypes(getTableColumnDataTypes())
             .build();
     ImportProcessor processor = importProcessorFactory.createImportProcessor(params);
@@ -169,9 +166,7 @@ public void onAllDataChunksCompleted() {
   /** Close resources properly once the process is completed */
   public void closeResources() {
     try {
-      if (distributedStorage != null) {
-        distributedStorage.close();
-      } else if (distributedTransactionManager != null) {
+      if (distributedTransactionManager != null) {
         distributedTransactionManager.close();
       }
     } catch (Throwable e) {
diff --git a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/dao/ScalarDbDao.java b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/dao/ScalarDbDao.java
index 6afceae89a..a56ceca1e0 100644
--- a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/dao/ScalarDbDao.java
+++ b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/dao/ScalarDbDao.java
@@ -1,6 +1,5 @@
 package com.scalar.db.dataloader.core.dataimport.dao;
 
-import com.scalar.db.api.DistributedStorage;
 import com.scalar.db.api.DistributedTransaction;
 import com.scalar.db.api.DistributedTransactionManager;
 import com.scalar.db.api.Get;
@@ -10,15 +9,13 @@
 import com.scalar.db.api.Result;
 import com.scalar.db.api.Scan;
 import com.scalar.db.api.ScanBuilder;
-import com.scalar.db.api.Scanner;
 import com.scalar.db.api.TransactionManagerCrudOperable;
 import com.scalar.db.dataloader.core.DataLoaderError;
 import com.scalar.db.dataloader.core.ScanRange;
-import com.scalar.db.exception.storage.ExecutionException;
 import com.scalar.db.exception.transaction.CrudException;
+import com.scalar.db.exception.transaction.UnknownTransactionStatusException;
 import com.scalar.db.io.Column;
 import com.scalar.db.io.Key;
-import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.NoSuchElementException;
@@ -29,22 +26,30 @@
 public class ScalarDbDao {
 
   /**
-   * Retrieve record from ScalarDB instance in storage mode
+   * Retrieves a record from a ScalarDB table using the specified partition and optional clustering
+   * keys while operating in storage mode through a {@link DistributedTransactionManager}.
    *
-   * @param namespace Namespace name
-   * @param table Table name
-   * @param partitionKey Partition key
-   * @param clusteringKey Optional clustering key for get
-   * @param storage Distributed storage for ScalarDB connection that is running in storage mode.
-   * @return Optional get result
-   * @throws ScalarDbDaoException if something goes wrong while reading the data
+   * <p>This method creates a {@link Get} operation for the given namespace and table, executes it
+   * using the provided transaction manager, and returns the result if the record exists.
+   *
+   * @param namespace the name of the ScalarDB namespace containing the target table
+   * @param table the name of the table to retrieve the record from
+   * @param partitionKey the partition key identifying the record's partition
+   * @param clusteringKey the optional clustering key identifying a specific record within the
+   *     partition; may be {@code null} if the table does not use clustering keys
+   * @param manager the {@link DistributedTransactionManager} instance used to perform the get
+   *     operation
+   * @return an {@link Optional} containing the {@link Result} if the record exists, or an empty
+   *     {@link Optional} if not found
+   * @throws ScalarDbDaoException if an error occurs while performing the get operation or
+   *     interacting with ScalarDB
    */
   public Optional<Result> get(
       String namespace,
       String table,
       Key partitionKey,
       Key clusteringKey,
-      DistributedStorage storage)
+      DistributedTransactionManager manager)
       throws ScalarDbDaoException {
 
     // Retrieving the key data for logging
@@ -52,8 +57,8 @@ public Optional<Result> get(
 
     try {
       Get get = createGetWith(namespace, table, partitionKey, clusteringKey);
-      return storage.get(get);
-    } catch (ExecutionException e) {
+      return manager.get(get);
+    } catch (CrudException | UnknownTransactionStatusException e) {
       throw new ScalarDbDaoException("error GET " + loggingKey, e);
     }
   }
@@ -117,15 +122,24 @@ public void put(
   }
 
   /**
-   * Save record in ScalarDB instance
+   * Saves a record into a ScalarDB table using the specified partition and optional clustering keys
+   * through a {@link DistributedTransactionManager}.
    *
-   * @param namespace Namespace name
-   * @param table Table name
-   * @param partitionKey Partition key
-   * @param clusteringKey Optional clustering key
-   * @param columns List of column values to be inserted or updated
-   * @param storage Distributed storage for ScalarDB connection that is running in storage mode
-   * @throws ScalarDbDaoException if something goes wrong while executing the transaction
+   * <p>This method constructs a {@link Put} operation with the provided key and column information,
+   * then executes it using the given transaction manager. The operation inserts a new record or
+   * updates an existing one if a record with the same primary key already exists.
+   *
+   * @param namespace the name of the ScalarDB namespace containing the target table
+   * @param table the name of the table where the record will be inserted or updated
+   * @param partitionKey the partition key identifying the record's partition
+   * @param clusteringKey the optional clustering key identifying a specific record within the
+   *     partition; may be {@code null} if the table does not use clustering keys
+   * @param columns the list of {@link Column} objects representing the column values to insert or
+   *     update
+   * @param manager the {@link DistributedTransactionManager} instance used to perform the put
+   *     operation
+   * @throws ScalarDbDaoException if an error occurs while executing the put operation or
+   *     interacting with ScalarDB
    */
   public void put(
       String namespace,
@@ -133,54 +147,17 @@ public void put(
       Key partitionKey,
       Key clusteringKey,
       List<Column<?>> columns,
-      DistributedStorage storage)
+      DistributedTransactionManager manager)
       throws ScalarDbDaoException {
     Put put = createPutWith(namespace, table, partitionKey, clusteringKey, columns);
     try {
-      storage.put(put);
-    } catch (ExecutionException e) {
+      manager.put(put);
+    } catch (CrudException | UnknownTransactionStatusException e) {
       throw new ScalarDbDaoException(
           DataLoaderError.ERROR_CRUD_EXCEPTION.buildMessage(e.getMessage()), e);
     }
   }
 
-  /**
-   * Scan a ScalarDB table
-   *
-   * @param namespace ScalarDB namespace
-   * @param table ScalarDB table name
-   * @param partitionKey Partition key used in ScalarDB scan
-   * @param range Optional range to set ScalarDB scan start and end values
-   * @param sorts Optional scan clustering key sorting values
-   * @param projections List of column projection to use during scan
-   * @param limit Scan limit value
-   * @param storage Distributed storage for ScalarDB connection that is running in storage mode
-   * @return List of ScalarDB scan results
-   * @throws ScalarDbDaoException if scan fails
-   */
-  public List<Result> scan(
-      String namespace,
-      String table,
-      Key partitionKey,
-      ScanRange range,
-      List<Scan.Ordering> sorts,
-      List<String> projections,
-      int limit,
-      DistributedStorage storage)
-      throws ScalarDbDaoException {
-    // Create scan
-    Scan scan = createScan(namespace, table, partitionKey, range, sorts, projections, limit);
-
-    // scan data
-    try {
-      try (Scanner scanner = storage.scan(scan)) {
-        return scanner.all();
-      }
-    } catch (ExecutionException | IOException e) {
-      throw new ScalarDbDaoException(DataLoaderError.ERROR_SCAN.buildMessage(e.getMessage()), e);
-    }
-  }
-
   /**
    * Scan a ScalarDB table
    *
diff --git a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/dao/ScalarDbStorageManager.java b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/dao/ScalarDbStorageManager.java
deleted file mode 100644
index 54185b9b3a..0000000000
--- a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/dao/ScalarDbStorageManager.java
+++ /dev/null
@@ -1,49 +0,0 @@
-package com.scalar.db.dataloader.core.dataimport.dao;
-
-import com.scalar.db.api.DistributedStorage;
-import com.scalar.db.api.DistributedStorageAdmin;
-import com.scalar.db.service.StorageFactory;
-import javax.annotation.Nullable;
-
-/**
- * A manager class for handling ScalarDB operations in storage mode.
- *
- * <p>Provides access to {@link DistributedStorage} for data operations and {@link
- * DistributedStorageAdmin} for administrative operations such as schema management.
- *
- * <p>This class is typically used when interacting with ScalarDB in a non-transactional,
- * storage-only configuration.
- */
-public class ScalarDbStorageManager {
-
-  @Nullable private final DistributedStorage storage;
-  private final DistributedStorageAdmin storageAdmin;
-
-  /**
-   * Constructs a {@code ScalarDbStorageManager} using the provided {@link StorageFactory}.
-   *
-   * @param storageFactory the factory used to create the ScalarDB storage and admin instances
-   */
-  public ScalarDbStorageManager(StorageFactory storageFactory) {
-    storage = storageFactory.getStorage();
-    storageAdmin = storageFactory.getStorageAdmin();
-  }
-
-  /**
-   * Returns distributed storage for ScalarDB connection that is running in storage mode
-   *
-   * @return distributed storage object
-   */
-  public DistributedStorage getDistributedStorage() {
-    return storage;
-  }
-
-  /**
-   * Returns distributed storage admin for ScalarDB admin operations
-   *
-   * @return distributed storage admin object
-   */
-  public DistributedStorageAdmin getDistributedStorageAdmin() {
-    return storageAdmin;
-  }
-}
diff --git a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/dao/ScalarDbTransactionManager.java b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/dao/ScalarDbTransactionManager.java
deleted file mode 100644
index e3b6f1bb85..0000000000
--- a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/dao/ScalarDbTransactionManager.java
+++ /dev/null
@@ -1,35 +0,0 @@
-package com.scalar.db.dataloader.core.dataimport.dao;
-
-import com.scalar.db.api.DistributedTransactionManager;
-import com.scalar.db.service.TransactionFactory;
-
-/**
- * A manager class for handling ScalarDB operations in transaction mode.
- *
- * <p>Provides access to {@link DistributedTransactionManager} for data operations
- *
- * <p>This class is typically used when interacting with ScalarDB in a transactional configuration.
- */
-public class ScalarDbTransactionManager {
-
-  private final DistributedTransactionManager transactionManager;
-
-  /**
-   * Constructs a {@code ScalarDbTransactionManager} using the provided {@link TransactionFactory}.
-   *
-   * @param transactionFactory the factory used to create the ScalarDB storage and admin instances
-   */
-  public ScalarDbTransactionManager(TransactionFactory transactionFactory) {
-    transactionManager = transactionFactory.getTransactionManager();
-  }
-
-  /**
-   * Returns distributed Transaction manager for ScalarDB connection that is running in transaction
-   * mode
-   *
-   * @return distributed transaction manager object
-   */
-  public DistributedTransactionManager getDistributedTransactionManager() {
-    return transactionManager;
-  }
-}
diff --git a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/processor/ImportProcessor.java b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/processor/ImportProcessor.java
index 81daf9646e..a8c3def492 100644
--- a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/processor/ImportProcessor.java
+++ b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/processor/ImportProcessor.java
@@ -360,7 +360,7 @@ private ImportTaskResult processStorageRecord(int dataChunkId, ImportRow importR
             .dao(params.getDao())
             .build();
     ImportTaskResult importRecordResult =
-        new ImportStorageTask(taskParams, params.getDistributedStorage()).execute();
+        new ImportStorageTask(taskParams, params.getDistributedTransactionManager()).execute();
 
     ImportTaskResult modifiedTaskResult =
         ImportTaskResult.builder()
diff --git a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/processor/ImportProcessorParams.java b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/processor/ImportProcessorParams.java
index 2d85325044..49448b6c4b 100644
--- a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/processor/ImportProcessorParams.java
+++ b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/processor/ImportProcessorParams.java
@@ -1,6 +1,5 @@
 package com.scalar.db.dataloader.core.dataimport.processor;
 
-import com.scalar.db.api.DistributedStorage;
 import com.scalar.db.api.DistributedTransactionManager;
 import com.scalar.db.api.TableMetadata;
 import com.scalar.db.dataloader.core.ScalarDbMode;
@@ -36,9 +35,6 @@ public class ImportProcessorParams {
   /** Data Access Object for ScalarDB operations. */
   ScalarDbDao dao;
 
-  /** Storage interface for non-transactional operations. */
-  DistributedStorage distributedStorage;
-
   /** Transaction manager for handling transactional operations. */
   DistributedTransactionManager distributedTransactionManager;
 }
diff --git a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/task/ImportStorageTask.java b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/task/ImportStorageTask.java
index e847cc3a34..378cc0f3f4 100644
--- a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/task/ImportStorageTask.java
+++ b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/task/ImportStorageTask.java
@@ -1,6 +1,6 @@
 package com.scalar.db.dataloader.core.dataimport.task;
 
-import com.scalar.db.api.DistributedStorage;
+import com.scalar.db.api.DistributedTransactionManager;
 import com.scalar.db.api.Result;
 import com.scalar.db.dataloader.core.dataimport.dao.ScalarDbDaoException;
 import com.scalar.db.io.Column;
@@ -9,74 +9,77 @@
 import java.util.Optional;
 
 /**
- * An import task that interacts with a {@link DistributedStorage} for data retrieval and storage
- * operations.
+ * An import task that performs data operations using a {@link DistributedTransactionManager}.
  *
- * <p>This class extends {@link ImportTask} and provides concrete implementations for fetching and
- * storing records using a {@link DistributedStorage} instance. It acts as a bridge between the
- * import process and the underlying distributed storage system.
+ * <p>This class extends {@link ImportTask} and provides concrete implementations for retrieving and
+ * storing records using a {@link DistributedTransactionManager} instance. It serves as a bridge
+ * between the import process and the underlying transactional data management layer in ScalarDB.
  *
- * <p>The task handles both read and write operations:
+ * <p>The task supports both read and write operations:
  *
  * <ul>
- *   <li>Reading existing records using partition and clustering keys
- *   <li>Storing new or updated records with their associated columns
+ *   <li>Retrieving existing records using partition and optional clustering keys.
+ *   <li>Inserting or updating records with their associated column values.
  * </ul>
  *
- * <p>All storage operations are performed through the provided {@link DistributedStorage} instance,
- * which must be properly initialized before creating this task.
+ * <p>All data operations are delegated to the DAO configured within {@link ImportTaskParams},
+ * executed through the provided {@link DistributedTransactionManager}. The manager must be properly
+ * initialized and connected before creating an instance of this class.
  */
 public class ImportStorageTask extends ImportTask {
 
-  private final DistributedStorage storage;
+  private final DistributedTransactionManager manager;
 
   /**
-   * Constructs an {@code ImportStorageTask} with the specified parameters and storage.
+   * Constructs an {@code ImportStorageTask} with the specified parameters and transaction manager.
    *
    * @param params the import task parameters containing configuration and DAO objects
-   * @param storage the distributed storage instance to be used for data operations
-   * @throws NullPointerException if either params or storage is null
+   * @param manager the {@link DistributedTransactionManager} instance used for transactional data
+   *     operations
+   * @throws NullPointerException if either {@code params} or {@code manager} is {@code null}
    */
-  public ImportStorageTask(ImportTaskParams params, DistributedStorage storage) {
+  public ImportStorageTask(ImportTaskParams params, DistributedTransactionManager manager) {
     super(params);
-    this.storage = storage;
+    this.manager = manager;
   }
 
   /**
-   * Retrieves a data record from the distributed storage using the specified keys.
+   * Retrieves a data record from the database using the specified keys.
    *
-   * <p>This method attempts to fetch a single record from the specified table using both partition
-   * and clustering keys. The operation is performed through the configured DAO using the associated
-   * storage instance.
+   * <p>This method attempts to fetch a single record from the given table using the provided
+   * partition and optional clustering keys. The retrieval is executed through the DAO configured in
+   * {@link ImportTaskParams}, utilizing the associated {@link DistributedTransactionManager}.
    *
-   * @param namespace the namespace of the table to query
-   * @param tableName the name of the table to query
+   * @param namespace the ScalarDB namespace of the target table
+   * @param tableName the name of the target table
    * @param partitionKey the partition key identifying the record's partition
-   * @param clusteringKey the clustering key for further record identification within the partition
-   * @return an {@link Optional} containing the {@link Result} if the record exists, otherwise an
-   *     empty {@link Optional}
-   * @throws ScalarDbDaoException if an error occurs during the retrieval operation, such as
-   *     connection issues or invalid table/namespace
+   * @param clusteringKey the clustering key for uniquely identifying the record within the
+   *     partition
+   * @return an {@link Optional} containing the {@link Result} if a record exists, or an empty
+   *     {@link Optional} if no matching record is found
+   * @throws ScalarDbDaoException if an error occurs during the retrieval operation, such as a
+   *     connection failure or invalid table/namespace
    */
   @Override
   protected Optional<Result> getDataRecord(
       String namespace, String tableName, Key partitionKey, Key clusteringKey)
       throws ScalarDbDaoException {
-    return params.getDao().get(namespace, tableName, partitionKey, clusteringKey, this.storage);
+    return params.getDao().get(namespace, tableName, partitionKey, clusteringKey, this.manager);
   }
 
   /**
-   * Saves a record into the distributed storage with the specified keys and columns.
+   * Saves or updates a record in the database using the specified keys and column values.
    *
    * <p>This method writes or updates a record in the specified table using the provided keys and
-   * column values. The operation is performed through the configured DAO using the associated
-   * storage instance.
+   * columns. The operation is performed through the DAO configured in {@link ImportTaskParams},
+   * utilizing the associated {@link DistributedTransactionManager}.
    *
-   * @param namespace the namespace of the target table
+   * @param namespace the ScalarDB namespace of the target table
    * @param tableName the name of the target table
    * @param partitionKey the partition key determining where the record will be stored
-   * @param clusteringKey the clustering key for organizing records within the partition
-   * @param columns the list of columns containing the record's data to be saved
+   * @param clusteringKey the clustering key for uniquely identifying the record within the
+   *     partition
+   * @param columns the list of columns containing the data to be stored
    * @throws ScalarDbDaoException if an error occurs during the save operation, such as connection
    *     issues, invalid data types, or constraint violations
    */
@@ -88,6 +91,6 @@ protected void saveRecord(
       Key clusteringKey,
       List<Column<?>> columns)
       throws ScalarDbDaoException {
-    params.getDao().put(namespace, tableName, partitionKey, clusteringKey, columns, this.storage);
+    params.getDao().put(namespace, tableName, partitionKey, clusteringKey, columns, this.manager);
   }
 }
diff --git a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/util/TableMetadataUtil.java b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/util/TableMetadataUtil.java
index 7cd1834d83..0eb814d408 100644
--- a/data-loader/core/src/main/java/com/scalar/db/dataloader/core/util/TableMetadataUtil.java
+++ b/data-loader/core/src/main/java/com/scalar/db/dataloader/core/util/TableMetadataUtil.java
@@ -1,12 +1,7 @@
 package com.scalar.db.dataloader.core.util;
 
-import com.scalar.db.api.TableMetadata;
 import com.scalar.db.dataloader.core.Constants;
 import com.scalar.db.dataloader.core.dataimport.controlfile.ControlFileTable;
-import com.scalar.db.transaction.consensuscommit.Attribute;
-import com.scalar.db.transaction.consensuscommit.ConsensusCommitUtils;
-import java.util.ArrayList;
-import java.util.List;
 import lombok.AccessLevel;
 import lombok.NoArgsConstructor;
 
@@ -38,37 +33,4 @@ public static String getTableLookupKey(ControlFileTable controlFileTable) {
         controlFileTable.getNamespace(),
         controlFileTable.getTable());
   }
-
-  /**
-   * Adds metadata columns to a list of projection columns for a ScalarDB table.
-   *
-   * @param tableMetadata The metadata of the ScalarDB table.
-   * @param projections A list of projection column names.
-   * @return A new list containing projection columns along with metadata columns.
-   */
-  public static List<String> populateProjectionsWithMetadata(
-      TableMetadata tableMetadata, List<String> projections) {
-    List<String> projectionMetadata = new ArrayList<>();
-    projections.forEach(
-        projection -> {
-          projectionMetadata.add(projection);
-          if (!isKeyColumn(projection, tableMetadata)) {
-            projectionMetadata.add(Attribute.BEFORE_PREFIX + projection);
-          }
-        });
-    projectionMetadata.addAll(ConsensusCommitUtils.getTransactionMetaColumns().keySet());
-    return projectionMetadata;
-  }
-
-  /**
-   * Checks whether a column is a key column (partition key or clustering key) in the table.
-   *
-   * @param column The name of the column to check.
-   * @param tableMetadata The metadata of the ScalarDB table.
-   * @return {@code true} if the column is a key column; {@code false} otherwise.
-   */
-  private static boolean isKeyColumn(String column, TableMetadata tableMetadata) {
-    return tableMetadata.getPartitionKeyNames().contains(column)
-        || tableMetadata.getClusteringKeyNames().contains(column);
-  }
 }
diff --git a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/CsvExportManagerTest.java b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/CsvExportManagerTest.java
index 26e6a88a85..31bbad5f2a 100644
--- a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/CsvExportManagerTest.java
+++ b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/CsvExportManagerTest.java
@@ -43,7 +43,7 @@ void setup() {
     manager = Mockito.mock(DistributedTransactionManager.class);
     mockData = UnitTestUtils.createTestTableMetadata();
     dao = Mockito.mock(ScalarDbDao.class);
-    producerTaskFactory = new ProducerTaskFactory(null, false, true);
+    producerTaskFactory = new ProducerTaskFactory(null, true);
   }
 
   @Test
@@ -88,7 +88,7 @@ void startExport_givenValidDataWithoutPartitionKey_shouldGenerateOutputFile()
   @Test
   void startExport_givenPartitionKey_shouldGenerateOutputFile()
       throws IOException, ScalarDbDaoException {
-    producerTaskFactory = new ProducerTaskFactory(",", false, false);
+    producerTaskFactory = new ProducerTaskFactory(",", false);
     exportManager = new CsvExportManager(manager, dao, producerTaskFactory);
     TransactionManagerCrudOperable.Scanner scanner =
         Mockito.mock(TransactionManagerCrudOperable.Scanner.class);
@@ -136,14 +136,15 @@ void startExport_givenPartitionKey_shouldGenerateOutputFile()
   @Test
   void startExport_givenNoHeaderRequired_shouldGenerateOutputFileWithoutHeader() throws Exception {
     String expectedFirstLine =
-        "9007199254740992,2147483647,true,0.000000000000000000000000000000000000000000001401298464324817,0.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000049,test value,YmxvYiB0ZXN0IHZhbHVl,2000-01-01,01:01:01,2000-01-01T01:01,1970-01-21T03:20:41.740Z";
+        "9007199254740992,2147483647,true,0.000000000000000000000000000000000000000000001401298464324817,0.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000049,test value,YmxvYiB0ZXN0IHZhbHVl,2000-01-01,01:01:01,2000-01-01T01:01,1970-01-21T03:20:41.740Z,0.000000000000000000000000000000000000000000001401298464324817,0.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000049,test value,YmxvYiB0ZXN0IHZhbHVl,txt value 464654654,2147483647,2147483647,9007199254740992,9007199254740992,test value,2147483647,2147483647,9007199254740992,9007199254740992";
 
     runExportAndAssertFirstLine(true, expectedFirstLine);
   }
 
   @Test
   void startExport_givenHeaderRequired_shouldGenerateOutputFileWithHeader() throws Exception {
-    String expectedFirstLine = "col1,col2,col3,col4,col5,col6,col7,col8,col9,col10,col11";
+    String expectedFirstLine =
+        "col1,col2,col3,col4,col5,col6,col7,col8,col9,col10,col11,before_col4,before_col5,before_col6,before_col7,tx_id,tx_state,tx_version,tx_prepared_at,tx_committed_at,before_tx_id,before_tx_state,before_tx_version,before_tx_prepared_at,before_tx_committed_at";
 
     runExportAndAssertFirstLine(false, expectedFirstLine);
   }
@@ -151,7 +152,7 @@ void startExport_givenHeaderRequired_shouldGenerateOutputFileWithHeader() throws
   private void runExportAndAssertFirstLine(boolean excludeHeader, String expectedFirstLine)
       throws Exception {
     // Arrange
-    producerTaskFactory = new ProducerTaskFactory(",", false, false);
+    producerTaskFactory = new ProducerTaskFactory(",", false);
     exportManager = new CsvExportManager(manager, dao, producerTaskFactory);
     TransactionManagerCrudOperable.Scanner scanner =
         Mockito.mock(TransactionManagerCrudOperable.Scanner.class);
diff --git a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/JsonExportManagerTest.java b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/JsonExportManagerTest.java
index b71d080323..6e4c6b30aa 100644
--- a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/JsonExportManagerTest.java
+++ b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/JsonExportManagerTest.java
@@ -43,7 +43,7 @@ void setup() {
     manager = Mockito.mock(DistributedTransactionManager.class);
     mockData = UnitTestUtils.createTestTableMetadata();
     dao = Mockito.mock(ScalarDbDao.class);
-    producerTaskFactory = new ProducerTaskFactory(null, false, true);
+    producerTaskFactory = new ProducerTaskFactory(null, true);
   }
 
   @Test
diff --git a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/JsonLineExportManagerTest.java b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/JsonLineExportManagerTest.java
index f2a6d1b360..7405a870f4 100644
--- a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/JsonLineExportManagerTest.java
+++ b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/JsonLineExportManagerTest.java
@@ -42,7 +42,7 @@ void setup() {
     manager = Mockito.mock(DistributedTransactionManager.class);
     mockData = UnitTestUtils.createTestTableMetadata();
     dao = Mockito.mock(ScalarDbDao.class);
-    producerTaskFactory = new ProducerTaskFactory(null, false, true);
+    producerTaskFactory = new ProducerTaskFactory(null, true);
   }
 
   @Test
diff --git a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/producer/CsvProducerTaskTest.java b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/producer/CsvProducerTaskTest.java
index 81a14a8292..2e0a7f334a 100644
--- a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/producer/CsvProducerTaskTest.java
+++ b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/producer/CsvProducerTaskTest.java
@@ -26,7 +26,7 @@ void setup() {
     mockMetadata = UnitTestUtils.createTestTableMetadata();
     projectedColumns = UnitTestUtils.getColumnsListOfMetadata();
     columnData = UnitTestUtils.getColumnData();
-    csvProducerTask = new CsvProducerTask(false, projectedColumns, mockMetadata, columnData, ",");
+    csvProducerTask = new CsvProducerTask(projectedColumns, mockMetadata, columnData, ",");
   }
 
   @Test
@@ -39,7 +39,7 @@ void process_withEmptyResultList_shouldReturnEmptyString() {
   @Test
   void process_withValidResultList_shouldReturnValidCsvString() {
     String expectedOutput =
-        "9007199254740992,2147483647,true,0.000000000000000000000000000000000000000000001401298464324817,0.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000049,test value,YmxvYiB0ZXN0IHZhbHVl,2000-01-01,01:01:01,2000-01-01T01:01,1970-01-21T03:20:41.740Z";
+        "9007199254740992,2147483647,true,0.000000000000000000000000000000000000000000001401298464324817,0.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000049,test value,YmxvYiB0ZXN0IHZhbHVl,2000-01-01,01:01:01,2000-01-01T01:01,1970-01-21T03:20:41.740Z,0.000000000000000000000000000000000000000000001401298464324817,0.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000049,test value,YmxvYiB0ZXN0IHZhbHVl,txt value 464654654,2147483647,2147483647,9007199254740992,9007199254740992,test value,2147483647,2147483647,9007199254740992,9007199254740992";
     Map<String, Column<?>> values = UnitTestUtils.createTestValues();
     Result result = new ResultImpl(values, mockMetadata);
     List<Result> resultList = new ArrayList<>();
@@ -50,7 +50,7 @@ void process_withValidResultList_shouldReturnValidCsvString() {
 
   @Test
   void process_withValidResultListWithMetadata_shouldReturnValidCsvString() {
-    csvProducerTask = new CsvProducerTask(true, projectedColumns, mockMetadata, columnData, ",");
+    csvProducerTask = new CsvProducerTask(projectedColumns, mockMetadata, columnData, ",");
     String expectedOutput =
         "9007199254740992,2147483647,true,0.000000000000000000000000000000000000000000001401298464324817,0.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000049,test value,YmxvYiB0ZXN0IHZhbHVl,2000-01-01,01:01:01,2000-01-01T01:01,1970-01-21T03:20:41.740Z,0.000000000000000000000000000000000000000000001401298464324817,0.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000049,test value,YmxvYiB0ZXN0IHZhbHVl,txt value 464654654,2147483647,2147483647,9007199254740992,9007199254740992,test value,2147483647,2147483647,9007199254740992,9007199254740992";
     Map<String, Column<?>> values = UnitTestUtils.createTestValues();
@@ -64,7 +64,7 @@ void process_withValidResultListWithMetadata_shouldReturnValidCsvString() {
   @Test
   void process_withValidResultList_withPartialProjections_shouldReturnValidCsvString() {
     projectedColumns = UnitTestUtils.getPartialColumnsListWithoutMetadata();
-    csvProducerTask = new CsvProducerTask(false, projectedColumns, mockMetadata, columnData, ",");
+    csvProducerTask = new CsvProducerTask(projectedColumns, mockMetadata, columnData, ",");
     String expectedOutput =
         "9007199254740992,2147483647,true,0.000000000000000000000000000000000000000000001401298464324817,0.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000049";
     Map<String, Column<?>> values = UnitTestUtils.createTestValues();
@@ -78,7 +78,7 @@ void process_withValidResultList_withPartialProjections_shouldReturnValidCsvStri
   @Test
   void process_withValidResultList_withPartialProjectionsAndMetadata_shouldReturnValidCsvString() {
     projectedColumns = UnitTestUtils.getPartialColumnsListWithMetadata();
-    csvProducerTask = new CsvProducerTask(true, projectedColumns, mockMetadata, columnData, ",");
+    csvProducerTask = new CsvProducerTask(projectedColumns, mockMetadata, columnData, ",");
     String expectedOutput =
         "9007199254740992,2147483647,true,0.000000000000000000000000000000000000000000001401298464324817,0.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000049,0.000000000000000000000000000000000000000000001401298464324817,0.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000049,test value,YmxvYiB0ZXN0IHZhbHVl,txt value 464654654,2147483647,2147483647,9007199254740992,9007199254740992,test value,2147483647,2147483647,9007199254740992,9007199254740992";
     Map<String, Column<?>> values = UnitTestUtils.createTestValues();
@@ -92,10 +92,9 @@ void process_withValidResultList_withPartialProjectionsAndMetadata_shouldReturnV
   @Test
   void
       process_withValidResultListWithNoProjectionSpecifiedWithoutMetadata_shouldReturnValidCsvString() {
-    csvProducerTask =
-        new CsvProducerTask(false, Collections.emptyList(), mockMetadata, columnData, ",");
+    csvProducerTask = new CsvProducerTask(Collections.emptyList(), mockMetadata, columnData, ",");
     String expectedOutput =
-        "9007199254740992,2147483647,true,0.000000000000000000000000000000000000000000001401298464324817,0.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000049,test value,YmxvYiB0ZXN0IHZhbHVl,2000-01-01,01:01:01,2000-01-01T01:01,1970-01-21T03:20:41.740Z";
+        "9007199254740992,2147483647,true,0.000000000000000000000000000000000000000000001401298464324817,0.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000049,test value,YmxvYiB0ZXN0IHZhbHVl,2000-01-01,01:01:01,2000-01-01T01:01,1970-01-21T03:20:41.740Z,0.000000000000000000000000000000000000000000001401298464324817,0.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000049,test value,YmxvYiB0ZXN0IHZhbHVl,txt value 464654654,2147483647,2147483647,9007199254740992,9007199254740992,test value,2147483647,2147483647,9007199254740992,9007199254740992";
     Map<String, Column<?>> values = UnitTestUtils.createTestValues();
     Result result = new ResultImpl(values, mockMetadata);
     List<Result> resultList = new ArrayList<>();
@@ -107,8 +106,7 @@ void process_withValidResultList_withPartialProjectionsAndMetadata_shouldReturnV
   @Test
   void
       process_withValidResultListWithNoProjectionSpecifiedWithMetadata_shouldReturnValidCsvString() {
-    csvProducerTask =
-        new CsvProducerTask(true, Collections.emptyList(), mockMetadata, columnData, ",");
+    csvProducerTask = new CsvProducerTask(Collections.emptyList(), mockMetadata, columnData, ",");
     String expectedOutput =
         "9007199254740992,2147483647,true,0.000000000000000000000000000000000000000000001401298464324817,0.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000049,test value,YmxvYiB0ZXN0IHZhbHVl,2000-01-01,01:01:01,2000-01-01T01:01,1970-01-21T03:20:41.740Z,0.000000000000000000000000000000000000000000001401298464324817,0.0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000049,test value,YmxvYiB0ZXN0IHZhbHVl,txt value 464654654,2147483647,2147483647,9007199254740992,9007199254740992,test value,2147483647,2147483647,9007199254740992,9007199254740992";
     Map<String, Column<?>> values = UnitTestUtils.createTestValues();
diff --git a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/producer/JsonLineProducerTaskTest.java b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/producer/JsonLineProducerTaskTest.java
index 6a450e3599..234854b170 100644
--- a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/producer/JsonLineProducerTaskTest.java
+++ b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/producer/JsonLineProducerTaskTest.java
@@ -26,8 +26,7 @@ void setup() {
     mockMetadata = UnitTestUtils.createTestTableMetadata();
     projectedColumns = UnitTestUtils.getColumnsListOfMetadata();
     columnData = UnitTestUtils.getColumnData();
-    jsonLineProducerTask =
-        new JsonLineProducerTask(false, projectedColumns, mockMetadata, columnData);
+    jsonLineProducerTask = new JsonLineProducerTask(projectedColumns, mockMetadata, columnData);
   }
 
   @Test
@@ -39,7 +38,7 @@ void process_withEmptyResultList_shouldReturnEmptyString() {
 
   @Test
   void process_withValidResultList_shouldReturnValidJsonLineString() {
-    ObjectNode rootNode = UnitTestUtils.getOutputDataWithoutMetadata();
+    ObjectNode rootNode = UnitTestUtils.getOutputDataWithMetadata();
     Map<String, Column<?>> values = UnitTestUtils.createTestValues();
     Result result = new ResultImpl(values, mockMetadata);
     List<Result> resultList = new ArrayList<>();
@@ -50,8 +49,7 @@ void process_withValidResultList_shouldReturnValidJsonLineString() {
 
   @Test
   void process_withValidResultListWithMetadata_shouldReturnValidJsonLineString() {
-    jsonLineProducerTask =
-        new JsonLineProducerTask(true, projectedColumns, mockMetadata, columnData);
+    jsonLineProducerTask = new JsonLineProducerTask(projectedColumns, mockMetadata, columnData);
     ObjectNode rootNode = UnitTestUtils.getOutputDataWithMetadata();
     Map<String, Column<?>> values = UnitTestUtils.createTestValues();
     Result result = new ResultImpl(values, mockMetadata);
@@ -64,8 +62,7 @@ void process_withValidResultListWithMetadata_shouldReturnValidJsonLineString() {
   @Test
   void process_withValidResultList_withPartialProjections_shouldReturnValidJsonLineString() {
     projectedColumns = UnitTestUtils.getPartialColumnsListWithoutMetadata();
-    jsonLineProducerTask =
-        new JsonLineProducerTask(false, projectedColumns, mockMetadata, columnData);
+    jsonLineProducerTask = new JsonLineProducerTask(projectedColumns, mockMetadata, columnData);
     ObjectNode rootNode = UnitTestUtils.getPartialOutputDataWithoutMetadata();
     Map<String, Column<?>> values = UnitTestUtils.createTestValues();
     Result result = new ResultImpl(values, mockMetadata);
@@ -79,8 +76,7 @@ void process_withValidResultList_withPartialProjections_shouldReturnValidJsonLin
   void
       process_withValidResultList_withPartialProjectionsAndMetadata_shouldReturnValidJsonLineString() {
     projectedColumns = UnitTestUtils.getPartialColumnsListWithMetadata();
-    jsonLineProducerTask =
-        new JsonLineProducerTask(true, projectedColumns, mockMetadata, columnData);
+    jsonLineProducerTask = new JsonLineProducerTask(projectedColumns, mockMetadata, columnData);
     ObjectNode rootNode = UnitTestUtils.getPartialOutputDataWithMetadata();
     Map<String, Column<?>> values = UnitTestUtils.createTestValues();
     Result result = new ResultImpl(values, mockMetadata);
@@ -94,8 +90,8 @@ void process_withValidResultList_withPartialProjections_shouldReturnValidJsonLin
   void
       process_withValidResultListWithNoProjectionSpecifiedWithoutMetadata_shouldReturnValidJsonLineString() {
     jsonLineProducerTask =
-        new JsonLineProducerTask(false, Collections.emptyList(), mockMetadata, columnData);
-    ObjectNode rootNode = UnitTestUtils.getOutputDataWithoutMetadata();
+        new JsonLineProducerTask(Collections.emptyList(), mockMetadata, columnData);
+    ObjectNode rootNode = UnitTestUtils.getOutputDataWithMetadata();
     Map<String, Column<?>> values = UnitTestUtils.createTestValues();
     Result result = new ResultImpl(values, mockMetadata);
     List<Result> resultList = new ArrayList<>();
@@ -108,7 +104,7 @@ void process_withValidResultList_withPartialProjections_shouldReturnValidJsonLin
   void
       process_withValidResultListWithNoProjectionSpecifiedWithMetadata_shouldReturnValidJsonLineString() {
     jsonLineProducerTask =
-        new JsonLineProducerTask(true, Collections.emptyList(), mockMetadata, columnData);
+        new JsonLineProducerTask(Collections.emptyList(), mockMetadata, columnData);
     ObjectNode rootNode = UnitTestUtils.getOutputDataWithMetadata();
     Map<String, Column<?>> values = UnitTestUtils.createTestValues();
     Result result = new ResultImpl(values, mockMetadata);
diff --git a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/producer/JsonProducerTaskTest.java b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/producer/JsonProducerTaskTest.java
index e14086d3c2..b8ff8f8698 100644
--- a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/producer/JsonProducerTaskTest.java
+++ b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/producer/JsonProducerTaskTest.java
@@ -26,8 +26,7 @@ void setup() {
     mockMetadata = UnitTestUtils.createTestTableMetadata();
     projectedColumns = UnitTestUtils.getColumnsListOfMetadata();
     columnData = UnitTestUtils.getColumnData();
-    jsonProducerTask =
-        new JsonProducerTask(false, projectedColumns, mockMetadata, columnData, true);
+    jsonProducerTask = new JsonProducerTask(projectedColumns, mockMetadata, columnData, true);
   }
 
   @Test
@@ -39,7 +38,7 @@ void process_withEmptyResultList_shouldReturnEmptyString() {
 
   @Test
   void process_withValidResultList_shouldReturnValidJsonString() {
-    ObjectNode rootNode = UnitTestUtils.getOutputDataWithoutMetadata();
+    ObjectNode rootNode = UnitTestUtils.getOutputDataWithMetadata();
     Map<String, Column<?>> values = UnitTestUtils.createTestValues();
     Result result = new ResultImpl(values, mockMetadata);
     List<Result> resultList = new ArrayList<>();
@@ -50,7 +49,7 @@ void process_withValidResultList_shouldReturnValidJsonString() {
 
   @Test
   void process_withValidResultListWithMetadata_shouldReturnValidJsonString() {
-    jsonProducerTask = new JsonProducerTask(true, projectedColumns, mockMetadata, columnData, true);
+    jsonProducerTask = new JsonProducerTask(projectedColumns, mockMetadata, columnData, true);
     ObjectNode rootNode = UnitTestUtils.getOutputDataWithMetadata();
     Map<String, Column<?>> values = UnitTestUtils.createTestValues();
     Result result = new ResultImpl(values, mockMetadata);
@@ -63,8 +62,7 @@ void process_withValidResultListWithMetadata_shouldReturnValidJsonString() {
   @Test
   void process_withValidResultList_withPartialProjections_shouldReturnValidJsonLineString() {
     projectedColumns = UnitTestUtils.getPartialColumnsListWithoutMetadata();
-    jsonProducerTask =
-        new JsonProducerTask(false, projectedColumns, mockMetadata, columnData, false);
+    jsonProducerTask = new JsonProducerTask(projectedColumns, mockMetadata, columnData, false);
     ObjectNode rootNode = UnitTestUtils.getPartialOutputDataWithoutMetadata();
     Map<String, Column<?>> values = UnitTestUtils.createTestValues();
     Result result = new ResultImpl(values, mockMetadata);
@@ -78,8 +76,7 @@ void process_withValidResultList_withPartialProjections_shouldReturnValidJsonLin
   void
       process_withValidResultList_withPartialProjectionsAndMetadata_shouldReturnValidJsonLineString() {
     projectedColumns = UnitTestUtils.getPartialColumnsListWithMetadata();
-    jsonProducerTask =
-        new JsonProducerTask(true, projectedColumns, mockMetadata, columnData, false);
+    jsonProducerTask = new JsonProducerTask(projectedColumns, mockMetadata, columnData, false);
     ObjectNode rootNode = UnitTestUtils.getPartialOutputDataWithMetadata();
     Map<String, Column<?>> values = UnitTestUtils.createTestValues();
     Result result = new ResultImpl(values, mockMetadata);
@@ -93,8 +90,8 @@ void process_withValidResultList_withPartialProjections_shouldReturnValidJsonLin
   void
       process_withValidResultListWithNoProjectionSpecifiedWithoutMetadata_shouldReturnValidJsonLineString() {
     jsonProducerTask =
-        new JsonProducerTask(false, Collections.emptyList(), mockMetadata, columnData, true);
-    ObjectNode rootNode = UnitTestUtils.getOutputDataWithoutMetadata();
+        new JsonProducerTask(Collections.emptyList(), mockMetadata, columnData, true);
+    ObjectNode rootNode = UnitTestUtils.getOutputDataWithMetadata();
     Map<String, Column<?>> values = UnitTestUtils.createTestValues();
     Result result = new ResultImpl(values, mockMetadata);
     List<Result> resultList = new ArrayList<>();
@@ -107,7 +104,7 @@ void process_withValidResultList_withPartialProjections_shouldReturnValidJsonLin
   void
       process_withValidResultListWithNoProjectionSpecifiedWithMetadata_shouldReturnValidJsonLineString() {
     jsonProducerTask =
-        new JsonProducerTask(true, Collections.emptyList(), mockMetadata, columnData, true);
+        new JsonProducerTask(Collections.emptyList(), mockMetadata, columnData, true);
     ObjectNode rootNode = UnitTestUtils.getOutputDataWithMetadata();
     Map<String, Column<?>> values = UnitTestUtils.createTestValues();
     Result result = new ResultImpl(values, mockMetadata);
diff --git a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/producer/ProducerTaskFactoryTest.java b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/producer/ProducerTaskFactoryTest.java
index f97e80a1d1..23fad1121a 100644
--- a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/producer/ProducerTaskFactoryTest.java
+++ b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataexport/producer/ProducerTaskFactoryTest.java
@@ -25,7 +25,7 @@ void setup() {
 
   @Test
   void createProducerTask_withJsonFileFormat_shouldReturnJsonProducerTask() {
-    ProducerTaskFactory producerTaskFactory = new ProducerTaskFactory(null, false, true);
+    ProducerTaskFactory producerTaskFactory = new ProducerTaskFactory(null, true);
     Assertions.assertEquals(
         JsonProducerTask.class,
         producerTaskFactory
@@ -35,7 +35,7 @@ void createProducerTask_withJsonFileFormat_shouldReturnJsonProducerTask() {
 
   @Test
   void createProducerTask_withJsonLinesFileFormat_shouldReturnJsonLineProducerTask() {
-    ProducerTaskFactory producerTaskFactory = new ProducerTaskFactory(null, false, false);
+    ProducerTaskFactory producerTaskFactory = new ProducerTaskFactory(null, false);
     Assertions.assertEquals(
         JsonLineProducerTask.class,
         producerTaskFactory
@@ -45,7 +45,7 @@ void createProducerTask_withJsonLinesFileFormat_shouldReturnJsonLineProducerTask
 
   @Test
   void createProducerTask_withCsvFileFormat_shouldReturnCsvProducerTask() {
-    ProducerTaskFactory producerTaskFactory = new ProducerTaskFactory(",", false, false);
+    ProducerTaskFactory producerTaskFactory = new ProducerTaskFactory(",", false);
     Assertions.assertEquals(
         CsvProducerTask.class,
         producerTaskFactory
diff --git a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataimport/ImportManagerTest.java b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataimport/ImportManagerTest.java
index bf348c5321..56e3517ce0 100644
--- a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataimport/ImportManagerTest.java
+++ b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataimport/ImportManagerTest.java
@@ -7,7 +7,6 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 
-import com.scalar.db.api.DistributedStorage;
 import com.scalar.db.api.DistributedTransactionManager;
 import com.scalar.db.api.TableMetadata;
 import com.scalar.db.dataloader.core.ScalarDbMode;
@@ -23,7 +22,6 @@ public class ImportManagerTest {
   private ImportManager importManager;
   private ImportEventListener listener1;
   private ImportEventListener listener2;
-  private DistributedStorage distributedStorage;
   private DistributedTransactionManager distributedTransactionManager;
 
   @BeforeEach
@@ -35,7 +33,6 @@ void setUp() {
 
     listener1 = mock(ImportEventListener.class);
     listener2 = mock(ImportEventListener.class);
-    distributedStorage = mock(DistributedStorage.class);
     distributedTransactionManager = mock(DistributedTransactionManager.class);
 
     importManager =
@@ -45,8 +42,7 @@ void setUp() {
             options,
             processorFactory,
             ScalarDbMode.STORAGE,
-            distributedStorage,
-            null); // Only one resource present
+            distributedTransactionManager);
     importManager.addListener(listener1);
     importManager.addListener(listener2);
   }
@@ -57,7 +53,7 @@ void onAllDataChunksCompleted_shouldNotifyListenersAndCloseStorage() {
 
     verify(listener1).onAllDataChunksCompleted();
     verify(listener2).onAllDataChunksCompleted();
-    verify(distributedStorage).close();
+    verify(distributedTransactionManager).close();
   }
 
   @Test
@@ -69,7 +65,7 @@ void onAllDataChunksCompleted_shouldAggregateListenerExceptionAndStillCloseResou
 
     assertTrue(thrown.getMessage().contains("Error during completion"));
     assertEquals("Listener1 failed", thrown.getCause().getMessage());
-    verify(distributedStorage).close();
+    verify(distributedTransactionManager).close();
   }
 
   @Test
@@ -81,7 +77,6 @@ void closeResources_shouldCloseTransactionManagerIfStorageIsNull() {
             mock(ImportOptions.class),
             mock(ImportProcessorFactory.class),
             ScalarDbMode.TRANSACTION,
-            null,
             distributedTransactionManager);
 
     managerWithTx.closeResources();
@@ -90,7 +85,7 @@ void closeResources_shouldCloseTransactionManagerIfStorageIsNull() {
 
   @Test
   void closeResources_shouldThrowIfResourceCloseFails() {
-    doThrow(new RuntimeException("Close failed")).when(distributedStorage).close();
+    doThrow(new RuntimeException("Close failed")).when(distributedTransactionManager).close();
 
     RuntimeException ex =
         assertThrows(RuntimeException.class, () -> importManager.closeResources());
diff --git a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataimport/processor/CsvImportProcessorTest.java b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataimport/processor/CsvImportProcessorTest.java
index 5457bc51d1..58eeaa8325 100644
--- a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataimport/processor/CsvImportProcessorTest.java
+++ b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataimport/processor/CsvImportProcessorTest.java
@@ -3,7 +3,6 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
-import com.scalar.db.api.DistributedStorage;
 import com.scalar.db.api.DistributedTransaction;
 import com.scalar.db.api.DistributedTransactionManager;
 import com.scalar.db.api.TableMetadata;
@@ -32,7 +31,6 @@ class CsvImportProcessorTest {
   @Mock TableColumnDataTypes tableColumnDataTypes;
 
   ScalarDbDao dao;
-  @Mock DistributedStorage distributedStorage;
   DistributedTransactionManager distributedTransactionManager;
   CsvImportProcessor csvImportProcessor;
 
@@ -65,7 +63,7 @@ void setup() throws ScalarDbDaoException, TransactionException {
                 "table",
                 UnitTestUtils.getPartitionKey(1),
                 UnitTestUtils.getClusteringKey(),
-                distributedStorage))
+                distributedTransactionManager))
         .thenReturn(UnitTestUtils.getResult(1));
     Mockito.when(
             dao.get(
@@ -84,7 +82,6 @@ void test_importProcessWithStorage() {
             .scalarDbMode(ScalarDbMode.STORAGE)
             .importOptions(importOptions)
             .dao(dao)
-            .distributedStorage(distributedStorage)
             .distributedTransactionManager(distributedTransactionManager)
             .tableColumnDataTypes(tableColumnDataTypes)
             .tableMetadataByTableName(tableMetadataByTableName)
@@ -103,7 +100,6 @@ void test_importProcessWithTransaction() {
             .scalarDbMode(ScalarDbMode.TRANSACTION)
             .importOptions(importOptions)
             .dao(dao)
-            .distributedStorage(distributedStorage)
             .distributedTransactionManager(distributedTransactionManager)
             .tableColumnDataTypes(tableColumnDataTypes)
             .tableMetadataByTableName(tableMetadataByTableName)
diff --git a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataimport/processor/ImportProcessorTest.java b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataimport/processor/ImportProcessorTest.java
index d60ebecb00..24c272a45d 100644
--- a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataimport/processor/ImportProcessorTest.java
+++ b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataimport/processor/ImportProcessorTest.java
@@ -10,7 +10,6 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.scalar.db.api.DistributedStorage;
 import com.scalar.db.api.DistributedTransaction;
 import com.scalar.db.api.DistributedTransactionManager;
 import com.scalar.db.api.TableMetadata;
@@ -62,7 +61,6 @@ class ImportProcessorTest {
   @Mock private ImportProcessorParams params;
   @Mock private ImportOptions importOptions;
   @Mock private ScalarDbDao dao;
-  @Mock private DistributedStorage distributedStorage;
   @Mock private DistributedTransactionManager distributedTransactionManager;
   @Mock private DistributedTransaction distributedTransaction;
   @Mock private TableColumnDataTypes tableColumnDataTypes;
@@ -87,7 +85,6 @@ void process_withStorageMode_shouldProcessAllDataChunks() {
     BufferedReader reader = new BufferedReader(new StringReader("test data"));
     when(params.getScalarDbMode()).thenReturn(ScalarDbMode.STORAGE);
     when(params.getDao()).thenReturn(dao);
-    when(params.getDistributedStorage()).thenReturn(distributedStorage);
     when(params.getTableColumnDataTypes()).thenReturn(tableColumnDataTypes);
 
     TestImportProcessor processor = new TestImportProcessor(params);
@@ -150,7 +147,6 @@ void process_withMultipleDataChunks_shouldUseThreadPool() {
     final int maxThreads = 4;
     when(importOptions.getMaxThreads()).thenReturn(maxThreads);
     when(params.getDao()).thenReturn(dao);
-    when(params.getDistributedStorage()).thenReturn(distributedStorage);
     when(params.getTableColumnDataTypes()).thenReturn(tableColumnDataTypes);
     when(params.getTableMetadataByTableName()).thenReturn(tableMetadataByTableName);
 
@@ -205,7 +201,6 @@ void process_withLargeNumberOfTasks_shouldWaitForAllTasksToComplete() {
     final int maxThreads = 2;
     when(importOptions.getMaxThreads()).thenReturn(maxThreads);
     when(params.getDao()).thenReturn(dao);
-    when(params.getDistributedStorage()).thenReturn(distributedStorage);
     when(params.getTableColumnDataTypes()).thenReturn(tableColumnDataTypes);
     when(params.getTableMetadataByTableName()).thenReturn(tableMetadataByTableName);
 
@@ -235,7 +230,6 @@ void process_withShutdown_shouldShutdownExecutorsGracefully() {
     // Arrange
     when(params.getScalarDbMode()).thenReturn(ScalarDbMode.STORAGE);
     when(params.getDao()).thenReturn(dao);
-    when(params.getDistributedStorage()).thenReturn(distributedStorage);
     when(params.getTableColumnDataTypes()).thenReturn(tableColumnDataTypes);
     when(params.getTableMetadataByTableName()).thenReturn(tableMetadataByTableName);
 
diff --git a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataimport/processor/JsonImportProcessorTest.java b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataimport/processor/JsonImportProcessorTest.java
index a5705d3684..2b1bc74ad5 100644
--- a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataimport/processor/JsonImportProcessorTest.java
+++ b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataimport/processor/JsonImportProcessorTest.java
@@ -3,7 +3,6 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
-import com.scalar.db.api.DistributedStorage;
 import com.scalar.db.api.DistributedTransaction;
 import com.scalar.db.api.DistributedTransactionManager;
 import com.scalar.db.api.TableMetadata;
@@ -32,7 +31,6 @@ class JsonImportProcessorTest {
   @Mock TableColumnDataTypes tableColumnDataTypes;
 
   ScalarDbDao dao;
-  @Mock DistributedStorage distributedStorage;
   DistributedTransactionManager distributedTransactionManager;
   JsonImportProcessor jsonImportProcessor;
 
@@ -65,7 +63,7 @@ void setup() throws ScalarDbDaoException, TransactionException {
                 "table",
                 UnitTestUtils.getPartitionKey(1),
                 UnitTestUtils.getClusteringKey(),
-                distributedStorage))
+                distributedTransactionManager))
         .thenReturn(UnitTestUtils.getResult(1));
     Mockito.when(
             dao.get(
@@ -84,7 +82,6 @@ void test_importProcessWithStorage() {
             .scalarDbMode(ScalarDbMode.STORAGE)
             .importOptions(importOptions)
             .dao(dao)
-            .distributedStorage(distributedStorage)
             .distributedTransactionManager(distributedTransactionManager)
             .tableColumnDataTypes(tableColumnDataTypes)
             .tableMetadataByTableName(tableMetadataByTableName)
@@ -103,7 +100,6 @@ void test_importProcessWithTransaction() {
             .scalarDbMode(ScalarDbMode.TRANSACTION)
             .importOptions(importOptions)
             .dao(dao)
-            .distributedStorage(distributedStorage)
             .distributedTransactionManager(distributedTransactionManager)
             .tableColumnDataTypes(tableColumnDataTypes)
             .tableMetadataByTableName(tableMetadataByTableName)
diff --git a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataimport/processor/JsonLinesImportProcessorTest.java b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataimport/processor/JsonLinesImportProcessorTest.java
index 30992f1d35..edb3b8ea2b 100644
--- a/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataimport/processor/JsonLinesImportProcessorTest.java
+++ b/data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataimport/processor/JsonLinesImportProcessorTest.java
@@ -3,7 +3,6 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
-import com.scalar.db.api.DistributedStorage;
 import com.scalar.db.api.DistributedTransaction;
 import com.scalar.db.api.DistributedTransactionManager;
 import com.scalar.db.api.TableMetadata;
@@ -32,7 +31,6 @@ class JsonLinesImportProcessorTest {
   @Mock TableColumnDataTypes tableColumnDataTypes;
 
   ScalarDbDao dao;
-  @Mock DistributedStorage distributedStorage;
   DistributedTransactionManager distributedTransactionManager;
   JsonLinesImportProcessor jsonLinesImportProcessor;
 
@@ -65,7 +63,7 @@ void setup() throws ScalarDbDaoException, TransactionException {
                 "table",
                 UnitTestUtils.getPartitionKey(1),
                 UnitTestUtils.getClusteringKey(),
-                distributedStorage))
+                distributedTransactionManager))
         .thenReturn(UnitTestUtils.getResult(1));
     Mockito.when(
             dao.get(
@@ -84,7 +82,6 @@ void test_importProcessWithStorage() {
             .scalarDbMode(ScalarDbMode.STORAGE)
             .importOptions(importOptions)
             .dao(dao)
-            .distributedStorage(distributedStorage)
             .distributedTransactionManager(distributedTransactionManager)
             .tableColumnDataTypes(tableColumnDataTypes)
             .tableMetadataByTableName(tableMetadataByTableName)
@@ -103,7 +100,6 @@ void test_importProcessWithTransaction() {
             .scalarDbMode(ScalarDbMode.TRANSACTION)
             .importOptions(importOptions)
             .dao(dao)
-            .distributedStorage(distributedStorage)
             .distributedTransactionManager(distributedTransactionManager)
             .tableColumnDataTypes(tableColumnDataTypes)
             .tableMetadataByTableName(tableMetadataByTableName)