Add transaction mode support for data export in dataloader core and CLI

data-loader/cli/src/main/java/com/scalar/db/dataloader/cli/command/dataexport/ExportCommand.java

@@ -5,6 +5,7 @@
 import static java.nio.file.StandardOpenOption.CREATE;
 
 import com.scalar.db.api.DistributedStorage;
+import com.scalar.db.api.DistributedTransactionManager;
 import com.scalar.db.api.TableMetadata;
 import com.scalar.db.common.error.CoreError;
 import com.scalar.db.dataloader.cli.exception.DirectoryValidationException;
@@ -13,6 +14,7 @@
 import com.scalar.db.dataloader.cli.util.InvalidFilePathException;
 import com.scalar.db.dataloader.core.ColumnKeyValue;
 import com.scalar.db.dataloader.core.FileFormat;
+import com.scalar.db.dataloader.core.ScalarDbMode;
 import com.scalar.db.dataloader.core.ScanRange;
 import com.scalar.db.dataloader.core.dataexport.CsvExportManager;
 import com.scalar.db.dataloader.core.dataexport.ExportManager;
@@ -28,7 +30,9 @@
 import com.scalar.db.dataloader.core.util.KeyUtils;
 import com.scalar.db.io.Key;
 import com.scalar.db.service.StorageFactory;
+import com.scalar.db.service.TransactionFactory;
 import java.io.BufferedWriter;
+import java.io.IOException;
 import java.nio.charset.Charset;
 import java.nio.file.Files;
 import java.nio.file.Paths;
@@ -62,14 +66,13 @@ public Integer call() throws Exception {
       validatePositiveValue(
           spec.commandLine(), maxThreads, CoreError.DATA_LOADER_INVALID_MAX_THREADS);
 
-      StorageFactory storageFactory = StorageFactory.create(scalarDbPropertiesFilePath);
-      TableMetadataService metaDataService =
-          new TableMetadataService(storageFactory.getStorageAdmin());
+      TableMetadataService tableMetadataService =
+          createTableMetadataService(scalarDbMode, scalarDbPropertiesFilePath);
       ScalarDbDao scalarDbDao = new ScalarDbDao();
 
-      ExportManager exportManager = createExportManager(storageFactory, scalarDbDao, outputFormat);
-
-      TableMetadata tableMetadata = metaDataService.getTableMetadata(namespace, table);
+      ExportManager exportManager =
+          createExportManager(scalarDbMode, scalarDbDao, outputFormat, scalarDbPropertiesFilePath);
+      TableMetadata tableMetadata = tableMetadataService.getTableMetadata(namespace, table);
 
       Key partitionKey =
           partitionKeyValue != null ? getKeysFromList(partitionKeyValue, tableMetadata) : null;
@@ -127,11 +130,85 @@ private void validateOutputDirectory() throws DirectoryValidationException {
     }
   }
 
+  private TableMetadataService createTableMetadataService(
+      ScalarDbMode scalarDbMode, String scalarDbPropertiesFilePath) throws IOException {
+    if (scalarDbMode.equals(ScalarDbMode.TRANSACTION)) {
+      TransactionFactory transactionFactory = TransactionFactory.create(scalarDbPropertiesFilePath);
+      return new TableMetadataService(transactionFactory.getTransactionAdmin());
+    }
+    StorageFactory storageFactory = StorageFactory.create(scalarDbPropertiesFilePath);
+    return new TableMetadataService(storageFactory.getStorageAdmin());
+  }
+
+  /**
+   * Creates an {@link ExportManager} instance based on ScalarDB mode and file format.
+   *
+   * @param scalarDbMode The ScalarDB mode (TRANSACTION or STORAGE).
+   * @param scalarDbDao The DAO for accessing ScalarDB.
+   * @param fileFormat The output file format (CSV, JSON, JSONL).
+   * @param scalarDbPropertiesFilePath Path to the ScalarDB properties file.
+   * @return A configured {@link ExportManager}.
+   * @throws IOException If there is an error reading the properties file.
+   */
   private ExportManager createExportManager(
-      StorageFactory storageFactory, ScalarDbDao scalarDbDao, FileFormat fileFormat) {
+      ScalarDbMode scalarDbMode,
+      ScalarDbDao scalarDbDao,
+      FileFormat fileFormat,
+      String scalarDbPropertiesFilePath)
+      throws IOException {
     ProducerTaskFactory taskFactory =
         new ProducerTaskFactory(delimiter, includeTransactionMetadata, prettyPrintJson);
-    DistributedStorage storage = storageFactory.getStorage();
+    if (scalarDbMode.equals(ScalarDbMode.STORAGE)) {
+      DistributedStorage storage = StorageFactory.create(scalarDbPropertiesFilePath).getStorage();
+      return createExportManagerWithStorage(storage, scalarDbDao, fileFormat, taskFactory);
+    } else {
+      DistributedTransactionManager distributedTransactionManager =
+          TransactionFactory.create(scalarDbPropertiesFilePath).getTransactionManager();
+      return createExportManagerWithTransaction(
+          distributedTransactionManager, scalarDbDao, fileFormat, taskFactory);
+    }
+  }
+
+  /**
+   * Returns an {@link ExportManager} that uses {@link DistributedTransactionManager}.
+   *
+   * @param distributedTransactionManager distributed transaction manager object
+   * @param scalarDbDao The DAO for accessing ScalarDB.
+   * @param fileFormat The output file format (CSV, JSON, JSONL).
+   * @param taskFactory Producer task factory object
+   * @return A configured {@link ExportManager}.
+   */
+  private ExportManager createExportManagerWithTransaction(
+      DistributedTransactionManager distributedTransactionManager,
+      ScalarDbDao scalarDbDao,
+      FileFormat fileFormat,
+      ProducerTaskFactory taskFactory) {
+    switch (fileFormat) {
+      case JSON:
+        return new JsonExportManager(distributedTransactionManager, scalarDbDao, taskFactory);
+      case JSONL:
+        return new JsonLineExportManager(distributedTransactionManager, scalarDbDao, taskFactory);
+      case CSV:
+        return new CsvExportManager(distributedTransactionManager, scalarDbDao, taskFactory);
+      default:
+        throw new AssertionError("Invalid file format" + fileFormat);
+    }
+  }
+
+  /**
+   * Returns an {@link ExportManager} that uses {@link DistributedStorage}.
+   *
+   * @param storage distributed storage object
+   * @param scalarDbDao The DAO for accessing ScalarDB.
+   * @param fileFormat The output file format (CSV, JSON, JSONL).
+   * @param taskFactory Producer task factory object
+   * @return A configured {@link ExportManager}.
+   */
+  private ExportManager createExportManagerWithStorage(
+      DistributedStorage storage,
+      ScalarDbDao scalarDbDao,
+      FileFormat fileFormat,
+      ProducerTaskFactory taskFactory) {
     switch (fileFormat) {
       case JSON:
         return new JsonExportManager(storage, scalarDbDao, taskFactory);
@@ -155,7 +232,8 @@ private ExportOptions buildExportOptions(Key partitionKey, ScanRange scanRange)
             .maxThreadCount(maxThreads)
             .dataChunkSize(dataChunkSize)
             .prettyPrintJson(prettyPrintJson)
-            .scanRange(scanRange);
+            .scanRange(scanRange)
+            .scalarDbMode(scalarDbMode);
 
     if (projectionColumns != null) {
       builder.projectionColumns(projectionColumns);

data-loader/cli/src/main/java/com/scalar/db/dataloader/cli/command/dataexport/ExportCommandOptions.java

@@ -3,6 +3,7 @@
 import com.scalar.db.api.Scan;
 import com.scalar.db.dataloader.core.ColumnKeyValue;
 import com.scalar.db.dataloader.core.FileFormat;
+import com.scalar.db.dataloader.core.ScalarDbMode;
 import java.util.ArrayList;
 import java.util.List;
 import picocli.CommandLine;
@@ -144,4 +145,11 @@ public class ExportCommandOptions {
       description = "Size of the data chunk to process in a single task (default: 200)",
       defaultValue = "200")
   protected int dataChunkSize;
+
+  @CommandLine.Option(
+      names = {"--mode", "-sm"},
+      description = "ScalarDB mode (STORAGE, TRANSACTION) (default: STORAGE)",
+      paramLabel = "<MODE>",
+      defaultValue = "STORAGE")
+  protected ScalarDbMode scalarDbMode;
 }

data-loader/cli/src/test/java/com/scalar/db/dataloader/cli/command/dataexport/ExportCommandTest.java

@@ -4,6 +4,7 @@
 
 import com.scalar.db.common.error.CoreError;
 import com.scalar.db.dataloader.core.FileFormat;
+import com.scalar.db.dataloader.core.ScalarDbMode;
 import java.io.File;
 import java.nio.file.Paths;
 import org.junit.jupiter.api.AfterEach;
@@ -55,4 +56,19 @@ void call_withInvalidScalarDBConfigurationFile_shouldReturnOne() throws Exceptio
     exportCommand.outputFormat = FileFormat.JSON;
     Assertions.assertEquals(1, exportCommand.call());
   }
+
+  @Test
+  void call_withScalarDBModeTransaction_WithInvalidConfigurationFile_shouldReturnOne()
+      throws Exception {
+    ExportCommand exportCommand = new ExportCommand();
+    exportCommand.configFilePath = "scalardb.properties";
+    exportCommand.dataChunkSize = 100;
+    exportCommand.namespace = "scalar";
+    exportCommand.table = "asset";
+    exportCommand.outputDirectory = "";
+    exportCommand.outputFileName = "sample.json";
+    exportCommand.outputFormat = FileFormat.JSON;
+    exportCommand.scalarDbMode = ScalarDbMode.TRANSACTION;
+    Assertions.assertEquals(1, exportCommand.call());
+  }
 }

data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataexport/CsvExportManager.java

@@ -1,6 +1,7 @@
 package com.scalar.db.dataloader.core.dataexport;
 
 import com.scalar.db.api.DistributedStorage;
+import com.scalar.db.api.DistributedTransactionManager;
 import com.scalar.db.api.TableMetadata;
 import com.scalar.db.dataloader.core.dataexport.producer.ProducerTaskFactory;
 import com.scalar.db.dataloader.core.dataimport.dao.ScalarDbDao;
@@ -13,8 +14,17 @@
 
 public class CsvExportManager extends ExportManager {
   public CsvExportManager(
-      DistributedStorage storage, ScalarDbDao dao, ProducerTaskFactory producerTaskFactory) {
-    super(storage, dao, producerTaskFactory);
+      DistributedStorage distributedStorage,
+      ScalarDbDao dao,
+      ProducerTaskFactory producerTaskFactory) {
+    super(distributedStorage, dao, producerTaskFactory);
+  }
+
+  public CsvExportManager(
+      DistributedTransactionManager distributedTransactionManager,
+      ScalarDbDao dao,
+      ProducerTaskFactory producerTaskFactory) {
+    super(distributedTransactionManager, dao, producerTaskFactory);
   }
 
   /**