Merge branch 'main' into issue_2678

.github/workflows/cron-integration-test.yml

@@ -0,0 +1,100 @@
+name: Cron Integration Test
+
+# Controls when the workflow will run
+on:
+  schedule: # Runs by default on main branch
+    - cron: '0 19 * * *' # Runs every day at 19:00 PM UTC, equal to 03:00 AM the next day in GMT+8 time zone
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: ${{ contains(github.ref, 'main') }}
+
+jobs:
+  changes:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - uses: dorny/paths-filter@v2
+        id: filter
+        with:
+          filters: |
+            source_changes:
+              - .github/**
+              - api/**
+              - bin/**
+              - catalogs/**
+              - clients/**
+              - common/**
+              - conf/**
+              - core/**
+              - dev/**
+              - gradle/**
+              - integration-test/**
+              - meta/**
+              - server/**
+              - server-common/**
+              - spark-connector/**
+              - trino-connector/**
+              - web/**
+              - docs/open-api/**
+              - build.gradle.kts
+              - gradle.properties
+              - gradlew
+              - setting.gradle.kts
+    outputs:
+      source_changes: ${{ steps.filter.outputs.source_changes }}
+
+  # Integration test for AMD64 architecture
+  test-amd64-arch:
+    needs: changes
+    if: needs.changes.outputs.source_changes == 'true'
+    runs-on: ubuntu-latest
+    timeout-minutes: 60
+    strategy:
+      matrix:
+        architecture: [linux/amd64]
+        java-version: [ 8, 11, 17 ]
+        test-mode: [ embedded, deploy ]
+    env:
+      DOCKER_RUN_NAME: hive-amd64
+      PLATFORM: ${{ matrix.architecture }}
+    steps:
+      - uses: actions/checkout@v3
+
+      - uses: actions/setup-java@v3
+        with:
+          java-version: ${{ matrix.java-version }}
+          distribution: 'temurin'
+
+      - name: Set up QEMU
+        uses: docker/setup-qemu-action@v2
+
+      - name: Package Gravitino
+        run: |
+          ./gradlew build -x test -PjdkVersion=${{ matrix.java-version }}
+          ./gradlew compileDistribution -x test -PjdkVersion=${{ matrix.java-version }}
+
+      - name: Setup debug Github Action
+        if: ${{ contains(github.event.pull_request.labels.*.name, 'debug action') }}
+        uses: csexton/debugger-action@master
+
+      - name: Free up disk space
+        run: |
+          dev/ci/util_free_space.sh
+
+      - name: Integration Test
+        id: integrationTest
+        run: |
+          ./gradlew test --rerun-tasks -PskipTests -PtestMode=${{ matrix.test-mode }} -PjdkVersion=${{ matrix.java-version }}
+
+      - name: Upload integrate tests reports
+        uses: actions/upload-artifact@v3
+        if: ${{ failure() && steps.integrationTest.outcome == 'failure' }}
+        with:
+          name: integrate test reports
+          path: |
+            build/reports
+            integration-test/build/integration-test.log
+            distribution/package/logs/gravitino-server.out
+            distribution/package/logs/gravitino-server.log
+            catalogs/**/*.log

api/src/main/java/com/datastrato/gravitino/Catalog.java

@@ -27,7 +27,10 @@ enum Type {
     FILESET,
 
     /** Catalog Type for Message Queue, like kafka://topic */
-    MESSAGING
+    MESSAGING,
+
+    /** Catalog Type for test only. */
+    UNSUPPORTED
   }
 
   /**

api/src/main/java/com/datastrato/gravitino/NameIdentifier.java

@@ -113,6 +113,21 @@ public static NameIdentifier ofFileset(
     return NameIdentifier.of(metalake, catalog, schema, fileset);
   }
 
+  /**
+   * Create the topic {@link NameIdentifier} with the given metalake, catalog, schema and topic
+   * name.
+   *
+   * @param metalake The metalake name
+   * @param catalog The catalog name
+   * @param schema The schema name
+   * @param topic The topic name
+   * @return The created topic {@link NameIdentifier}
+   */
+  public static NameIdentifier ofTopic(
+      String metalake, String catalog, String schema, String topic) {
+    return NameIdentifier.of(metalake, catalog, schema, topic);
+  }
+
   /**
    * Check the given {@link NameIdentifier} is a metalake identifier. Throw an {@link
    * IllegalNameIdentifierException} if it's not.
@@ -170,6 +185,17 @@ public static void checkFileset(NameIdentifier ident) {
     Namespace.checkFileset(ident.namespace);
   }
 
+  /**
+   * Check the given {@link NameIdentifier} is a topic identifier. Throw an {@link
+   * IllegalNameIdentifierException} if it's not.
+   *
+   * @param ident The topic {@link NameIdentifier} to check.
+   */
+  public static void checkTopic(NameIdentifier ident) {
+    check(ident != null, "Topic identifier must not be null");
+    Namespace.checkTopic(ident.namespace);
+  }
+
   /**
    * Create a {@link NameIdentifier} from the given identifier string.
    *

api/src/main/java/com/datastrato/gravitino/Namespace.java

@@ -105,6 +105,18 @@ public static Namespace ofFileset(String metalake, String catalog, String schema
     return of(metalake, catalog, schema);
   }
 
+  /**
+   * Create a namespace for topic.
+   *
+   * @param metalake The metalake name
+   * @param catalog The catalog name
+   * @param schema The schema name
+   * @return A namespace for topic
+   */
+  public static Namespace ofTopic(String metalake, String catalog, String schema) {
+    return of(metalake, catalog, schema);
+  }
+
   /**
    * Check if the given metalake namespace is legal, throw an {@link IllegalNamespaceException} if
    * it's illegal.
@@ -170,6 +182,19 @@ public static void checkFileset(Namespace namespace) {
         namespace);
   }
 
+  /**
+   * Check if the given topic namespace is legal, throw an {@link IllegalNamespaceException} if it's
+   * illegal.
+   *
+   * @param namespace The topic namespace
+   */
+  public static void checkTopic(Namespace namespace) {
+    check(
+        namespace != null && namespace.length() == 3,
+        "Topic namespace must be non-null and have 3 levels, the input namespace is %s",
+        namespace);
+  }
+
   private Namespace(String[] levels) {
     this.levels = levels;
   }

api/src/main/java/com/datastrato/gravitino/authorization/Group.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2024 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+package com.datastrato.gravitino.authorization;
+
+import com.datastrato.gravitino.Auditable;
+import com.datastrato.gravitino.annotation.Evolving;
+import java.util.List;
+
+/** The interface of a Group. The Group is the entity which contains users. */
+@Evolving
+public interface Group extends Auditable {
+
+  /**
+   * The name of the group.
+   *
+   * @return The name of the group.
+   */
+  String name();
+
+  /**
+   * The roles of the group.
+   *
+   * @return The roles of the group.
+   */
+  List<String> roles();
+}

api/src/main/java/com/datastrato/gravitino/exceptions/GroupAlreadyExistsException.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2024 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+
+package com.datastrato.gravitino.exceptions;
+
+import com.google.errorprone.annotations.FormatMethod;
+import com.google.errorprone.annotations.FormatString;
+
+/** An exception thrown when a group already exists. */
+public class GroupAlreadyExistsException extends AlreadyExistsException {
+
+  /**
+   * Constructs a new exception with the specified detail message.
+   *
+   * @param message the detail message.
+   * @param args the arguments to the message.
+   */
+  @FormatMethod
+  public GroupAlreadyExistsException(@FormatString String message, Object... args) {
+    super(message, args);
+  }
+
+  /**
+   * Constructs a new exception with the specified detail message and cause.
+   *
+   * @param cause the cause.
+   * @param message the detail message.
+   * @param args the arguments to the message.
+   */
+  @FormatMethod
+  public GroupAlreadyExistsException(Throwable cause, String message, Object... args) {
+    super(cause, message, args);
+  }
+}

api/src/main/java/com/datastrato/gravitino/exceptions/NoSuchGroupException.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2024 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+
+package com.datastrato.gravitino.exceptions;
+
+import com.google.errorprone.annotations.FormatMethod;
+import com.google.errorprone.annotations.FormatString;
+
+/** Exception thrown when a group with specified name is not existed. */
+public class NoSuchGroupException extends NotFoundException {
+  /**
+   * Constructs a new exception with the specified detail message.
+   *
+   * @param message the detail message.
+   * @param args the arguments to the message.
+   */
+  @FormatMethod
+  public NoSuchGroupException(@FormatString String message, Object... args) {
+    super(message, args);
+  }
+
+  /**
+   * Constructs a new exception with the specified detail message and cause.
+   *
+   * @param cause the cause.
+   * @param message the detail message.
+   * @param args the arguments to the message.
+   */
+  @FormatMethod
+  public NoSuchGroupException(Throwable cause, String message, Object... args) {
+    super(cause, message, args);
+  }
+}

api/src/main/java/com/datastrato/gravitino/messaging/TopicCatalog.java

@@ -85,8 +85,7 @@ Topic alterTopic(NameIdentifier ident, TopicChange... changes)
    * Drop a topic from the catalog.
    *
    * @param ident A topic identifier.
-   * @return true If the topic is dropped, false otherwise.
-   * @throws NoSuchTopicException If the topic does not exist.
+   * @return true If the topic is dropped, false if the topic does not exist.
    */
-  boolean dropTopic(NameIdentifier ident) throws NoSuchTopicException;
+  boolean dropTopic(NameIdentifier ident);
 }

catalogs/catalog-hadoop/src/test/java/com/datastrato/gravitino/catalog/hadoop/TestHadoopCatalogOperations.java

@@ -8,7 +8,7 @@
 import static com.datastrato.gravitino.Configs.ENTITY_KV_STORE;
 import static com.datastrato.gravitino.Configs.ENTITY_STORE;
 import static com.datastrato.gravitino.Configs.ENTRY_KV_ROCKSDB_BACKEND_PATH;
-import static com.datastrato.gravitino.Configs.KV_DELETE_AFTER_TIME;
+import static com.datastrato.gravitino.Configs.STORE_DELETE_AFTER_TIME;
 import static com.datastrato.gravitino.Configs.STORE_TRANSACTION_MAX_SKEW_TIME;
 import static com.datastrato.gravitino.connector.BaseCatalog.CATALOG_BYPASS_PREFIX;
 
@@ -78,7 +78,7 @@ public static void setUp() {
 
     Assertions.assertEquals(ROCKS_DB_STORE_PATH, config.get(ENTRY_KV_ROCKSDB_BACKEND_PATH));
     Mockito.when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
-    Mockito.when(config.get(KV_DELETE_AFTER_TIME)).thenReturn(20 * 60 * 1000L);
+    Mockito.when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 * 1000L);
 
     store = EntityStoreFactory.createEntityStore(config);
     store.initialize(config);

catalogs/catalog-hive/src/main/java/com/datastrato/gravitino/catalog/hive/HiveTablePropertiesMetadata.java

@@ -57,11 +57,11 @@ public class HiveTablePropertiesMetadata extends BasePropertiesMetadata {
   @VisibleForTesting
   public static final String ORC_SERDE_CLASS = "org.apache.hadoop.hive.ql.io.orc.OrcSerde";
 
-  private static final String PARQUET_INPUT_FORMAT_CLASS =
+  public static final String PARQUET_INPUT_FORMAT_CLASS =
       "org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat";
-  private static final String PARQUET_OUTPUT_FORMAT_CLASS =
+  public static final String PARQUET_OUTPUT_FORMAT_CLASS =
       "org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat";
-  private static final String PARQUET_SERDE_CLASS =
+  public static final String PARQUET_SERDE_CLASS =
       "org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe";
   private static final String COLUMNAR_SERDE_CLASS =
       "org.apache.hadoop.hive.serde2.columnar.ColumnarSerDe";
@@ -89,7 +89,10 @@ public enum TableType {
     VIRTUAL_INDEX,
   }
 
-  enum StorageFormat {
+  // In embedded test mode, HiveTablePropertiesMetadata will be loaded by spark connector which has
+  // different classloaders with Hive catalog. If StorageFormat is package scope, it couldn't
+  // be accessed by Hive catalog related classes in same package, so making it public.
+  public enum StorageFormat {
     SEQUENCEFILE(
         SEQUENCEFILE_INPUT_FORMAT_CLASS, SEQUENCEFILE_OUTPUT_FORMAT_CLASS, LAZY_SIMPLE_SERDE_CLASS),
     TEXTFILE(TEXT_INPUT_FORMAT_CLASS, IGNORE_KEY_OUTPUT_FORMAT_CLASS, LAZY_SIMPLE_SERDE_CLASS),

catalogs/catalog-jdbc-common/src/main/java/com/datastrato/gravitino/catalog/jdbc/JdbcTablePropertiesMetadata.java

@@ -6,6 +6,7 @@
 
 import com.datastrato.gravitino.StringIdentifier;
 import com.datastrato.gravitino.connector.BasePropertiesMetadata;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Maps;
 import java.util.HashMap;
 import java.util.Map;
@@ -14,13 +15,15 @@ public abstract class JdbcTablePropertiesMetadata extends BasePropertiesMetadata
 
   public static final String COMMENT_KEY = "comment";
 
-  protected Map<String, String> transformToJdbcProperties(Map<String, String> properties) {
+  @VisibleForTesting
+  public Map<String, String> transformToJdbcProperties(Map<String, String> properties) {
     HashMap<String, String> resultProperties = Maps.newHashMap(properties);
     resultProperties.remove(StringIdentifier.ID_KEY);
     return resultProperties;
   }
 
-  protected Map<String, String> convertFromJdbcProperties(Map<String, String> properties) {
+  @VisibleForTesting
+  public Map<String, String> convertFromJdbcProperties(Map<String, String> properties) {
     return properties;
   }
 }

catalogs/catalog-jdbc-common/src/main/java/com/datastrato/gravitino/catalog/jdbc/converter/JdbcColumnDefaultValueConverter.java

@@ -19,6 +19,8 @@ public class JdbcColumnDefaultValueConverter {
   protected static final String NULL = "NULL";
   protected static final DateTimeFormatter DATE_TIME_FORMATTER =
       DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+  protected static final DateTimeFormatter DATE_FORMATTER =
+      DateTimeFormatter.ofPattern("yyyy-MM-dd");
 
   public String fromGravitino(Expression defaultValue) {
     if (DEFAULT_VALUE_NOT_SET.equals(defaultValue)) {

catalogs/catalog-jdbc-mysql/src/main/java/com/datastrato/gravitino/catalog/mysql/MysqlTablePropertiesMetadata.java

@@ -105,7 +105,7 @@ public Map<String, String> transformToJdbcProperties(Map<String, String> propert
   }
 
   @Override
-  protected Map<String, String> convertFromJdbcProperties(Map<String, String> properties) {
+  public Map<String, String> convertFromJdbcProperties(Map<String, String> properties) {
     BidiMap<String, String> mysqlConfigToGravitino = GRAVITINO_CONFIG_TO_MYSQL.inverseBidiMap();
     return Collections.unmodifiableMap(
         new HashMap<String, String>() {