Merge branch 'master' into fix/data-loader/raw-record-issue

Author: inv-jishnu

data-loader/cli/src/main/java/com/scalar/db/dataloader/cli/command/dataimport/ImportCommand.java

@@ -4,10 +4,12 @@
 import static com.scalar.db.dataloader.cli.util.CommandLineInputUtils.validatePositiveValue;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.scalar.db.api.DistributedTransaction;
 import com.scalar.db.api.DistributedTransactionAdmin;
 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;
@@ -67,20 +69,32 @@ public Integer call() throws Exception {
         spec.commandLine(), dataChunkQueueSize, DataLoaderError.INVALID_DATA_CHUNK_QUEUE_SIZE);
     ControlFile controlFile = parseControlFileFromPath(controlFilePath).orElse(null);
     ImportOptions importOptions = createImportOptions(controlFile);
-    ImportLoggerConfig config =
+    ImportLoggerConfig importLoggerConfig =
         ImportLoggerConfig.builder()
             .logDirectoryPath(logDirectory)
             .isLogRawSourceRecordsEnabled(importOptions.isLogRawRecord())
             .isLogSuccessRecordsEnabled(importOptions.isLogSuccessRecords())
             .prettyPrint(prettyPrint)
             .build();
-    LogWriterFactory logWriterFactory = createLogWriterFactory(config);
+    LogWriterFactory logWriterFactory = createLogWriterFactory(importLoggerConfig);
+    File configFile = new File(configFilePath);
+    TransactionFactory transactionFactory = TransactionFactory.create(configFile);
+
+    // Validate transaction mode configuration before proceeding
+    validateTransactionMode(transactionFactory);
+
     Map<String, TableMetadata> tableMetadataMap =
-        createTableMetadataMap(controlFile, namespace, tableName);
+        createTableMetadataMap(controlFile, namespace, tableName, transactionFactory);
     try (BufferedReader reader =
         Files.newBufferedReader(Paths.get(sourceFilePath), Charset.defaultCharset())) {
       ImportManager importManager =
-          createImportManager(importOptions, tableMetadataMap, reader, logWriterFactory, config);
+          createImportManager(
+              importOptions,
+              tableMetadataMap,
+              reader,
+              importLoggerConfig,
+              logWriterFactory,
+              transactionFactory);
       importManager.startImport();
     }
     return 0;
@@ -101,14 +115,16 @@ private LogWriterFactory createLogWriterFactory(ImportLoggerConfig config) {
    * @param controlFile control file
    * @param namespace Namespace
    * @param tableName Single table name
+   * @param transactionFactory transaction factory to use
    * @return {@code Map<String, TableMetadata>} a table metadata map
    * @throws ParameterException if one of the argument values is wrong
    */
   private Map<String, TableMetadata> createTableMetadataMap(
-      ControlFile controlFile, String namespace, String tableName)
-      throws IOException, TableMetadataException {
-    File configFile = new File(configFilePath);
-    TransactionFactory transactionFactory = TransactionFactory.create(configFile);
+      ControlFile controlFile,
+      String namespace,
+      String tableName,
+      TransactionFactory transactionFactory)
+      throws TableMetadataException {
     try (DistributedTransactionAdmin transactionAdmin = transactionFactory.getTransactionAdmin()) {
       TableMetadataService tableMetadataService = new TableMetadataService(transactionAdmin);
       Map<String, TableMetadata> tableMetadataMap = new HashMap<>();
@@ -133,18 +149,19 @@ private Map<String, TableMetadata> createTableMetadataMap(
    * @param importOptions import options
    * @param tableMetadataMap table metadata map
    * @param reader buffered reader with source data
+   * @param importLoggerConfig import logging config
    * @param logWriterFactory log writer factory object
-   * @param config import logging config
+   * @param transactionFactory transaction factory to use
    * @return ImportManager object
    */
   private ImportManager createImportManager(
       ImportOptions importOptions,
       Map<String, TableMetadata> tableMetadataMap,
       BufferedReader reader,
+      ImportLoggerConfig importLoggerConfig,
       LogWriterFactory logWriterFactory,
-      ImportLoggerConfig config)
+      TransactionFactory transactionFactory)
       throws IOException {
-    File configFile = new File(configFilePath);
     ImportProcessorFactory importProcessorFactory = new DefaultImportProcessorFactory();
     ImportManager importManager =
         new ImportManager(
@@ -153,11 +170,12 @@ private ImportManager createImportManager(
             importOptions,
             importProcessorFactory,
             scalarDbMode,
-            TransactionFactory.create(configFile).getTransactionManager());
+            transactionFactory.getTransactionManager());
     if (importOptions.getLogMode().equals(LogMode.SPLIT_BY_DATA_CHUNK)) {
-      importManager.addListener(new SplitByDataChunkImportLogger(config, logWriterFactory));
+      importManager.addListener(
+          new SplitByDataChunkImportLogger(importLoggerConfig, logWriterFactory));
     } else {
-      importManager.addListener(new SingleFileImportLogger(config, logWriterFactory));
+      importManager.addListener(new SingleFileImportLogger(importLoggerConfig, logWriterFactory));
     }
     return importManager;
   }
@@ -236,6 +254,48 @@ private void validateLogDirectory(String logDirectory) throws ParameterException
     }
   }
 
+  /**
+   * Validate transaction mode configuration by attempting to start and abort a transaction
+   *
+   * @param transactionFactory transaction factory to test
+   * @throws ParameterException if transaction mode is incompatible with the configured transaction
+   *     manager
+   */
+  void validateTransactionMode(TransactionFactory transactionFactory) {
+    // Only validate when in TRANSACTION mode
+    if (scalarDbMode != ScalarDbMode.TRANSACTION) {
+      return;
+    }
+
+    DistributedTransaction transaction = null;
+    try {
+      // Try to start a read only transaction to verify the transaction manager is properly
+      // configured
+      transaction = transactionFactory.getTransactionManager().startReadOnly();
+    } catch (UnsupportedOperationException e) {
+      // Transaction mode is not supported by the configured transaction manager
+      throw new ParameterException(
+          spec.commandLine(),
+          DataLoaderError.INVALID_TRANSACTION_MODE.buildMessage(e.getMessage()),
+          e);
+    } catch (Exception e) {
+      // Other exceptions - configuration or runtime error
+      throw new ParameterException(
+          spec.commandLine(),
+          DataLoaderError.TRANSACTION_MODE_VALIDATION_FAILED.buildMessage(e.getMessage()),
+          e);
+    } finally {
+      // Ensure transaction is aborted
+      if (transaction != null) {
+        try {
+          transaction.abort();
+        } catch (Exception ignored) {
+          // Ignore errors during cleanup
+        }
+      }
+    }
+  }
+
   /**
    * Generate control file from a valid control file path
    *

data-loader/cli/src/test/java/com/scalar/db/dataloader/cli/command/dataimport/ImportCommandTest.java

@@ -3,9 +3,14 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
+import com.scalar.db.api.DistributedTransactionManager;
 import com.scalar.db.dataloader.core.FileFormat;
+import com.scalar.db.dataloader.core.ScalarDbMode;
 import com.scalar.db.dataloader.core.dataimport.ImportMode;
+import com.scalar.db.service.TransactionFactory;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -279,4 +284,52 @@ void call_withoutMaxThreads_shouldDefaultToAvailableProcessors() {
     // Verify it was set to available processors
     assertEquals(Runtime.getRuntime().availableProcessors(), command.maxThreads);
   }
+
+  @Test
+  void validateTransactionMode_withUnsupportedOperation_shouldThrowException() throws Exception {
+    // Arrange - Mock TransactionFactory to throw UnsupportedOperationException
+    TransactionFactory mockFactory = mock(TransactionFactory.class);
+    DistributedTransactionManager mockManager = mock(DistributedTransactionManager.class);
+
+    when(mockFactory.getTransactionManager()).thenReturn(mockManager);
+    when(mockManager.startReadOnly())
+        .thenThrow(new UnsupportedOperationException("Transaction mode is not supported"));
+
+    ImportCommand command = new ImportCommand();
+    CommandLine cmd = new CommandLine(command);
+    command.spec = cmd.getCommandSpec();
+    command.scalarDbMode = ScalarDbMode.TRANSACTION;
+
+    // Act & Assert
+    CommandLine.ParameterException thrown =
+        assertThrows(
+            CommandLine.ParameterException.class,
+            () -> command.validateTransactionMode(mockFactory));
+
+    assertTrue(thrown.getMessage().contains("TRANSACTION mode is not compatible"));
+  }
+
+  @Test
+  void validateTransactionMode_withOtherException_shouldThrowException() throws Exception {
+    // Arrange - Mock TransactionFactory to throw a different exception
+    TransactionFactory mockFactory = mock(TransactionFactory.class);
+    DistributedTransactionManager mockManager = mock(DistributedTransactionManager.class);
+
+    when(mockFactory.getTransactionManager()).thenReturn(mockManager);
+    when(mockManager.startReadOnly()).thenThrow(new RuntimeException("Connection failed"));
+
+    ImportCommand command = new ImportCommand();
+    CommandLine cmd = new CommandLine(command);
+    command.spec = cmd.getCommandSpec();
+    command.scalarDbMode = ScalarDbMode.TRANSACTION;
+
+    // Act & Assert
+    CommandLine.ParameterException thrown =
+        assertThrows(
+            CommandLine.ParameterException.class,
+            () -> command.validateTransactionMode(mockFactory));
+
+    assertTrue(thrown.getMessage().contains("Failed to validate TRANSACTION mode"));
+    assertTrue(thrown.getMessage().contains("Connection failed"));
+  }
 }

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

@@ -216,6 +216,14 @@ public enum DataLoaderError implements ScalarDbError {
       "Cannot specify both deprecated option '%s' and new option '%s'. Please use only '%s'",
       "",
       ""),
+  INVALID_TRANSACTION_MODE(
+      Category.USER_ERROR,
+      "0058",
+      "TRANSACTION mode is not compatible with the current configuration. Please try with STORAGE mode or check your ScalarDB configuration. Details: %s",
+      "",
+      ""),
+  TRANSACTION_MODE_VALIDATION_FAILED(
+      Category.USER_ERROR, "0059", "Failed to validate TRANSACTION mode. Details: %s", "", ""),
 
   //
   // Errors for the internal error category

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

@@ -80,11 +80,11 @@ public ExportReport startExport(
       ExecutorService executorService =
           Executors.newFixedThreadPool(exportOptions.getMaxThreadCount());
 
-      BufferedWriter bufferedWriter = new BufferedWriter(writer);
       boolean isJson = exportOptions.getOutputFileFormat() == FileFormat.JSON;
 
       try (TransactionManagerCrudOperable.Scanner scanner =
-          createScanner(exportOptions, dao, manager)) {
+              createScanner(exportOptions, dao, manager);
+          BufferedWriter bufferedWriter = new BufferedWriter(writer)) {
 
         Iterator<Result> iterator = scanner.iterator();
         AtomicBoolean isFirstBatch = new AtomicBoolean(true);
@@ -106,18 +106,20 @@ public ExportReport startExport(
         executorService.shutdown();
         if (executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS)) {
           logger.info("All tasks completed");
+          processFooter(exportOptions, tableMetadata, bufferedWriter);
         } else {
           logger.error("Timeout occurred while waiting for tasks to complete");
           // TODO: handle this
         }
-        processFooter(exportOptions, tableMetadata, bufferedWriter);
       } catch (InterruptedException
           | IOException
           | UnknownTransactionStatusException
           | CrudException e) {
         logger.error("Error during export: ", e);
       } finally {
-        bufferedWriter.flush();
+        if (!executorService.isShutdown()) {
+          executorService.shutdownNow();
+        }
       }
     } catch (ExportOptionsValidationException | IOException | ScalarDbDaoException e) {
       logger.error("Error during export: {}", e.getMessage());

data-loader/core/src/main/java/com/scalar/db/dataloader/core/util/ColumnUtils.java

@@ -83,6 +83,9 @@ public static Column<?> createColumnFromValue(
       DataType dataType, ColumnInfo columnInfo, @Nullable String value)
       throws ColumnParsingException {
     String columnName = columnInfo.getColumnName();
+    if (value != null && !dataType.equals(DataType.TEXT) && value.equalsIgnoreCase("null")) {
+      value = null;
+    }
     try {
       switch (dataType) {
         case BOOLEAN:

data-loader/core/src/test/java/com/scalar/db/dataloader/core/util/ColumnUtilsTest.java

@@ -228,4 +228,85 @@ void getColumnsFromResult_withResultNull_withValidData_shouldReturnColumns()
             null, sourceRecord, false, mockMetadata, "namespace", "table");
     assertEquals(8, columns.size());
   }
+
+  /**
+   * Tests that the string "null" (lowercase) is correctly treated as null for numeric, boolean, and
+   * date/time types.
+   */
+  @Test
+  void createColumnFromValue_valueIsLowercaseNull_shouldReturnNullColumn()
+      throws ColumnParsingException {
+    String columnName = "testColumn";
+    ColumnInfo columnInfo = ColumnInfo.builder().columnName(columnName).build();
+
+    // Integer type
+    Column<?> intColumn = ColumnUtils.createColumnFromValue(DataType.INT, columnInfo, "null");
+    assertEquals(IntColumn.ofNull(columnName), intColumn);
+    intColumn = ColumnUtils.createColumnFromValue(DataType.INT, columnInfo, "Null");
+    assertEquals(IntColumn.ofNull(columnName), intColumn);
+
+    // Double type
+    Column<?> doubleColumn = ColumnUtils.createColumnFromValue(DataType.DOUBLE, columnInfo, "null");
+    assertEquals(DoubleColumn.ofNull(columnName), doubleColumn);
+    doubleColumn = ColumnUtils.createColumnFromValue(DataType.DOUBLE, columnInfo, "NULL");
+    assertEquals(DoubleColumn.ofNull(columnName), doubleColumn);
+
+    // Boolean type
+    Column<?> boolColumn = ColumnUtils.createColumnFromValue(DataType.BOOLEAN, columnInfo, "null");
+    assertEquals(BooleanColumn.ofNull(columnName), boolColumn);
+
+    boolColumn = ColumnUtils.createColumnFromValue(DataType.BOOLEAN, columnInfo, "nuLL");
+    assertEquals(BooleanColumn.ofNull(columnName), boolColumn);
+
+    // Date type
+    Column<?> dateColumn = ColumnUtils.createColumnFromValue(DataType.DATE, columnInfo, "null");
+    assertEquals(DateColumn.ofNull(columnName), dateColumn);
+    dateColumn = ColumnUtils.createColumnFromValue(DataType.DATE, columnInfo, "NULL");
+    assertEquals(DateColumn.ofNull(columnName), dateColumn);
+
+    // Time type
+    Column<?> timeColumn = ColumnUtils.createColumnFromValue(DataType.TIME, columnInfo, "null");
+    assertEquals(TimeColumn.ofNull(columnName), timeColumn);
+
+    timeColumn = ColumnUtils.createColumnFromValue(DataType.TIME, columnInfo, "nuLL");
+    assertEquals(TimeColumn.ofNull(columnName), timeColumn);
+
+    // Timestamp type
+    Column<?> timestampColumn =
+        ColumnUtils.createColumnFromValue(DataType.TIMESTAMP, columnInfo, "null");
+    assertEquals(TimestampColumn.ofNull(columnName), timestampColumn);
+    timestampColumn = ColumnUtils.createColumnFromValue(DataType.TIMESTAMP, columnInfo, "NULL");
+    assertEquals(TimestampColumn.ofNull(columnName), timestampColumn);
+
+    // Timestamp type
+    Column<?> timestamprtzColumn =
+        ColumnUtils.createColumnFromValue(DataType.TIMESTAMPTZ, columnInfo, "null");
+    assertEquals(TimestampTZColumn.ofNull(columnName), timestamprtzColumn);
+    timestamprtzColumn =
+        ColumnUtils.createColumnFromValue(DataType.TIMESTAMPTZ, columnInfo, "Null");
+    assertEquals(TimestampTZColumn.ofNull(columnName), timestamprtzColumn);
+  }
+
+  /**
+   * Tests that when the string value "null" is provided for TEXT columns, it is treated as a
+   * literal string and not converted to null.
+   */
+  @Test
+  void createColumnFromValue_valueIsNullString_shouldRemainLiteralForTextType()
+      throws ColumnParsingException {
+    String columnName = "textColumn";
+    ColumnInfo columnInfo = ColumnInfo.builder().columnName(columnName).build();
+
+    Column<?> textCol = ColumnUtils.createColumnFromValue(DataType.TEXT, columnInfo, "null");
+    assertEquals(TextColumn.of(columnName, "null"), textCol);
+
+    textCol = ColumnUtils.createColumnFromValue(DataType.TEXT, columnInfo, "NULL");
+    assertEquals(TextColumn.of(columnName, "NULL"), textCol);
+
+    textCol = ColumnUtils.createColumnFromValue(DataType.TEXT, columnInfo, "Null");
+    assertEquals(TextColumn.of(columnName, "Null"), textCol);
+
+    textCol = ColumnUtils.createColumnFromValue(DataType.TEXT, columnInfo, "nuLL");
+    assertEquals(TextColumn.of(columnName, "nuLL"), textCol);
+  }
 }