[apache#7185] Improvement(doris): Doris supports 'DISTRIBUTED BY HASH(columnName) BUCKETS AUTO' (apache#7186)

<!--
1. Title: [#<issue>] <type>(<scope>): <subject>
   Examples:
     - "[apache#123] feat(operator): support xxx"
     - "[apache#233] fix: check null before access result in xxx"
     - "[MINOR] refactor: fix typo in variable name"
     - "[MINOR] docs: fix typo in README"
     - "[apache#255] test: fix flaky test NameOfTheTest"
   Reference: https://www.conventionalcommits.org/en/v1.0.0/
2. If the PR is unfinished, please mark this PR as draft.
-->

### What changes were proposed in this pull request?

Doris supports 'DISTRIBUTED BY HASH(dt_date) BUCKETS AUTO'

### Why are the changes needed?

Fix: #([7185](apache#7185))

### Does this PR introduce _any_ user-facing change?

N/A

### How was this patch tested?


org.apache.gravitino.catalog.doris.utils.TestDorisUtils#testDistributedInfoPattern

org.apache.gravitino.catalog.doris.integration.test.CatalogDorisIT#testAllDistributionWithAuto

Author: sunxiaojian

api/src/main/java/org/apache/gravitino/rel/expressions/distributions/Distributions.java

@@ -26,6 +26,12 @@
 /** Helper methods to create distributions to pass into Apache Gravitino. */
 public class Distributions {
 
+  /**
+   * AUTO indicates that the number of buckets is automatically determined by the system (without
+   * the need for manual specification).
+   */
+  public static final int AUTO = -1;
+
   /** NONE is used to indicate that there is no distribution. */
   public static final Distribution NONE =
       new DistributionImpl(Strategy.NONE, 0, Expression.EMPTY_EXPRESSION);
@@ -74,6 +80,17 @@ public static Distribution of(Strategy strategy, int number, Expression... expre
     return new DistributionImpl(strategy, number, expressions);
   }
 
+  /**
+   * Create a distribution by the given strategy, the number of buckets is not used.
+   *
+   * @param strategy The strategy to use
+   * @param expressions The expressions to distribute by
+   * @return The created distribution
+   */
+  public static Distribution auto(Strategy strategy, Expression... expressions) {
+    return new DistributionImpl(strategy, AUTO, expressions);
+  }
+
   /**
    * Create a distribution on columns. Like distribute by (a) or (a, b), for complex like
    * distributing by (func(a), b) or (func(a), func(b)), please use {@link DistributionImpl.Builder}

catalogs/catalog-jdbc-doris/src/main/java/org/apache/gravitino/catalog/doris/operation/DorisTableOperations.java

@@ -134,7 +134,7 @@ protected String generateCreateTableSql(
     }
 
     if (distribution.number() != 0) {
-      sqlBuilder.append(" BUCKETS ").append(distribution.number());
+      sqlBuilder.append(" BUCKETS ").append(DorisUtils.toBucketNumberString(distribution.number()));
     }
 
     properties = appendNecessaryProperties(properties);

catalogs/catalog-jdbc-doris/src/main/java/org/apache/gravitino/catalog/doris/utils/DorisUtils.java

@@ -29,6 +29,7 @@
 import java.util.stream.Collectors;
 import org.apache.gravitino.rel.expressions.NamedReference;
 import org.apache.gravitino.rel.expressions.distributions.Distribution;
+import org.apache.gravitino.rel.expressions.distributions.Distributions;
 import org.apache.gravitino.rel.expressions.distributions.Distributions.DistributionImpl;
 import org.apache.gravitino.rel.expressions.distributions.Strategy;
 import org.apache.gravitino.rel.expressions.literals.Literal;
@@ -48,7 +49,7 @@ public final class DorisUtils {
 
   private static final Pattern DISTRIBUTION_INFO_PATTERN =
       Pattern.compile(
-          "DISTRIBUTED BY\\s+(HASH|RANDOM)\\s*(\\(([^)]+)\\))?\\s*(BUCKETS\\s+(\\d+))?");
+          "DISTRIBUTED BY\\s+(HASH|RANDOM)\\s*(\\(([^)]+)\\))?\\s*(BUCKETS\\s+(\\d+|AUTO))?");
 
   private static final String LIST_PARTITION = "LIST";
   private static final String RANGE_PARTITION = "RANGE";
@@ -202,11 +203,8 @@ public static Distribution extractDistributionInfoFromSql(String createTableSql)
                   .map(f -> f.substring(1, f.length() - 1))
                   .toArray(String[]::new);
 
-      // Default bucket number is 1.
-      int bucketNum = 1;
-      if (matcher.find(5)) {
-        bucketNum = Integer.valueOf(matcher.group(5));
-      }
+      // Default bucket number is 1, auto is -1.
+      int bucketNum = extractBucketNum(matcher);
 
       return new DistributionImpl.Builder()
           .withStrategy(Strategy.getByName(distributionType))
@@ -220,4 +218,21 @@ public static Distribution extractDistributionInfoFromSql(String createTableSql)
 
     throw new RuntimeException("Failed to extract distribution info in sql:" + createTableSql);
   }
+
+  private static int extractBucketNum(Matcher matcher) {
+    int bucketNum = 1;
+    if (matcher.find(5)) {
+      String bucketValue = matcher.group(5);
+      // Use -1 to indicate auto bucket.
+      bucketNum =
+          bucketValue.trim().toUpperCase().equals("AUTO")
+              ? Distributions.AUTO
+              : Integer.valueOf(bucketValue);
+    }
+    return bucketNum;
+  }
+
+  public static String toBucketNumberString(int number) {
+    return number == Distributions.AUTO ? "AUTO" : String.valueOf(number);
+  }
 }

catalogs/catalog-jdbc-doris/src/test/java/org/apache/gravitino/catalog/doris/integration/test/CatalogDorisIT.java

@@ -61,6 +61,7 @@
 import org.apache.gravitino.rel.expressions.NamedReference;
 import org.apache.gravitino.rel.expressions.distributions.Distribution;
 import org.apache.gravitino.rel.expressions.distributions.Distributions;
+import org.apache.gravitino.rel.expressions.distributions.Strategy;
 import org.apache.gravitino.rel.expressions.literals.Literal;
 import org.apache.gravitino.rel.expressions.literals.Literals;
 import org.apache.gravitino.rel.expressions.sorts.SortOrder;
@@ -1004,4 +1005,35 @@ void testAllDistribution() {
       tableCatalog.dropTable(tableIdentifier);
     }
   }
+
+  @Test
+  void testAllDistributionWithAuto() {
+    Distribution distribution =
+        Distributions.auto(Strategy.HASH, NamedReference.field(DORIS_COL_NAME1));
+
+    String tableName = GravitinoITUtils.genRandomName("test_distribution_table");
+    NameIdentifier tableIdentifier = NameIdentifier.of(schemaName, tableName);
+    Column[] columns = createColumns();
+    Index[] indexes = Indexes.EMPTY_INDEXES;
+    Map<String, String> properties = createTableProperties();
+    Transform[] partitioning = Transforms.EMPTY_TRANSFORM;
+    TableCatalog tableCatalog = catalog.asTableCatalog();
+    tableCatalog.createTable(
+        tableIdentifier,
+        columns,
+        table_comment,
+        properties,
+        partitioning,
+        distribution,
+        null,
+        indexes);
+    // load table
+    Table loadTable = tableCatalog.loadTable(tableIdentifier);
+
+    Assertions.assertEquals(distribution.strategy(), loadTable.distribution().strategy());
+    Assertions.assertEquals(distribution.number(), loadTable.distribution().number());
+    Assertions.assertArrayEquals(
+        distribution.expressions(), loadTable.distribution().expressions());
+    tableCatalog.dropTable(tableIdentifier);
+  }
 }

catalogs/catalog-jdbc-doris/src/test/java/org/apache/gravitino/catalog/doris/utils/TestDorisUtils.java

@@ -26,6 +26,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
+import org.apache.gravitino.rel.expressions.distributions.Distribution;
 import org.apache.gravitino.rel.expressions.literals.Literal;
 import org.apache.gravitino.rel.expressions.literals.Literals;
 import org.apache.gravitino.rel.expressions.transforms.Transform;
@@ -171,4 +172,17 @@ public void testGeneratePartitionSqlFragment() {
     partitionSqlFragment = DorisUtils.generatePartitionSqlFragment(partition);
     assertEquals("PARTITION `p7` VALUES IN ((\"1\",\"2\"),(\"3\",\"4\"))", partitionSqlFragment);
   }
+
+  @Test
+  public void testDistributedInfoPattern() {
+    String createTableSql =
+        "CREATE TABLE `testTable` (\n`col1` date NOT NULL\n) ENGINE=OLAP\n PARTITION BY RANGE(`col1`)\n()\n DISTRIBUTED BY HASH(`col1`) BUCKETS 2";
+    Distribution distribution = DorisUtils.extractDistributionInfoFromSql(createTableSql);
+    assertEquals(distribution.number(), 2);
+
+    String createTableSqlWithAuto =
+        "CREATE TABLE `testTable` (\n`col1` date NOT NULL\n) ENGINE=OLAP\n PARTITION BY RANGE(`col1`)\n()\n DISTRIBUTED BY HASH(`col1`) BUCKETS AUTO";
+    Distribution distribution2 = DorisUtils.extractDistributionInfoFromSql(createTableSqlWithAuto);
+    assertEquals(distribution2.number(), -1);
+  }
 }

common/src/main/java/org/apache/gravitino/dto/rel/DistributionDTO.java

@@ -163,7 +163,8 @@ public DistributionDTO build() {
       strategy = strategy == null ? Strategy.HASH : strategy;
 
       Preconditions.checkState(args != null, "expressions cannot be null");
-      Preconditions.checkState(number >= 0, "bucketNum must be greater than 0");
+      // Check if the number of buckets is greater than -1, -1 is auto.
+      Preconditions.checkState(number >= -1, "bucketNum must be greater than -1");
       return new DistributionDTO(strategy, number, args);
     }
   }

docs/table-partitioning-distribution-sort-order-indexes.md

@@ -84,9 +84,21 @@ Distributions.of(Strategy.HASH, 4, NamedReference.field("score"));
 ```
 
 </TabItem>
+
+<TabItem value="java" label="Java">
+if you want to use auto distribution, you can use the following code, it will set the number is -1.
+
+```java
+// Auto distribution with strategy and fields
+Distributions.auto(Strategy.HASH, NamedReference.field("score"));
+```
+</TabItem>
+
 </Tabs>
 
 
+
+
 ## Sort ordering
 
 To define a sorted order table, you should use the following three components to construct a valid sorted order table.