feat(fuse): add hash-distributed partition writes

Cargo.toml

@@ -355,6 +355,7 @@ memchr = { version = "2", default-features = false }
 micromarshal = "0.7.0"
 mlua = { version = "0.11", features = ["lua54", "vendored", "async", "serialize"] }
 mockall = "0.11.2"
+murmur3 = "0.5.2"
 mysql_async = { version = "0.34", default-features = false, features = ["native-tls-tls"] }
 naive-cityhash = "0.2.0"
 ndarray = "0.15.6"

src/query/ast/src/ast/statements/table.rs

@@ -162,7 +162,7 @@ pub struct CreateTableStmt {
     pub uri_location: Option<UriLocation>,
     pub cluster_by: Option<ClusterOption>,
     pub table_options: BTreeMap<String, String>,
-    pub iceberg_table_partition: Option<Vec<Identifier>>,
+    pub partition_by: Option<Vec<Expr>>,
     pub table_properties: Option<BTreeMap<String, String>>,
     pub as_query: Option<Box<Query>>,
     pub table_type: TableType,
@@ -211,6 +211,12 @@ impl Display for CreateTableStmt {
             write!(f, " {uri_location}")?;
         }
 
+        if let Some(partition_by) = &self.partition_by {
+            write!(f, " PARTITION BY(")?;
+            write_comma_separated_list(f, partition_by)?;
+            write!(f, ")")?;
+        }
+
         if let Some(cluster_by) = &self.cluster_by {
             write!(f, " {cluster_by}")?;
         }
@@ -221,12 +227,6 @@ impl Display for CreateTableStmt {
             write_space_separated_string_map(f, &self.table_options)?;
         }
 
-        if let Some(iceberg_table_partition) = &self.iceberg_table_partition {
-            write!(f, " PARTITION BY(")?;
-            write_comma_separated_list(f, iceberg_table_partition)?;
-            write!(f, ")")?;
-        }
-
         if let Some(table_properties) = &self.table_properties {
             write!(f, " PROPERTIES(")?;
             write_space_separated_string_map(f, table_properties)?;

src/query/ast/src/parser/statement.rs

@@ -1131,16 +1131,22 @@ pub fn statement_body(i: Input) -> IResult<Statement> {
             })
         },
     );
+    let create_table_partition_by = map(
+        rule! {
+            #table_option ~ PARTITION ~ ^BY ~ ^"(" ~ ^#comma_separated_list1(expr) ~ ^")"
+        },
+        |(table_options, _, _, _, exprs, _)| (table_options, exprs),
+    );
     let create_table = map_res(
         rule! {
             CREATE ~ ( OR ~ ^REPLACE )? ~ (TEMP| TEMPORARY|TRANSIENT)? ~ TABLE ~ ( IF ~ ^NOT ~ ^EXISTS )?
             ~ #dot_separated_idents_1_to_3
             ~ #create_table_source?
             ~ ( #engine )?
             ~ ( #uri_location )?
+            ~ #create_table_partition_by?
             ~ ( CLUSTER ~ ^BY ~ LINEAR? ~ ^"(" ~ ^#comma_separated_list1(expr) ~ ^")" )?
             ~ ( #table_option )?
-            ~ ( PARTITION ~ ^BY ~ ^"(" ~ ^#comma_separated_list1(ident) ~ ^")" )?
             ~ ( PROPERTIES ~  #connection_options )?
             ~ ( AS ~ ^#query )?
         },
@@ -1154,9 +1160,9 @@ pub fn statement_body(i: Input) -> IResult<Statement> {
             source,
             engine,
             uri_location,
+            opt_partition_by,
             opt_cluster_by,
             opt_table_options,
-            opt_iceberg_table_partition_by,
             opt_table_properties,
             opt_as_query,
         )| {
@@ -1168,6 +1174,10 @@ pub fn statement_body(i: Input) -> IResult<Statement> {
                 Some(TEMP) | Some(TEMPORARY) => TableType::Temporary,
                 _ => unreachable!(),
             };
+            let (mut table_options, partition_by) = opt_partition_by
+                .map(|(options, exprs)| (options, Some(exprs)))
+                .unwrap_or_default();
+            table_options.extend(opt_table_options.unwrap_or_default());
             Ok(Statement::CreateTable(CreateTableStmt {
                 create_option,
                 catalog,
@@ -1179,9 +1189,8 @@ pub fn statement_body(i: Input) -> IResult<Statement> {
                 cluster_by: opt_cluster_by.map(|(_, _, _, _, exprs, _)| ClusterOption {
                     cluster_exprs: exprs,
                 }),
-                table_options: opt_table_options.unwrap_or_default(),
-                iceberg_table_partition: opt_iceberg_table_partition_by
-                    .map(|(_, _, _, cols, _)| cols),
+                table_options,
+                partition_by,
                 table_properties: opt_table_properties.map(|(_, properties)| properties),
                 as_query: opt_as_query.map(|(_, query)| Box::new(query)),
                 table_type,

src/query/ast/src/visit/statement_table.rs

@@ -149,9 +149,9 @@ impl Walk for CreateTableStmt {
         if let Some(cluster_by) = &self.cluster_by {
             try_walk!(cluster_by.walk(visitor));
         }
-        if let Some(partitions) = &self.iceberg_table_partition {
-            for ident in partitions {
-                try_walk!(ident.walk(visitor));
+        if let Some(partitions) = &self.partition_by {
+            for expr in partitions {
+                try_walk!(expr.walk(visitor));
             }
         }
         if let Some(as_query) = &self.as_query {
@@ -173,9 +173,9 @@ impl WalkMut for CreateTableStmt {
         if let Some(cluster_by) = &mut self.cluster_by {
             try_walk!(cluster_by.walk_mut(visitor));
         }
-        if let Some(partitions) = &mut self.iceberg_table_partition {
-            for ident in partitions {
-                try_walk!(ident.walk_mut(visitor));
+        if let Some(partitions) = &mut self.partition_by {
+            for expr in partitions {
+                try_walk!(expr.walk_mut(visitor));
             }
         }
         if let Some(as_query) = &mut self.as_query {

src/query/ast/tests/it/parser.rs

@@ -1393,6 +1393,32 @@ fn test_file_format_trim_space_option() {
     assert!(displayed.contains("TRIM_SPACE = true".to_uppercase().as_str()));
 }
 
+#[test]
+fn test_create_table_options_before_partition_by() {
+    let cases = [
+        "CREATE TABLE t(c INT) ENGINE=ICEBERG LOCATION='s3://bucket/path' CONNECTION_NAME='conn' PARTITION BY (c)",
+        "CREATE TABLE iceberg.db.t(c INT) LOCATION='s3://bucket/path' PARTITION BY (c)",
+        "CREATE TABLE t(a INT) ENGINE=FUSE ROW_PER_BLOCK=1 PARTITION BY (a)",
+        "CREATE TABLE t(a INT) ROW_PER_BLOCK=1 PARTITION BY (a)",
+    ];
+    for sql in cases {
+        let tokens = tokenize_sql(sql).unwrap();
+        let (stmt, _) = parse_sql(&tokens, Dialect::PostgreSQL).unwrap();
+
+        let displayed = stmt.to_string();
+        let displayed_uppercase = displayed.to_uppercase();
+        let partition_pos = displayed_uppercase.find("PARTITION BY").unwrap();
+        for option in ["LOCATION", "CONNECTION_NAME", "ROW_PER_BLOCK"] {
+            if let Some(option_pos) = displayed_uppercase.find(option) {
+                assert!(partition_pos < option_pos);
+            }
+        }
+
+        let tokens = tokenize_sql(&displayed).unwrap();
+        parse_sql(&tokens, Dialect::PostgreSQL).unwrap();
+    }
+}
+
 #[test]
 fn test_stage_local_filesystem_uri_errors() {
     let cases = [

src/query/ast/tests/it/testdata/stmt-error.txt

@@ -94,8 +94,10 @@ error:
   --> SQL:1:55
   |
 1 | create table a (c1 decimal(38), c2 int) partition by ();
-  | ------                                                ^ unexpected `)`, expecting <Ident>, <LiteralString>, or `IDENTIFIER`
-  | |                                                      
+  | ------                                                ^
+  | |                                                     |
+  | |                                                     expecting `<LiteralString>`, '<LiteralCodeString>', '<LiteralInteger>', '<LiteralFloat>', 'TRUE', 'FALSE', or more ...
+  | |                                                     while parsing expression
   | while parsing `CREATE [OR REPLACE] TABLE [IF NOT EXISTS] [<database>.]<table> [<source>] [<table_options>]`