resolve conflict

Cargo.toml

@@ -29,6 +29,7 @@ members = [
   "src/utils/memcomparable",
   "src/utils/pgwire",
   "src/utils/static-link",
+  "src/utils/stats_alloc",
   "src/utils/value-encoding",
   "src/workspace-hack",
 ]

e2e_test/batch/join.slt

@@ -39,4 +39,9 @@ statement ok
 drop table t2;
 
 statement ok
-drop table t3;
+drop table t3;
+
+query I
+select count(*) from (values (1, 2), (3, 4)) as a, (values (9),(4),(1)) as b;
+----
+6

e2e_test/batch/local_mode.slt

@@ -1,4 +1,7 @@
 statement ok
 SET QUERY_MODE TO local;
 
-include ./boolean.slt.part
+#include ./boolean.slt.part
+
+statement ok
+SET QUERY_MODE TO distributed;

e2e_test/ddl/privilege.slt

@@ -0,0 +1,83 @@
+# Create a super user.
+statement ok
+CREATE USER user WITH SUPERUSER PASSWORD 'password';
+
+# Create a user.
+statement ok
+CREATE USER user1 WITH PASSWORD 'password1';
+
+# Create a database.
+statement ok
+CREATE DATABASE db1;
+
+# Create a schema.
+statement ok
+CREATE SCHEMA db1.schema1;
+
+# Grant privilege for user1.
+statement ok
+GRANT ALL ON DATABASE db1 TO user1 WITH GRANT OPTION GRANTED BY user;
+
+# Grant invalid privilege on database for user1.
+statement error
+GRANT INSERT ON DATABASE db1 TO user1 WITH GRANT OPTION GRANTED BY user;
+
+# Grant privilege on invalid database for user1.
+statement error
+GRANT ALL ON DATABASE db_invalid TO user1 WITH GRANT OPTION GRANTED BY user;
+
+# Grant privilege on database for invalid user.
+statement error
+GRANT ALL ON DATABASE db_invalid TO user_invalid WITH GRANT OPTION GRANTED BY user;
+
+# Grant privilege on schema for user1.
+statement ok
+GRANT CREATE ON SCHEMA db1.schema1 TO user1 WITH GRANT OPTION GRANTED BY user;
+
+# Grant privilege on all sources in schema for user1.
+statement ok
+GRANT ALL PRIVILEGES ON ALL SOURCES IN SCHEMA db1.schema1 TO user1 GRANTED BY user;
+
+# Grant privilege on all mviews in schema for user1.
+statement ok
+GRANT ALL PRIVILEGES ON ALL MATERIALIZED VIEWS IN SCHEMA db1.schema1 TO user1 GRANTED BY user;
+
+# Revoke privilege on all mviews in schema for user1.
+statement ok
+REVOKE ALL PRIVILEGES ON ALL MATERIALIZED VIEWS IN SCHEMA db1.schema1 FROM user1;
+
+# Revoke privilege on all sources in schema for user1.
+statement ok
+REVOKE ALL PRIVILEGES ON ALL SOURCES IN SCHEMA db1.schema1 FROM user1;
+
+# Revoke privilege on schema for user1.
+statement ok
+REVOKE CREATE ON SCHEMA db1.schema1 FROM user1;
+
+# Revoke GRANT OPTION FOR from database for user1.
+statement ok
+REVOKE GRANT OPTION FOR ALL ON DATABASE db1 from user1 GRANTED BY user;
+
+# Revoke privilege on database for user1.
+statement ok
+REVOKE ALL ON DATABASE db1 FROM user1;
+
+# Drop schema
+statement ok
+DROP SCHEMA db1.schema1;
+
+# Drop schema
+statement ok
+DROP SCHEMA db1.dev;
+
+# Drop database
+statement ok
+DROP DATABASE db1;
+
+# Drop user1
+statement ok
+DROP USER user1;
+
+# Drop user
+statement ok
+DROP USER user;

proto/hummock.proto

@@ -126,7 +126,7 @@ message CompactTask {
   repeated KeyRange splits = 2;
   // low watermark in 'ts-aware compaction'
   uint64 watermark = 3;
-  // compacion output, which will be added to [`target_level`] of LSM after compaction
+  // compaction output, which will be added to [`target_level`] of LSM after compaction
   repeated SstableInfo sorted_output_ssts = 4;
   // task id assigned by hummock storage service
   uint64 task_id = 5;
@@ -135,18 +135,11 @@ message CompactTask {
   bool is_target_ultimate_and_leveling = 7;
   CompactMetrics metrics = 8;
   bool task_status = 9;
-  repeated CompactionGroup.PrefixPair prefix_pairs = 10;
   // Hash mapping from virtual node to parallel unit. Since one compactor might deal with SSTs
   // with data for more than one relational state tables, here a vector is required.
   repeated common.ParallelUnitMapping vnode_mappings = 11;
-}
-
-message CompactionGroup {
-  message PrefixPair {
-    // key value with `prefix` belongs to compaction group `group_id`
-    uint64 group_id = 1;
-    bytes prefix = 2;
-  }
+  // compaction group the task belongs to
+  uint64 compaction_group_id = 12;
 }
 
 message LevelHandler {
@@ -160,7 +153,15 @@ message LevelHandler {
 }
 
 message CompactStatus {
-  repeated LevelHandler level_handlers = 1;
+  uint64 compaction_group_id = 1;
+  repeated LevelHandler level_handlers = 2;
+  CompactionConfig compaction_config = 3;
+}
+
+message CompactionGroup {
+  uint64 id = 1;
+  repeated bytes member_prefixes = 2;
+  CompactionConfig compaction_config = 3;
 }
 
 message CompactTaskAssignment {
@@ -240,3 +241,18 @@ service HummockManagerService {
 }
 
 service CompactorService {}
+
+message CompactionConfig {
+  enum CompactionMode {
+    RANGE = 0;
+    CONSISTENT_HASH = 1;
+  }
+  uint64 max_bytes_for_level_base = 1;
+  uint64 max_level = 2;
+  uint64 max_bytes_for_level_multiplier = 3;
+  uint64 max_compaction_bytes = 4;
+  uint64 min_compaction_bytes = 5;
+  uint64 level0_tigger_file_numer = 6;
+  uint64 level0_tier_compact_file_number = 7;
+  CompactionMode compaction_mode = 8;
+}

proto/user.proto

@@ -32,39 +32,7 @@ message UserInfo {
 
 /// GrantPrivilege defines a privilege granted to a user.
 message GrantPrivilege {
-  message GrantDatabase {
-    uint32 database_id = 1;
-  }
-
-  message GrantSchema {
-    uint32 database_id = 1;
-    uint32 schema_id = 2;
-  }
-
-  message GrantTable {
-    uint32 database_id = 1;
-    uint32 schema_id = 2;
-    uint32 table_id = 3;
-  }
-
-  message GrantSource {
-    uint32 database_id = 1;
-    uint32 schema_id = 2;
-    uint32 source_id = 3;
-  }
-
-  /// To support grant privilege on ALL TABLES IN SCHEMA schema_name.
-  message GrantAllTables {
-    uint32 database_id = 1;
-    uint32 schema_id = 2;
-  }
-
-  message GrantAllSources {
-    uint32 database_id = 1;
-    uint32 schema_id = 2;
-  }
-
-  enum Privilege {
+  enum Action {
     UNKNOWN = 0;
     SELECT = 1;
     INSERT = 2;
@@ -74,20 +42,20 @@ message GrantPrivilege {
     CONNECT = 6;
   }
 
-  message PrivilegeWithGrantOption {
-    Privilege privilege = 1;
+  message ActionWithGrantOption {
+    Action action = 1;
     bool with_grant_option = 2;
   }
 
-  oneof target {
-    GrantDatabase grant_database = 1;
-    GrantSchema grant_schema = 2;
-    GrantTable grant_table = 3;
-    GrantSource grant_source = 4;
-    GrantAllTables grant_all_tables = 5;
-    GrantAllSources grant_all_sources = 6;
+  oneof object {
+    uint32 database_id = 1;
+    uint32 schema_id = 2;
+    uint32 table_id = 3;
+    uint32 source_id = 4;
+    uint32 all_tables_schema_id = 5;
+    uint32 all_sources_schema_id = 6;
   }
-  repeated PrivilegeWithGrantOption privilege_with_opts = 7;
+  repeated ActionWithGrantOption action_with_opts = 7;
 }
 
 message CreateUserRequest {
@@ -109,7 +77,7 @@ message DropUserResponse {
 }
 
 message GrantPrivilegeRequest {
-  string user_name = 1;
+  repeated string users = 1;
   repeated GrantPrivilege privileges = 2;
   bool with_grant_option = 3;
 }
@@ -120,7 +88,7 @@ message GrantPrivilegeResponse {
 }
 
 message RevokePrivilegeRequest {
-  string user_name = 1;
+  repeated string users = 1;
   repeated GrantPrivilege privileges = 2;
   bool revoke_grant_option = 3;
 }

src/batch/src/executor/generate_series.rs

@@ -105,8 +105,9 @@ where
             }
 
             let arr = builder.finish()?;
+            let len = arr.len();
             let columns = vec![Column::new(Arc::new(arr.into()))];
-            let chunk: DataChunk = DataChunk::builder().columns(columns).build();
+            let chunk: DataChunk = DataChunk::new(columns, len);
 
             yield chunk;
         }

src/batch/src/executor/hash_agg.rs

@@ -232,8 +232,10 @@ impl<K: HashKey + Send + Sync> HashAggExecutor<K> {
                 .collect::<Result<Vec<_>>>()?;
 
             let mut has_next = false;
+            let mut array_len = 0;
             for (key, states) in result.by_ref().take(cardinality) {
                 has_next = true;
+                array_len += 1;
                 key.deserialize_to_builders(&mut group_builders[..])?;
                 states
                     .into_iter()
@@ -250,7 +252,7 @@ impl<K: HashKey + Send + Sync> HashAggExecutor<K> {
                 .map(|b| Ok(Column::new(Arc::new(b.finish()?))))
                 .collect::<Result<Vec<_>>>()?;
 
-            let output = DataChunk::builder().columns(columns).build();
+            let output = DataChunk::new(columns, array_len);
             yield output;
         }
     }

src/batch/src/executor/join/hash_join.rs

@@ -402,6 +402,7 @@ mod tests {
     struct DataChunkMerger {
         data_types: Vec<DataType>,
         array_builders: Vec<ArrayBuilderImpl>,
+        array_len: usize,
     }
 
     impl DataChunkMerger {
@@ -414,6 +415,7 @@ mod tests {
             Ok(Self {
                 data_types,
                 array_builders,
+                array_len: 0,
             })
         }
 
@@ -422,6 +424,7 @@ mod tests {
             for idx in 0..self.array_builders.len() {
                 self.array_builders[idx].append_array(data_chunk.column_at(idx).array_ref())?;
             }
+            self.array_len += data_chunk.capacity();
 
             Ok(())
         }
@@ -433,7 +436,7 @@ mod tests {
                 .map(|array_builder| array_builder.finish().map(|arr| Column::new(Arc::new(arr))))
                 .collect::<Result<Vec<Column>>>()?;
 
-            DataChunk::try_from(columns)
+            Ok(DataChunk::new(columns, self.array_len))
         }
     }