Backport to branch(3.13) : Add table metadata service (#2504)

Co-authored-by: inv-jishnu <[email protected]>
Co-authored-by: Jishnu J <[email protected]>

Author: feeblefakie

core/src/main/java/com/scalar/db/common/error/CoreError.java

@@ -700,6 +700,10 @@ public enum CoreError implements ScalarDbError {
       "The provided partition key order does not match the table schema. Required order: %s",
       "",
       ""),
+  DATA_LOADER_MISSING_NAMESPACE_OR_TABLE(
+      Category.USER_ERROR, "0165", "Missing namespace or table: %s, %s", "", ""),
+  DATA_LOADER_TABLE_METADATA_RETRIEVAL_FAILED(
+      Category.USER_ERROR, "0166", "Failed to retrieve table metadata. Details: %s", "", ""),
 
   //
   // Errors for the concurrency error category

data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/controlfile/ControlFileTable.java

@@ -0,0 +1,64 @@
+package com.scalar.db.dataloader.core.dataimport.controlfile;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.ArrayList;
+import java.util.List;
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * Represents the configuration for a single table in the control file, including its namespace,
+ * table name, and field mappings. This class is used to define how data from a control file maps to
+ * a specific table in ScalarDB.
+ */
+@Getter
+@Setter
+public class ControlFileTable {
+
+  /** The namespace of the table in ScalarDB. */
+  @JsonProperty("namespace")
+  private String namespace;
+
+  /** The name of the table in ScalarDB. */
+  @JsonProperty("table")
+  private String table;
+
+  /**
+   * A list of mappings defining the correspondence between control file fields and table columns.
+   */
+  @JsonProperty("mappings")
+  private final List<ControlFileTableFieldMapping> mappings;
+
+  /**
+   * Creates a new {@code ControlFileTable} instance with the specified namespace and table name.
+   * The mappings list is initialized as an empty list.
+   *
+   * @param namespace The namespace of the table in ScalarDB.
+   * @param table The name of the table in ScalarDB.
+   */
+  public ControlFileTable(String namespace, String table) {
+    this.namespace = namespace;
+    this.table = table;
+    this.mappings = new ArrayList<>();
+  }
+
+  /**
+   * Constructs a {@code ControlFileTable} instance using data from a serialized JSON object. This
+   * constructor is used for deserialization of API requests or control files.
+   *
+   * @param namespace The namespace of the table in ScalarDB.
+   * @param table The name of the table in ScalarDB.
+   * @param mappings A list of mappings that define the relationship between control file fields and
+   *     table columns.
+   */
+  @JsonCreator
+  public ControlFileTable(
+      @JsonProperty("namespace") String namespace,
+      @JsonProperty("table") String table,
+      @JsonProperty("mappings") List<ControlFileTableFieldMapping> mappings) {
+    this.namespace = namespace;
+    this.table = table;
+    this.mappings = mappings;
+  }
+}

data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/controlfile/ControlFileTableFieldMapping.java

@@ -0,0 +1,39 @@
+package com.scalar.db.dataloader.core.dataimport.controlfile;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * Represents the mapping of a single field in the control file to a column in a ScalarDB table.
+ * This class defines how data from a specific field in the input source should be mapped to the
+ * corresponding column in the database.
+ */
+@Getter
+@Setter
+public class ControlFileTableFieldMapping {
+
+  /** The name of the field in the input source (e.g., JSON or CSV). */
+  @JsonProperty("source_field")
+  private String sourceField;
+
+  /** The name of the column in the ScalarDB table that the field maps to. */
+  @JsonProperty("target_column")
+  private String targetColumn;
+
+  /**
+   * Constructs a {@code ControlFileTableFieldMapping} instance using data from a serialized JSON
+   * object. This constructor is primarily used for deserialization of control file mappings.
+   *
+   * @param sourceField The name of the field in the input source (e.g., JSON or CSV).
+   * @param targetColumn The name of the corresponding column in the ScalarDB table.
+   */
+  @JsonCreator
+  public ControlFileTableFieldMapping(
+      @JsonProperty("source_field") String sourceField,
+      @JsonProperty("target_column") String targetColumn) {
+    this.sourceField = sourceField;
+    this.targetColumn = targetColumn;
+  }
+}

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

@@ -0,0 +1,24 @@
+package com.scalar.db.dataloader.core.tablemetadata;
+
+/** A custom exception that encapsulates errors thrown by the TableMetaDataService */
+public class TableMetadataException extends Exception {
+
+  /**
+   * Class constructor
+   *
+   * @param message error message
+   * @param cause reason for exception
+   */
+  public TableMetadataException(String message, Throwable cause) {
+    super(message, cause);
+  }
+
+  /**
+   * Class constructor
+   *
+   * @param message error message
+   */
+  public TableMetadataException(String message) {
+    super(message);
+  }
+}

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

@@ -0,0 +1,22 @@
+package com.scalar.db.dataloader.core.tablemetadata;
+
+import lombok.Getter;
+
+/** Represents the request for metadata for a single ScalarDB table */
+@Getter
+public class TableMetadataRequest {
+
+  private final String namespace;
+  private final String table;
+
+  /**
+   * Class constructor
+   *
+   * @param namespace ScalarDB namespace
+   * @param table ScalarDB table name
+   */
+  public TableMetadataRequest(String namespace, String table) {
+    this.namespace = namespace;
+    this.table = table;
+  }
+}

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

@@ -0,0 +1,73 @@
+package com.scalar.db.dataloader.core.tablemetadata;
+
+import com.scalar.db.api.DistributedStorageAdmin;
+import com.scalar.db.api.TableMetadata;
+import com.scalar.db.common.error.CoreError;
+import com.scalar.db.dataloader.core.util.TableMetadataUtil;
+import com.scalar.db.exception.storage.ExecutionException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import lombok.RequiredArgsConstructor;
+
+/**
+ * Service for retrieving {@link TableMetadata} from ScalarDB. Provides methods to fetch metadata
+ * for individual tables or a collection of tables.
+ */
+@RequiredArgsConstructor
+public class TableMetadataService {
+
+  private final DistributedStorageAdmin storageAdmin;
+
+  /**
+   * Retrieves the {@link TableMetadata} for a specific namespace and table name.
+   *
+   * @param namespace The ScalarDB namespace.
+   * @param tableName The name of the table within the specified namespace.
+   * @return The {@link TableMetadata} object containing schema details of the specified table.
+   * @throws TableMetadataException If the table or namespace does not exist, or if an error occurs
+   *     while fetching the metadata.
+   */
+  public TableMetadata getTableMetadata(String namespace, String tableName)
+      throws TableMetadataException {
+    try {
+      TableMetadata tableMetadata = storageAdmin.getTableMetadata(namespace, tableName);
+      if (tableMetadata == null) {
+        throw new TableMetadataException(
+            CoreError.DATA_LOADER_MISSING_NAMESPACE_OR_TABLE.buildMessage(namespace, tableName));
+      }
+      return tableMetadata;
+    } catch (ExecutionException e) {
+      throw new TableMetadataException(
+          CoreError.DATA_LOADER_TABLE_METADATA_RETRIEVAL_FAILED.buildMessage(e.getMessage()), e);
+    }
+  }
+
+  /**
+   * Retrieves the {@link TableMetadata} for a collection of table metadata requests.
+   *
+   * <p>Each request specifies a namespace and table name. The method consolidates the metadata into
+   * a map keyed by a unique lookup key generated for each table.
+   *
+   * @param requests A collection of {@link TableMetadataRequest} objects specifying the tables to
+   *     retrieve metadata for.
+   * @return A map where the keys are unique lookup keys (namespace + table name) and the values are
+   *     the corresponding {@link TableMetadata} objects.
+   * @throws TableMetadataException If any of the requested tables or namespaces are missing, or if
+   *     an error occurs while fetching the metadata.
+   */
+  public Map<String, TableMetadata> getTableMetadata(Collection<TableMetadataRequest> requests)
+      throws TableMetadataException {
+    Map<String, TableMetadata> metadataMap = new HashMap<>();
+
+    for (TableMetadataRequest request : requests) {
+      String namespace = request.getNamespace();
+      String tableName = request.getTable();
+      TableMetadata tableMetadata = getTableMetadata(namespace, tableName);
+      String key = TableMetadataUtil.getTableLookupKey(namespace, tableName);
+      metadataMap.put(key, tableMetadata);
+    }
+
+    return metadataMap;
+  }
+}

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

@@ -0,0 +1,73 @@
+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;
+
+/** Utility class for handling ScalarDB table metadata operations. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class TableMetadataUtil {
+
+  /**
+   * Generates a unique lookup key for a table within a namespace.
+   *
+   * @param namespace The namespace of the table.
+   * @param tableName The name of the table.
+   * @return A formatted string representing the table lookup key.
+   */
+  public static String getTableLookupKey(String namespace, String tableName) {
+    return String.format(Constants.TABLE_LOOKUP_KEY_FORMAT, namespace, tableName);
+  }
+
+  /**
+   * Generates a unique lookup key for a table using control file table data.
+   *
+   * @param controlFileTable The control file table object containing namespace and table name.
+   * @return A formatted string representing the table lookup key.
+   */
+  public static String getTableLookupKey(ControlFileTable controlFileTable) {
+    return String.format(
+        Constants.TABLE_LOOKUP_KEY_FORMAT,
+        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);
+  }
+}

data-loader/core/src/test/java/com/scalar/db/dataloader/core/tablemetadata/TableMetadataServiceTest.java

@@ -0,0 +1,53 @@
+package com.scalar.db.dataloader.core.tablemetadata;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.scalar.db.api.DistributedStorageAdmin;
+import com.scalar.db.api.TableMetadata;
+import com.scalar.db.common.error.CoreError;
+import com.scalar.db.dataloader.core.UnitTestUtils;
+import com.scalar.db.exception.storage.ExecutionException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+class TableMetadataServiceTest {
+
+  DistributedStorageAdmin storageAdmin;
+  TableMetadataService tableMetadataService;
+
+  @BeforeEach
+  void setup() throws ExecutionException {
+    storageAdmin = Mockito.mock(DistributedStorageAdmin.class);
+    Mockito.when(storageAdmin.getTableMetadata("namespace", "table"))
+        .thenReturn(UnitTestUtils.createTestTableMetadata());
+    tableMetadataService = new TableMetadataService(storageAdmin);
+  }
+
+  @Test
+  void getTableMetadata_withValidNamespaceAndTable_shouldReturnTableMetadataMap()
+      throws TableMetadataException {
+
+    Map<String, TableMetadata> expected = new HashMap<>();
+    expected.put("namespace.table", UnitTestUtils.createTestTableMetadata());
+    TableMetadataRequest tableMetadataRequest = new TableMetadataRequest("namespace", "table");
+    Map<String, TableMetadata> output =
+        tableMetadataService.getTableMetadata(Collections.singleton(tableMetadataRequest));
+    Assertions.assertEquals(expected.get("namespace.table"), output.get("namespace.table"));
+  }
+
+  @Test
+  void getTableMetadata_withInvalidNamespaceAndTable_shouldThrowException() {
+    TableMetadataRequest tableMetadataRequest = new TableMetadataRequest("namespace2", "table2");
+    assertThatThrownBy(
+            () ->
+                tableMetadataService.getTableMetadata(Collections.singleton(tableMetadataRequest)))
+        .isInstanceOf(TableMetadataException.class)
+        .hasMessage(
+            CoreError.DATA_LOADER_MISSING_NAMESPACE_OR_TABLE.buildMessage("namespace2", "table2"));
+  }
+}

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

@@ -0,0 +1,29 @@
+package com.scalar.db.dataloader.core.util;
+
+import static com.scalar.db.dataloader.core.Constants.TABLE_LOOKUP_KEY_FORMAT;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.scalar.db.dataloader.core.dataimport.controlfile.ControlFileTable;
+import org.junit.jupiter.api.Test;
+
+/** Unit tests for TableMetadataUtils */
+class TableMetadataUtilTest {
+
+  private static final String NAMESPACE = "ns";
+  private static final String TABLE_NAME = "table";
+
+  @Test
+  void getTableLookupKey_ValidStringArgs_ShouldReturnLookupKey() {
+    String actual = TableMetadataUtil.getTableLookupKey(NAMESPACE, TABLE_NAME);
+    String expected = String.format(TABLE_LOOKUP_KEY_FORMAT, NAMESPACE, TABLE_NAME);
+    assertThat(actual).isEqualTo(expected);
+  }
+
+  @Test
+  void getTableLookupKey_ValidControlFileArg_ShouldReturnLookupKey() {
+    ControlFileTable controlFileTable = new ControlFileTable(NAMESPACE, TABLE_NAME);
+    String actual = TableMetadataUtil.getTableLookupKey(controlFileTable);
+    String expected = String.format(TABLE_LOOKUP_KEY_FORMAT, NAMESPACE, TABLE_NAME);
+    assertThat(actual).isEqualTo(expected);
+  }
+}